From 0129a2767ba08156a94e28d6ebd86b890443fd5f Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 22 Oct 2026 11:54:09 +0000 Subject: [PATCH 1/2] feat(pass2): coarse sketch groupings roll up from the finest build of the same sketch The rollup rule also offers sketch rollups: a coarse per-group sketch build read only by estimates with a guaranteed merged readout is answered by a rollup (SummaryMerge with group_by) of the finest build of the same algorithm over the same rows, sized for every reader. Every coarse grouping of one finest build is one choice. Guarantees compose through the rollup and Stage 3 prices its merge and output groups. --- .../tests/sketch_rollup_planning.rs | 261 ++++++++++ .../src/asap_aware_cse.rs | 467 ++++++++++++++++-- .../src/per_computation_candidates.rs | 58 +++ .../src/workload_candidates.rs | 11 +- crates/plan-selection/src/lib.rs | 18 +- .../analytical-resource-cost.md | 7 +- .../design_docs/proposals/planner-layering.md | 2 +- 7 files changed, 774 insertions(+), 50 deletions(-) create mode 100644 crates/integration-tests/tests/sketch_rollup_planning.rs diff --git a/crates/integration-tests/tests/sketch_rollup_planning.rs b/crates/integration-tests/tests/sketch_rollup_planning.rs new file mode 100644 index 000000000..524e78bc5 --- /dev/null +++ b/crates/integration-tests/tests/sketch_rollup_planning.rs @@ -0,0 +1,261 @@ +//! Sketch rollup in planning: Pass 2 offers answering a coarse sketch +//! grouping from the finest build of the same sketch over the same rows, +//! and Stage 3 selects it when one build and a merge cost less than two +//! builds over the raw rows. +use std::rc::Rc; + +use asap_executor::physical_planner::promql_rows; +use asap_frontend_sql::{lower_sql, SqlCatalog}; +use asap_integration_tests::fixtures::lower_promql; +use asap_plan_selection::{run_stage1_to_stage3_with, PlanningModels, Stage1ToStage3Run}; +use asap_types::ir::properties::{AccuracyMetric, AccuracyScope, AccuracyTarget}; +use asap_types::ir::schema::{DataType, Field, Schema}; +use asap_types::ir::{ASAPOp, OperatorNode, QueryRoot}; +use asap_types::workload::{ + DataWorkload, Predictability, QueryRecurrence, QueryRequirements, QueryWorkloadEntry, + RepeatedDemand, RepetitionInterval, +}; + +fn catalog() -> SqlCatalog { + SqlCatalog::new().with_table( + "http", + Schema::new(vec![ + Field::plain("ts", DataType::Timestamp, false), + Field::plain("region", DataType::Utf8, false), + Field::plain("service", DataType::Utf8, false), + Field::plain("user_id", DataType::Float64, false), + Field::plain("latency", DataType::Float64, false), + ]), + ) +} + +/// A predictable query repeating every `interval_ms` at `accuracy`. +fn entry(accuracy: &AccuracyTarget, interval_ms: u32) -> QueryWorkloadEntry { + QueryWorkloadEntry { + query: Default::default(), + requirements: QueryRequirements { + accuracy: accuracy.clone(), + max_latency_ms: None, + }, + predictability: Predictability::Predictable { known_at: None }, + recurrence: QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(RepetitionInterval( + interval_ms, + ))), + range_evaluation: None, + } +} + +/// `http` arrives at 2e6 rows/s (W-RollupHeavy's table), 10,000 series, +/// a million distinct users at most per evaluation. +fn data() -> DataWorkload { + serde_json::from_value(serde_json::json!({ + "arrival": {"ingesting": {"rows_per_second": 2_000_000.0}}, + "sources": {"http": { + "series_count": 10_000, + "sample_interval": 5, + "values_non_negative": true, + "max_distinct_per_evaluation": 1_000_000, + }}, + })) + .unwrap() +} + +/// Every stage's output for the SQL `queries` at `accuracy`. +async fn plan_sql(queries: &[&str], accuracy: AccuracyTarget) -> Stage1ToStage3Run { + let mut roots = Vec::new(); + for (i, query) in queries.iter().enumerate() { + let root = lower_sql(query, &catalog(), accuracy.clone()) + .await + .unwrap(); + roots.push((i, QueryRoot::Operator(root))); + } + run_stage1_to_stage3_with( + roots, + &vec![entry(&accuracy, 60_000); queries.len()], + &data(), + PlanningModels::builtin(), + asap_plan_selection::Stage1ToStage3Search::exhaustive(), + ) + .unwrap() +} + +/// The logical candidate `id`'s roots. +fn logical_roots(run: &Stage1ToStage3Run, id: &str) -> Vec> { + run.stage1_logical_asap_optimization + .iter() + .find(|c| c.id == id) + .unwrap() + .roots + .as_ref() + .unwrap() + .iter() + .map(|root| match root { + QueryRoot::Operator(node) => Rc::clone(node), + QueryRoot::Scalar(_) => panic!("operator root"), + }) + .collect() +} + +/// The summary builds and rollups (regrouping merges) `roots` read, each +/// once. +fn builds_and_rollups(roots: &[Rc]) -> (usize, usize) { + let mut seen = std::collections::HashSet::new(); + let (mut builds, mut rollups) = (0, 0); + for node in roots.iter().flat_map(OperatorNode::reachable) { + if !seen.insert(Rc::as_ptr(&node)) { + continue; + } + match node.asap() { + Some(ASAPOp::SummaryAgg { .. } | ASAPOp::PaneChain { .. }) => builds += 1, + Some(ASAPOp::SummaryMerge { + group_by: Some(_), .. + }) => rollups += 1, + _ => {} + } + } + (builds, rollups) +} + +/// The cheapest valid candidate's cost per second among the Stage 2 +/// candidates whose logical candidate does (`rolled_up`) or does not roll +/// up a sketch. +fn cheapest(run: &Stage1ToStage3Run, rolled_up: bool) -> Option { + let selection = run.stage3_plan_selection.as_ref().unwrap(); + run.stage2_physical_asap_optimization + .iter() + .filter(|c| { + let (_, rollups) = builds_and_rollups(&logical_roots(run, &c.from_logical)); + (rollups > 0) == rolled_up + }) + .filter_map(|c| selection.costs.get(&c.id)?.cost_per_second) + .min_by(f64::total_cmp) +} + +/// Distinct users by (region, service) and by service, both at relative +/// ε = 0.05 per value over the same window (W-RollupHeavy's targets): Pass 2 offers the coarse KMV rolled up +/// from the finer one, and Stage 3 selects a plan with one build and a +/// rollup, cheaper than any plan that builds both from the raw rows. +#[tokio::test] +async fn stage3_selects_a_distinct_count_rollup_when_it_is_cheaper() { + let accuracy = AccuracyTarget::max_error(0.05) + .failure_probability(0.05) + .accuracy_metric(AccuracyMetric::Relative) + .accuracy_scope(AccuracyScope::PerValue); + let window = "WHERE ts >= now() - INTERVAL '5 minutes'"; + let run = plan_sql( + &[ + &format!( + "SELECT region, service, COUNT(DISTINCT user_id) AS value FROM http {window} \ + GROUP BY region, service" + ), + &format!( + "SELECT service, COUNT(DISTINCT user_id) AS value FROM http {window} \ + GROUP BY service" + ), + ], + accuracy, + ) + .await; + let rolled_up = cheapest(&run, true).expect("a rollup candidate is priced"); + let separate = cheapest(&run, false).expect("separate builds are priced"); + assert!(rolled_up < separate, "{rolled_up} vs {separate}"); + let selected = run.selected().unwrap(); + assert_eq!( + builds_and_rollups(&logical_roots(&run, &selected.from_logical)), + (1, 1), + "{}", + selected.from_logical + ); +} + +/// p99 latency by (region, service), by region and by service at rank +/// ε = 0.05 (template 13's shape): both coarse groupings roll up from one +/// KLL, in one Pass 2 candidate, and Stage 3 selects it. +#[tokio::test] +async fn coarse_quantile_groupings_roll_up_from_one_kll() { + let accuracy = AccuracyTarget::max_error(0.05) + .failure_probability(0.05) + .accuracy_metric(AccuracyMetric::Rank) + .accuracy_scope(AccuracyScope::PerValue); + let window = "WHERE ts >= now() - INTERVAL '5 minutes'"; + let query = |keys: &str| { + format!( + "SELECT {keys}, approx_percentile_cont(latency, 0.99) AS value FROM http {window} \ + GROUP BY {keys}" + ) + }; + let run = plan_sql( + &[ + &query("region, service"), + &query("region"), + &query("service"), + ], + accuracy, + ) + .await; + let selected = run.selected().unwrap(); + assert_eq!( + builds_and_rollups(&logical_roots(&run, &selected.from_logical)), + (1, 2), + "{}", + selected.from_logical + ); +} + +/// #840's `quantile by (host, job)` and `quantile by (job)` over 10,000 +/// series, every 10 s at rank ε = 0.01, δ = 0.001: Pass 2 offers the +/// coarse KLL rolled up from the finer one, and its guarantee meets the +/// target, so Stage 3 prices it. +#[test] +fn promql_quantile_by_job_rolls_up_from_by_host_job() { + let accuracy = AccuracyTarget::max_error(0.01) + .failure_probability(0.001) + .accuracy_metric(AccuracyMetric::Rank); + let roots = [ + "quantile by (host, job) (0.5, data)", + "quantile by (job) (0.5, data)", + ] + .iter() + .enumerate() + .map(|(i, query)| { + let root = lower_promql(query, accuracy.clone()).unwrap(); + ( + i, + QueryRoot::Operator(promql_rows::with_series_identity(&root).unwrap()), + ) + }) + .collect(); + let data: DataWorkload = serde_json::from_value(serde_json::json!({ + "arrival": {"ingesting": {"rows_per_second": 666.67}}, + "sources": {"data": {"series_count": 10_000, "metric_type": "gauge", + "sample_interval": 15_000}}, + })) + .unwrap(); + let entry = entry(&accuracy, 10_000); + let run = run_stage1_to_stage3_with( + roots, + &[entry.clone(), entry], + &data, + PlanningModels::builtin(), + asap_plan_selection::Stage1ToStage3Search::exhaustive(), + ) + .unwrap(); + let rolled: Vec<_> = run + .stage1_logical_asap_optimization + .iter() + .filter(|c| { + c.pass2 + .as_ref() + .is_some_and(|p| !p.rollup_candidates.is_empty()) + }) + .collect(); + assert!(!rolled.is_empty(), "a rollup Pass 2 candidate"); + let selection = run.stage3_plan_selection.as_ref().unwrap(); + assert!( + run.stage2_physical_asap_optimization + .iter() + .filter(|c| rolled.iter().any(|l| l.id == c.from_logical)) + .any(|c| selection.costs.contains_key(&c.id)), + "a rollup plan is valid and priced" + ); +} diff --git a/crates/logical-asap-optimization/src/asap_aware_cse.rs b/crates/logical-asap-optimization/src/asap_aware_cse.rs index b428eb32c..0c06c9284 100644 --- a/crates/logical-asap-optimization/src/asap_aware_cse.rs +++ b/crates/logical-asap-optimization/src/asap_aware_cse.rs @@ -101,19 +101,33 @@ //! 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 candidate* ([`rollup_candidates`]) 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_pass2_candidates`] -//! returns one Pass 2 candidate per non-empty subset of the rollup -//! candidates, each read from -//! its coarsest finer build. +//! nobody asked for. A *rollup candidate* ([`rollup_candidates`]) is a +//! coarse build 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: +//! - An exact accumulator whose family merges across disjoint groups with +//! an exact merged value ([`rollup_kind`]: sums, counts, minima and +//! maxima; counter states merge only in time). The coarse build becomes +//! an exact accumulator over its coarsest finer build's finalized values, +//! grouped by the coarse keys (a count sums the finer counts), so the +//! value stays exact. +//! - A per-group sketch (KMV, HLL, KLL, DDSketch, Count-Min, Count Sketch, +//! UnivMon, at any size) read only by estimates whose merged readouts +//! keep a guarantee (not a heap's top-k or UnivMon's entropy). The coarse build becomes a rollup (a regrouping +//! `SummaryMerge`) of its finest finer build's states by the coarse +//! keys. The groups of one build partition its rows, so the rolled-up +//! state keeps the family's merged guarantee: the direct coarse build's +//! for KMV, HLL, DDSketch and the linear sketches, the same rank bound +//! for KLL. +//! +//! [`rollup_pass2_candidates`] returns one Pass 2 candidate per non-empty +//! subset of the rollup *choices* ([`rollup_choices`]): each exact rollup +//! alone, and every sketch rollup of one finest build together, since they +//! share its size. A chosen sketch choice sizes the finest build for every +//! reader, its own and each coarse one's (the strictest consumer, as the +//! summary-capability rule sizes a shared build), and composes every +//! guarantee above it again. So 15 groupings of one window are one choice, +//! not 2^14 subsets. //! //! # Exact top-k reuse rule //! @@ -131,8 +145,8 @@ use asap_types::ir::cse::share_common_sub_dags; use asap_types::ir::operator::operator_properties::{Reduction, Source}; use asap_types::ir::operator::AggIntent; use asap_types::ir::schema::{ - ExactKind, FamilyCapability, FieldDataType, GroupingStrategy, MergeInputRule, Schema, - SketchKind, PROMQL_SERIES_IDENTITY, + ExactKind, FamilyCapability, FieldDataType, GroupingStrategy, MergeInputRule, MergedError, + Schema, SketchKind, PROMQL_SERIES_IDENTITY, }; use asap_types::ir::SortKey; use asap_types::ir::{ @@ -1573,18 +1587,26 @@ fn rebuild( } /// 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. +/// coarse `SummaryAgg`, the finer node it reads, and the node that replaces +/// the coarse value or state. #[derive(Debug, Clone)] pub struct RollupCandidate { - /// The coarse exact accumulator build (`SummaryAgg`); every reader of - /// it is a `FinalizeExactAccumulator`, which the rollup replaces. + /// The coarse build (`SummaryAgg`). An exact accumulator's every + /// reader is a `FinalizeExactAccumulator`, which the rollup replaces; a + /// sketch's every reader is an estimate its family answers, and the + /// rollup replaces the build itself. pub coarse: Rc, - /// The finer build's `FinalizeExactAccumulator`, already in the - /// candidate, whose values the rollup reads. + /// For an exact accumulator, the finer build's + /// `FinalizeExactAccumulator`, whose values the rollup reads; for a + /// sketch, the finest build of the same algorithm over the same rows + /// (a `SummaryAgg`), whose states the rollup merges. Both are already + /// in the candidate. pub finer: Rc, - /// The finalized exact accumulator over `finer`'s values, grouped by - /// the coarse keys, with the coarse value's column names. + /// For an exact accumulator, the finalized exact accumulator over + /// `finer`'s values, grouped by the coarse keys, with the coarse value's + /// column names; for a sketch, the regrouping `SummaryMerge` of + /// `finer`'s states by the coarse keys, at `finer`'s size (a Pass 2 + /// candidate resizes `finer` for every reader). pub rollup: Rc, } @@ -1631,7 +1653,9 @@ pub fn rollup_kind(family: &FieldDataType) -> Option { /// The rollup candidates 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. +/// the same rows that the candidate finalizes; then each coarse sketch build +/// read only by estimates, with the finest build of its algorithm over the +/// same rows whose keys include its own ([`sketch_rollup_candidates`]). pub fn rollup_candidates(roots: &[QueryRoot]) -> Vec { let mut nodes = Vec::new(); let mut seen = std::collections::HashSet::new(); @@ -1687,9 +1711,144 @@ pub fn rollup_candidates(roots: &[QueryRoot]) -> Vec { } candidates.extend(best.map(|(_, candidate)| candidate)); } + candidates.extend(sketch_rollup_candidates(&nodes, &readers)); candidates } +/// The sketch rollup candidates among `nodes` (children first), whose +/// readers are `readers`: each per-group sketch build grouped by listed keys +/// and read only by estimates its family answers, answered from the finest +/// build ([`finer_keys`] with the most keys, and no build finer than it) of +/// the same algorithm (at any size), update and filter over the same rows, +/// whose family's states merge into a state with a guarantee +/// ([`regroupable`]). Every coarse build of one finest build is one +/// choice in [`rollup_pass2_candidates`], so the choices stay few however +/// many groupings a workload asks for. +fn sketch_rollup_candidates( + nodes: &[Rc], + readers: &HashMap<*const OperatorNode, Vec<&Rc>>, +) -> Vec { + let builds: Vec<&Rc> = nodes.iter().filter(|n| regroupable(n)).collect(); + // The finer build with the most keys; among several (groupings that + // are not nested), the one whose sorted key names come first, so the + // choice does not depend on query order. + let key_names = |build: &OperatorNode| -> Vec { + let mut names: Vec = build + .schema + .fields + .iter() + .filter(|field| field.is_plain()) + .map(column_name) + .collect(); + names.sort(); + names + }; + let finest = |coarse: &OperatorNode| { + builds + .iter() + .filter_map(|&fine| Some((finer_keys(coarse, fine)?, fine))) + .min_by(|(keys, fine), (other_keys, other)| { + other_keys + .cmp(keys) + .then_with(|| key_names(fine).cmp(&key_names(other))) + }) + .map(|(_, fine)| fine) + }; + let mut candidates = Vec::new(); + for &coarse in &builds { + let Some(ASAPOp::SummaryAgg { family, .. }) = coarse.asap() else { + continue; + }; + // Every reader is an estimate whose merged readout keeps a + // guarantee: not a heap's top-k or a UnivMon's entropy. + let capability = FamilyCapability::of(family); + let read_with_guarantee = readers.get(&Rc::as_ptr(coarse)).is_some_and(|readers| { + readers.iter().all(|reader| match reader.asap() { + Some(ASAPOp::SummaryEstimate { query, .. }) => { + capability.answers(query) + && matches!( + capability.merged_error(query), + Some(MergedError::Exact | MergedError::AsIfBuiltOverUnion) + ) + } + _ => false, + }) + }); + if !read_with_guarantee { + continue; + } + // A build finer than the one with the most keys would have more. + let Some(fine) = finest(coarse) else { + continue; + }; + if let Some(rollup) = regrouped(coarse, fine) { + candidates.push(RollupCandidate { + coarse: Rc::clone(coarse), + finer: Rc::clone(fine), + rollup, + }); + } + } + candidates +} + +/// Whether `node` is a per-group sketch build grouped by listed keys whose +/// states a rollup may merge across groups +/// ([`FamilyCapability::merges_groups`]). +fn regroupable(node: &OperatorNode) -> bool { + matches!( + node.asap(), + Some(ASAPOp::SummaryAgg { + family: family @ FieldDataType::Sketch(..), + reduction: Reduction::Reduce(by), + .. + }) if !by.is_without() && FamilyCapability::merges_groups(family) + ) +} + +/// The regrouping `SummaryMerge` of `fine`'s states by `coarse`'s keys, in +/// `coarse`'s order and with its column names; `None` when a key is not in +/// `fine`'s state. +fn regrouped(coarse: &OperatorNode, fine: &Rc) -> Option> { + let Some(ASAPOp::SummaryAgg { + reduction: Reduction::Reduce(by), + .. + }) = coarse.asap() + else { + return None; + }; + let keys = (0..by.keys().len()) + .map(|key| { + let name = column_name(&coarse.schema.fields[key]); + fine.schema + .fields + .iter() + .position(|field| field.is_plain() && column_name(field) == name) + }) + .collect::>>()?; + regroup_merge(fine, keys, coarse).ok() +} + +/// `SummaryMerge { [state], group_by: keys }`, its columns named as +/// `named`'s (the coarse build it stands for). +fn regroup_merge( + state: &Rc, + keys: Vec, + named: &OperatorNode, +) -> Result, asap_types::ir::SchemaDerivationError> { + let merge = OperatorNode::new(Operator::ASAP(ASAPOp::SummaryMerge { + children: vec![Rc::clone(state)], + group_by: Some(keys), + newest_panes: None, + }))?; + let mut schema = merge.schema.clone(); + for (field, coarse) in schema.fields.iter_mut().zip(&named.schema.fields) { + field.name = coarse.name.clone(); + field.table = coarse.table.clone(); + } + Ok(Rc::new(OperatorNode::with_schema(merge.operator, schema))) +} + /// Whether `node` is an exact accumulator build grouped by listed keys whose /// family merges across groups ([`rollup_kind`]). fn grouped_exact(node: &OperatorNode) -> bool { @@ -1705,8 +1864,9 @@ fn grouped_exact(node: &OperatorNode) -> bool { } /// 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. +/// family (a sketch's algorithm at any size), 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 { @@ -1741,7 +1901,7 @@ fn finer_keys(coarse: &OperatorNode, fine: &OperatorNode) -> Option { // 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 + (same_family(family, fine_family) && input == fine_input && same_filter && fine_keys.len() > coarse_keys.len() @@ -1887,25 +2047,61 @@ fn rolled_up( Some(finalized) } +/// The independent choices among rollup candidates `candidates` +/// ([`rollup_candidates`]), in order: each exact accumulator's alone, and +/// every sketch candidate of one finest build together, since they share +/// that build's size. +pub fn rollup_choices(candidates: &[RollupCandidate]) -> Vec> { + let mut choices: Vec> = Vec::new(); + for candidate in candidates { + let sketch = is_sketch_build(&candidate.finer); + match choices.iter_mut().find(|choice| { + sketch + && is_sketch_build(&choice[0].finer) + && Rc::ptr_eq(&choice[0].finer, &candidate.finer) + }) { + Some(choice) => choice.push(candidate.clone()), + None => choices.push(vec![candidate.clone()]), + } + } + choices +} + +fn is_sketch_build(node: &OperatorNode) -> bool { + matches!( + node.asap(), + Some(ASAPOp::SummaryAgg { + family: FieldDataType::Sketch(..), + .. + }) + ) +} + /// Every Pass 2 candidate of the rollup rule for the candidate with query /// roots `roots` and rollup candidates `candidates` ([`rollup_candidates`]): -/// one per non-empty subset of them, in binary-counter order (the first -/// the lowest bit). An error only for more than [`MAX_REUSE_CANDIDATES`] -/// of them; a Pass 2 candidate -/// whose composition fails carries its own error. +/// one per non-empty subset of their choices ([`rollup_choices`]), in +/// binary-counter order (the first the lowest bit). A chosen sketch choice +/// resizes its finest build for every reader, its own and each coarse +/// build's ([`sized_for_all`]: the strictest consumer's size), and each +/// coarse build becomes the regrouping merge of the resized build's +/// states. An error only for more than [`MAX_REUSE_CANDIDATES`] choices; a +/// Pass 2 candidate whose composition fails carries its own error. pub fn rollup_pass2_candidates( roots: &[QueryRoot], candidates: &[RollupCandidate], ) -> Result, PerComputationCandidateError> { - Ok(subsets(candidates, "rollup")? + Ok(subsets(&rollup_choices(candidates), "rollup")? .into_iter() - .map(|chosen| { + .map(|choices| { + let chosen: Vec = choices.iter().flatten().cloned().collect(); let values: HashMap<*const OperatorNode, &RollupCandidate> = chosen .iter() + .filter(|candidate| !is_sketch_build(&candidate.finer)) .map(|candidate| (Rc::as_ptr(&candidate.coarse), candidate)) .collect(); - // Every finalized value of a chosen coarse build is replaced. - let replaced = roots + // Every finalized value of a chosen coarse exact build is + // replaced. + let mut replaced: HashMap<*const OperatorNode, Rc> = roots .iter() .flat_map(root_operators) .flat_map(OperatorNode::reachable) @@ -1917,14 +2113,56 @@ pub fn rollup_pass2_candidates( _ => None, }) .collect(); + let roots = sketch_rollups(&choices).and_then(|sketches| { + replaced.extend(sketches); + replace_all(roots, &replaced) + }); RollupPass2Candidate { - roots: replace_all(roots, &replaced), + roots, rollup_candidates: chosen, } }) .collect()) } +/// The replacements of the sketch rollup choices among `choices` +/// ([`rollup_choices`]): each finest build by the build sized for it and +/// every coarse build it serves, and each coarse build by the regrouping +/// merge of that build's states. An error when a finest build and its +/// coarse builds have no shared size, or a node does not derive. +fn sketch_rollups( + choices: &[Vec], +) -> Result)>, PerComputationCandidateError> { + let unsupported = || { + PerComputationCandidateError::Unsupported( + "a sketch rollup whose finest and coarse builds have no shared size", + ) + }; + let mut replaced = Vec::new(); + for choice in choices { + let fine = &choice[0].finer; + if !is_sketch_build(fine) { + continue; + } + let mut members = vec![fine]; + members.extend(choice.iter().map(|candidate| &candidate.coarse)); + let sized = sized_for_all(&members)?.ok_or_else(unsupported)?; + replaced.push((Rc::as_ptr(fine), Rc::clone(&sized))); + for candidate in choice { + let Some(ASAPOp::SummaryMerge { + group_by: Some(keys), + .. + }) = candidate.rollup.asap() + else { + unreachable!("a sketch rollup candidate's rollup is a regrouping merge"); + }; + let merge = regroup_merge(&sized, keys.clone(), &candidate.coarse)?; + replaced.push((Rc::as_ptr(&candidate.coarse), merge)); + } + } + Ok(replaced) +} + /// Every non-empty subset of `candidates`, in binary-counter order (the /// first candidate the lowest bit); an error past [`MAX_REUSE_CANDIDATES`]. fn subsets( @@ -1935,7 +2173,7 @@ fn subsets( if n > MAX_REUSE_CANDIDATES { return Err(PerComputationCandidateError::TooManyCandidates { counted: match rule { - "rollup" => "rollup candidates", + "rollup" => "rollup choices", _ => "top-k reuse candidates", }, found: n, @@ -2824,7 +3062,7 @@ mod tests { assert_eq!( error.to_string(), format!( - "found {} rollup candidates, more than {MAX_REUSE_CANDIDATES}", + "found {} rollup choices, more than {MAX_REUSE_CANDIDATES}", MAX_REUSE_CANDIDATES + 1 ) ); @@ -3523,15 +3761,164 @@ mod tests { ] { assert_eq!(rollup_kind(&family), None, "{family:?}"); } - // A grouped KLL beside a finer one is no site. + } + + /// #840's `quantile by (host, job)` and `quantile by (job)`, both KLL, + /// the coarse one at a tighter ε: the coarse query reads the regrouping + /// merge of the finer KLL's states by `job`, so one KLL is built, sized + /// for its strictest reader (the larger `k`), and the coarse query keeps + /// a guarantee within its own target. + #[test] + fn a_coarse_sketch_rolls_up_from_the_finer_build() { + let roots = chose( + &[ + ("quantile by (host, job) (0.5, data)", 0.01), + ("quantile by (job) (0.5, data)", 0.001), + ], + SketchAlgorithm::Kll, + ); + let ks: Vec = kll_builds(&roots).into_iter().map(|(_, k)| k).collect(); + let rule_candidates = rollup_candidates(&roots); + let [rule_candidate] = rule_candidates.as_slice() else { + panic!("one candidate: {rule_candidates:?}") + }; + assert!(Rc::ptr_eq(&rule_candidate.coarse, &builds(&roots[1])[0])); + assert!(Rc::ptr_eq(&rule_candidate.finer, &builds(&roots[0])[0])); + let pass2_candidates = rollup_pass2_candidates(&roots, &rule_candidates).unwrap(); + let [pass2_candidate] = pass2_candidates.as_slice() else { + panic!("one Pass 2 candidate") + }; + let rolled = pass2_candidate.roots.as_ref().unwrap(); + // One KLL, at the larger k, read by both queries. + assert_eq!( + kll_builds(rolled) + .into_iter() + .map(|(_, k)| k) + .collect::>(), + [*ks.iter().max().unwrap()] + ); + let merge = OperatorNode::reachable(operator(&rolled[1])) + .into_iter() + .find(|n| matches!(n.asap(), Some(ASAPOp::SummaryMerge { .. }))) + .expect("the coarse query reads a rollup"); + let Some(ASAPOp::SummaryMerge { + children, group_by, .. + }) = merge.asap() + else { + unreachable!() + }; + assert_eq!(group_by, &Some(vec![1])); + assert!(Rc::ptr_eq(&children[0], &builds(&rolled[0])[0])); + 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); + } + // Each query's guarantee is the shared KLL's, within its target. + assert!( + root_bound(&rolled[1]) <= 0.001, + "{}", + root_bound(&rolled[1]) + ); + assert!(root_bound(&rolled[0]) <= 0.001); + } + + /// Every coarser grouping of one finest build is one choice: two coarse + /// builds of a three-grouping chain give one Pass 2 candidate, in which + /// both read the finest build's states. A finer build that is itself + /// coarse is not a source. + #[test] + fn coarse_groupings_of_one_finest_build_roll_up_together() { let roots = chose( &[ ("quantile by (job) (0.9, m)", 0.01), ("quantile by (job, instance) (0.9, m)", 0.01), + ("quantile by (job, instance, zone) (0.9, m)", 0.01), + ], + SketchAlgorithm::Kll, + ); + let rule_candidates = rollup_candidates(&roots); + assert_eq!(rule_candidates.len(), 2); + let finest = &builds(&roots[2])[0]; + assert!(rule_candidates + .iter() + .all(|candidate| Rc::ptr_eq(&candidate.finer, finest))); + assert_eq!(rollup_choices(&rule_candidates).len(), 1); + let pass2_candidates = rollup_pass2_candidates(&roots, &rule_candidates).unwrap(); + let [pass2_candidate] = pass2_candidates.as_slice() else { + panic!("one Pass 2 candidate") + }; + let rolled = pass2_candidate.roots.as_ref().unwrap(); + assert_eq!(kll_builds(rolled).len(), 1); + } + + /// A quantile over every series rolls up from the KLLs by job: a + /// rollup to no keys. + #[test] + fn a_global_quantile_rolls_up_to_no_keys() { + let roots = chose( + &[ + ("quantile(0.9, m)", 0.01), + ("quantile by (job) (0.9, m)", 0.01), + ], + SketchAlgorithm::Kll, + ); + let rule_candidates = rollup_candidates(&roots); + let [rule_candidate] = rule_candidates.as_slice() else { + panic!("one candidate: {rule_candidates:?}") + }; + assert!(matches!( + rule_candidate.rollup.asap(), + Some(ASAPOp::SummaryMerge { group_by: Some(keys), .. }) if keys.is_empty() + )); + let pass2 = rollup_pass2_candidates(&roots, &rule_candidates).unwrap(); + let rolled = pass2[0].roots.as_ref().unwrap(); + assert_eq!(kll_builds(rolled).len(), 1); + operator(&rolled[0]).validate_structure().unwrap(); + } + + /// Groupings that are not nested: `by (job)` rolls up from `by (job, + /// instance)` or `by (job, zone)`; the source is the one whose sorted + /// key names come first, whatever the query order. + #[test] + fn the_rollup_source_does_not_depend_on_query_order() { + for queries in [ + [ + "quantile by (job) (0.9, m)", + "quantile by (job, zone) (0.9, m)", + "quantile by (job, instance) (0.9, m)", + ], + [ + "quantile by (job) (0.9, m)", + "quantile by (job, instance) (0.9, m)", + "quantile by (job, zone) (0.9, m)", + ], + ] { + let roots = chose(&queries.map(|q| (q, 0.01)), SketchAlgorithm::Kll); + let rule_candidates = rollup_candidates(&roots); + let [rule_candidate] = rule_candidates.as_slice() else { + panic!("one candidate: {rule_candidates:?}") + }; + let instance = queries.iter().position(|q| q.contains("instance")).unwrap(); + assert!(Rc::ptr_eq( + &rule_candidate.finer, + &builds(&roots[instance])[0] + )); + } + } + + /// A rollup needs the same rows: a coarse KLL over another metric is + /// none. + #[test] + fn sketch_rollup_needs_the_same_rows() { + let other_metric = chose( + &[ + ("quantile by (job) (0.9, n)", 0.01), + ("quantile by (job, instance) (0.9, m)", 0.01), ], SketchAlgorithm::Kll, ); - assert!(rollup_candidates(&roots).is_empty()); + assert!(rollup_candidates(&other_metric).is_empty()); } /// Two exact top-k over the same per-job sums: both read one ranking, diff --git a/crates/logical-asap-optimization/src/per_computation_candidates.rs b/crates/logical-asap-optimization/src/per_computation_candidates.rs index 5b56e2a62..6320f6d71 100644 --- a/crates/logical-asap-optimization/src/per_computation_candidates.rs +++ b/crates/logical-asap-optimization/src/per_computation_candidates.rs @@ -1546,6 +1546,9 @@ fn with_guarantees( /// /// [`AccuracyScope::PerValue`]: asap_types::ir::properties::AccuracyScope::PerValue fn composed_guarantee(node: &OperatorNode) -> Option { + if let Some(guarantee) = rolled_up_guarantee(node) { + return guarantee; + } if let Some(guarantee) = merged_panes_guarantee(node) { return guarantee; } @@ -1691,6 +1694,61 @@ fn composed_guarantee(node: &OperatorNode) -> Option { } } +/// The guarantee of an estimate or finalized value read from a rollup (a +/// regrouping `SummaryMerge` of one input): the same read of the input +/// states, when the family's merged error for that read is exact or the +/// same as one build's over the union (the family-capability table). The +/// groups of one build partition its rows, so a rolled-up state is a merge +/// of disjoint parts, and a family whose bound holds for one build over +/// the union keeps it: KMV, HLL and DDSketch merge into the state a direct +/// coarse build would have, KLL into a state with the same rank bound, and +/// linear sketches into the direct build's counters, whose error is +/// relative to the coarse group's total. Each guarantee holds per output +/// point, so the number of groups does not change it. None when the merged +/// readout is approximate (heap top-k, entropy) or the family has no merge. +/// `None` for any node that does not read a rollup. +fn rolled_up_guarantee(node: &OperatorNode) -> Option> { + let state = match node.asap()? { + ASAPOp::SummaryEstimate { summary_input, .. } => summary_input, + ASAPOp::FinalizeExactAccumulator { child } => child, + _ => return None, + }; + let Some(ASAPOp::SummaryMerge { + children, + group_by: Some(_), + .. + }) = state.asap() + else { + return None; + }; + let [input] = children.as_slice() else { + return Some(None); + }; + let family = state.asap()?.produced_state()?; + let merged = match node.asap()? { + ASAPOp::SummaryEstimate { query, .. } => FamilyCapability::of(family).merged_error(query), + // An exact accumulator merged across groups is exact. + _ => FamilyCapability::merges_groups(family).then_some(MergedError::Exact), + }; + if !matches!( + merged, + Some(MergedError::Exact | MergedError::AsIfBuiltOverUnion) + ) { + return Some(None); + } + let reader = OperatorNode::with_schema( + node.operator.map_children(|child| { + if Rc::ptr_eq(child, state) { + Rc::clone(input) + } else { + Rc::clone(child) + } + }), + node.schema.clone(), + ); + Some(composed_guarantee(&reader)) +} + /// The guarantee of an estimate or finalized value read from merged /// tumbling panes ([`merged_chain`]): the guarantee of the same read of one /// build over the merged panes' window ([`merged_window_build`]: the whole diff --git a/crates/logical-asap-optimization/src/workload_candidates.rs b/crates/logical-asap-optimization/src/workload_candidates.rs index aec78e607..354fb8af4 100644 --- a/crates/logical-asap-optimization/src/workload_candidates.rs +++ b/crates/logical-asap-optimization/src/workload_candidates.rs @@ -12,7 +12,7 @@ use asap_types::workload::QueryWorkloadEntry; use crate::asap_aware_cse::{ capability_sharing_candidates, identical_expression_pass2_candidates, pane_candidates, - rollup_candidates, rollup_pass2_candidates, sharing_candidates, + rollup_candidates, rollup_choices, rollup_pass2_candidates, sharing_candidates, summary_capability_pass2_candidates, topk_reuse_candidates, topk_reuse_pass2_candidates, window_composition_pass2_candidates, with_union_scan_labels, PaneCandidate, RollupCandidate, SharedPaneChain, SharingCandidate, TopKReuseCandidate, MAX_REUSE_CANDIDATES, @@ -423,17 +423,18 @@ impl Pass2List<'_> { /// computation rolls up its merged state); each followed by its /// identical-expression Pass 2 candidates. They are listed eagerly, /// like every Pass 2 rule's: more than [`MAX_REUSE_CANDIDATES`] rollup - /// and top-k reuse candidates together list one error. + /// choices ([`rollup_choices`]) and top-k reuse candidates together list + /// one error. fn reuse(&mut self, rules: &Rules, roots: &[QueryRoot], evaluation_grid_ms: &[Option]) { let candidates = rollup_candidates(roots); // Every rollup subset is combined with every top-k subset: cap the // product, not each factor. - let found = candidates.len() + topk_reuse_candidates(roots).len(); + let found = rollup_choices(&candidates).len() + topk_reuse_candidates(roots).len(); if found > MAX_REUSE_CANDIDATES { self.out.push(Pass2::failed( rules, PerComputationCandidateError::TooManyCandidates { - counted: "rollup and top-k reuse candidates together", + counted: "rollup choices and top-k reuse candidates together", found, max: MAX_REUSE_CANDIDATES, }, @@ -683,7 +684,7 @@ mod tests { }; assert_eq!( error.to_string(), - format!("found 21 rollup and top-k reuse candidates together, more than {MAX_REUSE_CANDIDATES}") + format!("found 21 rollup choices and top-k reuse candidates together, more than {MAX_REUSE_CANDIDATES}") ); } } diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs index 1551b5193..a29d00d8d 100644 --- a/crates/plan-selection/src/lib.rs +++ b/crates/plan-selection/src/lib.rs @@ -2095,6 +2095,11 @@ fn at_most_one_row(node: &OperatorNode) -> bool { .any(|m| matches!(m, AggIntent::TopK { .. } | AggIntent::CountValues { .. })) } Operator::ASAP(ASAPOp::SummaryAgg { reduction, .. }) => global(reduction), + // A rollup to no keys gives one state. + Operator::ASAP(ASAPOp::SummaryMerge { + group_by: Some(keys), + .. + }) => keys.is_empty(), // A merge gives one state per group, whatever the panes it merges. Operator::ASAP(ASAPOp::SummaryMerge { .. }) => matches!( node.summary_build(), @@ -3601,10 +3606,16 @@ fn price_at_scale( ), ) } - Payload::ASAP(ASAPOp::SummaryMerge { newest_panes, .. }) => { + Payload::ASAP(ASAPOp::SummaryMerge { + newest_panes, + group_by, + .. + }) => { // One operation per input state read; one state per group // out, each pane of a chain being a state of its group. A - // merge of a chain's newest `n` panes reads `n` per group. + // merge of a chain's newest `n` panes reads `n` per group. A + // rollup's groups are those of its keys, at most its input + // groups. let mut group_count = 0; for producer in producers { let panes = match &nodes[producer].payload { @@ -3616,6 +3627,9 @@ fn price_at_scale( group_count = group_count.max(output[producer].rows / panes.max(1)); } let states = merged_states(*newest_panes, producers, &output, nodes); + if let Some(keys) = group_by { + group_count = group_count.min(groups(&Reduction::by(keys.clone()))); + } let out = EdgeStatistics { rows: group_count, bytes: input.bytes.div_ceil(input.rows.max(1)) * group_count, diff --git a/docs/design_docs/proposals/asap-aware-mapping/analytical-resource-cost.md b/docs/design_docs/proposals/asap-aware-mapping/analytical-resource-cost.md index 8cb186c37..31155197d 100644 --- a/docs/design_docs/proposals/asap-aware-mapping/analytical-resource-cost.md +++ b/docs/design_docs/proposals/asap-aware-mapping/analytical-resource-cost.md @@ -785,8 +785,11 @@ from logical group count alone. Summary merge, subtract, delete, and readout are separate physical operators. Their CPU and memory use the concrete summary state size and number of input -states. A plan using one of these operations is unavailable until the -corresponding formula and required evidence are present. +states. A merge costs one operation per input state, and outputs one state per +group: the input's groups, or for a rollup (`group_by`) the groups of its +keys, estimated like any `by (keys)` and at most the input's groups. A plan +using one of these operations is unavailable until the corresponding formula +and required evidence are present. Summary construction uses physical-input realization rules before it emits a `SummaryAgg`. The default rule consumes the logical aggregate's immediate diff --git a/docs/design_docs/proposals/planner-layering.md b/docs/design_docs/proposals/planner-layering.md index 9abfa7840..27a438436 100644 --- a/docs/design_docs/proposals/planner-layering.md +++ b/docs/design_docs/proposals/planner-layering.md @@ -543,7 +543,7 @@ their overlapping windows. | Identical-expression rule | The input and computation semantics are identical. | One common computation node serving multiple consumers. | | Summary-capability rule | The computations have the same summary input data and the same window, and one summary supports all requested computations and their accuracy requirements. | One summary build node feeding several estimation nodes, e.g. UnivMon → distinct count, entropy, L2 norm. | | Window-composition rule | The computations have the same summary input data, and one window summary can answer the requested windows within their accuracy requirements. | One window summary feeding per-query merge (where needed) and estimation nodes, e.g. a sliding-window or tumbling-window KLL, or an Exponential Histogram with a KLL per EH bucket. | -| Rollup rule | A coarse exact accumulator (`sum`, `min`, `max`) reads the same rows as a finer exact accumulator the workload already computes, grouped by a superset of its keys. | The finer accumulator's finalized values feeding an exact accumulator grouped by the coarse keys, e.g. `sum by (job)` from `sum by (job, instance)`. | +| Rollup rule | A coarse exact accumulator (`sum`, `min`, `max`) or per-group sketch reads the same rows as a finer build of the same family (a sketch at any size) the workload already computes, grouped by a superset of its keys. | Exact: the finer accumulator's finalized values feeding an exact accumulator grouped by the coarse keys, e.g. `sum by (job)` from `sum by (job, instance)`. Sketch: a rollup (a `SummaryMerge` with `group_by`) of the finest build's states by the coarse keys, e.g. a KLL per `job` merged from the KLLs per `(host, job)`. Every coarse grouping of one finest build rolls up together, and that build is sized for its strictest reader. Every coarse reader must be an estimate whose merged readout keeps a guarantee (not a heap's top-k or UnivMon's entropy). The rolled-up state keeps the family's merged guarantee: KMV, HLL, DDSketch, Count-Min and Count Sketch give the direct coarse build's state, UnivMon its total, distinct count and L2, and KLL keeps its rank bound. Stage 3 prices one merge per input state and the coarse groups' states. | | Exact top-k reuse rule | Exact top-k computations read the same input and partition with different `k`. | One ranking (`Limit(k_max)` over `Sort`) that each top-k reads its first `k` rows from. | The rules compare computations by their **summary input data**: what a summary for From b994c0d7b5fcc2b7d2ceec2d9b051a32ee2ec472 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 22 Oct 2026 12:16:25 +0000 Subject: [PATCH 2/2] test(devtools): the #840 workload selects Q4's KLL rolled up from Q3's --- crates/devtools/tests/stage_pipeline.rs | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/crates/devtools/tests/stage_pipeline.rs b/crates/devtools/tests/stage_pipeline.rs index 31515a0a3..cde2d4227 100644 --- a/crates/devtools/tests/stage_pipeline.rs +++ b/crates/devtools/tests/stage_pipeline.rs @@ -1673,8 +1673,21 @@ fn scans_projecting_different_labels_are_shared() { let selected_label = physical.iter().find(|c| c["id"] == selected).unwrap()["display_name"] .as_str() .unwrap(); + // Q4's KLL by job rolls up from Q3's KLL by (host, job), built once per + // evaluation for both. assert_eq!( selected_label, - "Q1 Kll · Q2 Kll · Q3 Kll · Q4 Kll · shared Scan (Q1, Q2) · query time, shared: Scan" + "Q1 Kll · Q2 Kll · Q3 Kll · Q4 Kll · Kll for Q4 rolled up from Q3 · shared Scan (Q1, Q2) \ + · query time, shared: Scan, Kll" + ); + let rolled_up = selection["costs"][selected]["cost_per_second"] + .as_f64() + .unwrap(); + let without_rollup = cost_of( + "Q1 Kll · Q2 Kll · Q3 Kll · Q4 Kll · shared Scan (Q1, Q2) · query time, shared: Scan", + ); + assert!( + rolled_up < without_rollup, + "{rolled_up} vs {without_rollup}" ); }