diff --git a/crates/cost/src/execution_demand.rs b/crates/cost/src/execution_demand.rs
index 82899da9d..252a53811 100644
--- a/crates/cost/src/execution_demand.rs
+++ b/crates/cost/src/execution_demand.rs
@@ -345,7 +345,11 @@ pub fn consumers(dag: &PhysicalASAPDAG) -> HashMap<PhysicalASAPNodeId, Vec<Consu
 /// (operator or scalar root), at its own rate, or a materialized node, at
 /// that node's rate; a walk stops at the first materialized node it meets,
 /// and a node read several times within one consumer's walk counts once for
-/// it. Its `multiplier` is that number of consumers.
+/// it. 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.
 pub fn computations(
     dag: &PhysicalASAPDAG,
     recurrence: &[QueryRecurrence],
@@ -410,10 +414,26 @@ pub fn computations(
             )?,
         });
     }
+    // The refresh window of each node refreshed at ingestion.
+    let refresh_windows = dag.refresh_windows(
+        materialized
+            .iter()
+            .filter(|(_, m)| matches!(m, Materialization::MatAtIngestion(_)))
+            .map(|(&id, _)| id),
+    );
     let mut count: HashMap<_, (u64, f64)> = HashMap::new();
     for (id, of_node) in consumers(dag) {
         let entry = count.entry(id).or_default();
+        // The refresh windows whose ingestion run already computes the node.
+        let mut runs = HashSet::new();
         for consumer in of_node {
+            if let Consumer::Materialized(node) = consumer {
+                if let Some(Some(window)) = refresh_windows.get(&node) {
+                    if !runs.insert(*window) {
+                        continue;
+                    }
+                }
+            }
             entry.0 += 1;
             entry.1 += match consumer {
                 Consumer::Query(query) => query_rates[query],
diff --git a/crates/cost/src/lib.rs b/crates/cost/src/lib.rs
index 6f28b2ce8..95e8949f2 100644
--- a/crates/cost/src/lib.rs
+++ b/crates/cost/src/lib.rs
@@ -79,7 +79,8 @@ pub struct ResourceUsage {
     pub average_memory_bytes: f64,
     /// The most memory live at once, assuming every evaluation overlaps:
     /// in-memory kept state plus each computation's working memory, once
-    /// per consumer that recomputes it.
+    /// per consumer that recomputes it, with the `MatAtIngestion` consumers
+    /// of one refresh window counted once.
     pub peak_memory_bytes: u64,
     /// Bytes kept on disk: state materialized on `Medium::Disk`, plus the
     /// raw data query-time scans read (the longest window of each source).
@@ -372,8 +373,9 @@ pub struct NodeCost {
     /// ([`MeasuredOperationCosts::ns_per_evaluation`]) is counted on the
     /// node the query reads first.
     pub cpu_core_seconds_per_second: f64,
-    /// The number of consumers it is computed for (1 when materialized,
-    /// [`execution_demand::computations`]).
+    /// The number of consumers it is computed for, with the
+    /// `MatAtIngestion` consumers of one refresh window counted once (1
+    /// when materialized, [`execution_demand::computations`]).
     pub multiplier: u64,
     /// How it was measured, e.g. "scan 1000 samples; x 1.0000/s; keeps
     /// ... bytes": one computation's work, the times it is computed per
diff --git a/crates/cost/tests/ingestion_refresh_runs.rs b/crates/cost/tests/ingestion_refresh_runs.rs
new file mode 100644
index 000000000..bdc09b2f8
--- /dev/null
+++ b/crates/cost/tests/ingestion_refresh_runs.rs
@@ -0,0 +1,289 @@
+//! How often a `NotMaterialized` node is computed for the views refreshed
+//! at ingestion above it ([`computations`]): once per refresh window, since
+//! a runtime refreshes the views of one refresh window in one ingestion run
+//! (issue #1151).
+
+use std::rc::Rc;
+use std::time::Duration;
+
+use asap_cost::execution_demand::computations;
+use asap_types::ir::operator::operator_properties::{Reduction, Source, TimeShift};
+use asap_types::ir::physical_export::{compile_physical_asap_workload, PhysicalASAPDAG};
+use asap_types::ir::scalar::ColumnRef;
+use asap_types::ir::schema::{
+    DataType, Field, FieldDataType, GroupingStrategy, Schema, SketchAlgorithm, SketchKind,
+    SketchParams, SummaryUpdate,
+};
+use asap_types::ir::{
+    apply_materialization_timings_to_roots, ASAPOp, MaterializationAssignment, NonASAPOp, Operator,
+    OperatorNode, QueryRoot, TimeRangeKind,
+};
+use asap_types::physical::{Materialization, Medium};
+use asap_types::workload::{QueryRecurrence, RepeatedDemand, RepetitionInterval};
+
+/// `m` with columns `ts`, `value` and `job`.
+fn scan() -> Rc<OperatorNode> {
+    OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan {
+        source: Source::TimeSeries { metric: "m".into() },
+        predicates: vec![],
+        schema: Schema::with_time_index(
+            vec![
+                Field::plain("ts", DataType::Timestamp, false),
+                Field::plain("value", DataType::Float64, false),
+                Field::plain("job", DataType::Utf8, true),
+            ],
+            0,
+            vec![],
+        ),
+    }))
+    .unwrap()
+}
+
+/// A quantile sketch of `algorithm` (KLL k = 200 or DDSketch α = 0.01).
+fn quantile_sketch(algorithm: SketchAlgorithm) -> FieldDataType {
+    let params = match algorithm {
+        SketchAlgorithm::Kll => SketchParams::Kll {
+            k: 200,
+            failure_probability: None,
+        },
+        _ => SketchParams::DDSketch { alpha: 0.01 },
+    };
+    FieldDataType::Sketch(SketchKind::new(algorithm, params), Default::default())
+}
+
+/// `tumbling_window_count` panes of `tumbling_window_width_ms` of a quantile sketch of
+/// `algorithm` per `job` over `input`.
+fn chain(
+    input: &Rc<OperatorNode>,
+    algorithm: SketchAlgorithm,
+    tumbling_window_width_ms: u64,
+    tumbling_window_count: u32,
+) -> Rc<OperatorNode> {
+    OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryTumblingWindowChain {
+        child: Rc::clone(input),
+        family: quantile_sketch(algorithm),
+        input: SummaryUpdate::column(ColumnRef::Named("value".into())),
+        reduction: Reduction::by(vec![2]),
+        grouping: GroupingStrategy::default(),
+        filter: None,
+        tumbling_window_width_ms,
+        tumbling_window_count,
+    }))
+    .unwrap()
+}
+
+/// One query per summary of `summaries`, merging it, with each summary
+/// materialized as `materialization`: the DAG Stage 3 costs, chains
+/// unexpanded.
+fn dag_of(summaries: &[Rc<OperatorNode>], materialization: Materialization) -> PhysicalASAPDAG {
+    let mut assignment = MaterializationAssignment::default();
+    let roots: Vec<QueryRoot> = summaries
+        .iter()
+        .map(|summary| {
+            if materialization != Materialization::NotMaterialized {
+                assignment.set(summary, materialization);
+            }
+            QueryRoot::Operator(
+                OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge {
+                    children: vec![Rc::clone(summary)],
+                    newest_tumbling_windows: None,
+                    group_by: None,
+                }))
+                .unwrap(),
+            )
+        })
+        .collect();
+    let timed = apply_materialization_timings_to_roots(&roots, &assignment).unwrap();
+    compile_physical_asap_workload(&timed).unwrap()
+}
+
+/// `(multiplier, per_second)` of `dag`'s `Scan`, each query every 10 s.
+fn scan_computations(dag: &PhysicalASAPDAG) -> (u64, f64) {
+    computations_of(dag, |op| matches!(op, NonASAPOp::Scan { .. }))
+}
+
+/// `(multiplier, per_second)` of the one node of `dag` whose operator
+/// `is_node` picks, each query every 10 s.
+fn computations_of(
+    dag: &PhysicalASAPDAG,
+    is_node: impl Fn(&NonASAPOp<usize>) -> bool,
+) -> (u64, f64) {
+    let every_10_s =
+        QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(RepetitionInterval(10_000)));
+    let recurrence = vec![every_10_s; dag.roots.len()];
+    let times = computations(dag, &recurrence, &[], 3600.0, None).unwrap();
+    let node = dag
+        .nodes
+        .iter()
+        .find(|node| matches!(&node.payload, Operator::NonASAP(op) if is_node(op)))
+        .unwrap();
+    times[&node.id]
+}
+
+/// `child` evaluated `offset_ms` earlier.
+fn shifted(child: &Rc<OperatorNode>, offset_ms: i64) -> Rc<OperatorNode> {
+    OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeShift {
+        shift: TimeShift {
+            offset_ms,
+            at: None,
+        },
+        child: Rc::clone(child),
+    }))
+    .unwrap()
+}
+
+const AT_INGESTION: Materialization = Materialization::MatAtIngestion(Medium::Memory);
+
+/// A KLL chain and a DDSketch chain of 10 s panes over one `Scan`, both
+/// kept at ingestion, are refreshed in one ingestion run per pane: the
+/// `Scan` is computed once per second of arriving data, not twice.
+#[test]
+fn a_scan_under_two_chains_of_one_pane_width_is_computed_once() {
+    let scan = scan();
+    let dag = dag_of(
+        &[
+            chain(&scan, SketchAlgorithm::Kll, 10_000, 6),
+            chain(&scan, SketchAlgorithm::DDSketch, 10_000, 30),
+        ],
+        AT_INGESTION,
+    );
+    assert_eq!(scan_computations(&dag), (1, 1.0));
+}
+
+/// With 10 s and 60 s panes the chains need different runs (the 10 s chain
+/// needs five pane ends the 60 s chain does not): the `Scan` is computed
+/// once for each.
+#[test]
+fn a_scan_under_chains_of_two_pane_widths_is_computed_for_each() {
+    let scan = scan();
+    let dag = dag_of(
+        &[
+            chain(&scan, SketchAlgorithm::Kll, 10_000, 6),
+            chain(&scan, SketchAlgorithm::DDSketch, 60_000, 5),
+        ],
+        AT_INGESTION,
+    );
+    assert_eq!(scan_computations(&dag), (2, 2.0));
+}
+
+/// Query consumers are unchanged: with both chains built at query time
+/// (`NotMaterialized`), each query computes the `Scan` at its own rate.
+#[test]
+fn query_consumers_still_count_once_each() {
+    let scan = scan();
+    let range = |width_ms| {
+        OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeRange {
+            range: Duration::from_millis(width_ms),
+            kind: TimeRangeKind::Range,
+            child: Rc::clone(&scan),
+        }))
+        .unwrap()
+    };
+    let build = |input: Rc<OperatorNode>, algorithm| {
+        OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg {
+            child: input,
+            family: quantile_sketch(algorithm),
+            input: SummaryUpdate::column(ColumnRef::Named("value".into())),
+            reduction: Reduction::by(vec![2]),
+            grouping: GroupingStrategy::default(),
+            filter: None,
+        }))
+        .unwrap()
+    };
+    let dag = dag_of(
+        &[
+            build(range(60_000), SketchAlgorithm::Kll),
+            build(range(60_000), SketchAlgorithm::DDSketch),
+        ],
+        Materialization::NotMaterialized,
+    );
+    assert_eq!(scan_computations(&dag), (2, 0.2));
+}
+
+/// Two 10 s chains and a 60 s chain over one `Scan`: two refresh windows,
+/// two ingestion runs.
+#[test]
+fn a_scan_under_chains_of_10_10_and_60_s_is_computed_for_each_width() {
+    let scan = scan();
+    let dag = dag_of(
+        &[
+            chain(&scan, SketchAlgorithm::Kll, 10_000, 6),
+            chain(&scan, SketchAlgorithm::DDSketch, 10_000, 30),
+            chain(&scan, SketchAlgorithm::Kll, 60_000, 5),
+        ],
+        AT_INGESTION,
+    );
+    assert_eq!(scan_computations(&dag), (2, 2.0));
+}
+
+/// A 10 s chain over the `Scan` and a 10 s chain over the `Scan` shifted by
+/// 5 s end their panes on different grids (phases 0 and 5 s): two runs.
+#[test]
+fn chains_of_one_width_and_two_phases_are_computed_apart() {
+    let scan = scan();
+    let dag = dag_of(
+        &[
+            chain(&scan, SketchAlgorithm::Kll, 10_000, 6),
+            chain(
+                &shifted(&scan, 5_000),
+                SketchAlgorithm::DDSketch,
+                10_000,
+                30,
+            ),
+        ],
+        AT_INGESTION,
+    );
+    assert_eq!(scan_computations(&dag), (2, 2.0));
+}
+
+/// Two 10 s chains over one `TimeShift(10 s)` of the `Scan` (phase 0 for
+/// both): the shift and the `Scan` below it run once per run.
+#[test]
+fn nodes_below_a_shared_intermediate_node_are_computed_once_per_run() {
+    let shift = shifted(&scan(), 10_000);
+    let dag = dag_of(
+        &[
+            chain(&shift, SketchAlgorithm::Kll, 10_000, 6),
+            chain(&shift, SketchAlgorithm::DDSketch, 10_000, 30),
+        ],
+        AT_INGESTION,
+    );
+    assert_eq!(scan_computations(&dag), (1, 1.0));
+    assert_eq!(
+        computations_of(&dag, |op| matches!(op, NonASAPOp::TimeShift { .. })),
+        (1, 1.0)
+    );
+}
+
+/// Two summaries kept at ingestion over rows with no time window
+/// (`Values`) have no refresh window: each counts alone.
+#[test]
+fn consumers_of_unbounded_windows_count_alone() {
+    let rows = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Values {
+        rows: vec![],
+        schema: scan().schema.clone(),
+    }))
+    .unwrap();
+    let build = |algorithm| {
+        OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg {
+            child: Rc::clone(&rows),
+            family: quantile_sketch(algorithm),
+            input: SummaryUpdate::column(ColumnRef::Named("value".into())),
+            reduction: Reduction::by(vec![2]),
+            grouping: GroupingStrategy::default(),
+            filter: None,
+        }))
+        .unwrap()
+    };
+    let dag = dag_of(
+        &[
+            build(SketchAlgorithm::Kll),
+            build(SketchAlgorithm::DDSketch),
+        ],
+        AT_INGESTION,
+    );
+    assert_eq!(
+        computations_of(&dag, |op| matches!(op, NonASAPOp::Values { .. })),
+        (2, 2.0)
+    );
+}
diff --git a/crates/integration-tests/tests/materialization_e2e.rs b/crates/integration-tests/tests/materialization_e2e.rs
index bcce2c805..b342f3a16 100644
--- a/crates/integration-tests/tests/materialization_e2e.rs
+++ b/crates/integration-tests/tests/materialization_e2e.rs
@@ -117,8 +117,8 @@ fn floats(result: &[Vec<Batch>]) -> Vec<f64> {
 }
 
 /// Execute a timed DAG over one raw-source batch per `Scan`
