refactor(planner)!: Pass 1 candidates are exact or a summary; fused computations become the outer target's FusedComputation

[zzylol]

Why

ComputationCandidate mixed two questions: which computation a Pass 1 target computes and how it is computed. WholeExpression and NestedSumMerge really computed a different, fused computation (the outer aggregate together with the one beneath it), and AvgAsSumCount was just two exact states. The design doc's Pass 1 table offers each target "exact or a summary"; the type should say the same.

What

• ComputationCandidate now only says how a target is computed: Exact, ExactAccumulator(Vec<ExactKind>) or Summary(SketchAlgorithm).
• avg is ExactAccumulator([Sum, Count]). Composition refuses any other order.
• The whole-expression top-k and the nested-sum merge are the outer target's fused computation: fused: Option<FusedComputation { absorbs, candidates }>, which absorbs the target beneath.
  • The whole-expression top-k's candidates are Summary(CmsWithHeap) and Summary(CountSketchWithHeap).
  • The nested-sum merge's candidate is ExactAccumulator([Sum]).
• Removed: WholeExpression, NestedSumMerge, AvgAsSumCount and ComputationCandidate::absorbs_inner. ComputationTarget::absorbs is replaced by FusedComputation::absorbs.
• ComputationTarget::target is renamed aggregate.
• ExactAccumulator holds only the accumulator kinds, with no params; ExactParams itself stays (see How).
• Behaviour does not change. Enumeration order, candidate ids, counts, plans, costs, rejections and labels are byte-identical (see Verification).
• devtools candidate_labels reads the new shape. The design doc's Pass 1 section describes fused computations. No viewer change: the viewer shows only display_name, never candidate kinds.

How

Fused-computation representation. A target with a fusion carries fused: Some(FusedComputation { absorbs, candidates }). absorbs is the index in targets of the inner target it replaces, and candidates says how the fused computation is computed. A FusedComputation holds no aggregate, measure, filter or HAVING; it is always the outer target's measure. So it cannot be in impossible states (a fused computation with its own fusion, or one with no absorbed target), and no target is cloned.

A whole-workload choice keeps one digit per target. A digit ranges over the target's own candidates and then its fused computation's (ComputationTarget::choice_count). ComputationTarget::computation_for(index) returns the chosen candidate together with Some(absorbed target) when it is the fused computation's.

A fused group contributes the unfused pair (outer × inner) plus the fused candidates to the Cartesian product, so combinations = n·m + f. The enumeration is unchanged: absorbed_choice_varies still skips a fused pick with a non-first inner digit.

Alternatives considered:

• A fused computation as a flat peer in targets. It needs a "not computed" digit (Vec<Option<usize>> choices) everywhere, and a group-aware enumerator to keep the candidate order and ids.
• A boxed ComputationTarget as the fused alternative. This was the previous revision of this PR. It allowed impossible states and cloned the target.

Composition. compose_logical_candidate groups each target's computation_for result by target.aggregate, as a private ChosenComputation { target, candidate, absorbs }. In substitute, the aggregate, measure and samples_of come from the outer target. A candidate with absorbs: Some(_) is fused:

• Summary(heap) builds the whole-expression heap over the absorbed input.
• ExactAccumulator builds the nested Sum accumulator.

Otherwise, a one-kind ExactAccumulator is finalized, and [Sum, Count] of an avg is the sum/count division, as before. Any other list is Unsupported.

Why ExactParams stays. Composition builds FieldDataType::ExactAggregate(kind, params) from the kind (a private exact_family(kind)). ExactParams carries no information: the kernels and capability.rs only check that it matches kind. But it is part of a serialized format. FieldDataType is Serialize/Deserialize, so every exact SummaryAgg in the IR JSON (the stage_pipeline document) is written as {"ExactAggregate": ["Sum", "Sum"]}. The stage viewer (tools/dag-viewer/stages.js, covered by test_viewer.py) reads that array. Dropping the second element changes the wire format for the viewer and saved documents, so it belongs in its own format change.

