* It is essential to ensure the maximum buffer capacity is sufficient to accommodate all locally buffered data. *
- * This method should only be used when communicating via the HTTP protocol. Calling this method is illegal when
- * communicating over the TCP protocol.
+ * This method is supported for the HTTP and WebSocket (QWP) protocols. Calling this method is illegal when
+ * communicating over the TCP protocol. On WebSocket, {@link Sender#close()} still flushes pending rows.
*
* @return this instance for method chaining
* @see #autoFlushRows(int)
@@ -3561,9 +3561,14 @@ private QwpWebSocketSender buildWebSocket(HttpTokenProvider tokenProvider) {
int actualAutoFlushRows = autoFlushRows == PARAMETER_NOT_SET_EXPLICITLY ? DEFAULT_WS_AUTO_FLUSH_ROWS : autoFlushRows;
int actualAutoFlushBytes = autoFlushBytes == PARAMETER_NOT_SET_EXPLICITLY ? DEFAULT_WS_AUTO_FLUSH_BYTES : autoFlushBytes;
- long actualAutoFlushIntervalNanos = autoFlushIntervalMillis == PARAMETER_NOT_SET_EXPLICITLY
- ? DEFAULT_WS_AUTO_FLUSH_INTERVAL_NANOS
- : TimeUnit.MILLISECONDS.toNanos(autoFlushIntervalMillis);
+ long actualAutoFlushIntervalNanos;
+ if (autoFlushIntervalMillis == PARAMETER_NOT_SET_EXPLICITLY) {
+ actualAutoFlushIntervalNanos = DEFAULT_WS_AUTO_FLUSH_INTERVAL_NANOS;
+ } else if (autoFlushIntervalMillis == Integer.MAX_VALUE) {
+ actualAutoFlushIntervalNanos = 0; // explicit off: QwpWebSocketSender treats 0 as disabled
+ } else {
+ actualAutoFlushIntervalNanos = TimeUnit.MILLISECONDS.toNanos(autoFlushIntervalMillis);
+ }
Supplier
+ * Every per-frame check runs per frame, in the old order. A frame that is
+ * not a plain send (orphan tail, corrupt length, torn dictionary, segment
+ * hand-off) ends a non-empty batch BEFORE it: the batch is flushed and the
+ * next call meets that frame with an empty batch, exactly as the
+ * one-frame-per-call path did after its previous frame went out.
*/
- private boolean trySendOne() {
- if (orphanSkipTipFsn >= 0 && fsnAtZero + nextWireSeq >= orphanSkipStartFsn) {
- // The send cursor reached the orphaned deferred tail. Its frames
- // belong to an aborted transaction and must never be transmitted
- // (a commit-bearing frame from this session would commit them --
- // partial-transaction resurrection).
- if (!tryRetireOrphanTail()) {
- // Frames below the tail still await server acks; keep ticking
- // (tryReceiveAcks runs every loop iteration) until the ack
- // watermark reaches the tail's lower edge.
- return false;
- }
- if (nextWireSeq == 0) {
- // Reached only when the tail was NOT retirable at connection setup --
- // both setup sites (swapClient, positionCursorForStart) call
- // tryRetireOrphanTail first -- which means frames below it still needed
- // acks, which means they were SENT on this connection. So nextWireSeq is
- // never 0 here in practice and this arm is effectively dead; it is kept
- // as a cheap correctness guard rather than removed.
- //
- // NB the mapping is untouched only while NO wire sequence has been
- // consumed. A dictionary catch-up consumes sequences 0..n-1 before any
- // data frame, so after one this test is false and the recycle below --
- // which re-anchors the mapping from scratch -- is the correct path, not
- // an avoidable cost.
- try {
- positionCursorForStart();
- } catch (CatchUpSendException e) {
- // Re-anchor's catch-up send failed. fail() here is a fresh,
- // non-re-entrant connectLoop entry from the I/O loop body --
- // the same recovery a normal trySendOne send failure takes.
- // Preserve the wrapper on a cap gap so connectLoop's
- // isCatchUpCapGap keeps the orphan settle episode (attempt count
- // + dwell anchor) alive across the re-anchor recycle; an ordinary
- // failure unwraps to the raw cause and restarts the episode, like
- // any normal send failure.
- fail(isCatchUpCapGap(e) ? e : e.getCause());
+ private boolean trySendBatch() {
+ // Appending advances these speculatively; reconnect must snapshot only
+ // completed flushes and retain any older replay target after a failure.
+ final long batchStartWireSeq = nextWireSeq;
+ final long batchStartReplayTargetFsn = replayTargetFsn;
+ int batchFrames = 0;
+ int batchReplayed = 0;
+ long batchBytes = 0;
+ while (true) {
+ if (orphanSkipTipFsn >= 0 && fsnAtZero + nextWireSeq >= orphanSkipStartFsn) {
+ if (batchFrames > 0) {
+ break;
+ }
+ // The send cursor reached the orphaned deferred tail. Its frames
+ // belong to an aborted transaction and must never be transmitted
+ // (a commit-bearing frame from this session would commit them --
+ // partial-transaction resurrection).
+ if (!tryRetireOrphanTail()) {
+ // Frames below the tail still await server acks; keep ticking
+ // (tryReceiveAcks runs every loop iteration) until the ack
+ // watermark reaches the tail's lower edge.
return false;
}
- return true;
+ if (nextWireSeq == 0) {
+ // Reached only when the tail was NOT retirable at connection setup --
+ // both setup sites (swapClient, positionCursorForStart) call
+ // tryRetireOrphanTail first -- which means frames below it still needed
+ // acks, which means they were SENT on this connection. So nextWireSeq is
+ // never 0 here in practice and this arm is effectively dead; it is kept
+ // as a cheap correctness guard rather than removed.
+ //
+ // NB the mapping is untouched only while NO wire sequence has been
+ // consumed. A dictionary catch-up consumes sequences 0..n-1 before any
+ // data frame, so after one this test is false and the recycle below --
+ // which re-anchors the mapping from scratch -- is the correct path, not
+ // an avoidable cost.
+ try {
+ positionCursorForStart();
+ } catch (CatchUpSendException e) {
+ // Re-anchor's catch-up send failed. fail() here is a fresh,
+ // non-re-entrant connectLoop entry from the I/O loop body --
+ // the same recovery a normal trySendBatch send failure takes.
+ // Preserve the wrapper on a cap gap so connectLoop's
+ // isCatchUpCapGap keeps the orphan settle episode (attempt count
+ // + dwell anchor) alive across the re-anchor recycle; an ordinary
+ // failure unwraps to the raw cause and restarts the episode, like
+ // any normal send failure.
+ fail(isCatchUpCapGap(e) ? e : e.getCause());
+ return false;
+ }
+ return true;
+ }
+ // Frames were already sent on this connection: the linear
+ // fsnAtZero + wireSeq mapping cannot express the gap the retired
+ // tail leaves behind. Recycle; swapClient re-anchors the mapping
+ // at ackedFsn + 1 = the first frame past the tail. One reconnect,
+ // once per recovery -- and only on the slow path (unacked
+ // committed frames existed below the tail).
+ fail(new LineSenderException(
+ "recycling connection after retiring orphaned deferred tail (wire-seq realignment)"));
+ return false;
}
- // Frames were already sent on this connection: the linear
- // fsnAtZero + wireSeq mapping cannot express the gap the retired
- // tail leaves behind. Recycle; swapClient re-anchors the mapping
- // at ackedFsn + 1 = the first frame past the tail. One reconnect,
- // once per recovery -- and only on the slow path (unacked
- // committed frames existed below the tail).
- fail(new LineSenderException(
- "recycling connection after retiring orphaned deferred tail (wire-seq realignment)"));
- return false;
- }
- long pub = sendingSegment.publishedOffset();
- if (sendOffset >= pub) {
- // Nothing more in the current segment. If it's a sealed segment
- // (no longer the live active), advance to the next one.
- if (sendingSegment != engine.activeSegment()) {
- // The producer can publish the current segment's last frame
- // between our first publishedOffset() read and the active
- // segment check above. Re-read before leaving the segment, or
- // that frame is skipped permanently.
- pub = sendingSegment.publishedOffset();
- if (sendOffset >= pub) {
- MmapSegment next = advanceSegment();
- if (next != sendingSegment) {
- sendingSegment = next;
- return true; // let the next iteration try sending
+ long pub = sendingSegment.publishedOffset();
+ if (sendOffset >= pub) {
+ // Nothing more in the current segment. If it's a sealed segment
+ // (no longer the live active), advance to the next one.
+ if (sendingSegment != engine.activeSegment()) {
+ // The producer can publish the current segment's last frame
+ // between our first publishedOffset() read and the active
+ // segment check above. Re-read before leaving the segment, or
+ // that frame is skipped permanently.
+ pub = sendingSegment.publishedOffset();
+ if (sendOffset >= pub) {
+ if (batchFrames > 0) {
+ break;
+ }
+ MmapSegment next = advanceSegment();
+ if (next != sendingSegment) {
+ sendingSegment = next;
+ return true; // let the next iteration try sending
+ }
}
+ } else {
+ break;
}
- } else {
+ }
+ // At least the frame header is published; check we have the full frame.
+ if (sendOffset + MmapSegment.FRAME_HEADER_SIZE > pub) {
+ break;
+ }
+ long base = sendingSegment.address();
+ // Frame layout: [u32 crc][u32 payloadLen][payload].
+ int payloadLen = Unsafe.getUnsafe().getInt(base + sendOffset + 4);
+ if (payloadLen < 0) {
+ if (batchFrames > 0) {
+ break;
+ }
+ fail(new LineSenderException(
+ "negative payloadLen at offset " + sendOffset
+ + " in segment baseSeq=" + sendingSegment.baseSeq()));
return false;
}
+ long frameEnd = sendOffset + MmapSegment.FRAME_HEADER_SIZE + payloadLen;
+ if (frameEnd > pub) {
+ break; // payload not fully published yet
+ }
+ long wireLen = WebSocketFrameWriter.headerSize(payloadLen, true) + (long) payloadLen;
+ if (batchFrames > 0 && batchBytes + wireLen > client.batchByteCap()) {
+ break; // fixed byte cap: the next call carries on from this frame
+ }
+ long frameAddr = base + sendOffset + MmapSegment.FRAME_HEADER_SIZE;
+ // Torn-dictionary guard. sentDictCount is this loop's model of how many ids the
+ // CURRENT server has been told about. A frame whose delta starts ABOVE that
+ // coverage references ids the server was never given, and the server now rejects
+ // such a frame with STATUS_DICTIONARY_GAP rather than null-padding the hole. That
+ // rejection is retriable -- a wire recycle re-registers from id 0 -- but this
+ // frame cannot become sendable without one, and against a server old enough to
+ // null-pad it would land rows with silently NULL symbols. Fail terminally here,
+ // where the unreplayable range is still attributable.
+ //
+ // The guard, the mirror and the catch-up are ONE mechanism and are all
+ // ungated (see the constructor). A frame at deltaStart == sentDictCount is
+ // contiguous and extends the coverage, so a slot whose frames start at id 0
+ // replays cleanly with no persisted dictionary at all -- which is exactly why
+ // the mirror below must keep advancing even when the dictionary is missing.
+ // A gap here therefore means genuine loss: a host/power crash tore the
+ // unsynced .symbol-dict below the ids the surviving frames still reference
+ // (SF is process-crash but not host-crash durable).
+ int deltaStart = frameDeltaStart(frameAddr, payloadLen);
+ if (deltaStart > sentDictCount) {
+ if (batchFrames > 0) {
+ break;
+ }
+ recordFatal(new LineSenderException(
+ "recovered store-and-forward symbol dictionary is incomplete (likely a host crash): "
+ + "frame delta start " + deltaStart + " exceeds recovered dictionary size "
+ + sentDictCount + "; cannot replay without corrupting data -- resend required"));
+ return false;
+ }
+ try {
+ client.batchAppendBinary(frameAddr, payloadLen);
+ } catch (Throwable t) {
+ nextWireSeq = batchStartWireSeq;
+ replayTargetFsn = batchStartReplayTargetFsn;
+ fail(t);
+ return false;
+ }
+ // The bookkeeping below runs at append time, before the flush: the next
+ // frame's torn-dictionary guard needs this frame's coverage. If the flush
+ // fails, fail() re-anchors the wire state at ackedFsn + 1 on a fresh
+ // client, so an appended-but-unflushed frame is just another unacked
+ // frame; the persistent dictionary mirror is replayed by the catch-up.
+ //
+ // A real ring frame (data or commit) is going out on this connection,
+ // as opposed to only the dictionary catch-up. onClose / handleServerRejection
+ // key their poison-strike vs pre-send decision off this, not off nextWireSeq
+ // (which the catch-up advances).
+ dataFrameSentThisConnection = true;
+ if (deltaStart >= 0) {
+ // Mirror the symbols this frame just registered on the server, so a later
+ // reconnect can rebuild the whole dictionary and the guard above keeps an
+ // accurate view of the server's coverage. Idempotent on replay: a frame
+ // whose delta we already hold advances nothing.
+ //
+ // Ungated (see the constructor): this is the ONLY thing that advances
+ // sentDictCount from the frames themselves, so gating it while leaving the
+ // guard ungated froze the coverage at 0 and terminal'd a slot that replays
+ // perfectly from id 0.
+ accumulateSentDict(frameAddr, payloadLen, deltaStart);
+ }
+ sendOffset = frameEnd;
+ long fsnSent = fsnAtZero + nextWireSeq;
+ nextWireSeq++;
+ batchFrames++;
+ batchBytes += wireLen;
+ if (replayTargetFsn >= 0) {
+ batchReplayed++;
+ if (fsnSent >= replayTargetFsn) {
+ replayTargetFsn = -1L; // catch-up complete
+ }
+ }
}
- // At least the frame header is published; check we have the full frame.
- if (sendOffset + MmapSegment.FRAME_HEADER_SIZE > pub) {
- return false;
- }
- long base = sendingSegment.address();
- // Frame layout: [u32 crc][u32 payloadLen][payload].
- int payloadLen = Unsafe.getUnsafe().getInt(base + sendOffset + 4);
- if (payloadLen < 0) {
- fail(new LineSenderException(
- "negative payloadLen at offset " + sendOffset
- + " in segment baseSeq=" + sendingSegment.baseSeq()));
- return false;
- }
- long frameEnd = sendOffset + MmapSegment.FRAME_HEADER_SIZE + payloadLen;
- if (frameEnd > pub) {
- return false; // payload not fully published yet
- }
- long frameAddr = base + sendOffset + MmapSegment.FRAME_HEADER_SIZE;
- // Torn-dictionary guard. sentDictCount is this loop's model of how many ids the
- // CURRENT server has been told about. A frame whose delta starts ABOVE that
- // coverage references ids the server was never given, and the server now rejects
- // such a frame with STATUS_DICTIONARY_GAP rather than null-padding the hole. That
- // rejection is retriable -- a wire recycle re-registers from id 0 -- but this
- // frame cannot become sendable without one, and against a server old enough to
- // null-pad it would land rows with silently NULL symbols. Fail terminally here,
- // where the unreplayable range is still attributable.
- //
- // The guard, the mirror and the catch-up are ONE mechanism and are all
- // ungated (see the constructor). A frame at deltaStart == sentDictCount is
- // contiguous and extends the coverage, so a slot whose frames start at id 0
- // replays cleanly with no persisted dictionary at all -- which is exactly why
- // the mirror below must keep advancing even when the dictionary is missing.
- // A gap here therefore means genuine loss: a host/power crash tore the
- // unsynced .symbol-dict below the ids the surviving frames still reference
- // (SF is process-crash but not host-crash durable).
- int deltaStart = frameDeltaStart(frameAddr, payloadLen);
- if (deltaStart > sentDictCount) {
- recordFatal(new LineSenderException(
- "recovered store-and-forward symbol dictionary is incomplete (likely a host crash): "
- + "frame delta start " + deltaStart + " exceeds recovered dictionary size "
- + sentDictCount + "; cannot replay without corrupting data -- resend required"));
+ if (batchFrames == 0) {
return false;
}
try {
- client.sendBinary(frameAddr, payloadLen);
+ client.batchFlush();
} catch (Throwable t) {
+ nextWireSeq = batchStartWireSeq;
+ replayTargetFsn = batchStartReplayTargetFsn;
fail(t);
return false;
}
- // A real ring frame (data or commit) has now gone out on this connection,
- // as opposed to only the dictionary catch-up. onClose / handleServerRejection
- // key their poison-strike vs pre-send decision off this, not off nextWireSeq
- // (which the catch-up advances).
- dataFrameSentThisConnection = true;
- if (deltaStart >= 0) {
- // Mirror the symbols this frame just registered on the server, so a later
- // reconnect can rebuild the whole dictionary and the guard above keeps an
- // accurate view of the server's coverage. Idempotent on replay: a frame
- // whose delta we already hold advances nothing.
- //
- // Ungated (see the constructor): this is the ONLY thing that advances
- // sentDictCount from the frames themselves, so gating it while leaving the
- // guard ungated froze the coverage at 0 and terminal'd a slot that replays
- // perfectly from id 0.
- accumulateSentDict(frameAddr, payloadLen, deltaStart);
- }
lastFrameOrPingNanos = System.nanoTime();
- sendOffset = frameEnd;
- long fsnSent = fsnAtZero + nextWireSeq;
- nextWireSeq++;
- counters.framesSent.incrementAndGet();
- if (replayTargetFsn >= 0) {
- counters.framesReplayed.incrementAndGet();
- if (fsnSent >= replayTargetFsn) {
- replayTargetFsn = -1L; // catch-up complete
- }
+ counters.framesSent.addAndGet(batchFrames);
+ if (batchReplayed > 0) {
+ counters.framesReplayed.addAndGet(batchReplayed);
}
return true;
}
@@ -4288,7 +4347,7 @@ boolean isConnectInFlight() {
* current connection. Thrown by {@link #sendDictCatchUp}/{@link
* #sendCatchUpChunk} instead of calling {@link #fail}: the catch-up runs
* inside {@link #connectLoop} (via {@link #swapClient}) and, on the initial
- * connect, inside {@link #start()} / {@link #trySendOne} on the caller
+ * connect, inside {@link #start()} / {@link #trySendBatch} on the caller
* thread. Calling {@code fail()} from there would re-enter {@code
* connectLoop} -- corrupting the {@code fsnAtZero}/{@code nextWireSeq} wire
* mapping (a subsequent ACK then trims un-acked frames) and growing the
@@ -4297,7 +4356,7 @@ boolean isConnectInFlight() {
* thread and block {@code Sender} construction indefinitely. Each catch
* site instead turns it into ONE non-re-entrant reconnect: {@code
* connectLoop}'s own retry catch (swapClient path), a fresh {@code fail()}
- * from the I/O loop body (trySendOne path), or dropping the dead client so
+ * from the I/O loop body (trySendBatch path), or dropping the dead client so
* the I/O thread reconnects (start path). A JVM {@code Error} is never
* wrapped -- it must stay terminal. The {@code capGap} marker lets the reconnect
* loop preserve only consecutive incompatible-cap observations; ordinary catch-up
@@ -4607,7 +4666,7 @@ private void handleServerRejection(long wireSeq) {
// NACK is for a dictionary catch-up frame -- which occupies the
// already-acked wire sequences below replayStart -- not a data frame
// this connection sent. (dataFrameSentThisConnection can be true here
- // because trySendOne ships the head data frame before tryReceiveAcks
+ // because trySendBatch ships the head data frame before tryReceiveAcks
// reads the catch-up's NACK in the same loop iteration.) Attributing
// it a data FSN would key recordHeadRejectionStrike() off a
// below-baseline FSN -- negative when replayStart < catchUpFrames --
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
@@ -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/QuestDBBuilderTest.java b/core/src/test/java/io/questdb/client/test/QuestDBBuilderTest.java
index e338aa72..97a480b1 100644
--- a/core/src/test/java/io/questdb/client/test/QuestDBBuilderTest.java
+++ b/core/src/test/java/io/questdb/client/test/QuestDBBuilderTest.java
@@ -212,23 +212,27 @@ public void testMalformedEgressConfigRejectedAtBuildWithMinZero() {
"compression_level");
}
+ @Test
+ public void testAutoFlushOffAcceptedAtBuildWithMinZero() {
+ // auto_flush_interval=off and auto_flush=off are legal on WebSocket.
+ QuestDB.builder().fromConfig(
+ "ws::addr=127.0.0.1:1;auto_flush_interval=off;sender_pool_min=0;query_pool_min=0;").build().close();
+ QuestDB.builder().fromConfig(
+ "ws::addr=127.0.0.1:1;auto_flush=off;sender_pool_min=0;query_pool_min=0;").build().close();
+ }
+
@Test
public void testMalformedIngressConfigRejectedAtBuildWithMinZero() {
// sender_pool_min=0 pre-warms nothing, so build() never constructs a
// Sender -- yet it must still reject a malformed ingress key in the
// single cluster config up front. Covers a typed enum (tls_verify), a
// registry-STRING value that only the real Sender parse validates
- // (auto_flush_rows), and WebSocket build-time checks: auto_flush=off and
- // auto_flush_interval=off both disable auto-flush (unsupported on
- // WebSocket), and sf_durability=flush is not yet supported.
+ // (auto_flush_rows), and a WebSocket build-time check: sf_durability=flush
+ // is not yet supported.
assertBuildRejected(
"wss::addr=127.0.0.1:1;tls_verify=strict;sender_pool_min=0;query_pool_min=0;", "tls_verify");
assertBuildRejected(
"ws::addr=127.0.0.1:1;auto_flush_rows=abc;sender_pool_min=0;query_pool_min=0;", "auto_flush_rows");
- assertBuildRejected(
- "ws::addr=127.0.0.1:1;auto_flush_interval=off;sender_pool_min=0;query_pool_min=0;", "auto-flush");
- assertBuildRejected(
- "ws::addr=127.0.0.1:1;auto_flush=off;sender_pool_min=0;query_pool_min=0;", "auto-flush");
assertBuildRejected(
"ws::addr=127.0.0.1:1;sf_durability=flush;sender_pool_min=0;query_pool_min=0;", "not yet supported");
}
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java b/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java
index d0335886..cc8c78aa 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java
@@ -30,6 +30,7 @@
import io.questdb.client.cutlass.http.client.WebSocketClient;
import io.questdb.client.cutlass.http.client.WebSocketFrameHandler;
import io.questdb.client.cutlass.http.client.WebSocketSendBuffer;
+import io.questdb.client.network.IOOperation;
import io.questdb.client.network.PlainSocketFactory;
import io.questdb.client.network.Socket;
import io.questdb.client.network.SocketReadinessWaiter;
@@ -40,6 +41,7 @@
import java.lang.reflect.Field;
import java.lang.reflect.Method;
+import java.util.Arrays;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.atomic.AtomicReference;
@@ -47,6 +49,63 @@
public class WebSocketClientTest {
+ /**
+ * Batched frames go out back to back in ONE send(), each a well-formed
+ * masked binary frame with an exact-size header (no MAX_HEADER_SIZE gap
+ * between frames); an empty flush writes nothing.
+ */
+ @Test
+ public void testBatchFlushSendsAppendedFramesInOneWrite() throws Exception {
+ assertMemoryLeak(() -> {
+ CapturingSocket socket = new CapturingSocket();
+ // 7: 7-bit length, 200: 16-bit length, 0: empty payload
+ int[] lens = {7, 200, 0};
+ long payload = Unsafe.malloc(200, MemoryTag.NATIVE_DEFAULT);
+ try (CapturingWebSocketClient client = new CapturingWebSocketClient(socket)) {
+ setUpgradedTrue(client);
+ for (int i = 0; i < 200; i++) {
+ Unsafe.getUnsafe().putByte(payload + i, (byte) i);
+ }
+
+ client.batchFlush();
+ Assert.assertEquals("an empty flush must not write", 0, socket.sendCalls);
+
+ for (int len : lens) {
+ client.batchAppendBinary(payload, len);
+ }
+ Assert.assertEquals("appending must not write", 0, socket.sendCalls);
+ client.batchFlush();
+ Assert.assertEquals("the whole batch must be one write", 1, socket.sendCalls);
+
+ byte[] wire = socket.sent;
+ int pos = 0;
+ for (int len : lens) {
+ Assert.assertEquals("FIN + BINARY", (byte) 0x82, wire[pos]);
+ int maskPos;
+ if (len <= 125) {
+ Assert.assertEquals((byte) (0x80 | len), wire[pos + 1]);
+ maskPos = pos + 2;
+ } else {
+ Assert.assertEquals((byte) (0x80 | 126), wire[pos + 1]);
+ Assert.assertEquals(len, ((wire[pos + 2] & 0xFF) << 8) | (wire[pos + 3] & 0xFF));
+ maskPos = pos + 4;
+ }
+ int payloadPos = maskPos + 4;
+ for (int i = 0; i < len; i++) {
+ Assert.assertEquals((byte) i, (byte) (wire[payloadPos + i] ^ wire[maskPos + (i & 3)]));
+ }
+ pos = payloadPos + len;
+ }
+ Assert.assertEquals("no bytes beyond the 3 frames", wire.length, pos);
+
+ client.batchFlush();
+ Assert.assertEquals("a flushed batch is empty", 1, socket.sendCalls);
+ } finally {
+ Unsafe.free(payload, 200, MemoryTag.NATIVE_DEFAULT);
+ }
+ });
+ }
+
/**
* close() frees native memory (recv/fragment buffers, send buffers), so
* its guard must be a CAS, not a volatile check-then-act: two concurrent
@@ -130,6 +189,16 @@ public void testExtractMaxBatchSizeParsesPositive() throws Exception {
Assert.assertEquals(16 * 1024 * 1024, invokeExtractMaxBatchSize(response));
}
+ @Test
+ public void testFrameBurstIsParsedInPlace() throws Exception {
+ assertFrameBurstIsParsedInPlace(false);
+ }
+
+ @Test
+ public void testFrameBurstIsParsedInPlaceWithBlockingReceive() throws Exception {
+ assertFrameBurstIsParsedInPlace(true);
+ }
+
/**
* A frame handler may close() the client from inside its callback:
* CursorWebSocketSendLoop's NACK-recycle path (handleServerRejection /
@@ -349,6 +418,21 @@ public void testSendBinaryAssemblesTwoPayloadSlicesIntoOneMaskedFrame() throws E
});
}
+ /**
+ * doSend() must try send() first and poll for writability only when the
+ * socket reports full (send returns 0 / ERETRY); a pre-send ioWait costs
+ * an epoll_ctl + epoll_wait per frame even when the socket is writable.
+ */
+ @Test
+ public void testSendPollsForWritabilityOnlyOnBackpressure() throws Exception {
+ assertMemoryLeak(() -> {
+ // whole frame accepted by the first send(): no poll
+ assertSendBinaryWriteWaits(new CapturingSocket(), 0);
+ // 2 bytes accepted, socket full once, then the rest: exactly one poll
+ assertSendBinaryWriteWaits(new CapturingSocket(2, 0), 1);
+ });
+ }
+
@Test
public void testSendCloseFrameDoesNotClobberSendBuffer() throws Exception {
assertMemoryLeak(() -> {
@@ -410,6 +494,94 @@ public void testSendPingDoesNotClobberSendBuffer() throws Exception {
});
}
+ /**
+ * One recv delivering a burst of small frames (the shape of a server's
+ * cumulative ACKs) must be parsed in place: compacting the unread tail
+ * after every frame memmoved the rest of the burst once per frame. The
+ * burst fills the 64 KiB buffer and ends mid-frame, so the next recv must
+ * compact that tail BEFORE the grow check -- a tail left at the buffer end
+ * would read as "full" and grow the buffer for no reason.
+ */
+ private static void assertFrameBurstIsParsedInPlace(boolean blocking) throws Exception {
+ assertMemoryLeak(() -> {
+ final int frameLen = 13; // FIN|BINARY, len=11
+ final int frameCount = 5100; // > 64 KiB, so the first recv ends mid-frame
+ byte[] data = new byte[frameLen * frameCount];
+ for (int i = 0; i < frameCount; i++) {
+ data[i * frameLen] = (byte) 0x82;
+ data[i * frameLen + 1] = 11;
+ data[i * frameLen + 2] = (byte) i;
+ data[i * frameLen + 3] = (byte) (i >> 8);
+ }
+ try (FrameFeedWebSocketClient client = new FrameFeedWebSocketClient(data)) {
+ setUpgradedTrue(client);
+ final int bufSize = getIntField(client, "recvBufSize");
+ final long[] payload = {0, 0}; // first, last
+ final int[] frames = {0};
+ WebSocketFrameHandler handler = new WebSocketFrameHandler() {
+ @Override
+ public void onBinaryMessage(long payloadPtr, int payloadLen) {
+ Assert.assertEquals(11, payloadLen);
+ int tag = (Unsafe.getUnsafe().getByte(payloadPtr) & 0xFF)
+ | (Unsafe.getUnsafe().getByte(payloadPtr + 1) & 0xFF) << 8;
+ Assert.assertEquals(frames[0], tag);
+ if (frames[0]++ == 0) {
+ payload[0] = payloadPtr;
+ }
+ payload[1] = payloadPtr;
+ }
+
+ @Override
+ public void onClose(int code, String reason) {
+ Assert.fail("unexpected close frame");
+ }
+ };
+
+ int firstRecvFrames = bufSize / frameLen;
+ for (int i = 0; i < firstRecvFrames; i++) {
+ Assert.assertTrue(blocking
+ ? client.receiveFrame(handler, 1000)
+ : client.tryReceiveFrame(handler));
+ Assert.assertEquals("frame " + i + " must be parsed where recv put it",
+ (long) i * frameLen, payload[1] - payload[0]);
+ }
+ for (int i = firstRecvFrames; i < frameCount; i++) {
+ Assert.assertTrue(blocking
+ ? client.receiveFrame(handler, 1000)
+ : client.tryReceiveFrame(handler));
+ }
+ Assert.assertEquals(frameCount, frames[0]);
+ Assert.assertEquals("partial tail must not grow the recv buffer",
+ bufSize, getIntField(client, "recvBufSize"));
+ }
+ });
+ }
+
+ private static void assertSendBinaryWriteWaits(CapturingSocket socket, int expectedWriteWaits) throws Exception {
+ long payload = Unsafe.malloc(7, MemoryTag.NATIVE_DEFAULT);
+ try (CapturingWebSocketClient client = new CapturingWebSocketClient(socket)) {
+ setUpgradedTrue(client);
+ for (int i = 0; i < 7; i++) {
+ Unsafe.getUnsafe().putByte(payload + i, (byte) (i + 1));
+ }
+
+ client.sendBinary(payload, 7);
+
+ byte[] frame = socket.sent;
+ Assert.assertEquals(2 + 4 + 7, frame.length);
+ Assert.assertEquals((byte) 0x82, frame[0]);
+ Assert.assertEquals((byte) (0x80 | 7), frame[1]);
+ byte[] unmasked = new byte[7];
+ for (int i = 0; i < unmasked.length; i++) {
+ unmasked[i] = (byte) (frame[6 + i] ^ frame[2 + (i & 3)]);
+ }
+ Assert.assertArrayEquals(new byte[]{1, 2, 3, 4, 5, 6, 7}, unmasked);
+ Assert.assertEquals(expectedWriteWaits, client.writeWaits);
+ } finally {
+ Unsafe.free(payload, 7, MemoryTag.NATIVE_DEFAULT);
+ }
+ }
+
private static int getIntField(Object obj, String name) throws Exception {
Class clazz = obj.getClass();
while (clazz != null) {
@@ -500,20 +672,35 @@ public boolean wantsTlsWrite() {
}
}
+ /**
+ * Accumulates every byte send() accepts. The i-th send() call accepts at
+ * most acceptLimits[i] bytes (0 = socket full); calls past the script
+ * accept everything.
+ */
private static class CapturingSocket extends FakeSocket {
- private byte[] sent;
+ private final int[] acceptLimits;
+ private byte[] sent = new byte[0];
+ private int sendCalls;
+
+ CapturingSocket(int... acceptLimits) {
+ this.acceptLimits = acceptLimits;
+ }
@Override
public int send(long bufferPtr, int bufferLen) {
- sent = new byte[bufferLen];
- for (int i = 0; i < bufferLen; i++) {
- sent[i] = Unsafe.getUnsafe().getByte(bufferPtr + i);
+ int n = sendCalls < acceptLimits.length ? Math.min(acceptLimits[sendCalls], bufferLen) : bufferLen;
+ sendCalls++;
+ int offset = sent.length;
+ sent = Arrays.copyOf(sent, offset + n);
+ for (int i = 0; i < n; i++) {
+ sent[offset + i] = Unsafe.getUnsafe().getByte(bufferPtr + i);
}
- return bufferLen;
+ return n;
}
}
private static class CapturingWebSocketClient extends TestWebSocketClient {
+ private int writeWaits;
CapturingWebSocketClient(CapturingSocket socket) {
super(DefaultHttpClientConfiguration.INSTANCE, (nf, log) -> socket);
@@ -521,6 +708,9 @@ private static class CapturingWebSocketClient extends TestWebSocketClient {
@Override
protected void ioWait(int timeout, int op) {
+ if (op == IOOperation.WRITE) {
+ writeWaits++;
+ }
}
@Override
@@ -596,8 +786,8 @@ public boolean wantsTlsWrite() {
/**
* WebSocketClient over a socket pre-loaded with canned server frames;
- * sends always succeed. Used to drive a real tryReceiveFrame ->
- * tryParseFrame -> handler dispatch on the test thread.
+ * sends always succeed. Drives both receive APIs through real parsing
+ * and handler dispatch on the test thread.
*/
private static class FrameFeedWebSocketClient extends TestWebSocketClient {
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/AutoFlushIntervalClockTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/AutoFlushIntervalClockTest.java
new file mode 100644
index 00000000..b2522295
--- /dev/null
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/AutoFlushIntervalClockTest.java
@@ -0,0 +1,165 @@
+/*+*****************************************************************************
+ * ___ _ ____ ____
+ * / _ \ _ _ ___ ___| |_| _ \| __ )
+ * | | | | | | |/ _ \/ __| __| | | | _ \
+ * | |_| | |_| | __/\__ \ |_| |_| | |_) |
+ * \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ * Copyright (c) 2014-2019 Appsicle
+ * Copyright (c) 2019-2026 QuestDB
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ ******************************************************************************/
+
+package io.questdb.client.test.cutlass.qwp.client;
+
+import io.questdb.client.Sender;
+import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
+import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static io.questdb.client.test.tools.TestUtils.assertMemoryLeak;
+
+/**
+ * Pins how often the WebSocket sender reads its clock on the row path, via the
+ * injectable auto-flush clock, and that interval-based auto-flush still fires.
+ */
+public class AutoFlushIntervalClockTest {
+ private static final int N = 1000;
+
+ @Test
+ public void testDefaultIntervalWhenKeyAbsentAndZeroWhenOff() throws Exception {
+ assertMemoryLeak(() -> {
+ try (TestWebSocketServer server = startAckingServer()) {
+ try (Sender sender = Sender.fromConfig(cfg(server))) {
+ Assert.assertEquals("ws:: without auto_flush_interval keeps the 100 ms default",
+ TimeUnit.MILLISECONDS.toNanos(100),
+ ((QwpWebSocketSender) sender).getAutoFlushIntervalNanos());
+ }
+ try (Sender sender = Sender.fromConfig(cfg(server) + "auto_flush_interval=off;")) {
+ Assert.assertEquals(0L, ((QwpWebSocketSender) sender).getAutoFlushIntervalNanos());
+ }
+ }
+ });
+ }
+
+ @Test
+ public void testIntervalFlushesOnSecondRowOfAgedBatch() throws Exception {
+ assertMemoryLeak(() -> {
+ try (TestWebSocketServer server = startAckingServer();
+ Sender sender = Sender.fromConfig(cfg(server) + "auto_flush_rows=1000000;auto_flush_interval=100;")) {
+ QwpWebSocketSender ws = (QwpWebSocketSender) sender;
+ AtomicLong now = new AtomicLong();
+ ws.setNanoClockForTesting(now::get);
+ long fsn0 = ws.getCursorEngineForTesting().publishedFsn();
+
+ sender.table("t").longColumn("v", 1L).atNow();
+ Assert.assertEquals("first row must stay buffered",
+ fsn0, ws.getCursorEngineForTesting().publishedFsn());
+
+ now.addAndGet(TimeUnit.MILLISECONDS.toNanos(200));
+ sender.table("t").longColumn("v", 2L).atNow();
+ Assert.assertTrue("batch older than the interval must auto-flush on its second row",
+ ws.getCursorEngineForTesting().publishedFsn() > fsn0);
+ }
+ });
+ }
+
+ @Test
+ public void testIntervalOffReadsNoClock() throws Exception {
+ assertMemoryLeak(() -> {
+ try (TestWebSocketServer server = startAckingServer();
+ Sender sender = Sender.fromConfig(cfg(server) + "auto_flush_interval=off;")) {
+ AtomicLong reads = new AtomicLong();
+ ((QwpWebSocketSender) sender).setNanoClockForTesting(() -> reads.incrementAndGet() * 1_000L);
+ for (int i = 0; i < N; i++) {
+ sender.table("t").longColumn("v", i).atNow();
+ sender.flush();
+ }
+ Assert.assertEquals("auto_flush_interval=off must not read the clock on the row path",
+ 0L, reads.get());
+ }
+ });
+ }
+
+ @Test
+ public void testIntervalOffKeepsRowCapAndExplicitFlush() throws Exception {
+ assertMemoryLeak(() -> {
+ try (TestWebSocketServer server = startAckingServer();
+ Sender sender = Sender.fromConfig(cfg(server) + "auto_flush_interval=off;auto_flush_rows=10;")) {
+ QwpWebSocketSender ws = (QwpWebSocketSender) sender;
+ long fsn0 = ws.getCursorEngineForTesting().publishedFsn();
+ for (int i = 0; i < 9; i++) {
+ sender.table("t").longColumn("v", i).atNow();
+ }
+ Assert.assertEquals("9 rows stay below the row cap",
+ fsn0, ws.getCursorEngineForTesting().publishedFsn());
+ sender.table("t").longColumn("v", 9L).atNow();
+ long fsn1 = ws.getCursorEngineForTesting().publishedFsn();
+ Assert.assertTrue("the 10th row must trigger the row-cap flush", fsn1 > fsn0);
+
+ for (int i = 0; i < 3; i++) {
+ sender.table("t").longColumn("v", i).atNow();
+ }
+ Assert.assertEquals(fsn1, ws.getCursorEngineForTesting().publishedFsn());
+ sender.flush();
+ Assert.assertTrue("explicit flush() must send the partial batch",
+ ws.getCursorEngineForTesting().publishedFsn() > fsn1);
+ }
+ });
+ }
+
+ @Test
+ public void testOneClockReadPerSingleRowFlush() throws Exception {
+ assertMemoryLeak(() -> {
+ try (TestWebSocketServer server = startAckingServer();
+ Sender sender = Sender.fromConfig(cfg(server) + "auto_flush_rows=1000000;auto_flush_interval=100;")) {
+ AtomicLong reads = new AtomicLong();
+ ((QwpWebSocketSender) sender).setNanoClockForTesting(() -> reads.incrementAndGet() * 1_000L);
+ for (int i = 0; i < N; i++) {
+ sender.table("t").longColumn("v", i).atNow();
+ sender.flush();
+ }
+ Assert.assertEquals("one clock read (the batch timestamp) per single-row flush",
+ N, reads.get());
+ }
+ });
+ }
+
+ private static String cfg(TestWebSocketServer server) {
+ return "ws::addr=localhost:" + server.getPort() + ";";
+ }
+
+ private static TestWebSocketServer startAckingServer() throws Exception {
+ AtomicLong seq = new AtomicLong();
+ TestWebSocketServer server = new TestWebSocketServer(new TestWebSocketServer.WebSocketServerHandler() {
+ @Override
+ public void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) {
+ try {
+ client.sendBinary(CloseDrainTest.buildAck(seq.getAndIncrement()));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ });
+ server.start();
+ Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+ return server;
+ }
+}
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java
index fbc833d8..3b8d316c 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java
@@ -1118,8 +1118,6 @@ public void testFailedPublishThenNewSymbolPersistsSuffixWithoutDuplicating() thr
// atNow() and throws from there instead of from the flush() this test
// brackets. The batch and the branch under test are identical either way;
// only the throw site moves, so parking the thresholds costs no coverage.
- // (WebSocket rejects auto_flush=off outright, hence the large finite
- // interval.)
String cfg = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir
+ ";sf_max_segment_bytes=1024"
+ ";auto_flush_bytes=off;auto_flush_rows=1000000;auto_flush_interval=60000;";
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java
index 1d83dff2..c5dad492 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java
@@ -425,12 +425,21 @@ public void testDefaultIsAsync() {
}
@Test
- public void testDisableAutoFlush_notSupportedForWebSocket() {
- assertThrowsAny(
- Sender.builder(Sender.Transport.WEBSOCKET)
- .address(LOCALHOST)
- .disableAutoFlush(),
- "not supported for WebSocket");
+ public void testDisableAutoFlush_supportedForWebSocket() throws Exception {
+ assertMemoryLeak(() -> {
+ try (TestWebSocketServer server = new TestWebSocketServer(new TestWebSocketServer.WebSocketServerHandler() {
+ })) {
+ server.start();
+ Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+ try (QwpWebSocketSender sender = (QwpWebSocketSender) Sender.builder(Sender.Transport.WEBSOCKET)
+ .address(LOCALHOST + ":" + server.getPort())
+ .disableAutoFlush()
+ .build()) {
+ Assert.assertEquals(0, sender.getAutoFlushRows());
+ Assert.assertEquals(0L, sender.getAutoFlushIntervalNanos());
+ }
+ }
+ });
}
@Test
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
- * This deliberately does NOT pin trySendOne's in-place re-anchor arm: both connection
+ * This deliberately does NOT pin trySendBatch's in-place re-anchor arm: both connection
* setup sites call tryRetireOrphanTail first, so that arm is only reached when frames
* below the tail still needed acks -- which means they were sent on this connection.
* Verified by reverting its guard: this test still passes.
@@ -362,7 +362,7 @@ public void testRecoveredMaxSymbolIdSpansSealedSegments() throws Exception {
@Test
public void testRecoveredMaxSymbolIdExcludesOrphanTailFrames() throws Exception {
// recoveredMaxSymbolId must reflect only COMMITTED (transmitted) frames, not
- // the aborted orphan-tail frames trySendOne retires without ever sending. A
+ // the aborted orphan-tail frames trySendBatch retires without ever sending. A
// host crash that tears the persisted dictionary down to the committed ids
// while an orphan-tail frame introduced a HIGHER id must NOT over-reject the
// resume: the producer never reuses an orphan id on the wire (the tail retires
@@ -648,7 +648,7 @@ public void testSelfSufficientFrameCannotHideUnackedRecoveryGap() throws Excepti
/**
* In-memory transport emulating a healthy server: counts sends and
* answers every received frame with a cumulative empty-table STATUS_OK
- * for the highest wire seq seen. sendBinary and tryReceiveFrame both run
+ * for the highest wire seq seen. batchAppendBinary and tryReceiveFrame both run
* on the I/O thread; sentCount is volatile for test-thread assertions.
*/
private static final class AckingClient extends TestWebSocketClient {
@@ -660,7 +660,7 @@ private static final class AckingClient extends TestWebSocketClient {
}
@Override
- public void sendBinary(long dataPtr, int length) {
+ public void batchAppendBinary(long dataPtr, int length) {
sentCount++;
}
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopPoisonFrameTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopPoisonFrameTest.java
index ae083b8f..52d04d49 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopPoisonFrameTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopPoisonFrameTest.java
@@ -893,7 +893,7 @@ private static final class NackingClient extends TestWebSocketClient {
}
@Override
- public void sendBinary(long dataPtr, int length) {
+ public void batchAppendBinary(long dataPtr, int length) {
sentCount++;
}
@@ -940,7 +940,7 @@ private static final class ClosingMiddleboxClient extends TestWebSocketClient {
}
@Override
- public void sendBinary(long dataPtr, int length) {
+ public void batchAppendBinary(long dataPtr, int length) {
sentCount++;
}
@@ -978,7 +978,7 @@ private static final class OrderlyClosingClient extends TestWebSocketClient {
}
@Override
- public void sendBinary(long dataPtr, int length) {
+ public void batchAppendBinary(long dataPtr, int length) {
}
@Override
@@ -1016,7 +1016,7 @@ private static final class PreSendClosingClient extends TestWebSocketClient {
}
@Override
- public void sendBinary(long dataPtr, int length) {
+ public void batchAppendBinary(long dataPtr, int length) {
}
@Override
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopReplayMetricsTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopReplayMetricsTest.java
new file mode 100644
index 00000000..4c13f61a
--- /dev/null
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopReplayMetricsTest.java
@@ -0,0 +1,251 @@
+/*+*****************************************************************************
+ * ___ _ ____ ____
+ * / _ \ _ _ ___ ___| |_| _ \| __ )
+ * | | | | | | |/ _ \/ __| __| | | | _ \
+ * | |_| | |_| | __/\__ \ |_| |_| | |_) |
+ * \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ * Copyright (c) 2014-2019 Appsicle
+ * Copyright (c) 2019-2026 QuestDB
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ ******************************************************************************/
+
+package io.questdb.client.test.cutlass.qwp.client.sf.cursor;
+
+import io.questdb.client.DefaultHttpClientConfiguration;
+import io.questdb.client.cutlass.http.client.HttpClientException;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop;
+import io.questdb.client.cutlass.qwp.websocket.WebSocketFrameWriter;
+import io.questdb.client.network.Socket;
+import io.questdb.client.network.SocketReadinessWaiter;
+import io.questdb.client.std.MemoryTag;
+import io.questdb.client.std.Unsafe;
+import io.questdb.client.test.tools.TestUtils;
+import io.questdb.client.test.tools.TestWebSocketClient;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class CursorWebSocketSendLoopReplayMetricsTest {
+ private static final int FRAME_COUNT = 4;
+ private static final int PAYLOAD_LEN = 8;
+ private static final int WIRE_LEN = PAYLOAD_LEN + WebSocketFrameWriter.headerSize(PAYLOAD_LEN, true);
+
+ @Test
+ public void testAppendFailureDoesNotCountBufferedFramesAsReplays() throws Exception {
+ assertFailedBatchDoesNotCountAsSent(0, 3);
+ }
+
+ @Test
+ public void testPartialFlushDoesNotCountFailedBatchAsSent() throws Exception {
+ // Even one complete frame accepted by the socket does not complete the batch.
+ assertFailedBatchDoesNotCountAsSent(WIRE_LEN, Integer.MAX_VALUE);
+ }
+
+ @Test
+ public void testReplayTargetSurvivesAnotherFailedFlush() throws Exception {
+ TestUtils.assertMemoryLeak(() -> {
+ ScriptedSocket originalSocket = new ScriptedSocket(Integer.MAX_VALUE);
+ ScriptedSocket retrySocket = new ScriptedSocket(2 * WIRE_LEN);
+ ScriptedSocket replacementSocket = new ScriptedSocket(Integer.MAX_VALUE);
+ try (CursorSendEngine engine = new CursorSendEngine(null, 16_384);
+ BatchingClient original = new BatchingClient(originalSocket);
+ BatchingClient retry = new BatchingClient(retrySocket);
+ BatchingClient replacement = new BatchingClient(replacementSocket)) {
+ appendFrames(engine);
+ retry.batchCap = 2 * WIRE_LEN;
+ BatchingClient[] reconnectClients = {retry, replacement};
+ int[] reconnectIndex = {0};
+ try (CursorWebSocketSendLoop loop = new CursorWebSocketSendLoop(
+ original, engine, 0L, () -> reconnectClients[reconnectIndex[0]++], 0L, 0L)) {
+ loop.setRunningForTest(true);
+ loop.positionCursorForStartForTest();
+ Assert.assertTrue(loop.trySendBatchForTest());
+ Assert.assertEquals(FRAME_COUNT, loop.getTotalFramesSent());
+ Assert.assertEquals(0, loop.getTotalFramesReplayed());
+
+ loop.connectLoopForTest(new HttpClientException("test disconnect"), "reconnect", 0L);
+ Assert.assertTrue(loop.trySendBatchForTest());
+ Assert.assertEquals(2, loop.getTotalFramesReplayed());
+ // Appending the remaining two frames clears replayTargetFsn before
+ // this flush fails. The next connection must still replay all four.
+ Assert.assertFalse(loop.trySendBatchForTest());
+ Assert.assertEquals(2 * WIRE_LEN, retrySocket.acceptedBytes);
+ Assert.assertEquals(FRAME_COUNT + 2, loop.getTotalFramesSent());
+ Assert.assertEquals(2, loop.getTotalFramesReplayed());
+
+ Assert.assertTrue(loop.trySendBatchForTest());
+ Assert.assertEquals(FRAME_COUNT * WIRE_LEN, replacementSocket.acceptedBytes);
+ Assert.assertEquals(2 * FRAME_COUNT + 2, loop.getTotalFramesSent());
+ Assert.assertEquals(FRAME_COUNT + 2, loop.getTotalFramesReplayed());
+ Assert.assertEquals(2, loop.getTotalReconnects());
+ loop.checkError();
+ }
+ }
+ });
+ }
+
+ @Test
+ public void testZeroByteFlushFailureDoesNotCountFirstDeliveryAsReplay() throws Exception {
+ assertFailedBatchDoesNotCountAsSent(0, Integer.MAX_VALUE);
+ }
+
+ private static void appendFrames(CursorSendEngine engine) {
+ long payload = Unsafe.malloc(PAYLOAD_LEN, MemoryTag.NATIVE_DEFAULT);
+ try {
+ for (int i = 0; i < FRAME_COUNT; i++) {
+ Unsafe.getUnsafe().putLong(payload, i);
+ Assert.assertEquals(i, engine.appendBlocking(payload, PAYLOAD_LEN));
+ }
+ } finally {
+ Unsafe.free(payload, PAYLOAD_LEN, MemoryTag.NATIVE_DEFAULT);
+ }
+ }
+
+ private static void assertFailedBatchDoesNotCountAsSent(int acceptedBytes, int failAppendAt) throws Exception {
+ TestUtils.assertMemoryLeak(() -> {
+ ScriptedSocket firstSocket = new ScriptedSocket(acceptedBytes);
+ ScriptedSocket replacementSocket = new ScriptedSocket(Integer.MAX_VALUE);
+ try (CursorSendEngine engine = new CursorSendEngine(null, 16_384);
+ BatchingClient first = new BatchingClient(firstSocket);
+ BatchingClient replacement = new BatchingClient(replacementSocket)) {
+ appendFrames(engine);
+ first.failAppendAt = failAppendAt;
+ try (CursorWebSocketSendLoop loop = new CursorWebSocketSendLoop(
+ first, engine, 0L, () -> replacement, 0L, 0L)) {
+ loop.setRunningForTest(true);
+ loop.positionCursorForStartForTest();
+ Assert.assertFalse(loop.trySendBatchForTest());
+ Assert.assertEquals(acceptedBytes, firstSocket.acceptedBytes);
+ Assert.assertEquals(Math.min(FRAME_COUNT, failAppendAt), first.appendCalls);
+ if (failAppendAt <= FRAME_COUNT) {
+ Assert.assertEquals("append failure must happen before any socket write", 0, firstSocket.sendCalls);
+ } else {
+ Assert.assertTrue("flush must reach the socket", firstSocket.sendCalls > 0);
+ }
+ Assert.assertEquals(1, loop.getTotalReconnects());
+ Assert.assertEquals(0, loop.getTotalFramesSent());
+ Assert.assertEquals(0, loop.getTotalFramesReplayed());
+
+ Assert.assertTrue(loop.trySendBatchForTest());
+ Assert.assertEquals(FRAME_COUNT * WIRE_LEN, replacementSocket.acceptedBytes);
+ Assert.assertEquals(FRAME_COUNT, loop.getTotalFramesSent());
+ Assert.assertEquals("failed batches must not establish a replay boundary",
+ 0, loop.getTotalFramesReplayed());
+ Assert.assertFalse("all queued frames must have been sent", loop.trySendBatchForTest());
+ loop.checkError();
+ }
+ }
+ });
+ }
+
+ private static class BatchingClient extends TestWebSocketClient {
+ private int appendCalls;
+ private int batchCap = super.batchByteCap();
+ private int failAppendAt = Integer.MAX_VALUE;
+
+ private BatchingClient(ScriptedSocket socket) {
+ super(DefaultHttpClientConfiguration.INSTANCE, (nf, log) -> socket);
+ }
+
+ @Override
+ public void batchAppendBinary(long dataPtr, int length) {
+ if (++appendCalls == failAppendAt) {
+ throw new HttpClientException("test append failure");
+ }
+ // Skip the handshake but use the real send buffer and batchFlush/socket path.
+ getSendBuffer().appendMaskedBinaryFrame(dataPtr, length);
+ }
+
+ @Override
+ public int batchByteCap() {
+ return batchCap;
+ }
+
+ @Override
+ protected void ioWait(int timeout, int op) {
+ Assert.fail("scripted socket never blocks");
+ }
+
+ @Override
+ protected void setupIoWait() {
+ }
+ }
+
+ private static class ScriptedSocket implements Socket {
+ private final int bytesBeforeFailure;
+ private int acceptedBytes;
+ private int sendCalls;
+
+ private ScriptedSocket(int bytesBeforeFailure) {
+ this.bytesBeforeFailure = bytesBeforeFailure;
+ }
+
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public int getFd() {
+ return -1;
+ }
+
+ @Override
+ public boolean isClosed() {
+ return false;
+ }
+
+ @Override
+ public void of(int fd) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int recv(long bufferPtr, int bufferLen) {
+ return 0;
+ }
+
+ @Override
+ public int send(long bufferPtr, int bufferLen) {
+ sendCalls++;
+ if (acceptedBytes == bytesBeforeFailure) {
+ return -1;
+ }
+ int count = Math.min(bufferLen, bytesBeforeFailure - acceptedBytes);
+ acceptedBytes += count;
+ return count;
+ }
+
+ @Override
+ public void startTlsSession(CharSequence peerName, SocketReadinessWaiter waiter) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean supportsTls() {
+ return false;
+ }
+
+ @Override
+ public int tlsIO(int readinessFlags) {
+ return 0;
+ }
+
+ @Override
+ public boolean wantsTlsWrite() {
+ return false;
+ }
+ }
+}
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopRotationRaceTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopRotationRaceTest.java
index b21a018b..f1fd7e0f 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopRotationRaceTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopRotationRaceTest.java
@@ -33,7 +33,9 @@
import io.questdb.client.cutlass.qwp.client.sf.cursor.MmapSegment;
import io.questdb.client.cutlass.qwp.client.sf.cursor.SegmentBudget;
import io.questdb.client.cutlass.qwp.client.sf.cursor.SegmentManager;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.SfManifest;
import io.questdb.client.network.PlainSocketFactory;
+import io.questdb.client.std.FilesFacade;
import io.questdb.client.std.MemoryTag;
import io.questdb.client.std.Unsafe;
import io.questdb.client.test.tools.TestUtils;
@@ -41,8 +43,8 @@
import org.junit.Test;
/**
- * Probabilistic regression test for the send-loop rotation race fixed in
- * {@code CursorWebSocketSendLoop.trySendOne()} (PR #41): the loop could observe
+ * Regression tests for segment handoffs. The live-rotation test covers the race fixed in
+ * {@code CursorWebSocketSendLoop.trySendBatch()} (PR #41): the loop could observe
* a segment rotation without observing the final frame published into that
* segment, and advance past it — silently dropping the frame.
*
@@ -85,7 +87,7 @@ public void tailFrameSurvivesSegmentRotation() throws Exception {
SegmentManager manager = new SegmentManager(
SEGMENT_BYTES, new SegmentBudget(SegmentManager.UNLIMITED_TOTAL_BYTES));
manager.start();
- // sfDir == null => memory-only segments (no files/fds); same trySendOne() path.
+ // sfDir == null => memory-only segments (no files/fds); same trySendBatch() path.
CursorSendEngine engine = new CursorSendEngine(null, SEGMENT_BYTES, manager);
RecordingClient client = new RecordingClient(n);
CursorWebSocketSendLoop loop = new CursorWebSocketSendLoop(
@@ -119,6 +121,44 @@ public void tailFrameSurvivesSegmentRotation() throws Exception {
// The I/O thread is joined, so its writes to recorded[] are visible here.
assertContiguous(client, n);
+ Assert.assertEquals(n, loop.getTotalFramesSent());
+ });
+ }
+
+ @Test
+ public void testRecoveredSealedSegmentsFlushBeforeEmptyActive() throws Exception {
+ TestUtils.assertMemoryLeak(() -> {
+ final int sealedSegments = 3;
+ final int framesPerSegment = 20;
+ final int frames = sealedSegments * framesPerSegment;
+ String sfDir = TestUtils.createTmpDir("qdb-recovered-batch-");
+ try {
+ writeRecoveredSegments(sfDir, sealedSegments, framesPerSegment);
+ try (CursorSendEngine engine = new CursorSendEngine(sfDir, SEGMENT_BYTES);
+ RecordingClient client = new RecordingClient(frames);
+ CursorWebSocketSendLoop loop = new CursorWebSocketSendLoop(
+ client, engine, 0L, null, 0L, 0L)) {
+ Assert.assertTrue(engine.wasRecoveredFromDisk());
+ Assert.assertEquals(frames - 1, engine.publishedFsn());
+ Assert.assertEquals(frames, engine.activeSegment().baseSeq());
+ Assert.assertEquals(MmapSegment.HEADER_SIZE, engine.activeSegment().publishedOffset());
+ loop.setRunningForTest(true);
+ loop.positionCursorForStartForTest();
+
+ // Each sealed segment needs a send pass and an advance pass.
+ // No producer publishes into the empty active to trigger a later flush.
+ for (int i = 0; i < 2 * sealedSegments + 1; i++) {
+ loop.trySendBatchForTest();
+ loop.checkError();
+ }
+
+ assertContiguous(client, frames);
+ Assert.assertEquals(frames, loop.getTotalFramesSent());
+ Assert.assertFalse(loop.trySendBatchForTest());
+ }
+ } finally {
+ TestUtils.removeTmpDirRec(sfDir);
+ }
});
}
@@ -126,25 +166,48 @@ private static void assertContiguous(RecordingClient client, int expected) {
int count = client.recordedCount;
for (int i = 0; i < count; i++) {
if (client.recorded[i] != i) {
- Assert.fail("rotation race reordered/dropped a frame — "
+ Assert.fail("send loop reordered/dropped a frame — "
+ "fsn " + i + " on the wire was " + client.recorded[i]);
}
}
if (count != expected) {
- Assert.fail("rotation race dropped a frame — expected " + expected
+ Assert.fail("send loop dropped a frame — expected " + expected
+ " frames on the wire, got " + count);
}
}
+ private static void writeRecoveredSegments(String sfDir, int sealedSegments, int framesPerSegment) {
+ long payload = Unsafe.malloc(PAYLOAD_LEN, MemoryTag.NATIVE_DEFAULT);
+ try {
+ for (int s = 0; s <= sealedSegments; s++) {
+ String name = s == 0 ? "sf-initial.sfa" : String.format("sf-%016x.sfa", s);
+ long baseFsn = (long) s * framesPerSegment;
+ try (MmapSegment segment = MmapSegment.create(sfDir + "/" + name, baseFsn, SEGMENT_BYTES)) {
+ // The last segment is the empty active left by a crash after rotation.
+ if (s < sealedSegments) {
+ for (int i = 0; i < framesPerSegment; i++) {
+ Unsafe.getUnsafe().putLong(payload, baseFsn + i);
+ Assert.assertTrue(segment.tryAppend(payload, PAYLOAD_LEN) >= 0);
+ }
+ }
+ }
+ }
+ SfManifest.create(FilesFacade.INSTANCE, sfDir, 0, (long) sealedSegments * framesPerSegment).close();
+ } finally {
+ Unsafe.free(payload, PAYLOAD_LEN, MemoryTag.NATIVE_DEFAULT);
+ }
+ }
+
/**
- * In-memory transport, and nothing more: {@code sendBinary} captures the fsn
- * carried by each frame the loop ships. The other callbacks are inert — this
+ * In-memory transport: {@code batchAppendBinary} buffers each fsn and
+ * {@code batchFlush} publishes the delivered count. The other callbacks are inert — this
* test never receives ACKs and never reconnects. Only {@code sentCount} is
* read cross-thread (by the test, for progress) and so is volatile.
*/
private static final class RecordingClient extends TestWebSocketClient {
final long[] recorded;
volatile long sentCount;
+ private int appendedCount;
private int recordedCount;
RecordingClient(int capacity) {
@@ -153,10 +216,14 @@ private static final class RecordingClient extends TestWebSocketClient {
}
@Override
- public void sendBinary(long dataPtr, int length) {
- if (recordedCount < recorded.length) {
- recorded[recordedCount++] = Unsafe.getUnsafe().getLong(dataPtr);
- }
+ public void batchAppendBinary(long dataPtr, int length) {
+ Assert.assertTrue("more frames than expected", appendedCount < recorded.length);
+ recorded[appendedCount++] = Unsafe.getUnsafe().getLong(dataPtr);
+ }
+
+ @Override
+ public void batchFlush() {
+ recordedCount = appendedCount;
sentCount = recordedCount;
}
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopTornDictGuardTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopTornDictGuardTest.java
index c47c06cd..343a7036 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopTornDictGuardTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopTornDictGuardTest.java
@@ -52,7 +52,7 @@
/**
* White-box coverage of the send loop's torn-dictionary guard
- * ({@code trySendOne}: {@code deltaStart > sentDictCount}) firing on a genuine gap.
+ * ({@code trySendBatch}: {@code deltaStart > sentDictCount}) firing on a genuine gap.
*
* That guard is the drainer path's SOLE defense: an orphan drainer adopts a slot
* without running {@code Sender.build()}'s seed-time guard, so on a slot whose
@@ -84,6 +84,48 @@ public void tearDown() {
TestUtils.removeTmpDirRec(sfDir);
}
+ /**
+ * A gapped frame behind good frames ends the batch BEFORE it: the good frames
+ * are flushed first, and the guard fires on the next pass with nothing
+ * pending -- as when each frame had its own send. Firing mid-batch would latch
+ * the terminal with the good frames appended but never written.
+ */
+ @Test
+ public void testGuardFiresOnlyAfterFramesAheadOfTheGapAreFlushed() throws Exception {
+ assertMemoryLeak(() -> {
+ CountingClient client = new CountingClient();
+ try (CursorSendEngine engine = new CursorSendEngine(null, 16_384)) {
+ CursorWebSocketSendLoopCatchUpAlignmentTest.appendDeltaDictFrame(engine, 0, 'a');
+ CursorWebSocketSendLoopCatchUpAlignmentTest.appendDeltaDictFrame(engine, 1, 'b');
+ // the server only ever learns ids 0..1, so a delta starting at 5 is a gap
+ CursorWebSocketSendLoopCatchUpAlignmentTest.appendDeltaDictFrame(engine, 5, 'f');
+ CursorWebSocketSendLoop loop = newLoop(engine, client);
+ try {
+ loop.setRunningForTest(true);
+ loop.positionCursorForStartForTest();
+
+ Assert.assertTrue("the frames ahead of the gap must be sent", loop.trySendBatchForTest());
+ Assert.assertEquals(2, client.framesSent);
+ Assert.assertEquals("the frames ahead of the gap must be flushed", 1, client.flushes);
+ Assert.assertEquals(2, loop.getTotalFramesSent());
+ loop.checkError(); // no terminal yet
+
+ Assert.assertFalse("the gapped frame must be refused", loop.trySendBatchForTest());
+ Assert.assertEquals("no frame may reach the server through a gap", 2, client.framesSent);
+ try {
+ loop.checkError();
+ Assert.fail("the guard must latch a terminal error");
+ } catch (LineSenderException e) {
+ Assert.assertTrue("unexpected terminal: " + e.getMessage(),
+ e.getMessage().contains("incomplete"));
+ }
+ } finally {
+ loop.close();
+ }
+ }
+ });
+ }
+
@Test
public void testGuardFiresOnGenuineGapAndShipsNoFrame() throws Exception {
assertMemoryLeak(() -> {
@@ -104,7 +146,7 @@ public void testGuardFiresOnGenuineGapAndShipsNoFrame() throws Exception {
// before its first send (no catch-up: the mirror is empty).
loop.positionCursorForStartForTest();
- boolean sent = loop.trySendOneForTest();
+ boolean sent = loop.trySendBatchForTest();
Assert.assertFalse("the torn-dict guard must refuse to send the gapped frame", sent);
Assert.assertEquals("no frame may reach the server through a gap",
@@ -138,7 +180,7 @@ private static int countSegmentFiles(Path dir) {
}
// Constructs a recovery send loop that is never started -- the test drives
- // positionCursorForStart + trySendOne directly. The reconnect factory throws
+ // positionCursorForStart + trySendBatch directly. The reconnect factory throws
// because no reconnect is expected before the guard latches its terminal.
private CursorWebSocketSendLoop newLoop(CursorSendEngine engine, WebSocketClient client) {
return new CursorWebSocketSendLoop(
@@ -185,8 +227,9 @@ private void writeAndTearGappedSlot() throws Exception {
}
// Frame-counting stub transport: completes no real I/O. If the guard ever lets a
- // gapped frame through, sendBinary bumps framesSent and the test fails.
+ // gapped frame through, batchAppendBinary bumps framesSent and the test fails.
private static final class CountingClient extends TestWebSocketClient {
+ private int flushes;
private int framesSent;
CountingClient() {
@@ -204,10 +247,15 @@ public int getServerQwpVersion() {
}
@Override
- public void sendBinary(long dataPtr, int length) {
+ public void batchAppendBinary(long dataPtr, int length) {
framesSent++;
}
+ @Override
+ public void batchFlush() {
+ flushes++;
+ }
+
@Override
protected void ioWait(int timeout, int op) {
}
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;
+ }
+ }
}
diff --git a/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java b/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java
index 39695ec2..e4b94d8f 100644
--- a/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java
+++ b/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java
@@ -54,8 +54,7 @@ public void testEveryIngressKeyIsHonored() {
assertHonored("auto_flush_interval=250", "auto_flush_interval", 250);
// auto_flush=off reaches disableAutoFlush(), which sets the interval to
// MAX_VALUE; auto_flush_rows=off leaves the interval unset, so this field
- // distinguishes the two. That config is build-rejected on WebSocket; see
- // QuestDBBuilderTest.testMalformedIngressConfigRejectedAtBuildWithMinZero.
+ // distinguishes the two.
assertHonored("auto_flush=off", "auto_flush_interval", Integer.MAX_VALUE);
assertHonored("max_name_len=99", "max_name_len", 99);
assertHonored("transaction=on", "transaction", true);