From 2e392998fd1a3252449fbd8d8cc92912cde3495d Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 10:39:03 -0600 Subject: [PATCH 1/3] Unify window candidate planning and preserve evaluation alignment --- control_plane/src/clickhouse.rs | 14 +- control_plane/src/main.rs | 50 +- control_plane/src/physical/compiler.rs | 692 +++++++++++------- .../src/physical/compiler/windows.rs | 363 +++++++++ control_plane/src/physical/pane_reuse.rs | 237 ++++-- control_plane/src/query_plan.rs | 1 + control_plane/src/query_plan/logical.rs | 1 + crates/asap_types/src/aggregation_config.rs | 6 +- crates/asap_types/src/plan_publication.rs | 10 + crates/asap_types/src/query_plan.rs | 42 ++ .../drivers/ingest/prometheus_remote_write.rs | 1 + .../accelerator.rs | 1 + .../asap_clickhouse_query_engine/execution.rs | 15 +- .../asap_query_engine/exact_subqueries.rs | 1 + .../asap_query_engine/live_serve.rs | 1 + .../asap_query_engine/post_asap_readout.rs | 110 ++- .../asap_query_engine/summary_executor.rs | 107 ++- .../asapquery_compatibility_process_e2e.rs | 25 +- data_plane/tests/backend_process_e2e.rs | 8 +- data_plane/tests/support/physical_fixture.rs | 1 + .../control-plane/physical-compiler.md | 8 +- .../control-plane/plan-publication.md | 2 +- .../planning/repeated-dashboard-panes.md | 111 ++- .../planning/shared-window-panes.md | 81 +- ...asapquery-compatibility-demo-snapshot.json | 28 +- .../examples/asapquery-planning-snapshot.json | 28 +- tools/o11y-execution/discover_snapshot.py | 2 +- tools/o11y-execution/test_calibrate.py | 4 +- tools/o11y-execution/update_global_profile.py | 2 +- 29 files changed, 1421 insertions(+), 531 deletions(-) create mode 100644 control_plane/src/physical/compiler/windows.rs diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 4abd29e6c..4aba60b5d 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -259,9 +259,14 @@ pub async fn compile_automatic_clickhouse_workload( let config = materialize_selected_sql(node, family, query) .map_err(crate::query_plan::QueryPlanError::Invalid)?; let binding = MaterializationBinding { + full_window_slide_ms: matches!( + config.window_layout, + asap_types::WindowMaterializationLayout::FullWindow + ) + .then_some(config.slide_interval.saturating_mul(1_000)), materialization: config.policy_fingerprint().into(), output_grouping: PhysicalGrouping::Reduce(config.grouping_labels.names()), - window_ms: config.slide_interval * 1000, + window_ms: config.stored_window_ms(), pane_origin_ms: config.pane_origin_ms, readout_lookback_ms: Some(query.end_ms - query.start_ms), item_labels: config.aggregated_labels.labels.clone(), @@ -599,9 +604,14 @@ fn bind_selected_node( )); } Ok(MaterializationBinding { + full_window_slide_ms: matches!( + selected.window_layout, + asap_types::WindowMaterializationLayout::FullWindow + ) + .then_some(selected.slide_interval.saturating_mul(1_000)), materialization: selected.policy_fingerprint().into(), output_grouping: PhysicalGrouping::Reduce(selected.grouping_labels.names()), - window_ms: selected.slide_interval.saturating_mul(1000), + window_ms: selected.stored_window_ms(), pane_origin_ms: selected.pane_origin_ms, readout_lookback_ms: source_window.map(|seconds| seconds.saturating_mul(1000)), item_labels: selected.aggregated_labels.labels.clone(), diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 08790ef8d..03553f10b 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -522,7 +522,8 @@ struct PhysicalPlanQueryRequest { group_by: Vec, accuracy: types_v2::AccuracyTarget, lifecycle: physical::compiler::LifecyclePlanningInput, - window_implementations: Vec, + window_cost_model: physical::compiler::WindowCostModel, + evaluation_phase_ms: u64, #[serde(default)] runtime_policy: physical::compiler::RuntimeRulePolicy, } @@ -871,6 +872,8 @@ fn compile_physical_plan_request( .as_millis() as u64; let mut queries = Vec::with_capacity(request.queries.len()); let mut canonical_roots = Vec::with_capacity(request.queries.len()); + let mut window_models = Vec::new(); + let mut workload_entries = Vec::new(); for query in request.queries { if query.query_id.trim().is_empty() || query.metric.trim().is_empty() @@ -892,6 +895,29 @@ fn compile_physical_plan_request( Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string().into())), }; canonical_roots.push(std::rc::Rc::new(expr)); + window_models.push(query.window_cost_model); + workload_entries.push(planner_types::workload::RepeatingEntry { + query: planner_types::workload::Query(query.query_string.clone()), + demand: planner_types::workload::RepeatedDemand::FixedIntervalAt { + interval: planner_types::workload::RepetitionInterval( + query.lifecycle.evaluation_interval_ms, + ), + evaluation_phase: planner_types::workload::TimestampMs(query.evaluation_phase_ms), + }, + requirements: planner_types::workload::QueryRequirements { + accuracy: planner_types::workload::AccuracyRequirement::Explicit( + query.accuracy.clone(), + ), + ..Default::default() + }, + predictability: planner_types::workload::Predictability::Predictable { known_at: None }, + time_selection: planner_types::workload::TimeSelection { + lookback: Some(planner_types::workload::DurationMs( + query.window_secs.saturating_mul(1_000), + )), + ..Default::default() + }, + }); queries.push(physical::compiler::PlanningQuery { query_id: query.query_id, query_string: query.query_string, @@ -903,7 +929,7 @@ fn compile_physical_plan_request( group_by: query.group_by, accuracy: query.accuracy, lifecycle: query.lifecycle, - window_implementations: query.window_implementations, + window_implementations: Vec::new(), runtime_policy: query.runtime_policy, }); } @@ -919,10 +945,18 @@ fn compile_physical_plan_request( Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string().into())), }; + for (query, model) in queries.iter_mut().zip(window_models) { + physical::compiler::prepare_window_implementations(query, &model, request.target, 0) + .map_err(|error| (StatusCode::UNPROCESSABLE_ENTITY, error.to_string().into()))?; + } let planning_request = physical::compiler::PlanningRequest { - synthesized_window_queries: Default::default(), logical_selection, - query_workload: None, + query_workload: Some(planner_types::workload::QueryWorkload { + language: planner_types::workload::QueryLanguage::PromQL, + query_batch: None, + repeating_queries: Some(workload_entries), + data_workload: None, + }), queries, hybrid_execution: request.target == physical::compiler::PhysicalDeploymentTarget::BackendLocalRemoteWrite, @@ -2210,7 +2244,7 @@ mod api_tests { include_str!("../../docs/examples/asapquery-planning-snapshot.json"), ) .unwrap(); - let (planning, _) = snapshot.planning_request().unwrap(); + let (planning, _) = snapshot.clone().planning_request().unwrap(); let query = &planning.queries[0]; let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -2220,7 +2254,7 @@ mod api_tests { "queries": [{ "query_id": query.query_id, "query_string": query.query_string, "metric": "m", "window_secs": 60, "accuracy": query.accuracy, - "lifecycle": query.lifecycle, "window_implementations": [] + "lifecycle": query.lifecycle, "evaluation_phase_ms": 0, "window_cost_model": snapshot.implementation.window_cost_model }], "collector_ids": ["test"], "capability_snapshot_id": "test", "planner_revision": physical::compiler::PLANNER_REVISION, @@ -2274,7 +2308,7 @@ mod api_tests { include_str!("../../docs/examples/asapquery-compatibility-demo-snapshot.json"), ) .unwrap(); - let (planning, _) = snapshot.planning_request().unwrap(); + let (planning, _) = snapshot.clone().planning_request().unwrap(); let mut query = planning.queries[0].clone(); let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -2292,7 +2326,7 @@ mod api_tests { "queries": [{ "query_id": query.query_id, "query_string": query.query_string, "metric": metric, "window_secs": query.window_secs, "accuracy": query.accuracy, - "lifecycle": query.lifecycle, "window_implementations": query.window_implementations + "lifecycle": query.lifecycle, "evaluation_phase_ms": 0, "window_cost_model": { "implementation_id": "test", "cost": query.window_implementations[0].cost } }], "collector_ids": [], "capability_snapshot_id": "test", "planner_revision": physical::compiler::PLANNER_REVISION, diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 2ea40ab23..cd9d10319 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -37,6 +37,10 @@ use crate::query_plan::{ use crate::types_v2::AccuracyTarget; use planner_types::pre_asap::Source; +mod windows; +pub(super) use windows::gcd; +pub use windows::{prepare_window_implementations, WindowCostModel}; + pub const PLANNER_REVISION: &str = env!("ASAPPLANNER_REVISION"); pub const BACKEND_REVISION: &str = env!("ASAPQUERY_BACKEND_REVISION"); pub use asap_types::precompute_plan::BACKEND_COMPAT; @@ -114,6 +118,14 @@ pub struct WindowImplementationCandidate { pub slide_secs: u64, pub layout: asap_types::WindowMaterializationLayout, pub cost: ImplementationCostEvidence, + /// Only compiler-generated quotes may be repriced after changing their layout. + /// Serialized input always becomes provider evidence. + #[serde(skip)] + pub derived: bool, + /// This offer was generated for a restricted derived-maintenance cohort, + /// whose cadence cannot satisfy ordinary raw-state consumers of the same W. + #[serde(skip)] + pub cohort_only: bool, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] @@ -148,8 +160,6 @@ pub struct PlanningRequest { /// Original dashboard demand, in the same order as queries. None is legacy input. pub query_workload: Option, pub queries: Vec, - /// Compiler-owned quotes eligible for joint pane repricing; never inferred from model labels. - pub synthesized_window_queries: BTreeSet, pub evidence: HashMap, /// Fresh measured costs for Planner exact/summary composition sites, /// scoped to query IDs just like accuracy evidence. @@ -221,15 +231,11 @@ pub struct BackendLocalPlanningSnapshot { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(deny_unknown_fields)] pub struct BackendLocalImplementation { - /// Provider-priced concrete pane choices keyed by the registered PromQL text. - #[serde(default, skip_serializing_if = "HashMap::is_empty")] - pub window_candidates: HashMap>, pub lifecycle_costs: LifecycleCostEvidence, pub evidence_observed_at_unix_ms: u64, pub evidence_valid_for_ms: u64, pub horizon_seconds: f64, - pub window_implementation_id: String, - pub implementation_cost: ImplementationCostEvidence, + pub window_cost_model: WindowCostModel, #[serde(default, skip_serializing_if = "Option::is_none")] pub source_sample_interval_ms: Option, #[serde(default, skip_serializing_if = "u64_is_zero")] @@ -544,13 +550,6 @@ impl BackendLocalPlanningSnapshot { "QueryWorkload must contain at least one query".into(), )); } - for query in self.implementation.window_candidates.keys() { - if !entries.iter().any(|entry| &entry.query.0 == query) { - return Err(CompileError::Snapshot(format!( - "window candidates reference unregistered query `{query}`" - ))); - } - } let mut queries = Vec::with_capacity(entries.len()); let mut canonical_roots = Vec::with_capacity(entries.len()); let mut topk_evidence_by_id = HashMap::new(); @@ -595,14 +594,9 @@ impl BackendLocalPlanningSnapshot { horizon_seconds: self.implementation.horizon_seconds, costs: self.implementation.lifecycle_costs.clone(), }; - let derived_lifecycle = lifecycle.clone(); let post_asap = crate::planner_selection::keep_pre_asap(&parsed) .map_err(|error| CompileError::Snapshot(format!("query {index}: {error}")))?; canonical_roots.push(Rc::new(parsed)); - let mut cost = self.implementation.implementation_cost.clone(); - cost.workload_fingerprint = - canonical_promql(&query_string).map_err(CompileError::QueryPlan)?; - cost.horizon_seconds = self.implementation.horizon_seconds; let query_id = format!("compat-query-{index}"); if let Some(evidence) = self.implementation.topk_evidence.get(&query_string) { topk_evidence_by_id.insert(query_id.clone(), evidence.clone()); @@ -618,22 +612,7 @@ impl BackendLocalPlanningSnapshot { group_by: metadata.group_by_labels, accuracy, lifecycle, - window_implementations: self - .implementation - .window_candidates - .get(&query_string) - .cloned() - .unwrap_or_else(|| { - derived_window_candidates( - &self.implementation.window_implementation_id, - canonical_roots.last().expect("root pushed above"), - lookback_ms, - evaluation_interval_ms, - cost, - &derived_lifecycle, - self.implementation.query_staleness_margin_ms, - ) - }), + window_implementations: Vec::new(), runtime_policy: RuntimeRulePolicy::default(), }); } @@ -665,66 +644,13 @@ impl BackendLocalPlanningSnapshot { &exact_costs_by_id, self.implementation.erp.as_ref(), )?; - // Derived maintenance currently consumes full, non-overlapping source cohorts. - // Restrict only synthesized candidates; deployment-supplied evidence is authoritative. for query in &mut queries { - if self - .implementation - .window_candidates - .contains_key(&query.query_string) - { - continue; - } - let states = - collect_selected_materializations(&query.post_asap, true).map_err(|reason| { - CompileError::Query { - query_id: query.query_id.clone(), - reason, - } - })?; - let mut full_windows = BTreeSet::new(); - for state in &states { - if let Some(sources) = immutable_materialization_sources(&state.node) { - full_windows.insert(state.window_secs.unwrap_or(query.window_secs)); - for source in sources { - let (_, window, _) = - selected_input_contract(&source).map_err(|reason| { - CompileError::Query { - query_id: query.query_id.clone(), - reason, - } - })?; - full_windows.insert(window.unwrap_or(query.window_secs)); - } - } - } - let mut seen = BTreeSet::new(); - query.window_implementations.retain_mut(|candidate| { - if !full_windows.contains(&candidate.window_secs) { - return true; - } - if !seen.insert(candidate.window_secs) { - return false; - } - candidate.slide_secs = candidate.window_secs; - candidate.framework = SummaryWindowFramework::Tumbling; - candidate.layout = asap_types::WindowMaterializationLayout::Pane { - pane_secs: candidate.window_secs, - }; - candidate.implementation_id = format!( - "{}-{}s-derived-cohort", - self.implementation.window_implementation_id, candidate.window_secs - ); - candidate.cost = derived_window_cost( - &candidate.cost, - &query.lifecycle, - candidate.window_secs, - candidate.slide_secs, - &candidate.layout, - self.implementation.query_staleness_margin_ms, - ); - true - }); + prepare_window_implementations( + query, + &self.implementation.window_cost_model, + self.environment.target, + self.implementation.query_staleness_margin_ms, + )?; } // Composable lowering residualizes unsafe leaves individually; retain Planner siblings. Ok(( @@ -733,16 +659,6 @@ impl BackendLocalPlanningSnapshot { hybrid_execution: true, materialization_policy: None, query_workload: Some(workload), - synthesized_window_queries: queries - .iter() - .filter(|q| { - !self - .implementation - .window_candidates - .contains_key(&q.query_string) - }) - .map(|q| q.query_id.clone()) - .collect(), queries, evidence: topk_evidence_by_id, exact_composition_costs: exact_costs_by_id, @@ -1179,6 +1095,7 @@ impl PhysicalCompiler { } } } + let cohort_nodes = windows::cohort_nodes(&selected); for (ordinal, selected) in selected.into_iter().enumerate() { let mut branch_query = query.clone(); branch_query.window_secs = selected.window_secs.unwrap_or(query.window_secs); @@ -1186,9 +1103,14 @@ impl PhysicalCompiler { .group_by .clone() .unwrap_or_else(|| query.group_by.clone()); - branch_query - .window_implementations - .retain(|candidate| candidate.window_secs == branch_query.window_secs); + branch_query.window_implementations.retain(|candidate| { + candidate.window_secs == branch_query.window_secs + && if cohort_nodes.contains(&(Rc::as_ptr(&selected.node) as usize)) { + windows::is_full_cohort(candidate) + } else { + !candidate.cohort_only + } + }); let query = &branch_query; let lifecycle_costs = SummaryMaintenanceLifecycleCostInputs { build_cost: Some(Cost(query.lifecycle.costs.build)), @@ -1236,7 +1158,40 @@ impl PhysicalCompiler { let precompute_materialization = scoped_materialization(&aggregation, &selected.node)?; let materialization = precompute_materialization.policy_fingerprint(); - let state_consumers = consumers[&materialization] + let consumer_indices = consumers[&materialization] + .iter() + .copied() + .filter(|&index| { + let Some(workload) = &request.query_workload else { + return true; + }; + let other = &request.queries[index]; + if other.lifecycle.evaluation_interval_ms + != query.lifecycle.evaluation_interval_ms + { + return false; + } + let phases = workload + .entries() + .map(|entry| match entry.recurrence { + QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt { + evaluation_phase, + .. + }) => Some(evaluation_phase.0), + _ => None, + }) + .collect::>(); + match (phases[query_index], phases[index]) { + (Some(a), Some(b)) => { + let cadence = u64::from(query.lifecycle.evaluation_interval_ms); + let window = query.window_secs.saturating_mul(1_000); + a % cadence == b % cadence && a % window == b % window + } + _ => index == query_index, + } + }) + .collect::>(); + let state_consumers = consumer_indices .iter() .map(|index| &request.queries[*index]) .collect::>(); @@ -1246,12 +1201,10 @@ impl PhysicalCompiler { &model, &environment, &state_consumers, - request.query_workload.as_ref().map(|workload| { - ( - workload, - consumers[&materialization].iter().copied().collect(), - ) - }), + request + .query_workload + .as_ref() + .map(|workload| (workload, consumer_indices.clone())), )?; let window_implementation = query.window_implementations.iter() .find(|candidate| candidate.implementation_id == planner_selection.window_implementation_id @@ -1289,13 +1242,25 @@ impl PhysicalCompiler { .saturating_mul(1_000); runtime_materialization.pane_origin_ms = shared_pane_origin_ms( request.query_workload.as_ref(), - consumers[&materialization].iter().copied(), + [query_index], pane_width_ms, ) .map_err(|reason| CompileError::Query { query_id: query.query_id.clone(), reason, })?; + if matches!( + window_implementation.layout, + asap_types::WindowMaterializationLayout::FullWindow + ) { + runtime_materialization.pane_origin_ms = + runtime_materialization.pane_origin_ms.map(|end_phase| { + (i128::from(end_phase) + - i128::from(window_implementation.window_secs) * 1_000) + .rem_euclid(i128::from(pane_width_ms)) + as i64 + }); + } if let Some(source) = immutable_materialization_sources(&selected.node) { if environment.target != PhysicalDeploymentTarget::BackendLocalRemoteWrite { return Err(CompileError::Query { @@ -1610,6 +1575,8 @@ impl PhysicalCompiler { ))); } Ok(MaterializationBinding { + full_window_slide_ms: matches!(materialization.window_layout, asap_types::WindowMaterializationLayout::FullWindow) + .then_some(materialization.slide_interval.saturating_mul(1_000)), readout_lookback_ms: source_window.map(|seconds| seconds.saturating_mul(1_000)), materialization: fingerprint.into(), output_grouping: PhysicalGrouping::Reduce( @@ -2304,7 +2271,8 @@ fn validate_lifecycle_input( /// - **update fanout**: this is the layout's whole point (`worker.rs`'s /// `stores_full_windows` branch). A pane takes each sample exactly once; a /// full window takes it into every overlapping window that contains it, -/// `ceil(W / S)` of them. Charged at the supplied ingestion rate. +/// averaging `W / S` under the supplied stationary ingestion-rate model. +/// The maximum simultaneously open count is separately charged as residency. /// - **finalizations per read**: the mirror image. A full window is read /// whole; `W / P` panes are composed into one answer. Charged at the /// query's own evaluation cadence. @@ -2332,9 +2300,7 @@ pub(super) fn derived_window_cost( let window = window_secs.max(1) as f64; let slide = slide_secs.max(1) as f64; let (seal_interval, update_fanout, finalizations_per_read) = match layout { - asap_types::WindowMaterializationLayout::FullWindow => { - (slide, (window / slide).ceil(), 1.0) - } + asap_types::WindowMaterializationLayout::FullWindow => (slide, window / slide, 1.0), asap_types::WindowMaterializationLayout::Pane { pane_secs } => { let pane = (*pane_secs).max(1) as f64; (pane, 1.0, (window / pane).ceil()) @@ -2387,6 +2353,7 @@ pub(super) fn derived_window_cost( /// 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. +#[cfg(test)] fn range_selector_windows_secs(expr: &QueryExpr) -> BTreeSet { fn visit(expr: &QueryExpr, windows: &mut BTreeSet) { if let QueryExpr::TimeRange { range, .. } = expr { @@ -2432,6 +2399,7 @@ fn range_selector_windows_secs(expr: &QueryExpr) -> BTreeSet { /// Explicit snapshot candidates bypass this path. After logical selection, /// derived maintenance cohorts are restricted to their supported full windows; /// raw additive pane producers may subsequently be shared by Planner. +#[cfg(test)] fn derived_window_candidates( implementation_id: &str, expr: &QueryExpr, @@ -2445,74 +2413,23 @@ fn derived_window_candidates( 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; + let mut lifecycle = lifecycle.clone(); + lifecycle.evaluation_interval_ms = evaluation_interval_ms; windows .into_iter() - .flat_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 - }; - let window_label = if distinct { - format!("{implementation_id}-{window_secs}s") - } else { - implementation_id.to_string() - }; - // `Tumbling` pairs only with `Pane` in the validator's - // framework/layout table, so a non-sliding shape has no - // alternative to rank against and keeps its label unchanged. - let layouts: Vec<(String, asap_types::WindowMaterializationLayout)> = - if advances_within_window { - vec![ - ( - format!("{window_label}-pane-{slide_secs}s"), - asap_types::WindowMaterializationLayout::Pane { - pane_secs: slide_secs, - }, - ), - ( - format!("{window_label}-full-window"), - asap_types::WindowMaterializationLayout::FullWindow, - ), - ] - } else { - vec![( - window_label, - asap_types::WindowMaterializationLayout::Pane { - pane_secs: slide_secs, - }, - )] - }; - layouts - .into_iter() - .map(|(id, layout)| WindowImplementationCandidate { - implementation_id: id, - framework: if advances_within_window { - SummaryWindowFramework::Sliding - } else { - SummaryWindowFramework::Tumbling - }, - window_secs, - slide_secs, - cost: derived_window_cost( - &cost, - lifecycle, - window_secs, - slide_secs, - &layout, - staleness_margin_ms, - ), - layout, - }) - .collect::>() + .flat_map(|window| { + windows::derive( + &WindowCostModel { + implementation_id: implementation_id.into(), + cost: cost.clone(), + quotes: Vec::new(), + }, + &lifecycle, + window, + false, + PhysicalDeploymentTarget::BackendLocalRemoteWrite, + staleness_margin_ms, + ) }) .collect() } @@ -2546,30 +2463,7 @@ pub(super) fn validate_window_implementations( .layout .validate(candidate.window_secs, candidate.slide_secs) .is_ok() - && match (&candidate.framework, &candidate.layout) { - ( - SummaryWindowFramework::Tumbling | SummaryWindowFramework::Sliding, - asap_types::WindowMaterializationLayout::Pane { .. }, - ) - | ( - SummaryWindowFramework::Sliding, - asap_types::WindowMaterializationLayout::FullWindow, - ) => true, - ( - SummaryWindowFramework::Extension(name), - asap_types::WindowMaterializationLayout::HierarchicalRollup { .. }, - ) => name == "backend.exact-hierarchical-rollup.v1", - _ => false, - } - && match environment.target { - PhysicalDeploymentTarget::DistributedCollectors => { - matches!( - candidate.layout, - asap_types::WindowMaterializationLayout::FullWindow - ) || matches!(candidate.layout, asap_types::WindowMaterializationLayout::Pane { pane_secs } if pane_secs == candidate.window_secs) - } - PhysicalDeploymentTarget::BackendLocalRemoteWrite => true, - }; + && windows::supported(candidate, environment.target); if !valid { return Err(CompileError::Lifecycle { query_id: query.query_id.clone(), @@ -2790,11 +2684,21 @@ fn select_lifecycle( ..DataWorkload::default() }), }; - let (workload, indices) = - original_workload.unwrap_or((&workload, (0..consumers.len()).collect())); + let (workload, indices) = match original_workload { + Some((original, indices)) => { + let mut original = original.clone(); + // HTTP demand carries phases and requirements; its per-state source + // evidence comes from the same lifecycle input as window pricing. + if original.data_workload.is_none() { + original.data_workload = workload.data_workload; + } + (original, indices) + } + None => (workload, (0..consumers.len()).collect()), + }; let plan = plan_summary_maintenance_lifecycles( Rc::new(node.clone()), - WorkloadDemand::new(workload, &indices), + WorkloadDemand::new(&workload, &indices), environment.observed_at_unix_ms, Some(Horizon(query.lifecycle.horizon_seconds)), SummaryMaintenanceLifecycleCapabilities { @@ -4163,7 +4067,6 @@ pub(crate) mod tests { } Ok(PlanningRequest { logical_selection: Vec::new(), - synthesized_window_queries: BTreeSet::new(), hybrid_execution: false, materialization_policy: None, query_workload: None, @@ -4177,6 +4080,8 @@ pub(crate) mod tests { accuracy, lifecycle, window_implementations: vec![WindowImplementationCandidate { + derived: false, + cohort_only: false, implementation_id: "collector-tumbling-v1".into(), framework: SummaryWindowFramework::Tumbling, window_secs: 60, @@ -5359,10 +5264,87 @@ pub(crate) mod tests { } } + // A derived cohort must not impose its maintenance cadence on an unrelated raw leaf at the same W. + #[test] + fn raw_leaf_keeps_its_cadence_beside_a_same_window_derived_cohort() { + let mut snapshot = planning_snapshot(); + let model = snapshot.implementation.window_cost_model.clone(); + let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0]; + entry.query = Query("quantile(0.9, sum_over_time(m[1m]))".into()); + entry.demand = RepeatedDemand::FixedIntervalAt { + interval: RepetitionInterval(45_000), + evaluation_phase: planner_types::workload::TimestampMs(0), + }; + let mut right = snapshot.clone(); + right.query_workload.repeating_queries.as_mut().unwrap()[0].query = + Query("sum_over_time(n[1m])".into()); + let (mut request, env) = snapshot.planning_request().unwrap(); + let (right, _) = right.planning_request().unwrap(); + let left = request.queries[0].post_asap.clone(); + let right = right.queries[0].post_asap.clone(); + let right = Rc::new(SummaryNode { + expr: SummaryExpr::ValueOperation { + timing: planner_types::post_asap::ExecutionTiming::ReadTime, + operation: planner_types::post_asap::ValueOperation::FinalizeExactAccumulator, + child: right.clone(), + }, + schema: right.schema.clone(), + guarantee: None, + }); + request.queries[0].post_asap = Rc::new(SummaryNode { + expr: SummaryExpr::BinaryOp { + timing: planner_types::post_asap::ExecutionTiming::ReadTime, + lhs: left.clone(), + rhs: right, + operator: planner_types::post_asap::BinaryOperator { + kind: planner_types::pre_asap::BinaryOpKind::Arithmetic( + planner_types::pre_asap::ArithmeticOpKind::Add, + ), + vector_match: None, + }, + }, + schema: left.schema.clone(), + guarantee: None, + }); + let text = "quantile(0.9, sum_over_time(m[1m])) + sum_over_time(n[1m])"; + request.queries[0].query_string = text.into(); + request + .query_workload + .as_mut() + .unwrap() + .repeating_queries + .as_mut() + .unwrap()[0] + .query = Query(text.into()); + prepare_window_implementations(&mut request.queries[0], &model, env.target, 0).unwrap(); + request.queries[0] + .window_implementations + .retain(|candidate| { + matches!( + candidate.layout, + asap_types::WindowMaterializationLayout::Pane { .. } + ) + }); + let plan = PhysicalCompiler.compile(request, env).unwrap(); + assert!(plan + .precompute_plan + .materializations + .iter() + .any(|m| m.derived_input.is_some())); + let raw = plan + .precompute_plan + .materializations + .iter() + .find(|m| m.metric == "n") + .unwrap(); + assert_eq!(raw.slide_interval, 45); + assert_eq!(raw.window_layout.base_pane_secs(), 15); + } + // A full-window producer keeps overlapping accumulators alive even before publication. #[test] fn derived_window_regression_resident_cost() { - let template = planning_snapshot().implementation.implementation_cost; + let template = planning_snapshot().implementation.window_cost_model.cost; let mut lifecycle = planning_lifecycle(); lifecycle.costs = LifecycleCostEvidence { build: 0.0, @@ -5479,12 +5461,13 @@ pub(crate) mod tests { entry.query = Query(query.into()); entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact); let (derived, _) = snapshot.clone().planning_request().unwrap(); - snapshot.implementation.window_candidates.insert( - query.into(), - derived.queries[0].window_implementations.clone(), - ); + snapshot.implementation.window_cost_model.quotes = + derived.queries[0].window_implementations.clone(); let (request, env) = snapshot.planning_request().unwrap(); - assert!(request.synthesized_window_queries.is_empty()); + assert!(request.queries[0] + .window_implementations + .iter() + .any(|c| !c.derived)); let plan = PhysicalCompiler.compile(request, env).unwrap(); assert_eq!(plan.precompute_plan.materializations.len(), 2); } @@ -5507,7 +5490,7 @@ pub(crate) mod tests { // 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 cost = planning_snapshot().implementation.window_cost_model.cost; let expr = crate::query_parser::parse_query_expr_canonical( "quantile_over_time(0.5, data[5m])", AccuracyTarget::Exact, @@ -5524,11 +5507,11 @@ pub(crate) mod tests { .collect::>(), vec![ ( - "id-pane-30s", + "id-300s-slide-30s-pane-30s", asap_types::WindowMaterializationLayout::Pane { pane_secs: 30 } ), ( - "id-full-window", + "id-300s-slide-30s-full-window", asap_types::WindowMaterializationLayout::FullWindow ), ] @@ -5550,7 +5533,7 @@ pub(crate) mod tests { // not the magnitudes. #[test] fn derived_window_layout_prices_write_against_read_amplification() { - let cost = planning_snapshot().implementation.implementation_cost; + let cost = planning_snapshot().implementation.window_cost_model.cost; let expr = crate::query_parser::parse_query_expr_canonical( "quantile_over_time(0.5, data[5m])", AccuracyTarget::Exact, @@ -5598,7 +5581,7 @@ pub(crate) mod tests { // framework/layout table, so there is no alternative to price against it. #[test] fn tumbling_shapes_have_no_layout_alternative_to_rank() { - let cost = planning_snapshot().implementation.implementation_cost; + let cost = planning_snapshot().implementation.window_cost_model.cost; let expr = crate::query_parser::parse_query_expr_canonical( "quantile_over_time(0.5, data[5m])", AccuracyTarget::Exact, @@ -5614,7 +5597,7 @@ pub(crate) mod tests { 0, ); assert_eq!(derived.len(), 1); - assert_eq!(derived[0].implementation_id, "id"); + assert_eq!(derived[0].implementation_id, "id-300s-slide-300s-pane-300s"); assert_eq!(derived[0].framework, SummaryWindowFramework::Tumbling); } @@ -5624,7 +5607,7 @@ pub(crate) mod tests { 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; + let cost = planning_snapshot().implementation.window_cost_model.cost; for (lookback_ms, evaluation_ms) in [ (300_000, 30_000), (300_000, 300_000), @@ -5653,77 +5636,224 @@ pub(crate) mod tests { } } - // 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. + // Compilation keeps the demand grid while publishing independently sized panes or full windows. + #[test] + fn scheduled_window_layouts_preserve_cadence_phase_and_readout() { + for (evaluation, pane) in [(20, 20), (45, 15), (60, 60), (120, 60), (90, 30)] { + for phase in [0, 5_000] { + for full in [false, true] { + if full && evaluation == 60 { + continue; + } + let mut snapshot = planning_snapshot(); + let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0]; + entry.query = Query("sum_over_time(a[1m])".into()); + entry.requirements.accuracy = + AccuracyRequirement::Explicit(AccuracyTarget::Exact); + entry.demand = RepeatedDemand::FixedIntervalAt { + interval: RepetitionInterval(evaluation * 1_000), + evaluation_phase: planner_types::workload::TimestampMs(phase), + }; + let (mut request, env) = snapshot.planning_request().unwrap(); + request.queries[0].window_implementations.retain(|c| { + matches!( + c.layout, + asap_types::WindowMaterializationLayout::FullWindow + ) == full + }); + let plan = PhysicalCompiler.compile(request, env).unwrap(); + let config = &plan.precompute_plan.materializations[0]; + assert_eq!(config.slide_interval, u64::from(evaluation)); + assert_eq!(config.window_size, 60); + assert_eq!( + config.stored_window_ms(), + if full { 60_000 } else { pane * 1_000 } + ); + let entry = plan.query_plan.entries.values().next().unwrap(); + let binding = entry.materialization_bindings()[0]; + for tick in 3..6 { + let end = phase + u64::from(evaluation) * 1_000 * tick; + assert!( + binding.covers_range(end - 60_000, end), + "{evaluation} {phase} {full}" + ); + assert!(!binding.covers_range(end - 60_000 + 1, end + 1)); + } + } + } + } + } + + // Smaller common panes are selected only when saved maintenance exceeds extra merge work. + #[test] + fn different_cadences_share_common_panes_only_when_cheaper() { + for read_cost in [0.0, 1_000_000.0] { + let mut snapshot = planning_snapshot(); + snapshot.implementation.lifecycle_costs.read = read_cost; + snapshot + .implementation + .lifecycle_costs + .maintenance_per_update = 1.0; + let entries = snapshot.query_workload.repeating_queries.as_mut().unwrap(); + entries[0].query = Query("sum_over_time(a[1m])".into()); + entries[0].requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact); + entries[0].demand = RepeatedDemand::FixedIntervalAt { + interval: RepetitionInterval(20_000), + evaluation_phase: planner_types::workload::TimestampMs(0), + }; + let mut second = entries[0].clone(); + second.query = Query("sum_over_time(a[90s])".into()); + second.demand = RepeatedDemand::FixedIntervalAt { + interval: RepetitionInterval(30_000), + evaluation_phase: planner_types::workload::TimestampMs(0), + }; + entries.push(second); + let (mut request, env) = snapshot.planning_request().unwrap(); + for query in &mut request.queries { + query.window_implementations.retain(|c| { + matches!( + c.layout, + asap_types::WindowMaterializationLayout::Pane { .. } + ) + }); + } + let plan = PhysicalCompiler.compile(request, env).unwrap(); + assert_eq!( + plan.precompute_plan.materializations.len(), + if read_cost == 0.0 { 1 } else { 2 } + ); + if read_cost == 0.0 { + assert_eq!( + plan.precompute_plan.materializations[0].stored_window_ms(), + 10_000 + ); + assert_eq!(plan.lifecycle_estimates[0].expected_reads, 25.0); + assert_eq!(plan.lifecycle_estimates[0].expected_updates, 30_000.0); + } + } + } + + // An incompatible third consumer cannot disable sharing between the first two. #[test] - fn derived_window_candidate_stays_tumbling_without_a_dividing_cadence() { - let cost = planning_snapshot().implementation.implementation_cost; + fn sharing_keeps_profitable_subsets_with_other_phases_or_finer_cadences() { + for (third_phase, third_interval) in [(5_000, 20_000), (0, 1_000)] { + let mut snapshot = planning_snapshot(); + snapshot + .implementation + .lifecycle_costs + .maintenance_per_update = 1.0; + snapshot.implementation.lifecycle_costs.read = + if third_interval == 1_000 { 100.0 } else { 0.0 }; + let entries = snapshot.query_workload.repeating_queries.as_mut().unwrap(); + let template = entries[0].clone(); + entries.clear(); + for (window, phase, interval) in [ + (60, 0, 20_000), + (120, 0, 20_000), + (180, third_phase, third_interval), + ] { + let mut entry = template.clone(); + entry.query = Query(format!("sum_over_time(a[{window}s])")); + entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact); + entry.demand = RepeatedDemand::FixedIntervalAt { + interval: RepetitionInterval(interval), + evaluation_phase: planner_types::workload::TimestampMs(phase), + }; + entries.push(entry); + } + let (mut request, env) = snapshot.planning_request().unwrap(); + for query in &mut request.queries { + query.window_implementations.retain(|c| { + matches!( + c.layout, + asap_types::WindowMaterializationLayout::Pane { .. } + ) + }); + } + let plan = PhysicalCompiler.compile(request, env).unwrap(); + assert_eq!(plan.precompute_plan.materializations.len(), 2); + assert!(plan + .lifecycle_estimates + .iter() + .any(|e| e.consumer_query_ids.len() == 2)); + } + } + + // A subsecond cadence cannot be rounded into a different supported query schedule. + #[test] + fn fractional_cadence_uses_native_fallback_without_truncation() { + let mut snapshot = planning_snapshot(); + let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0]; + entry.query = Query("sum_over_time(a[1m])".into()); + entry.demand = RepeatedDemand::FixedIntervalAt { + interval: RepetitionInterval(1_500), + evaluation_phase: planner_types::workload::TimestampMs(0), + }; + let (request, env) = snapshot.planning_request().unwrap(); + assert!(request.queries[0].window_implementations.is_empty()); + let plan = PhysicalCompiler.compile(request, env).unwrap(); + assert!(plan.precompute_plan.materializations.is_empty()); + } + + // Every supported cadence has panes that exactly tile each scheduled range. + #[test] + fn derived_window_candidates_cover_dividing_nondividing_and_sparse_cadences() { let expr = crate::query_parser::parse_query_expr_canonical( - "quantile_over_time(0.5, data[5m])", + "sum_over_time(data[1m])", AccuracyTarget::Exact, ) .unwrap(); - for evaluation_ms in [300_000, 450_000, 45_000, 0] { - let derived = derived_window_candidates( + for (evaluation_secs, pane_secs) in [(20, 20), (45, 15), (60, 60), (120, 60), (90, 30)] { + let candidates = derived_window_candidates( "id", &expr, - 300_000, - evaluation_ms, - cost.clone(), + 60_000, + evaluation_secs * 1_000, + planning_snapshot().implementation.window_cost_model.cost, &planning_lifecycle(), 0, ); - let candidate = &derived[0]; + let pane = candidates + .iter() + .find(|c| { + matches!( + c.layout, + asap_types::WindowMaterializationLayout::Pane { .. } + ) + }) + .unwrap(); + assert_eq!(pane.slide_secs, u64::from(evaluation_secs)); assert_eq!( - ( - candidate.framework.clone(), - candidate.slide_secs, - candidate.layout.clone() - ), - ( - SummaryWindowFramework::Tumbling, - 300, - asap_types::WindowMaterializationLayout::Pane { pane_secs: 300 } - ), - "cadence {evaluation_ms}" + pane.layout, + asap_types::WindowMaterializationLayout::Pane { pane_secs } ); + assert!(pane + .layout + .validate(pane.window_secs, pane.slide_secs) + .is_ok()); } } - // Priced evidence is the evidence producer's to supply. A snapshot that - // carries its own candidates keeps them verbatim. + // A measured quote replaces only the matching shape, leaving other legal layouts available. #[test] - fn supplied_window_candidates_are_not_replaced_by_the_derivation() { + fn measured_window_quote_preserves_shape_and_price() { 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(), - &planning_lifecycle(), - 0, - ) - .remove(0); - supplied.framework = SummaryWindowFramework::Sliding; - supplied.slide_secs = 20; - supplied.layout = asap_types::WindowMaterializationLayout::Pane { pane_secs: 20 }; + let (derived, _) = snapshot.clone().planning_request().unwrap(); + let mut quote = derived.queries[0].window_implementations[0].clone(); + quote.implementation_id = "measured-pane".into(); + quote.cost.weighted_cost = 8.0; + quote.derived = false; snapshot .implementation - .window_candidates - .insert(query_string, vec![supplied.clone()]); + .window_cost_model + .quotes + .push(quote.clone()); let (request, _) = snapshot.planning_request().unwrap(); - assert_eq!(request.queries[0].window_implementations, vec![supplied]); + assert!(request.queries[0].window_implementations.contains("e)); + assert!(request.queries[0] + .window_implementations + .iter() + .any(|candidate| candidate.derived)); } // End to end: the retained-state count is derived from the pane width, so @@ -6305,13 +6435,15 @@ pub(crate) mod tests { query_workload, data_workload, implementation: BackendLocalImplementation { - window_candidates: HashMap::new(), lifecycle_costs: template.lifecycle.costs, evidence_observed_at_unix_ms: 9_500, evidence_valid_for_ms: 60_000, horizon_seconds: 300.0, - window_implementation_id: "backend-tumbling-v1".into(), - implementation_cost: template.window_implementations[0].cost.clone(), + window_cost_model: WindowCostModel { + implementation_id: "backend-tumbling-v1".into(), + cost: template.window_implementations[0].cost.clone(), + quotes: Vec::new(), + }, source_sample_interval_ms: None, query_staleness_margin_ms: 0, max_retained_summary_bytes: DEFAULT_RETAINED_SUMMARY_MEMORY_BUDGET_BYTES, diff --git a/control_plane/src/physical/compiler/windows.rs b/control_plane/src/physical/compiler/windows.rs new file mode 100644 index 000000000..cc8134ff7 --- /dev/null +++ b/control_plane/src/physical/compiler/windows.rs @@ -0,0 +1,363 @@ +//! Generate feasible layouts from selected state requirements before pricing them. +use super::*; +use asap_types::WindowMaterializationLayout; + +/// One input contract for snapshot and HTTP window planning. Quotes apply only +/// to their exact shape and workload; absent quotes use the lifecycle unit costs. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct WindowCostModel { + pub implementation_id: String, + pub cost: ImplementationCostEvidence, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub quotes: Vec, +} + +pub(in crate::physical) fn gcd(mut a: u64, mut b: u64) -> u64 { + while b != 0 { + (a, b) = (b, a % b); + } + a +} + +pub(super) fn is_full_cohort(candidate: &WindowImplementationCandidate) -> bool { + candidate.slide_secs == candidate.window_secs + && candidate.layout + == WindowMaterializationLayout::Pane { + pane_secs: candidate.window_secs, + } +} + +pub(super) fn cohort_nodes(states: &[SelectedMaterialization]) -> BTreeSet { + let mut nodes = BTreeSet::new(); + for state in states { + if let Some(sources) = immutable_materialization_sources(&state.node) { + nodes.insert(Rc::as_ptr(&state.node) as usize); + nodes.extend(sources.iter().map(|source| Rc::as_ptr(source) as usize)); + } + } + nodes +} + +pub(super) fn supported( + candidate: &WindowImplementationCandidate, + target: PhysicalDeploymentTarget, +) -> bool { + candidate + .layout + .validate(candidate.window_secs, candidate.slide_secs) + .is_ok() + && match (&candidate.framework, &candidate.layout) { + ( + SummaryWindowFramework::Tumbling | SummaryWindowFramework::Sliding, + WindowMaterializationLayout::Pane { pane_secs }, + ) => { + target == PhysicalDeploymentTarget::BackendLocalRemoteWrite + || *pane_secs == candidate.window_secs + } + (SummaryWindowFramework::Sliding, WindowMaterializationLayout::FullWindow) => { + // Collector schedulers currently support overlapping or tumbling windows only. + target == PhysicalDeploymentTarget::BackendLocalRemoteWrite + || candidate.slide_secs <= candidate.window_secs + } + _ => false, + } +} + +pub(super) fn derive( + model: &WindowCostModel, + lifecycle: &LifecyclePlanningInput, + window_secs: u64, + full_cohort: bool, + target: PhysicalDeploymentTarget, + staleness_margin_ms: u64, +) -> Vec { + let evaluation_ms = u64::from(lifecycle.evaluation_interval_ms); + // Runtime layouts have second precision. Never truncate a fractional cadence. + if window_secs == 0 || evaluation_ms == 0 || evaluation_ms % 1_000 != 0 { + return Vec::new(); + } + let evaluation_secs = evaluation_ms / 1_000; + let pane_secs = if full_cohort { + window_secs + } else { + gcd(window_secs, evaluation_secs) + }; + let slide_secs = if full_cohort { + window_secs + } else { + evaluation_secs + }; + let mut layouts = vec![WindowMaterializationLayout::Pane { pane_secs }]; + if !full_cohort && evaluation_secs != window_secs { + layouts.push(WindowMaterializationLayout::FullWindow); + } + layouts + .into_iter() + .filter_map(|layout| { + let suffix = match layout { + WindowMaterializationLayout::Pane { pane_secs } => format!("pane-{pane_secs}s"), + _ => "full-window".into(), + }; + let candidate = WindowImplementationCandidate { + implementation_id: format!( + "{}-{window_secs}s-slide-{slide_secs}s-{suffix}", + model.implementation_id + ), + framework: if slide_secs == window_secs { + SummaryWindowFramework::Tumbling + } else { + SummaryWindowFramework::Sliding + }, + window_secs, + slide_secs, + cost: derived_window_cost( + &model.cost, + lifecycle, + window_secs, + slide_secs, + &layout, + staleness_margin_ms, + ), + layout, + derived: true, + cohort_only: full_cohort && slide_secs != evaluation_secs, + }; + supported(&candidate, target).then_some(candidate) + }) + .collect() +} + +pub fn prepare_window_implementations( + query: &mut PlanningQuery, + model: &WindowCostModel, + target: PhysicalDeploymentTarget, + staleness_margin_ms: u64, +) -> Result<(), CompileError> { + if model.implementation_id.trim().is_empty() { + return Err(CompileError::Lifecycle { + query_id: query.query_id.clone(), + reason: "window cost model requires an implementation identity".into(), + }); + } + let mut model = model.clone(); + let fingerprint = canonical_promql(&query.query_string).map_err(CompileError::QueryPlan)?; + model.cost.workload_fingerprint = fingerprint.clone(); + model.cost.horizon_seconds = query.lifecycle.horizon_seconds; + let states = collect_selected_materializations( + &query.post_asap, + target == PhysicalDeploymentTarget::BackendLocalRemoteWrite, + ) + .map_err(|reason| CompileError::Query { + query_id: query.query_id.clone(), + reason, + })?; + let cohorts = cohort_nodes(&states); + let requirements = states + .iter() + .map(|state| { + ( + state.window_secs.unwrap_or(query.window_secs), + cohorts.contains(&(Rc::as_ptr(&state.node) as usize)), + ) + }) + .collect::>(); + let mut candidates = requirements + .iter() + .flat_map(|&(window, cohort)| { + derive( + &model, + &query.lifecycle, + window, + cohort, + target, + staleness_margin_ms, + ) + }) + .collect::>(); + let mut quote_shapes = BTreeSet::new(); + for quote in model + .quotes + .iter() + .filter(|q| q.cost.workload_fingerprint == fingerprint) + { + let shape = ( + quote.window_secs, + quote.slide_secs, + serde_json::to_string("e.layout).unwrap(), + ); + if !quote_shapes.insert(shape) { + return Err(CompileError::Lifecycle { + query_id: query.query_id.clone(), + reason: "duplicate measured window layout".into(), + }); + } + let applicable = requirements.iter().any(|&(window, cohort)| { + quote.window_secs == window + && if cohort { + is_full_cohort(quote) + } else { + quote.slide_secs.saturating_mul(1_000) + == u64::from(query.lifecycle.evaluation_interval_ms) + } + }); + if !applicable || !supported(quote, target) { + return Err(CompileError::Lifecycle { + query_id: query.query_id.clone(), + reason: format!( + "window quote `{}` does not match an executable state layout", + quote.implementation_id + ), + }); + } + candidates.retain(|candidate| { + !(candidate.window_secs == quote.window_secs + && candidate.slide_secs == quote.slide_secs + && candidate.layout == quote.layout) + }); + let mut quote = quote.clone(); + quote.derived = false; + quote.cohort_only = quote.slide_secs.saturating_mul(1_000) + != u64::from(query.lifecycle.evaluation_interval_ms); + candidates.push(quote); + } + let mut unique = BTreeMap::new(); + for candidate in &candidates { + if let Some(previous) = unique.insert(&candidate.implementation_id, candidate) { + if previous != candidate { + return Err(CompileError::Lifecycle { + query_id: query.query_id.clone(), + reason: "window implementation identity describes different offers".into(), + }); + } + } + } + let mut ids = BTreeSet::new(); + candidates.retain(|candidate| ids.insert(candidate.implementation_id.clone())); + query.window_implementations = candidates; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn snapshot() -> BackendLocalPlanningSnapshot { + serde_json::from_str(include_str!( + "../../../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap() + } + + // Target constraints eliminate infeasible offers before lifecycle selection. + #[test] + fn collector_generation_excludes_partial_panes_and_sparse_full_windows() { + let snapshot = snapshot(); + let model = snapshot.implementation.window_cost_model.clone(); + let (request, _) = snapshot.planning_request().unwrap(); + let mut lifecycle = request.queries[0].lifecycle.clone(); + for (interval, expected_count) in [ + (20_000, 1), + (45_000, 1), + (60_000, 1), + (90_000, 0), + (120_000, 1), + ] { + lifecycle.evaluation_interval_ms = interval; + let candidates = derive( + &model, + &lifecycle, + 60, + false, + PhysicalDeploymentTarget::DistributedCollectors, + 0, + ); + assert_eq!(candidates.len(), expected_count, "{interval}"); + assert!(candidates.iter().all(|candidate| supported( + candidate, + PhysicalDeploymentTarget::DistributedCollectors + ))); + } + } + + // A quote cannot authorize a pane that cuts scheduled ranges or an unimplemented rollup. + #[test] + fn quotes_cannot_bypass_layout_or_runtime_constraints() { + let snapshot = snapshot(); + let mut model = snapshot.implementation.window_cost_model.clone(); + let (request, _) = snapshot.planning_request().unwrap(); + let mut query = request.queries[0].clone(); + query.lifecycle.evaluation_interval_ms = 20_000; + prepare_window_implementations( + &mut query, + &model, + PhysicalDeploymentTarget::BackendLocalRemoteWrite, + 0, + ) + .unwrap(); + let mut quote = query.window_implementations[0].clone(); + quote.layout = WindowMaterializationLayout::Pane { pane_secs: 30 }; + model.quotes = vec![quote.clone()]; + assert!(prepare_window_implementations( + &mut query, + &model, + PhysicalDeploymentTarget::BackendLocalRemoteWrite, + 0 + ) + .is_err()); + quote.framework = + SummaryWindowFramework::Extension("backend.exact-hierarchical-rollup.v1".into()); + quote.layout = WindowMaterializationLayout::HierarchicalRollup { + base_pane_secs: 10, + levels_secs: vec![30], + }; + model.quotes = vec![quote]; + assert!(prepare_window_implementations( + &mut query, + &model, + PhysicalDeploymentTarget::BackendLocalRemoteWrite, + 0 + ) + .is_err()); + } + + // Conflicting identities and repeated shape evidence must fail rather than silently lose an offer. + #[test] + fn conflicting_quote_ids_and_duplicate_shapes_are_rejected() { + let snapshot = snapshot(); + let mut model = snapshot.implementation.window_cost_model.clone(); + let (request, _) = snapshot.planning_request().unwrap(); + let mut query = request.queries[0].clone(); + let mut quote = query.window_implementations[0].clone(); + quote.implementation_id = query.window_implementations[1].implementation_id.clone(); + model.quotes = vec![quote.clone()]; + assert!(prepare_window_implementations( + &mut query, + &model, + PhysicalDeploymentTarget::BackendLocalRemoteWrite, + 0 + ) + .is_err()); + quote.implementation_id = "duplicate-shape".into(); + model.quotes.push(quote); + assert!(prepare_window_implementations( + &mut query, + &model, + PhysicalDeploymentTarget::BackendLocalRemoteWrite, + 0 + ) + .is_err()); + } + + // External serialization never grants permission to reinterpret measured prices. + #[test] + fn serialized_generated_quote_loses_compiler_provenance() { + let (request, _) = snapshot().planning_request().unwrap(); + let candidate = &request.queries[0].window_implementations[0]; + assert!(candidate.derived); + let value = serde_json::to_value(candidate).unwrap(); + assert!(value.get("derived").is_none()); + let decoded: WindowImplementationCandidate = serde_json::from_value(value).unwrap(); + assert!(!decoded.derived); + } +} diff --git a/control_plane/src/physical/pane_reuse.rs b/control_plane/src/physical/pane_reuse.rs index ae34cb15f..713624be2 100644 --- a/control_plane/src/physical/pane_reuse.rs +++ b/control_plane/src/physical/pane_reuse.rs @@ -1,11 +1,12 @@ use super::compiler::{ - derived_window_cost, retained_state_count, CollectorMaterialization, + derived_window_cost, gcd, retained_state_count, CollectorMaterialization, MaterializationLifecycleEstimate, PlanningRequest, RuntimeRulePolicy, }; +use asap_types::WindowMaterializationLayout; use planner_types::post_asap::{PostAsapNodeId, SummaryWindowFramework}; use std::collections::{BTreeMap, BTreeSet}; -/// Lower Planner's costed pane-reuse groups without changing logical readouts. +/// Select and install cheaper shared panes without changing logical readouts. /// Only raw additive states are eligible; derived cohorts retain their full-window identity. #[allow(clippy::too_many_arguments)] pub(super) fn share_additive_panes( @@ -17,8 +18,7 @@ pub(super) fn share_additive_panes( policies: &mut BTreeMap, estimates: &mut BTreeMap, ) { - use asap_aware_mapping::pane_sharing::{select_shared_panes, PaneReuseCandidate}; - use asap_types::{AggregationType, WindowKind, WindowMaterializationLayout}; + use asap_types::{AggregationType, WindowKind}; let derived_sources = materializations .iter() .filter_map(|m| m.derived_input.as_ref()) @@ -26,7 +26,9 @@ pub(super) fn share_additive_panes( .collect::>(); let mut seen = BTreeSet::new(); let mut physical = Vec::new(); - let mut offers = Vec::new(); + let mut groups: Vec> = Vec::new(); + let mut keys = Vec::new(); + let mut member_consumers = Vec::new(); for m in materializations.iter() { let old = m.policy_fingerprint(); if !seen.insert(old) @@ -54,54 +56,70 @@ pub(super) fn share_additive_panes( continue; }; let query = &request.queries[first]; - // Explicitly priced implementations are not repriced or replaced. + let Some(estimate) = estimates.get(&old) else { + continue; + }; + // Provenance belongs to the selected layout, not to its request entry point. if consumers.iter().any(|index| { - let q = &request.queries[*index]; - q.lifecycle != query.lifecycle - || q.accuracy != query.accuracy - || !request.synthesized_window_queries.contains(&q.query_id) + !request.queries[*index] + .window_implementations + .iter() + .any(|candidate| { + candidate.implementation_id == estimate.window_implementation_id + && candidate.derived + }) }) { continue; } - let mut canonical = m.clone(); - canonical.window_size = pane_secs; - canonical.slide_interval = pane_secs; - canonical.window_type = WindowKind::Tumbling; let Some(policy) = policies.get(&old) else { continue; }; + let mut lifecycle = query.lifecycle.clone(); + lifecycle.evaluation_interval_ms = 0; + if consumers.iter().any(|index| { + let mut other = request.queries[*index].lifecycle.clone(); + other.evaluation_interval_ms = 0; + other != lifecycle || request.queries[*index].accuracy != query.accuracy + }) { + continue; + } + // Normalize only layout dimensions. Source, projection, grouping, state, + // policy and unit-cost evidence must still agree before considering reuse. + let mut canonical = m.clone(); + canonical.window_size = 1; + canonical.slide_interval = 1; + canonical.window_type = WindowKind::Tumbling; + canonical.window_layout = WindowMaterializationLayout::Pane { pane_secs: 1 }; + canonical.pane_origin_ms = Some(0); let key = ( canonical.policy_fingerprint(), - serde_json::to_string(&query.lifecycle).unwrap(), + serde_json::to_string(&lifecycle).unwrap(), serde_json::to_string(policy).unwrap(), + serde_json::to_string(&query.accuracy).unwrap(), ); - let mut maintenance = query.lifecycle.clone(); - maintenance.costs.read = 0.0; - let Some(template) = query.window_implementations.first() else { - continue; - }; - let producer_cost = derived_window_cost( - &template.cost, - &maintenance, - m.window_size, - m.slide_interval, - &m.window_layout, - request.query_staleness_margin_ms, - ) - .weighted_cost; - let read_cost = query.lifecycle.costs.read * query.lifecycle.horizon_seconds - / (f64::from(query.lifecycle.evaluation_interval_ms) / 1000.0) - * (m.window_size / pane_secs) as f64 - * consumers.len() as f64; - offers.push(PaneReuseCandidate { - compatibility: key, - lookback_ms: m.window_size.saturating_mul(1000), - producer_cost, - read_cost, - }); - physical.push((old, canonical)); + let index = physical.len(); + if let Some(group) = groups.iter_mut().find(|group| keys[group[0]] == key) { + group.push(index); + } else { + groups.push(vec![index]); + } + keys.push(key); + member_consumers.push(consumers); + physical.push((old, m.clone())); + } + let groups = select_shared_groups(request, &physical, &member_consumers, estimates, groups); + for group in &groups { + for &index in &group.members { + let canonical = &mut physical[index].1; + canonical.window_size = group.pane_secs; + canonical.slide_interval = group.pane_secs; + canonical.window_type = WindowKind::Tumbling; + canonical.window_layout = WindowMaterializationLayout::Pane { + pane_secs: group.pane_secs, + }; + canonical.pane_origin_ms = Some(group.origin); + } } - let groups = select_shared_panes(&offers); let mut target_counts = BTreeMap::new(); for group in &groups { *target_counts @@ -178,6 +196,8 @@ pub(super) fn share_additive_panes( producer.query_id = format!("state-{}", new.0); producer.window_secs = canonical.window_size; producer.slide_secs = canonical.slide_interval; + producer.window_layout = canonical.window_layout.clone(); + producer.pane_origin_ms = canonical.pane_origin_ms; producer.abstract_window_framework = SummaryWindowFramework::Tumbling; producer.window_implementation_id = estimates[&new].window_implementation_id.clone(); } @@ -185,3 +205,138 @@ pub(super) fn share_additive_panes( let mut seen = BTreeSet::new(); producers.retain(|producer| seen.insert(producer.materialization)); } + +struct SharedGroup { + members: Vec, + lookback_ms: u64, + cost: f64, + savings: f64, + pane_secs: u64, + origin: i64, +} + +fn select_shared_groups( + request: &PlanningRequest, + physical: &[( + asap_types::PolicyFingerprint, + asap_types::PrecomputeMaterialization, + )], + member_consumers: &[BTreeSet], + estimates: &BTreeMap, + groups: Vec>, +) -> Vec { + let price = |members: Vec| { + if members.len() < 2 { + return None; + } + let pane_secs = members + .iter() + .map(|&i| physical[i].1.window_layout.base_pane_secs()) + .reduce(gcd) + .unwrap(); + let origin = physical[members[0]] + .1 + .pane_origin_ms + .unwrap() + .rem_euclid((pane_secs * 1_000) as i64); + if members.iter().any(|&i| { + physical[i] + .1 + .pane_origin_ms + .unwrap() + .rem_euclid((pane_secs * 1_000) as i64) + != origin + }) { + return None; + } + let mut independent = 0.0; + let mut producer = 0.0_f64; + let mut reads = 0.0; + let mut lookback_ms = 0; + for &index in &members { + let (old, m) = &physical[index]; + let consumers = &member_consumers[index]; + let query = &request.queries[*consumers.first().unwrap()]; + let selected_id = &estimates[old].window_implementation_id; + let template = &query + .window_implementations + .iter() + .find(|candidate| &candidate.implementation_id == selected_id) + .unwrap() + .cost; + let mut maintenance = query.lifecycle.clone(); + maintenance.costs.read = 0.0; + independent += derived_window_cost( + template, + &maintenance, + m.window_size, + m.slide_interval, + &m.window_layout, + request.query_staleness_margin_ms, + ) + .weighted_cost; + producer = producer.max( + derived_window_cost( + template, + &maintenance, + m.window_size, + m.slide_interval, + &WindowMaterializationLayout::Pane { pane_secs }, + request.query_staleness_margin_ms, + ) + .weighted_cost, + ); + for &consumer in consumers { + let lifecycle = &request.queries[consumer].lifecycle; + let unit_reads = lifecycle.costs.read * lifecycle.horizon_seconds + / (f64::from(lifecycle.evaluation_interval_ms) / 1_000.0); + independent += + unit_reads * (m.window_size / m.window_layout.base_pane_secs()) as f64; + reads += unit_reads * (m.window_size / pane_secs) as f64; + } + lookback_ms = lookback_ms.max(m.window_size.saturating_mul(1_000)); + } + let cost = producer + reads; + if !independent.is_finite() || !cost.is_finite() || cost >= independent { + return None; + } + Some(SharedGroup { + members, + lookback_ms, + cost, + savings: independent - cost, + pane_secs, + origin, + }) + }; + let mut selected_groups = Vec::new(); + for mut remaining in groups { + while remaining.len() >= 2 { + if let Some(group) = price(remaining.clone()) { + selected_groups.push(group); + break; + } + // An incompatible phase or expensive fine-pane consumer must not + // prevent the remaining consumers from sharing profitable state. + let mut best: Option = None; + for (position, &left) in remaining.iter().enumerate() { + for &right in &remaining[position + 1..] { + if let Some(group) = price(vec![left, right]) { + if best + .as_ref() + .is_none_or(|previous| group.savings > previous.savings) + { + best = Some(group); + } + } + } + } + let Some(group) = best else { + break; + }; + remaining.retain(|index| !group.members.contains(index)); + selected_groups.push(group); + } + } + selected_groups +} diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index bcce290d2..fdb39f180 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -1266,6 +1266,7 @@ mod catalog_binding_tests { QueryNodeId(1), QueryPlanNode::ReadMaterialization { binding: MaterializationBinding { + full_window_slide_ms: None, item_labels: Vec::new(), materialization: config.policy_fingerprint().into(), output_grouping: PhysicalGrouping::PerEntity, diff --git a/control_plane/src/query_plan/logical.rs b/control_plane/src/query_plan/logical.rs index e802ea4ba..4baadd8e0 100644 --- a/control_plane/src/query_plan/logical.rs +++ b/control_plane/src/query_plan/logical.rs @@ -647,6 +647,7 @@ mod hybrid_tests { crate::physical::compiler::materialization_leaf_contract(node) .map_err(QueryPlanError::Invalid)?; Ok(MaterializationBinding { + full_window_slide_ms: None, item_labels: Vec::new(), materialization: asap_types::PolicyFingerprint( if spatial_filter.is_empty() { 7 } else { 8 }, diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs index 52f27b24d..27752371b 100644 --- a/crates/asap_types/src/aggregation_config.rs +++ b/crates/asap_types/src/aggregation_config.rs @@ -41,10 +41,8 @@ impl WindowMaterializationLayout { } pub fn validate(&self, window_secs: u64, slide_secs: u64) -> Result<(), String> { - if window_secs == 0 || slide_secs == 0 || slide_secs > window_secs { - return Err( - "window and slide must be positive and slide must not exceed window".into(), - ); + if window_secs == 0 || slide_secs == 0 { + return Err("window and slide must be positive".into()); } match self { Self::Pane { pane_secs } => { diff --git a/crates/asap_types/src/plan_publication.rs b/crates/asap_types/src/plan_publication.rs index 35ce812f5..3631b6cef 100644 --- a/crates/asap_types/src/plan_publication.rs +++ b/crates/asap_types/src/plan_publication.rs @@ -61,6 +61,16 @@ impl PhysicalPlanPublication { .get(&binding.materialization.fingerprint()) .copied() .ok_or("query binding has no precompute materialization")?; + let full_slide = matches!( + config.window_layout, + crate::WindowMaterializationLayout::FullWindow + ) + .then_some(config.slide_interval.saturating_mul(1_000)); + if binding.full_window_slide_ms != full_slide { + return Err( + "query full-window cadence differs from precompute definition".into(), + ); + } if config.stored_window_ms() != binding.window_ms { return Err("query pane differs from precompute stored window".into()); } diff --git a/crates/asap_types/src/query_plan.rs b/crates/asap_types/src/query_plan.rs index 8b0c48e8c..a8164d8bd 100644 --- a/crates/asap_types/src/query_plan.rs +++ b/crates/asap_types/src/query_plan.rs @@ -123,6 +123,17 @@ impl QueryPlan { "zero physical pane duration".into(), )); } + if binding.full_window_slide_ms.is_some() + != matches!( + identity.window_layout, + crate::WindowMaterializationLayout::FullWindow + ) + || binding.full_window_slide_ms == Some(0) + { + return Err(QueryPlanError::Invalid( + "query storage layout differs from catalog definition".into(), + )); + } if binding.pane_origin_ms != identity.pane_origin_ms { return Err(QueryPlanError::Invalid( "query pane origin differs from catalog definition".into(), @@ -464,6 +475,10 @@ pub enum FallbackPolicy { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(deny_unknown_fields)] pub struct MaterializationBinding { + /// Complete-window storage advances independently of its stored extent. + /// None denotes disjoint pane storage. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub full_window_slide_ms: Option, pub materialization: SummaryDefinitionId, /// Query operator grouping applied while folding those SIDs. pub output_grouping: PhysicalGrouping, @@ -484,6 +499,33 @@ pub struct MaterializationBinding { pub readout_lookback_ms: Option, } +impl MaterializationBinding { + /// Both range boundaries must identify complete stored state. Full windows + /// use a start grid; their end grid is displaced by the window width. + pub fn covers_range(&self, start_ms: u64, end_ms: u64) -> bool { + let Some(origin) = self.pane_origin_ms else { + return false; + }; + if self.window_ms == 0 || end_ms <= start_ms { + return false; + } + let start = i128::from(start_ms) - i128::from(origin); + match self.full_window_slide_ms { + Some(slide) => { + slide != 0 + && end_ms - start_ms == self.window_ms + && start.rem_euclid(i128::from(slide)) == 0 + } + None => { + start.rem_euclid(i128::from(self.window_ms)) == 0 + && (i128::from(end_ms) - i128::from(origin)) + .rem_euclid(i128::from(self.window_ms)) + == 0 + } + } + } +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(tag = "mode", content = "keys", rename_all = "snake_case")] pub enum PhysicalGrouping { diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index b37d4c798..bd6da65f6 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -1612,6 +1612,7 @@ mod tests { .next() .unwrap(); let binding = asap_types::query_plan::MaterializationBinding { + full_window_slide_ms: None, materialization: asap_types::PolicyFingerprint(policy).into(), output_grouping: asap_types::query_plan::PhysicalGrouping::Reduce(vec!["job".into()]), item_labels: vec![], diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs index c47a9c3fe..2ca7c7b70 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs @@ -584,6 +584,7 @@ mod tests { read, QueryPlanNode::ReadMaterialization { binding: MaterializationBinding { + full_window_slide_ms: None, materialization, output_grouping: PhysicalGrouping::Reduce(Vec::new()), item_labels: Vec::new(), diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs index 26183c1f6..42b162796 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs @@ -154,18 +154,7 @@ fn execute_relation_subtree( is_cumulative, ) .map_err(|error| format!("incomplete leaf coverage: {error:?}"))?; - let origin = binding - .pane_origin_ms - .ok_or_else(|| "incomplete leaf coverage: missing pane origin".to_owned())?; - let start = i64::try_from(t0_ms) - .map_err(|_| "incomplete leaf coverage: start exceeds i64".to_owned())?; - let end = i64::try_from(t1_ms) - .map_err(|_| "incomplete leaf coverage: end exceeds i64".to_owned())?; - let pane = i64::try_from(binding.window_ms) - .map_err(|_| "incomplete leaf coverage: pane exceeds i64".to_owned())?; - if pane <= 0 - || (start - origin).rem_euclid(pane) != 0 - || (end - origin).rem_euclid(pane) != 0 + if !binding.covers_range(t0_ms, t1_ms) || !complete_pane_coverage( leaf_outcome.coverage, (t0_ms, t1_ms), @@ -174,7 +163,7 @@ fn execute_relation_subtree( { return Err(format!( "incomplete leaf coverage: requested ({t0_ms}, {t1_ms}), observed {:?}, pane {} origin {}", - leaf_outcome.coverage, binding.window_ms, origin + leaf_outcome.coverage, binding.window_ms, binding.pane_origin_ms.unwrap_or(0) )); } } diff --git a/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs b/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs index aa1537b3e..c0ae6e712 100644 --- a/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs +++ b/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs @@ -936,6 +936,7 @@ mod tests { QueryNodeId(3), QueryPlanNode::ReadMaterialization { binding: MaterializationBinding { + full_window_slide_ms: None, item_labels: Vec::new(), materialization: MATERIALIZATION.into(), output_grouping: PhysicalGrouping::Reduce(vec!["job".into()]), diff --git a/data_plane/src/query_engines/asap_query_engine/live_serve.rs b/data_plane/src/query_engines/asap_query_engine/live_serve.rs index 95a36b7de..ab88856c6 100644 --- a/data_plane/src/query_engines/asap_query_engine/live_serve.rs +++ b/data_plane/src/query_engines/asap_query_engine/live_serve.rs @@ -439,6 +439,7 @@ mod tests { asap_types::query_plan::QueryNodeId(1), asap_types::query_plan::QueryPlanNode::ReadMaterialization { binding: asap_types::query_plan::MaterializationBinding { + full_window_slide_ms: None, item_labels: Vec::new(), materialization: policy.into(), output_grouping: asap_types::query_plan::PhysicalGrouping::PerEntity, diff --git a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs index 8632aba0a..814719c62 100644 --- a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs +++ b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs @@ -972,6 +972,7 @@ mod tests { QueryNodeId(1), QueryPlanNode::ReadMaterialization { binding: MaterializationBinding { + full_window_slide_ms: None, materialization: config.policy_fingerprint().into(), output_grouping: PhysicalGrouping::PerEntity, item_labels: vec![], @@ -1021,19 +1022,20 @@ mod tests { canonical, &node, asap_types::query_plan::InstantExecution { - lookback_ms: 60_000, + lookback_ms: 1_000, full_history: false, cumulative_readout: true, }, asap_types::query_plan::FallbackPolicy::ExactBackend, |_node, _family| { Ok(asap_types::query_plan::MaterializationBinding { + full_window_slide_ms: None, item_labels: Vec::new(), materialization: asap_types::PolicyFingerprint(123).into(), output_grouping: asap_types::query_plan::PhysicalGrouping::PerEntity, - window_ms: 60_000, - pane_origin_ms: Some(2_000), - readout_lookback_ms: Some(60_000), + window_ms: 1_000, + pane_origin_ms: Some(0), + readout_lookback_ms: Some(1_000), }) }, ) @@ -1121,6 +1123,104 @@ mod tests { assert_eq!(outcome.coverage, Some((2_000, 2_000))); } + // Exercise the compiled plan and runtime bucket assignment against raw integer samples. + #[test] + fn compiled_window_schedules_execute_exact_ranges() { + use crate::precompute_engine::window_manager::WindowManager; + use control_plane::physical::compiler::{BackendLocalPlanningSnapshot, PhysicalCompiler}; + for evaluation_secs in [20, 45, 60, 120, 90] { + for phase_ms in [0, 5_000] { + for full in [false, true] { + if full && evaluation_secs == 60 { + continue; + } + let mut snapshot: serde_json::Value = serde_json::from_str(include_str!( + "../../../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap(); + let entry = &mut snapshot["query_workload"]["repeating_queries"][0]; + entry["query"] = serde_json::json!("sum_over_time(a[1m])"); + entry["requirements"]["accuracy"]["explicit"] = serde_json::json!("Exact"); + entry["demand"]["fixed_interval_at"] = serde_json::json!({ + "interval": evaluation_secs * 1_000, "evaluation_phase": phase_ms + }); + let snapshot: BackendLocalPlanningSnapshot = + serde_json::from_value(snapshot).unwrap(); + let (mut request, env) = snapshot.planning_request().unwrap(); + request.queries[0].window_implementations.retain(|c| { + matches!( + c.layout, + asap_types::WindowMaterializationLayout::FullWindow + ) == full + }); + let plan = PhysicalCompiler.compile(request, env).unwrap(); + let config = &plan.precompute_plan.materializations[0]; + let manager = WindowManager::with_layout( + config.window_size, + config.slide_interval, + config.pane_origin_ms, + &config.window_layout, + ); + let mut buckets = BTreeMap::<(u64, u64), f64>::new(); + for second in 1..=800 { + for start in manager.stored_bucket_starts(second * 1_000 - 1) { + let (_, end) = manager.stored_bucket_bounds(start); + if start >= 0 && end <= 800_000 { + *buckets.entry((start as u64, end as u64)).or_default() += + second as f64; + } + } + } + let idx = SketchStore::new(); + idx.register(SketchInstanceMetadata { + sid: 7, + metric_name: "a".into(), + group_by_keys: Default::default(), + capability: Some(Capability::ExactAgg(asap_types::AggregationType::Sum)), + agg_kind: AggKind::ExactAgg { + agg_type: asap_types::AggregationType::Sum, + parameters_canonical: String::new(), + spatial_filter_canonical: String::new(), + }, + accuracy: None, + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: config.policy_fingerprint(), + }); + for (bounds, sum) in buckets { + idx.append_precompute( + 7, + BTreeMap::new(), + bounds, + Box::new( + crate::precompute_engine::operators::SumAccumulator::with_sum(sum), + ), + ); + } + let entry = plan.query_plan.entries.values().next().unwrap(); + for tick in 3..6 { + let end = phase_ms + evaluation_secs * 1_000 * tick; + let (outcome, _) = super::super::live_serve::serve_instant_from_query_plan( + &idx, entry, end, + ) + .unwrap_or_else(|error| { + panic!("E={evaluation_secs} phase={phase_ms} full={full}: {error:?}") + }); + let expected = ((end / 1_000 - 59)..=end / 1_000).sum::() as f64; + assert_eq!(outcome.series[0].1.last().unwrap().1, expected); + assert!(super::super::live_serve::serve_instant_from_query_plan( + &idx, + entry, + end + 1 + ) + .is_err()); + } + } + } + } + } + // Compile the two readouts, store one pane series, and execute the actual ratio. #[test] fn compiled_shared_sum_panes_preserve_each_lookback() { @@ -1262,6 +1362,7 @@ mod tests { asap_types::query_plan::QueryNodeId(1), QueryPlanNode::ReadMaterialization { binding: asap_types::query_plan::MaterializationBinding { + full_window_slide_ms: None, item_labels: Vec::new(), materialization: policy.into(), output_grouping: asap_types::query_plan::PhysicalGrouping::PerEntity, @@ -1358,6 +1459,7 @@ mod tests { asap_types::query_plan::QueryNodeId(1), QueryPlanNode::ReadMaterialization { binding: asap_types::query_plan::MaterializationBinding { + full_window_slide_ms: None, item_labels: Vec::new(), materialization: policy.into(), output_grouping: asap_types::query_plan::PhysicalGrouping::PerEntity, diff --git a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs index e03e702d2..237a691ae 100644 --- a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs +++ b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs @@ -405,28 +405,19 @@ impl SummaryValue { } } +#[cfg(test)] fn validate_binding_phase( binding: &asap_types::query_plan::MaterializationBinding, evaluation_ms: u64, ) -> Result<(), SummaryExecutorError> { - if i64::try_from(binding.window_ms).is_err() { - return Err(SummaryExecutorError::Unsupported( - "materialized pane width exceeds runtime timestamp range", - )); - } - planner_types::post_asap::validate_pane_coverage( - &planner_types::post_asap::PanePhaseBinding { - pane_width_ms: binding.window_ms, - pane_origin_ms: binding.pane_origin_ms, - }, - i64::try_from(evaluation_ms).ok(), - &planner_types::post_asap::BoundaryCoverage::PaneAligned, - ) - .map_err(|_| { - SummaryExecutorError::Unsupported( - "query evaluation phase does not match materialized pane origin", - ) - }) + let start = evaluation_ms.checked_sub(binding.readout_lookback_ms.unwrap_or(binding.window_ms)); + if start.is_some_and(|start| binding.covers_range(start, evaluation_ms)) { + Ok(()) + } else { + Err(SummaryExecutorError::Unsupported( + "query range does not match materialized window boundaries", + )) + } } impl QueryExecutionContext<'_> { @@ -456,7 +447,11 @@ impl QueryExecutionContext<'_> { "materialization population has unpublished input", )); } - validate_binding_phase(binding, self.t1_ms)?; + if !binding.covers_range(self.t0_ms, self.t1_ms) { + return Err(SummaryExecutorError::Unsupported( + "query range does not match materialized window boundaries", + )); + } enum Candidate { Sketch(DeltaSketchKind), @@ -529,7 +524,7 @@ impl QueryExecutionContext<'_> { matched_metadata += 1; match candidate { Candidate::Sketch(kind) => { - let Some(series) = self + let Some(mut series) = self .index .query_range(sid, self.t0_ms, self.t1_ms) .into_iter() @@ -538,6 +533,14 @@ impl QueryExecutionContext<'_> { check_panes(Vec::new())?; continue; }; + if binding.full_window_slide_ms.is_some() { + // Overlap lookup also returns neighboring complete windows. + // They overlap the answer and must never be merged into it. + series.samples.retain(|end, _| *end == self.t1_ms as i64); + if series.samples.is_empty() { + return Err(SummaryExecutorError::NoCandidates); + } + } check_panes(series.samples.keys().copied().collect())?; let key = match &binding.output_grouping { PhysicalGrouping::PerEntity => series.series_label_values.clone(), @@ -569,10 +572,12 @@ impl QueryExecutionContext<'_> { )); } } - if matches!( - agg_type, - AggregationType::MinMax | AggregationType::MultipleMinMax - ) { + if binding.full_window_slide_ms.is_none() + && matches!( + agg_type, + AggregationType::MinMax | AggregationType::MultipleMinMax + ) + { if let Some(series) = self.index.query_rollup_range( crate::storage_engines::sketch_db::index::RollupReduction::Max, sid, @@ -1394,6 +1399,7 @@ mod tests { #[test] fn pane_only_reads_require_the_planned_evaluation_phase() { let binding = asap_types::query_plan::MaterializationBinding { + full_window_slide_ms: None, item_labels: Vec::new(), materialization: asap_types::PolicyFingerprint(7).into(), output_grouping: asap_types::query_plan::PhysicalGrouping::PerEntity, @@ -1405,6 +1411,7 @@ mod tests { assert!(validate_binding_phase(&binding, 68_000).is_err()); let legacy = asap_types::query_plan::MaterializationBinding { + full_window_slide_ms: None, item_labels: Vec::new(), pane_origin_ms: None, ..binding @@ -1796,6 +1803,57 @@ mod tests { const T0: u64 = 1_000_000; const T1: u64 = 2_000_000; + // Neighboring overlapping complete windows are not additional answer panes. + #[test] + fn full_window_sketch_read_excludes_neighboring_windows() { + use asap_types::query_plan::{MaterializationBinding, PhysicalGrouping}; + let index = SketchStore::new(); + let fp = asap_types::PolicyFingerprint(703); + let mut metadata = kll_meta(1, "m", &[]); + metadata.policy_fp = fp; + index.register(metadata); + for (start, values) in [ + (0, vec![1_000.0, 2_000.0, 3_000.0]), + (20_000, vec![10.0, 20.0, 30.0]), + (40_000, vec![1_000.0, 2_000.0, 3_000.0]), + ] { + index.append_sample( + 1, + BTreeMap::new(), + (start, start + 60_000), + SketchSampleState { + bytes: encode_kll_items_proto(200, &values), + encoding: crate::storage_engines::sketch_db::index::SketchEncoding::ProtoFull, + }, + ); + } + let context = QueryExecutionContext { + index: &index, + t0_ms: 20_000, + t1_ms: 80_000, + is_cumulative: true, + allowed_materializations: Some(BTreeSet::from([fp])), + }; + let binding = MaterializationBinding { + full_window_slide_ms: Some(20_000), + materialization: fp.into(), + output_grouping: PhysicalGrouping::PerEntity, + item_labels: vec![], + window_ms: 60_000, + pane_origin_ms: Some(0), + readout_lookback_ms: Some(60_000), + }; + let states = context.read_bound_materialization(&binding).unwrap(); + let SummaryValue::Points(points, coverage) = context + .readout_bound(&states[0].1, &SketchQuery::Quantile { q: 0.5 }) + .unwrap() + else { + panic!("expected points"); + }; + assert_eq!(points, vec![(80_000, 20.0)]); + assert_eq!(coverage, Some((80_000, 80_000))); + } + /// One installed frequency summary merges panes before all four readouts. #[test] fn bound_univmon_merges_panes_for_four_readouts() { @@ -1843,6 +1901,7 @@ mod tests { allowed_materializations: Some(BTreeSet::from([fp])), }; let binding = MaterializationBinding { + full_window_slide_ms: None, materialization: fp.into(), output_grouping: PhysicalGrouping::PerEntity, item_labels: vec![], diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index 6d491f7f0..37ba2b0ff 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -789,19 +789,11 @@ async fn run_shared_dashboard(multi_pane: bool) { for entry in typed.query_workload.repeating_queries.as_mut().unwrap() { entry.time_selection.lookback = Some(planner_types::workload::DurationMs(10_000)); } - let (request, _) = typed.clone().planning_request().unwrap(); - for query in request.queries { - let mut candidates = query.window_implementations; - let mut small = candidates[0].clone(); - small.implementation_id = "five-second-pane".into(); - small.layout = asap_types::WindowMaterializationLayout::Pane { pane_secs: 5 }; - small.cost.weighted_cost = 0.0; - candidates[0].cost.weighted_cost = 10.0; - candidates.push(small); - typed - .implementation - .window_candidates - .insert(query.query_string, candidates); + for entry in typed.query_workload.repeating_queries.as_mut().unwrap() { + entry.demand = planner_types::workload::RepeatedDemand::FixedIntervalAt { + interval: planner_types::workload::RepetitionInterval(5_000), + evaluation_phase: planner_types::workload::TimestampMs(0), + }; } } let (request, environment) = typed.clone().planning_request().unwrap(); @@ -847,10 +839,9 @@ async fn run_shared_dashboard(multi_pane: bool) { assert_eq!(plan.precompute_plan.materializations.len(), 1); assert_eq!(plan.query_plan.entries.len(), 3); if multi_pane { - assert_eq!( - plan.lifecycle_estimates[0].window_implementation_id, - "five-second-pane" - ); + assert!(plan.lifecycle_estimates[0] + .window_implementation_id + .contains("pane")); for entry in plan.query_plan.entries.values() { assert_eq!(entry.instant.lookback_ms, 10_000); assert_eq!(entry.materialization_bindings()[0].window_ms, 5_000); diff --git a/data_plane/tests/backend_process_e2e.rs b/data_plane/tests/backend_process_e2e.rs index 00c427ab9..cd0f0e86a 100644 --- a/data_plane/tests/backend_process_e2e.rs +++ b/data_plane/tests/backend_process_e2e.rs @@ -484,9 +484,9 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { "window_secs": 1, "group_by": ["service"], "accuracy": {"Epsilon": 0.01}, - "window_implementations": [{ - "implementation_id": "collector-tumbling-v1", "framework": "tumbling", - "window_secs": 1, "slide_secs": 1, "layout": {"kind": "pane", "pane_secs": 1}, + "evaluation_phase_ms": 0, + "window_cost_model": { + "implementation_id": "collector-tumbling-v1", "cost": { "model_version": "process-e2e-v1", "workload_fingerprint": "shared-quantiles", "observed_at_unix_ms": observed_at_ms, "valid_for_ms": 60000, @@ -494,7 +494,7 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { "peak_memory_bytes": 4096, "network_bytes": 1024, "storage_bytes": 2048, "source_scan_bytes": 0 } - }], + }, "lifecycle": { "evaluation_interval_ms": 1000, "ingestion_rate_per_second": 100.0, diff --git a/data_plane/tests/support/physical_fixture.rs b/data_plane/tests/support/physical_fixture.rs index e896e9eee..b06e8bd2b 100644 --- a/data_plane/tests/support/physical_fixture.rs +++ b/data_plane/tests/support/physical_fixture.rs @@ -118,6 +118,7 @@ pub fn artifact(config: &StreamingConfig) -> PhysicalPlanInstallRequest { QueryNodeId(0), QueryPlanNode::ReadMaterialization { binding: MaterializationBinding { + full_window_slide_ms: None, materialization: config.policy_fingerprint().into(), output_grouping, item_labels: config.aggregated_labels.labels.clone(), diff --git a/docs/developer_docs/control-plane/physical-compiler.md b/docs/developer_docs/control-plane/physical-compiler.md index 97fdd621d..7bccd4a4c 100644 --- a/docs/developer_docs/control-plane/physical-compiler.md +++ b/docs/developer_docs/control-plane/physical-compiler.md @@ -95,7 +95,13 @@ pub struct TopKMembershipEvidence { Why this interface exists: it prevents adapters, protocols, and physical planning from each implementing their own query-to-summary mapping. -Each `WindowImplementationCandidate` carries a backend-owned implementation +Snapshot and HTTP adapters provide the same `WindowCostModel`. After logical +selection, one generator enumerates layouts allowed by each state's maintenance +requirements and deployment target, then derives or matches layout-specific costs. +See [repeated window planning](../planning/repeated-dashboard-panes.md) for inputs, +migration and cadence examples. + +Each internally generated `WindowImplementationCandidate` carries a backend-owned implementation identity, its Planner `SummaryWindowFramework`, concrete window/pane/state layout, and versioned workload-specific CPU, peak-memory, network, storage, scan, and calibrated weighted-cost evidence. The compiler collapses several diff --git a/docs/developer_docs/control-plane/plan-publication.md b/docs/developer_docs/control-plane/plan-publication.md index 203f29c11..dfae73481 100644 --- a/docs/developer_docs/control-plane/plan-publication.md +++ b/docs/developer_docs/control-plane/plan-publication.md @@ -12,7 +12,7 @@ plans, plus one target-specific `CollectorPlan` per Collector. - `queries`: query ID, PromQL, metric, window seconds, grouping labels, a typed `AccuracyTarget`, and lifecycle evidence (evaluation interval, ingestion rate/freshness, optimization horizon, and primitive state costs), - plus executor-feasible `window_implementations` with versioned, + plus `window_cost_model` and `evaluation_phase_ms` with versioned, workload-scoped physical cost evidence; - `collector_ids`: the required OpAMP agent IDs; - `capability_snapshot_id` and the exact `planner_revision`; diff --git a/docs/developer_docs/planning/repeated-dashboard-panes.md b/docs/developer_docs/planning/repeated-dashboard-panes.md index d3a7d328a..69bef94c7 100644 --- a/docs/developer_docs/planning/repeated-dashboard-panes.md +++ b/docs/developer_docs/planning/repeated-dashboard-panes.md @@ -1,67 +1,64 @@ -# Repeated dashboard pane selection +# Repeated dashboard window planning -The backend-local planning snapshot accepts `implementation.window_candidates`, -a map from the original registered PromQL text to a list of -`WindowImplementationCandidate` objects. Omitting a query keeps the existing -single lookback-sized candidate. Unknown query keys and empty candidate lists -are rejected. Each candidate supplies its own implementation ID, Tumbling -framework, query `window_secs`, `pane_secs`, state layout, and complete fresh -`ImplementationCostEvidence`. Pane sizes must be positive divisors of lookback. -Distributed Collector deployments still require pane size equal to lookback. +Snapshot and HTTP planning use one `window_cost_model` input with an +`implementation_id`, an implementation-cost metadata template (`cost`), and +optional measured `quotes`. HTTP queries also declare `evaluation_phase_ms`; +snapshots obtain it from the canonical workload's `fixed_interval_at` demand. +The former snapshot `window_candidates` and HTTP `window_implementations` inputs +are removed. There are no compatibility aliases. The snapshot's former +`window_implementation_id` and `implementation_cost` move under the model as +`implementation_id` and `cost` respectively. -For example, a 60-second query may offer 10-, 20-, and 60-second panes, with -separate provider quotes for each. The cost model returns the concrete ID in -Planner's `CompleteSummaryCandidateEstimate`; the compiler installs the ID -returned by Planner. It does not choose another size after planning. Actual -pane width enters the state fingerprint, precompute configuration, and query -binding; the query's 60-second lookback remains unchanged. Shared DAG consumers -must agree on the physical deployment contract. Distinct logical-window cohorts -that collide only after selecting a smaller shared pane are rejected until -joint lifecycle evidence is available for the resulting physical state. The -compiler never keeps only the first cohort's consumer count or cost quote. -Sharing within one already-priced logical cohort remains supported. +After logical selection, the compiler collects each maintained state's window +and executor constraints. It generates feasible layouts before pricing them. +Raw states offer `Pane { gcd(W,E) }` and, when distinct and supported, complete +windows advancing every E. A derived maintenance cohort and its sources offer +only the full, nonoverlapping cohort supported by that executor. Queries keep +their original W, E and evaluation phase regardless of storage representation. +Each state selects only candidates meeting its own requirements. -The original QueryWorkload, including recurrence, time scope, predictability, -requirements and data evidence, reaches lifecycle planning. Shared consumers' -reads are combined without multiplying state updates. Legacy direct requests -without a QueryWorkload retain their previous synthesized demand. Refresh cadence -alone does not prove evaluation phase alignment, so it is not a configuration -rejection condition. Candidate costs must account for the supplied workload; -the backend does not manufacture measured costs or an automatic pane-size cost -formula. Version 2 still requires complete workload-versus-exact quotes. +Measured quotes carry their workload fingerprint and exact window, slide and +layout. They replace the corresponding generated quote, or offer another legal +pane width, without bypassing capability checks. Serialization does not confer +compiler provenance: incoming quotes are always measured/provider evidence. +Unquoted layouts use supplied lifecycle unit costs multiplied by structural +counts. Layout changes never inherit another layout's measured scalar cost. +Workload-versus-exact deployment evidence is still required by snapshot version 2. -## Execution guarantees and limits +## Cases -Backend-local summary plans encode `promql_right_closed: true` in state -parameters and therefore in the fingerprint. Precompute workers assign boundary -samples to PromQL's `(start, end]` panes, while preserving original timestamps -inside accumulators. Legacy half-open states have a different identity. +All ranges below use PromQL's `(start,end]` convention. W is the lookback and E +is the query evaluation interval, both in seconds. Origins are zero unless stated. -At every instant/range evaluation, serving checks each binding's pane alignment. -Multi-pane reads require contiguous stored pane ends for every matched stored -series before merging state. Partial, missing/open, or interior missing panes -fail closed to exact fallback. Without explicit empty-pane completion evidence, -a sparse interval is conservatively a fallback rather than an assumed zero. -Whole panes outside the current lookback are excluded even when retained in -storage; this is logical window expiration, not a claim of immediate physical -state reclamation. Normal retention remains responsible for reclaiming state. +| Case | Query demand | Storage and behavior | +|---|---|---| +| 1. E divides W | W=60, E=20 | 20s panes: at 60 read `(0,20]+(20,40]+(40,60]`; at 80 read `(20,40]+(40,60]+(60,80]`. Complete windows are a separately priced alternative. | +| 2. E does not divide W | W=60, E=45 | 15s panes: at 90 read `(30,45]+(45,60]+(60,75]+(75,90]`. A 60s tumbling fallback cannot answer this boundary. | +| 3. E equals W | W=60, E=60 | One 60s pane per read: `(0,60]`, then `(60,120]`. | +| 4. E is a larger multiple | W=60, E=120 | 60s panes answer `(60,120]` and `(180,240]`. Query cadence stays 120s. Some continuously stored panes are unused. | +| 5. E is larger, not a multiple | W=60, E=90 | 30s panes answer `(30,60]+(60,90]`, then `(120,150]+(150,180]`. Complete windows may alternatively skip the gaps. | +| 6. Shifted demand | W=60, E=20, phase=5 | At 65 read `(5,25]+(25,45]+(45,65]`. Phase is part of the physical state identity. | +| 7. Ad hoc off-grid read | W=60, 20s epoch panes, evaluation=67 | `(7,67]` cannot be tiled; serving returns a capability miss for exact fallback. It never rounds the requested interval. | +| 8. Cross-query reuse | A: W=60/E=20; B: W=90/E=30 | Compare independent 20s/30s panes with one 10s producer. A merges 6 instead of 3 panes, B merges 9 instead of 3. Share only when the saved maintenance outweighs extra build, retention and read costs. | +| 9. Pane larger than E | W=60, E=20, proposed pane=30 | Illegal: it answers at 60 but cannot tile `(20,80]`. Hierarchical rollups remain unsupported and are rejected. | +| 10. Executor restrictions | Collector or derived cohort | Collectors exclude partial-window panes and unsupported sparse full-window schedules. Derived maintenance keeps its required full cohorts. Unsupported backend-local leaves or off-grid reads use exact fallback; an unsupported strict Collector plan is rejected. | +| 11. Measured costs | 20s pane costs 8; complete window costs 12 | Select the cheaper feasible quote. Rewriting into a shared 10s producer cannot reuse the measured cost 8. Explicitly quoted selected producers are protected from automatic repricing. | -This change does not add operators or min/max-specific paths. Existing operator, -source, and predicate limitations continue to apply. +Runtime layouts currently have whole-second precision. Fractional cadences are +not rounded: automatic backend-local planning leaves them to exact execution. +The integer cases above preserve E even when E exceeds W. -## Reproduction +## Serving and verification -From the repository root: +Bindings distinguish disjoint pane width from complete-window slide. Pane reads +require both boundaries on the pane grid. Complete-window reads require exactly +one window of width W, starting on its slide grid; the compiler converts the +query's end phase to the corresponding start phase. Sketch overlap lookup is +restricted to that window's end, so neighboring overlapping states are not merged. +Missing/open panes and unaligned boundaries remain capability misses. Sparse +series are not assumed to contain empty zero-valued panes without evidence. -```sh -cargo test -p control_plane --lib -cargo test -p data_plane --lib -- --test-threads=1 -cargo test -p data_plane --test asapquery_compatibility_process_e2e -- --test-threads=1 -``` - -The conformance tests change candidate costs and assert the selected ID and -installed pane width; execute advancing stored-state queries; reject missing and -partial panes; and run production RemoteWrite, planning, installation, SUM, -COUNT, and a ratio DAG with 5-second panes and 10-second lookbacks. Boundary -samples verify left exclusion and right inclusion. These are synthetic -correctness tests, not an o11ybench performance or benefit report. +Tests cover the five cadence relations, shifted phases, runtime bucket assignment, +raw-sample sums, off-grid misses, costed sharing, measured quotes and target +capabilities. The process dashboard test derives 5s panes from a 5s cadence for +10s lookbacks; it no longer injects a fabricated zero-cost candidate. diff --git a/docs/developer_docs/planning/shared-window-panes.md b/docs/developer_docs/planning/shared-window-panes.md index 15eea28ba..d2e0fe0b6 100644 --- a/docs/developer_docs/planning/shared-window-panes.md +++ b/docs/developer_docs/planning/shared-window-panes.md @@ -1,47 +1,38 @@ # Shared additive window panes -The fallback window provider prices each query range. After logical selection, -raw inputs of derived maintenance programs use full, non-overlapping windows, -matching the current maintenance executor. Explicit `window_candidates` remain -authoritative and are not rewritten. - -For backend-local raw SUM states, the backend offers compatible selected pane -producers to ASAPPlanner's `pane_sharing::select_shared_panes`. The compatibility -key preserves source, population predicate, projection, accumulator parameters, -grouping, partitioning, pane width/origin, runtime policy and lifecycle evidence. -Planner compares independent producers with one producer retained for the longest -lookback, charging every readout. Only beneficial groups are installed. - -The shared state's physical window is one pane. Logical readout windows stay on -their original QueryPlan bindings. Retention uses the largest bound lookback. -For `sum_over_time(a[1m]) / sum_over_time(a[10m])` evaluated every minute, one -60-second producer retains 11 published states; readouts merge one and ten panes. -Different metrics, predicates or phases remain separate. Derived inputs, explicit -window quotes, non-additive state and independently selected FullWindow layouts -are not rewritten. This pass reuses compatible selected panes; it does not search -all possible pane widths or re-optimize FullWindow choices jointly. - -Layout pricing charges both published retained states and open worker -accumulators. A full window of width W and step S keeps ceil(W/S) open states; -a pane producer keeps one. These residency costs are separate from update CPU. -Nonfinite arithmetic remains invalid evidence rather than becoming a zero quote. - -## Cross-repository validation - -The backend pins Planner commit `ca7546de792d74aee8231e9a1100ca893d9e86d3`, which provides -`pane_sharing::select_shared_panes`. Normal builds use the Git dependency. -For coordinated local development, the optional validation script exports only -the optimizer crate while retaining the pinned IR/frontend revision and restores -Cargo.lock after the run. - -```sh -python3 tools/test_shared_panes.py --planner /path/to/ASAPPlanner -- \ - test -p control_plane -python3 tools/test_shared_panes.py --planner /path/to/ASAPPlanner \ - --sketchlib /path/to/compatible/asap_sketchlib -- \ - test -p data_plane --lib compiled_shared_sum_panes_preserve_each_lookback -``` - -Use `--toolchain 1.98.0` if needed in the development environment. The data-plane -checkout requires sketchlib's standard-update guard and interpolated quantile -interfaces; validation used revision `8c03d7c` for those existing dependencies. +The compiler generates and prices layouts from selected state requirements. +See [window planning](repeated-dashboard-panes.md) for the input contract and all +cadence examples. + +After individual layout selection, backend-local raw SUM producers may share a +common-divisor pane. Compatibility preserves source, population predicate, +projection, accumulator parameters, grouping, partitioning, runtime policy, +accuracy and lifecycle unit-cost evidence. Evaluation intervals may differ; +each consumer's reads are charged at its own interval. Origins must agree modulo +the proposed common pane width. + +The optimizer compares the original producers with one producer at the common +pane width, retained for the longest lookback. It recalculates build, maintenance, +retention, retirement and read costs for the new layout. Only a finite, strictly +cheaper group is installed. Derived maintenance inputs, explicitly quoted selected +layouts, non-additive states and independently selected complete-window layouts +are not rewritten. This is a bounded comparison of selected panes, not exhaustive +search over all layouts or hierarchical rollups. + +The shared state's physical window is one pane. Logical readout windows remain +on the QueryPlan bindings. For `sum_over_time(a[1m]) / sum_over_time(a[10m])` +evaluated every minute, one 60s producer retains 11 published states, while its +readouts merge one and ten panes. For separate 60s/20s and 90s/30s demands, a +10s shared producer may win when maintenance is expensive; independent producers +remain when the extra reads or pane creation cost more. + +Costs include published retained states and open worker accumulators. Complete +windows have at most `ceil(W/E)` open states; their rate-model average update +fanout is `W/E`, including sparse schedules. Pane producers have one active state. +These are supplied unit costs and structural estimates, not measurements. + +If the entire compatibility group cannot share, the optimizer selects profitable +phase-compatible pairs, ordered by savings, and retries the remaining members. +An incompatible or expensive fine-cadence consumer therefore does not block reuse +between other consumers. This deterministic greedy selection does not claim a +globally optimal partition of all workload states. diff --git a/docs/examples/asapquery-compatibility-demo-snapshot.json b/docs/examples/asapquery-compatibility-demo-snapshot.json index 8641f93bd..87ecd811f 100644 --- a/docs/examples/asapquery-compatibility-demo-snapshot.json +++ b/docs/examples/asapquery-compatibility-demo-snapshot.json @@ -82,19 +82,21 @@ "evidence_observed_at_unix_ms": 9500, "evidence_valid_for_ms": 60000, "horizon_seconds": 300.0, - "window_implementation_id": "backend-tumbling-v1", - "implementation_cost": { - "model_version": "compat-cost-v1", - "workload_fingerprint": "asapquery-compatibility-demo", - "observed_at_unix_ms": 9500, - "valid_for_ms": 60000, - "horizon_seconds": 300.0, - "cpu_cost": 1.0, - "peak_memory_bytes": 4096, - "network_bytes": 0, - "storage_bytes": 2048, - "source_scan_bytes": 0, - "weighted_cost": 1.0 + "window_cost_model": { + "implementation_id": "backend-tumbling-v1", + "cost": { + "model_version": "compat-cost-v1", + "workload_fingerprint": "asapquery-compatibility-demo", + "observed_at_unix_ms": 9500, + "valid_for_ms": 60000, + "horizon_seconds": 300.0, + "cpu_cost": 1.0, + "peak_memory_bytes": 4096, + "network_bytes": 0, + "storage_bytes": 2048, + "source_scan_bytes": 0, + "weighted_cost": 1.0 + } }, "topk_evidence": { "topk(1, sum_over_time(asap_demo_gauge[5s]))": { diff --git a/docs/examples/asapquery-planning-snapshot.json b/docs/examples/asapquery-planning-snapshot.json index cd63e8bc8..b4147ae68 100644 --- a/docs/examples/asapquery-planning-snapshot.json +++ b/docs/examples/asapquery-planning-snapshot.json @@ -49,19 +49,21 @@ "evidence_observed_at_unix_ms": 9500, "evidence_valid_for_ms": 60000, "horizon_seconds": 300.0, - "window_implementation_id": "backend-tumbling-v1", - "implementation_cost": { - "model_version": "compat-cost-v1", - "workload_fingerprint": "replaced-with-canonical-promql", - "observed_at_unix_ms": 9500, - "valid_for_ms": 60000, - "horizon_seconds": 300.0, - "cpu_cost": 1.0, - "peak_memory_bytes": 1024, - "network_bytes": 0, - "storage_bytes": 512, - "source_scan_bytes": 0, - "weighted_cost": 1.0 + "window_cost_model": { + "implementation_id": "backend-tumbling-v1", + "cost": { + "model_version": "compat-cost-v1", + "workload_fingerprint": "replaced-with-canonical-promql", + "observed_at_unix_ms": 9500, + "valid_for_ms": 60000, + "horizon_seconds": 300.0, + "cpu_cost": 1.0, + "peak_memory_bytes": 1024, + "network_bytes": 0, + "storage_bytes": 512, + "source_scan_bytes": 0, + "weighted_cost": 1.0 + } } }, "environment": { diff --git a/tools/o11y-execution/discover_snapshot.py b/tools/o11y-execution/discover_snapshot.py index 7c640065e..ad72510c5 100644 --- a/tools/o11y-execution/discover_snapshot.py +++ b/tools/o11y-execution/discover_snapshot.py @@ -96,7 +96,7 @@ def evidence(value): implementation["query_staleness_margin_ms"] = max(0, (args.repetitions - 1) * args.interval_ms) implementation.update(evidence_observed_at_unix_ms=now, evidence_valid_for_ms=86400000, horizon_seconds=horizon) implementation["lifecycle_costs"] = dict.fromkeys(("build", "maintenance_per_update", "read", "retention_per_second", "retirement"), 1.0) - implementation["implementation_cost"].update(model_version="UNCALIBRATED-enumeration-only", observed_at_unix_ms=now, + implementation["window_cost_model"]["cost"].update(model_version="UNCALIBRATED-enumeration-only", observed_at_unix_ms=now, valid_for_ms=86400000, horizon_seconds=horizon, cpu_cost=1.0, peak_memory_bytes=0, network_bytes=0, storage_bytes=0, source_scan_bytes=0, weighted_cost=1.0) # Plan lifecycle is control-plane wall time. Historical event timestamps diff --git a/tools/o11y-execution/test_calibrate.py b/tools/o11y-execution/test_calibrate.py index bd7350b14..b98a77fbd 100644 --- a/tools/o11y-execution/test_calibrate.py +++ b/tools/o11y-execution/test_calibrate.py @@ -67,11 +67,11 @@ def test_shared_profile_preserves_measured_cpu_units(self): # Inclusive measured phases determine the coarse model; input count only normalizes updates. from update_global_profile import update self.measurements["candidates"][0]["resources"] = {"peak_memory_bytes": 123, "storage_bytes": 456} - snapshot = {"implementation": {"horizon_seconds": 60, "implementation_cost": {}}, + snapshot = {"implementation": {"horizon_seconds": 60, "window_cost_model": {"cost": {}}}, "workload_cost_evidence": {"old": True}} result, audit = update(snapshot, self.measurements, 20) self.assertEqual(result["implementation"]["lifecycle_costs"]["maintenance_per_update"], 0.5) - self.assertEqual(result["implementation"]["implementation_cost"]["cpu_cost"], 40) + self.assertEqual(result["implementation"]["window_cost_model"]["cost"]["cpu_cost"], 40) self.assertNotIn("workload_cost_evidence", result) self.assertIn("not measured zero", " ".join(audit["limitations"])) diff --git a/tools/o11y-execution/update_global_profile.py b/tools/o11y-execution/update_global_profile.py index 547b1ae49..c55fe1e0a 100644 --- a/tools/o11y-execution/update_global_profile.py +++ b/tools/o11y-execution/update_global_profile.py @@ -34,7 +34,7 @@ def phase(name): cpu = sum(phase(p) for p in ("install", "ingest_and_build", "residency", "retirement")) impl = snapshot["implementation"] impl["lifecycle_costs"] = costs - impl["implementation_cost"].update(model_version="measured-inclusive-coarse-cpu-only-v1", cpu_cost=cpu, + impl["window_cost_model"]["cost"].update(model_version="measured-inclusive-coarse-cpu-only-v1", cpu_cost=cpu, weighted_cost=cpu, peak_memory_bytes=int(max(nonnegative(r["resources"]["peak_memory_bytes"], "peak memory") for r in rows)), storage_bytes=int(max(nonnegative(r["resources"]["storage_bytes"], "storage") for r in rows)), network_bytes=0, source_scan_bytes=0) From 4795ff8411969c24f5d13bd0f6bc92b9f6011ab3 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 11:11:35 -0600 Subject: [PATCH 2/3] Align process fixtures with declared evaluation populations --- data_plane/tests/asapquery_compatibility_process_e2e.rs | 6 ++++++ data_plane/tests/support/durable_summary_process.rs | 3 +++ data_plane/tests/support/univmon_erp_process.rs | 6 ++++++ 3 files changed, 15 insertions(+) diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index 37ba2b0ff..64307e6f2 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -1248,6 +1248,12 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() (base + 1_700, 60.0), (base + 2_900, 5.0), (base + 4_200, 15.0), + // Keep both queried windows populated; missing counter + // panes deliberately use exact fallback. + (base + 5_500, 20.0), + (base + 6_700, 30.0), + (base + 7_900, 5.0), + (base + 9_200, 15.0), ], ), series_with_labels( diff --git a/data_plane/tests/support/durable_summary_process.rs b/data_plane/tests/support/durable_summary_process.rs index 626392ae3..f8a704906 100644 --- a/data_plane/tests/support/durable_summary_process.rs +++ b/data_plane/tests/support/durable_summary_process.rs @@ -13,6 +13,9 @@ async fn persisted_summary_restarts_without_live_reregistration() { .to_owned(); fixture["query_workload"]["repeating_queries"] = serde_json::json!([fixture["query_workload"]["repeating_queries"][2].clone()]); + // This restart fixture persists one complete five-second population. + fixture["query_workload"]["repeating_queries"][0]["demand"]["fixed_interval_at"]["interval"] = + serde_json::json!(5000); let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_value(fixture).unwrap(); let plan = quote_snapshot_for_test(snapshot).compile().unwrap(); diff --git a/data_plane/tests/support/univmon_erp_process.rs b/data_plane/tests/support/univmon_erp_process.rs index 7c496d4ab..29b5d572e 100644 --- a/data_plane/tests/support/univmon_erp_process.rs +++ b/data_plane/tests/support/univmon_erp_process.rs @@ -108,6 +108,8 @@ async fn measured_readout_evidence_selects_and_executes_univmon() { .map(|query| { let mut entry = template.clone(); entry["query"] = (*query).into(); + // ERP evidence is calibrated for one complete five-second population. + entry["demand"]["fixed_interval_at"]["interval"] = serde_json::json!(5000); entry["requirements"]["accuracy"] = serde_json::json!({"explicit": {"Epsilon": 0.2}}); entry }) @@ -291,6 +293,10 @@ async fn measured_readout_evidence_selects_and_executes_univmon() { assert!(observed.invalid_reason.is_none(), "{observed:?}"); assert!(!observed.populations.is_empty()); assert_eq!(observed.window_end_ms - observed.window_start_ms, 5000); + for population in &observed.populations { + assert_eq!(population.shape.event_count(), Some(raw.len() as u64)); + assert_eq!(population.shape.sorted_counts.len(), 128); + } assert!(plan .summary_catalog .materializations From 1807b88c8d8a0b7cf509c31986ccac5582c4f5bd Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 11:20:14 -0600 Subject: [PATCH 3/3] Declare actual transport windows and derive fixture sample counts --- .../asapquery_compatibility_process_e2e.rs | 17 +++++++++++++++-- .../e2e_controller_plans_and_backend_serves.rs | 9 ++++++++- 2 files changed, 23 insertions(+), 3 deletions(-) diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index 64307e6f2..a50a3452c 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -1710,8 +1710,21 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() .await .expect("metrics body"); assert!(metrics.contains("asap_remote_write_requests_total 4")); - assert!(metrics.contains("asap_remote_write_samples_total 36")); - assert!(metrics.contains("asap_remote_write_duplicates_total 33")); + let samples = |request: &WriteRequest| { + request + .timeseries + .iter() + .map(|series| series.samples.len()) + .sum::() + }; + assert!(metrics.contains(&format!( + "asap_remote_write_samples_total {}", + samples(&request) + samples(&watermark_advance) + ))); + assert!(metrics.contains(&format!( + "asap_remote_write_duplicates_total {}", + samples(&request) + ))); assert!(metrics.contains("asap_remote_write_rejected_requests_total 1")); } diff --git a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs index caf5a21f4..04e086531 100644 --- a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs +++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs @@ -63,10 +63,17 @@ fn phase_aligned_now_ns() -> u64 { } async fn post_full_config(client: &reqwest::Client, stack: &FullStack, json: &JsonValue) { - let runtime = data_plane::storage_engines::types::StreamingConfig::from_yaml_data( + let mut runtime = data_plane::storage_engines::types::StreamingConfig::from_yaml_data( &serde_yaml::to_value(json).unwrap(), ) .unwrap(); + // The transport payloads below contain one-second states. The legacy + // streaming emitter's default window is not their physical layout. + for config in runtime.aggregation_configs.values_mut() { + config.window_size = 1; + config.slide_interval = 1; + config.window_layout = asap_types::WindowMaterializationLayout::Pane { pane_secs: 1 }; + } let mut artifact = physical_fixture::artifact(&runtime); if runtime .aggregation_configs