[zzylol]

Why

Part of #1151, stacked on #1153 (refresh groups) and #1152 (contract). The runtime now refreshes the views of one refresh window in one ingestion run, so a Scan under a KLL chain and a DDSketch chain of 10 s panes is computed once per pane. asap_cost::execution_demand::computations still charges it once per MatAtIngestion consumer: (multiplier 2, 2.0/s), detail scan … x 2.0000/s. The old Stage 3 engine has its own copy of the per-consumer rule. Its certified bound (validate_observations) rejects an observation whose evaluated cost is below the modeled bound, so the rule and the engine must change together, in one PR.

What

• Cost rule (asap-cost): in computations, the MatAtIngestion consumers of a NotMaterialized node that share one refresh window (PhysicalASAPDAG::refresh_window) count as one consumer: +1 to the multiplier and +1.0/s to the rate.
  • A consumer with an unbounded window counts alone.
  • Query consumers and query-time materialized consumers are unchanged.
  • plan_cost / node_usage take the new (multiplier, per_second) unchanged, so ingestion CPU, scanned bytes, peak working memory and ingestion cores follow.
• Old Stage 3 engine (asap-plan-selection):
  • NodeCostProfile gets refresh_window, set for a pane chain maintained at ingestion.
  • All three places that count an unmaterialized node per consumer now count once per (node, refresh window):
    • DirectCostProfile::selected_cost, the modeled cost.
    • The equal-schedule encoder in direct_cost::encode.
    • The mixed-schedule encoder general_schedule::encode.
  • Each encoder adds one variable per (unmaterialized ingestion node, refresh window), bounded below by each use from that window and charged once at rate 1.
• The cost-model doc at the top of crates/plan-selection/src/lib.rs states the rule. So do NodeCost::multiplier and ResourceUsage::peak_memory_bytes in crates/cost/src/lib.rs, and, moved here from docs: refresh the ingestion-time views of one refresh window in one run (#1151) #1152, the "Refresh groups" paragraph of physical-planning-and-deployment.md §4 and the materialized_view.rs module doc, which say how Stage 3 prices the grouping.
• The engine builds a chain's refresh window with the shared RefreshWindow::of_window from asap-types, so a width-0 chain gets no window instead of a bad one.
• The integration-test helper planner_layering_common::computations follows the rule.
• RefreshWindow derives PartialOrd, Ord, so the engine keys its variables in a BTreeMap and creates them in a deterministic order.

How

computations computes dag.refresh_windows(MatAtIngestion nodes) once. While it adds up each node's consumers, it skips a MatAtIngestion consumer whose refresh window has already been counted for that node.

In the engine, a boundary maintained at ingestion now carries its refresh window. Its uses of an unmaterialized node go into runs[(node, window)] instead of one demand each. Each entry becomes one variable run ∈ [0, 1] with run ≥ use for every use, and run takes the place of those demands in the objective.

The direct profile admits no TimeShift at ingestion, so a chain's newest pane ends at the evaluation time, and its refresh window is (pane_width_ms, 0).

Key code interfaces added / modified

crates/cost/src/execution_demand.rs

/// ... A `NotMaterialized` node is computed once per consumer that reaches
/// it ([`consumers`]), at that consumer's rate. ... The `MatAtIngestion`
/// consumers of one refresh window ([`PhysicalASAPDAG::refresh_window`]) are
/// refreshed in one ingestion run, which computes the node once for all of
/// them: they count as one consumer (a consumer with an unbounded window
/// counts alone). Its `multiplier` is that number of consumers.
/// // signature unchanged, rule changed (excerpt of the doc)
pub fn computations(
    dag: &PhysicalASAPDAG,
    recurrence: &[QueryRecurrence],
    range_evaluations: &[Option<RangeEvaluation>],
    non_repeating_query_horizon_seconds: f64,
    root_rates: Option<&[f64]>,
) -> Result<HashMap<PhysicalASAPNodeId, (u64, f64)>, AnalyticalCostError>;

crates/types/src/ir/materialized_view.rs

/// The windows an ingestion run refreshes for a view
/// ([`PhysicalASAPDAG::refresh_window`]): ... (doc unchanged)
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] // was: without PartialOrd, Ord
pub struct RefreshWindow {
    /// Width of each window.
    pub width_ms: u64, // unchanged
    /// Where each window ends, modulo `width_ms`.
    pub phase_ms: u64, // unchanged
}

