From b67cf979b4f81b42b2c6c0139a40eb8f079b506b Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Mon, 2 Nov 2026 21:37:16 +0000 Subject: [PATCH 1/2] fix(stage2): keep a tumbling-window chain at ingestion time only on its readers' intervals A tumbling-window chain built at ingestion time keeps windows on the grid of its width anchored at the epoch. Stage 2's ingestion check skipped readers with no declared phase, so a 15 s chain could serve a query evaluated every 10 s and answer from windows ending up to 10 s early (#1173). Every reader at a fixed interval now needs an interval the width divides (IngestionPruneReason::PanesOffInterval); the chain stays a query-time candidate otherwise. Regression tests: the same query every 10 s and every 15 s answers as the query-time reference at every evaluation over one minute (runtime_install), and a 1-min chain read every 90 s is not maintained at ingestion time. Closes #1173 --- crates/physical-asap-optimization/src/lib.rs | 75 ++++++++++++++-- crates/planner/tests/runtime_install.rs | 95 ++++++++++++++++++++ 2 files changed, 165 insertions(+), 5 deletions(-) diff --git a/crates/physical-asap-optimization/src/lib.rs b/crates/physical-asap-optimization/src/lib.rs index 36dc19ae4..5d47d5cf7 100644 --- a/crates/physical-asap-optimization/src/lib.rs +++ b/crates/physical-asap-optimization/src/lib.rs @@ -61,8 +61,9 @@ //! as Prometheus assumes (it rejects out-of-order samples). Table (SQL) //! sources and mixed data, whose at-rest part may load in any order, are //! not taken to be ordered; -//! * it is a tumbling-pane chain whose readers of known phase all evaluate -//! on its pane boundaries, anchored at the epoch, or a whole-window build whose readers +//! * it is a tumbling-pane chain whose readers all evaluate on its pane +//! boundaries, anchored at the epoch (each fixed interval a multiple of the +//! pane width, and each known phase on that grid), or a whole-window build whose readers //! all run at one fixed interval with agreeing declared phases, over a //! window that is not shifted and whose evaluations all end on the grid //! of its width anchored at the epoch (the interval and a declared phase @@ -201,6 +202,11 @@ pub(crate) enum IngestionPruneReason { /// which are anchored at the epoch (e.g. 20 s panes read at phase 5 s): /// built once each, its panes cannot end at every evaluation. PanesOffPhase, + /// A tumbling-window chain read at a fixed interval that its width does + /// not divide (e.g. 15 s tumbling windows read every 10 s): built once + /// each on the grid anchored at the epoch, its tumbling windows cannot + /// end at every evaluation. + PanesOffInterval, /// Something below it can only run at query time. UpstreamAtQueryTime(String), /// It is not a summary build (e.g. a shared `Scan`): only a summary's @@ -815,9 +821,12 @@ fn ingestion_eligibility( } if let Some(tumbling_window_width_ms) = chain_pane_width_ms { // A chain's panes are built once each, on the grid anchored at the - // epoch (the plan carries no pane origin): every reader of known - // phase must evaluate on its boundaries. A reader of unknown phase - // is taken to run on them (`builds_one_pane_per_evaluation`). + // epoch (the plan carries no pane origin): every reader must + // evaluate on its boundaries. A reader at a fixed interval needs an + // interval the width divides, whatever its phase; a reader of known + // phase also needs that phase on the grid. A reader of unknown phase + // is taken to run at multiples of its interval + // (`builds_one_pane_per_evaluation`). let epoch_panes = PaneLayout { tumbling_window_width_ms, pane_origin_ms: Some(0), @@ -826,6 +835,12 @@ fn ingestion_eligibility( let QueryRecurrence::Repeated(demand) = &d.recurrence else { continue; }; + if d.recurrence + .fixed_interval_ms() + .is_some_and(|interval_ms| !interval_ms.is_multiple_of(tumbling_window_width_ms)) + { + return Err(Why::PanesOffInterval); + } let reader = plan_pane_phase(demand, tumbling_window_width_ms) .map_err(|_| Why::PanesOffPhase)?; let Some(phase_ms) = reader.pane_origin_ms else { @@ -1852,6 +1867,56 @@ mod tests { ); } + /// A chain of 1-min tumbling windows, anchored at the epoch, read every + /// 90 s ends a window at only every other evaluation (#1173), with or + /// without a declared phase on the grid: it is maintained at ingestion + /// time only when every reader's interval is a multiple of its width. + #[test] + fn a_tumbling_window_chain_is_maintained_only_on_its_readers_intervals() { + use IngestionPruneReason as Why; + let root = in_panes(summary_root("quantile_over_time(0.99, m[5m])")); + let every = |interval_ms: u64| QueryWorkloadEntry { + recurrence: QueryRecurrence::Repeated(RepeatedDemand::FixedInterval( + RepetitionInterval(interval_ms as u32), + )), + ..every_minute() + }; + let every_at_phase_0 = |interval_ms: u64| QueryWorkloadEntry { + recurrence: QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt { + interval: RepetitionInterval(interval_ms as u32), + evaluation_phase: TimestampMs(0), + }), + ..every_minute() + }; + let at_ingestion = |demands: [QueryWorkloadEntry; 2]| { + let mut cache = HashMap::new(); + // One chain read by both queries. + let implemented = map_roots(&[root.clone(), root.clone()], &mut |n| { + implement(n, &mut cache) + }) + .unwrap(); + materialization_candidates(&implemented, &demands, &arriving(), false) + .into_iter() + .find(|c| { + matches!( + c.node.asap(), + Some(ASAPOp::SummaryTumblingWindowChain { .. }) + ) + }) + .expect("the chain is a materialization candidate") + .at_ingestion + }; + assert_eq!(at_ingestion([every_minute(), every(2 * MINUTE)]), Ok(())); + assert_eq!( + at_ingestion([every_minute(), every(90_000)]), + Err(Why::PanesOffInterval) + ); + assert_eq!( + at_ingestion([every_minute(), every_at_phase_0(90_000)]), + Err(Why::PanesOffInterval) + ); + } + /// S4: maintaining state at ingestion time needs arriving data and /// queries that repeat and are predictable; otherwise the chain only /// runs at query time, and the materialization candidate says why. diff --git a/crates/planner/tests/runtime_install.rs b/crates/planner/tests/runtime_install.rs index 79f04426b..ec986a063 100644 --- a/crates/planner/tests/runtime_install.rs +++ b/crates/planner/tests/runtime_install.rs @@ -853,6 +853,101 @@ async fn assert_answers_as_reference_without_raw_data( } } +/// The same exact query twice, one copy every 10 s and one every 15 s +/// (hash-consed into one memo group), on a deployment that keeps no raw +/// data. A tumbling-window chain is built at ingestion time once per +/// tumbling window, on the grid of its width anchored at the epoch, so a +/// chain either copy reads must have a width dividing that copy's interval: +/// a 15 s chain read every 10 s ends up to 10 s before the evaluation +/// (#1173). Every evaluation of both copies over one minute answers as the +/// query-time reference. +#[tokio::test] +async fn queries_at_two_intervals_read_chains_on_their_own_grids() { + for query in [ + "sum by (job) (rate(m[1m]))", + "sum by (job) (sum_over_time(m[1m]))", + ] { + assert_copies_at_intervals_answer_as_reference(query, &[10_000, 15_000]).await; + } +} + +/// Copies of `query`, copy `q` every `intervals_ms[q]`, over 6 ingesting +/// series of `m` sampled every second, on a deployment that keeps no raw +/// data. Run by a runtime that refreshes each ingestion tumbling window +/// once, on the grid of its width anchored at the epoch, every evaluation +/// of every copy over one minute answers as the query-time reference. +async fn assert_copies_at_intervals_answer_as_reference(query: &str, intervals_ms: &[i64]) { + let entries: Vec = intervals_ms + .iter() + .map(|interval_ms| { + format!( + r#"{{"query": {query:?}, + "demand": {{"fixed_interval": {interval_ms}}}, + "requirements": {{}}, + "predictability": {{"predictable": {{"known_at": null}}}}}}"# + ) + }) + .collect(); + let workload: PlanningWorkload = serde_json::from_str(&format!( + r#"{{"query_workload": {{"language": "promql", "repeating_queries": [{}]}}, + "data_workload": {{"arrival": {{"ingesting": {{"rows_per_second": 6.0}}}}, + "sources": {{"m": {{"series_count": 6, "sample_interval": 1000}}}}}}}}"#, + entries.join(",") + )) + .unwrap(); + // A deployment that keeps no raw data samples: the windows are views. + let capabilities = DeploymentCapabilities { + raw_data_retained: false, + ..DeploymentCapabilities::default() + }; + let plan_json = plan_json_for(workload, &capabilities).await; + // Samples of `m`, `job` a (instances 1-3) and b (4-6), every second + // from 2 minutes before `EVALUATION_MS` to past the last evaluation; + // each value distinct, so a window off by a second answers differently. + let samples: Vec = (1..=6i64) + .flat_map(|instance| { + (0..200i64).map(move |step| RawDataSample { + labels: BTreeMap::from([ + ("__name__".to_string(), "m".to_string()), + ( + "job".to_string(), + if instance <= 3 { "a" } else { "b" }.to_string(), + ), + ("instance".to_string(), instance.to_string()), + ]), + timestamp_ms: EVALUATION_MS - 120_000 + step * 1_000, + value: (instance * 1000 + step * step % 997) as f64, + }) + }) + .collect(); + // Every evaluation of every copy over one minute, in time order. + let mut evaluations: Vec<(usize, i64)> = intervals_ms + .iter() + .enumerate() + .flat_map(|(query, &interval_ms)| { + (0..=60_000 / interval_ms).map(move |k| (query, EVALUATION_MS + k * interval_ms)) + }) + .collect(); + evaluations.sort_by_key(|&(query, evaluation_ms)| (evaluation_ms, query)); + // The states stored across every evaluation. + let mut store = SummaryStore::new(); + for (copy, evaluation_ms) in evaluations { + let run = run_data_plane_on( + &plan_json, + &samples, + evaluation_ms, + &mut store, + IngestionBuilds::Tumbling, + ); + assert_eq!(run.answers_per_query[copy].len(), 2, "one answer per job"); + assert_eq!( + run.answers_per_query[copy], run.reference_answers_per_query[copy], + "{query}, copy {copy} (every {} ms) at {evaluation_ms}", + intervals_ms[copy] + ); + } +} + /// A plan file of another DAG wire version is refused, and so is the /// reference of a tampered plan file, with an error rather than a panic. #[tokio::test] From fb47a129cde41d088cee9f2b31b2bee309d58813 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Mon, 2 Nov 2026 22:09:00 +0000 Subject: [PATCH 2/2] docs(stage2): cite the whole-window rule for a reader of unknown phase Review: builds_one_pane_per_evaluation decides keeping panes across evaluations, not the phase-0 assumption. --- crates/physical-asap-optimization/src/lib.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/physical-asap-optimization/src/lib.rs b/crates/physical-asap-optimization/src/lib.rs index 5d47d5cf7..01c25bf83 100644 --- a/crates/physical-asap-optimization/src/lib.rs +++ b/crates/physical-asap-optimization/src/lib.rs @@ -825,8 +825,8 @@ fn ingestion_eligibility( // evaluate on its boundaries. A reader at a fixed interval needs an // interval the width divides, whatever its phase; a reader of known // phase also needs that phase on the grid. A reader of unknown phase - // is taken to run at multiples of its interval - // (`builds_one_pane_per_evaluation`). + // is taken to run at multiples of its interval (phase 0), as for a + // whole-window build below. let epoch_panes = PaneLayout { tumbling_window_width_ms, pane_origin_ms: Some(0),