Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 14 additions & 8 deletions crates/lower/src/promql.rs
Original file line number Diff line number Diff line change
Expand Up @@ -480,15 +480,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),
Expand Down
56 changes: 56 additions & 0 deletions crates/lower/tests/promql_lowering.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 { .. })));
}
Loading