fix(planner): topk over a grouped sum of sum_over_time or count_over_time gets the fused heap

crates/devtools/src/candidate_labels.rs

@@ -86,7 +86,7 @@ pub fn computation_label(candidate: &ComputationCandidate) -> String {
 /// parentheses), each target listed under the first query that reads it.
 pub fn choice_label<Id>(inventory: &PerComputationCandidates<Id>, choice: &[usize]) -> String {
     let query_of_target = query_of_each_target(inventory);
-    // Each target's chosen candidate, with the target it absorbs when the
+    // Each target's chosen candidate, with the targets it absorbs when the
     // candidate is its fused computation's.
     let chosen: Vec<_> = inventory
         .targets
@@ -95,7 +95,12 @@ pub fn choice_label<Id>(inventory: &PerComputationCandidates<Id>, choice: &[usiz
         .map(|(target, &index)| target.computation_for(index).expect("a listed candidate"))
         .collect();
     // Targets a chosen fused computation absorbs are not computed.
-    let absorbed: Vec<usize> = chosen.iter().filter_map(|&(_, absorbs)| absorbs).collect();
+    let absorbed: Vec<usize> = chosen
+        .iter()
+        .filter_map(|&(_, absorbs)| absorbs)
+        .flatten()
+        .copied()
+        .collect();
     (0..inventory.roots.len())
         .map(|query| {
             let mut sketches = Vec::new();

crates/frontend-promql/src/promql.rs

@@ -50,6 +50,7 @@
 //! | `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]}` |
+//! | `topk(k, sum by (G) (count_over_time\|sum_over_time\|rate\|increase\|irate(…)))` | `Aggregate{[TopK{k}]}` over the explicit grouped `Aggregate{[Sum]}` and per-series aggregate |
 //! | `topk(k, <other>)` / `bottomk(k, …)` | `Sort{value} → Limit{k}` |
 //! | `m{f}` / `m{f}[w]` | `TimeRange{ingestion, Instant, Scan{predicates}}` / `TimeRange{w, Range, Scan}` |
 //! | `a OP b` | `BinaryOp{vector_match}` |
@@ -790,26 +791,33 @@ fn build_over_sub_dag(outer: Outer, keys: Vec<ColumnRef>, child: Unresolved) ->
             child: Rc::new(child),
         },
         Outer::TopK { k, descending } => {
-            let weighted_counter_ranking = matches!(
+            // A sum of per-series counter values, or of per-series sums or
+            // counts of raw data samples (`sum by (G) (sum_over_time(m[W]))`): a
+            // weighted ranking of the groups.
+            let weighted_group_ranking = matches!(
                 &child,
                 Unresolved::Aggregate {
                     measures,
                     child: sum_child,
                     ..
                 } if matches!(measures.as_slice(), [AggIntent::Sum { .. }])
-                    && matches!(sum_child.as_ref(), Unresolved::Aggregate { measures, .. }
+                    && matches!(sum_child.as_ref(), Unresolved::Aggregate { measures, reduction, .. }
                         if matches!(
                             measures.as_slice(),
                             [AggIntent::Rate | AggIntent::Increase | AggIntent::IRate]
-                        ))
+                        ) || (matches!(reduction, Reduction::PerEntity)
+                            && matches!(
+                                measures.as_slice(),
+                                [AggIntent::Sum { .. } | AggIntent::Count { .. }]
+                            )))
             );
             let direct_counter_ranking = matches!(&child, Unresolved::Aggregate {
                 measures, reduction: Reduction::PerEntity, ..
             } if matches!(
                 measures.as_slice(),
                 [AggIntent::Rate | AggIntent::Increase | AggIntent::IRate]
             ));
-            if descending && (weighted_counter_ranking || direct_counter_ranking) {
+            if descending && (weighted_group_ranking || direct_counter_ranking) {
                 return Ok(outer_aggregate(
                     keys,
                     AggIntent::TopK {

crates/logical-asap-optimization/src/realization.rs

@@ -305,9 +305,18 @@ pub(crate) fn realize_keyed_additive_summary_input(
         _ => None,
     };
     let counter_input = derivative_of.is_some();
+    // `sum by (G) (sum_over_time(m[W]))` or `... count_over_time ...`: the
+    // grouped sum of per-series sums (counts) of raw data samples is their sum
+    // (count) by `G`, so the heap reads those samples.
+    let samples = summed_per_series_samples(child);
+    // A raw data sample's count or a count's row weighs one.
+    let counted = match &samples {
+        Some((_, inner)) => matches!(inner, AggIntent::Count { .. }),
+        None => matches!(measures.as_slice(), [AggIntent::Count { .. }]),
+    };
     let weight = match measures.as_slice() {
-        [AggIntent::Count { .. }] => SummaryInputExpr::Constant(1.0),
-        [AggIntent::Sum { .. }] if counter_input => {
+        _ if counted => SummaryInputExpr::Constant(1.0),
+        [AggIntent::Sum { .. }] if counter_input || samples.is_some() => {
             SummaryInputExpr::Column(ColumnRef::SampleValue)
         }
         [AggIntent::Sum { col }] => SummaryInputExpr::Column(match col {
@@ -322,14 +331,12 @@ pub(crate) fn realize_keyed_additive_summary_input(
         }),
         _ => return None,
     };
-    let weight_domain = match measures.as_slice() {
-        [AggIntent::Count { .. }] => WeightDomain::NonNegative {
+    let weight_domain = match derivative_of {
+        _ if counted => WeightDomain::NonNegative {
             proof: NonNegativeWeightProof::UnitCount,
         },
-        _ => match derivative_of {
-            Some(Some(metric)) => WeightDomain::CounterDerivativeOfMetric { metric },
-            _ => WeightDomain::UnknownOrSigned,
-        },
+        Some(Some(metric)) => WeightDomain::CounterDerivativeOfMetric { metric },
+        _ => WeightDomain::UnknownOrSigned,
     };
     if matches!(heap_algorithm, SketchAlgorithm::CmsWithHeap)
         && !matches!(
@@ -384,7 +391,7 @@ pub(crate) fn realize_keyed_additive_summary_input(
         }
     };
     Some(PhysicalSummaryInput {
-        child: Rc::clone(raw_child),
+        child: samples.map_or_else(|| Rc::clone(raw_child), |(rows, _)| rows),
         input: SummaryUpdate {
             item: Some(item),
             weight,
@@ -393,6 +400,67 @@ pub(crate) fn realize_keyed_additive_summary_input(
     })
 }
 
+/// The raw data sample rows and the per-series `sum` or `count` over them when
+/// `grouped` is a grouped `sum` (not by `without`, nor by the metric name
+/// the per-series result drops) of the value of a per-series `sum` or
+/// `count` of raw data samples (`sum by (G) (sum_over_time(m[W]))`), neither
+/// filtered nor with `HAVING`; `None` otherwise. The grouped sum is then the
+/// sum (count) of those samples by `G`.
+fn summed_per_series_samples(grouped: &OperatorNode) -> Option<(Rc<OperatorNode>, AggIntent)> {
+    use asap_types::ir::operator::non_asap::any_measure_filtered;
+    let Some(NonASAPOp::Aggregate {
+        reduction: Reduction::Reduce(by),
+        measures,
+        filters,
+        having: None,
+        child: per_series,
+        ..
+    }) = grouped.non_asap()
+    else {
+        return None;
+    };
+    let [AggIntent::Sum { col }] = measures.as_slice() else {
+        return None;
+    };
+    let Some(NonASAPOp::Aggregate {
+        reduction: Reduction::PerEntity,
+        measures: inner,
+        filters: inner_filters,
+        having: None,
+        child: rows,
+        ..
+    }) = per_series.non_asap()
+    else {
+        return None;
+    };
+    let [inner @ (AggIntent::Sum { col: None } | AggIntent::Count { .. })] = inner.as_slice()
+    else {
+        return None;
+    };
+    if by.is_without()
+        || any_measure_filtered(filters)
+        || any_measure_filtered(inner_filters)
+        || matches!(rows.non_asap(), Some(NonASAPOp::Aggregate { .. }))
+    {
+        return None;
+    }
+    // The grouped sum reads the per-series value.
+    let value = resolve_column_ref(&ColumnRef::SampleValue, &per_series.schema).ok()?;
+    if col.is_some_and(|col| col != value) || by.keys().contains(&value) {
+        return None;
+    }
+    if by.keys().iter().any(|&key| {
+        per_series
+            .schema
+            .fields
+            .get(key)
+            .is_none_or(|field| field.name == "__name__")
+    }) {
+        return None;
+    }
+    Some((Rc::clone(rows), inner.clone()))
+}
+
 pub(crate) fn schema_column_ref(child: &OperatorNode, index: usize) -> Option<ColumnRef> {
     let column = child.schema.fields.get(index)?;
     Some(match &column.table {

crates/plan-selection/src/pass/query_candidates.rs

@@ -167,7 +167,7 @@ fn query_digits<Id>(
                 || per_computation.targets[t]
                     .fused
                     .as_ref()
-                    .is_some_and(|fused| read[fused.absorbs])
+                    .is_some_and(|fused| fused.absorbs.iter().any(|&a| read[a]))
         })
         .collect()
 }

crates/planner/tests/runtime_install.rs

@@ -2696,3 +2696,128 @@ async fn a_forced_heap_counter_topk_plan_compiles_and_answers_within_its_bound()
         }
     }
 }
+
+/// `data` samples of 40 regions with 3 series each, inside the last
+/// minute: region `r` (rank 1..=40) sums to `1000 / r^1.1` over
+/// its series (a skewed ranking of the regions) and holds `3 ·
+/// round(200 / r^1.1)` samples (a skewed count ranking), returned with each
+/// region's exact sum and exact count.
+fn skewed_region_raw_data_samples() -> (
+    Vec<RawDataSample>,
+    BTreeMap<String, f64>,
+    BTreeMap<String, f64>,
+) {
+    let mut samples = Vec::new();
+    let mut sums = BTreeMap::new();
+    let mut counts = BTreeMap::new();
+    for rank in 1..=40i64 {
+        let region = format!("r{rank:02}");
+        let sum = 1000.0 / (rank as f64).powf(1.1);
+        let per_series_steps = (200.0 / (rank as f64).powf(1.1)).round().max(1.0) as i64;
+        for series in 0..3 {
+            let instance = format!("{region}-{series}");
+            for step in 0..per_series_steps {
+                // Spread over (EVALUATION_MS - 59 s, EVALUATION_MS - 1 s].
+                let at = EVALUATION_MS - 1_000 - step * 58_000 / per_series_steps;
+                let mut s = sample("data", &instance, at, sum / 3.0 / per_series_steps as f64);
+                s.labels.insert("region".into(), region.clone());
+                samples.push(s);
+            }
+        }
+        sums.insert(region.clone(), sum);
+        counts.insert(region, (3 * per_series_steps) as f64);
+    }
+    samples.sort_by_key(|s| s.timestamp_ms);
+    (samples, sums, counts)
+}
+
+/// The ASAPQuery workloads' grouped top-k-sum shape,
+/// `topk(3, sum by (region) (<f>(data[1m])))` for `f` `sum_over_time` or
+/// `count_over_time`: each forced fused heap (one over the raw data samples,
+/// keyed by `region`, weighted by the sample value or by one) compiles and
+/// answers the true top 3 regions within its bound of their exact sums or
+/// counts, each named by its `region` label alone, as the exact top-k's.
+/// A Count-Min heap over sample values needs the gauge declared
+/// non-negative; a Count Sketch heap takes any gauge.
+#[tokio::test]
+async fn a_forced_heap_grouped_topk_sum_plan_compiles_and_answers_within_its_bound() {
+    let (samples, sums, counts) = skewed_region_raw_data_samples();
+    let gauge = r#""metric_type": "gauge""#;
+    let non_negative = r#""metric_type": "gauge", "values_non_negative": true"#;
+    for (function, truth, facts, algorithm) in [
+        (
+            "sum_over_time",
+            &sums,
+            non_negative,
+            SketchAlgorithm::CmsWithHeap,
+        ),
+        (
+            "sum_over_time",
+            &sums,
+            gauge,
+            SketchAlgorithm::CountSketchWithHeap,
+        ),
+        (
+            "count_over_time",
+            &counts,
+            gauge,
+            SketchAlgorithm::CmsWithHeap,
+        ),
+        (
+            "count_over_time",
+            &counts,
+            gauge,
+            SketchAlgorithm::CountSketchWithHeap,
+        ),
+    ] {
+        let query = format!("topk(3, sum by (region) ({function}(data[1m])))");
+        let workload = heap_topk_workload(&query, "data", facts);
+        let truth = BTreeMap::from([(String::new(), truth.clone())]);
+        let plan = plan_ignoring_guarantees(&workload, Some(algorithm.clone()))
+            .await
+            .unwrap_or_else(|error| panic!("{query} {algorithm:?}: {error}"));
+        let (built, width) = heap_of(&plan).expect("a heap plan");
+        assert_eq!(built, algorithm, "{query}");
+        plan.check_compiles().unwrap();
+        let plan_json = serde_json::to_string(&plan).unwrap();
+        let run = run_data_plane(
+            &plan_json,
+            &samples,
+            EVALUATION_MS,
+            &mut SummaryStore::new(),
+        );
+        for answers in [
+            &run.answers_per_query[0],
+            &run.reference_answers_per_query[0],
+        ] {
+            assert_heap_topk_answers(answers, &truth, (None, "region", 3), (&algorithm, width));
+            for (labels, _) in answers.iter() {
+                assert_eq!(
+                    labels.keys().collect::<Vec<_>>(),
+                    ["region"],
+                    "{query} {algorithm:?}: the group label alone"
+                );
+            }
+        }
+        // The installed plan (stored states) answers as its query-time
+        // reference does, within the heap's bound.
+        let bound = heap_bound(
+            &algorithm,
+            width,
+            &truth[""].values().copied().collect::<Vec<_>>(),
+        );
+        let reference: BTreeMap<_, _> = run.reference_answers_per_query[0]
+            .iter()
+            .map(|(labels, value)| (labels.clone(), *value))
+            .collect();
+        for (labels, value) in &run.answers_per_query[0] {
+            let expected = reference
+                .get(labels)
+                .unwrap_or_else(|| panic!("{query} {algorithm:?}: {labels:?} in the reference"));
+            assert!(
+                (value - expected).abs() <= bound,
+                "{query} {algorithm:?}: {value} vs {expected}"
+            );
+        }
+    }
+}

docs/design_docs/proposals/planner-layering.md

@@ -304,15 +304,17 @@ aggregations are separate targets: in `sum by (job) (rate(x[1m]))`, the
 per-series `rate` and the per-`job` `sum` are each a target, and their options
 multiply.
 
-A computation that fuses a target with the one directly beneath it is the
-outer target's **fused computation**, which absorbs the target beneath: the
+A computation that fuses a target with the targets beneath it is the
+outer target's **fused computation**, which absorbs those targets: the
 whole-expression top-*k* over a per-item `sum` or `count` (one heap summary
 over the inner aggregate's input), and the nested sum, a grouped `sum` over a
 `sum` (one exact Sum accumulator over the inner sum's input). Its options are
 offered beside the unfused pair's: in `topk by (job) (10, sum_over_time(m[1m]))`
 the options are the exact top-*k* over the exact per-series sum, or the fused
-computation's Count-Min or Count Sketch with a top-*k* heap. Like every
-target's options, a fused computation's are exact or a summary.
+computation's Count-Min or Count Sketch with a top-*k* heap. In
+`topk(10, sum by (job) (sum_over_time(m[1m])))` the heap reads the raw data samples
+beneath the grouped sum, so it absorbs both the grouped sum and the per-series
+sum. Like every target's options, a fused computation's are exact or a summary.
 
 Example for summary candidates:
 

docs/develop_docs/asap-aware-mapping-architecture.md

@@ -80,7 +80,7 @@ populations for instant top-k, quantile and sum) in Stage 1, and
    one exact option plus the summary families of `summary_candidates` that the
    accuracy target permits and the deployment supports. A whole-expression
    top-k or a nested sum also gets a fused computation that absorbs the
-   target beneath it.
+   targets beneath it.
 2. **Enumeration** takes one choice per target and composes the logical DAG.
    For each composed DAG, the Pass 2 rules find what can be shared, split into
    panes or reused, and each subset of their candidates becomes another

docs/develop_docs/asap-aware-mapping-contracts.md

@@ -46,10 +46,12 @@ asap-physical-asap-optimization -p asap-plan-selection -p asap-types
   dropped. Pass 1 prunes only what it can prove; accuracy that depends on
   data facts is left to Stage 3.
 - **Fused computations.** `FusedComputation { absorbs, candidates, pruned }`
-  replaces the target directly beneath (the whole-expression top-k, the
-  nested sum). A whole-workload choice keeps one digit per target; a fused
-  pick with a non-first digit on the absorbed target is skipped, so each
-  plan is enumerated once.
+  replaces the targets beneath (the whole-expression top-k, the nested sum).
+  It absorbs the target directly beneath and, for a heap over the raw data samples
+  beneath a grouped sum of per-series sums or counts, the per-series target
+  too. A whole-workload choice keeps one digit per target; a fused pick with
+  a non-first digit on an absorbed target is skipped, so each plan is
+  enumerated once.
 - **Pass 2.** Each rule adds candidates; none removes a Pass 1 candidate, adds
   a summary family, or adds a readout. So a deployment that does not support a
   family never sees it again after Pass 1. A Pass 2 candidate records what it