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..aa13b1b22 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`]). @@ -1912,10 +2074,13 @@ fn finer_keys(coarse: &OperatorNode, fine: &OperatorNode) -> Option { && 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()) + // 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()) } 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