Let QwpTableBuffer handle addSymbol fully

This required injecting the Sender into QwpTableBuffer, so it can
add a symbol to the global symbol table

Author: mtopolnik

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

@@ -42,6 +42,7 @@
 import io.questdb.client.std.ObjList;
 import io.questdb.client.std.bytes.DirectByteSlice;
 import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.TestOnly;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -696,6 +697,21 @@ public int getMaxSentSymbolId() {
         return maxSentSymbolId;
     }
 
+    /**
+     * Registers a symbol value in the global dictionary and returns its global ID.
+     * Called from {@link QwpTableBuffer.ColumnBuffer#addSymbol(CharSequence)}.
+     *
+     * @param symbol the symbol value to register
+     * @return the global symbol ID
+     */
+    public int getOrAddGlobalSymbol(String symbol) {
+        int globalId = globalSymbolDictionary.getOrAddSymbol(symbol);
+        if (globalId > currentBatchMaxSymbolId) {
+            currentBatchMaxSymbolId = globalId;
+        }
+        return globalId;
+    }
+
     /**
      * Returns the number of pending rows not yet flushed.
      * For testing.
@@ -861,19 +877,7 @@ public QwpWebSocketSender symbol(CharSequence columnName, CharSequence value) {
         checkNotClosed();
         checkTableSelected();
         QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(columnName), TYPE_SYMBOL, true);
-
-        if (value != null) {
-            // Register symbol in global dictionary and track max ID for delta calculation
-            String symbolValue = value.toString();
-            int globalId = globalSymbolDictionary.getOrAddSymbol(symbolValue);
-            if (globalId > currentBatchMaxSymbolId) {
-                currentBatchMaxSymbolId = globalId;
-            }
-            // Store global ID in the column buffer
-            col.addSymbolWithGlobalId(symbolValue, globalId);
-        } else {
-            col.addSymbol(null);
-        }
+        col.addSymbol(value);
         return this;
     }
 
@@ -891,7 +895,7 @@ public QwpWebSocketSender table(CharSequence tableName) {
         currentTableName = tableName.toString();
         currentTableBuffer = tableBuffers.get(currentTableName);
         if (currentTableBuffer == null) {
-            currentTableBuffer = new QwpTableBuffer(currentTableName);
+            currentTableBuffer = new QwpTableBuffer(currentTableName, this);
             tableBuffers.put(currentTableName, currentTableBuffer);
         }
         // Both modes accumulate rows until flush

core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java

@@ -29,6 +29,7 @@
 import io.questdb.client.cutlass.line.array.ArrayBufferAppender;
 import io.questdb.client.cutlass.line.array.DoubleArray;
 import io.questdb.client.cutlass.line.array.LongArray;
+import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
 import io.questdb.client.std.CharSequenceIntHashMap;
 import io.questdb.client.std.Chars;
 import io.questdb.client.std.Decimal128;
@@ -59,6 +60,7 @@ public class QwpTableBuffer implements QuietCloseable {
 
     private final CharSequenceIntHashMap columnNameToIndex;
     private final ObjList<ColumnBuffer> columns;
+    private final QwpWebSocketSender sender;
     private final String tableName;
     private QwpColumnDef[] cachedColumnDefs;
     private int columnAccessCursor; // tracks expected next column index
@@ -70,7 +72,18 @@ public class QwpTableBuffer implements QuietCloseable {
     private boolean schemaHashComputed;
 
     public QwpTableBuffer(String tableName) {
+        this(tableName, null);
+    }
+
+    /**
+     * Use this constructor overload to allow writing to a symbol column.
+     * {@link ColumnBuffer#addSymbol(CharSequence)} needs the sender to
+     * call {@link QwpWebSocketSender#getOrAddGlobalSymbol(String)}, registering
+     * the symbol in the global dictionary shared with the server.
+     */
+    public QwpTableBuffer(String tableName, QwpWebSocketSender sender) {
         this.tableName = tableName;
+        this.sender = sender;
         this.columns = new ObjList<>();
         this.columnNameToIndex = new CharSequenceIntHashMap();
         this.rowCount = 0;
@@ -322,6 +335,7 @@ private static void assertColumnType(CharSequence name, byte type, ColumnBuffer
 
     private ColumnBuffer createColumn(CharSequence name, byte type, boolean nullable) {
         ColumnBuffer col = new ColumnBuffer(Chars.toString(name), type, nullable);
+        col.sender = sender;
         int index = columns.size();
         col.index = index;
         columns.add(col);
@@ -534,6 +548,7 @@ public static class ColumnBuffer implements QuietCloseable {
         private int nullBufCapRows;
         // Off-heap null bitmap (bit-packed, 1 bit per row)
         private long nullBufPtr;
+        private QwpWebSocketSender sender;
         private int size;         // Total row count (including nulls)
         private OffHeapAppendMemory stringData;
         // Off-heap storage for string/varchar column data
@@ -1003,6 +1018,12 @@ public void addSymbol(CharSequence value) {
                 addNull();
                 return;
             }
+            if (sender != null) {
+                String symbolValue = value.toString();
+                int globalId = sender.getOrAddGlobalSymbol(symbolValue);
+                addSymbolWithGlobalId(symbolValue, globalId);
+                return;
+            }
             ensureNullBitmapCapacity();
             int idx = getOrAddLocalSymbol(value);
             dataBuffer.putInt(idx);
@@ -1015,7 +1036,6 @@ public void addSymbolUtf8(long ptr, int len) {
                 addNull();
                 return;
             }
-            ensureNullBitmapCapacity();
             StringSink lookupSink = symbolLookupSink;
             if (lookupSink == null) {
                 symbolLookupSink = lookupSink = new StringSink(Math.max(16, len));
@@ -1027,6 +1047,13 @@ public void addSymbolUtf8(long ptr, int len) {
                 Utf8s.stringFromUtf8Bytes(ptr, ptr + len);
                 throw new AssertionError("unreachable");
             }
+            if (sender != null) {
+                String symbolValue = lookupSink.toString();
+                int globalId = sender.getOrAddGlobalSymbol(symbolValue);
+                addSymbolWithGlobalId(symbolValue, globalId);
+                return;
+            }
+            ensureNullBitmapCapacity();
             int idx = getOrAddLocalSymbol(lookupSink);
             dataBuffer.putInt(idx);
             valueCount++;