feat(promql): lower group() and count_values() aggregation operators (#49)

crates/frontend-promql/src/promql.rs

@@ -33,6 +33,7 @@
 //! | `group` / `offset` / `@` / `info` | **rejected** — distinct semantics with no intent-algebra representation yet (`info` label-join → #84) |
 //! | `OUTER by (dims) (…)` | `Aggregate.keys = dims` (→ positional `Aggregate.by` in L3; generic `topk by`/`bottomk` grouping → `Sort.partition_by`) |
 //! | `count by (d) (…)` | `Aggregate{[CountDistinct], …}` (→ `Cardinality`) |
+//! | `group(v)` / `count_values("l", v)` | `Aggregate{[Group]}` (constant 1) / `Aggregate{[CountValues{l}]}` (group-by-value + count, new label `l`) — issue #49; `limitk`/`limit_ratio` series sampling → #86 |
 //! | `topk(k, count_over_time(…))` | `TopK{k, by}` (heavy-hitter intent) |
 //! | `topk(k, <other>)` / `bottomk(k, …)` | `Sort{value} → Limit{k}` |
 //! | `m{f}` | `Filter(Source)` |
@@ -68,6 +69,9 @@ enum Outer {
     None,
     Plain(OuterIntent),
     Count,
+    /// `count_values("l", v)` — group by value + count, emitting the value as a
+    /// new label `l` (issue #49).
+    CountValues { label: String },
     TopK { k: u64, descending: bool },
 }
 
@@ -80,6 +84,8 @@ enum OuterIntent {
     StdDev,
     Variance,
     Quantile(f64),
+    /// `group(v)` — constant 1 per group (issue #49).
+    Group,
 }
 
 #[derive(Debug, Clone)]
@@ -355,11 +361,14 @@ fn outer_kind(agg: &AggregateExpr) -> Result<Outer> {
         Outer::Plain(OuterIntent::Sum)
     } else if op == token::T_GROUP {
         // `group(v)` yields a constant 1 per group (presence), not a sum of
-        // values. Folding it onto `Sum` changed the result; reject until a
-        // distinct group-presence intent exists.
-        return Err(LoweringError::UnsupportedAggregateOp(
-            "`group` (constant-1 presence) is not `sum`; no distinct intent yet".into(),
-        ));
+        // values — a distinct intent, never folded onto `Sum` (issue #49).
+        Outer::Plain(OuterIntent::Group)
+    } else if op == token::T_COUNT_VALUES {
+        // `count_values("l", v)` groups by sample value and counts, emitting the
+        // value as a new label `l` (the string parameter) — issue #49.
+        Outer::CountValues {
+            label: str_param(agg)?,
+        }
     } else if op == token::T_AVG {
         Outer::Plain(OuterIntent::Avg)
     } else if op == token::T_MIN {
@@ -394,6 +403,9 @@ fn build_over_subtree(outer: Outer, keys: Vec<ColumnRef>, child: L2) -> Result<L
         Outer::None => child,
         Outer::Plain(intent) => outer_aggregate(keys, outer_func(&intent), child),
         Outer::Count => outer_aggregate(keys, AggFunc::CountDistinct, child),
+        Outer::CountValues { label } => {
+            outer_aggregate(keys, AggFunc::CountValues { label }, child)
+        }
         Outer::TopK { k, descending } => {
             let sorted = L2::Sort {
                 keys: vec![L2SortKey {
@@ -849,6 +861,17 @@ fn build(inner: Inner, keys: Vec<ColumnRef>, outer: Outer) -> Result<L2> {
                 outer_aggregate(keys, AggFunc::CountDistinct, inner_agg)
             }
         }),
+        Outer::CountValues { label } => {
+            let func = AggFunc::CountValues { label };
+            Ok(match &inner.func {
+                None => windowed_aggregate(inner, keys, func),
+                Some(f) => {
+                    let inner_f = inner_func(f);
+                    let inner_agg = windowed_aggregate(inner, vec![], inner_f);
+                    outer_aggregate(keys, func, inner_agg)
+                }
+            })
+        }
         Outer::TopK { k, descending } => {
             // Heavy-hitter only when ranking by frequency (`count`): that is a
             // first-class aggregate intent → `TopK`. Any other ranking (topk
@@ -999,6 +1022,28 @@ fn outer_func(o: &OuterIntent) -> AggFunc {
         OuterIntent::StdDev => AggFunc::StdDev { population: true },
         OuterIntent::Variance => AggFunc::Variance { population: true },
         OuterIntent::Quantile(q) => AggFunc::Quantile(*q),
+        OuterIntent::Group => AggFunc::Group,
+    }
+}
+
+/// A `count_values` string parameter (the synthesized label name). PromQL wraps
+/// it in a `StringLiteral`, possibly parenthesised (`count_values((("v")), …)`).
+fn str_param(agg: &AggregateExpr) -> Result<String> {
+    fn unwrap_str(expr: &Expr) -> Result<String> {
+        match expr {
+            Expr::StringLiteral(s) => Ok(s.val.clone()),
+            Expr::Paren(p) => unwrap_str(&p.expr),
+            other => Err(LoweringError::InvalidParameter(format!(
+                "`count_values` label must be a string literal, got {:?}",
+                std::mem::discriminant(other)
+            ))),
+        }
+    }
+    match &agg.param {
+        Some(e) => unwrap_str(e),
+        None => Err(LoweringError::MissingArgument(
+            "`count_values` label parameter".into(),
+        )),
     }
 }
 

crates/frontend-promql/tests/promql_conformance.rs

@@ -310,10 +310,13 @@ fn sum_without_is_rejected() {
 }
 
 #[test]
-fn group_aggregator_is_rejected() {
+fn group_aggregator_lowers_to_a_distinct_intent() {
     // SEMANTICS (PromQL): `group(v)` returns a constant 1 per group (presence),
-    // NOT a sum. Rather than fold it onto `Sum` (wrong value) we reject it.
-    let _ = rejected("group by (job) (up)");
+    // NOT a sum. It now lowers to a distinct `Group` intent (never folded onto
+    // `Sum`) — see §S. Regression guard that it is not a `Sum`.
+    let qe = ok("group by (job) (up)");
+    assert!(has(&qe, |i| *i == AggIntent::Group));
+    assert!(!has(&qe, |i| matches!(i, AggIntent::Sum { .. })));
 }
 
 // ─────────────────────────────────────────────────────────────────────────────
@@ -1278,3 +1281,92 @@ fn info_is_rejected__GAP() {
     // so adding support flips this deliberately.
     let _ = rejected("info(rate(http_requests_total[5m]))");
 }
+
+// ─────────────────────────────────────────────────────────────────────────────
+// S. Extended aggregation operators: group / count_values (aggregators.test; #49)
+// ─────────────────────────────────────────────────────────────────────────────
+
+#[test]
+fn group_lowers_to_a_constant_group_intent() {
+    // SEMANTICS: `group(v)` yields a constant 1 per group — a distinct intent,
+    // NOT folded onto `sum` (which would return the value sum instead of 1).
+    let qe = ok("group(up)");
+    let QueryExpr::Aggregate { aggs, .. } = &qe else {
+        panic!("expected an Aggregate, got {qe:?}");
+    };
+    assert!(matches!(aggs.as_slice(), [AggIntent::Group]));
+    // Output column is the constant-1 `group` value.
+    let sch = qe.output_schema().unwrap();
+    assert!(sch.columns.iter().any(|c| c.name == "group"));
+}
+
+#[test]
+fn group_by_keeps_the_grouping_keys() {
+    // `group by (job) (up)` — the grouping keys ride on `Aggregate.by`.
+    let qe = ok("group by (job) (up)");
+    let sch = qe.output_schema().unwrap();
+    assert!(sch.columns.iter().any(|c| c.name == "job"));
+    assert!(has(&qe, |i| *i == AggIntent::Group));
+}
+
+#[test]
+fn count_values_groups_by_value_and_synthesizes_a_label() {
+    // SEMANTICS: `count_values("l", v)` groups the input series by their sample
+    // value, counts each distinct value, and emits that value as a new label
+    // `l`. The intent carries the label; schema gains a `Utf8` `l` column.
+    let qe = ok(r#"count_values("version", build_version)"#);
+    let QueryExpr::Aggregate { aggs, .. } = &qe else {
+        panic!("expected an Aggregate, got {qe:?}");
+    };
+    assert!(
+        matches!(aggs.as_slice(), [AggIntent::CountValues { label }] if label == "version")
+    );
+    let sch = qe.output_schema().unwrap();
+    let version = sch
+        .columns
+        .iter()
+        .find(|c| c.name == "version")
+        .expect("synthesized `version` label column");
+    assert_eq!(version.dtype, DataType::Utf8, "the value becomes a string label");
+    assert!(
+        sch.columns.iter().any(|c| c.name == "count"),
+        "and a count column"
+    );
+}
+
+#[test]
+fn count_values_accepts_a_parenthesised_label_and_by_grouping() {
+    // `count_values by (job) ((("v")), m)` — nested parens around the string
+    // param, plus `by` grouping. Both survive.
+    let qe = ok(r#"count_values by (job) ((("v")), m)"#);
+    assert!(has(
+        &qe,
+        |i| matches!(i, AggIntent::CountValues { label } if label == "v")
+    ));
+    let sch = qe.output_schema().unwrap();
+    assert!(sch.columns.iter().any(|c| c.name == "job"));
+    assert!(sch.columns.iter().any(|c| c.name == "v"));
+}
+
+#[test]
+fn count_values_label_colliding_with_a_group_key_is_not_duplicated() {
+    // `count_values by (job)("job", v)` — the synthesized label name collides
+    // with a group-by key. PromQL's synthesized label takes precedence; the
+    // output must carry a single `job` column, never two.
+    let qe = ok(r#"count_values by (job) ("job", version)"#);
+    let sch = qe.output_schema().unwrap();
+    let jobs = sch.columns.iter().filter(|c| c.name == "job").count();
+    assert_eq!(jobs, 1, "collision deduped, got {:?}", sch.columns);
+    assert!(sch.columns.iter().any(|c| c.name == "count"));
+}
+
+#[test]
+fn limitk_and_limit_ratio_are_rejected__GAP() {
+    // `limitk`/`limit_ratio` are series-*sampling* operators — they return a
+    // deterministic-but-unordered subset of the input series unchanged. Modeling
+    // them as `topk` (order-by-value + limit) would change *which* series pass
+    // through, so they are cleanly rejected pending a sampling-selection node
+    // (follow-up #86), not mislowered.
+    let _ = rejected("limitk(2, http_requests)");
+    let _ = rejected("limit_ratio(0.1, http_requests)");
+}

crates/frontend-promql/tests/promql_equivalence.rs

@@ -174,9 +174,10 @@ fn changes_and_resets_are_not_count_over_time() {
 #[test]
 fn group_is_not_sum() {
     // PromQL `group` returns a constant 1 per group; it previously collapsed
-    // onto `sum` (sum of values).
-    assert_rejected("group(up)");
-    assert_rejected("group by (job) (up)");
+    // onto `sum` (sum of values). It now lowers to its own `Group` intent
+    // (issue #49) — distinct L3 from `sum`, not merged.
+    assert_distinct("group(up)", "sum(up)");
+    assert_distinct("group by (job) (up)", "sum by (job) (up)");
 }
 
 #[test]

crates/ir/src/intent_algebra/agg_intent.rs

@@ -176,6 +176,23 @@ pub enum AggIntent {
     /// (`time()` is the evaluation time itself — a `QueryExpr::EvalTime` leaf,
     /// not this.)
     TimeFn(TimeFunc),
+
+    // ── Extended aggregation operators (issue #49) ───────────────────────
+    /// PromQL `group(v)` — a constant `1` per group ("group presence"). The
+    /// grouping keys ride on the enclosing `Aggregate.by`. Deliberately NOT
+    /// aliased to `Sum`/`Count`: the output value is always 1, independent of
+    /// the input values (folding it onto `Sum` would change the result).
+    Group,
+    /// PromQL `count_values("l", v)` — group the input series by their sample
+    /// *value* and count each distinct value, emitting that value as a new
+    /// label `l`. Output = one series per distinct value, with labels
+    /// `by-keys ∪ {l}` and value = the count. Unlike every other reducer this
+    /// adds a synthesized `Utf8` label column, so `Aggregate` schema derivation
+    /// special-cases it (two output columns, not one).
+    CountValues {
+        /// The name of the synthesized label carrying the stringified value.
+        label: String,
+    },
 }
 
 /// Time / calendar accessor functions (issue #46), evaluated over a timestamp.
@@ -363,6 +380,12 @@ impl AggIntent {
             AggIntent::AbsentOverTime => col("absent_over_time", DataType::Float64, false),
             AggIntent::PresentOverTime => col("present_over_time", DataType::Float64, false),
             AggIntent::TimeFn(_) => col("value", DataType::Float64, false),
+            // `group` — constant 1 per group.
+            AggIntent::Group => col("group", DataType::Float64, false),
+            // `count_values` — the *value* column (the per-value count). The
+            // synthesized `label` column is added alongside it by `Aggregate`
+            // schema derivation, which special-cases this intent.
+            AggIntent::CountValues { .. } => col("count", DataType::Int64, false),
         }
     }
 }
@@ -403,6 +426,8 @@ pub fn agg_is_exact(op: &AggIntent) -> bool {
             | AggIntent::Avg { .. }
             | AggIntent::Min { .. }
             | AggIntent::Max { .. }
+            | AggIntent::Group
+            | AggIntent::CountValues { .. }
     )
 }
 

crates/ir/src/intent_algebra/query_expr.rs

@@ -495,6 +495,24 @@ impl QueryExpr {
                 // value probe (PromQL's single-column convention). A non-empty
                 // `output_names[i]` overrides the synthetic output column name.
                 for (i, intent) in aggs.iter().enumerate() {
+                    // `count_values("l", v)` emits TWO columns: the synthesized
+                    // `Utf8` label `l` (the stringified sample value it groups
+                    // by) and the per-value count. `output_names[i]` still
+                    // overrides the count column's name if set. If `l` collides
+                    // with a group-by key of the same name, PromQL's synthesized
+                    // label takes precedence — emit a single column, never a
+                    // duplicate.
+                    if let AggIntent::CountValues { label } = intent {
+                        if !out_cols.iter().any(|c| c.name == *label) {
+                            out_cols.push(Column::new(label.clone(), DataType::Utf8, false));
+                        }
+                        let mut cnt = intent.output_column(&probe);
+                        if let Some(name) = output_names.get(i).filter(|s| !s.is_empty()) {
+                            cnt.name = name.clone();
+                        }
+                        out_cols.push(cnt);
+                        continue;
+                    }
                     let in_col = intent
                         .input_col()
                         .and_then(|id| in_schema.columns.get(id))
@@ -505,7 +523,12 @@ impl QueryExpr {
                     }
                     out_cols.push(out);
                 }
-                let unique_keys = if by.is_empty() {
+                // `count_values` groups by (by-keys ∪ the synthesized value
+                // label), so the by-keys alone are not a unique key — be
+                // conservative and claim none.
+                let has_count_values =
+                    aggs.iter().any(|a| matches!(a, AggIntent::CountValues { .. }));
+                let unique_keys = if by.is_empty() || has_count_values {
                     Vec::new()
                 } else {
                     vec![(0..by.len()).collect()]

crates/l2/src/lower.rs

@@ -601,6 +601,10 @@ fn agg_func_to_intent(func: &AggFunc, acc: &AccuracyTarget, col: Option<ColumnId
         AggFunc::AbsentOverTime => AggIntent::AbsentOverTime,
         AggFunc::PresentOverTime => AggIntent::PresentOverTime,
         AggFunc::TimeFn(f) => AggIntent::TimeFn(*f),
+        AggFunc::Group => AggIntent::Group,
+        AggFunc::CountValues { label } => AggIntent::CountValues {
+            label: label.clone(),
+        },
     }
 }
 

crates/l2/src/relational.rs

@@ -159,6 +159,12 @@ pub enum AggFunc {
     /// PromQL time / calendar accessor (`timestamp`/`hour`/`day_of_week`/…) →
     /// `AggIntent::TimeFn` (issue #46).
     TimeFn(TimeFunc),
+    /// PromQL `group(v)` → `AggIntent::Group` (issue #49).
+    Group,
+    /// PromQL `count_values("l", v)` → `AggIntent::CountValues` (issue #49).
+    CountValues {
+        label: String,
+    },
 }
 
 /// The Layer-2 relational query IR.