[zzylol]

Stacked on #1235 (base refactor/1207-stage2). Part of #1207 (PR 6 of the restack plan). One commit, 7d1bd5590.

Why

After PR 5, Stage 3's select_plan, the executor's θ and chain handling, the plan's retention and the installer were all todo!(): no workload could be planned or installed. This PR is the baseline of the formulation (concepts doc §3 "Stage 3: baseline selection and plan output", "Plan output", "Executing the plan"): Stage 3 selects with the MILP over $x, y, z, t$ and builds the PhysicalASAPDAG itself; the executor runs what the planner may choose (θ from the node, native chains, a time-indexed store).

What

• Stage 3 (asap-stage3-plan-selection): $w_v$ and $c_{u,c,i}$ from asap-cost; MILP (A)–(F) with (G)/(H) by allocating $y$ only on admissible edges, (E) only inside strongly connected components, and "a query-time result that is not materialized has at most one consumer"; the plan is built from ${x=1}$, ${y=1}$, ${z=1}$; Selection { accepted_accuracy_per_query, pruned_candidates, milp_* }.
• asap-cost: θ from PhysicalASAPDAGNode::parameters (missing θ for a sketch is an error everywhere); chains, TimeRange over a chain, query-time scans and raw data materialized at ingestion priced from the node's retention_ms ($h_v$); plan_cost(dag, entries, data, settings).
• Executor: θ threaded into every summary constructor; chains run natively (an ingestion run builds the newest window; TimeRange/TimeShift over a chain select windows by window end); one time-indexed append-only store with retention (InMemoryDataStore); TimeShift { at: End | Start | T } above a TimeRange evaluates its child at the request's end/start/T, once per run; ScanWindow deleted.
• Types / installer: materialized_view.rs todos filled for native chains; a Scan that keeps raw data (query time, or materialized at ingestion) is raw-data retention, not a view; chain expansion and set_retention deleted; plan_installable installs PlanOutput::plan with its plan_cost report.
• Devtools/integration PR 6 todos (price, peak-memory warning, avg from sum and count).

How

