diff --git a/crates/physical-asap-optimization/src/lib.rs b/crates/physical-asap-optimization/src/lib.rs index 0c34f3b12..4f14355e0 100644 --- a/crates/physical-asap-optimization/src/lib.rs +++ b/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), 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..58c9b75c5 100644 --- a/docs/design_docs/proposals/planner-layering.md +++ b/docs/design_docs/proposals/planner-layering.md @@ -850,6 +850,18 @@ 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, 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 evaluate at multiples of its +interval, as for a pane chain. + #### Physical operator implementation Physical operator implementation converts every node to physical operators, for