diff --git a/Cargo.lock b/Cargo.lock index b09d2af0..9eb7cadb 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=d0701dd4d4f0acb267b004c42ba30b2c9547ff7c#d0701dd4d4f0acb267b004c42ba30b2c9547ff7c" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=378a7547ede629a64e84c9f7c810226ce196cce9#378a7547ede629a64e84c9f7c810226ce196cce9" 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=d0701dd4d4f0acb267b004c42ba30b2c9547ff7c#d0701dd4d4f0acb267b004c42ba30b2c9547ff7c" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=378a7547ede629a64e84c9f7c810226ce196cce9#378a7547ede629a64e84c9f7c810226ce196cce9" 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=d0701dd4d4f0acb267b004c42ba30b2c9547ff7c#d0701dd4d4f0acb267b004c42ba30b2c9547ff7c" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=378a7547ede629a64e84c9f7c810226ce196cce9#378a7547ede629a64e84c9f7c810226ce196cce9" dependencies = [ "serde", "serde_json", diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index f058a840..0c276d99 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 = "d0701dd4d4f0acb267b004c42ba30b2c9547ff7c" } -asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "d0701dd4d4f0acb267b004c42ba30b2c9547ff7c" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" } +asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" } # 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 = "d0701dd4d4f0acb267b004c42ba30b2c9547ff7c" } +asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 961e5d8c..09e25f1c 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 = "d0701dd4d4f0acb267b004c42ba30b2c9547ff7c"; +pub const PLANNER_REVISION: &str = "378a7547ede629a64e84c9f7c810226ce196cce9"; #[derive(Debug, Clone)] pub struct PlanningQuery { @@ -1544,6 +1544,7 @@ impl BackendLocalPlanningSnapshot { )); } let mut queries = Vec::with_capacity(entries.len()); + let mut canonical_roots = Vec::with_capacity(entries.len()); for (index, entry) in entries.into_iter().enumerate() { let evaluation_interval_ms = match entry.recurrence { QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(interval)) => interval.0, @@ -1589,8 +1590,9 @@ impl BackendLocalPlanningSnapshot { horizon_seconds: self.implementation.horizon_seconds, costs: self.implementation.lifecycle_costs.clone(), }; - let post_asap = select_post_asap(&parsed, accuracy.clone(), &lifecycle, None) + 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)?; @@ -1617,6 +1619,25 @@ impl BackendLocalPlanningSnapshot { runtime_policy: RuntimeRulePolicy::default(), }); } + // A cohort shares the same end-to-end requirement, not an inferred + // weakest common accuracy. Different targets are searched separately. + let mut cohorts: Vec<(AccuracyTarget, Vec<(usize, Rc)>)> = Vec::new(); + for (index, root) in canonical_roots.into_iter().enumerate() { + let accuracy = &queries[index].accuracy; + if let Some((_, roots)) = cohorts.iter_mut().find(|(target, _)| target == accuracy) { + roots.push((index, root)); + } else { + cohorts.push((accuracy.clone(), vec![(index, root)])); + } + } + for (accuracy, roots) in cohorts { + let model = ControlPlaneCostModel::new(accuracy.clone()); + let selected = crate::planner_selection::select_workload(roots, accuracy, &model) + .map_err(|error| CompileError::Snapshot(error.to_string()))?; + for (index, node) in selected { + queries[index].post_asap = node; + } + } PhysicalCompiler.compile( PlanningRequest { queries, @@ -1631,7 +1652,7 @@ impl BackendLocalPlanningSnapshot { impl PhysicalCompiler { pub fn compile( &self, - request: PlanningRequest, + mut request: PlanningRequest, environment: DeploymentEnvironment, ) -> Result { if request.planner_revision != PLANNER_REVISION { @@ -1649,6 +1670,15 @@ impl PhysicalCompiler { }); } + let roots = request + .queries + .iter() + .enumerate() + .map(|(id, query)| (id, Rc::clone(&query.post_asap))) + .collect(); + for (id, root) in planner_types::post_asap::share_common_summary_subtrees(roots) { + request.queries[id].post_asap = root; + } let mut aggregations = Vec::with_capacity(request.queries.len()); let mut readouts = Vec::with_capacity(request.queries.len()); let mut collector_materializations = Vec::with_capacity(request.queries.len()); @@ -1657,9 +1687,9 @@ impl PhysicalCompiler { let mut runtime_policies = BTreeMap::new(); // Binding is established while compiling the physical // materializations, then consumed by QueryPlan lowering. The key is - // the planner DAG node identity within a query; serving never scans + // the planner DAG node identity across the workload; serving never scans // BackendPlan candidates to rediscover this decision. - let mut node_bindings = HashMap::<(String, usize), asap_types::PolicyFingerprint>::new(); + let mut node_bindings = HashMap::::new(); for query in &request.queries { let evidence = request.evidence.get(&query.query_id); @@ -1693,6 +1723,9 @@ impl PhysicalCompiler { query_id: query.query_id.clone(), reason, })?; + if selected.is_empty() { + continue; + } let planner_selection = select_lifecycle(query, &node, &model, &environment)?; let window_implementation = query .window_implementations @@ -1743,7 +1776,7 @@ impl PhysicalCompiler { let precompute_materialization = backend_plan::aggregation_config_for_materialization(&aggregation)?; let materialization = precompute_materialization.policy_fingerprint(); - let binding_key = (query.query_id.clone(), selected.node_identity); + let binding_key = selected.node_identity; if let Some(existing) = node_bindings.insert(binding_key, materialization) { if existing != materialization { return Err(CompileError::Query { @@ -1772,7 +1805,7 @@ impl PhysicalCompiler { }); } let collector_materialization = CollectorMaterialization { - query_id: query.query_id.clone(), + query_id: format!("state-{}", materialization.0), materialization, metric: metric.clone(), algorithm: physical_algorithm, @@ -1810,7 +1843,7 @@ impl PhysicalCompiler { } } - let plan_id = stable_plan_id(&plan_materializations); + let plan_id = stable_workload_plan_id(&plan_materializations, &request.queries); let envelope = PlanEnvelope { plan_id, plan_version: environment.plan_version, @@ -1925,7 +1958,7 @@ impl PhysicalCompiler { ) })?; let fingerprint = node_bindings - .get(&(query.query_id.clone(), node as *const SummaryNode as usize)) + .get(&(node as *const SummaryNode as usize)) .copied() .ok_or_else(|| { crate::query_plan::QueryPlanError::Invalid(format!( @@ -2427,6 +2460,20 @@ fn stable_plan_id(materializations: &[CollectorMaterialization]) -> u64 { hasher.finish() } +fn stable_workload_plan_id( + materializations: &[CollectorMaterialization], + queries: &[PlanningQuery], +) -> u64 { + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + stable_plan_id(materializations).hash(&mut hasher); + for query in queries { + query.query_id.hash(&mut hasher); + query.query_string.hash(&mut hasher); + } + hasher.finish() +} + #[cfg(test)] mod tests { use super::*; @@ -2626,6 +2673,32 @@ mod tests { } } + #[test] + fn canonical_snapshot_preserves_shared_bindings_after_serialization() { + // Two different registered readouts survive publication with one state. + let mut snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap(); + let entries = snapshot.query_workload.repeating_queries.as_mut().unwrap(); + let mut second = entries[0].clone(); + second.query = Query("quantile_over_time(0.90, m[1m])".into()); + entries.push(second); + let bundle = snapshot.compile().unwrap(); + assert_eq!(bundle.query_plan.entries.len(), 2); + assert_eq!(bundle.precompute_plan.materializations.len(), 1); + let query_plan: QueryPlan = + serde_json::from_slice(&serde_json::to_vec(&bundle.query_plan).unwrap()).unwrap(); + let bindings = query_plan + .entries + .values() + .flat_map(|entry| entry.materialization_bindings()) + .map(|binding| binding.materialization) + .collect::>(); + assert_eq!(bindings.len(), 1); + query_plan.validate(&bindings).unwrap(); + } + #[test] fn compiles_one_decision_into_matching_collector_and_backend_views() { let bundle = PhysicalCompiler diff --git a/control_plane/src/physical/post_asap/cost_model.rs b/control_plane/src/physical/post_asap/cost_model.rs index f5282fa9..1d9bd5ae 100644 --- a/control_plane/src/physical/post_asap/cost_model.rs +++ b/control_plane/src/physical/post_asap/cost_model.rs @@ -286,17 +286,15 @@ impl CostModel for ControlPlaneCostModel { } AggIntent::TopK { .. } => { let preferred = self.topk_family_order(&intent_accuracy(intent)); - let mut ranked = Vec::with_capacity(candidates.len()); - for kind in preferred { - if candidates.contains(&kind) && !ranked.contains(&kind) { - ranked.push(kind); - } - } - for kind in candidates { - if !ranked.contains(kind) { - ranked.push(kind.clone()); - } - } + // Distinct grouping implementations can use the same algorithm. + // Ranking must preserve the candidate multiset, not deduplicate it. + let mut ranked = candidates.to_vec(); + ranked.sort_by_key(|kind| { + preferred + .iter() + .position(|candidate| candidate == kind) + .unwrap_or(preferred.len()) + }); ranked } // Cardinality → Hll, Count → Cms: control_plane only ever @@ -709,6 +707,26 @@ mod tests { ); } + // Workload search can enumerate multiple layouts for the same algorithm. + #[test] + fn topk_ranking_preserves_duplicate_candidates() { + let model = ControlPlaneCostModel::new(AccuracyTarget::Exact); + let intent = AggIntent::TopK { + k: 1, + accuracy: eps(0.01), + }; + let mut candidates = vec![ + SketchAlgorithm::CmsWithHeap, + SketchAlgorithm::CountSketchWithHeap, + SketchAlgorithm::CmsWithHeap, + SketchAlgorithm::CountSketchWithHeap, + ]; + let mut ranked = model.rank_candidates(&intent, &candidates); + candidates.sort(); + ranked.sort(); + assert_eq!(ranked, candidates); + } + #[test] fn topk_loose_tier_prefers_cheaper_cms_heap() { let model = ControlPlaneCostModel::new(AccuracyTarget::Epsilon(0.1)); diff --git a/control_plane/src/planner_selection.rs b/control_plane/src/planner_selection.rs index f31982fd..efd85feb 100644 --- a/control_plane/src/planner_selection.rs +++ b/control_plane/src/planner_selection.rs @@ -20,6 +20,8 @@ use thiserror::Error; #[derive(Debug, Error)] pub enum SelectionError { + #[error("ASAPPlanner workload materialization failed: {0}")] + Workload(String), #[error("failed to derive the pre-ASAP schema: {0}")] Schema(#[from] QueryExprError), #[error("ASAPPlanner produced no legal summary candidate for the target")] @@ -149,6 +151,40 @@ pub fn select_summary_default(expr: &QueryExpr) -> Result, Selec select_summary(expr, &asap_aware_mapping::DefaultCostModel) } +/// Search a same-requirement workload cohort through Planner's canonical CSE +/// and replacement inventory. Physical implementation compatibility is checked +/// later, before publication; this function never assigns runtime identities. +pub fn select_workload( + roots: Vec<(usize, Rc)>, + accuracy: AccuracyTarget, + cost_model: &dyn CostModel, +) -> Result)>, SelectionError> { + let strategies = asap_aware_mapping::default_strategies_with(cost_model); + let space = asap_aware_mapping::search_workload_with_targets( + roots + .into_iter() + .map(|(id, root)| (id, root, Some(accuracy.clone()))) + .collect(), + &strategies, + &asap_aware_mapping::DefaultAccuracyModel, + ); + let selection = space.global_selection(cost_model); + let roots = space + .roots + .iter() + .map(|(id, root)| { + selection + .materialize(root) + .map_err(|error| SelectionError::Workload(error.to_string()))? + .map(|node| (*id, node)) + .ok_or_else(|| SelectionError::Workload(format!("missing query root {id}"))) + }) + .collect::, _>>()?; + Ok(planner_types::post_asap::share_common_summary_subtrees( + roots, + )) +} + /// Select from Planner's legal candidates with deployment-supplied accuracy /// models and typed evidence (for example a TopK membership certificate). pub fn select_summary_with_evidence( @@ -190,3 +226,93 @@ pub fn select_summary_or_keep( result => result, } } + +#[cfg(test)] +mod workload_tests { + use super::*; + use crate::physical::post_asap::cost_model::ControlPlaneCostModel; + + fn plan(queries: &[&str], accuracy: AccuracyTarget) -> Vec<(usize, Rc)> { + let roots = queries + .iter() + .enumerate() + .map(|(index, query)| { + ( + index, + Rc::new( + crate::query_parser::parse_query_expr_canonical(query, accuracy.clone()) + .unwrap(), + ), + ) + }) + .collect(); + select_workload( + roots, + accuracy.clone(), + &ControlPlaneCostModel::new(accuracy), + ) + .unwrap() + } + + // Distinct quantile roots retain their readouts while sharing one selected sketch. + #[test] + fn quantile_roots_share_selected_producer() { + let roots = plan( + &[ + "quantile_over_time(0.90, m[1m])", + "quantile_over_time(0.99, m[1m])", + ], + AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }, + ); + let SummaryExpr::SummaryEstimate { + summary_input: first, + query: q1, + } = &roots[0].1.expr + else { + panic!("{:?}", roots[0].1) + }; + let SummaryExpr::SummaryEstimate { + summary_input: second, + query: q2, + } = &roots[1].1.expr + else { + panic!("{:?}", roots[1].1) + }; + assert!(Rc::ptr_eq(first, second)); + assert_ne!(q1, q2); + } + + // Sharing must not collapse different source or logical-window requirements. + #[test] + fn distinct_windows_and_sources_do_not_share() { + let roots = plan( + &[ + "sum_over_time(m[1m])", + "sum_over_time(m[2m])", + "sum_over_time(n[1m])", + ], + AccuracyTarget::Exact, + ); + assert!(!Rc::ptr_eq(&roots[0].1, &roots[1].1)); + assert!(!Rc::ptr_eq(&roots[0].1, &roots[2].1)); + } + + // Arithmetic keeps its exact operands visible and reuses the standalone sum. + #[test] + fn weighted_mean_retains_shared_sum_operand() { + let roots = plan( + &[ + "sum_over_time(m[1m])", + "sum_over_time(m[1m]) / count_over_time(m[1m])", + ], + AccuracyTarget::Exact, + ); + let SummaryExpr::BinaryOp { lhs, .. } = &roots[1].1.expr else { + panic!("{:?}", roots[1].1) + }; + assert!(Rc::ptr_eq(&roots[0].1, lhs)); + } +} diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml index a7337d45..2d520078 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 = "d0701dd4d4f0acb267b004c42ba30b2c9547ff7c" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" } diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 0fdfaf3a..50b3a0d9 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 = "d0701dd4d4f0acb267b004c42ba30b2c9547ff7c" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" } # Shared external (workspace) serde.workspace = true