fix(stage2): keep a tumbling-window chain at ingestion time only on its readers' intervals (#1173)

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 (phase 0), as for a
+        // whole-window build below.
         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.

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<String> = 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<RawDataSample> = (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]