diff --git a/Cargo.lock b/Cargo.lock index 9acd1f3b..2bc02df9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -980,10 +980,12 @@ dependencies = [ "promql-parser 0.9.0", "prost", "prost-build", + "regex", "reqwest 0.12.28", "serde", "serde_json", "serde_yaml", + "sqlparser", "thiserror 1.0.69", "tokio", "tokio-stream", @@ -993,6 +995,7 @@ dependencies = [ "tower 0.4.13", "tracing", "tracing-subscriber", + "xxhash-rust", "zstd", ] @@ -1198,6 +1201,7 @@ dependencies = [ "base64 0.21.7", "bincode", "chrono", + "chrono-tz", "clap 4.6.1", "control_plane", "crc32fast", diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 47d3bbf1..ceb99b89 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -24,6 +24,9 @@ thiserror = "1" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } chrono = { version = "0.4", features = ["serde"] } +regex = "1" +sqlparser = "0.51" +xxhash-rust = { version = "0.8", features = ["xxh64"] } # Private mirror of GreptimeTeam/promql-parser (Apache-2.0), matching # ASAPPlanner's frontend-promql historically -- carries local patches # (limitk / limit_ratio, a batch of experimental Prometheus functions) diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 4f547b5d..36773b72 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -11,12 +11,14 @@ use planner_types::workload::SqlDialect; use crate::physical::compiler::{PrecomputePlan, TransmissionPlan}; use crate::physical::post_asap::{cost_model::ControlPlaneCostModel, PhysicalExpr}; use crate::query_plan::{ - ClickHousePlanningContext, FallbackPolicy, FixedEvaluationRange, InstantExecution, - MaterializationBinding, PhysicalGrouping, QueryLanguage, QueryPlan, QueryPlanEntry, + BoundClickHouseQuery, ClickHousePlanningContext, FallbackPolicy, FixedEvaluationRange, + InstantExecution, MaterializationBinding, PhysicalGrouping, QueryLanguage, QueryPlan, + QueryPlanEntry, }; use asap_types::summary_catalog::SummaryCatalog; use serde::{Deserialize, Serialize}; use std::collections::HashMap; +use std::rc::Rc; #[derive(Debug, thiserror::Error)] pub enum ClickHousePlanningError { @@ -43,8 +45,20 @@ 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()); + 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), @@ -88,6 +102,24 @@ pub struct ClickHouseSqlWorkloadEntry { pub start_ms: u64, pub end_ms: u64, pub cumulative: bool, + /// Exact subtree artifacts emitted alongside ASAPPlanner's selected DAG. + #[serde(default)] + pub exact_subtrees: Vec, +} + +#[derive(Debug, Clone, Deserialize)] +pub struct ClickHouseExactSubtreeArtifact { + pub canonical_subtree: serde_json::Value, + pub canonical_subtree_fingerprint: u64, + pub query: BoundClickHouseQuery, +} + +pub fn canonical_subtree_fingerprint( + canonical_subtree: &serde_json::Value, +) -> Result { + let encoded = serde_json::to_vec(canonical_subtree) + .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?; + Ok(xxhash_rust::xxh64::xxh64(&encoded, 0)) } /// Physical-plan components produced for the normal atomic install path. @@ -125,7 +157,22 @@ pub async fn compile_clickhouse_workload( "SQL did not produce a summary DAG".into(), )); }; - let executable = QueryPlanEntry::compile_bound_relational( + let mut fingerprints = std::collections::BTreeSet::new(); + for artifact in &query.exact_subtrees { + let actual = canonical_subtree_fingerprint(&artifact.canonical_subtree)?; + if actual != artifact.canonical_subtree_fingerprint { + return Err(ClickHousePlanningError::Lower( + "exact SQL artifact fingerprint differs from its canonical subtree".into(), + )); + } + if !fingerprints.insert(actual) { + return Err(ClickHousePlanningError::Lower( + "duplicate exact SQL subtree artifact".into(), + )); + } + } + let mut used = std::collections::BTreeSet::new(); + let executable = QueryPlanEntry::compile_bound_relational_with_external( query.sql.clone(), planned.canonical_sql.clone(), &root, @@ -141,8 +188,41 @@ pub async fn compile_clickhouse_workload( }, FallbackPolicy::ExactBackend, |node, family| bind_selected_node(node, family, query, request), + |cut, schema| { + let canonical_subtree = serde_json::to_value(cut).map_err(|error| { + crate::query_plan::QueryPlanError::Invalid(error.to_string()) + })?; + let fingerprint = + canonical_subtree_fingerprint(&canonical_subtree).map_err(|error| { + crate::query_plan::QueryPlanError::Invalid(error.to_string()) + })?; + let Some((index, artifact)) = + query + .exact_subtrees + .iter() + .enumerate() + .find(|(_, artifact)| { + artifact.canonical_subtree_fingerprint == fingerprint + && artifact.canonical_subtree == canonical_subtree + }) + else { + return Ok(None); + }; + if &artifact.query.output_schema != schema { + return Err(crate::query_plan::QueryPlanError::Invalid( + "exact SQL artifact schema differs from the Planner cut".into(), + )); + } + used.insert(index); + Ok(Some(artifact.query.clone())) + }, ) .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?; + if used.len() != query.exact_subtrees.len() { + return Err(ClickHousePlanningError::Lower( + "one or more exact SQL subtree artifacts were not used by the Planner DAG".into(), + )); + } if executable .nodes .values() @@ -377,4 +457,121 @@ mod tests { .contains("ambiguous") ); } + + #[tokio::test] + async fn production_compile_publishes_only_the_planner_matched_exact_cut() { + use crate::physical::compiler::{PlanEnvelope, BACKEND_COMPAT, PLANNER_REVISION}; + use planner_types::{ + post_asap::{SummaryExpr, SummaryNode}, + pre_asap::{Column, DataType, Schema}, + }; + use std::{collections::BTreeMap, rc::Rc}; + + fn exact_cut( + node: &Rc, + ) -> Option<(&QueryExpr, &planner_types::post_asap::SummarySchema)> { + match &node.expr { + SummaryExpr::KeepPreAsap(expr) => Some((expr, &node.schema)), + SummaryExpr::ValueOperation { child, .. } => exact_cut(child), + _ => None, + } + } + + let table = Schema::with_time_index( + vec![ + Column::new("timestamp", DataType::Timestamp, false), + Column::new("value", DataType::Float64, false), + ], + 0, + vec![], + ); + let tables = HashMap::from([("requests".into(), table)]); + let planned = plan_clickhouse_sql( + "SELECT timestamp, value FROM requests", + &SqlCatalog { + tables: tables.clone(), + }, + AccuracyTarget::Exact, + ) + .await + .unwrap(); + let PhysicalExpr::Committed(crate::physical::post_asap::PostAsapPlan::Summary(root)) = + planned.physical + else { + panic!("summary plan") + }; + let (cut, schema) = exact_cut(&root).expect("Planner exact cut"); + let canonical_subtree = serde_json::to_value(cut).unwrap(); + let fingerprint = canonical_subtree_fingerprint(&canonical_subtree).unwrap(); + + let sds = SummaryCatalog::from_materializations(91, 1, &[]).unwrap(); + let envelope = PlanEnvelope { + plan_id: 91, + plan_version: 1, + generated_at_unix_ms: 0, + activation_unix_ms: 0, + expiry_unix_ms: None, + backend_compat: BACKEND_COMPAT.into(), + planner_revision: PLANNER_REVISION.into(), + capability_snapshot_id: "sql-exact-cut".into(), + }; + let mut precompute_plan = PrecomputePlan::build(envelope.clone(), vec![], &[]).unwrap(); + precompute_plan.summary_catalog = Some(sds.reference().unwrap()); + let mut transmission_plan = + TransmissionPlan::build(envelope, &precompute_plan, &BTreeMap::new()).unwrap(); + transmission_plan.summary_catalog = Some(sds.reference().unwrap()); + let workload = ClickHouseSqlWorkload { + sds: sds.clone(), + precompute_plan, + transmission_plan, + tables, + accuracy: AccuracyTarget::Exact, + queries: vec![ClickHouseSqlWorkloadEntry { + sql: "SELECT timestamp, value FROM requests".into(), + start_ms: 0, + end_ms: 2_000, + cumulative: false, + exact_subtrees: vec![ClickHouseExactSubtreeArtifact { + canonical_subtree, + canonical_subtree_fingerprint: fingerprint, + query: BoundClickHouseQuery { + sql: "SELECT timestamp, value FROM requests WHERE timestamp >= {from:Int64} AND timestamp < {to:Int64}".into(), + parameters: BTreeMap::new(), + start_parameter: Some("from".into()), + end_parameter: Some("to".into()), + output_schema: schema.clone(), + }, + }], + }], + }; + let bundle = compile_clickhouse_workload(&workload).await.unwrap(); + bundle.query_plan.validate_against_catalog(&sds).unwrap(); + assert!(bundle + .query_plan + .entries + .values() + .next() + .unwrap() + .nodes + .values() + .any(|node| matches!( + node, + crate::query_plan::QueryPlanNode::ExternalSqlLeaf { .. } + ))); + + let mut unused = workload; + let duplicate = unused.queries[0].exact_subtrees[0].clone(); + unused.queries[0].exact_subtrees.push(duplicate); + assert!(compile_clickhouse_workload(&unused).await.is_err()); + unused.queries[0].exact_subtrees.pop(); + unused.queries[0].exact_subtrees[0].canonical_subtree = + serde_json::to_value(QueryExpr::::Literal( + planner_types::pre_asap::ScalarValue::Float64(1.0), + )) + .unwrap(); + unused.queries[0].exact_subtrees[0].canonical_subtree_fingerprint = + canonical_subtree_fingerprint(&unused.queries[0].exact_subtrees[0].canonical_subtree) + .unwrap(); + assert!(compile_clickhouse_workload(&unused).await.is_err()); + } } diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index 28dedf01..fdfcc67e 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -11,7 +11,7 @@ use std::collections::{BTreeMap, BTreeSet}; use std::rc::Rc; use planner_types::post_asap::{SketchQuery, SummaryExpr, SummaryFamilyType, SummaryNode}; -use planner_types::pre_asap::Reduction; +use planner_types::pre_asap::{QueryExpr, Reduction}; use serde::{Deserialize, Serialize}; use thiserror::Error; @@ -153,6 +153,13 @@ impl QueryPlan { )); } } + if entry + .nodes + .values() + .any(|node| matches!(node, QueryPlanNode::RelationalJoin { .. })) + { + entry.catalog_relation_schema(entry.root, catalog)?; + } } Ok(()) } @@ -204,6 +211,145 @@ impl QueryPlan { } } +impl QueryPlanEntry { + /// Derive the relation carried by a mixed SQL subtree from its catalog + /// binding and executable readout. A declared parent schema must never be + /// used to reinterpret summary rows. + fn catalog_relation_schema( + &self, + id: QueryNodeId, + catalog: &crate::physical::summary_catalog::SummaryCatalog, + ) -> Result { + use planner_types::{ + post_asap::{SummaryField, SummarySchema}, + pre_asap::DataType, + }; + + let node = self.nodes.get(&id).ok_or_else(|| { + QueryPlanError::Invalid(format!( + "mixed SQL subtree references missing node {}", + id.0 + )) + })?; + match node { + QueryPlanNode::ExternalSqlLeaf { query } => Ok(query.output_schema.clone()), + QueryPlanNode::ExactReadout { input, .. } => { + let Some(QueryPlanNode::ReadMaterialization { binding }) = self.nodes.get(input) + else { + return Err(QueryPlanError::Invalid( + "mixed SQL exact readout must directly consume a catalog materialization" + .into(), + )); + }; + let identity = catalog + .materializations + .get(&binding.materialization) + .ok_or_else(|| { + QueryPlanError::Invalid( + "mixed SQL readout references absent catalog materialization".into(), + ) + })?; + let data = catalog + .data_descriptors + .get(&identity.data_descriptor_id) + .ok_or_else(|| { + QueryPlanError::Invalid( + "mixed SQL readout references absent data descriptor".into(), + ) + })?; + let labels = match &binding.output_grouping { + PhysicalGrouping::PerEntity => { + data.group_by_keys.iter().cloned().collect::>() + } + PhysicalGrouping::Reduce(labels) => { + if labels + .iter() + .any(|label| !data.group_by_keys.contains(label)) + { + return Err(QueryPlanError::Invalid( + "mixed SQL readout grouping is not provided by its data descriptor" + .into(), + )); + } + labels.clone() + } + }; + let mut fields = labels + .into_iter() + .map(|name| SummaryField { + name, + dtype: SummaryFamilyType::Plain(DataType::Utf8), + nullable: false, + }) + .collect::>(); + let time_index = fields.len(); + fields.push(SummaryField { + name: "timestamp".into(), + dtype: SummaryFamilyType::Plain(DataType::Timestamp), + nullable: false, + }); + fields.push(SummaryField { + name: "value".into(), + dtype: SummaryFamilyType::Plain(DataType::Float64), + nullable: false, + }); + Ok(SummarySchema { + fields, + time_index: Some(time_index), + }) + } + QueryPlanNode::Relational { + input, + input_schema, + output_schema, + .. + } => { + let actual = self.catalog_relation_schema(*input, catalog)?; + if actual != *input_schema { + return Err(QueryPlanError::Invalid( + "mixed SQL relational input schema differs from its executable child" + .into(), + )); + } + Ok(output_schema.clone()) + } + QueryPlanNode::RelationalJoin { + inputs, + left_schema, + right_schema, + output_schema, + .. + } => { + let left = self.catalog_relation_schema(inputs[0], catalog)?; + let right = self.catalog_relation_schema(inputs[1], catalog)?; + if left != *left_schema || right != *right_schema { + return Err(QueryPlanError::Invalid( + "mixed SQL join child schema differs from its executable subtree".into(), + )); + } + let expected_fields = left_schema + .fields + .iter() + .chain(&right_schema.fields) + .cloned() + .collect::>(); + if output_schema.fields != expected_fields + || output_schema.time_index != left_schema.time_index + { + return Err(QueryPlanError::Invalid( + "mixed SQL join output schema is not the strict child concatenation".into(), + )); + } + Ok(output_schema.clone()) + } + _ => Err(QueryPlanError::Invalid( + "mixed SQL summary subtree schema cannot be derived from catalog and readout" + .into(), + )), + } + } +} + #[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum QueryLanguage { @@ -286,6 +432,37 @@ pub struct InstantExecution { } impl QueryPlanEntry { + /// Bind a planner-selected exact cut. Callers must supply the node ID from + /// the compiled canonical DAG; this method never infers a SQL cut. + pub fn bind_external_sql_leaf( + &mut self, + node_id: QueryNodeId, + query: BoundClickHouseQuery, + ) -> Result<(), QueryPlanError> { + if self.language != QueryLanguage::ClickHouseSql { + return Err(QueryPlanError::Invalid( + "cannot bind a SQL leaf into a non-ClickHouse plan".into(), + )); + } + query.validate()?; + match self.nodes.get(&node_id) { + Some(QueryPlanNode::ExactFallback { .. }) => {} + Some(_) => { + return Err(QueryPlanError::Invalid( + "external SQL binding must replace a planner exact-fallback cut".into(), + )) + } + None => { + return Err(QueryPlanError::Invalid( + "external SQL cut node is absent".into(), + )) + } + } + self.nodes + .insert(node_id, QueryPlanNode::ExternalSqlLeaf { query }); + Ok(()) + } + /// Materializations this executable DAG reads, in stable node order. /// Serving uses this set for readiness accounting; it never performs a /// catalog candidate search to reconstruct dependencies. @@ -331,6 +508,7 @@ impl QueryPlanEntry { bind: &mut bind, logical_source: None, preserve_relational: false, + bind_external: None, }; let root = compiler.lower(root)?; Ok(Self { @@ -367,6 +545,49 @@ impl QueryPlanEntry { bind: &mut bind, logical_source: None, preserve_relational: true, + bind_external: None, + }; + let root = compiler.lower(root)?; + Ok(Self { + language: QueryLanguage::ClickHouseSql, + query_id, + canonical_query, + fixed_evaluation: Some(fixed_evaluation), + root, + nodes: compiler.nodes, + instant, + fallback, + }) + } + + pub fn compile_bound_relational_with_external( + query_id: String, + canonical_query: String, + root: &Rc, + fixed_evaluation: FixedEvaluationRange, + instant: InstantExecution, + fallback: FallbackPolicy, + mut bind: F, + mut bind_external: E, + ) -> Result + where + F: FnMut( + &SummaryNode, + &SummaryFamilyType, + ) -> Result, + E: FnMut( + &QueryExpr, + &planner_types::post_asap::SummarySchema, + ) -> Result, QueryPlanError>, + { + let mut compiler = DagCompiler { + next_id: 0, + nodes: BTreeMap::new(), + seen: BTreeMap::new(), + bind: &mut bind, + logical_source: None, + preserve_relational: true, + bind_external: Some(&mut bind_external), }; let root = compiler.lower(root)?; Ok(Self { @@ -404,6 +625,7 @@ impl QueryPlanEntry { bind: &mut bind, logical_source: Some(canonical_query.clone()), preserve_relational: false, + bind_external: None, }; let root = compiler.lower(root)?; let mut entry = Self { @@ -429,6 +651,101 @@ impl QueryPlanEntry { ))); } for (id, node) in &self.nodes { + if let QueryPlanNode::ExternalSqlLeaf { query } = node { + if self.language != QueryLanguage::ClickHouseSql { + return Err(QueryPlanError::Invalid( + "external SQL leaf is only valid in a ClickHouse query plan".into(), + )); + } + query.validate()?; + } + if let QueryPlanNode::RelationalJoin { + inputs, + join_kind, + pred, + left_schema, + right_schema, + output_schema, + .. + } = node + { + if self.language != QueryLanguage::ClickHouseSql { + return Err(QueryPlanError::Invalid( + "relational SQL join is only valid in a ClickHouse query plan".into(), + )); + } + if !matches!(join_kind, planner_types::pre_asap::JoinKind::Inner) { + return Err(QueryPlanError::Invalid( + "only inner external SQL joins are executable".into(), + )); + } + let predicate: planner_types::pre_asap::Predicate = + serde_json::from_value(pred.clone()).map_err(|error| { + QueryPlanError::Invalid(format!("invalid join predicate: {error}")) + })?; + let planner_types::pre_asap::QueryExpr::Compare { + left, + op: planner_types::pre_asap::CompareOpKind::Eq, + right, + } = predicate.0.as_ref() + else { + return Err(QueryPlanError::Invalid( + "external SQL join requires an equality predicate".into(), + )); + }; + let ( + planner_types::pre_asap::QueryExpr::Column(left), + planner_types::pre_asap::QueryExpr::Column(right), + ) = (left.as_ref(), right.as_ref()) + else { + return Err(QueryPlanError::Invalid( + "external SQL join keys must be columns".into(), + )); + }; + if *left >= left_schema.fields.len() + || *right < left_schema.fields.len() + || *right - left_schema.fields.len() >= right_schema.fields.len() + || output_schema.fields.len() + != left_schema.fields.len() + right_schema.fields.len() + { + return Err(QueryPlanError::Invalid( + "external SQL join schema or key bounds are invalid".into(), + )); + } + let mut expected_fields = left_schema.fields.clone(); + expected_fields.extend(right_schema.fields.clone()); + let expected_time_index = left_schema.time_index.or_else(|| { + right_schema + .time_index + .map(|index| left_schema.fields.len() + index) + }); + if output_schema.fields != expected_fields + || output_schema.time_index != expected_time_index + { + return Err(QueryPlanError::Invalid( + "relational join output schema is not the strict input concatenation" + .into(), + )); + } + for (input, declared) in [(inputs[0], left_schema), (inputs[1], right_schema)] { + let actual = match self.nodes.get(&input) { + Some(QueryPlanNode::ExternalSqlLeaf { query }) => { + Some(&query.output_schema) + } + Some(QueryPlanNode::Relational { output_schema, .. }) + | Some(QueryPlanNode::RelationalJoin { output_schema, .. }) => { + Some(output_schema) + } + _ => None, + }; + if actual.is_some_and(|actual| actual != declared) { + return Err(QueryPlanError::Invalid( + "relational join child output differs from its declared input schema" + .into(), + )); + } + } + } if let QueryPlanNode::Logical { operator, inputs } = node { operator.validate(inputs.len())?; } @@ -522,6 +839,96 @@ pub struct MaterializationBinding { pub readout_lookback_ms: Option, } +/// A planner-verified ClickHouse subtree boundary. The SQL must produce the +/// declared relation when evaluated over the bound request interval. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct BoundClickHouseQuery { + pub sql: String, + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub parameters: BTreeMap, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub start_parameter: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub end_parameter: Option, + pub output_schema: planner_types::post_asap::SummarySchema, +} + +impl BoundClickHouseQuery { + fn validate(&self) -> Result<(), QueryPlanError> { + if self.sql.trim().is_empty() { + return Err(QueryPlanError::Invalid( + "ClickHouse external subtree has empty SQL".into(), + )); + } + if self.output_schema.fields.is_empty() { + return Err(QueryPlanError::Invalid( + "ClickHouse external subtree has empty output schema".into(), + )); + } + let placeholder = regex::Regex::new(r"\{([A-Za-z_][A-Za-z0-9_]*):[^{}]+\}") + .expect("static placeholder regex"); + let parseable_sql = placeholder.replace_all(&self.sql, "0"); + let statements = sqlparser::parser::Parser::parse_sql( + &sqlparser::dialect::ClickHouseDialect {}, + &parseable_sql, + ) + .map_err(|error| QueryPlanError::Invalid(format!("invalid external SQL: {error}")))?; + let [sqlparser::ast::Statement::Query(parsed_query)] = statements.as_slice() else { + return Err(QueryPlanError::Invalid( + "ClickHouse external subtree must be one read-only SELECT".into(), + )); + }; + if parsed_query.format_clause.is_some() { + return Err(QueryPlanError::Invalid( + "ClickHouse external subtree must not choose its wire format".into(), + )); + } + let declared = placeholder + .captures_iter(&self.sql) + .map(|capture| capture[1].to_owned()) + .collect::>(); + let mut supplied = self.parameters.keys().cloned().collect::>(); + supplied.extend( + [&self.start_parameter, &self.end_parameter] + .into_iter() + .flatten() + .cloned(), + ); + if declared != supplied { + return Err(QueryPlanError::Invalid( + "ClickHouse placeholder declarations and bindings differ".into(), + )); + } + for name in [&self.start_parameter, &self.end_parameter] + .into_iter() + .flatten() + { + if name.is_empty() || self.parameters.contains_key(name) || name.starts_with("param_") { + return Err(QueryPlanError::Invalid( + "ClickHouse time parameter is empty or shadows a static parameter".into(), + )); + } + } + if self + .parameters + .keys() + .any(|name| name.starts_with("param_")) + { + return Err(QueryPlanError::Invalid( + "ClickHouse bindings use logical placeholder names, without the HTTP param_ prefix" + .into(), + )); + } + if self.start_parameter.is_some() && self.start_parameter == self.end_parameter { + return Err(QueryPlanError::Invalid( + "ClickHouse start and end parameters must differ".into(), + )); + } + Ok(()) + } +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(tag = "mode", content = "keys", rename_all = "snake_case")] pub enum PhysicalGrouping { @@ -532,6 +939,19 @@ pub enum PhysicalGrouping { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(tag = "op", rename_all = "snake_case", deny_unknown_fields)] pub enum QueryPlanNode { + /// Zero-input SQL-private exact subtree. Only ClickHouse query entries may + /// carry this node; relational parents consume its typed relation. + ExternalSqlLeaf { + query: BoundClickHouseQuery, + }, + 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, /// Serialized planner-owned operation. Keeping the wire form here makes @@ -586,10 +1006,11 @@ pub enum QueryPlanNode { impl QueryPlanNode { pub fn inputs(&self) -> &[QueryNodeId] { match self { - Self::Scalar { .. } | Self::ReadMaterialization { .. } | Self::ExactFallback { .. } => { - &[] - } - Self::Binary { inputs, .. } => inputs, + Self::Scalar { .. } + | Self::ReadMaterialization { .. } + | Self::ExternalSqlLeaf { .. } + | Self::ExactFallback { .. } => &[], + Self::Binary { inputs, .. } | Self::RelationalJoin { inputs, .. } => inputs, Self::ReduceSum { input, .. } | Self::Relational { input, .. } | Self::SummaryEstimate { input, .. } @@ -657,6 +1078,12 @@ struct DagCompiler<'a, F> { bind: &'a mut F, logical_source: Option, preserve_relational: bool, + bind_external: Option< + &'a mut dyn FnMut( + &QueryExpr, + &planner_types::post_asap::SummarySchema, + ) -> Result, QueryPlanError>, + >, } impl DagCompiler<'_, F> @@ -688,7 +1115,8 @@ where } } QueryPlanNode::CandidateTopK { inputs, .. } - | QueryPlanNode::Binary { inputs, .. } => { + | QueryPlanNode::Binary { inputs, .. } + | QueryPlanNode::RelationalJoin { inputs, .. } => { for input in inputs { *input = remap[input]; } @@ -699,6 +1127,7 @@ where | QueryPlanNode::Relational { input, .. } => *input = remap[input], QueryPlanNode::Scalar { .. } | QueryPlanNode::ReadMaterialization { .. } + | QueryPlanNode::ExternalSqlLeaf { .. } | QueryPlanNode::ExactFallback { .. } => {} } self.nodes.insert(remap[&local], physical); @@ -747,6 +1176,19 @@ where } let physical = match &node.expr { + SummaryExpr::KeepPreAsap(expr) if self.preserve_relational => { + match self.bind_external.as_mut() { + Some(bind) => match bind(expr, &node.schema)? { + Some(query) => QueryPlanNode::ExternalSqlLeaf { query }, + None => QueryPlanNode::ExactFallback { + reason: "planner exact SQL cut has no bound external artifact".into(), + }, + }, + None => QueryPlanNode::ExactFallback { + reason: "planner exact SQL cut has no external binder".into(), + }, + } + } SummaryExpr::ValueOperation { child, operation, .. } if self.preserve_relational @@ -1599,6 +2041,120 @@ mod catalog_binding_tests { binding } + fn relation_schema( + names: &[(&str, planner_types::pre_asap::DataType)], + ) -> planner_types::post_asap::SummarySchema { + use planner_types::post_asap::{SummaryFamilyType, SummaryField, SummarySchema}; + SummarySchema { + fields: names + .iter() + .map(|(name, dtype)| SummaryField { + name: (*name).into(), + dtype: SummaryFamilyType::Plain(dtype.clone()), + nullable: false, + }) + .collect(), + time_index: names + .iter() + .position(|(_, dtype)| *dtype == planner_types::pre_asap::DataType::Timestamp), + } + } + + #[test] + fn mixed_join_schema_is_derived_from_catalog_readout() { + use planner_types::pre_asap::{CompareOpKind, DataType, JoinKind, Predicate, QueryExpr}; + + let (mut plan, catalog) = fixture(); + let entry = plan.entries.values_mut().next().unwrap(); + entry.language = QueryLanguage::ClickHouseSql; + entry.fixed_evaluation = Some(FixedEvaluationRange { + start_ms: 0, + end_ms: 10_000, + cumulative: false, + }); + let readout = QueryNodeId(2); + let external = QueryNodeId(3); + let join = QueryNodeId(4); + let left = relation_schema(&[ + ("job", DataType::Utf8), + ("timestamp", DataType::Timestamp), + ("value", DataType::Float64), + ]); + let right = relation_schema(&[("divisor", DataType::Float64)]); + let mut joined = left.clone(); + joined.fields.extend(right.fields.clone()); + entry.nodes.insert( + readout, + QueryPlanNode::ExactReadout { + input: QueryNodeId(1), + readout: ExactReadout::Sum, + }, + ); + entry.nodes.insert( + external, + QueryPlanNode::ExternalSqlLeaf { + query: BoundClickHouseQuery { + sql: "SELECT 1.0 AS divisor".into(), + parameters: BTreeMap::new(), + start_parameter: None, + end_parameter: None, + output_schema: right.clone(), + }, + }, + ); + entry.nodes.insert( + join, + QueryPlanNode::RelationalJoin { + inputs: [readout, external], + join_kind: JoinKind::Inner, + pred: serde_json::to_value(Predicate(Rc::new(QueryExpr::Compare { + left: Rc::new(QueryExpr::Column(2)), + op: CompareOpKind::Eq, + right: Rc::new(QueryExpr::Column(3)), + }))) + .unwrap(), + left_schema: left, + right_schema: right, + output_schema: joined, + }, + ); + entry.root = join; + let canonical = entry.canonical_query.clone(); + let entry = plan.entries.pop_first().unwrap().1; + plan.entries.insert( + QueryPlan::catalog_key(QueryLanguage::ClickHouseSql, &canonical), + entry, + ); + plan.clickhouse_context = Some(ClickHousePlanningContext { + tables: Default::default(), + accuracy: planner_types::types::AccuracyTarget::Exact, + }); + plan.validate_against_catalog(&catalog).unwrap(); + + let QueryPlanNode::RelationalJoin { + left_schema, + output_schema, + .. + } = plan + .entries + .values_mut() + .next() + .unwrap() + .nodes + .get_mut(&join) + .unwrap() + else { + unreachable!() + }; + left_schema.fields[0].name = "invented_by_parent".into(); + output_schema.fields[0].name = "invented_by_parent".into(); + assert!(plan + .validate_against_catalog(&catalog) + .unwrap_err() + .to_string() + .contains("executable subtree")); + } + // One pane ID is compatible with a longer semantic readout window. #[test] fn catalog_binding_round_trip_preserves_pane_and_readout_windows() { @@ -1691,4 +2247,90 @@ mod catalog_binding_tests { .validate_against_catalog(&counter_catalog) .unwrap(); } + + #[test] + fn planner_must_bind_external_sql_at_an_explicit_fallback_cut() { + let cut = QueryNodeId(0); + let schema = planner_types::post_asap::SummarySchema { + fields: vec![planner_types::post_asap::SummaryField { + name: "value".into(), + dtype: planner_types::post_asap::SummaryFamilyType::Plain( + planner_types::pre_asap::DataType::Float64, + ), + nullable: false, + }], + time_index: None, + }; + let mut entry = QueryPlanEntry { + language: QueryLanguage::ClickHouseSql, + query_id: "mixed".into(), + canonical_query: "mixed".into(), + fixed_evaluation: Some(FixedEvaluationRange { + start_ms: 1, + end_ms: 2, + cumulative: false, + }), + root: cut, + nodes: BTreeMap::from([( + cut, + QueryPlanNode::ExactFallback { + reason: "planner exact cut".into(), + }, + )]), + instant: InstantExecution { + lookback_ms: 1, + full_history: false, + cumulative_readout: false, + }, + fallback: FallbackPolicy::ExactBackend, + }; + entry + .bind_external_sql_leaf( + cut, + BoundClickHouseQuery { + sql: "SELECT value FROM exact_source WHERE ts >= {from:UInt64} AND ts < {to:UInt64}".into(), + parameters: BTreeMap::new(), + start_parameter: Some("from".into()), + end_parameter: Some("to".into()), + output_schema: schema.clone(), + }, + ) + .unwrap(); + entry.validate(&BTreeSet::new()).unwrap(); + assert!(matches!( + entry.nodes[&cut], + QueryPlanNode::ExternalSqlLeaf { .. } + )); + assert!(entry + .bind_external_sql_leaf( + cut, + BoundClickHouseQuery { + sql: "SELECT 2".into(), + parameters: BTreeMap::new(), + start_parameter: None, + end_parameter: None, + output_schema: planner_types::post_asap::SummarySchema { + fields: vec![], + time_index: None, + }, + } + ) + .is_err()); + let bad = BoundClickHouseQuery { + sql: "SELECT value FROM t FORMAT JSONCompact".into(), + parameters: BTreeMap::new(), + start_parameter: None, + end_parameter: None, + output_schema: schema.clone(), + }; + assert!(bad.validate().is_err()); + let missing = BoundClickHouseQuery { + sql: "SELECT value FROM t WHERE ts >= {from:Int64}".into(), + parameters: BTreeMap::new(), + start_parameter: None, + end_parameter: None, + output_schema: schema, + }; + assert!(missing.validate().is_err()); + } } diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 38619a2b..789d52c4 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -51,6 +51,7 @@ tracing.workspace = true tracing-subscriber.workspace = true clap.workspace = true chrono.workspace = true +chrono-tz = "0.10" promql-parser.workspace = true tokio.workspace = true arc-swap.workspace = true diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index d1a27271..97499a21 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -1383,18 +1383,6 @@ async fn main() -> Result<()> { let victoria_task = victoria_server.map(|server| tokio::spawn(server.run())); - let clickhouse_accelerator = if args.clickhouse_http_port.is_some() { - let accelerator = Arc::new( - data_plane::query_engines::asap_clickhouse_query_engine::accelerator::CatalogClickHouseAccelerator::with_active_physical_plan( - sketch_index.clone(), - active_physical_plan.clone(), - ), - ); - Some(accelerator) - } else { - None - }; - let clickhouse_server_handle = args.clickhouse_http_port.map(|port| { let fallback = Arc::new( data_plane::query_engines::asap_clickhouse_query_engine::ClickHouseHttpFallback::new( @@ -1402,6 +1390,13 @@ async fn main() -> Result<()> { args.clickhouse_database.clone(), ), ); + let accelerator = Arc::new( + data_plane::query_engines::asap_clickhouse_query_engine::accelerator::CatalogClickHouseAccelerator::with_active_physical_plan( + sketch_index.clone(), + active_physical_plan.clone(), + ) + .with_exact_backend(fallback.clone()), + ); let clickhouse_server = data_plane::query_engines::asap_clickhouse_query_engine::ClickHouseHttpServer { listen_address: format!("0.0.0.0:{port}"), @@ -1409,10 +1404,7 @@ async fn main() -> Result<()> { }; info!("Starting ClickHouse-compatible HTTP proxy on port {port}"); tokio::spawn(async move { - let result = match clickhouse_accelerator { - Some(accelerator) => clickhouse_server.run_with_accelerator(accelerator).await, - None => clickhouse_server.run().await, - }; + let result = clickhouse_server.run_with_accelerator(accelerator).await; if let Err(error) = result { error!("ClickHouse HTTP server error: {error}"); } 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 377e572f..dbb6287a 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 @@ -5,12 +5,15 @@ use axum::{ body::Bytes, http::{HeaderMap, HeaderValue, StatusCode}, }; -use std::sync::Arc; +use std::{collections::BTreeMap, sync::Arc}; use super::{ clickhouse_result_adapter::ClickHouseFormat, - execution::{execute_sql_dag, ClickHouseDagFallback, ClickHouseDagOutcome}, - fallback::ClickHouseRawResponse, + execution::{ + execute_sql_dag_with_external, ClickHouseDagFallback, ClickHouseDagOutcome, + PreparedSqlLeaves, + }, + fallback::{ClickHouseExactBackend, ClickHouseRawResponse}, request::ClickHouseQueryRequest, server::{ ClickHouseAccelerationFallback, ClickHouseAccelerationOutcome, ClickHouseAccelerator, @@ -21,6 +24,7 @@ use crate::storage_engines::sketch_db::index::SketchStore; pub struct CatalogClickHouseAccelerator { pub store: Arc, active_physical_plan: Option, + exact_backend: Option>, } impl CatalogClickHouseAccelerator { @@ -28,6 +32,7 @@ impl CatalogClickHouseAccelerator { Self { store, active_physical_plan: None, + exact_backend: None, } } @@ -39,6 +44,81 @@ impl CatalogClickHouseAccelerator { accelerator.active_physical_plan = Some(active); accelerator } + + pub fn with_exact_backend(mut self, exact_backend: Arc) -> Self { + self.exact_backend = Some(exact_backend); + self + } + + async fn prepare_external_sql( + &self, + entry: &control_plane::query_plan::QueryPlanEntry, + start_ms: u64, + end_ms: u64, + request_context: Option<&ClickHouseQueryRequest>, + ) -> Result { + let mut prepared = PreparedSqlLeaves::new(); + let external = entry + .nodes + .iter() + .filter_map(|(id, node)| match node { + control_plane::query_plan::QueryPlanNode::ExternalSqlLeaf { query } => { + Some((*id, query)) + } + _ => None, + }) + .collect::>(); + if external.is_empty() { + return Ok(prepared); + } + let backend = self + .exact_backend + .as_ref() + .ok_or_else(|| "ClickHouse exact subtree endpoint unavailable".to_owned())?; + for (id, bound) in external { + let mut parameters = bound + .parameters + .iter() + .map(|(name, value)| (format!("param_{name}"), value.clone())) + .collect::>(); + if let Some(name) = &bound.start_parameter { + parameters.insert(format!("param_{name}"), start_ms.to_string()); + } + if let Some(name) = &bound.end_parameter { + parameters.insert(format!("param_{name}"), end_ms.to_string()); + } + parameters.insert("default_format".into(), "JSONCompact".into()); + if let Some(database) = request_context.and_then(ClickHouseQueryRequest::database) { + parameters.insert("database".into(), database.into()); + } + let request = ClickHouseQueryRequest { + method: axum::http::Method::POST, + sql: bound.sql.clone(), + body: Bytes::from(bound.sql.clone()), + parameters, + headers: request_context + .map(|request| request.headers.clone()) + .unwrap_or_default(), + }; + let response = backend + .execute(&request) + .await + .map_err(|error| error.to_string())?; + if !response.status.is_success() { + return Err(format!( + "ClickHouse external subtree returned HTTP {}", + response.status + )); + } + let relation = super::relational_adapter::ClickHouseRelation::from_json_compact( + &bound.output_schema, + &response.body, + ) + .map_err(|error| error.to_string())?; + prepared.insert(id, relation); + } + Ok(prepared) + } } fn requested_format(request: &ClickHouseQueryRequest) -> Result { @@ -109,10 +189,22 @@ impl ClickHouseAccelerator for CatalogClickHouseAccelerator { ), ); }; - match execute_sql_dag( + let prepared = match self + .prepare_external_sql(entry, range.start_ms, range.end_ms, Some(request)) + .await + { + Ok(prepared) => prepared, + Err(error) => { + return ClickHouseAccelerationOutcome::Fallback( + ClickHouseAccelerationFallback::Execution(error), + ) + } + }; + match execute_sql_dag_with_external( self.store.as_ref(), entry, catalog.as_ref(), + &prepared, range.start_ms, range.end_ms, range.cumulative, @@ -180,6 +272,38 @@ mod tests { rc::Rc, }; + struct FixedExactSubtree; + + #[async_trait] + impl ClickHouseExactBackend for FixedExactSubtree { + async fn execute( + &self, + request: &ClickHouseQueryRequest, + ) -> Result + { + assert_eq!( + request.sql, + "SELECT toInt64(2000) AS timestamp, toFloat64(10) AS divisor WHERE {from:UInt64} <= {to:UInt64}" + ); + assert_eq!(request.parameters.get("param_from"), Some(&"0".into())); + assert_eq!(request.parameters.get("param_to"), Some(&"2000".into())); + Ok(ClickHouseRawResponse { + status: StatusCode::OK, + headers: HeaderMap::new(), + body: Bytes::from_static( + br#"{"meta":[{"name":"timestamp","type":"Int64"},{"name":"divisor","type":"Float64"}],"data":[[2000,10.0]],"rows":1}"#, + ), + }) + } + + async fn ping( + &self, + ) -> Result + { + unreachable!() + } + } + fn relation_schema(names: &[(&str, DataType)]) -> SummarySchema { SummarySchema { fields: names @@ -498,6 +622,248 @@ mod tests { )); } + #[tokio::test] + async fn summary_store_and_clickhouse_leaf_join_project_end_to_end() { + use control_plane::query_plan::BoundClickHouseQuery; + + let (accelerator, _) = fixture(2_000).await; + let physical = accelerator + .active_physical_plan + .as_ref() + .unwrap() + .snapshot(); + let mut entry = physical.query_plan.entries.values().next().unwrap().clone(); + let left = QueryNodeId(1); + let external = QueryNodeId(6); + let join = QueryNodeId(7); + let project = QueryNodeId(8); + let left_schema = relation_schema(&[ + ("timestamp", DataType::Timestamp), + ("value", DataType::Float64), + ]); + let right_schema = relation_schema(&[ + ("timestamp", DataType::Timestamp), + ("divisor", DataType::Float64), + ]); + let joined_schema = relation_schema(&[ + ("timestamp", DataType::Timestamp), + ("value", DataType::Float64), + ("timestamp", DataType::Timestamp), + ("divisor", DataType::Float64), + ]); + let output_schema = relation_schema(&[ + ("timestamp", DataType::Timestamp), + ("ratio", DataType::Float64), + ]); + entry + .nodes + .retain(|id, _| *id == QueryNodeId(0) || *id == left); + entry.nodes.insert( + external, + QueryPlanNode::ExternalSqlLeaf { + query: BoundClickHouseQuery { + sql: "SELECT toInt64(2000) AS timestamp, toFloat64(10) AS divisor WHERE {from:UInt64} <= {to:UInt64}".into(), + parameters: BTreeMap::new(), + start_parameter: Some("from".into()), + end_parameter: Some("to".into()), + output_schema: right_schema.clone(), + }, + }, + ); + entry.nodes.insert( + join, + QueryPlanNode::RelationalJoin { + inputs: [left, external], + join_kind: planner_types::pre_asap::JoinKind::Inner, + pred: serde_json::to_value(Predicate(Rc::new(QueryExpr::Compare { + left: Rc::new(QueryExpr::Column(0)), + op: CompareOpKind::Eq, + right: Rc::new(QueryExpr::Column(2)), + }))) + .unwrap(), + left_schema, + right_schema, + output_schema: joined_schema.clone(), + }, + ); + entry.nodes.insert( + project, + QueryPlanNode::Relational { + input: join, + operation: serde_json::to_value(ValueOperation::Project { + cols: vec![ + ProjectItem { + alias: Some("timestamp".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, + }) + .unwrap(), + input_schema: joined_schema, + output_schema, + }, + ); + entry.root = project; + let accelerator = accelerator.with_exact_backend(Arc::new(FixedExactSubtree)); + let prepared = accelerator + .prepare_external_sql(&entry, 0, 2_000, None) + .await + .unwrap(); + let ClickHouseDagOutcome::Accelerated(result) = execute_sql_dag_with_external( + accelerator.store.as_ref(), + &entry, + physical.summary_catalog.as_ref().unwrap(), + &prepared, + 0, + 2_000, + true, + ) else { + panic!("mixed DAG should execute") + }; + assert_eq!( + String::from_utf8(result.encode(ClickHouseFormat::TabSeparated).unwrap()).unwrap(), + "1970-01-01T00:00:02\t0.5\n" + ); + // The same published leaf is also exercised against an actual + // ClickHouse HTTP endpoint in differential/integration environments. + if let Ok(base_url) = std::env::var("CLICKHOUSE_URL") { + let real = CatalogClickHouseAccelerator::empty(accelerator.store.clone()) + .with_exact_backend(Arc::new( + super::super::fallback::ClickHouseHttpFallback::new(base_url, "default".into()), + )); + let mut headers = HeaderMap::new(); + if let Ok(user) = std::env::var("CLICKHOUSE_USER") { + headers.insert("x-clickhouse-user", user.parse().unwrap()); + } + if let Ok(password) = std::env::var("CLICKHOUSE_PASSWORD") { + headers.insert("x-clickhouse-key", password.parse().unwrap()); + } + let context = ClickHouseQueryRequest { + method: Method::POST, + sql: String::new(), + body: Bytes::new(), + parameters: BTreeMap::new(), + headers, + }; + let prepared = real + .prepare_external_sql(&entry, 0, 2_000, Some(&context)) + .await + .unwrap(); + let ClickHouseDagOutcome::Accelerated(result) = execute_sql_dag_with_external( + real.store.as_ref(), + &entry, + physical.summary_catalog.as_ref().unwrap(), + &prepared, + 0, + 2_000, + true, + ) else { + panic!("real ClickHouse mixed DAG should execute") + }; + assert_eq!( + String::from_utf8(result.encode(ClickHouseFormat::TabSeparated).unwrap()).unwrap(), + "1970-01-01T00:00:02\t0.5\n" + ); + } + let QueryPlanNode::ExternalSqlLeaf { query } = entry.nodes.get_mut(&external).unwrap() + else { + unreachable!() + }; + query.output_schema.fields[0].name = "wrong_timestamp".into(); + assert!(matches!( + execute_sql_dag_with_external( + accelerator.store.as_ref(), + &entry, + physical.summary_catalog.as_ref().unwrap(), + &prepared, + 0, + 2_000, + true, + ), + ClickHouseDagOutcome::Fallback(_) + )); + } + + #[tokio::test] + async fn real_clickhouse_datetime64_external_leaf_is_typed() { + use control_plane::query_plan::BoundClickHouseQuery; + let Ok(base_url) = std::env::var("CLICKHOUSE_URL") else { + return; + }; + let schema = relation_schema(&[("timestamp", DataType::Timestamp)]); + let leaf = QueryNodeId(0); + let entry = QueryPlanEntry { + language: QueryLanguage::ClickHouseSql, + query_id: "datetime64".into(), + canonical_query: "datetime64".into(), + fixed_evaluation: Some(FixedEvaluationRange { + start_ms: 0, + end_ms: 2_000, + cumulative: false, + }), + root: leaf, + nodes: BTreeMap::from([( + leaf, + QueryPlanNode::ExternalSqlLeaf { + query: BoundClickHouseQuery { + sql: "SELECT toDateTime64('1970-01-01 00:00:00.123', 3, 'UTC') AS timestamp WHERE {from:UInt64} <= {to:UInt64}".into(), + parameters: BTreeMap::new(), + start_parameter: Some("from".into()), + end_parameter: Some("to".into()), + output_schema: schema, + }, + }, + )]), + instant: InstantExecution { + lookback_ms: 2_000, + full_history: false, + cumulative_readout: false, + }, + fallback: FallbackPolicy::ExactBackend, + }; + let accelerator = CatalogClickHouseAccelerator::empty(Arc::new(SketchStore::new())) + .with_exact_backend(Arc::new( + super::super::fallback::ClickHouseHttpFallback::new(base_url, "default".into()), + )); + let mut headers = HeaderMap::new(); + if let Ok(user) = std::env::var("CLICKHOUSE_USER") { + headers.insert("x-clickhouse-user", user.parse().unwrap()); + } + if let Ok(password) = std::env::var("CLICKHOUSE_PASSWORD") { + headers.insert("x-clickhouse-key", password.parse().unwrap()); + } + let context = ClickHouseQueryRequest { + method: Method::POST, + sql: String::new(), + body: Bytes::new(), + parameters: BTreeMap::new(), + headers, + }; + let prepared = accelerator + .prepare_external_sql(&entry, 0, 2_000, Some(&context)) + .await + .unwrap(); + let encoded = prepared[&leaf] + .clone() + .into_result() + .unwrap() + .encode(ClickHouseFormat::TabSeparated) + .unwrap(); + assert_eq!( + String::from_utf8(encoded).unwrap(), + "1970-01-01T00:00:00.123\n" + ); + } + #[tokio::test(flavor = "current_thread")] async fn real_clickhouse_reader_enters_backfill_service_lifecycle() { let Ok(base_url) = std::env::var("CLICKHOUSE_URL") else { 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 61985220..ef071576 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 @@ -12,7 +12,9 @@ use crate::{ use asap_types::summary_catalog::SummaryCatalog; use control_plane::query_plan::{QueryNodeId, QueryPlanEntry, QueryPlanNode}; use planner_types::post_asap::ValueOperation; -use std::collections::BTreeSet; +use std::collections::{BTreeMap, BTreeSet}; + +pub type PreparedSqlLeaves = BTreeMap; #[derive(Debug, Clone, PartialEq, Eq)] pub enum ClickHouseDagFallback { @@ -29,6 +31,144 @@ pub enum ClickHouseDagOutcome { Fallback(ClickHouseDagFallback), } +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 reachable_entry(entry: &QueryPlanEntry, root: QueryNodeId) -> Result { + let reachable = entry + .topological_order_from(root) + .map_err(|error| error.to_string())?; + let mut subtree = entry.clone(); + subtree.root = root; + subtree.nodes.retain(|id, _| reachable.contains(id)); + Ok(subtree) +} + +fn execute_relation_subtree( + index: &SketchStore, + entry: &QueryPlanEntry, + root: QueryNodeId, + expected_schema: &planner_types::post_asap::SummarySchema, + prepared: &PreparedSqlLeaves, + t0_ms: u64, + t1_ms: u64, + is_cumulative: bool, +) -> Result { + match entry.nodes.get(&root) { + Some(QueryPlanNode::ExternalSqlLeaf { query }) => { + if &query.output_schema != expected_schema { + return Err("external SQL leaf schema differs from its parent edge".into()); + } + prepared + .get(&root) + .cloned() + .ok_or_else(|| "published external SQL leaf was not prepared".into()) + } + Some(QueryPlanNode::Relational { + input, + operation, + input_schema, + output_schema, + }) => { + let input = execute_relation_subtree( + index, + entry, + *input, + input_schema, + prepared, + 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, + prepared, + t0_ms, + t1_ms, + is_cumulative, + )?; + let right = execute_relation_subtree( + index, + entry, + inputs[1], + right_schema, + prepared, + 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 subtree = reachable_entry(entry, root)?; + let outcome = + execute_query_plan_from_readout(index, &subtree, root, t0_ms, t1_ms, is_cumulative) + .map_err(|error| format!("summary subtree failed: {error:?}"))?; + for binding in subtree.materialization_bindings() { + let origin = binding + .pane_origin_ms + .ok_or_else(|| "summary leaf has no pane origin".to_owned())?; + let start = i64::try_from(t0_ms).map_err(|_| "start exceeds i64".to_owned())?; + let end = i64::try_from(t1_ms).map_err(|_| "end exceeds i64".to_owned())?; + let pane = i64::try_from(binding.window_ms) + .map_err(|_| "pane duration 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 summary leaf coverage for pane {} origin {}", + binding.window_ms, origin + )); + } + } + 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)), + } +} + fn complete_pane_coverage( coverage: Option<(u64, u64)>, requested: (u64, u64), @@ -49,12 +189,75 @@ pub fn execute_sql_dag( t0_ms: u64, t1_ms: u64, is_cumulative: bool, +) -> ClickHouseDagOutcome { + execute_sql_dag_with_external( + index, + entry, + sds, + &PreparedSqlLeaves::new(), + t0_ms, + t1_ms, + is_cumulative, + ) +} + +pub fn execute_sql_dag_with_external( + index: &SketchStore, + entry: &QueryPlanEntry, + sds: &SummaryCatalog, + prepared: &PreparedSqlLeaves, + t0_ms: u64, + t1_ms: u64, + is_cumulative: bool, ) -> ClickHouseDagOutcome { if let Err(error) = validate_payload(Some(sds), entry, sds.plan_id, sds.plan_version) { return ClickHouseDagOutcome::Fallback(ClickHouseDagFallback::UnsupportedPlan( error.to_string(), )); } + let has_mixed_relational = entry.topological_order().is_ok_and(|ids| { + ids.iter().any(|id| { + matches!( + entry.nodes.get(id), + Some(QueryPlanNode::RelationalJoin { .. } | QueryPlanNode::ExternalSqlLeaf { .. }) + ) + }) + }); + if has_mixed_relational { + let root_schema = match entry.nodes.get(&entry.root) { + Some(QueryPlanNode::Relational { output_schema, .. }) + | Some(QueryPlanNode::RelationalJoin { output_schema, .. }) => output_schema, + Some(QueryPlanNode::ExternalSqlLeaf { query }) => &query.output_schema, + _ => { + return ClickHouseDagOutcome::Fallback(ClickHouseDagFallback::UnsupportedPlan( + "mixed SQL plan root has no relation schema".into(), + )) + } + }; + let relation = match execute_relation_subtree( + index, + entry, + entry.root, + root_schema, + prepared, + t0_ms, + t1_ms, + is_cumulative, + ) { + Ok(relation) => relation, + Err(error) => { + return ClickHouseDagOutcome::Fallback(ClickHouseDagFallback::UnsupportedPlan( + error, + )) + } + }; + 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 69402b65..fb4fd83c 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 @@ -9,6 +9,7 @@ use arrow::{ datatypes::{DataType as ArrowDataType, Field, Schema}, record_batch::RecordBatch, }; +use chrono::{DateTime, NaiveDateTime, TimeZone}; use planner_types::{ post_asap::{SummaryFamilyType, SummaryNode, SummarySchema, ValueOperation}, pre_asap::{ArithmeticOpKind, CompareOpKind, DataType, QueryExpr, ScalarValue, SortKey}, @@ -46,7 +47,104 @@ enum Cell { Timestamp(i64), } -#[derive(Debug)] +fn cells_equal(left: &Cell, right: &Cell) -> bool { + match (left, right) { + (Cell::Null, _) | (_, Cell::Null) => false, + (Cell::Int64(left), Cell::Int64(right)) => left == right, + (Cell::Float64(left), Cell::Float64(right)) => left.to_bits() == right.to_bits(), + (Cell::Utf8(left), Cell::Utf8(right)) => left == right, + (Cell::Bool(left), Cell::Bool(right)) => left == right, + (Cell::Timestamp(left), Cell::Timestamp(right)) => left == right, + _ => false, + } +} + +fn json_cell( + value: &serde_json::Value, + dtype: &DataType, + nullable: bool, + clickhouse_type: &str, +) -> Result { + if value.is_null() && nullable { + return Ok(Cell::Null); + } + let invalid = || { + ClickHouseRelationalError::Invalid(format!( + "external value {value} does not match {dtype:?}" + )) + }; + match dtype { + DataType::Int64 => value.as_i64().map(Cell::Int64).ok_or_else(invalid), + DataType::Float64 => value.as_f64().map(Cell::Float64).ok_or_else(invalid), + DataType::Utf8 => value + .as_str() + .map(|value| Cell::Utf8(value.into())) + .ok_or_else(invalid), + DataType::Bool => value.as_bool().map(Cell::Bool).ok_or_else(invalid), + DataType::Timestamp => parse_clickhouse_timestamp(value, clickhouse_type) + .map(Cell::Timestamp) + .ok_or_else(invalid), + } +} + +fn parse_clickhouse_timestamp(value: &serde_json::Value, clickhouse_type: &str) -> Option { + let clickhouse_type = clickhouse_type + .strip_prefix("Nullable(") + .and_then(|value| value.strip_suffix(')')) + .unwrap_or(clickhouse_type); + if clickhouse_type == "Int64" { + return value.as_i64(); + } + let text = value.as_str()?; + if let Ok(timestamp) = DateTime::parse_from_rfc3339(text) { + return Some(timestamp.timestamp_millis()); + } + let (scale, timezone) = if clickhouse_type == "DateTime" { + (0, "UTC") + } else if let Some(timezone) = clickhouse_type + .strip_prefix("DateTime(") + .and_then(|value| value.strip_suffix(')')) + { + (0, timezone.trim().trim_matches('\'')) + } else { + let args = clickhouse_type + .strip_prefix("DateTime64(")? + .strip_suffix(')')?; + let mut args = args.split(',').map(str::trim); + let scale = args.next()?.parse::().ok()?; + if scale > 9 { + return None; + } + let timezone = args.next().unwrap_or("UTC").trim_matches('\''); + if args.next().is_some() { + return None; + } + (scale, timezone) + }; + let fraction_digits = text + .split_once('.') + .map(|(_, fraction)| fraction.len()) + .unwrap_or(0); + if fraction_digits != scale { + return None; + } + let naive = NaiveDateTime::parse_from_str( + text, + if scale == 0 { + "%Y-%m-%d %H:%M:%S" + } else { + "%Y-%m-%d %H:%M:%S%.f" + }, + ) + .ok()?; + let timezone: chrono_tz::Tz = timezone.parse().ok()?; + timezone + .from_local_datetime(&naive) + .single() + .map(|value| value.timestamp_millis()) +} + +#[derive(Clone, Debug)] pub struct ClickHouseRelation { rows: Vec>, fields: Vec<(String, DataType, bool)>, @@ -54,6 +152,81 @@ pub struct ClickHouseRelation { } impl ClickHouseRelation { + pub fn from_json_compact( + schema: &SummarySchema, + body: &[u8], + ) -> Result { + let document: serde_json::Value = serde_json::from_slice(body) + .map_err(|error| ClickHouseRelationalError::Invalid(error.to_string()))?; + let meta = document + .get("meta") + .and_then(serde_json::Value::as_array) + .ok_or_else(|| { + ClickHouseRelationalError::Invalid( + "ClickHouse JSONCompact response has no typed metadata".into(), + ) + })?; + let data = document + .get("data") + .and_then(serde_json::Value::as_array) + .ok_or_else(|| { + ClickHouseRelationalError::Invalid( + "ClickHouse JSONCompact response has no data rows".into(), + ) + })?; + let fields = fields_from_schema(schema); + if meta.len() != fields.len() + || meta + .iter() + .zip(&fields) + .any(|(actual, (name, dtype, nullable))| { + actual.get("name").and_then(serde_json::Value::as_str) != Some(name) + || !clickhouse_type_matches( + actual.get("type").and_then(serde_json::Value::as_str), + dtype, + *nullable, + ) + }) + { + return Err(ClickHouseRelationalError::Invalid( + "ClickHouse external metadata differs from its planned schema".into(), + )); + } + let mut rows = Vec::with_capacity(data.len()); + for encoded in data { + let encoded = encoded.as_array().ok_or_else(|| { + ClickHouseRelationalError::Invalid( + "ClickHouse JSONCompact row is not an array".into(), + ) + })?; + if encoded.len() != fields.len() { + return Err(ClickHouseRelationalError::Invalid( + "ClickHouse external row differs from its planned schema".into(), + )); + } + rows.push( + encoded + .iter() + .zip(&fields) + .zip(meta) + .map(|((value, (_, dtype, nullable)), metadata)| { + json_cell( + value, + dtype, + *nullable, + metadata["type"].as_str().expect("metadata validated"), + ) + }) + .collect::, _>>()?, + ); + } + Ok(Self { + rows, + fields, + coverage: None, + }) + } + pub fn from_series_rows( schema: &SummarySchema, series: Vec<(BTreeMap, Vec<(i64, f64)>)>, @@ -90,9 +263,99 @@ impl ClickHouseRelation { } } +fn clickhouse_type_matches(actual: Option<&str>, expected: &DataType, nullable: bool) -> bool { + let Some(mut actual) = actual else { + return false; + }; + if nullable { + let Some(inner) = actual + .strip_prefix("Nullable(") + .and_then(|value| value.strip_suffix(')')) + else { + return false; + }; + actual = inner; + } else if actual.starts_with("Nullable(") { + return false; + } + match expected { + DataType::Int64 => actual == "Int64", + DataType::Float64 => actual == "Float64", + DataType::Utf8 => actual == "String", + DataType::Bool => actual == "Bool", + DataType::Timestamp => { + actual == "Int64" + || actual == "DateTime" + || actual.starts_with("DateTime(") + || actual.starts_with("DateTime64(") + } + } +} + 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((ls, le)), Some((rs, re))) => { + (ls.max(rs) <= le.min(re)).then_some((ls.max(rs), le.min(re))) + } + (Some(value), None) | (None, Some(value)) => Some(value), + (None, None) => 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 rows = Vec::new(); + for left_row in &left.rows { + let left_value = + left_row + .get(*left_key) + .ok_or(ClickHouseRelationalError::ColumnOutOfRange( + *left_key, + left_row.len(), + ))?; + for right_row in &right.rows { + let right_value = right_row.get(right_local_key).ok_or( + ClickHouseRelationalError::ColumnOutOfRange(right_local_key, right_row.len()), + )?; + if cells_equal(left_value, right_value) { + let mut joined = left_row.clone(); + joined.extend(right_row.iter().cloned()); + rows.push(joined); + } + } + } + Ok(ClickHouseRelation { + rows, + fields: fields_from_schema(output_schema), + coverage, + }) + } + pub fn apply_filter( &self, pred: &planner_types::pre_asap::Predicate, 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 b2c1d8f5..fc035c3c 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 @@ -287,7 +287,9 @@ impl QueryNodeRuntime for PhysicalQueryRuntime<'_> { } QueryPlanNode::Logical { .. } | QueryPlanNode::CandidateTopK { .. } - | QueryPlanNode::Relational { .. } => Err(PhysicalNodeError::Fallback( + | QueryPlanNode::Relational { .. } + | QueryPlanNode::RelationalJoin { .. } + | QueryPlanNode::ExternalSqlLeaf { .. } => Err(PhysicalNodeError::Fallback( "logical node requires installed logical runtime".into(), )), QueryPlanNode::ExactFallback { reason } => { diff --git a/docs/developer_docs/query-engine/clickhouse-sql-support.md b/docs/developer_docs/query-engine/clickhouse-sql-support.md index 4bfd87bb..45c176d4 100644 --- a/docs/developer_docs/query-engine/clickhouse-sql-support.md +++ b/docs/developer_docs/query-engine/clickhouse-sql-support.md @@ -62,6 +62,28 @@ 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. +### Exact subtree artifacts + +ASAPPlanner currently identifies a canonical `KeepPreAsap` cut but does not +render that cut back to ClickHouse SQL. The workload generator must therefore +provide an `exact_subtrees` artifact containing the canonical subtree, its +fingerprint, the equivalent exact SQL, parameter contract, and result schema. +This is an explicit input to compilation; it is not generated automatically by +the control plane or inferred at serving time. + +The compiler matches the full canonical subtree and fingerprint and consumes +every artifact exactly once. It also requires the artifact schema to equal the +Planner cut schema. Missing, duplicate, mismatched, or unused artifacts prevent +publication of the accelerated SQL entry. Such a workload remains on the +whole-query ClickHouse fallback path until a complete artifact set is supplied. + +For a mixed plan, publication additionally derives every summary child's +relation schema from its SummaryCatalog `DataDescriptor`, physical grouping, +and exact readout. The derived schema must exactly equal the corresponding Join +input schema. If the catalog and readout do not determine a schema, the mixed +plan is rejected rather than interpreting summary rows using a parent-declared +schema. + ## Execution and fallback The accelerated path supports the relational operations represented by the