feat(planner): run the #509 stage pipeline with tree-DP selection; retire MajorPass

[zzylol]

Why

#572 (update of 2026-10-04): following #509, only Stage 3 prices plans. MajorPass is retired, and the facade runs the stage pipeline (Stage 1 logical_candidates → Stage 2 physical_candidates → Stage 3 plan_selection). Selection uses a tree dynamic program, not greedy search.

What

• StagePipeline (pass/stage_pipeline.rs) replaces MajorPass. It is the registry built-in "stage-pipeline" and the facade default.
  • It adds PromQL series identity to each root, merges identical input sub-DAGs, takes the Stage 1 inventory, runs select_plan, and merges identical producers after composition.
  • Scalar roots pass through unchanged. PlanOutput.selection reports how the plan was chosen.
• plan_selection::select_plan is the tree DP; see Selection algorithm below.
• plan_selection::select_exhaustive builds and prices every combination, in mixed-radix order, capped at 64. The devtool stage_pipeline and the tests use it; the enumeration now lives in logical_candidates (enumerate_choices, choice_index).
• PlanningModels moves to plan_selection and is re-exported from pass. Its doc states that cost is unused because Stage 3 prices analytically.
• Per-candidate rejection: a failure to compose, run Stage 2, or price rejects that one candidate with its reason; it no longer fails the plan. SelectionError now has only NoValidCandidate.
• Top-k consistency (needed for assumption 2):
  • An exact Aggregate{TopK} now derives the selected-rows schema that fix(ir): top-k sketch readouts return selected rows; Example 1 expectations follow planner output #579 gave sketch readouts: partition keys + item identity + value: Float64. It previously had a packed topk_<k>: Utf8 column, which made count(topk …) over a pass-through top-k fail to compose.
  • Stage 3 sizes every top-k realization at min(input rows, k × groups).
• pass/major.rs is deleted.

Selection algorithm

best(t, c) = local(t, c) + Σ_{u beneath t, read by c} min_c' best(u, c')

• local(t, c) is the change in workload cost when only t takes alternative c.
• A choice is admissible when that one-target candidate builds and passes Stage 3's accuracy check.
• Every Stage 1 realization reads its target's rewritten input, so no choice drops an inner target. The program reduces to a minimum per target over a forest.

The program is exact under three assumptions:

1. Cost is additive per node. It holds: price sums per-node costs.
2. A target's choice changes only its own nodes. Before this PR it failed for top-k with a consumer: Sort → Limit and the sketch readout reported different row counts, and the schemas also differed. After the two top-k fixes it holds; see the probe below.
3. Shared nodes don't depend on choices. It holds: a scan is priced from its time range and the data workload, and neither is a target.

Guard. For every target and the target beneath it, every pair of choices is built. The pair must cost base + local + local (relative tolerance 1e-9) and be admissible exactly when both single choices are. On a mismatch:

• with at most 64 combinations, every combination is built and priced (SelectionMethod::Exhaustive);
• otherwise the program's result is returned with TreeDpNotGuaranteedOptimal { reason }.

Full check. The winner is then built with identical producers merged and checked by Stage 3. If that fails, every combination is built (≤ 64), or the result is flagged.

Probe of assumption 2: every combination is priced and compared with base + the sum of single-target changes. Ingestion is 1 row/s.

Workload	Combinations	Worst gap, 3 series	Worst gap, 1M series	Failing
count(topk by (job) (10, sum_over_time(m[1m])))	30	6.5e-19 (was 3.3e-3)	3.5e-18	0 (was 2)
sum(topk by (job) …)	12	2.2e-19	0	0
quantile(0.5, topk by (job) …)	18	2.2e-19	0	0
max(topk(3, …)) / count(topk(3, …))	12 / 30	0 / 2.2e-19	0 / 2.2e-19	0
sum by (job) (rate(m[1m]))	4	1.6e-19	0	0
topk by (job) (10, sum_over_time(m[1m]))	6	2.2e-19	0	0

