diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index b57f00e8..9123c9ff 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -636,16 +636,13 @@ impl BackendLocalPlanningSnapshot { .get(&query_string) .cloned() .unwrap_or_else(|| { - vec![WindowImplementationCandidate { - implementation_id: self.implementation.window_implementation_id.clone(), - framework: SummaryWindowFramework::Tumbling, - window_secs: lookback_ms / 1_000, - slide_secs: lookback_ms / 1_000, - layout: asap_types::WindowMaterializationLayout::Pane { - pane_secs: lookback_ms / 1_000, - }, + derived_window_candidates( + &self.implementation.window_implementation_id, + canonical_roots.last().expect("root pushed above"), + lookback_ms, + evaluation_interval_ms, cost, - }] + ) }), runtime_policy: RuntimeRulePolicy::default(), }); @@ -2188,6 +2185,138 @@ fn validate_lifecycle_input( Ok(()) } +/// Every distinct range-selector window in `expr`, as seconds. +/// +/// A single query can carry several. `sum(sum_over_time(a[1m])) / sum(sum_over_time(b[5m]))` +/// has two, and each one becomes its own materialization with its own window. +/// `time_selection.lookback` is the workload's declared range and is not +/// required to equal any of them. +fn range_selector_windows_secs(expr: &QueryExpr) -> BTreeSet { + fn visit(expr: &QueryExpr, windows: &mut BTreeSet) { + if let QueryExpr::TimeRange { range, .. } = expr { + let secs = range.as_secs(); + if secs != 0 { + windows.insert(secs); + } + } + match expr { + QueryExpr::PromqlScalarBridge(child) + | QueryExpr::PromqlVectorFromScalar(child) + | QueryExpr::PromqlScalarFromVector(child) + | QueryExpr::PromqlRelabel { child, .. } + | QueryExpr::PromqlSeriesSample { child, .. } + | QueryExpr::Filter { child, .. } + | QueryExpr::Project { child, .. } + | QueryExpr::Aggregate { child, .. } + | QueryExpr::Dedup { child, .. } + | QueryExpr::Sort { child, .. } + | QueryExpr::Limit { child, .. } + | QueryExpr::PromqlSubquery { child, .. } + | QueryExpr::TimeRange { child, .. } + | QueryExpr::TimeShift { child, .. } => visit(child, windows), + QueryExpr::BinaryOp { + lhs: left, + rhs: right, + .. + } + | QueryExpr::Join { left, right, .. } + | QueryExpr::SetOp { left, right, .. } => { + visit(left, windows); + visit(right, windows); + } + _ => {} + } + } + let mut windows = BTreeSet::new(); + visit(expr, &mut windows); + windows +} + +/// The window implementations to plan with when the snapshot priced none for +/// this query. +/// +/// These are *shapes*, not cost quotes. `ImplementationCostEvidence` is +/// measured evidence: its `weighted_cost` doc puts pricing update CPU, +/// query-time merges, retained memory, storage, scans and network on the +/// evidence producer. So this never synthesizes competing candidates for one +/// window to rank against each other — one shape per window, selected by a +/// `min_by` over a one-element list where the cost cannot change the outcome. +/// Ranking `Pane` against `FullWindow` requires a snapshot supplying both with +/// their own priced evidence in `window_candidates`. +/// +/// **One candidate per range-selector window, not one per query.** A state +/// whose window has no candidate is dropped from selection outright +/// (`hybrid_execution`'s filter), and the surviving candidates are narrowed to +/// the selected window before validation. Deriving a single candidate from +/// `time_selection.lookback` therefore silently costs every operand whose own +/// range differs from it its summary: `sum(sum_over_time(a[1m])) / +/// sum(sum_over_time(b[5m]))` under a 1m lookback kept `a` and dropped `b` to +/// exact execution, with nothing reported. +/// +/// Within one window, the shape follows the query's evaluation cadence. A +/// workload evaluated every 30s over a 5m window needs its state to advance +/// every 30s; one 5m tumbling window answers with results that only change +/// once every five minutes. `evaluation_interval_ms` already reaches this +/// function — it was read for lifecycle costing and then dropped here. +/// +/// The guards are the validator's own rules, so a bad shape is a compile error +/// rather than a silent plan: `WindowMaterializationLayout::validate` requires +/// the pane to divide both window and slide (45s into 300s has no such pane), +/// and the framework/layout table admits `Tumbling + Pane` and `Sliding + Pane`. +/// `pane_secs == slide_secs` is the coarsest legal pane for a cadence, so it +/// is the one with the fewest query-time merges. `FullWindow` is the other +/// legal `Sliding` layout and is deliberately not emitted alongside it: +/// preferring it is a write-amplification-versus-read-amplification tradeoff, +/// which is a cost comparison, and there is no second quote to compare. +fn derived_window_candidates( + implementation_id: &str, + expr: &QueryExpr, + lookback_ms: u64, + evaluation_interval_ms: u32, + cost: ImplementationCostEvidence, +) -> Vec { + let mut windows = range_selector_windows_secs(expr); + if windows.is_empty() { + windows.insert(lookback_ms / 1_000); + } + // `window_implementation_id` reaches lifecycle estimates and cost + // manifests, so one label must not describe several shapes. A query with a + // single window keeps the snapshot's identity untouched. + let distinct = windows.len() > 1; + windows + .into_iter() + .map(|window_secs| { + let evaluation_secs = u64::from(evaluation_interval_ms) / 1_000; + let advances_within_window = evaluation_secs != 0 + && evaluation_secs < window_secs + && window_secs.is_multiple_of(evaluation_secs); + let slide_secs = if advances_within_window { + evaluation_secs + } else { + window_secs + }; + WindowImplementationCandidate { + implementation_id: if distinct { + format!("{implementation_id}-{window_secs}s") + } else { + implementation_id.to_string() + }, + framework: if advances_within_window { + SummaryWindowFramework::Sliding + } else { + SummaryWindowFramework::Tumbling + }, + window_secs, + slide_secs, + layout: asap_types::WindowMaterializationLayout::Pane { + pane_secs: slide_secs, + }, + cost: cost.clone(), + } + }) + .collect() +} + pub(super) fn validate_window_implementations( query: &PlanningQuery, environment: &DeploymentEnvironment, @@ -4861,15 +4990,9 @@ mod tests { let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0]; entry.query = Query("sum(sum_over_time(a[1m])) / sum(sum_over_time(b[5m]))".into()); entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact); - let (mut request, env) = snapshot.planning_request().unwrap(); - let query = &mut request.queries[0]; - let mut five_minutes = query.window_implementations[0].clone(); - five_minutes.implementation_id = "five-minute-evidence".into(); - five_minutes.window_secs = 300; - five_minutes.framework = SummaryWindowFramework::Sliding; - five_minutes.slide_secs = 60; - five_minutes.layout = asap_types::WindowMaterializationLayout::Pane { pane_secs: 60 }; - query.window_implementations.push(five_minutes); + // The derivation now covers both range selectors, so this no longer + // needs a hand-supplied 5m candidate to keep `b` from falling back. + let (request, env) = snapshot.planning_request().unwrap(); let plan = PhysicalCompiler.compile(request, env).unwrap(); let bindings = plan .query_plan @@ -4890,16 +5013,199 @@ mod tests { ) }) .collect::>(); + // `window_ms` is the stored pane width, `readout_lookback_ms` the + // semantic range. Each operand keeps its own range -- 1m for `a`, 5m + // for `b` -- while both store 10s panes, because the snapshot + // evaluates every 10s and the derivation now covers both selectors. assert_eq!( actual, - BTreeSet::from([("a", 60_000, Some(60_000)), ("b", 60_000, Some(300_000)),]) + BTreeSet::from([("a", 10_000, Some(60_000)), ("b", 10_000, Some(300_000)),]) ); assert_eq!(plan.precompute_plan.materializations.len(), 2); } - // A filtered denominator is a typed residual while its summary sibling remains installed. + fn planning_snapshot() -> BackendLocalPlanningSnapshot { + serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap() + } + + // A workload evaluated more often than its window is wide must advance its + // state at that cadence. Planning it as one lookback-wide tumbling window + // answers with results that only change once per window. + #[test] + fn derived_window_candidate_follows_the_evaluation_cadence() { + let cost = planning_snapshot().implementation.implementation_cost; + let expr = crate::query_parser::parse_query_expr_canonical( + "quantile_over_time(0.5, data[5m])", + AccuracyTarget::Exact, + ) + .unwrap(); + let derived = derived_window_candidates("id", &expr, 300_000, 30_000, cost); + assert_eq!(derived.len(), 1); + let candidate = &derived[0]; + assert_eq!(candidate.framework, SummaryWindowFramework::Sliding); + assert_eq!((candidate.window_secs, candidate.slide_secs), (300, 30)); + assert_eq!( + candidate.layout, + asap_types::WindowMaterializationLayout::Pane { pane_secs: 30 } + ); + } + + // Every shape this function can emit must survive the validator, or a bad + // derivation would reach a plan instead of a compile error. #[test] - fn composable_binary_retains_summary_sibling_of_prometheus_filtered_subtree() { + fn derived_window_candidate_shapes_are_accepted_by_validation() { + let snapshot = planning_snapshot(); + let (request, environment) = snapshot.planning_request().unwrap(); + let cost = planning_snapshot().implementation.implementation_cost; + for (lookback_ms, evaluation_ms) in [ + (300_000, 30_000), + (300_000, 300_000), + (300_000, 45_000), + (60_000, 90_000), + ] { + let mut query = request.queries[0].clone(); + query.window_secs = lookback_ms / 1_000; + let expr = crate::query_parser::parse_query_expr_canonical( + &format!("quantile_over_time(0.5, data[{}s])", lookback_ms / 1_000), + AccuracyTarget::Exact, + ) + .unwrap(); + query.window_implementations = derived_window_candidates( + "derived", + &expr, + lookback_ms, + evaluation_ms, + cost.clone(), + ); + validate_window_implementations(&query, &environment).unwrap_or_else(|error| { + panic!("lookback {lookback_ms} cadence {evaluation_ms}: {error:?}") + }); + } + } + + // A cadence that cannot divide the window has no pane width dividing both, + // and one at or above the window has nothing to slide within. Both keep the + // previous tumbling shape rather than emitting something unschedulable. + #[test] + fn derived_window_candidate_stays_tumbling_without_a_dividing_cadence() { + let cost = planning_snapshot().implementation.implementation_cost; + let expr = crate::query_parser::parse_query_expr_canonical( + "quantile_over_time(0.5, data[5m])", + AccuracyTarget::Exact, + ) + .unwrap(); + for evaluation_ms in [300_000, 450_000, 45_000, 0] { + let derived = + derived_window_candidates("id", &expr, 300_000, evaluation_ms, cost.clone()); + let candidate = &derived[0]; + assert_eq!( + ( + candidate.framework.clone(), + candidate.slide_secs, + candidate.layout.clone() + ), + ( + SummaryWindowFramework::Tumbling, + 300, + asap_types::WindowMaterializationLayout::Pane { pane_secs: 300 } + ), + "cadence {evaluation_ms}" + ); + } + } + + // Priced evidence is the evidence producer's to supply. A snapshot that + // carries its own candidates keeps them verbatim. + #[test] + fn supplied_window_candidates_are_not_replaced_by_the_derivation() { + let mut snapshot = planning_snapshot(); + let query_string = snapshot.query_workload.repeating_queries.as_ref().unwrap()[0] + .query + .0 + .clone(); + let expr = crate::query_parser::parse_query_expr_canonical( + "quantile_over_time(0.99, m[1m])", + AccuracyTarget::Exact, + ) + .unwrap(); + let mut supplied = derived_window_candidates( + "supplied", + &expr, + 60_000, + 60_000, + snapshot.implementation.implementation_cost.clone(), + ) + .remove(0); + supplied.framework = SummaryWindowFramework::Sliding; + supplied.slide_secs = 20; + supplied.layout = asap_types::WindowMaterializationLayout::Pane { pane_secs: 20 }; + snapshot + .implementation + .window_candidates + .insert(query_string, vec![supplied.clone()]); + let (request, _) = snapshot.planning_request().unwrap(); + assert_eq!(request.queries[0].window_implementations, vec![supplied]); + } + + // End to end: the retained-state count is derived from the pane width, so + // fixing the shape fixes it too. Six 10s panes cover the 1m lookback, plus + // the one still being filled. + #[test] + fn retained_state_count_follows_the_derived_pane_width() { + let snapshot = planning_snapshot(); + let (request, environment) = snapshot.planning_request().unwrap(); + let plan = PhysicalCompiler.compile(request, environment).unwrap(); + assert_eq!( + plan.precompute_plan.materializations[0].num_aggregates_to_retain, + Some(7) + ); + } + + // The reported case: two 5m-lookback quantiles evaluated every 30s. The + // whole chain has to land — sliding framework, 30s panes, and the retained + // count that falls out of the pane width — or the answer only changes once + // every five minutes. + #[test] + fn five_minute_lookback_evaluated_every_thirty_seconds_slides_by_thirty() { + let mut snapshot = planning_snapshot(); + { + let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0]; + entry.query = Query("quantile_over_time(0.5, data[5m])".into()); + entry.time_selection.lookback = Some(DurationMs(300_000)); + entry.demand = RepeatedDemand::FixedIntervalAt { + interval: RepetitionInterval(30_000), + evaluation_phase: planner_types::workload::TimestampMs(0), + }; + } + let (request, environment) = snapshot.planning_request().unwrap(); + let plan = PhysicalCompiler.compile(request, environment).unwrap(); + let materialization = &plan.precompute_plan.materializations[0]; + assert_eq!( + ( + materialization.window_size, + materialization.slide_interval, + materialization.window_type, + materialization.window_layout.clone(), + materialization.num_aggregates_to_retain, + ), + ( + 300, + 30, + asap_types::WindowKind::Sliding, + asap_types::WindowMaterializationLayout::Pane { pane_secs: 30 }, + // Ten 30s panes cover the 5m lookback, plus the one still filling. + Some(11), + ) + ); + } + + // A filtered operand gets a summary over its own filtered population, + // with each operand keeping its own range. + #[test] + fn composable_binary_summarizes_each_prometheus_filtered_operand() { use crate::query_plan::{logical::LogicalOperator, QueryPlanNode}; let mut snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!( "../../../docs/examples/asapquery-planning-snapshot.json" @@ -4913,20 +5219,35 @@ mod tests { let plan = PhysicalCompiler.compile(request, env).unwrap(); let query = plan.query_plan.entries.values().next().unwrap(); let bindings = query.materialization_bindings(); - assert_eq!(bindings.len(), 1); - let identity = &plan.summary_catalog.materializations[&bindings[0].materialization]; - let data = &plan.summary_catalog.data_descriptors[&identity.data_descriptor_id]; + // Both operands now hold a summary. The filtered denominator is no + // longer a typed residual: its 5m range has a candidate, so it gets + // its own summary over the filtered population rather than exact + // execution. Nothing about the filter forced the residual -- the + // missing 5m window candidate did, and this test previously pinned + // that artifact as intended behavior. + let bound = bindings + .iter() + .map(|binding| { + let identity = &plan.summary_catalog.materializations[&binding.materialization]; + let data = &plan.summary_catalog.data_descriptors[&identity.data_descriptor_id]; + ( + data.time_series_metric().unwrap(), + data.population_filter_canonical.clone(), + binding.readout_lookback_ms, + ) + }) + .collect::>(); assert_eq!( - (data.time_series_metric().unwrap(), bindings[0].window_ms), - ("a", 60_000) + bound, + BTreeSet::from([ + ("a", String::new(), Some(60_000)), + ("b", "{job!=\"x\"}".to_string(), Some(300_000)), + ]) ); assert!(!query .nodes .values() .any(|node| matches!(node, QueryPlanNode::ExactFallback { .. }))); - assert!(query.nodes.values().any(|node| matches!(node, - QueryPlanNode::Logical { operator: LogicalOperator::ExactSubquery { query }, .. } - if query == "sum_over_time(b{job!=\"x\"}[5m])" || query == "sum(sum_over_time(b{job!=\"x\"}[5m]))"))); assert!(!query.nodes.values().any(|node| matches!( node, QueryPlanNode::Logical { @@ -4934,7 +5255,7 @@ mod tests { .. } ))); - assert_eq!(plan.precompute_plan.materializations.len(), 1); + assert_eq!(plan.precompute_plan.materializations.len(), 2); } #[test] diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs index ecf98038..0a82532a 100644 --- a/control_plane/src/physical/workload_cost.rs +++ b/control_plane/src/physical/workload_cost.rs @@ -991,7 +991,10 @@ mod tests { .count(), 1 ); - assert!(manifest + // Both ranges are summarized now, so nothing in this workload is + // priced as exact backend execution. The 5m operand used to land there + // only because its window had no implementation candidate. + assert!(!manifest .components .values() .any(|demand| demand.implementation.get("location")