From e1fe71ba4de55d7593304b2b85fa229b4c4db8db Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 14 May 2026 15:37:43 -0600 Subject: [PATCH] refactor(intent_algebra): reconcile the two L3-construction paths onto lower_to_canonical MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit There were two ways to build a canonical L3 `QueryExpr`: * `convert_root` — the real path: relational L2 tree → canonical L3, via `lower_to_canonical`. Faithful (window-swap, Partition, etc.). * `lower_parsed_query` (`intent_algebra/lower.rs`) — reconstructed a canonical tree from the *flat* `ParsedQuery` summary. Lossy: it dropped exact aggregates entirely (no `Aggregate` node for `rate`/`sum`/`increase`/bare selectors) and threaded an explicit accuracy the real path doesn't take. Only `asap_tier_analysis` and one `sketch_algebra` test helper used the lossy path. Both are switched to `parse_query_expr_canonical` (the real path) and `lower_parsed_query` + its module are deleted. **Behavior change** (intended — see PR discussion): on the real path, `rate`/`sum`/`increase`/bare-selector lower to `AggIntent::Sum`, and `capability_for(&Sum)` returns `Capability::ExactAgg(Sum)` — so they are now ASAP-tier-answerable from exact-precompute state instead of unconditionally archive-routed. `lower_parsed_query` was *masking* the `ExactAgg` capability by dropping those intents. The 7 `asap_tier_analysis` tests that pinned the old "route to archive" behavior are rewritten to pin the new ExactAgg routing. **L4 binder fix**: feeding `sketch_algebra::bind_query_expr` the real canonical IR exposed that its bottom-up walk didn't recurse into `Window` — so `Window { Aggregate }` (the canonical windowed-sketch shape) never reached the `Bind*` rules. `bind_recursive` now recognizes `Window { Aggregate { .. } }`, pushes the window under the aggregate, and re-dispatches — the window rides along inside the bound node's `Logical(..)` child, exactly as it did when the aggregate sat on top. Finally: with the old `intent_algebra/lower.rs` gone, the name is free, so `lower_to_canonical.rs` → `lower.rs` — `intent_algebra::lower` is now the single L2→L3 lowering, matching design.md's `core::lower`. Co-Authored-By: Claude Opus 4.7 (1M context) --- control_plane/docs/design.md | 16 +- control_plane/src/asap_tier_analysis.rs | 119 +- control_plane/src/intent_algebra/binder.rs | 4 +- .../src/intent_algebra/column_resolution.rs | 2 +- control_plane/src/intent_algebra/lower.rs | 1068 ++++++++++++----- .../src/intent_algebra/lower_to_canonical.rs | 844 ------------- control_plane/src/intent_algebra/mod.rs | 8 +- .../src/intent_algebra/relational.rs | 6 +- control_plane/src/lib.rs | 2 +- control_plane/src/physical/window_fusion.rs | 8 +- control_plane/src/sketch_algebra/lower.rs | 42 +- control_plane/src/sketch_algebra/tests.rs | 16 +- 12 files changed, 910 insertions(+), 1225 deletions(-) delete mode 100644 control_plane/src/intent_algebra/lower_to_canonical.rs diff --git a/control_plane/docs/design.md b/control_plane/docs/design.md index b2771b84..2b55018c 100644 --- a/control_plane/docs/design.md +++ b/control_plane/docs/design.md @@ -45,7 +45,7 @@ DataCollector/controller already documents its query→sketch translation as a 5 |---|-------|--------------|-------------------| | 1 | **Query Language** | Parse raw strings (PromQL, SQL, DataFusion, ElasticDSL, …) into a language-specific AST | DC `controller/src/query_parser/{promql,sql}.rs` (each parses straight to the L2 relational tree below); asap-planner-rs pulls `promql-parser` + `sqlparser` directly; asap-fusion consumes a pre-built DataFusion `LogicalPlan` (its L1 happens upstream) | | 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/intent_algebra/relational.rs` — the L2 relational `QueryExpr` tree the `query_parser` front ends emit (one shared tree for PromQL + SQL today; a per-language split is future work); 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/lower_to_canonical.rs` (the L2→L3 converter: folds the `relational` L2 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/relational.rs`.* | +| 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/lower.rs` (the L2→L3 converter: folds the `relational` L2 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/relational.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/{promql,sql}.rs` (L1 parsers — emit the L2 tree directly) | > | `crates/core/logical_plan/` | `controller/src/intent_algebra/relational.rs` (the L2 relational `QueryExpr` tree) | -> | `crates/core/intent_algebra/` | `controller/src/intent_algebra/` (with `relational` carrying the L2 relational IR and `lower_to_canonical` lowering it to the canonical L3 types) | +> | `crates/core/intent_algebra/` | `controller/src/intent_algebra/` (with `relational` carrying the L2 relational IR and `lower` lowering 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/intent_algebra/relational.rs` (the L2 relational `QueryExpr` tree) | > | `core::intent_algebra` | `controller/src/intent_algebra/` (canonical `query_expr` / `agg_intent`) + `controller/src/intent_algebra/relational.rs` (the L2 relational IR) | > | `core::sketch_algebra` | `controller/src/sketch_algebra/` | -> | `core::lower` | per-layer: `controller/src/intent_algebra/{lower,lower_to_canonical}.rs` + `controller/src/sketch_algebra/lower.rs` | +> | `core::lower` | per-layer: `controller/src/intent_algebra/{lower,lower}.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/` | @@ -329,7 +329,7 @@ Each returns a language-flavored AST type. No sketch awareness. > tree directly** (`intent_algebra::relational::QueryExpr`: `Aggregate` > / `Window` / `Filter` / `Join` / `Sort` / `Limit` / …). That tree > *is* L2. `query_parser::parse_query_expr_canonical` then lowers it to -> the canonical L3 IR via `intent_algebra::lower_to_canonical` +> the canonical L3 IR via `intent_algebra::lower` > (sketch-fusion + the Binder folded in); `parse_query` projects the > flat `ParsedQuery` summary the legacy analyzer / pipeline consume. > @@ -598,7 +598,7 @@ The schema source is the **`SchemaCatalog`** seam: only the catalog impl swaps. Placement: today the Binder runs at the L2→L3 (relational → canonical) -conversion boundary (`lower_to_canonical::convert_root` calls it). +conversion boundary (`lower::convert_root` calls it). Once the `relational` L2 IR is retired it moves into the `core::lower` L1→L2→L3 passes proper — the `lower_*(ast, schema)` signatures below already anticipate a schema parameter at that point. @@ -787,7 +787,7 @@ The `schema` parameter on each pass is supplied by the **Binder** — see §6 "The Binder — name resolution as an explicit pass". The Binder is the single place symbolic names become positional `ColumnId`s; the `lower_*` passes consume its output and never resolve names ad-hoc. Today the -Binder runs at the L2→L3 boundary inside `lower_to_canonical`; it folds +Binder runs at the L2→L3 boundary inside `lower`; it folds into these `lower_*` signatures once the `relational` L2 IR is retired. ### `core::pipeline` — orchestration @@ -1856,7 +1856,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/relational.rs` | -| `controller/src/algebra/lower.rs` | `controller/src/intent_algebra/lower_to_canonical.rs` (the L2→L3 lowering + converter) | +| `controller/src/algebra/lower.rs` | `controller/src/intent_algebra/lower.rs` (the L2→L3 lowering + 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` | @@ -1883,6 +1883,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/relational.rs` carries the **L2 relational** `QueryExpr` the `query_parser` front ends emit; `lower_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` variants and the standalone sketch-lowering pass that once produced them have been retired. Several controller modules (query_parser, physical/, optimizer/, emit/) still reference `relational` for shared leaf types; fully *deleting* the L2 tree is gated on the canonical `Predicate` covering the remaining `ScalarExpr` variants — but it is the real, current L2 IR (formerly misnamed `legacy_expr`), not removable debt. +- `controller/src/intent_algebra/relational.rs` carries the **L2 relational** `QueryExpr` the `query_parser` front ends emit; `lower.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` variants and the standalone sketch-lowering pass that once produced them have been retired. Several controller modules (query_parser, physical/, optimizer/, emit/) still reference `relational` for shared leaf types; fully *deleting* the L2 tree is gated on the canonical `Predicate` covering the remaining `ScalarExpr` variants — but it is the real, current L2 IR (formerly misnamed `legacy_expr`), not removable debt. - `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/asap_tier_analysis.rs b/control_plane/src/asap_tier_analysis.rs index c5224dc3..d6fd7996 100644 --- a/control_plane/src/asap_tier_analysis.rs +++ b/control_plane/src/asap_tier_analysis.rs @@ -51,8 +51,7 @@ use promql_parser::parser::{self, Expr, VectorSelector}; use crate::intent_algebra::agg_intent::AggIntent; use crate::intent_algebra::query_expr::QueryExpr; -use crate::query_parser::parse_query; -use crate::types_v2::AccuracyTarget; +use crate::query_parser::{parse_query, parse_query_expr_canonical}; pub use crate::sketch_algebra::capability::{capability_for, Capability, SketchKindHandle}; @@ -154,16 +153,15 @@ pub fn analyze_promql_for_asap_tier(metricsql: &str) -> ASAPTierAnalysis { // PromQL function names. let trace = trace_from_promql(metricsql); - // Step 2: pick a sane default accuracy. Warm-tier analysis only - // cares about whether the AggIntent has a sketch binding, and the - // lowerer maps `parsed.exact_required = true` to `AccuracyTarget::Exact` - // anyway. Anything non-exact unlocks the same set of bindings, so - // we pick a mid-range epsilon as the analysis-time default; the - // real per-query accuracy bound comes from QueryWorkload further - // downstream. - let accuracy = AccuracyTarget::Epsilon(0.01); - - let expr = match crate::intent_algebra::lower::lower_parsed_query(&parsed, accuracy) { + // Step 2: lower to the canonical L3 `QueryExpr` via the real parse + // path. `parse_query` (above) already ran this conversion internally + // to build its flat summary; we re-run it here to get the *tree* + // itself, which the flat `ParsedQuery` doesn't carry. The walk below + // only inspects `AggIntent` kinds + accuracy; the converter pins + // sketch-eligible intents at a non-exact epsilon, which is all + // warm-tier analysis needs (the real per-query accuracy bound comes + // from QueryWorkload further downstream). + let expr = match parse_query_expr_canonical(metricsql) { Ok(e) => e, Err(e) => { return ASAPTierAnalysis { @@ -611,6 +609,7 @@ pub fn find_matching_policies( #[cfg(test)] mod tests { use super::*; + use promql_utilities::query_logics::enums::AggregationType; fn keys(items: &[&str]) -> BTreeSet { items.iter().map(|s| s.to_string()).collect() @@ -699,70 +698,73 @@ mod tests { assert!(!a.candidates.is_empty(), "expected at least one candidate"); } - // ── Unsupported / rejected shapes ──────────────────────────────────── + // ── ExactAgg-routed shapes ─────────────────────────────────────────── + // + // `rate` / `irate` / `increase` / `sum` and the bare selector all + // lower (via `lower`) to `AggIntent::Sum`, and + // `capability_for(&Sum)` returns `Capability::ExactAgg(Sum)` — so + // they are ASAP-tier-answerable from exact-precompute state. (The + // older `lower_parsed_query` path *dropped* these intents, masking + // the `ExactAgg` capability and routing everything to archive.) #[test] - fn reject_bare_vector_selector() { + fn bare_vector_selector_binds_to_exact_agg() { + // The PromQL parser models a bare selector as `Aggregate { Sum }` + // over the sample value; `Sum` carries an `ExactAgg` capability. let a = analyze_promql_for_asap_tier("http_requests_total{zone=\"z0\"}"); + assert!(a.unsupported.is_none(), "{a:?}"); + assert_eq!(a.candidates.len(), 1); assert_eq!( - a.unsupported, - Some(UnsupportedReason::NoCallNodeFound), - "{a:?}" + a.candidates[0].required_capability, + Capability::ExactAgg(AggregationType::Sum) ); - assert!(a.candidates.is_empty()); } #[test] - fn reject_rate_function() { - // `rate(...)` lowers to `AggIntent::Rate{...}` and - // `capability_for(&Rate{..})` returns None. + fn rate_binds_to_exact_agg() { let a = analyze_promql_for_asap_tier("rate(http_requests_total[5m])"); - match a.unsupported { - Some(UnsupportedReason::UnsupportedAggIntent(kind)) => assert_eq!(kind, "rate"), - other => panic!("expected UnsupportedAggIntent(rate), got {other:?}"), - } + assert!(a.unsupported.is_none(), "{a:?}"); + assert_eq!( + a.candidates[0].required_capability, + Capability::ExactAgg(AggregationType::Sum) + ); } #[test] - fn reject_irate_function() { + fn irate_binds_to_exact_agg() { + // `irate` shares `AggFunc::Rate` with `rate` in + // `query_parser::promql`; both lower to `AggIntent::Sum`. let a = analyze_promql_for_asap_tier("irate(http_requests_total[5m])"); - // `irate` lowers to `AggIntent::Rate{...}` via the - // control plane's PromQL parser (irate / rate share an AggFunc - // in `query_parser::promql`). The capability bridge returns - // None either way. - match a.unsupported { - Some(UnsupportedReason::UnsupportedAggIntent(kind)) => { - assert!( - kind == "rate" || kind == "irate", - "unexpected intent kind: {kind}" - ); - } - other => panic!("expected UnsupportedAggIntent, got {other:?}"), - } + assert!(a.unsupported.is_none(), "{a:?}"); + assert_eq!( + a.candidates[0].required_capability, + Capability::ExactAgg(AggregationType::Sum) + ); } #[test] - fn reject_increase_function() { + fn increase_binds_to_exact_agg() { let a = analyze_promql_for_asap_tier("increase(http_requests_total[5m])"); - match a.unsupported { - Some(UnsupportedReason::UnsupportedAggIntent(kind)) => { - assert_eq!(kind, "increase"); - } - other => panic!("expected UnsupportedAggIntent(increase), got {other:?}"), - } + assert!(a.unsupported.is_none(), "{a:?}"); + assert_eq!( + a.candidates[0].required_capability, + Capability::ExactAgg(AggregationType::Sum) + ); } #[test] - fn reject_sum_by_bare_metric() { - // `sum by (zone) (metric)` lowers to `AggIntent::Sum`; bridge - // returns None — Sum-over-CountSketch is a follow-up. + fn sum_by_binds_to_exact_agg() { let a = analyze_promql_for_asap_tier("sum by (zone) (http_requests_total)"); - match a.unsupported { - Some(UnsupportedReason::UnsupportedAggIntent(kind)) => assert_eq!(kind, "sum"), - other => panic!("expected UnsupportedAggIntent(sum), got {other:?}"), - } + assert!(a.unsupported.is_none(), "{a:?}"); + assert_eq!( + a.candidates[0].required_capability, + Capability::ExactAgg(AggregationType::Sum) + ); + assert_eq!(a.candidates[0].group_by_keys, keys(&["zone"])); } + // ── Unsupported / rejected shapes ──────────────────────────────────── + #[test] fn unparseable_promql_surfaces_clean_error() { let a = analyze_promql_for_asap_tier("@@@ this is not promql @@@"); @@ -804,14 +806,19 @@ mod tests { #[test] fn is_asap_tier_answerable_false_for_unsupported() { - let a = analyze_promql_for_asap_tier("rate(m[5m])"); + // `count_over_time(...)` without an outer `count by (...)` lowers + // to `AggIntent::Count { accuracy: Exact }` — exact counts have + // no ASAP-tier sketch, so `capability_for` returns `None`. + let a = analyze_promql_for_asap_tier("count_over_time(m[5m])"); assert!(!a.is_asap_tier_answerable()); } #[test] - fn is_asap_tier_answerable_false_for_bare_selector() { + fn is_asap_tier_answerable_true_for_bare_selector() { + // A bare selector lowers to `Aggregate { Sum }`, which carries an + // `ExactAgg` capability — so it is ASAP-tier-answerable. let a = analyze_promql_for_asap_tier("m{zone=\"z0\"}"); - assert!(!a.is_asap_tier_answerable()); + assert!(a.is_asap_tier_answerable()); } // ── Cardinality / count_over_time real-PromQL acceptance ──────────── diff --git a/control_plane/src/intent_algebra/binder.rs b/control_plane/src/intent_algebra/binder.rs index 814bf27a..2d55aa44 100644 --- a/control_plane/src/intent_algebra/binder.rs +++ b/control_plane/src/intent_algebra/binder.rs @@ -17,7 +17,7 @@ //! Our canonical L3 IR (`query_expr::QueryExpr`) already commits to //! positional column identity — `Aggregate.by: Vec` — exactly //! like RisingWave's `InputRef`. What was missing was the *pass* that -//! produces it: resolution was smeared into `lower_to_canonical::convert` +//! produces it: resolution was smeared into `lower::convert` //! with a hardcoded synthesized `(ts, value)` schema, so any query //! referencing a real label / column name (`price`, `host`, …) errored. //! @@ -46,7 +46,7 @@ //! ## Where it sits //! //! Today the Binder runs at the L2→L3 (legacy → canonical) conversion -//! boundary — `lower_to_canonical::convert_root` calls it. Once the +//! boundary — `lower::convert_root` calls it. Once the //! legacy IR is retired it moves into the `core::lower` L1→L2→L3 passes //! proper (the `lower_*(ast, schema)` signatures in design.md §6). diff --git a/control_plane/src/intent_algebra/column_resolution.rs b/control_plane/src/intent_algebra/column_resolution.rs index 6fae9016..b9171a2b 100644 --- a/control_plane/src/intent_algebra/column_resolution.rs +++ b/control_plane/src/intent_algebra/column_resolution.rs @@ -174,7 +174,7 @@ pub fn resolve_column_refs( /// [`resolve_column_refs`] but skips the `ColumnRef::Named` wrapping — /// the legacy `Aggregate.keys` field is already a `Vec`. /// -/// Used by `lower_to_canonical::convert` to translate a legacy +/// Used by `lower::convert` to translate a legacy /// `Aggregate.keys: Vec` into the canonical `by: Vec`. pub fn resolve_named_keys( keys: &[String], diff --git a/control_plane/src/intent_algebra/lower.rs b/control_plane/src/intent_algebra/lower.rs index 1894e4c9..d8f99f6f 100644 --- a/control_plane/src/intent_algebra/lower.rs +++ b/control_plane/src/intent_algebra/lower.rs @@ -1,234 +1,514 @@ -//! Lowering from `query_parser::ParsedQuery` → L3 [`QueryExpr`]. +//! The Layer-2 → canonical L3 IR converter. //! -//! Phase B exposes this as a standalone function so callers can opt into -//! the L3 IR without touching the existing `Analyzer::analyze` → -//! `QueryWorkload` pipeline. Wiring the analyzer to *also* emit a -//! `QueryExpr` is the follow-up phase's job. +//! Recursively converts a *whole* `relational::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`]. //! -//! Per-language scope. The `parse_query` entry point in -//! `query_parser::mod` already dispatches PromQL vs SQL; by the time we -//! see a `ParsedQuery` the language distinction has been collapsed onto -//! the flat summary. For PromQL (the only language this crate consumes -//! per the orchestrator spec), the lowering is: +//! ## Variant mapping //! -//! ```text -//! Scan{TimeSeries} → [Window?] → Aggregate{ by, [intent_per_aggregation] } -//! ``` +//! | relational `QueryExpr` | canonical `QueryExpr` | +//! |---|---| +//! | `Source(spec)` | `Scan { TimeSeries, label_filters: [], schema }` | +//! | `Ref(name)` | `Ref { name }` | +//! | `Filter` | `Filter` (pred via [`convert_scalar`]) | +//! | `Project` | `Project` (each item's expr via [`convert_scalar`])| +//! | `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) | +//! | `Partition` | `Partition` | +//! | `Distinct` | `Distinct` | +//! | `TopK` | `Aggregate { aggs: [AggIntent::TopK] }` (HeavyHitter)| +//! | `Merge` | `Merge` | +//! | `Join` | `Join` (None pred → `Literal(Bool(true))`) | +//! | `SetOp` | `SetOp` | +//! | `Sort` | `Sort` | +//! | `Limit` | `Limit` | +//! | `LetBinding` | `LetBinding` (relational `body` → canonical `child`)| +//! | `PromQLSubquery` | `Subquery` | +//! | `BinaryOp` | `BinaryOp` | //! -//! The window is emitted only when `ParsedQuery.time_window` is non-zero -//! and the query has at least one aggregation (a bare metric selector -//! lowers to a `Scan` alone). Workload-level CSE that produces fan-in -//! (multi-root with `LetBinding` / `Ref`) is a follow-up; this lowers -//! each query independently. +//! The single-statistic sketchable `Aggregate` fusion (`Window`-swap, +//! `Partition` wrap, `StdDev` / `Variance` fan-out) is done directly in +//! canonical terms inside the [`convert`] `Aggregate` arm — there is no +//! intermediate sketch-fused L2-or-L3 IR. +//! +//! ## Schema threading +//! +//! [`convert`] takes a `&Schema` — the schema in scope at the node — and +//! 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 L2→L3 lowering also doesn't handle — proper +//! bottom-up schema flow lands with the canonical `output_schema_in` +//! wiring downstream. #![allow(dead_code)] -use std::collections::HashMap; +use std::time::Duration; + use thiserror::Error; use crate::intent_algebra::agg_intent::AggIntent; +use crate::intent_algebra::binder::Binder; +use crate::intent_algebra::column_resolution::{ + resolve_named_keys, ResolveError, +}; +use crate::intent_algebra::relational::{ + AggFunc, ColumnRef as LColumnRef, PartitionKeys as LPartitionKeys, QueryExpr as LQueryExpr, + ScalarExpr as LScalarExpr, +}; use crate::intent_algebra::query_expr::{ - LabelFilter, QueryExpr, QueryExprError, Source, WindowKind, + from_legacy_scalar, ColumnRef as CColumnRef, HavingPredicate, LiteralValue, + PartitionKeys as CPartitionKeys, Predicate, ProjectItem as CProjectItem, + QueryExpr as CQueryExpr, QueryExprError, Source, WindowKind as CWindowKind, }; -use crate::intent_algebra::schema::{Column, ColumnId, DataType, Schema}; -use crate::query_parser::ParsedQuery; -use crate::types::AggType; -use crate::types_v2::AccuracyTarget; +use crate::intent_algebra::schema::Schema; +use crate::types_v2::{AccuracyTarget, BindingName}; -/// Errors returned by [`lower_parsed_query`]. +/// Errors produced while converting a legacy `QueryExpr` to canonical. +/// +/// `PartialEq` is not derived because [`QueryExprError`] (carried by +/// `Scalar`) does not derive it — callers compare via +/// `matches!(err, ConvertError::Variant { .. })`. #[derive(Debug, Error)] -pub enum LoweringError { - /// `ParsedQuery.metric_name` was empty — nothing to scan. - #[error("empty metric name in parsed query")] - EmptyMetricName, - /// `Aggregate` schema-derivation invariant failed inside the lowered - /// tree. Indicates the lowering logic produced an inconsistent shape; - /// surfaced rather than panicked. - #[error("post-lower schema check failed: {0}")] - PostLowerSchema(#[from] QueryExprError), +pub enum ConvertError { + /// 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 + /// `AggFunc::Custom(_)` triggers this today. + #[error("AggItem `{alias}` uses non-canonical func ({func_dbg}) — no AggIntent equivalent")] + NoCanonicalIntent { alias: String, func_dbg: String }, + /// A legacy `ScalarExpr` leaf failed to translate. Unreachable in + /// practice — `convert_scalar` handles every variant — but kept as a + /// typed boundary around [`from_legacy_scalar`]. + #[error("scalar translation failed: {0}")] + Scalar(QueryExprError), } -/// Lower a `ParsedQuery` into the L3 `QueryExpr` DAG. Single-rooted -/// (one query in, one root out). Workload-level CSE is a follow-up. +/// Lower a legacy Layer-2 `QueryExpr` tree to the canonical L3 IR. /// -/// The accuracy target is supplied separately because `ParsedQuery` -/// carries the legacy `accuracy_sla: f64` only on the `QueryWorkload` -/// downstream of analyzer; passing it explicitly keeps this function -/// orthogonal to the analyzer wiring. -pub fn lower_parsed_query( - parsed: &ParsedQuery, - accuracy: AccuracyTarget, -) -> Result { - if parsed.metric_name.trim().is_empty() { - return Err(LoweringError::EmptyMetricName); - } +/// 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 +/// schema every `ColumnId` indexes into. Because the Binder guarantees +/// every referenced name is in scope, the per-arm positional resolution +/// 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) +} - // ── Scan ───────────────────────────────────────────────────────────── - let scan_schema = scan_schema_for(parsed); - let label_filters = parsed - .label_filters - .iter() - .map(|(k, v)| LabelFilter { - label: k.clone(), - equals: v.clone(), - }) - .collect::>(); - let mut node = QueryExpr::Scan { - source: Source::TimeSeries { - metric: parsed.metric_name.clone(), +/// Convert a legacy `QueryExpr` tree to canonical against an explicit +/// inherited `schema`. The schema is threaded unchanged to every child — +/// see the module doc on schema threading. +pub fn convert(legacy: &LQueryExpr, schema: &Schema) -> Result { + Ok(match legacy { + LQueryExpr::Source(spec) => CQueryExpr::Scan { + source: Source::TimeSeries { + metric: spec.name.clone(), + }, + label_filters: Vec::new(), + // Carry the Binder's complete schema — the same self-contained + // scope every `ColumnId` in this tree resolves against. + schema: schema.clone(), }, - label_filters, - schema: scan_schema.clone(), - }; - // Bare metric selector (no aggregations, no window) — return the Scan - // directly. PromQL `up{job="api"}` lowers here. - if parsed.aggregations.is_empty() && parsed.time_window.is_zero() { - return Ok(node); - } + LQueryExpr::Ref(name) => CQueryExpr::Ref { + name: BindingName::new(name.clone()), + }, - // ── Window (optional) ─────────────────────────────────────────────── - if !parsed.time_window.is_zero() { - node = QueryExpr::Window { - // PromQL `[5m]` is canonically Sliding (per design.md §6 - // line ~308: "PromQL `[5m]` and streaming windows lower - // here"). SQL `TUMBLE` would lower to `Tumbling` — out of - // scope for the PromQL-only Phase B. - kind: WindowKind::Sliding, - size: parsed.time_window, - slide: None, - child: Box::new(node), - }; - } + LQueryExpr::Filter { pred, input } => CQueryExpr::Filter { + pred: convert_scalar(pred, schema)?, + child: Box::new(convert(input, schema)?), + }, - // ── Aggregate (optional, when intents present) ───────────────────── - if !parsed.aggregations.is_empty() { - let aggs = build_intents(parsed, accuracy); - let by = group_by_column_ids(&scan_schema, &parsed.group_by_labels); - node = QueryExpr::Aggregate { - by, + LQueryExpr::Project { cols, input } => { + let cols = cols + .iter() + .map(|pi| { + Ok(CProjectItem { + alias: pi.alias.clone(), + expr: convert_scalar(&pi.expr, schema)?, + }) + }) + .collect::, ConvertError>>()?; + CQueryExpr::Project { + cols, + child: Box::new(convert(input, schema)?), + } + } + + LQueryExpr::Aggregate { + keys, aggs, - having: None, - child: Box::new(node), - }; - } + having, + input, + } => { + // 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(*)`: 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, + }); + continue; + } + let mapped = agg_func_to_intents(&item.func); + if mapped.is_empty() { + return Err(ConvertError::NoCanonicalIntent { + alias: item.alias.clone(), + func_dbg: format!("{:?}", item.func), + }); + } + intents.extend(mapped); + } + let having = match having { + None => None, + Some(se) => Some(HavingPredicate(format!("{:?}", convert_scalar(se, schema)?))), + }; + CQueryExpr::Aggregate { + by, + aggs: intents, + having, + child: Box::new(convert(input, schema)?), + } + } - // Validate the lowered tree has a derivable output schema. Catches - // `Window` without time_index, out-of-range `by` ids, etc. The - // discard is intentional — we only want the validation side-effect. - let _ = node.output_schema()?; + LQueryExpr::Window { + duration, + slide, + input, + } => CQueryExpr::Window { + kind: if slide.is_some() { + CWindowKind::Sliding + } else { + CWindowKind::Tumbling + }, + size: *duration, + slide: *slide, + child: Box::new(convert(input, schema)?), + }, - Ok(node) -} + LQueryExpr::Partition { keys, input } => CQueryExpr::Partition { + keys: match keys { + LPartitionKeys::By(k) => CPartitionKeys::By(k.clone()), + LPartitionKeys::Without(k) => CPartitionKeys::Without(k.clone()), + }, + child: Box::new(convert(input, schema)?), + }, -/// Synthesise the output schema of the `Scan` for this `ParsedQuery`. -/// -/// PromQL convention: a leaf produces `(ts, value, *labels)` with `ts` at -/// position 0 (the time index) and `value` at position 1. Group-by -/// labels and equality-filter labels are projected into positions 2..N -/// in alphabetical order (deterministic). The unique-key set is -/// `[ts, *labels]` — one sample per (timestamp, label-tuple) pair. -fn scan_schema_for(parsed: &ParsedQuery) -> Schema { - let mut columns = vec![ - Column { - name: "ts".into(), - dtype: DataType::Timestamp, - nullable: false, + LQueryExpr::Distinct { cols, input } => CQueryExpr::Distinct { + cols: cols.iter().map(convert_column_ref).collect(), + child: Box::new(convert(input, schema)?), }, - Column { - name: "value".into(), - dtype: DataType::Float64, - nullable: false, + + LQueryExpr::TopK { k, by, input } => { + // γ4 classification default: heavy-hitter intent. The generic + // `Sort + Limit` case never reaches a legacy `TopK` node — + // see `topk_bridge` module doc. + let by = resolve_named_keys(by, schema)?; + CQueryExpr::Aggregate { + by, + aggs: vec![AggIntent::TopK { + k: *k as usize, + accuracy: AccuracyTarget::Epsilon(0.05), + }], + having: None, + child: Box::new(convert(input, schema)?), + } + } + + LQueryExpr::Merge { inputs } => CQueryExpr::Merge { + children: inputs + .iter() + .map(|i| convert(i, schema)) + .collect::, _>>()?, + }, + + LQueryExpr::Join { + kind, + pred, + left, + right, + } => CQueryExpr::Join { + kind: kind.clone(), + // Canonical `Join` requires a predicate; a legacy `None` pred + // is a CROSS JOIN — model it as the tautology `true`. + pred: match pred { + Some(se) => convert_scalar(se, schema)?, + None => Predicate::Literal(LiteralValue::Bool(true)), + }, + left: Box::new(convert(left, schema)?), + right: Box::new(convert(right, schema)?), }, - ]; - // Collect label names from group_by + filters, dedup + sort for - // determinism. - let mut labels: Vec = parsed.group_by_labels.to_vec(); - for k in parsed.label_filters.keys() { - if !labels.contains(k) { - labels.push(k.clone()); + + LQueryExpr::SetOp { + kind, + all, + left, + right, + } => CQueryExpr::SetOp { + kind: kind.clone(), + all: *all, + left: Box::new(convert(left, schema)?), + right: Box::new(convert(right, schema)?), + }, + + LQueryExpr::Sort { keys, input } => CQueryExpr::Sort { + // `SortKey` is the single canonical type (deduped in PR 1). + keys: keys.clone(), + child: Box::new(convert(input, schema)?), + }, + + LQueryExpr::Limit { n, offset, input } => CQueryExpr::Limit { + n: *n as usize, + offset: *offset as usize, + child: Box::new(convert(input, schema)?), + }, + + LQueryExpr::LetBinding { name, expr, body } => CQueryExpr::LetBinding { + name: BindingName::new(name.clone()), + expr: Box::new(convert(expr, schema)?), + // legacy `body` is the canonical `child`. + child: Box::new(convert(body, schema)?), + }, + + LQueryExpr::PromQLSubquery { + range, + resolution, + input, + } => CQueryExpr::Subquery { + range: *range, + resolution: *resolution, + child: Box::new(convert(input, schema)?), + }, + + LQueryExpr::BinaryOp { + op, + lhs, + rhs, + vector_match, + } => CQueryExpr::BinaryOp { + // `BinaryOpKind` / `VectorMatch` are the single canonical + // types (deduped in PR 1). + op: op.clone(), + lhs: Box::new(convert(lhs, schema)?), + rhs: Box::new(convert(rhs, schema)?), + vector_match: vector_match.clone(), + }, + }) +} + +/// Translate a legacy `ScalarExpr` to a canonical `Predicate`, recursing +/// through every composite variant so a nested `ScalarSubquery` (which +/// carries a legacy `QueryExpr` sub-tree) can be converted via [`convert`]. +/// `from_legacy_scalar` alone cannot do this — it has no converter to +/// recurse with — which is why `ScalarSubquery` was the one arm it defers. +pub fn convert_scalar(se: &LScalarExpr, schema: &Schema) -> Result { + match se { + LScalarExpr::ScalarSubquery(inner) => { + Ok(Predicate::ScalarSubquery(Box::new(convert(inner, schema)?))) + } + LScalarExpr::BinaryOp { op, lhs, rhs } => Ok(Predicate::BinaryOp { + op: op.clone(), + lhs: Box::new(convert_scalar(lhs, schema)?), + rhs: Box::new(convert_scalar(rhs, schema)?), + }), + LScalarExpr::IsNull { expr, negated } => Ok(Predicate::IsNull { + expr: Box::new(convert_scalar(expr, schema)?), + negated: *negated, + }), + LScalarExpr::FunctionCall { name, args } => Ok(Predicate::FunctionCall { + name: name.clone(), + args: args + .iter() + .map(|a| convert_scalar(a, schema)) + .collect::, _>>()?, + }), + LScalarExpr::InList { + expr, + list, + negated, + } => Ok(Predicate::InList { + expr: Box::new(convert_scalar(expr, schema)?), + list: list + .iter() + .map(|a| convert_scalar(a, schema)) + .collect::, _>>()?, + negated: *negated, + }), + LScalarExpr::Between { + expr, + low, + high, + negated, + } => Ok(Predicate::Between { + expr: Box::new(convert_scalar(expr, schema)?), + low: Box::new(convert_scalar(low, schema)?), + high: Box::new(convert_scalar(high, schema)?), + negated: *negated, + }), + // `Column` / `Literal` are subquery-free leaves — `from_legacy_scalar` + // translates them directly (and cannot fail for these arms). + LScalarExpr::Column(_) | LScalarExpr::Literal(_) => { + from_legacy_scalar(se).map_err(ConvertError::Scalar) } } - labels.sort(); - for label in &labels { - columns.push(Column { - name: label.clone(), - dtype: DataType::Utf8, - nullable: false, - }); - } - // unique_keys = [ts, *labels] — one sample per (ts, label-tuple). - let mut uk: Vec = vec![0]; // ts - for i in 2..columns.len() { - uk.push(i); - } - Schema::with_time_index(columns, 0, vec![uk]) } -/// Resolve `group_by_labels` (names) to column ids in the scan schema. -/// Labels missing from the schema are silently dropped — they wouldn't -/// have been preserved by the scan anyway. -fn group_by_column_ids(schema: &Schema, group_by_labels: &[String]) -> Vec { - group_by_labels - .iter() - .filter_map(|name| schema.column_id(name)) - .collect() +/// 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 { + match c { + LColumnRef::Named(n) => CColumnRef::Named(n.clone()), + LColumnRef::SampleValue => CColumnRef::SampleValue, + LColumnRef::Wildcard => CColumnRef::Wildcard, + } } -/// Translate the legacy `(AggType, Vec, exact_required)` triple -/// from `ParsedQuery` into typed `AggIntent` entries. -/// -/// Each `AggType::Quantile` fans out to one `AggIntent::Quantile{q}` per -/// `parsed.quantiles` entry — `quantile_over_time(0.95)` and -/// `quantile_over_time(0.99)` co-occurring on the same metric produce -/// two intents on the same `Aggregate`. -fn build_intents(parsed: &ParsedQuery, accuracy: AccuracyTarget) -> Vec { - let effective_accuracy = if parsed.exact_required { - AccuracyTarget::Exact - } else { - accuracy.clone() - }; +// ── AggFunc → AggIntent sketch mapping ─────────────────────────────────────── - // Track which quantiles we've already emitted to avoid duplicates - // when multiple legacy `AggType` rows imply the same φ (e.g. Min → - // q=0.0, Max → q=1.0 historically lived under `AggType::Quantile`). - let mut emitted_q = HashMap::::new(); - let mut out = Vec::new(); - for agg in &parsed.aggregations { - match agg { - AggType::Quantile => { - if parsed.quantiles.is_empty() { - out.push(AggIntent::Quantile { - q: 0.5, - accuracy: effective_accuracy.clone(), - }); - } else { - for &q in &parsed.quantiles { - let key = (q * 1e9) as u64; - if emitted_q.insert(key, true).is_none() { - out.push(AggIntent::Quantile { - q, - accuracy: effective_accuracy.clone(), - }); - } - } - } - } - AggType::Cardinality => out.push(AggIntent::Cardinality { - accuracy: effective_accuracy.clone(), - }), - AggType::Frequency => out.push(AggIntent::Frequency { - accuracy: effective_accuracy.clone(), - }), +/// 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::relational::{ + 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![], } - if out.is_empty() && parsed.exact_required { - // PromQL `sum`, `rate`, … — no sketch intent fires; the lowering - // surfaces this as `AggIntent::Sum` so the L3 schema is - // well-formed (an `Aggregate` with empty aggs is meaningless). - out.push(AggIntent::Sum); - } - out } // ── Tests ───────────────────────────────────────────────────────────────────── @@ -236,117 +516,329 @@ fn build_intents(parsed: &ParsedQuery, accuracy: AccuracyTarget) -> Vec LQueryExpr { + LQueryExpr::Source(SourceSpec { name: name.into() }) + } + + fn agg_item(alias: &str, func: AggFunc) -> AggItem { + AggItem { + alias: alias.into(), + func, + col: LColumnRef::SampleValue, + distinct: false, + } + } #[test] - fn lower_promql_basic() { - let parsed = parse_query( - "quantile_over_time(0.99, http_request_duration_seconds{service=\"api\"}[5m])", - ) - .expect("parse"); - let expr = lower_parsed_query(&parsed, AccuracyTarget::Epsilon(0.01)).expect("lower"); - - // Root: Aggregate { aggs: [Quantile{0.99, ε=0.01}] } - match &expr { - QueryExpr::Aggregate { aggs, child, .. } => { - assert_eq!(aggs.len(), 1); - match &aggs[0] { - AggIntent::Quantile { q, accuracy } => { - assert!((*q - 0.99).abs() < 1e-9); - assert_eq!(*accuracy, AccuracyTarget::Epsilon(0.01)); + fn source_becomes_scan_with_synthesized_schema() { + let c = convert_root(&src("http_requests_total")).unwrap(); + match c { + CQueryExpr::Scan { + source, + label_filters, + schema, + } => { + assert!(matches!(source, Source::TimeSeries { metric } if metric == "http_requests_total")); + assert!(label_filters.is_empty()); + assert_eq!(schema.columns.len(), 2); // ts, value + } + other => panic!("expected Scan, got {other:?}"), + } + } + + #[test] + fn ref_and_let_binding_round_trip_names() { + let legacy = LQueryExpr::LetBinding { + name: "cte".into(), + expr: Box::new(src("m")), + body: Box::new(LQueryExpr::Ref("cte".into())), + }; + match convert_root(&legacy).unwrap() { + CQueryExpr::LetBinding { name, child, .. } => { + assert_eq!(name.as_str(), "cte"); + assert!(matches!(*child, CQueryExpr::Ref { name } if name.as_str() == "cte")); + } + other => panic!("expected LetBinding, got {other:?}"), + } + } + + #[test] + fn window_over_aggregate_full_tree() { + // Window { Aggregate { keys: [], aggs: [Sum], Source } } + let legacy = LQueryExpr::Window { + duration: Duration::from_secs(300), + slide: None, + input: Box::new(LQueryExpr::Aggregate { + keys: vec![], + aggs: vec![agg_item("s", AggFunc::Sum)], + having: None, + input: Box::new(src("m")), + }), + }; + match convert_root(&legacy).unwrap() { + CQueryExpr::Window { + kind, size, child, .. + } => { + assert_eq!(kind, CWindowKind::Tumbling); + assert_eq!(size, Duration::from_secs(300)); + match *child { + CQueryExpr::Aggregate { aggs, child, .. } => { + assert!(matches!(aggs.as_slice(), [AggIntent::Sum])); + assert!(matches!(*child, CQueryExpr::Scan { .. })); } - other => panic!("expected Quantile intent, got {other:?}"), + other => panic!("expected Aggregate, got {other:?}"), } - // Mid: Window - match child.as_ref() { - QueryExpr::Window { - kind, size, child, .. - } => { - assert_eq!(*kind, WindowKind::Sliding); - assert_eq!(*size, std::time::Duration::from_secs(300)); - // Leaf: Scan - match child.as_ref() { - QueryExpr::Scan { - source, - label_filters, - schema, - } => { - match source { - Source::TimeSeries { metric } => { - assert_eq!(metric, "http_request_duration_seconds") - } - _ => panic!("expected TimeSeries source"), - } - assert_eq!(label_filters.len(), 1); - assert_eq!(label_filters[0].label, "service"); - assert_eq!(label_filters[0].equals, "api"); - // Schema: ts (time_index), value, service - assert_eq!(schema.time_index, Some(0)); - assert_eq!(schema.columns[0].name, "ts"); - assert_eq!(schema.columns[1].name, "value"); - assert!(schema.has_unique_key()); - } - other => panic!("expected Scan, got {other:?}"), - } - } - other => panic!("expected Window, got {other:?}"), + } + other => panic!("expected Window, got {other:?}"), + } + } + + #[test] + fn sliding_window_carries_slide() { + let legacy = LQueryExpr::Window { + duration: Duration::from_secs(600), + slide: Some(Duration::from_secs(60)), + input: Box::new(src("m")), + }; + match convert_root(&legacy).unwrap() { + CQueryExpr::Window { kind, slide, .. } => { + assert_eq!(kind, CWindowKind::Sliding); + assert_eq!(slide, Some(Duration::from_secs(60))); + } + other => panic!("expected Window, got {other:?}"), + } + } + + #[test] + 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(), "no GROUP BY → empty `by`: {by:?}"); + assert!(matches!(aggs.as_slice(), [AggIntent::Sum])); + } + other => panic!("expected Aggregate, got {other:?}"), + } + } + + #[test] + 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() { + CQueryExpr::Aggregate { by, .. } => assert!(by.is_empty()), + other => panic!("expected Aggregate, got {other:?}"), + } + } + + #[test] + 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 by, ref aggs, .. } + if by.is_empty() + && matches!(aggs.as_slice(), [AggIntent::Quantile { .. }]) + )); + } + other => panic!("expected Window, got {other:?}"), + } + } + + #[test] + fn topk_folds_into_aggregate_with_topk_intent() { + let legacy = LQueryExpr::TopK { + k: 5, + by: vec![], + input: Box::new(src("m")), + }; + match convert_root(&legacy).unwrap() { + CQueryExpr::Aggregate { by, aggs, .. } => { + assert!(by.is_empty()); + assert!(matches!(aggs.as_slice(), [AggIntent::TopK { k: 5, .. }])); + } + other => panic!("expected Aggregate, got {other:?}"), + } + } + + #[test] + fn custom_agg_func_errors() { + let legacy = LQueryExpr::Aggregate { + keys: vec![], + aggs: vec![agg_item("u", AggFunc::Custom("my_udf".into()))], + having: None, + input: Box::new(src("m")), + }; + assert!(matches!( + convert_root(&legacy).unwrap_err(), + ConvertError::NoCanonicalIntent { .. } + )); + } + + #[test] + fn filter_pred_with_scalar_subquery_recurses() { + // Filter { pred: ScalarSubquery(Ref("cte")), Source } — the + // ScalarSubquery arm `from_legacy_scalar` defers is handled here + // by recursing through `convert`. + let legacy = LQueryExpr::Filter { + pred: LScalarExpr::ScalarSubquery(Box::new(LQueryExpr::Ref("cte".into()))), + input: Box::new(src("m")), + }; + match convert_root(&legacy).unwrap() { + CQueryExpr::Filter { pred, .. } => match pred { + Predicate::ScalarSubquery(inner) => { + assert!(matches!(*inner, CQueryExpr::Ref { name } if name.as_str() == "cte")); } + other => panic!("expected ScalarSubquery, got {other:?}"), + }, + other => panic!("expected Filter, got {other:?}"), + } + } + + #[test] + fn project_translates_each_item_expr() { + let legacy = LQueryExpr::Project { + cols: vec![LProjectItem { + alias: Some("v".into()), + expr: LScalarExpr::Column("value".into()), + }], + input: Box::new(src("m")), + }; + match convert_root(&legacy).unwrap() { + CQueryExpr::Project { cols, .. } => { + assert_eq!(cols.len(), 1); + assert_eq!(cols[0].alias.as_deref(), Some("v")); + assert!(matches!(&cols[0].expr, Predicate::Column(_))); } - other => panic!("expected Aggregate at root, got {other:?}"), + other => panic!("expected Project, got {other:?}"), } + } - // The lowered tree's output schema is derivable end-to-end. - let schema = expr.output_schema().expect("output_schema"); - assert!( - schema.columns.iter().any(|c| c.name == "quantile_0_99"), - "output schema should contain quantile_0_99 column, got {:?}", - schema.columns - ); + #[test] + fn binary_op_converts_both_sides() { + let legacy = LQueryExpr::BinaryOp { + op: crate::intent_algebra::query_expr::BinaryOpKind::Add, + lhs: Box::new(src("a")), + rhs: Box::new(src("b")), + vector_match: None, + }; + match convert_root(&legacy).unwrap() { + CQueryExpr::BinaryOp { lhs, rhs, .. } => { + assert!(matches!(*lhs, CQueryExpr::Scan { .. })); + assert!(matches!(*rhs, CQueryExpr::Scan { .. })); + } + other => panic!("expected BinaryOp, got {other:?}"), + } } #[test] - fn lower_promql_with_group_by() { - let parsed = - parse_query("sum by (host) (quantile_over_time(0.95, latency{env=\"prod\"}[1m]))") - .expect("parse"); - let expr = lower_parsed_query(&parsed, AccuracyTarget::Epsilon(0.05)).expect("lower"); - let schema = expr.output_schema().expect("output_schema"); - // `host` is a group-by → it lands in unique_keys at position 0 - // in the output (positional projection of `by`). - assert!(schema.has_unique_key()); - assert!(schema.columns.iter().any(|c| c.name == "host")); + fn cross_join_none_pred_becomes_true_literal() { + let legacy = LQueryExpr::Join { + kind: crate::intent_algebra::query_expr::JoinKind::Cross, + pred: None, + left: Box::new(src("a")), + right: Box::new(src("b")), + }; + match convert_root(&legacy).unwrap() { + CQueryExpr::Join { pred, .. } => { + assert!(matches!(pred, Predicate::Literal(LiteralValue::Bool(true)))); + } + other => panic!("expected Join, got {other:?}"), + } } #[test] - fn lower_promql_cardinality() { - let parsed = - parse_query("count by (user_id) (count_over_time(active_users{env=\"prod\"}[5m]))") - .expect("parse"); - let expr = lower_parsed_query(&parsed, AccuracyTarget::Epsilon(0.01)).expect("lower"); - let schema = expr.output_schema().expect("output_schema"); - // Cardinality intent → output column named `cardinality` of dtype Int64. - let card = schema - .columns - .iter() - .find(|c| c.name == "cardinality") - .expect("cardinality column should be present"); - assert!(matches!(card.dtype, DataType::Int64)); + fn promql_subquery_becomes_canonical_subquery() { + let legacy = LQueryExpr::PromQLSubquery { + range: Duration::from_secs(3600), + resolution: Some(Duration::from_secs(60)), + input: Box::new(src("m")), + }; + match convert_root(&legacy).unwrap() { + CQueryExpr::Subquery { + range, resolution, .. + } => { + assert_eq!(range, Duration::from_secs(3600)); + assert_eq!(resolution, Some(Duration::from_secs(60))); + } + other => panic!("expected Subquery, got {other:?}"), + } } #[test] - fn lower_empty_metric_errors() { - let parsed = ParsedQuery { - metric_name: String::new(), - aggregations: vec![], - group_by_labels: vec![], - label_filters: HashMap::new(), - time_window: std::time::Duration::ZERO, - exact_required: false, - quantiles: vec![], - hint: None, + fn nested_tree_converts_recursively() { + // Sort { Limit { Filter { Window { Source } } } } — exercises a + // deep pass-through chain in one shot. + let legacy = LQueryExpr::Sort { + keys: vec![], + input: Box::new(LQueryExpr::Limit { + n: 10, + offset: 0, + input: Box::new(LQueryExpr::Filter { + pred: LScalarExpr::Literal( + crate::intent_algebra::relational::LiteralValue::Bool(true), + ), + input: Box::new(LQueryExpr::Window { + duration: Duration::from_secs(60), + slide: None, + input: Box::new(src("m")), + }), + }), + }), + }; + let c = convert_root(&legacy).unwrap(); + // Walk down and assert the spine survived. + let CQueryExpr::Sort { child, .. } = c else { + panic!("expected Sort") + }; + let CQueryExpr::Limit { n, child, .. } = *child else { + panic!("expected Limit") + }; + assert_eq!(n, 10); + let CQueryExpr::Filter { child, .. } = *child else { + panic!("expected Filter") + }; + let CQueryExpr::Window { child, .. } = *child else { + panic!("expected Window") }; - let err = lower_parsed_query(&parsed, AccuracyTarget::Exact).unwrap_err(); - assert!(matches!(err, LoweringError::EmptyMetricName)); + assert!(matches!(*child, CQueryExpr::Scan { .. })); } } diff --git a/control_plane/src/intent_algebra/lower_to_canonical.rs b/control_plane/src/intent_algebra/lower_to_canonical.rs deleted file mode 100644 index d8f99f6f..00000000 --- a/control_plane/src/intent_algebra/lower_to_canonical.rs +++ /dev/null @@ -1,844 +0,0 @@ -//! The Layer-2 → canonical L3 IR converter. -//! -//! Recursively converts a *whole* `relational::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 -//! -//! | relational `QueryExpr` | canonical `QueryExpr` | -//! |---|---| -//! | `Source(spec)` | `Scan { TimeSeries, label_filters: [], schema }` | -//! | `Ref(name)` | `Ref { name }` | -//! | `Filter` | `Filter` (pred via [`convert_scalar`]) | -//! | `Project` | `Project` (each item's expr via [`convert_scalar`])| -//! | `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) | -//! | `Partition` | `Partition` | -//! | `Distinct` | `Distinct` | -//! | `TopK` | `Aggregate { aggs: [AggIntent::TopK] }` (HeavyHitter)| -//! | `Merge` | `Merge` | -//! | `Join` | `Join` (None pred → `Literal(Bool(true))`) | -//! | `SetOp` | `SetOp` | -//! | `Sort` | `Sort` | -//! | `Limit` | `Limit` | -//! | `LetBinding` | `LetBinding` (relational `body` → canonical `child`)| -//! | `PromQLSubquery` | `Subquery` | -//! | `BinaryOp` | `BinaryOp` | -//! -//! The single-statistic sketchable `Aggregate` fusion (`Window`-swap, -//! `Partition` wrap, `StdDev` / `Variance` fan-out) is done directly in -//! canonical terms inside the [`convert`] `Aggregate` arm — there is no -//! intermediate sketch-fused L2-or-L3 IR. -//! -//! ## Schema threading -//! -//! [`convert`] takes a `&Schema` — the schema in scope at the node — and -//! 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 L2→L3 lowering also doesn't handle — proper -//! bottom-up schema flow lands with the canonical `output_schema_in` -//! wiring downstream. - -#![allow(dead_code)] - -use std::time::Duration; - -use thiserror::Error; - -use crate::intent_algebra::agg_intent::AggIntent; -use crate::intent_algebra::binder::Binder; -use crate::intent_algebra::column_resolution::{ - resolve_named_keys, ResolveError, -}; -use crate::intent_algebra::relational::{ - 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, - PartitionKeys as CPartitionKeys, Predicate, ProjectItem as CProjectItem, - QueryExpr as CQueryExpr, QueryExprError, Source, WindowKind as CWindowKind, -}; -use crate::intent_algebra::schema::Schema; -use crate::types_v2::{AccuracyTarget, BindingName}; - -/// Errors produced while converting a legacy `QueryExpr` to canonical. -/// -/// `PartialEq` is not derived because [`QueryExprError`] (carried by -/// `Scalar`) does not derive it — callers compare via -/// `matches!(err, ConvertError::Variant { .. })`. -#[derive(Debug, Error)] -pub enum ConvertError { - /// 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 - /// `AggFunc::Custom(_)` triggers this today. - #[error("AggItem `{alias}` uses non-canonical func ({func_dbg}) — no AggIntent equivalent")] - NoCanonicalIntent { alias: String, func_dbg: String }, - /// A legacy `ScalarExpr` leaf failed to translate. Unreachable in - /// practice — `convert_scalar` handles every variant — but kept as a - /// typed boundary around [`from_legacy_scalar`]. - #[error("scalar translation failed: {0}")] - Scalar(QueryExprError), -} - -/// Lower a legacy Layer-2 `QueryExpr` tree to the canonical L3 IR. -/// -/// 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 -/// schema every `ColumnId` indexes into. Because the Binder guarantees -/// every referenced name is in scope, the per-arm positional resolution -/// 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) -} - -/// Convert a legacy `QueryExpr` tree to canonical against an explicit -/// inherited `schema`. The schema is threaded unchanged to every child — -/// see the module doc on schema threading. -pub fn convert(legacy: &LQueryExpr, schema: &Schema) -> Result { - Ok(match legacy { - LQueryExpr::Source(spec) => CQueryExpr::Scan { - source: Source::TimeSeries { - metric: spec.name.clone(), - }, - label_filters: Vec::new(), - // Carry the Binder's complete schema — the same self-contained - // scope every `ColumnId` in this tree resolves against. - schema: schema.clone(), - }, - - LQueryExpr::Ref(name) => CQueryExpr::Ref { - name: BindingName::new(name.clone()), - }, - - LQueryExpr::Filter { pred, input } => CQueryExpr::Filter { - pred: convert_scalar(pred, schema)?, - child: Box::new(convert(input, schema)?), - }, - - LQueryExpr::Project { cols, input } => { - let cols = cols - .iter() - .map(|pi| { - Ok(CProjectItem { - alias: pi.alias.clone(), - expr: convert_scalar(&pi.expr, schema)?, - }) - }) - .collect::, ConvertError>>()?; - CQueryExpr::Project { - cols, - child: Box::new(convert(input, schema)?), - } - } - - LQueryExpr::Aggregate { - keys, - aggs, - having, - input, - } => { - // 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(*)`: 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, - }); - continue; - } - let mapped = agg_func_to_intents(&item.func); - if mapped.is_empty() { - return Err(ConvertError::NoCanonicalIntent { - alias: item.alias.clone(), - func_dbg: format!("{:?}", item.func), - }); - } - intents.extend(mapped); - } - let having = match having { - None => None, - Some(se) => Some(HavingPredicate(format!("{:?}", convert_scalar(se, schema)?))), - }; - CQueryExpr::Aggregate { - by, - aggs: intents, - having, - child: Box::new(convert(input, schema)?), - } - } - - LQueryExpr::Window { - duration, - slide, - input, - } => CQueryExpr::Window { - kind: if slide.is_some() { - CWindowKind::Sliding - } else { - CWindowKind::Tumbling - }, - size: *duration, - slide: *slide, - child: Box::new(convert(input, schema)?), - }, - - LQueryExpr::Partition { keys, input } => CQueryExpr::Partition { - keys: match keys { - LPartitionKeys::By(k) => CPartitionKeys::By(k.clone()), - LPartitionKeys::Without(k) => CPartitionKeys::Without(k.clone()), - }, - child: Box::new(convert(input, schema)?), - }, - - LQueryExpr::Distinct { cols, input } => CQueryExpr::Distinct { - cols: cols.iter().map(convert_column_ref).collect(), - child: Box::new(convert(input, schema)?), - }, - - LQueryExpr::TopK { k, by, input } => { - // γ4 classification default: heavy-hitter intent. The generic - // `Sort + Limit` case never reaches a legacy `TopK` node — - // see `topk_bridge` module doc. - let by = resolve_named_keys(by, schema)?; - CQueryExpr::Aggregate { - by, - aggs: vec![AggIntent::TopK { - k: *k as usize, - accuracy: AccuracyTarget::Epsilon(0.05), - }], - having: None, - child: Box::new(convert(input, schema)?), - } - } - - LQueryExpr::Merge { inputs } => CQueryExpr::Merge { - children: inputs - .iter() - .map(|i| convert(i, schema)) - .collect::, _>>()?, - }, - - LQueryExpr::Join { - kind, - pred, - left, - right, - } => CQueryExpr::Join { - kind: kind.clone(), - // Canonical `Join` requires a predicate; a legacy `None` pred - // is a CROSS JOIN — model it as the tautology `true`. - pred: match pred { - Some(se) => convert_scalar(se, schema)?, - None => Predicate::Literal(LiteralValue::Bool(true)), - }, - left: Box::new(convert(left, schema)?), - right: Box::new(convert(right, schema)?), - }, - - LQueryExpr::SetOp { - kind, - all, - left, - right, - } => CQueryExpr::SetOp { - kind: kind.clone(), - all: *all, - left: Box::new(convert(left, schema)?), - right: Box::new(convert(right, schema)?), - }, - - LQueryExpr::Sort { keys, input } => CQueryExpr::Sort { - // `SortKey` is the single canonical type (deduped in PR 1). - keys: keys.clone(), - child: Box::new(convert(input, schema)?), - }, - - LQueryExpr::Limit { n, offset, input } => CQueryExpr::Limit { - n: *n as usize, - offset: *offset as usize, - child: Box::new(convert(input, schema)?), - }, - - LQueryExpr::LetBinding { name, expr, body } => CQueryExpr::LetBinding { - name: BindingName::new(name.clone()), - expr: Box::new(convert(expr, schema)?), - // legacy `body` is the canonical `child`. - child: Box::new(convert(body, schema)?), - }, - - LQueryExpr::PromQLSubquery { - range, - resolution, - input, - } => CQueryExpr::Subquery { - range: *range, - resolution: *resolution, - child: Box::new(convert(input, schema)?), - }, - - LQueryExpr::BinaryOp { - op, - lhs, - rhs, - vector_match, - } => CQueryExpr::BinaryOp { - // `BinaryOpKind` / `VectorMatch` are the single canonical - // types (deduped in PR 1). - op: op.clone(), - lhs: Box::new(convert(lhs, schema)?), - rhs: Box::new(convert(rhs, schema)?), - vector_match: vector_match.clone(), - }, - }) -} - -/// Translate a legacy `ScalarExpr` to a canonical `Predicate`, recursing -/// through every composite variant so a nested `ScalarSubquery` (which -/// carries a legacy `QueryExpr` sub-tree) can be converted via [`convert`]. -/// `from_legacy_scalar` alone cannot do this — it has no converter to -/// recurse with — which is why `ScalarSubquery` was the one arm it defers. -pub fn convert_scalar(se: &LScalarExpr, schema: &Schema) -> Result { - match se { - LScalarExpr::ScalarSubquery(inner) => { - Ok(Predicate::ScalarSubquery(Box::new(convert(inner, schema)?))) - } - LScalarExpr::BinaryOp { op, lhs, rhs } => Ok(Predicate::BinaryOp { - op: op.clone(), - lhs: Box::new(convert_scalar(lhs, schema)?), - rhs: Box::new(convert_scalar(rhs, schema)?), - }), - LScalarExpr::IsNull { expr, negated } => Ok(Predicate::IsNull { - expr: Box::new(convert_scalar(expr, schema)?), - negated: *negated, - }), - LScalarExpr::FunctionCall { name, args } => Ok(Predicate::FunctionCall { - name: name.clone(), - args: args - .iter() - .map(|a| convert_scalar(a, schema)) - .collect::, _>>()?, - }), - LScalarExpr::InList { - expr, - list, - negated, - } => Ok(Predicate::InList { - expr: Box::new(convert_scalar(expr, schema)?), - list: list - .iter() - .map(|a| convert_scalar(a, schema)) - .collect::, _>>()?, - negated: *negated, - }), - LScalarExpr::Between { - expr, - low, - high, - negated, - } => Ok(Predicate::Between { - expr: Box::new(convert_scalar(expr, schema)?), - low: Box::new(convert_scalar(low, schema)?), - high: Box::new(convert_scalar(high, schema)?), - negated: *negated, - }), - // `Column` / `Literal` are subquery-free leaves — `from_legacy_scalar` - // translates them directly (and cannot fail for these arms). - LScalarExpr::Column(_) | LScalarExpr::Literal(_) => { - from_legacy_scalar(se).map_err(ConvertError::Scalar) - } - } -} - -/// 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 { - match c { - LColumnRef::Named(n) => CColumnRef::Named(n.clone()), - LColumnRef::SampleValue => CColumnRef::SampleValue, - LColumnRef::Wildcard => CColumnRef::Wildcard, - } -} - -// ── AggFunc → AggIntent sketch mapping ─────────────────────────────────────── - -/// 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::relational::{ - 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)] -mod tests { - use super::*; - use crate::intent_algebra::relational::{ - AggFunc, AggItem, ColumnRef as LColumnRef, ProjectItem as LProjectItem, SourceSpec, - }; - - fn src(name: &str) -> LQueryExpr { - LQueryExpr::Source(SourceSpec { name: name.into() }) - } - - fn agg_item(alias: &str, func: AggFunc) -> AggItem { - AggItem { - alias: alias.into(), - func, - col: LColumnRef::SampleValue, - distinct: false, - } - } - - #[test] - fn source_becomes_scan_with_synthesized_schema() { - let c = convert_root(&src("http_requests_total")).unwrap(); - match c { - CQueryExpr::Scan { - source, - label_filters, - schema, - } => { - assert!(matches!(source, Source::TimeSeries { metric } if metric == "http_requests_total")); - assert!(label_filters.is_empty()); - assert_eq!(schema.columns.len(), 2); // ts, value - } - other => panic!("expected Scan, got {other:?}"), - } - } - - #[test] - fn ref_and_let_binding_round_trip_names() { - let legacy = LQueryExpr::LetBinding { - name: "cte".into(), - expr: Box::new(src("m")), - body: Box::new(LQueryExpr::Ref("cte".into())), - }; - match convert_root(&legacy).unwrap() { - CQueryExpr::LetBinding { name, child, .. } => { - assert_eq!(name.as_str(), "cte"); - assert!(matches!(*child, CQueryExpr::Ref { name } if name.as_str() == "cte")); - } - other => panic!("expected LetBinding, got {other:?}"), - } - } - - #[test] - fn window_over_aggregate_full_tree() { - // Window { Aggregate { keys: [], aggs: [Sum], Source } } - let legacy = LQueryExpr::Window { - duration: Duration::from_secs(300), - slide: None, - input: Box::new(LQueryExpr::Aggregate { - keys: vec![], - aggs: vec![agg_item("s", AggFunc::Sum)], - having: None, - input: Box::new(src("m")), - }), - }; - match convert_root(&legacy).unwrap() { - CQueryExpr::Window { - kind, size, child, .. - } => { - assert_eq!(kind, CWindowKind::Tumbling); - assert_eq!(size, Duration::from_secs(300)); - match *child { - CQueryExpr::Aggregate { aggs, child, .. } => { - assert!(matches!(aggs.as_slice(), [AggIntent::Sum])); - assert!(matches!(*child, CQueryExpr::Scan { .. })); - } - other => panic!("expected Aggregate, got {other:?}"), - } - } - other => panic!("expected Window, got {other:?}"), - } - } - - #[test] - fn sliding_window_carries_slide() { - let legacy = LQueryExpr::Window { - duration: Duration::from_secs(600), - slide: Some(Duration::from_secs(60)), - input: Box::new(src("m")), - }; - match convert_root(&legacy).unwrap() { - CQueryExpr::Window { kind, slide, .. } => { - assert_eq!(kind, CWindowKind::Sliding); - assert_eq!(slide, Some(Duration::from_secs(60))); - } - other => panic!("expected Window, got {other:?}"), - } - } - - #[test] - 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(), "no GROUP BY → empty `by`: {by:?}"); - assert!(matches!(aggs.as_slice(), [AggIntent::Sum])); - } - other => panic!("expected Aggregate, got {other:?}"), - } - } - - #[test] - 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() { - CQueryExpr::Aggregate { by, .. } => assert!(by.is_empty()), - other => panic!("expected Aggregate, got {other:?}"), - } - } - - #[test] - 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 by, ref aggs, .. } - if by.is_empty() - && matches!(aggs.as_slice(), [AggIntent::Quantile { .. }]) - )); - } - other => panic!("expected Window, got {other:?}"), - } - } - - #[test] - fn topk_folds_into_aggregate_with_topk_intent() { - let legacy = LQueryExpr::TopK { - k: 5, - by: vec![], - input: Box::new(src("m")), - }; - match convert_root(&legacy).unwrap() { - CQueryExpr::Aggregate { by, aggs, .. } => { - assert!(by.is_empty()); - assert!(matches!(aggs.as_slice(), [AggIntent::TopK { k: 5, .. }])); - } - other => panic!("expected Aggregate, got {other:?}"), - } - } - - #[test] - fn custom_agg_func_errors() { - let legacy = LQueryExpr::Aggregate { - keys: vec![], - aggs: vec![agg_item("u", AggFunc::Custom("my_udf".into()))], - having: None, - input: Box::new(src("m")), - }; - assert!(matches!( - convert_root(&legacy).unwrap_err(), - ConvertError::NoCanonicalIntent { .. } - )); - } - - #[test] - fn filter_pred_with_scalar_subquery_recurses() { - // Filter { pred: ScalarSubquery(Ref("cte")), Source } — the - // ScalarSubquery arm `from_legacy_scalar` defers is handled here - // by recursing through `convert`. - let legacy = LQueryExpr::Filter { - pred: LScalarExpr::ScalarSubquery(Box::new(LQueryExpr::Ref("cte".into()))), - input: Box::new(src("m")), - }; - match convert_root(&legacy).unwrap() { - CQueryExpr::Filter { pred, .. } => match pred { - Predicate::ScalarSubquery(inner) => { - assert!(matches!(*inner, CQueryExpr::Ref { name } if name.as_str() == "cte")); - } - other => panic!("expected ScalarSubquery, got {other:?}"), - }, - other => panic!("expected Filter, got {other:?}"), - } - } - - #[test] - fn project_translates_each_item_expr() { - let legacy = LQueryExpr::Project { - cols: vec![LProjectItem { - alias: Some("v".into()), - expr: LScalarExpr::Column("value".into()), - }], - input: Box::new(src("m")), - }; - match convert_root(&legacy).unwrap() { - CQueryExpr::Project { cols, .. } => { - assert_eq!(cols.len(), 1); - assert_eq!(cols[0].alias.as_deref(), Some("v")); - assert!(matches!(&cols[0].expr, Predicate::Column(_))); - } - other => panic!("expected Project, got {other:?}"), - } - } - - #[test] - fn binary_op_converts_both_sides() { - let legacy = LQueryExpr::BinaryOp { - op: crate::intent_algebra::query_expr::BinaryOpKind::Add, - lhs: Box::new(src("a")), - rhs: Box::new(src("b")), - vector_match: None, - }; - match convert_root(&legacy).unwrap() { - CQueryExpr::BinaryOp { lhs, rhs, .. } => { - assert!(matches!(*lhs, CQueryExpr::Scan { .. })); - assert!(matches!(*rhs, CQueryExpr::Scan { .. })); - } - other => panic!("expected BinaryOp, got {other:?}"), - } - } - - #[test] - fn cross_join_none_pred_becomes_true_literal() { - let legacy = LQueryExpr::Join { - kind: crate::intent_algebra::query_expr::JoinKind::Cross, - pred: None, - left: Box::new(src("a")), - right: Box::new(src("b")), - }; - match convert_root(&legacy).unwrap() { - CQueryExpr::Join { pred, .. } => { - assert!(matches!(pred, Predicate::Literal(LiteralValue::Bool(true)))); - } - other => panic!("expected Join, got {other:?}"), - } - } - - #[test] - fn promql_subquery_becomes_canonical_subquery() { - let legacy = LQueryExpr::PromQLSubquery { - range: Duration::from_secs(3600), - resolution: Some(Duration::from_secs(60)), - input: Box::new(src("m")), - }; - match convert_root(&legacy).unwrap() { - CQueryExpr::Subquery { - range, resolution, .. - } => { - assert_eq!(range, Duration::from_secs(3600)); - assert_eq!(resolution, Some(Duration::from_secs(60))); - } - other => panic!("expected Subquery, got {other:?}"), - } - } - - #[test] - fn nested_tree_converts_recursively() { - // Sort { Limit { Filter { Window { Source } } } } — exercises a - // deep pass-through chain in one shot. - let legacy = LQueryExpr::Sort { - keys: vec![], - input: Box::new(LQueryExpr::Limit { - n: 10, - offset: 0, - input: Box::new(LQueryExpr::Filter { - pred: LScalarExpr::Literal( - crate::intent_algebra::relational::LiteralValue::Bool(true), - ), - input: Box::new(LQueryExpr::Window { - duration: Duration::from_secs(60), - slide: None, - input: Box::new(src("m")), - }), - }), - }), - }; - let c = convert_root(&legacy).unwrap(); - // Walk down and assert the spine survived. - let CQueryExpr::Sort { child, .. } = c else { - panic!("expected Sort") - }; - let CQueryExpr::Limit { n, child, .. } = *child else { - panic!("expected Limit") - }; - assert_eq!(n, 10); - let CQueryExpr::Filter { child, .. } = *child else { - panic!("expected Filter") - }; - let CQueryExpr::Window { child, .. } = *child else { - panic!("expected Window") - }; - assert!(matches!(*child, CQueryExpr::Scan { .. })); - } -} diff --git a/control_plane/src/intent_algebra/mod.rs b/control_plane/src/intent_algebra/mod.rs index 671dcc3b..b186f47a 100644 --- a/control_plane/src/intent_algebra/mod.rs +++ b/control_plane/src/intent_algebra/mod.rs @@ -34,8 +34,6 @@ //! workload-level CSE lives one layer up in `types_v2::WorkloadPlan`. //! - [`Schema`] — typed schema flowing on every L3 edge. `unique_keys` is //! the load-bearing field for CSE legality (`design.md` §6 line ~1284). -//! - [`lower_parsed_query`] — `query_parser::ParsedQuery` → [`QueryExpr`] -//! single-query lowering. //! //! Phase F adds the CSE surface that consumes `Schema::unique_keys`: //! @@ -75,7 +73,6 @@ pub mod agg_intent; pub mod cse; -pub mod lower; pub mod query_expr; pub mod schema; @@ -91,8 +88,8 @@ pub mod relational; // raw `relational::QueryExpr` tree and route it through `convert_root`, // which lowers it (single-statistic sketchable `Aggregate` fusion folded // in) onto the canonical IR. -pub mod lower_to_canonical; -pub use lower_to_canonical::{convert, convert_root, ConvertError}; +pub mod lower; +pub use lower::{convert, convert_root, ConvertError}; // Step γ7: the L3 Binder — name resolution as an explicit pass. Produces // the complete self-contained `Schema` every `ColumnId` indexes into; @@ -109,7 +106,6 @@ pub use agg_intent::{ default_quantile, AggIntent, }; pub use cse::{dedupe_subtrees, CseWorkloadPlan}; -pub use lower::{lower_parsed_query, LoweringError}; pub use query_expr::{ from_legacy_scalar, BinaryOpKind, BindingScope, ColumnRef, GroupSide, HavingPredicate, JoinKind, LabelFilter, LiteralValue, PartitionKeys, Predicate, ProjectItem, QueryExpr, diff --git a/control_plane/src/intent_algebra/relational.rs b/control_plane/src/intent_algebra/relational.rs index a45b0e78..84d446fd 100644 --- a/control_plane/src/intent_algebra/relational.rs +++ b/control_plane/src/intent_algebra/relational.rs @@ -217,7 +217,7 @@ pub enum FilterVal { /// /// This is purely a Layer-2 *relational* IR — the sketch-fused /// `SketchAgg` / `WindowedAgg` variants were removed once the -/// `lower_to_canonical` converter learned to fold the single-statistic +/// `lower` 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 [`ScalarExpr`] /// still carries four variants (`FunctionCall`, `ScalarSubquery`, @@ -272,7 +272,7 @@ pub enum QueryExpr { // The sketch-fused `SketchAgg` / `WindowedAgg` variants were removed: // they were never parser output — only an intermediate of an earlier - // sketch-lowering pass — and `lower_to_canonical` now folds the + // sketch-lowering pass — and `lower` now folds the // single-statistic sketchable `Aggregate` straight into canonical // shapes (`Aggregate { by: [] }` / `Window { Aggregate }`). @@ -503,7 +503,7 @@ impl AggFunc { /// /// Canonical Quantile is single-φ; this helper returns one canonical /// intent. Callers that need multi-φ behaviour build the merge fan-out - /// themselves (cf. `lower_to_canonical::agg_func_to_intents`). + /// themselves (cf. `lower::agg_func_to_intents`). pub fn to_sketch_op(&self) -> Option { match self { AggFunc::Quantile(phi) => Some(default_quantile(*phi)), diff --git a/control_plane/src/lib.rs b/control_plane/src/lib.rs index c601171a..29e6c337 100644 --- a/control_plane/src/lib.rs +++ b/control_plane/src/lib.rs @@ -42,7 +42,7 @@ //! lookup table (Phase 4 / 5 use this to route raw-name PromQL). //! - `intent_algebra` — `AggIntent` + `QueryExpr` DAG (canonical L3 IR; //! `relational` carries the L2 relational IR the parsers emit and -//! `lower_to_canonical` lowers it to the canonical L3 types). +//! `lower` lowers it to the canonical L3 types). //! - `query_parser` — front-end parsers (Layer 1): `promql.rs` / `sql.rs` //! emit the L2 relational `relational::QueryExpr` tree. //! - `physical` — L5 framework (allocator, planner, plan, sketch_catalog, diff --git a/control_plane/src/physical/window_fusion.rs b/control_plane/src/physical/window_fusion.rs index 7437eb73..e711ccc2 100644 --- a/control_plane/src/physical/window_fusion.rs +++ b/control_plane/src/physical/window_fusion.rs @@ -10,7 +10,7 @@ //! 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. -//! `lower_to_canonical` produces exactly that shape when it folds a +//! `lower` 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 @@ -53,7 +53,7 @@ pub struct FusedWindowSketch<'a> { } /// Recognize the canonical `Window { child: Aggregate { aggs: [one], -/// having: None, .. } }` shape — the stacked form `lower_to_canonical` +/// having: None, .. } }` shape — the stacked form `lower` /// folds a legacy `WindowedAgg` into — and return a [`FusedWindowSketch`] /// view of it. Returns `None` for any other shape (a multi-intent /// `Aggregate`, an `Aggregate` with a `having` clause, a `Window` over a @@ -172,7 +172,7 @@ pub fn fused_sketch_decision( // // Pins `recognize_windowed_sketch` + `fused_sketch_decision` against the // canonical planner: a `Window { Aggregate }` fused sketch — the shape -// `lower_to_canonical` folds a single-statistic sketchable `Aggregate` +// `lower` 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`. @@ -206,7 +206,7 @@ mod tests { } /// The canonical `Window { Aggregate { by: [], aggs: [agg] } }` shape — - /// the fold `lower_to_canonical` produces for a windowed single-sketch + /// the fold `lower` produces for a windowed single-sketch /// aggregate. fn windowed_sketch( agg: AggIntent, diff --git a/control_plane/src/sketch_algebra/lower.rs b/control_plane/src/sketch_algebra/lower.rs index 2a915c0d..c8f9ebdd 100644 --- a/control_plane/src/sketch_algebra/lower.rs +++ b/control_plane/src/sketch_algebra/lower.rs @@ -72,10 +72,46 @@ fn bind_recursive(expr: &QueryExpr, accuracy: &AccuracyTarget) -> PhysicalExpr { child: Box::new(bind_recursive(child, accuracy)), }, QueryExpr::Ref { name } => PhysicalExpr::Ref { name: name.clone() }, + // The canonical L3 IR places `Window` *above* a single-statistic + // sketchable `Aggregate` (`lower`'s window-swap). + // The `Bind*` rules match `Aggregate` with the window as its + // *child*, so push the window down under the aggregate and + // re-dispatch — the window then rides along inside the bound + // node's `Logical(...)` child, exactly as it did when the + // aggregate sat on top. A `Window` over anything else stays a + // logical pass-through. + QueryExpr::Window { + kind, + size, + slide, + child, + } if matches!(child.as_ref(), QueryExpr::Aggregate { .. }) => { + let QueryExpr::Aggregate { + by, + aggs, + having, + child: agg_child, + } = child.as_ref() + else { + unreachable!("guarded by the `matches!` above") + }; + let pushed = QueryExpr::Aggregate { + by: by.clone(), + aggs: aggs.clone(), + having: having.clone(), + child: Box::new(QueryExpr::Window { + kind: kind.clone(), + size: *size, + slide: *slide, + child: agg_child.clone(), + }), + }; + bind_recursive(&pushed, accuracy) + } // For `Aggregate`, the rule dispatcher already had a chance and - // declined. For `Scan` and `Window`, the recursive walk is a - // no-op (no binding rule applies to these shapes today). In all - // cases, wrap the L3 sub-tree as a logical pass-through. + // declined. For `Scan` and a `Window` over a non-`Aggregate` + // child, no binding rule applies — wrap the L3 sub-tree as a + // logical pass-through. QueryExpr::Aggregate { .. } | QueryExpr::Scan { .. } | QueryExpr::Window { .. } => { PhysicalExpr::Logical(expr.clone()) } diff --git a/control_plane/src/sketch_algebra/tests.rs b/control_plane/src/sketch_algebra/tests.rs index c643728b..47a2bcb7 100644 --- a/control_plane/src/sketch_algebra/tests.rs +++ b/control_plane/src/sketch_algebra/tests.rs @@ -542,16 +542,14 @@ fn phase_b_pattern_archive_only_routes_to_archive() { // representative query string from the corresponding fixture YAML and // pins the expected (sketch_kind | archive-only) outcome. -/// Helper: parse a PromQL string, lower to L3, bind to L4. Returns the -/// produced `PhysicalExpr` for assertion. The control plane's `parse_query` -/// returns a `ParsedQuery`; `lower_parsed_query` builds the L3 IR from -/// it under the supplied accuracy target; `bind_query_expr` is the L3→L4 -/// bottom-up walk. +/// Helper: parse a PromQL string to the canonical L3 `QueryExpr`, bind to +/// L4. Returns the produced `PhysicalExpr` for assertion. +/// `parse_query_expr_canonical` is the real parse path (L1 → L2 relational +/// → L3 canonical via `lower`); `bind_query_expr` is the +/// L3→L4 bottom-up walk under the supplied accuracy target. fn pipeline_l1_to_l4(query: &str, accuracy: AccuracyTarget) -> PhysicalExpr { - let parsed = - crate::query_parser::parse_query(query).unwrap_or_else(|e| panic!("parse {query}: {e}")); - let qe = crate::intent_algebra::lower_parsed_query(&parsed, accuracy.clone()) - .unwrap_or_else(|e| panic!("lower {query}: {e}")); + let qe = crate::query_parser::parse_query_expr_canonical(query) + .unwrap_or_else(|e| panic!("parse {query}: {e}")); bind_query_expr(&qe, accuracy).unwrap_or_else(|e| panic!("bind {query}: {e}")) }