diff --git a/control_plane/src/intent_algebra/column_resolution.rs b/control_plane/src/intent_algebra/column_resolution.rs index a96eff0f..881c24cd 100644 --- a/control_plane/src/intent_algebra/column_resolution.rs +++ b/control_plane/src/intent_algebra/column_resolution.rs @@ -174,10 +174,8 @@ pub fn resolve_column_refs( /// [`resolve_column_refs`] but skips the `ColumnRef::Named` wrapping — /// the legacy `Aggregate.keys` field is already a `Vec`. /// -/// Step γ1: used by the legacy→canonical `Aggregate` bridge -/// ([`crate::intent_algebra::aggregate_bridge::bridge_aggregate_to_canonical`]) -/// to translate the legacy `keys: Vec` into the canonical -/// `by: Vec` form. +/// Used by `legacy_to_canonical::convert` to translate a legacy +/// `Aggregate.keys: Vec` into the canonical `by: Vec`. pub fn resolve_named_keys( keys: &[String], schema: &Schema, diff --git a/control_plane/src/intent_algebra/legacy_lower.rs b/control_plane/src/intent_algebra/legacy_lower.rs deleted file mode 100644 index 35235ff3..00000000 --- a/control_plane/src/intent_algebra/legacy_lower.rs +++ /dev/null @@ -1,556 +0,0 @@ -//! Layer 2→3 lowering: convert relational Aggregate nodes to sketch AggIntent. -//! -//! This pass walks a [`QueryExpr`] tree and converts `Aggregate { AggFunc }` -//! nodes to `SketchAgg { AggIntent }` where the aggregation can benefit from -//! sketch-based execution. It is shared by both the PromQL and SQL parsers. -//! -//! # What gets lowered -//! -//! A single-agg `Aggregate` node whose `AggFunc` maps to a sketch intent is -//! replaced by `SketchAgg { AggIntent }`. If the `Aggregate` had GROUP BY -//! keys, they become a wrapping `Partition` node. -//! -//! Multi-agg `Aggregate` nodes or those with HAVING clauses pass through -//! unchanged — the physical planner handles them. - -use crate::intent_algebra::legacy_expr::*; -use crate::intent_algebra::{infer_schema_for_root, Schema}; - -/// Lower a Layer 2 `QueryExpr` (relational operators only) to Layer 3 -/// (sketch algebra with `AggIntent`). -/// -/// This pass walks the tree and converts `Aggregate { AggFunc }` nodes -/// to `SketchAgg { AggIntent }` where the aggregation can benefit from -/// sketch-based execution. -/// -/// Step β: derives the root-level [`Schema`] from the outermost `Source` -/// leaf (via [`infer_schema_for_root`]) and threads it through the -/// recursive descent. The lowering pass is structurally schema-agnostic -/// today (it preserves `ColumnRef::Named(_)` verbatim); Step γ will -/// migrate the per-variant column-list rewrites onto positional -/// `ColumnId` once the canonical aggregator surface lands consumer-side. -pub fn lower_to_sketch_algebra(expr: QueryExpr) -> QueryExpr { - let schema = infer_schema_for_root(&expr); - lower_to_sketch_algebra_with_schema(expr, &schema) -} - -/// Variant of [`lower_to_sketch_algebra`] that takes an explicit -/// inherited [`Schema`]. The public entry point derives it from the -/// outermost source; callers that already hold a Schema (the optimiser -/// pipeline, for instance) can pass it directly. -pub fn lower_to_sketch_algebra_with_schema( - expr: QueryExpr, - parent_schema: &Schema, -) -> QueryExpr { - match expr { - QueryExpr::Aggregate { keys, aggs, having, input } => { - let input = lower_to_sketch_algebra_with_schema(*input, parent_schema); - lower_aggregate(keys, aggs, having, Box::new(input)) - } - - // ── Single-input nodes: recurse ────────────────────────────────── - QueryExpr::Filter { pred, input } => QueryExpr::Filter { - pred, - input: Box::new(lower_to_sketch_algebra_with_schema(*input, parent_schema)), - }, - QueryExpr::Project { cols, input } => QueryExpr::Project { - cols, - input: Box::new(lower_to_sketch_algebra_with_schema(*input, parent_schema)), - }, - QueryExpr::Window { duration, slide, input } => { - let lowered_input = lower_to_sketch_algebra_with_schema(*input, parent_schema); - // Fuse Window + SketchAgg → WindowedAgg (the window defines sketch lifecycle). - if let QueryExpr::SketchAgg { op, col, input: sketch_input } = lowered_input { - let window = WindowSpec { - kind: match slide { - Some(s) => WindowKind::Sliding { size: duration, slide: s }, - None => WindowKind::Tumbling { size: duration }, - }, - time_col: None, - }; - QueryExpr::WindowedAgg { agg: op, window, col, input: sketch_input } - } else { - QueryExpr::Window { duration, slide, input: Box::new(lowered_input) } - } - } - QueryExpr::SketchAgg { op, col, input } => QueryExpr::SketchAgg { - op, - col, - input: Box::new(lower_to_sketch_algebra_with_schema(*input, parent_schema)), - }, - QueryExpr::WindowedAgg { agg, window, col, input } => QueryExpr::WindowedAgg { - agg, - window, - col, - input: Box::new(lower_to_sketch_algebra_with_schema(*input, parent_schema)), - }, - QueryExpr::Partition { keys, input } => QueryExpr::Partition { - keys, - input: Box::new(lower_to_sketch_algebra_with_schema(*input, parent_schema)), - }, - QueryExpr::Distinct { cols, input } => QueryExpr::Distinct { - cols, - input: Box::new(lower_to_sketch_algebra_with_schema(*input, parent_schema)), - }, - QueryExpr::TopK { k, by, input } => QueryExpr::TopK { - k, - by, - input: Box::new(lower_to_sketch_algebra_with_schema(*input, parent_schema)), - }, - QueryExpr::Sort { keys, input } => QueryExpr::Sort { - keys, - input: Box::new(lower_to_sketch_algebra_with_schema(*input, parent_schema)), - }, - QueryExpr::Limit { n, offset, input } => QueryExpr::Limit { - n, - offset, - input: Box::new(lower_to_sketch_algebra_with_schema(*input, parent_schema)), - }, - QueryExpr::PromQLSubquery { range, resolution, input } => QueryExpr::PromQLSubquery { - range, - resolution, - input: Box::new(lower_to_sketch_algebra_with_schema(*input, parent_schema)), - }, - - // ── Two-input nodes: recurse into both ────────────────────────── - QueryExpr::BinaryOp { op, lhs, rhs, vector_match } => QueryExpr::BinaryOp { - op, - lhs: Box::new(lower_to_sketch_algebra_with_schema(*lhs, parent_schema)), - rhs: Box::new(lower_to_sketch_algebra_with_schema(*rhs, parent_schema)), - vector_match, - }, - QueryExpr::Join { kind, pred, left, right } => QueryExpr::Join { - kind, - pred, - left: Box::new(lower_to_sketch_algebra_with_schema(*left, parent_schema)), - right: Box::new(lower_to_sketch_algebra_with_schema(*right, parent_schema)), - }, - QueryExpr::SetOp { kind, all, left, right } => QueryExpr::SetOp { - kind, - all, - left: Box::new(lower_to_sketch_algebra_with_schema(*left, parent_schema)), - right: Box::new(lower_to_sketch_algebra_with_schema(*right, parent_schema)), - }, - - // ── Multi-input / container nodes ─────��────────────────────────── - QueryExpr::Merge { inputs } => QueryExpr::Merge { - inputs: inputs - .into_iter() - .map(|i| lower_to_sketch_algebra_with_schema(i, parent_schema)) - .collect(), - }, - QueryExpr::LetBinding { name, expr, body } => QueryExpr::LetBinding { - name, - expr: Box::new(lower_to_sketch_algebra_with_schema(*expr, parent_schema)), - body: Box::new(lower_to_sketch_algebra_with_schema(*body, parent_schema)), - }, - - // ── Leaf nodes: pass through ────────���──────────────────────────── - QueryExpr::Source(_) | QueryExpr::Ref(_) => expr, - } -} - -/// Try to lower a single `Aggregate` node to `SketchAgg`. -/// -/// # Fan-out -/// -/// Step α replaced the legacy `AggIntent` with the canonical single-φ form -/// and dropped the `Extrema { min, max }` enum (split into `Min` / `Max`). -/// The two multi-intent legacy `AggFunc`s — `StdDev` and `Variance`, which -/// historically lowered to a `Quantile { quantiles: vec![0.25, 0.75] }` — -/// now fan out into two sibling `SketchAgg` (or `WindowedAgg`) nodes -/// wrapped in a `QueryExpr::Merge`. The F1 strategy lets every consumer -/// keep matching single-intent `SketchAgg::op` patterns unchanged. -fn lower_aggregate( - keys: Vec, - aggs: Vec, - having: Option, - input: Box, -) -> QueryExpr { - // Only lower single-agg Aggregates without HAVING. - if aggs.len() == 1 && having.is_none() { - let agg = &aggs[0]; - - // COUNT(*) without GROUP BY is a simple row count — no sketch benefit. - if matches!(agg.func, AggFunc::Count) && keys.is_empty() { - return QueryExpr::Aggregate { keys, aggs, having, input }; - } - - let intents = agg_func_to_intents(&agg.func); - if !intents.is_empty() { - // Build one SketchAgg / WindowedAgg per fanned-out intent. The - // input subtree is cloned for each sibling (Merge children own - // their own input) so the structure mirrors what a sketch - // physical planner would emit for a multi-intent aggregate. - let col = agg.col.clone(); - let sketch_nodes: Vec = match *input { - QueryExpr::Window { duration, slide, input: ref win_input } => { - let window = WindowSpec { - kind: match slide { - Some(s) => WindowKind::Sliding { size: duration, slide: s }, - None => WindowKind::Tumbling { size: duration }, - }, - time_col: None, - }; - intents.into_iter().map(|intent| QueryExpr::WindowedAgg { - agg: intent, - window: window.clone(), - col: col.clone(), - input: win_input.clone(), - }).collect() - } - ref other => { - let inp_boxed: Box = Box::new(other.clone()); - intents.into_iter().map(|intent| QueryExpr::SketchAgg { - op: intent, - col: col.clone(), - input: inp_boxed.clone(), - }).collect() - } - }; - let sketch = if sketch_nodes.len() == 1 { - sketch_nodes.into_iter().next().unwrap() - } else { - QueryExpr::Merge { inputs: sketch_nodes } - }; - - if keys.is_empty() { - return sketch; - } else { - return QueryExpr::Partition { - keys: PartitionKeys::By(keys), - input: Box::new(sketch), - }; - } - } - } - - // Multi-agg or non-sketchable: keep as relational Aggregate. - QueryExpr::Aggregate { keys, aggs, having, input } -} - -/// Map an [`AggFunc`] to the canonical [`AggIntent`]s needed for sketch -/// execution. Returns an empty vec for non-sketchable functions -/// (`Custom`), one intent for single-statistic functions, and N intents -/// for the StdDev / Variance fan-out (Step α F1 strategy: callers wrap -/// the resulting list in a `QueryExpr::Merge` of sibling SketchAggs). -/// -/// Step γ1 exposed this helper publicly so the legacy→canonical -/// `Aggregate` bridge (`aggregate_bridge::legacy_aggregate_to_canonical`) -/// can reuse the same `AggFunc` → `AggIntent` translation. The fan-out -/// semantics (StdDev / Variance → two siblings) are wrapped one layer -/// up (sketch lowering) before the bridge is called, so each `AggItem` -/// at the bridge level produces a single intent vector that's mostly -/// length 1. -pub(crate) fn agg_func_to_intents(func: &AggFunc) -> Vec { - use crate::intent_algebra::legacy_expr::{ - default_cardinality, default_frequency, default_quantile, - }; - use crate::types_v2::AccuracyTarget; - match func { - AggFunc::Quantile(phi) => vec![default_quantile(*phi)], - AggFunc::CountDistinct => vec![default_cardinality()], - AggFunc::HeavyHitters { .. } => vec![default_frequency()], - AggFunc::Count => vec![default_frequency()], - AggFunc::Avg => vec![AggIntent::Quantile { - q: 0.5, - accuracy: AccuracyTarget::Epsilon(0.01), - }], - AggFunc::Min => vec![AggIntent::Min], - AggFunc::Max => vec![AggIntent::Max], - // StdDev / Variance: legacy carried two quantiles in a single - // Quantile intent; Step α F1 fans them out into two siblings. - AggFunc::StdDev { .. } | AggFunc::Variance { .. } => vec![ - AggIntent::Quantile { q: 0.25, accuracy: AccuracyTarget::Epsilon(0.01) }, - AggIntent::Quantile { q: 0.75, accuracy: AccuracyTarget::Epsilon(0.01) }, - ], - AggFunc::Sum | AggFunc::Rate | AggFunc::Increase | AggFunc::Delta => { - vec![AggIntent::Sum] - } - AggFunc::Custom(_) => vec![], - } -} - -// ── Tests ───────────���───────────────────────────��───────────────────────────── - -#[cfg(test)] -mod tests { - use super::*; - use std::time::Duration; - - fn src(name: &str) -> QueryExpr { - QueryExpr::Source(SourceSpec { name: name.into() }) - } - - fn make_agg(func: AggFunc, input: QueryExpr) -> QueryExpr { - QueryExpr::Aggregate { - keys: vec![], - aggs: vec![AggItem { - alias: "v".into(), - func, - col: ColumnRef::SampleValue, - distinct: false, - }], - having: None, - input: Box::new(input), - } - } - - fn make_agg_with_keys(func: AggFunc, keys: Vec, input: QueryExpr) -> QueryExpr { - QueryExpr::Aggregate { - keys, - aggs: vec![AggItem { - alias: "v".into(), - func, - col: ColumnRef::SampleValue, - distinct: false, - }], - having: None, - input: Box::new(input), - } - } - - // ── Single-agg lowering ────────────────────────────────────────────────── - - #[test] - fn quantile_lowered_to_sketch_agg() { - let expr = make_agg(AggFunc::Quantile(0.99), src("m")); - let lowered = lower_to_sketch_algebra(expr); - assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Quantile { .. }, .. })); - } - - #[test] - fn count_distinct_lowered_to_cardinality() { - let expr = make_agg(AggFunc::CountDistinct, src("m")); - let lowered = lower_to_sketch_algebra(expr); - assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Cardinality { .. }, .. })); - } - - #[test] - fn count_without_group_by_stays_aggregate() { - // COUNT(*) without GROUP BY is a simple row count — no sketch benefit. - let expr = make_agg(AggFunc::Count, src("m")); - let lowered = lower_to_sketch_algebra(expr); - assert!(matches!(lowered, QueryExpr::Aggregate { .. })); - } - - #[test] - fn count_with_group_by_lowered_to_frequency() { - let expr = make_agg_with_keys(AggFunc::Count, vec!["region".into()], src("m")); - let lowered = lower_to_sketch_algebra(expr); - match &lowered { - QueryExpr::Partition { input, .. } => { - assert!(matches!(input.as_ref(), QueryExpr::SketchAgg { op: AggIntent::Frequency { .. }, .. })); - } - other => panic!("expected Partition(SketchAgg), got {other:?}"), - } - } - - #[test] - fn sum_lowered_to_sum() { - let expr = make_agg(AggFunc::Sum, src("m")); - let lowered = lower_to_sketch_algebra(expr); - assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Sum, .. })); - } - - #[test] - fn avg_lowered_to_quantile_p50() { - let expr = make_agg(AggFunc::Avg, src("m")); - let lowered = lower_to_sketch_algebra(expr); - match &lowered { - QueryExpr::SketchAgg { op: AggIntent::Quantile { q, .. }, .. } => { - assert!((*q - 0.5).abs() < 1e-9); - } - other => panic!("expected SketchAgg(Quantile), got {other:?}"), - } - } - - #[test] - fn min_lowered_to_min() { - let expr = make_agg(AggFunc::Min, src("m")); - let lowered = lower_to_sketch_algebra(expr); - assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Min, .. })); - } - - #[test] - fn max_lowered_to_max() { - let expr = make_agg(AggFunc::Max, src("m")); - let lowered = lower_to_sketch_algebra(expr); - assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Max, .. })); - } - - #[test] - fn stddev_fans_out_to_merge_of_quantile_siblings() { - // StdDev / Variance historically carried two quantiles in a - // single legacy intent; Step α's F1 fan-out emits a Merge of two - // single-φ SketchAgg siblings. - let expr = make_agg(AggFunc::StdDev { population: false }, src("m")); - let lowered = lower_to_sketch_algebra(expr); - match &lowered { - QueryExpr::Merge { inputs } => { - assert_eq!(inputs.len(), 2); - let mut qs: Vec = inputs.iter().filter_map(|node| match node { - QueryExpr::SketchAgg { op: AggIntent::Quantile { q, .. }, .. } => Some(*q), - _ => None, - }).collect(); - qs.sort_by(|a, b| a.partial_cmp(b).unwrap()); - assert_eq!(qs, vec![0.25, 0.75]); - } - other => panic!("expected Merge of two SketchAgg, got {other:?}"), - } - } - - #[test] - fn custom_func_not_lowered() { - let expr = make_agg(AggFunc::Custom("my_udf".into()), src("m")); - let lowered = lower_to_sketch_algebra(expr); - assert!(matches!(lowered, QueryExpr::Aggregate { .. })); - } - - // ── GROUP BY keys become Partition ────��─────────────────────────────────── - - #[test] - fn group_by_wraps_with_partition() { - let expr = make_agg_with_keys( - AggFunc::Quantile(0.5), - vec!["host".into()], - src("m"), - ); - let lowered = lower_to_sketch_algebra(expr); - match &lowered { - QueryExpr::Partition { keys, input } => { - assert_eq!(keys.keys(), &["host".to_string()]); - assert!(matches!(input.as_ref(), QueryExpr::SketchAgg { .. })); - } - other => panic!("expected Partition, got {other:?}"), - } - } - - // ── Multi-agg not lowered ──────────────────────────────────────────────── - - #[test] - fn multi_agg_not_lowered() { - let expr = QueryExpr::Aggregate { - keys: vec![], - aggs: vec![ - AggItem { - alias: "c".into(), - func: AggFunc::Count, - col: ColumnRef::Wildcard, - distinct: false, - }, - AggItem { - alias: "s".into(), - func: AggFunc::Sum, - col: ColumnRef::Named("x".into()), - distinct: false, - }, - ], - having: None, - input: Box::new(src("m")), - }; - let lowered = lower_to_sketch_algebra(expr); - assert!(matches!(lowered, QueryExpr::Aggregate { .. })); - } - - // ── Recursive lowering ───────────────────��────────────────────���────────── - - #[test] - fn window_wrapping_aggregate_fuses_to_windowed_agg() { - // Aggregate inside a Window → fused WindowedAgg. - let expr = QueryExpr::Window { - duration: Duration::from_secs(300), - slide: None, - input: Box::new(make_agg(AggFunc::Quantile(0.99), src("m"))), - }; - let lowered = lower_to_sketch_algebra(expr); - match &lowered { - QueryExpr::WindowedAgg { agg, window, .. } => { - assert!(matches!(agg, AggIntent::Quantile { .. })); - assert!(matches!(window.kind, WindowKind::Tumbling { .. })); - } - other => panic!("expected WindowedAgg, got {other:?}"), - } - } - - #[test] - fn sliding_window_fuses_to_windowed_agg_sliding() { - let expr = QueryExpr::Window { - duration: Duration::from_secs(300), - slide: Some(Duration::from_secs(60)), - input: Box::new(make_agg(AggFunc::Quantile(0.5), src("m"))), - }; - let lowered = lower_to_sketch_algebra(expr); - match &lowered { - QueryExpr::WindowedAgg { window, .. } => { - assert!(matches!(window.kind, WindowKind::Sliding { .. })); - } - other => panic!("expected WindowedAgg(Sliding), got {other:?}"), - } - } - - #[test] - fn window_wrapping_non_sketchable_stays_separate() { - // Custom func is not sketchable → Window stays, Aggregate stays. - let expr = QueryExpr::Window { - duration: Duration::from_secs(300), - slide: None, - input: Box::new(make_agg(AggFunc::Custom("my_udf".into()), src("m"))), - }; - let lowered = lower_to_sketch_algebra(expr); - assert!(matches!(lowered, QueryExpr::Window { .. })); - } - - #[test] - fn lowering_recurses_into_binary_op() { - let expr = QueryExpr::BinaryOp { - op: BinaryOpKind::Add, - lhs: Box::new(make_agg(AggFunc::Sum, src("a"))), - rhs: Box::new(make_agg(AggFunc::CountDistinct, src("b"))), - vector_match: None, - }; - let lowered = lower_to_sketch_algebra(expr); - match &lowered { - QueryExpr::BinaryOp { lhs, rhs, .. } => { - assert!(matches!(lhs.as_ref(), QueryExpr::SketchAgg { op: AggIntent::Sum, .. })); - assert!(matches!(rhs.as_ref(), QueryExpr::SketchAgg { op: AggIntent::Cardinality { .. }, .. })); - } - other => panic!("expected BinaryOp, got {other:?}"), - } - } - - #[test] - fn lowering_recurses_into_topk() { - let expr = QueryExpr::TopK { - k: 10, - by: vec!["symbol".into()], - input: Box::new(make_agg_with_keys( - AggFunc::Count, - vec!["symbol".into()], - src("m"), - )), - }; - let lowered = lower_to_sketch_algebra(expr); - match &lowered { - QueryExpr::TopK { input, .. } => { - // Count with GROUP BY is lowered to Partition(SketchAgg(Frequency)) - assert!(matches!(input.as_ref(), QueryExpr::Partition { .. })); - } - other => panic!("expected TopK, got {other:?}"), - } - } - - #[test] - fn rate_lowered_to_sum() { - let expr = make_agg(AggFunc::Rate, src("m")); - let lowered = lower_to_sketch_algebra(expr); - assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Sum, .. })); - } - - #[test] - fn delta_lowered_to_sum() { - let expr = make_agg(AggFunc::Delta, src("m")); - let lowered = lower_to_sketch_algebra(expr); - assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Sum, .. })); - } -} diff --git a/control_plane/src/intent_algebra/legacy_to_canonical.rs b/control_plane/src/intent_algebra/legacy_to_canonical.rs index 7e281282..f67e9056 100644 --- a/control_plane/src/intent_algebra/legacy_to_canonical.rs +++ b/control_plane/src/intent_algebra/legacy_to_canonical.rs @@ -48,8 +48,8 @@ //! //! [`convert`] takes a `&Schema` — the schema in scope at the node — and //! threads it unchanged to every child, matching the established -//! `legacy_lower` / `column_resolution::infer_schema_for_root` -//! convention. The synthesized source schema is `(ts, value)` and almost +//! `column_resolution::infer_schema_for_root` convention. The +//! synthesized source schema is `(ts, value)` and almost //! every legacy operator is schema-pass-through, so threading the root //! schema down is correct except for the nested-schema-transform case //! (an `Aggregate` below another `Aggregate`), which the legacy stack @@ -68,10 +68,9 @@ use crate::intent_algebra::column_resolution::{ resolve_named_keys, ResolveError, }; use crate::intent_algebra::legacy_expr::{ - AggFunc, ColumnRef as LColumnRef, PartitionKeys as LPartitionKeys, QueryExpr as LQueryExpr, - ScalarExpr as LScalarExpr, WindowKind as LWindowKind, WindowSpec, + AggFunc, AggItem, ColumnRef as LColumnRef, PartitionKeys as LPartitionKeys, + QueryExpr as LQueryExpr, ScalarExpr as LScalarExpr, WindowKind as LWindowKind, WindowSpec, }; -use crate::intent_algebra::legacy_lower::agg_func_to_intents; use crate::intent_algebra::query_expr::{ from_legacy_scalar, ColumnRef as CColumnRef, HavingPredicate, LiteralValue, PartitionKeys as CPartitionKeys, Predicate, ProjectItem as CProjectItem, @@ -103,7 +102,20 @@ pub enum ConvertError { Scalar(QueryExprError), } -/// Convert a legacy `QueryExpr` tree to canonical. +/// Lower a legacy `QueryExpr` tree all the way to canonical. +/// +/// Two internal walks, one public entry: +/// +/// 1. [`lower_to_sketch_algebra`] folds Layer-2 relational `Aggregate +/// { AggFunc }` shapes into the sketch-fused legacy Layer-3 form +/// (`SketchAgg` / `WindowedAgg` / `Partition`). This is idempotent on +/// input that is already Layer 3 — the `SketchAgg` / `WindowedAgg` +/// arms are pass-through — so callers may hand `convert_root` either +/// level. +/// 2. [`convert`] maps that legacy Layer-3 tree onto the canonical IR. +/// +/// The legacy Layer-3 fused IR never escapes this module: it is purely +/// the intermediate between these two walks. /// /// The inherited schema comes from the [`Binder`] — the explicit L3 /// name-resolution pass — which builds the complete, self-contained @@ -112,8 +124,9 @@ pub enum ConvertError { /// below (`resolve_column_ref` / `resolve_named_keys`) is **total**: it /// cannot raise `ConvertError::Resolve` on a well-formed legacy tree. pub fn convert_root(legacy: &LQueryExpr) -> Result { - let schema = Binder::new().bind(legacy); - convert(legacy, &schema) + let lowered = lower_to_sketch_algebra(legacy.clone()); + let schema = Binder::new().bind(&lowered); + convert(&lowered, &schema) } /// Convert a legacy `QueryExpr` tree to canonical against an explicit @@ -442,6 +455,315 @@ fn convert_column_ref(c: &LColumnRef) -> CColumnRef { } } +// ── Layer 2 → Layer 3 sketch lowering ──────────────────────────────────────── +// +// Formerly `intent_algebra::legacy_lower`. Folded in here because its sole +// caller is `convert_root` above (the `query_parser` entry points emit raw +// Layer-2 trees and route them straight through `convert_root`). Keeping it +// private to this module means the sketch-fused legacy Layer-3 IR +// (`SketchAgg` / `WindowedAgg`) is no longer a surface anything outside the +// converter can construct or observe. + +/// Lower a Layer-2 legacy `QueryExpr` (relational operators only) to the +/// Layer-3 sketch algebra: `Aggregate { AggFunc }` nodes whose function +/// maps to a sketch intent become `SketchAgg { AggIntent }` (or, fused +/// under a `Window`, `WindowedAgg`). Multi-agg `Aggregate`s, those with +/// `HAVING`, and non-sketchable functions pass through unchanged. +/// +/// Idempotent on input that is already Layer 3: the `SketchAgg` / +/// `WindowedAgg` arms simply recurse. +fn lower_to_sketch_algebra(expr: LQueryExpr) -> LQueryExpr { + match expr { + LQueryExpr::Aggregate { + keys, + aggs, + having, + input, + } => { + let input = lower_to_sketch_algebra(*input); + lower_aggregate(keys, aggs, having, Box::new(input)) + } + + // ── Single-input nodes: recurse ────────────────────────────────── + LQueryExpr::Filter { pred, input } => LQueryExpr::Filter { + pred, + input: Box::new(lower_to_sketch_algebra(*input)), + }, + LQueryExpr::Project { cols, input } => LQueryExpr::Project { + cols, + input: Box::new(lower_to_sketch_algebra(*input)), + }, + LQueryExpr::Window { + duration, + slide, + input, + } => { + let lowered_input = lower_to_sketch_algebra(*input); + // Fuse Window + SketchAgg → WindowedAgg (the window defines the + // sketch lifecycle). + if let LQueryExpr::SketchAgg { + op, + col, + input: sketch_input, + } = lowered_input + { + let window = WindowSpec { + kind: match slide { + Some(s) => LWindowKind::Sliding { + size: duration, + slide: s, + }, + None => LWindowKind::Tumbling { size: duration }, + }, + time_col: None, + }; + LQueryExpr::WindowedAgg { + agg: op, + window, + col, + input: sketch_input, + } + } else { + LQueryExpr::Window { + duration, + slide, + input: Box::new(lowered_input), + } + } + } + LQueryExpr::SketchAgg { op, col, input } => LQueryExpr::SketchAgg { + op, + col, + input: Box::new(lower_to_sketch_algebra(*input)), + }, + LQueryExpr::WindowedAgg { + agg, + window, + col, + input, + } => LQueryExpr::WindowedAgg { + agg, + window, + col, + input: Box::new(lower_to_sketch_algebra(*input)), + }, + LQueryExpr::Partition { keys, input } => LQueryExpr::Partition { + keys, + input: Box::new(lower_to_sketch_algebra(*input)), + }, + LQueryExpr::Distinct { cols, input } => LQueryExpr::Distinct { + cols, + input: Box::new(lower_to_sketch_algebra(*input)), + }, + LQueryExpr::TopK { k, by, input } => LQueryExpr::TopK { + k, + by, + input: Box::new(lower_to_sketch_algebra(*input)), + }, + LQueryExpr::Sort { keys, input } => LQueryExpr::Sort { + keys, + input: Box::new(lower_to_sketch_algebra(*input)), + }, + LQueryExpr::Limit { n, offset, input } => LQueryExpr::Limit { + n, + offset, + input: Box::new(lower_to_sketch_algebra(*input)), + }, + LQueryExpr::PromQLSubquery { + range, + resolution, + input, + } => LQueryExpr::PromQLSubquery { + range, + resolution, + input: Box::new(lower_to_sketch_algebra(*input)), + }, + + // ── Two-input nodes: recurse into both ────────────────────────── + LQueryExpr::BinaryOp { + op, + lhs, + rhs, + vector_match, + } => LQueryExpr::BinaryOp { + op, + lhs: Box::new(lower_to_sketch_algebra(*lhs)), + rhs: Box::new(lower_to_sketch_algebra(*rhs)), + vector_match, + }, + LQueryExpr::Join { + kind, + pred, + left, + right, + } => LQueryExpr::Join { + kind, + pred, + left: Box::new(lower_to_sketch_algebra(*left)), + right: Box::new(lower_to_sketch_algebra(*right)), + }, + LQueryExpr::SetOp { + kind, + all, + left, + right, + } => LQueryExpr::SetOp { + kind, + all, + left: Box::new(lower_to_sketch_algebra(*left)), + right: Box::new(lower_to_sketch_algebra(*right)), + }, + + // ── Multi-input / container nodes ─────────────────────────────── + LQueryExpr::Merge { inputs } => LQueryExpr::Merge { + inputs: inputs.into_iter().map(lower_to_sketch_algebra).collect(), + }, + LQueryExpr::LetBinding { name, expr, body } => LQueryExpr::LetBinding { + name, + expr: Box::new(lower_to_sketch_algebra(*expr)), + body: Box::new(lower_to_sketch_algebra(*body)), + }, + + // ── Leaf nodes: pass through ──────────────────────────────────── + LQueryExpr::Source(_) | LQueryExpr::Ref(_) => expr, + } +} + +/// Lower a single `Aggregate` node to `SketchAgg` / `WindowedAgg` where the +/// aggregation benefits from a sketch. +/// +/// Single-statistic functions produce one sketch node; `StdDev` / `Variance` +/// fan out into two sibling quantile sketches wrapped in a `Merge` (Step α +/// F1 strategy). `GROUP BY` keys become a wrapping `Partition`. Multi-agg, +/// `HAVING`-bearing, and non-sketchable (`Custom`) aggregates pass through +/// as a relational `Aggregate`. +fn lower_aggregate( + keys: Vec, + aggs: Vec, + having: Option, + input: Box, +) -> LQueryExpr { + // Only lower single-agg Aggregates without HAVING. + if aggs.len() == 1 && having.is_none() { + let agg = &aggs[0]; + + // COUNT(*) without GROUP BY is a simple row count — no sketch benefit. + if matches!(agg.func, AggFunc::Count) && keys.is_empty() { + return LQueryExpr::Aggregate { + keys, + aggs, + having, + input, + }; + } + + let intents = agg_func_to_intents(&agg.func); + if !intents.is_empty() { + let col = agg.col.clone(); + let sketch_nodes: Vec = match *input { + LQueryExpr::Window { + duration, + slide, + input: ref win_input, + } => { + let window = WindowSpec { + kind: match slide { + Some(s) => LWindowKind::Sliding { + size: duration, + slide: s, + }, + None => LWindowKind::Tumbling { size: duration }, + }, + time_col: None, + }; + intents + .into_iter() + .map(|intent| LQueryExpr::WindowedAgg { + agg: intent, + window: window.clone(), + col: col.clone(), + input: win_input.clone(), + }) + .collect() + } + ref other => { + let inp_boxed: Box = Box::new(other.clone()); + intents + .into_iter() + .map(|intent| LQueryExpr::SketchAgg { + op: intent, + col: col.clone(), + input: inp_boxed.clone(), + }) + .collect() + } + }; + let sketch = if sketch_nodes.len() == 1 { + sketch_nodes.into_iter().next().unwrap() + } else { + LQueryExpr::Merge { + inputs: sketch_nodes, + } + }; + + return if keys.is_empty() { + sketch + } else { + LQueryExpr::Partition { + keys: LPartitionKeys::By(keys), + input: Box::new(sketch), + } + }; + } + } + + // Multi-agg or non-sketchable: keep as relational Aggregate. + LQueryExpr::Aggregate { + keys, + aggs, + having, + input, + } +} + +/// Map an [`AggFunc`] to the canonical [`AggIntent`]s needed for sketch +/// execution. Empty for non-sketchable functions (`Custom`); one intent +/// for single-statistic functions; two for the `StdDev` / `Variance` +/// fan-out (the caller wraps the pair in a `Merge` of sibling sketches). +fn agg_func_to_intents(func: &AggFunc) -> Vec { + use crate::intent_algebra::legacy_expr::{ + default_cardinality, default_frequency, default_quantile, + }; + match func { + AggFunc::Quantile(phi) => vec![default_quantile(*phi)], + AggFunc::CountDistinct => vec![default_cardinality()], + AggFunc::HeavyHitters { .. } => vec![default_frequency()], + AggFunc::Count => vec![default_frequency()], + AggFunc::Avg => vec![AggIntent::Quantile { + q: 0.5, + accuracy: AccuracyTarget::Epsilon(0.01), + }], + AggFunc::Min => vec![AggIntent::Min], + AggFunc::Max => vec![AggIntent::Max], + // StdDev / Variance: legacy carried two quantiles in a single + // Quantile intent; Step α F1 fans them out into two siblings. + AggFunc::StdDev { .. } | AggFunc::Variance { .. } => vec![ + AggIntent::Quantile { + q: 0.25, + accuracy: AccuracyTarget::Epsilon(0.01), + }, + AggIntent::Quantile { + q: 0.75, + accuracy: AccuracyTarget::Epsilon(0.01), + }, + ], + AggFunc::Sum | AggFunc::Rate | AggFunc::Increase | AggFunc::Delta => { + vec![AggIntent::Sum] + } + AggFunc::Custom(_) => vec![], + } +} + // ── Tests ───────────────────────────────────────────────────────────────────── #[cfg(test)] @@ -765,3 +1087,256 @@ mod tests { assert!(matches!(*child, CQueryExpr::Scan { .. })); } } + +#[cfg(test)] +mod lower_tests { + //! Ported from the former `intent_algebra::legacy_lower` module — + //! exercises the private Layer-2 → Layer-3 `lower_to_sketch_algebra` + //! sketch-fusion pass that `convert_root` now runs internally. + use super::lower_to_sketch_algebra; + use crate::intent_algebra::legacy_expr::*; + use std::time::Duration; + + fn src(name: &str) -> QueryExpr { + QueryExpr::Source(SourceSpec { name: name.into() }) + } + + fn make_agg(func: AggFunc, input: QueryExpr) -> QueryExpr { + QueryExpr::Aggregate { + keys: vec![], + aggs: vec![AggItem { + alias: "v".into(), + func, + col: ColumnRef::SampleValue, + distinct: false, + }], + having: None, + input: Box::new(input), + } + } + + fn make_agg_with_keys(func: AggFunc, keys: Vec, input: QueryExpr) -> QueryExpr { + QueryExpr::Aggregate { + keys, + aggs: vec![AggItem { + alias: "v".into(), + func, + col: ColumnRef::SampleValue, + distinct: false, + }], + having: None, + input: Box::new(input), + } + } + + #[test] + fn quantile_lowered_to_sketch_agg() { + let expr = make_agg(AggFunc::Quantile(0.99), src("m")); + let lowered = lower_to_sketch_algebra(expr); + assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Quantile { .. }, .. })); + } + + #[test] + fn count_distinct_lowered_to_cardinality() { + let expr = make_agg(AggFunc::CountDistinct, src("m")); + let lowered = lower_to_sketch_algebra(expr); + assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Cardinality { .. }, .. })); + } + + #[test] + fn count_without_group_by_stays_aggregate() { + let expr = make_agg(AggFunc::Count, src("m")); + let lowered = lower_to_sketch_algebra(expr); + assert!(matches!(lowered, QueryExpr::Aggregate { .. })); + } + + #[test] + fn count_with_group_by_lowered_to_frequency() { + let expr = make_agg_with_keys(AggFunc::Count, vec!["region".into()], src("m")); + let lowered = lower_to_sketch_algebra(expr); + match &lowered { + QueryExpr::Partition { input, .. } => { + assert!(matches!(input.as_ref(), QueryExpr::SketchAgg { op: AggIntent::Frequency { .. }, .. })); + } + other => panic!("expected Partition(SketchAgg), got {other:?}"), + } + } + + #[test] + fn sum_lowered_to_sum() { + let expr = make_agg(AggFunc::Sum, src("m")); + let lowered = lower_to_sketch_algebra(expr); + assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Sum, .. })); + } + + #[test] + fn avg_lowered_to_quantile_p50() { + let expr = make_agg(AggFunc::Avg, src("m")); + let lowered = lower_to_sketch_algebra(expr); + match &lowered { + QueryExpr::SketchAgg { op: AggIntent::Quantile { q, .. }, .. } => { + assert!((*q - 0.5).abs() < 1e-9); + } + other => panic!("expected SketchAgg(Quantile), got {other:?}"), + } + } + + #[test] + fn min_lowered_to_min() { + let expr = make_agg(AggFunc::Min, src("m")); + let lowered = lower_to_sketch_algebra(expr); + assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Min, .. })); + } + + #[test] + fn max_lowered_to_max() { + let expr = make_agg(AggFunc::Max, src("m")); + let lowered = lower_to_sketch_algebra(expr); + assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Max, .. })); + } + + #[test] + fn stddev_fans_out_to_merge_of_quantile_siblings() { + let expr = make_agg(AggFunc::StdDev { population: false }, src("m")); + let lowered = lower_to_sketch_algebra(expr); + match &lowered { + QueryExpr::Merge { inputs } => { + assert_eq!(inputs.len(), 2); + let mut qs: Vec = inputs.iter().filter_map(|node| match node { + QueryExpr::SketchAgg { op: AggIntent::Quantile { q, .. }, .. } => Some(*q), + _ => None, + }).collect(); + qs.sort_by(|a, b| a.partial_cmp(b).unwrap()); + assert_eq!(qs, vec![0.25, 0.75]); + } + other => panic!("expected Merge of two SketchAgg, got {other:?}"), + } + } + + #[test] + fn custom_func_not_lowered() { + let expr = make_agg(AggFunc::Custom("my_udf".into()), src("m")); + let lowered = lower_to_sketch_algebra(expr); + assert!(matches!(lowered, QueryExpr::Aggregate { .. })); + } + + #[test] + fn group_by_wraps_with_partition() { + let expr = make_agg_with_keys(AggFunc::Quantile(0.5), vec!["host".into()], src("m")); + let lowered = lower_to_sketch_algebra(expr); + match &lowered { + QueryExpr::Partition { keys, input } => { + assert_eq!(keys.keys(), &["host".to_string()]); + assert!(matches!(input.as_ref(), QueryExpr::SketchAgg { .. })); + } + other => panic!("expected Partition, got {other:?}"), + } + } + + #[test] + fn multi_agg_not_lowered() { + let expr = QueryExpr::Aggregate { + keys: vec![], + aggs: vec![ + AggItem { alias: "c".into(), func: AggFunc::Count, col: ColumnRef::Wildcard, distinct: false }, + AggItem { alias: "s".into(), func: AggFunc::Sum, col: ColumnRef::Named("x".into()), distinct: false }, + ], + having: None, + input: Box::new(src("m")), + }; + let lowered = lower_to_sketch_algebra(expr); + assert!(matches!(lowered, QueryExpr::Aggregate { .. })); + } + + #[test] + fn window_wrapping_aggregate_fuses_to_windowed_agg() { + let expr = QueryExpr::Window { + duration: Duration::from_secs(300), + slide: None, + input: Box::new(make_agg(AggFunc::Quantile(0.99), src("m"))), + }; + let lowered = lower_to_sketch_algebra(expr); + match &lowered { + QueryExpr::WindowedAgg { agg, window, .. } => { + assert!(matches!(agg, AggIntent::Quantile { .. })); + assert!(matches!(window.kind, WindowKind::Tumbling { .. })); + } + other => panic!("expected WindowedAgg, got {other:?}"), + } + } + + #[test] + fn sliding_window_fuses_to_windowed_agg_sliding() { + let expr = QueryExpr::Window { + duration: Duration::from_secs(300), + slide: Some(Duration::from_secs(60)), + input: Box::new(make_agg(AggFunc::Quantile(0.5), src("m"))), + }; + let lowered = lower_to_sketch_algebra(expr); + match &lowered { + QueryExpr::WindowedAgg { window, .. } => { + assert!(matches!(window.kind, WindowKind::Sliding { .. })); + } + other => panic!("expected WindowedAgg(Sliding), got {other:?}"), + } + } + + #[test] + fn window_wrapping_non_sketchable_stays_separate() { + let expr = QueryExpr::Window { + duration: Duration::from_secs(300), + slide: None, + input: Box::new(make_agg(AggFunc::Custom("my_udf".into()), src("m"))), + }; + let lowered = lower_to_sketch_algebra(expr); + assert!(matches!(lowered, QueryExpr::Window { .. })); + } + + #[test] + fn lowering_recurses_into_binary_op() { + let expr = QueryExpr::BinaryOp { + op: BinaryOpKind::Add, + lhs: Box::new(make_agg(AggFunc::Sum, src("a"))), + rhs: Box::new(make_agg(AggFunc::CountDistinct, src("b"))), + vector_match: None, + }; + let lowered = lower_to_sketch_algebra(expr); + match &lowered { + QueryExpr::BinaryOp { lhs, rhs, .. } => { + assert!(matches!(lhs.as_ref(), QueryExpr::SketchAgg { op: AggIntent::Sum, .. })); + assert!(matches!(rhs.as_ref(), QueryExpr::SketchAgg { op: AggIntent::Cardinality { .. }, .. })); + } + other => panic!("expected BinaryOp, got {other:?}"), + } + } + + #[test] + fn lowering_recurses_into_topk() { + let expr = QueryExpr::TopK { + k: 10, + by: vec!["symbol".into()], + input: Box::new(make_agg_with_keys(AggFunc::Count, vec!["symbol".into()], src("m"))), + }; + let lowered = lower_to_sketch_algebra(expr); + match &lowered { + QueryExpr::TopK { input, .. } => { + assert!(matches!(input.as_ref(), QueryExpr::Partition { .. })); + } + other => panic!("expected TopK, got {other:?}"), + } + } + + #[test] + fn rate_lowered_to_sum() { + let expr = make_agg(AggFunc::Rate, src("m")); + let lowered = lower_to_sketch_algebra(expr); + assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Sum, .. })); + } + + #[test] + fn delta_lowered_to_sum() { + let expr = make_agg(AggFunc::Delta, src("m")); + let lowered = lower_to_sketch_algebra(expr); + assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Sum, .. })); + } +} diff --git a/control_plane/src/intent_algebra/mod.rs b/control_plane/src/intent_algebra/mod.rs index f129739c..f37afa04 100644 --- a/control_plane/src/intent_algebra/mod.rs +++ b/control_plane/src/intent_algebra/mod.rs @@ -80,20 +80,20 @@ pub mod query_expr; pub mod schema; // Refactor 2026-05 (`refactor/controller-layered-cleanup`): the -// pre-existing legacy L3+ IR formerly at `controller/src/algebra/expr.rs` -// + `controller/src/algebra/lower.rs` lives here while a separate -// follow-up unifies it with the canonical `query_expr` / `lower` -// modules above. These two `legacy_*` modules carry the heavy -// `QueryExpr` / `AggIntent` types used by the planner, allocator, -// physical planner, query_parser, and language_logical_plan modules -// today. +// pre-existing legacy L2 relational IR formerly at +// `controller/src/algebra/expr.rs` lives here while a separate follow-up +// unifies it with the canonical `query_expr` module above. It still +// carries the heavy `QueryExpr` type the `query_parser` modules emit and +// the planner / allocator / physical planner / language_logical_plan +// modules consume today. pub mod column_resolution; pub mod legacy_expr; -pub mod legacy_lower; -// Step γ7 keystone: the composable legacy → canonical full-tree -// converter. Producers route their output through `convert_root`; the -// consumer-flip PRs then become pure canonical pattern-match rewrites. +// Step γ7 keystone: the legacy → canonical full-tree converter. The +// `query_parser` entry points emit raw Layer-2 trees and route them +// through `convert_root`, which first folds them into the sketch-fused +// legacy Layer-3 form (the former `legacy_lower` pass, now private to +// this module) and then maps that onto the canonical IR. pub mod legacy_to_canonical; pub use legacy_to_canonical::{convert as convert_legacy, convert_root, ConvertError}; @@ -123,9 +123,9 @@ pub use schema::{cse_reuse_is_legal, Column, ColumnId, CseError, DataType, Schem // Step β plumbing: schema-driven column resolution helpers used by the // legacy planning stack (`optimizer/engine.rs`, `physical/{allocator, -// planner, stage_split}.rs`, `query_parser/*`, `intent_algebra:: -// legacy_lower`) to carry an inherited `Schema` alongside every legacy -// `QueryExpr` traversal. Consumers call `resolve_column_ref` at the point +// planner, stage_split}.rs`, `query_parser/*`) to carry an inherited +// `Schema` alongside every legacy `QueryExpr` traversal. Consumers call +// `resolve_column_ref` at the point // where they need a positional `ColumnId` — Step γ migrates variants // one at a time onto the canonical positional form. pub use column_resolution::{ @@ -133,45 +133,10 @@ pub use column_resolution::{ resolve_column_refs, resolve_named_keys, ResolveError, }; -// Step γ1 bridge: one-way `legacy_expr::QueryExpr::Aggregate` → -// canonical `query_expr::QueryExpr::Aggregate` helper. Consumers that -// need canonical-shape pattern-matching (`by: Vec`, `aggs: -// Vec`) call this on an inherited Schema; the legacy variant -// remains in `legacy_expr.rs` until every consumer entry point migrates -// (Step γ7 or later). -pub mod aggregate_bridge; -pub use aggregate_bridge::{ - bridge_aggregate_to_canonical, BridgeError, BridgedAggregate, -}; - -// Step γ4 bridge: one-way `legacy_expr::QueryExpr::TopK` → one of two -// canonical shapes (heavy-hitter `AggIntent::TopK` vs generic -// `Sort + Limit`) per design.md §6 "What was removed" row 1. Consumers -// call this on an inherited Schema; the legacy variant remains in -// `legacy_expr.rs` until every consumer entry point migrates (Step γ7 -// or later). -pub mod topk_bridge; -pub use topk_bridge::{bridge_topk, BridgeError as TopKBridgeError, BridgedTopK}; - -// Step γ3 bridge: one-way `legacy_expr::QueryExpr::WindowedAgg` → -// canonical `Window { child: Aggregate { .. } }` helper. Sidesteps the -// 20+ consumers that depend on the fused `WindowedAgg`'s window-sketch -// lifecycle invariant — they continue matching the legacy variant; only -// consumers that want the canonical stacked view call this bridge. The -// legacy variant remains in `legacy_expr.rs` until Step γ7 retires it. -pub mod windowed_agg_bridge; -pub use windowed_agg_bridge::{ - bridge_windowed_agg_to_canonical, output_schema_for_windowed_agg, - BridgeError as WindowedAggBridgeError, BridgedWindowedAgg, -}; - -// Step γ2 bridge: one-way `legacy_expr::QueryExpr::SketchAgg` → -// canonical-shape `BridgedAggregate` helper. Companion to the γ1 -// Aggregate bridge — SketchAgg's single-intent / single-column shape -// maps onto the same `BridgedAggregate` carrier (with `having: None`) -// so consumers can match on `by` / `aggs` regardless of which legacy -// variant the data came from. -pub mod sketch_agg_bridge; -pub use sketch_agg_bridge::{ - bridge_sketch_agg_to_canonical, output_schema_for_sketch_agg, -}; +// Step γ7 (PR 13): the four γ1–γ4 one-way "approach (c)" bridges +// (`aggregate_bridge`, `topk_bridge`, `windowed_agg_bridge`, +// `sketch_agg_bridge`) were deleted. They returned canonical-shape data +// *minus the child* — useful only as a non-composable migration aid +// while consumers still pattern-matched legacy variants. Every consumer +// now runs on the canonical IR via the composable `legacy_to_canonical` +// converter, so the bridges had zero remaining call sites. diff --git a/control_plane/src/query_parser/mod.rs b/control_plane/src/query_parser/mod.rs index e1d92c9b..cb19ba3f 100644 --- a/control_plane/src/query_parser/mod.rs +++ b/control_plane/src/query_parser/mod.rs @@ -4,7 +4,7 @@ //! //! | Function | Returns | Use | //! |---|---|---| -//! | [`parse_query_expr`] | `QueryExpr` | Full algebra IR | +//! | [`parse_query_expr_canonical`] | canonical `query_expr::QueryExpr` | Full algebra IR | //! | [`parse_query`] | `ParsedQuery` | Backward compat with existing analyzer | //! //! # Supported PromQL patterns (via `promql-parser` AST) @@ -95,46 +95,39 @@ pub enum QueryHint { // ── Public entry points ─────────────────────────────────────────────────────── -/// Parse a raw query string (PromQL or SQL) into the **legacy** L3 +/// Parse a raw query string (PromQL or SQL) into the **legacy Layer-2** /// [`legacy_expr::QueryExpr`](crate::intent_algebra::legacy_expr::QueryExpr) IR. /// -/// Both parsers emit Layer 2 relational operators (`Aggregate { AggFunc }`). -/// The shared lowering pass converts `Aggregate` → `SketchAgg { AggIntent }` -/// where applicable. +/// Both parsers emit Layer-2 relational operators (`Aggregate { AggFunc }`, +/// `Window`, `Filter`, `Join`, …). 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. /// -/// Step γ7: this still returns the *legacy* IR — it remains the internal -/// L2→L3 path that the producers (`sql.rs` / `promql.rs` / `legacy_lower`) -/// and `language_logical_plan` build on. New consumers use -/// [`parse_query_expr_canonical`]; the legacy path is removed once every -/// consumer migrates. -pub fn parse_query_expr( +/// 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(); - let layer2 = if upper.starts_with("SELECT") || upper.starts_with("WITH") { - sql::parse_sql_expr(q)? + if upper.starts_with("SELECT") || upper.starts_with("WITH") { + sql::parse_sql_expr(q) } else { - promql::parse_promql_expr(q)? - }; - // Layer 2 → Layer 3 lowering (shared by both languages). - Ok(crate::intent_algebra::legacy_lower::lower_to_sketch_algebra(layer2)) + promql::parse_promql_expr(q) + } } /// Parse a raw query string (PromQL or SQL) into the **canonical** L3 /// [`query_expr::QueryExpr`](crate::intent_algebra::query_expr::QueryExpr) IR. /// -/// This is the canonical-IR twin of [`parse_query_expr`]: it parses via the -/// exact same path (it calls [`parse_query_expr`] internally to obtain the -/// legacy [`QueryExpr`]) and then runs the result through -/// [`intent_algebra::convert_root`](crate::intent_algebra::convert_root) to -/// produce a canonical `query_expr::QueryExpr` tree. -/// -/// Both entry points coexist deliberately during the Step γ7 legacy-IR -/// retirement: [`parse_query_expr`] keeps returning the legacy `QueryExpr` -/// so every existing `optimizer/` / `physical/` / `main.rs` consumer keeps -/// compiling, while new canonical-IR consumers migrate onto this entry -/// point one at a time. A later PR removes the legacy path entirely. +/// This is the single public algebra-IR entry point. It parses the query +/// into the crate-internal legacy Layer-2 tree via [`parse_query_expr`], +/// then runs that through +/// [`intent_algebra::convert_root`](crate::intent_algebra::convert_root), +/// which folds the Layer-2 → Layer-3 sketch lowering and the +/// legacy → canonical conversion into one entry. The legacy IR is never +/// observable to callers. pub fn parse_query_expr_canonical( query: &str, ) -> anyhow::Result { @@ -525,69 +518,95 @@ mod tests { } #[test] - fn legacy_entry_point_still_returns_legacy_type() { - // `parse_query_expr` is untouched: it still returns the legacy - // `QueryExpr` so every existing optimizer / physical consumer keeps - // compiling. Asserting on a legacy-only variant proves the type. - let legacy = parse_query_expr( + fn legacy_entry_point_returns_raw_layer2() { + use crate::intent_algebra::legacy_expr::{AggFunc, QueryExpr as LQueryExpr}; + // `parse_query_expr` is the crate-internal language-dispatch front + // door: it returns the raw legacy Layer-2 relational tree with no + // sketch lowering applied — the L2→L3 fusion now lives inside + // `convert_root`. + let layer2 = parse_query_expr( "quantile_over_time(0.99, http_request_duration{env=\"prod\"}[5m])" ).unwrap(); - // `WindowedAgg` exists only on the legacy IR — the canonical IR - // splits it into `Window { Aggregate }`. - assert!(matches!( - legacy, - crate::intent_algebra::legacy_expr::QueryExpr::WindowedAgg { .. } - )); + // Raw Layer 2: an `Aggregate { AggFunc::Quantile }` sitting + // *directly* over a `Window` — un-fused, un-lowered. + match layer2 { + LQueryExpr::Aggregate { aggs, input, .. } => { + assert!(matches!( + aggs.as_slice(), + [item] if matches!(item.func, AggFunc::Quantile(_)) + )); + assert!(matches!(*input, LQueryExpr::Window { .. })); + } + other => panic!("expected raw Layer-2 Aggregate, got {other:?}"), + } } } #[cfg(test)] mod doc_verify_all { - // These tests pin the *legacy* L2→L3 parse path (`parse_query_expr`), - // so `QueryExpr` here is the legacy IR — not the canonical one that - // `super::*` would bring in. - use super::parse_query_expr; - use crate::intent_algebra::legacy_expr::*; + // These tests pin the design.md §6 worked examples against the + // canonical IR that `parse_query_expr_canonical` produces — the only + // algebra IR the parse path now emits. The legacy `WindowedAgg` / + // `SketchAgg` fusion the doc text once showed is folded by + // `convert_root` into the canonical `Window { Aggregate }` / + // `Aggregate { by: [], .. }` stacked forms. + use super::parse_query_expr_canonical; + use crate::intent_algebra::query_expr::QueryExpr; + use crate::intent_algebra::AggIntent; #[test] fn example4_promql_quantile() { - let expr = parse_query_expr( + let expr = parse_query_expr_canonical( "quantile_over_time(0.99, http_request_duration{env=\"prod\"}[5m])" ).unwrap(); - // Doc: WindowedAgg { Quantile([0.99]), Tumbling(5m), Filter(Source) } - assert!(matches!(&expr, QueryExpr::WindowedAgg { agg: AggIntent::Quantile { .. }, .. })); + // Canonical fold of the legacy `WindowedAgg { Quantile }`: + // `Window { Aggregate { by: [], [Quantile] } }`. + match &expr { + QueryExpr::Window { child, .. } => match child.as_ref() { + QueryExpr::Aggregate { by, aggs, .. } => { + assert!(by.is_empty()); + assert!(matches!(aggs.as_slice(), [AggIntent::Quantile { .. }])); + } + other => panic!("expected Aggregate under Window, got {other:?}"), + }, + other => panic!("expected Window, got {other:?}"), + } } #[test] fn example5_promql_topk() { - let expr = parse_query_expr( + let expr = parse_query_expr_canonical( "topk by (service) (10, count_over_time(requests{env=\"prod\"}[1m]))" ).unwrap(); - // Doc: TopK { 10, Partition { ["service"], WindowedAgg { Frequency } } } + // The legacy `TopK` folds to a canonical `Aggregate` carrying an + // `AggIntent::TopK`, over the `Partition { Window { Aggregate } }` + // the grouped windowed frequency sketch lowers to. match &expr { - QueryExpr::TopK { k: 10, input, .. } => { - match input.as_ref() { - QueryExpr::Partition { keys, input: inner } => { - assert_eq!(keys.keys(), &["service".to_string()]); - assert!(matches!(inner.as_ref(), QueryExpr::WindowedAgg { agg: AggIntent::Frequency { .. }, .. })); - } - other => panic!("expected Partition, got {other:?}"), - } + QueryExpr::Aggregate { aggs, child, .. } => { + assert!(matches!(aggs.as_slice(), [AggIntent::TopK { k: 10, .. }])); + assert!(matches!(child.as_ref(), QueryExpr::Partition { .. })); } - other => panic!("expected TopK, got {other:?}"), + other => panic!("expected Aggregate with TopK intent, got {other:?}"), } } #[test] fn example6_sql_avg() { - let expr = parse_query_expr( + let expr = parse_query_expr_canonical( "SELECT symbol, AVG(price) FROM trades GROUP BY symbol" ).unwrap(); - // Doc: Partition { ["symbol"], SketchAgg { Quantile([0.5]), Source } } + // Canonical fold of `Partition { ["symbol"], SketchAgg { Quantile } }`: + // `Partition { ["symbol"], Aggregate { by: [], [Quantile] } }`. match &expr { - QueryExpr::Partition { keys, input } => { + QueryExpr::Partition { keys, child } => { assert_eq!(keys.keys(), &["symbol".to_string()]); - assert!(matches!(input.as_ref(), QueryExpr::SketchAgg { op: AggIntent::Quantile { .. }, .. })); + 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:?}"), } @@ -595,26 +614,28 @@ mod doc_verify_all { #[test] fn example7_sql_tumble() { - let expr = parse_query_expr( + 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(); - // Doc: Limit { 10, Sort { Partition { ["region"], WindowedAgg { Cardinality } } } } - match &expr { - QueryExpr::Limit { n: 10, input, .. } => { - match input.as_ref() { - QueryExpr::Sort { input: sort_inner, .. } => { - match sort_inner.as_ref() { - QueryExpr::Partition { keys, input: part_inner } => { - assert_eq!(keys.keys(), &["region".to_string()]); - assert!(matches!(part_inner.as_ref(), QueryExpr::WindowedAgg { agg: AggIntent::Cardinality { .. }, .. })); - } - other => panic!("expected Partition, got {other:?}"), - } - } - other => panic!("expected Sort, got {other:?}"), - } - } - other => panic!("expected Limit, got {other:?}"), - } + // 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 index 9b8df2d5..d571d5b9 100644 --- a/control_plane/src/query_parser/sql.rs +++ b/control_plane/src/query_parser/sql.rs @@ -884,56 +884,76 @@ mod tests { // ── TUMBLE / HOP windows ──────────────────────────────────────────────── - /// Helper that runs the full pipeline (parse + lower), not just Layer 2. - fn parse_full(sql: &str) -> QueryExpr { - super::super::parse_query_expr(sql) - .unwrap_or_else(|e| panic!("parse_query_expr failed: {e}\nSQL: {sql}")) + /// 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}")) } - fn has_windowed_agg(e: &QueryExpr) -> bool { + /// 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 { - QueryExpr::WindowedAgg { .. } => true, - QueryExpr::Partition { input, .. } - | QueryExpr::TopK { input, .. } - | QueryExpr::Sort { input, .. } - | QueryExpr::Limit { input, .. } => has_windowed_agg(input), - _ => false, + 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( + 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 WindowedAgg in tree, got {expr:?}" + "expected canonical Window{{Aggregate}} in tree, got {expr:?}" ); } #[test] fn hop_in_group_by_produces_windowed_agg() { - let expr = parse_full( + 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 WindowedAgg in tree, got {expr:?}" + "expected canonical Window{{Aggregate}} in tree, got {expr:?}" ); } #[test] fn time_bucket_in_group_by_produces_windowed_agg() { - let expr = parse_full( + 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 WindowedAgg in tree, got {expr:?}" + "expected canonical Window{{Aggregate}} in tree, got {expr:?}" ); }