crates/plan-selection/src/search/milp/direct_cost.rs (crate-internal; the other fields carry no doc comments in the code)

#[derive(Debug, Clone, Default)]
pub(crate) struct NodeCostProfile { // type stays
    pub per_request: bool, // unchanged
    pub execution_cost: f64, // unchanged
    pub ingestion: bool, // unchanged
    pub standing_cost_per_second: f64, // unchanged
    pub work_seconds: f64, // unchanged
    /// Certified fixed-worker merge delay; the query pays the maximum reached delay.
    pub compaction_seconds: f64, // unchanged
    pub held_memory_cost_per_second: f64, // unchanged
    pub materialized: bool, // unchanged
    pub raw_retention: Option<(String, f64)>, // unchanged
    /// For a node maintained at ingestion, the refresh window of its
    /// ingestion runs: one run computes an unmaterialized node below the
    /// nodes of one refresh window once for all of them.
    pub refresh_window: Option<RefreshWindow>, // new; set in direct_profile.rs from RefreshWindow::of_window(0, pane_width_ms)
}

crates/integration-tests/tests/planner_layering_common/mod.rs

/// How many times each node is computed per evaluation of its consumers ...
/// The `MatAtIngestion` consumers of one refresh window share one ingestion
/// run and count once (#1151).
/// // signature unchanged, rule changed (excerpt of the doc)
pub fn computations(dag: &PhysicalASAPDAG) -> BTreeMap<NodeId, u64>;

Usage (crates/cost/tests/ingestion_refresh_runs.rs, run at this commit):

// One Scan of `m`; a KLL chain (6 × 10 s panes) and a DDSketch chain (30 × 10 s
// panes) over it, both MatAtIngestion, each merged by one query every 10 s:
// the unexpanded DAG Stage 3 costs.
let dag = dag_of(&[chain(&scan, Kll, 10_000, 6), chain(&scan, DDSketch, 10_000, 30)], AT_INGESTION);
// Both chains have refresh window (10 000 ms, 0): one ingestion run per pane.
let times = computations(&dag, &[every_10_s; 2], &[], 3600.0, None)?;
assert_eq!(times[&scan_id], (1, 1.0)); // was (2, 2.0)
// With a 60 s DDSketch chain instead, two refresh windows, two runs: (2, 2.0).
// Stage 3 then prices the Scan's ingestion CPU and scanned bytes at 1.0/s,
// and the old engine's modeled cost equals that price (see Verification).

Major algorithm changes

Ingestion demand of a shared stateless node

