From ec6dc46416888ed1d7dcbfdef8d86b8cc95dc5ca Mon Sep 17 00:00:00 2001 From: Zeying Zhu <50204836+zzylol@users.noreply.github.com> Date: Thu, 22 Oct 2026 19:20:13 +0000 Subject: [PATCH 1/3] fix(stage2): keep a window equal to its interval at ingestion time; whole-window builds end on the epoch grid A whole-window build (one window per evaluation) may now be maintained at ingestion time for readers declared with a fixed interval and no phase, as a pane chain's readers already are: a reader of unknown phase is taken to run on the epoch grid. The build must end on the grid of its window's width anchored at the epoch, where the runtime builds it: the interval and any known phase are multiples of the window. So m[5m] every 5 minutes (PromQL or SQL) is one tumbling window kept at ingestion time and plans without raw data, and m[20s] every 20 s at phase 5 s no longer reads a window off by 5 s (it uses 5 s panes). Closes #893 Closes #828 --- crates/physical-asap-optimization/src/lib.rs | 124 ++++++++++++------ crates/planner/tests/runtime_install.rs | 117 +++++++++++++++++ crates/planner/tests/sql_table_scans.rs | 76 +++++++++-- .../design_docs/proposals/planner-layering.md | 11 ++ 4 files changed, 283 insertions(+), 45 deletions(-) diff --git a/crates/physical-asap-optimization/src/lib.rs b/crates/physical-asap-optimization/src/lib.rs index 0c34f3b12..34293548b 100644 --- a/crates/physical-asap-optimization/src/lib.rs +++ b/crates/physical-asap-optimization/src/lib.rs @@ -52,10 +52,14 @@ //! 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 and phase, 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 known phase are multiples +//! of the window; a reader of unknown phase is taken to run on the grid, +//! 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 +204,13 @@ 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, or every 30 s): the + /// interval or the known 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 +789,59 @@ 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 known phases agree; a reader of unknown phase is + // taken to run on the grid, as a chain's is. + 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); + } + if *interval_ms.get_or_insert(interval) != interval { return Err(Why::WindowNotFixed); } - cadence = Some((interval.0, evaluation_phase.0)); + 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()).map_err(|_| Why::WindowNotFixed)?; + let on_grid = |ms: u64| window_ms > 0 && 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 +1699,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 +1732,24 @@ 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, every_minute()), - Err(Why::WindowNotFixed) + mat_candidates(&root, at(90_000)), + Err(Why::WindowOffEpochGrid) ); - assert_eq!(mat_candidates(&root, at(30_000)), Err(Why::WindowNotFixed)); let candidates = optimize_physical_asap( "L1", std::slice::from_ref(&root), diff --git a/crates/planner/tests/runtime_install.rs b/crates/planner/tests/runtime_install.rs index 61d9debe8..13e856f23 100644 --- a/crates/planner/tests/runtime_install.rs +++ b/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 = (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] diff --git a/crates/planner/tests/sql_table_scans.rs b/crates/planner/tests/sql_table_scans.rs index c07f95a31..e85538f47 100644 --- a/crates/planner/tests/sql_table_scans.rs +++ b/crates/planner/tests/sql_table_scans.rs @@ -45,15 +45,15 @@ fn http_schema() -> Schema { ) } -/// Repeating SQL `queries`, each every minute with `accuracy`, over `http` -/// ingesting 100 rows per second, whose 3 services and 100,000 rows held -/// are declared. -fn workload(queries: &[&str], accuracy: &str) -> PlanningWorkload { +/// Repeating SQL `queries`, each every `interval_ms` with `accuracy`, over +/// `http` ingesting 100 rows per second, whose 3 services and 100,000 rows +/// held are declared. +fn workload(queries: &[&str], accuracy: &str, interval_ms: u64) -> PlanningWorkload { let entries: Vec = queries .iter() .map(|query| { format!( - r#"{{"query": {query:?}, "demand": {{"fixed_interval": 60000}}, + r#"{{"query": {query:?}, "demand": {{"fixed_interval": {interval_ms}}}, "requirements": {{"accuracy": {accuracy}}}, "predictability": {{"predictable": {{"known_at": null}}}}}}"# ) @@ -83,26 +83,42 @@ async fn plan(query: &str, accuracy: &str) -> InstallablePlan { } async fn plan_all(queries: &[&str], accuracy: &str) -> InstallablePlan { - plan_for(queries, accuracy, &DeploymentCapabilities::default()).await + plan_for( + queries, + accuracy, + MINUTE_MS as u64, + &DeploymentCapabilities::default(), + ) + .await } /// [`plan_all`] for a deployment that keeps no raw data: every window is /// read by state maintained at ingestion time. async fn plan_without_raw_data(queries: &[&str], accuracy: &str) -> InstallablePlan { + plan_without_raw_data_every(queries, accuracy, MINUTE_MS as u64).await +} + +/// [`plan_without_raw_data`] with each query every `interval_ms`. +async fn plan_without_raw_data_every( + queries: &[&str], + accuracy: &str, + interval_ms: u64, +) -> InstallablePlan { let deployment = DeploymentCapabilities { raw_data_retained: false, ..DeploymentCapabilities::default() }; - plan_for(queries, accuracy, &deployment).await + plan_for(queries, accuracy, interval_ms, &deployment).await } async fn plan_for( queries: &[&str], accuracy: &str, + interval_ms: u64, deployment: &DeploymentCapabilities, ) -> InstallablePlan { let catalog = SqlCatalog::new().with_table("http", http_schema()); - let workload = workload(queries, accuracy); + let workload = workload(queries, accuracy, interval_ms); plan_installable(UserInput::new( &workload, FrontendInput::Sql { catalog: &catalog }, @@ -589,3 +605,47 @@ async fn ingestion_time_row_counts_read_each_table_row_once() { assert_eq!(by_key(answers[query].clone()), expected, "query {query}"); } } + +/// A 5-minute distinct count every 5 minutes (window equal to interval, +/// the W-RollupHeavy query that had no plan) and an exact 5-minute row +/// count every 5 minutes, for a deployment that keeps no raw data: each is +/// one tumbling window built at ingestion time from the table rows, read +/// no raw data at query time, and answers exactly. +#[tokio::test] +async fn a_window_equal_to_its_interval_is_built_at_ingestion_time_from_table_rows() { + let five_minutes_ms = 5 * MINUTE_MS as u64; + let plan = plan_without_raw_data_every( + &[DISTINCT_USERS_LAST_5_MINUTES], + APPROXIMATE, + five_minutes_ms, + ) + .await; + assert!( + plan.stored_outputs + .iter() + .any(|o| o.built_at == StoredOutputBuild::IngestionRun + && !o.inputs.raw_data_sample_windows.is_empty()), + "built at ingestion time from table rows: {:?}", + plan.stored_outputs + ); + assert!( + plan.query_time_raw_windows.is_empty(), + "no raw data at query time" + ); + assert_eq!( + by_key(run_data_plane(&plan).remove(0)), + expected_distinct_users() + ); + + let row_count = "SELECT service, COUNT(*) AS value FROM http \ + WHERE ts >= now() - INTERVAL '5 minutes' GROUP BY service"; + let plan = plan_without_raw_data_every(&[row_count], EXACT, five_minutes_ms).await; + assert!( + plan.query_time_raw_windows.is_empty(), + "no raw data at query time" + ); + let expected: BTreeMap = (0..3) + .map(|s| (format!("s{s}"), shown(&Value::Int64(100)))) + .collect(); + assert_eq!(by_key(run_data_plane(&plan).remove(0)), expected); +} diff --git a/docs/design_docs/proposals/planner-layering.md b/docs/design_docs/proposals/planner-layering.md index 505937623..03dc4355e 100644 --- a/docs/design_docs/proposals/planner-layering.md +++ b/docs/design_docs/proposals/planner-layering.md @@ -850,6 +850,17 @@ tumbling-pane chain that builds exactly one new pane per evaluation also gets `MatAtQueryKeptAcrossEvaluations(Memory)`: the panes built at query time are kept for the next evaluations. Nothing is placed on disk yet. +A runtime builds every ingestion-time state on the grid of its width anchored +at the epoch, so an ingestion-time build must end on that grid at every +evaluation of its readers. For a pane chain, the pane width divides each +known evaluation phase. For a whole-window build (one window per evaluation, +no pane chain), the readers share one fixed interval and phase, and the +interval and the phase are multiples of the window: `m[5m]` every 5 minutes +is one tumbling 5-minute window maintained at ingestion time and read at each +evaluation, as a one-pane chain would be, while `m[20s]` every 20 s at phase +5 s, or every 30 s, is not kept whole (Pass 2 offers 5 s or 10 s panes). A +reader whose phase is not declared is taken to run on the grid. + #### Physical operator implementation Physical operator implementation converts every node to physical operators, for From 940ad49c8f428fdb8194aef7ea27530c8d7ae96e Mon Sep 17 00:00:00 2001 From: Zeying Zhu <50204836+zzylol@users.noreply.github.com> Date: Thu, 22 Oct 2026 19:33:49 +0000 Subject: [PATCH 2/3] review: multi-reader whole-window cases, zero-width window is WindowNotFixed, wording on undeclared phases --- crates/physical-asap-optimization/src/lib.rs | 59 +++++++++++++++---- .../design_docs/proposals/planner-layering.md | 7 ++- 2 files changed, 52 insertions(+), 14 deletions(-) diff --git a/crates/physical-asap-optimization/src/lib.rs b/crates/physical-asap-optimization/src/lib.rs index 34293548b..08563c5b6 100644 --- a/crates/physical-asap-optimization/src/lib.rs +++ b/crates/physical-asap-optimization/src/lib.rs @@ -52,11 +52,11 @@ //! 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 and phase, 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 known phase are multiples -//! of the window; a reader of unknown phase is taken to run on the grid, -//! as a chain's is), and none is a range evaluation of more than one +//! 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; @@ -208,8 +208,9 @@ pub(crate) enum IngestionPruneReason { 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, or every 30 s): the - /// interval or the known phase is not a multiple of the window. + /// 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. @@ -792,8 +793,9 @@ fn ingestion_eligibility( // 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 known phases agree; a reader of unknown phase is - // taken to run on the grid, as a chain's is. + // 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() { @@ -836,8 +838,11 @@ fn ingestion_eligibility( if matches!(child.non_asap(), Some(NonASAPOp::TimeShift { .. })) { return Err(Why::WindowNotFixed); } - let window_ms = u64::try_from(range.as_millis()).map_err(|_| Why::WindowNotFixed)?; - let on_grid = |ms: u64| window_ms > 0 && ms.is_multiple_of(window_ms); + 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 { @@ -1750,6 +1755,38 @@ mod tests { 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_of(&[at(MINUTE as u32), at(2 * MINUTE as u32)]), + Err(Why::WindowNotFixed) + ); let candidates = optimize_physical_asap( "L1", std::slice::from_ref(&root), diff --git a/docs/design_docs/proposals/planner-layering.md b/docs/design_docs/proposals/planner-layering.md index 03dc4355e..0de5c469f 100644 --- a/docs/design_docs/proposals/planner-layering.md +++ b/docs/design_docs/proposals/planner-layering.md @@ -854,12 +854,13 @@ A runtime builds every ingestion-time state on the grid of its width anchored at the epoch, so an ingestion-time build must end on that grid at every evaluation of its readers. For a pane chain, the pane width divides each known evaluation phase. For a whole-window build (one window per evaluation, -no pane chain), the readers share one fixed interval and phase, and the -interval and the phase are multiples of the window: `m[5m]` every 5 minutes +no pane chain), the readers share one fixed interval, their declared phases +agree, and the interval and a declared phase are multiples of the window: `m[5m]` every 5 minutes is one tumbling 5-minute window maintained at ingestion time and read at each evaluation, as a one-pane chain would be, while `m[20s]` every 20 s at phase 5 s, or every 30 s, is not kept whole (Pass 2 offers 5 s or 10 s panes). A -reader whose phase is not declared is taken to run on the grid. +reader whose phase is not declared is taken to evaluate at multiples of its +interval, as for a pane chain. #### Physical operator implementation From 6a67c5cefdeeb3a80e180587d4fa349665a9871d Mon Sep 17 00:00:00 2001 From: Zeying Zhu <50204836+zzylol@users.noreply.github.com> Date: Thu, 22 Oct 2026 19:35:31 +0000 Subject: [PATCH 3/3] docs: re-wrap the whole-window eligibility text --- crates/physical-asap-optimization/src/lib.rs | 5 +++-- docs/design_docs/proposals/planner-layering.md | 6 +++--- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/crates/physical-asap-optimization/src/lib.rs b/crates/physical-asap-optimization/src/lib.rs index 08563c5b6..4f14355e0 100644 --- a/crates/physical-asap-optimization/src/lib.rs +++ b/crates/physical-asap-optimization/src/lib.rs @@ -56,8 +56,9 @@ //! 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). +//! 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. diff --git a/docs/design_docs/proposals/planner-layering.md b/docs/design_docs/proposals/planner-layering.md index 0de5c469f..58c9b75c5 100644 --- a/docs/design_docs/proposals/planner-layering.md +++ b/docs/design_docs/proposals/planner-layering.md @@ -855,9 +855,9 @@ at the epoch, so an ingestion-time build must end on that grid at every evaluation of its readers. For a pane chain, the pane width divides each known evaluation phase. For a whole-window build (one window per evaluation, no pane chain), the readers share one fixed interval, their declared phases -agree, and the interval and a declared phase are multiples of the window: `m[5m]` every 5 minutes -is one tumbling 5-minute window maintained at ingestion time and read at each -evaluation, as a one-pane chain would be, while `m[20s]` every 20 s at phase +agree, and the interval and a declared phase are multiples of the window: +`m[5m]` every 5 minutes is one tumbling 5-minute window maintained at +ingestion time and read at each evaluation, as a one-pane chain would be, while `m[20s]` every 20 s at phase 5 s, or every 30 s, is not kept whole (Pass 2 offers 5 s or 10 s panes). A reader whose phase is not declared is taken to evaluate at multiples of its interval, as for a pane chain.