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 43a56322..6d45f55c 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/main/java/io/questdb/client/cutlass/qwp/client/QwpColumnWriter.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpColumnWriter.java index f8cc669f..b2e57788 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/client/QwpWebSocketSender.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java index 4c8e731a..1fc43484 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 1d8cc396..02f95b91 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,14 +64,19 @@ 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; + // 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) { @@ -85,6 +91,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 +182,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. *

@@ -264,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. *

@@ -386,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( @@ -395,6 +435,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) + @@ -1465,9 +1549,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(); @@ -1482,10 +1575,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(); } @@ -1616,7 +1709,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(); } @@ -1859,7 +1952,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/client/MicrobatchBufferTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/MicrobatchBufferTest.java index e97e91b7..823cff85 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(() -> { 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 44f711a3..9767eaf9 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/client/QwpWebSocketSenderTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java index d90fd478..0df432cc 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 1d1aaebf..44ceb6d4 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)) { 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 ab14c18d..9c3f5861 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; @@ -902,6 +903,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(() -> { @@ -1388,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(() -> { @@ -1546,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; + } + } }