-/// (`batch(schema, metric)`): each materialized view refreshed in an
-/// ingestion run over `ingestion`, then the query at `evaluation_time_ms`
+/// (`batch(schema, metric)`): each refresh group of materialized views
+/// refreshed in one ingestion run over `ingestion`, then the query at `evaluation_time_ms`
 /// reading the views' states. Returns the root's values and the views.
 fn execute(
     dag: &PhysicalASAPDAG,
@@ -138,12 +138,16 @@ fn execute(
             batch(schema, metric).to_rows()
         })
     };
-    // Each view's ingestion run refreshes it from the raw data.
+    // Each refresh group's ingestion run refreshes its views from the raw
+    // data, compiled once with all of them as roots.
     let mut store = raw(&dag);
-    for view in &views {
-        let refresh = compile(&dag, &[], &[view.node_id as u64]).unwrap();
-        let states = physical_common::execute(&refresh, raw(&dag), ingestion.clone()).remove(0);
-        store = store.with_materialized_view(view.node_id, states).unwrap();
+    for group in dag.ingestion_refresh_groups().unwrap() {
+        let roots: Vec<u64> = group.views.iter().map(|view| view.node_id as u64).collect();
+        let refresh = compile(&dag, &[], &roots).unwrap();
+        let states = physical_common::execute(&refresh, raw(&dag), ingestion.clone());
+        for (view, states) in group.views.iter().zip(states) {
+            store = store.with_materialized_view(view.node_id, states).unwrap();
+        }
     }
     let query = compile(&dag, &views, &[dag.roots[0] as u64]).unwrap();
     let result = physical_common::execute(&query, store, Scope::Query { evaluation_time_ms });
diff --git a/crates/integration-tests/tests/planner_layering_common/mod.rs b/crates/integration-tests/tests/planner_layering_common/mod.rs
index 4fcf3507f..efa47f81b 100644
--- a/crates/integration-tests/tests/planner_layering_common/mod.rs
+++ b/crates/integration-tests/tests/planner_layering_common/mod.rs
@@ -774,7 +774,8 @@ pub fn assert_each_node_charged_per_consuming_query(run: &Run) {
 /// node; for a `NotMaterialized` one, the number of consumers that reach
 /// it. A consumer is a query (operator or scalar root) or a materialized
 /// node, whose walk starts at its inputs; a walk stops at materialized
-/// nodes and counts each node once.
+/// nodes and counts each node once. The `MatAtIngestion` consumers of one
+/// refresh window share one ingestion run and count once (#1151).
 pub fn computations(dag: &PhysicalASAPDAG) -> BTreeMap<NodeId, u64> {
     let materialized: BTreeSet<_> = dag
         .nodes
@@ -782,29 +783,46 @@ pub fn computations(dag: &PhysicalASAPDAG) -> BTreeMap<NodeId, u64> {
         .filter(|n| n.materialization != Materialization::NotMaterialized)
         .map(|n| n.id)
         .collect();
+    // Each walk's start and, for a consumer refreshed at ingestion, its
+    // refresh window.
     let starts = dag
         .roots
         .iter()
-        .map(|&root| vec![root])
+        .map(|&root| (vec![root], None))
         .chain(
             dag.scalar_roots
                 .iter()
-                .map(|(_, expr)| expr.operator_refs().into_iter().copied().collect()),
+                .map(|(_, expr)| (expr.operator_refs().into_iter().copied().collect(), None)),
         )
-        .chain(materialized.iter().map(|&id| dag.producers(id)));
+        .chain(materialized.iter().map(|&id| {
+            let at_ingestion = matches!(
+                dag.nodes
+                    .iter()
+                    .find(|n| n.id == id)
+                    .unwrap()
+                    .materialization,
+                Materialization::MatAtIngestion(_)
+            );
+            let window = at_ingestion.then(|| dag.refresh_window(id)).flatten();
+            (dag.producers(id), window)
+        }));
     let mut times: BTreeMap<_, _> = dag
         .nodes
         .iter()
         .map(|n| (n.id, u64::from(materialized.contains(&n.id))))
         .collect();
-    for start in starts {
+    // The (node, refresh window) pairs whose ingestion run is counted.
+    let mut runs = HashSet::new();
+    for (start, window) in starts {
         let mut seen = HashSet::new();
         let mut stack = start;
         while let Some(id) = stack.pop() {
             if materialized.contains(&id) || !seen.insert(id) {
                 continue;
             }
-            *times.get_mut(&id).expect("node") += 1;
+            if window.is_none_or(|window| runs.insert((id, window))) {
+                *times.get_mut(&id).expect("node") += 1;
+            }
             stack.extend(dag.producers(id));
         }
     }
diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs
index b30e75ffd..a4cf7a76f 100644
--- a/crates/plan-selection/src/lib.rs
+++ b/crates/plan-selection/src/lib.rs
@@ -43,7 +43,12 @@
 //!   ([`NodeCost::multiplier`](asap_cost::NodeCost::multiplier)) at the sum of their rates. Its consumers are
 //!   the queries whose sub-DAG reaches it and the materialized nodes above
 //!   it, counted without looking through a materialized node: a node below
-//!   one is computed when that node is. Stage 2 keeps a shared node one
+//!   one is computed when that node is. The `MatAtIngestion` consumers of
+//!   one refresh window (`PhysicalASAPDAG::refresh_window`: for a pane
+//!   chain, its pane width) are refreshed in one ingestion run and count as
+//!   one consumer: a `Scan` under a KLL chain and a DDSketch chain of 10 s
+//!   panes is computed once per second of arriving data, under chains of
+//!   10 s and 60 s panes twice. Stage 2 keeps a shared node one
 //!   node (it copies it only for readers at two timings). A shared
 //!   `NotMaterialized` input costs what reading it separately does (#509
 //!   Example 4) only when every consumer reads the same window: a query-time `Scan` is priced over the longest window
diff --git a/crates/plan-selection/src/search/costing/direct_profile.rs b/crates/plan-selection/src/search/costing/direct_profile.rs
index 58fc20ac9..942ad5324 100644
--- a/crates/plan-selection/src/search/costing/direct_profile.rs
+++ b/crates/plan-selection/src/search/costing/direct_profile.rs
@@ -9,6 +9,7 @@ use crate::search::milp::direct_cost::{DirectCostProfile, NodeCostProfile, Query
 use crate::search::refinement::planner::CandidatePlanningContext;
 use asap_cost::CostModel;
 use asap_cost::{NodeUsage, ProducerOutput};
+use asap_types::ir::materialized_view::RefreshWindow;
 use asap_types::ir::physical_export::PhysicalASAPDAGNode;
 use asap_types::ir::properties::execution_data_state::{DataPrimitive, ExecutionDataState};
 use asap_types::ir::properties::{AccuracyScope, ExecutionTiming};
@@ -418,6 +419,7 @@ fn extract_scan_class(
         ingestion: false,
         standing_cost_per_second: 0.0,
         raw_retention: None,
+        refresh_window: None,
     };
     let profiles = cache
         .iter()
@@ -484,6 +486,18 @@ fn profile_for_priced(
         }),
         materialized: node.materialization != Materialization::NotMaterialized,
         raw_retention,
+        // A chain maintained at ingestion refreshes its newest pane. The
+        // profile admits no `TimeShift` at ingestion, so that pane ends at
+        // the evaluation time: phase 0.
+        refresh_window: match &node.payload {
+            asap_types::ir::Operator::ASAP(ASAPOp::SummaryTumblingWindowChain {
+                tumbling_window_width_ms,
+                ..
+            }) if node.materialization == Materialization::MatAtIngestion(Medium::Memory) => {
+                RefreshWindow::of_window(0, i64::try_from(*tumbling_window_width_ms).unwrap_or(0))
+            }
+            _ => None,
+        },
     })
 }
 
@@ -1911,6 +1925,239 @@ mod tests {
         }
     }
 
+    /// A KLL chain (p90 over a minute) and a DDSketch chain (p99 over five
+    /// minutes) kept at ingestion over one `Scan`, each in panes of its
+    /// query's interval. With equal pane widths, one ingestion run per pane
+    /// computes the `Scan` for both chains, and the old Stage 3 engine's
+    /// modeled cost equals the real evaluator's with the `Scan` computed
+    /// once, for one schedule (both every 10 s) and for mixed schedules
+    /// (every 20 s and every 40 s, 20 s panes). With 10 s and 60 s panes the
+    /// chains run apart and both count the `Scan` twice (#1151).
+    #[test]
+    fn chains_of_one_refresh_window_share_their_scan_in_the_direct_profile() {
+        // ((p90 interval, its pane width), (p99 interval, its pane width), Scan computations)
+        for (kll, dd, scan_computations) in [
+            ((10_000, 10_000), (10_000, 10_000), 1),
+            ((20_000, 20_000), (40_000, 20_000), 1),
+            ((10_000, 10_000), (60_000, 60_000), 2),
+        ] {
+            let (modeled, evaluated, multiplier) = two_chains_over_one_scan(kll, dd);
+            let case = format!("p90 (every, panes) {kll:?} ms, p99 {dd:?} ms");
+            assert!(
+                (evaluated - modeled).abs() <= 1e-9 * evaluated.max(1e-15),
+                "{case}: direct {modeled}, evaluator {evaluated}"
+            );
+            assert_eq!(multiplier, scan_computations, "{case}");
+        }
+    }
+
+    /// [`chains_of_one_refresh_window_share_their_scan_in_the_direct_profile`]
+    /// with the p90 and the p99 each `(interval in ms, pane width in ms)`:
+    /// the direct profile's modeled cost of the selected plan, the
+    /// evaluator's cost of it, and the evaluator's multiplier of the `Scan`.
+    fn two_chains_over_one_scan(kll: (u64, u64), dd: (u64, u64)) -> (f64, f64, u64) {
+        use asap_logical_asap_optimization::test_support::{
+            compose_logical_candidate, generate_per_computation_candidates, pane_candidates,
+            window_composition_pass2_candidates,
+        };
+        use asap_types::ir::schema::SketchAlgorithm;
+        use asap_types::stages::HowToComputeAggregateMeasure;
+        use std::collections::HashMap;
+        use std::rc::Rc;
+        /// `node` and every node below it as candidates of `configured`, one
+        /// per node (`seen`, by address), at ingestion below a pane chain.
+        fn append(
+            node: &Rc<OperatorNode>,
+            configured: &mut PhysicalCandidatesWithCosts,
+            ingestion: bool,
+            seen: &mut HashMap<*const OperatorNode, PhysicalCandidateId>,
+        ) -> PhysicalCandidateId {
+            if let Some(id) = seen.get(&Rc::as_ptr(node)) {
+                return *id;
+            }
+            let chain = matches!(node.asap(), Some(ASAPOp::SummaryTumblingWindowChain { .. }));
+            let ingestion = ingestion || chain;
+            let mut inputs = Vec::new();
+            node.operator
+                .map_children(|child| inputs.push(append(child, configured, ingestion, seen)));
+            let id = PhysicalCandidateId(configured.physical.candidates.len());
+            configured.push_for_tests(
+                Rc::clone(node),
+                if ingestion {
+                    ExecutionTiming::IngestionTime
+                } else {
+                    ExecutionTiming::QueryTime
+                },
+                if chain {
+                    Materialization::MatAtIngestion(Medium::Memory)
+                } else {
+                    Materialization::NotMaterialized
+                },
+                0.0,
+                inputs
+                    .into_iter()
+                    .map(|supplier| InputProducerCandidates {
+                        candidates: vec![supplier].into(),
+                    })
+                    .collect(),
+            );
+            seen.insert(Rc::as_ptr(node), id);
+            id
+        }
+        let target = AccuracyTarget::max_error(0.05)
+            .failure_probability(0.01)
+            .accuracy_scope(AccuracyScope::PerValue);
+        // (query, interval in ms, sketch, pane width in ms)
+        let queries = [
+            (
+                "quantile_over_time(0.9, latency[1m])",
+                kll.0,
+                SketchAlgorithm::Kll,
+                kll.1,
+            ),
+            (
+                "quantile_over_time(0.99, latency[5m])",
+                dd.0,
+                SketchAlgorithm::DDSketch,
+                dd.1,
+            ),
+        ];
+        let originals: Vec<QueryRoot> = queries
+            .iter()
+            .map(|(query, ..)| QueryRoot::Operator(lower_promql(query, target.clone())))
+            .collect();
+        let generated = generate_per_computation_candidates(
+            &asap_logical_asap_optimization::generate_dag_of_logical_memo_groups(&originals),
+        )
+        .unwrap();
+        // Per target, the way to compute it with its query's sketch.
+        let choices: Vec<usize> = generated
+            .targets
+            .iter()
+            .zip(&queries)
+            .map(|(target, (_, _, algorithm, _))| {
+                target
+                    .ways_to_compute
+                    .iter()
+                    .position(|way| {
+                        matches!(way, HowToComputeAggregateMeasure::Summary(found) if found == algorithm)
+                    })
+                    .unwrap()
+            })
+            .collect();
+        let roots = compose_logical_candidate(&generated, &choices).unwrap();
+        let grid: Vec<Option<u64>> = queries
+            .iter()
+            .map(|(_, interval, ..)| Some(*interval))
+            .collect();
+        // Each build in its query's pane width.
+        let panes: Vec<_> = pane_candidates(&roots, &grid)
+            .into_iter()
+            .filter(|pane| {
+                (pane.window_ms(), pane.tumbling_window_width_ms) == (60_000, kll.1)
+                    || (pane.window_ms(), pane.tumbling_window_width_ms) == (300_000, dd.1)
+            })
+            .collect();
+        assert_eq!(panes.len(), 2, "{panes:?}");
+        let roots = window_composition_pass2_candidates(&roots, &panes)
+            .unwrap()
+            .into_iter()
+            .find(|candidate| {
+                candidate.pane_candidates.len() == 2 && candidate.shared_chains.is_empty()
+            })
+            .unwrap()
+            .roots
+            .unwrap();
+        let data: DataWorkload = serde_json::from_value(serde_json::json!({
+            "arrival": {"ingesting": {"rows_per_second": 100.0}},
+            "sources": {"latency": {"series_count": 2, "rows_held": 30_000,
+                "values_non_negative": true}}
+        }))
+        .unwrap();
+        let sized = crate::sized_as_pass1(&roots, &[target.clone(), target.clone()], &data);
+        let mut configured = PhysicalCandidatesWithCosts::empty_for_tests();
+        let mut seen = HashMap::new();
+        for root in &sized {
+            let QueryRoot::Operator(root) = root else {
+                panic!("operator")
+            };
+            let id = append(root, &mut configured, false, &mut seen);
+            configured.dag.query_roots.push(QueryOutputCandidates {
+                candidates: vec![id],
+            });
+        }
+        configured.sync_physical_for_tests();
+        let scans: Vec<usize> = (0..configured.physical.candidates.len())
+            .filter(|&id| matches!(configured.node(id).non_asap(), Some(NonASAPOp::Scan { .. })))
+            .collect();
+        assert_eq!(scans.len(), 1, "one shared Scan");
+        let entries: Vec<QueryWorkloadEntry> = queries
+            .iter()
+            .map(|(query, interval, ..)| QueryWorkloadEntry {
+                query: Query((*query).into()),
+                requirements: QueryRequirements {
+                    accuracy: target.clone(),
+                    max_latency_ms: None,
+                },
+                predictability: Predictability::Predictable { known_at: None },
+                recurrence: QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(
+                    RepetitionInterval(*interval as u32),
+                )),
+                range_evaluation: None,
+            })
+            .collect();
+        let mut settings = asap_cost::CostInputs::default();
+        settings.resource_speed.ms_per_cpu_op *= 1000.0;
+        settings.resource_speed.ms_per_scanned_byte *= 1000.0;
+        let context = CandidatePlanningContext {
+            deadline: None,
+            refine_materialization: false,
+            query_roots: &originals,
+            realizer: &crate::search::test_realizer(&originals, &entries, &data),
+            entries: &entries,
+            data: &data,
+            models: crate::SelectionModels::builtin().with_cost_inputs(&settings),
+            calibration: asap_cost::CalibrationFit::default(),
+        };
+        let extracted = extract(&configured, &context).unwrap().unwrap();
+        assert!(extracted.supported.iter().all(|supported| *supported));
+        let proposal = crate::search::milp::model::solve_direct_cost_proposal_with_metrics(
+            &configured.dag,
+            &[],
+            None,
+            &mut Default::default(),
+            &extracted.profile,
+        )
+        .unwrap()
+        .unwrap();
+        let mut rejections =
+            crate::search::refinement::planner::CandidateRejectionSummary::default();
+        let evaluated = crate::search::refinement::planner::evaluate_configured_candidate(
+            &configured,
+            &context,
+            &proposal.selected,
+            0,
+            &mut rejections,
+            &mut Vec::new(),
+        )
+        .unwrap()
+        .unwrap_or_else(|| panic!("{rejections:?}"));
+        let costs = &evaluated.0.costs[&evaluated.0.selected];
+        let scan_multiplier = costs
+            .per_node
+            .iter()
+            .find(|(_, cost)| cost.operator == asap_cost::OperatorKind::Scan)
+            .map(|(_, cost)| cost.multiplier)
+            .unwrap();
+        (
+            extracted
+                .profile
+                .selected_cost(&configured.dag, &proposal.selected),
+            costs.cost_per_second.unwrap(),
+            scan_multiplier,
+        )
+    }
+
     // Real shared KLL/DDS plans verify schedule unions and held-memory lifetime,
     // including two equal frequencies whose different phases cannot coalesce.
     #[test]