Equivalence tests (planner/tests/stage_pipeline_selection.rs) check that the program's choice equals the exhaustive winner's, with method TreeDp:

Workload	Combinations	Winner
#509 Example 1	24	P20
sum by (job) (rate(x[1m]))	4	[1, 0]
count(topk by (job) (10, sum_over_time(m[1m]))), 3 and 1M series	30	[0,0,1] and [0,0,0]
two PromQL queries on one source	24	[1,1,0,1]
SQL distinct count + percentile	15	[0, 0]

Each guard path has a unit test: coupling with ≤ 64 combinations → exhaustive; coupling with > 64 → flagged; uncoupled → TreeDp. The Example 1 program needs 9 builds where enumeration needs 24.

Before / after (#509 Example 1 through e2e_plan)

Before this PR. MajorPass ran the replacement search with DefaultCostModel.

• q1 root: SummaryAgg(Sum) over FinalizeExactAccumulator(Rate).
• q2: Limit ← Sort ← FinalizeExactAccumulator ← SummaryAgg.
• No Stage 3 cost or selection was reported. Only the devtool showed Stage 3's choice (P20).

After this PR. StagePipeline selects P20 by TreeDp, at 79.4011 cpu_ms per workload evaluation.

• q1 root: FinalizeExactAccumulator(Sum) over FinalizeExactAccumulator(Rate).
• q2: Limit ← Sort ← FinalizeExactAccumulator ← SummaryAgg.
• PlanOutput.selection reports the choice.

The regenerated viewer fixture selects the same plan (P20, same total). Only the top-k schemas in it changed.

Behaviour changes (follow-up: #580)

Change	Where	Status
Aggregate{TopK} schema: topk_<k>: Utf8 → partition keys + $promql_series_identity (SQL: item columns) + value: Float64	user-visible schema of exact top-k	intended
PromQL plans carry $promql_series_identity	facade output schemas	intended (as Example 1 / devtool)
Query-time summaries are never cheaper than raw under Stage 3, so raw plans are selected where MajorPass chose summaries	SQL distinct/percentile, PromQL quantiles, SQL SUM batch	accepted; until Stage 2 materialization
No Pass 2 sharing (strictest-consumer sizing, one KLL for p50/p99)	summary_sharing.rs	ignored, #580
Stage 1 lacks rollup, avg rewrite, top-k reuse, exact composition, Hydra, maintained populations	—	#580
PlanningModels.cost unused by the default pass	—	documented, #580

Tests:

• Ignored with Pass 2 sharing and Stage 1 coverage parity with the retired MajorPass #580:
  • Pass 2: quantiles_share_one_producer_sized_for_the_strictest_consumer, cross_series_p50_and_p99_share_one_producer, sql_p50_and_p99_share_one_producer.
  • Raw plan selected: quantiles_with_equal_params_share_one_producer, different_producers_are_not_shared, identical_ungrouped_queries_share_their_producers, identical_sql_percentiles_share_one_producer.
• Expectation changed:
  • operator_design_examples::batch_planning_replaces_and_shares_summary_operators → batch_planning_selects_and_executes_each_plan: Stage 3 picks the raw SUM, and both plans still execute to 31 and 60.
  • e2e_plan::facade_plans_match_cost_only_selection → facade_plans_match_exhaustive_stage_pipeline_selection: it no longer asserts that summaries are chosen.
  • planspace_series_identity_heap::carries_identity ignores the root, because a top-k root now returns identity whatever realizes it.
• Deleted: none. pass/major.rs had no tests.

planner_layering_example1.rs and its acceptance doc are untouched; they still copy the enumeration and could switch to enumerate_choices. Gate: fmt, clippy -D warnings, cargo test --workspace (1533 passed, 12 ignored) and the viewer tests are green.

🤖 Generated with Claude Code