Refresh ingestion-time views that share a refresh window in one run, and charge a shared stateless ingestion node once per run #1151

Example

One metric data with 1,000 series, one sample per series per second (1,000 rows per second), and two queries that both read 10 s panes (pane width = gcd(window, interval), #1095 §2.6):

q1: quantile_over_time(0.9,  data[1m])   # every 10 s, max_error 0.01, rank error
q2: quantile_over_time(0.99, data[5m])   # every 10 s, max_error 0.01, relative error

Take a plan that keeps a KLL pane chain for q1 and a DDSketch pane chain for q2, both maintained at ingestion time with 10 s panes, both over the same ingestion Scan(data):

            q1                      q2
            │                       │
      merge 6 panes            merge 30 panes
            │                       │
   [KLL PaneChain 10 s]    [DDSketch PaneChain 10 s]     MatAtIngestion
             ╲                     ╱
              ╲                   ╱
               [ Scan(data) ]                            NotMaterialized, ingestion time

The numbers are derived from the code paths named below, not from a run. With a 10 s KLL chain and a 60 s DDSketch chain instead, the two views need different runs (the 10 s view needs five pane ends the 60 s view does not), so the Scan stays (2, 2.0/s) and two reads, before and after.

Why it happens

• Execution. The executor already shares a node between roots within one execute call (crates/executor/src/runtime/mod.rs execute/build: one producer per node id; tests runtime/tests.rs shared_source_backpressure_and_reader_drop, diamond_and_run_isolation), and compile takes several roots. But the deployment contract refreshes each view in its own run: docs/design_docs/physical-planning-and-deployment.md §4 running example (refresh = compile(dag, [], [pane_0])) and the module doc of crates/types/src/ir/materialized_view.rs ("compiles the DAG with the executor for that view's node"). The Planner's stand-in runtimes do the same: crates/planner/tests/runtime_install.rs (for view in &views … run(dag, …, &[view.node_id], scope)) and crates/integration-tests/tests/materialization_e2e.rs (one compile(&dag, &[], &[view.node_id]) per view). So does the backend data plane, one builder and one compiled DAG per summary definition.
• Cost. computations (crates/cost/src/execution_demand.rs) counts a NotMaterialized node once per consumer (consumers), and each MatAtIngestion consumer adds rate 1.0. That matches per-view runs. [Feature] Stage 3: no execution-scope copies, exact candidate-level MILP, one cost formula #1095 §2.5 O3(b) assumes one ingestion pipeline I that runs a stateless ingestion candidate once whoever reads it, which matches neither today's execution nor the 10 s + 60 s case; §8 G5 lists the mismatch.

When two views can share a run

An ingestion run over (window_start, window_end] builds, for every root, the windows that end at window_end: a windowed raw Scan reads (window_start, window_end], and each TimeRange(w) keeps (window_end - w, window_end]. One view whose windows end at different offsets is already refused (IngestionBuildReadsWindowsOfDifferentOffsets). Two AtIngestion views can be built in one run, with the run's interval equal to each one's window, exactly when they have the same refresh window:

/// The windows an ingestion run refreshes for a view: width `width_ms`, ending
/// `phase_ms` (mod `width_ms`) after a multiple of `width_ms`, relative to the
/// evaluation time. Derived from the DAG.
pub struct RefreshWindow { pub width_ms: u64, pub phase_ms: u64 }

• For an installed view: its view_window (end_offset o, width W) gives RefreshWindow { W, o mod W }. All panes of a chain share their newest pane's refresh window.
• For a MatAtIngestion PaneChain on the unexpanded DAG: { pane_width_ms, s mod pane_width_ms }, s = the shifts between the chain and its Scan.
• Source, summary family and medium do not matter. A view with an unbounded window is its own group.
• A refresh group is the set of AtIngestion views with one refresh window. Every pane chain Pass 2 builds today has phase 0, so in practice a refresh group is "same pane width".

The key is computed from the PhysicalASAPDAG alone (time ranges, shifts, pane widths), like materialized_views() and retention. Sealing, lateness, threading and the grid's epoch alignment stay the deployment's.

Proposal

1. Deployment contract

• physical-planning-and-deployment.md §4 and the materialized_view.rs module doc: a runtime refreshes the AtIngestion views of one refresh group in one ingestion run, compiled once with all of them as roots and only views of other groups (or views the group reads) as materialized-view slots. A view of the group read by another view of the group is computed in the run, not read as a slot. The running example shows the KLL and DDSketch chains refreshed in one run.
• The plan's cost assumes this grouping. A runtime that splits a group (for example, sources that seal at different times) computes the same states at a higher cost than priced, the same status pane reuse has today.
• New derived API in asap-types, next to materialized_views(): PhysicalASAPDAG::refresh_window(node) -> Option<RefreshWindow> and PhysicalASAPDAG::ingestion_refresh_groups() -> Vec<IngestionRefreshGroup { refresh_window, views }> in dependency order; InstallablePlan::ingestion_refresh_groups() delegates. Nothing new in the plan file, so no wire-version bump.
• InstallablePlan's compile check (executor_compiles) compiles each group's refresh.

2. Cost rule

In computations, for a NotMaterialized node, group its MatAtIngestion consumers by refresh window: each distinct refresh window adds 1 to the multiplier and 1.0 to the rate. Query consumers and query-time materialized consumers are unchanged. plan_cost and node_usage take the new (multiplier, per_second) as they are; ingestion CPU, scanned bytes, peak working memory and ingestion cores follow.

The current Stage 3 engine has its own copy of the per-consumer rule: in crates/plan-selection/src/search/milp/general_schedule.rs, an unmaterialized node gets one demand per ingestion consumer. With a certified direct profile, validate_observations (milp/model.rs) rejects an observation whose evaluated cost is below the modeled bound, so the old engine must change in the same PR: one variable per (unmaterialized ingestion node, refresh window), bounded below by each consumer's both(reached, used) in that window and charged once. The integration-test helper planner_layering_common::computations, which asserts the old multipliers, and the cost-model doc at the top of crates/plan-selection/src/lib.rs change too.

3. Stage 3 MILP (#1095)

Replace the single ingestion context I by one context I_r per refresh window r of the ingestion-time shareable candidates. r(m) is a constant per candidate (§2.6 already makes each pane width its own group).

• O3(b): one ingestion pipeline I_r per refresh window.
• (2): at most one candidate per group in each I_r.
• (3): a boundary read p[m,g,c] ≤ p[I_{r(m)},g,c].
• (4): p[I_r,g,c'] ≤ Σ_{m : r(m)=r} p[m,g,c'].
• Cost: e(c) on each p[I_r,g,c] at rate 1. A scan read by chains of two refresh windows is charged twice, as the evaluator does.
• G5 is resolved by the rule; the prototype's (1, 1.0) hook in computations is dropped, because it under-charges two chains of different widths (e.g. the 1 s and 30 s chains of random-mixed-seed-862-w01-q3-q8-mvp).
• The p variables of ingestion groups multiply by the number of refresh windows; those groups are only Scan, Filter and Project.

The prototype (#1144) adopts this in a follow-up step of #1095: ContextKind::Ingestion(RefreshWindow), one ingestion context per window, and the (1, 1.0) hook removed.

Alternative not chosen

A deployment that updates states sample by sample passes each sample once whatever the pane widths. For it, one charge per ingestion node whatever the windows (the G5 hook) would be right. We choose the batch-run contract, which the executor and the stand-in runtimes implement. Under it, a per-sample deployment runs cheaper than priced, never more expensive.

Scope and acceptance

In scope: the doc contract, the derived refresh-group API, the compile check, the Planner's stand-in runtimes, the cost rule, the current Stage 3 engine's copy of it, the #1095 text. Out of scope: sharing runs across different widths, the prototype MILP code (tracked in #1095), the backend data plane (a follow-up in its repository, after its re-pin to current Planner main).

Done when:

• refresh_window of an unexpanded MatAtIngestion chain equals that of each expanded pane; 10 s KLL and 10 s DDSketch panes are one group, a 60 s chain another, an unbounded ingestion view alone.
• A stand-in runtime test with a counting DataStore (as crates/executor/tests/raw_scan.rs CountingStore): KLL and DDSketch chains over one Scan make one read_raw_data call per pane (two today), store the same states as per-view runs, and answer q1 and q2 unchanged.
• computations: a shared Scan under two equal-width ingestion chains is (1, 1.0) (fails today with (2, 2.0)); under 10 s and 60 s chains it stays (2, 2.0); query consumers still count once each.
• The current Stage 3 engine's modeled cost equals plan_cost on the two-chain case (a direct_price_options_tests case).
• corpus_snapshots and sketch_plan_snapshots regenerated. Expected: no cost diff for the selected plans. In the committed snapshots no selected node has multiplier > 1 at ingestion: o11y_bench/q19 and q25 have two 60 s chains but on two Scans with different predicates. Any plan change is listed in the PR.

PR plan

1. Docs: physical-planning-and-deployment.md §4 (contract, running example with two chains, responsibility table) and the materialized_view.rs module doc.
2. Refresh groups: RefreshWindow, refresh_window, ingestion_refresh_groups in asap-types; InstallablePlan::ingestion_refresh_groups; executor_compiles compiles each group; runtime_install.rs and materialization_e2e.rs refresh per group; the counting-store test. No cost change.
3. Cost rule: computations by refresh window; general_schedule.rs OR variable per (node, refresh window); test helper; cost-model doc in plan-selection/src/lib.rs; snapshots regenerated. One PR, because the certified bound check fails if the rule and the engine disagree.
4. [Feature] Stage 3: no execution-scope copies, exact candidate-level MILP, one cost formula #1095 text: O3(b), §3 rows (2)–(4) and the cost row, G5 closed.
5. Backend follow-up (other repository, filed there): one compiled DAG and one sample buffer per refresh group.

Status and related work

• Decided: change the deployment contract and the cost rule together.
• Related: [Feature] Stage 3: no execution-scope copies, exact candidate-level MILP, one cost formula #1095 (§2.5 O3(b), §2.6 pane widths, §8 G5), [Refactor] Pane-width rounds without cross-stage calls: rerun Pass 2 per round, or pass an incremental Pass 2 state #1048 (pane-width rounds), Share one scan of a source across readers with different predicates, labels and time ranges (subsumption derivations) #951 (shared scans; merging the two Scans of q19/q25 would make them share a run too), feat(plan-selection): group selection MILP over execution contexts as Stage 3's default search (#1095) #1144 (prototype).

🤖 Generated with Claude Code