[zzylol]

Stacked on #1235 (PR 5, Stage 2). Part of #1207 (PR 6 of the order), #1204. Draft; not to be merged on its own (one merge unit, PRs 0–9).

Why

After PR 5, planning stops at select_plan's todo!(): Stage 3, the realizer, the plan export and the executor's θ paths were stubs. #1207's order makes PR 6 the first end-to-end PR: the baseline formulation of the concepts doc §3 ("Stage 3: baseline selection and realization") as named code, with the export and the backend (executor) changed together.

What

• Stage 3 baseline MILP (crates/stage3-plan-selection): variables $x_v$, $y_{u,c,i}$, $z_{q,v}$, $t_v$ (SelectionVariables) and one function per constraint family (A)–(F); (G)/(H) hold because $y$ exists only on admissible edges; (I) are the domains. Objective $\min \sum_v w_v x_v$, $w_v$ under cost model 1 from asap-cost. Decoded into DagOfSelectedCandidates, realized once, priced with plan_cost.
• Realizer (Stage2AssignmentRealizer): one operator node per selected node over its selected producers; a tumbling-window layout becomes a SummaryTumblingWindowChain read through a whole-chain SummaryMerge. Each exported node carries its candidate's timing, lifetime, θ (parameters) and its memo group's summary coverage.
• Export / wire format: PhysicalASAPDAGNode::summary_coverage (new); parameters is filled; plan-file wire version 14. Materialized-view coverage is read from the node (derived only in the memo); installation expands chains with each tumbling window's own coverage slice.
• Backend: the executor builds, checks and runs every summary with θ from the exported node (all PR 0 todo!()s); the cost crate sizes states from the node's θ.
• Every todo!("#1207 PR 6 …") of the stack is implemented (Stage 3, realizer, installation, executor, cost, materialized-view coverage, integration helpers, fuzz pricing).

