diff --git a/Cargo.lock b/Cargo.lock index fcc9df5a..b3ada672 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -123,15 +123,6 @@ version = "1.0.102" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" -[[package]] -name = "ar_archive_writer" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7eb93bbb63b9c227414f6eb3a0adfddca591a8ce1e9b60661bb08969b87e340b" -dependencies = [ - "object", -] - [[package]] name = "arc-swap" version = "1.9.1" @@ -867,7 +858,6 @@ dependencies = [ "serde", "serde_json", "serde_yaml", - "sqlparser", "thiserror 1.0.69", "tokio", "tokio-stream", @@ -2385,15 +2375,6 @@ dependencies = [ "objc2-core-foundation", ] -[[package]] -name = "object" -version = "0.37.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ff76201f031d8863c38aa7f905eca4f53abbfa15f609db4277d44cd8938f33fe" -dependencies = [ - "memchr", -] - [[package]] name = "once_cell" version = "1.21.4" @@ -2813,16 +2794,6 @@ version = "3.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "95067976aca6421a523e491fce939a3e65249bac4b977adee0ee9771568e8aa3" -[[package]] -name = "psm" -version = "0.1.31" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "645dbe486e346d9b5de3ef16ede18c26e6c70ad97418f4874b8b1889d6e761ea" -dependencies = [ - "ar_archive_writer", - "cc", -] - [[package]] name = "quick-xml" version = "0.38.4" @@ -2988,26 +2959,6 @@ dependencies = [ "crossbeam-utils", ] -[[package]] -name = "recursive" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0786a43debb760f491b1bc0269fe5e84155353c67482b9e60d0cfb596054b43e" -dependencies = [ - "recursive-proc-macro-impl", - "stacker", -] - -[[package]] -name = "recursive-proc-macro-impl" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "76009fbe0614077fc1a2ce255e3a1881a2e3a3527097d5dc6d8212c585e7e38b" -dependencies = [ - "quote", - "syn 2.0.117", -] - [[package]] name = "redox_syscall" version = "0.5.18" @@ -3566,35 +3517,12 @@ dependencies = [ "vob", ] -[[package]] -name = "sqlparser" -version = "0.61.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dbf5ea8d4d7c808e1af1cbabebca9a2abe603bcefc22294c5b95018d53200cb7" -dependencies = [ - "log", - "recursive", -] - [[package]] name = "stable_deref_trait" version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" -[[package]] -name = "stacker" -version = "0.1.24" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "640c8cdd92b6b12f5bcb1803ca3bbf5ab96e5e6b6b96b9ab77dabe9e880b3190" -dependencies = [ - "cc", - "cfg-if", - "libc", - "psm", - "windows-sys 0.61.2", -] - [[package]] name = "static_assertions" version = "1.1.0" diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 6f448416..24916bd5 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -24,7 +24,6 @@ thiserror = "1" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } chrono = { version = "0.4", features = ["serde"] } -sqlparser = "0.61" promql-parser = "0.8" prost = "0.13" bytes = "1" diff --git a/control_plane/src/physical/planner.rs b/control_plane/src/physical/planner.rs index 6ea5accb..e1b882d5 100644 --- a/control_plane/src/physical/planner.rs +++ b/control_plane/src/physical/planner.rs @@ -8,7 +8,7 @@ //! - [`PhysicalAggOp`] — resolved AggIntent → concrete SketchType + SketchParams //! - [`PhysicalOp`] — a physical operator (sketch build, merge, exchange, eval, etc.) //! - [`PhysicalNode`] — a node in the physical plan tree (operator + placement + cost) -//! - [`Placement`] — where a physical operator runs (Agent, Backend, PromSketch, DB, etc.) +//! - [`Placement`] — where a physical operator runs (Agent, Backend, PromSketch, QueryEngine) //! //! Step γ7: the planner consumes the canonical `query_expr::QueryExpr`. //! The legacy `SketchAgg` / `WindowedAgg` / `TopK` variants are gone — they @@ -119,8 +119,6 @@ pub enum PhysicalOp { TopK { k: u64 }, /// Hash-partitioned aggregation. HashAggregate { keys: Vec }, - /// SQL query to database. - DbQuery { sql: String }, /// Passthrough — no transformation. Passthrough, } @@ -132,8 +130,6 @@ pub enum PhysicalWindow { OtelTumblingFlush { duration: Duration }, /// PromSketch ExponentialHistogram: time-decaying buckets. PromSketchEH { eh_k: usize, time_window: Duration }, - /// Database-side: `GROUP BY time_bucket(interval, ts)`. - SqlTimeBucket { interval: Duration, time_col: String }, /// No windowing (unbounded / landmark). None, } @@ -155,8 +151,6 @@ pub enum ExchangeFormat { Otlp, /// Sketch-specific binary (merged sketch bytes). SketchBinary, - /// Raw samples (for non-sketch path). - RawSamples, } /// Where a physical operator runs. @@ -170,8 +164,6 @@ pub enum Placement { PromSketchStore, /// General query engine (ASAPQuery). QueryEngine, - /// Database (ClickHouse, TimescaleDB, etc.). - Database, } // ── Physical plan tree ────────────────────────────────────────────────────── @@ -297,7 +289,7 @@ fn plan_node(expr: &QueryExpr, config: &PhysicalPlannerConfig) -> PhysicalNode { // WindowedAgg-inner / TopK all into Aggregate, so dispatch on shape: // * single TopK intent, no HAVING → TopK at QueryEngine // * single other intent, no HAVING → sketch build, budget-placed - // * multi-intent or HAVING → exact DbQuery at Database + // * multi-intent or HAVING → exact HashAggregate at QueryEngine QueryExpr::Aggregate { by, aggs, @@ -340,13 +332,14 @@ fn plan_node(expr: &QueryExpr, config: &PhysicalPlannerConfig) -> PhysicalNode { insert_exchange_if_needed(&mut node); return node; } - // Multi-intent / HAVING aggregate → exact at Database. + // Multi-intent / HAVING aggregate → no single sketch can serve + // it; fall back to an exact hash aggregation at the query engine. let child = plan_node(child, config); let mut node = PhysicalNode { - op: PhysicalOp::DbQuery { - sql: format!("GROUP BY {by:?}"), + op: PhysicalOp::HashAggregate { + keys: by.iter().map(|id| format!("{id:?}")).collect(), }, - placement: Placement::Database, + placement: Placement::QueryEngine, cost: PhysicalCost::default(), children: vec![child], }; @@ -511,7 +504,6 @@ fn insert_exchange_if_needed(node: &mut PhysicalNode) { (Placement::BackendCollector, Placement::QueryEngine) => { ExchangeFormat::SketchBinary } - (Placement::AgentCollector, Placement::Database) => ExchangeFormat::RawSamples, _ => ExchangeFormat::Otlp, }; // Wrap the child in an Exchange node @@ -727,8 +719,9 @@ mod tests { } #[test] - fn plan_multi_intent_aggregate_at_database() { - // Multi-intent Aggregate → exact DbQuery at Database. + fn plan_multi_intent_aggregate_at_query_engine() { + // Multi-intent Aggregate → exact HashAggregate at QueryEngine + // (no single sketch serves multiple intents). let expr = QueryExpr::Aggregate { by: vec![], aggs: vec![AggIntent::Sum, AggIntent::Min], @@ -736,8 +729,8 @@ mod tests { child: Box::new(scan("trades")), }; let node = plan(&expr, &default_config()); - assert_eq!(node.placement, Placement::Database); - assert!(matches!(node.op, PhysicalOp::DbQuery { .. })); + assert_eq!(node.placement, Placement::QueryEngine); + assert!(matches!(node.op, PhysicalOp::HashAggregate { .. })); } #[test] diff --git a/control_plane/src/physical/window_fusion.rs b/control_plane/src/physical/window_fusion.rs index e711ccc2..42f2ca0d 100644 --- a/control_plane/src/physical/window_fusion.rs +++ b/control_plane/src/physical/window_fusion.rs @@ -125,12 +125,6 @@ pub fn resolve_window_canonical( eh_k: 50, time_window: size, }, - (WindowKind::Tumbling, Placement::Database) => PhysicalWindow::SqlTimeBucket { - interval: size, - // Canonical `Window` carries no `time_col`; the legacy - // `resolve_window` defaults the same way when it is `None`. - time_col: "ts".into(), - }, // Fallback: tumbling-at-size for any other (kind, placement) // combo — mirrors the legacy `resolve_window` fallback arm. _ => PhysicalWindow::OtelTumblingFlush { duration: size }, diff --git a/control_plane/src/query_parser/mod.rs b/control_plane/src/query_parser/mod.rs index dd26d1d2..b30da48c 100644 --- a/control_plane/src/query_parser/mod.rs +++ b/control_plane/src/query_parser/mod.rs @@ -1,4 +1,4 @@ -//! SP-1 query workload extraction — PromQL and SQL parsers. +//! SP-1 query workload extraction — PromQL parser. //! //! # Entry points //! @@ -16,19 +16,8 @@ //! - `count(*_over_time(…) by (dims))` — cardinality //! - `changes/resets(m{f}[w])` //! - Bare metric selector / binary op → `exact_required` -//! -//! # Supported SQL patterns (doc §SQL Operators) -//! - `COUNT(*)` with/without GROUP BY → frequency / exact -//! - `COUNT(DISTINCT col)` ± GROUP BY → cardinality / Hydra -//! - `AVG/MIN/MAX(col)` ± GROUP BY → quantile / exact extrema -//! - `SUM(col)` → exact -//! - ORDER BY … DESC LIMIT k → heavy-hitter CountSketch -//! - Multiple aggs in one SELECT → all ops collected (Merge) -//! - JOIN … ON key → backend-side Join (sketch-aware push-down: see physical planner) -//! - UNION ALL → Merge (sketch linearity) pub mod promql; -pub mod sql; use std::collections::HashMap; use std::time::Duration; @@ -46,7 +35,7 @@ use crate::types::AggType; /// Produced by [`parse_query`] via [`QueryExpr`] tree walking. #[derive(Debug, Clone)] pub struct ParsedQuery { - /// Metric name (PromQL: from selector; SQL: FROM clause table). + /// Metric name (from the PromQL selector). pub metric_name: String, /// Aggregation types inferred from the query. pub aggregations: Vec, @@ -89,30 +78,24 @@ pub enum QueryHint { // ── Public entry points ─────────────────────────────────────────────────────── -/// Parse a raw query string (PromQL or SQL) into the **legacy Layer-2** +/// Parse a PromQL query string into the **legacy Layer-2** /// [`relational::QueryExpr`](crate::intent_algebra::relational::QueryExpr) IR. /// -/// Both parsers emit Layer-2 relational operators (`Aggregate { AggFunc }`, -/// `Window`, `Filter`, `Join`, …). The Layer-2 → Layer-3 sketch lowering +/// The parser emits Layer-2 relational operators (`Aggregate { AggFunc }`, +/// `Window`, `Filter`, …). The Layer-2 → Layer-3 sketch lowering /// and the conversion to the canonical IR both live inside /// [`intent_algebra::convert_root`](crate::intent_algebra::convert_root) — -/// this function is just the language-dispatch front door. +/// this function is just the parse front door. /// /// Internal to the crate: the only caller is /// [`parse_query_expr_canonical`], which is the public canonical-IR entry. pub(crate) fn parse_query_expr( query: &str, ) -> anyhow::Result { - let q = query.trim(); - let upper = q.to_ascii_uppercase(); - if upper.starts_with("SELECT") || upper.starts_with("WITH") { - sql::parse_sql_expr(q) - } else { - promql::parse_promql_expr(q) - } + promql::parse_promql_expr(query.trim()) } -/// Parse a raw query string (PromQL or SQL) into the **canonical** L3 +/// Parse a PromQL query string into the **canonical** L3 /// [`query_expr::QueryExpr`](crate::intent_algebra::query_expr::QueryExpr) IR. /// /// This is the single public algebra-IR entry point. It parses the query @@ -132,7 +115,7 @@ pub fn parse_query_expr_canonical( Ok(canonical) } -/// Parse a raw query string (PromQL or SQL) into a [`ParsedQuery`]. +/// Parse a PromQL query string into a [`ParsedQuery`]. /// /// This is the backward-compatible entry point for the existing /// [`crate::analyzer::Analyzer`]. Internally it parses via @@ -446,13 +429,7 @@ pub(super) fn debs_hint( mod tests { use super::*; - // Smoke tests for the unified entry point. - - #[test] - fn sql_dispatched_correctly() { - let pq = parse_query("SELECT COUNT(*) FROM hits GROUP BY AdvEngineID").unwrap(); - assert!(pq.aggregations.contains(&AggType::Frequency)); - } + // Smoke tests for the parse entry point. #[test] fn promql_dispatched_correctly() { @@ -584,52 +561,4 @@ mod doc_verify_all { } } - #[test] - fn example6_sql_avg() { - let expr = parse_query_expr_canonical( - "SELECT symbol, AVG(price) FROM trades GROUP BY symbol" - ).unwrap(); - // Canonical fold of `Partition { ["symbol"], SketchAgg { Quantile } }`: - // `Partition { ["symbol"], Aggregate { by: [], [Quantile] } }`. - match &expr { - QueryExpr::Partition { keys, child } => { - assert_eq!(keys.keys(), &["symbol".to_string()]); - match child.as_ref() { - QueryExpr::Aggregate { by, aggs, .. } => { - assert!(by.is_empty()); - assert!(matches!(aggs.as_slice(), [AggIntent::Quantile { .. }])); - } - other => panic!("expected Aggregate under Partition, got {other:?}"), - } - } - other => panic!("expected Partition, got {other:?}"), - } - } - - #[test] - fn example7_sql_tumble() { - let expr = parse_query_expr_canonical( - "SELECT region, COUNT(DISTINCT user_id) AS cnt FROM sessions GROUP BY region, TUMBLE(ts, INTERVAL '5' MINUTE) ORDER BY cnt DESC LIMIT 10" - ).unwrap(); - // Canonical fold of - // `Limit { Sort { Partition { ["region"], WindowedAgg { Cardinality } } } }`. - let QueryExpr::Limit { n: 10, child, .. } = &expr else { - panic!("expected Limit, got {expr:?}") - }; - let QueryExpr::Sort { child: sort_child, .. } = child.as_ref() else { - panic!("expected Sort, got {child:?}") - }; - let QueryExpr::Partition { keys, child: part_child } = sort_child.as_ref() else { - panic!("expected Partition, got {sort_child:?}") - }; - assert_eq!(keys.keys(), &["region".to_string()]); - let QueryExpr::Window { child: win_child, .. } = part_child.as_ref() else { - panic!("expected Window, got {part_child:?}") - }; - assert!(matches!( - win_child.as_ref(), - QueryExpr::Aggregate { aggs, .. } - if matches!(aggs.as_slice(), [AggIntent::Cardinality { .. }]) - )); - } } diff --git a/control_plane/src/query_parser/sql.rs b/control_plane/src/query_parser/sql.rs deleted file mode 100644 index cd6c9eee..00000000 --- a/control_plane/src/query_parser/sql.rs +++ /dev/null @@ -1,1020 +0,0 @@ -//! Layers 1→3 lowering: SQL string → QueryExpr (sketch logical plan). -//! -//! - **Layer 1**: the `sqlparser` crate parses the SQL string into a -//! language-specific AST (`sqlparser::ast::Statement`). -//! - **Layer 2**: the extraction functions (`extract_query_expr`, `extract_select_qe`) -//! interpret SQL semantics (SELECT projection, GROUP BY, WHERE, JOIN, ORDER BY, -//! LIMIT, UNION ALL) and lower them to the sketch algebra. -//! - **Layer 3**: the output is a `QueryExpr` tree with **relational operators only** -//! (`Source`, `Filter`, `Aggregate`, `Join`, `Sort`, `Limit`, `SetOp`). -//! -//! # Key difference from the PromQL parser -//! -//! The SQL parser does **not** emit `SketchAgg` or `WindowedAgg` nodes. It emits -//! generic `Aggregate { func: Avg/Count/CountDistinct/... }` nodes. Sketch assignment -//! happens later: -//! - **Layer 4 (optimizer)**: R5 TopKFusion rewrites `Limit(Sort(Aggregate))` → `TopK`; -//! R9 HydraConversion rewrites multi-key `CountDistinct` → `PerPartition`. -//! - **Layer 5 (physical planner / stage-split)**: `assign_agg_func` maps each `AggFunc` -//! to an `AggIntent` (e.g., `CountDistinct` → `Cardinality`, `Quantile(φ)` → `Quantile`). -//! -//! This means the SQL path goes: relational plan → optimizer rewrites → physical -//! sketch assignment, whereas PromQL goes: sketch plan directly → optimizer → physical. -//! -//! # Algorithm -//! -//! For each FUNCTION edge in the SELECT projection (leaf → root): -//! 1. Collect context: GROUP BY, WHERE, HAVING, JOIN, DISTINCT, UNION ALL -//! 2. Emit the corresponding `QueryExpr` node: -//! - WHERE predicates → `Filter { ScalarExpr }` -//! - GROUP BY + aggs → `Aggregate { keys, aggs }` -//! - ORDER BY + LIMIT → `Sort` + `Limit` (→ `TopK` via optimizer R5) -//! - JOIN … ON → `Join { kind, pred }` -//! - UNION ALL → `SetOp { Union, all: true }` -//! -//! Multiple aggregations in one SELECT each emit their own `AggItem`, -//! collected inside a single `Aggregate` node. - -use std::collections::HashMap; -use std::time::Duration; - -use anyhow::{anyhow, Context}; -use sqlparser::ast::{ - BinaryOperator, DuplicateTreatment, Expr, FunctionArg, FunctionArgExpr, FunctionArgumentList, - FunctionArguments, GroupByExpr, Join, JoinConstraint, JoinOperator, LimitClause, ObjectName, - OrderBy, OrderByExpr, OrderByKind, Query, Select, SelectItem, SetExpr, SetOperator, Statement, - TableFactor, Value, ValueWithSpan, -}; -use sqlparser::dialect::GenericDialect; - -use crate::intent_algebra::relational::{ - AggFunc, AggItem as AlgAggItem, BinaryOpKind, ColumnRef, JoinKind, LiteralValue, ProjectItem, - QueryExpr, ScalarExpr, SetOpKind, SortKey, SourceSpec, -}; - -// ── Public entry point ──────────────────────────────────────────────────────── - -/// Parse a SQL SELECT statement into a [`QueryExpr`] tree. -/// -/// Preserves `Sort`, `Limit`, `Join`, and `SetOp` nodes natively so the -/// [`crate::algebra`] optimizer and allocator can reason about them. -pub fn parse_sql_expr(sql: &str) -> anyhow::Result { - let dialect = GenericDialect {}; - let mut stmts = sqlparser::parser::Parser::parse_sql(&dialect, sql) - .with_context(|| format!("SQL parse error: {sql:?}"))?; - let stmt = stmts - .pop() - .ok_or_else(|| anyhow!("no SQL statement found"))?; - let query = match stmt { - Statement::Query(q) => *q, - other => return Err(anyhow!("expected SELECT, got {:?}", other)), - }; - extract_query_expr(&query) -} - -// ── Query-level dispatch ────────────────────────────────────────────────────── - -fn extract_query_expr(query: &Query) -> anyhow::Result { - let order_by: Vec = match &query.order_by { - Some(OrderBy { - kind: OrderByKind::Expressions(exprs), - .. - }) => exprs.clone(), - _ => vec![], - }; - let (limit_n, offset_n) = match &query.limit_clause { - Some(LimitClause::LimitOffset { - limit: Some(e), - offset, - .. - }) => ( - Some(e.clone()), - offset.as_ref().and_then(|o| expr_to_u64(&o.value)), - ), - Some(LimitClause::OffsetCommaLimit { - limit: e, offset, .. - }) => (Some(e.clone()), Some(expr_to_u64(offset).unwrap_or(0))), - _ => (None, None), - }; - let limit_val = limit_n.as_ref().and_then(|e| expr_to_u64(e)); - let offset_val = offset_n.unwrap_or(0); - - let body = extract_set_expr_qe(query.body.as_ref(), &order_by, limit_val, offset_val)?; - Ok(body) -} - -fn extract_set_expr_qe( - set_expr: &SetExpr, - order_by: &[OrderByExpr], - limit_n: Option, - offset_n: u64, -) -> anyhow::Result { - match set_expr { - SetExpr::Select(sel) => extract_select_qe(sel, order_by, limit_n, offset_n), - SetExpr::Query(inner) => extract_query_expr(inner), - - // UNION / INTERSECT / EXCEPT - SetExpr::SetOperation { - left, - right, - op, - set_quantifier, - } => { - use sqlparser::ast::{SetOperator, SetQuantifier}; - let left_qe = extract_set_expr_qe(left, &[], None, 0)?; - let right_qe = extract_set_expr_qe(right, &[], None, 0)?; - let kind = match op { - SetOperator::Union => SetOpKind::Union, - SetOperator::Intersect => SetOpKind::Intersect, - SetOperator::Except | SetOperator::Minus => SetOpKind::Except, - }; - let all = matches!(set_quantifier, SetQuantifier::All | SetQuantifier::ByName); - Ok(QueryExpr::SetOp { - kind, - all, - left: Box::new(left_qe), - right: Box::new(right_qe), - }) - } - other => Err(anyhow!("unsupported query body: {:?}", other)), - } -} - -// ── SELECT-level extraction ─────────────────────────────────────────────────── - -fn extract_select_qe( - sel: &Select, - order_by: &[OrderByExpr], - limit_n: Option, - offset_n: u64, -) -> anyhow::Result { - let metric_name = extract_table_name(sel)?; - let where_scalar = sel.selection.as_ref().map(sql_expr_to_scalar); - let group_keys = extract_group_by(&sel.group_by); - let having_scalar = sel.having.as_ref().map(sql_expr_to_scalar); - let agg_items = collect_agg_items_qe(&sel.projection); - let join_qe = extract_join_qe(sel); - let window_spec = extract_group_by_window(&sel.group_by); - - let source = QueryExpr::Source(SourceSpec { - name: metric_name.clone(), - }); - - // WHERE → Filter - let after_where = match where_scalar { - Some(pred) => QueryExpr::Filter { - pred, - input: Box::new(source), - }, - None => source, - }; - - // JOIN - let after_join = if let Some((inner_table, join_kind, join_pred)) = join_qe { - let inner_source = QueryExpr::Source(SourceSpec { name: inner_table }); - QueryExpr::Join { - kind: join_kind, - pred: join_pred, - left: Box::new(after_where), - right: Box::new(inner_source), - } - } else { - after_where - }; - - // TUMBLE / HOP → Window node wrapping the source - let after_window = if let Some(ws) = window_spec { - QueryExpr::Window { - duration: ws.size, - slide: ws.slide, - input: Box::new(after_join), - } - } else { - after_join - }; - - // GROUP BY + aggs OR bare projection - let after_agg = if agg_items.is_empty() { - // No aggregation — bare projection with possible DISTINCT. - let cols = collect_project_items(&sel.projection); - QueryExpr::Project { - cols, - input: Box::new(after_window), - } - } else { - let having = having_scalar; - QueryExpr::Aggregate { - keys: group_keys, - aggs: agg_items, - having, - input: Box::new(after_window), - } - }; - - // ORDER BY → Sort - let after_sort = if order_by.is_empty() { - after_agg - } else { - let keys: Vec = order_by - .iter() - .map(|o| SortKey { - col: expr_to_col_name(&o.expr).unwrap_or_else(|| "?".into()), - desc: matches!(o.options.asc, Some(false) | None), - nulls_first: None, - }) - .collect(); - QueryExpr::Sort { - keys, - input: Box::new(after_agg), - } - }; - - // LIMIT / OFFSET - let result = match limit_n { - Some(n) => QueryExpr::Limit { - n, - offset: offset_n, - input: Box::new(after_sort), - }, - None => after_sort, - }; - - Ok(result) -} - -// ── Aggregation item collection ────────────────────────────────────────────── - -fn collect_agg_items_qe(projection: &[SelectItem]) -> Vec { - let mut out = Vec::new(); - for item in projection { - let (expr, alias) = match item { - SelectItem::UnnamedExpr(e) => (e, None), - SelectItem::ExprWithAlias { expr, alias } => (expr, Some(alias.value.clone())), - _ => continue, - }; - collect_agg_from_expr_qe(expr, alias, &mut out); - } - out -} - -fn collect_agg_from_expr_qe(expr: &Expr, alias: Option, out: &mut Vec) { - match expr { - Expr::Function(f) => { - let fn_name = f - .name - .0 - .last() - .and_then(|i| i.as_ident()) - .map(|id| id.value.to_uppercase()) - .unwrap_or_default(); - - let (distinct, args) = match &f.args { - FunctionArguments::List(FunctionArgumentList { - duplicate_treatment, - args, - .. - }) => { - let is_distinct = - matches!(duplicate_treatment, Some(DuplicateTreatment::Distinct)); - (is_distinct, args.as_slice()) - } - _ => (false, &[][..]), - }; - - let col = first_col_from_args(args); - - let func = match fn_name.as_str() { - "COUNT" if distinct => AggFunc::CountDistinct, - "COUNT" => AggFunc::Count, - "SUM" => AggFunc::Sum, - "AVG" => AggFunc::Avg, - "MIN" => AggFunc::Min, - "MAX" => AggFunc::Max, - _ => return, - }; - - out.push(AlgAggItem { - alias: alias.unwrap_or_else(|| fn_name.to_lowercase()), - func, - col, - distinct, - }); - } - Expr::BinaryOp { left, right, .. } => { - collect_agg_from_expr_qe(left, None, out); - collect_agg_from_expr_qe(right, None, out); - } - Expr::Nested(inner) => collect_agg_from_expr_qe(inner, alias, out), - _ => {} - } -} - -fn collect_project_items(projection: &[SelectItem]) -> Vec { - projection - .iter() - .filter_map(|item| match item { - SelectItem::UnnamedExpr(e) => Some(ProjectItem { - alias: None, - expr: sql_expr_to_scalar(e), - }), - SelectItem::ExprWithAlias { expr, alias } => Some(ProjectItem { - alias: Some(alias.value.clone()), - expr: sql_expr_to_scalar(expr), - }), - SelectItem::Wildcard(_) => Some(ProjectItem { - alias: None, - expr: ScalarExpr::Column("*".into()), - }), - _ => None, - }) - .collect() -} - -// ── AST helpers: aggregation arguments ─────────────────────────────────────── - -fn first_col_from_args(args: &[FunctionArg]) -> ColumnRef { - for arg in args { - match arg { - FunctionArg::Unnamed(FunctionArgExpr::Wildcard) => return ColumnRef::Wildcard, - FunctionArg::Unnamed(FunctionArgExpr::Expr(Expr::Identifier(id))) => { - return ColumnRef::Named(id.value.clone()); - } - FunctionArg::Unnamed(FunctionArgExpr::Expr(Expr::CompoundIdentifier(parts))) => { - if let Some(last) = parts.last() { - return ColumnRef::Named(last.value.clone()); - } - } - _ => {} - } - } - ColumnRef::Wildcard -} - -// ── AST helpers: GROUP BY ────────────────────────────────────────────────────── - -/// Window spec extracted from a TUMBLE() or HOP() call in GROUP BY. -struct SqlWindowSpec { - size: Duration, - slide: Option, - time_col: Option, -} - -fn extract_group_by(group_by: &GroupByExpr) -> Vec { - let exprs = match group_by { - GroupByExpr::All(_) => return vec![], - GroupByExpr::Expressions(e, _) => e, - }; - exprs - .iter() - .filter_map(|e| match e { - Expr::Identifier(id) => Some(id.value.clone()), - Expr::CompoundIdentifier(parts) => parts.last().map(|i| i.value.clone()), - // Skip TUMBLE/HOP function calls — extracted separately. - Expr::Function(f) => { - let name = f - .name - .0 - .last() - .and_then(|i| i.as_ident()) - .map(|id| id.value.to_uppercase()) - .unwrap_or_default(); - if name == "TUMBLE" || name == "HOP" || name == "TIME_BUCKET" { - None - } else { - None // unknown function in GROUP BY — skip - } - } - _ => None, - }) - .collect() -} - -/// Extract a TUMBLE / HOP / time_bucket window from the GROUP BY clause. -/// -/// Supported forms: -/// - `TUMBLE(ts, INTERVAL '5' MINUTE)` → Tumbling { size: 5m } -/// - `HOP(ts, INTERVAL '1' MINUTE, INTERVAL '5' MINUTE)` → Sliding { slide: 1m, size: 5m } -/// - `time_bucket('5 minutes', ts)` → Tumbling { size: 5m } -fn extract_group_by_window(group_by: &GroupByExpr) -> Option { - let exprs = match group_by { - GroupByExpr::Expressions(e, _) => e, - _ => return None, - }; - for expr in exprs { - if let Expr::Function(f) = expr { - let name = f - .name - .0 - .last() - .and_then(|i| i.as_ident()) - .map(|id| id.value.to_uppercase()) - .unwrap_or_default(); - - let args = match &f.args { - FunctionArguments::List(FunctionArgumentList { args, .. }) => args, - _ => continue, - }; - - match name.as_str() { - "TUMBLE" if args.len() >= 2 => { - // TUMBLE(ts_col, interval) - let time_col = func_arg_to_col_name(&args[0]); - let size = func_arg_to_duration(&args[1])?; - return Some(SqlWindowSpec { - size, - slide: None, - time_col, - }); - } - "HOP" if args.len() >= 3 => { - // HOP(ts_col, slide_interval, size_interval) - let time_col = func_arg_to_col_name(&args[0]); - let slide = func_arg_to_duration(&args[1])?; - let size = func_arg_to_duration(&args[2])?; - return Some(SqlWindowSpec { - size, - slide: Some(slide), - time_col, - }); - } - "TIME_BUCKET" if args.len() >= 2 => { - // time_bucket('5 minutes', ts_col) — first arg is interval string - let size = func_arg_to_duration(&args[0])?; - let time_col = func_arg_to_col_name(&args[1]); - return Some(SqlWindowSpec { - size, - slide: None, - time_col, - }); - } - _ => {} - } - } - } - None -} - -fn func_arg_to_col_name(arg: &FunctionArg) -> Option { - match arg { - FunctionArg::Unnamed(FunctionArgExpr::Expr(Expr::Identifier(id))) => Some(id.value.clone()), - FunctionArg::Unnamed(FunctionArgExpr::Expr(Expr::CompoundIdentifier(parts))) => { - parts.last().map(|i| i.value.clone()) - } - _ => None, - } -} - -fn func_arg_to_duration(arg: &FunctionArg) -> Option { - match arg { - FunctionArg::Unnamed(FunctionArgExpr::Expr(expr)) => expr_to_duration(expr), - _ => None, - } -} - -fn expr_to_duration(expr: &Expr) -> Option { - match expr { - // INTERVAL '5' MINUTE - Expr::Interval(iv) => { - let val_str = match iv.value.as_ref() { - Expr::Value(vws) => match &vws.value { - Value::SingleQuotedString(s) | Value::DoubleQuotedString(s) => s.clone(), - Value::Number(n, _) => n.clone(), - _ => return None, - }, - _ => return None, - }; - let val: u64 = val_str.trim().parse().ok()?; - let unit = iv.leading_field.as_ref()?; - let secs = match unit { - sqlparser::ast::DateTimeField::Second => val, - sqlparser::ast::DateTimeField::Minute => val * 60, - sqlparser::ast::DateTimeField::Hour => val * 3600, - sqlparser::ast::DateTimeField::Day => val * 86400, - _ => return None, - }; - Some(Duration::from_secs(secs)) - } - // '5 minutes' string (time_bucket style) - Expr::Value(vws) => match &vws.value { - Value::SingleQuotedString(s) | Value::DoubleQuotedString(s) => parse_duration_string(s), - _ => None, - }, - _ => None, - } -} - -fn parse_duration_string(s: &str) -> Option { - let s = s.trim().to_lowercase(); - // Try "Nm", "Ns", "Nmin", "N minutes", "N seconds", "N hours" - let (num_str, unit) = if let Some(n) = s.strip_suffix("minutes") { - (n.trim(), 60u64) - } else if let Some(n) = s.strip_suffix("minute") { - (n.trim(), 60) - } else if let Some(n) = s.strip_suffix("min") { - (n.trim(), 60) - } else if let Some(n) = s.strip_suffix("hours") { - (n.trim(), 3600) - } else if let Some(n) = s.strip_suffix("hour") { - (n.trim(), 3600) - } else if let Some(n) = s.strip_suffix('h') { - (n.trim(), 3600) - } else if let Some(n) = s.strip_suffix("seconds") { - (n.trim(), 1) - } else if let Some(n) = s.strip_suffix("second") { - (n.trim(), 1) - } else if let Some(n) = s.strip_suffix('s') { - (n.trim(), 1) - } else if let Some(n) = s.strip_suffix('m') { - (n.trim(), 60) - } else { - return None; - }; - let n: u64 = num_str.parse().ok()?; - Some(Duration::from_secs(n * unit)) -} - -// ── AST helpers: table name ─────────────────────────────────────────────────── - -fn extract_table_name(sel: &Select) -> anyhow::Result { - sel.from - .first() - .and_then(|t| match &t.relation { - TableFactor::Table { name, .. } => Some(object_name_str(name)), - _ => None, - }) - .ok_or_else(|| anyhow!("could not determine table name from FROM clause")) -} - -fn object_name_str(name: &ObjectName) -> String { - name.0 - .iter() - .map(|i| i.as_ident().map(|id| id.value.as_str()).unwrap_or("")) - .collect::>() - .join(".") -} - -// ── SQL Expr → ScalarExpr ───────────────────────────────────────────────────── - -fn sql_expr_to_scalar(expr: &Expr) -> ScalarExpr { - match expr { - Expr::Identifier(id) => ScalarExpr::Column(id.value.clone()), - Expr::CompoundIdentifier(parts) => ScalarExpr::Column( - parts - .iter() - .map(|i| i.value.as_str()) - .collect::>() - .join("."), - ), - Expr::Value(vws) => sql_value_to_scalar(&vws.value), - Expr::BinaryOp { left, op, right } => { - let lhs = sql_expr_to_scalar(left); - let rhs = sql_expr_to_scalar(right); - let bop = sql_binop_to_algebra(op); - ScalarExpr::BinaryOp { - op: bop, - lhs: Box::new(lhs), - rhs: Box::new(rhs), - } - } - Expr::IsNull(inner) => ScalarExpr::IsNull { - expr: Box::new(sql_expr_to_scalar(inner)), - negated: false, - }, - Expr::IsNotNull(inner) => ScalarExpr::IsNull { - expr: Box::new(sql_expr_to_scalar(inner)), - negated: true, - }, - Expr::Between { - expr, - negated, - low, - high, - } => ScalarExpr::Between { - expr: Box::new(sql_expr_to_scalar(expr)), - low: Box::new(sql_expr_to_scalar(low)), - high: Box::new(sql_expr_to_scalar(high)), - negated: *negated, - }, - Expr::InList { - expr, - list, - negated, - } => ScalarExpr::InList { - expr: Box::new(sql_expr_to_scalar(expr)), - list: list.iter().map(sql_expr_to_scalar).collect(), - negated: *negated, - }, - Expr::Like { - expr, - pattern, - negated, - .. - } => { - let op = if *negated { - BinaryOpKind::NotLike - } else { - BinaryOpKind::Like - }; - ScalarExpr::BinaryOp { - op, - lhs: Box::new(sql_expr_to_scalar(expr)), - rhs: Box::new(sql_expr_to_scalar(pattern)), - } - } - Expr::Nested(inner) => sql_expr_to_scalar(inner), - Expr::Function(f) => { - let name = f - .name - .0 - .last() - .and_then(|i| i.as_ident()) - .map(|id| id.value.clone()) - .unwrap_or_default(); - ScalarExpr::FunctionCall { name, args: vec![] } - } - _ => ScalarExpr::Column("?".into()), // unknown expr → opaque column ref - } -} - -fn sql_value_to_scalar(v: &Value) -> ScalarExpr { - match v { - Value::SingleQuotedString(s) | Value::DoubleQuotedString(s) => { - ScalarExpr::Literal(LiteralValue::Str(s.clone())) - } - Value::Number(n, _) => { - if let Ok(i) = n.parse::() { - ScalarExpr::Literal(LiteralValue::Int(i)) - } else if let Ok(f) = n.parse::() { - ScalarExpr::Literal(LiteralValue::Float(f)) - } else { - ScalarExpr::Literal(LiteralValue::Null) - } - } - Value::Boolean(b) => ScalarExpr::Literal(LiteralValue::Bool(*b)), - Value::Null => ScalarExpr::Literal(LiteralValue::Null), - _ => ScalarExpr::Literal(LiteralValue::Null), - } -} - -fn sql_binop_to_algebra(op: &BinaryOperator) -> BinaryOpKind { - match op { - BinaryOperator::Plus => BinaryOpKind::Add, - BinaryOperator::Minus => BinaryOpKind::Sub, - BinaryOperator::Multiply => BinaryOpKind::Mul, - BinaryOperator::Divide => BinaryOpKind::Div, - BinaryOperator::Modulo => BinaryOpKind::Mod, - BinaryOperator::Eq => BinaryOpKind::Eq, - BinaryOperator::NotEq => BinaryOpKind::Ne, - BinaryOperator::Lt => BinaryOpKind::Lt, - BinaryOperator::LtEq => BinaryOpKind::Le, - BinaryOperator::Gt => BinaryOpKind::Gt, - BinaryOperator::GtEq => BinaryOpKind::Ge, - BinaryOperator::And => BinaryOpKind::And, - BinaryOperator::Or => BinaryOpKind::Or, - BinaryOperator::BitwiseAnd => BinaryOpKind::BitAnd, - BinaryOperator::BitwiseOr => BinaryOpKind::BitOr, - BinaryOperator::BitwiseXor => BinaryOpKind::BitXor, - BinaryOperator::StringConcat => BinaryOpKind::Concat, - _ => BinaryOpKind::Eq, // unknown → eq - } -} - -// ── JOIN → QueryExpr::Join ──────────────────────────────────────────────────── - -fn extract_join_qe(sel: &Select) -> Option<(String, JoinKind, Option)> { - let table_with_joins = sel.from.first()?; - let join = table_with_joins.joins.first()?; - let inner_table = match &join.relation { - TableFactor::Table { name, .. } => object_name_str(name), - _ => return None, - }; - let (kind, pred) = match &join.join_operator { - JoinOperator::Inner(c) => (JoinKind::Inner, join_constraint_to_scalar(c)), - JoinOperator::LeftOuter(c) => (JoinKind::LeftOuter, join_constraint_to_scalar(c)), - JoinOperator::RightOuter(c) => (JoinKind::RightOuter, join_constraint_to_scalar(c)), - JoinOperator::FullOuter(c) => (JoinKind::FullOuter, join_constraint_to_scalar(c)), - JoinOperator::CrossJoin(_) => (JoinKind::Cross, None), - _ => return None, - }; - Some((inner_table, kind, pred)) -} - -fn join_constraint_to_scalar(c: &JoinConstraint) -> Option { - match c { - JoinConstraint::On(e) => Some(sql_expr_to_scalar(e)), - _ => None, - } -} - -// ── Misc helpers ────────────────────────────────────────────────────────────── - -fn expr_to_u64(expr: &Expr) -> Option { - match expr { - Expr::Value(vws) => match &vws.value { - Value::Number(n, _) => n.parse::().ok(), - _ => None, - }, - _ => None, - } -} - -fn expr_to_col_name(expr: &Expr) -> Option { - match expr { - Expr::Identifier(id) => Some(id.value.clone()), - Expr::CompoundIdentifier(parts) => parts.last().map(|i| i.value.clone()), - _ => None, - } -} - -// ── Tests ───────────────────────────────────────────────────────────────────── - -#[cfg(test)] -mod tests { - use super::parse_sql_expr; - use crate::intent_algebra::relational::QueryExpr; - use crate::types::AggType; - - fn parse(sql: &str) -> QueryExpr { - parse_sql_expr(sql).unwrap_or_else(|e| panic!("parse_sql_expr failed: {e}\nSQL: {sql}")) - } - - fn pq(sql: &str) -> super::super::ParsedQuery { - super::super::parse_query(sql) - .unwrap_or_else(|e| panic!("parse_query failed: {e}\nSQL: {sql}")) - } - - // ── Basic aggregations ──────────────────────────────────────────────────── - - #[test] - fn count_star_no_group_is_exact() { - let pq = pq("SELECT COUNT(*) FROM hits"); - assert!(pq.exact_required); - assert!(pq.aggregations.is_empty()); - } - - #[test] - fn count_star_group_by_is_frequency() { - let pq = pq( - "SELECT AdvEngineID, COUNT(*) FROM hits WHERE AdvEngineID <> 0 GROUP BY AdvEngineID", - ); - assert!(pq.aggregations.contains(&AggType::Frequency)); - assert!(pq.group_by_labels.contains(&"AdvEngineID".to_string())); - } - - #[test] - fn count_distinct_is_cardinality() { - let pq = pq("SELECT COUNT(DISTINCT UserID) FROM hits"); - assert!(pq.aggregations.contains(&AggType::Cardinality)); - assert!(!pq.exact_required); - } - - #[test] - fn count_star_order_by_desc_limit_is_topk() { - let pq = pq("SELECT SearchPhrase, COUNT(*) AS c FROM hits \ - WHERE SearchPhrase <> '' GROUP BY SearchPhrase ORDER BY c DESC LIMIT 10"); - assert!(pq.aggregations.contains(&AggType::Frequency)); - } - - #[test] - fn avg_with_group_by_is_quantile_p50() { - let pq = pq("SELECT symbol, AVG(last) FROM hits GROUP BY symbol"); - assert!(pq.aggregations.contains(&AggType::Quantile)); - assert!(pq.quantiles.contains(&0.5)); - } - - #[test] - fn min_max_with_group_by_are_extremes() { - let pq = pq("SELECT symbol, MIN(last), MAX(last) FROM hits GROUP BY symbol"); - assert!(pq.aggregations.contains(&AggType::Quantile)); - assert!(pq.quantiles.contains(&0.0)); - assert!(pq.quantiles.contains(&1.0)); - } - - #[test] - fn min_max_no_group_by_is_exact_minmax() { - let pq = pq("SELECT MIN(EventDate), MAX(EventDate) FROM hits"); - // MIN/MAX map to Quantile in legacy AggType - assert!(pq.aggregations.contains(&AggType::Quantile)); - } - - #[test] - fn sum_is_always_exact() { - let pq = pq("SELECT SUM(AdvEngineID) FROM hits"); - assert!(pq.exact_required); - } - - // ── WHERE predicates ────────────────────────────────────────────────────── - - #[test] - fn where_equality_captured() { - let pq = pq("SELECT COUNT(*) FROM hits WHERE sectype = 'E' GROUP BY symbol"); - assert_eq!( - pq.label_filters.get("sectype").map(String::as_str), - Some("E") - ); - } - - #[test] - fn where_inequality_captured() { - let pq = pq("SELECT COUNT(*) FROM hits WHERE AdvEngineID <> 0 GROUP BY AdvEngineID"); - assert!(pq.aggregations.contains(&AggType::Frequency)); - } - - // ── Multi-aggregation ───────────────────────────────────────────────────── - - #[test] - fn multi_agg_collects_all() { - let pq = pq( - "SELECT RegionID, SUM(AdvEngineID), COUNT(*) AS c, AVG(ResolutionWidth), COUNT(DISTINCT UserID) \ - FROM hits GROUP BY RegionID ORDER BY c DESC LIMIT 10", - ); - assert!( - pq.aggregations.contains(&AggType::Cardinality), - "missing cardinality" - ); - assert!( - pq.aggregations.contains(&AggType::Frequency), - "missing frequency" - ); - assert!( - pq.aggregations.contains(&AggType::Quantile), - "missing quantile" - ); - // SUM adds exact_required alongside sketch ops - assert!(pq.exact_required, "SUM should set exact_required"); - } - - // ── Table / metric name ─────────────────────────────────────────────────── - - #[test] - fn dotted_table_name() { - let pq = pq("SELECT COUNT(*) FROM financial.last_trade_price GROUP BY symbol"); - assert_eq!(pq.metric_name, "financial.last_trade_price"); - } - - // ── DEBS SQL variants ───────────────────────────────────────────────────── - - #[test] - fn debs_q6_cardinality() { - use super::super::QueryHint; - let pq = pq("SELECT COUNT(DISTINCT symbol) FROM financial.last_trade_price"); - assert!(pq.aggregations.contains(&AggType::Cardinality)); - assert!(matches!(pq.hint, Some(QueryHint::DebsCardinality))); - } - - #[test] - fn debs_q3_topk() { - let pq = pq( - "SELECT symbol, COUNT(*) AS c FROM financial.last_trade_price \ - GROUP BY symbol ORDER BY c DESC LIMIT 10", - ); - assert!(pq.aggregations.contains(&AggType::Frequency)); - } - - // ── COUNT(DISTINCT) with GROUP BY ──────────────────────────────────────── - - #[test] - fn count_distinct_with_group_by() { - let pq = pq( - "SELECT RegionID, COUNT(DISTINCT UserID) AS u FROM hits GROUP BY RegionID ORDER BY u DESC LIMIT 10", - ); - assert!(pq.aggregations.contains(&AggType::Cardinality)); - assert!(pq.group_by_labels.contains(&"RegionID".to_string())); - } - - // ── TUMBLE / HOP windows ──────────────────────────────────────────────── - - /// Helper that runs the full canonical pipeline (parse + lower + - /// convert), not just the raw Layer-2 parse. - fn parse_full_canonical(sql: &str) -> crate::intent_algebra::query_expr::QueryExpr { - super::super::parse_query_expr_canonical(sql) - .unwrap_or_else(|e| panic!("parse_query_expr_canonical failed: {e}\nSQL: {sql}")) - } - - /// True if the tree contains the canonical fold of a legacy - /// `WindowedAgg` — a `Window` directly over an `Aggregate`. - fn has_windowed_agg(e: &crate::intent_algebra::query_expr::QueryExpr) -> bool { - use crate::intent_algebra::query_expr::QueryExpr as CQ; - match e { - CQ::Window { child, .. } if matches!(child.as_ref(), CQ::Aggregate { .. }) => true, - CQ::Window { child, .. } - | CQ::Partition { child, .. } - | CQ::Aggregate { child, .. } - | CQ::Filter { child, .. } - | CQ::Sort { child, .. } - | CQ::Limit { child, .. } - | CQ::Distinct { child, .. } - | CQ::Project { child, .. } - | CQ::Subquery { child, .. } => has_windowed_agg(child), - CQ::Merge { children } => children.iter().any(has_windowed_agg), - CQ::Join { left, right, .. } - | CQ::SetOp { left, right, .. } - | CQ::BinaryOp { - lhs: left, - rhs: right, - .. - } => has_windowed_agg(left) || has_windowed_agg(right), - CQ::LetBinding { expr, child, .. } => { - has_windowed_agg(expr) || has_windowed_agg(child) - } - CQ::Scan { .. } | CQ::Ref { .. } => false, - } - } - - #[test] - fn tumble_in_group_by_produces_windowed_agg() { - let expr = parse_full_canonical( - "SELECT symbol, AVG(price) FROM trades \ - GROUP BY symbol, TUMBLE(ts, INTERVAL '5' MINUTE)", - ); - assert!( - has_windowed_agg(&expr), - "expected canonical Window{{Aggregate}} in tree, got {expr:?}" - ); - } - - #[test] - fn hop_in_group_by_produces_windowed_agg() { - let expr = parse_full_canonical( - "SELECT symbol, COUNT(*) FROM trades \ - GROUP BY symbol, HOP(ts, INTERVAL '1' MINUTE, INTERVAL '5' MINUTE)", - ); - assert!( - has_windowed_agg(&expr), - "expected canonical Window{{Aggregate}} in tree, got {expr:?}" - ); - } - - #[test] - fn time_bucket_in_group_by_produces_windowed_agg() { - let expr = parse_full_canonical( - "SELECT symbol, AVG(price) FROM trades \ - GROUP BY symbol, time_bucket('5 minutes', ts)", - ); - assert!( - has_windowed_agg(&expr), - "expected canonical Window{{Aggregate}} in tree, got {expr:?}" - ); - } - - #[test] - fn tumble_layer2_emits_window_node() { - // Layer 2 only (no lowering): should be Aggregate { input: Window { Source } } - let expr = parse( - "SELECT symbol, AVG(price) FROM trades \ - GROUP BY symbol, TUMBLE(ts, INTERVAL '5' MINUTE)", - ); - match &expr { - QueryExpr::Aggregate { input, .. } => { - assert!( - matches!(input.as_ref(), QueryExpr::Window { .. }), - "expected Window inside Aggregate, got {input:?}" - ); - } - other => panic!("expected Aggregate, got {other:?}"), - } - } - - // ── UNION ALL → SetOp ───────────────────────────────────────────────────── - - #[test] - fn union_all_produces_set_op() { - let expr = parse( - "SELECT COUNT(DISTINCT UserID) FROM R \ - UNION ALL \ - SELECT COUNT(DISTINCT UserID) FROM S", - ); - assert!(matches!(expr, QueryExpr::SetOp { .. })); - } - - // ── Complex queries ─────────────────────────────────────────────────────── - - #[test] - fn complex_multi_agg_multi_dim_group_by_topk() { - let pq = pq( - "SELECT region, dc, COUNT(*) AS c, COUNT(DISTINCT UserID), AVG(ResponseTime) \ - FROM hits WHERE env = 'prod' GROUP BY region, dc ORDER BY c DESC LIMIT 5", - ); - assert!(pq.aggregations.contains(&AggType::Frequency)); - assert!(pq.aggregations.contains(&AggType::Cardinality)); - assert!(pq.aggregations.contains(&AggType::Quantile)); - assert!(pq.group_by_labels.contains(&"region".to_string())); - assert!(pq.group_by_labels.contains(&"dc".to_string())); - assert_eq!( - pq.label_filters.get("env").map(String::as_str), - Some("prod") - ); - assert!(pq.quantiles.contains(&0.5)); - } - - #[test] - fn complex_union_all_hll_with_where_on_each_branch() { - let pq = pq( - "SELECT region, COUNT(DISTINCT UserID) FROM sessions WHERE status = 'active' GROUP BY region \ - UNION ALL \ - SELECT region, COUNT(DISTINCT UserID) FROM sessions WHERE status = 'expired' GROUP BY region", - ); - assert!(pq.aggregations.contains(&AggType::Cardinality)); - assert!(pq.group_by_labels.contains(&"region".to_string())); - } -} diff --git a/control_plane/src/types_v2.rs b/control_plane/src/types_v2.rs index c672bd73..f1a80836 100644 --- a/control_plane/src/types_v2.rs +++ b/control_plane/src/types_v2.rs @@ -35,19 +35,13 @@ use serde::{Deserialize, Serialize}; /// Source language the raw query string is written in. Drives which L1 /// parser the control plane dispatches to. /// -/// Today the control plane only consumes `PromQL` and `Sql` (see -/// `query_parser/{promql,sql}.rs`); `ElasticDsl` is reserved for the -/// future ElasticDSL deployment model described in `design.md` §3. +/// The control plane consumes `PromQL` only (see `query_parser/promql.rs`). #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)] #[serde(rename_all = "snake_case")] pub enum QueryLanguage { /// Prometheus query language. Parsed via `promql-parser`. #[serde(rename = "prom_ql", alias = "prom_q_l")] PromQL, - /// SQL. Parsed via `sqlparser`. - Sql, - /// Elasticsearch DSL. Reserved — no L1 parser yet. - ElasticDsl, } // ── AccuracyTarget ──────────────────────────────────────────────────────────── @@ -110,7 +104,7 @@ impl AccuracyTarget { #[serde(tag = "kind", rename_all = "snake_case")] pub enum QueryShape { /// Evaluate once. Plan, execute, return result, discard state. - /// SQL ad-hoc queries; one-off PromQL via `POST /plan`. + /// One-off PromQL via `POST /plan`. #[default] OneShot, /// Continuous query — output stream that the executor keeps emitting @@ -143,8 +137,8 @@ pub enum QueryShape { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] #[serde(rename_all = "snake_case")] pub enum DataShape { - /// Bounded relation, fully materialised at plan time. SQL tables, - /// Parquet / CSV files, in-process columnar tables. + /// Bounded relation, fully materialised at plan time. Parquet / CSV + /// files, in-process columnar tables. Batch, /// Append-only stream — events arrive over time, never updated or /// deleted. Metrics, logs, event streams. The common case for the @@ -270,15 +264,10 @@ mod tests { #[test] fn query_language_serde_roundtrip() { - for variant in [ - QueryLanguage::PromQL, - QueryLanguage::Sql, - QueryLanguage::ElasticDsl, - ] { - let json = serde_json::to_string(&variant).unwrap(); - let back: QueryLanguage = serde_json::from_str(&json).unwrap(); - assert_eq!(variant, back, "round-trip failed for {variant:?}"); - } + let variant = QueryLanguage::PromQL; + let json = serde_json::to_string(&variant).unwrap(); + let back: QueryLanguage = serde_json::from_str(&json).unwrap(); + assert_eq!(variant, back, "round-trip failed for {variant:?}"); } #[test] diff --git a/data_plane/src/drivers/query/adapters/traits.rs b/data_plane/src/drivers/query/adapters/traits.rs index f0887257..24ac3fa6 100644 --- a/data_plane/src/drivers/query/adapters/traits.rs +++ b/data_plane/src/drivers/query/adapters/traits.rs @@ -74,7 +74,7 @@ pub trait QueryRequestAdapter: Send + Sync { form_params: Form>, ) -> Result; - /// Parse a POST request with JSON body (for Elasticsearch) + /// Parse a POST request with a JSON body. /// Default implementation returns an error - adapters that support JSON /// POST requests should override this method. async fn parse_json_post_request( @@ -128,11 +128,7 @@ pub trait QueryResponseAdapter: Send + Sync { async fn format_unsupported_query_response(&self) -> Result; } -/// Adapter trait for HTTP-based query protocols -/// (Prometheus HTTP, ClickHouse HTTP, etc.) -/// -/// For non-HTTP protocols (Flight SQL, native protocols), -/// define separate adapter traits. +/// Adapter trait for HTTP-based query protocols (Prometheus HTTP). /// /// Note: Fallback logic is handled separately via FallbackClient #[async_trait] diff --git a/data_plane/src/drivers/query/fallback/mod.rs b/data_plane/src/drivers/query/fallback/mod.rs index b0fcfee8..bfdcca46 100644 --- a/data_plane/src/drivers/query/fallback/mod.rs +++ b/data_plane/src/drivers/query/fallback/mod.rs @@ -13,7 +13,7 @@ use crate::drivers::query::adapters::ParsedQueryRequest; pub enum FallbackResponse { /// JSON response (used by Prometheus, etc.) Json(Value), - /// Plain text response (used by ClickHouse TSV, etc.) + /// Plain text response. Text(String), } diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index b61368fe..137e17a4 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -1467,7 +1467,7 @@ async fn handle_instant_query_post( let tenant = extract_tenant(&headers); let parsed_request = if content_type.contains("application/json") { - // Handle JSON POST (Elasticsearch) + // Handle JSON POST debug!("Parsing as JSON POST request"); match state.adapter.parse_json_post_request(body).await { Ok(req) => { diff --git a/data_plane/src/query_engines/asap_query_engine/mod.rs b/data_plane/src/query_engines/asap_query_engine/mod.rs index 06891a44..8ec546b9 100644 --- a/data_plane/src/query_engines/asap_query_engine/mod.rs +++ b/data_plane/src/query_engines/asap_query_engine/mod.rs @@ -1,7 +1,7 @@ //! Warm-tier sketch query engine. //! -//! `ASAPQueryEngine` is the long-standing PromQL/SQL/Elasticsearch-DSL -//! query path that answers from the in-memory sketch DB +//! `ASAPQueryEngine` is the PromQL query path that answers from the +//! in-memory sketch DB //! ([`crate::storage_engines::sketch_db::SketchStore`]) and its //! per-`agg_id` precomputed accumulators. It returns ε/δ-bounded //! approximate answers for sketch-resident queries and `None` on