diff --git a/controller/src/intent_algebra/mod.rs b/controller/src/intent_algebra/mod.rs index 5730fe69..615f521b 100644 --- a/controller/src/intent_algebra/mod.rs +++ b/controller/src/intent_algebra/mod.rs @@ -147,3 +147,14 @@ pub use windowed_agg_bridge::{ bridge_windowed_agg_to_canonical, output_schema_for_windowed_agg, BridgeError as WindowedAggBridgeError, BridgedWindowedAgg, }; + +// Step γ2 bridge: one-way `legacy_expr::QueryExpr::SketchAgg` → +// canonical-shape `BridgedAggregate` helper. Companion to the γ1 +// Aggregate bridge — SketchAgg's single-intent / single-column shape +// maps onto the same `BridgedAggregate` carrier (with `having: None`) +// so consumers can match on `by` / `aggs` regardless of which legacy +// variant the data came from. +pub mod sketch_agg_bridge; +pub use sketch_agg_bridge::{ + bridge_sketch_agg_to_canonical, output_schema_for_sketch_agg, +}; diff --git a/controller/src/intent_algebra/sketch_agg_bridge.rs b/controller/src/intent_algebra/sketch_agg_bridge.rs new file mode 100644 index 00000000..cb1667ce --- /dev/null +++ b/controller/src/intent_algebra/sketch_agg_bridge.rs @@ -0,0 +1,327 @@ +//! Step γ2 bridge: legacy `SketchAgg` node → canonical +//! `intent_algebra::query_expr::QueryExpr::Aggregate` shape data. +//! +//! Companion to the Step γ1 [`crate::intent_algebra::aggregate_bridge`] +//! module. Same approach **(c)** per the migration spec: the legacy +//! [`crate::intent_algebra::legacy_expr::QueryExpr::SketchAgg`] variant +//! remains the L2 emit shape (parsers, optimizer rules, allocator, stage +//! splitter still match on `SketchAgg` unchanged); this module adds a +//! one-way builder helper that converts the legacy single-intent / +//! single-column shape into the canonical-shape data — +//! `by: Vec`, `aggs: Vec` — that downstream +//! canonical-shape consumers (sketch_algebra `Bind*` rules, cost model, +//! …) already match on. +//! +//! ## Structural note: SketchAgg vs Aggregate +//! +//! `Aggregate { keys: Vec, aggs: Vec, having, input }` +//! and `SketchAgg { op: AggIntent, col: ColumnRef, input }` carry the +//! same semantic shape on the canonical side: a tuple of group-by +//! columns + a list of intents. The differences on the legacy side: +//! +//! | Aggregate | SketchAgg | +//! |---|---| +//! | `keys: Vec` (multi-col) | `col: ColumnRef` (single-col) | +//! | `aggs: Vec` w/ AggFunc | `op: AggIntent` already canonical | +//! | `having: Option` | no HAVING — always `None` | +//! +//! Because `SketchAgg.op` is already canonical [`AggIntent`] (Step α, +//! PR #138 — the variant lives in `legacy_expr` only because its +//! single-column / single-intent shape doesn't match the multi-column / +//! multi-intent canonical `Aggregate`; the per-field types are already +//! migrated), the bridge here is much thinner than γ1's: +//! no `AggFunc → AggIntent` translation, no fan-out (Step α F1 happens +//! at parse time before SketchAgg construction), no HAVING translation. +//! +//! ## Wildcard handling +//! +//! `SketchAgg { col: ColumnRef::Wildcard, .. }` is the COUNT(*) shape — +//! the sketch operates over every row, not on a specific column value. +//! In canonical-shape terms this is a global aggregate with an empty +//! `by`. The bridge surfaces this by emitting `by: vec![]` for the +//! Wildcard arm. `SampleValue` and `Named` arms resolve through +//! [`resolve_column_ref`] like any other positional column reference. +//! +//! ## What the bridge produces +//! +//! ```text +//! legacy::QueryExpr::SketchAgg { op, col, input } +//! │ +//! │ with parent_schema: &Schema ← Step β plumbing +//! ▼ +//! BridgedAggregate { +//! by: Vec, // [resolve(col)] or [] for Wildcard +//! aggs: vec![op.clone()], // single-intent +//! having: None, // SketchAgg has no HAVING +//! // child intentionally NOT carried — consumers walk legacy `input` directly. +//! } +//! ``` + +use crate::intent_algebra::agg_intent::AggIntent; +use crate::intent_algebra::aggregate_bridge::{BridgeError, BridgedAggregate}; +use crate::intent_algebra::column_resolution::resolve_column_ref; +use crate::intent_algebra::legacy_expr::ColumnRef; +use crate::intent_algebra::schema::{ColumnId, Schema}; + +/// Translate the canonical-shape fields of a legacy +/// `legacy_expr::QueryExpr::SketchAgg { op, col, input }` against the +/// inherited schema. +/// +/// Returns a [`BridgedAggregate`] — the same canonical-shape carrier the +/// γ1 Aggregate bridge produces — so downstream consumers can match on +/// `by` / `aggs` regardless of which legacy variant the data came from. +/// The legacy `input` subtree is left to the caller's own recursion. +/// +/// ## Schema flow +/// +/// `parent_schema` is the schema in scope at the legacy `SketchAgg` node +/// — its INPUT schema. The bridge resolves `col` against it. To get the +/// `SketchAgg`'s OUTPUT schema (which is what consumers should pass +/// downward when threading further down, or upward when the `SketchAgg` +/// is itself a child of another node), call +/// [`output_schema_for_sketch_agg`] on the `BridgedAggregate.by` / +/// `.aggs` together with `parent_schema`. +/// +/// ## Resolution rules +/// +/// * `ColumnRef::Named(name)` → `by = vec![schema.column_id(name)?]`. +/// * `ColumnRef::SampleValue` → `by = vec![schema.column_id("value")?]`. +/// * `ColumnRef::Wildcard` → `by = vec![]` (global / row-count +/// semantics). NOT an error here, unlike the bare +/// [`resolve_column_ref`] which surfaces +/// [`crate::intent_algebra::column_resolution::ResolveError::WildcardNotPositional`]. +/// +/// ## Errors +/// +/// Returns [`BridgeError::Key`] (wrapping `ResolveError::NotFound` or +/// `ResolveError::NoSampleValue`) when the named / sample-value column +/// is absent from the inherited schema. The conservative fallback: +/// callers keep the legacy SketchAgg at this site and log the deferral. +/// +/// SketchAgg has no `having` clause, so [`BridgeError::HavingDeferred`] +/// is unreachable. `op` is already canonical [`AggIntent`] (Step α), so +/// [`BridgeError::NoCanonicalIntent`] is also unreachable. +pub fn bridge_sketch_agg_to_canonical( + op: &AggIntent, + col: &ColumnRef, + parent_schema: &Schema, +) -> Result { + // Resolve `col` against the inherited schema, special-casing + // Wildcard → empty `by` (global / row-count semantics). + let by: Vec = match col { + ColumnRef::Wildcard => Vec::new(), + _ => vec![resolve_column_ref(col, parent_schema)?], + }; + + Ok(BridgedAggregate { + by, + aggs: vec![op.clone()], + having: None, + }) +} + +/// Output schema produced by `QueryExpr::SketchAgg { op, col, input }` +/// when its `input` carries `input_schema` as its output schema. +/// +/// Thin wrapper over +/// [`crate::intent_algebra::column_resolution::output_schema_for_aggregate`] +/// — SketchAgg and Aggregate produce the same shape (group-by columns +/// preserved positionally + one column per intent; time axis stripped; +/// unique_keys = `[by]` when `by` is non-empty, otherwise empty). Kept +/// as a sibling helper so callers descending into a still-legacy +/// `SketchAgg.input` can derive the right schema for the child without +/// first having to translate the whole subtree to canonical, mirroring +/// the γ1 [`output_schema_for_aggregate`] pattern. +/// +/// `by` ids are silently clamped to in-range — out-of-range ids are +/// dropped from the output. (Callers that need the strict check should +/// resolve `by` ids upstream via [`bridge_sketch_agg_to_canonical`] +/// which DOES surface `NotFound`.) +/// +/// # Example +/// +/// ```ignore +/// use controller::intent_algebra::{ +/// column_resolution::infer_source_schema, +/// sketch_agg_bridge::output_schema_for_sketch_agg, +/// agg_intent::AggIntent, +/// }; +/// let input = infer_source_schema("http_requests_total"); +/// // SketchAgg over Wildcard (COUNT(*)-style) → global aggregate, by = []. +/// let aggs = vec![AggIntent::Sum]; +/// let out = output_schema_for_sketch_agg(&input, &[], &aggs); +/// assert_eq!(out.columns.len(), 1); // sum +/// assert!(out.time_index.is_none()); // time axis stripped +/// ``` +pub fn output_schema_for_sketch_agg( + input: &Schema, + by: &[ColumnId], + aggs: &[AggIntent], +) -> Schema { + // SketchAgg's output shape is identical to Aggregate's — delegate. + crate::intent_algebra::column_resolution::output_schema_for_aggregate(input, by, aggs) +} + +// ── Tests ───────────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + use crate::intent_algebra::column_resolution::{infer_source_schema, ResolveError}; + use crate::intent_algebra::schema::{Column, DataType}; + use crate::types_v2::AccuracyTarget; + + #[test] + fn bridge_single_intent_sample_value() { + // SketchAgg { op: Sum, col: SampleValue, .. } against a default + // time-series schema → by = [1] (the `value` column), aggs = [Sum]. + let s = infer_source_schema("m"); + let b = bridge_sketch_agg_to_canonical( + &AggIntent::Sum, + &ColumnRef::SampleValue, + &s, + ) + .unwrap(); + assert_eq!(b.by, vec![1usize]); + assert_eq!(b.aggs.len(), 1); + assert!(matches!(b.aggs[0], AggIntent::Sum)); + assert!(b.having.is_none()); + } + + #[test] + fn bridge_wildcard_emits_empty_by() { + // Wildcard → global aggregate (by = []), single intent preserved. + let s = infer_source_schema("m"); + let b = bridge_sketch_agg_to_canonical( + &AggIntent::Count { accuracy: AccuracyTarget::Exact }, + &ColumnRef::Wildcard, + &s, + ) + .unwrap(); + assert!(b.by.is_empty()); + assert_eq!(b.aggs.len(), 1); + assert!(matches!(b.aggs[0], AggIntent::Count { .. })); + assert!(b.having.is_none()); + } + + #[test] + fn bridge_named_column_resolves_positionally() { + // SketchAgg { op: Max, col: Named("host"), .. } with host column + // at position 2 → by = [2]. + let mut s = infer_source_schema("m"); + s.columns.push(Column { + name: "host".into(), + dtype: DataType::Utf8, + nullable: false, + }); + let b = bridge_sketch_agg_to_canonical( + &AggIntent::Max, + &ColumnRef::Named("host".into()), + &s, + ) + .unwrap(); + assert_eq!(b.by, vec![2usize]); + assert_eq!(b.aggs.len(), 1); + assert!(matches!(b.aggs[0], AggIntent::Max)); + } + + #[test] + fn bridge_unresolvable_named_column_surfaces_key_error() { + // Named column absent from schema → BridgeError::Key(NotFound). + let s = infer_source_schema("m"); + let err = bridge_sketch_agg_to_canonical( + &AggIntent::Sum, + &ColumnRef::Named("missing".into()), + &s, + ) + .unwrap_err(); + assert!(matches!(err, BridgeError::Key(ResolveError::NotFound { .. }))); + } + + #[test] + fn bridge_sample_value_missing_surfaces_key_error() { + // SampleValue against a schema with no `value` column → + // BridgeError::Key(NoSampleValue). + let s = Schema { + columns: vec![Column { + name: "only".into(), + dtype: DataType::Utf8, + nullable: false, + }], + time_index: None, + unique_keys: Vec::new(), + }; + let err = bridge_sketch_agg_to_canonical( + &AggIntent::Sum, + &ColumnRef::SampleValue, + &s, + ) + .unwrap_err(); + assert!(matches!(err, BridgeError::Key(ResolveError::NoSampleValue { .. }))); + } + + #[test] + fn bridge_preserves_canonical_intent_unchanged() { + // The bridge passes `op` through verbatim — no fan-out (Step α + // F1 happens at parse time before SketchAgg construction) and no + // AggFunc translation (op is already canonical AggIntent). + let s = infer_source_schema("m"); + let intent = AggIntent::Quantile { + q: 0.95, + accuracy: AccuracyTarget::Exact, + }; + let b = bridge_sketch_agg_to_canonical(&intent, &ColumnRef::SampleValue, &s).unwrap(); + assert_eq!(b.aggs.len(), 1); + match &b.aggs[0] { + AggIntent::Quantile { q, .. } => assert!((*q - 0.95).abs() < 1e-9), + other => panic!("expected Quantile, got {other:?}"), + } + } + + #[test] + fn bridge_output_schema_matches_canonical_aggregate_shape() { + // Verify the bridged-data → output_schema_for_sketch_agg path + // produces the same shape the canonical + // QueryExpr::Aggregate::output_schema_in would for an equivalent + // single-intent / single-column input. The shared helper is + // output_schema_for_aggregate, so this test pins the equivalence. + let mut s = infer_source_schema("m"); + s.columns.push(Column { + name: "host".into(), + dtype: DataType::Utf8, + nullable: false, + }); + let b = bridge_sketch_agg_to_canonical( + &AggIntent::Sum, + &ColumnRef::Named("host".into()), + &s, + ) + .unwrap(); + let out = output_schema_for_sketch_agg(&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"); + // Time axis stripped — SketchAgg produces one row per group. + assert!(out.time_index.is_none()); + // unique_keys = [[0]] (the by tuple is unique by construction). + assert_eq!(out.unique_keys, vec![vec![0]]); + } + + #[test] + fn bridge_output_schema_wildcard_global_has_no_unique_keys() { + // Wildcard → by = [] → global aggregate → empty unique_keys. + let s = infer_source_schema("m"); + let b = bridge_sketch_agg_to_canonical( + &AggIntent::Count { accuracy: AccuracyTarget::Exact }, + &ColumnRef::Wildcard, + &s, + ) + .unwrap(); + let out = output_schema_for_sketch_agg(&s, &b.by, &b.aggs); + assert_eq!(out.columns.len(), 1); + assert_eq!(out.columns[0].name, "count"); + assert!(out.unique_keys.is_empty()); + assert!(out.time_index.is_none()); + } +} diff --git a/controller/src/physical/allocator.rs b/controller/src/physical/allocator.rs index d2659f6a..6818763c 100644 --- a/controller/src/physical/allocator.rs +++ b/controller/src/physical/allocator.rs @@ -598,11 +598,37 @@ impl SketchAllocator { col: crate::intent_algebra::legacy_expr::ColumnRef, child: PlanNode, budget: &mut BudgetState, - // Step β: schema in scope at this SketchAgg node. Unused today — - // Step γ wires sketch-placement rules that consult column types - // (e.g. KLL vs DDSketch for `value: Float64`). - _parent_schema: &Schema, + // Step β: schema in scope at this SketchAgg node. Step γ2 wires + // this schema through the SketchAgg bridge so the annotation + // rationale can carry canonical-shape info; future Step γ wiring + // (sketch-placement rules that consult column types, e.g. KLL vs + // DDSketch for `value: Float64`) will read it directly. + parent_schema: &Schema, ) -> PlanNode { + // Step γ2 demonstration: bridge the legacy SketchAgg to canonical- + // shape data and capture a suffix describing the canonical intent + // kind + group-by column count. The legacy emit shape is unchanged + // — `expr` still carries the legacy SketchAgg variant (approach (c) + // per the migration spec). The bridge fails gracefully when `col` + // is a Named/SampleValue that doesn't resolve against the inherited + // schema; the legacy annotation text stands alone in that case. + let canonical_suffix: String = match + crate::intent_algebra::bridge_sketch_agg_to_canonical( + &op, &col, parent_schema, + ) + { + Ok(b) => { + let kinds: Vec<&'static str> = b.aggs.iter() + .map(canonical_intent_kind_str) + .collect(); + format!( + " [canonical: intents={:?}, group_by_cols={}]", + kinds, b.by.len() + ) + } + Err(_e) => String::new(), + }; + // Exact non-mergeable (Avg) → always Db. if matches!(&op, AggIntent::Avg) { return PlanNode { @@ -618,7 +644,9 @@ impl SketchAllocator { ..Default::default() }, annotation: NodeAnnotation { - rationale: "Avg is not mergeable — must run at Db".into(), + rationale: format!( + "Avg is not mergeable — must run at Db{canonical_suffix}" + ), ..Default::default() }, children: vec![child], @@ -640,7 +668,9 @@ impl SketchAllocator { ..Default::default() }, annotation: NodeAnnotation { - rationale: "Exact(Sum/Count/Min/Max) merged at Backend".into(), + rationale: format!( + "Exact(Sum/Count/Min/Max) merged at Backend{canonical_suffix}" + ), ..Default::default() }, children: vec![child], @@ -671,7 +701,9 @@ impl SketchAllocator { annotation: NodeAnnotation { sketch_type: Some(sketch_type), sketch_params: Some(params), - rationale: "Sketch at Agent (within budget)".into(), + rationale: format!( + "Sketch at Agent (within budget){canonical_suffix}" + ), ..Default::default() }, children: vec![child], @@ -697,7 +729,9 @@ impl SketchAllocator { annotation: NodeAnnotation { sketch_type: Some(sketch_type), sketch_params: Some(params), - rationale: "Sketch demoted to Backend (Agent budget exceeded)".into(), + rationale: format!( + "Sketch demoted to Backend (Agent budget exceeded){canonical_suffix}" + ), budget_demotion: true, ..Default::default() }, @@ -723,7 +757,9 @@ impl SketchAllocator { annotation: NodeAnnotation { sketch_type: Some(sketch_type), sketch_params: Some(params), - rationale: "Sketch demoted to Precompute (Agent+Backend budgets exceeded)".into(), + rationale: format!( + "Sketch demoted to Precompute (Agent+Backend budgets exceeded){canonical_suffix}" + ), budget_demotion: true, ..Default::default() },