Fix stale datagram estimate on timestamp rename

TableOptionsImpl.designatedTimestamp() invalidated the cached base
estimate only on the first name set. QwpTableBuffer permits renaming
the designated timestamp while rowCount == 0, and the cache (keyed on
column count alone) folds in the trailer size, which depends on the
name length. After a flush kept the buffer's name and cache alive, a
rename reused the shorter-trailer estimate, so a bounded UDP sender
could emit datagrams above maxDatagramSize. The fix invalidates the
cache on every accepted set/change; the committedDatagramEstimate bump
stays under previousName == null because a rename requires
rowCount == 0, where the committed estimate is already 0.

The regression test asserts committedDatagramEstimate covers the
actual datagram after a short-name flush and a long-name rename
(fails at 43 vs 155 bytes without the fix).

A new encoder test mirrors flushPendingRowsSplit: consecutive
single-table messages on a reused encoder each carry the
FLAG_TABLE_OPTIONS flag and trailer, and a table without options does
not inherit them.

The FLAG_TABLE_OPTIONS javadoc now records the forward-compat
guarantee the ungated emission relies on: pre-table-options servers
validate only magic, version and payload length, and iterate exactly
the header-declared table count, so unknown flag bits and trailer
bytes after the last table body are ignored (verified against the
server decoder at 51b235b9ca~1).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: jerrinot

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

@@ -1550,15 +1550,17 @@ public TableOptions designatedTimestamp(CharSequence columnName) {
             }
             String previousName = tableBuffer.getDesignatedTimestampName();
             tableBuffer.setDesignatedTimestampName(columnName);
-            if (previousName == null) {
-                if (currentTableHeadroomState != null) {
-                    currentTableHeadroomState.invalidateBaseEstimate();
-                }
-                if (trackDatagramEstimate && committedDatagramEstimate > 0) {
-                    committedDatagramEstimate += QwpColumnWriter.getSingleTableOptionsTrailerSize(
-                            tableBuffer.getDesignatedTimestampName()
-                    );
-                }
+            // The cached base estimate folds in the trailer size, which depends on
+            // the name length, so every accepted set/change must invalidate it. The
+            // committedDatagramEstimate bump stays under previousName == null: a name
+            // change requires rowCount == 0, where the committed estimate is already 0.
+            if (currentTableHeadroomState != null) {
+                currentTableHeadroomState.invalidateBaseEstimate();
+            }
+            if (previousName == null && trackDatagramEstimate && committedDatagramEstimate > 0) {
+                committedDatagramEstimate += QwpColumnWriter.getSingleTableOptionsTrailerSize(
+                        tableBuffer.getDesignatedTimestampName()
+                );
             }
             return this;
         }

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

