feat(planner): physical ASAP-aware optimization and plan selection for candidate logical ASAP DAGs

[zzylol]

Why

#509 has four stages. Before this PR the planner stopped after logical ASAP-aware optimization (#575): it produced candidate logical ASAP DAGs but no physical plans, and it selected no plan. This PR adds Stage 2 (physical ASAP-aware optimization) and Stage 3 (plan selection) as an MVP, so a workload goes from the frontends to one selected plan.

What

• Physical ASAP-aware optimization (stage2_physical_asap_optimization::optimize_physical_asap): builds one candidate physical ASAP DAG for each candidate logical ASAP DAG.
  • Materialization: every sub-DAG gets refactor(planner): replace the summary lifecycle with #509 materialization options #667's Materialization::NotMaterialized and runs at query time. Enumerating MatAtIngestion and MatAtQuery is left to Stage 2 materialization: ingestion vs query time per summary #604.
  • Physical operator implementation: exact Aggregate{[TopK]} becomes a per-group Sort (descending, by the ranked measure) followed by a per-group Limit, the shape the PromQL frontend uses for topk. The ranked measure is found by looking through feat(planner): compose candidate logical ASAP DAGs from Pass 1 per-computation candidates #575's composition (a renaming Project, a Filter, FinalizeExactAccumulator or SummaryEstimate) to the reduction: under a Reduce it is the measure column, otherwise the sample value. A restoring Project over the Limit gives roots and consumers the logical [keys.., topk_k] columns; topk_k keeps the ranked value's type and nullability. A summary stays SummaryAgg → SummaryEstimate.
  • Unsupported exact top-k shapes (other measures, a filtered measure, HAVING, a per-entity reduction) fail with PhysicalASAPOptimizationError::ExactTopK and their reason.
  • Query roots are taken as they are, scalar roots included (refactor(planner): plan the workload on the unified operator IR #542). Every node keeps the guarantee feat(planner): compose candidate logical ASAP DAGs from Pass 1 per-computation candidates #575 composed for it. A node shared by two queries stays one node.
  • A candidate it cannot implement fails with PhysicalASAPOptimizationError. The caller prunes that candidate alone and records the error as its reason.
• Plan selection (stage3_plan_selection::select_plan): the only stage that computes cost.
  • Accuracy acceptance uses output-level evidence only. A query's output is accepted when its root guarantee (composed by feat(planner): compose candidate logical ASAP DAGs from Pass 1 per-computation candidates #575 with the built-in theoretical model) meets the target, or when the deployment's empirical output accuracy, AccuracyEvidenceProvider::output_guarantee (new, None by default), shows that it does. Checks on individual estimates inside the plan never accept on their own: a difference of two quantile estimates is rejected under the built-in models.
  • A scalar root that is exactly scalar(v) inherits v's guarantee. Any other scalar expression over approximate inputs has no guarantee until Compose accuracy guarantees through arithmetic and comparisons over approximate values #669.
  • When a target is given, Count-Min (and Count-Min with heap) is accepted only over update weights proven non-negative. A query without a target constrains nothing.
  • Every accepted candidate is priced node by node with analytical_cost, and the per-node costs are summed. The cheapest is selected. Every other candidate is listed with a reason: invalid (with the failed check) or valid but costlier.
• stage_pipeline devtool emits four documents: stage0_query_language_frontends, stage1_logical_asap_optimization, stage2_physical_asap_optimization (candidates plus pruned with reasons, no cost), and stage3_plan_selection (costs, selected, rejected). When plan selection fails, it still emits stage3_plan_selection, with selected: null and the rejections (or an error for any other failure). PromQL rows carry the series identity, as per-series state needs at runtime. The Example 1 fixture tools/dag-viewer/examples/planner-layering-example1.json is regenerated.

How

Physical ASAP-aware optimization rewrites bottom-up with one memo for the whole workload, so sharing between query roots is kept. It then applies apply_materialization_timings with MaterializationAssignment::with_default(NotMaterialized) and exports the timed roots as one PhysicalASAPDAG.

Plan selection walks each query root, first applying the Count-Min weight check, then the output-level accuracy check. Deployment evidence is read only here; #575's composition never reads it. Pricing charges each node once per computation: a materialized node once, a NotMaterialized node once for each consumer that reaches it (query roots, scalar roots, materialized nodes), so a shared node read by two queries costs twice until it is materialized.

Before this PR

stage_pipeline on Example 1 stops at 3 candidate logical ASAP DAGs. There is no physical plan, no cost, and no selected plan.

After this PR

Example 1 gives 3 → 3 → 1:

• Logical ASAP-aware optimization: 3 candidates. Q1 exact in all three. Q2 exact, Count-Min with heap, or CountSketch with heap.
• Physical ASAP-aware optimization: 3 candidates, none pruned, all at query time. Exact Q2 is sort (partition by job) → limit 10, under a Project restoring [job, topk_10].
• Plan selection:
  • P1 (Q1 exact (Sum acc, Rate acc) · Q2 exact (Sum acc)) is selected, at ≈79.40 cpu_ms_per_workload_evaluation.
  • P2 is invalid: q2: CmsWithHeap needs non-negative update weights, and these are not proven non-negative.
  • P3 is invalid: q2: the output has no guarantee, and the deployment has no evidence that it meets target EpsilonDelta { epsilon: 0.01, delta: 0.001 }.

Evidence

• stage2_physical_asap_optimization unit tests: exact top-k becomes sort then limit and a shared child stays shared; PromQL top-k over series sorts by the sample value, descending; top-k over a grouped aggregate sorts by its measure; a consumer of an exact top-k reads its logical columns; top-k with an extra measure or HAVING fails with ExactTopK; every node runs at query time.
• crates/integration-tests/tests/stage2_physical_asap_optimization_sql.rs (new): SQL heavy-hitter queries run through every Pass 1 choice and composition under exact and approximate targets; every exact top-k sorts by the ranked measure, descending, and every root outputs the logical columns — a group key named value, an integer group key, two group keys, and a SUM ranking.
• stage3_plan_selection tests:
  • a candidate whose output guarantee misses the target is invalid with a reason, and the exact plan is selected;
  • Count-Min over signed weights is invalid; count_min_is_admitted_without_a_target; not_materialized_shared_nodes_are_charged_per_query;
  • per-node costs cover the DAG and sum to the total (CountSketch with heap admitted through deployment evidence);
  • combined estimates (a difference of quantile estimates) need evidence about the output, and are accepted when the deployment provides it;
  • scalar(v) inherits v's guarantee and no other scalar expression does.
• stage_pipeline tests: a failed selection lists the rejections with selected: null; a run without physical candidates still emits stage3_plan_selection.
• crates/devtools/tests/stage_pipeline.rs: the four-document output validates; physical ASAP-aware optimization implements or prunes each composed logical candidate exactly once with no cost; plan selection accounts for every physical candidate once and costs exactly the valid ones; the committed fixture is current.

Verification

• cargo fmt --all -- --check: OK
• cargo clippy --workspace --all-targets -- -D warnings: OK
• cargo test --workspace: 1610 passed, 0 failed, 2 ignored
• dag-viewer test_render: 29 OK
• MetricsQL consumer script: OK

Architectural decisions

• Physical ASAP-aware optimization computes no cost; plan selection is the only stage that does, and the only stage that discards valid candidates.
• A candidate physical ASAP-aware optimization cannot implement is pruned with its reason instead of failing the whole run.
• Accuracy acceptance is output-level only. AccuracyModel stays theoretical; the deployment's empirical accuracy enters through AccuracyEvidenceProvider::output_guarantee and only in plan selection.
• Exact top-k has one implementation for roots and consumers: Sort → Limit under a Project restoring [keys.., topk_k].
• Costs are summed per node of the physical DAG, each node priced once per computation: a materialized node (MatAtIngestion/MatAtQuery) once, a NotMaterialized node once per consumer that reaches it (each query root, scalar root or materialized node; counting stops at a materialized node). Each node's multiplier is in the stage_pipeline output. Everything is NotMaterialized until Stage 2 materialization: ingestion vs query time per summary #604, so a shared node is charged once per consuming query (user decision).

Later changes: #674 renamed the modules, test file and document keys by stage number and name (stage2_physical_asap_optimization, stage3_plan_selection, stage0_query_language_frontends, …) and made these stages part of the planner pipeline, which the facade runs by default (run_stage1_to_stage3 / Stage1ToStage3Pass, #675 names); #577 replaced the enumeration cap with the display cap --max-displayed, so plan selection prices every candidate.

Limitations and follow-up

• Materialization offers only NotMaterialized. Enumeration of MatAtIngestion / MatAtQuery and the timing split across query roots: Stage 2 materialization: ingestion vs query time per summary #604.
• No latency check (Q2's 100 ms bound is not enforced) and no deployment capability checks.
• Costs use illustrative statistics, not calibrated measurements (MVP). Join, concat and binary nodes are priced from their first input only.
• A cost error on one candidate aborts selection; it is reachable only from bad declared rates.
• Top-k results drop the item columns, and heavy-hitter composition resolves columns by name: Top-k results drop the item columns (port/job); heavy-hitter candidates resolve columns by name #672.
• Scalar expressions over approximate inputs have no guarantee, and Top-K output has none either: Compose accuracy guarantees through arithmetic and comparisons over approximate values #669, including a Top-K membership/band guarantee so Top-K sketches can be selected without deployment evidence.
• The dag-viewer does not show the pruned list yet: feat(dag-viewer): redesign the viewer around the stage pipeline #574.

Stack: tracker #528 · main → #576 → #577 → …

Human review — do not complete with an agent

• The MVP boundary is correct.
• New conceptual layers or public interfaces are necessary.
• The before/after description matches the intended product behavior.
• Human reviewer:
• Decision and rationale:

🤖 Generated with Claude Code

https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW