From 0c61e845f1198c5543b1c8009456827068b88a55 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 11 May 2026 15:34:46 -0600 Subject: [PATCH] =?UTF-8?q?refactor(legacy=5Fexpr):=20batch=201=20?= =?UTF-8?q?=E2=80=94=20D-purge=20+=20Dedup=E2=86=92Distinct=20rename?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lowest-risk first batch of the legacy_expr retirement plan (Plan agent's 13-PR sequence). Deletes 5 QueryExpr variants + 5 ScalarExpr variants that have ZERO constructor sites (only match arms), and renames `QueryExpr::Dedup { col }` → `QueryExpr::Distinct { cols: Vec }` with N-arity generalization. ## Deleted variants (zero constructor sites; only match arms) QueryExpr family: - `QueryExpr::JoinSketch { join_key, outer, inner }` — sketch-bound physical alternative; design.md says L3 must be intent-only and the sketch-bound shape belongs at L4 (`sketch_algebra::SketchExpr::SketchJoin`). No constructor sites; 5 consumer match arms removed (query_parser, physical/{stage_split,planner,allocator}, optimizer/engine). - `QueryExpr::Subquery { alias, expr }` — designed for SQL nested SELECT lowering, but no parser emits it. 6 consumer match arms. - `QueryExpr::WindowFunc { func, partition_by, order_by, frame, input }` — designed for SQL OVER() clauses, no parser emits it. 5 arms. - `SketchCoverage` (legacy_expr-internal enum) — referenced only by an in-file test. Deleted along with the test. ScalarExpr family: - `UnaryOp { op, input }` — 0 consumer arms. - `InSubquery { expr, subquery, negated }` — 0 arms. - `Case { operand, when_then, else_ }` — 0 arms. - `Cast { expr, to }` — 0 arms. - `VectorBinaryOp { op, lhs, rhs, vector_match }` — 0 arms (design.md merges PromQL vector ops into `QueryExpr::BinaryOp { vector_match }`). ## Also dropped (orphaned after the D-purge) - `UnaryOpKind` enum (only used by `UnaryOp`). - `WindowFuncKind`, `WindowFrame`, `FrameUnit`, `FrameBound` (only used by `WindowFunc`). - Legacy `DataType` enum (only used by `Cast`). The canonical `intent_algebra::schema::DataType` is untouched. Reversible if Batch 2's canonical lift needs them. ## Dedup → Distinct rename + N-arity `QueryExpr::Dedup { col: ColumnRef }` → `QueryExpr::Distinct { cols: Vec }`. Per design.md §6 "What was removed" row 7: *"Single-column-only spelling of SQL `DISTINCT`. Replacement: renamed to `Distinct { cols }`, generalised to N columns"*. Touched sites: - legacy_expr.rs enum + walk/source_name match arms - query_parser/mod.rs (sketch-mapping arm) - physical/stage_split.rs (walk + PromQL serializer — 2 sites) - physical/planner.rs (new `display_distinct_cols(&[ColumnRef])` helper) - physical/allocator.rs (incl. routing doc table) - optimizer/engine.rs (cost factor, deployment-stage map, HLLDedupElim pattern, recursive rewriter, 2 tests) - intent_algebra/legacy_lower.rs HLLDedupElim rule struct name kept (semantically still describes the rule); only test names/asserts updated. ## Canonical AggIntent verification `controller/src/intent_algebra/agg_intent.rs` already has no `HistogramQuantile` (removed in #133) and no `PromQLSubquery`. No change needed. ## Build + test - `cargo build --release -p controller` — clean - `cargo build --release -p query_engine_rust` — clean - `cargo test -p controller --lib` — **655 passed, 0 failed** (was 656/656; -1 because `sketch_coverage_classification` test went with the deleted `SketchCoverage` enum) - `cargo test -p query_engine_rust --lib -- engines::warm_tier` — 13/13 pass ## Grep verification - `JoinSketch | QueryExpr::Subquery | QueryExpr::WindowFunc | SketchCoverage | ScalarExpr::{UnaryOp,InSubquery,Case,Cast,VectorBinaryOp}` → zero matches across `controller/src/` and `asap-query-engine/src/`. (The 6 remaining `::Subquery\b` hits are `Expr::Subquery(sq)` from the external `promql_parser` / `sqlparser` AST libraries — not our type; pre-existing and out of scope.) - `QueryExpr::Dedup | Dedup {` → zero matches. ## Diff: 7 files, +66 / -323 (net -257 lines) Co-Authored-By: Claude Opus 4.7 (1M context) --- controller/src/intent_algebra/legacy_expr.rs | 177 +----------------- controller/src/intent_algebra/legacy_lower.rs | 20 +- controller/src/optimizer/engine.rs | 51 ++--- controller/src/physical/allocator.rs | 77 +------- controller/src/physical/planner.rs | 29 ++- controller/src/physical/stage_split.rs | 23 +-- controller/src/query_parser/mod.rs | 12 +- 7 files changed, 66 insertions(+), 323 deletions(-) diff --git a/controller/src/intent_algebra/legacy_expr.rs b/controller/src/intent_algebra/legacy_expr.rs index 35eb433a3..16cc4a666 100644 --- a/controller/src/intent_algebra/legacy_expr.rs +++ b/controller/src/intent_algebra/legacy_expr.rs @@ -241,17 +241,6 @@ pub enum FilterVal { Null, } -/// How completely a query can be served by sketches. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum SketchCoverage { - /// All aggregation columns are sketch-mapped. - Full, - /// Some columns are sketch-mapped; others require exact passthrough. - Partial, - /// No sketch applicable; query requires exact execution. - None, -} - // ── Relational algebra ──────────────────────────────────────────────────────── /// Full relational + sketch algebra — the sole query IR. @@ -334,9 +323,11 @@ pub enum QueryExpr { input: Box, }, - /// δ — deduplicate on `col` before sketch ingestion. - Dedup { - col: String, + /// δ — deduplicate on `cols` (a tuple of columns) before sketch ingestion. + /// SQL `SELECT DISTINCT` lowers to this; the column set may be empty + /// (full-row distinct) or multi-column. PromQL has no direct analog. + Distinct { + cols: Vec, input: Box, }, @@ -362,13 +353,6 @@ pub enum QueryExpr { right: Box, }, - /// Sketch-aware join push-down: pre-aggregate on inner side then merge. - JoinSketch { - join_key: String, - outer: Box, - inner: Box, - }, - // ── Set operators ───────────────────────────────────────────────────── /// UNION / INTERSECT / EXCEPT (with or without ALL). @@ -396,12 +380,6 @@ pub enum QueryExpr { // ── Subquery / CTE ──────────────────────────────────────────────────── - /// Inline subquery with an alias (SQL `(SELECT ...) AS alias`). - Subquery { - alias: String, - expr: Box, - }, - /// SQL `WITH name AS (expr) IN body` or PromQL recording rule binding. LetBinding { name: String, @@ -409,17 +387,6 @@ pub enum QueryExpr { body: Box, }, - // ── Window functions (analytic functions) ───────────────────────────── - - /// OVER (PARTITION BY … ORDER BY … frame) analytic functions. - WindowFunc { - func: WindowFuncKind, - partition_by: Vec, - order_by: Vec, - frame: Option, - input: Box, - }, - // ── PromQL-specific operators ───────────────────────────────────────── /// `histogram_quantile(φ, )` — converts an HLL / histogram @@ -467,12 +434,6 @@ pub enum ScalarExpr { rhs: Box, }, - /// Unary prefix operator (`NOT`, `-`, `+`). - UnaryOp { - op: UnaryOpKind, - input: Box, - }, - /// Named function call (e.g. `ABS(x)`, `DATE_TRUNC('hour', ts)`). FunctionCall { name: String, @@ -489,13 +450,6 @@ pub enum ScalarExpr { negated: bool, }, - /// `expr IN (SELECT …)` / `NOT IN (SELECT …)`. - InSubquery { - expr: Box, - subquery: Box, - negated: bool, - }, - /// `expr BETWEEN low AND high` or `NOT BETWEEN …`. Between { expr: Box, @@ -509,28 +463,6 @@ pub enum ScalarExpr { expr: Box, negated: bool, }, - - /// CASE WHEN … THEN … [ELSE …] END. - Case { - operand: Option>, - when_then: Vec<(ScalarExpr, ScalarExpr)>, - else_: Option>, - }, - - /// CAST(expr AS type). - Cast { - expr: Box, - to: DataType, - }, - - /// PromQL vector binary op between two instant-vector expressions where one - /// or both sides produce a scalar in the final result (e.g. `rate(…) > 0.5`). - VectorBinaryOp { - op: BinaryOpKind, - lhs: Box, - rhs: Box, - vector_match: Option, - }, } // ── Supporting enumerations ─────────────────────────────────────────────────── @@ -659,14 +591,6 @@ pub enum BinaryOpKind { Atan2, } -/// Unary prefix operators. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum UnaryOpKind { - Negate, - Not, - BitwiseNot, -} - /// JOIN variant. #[derive(Debug, Clone, PartialEq, Eq)] pub enum JoinKind { @@ -725,48 +649,6 @@ pub struct SortKey { pub nulls_first: Option, } -/// Analytic window function kinds. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum WindowFuncKind { - RowNumber, - Rank, - DenseRank, - PercentRank, - CumeDist, - NTile { n: u64 }, - Lag { offset: u64 }, - Lead { offset: u64 }, - FirstValue, - LastValue, - NthValue { n: u64 }, - /// User-defined analytic function. - Custom(String), -} - -/// ROWS / RANGE frame clause for analytic functions. -#[derive(Debug, Clone)] -pub struct WindowFrame { - pub unit: FrameUnit, - pub start: FrameBound, - pub end: Option, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum FrameUnit { - Rows, - Range, - Groups, -} - -#[derive(Debug, Clone)] -pub enum FrameBound { - UnboundedPreceding, - Preceding(u64), - CurrentRow, - Following(u64), - UnboundedFollowing, -} - /// Scalar literal. #[derive(Debug, Clone, PartialEq)] pub enum LiteralValue { @@ -778,30 +660,6 @@ pub enum LiteralValue { Duration(Duration), } -/// SQL / Arrow data types used in CAST expressions. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum DataType { - Boolean, - Int8, - Int16, - Int32, - Int64, - UInt8, - UInt16, - UInt32, - UInt64, - Float32, - Float64, - Utf8, - Binary, - Timestamp, - Date, - Interval, - List(Box), - Struct(Vec<(String, DataType)>), - Custom(String), -} - impl QueryExpr { /// Walk the expression tree depth-first and call `f` on every node. pub fn walk(&self, f: &mut F) { @@ -814,11 +672,10 @@ impl QueryExpr { | QueryExpr::SketchAgg { input, .. } | QueryExpr::WindowedAgg { input, .. } | QueryExpr::Partition { input, .. } - | QueryExpr::Dedup { input, .. } + | QueryExpr::Distinct { input, .. } | QueryExpr::TopK { input, .. } | QueryExpr::Sort { input, .. } | QueryExpr::Limit { input, .. } - | QueryExpr::WindowFunc { input, .. } | QueryExpr::HistogramQuantile { input, .. } | QueryExpr::PromQLSubquery { input, .. } => input.walk(f), @@ -828,13 +685,11 @@ impl QueryExpr { for i in inputs { i.walk(f); } } QueryExpr::Join { left, right, .. } - | QueryExpr::JoinSketch { outer: left, inner: right, .. } | QueryExpr::SetOp { left, right, .. } | QueryExpr::BinaryOp { lhs: left, rhs: right, .. } => { left.walk(f); right.walk(f); } - QueryExpr::Subquery { expr, .. } => expr.walk(f), QueryExpr::LetBinding { expr, body, .. } => { expr.walk(f); body.walk(f); @@ -864,20 +719,17 @@ impl QueryExpr { | QueryExpr::SketchAgg { input, .. } | QueryExpr::WindowedAgg { input, .. } | QueryExpr::Partition { input, .. } - | QueryExpr::Dedup { input, .. } + | QueryExpr::Distinct { input, .. } | QueryExpr::TopK { input, .. } | QueryExpr::Sort { input, .. } | QueryExpr::Limit { input, .. } | QueryExpr::Aggregate { input, .. } - | QueryExpr::WindowFunc { input, .. } | QueryExpr::HistogramQuantile { input, .. } | QueryExpr::PromQLSubquery { input, .. } => input.source_name(), QueryExpr::Merge { inputs } => inputs.first()?.source_name(), QueryExpr::Join { left, .. } - | QueryExpr::JoinSketch { outer: left, .. } | QueryExpr::SetOp { left, .. } | QueryExpr::BinaryOp { lhs: left, .. } => left.source_name(), - QueryExpr::Subquery { expr, .. } => expr.source_name(), QueryExpr::LetBinding { body, .. } => body.source_name(), QueryExpr::Ref(_) => None, } @@ -1212,19 +1064,4 @@ mod tests { assert!(!AggIntent::default_cardinality().is_exact()); } - #[test] - fn sketch_coverage_classification() { - let ops: Vec = vec![ - AggIntent::default_cardinality(), - AggIntent::Exact(ExactAgg::Sum), - ]; - let has_sketch = ops.iter().any(|o| !o.is_exact()); - let has_exact = ops.iter().any(|o| o.is_exact()); - let cov = match (has_sketch, has_exact) { - (true, false) => SketchCoverage::Full, - (true, true) => SketchCoverage::Partial, - _ => SketchCoverage::None, - }; - assert_eq!(cov, SketchCoverage::Partial); - } } diff --git a/controller/src/intent_algebra/legacy_lower.rs b/controller/src/intent_algebra/legacy_lower.rs index 4c37e4a02..8c92dedd9 100644 --- a/controller/src/intent_algebra/legacy_lower.rs +++ b/controller/src/intent_algebra/legacy_lower.rs @@ -68,8 +68,8 @@ pub fn lower_to_sketch_algebra(expr: QueryExpr) -> QueryExpr { keys, input: Box::new(lower_to_sketch_algebra(*input)), }, - QueryExpr::Dedup { col, input } => QueryExpr::Dedup { - col, + QueryExpr::Distinct { cols, input } => QueryExpr::Distinct { + cols, input: Box::new(lower_to_sketch_algebra(*input)), }, QueryExpr::TopK { k, by, input } => QueryExpr::TopK { @@ -86,13 +86,6 @@ pub fn lower_to_sketch_algebra(expr: QueryExpr) -> QueryExpr { offset, input: Box::new(lower_to_sketch_algebra(*input)), }, - QueryExpr::WindowFunc { func, partition_by, order_by, frame, input } => QueryExpr::WindowFunc { - func, - partition_by, - order_by, - frame, - input: Box::new(lower_to_sketch_algebra(*input)), - }, QueryExpr::HistogramQuantile { phi, input } => QueryExpr::HistogramQuantile { phi, input: Box::new(lower_to_sketch_algebra(*input)), @@ -116,11 +109,6 @@ pub fn lower_to_sketch_algebra(expr: QueryExpr) -> QueryExpr { left: Box::new(lower_to_sketch_algebra(*left)), right: Box::new(lower_to_sketch_algebra(*right)), }, - QueryExpr::JoinSketch { join_key, outer, inner } => QueryExpr::JoinSketch { - join_key, - outer: Box::new(lower_to_sketch_algebra(*outer)), - inner: Box::new(lower_to_sketch_algebra(*inner)), - }, QueryExpr::SetOp { kind, all, left, right } => QueryExpr::SetOp { kind, all, @@ -132,10 +120,6 @@ pub fn lower_to_sketch_algebra(expr: QueryExpr) -> QueryExpr { QueryExpr::Merge { inputs } => QueryExpr::Merge { inputs: inputs.into_iter().map(lower_to_sketch_algebra).collect(), }, - QueryExpr::Subquery { alias, expr } => QueryExpr::Subquery { - alias, - expr: Box::new(lower_to_sketch_algebra(*expr)), - }, QueryExpr::LetBinding { name, expr, body } => QueryExpr::LetBinding { name, expr: Box::new(lower_to_sketch_algebra(*expr)), diff --git a/controller/src/optimizer/engine.rs b/controller/src/optimizer/engine.rs index be4fb8d09..b4c00f74c 100644 --- a/controller/src/optimizer/engine.rs +++ b/controller/src/optimizer/engine.rs @@ -202,7 +202,7 @@ impl CostModel for DefaultCostModel { QueryExpr::Filter { .. } => 0.5, QueryExpr::TopK { k, .. } => (*k as f64).recip().min(0.1), QueryExpr::Partition { .. } => 0.8, // partition adds overhead - QueryExpr::Dedup { .. } => 0.9, + QueryExpr::Distinct { .. } => 0.9, _ => 1.0, }; @@ -226,7 +226,7 @@ impl CostModel for DefaultCostModel { | QueryExpr::Source(_) | QueryExpr::Filter { .. } | QueryExpr::Window { .. } => &dc.agent, QueryExpr::Partition { .. } | QueryExpr::Merge { .. } - | QueryExpr::Dedup { .. } => &dc.backend_collector, + | QueryExpr::Distinct { .. } => &dc.backend_collector, QueryExpr::TopK { .. } | QueryExpr::HistogramQuantile { .. } | QueryExpr::BinaryOp { .. } | QueryExpr::PromQLSubquery { .. } => &dc.backend_db, QueryExpr::Aggregate { .. } => &dc.original_db, @@ -396,7 +396,7 @@ impl RewriteRule for HLLDedupElim { fn try_rewrite(&self, expr: QueryExpr, _model: &dyn CostModel) -> Option { match expr { QueryExpr::SketchAgg { op: AggIntent::Cardinality { accuracy }, col, input } => { - if let QueryExpr::Dedup { input: inner, .. } = *input { + if let QueryExpr::Distinct { input: inner, .. } = *input { return Some(QueryExpr::SketchAgg { op: AggIntent::Cardinality { accuracy }, col, @@ -906,9 +906,9 @@ impl QueryOptimizer { let (new_input, c) = recurse!(input); (QueryExpr::Partition { keys, input: new_input }, c) } - QueryExpr::Dedup { col, input } => { + QueryExpr::Distinct { cols, input } => { let (new_input, c) = recurse!(input); - (QueryExpr::Dedup { col, input: new_input }, c) + (QueryExpr::Distinct { cols, input: new_input }, c) } QueryExpr::TopK { k, by, input } => { let (new_input, c) = recurse!(input); @@ -922,10 +922,6 @@ impl QueryOptimizer { let (new_input, c) = recurse!(input); (QueryExpr::Limit { n, offset, input: new_input }, c) } - QueryExpr::WindowFunc { func, partition_by, order_by, frame, input } => { - let (new_input, c) = recurse!(input); - (QueryExpr::WindowFunc { func, partition_by, order_by, frame, input: new_input }, c) - } QueryExpr::HistogramQuantile { phi, input } => { let (new_input, c) = recurse!(input); (QueryExpr::HistogramQuantile { phi, input: new_input }, c) @@ -946,11 +942,6 @@ impl QueryOptimizer { let (new_right, cr) = recurse!(right); (QueryExpr::Join { kind, pred, left: new_left, right: new_right }, cl || cr) } - QueryExpr::JoinSketch { join_key, outer, inner } => { - let (new_outer, co) = recurse!(outer); - let (new_inner, ci) = recurse!(inner); - (QueryExpr::JoinSketch { join_key, outer: new_outer, inner: new_inner }, co || ci) - } QueryExpr::SetOp { kind, all, left, right } => { let (new_left, cl) = recurse!(left); let (new_right, cr) = recurse!(right); @@ -961,10 +952,6 @@ impl QueryOptimizer { let (new_rhs, cr) = recurse!(rhs); (QueryExpr::BinaryOp { op, lhs: new_lhs, rhs: new_rhs, vector_match }, cl || cr) } - QueryExpr::Subquery { alias, expr } => { - let (new_expr, c) = recurse!(expr); - (QueryExpr::Subquery { alias, expr: new_expr }, c) - } QueryExpr::LetBinding { name, expr, body } => { let (new_expr, ce) = recurse!(expr); let (new_body, cb) = recurse!(body); @@ -1138,16 +1125,16 @@ mod tests { let expr = QueryExpr::SketchAgg { op: AggIntent::default_cardinality(), col: ColumnRef::Named("user_id".into()), - input: Box::new(QueryExpr::Dedup { - col: "user_id".into(), + input: Box::new(QueryExpr::Distinct { + cols: vec![ColumnRef::Named("user_id".into())], input: Box::new(src("events")), }), }; let (result, _) = opt().optimize(expr); assert!( !matches!(&result, QueryExpr::SketchAgg { input, .. } - if matches!(input.as_ref(), QueryExpr::Dedup { .. })), - "Dedup should be eliminated before HLL" + if matches!(input.as_ref(), QueryExpr::Distinct { .. })), + "Distinct should be eliminated before HLL" ); } @@ -1264,9 +1251,9 @@ mod tests { #[test] fn optimizer_chain_of_rewrites() { - // Filter(Window(Dedup(HLL(Source)))) → - // R1: Window(Filter(Dedup(HLL(Source)))) - // R3: Window(Filter(HLL(Source))) (HLL absorbs Dedup) + // Filter(Window(Distinct(HLL(Source)))) → + // R1: Window(Filter(Distinct(HLL(Source)))) + // R3: Window(Filter(HLL(Source))) (HLL absorbs Distinct) let expr = QueryExpr::Filter { pred: ScalarExpr::Literal(LiteralValue::Bool(true)), input: Box::new(QueryExpr::Window { @@ -1275,22 +1262,22 @@ mod tests { input: Box::new(QueryExpr::SketchAgg { op: AggIntent::default_cardinality(), col: ColumnRef::Named("uid".into()), - input: Box::new(QueryExpr::Dedup { - col: "uid".into(), + input: Box::new(QueryExpr::Distinct { + cols: vec![ColumnRef::Named("uid".into())], input: Box::new(src("events")), }), }), }), }; let (result, _iters) = opt().optimize(expr); - // The Dedup should be gone. - let mut dedup_found = false; + // The Distinct should be gone. + let mut distinct_found = false; result.walk(&mut |n| { - if matches!(n, QueryExpr::Dedup { .. }) { - dedup_found = true; + if matches!(n, QueryExpr::Distinct { .. }) { + distinct_found = true; } }); - assert!(!dedup_found, "Dedup should have been eliminated"); + assert!(!distinct_found, "Distinct should have been eliminated"); } // ── R2: MergeLifting ───────────────────────────────────────────────────── diff --git a/controller/src/physical/allocator.rs b/controller/src/physical/allocator.rs index 9649c8aa0..ec93d2024 100644 --- a/controller/src/physical/allocator.rs +++ b/controller/src/physical/allocator.rs @@ -15,17 +15,17 @@ //! //! | Node type | Default stage | Condition | //! |-----------|---------------|-----------| -//! | Source, Filter, Window, Partition, Dedup | Agent | Always | +//! | Source, Filter, Window, Partition, Distinct | Agent | Always | //! | SketchAgg (sketachable op, mergeable) | Agent | budget OK | //! | SketchAgg (sketchable, mergeable) | Backend | agent budget exceeded | //! | SketchAgg (sketchable, not mergeable: Avg) | Db | always | //! | SketchAgg (exact: Sum/Count/Min/Max) | Backend | mergeable | //! | TopK | Precompute | always | -//! | Merge, JoinSketch | Backend | always | +//! | Merge | Backend | always | //! | Aggregate, Project, Sort, Limit | Db | always | -//! | WindowFunc, HistogramQuantile | Db | always | +//! | HistogramQuantile | Db | always | //! | PromQLSubquery, BinaryOp | Precompute | has sketch children | -//! | LetBinding, Subquery | same as body/inner | propagated | +//! | LetBinding | same as body | propagated | use crate::intent_algebra::legacy_expr::QueryExpr; use super::plan::{ @@ -166,18 +166,18 @@ impl SketchAllocator { } } - QueryExpr::Dedup { col, input } => { + QueryExpr::Distinct { cols, input } => { let child = self.alloc_node(*input, budget); PlanNode { - expr: QueryExpr::Dedup { - col, + expr: QueryExpr::Distinct { + cols, input: Box::new(child.expr.clone()), }, stage: PipelineStage::Agent, mode: ExecutionMode::Passthrough, cost: CostEstimate::default(), annotation: NodeAnnotation { - rationale: "Dedup at Agent before sketch build".into(), + rationale: "Distinct at Agent before sketch build".into(), ..Default::default() }, children: vec![child], @@ -246,29 +246,6 @@ impl SketchAllocator { } } - QueryExpr::JoinSketch { join_key, outer, inner } => { - let outer_node = self.alloc_node(*outer, budget); - let inner_node = self.alloc_node(*inner, budget); - PlanNode { - expr: QueryExpr::JoinSketch { - join_key, - outer: Box::new(outer_node.expr.clone()), - inner: Box::new(inner_node.expr.clone()), - }, - stage: PipelineStage::Backend, - mode: ExecutionMode::Passthrough, - cost: CostEstimate { - bytes_per_sec: self.raw_bytes_per_sec * 0.2, - ..Default::default() - }, - annotation: NodeAnnotation { - rationale: "JoinSketch at Backend: pre-agg inner then merge".into(), - ..Default::default() - }, - children: vec![outer_node, inner_node], - } - } - // ── Exact / relational — Db ─────────────────────────────────── QueryExpr::Aggregate { keys, aggs, having, input } => { let child = self.alloc_node(*input, budget); @@ -388,24 +365,6 @@ impl SketchAllocator { } } - QueryExpr::WindowFunc { func, partition_by, order_by, frame, input } => { - let child = self.alloc_node(*input, budget); - PlanNode { - expr: QueryExpr::WindowFunc { - func, partition_by, order_by, frame, - input: Box::new(child.expr.clone()), - }, - stage: PipelineStage::Db, - mode: ExecutionMode::Exact, - cost: CostEstimate::default(), - annotation: NodeAnnotation { - rationale: "Analytic window function at Db".into(), - ..Default::default() - }, - children: vec![child], - } - } - // ── PromQL-specific ─────────────────────────────────────────── QueryExpr::HistogramQuantile { phi, input } => { let child = self.alloc_node(*input, budget); @@ -490,26 +449,6 @@ impl SketchAllocator { } // ── Scoping constructs — propagate body's stage ─────────────── - QueryExpr::Subquery { alias, expr } => { - let child = self.alloc_node(*expr, budget); - let stage = child.stage.clone(); - let mode = child.mode.clone(); - PlanNode { - expr: QueryExpr::Subquery { - alias, - expr: Box::new(child.expr.clone()), - }, - stage, - mode, - cost: CostEstimate::default(), - annotation: NodeAnnotation { - rationale: "Subquery inherits inner stage".into(), - ..Default::default() - }, - children: vec![child], - } - } - QueryExpr::LetBinding { name, expr, body } => { let expr_node = self.alloc_node(*expr, budget); let body_node = self.alloc_node(*body, budget); diff --git a/controller/src/physical/planner.rs b/controller/src/physical/planner.rs index ffb945c08..4795c87ec 100644 --- a/controller/src/physical/planner.rs +++ b/controller/src/physical/planner.rs @@ -343,10 +343,11 @@ fn plan_node(expr: &QueryExpr, config: &PhysicalPlannerConfig) -> PhysicalNode { } } - QueryExpr::Dedup { col, input } => { + QueryExpr::Distinct { cols, input } => { let child = plan_node(input, config); + let pred = format!("distinct({})", display_distinct_cols(cols)); let mut node = PhysicalNode { - op: PhysicalOp::Filter { pred: format!("dedup({col})") }, + op: PhysicalOp::Filter { pred }, placement: Placement::BackendCollector, cost: PhysicalCost::default(), children: vec![child], @@ -423,8 +424,7 @@ fn plan_node(expr: &QueryExpr, config: &PhysicalPlannerConfig) -> PhysicalNode { QueryExpr::Sort { input, .. } | QueryExpr::Limit { input, .. } | QueryExpr::Project { input, .. } - | QueryExpr::Window { input, .. } - | QueryExpr::WindowFunc { input, .. } => { + | QueryExpr::Window { input, .. } => { let child = plan_node(input, config); PhysicalNode { op: PhysicalOp::Passthrough, @@ -436,7 +436,6 @@ fn plan_node(expr: &QueryExpr, config: &PhysicalPlannerConfig) -> PhysicalNode { // ── Join: both children, QueryEngine placement ────────────── QueryExpr::Join { left, right, .. } - | QueryExpr::JoinSketch { outer: left, inner: right, .. } | QueryExpr::SetOp { left, right, .. } => { let l = plan_node(left, config); let r = plan_node(right, config); @@ -448,8 +447,7 @@ fn plan_node(expr: &QueryExpr, config: &PhysicalPlannerConfig) -> PhysicalNode { } } - // ── Subquery / LetBinding ─────────────────────────────────── - QueryExpr::Subquery { expr, .. } => plan_node(expr, config), + // ── LetBinding ────────────────────────────────────────────── QueryExpr::LetBinding { body, .. } => plan_node(body, config), QueryExpr::Ref(_) => PhysicalNode { op: PhysicalOp::Passthrough, @@ -485,6 +483,23 @@ fn decide_sketch_placement(resolved: &PhysicalAggOp, config: &PhysicalPlannerCon } /// If a node's child is at a different stage, insert an Exchange node between them. +/// Render a `Distinct { cols }` column tuple into a human-readable display +/// string for the `Filter { pred }` rationale. `Distinct { cols: [] }` is +/// whole-row SQL DISTINCT and prints as `*`. +fn display_distinct_cols(cols: &[ColumnRef]) -> String { + if cols.is_empty() { + return "*".into(); + } + cols.iter() + .map(|c| match c { + ColumnRef::Named(s) => s.clone(), + ColumnRef::SampleValue => "@value".into(), + ColumnRef::Wildcard => "*".into(), + }) + .collect::>() + .join(", ") +} + fn insert_exchange_if_needed(node: &mut PhysicalNode) { let parent_placement = node.placement.clone(); for child in &mut node.children { diff --git a/controller/src/physical/stage_split.rs b/controller/src/physical/stage_split.rs index 7ae161157..adf7167c4 100644 --- a/controller/src/physical/stage_split.rs +++ b/controller/src/physical/stage_split.rs @@ -147,8 +147,8 @@ fn walk(expr: &QueryExpr, plan: &mut StagedPlan, budgets: &StageResourceBudgets) walk(input, plan, budgets); } - // Dedup — absorbed at Backend (HLL dedup elimination is upstream). - QueryExpr::Dedup { input, .. } => { + // Distinct — absorbed at Backend (HLL dedup elimination is upstream). + QueryExpr::Distinct { input, .. } => { plan.backend.has_dedup = true; walk(input, plan, budgets); } @@ -197,13 +197,6 @@ fn walk(expr: &QueryExpr, plan: &mut StagedPlan, budgets: &StageResourceBudgets) walk(rhs, plan, budgets); } - // JoinSketch — outer and inner both walked; join itself at Backend. - QueryExpr::JoinSketch { outer, inner, .. } => { - plan.backend.has_merge = true; - walk(outer, plan, budgets); - walk(inner, plan, budgets); - } - // Join — Backend. QueryExpr::Join { left, right, .. } => { plan.backend.has_merge = true; @@ -219,9 +212,7 @@ fn walk(expr: &QueryExpr, plan: &mut StagedPlan, budgets: &StageResourceBudgets) } // Transparent / passthrough nodes — recurse into child. - QueryExpr::Project { input, .. } - | QueryExpr::Subquery { expr: input, .. } - | QueryExpr::WindowFunc { input, .. } => walk(input, plan, budgets), + QueryExpr::Project { input, .. } => walk(input, plan, budgets), QueryExpr::LetBinding { expr, body, .. } => { walk(expr, plan, budgets); @@ -486,16 +477,12 @@ fn promql_from_qe(expr: &QueryExpr, ctx: &mut PromQLCtx) -> String { } // ── Passthrough nodes ───────────────────────────────────────────────── - QueryExpr::Dedup { input, .. } - | QueryExpr::Project { input, .. } - | QueryExpr::WindowFunc { input, .. } => promql_from_qe(input, ctx), - - QueryExpr::Subquery { expr, .. } => promql_from_qe(expr, ctx), + QueryExpr::Distinct { input, .. } + | QueryExpr::Project { input, .. } => promql_from_qe(input, ctx), QueryExpr::LetBinding { body, .. } => promql_from_qe(body, ctx), // ── Join / SetOp — serialise the outer / left branch ───────────────── - QueryExpr::JoinSketch { outer, .. } => promql_from_qe(outer, ctx), QueryExpr::Join { left, .. } => promql_from_qe(left, ctx), QueryExpr::SetOp { left, .. } => promql_from_qe(left, ctx), } diff --git a/controller/src/query_parser/mod.rs b/controller/src/query_parser/mod.rs index c2ccbf4cf..94903dd3b 100644 --- a/controller/src/query_parser/mod.rs +++ b/controller/src/query_parser/mod.rs @@ -24,7 +24,7 @@ //! - `SUM(col)` → exact //! - ORDER BY … DESC LIMIT k → heavy-hitter CountSketch //! - Multiple aggs in one SELECT → all ops collected (Merge) -//! - JOIN … ON key → JoinSketch push-down +//! - JOIN … ON key → backend-side Join (sketch-aware push-down: see physical planner) //! - UNION ALL → Merge (sketch linearity) pub mod promql; @@ -188,14 +188,10 @@ impl QeCollector { self.visit(input); self.inside_topk = prev; } - QueryExpr::Dedup { input, .. } => self.visit(input), + QueryExpr::Distinct { input, .. } => self.visit(input), QueryExpr::Merge { inputs } => { for i in inputs { self.visit(i); } } - QueryExpr::JoinSketch { outer, inner, .. } => { - self.visit(outer); - self.visit(inner); - } QueryExpr::Aggregate { keys, aggs, input, .. } => { for k in keys { if !self.group_by_labels.contains(k) { @@ -212,15 +208,13 @@ impl QeCollector { | QueryExpr::Sort { input, .. } | QueryExpr::Limit { input, .. } | QueryExpr::HistogramQuantile { input, .. } - | QueryExpr::PromQLSubquery { input, .. } - | QueryExpr::WindowFunc { input, .. } => self.visit(input), + | QueryExpr::PromQLSubquery { input, .. } => self.visit(input), QueryExpr::Join { left, right, .. } | QueryExpr::SetOp { left, right, .. } | QueryExpr::BinaryOp { lhs: left, rhs: right, .. } => { self.visit(left); self.visit(right); } - QueryExpr::Subquery { expr, .. } => self.visit(expr), QueryExpr::LetBinding { expr, body, .. } => { self.visit(expr); self.visit(body);