Persist symbol delta from frame, skip re-encode

M1: persistNewSymbolsBeforePublish re-encoded each new symbol into a
scratch buffer even though beginMessage had just written the identical
[len][utf8] bytes into the frame's symbol-dict delta section. The encoder
now records that entry region's offset and length; the producer writes
those bytes straight to the slot .symbol-dict via a new
PersistedSymbolDict.appendRawEntries, so a high-cardinality batch no longer
re-encodes what the wire frame already carries. The common path (durable
size == the frame's delta start id) takes the direct copy; a retry after a
failed publish, where the durable size has run ahead of the wire baseline,
falls back to re-encoding just the remaining suffix. The persisted bytes
are byte-identical either way.

M2: document, on the sendDictCatchUp oversized-entry terminal, that in a
heterogeneous or rolling-cap cluster a symbol accepted under a larger cap
can hit this hard stop on failover to a smaller-cap node and will not
self-recover if a later node advertises a larger cap. Behavior is unchanged
-- the terminal keeps the homogeneous common case livelock-free; this only
records the tradeoff and the two ways to relax it (an ingest-side
symbol-size bound, or a settle budget across reconnects before latching).

Adds testAppendRawEntriesMatchesAppendSymbols; the file-mode persist,
recovery, and failed-publish suites cover both the direct-copy and
re-encode branches end to end.

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

Author: glasstiger

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

@@ -39,6 +39,14 @@ public class QwpWebSocketEncoder implements QuietCloseable {
 
     private final QwpColumnWriter columnWriter = new QwpColumnWriter();
     private NativeBufferWriter buffer;
+    // Byte offsets, within the buffer, of the symbol-dict delta ENTRY region
+    // ([len][utf8]... only, without the two section varints) that beginMessage
+    // last wrote. Let the producer persist those bytes straight to the slot's
+    // .symbol-dict instead of re-encoding the same symbols (see
+    // QwpWebSocketSender.persistNewSymbolsBeforePublish). Valid until the next
+    // beginMessage; stored as offsets so they survive a buffer realloc.
+    private int deltaEntriesEnd;
+    private int deltaEntriesStart;
     // QWP ingress always advertises Gorilla timestamp encoding. The column
     // writer still emits a per-column encoding byte and falls back to raw
     // values when delta-of-delta overflows int32.
@@ -75,10 +83,12 @@ public void beginMessage(
         payloadStart = buffer.getPosition();
         buffer.putVarint(deltaStart);
         buffer.putVarint(deltaCount);
+        deltaEntriesStart = buffer.getPosition();
         for (int id = deltaStart; id < deltaStart + deltaCount; id++) {
             String symbol = globalDict.getSymbol(id);
             buffer.putString(symbol);
         }
+        deltaEntriesEnd = buffer.getPosition();
         columnWriter.setBuffer(buffer);
     }
 
@@ -122,6 +132,22 @@ public QwpBufferWriter getBuffer() {
         return buffer;
     }
 
+    /**
+     * Byte length of the symbol-dict delta ENTRY region ({@code [len][utf8]...},
+     * excluding the two section varints) that {@link #beginMessage} last wrote.
+     */
+    public int getDeltaEntriesLen() {
+        return deltaEntriesEnd - deltaEntriesStart;
+    }
+
+    /**
+     * Byte offset, within {@link #getBuffer()}, of the symbol-dict delta ENTRY
+     * region {@link #beginMessage} last wrote.
+     */
+    public int getDeltaEntriesStart() {
+        return deltaEntriesStart;
+    }
+
     public void setDeferCommit(boolean defer) {
         if (defer) {
             flags |= FLAG_DEFER_COMMIT;

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

@@ -3681,24 +3681,41 @@ private void persistNewSymbolsBeforePublish() {
         if (pd == null) {
             return;
         }
-        // Persist [pd.size() .. currentBatchMaxSymbolId] as ONE write.
+        // Persist [pd.size() .. currentBatchMaxSymbolId] as ONE write, BEFORE the
+        // frame is published.
         //
         // Resume from the dictionary's own durable size, NOT sentMaxSymbolId+1:
-        // appendSymbols advances pd.size() only after a full write, whereas
+        // the persist advances pd.size() only after a full write, whereas
         // sentMaxSymbolId only advances after the WHOLE frame is published (via
         // advanceSentMaxSymbolId, after activeBuffer.write). If a prior persist
-        // threw (a short write -- disk full/quota), the frame was not published
-        // and sentMaxSymbolId stayed put, while the symbols before the failing
-        // one are already on disk. Keying the resume point off sentMaxSymbolId+1
-        // would then re-append that persisted prefix on the retry, duplicating
-        // entries and corrupting the dense id->symbol mapping recovery relies on
-        // (position i must be symbol id i). pd.size() resumes exactly past what is
-        // already durable (the write overwrites any torn trailing bytes at
-        // appendOffset), so the write-ahead is idempotent under retry. In the
-        // happy path pd.size() equals sentMaxSymbolId+1, so the range -- and the
-        // behaviour -- are unchanged; the single positioned write just replaces
-        // the previous one-syscall-per-new-symbol loop.
-        pd.appendSymbols(globalSymbolDictionary, pd.size(), currentBatchMaxSymbolId);
+        // threw (short write -- disk full/quota) or the publish threw, the frame
+        // was not published and sentMaxSymbolId stayed put, while the symbols
+        // before the failure are already on disk. Keying the resume point off
+        // sentMaxSymbolId+1 would re-append that persisted prefix on the retry,
+        // duplicating entries and corrupting the dense id->symbol mapping recovery
+        // relies on (position i must be symbol id i). pd.size() resumes exactly
+        // past what is already durable, so the write-ahead is idempotent.
+        int from = pd.size();
+        if (currentBatchMaxSymbolId < from) {
+            return; // nothing new to persist (warm batch, or an idempotent retry)
+        }
+        // Fast path: the frame the encoder just built already holds these symbols
+        // in its delta section as [len][utf8]... -- byte-identical to what
+        // PersistedSymbolDict stores. In the common case pd.size() equals the
+        // frame's delta start id (sentMaxSymbolId+1), so persist those bytes
+        // straight from the frame instead of re-encoding the symbols. After a
+        // failed publish the durable size has run ahead of the wire baseline, so
+        // the frame's delta covers MORE than remains to persist; then re-encode
+        // just the [from .. currentBatchMaxSymbolId] suffix.
+        if (from == sentMaxSymbolId + 1) {
+            QwpBufferWriter buffer = encoder.getBuffer();
+            pd.appendRawEntries(
+                    buffer.getBufferPtr() + encoder.getDeltaEntriesStart(),
+                    encoder.getDeltaEntriesLen(),
+                    currentBatchMaxSymbolId - from + 1);
+        } else {
+            pd.appendSymbols(globalSymbolDictionary, from, currentBatchMaxSymbolId);
+        }
     }
 
     private void resetSymbolDictStateForNewConnection() {

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

@@ -2192,6 +2192,15 @@ private int sendDictCatchUp() {
                 // Latch a terminal (the data must be resent after the cap is
                 // raised) rather than calling fail() -- which, from inside the
                 // catch-up, would re-enter connectLoop (see CatchUpSendException).
+                //
+                // Tradeoff (heterogeneous / rolling-cap clusters): a symbol
+                // accepted under a larger/absent cap can hit this on failover to a
+                // smaller-cap node, and the hard terminal does NOT self-recover if
+                // a later node advertises a larger cap -- the producer must be
+                // resumed after the data is resent (or the cap raised). Bounding
+                // symbol size at ingest, or a settle budget across reconnects
+                // before latching, would relax this, but both are larger changes;
+                // the terminal keeps the homogeneous common case livelock-free.
                 LineSenderException err = new LineSenderException(
                         "symbol dictionary entry too large for the server batch cap during catch-up ["
                                 + "entryBytes=" + entryBytes + ", budget=" + budget + ']');

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

@@ -148,6 +148,30 @@ public static void removeOrphan(String slotDir) {
         Files.remove(slotDir + "/" + FILE_NAME);
     }
 
+    /**
+     * Appends {@code count} already-encoded entries -- {@code [len varint][utf8]...},
+     * the exact layout {@link #appendSymbols} produces -- verbatim from {@code addr}
+     * in a single write. The producer uses this when it already holds those bytes
+     * (the symbol-dict delta section the frame encoder just wrote), so it does not
+     * re-encode the same symbols. Writes {@code len} bytes and advances {@code size}
+     * by {@code count}. Same durability/idempotency contract as {@link #appendSymbols}:
+     * no fsync, and a short write throws WITHOUT advancing {@code size}/{@code
+     * appendOffset}, so a retry keyed off {@link #size()} re-persists the same range
+     * at the same offset. No-op when the range is empty or the dictionary is closed.
+     */
+    public void appendRawEntries(long addr, int len, int count) {
+        if (closed || count <= 0 || len <= 0) {
+            return;
+        }
+        long written = Files.write(fd, addr, len, appendOffset);
+        if (written != len) {
+            throw new IllegalStateException("short write to " + FILE_NAME
+                    + " [expected=" + len + ", actual=" + written + ']');
+        }
+        appendOffset += len;
+        size += count;
+    }
+
     /**
      * Appends one symbol, extending the on-disk dictionary. The caller appends a
      * frame's new symbols BEFORE publishing that frame, so the write ordering

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/PersistedSymbolDictTest.java

@@ -91,6 +91,62 @@ public void testAppendPersistsAcrossReopen() throws Exception {
         });
     }
 
+    @Test
+    public void testAppendRawEntriesMatchesAppendSymbols() throws Exception {
+        // M1: the producer persists the frame's already-encoded delta bytes via
+        // appendRawEntries instead of re-encoding the symbols. Those bytes are the
+        // same [len][utf8]... layout appendSymbols writes, so both must produce an
+        // identical, recoverable dictionary. Encode a range with appendSymbols,
+        // reopen to grab its on-disk entry bytes, replay them through
+        // appendRawEntries into a fresh dict, and assert the recovered symbols
+        // match -- including an empty entry mid-range.
+        assertMemoryLeak(() -> {
+            Path src = Files.createTempDirectory("qwp-symdict-src");
+            Path dst = Files.createTempDirectory("qwp-symdict-dst");
+            try {
+                GlobalSymbolDictionary dict = new GlobalSymbolDictionary();
+                dict.getOrAddSymbol("AAPL"); // id 0
+                dict.getOrAddSymbol("");     // id 1 -- empty entry mid-range
+                dict.getOrAddSymbol("MSFT"); // id 2
+
+                PersistedSymbolDict encoded = PersistedSymbolDict.open(src.toString());
+                encoded.appendSymbols(dict, 0, 2);
+                encoded.close();
+
+                // Reopen to obtain the on-disk entry region [len][utf8]... verbatim,
+                // then replay it byte-for-byte into a fresh dict via appendRawEntries.
+                PersistedSymbolDict reopened = PersistedSymbolDict.open(src.toString());
+                try {
+                    PersistedSymbolDict raw = PersistedSymbolDict.open(dst.toString());
+                    try {
+                        raw.appendRawEntries(reopened.loadedEntriesAddr(),
+                                reopened.loadedEntriesLen(), reopened.size());
+                        Assert.assertEquals(3, raw.size());
+                    } finally {
+                        raw.close();
+                    }
+                } finally {
+                    reopened.close();
+                }
+
+                // The raw-appended dict must recover the same dense symbols.
+                PersistedSymbolDict recovered = PersistedSymbolDict.open(dst.toString());
+                try {
+                    Assert.assertEquals(3, recovered.size());
+                    ObjList<String> symbols = recovered.readLoadedSymbols();
+                    Assert.assertEquals("AAPL", symbols.getQuick(0));
+                    Assert.assertEquals("", symbols.getQuick(1));
+                    Assert.assertEquals("MSFT", symbols.getQuick(2));
+                } finally {
+                    recovered.close();
+                }
+            } finally {
+                rmDir(src);
+                rmDir(dst);
+            }
+        });
+    }
+
     @Test
     public void testAppendSymbolsBatchWritesDenseRange() throws Exception {
         // appendSymbols persists a whole id range in one write (the hot-path