From 0f37e746b3625ca3c61e157748c187899e492e51 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Fri, 16 Oct 2026 02:29:55 +0000 Subject: [PATCH 1/4] refactor(planner)!: split Pass 1 candidates into exact or summary; fused computations become targets ComputationCandidate now says only how a target is computed: Exact, ExactAccumulator(Vec) or Summary(SketchAlgorithm). The whole-expression top-k and the nested-sum merge become fused targets (ComputationTarget::fused_target) that absorb the target beneath; avg becomes ExactAccumulator([Sum, Count]). WholeExpression, NestedSumMerge, AvgAsSumCount and ComputationCandidate::absorbs_inner are removed, and ComputationTarget::target is renamed aggregate. Enumeration order, candidate counts, plans, costs and labels are unchanged. --- .../asap-aware-mapping/src/asap_aware_cse.rs | 21 +- .../src/pass/stage1_to_stage3.rs | 15 +- .../src/per_computation_candidates.rs | 572 ++++++++++-------- .../src/stage2_physical_asap_optimization.rs | 4 +- .../src/stage3_plan_selection.rs | 4 +- .../tests/per_computation_candidates.rs | 53 +- crates/devtools/src/candidate_labels.rs | 92 ++- crates/frontend-sql/tests/frequency_l2.rs | 4 +- .../tests/stage1_logical_asap_optimization.rs | 23 +- .../stage2_physical_asap_optimization_sql.rs | 7 +- crates/planner/tests/plan_stages.rs | 2 +- 11 files changed, 461 insertions(+), 336 deletions(-) diff --git a/crates/asap-aware-mapping/src/asap_aware_cse.rs b/crates/asap-aware-mapping/src/asap_aware_cse.rs index 2d530f5ee..edb489010 100644 --- a/crates/asap-aware-mapping/src/asap_aware_cse.rs +++ b/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 { 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 { @@ -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 = inventory .targets .iter() - .map(|t| t.choices.candidates.iter().position(heap).unwrap_or(0)) + .map(|t| { + (0..t.choice_count()) + .position(|i| t.chosen(i).is_some_and(|(_, c)| heap(c))) + .unwrap_or(0) + }) .collect(); let heaps: Vec = compose_logical_candidate(&inventory, &choice) .unwrap() diff --git a/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs b/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs index fdf3b4ee7..fd30a4071 100644 --- a/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs +++ b/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 target's (`targets[t].chosen(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( /// 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].chosen(choice[t])`: target `t`'s + /// candidate, or its fused target's); a Pass 2 candidate has its Pass 1 + /// candidate's. pub choice: Vec, pub roots: Result, PerComputationCandidateError>, /// For a Pass 2 candidate: what it changed of its Pass 1 candidate. @@ -576,12 +578,13 @@ pub fn run_stage2_and_stage3( } } -/// 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 target's: the last target varies fastest. /// `false` when a carry out of the first target ends the enumeration. fn next_choice(inventory: &PerComputationCandidates, 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; diff --git a/crates/asap-aware-mapping/src/per_computation_candidates.rs b/crates/asap-aware-mapping/src/per_computation_candidates.rs index fca22f881..dadd3c43c 100644 --- a/crates/asap-aware-mapping/src/per_computation_candidates.rs +++ b/crates/asap-aware-mapping/src/per_computation_candidates.rs @@ -9,11 +9,16 @@ //! nested aggregates and operators read by scalar roots are found too. //! Pass 1 does not consider sharing across computations; that is Pass 2. //! -//! - Each target has exactly one exact option: the exact accumulator -//! (`ExactAccumulator`, or `AvgAsSumCount` for `avg`) when it can be built, -//! otherwise plain `Exact`. Whether it is materialized is decided in Stage 2. -//! A grouped `sum` over a `sum` beneath it also gets the nested-sum merge -//! (`NestedSumMerge`, #680), one exact accumulator absorbing the inner sum. +//! - A candidate says how a target is computed: exact or a summary +//! ([`ComputationCandidate`]). Each target has exactly one exact option: +//! its exact accumulator states (`ExactAccumulator`; `[Sum, Count]` for +//! `avg`) when they can be built, otherwise plain `Exact`. Whether they are +//! materialized is decided in Stage 2. +//! - A computation that fuses a target with the target beneath it is its own +//! target, which absorbs the one beneath +//! ([`ComputationTarget::fused_target`]): the whole-expression top-k over a +//! per-item `sum` or `count` (`Summary` of a heap family) and the nested +//! sum, a grouped `sum` over a `sum` (`ExactAccumulator([Sum])`). //! - Summary candidates are unsized; they are sized after Pass 2 sharing. //! - Candidates are pruned only when provably unable to meet the accuracy //! requirement, and each pruned candidate keeps its [`PruneReason`]. @@ -31,22 +36,26 @@ //! //! [`compose_logical_candidate`] is Algorithm 1's //! `SUBSTITUTE(d, s_1 ↦ c_1, …)`: it picks one per-computation candidate per -//! target (`choice[i]` indexes `targets[i].choices.candidates`) and builds one -//! whole-workload candidate DAG. An aggregate whose chosen candidates are all -//! `Exact` is kept. Otherwise each of its measures is substituted on its own -//! over the aggregate's input: +//! target (`choice[i]` indexes `targets[i]`'s candidates, then its fused +//! target's, [`ComputationTarget::chosen`]) and builds one whole-workload +//! candidate DAG. An aggregate whose chosen candidates are all `Exact` is +//! kept. Otherwise each of its measures is substituted on its own over the +//! aggregate's input: //! -//! - `ExactAccumulator`: `SummaryAgg` with the exact family → +//! - `ExactAccumulator` with one state: `SummaryAgg` with the exact family → //! `FinalizeExactAccumulator`. +//! - `ExactAccumulator([Sum, Count])` of an `avg`: the finalized `sum` and +//! `count` accumulators divided: a PromQL `BinaryOp`, or for SQL rows a +//! join on the group keys (NULL keys match) and a SQL division, NULL where +//! the count is 0. //! - `Summary`: `SummaryAgg` → `SummaryEstimate`. -//! - `AvgAsSumCount`: the finalized `sum` and `count` accumulators divided: -//! a PromQL `BinaryOp`, or for SQL rows a join on the group keys (NULL keys -//! match) and a SQL division, NULL where the count is 0. -//! - `NestedSumMerge`: `SummaryAgg` with the exact `Sum` family over the -//! absorbed inner sum's input, grouped by the outer keys → -//! `FinalizeExactAccumulator`. //! - `Exact` (or a measure that is no target): a single-measure `Aggregate`. //! +//! A fused target's candidate reads the absorbed target's input instead: +//! the whole-expression top-k is one heap `SummaryAgg` keyed by the ranked +//! item → `SummaryEstimate`; the nested sum is `SummaryAgg` with the exact +//! `Sum` family grouped by the outer keys → `FinalizeExactAccumulator`. +//! //! The substituted output keeps the aggregate's column names. The measures //! of a multi-measure aggregate are joined back on the group keys (NULL keys //! match, as grouping treats them). A measure filter becomes @@ -115,49 +124,52 @@ use crate::realization::{ accuracy_budget, accuracy_target, default_size_params, summary_candidates, }; -/// One Pass 1 candidate for a computation (one measure of an aggregate). +/// How one Pass 1 target is computed: exact or a summary. /// -/// Each target has exactly one exact option: `ExactAccumulator` or -/// `AvgAsSumCount` when the intent has one and it can be built, otherwise +/// Each target has exactly one exact option: `ExactAccumulator` when the +/// intent has an exact mergeable state and it can be built, otherwise /// `Exact`. The plain form would duplicate "accumulator, not materialized"; -/// Stage 2/3 decide materialization and sharing. +/// Stage 2/3 decide materialization and sharing. Which computation is +/// computed is the target's ([`ComputationTarget`]); a fused computation is +/// its own target ([`ComputationTarget::fused_target`]). #[derive(Debug, Clone, PartialEq)] 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, - /// An exact mergeable accumulator state. - ExactAccumulator { - kind: ExactKind, - params: ExactParams, - }, + /// One or more exact mergeable accumulator states, finalized and + /// combined into the measure: one state for `sum`, `count`, `min`, + /// `max`, `rate`, `irate` and `increase`; `[Sum, Count]`, divided, for + /// `avg`. + ExactAccumulator(Vec), /// An unsized summary family, sized later for the measure's accuracy. Summary(SketchAlgorithm), - /// `avg` rewritten as exact `sum` and `count` accumulator states. - AvgAsSumCount, - /// A whole-expression top-k: one heap summary (Count-Min or CountSketch - /// with a heap) over the input of the per-item `sum` or `count` beneath - /// the top-k, keyed by the ranked item and weighted by the summed value. - /// It absorbs the target beneath ([`ComputationTarget::absorbs`]): that - /// aggregate is not computed. Where it applies it replaces the heap of - /// the same family over the exact per-item result. Unsized, like - /// `Summary`. - WholeExpression(SketchAlgorithm), - /// A nested sum merged (#680): a grouped `sum` over the value of a - /// `sum` beneath it (`sum by (job) (sum_over_time(m[5m]))`) as one exact - /// `Sum` accumulator over the inner sum's input, grouped by the outer - /// keys: the sum of the inner groups' sums is the sum of their rows. It - /// absorbs the inner target ([`ComputationTarget::absorbs`]), which is - /// not computed, and is offered beside the outer sum's own accumulator. - NestedSumMerge, } -impl ComputationCandidate { - /// Whether the candidate absorbs the target beneath its own - /// ([`ComputationTarget::absorbs`]): a whole-expression top-k or a - /// nested-sum merge. - pub fn absorbs_inner(&self) -> bool { - matches!(self, Self::WholeExpression(_) | Self::NestedSumMerge) +/// One exact mergeable accumulator state: which accumulator, and its +/// parameters (`FieldDataType::ExactAggregate`'s pair). +#[derive(Debug, Clone, PartialEq)] +pub struct ExactState { + /// Which accumulator: `Sum`, `Count`, `Min`, `Max`, `Rate`, `IRate` or + /// `Increase`. + pub kind: ExactKind, + /// Its parameters; none of them has any, so this mirrors `kind`. + pub params: ExactParams, +} + +impl ExactState { + /// The state of accumulator `kind`, with its (empty) parameters. + pub fn of(kind: ExactKind) -> Self { + let params = match kind { + ExactKind::Sum => ExactParams::Sum, + ExactKind::Count => ExactParams::Count, + ExactKind::Min => ExactParams::Min, + ExactKind::Max => ExactParams::Max, + ExactKind::Rate => ExactParams::Rate, + ExactKind::IRate => ExactParams::IRate, + ExactKind::Increase => ExactParams::Increase, + }; + Self { kind, params } } } @@ -185,12 +197,10 @@ impl ComputationChoices { .into_iter() .partition(|choice| *choice != ComputationCandidate::Exact && applies(choice)); self.candidates = kept; - if pruned.iter().any(|choice| { - matches!( - choice, - ComputationCandidate::ExactAccumulator { .. } | ComputationCandidate::AvgAsSumCount - ) - }) { + if pruned + .iter() + .any(|choice| matches!(choice, ComputationCandidate::ExactAccumulator(_))) + { self.candidates.insert(0, ComputationCandidate::Exact); } self.pruned.extend( @@ -231,11 +241,13 @@ pub struct HavingEntry { } /// One measure of an aggregate with at least one realization besides exact -/// execution. The target retains source, grouping, filters, input expressions -/// and evaluation context. +/// execution, or a fused computation of it ([`Self::fused_target`]). The +/// target retains source, grouping, filters, input expressions and +/// evaluation context. #[derive(Debug, Clone)] pub struct ComputationTarget { - pub target: Rc, + /// The `Aggregate` node the measure belongs to. + pub aggregate: Rc, /// Index into the aggregate's measures. pub measure: usize, pub choices: ComputationChoices, @@ -245,12 +257,17 @@ pub struct ComputationTarget { /// The aggregate's own `having`, then one entry per consumer `Filter` /// directly above the aggregate. pub having: Vec, - /// The target directly beneath this one that its absorbing candidates - /// ([`ComputationCandidate::absorbs_inner`]) absorb. A - /// whole-workload choice of such a candidate takes the absorbed target's - /// first candidate: the target is not computed, so its own choice does - /// not matter. + /// On a fused target only: the index (in + /// [`PerComputationCandidates::targets`]) of the target directly + /// beneath that it absorbs, which is not computed when the fused target + /// is chosen. `None` on every target of the inventory's list. pub absorbs: Option, + /// The fused alternative of this target, computing the same measure + /// together with the target beneath it, which it absorbs: a + /// whole-expression top-k over the per-item `sum` or `count` beneath, or + /// the nested-sum merge of a `sum` over a `sum`. `None` when no fusion + /// applies, and on a fused target itself. + pub fused_target: Option>, /// The metric whose raw samples, or plain sums of them, are every value /// the aggregate reads ([`samples_of_metric`]); `None` otherwise. A /// heap-sketch weight read from them carries the open requirement @@ -258,57 +275,89 @@ pub struct ComputationTarget { pub samples_of: Option, } +impl ComputationTarget { + /// How many values this target's digit of a whole-workload choice + /// takes: its own candidates, then its fused target's. + pub fn choice_count(&self) -> usize { + self.choices.candidates.len() + + self + .fused_target + .as_ref() + .map_or(0, |fused| fused.choices.candidates.len()) + } + + /// The target that computes the measure and its candidate when this + /// target's digit of a whole-workload choice is `index`: this target + /// and its own candidate `index`, or past them, its fused target and + /// that target's candidate. `None` past [`Self::choice_count`]. + pub fn chosen(&self, index: usize) -> Option<(&ComputationTarget, &ComputationCandidate)> { + let own = self.choices.candidates.len(); + match self.choices.candidates.get(index) { + Some(candidate) => Some((self, candidate)), + None => { + let fused = self.fused_target.as_deref()?; + Some((fused, fused.choices.candidates.get(index - own)?)) + } + } + } +} + /// Compact Pass 1 inventory; roots and nested producer dependencies are retained. #[derive(Debug, Clone)] pub struct PerComputationCandidates { pub roots: Vec<(Id, QueryRoot)>, + /// The unfused targets; a whole-workload choice has one digit per + /// target, ranging over [`ComputationTarget::choice_count`]. pub targets: Vec, } impl PerComputationCandidates { /// The number of whole-workload candidates: the size of the Cartesian - /// product of the targets' candidates, without the choices that pick an - /// absorbing candidate and anything but the first candidate of the - /// target it absorbs (the same candidate as with the first), - /// saturating at `usize::MAX`. An absorbed target has one absorber, and - /// an absorbing target is never absorbed, so each pair counts as - /// `(n − w) · m + w` for an absorber with `n` candidates, `w` of them - /// absorbing, over an absorbed target with `m`. + /// product of the targets' candidates, where a target with a fused + /// target and the target beneath that it absorbs count as their own + /// pairs plus the fused target's candidates (the absorbed target is not + /// computed then), saturating at `usize::MAX`. An absorbed target has + /// one fused target over it, and a fused target is never absorbed, so + /// each such pair counts as `n · m + f` for a target with `n` + /// candidates over an absorbed target with `m`, and `f` fused ones. pub fn combinations(&self) -> usize { - let absorbed: HashSet = self.targets.iter().filter_map(|t| t.absorbs).collect(); + let absorbed: HashSet = absorbed_targets(&self.targets).collect(); self.targets .iter() .enumerate() .filter(|(index, _)| !absorbed.contains(index)) .map(|(_, target)| { let n = target.choices.candidates.len(); - let Some(inner) = target.absorbs else { + let Some(fused) = &target.fused_target else { return n; }; - let whole = target - .choices - .candidates - .iter() - .filter(|c| c.absorbs_inner()) - .count(); - (n - whole) - .saturating_mul(self.targets[inner].choices.candidates.len()) - .saturating_add(whole) + let inner = fused.absorbs.expect("a fused target absorbs"); + n.saturating_mul(self.targets[inner].choices.candidates.len()) + .saturating_add(fused.choices.candidates.len()) }) .fold(1usize, usize::saturating_mul) } - /// Whether `choice` picks an absorbing candidate and a candidate other - /// than the first for the target it absorbs: the same whole-workload - /// candidate as the one with the first, so enumeration skips it. + /// Whether `choice` picks a fused target and a candidate other than the + /// first for the target it absorbs: the same whole-workload candidate as + /// the one with the first, so enumeration skips it. pub fn absorbed_choice_varies(&self, choice: &[usize]) -> bool { self.targets.iter().zip(choice).any(|(target, &index)| { - target.choices.candidates[index].absorbs_inner() - && target.absorbs.is_some_and(|inner| choice[inner] != 0) + target + .chosen(index) + .and_then(|(computing, _)| computing.absorbs) + .is_some_and(|inner| choice[inner] != 0) }) } } +/// The indices of `targets` some target's fused target absorbs. +fn absorbed_targets(targets: &[ComputationTarget]) -> impl Iterator + '_ { + targets + .iter() + .filter_map(|target| target.fused_target.as_ref()?.absorbs) +} + #[derive(Debug, Error)] pub enum PerComputationCandidateError { #[error(transparent)] @@ -341,20 +390,21 @@ pub enum PerComputationCandidateError { pub fn computation_candidates_for_intent( intent: &AggIntent, ) -> Result { - let exact = match intent { - AggIntent::Count { .. } => Some((ExactKind::Count, ExactParams::Count)), - AggIntent::Sum { .. } => Some((ExactKind::Sum, ExactParams::Sum)), - AggIntent::Min { .. } => Some((ExactKind::Min, ExactParams::Min)), - AggIntent::Max { .. } => Some((ExactKind::Max, ExactParams::Max)), - AggIntent::Rate => Some((ExactKind::Rate, ExactParams::Rate)), - AggIntent::IRate => Some((ExactKind::IRate, ExactParams::IRate)), - AggIntent::Increase => Some((ExactKind::Increase, ExactParams::Increase)), - _ => None, + let states = match intent { + AggIntent::Count { .. } => vec![ExactKind::Count], + AggIntent::Sum { .. } => vec![ExactKind::Sum], + AggIntent::Min { .. } => vec![ExactKind::Min], + AggIntent::Max { .. } => vec![ExactKind::Max], + AggIntent::Rate => vec![ExactKind::Rate], + AggIntent::IRate => vec![ExactKind::IRate], + AggIntent::Increase => vec![ExactKind::Increase], + AggIntent::Avg { .. } => vec![ExactKind::Sum, ExactKind::Count], + _ => vec![], }; - let exact = match exact { - Some((kind, params)) => ComputationCandidate::ExactAccumulator { kind, params }, - None if matches!(intent, AggIntent::Avg { .. }) => ComputationCandidate::AvgAsSumCount, - None => ComputationCandidate::Exact, + let exact = if states.is_empty() { + ComputationCandidate::Exact + } else { + ComputationCandidate::ExactAccumulator(states.into_iter().map(ExactState::of).collect()) }; let mut choices = ComputationChoices { candidates: vec![exact], @@ -533,16 +583,17 @@ pub fn generate_per_computation_candidates( } if matches!(intent, AggIntent::Avg { .. }) && nullable_avg_input(&node, intent) { choices.prune(PruneReason::NullableAvgInput, |choice| { - *choice == ComputationCandidate::AvgAsSumCount + matches!(choice, ComputationCandidate::ExactAccumulator(_)) }); } targets.push(ComputationTarget { - target: node.clone(), + aggregate: node.clone(), measure, choices, filter: filters.get(measure).cloned().flatten(), having: entries.clone(), absorbs: None, + fused_target: None, samples_of: node.non_asap().and_then(|op| match op { NonASAPOp::Aggregate { child, .. } => samples_of_metric(child), _ => None, @@ -550,54 +601,70 @@ pub fn generate_per_computation_candidates( }); } } - add_whole_expression_candidates(&mut targets, &consumers); - add_nested_sum_candidates(&mut targets, &consumers); + add_whole_expression_targets(&mut targets, &consumers); + add_nested_sum_targets(&mut targets, &consumers); Ok(PerComputationCandidates { roots, targets }) } -/// Nested-sum merge (#680): a grouped `sum` over the value of a `sum` -/// target beneath it gets [`ComputationCandidate::NestedSumMerge`] beside -/// its own candidates, absorbing the inner target. Applied only when the -/// inner sum has no other consumer, so absorbing it removes its work, when -/// neither has a filter or `HAVING` ([`nested_sum_input`]), and when the -/// outer sum is not itself absorbed (by a whole-expression top-k or a -/// nested-sum merge above it) and the inner sum absorbs nothing, so an -/// absorbing target is never absorbed and merges do not stack. -fn add_nested_sum_candidates( +/// Nested-sum merge: a grouped `sum` over the value of a `sum` target +/// beneath it gets a fused target ([`ComputationTarget::fused_target`]) +/// with the one candidate `ExactAccumulator([Sum])`: one exact `Sum` +/// accumulator over the inner sum's input, grouped by the outer keys (the +/// sum of the inner groups' sums is the sum of their rows), absorbing the +/// inner target. Applied only when the inner sum has no other consumer, so +/// absorbing it removes its work, when neither has a filter or `HAVING` +/// ([`nested_sum_input`]), and when the outer sum is not itself absorbed +/// and has no fused target yet, and the inner sum has none, so a fused +/// target is never absorbed and merges do not stack. +fn add_nested_sum_targets( targets: &mut [ComputationTarget], consumers: &HashMap<*const OperatorNode, Vec>>>, ) { - let mut absorbed: HashSet = targets.iter().filter_map(|t| t.absorbs).collect(); + let mut absorbed: HashSet = absorbed_targets(targets).collect(); for t in 0..targets.len() { let target = &targets[t]; - if absorbed.contains(&t) || target.absorbs.is_some() || !target.having.is_empty() { + if absorbed.contains(&t) || target.fused_target.is_some() || !target.having.is_empty() { continue; } - let Some(NonASAPOp::Aggregate { child, .. }) = target.target.non_asap() else { + let Some(NonASAPOp::Aggregate { child, .. }) = target.aggregate.non_asap() else { continue; }; let Some(inner) = targets .iter() - .position(|other| Rc::ptr_eq(&other.target, child)) + .position(|other| Rc::ptr_eq(&other.aggregate, child)) else { continue; }; - // An absorbing target is never absorbed: the inner sum must not - // absorb a sum of its own. + // A fused target is never absorbed: the inner sum must not have a + // fused target of its own. if consumers.get(&Rc::as_ptr(child)).map(Vec::len) != Some(1) || !targets[inner].having.is_empty() - || targets[inner].absorbs.is_some() - || nested_sum_input(&target.target).is_none() + || targets[inner].fused_target.is_some() + || nested_sum_input(&target.aggregate).is_none() { continue; } absorbed.insert(inner); - let target = &mut targets[t]; - target - .choices - .candidates - .push(ComputationCandidate::NestedSumMerge); - target.absorbs = Some(inner); + let merge = ComputationCandidate::ExactAccumulator(vec![ExactState::of(ExactKind::Sum)]); + targets[t].fused_target = Some(Box::new(fused_target(&targets[t], inner, vec![merge]))); + } +} + +/// The fused target of `target` absorbing `targets[inner]`, with +/// `candidates`: the same measure, filter, `HAVING` and samples. +fn fused_target( + target: &ComputationTarget, + inner: usize, + candidates: Vec, +) -> ComputationTarget { + ComputationTarget { + choices: ComputationChoices { + candidates, + pruned: vec![], + }, + absorbs: Some(inner), + fused_target: None, + ..target.clone() } } @@ -671,26 +738,29 @@ fn nested_sum_input(target: &OperatorNode) -> Option<(Rc, AggInten Some((Rc::clone(rows), inner_sum.clone(), partitions)) } -/// Whole-expression top-k (#509 Example 1's Q2): a top-k over a per-item -/// `sum` or `count` that is itself a target has its heap candidates -/// (Count-Min and CountSketch with a heap) replaced by -/// [`ComputationCandidate::WholeExpression`] ones of the same families, -/// which absorb the inner target. Applied only when the inner aggregate has -/// no other consumer, so absorbing it removes its work, when neither has a -/// filter or `HAVING`, and when the keyed-additive rule realizes it +/// Whole-expression top-k (Example 1's Q2): a top-k over a per-item `sum` +/// or `count` that is itself a target gets a fused target +/// ([`ComputationTarget::fused_target`]) absorbing the inner target, whose +/// candidates are the top-k's heap summaries (Count-Min and CountSketch +/// with a heap), each one heap over the inner aggregate's input, keyed by +/// the ranked item and weighted by the summed value. Those heaps over the +/// exact per-item result leave the top-k's own candidates: the fused +/// target replaces them. Applied only when the inner aggregate has no other +/// consumer, so absorbing it removes its work, when neither has a filter or +/// `HAVING`, and when the keyed-additive rule realizes it /// ([`whole_expression_input`]). -fn add_whole_expression_candidates( +fn add_whole_expression_targets( targets: &mut [ComputationTarget], consumers: &HashMap<*const OperatorNode, Vec>>>, ) { for t in 0..targets.len() { let target = &targets[t]; - let Some(NonASAPOp::Aggregate { child, .. }) = target.target.non_asap() else { + let Some(NonASAPOp::Aggregate { child, .. }) = target.aggregate.non_asap() else { continue; }; let Some(inner) = targets .iter() - .position(|other| Rc::ptr_eq(&other.target, child)) + .position(|other| Rc::ptr_eq(&other.aggregate, child)) else { continue; }; @@ -699,24 +769,30 @@ fn add_whole_expression_candidates( || !target.having.is_empty() || targets[inner].filter.is_some() || !targets[inner].having.is_empty() - || whole_expression_input(&target.target).is_none() + || whole_expression_input(&target.aggregate).is_none() { continue; } - let target = &mut targets[t]; - let mut replaced = false; - for candidate in &mut target.choices.candidates { - if let ComputationCandidate::Summary( - algorithm @ (SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap), - ) = candidate - { - *candidate = ComputationCandidate::WholeExpression(algorithm.clone()); - replaced = true; - } - } - if replaced { - target.absorbs = Some(inner); + let (heaps, own): (Vec<_>, Vec<_>) = targets[t] + .choices + .candidates + .iter() + .cloned() + .partition(|candidate| { + matches!( + candidate, + ComputationCandidate::Summary( + SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap + ) + ) + }); + if heaps.is_empty() { + continue; } + let fused = fused_target(&targets[t], inner, heaps); + let target = &mut targets[t]; + target.choices.candidates = own; + target.fused_target = Some(Box::new(fused)); } } @@ -879,15 +955,13 @@ pub fn compose_logical_candidate( } let mut chosen: Chosen = HashMap::new(); for (target, &index) in inventory.targets.iter().zip(choice) { - let candidate = target - .choices - .candidates - .get(index) + let (computing, candidate) = target + .chosen(index) .ok_or(PerComputationCandidateError::InvalidChoice)?; chosen - .entry(Rc::as_ptr(&target.target)) + .entry(Rc::as_ptr(&computing.aggregate)) .or_default() - .push((target, candidate)); + .push((computing, candidate)); } // Consumer `Filter`s whose key-only `HAVING` moves below a substituted // aggregate: each is replaced by its input. @@ -2427,27 +2501,33 @@ fn substitute( .iter() .enumerate() .map(|(measure, intent)| { - let (candidate, samples_of) = targets + // `fused`: the candidate is a fused target's, which absorbs the + // target beneath. + let (candidate, samples_of, fused) = targets .iter() .find(|(target, _)| target.measure == measure) .map_or( - (&ComputationCandidate::Exact, None), - |(target, candidate)| (*candidate, target.samples_of.as_deref()), + (&ComputationCandidate::Exact, None, false), + |(target, candidate)| { + ( + *candidate, + target.samples_of.as_deref(), + target.absorbs.is_some(), + ) + }, ); let filter = filters.get(measure).and_then(Option::as_ref).map(rewritten); let key = (Rc::as_ptr(target), measure); - if *candidate == ComputationCandidate::NestedSumMerge { - return nested_sum(target, memo); - } - if let ComputationCandidate::WholeExpression(algorithm) = candidate { - return whole_expression( - target, - intent, - algorithm, - samples_of, - (sizing, key), - memo, - ); + if fused { + return match candidate { + ComputationCandidate::ExactAccumulator(_) => nested_sum(target, memo), + ComputationCandidate::Summary(algorithm) => { + whole_expression(target, intent, algorithm, samples_of, (sizing, key), memo) + } + ComputationCandidate::Exact => Err(PerComputationCandidateError::Unsupported( + "plain exact execution of a fused target", + )), + }; } substitute_measure( target, @@ -2577,15 +2657,7 @@ pub(crate) fn exact_accumulator( input: &Rc, reduction: Reduction, ) -> Result, PerComputationCandidateError> { - let params = match kind { - ExactKind::Sum => ExactParams::Sum, - ExactKind::Count => ExactParams::Count, - ExactKind::Min => ExactParams::Min, - ExactKind::Max => ExactParams::Max, - ExactKind::Rate => ExactParams::Rate, - ExactKind::IRate => ExactParams::IRate, - ExactKind::Increase => ExactParams::Increase, - }; + let ExactState { kind, params } = ExactState::of(kind); let family = FieldDataType::ExactAggregate(kind, params); let read = measure_input(intent, input, None)?; Ok(OperatorNode::new_shared(Operator::ASAP( @@ -2628,8 +2700,9 @@ fn substitute_measure( }, ))?) }; - let finalized = |kind: ExactKind, params: ExactParams, intent: &AggIntent, read| { - let state = state(FieldDataType::ExactAggregate(kind, params), intent, read)?; + let finalized = |exact: &ExactState, intent: &AggIntent, read| { + let family = FieldDataType::ExactAggregate(exact.kind.clone(), exact.params.clone()); + let state = state(family, intent, read)?; Ok::<_, PerComputationCandidateError>(OperatorNode::new_shared(Operator::ASAP( ASAPOp::FinalizeExactAccumulator { child: state }, ))?) @@ -2645,9 +2718,9 @@ fn substitute_measure( having: None, }))? } - ComputationCandidate::ExactAccumulator { kind, params } => { + ComputationCandidate::ExactAccumulator(states) if states.len() == 1 => { let read = measure_input(intent, input, filter.as_ref())?; - finalized(kind.clone(), params.clone(), intent, read)? + finalized(&states[0], intent, read)? } ComputationCandidate::Summary(algorithm) => { let family = FieldDataType::Sketch( @@ -2665,15 +2738,11 @@ fn substitute_measure( query: statistic(intent)?, }))? } - ComputationCandidate::WholeExpression(_) | ComputationCandidate::NestedSumMerge => { - return Err(PerComputationCandidateError::Unsupported( - "an absorbing candidate is substituted with its aggregate", - )) - } - ComputationCandidate::AvgAsSumCount => { - let AggIntent::Avg { col } = intent else { + ComputationCandidate::ExactAccumulator(states) => { + let (AggIntent::Avg { col }, [sum_state, count_state]) = (intent, states.as_slice()) + else { return Err(PerComputationCandidateError::Unsupported( - "sum/count of a non-avg", + "several exact states other than avg's sum and count", )); }; let sum_intent = AggIntent::Sum { col: *col }; @@ -2681,10 +2750,9 @@ fn substitute_measure( // non-null (a nullable input prunes this candidate), so it counts // the rows the sum reads. let read = measure_input(&sum_intent, input, filter.as_ref())?; - let mut sum = finalized(ExactKind::Sum, ExactParams::Sum, &sum_intent, read.clone())?; + let mut sum = finalized(sum_state, &sum_intent, read.clone())?; let count = finalized( - ExactKind::Count, - ExactParams::Count, + count_state, &AggIntent::Count { accuracy: AccuracyTarget::exact(), }, @@ -3162,10 +3230,7 @@ mod tests { assert!(!choices.contains(&ComputationCandidate::Exact)); assert_eq!( choices[0], - ComputationCandidate::ExactAccumulator { - kind: ExactKind::Count, - params: ExactParams::Count - } + ComputationCandidate::ExactAccumulator(vec![ExactState::of(ExactKind::Count)]) ); assert!(choices.contains(&ComputationCandidate::Summary(SketchAlgorithm::UnivMon))); } @@ -3818,13 +3883,13 @@ mod tests { generate_per_computation_candidates(vec![(0, QueryRoot::Operator(root))]).unwrap() } - /// Every candidate of target `t` composed with every other target at its - /// first candidate, with the candidate. + /// Every candidate of target `t`, then of its fused target, composed + /// with every other target at its first candidate, with the candidate. fn composed_with( inventory: &PerComputationCandidates, t: usize, ) -> Vec<(ComputationCandidate, Rc)> { - (0..inventory.targets[t].choices.candidates.len()) + (0..inventory.targets[t].choice_count()) .map(|c| { let mut choice = vec![0; inventory.targets.len()]; choice[t] = c; @@ -3832,10 +3897,8 @@ mod tests { let QueryRoot::Operator(root) = &composed[0].1 else { panic!("operator root") }; - ( - inventory.targets[t].choices.candidates[c].clone(), - root.clone(), - ) + let (_, candidate) = inventory.targets[t].chosen(c).unwrap(); + (candidate.clone(), root.clone()) }) .collect() } @@ -3860,7 +3923,7 @@ mod tests { .iter() .position(|t| { matches!( - t.target.non_asap(), + t.aggregate.non_asap(), Some(NonASAPOp::Aggregate { measures, .. }) if matches!(measures.as_slice(), [AggIntent::TopK { .. }]) ) @@ -3868,25 +3931,29 @@ mod tests { .expect("a top-k target") } - /// Example 1's Q2: the top-k over a per-series `sum_over_time` gets - /// exactly three candidates, exact and a whole-expression Count-Min and - /// CountSketch heap, which replace the heaps over the exact sum. They - /// absorb the inner sum: composed, no aggregate is computed and the heap - /// reads the range selector's raw samples. + /// Example 1's Q2: the top-k over a per-series `sum_over_time` keeps + /// only its exact candidate and gets a fused whole-expression target + /// with a Count-Min and a CountSketch heap, which replace the heaps over + /// the exact sum. The fused target absorbs the inner sum: composed, no + /// aggregate is computed and the heap reads the range selector's raw + /// samples. #[test] fn whole_expression_topk_replaces_the_heap_over_the_exact_sum() { let inventory = inventory("topk by (job) (10, sum_over_time(m[1m]))"); let topk = topk_target(&inventory); let target = &inventory.targets[topk]; + assert_eq!(target.choices.candidates, [ComputationCandidate::Exact]); + assert_eq!(target.absorbs, None); + let fused = target.fused_target.as_deref().expect("a fused target"); assert_eq!( - target.choices.candidates, + fused.choices.candidates, [ - ComputationCandidate::Exact, - ComputationCandidate::WholeExpression(SketchAlgorithm::CmsWithHeap), - ComputationCandidate::WholeExpression(SketchAlgorithm::CountSketchWithHeap), + ComputationCandidate::Summary(SketchAlgorithm::CmsWithHeap), + ComputationCandidate::Summary(SketchAlgorithm::CountSketchWithHeap), ] ); - let inner = target.absorbs.expect("absorbs the inner sum"); + assert!(Rc::ptr_eq(&fused.aggregate, &target.aggregate)); + let inner = fused.absorbs.expect("absorbs the inner sum"); assert_eq!(inventory.targets[inner].choices.candidates.len(), 1); assert_eq!(inventory.combinations(), 3); for (candidate, root) in composed_with(&inventory, topk) { @@ -3930,7 +3997,9 @@ mod tests { let topk = topk_target(&inventory); let mut heaps = 0; for (candidate, root) in composed_with(&inventory, topk) { - if !matches!(candidate, ComputationCandidate::WholeExpression(_)) { + // The top-k's own candidate is exact; its summaries are the + // fused target's. + if !matches!(candidate, ComputationCandidate::Summary(_)) { continue; } for state in heap_states(&root) { @@ -4010,17 +4079,19 @@ mod tests { } } + /// The target whose fused target is a nested-sum merge. fn nested_sum_target(inventory: &PerComputationCandidates) -> Option { + let merge = ComputationCandidate::ExactAccumulator(vec![ExactState::of(ExactKind::Sum)]); inventory.targets.iter().position(|t| { - t.choices - .candidates - .contains(&ComputationCandidate::NestedSumMerge) + t.fused_target + .as_ref() + .is_some_and(|fused| fused.choices.candidates == [merge.clone()]) }) } - /// `sum by (job) (sum_over_time(m[5m]))` gets a nested-sum merge beside - /// its own accumulator: one exact Sum accumulator over the range's raw - /// samples, grouped by `job`, which absorbs the inner per-series sum. + /// `sum by (job) (sum_over_time(m[5m]))` gets a fused nested-sum target + /// beside its own accumulator: one exact Sum accumulator over the range's + /// raw samples, grouped by `job`, which absorbs the inner per-series sum. /// Composed, no per-series state is built, the value is exact and the /// output columns are the outer sum's. #[test] @@ -4028,17 +4099,11 @@ mod tests { let inventory = inventory("sum by (job) (sum_over_time(m[5m]))"); let outer = nested_sum_target(&inventory).expect("a nested-sum merge"); let target = &inventory.targets[outer]; - assert_eq!( - target.choices.candidates, - [ - ComputationCandidate::ExactAccumulator { - kind: ExactKind::Sum, - params: ExactParams::Sum - }, - ComputationCandidate::NestedSumMerge, - ] - ); - let inner = target.absorbs.expect("absorbs the inner sum"); + let sum = ComputationCandidate::ExactAccumulator(vec![ExactState::of(ExactKind::Sum)]); + assert_eq!(target.choices.candidates, std::slice::from_ref(&sum)); + let fused = target.fused_target.as_deref().expect("a fused target"); + assert_eq!(fused.choices.candidates, [sum]); + let inner = fused.absorbs.expect("absorbs the inner sum"); assert_eq!(inventory.targets[inner].choices.candidates.len(), 1); assert_eq!(inventory.combinations(), 2); let composed = composed_with(&inventory, outer); @@ -4116,17 +4181,17 @@ mod tests { } /// Merges do not stack: in `sum(sum by (job) (sum_over_time(m[5m])))` - /// a target that absorbs is never absorbed, and `combinations` counts - /// exactly the choices enumeration keeps. + /// a target with a fused target is never absorbed, and `combinations` + /// counts exactly the choices enumeration keeps. #[test] fn nested_sum_merges_do_not_stack() { let inventory = inventory("sum(sum by (job) (sum_over_time(m[5m])))"); - let absorbed: Vec = inventory.targets.iter().filter_map(|t| t.absorbs).collect(); + let absorbed: Vec = absorbed_targets(&inventory.targets).collect(); assert_eq!(absorbed.len(), 1, "one merge"); for &a in &absorbed { assert!( - inventory.targets[a].absorbs.is_none(), - "an absorbed target absorbs" + inventory.targets[a].fused_target.is_none(), + "an absorbed target has a fused target" ); } let mut enumerated = 0; @@ -4138,7 +4203,7 @@ mod tests { let mut carry = true; for (digit, target) in choice.iter_mut().zip(&inventory.targets).rev() { *digit += 1; - if *digit < target.choices.candidates.len() { + if *digit < target.choice_count() { carry = false; break; } @@ -4151,26 +4216,22 @@ mod tests { assert_eq!(inventory.combinations(), enumerated); } - /// `combinations` counts each choice once: picking a whole-expression - /// candidate with any but the first candidate of the target it absorbs - /// is the same candidate. An absorbed target with two candidates (an - /// approximate inner count) gives (1 exact + 0 summaries) · 2 + 2 - /// whole-expression = 4, not 3 · 2. + /// `combinations` counts each choice once: picking a fused + /// whole-expression candidate with any but the first candidate of the + /// target it absorbs is the same candidate. An absorbed target with four + /// candidates (an approximate inner count) gives 1 exact · 4 + 2 + /// whole-expression = 6, not 3 · 4. #[test] fn combinations_skip_choices_a_whole_expression_candidate_absorbs() { let inventory = inventory("topk by (job) (10, count_over_time(m[1m]))"); let topk = topk_target(&inventory); let target = &inventory.targets[topk]; - let Some(inner) = target.absorbs else { - panic!("absorbs the count: {:?}", target.choices.candidates) + let Some(fused) = &target.fused_target else { + panic!("a fused target: {:?}", target.choices.candidates) }; + let inner = fused.absorbs.expect("absorbs the count"); let n = target.choices.candidates.len(); - let whole = target - .choices - .candidates - .iter() - .filter(|c| matches!(c, ComputationCandidate::WholeExpression(_))) - .count(); + let whole = fused.choices.candidates.len(); let m = inventory.targets[inner].choices.candidates.len(); let mut enumerated = 0; let mut choice = vec![0; inventory.targets.len()]; @@ -4181,7 +4242,7 @@ mod tests { let mut carry = true; for (digit, target) in choice.iter_mut().zip(&inventory.targets).rev() { *digit += 1; - if *digit < target.choices.candidates.len() { + if *digit < target.choice_count() { carry = false; break; } @@ -4192,7 +4253,8 @@ mod tests { } } assert!(whole > 0 && m > 1, "{n} {whole} {m}"); - assert_eq!(inventory.combinations(), (n - whole) * m + whole); + assert_eq!(inventory.combinations(), n * m + whole); + assert_eq!(inventory.combinations(), 6); assert_eq!(inventory.combinations(), enumerated); } diff --git a/crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs b/crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs index 2094d7a19..3682609e2 100644 --- a/crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs +++ b/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() diff --git a/crates/asap-aware-mapping/src/stage3_plan_selection.rs b/crates/asap-aware-mapping/src/stage3_plan_selection.rs index fc0f4a9d4..e3d3eab4c 100644 --- a/crates/asap-aware-mapping/src/stage3_plan_selection.rs +++ b/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| { - (0..target.choices.candidates.len()).map(move |index| { + (0..target.choice_count()).map(move |index| { let mut choice = choice.clone(); choice.push(index); choice diff --git a/crates/asap-aware-mapping/tests/per_computation_candidates.rs b/crates/asap-aware-mapping/tests/per_computation_candidates.rs index 25231a576..c5da75d73 100644 --- a/crates/asap-aware-mapping/tests/per_computation_candidates.rs +++ b/crates/asap-aware-mapping/tests/per_computation_candidates.rs @@ -1,7 +1,7 @@ //! Frontend-to-Pass-1 acceptance: candidate discovery precedes empirical selection. use asap_aware_mapping::per_computation_candidates::{ computation_candidates_for_intent, generate_per_computation_candidates, ComputationCandidate, - HavingEntry, HavingPlacement, PerComputationCandidateError, PruneReason, + ExactState, HavingEntry, HavingPlacement, PerComputationCandidateError, PruneReason, }; use asap_types::{ ir::operator_properties::{Reduction, Source}, @@ -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,18 @@ 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![ + ExactState::of(ExactKind::Sum) + ])], + vec![ComputationCandidate::ExactAccumulator(vec![ + ExactState::of(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 +408,9 @@ 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![ + ExactState::of(ExactKind::Sum) + ])] ); assert_eq!( algorithms(&candidates[1].choices.candidates), @@ -510,7 +506,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 +571,10 @@ fn avg_rewrites_to_sum_and_count_states() { .targets; assert_eq!( targets[0].choices.candidates, - vec![ComputationCandidate::AvgAsSumCount] + vec![ComputationCandidate::ExactAccumulator(vec![ + ExactState::of(ExactKind::Sum), + ExactState::of(ExactKind::Count) + ])] ); assert_eq!( targets[1].choices.candidates, @@ -584,7 +583,10 @@ fn avg_rewrites_to_sum_and_count_states() { assert_eq!( targets[1].choices.pruned, vec![( - ComputationCandidate::AvgAsSumCount, + ComputationCandidate::ExactAccumulator(vec![ + ExactState::of(ExactKind::Sum), + ExactState::of(ExactKind::Count) + ]), PruneReason::NullableAvgInput )] ); @@ -692,10 +694,9 @@ 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![ + ExactState::of(ExactKind::Sum) + ])] && choices.pruned.is_empty())); assert_eq!(found.len(), 2, "{found:#?}"); } @@ -707,7 +708,7 @@ fn having_of( ) -> Vec { 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 +741,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); diff --git a/crates/devtools/src/candidate_labels.rs b/crates/devtools/src/candidate_labels.rs index cc8ca737f..564e6d3ad 100644 --- a/crates/devtools/src/candidate_labels.rs +++ b/crates/devtools/src/candidate_labels.rs @@ -59,46 +59,52 @@ pub fn choice_label(inventory: &PerComputationCandidates, 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 = inventory + // Each target's chosen candidate, with the target that computes it: the + // target itself or its fused target. + 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.chosen(index).expect("a listed candidate")) + .collect(); + // Targets a chosen fused target absorbs are not computed. + let absorbed: Vec = chosen + .iter() + .filter_map(|(computing, _)| computing.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, ((computing, candidate), &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(states) => { + // e.g. "Sum acc", "Sum/Count acc" for an avg, or + // "nested Sum acc" for a fused nested sum. + let kinds: Vec = states + .iter() + .map(|state| format!("{:?}", state.kind)) + .collect(); + let nested = if computing.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 +504,50 @@ mod tests { assert_eq!(distinct.len(), logical.len(), "{logical:#?}"); } + /// A fused target'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_targets_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::>() + }; + 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] diff --git a/crates/frontend-sql/tests/frequency_l2.rs b/crates/frontend-sql/tests/frequency_l2.rs index cc26b2a9e..e8304bb59 100644 --- a/crates/frontend-sql/tests/frequency_l2.rs +++ b/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:?}" ); diff --git a/crates/integration-tests/tests/stage1_logical_asap_optimization.rs b/crates/integration-tests/tests/stage1_logical_asap_optimization.rs index 3a89967ad..41d973748 100644 --- a/crates/integration-tests/tests/stage1_logical_asap_optimization.rs +++ b/crates/integration-tests/tests/stage1_logical_asap_optimization.rs @@ -8,6 +8,7 @@ use std::sync::Arc; use asap_aware_mapping::per_computation_candidates::{ compose_logical_candidate, generate_per_computation_candidates, ComputationCandidate, + ExactState, }; use asap_frontend_sql::{lower_sql, SqlCatalog}; use asap_physical_operators::{ @@ -18,7 +19,9 @@ use asap_physical_operators::{ }; use asap_types::ir::physical_export::PhysicalASAPOperatorPayload; use asap_types::ir::{ASAPOp, NonASAPOp, OperatorNode, QueryRoot, ScalarExpr}; -use asap_types::post_asap::{FieldDataType, SketchAlgorithm, SummaryInputExpr, SummaryUpdate}; +use asap_types::post_asap::{ + ExactKind, FieldDataType, SketchAlgorithm, SummaryInputExpr, SummaryUpdate, +}; use asap_types::pre_asap::schema::{DataType, Field, Schema}; use asap_types::pre_asap::{ColumnRef, CompareOpKind, ScalarValue}; use asap_types::types::AccuracyTarget; @@ -103,7 +106,7 @@ fn compose_all(label: &str, root: QueryRoot) -> usize { let radix: Vec<_> = inventory .targets .iter() - .map(|target| target.choices.candidates.len()) + .map(|target| target.choice_count()) .collect(); let total: usize = radix.iter().product(); let mut choice = vec![0; radix.len()]; @@ -168,7 +171,7 @@ fn composed_estimates(root: QueryRoot) -> Vec> { .iter() .enumerate() .flat_map(|(target, computation)| { - (1..computation.choices.candidates.len()).map(move |index| { + (1..computation.choice_count()).map(move |index| { let mut choice = vec![0; targets]; choice[target] = index; choice @@ -255,7 +258,7 @@ async fn subquery_in_a_measure_filter_is_substituted() { .iter() .position(|target| { target - .target + .aggregate .non_asap() .and_then(|op| match op { NonASAPOp::Aggregate { reduction, .. } => reduction.group_keys(), @@ -410,7 +413,12 @@ async fn avg_as_sum_and_count( .choices .candidates .iter() - .position(|c| *c == ComputationCandidate::AvgAsSumCount) + .position(|c| { + *c == ComputationCandidate::ExactAccumulator(vec![ + ExactState::of(ExactKind::Sum), + ExactState::of(ExactKind::Count), + ]) + }) .expect("avg as sum and count") }) .collect(); @@ -532,7 +540,10 @@ fn promql_avg_divides_sum_and_count_vectors() { generate_per_computation_candidates(vec![(0, promql("avg by (job) (m)"))]).unwrap(); assert_eq!( inventory.targets[0].choices.candidates[0], - ComputationCandidate::AvgAsSumCount + ComputationCandidate::ExactAccumulator(vec![ + ExactState::of(ExactKind::Sum), + ExactState::of(ExactKind::Count) + ]) ); let composed = compose_logical_candidate(&inventory, &[0]).unwrap(); let QueryRoot::Operator(root) = &composed[0].1 else { diff --git a/crates/integration-tests/tests/stage2_physical_asap_optimization_sql.rs b/crates/integration-tests/tests/stage2_physical_asap_optimization_sql.rs index fdd6b39bf..c74e192b5 100644 --- a/crates/integration-tests/tests/stage2_physical_asap_optimization_sql.rs +++ b/crates/integration-tests/tests/stage2_physical_asap_optimization_sql.rs @@ -82,7 +82,7 @@ async fn check_every_choice( choices = choices .into_iter() .flat_map(|prefix: Vec| { - (0..target.choices.candidates.len()).map(move |i| { + (0..target.choice_count()).map(move |i| { let mut choice = prefix.clone(); choice.push(i); choice @@ -99,11 +99,12 @@ async fn check_every_choice( // that candidate. Every other choice must compose. Err(e) => { let summarized_topk = inventory.targets.iter().zip(&choice).any(|(t, &i)| { - let NonASAPOp::Aggregate { measures, .. } = t.target.expect_non_asap() else { + let NonASAPOp::Aggregate { measures, .. } = t.aggregate.expect_non_asap() + else { return false; }; matches!(measures[t.measure], AggIntent::TopK { .. }) - && t.choices.candidates[i] != ComputationCandidate::Exact + && t.chosen(i).map(|(_, c)| c) != Some(&ComputationCandidate::Exact) }); assert!(summarized_topk, "{sql}, choice {choice:?}: {e}"); continue; diff --git a/crates/planner/tests/plan_stages.rs b/crates/planner/tests/plan_stages.rs index d05d6e4c4..33d50d249 100644 --- a/crates/planner/tests/plan_stages.rs +++ b/crates/planner/tests/plan_stages.rs @@ -56,7 +56,7 @@ fn enumerates_every_combination_and_selects_one() { .stage1_pass1_per_computation .targets .iter() - .map(|target| target.choices.candidates.len()) + .map(|target| target.choice_count()) .collect(); assert_eq!(sizes.len(), 2); assert_eq!(run.combinations, sizes[0] * sizes[1]); From 6eec1863b6a07de98504b8c9da1b3cefc8a80c38 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Fri, 16 Oct 2026 02:29:55 +0000 Subject: [PATCH 2/4] docs(planner-layering): describe fused Pass 1 targets --- docs/design_docs/proposals/planner-layering.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/docs/design_docs/proposals/planner-layering.md b/docs/design_docs/proposals/planner-layering.md index 7fb4e24ce..d70713d35 100644 --- a/docs/design_docs/proposals/planner-layering.md +++ b/docs/design_docs/proposals/planner-layering.md @@ -304,6 +304,16 @@ 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 a +**fused target** of its own, which absorbs the target beneath: 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 +target's Count-Min or Count Sketch with a top-*k* heap. Like every target's +options, a fused target's are exact or a summary. + Example for summary candidates: | Original computation | Per-computation candidates | From 59fe7a8d2b9abf8bad5c09be53e319f4c9179da9 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Fri, 16 Oct 2026 02:46:45 +0000 Subject: [PATCH 3/4] refactor(planner)!: ExactAccumulator holds accumulator kinds; delete ExactState ExactState.params only mirrored kind (no exact accumulator has parameters), so ExactAccumulator now holds Vec and ExactState and ExactState::of are deleted. ExactParams stays: it is part of FieldDataType::ExactAggregate's serialized IR JSON, which the stage viewer reads. --- .../src/per_computation_candidates.rs | 78 ++++++++----------- .../tests/per_computation_candidates.rs | 25 ++---- crates/devtools/src/candidate_labels.rs | 6 +- .../tests/stage1_logical_asap_optimization.rs | 10 +-- 4 files changed, 45 insertions(+), 74 deletions(-) diff --git a/crates/asap-aware-mapping/src/per_computation_candidates.rs b/crates/asap-aware-mapping/src/per_computation_candidates.rs index dadd3c43c..ccff74d35 100644 --- a/crates/asap-aware-mapping/src/per_computation_candidates.rs +++ b/crates/asap-aware-mapping/src/per_computation_candidates.rs @@ -137,40 +137,28 @@ 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, - /// One or more exact mergeable accumulator states, finalized and - /// combined into the measure: one state for `sum`, `count`, `min`, - /// `max`, `rate`, `irate` and `increase`; `[Sum, Count]`, divided, for - /// `avg`. - ExactAccumulator(Vec), + /// One or more exact mergeable accumulator states, by accumulator kind, + /// finalized and combined into the measure: one state for `sum`, + /// `count`, `min`, `max`, `rate`, `irate` and `increase`; `[Sum, Count]`, + /// divided, for `avg`. + ExactAccumulator(Vec), /// An unsized summary family, sized later for the measure's accuracy. Summary(SketchAlgorithm), } -/// One exact mergeable accumulator state: which accumulator, and its -/// parameters (`FieldDataType::ExactAggregate`'s pair). -#[derive(Debug, Clone, PartialEq)] -pub struct ExactState { - /// Which accumulator: `Sum`, `Count`, `Min`, `Max`, `Rate`, `IRate` or - /// `Increase`. - pub kind: ExactKind, - /// Its parameters; none of them has any, so this mirrors `kind`. - pub params: ExactParams, -} - -impl ExactState { - /// The state of accumulator `kind`, with its (empty) parameters. - pub fn of(kind: ExactKind) -> Self { - let params = match kind { - ExactKind::Sum => ExactParams::Sum, - ExactKind::Count => ExactParams::Count, - ExactKind::Min => ExactParams::Min, - ExactKind::Max => ExactParams::Max, - ExactKind::Rate => ExactParams::Rate, - ExactKind::IRate => ExactParams::IRate, - ExactKind::Increase => ExactParams::Increase, - }; - Self { kind, params } - } +/// `FieldDataType::ExactAggregate` of accumulator `kind`, with its parameters +/// (none has any, so they mirror `kind`). +fn exact_family(kind: ExactKind) -> FieldDataType { + let params = match kind { + ExactKind::Sum => ExactParams::Sum, + ExactKind::Count => ExactParams::Count, + ExactKind::Min => ExactParams::Min, + ExactKind::Max => ExactParams::Max, + ExactKind::Rate => ExactParams::Rate, + ExactKind::IRate => ExactParams::IRate, + ExactKind::Increase => ExactParams::Increase, + }; + FieldDataType::ExactAggregate(kind, params) } /// Why Pass 1 proved a candidate unable to realize its target. @@ -404,7 +392,7 @@ pub fn computation_candidates_for_intent( let exact = if states.is_empty() { ComputationCandidate::Exact } else { - ComputationCandidate::ExactAccumulator(states.into_iter().map(ExactState::of).collect()) + ComputationCandidate::ExactAccumulator(states) }; let mut choices = ComputationChoices { candidates: vec![exact], @@ -645,7 +633,7 @@ fn add_nested_sum_targets( continue; } absorbed.insert(inner); - let merge = ComputationCandidate::ExactAccumulator(vec![ExactState::of(ExactKind::Sum)]); + let merge = ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum]); targets[t].fused_target = Some(Box::new(fused_target(&targets[t], inner, vec![merge]))); } } @@ -2657,8 +2645,7 @@ pub(crate) fn exact_accumulator( input: &Rc, reduction: Reduction, ) -> Result, PerComputationCandidateError> { - let ExactState { kind, params } = ExactState::of(kind); - let family = FieldDataType::ExactAggregate(kind, params); + let family = exact_family(kind); let read = measure_input(intent, input, None)?; Ok(OperatorNode::new_shared(Operator::ASAP( ASAPOp::SummaryAgg { @@ -2700,9 +2687,8 @@ fn substitute_measure( }, ))?) }; - let finalized = |exact: &ExactState, intent: &AggIntent, read| { - let family = FieldDataType::ExactAggregate(exact.kind.clone(), exact.params.clone()); - let state = state(family, intent, read)?; + let finalized = |kind: &ExactKind, intent: &AggIntent, read| { + let state = state(exact_family(kind.clone()), intent, read)?; Ok::<_, PerComputationCandidateError>(OperatorNode::new_shared(Operator::ASAP( ASAPOp::FinalizeExactAccumulator { child: state }, ))?) @@ -2718,9 +2704,9 @@ fn substitute_measure( having: None, }))? } - ComputationCandidate::ExactAccumulator(states) if states.len() == 1 => { + ComputationCandidate::ExactAccumulator(kinds) if kinds.len() == 1 => { let read = measure_input(intent, input, filter.as_ref())?; - finalized(&states[0], intent, read)? + finalized(&kinds[0], intent, read)? } ComputationCandidate::Summary(algorithm) => { let family = FieldDataType::Sketch( @@ -2738,8 +2724,8 @@ fn substitute_measure( query: statistic(intent)?, }))? } - ComputationCandidate::ExactAccumulator(states) => { - let (AggIntent::Avg { col }, [sum_state, count_state]) = (intent, states.as_slice()) + ComputationCandidate::ExactAccumulator(kinds) => { + let (AggIntent::Avg { col }, [sum_kind, count_kind]) = (intent, kinds.as_slice()) else { return Err(PerComputationCandidateError::Unsupported( "several exact states other than avg's sum and count", @@ -2750,9 +2736,9 @@ fn substitute_measure( // non-null (a nullable input prunes this candidate), so it counts // the rows the sum reads. let read = measure_input(&sum_intent, input, filter.as_ref())?; - let mut sum = finalized(sum_state, &sum_intent, read.clone())?; + let mut sum = finalized(sum_kind, &sum_intent, read.clone())?; let count = finalized( - count_state, + count_kind, &AggIntent::Count { accuracy: AccuracyTarget::exact(), }, @@ -3230,7 +3216,7 @@ mod tests { assert!(!choices.contains(&ComputationCandidate::Exact)); assert_eq!( choices[0], - ComputationCandidate::ExactAccumulator(vec![ExactState::of(ExactKind::Count)]) + ComputationCandidate::ExactAccumulator(vec![ExactKind::Count]) ); assert!(choices.contains(&ComputationCandidate::Summary(SketchAlgorithm::UnivMon))); } @@ -4081,7 +4067,7 @@ mod tests { /// The target whose fused target is a nested-sum merge. fn nested_sum_target(inventory: &PerComputationCandidates) -> Option { - let merge = ComputationCandidate::ExactAccumulator(vec![ExactState::of(ExactKind::Sum)]); + let merge = ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum]); inventory.targets.iter().position(|t| { t.fused_target .as_ref() @@ -4099,7 +4085,7 @@ mod tests { let inventory = inventory("sum by (job) (sum_over_time(m[5m]))"); let outer = nested_sum_target(&inventory).expect("a nested-sum merge"); let target = &inventory.targets[outer]; - let sum = ComputationCandidate::ExactAccumulator(vec![ExactState::of(ExactKind::Sum)]); + let sum = ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum]); assert_eq!(target.choices.candidates, std::slice::from_ref(&sum)); let fused = target.fused_target.as_deref().expect("a fused target"); assert_eq!(fused.choices.candidates, [sum]); diff --git a/crates/asap-aware-mapping/tests/per_computation_candidates.rs b/crates/asap-aware-mapping/tests/per_computation_candidates.rs index c5da75d73..7df06f5be 100644 --- a/crates/asap-aware-mapping/tests/per_computation_candidates.rs +++ b/crates/asap-aware-mapping/tests/per_computation_candidates.rs @@ -1,7 +1,7 @@ //! Frontend-to-Pass-1 acceptance: candidate discovery precedes empirical selection. use asap_aware_mapping::per_computation_candidates::{ computation_candidates_for_intent, generate_per_computation_candidates, ComputationCandidate, - ExactState, HavingEntry, HavingPlacement, PerComputationCandidateError, PruneReason, + HavingEntry, HavingPlacement, PerComputationCandidateError, PruneReason, }; use asap_types::{ ir::operator_properties::{Reduction, Source}, @@ -224,11 +224,9 @@ fn example_1_exact_q1_keeps_exact_accumulators_only() { assert_eq!( found, vec![ + vec![ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum])], vec![ComputationCandidate::ExactAccumulator(vec![ - ExactState::of(ExactKind::Sum) - ])], - vec![ComputationCandidate::ExactAccumulator(vec![ - ExactState::of(ExactKind::Rate) + ExactKind::Rate ])], ] ); @@ -408,9 +406,7 @@ fn multi_measure_aggregate_has_one_target_per_measure() { ); assert_eq!( candidates[0].choices.candidates, - vec![ComputationCandidate::ExactAccumulator(vec![ - ExactState::of(ExactKind::Sum) - ])] + vec![ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum])] ); assert_eq!( algorithms(&candidates[1].choices.candidates), @@ -572,8 +568,8 @@ fn avg_rewrites_to_sum_and_count_states() { assert_eq!( targets[0].choices.candidates, vec![ComputationCandidate::ExactAccumulator(vec![ - ExactState::of(ExactKind::Sum), - ExactState::of(ExactKind::Count) + ExactKind::Sum, + ExactKind::Count ])] ); assert_eq!( @@ -583,10 +579,7 @@ fn avg_rewrites_to_sum_and_count_states() { assert_eq!( targets[1].choices.pruned, vec![( - ComputationCandidate::ExactAccumulator(vec![ - ExactState::of(ExactKind::Sum), - ExactState::of(ExactKind::Count) - ]), + ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum, ExactKind::Count]), PruneReason::NullableAvgInput )] ); @@ -694,9 +687,7 @@ fn example_1_exact_q2_prunes_topk_heaps() { ] ); assert!(found.iter().any(|choices| choices.candidates - == [ComputationCandidate::ExactAccumulator(vec![ - ExactState::of(ExactKind::Sum) - ])] + == [ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum])] && choices.pruned.is_empty())); assert_eq!(found.len(), 2, "{found:#?}"); } diff --git a/crates/devtools/src/candidate_labels.rs b/crates/devtools/src/candidate_labels.rs index 564e6d3ad..d3b4f45a7 100644 --- a/crates/devtools/src/candidate_labels.rs +++ b/crates/devtools/src/candidate_labels.rs @@ -93,10 +93,8 @@ pub fn choice_label(inventory: &PerComputationCandidates, choice: &[usiz ComputationCandidate::ExactAccumulator(states) => { // e.g. "Sum acc", "Sum/Count acc" for an avg, or // "nested Sum acc" for a fused nested sum. - let kinds: Vec = states - .iter() - .map(|state| format!("{:?}", state.kind)) - .collect(); + let kinds: Vec = + states.iter().map(|kind| format!("{kind:?}")).collect(); let nested = if computing.absorbs.is_some() { "nested " } else { diff --git a/crates/integration-tests/tests/stage1_logical_asap_optimization.rs b/crates/integration-tests/tests/stage1_logical_asap_optimization.rs index 41d973748..7f377b2f0 100644 --- a/crates/integration-tests/tests/stage1_logical_asap_optimization.rs +++ b/crates/integration-tests/tests/stage1_logical_asap_optimization.rs @@ -8,7 +8,6 @@ use std::sync::Arc; use asap_aware_mapping::per_computation_candidates::{ compose_logical_candidate, generate_per_computation_candidates, ComputationCandidate, - ExactState, }; use asap_frontend_sql::{lower_sql, SqlCatalog}; use asap_physical_operators::{ @@ -415,8 +414,8 @@ async fn avg_as_sum_and_count( .iter() .position(|c| { *c == ComputationCandidate::ExactAccumulator(vec![ - ExactState::of(ExactKind::Sum), - ExactState::of(ExactKind::Count), + ExactKind::Sum, + ExactKind::Count, ]) }) .expect("avg as sum and count") @@ -540,10 +539,7 @@ fn promql_avg_divides_sum_and_count_vectors() { generate_per_computation_candidates(vec![(0, promql("avg by (job) (m)"))]).unwrap(); assert_eq!( inventory.targets[0].choices.candidates[0], - ComputationCandidate::ExactAccumulator(vec![ - ExactState::of(ExactKind::Sum), - ExactState::of(ExactKind::Count) - ]) + ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum, ExactKind::Count]) ); let composed = compose_logical_candidate(&inventory, &[0]).unwrap(); let QueryRoot::Operator(root) = &composed[0].1 else { From 56f761415a88e0f16118f80fcaa97ea3f96a6bb5 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Fri, 16 Oct 2026 03:43:31 +0000 Subject: [PATCH 4/4] refactor(planner)!: fused computations as FusedComputation; computation_for Replace ComputationTarget::fused_target (a boxed, cloned target) and ComputationTarget::absorbs with fused: Option, so a fused computation can no longer carry its own fused computation or a missing absorbed index. ComputationTarget::chosen becomes computation_for(index), returning the candidate and the target it absorbs; composition and labels take aggregate, measure and samples from the outer target. Also refuse avg exact states other than [Sum, Count] in that order (regression test avg_states_out_of_order_are_refused), document ComputationTarget::choices and PerComputationCandidates::roots, and fix compose_logical_candidate's doc on what choice[i] indexes. --- .../asap-aware-mapping/src/asap_aware_cse.rs | 2 +- .../src/pass/stage1_to_stage3.rs | 12 +- .../src/per_computation_candidates.rs | 327 ++++++++++-------- crates/devtools/src/candidate_labels.rs | 29 +- .../stage2_physical_asap_optimization_sql.rs | 3 +- .../design_docs/proposals/planner-layering.md | 8 +- 6 files changed, 204 insertions(+), 177 deletions(-) diff --git a/crates/asap-aware-mapping/src/asap_aware_cse.rs b/crates/asap-aware-mapping/src/asap_aware_cse.rs index edb489010..5caee47f1 100644 --- a/crates/asap-aware-mapping/src/asap_aware_cse.rs +++ b/crates/asap-aware-mapping/src/asap_aware_cse.rs @@ -2420,7 +2420,7 @@ mod tests { .iter() .map(|t| { (0..t.choice_count()) - .position(|i| t.chosen(i).is_some_and(|(_, c)| heap(c))) + .position(|i| t.computation_for(i).is_some_and(|(c, _)| heap(c))) .unwrap_or(0) }) .collect(); diff --git a/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs b/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs index fd30a4071..3df6cecb3 100644 --- a/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs +++ b/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs @@ -66,8 +66,8 @@ use crate::stage3_plan_selection::{select_plan, Selection, SelectionError}; use crate::PlanningModels; /// One whole-workload CandidateLogicalASAPDAG: `choice[t]` picks target -/// `t`'s candidate, or its fused target's (`targets[t].chosen(choice[t])`), -/// of the Pass 1 inventory. `roots` are the +/// `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 { @@ -186,9 +186,9 @@ pub fn run_stage1_to_stage3( /// One Stage 1 candidate in enumeration order, before Stage 2 numbers it. #[derive(Debug)] pub struct Stage1Candidate { - /// The Pass 1 choice (`targets[t].chosen(choice[t])`: target `t`'s - /// candidate, or its fused target's); 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, pub roots: Result, PerComputationCandidateError>, /// For a Pass 2 candidate: what it changed of its Pass 1 candidate. @@ -579,7 +579,7 @@ pub fn run_stage2_and_stage3( } /// Mixed-radix increment of `choice`, each digit over its target's -/// candidates and then its fused target's: the last target varies fastest. +/// 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(inventory: &PerComputationCandidates, choice: &mut [usize]) -> bool { for (digit, target) in choice.iter_mut().zip(&inventory.targets).rev() { diff --git a/crates/asap-aware-mapping/src/per_computation_candidates.rs b/crates/asap-aware-mapping/src/per_computation_candidates.rs index ccff74d35..ecba85ef5 100644 --- a/crates/asap-aware-mapping/src/per_computation_candidates.rs +++ b/crates/asap-aware-mapping/src/per_computation_candidates.rs @@ -14,11 +14,11 @@ //! its exact accumulator states (`ExactAccumulator`; `[Sum, Count]` for //! `avg`) when they can be built, otherwise plain `Exact`. Whether they are //! materialized is decided in Stage 2. -//! - A computation that fuses a target with the target beneath it is its own -//! target, which absorbs the one beneath -//! ([`ComputationTarget::fused_target`]): the whole-expression top-k over a -//! per-item `sum` or `count` (`Summary` of a heap family) and the nested -//! sum, a grouped `sum` over a `sum` (`ExactAccumulator([Sum])`). +//! - A computation that fuses a target with the target beneath it is the +//! outer target's [`FusedComputation`], which absorbs the one beneath and +//! has its own candidates: the whole-expression top-k over a per-item +//! `sum` or `count` (`Summary` of a heap family) and the nested sum, a +//! grouped `sum` over a `sum` (`ExactAccumulator([Sum])`). //! - Summary candidates are unsized; they are sized after Pass 2 sharing. //! - Candidates are pruned only when provably unable to meet the accuracy //! requirement, and each pruned candidate keeps its [`PruneReason`]. @@ -37,7 +37,7 @@ //! [`compose_logical_candidate`] is Algorithm 1's //! `SUBSTITUTE(d, s_1 ↦ c_1, …)`: it picks one per-computation candidate per //! target (`choice[i]` indexes `targets[i]`'s candidates, then its fused -//! target's, [`ComputationTarget::chosen`]) and builds one whole-workload +//! computation's, [`ComputationTarget::computation_for`]) and builds one whole-workload //! candidate DAG. An aggregate whose chosen candidates are all `Exact` is //! kept. Otherwise each of its measures is substituted on its own over the //! aggregate's input: @@ -51,7 +51,7 @@ //! - `Summary`: `SummaryAgg` → `SummaryEstimate`. //! - `Exact` (or a measure that is no target): a single-measure `Aggregate`. //! -//! A fused target's candidate reads the absorbed target's input instead: +//! A fused computation's candidate reads the absorbed target's input instead: //! the whole-expression top-k is one heap `SummaryAgg` keyed by the ranked //! item → `SummaryEstimate`; the nested sum is `SummaryAgg` with the exact //! `Sum` family grouped by the outer keys → `FinalizeExactAccumulator`. @@ -130,8 +130,8 @@ use crate::realization::{ /// intent has an exact mergeable state and it can be built, otherwise /// `Exact`. The plain form would duplicate "accumulator, not materialized"; /// Stage 2/3 decide materialization and sharing. Which computation is -/// computed is the target's ([`ComputationTarget`]); a fused computation is -/// its own target ([`ComputationTarget::fused_target`]). +/// computed is the target's ([`ComputationTarget`]) or its fused +/// computation's ([`FusedComputation`]). #[derive(Debug, Clone, PartialEq)] pub enum ComputationCandidate { /// Plain exact execution of the original sub-DAG; the exact option only @@ -229,15 +229,16 @@ pub struct HavingEntry { } /// One measure of an aggregate with at least one realization besides exact -/// execution, or a fused computation of it ([`Self::fused_target`]). The -/// target retains source, grouping, filters, input expressions and -/// evaluation context. +/// execution. The target retains source, grouping, filters, input +/// expressions and evaluation context. #[derive(Debug, Clone)] pub struct ComputationTarget { /// The `Aggregate` node the measure belongs to. pub aggregate: Rc, /// Index into the aggregate's measures. pub measure: usize, + /// The measure's unranked candidates, the exact option first, and the + /// candidates Pass 1 pruned with their reasons. pub choices: ComputationChoices, /// The measure's own filter; only rows that pass it update the state /// (`SummaryAgg.filter`). @@ -245,17 +246,10 @@ pub struct ComputationTarget { /// The aggregate's own `having`, then one entry per consumer `Filter` /// directly above the aggregate. pub having: Vec, - /// On a fused target only: the index (in - /// [`PerComputationCandidates::targets`]) of the target directly - /// beneath that it absorbs, which is not computed when the fused target - /// is chosen. `None` on every target of the inventory's list. - pub absorbs: Option, - /// The fused alternative of this target, computing the same measure - /// together with the target beneath it, which it absorbs: a - /// whole-expression top-k over the per-item `sum` or `count` beneath, or - /// the nested-sum merge of a `sum` over a `sum`. `None` when no fusion - /// applies, and on a fused target itself. - pub fused_target: Option>, + /// The fused computation of this measure together with the target + /// directly beneath it, offered beside this target's own candidates; + /// `None` when no fusion applies. + pub fused: Option, /// The metric whose raw samples, or plain sums of them, are every value /// the aggregate reads ([`samples_of_metric`]); `None` otherwise. A /// heap-sketch weight read from them carries the open requirement @@ -263,28 +257,42 @@ pub struct ComputationTarget { pub samples_of: Option, } +/// A target's measure computed together with the target directly beneath +/// it, which it absorbs (that target is not computed): the whole-expression +/// top-k over a per-item `sum` or `count` (`Summary` of a heap family), or +/// the nested sum, a grouped `sum` over a `sum` (`ExactAccumulator([Sum])`). +/// It has the outer target's aggregate, measure, filter and `HAVING`. +#[derive(Debug, Clone, PartialEq)] +pub struct FusedComputation { + /// The index in [`PerComputationCandidates::targets`] of the inner + /// target it absorbs. + pub absorbs: usize, + /// How the fused computation is computed: exact or a summary. + pub candidates: Vec, +} + impl ComputationTarget { /// How many values this target's digit of a whole-workload choice - /// takes: its own candidates, then its fused target's. + /// takes: its own candidates, then its fused computation's. pub fn choice_count(&self) -> usize { self.choices.candidates.len() + self - .fused_target + .fused .as_ref() - .map_or(0, |fused| fused.choices.candidates.len()) + .map_or(0, |fused| fused.candidates.len()) } - /// The target that computes the measure and its candidate when this - /// target's digit of a whole-workload choice is `index`: this target - /// and its own candidate `index`, or past them, its fused target and - /// that target's candidate. `None` past [`Self::choice_count`]. - pub fn chosen(&self, index: usize) -> Option<(&ComputationTarget, &ComputationCandidate)> { + /// How the measure is computed when this target's digit of a + /// whole-workload choice is `index`: the candidate, and the target it + /// absorbs when the candidate is the fused computation's (past this + /// target's own candidates). `None` past [`Self::choice_count`]. + pub fn computation_for(&self, index: usize) -> Option<(&ComputationCandidate, Option)> { let own = self.choices.candidates.len(); match self.choices.candidates.get(index) { - Some(candidate) => Some((self, candidate)), + Some(candidate) => Some((candidate, None)), None => { - let fused = self.fused_target.as_deref()?; - Some((fused, fused.choices.candidates.get(index - own)?)) + let fused = self.fused.as_ref()?; + Some((fused.candidates.get(index - own)?, Some(fused.absorbs))) } } } @@ -293,21 +301,24 @@ impl ComputationTarget { /// Compact Pass 1 inventory; roots and nested producer dependencies are retained. #[derive(Debug, Clone)] pub struct PerComputationCandidates { + /// The frontends' query roots with their ids, in workload order; whole- + /// workload candidates are these roots with chosen candidates substituted. pub roots: Vec<(Id, QueryRoot)>, - /// The unfused targets; a whole-workload choice has one digit per - /// target, ranging over [`ComputationTarget::choice_count`]. + /// The targets; a whole-workload choice has one digit per target, + /// ranging over [`ComputationTarget::choice_count`]. pub targets: Vec, } impl PerComputationCandidates { /// The number of whole-workload candidates: the size of the Cartesian /// product of the targets' candidates, where a target with a fused - /// target and the target beneath that it absorbs count as their own - /// pairs plus the fused target's candidates (the absorbed target is not - /// computed then), saturating at `usize::MAX`. An absorbed target has - /// one fused target over it, and a fused target is never absorbed, so - /// each such pair counts as `n · m + f` for a target with `n` - /// candidates over an absorbed target with `m`, and `f` fused ones. + /// computation and the target beneath that it absorbs count as their + /// own pairs plus the fused computation's candidates (the absorbed + /// target is not computed then), saturating at `usize::MAX`. An absorbed + /// target is absorbed by one fused computation, and a target with one is + /// never absorbed, so each such pair counts as `n · m + f` for a target + /// with `n` candidates over an absorbed target with `m`, and `f` fused + /// ones. pub fn combinations(&self) -> usize { let absorbed: HashSet = absorbed_targets(&self.targets).collect(); self.targets @@ -316,34 +327,33 @@ impl PerComputationCandidates { .filter(|(index, _)| !absorbed.contains(index)) .map(|(_, target)| { let n = target.choices.candidates.len(); - let Some(fused) = &target.fused_target else { + let Some(fused) = &target.fused else { return n; }; - let inner = fused.absorbs.expect("a fused target absorbs"); - n.saturating_mul(self.targets[inner].choices.candidates.len()) - .saturating_add(fused.choices.candidates.len()) + n.saturating_mul(self.targets[fused.absorbs].choices.candidates.len()) + .saturating_add(fused.candidates.len()) }) .fold(1usize, usize::saturating_mul) } - /// Whether `choice` picks a fused target and a candidate other than the - /// first for the target it absorbs: the same whole-workload candidate as - /// the one with the first, so enumeration skips it. + /// Whether `choice` picks a fused computation and a candidate other + /// than the first for the target it absorbs: the same whole-workload + /// candidate as the one with the first, so enumeration skips it. pub fn absorbed_choice_varies(&self, choice: &[usize]) -> bool { self.targets.iter().zip(choice).any(|(target, &index)| { target - .chosen(index) - .and_then(|(computing, _)| computing.absorbs) + .computation_for(index) + .and_then(|(_, absorbed)| absorbed) .is_some_and(|inner| choice[inner] != 0) }) } } -/// The indices of `targets` some target's fused target absorbs. +/// The indices of `targets` some target's fused computation absorbs. fn absorbed_targets(targets: &[ComputationTarget]) -> impl Iterator + '_ { targets .iter() - .filter_map(|target| target.fused_target.as_ref()?.absorbs) + .filter_map(|target| Some(target.fused.as_ref()?.absorbs)) } #[derive(Debug, Error)] @@ -580,8 +590,7 @@ pub fn generate_per_computation_candidates( choices, filter: filters.get(measure).cloned().flatten(), having: entries.clone(), - absorbs: None, - fused_target: None, + fused: None, samples_of: node.non_asap().and_then(|op| match op { NonASAPOp::Aggregate { child, .. } => samples_of_metric(child), _ => None, @@ -595,15 +604,16 @@ pub fn generate_per_computation_candidates( } /// Nested-sum merge: a grouped `sum` over the value of a `sum` target -/// beneath it gets a fused target ([`ComputationTarget::fused_target`]) +/// beneath it gets a fused computation ([`ComputationTarget::fused`]) /// with the one candidate `ExactAccumulator([Sum])`: one exact `Sum` /// accumulator over the inner sum's input, grouped by the outer keys (the /// sum of the inner groups' sums is the sum of their rows), absorbing the /// inner target. Applied only when the inner sum has no other consumer, so /// absorbing it removes its work, when neither has a filter or `HAVING` /// ([`nested_sum_input`]), and when the outer sum is not itself absorbed -/// and has no fused target yet, and the inner sum has none, so a fused -/// target is never absorbed and merges do not stack. +/// and has no fused computation yet, and the inner sum has none, so a +/// target with a fused computation is never absorbed and merges do not +/// stack. fn add_nested_sum_targets( targets: &mut [ComputationTarget], consumers: &HashMap<*const OperatorNode, Vec>>>, @@ -611,7 +621,7 @@ fn add_nested_sum_targets( let mut absorbed: HashSet = absorbed_targets(targets).collect(); for t in 0..targets.len() { let target = &targets[t]; - if absorbed.contains(&t) || target.fused_target.is_some() || !target.having.is_empty() { + if absorbed.contains(&t) || target.fused.is_some() || !target.having.is_empty() { continue; } let Some(NonASAPOp::Aggregate { child, .. }) = target.aggregate.non_asap() else { @@ -623,36 +633,21 @@ fn add_nested_sum_targets( else { continue; }; - // A fused target is never absorbed: the inner sum must not have a - // fused target of its own. + // A target with a fused computation is never absorbed: the inner sum + // must not have one of its own. if consumers.get(&Rc::as_ptr(child)).map(Vec::len) != Some(1) || !targets[inner].having.is_empty() - || targets[inner].fused_target.is_some() + || targets[inner].fused.is_some() || nested_sum_input(&target.aggregate).is_none() { continue; } absorbed.insert(inner); let merge = ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum]); - targets[t].fused_target = Some(Box::new(fused_target(&targets[t], inner, vec![merge]))); - } -} - -/// The fused target of `target` absorbing `targets[inner]`, with -/// `candidates`: the same measure, filter, `HAVING` and samples. -fn fused_target( - target: &ComputationTarget, - inner: usize, - candidates: Vec, -) -> ComputationTarget { - ComputationTarget { - choices: ComputationChoices { - candidates, - pruned: vec![], - }, - absorbs: Some(inner), - fused_target: None, - ..target.clone() + targets[t].fused = Some(FusedComputation { + absorbs: inner, + candidates: vec![merge], + }); } } @@ -727,13 +722,13 @@ fn nested_sum_input(target: &OperatorNode) -> Option<(Rc, AggInten } /// Whole-expression top-k (Example 1's Q2): a top-k over a per-item `sum` -/// or `count` that is itself a target gets a fused target -/// ([`ComputationTarget::fused_target`]) absorbing the inner target, whose +/// or `count` that is itself a target gets a fused computation +/// ([`ComputationTarget::fused`]) absorbing the inner target, whose /// candidates are the top-k's heap summaries (Count-Min and CountSketch /// with a heap), each one heap over the inner aggregate's input, keyed by /// the ranked item and weighted by the summed value. Those heaps over the /// exact per-item result leave the top-k's own candidates: the fused -/// target replaces them. Applied only when the inner aggregate has no other +/// computation replaces them. Applied only when the inner aggregate has no other /// consumer, so absorbing it removes its work, when neither has a filter or /// `HAVING`, and when the keyed-additive rule realizes it /// ([`whole_expression_input`]). @@ -777,10 +772,12 @@ fn add_whole_expression_targets( if heaps.is_empty() { continue; } - let fused = fused_target(&targets[t], inner, heaps); let target = &mut targets[t]; target.choices.candidates = own; - target.fused_target = Some(Box::new(fused)); + target.fused = Some(FusedComputation { + absorbs: inner, + candidates: heaps, + }); } } @@ -921,12 +918,14 @@ fn with_samples_requirement(mut update: SummaryUpdate, metric: Option<&str>) -> update } -/// Build one whole-workload candidate (end of Pass 1): `choice[i]` indexes -/// `inventory.targets[i].choices.candidates`. An aggregate with a chosen -/// candidate other than `Exact` is replaced measure by measure: an exact -/// accumulator by `SummaryAgg` → `FinalizeExactAccumulator`, a summary by -/// `SummaryAgg` → `SummaryEstimate`, `avg` by its `sum` and `count` -/// accumulators divided, and any other measure by its own single-measure +/// Build one whole-workload candidate (end of Pass 1): `choice[i]` picks +/// `inventory.targets[i]`'s candidate, or past its own candidates its fused +/// computation's ([`ComputationTarget::computation_for`]). An aggregate with +/// a chosen candidate other than `Exact` is replaced measure by measure: an +/// exact accumulator by `SummaryAgg` → `FinalizeExactAccumulator`, a summary +/// by `SummaryAgg` → `SummaryEstimate`, `avg` by its `sum` and `count` +/// accumulators divided, a fused computation by one state over the absorbed +/// target's input, and any other measure by its own single-measure /// aggregate; several measures are joined back on the group keys. A summary /// is sized here for its measure's accuracy requirement; Pass 1 leaves it /// unsized; several approximate inputs of one query share its budget @@ -943,20 +942,24 @@ pub fn compose_logical_candidate( } let mut chosen: Chosen = HashMap::new(); for (target, &index) in inventory.targets.iter().zip(choice) { - let (computing, candidate) = target - .chosen(index) + let (candidate, absorbs) = target + .computation_for(index) .ok_or(PerComputationCandidateError::InvalidChoice)?; chosen - .entry(Rc::as_ptr(&computing.aggregate)) + .entry(Rc::as_ptr(&target.aggregate)) .or_default() - .push((computing, candidate)); + .push(ChosenComputation { + target, + candidate, + absorbs, + }); } // Consumer `Filter`s whose key-only `HAVING` moves below a substituted // aggregate: each is replaced by its input. let pushed_below: HashSet<_> = chosen .values() .filter(|targets| substituted(targets)) - .flat_map(|targets| &targets[0].0.having) + .flat_map(|targets| &targets[0].target.having) .filter(|entry| entry.placement == HavingPlacement::BelowAggregate) .filter_map(|entry| entry.filter.as_ref().map(Rc::as_ptr)) .collect(); @@ -1206,17 +1209,27 @@ fn compose_roots( Ok(roots) } -/// Each aggregate's targets with their chosen candidates. -type Chosen<'a> = - HashMap<*const OperatorNode, Vec<(&'a ComputationTarget, &'a ComputationCandidate)>>; +/// One target's chosen computation in a whole-workload choice. +struct ChosenComputation<'a> { + /// The target whose measure it computes. + target: &'a ComputationTarget, + /// How it is computed. + candidate: &'a ComputationCandidate, + /// The target beneath it absorbs when the candidate is the target's fused + /// computation's; `None` for one of the target's own candidates. + absorbs: Option, +} + +/// Each aggregate's targets with their chosen computations. +type Chosen<'a> = HashMap<*const OperatorNode, Vec>>; type Memo = HashMap<*const OperatorNode, Rc>; /// Whether any measure of the aggregate is substituted by something other than /// its plain exact computation. -fn substituted(targets: &[(&ComputationTarget, &ComputationCandidate)]) -> bool { +fn substituted(targets: &[ChosenComputation<'_>]) -> bool { targets .iter() - .any(|(_, candidate)| **candidate != ComputationCandidate::Exact) + .any(|chosen| *chosen.candidate != ComputationCandidate::Exact) } fn rewrite( @@ -2447,7 +2460,7 @@ fn estimate_bound_over_samples( /// columns. fn substitute( target: &Rc, - targets: &[(&ComputationTarget, &ComputationCandidate)], + targets: &[ChosenComputation<'_>], sizing: &Sizing, memo: &Memo, ) -> Result, PerComputationCandidateError> { @@ -2465,7 +2478,7 @@ fn substitute( }; // A key-only `HAVING` is a row predicate over the input: rows of a // dropped group never reach the state. - let having = &targets[0].0.having; + let having = &targets[0].target.having; // Operator refs in a predicate (a subquery) are this aggregate's // children, already rewritten. let rewritten = |pred: &Predicate| { @@ -2489,21 +2502,18 @@ fn substitute( .iter() .enumerate() .map(|(measure, intent)| { - // `fused`: the candidate is a fused target's, which absorbs the - // target beneath. + // `fused`: the candidate is the target's fused computation's, + // which absorbs the target beneath. let (candidate, samples_of, fused) = targets .iter() - .find(|(target, _)| target.measure == measure) - .map_or( - (&ComputationCandidate::Exact, None, false), - |(target, candidate)| { - ( - *candidate, - target.samples_of.as_deref(), - target.absorbs.is_some(), - ) - }, - ); + .find(|chosen| chosen.target.measure == measure) + .map_or((&ComputationCandidate::Exact, None, false), |chosen| { + ( + chosen.candidate, + chosen.target.samples_of.as_deref(), + chosen.absorbs.is_some(), + ) + }); let filter = filters.get(measure).and_then(Option::as_ref).map(rewritten); let key = (Rc::as_ptr(target), measure); if fused { @@ -2513,7 +2523,7 @@ fn substitute( whole_expression(target, intent, algorithm, samples_of, (sizing, key), memo) } ComputationCandidate::Exact => Err(PerComputationCandidateError::Unsupported( - "plain exact execution of a fused target", + "plain exact execution of a fused computation", )), }; } @@ -2725,10 +2735,11 @@ fn substitute_measure( }))? } ComputationCandidate::ExactAccumulator(kinds) => { - let (AggIntent::Avg { col }, [sum_kind, count_kind]) = (intent, kinds.as_slice()) + let (AggIntent::Avg { col }, [ExactKind::Sum, ExactKind::Count]) = + (intent, kinds.as_slice()) else { return Err(PerComputationCandidateError::Unsupported( - "several exact states other than avg's sum and count", + "several exact states other than avg's sum then count", )); }; let sum_intent = AggIntent::Sum { col: *col }; @@ -2736,9 +2747,9 @@ fn substitute_measure( // non-null (a nullable input prunes this candidate), so it counts // the rows the sum reads. let read = measure_input(&sum_intent, input, filter.as_ref())?; - let mut sum = finalized(sum_kind, &sum_intent, read.clone())?; + let mut sum = finalized(&ExactKind::Sum, &sum_intent, read.clone())?; let count = finalized( - count_kind, + &ExactKind::Count, &AggIntent::Count { accuracy: AccuracyTarget::exact(), }, @@ -3869,7 +3880,7 @@ mod tests { generate_per_computation_candidates(vec![(0, QueryRoot::Operator(root))]).unwrap() } - /// Every candidate of target `t`, then of its fused target, composed + /// Every candidate of target `t`, then of its fused computation, composed /// with every other target at its first candidate, with the candidate. fn composed_with( inventory: &PerComputationCandidates, @@ -3883,7 +3894,7 @@ mod tests { let QueryRoot::Operator(root) = &composed[0].1 else { panic!("operator root") }; - let (_, candidate) = inventory.targets[t].chosen(c).unwrap(); + let (candidate, _) = inventory.targets[t].computation_for(c).unwrap(); (candidate.clone(), root.clone()) }) .collect() @@ -3918,9 +3929,9 @@ mod tests { } /// Example 1's Q2: the top-k over a per-series `sum_over_time` keeps - /// only its exact candidate and gets a fused whole-expression target - /// with a Count-Min and a CountSketch heap, which replace the heaps over - /// the exact sum. The fused target absorbs the inner sum: composed, no + /// only its exact candidate and gets a fused whole-expression + /// computation with a Count-Min and a CountSketch heap, which replace the + /// heaps over the exact sum. The fused computation absorbs the inner sum: composed, no /// aggregate is computed and the heap reads the range selector's raw /// samples. #[test] @@ -3929,17 +3940,15 @@ mod tests { let topk = topk_target(&inventory); let target = &inventory.targets[topk]; assert_eq!(target.choices.candidates, [ComputationCandidate::Exact]); - assert_eq!(target.absorbs, None); - let fused = target.fused_target.as_deref().expect("a fused target"); + let fused = target.fused.as_ref().expect("a fused computation"); assert_eq!( - fused.choices.candidates, + fused.candidates, [ ComputationCandidate::Summary(SketchAlgorithm::CmsWithHeap), ComputationCandidate::Summary(SketchAlgorithm::CountSketchWithHeap), ] ); - assert!(Rc::ptr_eq(&fused.aggregate, &target.aggregate)); - let inner = fused.absorbs.expect("absorbs the inner sum"); + let inner = fused.absorbs; assert_eq!(inventory.targets[inner].choices.candidates.len(), 1); assert_eq!(inventory.combinations(), 3); for (candidate, root) in composed_with(&inventory, topk) { @@ -3984,7 +3993,7 @@ mod tests { let mut heaps = 0; for (candidate, root) in composed_with(&inventory, topk) { // The top-k's own candidate is exact; its summaries are the - // fused target's. + // fused computation's. if !matches!(candidate, ComputationCandidate::Summary(_)) { continue; } @@ -4065,17 +4074,41 @@ mod tests { } } - /// The target whose fused target is a nested-sum merge. + /// An avg's exact states are its sum then its count: any other order + /// is refused rather than divided the wrong way round. + #[test] + fn avg_states_out_of_order_are_refused() { + let mut inventory = inventory("avg by (job) (m)"); + assert_eq!( + inventory.targets[0].choices.candidates, + [ComputationCandidate::ExactAccumulator(vec![ + ExactKind::Sum, + ExactKind::Count + ])] + ); + assert!(compose_logical_candidate(&inventory, &[0]).is_ok()); + inventory.targets[0].choices.candidates = + vec![ComputationCandidate::ExactAccumulator(vec![ + ExactKind::Count, + ExactKind::Sum, + ])]; + assert!(matches!( + compose_logical_candidate(&inventory, &[0]), + Err(PerComputationCandidateError::Unsupported(_)) + )); + } + + /// The target whose fused computation is a nested-sum merge. fn nested_sum_target(inventory: &PerComputationCandidates) -> Option { let merge = ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum]); inventory.targets.iter().position(|t| { - t.fused_target + t.fused .as_ref() - .is_some_and(|fused| fused.choices.candidates == [merge.clone()]) + .is_some_and(|fused| fused.candidates == [merge.clone()]) }) } - /// `sum by (job) (sum_over_time(m[5m]))` gets a fused nested-sum target + /// `sum by (job) (sum_over_time(m[5m]))` gets a fused nested sum /// beside its own accumulator: one exact Sum accumulator over the range's /// raw samples, grouped by `job`, which absorbs the inner per-series sum. /// Composed, no per-series state is built, the value is exact and the @@ -4087,9 +4120,9 @@ mod tests { let target = &inventory.targets[outer]; let sum = ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum]); assert_eq!(target.choices.candidates, std::slice::from_ref(&sum)); - let fused = target.fused_target.as_deref().expect("a fused target"); - assert_eq!(fused.choices.candidates, [sum]); - let inner = fused.absorbs.expect("absorbs the inner sum"); + let fused = target.fused.as_ref().expect("a fused computation"); + assert_eq!(fused.candidates, [sum]); + let inner = fused.absorbs; assert_eq!(inventory.targets[inner].choices.candidates.len(), 1); assert_eq!(inventory.combinations(), 2); let composed = composed_with(&inventory, outer); @@ -4167,7 +4200,7 @@ mod tests { } /// Merges do not stack: in `sum(sum by (job) (sum_over_time(m[5m])))` - /// a target with a fused target is never absorbed, and `combinations` + /// a target with a fused computation is never absorbed, and `combinations` /// counts exactly the choices enumeration keeps. #[test] fn nested_sum_merges_do_not_stack() { @@ -4176,8 +4209,8 @@ mod tests { assert_eq!(absorbed.len(), 1, "one merge"); for &a in &absorbed { assert!( - inventory.targets[a].fused_target.is_none(), - "an absorbed target has a fused target" + inventory.targets[a].fused.is_none(), + "an absorbed target has a fused computation" ); } let mut enumerated = 0; @@ -4212,12 +4245,12 @@ mod tests { let inventory = inventory("topk by (job) (10, count_over_time(m[1m]))"); let topk = topk_target(&inventory); let target = &inventory.targets[topk]; - let Some(fused) = &target.fused_target else { - panic!("a fused target: {:?}", target.choices.candidates) + let Some(fused) = &target.fused else { + panic!("a fused computation: {:?}", target.choices.candidates) }; - let inner = fused.absorbs.expect("absorbs the count"); + let inner = fused.absorbs; let n = target.choices.candidates.len(); - let whole = fused.choices.candidates.len(); + let whole = fused.candidates.len(); let m = inventory.targets[inner].choices.candidates.len(); let mut enumerated = 0; let mut choice = vec![0; inventory.targets.len()]; diff --git a/crates/devtools/src/candidate_labels.rs b/crates/devtools/src/candidate_labels.rs index d3b4f45a7..df11501c7 100644 --- a/crates/devtools/src/candidate_labels.rs +++ b/crates/devtools/src/candidate_labels.rs @@ -65,24 +65,21 @@ pub fn choice_label(inventory: &PerComputationCandidates, choice: &[usiz .expect("every target is reached from a root") }) .collect(); - // Each target's chosen candidate, with the target that computes it: the - // target itself or its fused target. + // 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) - .map(|(target, &index)| target.chosen(index).expect("a listed candidate")) - .collect(); - // Targets a chosen fused target absorbs are not computed. - let absorbed: Vec = chosen - .iter() - .filter_map(|(computing, _)| computing.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 = 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, ((computing, candidate), &target_query)) in + for (t, (&(candidate, absorbs), &target_query)) in chosen.iter().zip(&query_of_target).enumerate() { if target_query != query || absorbed.contains(&t) { @@ -90,16 +87,12 @@ pub fn choice_label(inventory: &PerComputationCandidates, choice: &[usiz } match candidate { ComputationCandidate::Exact => {} - ComputationCandidate::ExactAccumulator(states) => { + 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 = - states.iter().map(|kind| format!("{kind:?}")).collect(); - let nested = if computing.absorbs.is_some() { - "nested " - } else { - "" - }; + 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) => { @@ -502,11 +495,11 @@ mod tests { assert_eq!(distinct.len(), logical.len(), "{logical:#?}"); } - /// A fused target's candidate is labeled apart from the unfused pair's: + /// 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_targets_and_avg_states_have_their_own_labels() { + 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::{ diff --git a/crates/integration-tests/tests/stage2_physical_asap_optimization_sql.rs b/crates/integration-tests/tests/stage2_physical_asap_optimization_sql.rs index c74e192b5..0baa48678 100644 --- a/crates/integration-tests/tests/stage2_physical_asap_optimization_sql.rs +++ b/crates/integration-tests/tests/stage2_physical_asap_optimization_sql.rs @@ -104,7 +104,8 @@ async fn check_every_choice( return false; }; matches!(measures[t.measure], AggIntent::TopK { .. }) - && t.chosen(i).map(|(_, c)| c) != Some(&ComputationCandidate::Exact) + && t.computation_for(i).map(|(c, _)| c) + != Some(&ComputationCandidate::Exact) }); assert!(summarized_topk, "{sql}, choice {choice:?}: {e}"); continue; diff --git a/docs/design_docs/proposals/planner-layering.md b/docs/design_docs/proposals/planner-layering.md index d70713d35..6da1ca611 100644 --- a/docs/design_docs/proposals/planner-layering.md +++ b/docs/design_docs/proposals/planner-layering.md @@ -304,15 +304,15 @@ 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 a -**fused target** of its own, which absorbs the target beneath: the +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 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 -target's Count-Min or Count Sketch with a top-*k* heap. Like every target's -options, a fused target's are exact or a summary. +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. Example for summary candidates: