From 3b2c65118f0e4e63345b8644fb2c94d7269c0edc Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 05:21:59 -0600 Subject: [PATCH 01/20] Use recursive workload selection for SQL roots --- control_plane/src/clickhouse.rs | 48 +++++++++++++++++++++++++++++++-- 1 file changed, 46 insertions(+), 2 deletions(-) diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index ba350cb0b..35ed8a153 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -16,6 +16,7 @@ use crate::query_plan::{ use asap_types::summary_catalog::SummaryCatalog; use serde::{Deserialize, Serialize}; use std::collections::{BTreeSet, HashMap}; +use std::rc::Rc; #[derive(Debug, thiserror::Error)] pub enum ClickHousePlanningError { @@ -42,8 +43,24 @@ pub async fn plan_clickhouse_sql( // SQL keeps relational parents such as Project and Filter above a // summary-capable Aggregate. Use ASAPPlanner's recursive selector here; // the PromQL deployment lowering retains its existing conservative rules. - let cost_model = ControlPlaneCostModel::new(accuracy); - let selected = crate::planner_selection::select_summary(&canonical, &cost_model)?; + let cost_model = ControlPlaneCostModel::new(accuracy.clone()); + // SQL commonly keeps relational operators (Project/Filter/Sort/Limit) + // above the summary-capable aggregate. Workload search materializes those + // residual parents around the selected inner summary; root-only candidate + // selection incorrectly rejects such queries before recursive mapping. + let selected = crate::planner_selection::select_workload( + vec![(0, Rc::new(canonical.clone()))], + accuracy, + &cost_model, + )? + .into_iter() + .next() + .map(|(_, node)| node) + .ok_or_else(|| { + crate::planner_selection::SelectionError::Workload( + "SQL workload search returned no root".into(), + ) + })?; let physical = PhysicalExpr::committed(selected); Ok(ClickHousePlannedQuery { canonical_sql: canonical_sql_identity(&canonical), @@ -245,6 +262,33 @@ fn select_materialization<'a>( mod tests { use super::*; use asap_types::{AggregationType, KeyByLabelNames, PrecomputeMaterialization, WindowKind}; + use planner_types::pre_asap::{Column, DataType, Schema}; + + #[tokio::test] + async fn relational_sql_root_recursively_selects_inner_summary() { + let schema = Schema::with_time_index( + vec![ + Column::new("metric", DataType::Utf8, false), + Column::new("labels", DataType::Utf8, false), + Column::new("ts_ms", DataType::Timestamp, false), + Column::new("value", DataType::Float64, false), + ], + 2, + vec![], + ); + let catalog = SqlCatalog::new().with_table("raw_samples", schema); + let planned = plan_clickhouse_sql( + "SELECT labels, max(value) AS value FROM raw_samples \ + WHERE metric='cache_refresh_lag_seconds' \ + AND ts_ms>1788848096000 AND ts_ms<=1788891296000 \ + GROUP BY labels ORDER BY labels", + &catalog, + AccuracyTarget::Exact, + ) + .await + .expect("relational parents must be retained around the selected aggregate"); + assert!(matches!(planned.physical, PhysicalExpr::Committed(_))); + } fn materialization( agg: AggregationType, From b499439d0da58c8f9029feeda04c6b56969f203e Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 05:25:28 -0600 Subject: [PATCH 02/20] Bind exact ClickHouse SQL to equivalent planning SQL --- control_plane/src/clickhouse.rs | 46 ++++++++++++++++--- .../accelerator.rs | 32 ++++++------- .../plan_catalog.rs | 23 ++++++++-- .../query-engine/clickhouse-sql-support.md | 16 +++++++ 4 files changed, 89 insertions(+), 28 deletions(-) diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 35ed8a153..09c8c73dd 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -40,14 +40,11 @@ pub async fn plan_clickhouse_sql( let canonical = lower_sql_dialect(sql, catalog, SqlDialect::ClickhouseSQL, accuracy.clone()) .await .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?; - // SQL keeps relational parents such as Project and Filter above a - // summary-capable Aggregate. Use ASAPPlanner's recursive selector here; - // the PromQL deployment lowering retains its existing conservative rules. - let cost_model = ControlPlaneCostModel::new(accuracy.clone()); // SQL commonly keeps relational operators (Project/Filter/Sort/Limit) // above the summary-capable aggregate. Workload search materializes those // residual parents around the selected inner summary; root-only candidate // selection incorrectly rejects such queries before recursive mapping. + let cost_model = ControlPlaneCostModel::new(accuracy.clone()); let selected = crate::planner_selection::select_workload( vec![(0, Rc::new(canonical.clone()))], accuracy, @@ -80,6 +77,17 @@ pub async fn canonicalize_clickhouse_sql( Ok(canonical_sql_identity(&canonical)) } +/// Stable identity of the SQL text accepted by the ClickHouse endpoint. +/// +/// This intentionally does not parse the request. ClickHouse's executable SQL +/// surface is larger than ASAPPlanner's planning surface (tuple fields, array +/// lambdas, and engine-specific functions are common examples). Publication +/// may bind such an exact request template to a separately supplied, +/// semantically equivalent planning SQL expression. +pub fn sql_request_template_identity(sql: &str) -> String { + sql.trim().trim_end_matches(';').trim_end().to_owned() +} + /// Identity derived from ASAPPlanner's resolved canonical AST. Equivalent SQL /// formatting therefore maps to one catalog key without reparsing at serving. pub fn canonical_sql_identity(canonical: &QueryExpr) -> String { @@ -100,7 +108,16 @@ pub struct ClickHouseSqlWorkload { #[derive(Debug, Deserialize)] pub struct ClickHouseSqlWorkloadEntry { + /// Exact SQL template received by the data plane and sent to ClickHouse on + /// fallback. pub sql: String, + /// Semantically equivalent SQL written against Planner's supported SQL + /// surface. When absent, `sql` is planned directly. + /// + /// The control plane treats this as an explicit workload contract; it + /// never guesses equivalence by simplifying engine-specific expressions. + #[serde(default)] + pub planning_sql: Option, pub start_ms: u64, pub end_ms: u64, pub cumulative: bool, @@ -133,7 +150,12 @@ pub async fn compile_clickhouse_workload( }; 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 planned = plan_clickhouse_sql( + query.planning_sql.as_deref().unwrap_or(&query.sql), + &catalog, + request.accuracy.clone(), + ) + .await?; let PhysicalExpr::Committed(crate::physical::post_asap::PostAsapPlan::Summary(root)) = planned.physical else { @@ -181,7 +203,8 @@ pub async fn compile_clickhouse_workload( }) .collect::, _>>()?; plans.push(serde_json::json!({ - "sql": planned.canonical_sql, + "sql": sql_request_template_identity(&query.sql), + "planning_sql": planned.canonical_sql, "runtime": { "start_ms": query.start_ms, "end_ms": query.end_ms, "cumulative": query.cumulative, "materializations": materializations, @@ -258,6 +281,17 @@ fn select_materialization<'a>( Ok(selected) } +#[cfg(test)] +mod planning_tests { + use super::*; + + #[test] + fn request_identity_accepts_clickhouse_only_syntax_without_parsing() { + let sql = "SELECT samples[1].1, arraySum(i -> samples[i].2, range(1, 3)) FROM raw"; + assert_eq!(sql_request_template_identity(&format!(" {sql}; ")), sql); + } +} + #[cfg(test)] mod tests { use super::*; 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 170331137..80c18be54 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 @@ -40,7 +40,12 @@ pub struct SqlRuntimePlan { #[derive(Debug, serde::Serialize, serde::Deserialize)] pub struct ClickHousePublishedPlan { + /// Exact ClickHouse request/fallback template used as the lookup key. pub sql: String, + /// ASAPPlanner canonical identity of the explicitly equivalent planning + /// SQL. Retained on wire for audit; serving never executes it as fallback. + #[serde(default)] + pub planning_sql: String, pub runtime: SqlRuntimePlan, pub descriptors: SdsDescriptorReferences, } @@ -229,19 +234,15 @@ impl ClickHouseAccelerator for CatalogClickHouseAccelerator { .as_ref() .map(|s| s.binder.clone()) .or_else(|| self.binder.read().unwrap().clone()); - let Some(binder) = binder else { + if binder.is_none() { return ClickHouseAccelerationOutcome::Fallback( ClickHouseAccelerationFallback::CatalogMiss, ); - }; - let canonical_sql = match binder.canonical_identity(&request.sql).await { - Ok(canonical) => canonical, - Err(error) => { - return ClickHouseAccelerationOutcome::Fallback( - ClickHouseAccelerationFallback::Planning(error.to_string()), - ) - } - }; + } + // Lookup uses the exact ClickHouse request template published by the + // control plane. Do not reparse it here: publication may deliberately + // pair ClickHouse-only exact SQL with equivalent Planner SQL. + let canonical_sql = control_plane::clickhouse::sql_request_template_identity(&request.sql); let (entry, generation) = if let Some(sidecar) = sidecar { ( sidecar.catalog.lookup(&canonical_sql), @@ -507,21 +508,16 @@ mod tests { vec![], ); let tables = HashMap::from([("requests".into(), table_schema)]); - let canonical_sql = control_plane::clickhouse::canonicalize_clickhouse_sql( + let canonical_sql = control_plane::clickhouse::sql_request_template_identity( "SELECT sum(value) FROM requests", - &control_plane::clickhouse::ClickHouseSqlCatalog { - tables: tables.clone(), - }, - planner_types::types::AccuracyTarget::Exact, - ) - .await - .unwrap(); + ); let bundle = ClickHousePlanBundle { sds: sds.clone(), tables, accuracy: planner_types::types::AccuracyTarget::Exact, plans: vec![ClickHousePublishedPlan { sql: canonical_sql, + planning_sql: String::new(), runtime: SqlRuntimePlan { start_ms: 0, end_ms, 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 index 8467f31c8..dda2531c0 100644 --- 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 @@ -24,7 +24,7 @@ impl SqlQueryFingerprint { /// 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(); + let key = control_plane::clickhouse::sql_request_template_identity(sql); SqlQueryFingerprint(format!( "clickhouse-sql:v1:{:016x}", xxh64(key.as_bytes(), 0) @@ -115,10 +115,13 @@ impl

SqlPlanCatalogGeneration

{ } pub fn lookup(&self, sql: &str) -> Option>> { - let normalized = sql.trim(); + let normalized = control_plane::clickhouse::sql_request_template_identity(sql); self.entries - .get(&fingerprint_sql(normalized)) - .filter(|entry| entry.sql_template.trim() == normalized) + .get(&fingerprint_sql(&normalized)) + .filter(|entry| { + control_plane::clickhouse::sql_request_template_identity(&entry.sql_template) + == normalized + }) .cloned() } @@ -319,6 +322,18 @@ mod tests { ); } + #[test] + fn clickhouse_only_template_does_not_require_planner_parsing() { + let sds = sds(3); + let exact = "SELECT samples[1].1, arraySum(i -> samples[i].2, range(1, 3)) FROM raw"; + let generation = + SqlPlanCatalogGeneration::build(&sds, [entry(&sds, exact, "dag")]).unwrap(); + assert_eq!( + generation.lookup(&format!(" {exact}; ")).unwrap().plan, + "dag" + ); + } + // A dangling descriptor reference rejects the whole candidate generation. #[test] fn rejects_missing_descriptor_and_stale_successor() { diff --git a/docs/developer_docs/query-engine/clickhouse-sql-support.md b/docs/developer_docs/query-engine/clickhouse-sql-support.md index 4bfd87bbf..0f0185958 100644 --- a/docs/developer_docs/query-engine/clickhouse-sql-support.md +++ b/docs/developer_docs/query-engine/clickhouse-sql-support.md @@ -62,6 +62,22 @@ workload and uses the ordinary physical-plan stage and activate endpoints. The data plane has no ClickHouse-specific stage or activate endpoints and no SQL plan token or startup sidecar bundle. +Each workload query may carry two ClickHouse-private SQL strings: + +- `sql` is the exact request template. It is also the query sent to ClickHouse + when coverage or execution requires fallback. +- `planning_sql`, when present, is a semantically equivalent rewrite restricted + to ASAPPlanner's SQL surface. When absent, `sql` is planned directly. + +This split is useful for exact ClickHouse expressions containing tuple access, +array lambdas, or engine-specific functions. The control plane never guesses +that a simpler aggregate is equivalent. Workload generation owns that explicit +equivalence assertion. The data plane binds the normalized exact template to +the published DAG without reparsing ClickHouse-only syntax. Whitespace around +the request and one trailing semicolon are ignored; the SQL body otherwise has +to match exactly, so a different predicate or literal fails closed to +ClickHouse. + ## Execution and fallback The accelerated path supports the relational operations represented by the From 5e375f28bfcd3d73a2ca75d8ef4e8b429954fee2 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 05:30:42 -0600 Subject: [PATCH 03/20] Compile bounded ClickHouse table summaries --- control_plane/src/clickhouse.rs | 168 +++++++++++++++++++++++++++++++- 1 file changed, 165 insertions(+), 3 deletions(-) diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 09c8c73dd..9c28fdd9d 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -229,9 +229,8 @@ fn bind_selected_node( query: &ClickHouseSqlWorkloadEntry, request: &ClickHouseSqlWorkload, ) -> Result { - let (metric, source_window, spatial_filter) = - crate::physical::compiler::materialization_leaf_contract(node) - .map_err(crate::query_plan::QueryPlanError::Invalid)?; + let (metric, source_window, spatial_filter) = clickhouse_materialization_leaf_contract(node) + .map_err(crate::query_plan::QueryPlanError::Invalid)?; let expected = crate::physical::compiler::physical_materialization_family(family); let selected = select_materialization( &request.precompute_plan.materializations, @@ -250,6 +249,103 @@ fn bind_selected_node( }) } +/// Resolve the table-shaped SQL source without teaching the shared PromQL +/// materialization contract about SQL's metric and timestamp columns. +fn clickhouse_materialization_leaf_contract( + node: &planner_types::post_asap::SummaryNode, +) -> Result<(String, Option, String), String> { + use planner_types::{ + post_asap::SummaryExpr, + pre_asap::{CompareOpKind, QueryExpr, ScalarValue, Source}, + }; + let SummaryExpr::SummaryAgg { child, .. } = &node.expr else { + return crate::physical::compiler::materialization_leaf_contract(node); + }; + let SummaryExpr::KeepPreAsap(expr) = &child.expr else { + return crate::physical::compiler::materialization_leaf_contract(node); + }; + let QueryExpr::Scan { + source: Source::Table { .. }, + predicates, + schema, + } = expr.as_ref() + else { + return crate::physical::compiler::materialization_leaf_contract(node); + }; + + fn comparisons<'a>(expr: &'a QueryExpr, out: &mut Vec<&'a QueryExpr>) { + if let QueryExpr::BoolAnd(children) = expr { + for child in children { + comparisons(child, out); + } + } else { + out.push(expr); + } + } + let mut metric = None; + let mut lower_ms = None; + let mut upper_ms = None; + let mut leaves = Vec::new(); + for predicate in predicates { + comparisons(&predicate.0, &mut leaves); + } + for leaf in leaves { + let QueryExpr::Compare { left, op, right } = leaf else { + return Err("SQL materialization predicate is not a comparison".into()); + }; + let QueryExpr::Column(column) = left.as_ref() else { + return Err("SQL materialization predicate must reference a column".into()); + }; + let name = schema + .columns + .get(*column) + .map(|column| column.name.as_str()) + .ok_or_else(|| { + "SQL materialization predicate references an unknown column".to_string() + })?; + match (name, op, right.as_ref()) { + ("metric", CompareOpKind::Eq, QueryExpr::Literal(ScalarValue::Utf8(value))) => { + metric = Some(value.clone()); + } + ( + name, + CompareOpKind::Gt | CompareOpKind::Ge, + QueryExpr::Literal(ScalarValue::Int64(value)), + ) if schema + .time_index + .is_some_and(|index| schema.columns[index].name == name) => + { + lower_ms = Some(*value); + } + ( + name, + CompareOpKind::Lt | CompareOpKind::Le, + QueryExpr::Literal(ScalarValue::Int64(value)), + ) if schema + .time_index + .is_some_and(|index| schema.columns[index].name == name) => + { + upper_ms = Some(*value); + } + _ => { + return Err(format!( + "unsupported SQL materialization predicate on {name}" + )) + } + } + } + let metric = metric.ok_or_else(|| "SQL table summary requires metric='...'".to_string())?; + let window_secs = match (lower_ms, upper_ms) { + (Some(lower), Some(upper)) if upper > lower && (upper - lower) % 1_000 == 0 => { + Some((upper - lower) as u64 / 1_000) + } + _ => { + return Err("SQL table summary requires a positive whole-second timestamp range".into()) + } + }; + Ok((metric, window_secs, String::new())) +} + fn select_materialization<'a>( materializations: &'a [asap_types::PrecomputeMaterialization], metric: &str, @@ -324,6 +420,72 @@ mod tests { assert!(matches!(planned.physical, PhysicalExpr::Committed(_))); } + #[tokio::test] + async fn q05_max_over_time_sql_compiles_to_publishable_summary_dag() { + let schema = Schema::with_time_index( + vec![ + Column::new("metric", DataType::Utf8, false), + Column::new("labels", DataType::Utf8, false), + Column::new("ts_ms", DataType::Timestamp, false), + Column::new("value", DataType::Float64, false), + ], + 2, + vec![], + ); + let mut config = PrecomputeMaterialization::new( + AggregationType::MinMax, + "max".into(), + Default::default(), + KeyByLabelNames::new(vec!["labels".into()]), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 43_200, + 60, + WindowKind::Tumbling, + String::new(), + "cache_refresh_lag_seconds".into(), + None, + None, + None, + ); + config.pane_origin_ms = Some(0); + let sds = SummaryCatalog::from_materializations(71, 1, &[config.clone()]).unwrap(); + let envelope = crate::physical::compiler::PlanEnvelope { + plan_id: 71, + plan_version: 1, + generated_at_unix_ms: 0, + activation_unix_ms: 0, + expiry_unix_ms: None, + backend_compat: "test".into(), + planner_revision: "test".into(), + capability_snapshot_id: "test".into(), + }; + let mut precompute_plan = + PrecomputePlan::build_backend_local(envelope.clone(), vec![config]).unwrap(); + precompute_plan.summary_catalog = Some(sds.reference().unwrap()); + let mut transmission_plan = + TransmissionPlan::build(envelope, &precompute_plan, &Default::default()).unwrap(); + transmission_plan.summary_catalog = precompute_plan.summary_catalog.clone(); + let request = ClickHouseSqlWorkload { + sds, + precompute_plan, + transmission_plan, + tables: HashMap::from([("raw_samples".into(), schema)]), + accuracy: AccuracyTarget::Exact, + queries: vec![ClickHouseSqlWorkloadEntry { + sql: "SELECT labels, max(value) AS value FROM raw_samples WHERE metric='cache_refresh_lag_seconds' AND ts_ms>1788848096000 AND ts_ms<=1788891296000 GROUP BY labels ORDER BY labels".into(), + start_ms: 1_788_848_096_000, + end_ms: 1_788_891_296_000, + cumulative: true, + }], + }; + let bundle = compile_clickhouse_workload(&request) + .await + .expect("q05 SQL must compile for atomic sidecar publication"); + assert_eq!(bundle.plans.len(), 1); + } + fn materialization( agg: AggregationType, metric: &str, From c0c1b7e97631146453dd77d23670e96898c76db0 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 05:31:50 -0600 Subject: [PATCH 04/20] Fix ClickHouse workload test contract --- control_plane/src/clickhouse.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 9c28fdd9d..6d156a12a 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -475,6 +475,7 @@ mod tests { accuracy: AccuracyTarget::Exact, queries: vec![ClickHouseSqlWorkloadEntry { sql: "SELECT labels, max(value) AS value FROM raw_samples WHERE metric='cache_refresh_lag_seconds' AND ts_ms>1788848096000 AND ts_ms<=1788891296000 GROUP BY labels ORDER BY labels".into(), + planning_sql: None, start_ms: 1_788_848_096_000, end_ms: 1_788_891_296_000, cumulative: true, From cf9ba0a3676d0b94e06c3b0e46c7e4b9caa41ef1 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 05:35:43 -0600 Subject: [PATCH 05/20] Exercise compiled q05 SQL through shared executor --- .../accelerator.rs | 130 +++++++++++++++++- 1 file changed, 129 insertions(+), 1 deletion(-) 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 80c18be54..745e643b6 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 @@ -318,7 +318,7 @@ impl ClickHouseAccelerator for CatalogClickHouseAccelerator { mod tests { use super::*; use crate::{ - precompute_engine::operators::SumAccumulator, + precompute_engine::operators::{MinMaxAccumulator, SumAccumulator}, storage_engines::sketch_db::index::{AggKind, Capability, SketchInstanceMetadata}, }; use asap_types::{AggregationType, KeyByLabelNames, PrecomputeMaterialization, WindowKind}; @@ -334,6 +334,7 @@ mod tests { QueryExpr, ScalarValue, Schema, SortKey, }, }; + use std::collections::BTreeMap; use std::rc::Rc; fn relation_schema(names: &[(&str, DataType)]) -> SummarySchema { @@ -581,6 +582,133 @@ mod tests { assert_eq!(response.body, "1970-01-01T00:00:02\t50.0\n"); } + #[tokio::test] + async fn q05_compiled_sidecar_executes_grouped_max_as_typed_clickhouse_result() { + let start_ms = 1_788_848_096_000; + let end_ms = 1_788_891_296_000; + let mut config = PrecomputeMaterialization::new( + AggregationType::MinMax, + "max".into(), + Default::default(), + KeyByLabelNames::new(vec!["labels".into()]), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 43_200, + 43_200, + WindowKind::Tumbling, + String::new(), + "cache_refresh_lag_seconds".into(), + None, + Some("raw_samples".into()), + Some("value".into()), + ); + config.pane_origin_ms = Some(start_ms as i64); + let sds = SummaryCatalog::from_materializations(72, 1, &[config.clone()]).unwrap(); + let envelope = control_plane::physical::compiler::PlanEnvelope { + plan_id: 72, + plan_version: 1, + generated_at_unix_ms: 0, + activation_unix_ms: 0, + expiry_unix_ms: None, + backend_compat: "test".into(), + planner_revision: "test".into(), + capability_snapshot_id: "test".into(), + }; + let mut precompute = + control_plane::physical::compiler::PrecomputePlan::build_backend_local( + envelope.clone(), + vec![config.clone()], + ) + .unwrap(); + precompute.summary_catalog = Some(sds.reference().unwrap()); + let mut transmission = control_plane::physical::compiler::TransmissionPlan::build( + envelope, + &precompute, + &Default::default(), + ) + .unwrap(); + transmission.summary_catalog = precompute.summary_catalog.clone(); + let schema = Schema::with_time_index( + vec![ + Column::new("metric", DataType::Utf8, false), + Column::new("labels", DataType::Utf8, false), + Column::new("ts_ms", DataType::Timestamp, false), + Column::new("value", DataType::Float64, false), + ], + 2, + vec![], + ); + let sql = "SELECT labels, max(value) AS value FROM raw_samples WHERE metric='cache_refresh_lag_seconds' AND ts_ms>1788848096000 AND ts_ms<=1788891296000 GROUP BY labels ORDER BY labels"; + let compiled = control_plane::clickhouse::compile_clickhouse_workload( + &control_plane::clickhouse::ClickHouseSqlWorkload { + sds: sds.clone(), + precompute_plan: precompute, + transmission_plan: transmission, + tables: HashMap::from([("raw_samples".into(), schema)]), + accuracy: planner_types::types::AccuracyTarget::Exact, + queries: vec![control_plane::clickhouse::ClickHouseSqlWorkloadEntry { + sql: sql.into(), + start_ms, + end_ms, + cumulative: true, + }], + }, + ) + .await + .unwrap(); + let bundle: ClickHousePlanBundle = + serde_json::from_value(serde_json::to_value(compiled).unwrap()).unwrap(); + let materialization = config.policy_fingerprint(); + let store = Arc::new(SketchStore::new()); + store.install_summary_catalog(Arc::new(sds)).unwrap(); + store.register(SketchInstanceMetadata { + sid: 72, + metric_name: config.metric.clone(), + group_by_keys: BTreeSet::from(["labels".into()]), + capability: Some(Capability::ExactAgg(AggregationType::MinMax)), + agg_kind: AggKind::ExactAgg { + agg_type: AggregationType::MinMax, + parameters_canonical: String::new(), + spatial_filter_canonical: String::new(), + }, + accuracy: None, + first_seen_unix_ms: start_ms as i64, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: materialization, + }); + for (labels, value) in [("{instance=a}", 7.0), ("{instance=b}", 11.0)] { + store.append_precompute( + 72, + BTreeMap::from([("labels".into(), labels.into())]), + (start_ms, end_ms), + Box::new(MinMaxAccumulator::with_value(value, "max".into())), + ); + } + let accelerator = CatalogClickHouseAccelerator::from_bundle(bundle, store).unwrap(); + let request = ClickHouseQueryRequest { + method: Method::GET, + sql: sql.into(), + body: Bytes::new(), + parameters: Default::default(), + headers: HeaderMap::new(), + }; + let ClickHouseAccelerationOutcome::Accelerated(response) = + accelerator.execute(&request).await + else { + panic!("compiled q05 sidecar did not reach the warm executor") + }; + assert_eq!( + response.headers["content-type"], + "text/tab-separated-values; charset=UTF-8" + ); + assert_eq!( + std::str::from_utf8(&response.body).unwrap(), + "{instance=a}\t7.0\n{instance=b}\t11.0\n" + ); + } + #[tokio::test] async fn incomplete_summary_store_coverage_falls_back() { let (accelerator, request) = fixture(3_000).await; From 13e310ef566b331db3f0eca5ad64660f0362d552 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 05:37:21 -0600 Subject: [PATCH 06/20] Fix q05 sidecar test planning contract --- .../query_engines/asap_clickhouse_query_engine/accelerator.rs | 1 + 1 file changed, 1 insertion(+) 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 745e643b6..0a6a176a1 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 @@ -649,6 +649,7 @@ mod tests { accuracy: planner_types::types::AccuracyTarget::Exact, queries: vec![control_plane::clickhouse::ClickHouseSqlWorkloadEntry { sql: sql.into(), + planning_sql: None, start_ms, end_ms, cumulative: true, From 1bdc72a5c55faeec9a7bf8fc4b09331141941390 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 05:45:09 -0600 Subject: [PATCH 07/20] Adopt planner temporal SQL aggregates --- Cargo.lock | 36 ++++++++++++++++----------------- control_plane/Cargo.toml | 10 ++++----- control_plane/src/clickhouse.rs | 24 ++++++++++++++++++++++ crates/asap_types/Cargo.toml | 2 +- data_plane/Cargo.toml | 6 +++--- 5 files changed, 51 insertions(+), 27 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9acd1f3bc..b622ad310 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -374,7 +374,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=26e580710cf9b44b282598719d6c5ea9a9cc62be#26e580710cf9b44b282598719d6c5ea9a9cc62be" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=63455ab580baad914aa2b20a62aba620d7ed6e7b#63455ab580baad914aa2b20a62aba620d7ed6e7b" dependencies = [ "asap-types", "serde", @@ -385,7 +385,7 @@ dependencies = [ [[package]] name = "asap-frontend-metricsql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=26e580710cf9b44b282598719d6c5ea9a9cc62be#26e580710cf9b44b282598719d6c5ea9a9cc62be" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=63455ab580baad914aa2b20a62aba620d7ed6e7b#63455ab580baad914aa2b20a62aba620d7ed6e7b" dependencies = [ "asap-types", "metricsql_parser", @@ -395,7 +395,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=26e580710cf9b44b282598719d6c5ea9a9cc62be#26e580710cf9b44b282598719d6c5ea9a9cc62be" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=63455ab580baad914aa2b20a62aba620d7ed6e7b#63455ab580baad914aa2b20a62aba620d7ed6e7b" dependencies = [ "asap-types", "promql-parser 0.10.0", @@ -404,7 +404,7 @@ dependencies = [ [[package]] name = "asap-frontend-sql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=26e580710cf9b44b282598719d6c5ea9a9cc62be#26e580710cf9b44b282598719d6c5ea9a9cc62be" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=63455ab580baad914aa2b20a62aba620d7ed6e7b#63455ab580baad914aa2b20a62aba620d7ed6e7b" dependencies = [ "asap-sql-function-catalog", "asap-types", @@ -426,12 +426,12 @@ dependencies = [ [[package]] name = "asap-sql-function-catalog" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=26e580710cf9b44b282598719d6c5ea9a9cc62be#26e580710cf9b44b282598719d6c5ea9a9cc62be" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=63455ab580baad914aa2b20a62aba620d7ed6e7b#63455ab580baad914aa2b20a62aba620d7ed6e7b" [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=26e580710cf9b44b282598719d6c5ea9a9cc62be#26e580710cf9b44b282598719d6c5ea9a9cc62be" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=63455ab580baad914aa2b20a62aba620d7ed6e7b#63455ab580baad914aa2b20a62aba620d7ed6e7b" dependencies = [ "serde", "serde_json", @@ -1711,7 +1711,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2260,7 +2260,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.6.3", + "socket2 0.5.10", "tokio", "tower-service", "tracing", @@ -2453,7 +2453,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi 0.5.2", "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2801,7 +2801,7 @@ dependencies = [ [[package]] name = "metricsql_common" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=26e580710cf9b44b282598719d6c5ea9a9cc62be#26e580710cf9b44b282598719d6c5ea9a9cc62be" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=63455ab580baad914aa2b20a62aba620d7ed6e7b#63455ab580baad914aa2b20a62aba620d7ed6e7b" dependencies = [ "chrono", ] @@ -2809,7 +2809,7 @@ dependencies = [ [[package]] name = "metricsql_parser" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=26e580710cf9b44b282598719d6c5ea9a9cc62be#26e580710cf9b44b282598719d6c5ea9a9cc62be" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=63455ab580baad914aa2b20a62aba620d7ed6e7b#63455ab580baad914aa2b20a62aba620d7ed6e7b" dependencies = [ "ahash", "chrono", @@ -3433,7 +3433,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ "heck 0.5.0", - "itertools 0.13.0", + "itertools 0.10.5", "log", "multimap", "once_cell", @@ -3453,7 +3453,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools 0.13.0", + "itertools 0.10.5", "proc-macro2", "quote", "syn 2.0.117", @@ -3551,7 +3551,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls 0.23.40", - "socket2 0.6.3", + "socket2 0.5.10", "thiserror 2.0.18", "tokio", "tracing", @@ -3588,7 +3588,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.6.3", + "socket2 0.5.10", "tracing", "windows-sys 0.52.0", ] @@ -3904,7 +3904,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4438,7 +4438,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -5260,7 +5260,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.48.0", ] [[package]] diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 47d3bbf16..7948a4268 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -91,8 +91,8 @@ asap_types.workspace = true # scaffolding, unaware that `data_plane`'s `summary_executor.rs` in *this* # repo is a real one. Vendored locally instead of chased upstream -- see # `data_plane/src/query_engines/asap_query_engine/summary_exec.rs`. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" } -asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } +asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } # L1 adoption (design-target-architecture.md Part B): the PromQL front # end itself, replacing control_plane's own query_parser/promql.rs. @@ -100,9 +100,9 @@ asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = # `planner-types`/`asap-aware-mapping` above -- these three MUST move # together (two revs of the same upstream repo's types in one workspace # resolve to distinct Rust types that won't unify). -asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" } -asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" } -asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" } +asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } +asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } +asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 6d156a12a..37b8d2306 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -420,6 +420,30 @@ mod tests { assert!(matches!(planned.physical, PhysicalExpr::Committed(_))); } + #[tokio::test] + async fn temporal_sql_reuses_existing_rate_summary_intent() { + let schema = Schema::with_time_index( + vec![ + Column::new("metric", DataType::Utf8, false), + Column::new("labels", DataType::Utf8, false), + Column::new("ts_ms", DataType::Timestamp, false), + Column::new("value", DataType::Float64, false), + ], + 2, + vec![], + ); + let catalog = SqlCatalog::new().with_table("raw_samples", schema); + let planned = plan_clickhouse_sql( + "SELECT labels, asap_rate(value, ts_ms, 300000) AS value \ + FROM raw_samples WHERE metric='requests_total' GROUP BY labels", + &catalog, + AccuracyTarget::Exact, + ) + .await + .expect("explicit temporal SQL must use the shared rate DAG"); + assert!(matches!(planned.physical, PhysicalExpr::Committed(_))); + } + #[tokio::test] async fn q05_max_over_time_sql_compiles_to_publishable_summary_dag() { let schema = Schema::with_time_index( diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml index 80ac2800a..3d8e76bc4 100644 --- a/crates/asap_types/Cargo.toml +++ b/crates/asap_types/Cargo.toml @@ -32,4 +32,4 @@ sha2 = "0.10" # exactly (`control_plane/Cargo.toml`) -- two different revs of the same # git dependency in one workspace resolve to two distinct Rust types that # won't unify. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 38619a2b7..ff55c310e 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -38,8 +38,8 @@ control_plane = { path = "../control_plane" } # reduction: Reduction, .. }`) are `pre_asap` types, in the same crate now # (not a separate `asap-ir` import). Query serving consumes the compiled # QueryPlan; these types are used at physical-plan compilation boundaries. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" } -asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } +asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } # Shared external (workspace) serde.workspace = true @@ -130,7 +130,7 @@ crc32fast = "1.4" # none of them. [dev-dependencies] -asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" } +asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } tempfile = "3.20.0" criterion = { version = "0.5", features = ["html_reports"] } tokio-tungstenite = "0.21" From 06943398846e69ae4a9b93a383615f528ec1a0ae Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 05:47:49 -0600 Subject: [PATCH 08/20] Verify q05 warm SQL in a backend process --- .../asap_clickhouse_query_engine/server.rs | 1 + .../tests/clickhouse_q05_process_e2e.rs | 280 ++++++++++++++++++ 2 files changed, 281 insertions(+) create mode 100644 data_plane/tests/clickhouse_q05_process_e2e.rs 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 9f5c26b6d..675ae5082 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 @@ -389,6 +389,7 @@ async fn execute_or_fallback(state: &ServerState, request: &ClickHouseQueryReque ClickHouseAccelerationOutcome::Fallback(reason) => tracing::info!( failure_stage = reason.stage(), failure_reason = reason.reason_code(), + failure_detail = ?reason, "ClickHouse acceleration routed to exact fallback" ), } diff --git a/data_plane/tests/clickhouse_q05_process_e2e.rs b/data_plane/tests/clickhouse_q05_process_e2e.rs new file mode 100644 index 000000000..8f34ef9de --- /dev/null +++ b/data_plane/tests/clickhouse_q05_process_e2e.rs @@ -0,0 +1,280 @@ +//! Optional OS-process E2E for a q05-class bounded SQL max query. +//! Run with `CLICKHOUSE_URL=http://127.0.0.1:8123 cargo test -p data_plane --test clickhouse_q05_process_e2e`. + +use asap_types::{AggregationType, KeyByLabelNames, PrecomputeMaterialization, WindowKind}; +use control_plane::physical::compiler::{PlanEnvelope, PrecomputePlan, TransmissionPlan}; +use planner_types::pre_asap::{Column, DataType, Schema}; +use std::io::Read; +use std::{collections::HashMap, process::Stdio, time::Duration}; + +fn free_port() -> u16 { + std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port() +} + +fn clickhouse_auth(request: reqwest::RequestBuilder) -> reqwest::RequestBuilder { + match std::env::var("CLICKHOUSE_USER") { + Ok(user) => request.basic_auth(user, std::env::var("CLICKHOUSE_PASSWORD").ok()), + Err(_) => request, + } +} + +struct Child(std::process::Child); +impl Drop for Child { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +#[tokio::test] +async fn q05_sql_is_planned_backfilled_and_served_warm_by_backend_process() { + let Ok(clickhouse) = std::env::var("CLICKHOUSE_URL") else { + eprintln!("skipping q05 process E2E because CLICKHOUSE_URL is unset"); + return; + }; + let client = reqwest::Client::new(); + let end_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + let start_ms = end_ms - 43_200_000; + let setup = [ + "CREATE DATABASE IF NOT EXISTS asap_q05_e2e".to_string(), + "DROP TABLE IF EXISTS asap_q05_e2e.raw_samples".to_string(), + "CREATE TABLE asap_q05_e2e.raw_samples(metric String, labels String, ts_ms Int64, value Float64) ENGINE=Memory".to_string(), + format!("INSERT INTO asap_q05_e2e.raw_samples VALUES ('cache_refresh_lag_seconds','cache_refresh_lag_seconds',{},7),('cache_refresh_lag_seconds','cache_refresh_lag_seconds',{},11),('other','other',{},99)", start_ms + 1_000, start_ms + 2_000, start_ms + 2_000), + ]; + for sql in setup { + let response = clickhouse_auth(client.post(&clickhouse)) + .body(sql) + .send() + .await + .unwrap(); + assert!( + response.status().is_success(), + "{}", + response.text().await.unwrap() + ); + } + + let sql = format!("SELECT max(value) AS value FROM raw_samples WHERE metric='cache_refresh_lag_seconds' AND ts_ms>={start_ms} AND ts_ms<{end_ms}"); + let mut config = PrecomputeMaterialization::new( + AggregationType::MinMax, + "max".into(), + Default::default(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 43_200, + 43_200, + WindowKind::Tumbling, + String::new(), + "cache_refresh_lag_seconds".into(), + None, + Some("raw_samples".into()), + Some("value".into()), + ); + config.pane_origin_ms = Some(start_ms as i64); + let sds = control_plane::physical::summary_catalog::SummaryCatalog::from_materializations( + 73, + 1, + &[config.clone()], + ) + .unwrap(); + let envelope = PlanEnvelope { + plan_id: 73, + plan_version: 1, + generated_at_unix_ms: 0, + activation_unix_ms: 0, + expiry_unix_ms: None, + backend_compat: "test".into(), + planner_revision: "test".into(), + capability_snapshot_id: "test".into(), + }; + let mut precompute = + PrecomputePlan::build_backend_local(envelope.clone(), vec![config.clone()]).unwrap(); + precompute.summary_catalog = Some(sds.reference().unwrap()); + let mut transmission = + TransmissionPlan::build(envelope, &precompute, &Default::default()).unwrap(); + transmission.summary_catalog = precompute.summary_catalog.clone(); + let schema = Schema::with_time_index( + vec![ + Column::new("metric", DataType::Utf8, false), + Column::new("labels", DataType::Utf8, false), + Column::new("ts_ms", DataType::Timestamp, false), + Column::new("value", DataType::Float64, false), + ], + 2, + vec![], + ); + let compiled = control_plane::clickhouse::compile_clickhouse_workload( + &control_plane::clickhouse::ClickHouseSqlWorkload { + sds: sds.clone(), + precompute_plan: precompute.clone(), + transmission_plan: transmission.clone(), + tables: HashMap::from([("raw_samples".into(), schema)]), + accuracy: planner_types::types::AccuracyTarget::Exact, + queries: vec![control_plane::clickhouse::ClickHouseSqlWorkloadEntry { + sql: sql.clone(), + start_ms, + end_ms, + cumulative: true, + }], + }, + ) + .await + .unwrap(); + + let http_port = free_port(); + let mut sql_port = free_port(); + while sql_port == http_port { + sql_port = free_port(); + } + let streaming_config = format!( + "{}/examples/promql/streaming_config.yaml", + env!("CARGO_MANIFEST_DIR") + ); + let mut command = std::process::Command::new(env!("CARGO_BIN_EXE_data_plane")); + command.args([ + "--http-port", + &http_port.to_string(), + "--streaming-config", + &streaming_config, + "--clickhouse-http-port", + &sql_port.to_string(), + "--clickhouse-url", + &clickhouse, + "--clickhouse-database", + "asap_q05_e2e", + "--clickhouse-backfill-table", + "raw_samples", + "--clickhouse-backfill-database", + "asap_q05_e2e", + "--clickhouse-backfill-timestamp-column", + "ts_ms", + "--enable-backfill-worker", + "--output-dir", + "/tmp/asap-q05-process-e2e", + ]); + if let Ok(user) = std::env::var("CLICKHOUSE_USER") { + command.args(["--clickhouse-user", &user]); + } + if let Ok(password) = std::env::var("CLICKHOUSE_PASSWORD") { + command.args(["--clickhouse-password", &password]); + } + let mut child = Child( + command + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(), + ); + let base = format!("http://127.0.0.1:{http_port}"); + for _ in 0..100 { + if client.get(format!("{base}/health")).send().await.is_ok() { + break; + } + if let Some(status) = child.0.try_wait().unwrap() { + let mut stderr = String::new(); + child + .0 + .stderr + .as_mut() + .unwrap() + .read_to_string(&mut stderr) + .unwrap(); + panic!("backend exited before ready ({status}): {stderr}"); + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + let install = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest { + summary_catalog: sds, + collector_plans: vec![], + precompute_plan: precompute, + transmission_plan: transmission, + query_plan: control_plane::query_plan::QueryPlan { + plan_id: 73, + plan_version: 1, + entries: Default::default(), + }, + metricsql_plan: None, + clickhouse_sql: Some(serde_json::to_value(compiled).unwrap()), + storage_routing: None, + adaptation_evidence: vec![], + }; + let response = client + .post(format!("{base}/api/v1/physical-plan")) + .json(&install) + .send() + .await + .unwrap(); + assert!( + response.status().is_success(), + "{}", + response.text().await.unwrap() + ); + let response = client + .post(format!("{base}/api/v1/physical-plan/activate")) + .json(&serde_json::json!({"plan_id":73,"plan_version":1})) + .send() + .await + .unwrap(); + assert!( + response.status().is_success(), + "{}", + response.text().await.unwrap() + ); + let agg_id = config.policy_fp_u64(); + let response: serde_json::Value = client + .post(format!("{base}/api/v1/db/backfill")) + .json( + &serde_json::json!({"agg_id":agg_id,"start_ms":start_ms,"end_ms":end_ms, + "source":{"Prometheus":{"url":"clickhouse://configured"}},"windows_total":1}), + ) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + let job = response["job_id"] + .as_u64() + .unwrap_or_else(|| panic!("backfill was not accepted: {response}")); + for _ in 0..100 { + let status: serde_json::Value = client + .get(format!("{base}/api/v1/db/backfill/jobs/{job}")) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + match status["status"].as_str() { + Some("complete") => break, + Some("failed") => panic!("backfill failed: {status}"), + _ => tokio::time::sleep(Duration::from_millis(100)).await, + } + } + let warm = client + .post(format!("http://127.0.0.1:{sql_port}/")) + .body(sql) + .send() + .await + .unwrap(); + assert_eq!( + warm.headers()["x-asap-execution"], + "warm", + "fallback headers: {:?}", + warm.headers() + ); + let warm_value: f64 = warm.text().await.unwrap().trim().parse().unwrap(); + let exact_value: f64 = clickhouse_auth(client.post(&clickhouse)) + .body(format!("SELECT max(value) FROM asap_q05_e2e.raw_samples WHERE metric='cache_refresh_lag_seconds' AND ts_ms>={start_ms} AND ts_ms<{end_ms} FORMAT TabSeparated")) + .send().await.unwrap().text().await.unwrap().trim().parse().unwrap(); + assert_eq!(warm_value, exact_value); +} From 52044650bc3f3c14dbc95a9e4d30de813810a442 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 05:52:22 -0600 Subject: [PATCH 09/20] Bind temporal SQL table summaries --- control_plane/src/clickhouse.rs | 109 +++++++++++++++++++++++++++++--- 1 file changed, 99 insertions(+), 10 deletions(-) diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 37b8d2306..bd64858f1 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -264,12 +264,31 @@ fn clickhouse_materialization_leaf_contract( let SummaryExpr::KeepPreAsap(expr) = &child.expr else { return crate::physical::compiler::materialization_leaf_contract(node); }; - let QueryExpr::Scan { - source: Source::Table { .. }, - predicates, - schema, - } = expr.as_ref() - else { + fn table_scan( + expr: &QueryExpr, + ) -> Option<( + &[planner_types::pre_asap::Predicate], + &planner_types::pre_asap::Schema, + )> { + match expr { + QueryExpr::Project { child, .. } => table_scan(child), + QueryExpr::Scan { + source: Source::Table { .. }, + predicates, + schema, + } => Some((predicates, schema)), + _ => None, + } + } + let (source, explicit_window) = match expr.as_ref() { + QueryExpr::TimeRange { child, range } + if range.as_millis() > 0 && range.as_millis() % 1_000 == 0 => + { + (child.as_ref(), Some(range.as_secs())) + } + source => (source, None), + }; + let Some((predicates, schema)) = table_scan(source) else { return crate::physical::compiler::materialization_leaf_contract(node); }; @@ -335,15 +354,19 @@ fn clickhouse_materialization_leaf_contract( } } let metric = metric.ok_or_else(|| "SQL table summary requires metric='...'".to_string())?; - let window_secs = match (lower_ms, upper_ms) { + let inferred_window = match (lower_ms, upper_ms) { (Some(lower), Some(upper)) if upper > lower && (upper - lower) % 1_000 == 0 => { Some((upper - lower) as u64 / 1_000) } + (None, None) => None, _ => { - return Err("SQL table summary requires a positive whole-second timestamp range".into()) + return Err("SQL timestamp range must provide compatible lower and upper bounds".into()) } }; - Ok((metric, window_secs, String::new())) + let window_secs = explicit_window.or(inferred_window).ok_or_else(|| { + "SQL table summary requires a positive whole-second timestamp range".to_string() + })?; + Ok((metric, Some(window_secs), String::new())) } fn select_materialization<'a>( @@ -441,7 +464,73 @@ mod tests { ) .await .expect("explicit temporal SQL must use the shared rate DAG"); - assert!(matches!(planned.physical, PhysicalExpr::Committed(_))); + let PhysicalExpr::Committed(crate::physical::post_asap::PostAsapPlan::Summary(root)) = + planned.physical + else { + panic!("rate SQL did not produce a summary DAG") + }; + let mut aggregate = root.as_ref(); + while let planner_types::post_asap::SummaryExpr::ValueOperation { child, .. } = + &aggregate.expr + { + aggregate = child; + } + assert_eq!( + clickhouse_materialization_leaf_contract(aggregate).unwrap(), + ("requests_total".into(), Some(300), String::new()) + ); + + let mut config = PrecomputeMaterialization::new( + AggregationType::Increase, + String::new(), + Default::default(), + KeyByLabelNames::new(vec!["labels".into()]), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 300, + 300, + WindowKind::Tumbling, + String::new(), + "requests_total".into(), + None, + Some("raw_samples".into()), + Some("value".into()), + ); + config.pane_origin_ms = Some(0); + let sds = SummaryCatalog::from_materializations(74, 1, &[config.clone()]).unwrap(); + let envelope = crate::physical::compiler::PlanEnvelope { + plan_id: 74, + plan_version: 1, + generated_at_unix_ms: 0, + activation_unix_ms: 0, + expiry_unix_ms: None, + backend_compat: "test".into(), + planner_revision: "test".into(), + capability_snapshot_id: "test".into(), + }; + let mut precompute = + PrecomputePlan::build_backend_local(envelope.clone(), vec![config]).unwrap(); + precompute.summary_catalog = Some(sds.reference().unwrap()); + let mut transmission = + TransmissionPlan::build(envelope, &precompute, &Default::default()).unwrap(); + transmission.summary_catalog = precompute.summary_catalog.clone(); + let bundle = compile_clickhouse_workload(&ClickHouseSqlWorkload { + sds, + precompute_plan: precompute, + transmission_plan: transmission, + tables: HashMap::from([("raw_samples".into(), catalog.tables["raw_samples"].clone())]), + accuracy: AccuracyTarget::Exact, + queries: vec![ClickHouseSqlWorkloadEntry { + sql: "SELECT labels, asap_rate(value, ts_ms, 300000) AS value FROM raw_samples WHERE metric='requests_total' GROUP BY labels".into(), + start_ms: 0, + end_ms: 300_000, + cumulative: true, + }], + }) + .await + .expect("rate SQL must compile into a publishable shared DAG"); + assert_eq!(bundle.plans.len(), 1); } #[tokio::test] From f42ecad7aafd0f591cc4eb3e1e956efb603eea3b Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 05:57:40 -0600 Subject: [PATCH 10/20] Enforce SQL counter series identity --- Cargo.lock | 26 +++++++++++++------------- control_plane/Cargo.toml | 10 +++++----- control_plane/src/clickhouse.rs | 5 +++-- crates/asap_types/Cargo.toml | 2 +- data_plane/Cargo.toml | 6 +++--- 5 files changed, 25 insertions(+), 24 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index b622ad310..eb363dd23 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -374,7 +374,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=63455ab580baad914aa2b20a62aba620d7ed6e7b#63455ab580baad914aa2b20a62aba620d7ed6e7b" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b4e2750a894262b0b188b2377443fb637aa7ea7b#b4e2750a894262b0b188b2377443fb637aa7ea7b" dependencies = [ "asap-types", "serde", @@ -385,7 +385,7 @@ dependencies = [ [[package]] name = "asap-frontend-metricsql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=63455ab580baad914aa2b20a62aba620d7ed6e7b#63455ab580baad914aa2b20a62aba620d7ed6e7b" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b4e2750a894262b0b188b2377443fb637aa7ea7b#b4e2750a894262b0b188b2377443fb637aa7ea7b" dependencies = [ "asap-types", "metricsql_parser", @@ -395,7 +395,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=63455ab580baad914aa2b20a62aba620d7ed6e7b#63455ab580baad914aa2b20a62aba620d7ed6e7b" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b4e2750a894262b0b188b2377443fb637aa7ea7b#b4e2750a894262b0b188b2377443fb637aa7ea7b" dependencies = [ "asap-types", "promql-parser 0.10.0", @@ -404,7 +404,7 @@ dependencies = [ [[package]] name = "asap-frontend-sql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=63455ab580baad914aa2b20a62aba620d7ed6e7b#63455ab580baad914aa2b20a62aba620d7ed6e7b" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b4e2750a894262b0b188b2377443fb637aa7ea7b#b4e2750a894262b0b188b2377443fb637aa7ea7b" dependencies = [ "asap-sql-function-catalog", "asap-types", @@ -426,12 +426,12 @@ dependencies = [ [[package]] name = "asap-sql-function-catalog" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=63455ab580baad914aa2b20a62aba620d7ed6e7b#63455ab580baad914aa2b20a62aba620d7ed6e7b" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b4e2750a894262b0b188b2377443fb637aa7ea7b#b4e2750a894262b0b188b2377443fb637aa7ea7b" [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=63455ab580baad914aa2b20a62aba620d7ed6e7b#63455ab580baad914aa2b20a62aba620d7ed6e7b" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b4e2750a894262b0b188b2377443fb637aa7ea7b#b4e2750a894262b0b188b2377443fb637aa7ea7b" dependencies = [ "serde", "serde_json", @@ -1711,7 +1711,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -2453,7 +2453,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi 0.5.2", "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -2801,7 +2801,7 @@ dependencies = [ [[package]] name = "metricsql_common" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=63455ab580baad914aa2b20a62aba620d7ed6e7b#63455ab580baad914aa2b20a62aba620d7ed6e7b" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b4e2750a894262b0b188b2377443fb637aa7ea7b#b4e2750a894262b0b188b2377443fb637aa7ea7b" dependencies = [ "chrono", ] @@ -2809,7 +2809,7 @@ dependencies = [ [[package]] name = "metricsql_parser" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=63455ab580baad914aa2b20a62aba620d7ed6e7b#63455ab580baad914aa2b20a62aba620d7ed6e7b" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b4e2750a894262b0b188b2377443fb637aa7ea7b#b4e2750a894262b0b188b2377443fb637aa7ea7b" dependencies = [ "ahash", "chrono", @@ -3904,7 +3904,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -4438,7 +4438,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -5260,7 +5260,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 7948a4268..103ffbf7e 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -91,8 +91,8 @@ asap_types.workspace = true # scaffolding, unaware that `data_plane`'s `summary_executor.rs` in *this* # repo is a real one. Vendored locally instead of chased upstream -- see # `data_plane/src/query_engines/asap_query_engine/summary_exec.rs`. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } -asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } +asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } # L1 adoption (design-target-architecture.md Part B): the PromQL front # end itself, replacing control_plane's own query_parser/promql.rs. @@ -100,9 +100,9 @@ asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = # `planner-types`/`asap-aware-mapping` above -- these three MUST move # together (two revs of the same upstream repo's types in one workspace # resolve to distinct Rust types that won't unify). -asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } -asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } -asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } +asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } +asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } +asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index bd64858f1..5dba62e4e 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -427,7 +427,7 @@ mod tests { Column::new("value", DataType::Float64, false), ], 2, - vec![], + vec![vec![1]], ); let catalog = SqlCatalog::new().with_table("raw_samples", schema); let planned = plan_clickhouse_sql( @@ -453,7 +453,7 @@ mod tests { Column::new("value", DataType::Float64, false), ], 2, - vec![], + vec![vec![1, 2]], ); let catalog = SqlCatalog::new().with_table("raw_samples", schema); let planned = plan_clickhouse_sql( @@ -523,6 +523,7 @@ mod tests { accuracy: AccuracyTarget::Exact, queries: vec![ClickHouseSqlWorkloadEntry { sql: "SELECT labels, asap_rate(value, ts_ms, 300000) AS value FROM raw_samples WHERE metric='requests_total' GROUP BY labels".into(), + planning_sql: None, start_ms: 0, end_ms: 300_000, cumulative: true, diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml index 3d8e76bc4..fc895daa7 100644 --- a/crates/asap_types/Cargo.toml +++ b/crates/asap_types/Cargo.toml @@ -32,4 +32,4 @@ sha2 = "0.10" # exactly (`control_plane/Cargo.toml`) -- two different revs of the same # git dependency in one workspace resolve to two distinct Rust types that # won't unify. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index ff55c310e..4284ba8d2 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -38,8 +38,8 @@ control_plane = { path = "../control_plane" } # reduction: Reduction, .. }`) are `pre_asap` types, in the same crate now # (not a separate `asap-ir` import). Query serving consumes the compiled # QueryPlan; these types are used at physical-plan compilation boundaries. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } -asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } +asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } # Shared external (workspace) serde.workspace = true @@ -130,7 +130,7 @@ crc32fast = "1.4" # none of them. [dev-dependencies] -asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "63455ab580baad914aa2b20a62aba620d7ed6e7b" } +asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } tempfile = "3.20.0" criterion = { version = "0.5", features = ["html_reports"] } tokio-tungstenite = "0.21" From 0f497bcf3a560c5bbd80414c06b815e736c08fad Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 06:03:54 -0600 Subject: [PATCH 11/20] Fix q05 process planning contract --- data_plane/tests/clickhouse_q05_process_e2e.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/data_plane/tests/clickhouse_q05_process_e2e.rs b/data_plane/tests/clickhouse_q05_process_e2e.rs index 8f34ef9de..cb9b5ddf4 100644 --- a/data_plane/tests/clickhouse_q05_process_e2e.rs +++ b/data_plane/tests/clickhouse_q05_process_e2e.rs @@ -121,6 +121,7 @@ async fn q05_sql_is_planned_backfilled_and_served_warm_by_backend_process() { accuracy: planner_types::types::AccuracyTarget::Exact, queries: vec![control_plane::clickhouse::ClickHouseSqlWorkloadEntry { sql: sql.clone(), + planning_sql: None, start_ms, end_ms, cumulative: true, From bfde5cf8d3291bd03e2deb603a5589e95f4c83c7 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 06:30:30 -0600 Subject: [PATCH 12/20] Execute SQL joins over selected summaries --- Cargo.lock | 16 +- control_plane/Cargo.toml | 10 +- control_plane/src/clickhouse.rs | 79 ++++++++++ control_plane/src/emit/mod.rs | 1 + .../src/physical/colored_dag/allocator.rs | 7 + .../src/physical/colored_dag/emitter.rs | 1 + control_plane/src/physical/compiler.rs | 6 + control_plane/src/physical/post_asap/tests.rs | 1 + control_plane/src/query_plan.rs | 33 +++- control_plane/src/query_plan/logical.rs | 4 + crates/asap_types/Cargo.toml | 2 +- data_plane/Cargo.toml | 6 +- .../asap_clickhouse_query_engine/execution.rs | 149 ++++++++++++++++++ .../relational_adapter.rs | 99 ++++++++++++ .../asap_query_engine/post_asap_readout.rs | 3 +- .../asap_query_engine/summary_exec.rs | 1 + 16 files changed, 398 insertions(+), 20 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index eb363dd23..bb9118d8b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -374,7 +374,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b4e2750a894262b0b188b2377443fb637aa7ea7b#b4e2750a894262b0b188b2377443fb637aa7ea7b" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c3658#16c365852efce5cab50bb500c9024582a82edd19" dependencies = [ "asap-types", "serde", @@ -385,7 +385,7 @@ dependencies = [ [[package]] name = "asap-frontend-metricsql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b4e2750a894262b0b188b2377443fb637aa7ea7b#b4e2750a894262b0b188b2377443fb637aa7ea7b" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c3658#16c365852efce5cab50bb500c9024582a82edd19" dependencies = [ "asap-types", "metricsql_parser", @@ -395,7 +395,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b4e2750a894262b0b188b2377443fb637aa7ea7b#b4e2750a894262b0b188b2377443fb637aa7ea7b" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c3658#16c365852efce5cab50bb500c9024582a82edd19" dependencies = [ "asap-types", "promql-parser 0.10.0", @@ -404,7 +404,7 @@ dependencies = [ [[package]] name = "asap-frontend-sql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b4e2750a894262b0b188b2377443fb637aa7ea7b#b4e2750a894262b0b188b2377443fb637aa7ea7b" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c3658#16c365852efce5cab50bb500c9024582a82edd19" dependencies = [ "asap-sql-function-catalog", "asap-types", @@ -426,12 +426,12 @@ dependencies = [ [[package]] name = "asap-sql-function-catalog" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b4e2750a894262b0b188b2377443fb637aa7ea7b#b4e2750a894262b0b188b2377443fb637aa7ea7b" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c3658#16c365852efce5cab50bb500c9024582a82edd19" [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b4e2750a894262b0b188b2377443fb637aa7ea7b#b4e2750a894262b0b188b2377443fb637aa7ea7b" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c3658#16c365852efce5cab50bb500c9024582a82edd19" dependencies = [ "serde", "serde_json", @@ -2801,7 +2801,7 @@ dependencies = [ [[package]] name = "metricsql_common" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b4e2750a894262b0b188b2377443fb637aa7ea7b#b4e2750a894262b0b188b2377443fb637aa7ea7b" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c3658#16c365852efce5cab50bb500c9024582a82edd19" dependencies = [ "chrono", ] @@ -2809,7 +2809,7 @@ dependencies = [ [[package]] name = "metricsql_parser" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b4e2750a894262b0b188b2377443fb637aa7ea7b#b4e2750a894262b0b188b2377443fb637aa7ea7b" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c3658#16c365852efce5cab50bb500c9024582a82edd19" dependencies = [ "ahash", "chrono", diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 103ffbf7e..0959c32ce 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -91,8 +91,8 @@ asap_types.workspace = true # scaffolding, unaware that `data_plane`'s `summary_executor.rs` in *this* # repo is a real one. Vendored locally instead of chased upstream -- see # `data_plane/src/query_engines/asap_query_engine/summary_exec.rs`. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } -asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } +asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } # L1 adoption (design-target-architecture.md Part B): the PromQL front # end itself, replacing control_plane's own query_parser/promql.rs. @@ -100,9 +100,9 @@ asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = # `planner-types`/`asap-aware-mapping` above -- these three MUST move # together (two revs of the same upstream repo's types in one workspace # resolve to distinct Rust types that won't unify). -asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } -asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } -asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } +asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } +asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } +asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 5dba62e4e..f9f1997cc 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -534,6 +534,85 @@ mod tests { assert_eq!(bundle.plans.len(), 1); } + #[tokio::test] + async fn ratio_sql_compiles_two_rate_summaries_and_relational_join() { + let schema = Schema::with_time_index( + vec![ + Column::new("metric", DataType::Utf8, false), + Column::new("labels", DataType::Utf8, false), + Column::new("ts_ms", DataType::Timestamp, false), + Column::new("value", DataType::Float64, false), + ], + 2, + vec![vec![2, 1]], + ); + let configs = ["errors_total", "requests_total"].map(|metric| { + let mut config = PrecomputeMaterialization::new( + AggregationType::Increase, + String::new(), + Default::default(), + KeyByLabelNames::new(vec!["labels".into()]), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 300, + 300, + WindowKind::Tumbling, + String::new(), + metric.into(), + None, + Some("raw_samples".into()), + Some("value".into()), + ); + config.pane_origin_ms = Some(0); + config + }); + let sds = SummaryCatalog::from_materializations(75, 1, &configs).unwrap(); + let envelope = crate::physical::compiler::PlanEnvelope { + plan_id: 75, + plan_version: 1, + generated_at_unix_ms: 0, + activation_unix_ms: 0, + expiry_unix_ms: None, + backend_compat: "test".into(), + planner_revision: "test".into(), + capability_snapshot_id: "test".into(), + }; + let mut precompute = + PrecomputePlan::build_backend_local(envelope.clone(), configs.to_vec()).unwrap(); + precompute.summary_catalog = Some(sds.reference().unwrap()); + let mut transmission = + TransmissionPlan::build(envelope, &precompute, &Default::default()).unwrap(); + transmission.summary_catalog = precompute.summary_catalog.clone(); + let sql = "SELECT a.labels, a.v / b.v AS ratio FROM \ + (SELECT labels, asap_rate(value, ts_ms, 300000) AS v FROM raw_samples WHERE metric='errors_total' GROUP BY labels) a \ + INNER JOIN \ + (SELECT labels, asap_rate(value, ts_ms, 300000) AS v FROM raw_samples WHERE metric='requests_total' GROUP BY labels) b \ + ON a.labels=b.labels"; + let bundle = compile_clickhouse_workload(&ClickHouseSqlWorkload { + sds, + precompute_plan: precompute, + transmission_plan: transmission, + tables: HashMap::from([("raw_samples".into(), schema)]), + accuracy: AccuracyTarget::Exact, + queries: vec![ClickHouseSqlWorkloadEntry { + sql: sql.into(), + start_ms: 0, + end_ms: 300_000, + cumulative: true, + }], + }) + .await + .expect("ratio SQL must compile"); + let plan: crate::query_plan::ExecutableQueryPlan = + serde_json::from_value(bundle.plans[0]["runtime"]["executable"].clone()).unwrap(); + assert!(plan.nodes.values().any(|node| matches!( + node, + crate::query_plan::QueryPlanNode::RelationalJoin { .. } + ))); + assert_eq!(plan.materialization_bindings().len(), 2); + } + #[tokio::test] async fn q05_max_over_time_sql_compiles_to_publishable_summary_dag() { let schema = Schema::with_time_index( diff --git a/control_plane/src/emit/mod.rs b/control_plane/src/emit/mod.rs index af4fe16f9..9732e9ded 100644 --- a/control_plane/src/emit/mod.rs +++ b/control_plane/src/emit/mod.rs @@ -326,6 +326,7 @@ fn extract_from_node(node: &Rc) -> Option { // Not surfaced by any `Bind*` path yet (gated on rules that // haven't landed — see `deployment_expr.rs`'s module docs). SummaryExpr::BinaryOp { .. } + | SummaryExpr::RelationalJoin { .. } | SummaryExpr::SummaryJoin { .. } | SummaryExpr::SummarySubtract { .. } | SummaryExpr::SummaryDelete { .. } diff --git a/control_plane/src/physical/colored_dag/allocator.rs b/control_plane/src/physical/colored_dag/allocator.rs index b66dcff33..5cbf74a21 100644 --- a/control_plane/src/physical/colored_dag/allocator.rs +++ b/control_plane/src/physical/colored_dag/allocator.rs @@ -184,6 +184,13 @@ impl ThreeStageWalker { } StageId::Backend } + SummaryExpr::RelationalJoin { left, right, .. } => { + for child in [left, right] { + let (cid, _) = self.visit_l4node(child)?; + self.dag.edges.push((id, cid)); + } + StageId::Backend + } SummaryExpr::ValueOperation { child, timing, .. } => { let (cid, child_stage) = self.visit_l4node(child)?; self.dag.edges.push((id, cid)); diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs index 2bd4bcf29..9d230e89a 100644 --- a/control_plane/src/physical/colored_dag/emitter.rs +++ b/control_plane/src/physical/colored_dag/emitter.rs @@ -117,6 +117,7 @@ fn classify(expr: &PhysicalExpr) -> NodeKind<'_> { SummaryExpr::SummaryEstimate { query, .. } => NodeKind::SketchEstimate { query }, SummaryExpr::SummaryMerge { .. } => NodeKind::SketchMerge, SummaryExpr::BinaryOp { .. } + | SummaryExpr::RelationalJoin { .. } | SummaryExpr::CandidateTopK { .. } | SummaryExpr::ValueOperation { .. } | SummaryExpr::SummaryJoin { .. } diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index b53e2d08b..12d9181e3 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -2805,6 +2805,7 @@ fn summary_agg_metric(node: &SummaryNode) -> Option { inner: right, .. } + | SummaryExpr::RelationalJoin { left, right, .. } | SummaryExpr::SummarySubtract { left, right } | SummaryExpr::BinaryOp { lhs: left, @@ -2952,6 +2953,7 @@ fn requires_exact_erp_fallback( walk(inner, out); } SummaryExpr::SummarySubtract { left, right } + | SummaryExpr::RelationalJoin { left, right, .. } | SummaryExpr::BinaryOp { lhs: left, rhs: right, @@ -3766,6 +3768,10 @@ fn collect_selected_materializations( SummaryExpr::ValueOperation { child, .. } => { walk(child, readout, composable, grouping.clone(), selected)?; } + SummaryExpr::RelationalJoin { left, right, .. } => { + walk(left, readout, composable, grouping.clone(), selected)?; + walk(right, readout, composable, grouping.clone(), selected)?; + } SummaryExpr::BinaryOp { lhs, rhs, .. } if composable || crate::query_plan::exact_value_executable(node) => { diff --git a/control_plane/src/physical/post_asap/tests.rs b/control_plane/src/physical/post_asap/tests.rs index 745b2ab2c..3735861c5 100644 --- a/control_plane/src/physical/post_asap/tests.rs +++ b/control_plane/src/physical/post_asap/tests.rs @@ -120,6 +120,7 @@ fn node_is_archive(node: &Rc) -> bool { candidates, values, .. } => node_is_archive(candidates) || node_is_archive(values), SummaryExpr::SummarySubtract { left, right } + | SummaryExpr::RelationalJoin { left, right, .. } | SummaryExpr::BinaryOp { lhs: left, rhs: right, diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index b9681aca2..aca421445 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -598,6 +598,14 @@ pub enum PhysicalGrouping { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(tag = "op", rename_all = "snake_case", deny_unknown_fields)] pub enum QueryPlanNode { + RelationalJoin { + inputs: [QueryNodeId; 2], + join_kind: planner_types::pre_asap::JoinKind, + pred: serde_json::Value, + left_schema: planner_types::post_asap::SummarySchema, + right_schema: planner_types::post_asap::SummarySchema, + output_schema: planner_types::post_asap::SummarySchema, + }, Relational { input: QueryNodeId, operation: serde_json::Value, @@ -653,7 +661,7 @@ impl QueryPlanNode { Self::Scalar { .. } | Self::ReadMaterialization { .. } | Self::ExactFallback { .. } => { &[] } - Self::Binary { inputs, .. } => inputs, + Self::Binary { inputs, .. } | Self::RelationalJoin { inputs, .. } => inputs, Self::ReduceSum { input, .. } | Self::Relational { input, .. } | Self::SummaryEstimate { input, .. } @@ -752,7 +760,8 @@ where } } QueryPlanNode::CandidateTopK { inputs, .. } - | QueryPlanNode::Binary { inputs, .. } => { + | QueryPlanNode::Binary { inputs, .. } + | QueryPlanNode::RelationalJoin { inputs, .. } => { for input in inputs { *input = remap[input]; } @@ -811,6 +820,26 @@ where } let physical = match &node.expr { + SummaryExpr::RelationalJoin { + left, + right, + kind, + pred, + } if self.preserve_relational => QueryPlanNode::RelationalJoin { + inputs: [self.lower(left)?, self.lower(right)?], + join_kind: kind.clone(), + pred: serde_json::to_value(pred).map_err(|error| { + QueryPlanError::Invalid(format!( + "cannot serialize relational join predicate: {error}" + )) + })?, + left_schema: left.schema.clone(), + right_schema: right.schema.clone(), + output_schema: node.schema.clone(), + }, + SummaryExpr::RelationalJoin { .. } => QueryPlanNode::ExactFallback { + reason: "read-time relational join requires the relational compiler".into(), + }, SummaryExpr::ValueOperation { child, operation, .. } if self.preserve_relational diff --git a/control_plane/src/query_plan/logical.rs b/control_plane/src/query_plan/logical.rs index d1e57226a..3540a1cef 100644 --- a/control_plane/src/query_plan/logical.rs +++ b/control_plane/src/query_plan/logical.rs @@ -1287,6 +1287,10 @@ pub fn materialization_candidate_keys( visit(original, lhs, keys)?; visit(original, rhs, keys)?; } + SummaryExpr::RelationalJoin { left, right, .. } => { + visit(original, left, keys)?; + visit(original, right, keys)?; + } SummaryExpr::CandidateTopK { candidates, values, .. } => { diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml index fc895daa7..e41cf8026 100644 --- a/crates/asap_types/Cargo.toml +++ b/crates/asap_types/Cargo.toml @@ -32,4 +32,4 @@ sha2 = "0.10" # exactly (`control_plane/Cargo.toml`) -- two different revs of the same # git dependency in one workspace resolve to two distinct Rust types that # won't unify. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 4284ba8d2..c0204cd7b 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -38,8 +38,8 @@ control_plane = { path = "../control_plane" } # reduction: Reduction, .. }`) are `pre_asap` types, in the same crate now # (not a separate `asap-ir` import). Query serving consumes the compiled # QueryPlan; these types are used at physical-plan compilation boundaries. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } -asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } +asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } # Shared external (workspace) serde.workspace = true @@ -130,7 +130,7 @@ crc32fast = "1.4" # none of them. [dev-dependencies] -asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "b4e2750a894262b0b188b2377443fb637aa7ea7b" } +asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } tempfile = "3.20.0" criterion = { version = "0.5", features = ["html_reports"] } tokio-tungstenite = "0.21" 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 890a134c8..bb40d6889 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 @@ -24,6 +24,102 @@ pub enum ClickHouseDagFallback { }, ResultEncoding(String), } + +fn apply_relational_operation( + operation: serde_json::Value, + output_schema: &planner_types::post_asap::SummarySchema, + relation: ClickHouseRelation, +) -> Result { + let adapter = ClickHouseRelationalAdapter; + if let Some(filter) = operation.get("Filter") { + let predicate = filter + .get("pred") + .cloned() + .ok_or_else(|| "published Filter lacks pred".to_owned()) + .and_then(|value| serde_json::from_value(value).map_err(|error| error.to_string()))?; + return adapter + .apply_filter(&predicate, relation) + .map_err(|error| error.to_string()); + } + let operation: ValueOperation = + serde_json::from_value(operation).map_err(|error| error.to_string())?; + adapter + .apply_operation(&operation, output_schema, relation) + .map_err(|error| error.to_string()) +} + +fn execute_relation_subtree( + index: &SketchStore, + entry: &ExecutableQueryPlan, + root: QueryNodeId, + expected_schema: &planner_types::post_asap::SummarySchema, + t0_ms: u64, + t1_ms: u64, + is_cumulative: bool, +) -> Result { + match entry.nodes.get(&root) { + Some(QueryPlanNode::Relational { + input, + operation, + input_schema, + output_schema, + }) => { + let input = execute_relation_subtree( + index, + entry, + *input, + input_schema, + t0_ms, + t1_ms, + is_cumulative, + )?; + apply_relational_operation(operation.clone(), output_schema, input) + } + Some(QueryPlanNode::RelationalJoin { + inputs, + join_kind, + pred, + left_schema, + right_schema, + output_schema, + }) => { + if !matches!(join_kind, planner_types::pre_asap::JoinKind::Inner) { + return Err("only inner relational joins are executable".into()); + } + let left = execute_relation_subtree( + index, + entry, + inputs[0], + left_schema, + t0_ms, + t1_ms, + is_cumulative, + )?; + let right = execute_relation_subtree( + index, + entry, + inputs[1], + right_schema, + t0_ms, + t1_ms, + is_cumulative, + )?; + let pred = serde_json::from_value(pred.clone()).map_err(|error| error.to_string())?; + ClickHouseRelationalAdapter + .apply_inner_equi_join(&pred, output_schema, left, right) + .map_err(|error| error.to_string()) + } + Some(_) => { + let executable = reachable_entry(entry, root)?; + let outcome = + execute_query_plan_payload_readout(index, &executable, t0_ms, t1_ms, is_cumulative) + .map_err(|error| format!("{error:?}"))?; + ClickHouseRelation::from_series_rows(expected_schema, outcome.series, outcome.coverage) + .map_err(|error| error.to_string()) + } + None => Err(format!("published DAG references missing node {}", root.0)), + } +} pub enum ClickHouseDagOutcome { Accelerated(ClickHouseQueryResult), Fallback(ClickHouseDagFallback), @@ -55,6 +151,59 @@ pub fn execute_sql_dag( error.to_string(), )); } + let has_relational_join = entry.topological_order().is_ok_and(|ids| { + ids.iter().any(|id| { + matches!( + entry.nodes.get(id), + Some(QueryPlanNode::RelationalJoin { .. }) + ) + }) + }); + if has_relational_join { + let root_schema = match entry.nodes.get(&entry.root) { + Some(QueryPlanNode::Relational { output_schema, .. }) + | Some(QueryPlanNode::RelationalJoin { output_schema, .. }) => output_schema, + _ => { + return ClickHouseDagOutcome::Fallback(ClickHouseDagFallback::UnsupportedPlan( + "relational join plan root has no relation schema".into(), + )) + } + }; + let relation = match execute_relation_subtree( + index, + entry, + entry.root, + root_schema, + t0_ms, + t1_ms, + is_cumulative, + ) { + Ok(relation) => relation, + Err(error) => { + return ClickHouseDagOutcome::Fallback(ClickHouseDagFallback::UnsupportedPlan( + error, + )) + } + }; + let pane_ms = entry + .materialization_bindings() + .iter() + .map(|binding| binding.window_ms) + .max() + .unwrap_or(0); + if !complete_pane_coverage(relation.coverage, (t0_ms, t1_ms), pane_ms) { + return ClickHouseDagOutcome::Fallback(ClickHouseDagFallback::IncompleteCoverage { + requested: (t0_ms, t1_ms), + observed: relation.coverage, + }); + } + return match relation.into_result() { + Ok(result) => ClickHouseDagOutcome::Accelerated(result), + Err(error) => ClickHouseDagOutcome::Fallback(ClickHouseDagFallback::ResultEncoding( + error.to_string(), + )), + }; + } let mut base_root = entry.root; let mut relational = Vec::new(); loop { diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/relational_adapter.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/relational_adapter.rs index 69402b658..7ddef54c8 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/relational_adapter.rs +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/relational_adapter.rs @@ -93,6 +93,39 @@ impl ClickHouseRelation { pub struct ClickHouseRelationalAdapter; impl ClickHouseRelationalAdapter { + pub fn apply_inner_equi_join( + &self, + pred: &planner_types::pre_asap::Predicate, + output_schema: &SummarySchema, + left: ClickHouseRelation, + right: ClickHouseRelation, + ) -> Result { + let coverage = match (left.coverage, right.coverage) { + (Some((left_start, left_end)), Some((right_start, right_end))) => { + let start = left_start.max(right_start); + let end = left_end.min(right_end); + (start <= end).then_some((start, end)) + } + _ => None, + }; + let mut rows = Vec::new(); + for left_row in &left.rows { + for right_row in &right.rows { + let mut joined = Vec::with_capacity(left_row.len() + right_row.len()); + joined.extend(left_row.iter().cloned()); + joined.extend(right_row.iter().cloned()); + if matches!(eval(&pred.0, &joined)?, Cell::Bool(true)) { + rows.push(joined); + } + } + } + Ok(ClickHouseRelation { + rows, + fields: fields_from_schema(output_schema), + coverage, + }) + } + pub fn apply_filter( &self, pred: &planner_types::pre_asap::Predicate, @@ -646,4 +679,70 @@ mod tests { let error = eval(&QueryExpr::BoolAnd(vec![]), &row).unwrap_err(); assert!(matches!(error, ClickHouseRelationalError::Unsupported(_))); } + + #[test] + fn inner_equi_join_feeds_typed_ratio_projection() { + let side_schema = schema(&[("service", DataType::Utf8), ("value", DataType::Float64)]); + let left = ClickHouseRelation { + rows: vec![vec![Cell::Utf8("api".into()), Cell::Float64(2.0)]], + fields: fields_from_schema(&side_schema), + coverage: Some((300_000, 600_000)), + }; + let right = ClickHouseRelation { + rows: vec![vec![Cell::Utf8("api".into()), Cell::Float64(10.0)]], + fields: fields_from_schema(&side_schema), + coverage: Some((300_000, 600_000)), + }; + let joined_schema = schema(&[ + ("service", DataType::Utf8), + ("left_value", DataType::Float64), + ("service", DataType::Utf8), + ("right_value", DataType::Float64), + ]); + let pred = planner_types::pre_asap::Predicate(Rc::new(QueryExpr::Compare { + left: Rc::new(QueryExpr::Column(0)), + op: CompareOpKind::Eq, + right: Rc::new(QueryExpr::Column(2)), + })); + let joined = ClickHouseRelationalAdapter + .apply_inner_equi_join(&pred, &joined_schema, left, right) + .unwrap(); + let output_schema = schema(&[("service", DataType::Utf8), ("ratio", DataType::Float64)]); + let projected = ClickHouseRelationalAdapter + .apply_operation( + &ValueOperation::Project { + cols: vec![ + ProjectItem { + alias: Some("service".into()), + expr: QueryExpr::Column(0), + }, + ProjectItem { + alias: Some("ratio".into()), + expr: QueryExpr::Arithmetic { + op: ArithmeticOpKind::Div, + left: Rc::new(QueryExpr::Column(1)), + right: Rc::new(QueryExpr::Column(3)), + }, + }, + ], + qualifier: None, + }, + &output_schema, + joined, + ) + .unwrap(); + assert_eq!(projected.coverage, Some((300_000, 600_000))); + let result = projected.into_result().unwrap(); + let batch = &result.batches[0]; + assert_eq!(batch.schema().field(1).name(), "ratio"); + assert_eq!( + batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 0.2 + ); + } } 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 787da8534..cd39121a0 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 @@ -297,7 +297,8 @@ impl QueryNodeRuntime for PhysicalQueryRuntime<'_> { } QueryPlanNode::Logical { .. } | QueryPlanNode::CandidateTopK { .. } - | QueryPlanNode::Relational { .. } => Err(PhysicalNodeError::Fallback( + | QueryPlanNode::Relational { .. } + | QueryPlanNode::RelationalJoin { .. } => Err(PhysicalNodeError::Fallback( "logical node requires installed logical runtime".into(), )), QueryPlanNode::ExactFallback { reason } => { diff --git a/data_plane/src/query_engines/asap_query_engine/summary_exec.rs b/data_plane/src/query_engines/asap_query_engine/summary_exec.rs index 5f8c30294..4ab734ea4 100644 --- a/data_plane/src/query_engines/asap_query_engine/summary_exec.rs +++ b/data_plane/src/query_engines/asap_query_engine/summary_exec.rs @@ -244,6 +244,7 @@ pub fn execute( } SummaryExpr::SummaryJoin { .. } => Err(ExecError::NotYetSupported("SummaryJoin")), + SummaryExpr::RelationalJoin { .. } => Err(ExecError::NotYetSupported("RelationalJoin")), // CandidateTopK is lowered to the deployed QueryPlan DAG, where both // row inputs retain labels for intersection and exact reranking. This // legacy generic adapter exposes opaque GroupKey values and cannot From cbe6189f0fc96ed6026454cd33ac8291d045930e Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 06:32:08 -0600 Subject: [PATCH 13/20] Use keyed lookup for relational equi joins --- .../relational_adapter.rs | 56 ++++++++++++++++++- 1 file changed, 54 insertions(+), 2 deletions(-) diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/relational_adapter.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/relational_adapter.rs index 7ddef54c8..6036a0ab8 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/relational_adapter.rs +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/relational_adapter.rs @@ -1,6 +1,10 @@ //! ClickHouse row semantics for planner-owned relational wrappers. -use std::{cmp::Ordering, collections::BTreeMap, sync::Arc}; +use std::{ + cmp::Ordering, + collections::{BTreeMap, HashMap}, + sync::Arc, +}; use arrow::{ array::{ @@ -46,6 +50,17 @@ enum Cell { Timestamp(i64), } +fn join_key(value: &Cell) -> String { + match value { + Cell::Null => "null".into(), + Cell::Int64(value) => format!("i:{value}"), + Cell::Float64(value) => format!("f:{:016x}", value.to_bits()), + Cell::Utf8(value) => format!("s:{value}"), + Cell::Bool(value) => format!("b:{value}"), + Cell::Timestamp(value) => format!("t:{value}"), + } +} + #[derive(Debug)] pub struct ClickHouseRelation { rows: Vec>, @@ -108,9 +123,46 @@ impl ClickHouseRelationalAdapter { } _ => None, }; + let QueryExpr::Compare { + left: key_left, + op: CompareOpKind::Eq, + right: key_right, + } = pred.0.as_ref() + else { + return Err(ClickHouseRelationalError::Unsupported( + "join predicate is not equality".into(), + )); + }; + let (QueryExpr::Column(left_key), QueryExpr::Column(right_key)) = + (key_left.as_ref(), key_right.as_ref()) + else { + return Err(ClickHouseRelationalError::Unsupported( + "join keys are not columns".into(), + )); + }; + let right_local_key = right_key.checked_sub(left.fields.len()).ok_or_else(|| { + ClickHouseRelationalError::Invalid("right join key points into left input".into()) + })?; + let mut right_index: HashMap>> = HashMap::new(); + for row in &right.rows { + let key = + row.get(right_local_key) + .ok_or(ClickHouseRelationalError::ColumnOutOfRange( + right_local_key, + row.len(), + ))?; + right_index.entry(join_key(key)).or_default().push(row); + } let mut rows = Vec::new(); for left_row in &left.rows { - for right_row in &right.rows { + let key = + left_row + .get(*left_key) + .ok_or(ClickHouseRelationalError::ColumnOutOfRange( + *left_key, + left_row.len(), + ))?; + for right_row in right_index.get(&join_key(key)).into_iter().flatten() { let mut joined = Vec::with_capacity(left_row.len() + right_row.len()); joined.extend(left_row.iter().cloned()); joined.extend(right_row.iter().cloned()); From 920f33700c06e09acdfcb54016925968d1b6fe1b Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 06:36:40 -0600 Subject: [PATCH 14/20] Pin reviewed relational SQL planner --- Cargo.lock | 26 +++++++++++++------------- control_plane/Cargo.toml | 10 +++++----- control_plane/src/clickhouse.rs | 1 + crates/asap_types/Cargo.toml | 2 +- data_plane/Cargo.toml | 6 +++--- 5 files changed, 23 insertions(+), 22 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index bb9118d8b..c0b64e396 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -374,7 +374,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c3658#16c365852efce5cab50bb500c9024582a82edd19" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c365852efce5cab50bb500c9024582a82edd19#16c365852efce5cab50bb500c9024582a82edd19" dependencies = [ "asap-types", "serde", @@ -385,7 +385,7 @@ dependencies = [ [[package]] name = "asap-frontend-metricsql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c3658#16c365852efce5cab50bb500c9024582a82edd19" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c365852efce5cab50bb500c9024582a82edd19#16c365852efce5cab50bb500c9024582a82edd19" dependencies = [ "asap-types", "metricsql_parser", @@ -395,7 +395,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c3658#16c365852efce5cab50bb500c9024582a82edd19" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c365852efce5cab50bb500c9024582a82edd19#16c365852efce5cab50bb500c9024582a82edd19" dependencies = [ "asap-types", "promql-parser 0.10.0", @@ -404,7 +404,7 @@ dependencies = [ [[package]] name = "asap-frontend-sql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c3658#16c365852efce5cab50bb500c9024582a82edd19" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c365852efce5cab50bb500c9024582a82edd19#16c365852efce5cab50bb500c9024582a82edd19" dependencies = [ "asap-sql-function-catalog", "asap-types", @@ -426,12 +426,12 @@ dependencies = [ [[package]] name = "asap-sql-function-catalog" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c3658#16c365852efce5cab50bb500c9024582a82edd19" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c365852efce5cab50bb500c9024582a82edd19#16c365852efce5cab50bb500c9024582a82edd19" [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c3658#16c365852efce5cab50bb500c9024582a82edd19" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c365852efce5cab50bb500c9024582a82edd19#16c365852efce5cab50bb500c9024582a82edd19" dependencies = [ "serde", "serde_json", @@ -1711,7 +1711,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2453,7 +2453,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi 0.5.2", "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2801,7 +2801,7 @@ dependencies = [ [[package]] name = "metricsql_common" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c3658#16c365852efce5cab50bb500c9024582a82edd19" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c365852efce5cab50bb500c9024582a82edd19#16c365852efce5cab50bb500c9024582a82edd19" dependencies = [ "chrono", ] @@ -2809,7 +2809,7 @@ dependencies = [ [[package]] name = "metricsql_parser" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c3658#16c365852efce5cab50bb500c9024582a82edd19" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=16c365852efce5cab50bb500c9024582a82edd19#16c365852efce5cab50bb500c9024582a82edd19" dependencies = [ "ahash", "chrono", @@ -3904,7 +3904,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4438,7 +4438,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -5260,7 +5260,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.48.0", ] [[package]] diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 0959c32ce..3d40432d6 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -91,8 +91,8 @@ asap_types.workspace = true # scaffolding, unaware that `data_plane`'s `summary_executor.rs` in *this* # repo is a real one. Vendored locally instead of chased upstream -- see # `data_plane/src/query_engines/asap_query_engine/summary_exec.rs`. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } -asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "16c365852efce5cab50bb500c9024582a82edd19" } +asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c365852efce5cab50bb500c9024582a82edd19" } # L1 adoption (design-target-architecture.md Part B): the PromQL front # end itself, replacing control_plane's own query_parser/promql.rs. @@ -100,9 +100,9 @@ asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = # `planner-types`/`asap-aware-mapping` above -- these three MUST move # together (two revs of the same upstream repo's types in one workspace # resolve to distinct Rust types that won't unify). -asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } -asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } -asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } +asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c365852efce5cab50bb500c9024582a82edd19" } +asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c365852efce5cab50bb500c9024582a82edd19" } +asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c365852efce5cab50bb500c9024582a82edd19" } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index f9f1997cc..e9b699d70 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -597,6 +597,7 @@ mod tests { accuracy: AccuracyTarget::Exact, queries: vec![ClickHouseSqlWorkloadEntry { sql: sql.into(), + planning_sql: None, start_ms: 0, end_ms: 300_000, cumulative: true, diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml index e41cf8026..d4197f54b 100644 --- a/crates/asap_types/Cargo.toml +++ b/crates/asap_types/Cargo.toml @@ -32,4 +32,4 @@ sha2 = "0.10" # exactly (`control_plane/Cargo.toml`) -- two different revs of the same # git dependency in one workspace resolve to two distinct Rust types that # won't unify. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "16c365852efce5cab50bb500c9024582a82edd19" } diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index c0204cd7b..af96a3b98 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -38,8 +38,8 @@ control_plane = { path = "../control_plane" } # reduction: Reduction, .. }`) are `pre_asap` types, in the same crate now # (not a separate `asap-ir` import). Query serving consumes the compiled # QueryPlan; these types are used at physical-plan compilation boundaries. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } -asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "16c365852efce5cab50bb500c9024582a82edd19" } +asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c365852efce5cab50bb500c9024582a82edd19" } # Shared external (workspace) serde.workspace = true @@ -130,7 +130,7 @@ crc32fast = "1.4" # none of them. [dev-dependencies] -asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c3658" } +asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "16c365852efce5cab50bb500c9024582a82edd19" } tempfile = "3.20.0" criterion = { version = "0.5", features = ["html_reports"] } tokio-tungstenite = "0.21" From c2f20d41cb478e374e91c93676c80faf57a4d2c3 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 06:39:13 -0600 Subject: [PATCH 15/20] Inspect relational joins in planner tools --- control_plane/examples/calibration_candidates.rs | 10 ++++++++++ control_plane/examples/offline_planner_replay.rs | 5 +++++ 2 files changed, 15 insertions(+) diff --git a/control_plane/examples/calibration_candidates.rs b/control_plane/examples/calibration_candidates.rs index 64ce8d3c6..607bc1501 100644 --- a/control_plane/examples/calibration_candidates.rs +++ b/control_plane/examples/calibration_candidates.rs @@ -74,6 +74,16 @@ fn planner_forest(queries: &[control_plane::physical::compiler::PlanningQuery]) vec![outer, inner], json!({"key_debug":format!("{key:?}"),"family_debug":format!("{family:?}")}), ), + SummaryExpr::RelationalJoin { + left, + right, + kind, + pred, + } => ( + "RelationalJoin", + vec![left, right], + json!({"kind_debug":format!("{kind:?}"),"predicate_debug":format!("{pred:?}")}), + ), SummaryExpr::SummarySubtract { left, right } => { ("SummarySubtract", vec![left, right], json!({})) } diff --git a/control_plane/examples/offline_planner_replay.rs b/control_plane/examples/offline_planner_replay.rs index 0ccf9d051..88d5baf21 100644 --- a/control_plane/examples/offline_planner_replay.rs +++ b/control_plane/examples/offline_planner_replay.rs @@ -69,6 +69,11 @@ fn inspect( inner: rhs, .. } + | SummaryExpr::RelationalJoin { + left: lhs, + right: rhs, + .. + } | SummaryExpr::SummarySubtract { left: lhs, right: rhs, From b6181d421dd1fa4d2f7d6aed51fc1ff04bb16735 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 06:40:34 -0600 Subject: [PATCH 16/20] docs(clickhouse): define asap_last compatibility boundary --- .../query-engine/clickhouse-sql-support.md | 50 +++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/docs/developer_docs/query-engine/clickhouse-sql-support.md b/docs/developer_docs/query-engine/clickhouse-sql-support.md index 0f0185958..f47d61e83 100644 --- a/docs/developer_docs/query-engine/clickhouse-sql-support.md +++ b/docs/developer_docs/query-engine/clickhouse-sql-support.md @@ -89,6 +89,56 @@ Catalog misses, SQL canonicalization failures, unsupported formats, incomplete pane coverage, and execution errors fail closed to exact ClickHouse. The proxy preserves the upstream status, safe headers, and response body. +### `asap_last` boundary + +`asap_last(value, timestamp, window_ms)` is intentionally unavailable on the +accelerated path. The SQL frontend must reject it during planning and the HTTP +boundary must send the original, equivalent ClickHouse SQL to exact fallback. +It must not map `asap_last` to `argMax`, `MinMax`, `Sum`, or `Increase`. + +The current shared contracts cannot represent its state without acquiring new +meaning: + +| Existing contract | Why it cannot represent `asap_last` | +| --- | --- | +| `ExactAggregate(MinMax)` / `MinMaxAccumulator` | Chooses the minimum or maximum **value**, rather than the value from the greatest event timestamp. | +| `ExactAggregate(Increase)` / `IncreaseAccumulator` | Owns counter reset, extrapolation, and rate/increase semantics. Although its implementation retains a last sample, publishing a gauge-last descriptor as `Increase` would make capability matching and persisted payload identity false. | +| `ExactAggregate(Sum)` | Loses both the last value and its event timestamp. | +| Opaque sketch payload | Still requires a shared `SketchAlgorithm` and corresponding merge/readout semantics; it is not an untyped extension slot. | +| Raw sample storage | Can answer an exact scan but is not a mergeable materialized summary and cannot back a `SummaryAgg` DAG node. | + +The `AggregationType`, `Statistic`, accumulator factory, SDS capability, +materialization metadata, persistence decoder, pane merger, and DAG readout all +dispatch through closed shared vocabularies. A ClickHouse-only plan sidecar can +name a private operation, but it cannot cause SummaryStore to build, decode, +merge, or query an otherwise unknown payload. Encoding private last state under +an existing type would silently change that type's identity and is forbidden. + +Full acceleration therefore requires an explicitly approved contract +extension. The smallest coherent design is: + +```text +SqlLast plan leaf (ClickHouse-owned syntax and result binding) + -> exact timestamped-value materialization descriptor + -> per-series LastState { event_timestamp, value } + -> pane/shard merge by greatest event_timestamp + -> finalize value, retaining the declared series-key columns +``` + +The table catalog must prove `(timestamp, complete series identity)` is unique, +as it does for SQL rate/increase planning. This removes equal-timestamp +ambiguity within one series. Missing identity proof, duplicate timestamps, +non-numeric values, partial window coverage, mismatched payload identity, or an +unknown decoder must fail closed. `argMax(arg, val)` remains the existing SQL +extension with its original row-selection semantics and is not rewritten to +this operation. + +Adding only `SqlLastPlanEntry` is insufficient. An implementation would also +need a distinct persisted payload identity, accumulator factory and serializer, +single- and multi-series materialization, pane merge, coverage-aware readout, +and capability validation. Those are changes to the shared SDS/ingest contract, +so they are outside the compatibility constraint for this work. + ClickHouse can also provide samples to the queued backfill service through the explicit `clickhouse://configured` source marker. Backfill populates the same SummaryStore instances used by other ingest sources; it does not introduce a From 3934fb7fd8f33592e097181a16ef7576098ee04a Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 06:39:45 -0600 Subject: [PATCH 17/20] Test compiled SQL ratio against SummaryStore --- .../tests/clickhouse_ratio_store_e2e.rs | 159 ++++++++++++++++++ 1 file changed, 159 insertions(+) create mode 100644 data_plane/tests/clickhouse_ratio_store_e2e.rs diff --git a/data_plane/tests/clickhouse_ratio_store_e2e.rs b/data_plane/tests/clickhouse_ratio_store_e2e.rs new file mode 100644 index 000000000..740c72972 --- /dev/null +++ b/data_plane/tests/clickhouse_ratio_store_e2e.rs @@ -0,0 +1,159 @@ +use std::collections::{BTreeMap, BTreeSet, HashMap}; + +use arrow::array::{Float64Array, StringArray}; +use asap_types::{AggregationType, KeyByLabelNames, PrecomputeMaterialization, WindowKind}; +use control_plane::{ + clickhouse::{compile_clickhouse_workload, ClickHouseSqlWorkload, ClickHouseSqlWorkloadEntry}, + physical::{ + compiler::{PlanEnvelope, PrecomputePlan, TransmissionPlan}, + summary_catalog::SummaryCatalog, + }, + query_plan::ExecutableQueryPlan, +}; +use data_plane::{ + precompute_engine::operators::IncreaseAccumulator, + query_engines::asap_clickhouse_query_engine::execution::{ + execute_sql_dag, ClickHouseDagOutcome, + }, + storage_engines::{ + sketch_db::index::{AggKind, Capability, SketchInstanceMetadata, SketchStore}, + types::Measurement, + }, +}; +use planner_types::pre_asap::{Column, DataType, Schema}; + +#[tokio::test] +async fn compiled_ratio_bundle_executes_two_real_store_summaries() { + let configs = ["errors_total", "requests_total"].map(|metric| { + let mut config = PrecomputeMaterialization::new( + AggregationType::Increase, + String::new(), + Default::default(), + KeyByLabelNames::new(vec!["labels".into()]), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 300, + 300, + WindowKind::Tumbling, + String::new(), + metric.into(), + None, + Some("raw_samples".into()), + Some("value".into()), + ); + config.pane_origin_ms = Some(0); + config + }); + let sds = SummaryCatalog::from_materializations(76, 1, &configs).unwrap(); + let envelope = PlanEnvelope { + plan_id: 76, + plan_version: 1, + generated_at_unix_ms: 0, + activation_unix_ms: 0, + expiry_unix_ms: None, + backend_compat: "test".into(), + planner_revision: "test".into(), + capability_snapshot_id: "test".into(), + }; + let mut precompute = + PrecomputePlan::build_backend_local(envelope.clone(), configs.to_vec()).unwrap(); + precompute.summary_catalog = Some(sds.reference().unwrap()); + let mut transmission = + TransmissionPlan::build(envelope, &precompute, &Default::default()).unwrap(); + transmission.summary_catalog = precompute.summary_catalog.clone(); + let sql = "SELECT a.labels, a.v / b.v AS ratio FROM \ + (SELECT labels, asap_rate(value, ts_ms, 300000) AS v FROM raw_samples WHERE metric='errors_total' GROUP BY labels) a \ + INNER JOIN \ + (SELECT labels, asap_rate(value, ts_ms, 300000) AS v FROM raw_samples WHERE metric='requests_total' GROUP BY labels) b \ + ON a.labels=b.labels"; + let compiled = compile_clickhouse_workload(&ClickHouseSqlWorkload { + sds: sds.clone(), + precompute_plan: precompute, + transmission_plan: transmission, + tables: HashMap::from([( + "raw_samples".into(), + Schema::with_time_index( + vec![ + Column::new("metric", DataType::Utf8, false), + Column::new("labels", DataType::Utf8, false), + Column::new("ts_ms", DataType::Timestamp, false), + Column::new("value", DataType::Float64, false), + ], + 2, + vec![vec![2, 1]], + ), + )]), + accuracy: planner_types::types::AccuracyTarget::Exact, + queries: vec![ClickHouseSqlWorkloadEntry { + sql: sql.into(), + start_ms: 0, + end_ms: 300_000, + cumulative: true, + }], + }) + .await + .unwrap(); + let plan: ExecutableQueryPlan = + serde_json::from_value(compiled.plans[0]["runtime"]["executable"].clone()).unwrap(); + + let store = SketchStore::new(); + store + .install_summary_catalog(std::sync::Arc::new(sds.clone())) + .unwrap(); + let group = BTreeMap::from([("labels".into(), "api".into())]); + for (sid, config, end_value) in [(11, &configs[0], 60.0), (12, &configs[1], 300.0)] { + store.register(SketchInstanceMetadata { + sid, + metric_name: config.metric.clone(), + group_by_keys: BTreeSet::from(["labels".into()]), + capability: Some(Capability::ExactAgg(AggregationType::Increase)), + agg_kind: AggKind::ExactAgg { + agg_type: AggregationType::Increase, + parameters_canonical: String::new(), + spatial_filter_canonical: String::new(), + }, + accuracy: None, + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: config.policy_fingerprint(), + }); + store.append_precompute( + sid, + group.clone(), + (0, 300_000), + Box::new(IncreaseAccumulator::new( + Measurement::new(0.0), + 0, + Measurement::new(end_value), + 300_000, + )), + ); + } + + let ClickHouseDagOutcome::Accelerated(result) = + execute_sql_dag(&store, &plan, &sds, 0, 300_000, true) + else { + panic!("compiled ratio plan did not execute warm") + }; + let batch = &result.batches[0]; + assert_eq!( + batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + "api" + ); + assert_eq!( + batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 0.2 + ); +} From 549cf648d16f2123b2770891360d0afd898f3b08 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 06:43:35 -0600 Subject: [PATCH 18/20] Fix ratio store planning contract --- data_plane/tests/clickhouse_ratio_store_e2e.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/data_plane/tests/clickhouse_ratio_store_e2e.rs b/data_plane/tests/clickhouse_ratio_store_e2e.rs index 740c72972..9c5d86223 100644 --- a/data_plane/tests/clickhouse_ratio_store_e2e.rs +++ b/data_plane/tests/clickhouse_ratio_store_e2e.rs @@ -87,6 +87,7 @@ async fn compiled_ratio_bundle_executes_two_real_store_summaries() { accuracy: planner_types::types::AccuracyTarget::Exact, queries: vec![ClickHouseSqlWorkloadEntry { sql: sql.into(), + planning_sql: None, start_ms: 0, end_ms: 300_000, cumulative: true, From 007b4bc2925bc8f655f4f4408c403470a68fa8a3 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 06:53:12 -0600 Subject: [PATCH 19/20] Validate coverage for each SQL join input --- .../asap_clickhouse_query_engine/execution.rs | 29 ++++++- .../tests/clickhouse_ratio_store_e2e.rs | 87 ++++++++++++++++--- 2 files changed, 103 insertions(+), 13 deletions(-) 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 bb40d6889..26088e093 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 @@ -113,7 +113,28 @@ fn execute_relation_subtree( let executable = reachable_entry(entry, root)?; let outcome = execute_query_plan_payload_readout(index, &executable, t0_ms, t1_ms, is_cumulative) - .map_err(|error| format!("{error:?}"))?; + .map_err(|error| format!("incomplete leaf coverage: {error:?}"))?; + for binding in executable.materialization_bindings() { + let origin = binding + .pane_origin_ms + .ok_or_else(|| "incomplete leaf coverage: missing pane origin".to_owned())?; + let start = i64::try_from(t0_ms) + .map_err(|_| "incomplete leaf coverage: start exceeds i64".to_owned())?; + let end = i64::try_from(t1_ms) + .map_err(|_| "incomplete leaf coverage: end exceeds i64".to_owned())?; + let pane = i64::try_from(binding.window_ms) + .map_err(|_| "incomplete leaf coverage: pane exceeds i64".to_owned())?; + if pane <= 0 + || (start - origin).rem_euclid(pane) != 0 + || (end - origin).rem_euclid(pane) != 0 + || !complete_pane_coverage(outcome.coverage, (t0_ms, t1_ms), binding.window_ms) + { + return Err(format!( + "incomplete leaf coverage: requested ({t0_ms}, {t1_ms}), observed {:?}, pane {} origin {}", + outcome.coverage, binding.window_ms, origin + )); + } + } ClickHouseRelation::from_series_rows(expected_schema, outcome.series, outcome.coverage) .map_err(|error| error.to_string()) } @@ -179,6 +200,12 @@ pub fn execute_sql_dag( is_cumulative, ) { Ok(relation) => relation, + Err(error) if error.contains("incomplete leaf coverage") => { + return ClickHouseDagOutcome::Fallback(ClickHouseDagFallback::IncompleteCoverage { + requested: (t0_ms, t1_ms), + observed: None, + }) + } Err(error) => { return ClickHouseDagOutcome::Fallback(ClickHouseDagFallback::UnsupportedPlan( error, diff --git a/data_plane/tests/clickhouse_ratio_store_e2e.rs b/data_plane/tests/clickhouse_ratio_store_e2e.rs index 9c5d86223..96af0e89e 100644 --- a/data_plane/tests/clickhouse_ratio_store_e2e.rs +++ b/data_plane/tests/clickhouse_ratio_store_e2e.rs @@ -24,7 +24,7 @@ use planner_types::pre_asap::{Column, DataType, Schema}; #[tokio::test] async fn compiled_ratio_bundle_executes_two_real_store_summaries() { - let configs = ["errors_total", "requests_total"].map(|metric| { + let mut configs = ["errors_total", "requests_total"].map(|metric| { let mut config = PrecomputeMaterialization::new( AggregationType::Increase, String::new(), @@ -45,6 +45,8 @@ async fn compiled_ratio_bundle_executes_two_real_store_summaries() { config.pane_origin_ms = Some(0); config }); + configs[1].slide_interval = 150; + configs[1].window_type = WindowKind::Sliding; let sds = SummaryCatalog::from_materializations(76, 1, &configs).unwrap(); let envelope = PlanEnvelope { plan_id: 76, @@ -120,17 +122,27 @@ async fn compiled_ratio_bundle_executes_two_real_store_summaries() { expires_at_ms: None, policy_fp: config.policy_fingerprint(), }); - store.append_precompute( - sid, - group.clone(), - (0, 300_000), - Box::new(IncreaseAccumulator::new( - Measurement::new(0.0), - 0, - Measurement::new(end_value), - 300_000, - )), - ); + let panes = if sid == 11 { + vec![(0, 300_000, 0.0, end_value)] + } else { + vec![ + (0, 150_000, 0.0, 150.0), + (150_000, 300_000, 150.0, end_value), + ] + }; + for (start, end, first, last) in panes { + store.append_precompute( + sid, + group.clone(), + (start, end), + Box::new(IncreaseAccumulator::new( + Measurement::new(first), + start as i64, + Measurement::new(last), + end as i64, + )), + ); + } } let ClickHouseDagOutcome::Accelerated(result) = @@ -157,4 +169,55 @@ async fn compiled_ratio_bundle_executes_two_real_store_summaries() { .value(0), 0.2 ); + + let incomplete = SketchStore::new(); + incomplete + .install_summary_catalog(std::sync::Arc::new(sds.clone())) + .unwrap(); + for (sid, config) in [(21, &configs[0]), (22, &configs[1])] { + incomplete.register(SketchInstanceMetadata { + sid, + metric_name: config.metric.clone(), + group_by_keys: BTreeSet::from(["labels".into()]), + capability: Some(Capability::ExactAgg(AggregationType::Increase)), + agg_kind: AggKind::ExactAgg { + agg_type: AggregationType::Increase, + parameters_canonical: String::new(), + spatial_filter_canonical: String::new(), + }, + accuracy: None, + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: config.policy_fingerprint(), + }); + } + incomplete.append_precompute( + 21, + group.clone(), + (0, 300_000), + Box::new(IncreaseAccumulator::new( + Measurement::new(0.0), + 0, + Measurement::new(60.0), + 300_000, + )), + ); + incomplete.append_precompute( + 22, + group, + (150_000, 300_000), + Box::new(IncreaseAccumulator::new( + Measurement::new(150.0), + 150_000, + Measurement::new(300.0), + 300_000, + )), + ); + assert!(matches!( + execute_sql_dag(&incomplete, &plan, &sds, 0, 300_000, true), + ClickHouseDagOutcome::Fallback( + data_plane::query_engines::asap_clickhouse_query_engine::execution::ClickHouseDagFallback::IncompleteCoverage { .. } + ) + )); } From 74941fdc3bb983a8d0937da91f7bb0ae358deb81 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 06:55:44 -0600 Subject: [PATCH 20/20] Isolate ClickHouse process E2E state --- .../tests/clickhouse_q05_process_e2e.rs | 23 +++++++++++++++---- 1 file changed, 19 insertions(+), 4 deletions(-) diff --git a/data_plane/tests/clickhouse_q05_process_e2e.rs b/data_plane/tests/clickhouse_q05_process_e2e.rs index cb9b5ddf4..8a2e3e8d5 100644 --- a/data_plane/tests/clickhouse_q05_process_e2e.rs +++ b/data_plane/tests/clickhouse_q05_process_e2e.rs @@ -140,6 +140,8 @@ async fn q05_sql_is_planned_backfilled_and_served_warm_by_backend_process() { "{}/examples/promql/streaming_config.yaml", env!("CARGO_MANIFEST_DIR") ); + let output_dir = tempfile::tempdir().unwrap(); + let output_dir_arg = output_dir.path().to_str().unwrap().to_owned(); let mut command = std::process::Command::new(env!("CARGO_BIN_EXE_data_plane")); command.args([ "--http-port", @@ -160,7 +162,7 @@ async fn q05_sql_is_planned_backfilled_and_served_warm_by_backend_process() { "ts_ms", "--enable-backfill-worker", "--output-dir", - "/tmp/asap-q05-process-e2e", + &output_dir_arg, ]); if let Ok(user) = std::env::var("CLICKHOUSE_USER") { command.args(["--clickhouse-user", &user]); @@ -176,8 +178,10 @@ async fn q05_sql_is_planned_backfilled_and_served_warm_by_backend_process() { .unwrap(), ); let base = format!("http://127.0.0.1:{http_port}"); + let mut ready = false; for _ in 0..100 { if client.get(format!("{base}/health")).send().await.is_ok() { + ready = true; break; } if let Some(status) = child.0.try_wait().unwrap() { @@ -193,6 +197,7 @@ async fn q05_sql_is_planned_backfilled_and_served_warm_by_backend_process() { } tokio::time::sleep(Duration::from_millis(50)).await; } + assert!(ready, "backend did not become ready before poll deadline"); let install = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest { summary_catalog: sds, collector_plans: vec![], @@ -246,7 +251,9 @@ async fn q05_sql_is_planned_backfilled_and_served_warm_by_backend_process() { let job = response["job_id"] .as_u64() .unwrap_or_else(|| panic!("backfill was not accepted: {response}")); - for _ in 0..100 { + let mut backfill_complete = false; + let mut last_backfill_status = serde_json::Value::Null; + for _ in 0..300 { let status: serde_json::Value = client .get(format!("{base}/api/v1/db/backfill/jobs/{job}")) .send() @@ -255,12 +262,20 @@ async fn q05_sql_is_planned_backfilled_and_served_warm_by_backend_process() { .json() .await .unwrap(); - match status["status"].as_str() { - Some("complete") => break, + last_backfill_status = status.clone(); + match status["job"]["status"].as_str() { + Some("complete") => { + backfill_complete = true; + break; + } Some("failed") => panic!("backfill failed: {status}"), _ => tokio::time::sleep(Duration::from_millis(100)).await, } } + assert!( + backfill_complete, + "backfill did not complete before poll deadline: {last_backfill_status}" + ); let warm = client .post(format!("http://127.0.0.1:{sql_port}/")) .body(sql)