[pull] master from cube-js:master

docs-mintlify/reference/configuration/environment-variables.mdx

@@ -195,11 +195,8 @@ and we don't recommend enabling it in production yet.
 
 Enables the preview fast-track path for Cube Store query queues. When enabled,
 Cube can atomically enqueue and retrieve a query when queue concurrency is
-available, reducing queue coordination round trips. It applies to queries at
-queue priority 10 and above, which is where user-facing queries and the
-pre-aggregation builds a request waits on are submitted; background refresh runs
-below that and keeps using the regular path. This requires Cube Store
-1.7.25 or newer; against an older version Cube keeps using the regular path.
+available, reducing queue coordination round trips. This requires Cube and Cube Store
+1.7.26 or newer; against an older version Cube keeps using the regular path.
 
 | Possible Values | Default in Development | Default in Production |
 | --------------- | ---------------------- | --------------------- |

packages/cubejs-base-driver/src/queue-driver.interface.ts

@@ -9,25 +9,11 @@ export type QueryKeyHash = string & { __type: 'QueryKeyHash' };
 export type QueryKeysTuple = [keyHash: QueryKeyHash, queueId: QueueId];
 export type GetActiveAndToProcessResponse = [active: QueryKeysTuple[], toProcess: QueryKeysTuple[]];
 export type QueryStageStateResponse = [active: string[], toProcess: string[]] | [active: string[], toProcess: string[], defs: Record<string, QueryDef>];
