Cut symbol-dict overhead and fix two blind guards

Five review findings on the delta symbol dictionary. The three performance
ones all sit on paths that scale with symbol cardinality -- the workload
this feature exists for introduces a new symbol per ROW -- and the two
guards were pinning nothing.

RecoveredFrameAnalysis.foldDelta issued one bounds check, one capacity
check and a ~12-byte copyMemory per SYMBOL. runningCoverage is
loop-invariant (it only advances after the walk) and id ascends, so
"id >= runningCoverage" is a step predicate: the entries a frame
contributes are always ONE contiguous run at the tail of its delta
section. Note where the run starts and copy it once. Recovery walks the
whole backlog, so this traded millions of stub-dispatched small copies for
one bulk copy per frame. It also drops the partial prefix the per-entry
version left behind on a markGap bail-out; that residue was already
unreachable (a gap pins coverage() at -1, and a later self-sufficient
frame resets the buffer), so not writing it is equivalent and leaves less
state to reason about. accumulateSentDict already solved the same problem
this way one file over.

Crc32c.updateUnsafe read each 8-byte slice-by-8 block with eight getByte
calls: 16 loads per block where 9 do. Consume it with two getInt loads.
getInt is an Unsafe intrinsic exactly as getByte is, so the catchable
InternalError this method exists for is unaffected; the loop becomes
little-endian, which costs nothing because the formats it checksums are
already little-endian throughout. This is the whole recovery CRC. The
differential test against the native intrinsic (8 alignments x 28 lengths
x 3 seeds, plus the chaining fuzz) proves it bit-identical.

The sent-dict entry-ends index was maintained incrementally by
accumulateSentDict and built eagerly by the constructor, yet
sendDictCatchUp is its ONLY reader. That put a store per new symbol on the
per-frame I/O path and retained 4 bytes per symbol for the connection's
whole life, to serve a reconnect that may never come. Build it lazily
instead: ensureSentDictEntryIndex() already notices sentDictIndexedCount
falling behind and rebuilds, at most once per reconnect, dwarfed by
shipping the same bytes over the wire. A connection that never reconnects
no longer allocates the index at all.

testCtorFreesSeededMirrorWhenFrameSeedThrows could not fail. The seam
threw BEFORE ensureSentDictCapacity -- the call that copy-on-writes the
borrowed prefix into a loop-OWNED allocation -- so the cleanup it guards,
gated on sentDictBytesOwned, freed nothing. Deleting that cleanup left the
test green. Move the seam past the grow: deleting the cleanup now fails
with a 4096-byte NATIVE_DEFAULT leak. The test's premise comment described
the old copy-based design and is corrected.

testPersistFailureSurfacesAsLineSenderException drove code production
never runs. FilesFacade.isMmapAllowed() defaults to (this == INSTANCE), so
installing any fault facade silently swapped the dictionary onto the
positioned-write fallback -- the act of injecting the fault replaced the
code under test. Fault a refused ff.allocate on the mmap append window
instead, with the facade opting into isMmapAllowed(): that is both the
real shape of ENOSPC here and the path production takes. Reverting the
wrap now escapes as "could not grow mmap append region", from the mapped
path.

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

