diff --git a/control_plane/docs/design.md b/control_plane/docs/design.md index db227b18..48cf2243 100644 --- a/control_plane/docs/design.md +++ b/control_plane/docs/design.md @@ -44,8 +44,8 @@ DataCollector/controller already documents its query→sketch translation as a 5 | # | Layer | What it does | Today's locations | |---|-------|--------------|-------------------| | 1 | **Query Language** | Parse raw strings (PromQL, SQL, DataFusion, ElasticDSL, …) into a language-specific AST | DC `controller/src/query_parser/{promql,sql}.rs` + the per-language façade `controller/src/query_parser/language/{promql,sql,elastic_dsl}/`; asap-planner-rs pulls `promql-parser` + `sqlparser` directly; asap-fusion consumes a pre-built DataFusion `LogicalPlan` (its L1 happens upstream). *Refactor 2026-05 absorbed `controller/src/query_language/` into `query_parser::language/`.* | -| 2 | **Language Logical Plan** | Per-language algebra tree (`Aggregate` / `Window` / `Filter` / `Sort` / `Limit`) preserving language semantics, **no sketch names, no sketch binding** | DC `controller/src/language_logical_plan/{lower,plan}.rs` + `controller/src/intent_algebra/legacy_lower.rs` (legacy L2→L3 lowering pending unification with `intent_algebra/lower.rs`); asap-fusion inherits DataFusion's `LogicalPlan` as its L2; asap-planner-rs has no L2 today (uses a template-pattern catalogue) — **Phase 4 builds one** | -| 3 | **Intent algebra** | Language- and deployment-independent IR: `QueryExpr` + `AggIntent`. Describes **intent only** — *what* to compute, with accuracy target. **No sketch type, no sketch parameters, no sketch-bound nodes** (`SketchAgg` / `SketchJoin` / `SketchSubtract` etc. live in the L4 IR `SketchExpr`). **No language-shaped operators** (no `HistogramQuantile`, no `PromQLSubquery` — those are PromQL L2 nodes that lower to data-model-agnostic shapes here). **One canonical form per plan** — no `WindowedAgg` (use `Window` over `Aggregate`). Heavy-hitter intents are first-class (`AggIntent::TopK`) so heavy-hitter sketches bind directly on the intent rather than on a generic `Sort + Limit` shape; generic `Sort + Limit` survives in `QueryExpr` for non-heavy-hitter cases (e.g. `ORDER BY name LIMIT 10`). Every edge carries a typed `Schema`. Data-model-agnostic — `QueryExpr::Scan` wraps a `Source` sum with `TimeSeries` / `Table` / `Join` variants so the same L3 IR covers ASAPQuery's time-series queries and asap-fusion's tabular queries. | DC `controller/src/intent_algebra/{agg_intent,query_expr,schema,lower,cse}.rs` (canonical L3) + `controller/src/intent_algebra/legacy_expr.rs` (legacy heavy-IR pending dedup with the canonical types); asap-fusion's 3-variant `SubPopulationAnalyticsType` maps to a subset; asap-planner-rs's 9-variant `Statistic` maps to a subset — **Phase 4 splits sketch binding out of planner's current fused L3+L4**. *Refactor 2026-05 absorbed `controller/src/algebra/expr.rs` into `intent_algebra/legacy_expr.rs`.* | +| 2 | **Language Logical Plan** | Per-language algebra tree (`Aggregate` / `Window` / `Filter` / `Sort` / `Limit`) preserving language semantics, **no sketch names, no sketch binding** | DC `controller/src/language_logical_plan/{lower,plan}.rs` + `controller/src/intent_algebra/legacy_expr.rs` (the legacy L2 relational IR the `query_parser` front ends emit); asap-fusion inherits DataFusion's `LogicalPlan` as its L2; asap-planner-rs has no L2 today (uses a template-pattern catalogue) — **Phase 4 builds one** | +| 3 | **Intent algebra** | Language- and deployment-independent IR: `QueryExpr` + `AggIntent`. Describes **intent only** — *what* to compute, with accuracy target. **No sketch type, no sketch parameters, no sketch-bound nodes** (`SketchAgg` / `SketchJoin` / `SketchSubtract` etc. live in the L4 IR `SketchExpr`). **No language-shaped operators** (no `HistogramQuantile`, no `PromQLSubquery` — those are PromQL L2 nodes that lower to data-model-agnostic shapes here). **One canonical form per plan** — no `WindowedAgg` (use `Window` over `Aggregate`). Heavy-hitter intents are first-class (`AggIntent::TopK`) so heavy-hitter sketches bind directly on the intent rather than on a generic `Sort + Limit` shape; generic `Sort + Limit` survives in `QueryExpr` for non-heavy-hitter cases (e.g. `ORDER BY name LIMIT 10`). Every edge carries a typed `Schema`. Data-model-agnostic — `QueryExpr::Scan` wraps a `Source` sum with `TimeSeries` / `Table` / `Join` variants so the same L3 IR covers ASAPQuery's time-series queries and asap-fusion's tabular queries. | DC `controller/src/intent_algebra/{agg_intent,query_expr,schema,lower,cse}.rs` (canonical L3) + `controller/src/intent_algebra/legacy_to_canonical.rs` (the L2→L3 converter: folds the legacy `legacy_expr` relational tree straight to the canonical IR, single-statistic sketchable `Aggregate` fusion included — no intermediate fused L3 IR); asap-fusion's 3-variant `SubPopulationAnalyticsType` maps to a subset; asap-planner-rs's 9-variant `Statistic` maps to a subset — **Phase 4 splits sketch binding out of planner's current fused L3+L4**. *Refactor 2026-05 absorbed `controller/src/algebra/expr.rs` into `intent_algebra/legacy_expr.rs`.* | | 4 | **Sketch algebra + optimizer** | Cost-aware algebraic rewrite rules under deployment constraints. **This is where sketch binding happens** — L4 rules take intent-only L3 (`QueryExpr`) and emit the sketch-bound IR (`SketchExpr`). ~12 rules in DC; a smaller targeted subset in planner; `SketchConfigRule` + `HashModeRule` in fusion. | Core provides the **rule engine driver** + `OptimizerRule` trait + a shared rule library + the sketch-bound IR `core::sketch_algebra::SketchExpr`; deployment models **pick** which rules to enable + supply their own deployment constraints. DC `controller/src/sketch_algebra/` (IR + binding rules) + `controller/src/optimizer/{engine,trait_def,baseline,rules/,cost/}.rs` (rule engine + cost models); fusion `src/optimizer/rules/`; planner's `map_statistic_to_precompute_operator`. *Refactor 2026-05 absorbed `controller/src/algebra/optimizer.rs` (→ `optimizer/engine.rs`) and `controller/src/planner/{cost_model,delta_cost_model,online_cost_model,pareto,tco,wire_cost,rules,baseline_planner}.rs` (→ `optimizer/{cost/,rules/,baseline.rs}`).* | | 5 | **Physical Execution Plan** | Assign ops to pipeline stages (edge / gateway / backend / object store); produce the deployment-specific artifact (OpAMP YAML, `streaming_config.yaml`, rewritten DataFusion `LogicalPlan`). **Sketch binding is already committed by L4**; L5 is about stage allocation + emission. | Core provides the **stage allocator framework** + `PhysicalPlanner` trait + the sketch catalogue; deployment models supply their own **topology** (3-stage / 1-stage / 0-stage) + their own **emitter** for the output format. DC `controller/src/physical/{allocator,planner,plan,sketch_catalog,stage_split,topology,colored_dag/}.rs` (allocator framework + sketch catalogue + typed three-stage colouring) + `controller/src/emit/{stage_config,otap,telegraf,agent,backend,asapquery_backend,precompute,trait_def}.rs` (per-deployment-model emitters + `PlanEmitter` trait) + `controller/src/pipeline.rs` (L1→…→L5 driver, formerly `analyzer.rs`); asap-planner-rs `output/generator.rs`; asap-fusion `src/executor/`. *Refactor 2026-05 absorbed `controller/src/algebra/{physical,allocator,plan,directory}.rs` and `controller/src/planner/stage_split.rs` and the legacy `controller/src/stage_split/` framework into `physical/`; absorbed `controller/src/config/` into `emit/` + `workload.rs`; renamed `analyzer.rs` to `pipeline.rs`.* | @@ -169,7 +169,7 @@ If something must be optional (e.g. OpAMP for deployments that don't run OTel co > |---|---| > | `crates/core/query_language/` | `controller/src/query_parser/language/` | > | `crates/core/logical_plan/` | `controller/src/language_logical_plan/` | -> | `crates/core/intent_algebra/` | `controller/src/intent_algebra/` (with `legacy_expr` / `legacy_lower` carrying the older `algebra::expr`-flavored IR pending unification) | +> | `crates/core/intent_algebra/` | `controller/src/intent_algebra/` (with `legacy_expr` carrying the legacy L2 relational IR and `legacy_to_canonical` converting it to the canonical L3 types) | > | `crates/core/sketch_algebra/` | `controller/src/sketch_algebra/` | > | `crates/core/optimizer/{engine,trait,rules,cost}/` | `controller/src/optimizer/{engine.rs,trait_def.rs,rules/,cost/,baseline.rs}` | > | `crates/core/physical/{planner_trait,stage_allocator,topology,executor,sketch_catalog}/` | `controller/src/physical/{planner,allocator,plan,stage_split,sketch_catalog,topology,colored_dag/}.rs` | @@ -289,7 +289,7 @@ Core is not a trait-stubs library. It ships real L1/L2/L3 code lifted from DC's > | `core::logical_plan` | `controller/src/language_logical_plan/` | > | `core::intent_algebra` | `controller/src/intent_algebra/` (canonical) + `controller/src/intent_algebra/legacy_{expr,lower}.rs` (legacy IR pending unification) | > | `core::sketch_algebra` | `controller/src/sketch_algebra/` | -> | `core::lower` | per-layer: `controller/src/language_logical_plan/lower.rs` + `controller/src/intent_algebra/{lower,legacy_lower}.rs` + `controller/src/sketch_algebra/lower.rs` | +> | `core::lower` | per-layer: `controller/src/language_logical_plan/lower.rs` + `controller/src/intent_algebra/{lower,legacy_to_canonical}.rs` + `controller/src/sketch_algebra/lower.rs` | > | `core::optimizer::engine` | `controller/src/optimizer/engine.rs` | > | `core::optimizer::trait` | `controller/src/optimizer/trait_def.rs` (placeholder) | > | `core::optimizer::rules` | `controller/src/optimizer/rules/` | @@ -1858,7 +1858,7 @@ The pre-refactor layout grew organically as the controller absorbed three legacy | Retired path | Replacement | |---|---| | `controller/src/algebra/expr.rs` | `controller/src/intent_algebra/legacy_expr.rs` | -| `controller/src/algebra/lower.rs` | `controller/src/intent_algebra/legacy_lower.rs` | +| `controller/src/algebra/lower.rs` | `controller/src/intent_algebra/legacy_to_canonical.rs` (the L2→L3 lowering, folded into the legacy→canonical converter) | | `controller/src/algebra/directory.rs` | `controller/src/physical/sketch_catalog.rs` | | `controller/src/algebra/physical.rs` | `controller/src/physical/planner.rs` | | `controller/src/algebra/allocator.rs` | `controller/src/physical/allocator.rs` | @@ -1885,6 +1885,6 @@ The pre-refactor layout grew organically as the controller absorbed three legacy **TODOs left from the refactor:** - `controller/src/emit/stage_config.rs` (3,020 lines, formerly `config/stage_config.rs`) was moved whole rather than split into `emit/opamp.rs` + `emit/streaming_config.rs` + `emit/inference_config.rs` per design.md §5. The monolith mixes OTel-collector YAML emit, ASAPQuery-backend JSON emit, storage-routing JSON emit, and shared internals; a clean split needs ownership reorganisation, not file renames. Tracked for a follow-up. -- `controller/src/intent_algebra/{legacy_expr,legacy_lower}.rs` carry the older `algebra::expr`-flavored `QueryExpr` / `AggIntent` / `WindowSpec` IR alongside the canonical `intent_algebra::{query_expr,agg_intent,lower}` types. Most of the controller still consumes the legacy types (query_parser, language_logical_plan, physical/, optimizer/, emit/); migrating each call site onto the canonical types is a follow-up. +- `controller/src/intent_algebra/legacy_expr.rs` carries the legacy **L2 relational** `QueryExpr` the `query_parser` front ends emit; `legacy_to_canonical.rs` is the single pass that converts it (sketch-fusion folded in) to the canonical `intent_algebra::{query_expr,agg_intent}` L3 types. The sketch-fused `SketchAgg` / `WindowedAgg` legacy variants and the standalone `legacy_lower` pass have been retired. Several controller modules (query_parser, language_logical_plan, physical/, optimizer/, emit/) still reference `legacy_expr` for shared leaf types; fully deleting the legacy tree is a follow-up gated on the canonical `Predicate` covering the remaining `ScalarExpr` variants. - `controller/src/optimizer/trait_def.rs` (`OptimizerRule`), `controller/src/emit/trait_def.rs` (`PlanEmitter`), `controller/src/deployment_model.rs` (`DeploymentModelRegistry`) ship as placeholders — the existing free-function emitters and concrete rule loops still drive behaviour. Migrating them onto the trait surfaces lands when the per-deployment-model crate split lands. diff --git a/control_plane/src/intent_algebra/binder.rs b/control_plane/src/intent_algebra/binder.rs index 1d2938b0..08db8415 100644 --- a/control_plane/src/intent_algebra/binder.rs +++ b/control_plane/src/intent_algebra/binder.rs @@ -50,7 +50,7 @@ //! legacy IR is retired it moves into the `core::lower` L1→L2→L3 passes //! proper (the `lower_*(ast, schema)` signatures in design.md §6). -use crate::intent_algebra::legacy_expr::{ColumnRef, QueryExpr as LQueryExpr}; +use crate::intent_algebra::legacy_expr::QueryExpr as LQueryExpr; use crate::intent_algebra::schema::{Column, DataType, Schema}; /// The DB / source-schema metadata source from design.md §6 "three @@ -166,19 +166,17 @@ fn default_leaf_columns() -> Vec { ] } -/// Walk the legacy tree and collect every distinct column / group-key -/// name the legacy → canonical converter resolves positionally: -/// `SketchAgg.col` / `WindowedAgg.col` (when `Named`), `Aggregate.keys`, +/// Walk the legacy tree and collect every distinct group-key name the +/// legacy → canonical converter resolves positionally: `Aggregate.keys`, /// `TopK.by`, and `Partition.keys`. Sorted + de-duplicated for a stable, /// deterministic column order. +/// +/// `AggItem.col` (the statistic's *input* column) is deliberately not +/// collected — the converter never resolves it positionally; it only +/// ever resolves group-by keys. fn collect_referenced_columns(tree: &LQueryExpr) -> Vec { let mut out: Vec = Vec::new(); tree.walk(&mut |node| match node { - LQueryExpr::SketchAgg { col, .. } | LQueryExpr::WindowedAgg { col, .. } => { - if let ColumnRef::Named(name) = col { - out.push(name.clone()); - } - } LQueryExpr::Aggregate { keys, .. } => out.extend(keys.iter().cloned()), LQueryExpr::TopK { by, .. } => out.extend(by.iter().cloned()), LQueryExpr::Partition { keys, .. } => out.extend(keys.keys().iter().cloned()), @@ -194,7 +192,6 @@ fn collect_referenced_columns(tree: &LQueryExpr) -> Vec { #[cfg(test)] mod tests { use super::*; - use crate::intent_algebra::agg_intent::AggIntent; use crate::intent_algebra::legacy_expr::{ AggFunc, AggItem, ColumnRef as LColumnRef, PartitionKeys, QueryExpr as LQueryExpr, SourceSpec, @@ -213,22 +210,6 @@ mod tests { assert_eq!(schema.time_index, Some(0)); } - #[test] - fn sketch_agg_named_col_lands_in_schema() { - // SketchAgg { col: Named("price") } over Source — "price" must be - // resolvable, i.e. present in the bound schema. - let tree = LQueryExpr::SketchAgg { - op: AggIntent::Sum, - col: LColumnRef::Named("price".into()), - input: Box::new(src("trades")), - }; - let schema = Binder::new().bind(&tree); - assert!(schema.column_id("price").is_some(), "price should be bound: {schema:?}"); - // floor still there. - assert!(schema.column_id("ts").is_some()); - assert!(schema.column_id("value").is_some()); - } - #[test] fn aggregate_keys_and_partition_keys_land_in_schema() { // Aggregate { keys: ["region"] } and Partition { By(["host"]) }. @@ -309,11 +290,7 @@ mod tests { } } } - let tree = LQueryExpr::SketchAgg { - op: AggIntent::Sum, - col: LColumnRef::Named("datacenter".into()), - input: Box::new(src("known_metric")), - }; + let tree = src("known_metric"); let schema = Binder::with_catalog(FixedCatalog).bind(&tree); // `datacenter` came from the catalog, not usage-synthesis — and it // is non-nullable, unlike a usage-derived column. diff --git a/control_plane/src/intent_algebra/legacy_expr.rs b/control_plane/src/intent_algebra/legacy_expr.rs index 087d77f2..3facfe01 100644 --- a/control_plane/src/intent_algebra/legacy_expr.rs +++ b/control_plane/src/intent_algebra/legacy_expr.rs @@ -57,12 +57,11 @@ use crate::types_v2::AccuracyTarget; // to `Aggregate { by, aggs }` is // Step γ's job.) // -// `SketchAgg.op` / `WindowedAgg.agg` carry canonical `AggIntent` directly; // PerPartition semantics ride on `PerPartitionWrap` (a thin legacy-only -// wrapper consumed by the four sites that still build it). Free helpers -// (`agg_to_legacy_agg_type`, `agg_is_mergeable`, etc.) mirror what the old -// `AggIntent::method()` API used to provide so the migration is a typed -// search-and-replace rather than a semantic rewrite. +// wrapper consumed by the `physical::sketch_catalog` per-partition sizing +// helpers). Free helpers (`agg_is_mergeable`, `agg_is_exact`, etc.) mirror +// what the old `AggIntent::method()` API used to provide so the migration +// is a typed search-and-replace rather than a semantic rewrite. /// Canonical L3 aggregation intent. Re-exported here so existing /// `legacy_expr::AggIntent` references keep working — the type is now the @@ -71,10 +70,9 @@ pub use crate::intent_algebra::agg_intent::AggIntent; /// Per-partition wrapper that historically lived on the legacy `AggIntent` /// enum as a `PerPartition { inner, keys }` variant. Canonical L3 represents -/// this shape via `QueryExpr::Aggregate { by: keys, aggs: [inner] }`, but -/// the legacy carriers (`SketchAgg` / `WindowedAgg`) still need an inline -/// place for `keys` until Step γ collapses them. This wrapper sits exactly -/// where the variant used to. +/// this shape via `QueryExpr::Aggregate { by: keys, aggs: [inner] }`; this +/// wrapper survives only as the input type for the +/// `physical::sketch_catalog` per-partition sizing helpers. #[derive(Debug, Clone, PartialEq)] pub struct PerPartitionWrap { pub inner: AggIntent, @@ -157,28 +155,11 @@ pub use crate::intent_algebra::agg_intent::{ pub use crate::sketch_algebra::capability::{countmin_accuracy, hll_accuracy}; -/// Unified window specification — captures all language-level window semantics. -#[derive(Debug, Clone, PartialEq)] -pub struct WindowSpec { - pub kind: WindowKind, - pub time_col: Option, -} - -#[derive(Debug, Clone, PartialEq)] -pub enum WindowKind { - /// Fixed-size, non-overlapping. - Tumbling { size: Duration }, - /// Fixed-size, overlapping (each sample belongs to ceil(size/slide) windows). - Sliding { size: Duration, slide: Duration }, - /// Gap-based: window closes after inactivity. - Session { gap: Duration }, -} -// Step γ7: the `Unbounded` / `Landmark` variants were removed — no -// producer ever constructed them (`legacy_lower` only builds -// `Tumbling` / `Sliding`; the parsers never emit a `WindowSpec` with -// either). They had no canonical `query_expr::WindowKind` equivalent, -// so dropping them retires dead code and makes the legacy → canonical -// `WindowKind` mapping total. +// The legacy `WindowSpec` / `WindowKind` types were removed alongside the +// `WindowedAgg` variant they fed. Layer-2 windows are carried by +// `QueryExpr::Window { duration, slide }` directly; window *kind* +// (Tumbling / Sliding / Session) is a canonical L3 concern — +// `query_expr::WindowKind`. /// A single filter predicate pushed down to the collector. #[derive(Debug, Clone)] @@ -216,36 +197,30 @@ pub enum FilterVal { // ── Relational algebra ──────────────────────────────────────────────────────── -/// Full relational + sketch algebra — the sole query IR. -/// -/// Every variant is a *node* in the logical query plan tree. Leaves are -/// [`QueryExpr::Source`] or [`QueryExpr::Ref`]. Interior nodes combine their -/// `input` child(ren) through the operator they implement. +/// The legacy **Layer-2 relational** query IR. /// -/// # Migration status (legacy_expr migration, Batch 2) +/// Every variant is a *node* in the per-language logical query plan tree +/// the `query_parser` front ends (`promql.rs` / `sql.rs`) emit. Leaves +/// are [`QueryExpr::Source`] or [`QueryExpr::Ref`]; interior nodes combine +/// their `input` child(ren) through the operator they implement. /// -/// The ten "A-classified" variants — `Filter`, `Project`, `Partition`, -/// `Distinct`, `Merge`, `Join`, `SetOp`, `Sort`, `Limit`, `BinaryOp` — -/// have canonical structural twins in -/// [`crate::intent_algebra::query_expr::QueryExpr`]. The canonical spelling -/// uses `child:` where these legacy variants use `input:`; the typed -/// [`crate::intent_algebra::Predicate`] replaces [`ScalarExpr`] in `Filter` -/// / `Join` / `Aggregate::having` (translation via -/// [`crate::intent_algebra::from_legacy_scalar`] for the four supported -/// scalar shapes). +/// # Relationship to the canonical L3 IR /// -/// These legacy variants stay here for now because: -/// 1. [`legacy_lower`] (Batch 13's target) reshape paths still construct -/// them internally. -/// 2. The legacy [`ScalarExpr`] retains four E-deferred variants -/// (`FunctionCall`, `ScalarSubquery`, `InList`, `Between`) that the -/// canonical `Predicate` doesn't cover yet — deleting the legacy -/// `Filter` would lose `ScalarExpr` expressiveness from consumers -/// that haven't migrated. +/// Most variants have canonical structural twins in +/// [`crate::intent_algebra::query_expr::QueryExpr`]. The canonical +/// spelling uses `child:` where these legacy variants use `input:`; the +/// typed [`crate::intent_algebra::Predicate`] replaces [`ScalarExpr`] in +/// `Filter` / `Join` / `Aggregate::having` (translation via +/// [`crate::intent_algebra::from_legacy_scalar`]). /// -/// Consumer-side redirect (legacy → canonical, with `input:` → `child:` and -/// `ScalarExpr` → `Predicate`) lands in subsequent batches once Batch 2's -/// additive lift is in place. +/// This is purely a Layer-2 *relational* IR — the sketch-fused +/// `SketchAgg` / `WindowedAgg` variants were removed once the +/// `legacy_to_canonical` converter learned to fold the single-statistic +/// sketchable `Aggregate` straight into canonical shapes. The remaining +/// reason the tree is not yet *deleted* outright is that the legacy +/// [`ScalarExpr`] still carries four variants (`FunctionCall`, +/// `ScalarSubquery`, `InList`, `Between`) the canonical `Predicate` +/// doesn't cover, and the parsers build `ScalarExpr` directly. #[derive(Debug, Clone)] pub enum QueryExpr { // ── Base relations ──────────────────────────────────────────────────── @@ -293,25 +268,11 @@ pub enum QueryExpr { input: Box, }, - /// γ+α specialisation for sketch aggregations (single sketch per node). - /// - /// Kept separate from [`Self::Aggregate`] so the allocator can reason - /// about which sketch type to use without parsing `AggFunc` variants. - SketchAgg { - op: AggIntent, - col: ColumnRef, - input: Box, - }, - - /// Core sketch algebra operator: windowed aggregation intent. - /// Bundles the window and the aggregation because in sketch systems - /// the window defines the sketch lifecycle (when to flush/reset). - WindowedAgg { - agg: AggIntent, - window: WindowSpec, - col: ColumnRef, - input: Box, - }, + // The sketch-fused `SketchAgg` / `WindowedAgg` variants were removed: + // they were never parser output — only an intermediate the old + // `legacy_lower` pass produced — and the legacy → canonical converter + // now folds the single-statistic sketchable `Aggregate` straight into + // canonical shapes (`Aggregate { by: [] }` / `Window { Aggregate }`). // ── Distributed / multi-stage operators ────────────────────────────── @@ -589,8 +550,6 @@ impl QueryExpr { QueryExpr::Filter { input, .. } | QueryExpr::Project { input, .. } | QueryExpr::Window { input, .. } - | QueryExpr::SketchAgg { input, .. } - | QueryExpr::WindowedAgg { input, .. } | QueryExpr::Partition { input, .. } | QueryExpr::Distinct { input, .. } | QueryExpr::TopK { input, .. } @@ -616,12 +575,16 @@ impl QueryExpr { } } - /// Returns `true` when the sub-tree contains at least one [`QueryExpr::SketchAgg`] - /// or [`QueryExpr::TopK`] node (i.e. sketch work is present). + /// Returns `true` when the sub-tree contains at least one + /// [`QueryExpr::TopK`] node — the only Layer-2 variant that names a + /// heavy-hitter sketch directly. (Single-statistic sketchable + /// `Aggregate`s are recognised as sketch work only once the converter + /// folds them; at Layer 2 they are indistinguishable from exact + /// aggregates.) pub fn has_sketch_work(&self) -> bool { let mut found = false; self.walk(&mut |n| { - if matches!(n, QueryExpr::SketchAgg { .. } | QueryExpr::WindowedAgg { .. } | QueryExpr::TopK { .. }) { + if matches!(n, QueryExpr::TopK { .. }) { found = true; } }); @@ -635,8 +598,6 @@ impl QueryExpr { QueryExpr::Filter { input, .. } | QueryExpr::Project { input, .. } | QueryExpr::Window { input, .. } - | QueryExpr::SketchAgg { input, .. } - | QueryExpr::WindowedAgg { input, .. } | QueryExpr::Partition { input, .. } | QueryExpr::Distinct { input, .. } | QueryExpr::TopK { input, .. } @@ -744,10 +705,10 @@ mod tests { // ── has_sketch_work ─────────────────────────────────────────────────────── #[test] - fn has_sketch_work_true_when_ddsketch_present() { - let qe = QueryExpr::SketchAgg { - op: default_quantile(0.5), - col: ColumnRef::SampleValue, + fn has_sketch_work_true_when_topk_present() { + let qe = QueryExpr::TopK { + k: 10, + by: vec![], input: Box::new(src("m")), }; assert!(qe.has_sketch_work()); @@ -847,7 +808,7 @@ mod tests { #[test] fn complex_nested_tree() { - // TopK(10, Partition(symbol, Window(5m, SketchAgg(CountSketch, Source(price))))) + // TopK(10, Partition(symbol, Window(5m, Aggregate(Count, Source(price))))) let qe = QueryExpr::TopK { k: 10, by: vec![], @@ -856,10 +817,16 @@ mod tests { input: Box::new(QueryExpr::Window { duration: Duration::from_secs(300), slide: None, - input: Box::new(QueryExpr::SketchAgg { - op: default_frequency(), - col: ColumnRef::Wildcard, - input: Box::new(src("price")), + input: Box::new(QueryExpr::Aggregate { + keys: vec![], + aggs: vec![AggItem { + alias: "c".into(), + func: AggFunc::Count, + col: ColumnRef::Wildcard, + distinct: false, + }], + having: None, + input: Box::new(src("price")), }), }), }), diff --git a/control_plane/src/intent_algebra/legacy_to_canonical.rs b/control_plane/src/intent_algebra/legacy_to_canonical.rs index f67e9056..5df062dd 100644 --- a/control_plane/src/intent_algebra/legacy_to_canonical.rs +++ b/control_plane/src/intent_algebra/legacy_to_canonical.rs @@ -1,24 +1,9 @@ -//! Step γ7 keystone: composable legacy → canonical L3 IR converter. +//! The legacy Layer-2 → canonical L3 IR converter. //! -//! Recursively converts a *whole* `legacy_expr::QueryExpr` tree into a -//! *whole* `query_expr::QueryExpr` tree. This is the single piece of -//! migration machinery the consumer-flip PRs build on: once every -//! producer routes its output through [`convert_root`], the consumer -//! migrations are pure pattern-match rewrites onto the canonical IR and -//! the legacy IR can be deleted. -//! -//! ## Relationship to the γ1–γ4 bridges -//! -//! The four one-way bridges (`aggregate_bridge`, `sketch_agg_bridge`, -//! `windowed_agg_bridge`, `topk_bridge`) are *approach (c)* helpers: -//! they return canonical-shape data *minus the child*, so a consumer can -//! pattern-match a single legacy node without converting the subtree. -//! That makes them non-composable — you cannot chain them into a full -//! tree conversion. This module absorbs their per-variant logic into a -//! recursive converter that *does* carry the child, and is therefore the -//! thing producers and consumers actually pivot on. The bridges stay in -//! place until their remaining annotation-only consumers migrate; PR 13 -//! deletes them. +//! Recursively converts a *whole* `legacy_expr::QueryExpr` tree (the raw +//! Layer-2 relational IR the `query_parser` front ends emit) into a +//! *whole* canonical `query_expr::QueryExpr` tree. This is the single +//! entry the parse path routes through — [`convert_root`]. //! //! ## Variant mapping //! @@ -28,10 +13,9 @@ //! | `Ref(name)` | `Ref { name }` | //! | `Filter` | `Filter` (pred via [`convert_scalar`]) | //! | `Project` | `Project` (each item's expr via [`convert_scalar`])| -//! | `Aggregate` | `Aggregate` (keys→by, AggFunc→AggIntent, having) | +//! | `Aggregate` (multi-agg / HAVING / `Custom` / un-grouped `COUNT(*)`) | plain `Aggregate` (keys→by, AggFunc→AggIntent) | +//! | `Aggregate` (single sketchable) | *fuses* — `Window`-input → `Window { Aggregate }`, `GROUP BY` → wrapping `Partition`, `StdDev`/`Variance` → `Merge` of sibling quantile aggregates. See the `Aggregate` arm. | //! | `Window` | `Window` (slide → Sliding else Tumbling) | -//! | `SketchAgg` | `Aggregate { by: [col], aggs: [op] }` | -//! | `WindowedAgg` | `Window { child: Aggregate { .. } }` | //! | `Partition` | `Partition` | //! | `Distinct` | `Distinct` | //! | `TopK` | `Aggregate { aggs: [AggIntent::TopK] }` (HeavyHitter)| @@ -44,17 +28,22 @@ //! | `PromQLSubquery` | `Subquery` | //! | `BinaryOp` | `BinaryOp` | //! +//! The single-statistic sketchable `Aggregate` fusion was, before the +//! legacy-IR retirement, a separate `legacy_lower` pass that produced +//! intermediate `SketchAgg` / `WindowedAgg` legacy Layer-3 nodes. Those +//! variants are gone — the fusion is now done directly in canonical +//! terms inside the [`convert`] `Aggregate` arm. +//! //! ## Schema threading //! //! [`convert`] takes a `&Schema` — the schema in scope at the node — and -//! threads it unchanged to every child, matching the established -//! `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 -//! also doesn't handle — proper bottom-up schema flow lands with the -//! canonical `output_schema_in` wiring downstream. +//! threads it unchanged to every child. The [`Binder`]-built schema is +//! complete and self-contained (`(ts, value)` plus every referenced +//! name), so threading the root schema down is correct except for the +//! nested-schema-transform case (an `Aggregate` below another +//! `Aggregate`), which the legacy stack also doesn't handle — proper +//! bottom-up schema flow lands with the canonical `output_schema_in` +//! wiring downstream. #![allow(dead_code)] @@ -68,8 +57,8 @@ use crate::intent_algebra::column_resolution::{ resolve_named_keys, ResolveError, }; use crate::intent_algebra::legacy_expr::{ - AggFunc, AggItem, ColumnRef as LColumnRef, PartitionKeys as LPartitionKeys, - QueryExpr as LQueryExpr, ScalarExpr as LScalarExpr, WindowKind as LWindowKind, WindowSpec, + AggFunc, ColumnRef as LColumnRef, PartitionKeys as LPartitionKeys, QueryExpr as LQueryExpr, + ScalarExpr as LScalarExpr, }; use crate::intent_algebra::query_expr::{ from_legacy_scalar, ColumnRef as CColumnRef, HavingPredicate, LiteralValue, @@ -86,9 +75,8 @@ use crate::types_v2::{AccuracyTarget, BindingName}; /// `matches!(err, ConvertError::Variant { .. })`. #[derive(Debug, Error)] pub enum ConvertError { - /// A column reference (`Aggregate` key, `SketchAgg` / `WindowedAgg` - /// column, `TopK` partition key) did not resolve against the - /// inherited schema. + /// A column reference (`Aggregate` key, `Partition` / `TopK` key) + /// did not resolve against the inherited schema. #[error("column resolution failed: {0}")] Resolve(#[from] ResolveError), /// An `AggItem.func` has no canonical `AggIntent` equivalent — only @@ -102,20 +90,13 @@ pub enum ConvertError { Scalar(QueryExprError), } -/// Lower a legacy `QueryExpr` tree all the way to canonical. -/// -/// Two internal walks, one public entry: +/// Lower a legacy Layer-2 `QueryExpr` tree to the canonical L3 IR. /// -/// 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. +/// A single recursive walk ([`convert`]): every Layer-2 relational node +/// maps onto its canonical counterpart, and the single-statistic +/// sketchable `Aggregate` fuses (window-swap, `Partition` wrap, `StdDev` +/// fan-out) directly in canonical terms — see the [`convert`] `Aggregate` +/// arm. There is no intermediate legacy Layer-3 IR. /// /// The inherited schema comes from the [`Binder`] — the explicit L3 /// name-resolution pass — which builds the complete, self-contained @@ -124,9 +105,8 @@ 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 lowered = lower_to_sketch_algebra(legacy.clone()); - let schema = Binder::new().bind(&lowered); - convert(&lowered, &schema) + let schema = Binder::new().bind(legacy); + convert(legacy, &schema) } /// Convert a legacy `QueryExpr` tree to canonical against an explicit @@ -175,17 +155,103 @@ pub fn convert(legacy: &LQueryExpr, schema: &Schema) -> Result { + // A single-statistic sketchable aggregate *fuses* — this is the + // former `legacy_lower::lower_aggregate` step, done directly in + // canonical terms rather than via an intermediate legacy L3 + // node: + // * input is a `Window` → emit `Window { Aggregate { by: [] } }` + // (the window-defines-sketch-lifecycle shape); + // * otherwise → emit `Aggregate { by: [] }`; + // * `StdDev` / `Variance` fan out into a `Merge` of two + // sibling quantile aggregates (Step α F1 strategy); + // * `GROUP BY` keys wrap the result in a `Partition`. + // + // Matching the *raw* `input` for `Window` is exact for every + // tree the parsers emit: they never nest an `Aggregate` + // directly over a `Window` directly over another sketchable + // `Aggregate`, the only shape where the raw vs. sketch-lowered + // input would differ. + // + // Multi-agg, `HAVING`-bearing, `Custom`, and un-grouped + // `COUNT(*)` aggregates fall through to the plain canonical + // `Aggregate` below. + if aggs.len() == 1 && having.is_none() { + let item = &aggs[0]; + let ungrouped_count = + matches!(item.func, AggFunc::Count) && keys.is_empty(); + if !ungrouped_count { + let intents = agg_func_to_intents(&item.func); + if !intents.is_empty() { + let nodes: Vec = match input.as_ref() { + LQueryExpr::Window { + duration, + slide, + input: win_input, + } => { + let kind = if slide.is_some() { + CWindowKind::Sliding + } else { + CWindowKind::Tumbling + }; + let win_child = convert(win_input, schema)?; + intents + .into_iter() + .map(|intent| CQueryExpr::Window { + kind: kind.clone(), + size: *duration, + slide: *slide, + child: Box::new(CQueryExpr::Aggregate { + by: Vec::new(), + aggs: vec![intent], + having: None, + child: Box::new(win_child.clone()), + }), + }) + .collect() + } + other => { + let child = convert(other, schema)?; + intents + .into_iter() + .map(|intent| CQueryExpr::Aggregate { + by: Vec::new(), + aggs: vec![intent], + having: None, + child: Box::new(child.clone()), + }) + .collect() + } + }; + let sketch = if nodes.len() == 1 { + nodes.into_iter().next().unwrap() + } else { + CQueryExpr::Merge { children: nodes } + }; + return Ok(if keys.is_empty() { + sketch + } else { + CQueryExpr::Partition { + keys: CPartitionKeys::By(keys.clone()), + child: Box::new(sketch), + } + }); + } + // `intents` empty → `Custom` func; fall through to the + // plain path, which raises `NoCanonicalIntent`. + } + } + + // Plain canonical `Aggregate`: multi-agg, `HAVING`, `Custom`, + // or the un-grouped `COUNT(*)` exact-row-count case. let by = resolve_named_keys(keys, schema)?; let mut intents: Vec = Vec::with_capacity(aggs.len()); for item in aggs { - // Faithful mapping for un-grouped `COUNT(*)`: `legacy_lower` - // deliberately leaves it un-lowered (no GROUP BY → exact - // row count, no sketch benefit). `agg_func_to_intents` is - // the *lowering* map and would pick the `Frequency` sketch - // — that loses the "this is exact" decision. Map it to - // `Count { Exact }` so downstream (`capability_for`, the - // `QeCollector`) sees it as exact, matching the legacy - // `Aggregate { AggItem { Count }, keys: [] }` semantics. + // Faithful mapping for un-grouped `COUNT(*)`: no GROUP BY + // means an exact row count, no sketch benefit. + // `agg_func_to_intents` is the *sketch* map and would pick + // `Frequency` — that loses the "this is exact" decision. Map + // it to `Count { Exact }` so downstream (`capability_for`, + // the `QeCollector`) sees it as exact. if matches!(item.func, AggFunc::Count) && keys.is_empty() { intents.push(AggIntent::Count { accuracy: AccuracyTarget::Exact, @@ -228,43 +294,6 @@ pub fn convert(legacy: &LQueryExpr, schema: &Schema) -> Result { - // `SketchAgg.col` is the sketch's *input* column, not a - // GROUP BY key — canonical `Aggregate.by` is the group-by - // tuple, which for a `SketchAgg` is always empty (grouping - // rides on the wrapping `Partition` node). The sketch input - // is the canonical implicit `value` column, so `col` carries - // no information the canonical IR needs. - CQueryExpr::Aggregate { - by: Vec::new(), - aggs: vec![op.clone()], - having: None, - child: Box::new(convert(input, schema)?), - } - } - - LQueryExpr::WindowedAgg { - agg, - window, - col: _, - input, - } => { - let (kind, size, slide) = map_window_kind(window); - // As with `SketchAgg`: `col` is the sketch input column, not - // a GROUP BY key — the inner canonical `Aggregate.by` is empty. - CQueryExpr::Window { - kind, - size, - slide, - child: Box::new(CQueryExpr::Aggregate { - by: Vec::new(), - aggs: vec![agg.clone()], - having: None, - child: Box::new(convert(input, schema)?), - }), - } - } - LQueryExpr::Partition { keys, input } => CQueryExpr::Partition { keys: match keys { LPartitionKeys::By(k) => CPartitionKeys::By(k.clone()), @@ -432,19 +461,6 @@ pub fn convert_scalar(se: &LScalarExpr, schema: &Schema) -> Result (CWindowKind, Duration, Option) { - match &window.kind { - LWindowKind::Tumbling { size } => (CWindowKind::Tumbling, *size, None), - LWindowKind::Sliding { size, slide } => { - (CWindowKind::Sliding, *size, Some(*slide)) - } - LWindowKind::Session { gap } => (CWindowKind::Session, *gap, None), - } -} - /// Legacy `ColumnRef` → canonical `ColumnRef`. The two enums have /// identical variants; the canonical one is what L3 nodes carry. fn convert_column_ref(c: &LColumnRef) -> CColumnRef { @@ -455,281 +471,13 @@ 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, - } - }; +// ── AggFunc → AggIntent sketch mapping ─────────────────────────────────────── - 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). +/// Map an [`AggFunc`] to the canonical [`AggIntent`]s the `convert` +/// `Aggregate` arm fuses on. 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 sketch aggregates). fn agg_func_to_intents(func: &AggFunc) -> Vec { use crate::intent_algebra::legacy_expr::{ default_cardinality, default_frequency, default_quantile, @@ -867,18 +615,18 @@ mod tests { } #[test] - fn sketch_agg_folds_into_aggregate() { - // SketchAgg { Sum, SampleValue, Source } → Aggregate { by: [], [Sum] }. - // `col` is the sketch *input*, not a GROUP BY key — `by` is empty - // (grouping rides on a wrapping `Partition`, not the SketchAgg). - let legacy = LQueryExpr::SketchAgg { - op: AggIntent::Sum, - col: LColumnRef::SampleValue, + fn single_sketchable_aggregate_folds_to_canonical_aggregate() { + // A single-statistic sketchable `Aggregate` over a non-`Window` + // input folds to a canonical `Aggregate { by: [], aggs: [intent] }`. + let legacy = LQueryExpr::Aggregate { + keys: vec![], + aggs: vec![agg_item("s", AggFunc::Sum)], + having: None, input: Box::new(src("m")), }; match convert_root(&legacy).unwrap() { CQueryExpr::Aggregate { by, aggs, .. } => { - assert!(by.is_empty(), "SketchAgg.col is not a group-by key: {by:?}"); + assert!(by.is_empty(), "no GROUP BY → empty `by`: {by:?}"); assert!(matches!(aggs.as_slice(), [AggIntent::Sum])); } other => panic!("expected Aggregate, got {other:?}"), @@ -886,11 +634,19 @@ mod tests { } #[test] - fn sketch_agg_named_col_still_has_empty_by() { - // Even a `Named` sketch-target column is not a group-by key. - let legacy = LQueryExpr::SketchAgg { - op: AggIntent::Sum, - col: LColumnRef::Named("price".into()), + fn aggregate_target_column_is_not_a_group_by_key() { + // The `AggItem.col` (the statistic's input column) is *not* a + // GROUP BY key — only `Aggregate.keys` is. A `Named` agg-target + // column therefore leaves the canonical `by` empty. + let legacy = LQueryExpr::Aggregate { + keys: vec![], + aggs: vec![AggItem { + alias: "s".into(), + func: AggFunc::Sum, + col: LColumnRef::Named("price".into()), + distinct: false, + }], + having: None, input: Box::new(src("trades")), }; match convert_root(&legacy).unwrap() { @@ -900,28 +656,28 @@ mod tests { } #[test] - fn windowed_agg_folds_into_window_over_aggregate() { - let legacy = LQueryExpr::WindowedAgg { - agg: AggIntent::Quantile { - q: 0.99, - accuracy: AccuracyTarget::Epsilon(0.01), - }, - window: WindowSpec { - kind: LWindowKind::Tumbling { - size: Duration::from_secs(300), - }, - time_col: None, - }, - col: LColumnRef::SampleValue, - input: Box::new(src("m")), + fn single_sketchable_aggregate_over_window_folds_to_window_over_aggregate() { + // A single-statistic sketchable `Aggregate` whose input is a + // `Window` folds to the `Window { Aggregate { by: [] } }` shape — + // the window-defines-sketch-lifecycle form. + let legacy = LQueryExpr::Aggregate { + keys: vec![], + aggs: vec![agg_item("q", AggFunc::Quantile(0.99))], + having: None, + input: Box::new(LQueryExpr::Window { + duration: Duration::from_secs(300), + slide: None, + input: Box::new(src("m")), + }), }; match convert_root(&legacy).unwrap() { CQueryExpr::Window { kind, child, .. } => { assert_eq!(kind, CWindowKind::Tumbling); assert!(matches!( *child, - CQueryExpr::Aggregate { ref aggs, .. } - if matches!(aggs.as_slice(), [AggIntent::Quantile { .. }]) + CQueryExpr::Aggregate { ref by, ref aggs, .. } + if by.is_empty() + && matches!(aggs.as_slice(), [AggIntent::Quantile { .. }]) )); } other => panic!("expected Window, got {other:?}"), @@ -1087,256 +843,3 @@ 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/physical/window_fusion.rs b/control_plane/src/physical/window_fusion.rs index 6ed763c1..a41b9c35 100644 --- a/control_plane/src/physical/window_fusion.rs +++ b/control_plane/src/physical/window_fusion.rs @@ -1,29 +1,26 @@ -//! Step γ7 safety net: the `Window { child: Aggregate }` fusion -//! recognizer. +//! The `Window { child: Aggregate }` fusion recognizer. //! //! ## The invariant this protects //! -//! The legacy IR fuses the window and the sketch aggregation into a -//! single `legacy_expr::QueryExpr::WindowedAgg` node, because in sketch -//! systems the window defines the sketch's lifecycle (when to flush / -//! reset). The legacy `physical::planner::plan_node` `WindowedAgg` arm -//! relies on that fused shape: it resolves the window into the *same* -//! `PhysicalOp::OtelSketchBuild { window }` node as the sketch build. +//! In sketch systems the window defines the sketch's lifecycle (when to +//! flush / reset), so a windowed sketch aggregation must be planned as a +//! *single* fused physical node — `PhysicalOp::OtelSketchBuild { window }` +//! — where the window and the sketch build are resolved together. //! -//! The canonical L3 IR has no `WindowedAgg` — `legacy_to_canonical` -//! folds it into the stacked `Window { child: Aggregate { .. } }` shape -//! (design.md §6: "one canonical form per plan"). That moves the -//! window-defines-sketch-lifecycle invariant out of the *IR shape* and -//! into a planner-side **peephole recognizer** — this module. +//! The canonical L3 IR keeps "one canonical form per plan" (design.md +//! §6): it has no fused windowed-aggregate variant — a windowed sketch is +//! the stacked `Window { child: Aggregate { aggs: [one], .. } }` shape. +//! `legacy_to_canonical` produces exactly that shape when it folds a +//! single-statistic sketchable `Aggregate` sitting over a `Window`. This +//! module is the planner-side **peephole recognizer** that puts the +//! window-defines-sketch-lifecycle invariant back: it matches the stacked +//! shape and reconstructs the fused physical node. //! //! [`recognize_windowed_sketch`] matches the stacked canonical shape and //! returns a [`FusedWindowSketch`] view; [`fused_sketch_decision`] -//! reconstructs the exact `(PhysicalOp, Placement)` the legacy -//! `WindowedAgg` arm produces. Nothing here is on the hot path yet — PR -//! 8 wires it into the canonical `plan_node`. The `#[cfg(test)]` -//! equivalence harness below pins the reconstruction against the live -//! legacy `WindowedAgg` arm *while that arm is still the source of -//! truth*, so the invariant is proven before any consumer flips. +//! reconstructs the `(PhysicalOp, Placement)` for the fused sketch build. +//! It is on the canonical planner's hot path; the `#[cfg(test)]` +//! equivalence harness below pins it end-to-end against `plan`. #![allow(dead_code)] @@ -170,22 +167,23 @@ pub fn fused_sketch_decision( (op, placement) } + // ── Equivalence harness ────────────────────────────────────────────────────── // // Pins `recognize_windowed_sketch` + `fused_sketch_decision` against the -// live legacy `WindowedAgg` arm of `physical::planner::plan_node`. While -// the legacy arm is still the source of truth (it is, until PR 8), this -// harness proves the canonical recognizer reconstructs exactly the same -// physical decision — the window-defines-sketch-lifecycle invariant -// survives the un-fused canonical IR shape. +// canonical planner: a `Window { Aggregate }` fused sketch — the shape +// `legacy_to_canonical` folds a single-statistic sketchable `Aggregate` +// over a `Window` into — produces a resolved (non-`None`) physical window, +// and the planner's own decision matches `fused_sketch_decision`. #[cfg(test)] mod tests { use super::*; - use crate::intent_algebra::convert_root; use crate::intent_algebra::legacy_expr::{ - default_cardinality, default_frequency, default_quantile, ColumnRef as LColumnRef, - QueryExpr as LQueryExpr, SourceSpec, WindowKind as LWindowKind, WindowSpec, + AggFunc, AggItem, ColumnRef as LColumnRef, QueryExpr as LQueryExpr, SourceSpec, + }; + use crate::intent_algebra::{ + convert_root, default_cardinality, default_frequency, default_quantile, }; use crate::optimizer::engine::DeploymentConstraints; use crate::physical::planner::{plan, PhysicalOp}; @@ -198,61 +196,51 @@ mod tests { } } - /// Build a legacy `WindowedAgg { agg, window, SampleValue, Source }`. - fn legacy_windowed_agg(agg: AggIntent, window: WindowSpec) -> LQueryExpr { - LQueryExpr::WindowedAgg { - agg, - window, - col: LColumnRef::SampleValue, - input: Box::new(LQueryExpr::Source(SourceSpec { name: "m".into() })), - } - } - - fn tumbling(secs: u64) -> WindowSpec { - WindowSpec { - kind: LWindowKind::Tumbling { - size: Duration::from_secs(secs), - }, - time_col: None, - } - } - - fn sliding(size_secs: u64, slide_secs: u64) -> WindowSpec { - WindowSpec { - kind: LWindowKind::Sliding { - size: Duration::from_secs(size_secs), - slide: Duration::from_secs(slide_secs), - }, - time_col: None, - } + /// A canonical `Scan` leaf — built through `convert_root` so it carries + /// the same Binder-built schema a real converted tree would. + fn canonical_scan(metric: &str) -> QueryExpr { + convert_root(&LQueryExpr::Source(SourceSpec { + name: metric.into(), + })) + .expect("convert source") } - fn session(gap_secs: u64) -> WindowSpec { - WindowSpec { - kind: LWindowKind::Session { - gap: Duration::from_secs(gap_secs), - }, - time_col: None, + /// The canonical `Window { Aggregate { by: [], aggs: [agg] } }` shape — + /// the fold `legacy_to_canonical` produces for a windowed single-sketch + /// aggregate. + fn windowed_sketch( + agg: AggIntent, + kind: WindowKind, + size: Duration, + slide: Option, + ) -> QueryExpr { + QueryExpr::Window { + kind, + size, + slide, + child: Box::new(QueryExpr::Aggregate { + by: Vec::new(), + aggs: vec![agg], + having: None, + child: Box::new(canonical_scan("m")), + }), } } - /// End-to-end fusion assertion: a legacy `WindowedAgg`, converted to - /// the canonical `Window { Aggregate }` shape and run through the - /// canonical planner, produces a *fused* `OtelSketchBuild` carrying a - /// resolved (non-`None`) physical window. This is the - /// window-defines-sketch-lifecycle invariant — now a planner peephole - /// (`recognize_windowed_sketch` + `fused_sketch_decision`, both wired - /// onto the hot path by PR 8) rather than an IR-shape property. - /// - /// Before PR 8 this harness compared the recognizer against the live - /// legacy `WindowedAgg` planner arm; that arm is now gone (the planner - /// is canonical), so the assertion is the canonical end-to-end result. - fn assert_equivalent(agg: AggIntent, window: WindowSpec) { + /// End-to-end fusion assertion: the canonical `Window { Aggregate }` + /// shape, run through the recognizer + `fused_sketch_decision` and + /// through the canonical planner, produces a *fused* `OtelSketchBuild` + /// carrying a resolved (non-`None`) physical window — the + /// window-defines-sketch-lifecycle invariant as a planner peephole. + fn assert_equivalent( + agg: AggIntent, + kind: WindowKind, + size: Duration, + slide: Option, + ) { let cfg = config(); - let legacy = legacy_windowed_agg(agg, window); - let canonical = convert_root(&legacy).expect("convert"); + let canonical = windowed_sketch(agg, kind, size, slide); - // The recognizer matches the converted `Window { Aggregate }` fold. let fused = recognize_windowed_sketch(&canonical) .expect("canonical Window{Aggregate} should be recognized as a fused sketch"); let (decided_op, _placement) = fused_sketch_decision(&fused, &cfg); @@ -288,50 +276,80 @@ mod tests { #[test] fn equivalence_tumbling_quantile() { - assert_equivalent(default_quantile(0.99), tumbling(300)); + assert_equivalent( + default_quantile(0.99), + WindowKind::Tumbling, + Duration::from_secs(300), + None, + ); } #[test] fn equivalence_sliding_quantile() { - assert_equivalent(default_quantile(0.5), sliding(600, 60)); + assert_equivalent( + default_quantile(0.5), + WindowKind::Sliding, + Duration::from_secs(600), + Some(Duration::from_secs(60)), + ); } #[test] fn equivalence_session_quantile() { - assert_equivalent(default_quantile(0.95), session(30)); + assert_equivalent( + default_quantile(0.95), + WindowKind::Session, + Duration::from_secs(30), + None, + ); } #[test] fn equivalence_tumbling_cardinality() { - assert_equivalent(default_cardinality(), tumbling(60)); + assert_equivalent( + default_cardinality(), + WindowKind::Tumbling, + Duration::from_secs(60), + None, + ); } #[test] fn equivalence_tumbling_frequency() { - assert_equivalent(default_frequency(), tumbling(120)); + assert_equivalent( + default_frequency(), + WindowKind::Tumbling, + Duration::from_secs(120), + None, + ); } #[test] fn equivalence_sum_intent() { - assert_equivalent(AggIntent::Sum, tumbling(300)); + assert_equivalent( + AggIntent::Sum, + WindowKind::Tumbling, + Duration::from_secs(300), + None, + ); } #[test] fn recognizer_rejects_bare_aggregate() { - // A canonical Aggregate with no enclosing Window is the unfused - // SketchAgg case — not a windowed sketch. - let legacy = LQueryExpr::SketchAgg { - op: AggIntent::Sum, - col: LColumnRef::SampleValue, - input: Box::new(LQueryExpr::Source(SourceSpec { name: "m".into() })), + // A canonical `Aggregate` with no enclosing `Window` is the unfused + // sketch case — not a windowed sketch. + let canonical = QueryExpr::Aggregate { + by: Vec::new(), + aggs: vec![AggIntent::Sum], + having: None, + child: Box::new(canonical_scan("m")), }; - let canonical = convert_root(&legacy).expect("convert"); assert!(recognize_windowed_sketch(&canonical).is_none()); } #[test] fn recognizer_rejects_window_over_non_aggregate() { - // Window directly over a Scan — no inner Aggregate to fuse. + // `Window` directly over a `Scan` — no inner `Aggregate` to fuse. let legacy = LQueryExpr::Window { duration: Duration::from_secs(60), slide: None, @@ -343,9 +361,24 @@ mod tests { #[test] fn recognizer_accepts_converted_windowed_agg() { - // Sanity: the shape legacy_to_canonical folds a WindowedAgg into - // is exactly what the recognizer matches. - let legacy = legacy_windowed_agg(default_quantile(0.99), tumbling(300)); + // The L2 `Aggregate { Quantile } over Window` shape the parsers + // emit converts to canonical `Window { Aggregate }` — exactly the + // shape the recognizer matches. + let legacy = LQueryExpr::Aggregate { + keys: vec![], + aggs: vec![AggItem { + alias: "q".into(), + func: AggFunc::Quantile(0.99), + col: LColumnRef::SampleValue, + distinct: false, + }], + having: None, + input: Box::new(LQueryExpr::Window { + duration: Duration::from_secs(300), + slide: None, + input: Box::new(LQueryExpr::Source(SourceSpec { name: "m".into() })), + }), + }; let canonical = convert_root(&legacy).expect("convert"); let fused = recognize_windowed_sketch(&canonical).expect("should recognize"); assert_eq!(fused.window_kind, WindowKind::Tumbling);