From b38da6284f2f1b2624715c674caa9b5cbea6daac Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 7 Sep 2026 16:02:44 -0600 Subject: [PATCH 1/2] Unify canonical and distributed workload selection entrypoints --- control_plane/src/main.rs | 13 +-- control_plane/src/physical/compiler.rs | 111 ++++++++++++++++++++----- control_plane/src/planner_selection.rs | 23 ++++- 3 files changed, 121 insertions(+), 26 deletions(-) diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 8e0ac885..50400b70 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -759,6 +759,7 @@ fn compile_physical_plan_request( .unwrap_or_default() .as_millis() as u64; let mut queries = Vec::with_capacity(request.queries.len()); + let mut canonical_roots = Vec::with_capacity(request.queries.len()); for query in request.queries { if query.query_id.trim().is_empty() || query.metric.trim().is_empty() @@ -773,15 +774,11 @@ fn compile_physical_plan_request( Ok(expr) => expr, Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string())), }; - let post_asap = match physical::compiler::select_post_asap( - &expr, - query.accuracy.clone(), - &query.lifecycle, - request.evidence.get(&query.query_id), - ) { + let post_asap = match control_plane::planner_selection::keep_pre_asap(&expr) { Ok(plan) => plan, Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string())), }; + canonical_roots.push(std::rc::Rc::new(expr)); queries.push(physical::compiler::PlanningQuery { query_id: query.query_id, query_string: query.query_string, @@ -798,6 +795,10 @@ fn compile_physical_plan_request( }); } + if let Err(error) = physical::compiler::select_workload_roots(&mut queries, canonical_roots, &request.evidence) { + return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string())); + } + let bundle = match physical::compiler::PhysicalCompiler.compile( physical::compiler::PlanningRequest { queries, diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 306c3452..5653c592 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -1619,25 +1619,7 @@ 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; - } - } + select_workload_roots(&mut queries, canonical_roots, &HashMap::new())?; PhysicalCompiler.compile( PlanningRequest { queries, @@ -2072,6 +2054,52 @@ fn summary_agg_metric(node: &SummaryNode) -> Option { .flatten() } +/// Shared selection boundary for canonical startup and compile-and-publish. +/// Certificate-bearing roots stay isolated: equal certificate values do not +/// establish that the certificate's source scope covers another query. +pub fn select_workload_roots( + queries: &mut [PlanningQuery], + roots: Vec>, + evidence: &HashMap, +) -> Result<(), CompileError> { + if roots.len() != queries.len() { + return Err(CompileError::Snapshot( + "canonical root/query mapping is incomplete".into(), + )); + } + let mut cohorts: Vec<(AccuracyTarget, Option, Vec<(usize, Rc)>)> = + Vec::new(); + for (index, root) in roots.into_iter().enumerate() { + let accuracy = &queries[index].accuracy; + let certificate_scope = evidence + .contains_key(&queries[index].query_id) + .then(|| queries[index].query_id.clone()); + if let Some((_, _, roots)) = cohorts + .iter_mut() + .find(|(target, scope, _)| target == accuracy && scope == &certificate_scope) + { + roots.push((index, root)); + } else { + cohorts.push((accuracy.clone(), certificate_scope, vec![(index, root)])); + } + } + for (accuracy, scope, roots) in cohorts { + let model = ControlPlaneCostModel::new(accuracy.clone()); + let certificate = scope.as_ref().and_then(|id| evidence.get(id)); + let selected = crate::planner_selection::select_workload_with_evidence( + roots, + accuracy, + &model, + &QueryEvidence(certificate), + ) + .map_err(|error| CompileError::Snapshot(error.to_string()))?; + for (index, node) in selected { + queries[index].post_asap = node; + } + } + Ok(()) +} + /// Planner-adapter selection step used before physical compilation. Keeping /// this separate makes the ownership boundary explicit: callers supply the /// selected post-ASAP DAG to [`PhysicalCompiler::compile`]. @@ -2611,6 +2639,51 @@ mod tests { request_with_evidence(query_id, promql, None).expect("post-ASAP selection") } + // Both production adapters preserve canonical root identity and select the + // whole evidence-free cohort, rather than independently binding roots. + #[test] + fn shared_selection_adapter_preserves_query_mapping() { + let mut workload = request("q90", "quantile_over_time(0.9, m[1m])"); + workload + .queries + .extend(request("q99", "quantile_over_time(0.99, m[1m])").queries); + let roots = workload + .queries + .iter() + .map(|query| { + Rc::new( + crate::query_parser::parse_query_expr_canonical( + &query.query_string, + query.accuracy.clone(), + ) + .unwrap(), + ) + }) + .collect(); + select_workload_roots(&mut workload.queries, roots, &workload.evidence).unwrap(); + let bundle = PhysicalCompiler + .compile(workload, environment(10000)) + .unwrap(); + assert_eq!(bundle.query_plan.entries.len(), 2); + assert_eq!(bundle.collector_plans[0].materializations.len(), 1); + assert_eq!( + bundle + .query_plan + .entries + .values() + .map(|entry| entry.query_id.as_str()) + .collect::>(), + BTreeSet::from(["q90", "q99"]) + ); + } + + // A broken input mapping must be rejected, never silently drop a root. + #[test] + fn shared_selection_rejects_incomplete_root_mapping() { + let mut workload = request("q", "quantile_over_time(0.9, m[1m])"); + assert!(select_workload_roots(&mut workload.queries, vec![], &workload.evidence).is_err()); + } + #[test] fn shared_materialization_is_emitted_once_for_every_runtime() { for target in [ diff --git a/control_plane/src/planner_selection.rs b/control_plane/src/planner_selection.rs index a41ef105..c464bfed 100644 --- a/control_plane/src/planner_selection.rs +++ b/control_plane/src/planner_selection.rs @@ -158,12 +158,33 @@ pub fn select_workload( roots: Vec<(usize, Rc)>, accuracy: AccuracyTarget, cost_model: &dyn CostModel, +) -> Result)>, SelectionError> { + select_workload_with_evidence( + roots, + accuracy, + cost_model, + &asap_aware_mapping::NoAccuracyEvidence, + ) +} + +/// The entire cohort uses the same scoped accuracy certificate; callers must +/// not spread one query's evidence to unrelated workload roots. +pub fn select_workload_with_evidence( + roots: Vec<(usize, Rc)>, + accuracy: AccuracyTarget, + cost_model: &dyn CostModel, + evidence: &dyn AccuracyEvidenceProvider, ) -> Result)>, SelectionError> { // Canonical CSE still runs inside search_workload_with_targets. Do not // offer CSE's per-invocation recompute alternative: this runtime currently // provisions continuously maintained, content-addressed state only. let strategies: Vec> = vec![ - Box::new(SketchAlgorithmStrategy::new(cost_model)), + Box::new(SketchAlgorithmStrategy::with_models_and_evidence( + cost_model, + &asap_aware_mapping::DefaultAccuracyModel, + &asap_aware_mapping::EqualSplitAllocator, + evidence, + )), Box::new(asap_aware_mapping::SemanticEquivalentRewriteStrategy), ]; let space = asap_aware_mapping::search_workload_with_targets( From 39d7ba804eb36e6ec7315d6cd2ee34d0fdbf4577 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 7 Sep 2026 16:20:55 -0600 Subject: [PATCH 2/2] Format the shared workload adapter call --- control_plane/src/main.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 50400b70..15680b18 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -795,7 +795,9 @@ fn compile_physical_plan_request( }); } - if let Err(error) = physical::compiler::select_workload_roots(&mut queries, canonical_roots, &request.evidence) { + if let Err(error) = + physical::compiler::select_workload_roots(&mut queries, canonical_roots, &request.evidence) + { return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string())); }