Cut three redundant walks out of recovery and the flush path

C9. The full-dict-fallback discard re-folded the whole recovered ring
even when the dictionary it discarded was EMPTY -- and that is the
ordinary client-upgrade path, not the rare one the comment described.
Every frame the shipped client ever wrote passes confirmedMaxId = -1, so
deltaStart is 0 and maxDeltaStart is 0; such a slot has no .symbol-dict,
so recovery creates a fresh empty one. All three discard conditions hold
on the first restart after upgrading, for every non-empty slot, and the
first fold was already keyed to that size-0 baseline -- so the re-fold
recomputed a bit-identical result over the entire backlog, with a
byte-at-a-time varint decode over full-dict frames that carry the whole
dictionary, inside build(). It now runs only when the discarded
dictionary held entries, which is the only case that can desynchronise
the fold from the baseline its consumers derive.

C11. PersistedSymbolDict.openExisting walked the file twice: once to
CRC every chunk and total the surviving entry bytes, then again --
re-decoding both header varints of every chunk -- purely to copy into an
exact-sized allocation. The entry region is by construction a subset of
the file, so the file length bounds it for free: one pass now validates
and copies together into a length-sized buffer, then shrinks to the exact
size. The trusted prefix is unchanged.

The same pass gained an entryBytes <= 0 guard. Every entry costs at least
its own length varint, so a positive entryCount inside a zero-byte region
is self-contradictory -- but the CRC only proves the bytes are what was
written, never that the header triple is consistent, so the scan accepted
such a chunk and left decodeLoadedSymbols to discover it later, as a
throw two layers up rather than a trimmed prefix.

C12. Each pass of the flush path re-walked tableBuffers.keys() and
re-probed the hash map per table: three probes per table on a plain
flush, five on a split, on the producer's thread. The non-empty tables
are now collected once into two reusable lists and every pass indexes
them directly -- which also removes the bodyIdx bookkeeping that had to
stay aligned with splitFrameBodyBytes by hand.

Regression tests, each verified to fail with its fix reverted:
testEmptyDictionaryDiscardDoesNotReFoldTheBacklog (expected:<1> but
was:<2>) with testPopulatedDictionaryDiscardStillReFolds as its
counterpart, and
testChunkClaimingEntriesInAZeroByteRegionEndsTheTrustedPrefix
(expected:<0> but was:<1>). Observing C9 needed a new engine-level
recoveryFoldCount: recoveryFramesVisited() cannot see a second fold
because the re-fold replaces the analysis instance.

C12 has no new test. Its behaviour is unchanged by construction and its
real risk -- index alignment across the collected lists and
splitFrameBodyBytes -- is already exercised by the two split tests that
drive two distinct tables (testSplitBatchShipsDeltaOnFirstFrameOnly,
testOversizedTableSplitStrandsNothing).

Full client suite: 2696 tests green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: glasstiger

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

@@ -175,6 +175,14 @@ public class QwpWebSocketSender implements Sender {
     // encode. flushPendingRowsSplit uses them both for preflight sizing and to walk
     // the staged body slices without encoding the batch a second time. Cleared and
     // repopulated on every flush; reused to stay zero-GC.
+    // The non-empty tables of the batch currently being flushed, collected ONCE per
+    // flush and then iterated by every pass. Each pass used to re-walk tableBuffers.keys()
+    // and re-probe the hash map per table -- 3 probes per table on a plain flush and 5 on
+    // a split, on the producer's thread. Reused across flushes to stay zero-GC. Held in
+    // lockstep so index i names the same table in both, and in the same order as
+    // splitFrameBodyBytes, which is what lets the split passes index them directly.
+    private final ObjList<QwpTableBuffer> flushTableBuffers = new ObjList<>();
+    private final ObjList<CharSequence> flushTableNames = new ObjList<>();
     private final IntList splitFrameBodyBytes = new IntList();
     private final CharSequenceObjHashMap<QwpTableBuffer> tableBuffers;
     // null means plain text (no TLS)
@@ -3214,19 +3222,26 @@ private void checkTableSelected() {
         }
     }
 
