feat(qwp): add QuestDB facade with pooled Sender and Query APIs (#28)

Author: bluestreak01

ci/validate-pr-title/validate.js

@@ -16,6 +16,7 @@ const allowedSubTypes = [
   "log",
   "core",
   "ilp",
+  "qwp",
   "http",
   "conf",
   "utils",

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

@@ -0,0 +1,79 @@
+/*+*****************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  Copyright (c) 2014-2019 Appsicle
+ *  Copyright (c) 2019-2026 QuestDB
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ ******************************************************************************/
+
+package io.questdb.client;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Async handle for a submitted {@link Query}. Returned by {@link Query#submit()}.
+ * <p>
+ * Lifecycle: the Completion is allocated once as a field on the per-thread
+ * {@link Query} instance and is reused on every {@code submit()}. It is
+ * single-flight: a new {@code submit()} cannot be issued on the same {@link Query}
+ * until the previous Completion resolves (via {@link #await()},
+ * {@link #await(long, TimeUnit)} returning {@code true}, or an explicit
+ * {@link #cancel()} that races to terminal).
+ * <p>
+ * Signaling: the Completion is signaled from the I/O thread of the pooled
+ * query client when the handler's terminal callback ({@code onEnd},
+ * {@code onError}, or {@code onExecDone}) returns.
+ */
+public interface Completion {
+
+    /**
+     * Blocks until the query completes. Rethrows any server-reported failure
+     * as a {@link QueryException}. Returns normally on success.
+     *
+     * @throws QueryException       if the server reported an error or
+     *                              {@link #cancel()} won the race
+     * @throws InterruptedException if the calling thread is interrupted
+     *                              while waiting
+     */
+    void await() throws InterruptedException;
+
+    /**
+     * Blocks up to the given timeout. Returns {@code true} if the query
+     * completed, {@code false} on timeout.
+     *
+     * @throws QueryException       if the server reported an error or
+     *                              {@link #cancel()} won the race
+     * @throws InterruptedException if the calling thread is interrupted
+     *                              while waiting
+     */
+    boolean await(long timeout, TimeUnit unit) throws InterruptedException;
+
+    /**
+     * Requests cancellation of the in-flight query. The handler's
+     * {@code onError} fires with a cancellation status. No-op if the query
+     * has already completed.
+     */
+    void cancel();
+
+    /**
+     * Returns true once the query has terminated (success, error, or cancel
+     * acknowledged).
+     */
+    boolean isDone();
+}

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

@@ -0,0 +1,75 @@
+/*+*****************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  Copyright (c) 2014-2019 Appsicle
+ *  Copyright (c) 2019-2026 QuestDB
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ ******************************************************************************/
+
+package io.questdb.client;
+
+import io.questdb.client.cutlass.qwp.client.QwpBindSetter;
+import io.questdb.client.cutlass.qwp.client.QwpColumnBatchHandler;
+
+/**
+ * Per-thread, reusable builder for one query. Obtained from
+ * {@link QuestDB#query()}: every call on the same thread returns the same
+ * instance, reset to empty.
+ * <p>
+ * Lifecycle: configure with {@link #sql}, optional {@link #binds}, and
+ * {@link #handler}, then call {@link #submit()} to obtain a {@link Completion}.
+ * After the Completion terminates, the next {@code QuestDB.query()} call on
+ * the same thread returns this same instance with its state reset.
+ * <p>
+ * Thread safety: not thread-safe. One in-flight query per thread.
+ */
+public interface Query {
+
+    /** Discards the current configuration without submitting. */
+    void abandon();
+
+    /**
+     * Sets the bind-value setter, invoked by the pooled query client when the
+     * QUERY_REQUEST frame is being prepared. Pass a reusable
+     * {@link QwpBindSetter} instance (or a stateless lambda hoisted to a
+     * field) to keep submission zero-allocation.
+     */
+    Query binds(QwpBindSetter binds);
+
+    /**
+     * Sets the result-batch handler. The handler is invoked on the pooled
+     * query client's I/O thread; if it touches caller state, it is
+     * responsible for its own synchronization.
+     */
+    Query handler(QwpColumnBatchHandler handler);
+
+    /**
+     * Sets the SQL text. The buffer is not retained past {@link #submit()}.
+     */
+    Query sql(CharSequence sql);
+
+    /**
+     * Submits the query for execution. Returns the {@link Completion} field
+     * cached on this instance; never allocates. Blocks up to the builder's
+     * configured acquire timeout if the query pool is exhausted.
+     *
+     * @return the single-flight Completion bound to this Query instance
+     */
+    Completion submit();
+}

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

@@ -0,0 +1,59 @@
+/*+*****************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  Copyright (c) 2014-2019 Appsicle
+ *  Copyright (c) 2019-2026 QuestDB
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ ******************************************************************************/
+
+package io.questdb.client;
+
+/**
+ * Thrown from {@link Completion#await()} / {@link Completion#await(long, java.util.concurrent.TimeUnit)}
+ * when the server reported an error for the corresponding {@link Query},
+ * when {@link Completion#cancel()} won the race, or when an unrecoverable
+ * transport failure occurred during submission.
+ * <p>
+ * The original wire-level status byte is exposed via {@link #getStatus()} so
+ * callers can distinguish cancellation from schema errors etc. without
+ * string-matching the message.
+ */
+public class QueryException extends RuntimeException {
+
+    private final byte status;
+
+    public QueryException(byte status, String message) {
+        super(message);
+        this.status = status;
+    }
+
+    public QueryException(byte status, String message, Throwable cause) {
+        super(message, cause);
+        this.status = status;
+    }
+
+    /**
+     * Returns the server-reported wire status byte (see QWP protocol
+     * definitions), or {@code 0} if this exception was raised by the client
+     * (for example, transport failure before any server response).
+     */
+    public byte getStatus() {
+        return status;
+    }
+}