diff --git a/crates/e2e/tests/l4_binding.rs b/crates/e2e/tests/l4_binding.rs index 3f04d260..33b4515f 100644 --- a/crates/e2e/tests/l4_binding.rs +++ b/crates/e2e/tests/l4_binding.rs @@ -8,7 +8,7 @@ use asap_frontend_promql::lower_promql; use asap_ir::intent_algebra::expr_ir::ColumnRef; -use asap_ir::intent_algebra::query_expr::QueryExpr; +use asap_ir::intent_algebra::query_expr::{QueryExpr, Reduction}; use asap_ir::intent_algebra::schema::DataType; use asap_ir::types::AccuracyTarget; use asap_plan::implement_tree; @@ -59,13 +59,17 @@ fn promql_quantile_of_rate_binds_kll_over_rate_accumulator() { "the Sketch(…) type must not propagate past the estimate" ); - // The quantile: KLL committed, k=200 sized from ε=0.01. + // The quantile: KLL committed, k=200 sized from ε=0.01. `quantile(...)` + // is an aggregation operator with no `by(...)`: a genuine full + // reduction, one output row — not to be confused with the inner rate's + // per-entity grouping below, even though both once collapsed to the + // same empty `by: []` (issue #163). let SummaryExpr::SummaryAgg { child, sketch, params, col, - by, + reduction, } = &sketch_input.expr else { panic!("expected SummaryAgg, got {:?}", sketch_input.expr); @@ -73,19 +77,24 @@ fn promql_quantile_of_rate_binds_kll_over_rate_accumulator() { assert_eq!(sketch, &SummaryKind::Kll); assert_eq!(params, &SummaryParams::Kll { k: 200 }); assert_eq!(col, &ColumnRef::SampleValue); - assert!(by.is_empty(), "global quantile — no group keys"); + assert_eq!( + reduction, + &Reduction::by(vec![]), + "global quantile — no group keys, full reduction" + ); assert_eq!( dtype(&sketch_input.schema, "quantile_0_99"), &L4DataType::Sketch(SummaryKind::Kll, SummaryParams::Kll { k: 200 }) ); // The rate: exact counter-reset-aware accumulator, per-series (labels - // and time axis preserved), no estimate wrapper. + // and time axis preserved), no estimate wrapper. `rate(...)` has no + // grouping concept at all — every entity stays its own summary. let SummaryExpr::SummaryAgg { child: leaf, sketch, params, - by, + reduction, .. } = &child.expr else { @@ -93,7 +102,7 @@ fn promql_quantile_of_rate_binds_kll_over_rate_accumulator() { }; assert_eq!(sketch, &SummaryKind::Rate); assert_eq!(params, &SummaryParams::Rate); - assert!(by.is_empty()); + assert_eq!(reduction, &Reduction::PerEntity); assert_eq!( dtype(&child.schema, "value"), &L4DataType::Sketch(SummaryKind::Rate, SummaryParams::Rate) @@ -131,14 +140,21 @@ fn promql_exact_workload_binds_accumulators_not_sketches() { .expect("lowering failed"); let root = implement_tree(&l3).expect("binding failed"); let SummaryExpr::SummaryAgg { - sketch, params, by, .. + sketch, + params, + reduction, + .. } = &root.expr else { panic!("expected SummaryAgg, got {:?}", root.expr); }; assert_eq!(sketch, &SummaryKind::Sum); assert_eq!(params, &SummaryParams::Sum); - assert_eq!(by, &vec![2], "job is col 2 in [ts, value, job]"); + assert_eq!( + reduction, + &Reduction::by(vec![2]), + "job is col 2 in [ts, value, job]" + ); assert_eq!( dtype(&root.schema, "job"), &L4DataType::Primitive(DataType::Utf8), diff --git a/crates/plan/src/bind.rs b/crates/plan/src/bind.rs index ab9b1ec2..6e2cb0a1 100644 --- a/crates/plan/src/bind.rs +++ b/crates/plan/src/bind.rs @@ -129,9 +129,7 @@ fn bind_summary_agg( let child_schema = child.output_schema_in(scope)?; // The single canonical L3 derivation (per-series vs cross-series, name // overrides) already computes the row shape; L4 only retypes the summary - // state column. `reduction` (read directly, not re-derived — issue #165) - // settles both the schema shape and, downstream of this PR, the - // `by`-emptiness ambiguity a bare `Vec` alone can't (issue #163). + // state column. let per_series = matches!(reduction, Reduction::PerEntity); let by: Vec = reduction .group_keys() @@ -148,13 +146,18 @@ fn bind_summary_agg( field.dtype = L4DataType::Sketch(kind.clone(), params.clone()); } + // `reduction` is carried onto `SummaryAgg` verbatim — not flattened to a + // bare `Vec` — so `SummaryExecutor::find_candidates` can tell + // a genuine empty-`by` reduction apart from a per-entity shape with no + // grouping concept at all (issue #163). `bind_summary_agg` is the single + // place that decides this; nothing downstream re-derives it. let agg = Rc::new(L4Node { expr: SummaryExpr::SummaryAgg { child: implement_tree_in_with(child, scope, cost_model)?, sketch: kind, params, col, - by, + reduction: reduction.clone(), }, schema: state_schema, }); @@ -343,7 +346,7 @@ mod tests { sketch, params, col, - by, + reduction, } = &sketch_input.expr else { panic!("expected SummaryAgg, got {:?}", sketch_input.expr); @@ -351,7 +354,7 @@ mod tests { assert_eq!(sketch, &SummaryKind::Kll); assert_eq!(params, &SummaryParams::Kll { k: 200 }); assert_eq!(col, &ColumnRef::SampleValue); - assert_eq!(by, &vec![2]); + assert_eq!(reduction, &Reduction::by(vec![2])); // SummaryAgg edge: the state column carries the committed (kind, params). assert_eq!( field(&sketch_input.schema, "quantile_0_99").dtype, @@ -554,6 +557,52 @@ mod tests { assert_eq!(root.schema.time_index, Some(0)); } + /// Issue #163, case 1: a bare per-series range function (e.g. + /// `quantile_over_time(...)`) binds to `SummaryAgg { reduction: + /// PerEntity, .. }` — proving the L3 `Reduction` this crate already + /// computes (issue #165) is carried onto the L4 node verbatim, not + /// flattened back into an ambiguous bare `Vec`. + #[test] + fn bare_per_series_aggregate_binds_summary_agg_with_per_entity_reduction() { + let q = agg_per_entity( + default_quantile(0.99), + QueryExpr::TimeRange { + range: Duration::from_secs(10), + child: Box::new(metric_scan(&["job"])), + }, + ); + let root = implement_tree(&q).unwrap(); + let SummaryExpr::SummaryEstimate { sketch_input, .. } = &root.expr else { + panic!("expected estimate root, got {:?}", root.expr); + }; + let SummaryExpr::SummaryAgg { reduction, .. } = &sketch_input.expr else { + panic!("expected SummaryAgg, got {:?}", sketch_input.expr); + }; + assert_eq!(reduction, &Reduction::PerEntity); + } + + /// Issue #163, case 2: an aggregation operator explicitly invoked with + /// no `by(...)` (e.g. `count(hll_metric)`) binds to `SummaryAgg { + /// reduction: Reduce(vec![]), .. }` — byte-identical `by: []` to the + /// previous test at the old `Vec` shape; `reduction` is what + /// tells them apart now. + #[test] + fn explicit_empty_by_aggregate_binds_summary_agg_with_reduce_reduction() { + let intent = AggIntent::Cardinality { + col: None, + accuracy: AccuracyTarget::Epsilon(0.01), + }; + let q = agg(vec![], intent, metric_scan(&["job"])); + let root = implement_tree(&q).unwrap(); + let SummaryExpr::SummaryEstimate { sketch_input, .. } = &root.expr else { + panic!("expected estimate root, got {:?}", root.expr); + }; + let SummaryExpr::SummaryAgg { reduction, .. } = &sketch_input.expr else { + panic!("expected SummaryAgg, got {:?}", sketch_input.expr); + }; + assert_eq!(reduction, &Reduction::by(vec![])); + } + #[test] fn nested_aggregates_bind_per_node() { // quantile(0.9, sum by (job) (m)) — the boundary fires per node over diff --git a/crates/sketch/src/exec.rs b/crates/sketch/src/exec.rs index 0872d5f7..16318fce 100644 --- a/crates/sketch/src/exec.rs +++ b/crates/sketch/src/exec.rs @@ -6,7 +6,7 @@ use std::collections::BTreeMap; -use asap_ir::intent_algebra::{ColumnId, ColumnRef, QueryExpr}; +use asap_ir::intent_algebra::{ColumnRef, QueryExpr, Reduction}; use crate::expr::{L4Node, SummaryExpr}; use crate::sketch::{SketchQuery, SummaryKind, SummaryParams}; @@ -23,28 +23,42 @@ pub trait SummaryExecutor { type Value; /// Deployment error type. type Error; - /// Opaque per-group identity for a `SummaryAgg`'s `by` columns — e.g. - /// a label-value map for a PromQL/time-series deployment. `Ord` so + /// Opaque per-group identity for a `SummaryAgg`'s grouping — e.g. a + /// label-value map for a PromQL/time-series deployment. `Ord` so /// `execute` can fold same-group handles/states deterministically; - /// `Default` for the ungrouped case (`by` empty, or a `Logical` leaf, - /// which has no grouping concept at all) — every outcome is always a - /// per-group list, and the ungrouped case is simply a list of one - /// entry under `GroupKey::default()`. + /// `Default` for the single-group case (`Reduction::Reduce` with an + /// empty `by`, or a `Logical` leaf, which has no grouping concept at + /// all) — every outcome is always a per-group list, and the + /// single-group case is simply a list of one entry under + /// `GroupKey::default()`. type GroupKey: Clone + Ord + Default; /// Resolve a `SummaryAgg` leaf to matching materialized-instance - /// handles, each tagged with the group (per `by`) it belongs to. Must - /// only return handles whose `(SummaryKind, SummaryParams)` is - /// exactly `(sketch, params)`. When `by` is empty there is exactly - /// one group — every returned handle should share one - /// deployment-chosen `GroupKey` (e.g. `GroupKey::default()`). + /// handles, each tagged with the group it belongs to. Must only + /// return handles whose `(SummaryKind, SummaryParams)` is exactly + /// `(sketch, params)`. + /// + /// `reduction` (the same [`Reduction`] the L3 `Aggregate` node this + /// was bound from carried) tells you which of two shapes to produce — + /// the two are **not** interchangeable (issue #163): + /// - `Reduction::Reduce(by)`: a genuine cross-series reduction. Tag + /// every returned handle with the `GroupKey` for its `by` values; + /// when `by` is empty there is exactly one group — every handle + /// should share one deployment-chosen `GroupKey` (e.g. + /// `GroupKey::default()`), and they will all be merged together. + /// - `Reduction::PerEntity`: there is no grouping concept for this + /// shape at all. Tag every returned handle with its own, distinct + /// `GroupKey` (its full entity/series identity) — handles must + /// never share a `GroupKey` here, since `execute` merges same-group + /// handles together and merging unrelated entities would silently + /// produce a wrong answer. #[allow(clippy::type_complexity)] fn find_candidates( &self, sketch: &SummaryKind, params: &SummaryParams, col: &ColumnRef, - by: &[ColumnId], + reduction: &Reduction, child: &L4Node, ) -> Result, Self::Error>; @@ -119,9 +133,9 @@ pub fn execute( sketch, params, col, - by, + reduction, } => { - let tagged = exec.find_candidates(sketch, params, col, by, child)?; + let tagged = exec.find_candidates(sketch, params, col, reduction, child)?; if tagged.is_empty() { return Err(ExecError::NoCandidates); } @@ -270,14 +284,27 @@ mod tests { }) } + /// Builds a `SummaryAgg` with `Reduction::Reduce(vec![])` (an ordinary, + /// ungrouped cross-series reduction) — the shape every pre-#163 test in + /// this module exercises. Use [`agg_node_with_reduction`] to exercise + /// `Reduction` itself. fn agg_node(sketch: SummaryKind, params: SummaryParams, child: Rc) -> Rc { + agg_node_with_reduction(sketch, params, child, Reduction::by(vec![])) + } + + fn agg_node_with_reduction( + sketch: SummaryKind, + params: SummaryParams, + child: Rc, + reduction: Reduction, + ) -> Rc { Rc::new(L4Node { expr: SummaryExpr::SummaryAgg { child, sketch, params, col: ColumnRef::SampleValue, - by: vec![], + reduction, }, schema: lift(vec!["value"]), }) @@ -306,11 +333,13 @@ mod tests { /// default), and `merge_states` is `sum`, so tests can assert on /// concrete numbers without any real sketch-math dependency. /// - /// `find_candidates` intentionally ignores `sketch`/`params`/`by`/ - /// `child` and returns every registered handle tagged with its own - /// registered group — real deployments filter on those; these tests - /// only exercise `execute`'s own tree-walking and grouping/merge - /// logic, not a real candidate search. + /// `find_candidates` intentionally ignores `sketch`/`params`/ + /// `reduction`/`child` and returns every registered handle tagged with + /// its own registered group — real deployments filter on those; these + /// tests only exercise `execute`'s own tree-walking and grouping/merge + /// logic, not a real candidate search. See + /// `reduction_is_passed_through_to_find_candidates_unmodified` below + /// for a test that *does* inspect `reduction`. struct MockExecutor { /// handle -> (group, value). values: RefCell>, @@ -355,7 +384,7 @@ mod tests { _sketch: &SummaryKind, _params: &SummaryParams, _col: &ColumnRef, - _by: &[ColumnId], + _reduction: &Reduction, _child: &L4Node, ) -> Result, Self::Error> { Ok(self @@ -610,4 +639,65 @@ mod tests { vec![("us-east".to_string(), 2.0), ("us-west".to_string(), 20.0)] ); } + + /// A wiring test for issue #163: `execute` must pass a `SummaryAgg`'s + /// `reduction` through to `find_candidates` completely unmodified — not + /// re-derive it, not silently drop it in favor of a bare `by` list. + /// `MockExecutor` above ignores `reduction` entirely (it isn't testing + /// this), so this test uses its own executor that records exactly what + /// it received. + struct ReductionSpyExecutor { + seen: RefCell>, + } + + impl SummaryExecutor for ReductionSpyExecutor { + type Handle = u64; + type State = f64; + type Value = f64; + type Error = String; + type GroupKey = u32; + + fn find_candidates( + &self, + _sketch: &SummaryKind, + _params: &SummaryParams, + _col: &ColumnRef, + reduction: &Reduction, + _child: &L4Node, + ) -> Result, Self::Error> { + self.seen.borrow_mut().push(reduction.clone()); + Ok(vec![(0, 0)]) + } + + fn fetch_state(&self, _handle: &Self::Handle) -> Result { + Ok(1.0) + } + + fn merge_states(&self, states: Vec) -> Result { + Ok(states.into_iter().sum()) + } + + fn readout( + &self, + state: &Self::State, + _query: &SketchQuery, + ) -> Result { + Ok(*state) + } + + fn logical(&self, _expr: &QueryExpr) -> Result { + Ok(-1.0) + } + } + + #[test] + fn reduction_is_passed_through_to_find_candidates_unmodified() { + let exec = ReductionSpyExecutor { + seen: RefCell::new(vec![]), + }; + let (kind, params) = sum(); + let tree = agg_node_with_reduction(kind, params, logical_node(), Reduction::PerEntity); + execute(&tree, &exec).unwrap(); + assert_eq!(exec.seen.borrow().as_slice(), &[Reduction::PerEntity]); + } } diff --git a/crates/sketch/src/expr.rs b/crates/sketch/src/expr.rs index 082e984c..596f0618 100644 --- a/crates/sketch/src/expr.rs +++ b/crates/sketch/src/expr.rs @@ -2,7 +2,7 @@ use std::rc::Rc; use super::schema::L4Schema; use super::sketch::{SketchQuery, SummaryKind, SummaryParams}; -use asap_ir::intent_algebra::{ColumnId, ColumnRef, QueryExpr}; +use asap_ir::intent_algebra::{ColumnRef, QueryExpr, Reduction}; // ── L4 DAG node ─────────────────────────────────────────────────────────────── @@ -36,16 +36,23 @@ pub enum SummaryExpr { /// Sketch aggregation. L4 chose `sketch` + `params` from the catalog /// for `AggIntent` under `DeploymentConstraints`. - /// Output schema: `by` columns (verbatim) + one `Sketch(sketch, params)` - /// field carrying partial sketch state per group. + /// Output schema: grouping columns (verbatim) + one `Sketch(sketch, + /// params)` field carrying partial sketch state per group. SummaryAgg { child: Rc, sketch: SummaryKind, params: SummaryParams, /// The column being summarised (fed into the sketch). col: ColumnRef, - /// GROUP BY keys (positional) carried through to the output schema. - by: Vec, + /// How this aggregation's output rows relate to `child`'s — the + /// same [`Reduction`] the L3 `Aggregate` node it was bound from + /// carried (issue #165), reused verbatim rather than flattened to + /// a bare `Vec`. `Reduction::Reduce(by)` with an empty + /// `by` is a genuine full reduction (merge every candidate into + /// one group); `Reduction::PerEntity` has no grouping concept at + /// all (never merge across entities) — the two collapsed to the + /// same ambiguous `by: []` before this field existed (issue #163). + reduction: Reduction, }, /// Sketch-aware join (KMV / theta for join-cardinality; join-sample for diff --git a/docs/l4node-execution-model.md b/docs/l4node-execution-model.md index 94ac88f7..22c3f979 100644 --- a/docs/l4node-execution-model.md +++ b/docs/l4node-execution-model.md @@ -32,8 +32,8 @@ pub enum SummaryExpr { child: Rc, sketch: SummaryKind, params: SummaryParams, - col: ColumnRef, // the column being summarised - by: Vec, // GROUP BY keys + col: ColumnRef, // the column being summarised + reduction: Reduction, // GROUP BY keys, or "no grouping concept" (#163) }, /// Sketch-aware join (KMV/theta for cardinality, join-sample for @@ -64,9 +64,10 @@ value-bearing field carries this dtype is a "sketch-state schema." The compatible only if both match exactly, which is what lets `SummarySubtract`/ `SummaryDelete`/`SummaryMerge` reject a family/param mismatch as a plan-time type error rather than a runtime one. Per-node schema rule -summary: `SummaryAgg` outputs `by` verbatim plus one `Sketch(sketch, -params)` field; `SummaryEstimate` outputs a plain row-shaped schema (the -sketch field doesn't survive it); `SummarySubtract`/`SummaryMerge` require +summary: `SummaryAgg` outputs its grouping columns verbatim plus one +`Sketch(sketch, params)` field; `SummaryEstimate` outputs a plain +row-shaped schema (the sketch field doesn't survive it); +`SummarySubtract`/`SummaryMerge` require every input's `Sketch(s, p)` to already agree, and the catalog capability flag (`subtractable`/`deletable`/`mergeable`) gates whether the node can fire at all. @@ -111,7 +112,7 @@ pub trait SummaryExecutor { type GroupKey: Clone + Ord + Default; // e.g. a label-value map fn find_candidates(&self, sketch: &SummaryKind, params: &SummaryParams, - col: &ColumnRef, by: &[ColumnId], child: &L4Node) + col: &ColumnRef, reduction: &Reduction, child: &L4Node) -> Result, Self::Error>; fn fetch_state(&self, handle: &Self::Handle) -> Result; fn merge_states(&self, states: Vec) -> Result; @@ -130,26 +131,48 @@ down to find it is deployment-specific `QueryExpr`-shape knowledge (mirrors how a real store's edge-fact extraction already works); `execute` doesn't interpret it. -**Grouping (`by`).** `SummaryAgg` carries `by: Vec` (the GROUP BY -columns), but `execute` has no schema-level knowledge of what a "group -value" looks like for a given deployment — that's `find_candidates`'s job: -it tags every handle it returns with the `GroupKey` (an opaque, -deployment-chosen type) that handle belongs to. `execute` groups those -tagged handles itself before folding (`fold_states`/`merge_states` only -ever combine same-group states), and every `ExecOutcome` — `State` or -`Value` — is a per-group list, never a single bare value. The ungrouped -case (`by` empty, or a `Logical` leaf, which has no grouping concept at -all) is simply a list of one entry under `GroupKey::default()`, not a -different shape — callers don't need to special-case it. +**Grouping (`reduction`).** `SummaryAgg` carries `reduction: Reduction` — +the same L3 type (`asap_ir::intent_algebra::Reduction`) the `Aggregate` +node it was bound from carried (issue #165), reused verbatim rather than +flattened to a bare `Vec`. `execute` has no schema-level +knowledge of what a "group value" looks like for a given deployment — +that's `find_candidates`'s job: it tags every handle it returns with the +`GroupKey` (an opaque, deployment-chosen type) that handle belongs to. +`execute` groups those tagged handles itself before folding +(`fold_states`/`merge_states` only ever combine same-group states), and +every `ExecOutcome` — `State` or `Value` — is a per-group list, never a +single bare value. A `Logical` leaf (which has no grouping concept +either, but for a different reason — it's deferred entirely to the +deployment) is simply a list of one entry under `GroupKey::default()`, +not a different shape — callers don't need to special-case it. + +`Reduction` is two variants, not a bare `Vec` (issue #163): +`Reduce(by)` is a genuine cross-series reduction — an *empty* `by` still +means exactly one group, so `find_candidates` must tag every handle it +returns with one shared `GroupKey` there. `PerEntity` means there is no +grouping concept for this shape at all (a bare per-series range function +like `quantile_over_time(...)`, with no aggregation operator and so no +`by(...)` to begin with) — `find_candidates` must tag every handle with +its own distinct `GroupKey` there, one per entity, and `execute` must +never merge across them. Both used to collapse to the same `by: []` on +`SummaryAgg`, with nothing in the trait able to tell them apart; +`asap-plan::bind` decides which `Reduction` a node gets once, at L3→L4 +binding, and carries it onto the L4 node unchanged — the same value that +already decided the node's per-entity-vs-cross-series *schema* — so the +two can't drift out of sync. + `SummaryMerge`'s `(SummaryKind, SummaryParams)` agreement check stays *global* across every group from every child (it's a planning-time property of the node, fixed before any group value is known); the actual state-folding happens *within* each group independently, never across groups. First landed narrower (a single ungrouped value per tree) in #155; -broadened to this shape in response to #159 once a real +broadened to a per-`GroupKey` shape in response to #159 once a real consumer (`data_plane`'s grouped PromQL queries — `sum by (zone)`, `quantile by (zone)(...)`, the common case, not an edge case) showed the -original shape would have silently merged every group's state together. +original shape would have silently merged every group's state together; +`by` was then replaced with `Reduction` in response to #163, once +implementing `find_candidates` for real showed an empty `by` alone +couldn't tell a per-entity shape from an explicit full reduction. ### Nested composition