Where fused computations are created. add_whole_expression_targets moves the top-k's heap Summary candidates (its last candidates) into the fused computation, so the top-k keeps [Exact]. add_nested_sum_targets adds [ExactAccumulator([Sum])]. The eligibility checks are unchanged.

Key code interfaces added / modified

crates/asap-aware-mapping/src/per_computation_candidates.rs

/// How one Pass 1 target is computed: exact or a summary. The type stays; three variants are removed.
pub enum ComputationCandidate {
    /// Plain exact execution of the original sub-DAG; the exact option only when the
    /// intent has no exact accumulator form or that form is pruned.
    Exact,                                          // unchanged
    /// One or more exact mergeable accumulator states, by kind, finalized and combined into
    /// the measure: one for sum/count/min/max/rate/irate/increase; [Sum, Count] for avg.
    ExactAccumulator(Vec<ExactKind>),               // was ExactAccumulator { kind, params }
    /// An unsized summary family, sized later for the measure's accuracy.
    Summary(SketchAlgorithm),                       // unchanged
    // removed: AvgAsSumCount      → ExactAccumulator([Sum, Count])
    // removed: WholeExpression(a) → Summary(a) in a FusedComputation
    // removed: NestedSumMerge     → ExactAccumulator([Sum]) in a FusedComputation
}
// removed: ComputationCandidate::absorbs_inner(&self) -> bool  (FusedComputation::absorbs says it)

/// One measure of an aggregate with a realization besides exact execution. The type
/// stays; one field is renamed, one replaced and one added.
pub struct ComputationTarget {
    /// The `Aggregate` node the measure belongs to.
    pub aggregate: Rc<OperatorNode>,           // was target
    /// Index into the aggregate's measures.
    pub measure: usize,                        // unchanged
    /// The measure's unranked candidates, the exact option first, and the pruned ones with reasons.
    pub choices: ComputationChoices,           // unchanged (doc added)
    /// The measure's own filter (`SummaryAgg.filter`); None when unfiltered.
    pub filter: Option<Predicate>,             // unchanged
    /// The aggregate's own `having`, then one entry per consumer `Filter` above it.
    pub having: Vec<HavingEntry>,              // unchanged
    /// The fused computation of this measure with the target beneath, offered beside the
    /// target's own candidates; None when no fusion applies.
    pub fused: Option<FusedComputation>,       // new
    // removed: absorbs: Option<usize>         (now FusedComputation::absorbs)
    /// The metric whose raw samples (or plain sums) the aggregate reads; None otherwise.
    pub samples_of: Option<String>,            // unchanged
}

/// A target's measure computed together with the target directly beneath, which it
/// absorbs: the whole-expression top-k (Summary of a heap family) or the nested sum
/// (ExactAccumulator([Sum])). It has the outer target's aggregate, measure, filter, HAVING. New.
pub struct FusedComputation {
    /// The index in `PerComputationCandidates::targets` of the inner target it absorbs.
    pub absorbs: usize,                          // new
    /// How the fused computation is computed: exact or a summary.
    pub candidates: Vec<ComputationCandidate>,   // new
}

impl ComputationTarget {
    /// How many values this target's choice digit takes: own candidates, then the fused computation's.
    pub fn choice_count(&self) -> usize;   // new
    /// For digit `index`: the candidate, and the target it absorbs when the candidate is the
    /// fused computation's (past the own candidates); None past `choice_count`. Used by
    /// composition, enumeration and devtools labels.
    pub fn computation_for(&self, index: usize) -> Option<(&ComputationCandidate, Option<usize>)>;   // new
}

/// Compact Pass 1 inventory. The type stays.
pub struct PerComputationCandidates<Id> {
    /// The frontends' query roots with their ids, in workload order.
    pub roots: Vec<(Id, QueryRoot)>,           // unchanged (doc added)
    /// The targets; a choice has one digit per target over `choice_count`.
    pub targets: Vec<ComputationTarget>,       // unchanged
}
impl<Id> PerComputationCandidates<Id> {
    /// Whole-workload candidate count: the product, a fused group counting n·m + f.
    pub fn combinations(&self) -> usize;                            // unchanged signature
    /// Whether `choice` picks a fused computation and a non-first candidate of the target it absorbs.
    pub fn absorbed_choice_varies(&self, choice: &[usize]) -> bool; // unchanged signature
}

