fix(stage2): a window equal to its interval is kept at ingestion time; whole-window builds end on the epoch grid

crates/physical-asap-optimization/src/lib.rs

@@ -52,10 +52,15 @@
 //!   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
-//!   all run at one fixed interval with a known phase, over a window no
-//!   longer than that interval and not shifted, and none is a range
-//!   evaluation of more than one step per request (one window per request
-//!   cannot serve every step);
+//!   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
+//!   are multiples of the window; a reader with no declared phase is taken
+//!   to evaluate at multiples of its interval, as a chain's is), and none
+//!   is a range evaluation of more than one step per request (one window
+//!   per request cannot serve every step).
+//!   So a window equal to its interval (`m[5m]` every 5 minutes) is one
+//!   tumbling window maintained at ingestion time;
 //! * everything below it can run at ingestion time.
 //!
 //! ## Range evaluations
@@ -200,9 +205,14 @@ pub(crate) enum IngestionPruneReason {
     /// continuously ingesting time-series (PromQL) sources are.
     CounterNeedsTimeOrder,
     /// A whole-window build whose readers do not all run at one fixed
-    /// interval with a known phase, or whose window is longer than that
-    /// interval, shifted, or not a range.
+    /// interval and one phase, or whose window is shifted or not a range.
     WindowNotFixed,
+    /// A whole-window build whose evaluations do not all end on the grid of
+    /// its window's width anchored at the epoch, where the backend builds
+    /// it (e.g. `m[20s]` every 20 s at phase 5 s, every 30 s, or `m[15m]`
+    /// every 5 minutes, which panes serve): the interval or the declared
+    /// phase is not a multiple of the window.
+    WindowOffEpochGrid,
     /// A whole-window build read by a range evaluation of more than one
     /// step per request: it keeps one window per request, not one per step.
     RangeEvaluatedWholeWindow,
@@ -781,36 +791,63 @@ fn ingestion_eligibility(
     }) {
         return Err(Why::RangeEvaluatedWholeWindow);
     }
-    // A whole-window build: one window per evaluation, at one known phase,
-    // that ends at the evaluation and fits in the interval.
-    let mut cadence = None;
+    // A whole-window build: one tumbling window per evaluation, built (like
+    // a chain's panes) on the grid of its width anchored at the epoch, so
+    // every evaluation must end on that grid. The readers share one fixed
+    // interval and their declared phases agree; a reader with no declared
+    // phase is taken to evaluate at multiples of its interval (phase 0), as
+    // a chain's is, which lies on the grid once the interval does.
+    let mut interval_ms = None;
+    let mut phase_ms = None;
     for d in demands.into_iter().flatten() {
-        let QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt {
-            interval,
-            evaluation_phase,
-        }) = &d.recurrence
-        else {
-            return Err(Why::WindowNotFixed);
+        let (interval, phase) = match &d.recurrence {
+            QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(interval)) => {
+                (u64::from(interval.0), None)
+            }
+            QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt {
+                interval,
+                evaluation_phase,
+            }) => (
+                u64::from(interval.0),
+                evaluation_phase.0.checked_rem(u64::from(interval.0)),
+            ),
+            _ => return Err(Why::WindowNotFixed),
         };
-        if cadence.is_some_and(|c| c != (interval.0, evaluation_phase.0)) {
+        if interval == 0 {
             return Err(Why::WindowNotFixed);
         }
-        cadence = Some((interval.0, evaluation_phase.0));
+        if *interval_ms.get_or_insert(interval) != interval {
+            return Err(Why::WindowNotFixed);
+        }
+        if let Some(phase) = phase {
+            if *phase_ms.get_or_insert(phase) != phase {
+                return Err(Why::WindowNotFixed);
+            }
+        }
     }
