From b00ee22a790815dcb71839df4f74ebf20dae7e55 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 01:08:16 -0600 Subject: [PATCH 1/6] Verify MetricsQL acceleration against VictoriaMetrics --- control_plane/src/physical/compiler.rs | 76 +++- .../metricsql_binder.rs | 14 + .../tests/victoriametrics_accelerated_e2e.rs | 336 ++++++++++++++++++ 3 files changed, 425 insertions(+), 1 deletion(-) create mode 100644 data_plane/tests/victoriametrics_accelerated_e2e.rs diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index ab20044c3..9d8f78b73 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -18,7 +18,7 @@ use planner_types::post_asap::{ SketchQuery, SummaryExpr, SummaryFamilyType, SummaryMaintenanceLifecycle, SummaryMaintenanceMode, SummaryNode, SummaryWindowFramework, }; -use planner_types::pre_asap::QueryExpr; +use planner_types::pre_asap::{AggIntent, QueryExpr, Reduction}; use planner_types::workload::{ AccuracyRequirement, DataArrival, DataWorkload, DurationMs, Evidence, EvidenceSource, Predictability, Query, QueryLanguage, QueryRecurrence, QueryRequirements, QueryTimeScope, @@ -2047,6 +2047,41 @@ fn preserve_invalid_exact_fallback_roots( Ok(()) } +/// Reject MetricsQL shapes whose VictoriaMetrics series-reduction semantics +/// are not represented faithfully by the current canonical executor. +pub fn validate_metricsql_acceleration_shape(expr: &QueryExpr) -> Result<(), &'static str> { + let QueryExpr::Aggregate { + reduction: Reduction::Reduce(_), + measures: outer_measures, + child, + .. + } = expr + else { + return Ok(()); + }; + let QueryExpr::Aggregate { + reduction: Reduction::PerEntity, + measures: inner_measures, + child: inner_child, + .. + } = child.as_ref() + else { + return Ok(()); + }; + if outer_measures.len() == 1 + && matches!(outer_measures[0], AggIntent::Sum { .. }) + && inner_measures.len() == 1 + && matches!(inner_child.as_ref(), QueryExpr::TimeRange { .. }) + { + return match inner_measures[0] { + AggIntent::Rate => Err("nested sum(rate(...)) has VictoriaMetrics extrapolation semantics that the shared DAG cannot yet preserve"), + AggIntent::Increase => Err("nested sum(increase(...)) has VictoriaMetrics extrapolation semantics that the shared DAG cannot yet preserve"), + _ => Ok(()), + }; + } + Ok(()) +} + impl PhysicalCompiler { pub fn compile( &self, @@ -2061,6 +2096,20 @@ impl PhysicalCompiler { request: PlanningRequest, environment: DeploymentEnvironment, ) -> Result { + for query in &request.queries { + let expr = asap_frontend_metricsql::lower_metricsql( + &query.query_string, + query.accuracy.clone(), + ) + .map_err(|error| CompileError::Query { + query_id: query.query_id.clone(), + reason: format!("frontend.metricsql.lowering: {error}"), + })?; + validate_metricsql_acceleration_shape(&expr).map_err(|reason| CompileError::Query { + query_id: query.query_id.clone(), + reason: format!("frontend.metricsql.unsupported: {reason}"), + })?; + } self.compile_language(request, environment, true) } @@ -4402,6 +4451,31 @@ mod tests { assert_eq!(entry.language, crate::query_plan::QueryLanguage::MetricsQl); } + #[test] + fn metricsql_non_equivalent_counter_rollups_fail_closed_before_publication() { + for (query, stage) in [ + ("sum(rate(m[5s]))", "rate"), + ("sum(increase(m[5s]))", "increase"), + ] { + let mut workload = request("vm-q", query); + let accuracy = workload.queries[0].accuracy.clone(); + let canonical = + asap_frontend_metricsql::lower_metricsql(query, accuracy.clone()).unwrap(); + workload.queries[0].post_asap = crate::planner_selection::select_summary( + &canonical, + &crate::physical::post_asap::cost_model::ForcedFamilyCostModel::new( + accuracy, + planner_types::post_asap::SketchAlgorithm::Kll, + ), + ) + .unwrap(); + let error = PhysicalCompiler + .compile_metricsql(workload, environment(10_000)) + .unwrap_err(); + assert!(error.to_string().contains(stage), "{error}"); + } + } + #[test] fn hybrid_erp_capability_miss_preserves_exact_subtree() { let mut workload = request("q", "quantile_over_time(0.9, m[1m])"); diff --git a/data_plane/src/query_engines/asap_victoriametrics_query_engine/metricsql_binder.rs b/data_plane/src/query_engines/asap_victoriametrics_query_engine/metricsql_binder.rs index 7139ccf65..d6d5ae2e1 100644 --- a/data_plane/src/query_engines/asap_victoriametrics_query_engine/metricsql_binder.rs +++ b/data_plane/src/query_engines/asap_victoriametrics_query_engine/metricsql_binder.rs @@ -26,6 +26,8 @@ pub fn bind_metricsql( ) -> Result { let canonical = asap_frontend_metricsql::lower_metricsql(query, accuracy.clone()) .map_err(|error| MetricsQlBindingError::Unsupported(error.to_string()))?; + control_plane::physical::compiler::validate_metricsql_acceleration_shape(&canonical) + .map_err(|reason| MetricsQlBindingError::Unsupported(reason.into()))?; let physical = bind_query_expr(&canonical, accuracy) .map_err(|error| MetricsQlBindingError::Physical(error.to_string()))?; Ok(MetricsQlBinding { @@ -80,4 +82,16 @@ mod tests { Err(MetricsQlBindingError::Unsupported(_)) )); } + + #[test] + fn empirically_non_equivalent_nested_rollups_fail_closed() { + for query in ["sum(rate(foo[5s]))", "sum(increase(foo[5s]))"] { + assert!(matches!( + bind_metricsql(query, accuracy()), + Err(MetricsQlBindingError::Unsupported(_)) + )); + } + bind_metricsql("sum(sum_over_time(foo[5s]))", accuracy()) + .expect("nested sum rollup is represented exactly after complete ingestion"); + } } diff --git a/data_plane/tests/victoriametrics_accelerated_e2e.rs b/data_plane/tests/victoriametrics_accelerated_e2e.rs new file mode 100644 index 000000000..cd1479a89 --- /dev/null +++ b/data_plane/tests/victoriametrics_accelerated_e2e.rs @@ -0,0 +1,336 @@ +//! Real VictoriaMetrics differential for the published MetricsQL DAG path. +//! +//! Run with a VictoriaMetrics instance, for example: +//! `VICTORIAMETRICS_URL=http://127.0.0.1:18428 cargo test -p data_plane --test victoriametrics_accelerated_e2e -- --ignored --nocapture`. + +use data_plane::drivers::ingest::prometheus_remote_write::{ + Label, Sample, TimeSeries, WriteRequest, +}; +use prost::Message; +use serde_json::Value; +use std::{ + io::Write, + net::TcpListener, + process::{Child, Command, Stdio}, + time::Duration, +}; + +struct ChildGuard(Child); +impl Drop for ChildGuard { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} +fn port() -> u16 { + TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port() +} +fn value(body: &Value) -> f64 { + body["data"]["result"][0]["value"][1] + .as_str() + .unwrap() + .parse() + .unwrap() +} +async fn write(client: &reqwest::Client, base: &str, metric: &str, samples: &[(i64, f64)]) { + let request = WriteRequest { + timeseries: vec![TimeSeries { + labels: vec![Label { + name: "__name__".into(), + value: metric.into(), + }], + samples: samples + .iter() + .map(|(timestamp, value)| Sample { + timestamp: *timestamp, + value: *value, + }) + .collect(), + exemplars: vec![], + histograms: vec![], + }], + }; + let body = snap::raw::Encoder::new() + .compress_vec(&request.encode_to_vec()) + .unwrap(); + client + .post(format!("{base}/api/v1/write")) + .header("content-encoding", "snappy") + .header("content-type", "application/x-protobuf") + .header("x-prometheus-remote-write-version", "0.1.0") + .body(body) + .send() + .await + .unwrap() + .error_for_status() + .unwrap(); +} + +#[tokio::test] +#[ignore = "requires a real VictoriaMetrics via VICTORIAMETRICS_URL"] +async fn real_victoriametrics_ingest_published_dag_is_differential() { + let metric = format!("vm_accelerated_gauge_{}", std::process::id()); + let query = format!("sum(sum_over_time({metric}[5s]))"); + let unsupported_query = format!("sum(increase({metric}[5s]))"); + let upstream = std::env::var("VICTORIAMETRICS_URL").expect("VICTORIAMETRICS_URL"); + let client = reqwest::Client::new(); + let now_ms = chrono::Utc::now().timestamp_millis(); + let base = now_ms / 20_000 * 20_000 - 20_000; + let samples = [ + (base + 500, 1.), + (base + 1700, 2.), + (base + 2900, 3.), + (base + 4200, 4.), + (base + 5400, 5.), + (base + 6600, 6.), + (base + 8100, 7.), + (base + 9400, 8.), + (base + 10500, 9.), + ]; + let at = (base + 10_000) as f64 / 1000.; + let import = samples + .iter() + .map(|(ts, v)| format!("{metric} {v} {}\n", *ts as f64 / 1000.0)) + .collect::(); + client + .post(format!("{upstream}/api/v1/import/prometheus")) + .body(import) + .send() + .await + .unwrap() + .error_for_status() + .unwrap(); + client + .post(format!("{upstream}/internal/force_flush")) + .send() + .await + .unwrap() + .error_for_status() + .unwrap(); + let mut direct_unsupported = None; + for _ in 0..200 { + let response = client + .get(format!("{upstream}/api/v1/query")) + .query(&[ + ("query", unsupported_query.as_str()), + ("time", &at.to_string()), + ]) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + if response["data"]["result"] + .as_array() + .is_some_and(|r| !r.is_empty()) + && (value(&response) - 4.0).abs() < 1e-9 + { + direct_unsupported = Some(response); + break; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + let direct_unsupported = direct_unsupported.expect("VictoriaMetrics import became visible"); + let mut fixture: Value = serde_json::from_str(include_str!( + "../../docs/examples/asapquery-compatibility-demo-snapshot.json" + )) + .unwrap(); + let mut demand = fixture["query_workload"]["repeating_queries"][0].clone(); + demand["query"] = query.clone().into(); + demand["time_selection"]["lookback"] = 5_000.into(); + fixture["query_workload"]["repeating_queries"] = serde_json::json!([demand]); + fixture["implementation"]["topk_evidence"] = serde_json::json!({}); + let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = + serde_json::from_value(fixture).unwrap(); + let (mut request, environment) = snapshot.planning_request().unwrap(); + request.hybrid_execution = false; + let accuracy = request.queries[0].accuracy.clone(); + let expr = asap_frontend_metricsql::lower_metricsql(&query, accuracy.clone()).unwrap(); + request.queries[0].query_string = query.clone(); + request.queries[0].post_asap = control_plane::planner_selection::select_summary( + &expr, + &control_plane::physical::post_asap::cost_model::ForcedFamilyCostModel::new( + accuracy, + planner_types::post_asap::SketchAlgorithm::Kll, + ), + ) + .unwrap(); + let plan = control_plane::physical::compiler::PhysicalCompiler + .compile_metricsql(request, environment) + .unwrap(); + assert!(plan + .query_plan + .entries + .values() + .any(|entry| { entry.language == control_plane::query_plan::QueryLanguage::MetricsQl })); + let binding = plan + .query_plan + .entries + .values() + .find(|entry| entry.language == control_plane::query_plan::QueryLanguage::MetricsQl) + .and_then(|entry| entry.materialization_bindings().into_iter().next()) + .expect("published MetricsQL DAG has a bound SummaryScan"); + assert_eq!(binding.readout_lookback_ms, Some(5_000)); + let artifact = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest { + summary_catalog: plan.summary_catalog, + collector_plans: plan.collector_plans, + precompute_plan: plan.precompute_plan, + transmission_plan: plan.transmission_plan, + query_plan: plan.query_plan, + storage_routing: None, + adaptation_evidence: vec![], + }; + let artifact: data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest = + serde_json::from_value(serde_json::to_value(&artifact).unwrap()).unwrap(); + assert_eq!( + artifact + .query_plan + .entries + .values() + .find(|entry| entry.language == control_plane::query_plan::QueryLanguage::MetricsQl) + .and_then(|entry| entry.materialization_bindings().into_iter().next()) + .and_then(|binding| binding.readout_lookback_ms), + Some(5_000), + "MetricsQL effective lookback survives publication serialization" + ); + let mut file = tempfile::NamedTempFile::new().unwrap(); + serde_json::to_writer(&mut file, &artifact).unwrap(); + file.flush().unwrap(); + let query_port = port(); + let vm_port = port(); + let output = tempfile::tempdir().unwrap(); + let mut child = ChildGuard( + Command::new(env!("CARGO_BIN_EXE_data_plane")) + .args([ + "--profile", + "asapquery", + "--physical-plan", + file.path().to_str().unwrap(), + "--http-port", + &query_port.to_string(), + "--victoriametrics-http-port", + &vm_port.to_string(), + "--victoriametrics-url", + &upstream, + "--forward-unsupported-queries", + "--prometheus-server", + &upstream, + "--output-dir", + output.path().to_str().unwrap(), + "--precompute-allowed-lateness-ms", + "0", + "--precompute-flush-interval-ms", + "100", + ]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(), + ); + let ingest = format!("http://127.0.0.1:{query_port}"); + let proxy = format!("http://127.0.0.1:{vm_port}"); + let mut ready = false; + for _ in 0..100 { + if let Some(status) = child.0.try_wait().unwrap() { + panic!("backend exited: {status}"); + } + if client + .get(format!("{proxy}/api/v1/health")) + .send() + .await + .is_ok_and(|r| r.status().is_success()) + { + ready = true; + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + assert!(ready, "VictoriaMetrics listener did not become ready"); + write(&client, &ingest, &metric, &samples[..8]).await; + write(&client, &ingest, &metric, &samples[8..]).await; + let mut fallback_pair = None; + for _ in 0..200 { + let fallback_response = client + .get(format!("{proxy}/api/v1/query")) + .query(&[ + ("query", unsupported_query.as_str()), + ("time", &at.to_string()), + ]) + .send() + .await + .unwrap(); + assert_eq!( + fallback_response + .headers() + .get("x-asap-execution") + .and_then(|value| value.to_str().ok()), + Some("exact_fallback") + ); + let fallback = fallback_response.json::().await.unwrap(); + if fallback["data"]["result"] + .as_array() + .is_some_and(|r| !r.is_empty()) + && (value(&fallback) - 4.0).abs() < 1e-9 + { + fallback_pair = Some(fallback); + break; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + let fallback = fallback_pair.expect("fallback query became visible"); + assert_eq!(value(&fallback), value(&direct_unsupported)); + let mut accelerated = None; + for _ in 0..60 { + let response = client + .get(format!("{proxy}/api/v1/query")) + .query(&[("query", query.as_str()), ("time", &at.to_string())]) + .send() + .await + .unwrap(); + if response + .headers() + .get("x-asap-execution") + .and_then(|v| v.to_str().ok()) + == Some("warm") + { + accelerated = Some(response.json::().await.unwrap()); + break; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + let accelerated = accelerated.expect("published MetricsQL DAG reached shared warm executor"); + let mut direct = None; + for _ in 0..200 { + let response: Value = client + .get(format!("{upstream}/api/v1/query")) + .query(&[("query", query.as_str()), ("time", &at.to_string())]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + if response["data"]["result"] + .as_array() + .is_some_and(|rows| !rows.is_empty()) + && (value(&response) - 26.0).abs() < 1e-9 + && (value(&response) - value(&accelerated)).abs() < 1e-9 + { + direct = Some(response); + break; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + let direct = + direct.expect("VictoriaMetrics import became query-visible with differential result"); + assert!( + (value(&accelerated) - value(&direct)).abs() < 1e-9, + "accelerated={accelerated}, direct={direct}" + ); + drop(child); +} From 02adac6054b564ed232e0e2df62cbff9256b45c8 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 01:23:42 -0600 Subject: [PATCH 2/6] Restore stable PromQL query plan contract --- control_plane/src/query_plan.rs | 465 ++++++++------------------------ 1 file changed, 106 insertions(+), 359 deletions(-) diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index 28dedf017..599718c91 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -22,66 +22,23 @@ use asap_types::{sds::SummaryDefinitionId, PolicyFingerprint}; pub struct QueryPlan { pub plan_id: u64, pub plan_version: u64, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub clickhouse_context: Option, pub entries: BTreeMap, } -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] -#[serde(deny_unknown_fields)] -pub struct ClickHousePlanningContext { - pub tables: std::collections::HashMap, - pub accuracy: planner_types::types::AccuracyTarget, -} - -#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] -#[serde(deny_unknown_fields)] -pub struct FixedEvaluationRange { - pub start_ms: u64, - pub end_ms: u64, - pub cumulative: bool, -} - impl QueryPlan { pub fn empty() -> Self { Self { plan_id: 0, plan_version: 0, - clickhouse_context: None, entries: BTreeMap::new(), } } pub fn lookup(&self, promql: &str) -> Result<&QueryPlanEntry, QueryPlanError> { let identity = canonical_promql(promql)?; - self.lookup_canonical(QueryLanguage::PromQl, &identity) - } - - pub fn lookup_canonical( - &self, - language: QueryLanguage, - identity: &str, - ) -> Result<&QueryPlanEntry, QueryPlanError> { - let key = Self::catalog_key(language, identity); self.entries - .get(&key) - .filter(|entry| entry.language == language) - .ok_or_else(|| QueryPlanError::QueryNotPlanned(identity.into())) - } - - pub fn catalog_key(language: QueryLanguage, identity: &str) -> String { - match language { - QueryLanguage::PromQl => identity.to_owned(), - QueryLanguage::MetricsQl => format!("metricsql:{identity}"), - QueryLanguage::ClickHouseSql => format!("clickhouse:{identity}"), - } - } - - pub fn lookup_clickhouse( - &self, - canonical_sql: &str, - ) -> Result<&QueryPlanEntry, QueryPlanError> { - self.lookup_canonical(QueryLanguage::ClickHouseSql, canonical_sql) + .get(&identity) + .ok_or(QueryPlanError::QueryNotPlanned(identity)) } /// Validate semantic bindings against the authoritative snapshot before use. @@ -164,55 +121,18 @@ impl QueryPlan { )); } for (identity, entry) in &self.entries { - let expected = Self::catalog_key(entry.language, &entry.canonical_query); - if identity != &expected { + if identity != &entry.canonical_promql { return Err(QueryPlanError::Invalid(format!( "query map key `{identity}` differs from entry identity `{}`", - entry.canonical_query + entry.canonical_promql ))); } - match entry.language { - QueryLanguage::PromQl | QueryLanguage::MetricsQl - if entry.fixed_evaluation.is_some() => - { - return Err(QueryPlanError::Invalid( - "PromQL query entry carries a ClickHouse fixed evaluation range".into(), - )); - } - QueryLanguage::ClickHouseSql => { - if self.clickhouse_context.is_none() { - return Err(QueryPlanError::Invalid( - "ClickHouse query entry has no planning context".into(), - )); - } - let Some(range) = entry.fixed_evaluation else { - return Err(QueryPlanError::Invalid( - "ClickHouse query entry has no fixed evaluation range".into(), - )); - }; - if range.end_ms <= range.start_ms { - return Err(QueryPlanError::Invalid( - "ClickHouse query entry has an empty evaluation range".into(), - )); - } - } - QueryLanguage::PromQl | QueryLanguage::MetricsQl => {} - } entry.validate(available)?; } Ok(()) } } -#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] -#[serde(rename_all = "snake_case")] -pub enum QueryLanguage { - #[default] - PromQl, - MetricsQl, - ClickHouseSql, -} - /// 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)] @@ -222,59 +142,61 @@ pub struct QueryNodeId(pub u64); #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(deny_unknown_fields)] pub struct QueryPlanEntry { - #[serde(default)] - pub language: QueryLanguage, pub query_id: String, - #[serde(alias = "canonical_promql")] - pub canonical_query: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub fixed_evaluation: Option, + pub canonical_promql: String, pub root: QueryNodeId, pub nodes: BTreeMap, pub instant: InstantExecution, pub fallback: FallbackPolicy, } -fn topological_order( - root: QueryNodeId, - nodes: &BTreeMap, -) -> 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 - ))); +/// Language-neutral executable projection of a compiled physical query DAG. +/// +/// This is deliberately separate from [`QueryPlanEntry`], whose serialized +/// `canonical_promql` identity remains part of the stable PromQL contract. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct ExecutableQueryPlan { + pub root: QueryNodeId, + pub nodes: BTreeMap, + pub instant: InstantExecution, + pub fallback: FallbackPolicy, +} + +impl QueryPlanEntry { + pub fn executable(&self) -> ExecutableQueryPlan { + ExecutableQueryPlan { + root: self.root, + nodes: self.nodes.clone(), + instant: self.instant.clone(), + fallback: self.fallback.clone(), } - 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)?; + } +} + +impl ExecutableQueryPlan { + pub fn materialization_bindings(&self) -> Vec<&MaterializationBinding> { + self.nodes + .values() + .filter_map(|node| match node { + QueryPlanNode::ReadMaterialization { binding } => Some(binding), + _ => None, + }) + .collect() + } + + /// Internal compatibility view for the existing executor. The supplied + /// identity is never serialized into the PromQL plan catalog. + pub fn execution_view(&self, query_id: String, source: String) -> QueryPlanEntry { + QueryPlanEntry { + query_id, + canonical_promql: source, + root: self.root, + nodes: self.nodes.clone(), + instant: self.instant.clone(), + fallback: self.fallback.clone(), } - visiting.remove(&id); - visited.insert(id); - out.push(id); - Ok(()) } - let mut out = Vec::with_capacity(nodes.len()); - visit( - root, - nodes, - &mut BTreeSet::new(), - &mut BTreeSet::new(), - &mut out, - )?; - Ok(out) } #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] @@ -299,57 +221,10 @@ impl QueryPlanEntry { .collect() } - pub fn topological_order(&self) -> Result, QueryPlanError> { - topological_order(self.root, &self.nodes) - } - - pub fn topological_order_from( - &self, - root: QueryNodeId, - ) -> Result, QueryPlanError> { - topological_order(root, &self.nodes) - } - pub fn compile_bound( query_id: String, - canonical_query: String, - root: &Rc, - instant: InstantExecution, - fallback: FallbackPolicy, - mut bind: F, - ) -> Result - where - F: FnMut( - &SummaryNode, - &SummaryFamilyType, - ) -> Result, - { - let mut compiler = DagCompiler { - next_id: 0, - nodes: BTreeMap::new(), - seen: BTreeMap::new(), - bind: &mut bind, - logical_source: None, - preserve_relational: false, - }; - let root = compiler.lower(root)?; - Ok(Self { - language: QueryLanguage::PromQl, - query_id, - canonical_query, - fixed_evaluation: None, - root, - nodes: compiler.nodes, - instant, - fallback, - }) - } - - pub fn compile_bound_relational( - query_id: String, - canonical_query: String, + canonical_promql: String, root: &Rc, - fixed_evaluation: FixedEvaluationRange, instant: InstantExecution, fallback: FallbackPolicy, mut bind: F, @@ -366,14 +241,11 @@ impl QueryPlanEntry { seen: BTreeMap::new(), bind: &mut bind, logical_source: None, - preserve_relational: true, }; let root = compiler.lower(root)?; Ok(Self { - language: QueryLanguage::ClickHouseSql, query_id, - canonical_query, - fixed_evaluation: Some(fixed_evaluation), + canonical_promql, root, nodes: compiler.nodes, instant, @@ -385,7 +257,7 @@ impl QueryPlanEntry { /// This is a distinct physical alternative; native execution remains available. pub fn compile_bound_composable( query_id: String, - canonical_query: String, + canonical_promql: String, root: &Rc, instant: InstantExecution, fallback: FallbackPolicy, @@ -402,15 +274,12 @@ impl QueryPlanEntry { nodes: BTreeMap::new(), seen: BTreeMap::new(), bind: &mut bind, - logical_source: Some(canonical_query.clone()), - preserve_relational: false, + logical_source: Some(canonical_promql.clone()), }; let root = compiler.lower(root)?; let mut entry = Self { - language: QueryLanguage::PromQl, query_id, - canonical_query, - fixed_evaluation: None, + canonical_promql, root, nodes: compiler.nodes, instant, @@ -490,6 +359,46 @@ impl QueryPlanEntry { } 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)] @@ -532,14 +441,6 @@ pub enum PhysicalGrouping { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(tag = "op", rename_all = "snake_case", deny_unknown_fields)] pub enum QueryPlanNode { - Relational { - input: QueryNodeId, - /// Serialized planner-owned operation. Keeping the wire form here makes - /// the published catalog Send + Sync even though the planner AST uses Rc. - operation: serde_json::Value, - input_schema: planner_types::post_asap::SummarySchema, - output_schema: planner_types::post_asap::SummarySchema, - }, Logical { operator: logical::LogicalOperator, inputs: Vec, @@ -591,7 +492,6 @@ impl QueryPlanNode { } Self::Binary { inputs, .. } => inputs, Self::ReduceSum { input, .. } - | Self::Relational { input, .. } | Self::SummaryEstimate { input, .. } | Self::ExactReadout { input, .. } => std::slice::from_ref(input), Self::SummaryMerge { inputs } | Self::Logical { inputs, .. } => inputs, @@ -656,7 +556,6 @@ struct DagCompiler<'a, F> { seen: BTreeMap, bind: &'a mut F, logical_source: Option, - preserve_relational: bool, } impl DagCompiler<'_, F> @@ -695,8 +594,7 @@ where } QueryPlanNode::SummaryEstimate { input, .. } | QueryPlanNode::ExactReadout { input, .. } - | QueryPlanNode::ReduceSum { input, .. } - | QueryPlanNode::Relational { input, .. } => *input = remap[input], + | QueryPlanNode::ReduceSum { input, .. } => *input = remap[input], QueryPlanNode::Scalar { .. } | QueryPlanNode::ReadMaterialization { .. } | QueryPlanNode::ExactFallback { .. } => {} @@ -747,28 +645,6 @@ where } let physical = match &node.expr { - SummaryExpr::ValueOperation { - child, operation, .. - } if self.preserve_relational - && matches!( - operation, - planner_types::post_asap::ValueOperation::Project { .. } - | planner_types::post_asap::ValueOperation::Filter { .. } - | planner_types::post_asap::ValueOperation::Sort { .. } - | planner_types::post_asap::ValueOperation::Limit { .. } - ) => - { - QueryPlanNode::Relational { - input: self.lower(child)?, - operation: serde_json::to_value(operation).map_err(|error| { - QueryPlanError::UnsupportedNode(format!( - "cannot serialize relational operation: {error}" - )) - })?, - input_schema: child.schema.clone(), - output_schema: node.schema.clone(), - } - } SummaryExpr::ValueOperation { child, operation: @@ -929,60 +805,8 @@ where }) }) .collect::, _>>()?; - let candidate_input = self.lower(candidates)?; - let value_input = if let Some(original) = &self.logical_source { - let parsed = promql_parser::parser::parse(original) - .map_err(|error| QueryPlanError::Invalid(error.to_string()))?; - let promql_parser::parser::Expr::Aggregate(aggregate) = parsed else { - return Err(QueryPlanError::Invalid( - "CandidateTopK requires a top-level PromQL aggregate".into(), - )); - }; - if aggregate.op.to_string() != "topk" { - return Err(QueryPlanError::Invalid( - "CandidateTopK requires a topk source expression".into(), - )); - } - fn item_label(node: &SummaryNode) -> Option { - match &node.expr { - SummaryExpr::SummaryEstimate { summary_input, .. } => { - item_label(summary_input) - } - SummaryExpr::SummaryAgg { input, .. } => match &input.item { - Some(planner_types::post_asap::SummaryInputExpr::Column( - planner_types::pre_asap::ColumnRef::Named(label), - )) => Some(label.clone()), - Some(planner_types::post_asap::SummaryInputExpr::Column( - planner_types::pre_asap::ColumnRef::Qualified { name, .. }, - )) => Some(name.clone()), - _ => None, - }, - _ => None, - } - } - let item_label = item_label(candidates).ok_or_else(|| { - QueryPlanError::Invalid( - "CandidateTopK membership has no named item label".into(), - ) - })?; - let value_id = QueryNodeId(self.next_id); - self.next_id += 1; - self.nodes.insert( - value_id, - QueryPlanNode::Logical { - operator: logical::LogicalOperator::CandidateExactSubquery { - query: aggregate.expr.to_string(), - item_label, - }, - inputs: vec![candidate_input], - }, - ); - value_id - } else { - self.lower(values)? - }; QueryPlanNode::CandidateTopK { - inputs: [candidate_input, value_input], + inputs: [self.lower(candidates)?, self.lower(values)?], k: u64::try_from(*k).map_err(|_| { QueryPlanError::Invalid("CandidateTopK k exceeds u64".into()) })?, @@ -1354,12 +1178,10 @@ mod tests { } #[test] - fn language_tag_preserves_query_entry_serde() { + fn executable_extraction_preserves_promql_entry_serde() { let entry = QueryPlanEntry { - language: crate::query_plan::QueryLanguage::PromQl, query_id: "q".into(), - canonical_query: canonical_promql("up").unwrap(), - fixed_evaluation: None, + canonical_promql: canonical_promql("up").unwrap(), root: QueryNodeId(0), nodes: BTreeMap::from([( QueryNodeId(0), @@ -1375,72 +1197,12 @@ mod tests { fallback: FallbackPolicy::ExactBackend, }; let before = serde_json::to_value(&entry).unwrap(); + let _payload = entry.executable(); assert_eq!(before, serde_json::to_value(&entry).unwrap()); - assert!(before.get("canonical_query").is_some()); + assert!(before.get("canonical_promql").is_some()); assert!(before.get("executable").is_none()); } - #[test] - fn language_catalog_keys_keep_equal_query_text_distinct() { - let base = QueryPlanEntry { - language: QueryLanguage::PromQl, - query_id: "prom".into(), - canonical_query: "shared".into(), - fixed_evaluation: None, - root: QueryNodeId(0), - nodes: BTreeMap::from([( - QueryNodeId(0), - QueryPlanNode::ExactFallback { - reason: "fixture".into(), - }, - )]), - instant: InstantExecution { - lookback_ms: 1, - full_history: false, - cumulative_readout: false, - }, - fallback: FallbackPolicy::ExactBackend, - }; - let mut metricsql = base.clone(); - metricsql.language = QueryLanguage::MetricsQl; - metricsql.query_id = "metrics".into(); - let mut clickhouse = base.clone(); - clickhouse.language = QueryLanguage::ClickHouseSql; - clickhouse.query_id = "sql".into(); - clickhouse.fixed_evaluation = Some(FixedEvaluationRange { - start_ms: 1, - end_ms: 2, - cumulative: true, - }); - let plan = QueryPlan { - plan_id: 0, - plan_version: 0, - clickhouse_context: Some(ClickHousePlanningContext { - tables: Default::default(), - accuracy: planner_types::types::AccuracyTarget::Exact, - }), - entries: [base, metricsql, clickhouse] - .into_iter() - .map(|entry| (QueryPlan::catalog_key(entry.language, "shared"), entry)) - .collect(), - }; - - assert_eq!(plan.entries.len(), 3); - assert_eq!( - plan.lookup_canonical(QueryLanguage::PromQl, "shared") - .unwrap() - .query_id, - "prom" - ); - assert_eq!( - plan.lookup_canonical(QueryLanguage::MetricsQl, "shared") - .unwrap() - .query_id, - "metrics" - ); - assert_eq!(plan.lookup_clickhouse("shared").unwrap().query_id, "sql"); - } - #[test] fn graph_validation_rejects_cycles() { let mut nodes = BTreeMap::new(); @@ -1451,10 +1213,8 @@ mod tests { }, ); let entry = QueryPlanEntry { - language: crate::query_plan::QueryLanguage::PromQl, query_id: "q".into(), - canonical_query: "up".into(), - fixed_evaluation: None, + canonical_promql: "up".into(), root: QueryNodeId(0), nodes, instant: InstantExecution { @@ -1477,10 +1237,8 @@ mod tests { reason: "prepared".into(), }; let entry = QueryPlanEntry { - language: crate::query_plan::QueryLanguage::PromQl, query_id: "q".into(), - canonical_query: "topk(2, rate(m[5m]))".into(), - fixed_evaluation: None, + canonical_promql: "topk(2, rate(m[5m]))".into(), root: QueryNodeId(2), nodes: BTreeMap::from([ (QueryNodeId(0), leaf.clone()), @@ -1548,10 +1306,8 @@ mod catalog_binding_tests { config.pane_origin_ms = Some(0); let catalog = SummaryCatalog::from_materializations(7, 2, &[config.clone()]).unwrap(); let entry = QueryPlanEntry { - language: crate::query_plan::QueryLanguage::PromQl, query_id: "q".into(), - canonical_query: "sum_over_time(m[1m])".into(), - fixed_evaluation: None, + canonical_promql: "sum_over_time(m[1m])".into(), root: QueryNodeId(1), nodes: BTreeMap::from([( QueryNodeId(1), @@ -1577,8 +1333,7 @@ mod catalog_binding_tests { QueryPlan { plan_id: 7, plan_version: 2, - clickhouse_context: None, - entries: BTreeMap::from([(entry.canonical_query.clone(), entry)]), + entries: BTreeMap::from([(entry.canonical_promql.clone(), entry)]), }, catalog, ) @@ -1603,17 +1358,9 @@ mod catalog_binding_tests { #[test] fn catalog_binding_round_trip_preserves_pane_and_readout_windows() { let (plan, catalog) = fixture(); - let wire = serde_json::to_vec(&plan).unwrap(); - let mut decoded: QueryPlan = serde_json::from_slice(&wire).unwrap(); + let mut decoded: QueryPlan = + serde_json::from_slice(&serde_json::to_vec(&plan).unwrap()).unwrap(); decoded.validate_against_catalog(&catalog).unwrap(); - assert_eq!( - decoded - .lookup("sum_over_time(m[1m])") - .unwrap() - .canonical_query, - "sum_over_time(m[1m])" - ); - assert!(String::from_utf8(wire).unwrap().contains("canonical_query")); assert_eq!(binding(&mut decoded).window_ms, 10_000); assert_eq!(binding(&mut decoded).readout_lookback_ms, Some(60_000)); } From 7d1fd08ca00df09aaaccf45dd4433c82a04345c9 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 02:04:54 -0600 Subject: [PATCH 3/6] Separate executable query catalogs by language --- control_plane/src/backend_client.rs | 24 ++ control_plane/src/clickhouse.rs | 60 +-- control_plane/src/emit/backend_push.rs | 1 - control_plane/src/main.rs | 11 +- control_plane/src/physical/compiler.rs | 43 ++- control_plane/src/physical/publication.rs | 6 + control_plane/src/physical/workload_cost.rs | 2 +- control_plane/src/query_plan.rs | 222 ++++++++++- control_plane/src/query_plan/logical.rs | 8 +- .../drivers/ingest/prometheus_remote_write.rs | 2 + data_plane/src/drivers/query/servers/http.rs | 220 +++++++---- data_plane/src/main.rs | 6 + .../accelerator.rs | 338 +++++++++++------ .../asap_clickhouse_query_engine/execution.rs | 35 +- .../asap_clickhouse_query_engine/mod.rs | 2 + .../plan_catalog.rs | 353 ++++++++++++++++++ .../sql_binder.rs | 33 ++ .../asap_query_engine/catalog_resolver.rs | 9 +- .../query_engines/asap_query_engine/engine.rs | 179 +++++---- .../asap_query_engine/exact_subqueries.rs | 22 +- .../asap_query_engine/live_serve.rs | 10 +- .../asap_query_engine/logical_dag.rs | 22 +- .../asap_query_engine/physical_dag.rs | 16 +- .../asap_query_engine/post_asap_readout.rs | 38 +- .../types/hot_reload_config.rs | 5 +- .../asapquery_compatibility_process_e2e.rs | 4 + ...e2e_controller_plans_and_backend_serves.rs | 2 +- data_plane/tests/support/physical_fixture.rs | 7 +- .../tests/victoriametrics_accelerated_e2e.rs | 33 +- 29 files changed, 1301 insertions(+), 412 deletions(-) create mode 100644 data_plane/src/query_engines/asap_clickhouse_query_engine/plan_catalog.rs create mode 100644 data_plane/src/query_engines/asap_clickhouse_query_engine/sql_binder.rs diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs index 2f6a798d4..349d6adf1 100644 --- a/control_plane/src/backend_client.rs +++ b/control_plane/src/backend_client.rs @@ -375,6 +375,28 @@ impl BackendClient { query_plan: &crate::query_plan::QueryPlan, storage_routing: Option, adaptation_evidence: &[crate::physical::compiler::RuntimeAdaptationEvidence], + ) -> std::result::Result<(), BackendPostError> { + self.post_physical_plan_with_sidecars( + precompute_plan, + transmission_plan, + query_plan, + storage_routing, + adaptation_evidence, + None, + None, + ) + .await + } + + pub async fn post_physical_plan_with_sidecars( + &self, + precompute_plan: &crate::physical::compiler::PrecomputePlan, + transmission_plan: &crate::physical::compiler::TransmissionPlan, + query_plan: &crate::query_plan::QueryPlan, + storage_routing: Option, + adaptation_evidence: &[crate::physical::compiler::RuntimeAdaptationEvidence], + metricsql_plan: Option<&crate::query_plan::MetricsQlPlanCatalog>, + clickhouse_sql: Option, ) -> std::result::Result<(), BackendPostError> { // Compatibility replanner has no Planner-selected query/collector DAG. // Still publish the actual catalog and bind every provided projection. @@ -407,6 +429,8 @@ impl BackendClient { "precompute_plan": precompute_plan, "transmission_plan": transmission_plan, "query_plan": query_plan, + "metricsql_plan": metricsql_plan, + "clickhouse_sql": clickhouse_sql, "storage_routing": storage_routing, "adaptation_evidence": adaptation_evidence, })) diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 4f547b5d0..ba350cb0b 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -11,12 +11,11 @@ use planner_types::workload::SqlDialect; use crate::physical::compiler::{PrecomputePlan, TransmissionPlan}; use crate::physical::post_asap::{cost_model::ControlPlaneCostModel, PhysicalExpr}; use crate::query_plan::{ - ClickHousePlanningContext, FallbackPolicy, FixedEvaluationRange, InstantExecution, - MaterializationBinding, PhysicalGrouping, QueryLanguage, QueryPlan, QueryPlanEntry, + FallbackPolicy, InstantExecution, MaterializationBinding, PhysicalGrouping, QueryPlanEntry, }; use asap_types::summary_catalog::SummaryCatalog; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; +use std::collections::{BTreeSet, HashMap}; #[derive(Debug, thiserror::Error)] pub enum ClickHousePlanningError { @@ -90,13 +89,13 @@ pub struct ClickHouseSqlWorkloadEntry { pub cumulative: bool, } -/// Physical-plan components produced for the normal atomic install path. +/// Wire bundle consumed by the independent backend SQL catalog. #[derive(Debug, Serialize)] pub struct ClickHouseCompiledBundle { pub sds: SummaryCatalog, pub tables: HashMap, pub accuracy: AccuracyTarget, - pub query_plan: QueryPlan, + pub plans: Vec, pub precompute_plan: PrecomputePlan, pub transmission_plan: TransmissionPlan, } @@ -115,7 +114,7 @@ pub async fn compile_clickhouse_workload( let catalog = SqlCatalog { tables: request.tables.clone(), }; - let mut entries = std::collections::BTreeMap::new(); + let mut plans = Vec::with_capacity(request.queries.len()); for query in &request.queries { let planned = plan_clickhouse_sql(&query.sql, &catalog, request.accuracy.clone()).await?; let PhysicalExpr::Committed(crate::physical::post_asap::PostAsapPlan::Summary(root)) = @@ -126,14 +125,7 @@ pub async fn compile_clickhouse_workload( )); }; let executable = QueryPlanEntry::compile_bound_relational( - query.sql.clone(), - planned.canonical_sql.clone(), &root, - FixedEvaluationRange { - start_ms: query.start_ms, - end_ms: query.end_ms, - cumulative: query.cumulative, - }, InstantExecution { lookback_ms: query.end_ms.saturating_sub(query.start_ms), full_history: query.start_ms == 0, @@ -153,6 +145,10 @@ pub async fn compile_clickhouse_workload( )); } let bindings = executable.materialization_bindings(); + let materializations = bindings + .iter() + .map(|binding| binding.materialization.fingerprint()) + .collect::>(); let identities = bindings .iter() .map(|binding| { @@ -167,37 +163,21 @@ pub async fn compile_clickhouse_workload( }) }) .collect::, _>>()?; - // Descriptor references are already represented by each DAG's - // MaterializationBinding and validated through SummaryCatalog. - let _descriptor_ids = identities - .iter() - .map(|identity| { - ( - &identity.summary_descriptor_id, - &identity.data_descriptor_id, - ) - }) - .collect::>(); - let identity = QueryPlan::catalog_key(QueryLanguage::ClickHouseSql, &planned.canonical_sql); - if entries.insert(identity.clone(), executable).is_some() { - return Err(ClickHousePlanningError::Lower(format!( - "duplicate canonical SQL query identity `{identity}`" - ))); - } + plans.push(serde_json::json!({ + "sql": planned.canonical_sql, + "runtime": { "start_ms": query.start_ms, "end_ms": query.end_ms, + "cumulative": query.cumulative, + "materializations": materializations, + "executable": executable }, + "descriptors": { "summaries": identities.iter().map(|identity| identity.summary_descriptor_id.clone()).collect::>(), + "data": identities.iter().map(|identity| identity.data_descriptor_id.clone()).collect::>() } + })); } Ok(ClickHouseCompiledBundle { sds: request.sds.clone(), tables: request.tables.clone(), accuracy: request.accuracy.clone(), - query_plan: QueryPlan { - plan_id: request.sds.plan_id, - plan_version: request.sds.plan_version, - clickhouse_context: Some(ClickHousePlanningContext { - tables: request.tables.clone(), - accuracy: request.accuracy.clone(), - }), - entries, - }, + plans, precompute_plan: request.precompute_plan.clone(), transmission_plan: request.transmission_plan.clone(), }) @@ -226,7 +206,7 @@ fn bind_selected_node( window_ms: selected.slide_interval.saturating_mul(1000), pane_origin_ms: selected.pane_origin_ms, readout_lookback_ms: source_window.map(|seconds| seconds.saturating_mul(1000)), - item_labels: selected.aggregated_labels.labels.clone(), + item_labels: Default::default(), }) } diff --git a/control_plane/src/emit/backend_push.rs b/control_plane/src/emit/backend_push.rs index feafed81b..f6fbd974b 100644 --- a/control_plane/src/emit/backend_push.rs +++ b/control_plane/src/emit/backend_push.rs @@ -241,7 +241,6 @@ async fn push_documents_coupled( let query_plan = crate::query_plan::QueryPlan { plan_id: precompute_plan.envelope.plan_id, plan_version: precompute_plan.envelope.plan_version, - clickhouse_context: None, entries: Default::default(), }; let transmission_plan = match crate::physical::compiler::TransmissionPlan::build( diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 0d79d28a2..dc7f4bcf2 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -822,13 +822,20 @@ async fn handle_compile_and_publish_clickhouse_plan( ) .into_response(); }; + let empty_query_plan = control_plane::query_plan::QueryPlan { + plan_id, + plan_version, + entries: Default::default(), + }; if let Err(error) = client - .post_physical_plan_typed( + .post_physical_plan_with_sidecars( &bundle.precompute_plan, &bundle.transmission_plan, - &bundle.query_plan, + &empty_query_plan, None, &[], + None, + Some(serde_json::to_value(&bundle).expect("SQL bundle serializes")), ) .await { diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 9d8f78b73..b53e2d08b 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -737,6 +737,7 @@ pub struct PhysicalPlan { pub precompute_plan: PrecomputePlan, pub transmission_plan: TransmissionPlan, pub query_plan: QueryPlan, + pub metricsql_plan: Option, /// Lifecycle component only, not a complete physical-plan comparison. pub lifecycle_estimates: Vec, pub cost_comparison: Option, @@ -2555,6 +2556,7 @@ impl PhysicalCompiler { .map(asap_types::PrecomputeMaterialization::policy_fingerprint) .collect(); let mut query_entries = BTreeMap::new(); + let mut metricsql_entries = BTreeMap::new(); for query in &request.queries { let canonical = if metricsql { asap_frontend_metricsql::canonical_metricsql(&query.query_string).map_err( @@ -2658,11 +2660,19 @@ impl PhysicalCompiler { // backend-local range index leaf. crate::query_plan::logical::finalize_residuals(&mut entry)?; } - if metricsql { - entry.language = crate::query_plan::QueryLanguage::MetricsQl; - } - let catalog_key = QueryPlan::catalog_key(entry.language, &canonical); - if query_entries.insert(catalog_key, entry).is_some() { + let duplicate = if metricsql { + let sidecar = crate::query_plan::MetricsQlPlanEntry { + query_id: entry.query_id.clone(), + canonical_metricsql: canonical.clone(), + executable: entry.executable(), + }; + metricsql_entries + .insert(canonical.clone(), sidecar) + .is_some() + } else { + query_entries.insert(canonical.clone(), entry).is_some() + }; + if duplicate { return Err(CompileError::Query { query_id: query.query_id.clone(), reason: format!("duplicate canonical query identity `{canonical}`"), @@ -2672,15 +2682,25 @@ impl PhysicalCompiler { let query_plan = QueryPlan { plan_id, plan_version: envelope.plan_version, - clickhouse_context: None, entries: query_entries, }; + let metricsql_plan = metricsql.then_some(crate::query_plan::MetricsQlPlanCatalog { + plan_id, + plan_version: envelope.plan_version, + entries: metricsql_entries, + }); for materialization in &mut precompute_plan.materializations { let fingerprint = materialization.policy_fingerprint(); let max_lookback_ms = query_plan .entries .values() .flat_map(QueryPlanEntry::materialization_bindings) + .chain( + metricsql_plan + .iter() + .flat_map(|catalog| catalog.entries.values()) + .flat_map(|entry| entry.executable.materialization_bindings()), + ) .filter(|binding| binding.materialization.fingerprint() == fingerprint) .filter_map(|binding| binding.readout_lookback_ms) .max(); @@ -2750,6 +2770,7 @@ impl PhysicalCompiler { precompute_plan, transmission_plan, query_plan, + metricsql_plan, lifecycle_estimates: lifecycle_estimates.into_values().collect(), cost_comparison: None, }) @@ -4431,7 +4452,7 @@ mod tests { } #[test] - fn metricsql_compilation_publishes_a_language_tagged_query_entry() { + fn metricsql_compilation_publishes_an_independent_sidecar_entry() { let query = "default_rollup(m[1m])"; let mut workload = request("vm-q", "last_over_time(m[1m])"); let accuracy = workload.queries[0].accuracy.clone(); @@ -4444,11 +4465,13 @@ mod tests { .unwrap(); let identity = asap_frontend_metricsql::canonical_metricsql(query).unwrap(); let entry = plan - .query_plan - .lookup_canonical(crate::query_plan::QueryLanguage::MetricsQl, &identity) + .metricsql_plan + .as_ref() + .unwrap() + .lookup(&identity) .unwrap(); assert_eq!(entry.query_id, "vm-q"); - assert_eq!(entry.language, crate::query_plan::QueryLanguage::MetricsQl); + assert!(plan.query_plan.entries.is_empty()); } #[test] diff --git a/control_plane/src/physical/publication.rs b/control_plane/src/physical/publication.rs index 3c45f6676..a1c3d2e1c 100644 --- a/control_plane/src/physical/publication.rs +++ b/control_plane/src/physical/publication.rs @@ -12,6 +12,10 @@ pub struct PhysicalPlanPublication { pub collector_plans: Vec, pub transmission_plan: TransmissionPlan, pub query_plan: QueryPlan, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub metricsql_plan: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub clickhouse_sql: Option, } impl PhysicalPlanPublication { /// Validate every plan against the shared catalog snapshot. @@ -95,6 +99,8 @@ impl PhysicalPlan { collector_plans: self.collector_plans.clone(), transmission_plan: self.transmission_plan.clone(), query_plan: self.query_plan.clone(), + metricsql_plan: self.metricsql_plan.clone(), + clickhouse_sql: None, }; artifact.validate()?; Ok(artifact) diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs index 4279485c3..912b30006 100644 --- a/control_plane/src/physical/workload_cost.rs +++ b/control_plane/src/physical/workload_cost.rs @@ -258,7 +258,7 @@ pub fn manifest( for node_id in entry.topological_order()? { add( format!("query:{}:{}", entry.query_id, node_id.0), - json!({"node": entry.nodes[&node_id], "query": entry.canonical_query, "instant": entry.instant}), + json!({"node": entry.nodes[&node_id], "query": entry.canonical_promql, "instant": entry.instant}), "query_evaluation", evaluations, ); diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index 599718c91..14c9050f2 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -163,6 +163,84 @@ pub struct ExecutableQueryPlan { pub fallback: FallbackPolicy, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct MetricsQlPlanEntry { + pub query_id: String, + pub canonical_metricsql: String, + pub executable: ExecutableQueryPlan, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct MetricsQlPlanCatalog { + pub plan_id: u64, + pub plan_version: u64, + pub entries: BTreeMap, +} + +impl MetricsQlPlanCatalog { + pub fn lookup(&self, identity: &str) -> Option<&MetricsQlPlanEntry> { + self.entries + .get(identity) + .filter(|entry| entry.canonical_metricsql == identity) + } +} + +pub trait ExecutablePlanView: Sync { + fn root(&self) -> QueryNodeId; + fn nodes(&self) -> &BTreeMap; + fn instant(&self) -> &InstantExecution; + fn fallback(&self) -> &FallbackPolicy; + fn topological_order_from( + &self, + root: QueryNodeId, + ) -> Result, QueryPlanError> { + topological_order(root, self.nodes()) + } + fn topological_order(&self) -> Result, QueryPlanError> { + self.topological_order_from(self.root()) + } + fn materialization_bindings(&self) -> Vec<&MaterializationBinding> { + self.nodes() + .values() + .filter_map(|node| match node { + QueryPlanNode::ReadMaterialization { binding } => Some(binding), + _ => None, + }) + .collect() + } +} + +impl ExecutablePlanView for QueryPlanEntry { + fn root(&self) -> QueryNodeId { + self.root + } + fn nodes(&self) -> &BTreeMap { + &self.nodes + } + fn instant(&self) -> &InstantExecution { + &self.instant + } + fn fallback(&self) -> &FallbackPolicy { + &self.fallback + } +} +impl ExecutablePlanView for ExecutableQueryPlan { + fn root(&self) -> QueryNodeId { + self.root + } + fn nodes(&self) -> &BTreeMap { + &self.nodes + } + fn instant(&self) -> &InstantExecution { + &self.instant + } + fn fallback(&self) -> &FallbackPolicy { + &self.fallback + } +} + impl QueryPlanEntry { pub fn executable(&self) -> ExecutableQueryPlan { ExecutableQueryPlan { @@ -185,17 +263,65 @@ impl ExecutableQueryPlan { .collect() } - /// Internal compatibility view for the existing executor. The supplied - /// identity is never serialized into the PromQL plan catalog. - pub fn execution_view(&self, query_id: String, source: String) -> QueryPlanEntry { - QueryPlanEntry { - query_id, - canonical_promql: source, - root: self.root, - nodes: self.nodes.clone(), - instant: self.instant.clone(), - fallback: self.fallback.clone(), + pub fn topological_order_from( + &self, + root: QueryNodeId, + ) -> Result, QueryPlanError> { + topological_order(root, &self.nodes) + } + pub fn topological_order(&self) -> Result, QueryPlanError> { + topological_order(self.root, &self.nodes) + } +} + +fn topological_order( + root: QueryNodeId, + nodes: &BTreeMap, +) -> 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::new(); + visit( + root, + nodes, + &mut BTreeSet::new(), + &mut BTreeSet::new(), + &mut out, + )?; + Ok(out) +} + +impl QueryPlanEntry { + pub fn topological_order_from( + &self, + root: QueryNodeId, + ) -> Result, QueryPlanError> { + topological_order(root, &self.nodes) } } @@ -241,6 +367,7 @@ impl QueryPlanEntry { seen: BTreeMap::new(), bind: &mut bind, logical_source: None, + preserve_relational: false, }; let root = compiler.lower(root)?; Ok(Self { @@ -253,6 +380,35 @@ impl QueryPlanEntry { }) } + pub fn compile_bound_relational( + root: &Rc, + instant: InstantExecution, + fallback: FallbackPolicy, + mut bind: F, + ) -> Result + where + F: FnMut( + &SummaryNode, + &SummaryFamilyType, + ) -> Result, + { + let mut compiler = DagCompiler { + next_id: 0, + nodes: BTreeMap::new(), + seen: BTreeMap::new(), + bind: &mut bind, + logical_source: None, + preserve_relational: true, + }; + let root = compiler.lower(root)?; + Ok(ExecutableQueryPlan { + root, + nodes: compiler.nodes, + instant, + fallback, + }) + } + /// Compile selected summary nodes and verified native residuals into one DAG. /// This is a distinct physical alternative; native execution remains available. pub fn compile_bound_composable( @@ -275,6 +431,7 @@ impl QueryPlanEntry { seen: BTreeMap::new(), bind: &mut bind, logical_source: Some(canonical_promql.clone()), + preserve_relational: false, }; let root = compiler.lower(root)?; let mut entry = Self { @@ -441,6 +598,12 @@ pub enum PhysicalGrouping { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(tag = "op", rename_all = "snake_case", deny_unknown_fields)] pub enum QueryPlanNode { + Relational { + input: QueryNodeId, + operation: serde_json::Value, + input_schema: planner_types::post_asap::SummarySchema, + output_schema: planner_types::post_asap::SummarySchema, + }, Logical { operator: logical::LogicalOperator, inputs: Vec, @@ -492,6 +655,7 @@ impl QueryPlanNode { } Self::Binary { inputs, .. } => inputs, Self::ReduceSum { input, .. } + | Self::Relational { input, .. } | Self::SummaryEstimate { input, .. } | Self::ExactReadout { input, .. } => std::slice::from_ref(input), Self::SummaryMerge { inputs } | Self::Logical { inputs, .. } => inputs, @@ -556,6 +720,7 @@ struct DagCompiler<'a, F> { seen: BTreeMap, bind: &'a mut F, logical_source: Option, + preserve_relational: bool, } impl DagCompiler<'_, F> @@ -594,7 +759,8 @@ where } QueryPlanNode::SummaryEstimate { input, .. } | QueryPlanNode::ExactReadout { input, .. } - | QueryPlanNode::ReduceSum { input, .. } => *input = remap[input], + | QueryPlanNode::ReduceSum { input, .. } + | QueryPlanNode::Relational { input, .. } => *input = remap[input], QueryPlanNode::Scalar { .. } | QueryPlanNode::ReadMaterialization { .. } | QueryPlanNode::ExactFallback { .. } => {} @@ -645,6 +811,28 @@ where } let physical = match &node.expr { + SummaryExpr::ValueOperation { + child, operation, .. + } if self.preserve_relational + && matches!( + operation, + planner_types::post_asap::ValueOperation::Project { .. } + | planner_types::post_asap::ValueOperation::Filter { .. } + | planner_types::post_asap::ValueOperation::Sort { .. } + | planner_types::post_asap::ValueOperation::Limit { .. } + ) => + { + QueryPlanNode::Relational { + input: self.lower(child)?, + operation: serde_json::to_value(operation).map_err(|error| { + QueryPlanError::UnsupportedNode(format!( + "cannot serialize relational operation: {error}" + )) + })?, + input_schema: child.schema.clone(), + output_schema: node.schema.clone(), + } + } SummaryExpr::ValueOperation { child, operation: @@ -1199,7 +1387,17 @@ mod tests { let before = serde_json::to_value(&entry).unwrap(); let _payload = entry.executable(); assert_eq!(before, serde_json::to_value(&entry).unwrap()); - assert!(before.get("canonical_promql").is_some()); + assert_eq!( + before, + serde_json::json!({ + "query_id": "q", + "canonical_promql": "up", + "root": 0, + "nodes": {"0": {"op": "exact_fallback", "reason": "fixture"}}, + "instant": {"lookback_ms": 1, "full_history": false, "cumulative_readout": false}, + "fallback": "exact_backend" + }) + ); assert!(before.get("executable").is_none()); } diff --git a/control_plane/src/query_plan/logical.rs b/control_plane/src/query_plan/logical.rs index 903f2e98b..d1e57226a 100644 --- a/control_plane/src/query_plan/logical.rs +++ b/control_plane/src/query_plan/logical.rs @@ -425,10 +425,8 @@ impl QueryPlanEntry { }; let root = lower.lower(&expr)?; let entry = Self { - language: super::QueryLanguage::PromQl, query_id, - canonical_query, - fixed_evaluation: None, + canonical_promql: canonical_query, root, nodes: lower.nodes, instant, @@ -451,7 +449,7 @@ impl QueryPlanEntry { } Self::compile_logical( self.query_id.clone(), - self.canonical_query.clone(), + self.canonical_promql.clone(), self.instant, self.fallback, ) @@ -1382,7 +1380,7 @@ pub fn externalize_residuals(entry: &mut QueryPlanEntry) -> Result<(), QueryPlan _ => {} } } - let expr = parser::parse(&entry.canonical_query).map_err(|e| invalid(e.to_string()))?; + let expr = parser::parse(&entry.canonical_promql).map_err(|e| invalid(e.to_string()))?; let mut expressions = Vec::new(); gather(&expr, &mut expressions); let mut witnesses = BTreeMap::new(); diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index 265ba9c69..fe3ba0f4c 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -803,6 +803,8 @@ mod tests { }, runtime_config: Arc::new(streaming), query_plan: Arc::new(control_plane::query_plan::QueryPlan::empty()), + metricsql_plan: None, + clickhouse_sql: None, storage_routing: Arc::new(BackendStorageRouting::empty()), }; HotReloadStreamingConfig::from_active(HotReloadActivePhysicalPlan::new(active)) diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index d3cf14d63..dd1acd474 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -109,6 +109,16 @@ fn extract_tenant(headers: &axum::http::HeaderMap) -> String { /// Copy only request context that is safe and meaningful for the configured /// Prometheus fallback. Hop-by-hop and client-controlled transport headers are /// intentionally excluded. +fn requires_tenant_scoped_metricsql_fallback( + adapter_name: &str, + headers: &HashMap, +) -> bool { + adapter_name == "VictoriaMetrics HTTP / MetricsQL" + && headers + .get("x-asap-tenant") + .is_some_and(|tenant| !tenant.trim().is_empty()) +} + fn extract_fallback_headers(headers: &HeaderMap) -> HashMap { const FORWARDED: [&str; 3] = ["authorization", "x-scope-orgid", "x-asap-tenant"]; FORWARDED @@ -722,6 +732,20 @@ async fn process_query_request( } let language_identity = state.adapter.canonical_plan_identity(&parsed_request.query); + // Published summary catalogs are currently global. A VM cluster tenant + // must never read that global catalog until planning publishes an + // explicitly tenant-scoped sidecar, so tenant routes fail closed to VM. + if requires_tenant_scoped_metricsql_fallback(state.adapter.adapter_name(), &headers) { + if let Some(fallback) = &state.fallback { + return match fallback + .execute_query_with_headers(parsed_request, headers) + .await + { + Ok(response) => response.into_response(), + Err(status) => status.into_response(), + }; + } + } if state.adapter.adapter_name() == "VictoriaMetrics HTTP / MetricsQL" && language_identity.is_err() { @@ -2264,6 +2288,18 @@ async fn process_range_query_request( let step_ms = (parsed_request.step * 1000.0) as u64; let language_identity = state.adapter.canonical_plan_identity(&parsed_request.query); + if requires_tenant_scoped_metricsql_fallback(state.adapter.adapter_name(), &forwarding_headers) + { + if let Some(fallback) = &state.fallback { + return match fallback + .execute_range_query_with_headers(parsed_request, forwarding_headers) + .await + { + Ok(response) => response.into_response(), + Err(status) => status.into_response(), + }; + } + } if state.adapter.adapter_name() == "VictoriaMetrics HTTP / MetricsQL" && language_identity.is_err() { @@ -2711,9 +2747,10 @@ mod tests { query_plan: Arc::new(control_plane::query_plan::QueryPlan { plan_id: 7, plan_version: 1, - clickhouse_context: None, entries: Default::default(), }), + metricsql_plan: None, + clickhouse_sql: None, storage_routing: Arc::new( crate::storage_engines::types::BackendStorageRouting::empty(), ), @@ -2859,6 +2896,27 @@ mod tests { server.start_test_server().await.unwrap() } + #[test] + fn victoriametrics_instant_and_range_fail_closed_for_each_tenant() { + for tenant in ["tenant-a", "tenant-b"] { + let headers = HashMap::from([("x-asap-tenant".to_string(), tenant.to_string())]); + assert!( + requires_tenant_scoped_metricsql_fallback( + "VictoriaMetrics HTTP / MetricsQL", + &headers + ), + "instant tenant {tenant} must bypass global catalog" + ); + assert!( + requires_tenant_scoped_metricsql_fallback( + "VictoriaMetrics HTTP / MetricsQL", + &headers + ), + "range tenant {tenant} must bypass global catalog" + ); + } + } + #[tokio::test] async fn victoriametrics_cluster_routes_are_registered() { let port = setup_victoriametrics_test_server().await; @@ -6006,6 +6064,10 @@ pub struct PhysicalPlanInstallRequest { pub precompute_plan: control_plane::physical::compiler::PrecomputePlan, pub transmission_plan: control_plane::physical::compiler::TransmissionPlan, pub query_plan: control_plane::query_plan::QueryPlan, + #[serde(default)] + pub metricsql_plan: Option, + #[serde(default)] + pub clickhouse_sql: Option, pub storage_routing: Option, #[serde(default)] pub adaptation_evidence: Vec, @@ -6084,6 +6146,46 @@ pub fn build_active_physical_plan( .query_plan .validate(&typed_fps) .map_err(|error| format!("QueryPlan validation error: {error}"))?; + let metricsql_plan = request + .metricsql_plan + .map(|catalog| { + if (catalog.plan_id, catalog.plan_version) != (envelope.plan_id, envelope.plan_version) + { + return Err("MetricsQL catalog belongs to another physical generation".to_string()); + } + for (identity, entry) in &catalog.entries { + if identity != &entry.canonical_metricsql { + return Err("MetricsQL catalog key differs from canonical identity".into()); + } + for binding in entry.executable.materialization_bindings() { + let materialization = request.precompute_plan.materializations.iter() + .find(|config| config.policy_fingerprint() == binding.materialization.fingerprint()) + .ok_or_else(|| "MetricsQL binding has no precompute definition".to_string())?; + if binding.window_ms != materialization.slide_interval.saturating_mul(1_000) { + return Err("MetricsQL physical pane duration differs from installed precompute definition".into()); + } + if binding.pane_origin_ms != materialization.pane_origin_ms { + return Err("MetricsQL physical pane origin differs from installed precompute definition".into()); + } + } + crate::query_engines::asap_query_engine::catalog_resolver::validate_payload( + Some(&request.summary_catalog), + &entry.executable, + envelope.plan_id, + envelope.plan_version, + ) + .map_err(|error| format!("MetricsQL executable validation error: {error}"))?; + } + Ok(Arc::new(catalog)) + }) + .transpose()?; + let clickhouse_sql = request.clickhouse_sql.map(|value| { + let bundle = serde_json::from_value(value) + .map_err(|error| format!("ClickHouse SQL sidecar decode error: {error}"))?; + crate::query_engines::asap_clickhouse_query_engine::accelerator::build_active_generation( + bundle, &request.summary_catalog, + ).map(Arc::new).map_err(|error| format!("ClickHouse SQL sidecar validation error: {error}")) + }).transpose()?; let storage_routing = match request.storage_routing.as_ref() { Some(value) => Arc::new( crate::storage_engines::types::BackendStorageRouting::from_json_payload(value) @@ -6098,6 +6200,8 @@ pub fn build_active_physical_plan( transmission_plan: request.transmission_plan, runtime_config: Arc::new(runtime_config), query_plan: Arc::new(request.query_plan), + metricsql_plan, + clickhouse_sql, storage_routing, }) } @@ -6177,17 +6281,15 @@ async fn handle_post_physical_plan( let plan_id = active.plan_id(); let materialization_count = active.precompute_plan.materializations.len(); let metricsql_query_count = active - .query_plan - .entries - .values() - .filter(|entry| entry.language == control_plane::query_plan::QueryLanguage::MetricsQl) - .count(); + .metricsql_plan + .as_ref() + .map(|catalog| catalog.entries.len()) + .unwrap_or(0); let clickhouse_plan_count = active - .query_plan - .entries - .values() - .filter(|entry| entry.language == control_plane::query_plan::QueryLanguage::ClickHouseSql) - .count(); + .clickhouse_sql + .as_ref() + .map(|generation| generation.catalog.len()) + .unwrap_or(0); let plan_version = active.plan_version(); let now = unix_time_ms(); if let Err(error) = lifecycle.stage(active, now) { @@ -6248,11 +6350,10 @@ async fn handle_activate_physical_plan( }; let activated = active_handle.snapshot(); let clickhouse_plan_count = activated - .query_plan - .entries - .values() - .filter(|entry| entry.language == control_plane::query_plan::QueryLanguage::ClickHouseSql) - .count(); + .clickhouse_sql + .as_ref() + .map(|generation| generation.catalog.len()) + .unwrap_or(0); if let Some(catalog) = activated.summary_catalog.as_ref() { if let Err(error) = state .sketch_index @@ -7126,6 +7227,8 @@ mod catalog_install_tests { precompute_plan: plan.precompute_plan, transmission_plan: plan.transmission_plan, query_plan: plan.query_plan, + metricsql_plan: plan.metricsql_plan, + clickhouse_sql: None, storage_routing: None, adaptation_evidence: vec![], } @@ -7145,44 +7248,10 @@ mod catalog_install_tests { let handle = crate::storage_engines::types::HotReloadActivePhysicalPlan::new(active); let before = handle.snapshot(); let mut candidate = request(); - candidate.query_plan.clickhouse_context = - Some(control_plane::query_plan::ClickHousePlanningContext { - tables: Default::default(), - accuracy: planner_types::types::AccuracyTarget::Exact, - }); - let (_, mut entry) = candidate - .query_plan - .entries - .pop_first() - .expect("fixture has a query entry"); - entry.language = control_plane::query_plan::QueryLanguage::ClickHouseSql; - entry.fixed_evaluation = Some(control_plane::query_plan::FixedEvaluationRange { - start_ms: 0, - end_ms: 1_000, - cumulative: true, - }); - let canonical = entry.canonical_query.clone(); - let binding = entry - .nodes - .values_mut() - .find_map(|node| match node { - control_plane::query_plan::QueryPlanNode::ReadMaterialization { binding } => { - Some(binding) - } - _ => None, - }) - .expect("fixture query reads a materialization"); - binding.pane_origin_ms = Some(999); - candidate - .query_plan - .entries - .insert(format!("clickhouse:{canonical}"), entry); - - let error = install(candidate).expect_err("invalid SQL binding must fail staging"); - assert!(error.contains("pane origin"), "{error}"); - let after = handle.snapshot(); - assert_eq!(after.plan_id(), before.plan_id()); - assert_eq!(after.plan_version(), before.plan_version()); + candidate.clickhouse_sql = Some(serde_json::json!({"invalid": true})); + let error = install(candidate).expect_err("invalid SQL sidecar must fail staging"); + assert!(error.contains("ClickHouse SQL sidecar"), "{error}"); + assert!(Arc::ptr_eq(&before, &handle.snapshot())); } // Installing transports the exact supplied snapshot, rather than rebuilding it. @@ -7252,19 +7321,36 @@ mod catalog_install_tests { #[test] fn catalog_install_applies_pane_origin_validation_to_metricsql_entries() { let mut request = request(); - let entry = request.query_plan.entries.values_mut().next().unwrap(); - entry.language = control_plane::query_plan::QueryLanguage::MetricsQl; - let binding = entry - .nodes - .values_mut() - .find_map(|node| match node { - control_plane::query_plan::QueryPlanNode::ReadMaterialization { binding } => { - Some(binding) - } - _ => None, - }) - .expect("demo has maintained summaries"); - binding.pane_origin_ms = Some(1); + let (_, entry) = request.query_plan.entries.iter().next().unwrap(); + let identity = "sum(rate(foo[5m]))".to_string(); + let mut executable = entry.executable(); + let binding = executable + .materialization_bindings() + .into_iter() + .next() + .unwrap() + .clone(); + for node in executable.nodes.values_mut() { + if let control_plane::query_plan::QueryPlanNode::ReadMaterialization { + binding: candidate, + } = node + { + *candidate = binding.clone(); + candidate.pane_origin_ms = Some(1); + } + } + request.metricsql_plan = Some(control_plane::query_plan::MetricsQlPlanCatalog { + plan_id: request.query_plan.plan_id, + plan_version: request.query_plan.plan_version, + entries: std::collections::BTreeMap::from([( + identity.clone(), + control_plane::query_plan::MetricsQlPlanEntry { + query_id: "vm".into(), + canonical_metricsql: identity, + executable, + }, + )]), + }); assert!(install(request).unwrap_err().contains("pane origin")); } } diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index d1a272718..770d56eec 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -567,6 +567,8 @@ async fn main() -> Result<()> { precompute_plan: plan.precompute_plan, transmission_plan: plan.transmission_plan, query_plan: plan.query_plan, + metricsql_plan: plan.metricsql_plan, + clickhouse_sql: None, storage_routing: None, adaptation_evidence: Vec::new(), }, @@ -780,6 +782,8 @@ async fn main() -> Result<()> { transmission_plan: initial_transmission_plan, runtime_config: streaming_config.clone(), query_plan: Arc::new(control_plane::query_plan::QueryPlan::empty()), + metricsql_plan: None, + clickhouse_sql: None, storage_routing: Arc::new( data_plane::storage_engines::types::BackendStorageRouting::empty(), ), @@ -1168,6 +1172,8 @@ async fn main() -> Result<()> { transmission_plan: current.transmission_plan.clone(), runtime_config: current.runtime_config.clone(), query_plan: current.query_plan.clone(), + metricsql_plan: current.metricsql_plan.clone(), + clickhouse_sql: current.clickhouse_sql.clone(), storage_routing: Arc::new(bootstrap_routing), }); } diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs index 377e572f6..170331137 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs @@ -1,24 +1,113 @@ //! Catalog-backed ClickHouse acceleration boundary. +use std::{ + collections::{BTreeSet, HashMap}, + sync::{Arc, Mutex, RwLock}, +}; + +use asap_types::{summary_catalog::SummaryCatalog, PolicyFingerprint}; use async_trait::async_trait; use axum::{ body::Bytes, http::{HeaderMap, HeaderValue, StatusCode}, }; -use std::sync::Arc; +use control_plane::query_plan::ExecutableQueryPlan; use super::{ clickhouse_result_adapter::ClickHouseFormat, execution::{execute_sql_dag, ClickHouseDagFallback, ClickHouseDagOutcome}, fallback::ClickHouseRawResponse, + plan_catalog::{ + SdsDescriptorReferences, SqlPlanCatalog, SqlPlanCatalogGeneration, SqlPlanEntry, + }, request::ClickHouseQueryRequest, server::{ ClickHouseAccelerationFallback, ClickHouseAccelerationOutcome, ClickHouseAccelerator, }, + sql_binder::ClickHouseSqlBinder, }; use crate::storage_engines::sketch_db::index::SketchStore; +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] +pub struct SqlRuntimePlan { + pub start_ms: u64, + pub end_ms: u64, + pub cumulative: bool, + pub materializations: BTreeSet, + /// Control-plane compiled and materialization-bound executable DAG. + pub executable: ExecutableQueryPlan, +} + +#[derive(Debug, serde::Serialize, serde::Deserialize)] +pub struct ClickHousePublishedPlan { + pub sql: String, + pub runtime: SqlRuntimePlan, + pub descriptors: SdsDescriptorReferences, +} + +/// Independently published ClickHouse planning bundle. SDS remains the +/// descriptor authority; SQL DAG metadata lives in this separate catalog. +#[derive(Debug, serde::Serialize, serde::Deserialize)] +pub struct ClickHousePlanBundle { + pub sds: SummaryCatalog, + pub tables: HashMap, + pub accuracy: planner_types::types::AccuracyTarget, + pub plans: Vec, +} + +#[derive(Clone, Debug)] +pub struct ClickHouseActiveGeneration { + pub catalog: Arc>, + pub binder: ClickHouseSqlBinder, +} + +pub fn build_active_generation( + bundle: ClickHousePlanBundle, + physical_sds: &SummaryCatalog, +) -> Result { + if bundle + .sds + .reference() + .map_err(|e| super::plan_catalog::SqlPlanCatalogError::InvalidSds(e.to_string()))? + != physical_sds + .reference() + .map_err(|e| super::plan_catalog::SqlPlanCatalogError::InvalidSds(e.to_string()))? + { + return Err(super::plan_catalog::SqlPlanCatalogError::InvalidSds( + "SQL sidecar SDS differs from physical publication SDS".into(), + )); + } + for plan in &bundle.plans { + crate::query_engines::asap_query_engine::catalog_resolver::validate_payload( + Some(physical_sds), + &plan.runtime.executable, + physical_sds.plan_id, + physical_sds.plan_version, + ) + .map_err(|e| super::plan_catalog::SqlPlanCatalogError::InvalidSds(e.to_string()))?; + } + let entries = bundle.plans.into_iter().map(|plan| SqlPlanEntry { + sql_template: plan.sql, + plan: plan.runtime, + descriptors: plan.descriptors, + }); + let generation = SqlPlanCatalogGeneration::build(physical_sds, entries)?; + Ok(ClickHouseActiveGeneration { + catalog: Arc::new(generation), + binder: ClickHouseSqlBinder::new( + control_plane::clickhouse::ClickHouseSqlCatalog { + tables: bundle.tables, + }, + bundle.accuracy, + ), + }) +} + pub struct CatalogClickHouseAccelerator { + pub catalog: Arc>, + binder: RwLock>, + staged_binder: Mutex>, + publication: RwLock<()>, pub store: Arc, active_physical_plan: Option, } @@ -26,6 +115,10 @@ pub struct CatalogClickHouseAccelerator { impl CatalogClickHouseAccelerator { pub fn empty(store: Arc) -> Self { Self { + catalog: Arc::new(SqlPlanCatalog::default()), + binder: RwLock::new(None), + staged_binder: Mutex::new(None), + publication: RwLock::new(()), store, active_physical_plan: None, } @@ -39,6 +132,78 @@ impl CatalogClickHouseAccelerator { accelerator.active_physical_plan = Some(active); accelerator } + + pub fn from_bundle( + bundle: ClickHousePlanBundle, + store: Arc, + ) -> Result { + let accelerator = Self::empty(store); + let staged = accelerator.stage_bundle(bundle)?; + accelerator.activate(staged.plan_id, staged.plan_version)?; + Ok(accelerator) + } + + pub fn stage_bundle( + &self, + bundle: ClickHousePlanBundle, + ) -> Result + { + let _publication = self.publication.write().unwrap(); + for plan in &bundle.plans { + crate::query_engines::asap_query_engine::catalog_resolver::validate_payload( + Some(&bundle.sds), + &plan.runtime.executable, + bundle.sds.plan_id, + bundle.sds.plan_version, + ) + .map_err(|error| { + super::plan_catalog::SqlPlanCatalogError::InvalidSds(error.to_string()) + })?; + } + let entries = bundle.plans.into_iter().map(|plan| SqlPlanEntry { + sql_template: plan.sql, + plan: plan.runtime, + descriptors: plan.descriptors, + }); + let generation = SqlPlanCatalogGeneration::build(&bundle.sds, entries)?; + let plan_id = generation.sds.plan_id; + let plan_version = generation.sds.plan_version; + let binder = ClickHouseSqlBinder::new( + control_plane::clickhouse::ClickHouseSqlCatalog { + tables: bundle.tables, + }, + bundle.accuracy, + ); + let ack = self.catalog.stage(generation)?; + *self.staged_binder.lock().unwrap() = Some((plan_id, plan_version, binder)); + Ok(ack) + } + + pub fn activate( + &self, + plan_id: u64, + plan_version: u64, + ) -> Result + { + let _publication = self.publication.write().unwrap(); + let mut staged = self.staged_binder.lock().unwrap(); + let Some((staged_id, staged_version, _)) = staged.as_ref() else { + return Err(super::plan_catalog::SqlPlanCatalogError::NotStaged { + plan_id, + plan_version, + }); + }; + if (*staged_id, *staged_version) != (plan_id, plan_version) { + return Err(super::plan_catalog::SqlPlanCatalogError::NotStaged { + plan_id, + plan_version, + }); + } + let ack = self.catalog.activate(plan_id, plan_version)?; + let (_, _, binder) = staged.take().expect("staged binder checked above"); + *self.binder.write().unwrap() = Some(binder); + Ok(ack) + } } fn requested_format(request: &ClickHouseQueryRequest) -> Result { @@ -58,25 +223,18 @@ fn requested_format(request: &ClickHouseQueryRequest) -> Result ClickHouseAccelerationOutcome { - let Some(physical) = self.active_physical_plan.as_ref().map(|h| h.snapshot()) else { - return ClickHouseAccelerationOutcome::Fallback( - ClickHouseAccelerationFallback::CatalogMiss, - ); - }; - let Some(context) = physical.query_plan.clickhouse_context.as_ref() else { + let physical = self.active_physical_plan.as_ref().map(|h| h.snapshot()); + let sidecar = physical.as_ref().and_then(|p| p.clickhouse_sql.clone()); + let binder = sidecar + .as_ref() + .map(|s| s.binder.clone()) + .or_else(|| self.binder.read().unwrap().clone()); + let Some(binder) = binder else { return ClickHouseAccelerationOutcome::Fallback( ClickHouseAccelerationFallback::CatalogMiss, ); }; - let canonical_sql = match control_plane::clickhouse::canonicalize_clickhouse_sql( - &request.sql, - &control_plane::clickhouse::ClickHouseSqlCatalog { - tables: context.tables.clone(), - }, - context.accuracy.clone(), - ) - .await - { + let canonical_sql = match binder.canonical_identity(&request.sql).await { Ok(canonical) => canonical, Err(error) => { return ClickHouseAccelerationOutcome::Fallback( @@ -84,7 +242,16 @@ impl ClickHouseAccelerator for CatalogClickHouseAccelerator { ) } }; - let Ok(entry) = physical.query_plan.lookup_clickhouse(&canonical_sql) else { + let (entry, generation) = if let Some(sidecar) = sidecar { + ( + sidecar.catalog.lookup(&canonical_sql), + Some(sidecar.catalog.clone()), + ) + } else { + let _publication = self.publication.read().unwrap(); + (self.catalog.lookup(&canonical_sql), self.catalog.active()) + }; + let Some(entry) = entry else { return ClickHouseAccelerationOutcome::Fallback( ClickHouseAccelerationFallback::CatalogMiss, ); @@ -97,25 +264,18 @@ impl ClickHouseAccelerator for CatalogClickHouseAccelerator { ) } }; - let Some(catalog) = physical.summary_catalog.as_ref() else { + let Some(generation) = generation else { return ClickHouseAccelerationOutcome::Fallback( ClickHouseAccelerationFallback::CatalogMiss, ); }; - let Some(range) = entry.fixed_evaluation else { - return ClickHouseAccelerationOutcome::Fallback( - ClickHouseAccelerationFallback::Execution( - "ClickHouse plan is missing its fixed evaluation range".into(), - ), - ); - }; match execute_sql_dag( self.store.as_ref(), - entry, - catalog.as_ref(), - range.start_ms, - range.end_ms, - range.cumulative, + &entry.plan.executable, + generation.catalog.as_ref(), + entry.plan.start_ms, + entry.plan.end_ms, + entry.plan.cumulative, ) { ClickHouseDagOutcome::Accelerated(result) => match result.encode(format) { Ok(body) => { @@ -160,13 +320,11 @@ mod tests { precompute_engine::operators::SumAccumulator, storage_engines::sketch_db::index::{AggKind, Capability, SketchInstanceMetadata}, }; - use asap_types::summary_catalog::SummaryCatalog; use asap_types::{AggregationType, KeyByLabelNames, PrecomputeMaterialization, WindowKind}; use axum::http::Method; use control_plane::query_plan::{ - ClickHousePlanningContext, ExactReadout, FallbackPolicy, FixedEvaluationRange, - InstantExecution, MaterializationBinding, PhysicalGrouping, QueryLanguage, QueryNodeId, - QueryPlan, QueryPlanEntry, QueryPlanNode, + ExactReadout, FallbackPolicy, InstantExecution, MaterializationBinding, PhysicalGrouping, + QueryNodeId, QueryPlanNode, }; use planner_types::{ post_asap::{SummaryFamilyType, SummaryField, SummarySchema, ValueOperation}, @@ -175,10 +333,7 @@ mod tests { QueryExpr, ScalarValue, Schema, SortKey, }, }; - use std::{ - collections::{BTreeMap, BTreeSet, HashMap}, - rc::Rc, - }; + use std::rc::Rc; fn relation_schema(names: &[(&str, DataType)]) -> SummarySchema { SummarySchema { @@ -205,7 +360,7 @@ mod tests { store: Arc, seed: bool, ) -> (CatalogClickHouseAccelerator, ClickHouseQueryRequest) { - let mut config = PrecomputeMaterialization::new( + let config = PrecomputeMaterialization::new( AggregationType::Sum, String::new(), Default::default(), @@ -222,8 +377,7 @@ mod tests { None, None, ); - config.pane_origin_ms = Some(0); - let sds = SummaryCatalog::from_materializations(41, 1, &[config.clone()]).unwrap(); + let sds = SummaryCatalog::from_materializations(41, 1, &[config]).unwrap(); let materialization = *sds.materializations.keys().next().unwrap(); let read = QueryNodeId(0); let readout = QueryNodeId(1); @@ -239,24 +393,16 @@ mod tests { let project = QueryNodeId(3); let sort = QueryNodeId(4); let root = QueryNodeId(5); - let executable = QueryPlanEntry { - language: QueryLanguage::ClickHouseSql, - query_id: "SELECT value FROM samples".into(), - canonical_query: "SELECT value FROM samples".into(), - fixed_evaluation: Some(FixedEvaluationRange { - start_ms: 0, - end_ms: 2_000, - cumulative: true, - }), + let executable = ExecutableQueryPlan { root, nodes: [ ( read, QueryPlanNode::ReadMaterialization { binding: MaterializationBinding { + item_labels: Default::default(), materialization, output_grouping: PhysicalGrouping::Reduce(Vec::new()), - item_labels: Vec::new(), window_ms: 1_000, pane_origin_ms: Some(0), readout_lookback_ms: None, @@ -348,6 +494,10 @@ mod tests { }, fallback: FallbackPolicy::ExactBackend, }; + let descriptors = SdsDescriptorReferences { + summaries: sds.summary_descriptors.keys().cloned().collect(), + data: sds.data_descriptors.keys().cloned().collect(), + }; let table_schema = Schema::with_time_index( vec![ Column::new("timestamp", DataType::Timestamp, false), @@ -366,35 +516,23 @@ mod tests { ) .await .unwrap(); - let entry = QueryPlanEntry { - query_id: "SELECT sum(value) FROM requests".into(), - canonical_query: canonical_sql.clone(), - language: QueryLanguage::ClickHouseSql, - fixed_evaluation: Some(FixedEvaluationRange { - start_ms: 0, - end_ms, - cumulative: true, - }), - root: executable.root, - nodes: executable.nodes, - instant: executable.instant, - fallback: executable.fallback, - }; - let query_plan = QueryPlan { - plan_id: 41, - plan_version: 1, - clickhouse_context: Some(ClickHousePlanningContext { - tables, - accuracy: planner_types::types::AccuracyTarget::Exact, - }), - entries: BTreeMap::from([( - QueryPlan::catalog_key(QueryLanguage::ClickHouseSql, &canonical_sql), - entry, - )]), + let bundle = ClickHousePlanBundle { + sds: sds.clone(), + tables, + accuracy: planner_types::types::AccuracyTarget::Exact, + plans: vec![ClickHousePublishedPlan { + sql: canonical_sql, + runtime: SqlRuntimePlan { + start_ms: 0, + end_ms, + cumulative: true, + materializations: BTreeSet::from([materialization.fingerprint()]), + executable, + }, + descriptors, + }], }; - store - .install_summary_catalog(Arc::new(sds.clone())) - .unwrap(); + store.install_summary_catalog(Arc::new(sds)).unwrap(); store.register(SketchInstanceMetadata { sid: 7, metric_name: "requests".into(), @@ -425,47 +563,7 @@ mod tests { Box::new(SumAccumulator::with_sum(3.0)), ); } - let envelope = control_plane::physical::compiler::PlanEnvelope { - plan_id: 41, - plan_version: 1, - generated_at_unix_ms: 0, - activation_unix_ms: 0, - expiry_unix_ms: None, - backend_compat: control_plane::physical::compiler::BACKEND_COMPAT.into(), - planner_revision: control_plane::physical::compiler::PLANNER_REVISION.into(), - capability_snapshot_id: "clickhouse-test".into(), - }; - let mut precompute = control_plane::physical::compiler::PrecomputePlan::build( - envelope.clone(), - vec![config], - &["fixture".into()], - ) - .unwrap(); - precompute.summary_catalog = Some(sds.reference().unwrap()); - let mut transmission = control_plane::physical::compiler::TransmissionPlan::build( - envelope.clone(), - &precompute, - &BTreeMap::new(), - ) - .unwrap(); - transmission.summary_catalog = Some(sds.reference().unwrap()); - let active = crate::drivers::query::servers::http::build_active_physical_plan( - crate::drivers::query::servers::http::PhysicalPlanInstallRequest { - summary_catalog: sds, - collector_plans: vec![], - precompute_plan: precompute, - transmission_plan: transmission, - query_plan, - storage_routing: None, - adaptation_evidence: vec![], - }, - Arc::new(crate::storage_engines::types::BackendStorageRouting::empty()), - ) - .unwrap(); - let accelerator = CatalogClickHouseAccelerator::with_active_physical_plan( - store, - crate::storage_engines::types::HotReloadActivePhysicalPlan::new(active), - ); + let accelerator = CatalogClickHouseAccelerator::from_bundle(bundle, store).unwrap(); let request = ClickHouseQueryRequest { method: Method::GET, sql: "SELECT sum(value) FROM requests".into(), diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs index 619852208..890a134c8 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs @@ -5,14 +5,14 @@ use super::{ }; use crate::{ query_engines::asap_query_engine::{ - catalog_resolver::validate_payload, post_asap_readout::execute_query_plan_from_readout, + catalog_resolver::validate_payload, post_asap_readout::execute_query_plan_payload_readout, }, storage_engines::sketch_db::index::SketchStore, }; use asap_types::summary_catalog::SummaryCatalog; -use control_plane::query_plan::{QueryNodeId, QueryPlanEntry, QueryPlanNode}; +use control_plane::query_plan::{ExecutableQueryPlan, QueryNodeId, QueryPlanNode}; use planner_types::post_asap::ValueOperation; -use std::collections::BTreeSet; +use std::collections::{BTreeMap, BTreeSet}; #[derive(Debug, Clone, PartialEq, Eq)] pub enum ClickHouseDagFallback { @@ -44,7 +44,7 @@ fn complete_pane_coverage( /// summary selection, or materialization candidate search. pub fn execute_sql_dag( index: &SketchStore, - entry: &QueryPlanEntry, + entry: &ExecutableQueryPlan, sds: &SummaryCatalog, t0_ms: u64, t1_ms: u64, @@ -75,12 +75,14 @@ pub fn execute_sql_dag( _ => break, } } - if let Err(detail) = validate_reachable(entry, base_root) { - return ClickHouseDagOutcome::Fallback(ClickHouseDagFallback::UnsupportedPlan(detail)); - } + let executable = match reachable_entry(entry, base_root) { + Ok(entry) => entry, + Err(detail) => { + return ClickHouseDagOutcome::Fallback(ClickHouseDagFallback::UnsupportedPlan(detail)) + } + }; let outcome = - match execute_query_plan_from_readout(index, entry, base_root, t0_ms, t1_ms, is_cumulative) - { + match execute_query_plan_payload_readout(index, &executable, t0_ms, t1_ms, is_cumulative) { Ok(outcome) => outcome, Err(error) => { let detail = format!("{error:?}"); @@ -179,7 +181,10 @@ pub fn execute_sql_dag( } } -fn validate_reachable(entry: &QueryPlanEntry, root: QueryNodeId) -> Result<(), String> { +fn reachable_entry( + entry: &ExecutableQueryPlan, + root: QueryNodeId, +) -> Result { let mut pending = vec![root]; let mut reachable = BTreeSet::new(); while let Some(id) = pending.pop() { @@ -192,7 +197,15 @@ fn validate_reachable(entry: &QueryPlanEntry, root: QueryNodeId) -> Result<(), S .ok_or_else(|| format!("published DAG references missing node {}", id.0))?; pending.extend(node.inputs()); } - Ok(()) + let nodes = reachable + .into_iter() + .map(|id| (id, entry.nodes[&id].clone())) + .collect::>(); + Ok(ExecutableQueryPlan { + root, + nodes, + ..entry.clone() + }) } #[cfg(test)] diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/mod.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/mod.rs index d437e06ba..bc56a3bc0 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/mod.rs +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/mod.rs @@ -1,9 +1,11 @@ pub mod clickhouse_result_adapter; pub mod execution; pub mod fallback; +pub mod plan_catalog; pub mod relational_adapter; pub mod request; pub mod server; +pub mod sql_binder; pub use fallback::{ClickHouseExactBackend, ClickHouseHttpFallback}; pub use server::{ diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/plan_catalog.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/plan_catalog.rs new file mode 100644 index 000000000..8467f31c8 --- /dev/null +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/plan_catalog.rs @@ -0,0 +1,353 @@ +//! Generation-aware SQL plan lookup, kept separate from SDS. +//! +//! Entries own backend execution templates and only hold foreign keys to the +//! immutable SDS snapshot. Staging validates all foreign keys before a whole +//! generation can become visible to readers. + +use std::collections::{BTreeMap, BTreeSet}; +use std::sync::{Arc, RwLock}; + +use asap_types::sds::{DataDescriptorId, SummaryDescriptorId}; +use asap_types::summary_catalog::{SummaryCatalog, SummaryCatalogReference}; +use xxhash_rust::xxh64::xxh64; + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct SqlQueryFingerprint(String); + +impl SqlQueryFingerprint { + pub fn canonical(&self) -> &str { + &self.0 + } +} + +/// Stable lookup identity for the exact SQL template submitted at planning. +/// SQL parsing and semantic canonicalization remain ASAPPlanner's job; keeping +/// the text alongside the hash makes hash collisions fail closed. +pub fn fingerprint_sql(sql: &str) -> SqlQueryFingerprint { + let key = sql.trim(); + SqlQueryFingerprint(format!( + "clickhouse-sql:v1:{:016x}", + xxh64(key.as_bytes(), 0) + )) +} + +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct SdsDescriptorReferences { + pub summaries: BTreeSet, + pub data: BTreeSet, +} + +impl SdsDescriptorReferences { + pub fn empty() -> Self { + Self { + summaries: BTreeSet::new(), + data: BTreeSet::new(), + } + } + + fn validate(&self, catalog: &SummaryCatalog) -> Result<(), SqlPlanCatalogError> { + for id in &self.summaries { + if !catalog.summary_descriptors.contains_key(id) { + return Err(SqlPlanCatalogError::MissingSummaryDescriptor( + id.canonical().to_owned(), + )); + } + } + for id in &self.data { + if !catalog.data_descriptors.contains_key(id) { + return Err(SqlPlanCatalogError::MissingDataDescriptor( + id.canonical().to_owned(), + )); + } + } + Ok(()) + } +} + +#[derive(Debug, Clone)] +pub struct SqlPlanEntry

{ + /// Collision guard and parameter-binding template identity. + pub sql_template: String, + /// Planner-produced execution DAG (or its runtime-owned compiled form). + pub plan: P, + pub descriptors: SdsDescriptorReferences, +} + +impl

SqlPlanEntry

{ + pub fn fingerprint(&self) -> SqlQueryFingerprint { + fingerprint_sql(&self.sql_template) + } +} + +#[derive(Debug, Clone)] +pub struct SqlPlanCatalogGeneration

{ + pub sds: SummaryCatalogReference, + pub catalog: Arc, + entries: BTreeMap>>, +} + +impl

SqlPlanCatalogGeneration

{ + pub fn build( + sds: &SummaryCatalog, + entries: impl IntoIterator>, + ) -> Result { + let sds_reference = sds + .reference() + .map_err(|error| SqlPlanCatalogError::InvalidSds(error.to_string()))?; + let mut indexed = BTreeMap::new(); + for entry in entries { + entry.descriptors.validate(sds)?; + let fingerprint = entry.fingerprint(); + if indexed + .insert(fingerprint.clone(), Arc::new(entry)) + .is_some() + { + return Err(SqlPlanCatalogError::DuplicateFingerprint( + fingerprint.canonical().to_owned(), + )); + } + } + Ok(Self { + sds: sds_reference, + catalog: Arc::new(sds.clone()), + entries: indexed, + }) + } + + pub fn lookup(&self, sql: &str) -> Option>> { + let normalized = sql.trim(); + self.entries + .get(&fingerprint_sql(normalized)) + .filter(|entry| entry.sql_template.trim() == normalized) + .cloned() + } + + pub fn len(&self) -> usize { + self.entries.len() + } +} + +#[derive(Debug, Clone, serde::Serialize, PartialEq, Eq)] +pub struct SqlPlanCatalogAck { + pub plan_id: u64, + pub plan_version: u64, + pub sds_snapshot_sha256: String, + pub entry_count: usize, + pub phase: SqlPlanCatalogPhase, +} + +#[derive(Debug, Clone, Copy, serde::Serialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum SqlPlanCatalogPhase { + Staged, + Active, +} + +impl SqlPlanCatalogAck { + fn for_generation

( + generation: &SqlPlanCatalogGeneration

, + phase: SqlPlanCatalogPhase, + ) -> Self { + Self { + plan_id: generation.sds.plan_id, + plan_version: generation.sds.plan_version, + sds_snapshot_sha256: generation.sds.snapshot_sha256.clone(), + entry_count: generation.len(), + phase, + } + } +} + +#[derive(Debug, thiserror::Error, PartialEq, Eq)] +pub enum SqlPlanCatalogError { + #[error("invalid SDS catalog: {0}")] + InvalidSds(String), + #[error("SQL plan references missing summary descriptor {0}")] + MissingSummaryDescriptor(String), + #[error("SQL plan references missing data descriptor {0}")] + MissingDataDescriptor(String), + #[error("duplicate SQL plan fingerprint {0}")] + DuplicateFingerprint(String), + #[error("another SQL plan generation is already staged")] + CandidateAlreadyStaged, + #[error("SQL plan generation {plan_id}/{plan_version} was not staged")] + NotStaged { plan_id: u64, plan_version: u64 }, + #[error("SQL plan generation {candidate} is not newer than active generation {active}")] + StaleGeneration { active: u64, candidate: u64 }, +} + +#[derive(Debug)] +struct CatalogState

{ + active: Option>>, + staged: Option>>, +} + +/// Two-phase publisher for SQL plans. A lookup snapshots one immutable active +/// generation, so readers cannot observe a mixture of SDS generations. +#[derive(Debug)] +pub struct SqlPlanCatalog

{ + state: RwLock>, +} + +impl

Default for SqlPlanCatalog

{ + fn default() -> Self { + Self { + state: RwLock::new(CatalogState { + active: None, + staged: None, + }), + } + } +} + +impl

SqlPlanCatalog

{ + pub fn stage( + &self, + generation: SqlPlanCatalogGeneration

, + ) -> Result { + let mut state = self.state.write().unwrap(); + if state.staged.is_some() { + return Err(SqlPlanCatalogError::CandidateAlreadyStaged); + } + if let Some(active) = state.active.as_ref() { + if generation.sds.plan_version <= active.sds.plan_version { + return Err(SqlPlanCatalogError::StaleGeneration { + active: active.sds.plan_version, + candidate: generation.sds.plan_version, + }); + } + } + let generation = Arc::new(generation); + let ack = + SqlPlanCatalogAck::for_generation(generation.as_ref(), SqlPlanCatalogPhase::Staged); + state.staged = Some(generation); + Ok(ack) + } + + pub fn activate( + &self, + plan_id: u64, + plan_version: u64, + ) -> Result { + let mut state = self.state.write().unwrap(); + let staged = state + .staged + .as_ref() + .filter(|generation| { + generation.sds.plan_id == plan_id && generation.sds.plan_version == plan_version + }) + .cloned() + .ok_or(SqlPlanCatalogError::NotStaged { + plan_id, + plan_version, + })?; + let ack = SqlPlanCatalogAck::for_generation(staged.as_ref(), SqlPlanCatalogPhase::Active); + state.active = Some(staged); + state.staged = None; + Ok(ack) + } + + pub fn active(&self) -> Option>> { + self.state.read().unwrap().active.clone() + } + + pub fn lookup(&self, sql: &str) -> Option>> { + self.active()?.lookup(sql) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use asap_types::{AggregationType, KeyByLabelNames, PrecomputeMaterialization, WindowKind}; + + fn sds(plan_version: u64) -> SummaryCatalog { + let config = PrecomputeMaterialization::new( + AggregationType::Sum, + String::new(), + Default::default(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 60, + 60, + WindowKind::Tumbling, + String::new(), + "requests".into(), + None, + None, + None, + ); + SummaryCatalog::from_materializations(7, plan_version, &[config]).unwrap() + } + + fn entry(catalog: &SummaryCatalog, sql: &str, plan: &str) -> SqlPlanEntry { + SqlPlanEntry { + sql_template: sql.into(), + plan: plan.into(), + descriptors: SdsDescriptorReferences { + summaries: catalog.summary_descriptors.keys().cloned().collect(), + data: catalog.data_descriptors.keys().cloned().collect(), + }, + } + } + + // Publication validates foreign keys before exposing an atomic generation. + #[test] + fn validates_stages_activates_and_looks_up() { + let sds = sds(3); + let generation = SqlPlanCatalogGeneration::build( + &sds, + [entry(&sds, "SELECT sum(value) FROM requests", "dag")], + ) + .unwrap(); + let catalog = SqlPlanCatalog::default(); + let staged = catalog.stage(generation).unwrap(); + assert_eq!((staged.plan_id, staged.plan_version), (7, 3)); + assert_eq!(staged.phase, SqlPlanCatalogPhase::Staged); + assert!(catalog.lookup("SELECT sum(value) FROM requests").is_none()); + let activated = catalog.activate(7, 3).unwrap(); + assert_eq!(activated.entry_count, 1); + assert_eq!(activated.phase, SqlPlanCatalogPhase::Active); + assert_eq!( + catalog + .lookup(" SELECT sum(value) FROM requests ") + .unwrap() + .plan, + "dag" + ); + } + + // A dangling descriptor reference rejects the whole candidate generation. + #[test] + fn rejects_missing_descriptor_and_stale_successor() { + let sds = sds(3); + let mut invalid = entry(&sds, "SELECT 1", "dag"); + invalid + .descriptors + .summaries + .insert(serde_json::from_value(serde_json::json!("missing-summary")).unwrap()); + assert!(matches!( + SqlPlanCatalogGeneration::build(&sds, [invalid]), + Err(SqlPlanCatalogError::MissingSummaryDescriptor(_)) + )); + + let catalog = SqlPlanCatalog::default(); + catalog + .stage(SqlPlanCatalogGeneration::build(&sds, [entry(&sds, "SELECT 1", "v3")]).unwrap()) + .unwrap(); + catalog.activate(7, 3).unwrap(); + assert_eq!( + catalog.stage( + SqlPlanCatalogGeneration::build(&sds, [entry(&sds, "SELECT 1", "another-v3")],) + .unwrap() + ), + Err(SqlPlanCatalogError::StaleGeneration { + active: 3, + candidate: 3 + }) + ); + assert_eq!(catalog.lookup("SELECT 1").unwrap().plan, "v3"); + } +} diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/sql_binder.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/sql_binder.rs new file mode 100644 index 000000000..21e73b2d9 --- /dev/null +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/sql_binder.rs @@ -0,0 +1,33 @@ +//! ClickHouse request binding through ASAPPlanner's SQL frontend. + +use control_plane::clickhouse::{ + plan_clickhouse_sql, ClickHousePlannedQuery, ClickHousePlanningError, ClickHouseSqlCatalog, +}; +use planner_types::types::AccuracyTarget; + +#[derive(Clone, Debug)] +pub struct ClickHouseSqlBinder { + catalog: ClickHouseSqlCatalog, + accuracy: AccuracyTarget, +} + +impl ClickHouseSqlBinder { + pub fn new(catalog: ClickHouseSqlCatalog, accuracy: AccuracyTarget) -> Self { + Self { catalog, accuracy } + } + + /// Parse SQL to the canonical `QueryExpr` and invoke the same physical + /// mapping used by PromQL. Summary selection remains planner-owned. + pub async fn bind(&self, sql: &str) -> Result { + plan_clickhouse_sql(sql, &self.catalog, self.accuracy.clone()).await + } + + pub async fn canonical_identity(&self, sql: &str) -> Result { + control_plane::clickhouse::canonicalize_clickhouse_sql( + sql, + &self.catalog, + self.accuracy.clone(), + ) + .await + } +} diff --git a/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs b/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs index 187f67aa7..9e53ddb55 100644 --- a/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs +++ b/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs @@ -6,7 +6,7 @@ use std::collections::{BTreeMap, BTreeSet}; use asap_types::sds::{SummaryDefinitionId, SummaryDescriptor, SummaryOperator}; use asap_types::summary_catalog::SummaryCatalog; use asap_types::AggregationType; -use control_plane::query_plan::{ExactReadout, QueryPlanEntry, QueryPlanNode, QueryReadout}; +use control_plane::query_plan::{ExactReadout, ExecutablePlanView, QueryPlanNode, QueryReadout}; use crate::query_engines::EngineError; @@ -108,9 +108,10 @@ impl ResolvedMaterialization<'_> { /// Capability matching follows installed state edges, including merges; it /// never searches the catalog for a replacement materialization. +#[cfg(test)] pub(crate) fn validate_entry( catalog: Option<&SummaryCatalog>, - entry: &QueryPlanEntry, + entry: &control_plane::query_plan::QueryPlanEntry, plan_id: u64, plan_version: u64, ) -> Result<(), EngineError> { @@ -119,7 +120,7 @@ pub(crate) fn validate_entry( pub(crate) fn validate_payload( catalog: Option<&SummaryCatalog>, - entry: &QueryPlanEntry, + entry: &dyn ExecutablePlanView, plan_id: u64, plan_version: u64, ) -> Result<(), EngineError> { @@ -133,7 +134,7 @@ pub(crate) fn validate_payload( let mut states = BTreeMap::new(); let mut resolved = BTreeMap::new(); for id in entry.topological_order().map_err(|e| miss(e.to_string()))? { - let node = &entry.nodes[&id]; + let node = &entry.nodes()[&id]; let state_ids = match node { QueryPlanNode::ReadMaterialization { binding } => { if !resolved.contains_key(&binding.materialization) { 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 655bc5642..6fab22ff4 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -1,9 +1,35 @@ use crate::storage_engines::types::StreamingConfig; +use control_plane::query_plan::ExecutablePlanView; use std::sync::Arc; use asap_types::query_requirements::QueryRequirements; use asap_types::KeyByLabelNames; +struct RootedPlan<'a> { + base: &'a dyn control_plane::query_plan::ExecutablePlanView, + root: control_plane::query_plan::QueryNodeId, + instant: control_plane::query_plan::InstantExecution, +} +impl control_plane::query_plan::ExecutablePlanView for RootedPlan<'_> { + fn root(&self) -> control_plane::query_plan::QueryNodeId { + self.root + } + fn nodes( + &self, + ) -> &std::collections::BTreeMap< + control_plane::query_plan::QueryNodeId, + control_plane::query_plan::QueryPlanNode, + > { + self.base.nodes() + } + fn instant(&self) -> &control_plane::query_plan::InstantExecution { + &self.instant + } + fn fallback(&self) -> &control_plane::query_plan::FallbackPolicy { + self.base.fallback() + } +} + #[derive(Clone)] struct QueryReadinessRequirement { materializations: Vec, @@ -11,7 +37,7 @@ struct QueryReadinessRequirement { } fn readiness_requirement( - entry: &control_plane::query_plan::QueryPlanEntry, + entry: &dyn control_plane::query_plan::ExecutablePlanView, ) -> QueryReadinessRequirement { let bindings = entry.materialization_bindings(); let mut materializations = bindings @@ -136,15 +162,17 @@ impl ASAPQueryEngine { "no active physical plan", ) })?; - let planned = physical - .query_plan - .lookup_canonical( - control_plane::query_plan::QueryLanguage::MetricsQl, - identity, - ) - .map_err(|error| { - crate::query_engines::EngineError::capability_miss("query_plan", error.to_string()) + let sidecar = physical + .metricsql_plan + .as_ref() + .and_then(|catalog| catalog.lookup(identity)) + .ok_or_else(|| { + crate::query_engines::EngineError::capability_miss( + "metricsql_plan", + "query not planned", + ) })?; + let planned = &sidecar.executable; let leaves = self.prepare_logical(&physical, planned, &[now_ms]).await?; let (mut result, mut stats) = self.execute_logical_entry(&physical, planned, &leaves, now_ms)?; @@ -168,15 +196,17 @@ impl ASAPQueryEngine { "no active physical plan", ) })?; - let planned = physical - .query_plan - .lookup_canonical( - control_plane::query_plan::QueryLanguage::MetricsQl, - identity, - ) - .map_err(|error| { - crate::query_engines::EngineError::capability_miss("query_plan", error.to_string()) + let sidecar = physical + .metricsql_plan + .as_ref() + .and_then(|catalog| catalog.lookup(identity)) + .ok_or_else(|| { + crate::query_engines::EngineError::capability_miss( + "metricsql_plan", + "query not planned", + ) })?; + let planned = &sidecar.executable; self.execute_logical_range(&physical, planned, start_ms, end_ms, step_ms) .await } @@ -224,10 +254,10 @@ impl ASAPQueryEngine { async fn prepare_logical( &self, physical: &crate::storage_engines::types::ActivePhysicalPlan, - entry: &control_plane::query_plan::QueryPlanEntry, + entry: &dyn control_plane::query_plan::ExecutablePlanView, times: &[u64], ) -> Result { - super::catalog_resolver::validate_entry( + super::catalog_resolver::validate_payload( physical.summary_catalog.as_deref(), entry, physical.query_plan.plan_id, @@ -250,15 +280,17 @@ impl ASAPQueryEngine { "candidate evaluation predates epoch", ) })?; - let mut subtree = entry.clone(); - subtree.root = input; - let reachable = subtree.topological_order().map_err(|error| { + let subtree = RootedPlan { + base: entry, + root: input, + instant: entry.instant().clone(), + }; + subtree.topological_order().map_err(|error| { crate::query_engines::EngineError::capability_miss( "installed_logical_dag", error.to_string(), ) })?; - subtree.nodes.retain(|id, _| reachable.contains(id)); let (result, _) = self.execute_logical_entry( physical, &subtree, @@ -288,7 +320,7 @@ impl ASAPQueryEngine { fn execute_logical_entry( &self, physical: &crate::storage_engines::types::ActivePhysicalPlan, - entry: &control_plane::query_plan::QueryPlanEntry, + entry: &dyn control_plane::query_plan::ExecutablePlanView, leaves: &super::logical_dag::PreparedLeaves, at: u64, ) -> Result< @@ -300,16 +332,17 @@ impl ASAPQueryEngine { > { use crate::query_engines::EngineError; super::logical_dag::execute_installed(entry, leaves, at, |root, evaluation_ms| { - let mut subtree = entry.clone(); - subtree.root = root; - let reachable = subtree.topological_order().map_err(|e| { + let reachable = entry.topological_order_from(root).map_err(|e| { EngineError::capability_miss("installed_logical_dag", e.to_string()) })?; - subtree.nodes.retain(|id, _| reachable.contains(id)); - let bindings: Vec<_> = subtree - .materialization_bindings() - .into_iter() - .cloned() + let bindings: Vec<_> = reachable + .iter() + .filter_map(|id| match &entry.nodes()[id] { + control_plane::query_plan::QueryPlanNode::ReadMaterialization { binding } => { + Some(binding.clone()) + } + _ => None, + }) .collect(); let windows: std::collections::BTreeSet> = bindings.iter().map(|b| b.readout_lookback_ms).collect(); @@ -319,13 +352,20 @@ impl ASAPQueryEngine { "bound subtree requires one explicit positive window", )); } - subtree.instant.lookback_ms = windows + let lookback_ms = windows .first() .copied() .flatten() .expect("explicit semantic lookback checked"); - subtree.instant.full_history = false; - subtree.instant.cumulative_readout = true; + let subtree = RootedPlan { + base: entry, + root, + instant: control_plane::query_plan::InstantExecution { + lookback_ms, + full_history: false, + cumulative_readout: true, + }, + }; let requirement = readiness_requirement(&subtree); let index = self.sketch_index.as_ref().ok_or_else(|| { EngineError::capability_miss("installed_logical_dag", "summary store unavailable") @@ -401,7 +441,7 @@ impl ASAPQueryEngine { async fn execute_logical_range( &self, physical: &crate::storage_engines::types::ActivePhysicalPlan, - entry: &control_plane::query_plan::QueryPlanEntry, + entry: &dyn control_plane::query_plan::ExecutablePlanView, start: u64, end: u64, step: u64, @@ -673,7 +713,7 @@ impl ASAPQueryEngine { { if let Some(physical) = self.physical_plan_snapshot() { if let Ok(entry) = physical.query_plan.lookup(query) { - if entry.nodes.values().any(|node| { + if entry.nodes().values().any(|node| { matches!( node, control_plane::query_plan::QueryPlanNode::Logical { .. } @@ -697,7 +737,7 @@ impl ASAPQueryEngine { let planned = match physical_plan.as_ref() { Some(physical_plan) => match physical_plan.query_plan.lookup(query) { Ok(query_entry) => { - super::catalog_resolver::validate_entry( + super::catalog_resolver::validate_payload( physical_plan.summary_catalog.as_deref(), query_entry, physical_plan.query_plan.plan_id, physical_plan.query_plan.plan_version, )?; @@ -3753,8 +3793,8 @@ mod range_stitch_tests { #[tokio::test] async fn active_metricsql_entry_reaches_the_shared_dag_executor() { use control_plane::query_plan::{ - FallbackPolicy, InstantExecution, QueryLanguage, QueryNodeId, QueryPlanEntry, - QueryPlanNode, + ExecutableQueryPlan, FallbackPolicy, InstantExecution, MetricsQlPlanCatalog, + MetricsQlPlanEntry, QueryNodeId, QueryPlanNode, }; let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_str(include_str!( @@ -3763,33 +3803,38 @@ mod range_stitch_tests { .unwrap(); let mut plan = snapshot.compile().unwrap(); let identity = asap_frontend_metricsql::canonical_metricsql("1 + 2").unwrap(); - plan.query_plan.entries.insert( - control_plane::query_plan::QueryPlan::catalog_key(QueryLanguage::MetricsQl, &identity), - QueryPlanEntry { - language: QueryLanguage::MetricsQl, - query_id: "vm-scalar".into(), - canonical_query: identity.clone(), - fixed_evaluation: None, - root: QueryNodeId(2), - nodes: std::collections::BTreeMap::from([ - (QueryNodeId(0), QueryPlanNode::Scalar { value: 1.0 }), - (QueryNodeId(1), QueryPlanNode::Scalar { value: 2.0 }), - ( - QueryNodeId(2), - QueryPlanNode::Binary { - inputs: [QueryNodeId(0), QueryNodeId(1)], - operator: planner_types::pre_asap::ArithmeticOpKind::Add, - }, - ), - ]), - instant: InstantExecution { - lookback_ms: 1, - full_history: false, - cumulative_readout: false, - }, - fallback: FallbackPolicy::ExactBackend, + let executable = ExecutableQueryPlan { + root: QueryNodeId(2), + nodes: std::collections::BTreeMap::from([ + (QueryNodeId(0), QueryPlanNode::Scalar { value: 1.0 }), + (QueryNodeId(1), QueryPlanNode::Scalar { value: 2.0 }), + ( + QueryNodeId(2), + QueryPlanNode::Binary { + inputs: [QueryNodeId(0), QueryNodeId(1)], + operator: planner_types::pre_asap::ArithmeticOpKind::Add, + }, + ), + ]), + instant: InstantExecution { + lookback_ms: 1, + full_history: false, + cumulative_readout: false, }, - ); + fallback: FallbackPolicy::ExactBackend, + }; + plan.metricsql_plan = Some(MetricsQlPlanCatalog { + plan_id: plan.query_plan.plan_id, + plan_version: plan.query_plan.plan_version, + entries: std::collections::BTreeMap::from([( + identity.clone(), + MetricsQlPlanEntry { + query_id: "vm-scalar".into(), + canonical_metricsql: identity.clone(), + executable, + }, + )]), + }); let mut active = crate::drivers::query::servers::http::build_active_physical_plan( crate::drivers::query::servers::http::PhysicalPlanInstallRequest { summary_catalog: plan.summary_catalog, @@ -3797,6 +3842,8 @@ mod range_stitch_tests { precompute_plan: plan.precompute_plan, transmission_plan: plan.transmission_plan, query_plan: plan.query_plan, + metricsql_plan: plan.metricsql_plan, + clickhouse_sql: None, storage_routing: None, adaptation_evidence: vec![], }, diff --git a/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs b/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs index dfa93b15a..1285a06c9 100644 --- a/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs +++ b/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs @@ -2,7 +2,7 @@ use super::logical_dag::{PreparedLeaf, PreparedLeaves, Value}; use crate::query_engines::EngineError; use control_plane::query_plan::{ - logical::LogicalOperator, QueryNodeId, QueryPlanEntry, QueryPlanNode, + logical::LogicalOperator, ExecutablePlanView, QueryNodeId, QueryPlanEntry, QueryPlanNode, }; use std::collections::{BTreeMap, BTreeSet}; @@ -15,13 +15,13 @@ fn miss(message: impl Into) -> EngineError { /// Traverse only the installed graph, including epoch-aligned nested subquery grids. fn leaves( - entry: &QueryPlanEntry, + entry: &dyn ExecutablePlanView, times: &[u64], ) -> Result, EngineError> { let mut pending = Vec::new(); for at in times { pending.push(( - entry.root, + entry.root(), i64::try_from(*at).map_err(|_| miss("timestamp overflow"))?, )); } @@ -35,7 +35,7 @@ fn leaves( return Err(miss("installed execution grid exceeds budget")); } let node = entry - .nodes + .nodes() .get(&id) .ok_or_else(|| miss("missing installed node"))?; match node { @@ -91,13 +91,13 @@ fn leaves( } pub(super) fn candidate_dependencies( - entry: &QueryPlanEntry, + entry: &dyn ExecutablePlanView, times: &[u64], ) -> Result, EngineError> { let mut result = Vec::new(); for ((id, at), operator) in leaves(entry, times)? { if let LogicalOperator::CandidateExactSubquery { item_label, .. } = operator { - let input = *entry.nodes[&id] + let input = *entry.nodes()[&id] .inputs() .first() .ok_or_else(|| miss("candidate exact subtree has no membership input"))?; @@ -259,7 +259,7 @@ fn parse_result(body: &serde_json::Value, at: i64) -> Result #[cfg(test)] pub(super) async fn prepare( - entry: &QueryPlanEntry, + entry: &dyn ExecutablePlanView, times: &[u64], endpoint: Option<&str>, client: &reqwest::Client, @@ -268,7 +268,7 @@ pub(super) async fn prepare( } pub(super) async fn prepare_with_candidates( - entry: &QueryPlanEntry, + entry: &dyn ExecutablePlanView, times: &[u64], endpoint: Option<&str>, client: &reqwest::Client, @@ -281,7 +281,7 @@ pub(super) async fn prepare_with_candidates( let (query, candidate_filtered) = match &operator { LogicalOperator::ExactSubquery { query } => (query.clone(), false), LogicalOperator::CandidateExactSubquery { query, item_label } => { - let candidate_input = entry.nodes[&id].inputs()[0]; + let candidate_input = entry.nodes()[&id].inputs()[0]; let candidate = prepared .get(&(candidate_input, at)) .ok_or_else(|| miss("candidate membership was not prepared"))?; @@ -383,10 +383,8 @@ mod tests { use control_plane::query_plan::{FallbackPolicy, InstantExecution}; fn entry(nodes: BTreeMap) -> QueryPlanEntry { QueryPlanEntry { - language: control_plane::query_plan::QueryLanguage::PromQl, query_id: "remote-cut".into(), - canonical_query: "a / b".into(), - fixed_evaluation: None, + canonical_promql: "a / b".into(), root: QueryNodeId(0), nodes, instant: InstantExecution { 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 aac4fca19..15e738d25 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 @@ -160,7 +160,7 @@ pub fn serve_instant_from_summary_executor( pub fn serve_from_query_plan( index: &SketchStore, - entry: &control_plane::query_plan::QueryPlanEntry, + entry: &dyn control_plane::query_plan::ExecutablePlanView, t0_ms: u64, t1_ms: u64, is_cumulative: bool, @@ -181,7 +181,7 @@ pub fn serve_from_query_plan( /// timestamps never leak into the public range response. pub fn serve_range_steps_from_query_plan( index: &SketchStore, - entry: &control_plane::query_plan::QueryPlanEntry, + entry: &dyn control_plane::query_plan::ExecutablePlanView, start_ms: u64, end_ms: u64, step_ms: u64, @@ -263,7 +263,7 @@ fn coverage_covers_closed_windows( pub fn serve_instant_from_query_plan( index: &SketchStore, - entry: &control_plane::query_plan::QueryPlanEntry, + entry: &dyn control_plane::query_plan::ExecutablePlanView, now_ms: u64, ) -> Result<(ASAPTierResult, u64), LoweringSkip> { if !summary_executor_live_enabled() { @@ -448,10 +448,8 @@ mod tests { ); } let entry = control_plane::query_plan::QueryPlanEntry { - language: control_plane::query_plan::QueryLanguage::PromQl, query_id: "q-sum".into(), - canonical_query: "sum_over_time(bytes[1s])".into(), - fixed_evaluation: None, + canonical_promql: "sum_over_time(bytes[1s])".into(), root: control_plane::query_plan::QueryNodeId(0), nodes: BTreeMap::from([ ( diff --git a/data_plane/src/query_engines/asap_query_engine/logical_dag.rs b/data_plane/src/query_engines/asap_query_engine/logical_dag.rs index b07ecf64a..2920daa43 100644 --- a/data_plane/src/query_engines/asap_query_engine/logical_dag.rs +++ b/data_plane/src/query_engines/asap_query_engine/logical_dag.rs @@ -8,7 +8,7 @@ use control_plane::query_plan::logical::{ Aggregation, BinaryOperation, Grouping, LogicalOperator, TemporalOperation, }; use control_plane::query_plan::{ - CandidateCompleteness, QueryNodeId, QueryPlanEntry, QueryPlanNode, + CandidateCompleteness, ExecutablePlanView, QueryNodeId, QueryPlanEntry, QueryPlanNode, }; use std::collections::{BTreeMap, BTreeSet}; @@ -86,7 +86,7 @@ pub(crate) struct PreparedLeaf { pub(crate) type PreparedLeaves = BTreeMap<(QueryNodeId, i64), PreparedLeaf>; pub(crate) fn execute_installed( - entry: &QueryPlanEntry, + entry: &dyn ExecutablePlanView, leaves: &PreparedLeaves, at: u64, callback: F, @@ -98,7 +98,7 @@ where } fn execute_values( - entry: &QueryPlanEntry, + entry: &dyn ExecutablePlanView, leaves: &PreparedLeaves, at: u64, callback: F, @@ -116,7 +116,7 @@ where warnings: Vec::new(), }; let at_signed = i64::try_from(at).map_err(|_| miss("evaluation timestamp overflow"))?; - let evaluated = evaluator.eval(entry.root, at_signed)?; + let evaluated = evaluator.eval(entry.root(), at_signed)?; if matches!(evaluated, Value::Scalar(_)) { // QueryResult currently models vectors/matrices only. Preserve a scalar // root's HTTP type by routing it to native, while scalar intermediates @@ -144,7 +144,7 @@ where } struct Evaluator<'a, F> { - entry: &'a QueryPlanEntry, + entry: &'a dyn ExecutablePlanView, leaves: &'a PreparedLeaves, stats: ExecutionStats, callback: F, @@ -178,7 +178,7 @@ impl Result> Evaluator<' } let node = self .entry - .nodes + .nodes() .get(&id) .ok_or_else(|| miss("missing installed node"))? .clone(); @@ -802,7 +802,7 @@ mod topk_tests { .unwrap(); control_plane::query_plan::logical::finalize_residuals(&mut entry).unwrap(); let leaf = entry - .nodes + .nodes() .iter() .find_map(|(id, node)| { matches!( @@ -856,10 +856,8 @@ mod topk_tests { let summary = QueryNodeId(0); let root = QueryNodeId(1); let entry = QueryPlanEntry { - language: control_plane::query_plan::QueryLanguage::PromQl, query_id: "summary-rate-topk".into(), - canonical_query: "topk(2, rate(requests_total[5m]))".into(), - fixed_evaluation: None, + canonical_promql: "topk(2, rate(requests_total[5m]))".into(), root, nodes: BTreeMap::from([ ( @@ -985,10 +983,8 @@ mod topk_tests { let value_id = QueryNodeId(1); let root = QueryNodeId(2); let entry = QueryPlanEntry { - language: control_plane::query_plan::QueryLanguage::PromQl, query_id: "candidate-topk".into(), - canonical_query: "topk(1, rate(requests_total[5m]))".into(), - fixed_evaluation: None, + canonical_promql: "topk(1, rate(requests_total[5m]))".into(), root, nodes: BTreeMap::from([ ( 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 index fa2448d0e..2b3e3aa03 100644 --- a/data_plane/src/query_engines/asap_query_engine/physical_dag.rs +++ b/data_plane/src/query_engines/asap_query_engine/physical_dag.rs @@ -6,7 +6,9 @@ use std::collections::BTreeMap; -use control_plane::query_plan::{QueryNodeId, QueryPlanEntry, QueryPlanNode}; +#[cfg(test)] +use control_plane::query_plan::QueryPlanEntry; +use control_plane::query_plan::{ExecutablePlanView, QueryNodeId, QueryPlanNode}; use thiserror::Error; pub trait QueryNodeRuntime { @@ -32,14 +34,14 @@ pub enum DagExecutionError { /// 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, + entry: &dyn ExecutablePlanView, runtime: &R, ) -> Result> { - execute_from(entry, entry.root, runtime) + execute_from(entry, entry.root(), runtime) } pub fn execute_from( - entry: &QueryPlanEntry, + entry: &dyn ExecutablePlanView, root: QueryNodeId, runtime: &R, ) -> Result> { @@ -49,7 +51,7 @@ pub fn execute_from( let mut outputs = BTreeMap::::new(); for id in order { let node = entry - .nodes + .nodes() .get(&id) .ok_or_else(|| DagExecutionError::InvalidGraph(format!("missing node {}", id.0)))?; let inputs = node @@ -141,10 +143,8 @@ mod tests { .into_iter() .collect(); let entry = QueryPlanEntry { - language: control_plane::query_plan::QueryLanguage::PromQl, query_id: "q".into(), - canonical_query: "up".into(), - fixed_evaluation: None, + canonical_promql: "up".into(), root, nodes, instant: InstantExecution { 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 b2c1d8f5a..787da8534 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 @@ -82,17 +82,27 @@ pub fn execute_post_asap_readout( /// Installed QueryPlan materialization resolution occurs before this legacy test helper. pub fn execute_query_plan_readout( index: &SketchStore, - entry: &control_plane::query_plan::QueryPlanEntry, + entry: &dyn control_plane::query_plan::ExecutablePlanView, t0_ms: u64, t1_ms: u64, is_cumulative: bool, ) -> Result { - execute_physical_query_payload(index, entry, entry.root, t0_ms, t1_ms, is_cumulative) + execute_physical_query_payload(index, entry, entry.root(), t0_ms, t1_ms, is_cumulative) +} + +pub fn execute_query_plan_payload_readout( + index: &SketchStore, + entry: &control_plane::query_plan::ExecutableQueryPlan, + t0_ms: u64, + t1_ms: u64, + is_cumulative: bool, +) -> Result { + execute_query_plan_readout(index, entry, t0_ms, t1_ms, is_cumulative) } pub fn execute_query_plan_from_readout( index: &SketchStore, - entry: &control_plane::query_plan::QueryPlanEntry, + entry: &dyn control_plane::query_plan::ExecutablePlanView, root: control_plane::query_plan::QueryNodeId, t0_ms: u64, t1_ms: u64, @@ -103,13 +113,13 @@ pub fn execute_query_plan_from_readout( pub fn execute_query_plan_instant( index: &SketchStore, - entry: &control_plane::query_plan::QueryPlanEntry, + entry: &dyn control_plane::query_plan::ExecutablePlanView, now_ms: u64, ) -> Result<(PostAsapReadoutOutcome, u64), LoweringSkip> { - let t0_ms = if entry.instant.full_history { + let t0_ms = if entry.instant().full_history { 0 } else { - now_ms.saturating_sub(entry.instant.lookback_ms) + now_ms.saturating_sub(entry.instant().lookback_ms) }; if entry .materialization_bindings() @@ -125,7 +135,7 @@ pub fn execute_query_plan_instant( entry, t0_ms, now_ms, - entry.instant.cumulative_readout, + entry.instant().cumulative_readout, )?; Ok((outcome, t0_ms)) } @@ -517,17 +527,17 @@ fn reduce_sum_values( fn execute_physical_query_plan( index: &SketchStore, - entry: &control_plane::query_plan::QueryPlanEntry, + entry: &dyn control_plane::query_plan::ExecutablePlanView, t0_ms: u64, t1_ms: u64, is_cumulative: bool, ) -> Result { - execute_physical_query_payload(index, entry, entry.root, t0_ms, t1_ms, is_cumulative) + execute_physical_query_payload(index, entry, entry.root(), t0_ms, t1_ms, is_cumulative) } fn execute_physical_query_payload( index: &SketchStore, - entry: &control_plane::query_plan::QueryPlanEntry, + entry: &dyn control_plane::query_plan::ExecutablePlanView, root: control_plane::query_plan::QueryNodeId, t0_ms: u64, t1_ms: u64, @@ -1048,10 +1058,8 @@ mod tests { } let entry = control_plane::query_plan::QueryPlanEntry { - language: control_plane::query_plan::QueryLanguage::PromQl, query_id: "q-rate".into(), - canonical_query: "rate(requests_total[1m])".into(), - fixed_evaluation: None, + canonical_promql: "rate(requests_total[1m])".into(), root: control_plane::query_plan::QueryNodeId(0), nodes: BTreeMap::from([ ( @@ -1144,10 +1152,8 @@ mod tests { idx.append_precompute(7, BTreeMap::new(), (0, 60_000), Box::new(accumulator)); let entry = control_plane::query_plan::QueryPlanEntry { - language: control_plane::query_plan::QueryLanguage::PromQl, query_id: "q-rate".into(), - canonical_query: "rate(requests_total[1m])".into(), - fixed_evaluation: None, + canonical_promql: "rate(requests_total[1m])".into(), root: control_plane::query_plan::QueryNodeId(0), nodes: BTreeMap::from([ ( 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 661d9dd6f..c8e6e9234 100644 --- a/data_plane/src/storage_engines/types/hot_reload_config.rs +++ b/data_plane/src/storage_engines/types/hot_reload_config.rs @@ -95,6 +95,8 @@ pub struct ActivePhysicalPlan { pub transmission_plan: control_plane::physical::compiler::TransmissionPlan, pub runtime_config: Arc, pub query_plan: Arc, + pub metricsql_plan: Option>, + pub clickhouse_sql: Option>, pub storage_routing: Arc, } @@ -690,9 +692,10 @@ mod tests { query_plan: Arc::new(control_plane::query_plan::QueryPlan { plan_id, plan_version, - clickhouse_context: None, entries: Default::default(), }), + metricsql_plan: None, + clickhouse_sql: None, storage_routing: Arc::new(crate::storage_engines::types::BackendStorageRouting::empty()), } } diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index c0e6a61dc..d5ffc08cf 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -168,6 +168,8 @@ async fn erp_measured_kll_collector_to_query_oracle() { precompute_plan: plan.precompute_plan, transmission_plan: plan.transmission_plan, query_plan: plan.query_plan, + metricsql_plan: None, + clickhouse_sql: None, storage_routing: None, adaptation_evidence: vec![], }; @@ -486,6 +488,8 @@ async fn registered_temporal_topk(algorithm: planner_types::post_asap::SketchAlg precompute_plan: plan.precompute_plan, transmission_plan: plan.transmission_plan, query_plan: plan.query_plan, + metricsql_plan: None, + clickhouse_sql: None, storage_routing: None, adaptation_evidence: vec![], }; 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 323057c44..28fc86ac9 100644 --- a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs +++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs @@ -86,7 +86,7 @@ async fn post_full_config(client: &reqwest::Client, stack: &FullStack, json: &Js .query_plan .entries .values_mut() - .filter(|e| e.canonical_query.starts_with("count(")) + .filter(|e| e.canonical_promql.starts_with("count(")) { entry.instant.lookback_ms = 1000; for node in entry.nodes.values_mut() { diff --git a/data_plane/tests/support/physical_fixture.rs b/data_plane/tests/support/physical_fixture.rs index 30de3d05d..46c0ab612 100644 --- a/data_plane/tests/support/physical_fixture.rs +++ b/data_plane/tests/support/physical_fixture.rs @@ -43,7 +43,6 @@ pub fn artifact(config: &StreamingConfig) -> PhysicalPlanInstallRequest { let mut query_plan = QueryPlan { plan_id: 1, plan_version: 1, - clickhouse_context: None, entries: BTreeMap::new(), }; for config in &precompute.materializations { @@ -104,10 +103,8 @@ pub fn artifact(config: &StreamingConfig) -> PhysicalPlanInstallRequest { query_plan.entries.insert( canonical.clone(), QueryPlanEntry { - language: control_plane::query_plan::QueryLanguage::PromQl, query_id: canonical.clone(), - canonical_query: canonical, - fixed_evaluation: None, + canonical_promql: canonical, root: QueryNodeId(1), nodes: BTreeMap::from([ ( @@ -147,6 +144,8 @@ pub fn artifact(config: &StreamingConfig) -> PhysicalPlanInstallRequest { precompute_plan: precompute, transmission_plan: transmission, query_plan, + metricsql_plan: None, + clickhouse_sql: None, storage_routing: None, adaptation_evidence: vec![], } diff --git a/data_plane/tests/victoriametrics_accelerated_e2e.rs b/data_plane/tests/victoriametrics_accelerated_e2e.rs index cd1479a89..e3d67bab4 100644 --- a/data_plane/tests/victoriametrics_accelerated_e2e.rs +++ b/data_plane/tests/victoriametrics_accelerated_e2e.rs @@ -163,17 +163,18 @@ async fn real_victoriametrics_ingest_published_dag_is_differential() { let plan = control_plane::physical::compiler::PhysicalCompiler .compile_metricsql(request, environment) .unwrap(); - assert!(plan - .query_plan + let metricsql_plan = plan.metricsql_plan.clone().expect("MetricsQL sidecar"); + let binding = metricsql_plan .entries .values() - .any(|entry| { entry.language == control_plane::query_plan::QueryLanguage::MetricsQl })); - let binding = plan - .query_plan - .entries - .values() - .find(|entry| entry.language == control_plane::query_plan::QueryLanguage::MetricsQl) - .and_then(|entry| entry.materialization_bindings().into_iter().next()) + .next() + .and_then(|entry| { + entry + .executable + .materialization_bindings() + .into_iter() + .next() + }) .expect("published MetricsQL DAG has a bound SummaryScan"); assert_eq!(binding.readout_lookback_ms, Some(5_000)); let artifact = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest { @@ -182,6 +183,8 @@ async fn real_victoriametrics_ingest_published_dag_is_differential() { precompute_plan: plan.precompute_plan, transmission_plan: plan.transmission_plan, query_plan: plan.query_plan, + metricsql_plan: plan.metricsql_plan, + clickhouse_sql: None, storage_routing: None, adaptation_evidence: vec![], }; @@ -189,11 +192,17 @@ async fn real_victoriametrics_ingest_published_dag_is_differential() { serde_json::from_value(serde_json::to_value(&artifact).unwrap()).unwrap(); assert_eq!( artifact - .query_plan + .metricsql_plan + .as_ref() + .unwrap() .entries .values() - .find(|entry| entry.language == control_plane::query_plan::QueryLanguage::MetricsQl) - .and_then(|entry| entry.materialization_bindings().into_iter().next()) + .next() + .and_then(|entry| entry + .executable + .materialization_bindings() + .into_iter() + .next()) .and_then(|binding| binding.readout_lookback_ms), Some(5_000), "MetricsQL effective lookback survives publication serialization" From 3bfbcbc30426951c3404ceaf1df8bb797fb100d2 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 02:21:13 -0600 Subject: [PATCH 4/6] Limit query plan imports to tests --- .../src/query_engines/asap_query_engine/exact_subqueries.rs | 4 +++- data_plane/src/query_engines/asap_query_engine/logical_dag.rs | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs b/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs index 1285a06c9..fa9e9b7b2 100644 --- a/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs +++ b/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs @@ -1,8 +1,10 @@ //! Fetch installed exact cuts from Prometheus before composing them with ASAP state. use super::logical_dag::{PreparedLeaf, PreparedLeaves, Value}; use crate::query_engines::EngineError; +#[cfg(test)] +use control_plane::query_plan::QueryPlanEntry; use control_plane::query_plan::{ - logical::LogicalOperator, ExecutablePlanView, QueryNodeId, QueryPlanEntry, QueryPlanNode, + logical::LogicalOperator, ExecutablePlanView, QueryNodeId, QueryPlanNode, }; use std::collections::{BTreeMap, BTreeSet}; diff --git a/data_plane/src/query_engines/asap_query_engine/logical_dag.rs b/data_plane/src/query_engines/asap_query_engine/logical_dag.rs index 2920daa43..2b253badd 100644 --- a/data_plane/src/query_engines/asap_query_engine/logical_dag.rs +++ b/data_plane/src/query_engines/asap_query_engine/logical_dag.rs @@ -7,8 +7,10 @@ use crate::storage_engines::types::KeyByLabelValues; use control_plane::query_plan::logical::{ Aggregation, BinaryOperation, Grouping, LogicalOperator, TemporalOperation, }; +#[cfg(test)] +use control_plane::query_plan::QueryPlanEntry; use control_plane::query_plan::{ - CandidateCompleteness, ExecutablePlanView, QueryNodeId, QueryPlanEntry, QueryPlanNode, + CandidateCompleteness, ExecutablePlanView, QueryNodeId, QueryPlanNode, }; use std::collections::{BTreeMap, BTreeSet}; From e0511a0566f71f1b607e756546b978dd3f52100a Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 02:43:46 -0600 Subject: [PATCH 5/6] Restore candidate TopK exact value witness --- control_plane/src/query_plan.rs | 54 ++++++++++++++++++++++++++++++++- 1 file changed, 53 insertions(+), 1 deletion(-) diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index 14c9050f2..b9681aca2 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -993,8 +993,60 @@ where }) }) .collect::, _>>()?; + let candidate_input = self.lower(candidates)?; + let value_input = if let Some(original) = &self.logical_source { + let parsed = promql_parser::parser::parse(original) + .map_err(|error| QueryPlanError::Invalid(error.to_string()))?; + let promql_parser::parser::Expr::Aggregate(aggregate) = parsed else { + return Err(QueryPlanError::Invalid( + "CandidateTopK requires a top-level PromQL aggregate".into(), + )); + }; + if aggregate.op.to_string() != "topk" { + return Err(QueryPlanError::Invalid( + "CandidateTopK requires a topk source expression".into(), + )); + } + fn item_label(node: &SummaryNode) -> Option { + match &node.expr { + SummaryExpr::SummaryEstimate { summary_input, .. } => { + item_label(summary_input) + } + SummaryExpr::SummaryAgg { input, .. } => match &input.item { + Some(planner_types::post_asap::SummaryInputExpr::Column( + planner_types::pre_asap::ColumnRef::Named(label), + )) => Some(label.clone()), + Some(planner_types::post_asap::SummaryInputExpr::Column( + planner_types::pre_asap::ColumnRef::Qualified { name, .. }, + )) => Some(name.clone()), + _ => None, + }, + _ => None, + } + } + let item_label = item_label(candidates).ok_or_else(|| { + QueryPlanError::Invalid( + "CandidateTopK membership has no named item label".into(), + ) + })?; + let value_id = QueryNodeId(self.next_id); + self.next_id += 1; + self.nodes.insert( + value_id, + QueryPlanNode::Logical { + operator: logical::LogicalOperator::CandidateExactSubquery { + query: aggregate.expr.to_string(), + item_label, + }, + inputs: vec![candidate_input], + }, + ); + value_id + } else { + self.lower(values)? + }; QueryPlanNode::CandidateTopK { - inputs: [self.lower(candidates)?, self.lower(values)?], + inputs: [candidate_input, value_input], k: u64::try_from(*k).map_err(|_| { QueryPlanError::Invalid("CandidateTopK k exceeds u64".into()) })?, From abe19211fb08ee3b10709a35271aa1840939054b Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 03:03:05 -0600 Subject: [PATCH 6/6] Expose ClickHouse acceleration route in responses --- .../asap_clickhouse_query_engine/server.rs | 61 +++++++++++++++++-- 1 file changed, 56 insertions(+), 5 deletions(-) diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/server.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/server.rs index c79d3b8c3..9f5c26b6d 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/server.rs +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/server.rs @@ -161,9 +161,11 @@ mod tests { request: &ClickHouseQueryRequest, ) -> Result { self.sql.lock().unwrap().push(request.sql.clone()); + let mut headers = HeaderMap::new(); + headers.insert("x-clickhouse-summary", "upstream".parse().unwrap()); Ok(ClickHouseRawResponse { status: StatusCode::OK, - headers: HeaderMap::new(), + headers, body: Bytes::from_static(b"1\n"), }) } @@ -191,6 +193,11 @@ mod tests { .await .unwrap(); assert_eq!(get_response.status(), StatusCode::OK); + assert_eq!( + get_response.headers().get("x-asap-execution").unwrap(), + "exact_fallback" + ); + assert_eq!(get_response.headers()["x-clickhouse-summary"], "upstream"); let post_response = app .clone() .oneshot( @@ -202,6 +209,11 @@ mod tests { ) .await .unwrap(); + assert_eq!( + post_response.headers()["x-asap-execution"], + "exact_fallback" + ); + assert_eq!(post_response.headers()["x-clickhouse-summary"], "upstream"); assert_eq!( post_response .into_body() @@ -211,6 +223,26 @@ mod tests { .to_bytes(), "1\n" ); + let grafana_response = app + .clone() + .oneshot( + axum::http::Request::builder() + .method("POST") + .uri("/?query=SELECT%203%20FORMAT%20JSON") + .header("user-agent", "Grafana") + .body(axum::body::Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!( + grafana_response.headers()["x-asap-execution"], + "exact_fallback" + ); + assert_eq!( + grafana_response.headers()["x-clickhouse-summary"], + "upstream" + ); let ping = app .oneshot( axum::http::Request::builder() @@ -224,7 +256,10 @@ mod tests { ping.into_body().collect().await.unwrap().to_bytes(), "Ok.\n" ); - assert_eq!(*fallback.sql.lock().unwrap(), vec!["SELECT 1", "SELECT 2"]); + assert_eq!( + *fallback.sql.lock().unwrap(), + vec!["SELECT 1", "SELECT 2", "SELECT 3 FORMAT JSON"] + ); } #[tokio::test] @@ -249,6 +284,8 @@ mod tests { .await .unwrap(); + assert_eq!(response.headers().get("x-asap-execution").unwrap(), "warm"); + assert_eq!( response.into_body().collect().await.unwrap().to_bytes(), "accelerated\n" @@ -283,6 +320,8 @@ mod tests { .unwrap(); assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.headers()["x-asap-execution"], "exact_fallback"); + assert_eq!(response.headers()["x-clickhouse-summary"], "upstream"); assert_eq!(*fallback.sql.lock().unwrap(), vec!["SELECT 1"]); } } @@ -339,17 +378,29 @@ async fn query_post( async fn execute_or_fallback(state: &ServerState, request: &ClickHouseQueryRequest) -> Response { match state.accelerator.execute(request).await { - ClickHouseAccelerationOutcome::Accelerated(response) => return raw_response(response), + ClickHouseAccelerationOutcome::Accelerated(response) => { + let mut response = raw_response(response); + response.headers_mut().insert( + "x-asap-execution", + "warm".parse().expect("static header value"), + ); + return response; + } ClickHouseAccelerationOutcome::Fallback(reason) => tracing::info!( failure_stage = reason.stage(), failure_reason = reason.reason_code(), "ClickHouse acceleration routed to exact fallback" ), } - match state.fallback.execute(request).await { + let mut response = match state.fallback.execute(request).await { Ok(v) => raw_response(v), Err(e) => (StatusCode::BAD_GATEWAY, e.to_string()).into_response(), - } + }; + response.headers_mut().insert( + "x-asap-execution", + "exact_fallback".parse().expect("static header value"), + ); + response } async fn ping(State(state): State) -> Response {