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 ─────────────────────────────────────────────────────────────────────