feat(qwp): recycle the symbol dictionary at a threshold to lift the cardinality cap

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

@@ -155,7 +155,10 @@ public int getOrAddSymbol(CharSequence symbol) {
                             + ". Rows using already-registered symbol values continue to work. To start a fresh "
                             + "dictionary, close this sender and build a new one (with store-and-forward the "
                             + "buffered backlog drains first). For unbounded-cardinality data use varchar "
-                            + "columns instead of symbol");
+                            + "columns instead of symbol. The automatic dictionary reset "
+                            + "(symbol_dict_reset, symbol_dict_reset_threshold) and "
+                            + "Sender.resetSymbolDictionary() avoid this cap, but both act only "
+                            + "on senders created via Sender.build()/fromConfig().");
         }
 
         // Assign new ID — toString() only for new symbols that must be stored

core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/BackgroundDrainer.java

@@ -789,7 +789,8 @@ public void run() {
                         maxHeadFrameRejections,
                         poisonMinEscalationWindowMillis,
                         catchUpCapGapMinEscalationWindowMillis,
-                        CursorWebSocketSendLoop.ReconnectPolicy.ORPHAN);
+                        CursorWebSocketSendLoop.ReconnectPolicy.ORPHAN,
+                        0L);
                 loop.start();
 
                 while (!stopRequestedOrInterrupted()) {

core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorSendEngine.java

@@ -1349,6 +1349,11 @@ public SlotLock getSlotLockForTesting() {
         return slotLock;
     }
 
+    @TestOnly
+    public Runnable getSlotLockReleaseListenerForTesting() {
+        return slotLockReleaseListener;
+    }
+
     @TestOnly
     public long getSyncIntervalNanosForTesting() {
         return syncIntervalNanos;

core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java

@@ -415,6 +415,14 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     // it is engine.ackedFsn() + 1, so the first replayed frame on the new
     // connection is wireSeq=0 and server-side cumulative ACKs still line up.
     private long fsnAtZero;
+    // Third coordinate: additive offset applied on top of the engine FSN
+    // (fsnAtZero already folded in) to produce the FSN this loop hands to a
+    // user-visible surface -- the progress dispatcher and every SenderError
+    // [fromFsn,toFsn] span. Fixed for the lifetime of one loop instance: 0
+    // for a loop built directly against a live engine, or the sender's
+    // fsnEpochBase snapshot when a symbol-dict recycle rebuilt the engine and
+    // restarted its internal FSNs at 0. Rule: external = externalFsnBase + raw.
+    private final long externalFsnBase;
     // Bounded-await backstop budget for close() (see
     // DEFAULT_CLOSE_SHUTDOWN_AWAIT_MILLIS). Overridable via
     // setShutdownAwaitTimeoutMillis so tests can exercise the timeout branch
@@ -712,7 +720,7 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
                 reconnectMaxBackoffMillis, durableAckMode,
                 durableAckKeepaliveIntervalMillis, maxHeadFrameRejections,
                 poisonMinEscalationWindowMillis, catchUpCapGapMinEscalationWindowMillis,
-                CatchUpCapGapPolicy.RETRY_FOREVER);
+                CatchUpCapGapPolicy.RETRY_FOREVER, 0L);
     }
 
     /**
@@ -730,7 +738,8 @@ private CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
                                    int maxHeadFrameRejections,
                                    long poisonMinEscalationWindowMillis,
                                    long catchUpCapGapMinEscalationWindowMillis,
-                                   CatchUpCapGapPolicy catchUpCapGapPolicy) {
+                                   CatchUpCapGapPolicy catchUpCapGapPolicy,
+                                   long externalFsnBase) {
         if (maxHeadFrameRejections < 1) {
             throw new IllegalArgumentException(
                     "maxHeadFrameRejections must be >= 1: " + maxHeadFrameRejections);
@@ -882,6 +891,7 @@ private CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
         // always outlive their borrower. Any growth copy-on-writes into loop-owned memory
         // (ensureSentDictCapacity), and releaseSentDictBytes frees only what the loop owns.
         this.fsnAtZero = fsnAtZero;
+        this.externalFsnBase = externalFsnBase;
         this.parkNanos = parkNanos;
         this.reconnectFactory = reconnectFactory;
         this.reconnectInitialBackoffMillis = reconnectInitialBackoffMillis;
@@ -923,6 +933,11 @@ private CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
      * establishing its first connection, then retries endpoint-policy failures
      * indefinitely after it has been live. An orphan drainer returns such failures
      * to its owner so the slot can follow its settle/quarantine policy.
+     * <p>
+     * {@code externalFsnBase} is the additive offset this loop folds into every
+     * user-visible FSN it produces (progress-dispatcher advances and
+     * {@link SenderError} spans) -- see {@link #externalFsnBase}. Pass {@code 0L}
+     * unless the caller is replacing an engine a symbol-dict recycle rebuilt.
      */
     public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
                                    long fsnAtZero, long parkNanos,