diff --git a/crates/plan-selection/src/search/milp/direct_cost.rs b/crates/plan-selection/src/search/milp/direct_cost.rs
index cc26f0526..9ff9253a2 100644
--- a/crates/plan-selection/src/search/milp/direct_cost.rs
+++ b/crates/plan-selection/src/search/milp/direct_cost.rs
@@ -4,8 +4,9 @@
 //! Profiles are supplied by physical costing; this module never infers prices
 //! from a logical template or treats accuracy as a monetary penalty.
 use crate::search::candidates::graph::{QueryRootAssignment, SelectedCandidateGraph};
+use asap_types::ir::materialized_view::RefreshWindow;
 use good_lp::{constraint, variable, Constraint, Expression, ProblemVariables, Variable};
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, BTreeSet};
 
 #[path = "general_schedule.rs"]
 mod general_schedule;
@@ -43,6 +44,10 @@ pub(crate) struct NodeCostProfile {
     pub held_memory_cost_per_second: f64,
     pub materialized: bool,
     pub raw_retention: Option<(String, f64)>,
+    /// 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>,
 }
 
 impl NodeCostProfile {
@@ -260,7 +265,9 @@ impl DirectCostProfile {
         // Per unmaterialized node, its executions per second and its held
         // time per second: each consumer (a query, or a materialized node)
         // computes its own copy at its rate and holds it for its query's
-        // latency, or for the materialized node's longest reader's.
+        // latency, or for the materialized node's longest reader's; the
+        // nodes maintained at ingestion of one refresh window share one
+        // ingestion run, which computes it once for all of them.
         let mut execution_rates = vec![0.0; self.nodes.len()];
         let mut held_seconds = vec![0.0; self.nodes.len()];
         let boundaries = selected
@@ -268,23 +275,34 @@ impl DirectCostProfile {
             .iter()
             .map(|root| {
                 let query = root.query_root;
-                (vec![root.candidate.0], rates[query], latency[query])
+                (vec![root.candidate.0], rates[query], latency[query], None)
             })
             .chain(
                 selected
                     .selected_candidates
                     .iter()
                     .filter(|id| self.nodes[id.0].materialized)
-                    .map(|id| (inputs[id.0].clone(), union_rate(id.0), longest(id.0))),
+                    .map(|id| {
+                        (
+                            inputs[id.0].clone(),
+                            union_rate(id.0),
+                            longest(id.0),
+                            self.nodes[id.0].refresh_window,
+                        )
+                    }),
             );
-        for (mut pending, rate, lifetime) in boundaries {
+        // The (node, refresh window) pairs whose ingestion run is counted.
+        let mut runs = BTreeSet::new();
+        for (mut pending, rate, lifetime, refresh_window) in boundaries {
             let mut seen = vec![false; self.nodes.len()];
             while let Some(id) = pending.pop() {
                 if self.nodes[id].materialized || std::mem::replace(&mut seen[id], true) {
                     continue;
                 }
-                execution_rates[id] += rate;
-                held_seconds[id] += rate * lifetime;
+                if refresh_window.is_none_or(|window| runs.insert((id, window))) {
+                    execution_rates[id] += rate;
+                    held_seconds[id] += rate * lifetime;
+                }
                 pending.extend(&inputs[id]);
             }
         }
@@ -621,14 +639,33 @@ pub(super) fn encode(
         roots,
         &check_deadline,
     )?;
+    // Each node's executions: one per boundary reaching it, except that
+    // the boundaries maintained at ingestion of one refresh window share one
+    // ingestion run, executed when any of them reaches the node.
     let mut boundary_count = vec![Expression::from(0.0); count];
+    let mut runs: BTreeMap<(usize, RefreshWindow), Vec<Variable>> = BTreeMap::new();
     for flow in &flows {
+        let refresh_window = flow
+            .boundary
+            .checked_sub(queries)
+            .and_then(|consumer| profile.nodes[consumer].refresh_window);
         for (id, reached) in flow.reached.iter().enumerate() {
-            if let Some(reached) = reached {
-                boundary_count[id] += *reached;
+            match (reached, refresh_window) {
+                (Some(reached), Some(window)) => {
+                    runs.entry((id, window)).or_default().push(*reached)
+                }
+                (Some(reached), None) => boundary_count[id] += *reached,
+                (None, _) => {}
             }
         }
     }
+    for ((id, _), reached) in runs {
+        let run = variables.add(variable().min(0).max(1));
+        for reached in reached {
+            constraints.push(constraint!(run >= reached));
+        }
+        boundary_count[id] += run;
+    }
     // Each materialized node's longest reader's latency, built on first use.
     let mut longest_of: BTreeMap<usize, Variable> = BTreeMap::new();
     let mut longest =
@@ -767,6 +804,7 @@ mod tests {
                     held_memory_cost_per_second: 3.0,
                     materialized: true,
                     raw_retention: None,
+                    refresh_window: None,
                 },
                 NodeCostProfile {
                     ingestion: false,
@@ -829,6 +867,7 @@ mod tests {
                     held_memory_cost_per_second: 4.0,
                     materialized: true,
                     raw_retention: None,
+                    refresh_window: None,
                 },
                 NodeCostProfile {
                     ingestion: false,
@@ -878,6 +917,7 @@ mod tests {
                         held_memory_cost_per_second: 3.0,
                         materialized: true,
                         raw_retention: None,
+                        refresh_window: None,
                     },
                     NodeCostProfile {
                         ingestion: false,
diff --git a/crates/plan-selection/src/search/milp/general_schedule.rs b/crates/plan-selection/src/search/milp/general_schedule.rs
index ecd9cbf5f..5906a5c49 100644
--- a/crates/plan-selection/src/search/milp/general_schedule.rs
+++ b/crates/plan-selection/src/search/milp/general_schedule.rs
@@ -67,7 +67,9 @@ pub(super) fn encode(
     }
     // Per node, its demands `(used, rate, holder)`: a materialized node's own
     // executions, and an unmaterialized node's per consumer, each held for
-    // its holder's lifetime.
+    // its holder's lifetime; the consumers maintained at ingestion of one
+    // refresh window share one ingestion run, one demand used when any of
+    // them uses the node.
     let mut demands: Vec<Vec<(Variable, f64, Holder)>> = materialized
         .iter()
         .enumerate()
@@ -78,6 +80,9 @@ pub(super) fn encode(
                 .collect()
         })
         .collect();
+    // Per (node, refresh window), the uses of the node by that window's
+    // consumers, and one of them, whose lifetime holds the run's result.
+    let mut runs: BTreeMap<(usize, RefreshWindow), (Vec<Variable>, usize)> = BTreeMap::new();
     for flow in boundary_flows(
         graph,
         profile,
@@ -102,15 +107,29 @@ pub(super) fn encode(
             } else {
                 let consumer = flow.boundary - queries;
                 for (used, rate) in &materialized[consumer] {
-                    demands[id].push((
-                        both(variables, constraints, reached, *used),
-                        *rate,
-                        Holder::Node(consumer),
-                    ));
+                    let use_of_node = both(variables, constraints, reached, *used);
+                    match profile.nodes[consumer].refresh_window {
+                        Some(window) => runs
+                            .entry((id, window))
+                            .or_insert_with(|| (Vec::new(), consumer))
+                            .0
+                            .push(use_of_node),
+                        None => demands[id].push((use_of_node, *rate, Holder::Node(consumer))),
+                    }
                 }
             }
         }
     }
+    // One ingestion run per (node, refresh window), at the ingestion rate:
+    // used when any consumer of the window uses the node (minimization keeps
+    // it at the largest use).
+    for ((id, _), (uses, consumer)) in runs {
+        let run = variables.add(variable().min(0).max(1));
+        for use_of_node in uses {
+            constraints.push(constraint!(run >= use_of_node));
+        }
+        demands[id].push((run, 1.0, Holder::Node(consumer)));
+    }
     let mut objective = Expression::from(profile.request_cost / divisor);
     let mut retained = BTreeMap::<&str, Vec<(usize, f64)>>::new();
     // Each materialized node's longest reader's latency, built on first use.
diff --git a/crates/plan-selection/src/search/milp/schedule_selection_tests.rs b/crates/plan-selection/src/search/milp/schedule_selection_tests.rs
index 81062594e..00d54c502 100644
--- a/crates/plan-selection/src/search/milp/schedule_selection_tests.rs
+++ b/crates/plan-selection/src/search/milp/schedule_selection_tests.rs
@@ -98,6 +98,7 @@ fn heterogeneous_intervals_price_boundary_rates_and_longest_reader_lifetimes() {
                 held_memory_cost_per_second: 4.0,
                 materialized: true,
                 raw_retention: None,
+                refresh_window: None,
             },
             NodeCostProfile {
                 ingestion: false,
@@ -168,6 +169,7 @@ fn aligned_and_offset_phases_use_evaluator_union_policy() {
                 held_memory_cost_per_second: 2.0,
                 materialized: true,
                 raw_retention: None,
+                refresh_window: None,
             }],
         };
         let result = solve_direct_cost_proposal_with_metrics(
@@ -222,6 +224,7 @@ fn heterogeneous_selection_matches_exhaustive_reader_assignments() {
                 held_memory_cost_per_second: 3.0,
                 materialized: true,
                 raw_retention: None,
+                refresh_window: None,
             },
             NodeCostProfile {
                 per_request: false,
@@ -233,6 +236,7 @@ fn heterogeneous_selection_matches_exhaustive_reader_assignments() {
                 held_memory_cost_per_second: 1.0,
                 materialized: true,
                 raw_retention: None,
+                refresh_window: None,
             },
             NodeCostProfile {
                 ingestion: false,
diff --git a/crates/planner/src/installable.rs b/crates/planner/src/installable.rs
index b8d2831dd..855398f5f 100644
--- a/crates/planner/src/installable.rs
+++ b/crates/planner/src/installable.rs
@@ -29,7 +29,9 @@ use std::rc::Rc;
 use asap_cost::{NodePriceSource, OperatorKind};
 use asap_logical_asap_optimization::DeploymentCapabilities;
 use asap_plan_selection::{SearchReport, Selection};
-use asap_types::ir::materialized_view::{InstallableDagError, MaterializedView, RawDataRetention};
+use asap_types::ir::materialized_view::{
+    IngestionRefreshGroup, InstallableDagError, MaterializedView, RawDataRetention,
+};
 use asap_types::ir::operator::pane_chain::expand_pane_chains_in_workload;
 use asap_types::ir::physical_export::{
     PhysicalASAPDAG, PhysicalASAPDAGEdge, PhysicalASAPDAGNode, PhysicalASAPDAGValidationError,
@@ -191,7 +193,8 @@ pub enum InstallablePlanError {
     #[error("pane retention for node {node_id}, which is not a pane chain of the selected plan")]
     PaneRetentionOfUnknownChain { node_id: PhysicalASAPNodeId },
     /// Unsatisfiable request: the executor cannot refresh materialized view
-    /// `node_id` from its inputs (`reason`).
+    /// `node_id` from its inputs (`reason`); for a refresh group, `node_id`
+    /// is its first view and `reason` names every view compiled with it.
     #[error("the executor cannot refresh materialized view {node_id} from its inputs: {reason}")]
     MaterializedViewNotExecutable {
         node_id: PhysicalASAPNodeId,
@@ -300,6 +303,15 @@ impl InstallablePlan {
         Ok(self.dag.materialized_views()?)
     }
 
+    /// 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> {
+        Ok(self.dag.ingestion_refresh_groups()?)
+    }
+
     /// How far back each raw-data source is kept for query-time scans
     /// ([`PhysicalASAPDAG::raw_data_retention`]): the raw data a runtime
     /// keeps, the raw-data counterpart of each view's retention.
@@ -326,8 +338,10 @@ impl InstallablePlan {
     }
 
     /// Whether the Planner's executor compiles this plan as a runtime runs
-    /// it: each view's refresh over the views it reads and the `Scan`s it
-    /// reads (from the runtime's data store), then the query run
+    /// it: each refresh group's views compiled together (every root its
+    /// ingestion runs may have) and each other view's refresh,
+    /// over the views it reads and the `Scan`s it reads (from the runtime's
+    /// data store), then the query run
     /// over the views it reads and the `Scan`s it reads outside them. The
     /// DAG's own `TimeRange`s are compiled, so a window the executor cannot
     /// run fails here. `Err` holds the executor's error, or why the views
@@ -446,12 +460,12 @@ impl<'a> IndexedDag<'a> {
         }
     }
 
-    /// The sub-DAG of `root` and every node below it, with their edges, in
-    /// the DAG's node and edge order, rooted at `root`: all a refresh of
-    /// `root` reads, at the cost of its size rather than the plan's.
-    fn dag_below(&self, root: PhysicalASAPNodeId) -> PhysicalASAPDAG {
+    /// The sub-DAG of `roots` and every node below them, with their edges,
+    /// in the DAG's node and edge order, rooted at `roots`: all a refresh of
+    /// `roots` reads, at the cost of its size rather than the plan's.
+    fn dag_below(&self, roots: &[PhysicalASAPNodeId]) -> PhysicalASAPDAG {
         let mut below = HashSet::new();
-        let mut pending = vec![root];
+        let mut pending = roots.to_vec();
         while let Some(id) = pending.pop() {
             if below.insert(id) {
                 pending.extend(self.inputs.get(&id).into_iter().flatten().copied());
@@ -467,7 +481,7 @@ impl<'a> IndexedDag<'a> {
         PhysicalASAPDAG {
             nodes: ids.iter().map(|id| self.nodes[id].clone()).collect(),
             edges,
-            roots: vec![root],
+            roots: roots.to_vec(),
             scalar_roots: Vec::new(),
         }
     }
@@ -503,10 +517,16 @@ fn reader_span_ms_by_node(
     spans
 }
 
-/// Whether the executor compiles `dag` as a runtime runs it: each of
-/// `views`' refreshes ([`executor_refreshes`]), the first in the whole plan,
-/// which runs the executor's checks of the whole plan once, every other over
-/// the nodes below it; then the query run ([`executor_runs_queries`]).
+/// Whether the executor compiles `dag` as a runtime runs it: each refresh
+/// ([`executor_refreshes`]): one per refresh group of the views refreshed
+/// at ingestion ([`PhysicalASAPDAG::ingestion_refresh_groups_of`] `views`)
+/// and one per other view of `views` (refreshed on first read), the first
+/// in the whole plan, which runs the executor's checks of the whole plan
+/// once, every other over the nodes below it; then the query run
+/// ([`executor_runs_queries`]). A group is compiled with all its views as
+/// roots: a compile check covering every run's roots (the views whose
+/// state for the run's window is not stored yet, a subset), not the root
+/// set of any one run.
 fn executor_compiles(
     dag: &PhysicalASAPDAG,
     views: &[MaterializedView],
@@ -514,8 +534,22 @@ fn executor_compiles(
     let indexed = IndexedDag::new(dag);
     let views_by_node: HashMap<PhysicalASAPNodeId, &MaterializedView> =
         views.iter().map(|view| (view.node_id, view)).collect();
-    for (position, view) in views.iter().enumerate() {
-        executor_refreshes(view, &views_by_node, &indexed, position == 0)?;
+    let groups = dag.ingestion_refresh_groups_of(views);
+    let refreshed_at_ingestion: HashSet<PhysicalASAPNodeId> = groups
+        .iter()
+        .flat_map(|group| group.views.iter().map(|view| view.node_id))
+        .collect();
+    let refreshes = groups
+        .iter()
+        .map(|group| group.views.iter().collect::<Vec<_>>())
+        .chain(
+            views
+                .iter()
+                .filter(|view| !refreshed_at_ingestion.contains(&view.node_id))
+                .map(|view| vec![view]),
+        );
+    for (position, refreshed) in refreshes.enumerate() {
+        executor_refreshes(&refreshed, &views_by_node, &indexed, position == 0)?;
     }
     executor_runs_queries(&indexed, &views_by_node)
 }
@@ -550,41 +584,49 @@ fn executor_runs_queries(
     })
 }
 
-/// Whether the executor compiles `view`'s refresh with the views it reads
-/// (among `views_by_node`) as materialized view slots, as a runtime runs it;
-/// the executor's reason when it does not.
+/// Whether the executor compiles the refresh of `refreshed` (one view, or
+/// every view of one refresh group) as roots, with the views they read
+/// (among `views_by_node`) that are not among them as materialized view
+/// slots; the executor's reason, naming the roots, when it does not.
 fn executor_refreshes(
-    view: &MaterializedView,
+    refreshed: &[&MaterializedView],
     views_by_node: &HashMap<PhysicalASAPNodeId, &MaterializedView>,
     indexed: &IndexedDag<'_>,
     in_whole_plan: bool,
 ) -> Result<(), InstallablePlanError> {
     use asap_executor::physical_planner::compile;
+    let roots: Vec<PhysicalASAPNodeId> = refreshed.iter().map(|view| view.node_id).collect();
     // The refresh reads only the nodes below it: compile them, or the whole
     // plan once, for the executor's checks of the whole plan.
     let sub_dag;
     let below = match in_whole_plan {
         true => indexed.dag,
         false => {
-            sub_dag = indexed.dag_below(view.node_id);
+            sub_dag = indexed.dag_below(&roots);
             &sub_dag
         }
     };
     #[cfg(test)]
     tests::COMPILED_NODES.with(|count| count.set(count.get() + below.nodes.len()));
     // Raw data is read at each `Scan`, a raw data slot `compile` makes.
-    let read: Vec<MaterializedView> = view
-        .inputs
-        .materialized_views
+    let root_set: HashSet<PhysicalASAPNodeId> = roots.iter().copied().collect();
+    let mut read_ids = HashSet::new();
+    let read: Vec<MaterializedView> = refreshed
         .iter()
+        .flat_map(|view| &view.inputs.materialized_views)
+        .filter(|id| !root_set.contains(id) && read_ids.insert(**id))
         .filter_map(|id| views_by_node.get(id).map(|read| (*read).clone()))
         .collect();
-    compile(below, &read, &[view.node_id as u64])
+    let root_ids: Vec<u64> = roots.iter().map(|&id| id as u64).collect();
+    compile(below, &read, &root_ids)
         .map(|_| ())
         .map_err(
             |error| InstallablePlanError::MaterializedViewNotExecutable {
-                node_id: view.node_id,
-                reason: error.to_string(),
+                node_id: roots[0],
+                reason: match roots.as_slice() {
+                    [_] => error.to_string(),
+                    _ => format!("refreshed with views {roots:?}: {error}"),
+                },
             },
         )
 }
@@ -759,10 +801,11 @@ mod tests {
     /// A p99 panel over a 300 s window every second, with no raw data kept:
     /// one DDSketch per series in each of 300 one-second panes built at
     /// ingestion, about 300 materialized views in a DAG of about 900 nodes.
-    /// Each view's refresh is compiled over the nodes below it, and the
-    /// whole plan's checks run once, so the nodes handed to the executor
-    /// grow with the plan, not with the plan times its views (before, every
-    /// refresh compiled the whole plan). The count
+    /// Each refresh (here one refresh group of all the panes) is compiled
+    /// over the nodes below its views, and the whole plan's checks run once,
+    /// so the nodes handed to the executor grow with the plan, not with the
+    /// plan times its views (before, every refresh compiled the whole
+    /// plan). The count
     /// covers both `plan_installable`'s build and `check_compiles`.
     #[tokio::test]
     async fn view_refreshes_compile_only_the_nodes_below_them() {
diff --git a/crates/planner/src/lib.rs b/crates/planner/src/lib.rs
index dec904ca6..d0937d125 100644
--- a/crates/planner/src/lib.rs
+++ b/crates/planner/src/lib.rs
@@ -82,8 +82,8 @@ pub use asap_plan_selection::{
     AcceptedAccuracy, ErrorSelectingPlan, RejectionCause, SearchReport, SearchStatus, Selection,
 };
 pub use asap_types::ir::materialized_view::{
-    InstallableDagError, MaterializedView, MaterializedViewInputs, MaterializedViewRefresh,
-    RawDataRetention, RawDataWindow, ViewCoverage,
+    IngestionRefreshGroup, InstallableDagError, MaterializedView, MaterializedViewInputs,
+    MaterializedViewRefresh, RawDataRetention, RawDataWindow, RefreshWindow, ViewCoverage,
 };
 pub use asap_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPNodeId};
 pub use asap_types::ir::properties::summary_coverage::{ColumnIdentity, Constraint, SelectionBox};
diff --git a/crates/planner/tests/runtime_install.rs b/crates/planner/tests/runtime_install.rs
index c98e8b7e0..67abf04d6 100644
--- a/crates/planner/tests/runtime_install.rs
+++ b/crates/planner/tests/runtime_install.rs
@@ -4,21 +4,27 @@
 //! the compiler checks that no internal is used (CI runs
 //! `cargo check -p asap-planner --tests --locked` on its own for that).
 
+use std::cell::Cell;
 use std::collections::{BTreeMap, BTreeSet};
 use std::sync::Arc;
 
 use crate::time_series;
 use asap_executor::physical_planner::compile;
-use asap_executor::runtime::{Limits, RunContext, Scope};
+use asap_executor::runtime::{Limits, OutputStream, RunContext, Scope};
+use asap_executor::sources::{
+    DataStore, InMemoryDataStore, MaterializedViewRequest, RawDataRequest,
+};
 use asap_executor::values::{Batch, Value};
+use asap_executor::Error as ExecutorError;
 use asap_physical_asap_optimization::test_support::split_shared_by_timing;
 use asap_planner::{
     plan_installable, run_planning, CostInputs, DeploymentCapabilities, FrontendInput,
     InstallablePlan, InstallablePlanError, MaterializedView, MaterializedViewRefresh, OperatorKind,
     OptimizationInput, OptimizationPass, OptimizeError, PhysicalASAPDAG, PlanOutput,
-    PlanSearchSettings, PlanningModels, PlanningWorkload, SelectedPlanReport, SelectionBox,
-    Stage1ToStage3Pass, UserInput,
+    PlanSearchSettings, PlanningModels, PlanningWorkload, RefreshWindow, SelectedPlanReport,
+    SelectionBox, Stage1ToStage3Pass, UserInput,
 };
+use asap_types::ir::operator::Source;
 use asap_types::ir::schema::schema_support::with_promql_series_identity;
 use asap_types::ir::schema::{FieldDataType, SketchAlgorithm, SketchParams};
 use asap_types::ir::{
@@ -99,6 +105,35 @@ fn remote_write_samples() -> Vec<RawDataSample> {
     samples
 }
 
+thread_local! {
+    /// How many `read_raw_data` calls the runs on this thread made.
+    static RAW_DATA_READS: Cell<usize> = const { Cell::new(0) };
+}
+
+/// A store counting its `read_raw_data` calls in [`RAW_DATA_READS`].
+struct CountingStore(InMemoryDataStore);
+
+impl DataStore for CountingStore {
+    fn read_raw_data(
+        &self,
+        source: &Source,
+        request: &RawDataRequest,
+        context: RunContext,
+    ) -> Result<OutputStream<'_, Batch>, ExecutorError> {
+        RAW_DATA_READS.with(|reads| reads.set(reads.get() + 1));
+        self.0.read_raw_data(source, request, context)
+    }
+
+    fn read_materialized_view(
+        &self,
+        view: &MaterializedView,
+        request: &MaterializedViewRequest,
+        context: RunContext,
+    ) -> Result<OutputStream<'_, Batch>, ExecutorError> {
+        self.0.read_materialized_view(view, request, context)
+    }
+}
+
 /// Compile `ids` of `dag` over the stored states `views` (by node id) and
 /// run them in `scope` over the raw `samples`, as a runtime does with the
 /// executor: each id's output batches.
@@ -129,7 +164,9 @@ fn run(
     for (id, batch) in views {
         store = store.with_materialized_view(id as _, vec![batch]).unwrap();
     }
-    let instance = compiled.instantiate(Arc::new(store)).unwrap();
+    let instance = compiled
+        .instantiate(Arc::new(CountingStore(store)))
+        .unwrap();
     block_on(async {
         // The run's scope (ingestion window or query time) and limits.
         let context = RunContext::new(scope, Limits::default()).unwrap();
@@ -252,6 +289,8 @@ struct DataPlaneRun {
     /// How many materialized views this evaluation refreshed (ingestion or
     /// query runs); the others it read came from the store.
     outputs_built: usize,
+    /// How many `read_raw_data` calls those refreshes made.
+    raw_data_reads_refreshing: usize,
 }
 
 /// One summary state this test stores: its refreshed state, and how long
@@ -309,8 +348,9 @@ fn query_roots(plan: &InstallablePlan) -> Vec<u64> {
 
 /// The data plane at evaluation time `evaluation_ms`: from `plan_json`,
 /// `samples` and the states earlier evaluations left in `store`, refresh each
-/// materialized view the store lacks once, in the plan's order, from its
-/// raw-data windows and the views it reads; answer every query; drop the
+/// materialized view the store lacks once, from its raw-data windows and the
+/// views it reads: the views refreshed at ingestion by refresh group, one
+/// ingestion run per group and span, then the others; answer every query; drop the
 /// states past their retention; and answer the queries again with the
 /// query-time reference.
 fn run_data_plane(
@@ -359,9 +399,9 @@ fn run_data_plane_on(
     let mut inputs = BTreeMap::new();
     // How many materialized views this evaluation refreshes.
     let mut outputs_built = 0;
-    // Each view's state and the time span of its data at this
+    // Each view's state key and the time span of its data at this
     // evaluation, `(start_ms, end_ms]`, by node id.
-    let mut built: BTreeMap<u64, (Batch, (i64, i64))> = BTreeMap::new();
+    let mut planned: BTreeMap<u64, (String, (i64, i64))> = BTreeMap::new();
     for view in &views {
         // The evaluation its windows end at: this one, or for a tumbling
         // at-ingestion view the newest point of its grid at or before it.
@@ -386,31 +426,81 @@ fn run_data_plane_on(
                 view.inputs
                     .materialized_views
                     .iter()
-                    .map(|id| built[&(*id as u64)].1),
+                    .map(|id| planned[&(*id as u64)].1),
             )
             .reduce(|(a, b), (c, d)| (a.min(c), b.max(d)))
             .expect("a materialized view reads something");
-        let key = state_key(view, built_for_ms);
-        if !store.contains_key(&key) {
-            // The views its refresh reads; its raw data is bound at its `Scan`s.
-            let mut sources = BTreeMap::new();
-            for id in &view.inputs.materialized_views {
-                sources.insert(*id as u64, built[&(*id as u64)].0.clone());
-            }
-            // The run that refreshes it: an ingestion run over its data's span
-            // once its inputs are complete (each window ends at the run's
-            // end), or the query run of this evaluation.
-            let scope = match view.refresh() {
-                MaterializedViewRefresh::AtIngestion => Scope::Ingestion {
-                    window_start_ms: span.0,
-                    window_end_ms: span.1,
-                },
-                MaterializedViewRefresh::OnFirstRead => Scope::Query {
-                    evaluation_time_ms: evaluation_ms,
-                },
+        planned.insert(view.node_id as u64, (state_key(view, built_for_ms), span));
+    }
+    // The runs that refresh the views the store lacks: for each refresh
+    // group, one ingestion run per span its views' data covers, with every
+    // view of the group over that span as a root (a node they share runs
+    // once); then each view refreshed on first read in this evaluation's
+    // query run.
+    let mut refreshes: Vec<(Vec<&MaterializedView>, Scope)> = Vec::new();
+    for group in installed.ingestion_refresh_groups().unwrap() {
+        let mut by_span: BTreeMap<(i64, i64), Vec<&MaterializedView>> = BTreeMap::new();
+        for member in &group.views {
+            let view = views
+                .iter()
+                .find(|view| view.node_id == member.node_id)
+                .unwrap();
+            by_span
+                .entry(planned[&(view.node_id as u64)].1)
+                .or_default()
+                .push(view);
+        }
+        for (span, members) in by_span {
+            let scope = Scope::Ingestion {
+                window_start_ms: span.0,
+                window_end_ms: span.1,
             };
-            // The batches that run produced, in one batch, as the store keeps it.
-            let batches = run(dag, sources, samples, &[view.node_id as u64], scope).remove(0);
+            refreshes.push((members, scope));
+        }
+    }
+    for view in views
+        .iter()
+        .filter(|view| view.refresh() == MaterializedViewRefresh::OnFirstRead)
+    {
+        let scope = Scope::Query {
+            evaluation_time_ms: evaluation_ms,
+        };
+        refreshes.push((vec![view], scope));
+    }
+    let raw_data_reads_before = RAW_DATA_READS.with(Cell::get);
+    for (members, scope) in refreshes {
+        // The views of the run whose state the store lacks, one per state.
+        let mut roots: Vec<&MaterializedView> = Vec::new();
+        for view in members {
+            let key = &planned[&(view.node_id as u64)].0;
+            if !store.contains_key(key)
+                && !roots
+                    .iter()
+                    .any(|root| &planned[&(root.node_id as u64)].0 == key)
+            {
+                roots.push(view);
+            }
+        }
+        if roots.is_empty() {
+            continue;
+        }
+        // The views the run reads that it does not refresh, from the
+        // store; its raw data is bound at its `Scan`s.
+        let mut sources = BTreeMap::new();
+        for id in roots
+            .iter()
+            .flat_map(|view| &view.inputs.materialized_views)
+        {
+            if !roots.iter().any(|root| root.node_id == *id) {
+                let key = &planned[&(*id as u64)].0;
+                sources.insert(*id as u64, store[key].state.clone());
+            }
+        }
+        let root_ids: Vec<u64> = roots.iter().map(|view| view.node_id as u64).collect();
+        // Each root's batches, in one batch per root, as the store keeps it.
+        let outputs = run(dag, sources, samples, &root_ids, scope);
+        for (view, batches) in roots.into_iter().zip(outputs) {
+            let (key, span) = &planned[&(view.node_id as u64)];
             let schema = batches[0].schema().clone();
             let state_rows = batches.iter().flat_map(|b| b.to_rows().to_vec()).collect();
             store.insert(
@@ -423,11 +513,17 @@ fn run_data_plane_on(
             );
             outputs_built += 1;
         }
+    }
+    let raw_data_reads_refreshing = RAW_DATA_READS.with(Cell::get) - raw_data_reads_before;
+    // Each view's state and the span of its data, by node id.
+    let mut built: BTreeMap<u64, (Batch, (i64, i64))> = BTreeMap::new();
+    for view in &views {
+        let (key, span) = &planned[&(view.node_id as u64)];
         // One stored state serves every view with its key, so it is kept
         // for the largest retention any of them requires.
-        let stored = store.get_mut(&key).unwrap();
+        let stored = store.get_mut(key).unwrap();
         stored.retention_ms = stored.retention_ms.max(view.retention_ms());
-        built.insert(view.node_id as u64, (stored.state.clone(), span));
+        built.insert(view.node_id as u64, (stored.state.clone(), *span));
     }
     // The query run reads the views it reaches from the queries' roots
     // without passing another view; those below are read only through the
@@ -489,6 +585,122 @@ fn run_data_plane_on(
         answers_per_query,
         reference_answers_per_query,
         outputs_built,
+        raw_data_reads_refreshing,
+    }
+}
+
+/// The sketch algorithm of the panes view `id` of `plan` builds.
+fn pane_sketch(plan: &InstallablePlan, id: usize) -> SketchAlgorithm {
+    let node = plan.dag.nodes.iter().find(|node| node.id == id).unwrap();
+    match &node.payload {
+        asap_executor::planner::ir::physical_export::PhysicalASAPOperatorPayload::ASAP(
+            ASAPOp::SummaryAgg {
+                family: FieldDataType::Sketch(kind, _),
+                ..
+            },
+        ) => *kind.algorithm(),
+        other => panic!("not a sketch pane: {other:?}"),
+    }
+}
+
+/// Issue #1151's example: a p90 at rank error 0.01 over a minute and a p99
+/// at relative error 0.01 over five minutes, both every 10 s, over 1,000
+/// ingesting non-negative `latency_ms` gauges sampled every 10 s, on a
+/// deployment that keeps no raw data: a KLL chain and a DDSketch chain,
+/// both of 10 s panes, kept at ingestion over one `Scan`. Their panes are one refresh group, so each
+/// 10 s window is read from raw data once, in one ingestion run that builds
+/// both panes ending there: 30 reads for the 30 pane windows of the first
+/// evaluation (36 when each pane view had its own run). Each state equals
+/// the one a run of its view alone builds, and both queries answer as the
+/// query-time reference.
+#[tokio::test]
+async fn panes_of_one_refresh_window_read_each_raw_window_once() {
+    let entry = |query: &str, metric: &str| {
+        format!(
+            r#"{{"query": {query:?}, "demand": {{"fixed_interval": 10000}},
+              "requirements": {{"accuracy": {{"accuracy_metric": {metric:?}, "max_error": 0.01,
+                  "failure_probability": 0.01, "accuracy_scope": "whole_result"}}}},
+              "predictability": {{"predictable": {{"known_at": null}}}}}}"#
+        )
+    };
+    let workload: PlanningWorkload = serde_json::from_str(&format!(
+        r#"{{"query_workload": {{"language": "promql", "repeating_queries": [{}, {}]}},
+            "data_workload": {{"arrival": {{"ingesting": {{"rows_per_second": 100.0}}}},
+              "sources": {{"latency_ms": {{"series_count": 1000, "sample_interval": 10000,
+                "metric_type": "gauge", "values_non_negative": true}}}}}}}}"#,
+        entry("quantile_over_time(0.9, latency_ms[1m])", "rank"),
+        entry("quantile_over_time(0.99, latency_ms[5m])", "relative"),
+    ))
+    .unwrap();
+    let capabilities = DeploymentCapabilities {
+        raw_data_retained: false,
+        ..DeploymentCapabilities::default()
+    };
+    let plan_json = plan_json_for(workload, &capabilities).await;
+    let installed: InstallablePlan = serde_json::from_str(&plan_json).unwrap();
+    installed.check_compiles().unwrap();
+    // One refresh group: the six 10 s KLL panes and the thirty 10 s
+    // DDSketch panes.
+    let groups = installed.ingestion_refresh_groups().unwrap();
+    assert_eq!(groups.len(), 1, "{groups:#?}");
+    assert_eq!(
+        groups[0].refresh_window,
+        Some(RefreshWindow {
+            width_ms: 10_000,
+            phase_ms: 0
+        })
+    );
+    let sketches: Vec<SketchAlgorithm> = groups[0]
+        .views
+        .iter()
+        .map(|view| pane_sketch(&installed, view.node_id))
+        .collect();
+    let count = |algorithm| sketches.iter().filter(|a| **a == algorithm).count();
+    assert_eq!(count(SketchAlgorithm::Kll), 6, "{sketches:?}");
+    assert_eq!(count(SketchAlgorithm::DDSketch), 30, "{sketches:?}");
+
+    let samples = remote_write_samples();
+    let mut store = SummaryStore::new();
+    let evaluated = run_data_plane(&plan_json, &samples, EVALUATION_MS, &mut store);
+    assert_eq!(evaluated.outputs_built, 36);
+    assert_eq!(evaluated.raw_data_reads_refreshing, 30);
+    assert_eq!(
+        evaluated.answers_per_query,
+        evaluated.reference_answers_per_query
+    );
+    assert_eq!(evaluated.answers_per_query[0].len(), 2);
+
+    // Each pane's state equals the one an ingestion run of its view alone
+    // builds.
+    for view in &groups[0].views {
+        let span = view.inputs.raw_data_windows()[0].bounds_ms(EVALUATION_MS);
+        let scope = Scope::Ingestion {
+            window_start_ms: span.0,
+            window_end_ms: span.1,
+        };
+        let alone = run(
+            &installed.dag,
+            BTreeMap::new(),
+            &samples,
+            &[view.node_id as u64],
+            scope,
+        )
+        .remove(0);
+        // `Value` has no equality; its debug form shows each state.
+        let rows = format!(
+            "{:?}",
+            alone
+                .iter()
+                .flat_map(|b| b.to_rows().to_vec())
+                .collect::<Vec<_>>()
+        );
+        let stored = &store[&state_key(view, EVALUATION_MS)].state;
+        assert_eq!(
+            format!("{:?}", stored.to_rows()),
+            rows,
+            "view {}",
+            view.node_id
+        );
     }
 }
 
diff --git a/crates/types/src/ir/materialized_view.rs b/crates/types/src/ir/materialized_view.rs
index 6c2cc7420..0c932c78c 100644
--- a/crates/types/src/ir/materialized_view.rs
+++ b/crates/types/src/ir/materialized_view.rs
@@ -18,10 +18,13 @@
 //! before it. A runtime refreshes a view once all of its inputs are
 //! complete: each raw-data window's samples have arrived (sealed) and each
 //! view it reads is refreshed. It compiles the DAG with the executor for
-//! that view's node, with each view it reads as a materialized view slot
-//! and raw data read by `Source` at each `Scan`. It refreshes the view in an
-//! ingestion run ([`MaterializedViewRefresh::AtIngestion`]) or in the query
-//! run of the evaluation that first reads it
+//! that view's node (for a view refreshed at ingestion, together with the
+//! rest of its refresh group, below), with each view it reads that the run
+//! does not refresh as a materialized view slot and raw data read by
+//! `Source` at each `Scan`. It
+//! refreshes the view in an ingestion run
+//! ([`MaterializedViewRefresh::AtIngestion`]) or in the query run of the
+//! evaluation that first reads it
 //! ([`MaterializedViewRefresh::OnFirstRead`]), keeps the result, and binds
 //! it the same way wherever the DAG reads that node. An ingestion run over
 //! `(window_start_ms, window_end_ms]` refreshes the windows ending at
@@ -35,6 +38,24 @@
 //! A query run binds the views it reaches from the roots without passing
 //! another view; a view only other views read is read through them.
 //!
+//! ## Refresh groups
+//!
+//! Views refreshed in an ingestion run are refreshed by **refresh group**
+//! ([`PhysicalASAPDAG::ingestion_refresh_groups`]): the views whose windows
+//! have one width `W` and end at one offset `p` modulo `W` (their
+//! [`RefreshWindow`]; a view with an unbounded window is a group of its
+//! own). A runtime refreshes a group with one ingestion run per window
+//! `(e - W, e]`, `e ≡ p (mod W)`. The run's roots are the views of the group
+//! whose state for the window ending at `e` is not stored yet: in steady
+//! state the newest pane of each chain, since older panes are the same
+//! states, stored by earlier runs. A view of the group that a root reads is
+//! computed in the run; views of other groups are materialized view slots.
+//! A node several roots share runs once per run, so a `Scan` below a KLL
+//! pane chain and a DDSketch pane chain of equal pane width reads each
+//! pane's raw data once. Stage 3's cost assumes this grouping; a runtime
+//! that splits a group computes the same states at a higher cost than
+//! priced.
+//!
 //! Each view's `coverage` says what its stored state is: a stable id of its
 //! definition (what it computes) and its selection (which rows went in,
 //! with time relative to the evaluation time). Two views with the same
@@ -50,8 +71,8 @@ use serde::{Deserialize, Serialize};
 use thiserror::Error;
 
 use super::physical_export::{
-    PhysicalASAPDAG, PhysicalASAPDAGNode, PhysicalASAPDAGValidationError, PhysicalASAPNodeId,
-    ScanWindow,
+    offset_below, PhysicalASAPDAG, PhysicalASAPDAGNode, PhysicalASAPDAGValidationError,
+    PhysicalASAPNodeId, ScanWindow,
 };
 use crate::ir::operator::asap::ASAPOp;
 use crate::ir::operator::node::{Operator, OperatorNode};
@@ -228,6 +249,49 @@ pub enum MaterializedViewRefresh {
     OnFirstRead,
 }
 
+/// 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, PartialOrd, Ord)]
+pub struct RefreshWindow {
+    /// Width of each window.
+    pub width_ms: u64,
+    /// Where each window ends, modulo `width_ms`.
+    pub phase_ms: u64,
+}
+
+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> {
+        let phase_ms = end_offset_ms
+            .checked_rem_euclid(width_ms)
+            .filter(|_| width_ms > 0)?;
+        Some(RefreshWindow {
+            width_ms: u64::try_from(width_ms).ok()?,
+            phase_ms: u64::try_from(phase_ms).ok()?,
+        })
+    }
+}
+
+/// 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 {
+    /// 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>,
+    /// The views, in dependency order.
+    pub views: Vec<MaterializedView>,
+}
+
 /// How far back from the evaluation time a runtime keeps a raw-data source
 /// for the DAG's query-time scans ([`PhysicalASAPDAG::raw_data_retention`]).
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -371,6 +435,147 @@ impl PhysicalASAPDAG {
         Ok(views)
     }
 
+    /// 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> {
+        Ok(self.ingestion_refresh_groups_of(&self.materialized_views()?))
+    }
+
+    /// [`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> {
+        let views: Vec<MaterializedView> = views
+            .iter()
+            .filter(|view| view.refresh == MaterializedViewRefresh::AtIngestion)
+            .cloned()
+            .collect();
+        let refresh_windows = self.refresh_windows(views.iter().map(|view| view.node_id));
+        let mut groups: Vec<IngestionRefreshGroup> = Vec::new();
+        let mut group_by_window: HashMap<RefreshWindow, usize> = HashMap::new();
+        let mut group_by_view: HashMap<PhysicalASAPNodeId, usize> = HashMap::new();
+        for view in views {
+            let refresh_window = refresh_windows[&view.node_id];
+            let group = match refresh_window.and_then(|window| group_by_window.get(&window)) {
+                Some(&group) => group,
+                None => {
+                    groups.push(IngestionRefreshGroup {
+                        refresh_window,
+                        views: Vec::new(),
+                    });
+                    if let Some(window) = refresh_window {
+                        group_by_window.insert(window, groups.len() - 1);
+                    }
+                    groups.len() - 1
+                }
+            };
+            group_by_view.insert(view.node_id, group);
+            groups[group].views.push(view);
+        }
+        // The other groups each group reads views of.
+        let reads: Vec<Vec<usize>> = groups
+            .iter()
+            .enumerate()
+            .map(|(group, of)| {
+                of.views
+                    .iter()
+                    .flat_map(|view| &view.inputs.materialized_views)
+                    .filter_map(|read| group_by_view.get(read).copied())
+                    .filter(|&read| read != group)
+                    .collect()
+            })
+            .collect();
+        // Each group after those it reads: a depth-first postorder. A view's
+        // window holds the windows of the views it reads, all ending where
+        // it ends, so two groups never read each other.
+        fn visit(group: usize, reads: &[Vec<usize>], placed: &mut [bool], order: &mut Vec<usize>) {
+            if std::mem::replace(&mut placed[group], true) {
+                return;
+            }
+            for &read in &reads[group] {
+                visit(read, reads, placed, order);
+            }
+            order.push(group);
+        }
+        let mut placed = vec![false; groups.len()];
+        let mut order = Vec::with_capacity(groups.len());
+        for group in 0..groups.len() {
+            visit(group, &reads, &mut placed, &mut order);
+        }
+        let mut groups: Vec<Option<IngestionRefreshGroup>> = groups.into_iter().map(Some).collect();
+        order
+            .into_iter()
+            .filter_map(|group| groups[group].take())
+            .collect()
+    }
+
+    /// The refresh window of `node` ([`RefreshWindow::of_window`]): for a
+    /// `SummaryTumblingWindowChain` (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> {
+        self.refresh_windows([node])[&node]
+    }
+
+    /// [`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>> {
+        let payloads = self.payloads_by_id();
+        let (chains, others): (Vec<_>, Vec<_>) = nodes.into_iter().partition(|id| {
+            matches!(
+                payloads.get(id),
+                Some(Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { .. }))
+            )
+        });
+        let mut windows: HashMap<PhysicalASAPNodeId, Option<RefreshWindow>> = self
+            .view_windows(others)
+            .into_iter()
+            .map(|(id, window)| {
+                let refresh_window = match window {
+                    ScanWindow::Within {
+                        end_offset_ms,
+                        width_ms,
+                    } => RefreshWindow::of_window(end_offset_ms, width_ms),
+                    ScanWindow::Unbounded => None,
+                };
+                (id, refresh_window)
+            })
+            .collect();
+        for chain in chains {
+            let Some(Operator::ASAP(ASAPOp::SummaryTumblingWindowChain {
+                child,
+                tumbling_window_width_ms,
+                ..
+            })) = payloads.get(&chain)
+            else {
+                continue;
+            };
+            // The newest pane ends at the offsets of the shifts below the chain.
+            let end_offset_ms = offset_below(*child, |id| payloads.get(&id).copied());
+            let width_ms = i64::try_from(*tumbling_window_width_ms).unwrap_or(0);
+            windows.insert(
+                chain,
+                end_offset_ms.and_then(|offset| RefreshWindow::of_window(offset, width_ms)),
+            );
+        }
+        windows
+    }
+
     /// Sets `retention_ms` on every materialized view and every query-time
     /// `Scan` of an installed DAG (pane chains expanded), and returns the
     /// views ([`Self::materialized_views`]); `Err` when `deadline` passes.
diff --git a/crates/types/src/ir/physical_export.rs b/crates/types/src/ir/physical_export.rs
index 8cf99d175..03263faf9 100644
--- a/crates/types/src/ir/physical_export.rs
+++ b/crates/types/src/ir/physical_export.rs
@@ -318,6 +318,25 @@ pub fn time_range_window<'a>(
     }
 }
 
+/// 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> {
+    let mut end_offset_ms = 0i64;
+    let mut below = payload_of(node);
+    while let Some(Operator::NonASAP(NonASAPOp::TimeShift { shift, child })) = below {
+        if shift.at.is_some() {
+            return None;
+        }
+        end_offset_ms = end_offset_ms.checked_add(shift.offset_ms)?;
+        below = payload_of(*child);
+    }
+    Some(end_offset_ms)
+}
+
 /// Why `PhysicalASAPDAG::with_every_node_at_query_time` failed.
 #[derive(Debug, Error)]
 pub enum ReferencePlanError {
@@ -472,10 +491,11 @@ impl PhysicalASAPDAG {
 
     /// The interval of data the output of `node` summarizes, back from the
     /// evaluation time `T`: the union of the windows
-    /// ([`time_range_window`]) of every `TimeRange` below it. For an
-    /// expanded pane, its pane; for a view built from panes, all of them.
-    /// `Unbounded` when no `TimeRange` is below it or one of them is
-    /// unbounded (a whole SQL table, an `@` anchor). O(nodes + edges).
+    /// ([`time_range_window`]) of every `TimeRange` below it, and of every
+    /// unexpanded `SummaryTumblingWindowChain` (all its panes). For an expanded pane, its
+    /// pane; for a view built from panes, all of them. `Unbounded` when no
+    /// `TimeRange` or chain is below it or one of them is unbounded (a whole
+    /// SQL table, an `@` anchor). O(nodes + edges).
     #[cfg(any(test, feature = "test-support"))]
     pub fn view_window(&self, node: PhysicalASAPNodeId) -> ScanWindow {
         self.view_windows([node])[&node]
@@ -500,12 +520,35 @@ impl PhysicalASAPDAG {
                 if !seen.insert(id) {
                     continue;
                 }
-                if let Some(Operator::NonASAP(NonASAPOp::TimeRange { .. })) = payloads.get(&id) {
-                    let read = time_range_window(id, |id| payloads.get(&id).copied());
-                    window = Some(window.map_or(read, |window| window.union(read)));
-                    continue;
-                }
-                pending.extend(inputs.get(&id).into_iter().flatten());
+                let read = match payloads.get(&id) {
+                    Some(Operator::NonASAP(NonASAPOp::TimeRange { .. })) => {
+                        time_range_window(id, |id| payloads.get(&id).copied())
+                    }
+                    Some(Operator::ASAP(ASAPOp::SummaryTumblingWindowChain {
+                        child,
+                        tumbling_window_width_ms,
+                        tumbling_window_count,
+                        ..
+                    })) => {
+                        let width_ms =
+                            tumbling_window_width_ms.checked_mul(u64::from(*tumbling_window_count));
+                        match (
+                            offset_below(*child, |id| payloads.get(&id).copied()),
+                            width_ms,
+                        ) {
+                            (Some(end_offset_ms), Some(width_ms)) => ScanWindow::Within {
+                                end_offset_ms,
+                                width_ms: i64::try_from(width_ms).unwrap_or(i64::MAX),
+                            },
+                            _ => ScanWindow::Unbounded,
+                        }
+                    }
+                    _ => {
+                        pending.extend(inputs.get(&id).into_iter().flatten());
+                        continue;
+                    }
+                };
+                window = Some(window.map_or(read, |window| window.union(read)));
             }
             window.unwrap_or(ScanWindow::Unbounded)
         };
@@ -516,7 +559,9 @@ impl PhysicalASAPDAG {
     }
 
     /// Each node's operator, by id.
-    fn payloads_by_id(&self) -> HashMap<PhysicalASAPNodeId, &PhysicalASAPOperatorPayload> {
+    pub(crate) fn payloads_by_id(
+        &self,
+    ) -> HashMap<PhysicalASAPNodeId, &PhysicalASAPOperatorPayload> {
         self.nodes
             .iter()
             .map(|node| (node.id, &node.payload))
diff --git a/crates/types/tests/materialized_view.rs b/crates/types/tests/materialized_view.rs
index cb32dcd00..40e58273d 100644
--- a/crates/types/tests/materialized_view.rs
+++ b/crates/types/tests/materialized_view.rs
@@ -6,18 +6,24 @@ use std::collections::HashMap;
 use std::rc::Rc;
 use std::time::Duration;
 
-use asap_types::ir::materialized_view::{InstallableDagError, MaterializedViewRefresh};
+use asap_types::ir::materialized_view::{
+    InstallableDagError, MaterializedViewRefresh, RefreshWindow,
+};
 use asap_types::ir::operator::operator_properties::{Reduction, Source};
+use asap_types::ir::operator::pane_chain::expand_pane_chains_in_workload;
 use asap_types::ir::operator::TimeShift;
-use asap_types::ir::physical_export::{compile_physical_asap_dag, PhysicalASAPDAG};
+use asap_types::ir::physical_export::{
+    compile_physical_asap_dag, compile_physical_asap_workload, PhysicalASAPDAG,
+};
 use asap_types::ir::scalar::ColumnRef;
 use asap_types::ir::schema::{
     DataType, Field, FieldDataType, GroupingStrategy, Schema, SketchAlgorithm, SketchKind,
     SketchParams, SummaryUpdate,
 };
 use asap_types::ir::{
-    apply_materialization_timings, ASAPOp, MaterializationAssignment, NodeToTimedNodeCache,
-    NonASAPOp, Operator, OperatorNode, TimeRangeKind,
+    apply_materialization_timings, apply_materialization_timings_to_roots, ASAPOp,
+    MaterializationAssignment, NodeToTimedNodeCache, NonASAPOp, Operator, OperatorNode, QueryRoot,
+    TimeRangeKind,
 };
 use asap_types::physical::{Materialization, Medium};
 
@@ -57,10 +63,9 @@ fn pane(scan: &Rc<OperatorNode>, shift_ms: i64) -> Rc<OperatorNode> {
     .unwrap()
 }
 
-/// The panes at offsets 0 and 1 min kept at ingestion, their merge kept
-/// across evaluations, exported: `(dag, merge's node id)`.
-fn panes_and_kept_merge() -> (PhysicalASAPDAG, usize) {
-    let scan = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan {
+/// `m` with columns `ts`, `value` and `job`.
+fn scan_m() -> Rc<OperatorNode> {
+    OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan {
         source: Source::TimeSeries { metric: "m".into() },
         predicates: vec![],
         schema: Schema::with_time_index(
@@ -73,7 +78,13 @@ fn panes_and_kept_merge() -> (PhysicalASAPDAG, usize) {
             vec![],
         ),
     }))
-    .unwrap();
+    .unwrap()
+}
+
+/// The panes at offsets 0 and 1 min kept at ingestion, their merge kept
+/// across evaluations, exported: `(dag, merge's node id)`.
+fn panes_and_kept_merge() -> (PhysicalASAPDAG, usize) {
+    let scan = scan_m();
     let panes = [pane(&scan, 0), pane(&scan, 60_000)];
     let merge = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge {
         children: panes.to_vec(),
@@ -175,3 +186,308 @@ fn the_derivation_stops_at_its_deadline() {
         Err(InstallableDagError::DeadlineExceeded)
     );
 }
+
+/// A quantile sketch of `algorithm` (KLL k = 200 or DDSketch α = 0.01).
+fn quantile_sketch(algorithm: SketchAlgorithm) -> FieldDataType {
+    let params = match algorithm {
+        SketchAlgorithm::Kll => SketchParams::Kll {
+            k: 200,
+            failure_probability: None,
+        },
+        _ => SketchParams::DDSketch { alpha: 0.01 },
+    };
+    FieldDataType::Sketch(SketchKind::new(algorithm, params), Default::default())
+}
+
+/// `tumbling_window_count` panes of `tumbling_window_width_ms` of `family` per `job` over `input`.
+fn chain(
+    input: &Rc<OperatorNode>,
+    family: FieldDataType,
+    tumbling_window_width_ms: u64,
+    tumbling_window_count: u32,
+) -> Rc<OperatorNode> {
+    OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryTumblingWindowChain {
+        child: Rc::clone(input),
+        family,
+        input: SummaryUpdate::column(ColumnRef::Named("value".into())),
+        reduction: Reduction::by(vec![2]),
+        grouping: GroupingStrategy::default(),
+        filter: None,
+        tumbling_window_width_ms,
+        tumbling_window_count,
+    }))
+    .unwrap()
+}
+
+/// Three queries over one `Scan` of `m`, each merging a chain kept at
+/// ingestion: a KLL of six 10 s panes, a DDSketch of thirty 10 s panes, and
+/// a KLL of five 60 s panes. The DAG Stage 3 costs (chains unexpanded) and
+/// the installed DAG (chains expanded into panes, retention set).
+fn three_chains_over_one_scan() -> (PhysicalASAPDAG, PhysicalASAPDAG) {
+    let scan = scan_m();
+    let chains = [
+        chain(&scan, quantile_sketch(SketchAlgorithm::Kll), 10_000, 6),
+        chain(
+            &scan,
+            quantile_sketch(SketchAlgorithm::DDSketch),
+            10_000,
+            30,
+        ),
+        chain(&scan, quantile_sketch(SketchAlgorithm::Kll), 60_000, 5),
+    ];
+    let mut assignment = MaterializationAssignment::default();
+    let roots: Vec<QueryRoot> = chains
+        .iter()
+        .map(|chain| {
+            assignment.set(chain, Materialization::MatAtIngestion(Medium::Memory));
+            QueryRoot::Operator(
+                OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge {
+                    children: vec![Rc::clone(chain)],
+                    newest_tumbling_windows: None,
+                    group_by: None,
+                }))
+                .unwrap(),
+            )
+        })
+        .collect();
+    let timed = apply_materialization_timings_to_roots(&roots, &assignment).unwrap();
+    let unexpanded = compile_physical_asap_workload(&timed).unwrap();
+    let expanded = expand_pane_chains_in_workload(&timed).unwrap();
+    let mut installed = compile_physical_asap_workload(&expanded.roots).unwrap();
+    installed
+        .set_retention(&HashMap::new(), &HashMap::new(), None)
+        .unwrap();
+    (unexpanded, installed)
+}
+
+/// The width and sketch of the panes `node` of `dag` builds.
+fn pane_shape(dag: &PhysicalASAPDAG, node: usize) -> (u64, SketchAlgorithm) {
+    let payload = &dag.nodes.iter().find(|n| n.id == node).unwrap().payload;
+    let (family, width_ms) = match payload {
+        Operator::ASAP(ASAPOp::SummaryTumblingWindowChain {
+            family,
+            tumbling_window_width_ms,
+            ..
+        }) => (family, *tumbling_window_width_ms),
+        Operator::ASAP(ASAPOp::SummaryAgg { family, child, .. }) => {
+            let range = &dag.nodes.iter().find(|n| n.id == *child).unwrap().payload;
+            let Operator::NonASAP(NonASAPOp::TimeRange { range, .. }) = range else {
+                panic!("a pane reads a TimeRange")
+            };
+            (family, range.as_millis() as u64)
+        }
+        other => panic!("not a pane: {other:?}"),
+    };
+    let FieldDataType::Sketch(kind, _) = family else {
+        panic!("a sketch")
+    };
+    (width_ms, *kind.algorithm())
+}
+
+/// A chain kept at ingestion has its newest pane's refresh window on the
+/// DAG Stage 3 costs, and each of its expanded panes has the same one.
+#[test]
+fn a_chain_and_each_of_its_panes_have_one_refresh_window() {
+    let (unexpanded, installed) = three_chains_over_one_scan();
+    let window_of = |width_ms| RefreshWindow {
+        width_ms,
+        phase_ms: 0,
+    };
+    let chains: Vec<usize> = unexpanded
+        .nodes
+        .iter()
+        .filter(|node| {
+            matches!(
+                node.payload,
+                Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { .. })
+            )
+        })
+        .map(|node| node.id)
+        .collect();
+    assert_eq!(chains.len(), 3);
+    for chain in chains {
+        let (width_ms, _) = pane_shape(&unexpanded, chain);
+        assert_eq!(unexpanded.refresh_window(chain), Some(window_of(width_ms)));
+    }
+    let views = installed.materialized_views().unwrap();
+    assert_eq!(views.len(), 6 + 30 + 5);
+    for view in &views {
+        let (width_ms, _) = pane_shape(&installed, view.node_id);
+        assert_eq!(
+            installed.refresh_window(view.node_id),
+            Some(window_of(width_ms))
+        );
+    }
+}
+
+/// The 10 s KLL and 10 s DDSketch panes are one refresh group, the 60 s
+/// KLL panes another; the merges, at query time, are in none.
+#[test]
+fn panes_of_one_width_form_one_refresh_group() {
+    let (_, installed) = three_chains_over_one_scan();
+    let groups = installed.ingestion_refresh_groups().unwrap();
+    let shapes: Vec<_> = groups
+        .iter()
+        .map(|group| {
+            let mut shapes: Vec<_> = group
+                .views
+                .iter()
+                .map(|view| pane_shape(&installed, view.node_id))
+                .collect();
+            shapes.sort_by_key(|(width_ms, algorithm)| (*width_ms, format!("{algorithm:?}")));
+            shapes.dedup();
+            (group.refresh_window, shapes)
+        })
+        .collect();
+    assert_eq!(
+        shapes,
+        [
+            (
+                Some(RefreshWindow {
+                    width_ms: 10_000,
+                    phase_ms: 0
+                }),
+                vec![
+                    (10_000, SketchAlgorithm::DDSketch),
+                    (10_000, SketchAlgorithm::Kll)
+                ]
+            ),
+            (
+                Some(RefreshWindow {
+                    width_ms: 60_000,
+                    phase_ms: 0
+                }),
+                vec![(60_000, SketchAlgorithm::Kll)]
+            ),
+        ]
+    );
+    assert_eq!(groups[0].views.len(), 6 + 30);
+    assert_eq!(groups[1].views.len(), 5);
+}
+
+/// A pane whose window ends 30 s after a multiple of its 60 s width (a
+/// `TimeShift(90 s)` below its range) is refreshed on another grid than a
+/// pane at no offset: another refresh group.
+#[test]
+fn a_shifted_pane_has_its_own_phase() {
+    let scan = scan_m();
+    let panes = [pane(&scan, 0), pane(&scan, 90_000)];
+    let mut assignment = MaterializationAssignment::default();
+    for pane in &panes {
+        assignment.set(pane, Materialization::MatAtIngestion(Medium::Memory));
+    }
+    let roots: Vec<QueryRoot> = panes
+        .iter()
+        .map(|pane| QueryRoot::Operator(Rc::clone(pane)))
+        .collect();
+    let timed = apply_materialization_timings_to_roots(&roots, &assignment).unwrap();
+    let mut dag = compile_physical_asap_workload(&timed).unwrap();
+    dag.set_retention(&HashMap::new(), &HashMap::new(), None)
+        .unwrap();
+    let mut windows: Vec<(u64, u64)> = dag
+        .ingestion_refresh_groups()
+        .unwrap()
+        .iter()
+        .map(|group| {
+            let window = group.refresh_window.unwrap();
+            (window.width_ms, window.phase_ms)
+        })
+        .collect();
+    windows.sort_unstable();
+    assert_eq!(windows, [(60_000, 0), (60_000, 30_000)]);
+}
+
+/// Two views kept at ingestion over rows with no time window (`Values`)
+/// have no refresh window: each is a group of its own.
+#[test]
+fn views_of_unbounded_windows_are_groups_of_their_own() {
+    let rows = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Values {
+        rows: vec![],
+        schema: scan_m().schema.clone(),
+    }))
+    .unwrap();
+    let mut assignment = MaterializationAssignment::default();
+    let roots: Vec<QueryRoot> = [SketchAlgorithm::Kll, SketchAlgorithm::DDSketch]
+        .into_iter()
+        .map(|algorithm| {
+            let build = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg {
+                child: Rc::clone(&rows),
+                family: quantile_sketch(algorithm),
+                input: SummaryUpdate::column(ColumnRef::Named("value".into())),
+                reduction: Reduction::by(vec![2]),
+                grouping: GroupingStrategy::default(),
+                filter: None,
+            }))
+            .unwrap();
+            assignment.set(&build, Materialization::MatAtIngestion(Medium::Memory));
+            QueryRoot::Operator(build)
+        })
+        .collect();
+    let timed = apply_materialization_timings_to_roots(&roots, &assignment).unwrap();
+    let mut dag = compile_physical_asap_workload(&timed).unwrap();
+    dag.set_retention(&HashMap::new(), &HashMap::new(), None)
+        .unwrap();
+    let groups = dag.ingestion_refresh_groups().unwrap();
+    assert_eq!(groups.len(), 2);
+    for group in &groups {
+        assert_eq!(group.refresh_window, None);
+        assert_eq!(group.views.len(), 1);
+    }
+}
+
+/// A rollup across jobs of a pane, both kept at ingestion, reads the pane's
+/// window: one refresh group, the pane listed before the rollup that reads
+/// it, so a run whose roots hold both computes the pane in the run.
+#[test]
+fn a_view_reading_a_view_of_its_group_follows_it() {
+    let scan = scan_m();
+    let pane = pane(&scan, 0);
+    let rollup = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge {
+        children: vec![Rc::clone(&pane)],
+        newest_tumbling_windows: None,
+        group_by: Some(vec![]),
+    }))
+    .unwrap();
+    let mut assignment = MaterializationAssignment::default();
+    for view in [&pane, &rollup] {
+        assignment.set(view, Materialization::MatAtIngestion(Medium::Memory));
+    }
+    let timed = apply_materialization_timings_to_roots(
+        &[QueryRoot::Operator(Rc::clone(&rollup))],
+        &assignment,
+    )
+    .unwrap();
+    let mut dag = compile_physical_asap_workload(&timed).unwrap();
+    dag.set_retention(&HashMap::new(), &HashMap::new(), None)
+        .unwrap();
+    let groups = dag.ingestion_refresh_groups().unwrap();
+    assert_eq!(groups.len(), 1);
+    let [pane_view, rollup_view] = groups[0].views.as_slice() else {
+        panic!("{groups:#?}")
+    };
+    assert_eq!(rollup_view.node_id, dag.roots[0]);
+    assert_eq!(rollup_view.inputs.materialized_views, [pane_view.node_id]);
+    assert_eq!(
+        groups[0].refresh_window,
+        Some(RefreshWindow {
+            width_ms: 60_000,
+            phase_ms: 0
+        })
+    );
+}
+
+/// On the DAG Stage 3 costs (chains unexpanded), a node above a chain has
+/// the window of all the chain's panes: a merge of six 10 s panes, refreshed
+/// at ingestion, has refresh window (60 s, 0).
+#[test]
+fn a_node_above_an_unexpanded_chain_has_the_window_of_its_panes() {
+    let (unexpanded, _) = three_chains_over_one_scan();
+    let merge_of_kll_10s = unexpanded.roots[0];
+    assert_eq!(
+        unexpanded.refresh_window(merge_of_kll_10s),
+        Some(RefreshWindow {
+            width_ms: 60_000,
+            phase_ms: 0
+        })
+    );
+}
diff --git a/docs/design_docs/physical-planning-and-deployment.md b/docs/design_docs/physical-planning-and-deployment.md
index 00c574e7a..7fced770c 100644
--- a/docs/design_docs/physical-planning-and-deployment.md
+++ b/docs/design_docs/physical-planning-and-deployment.md
@@ -288,8 +288,9 @@ window; that requires supported boundary handling or a different candidate.
 ## 3. Materialization → Physical DAG
 
 Operator names in the diagrams of this section and §4 (`NativeKllBuild`,
-`NativeKllMerge`, `NativeP50`, `NativeQuantile`, `KllStateOutput`) are
-illustrative. The executor implements them as `SummaryAgg` (a KLL build),
+`NativeDDSketchBuild`, `NativeKllMerge`, `NativeP50`, `NativeQuantile`,
+`KllStateOutput`) are illustrative. The executor implements them as
+`SummaryAgg` (a KLL or DDSketch build),
 `SummaryMerge` and `SummaryEvaluation` (a quantile readout).
 
 The **Physical Plan Compiler** consumes both computation semantics and maintenance
@@ -448,6 +449,9 @@ impl InstallablePlan {
     /// Every materialized view, in dependency order: its node, refresh mode
     /// (at ingestion, or on first read and kept), coverage, inputs and retention.
     pub fn materialized_views(&self) -> Result<Vec<MaterializedView>, InstallablePlanError>;
+    /// The views refreshed at ingestion, by refresh group (one refresh window
+    /// each), each group after the groups it reads.
+    pub fn ingestion_refresh_groups(&self) -> Result<Vec<IngestionRefreshGroup>, InstallablePlanError>;
     /// Per raw-data `Source` read at query time: how far back raw data must be kept.
     pub fn raw_data_retention(&self) -> Result<Vec<RawDataRetention>, InstallablePlanError>;
 }
@@ -455,9 +459,17 @@ pub struct RawDataRetention {
     pub source: Source,
     pub lookback_ms: Option<u64>,     // None = unbounded
 }
+pub struct IngestionRefreshGroup {
+    pub refresh_window: Option<RefreshWindow>, // None = an unbounded window, a group of its own
+    pub views: Vec<MaterializedView>,          // in dependency order
+}
+pub struct RefreshWindow {
+    pub width_ms: u64,                // width of each window
+    pub phase_ms: u64,                // where each window ends, modulo width_ms
+}
 ```
 
-Both derivations are methods on `PhysicalASAPDAG` in `asap-types`, so a
+These derivations are methods on `PhysicalASAPDAG` in `asap-types`, so a
 runtime needs no dependency on the planner crate.
 
 A **materialized view** is a node whose output is kept across evaluations
@@ -574,21 +586,72 @@ ingestion window, or a query evaluation time `T`.
 - In an ingestion run over `(window_start, window_end]`, each `TimeRange(w)` keeps `(window_end - w, window_end]`. A windowed raw data slot asks for `(window_start, window_end]`, and a materialized view slot for `(window_end - w, window_end]`. A view refreshed at ingestion whose raw windows, or the windows of the views it reads, end at different offsets cannot be built in one run, so installation refuses it.
 - A store never implements PromQL or SQL semantics. The compile check compiles these operators, so a plan whose window the executor cannot run is rejected before it is installed.
 
+### Refresh groups
+
+Two views refreshed at ingestion can be built in one ingestion run, with the
+run's interval equal to each one's window, exactly when they have the same
+**refresh window**: windows of width `W` that end `p` after a multiple of `W`
+(`0 ≤ p < W`), relative to the evaluation time. A view whose `view_window` is
+`(T - o - W, T - o]` has the refresh window `(W, o mod W)`. All panes of a
+pane chain have their newest pane's refresh window. The source, the summary
+family and where the state is kept do not enter. A view with an unbounded
+window is a group of its own.
+
+A **refresh group** is the set of views refreshed at ingestion that have one
+refresh window `(W, p)` (`PhysicalASAPDAG::refresh_window`,
+`ingestion_refresh_groups`). A runtime refreshes a refresh group with one
+ingestion run per window `(e - W, e]`, `e ≡ p (mod W)`:
+
+- The run's roots are the views of the group whose state for the window
+  ending at `e` is not stored yet. Pane `i` of a chain at evaluation `T` is
+  the same state as the newest pane at `T - i·W`, so in steady state the
+  roots are the newest pane of each chain of the group; the older panes are
+  already stored.
+- A view of the group that a root reads is computed in the run, not read
+  from a slot. The views of other groups that the roots read are
+  materialized view slots.
+- The executor runs a node that several roots share once per run, so a `Scan`
+  below several views of the group reads its raw data once per window.
+- Views of different refresh windows are refreshed in different runs. With a
+  10-second and a 60-second pane chain over one `Scan`, the 10-second chain
+  needs five pane ends the 60-second chain does not, so the `Scan` is read
+  once for each chain.
+
+The refresh window is derived from the DAG alone (time ranges, shifts and pane
+widths), like the views and their retention. Sealing, lateness, threading and
+the alignment of the window grid stay the deployment's.
+
+Stage 3's cost assumes this grouping (`asap_cost::execution_demand::computations`):
+a stateless ingestion-time node is charged once per refresh window of the
+views above it, not once per view. A runtime that splits a group (for
+example, because its sources seal at different times) computes the same
+states at a higher cost than priced, the same status pane reuse has. A
+runtime that updates states sample by sample passes each sample once
+whatever the windows, so it runs cheaper than priced.
+
 ### Running example
 
 For the KLL p50/p99 example, with one-minute panes materialized at ingestion
-over `Source::TimeSeries { metric: "latency" }`:
+over `Source::TimeSeries { metric: "latency" }`, plus a p99.9 with relative
+error answered by a DDSketch chain with one-minute panes over the same `Scan`:
 
 ```text
 Install (once):
-    views     = plan.materialized_views()?          // [pane nodes, refresh = at ingestion, retention]
-    refresh   = compile(dag, [], [pane_0])           // Scan → TimeRange(1m) → NativeKllBuild(k=200)
-    query     = compile(dag, views[pane_0..pane_4], [p50, p99])  // 5 materialized view slots → NativeKllMerge → NativeP50, NativeP99
-
-Each complete pane (12:04, 12:05], in an ingestion run:
+    views     = plan.materialized_views()?          // [KLL and DDSketch pane nodes, refresh = at ingestion, retention]
+    groups    = plan.ingestion_refresh_groups()?    // [refresh window (1m, 0): ten KLL and DDSketch pane views]
+    refresh   = compile(dag, [], [kll_pane_0, dd_pane_0])  // the newest pane of each chain
+                // Scan → TimeRange(1m) → NativeKllBuild(k=200)
+                //                      → NativeDDSketchBuild(α=0.01)
+    query     = compile(dag, views[kll_pane_0..kll_pane_4], [p50, p99])  // 5 materialized view slots → NativeKllMerge → NativeP50, NativeP99
+    (the p99.9 query DAG reads dd_pane_0..dd_pane_4 the same way)
+
+Each complete pane (12:04, 12:05], in one ingestion run whose roots are the
+group's views with no stored state for the pane ending 12:05 (kll_pane_0 and
+dd_pane_0; the older panes were stored by earlier runs):
     refresh.instantiate(store)
