From 6c9a62946c1d09b9197cfa337c5b747b110f70c9 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:16:05 +0000 Subject: [PATCH 1/6] feat(planner): nested sum merge as a Pass 1 candidate (#680) A grouped sum over the value of a sum beneath it, such as sum by (job) (sum_over_time(m[5m])), gets ComputationCandidate::NestedSumMerge beside its own accumulator: one exact Sum accumulator over the inner sum's input, grouped by the outer keys, which absorbs the inner target like the whole-expression top-k does. Plan selection prices it against the nested accumulators. --- .../src/pass/stage1_to_stage3.rs | 10 +- .../src/per_computation_candidates.rs | 307 +++++++++++++++++- 2 files changed, 293 insertions(+), 24 deletions(-) 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 04fd41ae..c1a5ed77 100644 --- a/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs +++ b/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs @@ -631,12 +631,7 @@ fn display_name( .targets .iter() .zip(choice) - .filter(|(target, &index)| { - matches!( - target.choices.candidates[index], - ComputationCandidate::WholeExpression(_) - ) - }) + .filter(|(target, &index)| target.choices.candidates[index].absorbs_inner()) .filter_map(|(target, _)| target.absorbs) .collect(); (0..inventory.roots.len()) @@ -660,6 +655,9 @@ fn display_name( ComputationCandidate::AvgAsSumCount => { accumulators.push("Sum/Count acc".into()) } + ComputationCandidate::NestedSumMerge => { + accumulators.push("nested Sum acc".into()) + } ComputationCandidate::Summary(algorithm) | ComputationCandidate::WholeExpression(algorithm) => { sketches.push(sketch_name(algorithm)) diff --git a/crates/asap-aware-mapping/src/per_computation_candidates.rs b/crates/asap-aware-mapping/src/per_computation_candidates.rs index 7c3f0964..0fabc7b2 100644 --- a/crates/asap-aware-mapping/src/per_computation_candidates.rs +++ b/crates/asap-aware-mapping/src/per_computation_candidates.rs @@ -12,6 +12,8 @@ //! - 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. //! - 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`]. @@ -40,6 +42,9 @@ //! - `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`. //! //! The substituted output keeps the aggregate's column names. The measures @@ -138,6 +143,22 @@ pub enum ComputationCandidate { /// 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) + } } /// Why Pass 1 proved a candidate unable to realize its target. @@ -224,8 +245,8 @@ 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 - /// [`ComputationCandidate::WholeExpression`] candidates absorb. A + /// 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. @@ -246,13 +267,13 @@ pub struct PerComputationCandidates { impl PerComputationCandidates { /// The number of whole-workload candidates: the size of the Cartesian - /// product of the targets' candidates, without the choices that pick a - /// whole-expression candidate and anything but the first candidate of - /// the target it absorbs (the same candidate as with the first), + /// 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 - /// whole-expression, over an absorbed target with `m`. + /// absorbing, over an absorbed target with `m`. pub fn combinations(&self) -> usize { let absorbed: HashSet = self.targets.iter().filter_map(|t| t.absorbs).collect(); self.targets @@ -268,7 +289,7 @@ impl PerComputationCandidates { .choices .candidates .iter() - .filter(|c| matches!(c, ComputationCandidate::WholeExpression(_))) + .filter(|c| c.absorbs_inner()) .count(); (n - whole) .saturating_mul(self.targets[inner].choices.candidates.len()) @@ -277,16 +298,13 @@ impl PerComputationCandidates { .fold(1usize, usize::saturating_mul) } - /// Whether `choice` picks a whole-expression 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 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. pub fn absorbed_choice_varies(&self, choice: &[usize]) -> bool { self.targets.iter().zip(choice).any(|(target, &index)| { - matches!( - target.choices.candidates[index], - ComputationCandidate::WholeExpression(_) - ) && target.absorbs.is_some_and(|inner| choice[inner] != 0) + target.choices.candidates[index].absorbs_inner() + && target.absorbs.is_some_and(|inner| choice[inner] != 0) }) } } @@ -533,9 +551,107 @@ pub fn generate_per_computation_candidates( } } add_whole_expression_candidates(&mut targets, &consumers); + add_nested_sum_candidates(&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 above +/// it), so an absorbing target is never absorbed. +fn add_nested_sum_candidates( + targets: &mut [ComputationTarget], + consumers: &HashMap<*const OperatorNode, Vec>>>, +) { + let absorbed: HashSet = targets.iter().filter_map(|t| t.absorbs).collect(); + for t in 0..targets.len() { + let target = &targets[t]; + if absorbed.contains(&t) || target.absorbs.is_some() || !target.having.is_empty() { + continue; + } + let Some(NonASAPOp::Aggregate { child, .. }) = target.target.non_asap() else { + continue; + }; + let Some(inner) = targets + .iter() + .position(|other| Rc::ptr_eq(&other.target, child)) + else { + continue; + }; + if consumers.get(&Rc::as_ptr(child)).map(Vec::len) != Some(1) + || !targets[inner].having.is_empty() + || nested_sum_input(&target.target).is_none() + { + continue; + } + let target = &mut targets[t]; + target + .choices + .candidates + .push(ComputationCandidate::NestedSumMerge); + target.absorbs = Some(inner); + } +} + +/// The input rows, the inner sum and the outer grouping (moved onto the +/// input rows by column, [`partitions_over`]) of a nested-sum merge of +/// `target`, or `None` when it does not apply: `target` is a grouped `sum` +/// (not by `without`) of the value column of a per-series or grouped `sum` +/// directly beneath it, neither filtered nor with `HAVING`. +fn nested_sum_input(target: &OperatorNode) -> Option<(Rc, AggIntent, Reduction)> { + use asap_types::ir::non_asap::any_measure_filtered; + let Some(NonASAPOp::Aggregate { + child: inner, + reduction: reduction @ Reduction::Reduce(by), + measures, + filters, + having: None, + .. + }) = target.non_asap() + else { + return None; + }; + let [AggIntent::Sum { col }] = measures.as_slice() else { + return None; + }; + let Some(NonASAPOp::Aggregate { + child: rows, + reduction: inner_reduction, + measures: inner_measures, + filters: inner_filters, + having: None, + .. + }) = inner.non_asap() + else { + return None; + }; + let [inner_sum @ AggIntent::Sum { .. }] = inner_measures.as_slice() else { + return None; + }; + if by.is_without() || any_measure_filtered(filters) || any_measure_filtered(inner_filters) { + return None; + } + // The inner sum's value column: a per-series sum keeps its labels and + // carries the value as the sample value; a grouped one ends with it. + let value = match inner_reduction { + Reduction::PerEntity => resolve_column_ref(&ColumnRef::SampleValue, &inner.schema).ok()?, + Reduction::Reduce(keys) if !keys.is_without() => keys.len(), + Reduction::Reduce(_) => return None, + }; + let summed = match col { + None => resolve_column_ref(&ColumnRef::SampleValue, &inner.schema).ok()?, + Some(column) => *column, + }; + if summed != value || by.keys().contains(&value) { + return None; + } + let partitions = partitions_over(reduction, &inner.schema, &rows.schema)?; + 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 @@ -657,7 +773,11 @@ fn whole_expression_input( /// output), as indices into `input` (the rows the heap reads), matched by /// column. `None` when a key is missing or ambiguous in `input`, or the keys /// are given by `without`: `input` carries columns `inner` dropped. -fn partitions_over(reduction: &Reduction, inner: &Schema, input: &Schema) -> Option { +pub(crate) fn partitions_over( + reduction: &Reduction, + inner: &Schema, + input: &Schema, +) -> Option { let Reduction::Reduce(keys) = reduction else { return Some(Reduction::PerEntity); }; @@ -2288,6 +2408,9 @@ fn substitute( ); 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, @@ -2397,6 +2520,58 @@ fn whole_expression( ))?) } +/// A nested-sum merge for `target` ([`nested_sum_input`]): one exact `Sum` +/// accumulator over the absorbed sum's rewritten input, grouped as +/// `target`, finalized. +fn nested_sum( + target: &OperatorNode, + memo: &Memo, +) -> Result, PerComputationCandidateError> { + let (rows, intent, reduction) = nested_sum_input(target).ok_or( + PerComputationCandidateError::Unsupported("nested-sum merge input"), + )?; + let state = exact_accumulator( + ExactKind::Sum, + &intent, + &memo[&Rc::as_ptr(&rows)], + reduction, + )?; + Ok(OperatorNode::new_shared(Operator::ASAP( + ASAPOp::FinalizeExactAccumulator { child: state }, + ))?) +} + +/// An unfiltered exact accumulator of `kind` for `intent` over `input`, +/// grouped by `reduction`. +pub(crate) fn exact_accumulator( + kind: ExactKind, + intent: &AggIntent, + 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 family = FieldDataType::ExactAggregate(kind, params); + let read = measure_input(intent, input, None)?; + Ok(OperatorNode::new_shared(Operator::ASAP( + ASAPOp::SummaryAgg { + input: summary_update(intent, &family, &reduction, &input.schema, read)?, + child: Rc::clone(input), + family, + reduction, + grouping: GroupingStrategy::default(), + filter: None, + }, + ))?) +} + /// One measure of `target` over `input`: its group keys, then its value. fn substitute_measure( target: &OperatorNode, @@ -2462,9 +2637,9 @@ fn substitute_measure( query: statistic(intent)?, }))? } - ComputationCandidate::WholeExpression(_) => { + ComputationCandidate::WholeExpression(_) | ComputationCandidate::NestedSumMerge => { return Err(PerComputationCandidateError::Unsupported( - "whole-expression top-k is substituted with its aggregate", + "an absorbing candidate is substituted with its aggregate", )) } ComputationCandidate::AvgAsSumCount => { @@ -3803,6 +3978,102 @@ mod tests { } } + fn nested_sum_target(inventory: &PerComputationCandidates) -> Option { + inventory.targets.iter().position(|t| { + t.choices + .candidates + .contains(&ComputationCandidate::NestedSumMerge) + }) + } + + /// `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. + /// Composed, no per-series state is built, the value is exact and the + /// output columns are the outer sum's. + #[test] + fn nested_sum_merges_into_one_exact_accumulator() { + 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"); + assert_eq!(inventory.targets[inner].choices.candidates.len(), 1); + assert_eq!(inventory.combinations(), 2); + let composed = composed_with(&inventory, outer); + let (_, separate) = &composed[0]; + let (_, merged) = &composed[1]; + let states: Vec<_> = OperatorNode::reachable(merged) + .into_iter() + .filter(|n| matches!(n.asap(), Some(ASAPOp::SummaryAgg { .. }))) + .collect(); + let [state] = states.as_slice() else { + panic!("one accumulator: {states:?}") + }; + let Some(ASAPOp::SummaryAgg { + child, + family, + reduction, + .. + }) = state.asap() + else { + unreachable!() + }; + assert_eq!( + *family, + FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum) + ); + assert!(matches!( + child.non_asap(), + Some(NonASAPOp::TimeRange { .. }) + )); + let Reduction::Reduce(keys) = reduction else { + panic!("grouped by job: {reduction:?}") + }; + assert_eq!( + keys.keys() + .iter() + .map(|&k| child.schema.fields[k].name.as_str()) + .collect::>(), + ["job"] + ); + assert!(merged + .guarantee + .as_ref() + .is_some_and(AccuracyGuarantee::is_exact)); + let names = |node: &OperatorNode| { + node.schema + .fields + .iter() + .map(|f| f.name.clone()) + .collect::>() + }; + assert_eq!(names(merged), names(separate)); + merged.validate_structure().unwrap(); + } + + /// The merge needs a sum over a sum's value: an outer sum over a rate, + /// an outer count, or an outer sum over an inner `max` gets none. + #[test] + fn nested_sum_merge_needs_a_sum_of_sums() { + for query in [ + "sum by (job) (rate(m[5m]))", + "count by (job) (sum_over_time(m[5m]))", + "sum by (job) (max_over_time(m[5m]))", + ] { + assert_eq!(nested_sum_target(&inventory(query)), None, "{query}"); + } + } + /// `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 From 14011d3697aa23fa4f8a3066e5067f463149e3dc Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:16:05 +0000 Subject: [PATCH 2/6] feat(planner): Pass 2 rollup and exact top-k reuse rules (#680) Rollup answers a coarse exact accumulator (sum, count, min, max) from a finer one the candidate already computes over the same rows, as an exact accumulator over the finer finalized values grouped by the coarse keys. The family-capability table decides which families merge across groups; counter states, IRate and sketches do not roll up. Exact top-k reuse reads every exact top-k of one input and partition from one ranking, Limit(k_max) over Sort. Sketch heaps are not reused. Stage 1 lists both rules' candidates after the window-composition ones, each followed by its identical-expression candidates. --- .../asap-aware-mapping/src/asap_aware_cse.rs | 865 +++++++++++++++++- .../src/pass/stage1_to_stage3.rs | 224 ++++- 2 files changed, 1041 insertions(+), 48 deletions(-) diff --git a/crates/asap-aware-mapping/src/asap_aware_cse.rs b/crates/asap-aware-mapping/src/asap_aware_cse.rs index 6a975b13..82442281 100644 --- a/crates/asap-aware-mapping/src/asap_aware_cse.rs +++ b/crates/asap-aware-mapping/src/asap_aware_cse.rs @@ -4,7 +4,8 @@ //! //! Its three rules are implemented: the identical-expression rule, the //! summary-capability rule and, for tumbling panes of one query, the -//! window-composition rule. +//! window-composition rule; so are two reuse rules (#680), rollup and exact +//! top-k reuse. //! //! # Identical-expression rule //! @@ -62,6 +63,32 @@ //! the panes are rebuilt at each evaluation until Stage 2 keeps them. A pane //! width shared by several queries, sliding windows and Exponential //! Histograms are not implemented. +//! +//! # Rollup rule +//! +//! A coarse grouped computation can be answered from a finer one the +//! candidate already computes over the same rows: `sum by (job)` from +//! `sum by (job, instance)`. The rule never introduces a finer grouping +//! nobody asked for. A *rollup site* ([`rollup_sites`]) is a coarse exact +//! accumulator build whose family merges across disjoint groups in the +//! family-capability table with an exact merged value ([`rollup_kind`]: +//! sums, counts, minima and maxima; counter states merge only in time, and +//! a sketch's states would need a merge that regroups them, which the IR +//! does not have), and a finer build of the same family over the same rows +//! (a PromQL scan may project other labels), whose group keys include the +//! coarse ones. The coarse build becomes an exact accumulator over the +//! finer one's finalized values, grouped by the coarse keys (a count sums +//! the finer counts), so the value stays exact. [`rollup_variants`] +//! returns one candidate per non-empty subset of the sites, each read from +//! its coarsest finer build. +//! +//! # Exact top-k reuse rule +//! +//! Exact top-k computations of one input and partition with different `k` +//! read one ranking: the rows of the largest top-k, `Limit(k_max)` over +//! `Sort`, which hold every smaller top-k ([`topk_reuse_sites`], +//! [`topk_reuse_variants`]). Only exact computations over exact input are +//! reused; a sketch's heap is not, as its membership has no guarantee yet. use std::collections::HashMap; #[cfg(test)] @@ -69,12 +96,17 @@ use std::collections::HashSet; use std::rc::Rc; use asap_types::ir::cse::share_common_sub_dags; +use asap_types::ir::operator_properties::{Reduction, Source}; +use asap_types::ir::SortKey; use asap_types::ir::{ ASAPOp, NonASAPOp, Operator, OperatorNode, OperatorResultKind, QueryRoot, TimeRangeKind, }; +use asap_types::ir::{OperatorResultKind as ResultKind, ScalarExpr}; use asap_types::post_asap::{ - FamilyCapability, FieldDataType, GroupingStrategy, SketchAlgorithm, SketchKind, + ExactKind, FamilyCapability, FieldDataType, GroupingStrategy, MergedError, SelectionRelation, + SketchAlgorithm, SketchKind, }; +use asap_types::pre_asap::AggIntent; use crate::accuracy::estimators::shared_size_params; use crate::per_computation_candidates::{with_composed_guarantees, PerComputationCandidateError}; @@ -855,6 +887,556 @@ fn rebuild( Ok(out) } +/// One coarse build the rollup rule can answer from a finer one: the +/// coarse `SummaryAgg`, the finalized finer build it reads, and the value +/// that replaces the coarse build's finalized value. +#[derive(Debug, Clone)] +pub struct RollupSite { + pub coarse: Rc, + pub finer: Rc, + /// The finalized exact accumulator over `finer`'s values, grouped by + /// the coarse keys, with the coarse value's column names. + pub rollup: Rc, +} + +/// A Pass 2 candidate of the rollup rule: the candidate's roots with each +/// coarse build in `sites` (indices into [`rollup_sites`]) answered from its +/// finer build, and every guarantee composed again; the error when that +/// fails. +#[derive(Debug)] +pub struct RollupVariant { + pub sites: Vec, + pub roots: Result, PerComputationCandidateError>, +} + +/// The most sites of one reuse rule a candidate may have: every non-empty +/// subset is a candidate, about a million at 20. +pub const MAX_REUSE_SITES: usize = 20; + +/// The exact accumulator that merges finalized values of `family` across +/// disjoint groups, by the family-capability table: a family whose states +/// merge over disjoint (or overlapping) selections into an exact state. +/// `None` for a family without such a merge: counter states merge only +/// consecutive windows, and sketch states have no regrouping merge in the IR. +pub fn rollup_kind(family: &FieldDataType) -> Option { + let FieldDataType::ExactAggregate(kind, _) = family else { + return None; + }; + let merge = FamilyCapability::of(family).merge?; + if merge.inputs == SelectionRelation::DisjointInTime || merge.error != MergedError::Exact { + return None; + } + match kind { + // Counts add up. + ExactKind::Sum | ExactKind::Count => Some(ExactKind::Sum), + ExactKind::Min => Some(ExactKind::Min), + ExactKind::Max => Some(ExactKind::Max), + ExactKind::Rate | ExactKind::IRate | ExactKind::Increase => None, + } +} + +/// The rollup sites of the candidate with query roots `roots`, children +/// first: each coarse exact accumulator build read only through its +/// finalized value, with its coarsest finer build of the same family over +/// the same rows that the candidate finalizes. +pub fn rollup_sites(roots: &[QueryRoot]) -> Vec { + let mut nodes = Vec::new(); + let mut seen = std::collections::HashSet::new(); + for node in roots.iter().flat_map(root_operators) { + collect_children_first(node, &mut seen, &mut nodes); + } + let mut readers: HashMap<*const OperatorNode, Vec<&Rc>> = HashMap::new(); + for node in &nodes { + for child in node.children() { + readers.entry(Rc::as_ptr(child)).or_default().push(node); + } + } + let finalized = |build: &Rc| -> Vec<&Rc> { + readers + .get(&Rc::as_ptr(build)) + .map_or_else(Vec::new, |readers| { + readers + .iter() + .filter(|r| matches!(r.asap(), Some(ASAPOp::FinalizeExactAccumulator { .. }))) + .copied() + .collect() + }) + }; + let builds: Vec<&Rc> = nodes.iter().filter(|n| grouped_exact(n)).collect(); + let mut sites = Vec::new(); + for &coarse in &builds { + let finals = finalized(coarse); + // Every reader of the coarse state reads its finalized value. + if finals.is_empty() || finals.len() != readers[&Rc::as_ptr(coarse)].len() { + continue; + } + let mut best: Option<(usize, RollupSite)> = None; + for &fine in &builds { + let Some(keys) = finer_keys(coarse, fine) else { + continue; + }; + if best.as_ref().is_some_and(|(k, _)| *k <= keys) { + continue; + } + let Some(&finer) = finalized(fine).first() else { + continue; + }; + if let Some(rollup) = rolled_up(coarse, finer, finals[0]) { + best = Some(( + keys, + RollupSite { + coarse: Rc::clone(coarse), + finer: Rc::clone(finer), + rollup, + }, + )); + } + } + sites.extend(best.map(|(_, site)| site)); + } + sites +} + +/// Whether `node` is an exact accumulator build grouped by listed keys whose +/// family merges across groups ([`rollup_kind`]). +fn grouped_exact(node: &OperatorNode) -> bool { + matches!( + node.asap(), + Some(ASAPOp::SummaryAgg { + family, + reduction: Reduction::Reduce(by), + grouping: GroupingStrategy::PerSubpopulationInstance, + .. + }) if !by.is_without() && rollup_kind(family).is_some() + ) +} + +/// The number of group keys of `fine` when it can answer `coarse`: the same +/// family, update and filter over the same rows, and group keys (by column) +/// that strictly include the coarse ones. +fn finer_keys(coarse: &OperatorNode, fine: &OperatorNode) -> Option { + let ( + Some(ASAPOp::SummaryAgg { + child, + family, + input, + reduction: Reduction::Reduce(by), + filter, + .. + }), + Some(ASAPOp::SummaryAgg { + child: fine_child, + family: fine_family, + input: fine_input, + reduction: Reduction::Reduce(fine_by), + filter: fine_filter, + .. + }), + ) = (coarse.asap(), fine.asap()) + else { + return None; + }; + let names = |node: &OperatorNode, keys: &[usize]| -> Option> { + keys.iter() + .map(|&k| node.schema.fields.get(k).map(column_name)) + .collect() + }; + let coarse_keys = names(child, by.keys())?; + let fine_keys = names(fine_child, fine_by.keys())?; + let same_filter = match (filter, fine_filter) { + (None, None) => true, + // A filter indexes its input's columns: equal only over equal rows. + (a, b) => a == b && (Rc::ptr_eq(child, fine_child) || child == fine_child), + }; + (family == fine_family + && input == fine_input + && same_filter + && fine_keys.len() > coarse_keys.len() + && coarse_keys.iter().all(|key| fine_keys.contains(key)) + && same_rows(child, fine_child) + // A global SQL aggregate has a row over no input; a rollup of no + // finer group has none. + && !(coarse_keys.is_empty() && child.result_kind == ResultKind::Relation)) + .then_some(fine_keys.len()) +} + +type ColumnName = (Option, String); + +fn column_name(field: &asap_types::pre_asap::Field) -> ColumnName { + (field.table.clone(), field.name.clone()) +} + +/// Whether `a` and `b` output the same rows: the same or equal nodes, or +/// time-series scans of one metric without predicates (projecting perhaps +/// other labels) under the same time ranges and shifts. +fn same_rows(a: &Rc, b: &Rc) -> bool { + if Rc::ptr_eq(a, b) || a == b { + return true; + } + match (a.non_asap(), b.non_asap()) { + ( + Some(NonASAPOp::Scan { + source: source @ Source::TimeSeries { .. }, + predicates, + .. + }), + Some(NonASAPOp::Scan { + source: other, + predicates: other_predicates, + .. + }), + ) => source == other && predicates.is_empty() && other_predicates.is_empty(), + ( + Some(NonASAPOp::TimeRange { range, kind, child }), + Some(NonASAPOp::TimeRange { + range: other_range, + kind: other_kind, + child: other, + }), + ) => range == other_range && kind == other_kind && same_rows(child, other), + ( + Some(NonASAPOp::TimeShift { shift, child }), + Some(NonASAPOp::TimeShift { + shift: other_shift, + child: other, + }), + ) => shift == other_shift && same_rows(child, other), + _ => false, + } +} + +/// The finalized value of `coarse` (`coarse_value`) as an exact +/// accumulator over `finer` (the finer build's finalized values), grouped by +/// the coarse keys, finalized, with `coarse_value`'s column names; `None` +/// when it cannot be built or differs from `coarse_value` in type. +fn rolled_up( + coarse: &OperatorNode, + finer: &Rc, + coarse_value: &OperatorNode, +) -> Option> { + use crate::per_computation_candidates::{exact_accumulator, partitions_over}; + let Some(ASAPOp::SummaryAgg { + child, + family, + reduction, + .. + }) = coarse.asap() + else { + return None; + }; + let Some(ASAPOp::FinalizeExactAccumulator { child: fine }) = finer.asap() else { + return None; + }; + let Some(ASAPOp::SummaryAgg { + reduction: Reduction::Reduce(fine_by), + .. + }) = fine.asap() + else { + return None; + }; + // A grouped accumulator outputs its keys, then its value. + let value = fine_by.len(); + let regrouped = partitions_over(reduction, &child.schema, &finer.schema)?; + if regrouped + .group_keys() + .is_some_and(|keys| keys.iter().any(|&k| k >= value)) + { + return None; + } + let kind = rollup_kind(family)?; + let intent = match kind { + ExactKind::Min => AggIntent::Min { col: Some(value) }, + ExactKind::Max => AggIntent::Max { col: Some(value) }, + _ => AggIntent::Sum { col: Some(value) }, + }; + let state = exact_accumulator(kind, &intent, finer, regrouped).ok()?; + let mut finalized = + OperatorNode::new_shared(Operator::ASAP(ASAPOp::FinalizeExactAccumulator { + child: state, + })) + .ok()?; + let types = |node: &OperatorNode| -> Vec { + node.schema.fields.iter().map(|f| f.dtype.clone()).collect() + }; + if types(&finalized) != types(coarse_value) { + return None; + } + // A count summed is named after the sum: rename to the coarse columns. + if finalized.schema.fields != coarse_value.schema.fields { + let cols = coarse_value + .schema + .fields + .iter() + .enumerate() + .map(|(column, field)| asap_types::ir::ProjectItem { + alias: Some(field.name.clone()), + expr: ScalarExpr::Column(column), + }) + .collect(); + finalized = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Project { + cols, + qualifier: None, + child: finalized, + })) + .ok()?; + } + Some(finalized) +} + +/// Every Pass 2 candidate of the rollup rule for the candidate with query +/// roots `roots` and sites `sites` ([`rollup_sites`]): one per non-empty +/// subset of the sites, in binary-counter order (the first site the lowest +/// bit). An error only for more than [`MAX_REUSE_SITES`] sites; a variant +/// whose composition fails carries its own error. +pub fn rollup_variants( + roots: &[QueryRoot], + sites: &[RollupSite], +) -> Result, PerComputationCandidateError> { + Ok(subsets(sites.len(), "rollup")? + .into_iter() + .map(|chosen| { + let values: HashMap<*const OperatorNode, &RollupSite> = chosen + .iter() + .map(|&i| (Rc::as_ptr(&sites[i].coarse), &sites[i])) + .collect(); + // Every finalized value of a chosen coarse build is replaced. + let replaced = roots + .iter() + .flat_map(root_operators) + .flat_map(OperatorNode::reachable) + .filter_map(|node| match node.asap() { + Some(ASAPOp::FinalizeExactAccumulator { child }) => { + let site = values.get(&Rc::as_ptr(child))?; + Some((Rc::as_ptr(&node), Rc::clone(&site.rollup))) + } + _ => None, + }) + .collect(); + RollupVariant { + roots: replace_all(roots, &replaced), + sites: chosen, + } + }) + .collect()) +} + +/// Every non-empty subset of `n` sites, in binary-counter order; an error +/// past [`MAX_REUSE_SITES`]. +fn subsets(n: usize, rule: &'static str) -> Result>, PerComputationCandidateError> { + if n > MAX_REUSE_SITES { + return Err(PerComputationCandidateError::Unsupported(match rule { + "rollup" => "more than 20 rollup sites (2^n candidates)", + _ => "more than 20 top-k reuse sites (2^n candidates)", + })); + } + Ok((1u64..(1u64 << n)) + .map(|mask| (0..n).filter(|&i| mask >> i & 1 == 1).collect()) + .collect()) +} + +/// `roots` with every node in `replaced` swapped for its replacement, and +/// the nodes a replacement reads rebuilt too (a rollup may read a build +/// another rollup replaces); every guarantee composed again. +fn replace_all( + roots: &[QueryRoot], + replaced: &HashMap<*const OperatorNode, Rc>, +) -> Result, PerComputationCandidateError> { + fn walk( + node: &Rc, + replaced: &HashMap<*const OperatorNode, Rc>, + rebuilt: &mut HashMap<*const OperatorNode, Rc>, + ) -> Result, PerComputationCandidateError> { + if let Some(done) = rebuilt.get(&Rc::as_ptr(node)) { + return Ok(Rc::clone(done)); + } + let source = replaced.get(&Rc::as_ptr(node)).unwrap_or(node); + let mut changed = !Rc::ptr_eq(source, node); + let mut children = HashMap::new(); + for child in source.children() { + let new = walk(child, replaced, rebuilt)?; + changed |= !Rc::ptr_eq(&new, child); + children.insert(Rc::as_ptr(child), new); + } + let out = if !changed { + Rc::clone(node) + } else if children + .iter() + .all(|(old, new)| std::ptr::eq(*old, Rc::as_ptr(new))) + { + Rc::clone(source) + } else { + Rc::new(source.with_new_children(|child| children[&Rc::as_ptr(child)].clone())?) + }; + rebuilt.insert(Rc::as_ptr(node), Rc::clone(&out)); + Ok(out) + } + let mut rebuilt = HashMap::new(); + let mut out = Vec::new(); + for root in roots { + out.push(match root { + QueryRoot::Operator(node) => QueryRoot::Operator(walk(node, replaced, &mut rebuilt)?), + QueryRoot::Scalar(expr) => { + let mut nodes = HashMap::new(); + for node in expr.operator_refs() { + nodes.insert(Rc::as_ptr(node), walk(node, replaced, &mut rebuilt)?); + } + QueryRoot::Scalar( + expr.map_operator_refs(&mut |node| nodes[&Rc::as_ptr(node)].clone()), + ) + } + }); + } + with_composed_guarantees(&out) +} + +/// Exact top-k computations of one input and partition with at least two +/// different `k`, in the order a children-first walk meets them, and the +/// ranking they all read in a reuse candidate: `Limit(k_max)` over `Sort` +/// of the input, partitioned as they are. +#[derive(Debug, Clone)] +pub struct TopKReuseSite { + pub members: Vec>, + pub ranking: Rc, +} + +/// A Pass 2 candidate of the exact top-k reuse rule: the candidate's roots +/// with every member of each site in `sites` (indices into +/// [`topk_reuse_sites`]) reading its site's ranking, and every guarantee +/// composed again; the error when that fails. +#[derive(Debug)] +pub struct TopKReuseVariant { + pub sites: Vec, + pub roots: Result, PerComputationCandidateError>, +} + +/// The exact top-k reuse sites of the candidate with query roots `roots`: +/// plain (exact) `Aggregate` top-k nodes over an exact input, unfiltered and +/// grouped by listed keys, grouped by equal input and partition. +pub fn topk_reuse_sites(roots: &[QueryRoot]) -> Vec { + let mut nodes = Vec::new(); + let mut seen = std::collections::HashSet::new(); + for node in roots.iter().flat_map(root_operators) { + collect_children_first(node, &mut seen, &mut nodes); + } + let mut groups: Vec>> = Vec::new(); + for node in nodes.into_iter().filter(|n| exact_topk(n).is_some()) { + let same = |member: &Rc| { + let (Some((child, by, _)), Some((other, other_by, _))) = + (exact_topk(member), exact_topk(&node)) + else { + return false; + }; + by == other_by && (Rc::ptr_eq(child, other) || child == other) + }; + match groups.iter_mut().find(|g| same(&g[0])) { + Some(group) => group.push(node), + None => groups.push(vec![node]), + } + } + groups + .into_iter() + .filter_map(|members| { + let ks: Vec = members + .iter() + .filter_map(|m| exact_topk(m)) + .map(|t| t.2) + .collect(); + let largest = *ks.iter().max()?; + if ks.iter().all(|&k| k == largest) { + return None; + } + let (child, by, _) = exact_topk(&members[0])?; + let ranking = ranking(child, by, largest)?; + Some(TopKReuseSite { members, ranking }) + }) + .collect() +} + +/// The input, partition and `k` of an exact top-k over an exact input. +fn exact_topk(node: &OperatorNode) -> Option<(&Rc, &Reduction, usize)> { + let Some(NonASAPOp::Aggregate { + child, + reduction: reduction @ Reduction::Reduce(by), + measures, + filters, + having: None, + .. + }) = node.non_asap() + else { + return None; + }; + let [AggIntent::TopK { k, .. }] = measures.as_slice() else { + return None; + }; + let exact = child + .guarantee + .as_ref() + .is_some_and(asap_types::post_asap::AccuracyGuarantee::is_exact); + (exact && !by.is_without() && filters.iter().all(Option::is_none)) + .then_some((child, reduction, *k)) +} + +/// The `k` highest-ranked rows of `child` per partition: `Limit` over a +/// descending `Sort` by its ranked column, as exact top-k runs. +fn ranking(child: &Rc, by: &Reduction, k: usize) -> Option> { + let Reduction::Reduce(keys) = by else { + return None; + }; + let value = crate::realization::ranked_column(child).ok()?; + let sorted = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Sort { + keys: vec![SortKey { + expr: ScalarExpr::Column(value), + ascending: false, + nulls_first: false, + }], + partition_by: keys.clone(), + child: Rc::clone(child), + })) + .ok()?; + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Limit { + n: Some(k), + offset: 0, + partition_by: keys.clone(), + child: sorted, + })) + .ok() +} + +/// Every Pass 2 candidate of the exact top-k reuse rule for the candidate +/// with query roots `roots` and sites `sites` ([`topk_reuse_sites`]): one +/// per non-empty subset of the sites, in binary-counter order. An error +/// only for more than [`MAX_REUSE_SITES`] sites. +pub fn topk_reuse_variants( + roots: &[QueryRoot], + sites: &[TopKReuseSite], +) -> Result, PerComputationCandidateError> { + Ok(subsets(sites.len(), "top-k reuse")? + .into_iter() + .map(|chosen| { + let mut replaced = HashMap::new(); + let mut failed = None; + for &i in &chosen { + let site = &sites[i]; + for member in &site.members { + match member.with_new_children(|_| Rc::clone(&site.ranking)) { + Ok(node) => { + replaced.insert(Rc::as_ptr(member), Rc::new(node)); + } + Err(error) => failed = Some(error.into()), + } + } + } + TopKReuseVariant { + roots: match failed { + Some(error) => Err(error), + None => replace_all(roots, &replaced), + }, + sites: chosen, + } + }) + .collect()) +} + fn root_operators(root: &QueryRoot) -> Vec<&Rc> { match root { QueryRoot::Operator(node) => vec![node], @@ -1434,4 +2016,283 @@ mod tests { .collect(); assert_eq!(subsets, [vec![0], vec![1], vec![0, 1]]); } + /// The composed candidate of `queries` with every target at its exact + /// option (an exact accumulator, or plain exact execution). + 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 { .. }) + }) + } + + fn operator(root: &QueryRoot) -> &Rc { + let QueryRoot::Operator(node) = root else { + panic!("operator root") + }; + node + } + + fn builds(root: &QueryRoot) -> Vec> { + OperatorNode::reachable(operator(root)) + .into_iter() + .filter(|n| matches!(n.asap(), Some(ASAPOp::SummaryAgg { .. }))) + .collect() + } + + /// `sum by (job)` is answered from `sum by (job, instance)`, which the + /// workload computes anyway: the coarse query reads the finer query's + /// finalized sums, summed by `job`, so it builds nothing over the raw + /// rows; both values stay exact and keep their columns. + #[test] + fn rollup_from_a_finer_exact_sum() { + let roots = exact(&["sum by (job) (m)", "sum by (job, instance) (m)"]); + let sites = rollup_sites(&roots); + let [site] = sites.as_slice() else { + panic!("one site: {sites:?}") + }; + assert!(Rc::ptr_eq(&site.coarse, &builds(&roots[0])[0])); + let variants = rollup_variants(&roots, &sites).unwrap(); + let [variant] = variants.as_slice() else { + panic!("one variant") + }; + let rolled = variant.roots.as_ref().unwrap(); + let coarse = builds(&rolled[0]); + let fine = builds(&rolled[1]); + assert_eq!(fine.len(), 1); + // The finer build, then the sum over its values. + assert_eq!(coarse.len(), 2); + assert!(Rc::ptr_eq(&coarse[0], &fine[0]) || Rc::ptr_eq(&coarse[1], &fine[0])); + let rollup = coarse.iter().find(|b| !Rc::ptr_eq(b, &fine[0])).unwrap(); + let Some(ASAPOp::SummaryAgg { child, family, .. }) = rollup.asap() else { + unreachable!() + }; + assert!(matches!( + child.asap(), + Some(ASAPOp::FinalizeExactAccumulator { .. }) + )); + assert_eq!( + *family, + FieldDataType::ExactAggregate(ExactKind::Sum, asap_types::post_asap::ExactParams::Sum) + ); + for (before, after) in roots.iter().zip(rolled) { + let (before, after) = (operator(before), operator(after)); + after.validate_structure().unwrap(); + assert_eq!(before.schema, after.schema); + assert!(after + .guarantee + .as_ref() + .is_some_and(asap_types::post_asap::AccuracyGuarantee::is_exact)); + } + } + + /// Counts roll up as the sum of the finer counts; minima as minima. + #[test] + fn counts_and_extrema_roll_up() { + for (query, kind) in [ + ("count by (job) (m)", ExactKind::Sum), + ("min by (job) (m)", ExactKind::Min), + ("max by (job) (m)", ExactKind::Max), + ] { + let finer = query.replace("(job)", "(job, instance)"); + let roots = exact(&[query, &finer]); + let sites = rollup_sites(&roots); + let [site] = sites.as_slice() else { + panic!("{query}: one site: {sites:?}") + }; + let mut value = Rc::clone(&site.rollup); + if let Some(NonASAPOp::Project { child, .. }) = value.non_asap() { + value = Rc::clone(child); + } + let Some(ASAPOp::FinalizeExactAccumulator { child: state }) = value.asap() else { + panic!("a finalized value") + }; + let Some(ASAPOp::SummaryAgg { + family: FieldDataType::ExactAggregate(rolled, _), + .. + }) = state.asap() + else { + panic!("an exact accumulator") + }; + assert_eq!(*rolled, kind, "{query}"); + let variant = rollup_variants(&roots, &sites).unwrap().remove(0); + assert_eq!( + operator(&variant.roots.unwrap()[0]).schema, + operator(&roots[0]).schema, + "{query}" + ); + } + } + + /// The rule reuses only a finer computation the workload already has: + /// keys that are not a superset, another metric, window or measure, or + /// a lone query, give no site. + #[test] + fn rollup_needs_a_finer_computation_of_the_same_rows() { + for queries in [ + &["sum by (job) (m)"][..], + &["sum by (job) (m)", "sum by (instance) (m)"], + &["sum by (job) (m)", "sum by (job, instance) (n)"], + &["sum by (job) (m)", "max by (job, instance) (m)"], + &[ + "sum by (job) (sum_over_time(m[5m]))", + "sum by (job, instance) (sum_over_time(m[1m]))", + ], + ] { + assert!(rollup_sites(&exact(queries)).is_empty(), "{queries:?}"); + } + } + + /// Rollup follows the family-capability table: sums, counts and + /// extrema merge exactly across groups; counter states merge only in + /// time, IRate not at all; a heap sketch's merged top-k has no + /// guarantee, and no sketch's states can be regrouped in the IR. + #[test] + fn rollup_is_refused_for_families_without_an_exact_merge_across_groups() { + use asap_types::post_asap::{ExactParams as P, SketchParams}; + let exact = |kind, params| FieldDataType::ExactAggregate(kind, params); + assert_eq!( + rollup_kind(&exact(ExactKind::Sum, P::Sum)), + Some(ExactKind::Sum) + ); + assert_eq!( + rollup_kind(&exact(ExactKind::Count, P::Count)), + Some(ExactKind::Sum) + ); + for family in [ + exact(ExactKind::Rate, P::Rate), + exact(ExactKind::Increase, P::Increase), + exact(ExactKind::IRate, P::IRate), + FieldDataType::Sketch( + SketchKind::new( + SketchAlgorithm::CmsWithHeap, + SketchParams::CmsWithHeap { + width: 64, + depth: 5, + heap_size: 10, + }, + ), + GroupingStrategy::default(), + ), + FieldDataType::Sketch( + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 200, + failure_probability: None, + }, + ), + GroupingStrategy::default(), + ), + ] { + assert_eq!(rollup_kind(&family), None, "{family:?}"); + } + // A grouped KLL beside a finer one is no site. + let roots = chose( + &[ + ("quantile by (job) (0.9, m)", 0.01), + ("quantile by (job, instance) (0.9, m)", 0.01), + ], + SketchAlgorithm::Kll, + ); + assert!(rollup_sites(&roots).is_empty()); + } + + /// Two exact top-k over the same per-job sums: both read one ranking, + /// the 10 highest sums, and the top-5 is computed from it; each keeps + /// its `k`, output and exact value. + #[test] + fn exact_topk_reuses_the_larger_ranking() { + let roots = exact(&["topk(5, sum by (job) (m))", "topk(10, sum by (job) (m))"]); + let sites = topk_reuse_sites(&roots); + let [site] = sites.as_slice() else { + panic!("one site: {sites:?}") + }; + assert_eq!(site.members.len(), 2); + let Some(NonASAPOp::Limit { n: Some(10), .. }) = site.ranking.non_asap() else { + panic!("{:?}", site.ranking.operator) + }; + let variant = topk_reuse_variants(&roots, &sites).unwrap().remove(0); + let reused = variant.roots.unwrap(); + let read = |root: &QueryRoot| { + let Some(NonASAPOp::Aggregate { + child, measures, .. + }) = operator(root).non_asap() + else { + panic!("top-k aggregate") + }; + (Rc::clone(child), measures[0].clone()) + }; + let ((a, small), (b, large)) = (read(&reused[0]), read(&reused[1])); + assert!(Rc::ptr_eq(&a, &b), "one ranking"); + assert!(matches!(small, AggIntent::TopK { k: 5, .. })); + assert!(matches!(large, AggIntent::TopK { k: 10, .. })); + for (before, after) in roots.iter().zip(&reused) { + let (before, after) = (operator(before), operator(after)); + after.validate_structure().unwrap(); + assert_eq!(before.schema, after.schema); + assert!(after + .guarantee + .as_ref() + .is_some_and(asap_types::post_asap::AccuracyGuarantee::is_exact)); + } + } + + /// Top-k over other partitions or inputs, or of one `k`, share no + /// ranking; a top-k a heap sketch answers is not reused. + #[test] + fn topk_reuse_needs_exact_topk_of_one_input_and_partition() { + for queries in [ + &[ + "topk(5, sum by (job) (m))", + "topk by (job) (10, sum by (job, instance) (m))", + ][..], + &["topk(5, sum by (job) (m))", "topk(10, sum by (job) (n))"], + &["topk(5, sum by (job) (m))", "topk(5, max by (job) (m))"], + ] { + assert!(topk_reuse_sites(&exact(queries)).is_empty(), "{queries:?}"); + } + use crate::per_computation_candidates::{ + compose_logical_candidate, generate_per_computation_candidates, + ComputationCandidate as C, + }; + let inventory = generate_per_computation_candidates( + [ + "topk by (job) (5, sum_over_time(m[1m]))", + "topk by (job) (10, sum_over_time(m[1m]))", + ] + .iter() + .enumerate() + .map(|(i, q)| { + let root = lower_promql( + q, + asap_types::types::AccuracyTarget::epsilon_delta(0.01, 0.01), + ); + (i, QueryRoot::Operator(root)) + }) + .collect(), + ) + .unwrap(); + let heap = |c: &C| { + matches!( + c, + C::WholeExpression(SketchAlgorithm::CountSketchWithHeap) + | C::Summary(SketchAlgorithm::CountSketchWithHeap) + ) + }; + let choice: Vec = inventory + .targets + .iter() + .map(|t| t.choices.candidates.iter().position(heap).unwrap_or(0)) + .collect(); + let heaps: Vec = compose_logical_candidate(&inventory, &choice) + .unwrap() + .into_iter() + .map(|(_, root)| root) + .collect(); + assert!(OperatorNode::reachable(operator(&heaps[0])) + .iter() + .any(|n| matches!(n.asap(), Some(ASAPOp::SummaryEstimate { .. })))); + assert!(topk_reuse_sites(&heaps).is_empty()); + } } 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 c1a5ed77..8c5c4cf1 100644 --- a/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs +++ b/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs @@ -19,6 +19,10 @@ //! those of the window-composition rule ([`window_composition_variants`]: //! one per non-empty subset of the builds a repeating query could build //! as tumbling panes), each followed by its own identical-expression +//! candidates, and, for the composed candidate, those of the reuse rules +//! (exact top-k reuse, then each rollup candidate followed by its own +//! exact top-k reuse candidates: [`topk_reuse_variants`], +//! [`rollup_variants`]), each followed by its own identical-expression //! candidates; every distinct candidate once; the independent candidate //! is kept; //! 2. Stage 2, physical ASAP-aware optimization of every Stage 1 candidate @@ -45,8 +49,9 @@ use asap_types::workload::{DataWorkload, QueryLanguage, QueryRecurrence, Repeate use super::{OptimizationInput, OptimizationPass, OptimizeError, PlanOutput, QueryPlan}; use crate::asap_aware_cse::{ - capability_groups, identical_expression_variants, pane_sites, sharing_groups, - summary_capability_variants, window_composition_variants, + capability_groups, identical_expression_variants, pane_sites, rollup_sites, rollup_variants, + sharing_groups, summary_capability_variants, topk_reuse_sites, topk_reuse_variants, + window_composition_variants, }; use crate::per_computation_candidates::{ compose_logical_candidate, generate_per_computation_candidates, ComputationCandidate, @@ -74,7 +79,9 @@ pub struct LogicalCandidate { /// summary build set in `summaries` made one build sized for all its /// consumers (the summary-capability rule, e.g. "UnivMon for Q1, Q3"), /// then each build in `panes` built as tumbling panes (the -/// window-composition rule, e.g. "Kll for Q1 in 5 × 1m panes"), then the +/// window-composition rule, e.g. "Kll for Q1 in 5 × 1m panes"), or each +/// computation in `reused` answered from another (the rollup and exact +/// top-k reuse rules, e.g. "Sum acc for Q1 rolled up from Q2"), then the /// nodes of each group in `shared` (named by operator kind, children first) /// made one node (the identical-expression rule). It has `from_logical`'s /// `choice`. @@ -84,6 +91,7 @@ pub struct Pass2Candidate { pub from_logical: String, pub summaries: Vec, pub panes: Vec, + pub reused: Vec, pub shared: Vec, } @@ -191,6 +199,9 @@ pub struct Pass2Of { pub summaries: Vec, /// The builds built as tumbling panes (window-composition rule). pub panes: Vec, + /// The computations answered from another (rollup and exact top-k + /// reuse rules). + pub reused: Vec, /// The identical groups shared (operator kinds, children first). pub shared: Vec, } @@ -199,7 +210,8 @@ pub struct Pass2Of { /// candidate at a time: each composed Pass 1 candidate (every choice of a /// per-computation candidate for every target, the last target varying /// fastest), followed by its Pass 2 candidates: the identical-expression -/// rule's, then the window-composition rule's, then each summary-capability +/// rule's, then the window-composition rule's, then the reuse rules' +/// (exact top-k reuse and rollup), then each summary-capability /// candidate followed by its own identical-expression and /// window-composition candidates; every window-composition candidate is /// followed by its own identical-expression candidates. `recurrence[q]` is @@ -243,8 +255,12 @@ pub fn stage1_logical_asap_optimization<'a, Id: Clone>( pass2_of: None, }); for Pass2 { - summaries, - panes, + rules: + Rules { + summaries, + panes, + reused, + }, shared, roots, } in pass2 @@ -256,6 +272,9 @@ pub fn stage1_logical_asap_optimization<'a, Id: Clone>( if !panes.is_empty() { name += &format!(" · {}", panes.join(", ")); } + if !reused.is_empty() { + name += &format!(" · {}", reused.join(", ")); + } if !shared.is_empty() { name += &format!(" · shared {}", shared.join(", ")); } @@ -267,6 +286,7 @@ pub fn stage1_logical_asap_optimization<'a, Id: Clone>( from, summaries, panes, + reused, shared, }), }); @@ -277,21 +297,44 @@ pub fn stage1_logical_asap_optimization<'a, Id: Clone>( }) } -/// A Pass 2 candidate: the summary builds it shares (summary-capability -/// rule), the builds it builds as tumbling panes (window-composition rule), -/// the identical groups it shares, and its roots. +/// A Pass 2 candidate: what the rules other than the identical-expression +/// rule changed, the identical groups it shares, and its roots. struct Pass2 { - summaries: Vec, - panes: Vec, + rules: Rules, shared: Vec, roots: Result, PerComputationCandidateError>, } +/// What the rules other than the identical-expression rule changed: the +/// summary builds shared (summary-capability rule), the builds built as +/// tumbling panes (window-composition rule) and the computations answered +/// from another (rollup and exact top-k reuse rules). +#[derive(Debug, Clone, Default)] +struct Rules { + summaries: Vec, + panes: Vec, + reused: Vec, +} + +impl Rules { + fn with_panes(&self, panes: Vec) -> Self { + Self { + panes, + ..self.clone() + } + } + + fn with_reused(&self, reused: &[String]) -> Self { + let mut rules = self.clone(); + rules.reused.extend_from_slice(reused); + rules + } +} + impl Pass2 { - fn failed(summaries: &[String], panes: &[String], error: PerComputationCandidateError) -> Self { + fn failed(rules: &Rules, error: PerComputationCandidateError) -> Self { Self { - summaries: summaries.to_vec(), - panes: panes.to_vec(), + rules: rules.clone(), shared: Vec::new(), roots: Err(error), } @@ -317,13 +360,15 @@ fn pass2_candidates(roots: &[QueryRoot], interval_ms: &[Option]) -> Vec variants, Err(error) => { - found.out.push(Pass2::failed(&[], &[], error)); + found.out.push(Pass2::failed(&none, error)); return found.out; } }; @@ -344,15 +389,19 @@ fn pass2_candidates(roots: &[QueryRoot], interval_ms: &[Option]) -> Vec { - found.push(&summaries, &[], Vec::new(), &roots); - found.identical_variants(&summaries, &[], &roots); - found.window_variants(&summaries, &roots, interval_ms); + found.push(&rules, Vec::new(), &roots); + found.identical_variants(&rules, &roots); + found.window_variants(&rules, &roots, interval_ms); } // Listed as a candidate that cannot be composed; the other // variants go on. - Err(error) => found.out.push(Pass2::failed(&summaries, &[], error)), + Err(error) => found.out.push(Pass2::failed(&rules, error)), } } found.out @@ -366,29 +415,22 @@ struct Pass2List { } impl Pass2List { - fn push( - &mut self, - summaries: &[String], - panes: &[String], - shared: Vec, - roots: &[QueryRoot], - ) { + fn push(&mut self, rules: &Rules, shared: Vec, roots: &[QueryRoot]) { let flat = flatten(roots).0; if self.seen.contains(&flat) { return; } self.seen.push(flat); self.out.push(Pass2 { - summaries: summaries.to_vec(), - panes: panes.to_vec(), + rules: rules.clone(), shared, roots: Ok(roots.to_vec()), }); } - /// The identical-expression candidates of `roots`, which shares - /// `summaries` and builds `panes` in tumbling panes. - fn identical_variants(&mut self, summaries: &[String], panes: &[String], roots: &[QueryRoot]) { + /// The identical-expression candidates of `roots`, which the other + /// rules changed as `rules` says. + fn identical_variants(&mut self, rules: &Rules, roots: &[QueryRoot]) { let groups = sharing_groups(roots); for variant in identical_expression_variants(roots, &groups) { let shared = variant @@ -396,23 +438,19 @@ impl Pass2List { .iter() .map(|&g| format!("{} #{}", groups[g].nodes[0].operator.kind_name(), g + 1)) .collect(); - self.push(summaries, panes, shared, &variant.roots); + self.push(rules, shared, &variant.roots); } } - /// The window-composition candidates of `roots`, which shares - /// `summaries`, each followed by its identical-expression candidates. - fn window_variants( - &mut self, - summaries: &[String], - roots: &[QueryRoot], - interval_ms: &[Option], - ) { + /// The window-composition candidates of `roots`, which the other rules + /// changed as `rules` says, each followed by its identical-expression + /// candidates. + fn window_variants(&mut self, rules: &Rules, roots: &[QueryRoot], interval_ms: &[Option]) { let sites = pane_sites(roots, interval_ms); let variants = match window_composition_variants(roots, &sites) { Ok(variants) => variants, Err(error) => { - self.out.push(Pass2::failed(summaries, &[], error)); + self.out.push(Pass2::failed(rules, error)); return; } }; @@ -433,15 +471,108 @@ impl Pass2List { ) }) .collect(); + let rules = rules.with_panes(panes); match variant.roots { Ok(roots) => { - self.push(summaries, &panes, Vec::new(), &roots); - self.identical_variants(summaries, &panes, &roots); + self.push(&rules, Vec::new(), &roots); + self.identical_variants(&rules, &roots); } - Err(error) => self.out.push(Pass2::failed(summaries, &panes, error)), + Err(error) => self.out.push(Pass2::failed(&rules, error)), } } } + + /// The reuse candidates of `roots`: the exact top-k reuse rule's, then + /// each rollup candidate followed by its own exact top-k reuse + /// candidates; each followed by its identical-expression candidates. + fn reuse_variants(&mut self, rules: &Rules, roots: &[QueryRoot]) { + self.topk_variants(rules, roots); + let sites = rollup_sites(roots); + let variants = match rollup_variants(roots, &sites) { + Ok(variants) => variants, + Err(error) => { + self.out.push(Pass2::failed(rules, error)); + return; + } + }; + for variant in variants { + let reused: Vec = variant + .sites + .iter() + .map(|&i| { + let site = &sites[i]; + format!( + "{} for {} rolled up from {}", + state_name(&site.coarse), + queries_reading(roots, &site.coarse), + queries_reading(roots, &site.finer) + ) + }) + .collect(); + let rules = rules.with_reused(&reused); + match variant.roots { + Ok(roots) => { + self.push(&rules, Vec::new(), &roots); + self.identical_variants(&rules, &roots); + self.topk_variants(&rules, &roots); + } + Err(error) => self.out.push(Pass2::failed(&rules, error)), + } + } + } + + /// The exact top-k reuse candidates of `roots`, each followed by its + /// identical-expression candidates. + fn topk_variants(&mut self, rules: &Rules, roots: &[QueryRoot]) { + let sites = topk_reuse_sites(roots); + let variants = match topk_reuse_variants(roots, &sites) { + Ok(variants) => variants, + Err(error) => { + self.out.push(Pass2::failed(rules, error)); + return; + } + }; + for variant in variants { + let reused: Vec = variant + .sites + .iter() + .map(|&i| { + let site = &sites[i]; + let k = match site.ranking.non_asap() { + Some(asap_types::ir::NonASAPOp::Limit { n: Some(k), .. }) => *k, + _ => 0, + }; + let mut queries: Vec = site + .members + .iter() + .flat_map(|m| readers_of(roots, m)) + .collect(); + queries.sort_unstable(); + queries.dedup(); + let queries: Vec = + queries.iter().map(|q| format!("Q{}", q + 1)).collect(); + format!("top-k for {} from one top-{k}", queries.join(", ")) + }) + .collect(); + let rules = rules.with_reused(&reused); + match variant.roots { + Ok(roots) => { + self.push(&rules, Vec::new(), &roots); + self.identical_variants(&rules, &roots); + } + Err(error) => self.out.push(Pass2::failed(&rules, error)), + } + } + } +} + +/// The queries (`Q1, Q2`) that read `node`. +fn queries_reading(roots: &[QueryRoot], node: &Rc) -> String { + readers_of(roots, node) + .iter() + .map(|q| format!("Q{}", q + 1)) + .collect::>() + .join(", ") } /// The state a summary build keeps, e.g. "Kll" or "Rate acc". @@ -535,6 +666,7 @@ pub fn run_stage2_and_stage3( from_logical: format!("L{}", of.from + 1), summaries: of.summaries, panes: of.panes, + reused: of.reused, shared: of.shared, }); } From efdceaa92fb0319740b024796cf2e97b42859a09 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:16:05 +0000 Subject: [PATCH 3/6] test(planner): reuse candidates selected and executed end to end (#680) The nested-sum merge is selected and answers as the nested sums; rollup and exact top-k reuse are selected once the shared finer computation is computed once per evaluation, a stand-in for Stage 2 materialization. --- crates/integration-tests/tests/reuse_rules.rs | 292 ++++++++++++++++++ 1 file changed, 292 insertions(+) create mode 100644 crates/integration-tests/tests/reuse_rules.rs diff --git a/crates/integration-tests/tests/reuse_rules.rs b/crates/integration-tests/tests/reuse_rules.rs new file mode 100644 index 00000000..3e26cc1d --- /dev/null +++ b/crates/integration-tests/tests/reuse_rules.rs @@ -0,0 +1,292 @@ +//! Reuse in Stage 1 (#680): the nested-sum merge (a Pass 1 candidate) is +//! selected over the nested sums and answers the same; the rollup and exact +//! top-k reuse candidates (Pass 2) are selected once the finer computation +//! they read is materialized for all its readers. +mod physical_common; + +use std::collections::BTreeMap; +use std::rc::Rc; +use std::sync::Arc; + +use asap_aware_mapping::stage3_plan_selection::select_plan; +use asap_aware_mapping::{run_stage1_to_stage3, PlanningModels, Stage1ToStage3Run}; +use asap_devtools::planner_layering_examples::shared_data_workload; +use asap_integration_tests::fixtures::lower_promql; +use asap_physical_operators::physical_planner::{compile, promql_rows, InputContract}; +use asap_physical_operators::runtime::Scope; +use asap_physical_operators::values::{Batch, Value}; +use asap_types::ir::physical_export::{ + compile_physical_asap_dag_with_node_ids, PhysicalASAPOperatorPayload, +}; +use asap_types::ir::{ + apply_materialization_timings, ASAPOp, Materialization, MaterializationAssignment, Medium, + NonASAPOp, Operator, OperatorNode, QueryRoot, TimingMemo, +}; +use asap_types::post_asap::Schema; +use asap_types::types::AccuracyTarget; +use asap_types::workload::{ + DataArrival, DataWorkload, QueryRecurrence, RepeatedDemand, RepetitionInterval, +}; + +const EVALUATION_MS: i64 = 300_000; + +/// Every stage's output for `queries`, exact, each with `recurrence`. +fn plan(queries: &[&str], recurrence: QueryRecurrence) -> Stage1ToStage3Run { + let roots = queries + .iter() + .enumerate() + .map(|(i, query)| { + let root = lower_promql(query, AccuracyTarget::exact()).unwrap(); + let root = promql_rows::with_series_identity(&root).unwrap(); + (i, QueryRoot::Operator(root)) + }) + .collect(); + run_stage1_to_stage3( + roots, + &vec![Some(AccuracyTarget::exact()); queries.len()], + &vec![recurrence; queries.len()], + &data(), + PlanningModels::builtin(), + ) + .unwrap() +} + +/// The #509 shared data workload: 1,000,000 series sampled every 15 s. +fn data() -> DataWorkload { + shared_data_workload(DataArrival::ContinuouslyIngesting) +} + +/// Every minute. +fn repeating() -> QueryRecurrence { + QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(RepetitionInterval(60_000))) +} + +fn logical(run: &Stage1ToStage3Run, id: &str) -> Rc { + let candidate = run + .stage1_logical_asap_optimization + .iter() + .find(|c| c.id == id) + .unwrap(); + match &candidate.roots.as_ref().unwrap()[0] { + QueryRoot::Operator(node) => Rc::clone(node), + QueryRoot::Scalar(_) => panic!("operator root"), + } +} + +/// Two jobs of two series each, a sample every 15 s over [0, 5m). +fn samples() -> Vec<(&'static str, &'static str, i64, f64)> { + let mut rows = vec![]; + for (s, (job, series)) in [("a", "a1"), ("a", "a2"), ("b", "b1"), ("b", "b2")] + .into_iter() + .enumerate() + { + for step in 0..20i64 { + rows.push(( + job, + series, + step * 15_000, + (s as i64 * 100 + step) as f64 + 0.5, + )); + } + } + rows +} + +/// Execute `root` at query time, every raw `TimeRange` input fed `samples`. +fn execute(root: &Rc) -> Vec { + let timed = apply_materialization_timings( + root, + &MaterializationAssignment::default(), + &mut TimingMemo::new(), + ) + .unwrap(); + let compiled = compile_physical_asap_dag_with_node_ids(&timed).unwrap(); + let dag = compiled.dag; + let inputs: BTreeMap = dag + .nodes + .iter() + .filter(|node| { + matches!( + compiled.node_ids.operator_node(node.id).unwrap().operator, + Operator::NonASAP(NonASAPOp::TimeRange { .. }) + ) + }) + .map(|node| { + let schema: Arc = Arc::new(node.output_schema.clone()); + let rows = samples() + .into_iter() + .map(|(job, series, ts, value)| { + let labels = BTreeMap::from([ + ("__name__".to_string(), "m".to_string()), + ("job".to_string(), job.to_string()), + ("series".to_string(), series.to_string()), + ]); + promql_rows::series_row(&schema, &labels, ts, value).unwrap() + }) + .collect(); + (node.id as u64, Batch::try_new(schema, rows).unwrap()) + }) + .collect(); + let contracts = inputs + .iter() + .map(|(&id, batch)| (id, InputContract::bounded(batch.schema().clone()))) + .collect(); + let plan = compile(&dag, contracts, &[dag.roots[0] as u64]).unwrap(); + let mut rows: Vec = physical_common::execute( + &plan, + inputs, + Scope::Query { + evaluation_time_ms: EVALUATION_MS, + revision: 1, + }, + ) + .remove(0) + .iter() + .flat_map(|batch| batch.rows().to_vec()) + .map(|row: Vec| format!("{row:?}")) + .collect(); + rows.sort(); + rows +} + +/// `sum by (job) (sum_over_time(m[5m]))`: the nested sums (L1) and their +/// merge into one accumulator over the raw samples (L2). Plan selection +/// picks the merge, which builds no per-series state and answers the same +/// per-job sums. +#[test] +fn nested_sum_merge_is_selected_and_answers_as_the_nested_sums() { + // Evaluated once, so no tumbling-pane candidates. + let run = plan( + &["sum by (job) (sum_over_time(m[5m]))"], + QueryRecurrence::Unknown, + ); + assert_eq!(run.stage1_logical_asap_optimization.len(), 2); + let selected = run.selected().expect("a plan"); + assert_eq!(selected.from_logical, "L2"); + let (nested, merged) = (logical(&run, "L1"), logical(&run, "L2")); + let builds = |root: &Rc| { + OperatorNode::reachable(root) + .iter() + .filter(|n| matches!(n.asap(), Some(ASAPOp::SummaryAgg { .. }))) + .count() + }; + assert_eq!((builds(&nested), builds(&merged)), (2, 1)); + let answer = execute(&merged); + assert_eq!(answer.len(), 2, "one sum per job: {answer:?}"); + assert_eq!(answer, execute(&nested)); +} + +/// A stand-in for Stage 2 materialization (G8): the highest computed nodes +/// (not a raw scan or time range) that more than one query reads are +/// computed once per evaluation (`MatAtQuery`). Returns what the selected candidate +/// reuses. +fn materialize_shared(run: &Stage1ToStage3Run) -> Vec { + let mut candidates = run.stage2_physical_asap_optimization.clone(); + for candidate in &mut candidates { + let dag = &mut candidate.dag; + let mut readers: BTreeMap<_, Vec> = BTreeMap::new(); + for (query, &root) in dag.roots.iter().enumerate() { + let mut stack = vec![root]; + while let Some(id) = stack.pop() { + let queries = readers.entry(id).or_default(); + if queries.contains(&query) { + continue; + } + queries.push(query); + stack.extend( + dag.edges + .iter() + .filter(|e| e.consumer == id) + .map(|e| e.producer), + ); + } + } + let shared = |id| readers.get(&id).is_some_and(|q| q.len() > 1); + // The highest shared nodes: read by a root or an unshared node. + let highest: Vec<_> = dag + .nodes + .iter() + .map(|node| node.id) + .filter(|&id| { + shared(id) + && (dag.roots.contains(&id) + || dag + .edges + .iter() + .any(|e| e.producer == id && !shared(e.consumer))) + }) + .collect(); + for node in &mut dag.nodes { + let raw = matches!( + node.payload, + PhysicalASAPOperatorPayload::NonASAP( + NonASAPOp::Scan { .. } | NonASAPOp::TimeRange { .. } + ) + ); + if !raw && highest.contains(&node.id) { + node.materialization = Materialization::MatAtQuery(Medium::Memory); + } + } + } + let queries = run.stage1_pass1_per_computation.roots.len(); + let selection = select_plan( + &candidates, + &vec![None; queries], + &vec![repeating(); queries], + &data(), + PlanningModels::builtin(), + ) + .unwrap(); + let from = &candidates + .iter() + .find(|c| c.id == selection.selected) + .unwrap() + .from_logical; + run.stage1_pass2 + .iter() + .find(|p| &p.id == from) + .map_or_else(Vec::new, |p| p.reused.clone()) +} + +/// The reuse candidates of `queries`, and what the selected candidate +/// reuses when every shared computed node is computed once. +fn selected_reuse(queries: &[&str]) -> Vec { + let run = plan(queries, repeating()); + assert!(run.stage2_pruned.is_empty(), "{:?}", run.stage2_pruned); + let reused: Vec<_> = run + .stage1_pass2 + .iter() + .filter(|p| !p.reused.is_empty()) + .collect(); + assert!(!reused.is_empty(), "{queries:?}"); + let selected = run.selected().unwrap(); + assert!( + !reused.iter().any(|p| p.id == selected.from_logical), + "{queries:?}: not cheaper while every query recomputes what it reads" + ); + materialize_shared(&run) +} + +/// `sum by (job)` beside `sum by (job, instance)`: with the finer sums +/// computed once, plan selection answers the coarse sum from them. +#[test] +fn rollup_is_selected_when_the_finer_sums_are_computed_once() { + assert_eq!( + selected_reuse(&["sum by (job) (m)", "sum by (job, instance) (m)"]), + ["Sum acc for Q1 rolled up from Q2"] + ); +} + +/// Two exact top-k of the per-series rates: with the shared ranking +/// computed once, plan selection reads both from one top-10 ranking +/// instead of sorting every series twice. +#[test] +fn exact_topk_reuse_is_selected_when_the_ranking_is_computed_once() { + assert_eq!( + selected_reuse(&[ + "topk by (job) (5, rate(m[1m]))", + "topk by (job) (10, rate(m[1m]))", + ]), + ["top-k for Q1, Q2 from one top-10"] + ); +} From bc0647d1823661017dc8491f0a7b1255ca0970a8 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 15 Oct 2026 19:25:38 +0000 Subject: [PATCH 4/6] fix(planner): nested sum merge never groups by the dropped metric name, never stacks (#680) A per-series sum drops __name__ while the raw samples keep it, so sum by (__name__) (sum_over_time(m[5m])) merged would group under the metric's name where the nested sums group under none: no merge is offered then. A target that absorbs is now never absorbed, so merges of three nested sums do not stack and combinations() counts what enumeration keeps. --- .../src/per_computation_candidates.rs | 72 +++++++++++++++++-- 1 file changed, 68 insertions(+), 4 deletions(-) diff --git a/crates/asap-aware-mapping/src/per_computation_candidates.rs b/crates/asap-aware-mapping/src/per_computation_candidates.rs index 0fabc7b2..dd273ba5 100644 --- a/crates/asap-aware-mapping/src/per_computation_candidates.rs +++ b/crates/asap-aware-mapping/src/per_computation_candidates.rs @@ -560,13 +560,14 @@ pub fn generate_per_computation_candidates( /// 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 above -/// it), so an absorbing target is never absorbed. +/// 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( targets: &mut [ComputationTarget], consumers: &HashMap<*const OperatorNode, Vec>>>, ) { - let absorbed: HashSet = targets.iter().filter_map(|t| t.absorbs).collect(); + let mut absorbed: HashSet = targets.iter().filter_map(|t| t.absorbs).collect(); for t in 0..targets.len() { let target = &targets[t]; if absorbed.contains(&t) || target.absorbs.is_some() || !target.having.is_empty() { @@ -581,12 +582,16 @@ fn add_nested_sum_candidates( else { continue; }; + // An absorbing target is never absorbed: the inner sum must not + // absorb a sum 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() { continue; } + absorbed.insert(inner); let target = &mut targets[t]; target .choices @@ -600,7 +605,8 @@ fn add_nested_sum_candidates( /// input rows by column, [`partitions_over`]) of a nested-sum merge of /// `target`, or `None` when it does not apply: `target` is a grouped `sum` /// (not by `without`) of the value column of a per-series or grouped `sum` -/// directly beneath it, neither filtered nor with `HAVING`. +/// directly beneath it, neither filtered nor with `HAVING`, and not grouped +/// by the metric name a per-series inner sum drops. fn nested_sum_input(target: &OperatorNode) -> Option<(Rc, AggIntent, Reduction)> { use asap_types::ir::non_asap::any_measure_filtered; let Some(NonASAPOp::Aggregate { @@ -648,6 +654,19 @@ fn nested_sum_input(target: &OperatorNode) -> Option<(Rc, AggInten if summed != value || by.keys().contains(&value) { return None; } + // A per-series sum drops the metric name, which the raw rows keep: + // grouping them by it would name groups the nested sums leave unnamed. + if matches!(inner_reduction, Reduction::PerEntity) + && by.keys().iter().any(|&key| { + inner + .schema + .fields + .get(key) + .is_some_and(|f| f.name == "__name__") + }) + { + return None; + } let partitions = partitions_over(reduction, &inner.schema, &rows.schema)?; Some((Rc::clone(rows), inner_sum.clone(), partitions)) } @@ -4074,6 +4093,51 @@ mod tests { } } + /// A per-series sum drops the metric name, but the raw samples keep it: + /// grouping the merge by `__name__` would answer the metric's name where + /// the nested sums answer none, so no merge is offered. + #[test] + fn nested_sum_merge_never_groups_by_the_dropped_metric_name() { + let query = "sum by (__name__) (sum_over_time(m[5m]))"; + assert_eq!(nested_sum_target(&inventory(query)), None); + } + + /// 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. + #[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(); + assert_eq!(absorbed.len(), 1, "one merge"); + for &a in &absorbed { + assert!( + inventory.targets[a].absorbs.is_none(), + "an absorbed target absorbs" + ); + } + let mut enumerated = 0; + let mut choice = vec![0; inventory.targets.len()]; + loop { + if !inventory.absorbed_choice_varies(&choice) { + enumerated += 1; + } + let mut carry = true; + for (digit, target) in choice.iter_mut().zip(&inventory.targets).rev() { + *digit += 1; + if *digit < target.choices.candidates.len() { + carry = false; + break; + } + *digit = 0; + } + if carry { + break; + } + } + 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 From def6f94e971bc2f09ee750507e843617f4d5271c Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 15 Oct 2026 19:25:38 +0000 Subject: [PATCH 5/6] fix(planner): rollup only over Float64 values; cap rollup and top-k sites together (#680) Exact accumulators execute only Float64 updates, so integer counts are not summed again. Every rollup subset is combined with every top-k reuse subset, so more than 20 sites together list one error instead of 2^n candidates. Document the reuse sites' and variants' fields. --- .../asap-aware-mapping/src/asap_aware_cse.rs | 40 +++++++++- .../src/pass/stage1_to_stage3.rs | 76 ++++++++++++++++++- 2 files changed, 110 insertions(+), 6 deletions(-) diff --git a/crates/asap-aware-mapping/src/asap_aware_cse.rs b/crates/asap-aware-mapping/src/asap_aware_cse.rs index 82442281..8960141e 100644 --- a/crates/asap-aware-mapping/src/asap_aware_cse.rs +++ b/crates/asap-aware-mapping/src/asap_aware_cse.rs @@ -892,7 +892,11 @@ fn rebuild( /// that replaces the coarse build's finalized value. #[derive(Debug, Clone)] pub struct RollupSite { + /// The coarse exact accumulator build (`SummaryAgg`); every reader of + /// it is a `FinalizeExactAccumulator`, which the rollup replaces. pub coarse: Rc, + /// The finer build's `FinalizeExactAccumulator`, already in the + /// candidate, whose values the rollup reads. pub finer: Rc, /// The finalized exact accumulator over `finer`'s values, grouped by /// the coarse keys, with the coarse value's column names. @@ -905,7 +909,11 @@ pub struct RollupSite { /// fails. #[derive(Debug)] pub struct RollupVariant { + /// The sites rolled up (indices into [`rollup_sites`]), ascending. pub sites: Vec, + /// The candidate's roots with those sites rolled up and every guarantee + /// composed again; `Err` when the rebuilt DAG does not derive or + /// compose, which drops only this candidate. pub roots: Result, PerComputationCandidateError>, } @@ -1108,7 +1116,9 @@ fn same_rows(a: &Rc, b: &Rc) -> bool { /// The finalized value of `coarse` (`coarse_value`) as an exact /// accumulator over `finer` (the finer build's finalized values), grouped by /// the coarse keys, finalized, with `coarse_value`'s column names; `None` -/// when it cannot be built or differs from `coarse_value` in type. +/// when it cannot be built, the finer values are not Float64 (the only +/// update type exact accumulators execute), or it differs from +/// `coarse_value` in type. fn rolled_up( coarse: &OperatorNode, finer: &Rc, @@ -1136,6 +1146,15 @@ fn rolled_up( }; // A grouped accumulator outputs its keys, then its value. let value = fine_by.len(); + // Exact accumulators update on Float64 values only: an integer count is + // not summed again. + if finer.schema.fields.get(value).map(|f| &f.dtype) + != Some(&FieldDataType::Plain( + asap_types::pre_asap::DataType::Float64, + )) + { + return None; + } let regrouped = partitions_over(reduction, &child.schema, &finer.schema)?; if regrouped .group_keys() @@ -1295,7 +1314,11 @@ fn replace_all( /// of the input, partitioned as they are. #[derive(Debug, Clone)] pub struct TopKReuseSite { + /// The plain `Aggregate` top-k nodes of one input and partition, in + /// children-first order; at least two different `k`. pub members: Vec>, + /// `Limit(k_max)` over a descending `Sort` of their input, partitioned + /// as they are: what every member reads in a reuse candidate. pub ranking: Rc, } @@ -1305,7 +1328,11 @@ pub struct TopKReuseSite { /// composed again; the error when that fails. #[derive(Debug)] pub struct TopKReuseVariant { + /// The sites reused (indices into [`topk_reuse_sites`]), ascending. pub sites: Vec, + /// The candidate's roots with every member of those sites reading its + /// ranking and every guarantee composed again; `Err` when the rebuilt + /// DAG does not derive or compose, which drops only this candidate. pub roots: Result, PerComputationCandidateError>, } @@ -2086,11 +2113,16 @@ mod tests { } } - /// Counts roll up as the sum of the finer counts; minima as minima. + /// Minima roll up as minima, maxima as maxima. Integer counts do not + /// roll up: exact accumulators execute only Float64 updates. #[test] - fn counts_and_extrema_roll_up() { + fn extrema_roll_up_and_integer_counts_do_not() { + assert!(rollup_sites(&exact(&[ + "count by (job) (m)", + "count by (job, instance) (m)" + ])) + .is_empty()); for (query, kind) in [ - ("count by (job) (m)", ExactKind::Sum), ("min by (job) (m)", ExactKind::Min), ("max by (job) (m)", ExactKind::Max), ] { 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 8c5c4cf1..e896189a 100644 --- a/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs +++ b/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs @@ -51,7 +51,7 @@ use super::{OptimizationInput, OptimizationPass, OptimizeError, PlanOutput, Quer use crate::asap_aware_cse::{ capability_groups, identical_expression_variants, pane_sites, rollup_sites, rollup_variants, sharing_groups, summary_capability_variants, topk_reuse_sites, topk_reuse_variants, - window_composition_variants, + window_composition_variants, MAX_REUSE_SITES, }; use crate::per_computation_candidates::{ compose_logical_candidate, generate_per_computation_candidates, ComputationCandidate, @@ -485,9 +485,22 @@ impl Pass2List { /// The reuse candidates of `roots`: the exact top-k reuse rule's, then /// each rollup candidate followed by its own exact top-k reuse /// candidates; each followed by its identical-expression candidates. + /// They are listed eagerly, like every Pass 2 rule's: more than + /// [`MAX_REUSE_SITES`] rollup and top-k sites together list one error. fn reuse_variants(&mut self, rules: &Rules, roots: &[QueryRoot]) { - self.topk_variants(rules, roots); let sites = rollup_sites(roots); + // Every rollup subset is combined with every top-k subset: cap the + // product, not each factor. + if sites.len() + topk_reuse_sites(roots).len() > MAX_REUSE_SITES { + self.out.push(Pass2::failed( + rules, + PerComputationCandidateError::Unsupported( + "more than 20 rollup and top-k reuse sites together (2^n candidates)", + ), + )); + return; + } + self.topk_variants(rules, roots); let variants = match rollup_variants(roots, &sites) { Ok(variants) => variants, Err(error) => { @@ -900,3 +913,62 @@ impl OptimizationPass for Stage1ToStage3Pass { Ok(output) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_support::lower_promql; + + /// The composed candidate of `queries`, every target at its first + /// (exact) candidate. + fn exact(queries: &[String]) -> Vec { + let roots = queries + .iter() + .enumerate() + .map(|(i, q)| { + let root = lower_promql(q, asap_types::types::AccuracyTarget::exact()); + ( + i, + QueryRoot::Operator(with_promql_series_identity(&root).unwrap()), + ) + }) + .collect(); + let inventory = generate_per_computation_candidates(roots).unwrap(); + compose_logical_candidate(&inventory, &vec![0; inventory.targets.len()]) + .unwrap() + .into_iter() + .map(|(_, root)| root) + .collect() + } + + /// Rollup and exact top-k reuse sites multiply (every subset of one + /// with every subset of the other): past 20 sites together the + /// candidate lists one error instead of enumerating 2^21 candidates. + #[test] + fn reuse_sites_past_20_together_are_an_error() { + let mut queries = Vec::new(); + for i in 0..11 { + queries.push(format!("sum by (job) (m{i})")); + queries.push(format!("sum by (job, instance) (m{i})")); + } + for i in 0..10 { + queries.push(format!("topk by (job) (5, sum by (job, instance) (t{i}))")); + queries.push(format!("topk by (job) (10, sum by (job, instance) (t{i}))")); + } + let roots = exact(&queries); + assert_eq!(rollup_sites(&roots).len(), 11); + assert_eq!(topk_reuse_sites(&roots).len(), 10); + let mut found = Pass2List { + seen: vec![flatten(&roots).0], + out: Vec::new(), + }; + found.reuse_variants(&Rules::default(), &roots); + let [failed] = found.out.as_slice() else { + panic!("one error, {} candidates", found.out.len()) + }; + assert!(matches!( + failed.roots, + Err(PerComputationCandidateError::Unsupported(_)) + )); + } +} From b9957817e6f1f4b3f3b563e8d3a3d5c7d1d5dcb7 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 15 Oct 2026 19:25:38 +0000 Subject: [PATCH 6/6] test(planner): reuse candidates execute to the rows they replace (#680) Every candidate of sum by (__name__) and sum by (job) over sum_over_time answers alike; rollups of sums, minima and maxima and exact top-k reuse answer, query by query, as the candidate they reuse from. --- crates/integration-tests/tests/reuse_rules.rs | 86 +++++++++++++++++++ 1 file changed, 86 insertions(+) diff --git a/crates/integration-tests/tests/reuse_rules.rs b/crates/integration-tests/tests/reuse_rules.rs index 3e26cc1d..83f7d34d 100644 --- a/crates/integration-tests/tests/reuse_rules.rs +++ b/crates/integration-tests/tests/reuse_rules.rs @@ -290,3 +290,89 @@ fn exact_topk_reuse_is_selected_when_the_ranking_is_computed_once() { ["top-k for Q1, Q2 from one top-10"] ); } + +/// Query `q`'s root in the physical candidate Stage 2 made of `logical`. +fn physical_root(run: &Stage1ToStage3Run, logical: &str, q: usize) -> Rc { + let candidate = run + .stage2_physical_asap_optimization + .iter() + .find(|c| c.from_logical == logical) + .unwrap_or_else(|| panic!("{logical} is implemented")); + match &candidate.roots[q] { + QueryRoot::Operator(node) => Rc::clone(node), + QueryRoot::Scalar(_) => panic!("operator root"), + } +} + +/// Every Stage 1 candidate of `query` executes to the same rows. +fn every_candidate_answers_alike(query: &str) { + let run = plan(&[query], QueryRecurrence::Unknown); + assert!(run.stage2_pruned.is_empty(), "{:?}", run.stage2_pruned); + let answers: Vec<(String, Vec)> = run + .stage1_logical_asap_optimization + .iter() + .map(|c| (c.id.clone(), execute(&physical_root(&run, &c.id, 0)))) + .collect(); + for (id, answer) in &answers[1..] { + assert_eq!(answer, &answers[0].1, "{query}: {id} differs from L1"); + } +} + +/// A per-series sum drops the metric name, so grouping by `__name__` over +/// the nested sums groups every series under the empty name; a merge over +/// the raw samples would group them under `m`. Every candidate answers as +/// the nested sums. +#[test] +fn nested_sum_by_the_metric_name_answers_as_the_nested_sums() { + every_candidate_answers_alike("sum by (__name__) (sum_over_time(m[5m]))"); + every_candidate_answers_alike("sum by (job) (sum_over_time(m[5m]))"); +} + +/// Every reuse candidate of `queries` executes, query by query, to the rows +/// of the candidate it reuses from. +fn reuse_answers_alike(queries: &[&str]) { + let run = plan(queries, repeating()); + assert!(run.stage2_pruned.is_empty(), "{:?}", run.stage2_pruned); + let reused: Vec<_> = run + .stage1_pass2 + .iter() + .filter(|p| !p.reused.is_empty()) + .collect(); + assert!(!reused.is_empty(), "{queries:?}"); + for candidate in reused { + for q in 0..queries.len() { + assert_eq!( + execute(&physical_root(&run, &candidate.id, q)), + execute(&physical_root(&run, &candidate.from_logical, q)), + "{queries:?} Q{}: {:?}", + q + 1, + candidate.reused + ); + } + } +} + +/// Coarse sums (over merged nested sums), minima and maxima rolled up from +/// the finer ones answer as computed from the raw samples. +#[test] +fn rollups_answer_as_the_coarse_computations() { + for queries in [ + [ + "sum by (job) (sum_over_time(m[5m]))", + "sum by (job, series) (sum_over_time(m[5m]))", + ], + ["min by (job) (m)", "min by (job, series) (m)"], + ["max by (job) (m)", "max by (job, series) (m)"], + ] { + reuse_answers_alike(&queries); + } +} + +/// Top-k read from one larger ranking answer as computed alone. +#[test] +fn exact_topk_reuse_answers_as_each_topk() { + reuse_answers_alike(&[ + "topk by (job) (1, sum_over_time(m[5m]))", + "topk by (job) (2, sum_over_time(m[5m]))", + ]); +}