From 63455ab580baad914aa2b20a62aba620d7ed6e7b Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 05:39:49 -0600 Subject: [PATCH 1/2] feat(sql): lower explicit temporal aggregates --- crates/frontend-sql/src/sql/mod.rs | 198 +++++++++++++++++- crates/frontend-sql/tests/sql_lowering.rs | 113 ++++++++++ .../tests/sql_to_post_asap.rs | 40 +++- crates/sql-function-catalog/src/lib.rs | 19 ++ 4 files changed, 358 insertions(+), 12 deletions(-) diff --git a/crates/frontend-sql/src/sql/mod.rs b/crates/frontend-sql/src/sql/mod.rs index b7f94a1f..a2e03f64 100644 --- a/crates/frontend-sql/src/sql/mod.rs +++ b/crates/frontend-sql/src/sql/mod.rs @@ -7,11 +7,12 @@ //! unresolved `ColumnRef`s directly (issue #179) — the same tree shape //! [`resolve_root`](asap_types::pre_asap::resolve_root) binds to canonical, //! positional `QueryExpr`. Unlike PromQL's front end, SQL's -//! `Aggregate` nodes need no reduction-shape decision at construction time — -//! DataFusion's `Aggregate` plan node is always `Reduction::Reduce`, never -//! PromQL's per-series `PerEntity` (there is no windowed/subquery child -//! concept in SQL) — so this front end always builds `Reduce` directly. It -//! does still have to fold a `WHERE` directly over a bare table scan onto +//! Ordinary SQL `Aggregate` nodes are `Reduction::Reduce`. The explicit +//! `asap_rate`/`asap_increase`/`asap_last` bridge is the narrow exception: it +//! spells a time-series range reducer with an explicit value, time-index, and +//! window and therefore lowers to the same `TimeRange` + `PerEntity` shape as +//! its PromQL counterpart. The front end also has to fold a `WHERE` directly +//! over a bare table scan onto //! `Scan.predicates` itself (`filter_or_fold`) — canonical's invariant that a //! `Filter` never sits directly over a `Scan` — since front ends producing //! this shape are responsible for it now, not a converter. @@ -25,6 +26,7 @@ use std::rc::Rc; use std::sync::Arc; +use std::time::Duration; use datafusion::arrow::compute::kernels::cast_utils::parse_interval_month_day_nano; use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field}; @@ -57,7 +59,8 @@ use asap_types::pre_asap::query_expr::{ }; use asap_types::pre_asap::schema::{DataType, Schema}; use asap_types::pre_asap::{ - ColumnRef, CompareOpKind, JoinKind, RelationalSetOpKind, ScalarValue, WindowFuncKind, + resolve_column_ref, resolve_root, ColumnRef, CompareOpKind, JoinKind, RelationalSetOpKind, + ScalarValue, WindowFuncKind, }; use asap_types::types::AccuracyTarget; use asap_types::workload::SqlDialect; @@ -645,15 +648,30 @@ impl<'a> SqlLowerer<'a> { return self.lower_plan(&proj.input); } let child = Rc::new(self.lower_plan(&proj.input)?); + let temporal_input = plan_has_temporal_aggregate(&proj.input); let cols = proj .expr .iter() .map(|e| match e { - Expr::Alias(a) => df_expr_to_unresolved(&a.expr).map(|expr| ProjectItem { - expr, - alias: Some(a.name.clone()), - }), - _ => df_expr_to_unresolved(e).map(|expr| ProjectItem { expr, alias: None }), + Expr::Alias(a) => { + let expr = if temporal_input && is_temporal_output_column(&a.expr) { + Unresolved::Column(ColumnRef::Named("value".into())) + } else { + df_expr_to_unresolved(&a.expr)? + }; + Ok::, LoweringError>(ProjectItem { + expr, + alias: Some(a.name.clone()), + }) + } + _ => { + let expr = if temporal_input && is_temporal_output_column(e) { + Unresolved::Column(ColumnRef::Named("value".into())) + } else { + df_expr_to_unresolved(e)? + }; + Ok::, LoweringError>(ProjectItem { expr, alias: None }) + } }) .collect::, _>>()?; Ok(Unresolved::Project { @@ -666,6 +684,10 @@ impl<'a> SqlLowerer<'a> { fn lower_aggregate(&self, agg: &logical_expr::Aggregate) -> Result { let input = self.lower_plan(&agg.input)?; + if agg.aggr_expr.iter().any(is_temporal_aggregate) { + return self.lower_temporal_aggregate(agg, input); + } + // `GROUPING SETS`/`ROLLUP`/`CUBE` emit several grouping levels from one // scan. `Aggregate.by` is a single key set, so each level becomes its own // `Aggregate` and they are merged (issue #118). @@ -746,6 +768,110 @@ impl<'a> SqlLowerer<'a> { }) } + fn lower_temporal_aggregate( + &self, + agg: &logical_expr::Aggregate, + input: Unresolved, + ) -> Result { + if agg.aggr_expr.len() != 1 { + return Err(LoweringError::UnsupportedFeature( + "an ASAP temporal aggregate cannot share an Aggregate node with another reducer" + .into(), + )); + } + let Expr::AggregateFunction(call) = unalias(&agg.aggr_expr[0]) else { + unreachable!("is_temporal_aggregate accepted a non-aggregate expression") + }; + let name = call.func.name().to_lowercase(); + let [value, timestamp, window] = call.args.as_slice() else { + unreachable!("ASAP temporal UDAF signatures require exactly three arguments") + }; + + let value_ref = reducer_col(&name, std::slice::from_ref(value))?; + let timestamp_ref = reducer_col(&name, std::slice::from_ref(timestamp))?; + let Expr::Literal(window) = unalias(window) else { + return Err(LoweringError::InvalidExpression(format!( + "{name} window_ms must be a positive integer literal" + ))); + }; + let window_ms = scalar_positive_u64(window).ok_or_else(|| { + LoweringError::InvalidExpression(format!( + "{name} window_ms must be a positive integer literal" + )) + })?; + + let resolved_input = resolve_root(&input)?; + let input_schema = resolved_input.output_schema().map_err(|error| { + LoweringError::InvalidExpression(format!( + "cannot derive temporal aggregate input schema: {error}" + )) + })?; + let timestamp_id = resolve_column_ref(×tamp_ref, &input_schema).map_err(|error| { + LoweringError::InvalidExpression(format!("{name} timestamp argument: {error}")) + })?; + if input_schema.time_index != Some(timestamp_id) { + return Err(LoweringError::InvalidExpression(format!( + "{name} timestamp argument must name the input schema's time-index column" + ))); + } + let value_id = resolve_column_ref(&value_ref, &input_schema).map_err(|error| { + LoweringError::InvalidExpression(format!("{name} value argument: {error}")) + })?; + if value_id == timestamp_id + || !matches!( + input_schema.columns[value_id].dtype, + DataType::Int64 | DataType::Float64 + ) + { + return Err(LoweringError::InvalidExpression(format!( + "{name} value argument must name a numeric non-time column" + ))); + } + + let mut cols = vec![ + ProjectItem { + alias: Some("ts".into()), + expr: Unresolved::Column(timestamp_ref.clone()), + }, + ProjectItem { + alias: Some("value".into()), + expr: Unresolved::Column(value_ref.clone()), + }, + ]; + for group in &agg.group_expr { + let group_ref = expr_to_group_ref(group)?; + let group_name = named_ref(&group_ref).to_string(); + if group_name != named_ref(×tamp_ref) && group_name != named_ref(&value_ref) { + cols.push(ProjectItem { + alias: Some(group_name), + expr: Unresolved::Column(group_ref), + }); + } + } + let child = Unresolved::Project { + cols, + qualifier: None, + child: Rc::new(input), + }; + let child = Unresolved::TimeRange { + range: Duration::from_millis(window_ms), + child: Rc::new(child), + }; + let intent = match name.as_str() { + "asap_rate" => AggIntent::Rate, + "asap_increase" => AggIntent::Increase, + "asap_last" => AggIntent::LastOverTime, + _ => unreachable!("is_temporal_aggregate admitted {name}"), + }; + Ok(Unresolved::Aggregate { + reduction: Reduction::PerEntity, + measures: vec![intent], + output_names: vec![], + having: None, + child: Rc::new(child), + }) + } + /// `GROUP BY ROLLUP/CUBE/GROUPING SETS` — multi-level grouping (issue #118). /// /// One scan produces several grouping levels; `Aggregate.by` holds a single @@ -1292,6 +1418,56 @@ fn lower_agg_intent(expr: &Expr) -> Result, LoweringError> } } +fn temporal_aggregate_name(expr: &Expr) -> Option { + let Expr::AggregateFunction(call) = unalias(expr) else { + return None; + }; + let name = call.func.name().to_lowercase(); + matches!(name.as_str(), "asap_rate" | "asap_increase" | "asap_last").then_some(name) +} + +fn is_temporal_aggregate(expr: &Expr) -> bool { + temporal_aggregate_name(expr).is_some() +} + +fn is_temporal_output_column(expr: &Expr) -> bool { + let Expr::Column(col) = unalias(expr) else { + return false; + }; + let name = col.name.to_lowercase(); + ["asap_rate(", "asap_increase(", "asap_last("] + .iter() + .any(|prefix| name.starts_with(prefix)) +} + +fn plan_has_temporal_aggregate(plan: &LogicalPlan) -> bool { + match plan { + LogicalPlan::Aggregate(agg) => agg.aggr_expr.iter().any(is_temporal_aggregate), + LogicalPlan::Filter(filter) => plan_has_temporal_aggregate(&filter.input), + LogicalPlan::SubqueryAlias(alias) => plan_has_temporal_aggregate(&alias.input), + _ => false, + } +} + +fn named_ref(col: &ColumnRef) -> &str { + match col { + ColumnRef::Named(name) | ColumnRef::Qualified { name, .. } => name, + ColumnRef::SampleValue | ColumnRef::Wildcard => { + unreachable!("reducer_col only returns named column references") + } + } +} + +fn scalar_positive_u64(value: &DfScalarValue) -> Option { + match value { + DfScalarValue::Int64(Some(v)) if *v > 0 => Some(*v as u64), + DfScalarValue::Int32(Some(v)) if *v > 0 => Some(*v as u64), + DfScalarValue::UInt64(Some(v)) if *v > 0 => Some(*v), + DfScalarValue::UInt32(Some(v)) if *v > 0 => Some(*v as u64), + _ => None, + } +} + /// ClickHouse's row-selecting `argMax(arg, val)` / `argMin(arg, val)` — /// "return `arg`'s value from the row where `val` is maximal/minimal". /// `Some(name)` for `"argmax"`/`"argmin"`, `None` for every other name (the diff --git a/crates/frontend-sql/tests/sql_lowering.rs b/crates/frontend-sql/tests/sql_lowering.rs index 0b99cd41..1df0312d 100644 --- a/crates/frontend-sql/tests/sql_lowering.rs +++ b/crates/frontend-sql/tests/sql_lowering.rs @@ -1538,6 +1538,119 @@ async fn lower_clickhouse(sql: &str) -> QueryExpr { .unwrap_or_else(|e| panic!("lower failed for {sql:?}: {e}")) } +fn temporal_aggregate(qe: &QueryExpr) -> (&AggIntent, std::time::Duration, &QueryExpr) { + match qe { + QueryExpr::Aggregate { + reduction: Reduction::PerEntity, + measures, + child, + .. + } => { + let QueryExpr::TimeRange { range, child } = child.as_ref() else { + panic!("temporal Aggregate must directly wrap TimeRange, got {child:?}"); + }; + (&measures[0], *range, child) + } + QueryExpr::Project { child, .. } | QueryExpr::Filter { child, .. } => { + temporal_aggregate(child) + } + other => panic!("expected temporal Aggregate, got {other:?}"), + } +} + +#[tokio::test] +async fn explicit_temporal_aggregates_share_promql_intents_and_timerange() { + for (function, expected) in [ + ("asap_rate", AggIntent::Rate), + ("asap_increase", AggIntent::Increase), + ("asap_last", AggIntent::LastOverTime), + ] { + let sql = format!( + "SELECT service, {function}(latency, ts, 300000) AS v \ + FROM metrics WHERE service = 'api' GROUP BY service" + ); + let qe = lower_clickhouse(&sql).await; + let (intent, range, child) = temporal_aggregate(&qe); + assert_eq!(intent, &expected); + assert_eq!(range, std::time::Duration::from_secs(300)); + assert!(matches!(child, QueryExpr::Project { child, .. } + if matches!(child.as_ref(), QueryExpr::Scan { predicates, .. } if predicates.len() == 1))); + + let QueryExpr::Project { cols, .. } = &qe else { + panic!("SELECT list must remain a Project, got {qe:?}"); + }; + assert!(matches!(cols[0].expr, QueryExpr::Column(2))); + assert_eq!(cols[1].alias.as_deref(), Some("v")); + assert!(matches!(cols[1].expr, QueryExpr::Column(1))); + } +} + +#[tokio::test] +async fn temporal_aggregate_rejects_non_timestamp_and_non_positive_window() { + for sql in [ + "SELECT asap_rate(latency, bytes, 300000) FROM metrics", + "SELECT asap_rate(latency, ts, 0) FROM metrics", + "SELECT asap_rate(latency, ts, bytes) FROM metrics", + ] { + let err = lower_sql_dialect( + sql, + &catalog(), + SqlDialect::ClickhouseSQL, + AccuracyTarget::Exact, + ) + .await + .expect_err("invalid temporal arguments must fail closed"); + assert!( + format!("{err}").contains("timestamp argument") + || format!("{err}").contains("window_ms"), + "unexpected error for {sql}: {err}" + ); + } +} + +#[tokio::test] +async fn temporal_aggregate_rejects_mixed_reducers() { + let err = lower_sql_dialect( + "SELECT asap_rate(latency, ts, 300000), sum(bytes) FROM metrics", + &catalog(), + SqlDialect::ClickhouseSQL, + AccuracyTarget::Exact, + ) + .await + .expect_err("one child cannot carry temporal and ordinary aggregate semantics"); + assert!(format!("{err}").contains("cannot share an Aggregate node")); +} + +#[tokio::test] +async fn project_filter_and_outer_aggregate_preserve_temporal_child() { + let qe = lower_clickhouse( + "SELECT max(v) FROM (\ + SELECT service, asap_rate(latency, ts, 300000) AS v \ + FROM metrics WHERE bytes > 0 GROUP BY service\ + ) r WHERE v >= 0", + ) + .await; + let QueryExpr::Project { child, .. } = &qe else { + panic!("expected outer SELECT Project, got {qe:?}"); + }; + let QueryExpr::Aggregate { + reduction: Reduction::Reduce(_), + measures, + child, + .. + } = child.as_ref() + else { + panic!("expected outer Aggregate, got {child:?}"); + }; + assert!(matches!(measures.as_slice(), [AggIntent::Max { .. }])); + let QueryExpr::Filter { child, .. } = child.as_ref() else { + panic!("derived-table WHERE must remain above the inner query, got {child:?}"); + }; + let (intent, range, _) = temporal_aggregate(child); + assert_eq!(intent, &AggIntent::Rate); + assert_eq!(range, std::time::Duration::from_secs(300)); +} + #[tokio::test] async fn count_if_lowers_to_a_sum_over_a_derived_indicator_column() { // ClickHouse's `countIf(cond)` has no DataFusion equivalent at all, so it diff --git a/crates/integration-tests/tests/sql_to_post_asap.rs b/crates/integration-tests/tests/sql_to_post_asap.rs index 0d1f9843..1c1c2a15 100644 --- a/crates/integration-tests/tests/sql_to_post_asap.rs +++ b/crates/integration-tests/tests/sql_to_post_asap.rs @@ -25,7 +25,7 @@ use asap_aware_mapping::{ search_workload, DefaultCostModel, Replacement, ReplacementStrategy, ReplacementSubDAG, SketchAlgorithmStrategy, TargetSubDAG, }; -use asap_frontend_sql::{lower_sql, SqlCatalog}; +use asap_frontend_sql::{lower_sql, lower_sql_dialect, SqlCatalog}; use asap_types::post_asap::{ compile_executable_dag, ExactKind, ExactParams, ExecutableOperatorPayload, GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams, SketchQuery, SummaryExpr, SummaryFamilyType, @@ -35,6 +35,7 @@ use asap_types::pre_asap::expr_ir::ColumnRef; use asap_types::pre_asap::query_expr::{QueryExpr, Reduction}; use asap_types::pre_asap::schema::{Column, DataType, Schema}; use asap_types::types::AccuracyTarget; +use asap_types::workload::SqlDialect; /// This crate has no "bind me one tree" public API any more — /// `SketchAlgorithmStrategy::replacements` always returns every candidate, and @@ -94,6 +95,43 @@ async fn lower(sql: &str, accuracy: AccuracyTarget) -> QueryExpr { .unwrap_or_else(|e| panic!("lower failed for {sql:?}: {e}")) } +#[tokio::test] +async fn clickhouse_temporal_sql_reuses_the_rate_physical_summary() { + let pre_asap = lower_sql_dialect( + "SELECT service, asap_rate(latency, ts, 300000) AS v \ + FROM metrics WHERE bytes > 0 GROUP BY service", + &catalog(), + SqlDialect::ClickhouseSQL, + AccuracyTarget::Exact, + ) + .await + .expect("explicit temporal SQL must lower"); + let physical = realize(inner_aggregate(&pre_asap)).expect("rate must be physically planned"); + let SummaryExpr::SummaryAgg { + family, + reduction, + child, + .. + } = &physical.expr + else { + panic!("expected a shared SummaryAgg, got {:?}", physical.expr); + }; + assert_eq!( + family, + &SummaryFamilyType::ExactAggregate(ExactKind::Rate, ExactParams::Rate) + ); + assert_eq!(reduction, &Reduction::PerEntity); + let SummaryExpr::KeepPreAsap(raw) = &child.expr else { + panic!( + "expected a retained temporal SQL input, got {:?}", + child.expr + ); + }; + assert!(matches!(raw.as_ref(), QueryExpr::TimeRange { range, child } + if *range == std::time::Duration::from_secs(300) + && matches!(child.as_ref(), QueryExpr::Project { .. }))); +} + /// The `Aggregate` node beneath the identity `Project` DataFusion's planner /// always wraps a top-level aggregate in — see the module docs above. fn inner_aggregate(qe: &QueryExpr) -> &QueryExpr { diff --git a/crates/sql-function-catalog/src/lib.rs b/crates/sql-function-catalog/src/lib.rs index 39645717..8cae4308 100644 --- a/crates/sql-function-catalog/src/lib.rs +++ b/crates/sql-function-catalog/src/lib.rs @@ -269,6 +269,25 @@ pub struct ClickHouseBuiltin { /// natively understands before physical planning). See the module doc and /// `asap-frontend-sql::sql::ClickHouseBuiltinRewrite`. pub const CLICKHOUSE_BUILTINS: &[ClickHouseBuiltin] = &[ + // Explicit time-series reducers. These deliberately survive under their + // own names: the SQL frontend validates (value, timestamp, window_ms) and + // lowers the window to QueryExpr::TimeRange rather than pretending these + // are ordinary tabular aggregates. + ClickHouseBuiltin { + name: "asap_rate", + arity: Arity::Exact(3), + rewrite: RewriteKind::PassThrough, + }, + ClickHouseBuiltin { + name: "asap_increase", + arity: Arity::Exact(3), + rewrite: RewriteKind::PassThrough, + }, + ClickHouseBuiltin { + name: "asap_last", + arity: Arity::Exact(3), + rewrite: RewriteKind::PassThrough, + }, ClickHouseBuiltin { name: "uniqexact", arity: Arity::Exact(1), From b4e2750a894262b0b188b2377443fb637aa7ea7b Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 05:53:03 -0600 Subject: [PATCH 2/2] fix(sql): require proven temporal series identity --- crates/frontend-sql/src/sql/mod.rs | 68 +++++++++-- crates/frontend-sql/tests/sql_lowering.rs | 115 +++++++++++++++++- .../tests/sql_to_post_asap.rs | 80 ++++++------ crates/sql-function-catalog/src/lib.rs | 5 - 4 files changed, 215 insertions(+), 53 deletions(-) diff --git a/crates/frontend-sql/src/sql/mod.rs b/crates/frontend-sql/src/sql/mod.rs index a2e03f64..f2153bed 100644 --- a/crates/frontend-sql/src/sql/mod.rs +++ b/crates/frontend-sql/src/sql/mod.rs @@ -8,7 +8,7 @@ //! [`resolve_root`](asap_types::pre_asap::resolve_root) binds to canonical, //! positional `QueryExpr`. Unlike PromQL's front end, SQL's //! Ordinary SQL `Aggregate` nodes are `Reduction::Reduce`. The explicit -//! `asap_rate`/`asap_increase`/`asap_last` bridge is the narrow exception: it +//! `asap_rate`/`asap_increase` bridge is the narrow exception: it //! spells a time-series range reducer with an explicit value, time-index, and //! window and therefore lowers to the same `TimeRange` + `PerEntity` shape as //! its PromQL counterpart. The front end also has to fold a `WHERE` directly @@ -828,6 +828,53 @@ impl<'a> SqlLowerer<'a> { ))); } + let mut group_ids = Vec::with_capacity(agg.group_expr.len()); + let mut group_refs = Vec::with_capacity(agg.group_expr.len()); + for group in &agg.group_expr { + let group_ref = expr_to_group_ref(group)?; + let group_id = resolve_column_ref(&group_ref, &input_schema).map_err(|error| { + LoweringError::InvalidExpression(format!("{name} GROUP BY column: {error}")) + })?; + if group_id == timestamp_id || group_id == value_id { + return Err(LoweringError::InvalidExpression(format!( + "{name} GROUP BY cannot contain its timestamp or value column" + ))); + } + if group_ids.contains(&group_id) { + return Err(LoweringError::InvalidExpression(format!( + "{name} GROUP BY contains the same resolved column more than once" + ))); + } + group_ids.push(group_id); + group_refs.push(group_ref); + } + // Minimal series-identity contract without adding SQL-only metadata to + // the shared Schema: a declared row-unique key must contain the time + // index, and removing that index yields the complete series key. The + // GROUP BY must match that key exactly. A unique key that omits time is + // only row identity and proves nothing about time-series continuity. + let identifies_one_series = input_schema + .unique_keys + .iter() + .filter(|key| key.contains(×tamp_id)) + .any(|key| { + let mut series_key: Vec<_> = key + .iter() + .copied() + .filter(|id| *id != timestamp_id) + .collect(); + series_key.sort_unstable(); + series_key.dedup(); + let mut grouped = group_ids.clone(); + grouped.sort_unstable(); + series_key == grouped + }); + if !identifies_one_series { + return Err(LoweringError::InvalidExpression(format!( + "{name} GROUP BY must exactly match a declared series identity (a unique key without the time index)" + ))); + } + let mut cols = vec![ ProjectItem { alias: Some("ts".into()), @@ -838,15 +885,12 @@ impl<'a> SqlLowerer<'a> { expr: Unresolved::Column(value_ref.clone()), }, ]; - for group in &agg.group_expr { - let group_ref = expr_to_group_ref(group)?; + for group_ref in group_refs { let group_name = named_ref(&group_ref).to_string(); - if group_name != named_ref(×tamp_ref) && group_name != named_ref(&value_ref) { - cols.push(ProjectItem { - alias: Some(group_name), - expr: Unresolved::Column(group_ref), - }); - } + cols.push(ProjectItem { + alias: Some(group_name), + expr: Unresolved::Column(group_ref), + }); } let child = Unresolved::Project { cols, @@ -860,7 +904,7 @@ impl<'a> SqlLowerer<'a> { let intent = match name.as_str() { "asap_rate" => AggIntent::Rate, "asap_increase" => AggIntent::Increase, - "asap_last" => AggIntent::LastOverTime, + _ => unreachable!("is_temporal_aggregate admitted {name}"), }; Ok(Unresolved::Aggregate { @@ -1423,7 +1467,7 @@ fn temporal_aggregate_name(expr: &Expr) -> Option { return None; }; let name = call.func.name().to_lowercase(); - matches!(name.as_str(), "asap_rate" | "asap_increase" | "asap_last").then_some(name) + matches!(name.as_str(), "asap_rate" | "asap_increase").then_some(name) } fn is_temporal_aggregate(expr: &Expr) -> bool { @@ -1435,7 +1479,7 @@ fn is_temporal_output_column(expr: &Expr) -> bool { return false; }; let name = col.name.to_lowercase(); - ["asap_rate(", "asap_increase(", "asap_last("] + ["asap_rate(", "asap_increase("] .iter() .any(|prefix| name.starts_with(prefix)) } diff --git a/crates/frontend-sql/tests/sql_lowering.rs b/crates/frontend-sql/tests/sql_lowering.rs index 1df0312d..b91a4a4f 100644 --- a/crates/frontend-sql/tests/sql_lowering.rs +++ b/crates/frontend-sql/tests/sql_lowering.rs @@ -31,7 +31,7 @@ fn catalog() -> SqlCatalog { col("bytes", DataType::Int64), ], 0, - vec![], + vec![vec![0, 1]], ), ) .with_table( @@ -1563,7 +1563,6 @@ async fn explicit_temporal_aggregates_share_promql_intents_and_timerange() { for (function, expected) in [ ("asap_rate", AggIntent::Rate), ("asap_increase", AggIntent::Increase), - ("asap_last", AggIntent::LastOverTime), ] { let sql = format!( "SELECT service, {function}(latency, ts, 300000) AS v \ @@ -1621,6 +1620,118 @@ async fn temporal_aggregate_rejects_mixed_reducers() { assert!(format!("{err}").contains("cannot share an Aggregate node")); } +#[tokio::test] +async fn last_fails_closed_until_an_executable_summary_exists() { + let err = lower_sql_dialect( + "SELECT service, asap_last(latency, ts, 300000) FROM metrics GROUP BY service", + &catalog(), + SqlDialect::ClickhouseSQL, + AccuracyTarget::Exact, + ) + .await + .expect_err("last must not be advertised without an executable physical summary"); + assert!(format!("{err}").contains("Invalid function 'asap_last'")); +} + +#[tokio::test] +async fn temporal_grouping_requires_the_complete_declared_series_identity() { + let multi_series = SqlCatalog::new().with_table( + "samples", + Schema::with_time_index( + vec![ + col("ts", DataType::Timestamp), + col("service", DataType::Utf8), + col("instance", DataType::Utf8), + col("value", DataType::Float64), + ], + 0, + vec![vec![0, 1, 2]], + ), + ); + for sql in [ + "SELECT asap_rate(value, ts, 300000) FROM samples", + "SELECT service, asap_rate(value, ts, 300000) FROM samples GROUP BY service", + ] { + let err = lower_sql_dialect( + sql, + &multi_series, + SqlDialect::ClickhouseSQL, + AccuracyTarget::Exact, + ) + .await + .expect_err("partial identity must not merge counter series"); + assert!(format!("{err}").contains("declared series identity")); + } + + lower_sql_dialect( + "SELECT service, instance, asap_rate(value, ts, 300000) \ + FROM samples GROUP BY service, instance", + &multi_series, + SqlDialect::ClickhouseSQL, + AccuracyTarget::Exact, + ) + .await + .expect("the complete declared series identity is safe"); + + let row_id_only = SqlCatalog::new().with_table( + "samples", + Schema::with_time_index( + vec![ + col("ts", DataType::Timestamp), + col("service", DataType::Utf8), + col("value", DataType::Float64), + ], + 0, + vec![vec![1]], + ), + ); + lower_sql_dialect( + "SELECT service, asap_rate(value, ts, 300000) FROM samples GROUP BY service", + &row_id_only, + SqlDialect::ClickhouseSQL, + AccuracyTarget::Exact, + ) + .await + .expect_err("a row key without time does not prove a series identity"); +} + +#[tokio::test] +async fn temporal_grouping_rejects_value_time_and_duplicate_resolved_columns() { + for sql in [ + "SELECT asap_rate(latency, ts, 300000) FROM metrics GROUP BY ts", + "SELECT asap_rate(latency, ts, 300000) FROM metrics GROUP BY latency", + "SELECT m.service, asap_rate(m.latency, m.ts, 300000) \ + FROM metrics m GROUP BY m.service, service", + ] { + let err = lower_sql_dialect( + sql, + &catalog(), + SqlDialect::ClickhouseSQL, + AccuracyTarget::Exact, + ) + .await + .expect_err("unsafe or duplicate resolved grouping must fail closed"); + let message = format!("{err}"); + assert!( + message.contains("timestamp or value") + || message.contains("same resolved column more than once"), + "unexpected error for {sql}: {message}" + ); + } +} + +#[tokio::test] +async fn qualified_columns_are_validated_by_resolved_identity() { + let qe = lower_clickhouse( + "SELECT m.service, asap_increase(m.latency, m.ts, 300000) AS v \ + FROM metrics AS m GROUP BY m.service", + ) + .await; + let (intent, range, _) = temporal_aggregate(&qe); + assert_eq!(intent, &AggIntent::Increase); + assert_eq!(range, std::time::Duration::from_secs(300)); +} + #[tokio::test] async fn project_filter_and_outer_aggregate_preserve_temporal_child() { let qe = lower_clickhouse( diff --git a/crates/integration-tests/tests/sql_to_post_asap.rs b/crates/integration-tests/tests/sql_to_post_asap.rs index 1c1c2a15..8e7a517c 100644 --- a/crates/integration-tests/tests/sql_to_post_asap.rs +++ b/crates/integration-tests/tests/sql_to_post_asap.rs @@ -84,7 +84,7 @@ fn catalog() -> SqlCatalog { col("bytes", DataType::Int64), ], 0, - vec![], + vec![vec![0, 1]], ), ) } @@ -96,40 +96,52 @@ async fn lower(sql: &str, accuracy: AccuracyTarget) -> QueryExpr { } #[tokio::test] -async fn clickhouse_temporal_sql_reuses_the_rate_physical_summary() { - let pre_asap = lower_sql_dialect( - "SELECT service, asap_rate(latency, ts, 300000) AS v \ - FROM metrics WHERE bytes > 0 GROUP BY service", - &catalog(), - SqlDialect::ClickhouseSQL, - AccuracyTarget::Exact, - ) - .await - .expect("explicit temporal SQL must lower"); - let physical = realize(inner_aggregate(&pre_asap)).expect("rate must be physically planned"); - let SummaryExpr::SummaryAgg { - family, - reduction, - child, - .. - } = &physical.expr - else { - panic!("expected a shared SummaryAgg, got {:?}", physical.expr); - }; - assert_eq!( - family, - &SummaryFamilyType::ExactAggregate(ExactKind::Rate, ExactParams::Rate) - ); - assert_eq!(reduction, &Reduction::PerEntity); - let SummaryExpr::KeepPreAsap(raw) = &child.expr else { - panic!( - "expected a retained temporal SQL input, got {:?}", - child.expr +async fn clickhouse_temporal_sql_reuses_rate_and_increase_physical_summaries() { + for (function, expected) in [ + ( + "asap_rate", + SummaryFamilyType::ExactAggregate(ExactKind::Rate, ExactParams::Rate), + ), + ( + "asap_increase", + SummaryFamilyType::ExactAggregate(ExactKind::Increase, ExactParams::Increase), + ), + ] { + let sql = format!( + "SELECT service, {function}(latency, ts, 300000) AS v \ + FROM metrics WHERE bytes > 0 GROUP BY service" ); - }; - assert!(matches!(raw.as_ref(), QueryExpr::TimeRange { range, child } - if *range == std::time::Duration::from_secs(300) - && matches!(child.as_ref(), QueryExpr::Project { .. }))); + let pre_asap = lower_sql_dialect( + &sql, + &catalog(), + SqlDialect::ClickhouseSQL, + AccuracyTarget::Exact, + ) + .await + .expect("explicit temporal SQL must lower"); + let physical = + realize(inner_aggregate(&pre_asap)).expect("temporal reducer must be planned"); + let SummaryExpr::SummaryAgg { + family, + reduction, + child, + .. + } = &physical.expr + else { + panic!("expected a shared SummaryAgg, got {:?}", physical.expr); + }; + assert_eq!(family, &expected); + assert_eq!(reduction, &Reduction::PerEntity); + let SummaryExpr::KeepPreAsap(raw) = &child.expr else { + panic!( + "expected a retained temporal SQL input, got {:?}", + child.expr + ); + }; + assert!(matches!(raw.as_ref(), QueryExpr::TimeRange { range, child } + if *range == std::time::Duration::from_secs(300) + && matches!(child.as_ref(), QueryExpr::Project { .. }))); + } } /// The `Aggregate` node beneath the identity `Project` DataFusion's planner diff --git a/crates/sql-function-catalog/src/lib.rs b/crates/sql-function-catalog/src/lib.rs index 8cae4308..8991a8a6 100644 --- a/crates/sql-function-catalog/src/lib.rs +++ b/crates/sql-function-catalog/src/lib.rs @@ -283,11 +283,6 @@ pub const CLICKHOUSE_BUILTINS: &[ClickHouseBuiltin] = &[ arity: Arity::Exact(3), rewrite: RewriteKind::PassThrough, }, - ClickHouseBuiltin { - name: "asap_last", - arity: Arity::Exact(3), - rewrite: RewriteKind::PassThrough, - }, ClickHouseBuiltin { name: "uniqexact", arity: Arity::Exact(1),