Harden the symbol-dict fd close and split sizing

Three review follow-ups on the delta symbol-dictionary change:

- PersistedSymbolDict relinquishes fd (sets it to -1) before each
  in-branch ff.close and guards the catch-block closes with fd >= 0,
  so a throwing in-branch close cannot double-close via the catch.
  Completes the null-before-free discipline the buffers already use.

- QwpWebSocketEncoderTest pins the split path's context-free-body
  invariant: a table's body bytes must encode identically solo and
  combined, since flushPendingRowsSplit sizes each frame from the
  combined encode's splitFrameBodyBytes rather than re-measuring. A
  future column encoder that carried cross-table state now fails here
  instead of silently stranding a deferred prefix.

- CursorSendEngine imports QwpConstants instead of spelling it out
  fully-qualified eight times, matching its sibling send loop.

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

Author: glasstiger

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

@@ -24,6 +24,7 @@
 
 package io.questdb.client.cutlass.qwp.client.sf.cursor;
 
+import io.questdb.client.cutlass.qwp.protocol.QwpConstants;
 import io.questdb.client.std.Compat;
 import io.questdb.client.std.Files;
 import io.questdb.client.std.ObjList;
@@ -307,10 +308,10 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                 // commit would resurrect half a transaction -- see the WARN
                 // below. Computed before the I/O loop or producer append.
                 this.recoveredCommitBoundaryFsn = recovered.findLastFsnWithoutPayloadFlag(
-                        io.questdb.client.cutlass.qwp.protocol.QwpConstants.HEADER_OFFSET_FLAGS,
-                        io.questdb.client.cutlass.qwp.protocol.QwpConstants.FLAG_DEFER_COMMIT,
-                        io.questdb.client.cutlass.qwp.protocol.QwpConstants.MAGIC_MESSAGE,
-                        io.questdb.client.cutlass.qwp.protocol.QwpConstants.HEADER_SIZE
+                        QwpConstants.HEADER_OFFSET_FLAGS,
+                        QwpConstants.FLAG_DEFER_COMMIT,
+                        QwpConstants.MAGIC_MESSAGE,
+                        QwpConstants.HEADER_SIZE
                 );
                 if (publishedFsn >= 0 && recoveredCommitBoundaryFsn < publishedFsn) {
                     this.recoveredOrphanTipFsn = publishedFsn;
@@ -332,10 +333,10 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                 // returns 0 when no such frame carries a symbol, yielding -1 here.
                 // Computed before the I/O loop or producer append; single-threaded.
                 this.recoveredMaxSymbolId = recovered.maxSymbolDeltaEnd(
-                        io.questdb.client.cutlass.qwp.protocol.QwpConstants.MAGIC_MESSAGE,
-                        io.questdb.client.cutlass.qwp.protocol.QwpConstants.HEADER_OFFSET_FLAGS,
-                        io.questdb.client.cutlass.qwp.protocol.QwpConstants.FLAG_DELTA_SYMBOL_DICT,
-                        io.questdb.client.cutlass.qwp.protocol.QwpConstants.HEADER_SIZE,
+                        QwpConstants.MAGIC_MESSAGE,
+                        QwpConstants.HEADER_OFFSET_FLAGS,
+                        QwpConstants.FLAG_DELTA_SYMBOL_DICT,
+                        QwpConstants.HEADER_SIZE,
                         recoveredCommitBoundaryFsn) - 1L;
             } else {
                 // Fresh start with no recovered segments. Any stale

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

@@ -532,7 +532,9 @@ private static PersistedSymbolDict openExisting(FilesFacade ff, String filePath,
                 LOG.warn("symbol dict {} unreadable, bad magic or unknown version; recreating", filePath);
                 Unsafe.free(buf, len, MemoryTag.NATIVE_DEFAULT);
                 buf = 0L; // null after free so the catch below cannot double-free if ff.close throws
-                ff.close(fd);
+                int fdToClose = fd;
+                fd = -1; // relinquish before close so the catch cannot double-close if close throws
+                ff.close(fdToClose);
                 return null;
             }
             // Parse complete, CRC-valid entries after the header; stop at the first
@@ -622,7 +624,9 @@ private static PersistedSymbolDict openExisting(FilesFacade ff, String filePath,
                     entriesAddr = 0L;
                     entriesLen = 0;
                 }
-                ff.close(fd);
+                int fdToClose = fd;
+                fd = -1; // relinquish before close so the catch cannot double-close if close throws
+                ff.close(fdToClose);
                 return null;
             }
             return new PersistedSymbolDict(ff, fd, validLen, count, entriesAddr, entriesLen);
@@ -638,7 +642,9 @@ private static PersistedSymbolDict openExisting(FilesFacade ff, String filePath,
             if (entriesAddr != 0L) {
                 Unsafe.free(entriesAddr, entriesLen, MemoryTag.NATIVE_DEFAULT);
             }
-            ff.close(fd);
+            if (fd >= 0) { // a branch that already closed fd relinquished it to -1
+                ff.close(fd);
+            }
             LOG.warn("symbol dict {} recovery failed ({}); recreating", filePath, String.valueOf(t));
             return null;
         }
