fix(lower): topk over a bare selector ranks its own samples, preserving by-labels (#30)

`topk(k, <bare instant selector>) by (labels)` — e.g.
`topk(3, http_requests_total) by (job)` ("top-3 series per job") — mislowered.
The non-heavy-hitter path in `build`'s `Outer::TopK` else-branch defaulted a
bare selector argument (`inner.func == None`) to an implicit cross-series
`AggFunc::Sum`. That reducing `Sum` collapsed every label (including the `by`
partition keys) into a single `sum` column, so `Sort.partition_by = [job]` no
longer resolved at L3 — a regression surfaced reviewing the Partition →
`Sort.partition_by` reframe (#12, PR #18). It was also semantically wrong:
PromQL `topk` ranks the raw instant-vector samples, it does not sum them.

A bare selector now ranks over the `filtered_source` directly (label-
preserving), so `Sort.partition_by` ranks within each group. A range-vector-
function argument (`topk(k, rate(m[5m]))`) still reduces per series first —
also label-preserving — so those paths are unchanged, as are the heavy-hitter
`count_over_time` and `bottomk` cases.

Expected: `topk(3, http_requests_total) by (job)`
→ `Limit{3} → Sort{value desc, partition_by:[job]} → Scan`.

Tests: bare-selector topk by-label (ranks per group, no implicit Sum) and
bare-selector topk without `by` (ranks raw samples).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: zz_y

crates/lower/src/promql.rs

@@ -418,15 +418,21 @@ fn build(inner: Inner, keys: Vec<ColumnRef>, outer: Outer) -> Result<L2> {
                     input: Box::new(count_agg),
                 })
             } else {
-                let func = match &inner.func {
-                    Some(f) => inner_func(f),
-                    None => AggFunc::Sum,
+                // The base over which we rank. A range-vector-function argument
+                // (`topk(k, rate(m[5m]))`) reduces *per series* first — that is
+                // label-preserving, so the `by (host)` partition labels survive.
+                // A **bare instant selector** (`topk(k, m)`) ranks its own
+                // samples directly: it must NOT be wrapped in a reducing
+                // aggregate. Defaulting it to `Sum` was both semantically wrong
+                // (PromQL `topk` ranks the raw samples, it does not sum them) and
+                // destructive — the cross-series `Sum` collapses every label,
+                // including the `by (…)` partition keys, so they no longer
+                // resolve at L3 (issue #30). Keep the selector label-preserving so
+                // `Sort.partition_by` can rank within each group (issue #12).
+                let base = match inner.func.as_ref().map(inner_func) {
+                    Some(func) => windowed_aggregate(inner, vec![], func),
+                    None => filtered_source(inner.metric, inner.matchers),
                 };
-                // The grouping (`topk by (host)`) is per-group *ranking*, not a
-                // reduction — it rides on `Sort.partition_by` (→ positional L3
-                // `Sort.partition_by`), so the windowed reduction below stays
-                // label-preserving with no group keys (issue #12).
-                let base = windowed_aggregate(inner, vec![], func);
                 let sorted = L2::Sort {
                     keys: vec![L2SortKey {
                         expr: L2Expr::Column(ColumnRef::SampleValue),

crates/lower/tests/promql_lowering.rs

@@ -724,3 +724,59 @@ fn generic_topk_grouping_lowers_to_sort_partition_by() {
     assert_eq!(partition_by, &vec![2], "host is col 2 in [ts, value, host]");
     assert!(matches!(child.as_ref(), QueryExpr::Aggregate { by, .. } if by.is_empty()));
 }
+
+#[test]
+fn topk_over_bare_selector_by_label_ranks_per_group() {
+    // `topk(3, http_requests_total) by (job)` — top-3 series per `job`. A bare
+    // instant selector ranks its OWN samples; it must not be wrapped in an
+    // implicit cross-series `Sum`, which would collapse the `job` partition
+    // label before `Sort.partition_by` resolves it (issue #30 — follow-up to the
+    // Partition→Sort.partition_by reframe in #12). Expected:
+    //   Limit{3} → Sort{value desc, partition_by:[job]} → Scan
+    let q = lower("topk(3, http_requests_total) by (job)");
+    let QueryExpr::Limit { n, child, .. } = &q else {
+        panic!("expected Limit, got {q:?}");
+    };
+    assert_eq!(*n, 3);
+    let QueryExpr::Sort {
+        keys,
+        partition_by,
+        child,
+    } = child.as_ref()
+    else {
+        panic!("expected Sort, got {child:?}");
+    };
+    assert!(!keys[0].ascending, "topk ranks descending");
+    assert_eq!(partition_by, &vec![2], "job is col 2 in [ts, value, job]");
+    // No implicit reducing aggregate — the selector is label-preserving, so the
+    // sort is directly over the Scan (the `job` label survives to partition by).
+    assert!(
+        matches!(child.as_ref(), QueryExpr::Scan { .. }),
+        "ranking is over the bare Scan, not a reducing Aggregate, got {child:?}"
+    );
+    assert!(
+        !has_intent(&q, |i| matches!(i, AggIntent::Sum { .. })),
+        "no implicit Sum is introduced over a bare selector"
+    );
+}
+
+#[test]
+fn topk_over_bare_selector_ranks_raw_samples() {
+    // Even without `by`, `topk(3, m)` ranks the raw instant-vector samples — it
+    // does not sum them. The sort sits directly over the Scan, partition empty.
+    let q = lower("topk(3, http_requests_total)");
+    let QueryExpr::Limit { child, .. } = &q else {
+        panic!("expected Limit, got {q:?}");
+    };
+    let QueryExpr::Sort {
+        partition_by,
+        child,
+        ..
+    } = child.as_ref()
+    else {
+        panic!("expected Sort, got {child:?}");
+    };
+    assert!(partition_by.is_empty(), "no `by` → global ranking");
+    assert!(matches!(child.as_ref(), QueryExpr::Scan { .. }));
+    assert!(!has_intent(&q, |i| matches!(i, AggIntent::Sum { .. })));
+}