fix(sketch): SummaryAgg carries L3's Reduction directly, resolving the empty-by ambiguity

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,41 +59,50 @@ 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);
     };
     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 {
         panic!("expected inner SummaryAgg for rate, got {:?}", child.expr);
     };
     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),

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<ColumnId>` alone can't (issue #163).
+    // state column.
     let per_series = matches!(reduction, Reduction::PerEntity);
     let by: Vec<usize> = 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<ColumnId>` — 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,15 +346,15 @@ mod tests {
             sketch,
             params,
             col,
-            by,
+            reduction,
         } = &sketch_input.expr
         else {
             panic!("expected SummaryAgg, got {:?}", sketch_input.expr);
         };
         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<ColumnId>`.
+    #[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<ColumnId>` 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

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<Vec<(Self::GroupKey, Self::Handle)>, Self::Error>;
 
@@ -119,9 +133,9 @@ pub fn execute<E: SummaryExecutor>(
             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<L4Node>) -> Rc<L4Node> {
+        agg_node_with_reduction(sketch, params, child, Reduction::by(vec![]))
+    }
+
+    fn agg_node_with_reduction(
+        sketch: SummaryKind,
+        params: SummaryParams,
+        child: Rc<L4Node>,
+        reduction: Reduction,
+    ) -> Rc<L4Node> {
         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<HashMap<u64, (String, f64)>>,
@@ -355,7 +384,7 @@ mod tests {
             _sketch: &SummaryKind,
             _params: &SummaryParams,
             _col: &ColumnRef,
-            _by: &[ColumnId],
+            _reduction: &Reduction,
             _child: &L4Node,
         ) -> Result<Vec<(Self::GroupKey, Self::Handle)>, 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<Vec<Reduction>>,
+    }
+
+    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<Vec<(Self::GroupKey, Self::Handle)>, Self::Error> {
+            self.seen.borrow_mut().push(reduction.clone());
+            Ok(vec![(0, 0)])
+        }
+
+        fn fetch_state(&self, _handle: &Self::Handle) -> Result<Self::State, Self::Error> {
+            Ok(1.0)
+        }
+
+        fn merge_states(&self, states: Vec<Self::State>) -> Result<Self::State, Self::Error> {
+            Ok(states.into_iter().sum())
+        }
+
+        fn readout(
+            &self,
+            state: &Self::State,
+            _query: &SketchQuery,
+        ) -> Result<Self::Value, Self::Error> {
+            Ok(*state)
+        }
+
+        fn logical(&self, _expr: &QueryExpr) -> Result<Self::Value, Self::Error> {
+            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]);
+    }
 }