From 8ee7fa464469b194695d656814224d20a83ddb59 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Fri, 6 Nov 2026 21:31:24 +0000 Subject: [PATCH 1/2] refactor(stage0, types): memo hash-consing and candidate storage (#1207 PR 2) --- crates/frontend-common/src/canonicalize.rs | 2 +- crates/stage0-memo/src/lib.rs | 496 ++++++++++++++++++- crates/types/src/ir/schema/schema_support.rs | 2 +- crates/types/src/stages/candidates.rs | 479 +++++++++++++++++- crates/types/src/stages/hash_consing.rs | 59 +++ crates/types/src/stages/memo.rs | 8 +- crates/types/src/stages/mod.rs | 1 + 7 files changed, 1016 insertions(+), 31 deletions(-) create mode 100644 crates/types/src/stages/hash_consing.rs diff --git a/crates/frontend-common/src/canonicalize.rs b/crates/frontend-common/src/canonicalize.rs index 68a1eb47c..43495769f 100644 --- a/crates/frontend-common/src/canonicalize.rs +++ b/crates/frontend-common/src/canonicalize.rs @@ -1,7 +1,7 @@ //! Post-lowering canonicalization of the operator DAG. //! //! Rewrites equivalent spellings of a query into one shape, so that common -//! sub-DAG sharing (Stage 0 hash-consing, `cse` in `asap_stage0_memo`) can match +//! sub-DAG sharing (hash-consing, `Memo::find_or_add_group`) can match //! them by structural equality, whichever front end produced them. //! //! ## Heavy-hitter promotion diff --git a/crates/stage0-memo/src/lib.rs b/crates/stage0-memo/src/lib.rs index 9311a60dd..80697373c 100644 --- a/crates/stage0-memo/src/lib.rs +++ b/crates/stage0-memo/src/lib.rs @@ -1,21 +1,493 @@ //! `asap-stage0-memo` — #509 Stage 0: the memo groups of the workload's //! query roots (a [`Memo`] without candidates). Each node of the queries' //! DAGs becomes the group of its expression, found or added by -//! [`Memo::find_or_add_group`], so identical sub-expressions are one group. -//! Stage 1 starts from these groups. +//! [`Memo::find_or_add_group`], so identical sub-expressions, within one +//! query or across queries, are one group. Stage 1 starts from these groups. //! -//! It depends only on `asap-types`, where its output (`Memo`) lives; no -//! other stage crate depends on it (the frontends' tests aside, listed in -//! `crate_graph.rs`), and the planner facade calls it first. +//! It depends only on `asap-types`, where its output (`Memo`) lives and +//! where the one expression equality is; no other stage crate depends on +//! it, and the planner facade calls it first. -use asap_types::ir::{QueryRoot, SchemaDerivationError}; -use asap_types::stages::Memo; +use std::collections::HashMap; +use std::rc::Rc; + +use asap_types::ir::{OperatorNode, QueryRoot, SchemaDerivationError}; +use asap_types::stages::{Memo, MemoGroupId}; /// Stage 0: the memo groups of the workload's query roots, query `i` -/// being `roots[i]`, every group without candidates. Fails with the first -/// node that breaks the node rules over its children's groups -/// ([`Memo::find_or_add_group`]). +/// being `roots[i]` (added by [`Memo::add_query`], so each group lists the +/// queries whose root reaches it), every group without candidates. Fails with the first node that breaks the +/// node rules over its children's groups ([`Memo::find_or_add_group`]); the +/// facade validates the roots first, so that is a frontend bug. pub fn memo_of_queries(roots: &[QueryRoot]) -> Result { - let _ = roots; - todo!("#1207 PR 2: Stage 0 over Memo::find_or_add_group") + let mut memo = Memo::default(); + let mut group_of_node = GroupOfNode::new(); + for root in roots { + for node in root.operator_refs() { + group_of(&mut memo, &mut group_of_node, node)?; + } + } + let group = |node: &Rc| group_of_node[&Rc::as_ptr(node)]; + for root in roots { + memo.add_query(match root { + QueryRoot::Operator(node) => QueryRoot::Operator(group(node)), + QueryRoot::Scalar(expr) => { + QueryRoot::Scalar(expr.map_operator_refs(&mut |node| group(node))) + } + }); + } + Ok(memo) +} + +/// Each node of the queries' DAGs already visited, by address, to its +/// group: a node reached through several paths is visited once. The roots +/// outlive the map, so no address is reused. +type GroupOfNode = HashMap<*const OperatorNode, MemoGroupId>; + +/// The group of `node`: its operator over its children's groups (each +/// found or added first), found or added by [`Memo::find_or_add_group`]. +fn group_of( + memo: &mut Memo, + group_of_node: &mut GroupOfNode, + node: &Rc, +) -> Result { + if let Some(&group) = group_of_node.get(&Rc::as_ptr(node)) { + return Ok(group); + } + for child in node.children() { + group_of(memo, group_of_node, child)?; + } + let expression = node + .operator + .map_children(|child| group_of_node[&Rc::as_ptr(child)]); + let group = memo.find_or_add_group(expression)?; + group_of_node.insert(Rc::as_ptr(node), group); + Ok(group) +} + +#[cfg(test)] +mod tests { + use super::*; + use asap_types::ir::operator::agg_intent::AggIntent; + use asap_types::ir::operator::asap::ASAPOp; + use asap_types::ir::operator::operator_properties::{ + BinaryOpKind, GroupKeys, Reduction, Source, + }; + use asap_types::ir::properties::AccuracyTarget; + use asap_types::ir::scalar::expr_ir::{ColumnRef, CompareOpKind}; + use asap_types::ir::schema::sketch::{ + GroupingStrategy, SketchAlgorithm, SketchStatistic, SummaryUpdate, + }; + use asap_types::ir::schema::{DataType, Field, FieldDataType, Schema}; + use asap_types::ir::{BinaryOperator, NonASAPOp, Operator, ScalarExpr}; + + fn node(op: NonASAPOp) -> Rc { + OperatorNode::new_shared(Operator::NonASAP(op)).unwrap() + } + + /// `[ts, service, value, latency]`, no unique key. + fn scan() -> Rc { + node(NonASAPOp::Scan { + source: Source::TimeSeries { metric: "m".into() }, + predicates: vec![], + schema: Schema::with_time_index( + vec![ + Field::plain("ts", DataType::Timestamp, false), + Field::plain("service", DataType::Utf8, false), + Field::plain("value", DataType::Float64, false), + Field::plain("latency", DataType::Float64, false), + ], + 0, + vec![], + ), + }) + } + + fn quantile_agg_over( + child: Rc, + by: Vec, + col: Option, + q: f64, + accuracy: AccuracyTarget, + ) -> Rc { + node(NonASAPOp::Aggregate { + reduction: Reduction::by(by), + measures: vec![AggIntent::Quantile { col, q, accuracy }], + output_names: vec![], + filters: vec![], + having: None, + child, + }) + } + + fn quantile_agg(by: Vec, col: Option, q: f64) -> Rc { + quantile_agg_over(scan(), by, col, q, AccuracyTarget::exact()) + } + + fn compare(lhs: Rc, rhs: Rc) -> Rc { + node(NonASAPOp::BinaryOp { + operator: BinaryOperator { + checked_relative_division: false, + checked_finite_division: false, + kind: BinaryOpKind::Compare(CompareOpKind::Eq), + vector_match: None, + }, + return_bool: false, + lhs, + rhs, + }) + } + + /// The memo of `roots`, one query per root. + fn memo_of(roots: Vec>) -> Memo { + let roots: Vec = roots.into_iter().map(QueryRoot::Operator).collect(); + memo_of_queries(&roots).unwrap() + } + + /// Query `query`'s root group. + fn root(memo: &Memo, query: usize) -> MemoGroupId { + match &memo.query_roots[query] { + QueryRoot::Operator(group) => *group, + QueryRoot::Scalar(_) => panic!("expected an operator root"), + } + } + + /// Whether queries 0 and 1 of `[a, b]` are one group. + fn one_group(a: Rc, b: Rc) -> bool { + let memo = memo_of(vec![a, b]); + root(&memo, 0) == root(&memo, 1) + } + + /// Two queries sharing a sub-expression read one group for it, and + /// every group lists the queries that reach it and the groups that + /// read it. + #[test] + fn two_queries_sharing_a_sub_expression_get_one_group() { + let p50 = quantile_agg(vec![1], Some(2), 0.5); + let p50_vs_p90 = compare( + quantile_agg(vec![1], Some(2), 0.5), + quantile_agg(vec![1], Some(2), 0.9), + ); + let memo = memo_of(vec![p50, p50_vs_p90]); + // scan, p50, p90, compare. + assert_eq!(memo.groups().len(), 4); + let (scan, p50, compare) = (MemoGroupId(0), root(&memo, 0), root(&memo, 1)); + let p90 = MemoGroupId(2); + assert_eq!( + memo.groups()[compare.0 as usize].expression.children(), + vec![&p50, &p90] + ); + assert_eq!(memo.groups()[scan.0 as usize].reached_by_queries(), &[0, 1]); + assert_eq!(memo.groups()[p50.0 as usize].reached_by_queries(), &[0, 1]); + assert_eq!(memo.groups()[p90.0 as usize].reached_by_queries(), &[1]); + assert_eq!(memo.groups()[compare.0 as usize].reached_by_queries(), &[1]); + assert_eq!(memo.groups()[scan.0 as usize].parent_groups(), &[p50, p90]); + assert_eq!(memo.groups()[p50.0 as usize].parent_groups(), &[compare]); + assert!(memo + .groups() + .iter() + .all(|group| group.candidates.is_empty())); + } + + /// One query's repeated sub-expression is one group, read twice. + #[test] + fn a_query_s_repeated_sub_expression_is_one_group() { + let memo = memo_of(vec![compare( + quantile_agg(vec![1], Some(2), 0.5), + quantile_agg(vec![1], Some(2), 0.5), + )]); + assert_eq!(memo.groups().len(), 3); + let p50 = MemoGroupId(1); + assert_eq!( + memo.groups()[root(&memo, 0).0 as usize] + .expression + .children(), + vec![&p50, &p50] + ); + assert_eq!( + memo.groups()[p50.0 as usize].parent_groups(), + &[root(&memo, 0)] + ); + } + + /// A node reached through two paths is one group, visited once. + #[test] + fn an_already_shared_node_is_one_group() { + let p50 = quantile_agg(vec![1], Some(2), 0.5); + let memo = memo_of(vec![compare(Rc::clone(&p50), p50)]); + assert_eq!(memo.groups().len(), 3); + } + + /// Aggregates that differ in a column, a quantile or a filter are + /// different groups. + #[test] + fn aggregates_that_differ_in_an_argument_are_different_groups() { + use asap_types::ir::scalar::ScalarValue; + use asap_types::ir::Predicate; + assert!(!one_group( + quantile_agg(vec![1], Some(2), 0.5), + quantile_agg(vec![1], Some(3), 0.5) + )); + assert!(!one_group( + quantile_agg(vec![], Some(2), 0.5), + quantile_agg(vec![], Some(2), 0.9) + )); + let filtered = node(NonASAPOp::Aggregate { + reduction: Reduction::by(vec![1]), + measures: vec![AggIntent::Quantile { + col: Some(2), + q: 0.5, + accuracy: AccuracyTarget::exact(), + }], + output_names: vec![], + filters: vec![Some(Predicate(ScalarExpr::Compare { + left: Box::new(ScalarExpr::Column(3)), + op: CompareOpKind::Gt, + right: Box::new(ScalarExpr::Literal(ScalarValue::Float64(1.0))), + semantics: asap_types::ir::ExprSemantics::Sql, + }))], + having: None, + child: scan(), + }); + assert!(!one_group(quantile_agg(vec![1], Some(2), 0.5), filtered)); + } + + /// Equal nodes without a unique key are one group when they return the + /// same rows each time: a keyless scan, a global aggregate, a `without` + /// aggregate. + #[test] + fn equal_keyless_nodes_that_return_the_same_rows_are_one_group() { + 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("value", DataType::Float64, true), + ], + 0, + vec![], + ), + }) + }; + assert!(one_group(table_scan(), table_scan())); + assert!(one_group( + quantile_agg(vec![], Some(2), 0.9), + quantile_agg(vec![], Some(2), 0.9) + )); + let without = || { + node(NonASAPOp::Aggregate { + reduction: Reduction::Reduce(GroupKeys::without(vec![0])), + measures: vec![AggIntent::Count { + accuracy: AccuracyTarget::exact(), + }], + output_names: vec![], + filters: vec![], + having: None, + child: scan(), + }) + }; + assert!(one_group(without(), without())); + } + + /// Quantiles that differ only in their accuracy target are two groups + /// over one scan group. + #[test] + fn differing_accuracy_targets_are_two_groups_over_one_scan_group() { + let quantile = |epsilon| { + quantile_agg_over( + scan(), + vec![1], + Some(2), + 0.99, + AccuracyTarget::max_error(epsilon), + ) + }; + let memo = memo_of(vec![quantile(0.01), quantile(0.05)]); + assert_ne!(root(&memo, 0), root(&memo, 1)); + assert_eq!(memo.groups().len(), 3); + } + + /// `vector(scalar(sum by (service) (m)))`. + fn scalar_of_vector() -> Rc { + let sum_up = node(NonASAPOp::Aggregate { + reduction: Reduction::by(vec![1]), + measures: vec![AggIntent::Sum { col: Some(2) }], + output_names: vec![], + filters: vec![], + having: None, + child: scan(), + }); + node(NonASAPOp::PromqlVectorFromScalar( + ScalarExpr::PromqlScalarFromVector(sum_up), + )) + } + + /// A vector read through `scalar(v)` is a child group like any input, + /// one group across queries, and a scalar query's root reads it. + #[test] + fn a_scalar_referenced_vector_is_one_group_across_queries() { + let roots = vec![ + QueryRoot::Operator(scalar_of_vector()), + QueryRoot::Operator(scalar_of_vector()), + QueryRoot::Scalar(ScalarExpr::PromqlScalarFromVector(Rc::clone( + scalar_of_vector().children()[0], + ))), + ]; + let memo = memo_of_queries(&roots).unwrap(); + // scan, sum, bridge. + assert_eq!(memo.groups().len(), 3); + assert_eq!(root(&memo, 0), root(&memo, 1)); + let sum = MemoGroupId(1); + assert_eq!(memo.query_roots[2].operator_refs(), vec![&sum]); + assert_eq!( + memo.groups()[sum.0 as usize].reached_by_queries(), + &[0, 1, 2] + ); + } + + /// `SummaryAgg[DDSketch](scan)`. + fn summary_agg() -> Rc { + let family = FieldDataType::Sketch(SketchAlgorithm::DDSketch, GroupingStrategy::default()); + OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { + child: scan(), + family, + input: SummaryUpdate::column(ColumnRef::SampleValue), + reduction: Reduction::PerEntity, + grouping: GroupingStrategy::default(), + filter: None, + })) + .unwrap() + } + + /// Two quantile readouts of one summary are two groups over one state + /// group. + #[test] + fn readouts_share_their_state_group_but_not_each_other() { + let readout = |q| { + OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryEstimate { + summary_input: summary_agg(), + query: SketchStatistic::Quantile { q }, + })) + .unwrap() + }; + let memo = memo_of(vec![readout(0.95), readout(0.99)]); + let (p95, p99) = (root(&memo, 0), root(&memo, 1)); + assert_ne!(p95, p99); + assert_eq!( + memo.groups()[p95.0 as usize].expression.children(), + memo.groups()[p99.0 as usize].expression.children() + ); + assert_eq!(memo.groups().len(), 4); + } + + /// `DISTINCT` on every column returns the same rows each time and is one + /// group; a `Dedup` on some columns keeps an unspecified row per key. + #[test] + fn dedup_is_one_group_only_on_every_column() { + let dedup = |cols: Vec| { + node(NonASAPOp::Dedup { + cols, + child: scan(), + }) + }; + assert!(one_group(dedup(vec![]), dedup(vec![]))); + assert!(!one_group(dedup(vec![1]), dedup(vec![1]))); + } + + fn unordered_limit(child: Rc) -> Rc { + node(NonASAPOp::Limit { + n: Some(10), + offset: 0, + partition_by: GroupKeys::default(), + child, + }) + } + + /// Two unordered `LIMIT 10`s may return different rows: two groups, and + /// so are equal parents over them. + #[test] + fn unordered_limits_and_their_parents_are_different_groups() { + assert!(!one_group(unordered_limit(scan()), unordered_limit(scan()))); + let distinct_over_limit = || { + node(NonASAPOp::Dedup { + cols: vec![], + child: unordered_limit(scan()), + }) + }; + assert!(!one_group(distinct_over_limit(), distinct_over_limit())); + } + + /// A `LIMIT` over a `Sort` with the same partitions keeps the first rows + /// of a fixed order (one group); over a `Sort` partitioned differently, + /// which rows come first depends on input order (two groups). + #[test] + fn limits_over_a_sort_with_their_partitions_are_one_group() { + let limit = |sort_partitions: GroupKeys| { + node(NonASAPOp::Limit { + n: Some(10), + offset: 0, + partition_by: GroupKeys::default(), + child: node(NonASAPOp::Sort { + keys: vec![asap_types::ir::scalar::SortKey { + expr: ScalarExpr::Column(2), + ascending: false, + nulls_first: false, + }], + partition_by: sort_partitions, + child: scan(), + }), + }) + }; + assert!(one_group( + limit(GroupKeys::default()), + limit(GroupKeys::default()) + )); + assert!(!one_group( + limit(GroupKeys::by(vec![1])), + limit(GroupKeys::by(vec![1])) + )); + } + + /// A projection emitting the literal `value`. + fn literal(value: f64) -> Rc { + node(NonASAPOp::Project { + cols: vec![asap_types::ir::ProjectItem { + alias: Some("v".into()), + expr: ScalarExpr::literal_f64(value), + }], + qualifier: None, + child: scan(), + }) + } + + /// Signed zeros and different non-finite literals are different groups; + /// JSON's `null` for non-finite floats never decides equality. + #[test] + fn signed_zero_and_non_finite_literals_are_different_groups() { + for (a, b) in [ + (0.0, -0.0), + (-0.0, 0.0), + (f64::INFINITY, f64::NEG_INFINITY), + (f64::NAN, f64::NAN), + ] { + assert!(!one_group(literal(a), literal(b)), "{a} and {b}"); + } + assert!(one_group(literal(f64::INFINITY), literal(f64::INFINITY))); + } + + /// A frontend node that breaks the node rules (a readout of raw rows + /// instead of summary state) makes Stage 0 return the error. + #[test] + fn memo_of_queries_returns_the_error_of_a_node_that_breaks_the_node_rules() { + let readout_of_rows = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryEstimate { + summary_input: scan(), + query: SketchStatistic::Quantile { q: 0.5 }, + })) + .unwrap(); + assert!(memo_of_queries(&[QueryRoot::Operator(readout_of_rows)]).is_err()); + } } diff --git a/crates/types/src/ir/schema/schema_support.rs b/crates/types/src/ir/schema/schema_support.rs index 94e997390..61bc53e69 100644 --- a/crates/types/src/ir/schema/schema_support.rs +++ b/crates/types/src/ir/schema/schema_support.rs @@ -64,7 +64,7 @@ pub fn with_promql_series_identity( // A series has at most one sample per timestamp, so the full // identity and the timestamp key each row. Pass 2's // identical-expression rule reads this key: only a keyed - // node is legal to share (Stage 0 hash-consing, `cse` in `asap_stage0_memo`). + // node is legal to share (hash-consing, `Memo::find_or_add_group`). if let Some(time) = schema.time_index { schema.add_unique_key(vec![schema.fields.len() - 1, time]); } diff --git a/crates/types/src/stages/candidates.rs b/crates/types/src/stages/candidates.rs index dfad7cdee..d4b0d21bc 100644 --- a/crates/types/src/stages/candidates.rs +++ b/crates/types/src/stages/candidates.rs @@ -3,8 +3,9 @@ //! $r^{\log}_{c,i}$ (concepts doc `memo-groups-properties-requirements.md`, //! section 3; #1203). -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; +use super::hash_consing::{expression_hash, same_expression, same_value}; use super::{GroupProperties, MemoGroup, MemoGroupAsInput, MemoGroupId}; use crate::ir::properties::summary_coverage::SummaryCoverage; use crate::ir::schema::{ @@ -34,6 +35,10 @@ pub struct Memo { groups: Vec, /// Each query's root over groups, by index in the workload. pub query_roots: Vec>, + /// The groups [`Self::find_or_add_group`] may return, by + /// [`expression_hash`] of their expression: every group whose sharing + /// is legal ([`GroupProperties::shareable_when_equal`]). + groups_by_expression_hash: HashMap>, } impl Memo { @@ -46,34 +51,149 @@ impl Memo { /// groups and every group below them list it /// ([`MemoGroup::reached_by_queries`]). Stage 0 adds each query once. pub fn add_query(&mut self, root: QueryRoot) -> usize { - let _ = root; - todo!("#1207 PR 2") + let query = self.query_roots.len(); + let groups = root.operator_refs().into_iter().copied().collect(); + self.query_roots.push(root); + self.add_queries_below(groups, &[query]); + query } - /// Hash-consing: the group with this expression, added if new, with the - /// `GroupProperties` [`derive_group_properties`] derives. Fails when the - /// expression breaks the node rules (schema derivation, input - /// validation, merge check). + /// Hash-consing: the group whose expression is `expression` + /// (`same_expression`: operator, arguments and input groups), else a + /// new group for it, with the `GroupProperties` [`derive_group_properties`] + /// derives from its input groups, listed as a parent of each input + /// group. A group whose copies may return different rows (an unordered + /// `LIMIT`, [`GroupProperties::shareable_when_equal`]) is never returned + /// for another expression, so such an expression always gets a new + /// group. Fails, leaving the memo unchanged, when the expression breaks + /// the node rules (schema derivation, input validation, merge check). pub fn find_or_add_group( &mut self, expression: Operator, ) -> Result { - let _group_properties = derive_group_properties(self, &expression)?; - todo!("#1207 PR 2: hash-consing over groups") + let group_properties = derive_group_properties(self, &expression)?; + let hash = expression_hash(&expression); + let shareable = group_properties.shareable_when_equal; + if shareable { + if let Some(&group) = self + .groups_by_expression_hash + .get(&hash) + .and_then(|groups| { + groups.iter().find(|&&group| { + same_expression(&self.groups[group.0 as usize].expression, &expression) + }) + }) + { + return Ok(group); + } + } + let id = MemoGroupId( + u32::try_from(self.groups.len()).expect("a memo has at most u32::MAX groups"), + ); + for &input in expression.children() { + add_parent(&mut self.groups[input.0 as usize], id); + } + self.groups.push(MemoGroup { + expression, + group_properties, + parents: Vec::new(), + queries: Vec::new(), + candidates: Vec::new(), + }); + if shareable { + self.groups_by_expression_hash + .entry(hash) + .or_default() + .push(id); + } + Ok(id) } - /// Adds `candidate` to `group`'s candidates. Fails when its expression - /// breaks the node rules ([`derive_group_properties`]) or its output - /// schema or kind differs from the group's. + /// Adds `candidate` to `group`'s candidates and returns its id, unless + /// the group already has a candidate with the same expression + /// (`same_expression`) and logical properties: then that candidate's + /// id, and `candidate` is dropped. A new candidate makes `group` a parent + /// of each of its input groups, and `group`'s queries reach each input + /// group and every group below it. + /// + /// Fails, leaving the memo unchanged, when the candidate's expression + /// breaks the node rules a group's does ([`derive_group_properties`]), + /// or when its output schema or kind differs from the group's: then it + /// does not compute the group's result. pub fn add_candidate( &mut self, group: MemoGroupId, candidate: CandidateForMemoGroup, ) -> Result { - let _output = derive_group_properties(self, &candidate.expression.operator) + let output = derive_group_properties(self, &candidate.expression.operator) .map_err(CandidateError::Schema)?; - let _ = group; - todo!("#1207 PR 2") + let group_properties = &self.groups[group.0 as usize].group_properties; + if output.schema != group_properties.schema + || output.result_kind != group_properties.result_kind + { + return Err(CandidateError::OutputDiffersFromGroup { group }); + } + let candidates = &mut self.groups[group.0 as usize].candidates; + if let Some(index_in_group) = candidates + .iter() + .position(|known| same_expression_and_logical_properties(known, &candidate)) + { + return Ok(CandidateForMemoGroupId { + group, + index_in_group, + }); + } + let input_groups: Vec = candidate + .expression + .operator + .children() + .into_iter() + .copied() + .collect(); + candidates.push(candidate); + let index_in_group = candidates.len() - 1; + for &input in &input_groups { + add_parent(&mut self.groups[input.0 as usize], group); + } + let queries = self.groups[group.0 as usize].queries.clone(); + self.add_queries_below(input_groups, &queries); + Ok(CandidateForMemoGroupId { + group, + index_in_group, + }) + } + + /// Adds `queries` to each of `groups` and every group below them, through + /// the groups' expressions and their candidates' input groups. A group + /// that has every query already is not walked: every group below it has + /// them too. + fn add_queries_below(&mut self, groups: Vec, queries: &[usize]) { + let mut to_visit = groups; + while let Some(group) = to_visit.pop() { + let group = &mut self.groups[group.0 as usize]; + let mut added = false; + for &query in queries { + if let Err(position) = group.queries.binary_search(&query) { + group.queries.insert(position, query); + added = true; + } + } + if added { + to_visit.extend( + group + .expression + .children() + .into_iter() + .chain( + group + .candidates + .iter() + .flat_map(|candidate| candidate.expression.operator.children()), + ) + .copied(), + ); + } + } } /// The candidate `id`. @@ -165,6 +285,28 @@ pub enum CandidateError { OutputDiffersFromGroup { group: MemoGroupId }, } +/// Lists `parent` among `group`'s parents, once. +fn add_parent(group: &mut MemoGroup, parent: MemoGroupId) { + if !group.parents.contains(&parent) { + group.parents.push(parent); + } +} + +/// Whether `left` and `right` are one candidate of a group: the same +/// expression ([`same_expression`]) and logical properties (parameters by +/// [`same_value`], window layout). +fn same_expression_and_logical_properties( + left: &CandidateForMemoGroup, + right: &CandidateForMemoGroup, +) -> bool { + same_expression(&left.expression.operator, &right.expression.operator) + && same_value( + &left.logical_properties.parameters, + &right.logical_properties.parameters, + ) + && left.logical_properties.window_layout == right.logical_properties.window_layout +} + /// $v = (e_v, p^{\log}_v)$: one candidate of a memo group. #[derive(Debug, Clone)] pub struct CandidateForMemoGroup { @@ -1160,3 +1302,310 @@ mod tests { } } } + +/// The `Memo` methods: hash-consing groups ([`Memo::find_or_add_group`]) and +/// storing candidates ([`Memo::add_candidate`]), on the concepts doc's KLL +/// example (section 1: G1 the latency scan, G2 its KLL state). +#[cfg(test)] +mod memo_tests { + use super::*; + use crate::ir::operator::operator_properties::{GroupKeys, Reduction, Source}; + use crate::ir::scalar::expr_ir::ColumnRef; + use crate::ir::schema::{DataType, Field, Schema, SketchStatistic, SummaryUpdate}; + use crate::ir::NonASAPOp; + + /// `Scan metric[1m]`: `[ts, value]`. + fn scan(metric: &str) -> Operator { + Operator::NonASAP(NonASAPOp::Scan { + source: Source::TimeSeries { + metric: metric.into(), + }, + predicates: vec![], + schema: Schema::with_time_index( + vec![ + Field::plain("ts", DataType::Timestamp, false), + Field::plain("value", DataType::Float64, false), + ], + 0, + vec![], + ), + }) + } + + /// `SummaryAgg[KLL](input)`: no θ in the expression. + fn kll_state(input: MemoGroupId) -> Operator { + Operator::ASAP(ASAPOp::SummaryAgg { + child: input, + family: FieldDataType::Sketch(SketchAlgorithm::Kll, GroupingStrategy::default()), + input: SummaryUpdate::column(ColumnRef::SampleValue), + reduction: Reduction::PerEntity, + grouping: GroupingStrategy::default(), + filter: None, + }) + } + + /// `SummaryEstimate[q](state)`. + fn quantile_readout(state: MemoGroupId, q: f64) -> Operator { + Operator::ASAP(ASAPOp::SummaryEstimate { + summary_input: state, + query: SketchStatistic::Quantile { q }, + }) + } + + /// An unordered `LIMIT 10` over `input`, whose copies may return + /// different rows. + fn unordered_limit(input: MemoGroupId) -> Operator { + Operator::NonASAP(NonASAPOp::Limit { + n: Some(10), + offset: 0, + partition_by: GroupKeys::default(), + child: input, + }) + } + + /// `Project(input)` keeping its first column. + fn project_first_column(input: MemoGroupId) -> Operator { + Operator::NonASAP(NonASAPOp::Project { + cols: vec![crate::ir::ProjectItem { + alias: None, + expr: crate::ir::ScalarExpr::Column(0), + }], + qualifier: None, + child: input, + }) + } + + /// A candidate computing `operator` with `parameters` and no window. + fn candidate( + operator: Operator, + parameters: Option, + ) -> CandidateForMemoGroup { + let input_requirements = vec![LogicalRequirements::new(None); operator.children().len()]; + CandidateForMemoGroup { + // Built directly: the doc example's k = 200 and k = 345 are not + // in today's KLL list, which `LogicalProperties::new` checks. + logical_properties: LogicalProperties { + parameters, + window_layout: None, + }, + expression: CandidateExpression { operator }, + input_requirements, + fused_groups: vec![], + origin: OriginOfCandidateForMemoGroup::Pass1, + } + } + + /// KLL `k` (no recorded failure probability). + fn kll_k(k: u32) -> Option { + Some(SketchParams::Kll { + k, + failure_probability: None, + }) + } + + /// An equal expression over the same input groups is found, not added. + #[test] + fn find_or_add_group_returns_the_group_of_an_equal_expression() { + let mut memo = Memo::default(); + let g1 = memo.find_or_add_group(scan("latency")).unwrap(); + let g2 = memo.find_or_add_group(kll_state(g1)).unwrap(); + assert_eq!(memo.find_or_add_group(scan("latency")).unwrap(), g1); + assert_eq!(memo.find_or_add_group(kll_state(g1)).unwrap(), g2); + assert_ne!(memo.find_or_add_group(scan("errors")).unwrap(), g1); + assert_eq!(memo.groups().len(), 3); + assert_eq!( + memo.groups()[g2.0 as usize].expression.children(), + vec![&g1] + ); + } + + /// The same operator over different input groups is a different group. + #[test] + fn find_or_add_group_separates_equal_operators_over_different_inputs() { + let mut memo = Memo::default(); + let latency = memo.find_or_add_group(scan("latency")).unwrap(); + let errors = memo.find_or_add_group(scan("errors")).unwrap(); + assert_ne!( + memo.find_or_add_group(kll_state(latency)).unwrap(), + memo.find_or_add_group(kll_state(errors)).unwrap() + ); + } + + /// An expression whose copies may return different rows (an unordered + /// `LIMIT`) gets a new group every time. + #[test] + fn find_or_add_group_never_returns_an_unshareable_group() { + let mut memo = Memo::default(); + let g1 = memo.find_or_add_group(scan("latency")).unwrap(); + assert_ne!( + memo.find_or_add_group(unordered_limit(g1)).unwrap(), + memo.find_or_add_group(unordered_limit(g1)).unwrap() + ); + } + + /// A new group is a parent of each of its input groups, listed once. + #[test] + fn find_or_add_group_lists_the_new_group_among_its_inputs_parents() { + let mut memo = Memo::default(); + let g1 = memo.find_or_add_group(scan("latency")).unwrap(); + let g2 = memo.find_or_add_group(kll_state(g1)).unwrap(); + memo.find_or_add_group(kll_state(g1)).unwrap(); + assert_eq!(memo.groups()[g1.0 as usize].parent_groups(), &[g2]); + assert!(memo.groups()[g2.0 as usize].parent_groups().is_empty()); + } + + /// KLL builds with k = 200 and k = 345 are two candidates of one group + /// G2: θ is a logical property, not part of the expression. + #[test] + fn kll_builds_with_k_200_and_k_345_share_one_group() { + let mut memo = Memo::default(); + let g1 = memo.find_or_add_group(scan("latency")).unwrap(); + let k200_group = memo.find_or_add_group(kll_state(g1)).unwrap(); + let k200 = memo + .add_candidate(k200_group, candidate(kll_state(g1), kll_k(200))) + .unwrap(); + let k345_group = memo.find_or_add_group(kll_state(g1)).unwrap(); + let k345 = memo + .add_candidate(k345_group, candidate(kll_state(g1), kll_k(345))) + .unwrap(); + assert_eq!(k200_group, k345_group); + assert_ne!(k200, k345); + assert_eq!(memo.groups()[k200_group.0 as usize].candidates.len(), 2); + assert_eq!( + memo.candidate(k345).logical_properties.parameters(), + kll_k(345).as_ref() + ); + } + + /// A candidate with the same expression and logical properties as one + /// in the group is not added again; its id is returned. + #[test] + fn add_candidate_returns_the_existing_candidate_for_a_duplicate() { + let mut memo = Memo::default(); + let g1 = memo.find_or_add_group(scan("latency")).unwrap(); + let g2 = memo.find_or_add_group(kll_state(g1)).unwrap(); + let first = memo + .add_candidate(g2, candidate(kll_state(g1), kll_k(200))) + .unwrap(); + let again = memo + .add_candidate(g2, candidate(kll_state(g1), kll_k(200))) + .unwrap(); + assert_eq!(first, again); + assert_eq!( + first, + CandidateForMemoGroupId { + group: g2, + index_in_group: 0 + } + ); + let mut windowed = candidate(kll_state(g1), kll_k(200)); + windowed.logical_properties.window_layout = + Some(WindowLayout::Tumbling { width_ms: 10_000 }); + assert_ne!(memo.add_candidate(g2, windowed).unwrap(), first); + assert_eq!(memo.groups()[g2.0 as usize].candidates.len(), 2); + } + + /// A new candidate makes its group a parent of each group it reads, and + /// the group's queries reach those groups and every group below them. + #[test] + fn add_candidate_maintains_parents_and_queries() { + // Query 0 is G3 = SummaryEstimate[q=0.95](G2) over G2 = KLL(G1). + let mut memo = Memo::default(); + let g1 = memo.find_or_add_group(scan("latency")).unwrap(); + let g2 = memo.find_or_add_group(kll_state(g1)).unwrap(); + let g3 = memo.find_or_add_group(quantile_readout(g2, 0.95)).unwrap(); + assert_eq!(memo.add_query(QueryRoot::Operator(g3)), 0); + // A rule adds a state group G5 over another scan G4: no query + // reaches them yet. + let g4 = memo.find_or_add_group(scan("latency_copy")).unwrap(); + let g5 = memo.find_or_add_group(kll_state(g4)).unwrap(); + assert!(memo.groups()[g5.0 as usize].reached_by_queries().is_empty()); + // G3 gains a candidate reading G5: G5 and G4 now serve query 0. + memo.add_candidate(g3, candidate(quantile_readout(g5, 0.95), None)) + .unwrap(); + assert_eq!(memo.groups()[g5.0 as usize].parent_groups(), &[g3]); + assert_eq!(memo.groups()[g4.0 as usize].parent_groups(), &[g5]); + assert_eq!(memo.groups()[g5.0 as usize].reached_by_queries(), &[0]); + assert_eq!(memo.groups()[g4.0 as usize].reached_by_queries(), &[0]); + // Query 1 is G6 = Project(G3): it reaches G3 and everything below + // it, through expressions and candidates, in ascending order. + let g6 = memo.find_or_add_group(project_first_column(g3)).unwrap(); + assert_eq!(memo.add_query(QueryRoot::Operator(g6)), 1); + for group in [g1, g2, g3, g4, g5] { + assert_eq!( + memo.groups()[group.0 as usize].reached_by_queries(), + &[0, 1], + "group {group:?}" + ); + } + assert_eq!(memo.groups()[g3.0 as usize].parent_groups(), &[g6]); + } + + /// A candidate that computes another result than its group's (a + /// readout offered for a state group) is rejected and changes nothing. + #[test] + fn add_candidate_rejects_a_candidate_whose_output_differs_from_its_group() { + let mut memo = Memo::default(); + let g1 = memo.find_or_add_group(scan("latency")).unwrap(); + let g2 = memo.find_or_add_group(kll_state(g1)).unwrap(); + let before = format!("{memo:?}"); + assert!(matches!( + memo.add_candidate(g2, candidate(quantile_readout(g2, 0.5), None)), + Err(CandidateError::OutputDiffersFromGroup { group }) if group == g2 + )); + assert_eq!( + format!("{memo:?}"), + before, + "a rejected candidate changes nothing" + ); + } + + /// A candidate whose expression fails input validation (a readout of + /// raw rows instead of summary state) is rejected and changes nothing. + #[test] + fn add_candidate_rejects_a_candidate_whose_inputs_fail_validation() { + let mut memo = Memo::default(); + let g1 = memo.find_or_add_group(scan("latency")).unwrap(); + let g2 = memo.find_or_add_group(kll_state(g1)).unwrap(); + let g3 = memo.find_or_add_group(quantile_readout(g2, 0.95)).unwrap(); + let over_raw_rows = quantile_readout(g1, 0.95).map_children(|&group_id| { + MemoGroupAsInput { + memo: &memo, + group_id, + } + }); + assert!( + over_raw_rows.output_schema().is_ok(), + "fixture: the schema derives; only input validation rejects it" + ); + let before = format!("{memo:?}"); + assert!(matches!( + memo.add_candidate(g3, candidate(quantile_readout(g1, 0.95), None)), + Err(CandidateError::Schema(_)) + )); + assert_eq!( + format!("{memo:?}"), + before, + "a rejected candidate changes nothing" + ); + } + + /// An expression with no output schema (a group-by column out of range) + /// is an error, and no group is added. + #[test] + fn find_or_add_group_returns_an_error_for_an_expression_with_no_schema() { + let mut memo = Memo::default(); + let g1 = memo.find_or_add_group(scan("latency")).unwrap(); + let out_of_range = Operator::NonASAP(NonASAPOp::Aggregate { + reduction: Reduction::by(vec![9]), + measures: vec![crate::ir::operator::agg_intent::AggIntent::Sum { col: Some(1) }], + output_names: vec![], + filters: vec![], + having: None, + child: g1, + }); + assert!(memo.find_or_add_group(out_of_range).is_err()); + assert_eq!(memo.groups().len(), 1); + assert!(memo.groups()[g1.0 as usize].parent_groups().is_empty()); + } +} diff --git a/crates/types/src/stages/hash_consing.rs b/crates/types/src/stages/hash_consing.rs new file mode 100644 index 000000000..3a482dda0 --- /dev/null +++ b/crates/types/src/stages/hash_consing.rs @@ -0,0 +1,59 @@ +//! The one equality that makes two expressions one memo group, and two +//! candidates of a group one candidate: [`same_expression`] (operator, +//! arguments and input groups). [`Memo::find_or_add_group`] looks a group up +//! by it, Stage 0 included, and [`Memo::add_candidate`] finds a duplicate by +//! it. Parameters θ are not an argument of any operator (they are a +//! candidate's `LogicalProperties`), so builds of one family over one input +//! that differ only in θ are one group. +//! +//! [`Memo::find_or_add_group`]: super::Memo::find_or_add_group +//! [`Memo::add_candidate`]: super::Memo::add_candidate + +use std::collections::hash_map::DefaultHasher; +use std::hash::{Hash, Hasher}; + +use super::MemoGroupId; +use crate::ir::Operator; + +/// `expression` with every input group replaced by `()`: the operator and +/// its arguments. +fn operator_and_arguments(expression: &Operator) -> Operator<()> { + expression.map_children(|_| ()) +} + +/// Whether `left` and `right` are one expression: equal operators and +/// arguments ([`same_value`]) over the same input groups at the same +/// positions. +pub(super) fn same_expression(left: &Operator, right: &Operator) -> bool { + left.children() == right.children() + && same_value( + &operator_and_arguments(left), + &operator_and_arguments(right), + ) +} + +/// A hash consistent with [`same_expression`], used only to narrow the +/// groups compared: SipHash over the canonical JSON of the operator and its +/// arguments, then the input groups. A non-finite `f64` serializes as +/// `null`, which only widens one bucket; [`same_expression`] decides. +pub(super) fn expression_hash(expression: &Operator) -> u64 { + let mut hasher = DefaultHasher::new(); + serde_json::to_string(&operator_and_arguments(expression)) + .unwrap_or_default() + .hash(&mut hasher); + expression.children().hash(&mut hasher); + hasher.finish() +} + +/// Typed equality, confirmed by equal serializations: numeric `PartialEq` +/// alone conflates signed zeros, and `SketchParams`' equality omits the +/// recorded error and failure probability. The serialized check is +/// additional evidence, never a replacement for typed equality (JSON maps +/// non-finite floats to `null`). +pub(super) fn same_value(left: &T, right: &T) -> bool { + left == right + && match (serde_json::to_string(left), serde_json::to_string(right)) { + (Ok(left), Ok(right)) => left == right, + _ => false, + } +} diff --git a/crates/types/src/stages/memo.rs b/crates/types/src/stages/memo.rs index 153346093..d6842846f 100644 --- a/crates/types/src/stages/memo.rs +++ b/crates/types/src/stages/memo.rs @@ -25,10 +25,14 @@ pub struct MemoGroup { /// changes. pub group_properties: GroupProperties, /// The groups that read it ([`Self::parent_groups`]) through their - /// expression or a candidate's. + /// expression or a candidate's: kept up to date by + /// [`Memo::find_or_add_group`](super::Memo::find_or_add_group) and + /// [`Memo::add_candidate`](super::Memo::add_candidate). pub parents: Vec, /// The queries whose roots reach it ([`Self::reached_by_queries`]), - /// ascending. + /// ascending: kept up to date by + /// [`Memo::add_query`](super::Memo::add_query) and + /// [`Memo::add_candidate`](super::Memo::add_candidate). pub queries: Vec, /// Its candidates: empty after Stage 0, filled by Stage 1. pub candidates: Vec, diff --git a/crates/types/src/stages/mod.rs b/crates/types/src/stages/mod.rs index 596aee4f4..873c276a6 100644 --- a/crates/types/src/stages/mod.rs +++ b/crates/types/src/stages/mod.rs @@ -8,6 +8,7 @@ //! live here. mod candidates; +mod hash_consing; mod logical_memo_candidates; mod memo; mod per_computation_logical_memo_candidates; From e7109069646c25a970a73ecba5c5aa91f524f500 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Fri, 6 Nov 2026 21:37:46 +0000 Subject: [PATCH 2/2] =?UTF-8?q?test(types):=20memo=20summary=20coverage;?= =?UTF-8?q?=20asap-primitive-schema=20=C2=A74.2.2=20over=20memo=20groups?= =?UTF-8?q?=20(#1207=20PR=202)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/types/src/stages/candidates.rs | 8 +- crates/types/tests/main.rs | 1 + crates/types/tests/memo_summary_coverage.rs | 406 ++++++++++++++++++ .../proposals/asap-primitive-schema.md | 187 +++++--- 4 files changed, 530 insertions(+), 72 deletions(-) create mode 100644 crates/types/tests/memo_summary_coverage.rs diff --git a/crates/types/src/stages/candidates.rs b/crates/types/src/stages/candidates.rs index d4b0d21bc..17d72ada0 100644 --- a/crates/types/src/stages/candidates.rs +++ b/crates/types/src/stages/candidates.rs @@ -1568,11 +1568,9 @@ mod memo_tests { let g1 = memo.find_or_add_group(scan("latency")).unwrap(); let g2 = memo.find_or_add_group(kll_state(g1)).unwrap(); let g3 = memo.find_or_add_group(quantile_readout(g2, 0.95)).unwrap(); - let over_raw_rows = quantile_readout(g1, 0.95).map_children(|&group_id| { - MemoGroupAsInput { - memo: &memo, - group_id, - } + let over_raw_rows = quantile_readout(g1, 0.95).map_children(|&group_id| MemoGroupAsInput { + memo: &memo, + group_id, }); assert!( over_raw_rows.output_schema().is_ok(), diff --git a/crates/types/tests/main.rs b/crates/types/tests/main.rs index bf56559c0..8c1477bf9 100644 --- a/crates/types/tests/main.rs +++ b/crates/types/tests/main.rs @@ -2,6 +2,7 @@ //! its own copy of every dependency, which outgrew CI's disk. //! One file: `cargo test -p asap-types --test types ::`. +mod memo_summary_coverage; mod physical_export; mod planner_vocabulary; mod schema_rebuilding; diff --git a/crates/types/tests/memo_summary_coverage.rs b/crates/types/tests/memo_summary_coverage.rs new file mode 100644 index 000000000..cf96a6c45 --- /dev/null +++ b/crates/types/tests/memo_summary_coverage.rs @@ -0,0 +1,406 @@ +//! Group properties and summary coverage, derived once per memo group from +//! its input groups (`docs/design_docs/proposals/asap-primitive-schema.md` +//! §4.2.2). The memo of the worked example: +//! +//! ```text +//! G1 Scan latency +//! G2 Filter(job = "api")(G1); G3 TimeRange[1m](G2); G4 SummaryAgg[KLL](G3) +//! G5 Filter(job = "web")(G1); G6 TimeRange[1m](G5); G7 SummaryAgg[KLL](G6) +//! G8 Filter(job = "api" ∧ a + b > 10)(G1); G9 TimeRange[1m](G8); G10 SummaryAgg[KLL](G9) +//! ``` +use std::ops::Bound; +use std::time::Duration; + +use asap_types::ir::operator::operator_properties::{GroupKeys, Reduction, Source}; +use asap_types::ir::properties::summary_coverage::{ + ColumnIdentity, Constraint, CoverageError, SummaryCoverage, +}; +use asap_types::ir::scalar::expr_ir::{ArithmeticOpKind, CompareOpKind, ScalarValue}; +use asap_types::ir::scalar::ColumnRef; +use asap_types::ir::schema::{ + DataType, Field, FieldDataType, GroupingStrategy, MergeInputRule, Schema, SketchAlgorithm, + SketchStatistic, SummaryUpdate, +}; +use asap_types::ir::{ + ASAPOp, ExprSemantics, NonASAPOp, Operator, OperatorResultKind, Predicate, ScalarExpr, + SchemaDerivationError, TimeRangeKind, +}; +use asap_types::stages::{Memo, MemoGroupId}; +use asap_types::test_support::definition_id; + +const JOB: usize = 2; +const A: usize = 3; +const B: usize = 4; +const MINUTE_MS: i64 = 60_000; + +/// `Scan latency{predicates}`: `[ts, value, job, a, b]`. +fn scan(predicates: Vec>) -> Operator { + Operator::NonASAP(NonASAPOp::Scan { + source: Source::TimeSeries { + metric: "latency".into(), + }, + predicates: predicates.into_iter().map(Predicate).collect(), + schema: Schema::with_time_index( + vec![ + Field::plain("ts", DataType::Timestamp, false), + Field::plain("value", DataType::Float64, false), + Field::plain("job", DataType::Utf8, false), + Field::plain("a", DataType::Float64, false), + Field::plain("b", DataType::Float64, false), + ], + 0, + vec![], + ), + }) +} + +fn job_is(job: &str) -> ScalarExpr { + ScalarExpr::Compare { + left: Box::new(ScalarExpr::Column(JOB)), + op: CompareOpKind::Eq, + right: Box::new(ScalarExpr::Literal(ScalarValue::Utf8(job.into()))), + semantics: ExprSemantics::Sql, + } +} + +/// `a + b > 10`: it spans two columns, so it is no selection. +fn a_plus_b_above_10() -> ScalarExpr { + ScalarExpr::Compare { + left: Box::new(ScalarExpr::Arithmetic { + op: ArithmeticOpKind::Add, + left: Box::new(ScalarExpr::Column(A)), + right: Box::new(ScalarExpr::Column(B)), + semantics: ExprSemantics::Sql, + }), + op: CompareOpKind::Gt, + right: Box::new(ScalarExpr::Literal(ScalarValue::Float64(10.0))), + semantics: ExprSemantics::Sql, + } +} + +fn filter(pred: ScalarExpr, child: MemoGroupId) -> Operator { + Operator::NonASAP(NonASAPOp::Filter { + pred: Predicate(pred), + child, + }) +} + +fn range(width_ms: i64, child: MemoGroupId) -> Operator { + Operator::NonASAP(NonASAPOp::TimeRange { + range: Duration::from_millis(width_ms as u64), + kind: TimeRangeKind::Range, + child, + }) +} + +/// `SummaryAgg[KLL]` of the sample value per series. +fn kll(child: MemoGroupId) -> Operator { + Operator::ASAP(ASAPOp::SummaryAgg { + child, + family: FieldDataType::Sketch(SketchAlgorithm::Kll, GroupingStrategy::default()), + input: SummaryUpdate::column(ColumnRef::SampleValue), + reduction: Reduction::PerEntity, + grouping: GroupingStrategy::default(), + filter: None, + }) +} + +fn merge(children: Vec) -> Operator { + Operator::ASAP(ASAPOp::SummaryMerge { + children, + newest_tumbling_windows: None, + group_by: None, + }) +} + +/// KLL over the last minute of `G1` filtered by `pred`. +fn kll_of_last_minute_where( + memo: &mut Memo, + g1: MemoGroupId, + pred: ScalarExpr, +) -> MemoGroupId { + let filtered = memo.find_or_add_group(filter(pred, g1)).unwrap(); + let last_minute = memo.find_or_add_group(range(MINUTE_MS, filtered)).unwrap(); + memo.find_or_add_group(kll(last_minute)).unwrap() +} + +fn coverage(memo: &Memo, group: MemoGroupId) -> &SummaryCoverage { + memo.groups()[group.0 as usize] + .group_properties + .summary_coverage + .as_ref() + .expect("a summary state") +} + +/// `KLL` as an unlifted operator: its main input is `None`. +fn kll_unlifted() -> Operator> { + kll(MemoGroupId(0)).map_children(|_| None) +} + +fn jobs(values: &[&str]) -> Constraint { + Constraint::In( + values + .iter() + .map(|value| ScalarValue::Utf8((*value).into())) + .collect(), + ) +} + +fn last_minute() -> Option<(Bound, Bound)> { + Some((Bound::Excluded(-MINUTE_MS), Bound::Included(0))) +} + +/// A group's properties come from its operator and its input groups' +/// properties: the scan's schema, the state's kind and state column, the +/// readout's value column, and whether equal copies may be one group. +#[test] +fn group_properties_come_from_input_groups() { + let mut memo = Memo::default(); + let g1 = memo.find_or_add_group(scan(vec![])).unwrap(); + let g4 = kll_of_last_minute_where(&mut memo, g1, job_is("api")); + let readout = memo + .find_or_add_group(Operator::ASAP(ASAPOp::SummaryEstimate { + summary_input: g4, + query: SketchStatistic::Quantile { q: 0.99 }, + })) + .unwrap(); + let limit = memo + .find_or_add_group(Operator::NonASAP(NonASAPOp::Limit { + n: Some(1), + offset: 0, + partition_by: GroupKeys::default(), + child: g1, + })) + .unwrap(); + let properties = |group: MemoGroupId| &memo.groups()[group.0 as usize].group_properties; + assert_eq!(properties(g1).schema.fields.len(), 5); + assert_eq!(properties(g1).summary_coverage, None); + assert_eq!(properties(g4).result_kind, OperatorResultKind::State); + assert!(properties(g4) + .schema + .fields + .iter() + .any(|field| field.name == "state" && !field.is_plain())); + assert!(properties(g4).summary_coverage.is_some()); + assert_eq!( + properties(readout).result_kind, + OperatorResultKind::InstantVector + ); + assert!(properties(readout) + .schema + .fields + .iter() + .any(|field| field.name == "quantile" + && field.dtype == FieldDataType::Plain(DataType::Float64))); + assert!(properties(g4).shareable_when_equal); + assert!(!properties(limit).shareable_when_equal); +} + +/// G4: `job = "api"` and the minute are lifted; the definition is the KLL +/// over G1. +#[test] +fn a_summary_agg_over_a_job_filter_lifts_it_into_the_selection() { + let mut memo = Memo::default(); + let g1 = memo.find_or_add_group(scan(vec![])).unwrap(); + let g4 = kll_of_last_minute_where(&mut memo, g1, job_is("api")); + let coverage = coverage(&memo, g4); + assert_eq!(coverage.definition().input_group, Some(g1)); + assert_eq!( + coverage.definition().unlifted_operators, + vec![kll_unlifted()] + ); + let [selection] = coverage.selection() else { + panic!("one box: {coverage:?}") + }; + assert_eq!( + selection.columns()[&ColumnIdentity::new(None, "job")], + jobs(&["api"]) + ); + assert_eq!(selection.relative_time(), last_minute()); +} + +/// G7 has G4's definition and id and a disjoint selection, so +/// `SummaryMerge(G4, G7)` is a valid group covering job ∈ {api, web}. +#[test] +fn states_with_equal_definitions_and_disjoint_selections_merge() { + let mut memo = Memo::default(); + let g1 = memo.find_or_add_group(scan(vec![])).unwrap(); + let g4 = kll_of_last_minute_where(&mut memo, g1, job_is("api")); + let g7 = kll_of_last_minute_where(&mut memo, g1, job_is("web")); + assert_eq!( + coverage(&memo, g4).definition(), + coverage(&memo, g7).definition() + ); + assert_eq!( + definition_id(coverage(&memo, g4)), + definition_id(coverage(&memo, g7)) + ); + let merged = memo.find_or_add_group(merge(vec![g4, g7])).unwrap(); + let coverage = coverage(&memo, merged); + assert_eq!(coverage.definition().input_group, Some(g1)); + let [selection] = coverage.selection() else { + panic!("one box: {coverage:?}") + }; + assert_eq!( + selection.columns()[&ColumnIdentity::new(None, "job")], + jobs(&["api", "web"]) + ); +} + +/// Two states that select the same rows do not merge: KLL counts a row +/// once per input. +#[test] +fn states_with_overlapping_selections_do_not_merge() { + let mut memo = Memo::default(); + let g1 = memo.find_or_add_group(scan(vec![])).unwrap(); + let g4 = kll_of_last_minute_where(&mut memo, g1, job_is("api")); + let groups = memo.groups().len(); + assert!(matches!( + memo.find_or_add_group(merge(vec![g4, g4])), + Err(SchemaDerivationError::Coverage( + CoverageError::InputsOverlap { + required: MergeInputRule::Disjoint + } + )) + )); + assert_eq!(memo.groups().len(), groups); +} + +/// G10: `a + b > 10` spans two columns, so it stays as a `Filter` in the +/// definition; G10 does not merge with G4 although both select job = api. +#[test] +fn a_filter_that_is_no_selection_stays_unlifted() { + let mut memo = Memo::default(); + let g1 = memo.find_or_add_group(scan(vec![])).unwrap(); + let g4 = kll_of_last_minute_where(&mut memo, g1, job_is("api")); + let g10 = kll_of_last_minute_where( + &mut memo, + g1, + ScalarExpr::BoolAnd(vec![job_is("api"), a_plus_b_above_10()]), + ); + let coverage_of_g10 = coverage(&memo, g10); + assert_eq!(coverage_of_g10.definition().input_group, Some(g1)); + assert_eq!( + coverage_of_g10.definition().unlifted_operators, + vec![ + kll_unlifted(), + filter(a_plus_b_above_10(), MemoGroupId(0)).map_children(|_| None), + ] + ); + assert_eq!( + coverage_of_g10.selection()[0].columns()[&ColumnIdentity::new(None, "job")], + jobs(&["api"]) + ); + assert!(matches!( + memo.find_or_add_group(merge(vec![g4, g10])), + Err(SchemaDerivationError::Coverage( + CoverageError::DefinitionMismatch + )) + )); +} + +/// `a > (SELECT threshold)`: a scalar subquery reading group `threshold`. +fn a_above(threshold: MemoGroupId) -> ScalarExpr { + ScalarExpr::Compare { + left: Box::new(ScalarExpr::Column(A)), + op: CompareOpKind::Gt, + right: Box::new(ScalarExpr::ScalarSubquery(threshold)), + semantics: ExprSemantics::Sql, + } +} + +/// A residual `Filter` that reads a scalar subquery names the subquery's +/// group (`Some`) next to its main input (`None`). Two states with that +/// equal definition and disjoint selections merge. +#[test] +fn states_whose_definitions_read_the_same_subquery_merge() { + let mut memo = Memo::default(); + let g1 = memo.find_or_add_group(scan(vec![])).unwrap(); + let threshold = memo + .find_or_add_group(Operator::NonASAP(NonASAPOp::Values { + rows: vec![vec![ScalarExpr::Literal(ScalarValue::Float64(10.0))]], + schema: Schema::new(vec![Field::plain("threshold", DataType::Float64, false)]), + })) + .unwrap(); + let state_of_job = |memo: &mut Memo, job: &str| { + let pred = ScalarExpr::BoolAnd(vec![job_is(job), a_above(threshold)]); + kll_of_last_minute_where(memo, g1, pred) + }; + let api = state_of_job(&mut memo, "api"); + let web = state_of_job(&mut memo, "web"); + let definition = coverage(&memo, api).definition(); + assert_eq!(definition.input_group, Some(g1)); + assert_eq!( + definition.unlifted_operators, + vec![ + kll_unlifted(), + Operator::NonASAP(NonASAPOp::Filter { + pred: Predicate(a_above(threshold).map_operator_refs(&mut |group| Some(*group))), + child: None, + }), + ] + ); + assert_eq!(definition, coverage(&memo, web).definition()); + let merged = memo.find_or_add_group(merge(vec![api, web])).unwrap(); + assert_eq!( + coverage(&memo, merged).selection()[0].columns()[&ColumnIdentity::new(None, "job")], + jobs(&["api", "web"]) + ); +} + +/// PromQL label matchers are the scan's predicates: they are lifted too, so +/// `latency{job="api"}` and `latency{job="web"}` states merge. The scan, +/// without them, is the last unlifted operator. +#[test] +fn scan_predicates_lift_and_their_states_merge() { + let mut memo = Memo::default(); + let state_of_job = |memo: &mut Memo, job: &str| { + let series = memo.find_or_add_group(scan(vec![job_is(job)])).unwrap(); + let last_minute = memo.find_or_add_group(range(MINUTE_MS, series)).unwrap(); + memo.find_or_add_group(kll(last_minute)).unwrap() + }; + let api = state_of_job(&mut memo, "api"); + let web = state_of_job(&mut memo, "web"); + let definition = coverage(&memo, api).definition(); + assert_eq!(definition.input_group, None); + assert_eq!( + definition.unlifted_operators, + vec![kll_unlifted(), scan(vec![]).map_children(|_| None)] + ); + assert_eq!( + coverage(&memo, api).selection()[0].columns()[&ColumnIdentity::new(None, "job")], + jobs(&["api"]) + ); + let merged = memo.find_or_add_group(merge(vec![api, web])).unwrap(); + assert_eq!( + coverage(&memo, merged).selection()[0].columns()[&ColumnIdentity::new(None, "job")], + jobs(&["api", "web"]) + ); +} + +/// A tumbling-window chain of five minutes covers what one build over the +/// five-minute range does: the same definition and selection. +#[test] +fn a_tumbling_window_chain_covers_its_whole_window() { + let mut memo = Memo::default(); + let g1 = memo.find_or_add_group(scan(vec![])).unwrap(); + let chain = memo + .find_or_add_group(Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { + child: g1, + family: FieldDataType::Sketch(SketchAlgorithm::Kll, GroupingStrategy::default()), + input: SummaryUpdate::column(ColumnRef::SampleValue), + reduction: Reduction::PerEntity, + grouping: GroupingStrategy::default(), + filter: None, + tumbling_window_width_ms: MINUTE_MS as u64, + tumbling_window_count: 5, + })) + .unwrap(); + let five_minutes = memo.find_or_add_group(range(5 * MINUTE_MS, g1)).unwrap(); + let direct = memo.find_or_add_group(kll(five_minutes)).unwrap(); + assert_eq!(coverage(&memo, chain), coverage(&memo, direct)); + assert_eq!( + coverage(&memo, chain).selection()[0].relative_time(), + Some((Bound::Excluded(-5 * MINUTE_MS), Bound::Included(0))) + ); +} diff --git a/docs/design_docs/proposals/asap-primitive-schema.md b/docs/design_docs/proposals/asap-primitive-schema.md index d6920d4e6..ac0c7c1ac 100644 --- a/docs/design_docs/proposals/asap-primitive-schema.md +++ b/docs/design_docs/proposals/asap-primitive-schema.md @@ -112,10 +112,10 @@ A node in the physical data will represent the data or summary instance, so a no | Why not put it in the schema? | States worth merging cover different data but must have the same schema | below | | What is it based on? | Goldstein & Larson view matching (SIGMOD 2001), explained with an example | §4.1 | | What does it store? | `definition` (what is computed) + `selection` (which rows were taken) | §4.2.1 | -| How is it computed? | by the planner, from the sub-DAG the node covers, in one walk that yields both the definition and the selection | §4.2.2 | -| How expensive is it? | proportional to the few operators directly under the `SummaryAgg`, not to the whole sub-DAG; cached per node | §4.2.3 | +| How is it computed? | by the planner, from the memo group's input groups, in one walk that yields both the definition and the selection | §4.2.2 | +| How expensive is it? | proportional to the few operators directly under the `SummaryAgg`, not to the whole sub-DAG; derived once per memo group | §4.2.3 | | What uses it? | merge (in #646), rollup; slice, reuse (planned); subtract (reserved) | §5.6, §5.7 | -| Where is it in the code? | `OperatorNode::coverage()`, `SummaryCoverage::derive` | §6.1, §6.6 | +| Where is it in the code? | `GroupProperties::summary_coverage`, `SummaryCoverage::derive_from_inputs` | §6.6 | | What is left out? | the deployment and runtime implementation, e.g. SDS | §4.3 | **Why coverage is not part of the schema.** @@ -235,70 +235,109 @@ Formally, a state means `family(input(σ(C)))` for each group of `G`, where `C` #### 4.2.2 Deriving the definition and the selection -The planner computes the coverage of a node from the sub-DAG the node covers, not from a declaration. One walk down the sub-DAG produces both parts: every filter condition either moves into `selection` or stays in `definition`. The definition is described first, then how the planner decides which conditions move into the selection. +Coverage is derived for memo groups, never declared and never computed on operator-node trees. When `Memo::find_or_add_group` creates a summary-state group, `SummaryCoverage::derive_from_inputs` walks down from the group's `SummaryAgg` through its input groups' `expression`s in the memo and stores the result once in the group's `GroupProperties::summary_coverage`. One walk produces both parts: every filter condition either moves into the `selection` or stays in the `definition`. A `SummaryMerge` group reads its input groups' stored coverage; it is a valid group only when their definitions are equal and their selections are disjoint. -**Worked example.** p99 latency per job over table `t` (columns `job`, `region`, `latency`): +```rust +/// What a summary state computes, with every lifted selection removed. +pub struct SummaryStateDefinition { + /// The group the unlifted operators read: the first input under the + /// SummaryAgg that is not Filter / range TimeRange / TimeShift without @ + /// / direct-column Project. `None` when that input is a `Scan` with a + /// liftable predicate (a PromQL label matcher): it is lifted too, and the + /// `Scan` with the rest of its predicates is the last unlifted operator. + pub input_group: Option, + /// The operators not lifted into the selection, listed top down from the + /// SummaryAgg, each with only its residual arguments. `None` marks an + /// operator's main input (the next unlifted operator, or input_group); + /// `Some(g)` is another input group it reads, such as the group a scalar + /// subquery in a Filter reads. + pub unlifted_operators: Vec>>, +} -```sql -SELECT job, - approx_percentile_cont(latency, 0.99) FILTER (WHERE latency < 100) -FROM (SELECT job, region AS r, latency - FROM t - WHERE region = 'us' AND latency * 2 > 10) -GROUP BY job; +pub struct SummaryCoverage { + pub(crate) definition: SummaryStateDefinition, + /// Union of boxes over the output rows of the unlifted operators' input. + pub(crate) selection: Vec, +} + +impl SummaryCoverage { + pub(crate) fn derive_from_inputs( + operator: &Operator>, + ) -> Result; + /// A hash of the definition: equal definitions of one memo get equal ids. + pub(crate) fn definition_id(&self) -> String; +} ``` -The SQL frontend lowers it as follows: `FILTER (WHERE …)` becomes the aggregate's filter, `region AS r` becomes a `Project`, and the `WHERE` conditions are folded into the `Scan`'s predicates. With an error target, the planner answers the quantile from a KLL `SummaryAgg`, read out by a `SummaryEstimate`. The sub-DAG under that `SummaryEstimate`: +**Worked example.** Three queries over the series `latency` (labels `job`, `a`, `b`), each a KLL state over the last minute: ```text - ( SummaryEstimate p99 ) - ▲ - │ - [[ SummaryAgg ]] KLL(latency) by job, filter: latency < 100 - ▲ - │ - [ Project ] job, region AS r, latency - ▲ - │ - [ Scan t ] predicates: region = 'us' AND latency * 2 > 10 +G1 Scan latency +G2 Filter(job = "api")(G1); G3 TimeRange[1m](G2); G4 SummaryAgg[KLL](G3) ← Q1 +G5 Filter(job = "web")(G1); G6 TimeRange[1m](G5); G7 SummaryAgg[KLL](G6) ← Q2 +G8 Filter(job = "api" ∧ a + b > 10)(G1); G9 TimeRange[1m](G8); G10 SummaryAgg[KLL](G9) ← Q3 ``` -##### The definition - +**Case 1, G4: everything is lifted.** The walk starts at G4's `SummaryAgg` and reads each input group's `expression`: -The `definition` is the `SummaryAgg` together with its sub-DAG, with every condition that moves into `selection` (below) taken out. Everything else stays exactly as it is: +| Step | Group | Expression | What the walk does | +|---|---|---|---| +| 1 | G4 | `SummaryAgg[KLL](G3)` | the first unlifted operator, with no filter of its own | +| 2 | G3 | `TimeRange[1m](G2)` | the one range window: lifted as time `(−60 s, 0]` | +| 3 | G2 | `Filter(job = "api")(G1)` | `job = "api"` is a value set on one column: lifted, nothing of the `Filter` remains | +| 4 | G1 | `Scan latency` | not liftable, no predicate to lift: the walk stops; G1 is the `input_group` | -| In the sub-DAG | In the `definition` | -|---|---| -| a `Filter` whose conditions all move into `selection` | removed | -| a `Filter` with some conditions that stay | kept, with only the conditions that stay | -| `Scan.predicates` and `SummaryAgg.filter` | trimmed the same way | -| the one range `TimeRange`, and any `TimeShift`s, when they become relative time | removed | -| a `TimeShift` with no range `TimeRange`, or two or more range `TimeRange`s | unchanged: no time is taken out | -| any other operator | unchanged | +```text +G4.summary_coverage = SummaryCoverage { + definition: SummaryStateDefinition { + input_group: Some(G1), + unlifted_operators: [SummaryAgg[KLL]], + }, + selection: [ { job ∈ {api}, time (−60 s, 0] } ], +} +``` -So the `definition` holds what the state computes: the computation `C` with its remaining conditions, the summary family and its parameters, the input column, and the grouping `G`. +**Case 2, G7: the same definition, a disjoint selection.** The walk over G6 and G5 lifts `job = "web"` and the same minute, so G7's definition is G4's (`input_group: Some(G1)`, `unlifted_operators: [SummaryAgg[KLL]]`) and its `definition_id()` is the same; its selection is `{ job ∈ {web}, time (−60 s, 0] }`. The two selections are disjoint, so `SummaryMerge(G4, G7)` is a valid candidate (and group), covering `{ job ∈ {api, web}, time (−60 s, 0] }`. -In the worked example, `latency < 100` and `region = 'us'` move into `selection` (below), and `latency * 2 > 10` stays: +**Case 3, G10: `a + b > 10` cannot be lifted.** It spans two columns, so it is no selection. At G8 the walk lifts `job = "api"` and keeps the rest: ```text - ( SummaryEstimate p99 ) - ▲ - │ - [[ SummaryAgg ]] KLL(latency) by job ← filter removed - ▲ - │ - [ Project ] job, region AS r, latency ← unchanged - ▲ - │ - [ Scan t ] predicates: latency * 2 > 10 ← region = 'us' removed +G10.summary_coverage = SummaryCoverage { + definition: SummaryStateDefinition { + input_group: Some(G1), + unlifted_operators: [SummaryAgg[KLL], Filter(a + b > 10)], + }, + selection: [ { job ∈ {api}, time (−60 s, 0] } ], +} ``` -**When two definitions are equal.** Merging (§5.6) requires equal definitions. +Its definition differs from G4's, so G10 cannot merge with G4, even though both select `job = api`. + +**The fields.** + +- `input_group`: the group the definition reads, below everything the walk passed (`None` when the walk lifted a predicate out of a `Scan`). +- `unlifted_operators`: what the state computes over `input_group`, top down from the `SummaryAgg`, each operator with only the conditions that stayed. Each operator's main input is `None`; any other group it reads, such as the group of a scalar subquery in a `Filter`'s condition, is `Some(group)`. +- `selection`: which output rows of that computation the state took, as a union of boxes. +- `definition_id()`: a hash of the definition, the key a summary store files states under together with a concrete selection. + +##### The definition + +The `definition` is what remains of the operators the walk passed, with every condition that moves into `selection` (below) taken out: + +| On the walk | In `unlifted_operators` | +|---|---| +| a `Filter` whose conditions all move into `selection` | removed | +| a `Filter` with some conditions that stay | kept, with only the conditions that stay; a scalar subquery among them names its group | +| `SummaryAgg.filter` | trimmed the same way | +| the one range `TimeRange`, and any `TimeShift`s, when they become relative time | removed | +| a `TimeShift` with no range `TimeRange`, or two or more range `TimeRange`s | unchanged: no time is taken out | +| a direct-column `Project` | unchanged | +| a `Scan` with a predicate that moves into `selection` (a PromQL label matcher) | the last operator, with only the predicates that stay; `input_group` is `None` | +| any other operator (the first group the walk cannot pass) | not in the list: it is the `input_group` | -- They must have the same structure. Planning details are ignored: `guarantee` and `coverage_cache`. Operator nodes carry no timing (#1200), so a tumbling window built at ingestion time can merge with one built at query time. -- `SummaryUpdate.weight_domain` is compared too. It is computed from the rest, so it differs only if something is wrong. -- States over different tables never merge: a KLL over `m1` and one over `m2` have different definitions. To combine tables, put a `UNION ALL` with a column that marks the source table below one `SummaryAgg`; that column can then be used in `selection` or in the grouping. +A `SummaryTumblingWindowChain` covers what one build over its whole window does: the walk starts at that build's `SummaryAgg` over a range `TimeRange` of the whole window over the chain's input. A rollup (`SummaryMerge` with `group_by`) has its input's definition with the `SummaryAgg` grouped by the rollup's keys. + +**When two definitions are equal.** Merging (§5.6) requires equal definitions: the same `input_group` and the same `unlifted_operators`, compared as values. A definition with a scalar subquery compares like any other: two states whose `Filter(a > (SELECT threshold))` read the same group are equal there. Timing is not part of a group, so a tumbling window built at ingestion time can merge with one built at query time. States over different tables never merge: a KLL over `m1` and one over `m2` read different groups. To combine tables, put a `UNION ALL` with a column that marks the source table below one `SummaryAgg`; that column can then be used in `selection` or in the grouping. ##### The selection @@ -307,7 +346,7 @@ In the worked example, `latency < 100` and `region = 'us'` move into `selection` **Steps.** -1. **Collect the conditions.** Go down the sub-DAG from the `SummaryAgg` and collect every filter condition: from `Filter` nodes, from `Scan.predicates`, and from the `SummaryAgg`'s own `filter`. A condition `A AND B` counts as two conditions, `A` and `B`. +1. **Collect the conditions.** Go down the input groups from the `SummaryAgg` and collect every filter condition: from `Filter` groups, from the `Scan.predicates` where the walk stops, and from the `SummaryAgg`'s own `filter`. A condition `A AND B` counts as two conditions, `A` and `B`. 2. **Ask two questions about each condition:** - **Rule 1: would it pick the same rows if it were moved to just below the `SummaryAgg`?** `region = 'us'` below a `Project` that only renames columns: yes. `value > 5` below `rate`: no, because it filters the raw samples that `rate` reads, which changes the rate values. - **Rule 2: is it a simple condition on one column?** That is, a value set such as `region IN ('us', 'eu')`, or a range such as `latency < 100`. `latency * 2 > 10` is not: it is on an expression. @@ -316,7 +355,16 @@ In the worked example, `latency < 100` and `region = 'us'` move into `selection` - If either answer is no: leave it in the sub-DAG, so it is part of `definition`. The paper calls such conditions *residuals*. - One more case stays in `definition`: a condition on a column that already has a lifted condition it cannot be combined with (a forbidden-value set and a range, or values of different types). Conditions are taken in walk order: the `SummaryAgg.filter`, then the `Filter`s from top to bottom, then `Scan.predicates`. -In the worked example: +A SQL example, p99 latency per job over table `t` (columns `job`, `region`, `latency`), whose `WHERE` conditions the SQL front end folds into the `Scan`'s predicates and whose `region AS r` is a `Project`: + +```sql +SELECT job, + approx_percentile_cont(latency, 0.99) FILTER (WHERE latency < 100) +FROM (SELECT job, region AS r, latency + FROM t + WHERE region = 'us' AND latency * 2 > 10) +GROUP BY job; +``` | Condition | Found at | Rule 1: same rows at the `SummaryAgg`? | Rule 2: simple? | Result | |---|---|---|---|---| @@ -403,7 +451,7 @@ Any other shape stays in `definition`: #### 4.2.3 Cost of deriving coverage -**When it runs.** `coverage()` derives a node's coverage the first time it is called and caches it on the node, so each node pays once. Building a `SummaryMerge` (`OperatorNode::new`) derives its coverage to reject an invalid merge and caches the result, so `coverage()` and `validate_structure` do not derive it again. `validate_structure` derives it only for a merge that was not built through `new`. +**When it runs.** `find_or_add_group` derives a summary-state group's coverage once, when it creates the group, and stores it in `GroupProperties::summary_coverage`; `add_candidate` derives a candidate's the same way. A `SummaryMerge` reads its inputs' stored coverage, so each state is derived once. Operator-node trees carry no coverage. **Sizes used below.** @@ -886,9 +934,10 @@ OperatorNode §6.1 │ └── fields[i].dtype: FieldDataType ◀───────┘ §6.3 (the same type: the output │ └── Sketch(SketchKind, GroupingStrategy) `state` field has type `family`) │ └── algorithm, params §6.4 -├── guarantee: AccuracyGuarantee §6.5 (accuracy of an evaluation) -└── coverage() → SummaryCoverage §6.6 - └── definition: OperatorNode (again a node, so the same structure) +└── guarantee: AccuracyGuarantee §6.5 (accuracy of an evaluation) + +MemoGroup.group_properties.summary_coverage: SummaryCoverage §6.6 + └── definition: SummaryStateDefinition { input_group, unlifted_operators } ``` | Subsection | Defines | Used by | @@ -898,7 +947,7 @@ OperatorNode §6.1 | §6.3 Schema and field types | `Schema`, `Field`, `FieldDataType` | every node's `schema`; `SummaryAgg.family` | | §6.4 Summary-family parameters | `SketchKind`, `SketchParams`, `GroupingStrategy`, exact/sample/wavelet/model params | the payload of each non-plain `FieldDataType` variant | | §6.5 Update input and evaluations | `SummaryUpdate`, `SketchStatistic`, `MaintainedPopulation`, `PopulationStatistic` | fields of the `ASAPOp` variants in §6.2 | -| §6.6 Summary coverage | `SummaryCoverage`, `SelectionBox`, `Constraint` | `OperatorNode::coverage()` on summary nodes | +| §6.6 Summary coverage | `SummaryCoverage`, `SummaryStateDefinition`, `SelectionBox`, `Constraint` | `GroupProperties::summary_coverage` of summary-state memo groups | **Example: the p99 latency state of §5.1.** The `SummaryAgg` node (§6.1) holds `ASAPOp::SummaryAgg` (§6.2) with `family = Sketch(SketchKind::new(Kll, Kll { k: 200 }), PerSubpopulationInstance)` (§6.3, §6.4) and `input = SummaryUpdate::column(latency)` (§6.5). Its `schema` (§6.3) is `(job Utf8, state Sketch(KLL k=200))`: the `state` field has the same type as `family`. A `SummaryEstimate` above it holds `query = Quantile { q: 0.99 }` (§6.5), and its node carries the `guarantee` of the evaluation. `coverage()` on the `SummaryAgg` node returns a `SummaryCoverage` (§6.6). @@ -1342,16 +1391,16 @@ A finalized value's accuracy statement is `AccuracyGuarantee` (`ir/properties/gu ### 6.6 Summary coverage (`crates/types/src/ir/properties/summary_coverage.rs`) -The code for §4, as of #646. On `main` today `SummaryCoverage` is still the older declared form (`source` plus `regions`). +The code for §4. `SummaryStateDefinition` is shown in §4.2.2. ```rust pub struct SummaryCoverage { - /// What the state computes: the `SummaryAgg` and its sub-DAG, with the - /// conditions that went into `selection` taken out (§4.2.2). - pub definition: Rc, + /// What the state computes: its input group and the operators the walk + /// did not lift (§4.2.2). + pub(crate) definition: SummaryStateDefinition, /// Which rows went in: a union of boxes. A row is in the state if it is /// in at least one box. - pub selection: Vec, + pub(crate) selection: Vec, } /// One box: a row is in it when it meets every constraint. @@ -1386,9 +1435,10 @@ pub enum Constraint { } impl SummaryCoverage { - /// Computes the coverage of a summary node from its sub-DAG (§4.2.2). - /// Errors: `NotSummary` for a node that is not a `SummaryAgg`, - /// `SummaryTumblingWindowChain` or `SummaryMerge`; `TumblingWindowChainExpansion` for a `SummaryTumblingWindowChain` + /// The coverage of a summary-state group computed by `operator` over its + /// input groups (§4.2.2). Errors: `NotSummary` for an operator that is + /// not a `SummaryAgg`, `SummaryTumblingWindowChain`, `SummaryMerge` or a + /// `Filter` keeping whole states; `TumblingWindowChainExpansion` for a `SummaryTumblingWindowChain` /// that does not expand to one build over its whole window; for a /// merge, an input's own error, `EmptyMerge` (no inputs), /// `DefinitionMismatch` (inputs compute different things) or @@ -1398,9 +1448,12 @@ impl SummaryCoverage { /// not a group key of the inputs' build), `FamilyCannotRegroup` (counter /// 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; + pub(crate) fn derive_from_inputs( + operator: &Operator>, + ) -> Result; + pub(crate) fn definition_id(&self) -> String; } ``` -- `OperatorNode::new` and `validate_structure` reject an invalid `SummaryMerge` (different definitions, or selections that may overlap; #592 relaxes the overlap check per family). The derived coverage is cached on the node. +- `find_or_add_group` and `add_candidate` reject an invalid `SummaryMerge` (different definitions, or selections that may overlap, per the family's `MergeInputRule`). Operator-node trees have no coverage and no merge check. - A `SummaryAgg` always has coverage: what cannot go into `selection` stays in `definition`.