Not in this PR (#1207 rows 7–9): execution contexts $x^{\mathrm{mat}}v$, $x{\kappa,v}$, $y_{\kappa,u,c,i}$, $h_v$ and (R) (PR 7); accuracy (K) with $E_c$/$D_c$ (PR 8); tools/snapshots (PR 9).

Representation, rules, enumeration (#1207 (a)/(b)/(c))

• (a) Representation: unchanged memo; the MILP variables mirror it ($x$ per physical candidate, $y$ per admissible edge, $z$ per (query, candidate of its output group), $t$ per candidate).
• (b) Rules: none; Stage 3 only reads the memo and the edges.
• (c) Enumeration: the MILP selects the cheapest DAG; a reference enumerator (concepts doc Appendix A, over physical candidates) checks it on a small memo under 50 cost assignments × 2 schedules.

Notation → code

Notation	Code
$x_v$	SelectionVariables::x (binary; fixed 0 for an uncostable candidate)
$y_{u,c,i}$	SelectionVariables::y: HashMap<PhysicalEdgeId, Variable>
$z_{q,v}$	SelectionVariables::z: HashMap<(QueryId, PhysicalCandidateId), Variable>
$t_v$	SelectionVariables::t
(A)	input_satisfaction_constraints
(B)	producer_selection_constraints
(C), (D)	query_coverage_constraints
(E)	acyclicity_constraints (edges inside strongly connected components only)
(F)	utilization_constraints
(G), (H)	selection_variables: $y$ only on edges with satisfies_logical_requirement && satisfies_physical_requirement
(I)	selection_variables (binaries, $0 \le t_v \le
$w_v$	candidate_cost::candidate_costs
decode	milp::model::decode → DagOfSelectedCandidates
delivers_on (§2 alignment)	SelectionInputs::delivers_to_consumer (baseline: every query evaluation reaching the consumer)

How

1. candidate_costs: candidates costed producers first (Kahn's order over admissible edges). Each candidate is one PhysicalASAPDAGNode (a windowed build as the chain the realizer builds) priced by asap_cost::node_usage over one representative (first costed) producer per input, at a fixed rate: 1/s at ingestion; per request of the queries reaching its group for a query-time AcrossEvaluations result; per evaluation of each schedule of those queries otherwise. $w_v$ = node cost + raw-data disk of a query-time scan. Failures → PruneReason::UsageNotCostable.
2. solve_selection: builds (A)–(F), solves with HiGHS to proven optimality (gap 0, one thread), decodes.
3. select_plan: validates the DAG, realizes it, prices it with plan_cost, reports the objective as inexact (InexactObjective, term "fixed cost per candidate … (execution contexts: Tracking: implement the optimization formulation (memo candidates, properties, edges y_{u,c,i}, MILP) as named code per stage #1207 PR 7)") when it differs, and the solver bound only when the objective is exact.
4. Realizer: producers first; over_children maps a candidate's Operator<MemoGroupId> onto its producers' nodes in children() order; confirms update weights the data proves non-negative (so Selection::nodes_with_proven_non_negative_weights and the facade's re-application are deleted); exports with physical_asap_dag_with_assignments.
5. Installation: expand_tumbling_window_chains_in_workload over the realized roots; each tumbling window takes its chain's assignment with ViewCoverage::of_tumbling_window(index, width) and the chain's retention; realized nodes keep theirs; the new range/time shift run with the chain, not materialized.

Major algorithm changes

Stage 3 selection

• Problem. PR 0 removed feat(plan-selection): group selection MILP over execution contexts as Stage 3's default search (#1095) #1144's engine; Stage 3 was todo!(), so no workload planned.
• Before / after. Before (main): feat(plan-selection): group selection MILP over execution contexts as Stage 3's default search (#1095) #1144's context MILP with p[ctx,c], x[m], result slots, e binaries, budget shares. After: steps 1–3 of How.
• Correctness. (A)+(C)+(F)+(E) make every decoded node reachable from an output and acyclic; DagOfSelectedCandidates::validate checks it again. Tests: one per constraint family plus the_milp_agrees_with_the_reference_enumeration (MILP optimum = cheapest enumerated DAG, 100 cases).
• Complexity. Variables $|\mathcal C| + |\mathcal E^{adm}| + \sum_q |G_q| + |\mathcal C|$; constraints $\sum_c |\mathcal I(c)| + |\mathcal E^{adm}| + 2\sum_q|G_q| + |\mathcal E_{SCC}| + |\mathcal C|$. Costing is one pass over candidates and admissible edges.
• Assumptions and limits. The baseline's fixed-cost assumption: $w_v$ does not depend on who selects $v$ (a query-time candidate is charged once per schedule of every query reaching its group, whether or not they select it); retention of a chain is its window, not $h_v$. Accuracy is enforced only through Pass 2's parameter floors ((G)) until (K). Latency and budgets are not modeled (out of scope).

Deleted (replaced in this PR)

Deleted	Replaced by
Selection::nodes_with_proven_non_negative_weights, facade/devtools weight re-application	realizer confirms weights (bounds::confirmed_weights)
bounds::with_confirmed_weights, with_confirmed_weights_at	(unused after the above)
RealizedPlan::operator_node_of_selected_node	(unused)
Stage2AssignmentRealizer::{original_roots, query_time_results_retained}	roots from physical.query_roots; Stage 2 already pruned
PlanOutput::tumbling_window_retention_ms	PlanOutput::realized_plan
PhysicalASAPDAG::operator_nodes, node-tree coverage_of	PhysicalASAPDAGNode::summary_coverage from the memo
#[cfg(test-support)] on the chain expansion helpers	used by installation

New abstractions not in #1207 or the doc (pending approval)

• asap_types::ir::physical_export::{ExportedNodeAssignment, physical_asap_dag_with_assignments} — what the realizer assigns each exported node, and the exporter.
• PhysicalASAPDAGNode::summary_coverage: Option<ViewCoverage> (wire field), ViewCoverage::of, ViewCoverage::of_tumbling_window.
• expand_tumbling_window_chains_in_workload, TumblingWindowChainExpansion, ExpandedTumblingWindowChain (restored; PR 0 deleted them as unused).
• PlanOutput::realized_plan, InstallablePlanError::NoRealizedPlan; plan-file wire version 14.
• ScanWindow::union made public.
• test_support::set_summary_coverage_from_memo; executor test-support set_test_parameters, set_build_parameters, test_parameters.
• Executor signatures taking θ: validate_summary_kernel(.., parameters, ..), create_hydra_cms(family, parameters), StateColumnUpdater::new(family, parameters), Operator::{per_group_summary_build, per_group_keyed_summary_build, all_groups_summary_build}(.., parameters, ..), with_emit_empty_global(parameters); parameters on the summary-build and SummaryMerge operator kinds.
• asap_cost::state_bytes::summary_shape(family, parameters, rows) -> Result.
• Stage 3 private: SelectionInputs, SolvedSelection, SelectionMilpError, candidate_cost module, SelectionInputs::delivers_to_consumer (a baseline stand-in for PR 7's per-context delivers_on).
• Integration test helper planning::realize_from_root_candidates.

Key code interfaces added / modified

Permalinks at 09d69f1: stage3 select.rs, milp/model.rs, candidate_cost.rs, realize.rs, physical_export.rs, installable.rs

/// What the realizer assigns one exported node. // new
pub struct ExportedNodeAssignment {
    pub physical_properties: PhysicalProperties, // new: timing and lifetime
    pub parameters: Option<SketchParams>,       // new: θ of a summary build
    pub summary_coverage: Option<ViewCoverage>, // new: its memo group's coverage
}
/// Export `roots`, each node with `assignment_of(node)`; validated. // new
pub fn physical_asap_dag_with_assignments(
    roots: &[QueryRoot],
    assignment_of: &dyn Fn(&Rc<OperatorNode>) -> ExportedNodeAssignment,
) -> Result<(PhysicalASAPDAG, Vec<Rc<OperatorNode>>), PhysicalASAPDAGValidationError>;

pub struct PhysicalASAPDAGNode {
    // ... every existing field unchanged ...
    /// Its memo group's summary coverage; what a materialized view stores. // new
    pub summary_coverage: Option<ViewCoverage>,
}

/// Stage 2's realizer. // fields changed
pub struct Stage2AssignmentRealizer<'a> {
    pub entries: &'a [QueryWorkloadEntry], // unchanged
    pub data: &'a DataWorkload,            // unchanged
    // original_roots, query_time_results_retained: removed
}

pub struct Selection {
    pub selected_dag: DagOfSelectedCandidates, // unchanged
    // nodes_with_proven_non_negative_weights: removed
    pub cost: PlanCost,                                          // unchanged
    pub accepted_accuracy_per_query: Vec<AcceptedAccuracy>,      // unchanged
    pub pruned_candidates: Vec<PrunedCandidate<PhysicalCandidateId>>, // unchanged (UsageNotCostable)
    pub search_report: SearchReport,                             // unchanged
}

pub struct PlanOutput {
    // plans, scalar_roots, scalar_plans, selection: unchanged
    /// The selected plan as the realizer built it; `None` for a pass that realizes none. // was tumbling_window_retention_ms
    pub realized_plan: Option<RealizedPlan>,
}

Usage (end to end; the test runtime_install::the_plan_file_carries_theta_and_coverage_and_reads_back, which runs at 09d69f1):

// The user's workload: p99 of `latency_ms` over 60 min, every 10 min at phase 0, rank error 0.01, δ 0.01,
// over 1,000 ingesting series sampled every second; the deployment retains no raw data.
let json = plan_json(&["quantile_over_time(0.99, latency_ms[60m])"]).await;
//   plan_installable → e2e_plan → run_planning: Stage 0 memo → Pass 1/2 → Stage 2 candidates and edges
//   → select_plan: candidate_costs (w_v), solve_selection (x, y, z, t; (A)–(F)), decode,
//     Stage2AssignmentRealizer::realize (chain of 10 s… windows at ingestion, θ and coverage per node)
//   → installable_plan_of: chains expanded into tumbling windows, each with its coverage slice and retention.
let plan = InstallablePlan::from_json(&json).unwrap(); // the runtime reads the plan file (wire version 14)
assert_eq!(serde_json::to_string(&plan).unwrap(), json); // it round-trips unchanged
plan.check_compiles().unwrap(); // the executor compiles every refresh and the query run, θ from the nodes
// every SummaryAgg node has `parameters` (θ), every materialized view's node a `summary_coverage`.

Before this PR

Planning stopped at select_plan (todo!()); the executor could not build any sketch (θ todo!s); plan_installable was todo!(). On main, the #1144 engine selected plans.

After this PR

Every workload plans end to end through the baseline MILP, is realized, exported with θ and coverage, installed and compiled by the executor. Representative plans (cost model 1 defaults; fixed_interval_at 10 min, phase 0; 1,000 series at 1 s):

=== p99 over 60m (raw data retained: false)
cost/s 3.352e0; MILP 88 variables, 88 constraints; inexact objective: Some((3.3532199999999994, 3.3522133398133325))
  6 × ASAPSummaryAgg IngestionTime AcrossEvaluations(Memory) θ=Some(DDSketch { alpha: 0.1 })
  1 × ASAPSummaryEstimate QueryTime OneEvaluation(Memory) θ=None
  1 × ASAPSummaryMerge QueryTime None θ=None
  1 × NonASAPProject QueryTime OneEvaluation(Memory) θ=None
  1 × NonASAPScan IngestionTime None θ=None
  6 × NonASAPTimeRange IngestionTime None θ=None
  5 × NonASAPTimeShift IngestionTime None θ=None
=== p99 over 60m (raw data retained: true)
cost/s 3.829e-2; MILP 374 variables, 372 constraints; inexact objective: Some((0.03722, 0.03829272148881333))
  1 × ASAPSummaryAgg QueryTime None θ=Some(DDSketch { alpha: 0.1 })
  1 × ASAPSummaryEstimate QueryTime OneEvaluation(Memory) θ=None
  1 × NonASAPProject QueryTime None θ=None
  1 × NonASAPScan QueryTime OneEvaluation(Memory) θ=None
  1 × NonASAPTimeRange QueryTime OneEvaluation(Memory) θ=None
=== p99 and p50 over 10m (raw data retained: true)
cost/s 7.250e-3; MILP 522 variables, 521 constraints; inexact objective: Some((0.006206666666666666, 0.007250375378146667))
  1 × ASAPSummaryAgg QueryTime None θ=Some(Kll { k: 269, failure_probability: None })
  2 × ASAPSummaryEstimate QueryTime OneEvaluation(Memory) θ=None
  1 × NonASAPProject QueryTime None θ=None
  1 × NonASAPProject QueryTime OneEvaluation(Memory) θ=None
  1 × NonASAPScan QueryTime OneEvaluation(Memory) θ=None
  1 × NonASAPTimeRange QueryTime OneEvaluation(Memory) θ=None
=== sum by job of rate 5m, exact (raw data retained: true)
cost/s 3.905e-3; MILP 78 variables, 76 constraints; inexact objective: Some((0.0039050000000000005, 0.003905046906860001))
  1 × ASAPFinalizeExactAccumulator QueryTime None θ=None
  1 × ASAPSummaryAgg QueryTime None θ=None
  1 × NonASAPAggregate QueryTime None θ=None
  1 × NonASAPScan QueryTime None θ=None
  1 × NonASAPTimeRange QueryTime None θ=None

Plans differ from main (the user checks them with asap-planner --output text; the Stage 3 text printer is PR 9's). Known differences that show in tests are listed below.

Verification

• cargo fmt --all; cargo clippy --workspace --all-features --all-targets -D warnings clean; cargo clippy -p asap-types (no features) now warning-free (the PR 4 tumbling_window_chain.rs/input.rs warnings are gone: the expansion code is used); crate_graph 6/6 (stage 2 → asap-cost is a stage → shared-model edge).
• New tests: Stage 3 — one_producer_per_input_the_cheapest_and_its_inputs_selected ((A)–(D)), an_inadmissible_edge_is_never_selected ((G)/(H)), a_cycle_is_never_selected ((E)), a_free_candidate_nothing_reads_is_not_selected ((F)), an_uncostable_candidate_is_not_selected_and_an_uncostable_root_is_infeasible, a_windowed_state_is_read_only_by_queries_it_delivers_on, strongly_connected_components_group_cycles_only, the_milp_agrees_with_the_reference_enumeration; Stage 2 realizer — three tests (properties/θ/coverage, chain + merge + retention, invalid selection); cost — 5 restored sizing tests + 1; executor — θ tests restored/added (281/0/1); planner — the plan-file round trip.
• Independent review (fresh agent) of the diff; fixed: assignments of realized nodes under expanded chains, merge at ingestion read by a query, costing representatives/worklist, chain-shape agreement costing vs realizer, quadratic lookups, dead fields/functions, doc/terminology, solver bound vs inexact objective. Not taken: realizing twice (facade design from PR 0), executor merge θ-equality check, $t$ only on SCC members (doc fixes t: Vec), range charged though a chain bypasses it (objective is inexact anyway; PR 7).

Test table (cargo test -p <crate> --all-features --no-fail-fast, at 09d69f1; PR 5: 1190/206/9):

crate	passed	failed	ignored
asap-executor	281	0	1
asap-types	197	0	0
asap-frontend-promql	218	0	0
asap-frontend-common	49	0	0
asap-planner	24	5	1
asap-accuracy	54	0	0
asap-cost	37	0	0
asap-frontend-metricsql	15	0	0
asap-frontend-sql	175	0	0
asap-sql-function-catalog	10	0	0
asap-stage0-memo	13	0	0
asap-stage1-pass1-per-computation-candidates	33	0	0
asap-stage1-pass2-reuse-candidates	20	0	0
asap-stage2-physical-optimization	15	0	0
asap-stage3-plan-selection	11	0	0
metricsql_common	3	0	0
asap-devtools	43	51	3
asap-integration-tests	132	38	4
total	1330	94	9

Remaining failures, each with the later PR or the reason:

• asap-devtools (51): 37 stop at PR 9 todo!()s (asap_planner pruned-key/group printers, planning_stages_text); 12 planner_fuzz relations are inconclusive: a witness containing ASAP operators cannot be priced through Stages 0–3 (Pass 1 refuses ASAP input; pricing a fixed witness plan needs PR 9's tool rework), or the related workload is infeasible (below); 2 asap_planner CLI tests read the Stage 2/3 JSON the PR 9 printers write.
• asap-planner (5): view_refreshes_compile_only_the_nodes_below_them, ignoring_accuracy_guarantees_plans_the_undeclared_p99_query — avg by (…) (quantile_over_time(…)) with no raw data: Pass 2 gives the avg-over-quantile edges $\alpha^{\log}=0$ (metric rule), so only the exact plan exists, which needs raw data → infeasible (PR 4 spec question); summary_sharing::{identical_ungrouped_queries_share_their_producers, identical_sql_percentiles_share_one_producer} — different plans (baseline costs; accuracy not enforced until (K), PR 8); check_compiles_reports_the_executors_error — the selected top-k plan holds an Aggregate the executor cannot run (Stage 2 executor-support rule gap, PR 5).
• asap-integration-tests (38): (i) infeasible because Stage 2 prunes every candidate of an output (PerEntityAggregateNotExecutable with no executable alternative, e.g. the unsupported PromQL arm tests; PR 5/Pass 1); (ii) tests pinning a specific plan shape or sharing ratio (sharing_issue_regressions ×5, identical_exact_queries ×2, rollup_controlled, production_rollup_workloads, sql_to_post_asap ×6, promql_to_post_asap, promql_numeric_regressions, filtered_aggregates, sketch_rollup ×2, stage1_logical_asap_optimization ×3, exact_composition) — the baseline's fixed costs and the missing (K) select other plans or helper realizations find no plan of the pinned shape (PR 7 contexts, PR 8 accuracy, PR 9 re-baseline); (iii) kll_tumbling_window_execution ×3, materialized_views — hand-built/snapshot plans without θ or in the pre-14 wire format (PR 9 snapshots); (iv) joint_search_equivalence ×3 — infeasible joint workloads and the accuracy-policy hook (PR 8).
• No failure stops at a PR 6 todo!().