[zzylol]

Why

Part of #1151, stacked on #1152 (the deployment contract). With a KLL chain and a DDSketch chain of 10 s panes kept at ingestion over one Scan, the Planner's install check and its stand-in runtimes refresh each pane view in its own ingestion run, so each pane window is read from raw data twice. The contract in #1152 says to refresh the views of one refresh window in one run. This PR adds the derived API that names the groups, compiles each group as one run in the install check, and makes the stand-in runtimes refresh by group. It does not change the cost rule; that comes in the next PR.

What

• asap-types: next to materialized_views(). All are derived from the DAG. Nothing new goes in the plan file, so there is no wire-version bump.
  • RefreshWindow (with the one constructor RefreshWindow::of_window) and IngestionRefreshGroup.
  • PhysicalASAPDAG::refresh_window / refresh_windows.
  • PhysicalASAPDAG::ingestion_refresh_groups and ingestion_refresh_groups_of(views), for a caller that already derived the views.
  • view_windows now counts an unexpanded PaneChain below a node as all its panes, so a node above a chain on the DAG Stage 3 costs has a refresh window. The installed DAG has no chains, so nothing changes for the executor.
• asap-planner: InstallablePlan::ingestion_refresh_groups() delegates to the DAG.
  • The compile check (executor_compiles, behind plan_installable and check_compiles) now compiles one refresh per refresh group, plus one per view refreshed on first read. Before, it compiled one refresh per view.
  • It groups the views it was given (ingestion_refresh_groups_of) instead of deriving them again.
  • A group is compiled with all its views as roots. That is a compile check covering every run's roots, not any one run's root set.
  • A group compile error's reason names every view compiled with it.
• Stand-in runtimes refresh by group:
  • crates/planner/tests/runtime_install.rs: one ingestion run per (group, window end). Its roots are the group's views whose state for that window is not stored yet (the newest pane of each chain once older panes are stored). A view of the group that a root reads is computed in the run.
  • crates/integration-tests/tests/materialization_e2e.rs: one compile per group.
• New tests:
  • Types:
    • A chain and its expanded panes have one refresh window.
    • 10 s KLL and 10 s DDSketch panes are one group, 60 s panes another.
    • A shifted pane gets its own phase.
    • Views with unbounded windows are each a group of their own.
    • A rollup reading a pane of its own group follows it in the group, and reads it as an input view, not a view of another group.
    • A node above an unexpanded chain has the window of all its panes.
  • Runtime: a counting DataStore checks one read_raw_data call per pane window, the same states as per-view runs, and answers equal to the query-time reference.
• Docs: the plan-file block in §4 lists ingestion_refresh_groups, IngestionRefreshGroup and RefreshWindow. The running example calls it. §6 lists the new runtime test.

How

refresh_windows(nodes) gives each node's refresh window:

• For a PaneChain (the unexpanded DAG Stage 3 costs), it is (pane_width_ms, s mod pane_width_ms), where s is the sum of the TimeShift offsets below the chain. That is the newest pane's window.
• For any other node, it is (W, o mod W) of its view_window (T - o - W, T - o].
• It is None when the window is unbounded or empty.

Pane i of a chain reads TimeRange(W) ← TimeShift(i·W + s), so every expanded pane has its chain's refresh window.

ingestion_refresh_groups() works in three steps:

1. Take the views refreshed at ingestion from materialized_views() (dependency order).
2. Group them by refresh window. A None window is a group of its own, and each group keeps dependency order.
3. Order the groups by a depth-first postorder over "group A reads a view of group B". Two groups never read each other: a view's window contains the windows of the views it reads, and they all end where it ends.

Key code interfaces added / modified

crates/types/src/ir/materialized_view.rs

/// The windows an ingestion run refreshes for a view
/// ([`PhysicalASAPDAG::refresh_window`]): those `width_ms` wide that end
/// `phase_ms` (`< width_ms`) after a multiple of `width_ms`, relative to the
/// evaluation time. Views refreshed at ingestion with one refresh window
/// are refreshed together, in one ingestion run per window.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct RefreshWindow { // new
    /// Width of each window.
    pub width_ms: u64, // new
    /// Where each window ends, modulo `width_ms`.
    pub phase_ms: u64, // new
}

impl RefreshWindow {
    /// The refresh window of the window `(T - end_offset_ms - width_ms, T -
    /// end_offset_ms]`: `(width_ms, end_offset_ms mod width_ms)`; `None` for
    /// an empty window (`width_ms <= 0`).
    pub fn of_window(end_offset_ms: i64, width_ms: i64) -> Option<Self>; // new
}

