From 972d508c8ad44e1e5e28774ae5a90f7e6155033b Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 08:37:56 -0600 Subject: [PATCH 1/2] refactor(clickhouse): remove unused relational tree executor --- .../relational_adapter.rs | 216 +++--------------- data_plane/src/query_engines/canonical/mod.rs | 9 +- .../src/query_engines/canonical/relational.rs | 62 ----- 3 files changed, 33 insertions(+), 254 deletions(-) delete mode 100644 data_plane/src/query_engines/canonical/relational.rs 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 0bfc9fc91..b5ddf15df 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 @@ -14,18 +14,10 @@ use arrow::{ }; use chrono::{DateTime, NaiveDateTime, TimeZone}; use planner_types::{ - post_asap::{SummaryFamilyType, SummaryNode, SummarySchema, ValueOperation}, + post_asap::{SummaryFamilyType, SummarySchema, ValueOperation}, pre_asap::{ArithmeticOpKind, CompareOpKind, DataType, QueryExpr, ScalarValue, SortKey}, }; -use crate::query_engines::{ - asap_query_engine::{ - summary_exec::ExecOutcome, - summary_executor::{GroupState, SummaryValue}, - }, - canonical::relational::RelationalAdapter, -}; - use super::clickhouse_result_adapter::ClickHouseQueryResult; #[derive(Debug, thiserror::Error, PartialEq)] @@ -456,81 +448,6 @@ impl ClickHouseRelationalAdapter { } } -impl RelationalAdapter for ClickHouseRelationalAdapter -where - E: crate::query_engines::canonical::executor::SummaryExecutor< - GroupKey = BTreeMap, - State = GroupState, - Value = SummaryValue, - >, -{ - type Relation = ClickHouseRelation; - type Error = ClickHouseRelationalError; - - fn relation_from_outcome( - &self, - node: &SummaryNode, - outcome: ExecOutcome, - ) -> Result { - let fields = fields(node); - let mut rows = Vec::new(); - let mut coverage = None; - let mut coverage_complete = true; - match outcome { - ExecOutcome::Value(groups) => { - for (group, value) in groups { - match value.coverage() { - Some(next) if coverage_complete => { - coverage = intersect_coverage(coverage, Some(next)); - } - None => { - coverage = None; - coverage_complete = false; - } - Some(_) => {} - } - match value { - SummaryValue::Points(points, _) => { - for (timestamp, value) in points { - rows.push(row_from_value(&fields, &group, timestamp, value)?); - } - } - SummaryValue::TopK(_, _) => { - return Err(ClickHouseRelationalError::Unsupported( - "TopK summary rows".into(), - )); - } - } - } - } - ExecOutcome::State(groups) => { - for (group, state, family) in groups { - let value = exact_value(&state, &family)?; - rows.push(row_from_value(&fields, &group, 0, value)?); - } - } - } - Ok(ClickHouseRelation { - rows, - fields, - coverage, - }) - } - - fn apply( - &self, - node: &SummaryNode, - operation: &ValueOperation, - input: Self::Relation, - ) -> Result { - self.apply_operation(operation, &node.schema, input) - } -} - -fn fields(node: &SummaryNode) -> Vec<(String, DataType, bool)> { - fields_from_schema(&node.schema) -} - fn fields_from_schema(schema: &SummarySchema) -> Vec<(String, DataType, bool)> { schema .fields @@ -545,20 +462,6 @@ fn fields_from_schema(schema: &SummarySchema) -> Vec<(String, DataType, bool)> { .collect() } -fn exact_value( - state: &GroupState, - family: &SummaryFamilyType, -) -> Result { - if !matches!(family, SummaryFamilyType::ExactAggregate(..)) { - return Err(ClickHouseRelationalError::Unsupported( - "unfinalized non-exact summary state".into(), - )); - } - state.exact_value(&None).ok_or_else(|| { - ClickHouseRelationalError::Invalid("exact accumulator cannot be finalized".into()) - }) -} - fn row_from_value( fields: &[(String, DataType, bool)], group: &BTreeMap, @@ -923,14 +826,6 @@ fn cell_cmp(left: &Cell, right: &Cell) -> Option { } } -fn intersect_coverage(current: Option<(u64, u64)>, next: Option<(u64, u64)>) -> Option<(u64, u64)> { - match (current, next) { - (None, next) => next, - (current, None) => current, - (Some(left), Some(right)) => Some((left.0.max(right.0), left.1.min(right.1))), - } -} - fn arrow_type(dtype: &DataType) -> ArrowDataType { match dtype { DataType::Null => ArrowDataType::Null, @@ -1073,54 +968,12 @@ fn build_array( #[cfg(test)] mod tests { use super::*; - use crate::query_engines::canonical::{ - executor::SummaryExecutor, relational::execute_relational, - }; use planner_types::{ - post_asap::{ExecutionTiming, SketchQuery, SummaryExpr, SummaryField, SummarySchema}, - pre_asap::{ColumnRef, GroupKeys, Predicate, ProjectItem, Reduction}, + post_asap::{SummaryField, SummarySchema}, + pre_asap::{GroupKeys, Predicate, ProjectItem}, }; use std::rc::Rc; - struct MockExecutor; - - impl SummaryExecutor for MockExecutor { - type Handle = (); - type State = GroupState; - type Value = SummaryValue; - type Error = (); - type GroupKey = BTreeMap; - - fn find_candidates( - &self, - _: &SummaryFamilyType, - _: &ColumnRef, - _: &Reduction, - _: &SummaryNode, - ) -> Result, Self::Error> { - unreachable!() - } - - fn fetch_state(&self, _: &Self::Handle) -> Result { - unreachable!() - } - - fn merge_states(&self, _: Vec) -> Result { - unreachable!() - } - - fn readout(&self, _: &Self::State, _: &SketchQuery) -> Result { - unreachable!() - } - - fn logical(&self, _: &QueryExpr) -> Result { - Ok(SummaryValue::Points( - vec![(10, 2.0), (20, 3.0), (30, 1.0)], - Some((0, 40)), - )) - } - } - fn schema(fields: &[(&str, DataType)]) -> SummarySchema { SummarySchema { fields: fields @@ -1137,22 +990,6 @@ mod tests { } } - fn value_node( - child: Rc, - operation: ValueOperation, - schema: SummarySchema, - ) -> Rc { - Rc::new(SummaryNode { - expr: SummaryExpr::ValueOperation { - child, - operation, - timing: ExecutionTiming::ReadTime, - }, - schema, - guarantee: None, - }) - } - /// Map entries remain ordered pairs, including duplicate keys and null values. #[test] fn map_transport_retains_duplicate_keys_and_null_values() { @@ -1199,17 +1036,32 @@ mod tests { #[test] fn executes_filter_project_arithmetic_sort_and_limit_chain() { let input_schema = schema(&[("ts", DataType::Timestamp), ("sum", DataType::Float64)]); - let leaf = Rc::new(SummaryNode { - expr: SummaryExpr::KeepPreAsap(Rc::new(QueryExpr::Literal(ScalarValue::Int64(0)))), - schema: input_schema.clone(), - guarantee: None, - }); + let adapter = ClickHouseRelationalAdapter; + let input = ClickHouseRelation { + rows: vec![ + vec![Cell::Timestamp(10), Cell::Float64(2.0)], + vec![Cell::Timestamp(20), Cell::Float64(3.0)], + vec![Cell::Timestamp(30), Cell::Float64(1.0)], + ], + fields: fields_from_schema(&input_schema), + coverage: Some((0, 40)), + }; + let mut relation = adapter + .apply_filter( + &Predicate(QueryExpr::Compare { + left: Rc::new(QueryExpr::Column(1)), + op: CompareOpKind::Gt, + right: Rc::new(QueryExpr::Literal(ScalarValue::Float64(1.0))), + }), + input, + ) + .unwrap(); + assert_eq!(relation.rows.len(), 2); let projected_schema = schema(&[ ("bucket", DataType::Timestamp), ("score", DataType::Float64), ]); - let projected = value_node( - leaf, + for operation in [ ValueOperation::Project { cols: vec![ ProjectItem { @@ -1227,10 +1079,6 @@ mod tests { ], qualifier: None, }, - projected_schema.clone(), - ); - let sorted = value_node( - projected, ValueOperation::Sort { keys: vec![SortKey { expr: QueryExpr::Column(1), @@ -1239,16 +1087,12 @@ mod tests { }], partition_by: GroupKeys::none(), }, - projected_schema.clone(), - ); - let limited = value_node( - sorted, ValueOperation::Limit { n: 1, offset: 0 }, - projected_schema, - ); - - let relation = execute_relational(&limited, &MockExecutor, &ClickHouseRelationalAdapter) - .expect("supported SQL chain should execute"); + ] { + relation = adapter + .apply_operation(&operation, &projected_schema, relation) + .expect("installed SQL operators should execute"); + } assert_eq!(relation.coverage, Some((0, 40))); let result = relation.into_result().unwrap(); let batch = &result.batches[0]; diff --git a/data_plane/src/query_engines/canonical/mod.rs b/data_plane/src/query_engines/canonical/mod.rs index 61290336b..c170e5d0e 100644 --- a/data_plane/src/query_engines/canonical/mod.rs +++ b/data_plane/src/query_engines/canonical/mod.rs @@ -1,10 +1,7 @@ -//! Language-independent execution of ASAPPlanner's post-ASAP DAG. +//! Compatibility re-exports for the existing summary executor. //! -//! The implementation remains in its compatibility location while PromQL -//! callers migrate. These re-exports give SQL and PromQL one execution API -//! without changing the existing PromQL types or behavior. - -pub mod relational; +//! Installed SQL serving executes QueryPlan nodes through its typed relational +//! adapter. These aliases remain for callers of the summary execution API. pub mod executor { pub use crate::query_engines::asap_query_engine::summary_exec::{ diff --git a/data_plane/src/query_engines/canonical/relational.rs b/data_plane/src/query_engines/canonical/relational.rs deleted file mode 100644 index 816e97ba6..000000000 --- a/data_plane/src/query_engines/canonical/relational.rs +++ /dev/null @@ -1,62 +0,0 @@ -//! Relational query-time operators over the shared summary executor. -//! -//! The legacy executor intentionally keeps `Value` opaque. This companion -//! adapter lets a language runtime interpret planner-owned `ValueOperation` -//! nodes without widening that trait or changing existing PromQL behavior. - -use planner_types::post_asap::{SummaryExpr, SummaryNode, ValueOperation}; - -use super::executor::{execute, ExecError, ExecOutcome, SummaryExecutor}; - -pub trait RelationalAdapter { - type Relation; - type Error; - - fn relation_from_outcome( - &self, - node: &SummaryNode, - outcome: ExecOutcome, - ) -> Result; - - fn apply( - &self, - node: &SummaryNode, - operation: &ValueOperation, - input: Self::Relation, - ) -> Result; -} - -#[derive(Debug)] -pub enum RelationalExecError { - Executor(ExecError), - Adapter(AdapterError), -} - -/// Recursively evaluates query-time relational nodes and delegates every -/// summary/state node to the unchanged shared summary executor. -pub fn execute_relational( - node: &SummaryNode, - executor: &E, - adapter: &A, -) -> Result> -where - E: SummaryExecutor, - A: RelationalAdapter, -{ - match &node.expr { - SummaryExpr::ValueOperation { - child, operation, .. - } => { - let input = execute_relational(child, executor, adapter)?; - adapter - .apply(node, operation, input) - .map_err(RelationalExecError::Adapter) - } - _ => { - let outcome = execute(node, executor).map_err(RelationalExecError::Executor)?; - adapter - .relation_from_outcome(node, outcome) - .map_err(RelationalExecError::Adapter) - } - } -} From d313d4039ae4bbbf8794eb91b66c74ec28c0120a Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 08:41:50 -0600 Subject: [PATCH 2/2] test(clickhouse): construct the shared predicate wrapper --- .../asap_clickhouse_query_engine/relational_adapter.rs | 4 ++-- 1 file changed, 2 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 b5ddf15df..15945be21 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 @@ -1048,11 +1048,11 @@ mod tests { }; let mut relation = adapter .apply_filter( - &Predicate(QueryExpr::Compare { + &Predicate(Rc::new(QueryExpr::Compare { left: Rc::new(QueryExpr::Column(1)), op: CompareOpKind::Gt, right: Rc::new(QueryExpr::Literal(ScalarValue::Float64(1.0))), - }), + })), input, ) .unwrap();