+ * Call this after {@link #table(CharSequence)}. QWP senders retain options + * per table for the sender's lifetime. ILP transports cannot express table + * options. + *
+ * 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. + *
+ * 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. + *
+ * The hint is sticky per table. It may be set or re-declared only
+ * while the table has no buffered rows; setting 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.
diff --git a/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java b/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java
index 94018fff..0a2aca3d 100644
--- a/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java
+++ b/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
+ * 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;
/**
* Flag bit: payload region after the prelude is zstd-compressed. Set only
* when the handshake negotiated zstd compression. Mirror of the server-side
@@ -98,6 +110,10 @@ public final class QwpConstants {
* the ingress {@code STATUS_*} namespace (0x00-0x09).
*/
public static final byte STATUS_LIMIT_EXCEEDED = 0x0B;
+ /**
+ * Per-table option tag: designated timestamp column name.
+ */
+ public static final byte TABLE_OPTION_TAG_DESIGNATED_TIMESTAMP_NAME = 0x01;
/**
* Column type: BINARY (length-prefixed opaque bytes).
* Wire format: identical to VARCHAR — (N+1) x uint32 offsets + concatenated bytes.
diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java b/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java
index 0947e5e1..d1d81b2f 100644
--- a/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java
+++ b/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java
@@ -69,6 +69,7 @@ public class QwpTableBuffer implements QuietCloseable {
private int columnAccessCursor; // tracks expected next column index
private boolean columnDefsCacheValid;
private int committedColumnCount; // columns that existed at last nextRow()
+ private String designatedTimestampName;
private ColumnBuffer[] fastColumns; // plain array for O(1) sequential access
private int inProgressColumnCount;
private int rowCount;
@@ -125,6 +126,7 @@ public void clear() {
}
columns.clear();
columnNameToIndex.clear();
+ designatedTimestampName = null;
fastColumns = null;
columnAccessCursor = 0;
committedColumnCount = 0;
@@ -180,6 +182,14 @@ public QwpColumnDef[] getColumnDefs() {
return cachedColumnDefs;
}
+ /**
+ * Returns the create-only designated timestamp column name hint, or
+ * {@code null} when none was set.
+ */
+ public String getDesignatedTimestampName() {
+ return designatedTimestampName;
+ }
+
/**
* Returns an existing column with the given name and type, or {@code null} if absent.
*
@@ -353,6 +363,44 @@ public void rollbackUncommittedColumns() {
rebuildColumnAccessStructures();
}
+ /**
+ * Sets the create-only designated timestamp column name hint for this
+ * table. The value is sticky across rows and buffer resets. It may be
+ * set or re-declared only while the buffer holds no rows; any set while
+ * rows are buffered throws.
+ *
+ * @param columnName non-empty name of at most 127 UTF-8 bytes
+ */
+ public void setDesignatedTimestampName(CharSequence columnName) {
+ if (columnName == null || columnName.length() == 0) {
+ throw new LineSenderException("designated timestamp name cannot be empty");
+ }
+ int utf8Length = Utf8s.utf8Bytes(columnName);
+ if (utf8Length > MAX_COLUMN_NAME_LENGTH) {
+ throw new LineSenderException(
+ "designated timestamp name too long [maxLength=" + MAX_COLUMN_NAME_LENGTH
+ + ", utf8Length=" + utf8Length + ']'
+ );
+ }
+ if (designatedTimestampName == null) {
+ if (rowCount != 0) {
+ throw new LineSenderException(
+ "cannot set designated timestamp name for table '" + tableName
+ + "' after rows are buffered [new=" + columnName + ']'
+ );
+ }
+ designatedTimestampName = Chars.toString(columnName);
+ } else if (!Chars.equals(designatedTimestampName, columnName)) {
+ if (rowCount != 0) {
+ throw new LineSenderException(
+ "conflicting designated timestamp names for table '" + tableName
+ + "' [existing=" + designatedTimestampName + ", new=" + columnName + ']'
+ );
+ }
+ designatedTimestampName = Chars.toString(columnName);
+ }
+ }
+
private static void assertColumnType(CharSequence name, byte type, ColumnBuffer column) {
if (column.type != type) {
throw new LineSenderException(
diff --git a/core/src/main/java/io/questdb/client/impl/PooledSender.java b/core/src/main/java/io/questdb/client/impl/PooledSender.java
index 7b4e5f80..36531b85 100644
--- a/core/src/main/java/io/questdb/client/impl/PooledSender.java
+++ b/core/src/main/java/io/questdb/client/impl/PooledSender.java
@@ -59,6 +59,7 @@ public final class PooledSender implements Sender {
private final long generation;
private final SenderSlot slot;
+ private final PooledTableOptions tableOptions = new PooledTableOptions();
PooledSender(SenderSlot slot, long generation) {
this.slot = slot;
@@ -359,6 +360,11 @@ public Sender table(CharSequence table) {
return this;
}
+ @Override
+ public TableOptions tableOptions() {
+ return tableOptions.of(slot.live(generation).tableOptions());
+ }
+
@Override
public Sender timestampColumn(CharSequence name, long value, ChronoUnit unit) {
slot.live(generation).timestampColumn(name, value, unit);
@@ -399,4 +405,20 @@ public boolean hasSameSlotForTesting(PooledSender that) {
SenderSlot slot() {
return slot;
}
+
+ private final class PooledTableOptions implements TableOptions {
+ private TableOptions delegate;
+
+ @Override
+ public TableOptions designatedTimestamp(CharSequence columnName) {
+ slot.live(generation);
+ delegate.designatedTimestamp(columnName);
+ return this;
+ }
+
+ private TableOptions of(TableOptions delegate) {
+ this.delegate = delegate;
+ return this;
+ }
+ }
}
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java b/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java
index a44c57f2..5fd877f6 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java
@@ -128,6 +128,30 @@ public void testExtractMaxBatchSizeParsesPositive() throws Exception {
Assert.assertEquals(16 * 1024 * 1024, invokeExtractMaxBatchSize(response));
}
+ @Test
+ public void testExtractTableOptionsSupportAbsentHeaderReturnsFalse() throws Exception {
+ String response = "HTTP/1.1 101 Switching Protocols\r\n"
+ + "X-QWP-Version: 1\r\n"
+ + "\r\n";
+ Assert.assertFalse(invokeExtractTableOptionsSupport(response));
+ }
+
+ @Test
+ public void testExtractTableOptionsSupportFindsDesignatedTimestampTag() throws Exception {
+ String response = "HTTP/1.1 101 Switching Protocols\r\n"
+ + "X-QWP-Table-Options: 2, 1\r\n"
+ + "\r\n";
+ Assert.assertTrue(invokeExtractTableOptionsSupport(response));
+ }
+
+ @Test
+ public void testExtractTableOptionsSupportRejectsOtherTags() throws Exception {
+ String response = "HTTP/1.1 101 Switching Protocols\r\n"
+ + "X-QWP-Table-Options: 2\r\n"
+ + "\r\n";
+ Assert.assertFalse(invokeExtractTableOptionsSupport(response));
+ }
+
/**
* A frame handler may close() the client from inside its callback:
* CursorWebSocketSendLoop's NACK-recycle path (handleServerRejection /
@@ -393,6 +417,12 @@ private static int invokeExtractMaxBatchSize(String response) throws Exception {
return (int) m.invoke(null, response);
}
+ private static boolean invokeExtractTableOptionsSupport(String response) throws Exception {
+ Method m = WebSocketClient.class.getDeclaredMethod("extractTableOptionsSupport", String.class);
+ m.setAccessible(true);
+ return (boolean) m.invoke(null, response);
+ }
+
private static void setUpgradedTrue(Object obj) throws Exception {
Class clazz = obj.getClass();
while (clazz != null) {
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/line/LineHttpSenderInterfaceTest.java b/core/src/test/java/io/questdb/client/test/cutlass/line/LineHttpSenderInterfaceTest.java
index 0d8ae3d5..f840d5ff 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/line/LineHttpSenderInterfaceTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/line/LineHttpSenderInterfaceTest.java
@@ -25,6 +25,7 @@
package io.questdb.client.test.cutlass.line;
import io.questdb.client.Sender;
+import io.questdb.client.cutlass.line.LineSenderException;
import io.questdb.client.std.bytes.DirectByteSlice;
import org.junit.Assert;
import org.junit.Test;
@@ -72,4 +73,16 @@ public void testResetClearsBufferAndAllowsNewRows() {
sender.bufferView().size() > 0);
}
}
+
+ @Test
+ public void testTableOptionsThrows() {
+ try (Sender sender = Sender.fromConfig("http::addr=127.0.0.1:1;auto_flush=off;protocol_version=1;")) {
+ try {
+ sender.table("t").tableOptions();
+ Assert.fail("Expected LineSenderException");
+ } catch (LineSenderException e) {
+ Assert.assertTrue(e.getMessage().contains("table options are not supported by ILP"));
+ }
+ }
+ }
}
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpUdpSenderTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpUdpSenderTest.java
index f16bc267..6bd712fc 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpUdpSenderTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpUdpSenderTest.java
@@ -24,6 +24,7 @@
package io.questdb.client.test.cutlass.qwp.client;
+import io.questdb.client.Sender;
import io.questdb.client.cutlass.line.LineSenderException;
import io.questdb.client.cutlass.line.array.DoubleArray;
import io.questdb.client.cutlass.line.array.LongArray;
@@ -1677,6 +1678,171 @@ public void testSymbolPrefixFlushKeepsSingleRetainedDictionaryEntry() throws Exc
});
}
+ @Test
+ public void testTableOptionsBeforeTableThrows() throws Exception {
+ assertMemoryLeak(() -> {
+ CapturingNetworkFacade nf = new CapturingNetworkFacade();
+ try (QwpUdpSender sender = new QwpUdpSender(nf, 0, 0, 9000, 1)) {
+ assertThrowsContains("table()", sender::tableOptions);
+ }
+ });
+ }
+
+ @Test
+ public void testTableOptionsChainingReturnsSameInstance() throws Exception {
+ assertMemoryLeak(() -> {
+ CapturingNetworkFacade nf = new CapturingNetworkFacade();
+ try (QwpUdpSender sender = new QwpUdpSender(nf, 0, 0, 9000, 1)) {
+ Sender.TableOptions options = sender.table("t").tableOptions();
+
+ Assert.assertSame(options, options.designatedTimestamp("ts"));
+ Assert.assertSame(options, sender.tableOptions());
+ }
+ });
+ }
+
+ @Test
+ public void testTableOptionsDesignatedTimestampEmitsTrailer() throws Exception {
+ assertMemoryLeak(() -> {
+ CapturingNetworkFacade nf = new CapturingNetworkFacade();
+ try (QwpUdpSender sender = new QwpUdpSender(nf, 0, 0, 9000, 1)) {
+ sender.table("t").tableOptions().designatedTimestamp("ts");
+ sender.longColumn("x", 42).atNow();
+ sender.flush();
+ }
+
+ Assert.assertEquals(1, nf.packets.size());
+ byte[] packet = nf.packets.get(0);
+ Assert.assertEquals(FLAG_TABLE_OPTIONS, (byte) (packet[HEADER_OFFSET_FLAGS] & FLAG_TABLE_OPTIONS));
+ Assert.assertEquals(packet.length - HEADER_SIZE, Unsafe.byteArrayGetInt(packet, 8));
+
+ byte[] expectedTrailer = {
+ 4,
+ TABLE_OPTION_TAG_DESIGNATED_TIMESTAMP_NAME,
+ 2, 't', 's',
+ 5, 0, 0, 0
+ };
+ int trailerOffset = packet.length - expectedTrailer.length;
+ for (int i = 0; i < expectedTrailer.length; i++) {
+ Assert.assertEquals("trailer byte " + i, expectedTrailer[i], packet[trailerOffset + i]);
+ }
+ });
+ }
+
+ @Test
+ public void testTableOptionsDesignatedTimestampRejectedAfterRowsBuffered() throws Exception {
+ assertMemoryLeak(() -> {
+ CapturingNetworkFacade nf = new CapturingNetworkFacade();
+ int maxDatagramSize = 1024;
+ try (QwpUdpSender sender = new QwpUdpSender(nf, 0, 0, 9000, 1, maxDatagramSize)) {
+ sender.table("t");
+ sender.longColumn("x", 1).atNow();
+
+ // a late hint would grow the committed datagram past the cap
+ // with no re-check on the flush path, so it must be rejected
+ assertThrowsContains(
+ "after rows are buffered",
+ () -> sender.tableOptions().designatedTimestamp("event_ts")
+ );
+
+ sender.flush();
+ Assert.assertEquals(1, nf.packets.size());
+ Assert.assertEquals(0, (byte) (nf.packets.get(0)[HEADER_OFFSET_FLAGS] & FLAG_TABLE_OPTIONS));
+ assertPacketsWithinLimit(new RunResult(nf.packets, nf.lengths, nf.sendCount), maxDatagramSize);
+
+ // rowCount == 0 after the flush: the hint is accepted again
+ sender.tableOptions().designatedTimestamp("event_ts");
+ }
+ });
+ }
+
+ @Test
+ public void testTableOptionsOnlyRowIsDiscarded() throws Exception {
+ assertMemoryLeak(() -> {
+ CapturingNetworkFacade nf = new CapturingNetworkFacade();
+ try (QwpUdpSender sender = new QwpUdpSender(nf, 0, 0, 9000, 1)) {
+ sender.table("first").tableOptions().designatedTimestamp("ts");
+ sender.table("second");
+ sender.flush();
+ Assert.assertEquals(0, nf.sendCount);
+
+ sender.table("third").tableOptions().designatedTimestamp("event_ts");
+ }
+ Assert.assertEquals(0, nf.sendCount);
+ });
+ }
+
+ @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(() -> {
+ CapturingNetworkFacade nf = new CapturingNetworkFacade();
+ try (QwpUdpSender sender = new QwpUdpSender(nf, 0, 0, 9000, 1)) {
+ Sender.TableOptions retained = sender.table("a").tableOptions();
+ Assert.assertSame(retained, sender.table("b").tableOptions());
+
+ // the retained reference is the re-stamped per-sender flyweight:
+ // the call lands on the currently selected table, not on "a"
+ retained.designatedTimestamp("ts");
+ Assert.assertEquals("ts", sender.currentTableBufferForTest().getDesignatedTimestampName());
+
+ sender.table("a");
+ Assert.assertNull(sender.currentTableBufferForTest().getDesignatedTimestampName());
+ }
+ });
+ }
+
+ @Test
+ public void testTableOptionsStaleReferenceRejectedUntilTableReselected() throws Exception {
+ assertMemoryLeak(() -> {
+ CapturingNetworkFacade nf = new CapturingNetworkFacade();
+ try (QwpUdpSender sender = new QwpUdpSender(nf, 0, 0, 9000, 1)) {
+ Sender.TableOptions options = sender.table("a").tableOptions();
+ options.designatedTimestamp("ts");
+
+ sender.table("b");
+ assertThrowsContains("stale", () -> options.designatedTimestamp("other_ts"));
+
+ sender.table("a");
+ Assert.assertSame(options, options.designatedTimestamp("ts"));
+ Assert.assertEquals("ts", sender.currentTableBufferForTest().getDesignatedTimestampName());
+ }
+ });
+ }
+
@Test
public void testTableRejectsEmptyName() throws Exception {
assertMemoryLeak(() -> {
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketEncoderTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketEncoderTest.java
index 4d7d7931..59b80e66 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketEncoderTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketEncoderTest.java
@@ -1258,6 +1258,39 @@ public void testPayloadLengthPatched() throws Exception {
});
}
+ @Test
+ public void testPreTableOptionsEncodingIsByteIdenticalWhenNameIsAbsent() throws Exception {
+ assertMemoryLeak(() -> {
+ try (QwpWebSocketEncoder encoder = new QwpWebSocketEncoder();
+ QwpTableBuffer buffer = new QwpTableBuffer("t")) {
+ QwpTableBuffer.ColumnBuffer col = buffer.getOrCreateColumn("x", TYPE_LONG, false);
+ col.addLong(42);
+ buffer.nextRow();
+
+ int size = encoder.encode(buffer);
+ byte[] expected = {
+ 'Q', 'W', 'P', '1',
+ VERSION,
+ FLAG_GORILLA,
+ 1, 0,
+ 16, 0, 0, 0,
+ 1, 't',
+ 1,
+ 1,
+ 1, 'x', TYPE_LONG,
+ 0,
+ 42, 0, 0, 0, 0, 0, 0, 0
+ };
+
+ Assert.assertEquals(expected.length, size);
+ long address = encoder.getBuffer().getBufferPtr();
+ for (int i = 0; i < expected.length; i++) {
+ Assert.assertEquals("byte " + i, expected[i], Unsafe.getUnsafe().getByte(address + i));
+ }
+ }
+ });
+ }
+
@Test
public void testReset() throws Exception {
assertMemoryLeak(() -> {
@@ -1284,6 +1317,171 @@ public void testReset() throws Exception {
});
}
+ @Test
+ public void testTableOptionsTrailerEncoding() throws Exception {
+ assertMemoryLeak(() -> {
+ try (QwpWebSocketEncoder encoder = new QwpWebSocketEncoder();
+ QwpTableBuffer buffer = new QwpTableBuffer("t")) {
+ buffer.setDesignatedTimestampName("ts");
+ QwpTableBuffer.ColumnBuffer col = buffer.getOrCreateColumn("x", TYPE_LONG, false);
+ col.addLong(42);
+ buffer.nextRow();
+
+ int size = encoder.encode(buffer);
+ long address = encoder.getBuffer().getBufferPtr();
+ Assert.assertEquals(
+ FLAG_TABLE_OPTIONS,
+ (byte) (Unsafe.getUnsafe().getByte(address + HEADER_OFFSET_FLAGS) & FLAG_TABLE_OPTIONS)
+ );
+ Assert.assertEquals(size - HEADER_SIZE, Unsafe.getUnsafe().getInt(address + 8));
+
+ byte[] expectedTrailer = {
+ 4,
+ TABLE_OPTION_TAG_DESIGNATED_TIMESTAMP_NAME,
+ 2, 't', 's',
+ 5, 0, 0, 0
+ };
+ long trailerAddress = address + size - expectedTrailer.length;
+ for (int i = 0; i < expectedTrailer.length; i++) {
+ Assert.assertEquals(
+ "trailer byte " + i,
+ expectedTrailer[i],
+ Unsafe.getUnsafe().getByte(trailerAddress + i)
+ );
+ }
+ }
+ });
+ }
+
+ @Test
+ public void testTableOptionsTrailerEncodingWithMixedPresence() throws Exception {
+ assertMemoryLeak(() -> {
+ try (QwpWebSocketEncoder encoder = new QwpWebSocketEncoder();
+ QwpTableBuffer first = new QwpTableBuffer("first");
+ QwpTableBuffer second = new QwpTableBuffer("second");
+ QwpTableBuffer third = new QwpTableBuffer("third")) {
+ first.getOrCreateColumn("x", TYPE_LONG, false).addLong(1);
+ first.nextRow();
+ second.setDesignatedTimestampName("ts");
+ second.getOrCreateColumn("x", TYPE_LONG, false).addLong(2);
+ second.nextRow();
+ third.getOrCreateColumn("x", TYPE_LONG, false).addLong(3);
+ third.nextRow();
+
+ encoder.beginMessage(3, new GlobalSymbolDictionary(), -1, -1);
+ encoder.addTable(first);
+ encoder.addTable(second);
+ encoder.addTable(third);
+ int size = encoder.finishMessage();
+
+ long address = encoder.getBuffer().getBufferPtr();
+ Assert.assertEquals(
+ FLAG_TABLE_OPTIONS,
+ (byte) (Unsafe.getUnsafe().getByte(address + HEADER_OFFSET_FLAGS) & FLAG_TABLE_OPTIONS)
+ );
+ Assert.assertEquals(3, Unsafe.getUnsafe().getShort(address + 6));
+ Assert.assertEquals(size - HEADER_SIZE, Unsafe.getUnsafe().getInt(address + 8));
+
+ byte[] expectedTrailer = {
+ 0,
+ 4,
+ TABLE_OPTION_TAG_DESIGNATED_TIMESTAMP_NAME,
+ 2, 't', 's',
+ 0,
+ 7, 0, 0, 0
+ };
+ long trailerAddress = address + size - expectedTrailer.length;
+ for (int i = 0; i < expectedTrailer.length; i++) {
+ Assert.assertEquals(
+ "trailer byte " + i,
+ expectedTrailer[i],
+ Unsafe.getUnsafe().getByte(trailerAddress + i)
+ );
+ }
+ }
+ });
+ }
+
+ @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(() -> {
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java
index f5805bb9..02666859 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java
@@ -24,6 +24,7 @@
package io.questdb.client.test.cutlass.qwp.client;
+import io.questdb.client.Sender;
import io.questdb.client.cutlass.line.LineSenderException;
import io.questdb.client.cutlass.line.array.DoubleArray;
import io.questdb.client.cutlass.line.array.LongArray;
@@ -554,6 +555,19 @@ public void testOperationsAfterCloseThrow() throws Exception {
});
}
+ @Test
+ public void testOptionsOnlyRowIsDiscardedOnTableSwitch() throws Exception {
+ assertMemoryLeak(() -> {
+ try (QwpWebSocketSender sender = createUnconnectedSender()) {
+ sender.table("first").tableOptions().designatedTimestamp("ts");
+ sender.table("second");
+
+ Assert.assertEquals(0, sender.getTableBuffer("first").getRowCount());
+ Assert.assertEquals(0, sender.getTableBuffer("second").getRowCount());
+ }
+ });
+ }
+
@Test
public void testPendingBytesMatchesGroundTruthAcrossTableSwitches() throws Exception {
assertMemoryLeak(() -> {
@@ -691,6 +705,70 @@ public void testTableBeforeColumnsRequired() throws Exception {
});
}
+ @Test
+ public void testTableOptionsBeforeTableThrows() throws Exception {
+ assertMemoryLeak(() -> {
+ try (QwpWebSocketSender sender = createUnconnectedSender()) {
+ try {
+ sender.tableOptions();
+ Assert.fail("Expected LineSenderException");
+ } catch (LineSenderException e) {
+ Assert.assertTrue(e.getMessage().contains("table()"));
+ }
+ }
+ });
+ }
+
+ @Test
+ public void testTableOptionsChainingReturnsSameInstance() throws Exception {
+ assertMemoryLeak(() -> {
+ try (QwpWebSocketSender sender = createUnconnectedSender()) {
+ Sender.TableOptions options = sender.table("t").tableOptions();
+
+ Assert.assertSame(options, options.designatedTimestamp("ts"));
+ Assert.assertSame(options, sender.tableOptions());
+ }
+ });
+ }
+
+ @Test
+ public void testTableOptionsRetainedReferenceRetargetsOnNextTableOptionsCall() throws Exception {
+ assertMemoryLeak(() -> {
+ try (QwpWebSocketSender sender = createUnconnectedSender()) {
+ Sender.TableOptions retained = sender.table("a").tableOptions();
+ Assert.assertSame(retained, sender.table("b").tableOptions());
+
+ // the retained reference is the re-stamped per-sender flyweight:
+ // the call lands on the currently selected table, not on "a"
+ retained.designatedTimestamp("ts");
+ Assert.assertEquals("ts", sender.getTableBuffer("b").getDesignatedTimestampName());
+ Assert.assertNull(sender.getTableBuffer("a").getDesignatedTimestampName());
+ }
+ });
+ }
+
+ @Test
+ public void testTableOptionsStaleReferenceRejectedUntilTableReselected() throws Exception {
+ assertMemoryLeak(() -> {
+ try (QwpWebSocketSender sender = createUnconnectedSender()) {
+ Sender.TableOptions options = sender.table("a").tableOptions();
+ options.designatedTimestamp("ts");
+
+ sender.table("b");
+ try {
+ options.designatedTimestamp("other_ts");
+ Assert.fail("Expected LineSenderException");
+ } catch (LineSenderException e) {
+ Assert.assertTrue(e.getMessage().contains("stale"));
+ }
+
+ sender.table("a");
+ Assert.assertSame(options, options.designatedTimestamp("ts"));
+ Assert.assertEquals("ts", sender.getTableBuffer("a").getDesignatedTimestampName());
+ }
+ });
+ }
+
@Test
public void testTimestampColumnAfterCloseThrows() throws Exception {
assertMemoryLeak(() -> {
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpConstantsTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpConstantsTest.java
index ace6d4f1..bc3a9b45 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpConstantsTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpConstantsTest.java
@@ -37,6 +37,8 @@ public void testFlagBitPositions() {
// Verify flag bits are at correct positions
Assert.assertEquals(0x04, FLAG_GORILLA);
Assert.assertEquals(0x08, FLAG_DELTA_SYMBOL_DICT);
+ Assert.assertEquals(0x20, FLAG_TABLE_OPTIONS);
+ Assert.assertEquals(0x01, TABLE_OPTION_TAG_DESIGNATED_TIMESTAMP_NAME);
}
@Test
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpTableBufferTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpTableBufferTest.java
index d619a208..1f84cfa7 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpTableBufferTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpTableBufferTest.java
@@ -40,6 +40,7 @@
import java.nio.charset.StandardCharsets;
import static io.questdb.client.test.tools.TestUtils.assertMemoryLeak;
+import static io.questdb.client.test.tools.TestUtils.repeat;
import static org.junit.Assert.*;
public class QwpTableBufferTest {
@@ -1370,6 +1371,85 @@ public void testRetainInProgressRowFastClearsUnstagedNullableColumn() throws Exc
});
}
+ @Test
+ public void testDesignatedTimestampNameFirstSetAfterRowsBufferedThrows() throws Exception {
+ assertMemoryLeak(() -> {
+ try (QwpTableBuffer table = new QwpTableBuffer("test")) {
+ table.getOrCreateColumn("x", QwpConstants.TYPE_LONG, false).addLong(1);
+ table.nextRow();
+ try {
+ table.setDesignatedTimestampName("event_time");
+ fail("expected rejection of first set after rows are buffered");
+ } catch (LineSenderException e) {
+ assertTrue(e.getMessage().contains("after rows are buffered"));
+ }
+ assertNull(table.getDesignatedTimestampName());
+
+ table.reset();
+ table.setDesignatedTimestampName("event_time");
+ assertEquals("event_time", table.getDesignatedTimestampName());
+ }
+ });
+ }
+
+ @Test
+ public void testDesignatedTimestampNameIsStickyAndConsistent() throws Exception {
+ assertMemoryLeak(() -> {
+ try (QwpTableBuffer table = new QwpTableBuffer("test")) {
+ table.setDesignatedTimestampName("event_time");
+ table.setDesignatedTimestampName(new StringBuilder("event_time"));
+ table.reset();
+ assertEquals("event_time", table.getDesignatedTimestampName());
+
+ // no rows buffered: re-declaring with a different name is allowed
+ table.setDesignatedTimestampName("other_time");
+ assertEquals("other_time", table.getDesignatedTimestampName());
+
+ table.getOrCreateColumn("x", QwpConstants.TYPE_LONG, false).addLong(1);
+ table.nextRow();
+ try {
+ table.setDesignatedTimestampName("third_time");
+ fail("expected conflicting designated timestamp name");
+ } catch (LineSenderException e) {
+ assertTrue(e.getMessage().contains("conflicting designated timestamp names"));
+ }
+ table.setDesignatedTimestampName("other_time");
+ assertEquals("other_time", table.getDesignatedTimestampName());
+
+ table.reset();
+ table.setDesignatedTimestampName("fourth_time");
+ assertEquals("fourth_time", table.getDesignatedTimestampName());
+ }
+ });
+ }
+
+ @Test
+ public void testDesignatedTimestampNameValidatesUtf8Length() throws Exception {
+ assertMemoryLeak(() -> {
+ try (QwpTableBuffer table = new QwpTableBuffer("test")) {
+ try {
+ table.setDesignatedTimestampName("");
+ fail("expected empty designated timestamp name rejection");
+ } catch (LineSenderException e) {
+ assertTrue(e.getMessage().contains("cannot be empty"));
+ }
+
+ String maxBytes = repeat("\u00e9", 63) + "x";
+ table.setDesignatedTimestampName(maxBytes);
+ assertEquals(maxBytes, table.getDesignatedTimestampName());
+ }
+
+ try (QwpTableBuffer table = new QwpTableBuffer("test")) {
+ try {
+ table.setDesignatedTimestampName(repeat("\u00e9", 64));
+ fail("expected oversized designated timestamp name rejection");
+ } catch (LineSenderException e) {
+ assertTrue(e.getMessage().contains("utf8Length=128"));
+ }
+ }
+ });
+ }
+
private static void addSymbolUtf8(QwpTableBuffer.ColumnBuffer col, String value) {
byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
long ptr = copyToNative(bytes);
diff --git a/core/src/test/java/io/questdb/client/test/impl/SenderLeaseGenerationTest.java b/core/src/test/java/io/questdb/client/test/impl/SenderLeaseGenerationTest.java
index 7b5b627a..2837595e 100644
--- a/core/src/test/java/io/questdb/client/test/impl/SenderLeaseGenerationTest.java
+++ b/core/src/test/java/io/questdb/client/test/impl/SenderLeaseGenerationTest.java
@@ -25,6 +25,7 @@
package io.questdb.client.test.impl;
import io.questdb.client.Sender;
+import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
import io.questdb.client.impl.PooledSender;
import io.questdb.client.impl.SenderPool;
import io.questdb.client.test.tools.TestUtils;
@@ -118,6 +119,43 @@ public void testStaleGiveBackDoesNotEnqueueSlotTwice() throws Exception {
});
}
+ @Test
+ public void testStaleTableOptionsReferenceIsRejected() throws Exception {
+ TestUtils.assertMemoryLeak(() -> {
+ Class slotClass = Class.forName("io.questdb.client.impl.SenderSlot");
+ Constructor slotCtor = slotClass.getDeclaredConstructor(Sender.class, SenderPool.class, int.class);
+ slotCtor.setAccessible(true);
+ Method bump = slotClass.getDeclaredMethod("bumpGeneration");
+ bump.setAccessible(true);
+ Constructor