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}" ); } 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