feat(ilp): binary wire protocol

core/pom.xml

@@ -32,7 +32,7 @@
         <maven.javadoc.skip>false</maven.javadoc.skip>
         <outputPath>target</outputPath>
         <runtime.assembly>none</runtime.assembly>
-        <javac.target>11</javac.target>
+        <javac.target>17</javac.target>
         <argLine>-ea -Dfile.encoding=UTF-8 -XX:+UseParallelGC</argLine>
         <test.exclude>None</test.exclude>
         <test.include>%regex[.*[^o].class]</test.include><!-- exclude module-info.class-->

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

@@ -41,7 +41,7 @@ public BuildInformationHolder(Class<?> clazz) {
         String swVersion;
         try {
             final Attributes manifestAttributes = getManifestAttributes(clazz);
-            swVersion = getAttr(manifestAttributes, "QuestDB-Client-Version", "[DEVELOPMENT]");
+            swVersion = getAttr(manifestAttributes, "[DEVELOPMENT]");
         } catch (IOException e) {
             swVersion = UNKNOWN;
         }
@@ -57,8 +57,8 @@ public String getSwVersion() {
         return swVersion;
     }
 
-    private static String getAttr(final Attributes manifestAttributes, String attributeName, String defaultValue) {
-        final String value = manifestAttributes.getValue(attributeName);
+    private static String getAttr(final Attributes manifestAttributes, String defaultValue) {
+        final String value = manifestAttributes.getValue("QuestDB-Client-Version");
         return value != null ? value : defaultValue;
     }
 

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

@@ -54,6 +54,10 @@ default int getMaximumRequestBufferSize() {
         return Integer.MAX_VALUE;
     }
 
+    default int getMaximumResponseBufferSize() {
+        return Integer.MAX_VALUE;
+    }
+
     default NetworkFacade getNetworkFacade() {
         return NetworkFacadeImpl.INSTANCE;
     }

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

@@ -34,6 +34,7 @@
 import io.questdb.client.cutlass.line.http.AbstractLineHttpSender;
 import io.questdb.client.cutlass.line.tcp.DelegatingTlsChannel;
 import io.questdb.client.cutlass.line.tcp.PlainTcpLineChannel;
+import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
 import io.questdb.client.impl.ConfStringParser;
 import io.questdb.client.network.NetworkFacade;
 import io.questdb.client.network.NetworkFacadeImpl;
@@ -95,7 +96,7 @@
  *   2. Call {@link #reset()} to clear the internal buffers and start building a new row
  * <br>
  * Note: If the underlying error is permanent, retrying {@link #flush()} will fail again.
- * Use {@link #reset()} to discard the problematic data and continue with new data. See {@link LineSenderException#isRetryable()}
+ * Use {@link #reset()} to discard the problematic data and continue with new data.
  *
  */
 public interface Sender extends Closeable, ArraySender<Sender> {
@@ -108,7 +109,7 @@ public interface Sender extends Closeable, ArraySender<Sender> {
     /**
      * Create a Sender builder instance from a configuration string.
      * <br>
-     * This allows to use the configuration string as a template for creating a Sender builder instance and then
+     * This allows using the configuration string as a template for creating a Sender builder instance and then
      * tune options which are not available in the configuration string. Configurations options specified in the
      * configuration string cannot be overridden via the builder methods.
      * <p>
@@ -144,7 +145,12 @@ static LineSenderBuilder builder(CharSequence configurationString) {
      * @return Builder object to create a new Sender instance.
      */
     static LineSenderBuilder builder(Transport transport) {
-        return new LineSenderBuilder(transport == Transport.HTTP ? LineSenderBuilder.PROTOCOL_HTTP : LineSenderBuilder.PROTOCOL_TCP);
+        int protocol = switch (transport) {
+            case HTTP -> LineSenderBuilder.PROTOCOL_HTTP;
+            case TCP -> LineSenderBuilder.PROTOCOL_TCP;
+            case WEBSOCKET -> LineSenderBuilder.PROTOCOL_WEBSOCKET;
+        };
+        return new LineSenderBuilder(protocol);
     }
 
     /**
@@ -461,7 +467,15 @@ enum Transport {
          * and for use-cases where HTTP transport is not suitable, when communicating with a QuestDB server over a high-latency
          * network
          */
-        TCP
+        TCP,
+
+        /**
+         * Use WebSocket transport to communicate with a QuestDB server.
+         * <p>
+         * WebSocket transport uses the ILP v4 binary protocol for efficient data ingestion.
+         * It supports both synchronous and asynchronous modes with flow control.
+         */
+        WEBSOCKET
     }
 
     /**
@@ -508,12 +522,17 @@ final class LineSenderBuilder {
         private static final int DEFAULT_BUFFER_CAPACITY = 64 * 1024;
         private static final int DEFAULT_HTTP_PORT = 9000;
         private static final int DEFAULT_HTTP_TIMEOUT = 30_000;
+        private static final int DEFAULT_IN_FLIGHT_WINDOW_SIZE = 8;
         private static final int DEFAULT_MAXIMUM_BUFFER_CAPACITY = 100 * 1024 * 1024;
         private static final int DEFAULT_MAX_BACKOFF_MILLIS = 1_000;
         private static final int DEFAULT_MAX_NAME_LEN = 127;
         private static final long DEFAULT_MAX_RETRY_NANOS = TimeUnit.SECONDS.toNanos(10); // keep sync with the contract of the configuration method
         private static final long DEFAULT_MIN_REQUEST_THROUGHPUT = 100 * 1024; // 100KB/s, keep in sync with the contract of the configuration method
         private static final int DEFAULT_TCP_PORT = 9009;
+        private static final int DEFAULT_WEBSOCKET_PORT = 9000;
+        private static final int DEFAULT_WS_AUTO_FLUSH_BYTES = 1024 * 1024; // 1MB
+        private static final long DEFAULT_WS_AUTO_FLUSH_INTERVAL_NANOS = 100_000_000L; // 100ms
+        private static final int DEFAULT_WS_AUTO_FLUSH_ROWS = 500;
         private static final int MIN_BUFFER_SIZE = AuthUtils.CHALLENGE_LEN + 1; // challenge size + 1;
         // The PARAMETER_NOT_SET_EXPLICITLY constant is used to detect if a parameter was set explicitly in configuration parameters
         // where it matters. This is needed to detect invalid combinations of parameters. Why?
@@ -522,15 +541,19 @@ final class LineSenderBuilder {
         private static final int PARAMETER_NOT_SET_EXPLICITLY = -1;
         private static final int PROTOCOL_HTTP = 1;
         private static final int PROTOCOL_TCP = 0;
+        private static final int PROTOCOL_WEBSOCKET = 2;
         private final ObjList<String> hosts = new ObjList<>();
         private final IntList ports = new IntList();
+        private boolean asyncMode = false;
+        private int autoFlushBytes = PARAMETER_NOT_SET_EXPLICITLY;
         private int autoFlushIntervalMillis = PARAMETER_NOT_SET_EXPLICITLY;
         private int autoFlushRows = PARAMETER_NOT_SET_EXPLICITLY;
         private int bufferCapacity = PARAMETER_NOT_SET_EXPLICITLY;
         private String httpPath;
         private String httpSettingsPath;
         private int httpTimeout = PARAMETER_NOT_SET_EXPLICITLY;
         private String httpToken;
+        private int inFlightWindowSize = PARAMETER_NOT_SET_EXPLICITLY;
         private String keyId;
         private int maxBackoffMillis = PARAMETER_NOT_SET_EXPLICITLY;
         private int maxNameLength = PARAMETER_NOT_SET_EXPLICITLY;
@@ -658,6 +681,47 @@ public AdvancedTlsSettings advancedTls() {
             return new AdvancedTlsSettings();
         }
 
+        /**
+         * Enable asynchronous mode for WebSocket transport.
+         * <br>
+         * In async mode, rows are batched and sent asynchronously with flow control.
+         * This provides higher throughput at the cost of more complex error handling.
+         * <br>
+         * This is only used when communicating over WebSocket transport.
+         * <br>
+         * Default is synchronous mode (false).
+         *
+         * @param enabled whether to enable async mode
+         * @return this instance for method chaining
+         */
+        public LineSenderBuilder asyncMode(boolean enabled) {
+            this.asyncMode = enabled;
+            return this;
+        }
+
+        /**
+         * Set the maximum number of bytes per batch before auto-flushing.
+         * <br>
+         * This is only used when communicating over WebSocket transport.
+         * <br>
+         * Default value is 1MB.
+         *
+         * @param bytes maximum bytes per batch
+         * @return this instance for method chaining
+         */
+        public LineSenderBuilder autoFlushBytes(int bytes) {
+            if (this.autoFlushBytes != PARAMETER_NOT_SET_EXPLICITLY) {
+                throw new LineSenderException("auto flush bytes was already configured")
+                        .put("[bytes=").put(this.autoFlushBytes).put("]");
+            }
+            if (bytes < 0) {
+                throw new LineSenderException("auto flush bytes cannot be negative")
+                        .put("[bytes=").put(bytes).put("]");
+            }
+            this.autoFlushBytes = bytes;
+            return this;
+        }
+
         /**
          * Set the interval in milliseconds at which the Sender automatically flushes its buffer.
          * <br>
@@ -791,6 +855,40 @@ public Sender build() {
                         username, password, maxNameLength, actualMaxRetriesNanos, maxBackoffMillis, actualMinRequestThroughput, actualAutoFlushIntervalMillis, protocolVersion);
             }
 
+            if (protocol == PROTOCOL_WEBSOCKET) {
+                if (hosts.size() != 1 || ports.size() != 1) {
+                    throw new LineSenderException("only a single address (host:port) is supported for WebSocket transport");
+                }
+
+                int actualAutoFlushRows = autoFlushRows == PARAMETER_NOT_SET_EXPLICITLY ? DEFAULT_WS_AUTO_FLUSH_ROWS : autoFlushRows;
+                int actualAutoFlushBytes = autoFlushBytes == PARAMETER_NOT_SET_EXPLICITLY ? DEFAULT_WS_AUTO_FLUSH_BYTES : autoFlushBytes;
+                long actualAutoFlushIntervalNanos = autoFlushIntervalMillis == PARAMETER_NOT_SET_EXPLICITLY
+                        ? DEFAULT_WS_AUTO_FLUSH_INTERVAL_NANOS
+                        : TimeUnit.MILLISECONDS.toNanos(autoFlushIntervalMillis);
+                int actualInFlightWindowSize = inFlightWindowSize == PARAMETER_NOT_SET_EXPLICITLY ? DEFAULT_IN_FLIGHT_WINDOW_SIZE : inFlightWindowSize;
+
+                if (asyncMode) {
+                    return QwpWebSocketSender.connectAsync(
+                            hosts.getQuick(0),
+                            ports.getQuick(0),
+                            tlsEnabled,
+                            actualAutoFlushRows,
+                            actualAutoFlushBytes,
+                            actualAutoFlushIntervalNanos,
+                            actualInFlightWindowSize
+                    );
+                } else {
+                    return QwpWebSocketSender.connect(
+                            hosts.getQuick(0),
+                            ports.getQuick(0),
+                            tlsEnabled,
+                            actualAutoFlushRows,
+                            actualAutoFlushBytes,
+                            actualAutoFlushIntervalNanos
+                    );
+                }
+            }
+
             assert protocol == PROTOCOL_TCP;
 
             if (hosts.size() != 1 || ports.size() != 1) {
@@ -1048,6 +1146,29 @@ public LineSenderBuilder httpUsernamePassword(String username, String password)
             return this;
         }
 
+        /**
+         * Set the maximum number of batches that can be in-flight awaiting server acknowledgment.
+         * <br>
+         * This is only used when communicating over WebSocket transport with async mode enabled.
+         * <br>
+         * Default value is 8.
+         *
+         * @param size maximum number of in-flight batches
+         * @return this instance for method chaining
+         */
+        public LineSenderBuilder inFlightWindowSize(int size) {
+            if (this.inFlightWindowSize != PARAMETER_NOT_SET_EXPLICITLY) {
+                throw new LineSenderException("in-flight window size was already configured")
+                        .put("[size=").put(this.inFlightWindowSize).put("]");
+            }
+            if (size < 1) {
+                throw new LineSenderException("in-flight window size must be positive")
+                        .put("[size=").put(size).put("]");
+            }
+            this.inFlightWindowSize = size;
+            return this;
+        }
+
         /**
          * Configures the maximum backoff time between retry attempts when the Sender encounters recoverable errors.
          * <br>
@@ -1275,7 +1396,13 @@ private void configureDefaults() {
                 maximumBufferCapacity = protocol == PROTOCOL_HTTP ? DEFAULT_MAXIMUM_BUFFER_CAPACITY : bufferCapacity;
             }
             if (ports.size() == 0) {
-                ports.add(protocol == PROTOCOL_HTTP ? DEFAULT_HTTP_PORT : DEFAULT_TCP_PORT);
+                if (protocol == PROTOCOL_HTTP) {
+                    ports.add(DEFAULT_HTTP_PORT);
+                } else if (protocol == PROTOCOL_WEBSOCKET) {
+                    ports.add(DEFAULT_WEBSOCKET_PORT);
+                } else {
+                    ports.add(DEFAULT_TCP_PORT);
+                }
             }
             if (tlsValidationMode == null) {
                 tlsValidationMode = TlsValidationMode.DEFAULT;
@@ -1334,8 +1461,16 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
             } else if (Chars.equals("tcps", sink)) {
                 tcp();
                 tlsEnabled = true;
+            } else if (Chars.equals("ws", sink)) {
+                if (tlsEnabled) {
+                    throw new LineSenderException("cannot use ws protocol when TLS is enabled. use wss instead");
+                }
+                websocket();
+            } else if (Chars.equals("wss", sink)) {
+                websocket();
+                tlsEnabled = true;
             } else {
-                throw new LineSenderException("invalid schema [schema=").put(sink).put(", supported-schemas=[http, https, tcp, tcps]]");
+                throw new LineSenderException("invalid schema [schema=").put(sink).put(", supported-schemas=[http, https, tcp, tcps, ws, wss]]");
             }
 
             String tcpToken = null;
@@ -1357,7 +1492,9 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
                     address(sink);
                     if (ports.size() == hosts.size() - 1) {
                         // not set
-                        port(protocol == PROTOCOL_TCP ? DEFAULT_TCP_PORT : DEFAULT_HTTP_PORT);
+                        port(protocol == PROTOCOL_TCP ? DEFAULT_TCP_PORT
+                                : protocol == PROTOCOL_WEBSOCKET ? DEFAULT_WEBSOCKET_PORT
+                                : DEFAULT_HTTP_PORT);
                     }
                 } else if (Chars.equals("user", sink)) {
                     // deprecated key: user, new key: username
@@ -1414,7 +1551,7 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
                     } else if (protocol == PROTOCOL_HTTP) {
                         httpToken(sink.toString());
                     } else {
-                        throw new AssertionError();
+                        throw new LineSenderException("token is not supported for WebSocket protocol");
                     }
                 } else if (Chars.equals("retry_timeout", sink)) {
                     pos = getValue(configurationString, pos, sink, "retry_timeout");
@@ -1617,12 +1754,31 @@ private void validateParameters() {
                 if (autoFlushIntervalMillis != PARAMETER_NOT_SET_EXPLICITLY) {
                     throw new LineSenderException("auto flush interval is not supported for TCP protocol");
                 }
+            } else if (protocol == PROTOCOL_WEBSOCKET) {
+                if (privateKey != null) {
+                    throw new LineSenderException("TCP authentication is not supported for WebSocket protocol");
+                }
+                if (httpToken != null || username != null || password != null) {
+                    // TODO: WebSocket auth not yet implemented
+                    throw new LineSenderException("Authentication is not yet supported for WebSocket protocol");
+                }
+                if (inFlightWindowSize != PARAMETER_NOT_SET_EXPLICITLY && !asyncMode) {
+                    throw new LineSenderException("in-flight window size requires async mode");
+                }
             } else {
                 throw new LineSenderException("unsupported protocol ")
                         .put("[protocol=").put(protocol).put("]");
             }
         }
 
+        private void websocket() {
+            if (protocol != PARAMETER_NOT_SET_EXPLICITLY) {
+                throw new LineSenderException("protocol was already configured ")
+                        .put("[protocol=").put(protocol).put("]");
+            }
+            protocol = PROTOCOL_WEBSOCKET;
+        }
+
         public class AdvancedTlsSettings {
             /**
              * Configure a custom truststore. This is only needed when using {@link #enableTls()} when your default