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

.gitignore

@@ -26,6 +26,7 @@ core/CMakeCache.txt
 **/.project
 **/.settings
 **/.classpath
+**/.factorypath
 **/build
 **/CMakeFiles
 .envrc

core/src/main/java/io/questdb/client/SenderConnectionEvent.java

@@ -193,15 +193,19 @@ public enum Kind {
 
         /**
          * The active wire connection dropped and the reconnect loop is about
-         * to start. Fired once per outage, before any retry attempts.
+         * to start. Fired once per outage, before any retry attempts. Not
+         * fired for the deliberate reconnect a symbol-dictionary recycle
+         * performs.
          */
         DISCONNECTED,
 
         /**
          * A reconnect attempt succeeded against the same endpoint that was
          * previously active. Fired once per successful reconnect to the same
          * endpoint; {@link Kind#FAILED_OVER} is fired instead when the new
-         * endpoint differs.
+         * endpoint differs. Also fired — without a preceding
+         * {@link #DISCONNECTED} — for the reconnect that follows a
+         * symbol-dictionary recycle, which is deliberate, not an outage.
          */
         RECONNECTED,
 

core/src/main/java/io/questdb/client/SenderConnectionListener.java

@@ -49,7 +49,11 @@
  * ({@link SenderConnectionEvent.Kind#CONNECTED},
  * {@link SenderConnectionEvent.Kind#FAILED_OVER},
  * {@link SenderConnectionEvent.Kind#RECONNECTED}) are guaranteed to fire on
- * each transition. Failure events ({@code ENDPOINT_ATTEMPT_FAILED},
+ * each transition. A symbol-dictionary recycle tears the wire connection down
+ * and reconnects deliberately; that reconnect reports {@code RECONNECTED}
+ * (or {@code FAILED_OVER}) with no {@code DISCONNECTED} before it, so a
+ * listener pairing the two must not assume every success follows an outage.
+ * Failure events ({@code ENDPOINT_ATTEMPT_FAILED},
  * {@code ALL_ENDPOINTS_UNREACHABLE}) may be coalesced under inbox pressure.
  * The terminal event {@code AUTH_FAILED}
  * fires before the producer-thread {@code LineSenderException} is observable on

core/src/main/java/io/questdb/client/SenderErrorHandler.java

@@ -33,11 +33,13 @@
  *
  * <h2>Threading</h2>
  * Handlers normally run on a dedicated daemon dispatcher thread, never on the
- * I/O thread or the producer thread. One exception: a build()-time quarantine
+ * I/O thread or the producer thread. One exception: a slot quarantine
  * ({@link SenderError.Category#DATA_LOSS}) is dispatched synchronously on the
- * thread calling {@code build()} — the async dispatcher belongs to the
- * connected sender, which does not exist yet at build time. Handlers must not
- * block: for the build-time case, {@code build()} is waiting.
+ * calling thread — at {@code build()} time because the async dispatcher does
+ * not exist yet, and during a symbol-dictionary recycle rebuild (inside
+ * {@code table(...)}, on the producer thread) because a data-loss notice must
+ * not be dropped under inbox pressure. Handlers must not block or call back
+ * into the sender: in both cases the caller is waiting.
  * Slow handlers cannot stall publishing; if the bounded
  * inbox fills up, surplus notifications are dropped (visible via
  * {@code QwpWebSocketSender.getDroppedErrorNotifications()}).

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

@@ -155,7 +155,11 @@ 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(), and the reset "
+                            + "itself runs at a table() call once the backlog is acknowledged.");
         }
 
         // 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

@@ -1140,7 +1140,8 @@ public void run() {
                         maxHeadFrameRejections,
                         poisonMinEscalationWindowMillis,
                         catchUpCapGapMinEscalationWindowMillis,
-                        CursorWebSocketSendLoop.ReconnectPolicy.ORPHAN);
+                        CursorWebSocketSendLoop.ReconnectPolicy.ORPHAN,
+                        0L);
                 // Without this the loop's ridden-out reports -- above all
                 // "credential-unavailable", the one endpoint-policy failure an ORPHAN
                 // loop retries rather than latching -- are dispatched into a null, and

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

@@ -0,0 +1,77 @@
+/*+*****************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  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.sf.cursor;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Sender-lifetime observability counters shared by every
+ * {@link CursorWebSocketSendLoop} generation and every {@link CursorSendEngine}
+ * a {@code QwpWebSocketSender} attaches.
+ * <p>
+ * A symbol-dictionary recycle rebuilds both the loop and the engine; the
+ * counters they report must not restart with them, so the sender owns one
+ * instance for its whole life and hands it to each new loop and engine before
+ * use ({@code adoptCounters}). A loop or engine built standalone (tests,
+ * background drainers) starts on a fresh instance of its own, so its getters
+ * behave exactly as they did when the counters were per-instance fields.
+ * <p>
+ * Every field is incremented atomically from whichever I/O thread (or the
+ * caller thread inside a synchronous {@code start()}) is active, and read by
+ * any monitor thread; each read is one atomic load, so a value sampled across
+ * a recycle is never a torn sum.
+ */
+public final class CursorSendCounters {
+    /** ACK frames received and applied; {@code CursorWebSocketSendLoop#getTotalAcks()}. */
+    public final AtomicLong acks = new AtomicLong();
+    /** Producer appends that hit a full ring and parked; {@code CursorSendEngine#getTotalBackpressureStalls()}. */
+    public final AtomicLong backpressureStalls = new AtomicLong();
+    /** Frames re-sent inside post-reconnect catch-up windows. */
+    public final AtomicLong framesReplayed = new AtomicLong();
+    /** Binary frames issued to the wire, replays included. */
+    public final AtomicLong framesSent = new AtomicLong();
+    /** Reconnect attempts, failed and successful alike. */
+    public final AtomicLong reconnectAttempts = new AtomicLong();
+    /** Successful reconnects. */
+    public final AtomicLong reconnects = new AtomicLong();
+    /** Non-OK / non-DURABLE_ACK frames received from the server, retriable and terminal. */
+    public final AtomicLong serverErrors = new AtomicLong();
+
+    /**
+     * Folds every counter of {@code other} into this instance. Used once, when a
+     * loop or engine that started on its own default instance is handed the
+     * sender's shared one, so anything counted before adoption is carried
+     * rather than dropped.
+     */
+    public void addAll(CursorSendCounters other) {
+        acks.addAndGet(other.acks.get());
+        backpressureStalls.addAndGet(other.backpressureStalls.get());
+        framesReplayed.addAndGet(other.framesReplayed.get());
+        framesSent.addAndGet(other.framesSent.get());
+        reconnectAttempts.addAndGet(other.reconnectAttempts.get());
+        reconnects.addAndGet(other.reconnects.get());
+        serverErrors.addAndGet(other.serverErrors.get());
+    }
+}

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

@@ -84,12 +84,14 @@ public final class CursorSendEngine implements QuietCloseable {
     private static volatile ThreadFactory flockReleaseRetryThreadFactory =
             DEFAULT_FLOCK_RELEASE_RETRY_THREAD_FACTORY;
     private final long appendDeadlineNanos;
-    // Number of times appendBlocking observed BACKPRESSURE_NO_SPARE on its first
-    // ring.appendOrFsn attempt. One increment per blocking-call that had to wait
-    // for the manager (or for ACKs) — not one per spin-park. Producer-thread
-    // writer; volatile because the user may sample it from any thread.
-    private final java.util.concurrent.atomic.AtomicLong backpressureStallCount =
-            new java.util.concurrent.atomic.AtomicLong();
+    // Sender-lifetime observability counters; only backpressureStalls is used
+    // here: one increment per blocking appendBlocking call that had to wait
+    // for the manager (or for ACKs), not one per spin-park. Producer-thread
+    // writer; any thread may read it. A fresh instance by default;
+    // QwpWebSocketSender hands every engine it attaches its own shared
+    // instance via adoptCounters(), so a symbol-dictionary recycle's rebuilt
+    // engine keeps counting where the outgoing one stopped.
+    private CursorSendCounters counters = new CursorSendCounters();
     // Constructed before an owned manager acquires its native path scratch, so
     // callback allocation failure cannot orphan manager resources. A timed-out
     // close can then hand it to either manager path without allocating.
@@ -923,7 +925,7 @@ public long appendBlocking(long payloadAddr, int payloadLen) {
         }
         // First miss → record one stall (not one per spin) and start the
         // deadline clock.
-        backpressureStallCount.incrementAndGet();
+        counters.backpressureStalls.incrementAndGet();
         long deadlineNs = System.nanoTime() + appendDeadlineNanos;
         while (true) {
             long now = System.nanoTime();
@@ -937,7 +939,7 @@ public long appendBlocking(long payloadAddr, int payloadLen) {
                 lastBackpressureLogNs = now;
                 LOG.warn("cursor producer backpressured ({} stalls so far); waiting for I/O or periodic disk sync; "
                                 + "will throw after {} ms",
-                        backpressureStallCount.get(), appendDeadlineNanos / 1_000_000L);
+                        counters.backpressureStalls.get(), appendDeadlineNanos / 1_000_000L);
             }
             LockSupport.parkNanos(50_000L); // 50 µs
             fsn = ring.appendOrFsn(payloadAddr, payloadLen);
@@ -1349,6 +1351,11 @@ public SlotLock getSlotLockForTesting() {
         return slotLock;
     }
 
+    @TestOnly
+    public Runnable getSlotLockReleaseListenerForTesting() {
+        return slotLockReleaseListener;
+    }
+
     @TestOnly
     public long getSyncIntervalNanosForTesting() {
         return syncIntervalNanos;
@@ -1623,6 +1630,32 @@ public void setSlotLockReleaseListener(Runnable listener) {
         }
     }
 
+    /**
+     * Replaces this engine's counters with the sender's shared, sender-lifetime
+     * instance, folding this engine's own backpressure-stall count into it
+     * (the other counters are the loop's, and an engine handed between
+     * senders must not carry a previous sender's totals). Producer thread
+     * only, before the first {@link #appendBlocking} on this engine -- the
+     * same attach window {@link #setSlotLockReleaseListener} uses.
+     */
+    public void adoptCounters(CursorSendCounters shared) {
+        if (shared == counters) {
+            return;
+        }
+        shared.backpressureStalls.addAndGet(counters.backpressureStalls.get());
+        counters = shared;
+    }
+
+    /**
+     * The engine's current counters holder -- the DEFAULT instance until
+     * {@link #adoptCounters} replaces it. Test-only seam to observe what
+     * {@link #adoptCounters} folds and what it leaves alone.
+     */
+    @TestOnly
+    public CursorSendCounters getCountersForTesting() {
+        return counters;
+    }
+
     /**
      * Re-arms the shared terminal retry for an engine whose final watermark
      * barrier or confirmed flock release is still pending and no longer
@@ -1749,10 +1782,12 @@ public PersistedSymbolDict getPersistedSymbolDict() {
      * Number of times {@link #appendBlocking} hit
      * {@link SegmentRing#BACKPRESSURE_NO_SPARE} on its first attempt and
      * had to wait for the segment manager (or for ACKs) to free space.
-     * One increment per blocking-call, not per spin-park. Cumulative.
+     * One increment per blocking-call, not per spin-park. Cumulative, and
+     * carried across a symbol-dictionary recycle once the owning sender has
+     * adopted this engine (see {@link #adoptCounters}).
      */
     public long getTotalBackpressureStalls() {
-        return backpressureStallCount.get();
+        return counters.backpressureStalls.get();
     }
 
     /**