feat(qwp): stop resending the full symbol dictionary on every message

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

@@ -50,6 +50,37 @@ public GlobalSymbolDictionary(int initialCapacity) {
         this.idToSymbol = new ObjList<>(initialCapacity);
     }
 
+    /**
+     * Appends {@code symbol} at the next sequential id, matching a recovered /
+     * persisted dictionary's dense id order, WITHOUT de-duplicating.
+     * <p>
+     * Recovery ({@code QwpWebSocketSender.seedGlobalDictionaryFromPersisted})
+     * replays the persisted entries in id order to rebuild this dictionary. It must
+     * NOT collapse two source strings that decode to the same characters, because
+     * the persisted {@code .symbol-dict}, the on-wire delta and the I/O-thread
+     * catch-up mirror all key on the entry POSITION (id), not on the string. The
+     * only strings that collide this way are malformed lone UTF-16 surrogates,
+     * which the UTF-8 encoder maps to {@code '?'}: {@link #getOrAddSymbol} would
+     * de-dup them and leave this dictionary SHORTER than the persisted entry count,
+     * desyncing the producer's delta baseline from the catch-up mirror (which uses
+     * {@code pd.size()}) and silently misattributing later symbols. Appending
+     * unconditionally keeps {@link #size()} equal to that count. The reverse lookup
+     * keeps the highest id for a colliding string, which is harmless: both ids
+     * encode to the same bytes, so resolving either is equivalent.
+     *
+     * @param symbol the recovered symbol string (must not be null)
+     * @return the id assigned (the previous {@link #size()})
+     */
+    public int addRecoveredSymbol(String symbol) {
+        if (symbol == null) {
+            throw new IllegalArgumentException("symbol cannot be null");
+        }
+        int newId = idToSymbol.size();
+        symbolToId.put(symbol, newId);
+        idToSymbol.add(symbol);
+        return newId;
+    }
+
     /**
      * Clears all symbols from the dictionary.
      * <p>

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

@@ -76,6 +76,21 @@ public static int varintSize(long value) {
         return (64 - Long.numberOfLeadingZeros(value) + 6) / 7;
     }
 
+    /**
+     * Writes {@code value} as an unsigned LEB128 varint directly at native address
+     * {@code addr} and returns the address just past the last byte. The canonical
+     * raw-address varint writer shared by the SF cursor's persisted dictionary and
+     * catch-up frame builder.
+     */
+    public static long writeVarint(long addr, long value) {
+        while (value > 0x7F) {
+            Unsafe.getUnsafe().putByte(addr++, (byte) ((value & 0x7F) | 0x80));
+            value >>>= 7;
+        }
+        Unsafe.getUnsafe().putByte(addr++, (byte) value);
+        return addr;
+    }
+
     @Override
     public void close() {
         if (bufferPtr != 0) {
@@ -336,11 +351,7 @@ public void skip(int bytes) {
     }
 
     private static void writeVarintDirect(long addr, long value) {
-        while (value > 0x7F) {
-            Unsafe.getUnsafe().putByte(addr++, (byte) ((value & 0x7F) | 0x80));
-            value >>>= 7;
-        }
-        Unsafe.getUnsafe().putByte(addr, (byte) value);
+        writeVarint(addr, value);
     }
 
     private void encodeUtf8(CharSequence value, int utf8Len) {

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;