[zzylol]

Stacked on #1236 (base refactor/1207-stage3-baseline). Part of #1207 (PR 7 of the restack plan). One commit, 6e46d1fd5 (on PR 6 e53cf6e09; previous head 3b831f97f kept as backup/1207-contexts-3b831f97f). Rebuilt on the restacked PR 6 (7d1bd5590); the previous head 8a466a4c8 is kept as backup/1207-contexts-8a466a4c8 (its realizer merge, inexact-objective terms and TumblingWindowChainGeometry have no counterpart left after PR 6 and are not carried over).

Why

PR 6's selection runs each candidate at most once, at the combined rate of every query reaching its group, and sets retention after solving. The formulation (concepts doc §5 "Execution contexts, by example", "Cost and retention") runs a candidate once per execution context that needs it, a materialized one once in its own context, and makes retention a decision: $h_v$ with constraint (R) and $\pi_v h_v$ in the objective.

What

• Stage 3 selects over executions: $x_{\kappa,v}$ per execution of a candidate in a context ($x^{\mathrm{mat}}v$ = the execution of an AcrossEvaluations $v$ in Materialized(v)), $y{\kappa,u,c,i}$ per edge in the consumer's context (only where the producer delivers_on that context), $z$, $t$ per execution on strongly connected components only, and $h_v$. Constraints (A)–(F) per execution, the one-consumer rule per context (query-time producers, S1 unchanged), and (R).
• Objective $\sum_v W^{\mathrm{mat}}v x^{\mathrm{mat}}v + \sum \rho\kappa c{\kappa,v} x_{\kappa,v} + \sum \rho_\kappa c_{\kappa,u,c,i} y_{\kappa,u,c,i} + \sum_v \pi_v h_v$ (C1–C3).
• The plan has one node per selected execution (several per candidate when it runs in several contexts); retention_ms is $h_v$ from the solution. PR 6's post-solve retention step is deleted.
• types: delivers_on in Materialized(v) (was todo!()), with PhysicalCandidate::chain_width_ms.
• asap-cost aligned with the contexts (K1): a not-materialized node is computed once per evaluation grid of the queries reaching it (was once per query); a query-time chain builds one window per width (was once per request), from its input's rows (was the input divided by the retained windows).
• Retention fix found by the new end-to-end test: a chain that is not materialized is rebuilt in every evaluation, so it reads back what its own consumers read, not one window.

