diff --git a/core/src/main/java/io/questdb/client/QuestDB.java b/core/src/main/java/io/questdb/client/QuestDB.java index 8cfc437b..02a0cb6d 100644 --- a/core/src/main/java/io/questdb/client/QuestDB.java +++ b/core/src/main/java/io/questdb/client/QuestDB.java @@ -139,7 +139,11 @@ static QuestDB connect(CharSequence configurationString, HttpTokenProvider token * {@link Sender#close()} on the returned instance to release it back to * the pool. {@code close()} on a pooled Sender flushes pending rows * before returning to the pool; a real disconnect only happens at - * {@link #close()} on this {@code QuestDB} handle. + * {@link #close()} on this {@code QuestDB} handle. A Sender in strict + * transactional mode ({@code transaction=strict}, or a WebSocket config + * with no auto-flush trigger, such as {@code auto_flush=off}) commits only + * on {@link Sender#flush()}, so its {@code close()} discards the rows it + * did not flush instead. *
* Allocation: zero at steady state -- the returned instance is a
* pre-allocated decorator backed by a pre-allocated underlying Sender.
diff --git a/core/src/main/java/io/questdb/client/Sender.java b/core/src/main/java/io/questdb/client/Sender.java
index d16e9780..fb21442f 100644
--- a/core/src/main/java/io/questdb/client/Sender.java
+++ b/core/src/main/java/io/questdb/client/Sender.java
@@ -773,13 +773,46 @@ default void resetSymbolDictionary() {
* data format) and want to continue sending new data without retrying the problematic data.
* After calling this method, you can start building a new row by calling {@link #table(CharSequence)}.
*
- * Note: This method is only available for HTTP transport. TCP transport doesn't support
- * this operation.
+ * In WebSocket transactional mode ({@code transaction=on} or {@code strict}) it also
+ * discards the rows of the open transaction that auto-flush already sent, as
+ * {@link #rollback()} does, when the server supports QWP rollback.
+ *
+ * Note: the ILP TCP and UDP transports don't support this operation.
*
* @see #flush()
*/
void reset();
+ /**
+ * Discard the open transaction: the rows buffered since the last flush and, in
+ * WebSocket transactional mode ({@code transaction=on} or {@code strict}), the
+ * rows of the transaction that auto-flush already sent to the server, which
+ * the server discards on request.
+ *
+ * Like {@link #flush()}, this returns without waiting for the server, so a
+ * server outage makes it fail only when the store-and-forward buffer is full,
+ * as it would a flush. The request goes out ahead of anything sent later, so
+ * later rows never commit the discarded ones, even across reconnects. Call
+ * {@link #drain(long)} to wait until the server confirms.
+ *
+ * A {@code transaction=on} transaction that already committed in parts (see
+ * {@link LineSenderBuilder#transactional(boolean)}) keeps the parts it committed;
+ * this discards only the rest.
+ *
+ * Discarding rows already sent needs a server that supports QWP rollback. When
+ * the server does not, this method throws and aborts the transaction: every
+ * later flush throws too, and closing the sender discards the rows, because the
+ * server rolls back uncommitted rows when the connection closes.
+ *
+ * Only the WebSocket transport supports this method.
+ *
+ * @throws LineSenderException if the transport does not support rollback, or the
+ * server cannot discard rows already sent
+ */
+ default void rollback() {
+ throw new LineSenderException("rollback is only supported for WebSocket transport");
+ }
+
/**
* Add a column with a 16-bit signed integer value.
*
@@ -1075,7 +1108,10 @@ final class LineSenderBuilder {
private static final int DEFAULT_TCP_PORT = 9009;
private static final int DEFAULT_UDP_PORT = 9007;
private static final int DEFAULT_WEBSOCKET_PORT = 9000;
- private static final int DEFAULT_WS_AUTO_FLUSH_BYTES = 0;
+ // Unset auto_flush_bytes on WebSocket sizes the byte trigger from what one
+ // frame can carry, so no batch outgrows a frame; see
+ // QwpWebSocketSender.AUTO_FLUSH_BYTES_AUTO.
+ private static final int DEFAULT_WS_AUTO_FLUSH_BYTES = QwpWebSocketSender.AUTO_FLUSH_BYTES_AUTO;
private static final long DEFAULT_WS_AUTO_FLUSH_INTERVAL_NANOS = 100_000_000L; // 100ms
private static final int DEFAULT_WS_AUTO_FLUSH_ROWS = 1_000;
// Cadence at which the SF cursor I/O loop sends a keepalive PING
@@ -1118,10 +1154,12 @@ final class LineSenderBuilder {
private final IntList ports = new IntList();
private long authTimeoutMillis = QwpWebSocketSender.DEFAULT_AUTH_TIMEOUT_MS;
private int autoFlushBytes = PARAMETER_NOT_SET_EXPLICITLY;
- // Set by disableAutoFlush(). auto_flush_interval=off also stores the interval as
- // Integer.MAX_VALUE, so the WebSocket validation keys on this flag to reject only the
- // full disable and accept the interval-only opt-out.
- private boolean autoFlushDisabled;
+ // disableAutoFlush() (auto_flush=off) sets it. That call turns off every
+ // auto-flush trigger, so a positive auto_flush_bytes contradicts it, and only
+ // that conflict check reads this flag. auto_flush_rows=off with
+ // auto_flush_interval=off reaches the same trigger state, which is what
+ // hasAutoFlushTrigger() reads to pick the sender's mode.
+ private boolean isAutoFlushDisabled;
private int autoFlushIntervalMillis = PARAMETER_NOT_SET_EXPLICITLY;
private int autoFlushRows = PARAMETER_NOT_SET_EXPLICITLY;
private int bufferCapacity = PARAMETER_NOT_SET_EXPLICITLY;
@@ -1234,7 +1272,12 @@ public int getConnectTimeout() {
private long reconnectMaxDurationMillis = PARAMETER_NOT_SET_EXPLICITLY;
private int durableAckTiers = DurableAckTiers.NONE;
private int retryTimeoutMillis = PARAMETER_NOT_SET_EXPLICITLY;
+ private boolean strictTransactional;
private boolean transactional;
+ // True once transaction=... (or transactional()/strictTransactional())
+ // set the mode explicitly, so auto_flush=off can refuse to override an
+ // explicit transaction=off.
+ private boolean transactionModeSet;
private String senderId = DEFAULT_SENDER_ID;
// Per-append deadline for SF appendBlocking spin-then-throw. Used to
// be a hardcoded 30s constant; expose so tight-SLA users can lower
@@ -1385,7 +1428,13 @@ public LineSenderBuilder authTimeoutMillis(long millis) {
*
* This is only used when communicating over WebSocket transport.
*
- * Default value is 0, which disables byte-based auto-flush.
+ * By default the byte trigger fires at 90% of the largest frame the connection can carry: the server's
+ * advertised maximum batch size, or the store-and-forward segment when that is smaller. A positive value sets
+ * the trigger, capped at that size, and counts as an auto-flush trigger: unless the sender is transactional,
+ * every batch it sends commits. 0 turns the byte trigger off, and then every batch must fit one frame per
+ * table.
+ *
+ * You cannot set a positive value when auto-flush is disabled. See {@link #disableAutoFlush()}.
*
* @param bytes maximum bytes per batch
* @return this instance for method chaining
@@ -1397,6 +1446,8 @@ public LineSenderBuilder autoFlushBytes(int bytes) {
if (this.autoFlushBytes != PARAMETER_NOT_SET_EXPLICITLY) {
throw new LineSenderException("auto flush bytes was already configured")
.put("[bytes=").put(this.autoFlushBytes).put("]");
+ } else if (isAutoFlushDisabled && bytes > 0) {
+ throw new LineSenderException("cannot set auto flush bytes when auto-flush is already disabled");
}
if (bytes < 0) {
throw new LineSenderException("auto flush bytes cannot be negative")
@@ -1455,7 +1506,9 @@ public LineSenderBuilder autoFlushIntervalMillis(int autoFlushIntervalMillis) {
* Setting this to 1 means that the Sender will send each row to a server immediately after it is added. This
* effectively disables batching and may lead to a significant performance degradation.
*
- * Setting this to 0 disables row-based auto-flush. Interval-based auto-flush remains enabled.
+ * Setting this to 0 disables row-based auto-flush. Interval-based auto-flush remains enabled. On WebSocket, a
+ * sender whose interval trigger is off too, and that has no positive {@link #autoFlushBytes(int)}, has no
+ * auto-flush trigger left and works like one built with {@link #disableAutoFlush()}.
*
* You cannot set this value when auto-flush is disabled. See {@link #disableAutoFlush()}. * @@ -1868,9 +1921,17 @@ public LineSenderBuilder connectionListenerInboxCapacity(int capacity) { *
* 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 or WebSocket protocol. On WebSocket, {@code auto_flush_interval=off} turns off - * only the time-based trigger. + * This method can be used with the HTTP and WebSocket protocols; calling it is illegal when communicating over + * the TCP protocol. Turning off both the row and the interval trigger ({@code auto_flush_rows=off} and + * {@code auto_flush_interval=off}) has the same effect. + *
+ * On WebSocket, a sender with no auto-flush trigger is in strict transactional mode (see + * {@link #strictTransactional(boolean)}) unless {@link #transactional(boolean)} selects the mode: rows still go + * to the server in frames sized for the connection, so a batch can be any size, but only {@code flush()} + * commits them, and closing the sender or returning a pooled sender without a flush discards them. + * {@code transactional(true)} keeps such a sender non-strict, so closing it commits, and + * {@code transactional(false)} cannot be combined with it. Nor can a positive {@link #autoFlushBytes(int)}, + * which is an auto-flush trigger. * * @return this instance for method chaining * @see #autoFlushRows(int) @@ -1885,10 +1946,14 @@ public LineSenderBuilder disableAutoFlush() { throw new LineSenderException("auto flush interval was already configured ") .put("[autoFlushIntervalMillis=").put(this.autoFlushIntervalMillis).put("]"); } + if (this.autoFlushBytes > 0) { + throw new LineSenderException("auto flush bytes was already configured ") + .put("[autoFlushBytes=").put(this.autoFlushBytes).put("]"); + } this.autoFlushRows = AUTO_FLUSH_DISABLED; this.autoFlushIntervalMillis = Integer.MAX_VALUE; - this.autoFlushDisabled = true; + this.isAutoFlushDisabled = true; return this; } @@ -2757,12 +2822,54 @@ private LineSenderBuilder requestDurableAckTiers(int tiers) { return this; } + /** + * Enables strict transactional mode, the builder form of + * {@code transaction=strict}. Like {@link #transactional(boolean)}, auto-flush + * sends rows to the server without committing them and {@code flush()} + * commits, so a transaction can be any size. In addition the transaction + * never commits in parts: a row that would take a table past the server's + * {@code qwp.max.uncommitted.rows}, or the transaction past + * {@code sf_max_total_bytes}, throws + * {@link io.questdb.client.cutlass.qwp.client.TransactionTooLargeException} + * and is not added, and {@code close()} without {@code flush()} discards the + * transaction instead of committing it. + * A sender with no auto-flush trigger (see {@link #disableAutoFlush()}) is + * in this mode unless {@link #transactional(boolean)} selects the mode. + * WebSocket transport only. + * + * @param enabled true to enable strict transactional mode; false leaves + * {@link #transactional(boolean) transactional} mode as set + * @return this instance for method chaining + */ + public LineSenderBuilder strictTransactional(boolean enabled) { + if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) { + throw new LineSenderException("transactional is only supported for WebSocket transport"); + } + this.strictTransactional = enabled; + if (enabled) { + this.transactional = true; + } + this.transactionModeSet = true; + return this; + } + /** * Enables transactional mode. Auto-flush sends data to the server * with {@code FLAG_DEFER_COMMIT}; only an explicit {@code flush()} - * triggers the server-side WAL commit. This allows accumulating - * datasets larger than the server's recv buffer while committing - * atomically per table. + * (or {@code close()}) triggers the server-side WAL commit. This allows + * accumulating datasets larger than the server's recv buffer while + * committing atomically per table. A transaction too large for either side + * commits in parts: the server commits a table on its own once + * {@code qwp.max.uncommitted.rows} of its rows are uncommitted, and the + * sender, which holds the transaction in {@code sf_max_total_bytes} until + * the server commits it, commits the rows it already sent once the + * transaction outgrows that buffer, and logs a warning. Use + * {@link #strictTransactional(boolean)} to fail instead. + *
+ * A sender with no auto-flush trigger (see {@link #disableAutoFlush()}) is
+ * in strict transactional mode unless this method selects the mode:
+ * {@code true} keeps it in this non-strict mode, and {@code false} cannot
+ * be combined with it.
*
* @param enabled true to enable transactional mode
* @return this instance for method chaining
@@ -2772,6 +2879,10 @@ public LineSenderBuilder transactional(boolean enabled) {
throw new LineSenderException("transactional is only supported for WebSocket transport");
}
this.transactional = enabled;
+ if (!enabled) {
+ this.strictTransactional = false;
+ }
+ this.transactionModeSet = true;
return this;
}
@@ -3822,6 +3933,11 @@ private QwpWebSocketSender buildWebSocket(HttpTokenProvider tokenProvider) {
// dispatcher daemon, drainer pool, microbatch buffers and
// WebSocketClient inside the abandoned `connected`.
connected.setTransactional(transactional);
+ // With no auto-flush trigger only flush() commits: a strict transaction of
+ // any size, unless transaction= chose the mode.
+ if (strictTransactional || (!transactionModeSet && !hasAutoFlushTrigger())) {
+ connected.setStrictTransactional(true);
+ }
try {
final String rebuildSfDir = sfDir;
final String rebuildSenderId = senderId;
@@ -4230,8 +4346,10 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
transactional(true);
} else if (Chars.equalsIgnoreCase("off", sink)) {
transactional(false);
+ } else if (Chars.equalsIgnoreCase("strict", sink)) {
+ strictTransactional(true);
} else {
- throw new LineSenderException("invalid transaction [value=").put(sink).put(", allowed-values=[on, off]]");
+ throw new LineSenderException("invalid transaction [value=").put(sink).put(", allowed-values=[on, off, strict]]");
}
} else if (Chars.equals("sf_dir", sink)) {
if (protocol != PROTOCOL_WEBSOCKET) {
@@ -4682,8 +4800,10 @@ private LineSenderBuilder fromConfigWebSocket(CharSequence configurationString)
transactional(true);
} else if (s.equalsIgnoreCase("off")) {
transactional(false);
+ } else if (s.equalsIgnoreCase("strict")) {
+ strictTransactional(true);
} else {
- throw new LineSenderException("invalid transaction [value=").put(s).put(", allowed-values=[on, off]]");
+ throw new LineSenderException("invalid transaction [value=").put(s).put(", allowed-values=[on, off, strict]]");
}
}
s = view.getStr("request_durable_ack");
@@ -4821,7 +4941,7 @@ public java.util.Map
+ * Like {@link #flush()}, it hands the rollback frame to store-and-forward and
+ * returns without waiting for the server, so a server outage makes it fail
+ * only when the store-and-forward buffer is full, as it would a flush. The
+ * frame goes out behind the transaction's frames and ahead of anything sent
+ * later, on every connection and after a restart, so nothing sent later can
+ * commit the discarded rows. Call {@link #drain(long)} to wait for the server
+ * to confirm.
+ *
+ * A transaction=on transaction that already committed in parts, at the
+ * client buffer or the server's {@code qwp.max.uncommitted.rows}, keeps the
+ * parts it committed; this discards only the rest.
+ *
+ * @throws LineSenderException when frames of the open transaction were already
+ * sent and the server does not support QWP rollback
+ * (the transaction is then aborted: the next flush
+ * throws, and only closing the sender ends it)
+ */
+ @Override
+ public void rollback() {
+ checkNotClosed();
+ checkRecycleFailure();
+ discardPendingRows();
+ if (txnAborted) {
+ throw transactionAbortedError();
+ }
+ if (!hasDeferredMessages) {
+ clearTransactionState();
+ return;
+ }
+ if (!serverRollbackEnabled) {
+ txnAborted = true;
+ throw transactionAbortedError();
+ }
+ rollbackOpenTransaction("rollback()");
+ }
+
+ /**
+ * Pool hook for a strict-mode lease being returned: discards what the
+ * borrower left uncommitted, so the next borrower starts with no open
+ * transaction. Returns {@code false} when frames already sent could not be
+ * rolled back -- the server does not support QWP rollback, or the transaction
+ * was already aborted -- and the pool must retire the connection instead; the
+ * server rolls those frames back when the connection closes.
+ */
+ public boolean discardTransactionForReuse() {
+ checkNotClosed();
+ if (!txnAborted) {
+ warnDiscardedTransaction("returning the sender to the pool");
+ }
+ discardPendingRows();
+ if (txnAborted) {
+ return false;
+ }
+ if (!hasDeferredMessages) {
+ clearTransactionState();
+ return true;
+ }
+ if (!serverRollbackEnabled) {
+ // Only closing the connection discards the frames already sent, so the
+ // transaction is over. Aborting it also keeps the close() that retires
+ // the connection from warning about the same rows again.
+ txnAborted = true;
+ return false;
+ }
+ rollbackOpenTransaction("returning the sender to the pool");
+ return true;
+ }
+
+ private void discardPendingRows() {
// Reset ALL table buffers, not just the current one
ObjList
+ * No wait is needed to keep the rolled-back frames from ever committing. The
+ * store-and-forward log orders the rollback frame ahead of every frame
+ * published later, and every connection replays the log in that order, so a
+ * server that receives a later commit has rolled the transaction back first.
+ * Only a server that does not support rollback, reached by a reconnect
+ * before the frame was acked, could commit them: it would read the frame as
+ * an empty deferred frame. The I/O loop never sends it there; it discards
+ * the transaction by closing that connection instead (see
+ * {@code CursorWebSocketSendLoop.retireRollbackFrame}). The same holds for
+ * frames replayed after a restart or by a background drainer, which use the
+ * same loop.
+ *
+ * The rows a transaction=on transaction committed in parts stay committed;
+ * {@code action} names the call in the warning that says so.
+ */
+ private void rollbackOpenTransaction(String action) {
+ long committedRows = txnRowsCommittedInParts;
+ publishRollbackFrame();
+ hasDeferredMessages = false;
+ // The rollback frame gets an OK ack covering the transaction's frames, or the
+ // I/O loop acknowledges it when it discards the transaction itself, so
+ // close-time drains may wait for it like for a commit.
+ lastCommitBoundaryFsn = cursorEngine.publishedFsn();
+ clearTransactionState();
+ if (committedRows > 0) {
+ LOG.warn("{} discarded the rows of the open transaction sent since the sender last committed part of "
+ + "it, but the {} rows it committed when the transaction outgrew the client buffer stay "
+ + "committed", action, committedRows);
+ }
+ }
+
/**
* Advisory request to start a fresh symbol-dictionary epoch. Sets
* {@link #manualResetRequested}; if no flush is in flight
@@ -3400,6 +3707,8 @@ public void setCursorEngine(CursorSendEngine engine, boolean takeOwnership) {
engine.adoptCounters(counters);
engine.setSlotLockReleaseListener(this::onSlotLockReleased);
}
+ // The segment size bounds every frame; re-size the byte trigger for it.
+ recomputeFrameBudget();
}
/**
@@ -3549,8 +3858,37 @@ public void setRecycleDeferredCloseMaxWaitMillisForTesting(long millis) {
this.recycleDeferredCloseMaxWaitMillis = millis;
}
+ /**
+ * Strict transactional mode: auto-flush ships rows as FLAG_DEFER_COMMIT frames
+ * like {@link #setTransactional(boolean) transactional} mode, only an explicit
+ * flush commits, and the open transaction never commits in parts. A row that
+ * would push a table past the server's {@code qwp.max.uncommitted.rows}, or the
+ * open transaction past what the client buffer can hold alongside the commit,
+ * throws {@link TransactionTooLargeException} instead and is not added.
+ * {@link #close()} discards an uncommitted transaction rather than committing
+ * it. Enabling it enables transactional mode; disabling it leaves transactional
+ * mode on.
+ */
+ public void setStrictTransactional(boolean strict) {
+ this.strictTransactional = strict;
+ if (strict) {
+ this.transactional = true;
+ }
+ }
+
+ /**
+ * Transactional mode: auto-flush ships rows as FLAG_DEFER_COMMIT frames and
+ * only an explicit flush (or close) commits. A transaction that outgrows the
+ * client buffer commits in parts: before a flush that does not fit beside the
+ * frames already sent, the sender commits those frames and logs a warning.
+ * Disabling it also disables {@link #setStrictTransactional(boolean) strict}
+ * mode, which refuses rows instead.
+ */
public void setTransactional(boolean transactional) {
this.transactional = transactional;
+ if (!transactional) {
+ this.strictTransactional = false;
+ }
}
/**
@@ -3886,44 +4224,30 @@ private static List
- * Always updates {@link #serverMaxBatchSize} so the single-row hard guard
- * in {@link #sendRow} fires against the freshly advertised value. The
- * byte trigger, however, is only adjusted when the user left it enabled:
- * an explicit {@code auto_flush_bytes=off} (autoFlushBytes == 0) is
- * preserved even when the server advertises a cap, so applications that
- * opted out keep the contract they asked for.
+ * Always updates {@link #serverMaxBatchSize}, so the single-row guard in
+ * {@link #sendRow} fires against the freshly advertised value. See
+ * {@link #recomputeFrameBudget()} for the byte trigger.
*/
@TestOnly
public void applyServerBatchSizeLimit(int advertisedMaxBatchSize) {
serverMaxBatchSize = advertisedMaxBatchSize;
- if (autoFlushBytes <= 0) {
- // User opted out of byte-based auto-flush; respect that even when
- // the server advertises a cap. The single-row guard still protects
- // against oversize individual rows via serverMaxBatchSize.
- effectiveAutoFlushBytes = 0;
- return;
- }
- if (advertisedMaxBatchSize <= 0) {
- effectiveAutoFlushBytes = autoFlushBytes;
- return;
- }
- // Cap at 90% of the server's hard limit. Raw column-buffer bytes are
- // a conservative proxy for wire size (compression usually shrinks the
- // payload), but schema and dict-delta overhead can push the wire size
- // above the raw total in pathological cases.
- long safeBudget = (long) advertisedMaxBatchSize * 9 / 10;
- if (autoFlushBytes < safeBudget) {
- effectiveAutoFlushBytes = autoFlushBytes;
- } else {
- effectiveAutoFlushBytes = (int) safeBudget;
- }
+ recomputeFrameBudget();
+ }
+
+ /**
+ * Records the transaction limits the server advertised on the last
+ * foreground handshake: its per-table {@code qwp.max.uncommitted.rows}
+ * ({@code 0} when not advertised) and whether it confirmed rollback support.
+ */
+ @TestOnly
+ public void applyServerTransactionLimits(long maxUncommittedRows, boolean rollbackEnabled) {
+ serverMaxUncommittedRows = maxUncommittedRows;
+ serverRollbackEnabled = rollbackEnabled;
}
private void atMicros(long timestampMicros) {
@@ -4256,6 +4580,9 @@ private WebSocketClient connectWalk(ReconnectSupplier ctx, CursorWebSocketSendLo
newClient.setQwpMaxVersion(QwpConstants.VERSION);
newClient.setQwpClientId(QwpConstants.CLIENT_ID);
newClient.setQwpDurableAckTiers(durableAckTiers);
+ // Ask every server whether it accepts the rollback control frame;
+ // only one that confirms ever receives it.
+ newClient.setQwpRequestRollback(true);
newClient.setConnectTimeout(effectiveConnectTimeoutMs(background, connectTimeoutMs));
if (cancellation != null) {
// Publish the client we are about to block on so a
@@ -4460,6 +4787,8 @@ private WebSocketClient connectWalk(ReconnectSupplier ctx, CursorWebSocketSendLo
// endpoint's cap; an oversize row then escapes the producer-side
// guard and trips a wire-level ws-close[1009] downstream.
applyServerBatchSizeLimit(newClient.getServerMaxBatchSize());
+ applyServerTransactionLimits(
+ newClient.getServerMaxUncommittedRows(), newClient.isServerRollbackEnabled());
return newClient;
} catch (Error e) {
closeQuietlyOnError(newClient);
@@ -5189,8 +5518,12 @@ private void ensureNoInProgressRow() {
* @param deferCommit when true, the message carries FLAG_DEFER_COMMIT
* so the server appends rows but does not commit.
* Used by auto-flush in transactional mode.
+ * @param isAutoFlush true when a row triggered the flush: when the client
+ * buffer cannot take it, a strict transaction keeps the
+ * rows pending instead of throwing, and refuses the next
+ * row.
*/
- private void flushPendingRows(boolean deferCommit) {
+ private void flushPendingRows(boolean deferCommit, boolean isAutoFlush) {
if (pendingRowCount <= 0) {
return;
}
@@ -5202,6 +5535,7 @@ private void flushPendingRows(boolean deferCommit) {
if (tableCount == 0) {
clearPendingTables();
pendingBytes = 0;
+ pendingSymbolDictBytes = 0;
currentTableBufferSnapshotBytes = currentTableBuffer == null
? 0 : currentTableBuffer.getBufferedBytes();
pendingRowCount = 0;
@@ -5233,6 +5567,39 @@ private void flushPendingRows(boolean deferCommit) {
int combinedBodyStart = encodeCombinedFrame(tableCount, deferCommit, deltaBaseline);
int messageSize = encoder.finishMessage();
+ // The client buffer holds a transaction's frames until the server commits it,
+ // so it cannot trim them to make room. Before any frame of this flush reaches
+ // the ring, check that the buffer holds the flush beside them.
+ if (transactional && cursorEngine != null) {
+ if (strictTransactional) {
+ // A strict transaction must be able to commit as one: refuse a flush
+ // the client buffer could not hold together with the commit that has
+ // to follow it.
+ if (!hasTransactionRoom(cap, deltaBaseline, messageSize, deferCommit)) {
+ if (deferCommit) {
+ txnBufferFull = true;
+ if (isAutoFlush) {
+ // The rows stay pending, the one that triggered this flush
+ // included, and the room kept for the commit holds them. The
+ // next row is refused, so a refused row is never part of the
+ // transaction, as at the server's row limit.
+ return;
+ }
+ }
+ throw transactionTooLargeForClientBuffer("so the flush was not sent");
+ }
+ } else if (txnSimSegments > 0 && !hasTransactionRoom(cap, deltaBaseline, messageSize, deferCommit)) {
+ // transaction=on commits in parts instead: commit the frames already
+ // sent, which lets the buffer trim them, and send this flush as the
+ // start of the next part. A flush that does not fit even alone goes
+ // out as it is.
+ commitTransactionPart();
+ // The commit frame went through the encoder: encode the batch again.
+ combinedBodyStart = encodeCombinedFrame(tableCount, deferCommit, deltaBaseline);
+ messageSize = encoder.finishMessage();
+ }
+ }
+
// Full-dict over-cap fallback. A full-dictionary frame carries the whole
// dictionary from id 0, so its fixed overhead grows with lifetime symbol
// cardinality: eventually the section alone fills the cap, and before that
@@ -5302,6 +5669,7 @@ && splitFramesFit(cap, currentBatchMaxSymbolId)) {
if (!deferCommit) {
lastCommitBoundaryFsn = cursorEngine.publishedFsn();
}
+ accountTransactionRows(deferCommit);
resetTableBuffersAfterFlush();
} catch (Throwable t) {
@@ -5531,6 +5899,7 @@ private void flushPendingRowsSplit(
// committed the whole group.
lastCommitBoundaryFsn = cursorEngine.publishedFsn();
}
+ accountTransactionRows(deferCommit);
resetTableBuffersAfterFlush();
}
@@ -5554,6 +5923,7 @@ private void resetTableBuffersAfterFlush() {
flushTableBuffers.setPos(0);
currentBatchMaxSymbolId = -1;
pendingBytes = 0;
+ pendingSymbolDictBytes = 0;
currentTableBufferSnapshotBytes = 0;
pendingRowCount = 0;
firstPendingRowTimeNanos = 0;
@@ -6461,6 +6831,7 @@ private void sendCommitMessage() {
sealAndSwapBuffer();
hasDeferredMessages = false;
lastCommitBoundaryFsn = cursorEngine.publishedFsn();
+ clearTransactionState();
}
private SenderErrorHandler userErrorHandler() {
@@ -7110,7 +7481,17 @@ private void sealAndSwapBuffer() {
// (durable mmap append, returns once published).
try {
toSend.markSending();
+ // Transactional mode tracks how the open transaction's frames pack
+ // into segments; the first frame's placement depends on the active
+ // segment's room before the append.
+ boolean isTxnFrame = transactional
+ && isTransactionFrame(toSend.getBufferPtr(), toSend.getBufferPos());
+ long activeRemaining = isTxnFrame && txnSimSegments == 0
+ ? cursorEngine.activeSegment().capacityRemaining() : 0;
cursorEngine.appendBlocking(toSend.getBufferPtr(), toSend.getBufferPos());
+ if (isTxnFrame) {
+ recordTransactionFrame(toSend.getBufferPos(), activeRemaining);
+ }
toSend.markRecycled();
} catch (Throwable t) {
// appendBlocking failed synchronously on the user thread — the
@@ -7135,6 +7516,315 @@ private void sealAndSwapBuffer() {
}
}
+ /**
+ * Adds the rows a just-published FLAG_DEFER_COMMIT flush shipped to each
+ * flushed table's open-transaction count, or forgets the open transaction
+ * when the flush committed it.
+ */
+ private void accountTransactionRows(boolean deferred) {
+ if (!deferred) {
+ clearTransactionState();
+ return;
+ }
+ for (int i = 0, n = flushTableBuffers.size(); i < n; i++) {
+ QwpTableBuffer table = flushTableBuffers.getQuick(i);
+ long sentRows = table.getTxnSentRows();
+ if (sentRows == 0) {
+ txnTables.add(table);
+ }
+ table.setTxnSentRows(sentRows + table.getRowCount());
+ }
+ }
+
+ private void checkTransactionAcceptsRow() {
+ if (txnAborted) {
+ throw transactionAbortedError();
+ }
+ if (!strictTransactional) {
+ return;
+ }
+ if (txnBufferFull) {
+ throw transactionTooLargeForClientBuffer("so the row was not added");
+ }
+ long maxRows = serverMaxUncommittedRows;
+ if (maxRows <= 0) {
+ maxRows = QwpConstants.DEFAULT_SERVER_MAX_UNCOMMITTED_ROWS;
+ }
+ // The server commits a table on its own once its uncommitted rows reach
+ // maxRows, so a strict transaction holds at most maxRows - 1 rows per table.
+ long rows = currentTableBuffer.getTxnSentRows() + currentTableBuffer.getRowCount() + 1;
+ if (rows >= maxRows) {
+ throw new TransactionTooLargeException("transaction too large for the server")
+ .put(" [table=").put(currentTableName)
+ .put(", rows=").put(rows)
+ .put(", maxUncommittedRows=").put(maxRows)
+ .put("]; the server commits a table's rows on its own once qwp.max.uncommitted.rows "
+ + "of them are uncommitted, which would split the transaction, so the row was "
+ + "not added. Raise qwp.max.uncommitted.rows on the server");
+ }
+ }
+
+ /**
+ * Forgets the open transaction: its tables' row counts, the client-buffer
+ * simulation and the buffer-full latch. Called when a frame commits it or
+ * the rollback frame discards it; O(tables in the transaction).
+ */
+ private void clearTransactionState() {
+ for (int i = 0, n = txnTables.size(); i < n; i++) {
+ txnTables.getQuick(i).setTxnSentRows(0);
+ }
+ txnTables.setPos(0);
+ txnBufferFull = false;
+ txnBytes = 0;
+ txnRefusedBytes = 0;
+ txnRowsCommittedInParts = 0;
+ txnSimSegments = 0;
+ txnSimRemaining = 0;
+ }
+
+ /**
+ * transaction=on: commits the frames of the open transaction already sent,
+ * because the next flush does not fit beside them in the client buffer, which
+ * cannot trim them before the server commits them. The transaction goes on
+ * with the rows sent from here on, so it commits in parts, as it does when the
+ * server commits a table at {@code qwp.max.uncommitted.rows}. Logs the first
+ * part each transaction commits this way.
+ */
+ private void commitTransactionPart() {
+ long rows = sentTransactionRows();
+ long bytes = txnBytes;
+ long committedRows = txnRowsCommittedInParts;
+ sendCommitMessage();
+ // sendCommitMessage() forgot the transaction, but it goes on.
+ txnRowsCommittedInParts = committedRows + rows;
+ if (committedRows == 0) {
+ LOG.warn("the open transaction outgrew the client buffer, so the sender committed the {} rows it had "
+ + "already sent, and the rest of the transaction commits separately [transactionBytes={}, "
+ + "ringCapacityBytes={}]. A transaction=on transaction commits in parts when "
+ + "sf_max_total_bytes cannot hold it: raise sf_max_total_bytes or set sf_dir to keep such "
+ + "transactions whole, or use transaction=strict to refuse their rows instead. Logged once "
+ + "per transaction",
+ rows, bytes, cursorEngine.ringCapacityBytes());
+ }
+ }
+
+ /**
+ * Transactional mode: whether the ring under {@code sf_max_total_bytes} can hold
+ * a flush's frames together with the open transaction's frames already on it
+ * and, for a deferred flush, room for the frame that ends the transaction. The
+ * server acks none of a transaction's frames before its commit, so the ring
+ * cannot trim them; a transaction that does not fit would block its own commit.
+ * A strict transaction keeps room for a commit that carries the rows still
+ * pending when it ends, a transaction=on transaction for a bare commit frame:
+ * it commits its frames with one when the next flush does not fit.
+ *
+ * Mirrors how the ring packs frames: frames never span segments, and a frame
+ * that does not fit the current segment opens the next one. The flush's
+ * frame sizes are derived the way flushPendingRows and the split publish
+ * them; dictionary chunks of the full-dictionary fallback count as full
+ * frames. Runs before anything of the flush is published, so a refusal
+ * leaves the batch pending and the ring untouched. On a strict refusal it
+ * records the size the transaction would have reached in
+ * {@link #txnRefusedBytes}.
+ */
+ private boolean hasTransactionRoom(int cap, int deltaBaseline, int messageSize, boolean deferCommit) {
+ CursorSendEngine engine = cursorEngine;
+ long segmentPayload = segmentPayloadCapacity();
+ transactionFrameSizes.clear();
+ if (cap <= 0 || messageSize <= cap) {
+ transactionFrameSizes.add(messageSize);
+ } else {
+ int baseline = deltaBaseline;
+ if (!deltaDictEnabled && currentBatchMaxSymbolId > deltaBaseline
+ && !splitFramesFit(cap, deltaBaseline)
+ && splitFramesFit(cap, currentBatchMaxSymbolId)) {
+ // The full-dictionary fallback ships the section as chunks of at
+ // most cap bytes ahead of the table frames.
+ long dictBytes = encoder.getSplitMessageSize(0, deltaBaseline, currentBatchMaxSymbolId)
+ - QwpConstants.HEADER_SIZE;
+ long chunkPayload = Math.max(1, cap - QwpConstants.HEADER_SIZE - 16);
+ for (long chunks = dictBytes / chunkPayload + 1; chunks > 0; chunks--) {
+ transactionFrameSizes.add(cap);
+ }
+ baseline = currentBatchMaxSymbolId;
+ }
+ for (int i = 0, n = splitFrameBodyBytes.size(); i < n; i++) {
+ transactionFrameSizes.add(encoder.getSplitMessageSize(
+ splitFrameBodyBytes.getQuick(i), baseline, currentBatchMaxSymbolId));
+ if (deltaDictEnabled && currentBatchMaxSymbolId > baseline) {
+ baseline = currentBatchMaxSymbolId;
+ }
+ }
+ }
+ if (deferCommit && !strictTransactional) {
+ // Keep room for the bare commit or rollback frame that ends the
+ // transaction: the next flush that does not fit commits it with one.
+ transactionFrameSizes.add(COMMIT_FRAME_MAX_BYTES);
+ }
+ int segments = txnSimSegments;
+ long remaining = segments == 0 ? engine.activeSegment().capacityRemaining() : txnSimRemaining;
+ long bytes = txnBytes;
+ for (int i = 0, n = transactionFrameSizes.size(); i < n; i++) {
+ long size = transactionFrameSizes.getQuick(i);
+ bytes += size;
+ if (segments == 0 || size > remaining) {
+ if (segments > 0 || size > remaining) {
+ remaining = segmentPayload;
+ }
+ segments++;
+ }
+ remaining = Math.max(0, remaining - size - MmapSegment.FRAME_HEADER_SIZE);
+ }
+ if (deferCommit && strictTransactional) {
+ // Keep room for the commit: the rows pending when this transaction
+ // ends are at most a frame's worth plus one row, which a split
+ // ships as a few frames of at most cap bytes.
+ long reserve = cap > 0 ? Math.min(cap, segmentPayload) : segmentPayload;
+ for (int i = 0; i < STRICT_COMMIT_RESERVE_FRAMES; i++) {
+ if (reserve > remaining) {
+ remaining = segmentPayload;
+ segments++;
+ }
+ remaining = Math.max(0, remaining - reserve - MmapSegment.FRAME_HEADER_SIZE);
+ }
+ }
+ long maxSegments = Math.max(2, engine.ringCapacityBytes() / engine.segmentSizeBytes());
+ if (segments > maxSegments) {
+ if (strictTransactional) {
+ txnRefusedBytes = bytes;
+ }
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * True for a QWP frame that joins the open transaction: FLAG_DEFER_COMMIT
+ * set, and not the rollback control frame (which also carries the flag).
+ */
+ private static boolean isTransactionFrame(long payloadAddr, int payloadLen) {
+ if (payloadLen < QwpConstants.HEADER_SIZE
+ || Unsafe.getUnsafe().getInt(payloadAddr) != QwpConstants.MAGIC_MESSAGE) {
+ return false;
+ }
+ byte flags = Unsafe.getUnsafe().getByte(payloadAddr + QwpConstants.HEADER_OFFSET_FLAGS);
+ return (flags & QwpConstants.FLAG_DEFER_COMMIT) != 0 && (flags & QwpConstants.FLAG_ROLLBACK) == 0;
+ }
+
+ /**
+ * Re-sizes the byte trigger {@link #effectiveAutoFlushBytes} from the frame
+ * size the connection can carry: the smaller of the advertised
+ * X-QWP-Max-Batch-Size and the payload a store-and-forward segment can hold.
+ *
+ * {@code at()} and {@code atNow()} throw it for the row that would take the
+ * transaction past either limit, and that row is not part of the transaction.
+ * {@code flush()} throws it when the client buffer cannot hold the flush, and
+ * sends nothing. Either way nothing past the limit reaches the server, and the
+ * transaction keeps the rows accepted before the error.
+ */
+public class TransactionTooLargeException extends LineSenderException {
+
+ public TransactionTooLargeException(CharSequence message) {
+ super(message);
+ }
+}
diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorSendEngine.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorSendEngine.java
index ba46365a..16d9296d 100644
--- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorSendEngine.java
+++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorSendEngine.java
@@ -1900,6 +1900,17 @@ public int sealedSegmentsSnapshot(MmapSegment[] target) {
return ring.snapshotSealedSegments(target);
}
+ /**
+ * Bytes of segments the ring may hold under {@code sf_max_total_bytes}: the
+ * budget's capacity minus the side-file bytes it also charges (disk mode).
+ * In a sender pool the budget is shared, so this bounds the rings of all
+ * pooled senders together, not this ring alone.
+ */
+ public long ringCapacityBytes() {
+ SegmentBudget budget = manager.getBudget();
+ return budget.getCapacityBytes() - budget.getSideFileBytes();
+ }
+
/**
* Configured per-segment size in bytes.
*/
diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java
index 8a76edcc..e45e53c8 100644
--- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java
+++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java
@@ -591,6 +591,16 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
// and by the I/O thread afterwards -- never concurrently.
private long orphanSkipStartFsn = -1L;
private long orphanSkipTipFsn = -1L;
+ // Whether the current connection's server confirmed X-QWP-Rollback, so a rollback
+ // control frame may go out on it. setWireBaselineWithCatchUp sets it for every
+ // connection from the client it runs on; false until then, the safe side (see
+ // retireRollbackFrame). Written by start() on the user thread before the I/O thread
+ // exists, and by the I/O thread afterwards -- never concurrently.
+ private boolean isRollbackConfirmed;
+ // FSN of the last commit-bearing frame sent on the current connection, -1 when none.
+ // Tracked only while !isRollbackConfirmed, for retireRollbackFrame. Reset with
+ // isRollbackConfirmed; same threading.
+ private long lastSentCommitBearingFsn = -1L;
// Poison-frame detector state (I/O thread only). poisonFsn is the FSN of the
// frame implicated by the most recent server-active rejection: the NACK-named
// frame, or the OK-level head-of-line frame (highestOkFsn+1) for a
@@ -3198,6 +3208,10 @@ private void setWireBaselineWithCatchUp(long replayStart) {
// handleServerRejection can tell "only the catch-up went out" from "the
// head data frame went out".
dataFrameSentThisConnection = false;
+ // The server this connection reached decides whether a rollback frame may go
+ // out on it; until a client exists no frame goes out at all.
+ isRollbackConfirmed = client != null && client.isServerRollbackEnabled();
+ lastSentCommitBearingFsn = -1L;
// Do not gate solely on isDeltaDictEnabled(): a recovered delta slot can
// lose access to its side-file while its on-disk frames still start above
// zero. hasReplayDictionaryDependency preserves that distinction while
@@ -3230,6 +3244,26 @@ private int frameDeltaStart(long payloadAddr, int payloadLen) {
return encoded < 0L ? -1 : (int) (encoded >>> 3);
}
+ // True for a QWP frame carrying FLAG_DEFER_COMMIT. A short or foreign payload counts
+ // as commit-bearing, so retireRollbackFrame waits for its ack: the safe side, and the
+ // rule RecoveredFrameAnalysis applies to the orphaned deferred tail.
+ private static boolean isDeferredFrame(long payloadAddr, int payloadLen) {
+ return payloadLen >= QwpConstants.HEADER_SIZE
+ && Unsafe.getUnsafe().getInt(payloadAddr) == QwpConstants.MAGIC_MESSAGE
+ && (Unsafe.getUnsafe().getByte(payloadAddr + QwpConstants.HEADER_OFFSET_FLAGS)
+ & QwpConstants.FLAG_DEFER_COMMIT) != 0;
+ }
+
+ // True only for the rollback control frame (QwpWebSocketEncoder.encodeRollbackMessage):
+ // a bare QWP header whose flags byte equals ROLLBACK_FRAME_FLAGS, compared whole as the
+ // server matches it.
+ private static boolean isRollbackFrame(long payloadAddr, int payloadLen) {
+ return payloadLen == QwpConstants.HEADER_SIZE
+ && Unsafe.getUnsafe().getInt(payloadAddr) == QwpConstants.MAGIC_MESSAGE
+ && Unsafe.getUnsafe().getByte(payloadAddr + QwpConstants.HEADER_OFFSET_FLAGS)
+ == QwpConstants.ROLLBACK_FRAME_FLAGS;
+ }
+
// True only for a well-formed QWP frame this encoder produced that carries a
// delta symbol-dict section. The magic check keeps the dict logic from
// misreading non-QWP payloads (e.g. synthetic frames injected by tests) whose
@@ -4001,6 +4035,16 @@ private boolean trySendBatch() {
break; // fixed byte cap: the next call carries on from this frame
}
long frameAddr = base + sendOffset + MmapSegment.FRAME_HEADER_SIZE;
+ if (!isRollbackConfirmed && isRollbackFrame(frameAddr, payloadLen)) {
+ if (batchFrames > 0) {
+ break;
+ }
+ // This server cannot roll back, so the frame never goes out on this
+ // connection: the loop waits for acks, or it discarded the rolled-back
+ // transaction and recycled.
+ retireRollbackFrame();
+ return false;
+ }
// 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
@@ -4063,6 +4107,9 @@ private boolean trySendBatch() {
sendOffset = frameEnd;
long fsnSent = fsnAtZero + nextWireSeq;
nextWireSeq++;
+ if (!isRollbackConfirmed && !isDeferredFrame(frameAddr, payloadLen)) {
+ lastSentCommitBearingFsn = fsnSent;
+ }
batchFrames++;
batchBytes += wireLen;
if (replayTargetFsn >= 0) {
@@ -4114,6 +4161,45 @@ void positionCursorForStart() {
positionCursorAt(replayStart);
}
+ /**
+ * Retires the rollback control frame at the send cursor of a connection whose
+ * server did not confirm X-QWP-Rollback. The producer sends that frame only after
+ * a server confirmed rollback, so this connection is a reconnect that reached a
+ * server without rollback support, such as an older node after a failover, with
+ * the frame not yet acked. That server would read the frame as an empty deferred
+ * frame and commit the rolled-back transaction, whose frames it holds uncommitted
+ * on this connection, with the next commit frame. So the frame never goes out:
+ * closing the connection discards those frames instead, as every server does with
+ * the rows a closed connection leaves uncommitted.
+ *
+ * The close waits until the server acks every commit-bearing frame sent on this
+ * connection before the rollback frame: the self-acknowledge below covers those
+ * frames too, so a commit the server had not yet processed would be lost with the
+ * connection. Until then this returns without doing anything, and tryReceiveAcks
+ * runs every loop iteration. Once they are acked, the frames this connection holds
+ * unacked are exactly the rolled-back transaction's deferred frames. The cumulative
+ * self-acknowledge of the rollback frame retires them, and the recycle discards
+ * them server-side and re-anchors the next connection past the rollback frame, the
+ * way {@link #tryRetireOrphanTail} handles an orphaned deferred tail.
+ */
+ private void retireRollbackFrame() {
+ long rollbackFsn = fsnAtZero + nextWireSeq;
+ long ackedFsn = engine.ackedFsn();
+ if (ackedFsn < lastSentCommitBearingFsn) {
+ return;
+ }
+ LOG.info("server does not support QWP rollback: discarding the rolled-back transaction "
+ + "[fsn {}..{}] by closing the connection that holds it uncommitted",
+ externalFsnBase + ackedFsn + 1L, externalFsnBase + rollbackFsn);
+ boolean isAdvanced = engine.acknowledge(rollbackFsn);
+ // Every ack is clamped to the frames this loop sent, and the cursor stands on
+ // the rollback frame, so the self-acknowledge always advances the watermark.
+ assert isAdvanced : "rollback frame already acknowledged [fsn=" + rollbackFsn + ']';
+ dispatchProgress(rollbackFsn);
+ fail(new LineSenderException(
+ "recycling connection to discard a rolled-back transaction the server cannot roll back"));
+ }
+
/**
* Retires the recovered orphaned deferred tail once retirable.
*
diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentManager.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentManager.java
index 7d3f5f56..ba400016 100644
--- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentManager.java
+++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentManager.java
@@ -472,6 +472,14 @@ public synchronized boolean isWorkerReaped() {
return workerThread == null;
}
+ /**
+ * The {@code sf_max_total_bytes} budget this manager charges its rings'
+ * segments to; shared with other managers in a sender pool.
+ */
+ public SegmentBudget getBudget() {
+ return budget;
+ }
+
/**
* Stop tracking {@code ring}. Idempotent; safe to call from any thread.
*
diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpConstants.java b/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpConstants.java
index 0fa7a645..ba1ee0e5 100644
--- a/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpConstants.java
+++ b/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpConstants.java
@@ -36,6 +36,13 @@ public final class QwpConstants {
* Client identifier sent in the X-QWP-Client-Id upgrade header.
*/
public static final String CLIENT_ID = "java/1.0.2";
+ /**
+ * Default of the server's {@code qwp.max.uncommitted.rows}: the per-table
+ * count of rows a deferred-commit group may hold before the server commits
+ * them on its own. A strict transaction assumes this value when the server
+ * does not advertise {@code X-QWP-Max-Uncommitted-Rows}.
+ */
+ public static final long DEFAULT_SERVER_MAX_UNCOMMITTED_ROWS = 1_000_000L;
/**
* Flag bit: defer WAL commit. The server appends rows to WAL writers
* but skips the commit until a subsequent message without this flag.
@@ -50,6 +57,12 @@ public final class QwpConstants {
* Flag bit: Gorilla timestamp encoding enabled.
*/
public static final byte FLAG_GORILLA = 0x04;
+ /**
+ * Flag bit: roll back the connection's open deferred-commit group. Only ever
+ * set on the rollback control frame (see {@link #ROLLBACK_FRAME_FLAGS}), which
+ * the client sends only after the server confirmed {@code X-QWP-Rollback}.
+ */
+ public static final byte FLAG_ROLLBACK = 0x20;
/**
* Flag bit: payload region after the prelude is zstd-compressed. Set only
* when the handshake negotiated zstd compression. Mirror of the server-side
@@ -108,6 +121,13 @@ public final class QwpConstants {
* constant; used by the decoder to reject malformed wire bytes.
*/
public static final int MAX_TABLE_NAME_LENGTH = 127;
+ /**
+ * Flags byte of the rollback control frame: a bare 12-byte header with no
+ * table blocks and no payload. It carries {@link #FLAG_DEFER_COMMIT} too, so a
+ * server that does not know {@link #FLAG_ROLLBACK} treats it as an empty
+ * deferred frame instead of a commit.
+ */
+ public static final byte ROLLBACK_FRAME_FLAGS = FLAG_ROLLBACK | FLAG_DEFER_COMMIT;
/**
* Status byte on a {@code QUERY_ERROR} frame: the query was cancelled,
* either by a client {@code CANCEL} frame or by explicit server-side
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 02f95b91..7e7ce99b 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
@@ -78,6 +78,9 @@ public class QwpTableBuffer implements QuietCloseable {
// buffer is listed once. Producer thread only.
private boolean listedPending;
private int rowCount;
+ // Rows of this table the sender already shipped in FLAG_DEFER_COMMIT frames
+ // of the open transaction. Producer thread only.
+ private long txnSentRows;
public QwpTableBuffer(String tableName) {
this(tableName, null);
@@ -282,6 +285,14 @@ public String getTableName() {
return tableName;
}
+ /**
+ * Rows of this table already shipped in FLAG_DEFER_COMMIT frames of the
+ * sender's open transaction; 0 outside a transaction.
+ */
+ public long getTxnSentRows() {
+ return txnSentRows;
+ }
+
/**
* Returns true if a row is partially built (columns written but not yet committed
* via {@link #nextRow()}).
@@ -426,6 +437,13 @@ public void setListedPending(boolean listedPending) {
this.listedPending = listedPending;
}
+ /**
+ * Sets the count reported by {@link #getTxnSentRows()}.
+ */
+ public void setTxnSentRows(long txnSentRows) {
+ this.txnSentRows = txnSentRows;
+ }
+
private static void assertColumnType(CharSequence name, byte type, ColumnBuffer column) {
if (column.type != type) {
throw new LineSenderException(
diff --git a/core/src/main/java/io/questdb/client/impl/PooledSender.java b/core/src/main/java/io/questdb/client/impl/PooledSender.java
index a9ef418a..e25bad14 100644
--- a/core/src/main/java/io/questdb/client/impl/PooledSender.java
+++ b/core/src/main/java/io/questdb/client/impl/PooledSender.java
@@ -28,6 +28,7 @@
import io.questdb.client.Sender;
import io.questdb.client.cutlass.line.array.DoubleArray;
import io.questdb.client.cutlass.line.array.LongArray;
+import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
import io.questdb.client.std.Decimal128;
import io.questdb.client.std.Decimal256;
import io.questdb.client.std.Decimal64;
@@ -47,7 +48,9 @@
*
* Behaviour difference from a raw Sender: {@link #close()} flushes the buffer
* and returns the slot to the pool. The underlying Sender is only truly closed
- * when {@link io.questdb.client.QuestDB#close()} shuts the pool down.
+ * when {@link io.questdb.client.QuestDB#close()} shuts the pool down. A
+ * WebSocket Sender in strict transactional mode commits only on flush(), so
+ * there close() discards the open transaction instead of flushing it.
*
* Because the slot is reused across borrows, this wrapper -- not the slot --
* carries the lease identity. A stale handle (held after {@link #close()}, with
@@ -167,8 +170,18 @@ public void close() {
// above. A stale duplicate throws here, and the finally routes it
// to discardBroken, which drops it on the same stale-generation
// check under the pool lock.
- slot.live(generation).flush();
- flushed = true;
+ Sender live = slot.live(generation);
+ if (live instanceof QwpWebSocketSender && ((QwpWebSocketSender) live).isStrictTransactional()) {
+ // A strict transaction commits only on an explicit flush(), so
+ // returning the lease discards whatever the borrower left open.
+ // The next borrower must not inherit it, so a transaction the
+ // server cannot roll back retires the connection instead (the
+ // server rolls it back when the connection closes).
+ flushed = ((QwpWebSocketSender) live).discardTransactionForReuse();
+ } else {
+ live.flush();
+ flushed = true;
+ }
} finally {
if (flushed) {
slot.pool().giveBack(this);
@@ -341,6 +354,11 @@ public void reset() {
slot.live(generation).reset();
}
+ @Override
+ public void rollback() {
+ slot.live(generation).rollback();
+ }
+
@Override
public void resetSymbolDictionary() {
slot.live(generation).resetSymbolDictionary();
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 e304df95..d1571390 100644
--- a/core/src/test/java/io/questdb/client/test/QuestDBBuilderTest.java
+++ b/core/src/test/java/io/questdb/client/test/QuestDBBuilderTest.java
@@ -215,12 +215,20 @@ public void testMalformedEgressConfigRejectedAtBuildWithMinZero() {
@Test
public void testAutoFlushIntervalOffAcceptedAtBuildWithMinZero() {
// auto_flush_interval=off turns off only the time-based trigger and is legal on
- // WebSocket, alone or with the row trigger off too (only auto_flush=off is rejected).
+ // WebSocket: alone, with the row trigger off too (a sender with no auto-flush
+ // trigger, in strict transactional mode), and with a byte trigger left as the
+ // only one.
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_rows=off;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_rows=off;auto_flush_interval=off;auto_flush_bytes=65536;"
+ "sender_pool_min=0;query_pool_min=0;").build().close();
+ QuestDB.builder().fromConfig(
+ "ws::addr=127.0.0.1:1;auto_flush=off;auto_flush_bytes=off;sender_pool_min=0;query_pool_min=0;")
+ .build().close();
}
@Test
@@ -229,18 +237,24 @@ public void testMalformedIngressConfigRejectedAtBuildWithMinZero() {
// 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 disables
- // auto-flush (unsupported on WebSocket), also when auto_flush_bytes is set, and
- // sf_durability=flush is not yet supported.
+ // (auto_flush_rows), and WebSocket build-time checks: a sender with no
+ // auto-flush trigger is in strict transactional mode, so it cannot combine
+ // with transaction=off, auto_flush=off cannot combine with a byte trigger,
+ // and 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=off;sender_pool_min=0;query_pool_min=0;", "auto-flush");
+ "ws::addr=127.0.0.1:1;auto_flush=off;transaction=off;sender_pool_min=0;query_pool_min=0;",
+ "which on WebSocket selects strict transactional mode and cannot be combined with transaction=off");
+ assertBuildRejected(
+ "ws::addr=127.0.0.1:1;auto_flush_rows=off;auto_flush_interval=off;transaction=off;"
+ + "sender_pool_min=0;query_pool_min=0;",
+ "which on WebSocket selects strict transactional mode and cannot be combined with transaction=off");
assertBuildRejected(
"ws::addr=127.0.0.1:1;auto_flush=off;auto_flush_bytes=65536;sender_pool_min=0;query_pool_min=0;",
- "auto-flush");
+ "auto flush bytes was already configured [autoFlushBytes=65536]");
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 cc8c78aa..fdde0e56 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
@@ -189,6 +189,32 @@ public void testExtractMaxBatchSizeParsesPositive() throws Exception {
Assert.assertEquals(16 * 1024 * 1024, invokeExtractMaxBatchSize(response));
}
+ @Test
+ public void testExtractMaxUncommittedRowsAbsentHeaderReturnsZero() throws Exception {
+ String response = "HTTP/1.1 101 Switching Protocols\r\n"
+ + "X-QWP-Version: 1\r\n"
+ + "X-QWP-Max-Batch-Size: 16777216\r\n"
+ + "\r\n";
+ Assert.assertEquals(0, invokeExtractMaxUncommittedRows(response));
+ }
+
+ @Test
+ public void testExtractMaxUncommittedRowsMalformedOrNegativeReturnsZero() throws Exception {
+ // 0 means "not advertised": a strict transaction then assumes the default.
+ Assert.assertEquals(0, invokeExtractMaxUncommittedRows("HTTP/1.1 101 Switching Protocols\r\n"
+ + "X-QWP-Max-Uncommitted-Rows: lots\r\n\r\n"));
+ Assert.assertEquals(0, invokeExtractMaxUncommittedRows("HTTP/1.1 101 Switching Protocols\r\n"
+ + "X-QWP-Max-Uncommitted-Rows: -5\r\n\r\n"));
+ }
+
+ @Test
+ public void testExtractMaxUncommittedRowsParsesBeyondIntRange() throws Exception {
+ String response = "HTTP/1.1 101 Switching Protocols\r\n"
+ + "x-qwp-max-uncommitted-rows: 5000000000 \r\n"
+ + "\r\n";
+ Assert.assertEquals(5_000_000_000L, invokeExtractMaxUncommittedRows(response));
+ }
+
@Test
public void testFrameBurstIsParsedInPlace() throws Exception {
assertFrameBurstIsParsedInPlace(false);
@@ -602,6 +628,12 @@ private static int invokeExtractMaxBatchSize(String response) throws Exception {
return (int) m.invoke(null, response);
}
+ private static long invokeExtractMaxUncommittedRows(String response) throws Exception {
+ Method m = WebSocketClient.class.getDeclaredMethod("extractMaxUncommittedRows", String.class);
+ m.setAccessible(true);
+ return (long) m.invoke(null, response);
+ }
+
private static void setUpgradedTrue(Object obj) throws Exception {
Class> clazz = obj.getClass();
while (clazz != null) {
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..390168fd 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
@@ -1052,7 +1052,10 @@ public void testFailedPublishDoesNotDuplicatePersistedSymbols() throws Exception
// Small segment; a heavily padded row's frame cannot fit, so
// appendBlocking throws PAYLOAD_TOO_LARGE deterministically -- no
// backpressure timing needed. The server never acks (SilentHandler).
- String cfg = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir + ";sf_max_segment_bytes=1024;";
+ // auto_flush_bytes=off: the oversized row must reach flush() whole,
+ // as the byte trigger would otherwise publish it from at().
+ String cfg = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir
+ + ";sf_max_segment_bytes=1024;auto_flush_bytes=off;";
String pad = TestUtils.repeat("x", 2000); // frame >> 1024-byte segment
Sender sender = Sender.fromConfig(cfg);
try {
@@ -1187,9 +1190,12 @@ public void testRecoveryAfterFailedPublishReplaysGapFree() throws Exception {
int port = silent.getPort();
silent.start();
Assert.assertTrue(silent.awaitStart(5, TimeUnit.SECONDS));
+ // auto_flush_bytes=off: the oversized row must reach flush() whole,
+ // as the byte trigger would otherwise publish it from at().
String cfg = "ws::addr=localhost:" + port
+ ";sf_dir=" + sfDir
+ ";sf_max_segment_bytes=4096"
+ + ";auto_flush_bytes=off"
+ ";close_flush_timeout_millis=0;";
Sender s1 = Sender.fromConfig(cfg);
try {
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..f69a5851 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,28 @@ public void testDefaultIsAsync() {
}
@Test
- public void testDisableAutoFlush_notSupportedForWebSocket() {
- assertThrowsAny(
- Sender.builder(Sender.Transport.WEBSOCKET)
+ public void testDisableAutoFlush_rejectsPositiveAutoFlushBytes() {
+ // disableAutoFlush() turns every auto-flush trigger off, so a byte trigger
+ // contradicts it in either order, as a row or interval trigger does.
+ assertThrows("cannot set auto flush bytes when auto-flush is already disabled",
+ () -> Sender.builder(Sender.Transport.WEBSOCKET)
.address(LOCALHOST)
- .disableAutoFlush(),
- "not supported for WebSocket");
+ .disableAutoFlush()
+ .autoFlushBytes(65_536));
+ assertThrows("auto flush bytes was already configured [autoFlushBytes=65536]",
+ () -> Sender.builder(Sender.Transport.WEBSOCKET)
+ .address(LOCALHOST)
+ .autoFlushBytes(65_536)
+ .disableAutoFlush());
+ // 0 turns the byte trigger off, which agrees with it.
+ Assert.assertNotNull(Sender.builder(Sender.Transport.WEBSOCKET)
+ .address(LOCALHOST)
+ .disableAutoFlush()
+ .autoFlushBytes(0));
+ Assert.assertNotNull(Sender.builder(Sender.Transport.WEBSOCKET)
+ .address(LOCALHOST)
+ .autoFlushBytes(0)
+ .disableAutoFlush());
}
@Test
@@ -790,6 +806,50 @@ public void testNoAddress_fails() {
"address not set");
}
+ @Test
+ public void testNoAutoFlushTrigger_rejectsTransactionOff() {
+ // A WebSocket sender with no auto-flush trigger commits only on flush(): it is
+ // strict unless transactional(true) says otherwise, and transactional(false)
+ // contradicts it, however the triggers were turned off.
+ String expected = "auto-flush is off (auto_flush=off, or auto_flush_rows=off and auto_flush_interval=off), "
+ + "which on WebSocket selects strict transactional mode and cannot be combined with "
+ + "transaction=off; set transaction=on or turn on an auto-flush trigger";
+ assertThrowsAny(
+ Sender.builder(Sender.Transport.WEBSOCKET)
+ .address(LOCALHOST)
+ .disableAutoFlush()
+ .transactional(false),
+ expected);
+ assertThrowsAny(
+ Sender.builder(Sender.Transport.WEBSOCKET)
+ .address(LOCALHOST)
+ .transactional(false)
+ .disableAutoFlush(),
+ expected);
+ assertThrowsAny(
+ Sender.builder(Sender.Transport.WEBSOCKET)
+ .address(LOCALHOST)
+ .autoFlushRows(0)
+ .autoFlushIntervalMillis(Integer.MAX_VALUE)
+ .transactional(false),
+ expected);
+ assertThrowsAny(
+ Sender.builder(Sender.Transport.WEBSOCKET)
+ .address(LOCALHOST)
+ .autoFlushRows(0)
+ .autoFlushIntervalMillis(Integer.MAX_VALUE)
+ .autoFlushBytes(0)
+ .transactional(false),
+ expected);
+ // strictTransactional(false) keeps transactional mode as set: off here.
+ assertThrowsAny(
+ Sender.builder(Sender.Transport.WEBSOCKET)
+ .address(LOCALHOST)
+ .disableAutoFlush()
+ .strictTransactional(false),
+ expected);
+ }
+
@Test
public void testPortMismatch_fails() {
assertThrowsAny(
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpLargeTransactionTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpLargeTransactionTest.java
new file mode 100644
index 00000000..95d003c1
--- /dev/null
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpLargeTransactionTest.java
@@ -0,0 +1,1704 @@
+/*+*****************************************************************************
+ * ___ _ ____ ____
+ * / _ \ _ _ ___ ___| |_| _ \| __ )
+ * | | | | | | |/ _ \/ __| __| | | | _ \
+ * | |_| | |_| | __/\__ \ |_| |_| | |_) |
+ * \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ * 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 ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import io.questdb.client.QuestDB;
+import io.questdb.client.Sender;
+import io.questdb.client.cutlass.line.LineSenderException;
+import io.questdb.client.cutlass.qwp.client.BatchTooLargeForCapException;
+import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
+import io.questdb.client.cutlass.qwp.client.TransactionTooLargeException;
+import io.questdb.client.cutlass.qwp.protocol.QwpConstants;
+import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer;
+import io.questdb.client.test.tools.TestUtils;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import static io.questdb.client.test.tools.TestUtils.assertMemoryLeak;
+
+/**
+ * Batches of any size over QWP/WebSocket: the byte trigger that keeps every batch
+ * within one frame, transactional and strict transactional mode, and rollback.
+ *
+ * {@link TxnServer} answers frames the way the QuestDB server does: a
+ * FLAG_DEFER_COMMIT frame gets no ack until a commit frame closes its group, a
+ * commit frame gets a cumulative OK ack, and the rollback control frame gets a
+ * cumulative OK ack once the server negotiated rollback.
+ */
+public class QwpLargeTransactionTest {
+ private static final int CAP = 32 * 1024;
+ // The settings a refusal at the client buffer names.
+ private static final String CLIENT_BUFFER_REMEDY = "Raise sf_max_total_bytes, or set sf_dir to keep the "
+ + "transaction on disk";
+ // What closing a sender or returning a pooled lease logs when it discards an open
+ // strict transaction of 1000 rows, behind the action that discarded it.
+ private static final String DISCARD_WARNING = " discarded an uncommitted transaction of 1000 rows: a strict "
+ + "transactional sender (transaction=strict, or one with no auto-flush trigger) commits only on "
+ + "flush(). Call flush() before close() to keep the rows, or ignore this if the discard is intentional";
+ private static final String NO_TRIGGER_TRANSACTION_OFF_ERROR = "auto-flush is off (auto_flush=off, or "
+ + "auto_flush_rows=off and auto_flush_interval=off), which on WebSocket selects strict transactional "
+ + "mode and cannot be combined with transaction=off; set transaction=on or turn on an auto-flush trigger";
+ // ~100 encoded bytes per row: 1000 rows (the default row trigger) are ~100 KB,
+ // three times the 32 KiB cap.
+ private static final String PAD = TestUtils.repeat("x", 90);
+ // transaction=on in frames of exactly 280 rows (~28 KB): two fill a 64 KiB segment,
+ // so the 256 KiB client buffer holds eight frames of a transaction. A blocked
+ // append would wait 30 s.
+ private static final String TXN_ON_FRAMES_OF_280 = "transaction=on;auto_flush_rows=280;auto_flush_bytes=off;"
+ + "auto_flush_interval=off;sf_max_segment_bytes=64k;sf_max_total_bytes=256k;"
+ + "sf_append_deadline_millis=30000;";
+
+ @Test
+ public void testAutoFlushBytesAloneCommitsEachBatch() throws Exception {
+ // A positive auto_flush_bytes is an auto-flush trigger: with the row and
+ // interval triggers off, every batch it cuts commits on its own, as before.
+ // transaction=off may say so explicitly.
+ assertMemoryLeak(() -> {
+ String[] configs = {
+ "auto_flush_rows=off;auto_flush_interval=off;auto_flush_bytes=8192;",
+ "auto_flush_rows=off;auto_flush_interval=off;auto_flush_bytes=8192;transaction=off;"
+ };
+ for (String config : configs) {
+ TxnServer h = new TxnServer();
+ try (TestWebSocketServer server = start(h, CAP)) {
+ try (Sender s = Sender.fromConfig(conf(server) + config)) {
+ QwpWebSocketSender ws = (QwpWebSocketSender) s;
+ Assert.assertFalse(config, ws.isTransactional());
+ Assert.assertEquals(config, 8_192, ws.getEffectiveAutoFlushBytes());
+ // ~20 KB: the byte trigger sends at least two batches, and each
+ // commits before any flush().
+ writeRows(s, "t", 0, 200);
+ Assert.assertTrue(config, h.awaitFrames(2));
+ for (TxnFrame f : h.frames) {
+ Assert.assertTrue(config + ": " + h.frames, f.isCommit());
+ }
+ }
+ // close() commits the rest.
+ Assert.assertEquals(config, h.frames, h.committedFrames());
+ }
+ }
+ });
+ }
+
+ @Test
+ public void testAutoFlushOffRejectsPositiveAutoFlushBytes() throws Exception {
+ // auto_flush=off turns every auto-flush trigger off, so a byte trigger
+ // contradicts it, as a row or interval trigger already does. The ws:: parser
+ // applies the trigger keys before auto_flush, whatever their order.
+ // auto_flush_bytes=off is no trigger; see
+ // testNoAutoFlushTriggerWithAutoFlushBytesOffSendsOneFramePerFlush.
+ assertMemoryLeak(() -> {
+ assertConfigRejected("auto_flush=off;auto_flush_bytes=65536;",
+ "auto flush bytes was already configured [autoFlushBytes=65536]");
+ assertConfigRejected("auto_flush_bytes=65536;auto_flush=off;",
+ "auto flush bytes was already configured [autoFlushBytes=65536]");
+ });
+ }
+
+ @Test
+ public void testDefaultSenderSplitsWideRowsIntoFrames() throws Exception {
+ // Before: no byte trigger by default, so the 1000-row batch outgrew the cap
+ // and flush threw BatchTooLargeForCapException.
+ assertMemoryLeak(() -> {
+ TxnServer h = new TxnServer();
+ try (TestWebSocketServer server = start(h, CAP)) {
+ try (Sender s = Sender.fromConfig(conf(server))) {
+ writeRows(s, "t", 0, 2_000);
+ s.flush();
+ Assert.assertTrue(s.drain(5_000));
+ QwpWebSocketSender ws = (QwpWebSocketSender) s;
+ Assert.assertEquals((long) CAP * 9 / 10, ws.getEffectiveAutoFlushBytes());
+ }
+ assertFramesWithinCap(h, CAP);
+ Assert.assertTrue("rows must span frames: " + h.frames, h.frames.size() >= 7);
+ // Not transactional: every frame commits on its own.
+ Assert.assertEquals(h.frames.size(), h.commitFrames());
+ }
+ });
+ }
+
+ @Test
+ public void testExplicitAutoFlushBytesOffKeepsNoByteTrigger() throws Exception {
+ // auto_flush_bytes=off still opts out of the byte trigger, as documented.
+ assertMemoryLeak(() -> {
+ TxnServer h = new TxnServer();
+ try (TestWebSocketServer server = start(h, CAP)) {
+ try (Sender s = Sender.fromConfig(conf(server) + "auto_flush_bytes=off;")) {
+ Assert.assertEquals(0, ((QwpWebSocketSender) s).getEffectiveAutoFlushBytes());
+ try {
+ writeRows(s, "t", 0, 1_000);
+ Assert.fail("a 1000-row batch over the cap must be rejected");
+ } catch (BatchTooLargeForCapException expected) {
+ TestUtils.assertContains(expected.getMessage(), "serverMaxBatchSize=" + CAP);
+ }
+ s.reset();
+ }
+ Assert.assertEquals(0, h.frames.size());
+ }
+ });
+ }
+
+ @Test
+ public void testByteTriggerFollowsSegmentSize() throws Exception {
+ // The server accepts 48 KiB frames, but a 16 KiB segment cannot hold them:
+ // batches must be sized to the segment, not fail at the append.
+ assertMemoryLeak(() -> {
+ TxnServer h = new TxnServer();
+ try (TestWebSocketServer server = start(h, 48 * 1024)) {
+ try (Sender s = Sender.fromConfig(conf(server) + "sf_max_segment_bytes=16k;")) {
+ writeRows(s, "t", 0, 600);
+ s.flush();
+ Assert.assertTrue(s.drain(5_000));
+ int segmentPayload = 16 * 1024 - 24 - 8;
+ Assert.assertEquals(segmentPayload * 9 / 10, ((QwpWebSocketSender) s).getEffectiveAutoFlushBytes());
+ }
+ assertFramesWithinCap(h, 16 * 1024 - 24 - 8);
+ Assert.assertTrue(h.frames.size() >= 4);
+ }
+ });
+ }
+
+ @Test
+ public void testNewSymbolsCountTowardByteTrigger() throws Exception {
+ // Symbol columns buffer 4-byte ids; the 300-byte strings ride in the frame's
+ // dictionary section. Counting them keeps the frame within the cap.
+ assertMemoryLeak(() -> {
+ TxnServer h = new TxnServer();
+ try (TestWebSocketServer server = start(h, CAP)) {
+ try (Sender s = Sender.fromConfig(conf(server))) {
+ String prefix = TestUtils.repeat("s", 290);
+ for (int i = 0; i < 1_000; i++) {
+ s.table("t").symbol("sym", prefix + i).longColumn("v", i).atNow();
+ }
+ s.flush();
+ Assert.assertTrue(s.drain(5_000));
+ }
+ assertFramesWithinCap(h, CAP);
+ Assert.assertTrue(h.frames.size() >= 10);
+ }
+ });
+ }
+
+ @Test
+ public void testNoAutoFlushTriggerIsStrictTransactionOfAnySize() throws Exception {
+ // Every way to turn off all auto-flush triggers builds the same sender: a
+ // strict transaction of any size, which only flush() commits.
+ // auto_flush_rows=off with auto_flush_interval=off used to send each flush as
+ // one batch, which failed past the frame cap, and to commit on close().
+ assertMemoryLeak(() -> {
+ for (NoTriggerSpelling spelling : noTriggerSpellings()) {
+ TxnServer h = new TxnServer();
+ List
+ * It also models what the server commits: a commit frame commits the group open
+ * on its connection, while a rollback frame or a closed connection leaves that
+ * group uncommitted. On a connection that did not confirm rollback the rollback
+ * frame is what a server without rollback support makes of it, an empty deferred
+ * frame: the group stays open, and the next commit on the connection commits it.
+ */
+ private static final class TxnServer implements TestWebSocketServer.WebSocketServerHandler {
+ // Every frame received, over all connections, in arrival order.
+ final List
+ * Contract under test: the I/O loop never sends the rollback frame to such a
+ * server. Once the server acked every commit-bearing frame sent before it, the
+ * loop acknowledges the rollback frame itself and recycles the connection,
+ * which discards the rolled-back frames server-side, and the next connection
+ * starts past the rollback frame. A server that confirmed rollback receives the
+ * frame. The same holds for a log recovered after a restart and for the
+ * background drainer, which replay through the same loop.
+ *
+ * {@link ServerModel} is the oracle: it commits what the QuestDB server commits,
+ * and drops the frames a closed connection still had in flight, so a recycle
+ * before the acks of earlier commits shows as a lost commit.
+ */
+public class CursorWebSocketSendLoopRollbackTest {
+ // Random transaction logs per run of the fuzz test; raise it for a local soak.
+ private static final int FUZZ_ITERATIONS = Integer.getInteger("questdb.fuzz.rollback.iterations", 100);
+ private static final long SEGMENT_SIZE_BYTES = 64 * 1024;
+ private static final String TABLE = "t";
+
+ @Rule
+ public final TemporaryFolder temporaryFolder = TemporaryFolder.builder().assureDeletion().build();
+
+ @Test
+ public void testBackgroundDrainerRetiresRollbackOnServerWithoutRollback() throws Exception {
+ // An orphaned slot whose producer rolled back a transaction and committed the
+ // next one before it went away, with nothing acked yet.
+ String slotPath = temporaryFolder.newFolder("slot").getAbsolutePath();
+ TestUtils.assertMemoryLeak(() -> {
+ try (CursorSendEngine engine = new CursorSendEngine(slotPath, SEGMENT_SIZE_BYTES)) {
+ appendCommit(engine, 1);
+ appendDeferred(engine, 2);
+ appendRollback(engine);
+ appendDeferred(engine, 3);
+ appendCommit(engine, 4);
+ }
+ ServerModel server = new ServerModel();
+ List
@@ -5068,8 +5198,13 @@ private void validateParameters() {
if (protocolVersion != PARAMETER_NOT_SET_EXPLICITLY) {
throw new LineSenderException("protocol version is not supported for WebSocket protocol");
}
- if (autoFlushDisabled) {
- throw new LineSenderException("disabling auto-flush is not supported for WebSocket protocol");
+ if (transactionModeSet && !transactional && !hasAutoFlushTrigger()) {
+ // Only flush() can commit this sender's rows, and a non-transactional
+ // flush has to fit one frame per table.
+ throw new LineSenderException("auto-flush is off (auto_flush=off, or auto_flush_rows=off and "
+ + "auto_flush_interval=off), which on WebSocket selects strict transactional mode and "
+ + "cannot be combined with transaction=off; set transaction=on or turn on an auto-flush "
+ + "trigger");
}
if ((sfDurability == SfDurability.FLUSH || sfDurability == SfDurability.APPEND)) {
throw new LineSenderException(
diff --git a/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java b/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java
index 938d5175..ab479d60 100644
--- a/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java
+++ b/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java
@@ -84,6 +84,11 @@ public abstract class WebSocketClient implements QuietCloseable {
private static final String QWP_CONTENT_ENCODING_HEADER_NAME = "X-QWP-Content-Encoding:";
private static final String QWP_DURABLE_ACK_HEADER_NAME = "X-QWP-Durable-Ack:";
private static final String QWP_MAX_BATCH_SIZE_HEADER_NAME = "X-QWP-Max-Batch-Size:";
+ private static final String QWP_MAX_UNCOMMITTED_ROWS_HEADER_NAME = "X-QWP-Max-Uncommitted-Rows:";
+ private static final String QWP_ROLLBACK_HEADER_NAME = "X-QWP-Rollback:";
+ // Token the server echoes in X-QWP-Rollback when it accepts the rollback
+ // control frame on this connection.
+ private static final String QWP_ROLLBACK_TOKEN_ENABLED = "enabled";
private static final String QWP_VERSION_HEADER_NAME = "X-QWP-Version:";
private static final ThreadLocal
+ *
+ */
+ private void recomputeFrameBudget() {
+ long segmentPayload = segmentPayloadCapacity();
+ int advertised = serverMaxBatchSize;
+ long frameSize = Math.min(advertised > 0 ? advertised : FALLBACK_SERVER_MAX_BATCH_SIZE, segmentPayload);
+ long budget = frameSize * 9 / 10;
+ if (autoFlushBytes > 0) {
+ long limit = advertised > 0 ? budget
+ : (segmentPayload == Long.MAX_VALUE ? Long.MAX_VALUE : segmentPayload * 9 / 10);
+ effectiveAutoFlushBytes = (int) Math.min(autoFlushBytes, limit);
+ } else if (autoFlushBytes == AUTO_FLUSH_BYTES_AUTO) {
+ effectiveAutoFlushBytes = (int) budget;
+ } else {
+ effectiveAutoFlushBytes = 0;
+ }
+ }
+
+ /**
+ * Transactional mode: advances the client-buffer simulation by one
+ * open-transaction frame that just reached the ring. {@code activeRemaining}
+ * is the active segment's room before the append, needed for the
+ * transaction's first frame.
+ */
+ private void recordTransactionFrame(int payloadLen, long activeRemaining) {
+ txnBytes += payloadLen;
+ if (txnSimSegments == 0) {
+ txnSimRemaining = payloadLen > activeRemaining ? segmentPayloadCapacity() : activeRemaining;
+ txnSimSegments = 1;
+ } else if (payloadLen > txnSimRemaining) {
+ txnSimRemaining = segmentPayloadCapacity();
+ txnSimSegments++;
+ }
+ txnSimRemaining = Math.max(0, txnSimRemaining - payloadLen - MmapSegment.FRAME_HEADER_SIZE);
+ }
+
+ /**
+ * Largest frame payload a store-and-forward segment can hold, or
+ * {@link Long#MAX_VALUE} before an engine is attached.
+ */
+ private long segmentPayloadCapacity() {
+ CursorSendEngine engine = cursorEngine;
+ return engine == null
+ ? Long.MAX_VALUE
+ : engine.segmentSizeBytes() - MmapSegment.HEADER_SIZE - MmapSegment.FRAME_HEADER_SIZE;
+ }
+
+ /**
+ * Rows the open transaction's deferred frames already sent, over all its tables.
+ */
+ private long sentTransactionRows() {
+ long rows = 0;
+ for (int i = 0, n = txnTables.size(); i < n; i++) {
+ rows += txnTables.getQuick(i).getTxnSentRows();
+ }
+ return rows;
+ }
+
+ private LineSenderException transactionAbortedError() {
+ return new LineSenderException("transaction aborted: rollback could not undo the frames of the "
+ + "open transaction already sent, because the server does not support QWP rollback. Close "
+ + "the sender to discard them; the server rolls them back when the connection closes");
+ }
+
+ /**
+ * The error for a strict transaction the client buffer cannot hold: the size
+ * the transaction would have reached, the buffer, what was refused, and the
+ * settings that make room.
+ */
+ private LineSenderException transactionTooLargeForClientBuffer(String outcome) {
+ CursorSendEngine engine = cursorEngine;
+ return new TransactionTooLargeException("transaction too large for the client buffer")
+ .put(" [transactionBytes=").put(txnRefusedBytes)
+ .put(", ringCapacityBytes=").put(engine.ringCapacityBytes())
+ .put(", segmentBytes=").put(engine.segmentSizeBytes())
+ .put("]; the client holds a strict transaction until the server commits it, and "
+ + "sf_max_total_bytes cannot hold this one, ").put(outcome)
+ .put(". Raise sf_max_total_bytes, or set sf_dir to keep the transaction on disk");
+ }
+
+ /**
+ * Strict mode commits only on flush(), so closing the sender or returning a
+ * pooled lease discards the open transaction. Logs it when the transaction holds
+ * rows: a caller that relied on close() to commit, as a sender with auto-flush
+ * triggers does, would otherwise lose them without a trace. Counts the rows
+ * already sent in deferred frames and those still buffered; call it before
+ * discardPendingRows() clears the latter.
+ */
+ private void warnDiscardedTransaction(String action) {
+ long rows = pendingRowCount + sentTransactionRows();
+ if (rows > 0) {
+ LOG.warn("{} discarded an uncommitted transaction of {} rows: a strict transactional sender "
+ + "(transaction=strict, or one with no auto-flush trigger) commits only on flush(). Call "
+ + "flush() before close() to keep the rows, or ignore this if the discard is intentional",
+ action, rows);
+ }
+ }
+
/**
* Accumulates the current row.
* Rows buffer until flush (explicit or auto-flush).
@@ -7142,6 +7832,12 @@ private void sealAndSwapBuffer() {
private void sendRow() {
checkRecycleFailure();
ensureConnected();
+ if (transactional) {
+ // Before nextRow(): a refusal leaves the row uncommitted, so the
+ // at()/atNow() error path discards it and the transaction keeps the
+ // rows accepted before it.
+ checkTransactionAcceptsRow();
+ }
// Hard guard: a single row whose bytes exceed the server's wire cap
// would flush as an oversize WS frame the server closes with
@@ -7179,7 +7875,7 @@ private void sendRow() {
currentTableBufferSnapshotBytes = bufferedNow;
if (shouldAutoFlush()) {
- flushPendingRows(transactional);
+ flushPendingRows(transactional, true);
}
}
@@ -7193,7 +7889,7 @@ private boolean shouldAutoFlush() {
if (autoFlushRows > 0 && pendingRowCount >= autoFlushRows) {
return true;
}
- if (effectiveAutoFlushBytes > 0 && getPendingBytes() >= effectiveAutoFlushBytes) {
+ if (effectiveAutoFlushBytes > 0 && pendingBytes + pendingSymbolDictBytes >= effectiveAutoFlushBytes) {
return true;
}
// pendingRowCount == 1: sendRow() just timestamped this very row, so its
diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/TransactionTooLargeException.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/TransactionTooLargeException.java
new file mode 100644
index 00000000..c7cc37f8
--- /dev/null
+++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/TransactionTooLargeException.java
@@ -0,0 +1,51 @@
+/*+*****************************************************************************
+ * ___ _ ____ ____
+ * / _ \ _ _ ___ ___| |_| _ \| __ )
+ * | | | | | | |/ _ \/ __| __| | | | _ \
+ * | |_| | |_| | __/\__ \ |_| |_| | |_) |
+ * \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ * 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.cutlass.qwp.client;
+
+import io.questdb.client.cutlass.line.LineSenderException;
+
+/**
+ * A strict transaction ({@code transaction=strict}, or a sender with no auto-flush
+ * trigger such as {@code auto_flush=off}, on the WebSocket transport) outgrew a
+ * limit that would stop it from committing as one transaction: the server's
+ * per-table {@code qwp.max.uncommitted.rows}, past which the server commits the
+ * rows on its own, or the client's {@code sf_max_total_bytes}, which must hold the
+ * whole uncommitted transaction until the server commits it. To send transactions
+ * of that size, raise the limit the message names: {@code qwp.max.uncommitted.rows}
+ * on the server, or {@code sf_max_total_bytes} on the client, whose default is
+ * larger with {@code sf_dir} set.
+ * > connections = after.framesByConnection();
+ Assert.assertEquals("one connection for the discarded transaction, one for the next: "
+ + after.frames, 2, connections.size());
+ // The reader threads of the two connections may log their first frames in
+ // either order: tell them apart by what they carry.
+ List
> framesByConnection() {
+ List
> byConnection = new ArrayList<>();
+ for (int i = 0, n = connections.size(); i < n; i++) {
+ byConnection.add(new ArrayList<>());
+ }
+ for (TxnFrame f : frames) {
+ byConnection.get(f.connection).add(f);
+ }
+ return byConnection;
+ }
+
+ synchronized boolean awaitFrames(int count) throws InterruptedException {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ while (frames.size() < count) {
+ long left = deadline - System.nanoTime();
+ if (left <= 0) {
+ return false;
+ }
+ TimeUnit.NANOSECONDS.timedWait(this, left);
+ }
+ return true;
+ }
+
+ synchronized int commitFrames() {
+ int n = 0;
+ for (TxnFrame f : frames) {
+ if (f.isCommit()) {
+ n++;
+ }
+ }
+ return n;
+ }
+
+ synchronized int rollbackFrames() {
+ int n = 0;
+ for (TxnFrame f : frames) {
+ if (f.isRollback()) {
+ n++;
+ }
+ }
+ return n;
+ }
+
+ private static byte[] ack(long seq) {
+ byte[] buf = new byte[1 + 8 + 2];
+ ByteBuffer bb = ByteBuffer.wrap(buf).order(ByteOrder.LITTLE_ENDIAN);
+ bb.put((byte) 0x00);
+ bb.putLong(seq);
+ bb.putShort((short) 0);
+ return buf;
+ }
+
+ private static final class Connection {
+ // Deferred frames the connection holds uncommitted.
+ final List
> groups = new ArrayList<>();
+ private final List