From c63ec482cf36e60a3817f00ffaf9bb3dcb4870bb Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 22 Oct 2026 16:19:12 +0000 Subject: [PATCH 1/2] feat(pass2): share SQL table scans; a coarser grouping reads a finer grouping's pane chain, newest panes rolled up Two queries over one SQL table shared nothing: structural CSE shares a non-ASAP node only when its rows have a unique key, and a table scan's rows have none. Equal scans are now shared: each reader reads every row the scan outputs. The window-composition rule now also groups a build with a finer grouping's build over a window at least its own: one chain of the finer grouping's panes, sized for every reader, which the coarser build reads as its newest panes rolled up into its own groups. A global SQL percentile over 5 minutes reads the newest 30 of 120 per-city panes kept for a per-city percentile over 20 minutes. A rollup to no keys of a SQL relation's states now stands for the global SQL aggregate: the executor outputs one empty state when it has no input state, so the answer is one NULL row, as the global build gives. The coverage check no longer rejects it, and the rollup rule offers it. Part of #878 (problem 2). --- crates/devtools/src/candidate_labels.rs | 12 +- crates/executor/src/operators/mod.rs | 5 + crates/executor/src/operators/summary/mod.rs | 61 ++++- crates/executor/src/operators/unchecked.rs | 13 +- crates/executor/src/physical_planner/mod.rs | 30 +++ .../integration-tests/tests/sketch_rollup.rs | 11 +- .../tests/sql_two_percentiles.rs | 192 ++++++++++++++- .../src/asap_aware_cse.rs | 219 +++++++++++++++--- crates/types/src/ir/cse.rs | 36 ++- .../src/ir/properties/summary_coverage.rs | 10 +- crates/types/tests/summary_coverage.rs | 22 +- .../proposals/asap-primitive-schema.md | 9 +- .../design_docs/proposals/planner-layering.md | 23 +- 13 files changed, 569 insertions(+), 74 deletions(-) diff --git a/crates/devtools/src/candidate_labels.rs b/crates/devtools/src/candidate_labels.rs index 0bb68558e..1cbaabb42 100644 --- a/crates/devtools/src/candidate_labels.rs +++ b/crates/devtools/src/candidate_labels.rs @@ -216,8 +216,18 @@ pub fn pass2_label( .collect(); let shared = pass2.shared_pane_chains.iter().map(|shared| { let build = &candidates[shared.members[0]].build; + // The chain is one member's build in panes (a coarser member reads + // it rolled up). let chain = |node: &OperatorNode| { - is_pane_chain_of(node, build, shared.pane_width_ms, shared.pane_count, true) + shared.members.iter().any(|&m| { + is_pane_chain_of( + node, + &candidates[m].build, + shared.pane_width_ms, + shared.pane_count, + true, + ) + }) }; let mut text = format!( "one {} in {} × {} panes", diff --git a/crates/executor/src/operators/mod.rs b/crates/executor/src/operators/mod.rs index c618678e6..a81fa6c32 100644 --- a/crates/executor/src/operators/mod.rs +++ b/crates/executor/src/operators/mod.rs @@ -183,6 +183,11 @@ enum Kind { SummaryMerge { state: usize, groups: Vec, + /// Whether a merge of no input state outputs one empty state, as a + /// global SQL aggregate has one row over no input rows. Only a merge + /// with no groups; `false` (no output row) by default. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + empty_global_state: bool, }, Evaluation { state: usize, diff --git a/crates/executor/src/operators/summary/mod.rs b/crates/executor/src/operators/summary/mod.rs index 5e83da33a..1dc4a9268 100644 --- a/crates/executor/src/operators/summary/mod.rs +++ b/crates/executor/src/operators/summary/mod.rs @@ -278,11 +278,34 @@ impl Operator { .collect::>(); fields.push(input.fields[state].clone()); Ok(Self { - kind: Kind::SummaryMerge { state, groups }, + kind: Kind::SummaryMerge { + state, + groups, + empty_global_state: false, + }, inputs: vec![input], output: schema(fields), }) } + /// A merge into no groups that outputs one empty state when it has no + /// input state: the global SQL aggregate a rollup of a relation's + /// grouped states to no keys stands for, whose one row is NULL over no + /// input rows. + pub fn with_empty_global_state(mut self) -> Result { + match &mut self.kind { + Kind::SummaryMerge { + groups, + empty_global_state, + .. + } if groups.is_empty() => *empty_global_state = true, + _ => { + return Err(invalid( + "an empty global state needs a merge into no groups", + )) + } + } + Ok(self) + } pub fn evaluation( input: SchemaRef, state: usize, @@ -485,10 +508,20 @@ pub(super) fn execute_merge<'a>( let input_schema = operator.inputs[0].clone(); let input = inputs.pop().ok_or_else(|| invalid("input missing"))?; Ok(futures::stream::once(async move { - let Kind::SummaryMerge { state, groups } = &operator.kind else { + let Kind::SummaryMerge { + state, + groups, + empty_global_state, + } = &operator.kind + else { unreachable!() }; - merge_summary(input, *state, groups, &input_schema, &output, &context).await + let merge = Merge { + state_column: *state, + groups, + empty_global_state: *empty_global_state, + }; + merge_summary(input, merge, &input_schema, &output, &context).await }) .boxed_local()) } @@ -831,16 +864,27 @@ fn state_indices(batch: &Batch, state: usize) -> Result<(&StateColumn, &UInt32Ar } /// Merge the states of each group across all input rows. +/// A `SummaryMerge`'s parameters. +struct Merge<'a> { + state_column: usize, + groups: &'a [usize], + empty_global_state: bool, +} + async fn merge_summary( mut input: Input<'_, Batch>, - state_column: usize, - groups: &[usize], + merge: Merge<'_>, input_schema: &SchemaRef, output: &SchemaRef, context: &RunContext, ) -> Result { use crate::summary_kernels::exact::ScalarState; use crate::values::StateStorage; + let Merge { + state_column, + groups, + empty_global_state, + } = merge; let family = &input_schema.fields[state_column].dtype; let mut batches = Vec::new(); let mut reservations = Vec::new(); @@ -861,6 +905,13 @@ async fn merge_summary( .map(|(row, group)| (index, row, group)), ); } + // A global SQL aggregate's one row over no input rows: one empty state. + if entries.is_empty() && empty_global_state { + table.add_empty_group()?; + let mut updater = StateColumnUpdater::new(family)?; + updater.add_group()?; + return grouped_states(&table, updater.seal(), input_schema, output); + } let columns = batches .iter() .map(|batch| state_indices(batch, state_column)) diff --git a/crates/executor/src/operators/unchecked.rs b/crates/executor/src/operators/unchecked.rs index 568b9d623..75481f497 100644 --- a/crates/executor/src/operators/unchecked.rs +++ b/crates/executor/src/operators/unchecked.rs @@ -190,8 +190,17 @@ impl TryFrom for Operator { Kind::KeyedEvaluation { state, k } => { Operator::keyed_evaluation(input(0)?, state, k, output.clone())? } - Kind::SummaryMerge { state, groups } => { - Operator::summary_merge(input(0)?, state, groups)? + Kind::SummaryMerge { + state, + groups, + empty_global_state, + } => { + let merge = Operator::summary_merge(input(0)?, state, groups)?; + if empty_global_state { + merge.with_empty_global_state()? + } else { + merge + } } Kind::Evaluation { state, query } => Operator::evaluation(input(0)?, state, query)?, } diff --git a/crates/executor/src/physical_planner/mod.rs b/crates/executor/src/physical_planner/mod.rs index d4254f2de..552997967 100644 --- a/crates/executor/src/physical_planner/mod.rs +++ b/crates/executor/src/physical_planner/mod.rs @@ -626,6 +626,21 @@ fn compile_internal( continue; } } + // A rollup of a relation's states to no keys is a global SQL + // aggregate: one row, an empty state, over no input rows. + if let Payload::ASAP(ASAPOp::SummaryMerge { + group_by: Some(keys), + .. + }) = &node.payload + { + if keys.is_empty() && built_from_relation(&restored[&id]) { + let operator = compile_node(node, &schemas) + .and_then(Operator::with_empty_global_state) + .map_err(|error| invalid(format!("node {id}: {error}")))?; + physical_dag.add(id, inputs, operator)?; + continue; + } + } // A SQL table's trailing window (and a tumbling pane's shift // under it) runs as a filter on the table's time column against // the evaluation time. @@ -654,6 +669,21 @@ fn compile_internal( Ok(physical_dag) } +/// Whether every summary state `state` merges was built from a relation's +/// rows (a SQL table's), where a global aggregate has a row over no input +/// row; not from a PromQL vector's, where it has none. +fn built_from_relation(state: &OperatorNode) -> bool { + match state.asap() { + Some(ASAPOp::SummaryMerge { children, .. }) => { + children.iter().all(|child| built_from_relation(child)) + } + Some(ASAPOp::SummaryAgg { child, .. } | ASAPOp::PaneChain { child, .. }) => { + child.result_kind == planner_types::ir::OperatorResultKind::Relation + } + _ => false, + } +} + /// The native operator of `id` when it is a range `TimeRange` or a /// `TimeShift` over a SQL table's rows (a table `Scan` below any stacked /// `TimeShift`s), `None` for any other node. diff --git a/crates/integration-tests/tests/sketch_rollup.rs b/crates/integration-tests/tests/sketch_rollup.rs index 3e6d53129..bc2a20737 100644 --- a/crates/integration-tests/tests/sketch_rollup.rs +++ b/crates/integration-tests/tests/sketch_rollup.rs @@ -340,9 +340,9 @@ async fn quantiles_rolled_up_keep_the_sketch_bound() { } } -/// A regrouping merge keys only by group keys of its input states, does -/// not stand for a global SQL aggregate, and never merges counter states -/// of different groups. +/// A regrouping merge keys only by group keys of its input states (or by +/// none, a global aggregate), and never merges counter states of different +/// groups. #[tokio::test] async fn invalid_rollups_are_rejected() { let accuracy = AccuracyTarget::max_error(0.05) @@ -383,9 +383,8 @@ async fn invalid_rollups_are_rejected() { for keys in [vec![2], vec![1, 1]] { assert!(regroup(keys.clone()).is_err(), "{keys:?}"); } - assert!(regroup(vec![]) - .unwrap_err() - .contains(&CoverageError::GlobalRelationRegroup.to_string())); + // No keys: the global build over the same rows. + assert!(regroup(vec![]).unwrap().coverage().is_some()); // Both keys, swapped, are still group keys of the input. assert!(regroup(vec![1, 0]).unwrap().coverage().is_some()); // Rate states merge consecutive windows of one series, not groups. diff --git a/crates/integration-tests/tests/sql_two_percentiles.rs b/crates/integration-tests/tests/sql_two_percentiles.rs index 7e0d43564..50584dd71 100644 --- a/crates/integration-tests/tests/sql_two_percentiles.rs +++ b/crates/integration-tests/tests/sql_two_percentiles.rs @@ -2,7 +2,10 @@ //! the overall p50 and p90 over the last 5 minutes: every candidate that //! computes one of its percentiles exactly and the other from a sketch //! compiles, is priced, and answers the exact one as `percentile_cont` does -//! and the sketched one within the sketch's bound. +//! and the sketched one within the sketch's bound. With Q2, the p90 per +//! city over the last 20 minutes, Q1 and Q2 share the table scan, and Q2's +//! per-city pane chain serves Q1 through its newest panes rolled up across +//! cities. mod physical_common; mod planner_layering_common; @@ -20,6 +23,8 @@ use std::rc::Rc; const Q1: &str = "SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY value), \ percentile_cont(0.9) WITHIN GROUP (ORDER BY value) FROM metrics \ WHERE ts > now() - INTERVAL '5' MINUTE"; +const Q2: &str = "SELECT city, percentile_cont(0.9) WITHIN GROUP (ORDER BY value) \ + FROM metrics WHERE ts > now() - INTERVAL '20' MINUTE GROUP BY city"; /// Every query's accuracy target. const EPSILON: f64 = 0.01; const MINUTE_MS: i64 = 60_000; @@ -149,9 +154,17 @@ fn within_bound(algorithm: Option<&SketchAlgorithm>, q: f64, answer: f64, values } /// `root`'s answer rows at [`EVALUATION_TIME_MS`] over [`rows`], every node at -/// query time. +/// query time, each pane chain as its panes. fn answers(root: &Rc) -> Vec> { - let table = rows() + answers_over(root, rows()) +} + +/// [`answers`] over the table rows `table` (time, city, value). +fn answers_over( + root: &Rc, + table: Vec<(i64, Option, f64)>, +) -> Vec> { + let table = table .into_iter() .map(|(ts, city, value)| { vec![ @@ -161,7 +174,9 @@ fn answers(root: &Rc) -> Vec> { ] }) .collect(); - physical_common::execute_raw_rows_at(root, table, EVALUATION_TIME_MS) + // The executor runs a pane chain as its panes. + let root = asap_types::ir::operator::pane_chain::expand_pane_chains(root).unwrap(); + physical_common::execute_raw_rows_at(&root, table, EVALUATION_TIME_MS) } fn as_f64(value: &Value) -> f64 { @@ -273,3 +288,172 @@ async fn q1_with_one_exact_and_one_sketched_percentile_compiles_and_matches() { } } } + +/// The id and query roots of every Stage 1 candidate of `queries`. +async fn stage1(queries: &[&str]) -> Vec<(String, Vec)> { + use asap_logical_asap_optimization::per_computation_candidates::generate_per_computation_candidates; + use asap_logical_asap_optimization::workload_candidates::stage1_logical_asap_optimization; + let workload = workload(queries); + let entries: Vec<_> = workload.query_workload.entries().collect(); + let mut roots = Vec::new(); + for (q, entry) in entries.iter().enumerate() { + let root = lower_sql_dialect( + &entry.query.0, + &catalog(), + SqlDialect::DataFusionSQL, + entry.requirements.accuracy.clone(), + ) + .await + .unwrap(); + roots.push((q, QueryRoot::Operator(root))); + } + let per_computation = generate_per_computation_candidates(roots).unwrap(); + stage1_logical_asap_optimization(&per_computation, &entries) + .enumerate() + .filter_map(|(i, c)| Some((format!("L{}", i + 1), c.roots.ok()?))) + .collect() +} + +/// Q1 and Q2 read the same rows of `metrics`: a candidate reads them with +/// one scan. A table scan has no unique key, and is shared all the same: +/// each reader reads every row it outputs. +#[tokio::test] +async fn q1_and_q2_share_the_table_scan() { + let scans = |root: &QueryRoot| -> Vec<*const OperatorNode> { + OperatorNode::reachable(operator(root)) + .iter() + .filter(|n| matches!(n.non_asap(), Some(NonASAPOp::Scan { .. }))) + .map(Rc::as_ptr) + .collect() + }; + let candidates = stage1(&[Q1, Q2]).await; + assert!( + candidates.iter().any(|(_, roots)| { + let q2_scans = scans(&roots[1]); + scans(&roots[0]).iter().any(|scan| q2_scans.contains(scan)) + }), + "no candidate shares the scan" + ); +} + +/// Whether Q2's per-city pane chain (120 panes of 10 s) serves both of +/// Q1's percentiles in `roots`: Q1 merges each city's newest 30 panes (its +/// 5 minutes), then rolls the cities up into one state, which both of its +/// estimates read. +fn newest_panes_rolled_up(roots: &[QueryRoot]) -> bool { + let (q1, q2) = (operator(&roots[0]), operator(&roots[1])); + let q2_chains: Vec<*const OperatorNode> = OperatorNode::reachable(q2) + .iter() + .filter(|n| { + matches!( + n.asap(), + Some(ASAPOp::PaneChain { + pane_count: 120, + .. + }) + ) + }) + .map(Rc::as_ptr) + .collect(); + let rollups: Vec> = OperatorNode::reachable(q1) + .into_iter() + .filter(|n| match n.asap() { + Some(ASAPOp::SummaryMerge { + children, + group_by: Some(keys), + newest_panes: None, + }) => { + keys.is_empty() + && matches!( + children[0].asap(), + Some(ASAPOp::SummaryMerge { + children: chain, + newest_panes: Some(30), + group_by: None, + }) if q2_chains.contains(&Rc::as_ptr(&chain[0])) + ) + } + _ => false, + }) + .collect(); + OperatorNode::reachable(q1) + .iter() + .filter(|n| match n.asap() { + Some(ASAPOp::SummaryEstimate { summary_input, .. }) => { + rollups.iter().any(|r| Rc::ptr_eq(r, summary_input)) + } + _ => false, + }) + .count() + == 2 +} + +/// The values of [`rows`] of city `city` (no city for `None`) in the last +/// 20 minutes, sorted. +fn city_values(city: Option<&str>) -> Vec { + let mut values: Vec = rows() + .into_iter() + .filter(|(ts, row_city, _)| { + *ts > EVALUATION_TIME_MS - 20 * MINUTE_MS && row_city.as_deref() == city + }) + .map(|(_, _, value)| value) + .collect(); + values.sort_by(f64::total_cmp); + values +} + +/// Q2's per-city chain serves both of Q1's percentiles through its newest +/// 30 panes rolled up across cities. Each query's answers are within the +/// sketch's bound of the exact `percentile_cont`, and over a window with +/// no row Q1 still answers one NULL row, as the global SQL aggregate the +/// rollup stands for. +#[tokio::test] +async fn q2_pane_chain_serves_q1_through_newest_panes_and_a_rollup() { + let candidates = stage1(&[Q1, Q2]).await; + for algorithm in [SketchAlgorithm::Kll, SketchAlgorithm::DDSketch] { + let (id, roots) = candidates + .iter() + .find(|(_, roots)| { + newest_panes_rolled_up(roots) + && sketch_of(operator(&roots[0])) == Some(algorithm.clone()) + }) + .unwrap_or_else(|| panic!("no {algorithm:?} newest-panes rollup candidate")); + let q1 = answers(operator(&roots[0])); + let values = values_in(5 * MINUTE_MS, None); + assert_eq!(q1.len(), 1, "{id}"); + for (column, q) in [0.5, 0.9].into_iter().enumerate() { + let answer = as_f64(&q1[0][column]); + assert!( + within_bound(Some(&algorithm), q, answer, &values), + "{id}: Q1 p{} = {answer}, exact {}", + q * 100.0, + percentile_cont(q, &values) + ); + } + let older = rows() + .into_iter() + .filter(|(ts, _, _)| *ts <= EVALUATION_TIME_MS - 5 * MINUTE_MS) + .collect(); + let empty = answers_over(operator(&roots[0]), older); + assert!( + matches!(empty.as_slice(), [row] if matches!(row.as_slice(), [Value::Null, Value::Null])), + "{id}: {empty:?}" + ); + let q2 = answers(operator(&roots[1])); + assert_eq!(q2.len(), 4, "{id}: three cities and none"); + for row in q2 { + let city = match &row[0] { + Value::Utf8(city) => Some(city.to_string()), + Value::Null => None, + other => panic!("city: {other:?}"), + }; + let values = city_values(city.as_deref()); + let answer = as_f64(&row[1]); + assert!( + within_bound(Some(&algorithm), 0.9, answer, &values), + "{id}: Q2 p90 of {city:?} = {answer}, exact {}", + percentile_cont(0.9, &values) + ); + } + } +} diff --git a/crates/logical-asap-optimization/src/asap_aware_cse.rs b/crates/logical-asap-optimization/src/asap_aware_cse.rs index 9ece65c8f..abeccc4c9 100644 --- a/crates/logical-asap-optimization/src/asap_aware_cse.rs +++ b/crates/logical-asap-optimization/src/asap_aware_cse.rs @@ -12,7 +12,7 @@ //! A *sharing candidate* ([`SharingCandidate`]) of a whole-workload //! candidate is a set of two or more of its nodes with identical input and //! computation semantics that are legal to share: the `ir::cse` interning -//! merges only ASAP nodes and nodes with a unique key. +//! merges only ASAP nodes, scans, and nodes with a unique key. //! [`identical_expression_pass2_candidates`] returns one Pass 2 candidate //! per non-empty subset of the sharing candidates (Algorithm 2, line 31): //! each sharing candidate in the subset becomes one node serving every @@ -91,9 +91,12 @@ //! same panes, each at its own cadence. Every window ends at the evaluation //! time, so each shorter one is the chain's newest panes: its //! `SummaryMerge` merges only those (`newest_panes`), and has the guarantee -//! of one build over its own window. Stage 2 never keeps a chain read at -//! different intervals across evaluations from query time (each reader -//! would need one new pane per evaluation). Long-window segments, sliding +//! of one build over its own window. A build grouped by fewer keys also +//! reads the chain of a finer grouping over a window at least its own +//! ([`rollup_window_groups`]): it merges its newest panes, then rolls the +//! groups up into its own, as the rollup rule does. Stage 2 never keeps a +//! chain read at different intervals across evaluations from query time +//! (each reader would need one new pane per evaluation). Long-window segments, sliding //! windows and Exponential Histograms are not implemented. //! //! # Rollup rule @@ -148,11 +151,11 @@ use asap_types::ir::schema::{ ExactKind, FamilyCapability, FieldDataType, GroupingStrategy, MergeInputRule, MergedError, Schema, SketchKind, PROMQL_SERIES_IDENTITY, }; +use asap_types::ir::ScalarExpr; use asap_types::ir::SortKey; use asap_types::ir::{ ASAPOp, NonASAPOp, Operator, OperatorNode, OperatorResultKind, QueryRoot, TimeRangeKind, }; -use asap_types::ir::{OperatorResultKind as ResultKind, ScalarExpr}; use crate::accuracy::estimators::size_for_all_readers; use crate::per_computation_candidates::{recompose_guarantees, PerComputationCandidateError}; @@ -1265,6 +1268,13 @@ pub fn window_composition_pass2_candidates( if on_gcd_panes != on_member_panes { push_shared_chains(roots, &chosen, &on_gcd_panes, &mut pass2_candidates); } + // Coarser groupings read a finer grouping's chain, rolled up. + push_shared_chains( + roots, + &chosen, + &rollup_window_groups(roots, &chosen), + &mut pass2_candidates, + ); } Ok(pass2_candidates) } @@ -1396,6 +1406,135 @@ fn nested_window_groups(candidates: &[PaneCandidate]) -> Vec { shared } +/// The pane candidates (positions in `candidates`) that can read one +/// chain of a finer grouping's panes, each group with its chain: a host, +/// and every candidate whose group keys are the host's or fewer +/// ([`rolls_up_into`]), whose window is at most the host's and whose pane +/// width the host's divides. Each member merges the newest panes of its +/// window, and a coarser member then rolls the host's groups up into its +/// own (a regrouping `SummaryMerge`): `p90 per city over 20 minutes` in +/// 10 s panes serves `p90 over 5 minutes` as the newest 30 panes merged +/// across cities. A coarser member's every reader in `roots` is an +/// estimate whose merged readout keeps a guarantee, as for the rollup rule. +/// Hosts are tried by most group keys, then longest window, and each +/// candidate joins one group. Only groups with a coarser member are +/// listed; the others are [`nested_window_groups`]'s. +fn rollup_window_groups(roots: &[QueryRoot], candidates: &[PaneCandidate]) -> Vec { + let mut nodes = Vec::new(); + let mut seen = HashSet::new(); + for node in roots.iter().flat_map(root_operators) { + collect_children_first(node, &mut seen, &mut nodes); + } + let mut readers: HashMap<*const OperatorNode, Vec<&Rc>> = HashMap::new(); + for node in &nodes { + for child in node.children() { + readers.entry(Rc::as_ptr(child)).or_default().push(node); + } + } + let rolls_up_with_guarantee = |member: &PaneCandidate| match member.build.asap() { + Some(ASAPOp::SummaryAgg { family, .. }) => readers + .get(&Rc::as_ptr(&member.build)) + .is_some_and(|readers| read_with_merged_guarantee(family, readers)), + _ => false, + }; + let mut hosts: Vec = (0..candidates.len()).collect(); + hosts.sort_by_key(|&i| { + ( + std::cmp::Reverse(group_key_count(&candidates[i].build)), + std::cmp::Reverse(candidates[i].window_ms()), + i, + ) + }); + let mut grouped = vec![false; candidates.len()]; + let mut groups = Vec::new(); + for host in hosts { + if grouped[host] { + continue; + } + let coarser = |i: usize| { + group_key_count(&candidates[i].build) < group_key_count(&candidates[host].build) + }; + let reads_host = |i: usize| { + let (member, chain) = (&candidates[i], &candidates[host]); + i == host + || (!grouped[i] + && member.window_ms() <= chain.window_ms() + && member.pane_width_ms.is_multiple_of(chain.pane_width_ms) + && rolls_up_into(member, chain) + && (!coarser(i) || rolls_up_with_guarantee(member))) + }; + let members: Vec = (0..candidates.len()).filter(|&i| reads_host(i)).collect(); + if !members.iter().any(|&i| coarser(i)) { + continue; + } + for &i in &members { + grouped[i] = true; + } + groups.push(SharedPaneChain { + members, + pane_width_ms: candidates[host].pane_width_ms, + pane_count: candidates[host].pane_count, + }); + } + groups.sort_by_key(|group| group.members[0]); + groups +} + +/// The number of group keys of a build grouped by listed keys; 0 for any +/// other build. +fn group_key_count(build: &OperatorNode) -> usize { + match build.asap() { + Some(ASAPOp::SummaryAgg { + reduction: Reduction::Reduce(by), + .. + }) if !by.is_without() => by.len(), + _ => 0, + } +} + +/// Whether `member`'s build is `host`'s ([`same_build_over_one_input`]), +/// or the same but grouped by fewer of `host`'s keys, of a family whose +/// states merge across groups ([`FamilyCapability::merges_groups`]), so a +/// rollup of `host`'s states answers it. +fn rolls_up_into(member: &PaneCandidate, host: &PaneCandidate) -> bool { + if same_build_over_one_input(member, host) { + return true; + } + let ( + Some(ASAPOp::SummaryAgg { + child: range, + family, + input, + reduction: Reduction::Reduce(by), + grouping, + filter, + }), + Some(ASAPOp::SummaryAgg { + child: host_range, + family: host_family, + input: host_input, + reduction: Reduction::Reduce(host_by), + grouping: host_grouping, + filter: host_filter, + }), + ) = (member.build.asap(), host.build.asap()) + else { + return false; + }; + let (below, host_below) = (range.children()[0], host_range.children()[0]); + // Both ranges read the same rows below, so a key names one column. + !by.is_without() + && !host_by.is_without() + && by.len() < host_by.len() + && by.keys().iter().all(|key| host_by.keys().contains(key)) + && FamilyCapability::merges_groups(family) + && same_family(family, host_family) + && (Rc::ptr_eq(below, host_below) || below == host_below) + && input == host_input + && grouping == host_grouping + && filter == host_filter +} + /// Whether two pane candidates build one family (at any size) over the /// same input below their ranges, read the same way, with the same output /// column names. @@ -1455,7 +1594,16 @@ fn shared_chain_build( candidates: &[PaneCandidate], group: &[usize], ) -> Result>, PerComputationCandidateError> { - let host = longest(candidates, group); + // Of the longest windows, the finest grouping: a rollup group's host. + let host = *group + .iter() + .max_by_key(|&&i| { + ( + candidates[i].window_ms(), + group_key_count(&candidates[i].build), + ) + }) + .expect("a group has members"); let members: Vec<&Rc> = std::iter::once(host) .chain(group.iter().copied().filter(|&i| i != host)) .map(|i| &candidates[i].build) @@ -1482,12 +1630,21 @@ fn in_panes( for &i in &shared.members { let newest = u32::try_from(candidates[i].window_ms() / shared.pane_width_ms) .expect("a member's window is at most the chain's"); - let merge = OperatorNode::new(Operator::ASAP(ASAPOp::SummaryMerge { + let merge = Rc::new(OperatorNode::new(Operator::ASAP(ASAPOp::SummaryMerge { children: vec![Rc::clone(&chain)], newest_panes: (newest < shared.pane_count).then_some(newest), group_by: None, - }))?; - replaced.insert(Rc::as_ptr(&candidates[i].build), Rc::new(merge)); + }))?); + // A coarser member rolls the chain's groups up into its own. + let member = &candidates[i].build; + let read = if group_key_count(member) == group_key_count(build) { + merge + } else { + regrouped(member, &merge).ok_or(PerComputationCandidateError::Unsupported( + "a coarser pane build whose keys the shared chain does not hold", + ))? + }; + replaced.insert(Rc::as_ptr(member), read); } } for candidate in candidates { @@ -1764,22 +1921,10 @@ fn sketch_rollup_candidates( 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 { + if !readers + .get(&Rc::as_ptr(coarse)) + .is_some_and(|readers| read_with_merged_guarantee(family, readers)) + { continue; } // A build finer than the one with the most keys would have more. @@ -1797,6 +1942,23 @@ fn sketch_rollup_candidates( candidates } +/// Whether every one of `readers`, the readers of a build of `family`, is +/// an estimate whose merged readout keeps a guarantee: not a heap's top-k +/// or a UnivMon's entropy. +fn read_with_merged_guarantee(family: &FieldDataType, readers: &[&Rc]) -> bool { + let capability = FamilyCapability::of(family); + 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, + }) +} + /// Whether `node` is a per-group sketch build grouped by listed keys whose /// states a rollup may merge across groups /// ([`FamilyCapability::merges_groups`]). @@ -1911,11 +2073,8 @@ fn finer_keys(coarse: &OperatorNode, fine: &OperatorNode) -> Option { && same_filter && fine_keys.len() > coarse_keys.len() && coarse_keys.iter().all(|key| fine_keys.contains(key)) - && same_rows(child, fine_child) - // A global SQL aggregate has a row over no input; a rollup of no - // finer group has none. - && !(coarse_keys.is_empty() && child.result_kind == ResultKind::Relation)) - .then_some(fine_keys.len()) + && same_rows(child, fine_child)) + .then_some(fine_keys.len()) } type ColumnName = (Option, String); diff --git a/crates/types/src/ir/cse.rs b/crates/types/src/ir/cse.rs index c5c7227d8..8d2743bc0 100644 --- a/crates/types/src/ir/cse.rs +++ b/crates/types/src/ir/cse.rs @@ -40,7 +40,9 @@ //! shared across consumers when its row identity is stable across reads, so //! an ungrouped aggregate, a `without(..)` grouping, a `Concat`/`SetOp` that //! drops its keys, … is always inserted fresh even when it is structurally -//! identical to something already interned. An ASAP node (summary state and +//! identical to something already interned. A `Scan` is the exception: equal +//! scans read the same source rows, so every reader of one shared scan reads +//! the rows it would have read alone, a SQL table's keyless rows included. An ASAP node (summary state and //! its evaluations) has no such gate: equal operator, schema and guarantee make //! it shareable, exactly as post-ASAP sharing decided before this IR. //! @@ -56,6 +58,7 @@ use std::hash::{Hash, Hasher}; use std::rc::Rc; use crate::ir::operator::node::{Operator, OperatorNode}; +use crate::ir::NonASAPOp; /// [`structural_hash`]'s memoization cache: an already-hashed node's `Rc` /// pointer to its hash. A fresh cache is always correct; what matters is @@ -207,7 +210,9 @@ impl InternTable { // A node that is not legal to share is never *returned* as a match // for something else; it still occupies a fresh slot in the bucket // (harmless: later scans require legality of the new node too). - let reusable = node.is_asap() || node.schema.has_unique_key(); + let reusable = node.is_asap() + || node.schema.has_unique_key() + || matches!(node.non_asap(), Some(NonASAPOp::Scan { .. })); let bucket = self.buckets.entry(hash).or_default(); if reusable { if let Some(existing) = bucket @@ -406,6 +411,33 @@ mod tests { ); } + /// Two equal scans of a keyless table (SQL rows) are one node: each + /// reader reads every row the scan outputs. + #[test] + fn equal_scans_without_a_unique_key_are_shared() { + let table_scan = || { + node(NonASAPOp::Scan { + source: Source::Table { + table_ref: "metrics".into(), + }, + predicates: vec![], + schema: Schema::with_time_index( + vec![ + Field::plain("ts", DataType::Timestamp, false), + Field::plain("city", DataType::Utf8, true), + Field::plain("value", DataType::Float64, true), + ], + 0, + vec![], + ), + }) + }; + let a = table_scan(); + assert!(!a.schema.has_unique_key(), "fixture sanity: keyless rows"); + let (ra, rb) = two_roots(a, table_scan()); + assert!(Rc::ptr_eq(&ra, &rb)); + } + /// Grouped quantiles differing only in accuracy keep two roots over one shared scan. #[test] fn differing_accuracy_is_never_merged_but_the_scan_child_is_shared() { diff --git a/crates/types/src/ir/properties/summary_coverage.rs b/crates/types/src/ir/properties/summary_coverage.rs index 7336cc8dc..e535d843c 100644 --- a/crates/types/src/ir/properties/summary_coverage.rs +++ b/crates/types/src/ir/properties/summary_coverage.rs @@ -19,7 +19,7 @@ use serde::{Deserialize, Serialize}; use thiserror::Error; use crate::ir::operator::asap::ASAPOp; -use crate::ir::operator::node::{Operator, OperatorNode, OperatorResultKind}; +use crate::ir::operator::node::{Operator, OperatorNode}; use crate::ir::operator::non_asap::{NonASAPOp, TimeRangeKind}; use crate::ir::operator::operator_properties::Reduction; use crate::ir::scalar::expr_ir::{CompareOpKind, ScalarValue}; @@ -235,11 +235,6 @@ pub enum CoverageError { /// bad input. #[error("the regrouped build does not derive: {reason}")] RegroupedBuild { reason: String }, - /// A regrouping `SummaryMerge` to no keys over a relation's states: a - /// global SQL aggregate has one row over no input rows, a merge of no - /// states none. Bad input. - #[error("a regrouping merge to no keys cannot stand for a global SQL aggregate")] - GlobalRelationRegroup, } /// Coverage already derived during one validation, by node address. The @@ -394,9 +389,6 @@ fn regrouped(merged: SummaryCoverage, keys: &[ColumnId]) -> Result= 100` give `latency ∈ (−∞, ∞)`, which still excludes NULL), or value sets of one column (`{us}` and `{eu}` give `{us, eu}`). Gaps stay as separate boxes | | State or value? | state in, state out | @@ -1408,9 +1408,8 @@ impl SummaryCoverage { /// in time for counter states, as the family's `MergeInputRule` /// requires); for a rollup, also `NotAGroupKey` (a `group_by` column is /// not a group key of the inputs' build), `FamilyCannotRegroup` (counter - /// states, a family without a merge, or a Hydra), `GlobalRelationRegroup` - /// (no keys over a SQL relation) or `RegroupedBuild` (the coarse build - /// it stands for does not derive). + /// states, a family without a merge, or a Hydra) or `RegroupedBuild` + /// (the coarse build it stands for does not derive). pub fn derive(node: &OperatorNode) -> Result; } ``` diff --git a/docs/design_docs/proposals/planner-layering.md b/docs/design_docs/proposals/planner-layering.md index 82ac0d3c7..8270b128d 100644 --- a/docs/design_docs/proposals/planner-layering.md +++ b/docs/design_docs/proposals/planner-layering.md @@ -559,7 +559,10 @@ as one input: the implementation widens every scan of the metric without label matchers to the union of those labels. Time ranges, time shifts, and per-series summaries without a filter (with the nodes that read their state) keep the wider rows; every other consumer reads its own columns through a projection. -These candidates are kept only when they share a scan. +These candidates are kept only when they share a scan. A SQL table scan reads +every column of the table, so two queries over one table already have equal +scans; a scan is shared even though a table's rows have no unique key, since +each reader reads every row the scan outputs. The examples behind these rules: @@ -761,9 +764,21 @@ For example, `quantile_over_time(0.5, m[1m])` and 10-s KLL panes: the 1-min query merges the newest 6, the 5-min query all 30. And `quantile_over_time(0.5, m[15m])` every 1 min and every 5 min read one chain of fifteen 1-min panes instead of fifteen 1-min and three 5-min -panes. A family that merges exactly keeps each query's guarantee, since a -merge of fewer panes is one build over the shorter window, and each query -fails with at most the shared state's failure probability. Stage 3 prices +panes. A query grouped by fewer keys also reads the chain of a finer +grouping whose window is at least its own, on panes that divide its own, +when the family merges states of different groups: it merges its newest +panes, then rolls the groups up into its own (a regrouping `SummaryMerge` +over the newest-panes merge). For example, `SELECT city, +percentile_cont(0.9) … INTERVAL '20' MINUTE GROUP BY city` and the global +p50 and p90 over the last 5 minutes, both every 10 s, read one chain of 120 +per-city 10-s KLL panes: the 5-minute query merges each city's newest 30, +then all cities into one state that answers both its percentiles. The chain +is sized for every reader, here both queries' ε. A rollup to no keys over +a SQL table's states stands for a global SQL aggregate, so it has one +state, empty over no rows. A family that merges exactly keeps each query's +guarantee, since a merge of fewer panes is one build over the shorter +window, and each query fails with at most the shared state's failure +probability. Stage 3 prices each merge by the panes it reads, at its query's rate. The per-query chains stay candidates too. A chain read at different intervals is not kept across evaluations from query time: that option needs every reader to build one From ca473b72ced5896b819d7e7f6e3fe1a0345e1af4 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 22 Oct 2026 16:21:11 +0000 Subject: [PATCH 2/2] fix(pass2): keep refusing an exact rollup to no keys over a SQL relation (a count of no rows is 0, a sum of no counts NULL) --- crates/logical-asap-optimization/src/asap_aware_cse.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/crates/logical-asap-optimization/src/asap_aware_cse.rs b/crates/logical-asap-optimization/src/asap_aware_cse.rs index abeccc4c9..aa13b1b22 100644 --- a/crates/logical-asap-optimization/src/asap_aware_cse.rs +++ b/crates/logical-asap-optimization/src/asap_aware_cse.rs @@ -2073,7 +2073,13 @@ fn finer_keys(coarse: &OperatorNode, fine: &OperatorNode) -> Option { && same_filter && fine_keys.len() > coarse_keys.len() && coarse_keys.iter().all(|key| fine_keys.contains(key)) - && same_rows(child, fine_child)) + && same_rows(child, fine_child) + // A global SQL count over no rows is 0, but an exact rollup sums no + // finer value into NULL. A sketch rollup to no keys has one empty + // state over no rows, as the global build has. + && !(coarse_keys.is_empty() + && child.result_kind == OperatorResultKind::Relation + && matches!(family, FieldDataType::ExactAggregate(..)))) .then_some(fine_keys.len()) }