From 8cadbd70260c29694a604a21c34164cd87782adf Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 2 Sep 2026 21:19:47 -0600 Subject: [PATCH 1/6] feat(query): execute authoritative query plans --- control_plane/src/backend_client.rs | 2 + control_plane/src/emit/backend_push.rs | 16 +- control_plane/src/lib.rs | 1 + control_plane/src/main.rs | 2 + control_plane/src/physical/compiler.rs | 80 +++- control_plane/src/query_plan.rs | 374 ++++++++++++++++++ data_plane/src/drivers/query/servers/http.rs | 21 + data_plane/src/main.rs | 4 +- .../query_engines/asap_query_engine/engine.rs | 116 +++--- .../asap_query_engine/live_serve.rs | 38 +- .../asap_query_engine/post_asap_planner.rs | 6 + .../asap_query_engine/post_asap_readout.rs | 126 +++++- .../asap_query_engine/summary_executor.rs | 16 +- .../types/hot_reload_config.rs | 2 + 14 files changed, 748 insertions(+), 56 deletions(-) create mode 100644 control_plane/src/query_plan.rs 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..55fdfda71 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -644,6 +644,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 +723,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, diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 2592b8063..c6e1909cc 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; @@ -34,6 +34,9 @@ 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, QueryPlan, QueryPlanEntry, QueryPlanNode, +}; use crate::types_v2::AccuracyTarget; use planner_types::pre_asap::Source; @@ -42,6 +45,9 @@ pub const PLANNER_REVISION: &str = "5d0b6f6edcac65edc89a72051f37977ab0c83031"; #[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, @@ -154,6 +160,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 +178,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>); @@ -324,11 +333,68 @@ 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 kind = asap_types::SummaryKind::from(selected.kind); + let params = asap_types::SummaryParams::from(selected.params); + 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.kind == kind + && materialization.params == params + && 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 { + query_id: query.query_id.clone(), + canonical_promql: canonical.clone(), + root: QueryPlanNode::compile(&query.post_asap)?, + materializations: bound, + fallback: FallbackPolicy::ExactBackend, + }; + 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, }) } } @@ -611,6 +677,7 @@ 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, @@ -653,6 +720,17 @@ 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.materializations.len(), 1); + entry.root.to_summary_node().expect("executable DAG recipe"); + 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"); diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs new file mode 100644 index 000000000..0973b85b9 --- /dev/null +++ b/control_plane/src/query_plan.rs @@ -0,0 +1,374 @@ +//! Authoritative serving plan compiled from one selected post-ASAP DAG. +//! +//! A query request may be parsed only to obtain its stable identity. Family +//! selection, materialization matching and execution-shape construction all +//! happen here, before publication. The data plane therefore never invokes +//! ASAPPlanner or searches the materialization catalog while serving. + +use std::collections::{BTreeMap, BTreeSet}; +use std::rc::Rc; + +use planner_types::post_asap::{ + ExactKind, ExactParams, GroupingStrategy, SketchKind, SketchQuery, SummaryExpr, + SummaryFamilyType, SummaryField, SummaryNode, SummarySchema, +}; +use planner_types::pre_asap::{ColumnRef, DataType, QueryExpr, Reduction}; +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +use asap_types::{PolicyFingerprint, SummaryKind, SummaryParams}; + +#[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, + materializations: &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 + ))); + } + if entry.materializations.is_empty() { + return Err(QueryPlanError::Invalid(format!( + "query `{identity}` has no bound materialization" + ))); + } + if let Some(missing) = entry + .materializations + .iter() + .find(|fingerprint| !materializations.contains(fingerprint)) + { + return Err(QueryPlanError::Invalid(format!( + "query `{identity}` references absent materialization {}", + missing.0 + ))); + } + } + Ok(()) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct QueryPlanEntry { + pub query_id: String, + pub canonical_promql: String, + pub root: QueryPlanNode, + pub materializations: BTreeSet, + pub fallback: FallbackPolicy, +} + +#[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(tag = "op", rename_all = "snake_case", deny_unknown_fields)] +pub enum QueryPlanNode { + KeepPreAsap { + logical: serde_json::Value, + schema_names: Vec, + }, + SummaryAgg { + child: Box, + kind: SummaryKind, + params: SummaryParams, + col: ColumnRef, + reduction: Reduction, + schema_names: Vec, + }, + SummaryEstimate { + summary_input: Box, + query: QueryReadout, + schema_names: Vec, + }, + SummaryMerge { + children: Vec, + schema_names: Vec, + }, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum QueryReadout { + Quantile { + q: f64, + }, + PointCount { + key: ColumnRef, + value: Option, + }, + Cardinality, + TopK { + k: usize, + }, +} + +#[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 MVP query executor: {0}")] + UnsupportedNode(String), + #[error("invalid QueryPlan: {0}")] + Invalid(String), + #[error("invalid serialized pre-ASAP leaf: {0}")] + InvalidLogicalLeaf(String), +} + +/// Canonical textual identity used by both compilation and request lookup. +/// PromQL's parser/formatter normalizes whitespace and matcher formatting; +/// semantically different expressions retain different keys. +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())) +} + +impl QueryPlanNode { + pub fn compile(node: &SummaryNode) -> Result { + let schema_names = node + .schema + .fields + .iter() + .map(|field| field.name.clone()) + .collect(); + match &node.expr { + SummaryExpr::KeepPreAsap(logical) => Ok(Self::KeepPreAsap { + logical: serde_json::to_value(logical.as_ref()) + .map_err(|error| QueryPlanError::InvalidLogicalLeaf(error.to_string()))?, + schema_names, + }), + SummaryExpr::SummaryAgg { + child, + family, + col, + reduction, + .. + } => { + let (kind, params) = flatten_family(family)?; + Ok(Self::SummaryAgg { + child: Box::new(Self::compile(child)?), + kind, + params, + col: col.clone(), + reduction: reduction.clone(), + schema_names, + }) + } + SummaryExpr::SummaryEstimate { + summary_input, + query, + } => Ok(Self::SummaryEstimate { + summary_input: Box::new(Self::compile(summary_input)?), + query: query.clone().into(), + schema_names, + }), + SummaryExpr::SummaryMerge { children } => Ok(Self::SummaryMerge { + children: children + .iter() + .map(|child| Self::compile(child)) + .collect::, _>>()?, + schema_names, + }), + SummaryExpr::SummaryJoin { .. } => { + Err(QueryPlanError::UnsupportedNode("summary_join".into())) + } + SummaryExpr::SummarySubtract { .. } => { + Err(QueryPlanError::UnsupportedNode("summary_subtract".into())) + } + SummaryExpr::SummaryDelete { .. } => { + Err(QueryPlanError::UnsupportedNode("summary_delete".into())) + } + } + } + + /// Rebuild the planner-owned semantic node without making a planning + /// decision. Schema names are retained because group IDs are positional; + /// other schema/guarantee details are irrelevant to execution. + pub fn to_summary_node(&self) -> Result, QueryPlanError> { + let (expr, names) = match self { + Self::KeepPreAsap { + logical, + schema_names, + } => { + let logical: QueryExpr = serde_json::from_value(logical.clone()) + .map_err(|error| QueryPlanError::InvalidLogicalLeaf(error.to_string()))?; + (SummaryExpr::KeepPreAsap(Rc::new(logical)), schema_names) + } + Self::SummaryAgg { + child, + kind, + params, + col, + reduction, + schema_names, + } => ( + SummaryExpr::SummaryAgg { + child: child.to_summary_node()?, + family: expand_family(*kind, params.clone())?, + col: col.clone(), + reduction: reduction.clone(), + grouping: GroupingStrategy::default(), + }, + schema_names, + ), + Self::SummaryEstimate { + summary_input, + query, + schema_names, + } => ( + SummaryExpr::SummaryEstimate { + summary_input: summary_input.to_summary_node()?, + query: query.clone().into(), + }, + schema_names, + ), + Self::SummaryMerge { + children, + schema_names, + } => ( + SummaryExpr::SummaryMerge { + children: children + .iter() + .map(Self::to_summary_node) + .collect::, _>>()?, + }, + schema_names, + ), + }; + Ok(Rc::new(SummaryNode { + expr, + schema: execution_schema(names), + guarantee: None, + })) + } +} + +fn execution_schema(names: &[String]) -> SummarySchema { + SummarySchema { + fields: names + .iter() + .map(|name| SummaryField { + name: name.clone(), + dtype: SummaryFamilyType::Plain(DataType::Utf8), + nullable: true, + }) + .collect(), + time_index: None, + } +} + +fn flatten_family( + family: &SummaryFamilyType, +) -> Result<(SummaryKind, SummaryParams), QueryPlanError> { + match family { + SummaryFamilyType::ExactAggregate(kind, params) => { + Ok((kind.clone().into(), params.clone().into())) + } + SummaryFamilyType::Sketch(kind, _) => { + Ok((kind.clone().into(), kind.params().clone().into())) + } + other => Err(QueryPlanError::UnsupportedNode(format!( + "summary family {other:?}" + ))), + } +} + +fn expand_family( + kind: SummaryKind, + params: SummaryParams, +) -> Result { + if kind.is_exact() { + let exact = match (kind, params) { + (SummaryKind::Sum, SummaryParams::Sum) => (ExactKind::Sum, ExactParams::Sum), + (SummaryKind::Count, SummaryParams::Count) => (ExactKind::Count, ExactParams::Count), + (SummaryKind::MinMax, SummaryParams::MinMax) => { + (ExactKind::MinMax, ExactParams::MinMax) + } + (SummaryKind::Increase, SummaryParams::Increase) => { + (ExactKind::Increase, ExactParams::Increase) + } + (SummaryKind::Rate, SummaryParams::Rate) => (ExactKind::Rate, ExactParams::Rate), + (kind, params) => { + return Err(QueryPlanError::Invalid(format!( + "mismatched exact family {kind:?}/{params:?}" + ))) + } + }; + return Ok(SummaryFamilyType::ExactAggregate(exact.0, exact.1)); + } + let algorithm = kind + .as_sketch_kind() + .ok_or_else(|| QueryPlanError::Invalid(format!("not a sketch kind: {kind:?}")))?; + let sketch_params = params + .as_sketch_params() + .ok_or_else(|| QueryPlanError::Invalid(format!("not sketch params: {params:?}")))?; + Ok(SummaryFamilyType::Sketch( + SketchKind::new(algorithm, sketch_params), + GroupingStrategy::default(), + )) +} + +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 }, + } + } +} + +#[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() + ); + } +} 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/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..1911fb736 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 @@ -85,6 +85,57 @@ 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 { + let node = entry + .root + .to_summary_node() + .map_err(|error| LoweringSkip::InvalidQueryPlan(error.to_string()))?; + execute_bound_post_asap( + index, + &node, + t0_ms, + t1_ms, + is_cumulative, + entry.materializations.clone(), + ) +} + +pub fn execute_query_plan_instant( + index: &SketchStore, + entry: &control_plane::query_plan::QueryPlanEntry, + now_ms: u64, +) -> Result<(PostAsapReadoutOutcome, u64), LoweringSkip> { + const DEFAULT_LOOKBACK_MS: u64 = 5 * 60 * 1000; + let node = entry + .root + .to_summary_node() + .map_err(|error| LoweringSkip::InvalidQueryPlan(error.to_string()))?; + let hints = execution_hints(&node); + let t0_ms = if hints.full_history { + 0 + } else { + now_ms.saturating_sub(hints.lookback_ms.unwrap_or(DEFAULT_LOOKBACK_MS)) + }; + let outcome = execute_bound_post_asap( + index, + &node, + t0_ms, + now_ms, + hints.cumulative_readout, + entry.materializations.clone(), + )?; + Ok((outcome, t0_ms)) +} + /// 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. @@ -166,6 +217,59 @@ fn execute_planned_post_asap( } } +fn execute_bound_post_asap( + index: &SketchStore, + node: &planner_types::post_asap::SummaryNode, + t0_ms: u64, + t1_ms: u64, + is_cumulative: bool, + allowed_materializations: std::collections::BTreeSet, +) -> Result { + let ctx = QueryExecutionContext { + index, + t0_ms, + t1_ms, + is_cumulative, + allowed_materializations: Some(allowed_materializations), + }; + convert_execution_result(execute(node, &ctx), t1_ms) +} + +fn convert_execution_result( + result: Result< + ExecOutcome>, + crate::query_engines::asap_query_engine::summary_exec::ExecError< + crate::query_engines::asap_query_engine::summary_executor::SummaryExecutorError, + >, + >, + t1_ms: u64, +) -> Result { + match result { + Ok(ExecOutcome::Value(values)) => { + let mut coverage: Option<(u64, u64)> = 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 }) + } + Ok(ExecOutcome::State(groups)) => { + let mut coverage: Option<(u64, u64)> = None; + let mut series = Vec::new(); + for (group_key, state, _family) in &groups { + fold_coverage(&mut coverage, state.exact_coverage()); + let Some(value) = state.exact_value(&None) else { + continue; + }; + series.push((group_key.clone(), vec![(t1_ms as i64, value)])); + } + Ok(PostAsapReadoutOutcome { series, coverage }) + } + Err(error) => Err(LoweringSkip::ExecuteFailed(format!("{error:?}"))), + } +} + /// `SummaryValue::Points`/`TopK` -> `ASAPTierResult.series`'s row shape. /// `TopK`'s ranked-list-per-timestamp shape is pivoted into one row per /// item (each row = the group's label map plus an `item` label, one point @@ -243,7 +347,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 +408,26 @@ 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 entry = control_plane::query_plan::QueryPlanEntry { + query_id: "q-cardinality".into(), + canonical_promql: control_plane::query_plan::canonical_promql("count(unique_users)") + .unwrap(), + root: control_plane::query_plan::QueryPlanNode::compile(&node).unwrap(), + materializations: [asap_types::PolicyFingerprint(123)].into_iter().collect(), + fallback: control_plane::query_plan::FallbackPolicy::ExactBackend, + }; + 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..1d506da64 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 @@ -366,7 +366,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/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(), From 0bfc5a08bcfdc8f4ad04cc44fff94497f39d3161 Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 2 Sep 2026 22:57:30 -0600 Subject: [PATCH 2/6] fix(query-plan): bind backend plan at server boundary --- data_plane/tests/e2e_controller_plans_and_backend_serves.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) 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()) From 841fa1a231159bff2200ba490273805a971dbbce Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 3 Sep 2026 08:23:09 -0600 Subject: [PATCH 3/6] feat(planning): let Planner select abstract windows --- Cargo.lock | 6 +- control_plane/Cargo.toml | 6 +- control_plane/src/main.rs | 2 + control_plane/src/opamp/mod.rs | 5 + control_plane/src/physical/compiler.rs | 229 +++++++++++++++--- .../src/physical/post_asap/cost_model.rs | 55 ++++- crates/asap_types/Cargo.toml | 2 +- data_plane/Cargo.toml | 2 +- .../control-plane/physical-compiler.md | 55 +++-- 9 files changed, 300 insertions(+), 62 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5c94d4eb2..c0c4c3fb6 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=264937ec4a06e260920c7e583bffed34cc07dd64#264937ec4a06e260920c7e583bffed34cc07dd64" 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=264937ec4a06e260920c7e583bffed34cc07dd64#264937ec4a06e260920c7e583bffed34cc07dd64" 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=264937ec4a06e260920c7e583bffed34cc07dd64#264937ec4a06e260920c7e583bffed34cc07dd64" dependencies = [ "serde", "serde_json", diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index a4cafb246..2b25feb73 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 = "264937ec4a06e260920c7e583bffed34cc07dd64" } +asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "264937ec4a06e260920c7e583bffed34cc07dd64" } # 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 = "264937ec4a06e260920c7e583bffed34cc07dd64" } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 55fdfda71..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)] @@ -732,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 c6e1909cc..8779d8cb9 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -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::{ @@ -40,7 +40,7 @@ use crate::query_plan::{ use crate::types_v2::AccuracyTarget; use planner_types::pre_asap::Source; -pub const PLANNER_REVISION: &str = "5d0b6f6edcac65edc89a72051f37977ab0c83031"; +pub const PLANNER_REVISION: &str = "264937ec4a06e260920c7e583bffed34cc07dd64"; #[derive(Debug, Clone)] pub struct PlanningQuery { @@ -58,6 +58,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)] @@ -121,6 +155,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, } @@ -238,6 +276,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, @@ -246,13 +285,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 { .. } => { @@ -287,8 +336,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, }); } @@ -482,12 +535,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, @@ -545,31 +662,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, }) } @@ -684,6 +812,26 @@ mod tests { 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(), @@ -735,6 +883,15 @@ mod tests { 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 { @@ -751,6 +908,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..bed9195b4 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,38 @@ 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, + ) -> 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() + .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()], + }, + ) + } + fn rank_candidates( &self, intent: &AggIntent, diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml index fa125ac45..bce5ce677 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 = "264937ec4a06e260920c7e583bffed34cc07dd64" } diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index de6218d60..22e9e24c7 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 = "264937ec4a06e260920c7e583bffed34cc07dd64" } # Shared external (workspace) serde.workspace = true diff --git a/docs/developer_docs/control-plane/physical-compiler.md b/docs/developer_docs/control-plane/physical-compiler.md index 08814eabf..7fabda9fc 100644 --- a/docs/developer_docs/control-plane/physical-compiler.md +++ b/docs/developer_docs/control-plane/physical-compiler.md @@ -14,22 +14,24 @@ 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 @@ -42,8 +44,9 @@ PhysicalCompiler -------> PhysicalPlan - **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. @@ -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 @@ -269,13 +280,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 From 8311dd8d82bda2daec4caded05b77966737934d0 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 3 Sep 2026 13:49:05 -0600 Subject: [PATCH 4/6] feat(query): execute node-bound physical DAGs --- Cargo.lock | 6 +- control_plane/Cargo.toml | 6 +- control_plane/src/physical/compiler.rs | 116 +++- .../src/physical/post_asap/cost_model.rs | 12 + control_plane/src/query_parser/mod.rs | 2 +- control_plane/src/query_plan.rs | 537 ++++++++++-------- crates/asap_types/Cargo.toml | 2 +- data_plane/Cargo.toml | 2 +- .../query_engines/asap_query_engine/mod.rs | 1 + .../asap_query_engine/physical_dag.rs | 151 +++++ .../asap_query_engine/post_asap_readout.rs | 268 ++++++--- .../asap_query_engine/summary_executor.rs | 114 ++++ .../control-plane/physical-compiler.md | 29 +- 13 files changed, 904 insertions(+), 342 deletions(-) create mode 100644 data_plane/src/query_engines/asap_query_engine/physical_dag.rs diff --git a/Cargo.lock b/Cargo.lock index c0c4c3fb6..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=264937ec4a06e260920c7e583bffed34cc07dd64#264937ec4a06e260920c7e583bffed34cc07dd64" +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=264937ec4a06e260920c7e583bffed34cc07dd64#264937ec4a06e260920c7e583bffed34cc07dd64" +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=264937ec4a06e260920c7e583bffed34cc07dd64#264937ec4a06e260920c7e583bffed34cc07dd64" +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 2b25feb73..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 = "264937ec4a06e260920c7e583bffed34cc07dd64" } -asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "264937ec4a06e260920c7e583bffed34cc07dd64" } +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 = "264937ec4a06e260920c7e583bffed34cc07dd64" } +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/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 8779d8cb9..eaae0d01e 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -35,12 +35,13 @@ use crate::physical::colored_dag::emitter::{ }; use crate::physical::post_asap::cost_model::ControlPlaneCostModel; use crate::query_plan::{ - canonical_promql, FallbackPolicy, QueryPlan, QueryPlanEntry, QueryPlanNode, + canonical_promql, FallbackPolicy, InstantExecution, MaterializationBinding, PhysicalGrouping, + QueryPlan, QueryPlanEntry, }; use crate::types_v2::AccuracyTarget; use planner_types::pre_asap::Source; -pub const PLANNER_REVISION: &str = "264937ec4a06e260920c7e583bffed34cc07dd64"; +pub const PLANNER_REVISION: &str = "739753e33e096c01faccca8e7a1e3da5ad3aab9c"; #[derive(Debug, Clone)] pub struct PlanningQuery { @@ -423,13 +424,52 @@ impl PhysicalCompiler { .into(), }); } - let entry = QueryPlanEntry { - query_id: query.query_id.clone(), - canonical_promql: canonical.clone(), - root: QueryPlanNode::compile(&query.post_asap)?, - materializations: bound, - fallback: FallbackPolicy::ExactBackend, - }; + 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_kind, node_params| { + 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.kind == node_kind && m.params == node_params) + }) + .copied() + .ok_or_else(|| { + crate::query_plan::QueryPlanError::Invalid(format!( + "no exact physical binding for {node_kind:?}/{node_params:?}" + )) + })?; + Ok(MaterializationBinding { + materialization: fingerprint, + metric: metric.clone(), + kind: node_kind, + params: node_params, + 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(), @@ -452,6 +492,41 @@ impl PhysicalCompiler { } } +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`]. @@ -874,8 +949,27 @@ mod tests { .lookup("quantile_over_time( 0.99, m[1m] )") .expect("canonical QueryPlan lookup"); assert_eq!(entry.query_id, "q-quantile"); - assert_eq!(entry.materializations.len(), 1); - entry.root.to_summary_node().expect("executable DAG recipe"); + 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); diff --git a/control_plane/src/physical/post_asap/cost_model.rs b/control_plane/src/physical/post_asap/cost_model.rs index bed9195b4..244322bf7 100644 --- a/control_plane/src/physical/post_asap/cost_model.rs +++ b/control_plane/src/physical/post_asap/cost_model.rs @@ -226,6 +226,7 @@ impl CostModel for ControlPlaneCostModel { 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 @@ -241,12 +242,23 @@ impl CostModel for ControlPlaneCostModel { .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", + ), + ), }, ) } 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 index 0973b85b9..68d906d87 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -1,18 +1,15 @@ -//! Authoritative serving plan compiled from one selected post-ASAP DAG. +//! Authoritative backend-executable query DAG. //! -//! A query request may be parsed only to obtain its stable identity. Family -//! selection, materialization matching and execution-shape construction all -//! happen here, before publication. The data plane therefore never invokes -//! ASAPPlanner or searches the materialization catalog while serving. +//! 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::{ - ExactKind, ExactParams, GroupingStrategy, SketchKind, SketchQuery, SummaryExpr, - SummaryFamilyType, SummaryField, SummaryNode, SummarySchema, -}; -use planner_types::pre_asap::{ColumnRef, DataType, QueryExpr, Reduction}; +use planner_types::post_asap::{SketchQuery, SummaryExpr, SummaryFamilyType, SummaryNode}; +use planner_types::pre_asap::Reduction; use serde::{Deserialize, Serialize}; use thiserror::Error; @@ -40,10 +37,7 @@ impl QueryPlan { .ok_or(QueryPlanError::QueryNotPlanned(identity)) } - pub fn validate( - &self, - materializations: &BTreeSet, - ) -> Result<(), QueryPlanError> { + pub fn validate(&self, available: &BTreeSet) -> Result<(), QueryPlanError> { for (identity, entry) in &self.entries { if identity != &entry.canonical_promql { return Err(QueryPlanError::Invalid(format!( @@ -51,36 +45,147 @@ impl QueryPlan { entry.canonical_promql ))); } - if entry.materializations.is_empty() { - return Err(QueryPlanError::Invalid(format!( - "query `{identity}` has no bound materialization" - ))); - } - if let Some(missing) = entry - .materializations - .iter() - .find(|fingerprint| !materializations.contains(fingerprint)) - { - return Err(QueryPlanError::Invalid(format!( - "query `{identity}` references absent materialization {}", - missing.0 - ))); - } + 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: QueryPlanNode, - pub materializations: BTreeSet, + 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, + SummaryKind, + SummaryParams, + ) -> 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 { @@ -88,30 +193,53 @@ pub enum FallbackPolicy { Reject, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct MaterializationBinding { + pub materialization: PolicyFingerprint, + pub metric: String, + pub kind: SummaryKind, + pub params: SummaryParams, + /// 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 { - KeepPreAsap { - logical: serde_json::Value, - schema_names: Vec, - }, - SummaryAgg { - child: Box, - kind: SummaryKind, - params: SummaryParams, - col: ColumnRef, - reduction: Reduction, - schema_names: Vec, + ReadMaterialization { + binding: MaterializationBinding, }, SummaryEstimate { - summary_input: Box, + input: QueryNodeId, query: QueryReadout, - schema_names: Vec, }, SummaryMerge { - children: Vec, - schema_names: Vec, + 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)] @@ -121,7 +249,7 @@ pub enum QueryReadout { q: f64, }, PointCount { - key: ColumnRef, + key: planner_types::pre_asap::ColumnRef, value: Option, }, Cardinality, @@ -130,161 +258,121 @@ pub enum QueryReadout { }, } -#[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 MVP query executor: {0}")] - UnsupportedNode(String), - #[error("invalid QueryPlan: {0}")] - Invalid(String), - #[error("invalid serialized pre-ASAP leaf: {0}")] - InvalidLogicalLeaf(String), +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 }, + } + } } -/// Canonical textual identity used by both compilation and request lookup. -/// PromQL's parser/formatter normalizes whitespace and matcher formatting; -/// semantically different expressions retain different keys. -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())) +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 }, + } + } } -impl QueryPlanNode { - pub fn compile(node: &SummaryNode) -> Result { - let schema_names = node - .schema - .fields - .iter() - .map(|field| field.name.clone()) - .collect(); - match &node.expr { - SummaryExpr::KeepPreAsap(logical) => Ok(Self::KeepPreAsap { - logical: serde_json::to_value(logical.as_ref()) - .map_err(|error| QueryPlanError::InvalidLogicalLeaf(error.to_string()))?, - schema_names, - }), +struct DagCompiler<'a, F> { + next_id: u64, + nodes: BTreeMap, + seen: BTreeMap, + bind: &'a mut F, +} + +impl DagCompiler<'_, F> +where + F: FnMut( + &SummaryNode, + SummaryKind, + SummaryParams, + ) -> 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 { - child, family, - col, reduction, + child, .. } => { let (kind, params) = flatten_family(family)?; - Ok(Self::SummaryAgg { - child: Box::new(Self::compile(child)?), - kind, - params, - col: col.clone(), - reduction: reduction.clone(), - schema_names, - }) + let mut binding = (self.bind)(node, kind, params)?; + binding.output_grouping = physical_grouping(reduction, child)?; + QueryPlanNode::ReadMaterialization { binding } } SummaryExpr::SummaryEstimate { summary_input, query, - } => Ok(Self::SummaryEstimate { - summary_input: Box::new(Self::compile(summary_input)?), + } => QueryPlanNode::SummaryEstimate { + input: self.lower(summary_input)?, query: query.clone().into(), - schema_names, - }), - SummaryExpr::SummaryMerge { children } => Ok(Self::SummaryMerge { - children: children - .iter() - .map(|child| Self::compile(child)) - .collect::, _>>()?, - schema_names, - }), + }, + 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 { .. } => { - Err(QueryPlanError::UnsupportedNode("summary_join".into())) + return Err(QueryPlanError::UnsupportedNode("summary_join".into())) } SummaryExpr::SummarySubtract { .. } => { - Err(QueryPlanError::UnsupportedNode("summary_subtract".into())) + return Err(QueryPlanError::UnsupportedNode("summary_subtract".into())) } SummaryExpr::SummaryDelete { .. } => { - Err(QueryPlanError::UnsupportedNode("summary_delete".into())) + return Err(QueryPlanError::UnsupportedNode("summary_delete".into())) } - } - } - - /// Rebuild the planner-owned semantic node without making a planning - /// decision. Schema names are retained because group IDs are positional; - /// other schema/guarantee details are irrelevant to execution. - pub fn to_summary_node(&self) -> Result, QueryPlanError> { - let (expr, names) = match self { - Self::KeepPreAsap { - logical, - schema_names, - } => { - let logical: QueryExpr = serde_json::from_value(logical.clone()) - .map_err(|error| QueryPlanError::InvalidLogicalLeaf(error.to_string()))?; - (SummaryExpr::KeepPreAsap(Rc::new(logical)), schema_names) - } - Self::SummaryAgg { - child, - kind, - params, - col, - reduction, - schema_names, - } => ( - SummaryExpr::SummaryAgg { - child: child.to_summary_node()?, - family: expand_family(*kind, params.clone())?, - col: col.clone(), - reduction: reduction.clone(), - grouping: GroupingStrategy::default(), - }, - schema_names, - ), - Self::SummaryEstimate { - summary_input, - query, - schema_names, - } => ( - SummaryExpr::SummaryEstimate { - summary_input: summary_input.to_summary_node()?, - query: query.clone().into(), - }, - schema_names, - ), - Self::SummaryMerge { - children, - schema_names, - } => ( - SummaryExpr::SummaryMerge { - children: children - .iter() - .map(Self::to_summary_node) - .collect::, _>>()?, - }, - schema_names, - ), }; - Ok(Rc::new(SummaryNode { - expr, - schema: execution_schema(names), - guarantee: None, - })) + self.nodes.insert(id, physical); + Ok(id) } } -fn execution_schema(names: &[String]) -> SummarySchema { - SummarySchema { - fields: names - .iter() - .map(|name| SummaryField { - name: name.clone(), - dtype: SummaryFamilyType::Plain(DataType::Utf8), - nullable: true, - }) - .collect(), - time_index: None, - } +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)) } fn flatten_family( @@ -303,61 +391,22 @@ fn flatten_family( } } -fn expand_family( - kind: SummaryKind, - params: SummaryParams, -) -> Result { - if kind.is_exact() { - let exact = match (kind, params) { - (SummaryKind::Sum, SummaryParams::Sum) => (ExactKind::Sum, ExactParams::Sum), - (SummaryKind::Count, SummaryParams::Count) => (ExactKind::Count, ExactParams::Count), - (SummaryKind::MinMax, SummaryParams::MinMax) => { - (ExactKind::MinMax, ExactParams::MinMax) - } - (SummaryKind::Increase, SummaryParams::Increase) => { - (ExactKind::Increase, ExactParams::Increase) - } - (SummaryKind::Rate, SummaryParams::Rate) => (ExactKind::Rate, ExactParams::Rate), - (kind, params) => { - return Err(QueryPlanError::Invalid(format!( - "mismatched exact family {kind:?}/{params:?}" - ))) - } - }; - return Ok(SummaryFamilyType::ExactAggregate(exact.0, exact.1)); - } - let algorithm = kind - .as_sketch_kind() - .ok_or_else(|| QueryPlanError::Invalid(format!("not a sketch kind: {kind:?}")))?; - let sketch_params = params - .as_sketch_params() - .ok_or_else(|| QueryPlanError::Invalid(format!("not sketch params: {params:?}")))?; - Ok(SummaryFamilyType::Sketch( - SketchKind::new(algorithm, sketch_params), - GroupingStrategy::default(), - )) -} - -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 }, - } - } +#[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), } -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 }, - } - } +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)] @@ -371,4 +420,32 @@ mod tests { 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 bce5ce677..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 = "264937ec4a06e260920c7e583bffed34cc07dd64" } +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 22e9e24c7..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 = "264937ec4a06e260920c7e583bffed34cc07dd64" } +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/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_readout.rs b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs index 1911fb736..4b8e9f602 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; @@ -95,18 +97,7 @@ pub fn execute_query_plan_readout( t1_ms: u64, is_cumulative: bool, ) -> Result { - let node = entry - .root - .to_summary_node() - .map_err(|error| LoweringSkip::InvalidQueryPlan(error.to_string()))?; - execute_bound_post_asap( - index, - &node, - t0_ms, - t1_ms, - is_cumulative, - entry.materializations.clone(), - ) + execute_physical_query_plan(index, entry, t0_ms, t1_ms, is_cumulative) } pub fn execute_query_plan_instant( @@ -114,28 +105,170 @@ pub fn execute_query_plan_instant( entry: &control_plane::query_plan::QueryPlanEntry, now_ms: u64, ) -> Result<(PostAsapReadoutOutcome, u64), LoweringSkip> { - const DEFAULT_LOOKBACK_MS: u64 = 5 * 60 * 1000; - let node = entry - .root - .to_summary_node() - .map_err(|error| LoweringSkip::InvalidQueryPlan(error.to_string()))?; - let hints = execution_hints(&node); - let t0_ms = if hints.full_history { + let t0_ms = if entry.instant.full_history { 0 } else { - now_ms.saturating_sub(hints.lookback_ms.unwrap_or(DEFAULT_LOOKBACK_MS)) + now_ms.saturating_sub(entry.instant.lookback_ms) }; - let outcome = execute_bound_post_asap( + let outcome = execute_physical_query_plan( index, - &node, + entry, t0_ms, now_ms, - hints.cumulative_readout, - entry.materializations.clone(), + entry.instant.cumulative_readout, )?; Ok((outcome, t0_ms)) } +#[derive(Clone)] +enum PhysicalQueryOutput { + State( + Vec<( + BTreeMap, + GroupState, + asap_types::SummaryKind, + asap_types::SummaryParams, + )>, + ), + 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("summary merge inputs have different families")] + FamilyMismatch, + #[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 + .into_iter() + .map(|(key, state)| { + (key, state, binding.kind.clone(), binding.params.clone()) + }) + .collect(), + )) + } + 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 family: Option<(asap_types::SummaryKind, asap_types::SummaryParams)> = None; + 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, kind, params) in groups { + match &family { + None => family = Some((kind.clone(), params.clone())), + Some((expected_kind, expected_params)) + if expected_kind == kind && expected_params == params => {} + Some(_) => return Err(PhysicalNodeError::FamilyMismatch), + } + by_group.entry(key.clone()).or_default().push(state.clone()); + } + } + let (kind, params) = family.ok_or(PhysicalNodeError::ExpectedState)?; + by_group + .into_iter() + .map(|(key, states)| { + self.context + .merge_bound_states(states) + .map(|state| (key, state, kind.clone(), params.clone())) + .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. @@ -217,59 +350,6 @@ fn execute_planned_post_asap( } } -fn execute_bound_post_asap( - index: &SketchStore, - node: &planner_types::post_asap::SummaryNode, - t0_ms: u64, - t1_ms: u64, - is_cumulative: bool, - allowed_materializations: std::collections::BTreeSet, -) -> Result { - let ctx = QueryExecutionContext { - index, - t0_ms, - t1_ms, - is_cumulative, - allowed_materializations: Some(allowed_materializations), - }; - convert_execution_result(execute(node, &ctx), t1_ms) -} - -fn convert_execution_result( - result: Result< - ExecOutcome>, - crate::query_engines::asap_query_engine::summary_exec::ExecError< - crate::query_engines::asap_query_engine::summary_executor::SummaryExecutorError, - >, - >, - t1_ms: u64, -) -> Result { - match result { - Ok(ExecOutcome::Value(values)) => { - let mut coverage: Option<(u64, u64)> = 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 }) - } - Ok(ExecOutcome::State(groups)) => { - let mut coverage: Option<(u64, u64)> = None; - let mut series = Vec::new(); - for (group_key, state, _family) in &groups { - fold_coverage(&mut coverage, state.exact_coverage()); - let Some(value) = state.exact_value(&None) else { - continue; - }; - series.push((group_key.clone(), vec![(t1_ms as i64, value)])); - } - Ok(PostAsapReadoutOutcome { series, coverage }) - } - Err(error) => Err(LoweringSkip::ExecuteFailed(format!("{error:?}"))), - } -} - /// `SummaryValue::Points`/`TopK` -> `ASAPTierResult.series`'s row shape. /// `TopK`'s ranked-list-per-timestamp shape is pivoted into one row per /// item (each row = the group's label map plus an `item` label, one point @@ -415,14 +495,30 @@ mod tests { 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 entry = control_plane::query_plan::QueryPlanEntry { - query_id: "q-cardinality".into(), - canonical_promql: control_plane::query_plan::canonical_promql("count(unique_users)") - .unwrap(), - root: control_plane::query_plan::QueryPlanNode::compile(&node).unwrap(), - materializations: [asap_types::PolicyFingerprint(123)].into_iter().collect(), - fallback: control_plane::query_plan::FallbackPolicy::ExactBackend, - }; + 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, kind, params| { + Ok(control_plane::query_plan::MaterializationBinding { + materialization: asap_types::PolicyFingerprint(123), + metric: "unique_users".into(), + kind, + params, + 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()); 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 1d506da64..ff46b2d3a 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,120 @@ 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 { kind, config, .. } => { + summary_params_match(&binding.kind, &binding.params, *kind, config) + .then(|| to_delta_kind(*kind, config)) + .flatten() + .map(Candidate::Sketch) + } + AggKind::ExactAgg { agg_type, .. } => { + exact_agg_kind_match(&binding.kind, &binding.params, *agg_type) + .then_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 diff --git a/docs/developer_docs/control-plane/physical-compiler.md b/docs/developer_docs/control-plane/physical-compiler.md index 7fabda9fc..40129850a 100644 --- a/docs/developer_docs/control-plane/physical-compiler.md +++ b/docs/developer_docs/control-plane/physical-compiler.md @@ -9,7 +9,7 @@ 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. @@ -35,10 +35,10 @@ 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 @@ -48,7 +48,7 @@ PhysicalCompiler -------> PhysicalPlan 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 @@ -127,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 } ``` @@ -153,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 From 3714ee9c4112f012e9180dcb71fa7e33a77e4e23 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 3 Sep 2026 20:54:09 -0600 Subject: [PATCH 5/6] fix(stack): align QueryPlan with canonical planner families --- control_plane/src/physical/compiler.rs | 17 ++++---- control_plane/src/query_plan.rs | 40 ++++++------------- .../asap_query_engine/post_asap_readout.rs | 39 +++++------------- .../asap_query_engine/summary_executor.rs | 14 ++----- .../storage_engines/sketch_db/index/mod.rs | 5 +-- 5 files changed, 34 insertions(+), 81 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index eaae0d01e..676b1bfba 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -397,8 +397,10 @@ impl PhysicalCompiler { query_id: query.query_id.clone(), reason: "selected plan has no executable sketch materialization/readout".into(), })?; - let kind = asap_types::SummaryKind::from(selected.kind); - let params = asap_types::SummaryParams::from(selected.params); + 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"), @@ -410,8 +412,7 @@ impl PhysicalCompiler { 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.kind == kind - && materialization.params == params + && materialization.family == family && materialization.window.size_ms == query.window_secs.saturating_mul(1_000) && materialization.group_by == query.group_by) .then_some(route.materialization) @@ -434,7 +435,7 @@ impl PhysicalCompiler { cumulative_readout: true, }, FallbackPolicy::ExactBackend, - |node, node_kind, node_params| { + |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(), @@ -451,19 +452,17 @@ impl PhysicalCompiler { backend_plan .materializations .get(fingerprint) - .is_some_and(|m| m.kind == node_kind && m.params == node_params) + .is_some_and(|m| &m.family == node_family) }) .copied() .ok_or_else(|| { crate::query_plan::QueryPlanError::Invalid(format!( - "no exact physical binding for {node_kind:?}/{node_params:?}" + "no exact physical binding for {node_family:?}" )) })?; Ok(MaterializationBinding { materialization: fingerprint, metric: metric.clone(), - kind: node_kind, - params: node_params, sid_grouping: query.group_by.clone(), output_grouping: PhysicalGrouping::Reduce(query.group_by.clone()), window_ms: query.window_secs.saturating_mul(1_000), diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index 68d906d87..c80d07fad 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -13,7 +13,7 @@ use planner_types::pre_asap::Reduction; use serde::{Deserialize, Serialize}; use thiserror::Error; -use asap_types::{PolicyFingerprint, SummaryKind, SummaryParams}; +use asap_types::PolicyFingerprint; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(deny_unknown_fields)] @@ -88,8 +88,7 @@ impl QueryPlanEntry { where F: FnMut( &SummaryNode, - SummaryKind, - SummaryParams, + &SummaryFamilyType, ) -> Result, { let mut compiler = DagCompiler { @@ -198,8 +197,6 @@ pub enum FallbackPolicy { pub struct MaterializationBinding { pub materialization: PolicyFingerprint, pub metric: String, - pub kind: SummaryKind, - pub params: SummaryParams, /// Exact label-key layout of the stored materialization. pub sid_grouping: Vec, /// Query operator grouping applied while folding those SIDs. @@ -289,11 +286,7 @@ struct DagCompiler<'a, F> { impl DagCompiler<'_, F> where - F: FnMut( - &SummaryNode, - SummaryKind, - SummaryParams, - ) -> Result, + F: FnMut(&SummaryNode, &SummaryFamilyType) -> Result, { fn lower(&mut self, node: &Rc) -> Result { let identity = Rc::as_ptr(node) as usize; @@ -313,8 +306,15 @@ where child, .. } => { - let (kind, params) = flatten_family(family)?; - let mut binding = (self.bind)(node, kind, params)?; + 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 } } @@ -375,22 +375,6 @@ fn physical_grouping( Ok(PhysicalGrouping::Reduce(names)) } -fn flatten_family( - family: &SummaryFamilyType, -) -> Result<(SummaryKind, SummaryParams), QueryPlanError> { - match family { - SummaryFamilyType::ExactAggregate(kind, params) => { - Ok((kind.clone().into(), params.clone().into())) - } - SummaryFamilyType::Sketch(kind, _) => { - Ok((kind.clone().into(), kind.params().clone().into())) - } - other => Err(QueryPlanError::UnsupportedNode(format!( - "summary family {other:?}" - ))), - } -} - #[derive(Debug, Error)] pub enum QueryPlanError { #[error("invalid PromQL query identity: {0}")] 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 4b8e9f602..ba9538c75 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 @@ -122,14 +122,7 @@ pub fn execute_query_plan_instant( #[derive(Clone)] enum PhysicalQueryOutput { - State( - Vec<( - BTreeMap, - GroupState, - asap_types::SummaryKind, - asap_types::SummaryParams, - )>, - ), + State(Vec<(BTreeMap, GroupState)>), Value(Vec<(BTreeMap, SummaryValue)>), } @@ -139,8 +132,6 @@ enum PhysicalNodeError { Store(SummaryExecutorError), #[error("node expected summary state input")] ExpectedState, - #[error("summary merge inputs have different families")] - FamilyMismatch, #[error("physical fallback requested: {0}")] Fallback(String), } @@ -165,14 +156,7 @@ impl QueryNodeRuntime for PhysicalQueryRuntime<'_> { .context .read_bound_materialization(binding) .map_err(PhysicalNodeError::Store)?; - Ok(PhysicalQueryOutput::State( - groups - .into_iter() - .map(|(key, state)| { - (key, state, binding.kind.clone(), binding.params.clone()) - }) - .collect(), - )) + Ok(PhysicalQueryOutput::State(groups)) } QueryPlanNode::SummaryEstimate { query, .. } => { let [PhysicalQueryOutput::State(groups)] = inputs else { @@ -181,7 +165,7 @@ impl QueryNodeRuntime for PhysicalQueryRuntime<'_> { let query: planner_types::post_asap::SketchQuery = query.clone().into(); groups .iter() - .map(|(key, state, _, _)| { + .map(|(key, state)| { self.context .readout_bound(state, &query) .map(|value| (key.clone(), value)) @@ -191,30 +175,25 @@ impl QueryNodeRuntime for PhysicalQueryRuntime<'_> { .map(PhysicalQueryOutput::Value) } QueryPlanNode::SummaryMerge { .. } => { - let mut family: Option<(asap_types::SummaryKind, asap_types::SummaryParams)> = None; 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, kind, params) in groups { - match &family { - None => family = Some((kind.clone(), params.clone())), - Some((expected_kind, expected_params)) - if expected_kind == kind && expected_params == params => {} - Some(_) => return Err(PhysicalNodeError::FamilyMismatch), - } + for (key, state) in groups { by_group.entry(key.clone()).or_default().push(state.clone()); } } - let (kind, params) = family.ok_or(PhysicalNodeError::ExpectedState)?; + 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, kind.clone(), params.clone())) + .map(|state| (key, state)) .map_err(PhysicalNodeError::Store) }) .collect::, _>>() @@ -258,7 +237,7 @@ fn execute_physical_query_plan( PhysicalQueryOutput::State(groups) => { let mut coverage = None; let mut series = Vec::new(); - for (group_key, state, _, _) in &groups { + 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)])); 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 ff46b2d3a..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 @@ -338,16 +338,10 @@ impl QueryExecutionContext<'_> { return None; } match &meta.agg_kind { - AggKind::Sketch { kind, config, .. } => { - summary_params_match(&binding.kind, &binding.params, *kind, config) - .then(|| to_delta_kind(*kind, config)) - .flatten() - .map(Candidate::Sketch) - } - AggKind::ExactAgg { agg_type, .. } => { - exact_agg_kind_match(&binding.kind, &binding.params, *agg_type) - .then_some(Candidate::ExactAgg(*agg_type)) - } + AggKind::Sketch { + algorithm, config, .. + } => to_delta_kind(algorithm.clone(), config).map(Candidate::Sketch), + AggKind::ExactAgg { agg_type, .. } => Some(Candidate::ExactAgg(*agg_type)), } }) .flatten(); 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 { From f6b0a6a994d38b83bcd812336bd0c6cf5f61a385 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 3 Sep 2026 21:24:52 -0600 Subject: [PATCH 6/6] fix(query-plan): update bound-plan test fixture --- .../src/query_engines/asap_query_engine/post_asap_readout.rs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) 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 ba9538c75..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 @@ -485,12 +485,10 @@ mod tests { cumulative_readout: true, }, control_plane::query_plan::FallbackPolicy::ExactBackend, - |_node, kind, params| { + |_node, _family| { Ok(control_plane::query_plan::MaterializationBinding { materialization: asap_types::PolicyFingerprint(123), metric: "unique_users".into(), - kind, - params, sid_grouping: vec!["service".into()], output_grouping: control_plane::query_plan::PhysicalGrouping::PerEntity, window_ms: 60_000,