/// Build one whole-workload candidate; `choice[i]` picks `targets[i]`'s candidate or,
/// past them, its fused computation's. Returns `Unsupported` (an internal bug) for an
/// `ExactAccumulator` list it cannot finalize, e.g. avg states not in [Sum, Count] order.
pub fn compose_logical_candidate<Id: Clone>(
    inventory: &PerComputationCandidates<Id>,
    choice: &[usize],
) -> Result<Vec<(Id, QueryRoot)>, PerComputationCandidateError>;   // unchanged signature

crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs: the signatures are unchanged. LogicalCandidate::choice and Stage1Candidate::choice now index targets[t].computation_for(choice[t]). next_choice counts digits with choice_count.

crates/devtools/src/candidate_labels.rs: choice_label keeps its signature and its output. It labels ExactAccumulator as "Sum acc" or "Sum/Count acc", and as "nested Sum acc" when computation_for returns an absorbed target.

Usage, end to end. This was run at the head commit as a test through the devtools pipeline that stage_pipeline uses; every assert passed:

// 1. What the user writes: one exact PromQL panel, the per-job sum of each series'
//    5-minute sums, refreshed every 10 s.
let panel = RepeatingEntry {
    query: Query("sum by (job) (sum_over_time(m[5m]))".into()),
    demand: RepeatedDemand::FixedInterval(RepetitionInterval(10_000)),
    requirements: QueryRequirements { accuracy: AccuracyTarget::exact(), max_latency_ms: None },
    predictability: Predictability::Predictable { known_at: None },
};
// `workload`: the panel plus the example data facts (continuously ingesting);
// every stage reads it.
let workload = PlanningWorkload {
    query_workload: QueryWorkload {
        language: QueryLanguage::PromQL,
        query_batch: None,
        repeating_queries: Some(vec![panel]),
    },
    data_workload: Some(shared_data_workload(DataArrival::ContinuouslyIngesting)),
};
// `roots`: Stage 0's logical DAG, an outer per-job Sum over an inner per-series Sum.
let roots = lower_promql_roots(&workload).unwrap();
// `run`: Stages 1-3; steps 2-4 read its Pass 1 inventory, candidates and selection.
let run = run_planning_stages(&workload, roots, None).unwrap();

// 2. Stage 1, Pass 1. `inventory`: the per-computation candidates the choices index.
let inventory = &run.stage1_pass1_per_computation;
// `outer`: targets[0], the per-job sum; generate_per_computation_candidates gave it
// a fused nested sum.
let outer = &inventory.targets[0];
// `sum`: one exact Sum accumulator, the outer, inner and fused candidate alike.
let sum = ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum]);
assert_eq!(outer.choices.candidates, [sum.clone()]);
assert_eq!(inventory.targets[1].choices.candidates, [sum.clone()]); // the per-series sum
// `fused`: the outer target's fused computation, absorbing targets[1].
let fused = outer.fused.as_ref().unwrap();
assert_eq!((fused.absorbs, &fused.candidates[..]), (1, &[sum.clone()][..]));
// The outer digit takes 2 values (own + fused): 1·1 + 1 = 2 Pass 1 candidates.
assert_eq!(outer.choice_count(), 2);
assert_eq!(inventory.combinations(), 2);
// Digit 0 is the outer sum's own accumulator; digit 1 the fused one, absorbing target 1.
assert_eq!(outer.computation_for(0), Some((&sum, None)));
assert_eq!(outer.computation_for(1), Some((&sum, Some(1))));