/// The views refreshed at ingestion that share one refresh window
/// ([`PhysicalASAPDAG::ingestion_refresh_groups`]). A runtime refreshes
/// them with one ingestion run per window `(e - W, e]`; each run's roots are
/// the views whose state for the window ending at `e` is not stored yet (in
/// steady state the newest pane of each chain), and a view of the group a
/// root reads is computed in the run.
#[derive(Debug, Clone, PartialEq)]
pub struct IngestionRefreshGroup { // new
    /// The refresh window of every view of the group; `None` for a view
    /// whose window is unbounded or empty, which is a group of its own.
    pub refresh_window: Option<RefreshWindow>, // new
    /// The views, in dependency order.
    pub views: Vec<MaterializedView>, // new
}

impl PhysicalASAPDAG {
    /// The views refreshed at ingestion, by refresh group: the views with
    /// one [`RefreshWindow`] ([`Self::refresh_window`]) form one group,
    /// which a runtime refreshes with one ingestion run per window
    /// ([`IngestionRefreshGroup`]). Each group lists its views in dependency
    /// order and comes after the groups whose views it reads. Errs as
    /// [`Self::materialized_views`].
    pub fn ingestion_refresh_groups(&self) -> Result<Vec<IngestionRefreshGroup>, InstallableDagError>; // new
    /// [`Self::ingestion_refresh_groups`] of `views`, this DAG's
    /// materialized views ([`Self::materialized_views`]), for a caller that
    /// already derived them.
    pub fn ingestion_refresh_groups_of(&self, views: &[MaterializedView]) -> Vec<IngestionRefreshGroup>; // new
    /// The refresh window of `node` ([`RefreshWindow::of_window`]): for a
    /// `PaneChain` (a DAG with its chains unexpanded), its newest pane's,
    /// `(pane_width, s mod pane_width)` with `s` the sum of the offsets of
    /// the `TimeShift`s below it; for any other node, that of its window
    /// ([`Self::view_windows`], which counts a chain below it as all its
    /// panes). A chain and each of its expanded panes have the same one.
    /// `None` when the window is unbounded or empty. O(nodes + edges).
    pub fn refresh_window(&self, node: PhysicalASAPNodeId) -> Option<RefreshWindow>; // new
    /// [`Self::refresh_window`] of each of `nodes`, by id, with the DAG
    /// indexed once for all of them.
    pub fn refresh_windows(&self, nodes: impl IntoIterator<Item = PhysicalASAPNodeId>)
        -> HashMap<PhysicalASAPNodeId, Option<RefreshWindow>>; // new
}

crates/types/src/ir/physical_export.rs

impl PhysicalASAPDAG {
    /// `Self::view_window` of each of `nodes`, by id, with the DAG indexed
    /// once for all of them: O(nodes + edges + the nodes walked below each).
    pub fn view_windows(&self, nodes: impl IntoIterator<Item = PhysicalASAPNodeId>)
        -> HashMap<PhysicalASAPNodeId, ScanWindow>; // signature unchanged; now also counts an unexpanded PaneChain below as all its panes
    /// Each node's operator, by id.
    pub(crate) fn payloads_by_id(&self) -> HashMap<PhysicalASAPNodeId, &PhysicalASAPOperatorPayload>; // was private
}

/// How long before the evaluation time `node` is evaluated: the sum of the
/// offsets of the `TimeShift`s from `node` down; `None` when one of them has
/// an `@` anchor or the sum overflows.
pub(crate) fn offset_below<'a>(node: PhysicalASAPNodeId,
    payload_of: impl Fn(PhysicalASAPNodeId) -> Option<&'a PhysicalASAPOperatorPayload>) -> Option<i64>; // new

crates/planner/src/installable.rs, crates/planner/src/lib.rs

impl InstallablePlan {
    /// The views refreshed at ingestion, by refresh group, each group after
    /// the groups it reads ([`PhysicalASAPDAG::ingestion_refresh_groups`]):
    /// a runtime refreshes each group in one ingestion run per window.
    pub fn ingestion_refresh_groups(&self) -> Result<Vec<IngestionRefreshGroup>, InstallablePlanError>; // new
    /// Whether the Planner's executor compiles this plan as a runtime runs
    /// it: each refresh group's views compiled together (every root its
    /// ingestion runs may have) and each other view's refresh, ... then the
    /// query run ... // was: each view's refresh
    pub fn check_compiles(&self) -> Result<(), String>;
}

pub enum InstallablePlanError {
    /// Unsatisfiable request: the executor cannot refresh materialized view
    /// `node_id` from its inputs (`reason`); for a refresh group, `node_id`
    /// is its first view and `reason` names every view compiled with it.
    MaterializedViewNotExecutable { node_id: PhysicalASAPNodeId, reason: String }, // doc and reason changed
    // every other variant: unchanged
}

pub use asap_types::ir::materialized_view::{IngestionRefreshGroup, RefreshWindow}; // new re-exports

Usage (the new runtime test panes_of_one_refresh_window_read_each_raw_window_once, run at this commit):

