[zzylol]

Part of #1095

Why

#1151 (merged into feat/1095-stage3 as #1152–#1154) changed the cost rule for stateless ingestion work. The deployment refreshes all ingestion-time views of one refresh window in one run. So asap_cost::execution_demand::computations now charges a NotMaterialized ingestion-time node once per distinct refresh window of its MatAtIngestion consumers. A consumer with an unbounded window counts alone. #1095 §2.5 O3(b) was rewritten to match: one ingestion pipeline I_r per refresh window r.

The prototype (#1144) still had one ingestion context I. To keep its objective equal to plan_cost, it also had a hook in computations that charged such a node (1, 1.0), whoever read it. That hook under-charges a scan read by chains of different pane widths (§8 G5). This PR adopts I_r and removes the hook.

What

• Merges origin/feat/1095-stage3 (docs: refresh the ingestion-time views of one refresh window in one run (#1151) #1152–feat: charge a shared ingestion node once per refresh window (#1151) #1154). On the conflict in crates/cost/src/execution_demand.rs, it keeps feat: charge a shared ingestion node once per refresh window (#1151) #1154's rule and drops the prototype's hook. retained_panes and maintained_window_ms stay public, since groups.rs still uses them.
• The MILP has one ingestion context per refresh window of the ingestion-time shareable candidates. An ingestion-time shareable candidate whose window is unbounded or empty gets a context of its own.
• Rows, per [Feature] Stage 3: no execution-scope copies, exact candidate-level MILP, one cost formula #1095 §3:
  • row 3 (boundary read): p[m,g,c] ≤ p[I_{r(m)},g,c];
  • row 4: p[I_r,g,c'] ≤ Σ_{m: r(m)=r} p[m,g,c'];
  • cost: a candidate of I_r is charged e(c) once per I_r that selects it, at rate 1.
• Decode, warm start, the cross-context acyclicity ranks and the LP diagnosis labels resolve a read of an ingestion-time candidate to the I_r of the reading context.

How

r(m) must be computed exactly as asap-cost computes it on the realized DAG. So groups::refresh_windows builds a nodes-only DAG of the candidates below each ingestion-time shareable candidate, reading each input from its first producer candidate, as catalog_node does. It then calls the same function, PhysicalASAPDAG::refresh_windows. For that to work on a DAG without edges, PhysicalASAPDAG::view_windows now walks each operator's inputs (payload.children()) instead of the edge list. validate requires these to be identical in a valid DAG, so its results on real DAGs do not change.

execution_contexts puts a candidate of I_r into the seeds of I_{r(m)} only when the context of an ingestion-time shareable m reads it. Stage 2's compatible_timing lets only ingestion-time candidates read a NotMaterialized ingestion-time candidate. Each I_r is then expanded through the candidates it computes, as I was. Contexts are ordered: queries, then the I_r (by IngestionPipeline), then the shareable contexts. MilpInputs::ingestion_context_of_shareable records m → I_{r(m)}.

Key code interfaces added / modified

crates/plan-selection/src/groups.rs

/// Which execution context (#1095 §2.0). Stays.
pub(crate) enum ContextKind {
    /// A query evaluation. // unchanged
    Query(usize),
    /// An ingestion pipeline `I_r`. // was `Ingestion` (one `I`)
    Ingestion(IngestionPipeline),
    /// A shareable candidate's own inputs. // unchanged
    Shared(PhysicalCandidateId),
}

/// Which ingestion pipeline `I_r` (§2.5 O3(b)). New.
pub(crate) enum IngestionPipeline {
    /// The ingestion-time views of this refresh window. // new
    Window(RefreshWindow),
    /// An ingestion-time shareable candidate whose window is unbounded or empty: refreshed alone. // new
    Alone(PhysicalCandidateId),
}

/// The pipeline of ingestion-time shareable `shareable`, from its refresh window. // new
pub(crate) fn ingestion_pipeline(
    shareable: PhysicalCandidateId,
    refresh_windows: &HashMap<PhysicalCandidateId, Option<RefreshWindow>>,
) -> IngestionPipeline;

/// What the MILP is built from. Stays; fields not listed are unchanged.
pub(crate) struct MilpInputs {
    /// Queries, then the `I_r` some context reads, then the shareable contexts. // was: queries, `I` at index Q, shareable
    pub(crate) contexts: Vec<ExecutionContext>,
    /// Per shareable candidate with a context, its index in `contexts`. // unchanged
    pub(crate) context_of_shareable: BTreeMap<PhysicalCandidateId, usize>,
    /// Per ingestion-time shareable `m` whose context reads a candidate of an `I_r`: the index of `I_{r(m)}`. // new
    pub(crate) ingestion_context_of_shareable: BTreeMap<PhysicalCandidateId, usize>,
    /* every other field unchanged and not shown */
}

impl MilpInputs {
    /// Where `context` reads a candidate of an `I_r`: itself for an `I_r`, `I_{r(m)}` for ingestion-time `m`'s context, else `None`. // new
    pub(crate) fn ingestion_context_read_by(&self, context: usize) -> Option<usize>;
}

/// `execution_contexts`' result. New.
pub(crate) struct ExecutionContexts {
    /// See `MilpInputs::contexts`. // new
    pub(crate) contexts: Vec<ExecutionContext>,
    /// See `MilpInputs::context_of_shareable`. // new
    pub(crate) context_of_shareable: BTreeMap<PhysicalCandidateId, usize>,
    /// See `MilpInputs::ingestion_context_of_shareable`. // new
    pub(crate) ingestion_context_of_shareable: BTreeMap<PhysicalCandidateId, usize>,
}

/// The execution contexts and their options. // was: returned `(Vec<ExecutionContext>, BTreeMap<..>)`, no `refresh_windows`
pub(crate) fn execution_contexts(
    physical: &PhysicalASAPMemoCandidates,
    kind: &[RunKind],
    producer_candidates_per_input: &[Vec<Vec<PhysicalCandidateId>>],
    output_candidates: &[Vec<PhysicalCandidateId>],
    refresh_windows: &HashMap<PhysicalCandidateId, Option<RefreshWindow>>, // new
    usable: impl Fn(PhysicalCandidateId) -> bool,
) -> ExecutionContexts;

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

impl SelectionMilp<'_> {
    /// The index of `I` among the contexts. // removed (there is no single `I`)
    pub(crate) fn ingestion_context(&self) -> usize;
    /// `MilpInputs::ingestion_context_read_by`. // new
    pub(crate) fn ingestion_context_read_by(&self, context: usize) -> Option<usize>;
}

crates/types/src/ir/physical_export.rs

impl PhysicalASAPDAG {
    /// Each node's view window. It now walks each operator's inputs (a valid DAG's edges), so it also answers for a DAG of nodes alone. // signature unchanged
    pub fn view_windows(
        &self,
        nodes: impl IntoIterator<Item = PhysicalASAPNodeId>,
    ) -> HashMap<PhysicalASAPNodeId, ScanWindow>;
}

crates/cost/src/execution_demand.rs: no signature change. The hook None if n.output_state.timing == ExecutionTiming::IngestionTime && count.contains_key(&n.id) => (1, 1.0) is removed (merge resolution).

Usage (end to end, as the test ingestion_scans_under_chains_of_two_refresh_windows_are_charged_twice runs it):

// Two approximate quantile queries over metric `m` (100 series, 100 rows/s arriving):
// one every 10 s over 10 s, one every 60 s over 5 m.
let queries = [
    approximate("quantile_over_time(0.5, m[10s])", 10_000, 0.01),
    approximate("quantile_over_time(0.9, m[5m])", 60_000, 0.05),
];
// Stages 1 and 2 as the facade runs them; Stage 2 offers ingestion-time pane chains of many widths.
let fixture = through_stages_with(&queries, true, true);
// Stage 3: prepare_milp_inputs computes r(m) per ingestion-time chain and one I_r per window;
// the test keeps only the 10 s and 60 s chains and forbids query-time scans.
// solve_and_check solves, decodes, realizes through Stage 2 and asserts objective == plan_cost.
let (checked, scans) = ingestion_scans_under_refresh_windows(&queries, &[10_000, 60_000]);
// One ingestion scan per refresh window, each read only by chains of that window:
assert_eq!(scans, vec![BTreeSet::from([10_000]), BTreeSet::from([60_000])]);

With ASAP_STAGE3_SEARCH=milp, stage_pipeline reaches the same code. Its ASAP_STAGE3_REPORT lines report the model size; the Ingestion scope now holds every I_r.

Major algorithm changes

Ingestion contexts per refresh window (§2.5 O3(b), §3 rows 2–4, §3.5)

• Problem. With one I, a scan read by a 1 s chain and a 30 s chain ran once in I, and the MILP charged it once. Under feat: charge a shared ingestion node once per refresh window (#1151) #1154, plan_cost charges it once per refresh window. Before this PR, the hook in computations hid the difference by under-charging the plan. On random-mixed-seed-862 workload-01, the old selected plan read one ingestion scan from chains of 1 s, 20 s and 30 s panes (node 51, 8.32e-9 per second). It was charged once instead of three times.
• Before. 1. execution_contexts seeds one I with every ingestion-time candidate read across a boundary. 2. Row 3: p[ctx,c] ≤ p[I,c]. 3. Row 4 for I: read by any other context. 4. computations charged each such node (1, 1.0).
• After. 1. refresh_windows computes r(m) per ingestion-time shareable candidate with PhysicalASAPDAG::refresh_windows. 2. execution_contexts seeds I_{r(m)} from m's context only (Alone(m) for an unbounded window). 3. Row 3: p[m,c] ≤ p[I_{r(m)},c]. 4. Row 4: a candidate of I_r is read by the contexts of the m with r(m) = r. 5. Each I_r run is charged at rate 1. computations applies feat: charge a shared ingestion node once per refresh window (#1151) #1154's rule unchanged.
• Correctness. A node of I_r in the decoded DAG is read only by the m with r(m) = r, so computations counts it once. The same candidate selected in two I_r is two nodes, each counted once. r(m) comes from the same function computations calls. This holds as long as the candidates of one memo group read one window (see limits). The per-solve check (objective = plan_cost) passes on every test and on all seven §6.1 workloads.
• Complexity. Contexts grow from Q + 1 + |M| to Q + |distinct r| + |M|. A candidate read by chains of k windows gets up to k p columns instead of 1. Measured: +0 to +4 % variables (workload-01's largest round: 39,907 → 40,251). refresh_windows is linear in the candidates below the ingestion-time states, built once per prepare_milp_inputs_with.
• Assumptions and limits. r(m) is computed with each input read from its first producer candidate. That is exact when every producer candidate of an input reads one view window. refresh_windows debug-asserts this (producer_window_disagreements), because the per-solve equality check alone would miss a mismatch in a model with inexact terms (it reports InexactObjective). Tests check it on hand-built and real Stage 1–2 fixtures; no counterexample is known (a widened time range would be one). Only ingestion-time shareable contexts read a candidate of an I_r. If some other context read one, it would have no source and row 3 would force the read to 0 (Stage 2 never links one).

Before this PR

The prototype modeled one ingestion pipeline. It relied on a hook in asap-cost that charged stateless ingestion work once whoever read it, so plans sharing a scan across chains of different pane widths were under-priced (§8 G5).

After this PR

The MILP and plan_cost both use #1151's rule (one run per refresh window), and the hook is gone. A scan shared by chains of one refresh window is charged once; chains of k windows pay it k times. The solver now sees that cost and chooses between sharing and separate scans with it.

Verification

• New tests in milp/model/tests.rs, each with the per-solve objective = plan_cost check:
  • one_ingestion_scan_under_chains_of_one_refresh_window_is_charged_once: two 10 s chains → one scan;
  • ingestion_scans_under_chains_of_two_refresh_windows_are_charged_twice: 10 s + 60 s chains → two scans, one per window.
  • Checked by mutation: with every window mapped to one pipeline, the second test fails with MILP objective 1.1670e-1 != plan_cost 1.1696e-1.
• Review follow-ups (222fbd3):
  • execution_contexts_have_one_ingestion_pipeline_per_refresh_window (hand-built): two views of (10 s, 0) share one I_r; (60 s, 0) and (10 s, 5 s) get their own (the shifted one goes through a NotMaterialized shift and range); an unbounded view gets Alone.
  • producers_of_an_input_read_one_view_window (hand-built) and producers_of_an_input_read_one_view_window_through_stages (three Stage 1–2 fixtures).
  • asap-types a_view_has_its_refresh_window_on_a_dag_without_edges: a non-chain KLL over TimeRange(1 min) shifted 15 s has window (60 s, 15 s) with and without edges. It fails when the view_windows change is reverted.
  • Behavior is unchanged: row 4 now uses an inverse map I_r → contexts, the debug-assert is debug-only, and the label/wording changes are cosmetic. So no re-measurement.
• cargo test -p asap-plan-selection --lib: 322 passed. cargo test -p asap-cost, cargo test -p asap-types: pass. cargo test -p asap-devtools --test devtools stage_pipeline: 51 passed.
• cargo fmt --all; cargo clippy -p asap-types -p asap-cost -p asap-plan-selection --all-targets -- -D warnings and cargo clippy -p asap-plan-selection --all-targets --features lp-diagnosis -- -D warnings: clean.
• §6.1 re-measure (stage_pipeline, ASAP_STAGE3_SEARCH=milp, aws-fargate cost model, accuracy enforced, 300 s limit). Before = feat(plan-selection): selection MILP solved per independent part, LP gap diagnosis, tighter big-M bounds (#1095 §5 3a, §6.1) #1155 head; after = this PR:

Workload	Vars / binaries (before)	Build	Solve	Status, gap	Cost per second	plan_cost = objective	Wall	Peak RSS
example1	525 / 347 (501)	0.07 → 0.07 s	0.03 → 0.03 s	optimal → optimal	4.3770e+00 → 4.3770e+00	yes	2 → 2 s	54 → 53 MB
rangeheavy-t5-q05-q09	7,041 / 3,967 (6,982)	0.15 → 0.15 s	1.63 → 1.61 s	optimal → optimal	3.9901e-06 → 3.9901e-06	yes	38 → 38 s	119 → 116 MB
rangeheavy-t7-t4	4,667 / 2,804 (4,490)	0.03 → 0.03 s	2.49 → 2.05 s	optimal → optimal	2.1493e-07 → 2.1493e-07	yes	29 → 29 s	105 → 106 MB
rangeheavy-t8-t5	4,013 / 1,749	0.03 → 0.03 s	1.46 → 1.47 s	optimal → optimal	6.4822e-06 → 6.4822e-06	yes	330 → 323 s	301 → 298 MB
largest round (5×)	114,284 / 50,660	44.3 → 43.0 s	0.54 → 0.52 s	deadline, no dual bound (both)	incumbent 2.0661e-05 → 2.0661e-05
ratio-of-quantiles	52 / 29	0.00 → 0.00 s	0.05 → 0.05 s	optimal → optimal	2.4071e-06 → 2.4071e-06	yes	1 → 1 s	35 → 35 MB
rollupheavy-t11	8,847 / 5,049 (8,788)	0.17 → 0.17 s	3.69 → 3.64 s	optimal → optimal	1.4159e-06 → 1.4159e-06	yes	62 → 63 s	145 → 148 MB
workload-01 (selected round)	3,866 / 1,291 (39,907)	0.25 → 0.03 s	191.51 → 4.79 s	optimal → deadline, 5.8 %	1.4171e-06 → 1.4402e-06	yes	292 → 300 s	896 → 892 MB
largest round	40,251 / 13,491 (39,907)	0.2 → 0.3 s	191.51 → 200.44 s	optimal → deadline, 41.2 %	incumbent 1.4171e-06 → 2.2492e-06

Six workloads select the same plan at the same cost. Four of them have small variable increases from the extra I_r contexts. Example1, ratio-of-quantiles and rollupheavy-t11 have no ingestion-time node in the plan; t5-q05-q09, t7-t4 and t8-t5 have one chain over its own scan.

workload-01 (random-mixed-seed-862) changes by +2.31e-8 per second (+1.6 %). Two causes:

1. The old plan was under-charged by 1.66e-8. Its ingestion scan (node 51, 8.32e-9 per second) was read by four chains of three refresh windows: Sum at 1 s panes, Increase at 20 s, and Sum and Rate at 30 s. The hook charged it once; feat: charge a shared ingestion node once per refresh window (#1151) #1154 charges it three times (+2 × 8.32e-9). So under the current rule, the old plan costs ≈ 1.4337e-06.
2. The largest round no longer solves in time. Before, it was optimal just under the deadline (191.5 s). After, it has more variables and stops at a 41.2 % gap, so the selected plan is round 1's incumbent (5.8 % gap). That plan uses one scan per chain: four scans for the 1 s, 20 s and two 30 s chains (the 30 s Rate chain reads a cheaper scan of its own). It is 6.4e-9 above the old plan's corrected cost. With a 600 s limit, the largest round still ends at a 38.9 % gap, and the selection is the same 1.4402e-06.

The selected plan's objective equals plan_cost (checked per solve). Closing workload-01's gap belongs with the §5 steps (3b/3c).

🤖 Generated with Claude Code