@@ -934,13 +949,14 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
                                    int maxHeadFrameRejections,
                                    long poisonMinEscalationWindowMillis,
                                    long catchUpCapGapMinEscalationWindowMillis,
-                                   ReconnectPolicy reconnectPolicy) {
+                                   ReconnectPolicy reconnectPolicy,
+                                   long externalFsnBase) {
         this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
                 reconnectInitialBackoffMillis,
                 reconnectMaxBackoffMillis, durableAckMode,
                 durableAckKeepaliveIntervalMillis, maxHeadFrameRejections,
                 poisonMinEscalationWindowMillis, catchUpCapGapMinEscalationWindowMillis,
-                catchUpPolicyFor(reconnectPolicy));
+                catchUpPolicyFor(reconnectPolicy), externalFsnBase);
     }
 
     private static CatchUpCapGapPolicy catchUpPolicyFor(ReconnectPolicy reconnectPolicy) {
@@ -1497,6 +1513,25 @@ public boolean isRunning() {
         return running;
     }
 
+    /**
+     * Called by the sender before {@link #start()} when a prior loop of the
+     * same sender already reached the server: restores Invariant B's
+     * past-initialization classification (see {@link
+     * #endpointPolicyFailureIsTerminal()}) across a symbol-dict recycle's
+     * loop rebuild, where the constructor would otherwise seed a fresh
+     * {@code hasEverConnected = false} for the new loop instance (ASYNC
+     * startup always hands the constructor a null client). Public rather
+     * than package-private only because the owning sender lives in a
+     * different package; it is not part of the public {@code Sender} API.
+     * {@code hasEverConnected} is volatile, so this write needs no extra
+     * synchronization to be visible to the I/O thread -- callers still call
+     * it before {@code start()} so the invariant is established before the
+     * loop can observe any endpoint-policy failure.
+     */
+    public void markEverConnected() {
+        hasEverConnected = true;
+    }
+
     /**
      * Plug an async-delivery sink for {@link SenderConnectionEvent}
      * notifications. Connection events fire from
@@ -1786,8 +1821,8 @@ private void connectLoop(Throwable initial, String phase, long paceFirstAttemptM
                     // role rejects are classified into the transient branch below.
                     LOG.error("terminal upgrade error during {} -- won't retry: {}",
                             phase, e.getMessage());
-                    long fromFsn = engine.ackedFsn() + 1L;
-                    long toFsn = Math.max(fromFsn, engine.publishedFsn());
+                    long fromFsn = externalFsnBase + engine.ackedFsn() + 1L;
+                    long toFsn = Math.max(fromFsn, externalFsnBase + engine.publishedFsn());
                     SenderError err = new SenderError(
                             SenderError.Category.SECURITY_ERROR,
                             SenderError.Policy.TERMINAL,
@@ -1826,8 +1861,8 @@ private void connectLoop(Throwable initial, String phase, long paceFirstAttemptM
                         // volatile first-writer-wins latch observed by the owner.
                         capabilityGapTerminal = e;
                     }
-                    long fromFsn = engine.ackedFsn() + 1L;
-                    long toFsn = Math.max(fromFsn, engine.publishedFsn());
+                    long fromFsn = externalFsnBase + engine.ackedFsn() + 1L;
+                    long toFsn = Math.max(fromFsn, externalFsnBase + engine.publishedFsn());
                     SenderError err = new SenderError(
                             SenderError.Category.PROTOCOL_VIOLATION,
                             SenderError.Policy.TERMINAL,
@@ -1958,15 +1993,15 @@ private void connectLoop(Throwable initial, String phase, long paceFirstAttemptM
      * producer stays alive and no data is at risk.
      */
     private void dispatchRetriedEndpointPolicyFailure(SenderError.Category category, String message) {
-        long fromFsn = engine.ackedFsn() + 1L;
+        long fromFsn = externalFsnBase + engine.ackedFsn() + 1L;
         dispatchError(new SenderError(
                 category,
                 SenderError.Policy.RETRIABLE,
                 SenderError.NO_STATUS_BYTE,
                 message,
                 SenderError.NO_MESSAGE_SEQUENCE,
                 fromFsn,
-                Math.max(fromFsn, engine.publishedFsn()),
+                Math.max(fromFsn, externalFsnBase + engine.publishedFsn()),
                 null,
                 System.nanoTime()
         ));
@@ -2072,9 +2107,11 @@ private void haltOnPoisonedFrame(String lastRejection, long toFsnHint) {
         // the operator at those bytes would misattribute the poison. The
         // caller supplies the span end: a NACK names the exact frame, so the
         // span is that single frame; a non-orderly close cannot single one
-        // out, so it spans to publishedFsn.
-        long fromFsn = poisonFsn;
-        long toFsn = Math.max(fromFsn, toFsnHint);
+        // out, so it spans to publishedFsn. poisonFsn and toFsnHint are both
+        // raw internal FSNs (fsnAtZero already folded in by the caller where
+        // relevant); rebase both by externalFsnBase here.
+        long fromFsn = externalFsnBase + poisonFsn;
+        long toFsn = Math.max(fromFsn, externalFsnBase + toFsnHint);
         String msg = "frame at fsn=" + fromFsn + " rejected " + poisonStrikes
                 + " consecutive times with no acceptance at or beyond it -- poisoned frame, replay cannot succeed (last: "
                 + lastRejection + ')';
@@ -2098,12 +2135,14 @@ private void haltOnPoisonedFrame(String lastRejection, long toFsnHint) {
      * Notify the progress dispatcher that the ack watermark advanced to
      * {@code ackedFsn}. Caller must already have observed the advance via
      * {@link CursorSendEngine#acknowledge}'s boolean return; this method
-     * does no further filtering.
+     * does no further filtering. {@code ackedFsn} is the engine-relative FSN
+     * (fsnAtZero already folded in by the caller); this rebases it by
+     * {@link #externalFsnBase} before it reaches the user-visible dispatcher.
      */
     private void dispatchProgress(long ackedFsn) {
         SenderProgressDispatcher d = progressDispatcher;
         if (d != null) {
-            d.offer(ackedFsn);
+            d.offer(externalFsnBase + ackedFsn);
         }
     }
 
@@ -3842,8 +3881,8 @@ private void handlePreSendRejection(long wireSeq, byte status,
             // protocol-violation close path uses (see onClose above): there
             // is no FSN we can attribute the rejection to, so we report
             // the unacked range the producer can correlate against.
-            long fromFsn = engine.ackedFsn() + 1L;
-            long toFsn = Math.max(fromFsn, engine.publishedFsn());
+            long fromFsn = externalFsnBase + engine.ackedFsn() + 1L;
+            long toFsn = Math.max(fromFsn, externalFsnBase + engine.publishedFsn());
             String tableName = response.getTableEntryCount() == 1
                     ? response.getTableName(0)
                     : null;
@@ -3966,8 +4005,8 @@ private void handleServerRejection(long wireSeq) {
                     status & 0xFF,
                     response.getErrorMessage(),
                     wireSeq,
-                    fsn,
-                    fsn,
+                    externalFsnBase + fsn,
+                    externalFsnBase + fsn,
                     tableName,
                     System.nanoTime()
             );

core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpConstants.java

@@ -92,8 +92,17 @@ public final class QwpConstants {
      * <p>
      * NOT the result-direction cap: {@code QwpResultBatchDecoder.MAX_CONN_DICT_SIZE}
      * (8,388,608) governs server-to-client result batches and is unrelated.
-     */
-    public static final int MAX_SYMBOL_DICTIONARY_SIZE = 1_000_000;
+     * <p>
+     * Compatibility: servers released before QuestDB 10.0.0 cap their
+     * dictionary at 1,000,000, and QWP has no wire-level negotiation of the
+     * limit -- a dictionary this client lets grow past 1M is rejected by
+     * those servers as a terminal parse error. Reachable on defaults: each
+     * recycle raises the re-arm bar to twice the dictionary size at the
+     * swap, capped at half of this constant, so an unbounded-cardinality
+     * producer's dictionary grows to 1M entries per epoch. Only 10.0.0+
+     * servers are supported.
+     */
+    public static final int MAX_SYMBOL_DICTIONARY_SIZE = 2_000_000;
     /**
      * Maximum table name length in bytes. Mirrors the server's same-named
      * constant; used by the decoder to reject malformed wire bytes.

core/src/main/java/io/questdb/client/impl/ConfigSchema.java

@@ -90,6 +90,9 @@ public final class ConfigSchema {
         str("sf_max_segment_bytes", Side.INGRESS);
         str("sf_max_total_bytes", Side.INGRESS);
         str("sf_sync_interval_millis", Side.INGRESS);
+        str("symbol_dict_reset", Side.INGRESS);
+        str("symbol_dict_reset_max_wait_millis", Side.INGRESS);
+        str("symbol_dict_reset_threshold", Side.INGRESS);
         str("transaction", Side.INGRESS);
 
         // EGRESS -- the QwpQueryClient applies. Typed where there is a range or

core/src/main/java/io/questdb/client/impl/PooledSender.java

@@ -335,6 +335,11 @@ public void reset() {
         slot.live(generation).reset();
     }
 
+    @Override
+    public void resetSymbolDictionary() {
+        slot.live(generation).resetSymbolDictionary();
+    }
+
     @Override
     public Sender shortColumn(CharSequence name, short value) {
         slot.live(generation).shortColumn(name, value);

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictCeilingTest.java

@@ -45,7 +45,7 @@
 /**
  * The producer-side dictionary cap ({@code MAX_SYMBOL_DICTIONARY_SIZE}) as the
  * application sees it: {@code symbol()} with a value that would create the
- * 1,000,001st distinct entry throws BEFORE the row is buffered, the row is
+ * 2,000,001st distinct entry throws BEFORE the row is buffered, the row is
  * cancellable, and the sender keeps working with already-registered values --
  * the wire never carries the refused symbol.
  */
@@ -76,7 +76,7 @@ public void testSymbolPastCapThrowsAndSenderStaysUsable() throws Exception {
                         sender.table("t").symbol("s", "one-too-many");
                         Assert.fail("expected LineSenderException past the dictionary cap");
                     } catch (LineSenderException expected) {
-                        Assert.assertTrue(expected.getMessage().contains("1000000"));
+                        Assert.assertTrue(expected.getMessage().contains(String.valueOf(MAX_SYMBOL_DICTIONARY_SIZE)));
                     }
                     Assert.assertEquals("the refusal must not have grown the dictionary",
                             MAX_SYMBOL_DICTIONARY_SIZE, dict.size());
@@ -98,6 +98,54 @@ public void testSymbolPastCapThrowsAndSenderStaysUsable() throws Exception {
         });
     }
 
+    /**
+     * A threshold configured AT the cap, with automatic reset DISABLED, must
+     * behave exactly like the undecorated cap: the refusal still fires, and
+     * its message still names the reset valve even though this particular
+     * sender has it switched off -- the valve is documented for senders that
+     * want it, not conditioned on this sender having chosen it.
+     * <p>
+     * Out of scope here: whether {@code symbol_dict_reset=off} actually keeps
+     * {@code armIfEligible()} from arming. That only runs from the tail of a
+     * completed {@code flush()}, which this test never performs (the fill
+     * goes through the raw dictionary test accessor, and the one
+     * {@code Sender}-routed call throws inside {@code symbol()} before a row
+     * completes) -- an {@code isResetArmed()} assertion here would pass
+     * regardless of the knob, proving nothing. That arming-vs-flush property
+     * is pinned in {@code SymbolDictRecycleArmingTest.testArmsAtThreshold}.
+     */
+    @Test
+    public void testCapReachedWithResetDisabledStillThrowsAndNamesTheResetValve() throws Exception {
+        assertMemoryLeak(() -> {
+            AckAllHandler handler = new AckAllHandler();
+            try (TestWebSocketServer server = new TestWebSocketServer(handler)) {
+                int port = server.getPort();
+                server.start();
+                Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+                try (Sender sender = Sender.fromConfig("ws::addr=localhost:" + port
+                        + ";symbol_dict_reset=off;symbol_dict_reset_threshold=" + MAX_SYMBOL_DICTIONARY_SIZE + ";")) {
+                    QwpWebSocketSender ws = (QwpWebSocketSender) sender;
+                    GlobalSymbolDictionary dict = ws.getGlobalSymbolDictionaryForTest();
+                    for (int i = 0; i < MAX_SYMBOL_DICTIONARY_SIZE; i++) {
+                        dict.getOrAddSymbol("f" + i);
+                    }
+
+                    try {
+                        sender.table("t").symbol("s", "one-too-many");
+                        Assert.fail("expected LineSenderException past the dictionary cap");
+                    } catch (LineSenderException expected) {
+                        String message = expected.getMessage();
+                        Assert.assertTrue("message names the limit: " + message,
+                                message.contains(String.valueOf(MAX_SYMBOL_DICTIONARY_SIZE)));
+                        Assert.assertTrue("message points at the reset valve: " + message,
+                                message.contains("symbol_dict_reset") && message.contains("resetSymbolDictionary()"));
+                    }
+                }
+            }
+        });
+    }
+
     private static void waitFor(Condition condition, long timeoutMillis) throws Exception {
         long deadline = System.currentTimeMillis() + timeoutMillis;
         while (!condition.holds()) {