Make query egress symmetric with sender ingest

Reshape the QuestDB facade so reads and writes share one pooled-lease
model, and remove the thread-affine footguns on both sides.

Ingest:
- Remove QuestDB.sender() and releaseSender(), along with the entire
  thread-pin subsystem behind them (SenderPool.pinToCurrentThread,
  releaseCurrentThread, clearPinIfCurrent, the threadAffine ThreadLocal,
  and the PooledSender invalidated flag that existed only to make pinning
  safe). borrowSender() is now the only way to lease a Sender.

Egress:
- Add QuestDB.borrowQuery(), a closeable, non-allocating Query lease that
  mirrors borrowSender(). Each pooled QueryWorker owns one pre-allocated
  QueryImpl, handed out reset on borrow; submit() dispatches on the held
  worker (single-flight) and close() returns it to the pool. The worker
  no longer auto-releases per query.
- Remove query(), newQuery(), and executeSql(). Reads now connect at
  borrow time rather than submit time; under lazy_connect the read pool
  still defaults to min=0, so build() does not fail-fast while the server
  is down.

Test seams:
- Make the white-box seam constructors public and annotate @testonly
  where production never calls them (QuestDBImpl, SenderPool). The
  QueryClientPool connectHook ctor stays public without @testonly because
  QuestDBImpl constructs the query pool through it. Tests now call these
  constructors directly instead of via reflection.

Update the client tests, the usage example, and the startup/failover
design doc to the new API.

Author: bluestreak01

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

@@ -27,19 +27,27 @@
 import io.questdb.client.cutlass.qwp.client.QwpBindSetter;
 import io.questdb.client.cutlass.qwp.client.QwpColumnBatchHandler;
 
+import java.io.Closeable;
+
 /**
- * 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.
+ * A query handle leased from the {@link QuestDB} pool via
+ * {@link QuestDB#borrowQuery()}. The handle holds one pooled query client (one
+ * WebSocket + I/O thread) for the lifetime of the borrow; the caller MUST
+ * {@link #close()} it to release the client back to the pool (typically via
+ * try-with-resources).
+ * <p>
+ * Allocation: zero at steady state -- {@code borrowQuery()} returns a
+ * pre-allocated handle bound to the leased pool slot, and the
+ * {@link Completion} is a field on it reused across submits.
  * <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.
+ * {@link #handler}, then call {@link #submit()} to obtain a {@link Completion}
+ * and {@code await()} it before the next {@link #submit()}.
  * <p>
- * Thread safety: not thread-safe. One in-flight query per thread.
+ * Thread safety: not thread-safe and single-flight -- one in-flight query per
+ * handle. To run queries concurrently, borrow one handle per concurrent query.
  */