1. select_plan: selectable edges = admissible ∧ the producer delivers_on the evaluation context of every query reaching a query-time consumer (grid evaluation_grid_ms, phase 0; unknown times = a 1 ms grid).
2. cost::selection_costs: each candidate becomes one node (candidate_node, shared with plan building). Groups are visited in Memo::topological_order; each candidate is costed with asap_cost::node_usage over its input groups' logical outputs (per group, the most rows/bytes any costed candidate outputs, the θ of the largest state). That is $w_v$. For each selectable edge, $c_{u,c,i}$ = cost of the consumer with input $i$ read from $u$'s own output − $w_c$. Rates: ingestion time 1 s of rows per second; query time once per evaluation of the queries reaching the group (once per request when materialized). A node that retains rows gets retention_need_ms (most offset + range over its consumers' TimeShift/TimeRange paths, + range-evaluation span; a chain reads back its width).
3. milp::model::solve_selection: HiGHS, gap 0, one thread, coefficients scaled to max 1,000.
4. plan::plan_of: Kahn order over selected edges, PhysicalASAPDAG::new derives the edges; retention_ms = most need over the selected consumers (and the span of a query it answers).

Key code interfaces added / modified

Permalinks at 7d1bd5590.

types/src/ir/physical_export.rs

/// Stays. One plan structure (unchanged fields not repeated in this block are unchanged).
pub struct PhysicalASAPDAGNode {
    /// What a materialized node stores: its group's aggregate_coverage; set only on AcrossEvaluations nodes.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub summary_coverage: Option<ViewCoverage>, // new
}
impl PhysicalASAPDAG {
    /// The DAG of `nodes` (each after its inputs) with one edge per payload input; validated. Stage 3 builds its plan this way.
    pub fn new(nodes: Vec<PhysicalASAPDAGNode>, roots: Vec<PhysicalASAPNodeId>,
        scalar_roots: Vec<(usize, ScalarExpr<PhysicalASAPNodeId>)>) -> Result<Self, PhysicalASAPDAGValidationError>; // new
}

types/src/ir/properties/execution_data_state.rs

impl DataPrimitive {
    /// SummaryState for a summary build/chain/join/maintained population or any State result; Raw otherwise.
    pub fn of<C>(operator: &Operator<C>, result_kind: OperatorResultKind) -> Self; // new
}

types/src/stages/physical/memo_candidates.rs

impl PhysicalASAPMemoCandidates {
    /// Edge `id` of E.
    pub fn edge(&self, id: PhysicalEdgeId) -> &PhysicalEdge; // new
}

types/src/ir/materialized_view.rs

impl From<&SummaryCoverage> for ViewCoverage; // new (I2)
impl ViewCoverage { #[cfg(test-support)] pub fn new(definition_id: String, selection: Vec<SelectionBox>) -> Self; } // new, tests only
impl PhysicalASAPDAG {
    pub fn set_retention(..); // removed: Stage 3 sets retention_ms
    /// Each node's refresh window: a chain's (w, Σ shifts below mod w); others from the windows below.
    pub fn refresh_windows(&self, nodes: impl IntoIterator<Item = PhysicalASAPNodeId>) -> HashMap<PhysicalASAPNodeId, Option<RefreshWindow>>; // was todo!()
    /// Per source, the longest retention_ms of its raw-data Scans (query time, or materialized at ingestion).
    pub fn raw_data_retention(&self) -> Result<Vec<RawDataRetention>, InstallableDagError>; // was todo!()
}
impl MaterializedView { pub fn retention_ms(&self) -> u64; pub fn refresh(&self) -> MaterializedViewRefresh; } // was test-support only

cost/src/usage.rs, cost/src/state_bytes.rs

/// The plan's cost; recurrence/range evaluations from `entries`, formula scale = frozen_calibration_factor (1 unset).
pub fn plan_cost(dag: &PhysicalASAPDAG, entries: &[QueryWorkloadEntry], data: &DataWorkload, settings: &CostInputs)
    -> Result<(PlanCost, CalibrationFit), (PhysicalASAPNodeId, AnalyticalCostError)>; // was (dag, nodes, recurrence, range_evaluations, data, settings, formula_scale)
/// Err UnsupportedCandidate for a sketch without θ.
pub fn summary_shape(family: &FieldDataType, parameters: Option<&SketchParams>, input_rows: StateInputRows) -> Result<(f64, u64), AnalyticalCostError>; // was (family, input_rows) -> (f64, u64)
pub fn node_usage<'a>(node, producers, rate, ingestion_window_ms, data, settings, formula_scale) -> Result<NodeUsage<'a>, AnalyticalCostError>; // retained_tumbling_window_count removed (read from node.retention_ms)
pub fn tumbling_window_retention_ms(..); pub fn tumbling_window_retention_ms_of_plan(..);
pub fn retained_tumbling_windows(..); pub fn tumbling_windows_of(..); // removed (retention_ms on the node)
pub fn maintained_window_ms(..); pub fn node_map(..); pub fn query_reach(..); pub type NodeMap; // now private (no other crate uses them)

executor (I6)

pub fn validate_summary_kernel(family: &FieldDataType, parameters: Option<&SketchParams>, input: &SummaryUpdate) -> Result<(), String>; // θ added
pub fn create_hydra_cms(family: &FieldDataType, parameters: Option<&SketchParams>) -> Result<HydraCms, String>; // θ added
impl StateColumnUpdater { pub fn new(family: &FieldDataType, parameters: Option<&SketchParams>) -> Result<Self, Error>; } // θ added
impl Operator {
    pub fn per_group_summary_build(input, family, parameters: Option<SketchParams>, value, time, groups) -> Result<Self, Error>; // θ added (also per_group_keyed_summary_build, all_groups_summary_build)
    pub fn with_emit_empty_global(self, parameters: Option<SketchParams>) -> Result<Self, Error>; // θ added
    /// Evaluates its input at the request end/start or at T, minus offset_ms (TimeShift{at}).
    pub fn evaluate_at(input: SchemaRef, at: AtModifier, offset_ms: i64) -> Result<Self, Error>; // new
}
pub trait PhysicalOperator { /// The context the node's inputs run in; default: the run's.
    fn input_context(&self, context: &RunContext) -> Result<RunContext, Error>; } // new
impl InMemoryDataStore {
    /// Appends `view`'s rows for (start, end]; refuses overlap/out of order; drops rows older than retention_ms.
    pub fn append_materialized_view(&mut self, view: &MaterializedView, time_range_ms: (i64, i64), batches: Vec<Batch>) -> Result<(), Error>; // new
    pub fn with_materialized_view(..); // removed
}
#[cfg(feature = "test-support")] pub fn set_build_parameters<'a>(nodes: impl IntoIterator<Item = &'a mut PhysicalASAPDAGNode>, parameters: Option<&SketchParams>); // new, tests only
// sources/window.rs: ScanWindow, union, scan_windows, view_windows: removed (private)

planner/src/installable.rs

pub struct PredictedNodeUsage {
    pub node_id: PhysicalASAPNodeId, // was stage3_node_id + installed_node_ids (no expansion)
    pub operator: OperatorKind, pub timing: ExecutionTiming, pub cpu_core_seconds_per_second: f64,
    pub cost_per_second: Option<f64>, pub cost_source: NodeCostSource, // unchanged
}
pub enum InstallablePlanError { /* unchanged variants */
    /// Internal: plan_cost failed on the selected plan.
    PlanCostFailed { node_id: PhysicalASAPNodeId, reason: String }, // new
}

Wire version stays 14 (the stack's; nodes gain summary_coverage, build/merge operators an optional parameters).

Usage (planner test runtime_install::the_plan_carries_stage3s_predicted_usage, passing):

// The workload: p99 over 60 min, every 10 min at phase 0, rank error 0.01; 1,000 series at 1 sample/s.
let workload = workload(&["quantile_over_time(0.99, latency_ms[60m])"]); // fixed_interval_at {600000, 0}
// No raw data kept by the deployment: Stage 2 prunes query-time scans.
let capabilities = DeploymentCapabilities { raw_data_retained: false, ..Default::default() };
// Frontend → Stage 0–2 → select_plan → PhysicalASAPDAG → plan_cost report → install checks.
let (plan, report) = plan_installable(UserInput::new(&workload, FrontendInput::Promql, models)).await?;
// plan.dag: 0 Scan (IT, None) → 1 SummaryTumblingWindowChain[DDSketch α=0.1, w=600000] (IT, AcrossEvaluations,
//   retention_ms 3_600_000) → 2 TimeRange(3600s) → 3 SummaryMerge → 4 SummaryEstimate(0.99) → 5 Project (QT).
// report.per_node: chain 6.568 /s, scan 0.0042 /s, query-time nodes ~1e-5 /s each.
// The alternative, raw data materialized at ingestion (retention 1 h) + exact quantile, costs 14.4 /s and loses.

Major algorithm changes

Baseline selection. Problem: select_plan was todo!(). After: steps in How. Correctness: (A)–(F) as the doc; milp::model::tests check the cheaper producer by $w$+$c$, the one-consumer rule (infeasible without a OneEvaluation candidate) and (E) on a two-group cycle (Ga = m offset 1s, Gb = m offset 2s, each from the other). Complexity: one node_usage per candidate plus one per selectable edge; MILP size |C| + |E_sel| + Σ|G_q| variables. Limits: see spec questions S1–S4.

Native chains / time-indexed store. Before: the installer expanded a chain into N SummaryAgg windows. After: an ingestion run over (start, end] builds one window state (window_end = end); the store appends it and drops rows older than retention_ms; a query reads the windows with window_end ∈ (T − o − r, T − o] (w | r and w | o checked). Test executor/tests/tumbling_window_chain.rs: 24 refreshes of a 10-min exact-sum chain, queries through TimeRange(1h) and TimeShift(1h)→TimeRange(1h) equal direct sums; a read past 2 h retention fails.

Executor gaps (handoff watch-outs) and where they are covered

Watch-out	Covered by
θ from the node in every summary constructor	this PR
TimeRange/TimeShift over chains by window index; newest-n path	this PR (no newest-n merge path was left)
time-indexed store for non-chain AcrossEvaluations	this PR, partly: stored by the interval of the data it summarizes (S5)
TimeShift{at End/Start} above a TimeRange, once per request	this PR (Operator::evaluate_at)
filtered builds at ingestion (validate_ingestion_time_summary)	#1247 (X5, gap 45) — not duplicated; comment points there
ingestion reorder buffer for counter states (+ buffer memory in asap-cost)	#1247 (X5)
exact Reduce TopK	#1242 (X2, gap 12)
stddev/deriv per-entity, Reduce stddev	#1241 (X1, gaps 1–2), #1242 (gap 8)
rate-weighted heap updates (typed binding)	not done; no refusal remains (WeightDomain deleted in PR 5); listed as open

Deleted

Item	Why
PhysicalASAPDAG::set_retention	Stage 3 sets retention_ms
chain expansion in the installer (ExpandedWithRetention, reader_span_ms_by_node, PredictedNodeUsage::{stage3_node_id, installed_node_ids})	chains run natively
executor ScanWindow, union, scan_windows, view_windows	read bounds from the TimeRange/TimeShift nodes
asap-cost tumbling_window_retention_ms(_of_plan), retained_tumbling_windows, tumbling_windows_of	retention is on the node
InMemoryDataStore::with_materialized_view, RawDataWindow::lookback_ms, test_support::lookback_ms	replaced by append_materialized_view / retention checks
price(.., last_root_is_key_filter) parameter	unused
coverage_of (todo)	coverage read from node.summary_coverage

Pending approval (new pub items not in the doc)

1. PhysicalASAPDAG::new(nodes, roots, scalar_roots) -> Result<Self, PhysicalASAPDAGValidationError> — edges derived from payloads (the edge fields stay pub(crate)). Example: nodes [0 Scan, 1 TimeRange{child: 0}], roots [1] → one edge 0→1 with the scan's schema and data state. Also used by the test export.
2. DataPrimitive::of(operator, result_kind) — shared by Stage 3 and the test export. Example: SummaryMerge (result State) → SummaryState; Project → Raw.
3. PhysicalASAPMemoCandidates::edge(id) -> &PhysicalEdge (review D8).
4. plan_cost(dag, entries, data, settings) signature (four callers built the same vectors).
5. Executor: Operator::evaluate_at, PhysicalOperator::input_context, InMemoryDataStore::append_materialized_view, InstallablePlanError::PlanCostFailed, test-support ViewCoverage::new, set_build_parameters (signatures above). Example append_materialized_view(&view, (0, 600_000), vec![state]) stores the window ending at 600 000.
6. is_raw_data_scan (crate-private) semantics: a Scan materialized at ingestion keeps raw data with retention_ms and is reported by raw_data_retention, not as a view (it has no coverage).

Spec questions (code kept independent where possible)

• S1 IT producers and the one-consumer rule. The doc says a ℓ = None producer has at most one consumer per context. At ingestion there is no OneEvaluation, so two chains over one Scan (IT, None) would be infeasible; the baseline applies the rule to query-time producers only. Confirm, and say what PR 7 should do per pipeline context.
• S2 Retention in PR 6. The baseline needs retention_ms to cost and install chains and raw data, so PR 6 sets it post-solve as the largest need over the selected consumers (and costs $w_v$ with the largest need over the selectable ones). PR 7 keeps the same need function (cost::retention_need_ms) for (R) and $\pi h$. @ T stays unbounded (None).
• S3 θ through θ-less state candidates. A merge / TimeRange over a chain has no θ; its logical output takes the largest θ of its input group, so an estimate two hops above a build is costed at that θ. Exact once PR 4 Q1 (merge carries θ) is decided.
• S4 fixed_interval without a phase has no known grid, so no chain or stored result delivers on it; tests that expect sharing switched to fixed_interval_at { phase 0 } (joint_search, planner install tests). The issue-fixture JSONs ([Bug] A nested shorter window of an exact sum does not reuse the newest panes of the longer window's chain #924/[Bug] The same query at a longer repeat interval recomputes instead of reusing the shorter-interval result #925 sharing regressions) were not changed and still fail on this.
• S5 Non-chain AcrossEvaluations results are stored by the interval of data they summarize, not per evaluation time; rows from an @-anchored subtree keep the anchor time (not restamped to each step).
• C1 of PR 5 (QueryTimeResultsNotRetained) and Q1 of PR 4 are untouched.

Before this PR

Planning stopped at select_plan (todo!()); executor summaries panicked without θ; chains could not be run or installed.

After this PR

Every workload whose queries have known evaluation grids plans end to end to one PhysicalASAPDAG, installs, and runs on the executor with native chains.

Verification

cargo test per crate (--all-features --no-fail-fast), cargo fmt --check, cargo clippy --workspace --all-features --all-targets -D warnings clean, crate_graph passes.

Crate	PR 5 (616f765)	PR 6 (7d1bd55)
asap-types	204/0/0	204/0/0
asap-cost	31/1/0	32/0/0
asap-executor	209/62/1	281/0/1
asap-planner	8/18/1	26/3/1
asap-devtools	37/56/3	37/56/3
asap-integration-tests	100/50/4	135/15/4
asap-stage3-plan-selection	3/0/0	6/0/0
others	unchanged	unchanged
Total	1165/187/9	1294/74/9

By test name vs PR 5: 0 newly failing, 129 newly passing, 1 removed (values::weighted_state_tests::weighted_state_family_and_shape_must_match, replaced by state-kind checks).

Still failing (none at a PR 6 todo):

• devtools 56: all stop at Tracking: implement the optimization formulation (memo candidates, properties, edges y_{u,c,i}, MILP) as named code per stage #1207 PR 9 todos (pruned list printer 27, group node 9, fuzz plan cost/roots 15, group label 3, text output 2).
• planner 3: summary_sharing ×2 (plan-shape pins: an exact plan or a chain is chosen where the test expects SummaryAgg states; PR 9 M9); check_compiles_reports_the_executors_error (exact Reduce TopK, executor gap 12 → feat(executor): grouped stddev/variance, corr, group, count_values, exact topk and without grouping #1242).
• integration 15: plan-shape/accuracy pins (DDSketch chosen for rank targets: production_rollup ×2, rollup_controlled, promql_to_post_asap ×3, filtered_aggregates, sharing A filtered query (city = 'NYC') doesn't read its group from a grouped summary another query builds #902; PR 4/8 metric rule); sharing regressions [Bug] A nested shorter window of an exact sum does not reuse the newest panes of the longer window's chain #924/[Bug] The same query at a longer repeat interval recomputes instead of reusing the shorter-interval result #925 ×3 (fixed_interval without phase, S4; PR 7 contexts); identical_exact_queries (pin); joint latency-bound test (latency bounds are out of this design) and the stddev "unsupported" test (no executor-support check in the planner; X1 feat(executor): run exact per-series time functions (rate, changes, stddev_over_time, timestamp, irate state, …) #1241); materialized_views fixture in the old wire format (PR 9 re-baseline).

Self-review: no shims; candidate_node shared by costing and plan building; one retention_need_ms for costing and the plan; cost helpers used only inside asap-cost made private. Not done: RejectionCause keeps variants of the deleted engine (only Infeasible is produced; the viewer contract lists them — delete with the viewer in PR 9); ErrorSelectingPlan::Cost is unused for the same reason.