From 657f4e18ae75c69dd4db4057e14867f8a7d91247 Mon Sep 17 00:00:00 2001
From: waralexrom <108349432+waralexrom@users.noreply.github.com>
Date: Tue, 25 Aug 2026 17:25:12 +0200
Subject: [PATCH 1/3] fix(schema-compiler): a cube that extends another broke
 the parent's multi-stage measures (#11641)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* fix(schema-compiler): resolve inherited definitions per cube

`extends` hands the extending cube the very definitions of the cube it
extends: `allDefinitions` merges the parent's member objects by reference,
and a view's `default_filters` reach the extending view through the
prototype. References resolved per cube were written into those shared
objects, so the last cube prepared won and every other cube ended up
carrying member paths of a cube that is not its own.

Multi-stage members were hit through their nested `grain` and `filter`
objects — `prepareMembers` copies the member itself, but not what hangs off
it — which broke every multi-stage measure of the extended cube, including
queries that never mention the extending one. A parent's `grain.include:
[d]` came out as `["child.d"]`, and planning then either found no join path
to that cube or reported the grain dimension as unreachable, naming a
dimension the model already declares.

The same aliasing hit two more places: a view's default filter resolved to
a member of the view extending it, and a pre-aggregation's
`outputColumnTypes` names were scoped to the extending cube.

Copy each of them before writing the resolved references.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(tesseract): name the grain in the unreachable-dimension report

The report for a dimension a multi-stage member reads outside its grain
named the member, the dimension, and the declaration that fixes it — all of
which the model already says — so a grain holding something else was
indistinguishable from a grain the reader believed was right.

Spell out the grain the member is computed at. Granularities are named
separately (`orders.created_at (month)`) instead of being folded into the
symbol name, and the one case a listing cannot disambiguate — the grain
carrying the read dimension only under a granularity — gets a sentence of
its own.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../src/compiler/CubeEvaluator.ts             |  50 +++-
 .../unit/extends-shared-definitions.test.ts   | 255 ++++++++++++++++++
 .../multi_stage/multi_stage_query_planner.rs  |  89 +++++-
 .../common/integration_multi_stage.yaml       |  10 +
 .../integration/multi_stage/dimension_deps.rs |  30 +++
 5 files changed, 418 insertions(+), 16 deletions(-)
 create mode 100644 packages/cubejs-schema-compiler/test/unit/extends-shared-definitions.test.ts

diff --git a/packages/cubejs-schema-compiler/src/compiler/CubeEvaluator.ts b/packages/cubejs-schema-compiler/src/compiler/CubeEvaluator.ts
index 4b2fa35409886..2909d43bf8c2e 100644
--- a/packages/cubejs-schema-compiler/src/compiler/CubeEvaluator.ts
+++ b/packages/cubejs-schema-compiler/src/compiler/CubeEvaluator.ts
@@ -299,6 +299,12 @@ export class CubeEvaluator extends CubeSymbols {
       return `${cube.name}.${match.name}`;
     };
 
+    // A view extending another one inherits its `default_filters` entries by
+    // reference, so the resolved references have to go into a copy owned by this
+    // view. Written in place they would resolve to whichever view is prepared
+    // last, pointing the other views' filters at members they do not include.
+    cube.defaultFilters = (cube.defaultFilters as ViewDefaultValueFilter[]).map(f => ({ ...f }));
+
     for (const filter of cube.defaultFilters as ViewDefaultValueFilter[]) {
       const rawMember = this.evaluateReferences(cube.name, filter.member);
       const resolved = resolveViewMember('member', rawMember);
@@ -641,24 +647,33 @@ export class CubeEvaluator extends CubeSymbols {
               : {}),
           }));
         }
+        // `filter` and `grain` are nested objects, and a cube extending another
+        // one inherits them by reference, so the resolved references have to go
+        // into a copy owned by this cube. Written in place they would resolve to
+        // whichever cube is prepared last, pointing every member of the other
+        // cubes at that cube's dimensions.
         if (member.filter) {
-          if (typeof member.filter.exclude === 'function') {
-            member.filter.excludeReferences = this.evaluateReferences(cubeName, member.filter.exclude);
+          const filter = { ...member.filter };
+          if (typeof filter.exclude === 'function') {
+            filter.excludeReferences = this.evaluateReferences(cubeName, filter.exclude);
           }
-          if (typeof member.filter.keepOnly === 'function') {
-            member.filter.keepOnlyReferences = this.evaluateReferences(cubeName, member.filter.keepOnly);
+          if (typeof filter.keepOnly === 'function') {
+            filter.keepOnlyReferences = this.evaluateReferences(cubeName, filter.keepOnly);
           }
+          member.filter = filter;
         }
         if (member.grain) {
-          if (typeof member.grain.exclude === 'function') {
-            member.grain.excludeReferences = this.evaluateReferences(cubeName, member.grain.exclude);
+          const grain = { ...member.grain };
+          if (typeof grain.exclude === 'function') {
+            grain.excludeReferences = this.evaluateReferences(cubeName, grain.exclude);
           }
-          if (typeof member.grain.keepOnly === 'function') {
-            member.grain.keepOnlyReferences = this.evaluateReferences(cubeName, member.grain.keepOnly);
+          if (typeof grain.keepOnly === 'function') {
+            grain.keepOnlyReferences = this.evaluateReferences(cubeName, grain.keepOnly);
           }
-          if (typeof member.grain.include === 'function') {
-            member.grain.includeReferences = this.evaluateReferences(cubeName, member.grain.include);
+          if (typeof grain.include === 'function') {
+            grain.includeReferences = this.evaluateReferences(cubeName, grain.include);
           }
+          member.grain = grain;
         }
       }
     }
@@ -711,7 +726,7 @@ export class CubeEvaluator extends CubeSymbols {
   protected preparePreAggregations(cube: any, errorReporter: ErrorReporter) {
     if (cube.preAggregations) {
       // eslint-disable-next-line no-restricted-syntax
-      for (const preAggregation of Object.values(cube.preAggregations) as any) {
+      for (const [preAggregationName, preAggregation] of Object.entries(cube.preAggregations) as any) {
         // preAggregation is actually (PreAggregationDefinitionRollup | PreAggregationDefinitionOriginalSql)
         if (preAggregation.timeDimension) {
           preAggregation.timeDimensionReference = preAggregation.timeDimension;
@@ -767,10 +782,17 @@ export class CubeEvaluator extends CubeSymbols {
           delete preAggregation.buildRangeEnd;
         }
 
+        // `outputColumnTypes` names are resolved against this cube, and a cube
+        // extending another one inherits the pre-aggregation by reference, so
+        // both the entry and its columns have to be copies owned by this cube.
         if (preAggregation.outputColumnTypes) {
-          preAggregation.outputColumnTypes.forEach(column => {
-            column.name = this.evaluateReferences(cube.name, column.member, { originalSorting: true });
-          });
+          cube.preAggregations[preAggregationName] = {
+            ...preAggregation,
+            outputColumnTypes: preAggregation.outputColumnTypes.map(column => ({
+              ...column,
+              name: this.evaluateReferences(cube.name, column.member, { originalSorting: true }),
+            })),
+          };
         }
       }
     }
diff --git a/packages/cubejs-schema-compiler/test/unit/extends-shared-definitions.test.ts b/packages/cubejs-schema-compiler/test/unit/extends-shared-definitions.test.ts
new file mode 100644
index 0000000000000..0c239f617dd56
--- /dev/null
+++ b/packages/cubejs-schema-compiler/test/unit/extends-shared-definitions.test.ts
@@ -0,0 +1,255 @@
+import { PostgresQuery } from '../../src/adapter/PostgresQuery';
+import { prepareYamlCompiler } from './PrepareCompiler';
+
+// `extends` hands the extending cube the very definitions of the cube it extends,
+// so every reference resolved per cube — a multi-stage `grain:`/`filter:`, a view
+// default filter, a pre-aggregation's output column names — has to be written into
+// an object owned by that cube. Written into the shared one it resolves to whichever
+// cube is prepared last, and the other cubes end up carrying member paths of a cube
+// that is not theirs.
+describe('Multi-stage members of a cube that another cube extends', () => {
+  const baseFact = `
+  - name: base_fact
+    sql: "SELECT 1 AS id, 1 AS dim_id, '2026-01-01'::date AS d, 10 AS v"
+    joins:
+      - name: dims
+        sql: "{CUBE}.dim_id = {dims}.id"
+        relationship: many_to_one
+    dimensions:
+      - name: id
+        sql: "{CUBE}.id"
+        type: number
+        primary_key: true
+      - name: d
+        sql: "{CUBE}.d"
+        type: time
+      - name: v
+        sql: "{CUBE}.v"
+        type: number
+    measures:
+      - name: v_sum
+        sql: "{v}"
+        type: sum
+      - name: daily_v
+        multi_stage: true
+        sql: "{v_sum}"
+        type: number
+      - name: linked
+        multi_stage: true
+        sql: "{daily_v}"
+        type: sum
+        grain:
+          include:
+            - d
+      - name: combining
+        multi_stage: true
+        sql: "CASE WHEN {d} IS NOT NULL THEN {daily_v} ELSE 0 END"
+        type: max
+        grain:
+          include:
+            - d
+      - name: outer_combining
+        multi_stage: true
+        sql: "{combining}"
+        type: number
+      - name: v_sum_all_dates
+        multi_stage: true
+        sql: "{v_sum}"
+        type: number
+        filter:
+          exclude:
+            - d
+`;
+
+  const dims = `
+  - name: dims
+    sql: "SELECT 1 AS id, 'a' AS name"
+    dimensions:
+      - name: id
+        sql: "{CUBE}.id"
+        type: number
+        primary_key: true
+      - name: name
+        sql: "{CUBE}.name"
+        type: string
+`;
+
+  const childFact = `
+  - name: child_fact
+    extends: base_fact
+    sql: "SELECT 1 AS id, 1 AS dim_id, '2026-01-01'::date AS d, 10 AS v, 'x' AS tag"
+    dimensions:
+      - name: tag
+        sql: "{CUBE}.tag"
+        type: string
+`;
+
+  const model = (withChild: boolean) => `cubes:${dims}${baseFact}${withChild ? childFact : ''}`;
+
+  const compile = async (withChild: boolean) => {
+    const compilers = prepareYamlCompiler(model(withChild));
+    await compilers.compiler.compile();
+    return compilers;
+  };
+
+  const buildSql = async (withChild: boolean, query: any, useNativeSqlPlanner: boolean) => {
+    const compilers = await compile(withChild);
+    return new PostgresQuery(compilers, {
+      timezone: 'UTC',
+      useNativeSqlPlanner,
+      ...query,
+    }).buildSqlAndParams();
+  };
+
+  it('resolves grain references against the cube that declares the member', async () => {
+    const { cubeEvaluator } = await compile(true);
+
+    expect(cubeEvaluator.evaluatedCubes.base_fact.measures.linked.grain?.includeReferences)
+      .toEqual(['base_fact.d']);
+    expect(cubeEvaluator.evaluatedCubes.child_fact.measures.linked.grain?.includeReferences)
+      .toEqual(['child_fact.d']);
+  });
+
+  it('resolves filter references against the cube that declares the member', async () => {
+    const { cubeEvaluator } = await compile(true);
+
+    expect(cubeEvaluator.evaluatedCubes.base_fact.measures.v_sum_all_dates.filter?.excludeReferences)
+      .toEqual(['base_fact.d']);
+    expect(cubeEvaluator.evaluatedCubes.child_fact.measures.v_sum_all_dates.filter?.excludeReferences)
+      .toEqual(['child_fact.d']);
+  });
+
+  describe.each([
+    ['native', true],
+    ['legacy', false],
+  ])('%s planner', (_name, useNativeSqlPlanner) => {
+    // Planning the parent's member must not depend on the extending cube being
+    // there at all, so the plan is compared against the same model without it.
+    const expectSamePlanWithAndWithoutChild = async (query: any) => {
+      const [withoutChild] = await buildSql(false, query, useNativeSqlPlanner);
+      const [withChild] = await buildSql(true, query, useNativeSqlPlanner);
+      expect(withChild).toEqual(withoutChild);
+    };
+
+    it('plans a multi-stage measure with an explicit grain', async () => {
+      await expectSamePlanWithAndWithoutChild({ measures: ['base_fact.linked'] });
+      await expect(buildSql(true, { measures: ['child_fact.linked'] }, useNativeSqlPlanner)).resolves.toBeDefined();
+    });
+
+    it('plans a chained multi-stage measure reading a grain dimension', async () => {
+      await expectSamePlanWithAndWithoutChild({ measures: ['base_fact.outer_combining'] });
+      await expectSamePlanWithAndWithoutChild({
+        measures: ['base_fact.outer_combining'],
+        dimensions: ['dims.name'],
+      });
+    });
+
+    it('plans a multi-stage measure with a filter directive', async () => {
+      await expectSamePlanWithAndWithoutChild({
+        measures: ['base_fact.v_sum_all_dates'],
+        timeDimensions: [{
+          dimension: 'base_fact.d',
+          granularity: 'month',
+          dateRange: ['2026-01-01', '2026-01-31'],
+        }],
+      });
+    });
+
+    it('plans a plain measure of a cube that another cube extends', async () => {
+      await expect(buildSql(true, { measures: ['base_fact.v_sum'] }, useNativeSqlPlanner)).resolves.toBeDefined();
+    });
+  });
+});
+
+describe('View default filters of a view that another view extends', () => {
+  const model = `
+cubes:
+  - name: orders
+    sql: "SELECT 1 AS id, 'usd' AS currency"
+    dimensions:
+      - name: id
+        sql: "{CUBE}.id"
+        type: number
+        primary_key: true
+      - name: currency
+        sql: "{CUBE}.currency"
+        type: string
+    measures:
+      - name: count
+        type: count
+
+views:
+  - name: base_view
+    cubes:
+      - join_path: orders
+        includes:
+          - currency
+          - count
+    default_filters:
+      - member: currency
+        operator: equals
+        values: ["usd"]
+
+  - name: child_view
+    extends: base_view
+`;
+
+  it('resolves the filter member against the view that declares the filter', async () => {
+    const { compiler, cubeEvaluator } = prepareYamlCompiler(model);
+    await compiler.compile();
+
+    expect(cubeEvaluator.evaluatedCubes.base_view.defaultFilters?.map(f => f.memberReference))
+      .toEqual(['base_view.currency']);
+    expect(cubeEvaluator.evaluatedCubes.child_view.defaultFilters?.map(f => f.memberReference))
+      .toEqual(['child_view.currency']);
+  });
+});
+
+describe('Pre-aggregations of a cube that another cube extends', () => {
+  const model = `
+cubes:
+  - name: base
+    sql: "SELECT 1 AS id, '2026-01-01'::date AS d"
+    dimensions:
+      - name: id
+        sql: "{CUBE}.id"
+        type: number
+        primary_key: true
+      - name: d
+        sql: "{CUBE}.d"
+        type: time
+    measures:
+      - name: count
+        type: count
+    pre_aggregations:
+      - name: main
+        dimensions:
+          - id
+        measures:
+          - count
+        time_dimension: d
+        granularity: day
+        output_column_types:
+          - member: id
+            type: integer
+
+  - name: child
+    extends: base
+    sql: "SELECT 1 AS id, '2026-01-01'::date AS d, 'x' AS tag"
+    dimensions:
+      - name: tag
+        sql: "{CUBE}.tag"
+        type: string
+`;
+
+  it('resolves output column names against the cube that declares the pre-aggregation', async () => {
+    const { compiler, cubeEvaluator } = prepareYamlCompiler(model);
+    await compiler.compile();
+
+    const names = (cube: string) => (cubeEvaluator.evaluatedCubes[cube].preAggregations.main as any)
+      .outputColumnTypes.map((c: any) => c.name);
+
+    expect(names('base')).toEqual(['base.id']);
+    expect(names('child')).toEqual(['child.id']);
+  });
+});
diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/planners/multi_stage/multi_stage_query_planner.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/planners/multi_stage/multi_stage_query_planner.rs
index 845bbe7cba12b..97f9b488d943c 100644
--- a/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/planners/multi_stage/multi_stage_query_planner.rs
+++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/planners/multi_stage/multi_stage_query_planner.rs
@@ -414,15 +414,100 @@ impl MultiStageQueryPlanner {
         }) {
             return Ok(());
         }
+        let grain = Self::grain_members(grain_state, parent_state);
+        // A granularity of the very dimension the sql reads looks like a match in
+        // the listing, so the one case where reading the grain is not enough to
+        // tell them apart is spelled out.
+        let target = dimension.clone().resolve_reference_chain().full_name();
+        let hint = if grain
+            .iter()
+            .any(|m| Self::granular_time_dimension_base(m).as_ref() == Some(&target))
+        {
+            format!(
+                " The grain carries {target} at a granularity, which is a value of its own and not \
+                 the dimension itself."
+            )
+        } else {
+            String::new()
+        };
+        let grain = if grain.is_empty() {
+            "no dimensions".to_string()
+        } else {
+            grain
+                .iter()
+                .map(Self::describe_grain_member)
+                .collect::<Vec<_>>()
+                .join(", ")
+        };
         Err(CubeError::user(format!(
             "Multi-stage member {member} reads dimension {dimension}, which is not part of the \
-             grain it is computed at. Add {dimension} to `grain.include` of {member}, or remove \
-             it from the member's sql.",
+             grain it is computed at ({grain}).{hint} Add {dimension} to `grain.include` of \
+             {member}, or remove it from the member's sql.",
             member = member.full_name(),
             dimension = dimension.full_name(),
         )))
     }
 
+    // The grain the member is computed at: the stage's own dimensions and, when
+    // the assembly broadcasts back onto the query grid, the keys side.
+    fn grain_members(
+        grain_state: &QueryProperties,
+        parent_state: &QueryProperties,
+    ) -> Vec<Rc<MemberSymbol>> {
+        let mut members: Vec<Rc<MemberSymbol>> = Vec::new();
+        for state in [grain_state, parent_state] {
+            for dimension in state
+                .dimensions()
+                .iter()
+                .chain(state.time_dimensions().iter())
+            {
+                let resolved = dimension.clone().resolve_reference_chain();
+                if !members
+                    .iter()
+                    .any(|m| m.full_name() == resolved.full_name())
+                {
+                    members.push(resolved);
+                }
+            }
+        }
+        members
+    }
+
+    // A time dimension carries its granularity inside its name, which reads as a
+    // member of its own; the granularity is named separately instead.
+    fn describe_grain_member(member: &Rc<MemberSymbol>) -> String {
+        match member.as_ref() {
+            MemberSymbol::TimeDimension(time_dimension) => match time_dimension.granularity() {
+                Some(granularity) => format!(
+                    "{} ({})",
+                    time_dimension
+                        .base_symbol()
+                        .clone()
+                        .resolve_reference_chain()
+                        .full_name(),
+                    granularity
+                ),
+                None => member.full_name(),
+            },
+            _ => member.full_name(),
+        }
+    }
+
+    fn granular_time_dimension_base(member: &Rc<MemberSymbol>) -> Option<String> {
+        match member.as_ref() {
+            MemberSymbol::TimeDimension(time_dimension) => {
+                time_dimension.granularity().as_ref().map(|_| {
+                    time_dimension
+                        .base_symbol()
+                        .clone()
+                        .resolve_reference_chain()
+                        .full_name()
+                })
+            }
+            _ => None,
+        }
+    }
+
     /// Plans CASE-SWITCH dependencies: collects, per dependency, the
     /// union of switch values it covers and renders each dependency
     /// under a state with an equality filter on the switch member
diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/test_fixtures/schemas/yaml_files/common/integration_multi_stage.yaml b/rust/cube/cubesqlplanner/cubesqlplanner/src/test_fixtures/schemas/yaml_files/common/integration_multi_stage.yaml
index 43f33edd26866..86cd98c2bfddb 100644
--- a/rust/cube/cubesqlplanner/cubesqlplanner/src/test_fixtures/schemas/yaml_files/common/integration_multi_stage.yaml
+++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/test_fixtures/schemas/yaml_files/common/integration_multi_stage.yaml
@@ -414,6 +414,16 @@ cubes:
                 include:
                     - orders.created_at
 
+          # The declared grain is a dimension of another cube, so it does not
+          # supply the dimension the sql reads.
+          - name: amount_first_half_of_month_grain_of_other_cube
+            type: sum
+            sql: "CASE WHEN EXTRACT(DAY FROM {CUBE.created_at}) <= 15 THEN {CUBE.total_amount} ELSE 0 END"
+            multi_stage: true
+            grain:
+                include:
+                    - customers.city
+
           # Undeclared read one stage further out: the dimension is read by a
           # measure whose aggregated dependency is itself multi-stage.
           - name: amount_prev_month_first_half
diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/multi_stage/dimension_deps.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/multi_stage/dimension_deps.rs
index c5383476b0e58..aa8f17d2b6796 100644
--- a/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/multi_stage/dimension_deps.rs
+++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/multi_stage/dimension_deps.rs
@@ -58,6 +58,36 @@ async fn test_undeclared_time_dimension_read_is_reported() {
     );
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn test_report_spells_out_the_grain_the_member_is_computed_at() {
+    let message = expect_error("amount_first_half_of_month");
+
+    assert!(
+        message.contains("orders.created_at (month)"),
+        "The error must spell out the grain the member is computed at:\n{}",
+        message
+    );
+    assert!(
+        message.contains("at a granularity"),
+        "A granularity of the dimension the sql reads must not read as a match:\n{}",
+        message
+    );
+}
+
+/// A grain declared at another cube's dimension: naming the read dimension
+/// alone would repeat what the model already says, so the grain the member is
+/// actually computed at is what tells the two apart.
+#[tokio::test(flavor = "multi_thread")]
+async fn test_report_names_a_grain_declared_at_another_cube() {
+    let message = expect_error("amount_first_half_of_month_grain_of_other_cube");
+
+    assert!(
+        message.contains("orders.created_at") && message.contains("customers.city"),
+        "The error must name both the dimension the sql reads and the declared grain:\n{}",
+        message
+    );
+}
+
 /// The reading member consumes a time-shifted multi-stage measure, so its own
 /// grain is settled one stage above the leaf that would have to carry the
 /// dimension.

From cfaa5f8e3973b72954bc36e244f20b1d19f93a9e Mon Sep 17 00:00:00 2001
From: Dmitry Patsura <talk@dmtry.me>
Date: Tue, 25 Aug 2026 17:50:07 +0200
Subject: [PATCH 2/3] refactor(query-orchestrator): Simplify retrieval result
 (#11640)

---
 .../configuration/environment-variables.mdx   |  7 +-
 .../src/queue-driver.interface.ts             | 22 +----
 .../src/CubeStoreQueueDriver.ts               | 29 ++-----
 .../cubejs-query-orchestrator/DEVELOPMENT.md  | 11 +--
 .../LocalQueueDriverConnection.ts             | 26 ++----
 .../src/orchestrator/QueryQueue.ts            | 86 +++++++------------
 .../test/unit/QueryQueue.abstract.ts          | 66 ++++++++------
 7 files changed, 101 insertions(+), 146 deletions(-)

diff --git a/docs-mintlify/reference/configuration/environment-variables.mdx b/docs-mintlify/reference/configuration/environment-variables.mdx
index d5f60fa323368..b7b2053a04d0c 100644
--- a/docs-mintlify/reference/configuration/environment-variables.mdx
+++ b/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 |
 | --------------- | ---------------------- | --------------------- |
diff --git a/packages/cubejs-base-driver/src/queue-driver.interface.ts b/packages/cubejs-base-driver/src/queue-driver.interface.ts
index 5980031cd59b2..0d4b2c7efd01a 100644
--- a/packages/cubejs-base-driver/src/queue-driver.interface.ts
+++ b/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]>;
diff --git a/packages/cubejs-cubestore-driver/src/CubeStoreQueueDriver.ts b/packages/cubejs-cubestore-driver/src/CubeStoreQueueDriver.ts
index 95563eb392041..5067b71a82865 100644
--- a/packages/cubejs-cubestore-driver/src/CubeStoreQueueDriver.ts
+++ b/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;
diff --git a/packages/cubejs-query-orchestrator/DEVELOPMENT.md b/packages/cubejs-query-orchestrator/DEVELOPMENT.md
index f51d405843f47..6f58e7e518219 100644
--- a/packages/cubejs-query-orchestrator/DEVELOPMENT.md
+++ b/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
diff --git a/packages/cubejs-query-orchestrator/src/orchestrator/LocalQueueDriverConnection.ts b/packages/cubejs-query-orchestrator/src/orchestrator/LocalQueueDriverConnection.ts
index 55d4211440364..9ac867ae02c78 100644
--- a/packages/cubejs-query-orchestrator/src/orchestrator/LocalQueueDriverConnection.ts
+++ b/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,14 +283,7 @@ 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 };
@@ -298,14 +291,11 @@ export class LocalQueueDriverConnection implements QueueDriverConnectionInterfac
 
     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> {
diff --git a/packages/cubejs-query-orchestrator/src/orchestrator/QueryQueue.ts b/packages/cubejs-query-orchestrator/src/orchestrator/QueryQueue.ts
index d36558735f70c..8ad5b2f4f81d1 100644
--- a/packages/cubejs-query-orchestrator/src/orchestrator/QueryQueue.ts
+++ b/packages/cubejs-query-orchestrator/src/orchestrator/QueryQueue.ts
@@ -25,14 +25,11 @@ export type QueryHandlerFn = (query: QueryDef, cancelHandler: CancelHandlerFn) =
 export type StreamHandlerFn = (query: QueryDef, stream: QueryStream) => Promise<unknown>;
 export type QueryHandlersMap = Record<string, QueryHandlerFn>;
 
-export type RetrievedQuery = {
-  queryKeyHash: QueryKeyHash;
-  queueId: QueueId;
-  queueSize: number;
-  query: QueryDef;
-};
-
-export type SendProcessMessageFn = (retrieved: RetrievedQuery) => Promise<void> | void;
+export type SendProcessMessageFn = (
+  queryKeyHash: QueryKeyHash,
+  queueId: QueueId,
+  retrieved: RetrieveForProcessingSuccess
+) => Promise<void> | void;
 export type SendCancelMessageFn = (query: QueryDef, queueId: QueueId | null) => Promise<void> | void;
 
 export type ExecuteInQueueOptions = Omit<AddToQueueOptions, 'queueId'> & {
@@ -131,7 +128,9 @@ export class QueryQueue {
     this.orphanedTimeout = options.orphanedTimeout || 120;
     this.heartBeatInterval = options.heartBeatInterval || 30;
 
-    this.sendProcessMessageFn = options.sendProcessMessageFn || ((retrieved) => { this.executeQuery(retrieved); });
+    this.sendProcessMessageFn = options.sendProcessMessageFn || (
+      (queryKeyHash, queueId, retrieved) => { this.executeQuery(queryKeyHash, queueId, retrieved); }
+    );
     this.sendCancelMessageFn = options.sendCancelMessageFn || ((query, queueId) => { this.processCancel(query, queueId); });
     this.queryHandlers = options.queryHandlers;
     this.streamHandler = options.streamHandler;
@@ -307,13 +306,13 @@ export class QueryQueue {
 
       if (retrieved) {
         // The item is active already, there is nothing for reconcile to pick up
-        await this.dispatchQuery(this.retrievedQuery(queryKeyHash, queueId, retrieved));
+        await this.dispatchQuery(queryKeyHash, queueId, retrieved);
       } else {
         await this.reconcileQueue();
       }
 
       if (!added) {
-        const queryDef = retrieved ? retrieved[4] : await queueConnection.getQueryDef(queryKeyHash, queueId);
+        const queryDef = retrieved ? retrieved.def : await queueConnection.getQueryDef(queryKeyHash, queueId);
         if (queryDef) {
           waitingContext = {
             queueId,
@@ -328,7 +327,7 @@ export class QueryQueue {
 
       // A retrieval carries the active keys of its prefix, and a retrieved query is never pending,
       // so it has no place in the queue to report
-      const [active, toProcess] = retrieved ? [retrieved[2], undefined] : await queueConnection.getQueryStageState(true);
+      const [active, toProcess] = retrieved ? [retrieved.active, undefined] : await queueConnection.getQueryStageState(true);
 
       this.logger('Waiting for query', {
         ...waitingContext,
@@ -365,17 +364,6 @@ export class QueryQueue {
     }
   }
 
-  protected retrievedQuery(queryKeyHash: QueryKeyHash, queueId: QueueId, retrieved: RetrieveForProcessingSuccess): RetrievedQuery {
-    const [, , , queueSize, query] = retrieved;
-
-    return {
-      queryKeyHash,
-      queueId,
-      queueSize,
-      query,
-    };
-  }
-
   /**
    * `dispose` releases the listener and the timer, it's a no-op once the promise resolved.
    */
@@ -796,17 +784,21 @@ export class QueryQueue {
       return;
     }
 
-    await this.dispatchQuery(retrieved);
+    await this.dispatchQuery(queryKeyHashed, queueId, retrieved);
   }
 
-  protected async dispatchQuery(retrieved: RetrievedQuery): Promise<void> {
+  protected async dispatchQuery(
+    queryKeyHash: QueryKeyHash,
+    queueId: QueueId,
+    retrieved: RetrieveForProcessingSuccess
+  ): Promise<void> {
     try {
-      await this.sendProcessMessageFn(retrieved);
+      await this.sendProcessMessageFn(queryKeyHash, queueId, retrieved);
     } catch (e: any) {
       this.logger('Error while processing message', {
-        queueId: retrieved.queueId,
-        queryKey: retrieved.query.queryKey,
-        requestId: retrieved.query.requestId,
+        queueId,
+        queryKey: retrieved.def.queryKey,
+        requestId: retrieved.def.requestId,
         error: (e.stack || e).toString(),
         queuePrefix: this.redisQueuePrefix
       });
@@ -817,27 +809,16 @@ export class QueryQueue {
    * Atomically moves the query specified by `queryKeyHashed` to the active set. Returns `null`
    * when another node is already running the query or the concurrency budget is full.
    */
-  protected async retrieveQueryForProcessing(queryKeyHashed: QueryKeyHash, queueId: QueueId): Promise<RetrievedQuery | null> {
+  protected async retrieveQueryForProcessing(queryKeyHashed: QueryKeyHash, queueId: QueueId): Promise<RetrieveForProcessingSuccess | null> {
     const queueConnection = await this.queueDriver.createConnection();
 
-    let insertedCount;
-    let activeKeys;
-    let queueSize;
     let query;
-    let retrievalSucceeded;
 
     try {
-      const retrieveResult = await queueConnection.retrieveForProcessing(queryKeyHashed, queueId);
-      if (retrieveResult) {
-        [insertedCount, , activeKeys, queueSize, query, retrievalSucceeded] = retrieveResult;
-      }
-
-      const activated = activeKeys && activeKeys.indexOf(queryKeyHashed) !== -1;
-      if (!query) {
+      const retrieved = await queueConnection.retrieveForProcessing(queryKeyHashed, queueId);
+      if (!retrieved) {
         query = await queueConnection.getQueryDef(queryKeyHashed, null);
-      }
 
-      if (!query || !insertedCount || !activated || !retrievalSucceeded) {
         // TODO Ideally streaming queries should reconcile queue here after waiting on open slot however in practice continue wait timeout reconciles faster CPU-wise
         // if (query?.queryHandler === 'stream') {
         //   const [active] = await queueConnection.getQueryStageState(true);
@@ -852,22 +833,13 @@ export class QueryQueue {
           queryKey: query && query.queryKey || queryKeyHashed,
           requestId: query && query.requestId,
           queuePrefix: this.redisQueuePrefix,
-          retrievalSucceeded,
           query,
-          insertedCount,
-          activeKeys,
-          activated,
           queryExists: !!query
         });
         return null;
       }
 
-      return {
-        queryKeyHash: queryKeyHashed,
-        queueId,
-        queueSize,
-        query,
-      };
+      return retrieved;
     } catch (e: any) {
       this.logger('Queue storage error', {
         queueId,
@@ -888,8 +860,12 @@ export class QueryQueue {
    * the result. It's the counterpart of `sendProcessMessageFn` and the entry point for a custom
    * implementation which hands the query over to another process.
    */
-  public async executeQuery(retrieved: RetrievedQuery): Promise<void> {
-    const { queryKeyHash: queryKeyHashed, queueId, queueSize, query } = retrieved;
+  public async executeQuery(
+    queryKeyHashed: QueryKeyHash,
+    queueId: QueueId,
+    retrieved: RetrieveForProcessingSuccess
+  ): Promise<void> {
+    const { queueSize, def: query } = retrieved;
 
     const queueConnection = await this.queueDriver.createConnection();
 
diff --git a/packages/cubejs-query-orchestrator/test/unit/QueryQueue.abstract.ts b/packages/cubejs-query-orchestrator/test/unit/QueryQueue.abstract.ts
index 2dc205f4c1be7..72814fd4b71d2 100644
--- a/packages/cubejs-query-orchestrator/test/unit/QueryQueue.abstract.ts
+++ b/packages/cubejs-query-orchestrator/test/unit/QueryQueue.abstract.ts
@@ -71,9 +71,9 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) =>
           readable.pipe(stream);
         });
       },
-      sendProcessMessageFn: async (retrieved) => {
+      sendProcessMessageFn: async (queryKeyHash, queueId, retrieved) => {
         streamCallOrder.push('dispatch');
-        processMessagePromises.push(queue.executeQuery(retrieved));
+        processMessagePromises.push(queue.executeQuery(queryKeyHash, queueId, retrieved));
       },
       sendCancelMessageFn: async (query) => {
         processCancelPromises.push(queue.processCancel.bind(queue)(query));
@@ -479,10 +479,14 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) =>
         );
 
         const firstRetrieval = await connection.retrieveForProcessing(key, queueId);
-        expect(firstRetrieval?.[5]).toBe(true);
+        expect(firstRetrieval).toMatchObject({
+          active: [key],
+          queueSize: 0,
+          def: { queryKey: key },
+        });
 
         const secondRetrieval = await connection2.retrieveForProcessing(key, queueId);
-        expect(secondRetrieval).toStrictEqual([0, null, [key], 0, null, false]);
+        expect(secondRetrieval).toBeNull();
       } finally {
         await connection.getQueryAndRemove(key, null);
         queue.queueDriver.release(connection);
@@ -490,12 +494,14 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) =>
       }
     });
 
-    onlyLocalTest('a failed retrieval does not reserve a pending query', async () => {
+    test('a failed retrieval does not reserve a pending query', async () => {
       const connection = await queue.queueDriver.createConnection();
       const connection2 = await queue.queueDriver.createConnection();
       const priority = 10;
-      const firstKey = 'concurrency-first' as any;
-      const secondKey = 'concurrency-second' as any;
+      const firstKey: QueryKey = 'concurrency-first';
+      const secondKey: QueryKey = 'concurrency-second';
+      const firstHash = connection.redisHash(firstKey);
+      const secondHash = connection.redisHash(secondKey);
 
       try {
         const [, firstQueueId] = await connection.addToQueue(
@@ -509,20 +515,25 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) =>
           }
         );
 
-        expect((await connection.retrieveForProcessing(firstKey, firstQueueId))?.[5]).toBe(true);
-        expect(await connection2.retrieveForProcessing(secondKey, secondQueueId)).toStrictEqual([
-          0, null, [firstKey], 1, null, false
-        ]);
-        expect(await connection.getToProcessQueries()).toStrictEqual([[secondKey, secondQueueId]]);
+        expect(await connection.retrieveForProcessing(firstHash, firstQueueId)).toMatchObject({
+          active: [firstHash],
+          queueSize: 1,
+          def: { queryKey: firstKey },
+        });
+        expect(await connection2.retrieveForProcessing(secondHash, secondQueueId)).toBeNull();
+        expect(await connection.getToProcessQueries()).toStrictEqual([[secondHash, secondQueueId]]);
 
-        await connection.getQueryAndRemove(firstKey, firstQueueId);
+        await connection.getQueryAndRemove(firstHash, firstQueueId);
 
-        const secondRetrieval = await connection2.retrieveForProcessing(secondKey, secondQueueId);
-        expect(secondRetrieval?.[0]).toBe(1);
-        expect(secondRetrieval?.[5]).toBe(true);
+        const secondRetrieval = await connection2.retrieveForProcessing(secondHash, secondQueueId);
+        expect(secondRetrieval).toMatchObject({
+          active: [secondHash],
+          queueSize: 0,
+          def: { queryKey: secondKey },
+        });
       } finally {
-        await connection.getQueryAndRemove(firstKey, null);
-        await connection.getQueryAndRemove(secondKey, null);
+        await connection.getQueryAndRemove(firstHash, null);
+        await connection.getQueryAndRemove(secondHash, null);
         queue.queueDriver.release(connection);
         queue.queueDriver.release(connection2);
       }
@@ -540,7 +551,11 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) =>
             queueId: queue.generateQueueId(), stageQueryKey: key, requestId: '1'
           }
         );
-        expect((await connection.retrieveForProcessing(key, staleQueueId))?.[5]).toBe(true);
+        expect(await connection.retrieveForProcessing(key, staleQueueId)).toMatchObject({
+          active: [key],
+          queueSize: 0,
+          def: { queryKey },
+        });
         await connection.getQueryAndRemove(key, staleQueueId);
 
         const [, currentQueueId] = await connection.addToQueue(
@@ -550,12 +565,14 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) =>
         );
 
         if (options.cacheAndQueueDriver !== 'cubestore') {
-          expect(await connection.retrieveForProcessing(key, staleQueueId)).toStrictEqual([
-            0, null, [], 1, null, false
-          ]);
+          expect(await connection.retrieveForProcessing(key, staleQueueId)).toBeNull();
           expect(await connection.getToProcessQueries()).toStrictEqual([[key, currentQueueId]]);
         }
-        expect((await connection.retrieveForProcessing(key, currentQueueId))?.[5]).toBe(true);
+        expect(await connection.retrieveForProcessing(key, currentQueueId)).toMatchObject({
+          active: [key],
+          queueSize: 0,
+          def: { queryKey, query: ['new'] },
+        });
 
         const staleUpdateResult = await connection.optimisticQueryUpdate(key, { stale: true }, staleQueueId);
         if (options.cacheAndQueueDriver !== 'cubestore') {
@@ -756,8 +773,7 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) =>
           // concurrency is 1, the first query takes the only slot
           const [added1, , , , retrieved1] = await addToQueue(first, 1);
           expect(added1).toBe(1);
-          expect(retrieved1?.[5]).toBe(true);
-          expect(retrieved1?.[4].queryKey).toStrictEqual(first);
+          expect(retrieved1?.def.queryKey).toStrictEqual(first);
 
           // an active item is never retrieved twice
           const [added1again, , , , retrieved1again] = await addToQueue(first, 1);

From cfffdfe152c46c9cc3bcf696666be3b568e437ff Mon Sep 17 00:00:00 2001
From: Dmitry Patsura <talk@dmtry.me>
Date: Tue, 25 Aug 2026 20:14:57 +0200
Subject: [PATCH 3/3] chore(query-orchestrator): Queue - improve benchmarks
 (#11642)

---
 .../cubejs-query-orchestrator/DEVELOPMENT.md  |  54 +-
 .../cubejs-query-orchestrator/package.json    |   5 +-
 .../test/benchmarks/QueueBench.abstract.ts    | 784 ++++++++++++------
 .../test/benchmarks/QueueBenchWorker.ts       |  96 +--
 .../test/benchmarks/QueueCubestore.bench.ts   |   2 +-
 .../test/benchmarks/QueueMemory.bench.ts      |   2 +-
 .../test/benchmarks/instrument.ts             | 263 ++++++
 .../test/benchmarks/protocol.ts               |  36 +
 .../test/benchmarks/run-suite.ts              | 487 +++++++++++
 .../test/benchmarks/suites.ts                 | 258 ++++++
 yarn.lock                                     |   2 +-
 11 files changed, 1689 insertions(+), 300 deletions(-)
 create mode 100644 packages/cubejs-query-orchestrator/test/benchmarks/instrument.ts
 create mode 100644 packages/cubejs-query-orchestrator/test/benchmarks/protocol.ts
 create mode 100644 packages/cubejs-query-orchestrator/test/benchmarks/run-suite.ts
 create mode 100644 packages/cubejs-query-orchestrator/test/benchmarks/suites.ts

diff --git a/packages/cubejs-query-orchestrator/DEVELOPMENT.md b/packages/cubejs-query-orchestrator/DEVELOPMENT.md
index 6f58e7e518219..243a1aaafde3b 100644
--- a/packages/cubejs-query-orchestrator/DEVELOPMENT.md
+++ b/packages/cubejs-query-orchestrator/DEVELOPMENT.md
@@ -357,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.
diff --git a/packages/cubejs-query-orchestrator/package.json b/packages/cubejs-query-orchestrator/package.json
index 223f668c8da16..35f95486d71ce 100644
--- a/packages/cubejs-query-orchestrator/package.json
+++ b/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": {
diff --git a/packages/cubejs-query-orchestrator/test/benchmarks/QueueBench.abstract.ts b/packages/cubejs-query-orchestrator/test/benchmarks/QueueBench.abstract.ts
index f558043c5e217..80d41018d7353 100644
--- a/packages/cubejs-query-orchestrator/test/benchmarks/QueueBench.abstract.ts
+++ b/packages/cubejs-query-orchestrator/test/benchmarks/QueueBench.abstract.ts
@@ -1,10 +1,21 @@
-import { CubeStoreQueueDriver } from '@cubejs-backend/cubestore-driver';
 import crypto from 'crypto';
 import path from 'path';
 import { ChildProcess, fork } from 'child_process';
-import { createPromiseLock, MethodName, pausePromise } from '@cubejs-backend/shared';
-import { QueueDriverConnectionInterface, QueueDriverInterface, QueuePriority } from '@cubejs-backend/base-driver';
-import { LocalQueueDriver, QueryQueue, QueryQueueOptions } from '../../src';
+import { createPromiseLock, pausePromise } from '@cubejs-backend/shared';
+import { QueuePriority } from '@cubejs-backend/base-driver';
+import { ContinueWaitError, QueryQueueOptions, TimeoutError } from '../../src';
+import {
+  BenchCounters,
+  cloneMethods,
+  createBenchQueue,
+  createCounters,
+  driverCallsTotal,
+  MethodCounter,
+  mergeEvents,
+  mergeMethods,
+  percentiles,
+} from './instrument';
+import { counterSnapshot, ParentMessage, WorkerSnapshot } from './protocol';
 
 export type QueryQueueTestOptions = Pick<QueryQueueOptions, 'cacheAndQueueDriver' | 'cubeStoreDriverFactory'> & {
   beforeAll?: () => Promise<void>,
@@ -12,56 +23,161 @@ export type QueryQueueTestOptions = Pick<QueryQueueOptions, 'cacheAndQueueDriver
   workers?: number,
 };
 
-function patchQueueDriverConnectionForTrack(connection: QueueDriverConnectionInterface, counters: any): QueueDriverConnectionInterface {
-  function wrapAsyncMethod<M extends MethodName<QueueDriverConnectionInterface>>(methodName: M): any {
-    return async (...args: Parameters<QueueDriverConnectionInterface[M]>) => {
-      if (!(methodName in counters.methods)) {
-        counters.methods[methodName] = {
-          started: 1,
-          finished: 0,
-        };
-      } else {
-        counters.methods[methodName].started++;
-      }
+type PriorityBucket = { priority: number, weight: number };
+
+type BenchSettings = {
+  driver: string,
+  fastTrack: boolean,
+  workers: number,
+  concurrency: number,
+  totalQueries: number,
+  periodMs: number,
+  pushIntervalMs: number,
+  priority: QueuePriority,
+  priorityMix: PriorityBucket[] | null,
+  handlerLatencyMs: number,
+  queueResponseSize: number,
+  queuePayloadSize: number,
+  workerReconcileMs: number,
+  warmupQueries: number,
+  idleTailMs: number,
+  tickMs: number,
+};
 
-      const result = await (connection[methodName] as any)(...args);
-      counters.methods[methodName].finished++;
+type Phase = 'warmup' | 'measure' | 'drain' | 'idle';
 
-      return result;
-    };
-  }
+type Aggregate = {
+  methods: Record<string, MethodCounter>,
+  events: Record<string, number>,
+  handlersStarted: number,
+  handlersFinished: number,
+  fastTrack: { attempts: number, hits: number },
+  driverCalls: number,
+};
 
+const EMPTY_AGGREGATE = (): Aggregate => ({
+  methods: {},
+  events: {},
+  handlersStarted: 0,
+  handlersFinished: 0,
+  fastTrack: { attempts: 0, hits: 0 },
+  driverCalls: 0,
+});
+
+function toAggregate(source: Pick<BenchCounters, 'methods' | 'events' | 'handlersStarted' | 'handlersFinished' | 'fastTrack'>): Aggregate {
   return {
-    ...connection,
-    addToQueue: wrapAsyncMethod('addToQueue'),
-    getResult: wrapAsyncMethod('getResult'),
-    getQueriesToCancel: wrapAsyncMethod('getQueriesToCancel'),
-    getActiveAndToProcess: wrapAsyncMethod('getActiveAndToProcess'),
-    retrieveForProcessing: wrapAsyncMethod('retrieveForProcessing'),
-    getQueryDef: wrapAsyncMethod('getQueryDef'),
-    setResultAndRemoveQuery: wrapAsyncMethod('setResultAndRemoveQuery'),
-    getQueryStageState: wrapAsyncMethod('getQueryStageState'),
-    getResultBlocking: wrapAsyncMethod('getResultBlocking'),
-    optimisticQueryUpdate: wrapAsyncMethod('optimisticQueryUpdate'),
-    getQueryAndRemove: wrapAsyncMethod('getQueryAndRemove'),
-    release: connection.release,
+    methods: cloneMethods(source.methods),
+    events: { ...source.events },
+    handlersStarted: source.handlersStarted,
+    handlersFinished: source.handlersFinished,
+    fastTrack: { ...source.fastTrack },
+    driverCalls: driverCallsTotal(source),
   };
 }
 
-function patchQueueDriverForTrack(driver: QueueDriverInterface, counters: any): QueueDriverInterface {
-  return {
-    ...driver,
-    createConnection: async () => {
-      counters.connections++;
+function addAggregate(into: Aggregate, from: Aggregate): Aggregate {
+  mergeMethods(into.methods, from.methods);
+  mergeEvents(into.events, from.events);
+  into.handlersStarted += from.handlersStarted;
+  into.handlersFinished += from.handlersFinished;
+  into.fastTrack.attempts += from.fastTrack.attempts;
+  into.fastTrack.hits += from.fastTrack.hits;
+  into.driverCalls += from.driverCalls;
 
-      return patchQueueDriverConnectionForTrack(await driver.createConnection(), counters);
-    },
-    redisHash: (...args) => driver.redisHash(...args),
-    release: async (...args) => {
-      counters.connections--;
+  return into;
+}
 
-      return driver.release(...args);
+/**
+ * Warmup is charged against a baseline instead of a counter reset, so the workers need no reset
+ * round trip and the subtraction is identical on every source
+ */
+function subAggregate(a: Aggregate, b: Aggregate): Aggregate {
+  const methods: Record<string, MethodCounter> = {};
+  for (const [name, m] of Object.entries(a.methods)) {
+    const base = b.methods[name] || { started: 0, finished: 0 };
+    methods[name] = { started: m.started - base.started, finished: m.finished - base.finished };
+  }
+
+  const events: Record<string, number> = {};
+  for (const [name, count] of Object.entries(a.events)) {
+    events[name] = count - (b.events[name] || 0);
+  }
+
+  return {
+    methods,
+    events,
+    handlersStarted: a.handlersStarted - b.handlersStarted,
+    handlersFinished: a.handlersFinished - b.handlersFinished,
+    fastTrack: {
+      attempts: a.fastTrack.attempts - b.fastTrack.attempts,
+      hits: a.fastTrack.hits - b.fastTrack.hits,
     },
+    driverCalls: a.driverCalls - b.driverCalls,
+  };
+}
+
+function parsePriorityMix(raw: string | undefined): PriorityBucket[] | null {
+  if (!raw) {
+    return null;
+  }
+
+  const buckets = raw.split(',').map((part) => {
+    const [priority, weight] = part.split(':');
+    return { priority: parseInt(priority, 10), weight: parseInt(weight, 10) };
+  });
+
+  if (buckets.some((b) => Number.isNaN(b.priority) || Number.isNaN(b.weight) || b.weight <= 0)) {
+    throw new Error(`Malformed BENCH_PRIORITY_MIX: ${raw}, expected "10:50,0:50"`);
+  }
+
+  return buckets;
+}
+
+/**
+ * Largest-remainder apportionment, so an uneven mix still interleaves instead of arriving in blocks
+ */
+function pickBucket(buckets: PriorityBucket[], assigned: number[], index: number): number {
+  const total = buckets.reduce((acc, b) => acc + b.weight, 0);
+  let best = 0;
+  let bestScore = -Infinity;
+
+  for (let i = 0; i < buckets.length; i++) {
+    const score = (buckets[i].weight * (index + 1)) / total - assigned[i];
+    if (score > bestScore) {
+      bestScore = score;
+      best = i;
+    }
+  }
+
+  return best;
+}
+
+const envInt = (name: string, fallback: number) => parseInt(process.env[name] || `${fallback}`, 10);
+
+function readSettings(driver: string, workers: number): BenchSettings {
+  const totalQueries = envInt('BENCH_TOTAL_QUERIES', 1000);
+  // BENCH_PERIOD_MS spreads the queries evenly over that window instead of pushing them
+  // as fast as the loop allows, which is what decides whether the queue ever backlogs
+  const periodMs = envInt('BENCH_PERIOD_MS', 0);
+
+  return {
+    driver,
+    fastTrack: process.env.CUBEJS_QUEUE_FAST_TRACK === 'true',
+    workers,
+    concurrency: envInt('BENCH_CONCURRENCY', 50),
+    totalQueries,
+    periodMs,
+    pushIntervalMs: periodMs > 0 && totalQueries > 0 ? Math.max(1, Math.round(periodMs / totalQueries)) : 10,
+    priority: envInt('BENCH_PRIORITY', QueuePriority.Interactive),
+    priorityMix: parsePriorityMix(process.env.BENCH_PRIORITY_MIX),
+    handlerLatencyMs: envInt('BENCH_HANDLER_LATENCY_MS', 1500),
+    // eslint-disable-next-line no-bitwise
+    queueResponseSize: envInt('BENCH_RESPONSE_SIZE', 5 << 20),
+    queuePayloadSize: envInt('BENCH_PAYLOAD_SIZE', 256 * 1024),
+    workerReconcileMs: envInt('BENCH_WORKER_RECONCILE_MS', 50),
+    warmupQueries: envInt('BENCH_WARMUP_QUERIES', 0),
+    idleTailMs: envInt('BENCH_IDLE_TAIL_MS', 0),
+    tickMs: envInt('BENCH_TICK_MS', 1000),
   };
 }
 
@@ -71,88 +187,23 @@ export function QueryQueueBenchmark(name: string, options: QueryQueueTestOptions
       await options.beforeAll();
     }
 
-    const createBenchmark = async (benchSettings: { totalQueries: number, queueResponseSize: number, queuePayloadSize: number, currency: number, pushIntervalMs: number, priority: QueuePriority }) => {
-      const counters = {
-        connections: 0,
-        methods: {},
-        events: {},
-        queueStarted: 0,
-        queueResolved: 0,
-        handlersStarted: 0,
-        handlersFinished: 0,
-        queueDriverQueriesStarted: 0,
-      };
-
-      const queueDriverFactory = (driverType, queueDriverOptions) => {
-        switch (driverType) {
-          case 'memory':
-            return patchQueueDriverForTrack(
-              new LocalQueueDriver(
-                queueDriverOptions
-              ) as any,
-              counters
-            );
-          case 'cubestore':
-            return patchQueueDriverForTrack(
-              new CubeStoreQueueDriver(
-                async () => options.cubeStoreDriverFactory(),
-                queueDriverOptions
-              ),
-              counters
-            );
-          default:
-            throw new Error(`Unsupported driver: ${driverType}`);
-        }
-      };
+    const createBenchmark = async (benchSettings: BenchSettings) => {
+      const counters = createCounters();
 
       const tenantPrefix = crypto.randomBytes(6).toString('hex');
-      const queue = new QueryQueue(`${tenantPrefix}#test_query_queue`, {
-        queryHandlers: {
-          query: async (_query) => {
-            counters.handlersStarted++;
-            await pausePromise(1500);
-            counters.handlersFinished++;
-
-            return {
-              payload: 'a'.repeat(benchSettings.queueResponseSize),
-            };
-          },
-          stream: async (_query, _stream) => {
-            throw new Error('streaming handler is not supported for testing');
-          }
-        },
-        cancelHandlers: {
-          query: async (_query) => {
-            console.error('Cancel handler was called for query');
-          },
-        },
-        continueWaitTimeout: 60 * 2,
-        executionTimeout: 20,
-        orphanedTimeout: 60 * 5,
-        concurrency: benchSettings.currency,
-        logger: (event, _params) => {
-          // console.log(event, _params);
-          // console.log(event);
-
-          if (event in counters.events) {
-            counters.events[event]++;
-          } else {
-            counters.events[event] = 1;
-          }
-
-          if (event.includes('error')) {
-            console.log(event, _params);
-          }
-        },
-        queueDriverFactory,
-        ...options
-      });
-
-      // Spawn worker processes for multi-process simulation (CubeStore only)
-      type WorkerCounters = { handlersStarted: number; handlersFinished: number; events: Record<string, number> };
-      type WorkerState = { worker: ChildProcess; counters: WorkerCounters; prevFinished: number };
+      const queue = createBenchQueue(`${tenantPrefix}#test_query_queue`, counters, benchSettings, options);
+
+      type WorkerState = {
+        worker: ChildProcess,
+        latest: WorkerSnapshot,
+        baseline: Aggregate,
+        awaiting: { seq: number, resolve: () => void } | null,
+        alive: boolean,
+      };
       const workerStates: WorkerState[] = [];
       const numWorkers = options.workers || 0;