-public interface Query {
+public interface Query extends Closeable {
 
     /** Discards the current configuration without submitting. */
     void abandon();
@@ -52,6 +60,16 @@ public interface Query {
      */
     Query binds(QwpBindSetter binds);
 
+    /**
+     * Releases the leased pooled query client back to the pool. The caller
+     * MUST call this (typically via try-with-resources). If a submit is still
+     * in flight, {@code close()} cancels it and waits for the terminal event
+     * before returning the client. A real disconnect only happens at
+     * {@link QuestDB#close()}. Idempotent.
+     */
+    @Override
+    void close();
+
     /**
      * 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
@@ -65,11 +83,12 @@ public interface Query {
     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.
+     * Submits the query for execution on the leased client. Returns the
+     * {@link Completion} field cached on this handle; never allocates. The
+     * handle is single-flight: {@code await()} the returned Completion before
+     * the next {@code submit()}.
      *
-     * @return the single-flight Completion bound to this Query instance
+     * @return the single-flight Completion bound to this Query handle
      */
     Completion submit();
 }

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

@@ -24,8 +24,6 @@
 
 package io.questdb.client;
 
-import io.questdb.client.cutlass.qwp.client.QwpColumnBatchHandler;
-
 import java.io.Closeable;
 
 /**
@@ -34,9 +32,9 @@
  * share across threads.
  * <p>
  * Steady-state allocation is zero: pooled instances are pre-allocated and
- * reused, the per-thread {@link Query} handle is cached in a {@code ThreadLocal},
- * and the {@link Completion} associated with each query is a field on that
- * cached handle.
+ * reused, each borrowed {@link Query} handle is a pre-allocated front bound to
+ * its pool slot, and the {@link Completion} associated with each query is a
+ * field on that handle.
  * <p>
  * Configuration: one {@code ws}/{@code wss} string describes the whole cluster
  * (a single {@code addr} server list) and both the ingest and query pools
@@ -47,7 +45,7 @@
  * is up).
  * <p>
  * Thread safety: instances are safe to share. {@link #borrowSender()} and
- * {@link #query()} may be called concurrently from any thread; the pool
+ * {@link #borrowQuery()} may be called concurrently from any thread; the pool
  * guarantees mutual exclusion of pooled resources.
  */
 public interface QuestDB extends Closeable {
@@ -80,6 +78,31 @@ static QuestDB connect(CharSequence configurationString) {
         return builder().fromConfig(configurationString).build();
     }
 
+    /**
+     * Borrows a {@link Query} handle from the pool. The caller MUST call
+     * {@link Query#close()} on the returned instance to release it back to the
+     * pool (typically via try-with-resources). The handle leases one pooled
+     * query client (one WebSocket + I/O thread) for the borrow's lifetime;
+     * submit one or more queries on it, then close it.
+     * <p>
+     * Allocation: zero at steady state -- the returned instance is a
+     * pre-allocated handle bound to the leased pool slot.
+     * <p>
+     * Blocking: blocks up to the builder's
+     * {@link QuestDBBuilder#acquireTimeoutMillis(long) acquire timeout} when
+     * the pool is exhausted; throws on timeout.
+     * <p>
+     * Concurrency: a single handle is single-flight. To run queries
+     * concurrently, borrow one handle per concurrent query (up to
+     * {@code query_pool_max}).
+     *
+     * @return a Query handle leased from the pool; release with
+     * {@link Query#close()}
+     * @throws QueryException if the pool is exhausted beyond the acquire
+     *                        timeout, or if this handle is closed
+     */
+    Query borrowQuery();
+
     /**
      * Borrows a {@link Sender} from the pool. The caller MUST call
      * {@link Sender#close()} on the returned instance to release it back to
@@ -114,61 +137,4 @@ static QuestDB connect(CharSequence configurationString) {
      */
     @Override
     void close();
-
-    /**
-     * One-shot convenience for queries with no bind parameters. Equivalent to
-     * {@code query().sql(sql).handler(handler).submit()}. Returns the same
-     * thread-local {@link Completion} instance that {@link #query()} would,
-     * so this method is also zero-allocation at steady state.
-     *
-     * @param sql     the SQL text; the buffer is not retained after submit
-     * @param handler the result-batch handler; invoked on the pooled query
-     *                client's I/O thread
-     * @return a single-flight handle for the in-flight query
-     */
-    Completion executeSql(CharSequence sql, QwpColumnBatchHandler handler);
-
-    /**
-     * Allocates a fresh {@link Query} handle. Unlike {@link #query()}, this
-     * does NOT return the per-thread cached instance; every call allocates.
-     * <p>
-     * Use this when one thread needs to hold multiple in-flight queries
-     * concurrently (each {@code submit()} acquires its own worker from the
-     * query pool, so up to {@code queryPoolSize} concurrent queries on a
-     * single thread is fine). For the common case of one query at a time,
-     * prefer {@link #query()} -- it is allocation-free.
-     */
-    Query newQuery();
-
-    /**
-     * Opens a query builder for the calling thread. Returns the same
-     * thread-local instance on every call: callers do not need to cache it
-     * themselves. The returned {@code Query} is in a reset state and is not
-     * thread-safe -- one in-flight query per thread.
-     * <p>
-     * For multiple concurrent in-flight queries from a single thread, use
-     * {@link #newQuery()} instead.
-     */
-    Query query();
-
-    /**
-     * Releases the thread-affine {@link Sender} (if any) currently attached
-     * to the calling thread back to the pool. Call this on threads borrowed
-     * from pools you do not own (for example, Netty event loops) before they
-     * are recycled, to avoid pinning a {@link Sender} for the lifetime of
-     * a thread that no longer needs it.
-     */
-    void releaseSender();
-
-    /**
-     * Returns a {@link Sender} pinned to the calling thread. First call on
-     * a thread takes one from the pool and pins it; subsequent calls on the
-     * same thread return the same instance. The pin is released by
-     * {@link #releaseSender()} or by {@link #close()} on this handle.
-     * <p>
-     * Use this for long-lived, dedicated producer threads where borrow/return
-     * overhead would dominate. For short-lived or event-loop callers, prefer
-     * {@link #borrowSender()}.
-     */
-    Sender sender();
 }

core/src/main/java/io/questdb/client/impl/PooledSender.java

@@ -59,7 +59,6 @@ public final class PooledSender implements Sender {
     private final int slotIndex;
     private volatile long idleSinceMillis;
     private volatile boolean inUse;
-    private volatile boolean invalidated;
 
     PooledSender(Sender delegate, SenderPool pool, int slotIndex) {
         this.delegate = delegate;
@@ -141,15 +140,6 @@ public Sender charColumn(CharSequence name, char value) {
      * the owning {@code QuestDB} is closed.
      * <p>
      * Idempotent: a second call after a return is a no-op.
-     * <p>
-     * Clears the current thread's pin (if any) before the slot becomes
-     * borrowable again. Without this step a thread that pinned this
-     * wrapper and then closed it via the public {@link Sender#close()}
-     * (the natural try-with-resources idiom) would still hold the pin
-     * in its {@link ThreadLocal}; a subsequent {@code QuestDB.sender()}
-     * call on that thread would return the cached wrapper even though
-     * another consumer has since borrowed the slot, and the two
-     * consumers would write to the same underlying delegate.
      */
     @Override
     public void close() {
@@ -167,12 +157,6 @@ public void close() {
             flushed = true;
         } finally {
             inUse = false;
-            // Clear the pin BEFORE returning the slot. If we cleared
-            // after giveBack(), a concurrent borrower could grab the
-            // slot while this thread's pin still references it, and a
-            // re-pin on this thread would return the (now in-use)
-            // wrapper -- the same race this clear is meant to close.
-            pool.clearPinIfCurrent(this);
             if (flushed) {
                 pool.giveBack(this);
             } else {
@@ -401,19 +385,11 @@ boolean isInUse() {
         return inUse;
     }
 
-    boolean isInvalidated() {
-        return invalidated;
-    }
-
     void markIdleAt(long nowMillis) {
         idleSinceMillis = nowMillis;
     }
 
     void markInUse() {
         inUse = true;
     }
-
-    void markInvalidated() {
-        invalidated = true;
-    }
 }

core/src/main/java/io/questdb/client/impl/QueryClientPool.java

@@ -26,6 +26,7 @@
 
 import io.questdb.client.QueryException;
 import io.questdb.client.cutlass.qwp.client.QwpQueryClient;
+import org.jetbrains.annotations.TestOnly;
 
 import java.util.ArrayDeque;
 import java.util.ArrayList;
@@ -89,11 +90,12 @@ public QueryClientPool(
                 idleTimeoutMillis, maxLifetimeMillis, null);
     }
 
-    // Package-private constructor exposing the connectHook test seam: production
-    // passes null (-> the real QwpQueryClient.connect()). White-box tests in
-    // io.questdb.client.test.impl reach this by reflection to inject a hook that
-    // throws a non-RuntimeException Throwable from the native connect path.
-    QueryClientPool(
+    // Constructor exposing the connectHook seam. Production (QuestDBImpl) passes
+    // null -> the real QwpQueryClient.connect(); white-box tests pass a hook that
+    // throws a non-RuntimeException Throwable from the native connect path. This
+    // is the construction path QuestDBImpl uses, so it is a real (public) ctor,
+    // not test-only.
+    public QueryClientPool(
             String configurationString,
             int minSize,
             int maxSize,
@@ -106,13 +108,12 @@ public QueryClientPool(
                 idleTimeoutMillis, maxLifetimeMillis, connectHook, null);
     }
 
-    // Package-private constructor exposing both the connectHook and startHook
-    // test seams: production passes null for each (-> the real
-    // QwpQueryClient.connect() and QueryWorker.start()). White-box tests in
-    // io.questdb.client.test.impl reach this by reflection to inject a hook that
-    // throws a Throwable from either the native connect path (connectHook) or
-    // the worker thread-start path (startHook).
-    QueryClientPool(
+    // Constructor exposing both the connectHook and startHook seams. Production
+    // reaches it via the overload above (both null -> the real
+    // QwpQueryClient.connect() and QueryWorker.start()); white-box tests pass a
+    // hook that throws a Throwable from either the native connect path
+    // (connectHook) or the worker thread-start path (startHook).
+    public QueryClientPool(
             String configurationString,
             int minSize,
             int maxSize,
@@ -355,11 +356,12 @@ void release(QueryWorker w) {
         }
     }
 
-    // Package-private white-box accessor for tests: reports the current
-    // in-flight creation count under the pool lock. A non-zero value after a
-    // failed acquire() means the slot reservation was never released -- the
-    // capacity-shrink bug this guards against.
-    int inFlightCreations() {
+    // White-box accessor for tests: reports the current in-flight creation count
+    // under the pool lock. A non-zero value after a failed acquire() means the
+    // slot reservation was never released -- the capacity-shrink bug this guards
+    // against.
+    @TestOnly
+    public int inFlightCreations() {
         lock.lock();
         try {
             return inFlightCreations;