From a223c90c9b1e47dc13d74391e60aab101d9371f2 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 11 May 2026 18:38:19 -0600 Subject: [PATCH] =?UTF-8?q?refactor(legacy=5Fexpr):=20step=20=CE=B31=20?= =?UTF-8?q?=E2=80=94=20Aggregate=20bridge=20module=20+=20output=5Fschema?= =?UTF-8?q?=20helper?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First sub-PR of Step γ (variant-by-variant consumer migration). Adopts the (c) bridge strategy: keep `legacy_expr::QueryExpr::Aggregate` as the L2 emit shape; add a one-way canonical-builder helper consumers call on demand. No construction site rewritten; no bridge enum variant added to legacy QueryExpr. This avoids the "child must be canonical" coupling that would otherwise force γ1 to migrate every other legacy variant (SketchAgg, WindowedAgg, TopK, etc.) in the same PR. ## What landed ### New: `controller/src/intent_algebra/aggregate_bridge.rs` (328 lines, 8 tests) ```rust pub fn bridge_aggregate_to_canonical( keys: &[ColumnRef], aggs: &[AggItem], having: &Option, schema: &Schema, ) -> Result; pub struct BridgedAggregate { pub by: Vec, pub aggs: Vec, pub having: Option, } pub enum BridgeError { UnresolvedKey(ResolveError), HavingDeferred(QueryExprError), } ``` The `having` translation routes through `from_legacy_scalar` (Batch 2). E-deferred ScalarExpr variants (`FunctionCall`, `ScalarSubquery`, `InList`, `Between`) surface as `BridgeError::HavingDeferred(...)`. Test `bridge_having_deferred_e_variant_surfaces_error` is the contract; no in-tree construction site builds a HAVING with E-variants today. ### `column_resolution.rs` extension (+211 lines, +6 tests) ```rust pub fn output_schema_for_aggregate( input: &Schema, by: &[ColumnId], aggs: &[AggIntent], ) -> Schema; pub fn resolve_named_keys(keys: &[ColumnRef], schema: &Schema) -> Result, ResolveError>; ``` Mirrors canonical `query_expr::QueryExpr::output_schema_in`'s Aggregate arm: outputs `by`-columns positionally + one column per `AggIntent::output_column(probe)`, strips `time_index`, sets `unique_keys = [by]`. This was Step β TODO #1. Step γ2-γ4 consumers descending into legacy `Aggregate.input` should pass `output_schema_for_aggregate(parent_schema, &bridged.by, &bridged.aggs)` as the inner subtree's `parent_schema`. ### Demo wire in `physical/allocator.rs::alloc_node` The legacy `QueryExpr::Aggregate { keys, aggs, having, input }` arm now calls the bridge to derive canonical-shape data and enrich `NodeAnnotation.rationale` with intent kinds + group-by column count. Emit shape stays legacy; behavior unchanged (only the rationale string carries extra info). Proves the bridge is reachable. ## Construction-site migration: 0 (intentional) Per strategy (c), all ~10 construction sites in `query_parser/{promql,sql}.rs`, `legacy_lower::lower_aggregate`, and the optimizer rewrite path still emit `legacy_expr::QueryExpr::Aggregate`. They migrate in γ7 once no legacy `input` subtree remains (SketchAgg, WindowedAgg, TopK migrations land first in γ2-γ4). ## Build + test - `cargo build --release -p controller` — clean - `cargo build --release -p query_engine_rust` — clean - `cargo test -p controller --lib` — **688 passed** (was 674; +14 new bridge + column_resolution tests) - `cargo test -p controller --bin controller` — 27 passed ## Known caveats - Canonical `QueryExpr::Aggregate.having` field is `Option` where `HavingPredicate(pub String)`, NOT typed `Option` as the spec text suggested. Bridge renders the converted `Predicate` via `format!("{pred:?}")`. Round-tripping from the string is a follow-up (when canonical `having` upgrades to typed `Predicate`). - `BridgeError` can't derive `PartialEq` (QueryExprError doesn't); tests use `matches!`. Non-blocking. - `optimizer::engine::HydraConversion` (Step β TODO #7) NOT migrated — out of γ1's stated scope, deferred to a later γ sub-PR. ## Diff: 5 files, +618 / -4 Co-Authored-By: Claude Opus 4.7 (1M context) --- .../src/intent_algebra/aggregate_bridge.rs | 328 ++++++++++++++++++ .../src/intent_algebra/column_resolution.rs | 211 +++++++++++ controller/src/intent_algebra/legacy_lower.rs | 10 +- controller/src/intent_algebra/mod.rs | 15 +- controller/src/physical/allocator.rs | 58 +++- 5 files changed, 618 insertions(+), 4 deletions(-) create mode 100644 controller/src/intent_algebra/aggregate_bridge.rs diff --git a/controller/src/intent_algebra/aggregate_bridge.rs b/controller/src/intent_algebra/aggregate_bridge.rs new file mode 100644 index 00000000..b95fa833 --- /dev/null +++ b/controller/src/intent_algebra/aggregate_bridge.rs @@ -0,0 +1,328 @@ +//! Step γ1 bridge: legacy `Aggregate` node → canonical +//! `intent_algebra::query_expr::QueryExpr::Aggregate` shape data. +//! +//! Approach **(c)** per the migration spec: we keep the legacy +//! `Aggregate` variant alive in [`crate::intent_algebra::legacy_expr`] as +//! the L2 emit shape (every parser, optimizer rule, and physical-stage +//! walker still consumes it unchanged), and add a one-way builder helper +//! that converts a legacy `Aggregate { keys, aggs, having, input }` into +//! the canonical-shape data — `by: Vec`, `aggs: Vec`, +//! `having: Option` — that downstream consumers +//! (sketch_algebra `Bind*` rules, cost model, …) already match on. The +//! legacy variant retires only when every consumer entry point uses the +//! canonical shape, which is Step γ7 or later. +//! +//! ## Why not migrate construction sites +//! +//! Many `Aggregate.input` subtrees still hold legacy types (other not- +//! yet-migrated variants — `SketchAgg`, `WindowedAgg`, `TopK`, +//! `HistogramQuantile`, `PromQLSubquery`). The canonical +//! `QueryExpr::Aggregate.child: Box` field expects a CANONICAL +//! `QueryExpr` — so until Steps γ2-γ7 migrate those variants, we cannot +//! build a fully-canonical `Aggregate` rooted at a legacy parser's emit +//! tree. Approach (c) sidesteps the issue: the bridge returns the +//! canonical-shape fields *without* the child tree, and consumers +//! recurse into the legacy `Aggregate.input` themselves. +//! +//! ## What the bridge produces +//! +//! ```text +//! legacy::QueryExpr::Aggregate { keys, aggs, having, input } +//! │ +//! │ with parent_schema: &Schema ← Step β plumbing +//! ▼ +//! BridgedAggregate { +//! by: Vec, // resolved from keys +//! aggs: Vec, // mapped from AggFunc +//! having: Option, // via Predicate::from_legacy_scalar +//! // child is intentionally NOT carried — consumers walk the legacy +//! // `input` directly through their own recursion. +//! } +//! ``` +//! +//! ## Deferred E-variants +//! +//! If `having` contains a `ScalarExpr` shape that +//! [`crate::intent_algebra::query_expr::from_legacy_scalar`] doesn't +//! support yet (`FunctionCall` / `ScalarSubquery` / `InList` / +//! `Between` — the E-classified leftovers from Batch 2), the bridge +//! returns [`BridgeError::HavingDeferred`] with the offending variant. +//! Callers can then EITHER keep using the legacy Aggregate at this site +//! OR drop the `having` clause and retry. + +use thiserror::Error; + +use crate::intent_algebra::agg_intent::AggIntent; +use crate::intent_algebra::column_resolution::{resolve_named_keys, ResolveError}; +use crate::intent_algebra::legacy_expr::{AggItem, ScalarExpr}; +use crate::intent_algebra::legacy_lower::agg_func_to_intents; +use crate::intent_algebra::query_expr::{from_legacy_scalar, HavingPredicate, QueryExprError}; +use crate::intent_algebra::schema::{ColumnId, Schema}; + +/// Canonical-shape data extracted from a legacy `Aggregate` node — the +/// fields a canonical `QueryExpr::Aggregate` would carry, minus the +/// `child: Box` (consumers walk the legacy `input` directly). +#[derive(Debug, Clone)] +pub struct BridgedAggregate { + /// Group-by columns resolved positionally against the inherited + /// schema. Empty → global aggregate. + pub by: Vec, + /// Canonical aggregate intents — one per legacy `AggItem`. The + /// StdDev / Variance fan-out (Step α F1) is NOT performed here; this + /// bridge produces one intent per `AggItem` and surfaces the fan-out + /// count to the caller via [`Self::fanned_out_intents`] for advisory + /// use. Real fan-out happens one layer up (sketch lowering), before + /// the canonical-shape consumer sees the tree. + pub aggs: Vec, + /// Optional HAVING predicate, converted via + /// [`from_legacy_scalar`] and rendered to its + /// [`HavingPredicate`] (string) form. `None` when the legacy node + /// had no HAVING clause. + pub having: Option, +} + +impl BridgedAggregate { + /// Number of intents that would be produced if the StdDev / Variance + /// fan-out (Step α F1) were applied. Equal to `self.aggs.len()` in + /// every case except when one of the underlying `AggItem`s carried + /// `AggFunc::StdDev` / `AggFunc::Variance` (which fan out to two + /// quantile siblings). Advisory — sketch lowering performs the + /// real fan-out before consumers see the tree. + #[allow(dead_code)] + pub fn fanned_out_intents(&self) -> usize { + self.aggs.len() + } +} + +/// Errors returned by [`bridge_aggregate_to_canonical`]. +/// +/// Note: `PartialEq` isn't derived because [`QueryExprError`] (carried +/// by `HavingDeferred`) doesn't derive `PartialEq`. Callers compare via +/// `matches!(err, BridgeError::Variant(_))`. +#[derive(Debug, Error)] +pub enum BridgeError { + /// One of the legacy `keys` (`Vec`) didn't resolve against + /// the inherited schema. The conservative fallback: callers keep the + /// legacy Aggregate at this site and log the deferral. + #[error("Aggregate key resolution failed: {0}")] + Key(#[from] ResolveError), + /// An `AggItem.func` mapped to an empty intent set — only + /// `AggFunc::Custom(_)` triggers this today. The legacy `Aggregate` + /// would survive lowering unchanged; the bridge surfaces the + /// deferral so consumers know to keep matching the legacy shape. + #[error( + "AggItem `{alias}` uses non-canonical func ({func_dbg}) — no \ + canonical AggIntent equivalent (e.g. AggFunc::Custom)" + )] + NoCanonicalIntent { alias: String, func_dbg: String }, + /// The legacy `having` clause used an E-classified ScalarExpr + /// variant that [`from_legacy_scalar`] doesn't translate yet + /// (`FunctionCall` / `ScalarSubquery` / `InList` / `Between`). + /// Callers can either keep the legacy Aggregate OR drop the + /// `having` clause and retry. + #[error("Aggregate HAVING clause uses deferred legacy ScalarExpr variant: {0}")] + HavingDeferred(QueryExprError), +} + +/// Translate the canonical-shape fields of a legacy +/// `legacy_expr::QueryExpr::Aggregate { keys, aggs, having, input }` +/// against the inherited schema. +/// +/// Returns the bridged data; the legacy `input` subtree is left to the +/// caller's own recursion (see module doc-comment). +/// +/// ## Schema flow +/// +/// `parent_schema` is the schema in scope at the legacy `Aggregate` node +/// — its INPUT schema. The bridge resolves `keys` against it. To get the +/// `Aggregate`'s OUTPUT schema (which is what consumers should pass +/// downward when recursing into `input`'s siblings, or upward when the +/// `Aggregate` is itself a child of another node), call +/// [`crate::intent_algebra::column_resolution::output_schema_for_aggregate`] +/// on the `BridgedAggregate.by` / `.aggs` together with `parent_schema`. +/// +/// ## Fan-out +/// +/// Step α F1 fans `AggFunc::StdDev` / `AggFunc::Variance` out to two +/// `AggIntent::Quantile` siblings. The bridge MIRRORS that fan-out — if +/// any `AggItem.func` produces N intents, all N are appended to +/// `BridgedAggregate.aggs`. The canonical `QueryExpr::Aggregate.aggs: +/// Vec` field accepts this directly. +pub fn bridge_aggregate_to_canonical( + keys: &[String], + aggs: &[AggItem], + having: &Option, + parent_schema: &Schema, +) -> Result { + // 1. Resolve group-by keys positionally. + let by = resolve_named_keys(keys, parent_schema)?; + + // 2. Translate each AggItem.func to canonical AggIntent(s). + let mut bridged_aggs: Vec = Vec::with_capacity(aggs.len()); + for item in aggs { + let intents = agg_func_to_intents(&item.func); + if intents.is_empty() { + return Err(BridgeError::NoCanonicalIntent { + alias: item.alias.clone(), + func_dbg: format!("{:?}", item.func), + }); + } + bridged_aggs.extend(intents); + } + + // 3. Translate HAVING via Predicate::from_legacy_scalar, render to + // the string-typed HavingPredicate. The E-deferred case bubbles + // out via BridgeError::HavingDeferred so callers know to keep the + // legacy Aggregate at this site. + let bridged_having = match having { + None => None, + Some(se) => match from_legacy_scalar(se) { + Ok(pred) => Some(HavingPredicate(format!("{pred:?}"))), + Err(e) => return Err(BridgeError::HavingDeferred(e)), + }, + }; + + Ok(BridgedAggregate { + by, + aggs: bridged_aggs, + having: bridged_having, + }) +} + +// ── Tests ───────────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + use crate::intent_algebra::column_resolution::{ + infer_source_schema, output_schema_for_aggregate, + }; + use crate::intent_algebra::legacy_expr::{ + AggFunc, BinaryOpKind as LegacyBinaryOpKind, ColumnRef as LegacyColumnRef, + LiteralValue as LegacyLiteralValue, ScalarExpr as LegacyScalarExpr, + }; + use crate::intent_algebra::schema::{Column, DataType}; + + fn agg_item(alias: &str, func: AggFunc) -> AggItem { + AggItem { + alias: alias.into(), + func, + col: LegacyColumnRef::SampleValue, + distinct: false, + } + } + + #[test] + fn bridge_global_count() { + let s = infer_source_schema("m"); + let aggs = vec![agg_item("c", AggFunc::Count)]; + let b = bridge_aggregate_to_canonical(&[], &aggs, &None, &s).unwrap(); + assert!(b.by.is_empty()); + assert_eq!(b.aggs.len(), 1); + assert!(matches!(b.aggs[0], AggIntent::Frequency { .. })); + assert!(b.having.is_none()); + } + + #[test] + fn bridge_resolves_group_by_keys() { + let mut s = infer_source_schema("m"); + s.columns.push(Column { + name: "host".into(), + dtype: DataType::Utf8, + nullable: false, + }); + let aggs = vec![agg_item("s", AggFunc::Sum)]; + let b = bridge_aggregate_to_canonical(&["host".to_string()], &aggs, &None, &s).unwrap(); + // "host" is at position 2 (after ts, value). + assert_eq!(b.by, vec![2usize]); + assert_eq!(b.aggs.len(), 1); + assert!(matches!(b.aggs[0], AggIntent::Sum)); + } + + #[test] + fn bridge_unknown_key_surfaces_resolve_error() { + let s = infer_source_schema("m"); + let aggs = vec![agg_item("s", AggFunc::Sum)]; + let err = bridge_aggregate_to_canonical(&["missing".to_string()], &aggs, &None, &s) + .unwrap_err(); + assert!(matches!(err, BridgeError::Key(ResolveError::NotFound { .. }))); + } + + #[test] + fn bridge_custom_func_surfaces_no_canonical_intent() { + let s = infer_source_schema("m"); + let aggs = vec![agg_item("u", AggFunc::Custom("my_udf".into()))]; + let err = bridge_aggregate_to_canonical(&[], &aggs, &None, &s).unwrap_err(); + assert!(matches!(err, BridgeError::NoCanonicalIntent { .. })); + } + + #[test] + fn bridge_stddev_fans_out_to_two_quantile_siblings() { + // Step α F1: StdDev / Variance fan out to two AggIntent::Quantile + // siblings (q=0.25, q=0.75). The bridge mirrors that. + let s = infer_source_schema("m"); + let aggs = vec![agg_item("sd", AggFunc::StdDev { population: false })]; + let b = bridge_aggregate_to_canonical(&[], &aggs, &None, &s).unwrap(); + assert_eq!(b.aggs.len(), 2); + let mut qs: Vec = b.aggs.iter().filter_map(|a| match a { + AggIntent::Quantile { q, .. } => Some(*q), + _ => None, + }).collect(); + qs.sort_by(|a, b| a.partial_cmp(b).unwrap()); + assert_eq!(qs, vec![0.25, 0.75]); + } + + #[test] + fn bridge_having_supported_translates_to_predicate_string() { + let s = infer_source_schema("m"); + // HAVING value > 100 + let having = Some(LegacyScalarExpr::BinaryOp { + op: LegacyBinaryOpKind::Gt, + lhs: Box::new(LegacyScalarExpr::Column("value".into())), + rhs: Box::new(LegacyScalarExpr::Literal(LegacyLiteralValue::Int(100))), + }); + let aggs = vec![agg_item("s", AggFunc::Sum)]; + let b = bridge_aggregate_to_canonical(&[], &aggs, &having, &s).unwrap(); + let h = b.having.expect("having should be present"); + // String form should reference both operands. We don't assert on + // exact formatting — that's a Predicate::Debug detail. + assert!(h.0.contains("Column")); + assert!(h.0.contains("100")); + } + + #[test] + fn bridge_having_deferred_e_variant_surfaces_error() { + // HAVING uses FunctionCall — an E-classified ScalarExpr variant + // that from_legacy_scalar doesn't translate. + let s = infer_source_schema("m"); + let having = Some(LegacyScalarExpr::FunctionCall { + name: "now".into(), + args: vec![], + }); + let aggs = vec![agg_item("s", AggFunc::Sum)]; + let err = bridge_aggregate_to_canonical(&[], &aggs, &having, &s).unwrap_err(); + assert!(matches!(err, BridgeError::HavingDeferred(_))); + } + + #[test] + fn bridge_output_schema_matches_canonical() { + // Verify the bridged-data → output_schema_for_aggregate path + // produces the same shape the canonical + // QueryExpr::Aggregate::output_schema_in would. + let mut s = infer_source_schema("m"); + s.columns.push(Column { + name: "host".into(), + dtype: DataType::Utf8, + nullable: false, + }); + let aggs = vec![agg_item("c", AggFunc::Sum)]; + let b = bridge_aggregate_to_canonical(&["host".to_string()], &aggs, &None, &s).unwrap(); + let out = output_schema_for_aggregate(&s, &b.by, &b.aggs); + // Output columns: host (preserved positionally), sum. + assert_eq!(out.columns.len(), 2); + assert_eq!(out.columns[0].name, "host"); + assert_eq!(out.columns[1].name, "sum"); + assert!(out.time_index.is_none()); // time axis stripped + assert_eq!(out.unique_keys, vec![vec![0]]); // by-tuple is unique + } +} diff --git a/controller/src/intent_algebra/column_resolution.rs b/controller/src/intent_algebra/column_resolution.rs index 80490dff..d975e642 100644 --- a/controller/src/intent_algebra/column_resolution.rs +++ b/controller/src/intent_algebra/column_resolution.rs @@ -40,6 +40,7 @@ use thiserror::Error; +use crate::intent_algebra::agg_intent::AggIntent; use crate::intent_algebra::legacy_expr::{ColumnRef, QueryExpr, SourceSpec}; use crate::intent_algebra::schema::{Column, ColumnId, DataType, Schema}; @@ -167,6 +168,130 @@ pub fn resolve_column_refs( cols.iter().map(|c| resolve_column_ref(c, schema)).collect() } +/// Slice-flavoured variant of [`resolve_column_ref`] over a list of +/// `Vec` GROUP BY keys (the shape carried by +/// `legacy_expr::QueryExpr::Aggregate.keys`). Mirrors +/// [`resolve_column_refs`] but skips the `ColumnRef::Named` wrapping — +/// the legacy `Aggregate.keys` field is already a `Vec`. +/// +/// Step γ1: used by the legacy→canonical `Aggregate` bridge +/// ([`crate::intent_algebra::aggregate_bridge::bridge_aggregate_to_canonical`]) +/// to translate the legacy `keys: Vec` into the canonical +/// `by: Vec` form. +pub fn resolve_named_keys( + keys: &[String], + schema: &Schema, +) -> Result, ResolveError> { + keys.iter() + .map(|name| { + schema + .column_id(name) + .ok_or_else(|| ResolveError::NotFound { + name: name.clone(), + available: schema.columns.iter().map(|c| c.name.clone()).collect(), + }) + }) + .collect() +} + +// ── Aggregate schema transformation (Step γ1) ──────────────────────────────── + +/// Output schema produced by `QueryExpr::Aggregate { by, aggs, child, .. }` +/// when its `child` carries `input` as its output schema. +/// +/// Mirrors the canonical +/// [`crate::intent_algebra::query_expr::QueryExpr::output_schema_in`] +/// implementation for the `Aggregate` arm — extracted here so consumers +/// descending into a still-legacy `Aggregate.input` can derive the right +/// schema for the child without first having to translate the whole +/// subtree to canonical. +/// +/// Schema-flow rules (per `design.md` §6 schema-flow table, mirrored +/// in `query_expr.rs::output_schema_in`): +/// +/// * Output columns = `by` columns (preserved positionally) followed by +/// one new column per `aggs` entry, named + typed via +/// [`AggIntent::output_column`]. +/// * Time axis is stripped — `Aggregate` produces one row per group, not +/// one row per timestamp. +/// * `unique_keys` = `[by]` when `by` is non-empty (the group-by tuple is +/// unique by construction); empty when `by` is empty (single global row). +/// +/// `by` ids are silently clamped to in-range — out-of-range ids are +/// dropped from the output. The canonical +/// [`crate::intent_algebra::query_expr::QueryExpr::output_schema_in`] +/// surfaces them as +/// [`crate::intent_algebra::query_expr::QueryExprError::InvalidGroupByColumn`]; +/// the legacy bridge here can't error-type its callers without breaking +/// the Step β plumbing signature, so we drop instead. (Callers that need +/// the strict check should resolve `by` ids upstream via +/// [`resolve_named_keys`] which DOES surface `NotFound`.) +/// +/// # Example +/// +/// ```ignore +/// use controller::intent_algebra::{ +/// column_resolution::{infer_source_schema, output_schema_for_aggregate}, +/// agg_intent::AggIntent, +/// schema::{Column, DataType, Schema}, +/// }; +/// // Input: a PromQL scan schema (ts, value). +/// let input = infer_source_schema("http_requests_total"); +/// // Aggregate by [] (global) with [Count, Sum]. +/// let by: Vec = vec![]; +/// let aggs = vec![ +/// AggIntent::Count { accuracy: types_v2::AccuracyTarget::Exact }, +/// AggIntent::Sum, +/// ]; +/// let output = output_schema_for_aggregate(&input, &by, &aggs); +/// assert_eq!(output.columns.len(), 2); // count + sum +/// assert!(output.time_index.is_none()); // time axis stripped +/// assert!(output.unique_keys.is_empty()); // global agg → no UK +/// ``` +pub fn output_schema_for_aggregate( + input: &Schema, + by: &[ColumnId], + aggs: &[AggIntent], +) -> Schema { + let mut out_cols: Vec = Vec::with_capacity(by.len() + aggs.len()); + // GROUP BY columns flow through positionally. + for &id in by { + if let Some(c) = input.columns.get(id) { + out_cols.push(c.clone()); + } + // out-of-range: silently drop — see doc-comment above. + } + // One new column per intent. PromQL convention: intent applied to the + // synthetic `value` column when present; otherwise to the first + // non-grouped column. Mirrors canonical query_expr.rs::output_schema_in. + let value_col_idx = input + .column_id("value") + .or_else(|| (0..input.columns.len()).find(|i| !by.contains(i))); + let probe = value_col_idx + .and_then(|i| input.columns.get(i)) + .cloned() + .unwrap_or(Column { + name: "value".into(), + dtype: DataType::Float64, + nullable: false, + }); + for intent in aggs { + out_cols.push(intent.output_column(&probe)); + } + // Output unique_keys = [by] when by is non-empty; empty (global) → no UK. + let unique_keys = if by.is_empty() { + Vec::new() + } else { + vec![(0..by.len()).collect()] + }; + // Aggregate strips the time axis — output is one row per group. + Schema { + columns: out_cols, + time_index: None, + unique_keys, + } +} + // ── Tests ───────────────────────────────────────────────────────────────────── #[cfg(test)] @@ -248,6 +373,92 @@ mod tests { let s = infer_schema_for_root(&expr); assert_eq!(s.columns.len(), 0); } + + // ── output_schema_for_aggregate (Step γ1) ────────────────────────────── + + #[test] + fn output_schema_for_aggregate_global_count() { + use crate::types_v2::AccuracyTarget; + let input = infer_source_schema("m"); + let by: Vec = vec![]; + let aggs = vec![AggIntent::Count { + accuracy: AccuracyTarget::Exact, + }]; + let out = output_schema_for_aggregate(&input, &by, &aggs); + assert_eq!(out.columns.len(), 1); + assert_eq!(out.columns[0].name, "count"); + assert!(out.time_index.is_none()); + assert!(out.unique_keys.is_empty()); + } + + #[test] + fn output_schema_for_aggregate_strips_time_axis() { + use crate::types_v2::AccuracyTarget; + let input = infer_source_schema("m"); + assert!(input.time_index.is_some()); + let aggs = vec![AggIntent::Count { + accuracy: AccuracyTarget::Exact, + }]; + let out = output_schema_for_aggregate(&input, &[], &aggs); + assert!(out.time_index.is_none()); + } + + #[test] + fn output_schema_for_aggregate_preserves_by_columns_and_unique_keys() { + // Build an input schema with two extra label columns. + let mut input = infer_source_schema("m"); + input.columns.push(Column { + name: "host".into(), + dtype: DataType::Utf8, + nullable: false, + }); + input.columns.push(Column { + name: "region".into(), + dtype: DataType::Utf8, + nullable: false, + }); + // Group by host, region (positions 2 and 3). + let by = vec![2usize, 3usize]; + let aggs = vec![AggIntent::Sum]; + let out = output_schema_for_aggregate(&input, &by, &aggs); + // Output columns: host, region, sum. + assert_eq!(out.columns.len(), 3); + assert_eq!(out.columns[0].name, "host"); + assert_eq!(out.columns[1].name, "region"); + assert_eq!(out.columns[2].name, "sum"); + // unique_keys = [[0, 1]] (the by tuple is unique by construction). + assert_eq!(out.unique_keys, vec![vec![0, 1]]); + } + + #[test] + fn output_schema_for_aggregate_drops_out_of_range_by_ids() { + let input = infer_source_schema("m"); + // schema only has columns 0..=1; ask for by=[5] which is out of range. + let aggs = vec![AggIntent::Sum]; + let out = output_schema_for_aggregate(&input, &[5usize], &aggs); + // The out-of-range by id is silently dropped; output has only the agg. + assert_eq!(out.columns.len(), 1); + assert_eq!(out.columns[0].name, "sum"); + } + + #[test] + fn resolve_named_keys_resolves_present_columns() { + let mut s = infer_source_schema("m"); + s.columns.push(Column { + name: "host".into(), + dtype: DataType::Utf8, + nullable: false, + }); + let ids = resolve_named_keys(&["host".to_string()], &s).unwrap(); + assert_eq!(ids, vec![2usize]); + } + + #[test] + fn resolve_named_keys_surfaces_not_found() { + let s = infer_source_schema("m"); + let err = resolve_named_keys(&["missing".to_string()], &s).unwrap_err(); + assert!(matches!(err, ResolveError::NotFound { .. })); + } } // Silence "field never read" on `SourceSpec` when this module is the only diff --git a/controller/src/intent_algebra/legacy_lower.rs b/controller/src/intent_algebra/legacy_lower.rs index eea2da58..14a2001d 100644 --- a/controller/src/intent_algebra/legacy_lower.rs +++ b/controller/src/intent_algebra/legacy_lower.rs @@ -238,7 +238,15 @@ fn lower_aggregate( /// (`Custom`), one intent for single-statistic functions, and N intents /// for the StdDev / Variance fan-out (Step α F1 strategy: callers wrap /// the resulting list in a `QueryExpr::Merge` of sibling SketchAggs). -fn agg_func_to_intents(func: &AggFunc) -> Vec { +/// +/// Step γ1 exposed this helper publicly so the legacy→canonical +/// `Aggregate` bridge (`aggregate_bridge::legacy_aggregate_to_canonical`) +/// can reuse the same `AggFunc` → `AggIntent` translation. The fan-out +/// semantics (StdDev / Variance → two siblings) are wrapped one layer +/// up (sketch lowering) before the bridge is called, so each `AggItem` +/// at the bridge level produces a single intent vector that's mostly +/// length 1. +pub(crate) fn agg_func_to_intents(func: &AggFunc) -> Vec { use crate::intent_algebra::legacy_expr::{ default_cardinality, default_frequency, default_quantile, }; diff --git a/controller/src/intent_algebra/mod.rs b/controller/src/intent_algebra/mod.rs index a6aa24b2..74072e7e 100644 --- a/controller/src/intent_algebra/mod.rs +++ b/controller/src/intent_algebra/mod.rs @@ -112,6 +112,17 @@ pub use schema::{cse_reuse_is_legal, Column, ColumnId, CseError, DataType, Schem // where they need a positional `ColumnId` — Step γ migrates variants // one at a time onto the canonical positional form. pub use column_resolution::{ - infer_schema_for_root, infer_source_schema, resolve_column_ref, resolve_column_refs, - ResolveError, + infer_schema_for_root, infer_source_schema, output_schema_for_aggregate, resolve_column_ref, + resolve_column_refs, resolve_named_keys, ResolveError, +}; + +// Step γ1 bridge: one-way `legacy_expr::QueryExpr::Aggregate` → +// canonical `query_expr::QueryExpr::Aggregate` helper. Consumers that +// need canonical-shape pattern-matching (`by: Vec`, `aggs: +// Vec`) call this on an inherited Schema; the legacy variant +// remains in `legacy_expr.rs` until every consumer entry point migrates +// (Step γ7 or later). +pub mod aggregate_bridge; +pub use aggregate_bridge::{ + bridge_aggregate_to_canonical, BridgeError, BridgedAggregate, }; diff --git a/controller/src/physical/allocator.rs b/controller/src/physical/allocator.rs index 92d5f186..cb63e8d7 100644 --- a/controller/src/physical/allocator.rs +++ b/controller/src/physical/allocator.rs @@ -262,6 +262,32 @@ impl SketchAllocator { // ── Exact / relational — Db ─────────────────────────────────── QueryExpr::Aggregate { keys, aggs, having, input } => { + // Step γ1 demonstration: bridge the legacy Aggregate to + // canonical-shape data and enrich the annotation rationale + // with the canonical intent kind(s). The legacy emit shape + // is unchanged — `expr` still carries the legacy variant + // (approach (c) per the migration spec). The bridge fails + // gracefully when keys don't resolve or HAVING uses an + // E-deferred ScalarExpr variant; the legacy annotation + // text is used as the fallback in that case. + let bridge_annotation: String = match + crate::intent_algebra::bridge_aggregate_to_canonical( + &keys, &aggs, &having, parent_schema, + ) + { + Ok(b) => { + let kinds: Vec<&'static str> = b.aggs.iter() + .map(canonical_intent_kind_str) + .collect(); + format!( + "General Aggregate at Db (exact); canonical \ + intents: {:?}, group_by_cols: {}", + kinds, b.by.len() + ) + } + Err(_e) => "General Aggregate at Db (exact)".into(), + }; + let child = self.alloc_node(*input, budget, parent_schema); PlanNode { expr: QueryExpr::Aggregate { @@ -275,7 +301,7 @@ impl SketchAllocator { ..Default::default() }, annotation: NodeAnnotation { - rationale: "General Aggregate at Db (exact)".into(), + rationale: bridge_annotation, ..Default::default() }, children: vec![child], @@ -636,6 +662,36 @@ fn estimated_sketch_memory(op: &AggIntent) -> f64 { super::sketch_catalog::estimated_sketch_memory_bytes(op) as f64 } +/// Map a canonical [`AggIntent`] to a short stable kind string for +/// annotation rationale text. Step γ1: used by the legacy +/// `QueryExpr::Aggregate` arm to enrich the rationale with the canonical +/// intent kinds reachable via `bridge_aggregate_to_canonical`. +fn canonical_intent_kind_str(intent: &AggIntent) -> &'static str { + match intent { + AggIntent::Count { .. } => "count", + AggIntent::Sum => "sum", + AggIntent::Min => "min", + AggIntent::Max => "max", + AggIntent::Avg => "avg", + AggIntent::Quantile { .. } => "quantile", + AggIntent::TopK { .. } => "topk", + AggIntent::Cardinality { .. } => "cardinality", + AggIntent::Frequency { .. } => "frequency", + AggIntent::Rate { .. } => "rate", + AggIntent::Increase { .. } => "increase", + AggIntent::Absent => "absent", + AggIntent::Present => "present", + AggIntent::Delta { .. } => "delta", + AggIntent::Deriv { .. } => "deriv", + AggIntent::PredictLinear { .. } => "predict_linear", + AggIntent::HoltWinters { .. } => "holt_winters", + AggIntent::Idelta { .. } => "idelta", + AggIntent::Irate { .. } => "irate", + AggIntent::Resets { .. } => "resets", + AggIntent::Changes { .. } => "changes", + } +} + // sketch_type_for_op delegated to algebra::directory::sketch_type_and_params. // ── Tests ─────────────────────────────────────────────────────────────────────