+      const shutdown = { requested: false };
+      const diedEarly = { count: 0 };
 
       if (numWorkers > 0) {
         const workerPath = path.resolve(__dirname, 'QueueBenchWorker.js');
@@ -165,13 +216,20 @@ export function QueryQueueBenchmark(name: string, options: QueryQueueTestOptions
 
           const state: WorkerState = {
             worker: w,
-            counters: { handlersStarted: 0, handlersFinished: 0, events: {} },
-            prevFinished: 0,
+            latest: counterSnapshot(createCounters()),
+            baseline: EMPTY_AGGREGATE(),
+            awaiting: null,
+            alive: true,
           };
 
-          w.on('message', (msg: { type: string; data?: WorkerCounters }) => {
-            if (msg.type === 'counters' && msg.data) {
-              state.counters = msg.data;
+          w.on('message', (msg: ParentMessage) => {
+            if (msg.type === 'counters') {
+              state.latest = msg.data;
+
+              if (state.awaiting && (state.awaiting.seq === msg.seq || msg.seq === -1)) {
+                state.awaiting.resolve();
+                state.awaiting = null;
+              }
             }
           });
 
@@ -179,14 +237,30 @@ export function QueryQueueBenchmark(name: string, options: QueryQueueTestOptions
             console.error(`[Worker ${i}] error:`, err);
           });
 
+          // Without this a dead worker keeps its last snapshot and every tick waits the full
+          // timeout for a reply that cannot come, while the run reports as if it were still there
+          w.on('exit', (code, signal) => {
+            if (!shutdown.requested) {
+              // Counted into the result too: on stderr alone this reads as a healthy run in
+              // the .jsonl, and a degraded point would get compared against a whole one
+              diedEarly.count++;
+              console.error(`[Worker ${i}] exited early with ${signal || `code ${code}`} — its counters stop here`);
+            }
+
+            state.alive = false;
+            state.awaiting?.resolve();
+            state.awaiting = null;
+          });
+
           w.send({
             type: 'start',
             tenantPrefix,
             benchSettings: {
               queueResponseSize: benchSettings.queueResponseSize,
-              currency: benchSettings.currency,
+              concurrency: benchSettings.concurrency,
+              handlerLatencyMs: benchSettings.handlerLatencyMs,
             },
-            reconcileInterval: 50,
+            reconcileIntervalMs: benchSettings.workerReconcileMs,
           });
 
           workerStates.push(state);
@@ -195,140 +269,388 @@ export function QueryQueueBenchmark(name: string, options: QueryQueueTestOptions
         console.log(`Spawned ${numWorkers} worker processes`);
       }
 
-      const processingPromisses = [];
+      let tickSeq = 0;
 
-      async function awaitProcessing() {
-        // process query can call reconcileQueue
-        while (await queue.shutdown() || processingPromisses.length) {
-          console.log('awaitProcessing', {
-            counters,
-            processingPromisses: processingPromisses.length
+      async function collectWorkerSnapshots(timeoutMs = 200): Promise<number> {
+        if (workerStates.length === 0) {
+          return 0;
+        }
+
+        const seq = ++tickSeq;
+        const requestedAt = Date.now();
+
+        const waits = workerStates.filter((ws) => ws.alive).map((ws) => new Promise<void>((resolve) => {
+          if (ws.awaiting) {
+            ws.awaiting.resolve();
+          }
+
+          if (!ws.worker.connected) {
+            ws.alive = false;
+            resolve();
+
+            return;
+          }
+
+          ws.awaiting = { seq, resolve };
+          // Never reject: this promise loses the race below whenever a worker is slow, and a
+          // rejection settled after the loser is dropped would surface as an unhandled rejection
+          ws.worker.send({ type: 'tickRequest', seq }, (err) => {
+            if (err) {
+              ws.awaiting = null;
+              resolve();
+            }
           });
-          await Promise.all(processingPromisses.splice(0));
+        }));
+
+        if (waits.length === 0) {
+          return 0;
         }
 
-        // Shutdown worker processes
-        if (workerStates.length > 0) {
-          await Promise.all(workerStates.map((ws) => new Promise<void>((resolve) => {
-            const onMessage = (msg: { type: string; data?: WorkerCounters }) => {
-              if (msg.type === 'counters' && msg.data) {
-                ws.counters = msg.data;
-              }
-              if (msg.type === 'done') {
-                ws.worker.removeListener('message', onMessage);
-                resolve();
-              }
-            };
+        await Promise.race([Promise.all(waits), pausePromise(timeoutMs)]);
 
-            ws.worker.on('message', onMessage);
-            ws.worker.send({ type: 'shutdown' });
-          })));
+        return Date.now() - requestedAt;
+      }
+
+      function snapshotAggregate(): Aggregate {
+        const total = toAggregate(counters);
+        for (const ws of workerStates) {
+          addAggregate(total, toAggregate(ws.latest));
         }
+
+        return total;
       }
 
-      const progressIntervalId = setInterval(() => {
-        console.log('running', {
-          ...counters,
-          processingPromisses: processingPromisses.length,
-          benchSettings,
-          ...(workerStates.length > 0 ? {
-            workers: workerStates.map((ws, i) => {
-              const finishedSinceLastTick = ws.counters.handlersFinished - ws.prevFinished;
-              ws.prevFinished = ws.counters.handlersFinished;
-              return {
-                id: i,
-                handlersStarted: ws.counters.handlersStarted,
-                handlersFinished: ws.counters.handlersFinished,
-                processing: ws.counters.handlersStarted - ws.counters.handlersFinished,
-                processedFromLastEvent: finishedSinceLastTick,
-              };
-            }),
-          } : {}),
-        });
-      }, 1000);
-
-      const lock = createPromiseLock();
-
-      const pusherIntervalId = setInterval(async () => {
-        if (counters.queueStarted >= benchSettings.totalQueries) {
-          lock.resolve();
-          clearInterval(pusherIntervalId);
+      const runStartedAt = Date.now();
+      let phase: Phase = benchSettings.warmupQueries > 0 ? 'warmup' : 'measure';
+      let baseline = EMPTY_AGGREGATE();
+      let baselineMain = EMPTY_AGGREGATE();
 
-          return;
+      let pushed = 0;
+      let completed = 0;
+      const failed = { continueWait: 0, timeout: 0, other: 0 };
+      const errorSamples: string[] = [];
+      let latenciesByPriority: Record<number, number[]> = {};
+      const latencies = () => Object.values(latenciesByPriority).flat();
+
+      const inFlight = () => pushed - completed - failed.continueWait - failed.timeout - failed.other;
+
+      const processingPromisses: Promise<null>[] = [];
+
+      const bucketAssigned = benchSettings.priorityMix ? benchSettings.priorityMix.map(() => 0) : [];
+
+      function priorityFor(index: number): QueuePriority {
+        if (!benchSettings.priorityMix) {
+          return benchSettings.priority;
         }
 
-        counters.queueStarted++;
+        const bucket = pickBucket(benchSettings.priorityMix, bucketAssigned, index);
+        bucketAssigned[bucket]++;
+
+        return benchSettings.priorityMix[bucket].priority;
+      }
+
+      function pushOne(index: number) {
+        pushed++;
 
+        const priority = priorityFor(index);
         const queueId = crypto.randomBytes(12).toString('hex');
+        const startedAt = process.hrtime.bigint();
+
         const running = (async () => {
           try {
             await queue.executeInQueue('query', queueId, {
-              // eslint-disable-next-line no-bitwise
               payload: {
                 large_str: 'a'.repeat(benchSettings.queuePayloadSize)
               },
               orphanedTimeout: 120
-            }, benchSettings.priority, {
+            }, priority, {
               stageQueryKey: 1,
               requestId: 'request-id',
               spanId: 'span-id'
             });
-          } catch (e) {
-            console.error(e);
-          }
 
-          counters.queueResolved++;
+            completed++;
+
+            const latencyMs = Number(process.hrtime.bigint() - startedAt) / 1e6;
+            (latenciesByPriority[priority] ||= []).push(latencyMs);
+          } catch (e: any) {
+            if (e instanceof ContinueWaitError) {
+              failed.continueWait++;
+            } else if (e instanceof TimeoutError) {
+              failed.timeout++;
+            } else {
+              failed.other++;
+            }
 
-          // losing memory for a result
+            if (errorSamples.length < 3) {
+              errorSamples.push(`${e?.constructor?.name}: ${e?.message}`);
+            }
+          }
+
+          // The result is dropped rather than returned, so 1000 payloads are not held alive
+          // by the promise array until the run ends
           return null;
         })();
 
         processingPromisses.push(running);
-        await running;
-      }, benchSettings.pushIntervalMs);
-
-      await lock.promise;
-      await awaitProcessing();
-      clearInterval(progressIntervalId);
-
-      const workerAgg = workerStates.reduce(
-        (acc, ws) => ({
-          handlersStarted: acc.handlersStarted + ws.counters.handlersStarted,
-          handlersFinished: acc.handlersFinished + ws.counters.handlersFinished,
-        }),
-        { handlersStarted: 0, handlersFinished: 0 }
-      );
-
-      console.dir({
-        message: 'Result',
-        benchSettings,
-        ...counters,
-        ...(workerStates.length > 0 ? {
-          workers: workerStates.map((ws, i) => ({
-            id: i,
-            ...ws.counters,
-            processing: ws.counters.handlersStarted - ws.counters.handlersFinished,
-          })),
-          totalHandlersStarted: counters.handlersStarted + workerAgg.handlersStarted,
-          totalHandlersFinished: counters.handlersFinished + workerAgg.handlersFinished,
-        } : {}),
-      }, { depth: null });
+      }
+
+      function runPusher(count: number, intervalMs: number): Promise<void> {
+        if (count <= 0) {
+          return Promise.resolve();
+        }
+
+        const lock = createPromiseLock();
+        let index = 0;
+
+        const pusherIntervalId = setInterval(() => {
+          if (index >= count) {
+            clearInterval(pusherIntervalId);
+            lock.resolve();
+
+            return;
+          }
+
+          pushOne(index);
+          index++;
+        }, intervalMs);
+
+        return lock.promise as Promise<void>;
+      }
+
+      async function drain() {
+        // process query can call reconcileQueue
+        while (await queue.shutdown() || processingPromisses.length) {
+          await Promise.all(processingPromisses.splice(0));
+        }
+      }
+
+      let prevDriverCalls = 0;
+      let prevHandlersFinished = 0;
+      let prevTickAt = runStartedAt;
+      let peakDriverCallsPerSec = 0;
+      let ticking = false;
+
+      const tickIntervalId = setInterval(async () => {
+        if (ticking) {
+          return;
+        }
+        ticking = true;
+
+        try {
+          const workerSnapshotAgeMs = await collectWorkerSnapshots();
+          const now = Date.now();
+          const agg = subAggregate(snapshotAggregate(), baseline);
+
+          const driverCallsDelta = agg.driverCalls - prevDriverCalls;
+          const elapsedSinceTick = Math.max(1, now - prevTickAt);
+          const perSec = (driverCallsDelta * 1000) / elapsedSinceTick;
+
+          if (phase === 'measure' || phase === 'drain') {
+            peakDriverCallsPerSec = Math.max(peakDriverCallsPerSec, perSec);
+          }
+
+          console.log(`BENCH_TICK ${JSON.stringify({
+            runId: process.env.BENCH_RUN_ID || null,
+            tMs: now - runStartedAt,
+            phase,
+            pushed,
+            inFlight: inFlight(),
+            completed,
+            failed: failed.continueWait + failed.timeout + failed.other,
+            driverCallsTotal: agg.driverCalls,
+            driverCallsDelta,
+            driverCallsPerSec: Math.round(perSec),
+            handlersFinished: agg.handlersFinished,
+            handlersFinishedDelta: agg.handlersFinished - prevHandlersFinished,
+            fastTrackAttempts: agg.fastTrack.attempts,
+            fastTrackHits: agg.fastTrack.hits,
+            workerSnapshotAgeMs,
+          })}`);
+
+          prevDriverCalls = agg.driverCalls;
+          prevHandlersFinished = agg.handlersFinished;
+          prevTickAt = now;
+        } finally {
+          ticking = false;
+        }
+      }, benchSettings.tickMs);
+
+      if (benchSettings.warmupQueries > 0) {
+        await runPusher(benchSettings.warmupQueries, Math.min(benchSettings.pushIntervalMs, 50));
+        await drain();
+
+        await collectWorkerSnapshots();
+        baseline = snapshotAggregate();
+        baselineMain = toAggregate(counters);
+        for (const ws of workerStates) {
+          ws.baseline = toAggregate(ws.latest);
+        }
+
+        pushed = 0;
+        completed = 0;
+        failed.continueWait = 0;
+        failed.timeout = 0;
+        failed.other = 0;
+        latenciesByPriority = {};
+        errorSamples.splice(0);
+        prevDriverCalls = 0;
+        prevHandlersFinished = 0;
+      }
+
+      phase = 'measure';
+      const measureStartedAt = Date.now();
+
+      await runPusher(benchSettings.totalQueries, benchSettings.pushIntervalMs);
+      const pushEndedAt = Date.now();
+
+      phase = 'drain';
+      await drain();
+      const drainEndedAt = Date.now();
+
+      // Everything the run reports as its cost comes from this one snapshot, taken after drain
+      // and before the idle tail: worker polling during the tail is never charged to the
+      // queries, and because the three views come from a single instant the per-process
+      // breakdown adds up to the total exactly. Without a fresh pull the idle delta would
+      // absorb up to a tick of worker polling that predates it.
+      await collectWorkerSnapshots();
+      const measured = subAggregate(snapshotAggregate(), baseline);
+      const measuredMain = subAggregate(toAggregate(counters), baselineMain);
+      const measuredWorkers = workerStates.map((ws) => subAggregate(toAggregate(ws.latest), ws.baseline));
+
+      if (benchSettings.idleTailMs > 0) {
+        phase = 'idle';
+        await pausePromise(benchSettings.idleTailMs);
+      }
+
+      clearInterval(tickIntervalId);
+      await collectWorkerSnapshots();
+
+      const idleDriverCalls = benchSettings.idleTailMs > 0
+        ? subAggregate(snapshotAggregate(), baseline).driverCalls - measured.driverCalls
+        : 0;
+
+      shutdown.requested = true;
+
+      if (workerStates.length > 0) {
+        await Promise.all(workerStates.map((ws) => new Promise<void>((resolve) => {
+          if (!ws.alive || !ws.worker.connected) {
+            resolve();
+
+            return;
+          }
+
+          const onMessage = (msg: ParentMessage) => {
+            if (msg.type === 'counters') {
+              ws.latest = msg.data;
+            }
+            if (msg.type === 'done') {
+              ws.worker.removeListener('message', onMessage);
+              resolve();
+            }
+          };
+
+          ws.worker.on('message', onMessage);
+          // A worker that dies before answering would otherwise hold this promise open forever
+          ws.worker.once('exit', resolve);
+          ws.worker.send({ type: 'shutdown' }, (err) => {
+            if (err) {
+              resolve();
+            }
+          });
+        })));
+      }
+
+      const pushWindowMs = pushEndedAt - measureStartedAt;
+      const capacityQps = (benchSettings.concurrency * 1000) / benchSettings.handlerLatencyMs;
+      const targetRateQps = benchSettings.periodMs > 0
+        ? (benchSettings.totalQueries * 1000) / benchSettings.periodMs
+        : null;
+      const actualRateQps = pushWindowMs > 0 ? (pushed * 1000) / pushWindowMs : null;
+      const processes = benchSettings.workers + 1;
+
+      const round = (v: number | null, digits = 3) => (v === null ? null : Number(v.toFixed(digits)));
+
+      const result = {
+        runId: process.env.BENCH_RUN_ID || null,
+        suite: process.env.BENCH_SUITE || null,
+        label: process.env.BENCH_LABEL || null,
+        axis: process.env.BENCH_AXIS ? JSON.parse(process.env.BENCH_AXIS) : null,
+        settings: benchSettings,
+        derived: {
+          capacityQps: round(capacityQps),
+          targetRateQps: round(targetRateQps),
+          actualRateQps: round(actualRateQps),
+          targetRho: round(targetRateQps === null ? null : targetRateQps / capacityQps),
+          actualRho: round(actualRateQps === null ? null : actualRateQps / capacityQps),
+        },
+        timing: {
+          startedAt: new Date(runStartedAt).toISOString(),
+          measureStartedAtMs: measureStartedAt - runStartedAt,
+          pushWindowMs,
+          drainMs: drainEndedAt - pushEndedAt,
+          elapsedMs: drainEndedAt - measureStartedAt,
+          idleTailMs: benchSettings.idleTailMs,
+        },
+        outcome: {
+          pushed,
+          completed,
+          inFlightAtEnd: inFlight(),
+          failed: { ...failed, total: failed.continueWait + failed.timeout + failed.other },
+          errorSamples,
+          workersDiedEarly: diedEarly.count,
+        },
+        latencyMs: percentiles(latencies()),
+        latencyMsByPriority: Object.fromEntries(
+          Object.entries(latenciesByPriority).map(([priority, samples]) => [priority, percentiles(samples)])
+        ),
+        driverCalls: {
+          total: measured.driverCalls,
+          perQuery: completed > 0 ? round(measured.driverCalls / completed) : null,
+          peakPerSec: Math.round(peakDriverCallsPerSec),
+          byMethod: measured.methods,
+          main: measuredMain.methods,
+          workers: measuredWorkers.map((agg) => agg.methods),
+          fastTrack: {
+            ...measured.fastTrack,
+            missRate: measured.fastTrack.attempts > 0
+              ? round(1 - measured.fastTrack.hits / measured.fastTrack.attempts)
+              : null,
+          },
+        },
+        events: {
+          merged: measured.events,
+          main: measuredMain.events,
+          workers: measuredWorkers.map((agg) => agg.events),
+        },
+        handlers: {
+          started: measured.handlersStarted,
+          finished: measured.handlersFinished,
+          main: measuredMain.handlersFinished,
+          workers: measuredWorkers.map((agg) => agg.handlersFinished),
+        },
+        idle: {
+          driverCalls: idleDriverCalls,
+          callsPerSecPerProcess: benchSettings.idleTailMs > 0
+            ? round((idleDriverCalls * 1000) / benchSettings.idleTailMs / processes)
+            : null,
+        },
+        connections: counters.connections,
+      };
+
+      // stdout is a pipe under the suite runner, and a pipe write is asynchronous on POSIX —
+      // the process.exit below would be free to drop a half-written line. console.log offers
+      // no completion callback, so this one line goes out through write().
+      await new Promise<void>((resolve) => {
+        process.stdout.write(`BENCH_RESULT ${JSON.stringify(result)}\n`, () => resolve());
+      });
+
+      if (!process.env.BENCH_RUN_ID) {
+        console.dir({ message: 'Result', ...result }, { depth: null });
+      }
     };
 
-    const totalQueries = parseInt(process.env.BENCH_TOTAL_QUERIES || '1000', 10);
-    // BENCH_PERIOD_MS spreads the queries evenly over that window instead of pushing them
-    // as fast as the loop allows, which is what decides whether the queue ever backlogs
-    const periodMs = parseInt(process.env.BENCH_PERIOD_MS || '0', 10);
-
-    await createBenchmark({
-      currency: parseInt(process.env.BENCH_CONCURRENCY || '50', 10),
-      totalQueries,
-      pushIntervalMs: periodMs > 0 ? Math.max(1, Math.round(periodMs / totalQueries)) : 10,
-      priority: parseInt(process.env.BENCH_PRIORITY || `${QueuePriority.Interactive}`, 10),
-      // eslint-disable-next-line no-bitwise
-      queueResponseSize: parseInt(process.env.BENCH_RESPONSE_SIZE || `${5 << 20}`, 10),
-      queuePayloadSize: parseInt(process.env.BENCH_PAYLOAD_SIZE || `${256 * 1024}`, 10),
-    });
+    await createBenchmark(readSettings(name, options.workers || 0));
 
     if (options.afterAll) {
       await options.afterAll();
diff --git a/packages/cubejs-query-orchestrator/test/benchmarks/QueueBenchWorker.ts b/packages/cubejs-query-orchestrator/test/benchmarks/QueueBenchWorker.ts
index 624c1e48a2e1d..a5f4b01c860d5 100644
--- a/packages/cubejs-query-orchestrator/test/benchmarks/QueueBenchWorker.ts
+++ b/packages/cubejs-query-orchestrator/test/benchmarks/QueueBenchWorker.ts
@@ -2,94 +2,68 @@
 import 'source-map-support/register';
 
 import { CubeStoreDriver } from '@cubejs-backend/cubestore-driver';
-import { pausePromise } from '@cubejs-backend/shared';
 import { QueryQueue } from '../../src';
+import { countEvent, createBenchQueue, createCounters } from './instrument';
+import { counterSnapshot, WorkerMessage, WorkerStartMessage } from './protocol';
 
 if (!process.send) {
   throw new Error('QueueBenchWorker must be run as a child process with IPC');
 }
 
-const counters = {
-  handlersStarted: 0,
-  handlersFinished: 0,
-  events: {} as Record<string, number>,
-};
+// A killed run must not leave a worker polling Cube Store forever
+process.on('disconnect', () => process.exit(0));
+
+const RECONCILE_ERROR_EVENT = 'bench reconcile poll error';
+
+const counters = createCounters();
 
 let cubeStoreDriver: CubeStoreDriver;
 let queue: QueryQueue;
 let reconcileId: ReturnType<typeof setInterval>;
-let progressId: ReturnType<typeof setInterval>;
 
-process.on('message', async (msg: { type: string; tenantPrefix?: string; benchSettings?: { queueResponseSize: number; currency: number; handlerLatencyMs?: number }; reconcileInterval?: number }) => {
+process.on('message', async (msg: WorkerMessage) => {
   if (msg.type === 'start') {
-    const { tenantPrefix, benchSettings, reconcileInterval } = msg as Required<typeof msg>;
+    const { tenantPrefix, benchSettings, reconcileIntervalMs } = msg as WorkerStartMessage;
 
     cubeStoreDriver = new CubeStoreDriver({});
 
-    queue = new QueryQueue(`${tenantPrefix}#test_query_queue`, {
-      queryHandlers: {
-        query: async () => {
-          counters.handlersStarted++;
-          await pausePromise(benchSettings.handlerLatencyMs || 1500);
-          counters.handlersFinished++;
-
-          return {
-            payload: 'a'.repeat(benchSettings.queueResponseSize),
-          };
-        },
-        stream: async () => {
-          throw new Error('streaming handler is not supported for testing');
-        }
+    queue = createBenchQueue(
+      `${tenantPrefix}#test_query_queue`,
+      counters,
+      benchSettings,
+      {
+        cacheAndQueueDriver: 'cubestore',
+        cubeStoreDriverFactory: async () => cubeStoreDriver,
       },
-      cancelHandlers: {
-        query: async () => {
-          console.error('[Worker] Cancel handler was called for query');
-        },
-      },
-      continueWaitTimeout: 60 * 2,
-      executionTimeout: 20,
-      orphanedTimeout: 60 * 5,
-      concurrency: benchSettings.currency,
-      cacheAndQueueDriver: 'cubestore',
-      cubeStoreDriverFactory: async () => cubeStoreDriver,
-      logger: (event, _params) => {
-        if (event in counters.events) {
-          counters.events[event]++;
-        } else {
-          counters.events[event] = 1;
-        }
+      '[Worker] ',
+    );
 
-        if (event.includes('error')) {
-          console.log('[Worker]', event, _params);
-        }
-      },
-    });
-
-    // Periodically reconcile to pick up pending queries from CubeStore
+    // A worker never submits, so it has no submit-time reconcile to bootstrap from. This poll is
+    // an artifact of the harness — production reconcile is event-driven — which is why its
+    // interval is a setting and lands in the reported run settings.
     reconcileId = setInterval(() => {
-      queue.reconcileQueue();
-    }, reconcileInterval);
-
-    // Report counters to main process periodically
-    progressId = setInterval(() => {
-      process.send!({
-        type: 'counters',
-        data: { ...counters, events: { ...counters.events } },
+      // reconcileQueue rethrows, and an unhandled rejection takes the process down under
+      // Node's default policy — the run would then keep reporting as if the worker were here
+      queue.reconcileQueue().catch((e) => {
+        countEvent(counters, RECONCILE_ERROR_EVENT);
+        console.error('[Worker]', RECONCILE_ERROR_EVENT, e);
       });
-    }, 1000);
+    }, reconcileIntervalMs);
+  }
+
+  // Answering on request instead of pushing on a timer keeps the worker numbers inside the tick
+  // they belong to
+  if (msg.type === 'tickRequest') {
+    process.send!({ type: 'counters', seq: msg.seq, data: counterSnapshot(counters) });
   }
 
   if (msg.type === 'shutdown') {
     clearInterval(reconcileId);
-    clearInterval(progressId);
 
     await queue.shutdown();
     await cubeStoreDriver.release();
 
-    process.send!({
-      type: 'counters',
-      data: { ...counters, events: { ...counters.events } },
-    });
+    process.send!({ type: 'counters', seq: -1, data: counterSnapshot(counters) });
     process.send!({ type: 'done' });
     process.disconnect();
     process.exit(0);
diff --git a/packages/cubejs-query-orchestrator/test/benchmarks/QueueCubestore.bench.ts b/packages/cubejs-query-orchestrator/test/benchmarks/QueueCubestore.bench.ts
index 441a6b8aaa718..24bfd0d0871ac 100644
--- a/packages/cubejs-query-orchestrator/test/benchmarks/QueueCubestore.bench.ts
+++ b/packages/cubejs-query-orchestrator/test/benchmarks/QueueCubestore.bench.ts
@@ -28,7 +28,7 @@ const beforeAll = async () => {
 const workers = parseInt(process.env.WORKERS || '2', 10);
 
 QueryQueueBenchmark(
-  `CubeStore Queue (workers: ${workers})`,
+  'cubestore',
   {
     cacheAndQueueDriver: 'cubestore',
     cubeStoreDriverFactory,
diff --git a/packages/cubejs-query-orchestrator/test/benchmarks/QueueMemory.bench.ts b/packages/cubejs-query-orchestrator/test/benchmarks/QueueMemory.bench.ts
index 59f3546264b60..5b0b5b06c459f 100644
--- a/packages/cubejs-query-orchestrator/test/benchmarks/QueueMemory.bench.ts
+++ b/packages/cubejs-query-orchestrator/test/benchmarks/QueueMemory.bench.ts
@@ -12,7 +12,7 @@ const beforeAll = async () => {
 };
 
 QueryQueueBenchmark(
-  'Memory Queue',
+  'memory',
   {
     cacheAndQueueDriver: 'memory',
     beforeAll,
diff --git a/packages/cubejs-query-orchestrator/test/benchmarks/instrument.ts b/packages/cubejs-query-orchestrator/test/benchmarks/instrument.ts
new file mode 100644
index 0000000000000..036909be0fe15
--- /dev/null
+++ b/packages/cubejs-query-orchestrator/test/benchmarks/instrument.ts
@@ -0,0 +1,263 @@
+import { CubeStoreQueueDriver } from '@cubejs-backend/cubestore-driver';
+import { MethodName, pausePromise } from '@cubejs-backend/shared';
+import {
+  AddToQueueResponse,
+  QueueDriverConnectionInterface,
+  QueueDriverInterface,
+  QueuePriority,
+} from '@cubejs-backend/base-driver';
+import { LocalQueueDriver, QueryQueue, QueryQueueOptions } from '../../src';
+
+export type MethodCounter = { started: number, finished: number };
+
+export type BenchCounters = {
+  connections: number,
+  methods: Record<string, MethodCounter>,
+  events: Record<string, number>,
+  handlersStarted: number,
+  handlersFinished: number,
+  fastTrack: { attempts: number, hits: number },
+};
+
+export function createCounters(): BenchCounters {
+  return {
+    connections: 0,
+    methods: {},
+    events: {},
+    handlersStarted: 0,
+    handlersFinished: 0,
+    fastTrack: { attempts: 0, hits: 0 },
+  };
+}
+
+export function countEvent(counters: Pick<BenchCounters, 'events'>, event: string) {
+  counters.events[event] = (counters.events[event] || 0) + 1;
+}
+
+export function driverCallsTotal(counters: Pick<BenchCounters, 'methods'>): number {
+  return Object.values(counters.methods).reduce((acc, m) => acc + m.started, 0);
+}
+
+export function mergeMethods(into: Record<string, MethodCounter>, from: Record<string, MethodCounter>) {
+  for (const [name, m] of Object.entries(from)) {
+    if (name in into) {
+      into[name].started += m.started;
+      into[name].finished += m.finished;
+    } else {
+      into[name] = { started: m.started, finished: m.finished };
+    }
+  }
+
+  return into;
+}
+
+export function mergeEvents(into: Record<string, number>, from: Record<string, number>) {
+  for (const [name, count] of Object.entries(from)) {
+    into[name] = (into[name] || 0) + count;
+  }
+
+  return into;
+}
+
+export function cloneMethods(methods: Record<string, MethodCounter>): Record<string, MethodCounter> {
+  return mergeMethods({}, methods);
+}
+
+export type Percentiles = { count: number, mean: number, p50: number, p90: number, p95: number, p99: number, max: number };
+
+export function percentiles(samples: number[]): Percentiles {
+  if (samples.length === 0) {
+    return { count: 0, mean: 0, p50: 0, p90: 0, p95: 0, p99: 0, max: 0 };
+  }
+
+  const sorted = [...samples].sort((a, b) => a - b);
+  const at = (q: number) => Math.round(sorted[Math.min(sorted.length - 1, Math.max(0, Math.ceil(q * sorted.length) - 1))]);
+
+  return {
+    count: sorted.length,
+    mean: Math.round(sorted.reduce((a, b) => a + b, 0) / sorted.length),
+    p50: at(0.5),
+    p90: at(0.9),
+    p95: at(0.95),
+    p99: at(0.99),
+    max: Math.round(sorted[sorted.length - 1]),
+  };
+}
+
+/**
+ * Asking the connection itself, rather than reading the env, covers the version capability check
+ * and the priority floor too, so a driver that cannot fast track never registers an attempt and
+ * the miss rate stays a property of the queue rather than of the deployment.
+ */
+async function fastTrackEligible(connection: QueueDriverConnectionInterface, priority: QueuePriority): Promise<boolean> {
+  const { useFastTrack } = connection as any;
+
+  return typeof useFastTrack === 'function' ? useFastTrack.call(connection, priority) : false;
+}
+
+/** `addToQueue` is wrapped separately, so that it can also record the fast track outcome */
+const TRACKED_METHODS: MethodName<QueueDriverConnectionInterface>[] = [
+  'getResult',
+  'getQueriesToCancel',
+  'getActiveAndToProcess',
+  'retrieveForProcessing',
+  'getQueryDef',
+  'setResultAndRemoveQuery',
+  'getQueryStageState',
+  'getResultBlocking',
+  'optimisticQueryUpdate',
+  'getQueryAndRemove',
+];
+
+function patchQueueDriverConnectionForTrack(connection: QueueDriverConnectionInterface, counters: BenchCounters): QueueDriverConnectionInterface {
+  function wrapAsyncMethod<M extends MethodName<QueueDriverConnectionInterface>>(methodName: M): any {
+    return async (...args: Parameters<QueueDriverConnectionInterface[M]>) => {
+      if (!(methodName in counters.methods)) {
+        counters.methods[methodName] = {
+          started: 1,
+          finished: 0,
+        };
+      } else {
+        counters.methods[methodName].started++;
+      }
+
+      const result = await (connection[methodName] as any)(...args);
+      counters.methods[methodName].finished++;
+
+      return result;
+    };
+  }
+
+  const trackedAddToQueue = wrapAsyncMethod('addToQueue');
+
+  const tracked: Record<string, any> = {
+    ...Object.fromEntries(TRACKED_METHODS.map((methodName) => [methodName, wrapAsyncMethod(methodName)])),
+    addToQueue: async (...args: Parameters<QueueDriverConnectionInterface['addToQueue']>) => {
+      const eligible = await fastTrackEligible(connection, args[3]);
+      if (eligible) {
+        counters.fastTrack.attempts++;
+      }
+
+      const result: AddToQueueResponse = await trackedAddToQueue(...args);
+      if (eligible && result[4]) {
+        counters.fastTrack.hits++;
+      }
+
+      return result;
+    },
+  };
+
+  // Spreading the connection would copy own properties only and silently drop every method that
+  // lives on the prototype and is not re-listed above — `updateHeartBeat` among them, which
+  // QueryQueue calls on a timer during a long handler
+  return new Proxy(connection, {
+    get: (target, prop) => {
+      if (typeof prop === 'string' && prop in tracked) {
+        return tracked[prop];
+      }
+
+      const value = Reflect.get(target, prop, target);
+
+      return typeof value === 'function' ? value.bind(target) : value;
+    },
+  });
+}
+
+export function patchQueueDriverForTrack(driver: QueueDriverInterface, counters: BenchCounters): QueueDriverInterface {
+  return {
+    ...driver,
+    createConnection: async () => {
+      counters.connections++;
+
+      return patchQueueDriverConnectionForTrack(await driver.createConnection(), counters);
+    },
+    redisHash: (...args) => driver.redisHash(...args),
+    release: async (...args) => {
+      counters.connections--;
+
+      return driver.release(...args);
+    },
+  };
+}
+
+/**
+ * Both the main process and the workers count through the same wrapper, otherwise every run with
+ * `workers > 0` reports the driver calls of the main process only
+ */
+export function makeQueueDriverFactory(counters: BenchCounters, cubeStoreDriverFactory?: () => Promise<any>) {
+  return (driverType: string, queueDriverOptions: any) => {
+    switch (driverType) {
+      case 'memory':
+        return patchQueueDriverForTrack(
+          new LocalQueueDriver(queueDriverOptions) as any,
+          counters
+        );
+      case 'cubestore':
+        return patchQueueDriverForTrack(
+          new CubeStoreQueueDriver(
+            async () => cubeStoreDriverFactory(),
+            queueDriverOptions
+          ),
+          counters
+        );
+      default:
+        throw new Error(`Unsupported driver: ${driverType}`);
+    }
+  };
+}
+
+export type BenchQueueSettings = {
+  queueResponseSize: number,
+  concurrency: number,
+  handlerLatencyMs: number,
+};
+
+/**
+ * The submitter and the workers have to run the same queue: any difference between the two
+ * would show up in the results as a difference between processes
+ */
+export function createBenchQueue(
+  queueName: string,
+  counters: BenchCounters,
+  settings: BenchQueueSettings,
+  options: Pick<QueryQueueOptions, 'cacheAndQueueDriver' | 'cubeStoreDriverFactory'>,
+  logPrefix = '',
+): QueryQueue {
+  return new QueryQueue(queueName, {
+    queryHandlers: {
+      query: async () => {
+        counters.handlersStarted++;
+        await pausePromise(settings.handlerLatencyMs);
+        counters.handlersFinished++;
+
+        return {
+          payload: 'a'.repeat(settings.queueResponseSize),
+        };
+      },
+      stream: async () => {
+        throw new Error('streaming handler is not supported for testing');
+      },
+    },
+    cancelHandlers: {
+      query: async () => {
+        console.error(`${logPrefix}Cancel handler was called for query`);
+      },
+    },
+    continueWaitTimeout: 60 * 2,
+    executionTimeout: 20,
+    orphanedTimeout: 60 * 5,
+    concurrency: settings.concurrency,
+    logger: (event, params) => {
+      countEvent(counters, event);
+
+      // stderr, not stdout: stdout carries the BENCH_RESULT/BENCH_TICK lines the suite
+      // runner parses, and it is shared with every worker
+      if (event.includes('error')) {
+        console.error(`${logPrefix}${event}`, params);
+      }
+    },
+    queueDriverFactory: makeQueueDriverFactory(counters, options.cubeStoreDriverFactory),
+    cacheAndQueueDriver: options.cacheAndQueueDriver,
+    cubeStoreDriverFactory: options.cubeStoreDriverFactory,
+  });
+}
diff --git a/packages/cubejs-query-orchestrator/test/benchmarks/protocol.ts b/packages/cubejs-query-orchestrator/test/benchmarks/protocol.ts
new file mode 100644
index 0000000000000..c35ce80ba8438
--- /dev/null
+++ b/packages/cubejs-query-orchestrator/test/benchmarks/protocol.ts
@@ -0,0 +1,36 @@
+import { BenchCounters, BenchQueueSettings, cloneMethods, MethodCounter } from './instrument';
+
+export type WorkerStartMessage = {
+  type: 'start',
+  tenantPrefix: string,
+  benchSettings: BenchQueueSettings,
+  reconcileIntervalMs: number,
+};
+
+export type WorkerMessage =
+  | WorkerStartMessage
+  | { type: 'tickRequest', seq: number }
+  | { type: 'shutdown' };
+
+export type WorkerSnapshot = {
+  handlersStarted: number,
+  handlersFinished: number,
+  methods: Record<string, MethodCounter>,
+  events: Record<string, number>,
+  fastTrack: { attempts: number, hits: number },
+};
+
+export type ParentMessage =
+  | { type: 'counters', seq: number, data: WorkerSnapshot }
+  | { type: 'done' };
+
+/** Detached from the live counters, so a snapshot in flight over IPC cannot keep moving */
+export function counterSnapshot(counters: Omit<BenchCounters, 'connections'>): WorkerSnapshot {
+  return {
+    handlersStarted: counters.handlersStarted,
+    handlersFinished: counters.handlersFinished,
+    methods: cloneMethods(counters.methods),
+    events: { ...counters.events },
+    fastTrack: { ...counters.fastTrack },
+  };
+}
diff --git a/packages/cubejs-query-orchestrator/test/benchmarks/run-suite.ts b/packages/cubejs-query-orchestrator/test/benchmarks/run-suite.ts
new file mode 100644
index 0000000000000..d4d25c9fac5a0
--- /dev/null
+++ b/packages/cubejs-query-orchestrator/test/benchmarks/run-suite.ts
@@ -0,0 +1,487 @@
+// eslint-disable-next-line import/no-extraneous-dependencies
+import 'source-map-support/register';
+
+import fs from 'fs';
+import path from 'path';
+import readline from 'readline';
+import { fork } from 'child_process';
+// eslint-disable-next-line import/no-extraneous-dependencies
+import yargs from 'yargs';
+// eslint-disable-next-line import/no-extraneous-dependencies
+import { hideBin } from 'yargs/helpers';
+import { pausePromise } from '@cubejs-backend/shared';
+import { driverCallsTotal } from './instrument';
+import { BenchRun, estimateRunMs, rhoOf, Suite, SUITES, suiteByName } from './suites';
+
+type Args = {
+  suites: string[],
+  driver?: 'cubestore' | 'memory',
+  fastTrack: boolean[],
+  settleMs: number,
+  runTimeoutMs: number,
+  only?: string[],
+  repeat: number,
+  out?: string,
+  dryRun: boolean,
+  report?: string,
+  list: boolean,
+};
+
+const commaList = (v: unknown): string[] => String(v).split(',').map((s) => s.trim()).filter(Boolean);
+
+const FAST_TRACK_PASSES: Record<string, boolean> = { on: true, true: true, off: false, false: false };
+
+/**
+ * Every other option is validated by yargs — `choices`, `check`, `strict`. Mapping an
+ * unrecognised token to `off` would run a pass the caller did not ask for and label it as if
+ * they had.
+ */
+function parseFastTrackPasses(v: unknown): boolean[] {
+  const passes = commaList(v).map((token) => {
+    const pass = FAST_TRACK_PASSES[token.toLowerCase()];
+    if (pass === undefined) {
+      throw new Error(`--fast-track: expected on/off, got "${token}"`);
+    }
+
+    return pass;
+  });
+
+  if (passes.length === 0) {
+    throw new Error('--fast-track needs at least one pass, e.g. --fast-track=off,on');
+  }
+
+  return passes;
+}
+
+function parseArgs(argv: string[]): Args {
+  const parsed = yargs(argv)
+    .scriptName('bench:suite')
+    // A variadic positional is only collected by a command builder, and angle brackets would make
+    // it required, which --list and --report do not satisfy
+    .command('$0 [suites..]', 'Run queue benchmark suites, off and on, into a .jsonl', (y) => y
+      .positional('suites', {
+        describe: 'Suite names, or "all" for every suite except the smoke one',
+        type: 'string',
+        array: true,
+        default: [] as string[],
+      }))
+    .example('$0 S1 --dry-run', 'price the sweep before starting it')
+    .example('$0 S1 S6 --settle=5000', 'run two suites back to back')
+    .example('$0 S1 --only=rho=2.5 --repeat=3', 'one point, three times')
+    .example('$0 --report results.jsonl', 'redraw the table from a finished run')
+    .option('driver', {
+      describe: 'Override the driver the suite declares',
+      choices: ['cubestore', 'memory'] as const,
+    })
+    .option('fast-track', {
+      describe: 'Which passes to run for each configuration',
+      default: 'off,on',
+      coerce: parseFastTrackPasses,
+    })
+    .option('settle', {
+      describe: 'Pause between runs, ms — lets the previous run\'s connections go away',
+      type: 'number',
+      default: 3000,
+    })
+    .option('run-timeout', {
+      describe: 'Kill a run after this long, ms',
+      type: 'number',
+      default: 30 * 60 * 1000,
+    })
+    .option('only', {
+      describe: 'Comma separated substrings; keeps just the matching runs',
+      coerce: commaList,
+    })
+    // Above capacity a single pair is not enough to conclude from — the same point, several times
+    .option('repeat', {
+      describe: 'Run each selected point this many times, labelled #1, #2, …',
+      type: 'number',
+      default: 1,
+    })
+    .option('out', {
+      describe: 'Write the .jsonl here instead of .context/bench-results',
+      type: 'string',
+    })
+    .option('dry-run', {
+      describe: 'Print the matrix with computed \u03c1 and estimated wall clock, run nothing',
+      type: 'boolean',
+      default: false,
+    })
+    .option('report', {
+      describe: 'Redraw the summary from an existing .jsonl and exit',
+      type: 'string',
+    })
+    .option('list', {
+      describe: 'List the suites and exit',
+      type: 'boolean',
+      default: false,
+    })
+    .check((a) => {
+      if (!a.list && !a.report && (a.suites as string[]).length === 0) {
+        throw new Error('Name at least one suite, or pass --list / --report');
+      }
+      if (a.repeat < 1) {
+        throw new Error('--repeat must be at least 1');
+      }
+
+      return true;
+    })
+    .strict()
+    .wrap(Math.min(120, process.stdout.columns || 120))
+    .parseSync();
+
+  return {
+    suites: parsed.suites as string[],
+    driver: parsed.driver,
+    fastTrack: parsed.fastTrack,
+    settleMs: parsed.settle,
+    runTimeoutMs: parsed.runTimeout,
+    only: parsed.only,
+    repeat: parsed.repeat,
+    out: parsed.out,
+    dryRun: parsed.dryRun,
+    report: parsed.report,
+    list: parsed.list,
+  };
+}
+
+// dist/test/benchmarks -> dist -> package -> packages -> repo root
+const repoRoot = path.resolve(__dirname, '../../../../..');
+
+function resultsDir(): string {
+  const dir = path.resolve(repoRoot, '.context/bench-results');
+  fs.mkdirSync(dir, { recursive: true });
+
+  return dir;
+}
+
+// Seconds included: the sink appends, so a minute-granularity name silently merges two sweeps
+// of the same suites into one file and the summary then keeps only the last pair per label
+function stamp(): string {
+  return new Date().toISOString().replace(/[:.]/g, '-').slice(0, 19);
+}
+
+type RunRecord = any;
+
+/**
+ * A throw inside the readline listener escapes the run loop entirely — no sink.end(), no
+ * summary, and every remaining point in the sweep skipped. A corrupt line is reachable: the
+ * bench child shares its stdout pipe with every worker it forks, and a pipe only guarantees
+ * atomic writes up to PIPE_BUF.
+ */
+function parseBenchLine(line: string, prefix: string, onError: (message: string) => void): any | null {
+  try {
+    return JSON.parse(line.slice(prefix.length));
+  } catch (e: any) {
+    onError(`unparseable ${prefix.trim()} line (${e.message}): ${line.slice(0, 200)}`);
+
+    return null;
+  }
+}
+
+function selectRuns(suite: Suite, args: Args): BenchRun[] {
+  return args.only ? suite.runs.filter((r) => args.only!.some((o) => r.label.includes(o))) : suite.runs;
+}
+
+async function executeRun(suite: Suite, benchRun: BenchRun, fastTrack: boolean, args: Args, sink: fs.WriteStream): Promise<RunRecord> {
+  const driver = args.driver || suite.driver;
+  const entry = path.resolve(__dirname, driver === 'memory' ? 'QueueMemory.bench.js' : 'QueueCubestore.bench.js');
+  const runId = `${suite.name}/${benchRun.label}/${fastTrack ? 'on' : 'off'}`;
+
+  console.log(`\n=== ${runId} — ${JSON.stringify(benchRun.axis)} — est. ${Math.round(estimateRunMs(benchRun.env) / 1000)}s ===`);
+
+  const child = fork(entry, [], {
+    execArgv: process.execArgv,
+    stdio: ['inherit', 'pipe', 'inherit', 'ipc'],
+    env: {
+      ...process.env,
+      ...benchRun.env,
+      CUBEJS_QUEUE_FAST_TRACK: `${fastTrack}`,
+      BENCH_RUN_ID: runId,
+      BENCH_SUITE: suite.name,
+      BENCH_LABEL: benchRun.label,
+      BENCH_AXIS: JSON.stringify(benchRun.axis),
+    },
+  });
+
+  let result: RunRecord = null;
+  let malformed = 0;
+  const onMalformed = (message: string) => {
+    malformed++;
+    console.error(`  !! ${runId}: ${message}`);
+  };
+
+  const rl = readline.createInterface({ input: child.stdout! });
+  rl.on('line', (line) => {
+    if (line.startsWith('BENCH_RESULT ')) {
+      const parsed = parseBenchLine(line, 'BENCH_RESULT ', onMalformed);
+      if (parsed) {
+        result = parsed;
+        sink.write(`${JSON.stringify({ type: 'run', ...result })}\n`);
+      }
+    } else if (line.startsWith('BENCH_TICK ')) {
+      const parsed = parseBenchLine(line, 'BENCH_TICK ', onMalformed);
+      if (parsed) {
+        sink.write(`${JSON.stringify({ type: 'tick', ...parsed })}\n`);
+      }
+    } else {
+      console.log(`  | ${line}`);
+    }
+  });
+
+  const timeoutMs = Math.max(args.runTimeoutMs, estimateRunMs(benchRun.env) * 2 + 120000);
+  let timedOut = false;
+  const timer = setTimeout(() => {
+    timedOut = true;
+    child.kill('SIGKILL');
+  }, timeoutMs);
+
+  // The child exits right after printing, so its last lines can still be unread at that point
+  const drained = new Promise<void>((resolve) => rl.on('close', resolve));
+  const code = await new Promise<number | null>((resolve) => child.on('exit', resolve));
+  await drained;
+  clearTimeout(timer);
+
+  if (!result) {
+    let error: string;
+    if (timedOut) {
+      error = `timed out after ${timeoutMs}ms`;
+    } else if (malformed > 0) {
+      error = `exited with code ${code}, and ${malformed} line(s) were unparseable`;
+    } else {
+      error = `exited with code ${code} without a BENCH_RESULT`;
+    }
+    console.error(`  !! ${runId}: ${error}`);
+    const failure = { runId, suite: suite.name, label: benchRun.label, axis: benchRun.axis, settings: { fastTrack }, error };
+    sink.write(`${JSON.stringify({ type: 'run', ...failure })}\n`);
+
+    return failure;
+  }
+
+  return result;
+}
+
+const fmt = (v: any, digits = 2) => (typeof v === 'number' ? Number(v.toFixed(digits)) : '—');
+
+function markdown(header: string[], rows: (string | number)[][]): string {
+  return [
+    `| ${header.join(' | ')} |`,
+    `|${header.map(() => '---').join('|')}|`,
+    ...rows.map((row) => `| ${row.join(' | ')} |`),
+  ].join('\n');
+}
+
+/**
+ * The workers poll reconcile on a timer that production does not have, and at a low ρ that poll is
+ * most of the traffic — it dilutes the headline badly. The submitting process is the honest view.
+ */
+function mainCallsPerQuery(record: RunRecord): number | null {
+  const methods = record?.driverCalls?.main;
+  const completed = record?.outcome?.completed;
+  if (!methods || !completed) {
+    return null;
+  }
+
+  return driverCallsTotal({ methods }) / completed;
+}
+
+/**
+ * Where queries drop, per-completed is an efficiency reading and per-pushed is the cost one — the
+ * two diverge sharply and quoting only the first turns a flat cost into an apparent saving
+ */
+function callsPerPushed(record: RunRecord): number | null {
+  const total = record?.driverCalls?.total;
+  const pushed = record?.outcome?.pushed;
+
+  return total && pushed ? total / pushed : null;
+}
+
+function pct(off: number | null | undefined, on: number | null | undefined): string {
+  if (!off || on === null || on === undefined) {
+    return '—';
+  }
+
+  const delta = ((on - off) / off) * 100;
+
+  return `${delta >= 0 ? '+' : ''}${delta.toFixed(1)}%`;
+}
+
+function markdownTable(records: RunRecord[]): string {
+  const byLabel = new Map<string, { off?: RunRecord, on?: RunRecord }>();
+  for (const r of records) {
+    const key = `${r.suite}/${r.label}`;
+    const pair = byLabel.get(key) || {};
+    if (r.settings?.fastTrack) {
+      pair.on = r;
+    } else {
+      pair.off = r;
+    }
+    byLabel.set(key, pair);
+  }
+
+  const header = ['run', 'ρ', 'rate q/s', 'done off→on', 'fail off→on', 'calls/pushed off→on', 'main calls/q off→on', 'Δ main', 'peak calls/s off→on', 'p95 ms off→on', 'elapsed s off→on', 'FT hit%', 'lost'];
+  const rows: (string | number)[][] = [];
+
+  for (const [key, { off, on }] of byLabel) {
+    const sample = on || off;
+    const missRate = on?.driverCalls?.fastTrack?.missRate;
+    const mainOff = mainCallsPerQuery(off);
+    const mainOn = mainCallsPerQuery(on);
+    // A run that lost a worker measured fewer processes than its axis claims
+    const degraded = (off?.outcome?.workersDiedEarly ?? 0) > 0 || (on?.outcome?.workersDiedEarly ?? 0) > 0;
+
+    if (off?.error || on?.error) {
+      rows.push([key, `off: ${off?.error || 'ok'}, on: ${on?.error || 'ok'}`, ...header.slice(2).map(() => '—')]);
+    } else if (sample) {
+      rows.push([
+        degraded ? `⚠ ${key}` : key,
+        fmt(sample.derived?.actualRho),
+        fmt(sample.derived?.actualRateQps),
+        `${off?.outcome?.completed ?? '—'}→${on?.outcome?.completed ?? '—'}`,
+        `${off?.outcome?.failed?.total ?? '—'}→${on?.outcome?.failed?.total ?? '—'}`,
+        `${fmt(callsPerPushed(off))}→${fmt(callsPerPushed(on))}`,
+        `${fmt(mainOff)}→${fmt(mainOn)}`,
+        pct(mainOff, mainOn),
+        `${off?.driverCalls?.peakPerSec ?? '—'}→${on?.driverCalls?.peakPerSec ?? '—'}`,
+        `${fmt(off?.latencyMs?.p95, 0)}→${fmt(on?.latencyMs?.p95, 0)}`,
+        `${fmt((off?.timing?.elapsedMs ?? 0) / 1000, 1)}→${fmt((on?.timing?.elapsedMs ?? 0) / 1000, 1)}`,
+        typeof missRate === 'number' ? `${((1 - missRate) * 100).toFixed(1)}%` : '—',
+        `${off?.events?.merged?.['Orphaned execution result'] ?? 0}→${on?.events?.merged?.['Orphaned execution result'] ?? 0}`,
+      ]);
+    }
+  }
+
+  return markdown(header, rows);
+}
+
+function idleTable(records: RunRecord[]): string | null {
+  const idle = records.filter((r) => r.idle?.callsPerSecPerProcess !== null && r.idle?.callsPerSecPerProcess !== undefined);
+  if (idle.length === 0) {
+    return null;
+  }
+
+  return markdown(
+    ['run', 'fast track', 'workers', 'reconcile ms', 'idle calls', 'calls/s/process'],
+    idle.map((r) => [
+      `${r.suite}/${r.label}`,
+      r.settings.fastTrack ? 'on' : 'off',
+      r.settings.workers,
+      r.settings.workerReconcileMs,
+      r.idle.driverCalls,
+      fmt(r.idle.callsPerSecPerProcess),
+    ]),
+  );
+}
+
+function report(records: RunRecord[]) {
+  console.log('\n');
+  console.log(markdownTable(records));
+
+  const idle = idleTable(records);
+  if (idle) {
+    console.log('\nIdle floor\n');
+    console.log(idle);
+  }
+}
+
+function dryRun(suites: Suite[], args: Args) {
+  let totalMs = 0;
+  const rows: (string | number)[][] = [];
+
+  for (const suite of suites) {
+    for (const benchRun of selectRuns(suite, args)) {
+      const est = estimateRunMs(benchRun.env);
+      totalMs += (est + args.settleMs) * args.fastTrack.length * args.repeat;
+
+      rows.push([
+        suite.name,
+        benchRun.label,
+        JSON.stringify(benchRun.axis),
+        fmt(rhoOf(benchRun.env)),
+        benchRun.env.BENCH_TOTAL_QUERIES,
+        fmt(parseInt(benchRun.env.BENCH_PERIOD_MS || '0', 10) / 1000, 0),
+        Math.round(est / 1000),
+      ]);
+    }
+  }
+
+  console.log(markdown(['suite', 'run', 'axis', 'ρ', 'queries', 'period s', 'est. s per pass'], rows));
+  console.log(`\n${args.fastTrack.length} pass(es) per run — estimated total ${(totalMs / 60000).toFixed(0)} min`);
+}
+
+(async () => {
+  const args = parseArgs(hideBin(process.argv));
+
+  if (args.list) {
+    for (const suite of SUITES) {
+      console.log(`${suite.name.padEnd(6)} ${suite.runs.length} runs, ${suite.driver} — ${suite.description}`);
+    }
+
+    return;
+  }
+
+  if (args.report) {
+    const records: RunRecord[] = [];
+    let unreadable = 0;
+
+    // A sweep that was killed leaves a truncated last line, and one throw here used to lose
+    // the whole file rather than the one record
+    for (const line of fs.readFileSync(args.report, 'utf-8').split('\n').filter((l) => l.trim())) {
+      try {
+        const record = JSON.parse(line);
+        if (record.type === 'run') {
+          records.push(record);
+        }
+      } catch (e: any) {
+        unreadable++;
+      }
+    }
+
+    if (unreadable > 0) {
+      console.error(`!! skipped ${unreadable} unreadable line(s) in ${args.report}`);
+    }
+
+    report(records);
+
+    return;
+  }
+
+  const suites = args.suites.includes('all')
+    ? SUITES.filter((s) => s.name !== 'smoke')
+    : args.suites.map(suiteByName);
+
+  if (args.dryRun) {
+    dryRun(suites, args);
+
+    return;
+  }
+
+  const outPath = args.out || path.resolve(resultsDir(), `${suites.map((s) => s.name).join('+')}-${stamp()}.jsonl`);
+  const sink = fs.createWriteStream(outPath, { flags: 'a' });
+  console.log(`Writing to ${outPath}`);
+
+  const records: RunRecord[] = [];
+
+  for (const suite of suites) {
+    for (const benchRun of selectRuns(suite, args)) {
+      for (let pass = 0; pass < args.repeat; pass++) {
+        for (const fastTrack of args.fastTrack) {
+          const labelled = args.repeat > 1
+            ? { ...benchRun, label: `${benchRun.label}#${pass + 1}` }
+            : benchRun;
+          records.push(await executeRun(suite, labelled, fastTrack, args, sink));
+          // Let the previous run's connections and Cube Store's own bookkeeping settle
+          await pausePromise(args.settleMs);
+        }
+      }
+    }
+  }
+
+  await new Promise<void>((resolve) => sink.end(resolve));
+
+  report(records);
+  console.log(`\nResults: ${outPath}`);
+})().catch((e) => {
+  console.error(e);
+  process.exit(1);
+});
diff --git a/packages/cubejs-query-orchestrator/test/benchmarks/suites.ts b/packages/cubejs-query-orchestrator/test/benchmarks/suites.ts
new file mode 100644
index 0000000000000..eb4056edf8641
--- /dev/null
+++ b/packages/cubejs-query-orchestrator/test/benchmarks/suites.ts
@@ -0,0 +1,258 @@
+import { QueuePriority } from '@cubejs-backend/base-driver';
+
+export type BenchRun = {
+  label: string,
+  axis: Record<string, number | string>,
+  env: Record<string, string>,
+};
+
+export type Suite = {
+  name: string,
+  description: string,
+  driver: 'cubestore' | 'memory',
+  runs: BenchRun[],
+};
+
+/**
+ * The published fast track numbers were all taken at reduced payloads and that was never written
+ * down anywhere. These are the same reduced values, stated once; S7 owns the payload axis.
+ */
+const DEFAULTS: Record<string, string> = {
+  WORKERS: '2',
+  BENCH_CONCURRENCY: '10',
+  BENCH_HANDLER_LATENCY_MS: '1500',
+  BENCH_RESPONSE_SIZE: `${64 * 1024}`,
+  BENCH_PAYLOAD_SIZE: `${16 * 1024}`,
+  BENCH_PRIORITY: `${QueuePriority.Interactive}`,
+  BENCH_WORKER_RECONCILE_MS: '50',
+  BENCH_WARMUP_QUERIES: '20',
+  BENCH_IDLE_TAIL_MS: '0',
+  BENCH_TICK_MS: '1000',
+};
+
+const num = (env: Record<string, string>, key: string) => parseInt(env[key] ?? DEFAULTS[key], 10);
+
+function run(label: string, axis: Record<string, number | string>, env: Record<string, string>): BenchRun {
+  return { label, axis, env: { ...DEFAULTS, ...env } };
+}
+
+export function capacityQps(concurrency: number, handlerLatencyMs: number): number {
+  return (concurrency * 1000) / handlerLatencyMs;
+}
+
+function periodForRho(totalQueries: number, rho: number, concurrency: number, handlerLatencyMs: number): number {
+  return Math.round((totalQueries * 1000) / (rho * capacityQps(concurrency, handlerLatencyMs)));
+}
+
+export function rhoOf(env: Record<string, string>): number | null {
+  const periodMs = num(env, 'BENCH_PERIOD_MS');
+  const total = num(env, 'BENCH_TOTAL_QUERIES');
+  if (!periodMs || !total) {
+    return null;
+  }
+
+  const rate = (total * 1000) / periodMs;
+
+  return rate / capacityQps(num(env, 'BENCH_CONCURRENCY'), num(env, 'BENCH_HANDLER_LATENCY_MS'));
+}
+
+/** Wall clock a run cannot go below: the arrival window, or the time capacity needs to chew through it */
+export function estimateRunMs(env: Record<string, string>): number {
+  const total = num(env, 'BENCH_TOTAL_QUERIES');
+  const capacity = capacityQps(num(env, 'BENCH_CONCURRENCY'), num(env, 'BENCH_HANDLER_LATENCY_MS'));
+  const drainMs = total > 0 ? (total / capacity) * 1000 : 0;
+  const warmupMs = num(env, 'BENCH_WARMUP_QUERIES') > 0
+    ? (num(env, 'BENCH_WARMUP_QUERIES') / capacity) * 1000 + 2000
+    : 0;
+
+  return Math.round(Math.max(num(env, 'BENCH_PERIOD_MS') || 0, drainMs) + num(env, 'BENCH_IDLE_TAIL_MS') + warmupMs);
+}
+
+const S1: Suite = {
+  name: 'S1',
+  description: 'ρ-sweep — the main chart. Driver calls per completed query against the load factor, points clustered around ρ=1 where the knee is.',
+  driver: 'cubestore',
+  runs: [0.5, 0.75, 0.9, 1.0, 1.25, 1.75, 2.5].map((rho) => run(
+    `rho=${rho}`,
+    { rho },
+    {
+      BENCH_TOTAL_QUERIES: '1000',
+      BENCH_PERIOD_MS: `${periodForRho(1000, rho, 10, 1500)}`,
+    }
+  )),
+};
+
+const S2: Suite = {
+  name: 'S2',
+  description: 'Fan-out at underload. The old report called the effect flat across processes, but measured it at ρ≈15 where the fast track degenerates.',
+  driver: 'cubestore',
+  runs: [0, 1, 2, 3, 4, 5].map((workers) => run(
+    `workers=${workers}`,
+    { workers, rho: 0.8 },
+    {
+      WORKERS: `${workers}`,
+      BENCH_TOTAL_QUERIES: '500',
+      BENCH_PERIOD_MS: `${periodForRho(500, 0.8, 10, 1500)}`,
+    }
+  )),
+};
+
+const S3: Suite = {
+  name: 'S3',
+  description: 'Burst then silence. The one shape where the fast track can be a net loss: an ADD_AND_RETRIEVE that comes back empty is a round trip spent for nothing.',
+  driver: 'cubestore',
+  runs: [50, 200].map((concurrency) => run(
+    `burst-c${concurrency}`,
+    { concurrency },
+    {
+      BENCH_CONCURRENCY: `${concurrency}`,
+      BENCH_TOTAL_QUERIES: '1000',
+      BENCH_PERIOD_MS: '5000',
+      BENCH_IDLE_TAIL_MS: '60000',
+    }
+  )),
+};
+
+const S4: Suite = {
+  name: 'S4',
+  description: 'Priority mix, half Interactive half Background. Checks that background never fast-tracks and that it does not starve while interactive does.',
+  driver: 'cubestore',
+  runs: [0.8, 1.5].map((rho) => run(
+    `mix-rho=${rho}`,
+    { rho },
+    {
+      BENCH_PRIORITY_MIX: '10:50,0:50',
+      BENCH_TOTAL_QUERIES: '1000',
+      BENCH_PERIOD_MS: `${periodForRho(1000, rho, 10, 1500)}`,
+    }
+  )),
+};
+
+function idleRun(workers: number, reconcileMs: number): BenchRun {
+  // Nothing polls without a worker, so the interval is not an axis of the control run
+  const polled = workers > 0;
+
+  return run(
+    polled ? `idle-w${workers}-r${reconcileMs}` : `idle-w${workers}`,
+    polled ? { workers, reconcileMs } : { workers },
+    {
+      WORKERS: `${workers}`,
+      BENCH_TOTAL_QUERIES: '0',
+      BENCH_PERIOD_MS: '0',
+      BENCH_WARMUP_QUERIES: '0',
+      BENCH_IDLE_TAIL_MS: '60000',
+      BENCH_WORKER_RECONCILE_MS: `${reconcileMs}`,
+    }
+  );
+}
+
+const S5: Suite = {
+  name: 'S5',
+  description: 'Idle floor — driver calls per second per process on an empty queue. On idle the worker reconcile poll is the entire traffic, so both a harness-fast and a realistic interval are measured.',
+  driver: 'cubestore',
+  runs: [
+    // Nothing polls here: the submitter has no timer of its own and there are no workers, so this
+    // run is zero by construction. It is the control that says the rest of the table is the poll
+    // and nothing else — and it carries no reconcile axis, because that setting only reaches workers
+    idleRun(0, 50),
+    ...[1, 2, 4].flatMap((workers) => [50, 1000].map((reconcileMs) => idleRun(workers, reconcileMs))),
+  ],
+};
+
+const S6: Suite = {
+  name: 'S6',
+  description: 'Handler latency sweep at ρ=0.8. The shorter the handler, the larger the overhead share — the saving as a fraction of wall clock should peak at 100ms.',
+  driver: 'cubestore',
+  runs: [
+    { latencyMs: 100, windowMs: 60000 },
+    { latencyMs: 500, windowMs: 60000 },
+    { latencyMs: 1500, windowMs: 60000 },
+    // 1.6 q/s over a minute is too few samples for a p99, so this point gets a longer window
+    { latencyMs: 5000, windowMs: 180000 },
+  ].map(({ latencyMs, windowMs }) => {
+    const total = Math.round(0.8 * capacityQps(10, latencyMs) * (windowMs / 1000));
+
+    return run(
+      `L=${latencyMs}ms`,
+      { latencyMs, rho: 0.8 },
+      {
+        BENCH_HANDLER_LATENCY_MS: `${latencyMs}`,
+        BENCH_TOTAL_QUERIES: `${total}`,
+        BENCH_PERIOD_MS: `${windowMs}`,
+      }
+    );
+  }),
+};
+
+const S7: Suite = {
+  name: 'S7',
+  description: 'Payload sweep. ADD_AND_RETRIEVE carries the query def inline, so on fat payloads the saved round trips can be eaten by the bytes.',
+  driver: 'cubestore',
+  runs: [
+    // eslint-disable-next-line no-bitwise
+    ...[0, 256 * 1024, 5 << 20, 20 << 20].map((responseSize) => run(
+      `response=${responseSize}`,
+      { responseSize, axis: 'response' },
+      {
+        BENCH_RESPONSE_SIZE: `${responseSize}`,
+        BENCH_PAYLOAD_SIZE: `${16 * 1024}`,
+        BENCH_TOTAL_QUERIES: '300',
+        BENCH_PERIOD_MS: `${periodForRho(300, 0.8, 10, 1500)}`,
+      }
+    )),
+    ...[0, 16 * 1024, 256 * 1024, 1024 * 1024].map((payloadSize) => run(
+      `payload=${payloadSize}`,
+      { payloadSize, axis: 'payload' },
+      {
+        BENCH_RESPONSE_SIZE: `${64 * 1024}`,
+        BENCH_PAYLOAD_SIZE: `${payloadSize}`,
+        BENCH_TOTAL_QUERIES: '300',
+        BENCH_PERIOD_MS: `${periodForRho(300, 0.8, 10, 1500)}`,
+      }
+    )),
+  ],
+};
+
+const S8: Suite = {
+  name: 'S8',
+  description: 'Crossing ρ=1 from the capacity side instead of the arrival side, at a fixed 8.33 q/s. Tests whether what decides is the budget or the ratio.',
+  driver: 'cubestore',
+  runs: [5, 10, 15, 20, 50].map((concurrency) => run(
+    `c=${concurrency}`,
+    { concurrency, rho: Number((8.3333 / capacityQps(concurrency, 1500)).toFixed(3)) },
+    {
+      BENCH_CONCURRENCY: `${concurrency}`,
+      BENCH_TOTAL_QUERIES: '1000',
+      BENCH_PERIOD_MS: '120000',
+    }
+  )),
+};
+
+const SMOKE: Suite = {
+  name: 'smoke',
+  description: 'Two-minute self-check on the memory driver — verifies ticks, percentiles and the result line without a Cube Store.',
+  driver: 'memory',
+  runs: [0.5, 1.5].map((rho) => run(
+    `smoke-rho=${rho}`,
+    { rho },
+    {
+      WORKERS: '0',
+      BENCH_CONCURRENCY: '5',
+      BENCH_HANDLER_LATENCY_MS: '100',
+      BENCH_TOTAL_QUERIES: '200',
+      BENCH_PERIOD_MS: `${periodForRho(200, rho, 5, 100)}`,
+      BENCH_WARMUP_QUERIES: '10',
+    }
+  )),
+};
+
+export const SUITES: Suite[] = [S1, S2, S3, S4, S5, S6, S7, S8, SMOKE];
+
+export function suiteByName(name: string): Suite {
+  const suite = SUITES.find((s) => s.name.toLowerCase() === name.toLowerCase());
+  if (!suite) {
+    throw new Error(`Unknown suite: ${name}. Known: ${SUITES.map((s) => s.name).join(', ')}`);
+  }
+
+  return suite;
+}
diff --git a/yarn.lock b/yarn.lock
index a0e0d26cf1d44..08be06a73eb37 100644
--- a/yarn.lock
+++ b/yarn.lock
@@ -8781,7 +8781,7 @@
   resolved "https://registry.yarnpkg.com/@types/yargs-parser/-/yargs-parser-20.2.1.tgz#3b9ce2489919d9e4fea439b76916abc34b2df129"
   integrity sha512-7tFImggNeNBVMsn0vLrpn1H1uPrUBdnARPTpZoitY37ZrdJREzf7I16tMrlK3hen349gr1NYh8CmZQa7CTG6Aw==
 
-"@types/yargs@^17.0.8":
+"@types/yargs@^17.0.31", "@types/yargs@^17.0.8":
   version "17.0.31"
   resolved "https://registry.yarnpkg.com/@types/yargs/-/yargs-17.0.31.tgz#8fd0089803fd55d8a285895a18b88cb71a99683c"
   integrity sha512-bocYSx4DI8TmdlvxqGpVNXOgCNR1Jj0gNPhhAY+iz1rgKDAaYrAYdFYnhDV1IFuiuVc9HkOwyDcFxaTElF3/wg==