diff --git a/crates/devtools/src/candidate_labels.rs b/crates/devtools/src/candidate_labels.rs index f99b0d619..286cf24b1 100644 --- a/crates/devtools/src/candidate_labels.rs +++ b/crates/devtools/src/candidate_labels.rs @@ -86,7 +86,7 @@ pub fn computation_label(candidate: &ComputationCandidate) -> String { /// parentheses), each target listed under the first query that reads it. pub fn choice_label(inventory: &PerComputationCandidates, choice: &[usize]) -> String { let query_of_target = query_of_each_target(inventory); - // Each target's chosen candidate, with the target it absorbs when the + // Each target's chosen candidate, with the targets it absorbs when the // candidate is its fused computation's. let chosen: Vec<_> = inventory .targets @@ -95,7 +95,12 @@ pub fn choice_label(inventory: &PerComputationCandidates, choice: &[usiz .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(); + let absorbed: Vec = chosen + .iter() + .filter_map(|&(_, absorbs)| absorbs) + .flatten() + .copied() + .collect(); (0..inventory.roots.len()) .map(|query| { let mut sketches = Vec::new(); diff --git a/crates/frontend-promql/src/promql.rs b/crates/frontend-promql/src/promql.rs index 863c4c0af..ac0bbc0c9 100644 --- a/crates/frontend-promql/src/promql.rs +++ b/crates/frontend-promql/src/promql.rs @@ -50,6 +50,7 @@ //! | `group(v)` / `count_values("l", v)` | `Aggregate{[Group]}` (constant 1) / `Aggregate{[CountValues{l}]}` (group-by-value + count, new label `l`) — issue #49 | //! | `limitk(k, v)` / `limit_ratio(r, v)` | `PromqlSeriesSample{LimitK(k) \| LimitRatio(r)}` — series-sampling selection, whole series kept unchanged (issue #86) | //! | `topk(k, count_over_time(…))` / `topk(k, sum_over_time(…))` | `Aggregate{[TopK{k}]}` (heavy-hitter intent) over the explicit inner `Aggregate{[Count/Sum]}` | +//! | `topk(k, sum by (G) (count_over_time\|sum_over_time\|rate\|increase\|irate(…)))` | `Aggregate{[TopK{k}]}` over the explicit grouped `Aggregate{[Sum]}` and per-series aggregate | //! | `topk(k, )` / `bottomk(k, …)` | `Sort{value} → Limit{k}` | //! | `m{f}` / `m{f}[w]` | `TimeRange{ingestion, Instant, Scan{predicates}}` / `TimeRange{w, Range, Scan}` | //! | `a OP b` | `BinaryOp{vector_match}` | @@ -790,18 +791,25 @@ fn build_over_sub_dag(outer: Outer, keys: Vec, child: Unresolved) -> child: Rc::new(child), }, Outer::TopK { k, descending } => { - let weighted_counter_ranking = matches!( + // A sum of per-series counter values, or of per-series sums or + // counts of raw data samples (`sum by (G) (sum_over_time(m[W]))`): a + // weighted ranking of the groups. + let weighted_group_ranking = matches!( &child, Unresolved::Aggregate { measures, child: sum_child, .. } if matches!(measures.as_slice(), [AggIntent::Sum { .. }]) - && matches!(sum_child.as_ref(), Unresolved::Aggregate { measures, .. } + && matches!(sum_child.as_ref(), Unresolved::Aggregate { measures, reduction, .. } if matches!( measures.as_slice(), [AggIntent::Rate | AggIntent::Increase | AggIntent::IRate] - )) + ) || (matches!(reduction, Reduction::PerEntity) + && matches!( + measures.as_slice(), + [AggIntent::Sum { .. } | AggIntent::Count { .. }] + ))) ); let direct_counter_ranking = matches!(&child, Unresolved::Aggregate { measures, reduction: Reduction::PerEntity, .. @@ -809,7 +817,7 @@ fn build_over_sub_dag(outer: Outer, keys: Vec, child: Unresolved) -> measures.as_slice(), [AggIntent::Rate | AggIntent::Increase | AggIntent::IRate] )); - if descending && (weighted_counter_ranking || direct_counter_ranking) { + if descending && (weighted_group_ranking || direct_counter_ranking) { return Ok(outer_aggregate( keys, AggIntent::TopK { diff --git a/crates/logical-asap-optimization/src/per_computation_candidates.rs b/crates/logical-asap-optimization/src/per_computation_candidates.rs index 363340b8a..d5170c38e 100644 --- a/crates/logical-asap-optimization/src/per_computation_candidates.rs +++ b/crates/logical-asap-optimization/src/per_computation_candidates.rs @@ -14,8 +14,8 @@ //! 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 the -//! outer target's [`FusedComputation`], which absorbs the one beneath and +//! - A computation that fuses a target with the targets beneath it is the +//! outer target's [`FusedComputation`], which absorbs the ones 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])`). @@ -54,9 +54,10 @@ //! - `Summary`: `SummaryAgg` → `SummaryEstimate`. //! - `Exact` (or a measure that is no target): a single-measure `Aggregate`. //! -//! A fused computation's candidate reads the absorbed target's input instead: +//! A fused computation's candidate reads the absorbed targets' 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 +//! item → `SummaryEstimate` (over a grouped sum of per-series sums or counts, +//! it reads the per-series target's input, the raw data samples); 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 @@ -278,16 +279,20 @@ pub struct ComputationTarget { pub raw_data_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`. +/// A target's measure computed together with the targets beneath it, which +/// it absorbs (they are 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, + /// The indices in [`PerComputationCandidates::targets`] of the targets + /// it absorbs: the one directly beneath, then, for a whole-expression + /// heap over the raw data samples beneath a grouped sum of per-series sums or + /// counts (`topk(k, sum by (G) (sum_over_time(m[W])))`), the per-series + /// target beneath that, then the targets any of these targets' own fused + /// computations absorb. None of them is computed when this is picked. + pub absorbs: Vec, /// How the fused computation is computed: exact or a summary. Empty /// when Pass 1 pruned every one ([`Self::pruned`]); the fusion is then /// not offered. @@ -310,16 +315,22 @@ impl ComputationTarget { } /// How the measure is computed when this target's digit of a - /// whole-workload choice is `index`: the candidate, and the target it + /// whole-workload choice is `index`: the candidate, and the targets 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)> { + pub fn computation_for( + &self, + index: usize, + ) -> Option<(&ComputationCandidate, Option<&[usize]>)> { let own = self.choices.candidates.len(); match self.choices.candidates.get(index) { Some(candidate) => Some((candidate, None)), None => { let fused = self.fused.as_ref()?; - Some((fused.candidates.get(index - own)?, Some(fused.absorbs))) + Some(( + fused.candidates.get(index - own)?, + Some(fused.absorbs.as_slice()), + )) } } } @@ -339,39 +350,54 @@ pub struct PerComputationCandidates { impl PerComputationCandidates { /// The number of whole-workload candidates: the size of the Cartesian /// product of the targets' candidates, where a target with a fused - /// 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. + /// computation and the targets beneath that it absorbs count as their + /// own combinations plus the fused computation's candidates (the + /// absorbed targets are not computed then), saturating at `usize::MAX`. + /// A target with `n` candidates and `f` fused ones counts `n · Π m + f`, + /// the product over the targets it absorbs that no other of them + /// absorbs, each counted the same way. It is exactly the number of + /// choices enumeration keeps ([`Self::absorbed_choice_varies`]). pub fn combinations(&self) -> usize { 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(fused) = &target.fused else { - return n; - }; - n.saturating_mul(self.targets[fused.absorbs].choices.candidates.len()) - .saturating_add(fused.candidates.len()) - }) + (0..self.targets.len()) + .filter(|index| !absorbed.contains(index)) + .map(|index| self.combinations_of(index)) .fold(1usize, usize::saturating_mul) } + /// [`Self::combinations`] of target `index` with the targets it absorbs. + fn combinations_of(&self, index: usize) -> usize { + let target = &self.targets[index]; + let n = target.choices.candidates.len(); + let Some(fused) = &target.fused else { + return n; + }; + let absorbed_by_another = |a: usize| { + fused.absorbs.iter().any(|&b| { + self.targets[b] + .fused + .as_ref() + .is_some_and(|inner| inner.absorbs.contains(&a)) + }) + }; + fused + .absorbs + .iter() + .filter(|&&a| !absorbed_by_another(a)) + .map(|&a| self.combinations_of(a)) + .fold(n, usize::saturating_mul) + .saturating_add(fused.candidates.len()) + } + /// Whether `choice` picks a fused computation and a candidate other - /// than the first for the target it absorbs: the same whole-workload + /// than the first for a 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 .computation_for(index) .and_then(|(_, absorbed)| absorbed) - .is_some_and(|inner| choice[inner] != 0) + .is_some_and(|inner| inner.iter().any(|&i| choice[i] != 0)) }) } @@ -417,7 +443,8 @@ impl PerComputationCandidates { fn absorbed_targets(targets: &[ComputationTarget]) -> impl Iterator + '_ { targets .iter() - .filter_map(|target| Some(target.fused.as_ref()?.absorbs)) + .filter_map(|target| target.fused.as_ref()) + .flat_map(|fused| fused.absorbs.iter().copied()) } #[derive(Debug, Error)] @@ -684,9 +711,44 @@ pub fn generate_per_computation_candidates( add_whole_expression_targets(&mut targets, &consumers); prune_topk_heaps_over_one_row_per_item(&mut targets); add_nested_sum_targets(&mut targets, &consumers); + close_absorptions(&mut targets); Ok(PerComputationCandidates { roots, targets }) } +/// Adds to each fused computation's `absorbs` the targets its absorbed +/// targets' own fused computations absorb, transitively: none of them is +/// computed when the fused computation is picked, so enumeration keeps only +/// their first choice ([`PerComputationCandidates::absorbed_choice_varies`]) +/// and [`PerComputationCandidates::combinations`] counts exactly those. +fn close_absorptions(targets: &mut [ComputationTarget]) { + loop { + let mut changed = false; + for t in 0..targets.len() { + let Some(fused) = &targets[t].fused else { + continue; + }; + let beneath: Vec = fused + .absorbs + .iter() + .filter_map(|&a| targets[a].fused.as_ref()) + .flat_map(|inner| inner.absorbs.iter().copied()) + .filter(|b| !fused.absorbs.contains(b)) + .collect(); + if let Some(fused) = targets[t].fused.as_mut().filter(|_| !beneath.is_empty()) { + for b in beneath { + if !fused.absorbs.contains(&b) { + fused.absorbs.push(b); + } + } + changed = true; + } + } + if !changed { + return; + } + } +} + /// Nested-sum merge: a grouped `sum` over the value of a `sum` target /// beneath it gets a fused computation ([`ComputationTarget::fused`]) /// with the one candidate `ExactAccumulator([Sum])`: one exact `Sum` @@ -694,15 +756,17 @@ pub fn generate_per_computation_candidates( /// 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 computation yet, and the inner sum has none, so a -/// target with a fused computation is never absorbed and merges do not -/// stack. +/// ([`nested_sum_input`]), and when the outer sum is not itself absorbed by +/// a merge and has no fused computation yet, and the inner sum has none, so +/// merges do not stack. An outer sum a whole-expression top-k absorbs +/// (`topk(k, sum by (G) (sum_over_time(m[W])))`) keeps its merge for the +/// top-k's own candidates. fn add_nested_sum_targets( targets: &mut [ComputationTarget], consumers: &HashMap<*const OperatorNode, Vec>>>, ) { - let mut absorbed: HashSet = absorbed_targets(targets).collect(); + // The targets a merge absorbs. + let mut absorbed: HashSet = HashSet::new(); for t in 0..targets.len() { let target = &targets[t]; if absorbed.contains(&t) || target.fused.is_some() || !target.having.is_empty() { @@ -717,8 +781,8 @@ fn add_nested_sum_targets( else { continue; }; - // A target with a fused computation is never absorbed: the inner sum - // must not have one of its own. + // Merges do not stack: the inner sum must not have a fused + // computation of its own. if consumers.get(&Rc::as_ptr(child)).map(Vec::len) != Some(1) || !targets[inner].having.is_empty() || targets[inner].fused.is_some() @@ -729,7 +793,7 @@ fn add_nested_sum_targets( absorbed.insert(inner); let merge = ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum]); targets[t].fused = Some(FusedComputation { - absorbs: inner, + absorbs: vec![inner], candidates: vec![merge], pruned: Vec::new(), }); @@ -813,7 +877,9 @@ fn nested_sum_input(target: &OperatorNode) -> Option<(Rc, AggInten /// 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 -/// computation replaces them. Applied only when the inner aggregate has no other +/// computation replaces them. Over a grouped sum of per-series sums or +/// counts of raw data samples, the heap reads those samples and absorbs the +/// per-series target too. 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`]). @@ -837,10 +903,32 @@ fn add_whole_expression_targets( || !target.having.is_empty() || targets[inner].filter.is_some() || !targets[inner].having.is_empty() - || whole_expression_input(&target.aggregate).is_none() { continue; } + let Some((input, ..)) = whole_expression_input(&target.aggregate) else { + continue; + }; + // A heap over the raw data samples beneath a grouped sum of per-series + // sums or counts also absorbs the per-series target, when the + // grouped sum is its only consumer. + let mut absorbs = vec![inner]; + if let Some(NonASAPOp::Aggregate { + child: per_series, .. + }) = targets[inner].aggregate.non_asap() + { + let reads_below = per_series + .children() + .iter() + .any(|&rows| Rc::ptr_eq(rows, &input)); + if reads_below && consumers.get(&Rc::as_ptr(per_series)).map(Vec::len) == Some(1) { + absorbs.extend( + targets + .iter() + .position(|other| Rc::ptr_eq(&other.aggregate, per_series)), + ); + } + } let (heaps, own): (Vec<_>, Vec<_>) = targets[t] .choices .candidates @@ -860,7 +948,7 @@ fn add_whole_expression_targets( let target = &mut targets[t]; target.choices.candidates = own; target.fused = Some(FusedComputation { - absorbs: inner, + absorbs, candidates: heaps, pruned: Vec::new(), }); @@ -1084,7 +1172,7 @@ pub fn compose_logical_candidate( .push(ChosenComputation { target, candidate, - absorbs, + fused: absorbs.is_some(), }); } // Consumer `Filter`s whose key-only `HAVING` moves below a substituted @@ -1348,9 +1436,9 @@ struct ChosenComputation<'a> { 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, + /// Whether the candidate is the target's fused computation's, which + /// absorbs the targets beneath; `false` for one of its own candidates. + fused: bool, } /// Each aggregate's targets with their chosen computations. @@ -2678,7 +2766,7 @@ fn substitute( ( chosen.candidate, chosen.target.raw_data_samples_of.as_deref(), - chosen.absorbs.is_some(), + chosen.fused, ) }); let filter = filters.get(measure).and_then(Option::as_ref).map(rewritten); @@ -4208,7 +4296,7 @@ mod tests { ComputationCandidate::Summary(SketchAlgorithm::CountSketchWithHeap), ] ); - let inner = fused.absorbs; + let inner = fused.absorbs[0]; assert_eq!(inventory.targets[inner].choices.candidates.len(), 1); assert_eq!(inventory.combinations(), 3); for (candidate, root) in composed_with(&inventory, topk) { @@ -4281,6 +4369,150 @@ mod tests { ); } + /// How many whole-workload choices enumeration keeps: every digit over + /// its target's `choice_count`, less those [`absorbed_choice_varies`] + /// skips. + /// + /// [`absorbed_choice_varies`]: PerComputationCandidates::absorbed_choice_varies + fn enumerated_choices(inventory: &PerComputationCandidates) -> usize { + 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.choice_count() { + carry = false; + break; + } + *digit = 0; + } + if carry { + return enumerated; + } + } + } + + /// `topk(k, sum by (G) (sum_over_time(m[W])))` and its `count_over_time` + /// form rank the groups `G` by the sum (count) of their raw data samples: the + /// top-k keeps its exact candidate and gets the fused whole-expression + /// Count-Min and CountSketch heaps, which absorb both the grouped sum and + /// the per-series sum (count) beneath it. Composed, no aggregate is + /// computed: the heap reads the range selector's raw data samples, keyed by + /// `G` (less the top-k's own `by` labels, its partitions), weighted by + /// the sample value (a sum, an open requirement on `m`) or by one (a + /// count, proven non-negative). + #[test] + fn whole_expression_topk_over_a_grouped_sum_of_range_sums_reads_the_raw_data_samples() { + let samples = WeightDomain::RawDataSamplesOfMetric { metric: "m".into() }; + let unit = WeightDomain::NonNegative { + proof: NonNegativeWeightProof::UnitCount, + }; + let value = SummaryInputExpr::Column(ColumnRef::SampleValue); + let one = SummaryInputExpr::Constant(1.0); + for (query, weight, domain, partitions) in [ + ( + "topk(3, sum by (region) (sum_over_time(m[1m])))", + &value, + &samples, + &[][..], + ), + ( + "topk(3, sum by (region) (count_over_time(m[1m])))", + &one, + &unit, + &[], + ), + ( + "topk by (job) (3, sum by (job, region) (sum_over_time(m[1m])))", + &value, + &samples, + &["job"], + ), + ] { + let inventory = inventory(query); + let topk = topk_target(&inventory); + let target = &inventory.targets[topk]; + assert_eq!( + target.choices.candidates, + [ComputationCandidate::Exact], + "{query}" + ); + let fused = target.fused.as_ref().expect("a fused computation"); + assert_eq!( + fused.candidates, + [ + ComputationCandidate::Summary(SketchAlgorithm::CmsWithHeap), + ComputationCandidate::Summary(SketchAlgorithm::CountSketchWithHeap), + ], + "{query}" + ); + assert_eq!(fused.absorbs.len(), 2, "{query}: both aggregates beneath"); + assert_eq!( + inventory.combinations(), + enumerated_choices(&inventory), + "{query}" + ); + let mut heaps = 0; + for (candidate, root) in composed_with(&inventory, topk) { + if candidate == ComputationCandidate::Exact { + continue; + } + assert!( + !OperatorNode::reachable(&root).iter().any(|n| matches!( + n.non_asap(), + Some(NonASAPOp::Aggregate { .. }) + ) || matches!( + n.asap(), + Some(ASAPOp::FinalizeExactAccumulator { .. }) + )), + "{query} {candidate:?}: no sum is computed" + ); + let [state] = &heap_states(&root)[..] else { + panic!("{query}: one heap") + }; + let Some(ASAPOp::SummaryAgg { + child, + input, + reduction, + .. + }) = state.asap() + else { + unreachable!() + }; + assert!( + matches!(child.non_asap(), Some(NonASAPOp::TimeRange { .. })), + "{query}: the heap reads the raw data samples" + ); + assert_eq!( + input.item, + Some(SummaryInputExpr::Column(ColumnRef::Named("region".into()))), + "{query}" + ); + assert_eq!(&input.weight, weight, "{query}"); + assert_eq!(&input.weight_domain, domain, "{query}"); + let names: Vec<_> = reduction + .group_keys() + .map_or(&[][..], |keys| keys.keys()) + .iter() + .map(|&index| child.schema.fields[index].name.as_str()) + .collect(); + assert_eq!(names, partitions, "{query}"); + // The answer has the top-k's columns, `region` among them. + assert!( + root.schema.fields.iter().any(|f| f.name == "region"), + "{query}: {:?}", + root.schema + ); + heaps += 1; + } + assert_eq!(heaps, 2, "{query}"); + } + } + /// The weight domain of every heap-sketch update in every candidate of /// `query`'s top-k. fn heap_weight_domains(query: &str) -> Vec { @@ -4386,7 +4618,7 @@ mod tests { assert_eq!(target.choices.candidates, std::slice::from_ref(&sum)); let fused = target.fused.as_ref().expect("a fused computation"); assert_eq!(fused.candidates, [sum]); - let inner = fused.absorbs; + let inner = fused.absorbs[0]; assert_eq!(inventory.targets[inner].choices.candidates.len(), 1); assert_eq!(inventory.combinations(), 2); let composed = composed_with(&inventory, outer); @@ -4477,26 +4709,70 @@ mod tests { "an absorbed target has a fused computation" ); } - 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.choice_count() { - carry = false; - break; - } - *digit = 0; - } - if carry { - break; + let enumerated = enumerated_choices(&inventory); + assert_eq!(inventory.combinations(), enumerated); + } + + /// The shapes a heap over the raw data samples cannot answer do not get one: + /// `bottomk` stays a sort and limit (no top-k target), and a grouped sum + /// `without` labels or by the metric name (which the per-series sum drops) + /// gets no heap that absorbs the per-series sum and reads the samples. + #[test] + fn grouped_range_sums_a_raw_data_samples_heap_cannot_rank_get_no_fused_heap() { + let is_topk = |t: &ComputationTarget| { + matches!( + t.aggregate.non_asap(), + Some(NonASAPOp::Aggregate { measures, .. }) + if matches!(measures.as_slice(), [AggIntent::TopK { .. }]) + ) + }; + let bottom = inventory("bottomk(3, sum by (region) (sum_over_time(m[1m])))"); + assert!( + !bottom.targets.iter().any(is_topk), + "bottomk has no top-k target" + ); + for query in [ + "topk(3, sum without (instance) (sum_over_time(m[1m])))", + "topk(3, sum by (__name__) (sum_over_time(m[1m])))", + ] { + let inventory = inventory(query); + for target in inventory.targets.iter().filter(|t| is_topk(t)) { + assert!( + target.fused.as_ref().is_none_or(|f| f.absorbs.len() == 1), + "{query}: {:?}", + target.fused + ); } } - assert_eq!(inventory.combinations(), enumerated); + } + + /// A fused computation picked leaves every target beneath it uncomputed, + /// also one that an absorbed target's own merge absorbs: Pass 1 closes + /// `absorbs` over those, so enumeration keeps only their first choice and + /// `combinations` still counts exactly the choices kept. Here the top-k + /// heap is made to absorb the grouped sum only, whose merge absorbs the + /// per-series sum. + #[test] + fn a_fused_computation_absorbs_what_its_absorbed_targets_merges_absorb() { + let mut inventory = inventory("topk(10, sum by (job) (sum_over_time(m[1m])))"); + let topk = topk_target(&inventory); + let fused = inventory.targets[topk] + .fused + .as_mut() + .expect("a fused heap"); + let grouped = fused.absorbs[0]; + fused.absorbs.truncate(1); + let per_series = inventory.targets[grouped] + .fused + .as_ref() + .expect("the grouped sum's merge") + .absorbs[0]; + close_absorptions(&mut inventory.targets); + assert_eq!( + inventory.targets[topk].fused.as_ref().unwrap().absorbs, + vec![grouped, per_series] + ); + assert_eq!(inventory.combinations(), enumerated_choices(&inventory)); } /// `combinations` counts each choice once: picking a fused @@ -4512,29 +4788,11 @@ mod tests { let Some(fused) = &target.fused else { panic!("a fused computation: {:?}", target.choices.candidates) }; - let inner = fused.absorbs; + let inner = fused.absorbs[0]; let n = target.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()]; - 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.choice_count() { - carry = false; - break; - } - *digit = 0; - } - if carry { - break; - } - } + let enumerated = enumerated_choices(&inventory); assert!(whole > 0 && m > 1, "{n} {whole} {m}"); assert_eq!(inventory.combinations(), n * m + whole); assert_eq!(inventory.combinations(), 6); diff --git a/crates/logical-asap-optimization/src/realization.rs b/crates/logical-asap-optimization/src/realization.rs index 9442edc29..40c524f55 100644 --- a/crates/logical-asap-optimization/src/realization.rs +++ b/crates/logical-asap-optimization/src/realization.rs @@ -305,9 +305,18 @@ pub(crate) fn realize_keyed_additive_summary_input( _ => None, }; let counter_input = derivative_of.is_some(); + // `sum by (G) (sum_over_time(m[W]))` or `... count_over_time ...`: the + // grouped sum of per-series sums (counts) of raw data samples is their sum + // (count) by `G`, so the heap reads those samples. + let samples = summed_per_series_samples(child); + // A raw data sample's count or a count's row weighs one. + let counted = match &samples { + Some((_, inner)) => matches!(inner, AggIntent::Count { .. }), + None => matches!(measures.as_slice(), [AggIntent::Count { .. }]), + }; let weight = match measures.as_slice() { - [AggIntent::Count { .. }] => SummaryInputExpr::Constant(1.0), - [AggIntent::Sum { .. }] if counter_input => { + _ if counted => SummaryInputExpr::Constant(1.0), + [AggIntent::Sum { .. }] if counter_input || samples.is_some() => { SummaryInputExpr::Column(ColumnRef::SampleValue) } [AggIntent::Sum { col }] => SummaryInputExpr::Column(match col { @@ -322,14 +331,12 @@ pub(crate) fn realize_keyed_additive_summary_input( }), _ => return None, }; - let weight_domain = match measures.as_slice() { - [AggIntent::Count { .. }] => WeightDomain::NonNegative { + let weight_domain = match derivative_of { + _ if counted => WeightDomain::NonNegative { proof: NonNegativeWeightProof::UnitCount, }, - _ => match derivative_of { - Some(Some(metric)) => WeightDomain::CounterDerivativeOfMetric { metric }, - _ => WeightDomain::UnknownOrSigned, - }, + Some(Some(metric)) => WeightDomain::CounterDerivativeOfMetric { metric }, + _ => WeightDomain::UnknownOrSigned, }; if matches!(heap_algorithm, SketchAlgorithm::CmsWithHeap) && !matches!( @@ -384,7 +391,7 @@ pub(crate) fn realize_keyed_additive_summary_input( } }; Some(PhysicalSummaryInput { - child: Rc::clone(raw_child), + child: samples.map_or_else(|| Rc::clone(raw_child), |(rows, _)| rows), input: SummaryUpdate { item: Some(item), weight, @@ -393,6 +400,67 @@ pub(crate) fn realize_keyed_additive_summary_input( }) } +/// The raw data sample rows and the per-series `sum` or `count` over them when +/// `grouped` is a grouped `sum` (not by `without`, nor by the metric name +/// the per-series result drops) of the value of a per-series `sum` or +/// `count` of raw data samples (`sum by (G) (sum_over_time(m[W]))`), neither +/// filtered nor with `HAVING`; `None` otherwise. The grouped sum is then the +/// sum (count) of those samples by `G`. +fn summed_per_series_samples(grouped: &OperatorNode) -> Option<(Rc, AggIntent)> { + use asap_types::ir::operator::non_asap::any_measure_filtered; + let Some(NonASAPOp::Aggregate { + reduction: Reduction::Reduce(by), + measures, + filters, + having: None, + child: per_series, + .. + }) = grouped.non_asap() + else { + return None; + }; + let [AggIntent::Sum { col }] = measures.as_slice() else { + return None; + }; + let Some(NonASAPOp::Aggregate { + reduction: Reduction::PerEntity, + measures: inner, + filters: inner_filters, + having: None, + child: rows, + .. + }) = per_series.non_asap() + else { + return None; + }; + let [inner @ (AggIntent::Sum { col: None } | AggIntent::Count { .. })] = inner.as_slice() + else { + return None; + }; + if by.is_without() + || any_measure_filtered(filters) + || any_measure_filtered(inner_filters) + || matches!(rows.non_asap(), Some(NonASAPOp::Aggregate { .. })) + { + return None; + } + // The grouped sum reads the per-series value. + let value = resolve_column_ref(&ColumnRef::SampleValue, &per_series.schema).ok()?; + if col.is_some_and(|col| col != value) || by.keys().contains(&value) { + return None; + } + if by.keys().iter().any(|&key| { + per_series + .schema + .fields + .get(key) + .is_none_or(|field| field.name == "__name__") + }) { + return None; + } + Some((Rc::clone(rows), inner.clone())) +} + pub(crate) fn schema_column_ref(child: &OperatorNode, index: usize) -> Option { let column = child.schema.fields.get(index)?; Some(match &column.table { diff --git a/crates/plan-selection/src/pass/query_candidates.rs b/crates/plan-selection/src/pass/query_candidates.rs index 46a582bc4..1647bb116 100644 --- a/crates/plan-selection/src/pass/query_candidates.rs +++ b/crates/plan-selection/src/pass/query_candidates.rs @@ -167,7 +167,7 @@ fn query_digits( || per_computation.targets[t] .fused .as_ref() - .is_some_and(|fused| read[fused.absorbs]) + .is_some_and(|fused| fused.absorbs.iter().any(|&a| read[a])) }) .collect() } diff --git a/crates/planner/tests/runtime_install.rs b/crates/planner/tests/runtime_install.rs index 84ff3637f..8e577077e 100644 --- a/crates/planner/tests/runtime_install.rs +++ b/crates/planner/tests/runtime_install.rs @@ -2696,3 +2696,128 @@ async fn a_forced_heap_counter_topk_plan_compiles_and_answers_within_its_bound() } } } + +/// `data` samples of 40 regions with 3 series each, inside the last +/// minute: region `r` (rank 1..=40) sums to `1000 / r^1.1` over +/// its series (a skewed ranking of the regions) and holds `3 · +/// round(200 / r^1.1)` samples (a skewed count ranking), returned with each +/// region's exact sum and exact count. +fn skewed_region_raw_data_samples() -> ( + Vec, + BTreeMap, + BTreeMap, +) { + let mut samples = Vec::new(); + let mut sums = BTreeMap::new(); + let mut counts = BTreeMap::new(); + for rank in 1..=40i64 { + let region = format!("r{rank:02}"); + let sum = 1000.0 / (rank as f64).powf(1.1); + let per_series_steps = (200.0 / (rank as f64).powf(1.1)).round().max(1.0) as i64; + for series in 0..3 { + let instance = format!("{region}-{series}"); + for step in 0..per_series_steps { + // Spread over (EVALUATION_MS - 59 s, EVALUATION_MS - 1 s]. + let at = EVALUATION_MS - 1_000 - step * 58_000 / per_series_steps; + let mut s = sample("data", &instance, at, sum / 3.0 / per_series_steps as f64); + s.labels.insert("region".into(), region.clone()); + samples.push(s); + } + } + sums.insert(region.clone(), sum); + counts.insert(region, (3 * per_series_steps) as f64); + } + samples.sort_by_key(|s| s.timestamp_ms); + (samples, sums, counts) +} + +/// The ASAPQuery workloads' grouped top-k-sum shape, +/// `topk(3, sum by (region) ((data[1m])))` for `f` `sum_over_time` or +/// `count_over_time`: each forced fused heap (one over the raw data samples, +/// keyed by `region`, weighted by the sample value or by one) compiles and +/// answers the true top 3 regions within its bound of their exact sums or +/// counts, each named by its `region` label alone, as the exact top-k's. +/// A Count-Min heap over sample values needs the gauge declared +/// non-negative; a Count Sketch heap takes any gauge. +#[tokio::test] +async fn a_forced_heap_grouped_topk_sum_plan_compiles_and_answers_within_its_bound() { + let (samples, sums, counts) = skewed_region_raw_data_samples(); + let gauge = r#""metric_type": "gauge""#; + let non_negative = r#""metric_type": "gauge", "values_non_negative": true"#; + for (function, truth, facts, algorithm) in [ + ( + "sum_over_time", + &sums, + non_negative, + SketchAlgorithm::CmsWithHeap, + ), + ( + "sum_over_time", + &sums, + gauge, + SketchAlgorithm::CountSketchWithHeap, + ), + ( + "count_over_time", + &counts, + gauge, + SketchAlgorithm::CmsWithHeap, + ), + ( + "count_over_time", + &counts, + gauge, + SketchAlgorithm::CountSketchWithHeap, + ), + ] { + let query = format!("topk(3, sum by (region) ({function}(data[1m])))"); + let workload = heap_topk_workload(&query, "data", facts); + let truth = BTreeMap::from([(String::new(), truth.clone())]); + let plan = plan_ignoring_guarantees(&workload, Some(algorithm.clone())) + .await + .unwrap_or_else(|error| panic!("{query} {algorithm:?}: {error}")); + let (built, width) = heap_of(&plan).expect("a heap plan"); + assert_eq!(built, algorithm, "{query}"); + plan.check_compiles().unwrap(); + let plan_json = serde_json::to_string(&plan).unwrap(); + let run = run_data_plane( + &plan_json, + &samples, + EVALUATION_MS, + &mut SummaryStore::new(), + ); + for answers in [ + &run.answers_per_query[0], + &run.reference_answers_per_query[0], + ] { + assert_heap_topk_answers(answers, &truth, (None, "region", 3), (&algorithm, width)); + for (labels, _) in answers.iter() { + assert_eq!( + labels.keys().collect::>(), + ["region"], + "{query} {algorithm:?}: the group label alone" + ); + } + } + // The installed plan (stored states) answers as its query-time + // reference does, within the heap's bound. + let bound = heap_bound( + &algorithm, + width, + &truth[""].values().copied().collect::>(), + ); + let reference: BTreeMap<_, _> = run.reference_answers_per_query[0] + .iter() + .map(|(labels, value)| (labels.clone(), *value)) + .collect(); + for (labels, value) in &run.answers_per_query[0] { + let expected = reference + .get(labels) + .unwrap_or_else(|| panic!("{query} {algorithm:?}: {labels:?} in the reference")); + assert!( + (value - expected).abs() <= bound, + "{query} {algorithm:?}: {value} vs {expected}" + ); + } + } +} diff --git a/docs/design_docs/proposals/planner-layering.md b/docs/design_docs/proposals/planner-layering.md index e2206f8fa..875c38917 100644 --- a/docs/design_docs/proposals/planner-layering.md +++ b/docs/design_docs/proposals/planner-layering.md @@ -304,15 +304,17 @@ 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 the -outer target's **fused computation**, which absorbs the target beneath: the +A computation that fuses a target with the targets beneath it is the +outer target's **fused computation**, which absorbs those targets: 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 -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. +computation's Count-Min or Count Sketch with a top-*k* heap. In +`topk(10, sum by (job) (sum_over_time(m[1m])))` the heap reads the raw data samples +beneath the grouped sum, so it absorbs both the grouped sum and the per-series +sum. Like every target's options, a fused computation's are exact or a summary. Example for summary candidates: diff --git a/docs/develop_docs/asap-aware-mapping-architecture.md b/docs/develop_docs/asap-aware-mapping-architecture.md index 7d586e9bf..924c4de70 100644 --- a/docs/develop_docs/asap-aware-mapping-architecture.md +++ b/docs/develop_docs/asap-aware-mapping-architecture.md @@ -80,7 +80,7 @@ populations for instant top-k, quantile and sum) in Stage 1, and one exact option plus the summary families of `summary_candidates` that the accuracy target permits and the deployment supports. A whole-expression top-k or a nested sum also gets a fused computation that absorbs the - target beneath it. + targets beneath it. 2. **Enumeration** takes one choice per target and composes the logical DAG. For each composed DAG, the Pass 2 rules find what can be shared, split into panes or reused, and each subset of their candidates becomes another diff --git a/docs/develop_docs/asap-aware-mapping-contracts.md b/docs/develop_docs/asap-aware-mapping-contracts.md index 81051a8ae..c5f5b527a 100644 --- a/docs/develop_docs/asap-aware-mapping-contracts.md +++ b/docs/develop_docs/asap-aware-mapping-contracts.md @@ -46,10 +46,12 @@ asap-physical-asap-optimization -p asap-plan-selection -p asap-types dropped. Pass 1 prunes only what it can prove; accuracy that depends on data facts is left to Stage 3. - **Fused computations.** `FusedComputation { absorbs, candidates, pruned }` - replaces the target directly beneath (the whole-expression top-k, the - nested sum). A whole-workload choice keeps one digit per target; a fused - pick with a non-first digit on the absorbed target is skipped, so each - plan is enumerated once. + replaces the targets beneath (the whole-expression top-k, the nested sum). + It absorbs the target directly beneath and, for a heap over the raw data samples + beneath a grouped sum of per-series sums or counts, the per-series target + too. A whole-workload choice keeps one digit per target; a fused pick with + a non-first digit on an absorbed target is skipped, so each plan is + enumerated once. - **Pass 2.** Each rule adds candidates; none removes a Pass 1 candidate, adds a summary family, or adds a readout. So a deployment that does not support a family never sees it again after Pass 1. A Pass 2 candidate records what it