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 (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 followed by a per-group Limit, the shape the PromQL frontend uses for topk. A summary stays SummaryAgg → SummaryEstimate.
  • 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 (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.
  • 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: language_specific_frontends, logical_asap_optimization, physical_asap_optimization (candidates plus pruned with reasons, no cost), and plan_selection (costs, selected, rejected). 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 of the physical DAG once, so a shared node costs the same whatever the number of queries reading it.

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.
• Plan selection:
  • P1 (Q1 exact · Q2 exact) is selected, at 79.4 cpu_ms per workload evaluation.
  • P2 is invalid: CmsWithHeap needs non-negative update weights, and these are not proven non-negative.
  • P3 is invalid: the Top-K output has no guarantee, and the deployment has no evidence that it meets the target.

Evidence

• physical_asap_optimization tests: exact top-k becomes sort then limit and a shared child stays shared; every node runs at query time.
• 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;
  • 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.
• 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: 1593 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.
• Costs are summed per node of the physical DAG, so shared work is charged once.

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.
• 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 → #575 → #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