diff --git a/Cargo.lock b/Cargo.lock index 5c94d4eb2..e84f81018 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=5d0b6f6edcac65edc89a72051f37977ab0c83031#5d0b6f6edcac65edc89a72051f37977ab0c83031" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=739753e33e096c01faccca8e7a1e3da5ad3aab9c#739753e33e096c01faccca8e7a1e3da5ad3aab9c" 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=5d0b6f6edcac65edc89a72051f37977ab0c83031#5d0b6f6edcac65edc89a72051f37977ab0c83031" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=739753e33e096c01faccca8e7a1e3da5ad3aab9c#739753e33e096c01faccca8e7a1e3da5ad3aab9c" 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=5d0b6f6edcac65edc89a72051f37977ab0c83031#5d0b6f6edcac65edc89a72051f37977ab0c83031" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=739753e33e096c01faccca8e7a1e3da5ad3aab9c#739753e33e096c01faccca8e7a1e3da5ad3aab9c" dependencies = [ "serde", "serde_json", diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index a4cafb246..bafd44eea 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 = "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/ProjectASAP/ASAPPlanner", rev = "5d0b6f6edcac65edc89a72051f37977ab0c83031" } -asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "5d0b6f6edcac65edc89a72051f37977ab0c83031" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "739753e33e096c01faccca8e7a1e3da5ad3aab9c" } +asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "739753e33e096c01faccca8e7a1e3da5ad3aab9c" } # 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 = "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/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 = "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/ProjectASAP/ASAPPlanner", rev = "5d0b6f6edcac65edc89a72051f37977ab0c83031" } +asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "739753e33e096c01faccca8e7a1e3da5ad3aab9c" } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs index 0165d0429..028fef543 100644 --- a/control_plane/src/backend_client.rs +++ b/control_plane/src/backend_client.rs @@ -367,6 +367,7 @@ impl BackendClient { &self, precompute_plan: &crate::physical::compiler::PrecomputePlan, backend_plan: Vec, + query_plan: &crate::query_plan::QueryPlan, storage_routing: Option, ) -> std::result::Result<(), BackendPostError> { let url = derive_physical_plan_url(&self.endpoint); @@ -376,6 +377,7 @@ impl BackendClient { .json(&serde_json::json!({ "precompute_plan": precompute_plan, "backend_plan": backend_plan, + "query_plan": query_plan, "storage_routing": storage_routing, })) .send() diff --git a/control_plane/src/emit/backend_push.rs b/control_plane/src/emit/backend_push.rs index e8becfe47..a824d0641 100644 --- a/control_plane/src/emit/backend_push.rs +++ b/control_plane/src/emit/backend_push.rs @@ -240,10 +240,24 @@ async fn push_documents_coupled( return (false, false, 0); } }; + // The compatibility emitter has no Planner-selected query catalog. It + // may still install producer/storage state, but publishes an empty + // QueryPlan so every serving request fails closed to the exact tier. + let query_plan = crate::query_plan::QueryPlan { + plan_id: crate::backend_plan::BackendPlan::decode(&plan_bytes) + .map(|plan| plan.plan_id) + .unwrap_or_default(), + entries: Default::default(), + }; for attempt in 1..=RETRY_MAX_ATTEMPTS { match client - .post_physical_plan_typed(precompute_plan, plan_bytes.clone(), Some(routing.clone())) + .post_physical_plan_typed( + precompute_plan, + plan_bytes.clone(), + &query_plan, + Some(routing.clone()), + ) .await { Ok(()) => return (true, true, attempt), diff --git a/control_plane/src/lib.rs b/control_plane/src/lib.rs index 0c443f670..e65c2b192 100644 --- a/control_plane/src/lib.rs +++ b/control_plane/src/lib.rs @@ -68,6 +68,7 @@ pub mod physical; pub mod pipeline; pub mod planner_selection; pub mod query_parser; +pub mod query_plan; pub mod query_planning; pub mod replan; pub mod runtime_samples; diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index a0a84b521..0157126f0 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -582,6 +582,7 @@ struct PhysicalPlanQueryRequest { group_by: Vec, accuracy: types_v2::AccuracyTarget, lifecycle: physical::compiler::LifecyclePlanningInput, + window_implementations: Vec, } #[derive(Debug, Deserialize)] @@ -644,6 +645,7 @@ async fn handle_compile_and_publish_physical_plan( .post_physical_plan_typed( &bundle.precompute_plan, bundle.backend_plan.encode_to_vec(), + &bundle.query_plan, None, ) .await @@ -722,6 +724,7 @@ fn compile_physical_plan_request( }; queries.push(physical::compiler::PlanningQuery { query_id: query.query_id, + query_string: query.query_string, post_asap, source: planner_types::pre_asap::Source::TimeSeries { metric: query.metric, @@ -730,6 +733,7 @@ fn compile_physical_plan_request( group_by: query.group_by, accuracy: query.accuracy, lifecycle: query.lifecycle, + window_implementations: query.window_implementations, }); } diff --git a/control_plane/src/opamp/mod.rs b/control_plane/src/opamp/mod.rs index abae284be..954341ab0 100644 --- a/control_plane/src/opamp/mod.rs +++ b/control_plane/src/opamp/mod.rs @@ -1000,6 +1000,11 @@ mod tests { parameters: serde_json::json!({"precision": 14}), group_by: vec!["service".into()], window_secs: 60, + abstract_window_framework: + planner_types::post_asap::SummaryWindowFramework::Tumbling, + window_implementation_id: "collector-tumbling-v1".into(), + pane_secs: 60, + state_layout: "anchored-pane-v1".into(), evidence_source: None, lifecycle: crate::physical::compiler::CollectorLifecycle { kind: "continuously_maintained".into(), diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 2592b8063..676b1bfba 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -4,7 +4,7 @@ //! deployment decision: evidence freshness, target capabilities, windows, the //! Collector execution projection, and the matching BackendPlan. -use std::collections::HashMap; +use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::rc::Rc; use asap_aware_mapping::cost_model::Cost; @@ -16,7 +16,7 @@ use asap_aware_mapping::{ use planner_types::post_asap::{ CompositionOperator, EvaluationSchedule, OutputRepresentation, SketchQuery, SummaryExpr, SummaryFamilyType, SummaryMaintenanceLifecycle, SummaryMaintenanceLifecycleGuarantee, - SummaryMaintenanceMode, SummaryNode, + SummaryMaintenanceMode, SummaryNode, SummaryWindowFramework, }; use planner_types::pre_asap::QueryExpr; use planner_types::workload::{ @@ -34,14 +34,21 @@ use crate::physical::colored_dag::emitter::{ AggregationInput, BackendAggregation, BackendReadout, BackendStageConfig, }; use crate::physical::post_asap::cost_model::ControlPlaneCostModel; +use crate::query_plan::{ + canonical_promql, FallbackPolicy, InstantExecution, MaterializationBinding, PhysicalGrouping, + QueryPlan, QueryPlanEntry, +}; use crate::types_v2::AccuracyTarget; use planner_types::pre_asap::Source; -pub const PLANNER_REVISION: &str = "5d0b6f6edcac65edc89a72051f37977ab0c83031"; +pub const PLANNER_REVISION: &str = "739753e33e096c01faccca8e7a1e3da5ad3aab9c"; #[derive(Debug, Clone)] pub struct PlanningQuery { pub query_id: String, + /// Catalog expression used only to build the stable QueryPlan identity. + /// The selected implementation comes from `post_asap`, never this text. + pub query_string: String, /// Planner-selected post-ASAP DAG. The physical compiler must not /// re-select a summary family from pre-ASAP input. pub post_asap: Rc, @@ -52,6 +59,40 @@ pub struct PlanningQuery { pub group_by: Vec, pub accuracy: AccuracyTarget, pub lifecycle: LifecyclePlanningInput, + /// Executor-feasible concrete realizations offered to Planner for its + /// abstract window-framework decision. The compiler retains physical + /// identities and exposes only framework + complete weighted cost to + /// Planner. An empty or stale set fails closed. + pub window_implementations: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct ImplementationCostEvidence { + pub model_version: String, + pub workload_fingerprint: String, + pub observed_at_unix_ms: u64, + pub valid_for_ms: u64, + pub horizon_seconds: f64, + pub cpu_cost: f64, + pub peak_memory_bytes: u64, + pub network_bytes: u64, + pub storage_bytes: u64, + pub source_scan_bytes: u64, + /// Dimensionally calibrated scalar passed to Planner for comparison. + pub weighted_cost: f64, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct WindowImplementationCandidate { + /// Backend-owned identity; never copied into Planner IR. + pub implementation_id: String, + pub framework: SummaryWindowFramework, + pub window_secs: u64, + pub pane_secs: u64, + pub state_layout: String, + pub cost: ImplementationCostEvidence, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] @@ -115,6 +156,10 @@ pub struct CollectorMaterialization { pub parameters: Value, pub group_by: Vec, pub window_secs: u64, + pub abstract_window_framework: SummaryWindowFramework, + pub window_implementation_id: String, + pub pane_secs: u64, + pub state_layout: String, pub evidence_source: Option, pub lifecycle: CollectorLifecycle, } @@ -154,6 +199,7 @@ pub struct PhysicalPlan { pub collector_plans: Vec, pub precompute_plan: PrecomputePlan, pub backend_plan: BackendPlan, + pub query_plan: QueryPlan, } #[derive(Debug, Error)] @@ -171,6 +217,8 @@ pub enum CompileError { Lifecycle { query_id: String, reason: String }, #[error("failed to construct BackendPlan: {0}")] BackendPlan(#[from] anyhow::Error), + #[error("failed to construct QueryPlan: {0}")] + QueryPlan(#[from] crate::query_plan::QueryPlanError), } struct QueryEvidence<'a>(Option<&'a TopKMembershipEvidence>); @@ -229,6 +277,7 @@ impl PhysicalCompiler { retention_cost_rate: Some(CostRate(query.lifecycle.costs.retention_per_second)), retirement_cost: Some(Cost(query.lifecycle.costs.retirement)), }; + let window_costs = validate_window_implementations(query, &environment)?; let model = ControlPlaneCostModel::new(query.accuracy.clone()) .with_summary_maintenance( lifecycle_costs, @@ -237,13 +286,23 @@ impl PhysicalCompiler { merge: true, delete: false, }, - ); + ) + .with_window_framework_costs(window_costs); let node = query.post_asap.clone(); let selected = extract_selected(&node).ok_or_else(|| CompileError::Query { query_id: query.query_id.clone(), reason: "selected plan has no executable sketch materialization/readout".into(), })?; - let lifecycle = select_lifecycle(query, &node, &model, &environment)?; + let planner_selection = select_lifecycle(query, &node, &model, &environment)?; + 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)) + .ok_or_else(|| CompileError::Lifecycle { + query_id: query.query_id.clone(), + reason: "Planner selected a window framework without a retained concrete implementation".into(), + })?; let metric = match &query.source { Source::TimeSeries { metric } => metric.clone(), Source::Table { .. } => { @@ -278,8 +337,12 @@ impl PhysicalCompiler { parameters: sketch_params_json(&selected.params), group_by: query.group_by.clone(), window_secs: query.window_secs, + abstract_window_framework: planner_selection.window_framework, + window_implementation_id: window_implementation.implementation_id.clone(), + pane_secs: window_implementation.pane_secs, + state_layout: window_implementation.state_layout.clone(), evidence_source: evidence.map(|e| e.source.clone()), - lifecycle, + lifecycle: planner_selection.lifecycle, }); } @@ -324,15 +387,145 @@ impl PhysicalCompiler { .map(backend_plan::aggregation_config_for_materialization) .collect::, _>>()?, }; + let materialization_fingerprints: BTreeSet<_> = + backend_plan.materializations.keys().copied().collect(); + let mut query_entries = BTreeMap::new(); + for query in &request.queries { + let canonical = canonical_promql(&query.query_string)?; + let selected = + extract_selected(&query.post_asap).ok_or_else(|| CompileError::Query { + query_id: query.query_id.clone(), + reason: "selected plan has no executable sketch materialization/readout".into(), + })?; + let family = SummaryFamilyType::Sketch( + selected.kind, + planner_types::post_asap::GroupingStrategy::PerSubpopulationInstance, + ); + let metric = match &query.source { + Source::TimeSeries { metric } => metric, + Source::Table { .. } => unreachable!("table source rejected above"), + }; + let bound: BTreeSet<_> = backend_plan + .routing + .iter() + .filter_map(|route| { + let materialization = backend_plan.materializations.get(&route.materialization)?; + (route.storage_backend == backend_plan::StorageBackend::SketchStore + && matches!(&materialization.source, Source::TimeSeries { metric: m } if m == metric) + && materialization.family == family + && materialization.window.size_ms == query.window_secs.saturating_mul(1_000) + && materialization.group_by == query.group_by) + .then_some(route.materialization) + }) + .collect(); + if bound.is_empty() { + return Err(CompileError::Query { + query_id: query.query_id.clone(), + reason: "compiled BackendPlan has no exact materialization for QueryPlan" + .into(), + }); + } + let entry = QueryPlanEntry::compile_bound( + query.query_id.clone(), + canonical.clone(), + &query.post_asap, + InstantExecution { + lookback_ms: query.window_secs.saturating_mul(1_000), + full_history: false, + cumulative_readout: true, + }, + FallbackPolicy::ExactBackend, + |node, node_family| { + let planned_metric = summary_agg_metric(node).ok_or_else(|| { + crate::query_plan::QueryPlanError::Invalid( + "materialized node has no unique time-series source".into(), + ) + })?; + if planned_metric != *metric { + return Err(crate::query_plan::QueryPlanError::Invalid(format!( + "catalog metric `{metric}` disagrees with post-ASAP source `{planned_metric}`" + ))); + } + let fingerprint = bound + .iter() + .find(|fingerprint| { + backend_plan + .materializations + .get(fingerprint) + .is_some_and(|m| &m.family == node_family) + }) + .copied() + .ok_or_else(|| { + crate::query_plan::QueryPlanError::Invalid(format!( + "no exact physical binding for {node_family:?}" + )) + })?; + Ok(MaterializationBinding { + materialization: fingerprint, + metric: metric.clone(), + sid_grouping: query.group_by.clone(), + output_grouping: PhysicalGrouping::Reduce(query.group_by.clone()), + window_ms: query.window_secs.saturating_mul(1_000), + }) + }, + )?; + if query_entries.insert(canonical.clone(), entry).is_some() { + return Err(CompileError::Query { + query_id: query.query_id.clone(), + reason: format!("duplicate canonical query identity `{canonical}`"), + }); + } + } + let query_plan = QueryPlan { + plan_id, + entries: query_entries, + }; + query_plan.validate(&materialization_fingerprints)?; Ok(PhysicalPlan { envelope, collector_plans, precompute_plan, backend_plan, + query_plan, }) } } +fn summary_agg_metric(node: &SummaryNode) -> Option { + fn walk(node: &SummaryNode, metrics: &mut BTreeSet) { + match &node.expr { + SummaryExpr::KeepPreAsap(expr) => { + let parsed = crate::query_parser::qe_to_parsed_query(expr); + if !parsed.metric_name.is_empty() { + metrics.insert(parsed.metric_name); + } + } + SummaryExpr::SummaryAgg { child, .. } => walk(child, metrics), + SummaryExpr::SummaryEstimate { summary_input, .. } => walk(summary_input, metrics), + SummaryExpr::SummaryMerge { children } => { + for child in children { + walk(child, metrics); + } + } + SummaryExpr::SummaryJoin { + outer: left, + inner: right, + .. + } + | SummaryExpr::SummarySubtract { left, right } => { + walk(left, metrics); + walk(right, metrics); + } + SummaryExpr::SummaryDelete { summary_input, .. } => walk(summary_input, metrics), + } + } + let mut metrics = BTreeSet::new(); + walk(node, &mut metrics); + (metrics.len() == 1) + .then(|| metrics.into_iter().next()) + .flatten() +} + /// 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`]. @@ -416,12 +609,76 @@ fn validate_lifecycle_input( Ok(()) } +fn validate_window_implementations( + query: &PlanningQuery, + environment: &DeploymentEnvironment, +) -> Result, CompileError> { + let mut ids = BTreeSet::new(); + let mut cheapest = BTreeMap::::new(); + for candidate in &query.window_implementations { + let evidence = &candidate.cost; + let age = environment + .observed_at_unix_ms + .saturating_sub(evidence.observed_at_unix_ms); + let valid = !candidate.implementation_id.trim().is_empty() + && ids.insert(candidate.implementation_id.clone()) + && !candidate.state_layout.trim().is_empty() + && !evidence.model_version.trim().is_empty() + && !evidence.workload_fingerprint.trim().is_empty() + && evidence.valid_for_ms != 0 + && age <= environment.max_evidence_age_ms.min(evidence.valid_for_ms) + && evidence.horizon_seconds.is_finite() + && (evidence.horizon_seconds - query.lifecycle.horizon_seconds).abs() <= f64::EPSILON + && evidence.cpu_cost.is_finite() + && evidence.cpu_cost >= 0.0 + && evidence.weighted_cost.is_finite() + && evidence.weighted_cost >= 0.0 + && candidate.window_secs == query.window_secs + && 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; + if !valid { + return Err(CompileError::Lifecycle { + query_id: query.query_id.clone(), + reason: format!( + "window implementation `{}` has incomplete, stale, incompatible, or duplicate physical evidence", + candidate.implementation_id + ), + }); + } + cheapest + .entry(candidate.framework.clone()) + .and_modify(|cost| *cost = cost.min(evidence.weighted_cost)) + .or_insert(evidence.weighted_cost); + } + if cheapest.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()) +} + +struct PlannerPhysicalSelection { + lifecycle: CollectorLifecycle, + window_framework: SummaryWindowFramework, +} + fn select_lifecycle( query: &PlanningQuery, node: &SummaryNode, model: &ControlPlaneCostModel, environment: &DeploymentEnvironment, -) -> Result { +) -> Result { let workload = QueryWorkload { language: QueryLanguage::PromQL, query_batch: None, @@ -479,31 +736,42 @@ fn select_lifecycle( query_id: query.query_id.clone(), reason: "latest ASAPPlanner selected no executable Collector lifecycle".into(), })?; - Ok(CollectorLifecycle { - kind: match guarantee.summary_maintenance_lifecycle { - SummaryMaintenanceLifecycle::Ephemeral => "ephemeral", - SummaryMaintenanceLifecycle::Prepared { .. } => "prepared", - SummaryMaintenanceLifecycle::Shared { .. } => "shared", - SummaryMaintenanceLifecycle::ContinuouslyMaintained => "continuously_maintained", - } - .into(), - maintenance_mode: match guarantee.summary_maintenance_mode { - SummaryMaintenanceMode::DirectBuild => "direct_build", - SummaryMaintenanceMode::Incremental => "incremental", - } - .into(), - evaluation_schedule: match guarantee.evaluation_schedule { - EvaluationSchedule::OneShot => "one_shot", - EvaluationSchedule::PerUpdate => "per_update", - EvaluationSchedule::OnRead => "on_read", - } - .into(), - output_representation: match guarantee.output_representation { - OutputRepresentation::PlainRows => "plain_rows", - OutputRepresentation::SummaryState => "summary_state", - OutputRepresentation::FinalizedValue => "finalized_value", - } - .into(), + let window_framework = plan + .deployments + .first() + .and_then(|deployment| deployment.selected_window_framework.clone()) + .ok_or_else(|| CompileError::Lifecycle { + query_id: query.query_id.clone(), + reason: "latest ASAPPlanner selected no window framework from the supplied physical evidence".into(), + })?; + Ok(PlannerPhysicalSelection { + lifecycle: CollectorLifecycle { + kind: match guarantee.summary_maintenance_lifecycle { + SummaryMaintenanceLifecycle::Ephemeral => "ephemeral", + SummaryMaintenanceLifecycle::Prepared { .. } => "prepared", + SummaryMaintenanceLifecycle::Shared { .. } => "shared", + SummaryMaintenanceLifecycle::ContinuouslyMaintained => "continuously_maintained", + } + .into(), + maintenance_mode: match guarantee.summary_maintenance_mode { + SummaryMaintenanceMode::DirectBuild => "direct_build", + SummaryMaintenanceMode::Incremental => "incremental", + } + .into(), + evaluation_schedule: match guarantee.evaluation_schedule { + EvaluationSchedule::OneShot => "one_shot", + EvaluationSchedule::PerUpdate => "per_update", + EvaluationSchedule::OnRead => "on_read", + } + .into(), + output_representation: match guarantee.output_representation { + OutputRepresentation::PlainRows => "plain_rows", + OutputRepresentation::SummaryState => "summary_state", + OutputRepresentation::FinalizedValue => "finalized_value", + } + .into(), + }, + window_framework, }) } @@ -611,12 +879,33 @@ mod tests { Ok(PlanningRequest { queries: vec![PlanningQuery { query_id: query_id.into(), + query_string: promql.into(), post_asap, source: Source::TimeSeries { metric: "m".into() }, window_secs: 60, group_by: vec![], accuracy, lifecycle, + window_implementations: vec![WindowImplementationCandidate { + implementation_id: "collector-tumbling-v1".into(), + framework: SummaryWindowFramework::Tumbling, + window_secs: 60, + pane_secs: 60, + state_layout: "anchored-pane-v1".into(), + cost: ImplementationCostEvidence { + model_version: "test-cost-v1".into(), + workload_fingerprint: "test-workload".into(), + observed_at_unix_ms: 9_500, + valid_for_ms: 60_000, + horizon_seconds: 300.0, + cpu_cost: 1.0, + peak_memory_bytes: 1_024, + network_bytes: 512, + storage_bytes: 512, + source_scan_bytes: 0, + weighted_cost: 1.0, + }, + }], }], evidence: evidence_by_query, planner_revision: PLANNER_REVISION.into(), @@ -653,10 +942,49 @@ mod tests { SummaryMaintenanceLifecycle::ContinuouslyMaintained ); assert_eq!(bundle.backend_plan.routing.len(), 1); + assert_eq!(bundle.query_plan.plan_id, bundle.envelope.plan_id); + let entry = bundle + .query_plan + .lookup("quantile_over_time( 0.99, m[1m] )") + .expect("canonical QueryPlan lookup"); + assert_eq!(entry.query_id, "q-quantile"); + assert_eq!( + entry + .nodes + .values() + .filter(|node| matches!( + node, + crate::query_plan::QueryPlanNode::ReadMaterialization { .. } + )) + .count(), + 1 + ); + entry + .validate( + &bundle + .backend_plan + .materializations + .keys() + .copied() + .collect(), + ) + .expect("executable physical DAG"); + let wire = serde_json::to_vec(&bundle.query_plan).expect("serialize QueryPlan"); + let decoded: QueryPlan = serde_json::from_slice(&wire).expect("deserialize QueryPlan"); + assert_eq!(decoded, bundle.query_plan); for plan in &bundle.collector_plans { assert_eq!(plan.envelope, bundle.envelope); assert_eq!(plan.materializations[0].metric, "m"); assert_eq!(plan.materializations[0].window_secs, 60); + assert_eq!( + plan.materializations[0].abstract_window_framework, + SummaryWindowFramework::Tumbling + ); + assert_eq!( + plan.materializations[0].window_implementation_id, + "collector-tumbling-v1" + ); + assert_eq!(plan.materializations[0].pane_secs, 60); assert_eq!( plan.materializations[0].lifecycle, CollectorLifecycle { @@ -673,6 +1001,16 @@ mod tests { } } + #[test] + fn missing_window_implementation_evidence_fails_closed() { + let mut request = request("q-window", "quantile_over_time(0.99, m[1m])"); + request.queries[0].window_implementations.clear(); + let error = PhysicalCompiler + .compile(request, environment(10_000)) + .expect_err("Planner must not receive a zero-cost invented window"); + assert!(matches!(error, CompileError::Lifecycle { .. })); + } + #[test] fn topk_fails_closed_without_membership_evidence() { assert!(request_with_evidence("q-topk", "topk(5, m)", None).is_err()); diff --git a/control_plane/src/physical/post_asap/cost_model.rs b/control_plane/src/physical/post_asap/cost_model.rs index eb8a093fe..244322bf7 100644 --- a/control_plane/src/physical/post_asap/cost_model.rs +++ b/control_plane/src/physical/post_asap/cost_model.rs @@ -35,11 +35,14 @@ #![allow(dead_code)] +use asap_aware_mapping::cost_model::{Cost, CostedSummaryDeployment}; use asap_aware_mapping::{ - CostModel, Implementation, SummaryMaintenanceCapabilities, - SummaryMaintenanceLifecycleCostInputs, + CompleteSummaryCandidateEstimate, CostModel, Horizon, Implementation, + SummaryMaintenanceCapabilities, SummaryMaintenanceLifecycleCostInputs, +}; +use planner_types::post_asap::{ + SketchAlgorithm, SketchParams, SketchQuery, SummaryWindowFramework, }; -use planner_types::post_asap::{SketchAlgorithm, SketchParams, SketchQuery}; use planner_types::pre_asap::expr_ir::ColumnRef; use crate::physical::deployment_cost::wire::WireCostTable; @@ -55,6 +58,7 @@ pub struct ControlPlaneCostModel { pub workload_accuracy: AccuracyTarget, lifecycle_costs: SummaryMaintenanceLifecycleCostInputs, summary_maintenance: SummaryMaintenanceCapabilities, + window_framework_costs: Vec<(SummaryWindowFramework, Cost)>, } impl ControlPlaneCostModel { @@ -63,9 +67,22 @@ impl ControlPlaneCostModel { workload_accuracy, lifecycle_costs: SummaryMaintenanceLifecycleCostInputs::default(), summary_maintenance: SummaryMaintenanceCapabilities::default(), + window_framework_costs: Vec::new(), } } + /// 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. + pub fn with_window_framework_costs( + mut self, + costs: Vec<(SummaryWindowFramework, Cost)>, + ) -> Self { + self.window_framework_costs = costs; + self + } + pub fn with_summary_maintenance( mut self, lifecycle_costs: SummaryMaintenanceLifecycleCostInputs, @@ -202,6 +219,50 @@ impl CostModel for ControlPlaneCostModel { self.summary_maintenance } + fn complete_summary_candidate_estimate( + &self, + _root: &planner_types::post_asap::SummaryNode, + _target: Option<&planner_types::pre_asap::QueryExpr>, + deployments: &[CostedSummaryDeployment<'_>], + _horizon: Option, + _expected_reads: Option, + _required_accuracy: &[AccuracyTarget], + ) -> Option { + // One compiler candidate describes the complete concrete realization + // of this query DAG. The current executor exposes one anchored window + // framework across all reachable summary states; represent that full + // per-state assignment explicitly rather than relying on traversal + // order or leaving any state uncosted. + if deployments.is_empty() || self.window_framework_costs.is_empty() { + return None; + } + let lifecycle_cost: f64 = deployments + .iter() + .map(|deployment| deployment.selected_cost.0) + .sum(); + self.window_framework_costs + .iter() + // GOS/error propagation is introduced by the later adaptation + // slice. Until then, do not claim an approximate exponential + // histogram window is exact. + .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)) + .map( + |(framework, physical_cost)| CompleteSummaryCandidateEstimate { + cost: Cost(lifecycle_cost + physical_cost.0), + window_frameworks: vec![Some(framework.clone()); deployments.len()], + window_accuracy_guarantee: Some( + planner_types::post_asap::ResultGuarantee::exact( + "backend exact window implementation", + ), + ), + }, + ) + } + fn rank_candidates( &self, intent: &AggIntent, diff --git a/control_plane/src/query_parser/mod.rs b/control_plane/src/query_parser/mod.rs index 53426c3fe..5b4524248 100644 --- a/control_plane/src/query_parser/mod.rs +++ b/control_plane/src/query_parser/mod.rs @@ -118,7 +118,7 @@ pub fn parse_query(query: &str, accuracy: AccuracyTarget) -> anyhow::Result ParsedQuery { +pub(crate) fn qe_to_parsed_query(qe: &QueryExpr) -> ParsedQuery { // The Binder-built `Scan.schema` is the complete, self-contained // column universe every `ColumnId` in the tree indexes into. We grab // it up-front so the `Aggregate` walk can recover group-by *names* diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs new file mode 100644 index 000000000..c80d07fad --- /dev/null +++ b/control_plane/src/query_plan.rs @@ -0,0 +1,435 @@ +//! Authoritative backend-executable query DAG. +//! +//! ASAPPlanner owns semantic post-ASAP IR. Physical compilation binds every +//! maintained-summary leaf to one materialization and lowers edges to stable +//! node IDs. Serving executes this graph without reconstructing Planner IR or +//! searching for compatible materializations. + +use std::collections::{BTreeMap, BTreeSet}; +use std::rc::Rc; + +use planner_types::post_asap::{SketchQuery, SummaryExpr, SummaryFamilyType, SummaryNode}; +use planner_types::pre_asap::Reduction; +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +use asap_types::PolicyFingerprint; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct QueryPlan { + pub plan_id: u64, + pub entries: BTreeMap, +} + +impl QueryPlan { + pub fn empty() -> Self { + Self { + plan_id: 0, + entries: BTreeMap::new(), + } + } + + pub fn lookup(&self, promql: &str) -> Result<&QueryPlanEntry, QueryPlanError> { + let identity = canonical_promql(promql)?; + self.entries + .get(&identity) + .ok_or(QueryPlanError::QueryNotPlanned(identity)) + } + + pub fn validate(&self, available: &BTreeSet) -> Result<(), QueryPlanError> { + for (identity, entry) in &self.entries { + if identity != &entry.canonical_promql { + return Err(QueryPlanError::Invalid(format!( + "query map key `{identity}` differs from entry identity `{}`", + entry.canonical_promql + ))); + } + entry.validate(available)?; + } + Ok(()) + } +} + +/// Stable identity inside one query entry. Edges are IDs so common +/// subexpressions remain shared after serialization. +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord, Hash)] +#[serde(transparent)] +pub struct QueryNodeId(pub u64); + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct QueryPlanEntry { + pub query_id: String, + pub canonical_promql: String, + pub root: QueryNodeId, + pub nodes: BTreeMap, + pub instant: InstantExecution, + pub fallback: FallbackPolicy, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct InstantExecution { + pub lookback_ms: u64, + pub full_history: bool, + pub cumulative_readout: bool, +} + +impl QueryPlanEntry { + pub fn compile_bound( + query_id: String, + canonical_promql: String, + root: &Rc, + instant: InstantExecution, + fallback: FallbackPolicy, + mut bind: F, + ) -> Result + where + F: FnMut( + &SummaryNode, + &SummaryFamilyType, + ) -> Result, + { + let mut compiler = DagCompiler { + next_id: 0, + nodes: BTreeMap::new(), + seen: BTreeMap::new(), + bind: &mut bind, + }; + let root = compiler.lower(root)?; + Ok(Self { + query_id, + canonical_promql, + root, + nodes: compiler.nodes, + instant, + fallback, + }) + } + + /// Validate references, bindings, reachability, and cycles before activation. + pub fn validate(&self, available: &BTreeSet) -> Result<(), QueryPlanError> { + if !self.nodes.contains_key(&self.root) { + return Err(QueryPlanError::Invalid(format!( + "query `{}` has missing root {}", + self.query_id, self.root.0 + ))); + } + for (id, node) in &self.nodes { + for input in node.inputs() { + if !self.nodes.contains_key(input) { + return Err(QueryPlanError::Invalid(format!( + "query `{}` node {} references missing input {}", + self.query_id, id.0, input.0 + ))); + } + } + if let QueryPlanNode::ReadMaterialization { binding } = node { + if !available.contains(&binding.materialization) { + return Err(QueryPlanError::Invalid(format!( + "query `{}` node {} references absent materialization {}", + self.query_id, id.0, binding.materialization.0 + ))); + } + } + } + let order = self.topological_order()?; + if order.len() != self.nodes.len() { + return Err(QueryPlanError::Invalid(format!( + "query `{}` contains unreachable nodes", + self.query_id + ))); + } + Ok(()) + } + + /// Return reachable nodes with every input before its consumer. + pub fn topological_order(&self) -> Result, QueryPlanError> { + fn visit( + id: QueryNodeId, + nodes: &BTreeMap, + visiting: &mut BTreeSet, + visited: &mut BTreeSet, + out: &mut Vec, + ) -> Result<(), QueryPlanError> { + if visited.contains(&id) { + return Ok(()); + } + if !visiting.insert(id) { + return Err(QueryPlanError::Invalid(format!( + "cycle detected at query node {}", + id.0 + ))); + } + let node = nodes + .get(&id) + .ok_or_else(|| QueryPlanError::Invalid(format!("missing query node {}", id.0)))?; + for input in node.inputs() { + visit(*input, nodes, visiting, visited, out)?; + } + visiting.remove(&id); + visited.insert(id); + out.push(id); + Ok(()) + } + let mut out = Vec::with_capacity(self.nodes.len()); + visit( + self.root, + &self.nodes, + &mut BTreeSet::new(), + &mut BTreeSet::new(), + &mut out, + )?; + Ok(out) + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum FallbackPolicy { + ExactBackend, + Reject, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct MaterializationBinding { + pub materialization: PolicyFingerprint, + pub metric: String, + /// Exact label-key layout of the stored materialization. + pub sid_grouping: Vec, + /// Query operator grouping applied while folding those SIDs. + pub output_grouping: PhysicalGrouping, + pub window_ms: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "mode", content = "keys", rename_all = "snake_case")] +pub enum PhysicalGrouping { + PerEntity, + Reduce(Vec), +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(tag = "op", rename_all = "snake_case", deny_unknown_fields)] +pub enum QueryPlanNode { + ReadMaterialization { + binding: MaterializationBinding, + }, + SummaryEstimate { + input: QueryNodeId, + query: QueryReadout, + }, + SummaryMerge { + inputs: Vec, + }, + ExactFallback { + reason: String, + }, +} + +impl QueryPlanNode { + pub fn inputs(&self) -> &[QueryNodeId] { + match self { + Self::ReadMaterialization { .. } | Self::ExactFallback { .. } => &[], + Self::SummaryEstimate { input, .. } => std::slice::from_ref(input), + Self::SummaryMerge { inputs } => inputs, + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum QueryReadout { + Quantile { + q: f64, + }, + PointCount { + key: planner_types::pre_asap::ColumnRef, + value: Option, + }, + Cardinality, + TopK { + k: usize, + }, +} + +impl From for QueryReadout { + fn from(query: SketchQuery) -> Self { + match query { + SketchQuery::Quantile { q } => Self::Quantile { q }, + SketchQuery::PointCount { key, value } => Self::PointCount { key, value }, + SketchQuery::Cardinality => Self::Cardinality, + SketchQuery::TopK { k } => Self::TopK { k }, + } + } +} + +impl From for SketchQuery { + fn from(query: QueryReadout) -> Self { + match query { + QueryReadout::Quantile { q } => Self::Quantile { q }, + QueryReadout::PointCount { key, value } => Self::PointCount { key, value }, + QueryReadout::Cardinality => Self::Cardinality, + QueryReadout::TopK { k } => Self::TopK { k }, + } + } +} + +struct DagCompiler<'a, F> { + next_id: u64, + nodes: BTreeMap, + seen: BTreeMap, + bind: &'a mut F, +} + +impl DagCompiler<'_, F> +where + F: FnMut(&SummaryNode, &SummaryFamilyType) -> Result, +{ + fn lower(&mut self, node: &Rc) -> Result { + let identity = Rc::as_ptr(node) as usize; + if let Some(id) = self.seen.get(&identity) { + return Ok(*id); + } + let id = QueryNodeId(self.next_id); + self.next_id += 1; + self.seen.insert(identity, id); + let physical = match &node.expr { + SummaryExpr::KeepPreAsap(_) => QueryPlanNode::ExactFallback { + reason: "post-ASAP node requires exact execution".into(), + }, + SummaryExpr::SummaryAgg { + family, + reduction, + child, + .. + } => { + if !matches!( + family, + SummaryFamilyType::ExactAggregate(..) | SummaryFamilyType::Sketch(..) + ) { + return Err(QueryPlanError::UnsupportedNode(format!( + "summary family {family:?}" + ))); + } + let mut binding = (self.bind)(node, family)?; + binding.output_grouping = physical_grouping(reduction, child)?; + QueryPlanNode::ReadMaterialization { binding } + } + SummaryExpr::SummaryEstimate { + summary_input, + query, + } => QueryPlanNode::SummaryEstimate { + input: self.lower(summary_input)?, + query: query.clone().into(), + }, + SummaryExpr::SummaryMerge { children } => { + if children.is_empty() { + return Err(QueryPlanError::UnsupportedNode( + "empty summary_merge".into(), + )); + } + QueryPlanNode::SummaryMerge { + inputs: children + .iter() + .map(|child| self.lower(child)) + .collect::>()?, + } + } + SummaryExpr::SummaryJoin { .. } => { + return Err(QueryPlanError::UnsupportedNode("summary_join".into())) + } + SummaryExpr::SummarySubtract { .. } => { + return Err(QueryPlanError::UnsupportedNode("summary_subtract".into())) + } + SummaryExpr::SummaryDelete { .. } => { + return Err(QueryPlanError::UnsupportedNode("summary_delete".into())) + } + }; + self.nodes.insert(id, physical); + Ok(id) + } +} + +fn physical_grouping( + reduction: &Reduction, + child: &SummaryNode, +) -> Result { + let Some(keys) = reduction.group_keys() else { + return Ok(PhysicalGrouping::PerEntity); + }; + let names = keys + .keys() + .iter() + .map(|&id| { + child + .schema + .fields + .get(id) + .map(|f| f.name.clone()) + .ok_or_else(|| QueryPlanError::Invalid(format!("unresolved grouping column {id}"))) + }) + .collect::>()?; + Ok(PhysicalGrouping::Reduce(names)) +} + +#[derive(Debug, Error)] +pub enum QueryPlanError { + #[error("invalid PromQL query identity: {0}")] + InvalidPromql(String), + #[error("query is absent from the active QueryPlan: {0}")] + QueryNotPlanned(String), + #[error("post-ASAP DAG cannot be represented by the query executor: {0}")] + UnsupportedNode(String), + #[error("invalid QueryPlan: {0}")] + Invalid(String), +} + +pub fn canonical_promql(query: &str) -> Result { + promql_parser::parser::parse(query.trim()) + .map(|expr| expr.to_string()) + .map_err(|error| QueryPlanError::InvalidPromql(error.to_string())) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn canonical_identity_ignores_formatting() { + assert_eq!( + canonical_promql("sum by (service) ( rate(http_requests_total[5m]) )").unwrap(), + canonical_promql("sum by(service)(rate(http_requests_total[5m]))").unwrap() + ); + } + + #[test] + fn graph_validation_rejects_cycles() { + let mut nodes = BTreeMap::new(); + nodes.insert( + QueryNodeId(0), + QueryPlanNode::SummaryMerge { + inputs: vec![QueryNodeId(0)], + }, + ); + let entry = QueryPlanEntry { + query_id: "q".into(), + canonical_promql: "up".into(), + root: QueryNodeId(0), + nodes, + instant: InstantExecution { + lookback_ms: 0, + full_history: false, + cumulative_readout: false, + }, + fallback: FallbackPolicy::Reject, + }; + assert!(entry + .validate(&BTreeSet::new()) + .unwrap_err() + .to_string() + .contains("cycle")); + } +} diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml index fa125ac45..0ea754f36 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 = "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/ProjectASAP/ASAPPlanner", rev = "5d0b6f6edcac65edc89a72051f37977ab0c83031" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "739753e33e096c01faccca8e7a1e3da5ad3aab9c" } diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index de6218d60..e42164a84 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) -- `find_candidates` still needs to # walk/match them directly. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "5d0b6f6edcac65edc89a72051f37977ab0c83031" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "739753e33e096c01faccca8e7a1e3da5ad3aab9c" } # Shared external (workspace) serde.workspace = true diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index ddc8e1eec..48a35403f 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -5271,6 +5271,7 @@ async fn handle_post_backend_plan( struct PhysicalPlanInstallRequest { precompute_plan: control_plane::physical::compiler::PrecomputePlan, backend_plan: Vec, + query_plan: control_plane::query_plan::QueryPlan, storage_routing: Option, } @@ -5325,6 +5326,15 @@ async fn handle_post_physical_plan( ) .into_response(); } + if request.query_plan.plan_id != new_plan.plan_id { + return ( + StatusCode::UNPROCESSABLE_ENTITY, + axum::Json(serde_json::json!({ + "status": "error", "error": "QueryPlan and BackendPlan plan_id differ" + })), + ) + .into_response(); + } let config_fps: BTreeSet = new_config.aggregation_configs.keys().copied().collect(); let plan_fps: BTreeSet = new_plan.materializations.keys().map(|fp| fp.0).collect(); if config_fps != plan_fps { @@ -5337,6 +5347,16 @@ async fn handle_post_physical_plan( ) .into_response(); } + let typed_plan_fps: BTreeSet<_> = new_plan.materializations.keys().copied().collect(); + if let Err(error) = request.query_plan.validate(&typed_plan_fps) { + return ( + StatusCode::UNPROCESSABLE_ENTITY, + axum::Json(serde_json::json!({ + "status": "error", "error": format!("QueryPlan validation error: {error}") + })), + ) + .into_response(); + } let new_routing = match request.storage_routing.as_ref() { Some(value) => match crate::storage_engines::types::BackendStorageRouting::from_json_payload(value) { Ok(routing) => Some(routing), @@ -5356,6 +5376,7 @@ async fn handle_post_physical_plan( precompute_plan: request.precompute_plan, runtime_config: Arc::new(new_config), backend_plan: Arc::new(new_plan), + query_plan: Arc::new(request.query_plan), storage_routing: new_routing, }; let generated = active.backend_plan.generated_at_unix_ms; diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index cb6209cf9..5951742f8 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -442,6 +442,7 @@ async fn main() -> Result<()> { precompute_plan: initial_precompute_plan, runtime_config: streaming_config.clone(), backend_plan: Arc::new(initial_backend_plan), + query_plan: Arc::new(control_plane::query_plan::QueryPlan::empty()), storage_routing: Arc::new( data_plane::storage_engines::types::BackendStorageRouting::empty(), ), @@ -473,7 +474,7 @@ async fn main() -> Result<()> { // EngineError::CapabilityMiss when the ASAP tier is empty // / ghost / unknown. .with_sketch_index(sketch_index.clone()) - .with_hot_reload_backend_plan(hot_reload_backend_plan.clone()); + .with_active_physical_plan(active_physical_plan.clone()); if let Some(control_plane_endpoint) = args.control_plane_endpoint.as_ref() { info!( "Capability-miss notifications enabled → {}", @@ -806,6 +807,7 @@ async fn main() -> Result<()> { precompute_plan: current.precompute_plan.clone(), runtime_config: current.runtime_config.clone(), backend_plan: current.backend_plan.clone(), + query_plan: current.query_plan.clone(), storage_routing: Arc::new(bootstrap_routing), }); } diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index d4cd924ac..9e9592922 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -56,14 +56,9 @@ pub struct ASAPQueryEngine { /// the rest of the routing matrix. archive_engine: Option>, - /// BackendPlan wire format (design-backend-plan-wire-format.md). When - /// `Some`, `post_asap_planner.rs`'s serving-time family/params lookup - /// prefers reading the installed plan's materializations directly - /// over reconstructing from `SketchStore` metadata - /// (`ObservedFamilyCostModel`). `None` when not wired up (unit - /// tests, legacy callers), which falls back to `SketchStore` - /// reconstruction only. - hot_reload_backend_plan: Option, + /// Generation-consistent physical snapshot used by the production query + /// path. QueryPlan and BackendPlan must never be sampled separately. + active_physical_plan: Option, } impl ASAPQueryEngine { @@ -94,31 +89,25 @@ impl ASAPQueryEngine { control_plane_client: None, sketch_index: None, archive_engine: None, - hot_reload_backend_plan: None, + active_physical_plan: None, } } - /// Attach a `HotReloadBackendPlan` handle so serving-time family/params - /// lookups prefer the control plane's installed `BackendPlan` over - /// `SketchStore` reconstruction (see this struct's field doc). - /// Without this call, lookups fall back to `SketchStore` - /// reconstruction unconditionally. - pub fn with_hot_reload_backend_plan( + pub fn with_active_physical_plan( mut self, - handle: crate::storage_engines::types::HotReloadBackendPlan, + handle: crate::storage_engines::types::HotReloadActivePhysicalPlan, ) -> Self { - self.hot_reload_backend_plan = Some(handle); + self.active_physical_plan = Some(handle); self } - /// Snapshot of the currently installed `BackendPlan`, if a hot-reload - /// handle is wired up. `None` otherwise — callers fall back to the - /// `SketchStore`-reconstruction path. - fn backend_plan_snapshot(&self) -> Option> { - self.hot_reload_backend_plan + fn physical_plan_snapshot( + &self, + ) -> Option> { + self.active_physical_plan .as_ref() - .map(|h| h.snapshot()) - .filter(|plan| plan.plan_id != 0) + .map(|handle| handle.snapshot()) + .filter(|plan| plan.backend_plan.plan_id != 0) } /// Phase-5 hybrid-stitch builder — attach an archive engine the @@ -320,32 +309,49 @@ impl ASAPQueryEngine { )); }; - let backend_plan_snap = self.backend_plan_snapshot(); - let result = - crate::query_engines::asap_query_engine::live_serve::serve_from_summary_executor( + let planned = match self.physical_plan_snapshot() { + Some(physical_plan) => match physical_plan.query_plan.lookup(query) { + Ok(query_entry) => { + crate::query_engines::asap_query_engine::live_serve::serve_from_query_plan( + idx, + query_entry, + start_ms, + end_ms, + false, + ) + } + Err(reason) => Err(crate::query_engines::asap_query_engine::post_asap_planner::LoweringSkip::QueryNotPlanned(reason.to_string())), + }, + #[cfg(test)] + None => crate::query_engines::asap_query_engine::live_serve::serve_from_summary_executor( idx, query, start_ms, end_ms, false, control_plane::types_v2::AccuracyTarget::Epsilon(0.01), - backend_plan_snap.as_deref(), - ) - .map_err(|reason| { - if let Some(req) = Self::requirements_from_query_str(query) { - crate::drivers::control_plane_client::spawn_capability_miss_notify( - &self.control_plane_client, - &req, - ); - } - crate::query_engines::EngineError::capability_miss( - crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), - format!( - "post-ASAP/BackendPlan resolver could not serve `{query}` over \ + None, + ), + #[cfg(not(test))] + None => Err(crate::query_engines::asap_query_engine::post_asap_planner::LoweringSkip::QueryNotPlanned( + "no active physical QueryPlan".into(), + )), + }; + let result = planned.map_err(|reason| { + if let Some(req) = Self::requirements_from_query_str(query) { + crate::drivers::control_plane_client::spawn_capability_miss_notify( + &self.control_plane_client, + &req, + ); + } + crate::query_engines::EngineError::capability_miss( + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), + format!( + "post-ASAP/BackendPlan resolver could not serve `{query}` over \ [{start_ms}, {end_ms}]: {reason:?} — failing over to archive" - ), - ) - })?; + ), + ) + })?; // Matrix shape — the range_query wire format requires it. let warm_qr = asap_tier_result_to_query_result(result.clone(), end_ms, true); @@ -550,13 +556,23 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu .duration_since(std::time::SystemTime::UNIX_EPOCH) .map(|d| d.as_millis() as u64) .unwrap_or(0); - let backend_plan = self.backend_plan_snapshot(); - let (result, t0_ms) = - crate::query_engines::asap_query_engine::live_serve::serve_instant_from_summary_executor( - idx, query, now_ms, backend_plan.as_deref(), - ) - .map_err(|reason| { - tracing::debug!(query, ?reason, "post-ASAP warm serving capability miss"); + let planned = match self.physical_plan_snapshot() { + Some(physical_plan) => match physical_plan.query_plan.lookup(query) { + Ok(query_entry) => crate::query_engines::asap_query_engine::live_serve::serve_instant_from_query_plan( + idx, query_entry, now_ms, + ), + Err(reason) => Err(crate::query_engines::asap_query_engine::post_asap_planner::LoweringSkip::QueryNotPlanned(reason.to_string())), + }, + #[cfg(test)] + None => crate::query_engines::asap_query_engine::live_serve::serve_instant_from_summary_executor( + idx, query, now_ms, None, + ), + #[cfg(not(test))] + None => Err(crate::query_engines::asap_query_engine::post_asap_planner::LoweringSkip::QueryNotPlanned( + "no active physical QueryPlan".into(), + )), + }; + let (result, t0_ms) = planned.map_err(|reason| { if let Some(req) = Self::requirements_from_query_str(query) { crate::drivers::control_plane_client::spawn_capability_miss_notify( &self.control_plane_client, 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 8032b0d02..c5f78c028 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 @@ -22,7 +22,8 @@ use control_plane::types_v2::AccuracyTarget; use crate::query_engines::asap_query_engine::post_asap_planner::LoweringSkip; use crate::query_engines::asap_query_engine::post_asap_readout::{ - execute_post_asap_instant, execute_post_asap_readout, + execute_post_asap_instant, execute_post_asap_readout, execute_query_plan_instant, + execute_query_plan_readout, }; use crate::storage_engines::sketch_db::index::SketchStore; use crate::storage_engines::sketch_db::query::ASAPTierResult; @@ -175,6 +176,41 @@ pub fn serve_instant_from_summary_executor( )) } +pub fn serve_from_query_plan( + index: &SketchStore, + entry: &control_plane::query_plan::QueryPlanEntry, + t0_ms: u64, + t1_ms: u64, + is_cumulative: bool, +) -> Result { + if !summary_executor_live_enabled() { + return Err(LoweringSkip::Disabled); + } + let outcome = execute_query_plan_readout(index, entry, t0_ms, t1_ms, is_cumulative)?; + Ok(ASAPTierResult { + series: outcome.series, + coverage: outcome.coverage, + }) +} + +pub fn serve_instant_from_query_plan( + index: &SketchStore, + entry: &control_plane::query_plan::QueryPlanEntry, + now_ms: u64, +) -> Result<(ASAPTierResult, u64), LoweringSkip> { + if !summary_executor_live_enabled() { + return Err(LoweringSkip::Disabled); + } + let (outcome, t0_ms) = execute_query_plan_instant(index, entry, now_ms)?; + Ok(( + ASAPTierResult { + series: outcome.series, + coverage: outcome.coverage, + }, + t0_ms, + )) +} + #[cfg(test)] mod tests { use super::*; diff --git a/data_plane/src/query_engines/asap_query_engine/mod.rs b/data_plane/src/query_engines/asap_query_engine/mod.rs index ba046bf9b..6736eaca8 100644 --- a/data_plane/src/query_engines/asap_query_engine/mod.rs +++ b/data_plane/src/query_engines/asap_query_engine/mod.rs @@ -11,6 +11,7 @@ pub mod engine; pub mod live_serve; +pub mod physical_dag; pub mod post_asap_planner; pub mod post_asap_readout; pub mod summary_exec; diff --git a/data_plane/src/query_engines/asap_query_engine/physical_dag.rs b/data_plane/src/query_engines/asap_query_engine/physical_dag.rs new file mode 100644 index 000000000..2d87bf24d --- /dev/null +++ b/data_plane/src/query_engines/asap_query_engine/physical_dag.rs @@ -0,0 +1,151 @@ +//! Graph traversal for an installed physical QueryPlan. +//! +//! This module owns dependency ordering and memoization only. Physical node +//! definitions live in `control_plane`; store and operator semantics are +//! supplied by a runtime adapter. + +use std::collections::BTreeMap; + +use control_plane::query_plan::{QueryNodeId, QueryPlanEntry, QueryPlanNode}; +use thiserror::Error; + +pub trait QueryNodeRuntime { + type Output: Clone; + type Error; + + fn execute_node( + &self, + id: QueryNodeId, + node: &QueryPlanNode, + inputs: &[Self::Output], + ) -> Result; +} + +#[derive(Debug, Error)] +pub enum DagExecutionError { + #[error("invalid physical query graph: {0}")] + InvalidGraph(String), + #[error("query node {node_id} failed")] + Node { node_id: u64, source: E }, +} + +/// Execute each reachable node exactly once. A diamond-shaped DAG therefore +/// performs one store read for the shared leaf, not one read per parent path. +pub fn execute( + entry: &QueryPlanEntry, + runtime: &R, +) -> Result> { + let order = entry + .topological_order() + .map_err(|error| DagExecutionError::InvalidGraph(error.to_string()))?; + let mut outputs = BTreeMap::::new(); + for id in order { + let node = entry + .nodes + .get(&id) + .ok_or_else(|| DagExecutionError::InvalidGraph(format!("missing node {}", id.0)))?; + let inputs = node + .inputs() + .iter() + .map(|input| { + outputs.get(input).cloned().ok_or_else(|| { + DagExecutionError::InvalidGraph(format!( + "node {} ran before input {}", + id.0, input.0 + )) + }) + }) + .collect::, _>>()?; + let output = + runtime + .execute_node(id, node, &inputs) + .map_err(|source| DagExecutionError::Node { + node_id: id.0, + source, + })?; + outputs.insert(id, output); + } + outputs.remove(&entry.root).ok_or_else(|| { + DagExecutionError::InvalidGraph(format!("root {} produced no output", entry.root.0)) + }) +} + +#[cfg(test)] +mod tests { + use std::cell::RefCell; + use std::collections::BTreeMap; + + use control_plane::query_plan::{FallbackPolicy, InstantExecution, QueryReadout}; + + use super::*; + + struct CountingRuntime(RefCell>); + + impl QueryNodeRuntime for CountingRuntime { + type Output = usize; + type Error = std::convert::Infallible; + + fn execute_node( + &self, + id: QueryNodeId, + _node: &QueryPlanNode, + inputs: &[usize], + ) -> Result { + *self.0.borrow_mut().entry(id).or_default() += 1; + Ok(1 + inputs.iter().sum::()) + } + } + + #[test] + fn shared_node_is_executed_once() { + let shared = QueryNodeId(0); + let left = QueryNodeId(1); + let right = QueryNodeId(2); + let root = QueryNodeId(3); + let nodes = [ + ( + shared, + QueryPlanNode::ExactFallback { + reason: "leaf".into(), + }, + ), + ( + left, + QueryPlanNode::SummaryEstimate { + input: shared, + query: QueryReadout::Cardinality, + }, + ), + ( + right, + QueryPlanNode::SummaryEstimate { + input: shared, + query: QueryReadout::Cardinality, + }, + ), + ( + root, + QueryPlanNode::SummaryMerge { + inputs: vec![left, right], + }, + ), + ] + .into_iter() + .collect(); + let entry = QueryPlanEntry { + query_id: "q".into(), + canonical_promql: "up".into(), + root, + nodes, + instant: InstantExecution { + lookback_ms: 0, + full_history: false, + cumulative_readout: false, + }, + fallback: FallbackPolicy::Reject, + }; + let runtime = CountingRuntime(RefCell::new(BTreeMap::new())); + assert_eq!(execute(&entry, &runtime).unwrap(), 5); + assert!(runtime.0.borrow().values().all(|count| *count == 1)); + } +} diff --git a/data_plane/src/query_engines/asap_query_engine/post_asap_planner.rs b/data_plane/src/query_engines/asap_query_engine/post_asap_planner.rs index ab4aeea69..0fb403b32 100644 --- a/data_plane/src/query_engines/asap_query_engine/post_asap_planner.rs +++ b/data_plane/src/query_engines/asap_query_engine/post_asap_planner.rs @@ -62,6 +62,12 @@ use crate::storage_engines::sketch_db::index::SketchStore; pub enum LoweringSkip { /// Operational kill switch disabled warm DAG execution. Disabled, + /// No exact identity exists in the active, control-plane-compiled + /// QueryPlan. This is a catalog miss, not an invitation to re-plan. + QueryNotPlanned(String), + /// The installed QueryPlan entry could not be reconstructed or failed + /// its internal contract. The request must fail closed. + InvalidQueryPlan(String), /// `parse_query_expr_canonical` failed — same failure mode the legacy /// parsing path tolerates and reports as an archive fallback reason. ParseFailed(String), 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 439ca8ec9..a62c8b977 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 @@ -5,13 +5,15 @@ use std::collections::BTreeMap; use crate::query_engines::asap_query_engine::summary_exec::{execute, ExecOutcome}; +use control_plane::query_plan::{QueryNodeId, QueryPlanNode}; use control_plane::types_v2::AccuracyTarget; +use crate::query_engines::asap_query_engine::physical_dag::{self, QueryNodeRuntime}; use crate::query_engines::asap_query_engine::post_asap_planner::{ execution_hints, plan_promql_to_post_asap, resolve_materializations_for_post_asap, LoweringSkip, }; use crate::query_engines::asap_query_engine::summary_executor::{ - QueryExecutionContext, SummaryValue, + GroupState, QueryExecutionContext, SummaryExecutorError, SummaryValue, }; use crate::storage_engines::sketch_db::index::SketchStore; @@ -85,6 +87,167 @@ pub fn execute_post_asap_readout( ) } +/// Execute an already-bound QueryPlan entry. This is the production serving +/// path: no PromQL lowering, planner cost model, observed-family lookup, or +/// BackendPlan materialization search occurs here. +pub fn execute_query_plan_readout( + index: &SketchStore, + entry: &control_plane::query_plan::QueryPlanEntry, + t0_ms: u64, + t1_ms: u64, + is_cumulative: bool, +) -> Result { + execute_physical_query_plan(index, entry, t0_ms, t1_ms, is_cumulative) +} + +pub fn execute_query_plan_instant( + index: &SketchStore, + entry: &control_plane::query_plan::QueryPlanEntry, + now_ms: u64, +) -> Result<(PostAsapReadoutOutcome, u64), LoweringSkip> { + let t0_ms = if entry.instant.full_history { + 0 + } else { + now_ms.saturating_sub(entry.instant.lookback_ms) + }; + let outcome = execute_physical_query_plan( + index, + entry, + t0_ms, + now_ms, + entry.instant.cumulative_readout, + )?; + Ok((outcome, t0_ms)) +} + +#[derive(Clone)] +enum PhysicalQueryOutput { + State(Vec<(BTreeMap, GroupState)>), + Value(Vec<(BTreeMap, SummaryValue)>), +} + +#[derive(Debug, thiserror::Error)] +enum PhysicalNodeError { + #[error("materialization/store operation failed: {0:?}")] + Store(SummaryExecutorError), + #[error("node expected summary state input")] + ExpectedState, + #[error("physical fallback requested: {0}")] + Fallback(String), +} + +struct PhysicalQueryRuntime<'a> { + context: QueryExecutionContext<'a>, +} + +impl QueryNodeRuntime for PhysicalQueryRuntime<'_> { + type Output = PhysicalQueryOutput; + type Error = PhysicalNodeError; + + fn execute_node( + &self, + _id: QueryNodeId, + node: &QueryPlanNode, + inputs: &[Self::Output], + ) -> Result { + match node { + QueryPlanNode::ReadMaterialization { binding } => { + let groups = self + .context + .read_bound_materialization(binding) + .map_err(PhysicalNodeError::Store)?; + Ok(PhysicalQueryOutput::State(groups)) + } + QueryPlanNode::SummaryEstimate { query, .. } => { + let [PhysicalQueryOutput::State(groups)] = inputs else { + return Err(PhysicalNodeError::ExpectedState); + }; + let query: planner_types::post_asap::SketchQuery = query.clone().into(); + groups + .iter() + .map(|(key, state)| { + self.context + .readout_bound(state, &query) + .map(|value| (key.clone(), value)) + .map_err(PhysicalNodeError::Store) + }) + .collect::, _>>() + .map(PhysicalQueryOutput::Value) + } + QueryPlanNode::SummaryMerge { .. } => { + let mut by_group: BTreeMap, Vec> = + BTreeMap::new(); + for input in inputs { + let PhysicalQueryOutput::State(groups) = input else { + return Err(PhysicalNodeError::ExpectedState); + }; + for (key, state) in groups { + by_group.entry(key.clone()).or_default().push(state.clone()); + } + } + if by_group.is_empty() { + return Err(PhysicalNodeError::ExpectedState); + } + by_group + .into_iter() + .map(|(key, states)| { + self.context + .merge_bound_states(states) + .map(|state| (key, state)) + .map_err(PhysicalNodeError::Store) + }) + .collect::, _>>() + .map(PhysicalQueryOutput::State) + } + QueryPlanNode::ExactFallback { reason } => { + Err(PhysicalNodeError::Fallback(reason.clone())) + } + } + } +} + +fn execute_physical_query_plan( + index: &SketchStore, + entry: &control_plane::query_plan::QueryPlanEntry, + t0_ms: u64, + t1_ms: u64, + is_cumulative: bool, +) -> Result { + let runtime = PhysicalQueryRuntime { + context: QueryExecutionContext { + index, + t0_ms, + t1_ms, + is_cumulative, + allowed_materializations: None, + }, + }; + let output = physical_dag::execute(entry, &runtime) + .map_err(|error| LoweringSkip::ExecuteFailed(error.to_string()))?; + match output { + PhysicalQueryOutput::Value(values) => { + let mut coverage = None; + let mut series = Vec::new(); + for (group_key, value) in &values { + fold_coverage(&mut coverage, value.coverage()); + series.extend(summary_value_to_series(group_key, value)); + } + Ok(PostAsapReadoutOutcome { series, coverage }) + } + PhysicalQueryOutput::State(groups) => { + let mut coverage = None; + let mut series = Vec::new(); + for (group_key, state) in &groups { + fold_coverage(&mut coverage, state.exact_coverage()); + if let Some(value) = state.exact_value(&None) { + series.push((group_key.clone(), vec![(t1_ms as i64, value)])); + } + } + Ok(PostAsapReadoutOutcome { series, coverage }) + } + } +} + /// Plan and execute an instant query without consulting the legacy candidate /// analyzer. Lookback and cumulative-vs-per-window behavior come from the /// post-ASAP DAG itself. @@ -243,7 +406,7 @@ mod tests { first_seen_unix_ms: 0, retired_at_ms: None, expires_at_ms: None, - policy_fp: asap_types::PolicyFingerprint::UNSET, + policy_fp: asap_types::PolicyFingerprint(123), }); use asap_sketchlib::{HllSketch, HllVariant, MessagePackCodec}; let mut sk = HllSketch::new(HllVariant::Regular, 14); @@ -304,6 +467,40 @@ mod tests { idx } + #[test] + fn formal_query_plan_executes_only_its_bound_policy() { + let idx = SketchStore::new(); + register_hll(&idx, 1, "api", &["a", "b"]); + register_hll(&idx, 2, "worker", &["b", "c"]); + let node = plan_promql_to_post_asap(&idx, "count(unique_users)", accuracy(), None) + .expect("compile-stage fixture"); + let canonical = control_plane::query_plan::canonical_promql("count(unique_users)").unwrap(); + let entry = control_plane::query_plan::QueryPlanEntry::compile_bound( + "q-cardinality".into(), + canonical, + &node, + control_plane::query_plan::InstantExecution { + lookback_ms: 60_000, + full_history: false, + cumulative_readout: true, + }, + control_plane::query_plan::FallbackPolicy::ExactBackend, + |_node, _family| { + Ok(control_plane::query_plan::MaterializationBinding { + materialization: asap_types::PolicyFingerprint(123), + metric: "unique_users".into(), + sid_grouping: vec!["service".into()], + output_grouping: control_plane::query_plan::PhysicalGrouping::PerEntity, + window_ms: 60_000, + }) + }, + ) + .unwrap(); + let result = execute_query_plan_readout(&idx, &entry, 1_000, 2_000, true) + .expect("execute formal QueryPlan"); + assert!(!result.series.is_empty()); + } + #[test] fn bare_range_function_keeps_one_series_per_entity() { let idx = ddsketch_fixture(); 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 1c374624b..cc7efc88f 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 @@ -306,6 +306,114 @@ impl SummaryValue { } } +impl QueryExecutionContext<'_> { + /// Resolve exactly one compiler-bound materialization. This is the formal + /// QueryPlan path: fingerprint -> SID is the only lookup; metadata checks + /// are integrity checks and never broaden the candidate set. + pub fn read_bound_materialization( + &self, + binding: &control_plane::query_plan::MaterializationBinding, + ) -> Result, GroupState)>, SummaryExecutorError> { + use control_plane::query_plan::PhysicalGrouping; + + enum Candidate { + Sketch(DeltaSketchKind), + ExactAgg(AggregationType), + } + + let required_keys: BTreeSet<_> = binding.sid_grouping.iter().cloned().collect(); + let mut sids = self.index.sids_for_policy(binding.materialization); + sids.sort_unstable(); + sids.dedup(); + let mut by_group: BTreeMap, Vec> = BTreeMap::new(); + + for sid in sids { + let candidate = self + .index + .with_instance(sid, |meta| { + if meta.policy_fp != binding.materialization + || meta.metric_name != binding.metric + || meta.group_by_keys != required_keys + { + return None; + } + match &meta.agg_kind { + AggKind::Sketch { + algorithm, config, .. + } => to_delta_kind(algorithm.clone(), config).map(Candidate::Sketch), + AggKind::ExactAgg { agg_type, .. } => Some(Candidate::ExactAgg(*agg_type)), + } + }) + .flatten(); + let Some(candidate) = candidate else { continue }; + match candidate { + Candidate::Sketch(kind) => { + let Some(series) = self + .index + .query_range(sid, self.t0_ms, self.t1_ms) + .into_iter() + .next() + else { + continue; + }; + let key = match &binding.output_grouping { + PhysicalGrouping::PerEntity => series.series_label_values.clone(), + PhysicalGrouping::Reduce(keys) => { + project_group_key(keys, &series.series_label_values) + } + }; + by_group.entry(key).or_default().push(GroupState::Sketch { + entries: vec![Rc::new(series)], + kind, + }); + } + Candidate::ExactAgg(agg_type) => { + let Some((labels, windows)) = self + .index + .query_exact_agg_range(sid, self.t0_ms, self.t1_ms) + .into_iter() + .next() + else { + continue; + }; + let key = match &binding.output_grouping { + PhysicalGrouping::PerEntity => labels, + PhysicalGrouping::Reduce(keys) => project_group_key(keys, &labels), + }; + by_group.entry(key).or_default().push(GroupState::ExactAgg { + entries: vec![Rc::new(windows)], + agg_type, + }); + } + } + } + if by_group.is_empty() { + return Err(SummaryExecutorError::NoCandidates); + } + by_group + .into_iter() + .map(|(key, states)| { + ::merge_states(self, states).map(|state| (key, state)) + }) + .collect() + } + + pub fn readout_bound( + &self, + state: &GroupState, + query: &SketchQuery, + ) -> Result { + ::readout(self, state, query) + } + + pub fn merge_bound_states( + &self, + states: Vec, + ) -> Result { + ::merge_states(self, states) + } +} + /// Fold `w_end` (a raw window-end timestamp, may be negative pre-epoch /// in principle) into a running `(min, max)` coverage accumulator — /// shared by `readout_cumulative`/`readout_per_window` so both compute @@ -366,7 +474,21 @@ impl<'a> SummaryExecutor for QueryExecutionContext<'a> { ExactAgg(AggregationType), } - let candidate_sids = self.index.instances_matching(&metric, &required_keys); + // A compiled QueryPlan resolves materializations before activation. + // Serving follows the direct fingerprint -> SID reverse index; it + // never scans metric registrations to discover a compatible family. + let candidate_sids = if let Some(allowed) = &self.allowed_materializations { + let mut sids: Vec<_> = allowed + .iter() + .flat_map(|fingerprint| self.index.sids_for_policy(*fingerprint)) + .collect(); + sids.sort_unstable(); + sids.dedup(); + sids + } else { + // Compatibility-only callers without a formal QueryPlan. + self.index.instances_matching(&metric, &required_keys) + }; let mut out = Vec::new(); for sid in candidate_sids { let candidate = self diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index 5fbe90ae4..e06ddeb33 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -2231,10 +2231,7 @@ impl crate::storage_engines::sketch_db::index::persistence::EpochSource for Sket encoding_to_tag(s.encoding), s.bytes.clone(), ), - AggPayload::ExactAgg(p) => (p.type_name().to_string(), 0u8, { - use asap_types::traits::SerializableToSink; - p.serialize_to_bytes() - }), + AggPayload::ExactAgg(p) => (p.type_name().to_string(), 0u8, p.serialize_to_bytes()), }; approx_bytes += payload.approx_bytes(); entries.push(EpochSnapshotEntry { diff --git a/data_plane/src/storage_engines/types/hot_reload_config.rs b/data_plane/src/storage_engines/types/hot_reload_config.rs index 62322bb3d..8d2f60f43 100644 --- a/data_plane/src/storage_engines/types/hot_reload_config.rs +++ b/data_plane/src/storage_engines/types/hot_reload_config.rs @@ -88,6 +88,7 @@ pub struct ActivePhysicalPlan { pub precompute_plan: control_plane::physical::compiler::PrecomputePlan, pub runtime_config: Arc, pub backend_plan: Arc, + pub query_plan: Arc, pub storage_routing: Arc, } @@ -117,6 +118,7 @@ impl std::fmt::Debug for HotReloadActivePhysicalPlan { let snapshot = self.snapshot(); f.debug_struct("HotReloadActivePhysicalPlan") .field("plan_id", &snapshot.backend_plan.plan_id) + .field("query_count", &snapshot.query_plan.entries.len()) .field( "materializations", &snapshot.precompute_plan.materializations.len(), 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 054f59b89..878f1fc98 100644 --- a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs +++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs @@ -360,8 +360,7 @@ async fn start_full_stack(otlp_http_port: u16, otlp_grpc_port: u16) -> FullStack // `sketch_index` via OTLP ingest (the engine's // `precompute_engine` shares the Arc), but the query // path can't see them without this binding. - .with_sketch_index(sketch_index.clone()) - .with_hot_reload_backend_plan(hot_reload_backend_plan.clone()), + .with_sketch_index(sketch_index.clone()), ); let server = HttpServer::new(http_config, query_engine, sketch_index) .with_hot_reload_config(hot_reload.clone()) diff --git a/docs/developer_docs/control-plane/physical-compiler.md b/docs/developer_docs/control-plane/physical-compiler.md index 08814eabf..40129850a 100644 --- a/docs/developer_docs/control-plane/physical-compiler.md +++ b/docs/developer_docs/control-plane/physical-compiler.md @@ -9,43 +9,46 @@ The compiler consumes ASAPPlanner types pinned to the revision exposed as `physical::compiler::PLANNER_REVISION`, selects from Planner's legal candidate space with backend-owned cost and evidence inputs, and emits one `PhysicalPlan`. The plan contains CollectorPlan, -PrecomputePlan, and BackendPlan projections compiled from the same decision +PrecomputePlan, BackendPlan, and QueryPlan projections compiled from the same decision for every target collector. Legacy `StageAllocator`/`ThreeStageEmitter` paths remain for older publication flows; they are not a second semantic planner. -ASAPPlanner owns logical semantics and summary selection. In particular, a -deployment override may choose only a family compatible with the selected -statistic; the physical compiler must reject or ignore an incompatible -override, never change the statistic to make the override fit. Upgrading the -Planner pin is a whole-interface migration because newer Planner revisions -change the post-ASAP family, reduction, grouping, and maintenance types. +ASAPPlanner owns abstract semantics and selection: summary family and +parameters, summary-maintenance lifecycle, and summary-window framework. The +backend enumerates executor-feasible concrete implementations and supplies +complete workload-scoped cost evidence to Planner. It then retains the +concrete identity corresponding to Planner's selected abstract framework. +Missing or stale implementation evidence makes the candidate unavailable; the +compiler never invents a framework or assigns it an optimistic zero cost. ## 1. Code architecture The control plane has three public layers: ```text -PlanningRequest - | - v -ASAPPlanner candidate selection +PlanningRequest + DataWorkload + concrete implementation evidence + | ^ + | abstract candidates | complete physical costs + v | +ASAPPlanner selection <---------- PhysicalCompiler | v PhysicalCompiler -------> PhysicalPlan - | | | - v v v - Collector Precompute Backend - Plan Plan Plan + | | | | + v v v v + Collector Precompute Backend Query + Plan Plan Plan DAG ``` - **Planner selection boundary** is `planner_selection::select_summary_with_evidence`. It enumerates Planner's candidates and commits only a legal candidate. -- **Physical compiler** adds backend-owned placement, windows, transport, and - runtime capabilities without changing logical semantics. +- **Physical compiler** enumerates concrete window/pane/state-layout, + placement, transport, and runtime implementations without changing the + Planner-owned abstract framework. - **PhysicalPlan** is the only output passed to publication. Its CollectorPlan, - PrecomputePlan, and BackendPlan projections are created together and share identities. + PrecomputePlan, BackendPlan, and QueryPlan projections are created together and share identities. Logical query parsing, summary alternatives, guarantees, and candidate search remain public ASAPPlanner interfaces. Runtime publication is documented in @@ -70,7 +73,7 @@ Input definitions: | Field | Definition | | --- | --- | -| `queries` | Canonical `QueryExpr`, source, window, grouping labels, accuracy, and stable query ID. | +| `queries` | Canonical `QueryExpr`, source, grouping labels, accuracy, stable query ID, `DataWorkload`, and executor-feasible `window_implementations`. | | `evidence` | Optional typed TopK membership certificates keyed by query ID. | | `planner_revision` | Immutable Planner build/revision used for reproducibility. | @@ -91,6 +94,14 @@ 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 +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 +physical realizations of one framework to the cheapest complete one before +calling Planner, then resolves Planner's result back to that retained concrete +identity. Physical identities never enter post-ASAP IR. + ### Physical compiler ```rust @@ -116,6 +127,7 @@ pub struct PhysicalPlan { pub collector_plans: Vec, // complete per-target projections pub precompute_plan: PrecomputePlan, // backend streaming materializations pub backend_plan: BackendPlan, + pub query_plan: QueryPlan, // node-ID physical serving DAG } ``` @@ -142,6 +154,22 @@ Output definitions: | `collector_plans` | One plan per targeted collector, following ASAPCollector's public CollectorPlan schema. | | `precompute_plan` | Aggregation definitions emitted to `/api/v1/streaming-config`; contains no query-string jobs. | | `backend_plan` | Matching data-plane materialization and routing contract. | +| `query_plan` | Canonical query identity, explicit fallback policy, node-ID DAG, and exact per-node materialization bindings. | + +### QueryPlan execution boundary + +`QueryPlan` is physical and executable; it is not a serialized copy of +post-ASAP IR. Each `ReadMaterialization` node binds one policy fingerprint, +metric, family/parameters, stored SID grouping layout, output reduction, and +window. The data plane resolves only that fingerprint through the +`policy_fp -> SID` reverse index and verifies SID metadata exactly. It never +scans the catalog for a serving-time candidate. + +Graph traversal is separate from node definitions and store semantics. +Activation validates roots, edges, bindings, reachability, and cycles. +Execution uses the validated topological order and memoizes every node result, +so a shared node in a diamond DAG performs one store/operator execution. A +typed node failure follows the entry's explicit fallback route. ### What the compiler puts in CollectorPlan @@ -269,13 +297,17 @@ cross-consistent; it does not mean they have been activated. For delta, verify duplicate, missing, reordered, and recovery-checkpoint cases. -### Add a physical window policy - -1. Add the public policy variant with anchor, size, slide, and lateness. -2. Prove it covers the selected logical range without changing semantics. -3. Include it in materialization identity. -4. Verify generated collector/backend windows are identical and incompatible - query ranges fail compilation. +### Add a window implementation + +1. Use an existing Planner-owned `SummaryWindowFramework`; new abstract + frameworks must first be added to Planner. +2. Advertise an executor-feasible concrete implementation with complete, + fresh `DataWorkload` evidence. +3. Prove its panes and state layout implement the framework and cover the + selected query-time range. +4. Retain the concrete implementation identity in the physical plan and + materialization identity, never in Planner IR. +5. Verify missing evidence and incompatible executor semantics fail closed. ### Required output checks