Optimize symbol dictionary recovery and catch-up

Author: glasstiger

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

@@ -137,6 +137,10 @@ public final class CursorSendEngine implements QuietCloseable {
     // constructor, closed by {@link #close()}. When null in disk mode the engine
     // reports delta encoding as unavailable and the sender keeps full-dict frames.
     private final PersistedSymbolDict persistedSymbolDict;
+    // Engine-owned output of the single ordered recovery walk. It is retained
+    // because both producer seeding and every recycled send loop need the same
+    // frame-rebuilt symbol suffix. Null for fresh and memory-only engines.
+    private final RecoveredFrameAnalysis recoveredFrameAnalysis;
     // close() is publicly callable from any thread (Sender.close from a user
     // thread, JVM shutdown hooks, test cleanup). volatile + synchronized
     // close() makes the check-and-set atomic and gives readers a fence.
@@ -248,6 +252,7 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
         SegmentRing ringInProgress = null;
         AckWatermark watermarkInProgress = null;
         PersistedSymbolDict persistedDictInProgress = null;
+        RecoveredFrameAnalysis recoveredFrameAnalysisInProgress = null;
         try {
             // Disk mode: try to recover any *.sfa files left behind by a prior
             // session before deciding to start fresh. Without this the engine
@@ -329,6 +334,12 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                 if (seed >= 0) {
                     recovered.acknowledge(seed);
                 }
+                // Fold the whole recovered ring once. The result checkpoints all
+                // running metadata and raw symbol bytes at each commit-bearing
+                // frame, so its final snapshot excludes an orphan deferred tail
+                // without requiring a second bounded scan.
+                recoveredFrameAnalysisInProgress = recovered.analyzeRecovery(
+                        persistedDictInProgress == null ? 0 : persistedDictInProgress.size());
                 // Locate the last commit-bearing frame below a potentially
                 // orphaned FLAG_DEFER_COMMIT tail. A producer that crashed (or
                 // closed) mid-transaction leaves deferred frames with no
@@ -338,12 +349,7 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                 // drainOnClose), and (b) replaying them into a NEW session's
                 // commit would resurrect half a transaction -- see the WARN
                 // below. Computed before the I/O loop or producer append.
-                this.recoveredCommitBoundaryFsn = recovered.findLastFsnWithoutPayloadFlag(
-                        QwpConstants.HEADER_OFFSET_FLAGS,
-                        QwpConstants.FLAG_DEFER_COMMIT,
-                        QwpConstants.MAGIC_MESSAGE,
-                        QwpConstants.HEADER_SIZE
-                );
+                this.recoveredCommitBoundaryFsn = recoveredFrameAnalysisInProgress.commitBoundaryFsn();
                 if (publishedFsn >= 0 && recoveredCommitBoundaryFsn < publishedFsn) {
                     this.recoveredOrphanTipFsn = publishedFsn;
                     LOG.warn("recovered SF log ends with {} deferred frame(s) whose transaction was never "
@@ -363,12 +369,7 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                 // this and over-reject an otherwise-recoverable slot. maxSymbolDeltaEnd
                 // returns 0 when no such frame carries a symbol, yielding -1 here.
                 // Computed before the I/O loop or producer append; single-threaded.
-                this.recoveredMaxSymbolId = recovered.maxSymbolDeltaEnd(
-                        QwpConstants.MAGIC_MESSAGE,
-                        QwpConstants.HEADER_OFFSET_FLAGS,
-                        QwpConstants.FLAG_DELTA_SYMBOL_DICT,
-                        QwpConstants.HEADER_SIZE,
-                        recoveredCommitBoundaryFsn) - 1L;
+                this.recoveredMaxSymbolId = recoveredFrameAnalysisInProgress.maxDeltaEnd() - 1L;
                 // Full-dict-fallback recovery. When the persisted .symbol-dict is a
                 // SUBSET of the ids the surviving frames reference
                 // (recoveredMaxSymbolId >= its size) YET every such frame is
@@ -388,12 +389,7 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                 // dictionary. The recoveredMaxSymbolId >= size guard means this never
                 // fires for a slot whose dictionary is intact, nor for an empty slot
                 // (recoveredMaxSymbolId == -1). Single-threaded; before the I/O loop.
-                this.recoveredMaxSymbolDeltaStart = recovered.maxSymbolDeltaStart(
-                        QwpConstants.MAGIC_MESSAGE,
-                        QwpConstants.HEADER_OFFSET_FLAGS,
-                        QwpConstants.FLAG_DELTA_SYMBOL_DICT,
-                        QwpConstants.HEADER_SIZE,
-                        recoveredCommitBoundaryFsn);
+                this.recoveredMaxSymbolDeltaStart = recoveredFrameAnalysisInProgress.maxDeltaStart();
                 if (persistedDictInProgress != null
                         && recoveredMaxSymbolId >= persistedDictInProgress.size()
                         && recoveredMaxSymbolDeltaStart == 0L) {
@@ -451,6 +447,7 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
             this.ring = ringInProgress;
             this.watermark = watermarkInProgress;
             this.persistedSymbolDict = persistedDictInProgress;
+            this.recoveredFrameAnalysis = recoveredFrameAnalysisInProgress;
         } catch (Throwable t) {
             // Stop an owned manager before freeing the ring and watermark it may
             // touch, then release the slot lock. Each cleanup is in its own
@@ -482,6 +479,12 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                 } catch (Throwable ignored) {
                 }
             }
+            if (recoveredFrameAnalysisInProgress != null) {
+                try {
+                    recoveredFrameAnalysisInProgress.close();
+                } catch (Throwable ignored) {
+                }
+            }
             if (acquiredLock != null) {
                 try {
                     acquiredLock.close();
@@ -667,6 +670,12 @@ public synchronized void close() {
                 } catch (Throwable ignored) {
                 }
             }
+            if (recoveredFrameAnalysis != null) {
+                try {
+                    recoveredFrameAnalysis.close();
+                } catch (Throwable ignored) {
+                }
+            }
             if (fullyDrained) {
                 try {
                     unlinkAllSegmentFiles(sfDir);
@@ -716,6 +725,11 @@ public synchronized void close() {
      * transmitted, so their ids never reach a server and must not inflate the baseline.
      */
     public long collectReplaySymbolsAbove(int baseline, ObjList<String> out) {
+        if (recoveredFrameAnalysis != null
+                && recoveredFrameAnalysis.baseline() == baseline) {
+            recoveredFrameAnalysis.appendDecodedSymbols(out);
+            return recoveredFrameAnalysis.coverage();
+        }
         if (ring == null) {
             return baseline;
         }
@@ -738,6 +752,40 @@ public long collectReplaySymbolsAbove(int baseline, ObjList<String> out) {
         );
     }
 
+    long recoveredSymbolCoverage(int baseline) {
+        return checkedRecoveryAnalysis(baseline).coverage();
+    }
+
+    int recoveredSymbolSuffixCount(int baseline) {
+        return checkedRecoveryAnalysis(baseline).rawCount();
+    }
+
+    int recoveredSymbolSuffixLen(int baseline) {
+        return checkedRecoveryAnalysis(baseline).rawLen();
+    }
+
+    void copyRecoveredSymbolSuffix(int baseline, long target) {
+        RecoveredFrameAnalysis analysis = checkedRecoveryAnalysis(baseline);
+        int len = analysis.rawLen();
+        if (len > 0) {
+            io.questdb.client.std.Unsafe.getUnsafe().copyMemory(analysis.rawAddr(), target, len);
+        }
+    }
+
+    @TestOnly
+    public long recoveryFramesVisited() {
+        return recoveredFrameAnalysis == null ? 0L : recoveredFrameAnalysis.framesVisited();
+    }
+
+    private RecoveredFrameAnalysis checkedRecoveryAnalysis(int baseline) {
+        if (recoveredFrameAnalysis == null || recoveredFrameAnalysis.baseline() != baseline) {
+            throw new IllegalStateException("recovery symbol baseline mismatch [expected="
+                    + (recoveredFrameAnalysis == null ? "none" : recoveredFrameAnalysis.baseline())
+                    + ", actual=" + baseline + ']');
+        }
+        return recoveredFrameAnalysis;
+    }
+
     /**
      * Pass-through to {@link SegmentRing#findSegmentContaining(long)}.
      */

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

@@ -44,7 +44,6 @@
 import io.questdb.client.std.MemoryTag;
 import io.questdb.client.std.QuietCloseable;
 import io.questdb.client.std.str.Utf8s;
-import io.questdb.client.std.ObjList;
 import io.questdb.client.std.Unsafe;
 import org.jetbrains.annotations.TestOnly;
 import org.slf4j.Logger;
@@ -195,7 +194,7 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
      * Throttle "reconnect attempt N failed" WARN logs to one per 5 s.
      */
     private static final long RECONNECT_LOG_THROTTLE_NANOS = 5_000_000_000L;
-    // Test seam: when true, appendSymbolToMirror throws instead of appending,
+    // Test seam: when true, recovery mirror seeding throws before growing,
     // simulating the native realloc OOM (or MAX_SENT_DICT_BYTES ceiling) that
     // ensureSentDictCapacity can raise while the constructor seeds the recovery
     // mirror. Lets a test prove the constructor frees the already-malloc'd mirror on
@@ -300,6 +299,13 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     private int sentDictBytesCapacity;
     private int sentDictBytesLen;
     private int sentDictCount;
+    // Reusable staging frame for dictionary catch-up chunks. A reconnect may split a
+    // large dictionary into many chunks, and every later reconnect repeats that split;
+    // retaining one growable loop-owned buffer turns the old malloc/free-per-chunk path
+    // into amortized growth only. I/O-thread-owned, freed beside sentDictBytesAddr.
+    private long catchUpFrameAddr;
+    private int catchUpFrameCapacity;
+    private int catchUpFrameGrowthCount;
     // Orphan-policy cap-gap attempts -- catch-ups that reached a node and found an entry
     // too large for its batch cap -- with no intervening successful catch-up or unrelated
     // reconnect state (see MAX_CATCHUP_CAP_GAP_ATTEMPTS for the full budget accounting).
@@ -750,18 +756,28 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
         // catch-up-free.
         if (engine.recoveredMaxSymbolDeltaStart() > 0L) {
             // The prefix seed above may already have malloc'd the mirror, and
-            // appendSymbolToMirror -> ensureSentDictCapacity can still throw here (a
-            // native realloc OOM, or the MAX_SENT_DICT_BYTES ceiling). Such a throw
+            // ensureSentDictCapacity can still throw here (a native realloc OOM, or
+            // the MAX_SENT_DICT_BYTES ceiling). Such a throw
             // propagates out of the constructor, so the half-built loop is never
             // assigned to a reference -- neither ensureConnected's catch nor
             // BackgroundDrainer's finally can close() it -- and the mirror would leak.
             // Free it on any throw so the constructor leaves nothing behind, mirroring
             // the loopNeverRan free in close() / ioLoop's exit.
             try {
-                ObjList<String> fromFrames = new ObjList<>();
-                if (engine.collectReplaySymbolsAbove(sentDictCount, fromFrames) >= 0) {
-                    for (int i = 0, n = fromFrames.size(); i < n; i++) {
-                        appendSymbolToMirror(fromFrames.getQuick(i));
+                int baseline = sentDictCount;
+                if (engine.recoveredSymbolCoverage(baseline) >= 0L) {
+                    int suffixLen = engine.recoveredSymbolSuffixLen(baseline);
+                    int suffixCount = engine.recoveredSymbolSuffixCount(baseline);
+                    if (suffixLen > 0) {
+                        if (forceMirrorSeedFailureForTest) {
+                            throw new LineSenderException(
+                                    "simulated mirror seed allocation failure (test only)");
+                        }
+                        ensureSentDictCapacity((long) sentDictBytesLen + suffixLen);
+                        engine.copyRecoveredSymbolSuffix(
+                                baseline, sentDictBytesAddr + sentDictBytesLen);
+                        sentDictBytesLen += suffixLen;
+                        sentDictCount += suffixCount;
                     }
                 }
             } catch (Throwable t) {
@@ -1120,6 +1136,9 @@ public synchronized void close() {
             // buffer and drive setWireBaselineWithCatchUp into a null-mirror catch-up.
             sentDictCount = 0;
         }
+        if (loopNeverRan) {
+            freeCatchUpFrameBuffer();
+        }
     }
 
     /**
@@ -2030,6 +2049,7 @@ private void ioLoop() {
                 sentDictBytesLen = 0;
                 sentDictCount = 0; // keep the mirror all-or-nothing (see close())
             }
+            freeCatchUpFrameBuffer();
             shutdownLatch.countDown();
             // Failed-stop hand-off (see delegateEngineClose): the owner could
             // not free the engine safely while this thread was alive, so the
@@ -2613,7 +2633,8 @@ private void sendCatchUpChunk(int deltaStart, int deltaCount, long symbolsAddr,
         }
         int payloadLen = (int) payloadLenL;
         int frameLen = (int) frameLenL;
-        long frame = Unsafe.malloc(frameLen, MemoryTag.NATIVE_DEFAULT);
+        ensureCatchUpFrameCapacity(frameLen);
+        long frame = catchUpFrameAddr;
         try {
             Unsafe.getUnsafe().putByte(frame, (byte) 'Q');
             Unsafe.getUnsafe().putByte(frame + 1, (byte) 'W');
@@ -2647,14 +2668,42 @@ private void sendCatchUpChunk(int deltaStart, int deltaCount, long symbolsAddr,
                 throw (Error) t;
             }
             throw new CatchUpSendException(t);
-        } finally {
-            Unsafe.free(frame, frameLen, MemoryTag.NATIVE_DEFAULT);
         }
         nextWireSeq++; // this catch-up chunk consumed a wire sequence
         lastFrameOrPingNanos = System.nanoTime();
         totalFramesSent.incrementAndGet();
     }
 
+    @TestOnly
+    public int catchUpFrameGrowthCount() {
+        return catchUpFrameGrowthCount;
+    }
+
+    private void ensureCatchUpFrameCapacity(int required) {
+        if (catchUpFrameCapacity >= required) {
+            return;
+        }
+        long newCapacity = Math.max(required, Math.max(4096L, (long) catchUpFrameCapacity * 2L));
+        if (newCapacity > MAX_SENT_DICT_BYTES) {
+            newCapacity = MAX_SENT_DICT_BYTES;
+        }
+        catchUpFrameAddr = Unsafe.realloc(
+                catchUpFrameAddr,
+                catchUpFrameCapacity,
+                (int) newCapacity,
+                MemoryTag.NATIVE_DEFAULT);
+        catchUpFrameCapacity = (int) newCapacity;
+        catchUpFrameGrowthCount++;
+    }
+
+    private void freeCatchUpFrameBuffer() {
+        if (catchUpFrameAddr != 0L) {
+            Unsafe.free(catchUpFrameAddr, catchUpFrameCapacity, MemoryTag.NATIVE_DEFAULT);
+            catchUpFrameAddr = 0L;
+            catchUpFrameCapacity = 0;
+        }
+    }
+
     private boolean tryReceiveAcks() {
         boolean any = false;
         try {

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

@@ -639,6 +639,20 @@ public long collectReplaySymbolsAbove(
         return coverage;
     }
 
+    /**
+     * Feeds every recovered frame in this segment to the engine's single-pass
+     * recovery fold. Segments are visited oldest-first by {@link SegmentRing}.
+     */
+    void scanRecovery(RecoveredFrameAnalysis analysis) {
+        long off = HEADER_SIZE;
+        long frames = frameCount;
+        for (long i = 0; i < frames; i++) {
+            int payloadLen = Unsafe.getUnsafe().getInt(mmapAddress + off + 4);
+            analysis.accept(baseSeq + i, mmapAddress + off + FRAME_HEADER_SIZE, payloadLen);
+            off += FRAME_HEADER_SIZE + payloadLen;
+        }
+    }
+
     /**
      * Highest {@code deltaStart + deltaCount} (one past the highest symbol id) any
      * delta-flagged frame in this segment carries, or {@code 0} only when NO