perf(qwp): speed up frequent flushes on the WebSocket sender

core/src/main/java/io/questdb/client/cutlass/qwp/client/MicrobatchBuffer.java

@@ -68,7 +68,6 @@ public class MicrobatchBuffer implements QuietCloseable {
     private int bufferPos;
     // Native memory buffer
     private long bufferPtr;
-    private long firstRowTimeNanos;
     // For waiting on recycle (user thread waits for I/O thread to finish)
     private volatile Thread recycleWaiter;
     // Row tracking
@@ -89,7 +88,6 @@ public MicrobatchBuffer(int initialCapacity) {
         this.bufferPtr = Unsafe.malloc(initialCapacity, MemoryTag.NATIVE_ILP_RSS);
         this.bufferPos = 0;
         this.rowCount = 0;
-        this.firstRowTimeNanos = 0;
         this.batchId = nextBatchId.getAndIncrement();
     }
 
@@ -195,16 +193,6 @@ public void ensureCapacity(int requiredCapacity) {
         }
     }
 
-    /**
-     * Returns the age of the first row in nanoseconds, or 0 if no rows.
-     */
-    public long getAgeNanos() {
-        if (rowCount == 0) {
-            return 0;
-        }
-        return System.nanoTime() - firstRowTimeNanos;
-    }
-
     /**
      * Returns the batch ID for this buffer.
      */
@@ -256,15 +244,12 @@ public boolean hasData() {
     }
 
     /**
-     * Increments the row count and records the first row time if this is the first row.
+     * Increments the row count.
      */
     public void incrementRowCount() {
         if (state != STATE_FILLING) {
             throw new IllegalStateException("Cannot increment row count when state is " + stateName(state));
         }
-        if (rowCount == 0) {
-            firstRowTimeNanos = System.nanoTime();
-        }
         rowCount++;
     }
 
@@ -348,7 +333,6 @@ public void reset() {
         }
         bufferPos = 0;
         rowCount = 0;
-        firstRowTimeNanos = 0;
         batchId = nextBatchId.getAndIncrement();
         recycleWaiter = null;
         state = STATE_FILLING;

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpColumnWriter.java

@@ -25,7 +25,6 @@
 package io.questdb.client.cutlass.qwp.client;
 
 import io.questdb.client.cutlass.line.LineSenderException;
-import io.questdb.client.cutlass.qwp.protocol.QwpColumnDef;
 import io.questdb.client.cutlass.qwp.protocol.QwpGorillaEncoder;
 import io.questdb.client.cutlass.qwp.protocol.QwpTableBuffer;
 import io.questdb.client.std.Unsafe;
@@ -128,6 +127,13 @@ private void encodeColumn(
         }
     }
 
