Add QWP version negotiation to WebSocket upgrade

The client sends X-QWP-Max-Version and X-QWP-Client-Id
headers in the WebSocket upgrade request. After the server
responds with X-QWP-Version, the client reads the selected
version and uses it in the version byte of all subsequent
QWP message headers.

QwpWebSocketEncoder.writeHeader() now uses a configurable
version field instead of the hardcoded VERSION_1 constant.
QwpWebSocketSender.ensureConnected() sets the version on
the encoder after reading it from the upgrade response.

QwpConstants gains MAX_SUPPORTED_VERSION and CLIENT_ID
constants for the negotiation headers.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: mtopolnik

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

@@ -98,6 +98,10 @@ public abstract class WebSocketClient implements QuietCloseable {
     // Connection state
     private CharSequence host;
     private int port;
+    // QWP version negotiation
+    private String qwpClientId;
+    private int qwpMaxVersion = 1;
+    private int serverQwpVersion = 1;
     // Receive buffer (native memory)
     private long recvBufPtr;
     private int recvBufSize;
@@ -231,6 +235,13 @@ public int getPort() {
         return port;
     }
 
+    /**
+     * Returns the QWP version selected by the server during the upgrade handshake.
+     */
+    public int getServerQwpVersion() {
+        return serverQwpVersion;
+    }
+
     /**
      * Gets the send buffer for building WebSocket frames.
      * <p>
@@ -380,6 +391,20 @@ public boolean tryReceiveFrame(WebSocketFrameHandler handler) {
         return result != null && result;
     }
 
+    /**
+     * Sets the QWP client identifier sent in the X-QWP-Client-Id upgrade header.
+     */
+    public void setQwpClientId(String clientId) {
+        this.qwpClientId = clientId;
+    }
+
+    /**
+     * Sets the maximum QWP version this client supports, sent in the X-QWP-Max-Version upgrade header.
+     */
+    public void setQwpMaxVersion(int maxVersion) {
+        this.qwpMaxVersion = maxVersion;
+    }
+
     /**
      * Performs WebSocket upgrade handshake.
      *
@@ -423,6 +448,14 @@ public void upgrade(CharSequence path, int timeout, CharSequence authorizationHe
         sendBuffer.putAscii(handshakeKey);
         sendBuffer.putAscii("\r\n");
         sendBuffer.putAscii("Sec-WebSocket-Version: 13\r\n");
+        sendBuffer.putAscii("X-QWP-Max-Version: ");
+        sendBuffer.putAscii(Integer.toString(qwpMaxVersion));
+        sendBuffer.putAscii("\r\n");
+        if (qwpClientId != null) {
+            sendBuffer.putAscii("X-QWP-Client-Id: ");
+            sendBuffer.putAscii(qwpClientId);
+            sendBuffer.putAscii("\r\n");
+        }
         if (authorizationHeader != null) {
             sendBuffer.putAscii("Authorization: ");
             sendBuffer.putAscii(authorizationHeader);
@@ -889,6 +922,30 @@ private void validateUpgradeResponse(int headerEnd) {
         if (!containsHeaderValue(response, "Sec-WebSocket-Accept:", expectedAccept, false)) {
             throw new HttpClientException("Invalid Sec-WebSocket-Accept header");
         }
+
+        // Extract X-QWP-Version (optional — defaults to 1 if absent)
+        serverQwpVersion = extractIntHeader(response, "X-QWP-Version:", 1);
+    }
+
+    private static int extractIntHeader(String response, String headerName, int defaultValue) {
+        int headerLen = headerName.length();
+        int responseLen = response.length();
+        for (int i = 0; i <= responseLen - headerLen; i++) {
+            if (response.regionMatches(true, i, headerName, 0, headerLen)) {
+                int valueStart = i + headerLen;
+                int lineEnd = response.indexOf('\r', valueStart);
+                if (lineEnd < 0) {
+                    lineEnd = responseLen;
+                }
+                String value = response.substring(valueStart, lineEnd).trim();
+                try {
+                    return Integer.parseInt(value);
+                } catch (NumberFormatException e) {
+                    return defaultValue;
+                }
+            }
+        }
+        return defaultValue;
     }
 
     protected void dieWaiting(int n) {

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

@@ -42,6 +42,7 @@ public class QwpWebSocketEncoder implements QuietCloseable {
     private byte flags;
     private int payloadStart;
     private byte savedFlags;
+    private byte version = VERSION_1;
 
     public QwpWebSocketEncoder() {
         this.buffer = new NativeBufferWriter();
@@ -133,12 +134,16 @@ public void setGorillaEnabled(boolean enabled) {
         }
     }
 
+    public void setVersion(byte version) {
+        this.version = version;
+    }
+
     public void writeHeader(int tableCount, int payloadLength) {
         buffer.putByte((byte) 'Q');
         buffer.putByte((byte) 'W');
         buffer.putByte((byte) 'P');
         buffer.putByte((byte) '1');
-        buffer.putByte(VERSION_1);
+        buffer.putByte(version);
         buffer.putByte(flags);
         buffer.putShort((short) tableCount);
         buffer.putInt(payloadLength);

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

@@ -1014,6 +1014,8 @@ private void ensureConnected() {
 
             // Connect and upgrade to WebSocket
             try {
+                client.setQwpMaxVersion(MAX_SUPPORTED_VERSION);
+                client.setQwpClientId(CLIENT_ID);
                 client.connect(host, port);
                 client.upgrade(WRITE_PATH, authorizationHeader);
             } catch (Exception e) {
@@ -1041,11 +1043,15 @@ private void ensureConnected() {
             }
             // Sync mode (window=1): no send queue - we send and read ACKs synchronously
 
+            // Use the version selected by the server
+            encoder.setVersion((byte) client.getServerQwpVersion());
+
             // Clear sent schema hashes - server starts fresh on each connection
             sentSchemaHashes.clear();
 
             connected = true;
-            LOG.info("Connected to WebSocket [host={}, port={}, windowSize={}]", host, port, inFlightWindowSize);
+            LOG.info("Connected to WebSocket [host={}, port={}, windowSize={}, qwpVersion={}]",
+                    host, port, inFlightWindowSize, client.getServerQwpVersion());
         }
     }
 

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

@@ -240,10 +240,18 @@ public final class QwpConstants {
      * Column type: VARCHAR (length-prefixed UTF-8, aux storage).
      */
     public static final byte TYPE_VARCHAR = 0x0F;
+    /**
+     * Client identifier sent in the X-QWP-Client-Id upgrade header.
+     */
+    public static final String CLIENT_ID = "java/1.0.2";
     /**
      * Current protocol version.
      */
     public static final byte VERSION_1 = 1;
+    /**
+     * Maximum protocol version supported by this client.
+     */
+    public static final byte MAX_SUPPORTED_VERSION = VERSION_1;
 
     private QwpConstants() {
         // utility class

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

@@ -42,6 +42,34 @@
  */
 public class QwpWebSocketEncoderTest {
 
+    @Test
+    public void testVersionByteInHeader() throws Exception {
+        assertMemoryLeak(() -> {
+            try (QwpWebSocketEncoder encoder = new QwpWebSocketEncoder();
+                 QwpTableBuffer buffer = new QwpTableBuffer("test")) {
+
+                QwpTableBuffer.ColumnBuffer col = buffer.getOrCreateColumn("x", TYPE_LONG, false);
+                col.addLong(42);
+                buffer.nextRow();
+
+                // Default version
+                int size = encoder.encode(buffer, false);
+                Assert.assertTrue(size > 0);
+                long ptr = encoder.getBuffer().getBufferPtr();
+                Assert.assertEquals(1, Unsafe.getUnsafe().getByte(ptr + 4));
+
+                // Custom version
+                buffer.reset();
+                col = buffer.getOrCreateColumn("x", TYPE_LONG, false);
+                col.addLong(42);
+                buffer.nextRow();
+                encoder.setVersion((byte) 3);
+                encoder.encode(buffer, false);
+                Assert.assertEquals(3, Unsafe.getUnsafe().getByte(ptr + 4));
+            }
+        });
+    }
+
     @Test
     public void testBufferResetAndReuse() throws Exception {
         assertMemoryLeak(() -> {