Name designated timestamps in QWP auto-create

QWP senders let callers name the designated timestamp column that the
server creates when it auto-creates a missing table. The sender API
exposes the hint through a Sender.TableOptions flyweight obtained via
sender.table(name).tableOptions().designatedTimestamp(col), keeping
Sender itself free of flat per-option methods so future create-time
hints (partitioning, storage policies) can land beside it.

The wire encoding sets header flag 0x20 and appends a per-table TLV
options trailer (tag 0x01 = designated timestamp name) plus a u32
length footer after the table blocks. Old servers ignore the unknown
flag and the trailing bytes, so frames degrade to the conventional
timestamp column name; when no hint is set the emitted bytes stay
byte-identical to the legacy encoding, which a golden-bytes test pins.
Both the WebSocket and UDP transports carry the trailer, and
store-and-forward replays the self-describing frames safely against
older servers.

The WebSocket client reads the X-QWP-Table-Options capability header
during the upgrade and logs one warning when a configured hint reaches
a server that does not advertise support. ILP senders reject the API.

The hint is sticky per table, validated to 127 UTF-8 bytes, and
re-declarable while a table buffer holds no rows, so pooled sender
borrowers can re-declare it instead of hitting an unclearable
conflict. The pooled sender guards stale handles behind its lease
generation. Tests cover trailer layouts and mixed presence, UDP
framing, upgrade-header parsing, stale and retained flyweight
references, options-only rows, and lease rejection.

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

Author: jerrinot

core/src/main/java/io/questdb/client/Sender.java