@@ -3905,15 +3905,16 @@ private void persistNewSymbolsBeforePublish() {
                 pd.appendSymbols(globalSymbolDictionary, from, currentBatchMaxSymbolId);
             }
         } catch (Throwable t) {
-            // A short write (disk full / quota) to the persisted dictionary throws
-            // a low-level IllegalStateException. Surface it as a LineSenderException
-            // -- like every other flush-path failure, e.g. the cursor append in
-            // sealAndSwapBuffer -- so a caller catching LineSenderException around
-            // flush() also catches a disk-full during the write-ahead persist. The
-            // persist ran before publish and pd.size() did not advance on the short
-            // write, so the still-buffered rows re-persist the same range
-            // idempotently on retry. A JVM Error is never a persist failure; let it
-            // propagate.
+            // A failed write to the persisted dictionary throws a low-level
+            // IllegalStateException: in production that is ff.allocate refusing to grow
+            // the mmap append window (how a full disk / exhausted quota surfaces there),
+            // and behind an injected facade a short positioned write. Surface it as a
+            // LineSenderException -- like every other flush-path failure, e.g. the cursor
+            // append in sealAndSwapBuffer -- so a caller catching LineSenderException
+            // around flush() also catches a disk-full during the write-ahead persist. The
+            // persist ran before publish and pd.size() did not advance on the failure, so
+            // the still-buffered rows re-persist the same range idempotently on retry.
+            // A JVM Error is never a persist failure; let it propagate.
             if (t instanceof Error) {
                 throw (Error) t;
             }

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

@@ -197,12 +197,14 @@ 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, 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
-    // such a throw rather than leaking it. Production never sets it; volatile only so
-    // a test thread's write is visible to the loop under test.
+    // Test seam: when true, recovery mirror seeding throws immediately AFTER
+    // ensureSentDictCapacity has grown (and therefore taken ownership of) the mirror,
+    // standing in for the copyRecoveredSymbolSuffix-adjacent failure that leaves a
+    // freshly malloc'd mirror with no owner. It sits after the grow deliberately: the
+    // constructor's cleanup only frees an OWNED mirror, so a seam before the grow
+    // would leave nothing to free and the leak guard would pass with the cleanup
+    // deleted. Production never sets it; volatile only so a test thread's write is
+    // visible to the loop under test.
     @TestOnly
     static volatile boolean forceMirrorSeedFailureForTest;
     // Pre-converted to nanos for the comparison in sendDurableAckKeepaliveIfDue.
@@ -784,21 +786,30 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
                     int suffixLen = engine.recoveredSymbolSuffixLen(baseline);
                     int suffixCount = engine.recoveredSymbolSuffixCount(baseline);
                     if (suffixLen > 0) {
+                        ensureSentDictCapacity((long) sentDictBytesLen + suffixLen);
                         if (forceMirrorSeedFailureForTest) {
+                            // Throw AFTER the grow, never before it. ensureSentDictCapacity
+                            // is what copy-on-writes a borrowed prefix into a loop-OWNED
+                            // allocation (sentDictBytesOwned = true), and the catch's
+                            // releaseSentDictBytes() is gated on exactly that flag. A seam
+                            // in front of the grow fires while the mirror is still borrowed
+                            // from PersistedSymbolDict, so the cleanup frees nothing and the
+                            // guard below passes even with the cleanup deleted -- it would
+                            // stop pinning the leak it exists for.
                             throw new LineSenderException(
                                     "simulated mirror seed allocation failure (test only)");
                         }
-                        ensureSentDictCapacity((long) sentDictBytesLen + suffixLen);
                         engine.copyRecoveredSymbolSuffix(
                                 baseline, sentDictBytesAddr + sentDictBytesLen);
                         sentDictBytesLen += suffixLen;
                         sentDictCount += suffixCount;
                     }
                 }
             }