Fixes (2026-11-09): Stage 3 edge costs went negative (an execution, its edge and its retention summed to −1,356,902/s on example 3a's shape): a candidate's baseline input was the most any candidate of the input group outputs, whatever its timing, so an ingestion-time chain was costed over the five years a materialized raw Scan retains for query-time consumers, which also inflated its $\pi_v$. The baseline is now over the producers the candidate may read (its admissible edges, baseline_inputs), and an ingestion-time consumer reads one batch of raw data, also from a Scan materialized at ingestion time (reads_one_batch_of: left out of the baseline, and its edge costs what reading the Scan that is not materialized does). Pinned by cost::tests::no_execution_reading_a_producer_costs_less_than_nothing (per execution + edge + $\pi_v$ × the least need ≥ 0). Concepts doc: "logical size of its inputs" says over which producers.

How

1. ExecutionContexts::new walks from each query output: the query's context is Evaluation { grid } (evaluation_grid: evaluation_grid_ms of its range evaluation at phase 0, 1 ms when unknown). For each consumer execution, its input context is its own, except: an ingestion-time AcrossEvaluations consumer → IngestionPipeline(refresh window) (chain width, else its group's grid), so ingestion builds of one window share inputs; a TimeShift { at: End | Start } → the grid of its queries' request end (start) times. Each admissible edge into the consumer whose producer delivers_on the input context becomes $(consumer, edge, producer)$; a producer with ℓ = None/OneEvaluation runs in that context, an AcrossEvaluations one in Materialized(u).
2. cost::execution_costs: logical outputs per group as in PR 6; each execution costed with node_usage at its rate executions_per_second (ingestion 1; evaluation context: evaluation_rate of the context's queries reaching the group, the @-anchored subtree once per request; Materialized(m): 1000/w for a query-time chain, else once per request). Edge cost = consumer's cost in its context with input $i$ from $u$ − its own. A candidate that retains rows (materialized, or a query-time Scan) is costed at the most any of its consumers reads back; $\pi_v$ = cost of one more window (chain) or second, $W^{\mathrm{mat}}_v$ = cost − $\pi_v h_v$; for a query-time scan only the raw data on disk is in $\pi_v h_v$. A candidate one of whose consumers reads it unbounded (a Scan read whole, @) has no $h_v$ and is costed retaining everything (PR 6 behaviour).
3. milp::model::solve_selection adds (R): $h_v \ge \mathrm{need}\cdot y$ per selected edge (edge_retention_need_ms) and $h_v \ge \mathrm{span}q\cdot z{q,v}$; $h$ is in seconds for scaling.
4. plan::plan_of builds one node per selected execution in Kahn order.

Key code interfaces added / modified

Permalinks at [3b831f97f] (before the 2026-11-09 fix)(https://github.com/ProjectASAP/ASAPPlanner/tree/3b831f97f/crates).

types/src/stages/physical/memo_candidates.rs

impl PhysicalCandidate {
    /// Whether a consumer running in `context` can read this producer; in `Materialized(v)` on v's own
    /// times (a chain v: its width at phase 0, else v's group grid; neither: false for a time-dependent producer).
    pub fn delivers_on(&self, candidates: &PhysicalASAPMemoCandidates, context: &ExecutionContext) -> bool; // Materialized(v) was todo!()
    /// Its tumbling-window width when it is a SummaryTumblingWindowChain; used by delivers_on and Stage 3's rates.
    pub fn chain_width_ms(&self, candidates: &PhysicalASAPMemoCandidates) -> Option<u64>; // new
}

stage3-plan-selection/src/milp/execution_contexts.rs (new, crate-private)

/// One run of a candidate in a context: a node of the plan. x^mat_v = x of (Materialized(v), v).
pub(crate) struct Execution {
    /// κ. // new
    pub(crate) context: ExecutionContextId,
    /// v. // new
    pub(crate) candidate: PhysicalCandidateId,
}
/// The distinct contexts, every execution and every y_{κ,u,c,i} reached from the query outputs.
pub(crate) struct ExecutionContexts {
    /// The distinct contexts, by ExecutionContextId (deduplicated by equality). // new
    contexts: Vec<ExecutionContext>,
    /// Every reached execution, ascending (index_of = binary search). // new
    executions: Vec<Execution>,
    /// Every y as (consumer execution, edge (u,c,i), producer execution); κ = the consumer's context. // new
    edges: Vec<(Execution, PhysicalEdgeId, Execution)>,
}
impl ExecutionContexts {
    /// Contexts, executions, edges, and per output the executions that may answer it.
    pub(crate) fn new(physical: &PhysicalASAPMemoCandidates, entries: &[QueryWorkloadEntry],
        outputs: &[(QueryId, MemoGroupId)]) -> (Self, Vec<(QueryId, Vec<Execution>)>);
    pub(crate) fn executions(&self) -> &[Execution];
    pub(crate) fn edges(&self) -> &[(Execution, PhysicalEdgeId, Execution)];
    pub(crate) fn index_of(&self, execution: Execution) -> usize;
}
/// A query's context grid (moved from select.rs `evaluation_context`).
pub(crate) fn evaluation_grid(entry: &QueryWorkloadEntry) -> EvaluationGrid;
/// ρ of an execution (see How 2).
pub(crate) fn executions_per_second(physical: &PhysicalASAPMemoCandidates, entries: &[QueryWorkloadEntry],
    contexts: &ExecutionContexts, execution: Execution, non_repeating_query_horizon_seconds: f64) -> Result<f64, String>;

stage3-plan-selection/src/cost.rs

/// The objective's coefficients per second. Replaces SelectionCosts { candidate, edge }.
pub(crate) struct ExecutionCosts {
    /// ρ_κ c_{κ,v}, or W^mat_v for an execution in Materialized(v), by execution position; Err = not costable (x = 0). // new
    pub(crate) per_execution: Vec<Result<f64, String>>,
    /// ρ_κ c_{κ,u,c,i}, by position in ExecutionContexts::edges; None = not costable (no y). // new
    pub(crate) per_edge: Vec<Option<f64>>,
    /// π_v per ms retained, for a candidate retaining rows only within a bound (it gets h_v and (R)). // new
    pub(crate) retention_cost_per_second: HashMap<PhysicalCandidateId, f64>,
}
pub(crate) fn execution_costs(physical, contexts: &ExecutionContexts, query_outputs: &[(QueryId, Vec<Execution>)],
    entries, data, cost_inputs, formula_scale) -> ExecutionCosts; // was selection_costs(physical, selectable, ..) -> SelectionCosts
/// need_{κ,u,c,i} of an edge (None = unbounded). // new
pub(crate) fn edge_retention_need_ms(physical, entries, edge: PhysicalEdgeId) -> Option<u64>;
/// A query reading an output directly reads back its range-evaluation span. // new
pub(crate) fn output_retention_need_ms(entry: &QueryWorkloadEntry) -> u64;

stage3-plan-selection/src/milp/model.rs

pub(crate) struct SelectionInputs<'a> {
    pub(crate) physical: &'a PhysicalASAPMemoCandidates, // unchanged
    /// Each query's entry, for the needs of (R). // new
    pub(crate) entries: &'a [QueryWorkloadEntry],
    /// The executions and edges x and y are over. // new
    pub(crate) contexts: &'a ExecutionContexts,
    pub(crate) costs: &'a ExecutionCosts, // was &SelectionCosts
    pub(crate) query_outputs: &'a [(QueryId, Vec<Execution>)], // was Vec<PhysicalCandidateId>
}
pub(crate) struct SelectionVariables {
    x: Vec<Variable>,                          // was by candidate; now by execution
    y: Vec<Option<Variable>>,                  // was HashMap<PhysicalEdgeId, Variable>; now by context edge
    z: HashMap<(QueryId, Execution), Variable>, // was keyed by PhysicalCandidateId
    t: Vec<Option<Variable>>,                  // by execution, on strongly connected components only
    h: HashMap<PhysicalCandidateId, Variable>, // new: h_v in seconds
}
pub(crate) struct SolvedSelection {
    pub(crate) executions: Vec<Execution>,                         // was candidates
    pub(crate) edges: Vec<(Execution, PhysicalEdgeId, Execution)>, // was Vec<PhysicalEdgeId>
    pub(crate) outputs: Vec<Execution>,                            // was Vec<PhysicalCandidateId>
    pub(crate) retention_ms: HashMap<PhysicalCandidateId, u64>,    // new: h_v
    pub(crate) variables: usize, pub(crate) constraints: usize,
    pub(crate) construction_ms: u64, pub(crate) solve_ms: u64,     // unchanged
}
fn retention_constraints(inputs: &SelectionInputs<'_>, variables: &SelectionVariables) -> Vec<Constraint>; // was todo!()
// removed: struct ExecutionContextVariables (merged into SelectionVariables)

cost/src/execution_demand.rs

/// Signature unchanged; a not-materialized node is computed once per evaluation grid of its queries
/// (multiplier = number of contexts); a query-time AcrossEvaluations chain at 1000 / width.
pub fn computations(dag, recurrence, range_evaluations, non_repeating_query_horizon_seconds, root_rates)
    -> Result<HashMap<PhysicalASAPNodeId, (u64, f64)>, AnalyticalCostError>;

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

// Two queries over the last hour of 1,000 latency series (1 sample/s), rank error 0.01:
// p99 every 10 min and p50 every 30 min, both at phase 0: two evaluation contexts (grids 600 s and 1800 s).
let workload = workload_every(&[
    ("quantile_over_time(0.99, latency_ms[60m])", 600_000),
    ("quantile_over_time(0.5, latency_ms[60m])", 1_800_000),
]);
// The deployment keeps no raw data, so Stage 2 prunes query-time scans.
let capabilities = DeploymentCapabilities { raw_data_retained: false, ..Default::default() };
// Frontend → Stages 0–2 → select_plan: ExecutionContexts::new finds κ600, κ1800 and Materialized(chain);
// the chain's input scan runs in the ingestion pipeline of the chain's window; (R) gives h_chain = 3,600 s.
let (plan, _) = plan_with_report_for(workload, &capabilities).await;
// plan.dag (asserted): one SummaryTumblingWindowChain (AcrossEvaluations, retention_ms 3_600_000),
// read by two TimeRange(3600s) nodes, one per context, and two distinct roots.

Major algorithm changes

Selection over execution contexts. Problem: PR 6 ran a candidate once at the summed rate of every query reaching its group: two queries on one grid that share a not-materialized scan were costed twice, and one query every 10 min plus one every 30 min could not run a not-materialized candidate once per grid. Before: $x_v, y_{u,c,i}$ over candidates, rates per candidate, retention post-solve. After: How 1–4. Correctness: model tests — one-consumer rule per context (infeasible without OneEvaluation when both queries are on one grid), a None scan runs once per context on two grids (a_result_that_is_not_materialized_runs_once_per_context_that_reads_it), a materialized chain runs once and $h$ = 300 s for consumers reading 1 m and 5 m (a_materialized_chain_runs_once_and_retains_what_its_consumers_read_back), (E) on a cycle; the planner test above end to end. Complexity: executions ≤ |C| × |contexts|; one node_usage per candidate, per execution, per context edge and per retaining candidate (one more); MILP variables |executions| + |context edges| + Σ|G_q| + |retaining|. Limits: see Spec questions.

Chain retention need (fix). Problem: retention_need_ms read one window back through any chain, so a chain rebuilt in each evaluation (ℓ = None) over raw data retained 10 min served a 60 min range (found by the new e2e test; with (R) that plan became the cheapest). After: only an AcrossEvaluations chain reads its newest window; any other passes through to what its consumers read. Test: a_chain_rebuilt_in_each_evaluation_reads_back_what_its_consumers_read (fails on the PR 6 rule: 10 s instead of 300 s).

asap-cost contexts (K1). Before: a None node once per query reaching it; a query-time chain once per request over input rows / retained windows. After: once per evaluation grid; one window per width. Tests: a_not_materialized_node_is_computed_once_per_evaluation_grid (two 10 s queries + one 30 s: multiplier 2, rate 1/10 + 1/30), a_query_time_chain_builds_one_window_per_width.

Deleted

Item	Why
cost::SelectionCosts, cost::selection_costs	per execution: ExecutionCosts, execution_costs
model::ExecutionContextVariables (#[allow(dead_code)]), the todo!() retention_constraints	merged into SelectionVariables; (R) implemented
per-candidate $x_v$ / unindexed $y_{u,c,i}$	$x_{\kappa,v}$, $y_{\kappa,u,c,i}$
select::evaluation_context, select::delivers_to_consumer	evaluation_grid and the per-context delivers_on walk in ExecutionContexts::new
plan_of's post-solve retention (output_span_ms, retention_need_ms over the selected consumers, its entries parameter)	retention_ms = $h_v$
delivers_on's todo!() for Materialized(v)	implemented

Pending approval (new pub items not in the doc)

1. PhysicalCandidate::chain_width_ms(&self, &PhysicalASAPMemoCandidates) -> Option<u64> — shared by delivers_on (own times of Materialized(v)) and Stage 3 (refresh window, ρ = 1000/w, Δ of π). Example: a SummaryTumblingWindowChain { tumbling_window_width_ms: 10_000, .. } candidate → Some(10_000); a SummaryAgg → None.
2. Crate-private shapes beyond the approved ExecutionContexts { contexts, executions, edges } / ExecutionCosts { per_execution, retention_cost_per_second }: edges holds (consumer execution, edge, producer execution) (the producer's execution depends on the consumer's input context, which differs from its own for ingestion builds and @ subtrees), and ExecutionCosts::per_edge holds $\rho_\kappa c_{\kappa,u,c,i}$ (C1). Example: Q every 10 s reading TimeRange(1m) over chain v (IT, w = 10 s) → edge ((κ10, range), v→range, (Materialized(v), v)); the chain's input edge ((Materialized(v), v), scan→v, (IngestionPipeline({10 s, 0}), scan)).

Spec questions (code kept independent where possible)

• P1 ρ per execution. The doc writes $\rho_\kappa$ per context. Costing every execution in κ at the rate of all of κ's queries would charge a node used only by an instant query at a range query's 61 evaluations per request when both share a grid. The code uses the queries of κ that reach the candidate's group (executions_per_second), which equals what plan_cost charges when every such query uses the node. Recommend: doc says $\rho_{\kappa,v}$ = evaluation rate of κ's queries reaching v.
• P2 @ T horizon. The workload has no horizon end, so need = horizon end − T + range cannot be computed; a producer read through TimeShift { at: T } keeps PR 6's behaviour (no $h_v$, retains everything). A field is needed, e.g. QueryWorkload { horizon_end: Option<TimestampMs>, .. }: with x @ 1700000000 over [5m] and horizon_end = 1700086400000, need = 86,400,000 + 300,000 ms; without it, per the decision, that candidate would not be offered.
• P3 Query-time scan read cost. A scan's executions are still costed reading what it retains at its largest need, in every context (PR 6 watch-out "move scan read onto edges"); only its raw data on disk is in $\pi_v h_v$. Moving the read onto the edges changes asap-cost's scan model; recommend a follow-up.
• P4 Unbounded consumers. A candidate one of whose selectable consumers reads it unbounded (a Scan read whole, @) has no $h_v$ and is costed retaining everything, even when that consumer is not selected.
• P5 asap-cost and @ subtrees. computations does not know a range query's @ end() subtree runs once per request; it charges it per step. Stage 3 charges once per request.
• S1 (ingestion-time None producers exempt from the one-consumer rule) unchanged, pending on refactor(stage3): baseline selection MILP and its PhysicalASAPDAG; executor θ, native chains and a time-indexed store (#1207 PR 6) #1236.

Before this PR

One run per candidate at the summed rate of all reaching queries; retention set after solving from the selected consumers; a None chain over raw data could be planned retaining one window for a 60 min range.

After this PR

Each candidate runs once per context that needs it (once in total when materialized); retention is chosen by the MILP under (R) and costed in the objective; the plan carries one node per execution.

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. Wall-clock tests stay ignored.

Crate	PR 6 (e53cf6e)	PR 7 (6e46d1f)
asap-types	204/0/0	205/0/0
asap-cost	33/0/0	35/0/0
asap-stage3-plan-selection	6/0/0	10/0/0
asap-planner	25/4/1	26/4/1
asap-devtools	37/56/3	37/56/3
asap-integration-tests	135/15/4	135/15/4
others	unchanged	unchanged
Total	1294/75/9	1302/75/9

By test name vs PR 6 e53cf6e: 0 newly failing, 9 newly passing (below plus the cost pin). Vs the previous PR 7 head 3b831f9 (1301/74/9): newly passing the cost pin and PR 6's merge pin; newly failing asap-planner installable::tests::view_refreshes_compile_only_the_nodes_below_them, as in PR 6 (its old plan held raw rows at ingestion that the deployment does not retain; PR 5 now prunes that Scan).

Earlier, vs PR 6 7d1bd55: 8 new tests pass (delivers_on_the_own_times_of_a_materialized_computation; asap-cost a_not_materialized_node_is_computed_once_per_evaluation_grid, a_query_time_chain_builds_one_window_per_width; Stage 3 a_query_time_result_that_is_not_materialized_has_one_consumer_per_context (renamed from ..._has_one_consumer), a_result_that_is_not_materialized_runs_once_per_context_that_reads_it, a_materialized_chain_runs_once_and_retains_what_its_consumers_read_back, a_chain_rebuilt_in_each_evaluation_reads_back_what_its_consumers_read; planner runtime_install::one_chain_serves_two_evaluation_contexts). The 74 failures are PR 6's, with the same causes (devtools PR 9 todos; plan-shape/accuracy pins; fixed_interval without phase, S4 — contexts do not change those, the queries have no known grid; executor gap 12).

Self-review: no shims or aliases; one retention_need_ms for costing and (R); evaluation_grid moved, not copied; the per-candidate cost path, PR 6's post-solve retention and the #[allow(dead_code)] context variables are deleted. Over-abstraction check: Execution and ExecutionContexts are the approved shapes; the triple in edges and per_edge are the only additions (Pending 2). Not done here: scan read cost on edges (P3).