@@ -742,6 +742,25 @@ default Sender shortColumn(CharSequence name, short value) {
      */
     Sender table(CharSequence table);
 
+    /**
+     * Returns create-time options for the currently selected table.
+     * <p>
+     * Call this after {@link #table(CharSequence)}. QWP senders retain options
+     * per table for the sender's lifetime. ILP transports cannot express table
+     * options.
+     * <p>
+     * The returned instance is a single per-sender object that this method and
+     * {@link #table(CharSequence)} re-target at the currently selected table.
+     * Do not retain the reference across table selections; use it immediately
+     * and obtain a fresh one after selecting a table.
+     *
+     * @return options for the currently selected table
+     * @throws LineSenderException if the transport is ILP or no table is selected
+     */
+    default TableOptions tableOptions() {
+        throw new LineSenderException("table options are not supported by ILP");
+    }
+
     /**
      * Add a column with a non-designated timestamp value.
      *
@@ -837,6 +856,32 @@ enum SfDurability {
         PERIODIC
     }
 
+    /**
+     * Create-time options for the table selected by {@link Sender#table(CharSequence)}.
+     */
+    interface TableOptions {
+
+        /**
+         * Sets the designated timestamp column name to use if the selected
+         * table is auto-created by QWP ingestion.
+         * <p>
+         * Supporting servers use this hint only while creating a missing table
+         * and silently ignore it for existing tables. Older QWP servers also
+         * silently ignore it and create the conventional {@code timestamp}
+         * column instead.
+         * <p>
+         * The hint is sticky per table. It may be re-declared with a different
+         * name while the table has no buffered rows; changing it while rows
+         * are buffered throws.
+         *
+         * @param columnName non-empty column name of at most 127 UTF-8 bytes
+         * @return this instance for method chaining
+         * @throws LineSenderException if the name is invalid or these options
+         *                             no longer refer to the selected table
+         */
+        TableOptions designatedTimestamp(CharSequence columnName);
+    }
+
     /**
      * Configure TLS mode.
      * Most users should not need to use anything but the default mode.

core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java

@@ -83,6 +83,7 @@ public abstract class WebSocketClient implements QuietCloseable {
     private static final String QWP_DURABLE_ACK_ENABLED_VALUE = "enabled";
     private static final String QWP_DURABLE_ACK_HEADER_NAME = "X-QWP-Durable-Ack:";
     private static final String QWP_MAX_BATCH_SIZE_HEADER_NAME = "X-QWP-Max-Batch-Size:";
+    private static final String QWP_TABLE_OPTIONS_HEADER_NAME = "X-QWP-Table-Options:";
     private static final String QWP_VERSION_HEADER_NAME = "X-QWP-Version:";
     private static final ThreadLocal<MessageDigest> SHA1_DIGEST = ThreadLocal.withInitial(() -> {
         try {
@@ -162,6 +163,7 @@ public abstract class WebSocketClient implements QuietCloseable {
     // the wire said without re-clamping so a misconfigured server is observable.
     private int serverNegotiatedZstdLevel;
     private int serverQwpVersion = 1;
+    private boolean serverTableOptionsSupported;
     private String upgradeRejectRole;
     // Server-advertised zone identifier from the most recent rejected upgrade,
     // captured from the X-QuestDB-Zone response header on a 421. Null when the
@@ -410,6 +412,14 @@ public boolean isServerDurableAckEnabled() {
         return serverDurableAckEnabled;
     }
 
+    /**
+     * Returns true when the server advertised support for the designated
+     * timestamp name table option in {@code X-QWP-Table-Options}.
+     */
+    public boolean isServerTableOptionsSupported() {
+        return serverTableOptionsSupported;
+    }
+
     /**
      * Receives and processes WebSocket frames.
      *
@@ -614,6 +624,7 @@ public void upgrade(CharSequence path, int timeout, CharSequence authorizationHe
         upgradeRejectRole = null;
         upgradeRejectZone = null;
         upgradeStatusCode = 0;
+        serverTableOptionsSupported = false;
 
         // Generate random key
         byte[] keyBytes = new byte[16];
@@ -822,6 +833,35 @@ private static int extractQwpVersion(String response) {
         return 1;
     }
 
+    private static boolean extractTableOptionsSupport(String response) {
+        int headerLen = QWP_TABLE_OPTIONS_HEADER_NAME.length();
+        int responseLen = response.length();
+        for (int i = 0; i <= responseLen - headerLen; i++) {
+            if (response.regionMatches(true, i, QWP_TABLE_OPTIONS_HEADER_NAME, 0, headerLen)) {
+                int valueStart = i + headerLen;
+                int lineEnd = response.indexOf('\r', valueStart);
+                if (lineEnd < 0) {
+                    lineEnd = responseLen;
+                }
+                String value = response.substring(valueStart, lineEnd);
+                int tokenStart = 0;
+                while (tokenStart <= value.length()) {
+                    int comma = value.indexOf(',', tokenStart);
+                    int tokenEnd = comma < 0 ? value.length() : comma;
+                    if ("1".equals(value.substring(tokenStart, tokenEnd).trim())) {
+                        return true;
+                    }
+                    if (comma < 0) {
+                        break;
+                    }
+                    tokenStart = comma + 1;
+                }
+                return false;
+            }
+        }
+        return false;
+    }
+
     private static String extractRoleHeader(String response) {
         int headerLen = QUESTDB_ROLE_HEADER_NAME.length();
         int responseLen = response.length();
@@ -1360,6 +1400,10 @@ private void validateUpgradeResponse(int headerEnd) {
         // sender falls back to its locally configured byte budget in that case.
         serverMaxBatchSize = extractMaxBatchSize(response);
 
+        // Extract the supported per-table option tags. Tag 1 is the
+        // designated timestamp column name hint.
+        serverTableOptionsSupported = extractTableOptionsSupport(response);
+
         // Extract X-QWP-Content-Encoding (optional). Surfaces what level the
         // server actually applied -- which may differ from what this client
         // asked for if the server has qwp.egress.compression.force.level set

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

@@ -289,6 +289,20 @@ private void writeTableHeader(String tableName, int rowCount, QwpColumnDef[] col
         }
     }
 
+    private void writeTableOptions(QwpTableBuffer tableBuffer) {
+        String designatedTimestampName = tableBuffer.getDesignatedTimestampName();
+        if (designatedTimestampName == null) {
+            buffer.putVarint(0);
+            return;
+        }
+
+        int nameLength = NativeBufferWriter.utf8Length(designatedTimestampName);
+        int blockLength = 1 + NativeBufferWriter.varintSize(nameLength) + nameLength;
+        buffer.putVarint(blockLength);
+        buffer.putByte(TABLE_OPTION_TAG_DESIGNATED_TIMESTAMP_NAME);
+        buffer.putString(designatedTimestampName);
+    }
+
     private void writeTimestampColumn(long addr, int count, boolean useGorilla) {
         if (useGorilla && count > 2) {
             // Single pass: check feasibility and compute encoded size together
@@ -348,6 +362,21 @@ void encodeTable(
         }
     }
 
+    int encodeTableOptions(QwpTableBuffer tableBuffer) {
+        int start = buffer.getPosition();
+        writeTableOptions(tableBuffer);
+        return buffer.getPosition() - start;
+    }
+
+    static int getSingleTableOptionsTrailerSize(CharSequence designatedTimestampName) {
+        if (designatedTimestampName == null) {
+            return 0;
+        }
+        int nameLength = NativeBufferWriter.utf8Length(designatedTimestampName);
+        int blockLength = 1 + NativeBufferWriter.varintSize(nameLength) + nameLength;
+        return NativeBufferWriter.varintSize(blockLength) + blockLength + Integer.BYTES;
+    }
+
     void setBuffer(QwpBufferWriter buffer) {
         this.buffer = buffer;
     }

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

@@ -81,6 +81,7 @@ public class QwpUdpSender implements Sender {
     private final SegmentedNativeBufferWriter payloadWriter;
     private final CharSequenceObjHashMap<QwpTableBuffer> tableBuffers;
     private final CharSequenceObjHashMap<TableHeadroomState> tableHeadroomStates;
+    private final TableOptionsImpl tableOptions = new TableOptionsImpl();
     private final boolean trackDatagramEstimate;
     private QwpTableBuffer.ColumnBuffer cachedTimestampColumn;
     private QwpTableBuffer.ColumnBuffer cachedTimestampNanosColumn;
@@ -591,6 +592,13 @@ public Sender table(CharSequence tableName) {
         return this;
     }
 
+    @Override
+    public TableOptions tableOptions() {
+        checkNotClosed();
+        checkTableSelected();
+        return tableOptions.of(currentTableBuffer);
+    }
+
     @Override
     public Sender timestampColumn(CharSequence columnName, long value, ChronoUnit unit) {
         checkNotClosed();
@@ -866,6 +874,15 @@ private void appendLongArrayValue(QwpTableBuffer.ColumnBuffer column, Object val
         throw new LineSenderException("unsupported long array type");
     }
 
+    private void appendTableOptionsForUdp(QwpTableBuffer tableBuffer) {
+        if (tableBuffer.getDesignatedTimestampName() == null) {
+            return;
+        }
+        int trailerStart = payloadWriter.getPosition();
+        columnWriter.encodeTableOptions(tableBuffer);
+        payloadWriter.putInt(payloadWriter.getPosition() - trailerStart);
+    }
+
     private void atMicros(long timestampMicros) {
         try {
             stageDesignatedTimestampValue(timestampMicros, false);
@@ -1007,6 +1024,7 @@ private int encodeCommittedPrefixPayloadForUdp(QwpTableBuffer tableBuffer) {
                 false,
                 false
         );
+        appendTableOptionsForUdp(tableBuffer);
         payloadWriter.finish();
         return payloadWriter.getPosition();
     }
@@ -1015,6 +1033,7 @@ private int encodeTablePayloadForUdp(QwpTableBuffer tableBuffer) {
         payloadWriter.reset();
         columnWriter.setBuffer(payloadWriter);
         columnWriter.encodeTable(tableBuffer, false, false);
+        appendTableOptionsForUdp(tableBuffer);
         payloadWriter.finish();
         return payloadWriter.getPosition();
     }
@@ -1106,6 +1125,9 @@ private long estimateBaseForCurrentSchema() {
                 estimate += 1;
             }
         }
+        estimate += QwpColumnWriter.getSingleTableOptionsTrailerSize(
+                currentTableBuffer.getDesignatedTimestampName()
+        );
         if (currentTableHeadroomState != null) {
             currentTableHeadroomState.cacheBaseEstimate(currentTableBuffer.getColumnCount(), estimate);
         }
@@ -1218,17 +1240,17 @@ private void rollbackCurrentRowToCommittedState() {
 
     private void sendCommittedPrefix(CharSequence tableName, QwpTableBuffer tableBuffer) {
         int payloadLength = encodeCommittedPrefixPayloadForUdp(tableBuffer);
-        sendEncodedPayload(tableName, payloadLength);
+        sendEncodedPayload(tableName, tableBuffer, payloadLength);
     }
 
-    private void sendEncodedPayload(CharSequence tableName, int payloadLength) {
+    private void sendEncodedPayload(CharSequence tableName, QwpTableBuffer tableBuffer, int payloadLength) {
         headerBuffer.reset();
         headerBuffer.putByte((byte) 'Q');
         headerBuffer.putByte((byte) 'W');
         headerBuffer.putByte((byte) 'P');
         headerBuffer.putByte((byte) '1');
         headerBuffer.putByte(VERSION);
-        headerBuffer.putByte((byte) 0);
+        headerBuffer.putByte(tableBuffer.getDesignatedTimestampName() == null ? 0 : FLAG_TABLE_OPTIONS);
         headerBuffer.putShort((short) 1);
         headerBuffer.putInt(payloadLength);
 
@@ -1251,7 +1273,7 @@ private void sendEncodedPayload(CharSequence tableName, int payloadLength) {
 
     private void sendWholeTableBuffer(CharSequence tableName, QwpTableBuffer tableBuffer) {
         int payloadLength = encodeTablePayloadForUdp(tableBuffer);
-        sendEncodedPayload(tableName, payloadLength);
+        sendEncodedPayload(tableName, tableBuffer, payloadLength);
         tableBuffer.reset();
     }
 
@@ -1483,6 +1505,11 @@ long getCachedBaseEstimate(int currentSchemaColumnCount) {
             return cachedBaseEstimateColumnCount == currentSchemaColumnCount ? cachedBaseEstimate : -1;
         }
 
+        void invalidateBaseEstimate() {
+            cachedBaseEstimate = -1;
+            cachedBaseEstimateColumnCount = -1;
+        }
+
         long predictNextRowGrowth() {
             if (committedSampleCount < 2) {
                 return 0;
@@ -1509,4 +1536,36 @@ void recordCommittedRow(int currentSchemaColumnCount, long rowDatagramGrowth) {
             }
         }
     }
+
+    private final class TableOptionsImpl implements TableOptions {
+        private QwpTableBuffer tableBuffer;
+
+        @Override
+        public TableOptions designatedTimestamp(CharSequence columnName) {
+            checkNotClosed();
+            if (tableBuffer != currentTableBuffer) {
+                throw new LineSenderException(
+                        "table options reference is stale; call tableOptions() after table()"
+                );
+            }
+            String previousName = tableBuffer.getDesignatedTimestampName();
+            tableBuffer.setDesignatedTimestampName(columnName);
+            if (previousName == null) {
+                if (currentTableHeadroomState != null) {
+                    currentTableHeadroomState.invalidateBaseEstimate();
+                }
+                if (trackDatagramEstimate && committedDatagramEstimate > 0) {
+                    committedDatagramEstimate += QwpColumnWriter.getSingleTableOptionsTrailerSize(
+                            tableBuffer.getDesignatedTimestampName()
+                    );
+                }
+            }
+            return this;
+        }
+
+        private TableOptions of(QwpTableBuffer tableBuffer) {
+            this.tableBuffer = tableBuffer;
+            return this;
+        }
+    }
 }