-    private int countNonEmptyTables(ObjList<CharSequence> keys) {
-        int tableCount = 0;
+    /**
+     * Collects the batch's non-empty tables into {@link #flushTableNames} /
+     * {@link #flushTableBuffers} and returns how many there are. One walk of the key
+     * list and one hash probe per table, for the whole flush.
+     */
+    private int collectNonEmptyTables(ObjList<CharSequence> keys) {
+        flushTableNames.clear();
+        flushTableBuffers.clear();
         for (int i = 0, n = keys.size(); i < n; i++) {
             CharSequence tableName = keys.getQuick(i);
             if (tableName == null) {
                 continue;
             }
             QwpTableBuffer tableBuffer = tableBuffers.get(tableName);
             if (tableBuffer != null && tableBuffer.getRowCount() > 0) {
-                tableCount++;
+                flushTableNames.add(tableName);
+                flushTableBuffers.add(tableBuffer);
             }
         }
-        return tableCount;
+        return flushTableBuffers.size();
     }
 
     private Endpoint currentEndpoint() {
@@ -3552,7 +3567,7 @@ private void flushPendingRows(boolean deferCommit) {
         cachedTimestampNanosColumn = null;
 
         ObjList<CharSequence> keys = tableBuffers.keys();
-        int tableCount = countNonEmptyTables(keys);
+        int tableCount = collectNonEmptyTables(keys);
         if (tableCount == 0) {
             pendingBytes = 0;
             currentTableBufferSnapshotBytes = currentTableBuffer == null
@@ -3584,19 +3599,12 @@ private void flushPendingRows(boolean deferCommit) {
         splitFrameBodyBytes.clear();
         int combinedBodyStart = encoder.getBuffer().getPosition();
         int bodyStart = combinedBodyStart;
-        for (int i = 0, n = keys.size(); i < n; i++) {
-            CharSequence tableName = keys.getQuick(i);
-            if (tableName == null) {
-                continue;
-            }
-            QwpTableBuffer tableBuffer = tableBuffers.get(tableName);
-            if (tableBuffer == null || tableBuffer.getRowCount() == 0) {
-                continue;
-            }
+        for (int i = 0; i < tableCount; i++) {
+            QwpTableBuffer tableBuffer = flushTableBuffers.getQuick(i);
 
             if (LOG.isDebugEnabled()) {
                 LOG.debug("Encoding table [name={}, rows={}, batchMaxId={}]",
-                        tableName, tableBuffer.getRowCount(), currentBatchMaxSymbolId);
+                        flushTableNames.getQuick(i), tableBuffer.getRowCount(), currentBatchMaxSymbolId);
             }
 
             encoder.addTable(tableBuffer);
@@ -3618,7 +3626,7 @@ private void flushPendingRows(boolean deferCommit) {
         if (cap > 0 && messageSize > cap) {
             // Keep the completed combined frame staged in the encoder while the
             // split path copies its delta entries and table-body slices.
-            flushPendingRowsSplit(keys, deferCommit, combinedBodyStart, cap);
+            flushPendingRowsSplit(deferCommit, combinedBodyStart, cap);
             return;
         }
 
@@ -3640,7 +3648,7 @@ private void flushPendingRows(boolean deferCommit) {
             lastCommitBoundaryFsn = cursorEngine.publishedFsn();
         }
 
-        resetTableBuffersAfterFlush(keys);
+        resetTableBuffersAfterFlush();
     }
 
     /**
@@ -3671,7 +3679,6 @@ private void flushPendingRows(boolean deferCommit) {
      *                    last message omits the flag.
      */
     private void flushPendingRowsSplit(
-            ObjList<CharSequence> keys,
             boolean deferCommit,
             int combinedBodyStart,
             int cap
@@ -3695,22 +3702,12 @@ private void flushPendingRowsSplit(
         // already performed. simBaseline mirrors the publish loop's baseline advance
         // (advanceSentMaxSymbolId), so each size equals the frame the staged-slice
         // assembler will build; this pass mutates no delta/persist state.
-        int nonEmptyCount = 0;
+        int nonEmptyCount = flushTableBuffers.size();
         int simBaseline = symbolDeltaBaseline();
-        int bodyIdx = 0;
-        for (int i = 0, n = keys.size(); i < n; i++) {
-            CharSequence tableName = keys.getQuick(i);
-            if (tableName == null) {
-                continue;
-            }
-            QwpTableBuffer tableBuffer = tableBuffers.get(tableName);
-            if (tableBuffer == null || tableBuffer.getRowCount() == 0) {
-                continue;
-            }
-            nonEmptyCount++;
+        for (int i = 0; i < nonEmptyCount; i++) {
+            CharSequence tableName = flushTableNames.getQuick(i);
             int messageSize = encoder.getSplitMessageSize(
-                    splitFrameBodyBytes.getQuick(bodyIdx), simBaseline, currentBatchMaxSymbolId);
-            bodyIdx++;
+                    splitFrameBodyBytes.getQuick(i), simBaseline, currentBatchMaxSymbolId);
             if (messageSize > cap) {
                 // The batch stays BUFFERED: this throw precedes every publish, and a
                 // rejected flush must not silently discard the caller's rows (see
@@ -3737,24 +3734,14 @@ private void flushPendingRowsSplit(
             }
         }
 
-        int sent = 0;
-        bodyIdx = 0;
         int tableBodyOffset = combinedBodyStart;
-        for (int i = 0, n = keys.size(); i < n; i++) {
-            CharSequence tableName = keys.getQuick(i);
-            if (tableName == null) {
-                continue;
-            }
-            QwpTableBuffer tableBuffer = tableBuffers.get(tableName);
-            if (tableBuffer == null || tableBuffer.getRowCount() == 0) {
-                continue;
-            }
+        for (int i = 0; i < nonEmptyCount; i++) {
+            CharSequence tableName = flushTableNames.getQuick(i);
 
-            sent++;
-            boolean isLast = (sent == nonEmptyCount);
+            boolean isLast = (i == nonEmptyCount - 1);
             boolean deferThis = deferCommit || !isLast;
 
-            int tableBodyLength = splitFrameBodyBytes.getQuick(bodyIdx++);
+            int tableBodyLength = splitFrameBodyBytes.getQuick(i);
             // Persist before touching activeBuffer. If the write-ahead fails, the
             // caller can retry with both the source rows and active microbatch
             // unchanged. The first frame carries the batch's new symbols; later
@@ -3800,21 +3787,16 @@ private void flushPendingRowsSplit(
             // committed the whole group.
             lastCommitBoundaryFsn = cursorEngine.publishedFsn();
         }
-        resetTableBuffersAfterFlush(keys);
+        resetTableBuffersAfterFlush();
     }
 
-    private void resetTableBuffersAfterFlush(ObjList<CharSequence> keys) {
-        for (int i = 0, n = keys.size(); i < n; i++) {
-            CharSequence tableName = keys.getQuick(i);
-            if (tableName == null) {
-                continue;
-            }
-            QwpTableBuffer tableBuffer = tableBuffers.get(tableName);
-            if (tableBuffer == null || tableBuffer.getRowCount() == 0) {
-                continue;
-            }
-            tableBuffer.reset();
+    private void resetTableBuffersAfterFlush() {
+        for (int i = 0, n = flushTableBuffers.size(); i < n; i++) {
+            flushTableBuffers.getQuick(i).reset();
         }
+        // Drop the references; the next flush re-collects.
+        flushTableNames.clear();
+        flushTableBuffers.clear();
         currentBatchMaxSymbolId = -1;
         pendingBytes = 0;
         currentTableBufferSnapshotBytes = 0;

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

@@ -108,6 +108,10 @@ public final class CursorSendEngine implements QuietCloseable {
     // against the recovered dictionary size to fail clean instead. Computed once
     // in the constructor's recovery branch; -1 elsewhere.
     private long recoveredMaxSymbolId = -1L;
+    // How many times the constructor folded the recovered ring. Observable because
+    // recoveryFramesVisited() cannot be: the full-dict-fallback re-fold REPLACES the
+    // analysis instance, resetting that counter, so a second fold is invisible through it.
+    private int recoveryFoldCount;
     // Highest deltaStart across the recovered COMMITTED frames; 0 when none carries a symbol
     // dictionary. ZERO means every surviving frame is SELF-SUFFICIENT -- it re-registers its
     // dictionary from id 0 -- so the slot replays with no dictionary at all and the send loop
@@ -348,6 +352,7 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                 // without requiring a second bounded scan.
                 recoveredFrameAnalysisInProgress = recovered.analyzeRecovery(
                         persistedDictInProgress == null ? 0 : persistedDictInProgress.size());
+                recoveryFoldCount++;
                 // 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
@@ -401,6 +406,19 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                 if (persistedDictInProgress != null
                         && recoveredMaxSymbolId >= persistedDictInProgress.size()
                         && recoveredMaxSymbolDeltaStart == 0L) {
+                    // Only a NON-EMPTY discarded dictionary can desynchronise the fold.
+                    // The first fold was keyed to this very size, so when it is 0 the
+                    // re-fold below recomputes the identical baseline-0 result -- and
+                    // that is the OVERWHELMINGLY common case, not the rare one the
+                    // comment used to describe: every frame the shipped client ever
+                    // wrote passes confirmedMaxId = -1, hence deltaStart == 0, hence
+                    // maxDeltaStart == 0; and such a slot has no .symbol-dict, so
+                    // recovery creates a fresh EMPTY one. All three conditions therefore
+                    // hold on the FIRST restart after upgrading, for every non-empty
+                    // slot, and the re-fold was a second O(payload bytes) walk -- with a
+                    // byte-at-a-time varint decode over full-dict frames, which carry the
+                    // whole dictionary -- for no result change at all.
+                    int discardedSize = persistedDictInProgress.size();
                     persistedDictInProgress.close();
                     persistedDictInProgress = null;
                     // Re-fold at baseline 0. The analysis above was keyed to the
@@ -427,11 +445,14 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                     // Re-folding rather than keeping the dictionary preserves the discard's
                     // whole point -- the slot recovers in full-dict mode, exactly as it was
                     // written, with producer, mirror and replay guard all anchored at 0.
-                    recoveredFrameAnalysisInProgress.close();
-                    recoveredFrameAnalysisInProgress = recovered.analyzeRecovery(0);
-                    this.recoveredCommitBoundaryFsn = recoveredFrameAnalysisInProgress.commitBoundaryFsn();
-                    this.recoveredMaxSymbolId = recoveredFrameAnalysisInProgress.maxDeltaEnd() - 1L;
-                    this.recoveredMaxSymbolDeltaStart = recoveredFrameAnalysisInProgress.maxDeltaStart();
+                    if (discardedSize > 0) {
+                        recoveredFrameAnalysisInProgress.close();
+                        recoveredFrameAnalysisInProgress = recovered.analyzeRecovery(0);
+                        recoveryFoldCount++;
+                        this.recoveredCommitBoundaryFsn = recoveredFrameAnalysisInProgress.commitBoundaryFsn();
+                        this.recoveredMaxSymbolId = recoveredFrameAnalysisInProgress.maxDeltaEnd() - 1L;
+                        this.recoveredMaxSymbolDeltaStart = recoveredFrameAnalysisInProgress.maxDeltaStart();
+                    }
                 }
             } else {
                 // Fresh start with no recovered segments. Any stale
@@ -809,6 +830,11 @@ void copyRecoveredSymbolSuffix(int baseline, long target) {
         }
     }
 
+    @TestOnly
+    public int recoveryFoldCount() {
+        return recoveryFoldCount;
+    }
+
     @TestOnly
     public long recoveryFramesVisited() {
         return recoveredFrameAnalysis == null ? 0L : recoveredFrameAnalysis.framesVisited();

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

@@ -670,6 +670,7 @@ private static PersistedSymbolDict openExisting(FilesFacade ff, String filePath,
         boolean mappedInput = ff.isMmapAllowed();
         long inputAddr = 0L;
         long entriesAddr = 0L;
+        int entriesCapacity = 0;
         int entriesLen = 0;
         try {
             if (mappedInput) {
@@ -692,15 +693,25 @@ private static PersistedSymbolDict openExisting(FilesFacade ff, String filePath,
                     || Unsafe.getUnsafe().getByte(inputAddr + 4) != VERSION) {
                 throw new IllegalStateException("bad magic or unknown symbol dictionary version");
             }
-            // First pass validates chunks and totals only the entry bytes that
-            // survive. Production reads through a file-backed mapping, so this
-            // pass does not allocate an anonymous whole-file copy. The second
-            // pass copies directly into one exact-sized retained allocation.
-            RecoveryScan scan = scanRecoveredChunks(inputAddr, len);
+            // ONE pass: validate each chunk's CRC and, once proven good, copy its
+            // entries straight out. The entry region is by construction a subset of
+            // the file, so `len` is a safe upper bound to allocate against -- it
+            // overshoots only by the chunk headers and CRCs -- and the exact size is
+            // reclaimed by the shrink below. The previous shape walked the file
+            // TWICE, re-decoding both header varints of every chunk on the second
+            // pass purely to size one allocation it could have bounded for free.
+            entriesCapacity = len;
+            entriesAddr = Unsafe.malloc(entriesCapacity, MemoryTag.NATIVE_DEFAULT);
+            RecoveryScan scan = scanAndCopyRecoveredChunks(inputAddr, len, entriesAddr);
             entriesLen = scan.entriesLen;
-            if (entriesLen > 0) {
-                entriesAddr = Unsafe.malloc(entriesLen, MemoryTag.NATIVE_DEFAULT);
-                copyRecoveredEntries(inputAddr, scan.validLen, entriesAddr, entriesLen);
+            if (entriesLen == 0) {
+                Unsafe.free(entriesAddr, entriesCapacity, MemoryTag.NATIVE_DEFAULT);
+                entriesAddr = 0L;
+                entriesCapacity = 0;
+            } else if (entriesLen < entriesCapacity) {
+                entriesAddr = Unsafe.realloc(
+                        entriesAddr, entriesCapacity, entriesLen, MemoryTag.NATIVE_DEFAULT);
+                entriesCapacity = entriesLen;
             }
             if (mappedInput) {
                 ff.munmap(inputAddr, fileLen, MemoryTag.MMAP_DEFAULT);
@@ -729,7 +740,8 @@ private static PersistedSymbolDict openExisting(FilesFacade ff, String filePath,
                 }
             }
             if (entriesAddr != 0L) {
-                Unsafe.free(entriesAddr, entriesLen, MemoryTag.NATIVE_DEFAULT);
+                // capacity, not entriesLen: the shrink may not have happened yet.
+                Unsafe.free(entriesAddr, entriesCapacity, MemoryTag.NATIVE_DEFAULT);
             }
             if (fd >= 0) {
                 ff.close(fd);
@@ -743,25 +755,15 @@ private static PersistedSymbolDict openExisting(FilesFacade ff, String filePath,
         }
     }
 
-    private static void copyRecoveredEntries(long inputAddr, int validLen, long entriesAddr, int entriesLen) {
-        Varint v = new Varint();
-        int diskPos = HEADER_SIZE;
-        int dstPos = 0;
-        while (diskPos < validLen) {
-            if (!v.decode(inputAddr, diskPos, validLen)
-                    || !v.decode(inputAddr, v.end, validLen)) {
-                throw new IllegalStateException("validated symbol dictionary chunk could not be decoded");
-            }
-            int entryBytes = (int) v.value;
-            int entriesStart = v.end;
-            Unsafe.getUnsafe().copyMemory(inputAddr + entriesStart, entriesAddr + dstPos, entryBytes);
-            dstPos += entryBytes;
-            diskPos = entriesStart + entryBytes + CRC_SIZE;
-        }
-        assert dstPos == entriesLen;
-    }
-
-    private static RecoveryScan scanRecoveredChunks(long inputAddr, int len) {
+    /**
+     * Validates every chunk and copies the entries of each proven-good one into
+     * {@code dstAddr}, in a single pass. {@code dstAddr} must have room for {@code len}
+     * bytes -- the entry region is a subset of the file, so that always suffices.
+     * Stops at the first chunk that is torn, fails its CRC, or is internally
+     * inconsistent, exactly as the two-pass version did, so the trusted prefix is
+     * unchanged.
+     */
+    private static RecoveryScan scanAndCopyRecoveredChunks(long inputAddr, int len, long dstAddr) {
         Varint v = new Varint();
         int count = 0;
         int diskPos = HEADER_SIZE;
@@ -788,10 +790,17 @@ private static RecoveryScan scanRecoveredChunks(long inputAddr, int len) {
                     chunkEndI - diskPos);
             if (crcCalc != crcStored
                     || entryCount <= 0
+                    // Every entry costs at least its own length varint, so a positive
+                    // entryCount inside a zero-byte region is self-contradictory. The CRC
+                    // proves the bytes are what was WRITTEN, never that the triple is
+                    // consistent, and decodeLoadedSymbols would only discover it later --
+                    // as a throw two layers up rather than a trimmed trusted prefix.
+                    || entryBytes <= 0
                     || (long) count + entryCount > Integer.MAX_VALUE
                     || entriesLen + entryBytes > Integer.MAX_VALUE) {
                 break;
             }
+            Unsafe.getUnsafe().copyMemory(inputAddr + entriesStart, dstAddr + entriesLen, entryBytes);
             entriesLen += entryBytes;
             diskPos = chunkEndI + CRC_SIZE;
             count += (int) entryCount;