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

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

@@ -2048,7 +2048,7 @@ mod tests {
     fn exact_accumulators_that_merge_get_panes() {
         use crate::per_computation_candidates::ComputationCandidate;
         let accumulator =
-            |c: &ComputationCandidate| matches!(c, ComputationCandidate::ExactAccumulator { .. });
+            |c: &ComputationCandidate| matches!(c, ComputationCandidate::ExactAccumulator(_));
         for query in [
             "rate(m[5m])",
             "increase(m[5m])",
@@ -2153,9 +2153,7 @@ mod tests {
     fn exact(queries: &[&str]) -> Vec<QueryRoot> {
         use crate::per_computation_candidates::ComputationCandidate as C;
         let queries: Vec<(&str, f64)> = queries.iter().map(|q| (*q, 0.01)).collect();
-        chose_where(&queries, |c| {
-            matches!(c, C::Exact | C::ExactAccumulator { .. })
-        })
+        chose_where(&queries, |c| matches!(c, C::Exact | C::ExactAccumulator(_)))
     }
 
     fn operator(root: &QueryRoot) -> &Rc<OperatorNode> {
@@ -2415,17 +2413,16 @@ mod tests {
             .collect(),
         )
         .unwrap();
-        let heap = |c: &C| {
-            matches!(
-                c,
-                C::WholeExpression(SketchAlgorithm::CountSketchWithHeap)
-                    | C::Summary(SketchAlgorithm::CountSketchWithHeap)
-            )
-        };
+        // The CountSketch heap, the top-k's fused whole-expression one.
+        let heap = |c: &C| matches!(c, C::Summary(SketchAlgorithm::CountSketchWithHeap));
         let choice: Vec<usize> = inventory
             .targets
             .iter()
-            .map(|t| t.choices.candidates.iter().position(heap).unwrap_or(0))
+            .map(|t| {
+                (0..t.choice_count())
+                    .position(|i| t.computation_for(i).is_some_and(|(c, _)| heap(c)))
+                    .unwrap_or(0)
+            })
             .collect();
         let heaps: Vec<QueryRoot> = compose_logical_candidate(&inventory, &choice)
             .unwrap()

crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs

@@ -65,8 +65,9 @@ use crate::stage2_physical_asap_optimization::{optimize_physical_asap, PhysicalC
 use crate::stage3_plan_selection::{select_plan, Selection, SelectionError};
 use crate::PlanningModels;
 
-/// One whole-workload CandidateLogicalASAPDAG: `choice[t]` indexes
-/// `targets[t].choices.candidates` of the Pass 1 inventory. `roots` are the
+/// One whole-workload CandidateLogicalASAPDAG: `choice[t]` picks target
+/// `t`'s candidate, or its fused computation's
+/// (`targets[t].computation_for(choice[t])`), of the Pass 1 inventory. `roots` are the
 /// composed query roots in input order, or why the choice did not compose.
 #[derive(Debug)]
 pub struct LogicalCandidate {
@@ -185,8 +186,9 @@ pub fn run_stage1_to_stage3<Id: Clone + std::fmt::Debug>(
 /// One Stage 1 candidate in enumeration order, before Stage 2 numbers it.
 #[derive(Debug)]
 pub struct Stage1Candidate {
-    /// The Pass 1 choice (`targets[t].choices.candidates[choice[t]]`); a
-    /// Pass 2 candidate has its Pass 1 candidate's.
+    /// The Pass 1 choice (`targets[t].computation_for(choice[t])`: target
+    /// `t`'s candidate, or its fused computation's); a Pass 2 candidate has
+    /// its Pass 1 candidate's.
     pub choice: Vec<usize>,
     pub roots: Result<Vec<QueryRoot>, PerComputationCandidateError>,
     /// For a Pass 2 candidate: what it changed of its Pass 1 candidate.
@@ -576,12 +578,13 @@ pub fn run_stage2_and_stage3<Id: Clone + std::fmt::Debug>(
     }
 }
 
-/// Mixed-radix increment of `choice`: the last target varies fastest.
+/// Mixed-radix increment of `choice`, each digit over its target's
+/// candidates and then its fused computation's: the last target varies fastest.
 /// `false` when a carry out of the first target ends the enumeration.
 fn next_choice<Id>(inventory: &PerComputationCandidates<Id>, choice: &mut [usize]) -> bool {
     for (digit, target) in choice.iter_mut().zip(&inventory.targets).rev() {
         *digit += 1;
-        if *digit < target.choices.candidates.len() {
+        if *digit < target.choice_count() {
             return true;
         }
         *digit = 0;

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

@@ -896,7 +896,7 @@ mod tests {
         let choice: Vec<_> = inventory
             .targets
             .iter()
-            .map(|t| t.choices.candidates.len() - 1)
+            .map(|t| t.choice_count() - 1)
             .collect();
         let roots: Vec<_> =
             crate::per_computation_candidates::compose_logical_candidate(&inventory, &choice)
@@ -939,7 +939,7 @@ mod tests {
         let choice: Vec<_> = inventory
             .targets
             .iter()
-            .map(|t| t.choices.candidates.len() - 1)
+            .map(|t| t.choice_count() - 1)
             .collect();
         crate::per_computation_candidates::compose_logical_candidate(&inventory, &choice)
             .unwrap()

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

@@ -2633,7 +2633,7 @@ mod tests {
         let topk = inventory
             .targets
             .iter()
-            .position(|t| t.choices.candidates.len() > 2)
+            .position(|t| t.choice_count() > 2)
             .unwrap();
         [0, 1, 2]
             .into_iter()
@@ -3073,7 +3073,7 @@ mod tests {
             choices = choices
                 .into_iter()
                 .flat_map(|choice: Vec<usize>| {
-                    (0..target.choices.candidates.len()).map(move |index| {
+                    (0..target.choice_count()).map(move |index| {
                         let mut choice = choice.clone();
                         choice.push(index);
                         choice

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

@@ -9,8 +9,7 @@ use asap_types::{
         ASAPOp, ExprSemantics, NonASAPOp, Operator, OperatorNode, Predicate, QueryRoot, ScalarExpr,
     },
     post_asap::{
-        ExactKind, ExactParams, GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams,
-        SummaryUpdate,
+        ExactKind, GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams, SummaryUpdate,
     },
     pre_asap::{
         AggIntent, ColumnRef, CompareOpKind, DataType, Field, FieldDataType, ScalarValue, Schema,
@@ -172,7 +171,7 @@ fn scalar_root_producers_are_discovered_once() {
         root.validate_structure().unwrap();
     }
     assert_eq!(candidates.targets.len(), 1);
-    assert!(Rc::ptr_eq(&candidates.targets[0].target, &producer));
+    assert!(Rc::ptr_eq(&candidates.targets[0].aggregate, &producer));
     assert!(producer.timing.is_none());
     assert!(producer.guarantee.is_none());
     producer.validate_structure().unwrap();
@@ -225,20 +224,16 @@ fn example_1_exact_q1_keeps_exact_accumulators_only() {
     assert_eq!(
         found,
         vec![
-            vec![ComputationCandidate::ExactAccumulator {
-                kind: ExactKind::Sum,
-                params: ExactParams::Sum
-            }],
-            vec![ComputationCandidate::ExactAccumulator {
-                kind: ExactKind::Rate,
-                params: ExactParams::Rate
-            }],
+            vec![ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum])],
+            vec![ComputationCandidate::ExactAccumulator(vec![
+                ExactKind::Rate
+            ])],
         ]
     );
     assert!(candidates
         .targets
         .iter()
-        .all(|target| target.target.timing.is_none()));
+        .all(|target| target.aggregate.timing.is_none()));
 }
 
 /// Physical annotations, ASAP operators and invalid probability requirements
@@ -411,10 +406,7 @@ fn multi_measure_aggregate_has_one_target_per_measure() {
     );
     assert_eq!(
         candidates[0].choices.candidates,
-        vec![ComputationCandidate::ExactAccumulator {
-            kind: ExactKind::Sum,
-            params: ExactParams::Sum
-        }]
+        vec![ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum])]
     );
     assert_eq!(
         algorithms(&candidates[1].choices.candidates),
@@ -510,7 +502,7 @@ fn key_only_having_is_pushed_below_and_keeps_candidates() {
         .unwrap()
         .targets;
     assert_eq!(targets.len(), 1);
-    assert!(Rc::ptr_eq(&targets[0].target, &node));
+    assert!(Rc::ptr_eq(&targets[0].aggregate, &node));
     assert_eq!(
         targets[0].having,
         vec![HavingEntry {
@@ -575,7 +567,10 @@ fn avg_rewrites_to_sum_and_count_states() {
     .targets;
     assert_eq!(
         targets[0].choices.candidates,
-        vec![ComputationCandidate::AvgAsSumCount]
+        vec![ComputationCandidate::ExactAccumulator(vec![
+            ExactKind::Sum,
+            ExactKind::Count
+        ])]
     );
     assert_eq!(
         targets[1].choices.candidates,
@@ -584,7 +579,7 @@ fn avg_rewrites_to_sum_and_count_states() {
     assert_eq!(
         targets[1].choices.pruned,
         vec![(
-            ComputationCandidate::AvgAsSumCount,
+            ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum, ExactKind::Count]),
             PruneReason::NullableAvgInput
         )]
     );
@@ -692,10 +687,7 @@ fn example_1_exact_q2_prunes_topk_heaps() {
         ]
     );
     assert!(found.iter().any(|choices| choices.candidates
-        == [ComputationCandidate::ExactAccumulator {
-            kind: ExactKind::Sum,
-            params: ExactParams::Sum
-        }]
+        == [ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum])]
         && choices.pruned.is_empty()));
     assert_eq!(found.len(), 2, "{found:#?}");
 }
@@ -707,7 +699,7 @@ fn having_of(
 ) -> Vec<HavingEntry> {
     let target = targets
         .iter()
-        .find(|target| match target.target.non_asap() {
+        .find(|target| match target.aggregate.non_asap() {
             Some(NonASAPOp::Aggregate { measures, .. }) => is(&measures[target.measure]),
             _ => false,
         })
@@ -740,7 +732,7 @@ fn scalar_reference_filter_is_not_having() {
         .unwrap()
         .targets;
     assert_eq!(targets.len(), 1);
-    assert!(Rc::ptr_eq(&targets[0].target, &agg));
+    assert!(Rc::ptr_eq(&targets[0].aggregate, &agg));
     assert!(targets[0].having.is_empty(), "{:?}", targets[0].having);
 
     let agg = aggregate_over(scan(false), count(), vec![], None);

crates/devtools/src/candidate_labels.rs

@@ -59,46 +59,43 @@ pub fn choice_label<Id>(inventory: &PerComputationCandidates<Id>, choice: &[usiz
                     root_operators(root).into_iter().any(|node| {
                         OperatorNode::reachable(node)
                             .iter()
-                            .any(|n| Rc::ptr_eq(n, &target.target))
+                            .any(|n| Rc::ptr_eq(n, &target.aggregate))
                     })
                 })
                 .expect("every target is reached from a root")
         })
         .collect();
-    // Targets a chosen whole-expression candidate absorbs are not computed.
-    let absorbed: Vec<usize> = inventory
+    // Each target's chosen candidate, with the target it absorbs when the
+    // candidate is its fused computation's.
+    let chosen: Vec<_> = inventory
         .targets
         .iter()
         .zip(choice)
-        .filter(|(target, &index)| target.choices.candidates[index].absorbs_inner())
-        .filter_map(|(target, _)| target.absorbs)
+        .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();
     (0..inventory.roots.len())
         .map(|query| {
             let mut sketches = Vec::new();
             let mut accumulators = Vec::new();
-            for (t, (target, (&target_query, &index))) in inventory
-                .targets
-                .iter()
-                .zip(query_of_target.iter().zip(choice))
-                .enumerate()
+            for (t, (&(candidate, absorbs), &target_query)) in
+                chosen.iter().zip(&query_of_target).enumerate()
             {
                 if target_query != query || absorbed.contains(&t) {
                     continue;
                 }
-                match &target.choices.candidates[index] {
+                match candidate {
                     ComputationCandidate::Exact => {}
-                    ComputationCandidate::ExactAccumulator { kind, .. } => {
-                        accumulators.push(format!("{kind:?} acc"))
-                    }
-                    ComputationCandidate::AvgAsSumCount => {
-                        accumulators.push("Sum/Count acc".into())
-                    }
-                    ComputationCandidate::NestedSumMerge => {
-                        accumulators.push("nested Sum acc".into())
+                    ComputationCandidate::ExactAccumulator(kinds) => {
+                        // e.g. "Sum acc", "Sum/Count acc" for an avg, or
+                        // "nested Sum acc" for a fused nested sum.
+                        let kinds: Vec<String> =
+                            kinds.iter().map(|kind| format!("{kind:?}")).collect();
+                        let nested = if absorbs.is_some() { "nested " } else { "" };
+                        accumulators.push(format!("{nested}{} acc", kinds.join("/")))
                     }
-                    ComputationCandidate::Summary(algorithm)
-                    | ComputationCandidate::WholeExpression(algorithm) => {
+                    ComputationCandidate::Summary(algorithm) => {
                         sketches.push(sketch_name(algorithm))
                     }
                 }
@@ -498,6 +495,50 @@ mod tests {
         assert_eq!(distinct.len(), logical.len(), "{logical:#?}");
     }
 
+    /// A fused computation's candidate is labeled apart from the unfused pair's:
+    /// the nested sum as "nested Sum acc", beside the two separate sums; an
+    /// avg's exact states as "Sum/Count acc".
+    #[test]
+    fn fused_computations_and_avg_states_have_their_own_labels() {
+        use crate::planner_layering_examples::shared_data_workload;
+        use asap_types::types::AccuracyTarget;
+        use asap_types::workload::{
+            DataArrival, PlanningWorkload, Predictability, Query, QueryLanguage, QueryRequirements,
+            QueryWorkload, RepeatedDemand, RepeatingEntry, RepetitionInterval,
+        };
+        let labels_of = |query: &str| {
+            let workload = PlanningWorkload {
+                query_workload: QueryWorkload {
+                    language: QueryLanguage::PromQL,
+                    query_batch: None,
+                    repeating_queries: Some(vec![RepeatingEntry {
+                        query: Query(query.into()),
+                        demand: RepeatedDemand::FixedInterval(RepetitionInterval(10_000)),
+                        requirements: QueryRequirements {
+                            accuracy: AccuracyTarget::exact(),
+                            max_latency_ms: None,
+                        },
+                        predictability: Predictability::Predictable { known_at: None },
+                    }]),
+                },
+                data_workload: Some(shared_data_workload(DataArrival::ContinuouslyIngesting)),
+            };
+            let run = run_planning_stages(&workload, lower_promql_roots(&workload).unwrap(), None)
+                .unwrap();
+            let labels = candidate_labels(&run);
+            run.stage1_logical_asap_optimization
+                .iter()
+                .filter(|c| c.pass2.is_none())
+                .map(|c| labels[&c.id].clone())
+                .collect::<Vec<_>>()
+        };
+        assert_eq!(
+            labels_of("sum by (job) (sum_over_time(m[5m]))"),
+            ["Q1 exact (Sum acc, Sum acc)", "Q1 exact (nested Sum acc)"]
+        );
+        assert_eq!(labels_of("avg by (job) (m)"), ["Q1 exact (Sum/Count acc)"]);
+    }
+
     /// A summary build read by two queries can be built once per evaluation
     /// for both: its label says "query time, shared".
     #[test]

crates/frontend-sql/tests/frequency_l2.rs

@@ -162,13 +162,13 @@ async fn stage1_offers_exact_and_summary_l2_candidates() {
     let l2 = inventory
         .targets
         .iter()
-        .find(|t| has_l2(&t.target))
+        .find(|t| has_l2(&t.aggregate))
         .expect("L2 target");
     let candidates = &l2.choices.candidates;
     assert!(
         candidates.iter().any(|c| matches!(
             c,
-            ComputationCandidate::Exact | ComputationCandidate::ExactAccumulator { .. }
+            ComputationCandidate::Exact | ComputationCandidate::ExactAccumulator(_)
         )),
         "{candidates:?}"
     );