-export type RetrieveForProcessingSuccess = [
-  added: unknown,
-  // Identifies the retrieved generation of the queue item.
-  queueId: QueueId | null,
+export type RetrieveForProcessingSuccess = {
   active: QueryKeyHash[],
-  pending: number,
+  queueSize: number,
   def: QueryDef,
-  retrieved: true
-];
-export type RetrieveForProcessingFail = [
-  added: unknown,
-  // Null when no queue item was retrieved.
-  queueId: QueueId | null,
-  active: QueryKeyHash[],
-  pending: number,
-  def: null,
-  retrieved: false
-];
-export type RetrieveForProcessingResponse = RetrieveForProcessingSuccess | RetrieveForProcessingFail | null;
+};
 export type AddToQueueResponse = [
   added: number,
   queueId: QueueId | null,
@@ -106,7 +92,7 @@ export interface QueueDriverConnectionInterface {
   updateHeartBeat(hash: QueryKeyHash, queueId: QueueId | null): Promise<void>;
   // Atomically moves a queue item to active. Returns null when another node is already
   // processing the query or the concurrency budget is full.
-  retrieveForProcessing(hash: QueryKeyHash, queueId: QueueId): Promise<RetrieveForProcessingResponse>;
+  retrieveForProcessing(hash: QueryKeyHash, queueId: QueueId): Promise<RetrieveForProcessingSuccess | null>;
   optimisticQueryUpdate(hash: QueryKeyHash, toUpdate: unknown, queueId: QueueId): Promise<boolean>;
   cancelQuery(queryKey: QueryKey, queueId: QueueId | null): Promise<QueryDef | null>;
   getQueryAndRemove(hash: QueryKeyHash, queueId: QueueId | null): Promise<[QueryDef]>;

packages/cubejs-cubestore-driver/src/CubeStoreQueueDriver.ts

@@ -4,7 +4,6 @@ import {
   QueueDriverConnectionInterface,
   QueryStageStateResponse,
   QueryDef,
-  RetrieveForProcessingResponse,
   RetrieveForProcessingSuccess,
   QueueDriverOptions,
   AddToQueueQuery,
@@ -357,30 +356,20 @@ export class CubestoreQueueDriverConnection implements QueueDriverConnectionInte
       return null;
     }
 
-    return [
-      1,
-      row.id ? parseInt(row.id, 10) : null,
-      this.decodeActiveKeysFromRow(row.active),
-      parseInt(row.pending, 10),
-      this.decodeQueryDefFromRow(row as { payload: string, extra?: string | null }, method),
-      true
-    ];
+    return {
+      active: this.decodeActiveKeysFromRow(row.active),
+      queueSize: parseInt(row.pending, 10),
+      def: this.decodeQueryDefFromRow(row as { payload: string, extra?: string | null }, method),
+    };
   }
 
-  public async retrieveForProcessing(hash: QueryKeyHash, _queueId: QueueId): Promise<RetrieveForProcessingResponse> {
-    const rows = await this.driver.query<CubeStoreRetrieveResponse>('QUEUE RETRIEVE EXTENDED CONCURRENCY ? ?', [
+  public async retrieveForProcessing(hash: QueryKeyHash, _queueId: QueueId): Promise<RetrieveForProcessingSuccess | null> {
+    const rows = await this.driver.query<CubeStoreRetrieveResponse>('QUEUE RETRIEVE CONCURRENCY ? ?', [
       this.options.concurrency,
       this.prefixKey(hash),
     ]);
-    if (rows && rows.length) {
-      return this.decodeRetrievedFromRow(rows[0], 'retrieveForProcessing') || [
-        0,
-        null,
-        this.decodeActiveKeysFromRow(rows[0].active),
-        parseInt(rows[0].pending, 10),
-        null,
-        false
-      ];
+    if (rows.length) {
+      return this.decodeRetrievedFromRow(rows[0], 'retrieveForProcessing');
     }
 
     return null;

packages/cubejs-query-orchestrator/DEVELOPMENT.md

@@ -79,7 +79,8 @@ enum ResultStatus {
 
 `EXTENDED` on `QUEUE RETRIEVE` changes only the failure shape: without it a failed retrieval
 returns zero rows, with it a single row where `payload` and `id` are `NULL` but `pending`
-and `active` are filled. The driver always sends `EXTENDED`.
+and `active` are filled. The driver uses the non-extended form and treats zero rows as a failed
+retrieval.
 
 ## Enqueue and wait: `executeInQueue`
 
@@ -221,13 +222,13 @@ sequenceDiagram
     participant QueryOrchestrator
 
     QueryQueue->>QueueDriver: retrieveForProcessing
-    QueueDriver->>CubeStore: QUEUE RETRIEVE EXTENDED CONCURRENCY ?n ?path
+    QueueDriver->>CubeStore: QUEUE RETRIEVE CONCURRENCY ?n ?path
     CubeStore-->>QueueDriver: RetrieveResponse
-    QueueDriver-->>QueryQueue: [added, queueId, activeKeys, queueSize, def, retrieved]
+    QueueDriver-->>QueryQueue: { active, queueSize, def } | null
     Note over QueueDriver,CubeStore: The retrieval is atomic in Cube Store:<br/>only one node moves the item to active
 
-    alt def && added && activeKeys includes our key && retrieved
-        QueryQueue-)Background: sendProcessMessageFn(RetrievedQuery)
+    alt retrieved
+        QueryQueue-)Background: sendProcessMessageFn(queryKeyHash, queueId, retrieved)
         Note over QueryQueue,Background: Detached from here on: the hand-off returns,<br/>the execution keeps running
 
         Background->>QueueDriver: optimisticQueryUpdate
@@ -356,7 +357,53 @@ returns items highest priority first (oldest first within a priority) and reconc
 The fast track selects itself, so it is priority blind with nothing to compensate. That is
 what the `active + pending < concurrency` condition rules out: retrieving leaves a free slot
 for every item already pending, so no item is jumped over, and once the budget gets tight
-the fast track steps aside and lets reconcile pick by priority. Note that a retrieved item
-goes straight to active and never becomes pending, so a burst onto an idle queue still
-fast-tracks every query — items only start accumulating in pending once the concurrency
-budget is exhausted, which is exactly when the condition should stop firing.
+the fast track steps aside and lets reconcile pick by priority. A retrieved item goes straight
+to active and never becomes pending, so a burst onto an idle queue fast-tracks exactly one
+concurrency budget's worth and no more: items start accumulating in pending the moment the
+budget is exhausted, which is when the condition stops firing. S3 measures this exactly —
+50 retrievals out of 1000 at concurrency 50, 200 out of 1000 at concurrency 200 — so on a
+burst the saving scales as `concurrency / burst size`, not with the size of the burst.
+
+## Benchmarks
+
+The harness lives in `test/benchmarks/` and is compiled by the ordinary `yarn tsc` — jest never
+picks it up (`testMatch` is `*.test.ts`). Runs go through the suite runner:
+
+```bash
+yarn tsc
+yarn bench:suite --list                  # what the suites are
+yarn bench:suite S1 --dry-run            # the matrix with computed ρ and wall clock, run this first
+yarn bench:suite S1                      # off and on, one pass each
+yarn bench:suite --report .context/bench-results/S1-….jsonl
+```
+
+Every run emits one `BENCH_RESULT {json}` line plus a `BENCH_TICK {json}` per second; the runner
+collects both into `.context/bench-results/<suites>-<stamp>.jsonl` and prints a markdown summary.
+A single run can also be driven straight from env vars against
+`dist/test/benchmarks/QueueCubestore.bench.js` (or `QueueMemory.bench.js`) — see `readSettings`
+in `QueueBench.abstract.ts` for the full list.
+
+The one axis that decides everything is the load factor:
+
+```
+ρ = arrival_rate / capacity,   capacity = concurrency / handler_latency
+```
+
+The fast track saves a round trip only while the concurrency budget has a free slot, so ρ is
+what the suites sweep. `driverCalls.fastTrack.missRate` is the direct measure of the cost side:
+an `ADD_AND_RETRIEVE` that comes back without a retrieval is a round trip spent for nothing.
+Eligibility is read off the connection's `useFastTrack`, so a driver that cannot fast track
+never registers an attempt.
+
+Two things about the numbers are artifacts of the harness rather than production behaviour:
+
+- Workers poll `reconcileQueue` on a timer (`BENCH_WORKER_RECONCILE_MS`, default 50ms) because a
+  worker never submits and so has no submit-time reconcile to bootstrap from. Production
+  reconcile is event-driven. This poll dominates `getQueriesToCancel` / `getActiveAndToProcess`
+  and is the entire traffic of the idle-floor suite, which is why S5 measures two intervals.
+- Payload defaults are 5MB responses / 256KB query bodies, but the suites deliberately run at
+  64KB / 16KB. S7 owns the payload axis.
+
+`driverCalls` is snapshotted after drain and before the idle tail, so on a suite with
+`BENCH_IDLE_TAIL_MS` the tail's polling lands in `idle.driverCalls` and nowhere else — and
+`main` plus `workers` add up to `total` exactly.

packages/cubejs-query-orchestrator/package.json

@@ -21,6 +21,7 @@
     "unit": "jest --runInBand --forceExit --coverage --verbose test/unit",
     "integration": "jest --runInBand --verbose test/integration",
     "integration:cubestore": "jest --runInBand --verbose test/integration/cubestore",
+    "bench:suite": "node dist/test/benchmarks/run-suite.js",
     "lint": "eslint src/* test/* --ext .ts,.js",
     "lint:fix": "eslint --fix src/* test/* --ext .ts,.js"
   },
@@ -41,9 +42,11 @@
     "@types/jest": "^29",
     "@types/node": "^22",
     "@types/ramda": "^0.27.32",
+    "@types/yargs": "^17.0.31",
     "jest": "^29",
     "ts-jest": "^29",
-    "typescript": "~5.2.2"
+    "typescript": "~5.2.2",
+    "yargs": "^17.7.1"
   },
   "license": "Apache-2.0",
   "eslintConfig": {

packages/cubejs-query-orchestrator/src/orchestrator/LocalQueueDriverConnection.ts

@@ -11,7 +11,7 @@ import {
   QueryKeysTuple,
   GetActiveAndToProcessResponse,
   QueryStageStateResponse,
-  RetrieveForProcessingResponse,
+  RetrieveForProcessingSuccess,
   QueueDriverOptions,
   QueuePriority
 } from '@cubejs-backend/base-driver';
@@ -272,7 +272,7 @@ export class LocalQueueDriverConnection implements QueueDriverConnectionInterfac
     }
   }
 
-  public async retrieveForProcessing(queryKeyHash: QueryKeyHash, queueId: QueueId): Promise<RetrieveForProcessingResponse> {
+  public async retrieveForProcessing(queryKeyHash: QueryKeyHash, queueId: QueueId): Promise<RetrieveForProcessingSuccess | null> {
     const query = this.state.queryDef[queryKeyHash];
     const activeKeys = this.queueArray(this.state.active) as QueryKeyHash[];
 
@@ -283,29 +283,19 @@ export class LocalQueueDriverConnection implements QueueDriverConnectionInterfac
       this.state.active[queryKeyHash] ||
       activeKeys.length >= this.concurrency
     ) {
-      return [
-        0,
-        null,
-        activeKeys,
-        Object.keys(this.state.toProcess).length,
-        null,
-        false
-      ];
+      return null;
     }
 
     this.state.active[queryKeyHash] = { key: queryKeyHash, order: Number(queueId), queueId };
     delete this.state.toProcess[queryKeyHash];
 
     this.state.heartBeat[queryKeyHash] = { key: queryKeyHash, order: new Date().getTime(), queueId };
 
-    return [
-      1,
-      query.queueId,
-      this.queueArray(this.state.active) as QueryKeyHash[],
-      Object.keys(this.state.toProcess).length,
-      query,
-      true
-    ];
+    return {
+      active: this.queueArray(this.state.active) as QueryKeyHash[],
+      queueSize: Object.keys(this.state.toProcess).length,
+      def: query,
+    };
   }
 
   public async optimisticQueryUpdate(queryKeyHash: QueryKeyHash, toUpdate: any, queueId: QueueId): Promise<boolean> {