// Issue #1151's example: p90 at rank error 0.01 over 1 m and p99 at relative
// error 0.01 over 5 m, both every 10 s, over 1,000 `latency_ms` gauges
// sampled every 10 s; the deployment keeps no raw data.
let workload: PlanningWorkload = /* the two entries above */;
// Stage 1-3 select a KLL chain (6 × 10 s panes) and a DDSketch chain
// (30 × 10 s panes), both MatAtIngestion over one Scan; installation expands
// the chains and compiles each refresh group (`executor_compiles`).
let (plan, _report) = plan_installable(UserInput::new(&workload, FrontendInput::Promql { histograms: None },
    PlanningModels::builtin().with_deployment_capabilities(&capabilities))).await?;
// One group: refresh window (10 000 ms, phase 0), 6 KLL + 30 DDSketch pane views.
let groups = plan.ingestion_refresh_groups()?;
assert_eq!(groups[0].refresh_window, Some(RefreshWindow { width_ms: 10_000, phase_ms: 0 }));
// The runtime: for each group, one ingestion run per window end; its roots are
// the group's views whose state for that window is not stored yet (the newest
// pane of each chain once older panes are stored); the Scan they share reads
// raw data once.
for group in &groups {
    for (span, views) in /* group.views bucketed by (start, end], those not stored */ {
        let roots: Vec<u64> = views.iter().map(|v| v.node_id as u64).collect();
        let refresh = compile(&plan.dag, &[], &roots)?; // KLL and DDSketch panes ending at `span.1`
        // execute with Scope::Ingestion { window_start_ms: span.0, window_end_ms: span.1 }
    }
}
// First evaluation: 36 pane states built with 30 `read_raw_data` calls (36
// with one run per view); each state equals that view's run alone; q1 and q2
// answer as the query-time reference.

Major algorithm changes

Refresh grouping (installation and stand-in runtimes)

• Problem. For the example above, the old per-view refresh made 36 ingestion runs and 36 read_raw_data calls at the first evaluation, one per pane view. The 6 KLL panes and the newest 6 DDSketch panes cover the same 6 windows.
• Before / after.
  • Before: for each view, compile(dag, reads, [view]) and run it.
  • After:
    1. ingestion_refresh_groups() groups the views by refresh window.
    2. Per group and window end, run once with roots = the group's views not yet stored for that window, and slots = the views they read that are not roots.
    3. The views refreshed on first read keep one run each.
    4. The install check compiles each group once with all its views as roots, a superset of every run's roots.
• Correctness. A run over (s, e] builds, for every root, the window ending at e. Views with one refresh window and the same span read exactly (s, e]. The executor runs each shared node once per execute. The test checks every grouped state against the state its view builds alone, and the answers against the query-time reference.
• Complexity. refresh_windows is O(nodes + edges). ingestion_refresh_groups adds O(views + reads) to materialized_views. executor_compiles now compiles fewer refreshes, each over the nodes below all of its roots. The existing 300-pane check (view_refreshes_compile_only_the_nodes_below_them) still holds the compiled nodes at ≤ 6× the plan.
• Assumptions and limits. Different refresh windows never share a run. Unbounded windows stay one view per group. The deployment still decides sealing, lateness, threading and grid alignment.

Before this PR

The install check and the Planner's stand-in runtimes compiled and ran one ingestion refresh per view, so a Scan under two equal-width chains was read once per chain per window. Nothing in the API named which views can share a run.

After this PR

A runtime reads the groups from InstallablePlan::ingestion_refresh_groups() (or the DAG method) and runs each group once per window. The install check compiles exactly those runs. The stand-in runtimes show one raw read per pane window for the example. The cost rule does not change yet.

Verification

• Failing first:
  • With the stand-in runtime run one view per run (main's behavior), the new runtime test fails: left: 36, right: 30 raw-data reads.
  • The types tests need the new API, which main lacks.
• cargo test -p asap-types --test types: 103 passed, 6 of them new in materialized_view.
• cargo test -p asap-planner --test planner runtime_install: 41 passed, 1 of them new. One unrelated test hit its MILP deadline under machine load (a_quantile_shared_with_a_ratio_is_sized_for_the_ratio) and passed on rerun.
• cargo test -p asap-executor: all passed. view_windows, which the executor calls, changed.
• cargo test -p asap-planner --lib: 11 passed, including view_refreshes_compile_only_the_nodes_below_them.
• cargo test -p asap-integration-tests --test integration materialization_e2e: 3 passed, 1 ignored (as before).
• cargo fmt --all --check and cargo clippy -p asap-types -p asap-planner -p asap-integration-tests -p asap-executor --all-targets -- -D warnings: clean.

Note: with 1 s samples (the issue's first example), the Planner picks 1 s panes for the DDSketch chain and 10 s panes for the KLL chain today, so they are two groups. The test samples every 10 s so that both chains get 10 s panes.

🤖 Generated with Claude Code