@@ -660,7 +666,9 @@ private static PersistedSymbolDict openFresh(FilesFacade ff, String filePath) {
             Unsafe.getUnsafe().putByte(hdr + 7, (byte) 0);
             long written = ff.write(fd, hdr, HEADER_SIZE, 0);
             if (written != HEADER_SIZE) {
-                ff.close(fd);
+                int fdToClose = fd;
+                fd = -1; // relinquish before close so the catch cannot double-close if close throws
+                ff.close(fdToClose);
                 ff.remove(filePath);
                 LOG.warn("symbol dict {} header write failed; proceeding without it", filePath);
                 return null;
@@ -670,7 +678,9 @@ private static PersistedSymbolDict openFresh(FilesFacade ff, String filePath) {
             // throwing; the Unsafe puts target a valid 8-byte buffer and an 8-byte
             // malloc cannot realistically OOM), but close the fd against a future
             // edit so it cannot leak -- mirroring openExisting's error handling.
-            ff.close(fd);
+            if (fd >= 0) { // the header-write branch relinquished fd to -1 before closing
+                ff.close(fd);
+            }
             LOG.warn("symbol dict {} creation failed ({}); proceeding without it", filePath, String.valueOf(t));
             return null;
         } finally {

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

@@ -1284,6 +1284,74 @@ public void testReset() throws Exception {
         });
     }
 
+    @Test
+    public void testTableBodyEncodingIsContextFree() throws Exception {
+        // The split-flush path (QwpWebSocketSender.flushPendingRowsSplit) sizes each
+        // per-table frame ARITHMETICALLY from splitFrameBodyBytes -- the body byte
+        // count captured during the COMBINED encode in flushPendingRows -- instead of
+        // re-encoding to measure. That is sound only while a table's body bytes are
+        // context-free: identical whether the table is encoded solo or as the k-th
+        // table after other tables and the delta section. Today every column encoder
+        // is stateless and symbol cells carry absolute global ids, so the property
+        // holds; this pins it so a future column encoder that ever carried
+        // cross-table state (which would break the arithmetic sizing and could strand
+        // a deferred prefix mid-split) fails loudly here rather than silently in
+        // production.
+        assertMemoryLeak(() -> {
+            try (QwpWebSocketEncoder encoder = new QwpWebSocketEncoder();
+                 QwpTableBuffer t0 = new QwpTableBuffer("alpha");
+                 QwpTableBuffer t1 = new QwpTableBuffer("bravo");
+                 QwpTableBuffer t2 = new QwpTableBuffer("charlie")) {
+                GlobalSymbolDictionary dict = new GlobalSymbolDictionary();
+                int aapl = dict.getOrAddSymbol("AAPL"); // 0
+                int goog = dict.getOrAddSymbol("GOOG"); // 1
+                int msft = dict.getOrAddSymbol("MSFT"); // 2
+
+                // Distinct schemas + a symbol column (absolute global ids) so a
+                // positional / cross-table encoder bug would shift the body bytes.
+                t0.getOrCreateColumn("sym", TYPE_SYMBOL, false).addSymbolWithGlobalId("AAPL", aapl);
+                t0.getOrCreateColumn("v", TYPE_LONG, false).addLong(1L);
+                t0.nextRow();
+
+                t1.getOrCreateColumn("sym", TYPE_SYMBOL, false).addSymbolWithGlobalId("GOOG", goog);
+                t1.getOrCreateColumn("d", TYPE_DOUBLE, false).addDouble(2.5);
+                t1.getOrCreateColumn("s", TYPE_VARCHAR, true).addString("hello");
+                t1.nextRow();
+
+                t2.getOrCreateColumn("sym", TYPE_SYMBOL, false).addSymbolWithGlobalId("MSFT", msft);
+                t2.getOrCreateDesignatedTimestampColumn(TYPE_TIMESTAMP).addLong(1_000_000L);
+                t2.nextRow();
+
+                QwpTableBuffer[] tables = {t0, t1, t2};
+                int confirmedMaxId = -1;
+                int batchMaxId = 2;
+
+                // Combined encode: capture each table's body bytes exactly as
+                // flushPendingRows' splitFrameBodyBytes does (position delta per addTable).
+                int[] combinedBody = new int[tables.length];
+                encoder.beginMessage(tables.length, dict, confirmedMaxId, batchMaxId);
+                int bodyStart = encoder.getBuffer().getPosition();
+                for (int i = 0; i < tables.length; i++) {
+                    encoder.addTable(tables[i]);
+                    int bodyEnd = encoder.getBuffer().getPosition();
+                    combinedBody[i] = bodyEnd - bodyStart;
+                    bodyStart = bodyEnd;
+                }
+
+                // Solo encode each table under the SAME baseline/batch max, as the
+                // split publish loop does, and assert the body bytes match the capture.
+                for (int i = 0; i < tables.length; i++) {
+                    encoder.beginMessage(1, dict, confirmedMaxId, batchMaxId);
+                    int soloBodyStart = encoder.getBuffer().getPosition();
+                    encoder.addTable(tables[i]);
+                    int soloBody = encoder.getBuffer().getPosition() - soloBodyStart;
+                    Assert.assertEquals("table " + i + " body must encode identically solo vs combined "
+                            + "(splitFrameBodyBytes relies on a context-free body)", combinedBody[i], soloBody);
+                }
+            }
+        });
+    }
+
     @Test
     public void testVersionByteInHeader() throws Exception {
         assertMemoryLeak(() -> {