From 29c86fef27dc5586fece8574afa8dda8e8d2d460 Mon Sep 17 00:00:00 2001 From: Sergei Minaev <5072859+jovfer@users.noreply.github.com> Date: Thu, 5 Nov 2026 13:25:55 +0000 Subject: [PATCH 1/4] perf(qwp): drop the unread first-row stamp of the microbatch buffer MicrobatchBuffer read System.nanoTime() on the first row of every microbatch to serve getAgeNanos(), which nothing calls. Drop the field, the stamp and the getter, together with the test of the getter. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DjhXUiBaimJEwHoXYv4CeN --- .../cutlass/qwp/client/MicrobatchBuffer.java | 18 +----------------- .../qwp/client/MicrobatchBufferTest.java | 18 ------------------ 2 files changed, 1 insertion(+), 35 deletions(-) diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/MicrobatchBuffer.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/MicrobatchBuffer.java index 43a563228..6d45f55c8 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/MicrobatchBuffer.java +++ b/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; diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/MicrobatchBufferTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/MicrobatchBufferTest.java index e97e91b70..823cff856 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/MicrobatchBufferTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/MicrobatchBufferTest.java @@ -311,24 +311,6 @@ public void testEnsureCapacityNoGrowth() throws Exception { }); } - @Test - public void testFirstRowTimeIsRecorded() throws Exception { - assertMemoryLeak(() -> { - try (MicrobatchBuffer buffer = new MicrobatchBuffer(1024)) { - Assert.assertEquals(0, buffer.getAgeNanos()); - - buffer.incrementRowCount(); - long age1 = buffer.getAgeNanos(); - Assert.assertTrue(age1 >= 0); - - Os.sleep(10); - - long age2 = buffer.getAgeNanos(); - Assert.assertTrue(age2 > age1); - } - }); - } - @Test public void testFullStateLifecycle() throws Exception { assertMemoryLeak(() -> { From 60e37fd1d7c4089fc58a2822e4b44ad0095b1db8 Mon Sep 17 00:00:00 2001 From: Sergei Minaev <5072859+jovfer@users.noreply.github.com> Date: Thu, 5 Nov 2026 13:27:26 +0000 Subject: [PATCH 2/4] perf(qwp): encode table and column names once per schema The table block header re-encoded the table name and every column name from Java strings on each frame. Cache the wire form of the table name (built in the constructor) and of the column definitions (rebuilt with the column definitions cache), and copy the two byte arrays with the row-count varint between them. The frame bytes are unchanged. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DjhXUiBaimJEwHoXYv4CeN --- .../cutlass/qwp/client/QwpColumnWriter.java | 22 ++--- .../cutlass/qwp/protocol/QwpTableBuffer.java | 67 +++++++++++++ .../qwp/client/QwpWebSocketEncoderTest.java | 28 ++++++ .../qwp/protocol/QwpTableBufferTest.java | 95 +++++++++++++++++++ 4 files changed, 201 insertions(+), 11 deletions(-) diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpColumnWriter.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpColumnWriter.java index f8cc669fb..b2e577888 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpColumnWriter.java +++ b/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); diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java b/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java index 1d8cc3966..b0a1e51c9 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java @@ -47,6 +47,7 @@ import io.questdb.client.std.str.StringSink; import io.questdb.client.std.str.Utf8s; +import java.nio.charset.StandardCharsets; import java.util.Arrays; import static io.questdb.client.cutlass.qwp.protocol.QwpConstants.*; @@ -63,12 +64,14 @@ public class QwpTableBuffer implements QuietCloseable { private static final int MAX_COLUMN_NAME_LENGTH = 127; private final LowerCaseCharSequenceIntHashMap columnNameToIndex; private final ObjList columns; + private final byte[] encodedTableName; private final QwpWebSocketSender sender; private final String tableName; private QwpColumnDef[] cachedColumnDefs; private int columnAccessCursor; // tracks expected next column index private boolean columnDefsCacheValid; private int committedColumnCount; // columns that existed at last nextRow() + private byte[] encodedColumnDefs; private ColumnBuffer[] fastColumns; // plain array for O(1) sequential access private int inProgressColumnCount; private int rowCount; @@ -85,6 +88,7 @@ public QwpTableBuffer(String tableName) { */ public QwpTableBuffer(String tableName, QwpWebSocketSender sender) { this.tableName = tableName; + this.encodedTableName = encodeName(tableName); this.sender = sender; this.columns = new ObjList<>(); this.columnNameToIndex = new LowerCaseCharSequenceIntHashMap(); @@ -175,11 +179,30 @@ public QwpColumnDef[] getColumnDefs() { ColumnBuffer col = columns.get(i); cachedColumnDefs[i] = new QwpColumnDef(col.name, col.type); } + encodedColumnDefs = encodeColumnDefs(cachedColumnDefs); columnDefsCacheValid = true; } return cachedColumnDefs; } + /** + * Returns the wire form of the column definitions: varint column count, then per + * column varint name length, UTF-8 name and the type code. Rebuilt together with + * {@link #getColumnDefs()}; do not modify the returned array. + */ + public byte[] getEncodedColumnDefs() { + getColumnDefs(); + return encodedColumnDefs; + } + + /** + * Returns the wire form of the table name: varint UTF-8 length, then the UTF-8 + * bytes. Built once in the constructor; do not modify the returned array. + */ + public byte[] getEncodedTableName() { + return encodedTableName; + } + /** * Returns an existing column with the given name and type, or {@code null} if absent. *

@@ -395,6 +418,50 @@ private static void assertColumnType(CharSequence name, byte type, ColumnBuffer } } + private static byte[] encodeColumnDefs(QwpColumnDef[] defs) { + byte[][] names = new byte[defs.length][]; + int size = varintSize(defs.length); + for (int i = 0; i < defs.length; i++) { + names[i] = defs[i].getName().getBytes(StandardCharsets.UTF_8); + size += varintSize(names[i].length) + names[i].length + 1; + } + byte[] out = new byte[size]; + int pos = putVarint(out, 0, defs.length); + for (int i = 0; i < defs.length; i++) { + pos = putVarint(out, pos, names[i].length); + System.arraycopy(names[i], 0, out, pos, names[i].length); + pos += names[i].length; + out[pos++] = defs[i].getWireTypeCode(); + } + return out; + } + + private static byte[] encodeName(String name) { + byte[] utf8 = name.getBytes(StandardCharsets.UTF_8); + byte[] out = new byte[varintSize(utf8.length) + utf8.length]; + int pos = putVarint(out, 0, utf8.length); + System.arraycopy(utf8, 0, out, pos, utf8.length); + return out; + } + + private static int putVarint(byte[] dst, int pos, long value) { + while (value > 0x7F) { + dst[pos++] = (byte) ((value & 0x7F) | 0x80); + value >>>= 7; + } + dst[pos++] = (byte) value; + return pos; + } + + private static int varintSize(long value) { + int size = 1; + while (value > 0x7F) { + size++; + value >>>= 7; + } + return size; + } + private ColumnBuffer createColumn(CharSequence name, byte type, boolean useNullBitmap) { if (columns.size() >= MAX_COLUMNS_PER_TABLE) { throw new LineSenderException("column count exceeds maximum: " + (columns.size() + 1) + diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketEncoderTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketEncoderTest.java index 44f711a3d..9767eaf9c 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketEncoderTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketEncoderTest.java @@ -1423,6 +1423,34 @@ public void testSplitMessageCopiesStagedTableBodies() throws Exception { }); } + @Test + public void testTableHeaderMatchesTheWireLayout() throws Exception { + assertMemoryLeak(() -> { + try (QwpWebSocketEncoder encoder = new QwpWebSocketEncoder(); + QwpTableBuffer buffer = new QwpTableBuffer("trades")) { + for (int i = 0; i < 300; i++) { + buffer.getOrCreateColumn("price", TYPE_DOUBLE, false).addDouble(i); + buffer.getOrCreateColumn("qty", TYPE_LONG, false).addLong(i); + buffer.nextRow(); + } + + encoder.encode(buffer); + + byte[] expected = { + 6, 't', 'r', 'a', 'd', 'e', 's', + (byte) 0xAC, 0x02, + 2, + 5, 'p', 'r', 'i', 'c', 'e', TYPE_DOUBLE, + 3, 'q', 't', 'y', TYPE_LONG + }; + long ptr = encoder.getBuffer().getBufferPtr(); + for (int i = 0; i < expected.length; i++) { + Assert.assertEquals("byte " + i, expected[i], Unsafe.getUnsafe().getByte(ptr + 12 + i)); + } + } + }); + } + @Test public void testVersionByteInHeader() throws Exception { assertMemoryLeak(() -> { diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpTableBufferTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpTableBufferTest.java index ab14c18d8..f591709c1 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpTableBufferTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpTableBufferTest.java @@ -902,6 +902,101 @@ public void testDoubleArrayWrapperVaryingDimensionality() throws Exception { }); } + @Test + public void testEncodedColumnDefsDropAColumnOfACancelledRow() throws Exception { + assertMemoryLeak(() -> { + try (QwpTableBuffer table = new QwpTableBuffer("t")) { + table.getOrCreateColumn("a", QwpConstants.TYPE_LONG, false).addLong(1); + table.nextRow(); + + table.getOrCreateColumn("c", QwpConstants.TYPE_DOUBLE, false).addDouble(1.5); + assertArrayEquals(new byte[]{ + 2, + 1, 'a', QwpConstants.TYPE_LONG, + 1, 'c', QwpConstants.TYPE_DOUBLE + }, table.getEncodedColumnDefs()); + + table.cancelCurrentRow(); + table.rollbackUncommittedColumns(); + assertArrayEquals(new byte[]{ + 1, + 1, 'a', QwpConstants.TYPE_LONG + }, table.getEncodedColumnDefs()); + + table.getOrCreateColumn("d", QwpConstants.TYPE_VARCHAR, true).addString("x"); + assertArrayEquals(new byte[]{ + 2, + 1, 'a', QwpConstants.TYPE_LONG, + 1, 'd', QwpConstants.TYPE_VARCHAR + }, table.getEncodedColumnDefs()); + } + }); + } + + @Test + public void testEncodedColumnDefsTrackTheSchema() throws Exception { + assertMemoryLeak(() -> { + try (QwpTableBuffer table = new QwpTableBuffer("t")) { + table.getOrCreateColumn("a", QwpConstants.TYPE_LONG, false).addLong(1); + table.getOrCreateColumn("b", QwpConstants.TYPE_VARCHAR, true).addString("x"); + table.nextRow(); + + byte[] before = table.getEncodedColumnDefs(); + assertArrayEquals(new byte[]{ + 2, + 1, 'a', QwpConstants.TYPE_LONG, + 1, 'b', QwpConstants.TYPE_VARCHAR + }, before); + + table.getOrCreateColumn("c", QwpConstants.TYPE_DOUBLE, false).addDouble(1.5); + assertArrayEquals(new byte[]{ + 3, + 1, 'a', QwpConstants.TYPE_LONG, + 1, 'b', QwpConstants.TYPE_VARCHAR, + 1, 'c', QwpConstants.TYPE_DOUBLE + }, table.getEncodedColumnDefs()); + assertArrayEquals(new byte[]{ + 2, + 1, 'a', QwpConstants.TYPE_LONG, + 1, 'b', QwpConstants.TYPE_VARCHAR + }, before); + } + }); + } + + @Test + public void testEncodedColumnNameLengthCountsUtf8Bytes() throws Exception { + assertMemoryLeak(() -> { + StringBuilder name = new StringBuilder(); + for (int i = 0; i < 100; i++) { + name.append('ö'); + } + try (QwpTableBuffer table = new QwpTableBuffer("t")) { + table.getOrCreateColumn(name.toString(), QwpConstants.TYPE_LONG, false).addLong(1); + table.nextRow(); + + byte[] encoded = table.getEncodedColumnDefs(); + assertEquals(1 + 2 + 200 + 1, encoded.length); + assertEquals(1, encoded[0]); + assertEquals((byte) 0xC8, encoded[1]); + assertEquals(0x01, encoded[2]); + assertEquals(QwpConstants.TYPE_LONG, encoded[encoded.length - 1]); + } + }); + } + + @Test + public void testEncodedTableNameIsUtf8LengthThenBytes() throws Exception { + assertMemoryLeak(() -> { + try (QwpTableBuffer table = new QwpTableBuffer("tëst")) { + assertArrayEquals( + new byte[]{5, 't', (byte) 0xC3, (byte) 0xAB, 's', 't'}, + table.getEncodedTableName() + ); + } + }); + } + @Test public void testGetExistingColumnReturnsNullWithoutCreatingColumn() throws Exception { assertMemoryLeak(() -> { From f90ed8e41666d18c3fee4d093f865655f97947fd Mon Sep 17 00:00:00 2001 From: Sergei Minaev <5072859+jovfer@users.noreply.github.com> Date: Thu, 5 Nov 2026 13:29:46 +0000 Subject: [PATCH 3/4] perf(qwp): reset a column buffer in proportion to what it held ColumnBuffer.reset() zeroed the null bitmap and cleared the local symbol dictionary on every flush. Zero the bitmap only after a batch that held nulls, and clear the dictionary only when it has entries. Tests pin the bitmap and dictionary behaviour across resets. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DjhXUiBaimJEwHoXYv4CeN --- .../cutlass/qwp/protocol/QwpTableBuffer.java | 17 +- .../qwp/protocol/QwpTableBufferTest.java | 168 ++++++++++++++++++ 2 files changed, 181 insertions(+), 4 deletions(-) diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java b/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java index b0a1e51c9..c3ff76839 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java @@ -1532,9 +1532,18 @@ public boolean isNull(int index) { return (Unsafe.getUnsafe().getLong(longAddr) & (1L << bitIndex)) != 0; } + /** + * Resets the column for the next batch. Work is proportional to what the column held: + * a bit is set in the null bitmap only while {@code hasNulls} is true ({@code markNull} + * is the only setter; {@code truncateTo}, the retain paths and + * {@code ensureNullBitmapCapacity} clear what they shrink or grow), so the bitmap is + * zeroed only after a batch with nulls. The local symbol dictionary is populated only + * without a sender (global ids leave it empty), so it is cleared only when non-empty. + */ public void reset() { size = 0; valueCount = 0; + boolean hadNulls = hasNulls; hasNulls = false; if (dataBuffer != null) { dataBuffer.truncate(); @@ -1549,10 +1558,10 @@ public void reset() { if (stringData != null) { stringData.truncate(); } - if (nullBufPtr != 0) { + if (nullBufPtr != 0 && hadNulls) { Vect.memset(nullBufPtr, (long) nullBufCapRows >>> 3, 0); } - if (symbolDict != null) { + if (symbolDict != null && symbolList.size() > 0) { symbolDict.clear(); symbolList.clear(); } @@ -1683,7 +1692,7 @@ public void truncateTo(int newSize) { geohashPrecision = -1; maxGlobalSymbolId = -1; storeGlobalSymbolIdsOnly = false; - if (symbolDict != null) { + if (symbolDict != null && symbolList.size() > 0) { symbolDict.clear(); symbolList.clear(); } @@ -1926,7 +1935,7 @@ private void resetEmptyMetadata() { geohashPrecision = -1; maxGlobalSymbolId = -1; storeGlobalSymbolIdsOnly = false; - if (symbolDict != null) { + if (symbolDict != null && symbolList.size() > 0) { symbolDict.clear(); symbolList.clear(); } diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpTableBufferTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpTableBufferTest.java index f591709c1..9c3f5861e 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpTableBufferTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpTableBufferTest.java @@ -28,6 +28,7 @@ import io.questdb.client.cutlass.line.LineSenderException; import io.questdb.client.cutlass.line.array.DoubleArray; import io.questdb.client.cutlass.line.array.LongArray; +import io.questdb.client.cutlass.qwp.client.QwpWebSocketEncoder; import io.questdb.client.cutlass.qwp.protocol.OffHeapAppendMemory; import io.questdb.client.cutlass.qwp.protocol.QwpConstants; import io.questdb.client.cutlass.qwp.protocol.QwpTableBuffer; @@ -1483,6 +1484,137 @@ public void testNonAsciiColumnNameCaseInsensitive() throws Exception { }); } + @Test + public void testResetAfterACancelledNullRowLeavesNoStaleBit() throws Exception { + assertMemoryLeak(() -> { + try (QwpTableBuffer table = new QwpTableBuffer("t")) { + QwpTableBuffer.ColumnBuffer col = table.getOrCreateColumn("v", QwpConstants.TYPE_LONG, true); + col.addLong(1L); + table.nextRow(); + col.addNull(); + table.cancelCurrentRow(); + table.reset(); + + col.addLong(1L); + table.nextRow(); + col.addLong(2L); + table.nextRow(); + col.addNull(); + table.nextRow(); + + assertFalse(col.isNull(0)); + assertFalse(col.isNull(1)); + assertTrue(col.isNull(2)); + int[] header = readNullHeader(table); + assertEquals(1, header[0]); + assertEquals(0b100, header[1]); + } + }); + } + + @Test + public void testResetAfterNullsBeyondTheInitialBitmapCapacityClearsEveryBit() throws Exception { + assertMemoryLeak(() -> { + try (QwpTableBuffer table = new QwpTableBuffer("t")) { + QwpTableBuffer.ColumnBuffer col = table.getOrCreateColumn("v", QwpConstants.TYPE_LONG, true); + for (int i = 0; i < 100; i++) { + if (i == 90) { + col.addNull(); + } else { + col.addLong(i); + } + table.nextRow(); + } + assertTrue(col.isNull(90)); + table.reset(); + + for (int i = 0; i < 100; i++) { + col.addLong(i); + table.nextRow(); + } + + assertFalse(col.isNull(90)); + assertEquals(100, col.getValueCount()); + assertEquals(0, readNullHeader(table)[0]); + } + }); + } + + @Test + public void testResetAfterNullsClearsTheBitmapForTheNextBatch() throws Exception { + assertMemoryLeak(() -> { + try (QwpTableBuffer table = new QwpTableBuffer("t")) { + QwpTableBuffer.ColumnBuffer col = table.getOrCreateColumn("v", QwpConstants.TYPE_LONG, true); + col.addNull(); + table.nextRow(); + col.addLong(1L); + table.nextRow(); + assertTrue(col.isNull(0)); + table.reset(); + + col.addLong(1L); + table.nextRow(); + col.addLong(2L); + table.nextRow(); + col.addNull(); + table.nextRow(); + + assertFalse(col.isNull(0)); + assertFalse(col.isNull(1)); + assertTrue(col.isNull(2)); + int[] header = readNullHeader(table); + assertEquals(1, header[0]); + assertEquals(0b100, header[1]); + } + }); + } + + @Test + public void testResetKeepsAnEmptyLocalDictionaryAndClearsAPopulatedOne() throws Exception { + assertMemoryLeak(() -> { + try (QwpTableBuffer table = new QwpTableBuffer("t")) { + QwpTableBuffer.ColumnBuffer col = table.getOrCreateColumn("sym", QwpConstants.TYPE_SYMBOL, true); + col.addSymbolWithGlobalId("a", 7); + table.nextRow(); + table.reset(); + assertEquals(0, col.getSymbolDictionarySize()); + assertEquals(-1, col.getMaxGlobalSymbolId()); + + col.addSymbol("x"); + table.nextRow(); + assertEquals(1, col.getSymbolDictionarySize()); + table.reset(); + assertEquals(0, col.getSymbolDictionarySize()); + assertFalse(col.hasSymbol("x")); + } + }); + } + + @Test + public void testResetWithoutNullsThenANullInTheNextBatch() throws Exception { + assertMemoryLeak(() -> { + try (QwpTableBuffer table = new QwpTableBuffer("t")) { + QwpTableBuffer.ColumnBuffer col = table.getOrCreateColumn("v", QwpConstants.TYPE_LONG, true); + col.addLong(1L); + table.nextRow(); + col.addLong(2L); + table.nextRow(); + table.reset(); + + col.addLong(1L); + table.nextRow(); + col.addNull(); + table.nextRow(); + + assertFalse(col.isNull(0)); + assertTrue(col.isNull(1)); + int[] header = readNullHeader(table); + assertEquals(1, header[0]); + assertEquals(0b10, header[1]); + } + }); + } + @Test public void testRetainInProgressRowFastClearsUnstagedNullableColumn() throws Exception { assertMemoryLeak(() -> { @@ -1641,4 +1773,40 @@ private static long[] readLongArraysLikeEncoder(QwpTableBuffer.ColumnBuffer col) } return result; } + + /** + * Encodes a single-column table and returns {null-header marker, first bitmap byte + * or 0 when the marker is 0}. + */ + private static int[] readNullHeader(QwpTableBuffer table) { + try (QwpWebSocketEncoder encoder = new QwpWebSocketEncoder()) { + encoder.encode(table); + long ptr = encoder.getBuffer().getBufferPtr() + 12; + long[] pos = {ptr}; + long nameLen = readVarint(pos); + pos[0] += nameLen; + readVarint(pos); // row count + long colCount = readVarint(pos); + for (long c = 0; c < colCount; c++) { + long len = readVarint(pos); + pos[0] += len + 1; // name + type byte + } + int marker = Unsafe.getUnsafe().getByte(pos[0]) & 0xFF; + int bitmap = marker == 1 ? Unsafe.getUnsafe().getByte(pos[0] + 1) & 0xFF : 0; + return new int[]{marker, bitmap}; + } + } + + private static long readVarint(long[] pos) { + long value = 0; + int shift = 0; + while (true) { + int b = Unsafe.getUnsafe().getByte(pos[0]++) & 0xFF; + value |= (long) (b & 0x7F) << shift; + if ((b & 0x80) == 0) { + return value; + } + shift += 7; + } + } } From ba25bf1dc24eb8bdeb51c0f927e3bf3ab7a8dd1e Mon Sep 17 00:00:00 2001 From: Sergei Minaev <5072859+jovfer@users.noreply.github.com> Date: Thu, 5 Nov 2026 13:33:22 +0000 Subject: [PATCH 4/4] perf(qwp): flush from the list of tables that received rows Every flush walked every table the sender had ever seen, with a hash probe per key, to find the ones with rows, and emptied the two flush lists with ObjList.clear(), which fills the whole backing array. sendRow() now lists a table in pendingTables when it receives its first row since the last successful flush; a flag on QwpTableBuffer keeps each table listed once. collectNonEmptyTables() walks that list, and the flush lists are emptied with setPos(0). The list and the flags are cleared together only when a flush succeeds or finds nothing to ship, so a flush that throws before publishing leaves the list for the retry. reset() leaves the list alone and the collect skips listed tables without rows; close() empties it. Tables in a frame now ship in the order they first received a row since the last flush, instead of the order the sender first saw them. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DjhXUiBaimJEwHoXYv4CeN --- .../qwp/client/QwpWebSocketSender.java | 56 ++-- .../cutlass/qwp/protocol/QwpTableBuffer.java | 17 ++ .../qwp/client/QwpWebSocketSenderTest.java | 242 ++++++++++++++++++ .../qwp/client/SelfSufficientFramesTest.java | 4 +- 4 files changed, 298 insertions(+), 21 deletions(-) diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java index 4c8e731aa..1fc434840 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java +++ b/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 flushTableBuffers = new ObjList<>(); private final ObjList 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 pendingTables = new ObjList<>(); private final IntList splitFrameBodyBytes = new IntList(); private final CharSequenceObjHashMap 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 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 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 diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java b/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java index c3ff76839..02f95b914 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java @@ -74,6 +74,9 @@ public class QwpTableBuffer implements QuietCloseable { private byte[] encodedColumnDefs; private ColumnBuffer[] fastColumns; // plain array for O(1) sequential access private int inProgressColumnCount; + // Set while the sender's pending-tables list holds this buffer, so the + // buffer is listed once. Producer thread only. + private boolean listedPending; private int rowCount; public QwpTableBuffer(String tableName) { @@ -287,6 +290,13 @@ public boolean hasInProgressRow() { return inProgressColumnCount > 0; } + /** + * Returns true while the sender's pending-tables list holds this buffer. + */ + public boolean isListedPending() { + return listedPending; + } + /** * Advances to the next row. *

@@ -409,6 +419,13 @@ public void rollbackUncommittedColumns() { rebuildColumnAccessStructures(); } + /** + * Marks whether the sender's pending-tables list holds this buffer. + */ + public void setListedPending(boolean listedPending) { + this.listedPending = listedPending; + } + private static void assertColumnType(CharSequence name, byte type, ColumnBuffer column) { if (column.type != type) { throw new LineSenderException( diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java index d90fd478a..0df432cc7 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java @@ -24,10 +24,12 @@ package io.questdb.client.test.cutlass.qwp.client; +import io.questdb.client.Sender; import io.questdb.client.cutlass.qwp.client.DurableAckTiers; import io.questdb.client.cutlass.line.LineSenderException; import io.questdb.client.cutlass.line.array.DoubleArray; import io.questdb.client.cutlass.line.array.LongArray; +import io.questdb.client.cutlass.qwp.client.BatchTooLargeForCapException; import io.questdb.client.cutlass.qwp.client.MicrobatchBuffer; import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender; import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine; @@ -41,9 +43,13 @@ import org.junit.Assert; import org.junit.Test; +import java.io.IOException; import java.lang.reflect.Field; import java.time.Instant; import java.time.temporal.ChronoUnit; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; import java.util.concurrent.TimeUnit; import static io.questdb.client.test.tools.TestUtils.assertMemoryLeak; @@ -283,6 +289,82 @@ public void testCancelRowNoOpWithoutTable() throws Exception { }); } + @Test + public void testCancelledRowDoesNotShipItsTable() throws Exception { + assertMemoryLeak(() -> { + FrameCapture capture = new FrameCapture(); + try (TestWebSocketServer server = new TestWebSocketServer(capture)) { + server.start(); + Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS)); + try (Sender sender = Sender.fromConfig("ws::addr=localhost:" + server.getPort() + + ";auto_flush_interval=60000;")) { + sender.table("alpha").longColumn("v", 1).atNow(); + sender.table("bravo").longColumn("v", 2); + sender.cancelRow(); + sender.flush(); + capture.awaitFrames(1, 5_000); + Assert.assertEquals(1, capture.tableCount(0)); + byte[] frame0 = capture.frame(0); + Assert.assertTrue("frame 0 carries alpha", indexOf(frame0, "alpha") >= 0); + Assert.assertEquals("frame 0 carries no bravo", -1, indexOf(frame0, "bravo")); + } + } + }); + } + + @Test + public void testCapRejectedFlushThenResetNeverShipsTheDroppedTable() throws Exception { + assertMemoryLeak(() -> { + FrameCapture capture = new FrameCapture(); + try (TestWebSocketServer server = new TestWebSocketServer(capture)) { + server.setAdvertisedMaxBatchSize(4 * 1024); + server.start(); + Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS)); + // no row or byte trigger: the whole batch reaches the explicit flush(), whose split + // pre-flight rejects it before publishing anything + try (Sender sender = Sender.fromConfig("ws::addr=localhost:" + server.getPort() + + ";auto_flush_rows=off;auto_flush_bytes=off;auto_flush_interval=60000;")) { + // warm-up, so the cap is known + sender.table("w").longColumn("v", 1).atNow(); + sender.flush(); + capture.awaitFrames(1, 5_000); + + sender.table("alpha").longColumn("v", 2).atNow(); + sender.table("bravo").longColumn("v", 3).atNow(); + // three 1.5 KB rows take charlie_dropped's block over the 4 KiB cap + String wide = wideValue(1500); + for (int i = 0; i < 3; i++) { + sender.table("charlie_dropped").stringColumn("s", wide).atNow(); + } + try { + sender.flush(); + Assert.fail("the batch exceeds the 4 KiB cap"); + } catch (BatchTooLargeForCapException expected) { + // thrown before anything was published + } + + // the three tables stay listed after reset(); only the ones that get rows again ship + sender.reset(); + sender.table("alpha").longColumn("v", 4).atNow(); + sender.table("bravo").longColumn("v", 5).atNow(); + sender.flush(); + capture.awaitFrames(2, 5_000); + Assert.assertEquals(2, capture.tableCount(1)); + byte[] frame1 = capture.frame(1); + int alphaAt = indexOf(frame1, "alpha"); + int bravoAt = indexOf(frame1, "bravo"); + Assert.assertTrue("frame 1 carries alpha", alphaAt >= 0); + Assert.assertTrue("frame 1 carries bravo", bravoAt >= 0); + Assert.assertTrue("tables ship in first-row order", alphaAt < bravoAt); + for (int i = 0, n = capture.frameCount(); i < n; i++) { + Assert.assertEquals("frame " + i + " must not carry the dropped table", + -1, indexOf(capture.frame(i), "charlie_dropped")); + } + } + } + }); + } + @Test public void testCloseIdemponent() throws Exception { assertMemoryLeak(() -> { @@ -333,6 +415,39 @@ public void testDoubleColumnAfterCloseThrows() throws Exception { }); } + @Test + public void testFlushAfterResetShipsOnlyTablesWithNewRows() throws Exception { + assertMemoryLeak(() -> { + FrameCapture capture = new FrameCapture(); + try (TestWebSocketServer server = new TestWebSocketServer(capture)) { + server.start(); + Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS)); + try (Sender sender = Sender.fromConfig("ws::addr=localhost:" + server.getPort() + + ";auto_flush_interval=60000;")) { + sender.table("alpha").longColumn("v", 1).atNow(); + sender.table("bravo").longColumn("v", 2).atNow(); + // both tables stay listed after reset(); the flush skips the one without rows + sender.reset(); + sender.table("bravo").longColumn("v", 3).atNow(); + sender.flush(); + capture.awaitFrames(1, 5_000); + Assert.assertEquals("only bravo has rows", 1, capture.tableCount(0)); + byte[] frame0 = capture.frame(0); + Assert.assertTrue("frame 0 carries bravo", indexOf(frame0, "bravo") >= 0); + Assert.assertEquals("frame 0 carries no alpha", -1, indexOf(frame0, "alpha")); + + sender.table("alpha").longColumn("v", 4).atNow(); + sender.flush(); + capture.awaitFrames(2, 5_000); + Assert.assertEquals("only alpha has rows", 1, capture.tableCount(1)); + byte[] frame1 = capture.frame(1); + Assert.assertTrue("frame 1 carries alpha", indexOf(frame1, "alpha") >= 0); + Assert.assertEquals("frame 1 carries no bravo", -1, indexOf(frame1, "bravo")); + } + } + }); + } + @Test public void testFlushAppendFailureDoesNotLeaveMicrobatchBufferInUse() throws Exception { assertMemoryLeak(() -> { @@ -378,6 +493,56 @@ public void testFlushAppendFailureDoesNotLeaveMicrobatchBufferInUse() throws Exc }); } + @Test + public void testFlushShipsOnlyTablesWithPendingRows() throws Exception { + assertMemoryLeak(() -> { + FrameCapture capture = new FrameCapture(); + try (TestWebSocketServer server = new TestWebSocketServer(capture)) { + server.start(); + Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS)); + try (Sender sender = Sender.fromConfig("ws::addr=localhost:" + server.getPort() + + ";auto_flush_interval=60000;")) { + sender.table("alpha").longColumn("v", 1).atNow(); + sender.table("bravo").longColumn("v", 2).atNow(); + sender.table("alpha").longColumn("v", 3).atNow(); + sender.flush(); + capture.awaitFrames(1, 5_000); + Assert.assertEquals(2, capture.tableCount(0)); + byte[] frame0 = capture.frame(0); + int alphaAt = indexOf(frame0, "alpha"); + int bravoAt = indexOf(frame0, "bravo"); + Assert.assertTrue("frame 0 carries alpha", alphaAt >= 0); + Assert.assertTrue("frame 0 carries bravo", bravoAt >= 0); + Assert.assertTrue("tables ship in first-row order", alphaAt < bravoAt); + + sender.table("charlie").longColumn("v", 4).atNow(); + sender.flush(); + capture.awaitFrames(2, 5_000); + Assert.assertEquals("only charlie has rows", 1, capture.tableCount(1)); + + sender.table("bravo").longColumn("v", 5).atNow(); + sender.flush(); + capture.awaitFrames(3, 5_000); + Assert.assertEquals("only bravo has rows", 1, capture.tableCount(2)); + byte[] frame2 = capture.frame(2); + Assert.assertTrue("frame 2 carries bravo", indexOf(frame2, "bravo") >= 0); + Assert.assertEquals("frame 2 carries no alpha", -1, indexOf(frame2, "alpha")); + Assert.assertEquals("frame 2 carries no charlie", -1, indexOf(frame2, "charlie")); + + // bravo was created after alpha but gets its row first in this batch + sender.table("bravo").longColumn("v", 6).atNow(); + sender.table("alpha").longColumn("v", 7).atNow(); + sender.flush(); + capture.awaitFrames(4, 5_000); + Assert.assertEquals(2, capture.tableCount(3)); + byte[] frame3 = capture.frame(3); + Assert.assertTrue("tables ship in first-row order, not creation order", + indexOf(frame3, "bravo") < indexOf(frame3, "alpha")); + } + } + }); + } + @Test public void testGeoHashColumnLongAfterCloseThrows() throws Exception { assertMemoryLeak(() -> { @@ -864,6 +1029,26 @@ private static MicrobatchBuffer getMicrobatchBuffer(QwpWebSocketSender sender, S return (MicrobatchBuffer) field.get(sender); } + private static int indexOf(byte[] frame, String ascii) { + int n = ascii.length(); + for (int i = 0, last = frame.length - n; i <= last; i++) { + int j = 0; + while (j < n && frame[i + j] == (byte) ascii.charAt(j)) { + j++; + } + if (j == n) { + return i; + } + } + return -1; + } + + private static String wideValue(int length) { + char[] chars = new char[length]; + Arrays.fill(chars, 'x'); + return new String(chars); + } + /** * Creates a sender without connecting. * For unit tests that don't need actual connectivity. @@ -872,4 +1057,61 @@ private QwpWebSocketSender createUnconnectedSender() { return QwpWebSocketSender.createForTesting("localhost", 9000); } + /** + * Records every QWP frame the server receives, in arrival order, and acks each + * binary message up to the last frame it carried. + */ + private static final class FrameCapture implements TestWebSocketServer.WebSocketServerHandler { + private static final int HEADER_SIZE = 12; + private static final int OFFSET_PAYLOAD_LENGTH = 8; + private static final int OFFSET_TABLE_COUNT = 6; + private final List frames = new ArrayList<>(); + private long nextSeq; + + void awaitFrames(int atLeast, long timeoutMillis) throws InterruptedException { + long deadline = System.currentTimeMillis() + timeoutMillis; + while (frameCount() < atLeast && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + } + Assert.assertTrue("expected at least " + atLeast + " frame(s) within " + timeoutMillis + + " ms, got " + frameCount(), frameCount() >= atLeast); + } + + synchronized byte[] frame(int index) { + return frames.get(index); + } + + synchronized int frameCount() { + return frames.size(); + } + + @Override + public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) { + int pos = 0; + int framesInMessage = 0; + while (pos + HEADER_SIZE <= data.length) { + int payloadLength = (data[pos + OFFSET_PAYLOAD_LENGTH] & 0xFF) + | ((data[pos + OFFSET_PAYLOAD_LENGTH + 1] & 0xFF) << 8) + | ((data[pos + OFFSET_PAYLOAD_LENGTH + 2] & 0xFF) << 16) + | ((data[pos + OFFSET_PAYLOAD_LENGTH + 3] & 0xFF) << 24); + if (payloadLength < 0 || pos + HEADER_SIZE + payloadLength > data.length) { + break; + } + frames.add(Arrays.copyOfRange(data, pos, pos + HEADER_SIZE + payloadLength)); + framesInMessage++; + pos += HEADER_SIZE + payloadLength; + } + nextSeq += Math.max(1, framesInMessage); + try { + client.sendBinary(CloseDrainTest.buildAck(nextSeq - 1)); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + synchronized int tableCount(int index) { + byte[] frame = frames.get(index); + return (frame[OFFSET_TABLE_COUNT] & 0xFF) | ((frame[OFFSET_TABLE_COUNT + 1] & 0xFF) << 8); + } + } } diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SelfSufficientFramesTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SelfSufficientFramesTest.java index 1d1aaebfa..44ceb6d42 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SelfSufficientFramesTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SelfSufficientFramesTest.java @@ -588,8 +588,8 @@ public void testSplitPreflightAdvancesBaselineSoLaterFramesArentSizedWithTheDelt // the first split frame) has a tiny body, so delta + body1 fits the cap. The // second table has a big body: body2 alone fits the cap, but delta + body2 does // NOT. The real code splits and ships both frames; the un-advanced pre-flight - // would size the second frame WITH the delta and throw. Table order is - // insertion order (CharSequenceObjHashMap.keys()), so t1 is the delta frame. + // would size the second frame WITH the delta and throw. Tables ship in the + // order they first received a row since the last flush, so t1 is the delta frame. assertMemoryLeak(() -> { CapturingHandler handler = new CapturingHandler(); try (TestWebSocketServer server = new TestWebSocketServer(handler)) {