• Problem. For a Scan under 10 s KLL and 10 s DDSketch chains, computations gave (2, 2.0), so Stage 3 priced the Scan twice. The runtime (feat: derive ingestion refresh groups and refresh each group in one run (#1151) #1153) runs it once per pane. The old engine modeled the same double charge.
• Before / after.
  • Before: computations gave each unmaterialized node +1, +rate per consumer. The engine pushed one demand both(reached, used) per consumer, at rate 1.0 for an ingestion consumer.
  • After:
    1. Group a node's MatAtIngestion consumers by refresh_window.
    2. Each group adds +1, +1.0. Every other consumer adds as before.
    3. The engine gives each (node, window) one variable y ≥ use_c for each consumer c of that window, charged execution_cost × 1.0 × y.
• Correctness.
  • Minimization drives y down to max_c use_c, the OR of the uses, so the objective equals selected_cost. selected_cost applies the same rule as computations.
  • direct_profile::tests::chains_of_one_refresh_window_share_their_scan_in_the_direct_profile checks modeled = evaluator (within 1e-9) and the Scan multiplier on three cases: equal schedules with 10 s panes (multiplier 1), mixed schedules every 20 s and every 40 s with 20 s panes (1, general encoder), and 10 s + 60 s panes (2).
  • lower_bounds.rs prices with (1, 1.0), so it is still a lower bound.
• Complexity. refresh_windows is O(nodes + edges), once per computations call. The engine adds at most one variable per (unmaterialized ingestion candidate, refresh window) and one constraint per use, and drops the same number of per-consumer demand terms. Ingestion groups are only Scan, Filter, Project and TimeRange.
• Assumptions and limits.
  • The batch-run contract of docs: refresh the ingestion-time views of one refresh window in one run (#1151) #1152. A runtime that updates states sample by sample runs cheaper than priced.
  • Different widths still pay once each, matching the evaluator.
  • The prototype MILP ([Feature] Stage 3: no execution-scope copies, exact candidate-level MILP, one cost formula #1095 / feat(plan-selection): group selection MILP over execution contexts as Stage 3's default search (#1095) #1144) is not touched here.

Snapshot diffs

Regenerated with UPDATE_SNAPSHOTS=1 cargo test --release -p asap-devtools --test corpus_snapshots -- --ignored. Both corpus_plans_match_snapshots and sketch_plans_match_snapshots passed, and no snapshot file changed: no selected plan, cost or status differs. This matches #1151's expectation. No committed plan has a node with multiplier > 1 at ingestion: the two 60 s chains of o11y_bench/q19 and q25 read two different Scans.

Before this PR

Stage 3 charged a stateless ingestion node once per ingestion consumer. That was right for per-view refreshes, and it over-charged the one run per refresh group that #1152/#1153 introduce.

After this PR

Stage 3 charges such a node once per refresh window of the views above it, in asap-cost and in the old Stage 3 engine alike. Equal-width chains over one source become cheaper together, so the cost of one chain's pane width now depends on the other chains' widths.

Verification

• Failing first:
  • asap-cost without the rule: a_scan_under_two_chains_of_one_pane_width_is_computed_once fails, left: (2, 2.0), right: (1, 1.0).
  • Rule only (the engine without refresh_window): the direct-profile test fails, "direct 0.051930…, evaluator 0.051670…". The modeled cost is above the evaluated one, which validate_observations would reject in production.
  • Engine only (refresh_window set, old rule): the same test fails, "direct 0.051670…, evaluator 0.051930…". The engine models the shared run and the evaluator does not.
  • Neither change: the same test fails on its multiplier assertion (Scan multiplier 2, expected 1).
• cargo test -p asap-cost: all passed, 7 of them new in ingestion_refresh_runs:
  • 10 s + 10 s → (1, 1.0); 10 s + 60 s → (2, 2.0); 10 / 10 / 60 s → (2, 2.0).
  • One width, phases 0 and 5 s → (2, 2.0).
  • Two chains through one shared TimeShift(10 s) → (1, 1.0) for the shift and for the Scan.
  • Unbounded consumers count alone; query consumers count once each.
• cargo test -p asap-plan-selection --lib: 285 passed. structural_seed_timeout_preserves_validated_incumbent (a 1 s wall-clock test) failed once under machine load and passed on rerun.
• cargo test -p asap-integration-tests --test integration -- planner_layering range_evaluation sql_to_physical reuse_rules materialization_e2e: 103 passed, 20 ignored.
• cargo test -p asap-devtools --test devtools stage_pipeline: 48 passed.
• Snapshots: see above; no diffs, also after the review fixes (rerun at 3476adf).
• cargo fmt --all --check and cargo clippy -p asap-types -p asap-cost -p asap-plan-selection -p asap-integration-tests --all-targets -- -D warnings: clean.

Remaining for #1151, tracked separately: the #1095 text (O3(b), rows (2)–(4), G5) and the backend follow-up in its repository.

Note: on the issue's first example (1 s samples), the Planner still picks 1 s panes for the DDSketch chain and 10 s panes for the KLL chain under the new rule. The saved scan does not outweigh the DDSketch pane-width choice there, so that plan has two refresh groups.

🤖 Generated with Claude Code