-        Scan calls read_raw_data(TimeSeries { latency }, time_range_ms = (12:04, 12:05])
-    store the KllState in the summary store under the view's coverage and interval
+        Scan calls read_raw_data(TimeSeries { latency }, time_range_ms = (12:04, 12:05]) once;
+        both builds read its rows
+    store the KllState and the DDSketchState in the summary store, each under its view's coverage and interval
 
 Query at T = 12:05 over (12:00, 12:05]:
     the installed DAG has the chain expanded: five pane views, pane i under TimeShift(i·1m)
@@ -636,7 +699,7 @@ The complete example makes the ownership boundary explicit:
 | **Physical Plan Compiler** | Lower to native KLL build, merge, and readout operators |
 | **Physical DAG** | Define precompute and query DAGs with typed input/output boundaries |
 | **Deployment interface** | Read the `latency` raw data slot with `read_raw_data`, and each pane's materialized view slot with `read_materialized_view`, from one `DataStore` |
-| **Deployment** | Schedule pane refreshes, store and resolve panes, run queries |
+| **Deployment** | Schedule one ingestion run per refresh group and window, store and resolve panes, run queries |
 
 ```text
 Logical:
@@ -662,7 +725,7 @@ does not introduce additional planning decisions.
 The physical layer does not own raw ingestion, storage formats, or decoding
 persisted bytes into typed state. It compiles computation over input slots:
 windows, summary build, merge (for example KLL merge), sketch estimates and
-exact finalization. The deployment schedules each pane's refresh, stores and
+exact finalization. The deployment schedules each refresh group's runs, stores and
 reads stored state, and serves raw data and views to the slots through its
 `DataStore`. The plan file is the only persisted handoff; a compiled
 physical DAG is derived from it by the executor and is not persisted.
@@ -680,6 +743,7 @@ operator/runtime fixtures:
 | --- | --- |
 | `kll_pane_execution::five_panes_roundtrip_and_shared_merge_runs_once` | One-minute pane refreshes from raw data → five materialized view slots → shared native merge → p50/p99; counts every sample once, checks adjacent aligned windows and instruments one merge start per run |
 | `kll_pane_execution::a_store_keyed_by_definition_and_time_range_answers_the_panes` | A store that finds states only by `(definition_id, time range)` answers the five-pane query: the panes share one definition and request the five ranges that tile the window |
+| `runtime_install::panes_of_one_refresh_window_read_each_raw_window_once` | A KLL chain and a DDSketch chain of 10 s panes over one `Scan` are one refresh group: each pane window is read from raw data once, in one ingestion run that builds both panes; each state equals that of a run of its view alone, and both queries answer as the query-time reference |
 | `kll_pane_execution::panes_reject_parameters_schema_and_missing_binding` | Parameter relabelling, incompatible schemas and missing views fail explicitly |
 | `precompute_candidates::grouped_rate_can_be_materialized_before_or_after_grouped_sum` | Stored Rate state and stored Sum state answer like query time; summing counters before Rate does not |
 | `sql_to_physical::sql_filter_grouped_sum_executes_and_rebinds` | SQL text → candidate search → physical compilation → shared Scan predicates and grouped summary execution; NULL samples are ignored and fresh bindings produce new results |