From aec203aea8e22005b7c86a3db7cf74f54981abe5 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 14:08:29 -0600 Subject: [PATCH 1/4] feat: select concrete dashboard panes and enforce window correctness --- Cargo.lock | 6 +- control_plane/Cargo.toml | 6 +- control_plane/src/emit/stage_config.rs | 5 + control_plane/src/main.rs | 1 + control_plane/src/physical/compiler.rs | 187 ++++++++++++++---- .../src/physical/post_asap/cost_model.rs | 39 ++-- control_plane/src/physical/workload_cost.rs | 18 ++ crates/asap_types/Cargo.toml | 2 +- data_plane/Cargo.toml | 2 +- data_plane/src/precompute_engine/worker.rs | 24 ++- .../asap_query_engine/post_asap_readout.rs | 105 ++++++++++ .../asap_query_engine/summary_exec.rs | 11 +- .../asap_query_engine/summary_executor.rs | 40 ++++ .../asapquery_compatibility_process_e2e.rs | 122 ++++++++++-- .../planning/repeated-dashboard-panes.md | 63 ++++++ 15 files changed, 554 insertions(+), 77 deletions(-) create mode 100644 docs/developer_docs/planning/repeated-dashboard-panes.md diff --git a/Cargo.lock b/Cargo.lock index 5f1fb4b7..e9332502 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -343,7 +343,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=c3410d14865497758d212e4265ad25c782187de1#c3410d14865497758d212e4265ad25c782187de1" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=abb2f20e27091ac0c60715dc42b9b59c75b3447c#abb2f20e27091ac0c60715dc42b9b59c75b3447c" dependencies = [ "asap-types", "serde", @@ -354,7 +354,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=c3410d14865497758d212e4265ad25c782187de1#c3410d14865497758d212e4265ad25c782187de1" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=abb2f20e27091ac0c60715dc42b9b59c75b3447c#abb2f20e27091ac0c60715dc42b9b59c75b3447c" dependencies = [ "asap-types", "promql-parser 0.10.0", @@ -374,7 +374,7 @@ dependencies = [ [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=c3410d14865497758d212e4265ad25c782187de1#c3410d14865497758d212e4265ad25c782187de1" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=abb2f20e27091ac0c60715dc42b9b59c75b3447c#abb2f20e27091ac0c60715dc42b9b59c75b3447c" dependencies = [ "serde", "serde_json", diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 52d7961c..b96327c7 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -93,8 +93,8 @@ asap_types.workspace = true # scaffolding, unaware that `data_plane`'s `summary_executor.rs` in *this* # repo is a real one. Vendored locally instead of chased upstream -- see # `data_plane/src/query_engines/asap_query_engine/summary_exec.rs`. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } -asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "abb2f20e27091ac0c60715dc42b9b59c75b3447c" } +asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "abb2f20e27091ac0c60715dc42b9b59c75b3447c" } # L1 adoption (design-target-architecture.md Part B): the PromQL front # end itself, replacing control_plane's own query_parser/promql.rs. @@ -102,7 +102,7 @@ asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = # `planner-types`/`asap-aware-mapping` above -- these three MUST move # together (two revs of the same upstream repo's types in one workspace # resolve to distinct Rust types that won't unify). -asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } +asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "abb2f20e27091ac0c60715dc42b9b59c75b3447c" } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index 6582b79b..5e9b62ad 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -2991,6 +2991,11 @@ pub(crate) fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonVa obj.insert("weight_mode".into(), JsonValue::String(mode.into())); } } + // PromQL range selectors are (start, end]. Encode the boundary convention + // in state identity so legacy half-open panes cannot satisfy this binding. + if matches!(agg.aggregation_input, AggregationInput::Raw) { + parameters["promql_right_closed"] = json!(true); + } let aggregation_input = match agg.aggregation_input { AggregationInput::SketchEnvelope => "sketch_envelope", AggregationInput::Raw => "raw", diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index be6dc44d..cf08aa43 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -826,6 +826,7 @@ fn compile_physical_plan_request( } let planning_request = physical::compiler::PlanningRequest { + query_workload: None, queries, evidence: request.evidence, planner_revision: request.planner_revision, diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index cbb2b40d..ed6136f6 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -42,7 +42,7 @@ use crate::query_plan::{ use crate::types_v2::AccuracyTarget; use planner_types::pre_asap::Source; -pub const PLANNER_REVISION: &str = "c3410d14865497758d212e4265ad25c782187de1"; +pub const PLANNER_REVISION: &str = "abb2f20e27091ac0c60715dc42b9b59c75b3447c"; #[derive(Debug, Clone)] pub struct PlanningQuery { @@ -126,6 +126,8 @@ pub struct LifecyclePlanningInput { #[derive(Debug, Clone, Default)] pub struct PlanningRequest { + /// Original dashboard demand, in the same order as queries. None is legacy input. + pub query_workload: Option, pub queries: Vec, pub evidence: HashMap, pub planner_revision: String, @@ -180,6 +182,9 @@ 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, @@ -1584,6 +1589,13 @@ 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(); @@ -1645,7 +1657,7 @@ impl BackendLocalPlanningSnapshot { } queries.push(PlanningQuery { query_id, - query_string, + query_string: query_string.clone(), post_asap, source: Source::TimeSeries { metric: metadata.metric_name, @@ -1654,20 +1666,28 @@ impl BackendLocalPlanningSnapshot { group_by: metadata.group_by_labels, accuracy, lifecycle, - window_implementations: vec![WindowImplementationCandidate { - implementation_id: self.implementation.window_implementation_id.clone(), - framework: SummaryWindowFramework::Tumbling, - window_secs: lookback_ms / 1_000, - pane_secs: lookback_ms / 1_000, - state_layout: self.implementation.state_layout.clone(), - cost, - }], + window_implementations: self + .implementation + .window_candidates + .get(&query_string) + .cloned() + .unwrap_or_else(|| { + vec![WindowImplementationCandidate { + implementation_id: self.implementation.window_implementation_id.clone(), + framework: SummaryWindowFramework::Tumbling, + window_secs: lookback_ms / 1_000, + pane_secs: lookback_ms / 1_000, + state_layout: self.implementation.state_layout.clone(), + cost, + }] + }), runtime_policy: RuntimeRulePolicy::default(), }); } select_workload_roots(&mut queries, canonical_roots, &topk_evidence_by_id)?; Ok(( PlanningRequest { + query_workload: Some(workload), queries, evidence: topk_evidence_by_id, planner_revision: PLANNER_REVISION.into(), @@ -1698,6 +1718,18 @@ impl PhysicalCompiler { }); } + if let Some(workload) = &request.query_workload { + let entries = workload.entries().collect::>(); + if entries.len() != request.queries.len() + || entries + .iter() + .zip(&request.queries) + .any(|(entry, query)| entry.query.0 != query.query_string) + { + return Err(CompileError::Snapshot("original workload and planning queries must have identical order and query text".into())); + } + } + let roots = request .queries .iter() @@ -1754,10 +1786,12 @@ impl PhysicalCompiler { SummaryMaintenanceCapabilities { incremental_update: true, merge: true, - delete: false, + // Expired whole panes are excluded when reading a moving scope. + delete: environment.target + == PhysicalDeploymentTarget::BackendLocalRemoteWrite, }, ) - .with_window_framework_costs(window_costs); + .with_window_implementation_costs(window_costs); if environment.target == PhysicalDeploymentTarget::DistributedCollectors && selected.iter().any(|state| { matches!( @@ -1797,7 +1831,7 @@ impl PhysicalCompiler { } _ => selected.algorithm.clone(), }; - let aggregation = physical_aggregation( + let mut aggregation = physical_aggregation( query, &selected, aggregation_id.clone(), @@ -1816,14 +1850,26 @@ impl PhysicalCompiler { &model, &environment, &state_consumers, + request.query_workload.as_ref().map(|workload| { + ( + workload, + consumers[&materialization].iter().copied().collect(), + ) + }), )?; let window_implementation = query.window_implementations.iter() - .filter(|candidate| candidate.framework == planner_selection.window_framework) - .min_by(|left, right| left.cost.weighted_cost.total_cmp(&right.cost.weighted_cost)) + .find(|candidate| candidate.implementation_id == planner_selection.window_implementation_id + && candidate.framework == planner_selection.window_framework) .ok_or_else(|| CompileError::Lifecycle { query_id: query.query_id.clone(), reason: "Planner selected a window framework without a retained concrete implementation".into(), })?; + // The logical consumer group is identified above. The installed state + // identity includes the actual pane width selected by Planner. + aggregation.window_secs = window_implementation.pane_secs; + let materialization = + backend_plan::aggregation_config_for_materialization(&aggregation)? + .policy_fingerprint(); lifecycle_estimates .entry(materialization) .or_insert_with(|| MaterializationLifecycleEstimate { @@ -1873,7 +1919,7 @@ impl PhysicalCompiler { algorithm: physical_algorithm, parameters: selected.parameters, group_by: query.group_by.clone(), - window_secs: query.window_secs, + window_secs: window_implementation.pane_secs, abstract_window_framework: planner_selection.window_framework.clone(), window_implementation_id: window_implementation.implementation_id.clone(), pane_secs: window_implementation.pane_secs, @@ -2037,7 +2083,9 @@ impl PhysicalCompiler { )) })?; if materialization.family != physical_materialization_family(node_family) - || materialization.window.size_ms != query.window_secs.saturating_mul(1_000) + || materialization.window.size_ms == 0 + || query.window_secs.saturating_mul(1_000) % materialization.window.size_ms + != 0 || materialization.group_by != query.group_by { return Err(crate::query_plan::QueryPlanError::Invalid(format!( @@ -2050,7 +2098,7 @@ impl PhysicalCompiler { metric: planned_metric, sid_grouping: query.group_by.clone(), output_grouping: PhysicalGrouping::Reduce(query.group_by.clone()), - window_ms: query.window_secs.saturating_mul(1_000), + window_ms: materialization.window.size_ms, }) }, )?; @@ -2280,9 +2328,9 @@ fn validate_lifecycle_input( fn validate_window_implementations( query: &PlanningQuery, environment: &DeploymentEnvironment, -) -> Result, CompileError> { +) -> Result, CompileError> { let mut ids = BTreeSet::new(); - let mut cheapest = BTreeMap::::new(); + let mut candidates = Vec::new(); for candidate in &query.window_implementations { let evidence = &candidate.cost; let age = environment @@ -2306,11 +2354,15 @@ fn validate_window_implementations( && candidate.pane_secs != 0 && candidate.pane_secs <= candidate.window_secs && candidate.window_secs % candidate.pane_secs == 0 - // Current Collector runtime contract is the MVP's anchored, - // tumbling implementation. Other Planner primitives become - // candidates only when an executor advertises full semantics. && candidate.framework == SummaryWindowFramework::Tumbling - && candidate.pane_secs == candidate.window_secs; + && match environment.target { + PhysicalDeploymentTarget::DistributedCollectors => { + candidate.pane_secs == candidate.window_secs + } + // Cadence does not establish phase alignment. Serving checks each + // actual interval and falls back when whole panes cannot cover it. + PhysicalDeploymentTarget::BackendLocalRemoteWrite => true, + }; if !valid { return Err(CompileError::Lifecycle { query_id: query.query_id.clone(), @@ -2320,24 +2372,23 @@ fn validate_window_implementations( ), }); } - cheapest - .entry(candidate.framework.clone()) - .and_modify(|cost| *cost = cost.min(evidence.weighted_cost)) - .or_insert(evidence.weighted_cost); + candidates.push(( + candidate.implementation_id.clone(), + candidate.framework.clone(), + Cost(evidence.weighted_cost), + )); } - if cheapest.is_empty() { + if candidates.is_empty() { return Err(CompileError::Lifecycle { query_id: query.query_id.clone(), reason: "no complete executor-feasible window implementation evidence".into(), }); } - Ok(cheapest - .into_iter() - .map(|(framework, cost)| (framework, Cost(cost))) - .collect()) + Ok(candidates) } struct PlannerPhysicalSelection { + window_implementation_id: String, lifecycle: CollectorLifecycle, window_framework: SummaryWindowFramework, expected_reads: f64, @@ -2351,6 +2402,7 @@ fn select_lifecycle( model: &ControlPlaneCostModel, environment: &DeploymentEnvironment, consumers: &[&PlanningQuery], + original_workload: Option<(&QueryWorkload, Vec)>, ) -> Result { // Current lifecycle evidence is per producer with one unit read cost. // Conflicting source/rate/horizon/cost snapshots cannot be averaged into @@ -2404,9 +2456,11 @@ fn select_lifecycle( ..DataWorkload::default() }), }; + let (workload, indices) = + original_workload.unwrap_or((&workload, (0..consumers.len()).collect())); let plan = plan_summary_maintenance_lifecycles( Rc::new(node.clone()), - WorkloadDemand::new(&workload, &(0..consumers.len()).collect::>()), + WorkloadDemand::new(workload, &indices), environment.observed_at_unix_ms, Some(Horizon(query.lifecycle.horizon_seconds)), SummaryMaintenanceLifecycleCapabilities { @@ -2438,6 +2492,12 @@ fn select_lifecycle( reason: "latest ASAPPlanner selected no window framework from the supplied physical evidence".into(), })?; Ok(PlannerPhysicalSelection { + window_implementation_id: plan.selected_physical_plan_id.clone().ok_or_else(|| { + CompileError::Lifecycle { + query_id: query.query_id.clone(), + reason: "Planner returned no concrete window implementation identity".into(), + } + })?, expected_reads: plan.expected_reads.ok_or_else(|| CompileError::Lifecycle { query_id: query.query_id.clone(), reason: "missing joint read demand".into(), @@ -2810,6 +2870,7 @@ mod tests { evidence_by_query.insert(query_id.to_string(), evidence); } Ok(PlanningRequest { + query_workload: None, queries: vec![PlanningQuery { query_id: query_id.into(), query_string: promql.into(), @@ -3342,6 +3403,7 @@ 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, @@ -3353,6 +3415,16 @@ mod tests { }, environment, }; + assert_eq!( + snapshot + .clone() + .planning_request() + .unwrap() + .0 + .query_workload + .as_ref(), + Some(&snapshot.query_workload) + ); let first = snapshot .clone() .compile() @@ -3580,6 +3652,53 @@ mod tests { )); } + // Concrete candidates with the same framework survive selection; changing + // their quoted costs changes installed state, not the query's lookback. + #[test] + fn tumbling_sizes_are_selected_by_cost_and_installed() { + for (small_cost, expected_secs, expected_id) in [(0.1, 10, "small"), (10.0, 60, "large")] { + let mut request = request("q", "sum_over_time(m[1m])"); + let query = &mut request.queries[0]; + let mut small = query.window_implementations[0].clone(); + small.implementation_id = "small".into(); + small.pane_secs = 10; + small.cost.weighted_cost = small_cost; + query.window_implementations[0].implementation_id = "large".into(); + query.window_implementations.push(small); + let mut env = environment(10_000); + env.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite; + env.collector_ids.clear(); + let bundle = PhysicalCompiler.compile(request, env).unwrap(); + let materialization = bundle + .backend_plan + .materializations + .values() + .next() + .unwrap(); + assert_eq!(materialization.window.size_ms, expected_secs * 1000); + let entry = bundle.query_plan.lookup("sum_over_time(m[1m])").unwrap(); + assert_eq!(entry.instant.lookback_ms, 60_000); + assert_eq!( + entry.materialization_bindings()[0].window_ms, + expected_secs * 1000 + ); + assert_eq!( + bundle.lifecycle_estimates[0].window_implementation_id, + expected_id + ); + } + } + + // Non-divisor panes cannot reconstruct a lookback from whole states. + #[test] + fn tumbling_sizes_reject_non_divisors() { + let mut request = request("q", "sum_over_time(m[1m])"); + request.queries[0].window_implementations[0].pane_secs = 7; + let mut env = environment(10_000); + env.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite; + assert!(PhysicalCompiler.compile(request, env).is_err()); + } + #[test] fn missing_window_implementation_evidence_fails_closed() { let mut request = request("q-window", "quantile_over_time(0.99, m[1m])"); diff --git a/control_plane/src/physical/post_asap/cost_model.rs b/control_plane/src/physical/post_asap/cost_model.rs index 9cdf0624..67ba378f 100644 --- a/control_plane/src/physical/post_asap/cost_model.rs +++ b/control_plane/src/physical/post_asap/cost_model.rs @@ -63,7 +63,7 @@ pub struct ControlPlaneCostModel { pub workload_accuracy: AccuracyTarget, lifecycle_costs: SummaryMaintenanceLifecycleCostInputs, summary_maintenance: SummaryMaintenanceCapabilities, - window_framework_costs: Vec<(SummaryWindowFramework, Cost)>, + window_framework_costs: Vec<(Option, SummaryWindowFramework, Cost)>, offline_evidence: Option, offline_frequency_comparison: Option<(OfflineComparisonEvidence, OfflineComparisonRequest)>, } @@ -213,15 +213,27 @@ impl ControlPlaneCostModel { costs.into_iter().map(|(algorithm, _)| algorithm).collect() } - /// Bind the cheapest complete, workload-scoped physical realization for - /// each Planner-owned abstract window framework. Concrete implementation - /// identities stay in the physical compiler; only framework and cost - /// cross into Planner's candidate comparison. + /// Keep concrete physical identities in Planner's complete-candidate estimate, + /// including distinct pane sizes using the same abstract window framework. + pub fn with_window_implementation_costs( + mut self, + costs: Vec<(String, SummaryWindowFramework, Cost)>, + ) -> Self { + self.window_framework_costs = costs + .into_iter() + .map(|(id, framework, cost)| (Some(id), framework, cost)) + .collect(); + self + } + pub fn with_window_framework_costs( mut self, costs: Vec<(SummaryWindowFramework, Cost)>, ) -> Self { - self.window_framework_costs = costs; + self.window_framework_costs = costs + .into_iter() + .map(|(framework, cost)| (None, framework, cost)) + .collect(); self } @@ -387,14 +399,19 @@ impl CostModel for ControlPlaneCostModel { // GOS/error propagation is introduced by the later adaptation // slice. Until then, do not claim an approximate exponential // histogram window is exact. - .filter(|(framework, _)| { + .filter(|(_, framework, _)| { !matches!(framework, SummaryWindowFramework::ExponentialHistogram) }) - .filter(|(_, cost)| cost.0.is_finite() && cost.0 >= 0.0) - .min_by(|left, right| left.1 .0.total_cmp(&right.1 .0)) + .filter(|(_, _, cost)| cost.0.is_finite() && cost.0 >= 0.0) + .min_by(|left, right| { + left.2 + .0 + .total_cmp(&right.2 .0) + .then_with(|| left.0.cmp(&right.0)) + }) .map( - |(framework, physical_cost)| CompleteSummaryCandidateEstimate { - physical_plan_id: None, + |(id, framework, physical_cost)| CompleteSummaryCandidateEstimate { + physical_plan_id: id.clone(), cost: Cost(lifecycle_cost + physical_cost.0), window_frameworks: vec![Some(framework.clone()); deployments.len()], window_accuracy_guarantee: Some( diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs index d2baa296..7079c875 100644 --- a/control_plane/src/physical/workload_cost.rs +++ b/control_plane/src/physical/workload_cost.rs @@ -515,6 +515,14 @@ mod tests { ] { let (mut request, env) = fixture().planning_request().unwrap(); request.queries[0].query_string = query.into(); + request + .query_workload + .as_mut() + .unwrap() + .repeating_queries + .as_mut() + .unwrap()[0] + .query = planner_types::workload::Query(query.into()); let exact = with_exact_alternative(request).unwrap().pop().unwrap(); let plan = PhysicalCompiler .compile(exact.clone(), env.clone()) @@ -656,6 +664,16 @@ mod tests { let mut second = shared.queries[0].clone(); second.query_id = "second-consumer".into(); second.query_string = "quantile_over_time(0.5, m[1m])".into(); + let entries = shared + .query_workload + .as_mut() + .unwrap() + .repeating_queries + .as_mut() + .unwrap(); + let mut demand = entries[0].clone(); + demand.query = planner_types::workload::Query(second.query_string.clone()); + entries.push(demand); shared.queries.push(second); let roots = shared .queries diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml index d093312d..b1d27c1c 100644 --- a/crates/asap_types/Cargo.toml +++ b/crates/asap_types/Cargo.toml @@ -30,4 +30,4 @@ xxhash-rust = { version = "0.8", features = ["xxh64"] } # exactly (`control_plane/Cargo.toml`) -- two different revs of the same # git dependency in one workspace resolve to two distinct Rust types that # won't unify. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "abb2f20e27091ac0c60715dc42b9b59c75b3447c" } diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index f8b43fbe..5e2d885e 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -38,7 +38,7 @@ control_plane = { path = "../control_plane" } # reduction: Reduction, .. }`) are `pre_asap` types, in the same crate now # (not a separate `asap-ir` import). Query serving consumes the compiled # QueryPlan; these types are used at physical-plan compilation boundaries. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "abb2f20e27091ac0c60715dc42b9b59c75b3447c" } # Shared external (workspace) serde.workspace = true diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs index 34ce3c56..8a443cab 100644 --- a/data_plane/src/precompute_engine/worker.rs +++ b/data_plane/src/precompute_engine/worker.rs @@ -395,18 +395,33 @@ impl Worker { } let state = self.group_states.get_mut(&sid).unwrap(); + // Keep original timestamps inside accumulators (notably rate/increase), + // shifting only pane membership and closure watermark for PromQL (a,b]. + let right_closed = state + .config + .parameters + .get("promql_right_closed") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false); + let pane_timestamp = |ts: i64| { + if right_closed { + ts.saturating_sub(1) + } else { + ts + } + }; // Find the timestamp span in this batch. A first batch may contain // several windows (Prometheus commonly sends catch-up samples after // startup), so its minimum timestamp is also the initial closure // scan boundary. let batch_min_ts = samples .iter() - .map(|(_, ts, _)| *ts) + .map(|(_, ts, _)| pane_timestamp(*ts)) .min() .unwrap_or(i64::MIN); let batch_max_ts = samples .iter() - .map(|(_, ts, _)| *ts) + .map(|(_, ts, _)| pane_timestamp(*ts)) .max() .unwrap_or(i64::MIN); let previous_event_time = state.max_event_time_ms; @@ -423,8 +438,9 @@ impl Worker { // Route each sample to its pane for (series_key, ts, val) in &samples { let too_late = previous_event_time != i64::MIN - && *ts < watermark_for_event_time(previous_event_time, allowed_lateness_ms); - let pane_start = state.window_manager.pane_start_for(*ts); + && pane_timestamp(*ts) + < watermark_for_event_time(previous_event_time, allowed_lateness_ms); + let pane_start = state.window_manager.pane_start_for(pane_timestamp(*ts)); let pane_end = pane_start + state.window_manager.slide_interval_ms(); let pane_closed = !state.active_panes.contains_key(&pane_start) && previous_closure_watermark >= pane_start + state.window_manager.window_size_ms(); 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 de53bf96..1a0cc1dd 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 @@ -110,6 +110,16 @@ pub fn execute_query_plan_instant( } else { now_ms.saturating_sub(entry.instant.lookback_ms) }; + for binding in entry.materialization_bindings() { + if binding.window_ms == 0 + || t0_ms % binding.window_ms != 0 + || now_ms % binding.window_ms != 0 + { + return Err(LoweringSkip::MaterializationNotReady( + "evaluation interval cuts a materialized pane".into(), + )); + } + } let outcome = execute_physical_query_plan( index, entry, @@ -896,6 +906,101 @@ mod tests { assert_eq!(outcome.coverage, Some((2_000, 2_000))); } + #[test] + fn repeated_multi_pane_reads_exclude_expired_state_and_reject_gaps() { + let idx = SketchStore::new(); + let policy = asap_types::PolicyFingerprint(777); + idx.register(SketchInstanceMetadata { + sid: 7, + metric_name: "requests_total".into(), + group_by_keys: std::collections::BTreeSet::new(), + 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: policy, + }); + // Old panes remain stored; each advancing query must select only its lookback. + for pane in 0..8 { + idx.append_precompute( + 7, + BTreeMap::new(), + (pane * 10_000, (pane + 1) * 10_000), + Box::new( + crate::precompute_engine::operators::SumAccumulator::with_sum( + (pane + 1) as f64, + ), + ), + ); + } + + let entry = control_plane::query_plan::QueryPlanEntry { + query_id: "q-rate".into(), + canonical_promql: "rate(requests_total[1m])".into(), + root: control_plane::query_plan::QueryNodeId(0), + nodes: BTreeMap::from([ + ( + control_plane::query_plan::QueryNodeId(0), + QueryPlanNode::ExactReadout { + input: control_plane::query_plan::QueryNodeId(1), + readout: control_plane::query_plan::ExactReadout::Sum, + }, + ), + ( + control_plane::query_plan::QueryNodeId(1), + QueryPlanNode::ReadMaterialization { + binding: control_plane::query_plan::MaterializationBinding { + materialization: policy, + metric: "requests_total".into(), + sid_grouping: vec![], + output_grouping: control_plane::query_plan::PhysicalGrouping::PerEntity, + window_ms: 10_000, + }, + }, + ), + ]), + instant: control_plane::query_plan::InstantExecution { + lookback_ms: 60_000, + full_history: false, + cumulative_readout: true, + }, + fallback: control_plane::query_plan::FallbackPolicy::ExactBackend, + }; + for (now, expected) in [(60_000, 21.0), (70_000, 27.0), (80_000, 33.0)] { + let (outcome, _) = execute_query_plan_instant(&idx, &entry, now).unwrap(); + assert_eq!(outcome.series[0].1[0].1, expected); + } + assert!( + execute_query_plan_instant(&idx, &entry, 70_001).is_err(), + "partial pane must fall back" + ); + assert!( + execute_query_plan_instant(&idx, &entry, 90_000).is_err(), + "open/missing trailing pane must fall back" + ); + // Both endpoints exist, but the missing interior pane is not evidence of zero samples. + let gap_idx = SketchStore::new(); + idx.with_instance(7, |meta| gap_idx.register(meta.clone())); + for pane in [0, 1, 3, 4, 5] { + gap_idx.append_precompute( + 7, + BTreeMap::new(), + (pane * 10_000, (pane + 1) * 10_000), + Box::new(crate::precompute_engine::operators::SumAccumulator::with_sum(1.0)), + ); + } + assert!( + execute_query_plan_instant(&gap_idx, &entry, 60_000).is_err(), + "interior gap must fall back" + ); + } + #[test] fn exact_query_plan_rate_uses_reset_aware_readout() { let idx = SketchStore::new(); diff --git a/data_plane/src/query_engines/asap_query_engine/summary_exec.rs b/data_plane/src/query_engines/asap_query_engine/summary_exec.rs index 2af5214d..523e0384 100644 --- a/data_plane/src/query_engines/asap_query_engine/summary_exec.rs +++ b/data_plane/src/query_engines/asap_query_engine/summary_exec.rs @@ -491,8 +491,7 @@ mod tests { assert_eq!(only(v), 7.0); } - /// Legacy candidate lookup cannot discard keyed items or constant weights - /// when adapting the planner's typed update contract. + /// Expressions outside the supported column/unit-weight contract fail before lookup. #[test] fn unsupported_update_semantics_fail_before_candidate_lookup() { use planner_types::post_asap::{SummaryInputExpr, SummaryUpdate}; @@ -500,12 +499,12 @@ mod tests { exec.register(7.0); for input in [ SummaryUpdate { - item: Some(SummaryInputExpr::Column(ColumnRef::Named("key".into()))), + item: Some(SummaryInputExpr::Constant(2.0)), weight: SummaryInputExpr::Column(ColumnRef::SampleValue), }, SummaryUpdate { item: None, - weight: SummaryInputExpr::Constant(1.0), + weight: SummaryInputExpr::Constant(2.0), }, ] { let mut tree = agg_node(sum(), logical_node()); @@ -516,9 +515,7 @@ mod tests { *update = input; assert!(matches!( execute(&tree, &exec), - Err(ExecError::NotYetSupported( - "keyed or non-column summary update" - )) + Err(ExecError::NotYetSupported("summary update expression")) )); } } 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 c20c8e9b..92666632 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 @@ -372,6 +372,42 @@ impl QueryExecutionContext<'_> { ExactAgg(AggregationType), } + // A missing pane can mean delayed ingestion, not an empty interval. + // Until explicit empty-pane completion exists, multi-pane reads require + // every pane for each stored series, before any cross-series merge. + let check_panes = |ends: Vec| -> Result<(), SummaryExecutorError> { + let width = binding.window_ms; + if width == 0 { + return Err(SummaryExecutorError::Unsupported("zero pane width")); + } + if self.t1_ms.saturating_sub(self.t0_ms) <= width { + return Ok(()); + } + if self.t0_ms % width != 0 || self.t1_ms % width != 0 { + return Err(SummaryExecutorError::Unsupported("partial pane interval")); + } + let mut expected = self.t0_ms.checked_add(width); + for end in ends { + let Ok(end) = u64::try_from(end) else { + continue; + }; + if end <= self.t0_ms { + continue; + } // delta decoding carry-in is not an answer pane + if Some(end) != expected { + return Err(SummaryExecutorError::Unsupported( + "missing materialized pane", + )); + } + expected = end.checked_add(width); + } + if expected != self.t1_ms.checked_add(width) { + return Err(SummaryExecutorError::Unsupported( + "incomplete materialized panes", + )); + } + Ok(()) + }; let required_keys: BTreeSet<_> = binding.sid_grouping.iter().cloned().collect(); let mut sids = self.index.sids_for_policy(binding.materialization); sids.sort_unstable(); @@ -407,8 +443,10 @@ impl QueryExecutionContext<'_> { .into_iter() .next() else { + check_panes(Vec::new())?; continue; }; + check_panes(series.samples.keys().copied().collect())?; let key = match &binding.output_grouping { PhysicalGrouping::PerEntity => series.series_label_values.clone(), PhysicalGrouping::Reduce(keys) => { @@ -427,8 +465,10 @@ impl QueryExecutionContext<'_> { .into_iter() .next() else { + check_panes(Vec::new())?; continue; }; + check_panes(windows.keys().copied().collect())?; let key = match &binding.output_grouping { PhysicalGrouping::PerEntity => labels, PhysicalGrouping::Reduce(keys) => project_group_key(keys, &labels), diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index edf36bc3..438d2a3e 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -121,6 +121,16 @@ fn is_warm(response: &Value) -> bool { // uneven instance sample counts and Remote Write retries. #[tokio::test] async fn shared_exact_dashboard_executes_selected_workload() { + run_shared_dashboard(false).await; +} + +// A selected 5s pane serves advancing 10s lookbacks through the production HTTP path. +#[tokio::test] +async fn repeated_dashboard_executes_multiple_selected_panes() { + run_shared_dashboard(true).await; +} + +async fn run_shared_dashboard(multi_pane: bool) { let fallback_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let fallback_address = fallback_listener.local_addr().unwrap(); tokio::spawn(async move { @@ -134,8 +144,16 @@ async fn shared_exact_dashboard_executes_selected_workload() { "../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) .unwrap(); - let sum = "sum by (service) (sum_over_time(asap_demo_gauge[5s]))"; - let count = "sum by (service) (count_over_time(asap_demo_gauge[5s]))"; + let sum = if multi_pane { + "sum by (service) (sum_over_time(asap_demo_gauge[10s]))" + } else { + "sum by (service) (sum_over_time(asap_demo_gauge[5s]))" + }; + let count = if multi_pane { + "sum by (service) (count_over_time(asap_demo_gauge[10s]))" + } else { + "sum by (service) (count_over_time(asap_demo_gauge[5s]))" + }; let mean = format!("{sum} / {count}"); let mut entry = snapshot["query_workload"]["repeating_queries"][2].clone(); snapshot["query_workload"]["repeating_queries"] = Value::Array( @@ -149,6 +167,25 @@ async fn shared_exact_dashboard_executes_selected_workload() { ); let mut typed: control_plane::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_value(snapshot.clone()).unwrap(); + if multi_pane { + 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.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); + } + } let (request, environment) = typed.clone().planning_request().unwrap(); let candidates = control_plane::physical::workload_cost::with_exact_alternative(request).unwrap(); @@ -189,6 +226,16 @@ async fn shared_exact_dashboard_executes_selected_workload() { assert!(plan.cost_comparison.is_some()); 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" + ); + 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); + } + } let snapshot_path = output_dir.path().join("snapshot.json"); std::fs::write(&snapshot_path, serde_json::to_vec(&snapshot).unwrap()).unwrap(); let port = unused_port(); @@ -237,7 +284,15 @@ async fn shared_exact_dashboard_executes_selected_workload() { }; let request = WriteRequest { timeseries: vec![ - labeled("api", "a", &[(base + 500, 10.0)]), + if multi_pane { + labeled( + "api", + "a", + &[(base, 10000.0), (base + 500, 10.0), (base + 5000, 6.0)], + ) + } else { + labeled("api", "a", &[(base + 500, 10.0)]) + }, labeled( "api", "b", @@ -258,22 +313,51 @@ async fn shared_exact_dashboard_executes_selected_workload() { assert_eq!(remote_write(&client, &backend, &advance).await, 204); let final_advance = WriteRequest { timeseries: vec![ - labeled("api", "a", &[(base + 10500, 1000.0)]), + if multi_pane { + labeled("api", "a", &[(base + 10000, 7.0), (base + 10500, 1000.0)]) + } else { + labeled("api", "a", &[(base + 10500, 1000.0)]) + }, labeled("api", "b", &[(base + 10500, 1000.0)]), labeled("worker", "c", &[(base + 10500, 1000.0)]), ], }; assert_eq!(remote_write(&client, &backend, &final_advance).await, 204); + if multi_pane { + let close_third = WriteRequest { + timeseries: vec![ + labeled("api", "a", &[(base + 15500, 9999.0)]), + labeled("api", "b", &[(base + 15500, 9999.0)]), + labeled("worker", "c", &[(base + 15500, 9999.0)]), + ], + }; + assert_eq!(remote_write(&client, &backend, &close_third).await, 204); + } + let evaluation = base + if multi_pane { 10000 } else { 5000 }; for (query, expected) in [ - (sum, [24.0, 24.0]), - (count, [4.0, 2.0]), - (mean.as_str(), [6.0, 12.0]), + ( + sum, + if multi_pane { + [237.0, 124.0] + } else { + [24.0, 24.0] + }, + ), + (count, if multi_pane { [8.0, 3.0] } else { [4.0, 2.0] }), + ( + mean.as_str(), + if multi_pane { + [237.0 / 8.0, 124.0 / 3.0] + } else { + [6.0, 12.0] + }, + ), ] { let result = wait_for_warm_instant( &client, &backend, query, - (base + 5000) as f64 / 1000.0, + evaluation as f64 / 1000.0, &output_dir.path().join("query_engine.log"), ) .await; @@ -290,7 +374,7 @@ async fn shared_exact_dashboard_executes_selected_workload() { ); assert_eq!( row["value"][0].as_f64().unwrap(), - (base + 5000) as f64 / 1000.0 + evaluation as f64 / 1000.0 ); } } @@ -298,8 +382,8 @@ async fn shared_exact_dashboard_executes_selected_workload() { .get(format!("{backend}/api/v1/query_range")) .query(&[ ("query", mean.clone()), - ("start", ((base + 5000) as f64 / 1000.0).to_string()), - ("end", ((base + 10000) as f64 / 1000.0).to_string()), + ("start", (evaluation as f64 / 1000.0).to_string()), + ("end", ((evaluation + 5000) as f64 / 1000.0).to_string()), ("step", "5".into()), ]) .send() @@ -312,14 +396,26 @@ async fn shared_exact_dashboard_executes_selected_workload() { for row in result["data"]["result"].as_array().unwrap() { let values = row["values"].as_array().unwrap(); assert_eq!(values.len(), 2, "{result}"); - assert_eq!(values[1][1], "100", "{result}"); + assert_eq!( + values[1][1], + if multi_pane { + if row["metric"]["service"] == "api" { + "441.4" + } else { + "550" + } + } else { + "100" + }, + "{result}" + ); } // Unaligned intervals cannot be answered by whole tumbling states. let result: Value = client .get(format!("{backend}/api/v1/query")) .query(&[ ("query", mean), - ("time", ((base + 5001) as f64 / 1000.0).to_string()), + ("time", ((evaluation + 1) as f64 / 1000.0).to_string()), ]) .send() .await diff --git a/docs/developer_docs/planning/repeated-dashboard-panes.md b/docs/developer_docs/planning/repeated-dashboard-panes.md new file mode 100644 index 00000000..0af2791f --- /dev/null +++ b/docs/developer_docs/planning/repeated-dashboard-panes.md @@ -0,0 +1,63 @@ +# Repeated dashboard pane selection + +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. + +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. + +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. + +## Execution guarantees and limits + +Backend-local raw plans encode `promql_right_closed: true` in state parameters +and therefore in the fingerprint. Raw workers assign boundary samples to +PromQL's `(start, end]` panes, while preserving original timestamps inside +accumulators. Legacy half-open states have a different identity. + +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. + +This change does not add operators or min/max-specific paths. Existing operator, +source, and predicate limitations continue to apply. + +## Reproduction + +From the repository root: + +```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. From 43c0379c5fa463bd293db7927b6eaec652b70a77 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 14:47:46 -0600 Subject: [PATCH 2/4] Reject shared panes without joint consumer lifecycle evidence --- control_plane/src/physical/compiler.rs | 47 +++++++++++++++++-- .../planning/repeated-dashboard-panes.md | 6 ++- 2 files changed, 47 insertions(+), 6 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index ed6136f6..331655c3 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -1751,7 +1751,8 @@ impl PhysicalCompiler { // BackendPlan candidates to rediscover this decision. let mut node_bindings = HashMap::::new(); let consumers = materialization_consumers(&request.queries, environment.target)?; - let mut lifecycle_estimates = BTreeMap::new(); + let mut lifecycle_estimates = + BTreeMap::::new(); for query in &request.queries { let evidence = request.evidence.get(&query.query_id); @@ -1870,14 +1871,27 @@ impl PhysicalCompiler { let materialization = backend_plan::aggregation_config_for_materialization(&aggregation)? .policy_fingerprint(); + let consumer_query_ids = state_consumers + .iter() + .map(|query| query.query_id.clone()) + .collect::>(); + // Lifecycle demand was priced before choosing pane width. Two + // distinct logical cohorts can now collide on one physical + // fingerprint; retaining either quote would omit consumers. + if lifecycle_estimates + .get(&materialization) + .is_some_and(|existing| existing.consumer_query_ids != consumer_query_ids) + { + return Err(CompileError::Lifecycle { + query_id: query.query_id.clone(), + reason: "selected pane coalesces distinct logical consumer cohorts; joint physical-pane lifecycle evidence is required".into(), + }); + } lifecycle_estimates .entry(materialization) .or_insert_with(|| MaterializationLifecycleEstimate { materialization, - consumer_query_ids: state_consumers - .iter() - .map(|query| query.query_id.clone()) - .collect(), + consumer_query_ids, window_implementation_id: window_implementation.implementation_id.clone(), horizon_seconds: query.lifecycle.horizon_seconds, expected_reads: planner_selection.expected_reads, @@ -3689,6 +3703,29 @@ mod tests { } } + // Distinct logical cohorts cannot silently coalesce using only the first quote. + #[test] + fn selected_panes_reject_unpriced_cross_cohort_coalescing() { + let mut workload = request("q20", "sum_over_time(m[20s])"); + let mut second = request("q40", "sum_over_time(m[40s])").queries.remove(0); + workload.queries[0].window_secs = 20; + workload.queries[0].window_implementations[0].window_secs = 20; + second.window_secs = 40; + second.lifecycle.evaluation_interval_ms = 20_000; + second.window_implementations[0].window_secs = 40; + workload.queries.push(second); + for query in &mut workload.queries { + query.window_implementations[0].pane_secs = 10; + query.window_implementations[0].implementation_id = "shared-ten-second-pane".into(); + } + let mut env = environment(10_000); + env.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite; + env.collector_ids.clear(); + assert!( + matches!(PhysicalCompiler.compile(workload, env), Err(CompileError::Lifecycle { reason, .. }) if reason.contains("distinct logical consumer cohorts")) + ); + } + // Non-divisor panes cannot reconstruct a lookback from whole states. #[test] fn tumbling_sizes_reject_non_divisors() { diff --git a/docs/developer_docs/planning/repeated-dashboard-panes.md b/docs/developer_docs/planning/repeated-dashboard-panes.md index 0af2791f..995ae04a 100644 --- a/docs/developer_docs/planning/repeated-dashboard-panes.md +++ b/docs/developer_docs/planning/repeated-dashboard-panes.md @@ -15,7 +15,11 @@ 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. +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. The original QueryWorkload, including recurrence, time scope, predictability, requirements and data evidence, reaches lifecycle planning. Shared consumers' From 03536ea476f37e2ef55967ab32e8379068e0514d Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 14:54:13 -0600 Subject: [PATCH 3/4] Preserve strict legacy summary contract documentation --- data_plane/src/query_engines/asap_query_engine/summary_exec.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/data_plane/src/query_engines/asap_query_engine/summary_exec.rs b/data_plane/src/query_engines/asap_query_engine/summary_exec.rs index 237521b4..10debfd4 100644 --- a/data_plane/src/query_engines/asap_query_engine/summary_exec.rs +++ b/data_plane/src/query_engines/asap_query_engine/summary_exec.rs @@ -495,7 +495,8 @@ mod tests { assert_eq!(only(v), 7.0); } - /// Expressions outside the supported column/unit-weight contract fail before lookup. + /// Legacy candidate lookup cannot discard keyed items or constant weights + /// when adapting the planner's typed update contract. #[test] fn unsupported_update_semantics_fail_before_candidate_lookup() { use planner_types::post_asap::{SummaryInputExpr, SummaryUpdate}; From 4b04801681cc257508ca81d086b11c4926aceb0c Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 15:07:49 -0600 Subject: [PATCH 4/4] Align pane regressions with safe pooled state and original lifecycle demand --- control_plane/src/physical/compiler.rs | 15 ++++++++++----- .../tests/asapquery_compatibility_process_e2e.rs | 5 +++++ 2 files changed, 15 insertions(+), 5 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index d535bc7e..b5e9278c 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -3924,7 +3924,7 @@ mod tests { #[test] fn tumbling_sizes_are_selected_by_cost_and_installed() { for (small_cost, expected_secs, expected_id) in [(0.1, 10, "small"), (10.0, 60, "large")] { - let mut request = request("q", "sum_over_time(m[1m])"); + let mut request = request("q", "sum(sum_over_time(m[1m]))"); let query = &mut request.queries[0]; let mut small = query.window_implementations[0].clone(); small.implementation_id = "small".into(); @@ -3943,7 +3943,10 @@ mod tests { .next() .unwrap(); assert_eq!(materialization.window.size_ms, expected_secs * 1000); - let entry = bundle.query_plan.lookup("sum_over_time(m[1m])").unwrap(); + let entry = bundle + .query_plan + .lookup("sum(sum_over_time(m[1m]))") + .unwrap(); assert_eq!(entry.instant.lookback_ms, 60_000); assert_eq!( entry.materialization_bindings()[0].window_ms, @@ -3959,8 +3962,10 @@ mod tests { // Distinct logical cohorts cannot silently coalesce using only the first quote. #[test] fn selected_panes_reject_unpriced_cross_cohort_coalescing() { - let mut workload = request("q20", "sum_over_time(m[20s])"); - let mut second = request("q40", "sum_over_time(m[40s])").queries.remove(0); + let mut workload = request("q20", "sum(sum_over_time(m[20s]))"); + let mut second = request("q40", "sum(sum_over_time(m[40s]))") + .queries + .remove(0); workload.queries[0].window_secs = 20; workload.queries[0].window_implementations[0].window_secs = 20; second.window_secs = 40; @@ -3982,7 +3987,7 @@ mod tests { // Non-divisor panes cannot reconstruct a lookback from whole states. #[test] fn tumbling_sizes_reject_non_divisors() { - let mut request = request("q", "sum_over_time(m[1m])"); + let mut request = request("q", "sum(sum_over_time(m[1m]))"); request.queries[0].window_implementations[0].pane_secs = 7; let mut env = environment(10_000); env.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite; diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index 114c58a4..24ed0e0d 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -1253,6 +1253,11 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() .unwrap(); snapshot.query_workload.repeating_queries.as_mut().unwrap()[0].query = planner_types::workload::Query("rate(asap_demo_counter_total[1m])".into()); + // This fixture constructs an old envelope counter over whole retired panes, + // not a real-time retracting raw counter (which Planner correctly refuses). + snapshot.query_workload.repeating_queries.as_mut().unwrap()[0] + .time_selection + .scope = planner_types::workload::QueryTimeScope::Unknown; let (mut legacy_request, mut environment) = snapshot.planning_request().unwrap(); let query = &mut legacy_request.queries[0]; let parsed = control_plane::query_parser::parse_query_expr_canonical(