From 88a2840fdf5561b04594a09a1e18265060681b09 Mon Sep 17 00:00:00 2001 From: Austin Barrington Date: Wed, 4 Nov 2026 04:02:45 +0000 Subject: [PATCH] Domain uses AST analysis, not the SQL translator timeseriesql::analysis holds the AST facts that hold whatever the backend: select_output_field_name (with default_field_alias / influxql_name), select_has_true_aggregate and the aggregate/window-transform predicates, SelectorKind, MultiSelector, BUCKET_COL and TIME_NS_COL. to_clickhouse re-exports them, so the translator is unchanged; domain::rollup, sharding::query_merge and sharding::multi_selector import the analysis module instead of to_clickhouse. --- hyperbytedb/src/domain/rollup.rs | 2 +- .../src/domain/sharding/multi_selector.rs | 2 +- .../src/domain/sharding/query_merge.rs | 2 +- hyperbytedb/src/timeseriesql/analysis.rs | 167 ++++++++++++++++++ hyperbytedb/src/timeseriesql/mod.rs | 1 + .../timeseriesql/to_clickhouse/aggregates.rs | 45 +---- .../src/timeseriesql/to_clickhouse/select.rs | 51 +----- .../timeseriesql/to_clickhouse/selectors.rs | 26 +-- 8 files changed, 176 insertions(+), 120 deletions(-) create mode 100644 hyperbytedb/src/timeseriesql/analysis.rs diff --git a/hyperbytedb/src/domain/rollup.rs b/hyperbytedb/src/domain/rollup.rs index 52f2c4e..01a4e55 100644 --- a/hyperbytedb/src/domain/rollup.rs +++ b/hyperbytedb/src/domain/rollup.rs @@ -92,7 +92,7 @@ type FieldRollupsResult = Result< /// Build rollup metadata for an MV destination from its SELECT fields. pub fn field_rollups_from_mv_select(fields: &[Field]) -> FieldRollupsResult { - use crate::timeseriesql::to_clickhouse::select_output_field_name; + use crate::timeseriesql::analysis::select_output_field_name; let mut field_types = HashMap::new(); let mut field_rollups = HashMap::new(); diff --git a/hyperbytedb/src/domain/sharding/multi_selector.rs b/hyperbytedb/src/domain/sharding/multi_selector.rs index 6a134d8..5f620fd 100644 --- a/hyperbytedb/src/domain/sharding/multi_selector.rs +++ b/hyperbytedb/src/domain/sharding/multi_selector.rs @@ -17,8 +17,8 @@ use indexmap::IndexMap; use serde_json::Value; use crate::domain::query_result::SeriesResult; +use crate::timeseriesql::analysis::{BUCKET_COL, MultiSelector, SelectorKind, TIME_NS_COL}; use crate::timeseriesql::ast::SelectStatement; -use crate::timeseriesql::to_clickhouse::{BUCKET_COL, MultiSelector, SelectorKind, TIME_NS_COL}; /// Remove the hidden ranking columns. #[must_use] diff --git a/hyperbytedb/src/domain/sharding/query_merge.rs b/hyperbytedb/src/domain/sharding/query_merge.rs index 36f9480..f12318c 100644 --- a/hyperbytedb/src/domain/sharding/query_merge.rs +++ b/hyperbytedb/src/domain/sharding/query_merge.rs @@ -10,11 +10,11 @@ use crate::domain::rollup::{ MeanRollupField, RollupCombine, aggregate_source_field_name, rollup_combine_from_field, }; use crate::error::HyperbytedbError; +use crate::timeseriesql::analysis::{select_has_true_aggregate, select_output_field_name}; use crate::timeseriesql::ast::{ BinaryExpr, BinaryOp, Dimension, DurationUnit, Expr, Field, FillOption, FunctionCall, SelectStatement, }; -use crate::timeseriesql::to_clickhouse::{select_has_true_aggregate, select_output_field_name}; /// Coordinator-side merge refuses unbounded DISTINCT / series fan-in. const SHARD_MERGE_MAX_DISTINCT_SERIES: usize = 500_000; diff --git a/hyperbytedb/src/timeseriesql/analysis.rs b/hyperbytedb/src/timeseriesql/analysis.rs new file mode 100644 index 0000000..49a2c43 --- /dev/null +++ b/hyperbytedb/src/timeseriesql/analysis.rs @@ -0,0 +1,167 @@ +//! Facts about an InfluxQL AST that hold whatever the query backend: output +//! column names, whether a SELECT aggregates, the TOP/BOTTOM/SAMPLE +//! selector, and the merge columns shared by the translator and the +//! sharded merge. The domain uses these; it does not use SQL generation. + +use crate::timeseriesql::ast::{Expr, Field, SelectStatement}; + +/// Each row's group bucket, in nanoseconds (0 without GROUP BY time()). +pub const BUCKET_COL: &str = "__bucket"; + +/// Each row's point time, in nanoseconds. +pub const TIME_NS_COL: &str = "__tns"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SelectorKind { + Top, + Bottom, + Sample, +} + +/// A statement's TOP, BOTTOM or SAMPLE call, parsed. +#[derive(Debug, Clone)] +pub struct MultiSelector { + pub kind: SelectorKind, + /// Output column of the selected values (`top` unless aliased). + pub output: String, + /// The field ranked. + pub field: String, + /// TOP/BOTTOM's tag arguments: at most one point per combination. + pub tags: Vec, + pub n: u64, +} + +/// Output column name for a SELECT field (explicit alias or Influx-style default). +#[must_use] +pub fn select_output_field_name(field: &Field) -> Option { + field + .alias + .clone() + .or_else(|| default_field_alias(&field.expr)) +} + +/// InfluxQL's name for a field without an alias (`Field.Name()` in InfluxDB): +/// a call is named after its function (`mean("v")` → `mean`), an expression +/// after the fields and calls in it joined by `_` (`"a" + "b"` → `a_b`, +/// `mean("v") * 2` → `mean`), a field reference after the field. +/// +/// A bare field reference gets no alias: its column already has that name. +pub fn default_field_alias(expr: &Expr) -> Option { + if matches!(expr, Expr::Identifier(_) | Expr::FieldRef { .. }) { + return None; + } + influxql_name(expr) +} + +pub fn influxql_name(expr: &Expr) -> Option { + let mut names = Vec::new(); + expr_name_parts(expr, &mut names); + (!names.is_empty()).then(|| names.join("_")) +} + +fn expr_name_parts(expr: &Expr, names: &mut Vec) { + match expr { + Expr::Call(func) => names.push(func.name.to_lowercase()), + Expr::Identifier(name) | Expr::FieldRef { name, .. } => names.push(name.clone()), + Expr::BinaryExpr(be) => { + expr_name_parts(&be.left, names); + expr_name_parts(&be.right, names); + } + Expr::UnaryExpr(_, inner) => expr_name_parts(inner, names), + _ => {} + } +} + +/// Whether any SELECT field contains a row-collapsing aggregate (not a bare window transform). +pub fn select_has_true_aggregate(stmt: &SelectStatement) -> bool { + stmt.fields.iter().any(|f| expr_contains_aggregate(&f.expr)) +} + +pub fn is_window_transform_call(name: &str) -> bool { + matches!( + name.to_ascii_uppercase().as_str(), + "DERIVATIVE" + | "NON_NEGATIVE_DERIVATIVE" + | "DIFFERENCE" + | "NON_NEGATIVE_DIFFERENCE" + | "MOVING_AVERAGE" + | "CUMULATIVE_SUM" + | "ELAPSED" + ) +} + +/// Whether an expression contains a row-collapsing aggregate. Window transforms +/// only count when they wrap a nested aggregate (e.g. `difference(mean(v))`). +pub fn expr_contains_aggregate(expr: &Expr) -> bool { + match expr { + Expr::Call(fc) if is_window_transform_call(&fc.name) => { + fc.args.first().is_some_and(|a| matches!(a, Expr::Call(_))) + } + Expr::Call(_) => true, + Expr::BinaryExpr(be) => { + expr_contains_aggregate(&be.left) || expr_contains_aggregate(&be.right) + } + Expr::UnaryExpr(_, e) => expr_contains_aggregate(e), + _ => false, + } +} + +/// Whether an expression contains a window transform applied directly to a raw +/// field (no nested aggregate) — a per-point transform. +pub fn expr_contains_raw_transform(expr: &Expr) -> bool { + match expr { + Expr::Call(fc) if is_window_transform_call(&fc.name) => { + !fc.args.first().is_some_and(|a| matches!(a, Expr::Call(_))) + } + Expr::Call(_) => false, + Expr::BinaryExpr(be) => { + expr_contains_raw_transform(&be.left) || expr_contains_raw_transform(&be.right) + } + Expr::UnaryExpr(_, e) => expr_contains_raw_transform(e), + _ => false, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::timeseriesql::ast::Statement; + + fn select(q: &str) -> crate::timeseriesql::ast::SelectStatement { + match crate::timeseriesql::parse(q).unwrap().remove(0) { + Statement::Select(s) => s, + _ => panic!("not a select"), + } + } + + #[test] + fn output_names_follow_influxql() { + let s = select(r#"SELECT mean("v") * 2, "a" + "b", "v", mean("v") AS m FROM t"#); + assert_eq!( + select_output_field_name(&s.fields[0]).as_deref(), + Some("mean") + ); + assert_eq!( + select_output_field_name(&s.fields[1]).as_deref(), + Some("a_b") + ); + assert_eq!(select_output_field_name(&s.fields[2]), None); + assert_eq!(select_output_field_name(&s.fields[3]).as_deref(), Some("m")); + } + + #[test] + fn window_transforms_are_not_aggregates() { + assert!(!select_has_true_aggregate(&select( + r#"SELECT difference("v") FROM m"# + ))); + assert!(select_has_true_aggregate(&select( + r#"SELECT mean("v") FROM m"# + ))); + } + + #[test] + fn merge_column_names_are_stable() { + assert_eq!(BUCKET_COL, "__bucket"); + assert_eq!(TIME_NS_COL, "__tns"); + } +} diff --git a/hyperbytedb/src/timeseriesql/mod.rs b/hyperbytedb/src/timeseriesql/mod.rs index c9e3c78..804cce0 100644 --- a/hyperbytedb/src/timeseriesql/mod.rs +++ b/hyperbytedb/src/timeseriesql/mod.rs @@ -1,3 +1,4 @@ +pub mod analysis; pub mod ast; pub mod ddl_parser; pub mod digest; diff --git a/hyperbytedb/src/timeseriesql/to_clickhouse/aggregates.rs b/hyperbytedb/src/timeseriesql/to_clickhouse/aggregates.rs index ba92f54..b906615 100644 --- a/hyperbytedb/src/timeseriesql/to_clickhouse/aggregates.rs +++ b/hyperbytedb/src/timeseriesql/to_clickhouse/aggregates.rs @@ -5,51 +5,8 @@ use crate::timeseriesql::ast::*; use super::conditions::{binary_op_to_clickhouse, format_float, quote_phys_identifier}; use super::select::translate_field_expr; +pub(super) use crate::timeseriesql::analysis::expr_contains_raw_transform; -pub(super) fn is_window_transform_call(name: &str) -> bool { - matches!( - name.to_ascii_uppercase().as_str(), - "DERIVATIVE" - | "NON_NEGATIVE_DERIVATIVE" - | "DIFFERENCE" - | "NON_NEGATIVE_DIFFERENCE" - | "MOVING_AVERAGE" - | "CUMULATIVE_SUM" - | "ELAPSED" - ) -} - -/// Whether an expression contains a row-collapsing aggregate. Window transforms -/// only count when they wrap a nested aggregate (e.g. `difference(mean(v))`). -pub(super) fn expr_contains_aggregate(expr: &Expr) -> bool { - match expr { - Expr::Call(fc) if is_window_transform_call(&fc.name) => { - fc.args.first().is_some_and(|a| matches!(a, Expr::Call(_))) - } - Expr::Call(_) => true, - Expr::BinaryExpr(be) => { - expr_contains_aggregate(&be.left) || expr_contains_aggregate(&be.right) - } - Expr::UnaryExpr(_, e) => expr_contains_aggregate(e), - _ => false, - } -} - -/// Whether an expression contains a window transform applied directly to a raw -/// field (no nested aggregate) — a per-point transform. -pub(super) fn expr_contains_raw_transform(expr: &Expr) -> bool { - match expr { - Expr::Call(fc) if is_window_transform_call(&fc.name) => { - !fc.args.first().is_some_and(|a| matches!(a, Expr::Call(_))) - } - Expr::Call(_) => false, - Expr::BinaryExpr(be) => { - expr_contains_raw_transform(&be.left) || expr_contains_raw_transform(&be.right) - } - Expr::UnaryExpr(_, e) => expr_contains_raw_transform(e), - _ => false, - } -} pub(super) fn translate_binary_expr( be: &BinaryExpr, use_fill: bool, diff --git a/hyperbytedb/src/timeseriesql/to_clickhouse/select.rs b/hyperbytedb/src/timeseriesql/to_clickhouse/select.rs index 41124df..b64019f 100644 --- a/hyperbytedb/src/timeseriesql/to_clickhouse/select.rs +++ b/hyperbytedb/src/timeseriesql/to_clickhouse/select.rs @@ -6,14 +6,15 @@ use std::fmt::Write; use super::SeriesJoin; use super::aggregates::{ - expr_contains_aggregate, expr_contains_raw_transform, translate_aggregate_call, - translate_binary_expr, + expr_contains_raw_transform, translate_aggregate_call, translate_binary_expr, }; use super::coalesce::build_coalesced_fact_view_filtered; use super::conditions::{ format_float, nanos_to_ch_timestamp, quote_identifier, quote_phys_identifier, quote_string, translate_expr, }; +pub(super) use crate::timeseriesql::analysis::{default_field_alias, influxql_name}; +pub use crate::timeseriesql::analysis::{select_has_true_aggregate, select_output_field_name}; pub fn translate_native_table( stmt: &SelectStatement, @@ -854,47 +855,6 @@ pub(super) fn translate_field( }) } -/// Output column name for a SELECT field (explicit alias or Influx-style default). -#[must_use] -pub fn select_output_field_name(field: &Field) -> Option { - field - .alias - .clone() - .or_else(|| default_field_alias(&field.expr)) -} - -/// InfluxQL's name for a field without an alias (`Field.Name()` in InfluxDB): -/// a call is named after its function (`mean("v")` → `mean`), an expression -/// after the fields and calls in it joined by `_` (`"a" + "b"` → `a_b`, -/// `mean("v") * 2` → `mean`), a field reference after the field. -/// -/// A bare field reference gets no alias: its column already has that name. -pub(super) fn default_field_alias(expr: &Expr) -> Option { - if matches!(expr, Expr::Identifier(_) | Expr::FieldRef { .. }) { - return None; - } - influxql_name(expr) -} - -fn influxql_name(expr: &Expr) -> Option { - let mut names = Vec::new(); - expr_name_parts(expr, &mut names); - (!names.is_empty()).then(|| names.join("_")) -} - -fn expr_name_parts(expr: &Expr, names: &mut Vec) { - match expr { - Expr::Call(func) => names.push(func.name.to_lowercase()), - Expr::Identifier(name) | Expr::FieldRef { name, .. } => names.push(name.clone()), - Expr::BinaryExpr(be) => { - expr_name_parts(&be.left, names); - expr_name_parts(&be.right, names); - } - Expr::UnaryExpr(_, inner) => expr_name_parts(inner, names), - _ => {} - } -} - /// The pre-InfluxQL default column name: `mean("v")` → `mean_v`. Only for /// materialized views stored before aliases were required, whose columns were /// created under these names. @@ -1053,8 +1013,3 @@ pub(super) fn translate_field_expr( Expr::Now => Ok("now64()".to_string()), } } - -/// Whether any SELECT field contains a row-collapsing aggregate (not a bare window transform). -pub fn select_has_true_aggregate(stmt: &SelectStatement) -> bool { - stmt.fields.iter().any(|f| expr_contains_aggregate(&f.expr)) -} diff --git a/hyperbytedb/src/timeseriesql/to_clickhouse/selectors.rs b/hyperbytedb/src/timeseriesql/to_clickhouse/selectors.rs index 3889c4c..1e2c60c 100644 --- a/hyperbytedb/src/timeseriesql/to_clickhouse/selectors.rs +++ b/hyperbytedb/src/timeseriesql/to_clickhouse/selectors.rs @@ -20,31 +20,7 @@ use super::select::{ build_from_source, select_output_field_name, select_tag_column_sql, time_bucket_expr, translate_field, with_output_names, }; - -/// Each row's group bucket, in nanoseconds (0 without GROUP BY time()). -pub const BUCKET_COL: &str = "__bucket"; -/// Each row's point time, in nanoseconds. -pub const TIME_NS_COL: &str = "__tns"; - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum SelectorKind { - Top, - Bottom, - Sample, -} - -/// A statement's TOP, BOTTOM or SAMPLE call, parsed. -#[derive(Debug, Clone)] -pub struct MultiSelector { - pub kind: SelectorKind, - /// Output column of the selected values (`top` unless aliased). - pub output: String, - /// The field ranked. - pub field: String, - /// TOP/BOTTOM's tag arguments: at most one point per combination. - pub tags: Vec, - pub n: u64, -} +pub use crate::timeseriesql::analysis::{BUCKET_COL, MultiSelector, SelectorKind, TIME_NS_COL}; fn kind_of(name: &str) -> Option { match name.to_ascii_uppercase().as_str() {