Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 45 additions & 0 deletions core/src/main/java/io/questdb/client/Sender.java
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*
Expand Down Expand Up @@ -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 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.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
*
Expand Down Expand Up @@ -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];
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -1007,6 +1024,7 @@ private int encodeCommittedPrefixPayloadForUdp(QwpTableBuffer tableBuffer) {
false,
false
);
appendTableOptionsForUdp(tableBuffer);
payloadWriter.finish();
return payloadWriter.getPosition();
}
Expand All @@ -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();
}
Expand Down Expand Up @@ -1106,6 +1125,9 @@ private long estimateBaseForCurrentSchema() {
estimate += 1;
}
}
estimate += QwpColumnWriter.getSingleTableOptionsTrailerSize(
currentTableBuffer.getDesignatedTimestampName()
);
if (currentTableHeadroomState != null) {
currentTableHeadroomState.cacheBaseEstimate(currentTableBuffer.getColumnCount(), estimate);
}
Expand Down Expand Up @@ -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);

Expand All @@ -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();
}

Expand Down Expand Up @@ -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;
Expand All @@ -1509,4 +1536,32 @@ 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()"
);
}
tableBuffer.setDesignatedTimestampName(columnName);
// The cached base estimate folds in the trailer size, which depends on
// the name length, so every accepted set/change must invalidate it. Any
// accepted set requires rowCount == 0, where the committed estimate is
// already 0, so no committedDatagramEstimate adjustment is needed.
if (currentTableHeadroomState != null) {
currentTableHeadroomState.invalidateBaseEstimate();
}
return this;
}

private TableOptions of(QwpTableBuffer tableBuffer) {
this.tableBuffer = tableBuffer;
return this;
}
}
}
Loading
Loading