+    private void putEncoded(byte[] bytes) {
+        int n = bytes.length;
+        buffer.ensureCapacity(n);
+        Unsafe.getUnsafe().copyMemory(bytes, Unsafe.BYTE_OFFSET, null, buffer.getWriteAddress(), n);
+        buffer.skip(n);
+    }
+
     private void writeBooleanColumn(long addr, int count) {
         int packedSize = (count + 7) / 8;
         for (int i = 0; i < packedSize; i++) {
@@ -279,14 +285,10 @@ private void writeSymbolColumnWithGlobalIds(QwpTableBuffer.ColumnBuffer col, int
         }
     }
 
-    private void writeTableHeader(String tableName, int rowCount, QwpColumnDef[] columns) {
-        buffer.putString(tableName);
+    private void writeTableHeader(QwpTableBuffer tableBuffer, int rowCount) {
+        putEncoded(tableBuffer.getEncodedTableName());
         buffer.putVarint(rowCount);
-        buffer.putVarint(columns.length);
-        for (QwpColumnDef col : columns) {
-            buffer.putString(col.getName());
-            buffer.putByte(col.getWireTypeCode());
-        }
+        putEncoded(tableBuffer.getEncodedColumnDefs());
     }
 
     private void writeTimestampColumn(long addr, int count, boolean useGorilla) {
@@ -328,9 +330,7 @@ void encodeTable(
             boolean useGlobalSymbols,
             boolean useGorilla
     ) {
-        QwpColumnDef[] columnDefs = tableBuffer.getColumnDefs();
-
-        writeTableHeader(tableBuffer.getTableName(), rowCount, columnDefs);
+        writeTableHeader(tableBuffer, rowCount);
 
         for (int i = 0; i < tableBuffer.getColumnCount(); i++) {
             QwpTableBuffer.ColumnBuffer col = tableBuffer.getColumn(i);

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java

@@ -246,6 +246,12 @@ public class QwpWebSocketSender implements Sender {
     // splitFrameBodyBytes, which is what lets the split passes index them directly.
     private final ObjList<QwpTableBuffer> flushTableBuffers = new ObjList<>();
     private final ObjList<CharSequence> flushTableNames = new ObjList<>();
+    // The tables that received a row since the last successful flush, in first-row
+    // order, each listed once (QwpTableBuffer.isListedPending() guards the add).
+    // collectNonEmptyTables() walks this instead of every table the sender has seen.
+    // Cleared, together with the flags, only when a flush succeeds or finds nothing
+    // to ship, so a flush that throws before publishing leaves it for the retry.
+    private final ObjList<QwpTableBuffer> pendingTables = new ObjList<>();
     private final IntList splitFrameBodyBytes = new IntList();
     private final CharSequenceObjHashMap<QwpTableBuffer> tableBuffers;
     // null means plain text (no TLS)
@@ -1902,6 +1908,7 @@ private void close0(boolean[] restoreInterrupt) {
                 }
             }
             tableBuffers.clear();
+            pendingTables.clear();
         } catch (Throwable t) {
             LOG.error("Error closing encoder or table buffers: {}", String.valueOf(t));
             terminalError = captureCloseError(terminalError, t);
@@ -4603,6 +4610,13 @@ private void checkTableSelected() {
         }
     }
 
+    private void clearPendingTables() {
+        for (int i = 0, n = pendingTables.size(); i < n; i++) {
+            pendingTables.getQuick(i).setListedPending(false);
+        }
+        pendingTables.setPos(0);
+    }
+
     private synchronized Throwable closeRemainingResources(Throwable terminalError) {
         if (closeCleanupStarted) {
             return terminalError;
@@ -4711,20 +4725,17 @@ private synchronized Throwable closeRemainingResources(Throwable terminalError)
 
     /**
      * Collects the batch's non-empty tables into {@link #flushTableNames} /
-     * {@link #flushTableBuffers} and returns how many there are. One walk of the key
-     * list and one hash probe per table, for the whole flush.
-     */
-    private int collectNonEmptyTables(ObjList<CharSequence> keys) {
-        flushTableNames.clear();
-        flushTableBuffers.clear();
-        for (int i = 0, n = keys.size(); i < n; i++) {
-            CharSequence tableName = keys.getQuick(i);
-            if (tableName == null) {
-                continue;
-            }
-            QwpTableBuffer tableBuffer = tableBuffers.get(tableName);
-            if (tableBuffer != null && tableBuffer.getRowCount() > 0) {
-                flushTableNames.add(tableName);
+     * {@link #flushTableBuffers} and returns how many there are. One walk of
+     * {@link #pendingTables}, in first-row order, for the whole flush; a listed
+     * table whose rows were since discarded is skipped.
+     */
+    private int collectNonEmptyTables() {
+        flushTableNames.setPos(0);
+        flushTableBuffers.setPos(0);
+        for (int i = 0, n = pendingTables.size(); i < n; i++) {
+            QwpTableBuffer tableBuffer = pendingTables.getQuick(i);
+            if (tableBuffer.getRowCount() > 0) {
+                flushTableNames.add(tableBuffer.getTableName());
                 flushTableBuffers.add(tableBuffer);
             }
         }
@@ -5187,9 +5198,9 @@ private void flushPendingRows(boolean deferCommit) {
         cachedTimestampColumn = null;
         cachedTimestampNanosColumn = null;
 
-        ObjList<CharSequence> keys = tableBuffers.keys();
-        int tableCount = collectNonEmptyTables(keys);
+        int tableCount = collectNonEmptyTables();
         if (tableCount == 0) {
+            clearPendingTables();
             pendingBytes = 0;
             currentTableBufferSnapshotBytes = currentTableBuffer == null
                     ? 0 : currentTableBuffer.getBufferedBytes();
@@ -5535,9 +5546,12 @@ private void resetTableBuffersAfterFlush() {
         for (int i = 0, n = flushTableBuffers.size(); i < n; i++) {
             flushTableBuffers.getQuick(i).reset();
         }
-        // Drop the references; the next flush re-collects.
-        flushTableNames.clear();
-        flushTableBuffers.clear();
+        // The next row lists its table again.
+        clearPendingTables();
+        // The next flush re-collects. setPos(0) rather than clear(), which fills the
+        // whole backing array.
+        flushTableNames.setPos(0);
+        flushTableBuffers.setPos(0);
         currentBatchMaxSymbolId = -1;
         pendingBytes = 0;
         currentTableBufferSnapshotBytes = 0;
@@ -7150,6 +7164,10 @@ private void sendRow() {
             firstPendingRowTimeNanos = nanoClock.getAsLong();
         }
         pendingRowCount++;
+        if (!currentTableBuffer.isListedPending()) {
+            currentTableBuffer.setListedPending(true);
+            pendingTables.add(currentTableBuffer);
+        }
 
         // Advance pendingBytes by the bytes the just-committed row added to
         // the current table. nextRow() accumulated the total during its