From 4ff2e0af5511c4fc350dcb958cb10f27666ef5e7 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 09:38:52 -0600 Subject: [PATCH 1/3] refactor(types): split ExactKind::MinMax into Min and Max `ExactKind::MinMax` was never a min/max pair: `function_rules` maps `AggIntent::Max` onto it and `AggIntent::Min` onto the separately added `ExactKind::Min`, so the name promised a two-sided accumulator that the variant never was. Consumers that match on the family had to carry the direction out-of-band (ASAPQuery-backend threads it through the `aggregationSubType` wire string) because the name gave no guarantee about which extremum the state holds. Rename the variant to `Max` in both `ExactKind` and `ExactParams`, so the two directions are symmetric variants and a stored minimum can no longer content-address onto a maximum. Co-Authored-By: Claude Opus 5 (1M context) --- crates/asap-aware-mapping/src/accuracy.rs | 2 +- crates/asap-aware-mapping/src/function_rules.rs | 2 +- crates/asap-aware-mapping/src/replacement.rs | 4 ++-- crates/asap-aware-mapping/src/rollup.rs | 2 +- crates/integration-tests/tests/exact_composition.rs | 6 +++--- crates/types/src/post_asap/schema.rs | 2 +- crates/types/src/post_asap/sketch.rs | 9 ++++----- 7 files changed, 13 insertions(+), 14 deletions(-) diff --git a/crates/asap-aware-mapping/src/accuracy.rs b/crates/asap-aware-mapping/src/accuracy.rs index a132eb39..1590d734 100644 --- a/crates/asap-aware-mapping/src/accuracy.rs +++ b/crates/asap-aware-mapping/src/accuracy.rs @@ -541,7 +541,7 @@ impl DefaultAccuracyModel { } let mut provenance = Vec::new(); let count = row_count(stats, &mut provenance); - let exact_local = ResultGuarantee::exact("ExactAggregate(MinMax)"); + let exact_local = ResultGuarantee::exact("ExactAggregate(Max)"); provenance.extend(composed_provenance( op, inputs, diff --git a/crates/asap-aware-mapping/src/function_rules.rs b/crates/asap-aware-mapping/src/function_rules.rs index e5c5f7ba..0b96e018 100644 --- a/crates/asap-aware-mapping/src/function_rules.rs +++ b/crates/asap-aware-mapping/src/function_rules.rs @@ -21,7 +21,7 @@ pub(crate) fn function_rules(intent: &AggIntent) -> Option { ), AggIntent::Max { .. } => ( CompositionOperator::ExactExtremum, - Some((ExactKind::MinMax, ExactParams::MinMax)), + Some((ExactKind::Max, ExactParams::Max)), ), AggIntent::Avg { .. } => (CompositionOperator::ExactAverage, None), AggIntent::Rate => ( diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index deef4095..d2ddb2a1 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -632,7 +632,7 @@ pub trait ReplacementStrategy { #[derive(Debug, Clone, PartialEq)] pub enum Implementation { /// An exact **mergeable** accumulator (partial state ≡ the value - /// itself: `Sum` / `Count` / `MinMax` / `Rate` / `Increase`). The + /// itself: `Sum` / `Count` / `Min` / `Max` / `Rate` / `Increase`). The /// built state *is* the answer already — no `SummaryEstimate` readout /// step. ExactAggregate { @@ -5921,7 +5921,7 @@ mod tests { // exact mergeable accumulators (A::Sum { col: None }, Acc(E::Sum)), (A::Min { col: None }, Acc(E::Min)), - (A::Max { col: None }, Acc(E::MinMax)), + (A::Max { col: None }, Acc(E::Max)), (A::Rate, Acc(E::Rate)), (A::IRate, Acc(E::IRate)), (A::Increase, Acc(E::Increase)), diff --git a/crates/asap-aware-mapping/src/rollup.rs b/crates/asap-aware-mapping/src/rollup.rs index a5abae8a..351515dd 100644 --- a/crates/asap-aware-mapping/src/rollup.rs +++ b/crates/asap-aware-mapping/src/rollup.rs @@ -143,7 +143,7 @@ fn bindable_grouped_aggregate( /// reasoning behind each arm. `None` for any intent this module does not /// (yet) know a correct combinator for — including every intent /// `agg_is_mergeable` permits but this module doesn't specifically handle -/// (`Rate`, and everything outside the `Sum`/`Count`/`MinMax`/`Increase` +/// (`Rate`, and everything outside the `Sum`/`Count`/`Min`/`Max`/`Increase` /// vocabulary `agg_is_mergeable`'s own doc names) — so `is_legal_rollup_source` /// (which calls this) is *strictly narrower* than `agg_is_mergeable` alone, /// deliberately: `agg_is_mergeable` answers "does *some* partial-state merge diff --git a/crates/integration-tests/tests/exact_composition.rs b/crates/integration-tests/tests/exact_composition.rs index 86b5ad1f..de138d48 100644 --- a/crates/integration-tests/tests/exact_composition.rs +++ b/crates/integration-tests/tests/exact_composition.rs @@ -261,7 +261,7 @@ fn every_exact_accumulator_is_finalized_before_an_outer_sketch() { AggIntent::Max { col: None }, Rc::new(metric_scan(&["zone"])), ), - ExactKind::MinMax, + ExactKind::Max, ), ( per_entity( @@ -616,8 +616,8 @@ fn readout_under_maintenance_is_rejected_at_construction() { expr: SummaryExpr::SummaryAgg { child: post, family: SummaryFamilyType::ExactAggregate( - ExactKind::MinMax, - asap_types::post_asap::ExactParams::MinMax, + ExactKind::Max, + asap_types::post_asap::ExactParams::Max, ), input: SummaryUpdate::column(asap_types::pre_asap::ColumnRef::SampleValue), reduction: Reduction::by(vec![]), diff --git a/crates/types/src/post_asap/schema.rs b/crates/types/src/post_asap/schema.rs index a61abcef..ffef2a05 100644 --- a/crates/types/src/post_asap/schema.rs +++ b/crates/types/src/post_asap/schema.rs @@ -24,7 +24,7 @@ pub enum SummaryFamilyType { /// pre-ASAP `DataType` (`Int64`/`Float64`/`Utf8`/`Bool`/`Timestamp`), /// passed through unchanged from a pre-ASAP edge. Plain(DataType), - /// Exact, mergeable accumulator state (`Sum`/`Count`/`MinMax`/`Rate`/ + /// Exact, mergeable accumulator state (`Sum`/`Count`/`Min`/`Max`/`Rate`/ /// `Increase`) — the partial state *is* the value; no readout needed. ExactAggregate(ExactKind, ExactParams), /// Approximate sketch state (KLL/CMS/HLL/…), read out via a diff --git a/crates/types/src/post_asap/sketch.rs b/crates/types/src/post_asap/sketch.rs index 6a14e126..9d870a6c 100644 --- a/crates/types/src/post_asap/sketch.rs +++ b/crates/types/src/post_asap/sketch.rs @@ -13,10 +13,10 @@ pub enum ExactKind { Sum, /// Exact count accumulator (mergeable by addition). Count, - /// Exact min/max accumulator (mergeable by comparison). - MinMax, - /// Exact minimum, distinct from the legacy maximum accumulator. + /// Exact minimum accumulator (mergeable by comparison). Min, + /// Exact maximum accumulator (mergeable by comparison). + Max, /// Exact increase accumulator (counter-reset-aware delta). Increase, /// Rate accumulator (increase / time window duration). @@ -33,9 +33,8 @@ pub enum ExactKind { pub enum ExactParams { Sum, Count, - MinMax, - /// Exact minimum, distinct from the legacy maximum accumulator. Min, + Max, Increase, Rate, IRate, From 029ff2fe041172c94c2d32c90b185bc83c5e8a57 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 10:27:59 -0600 Subject: [PATCH 2/3] feat(promql): restore a distinct-count idiom over mergeable cardinality Lowering `count(v)` to a row count (15e6cf8) matched PromQL, but it left no way to ask for a distinct count that reduces across series. The `Cardinality` intent survived only under `distinct_over_time(v[w])`, which is per-series: a consumer holding one mergeable cardinality state per source could answer the merged question and had no expression that asked it. Collapse `count(distinct_over_time(v[w]))` into a single `Aggregate{[Cardinality]}` reduced by the outer `by`. The nested reading -- count the series that have a distinct-count -- is not what anyone writes this expression for, and it discards the mergeability of the state underneath. The collapsed form needs its own reduction rule: `reduction_for` reads the surviving `TimeRange` as one row per series and answers `PerEntity`, which is right for a bare `distinct_over_time(v[w])` and wrong once an explicit aggregator wraps it. `windowed_reduce` keeps the outer aggregator's grouping, so an empty `by` reduces everything. Co-Authored-By: Claude Opus 5 (1M context) --- crates/frontend-promql/src/promql.rs | 40 +++++++++++++++++++ .../frontend-promql/tests/promql_lowering.rs | 39 ++++++++++++++++++ 2 files changed, 79 insertions(+) diff --git a/crates/frontend-promql/src/promql.rs b/crates/frontend-promql/src/promql.rs index 4a5861fa..765d82e4 100644 --- a/crates/frontend-promql/src/promql.rs +++ b/crates/frontend-promql/src/promql.rs @@ -46,6 +46,7 @@ //! | `group` / `offset` / `@` / `info` | **rejected** — distinct semantics with no intent-algebra representation yet (`info` label-join → #84) | //! | `OUTER by (dims) (…)` | `Aggregate.reduction = Reduce(by = dims)` (generic `topk by`/`bottomk` grouping → `Sort.partition_by`) | //! | `count by (d) (…)` | `Aggregate{[Count], …}` | +//! | `count by (d) (distinct_over_time(v[w]))` | `Aggregate{[Cardinality], …}` — the distinct-count idiom; one merged cardinality, not a count of series | //! | `group(v)` / `count_values("l", v)` | `Aggregate{[Group]}` (constant 1) / `Aggregate{[CountValues{l}]}` (group-by-value + count, new label `l`) — issue #49 | //! | `limitk(k, v)` / `limit_ratio(r, v)` | `PromqlSeriesSample{LimitK(k) \| LimitRatio(r)}` — series-sampling selection, whole series kept unchanged (issue #86) | //! | `topk(k, count_over_time(…))` / `topk(k, sum_over_time(…))` | `Aggregate{[TopK{k}]}` (heavy-hitter intent) over the explicit inner `Aggregate{[Count/Sum]}` | @@ -1454,6 +1455,21 @@ fn build(inner: Inner, keys: Vec, outer: Outer) -> Result accuracy: current_accuracy(), }, ), + // `count(distinct_over_time(v[w]))` is the distinct-count + // idiom: one cardinality over the window, reduced by the outer + // `by`. Nesting the two aggregates instead would count *series* + // that have a distinct-count -- an answer nobody writes this + // expression for, and one that throws away the mergeability of + // the cardinality state underneath. `count(v)` on its own stays + // a row count. + Some(InnerFunc::Cardinality) => windowed_reduce( + inner, + keys, + AggIntent::Cardinality { + col: None, + accuracy: current_accuracy(), + }, + ), Some(f) => { let inner_i = inner_intent(f); let inner_agg = windowed_aggregate(inner, vec![], inner_i); @@ -1621,6 +1637,30 @@ fn windowed_aggregate( } } +/// [`windowed_aggregate`] for a collapsed two-level form, where an explicit +/// outer aggregator supplies the grouping. [`reduction_for`]'s range-child +/// rule would read the surviving `TimeRange` as "one row per series" and +/// answer `PerEntity`, which is right for a bare `distinct_over_time(v[w])` +/// but wrong once the user wrote `count(...)` around it: that is a real +/// cross-series reduction, and an empty `by` means reduce everything. +fn windowed_reduce(inner: Inner, keys: Vec, intent: AggIntent) -> Unresolved { + let base = filtered_source(inner.metric, inner.matchers, inner.shift); + let child = match inner.window { + Some(w) => Unresolved::TimeRange { + range: w, + child: Rc::new(base), + }, + None => base, + }; + Unresolved::Aggregate { + reduction: Reduction::Reduce(GroupKeys::by(keys)), + measures: vec![intent], + output_names: vec![String::new()], + having: None, + child: Rc::new(child), + } +} + /// `Aggregate{reduction, [intent]}` directly over an existing Unresolved subtree — the /// OUTER level of a two-level aggregation such as `sum(rate(…))` or the /// `Aggregate{[Quantile]}` that wraps a `histogram_quantile` argument. diff --git a/crates/frontend-promql/tests/promql_lowering.rs b/crates/frontend-promql/tests/promql_lowering.rs index 128783f9..ba03468a 100644 --- a/crates/frontend-promql/tests/promql_lowering.rs +++ b/crates/frontend-promql/tests/promql_lowering.rs @@ -380,6 +380,45 @@ fn count_over_rate_keeps_both_levels() { )); } +#[test] +fn count_over_distinct_over_time_collapses_to_one_cardinality() { + // `count(v)` is a row count, but `count(distinct_over_time(v[w]))` is the + // distinct-count idiom -- a single merged cardinality reduced by the outer + // `by`, not a count of the series that happen to have one. Keeping it as + // one Aggregate is what lets mergeable cardinality state answer it across + // sources. + for (query, reduction) in [ + ( + "count(distinct_over_time(unique_users[5m]))", + Reduction::by(Vec::new()), + ), + ( + "count by (job) (distinct_over_time(unique_users[5m]))", + Reduction::by(vec![2]), + ), + ] { + let qe = lower(query); + let QueryExpr::Aggregate { + measures, + reduction: actual, + child, + .. + } = &qe + else { + panic!("expected a single Aggregate{{Cardinality}}, got {qe:?}"); + }; + assert!( + matches!(measures.as_slice(), [AggIntent::Cardinality { .. }]), + "{query}: {measures:?}" + ); + assert_eq!(actual, &reduction, "{query}"); + assert!( + !matches!(child.as_ref(), QueryExpr::Aggregate { .. }), + "{query}: the two levels must collapse into one, got {child:?}" + ); + } +} + // ── count / cardinality ─────────────────────────────────────────────────────── #[test] From 5e385aa813b76dc49269544e9162d248645c3c5b Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 15:06:43 -0600 Subject: [PATCH 3/3] fix(promql): preserve count over per-series cardinality --- crates/frontend-promql/src/promql.rs | 40 ------------------- .../frontend-promql/tests/promql_lowering.rs | 38 +++++++++++------- 2 files changed, 24 insertions(+), 54 deletions(-) diff --git a/crates/frontend-promql/src/promql.rs b/crates/frontend-promql/src/promql.rs index 765d82e4..4a5861fa 100644 --- a/crates/frontend-promql/src/promql.rs +++ b/crates/frontend-promql/src/promql.rs @@ -46,7 +46,6 @@ //! | `group` / `offset` / `@` / `info` | **rejected** — distinct semantics with no intent-algebra representation yet (`info` label-join → #84) | //! | `OUTER by (dims) (…)` | `Aggregate.reduction = Reduce(by = dims)` (generic `topk by`/`bottomk` grouping → `Sort.partition_by`) | //! | `count by (d) (…)` | `Aggregate{[Count], …}` | -//! | `count by (d) (distinct_over_time(v[w]))` | `Aggregate{[Cardinality], …}` — the distinct-count idiom; one merged cardinality, not a count of series | //! | `group(v)` / `count_values("l", v)` | `Aggregate{[Group]}` (constant 1) / `Aggregate{[CountValues{l}]}` (group-by-value + count, new label `l`) — issue #49 | //! | `limitk(k, v)` / `limit_ratio(r, v)` | `PromqlSeriesSample{LimitK(k) \| LimitRatio(r)}` — series-sampling selection, whole series kept unchanged (issue #86) | //! | `topk(k, count_over_time(…))` / `topk(k, sum_over_time(…))` | `Aggregate{[TopK{k}]}` (heavy-hitter intent) over the explicit inner `Aggregate{[Count/Sum]}` | @@ -1455,21 +1454,6 @@ fn build(inner: Inner, keys: Vec, outer: Outer) -> Result accuracy: current_accuracy(), }, ), - // `count(distinct_over_time(v[w]))` is the distinct-count - // idiom: one cardinality over the window, reduced by the outer - // `by`. Nesting the two aggregates instead would count *series* - // that have a distinct-count -- an answer nobody writes this - // expression for, and one that throws away the mergeability of - // the cardinality state underneath. `count(v)` on its own stays - // a row count. - Some(InnerFunc::Cardinality) => windowed_reduce( - inner, - keys, - AggIntent::Cardinality { - col: None, - accuracy: current_accuracy(), - }, - ), Some(f) => { let inner_i = inner_intent(f); let inner_agg = windowed_aggregate(inner, vec![], inner_i); @@ -1637,30 +1621,6 @@ fn windowed_aggregate( } } -/// [`windowed_aggregate`] for a collapsed two-level form, where an explicit -/// outer aggregator supplies the grouping. [`reduction_for`]'s range-child -/// rule would read the surviving `TimeRange` as "one row per series" and -/// answer `PerEntity`, which is right for a bare `distinct_over_time(v[w])` -/// but wrong once the user wrote `count(...)` around it: that is a real -/// cross-series reduction, and an empty `by` means reduce everything. -fn windowed_reduce(inner: Inner, keys: Vec, intent: AggIntent) -> Unresolved { - let base = filtered_source(inner.metric, inner.matchers, inner.shift); - let child = match inner.window { - Some(w) => Unresolved::TimeRange { - range: w, - child: Rc::new(base), - }, - None => base, - }; - Unresolved::Aggregate { - reduction: Reduction::Reduce(GroupKeys::by(keys)), - measures: vec![intent], - output_names: vec![String::new()], - having: None, - child: Rc::new(child), - } -} - /// `Aggregate{reduction, [intent]}` directly over an existing Unresolved subtree — the /// OUTER level of a two-level aggregation such as `sum(rate(…))` or the /// `Aggregate{[Quantile]}` that wraps a `histogram_quantile` argument. diff --git a/crates/frontend-promql/tests/promql_lowering.rs b/crates/frontend-promql/tests/promql_lowering.rs index ba03468a..fb2af49b 100644 --- a/crates/frontend-promql/tests/promql_lowering.rs +++ b/crates/frontend-promql/tests/promql_lowering.rs @@ -381,40 +381,50 @@ fn count_over_rate_keeps_both_levels() { } #[test] -fn count_over_distinct_over_time_collapses_to_one_cardinality() { - // `count(v)` is a row count, but `count(distinct_over_time(v[w]))` is the - // distinct-count idiom -- a single merged cardinality reduced by the outer - // `by`, not a count of the series that happen to have one. Keeping it as - // one Aggregate is what lets mergeable cardinality state answer it across - // sources. +fn count_over_distinct_over_time_preserves_both_aggregates() { + // One series with window samples [1, 2] produces one distinct-count + // result (value 2). The outer count counts that one series, yielding 1. for (query, reduction) in [ ( "count(distinct_over_time(unique_users[5m]))", - Reduction::by(Vec::new()), + Reduction::by(vec![]), ), ( "count by (job) (distinct_over_time(unique_users[5m]))", Reduction::by(vec![2]), ), ] { - let qe = lower(query); + let tree = lower(query); let QueryExpr::Aggregate { measures, reduction: actual, child, .. - } = &qe + } = &tree else { - panic!("expected a single Aggregate{{Cardinality}}, got {qe:?}"); + panic!("expected outer Count: {tree:?}"); }; assert!( - matches!(measures.as_slice(), [AggIntent::Cardinality { .. }]), - "{query}: {measures:?}" + matches!(measures.as_slice(), [AggIntent::Count { .. }]), + "{query}: {tree:?}" ); assert_eq!(actual, &reduction, "{query}"); + let QueryExpr::Aggregate { + measures, + reduction, + child, + .. + } = child.as_ref() + else { + panic!("expected inner per-series Cardinality: {tree:?}"); + }; + assert!( + matches!(measures.as_slice(), [AggIntent::Cardinality { .. }]), + "{query}: {tree:?}" + ); + assert_eq!(reduction, &Reduction::PerEntity, "{query}"); assert!( - !matches!(child.as_ref(), QueryExpr::Aggregate { .. }), - "{query}: the two levels must collapse into one, got {child:?}" + matches!(child.as_ref(), QueryExpr::TimeRange { range, .. } if range.as_secs() == 300) ); } }