-    let Some((interval_ms, _)) = cadence else {
+    let Some(interval_ms) = interval_ms else {
         return Err(Why::WindowNotFixed);
     };
-    match node.children()[0].non_asap() {
-        Some(NonASAPOp::TimeRange {
-            range,
-            kind: asap_types::ir::TimeRangeKind::Range,
-            child,
-        }) if range.as_millis() <= u128::from(interval_ms)
-            && !matches!(child.non_asap(), Some(NonASAPOp::TimeShift { .. })) =>
-        {
-            Ok(())
-        }
-        _ => Err(Why::WindowNotFixed),
+    let Some(NonASAPOp::TimeRange {
+        range,
+        kind: asap_types::ir::TimeRangeKind::Range,
+        child,
+    }) = node.children()[0].non_asap()
+    else {
+        return Err(Why::WindowNotFixed);
+    };
+    if matches!(child.non_asap(), Some(NonASAPOp::TimeShift { .. })) {
+        return Err(Why::WindowNotFixed);
+    }
+    let window_ms = u64::try_from(range.as_millis())
+        .ok()
+        .filter(|&ms| ms > 0)
+        .ok_or(Why::WindowNotFixed)?;
+    let on_grid = |ms: u64| ms.is_multiple_of(window_ms);
+    if on_grid(interval_ms) && phase_ms.is_none_or(on_grid) {
+        Ok(())
+    } else {
+        Err(Why::WindowOffEpochGrid)
     }
 }
 
@@ -1668,19 +1705,25 @@ mod tests {
     }
 
     /// A whole-window build is maintained only for one window per
-    /// evaluation at a known phase: a fixed interval with a phase, a window
-    /// no longer than it, and no filter.
+    /// evaluation on the epoch grid of its width, where the backend builds
+    /// it: a fixed interval that is a multiple of the window, a phase (if
+    /// known) on that grid, and no filter. `m[1m]` every minute with no
+    /// declared phase is one tumbling 1-minute window per evaluation, taken
+    /// to run on the grid like a chain's reader of unknown phase; at phase
+    /// 30 s its windows would end off the grid, and every 30 s or every
+    /// 90 s some evaluations would.
     #[test]
-    fn a_whole_window_build_needs_a_known_phase_and_a_window_within_the_interval() {
+    fn a_whole_window_build_needs_evaluations_on_the_epoch_grid_of_its_window() {
         use IngestionPruneReason as Why;
         let root = summary_root("quantile_over_time(0.99, m[1m])");
-        let at = |interval_ms: u32| QueryWorkloadEntry {
+        let at_phase = |interval_ms: u32, phase_ms: u64| QueryWorkloadEntry {
             recurrence: QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt {
                 interval: RepetitionInterval(interval_ms),
-                evaluation_phase: TimestampMs(0),
+                evaluation_phase: TimestampMs(phase_ms),
             }),
             ..every_minute()
         };
+        let at = |interval_ms: u32| at_phase(interval_ms, 0);
         let mat_candidates = |root: &QueryRoot, demand: QueryWorkloadEntry| {
             let mut memo = HashMap::new();
             let implemented =
@@ -1695,11 +1738,56 @@ mod tests {
             .at_ingestion
         };
         assert_eq!(mat_candidates(&root, at(MINUTE as u32)), Ok(()));
+        assert_eq!(mat_candidates(&root, every_minute()), Ok(()));
+        assert_eq!(mat_candidates(&root, at(2 * MINUTE as u32)), Ok(()));
+        assert_eq!(
+            mat_candidates(&root, at_phase(MINUTE as u32, 3 * MINUTE)),
+            Ok(())
+        );
+        assert_eq!(
+            mat_candidates(&root, at_phase(MINUTE as u32, 30_000)),
+            Err(Why::WindowOffEpochGrid)
+        );
+        assert_eq!(
+            mat_candidates(&root, at(30_000)),
+            Err(Why::WindowOffEpochGrid)
+        );
+        assert_eq!(
+            mat_candidates(&root, at(90_000)),
+            Err(Why::WindowOffEpochGrid)
+        );
+        // Several readers of one build: a declared phase on the grid and an
+        // undeclared one agree; two declared phases, or two intervals, do not.
+        let mat_candidates_of = |demands: &[QueryWorkloadEntry]| {
+            let roots = vec![root.clone(); demands.len()];
+            let mut memo = HashMap::new();
+            let implemented = map_roots(&roots, &mut |n| implement(n, &mut memo)).unwrap();
+            materialization_candidates(
+                &implemented,
+                demands,
+                &arriving(),
+                &DeploymentCapabilities::default(),
+            )
+            .into_iter()
+            .find(|c| matches!(c.node.asap(), Some(ASAPOp::SummaryAgg { .. })))
+            .expect("the whole-window build is a materialization candidate")
+            .at_ingestion
+        };
+        assert_eq!(
+            mat_candidates_of(&[at_phase(MINUTE as u32, MINUTE), every_minute()]),
+            Ok(())
+        );
+        assert_eq!(
+            mat_candidates_of(&[
+                at(MINUTE as u32),
+                at_phase(MINUTE as u32, 2 * MINUTE + MINUTE / 2)
+            ]),
+            Err(Why::WindowNotFixed)
+        );
         assert_eq!(
-            mat_candidates(&root, every_minute()),
+            mat_candidates_of(&[at(MINUTE as u32), at(2 * MINUTE as u32)]),
             Err(Why::WindowNotFixed)
         );
-        assert_eq!(mat_candidates(&root, at(30_000)), Err(Why::WindowNotFixed));
         let candidates = optimize_physical_asap(
             "L1",
             std::slice::from_ref(&root),

crates/planner/tests/runtime_install.rs

@@ -727,6 +727,123 @@ async fn assert_phased_copies_answer_as_reference(phases_ms: &[i64]) {
     }
 }
 
+/// A query whose window equals its interval, with no declared phase: one
+/// tumbling window per evaluation, maintained at ingestion time, answers
+/// every evaluation without raw data. `quantile_over_time(0.99, m[5m])`
+/// every 5 minutes at rank error 0.01 (a KLL per series) had no plan.
+#[tokio::test]
+async fn a_window_equal_to_its_interval_is_kept_at_ingestion_time() {
+    assert_answers_as_reference_without_raw_data(
+        "quantile_over_time(0.99, m[5m])",
+        r#"{"fixed_interval": 300000}"#,
+        RANK_ERROR_1_PERCENT,
+        &[0, 300_000, 600_000],
+    )
+    .await;
+    assert_answers_as_reference_without_raw_data(
+        "sum by (job) (sum_over_time(m[20s]))",
+        r#"{"fixed_interval": 20000}"#,
+        "{}",
+        &[0, 20_000, 40_000, 60_000],
+    )
+    .await;
+}
+
+/// The same 20 s window every 20 s at phase 5 s: a 20 s window on the
+/// epoch grid would end 5 s before every evaluation, so the plan must not
+/// read one; it still answers every evaluation as the reference.
+#[tokio::test]
+async fn a_window_equal_to_its_interval_at_an_offset_phase_answers_as_reference() {
+    assert_answers_as_reference_without_raw_data(
+        "sum by (job) (sum_over_time(m[20s]))",
+        r#"{"fixed_interval_at": {"interval": 20000, "evaluation_phase": 5000}}"#,
+        "{}",
+        &[5_000, 25_000, 45_000, 65_000],
+    )
+    .await;
+}
+
+/// Rank error 0.01 with failure probability 0.01, as `requirements`.
+const RANK_ERROR_1_PERCENT: &str = r#"{"accuracy": {"accuracy_metric": "rank", "max_error": 0.01,
+    "failure_probability": 0.01, "accuracy_scope": "whole_result"}}"#;
+
+/// Plans `query` with `demand` and `requirements` over 6 ingesting series of `m` sampled every
+/// second, for a deployment that keeps no raw data, and runs it at
+/// `EVALUATION_MS` plus each of `evaluation_offsets_ms` on a runtime that
+/// builds each ingestion state once, on the epoch grid of its width. The
+/// plan keeps state at ingestion time and reads no raw data at query time;
+/// every answer equals the query-time reference.
+async fn assert_answers_as_reference_without_raw_data(
+    query: &str,
+    demand: &str,
+    requirements: &str,
+    evaluation_offsets_ms: &[i64],
+) {
+    let workload: PlanningWorkload = serde_json::from_str(&format!(
+        r#"{{"query_workload": {{"language": "promql", "repeating_queries": [
+              {{"query": {query:?}, "demand": {demand}, "requirements": {requirements},
+                "predictability": {{"predictable": {{"known_at": null}}}}}}]}},
+            "data_workload": {{"arrival": {{"ingesting": {{"rows_per_second": 6.0}}}},
+              "sources": {{"m": {{"series_count": 6, "sample_interval": 1000}}}}}}}}"#
+    ))
+    .unwrap();
+    let capabilities = DeploymentCapabilities {
+        raw_data_retained: false,
+        ..DeploymentCapabilities::default()
+    };
+    let plan_json = plan_json_for(workload, &capabilities).await;
+    let plan: InstallablePlan = serde_json::from_str(&plan_json).unwrap();
+    assert!(
+        plan.stored_outputs
+            .iter()
+            .any(|output| output.built_at == StoredOutputBuild::IngestionRun),
+        "{:#?}",
+        plan.stored_outputs
+    );
+    assert!(
+        plan.query_time_raw_windows.is_empty(),
+        "no raw data at query time"
+    );
+    // Samples of `m`, `job` a (instances 1-3) and b (4-6), every second
+    // from 6 minutes before `EVALUATION_MS` to past the last evaluation;
+    // each value distinct, so a window off by a second answers differently.
+    let last_ms = EVALUATION_MS + evaluation_offsets_ms.iter().max().unwrap();
+    let steps = (last_ms - EVALUATION_MS + 360_000) / 1_000 + 10;
+    let samples: Vec<RawDataSample> = (1..=6i64)
+        .flat_map(|instance| {
+            (0..steps).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 - 360_000 + step * 1_000,
+                value: (instance * 1000 + step * step % 997) as f64,
+            })
+        })
+        .collect();
+    // The states stored across every evaluation.
+    let mut store = SummaryStore::new();
+    for offset_ms in evaluation_offsets_ms {
+        let evaluation_ms = EVALUATION_MS + offset_ms;
+        let run = run_data_plane_on(
+            &plan_json,
+            &samples,
+            evaluation_ms,
+            &mut store,
+            IngestionBuilds::Tumbling,
+        );
+        assert!(!run.answers_per_query[0].is_empty());
+        assert_eq!(
+            run.answers_per_query, run.reference_answers_per_query,
+            "{query} at {evaluation_ms}"
+        );
+    }
+}
+
 /// 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]