@@ -52,6 +52,14 @@ public final class QwpConstants {
     public static final byte FLAG_GORILLA = 0x04;
     /**
      * Flag bit: per-table options trailer enabled.
+     * <p>
+     * The client sets this flag whenever a table option is declared, without
+     * gating on the negotiated capability (UDP has no handshake, and
+     * store-and-forward frames must replay to arbitrary servers). This relies
+     * on the server-side forward-compat guarantee: pre-table-options servers
+     * validate only magic, version and payload length (unknown flag bits are
+     * ignored) and iterate exactly the header-declared table count, so trailer
+     * bytes after the last table body are never read.
      */
     public static final byte FLAG_TABLE_OPTIONS = 0x20;
     /**

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

@@ -1745,6 +1745,40 @@ public void testTableOptionsOnlyRowIsDiscarded() throws Exception {
         });
     }
 
+    @Test
+    public void testTableOptionsRenameInvalidatesCachedDatagramEstimate() throws Exception {
+        assertMemoryLeak(() -> {
+            CapturingNetworkFacade nf = new CapturingNetworkFacade();
+            StringBuilder longName = new StringBuilder();
+            for (int i = 0; i < 120; i++) {
+                longName.append('t');
+            }
+            int maxDatagramSize = 1024;
+            try (QwpUdpSender sender = new QwpUdpSender(nf, 0, 0, 9000, 1, maxDatagramSize)) {
+                sender.table("t").tableOptions().designatedTimestamp("a");
+                sender.longColumn("x", 1).atNow();
+                sender.flush();
+                Assert.assertEquals(1, nf.packets.size());
+
+                // rowCount == 0 after the flush, so re-declaring a different (much
+                // longer) name is accepted; the cached base estimate sized for "a"
+                // must not survive the rename
+                sender.table("t").tableOptions().designatedTimestamp(longName);
+                sender.longColumn("x", 2).atNow();
+                long estimate = sender.committedDatagramEstimateForTest();
+                sender.flush();
+
+                Assert.assertEquals(2, nf.packets.size());
+                int actual = nf.packets.get(1).length;
+                Assert.assertTrue(
+                        "committed estimate must cover the actual datagram [estimate=" + estimate + ", actual=" + actual + ']',
+                        estimate >= actual
+                );
+                assertPacketsWithinLimit(new RunResult(nf.packets, nf.lengths, nf.sendCount), maxDatagramSize);
+            }
+        });
+    }
+
     @Test
     public void testTableOptionsRetainedReferenceRetargetsOnNextTableOptionsCall() throws Exception {
         assertMemoryLeak(() -> {

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

@@ -1402,6 +1402,86 @@ public void testTableOptionsTrailerEncodingWithMixedPresence() throws Exception
         });
     }
 
+    @Test
+    public void testTableOptionsTrailerOnConsecutiveSplitMessages() throws Exception {
+        assertMemoryLeak(() -> {
+            try (QwpWebSocketEncoder encoder = new QwpWebSocketEncoder();
+                 QwpTableBuffer first = new QwpTableBuffer("first");
+                 QwpTableBuffer second = new QwpTableBuffer("second");
+                 QwpTableBuffer third = new QwpTableBuffer("third")) {
+                first.setDesignatedTimestampName("ts");
+                first.getOrCreateColumn("x", TYPE_LONG, false).addLong(1);
+                first.nextRow();
+                second.getOrCreateColumn("x", TYPE_LONG, false).addLong(2);
+                second.nextRow();
+                third.setDesignatedTimestampName("event_ts");
+                third.getOrCreateColumn("x", TYPE_LONG, false).addLong(3);
+                third.nextRow();
+
+                // mirrors flushPendingRowsSplit: one single-table message per
+                // table on the same reused encoder
+                GlobalSymbolDictionary dict = new GlobalSymbolDictionary();
+
+                encoder.beginMessage(1, dict, -1, -1);
+                encoder.addTable(first);
+                int size1 = encoder.finishMessage();
+                long address1 = encoder.getBuffer().getBufferPtr();
+                Assert.assertEquals(
+                        FLAG_TABLE_OPTIONS,
+                        (byte) (Unsafe.getUnsafe().getByte(address1 + HEADER_OFFSET_FLAGS) & FLAG_TABLE_OPTIONS)
+                );
+                Assert.assertEquals(size1 - HEADER_SIZE, Unsafe.getUnsafe().getInt(address1 + 8));
+                byte[] expectedTrailer1 = {
+                        4,
+                        TABLE_OPTION_TAG_DESIGNATED_TIMESTAMP_NAME,
+                        2, 't', 's',
+                        5, 0, 0, 0
+                };
+                long trailerAddress1 = address1 + size1 - expectedTrailer1.length;
+                for (int i = 0; i < expectedTrailer1.length; i++) {
+                    Assert.assertEquals(
+                            "message 1 trailer byte " + i,
+                            expectedTrailer1[i],
+                            Unsafe.getUnsafe().getByte(trailerAddress1 + i)
+                    );
+                }
+
+                // a table without options must not inherit the previous
+                // message's flag or trailer
+                encoder.beginMessage(1, dict, -1, -1);
+                encoder.addTable(second);
+                int size2 = encoder.finishMessage();
+                long address2 = encoder.getBuffer().getBufferPtr();
+                Assert.assertEquals(0, Unsafe.getUnsafe().getByte(address2 + HEADER_OFFSET_FLAGS) & FLAG_TABLE_OPTIONS);
+                Assert.assertEquals(size2 - HEADER_SIZE, Unsafe.getUnsafe().getInt(address2 + 8));
+
+                encoder.beginMessage(1, dict, -1, -1);
+                encoder.addTable(third);
+                int size3 = encoder.finishMessage();
+                long address3 = encoder.getBuffer().getBufferPtr();
+                Assert.assertEquals(
+                        FLAG_TABLE_OPTIONS,
+                        (byte) (Unsafe.getUnsafe().getByte(address3 + HEADER_OFFSET_FLAGS) & FLAG_TABLE_OPTIONS)
+                );
+                Assert.assertEquals(size3 - HEADER_SIZE, Unsafe.getUnsafe().getInt(address3 + 8));
+                byte[] expectedTrailer3 = {
+                        10,
+                        TABLE_OPTION_TAG_DESIGNATED_TIMESTAMP_NAME,
+                        8, 'e', 'v', 'e', 'n', 't', '_', 't', 's',
+                        11, 0, 0, 0
+                };
+                long trailerAddress3 = address3 + size3 - expectedTrailer3.length;
+                for (int i = 0; i < expectedTrailer3.length; i++) {
+                    Assert.assertEquals(
+                            "message 3 trailer byte " + i,
+                            expectedTrailer3[i],
+                            Unsafe.getUnsafe().getByte(trailerAddress3 + i)
+                    );
+                }
+            }
+        });
+    }
+
     @Test
     public void testVersionByteInHeader() throws Exception {
         assertMemoryLeak(() -> {