-            if (sentDictCount > 0) {
-                rebuildSentDictEntryIndex();
-            }
+            // The entry-ends index is NOT built here. sendDictCatchUp is its only
+            // reader and builds it on demand, so a recovered slot that drains without
+            // ever reconnecting -- the normal case -- never pays the O(n) walk nor
+            // retains the 4-bytes-per-symbol index.
         } catch (Throwable t) {
             releaseSentDictBytes();
             throw t;
@@ -2446,7 +2457,16 @@ private void accumulateSentDict(long payloadAddr, int payloadLen, int deltaStart
         if (deltaCount <= 0 || deltaStart > sentDictCount || deltaEnd <= sentDictCount) {
             return;
         }
-        ensureSentDictEntryIndex();
+        // Deliberately does NOT touch the entry-ends index. Only sendDictCatchUp reads
+        // it, so maintaining it here would put a store per new symbol on the per-frame
+        // I/O path -- and on the workload this feature targets (a new symbol per ROW)
+        // that is a store per row, plus 4 bytes per symbol retained for the connection's
+        // whole life, to serve a reconnect that may never happen. sendDictCatchUp calls
+        // ensureSentDictEntryIndex(), which notices sentDictIndexedCount has fallen
+        // behind sentDictCount and rebuilds; that O(n) walk runs at most once per
+        // reconnect and is dwarfed by shipping the same n bytes over the wire. A
+        // connection that never reconnects now never allocates the index at all.
+        //
         // Walk past the already-held prefix [deltaStart, sentDictCount), then copy
         // the new tail [sentDictCount, deltaEnd).
         int skip = sentDictCount - deltaStart;
@@ -2459,8 +2479,8 @@ private void accumulateSentDict(long payloadAddr, int payloadLen, int deltaStart
         }
         long regionStart = p;
         long newCount = deltaEnd - sentDictCount;
-        long requiredEntryCount = (long) sentDictCount + newCount;
-        ensureSentDictEntryEndsCapacity(requiredEntryCount);
+        // Walk the new tail only to find where it ends; the entry boundaries are not
+        // recorded here (see above -- the catch-up rebuilds them on demand).
         for (long i = 0; i < newCount; i++) {
             long len = readVarintAt(p, limit);
             p = varintEnd + len;
@@ -2469,9 +2489,6 @@ private void accumulateSentDict(long payloadAddr, int payloadLen, int deltaStart
                 // than corrupt the mirror.
                 return;
             }
-            Unsafe.getUnsafe().putInt(
-                    sentDictEntryEndsAddr + ((long) sentDictCount + i) * Integer.BYTES,
-                    sentDictBytesLen + (int) (p - regionStart));
         }
         int regionBytes = (int) (p - regionStart);
         // long sum: sentDictBytesLen + regionBytes can exceed Integer.MAX_VALUE on
@@ -2482,7 +2499,6 @@ private void accumulateSentDict(long payloadAddr, int payloadLen, int deltaStart
         Unsafe.getUnsafe().copyMemory(regionStart, sentDictBytesAddr + sentDictBytesLen, regionBytes);
         sentDictBytesLen += regionBytes;
         sentDictCount += (int) newCount;
-        sentDictIndexedCount = sentDictCount;
     }
 
     private void ensureSentDictCapacity(long required) {

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

@@ -221,8 +221,14 @@ public void close() {
         releaseRawStorage();
     }
 
-    private void appendRaw(long entryStart, int entryLen) {
-        long required = (long) runningRawLen + entryLen;
+    /**
+     * Appends one contiguous run of {@code count} wire entries -- {@code [len][utf8]}
+     * repeated, exactly as a delta section carries them -- in a single copy. Callers
+     * pass a whole frame's new-symbol suffix at once; see {@link #foldDelta} for why
+     * that suffix is always contiguous.
+     */
+    private void appendRaw(long addr, int len, int count) {
+        long required = (long) runningRawLen + len;
         if (required > MAX_RAW_BYTES) {
             throw new IllegalStateException("recovered symbol dictionary suffix exceeds maximum size "
                     + "[required=" + required + ", max=" + MAX_RAW_BYTES + ']');
@@ -235,9 +241,9 @@ private void appendRaw(long entryStart, int entryLen) {
             rawAddr = Unsafe.realloc(rawAddr, rawCapacity, (int) newCapacity, MemoryTag.NATIVE_DEFAULT);
             rawCapacity = (int) newCapacity;
         }
-        Unsafe.getUnsafe().copyMemory(entryStart, rawAddr + runningRawLen, entryLen);
-        runningRawLen += entryLen;
-        runningRawCount++;
+        Unsafe.getUnsafe().copyMemory(addr, rawAddr + runningRawLen, len);
+        runningRawLen += len;
+        runningRawCount += count;
     }
 
     private void foldDelta(long fsn, long p, long limit) {
@@ -292,7 +298,24 @@ private void foldDelta(long fsn, long p, long limit) {
             return;
         }
 
+        // runningCoverage is loop-invariant here -- it only advances after the walk --
+        // and id ascends from deltaStart, so `id >= runningCoverage` is a step
+        // predicate: the entries this frame contributes are always ONE contiguous run
+        // at the tail of its delta section. Note where that run starts and copy it in a
+        // single memcpy once the walk succeeds, rather than paying a bound check, a
+        // capacity check and a ~12-byte copyMemory per symbol. Recovery walks the whole
+        // backlog, and the workload this feature exists for introduces a new symbol per
+        // ROW, so that is millions of stub-dispatched small copies where one bulk copy
+        // per frame does the job.
+        //
+        // Deferring the copy past the markGap bail-outs also drops the partial prefix
+        // the per-entry version used to leave behind. That residue was already
+        // unreachable -- a gap pins coverage() at -1, and a later self-sufficient frame
+        // resets runningRawLen/runningRawCount -- so not writing it is equivalent, and
+        // leaves less state to reason about.
         long id = deltaStart;
+        long suffixStart = 0L;
+        int suffixCount = 0;
         for (long i = 0; i < deltaCount; i++, id++) {
             symbolEntriesVisited++;
             long entryStart = p;
@@ -310,9 +333,15 @@ private void foldDelta(long fsn, long p, long limit) {
             }
             p += symbolLen;
             if (id >= runningCoverage) {
-                appendRaw(entryStart, (int) (p - entryStart));
+                if (suffixCount == 0) {
+                    suffixStart = entryStart;
+                }
+                suffixCount++;
             }
         }
+        if (suffixCount > 0) {
+            appendRaw(suffixStart, (int) (p - suffixStart), suffixCount);
+        }
         if (deltaEnd > runningCoverage) {
             runningCoverage = deltaEnd;
         }

core/src/main/java/io/questdb/client/std/Crc32c.java

@@ -73,6 +73,14 @@ private Crc32c() {
      * mmap access fault into a catchable {@link InternalError}; the native
      * {@link #update} path cannot provide that guarantee because a SIGBUS raised
      * inside JNI aborts the JVM.
+     * <p>
+     * The hot loop consumes each 8-byte block with two {@code getInt}s rather than
+     * eight {@code getByte}s -- 9 loads per block instead of 16, for the same table
+     * work. {@code getInt} is an {@link Unsafe} intrinsic exactly as {@code getByte}
+     * is, so the fault-catchability above is unaffected. It does make the loop
+     * little-endian, which costs nothing here: the segment and dictionary formats this
+     * checksums are already little-endian throughout (their headers are read back with
+     * {@code getInt}), and the native twin this must agree with asserts the same.
      *
      * @param seed previous CRC value, or {@link #INIT} to start fresh
      * @param addr off-heap address of at least {@code len} readable bytes
@@ -84,22 +92,18 @@ public static int updateUnsafe(int seed, long addr, long len) {
         int crc = ~seed;
         int[] table = CRC32C_TABLE;
         while (len >= Long.BYTES) {
-            int b0 = Unsafe.getUnsafe().getByte(addr) & 0xFF;
-            int b1 = Unsafe.getUnsafe().getByte(addr + 1) & 0xFF;
-            int b2 = Unsafe.getUnsafe().getByte(addr + 2) & 0xFF;
-            int b3 = Unsafe.getUnsafe().getByte(addr + 3) & 0xFF;
-            int b4 = Unsafe.getUnsafe().getByte(addr + 4) & 0xFF;
-            int b5 = Unsafe.getUnsafe().getByte(addr + 5) & 0xFF;
-            int b6 = Unsafe.getUnsafe().getByte(addr + 6) & 0xFF;
-            int b7 = Unsafe.getUnsafe().getByte(addr + 7) & 0xFF;
-            crc = table[7 * 256 + ((crc ^ b0) & 0xFF)]
-                    ^ table[6 * 256 + (((crc >>> 8) ^ b1) & 0xFF)]
-                    ^ table[5 * 256 + (((crc >>> 16) ^ b2) & 0xFF)]
-                    ^ table[4 * 256 + (((crc >>> 24) ^ b3) & 0xFF)]
-                    ^ table[3 * 256 + b4]
-                    ^ table[2 * 256 + b5]
-                    ^ table[256 + b6]
-                    ^ table[b7];
+            // Little-endian: lo holds bytes 0..3 and hi bytes 4..7, ascending from the
+            // low byte, so the per-byte table lookups below stay in wire order.
+            int lo = Unsafe.getUnsafe().getInt(addr) ^ crc;
+            int hi = Unsafe.getUnsafe().getInt(addr + Integer.BYTES);
+            crc = table[7 * 256 + (lo & 0xFF)]
+                    ^ table[6 * 256 + ((lo >>> 8) & 0xFF)]
+                    ^ table[5 * 256 + ((lo >>> 16) & 0xFF)]
+                    ^ table[4 * 256 + (lo >>> 24)]
+                    ^ table[3 * 256 + (hi & 0xFF)]
+                    ^ table[2 * 256 + ((hi >>> 8) & 0xFF)]
+                    ^ table[256 + ((hi >>> 16) & 0xFF)]
+                    ^ table[hi >>> 24];
             addr += Long.BYTES;
             len -= Long.BYTES;
         }

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

@@ -822,12 +822,20 @@ public void testCommitMessageDoesNotShipUnpersistedLeakedSymbol() throws Excepti
     public void testPersistFailureSurfacesAsLineSenderException() throws Exception {
         // A dictionary write that fails mid-flush -- a full disk, an exhausted quota --
         // must reach the caller as a LineSenderException, like every other flush-path
-        // failure. PersistedSymbolDict throws a low-level IllegalStateException on a short
-        // write; persistNewSymbolsBeforePublish wraps it. Without the wrap the raw
-        // IllegalStateException sails straight past every user's
-        // `catch (LineSenderException)` around flush() and takes the application down.
+        // failure. PersistedSymbolDict throws a low-level IllegalStateException when it
+        // cannot grow its append window; persistNewSymbolsBeforePublish wraps it.
+        // Without the wrap the raw IllegalStateException sails straight past every
+        // user's `catch (LineSenderException)` around flush() and takes the application
+        // down.
         //
-        // Nothing could reach that translation before: PersistedSymbolDict has
+        // The fault is a REFUSED ff.allocate on the mmap append window, with the facade
+        // opting into isMmapAllowed(): that is both the real shape of ENOSPC here and
+        // the path production actually runs. A facade that inherits the default
+        // isMmapAllowed() (this == INSTANCE -> false) silently swaps the dictionary onto
+        // the positioned-write fallback, which production never executes -- so a
+        // short-write fault would prove this translation on dead code.
+        //
+        // Nothing could reach the translation at all before: PersistedSymbolDict has
         // facade-aware overloads, but CursorSendEngine called only the
         // FilesFacade.INSTANCE ones, so no test could inject a dictionary I/O fault
         // through the real producer path. The engine now takes a FilesFacade for exactly
@@ -842,7 +850,7 @@ public void testPersistFailureSurfacesAsLineSenderException() throws Exception {
                 String slot = Paths.get(sfDir, "default").toString();
                 Assert.assertEquals(0, io.questdb.client.std.Files.mkdir(sfDir,
                         io.questdb.client.std.Files.DIR_MODE_DEFAULT));
-                ShortDictWriteFacade ff = new ShortDictWriteFacade();
+                FullDiskDictFacade ff = new FullDiskDictFacade();
                 // The engine owns the dictionary; the fault facade reaches only it, so the
                 // segment files still write normally and the ONLY failure is the persist.
                 CursorSendEngine engine = new CursorSendEngine(
@@ -851,7 +859,7 @@ public void testPersistFailureSurfacesAsLineSenderException() throws Exception {
                 Sender sender = QwpWebSocketSender.connect(
                         "localhost", port, null, 0, 0, 0L, null, false, engine);
                 try {
-                    ff.armed = true; // the next dictionary append short-writes
+                    ff.armed = true; // the next dictionary append cannot grow its window
                     sender.table("m").symbol("s", "boom").longColumn("v", 1L).atNow();
                     try {
                         sender.flush();
@@ -1553,16 +1561,27 @@ public void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] dat
      * fills mid-flush. Offset 0 is left alone so the file header still writes -- only the
      * entry append fails, which is the path under test.
      */
-    private static final class ShortDictWriteFacade extends DelegatingFilesFacade {
+    /**
+     * Refuses to grow the symbol dictionary's mmap append window -- the production
+     * shape of a full disk, since PersistedSymbolDict reserves that window through
+     * {@code ff.allocate} and ENOSPC surfaces there as a refusal.
+     * <p>
+     * {@code isMmapAllowed()} opts IN deliberately. The inherited default is
+     * {@code this == INSTANCE}, so any wrapping facade otherwise routes the dictionary
+     * down the positioned-write fallback that production never executes -- injecting a
+     * fault would then silently replace the code under test.
+     */
+    private static final class FullDiskDictFacade extends DelegatingFilesFacade {
         boolean armed;
 
         @Override
-        public long write(int fd, long addr, long len, long offset) {
-            if (armed && offset > 0 && len > 1) {
-                armed = false;
-                return INSTANCE.write(fd, addr, len - 1, offset);
-            }
-            return INSTANCE.write(fd, addr, len, offset);
+        public boolean allocate(int fd, long size) {
+            return !armed && INSTANCE.allocate(fd, size);
+        }
+
+        @Override
+        public boolean isMmapAllowed() {
+            return true;
         }
     }