[zzylol]

Why

topk(k, sum by (G) (sum_over_time(m[W]))) ranks groups G by the sum of their raw data samples, and the count_over_time form ranks them by their sample count. A fused Count-Min or Count Sketch heap answers either one in a single pass over the samples: each sample adds its value (or 1) to its key G. Pass 1 offered no heap candidate for these queries, only the exact sort and limit (#832). After #815 and #831, only the series-ranking form topk by (G) (k, sum_over_time(m[W])) and the counter forms (sum by (G) (increase|rate(...))) got the fused heap.

What

• Pass 1 now gives topk(k, sum by (G) (sum_over_time(m[W]))) and topk(k, sum by (G) (count_over_time(m[W]))) a TopK target. The target keeps Exact and gets a fused CmsWithHeap and CountSketchWithHeap. This also works with an outer topk by (P), which partitions the heaps.
• Each heap reads the range selector's raw data samples, keyed by the G labels other than the outer by labels:
  • for a sum, it is weighted by the sample value. The weight carries the open RawDataSamplesOfMetric requirement, so Count-Min is admitted only over a gauge declared values_non_negative (fix(plan-selection): prove Count-Min weights over a gauge declared non-negative #815). Count Sketch takes any gauge.
  • for a count, it is weighted by 1, which is proven non-negative.
• The answer has the exact top-k's columns: the G labels and the value, without __name__.
• The exact top-k is still offered, and the grouped sum keeps its nested-sum merge (one exact accumulator per G over the samples). The selected exact plan is the same plan as before, plus one final Project. Measured with the converter: 1.24077e-8 before, 1.24078e-8 after.
• The counter forms (increase, rate) were already covered and are unchanged. bottomk still lowers to Sort + Limit.

How

• Frontend. build_over_sub_dag promotes a descending topk over sum to AggIntent::TopK when the summed child is a per-series sum or count, just as it does for rate, increase and irate.
• Keyed-additive rule. The new summed_per_series_samples recognises a grouped sum (not without, not by __name__, with no filter and no HAVING) of the value of a per-series sum or count of raw rows. realize_keyed_additive_summary_input then builds the heap over those rows instead of over the per-series results: a grouped sum of per-series sums is the sum of the samples by group.
• Fusion absorbs a list. FusedComputation.absorbs is now Vec<usize>. The heap absorbs the grouped sum and, when the grouped sum is the per-series target's only consumer, the per-series target too. Choices of the absorbed targets therefore do not multiply the heap candidates.
  • combinations counts n · Π m + f recursively, and it still equals the number of choices enumeration keeps.
  • absorbed_choice_varies checks every absorbed digit.
  • After the nested-sum merges are added, close_absorptions adds to each fused computation the targets its absorbed targets' own merges absorb. None of them is computed when the fused computation is picked, so the count stays exact for any nesting.
  • A target that a whole-expression heap absorbs may still get its own nested-sum merge, which serves the top-k's exact candidate.
• The fix(planner): Stage 3 rejects plans the executor cannot compile; no PromQL top-k heap over its instant vector #805 prune (TopKHeapOverOneRowPerItem) was not the cause, and it is unchanged. It runs after the fused heaps have moved out of the top-k's own candidates.

Key code interfaces added / modified

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

/// Stays: a target's measure computed together with the targets beneath it, which it absorbs.
pub struct FusedComputation {
    /// The targets it absorbs: the one directly beneath, then, for a whole-expression heap over
    /// the raw data samples beneath a grouped sum of per-series sums or counts, the per-series target.
    /// Then the targets any of these targets' own fused computations absorb.
    pub absorbs: Vec<usize>, // was `usize`
    /// How the fused computation is computed: exact or a summary.
    pub candidates: Vec<ComputationCandidate>, // unchanged
    /// The candidates Pass 1 pruned from it, with their reasons.
    pub pruned: Vec<(ComputationCandidate, PruneReason)>, // unchanged
}

impl ComputationTarget {
    /// The candidate for digit `index`, and the targets it absorbs when it is the fused
    /// computation's.
    pub fn computation_for(
        &self,
        index: usize,
    ) -> Option<(&ComputationCandidate, Option<&[usize]>)>; // was Option<usize>
}

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

/// New (private): the raw data sample rows and the per-series `sum` or `count` over them when
/// `grouped` is a grouped `sum` of a per-series `sum` or `count` of raw data samples; `None` otherwise.
fn summed_per_series_samples(grouped: &OperatorNode) -> Option<(Rc<OperatorNode>, AggIntent)>;

Usage

This walks through a_forced_heap_grouped_topk_sum_plan_compiles_and_answers_within_its_bound (crates/planner/tests/runtime_install.rs), run at the linked commit:

// The user's query: the top 3 regions by the 1-minute sum of their samples, within 0.05 at
// failure probability 0.3, every 10 s, over 10,000 `data` gauge series (declared non-negative)
// sampled every 100 ms.
let workload = heap_topk_workload(
    "topk(3, sum by (region) (sum_over_time(data[1m])))",
    "data",
    r#""metric_type": "gauge", "values_non_negative": true"#,
);
// Stage 0 lowers it to Aggregate{[TopK{3}]} over Aggregate{[Sum], by (region)} over the
// per-series Aggregate{[Sum]} over TimeRange{1m}. Pass 1 gives the TopK target `Exact`, plus a
// fused CmsWithHeap and CountSketchWithHeap that absorb both sums: one heap over the TimeRange
// rows, item `region`, weight the sample value (RawDataSamplesOfMetric{"data"}, confirmed by the
// declaration). Guarantees are ignored (a top-k estimate has none), no raw data samples are kept,
// and Stage 3 keeps only CmsWithHeap plans that the executor compiles (HeapPlansOnly).
let plan = plan_ignoring_guarantees(&workload, Some(SketchAlgorithm::CmsWithHeap)).await?;
// The heap the plan builds, and its width (for the ε·L1 bound).
let (built, width) = heap_of(&plan).expect("a heap plan"); // (CmsWithHeap, width)
plan.check_compiles()?;
// The stand-in runtime feeds 40 regions × 3 series of skewed samples, builds the stored states,
// and answers at EVALUATION_MS; it also runs the plan's query_time_reference over raw data samples.
let run = run_data_plane(&serde_json::to_string(&plan)?, &samples, EVALUATION_MS, &mut SummaryStore::new());
// Each answer: labels {region} only (no __name__) and the estimated sum. The true top 3
// (r01, r02, r03) are recalled, each within e/w · L1 of its exact sum (Count-Min never
// underestimates), and within that bound of the reference's answer.

The test also runs the other three combinations (Count Sketch over an undeclared gauge for the sum, and both heaps for count_over_time, weighted by 1).

Before this PR

topk(3, sum by (region) (sum_over_time(data[1m]))) was lowered to Limit{3}(Sort{value desc}(...)), so it had no top-k target. The forced Count-Min plan above failed:

plan selection: no valid candidate: [P1-m1 ExecutorCannotCompile "this deployment runs only CmsWithHeap top-k plans",
                                     P1 RawDataNotRetained "scans raw data at query time, ..."]

After this PR

The query has a top-k target with Exact and the two fused heaps over the raw data samples. Plan selection prices them beside the exact plan. Exhaustive stage_pipeline on the variant workload below (q07 rewritten as topk by (region, service) (4, sum by (region, service, job) (sum_over_time(data[20m]))), 10,000 series, guarantees ignored, no raw data) gave these costs in the built-in model:

Candidate	Cost/s	Note
P4-m1 exact, nested-sum merge (100 groups)	1.129	selected
P6-m1 CmsWithHeap	1.997	new
P2-m1 exact, per-series sums (10,000 groups)	3.984
P8-m1 CountSketchWithHeap	16.41	new

With only 100 groups, the exact plan stays the cheapest. The heaps win only when there are many more groups.

Verification

• New Pass 1 test whole_expression_topk_over_a_grouped_sum_of_range_sums_reads_the_raw_data_samples covers sum, count, and partitioned sum. It checks that the target keeps Exact and gets the two fused heaps, which absorb both aggregates. Each heap reads the TimeRange with item region, the right weight and weight domain, and the outer partitions, and combinations equals the enumerated choices. It failed before the fix (no top-k target).
• New Pass 1 tests grouped_range_sums_a_raw_data_samples_heap_cannot_rank_get_no_fused_heap (bottomk keeps Sort + Limit; sum without and sum by (__name__) get no heap over the raw data samples) and a_fused_computation_absorbs_what_its_absorbed_targets_merges_absorb (the closure keeps combinations equal to the enumerated choices).
• New runtime test a_forced_heap_grouped_topk_sum_plan_compiles_and_answers_within_its_bound covers Count-Min and Count Sketch for sum, and both for count. In each case the forced heap plan compiles and recalls the true top 3 on skewed data. Its estimates are within the ε bound of the exact values and of query_time_reference, and its answers carry region only. It failed before the fix (no valid candidate).
• cargo test --workspace --locked, cargo test -p asap-planner --no-default-features --features promql and cargo fmt --check pass locally. I did not wait for CI.
• Real workload (seed-862 workload-01, local converter asapquery_random_workload rebuilt on this branch, --no-raw --failure-probability 0.3 --check, three evaluations):

The rows marked variant use q07 rewritten to the new shape, topk by (region, service) (4, sum by (region, service, job) (sum_over_time(data[20m]))). Recall and max relative error are taken over the three evaluations.

Manifest	Queries	Flags	Selected	Cost/s	Recall	Max rel. error
as-is	q07		P1-m1 exact	3.716e-8	1.0	0
as-is	q07	--ignore-accuracy-guarantees	P2-m1 CmsWithHeap (w 55, d 2, heap 20)	1.923e-8	0.9875	0.0255
as-is	q06+q07		P1-m3 exact, exact	9.467e-8	1.0 / 1.0	0 / 0
as-is	q06+q07	--ignore-accuracy-guarantees	P9-m3 CmsWithHeap, CmsWithHeap	4.167e-8	q06 0.975–0.9875, q07 0.9875	q06 0.0237, q07 0.0255
variant	q07		P2-m1 exact, nested sum	1.263e-8	1.0	0
variant	q07	--ignore-accuracy-guarantees	P2-m1 exact, nested sum	1.263e-8	1.0	0
variant	q06+q07		P2-m3 exact, exact	7.015e-8	1.0 / 1.0	0 / 0
variant	q06+q07	--ignore-accuracy-guarantees	P6-m3 q06 CmsWithHeap, q07 exact	3.507e-8	q06 0.975–0.9875, q07 1.0	q06 0.0237, q07 0

Every check passes. The as-is q07 is the series-ranking form, which this PR does not change: its heap comes from #815 and #831. Without --ignore-accuracy-guarantees, no heap is selected, because a top-k estimate has no guarantee. For the variant, the new heaps are offered and priced, but the exact nested-sum plan over 100 groups is cheaper (see the table above).

Limitations: bottomk, other range functions, and SQL nested aggregates are out of scope.

Closes #832