// 3. Stage 1 enumeration: choice [0, 0] is the two separate sums; [1, 0] composes one
//    Sum accumulator over the raw samples, grouped by job. With Pass 2's pane candidates:
//    L1 [0,0] "Q1 exact (Sum acc, Sum acc)"
//    L2 [0,0] "Q1 exact (Sum acc, Sum acc) · Sum acc for Q1 in 30 × 10s panes"
//    L3 [1,0] "Q1 exact (nested Sum acc)"
//    L4 [1,0] "Q1 exact (nested Sum acc) · Sum acc for Q1 in 30 × 10s panes"
// `labels`: devtools' display names by candidate id, as stage_pipeline writes them.
let labels = candidate_labels(&run);
assert_eq!(labels["L3"], "Q1 exact (nested Sum acc)");

// 4. What the user gets back. `selection`: Stage 3's result. The fused nested sum in
//    panes, materialized at ingestion (P4-m1, 0.523 cost/s), beats the separate sums
//    (P1, 15.80 cost/s) and the unmaterialized merge (P4, 15.60 cost/s).
let selection = run.stage3_plan_selection.as_ref().unwrap();
assert_eq!(selection.selected, "P4-m1");

Before this PR

For the panel above, targets[outer].choices.candidates was [ExactAccumulator { kind: Sum, params: Sum }, NestedSumMerge], with absorbs: Some(1) on the outer target. The top-k in topk by (job) (10, sum_over_time(m[1m])) had [Exact, WholeExpression(CmsWithHeap), WholeExpression(CountSketchWithHeap)], and avg had [AvgAsSumCount]. The candidate enum carried both "which computation" and "how".

After this PR

The outer sum has [ExactAccumulator([Sum])] and fused: Some(FusedComputation { absorbs: 1, candidates: [ExactAccumulator([Sum])] }). The top-k has [Exact] and a fused computation with [Summary(CmsWithHeap), Summary(CountSketchWithHeap)]. avg has [ExactAccumulator([Sum, Count])]. Choices, candidate ids and every plan are the same.

Verification

• At head 56f7614: cargo nextest run -p asap-aware-mapping -p asap-devtools -p asap-planner -p asap-frontend-sql -p asap-integration-tests gives 1117 passed, 29 skipped.
• cargo fmt --all -- --check and cargo clippy --workspace --all-targets --all-features --locked -- -D warnings: clean.
• New regression test avg_states_out_of_order_are_refused. With ExactAccumulator([Count, Sum]) on an avg target it failed before the order check (composition divided count by sum). It passes after.
• New test fused_computations_and_avg_states_have_their_own_labels: the nested sum is labeled apart from the separate sums, and an avg is labeled "Sum/Count acc".
• Before/after behaviour check at base 262a43e and at this head, comparing all 24 outputs byte for byte (logical ids and choices, compose errors, physical ids, Stage 2 prunes, Stage 3 selected/costs/accepted/rejected/plan, and labels):
  • Examples 1, 2, 3a, 3b and 4 (Pattern A, monthly predictable), through run_planning_stages. Examples 1, 2, 3a and 3b also through stage_pipeline --max-displayed 100000. Logical candidates: 27, 82, 972, 5 and 12,500. Labels are distinct in each.
  • stage_pipeline exact and ε=0.01/δ=0.001 runs of: topk by (job) (10, sum_over_time(m[1m])), topk by (job) (10, count_over_time(m[1m])) (whole-expression top-k), sum by (job) (sum_over_time(m[5m])) and sum by (job, __name__) (…) (nested sum), avg by (job) (m), avg_over_time(m[5m]), sum by (job) (rate(x[1m])) and topk by (job) (10, rate(x[1m])). Also SQL AVG/SUM/COUNT over a non-null column, and AVG over a nullable one (pruned to Exact).
  • All outputs are identical, including every display_name.
• Tests that enumerate every choice count digits with choice_count(), so they still cover the fused candidates.
• Review: an independent review of 59fe7a8 was clean. This revision applies the simplifications chosen from it.

Limitations and follow-ups

• ExactParams stays because of the serialized IR format (see How). Removing it from FieldDataType::ExactAggregate is a separate wire-format change that would touch the viewer and the saved example documents.