diff --git a/crates/integration-tests/tests/sketch_rollup.rs b/crates/integration-tests/tests/sketch_rollup.rs index a5044bd34..6f219b8a6 100644 --- a/crates/integration-tests/tests/sketch_rollup.rs +++ b/crates/integration-tests/tests/sketch_rollup.rs @@ -10,6 +10,7 @@ use std::collections::BTreeMap; use std::rc::Rc; use std::sync::Arc; +use asap_accuracy::sizing::with_summaries_sized_for_failure_probability; use asap_executor::{ physical_planner::compile, runtime::{Limits, RunContext, Scope}, @@ -25,7 +26,8 @@ use asap_types::ir::properties::{AccuracyMetric, AccuracyTarget}; use asap_types::ir::schema::{ DataType, ExactKind, ExactParams, Field, FieldDataType, Schema, SketchAlgorithm, }; -use asap_types::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode}; +use asap_types::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode, QueryRoot}; +use asap_types::workload::DataWorkload; use futures::StreamExt; const REGIONS: [&str; 3] = ["us", "eu", "ap"]; @@ -224,6 +226,10 @@ fn rolled_up(coarse: &OperatorNode) -> Rc { /// rollup from `(region, job)`: the family, the direct root and the /// rolled-up root. Explicit rule fixtures keep this execution test independent /// of the search archive and its tighter, potentially much larger capacities. +/// Each candidate's summaries are sized for `accuracy`'s failure +/// probability, as Pass 1 sizes a memo candidate +/// ([`with_summaries_sized_for_failure_probability`]); composition alone +/// sizes them for the calibrated one. async fn direct_and_rolled_up( sql: &str, accuracy: AccuracyTarget, @@ -231,6 +237,21 @@ async fn direct_and_rolled_up( let root = lower_sql(sql, &catalog(), accuracy.clone()).await.unwrap(); let mut out = Vec::new(); for candidate in explicit_local_candidates(root) { + let candidate = match accuracy.failure_probability { + Some(failure_probability) => { + match &with_summaries_sized_for_failure_probability( + &[QueryRoot::Operator(candidate)], + failure_probability, + &DataWorkload::default(), + ) + .unwrap()[..] + { + [QueryRoot::Operator(sized)] => Rc::clone(sized), + _ => unreachable!("one operator root in, one out"), + } + } + None => candidate, + }; let Some(build) = build(&candidate) else { continue; }; @@ -279,14 +300,18 @@ async fn distinct_counts_rolled_up_equal_the_direct_build() { /// p99 latency per job. DDSketch's bucket counts add, so the merged state /// is the direct build's and so is its answer. KLL's compactions are -/// randomized, so its rolled-up answer differs, but stays within its rank -/// bound of the exact p99, as the direct build's does. +/// randomized (the executor's KLL seeds its coin per run), so its rolled-up +/// answer differs, but stays within its rank bound of the exact p99, as the +/// direct build's does. Each answer may miss the bound with the sizing's +/// failure probability: at the calibrated 0.01 (`k = 52`) each of the eight +/// KLL answers checked here may miss it with probability up to 0.01, so the +/// KLL is sized for 1e-6 (`k = 100`). #[tokio::test] async fn quantiles_rolled_up_keep_the_sketch_bound() { const EPSILON: f64 = 0.05; for metric in [AccuracyMetric::Rank, AccuracyMetric::Relative] { let accuracy = AccuracyTarget::max_error(EPSILON) - .failure_probability(0.01) + .failure_probability(1e-6) .accuracy_metric(metric); let pairs = direct_and_rolled_up( "SELECT job, approx_percentile_cont(latency, 0.99) AS value FROM http GROUP BY job", diff --git a/crates/logical-asap-optimization/src/lib.rs b/crates/logical-asap-optimization/src/lib.rs index 2a6c14468..bc887f1d2 100644 --- a/crates/logical-asap-optimization/src/lib.rs +++ b/crates/logical-asap-optimization/src/lib.rs @@ -64,6 +64,7 @@ mod added_groups; /// The deployment's capabilities: supported summaries and what it retains. mod deployment_capabilities; mod find_or_add_group; +mod operator_over_groups; mod own_candidate_sizing; mod pass1; mod pass2; diff --git a/crates/logical-asap-optimization/src/operator_over_groups.rs b/crates/logical-asap-optimization/src/operator_over_groups.rs new file mode 100644 index 000000000..dd5ea61a0 --- /dev/null +++ b/crates/logical-asap-optimization/src/operator_over_groups.rs @@ -0,0 +1,95 @@ +//! One operator over groups (#1203): a candidate a rule composes or +//! rewrites as a tree of operators is stored as its root operator alone, +//! every node beneath it found or added as a memo group +//! ([`over_child_groups`], Stage 0's node equality). Pass 1 and the +//! Pass 2 rules store their candidates through [`operator_over_groups`]. + +use std::collections::HashMap; +use std::rc::Rc; + +use asap_types::ir::OperatorNode; +use asap_types::stages::{Memo, MemoGroupId, OriginOfCandidateForMemoGroup}; + +use crate::added_groups::groups_read_through_added_groups; +use crate::find_or_add_group::over_child_groups; +use crate::own_candidate_sizing::size_own_candidate_for; + +/// `root`, a candidate of a group `queries` reach, as one operator over +/// groups: the exact guarantees below it removed +/// ([`without_exact_guarantees_below`]), each child found or added with +/// every node below it ([`over_child_groups`]: an added group is made +/// by `origin`, sized for `accuracy_budget_parts`), and every added group it +/// reads through sized for at least `accuracy_budget_parts` (a reader sized +/// for more parts may find a group added for fewer). Returns the operator +/// over the groups' nodes and those groups by child position. +pub(crate) fn operator_over_groups( + memo: &mut Memo, + root: &Rc, + queries: &[usize], + origin: &OriginOfCandidateForMemoGroup, + accuracy_budget_parts: usize, +) -> (Rc, Vec) { + let root = without_exact_guarantees_below(root); + let (node, children) = over_child_groups(memo, &root, queries, origin, accuracy_budget_parts); + for group in groups_read_through_added_groups(memo, &children).added_groups { + size_own_candidate_for(memo, group, accuracy_budget_parts); + } + (node, children) +} + +/// `root` with the exact guarantee removed from every node below it. Stage +/// 0's nodes carry no guarantee, and composition gives each exact node it +/// rebuilds the exact one ("exact over exact inputs"); that difference alone +/// must not split a composed node from the equal Stage 0 group +/// ([`find_or_add_group`](crate::find_or_add_group::find_or_add_group) +/// compares guarantees). An approximate +/// guarantee is kept, so builds that differ only in their failure +/// probability or guarantee stay different groups. `root`'s own guarantee +/// is the candidate's and is kept. +pub(crate) fn without_exact_guarantees_below(root: &Rc) -> Rc { + fn without_exact( + node: &Rc, + done: &mut HashMap<*const OperatorNode, Rc>, + ) -> Rc { + if let Some(found) = done.get(&Rc::as_ptr(node)) { + return Rc::clone(found); + } + let operator = node + .operator + .map_children(|child| without_exact(child, done)); + let exact = node.guarantee.as_ref().is_some_and(|g| g.is_exact()); + let changed = exact + || node + .children() + .iter() + .zip(operator.children()) + .any(|(a, b)| !Rc::ptr_eq(a, b)); + let rebuilt = if changed { + Rc::new(OperatorNode { + operator, + guarantee: if exact { None } else { node.guarantee.clone() }, + ..(**node).clone() + }) + } else { + Rc::clone(node) + }; + done.insert(Rc::as_ptr(node), Rc::clone(&rebuilt)); + rebuilt + } + let mut done = HashMap::new(); + let operator = root + .operator + .map_children(|child| without_exact(child, &mut done)); + if root + .children() + .iter() + .zip(operator.children()) + .all(|(a, b)| Rc::ptr_eq(a, b)) + { + return Rc::clone(root); + } + Rc::new(OperatorNode { + operator, + ..(**root).clone() + }) +} diff --git a/crates/logical-asap-optimization/src/pass1/add_pass1_candidates.rs b/crates/logical-asap-optimization/src/pass1/add_pass1_candidates.rs index 362ace932..46a4285f5 100644 --- a/crates/logical-asap-optimization/src/pass1/add_pass1_candidates.rs +++ b/crates/logical-asap-optimization/src/pass1/add_pass1_candidates.rs @@ -37,16 +37,15 @@ //! the resulting error contract. The connection planner may select a conservative //! larger configuration, and validates the recomposed query-root guarantee. +use asap_types::stages::hash_consing::equal_nodes; use asap_types::stages::HowToComputeAggregateMeasure; -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; use std::rc::Rc; use super::per_computation_candidates::{ compose_logical_candidate_with_accuracy_budget_parts, PerComputationCandidates, }; -use crate::added_groups::groups_read_through_added_groups; -use crate::find_or_add_group::over_child_groups; -use crate::own_candidate_sizing::size_own_candidate_for; +use crate::operator_over_groups::operator_over_groups; use asap_accuracy::sizing::{ per_input_failure_probability, with_summaries_sized_for_failure_probability, }; @@ -172,32 +171,29 @@ fn add_candidates_to_initial_groups( most_approximate_inputs[q] += summaries; } } - let mut structural_ids = NodeToStructuralIdCache::default(); for g in 0..initial_groups { - let candidates = match targets_of.get(&MemoGroupId(g as u32)) { - None => None, - Some(targets) => { - let max_accuracy_budget_parts = memo.groups()[g] - .queries - .iter() - .map(|&q| most_approximate_inputs[q]) - .max() - .unwrap_or(1) - .max(1); - target_candidates( - per_computation, - accuracy_targets, - data, - memo, - &mut structural_ids, - &logical_nodes, - MemoGroupId(g as u32), - targets, - max_accuracy_budget_parts, - )? - } - }; - memo.candidates[g] = candidates.unwrap_or_else(|| vec![own(&memo.groups()[g])]); + if let Some(targets) = targets_of.get(&MemoGroupId(g as u32)) { + let max_accuracy_budget_parts = memo.groups()[g] + .queries + .iter() + .map(|&q| most_approximate_inputs[q]) + .max() + .unwrap_or(1) + .max(1); + target_candidates( + per_computation, + accuracy_targets, + data, + memo, + &logical_nodes, + MemoGroupId(g as u32), + targets, + max_accuracy_budget_parts, + )?; + } + if memo.candidates[g].is_empty() { + memo.candidates[g] = vec![own(&memo.groups()[g])]; + } } memo.rebuild_parents(); Ok(()) @@ -208,27 +204,21 @@ fn add_candidates_to_initial_groups( /// target varying fastest), each the aggregate composed alone with that /// choice, its summaries sized for its queries' failure probability (query /// `q`'s target `accuracy_targets[q]`, over `data`); the groups beneath it -/// are found or added beneath Stage 0's groups. `None` when none composes. +/// are found or added ([`operator_over_groups`]), each added to `g` +/// ([`Memo::add_candidate`]). None is added when none composes. #[allow(clippy::too_many_arguments)] fn target_candidates( per_computation: &PerComputationCandidates, accuracy_targets: &[AccuracyTarget], data: &DataWorkload, memo: &mut Memo, - structural_ids: &mut NodeToStructuralIdCache, logical_nodes: &Rc<[Rc]>, g: MemoGroupId, targets: &[usize], max_accuracy_budget_parts: usize, -) -> Result>, SchemaDerivationError> { +) -> Result<(), SchemaDerivationError> { let aggregate = Rc::clone(&memo.groups()[g.0 as usize].logical_node); let queries = memo.groups()[g.0 as usize].queries.clone(); - let own_key = structural_ids.structural_id(&aggregate); - let mut candidates = Vec::new(); - // Each candidate's identity in the memo: its node's structural id, its - // input groups, the groups it fuses and its accuracy budget parts. A candidate - // equal to one already listed is not added again. - let mut known = HashSet::new(); let mut choice = vec![0; per_computation.targets.len()]; loop { let mut implements = vec![g]; @@ -260,67 +250,63 @@ fn target_candidates( if accuracy_budget_parts > 1 && !builds_a_sketch(&root) { break; } - let (node, children) = if structural_ids.structural_id(&root) == own_key { + let root = size_for_failure_probability( + &root, + &queries, + accuracy_budget_parts, + &per_computation.roots, + accuracy_targets, + data, + )?; + let (node, children) = operator_over_groups( + memo, + &root, + &queries, + &OriginOfCandidateForMemoGroup::Pass1, + accuracy_budget_parts, + ); + // A parent reads the candidate in the group's place, so it + // outputs the group's schema when it has the same columns, with + // the group's metadata (composition may derive a closed label set + // where the frontend's node leaves it open). A query's top-k root + // may output other columns (the ranked value beside the key): it + // keeps its own. + let same_columns = node.schema.fields == aggregate.schema.fields + && node.schema.time_index == aggregate.schema.time_index; + let node = if !same_columns || node.schema == aggregate.schema { + node + } else { + Rc::new(OperatorNode { + schema: aggregate.schema.clone(), + ..(*node).clone() + }) + }; + // The exact composition of the group's own node is the group's + // node itself (Stage 0's equality, its exact guarantee aside). + let own = node.guarantee.as_ref().is_none_or(|g| g.is_exact()) + && equal_nodes( + &OperatorNode { + guarantee: None, + ..(*node).clone() + }, + &aggregate, + ); + let (node, children) = if own { ( Rc::clone(&aggregate), memo.groups()[g.0 as usize].child_groups.clone(), ) } else { - let root = size_for_failure_probability( - &root, - &queries, - accuracy_budget_parts, - &per_computation.roots, - accuracy_targets, - data, - )?; - let root = without_exact_guarantees_below(&root); - let (node, children) = over_child_groups( - memo, - &root, - &queries, - &OriginOfCandidateForMemoGroup::Pass1, - accuracy_budget_parts, - ); - // A consumer sized for more parts may find a group added for - // fewer. - for group in groups_read_through_added_groups(memo, &children).added_groups { - size_own_candidate_for(memo, group, accuracy_budget_parts); - } - // A parent reads the candidate in the group's place, so it - // outputs the group's schema when it has the same columns, - // with the group's metadata (composition may derive a closed - // label set where the frontend's node leaves it open). A - // query's top-k root may output other columns (the ranked - // value beside the key): it keeps its own. - let group_node = &aggregate; - let same_columns = node.schema.fields == group_node.schema.fields - && node.schema.time_index == group_node.schema.time_index; - let node = if !same_columns || node.schema == group_node.schema { - node - } else { - Rc::new(OperatorNode { - schema: group_node.schema.clone(), - ..(*node).clone() - }) - }; (node, children) }; - if !known.insert(( - structural_ids.structural_id(&node), - children.clone(), - implements[1..].to_vec(), - accuracy_budget_parts, - )) { - continue; - } - candidates.push(CandidateForMemoGroup { + let candidate = CandidateForMemoGroup { fused_groups: implements[1..].to_vec(), root_node: node, input_groups: children, origin: OriginOfCandidateForMemoGroup::Pass1, accuracy_budget_parts, - }); + }; + memo.add_candidate(g, candidate); } // Mixed-radix over this group's targets only. let mut advanced = false; @@ -336,7 +322,7 @@ fn target_candidates( break; } } - Ok((!candidates.is_empty()).then_some(candidates)) + Ok(()) } /// `root`, a candidate of a group `queries` reach, with each approximate @@ -444,62 +430,6 @@ fn compose_alone( ) } -/// `root` with the exact guarantee removed from every node below it. Stage -/// 0's nodes carry no guarantee, and composition gives each exact node it -/// rebuilds the exact one ("exact over exact inputs"); that difference alone -/// must not split a composed node from the equal Stage 0 group -/// ([`find_or_add_group`](crate::find_or_add_group::find_or_add_group) compares guarantees). An approximate -/// guarantee is kept, so builds that differ only in their failure -/// probability or guarantee stay different groups. `root`'s own guarantee -/// is the candidate's and is kept. -fn without_exact_guarantees_below(root: &Rc) -> Rc { - fn without_exact( - node: &Rc, - done: &mut HashMap<*const OperatorNode, Rc>, - ) -> Rc { - if let Some(found) = done.get(&Rc::as_ptr(node)) { - return Rc::clone(found); - } - let operator = node - .operator - .map_children(|child| without_exact(child, done)); - let exact = node.guarantee.as_ref().is_some_and(|g| g.is_exact()); - let changed = exact - || node - .children() - .iter() - .zip(operator.children()) - .any(|(a, b)| !Rc::ptr_eq(a, b)); - let rebuilt = if changed { - Rc::new(OperatorNode { - operator, - guarantee: if exact { None } else { node.guarantee.clone() }, - ..(**node).clone() - }) - } else { - Rc::clone(node) - }; - done.insert(Rc::as_ptr(node), Rc::clone(&rebuilt)); - rebuilt - } - let mut done = HashMap::new(); - let operator = root - .operator - .map_children(|child| without_exact(child, &mut done)); - if root - .children() - .iter() - .zip(operator.children()) - .all(|(a, b)| Rc::ptr_eq(a, b)) - { - return Rc::clone(root); - } - Rc::new(OperatorNode { - operator, - ..(**root).clone() - }) -} - /// Whether `node`'s private nodes build a sketch, which its accuracy budget parts size. fn builds_a_sketch(node: &Rc) -> bool { OperatorNode::reachable(node).iter().any(|n| { @@ -518,86 +448,11 @@ fn builds_a_sketch(node: &Rc) -> bool { }) } -/// Interned structural identities: the operator over its children's identities, -/// without the schema, guarantee, timing and materialization composition -/// derives. Memoized per node (the cache keeps each node alive, so its -/// address is not reused). -#[derive(Default)] -pub(crate) struct NodeToStructuralIdCache { - id_of_node: HashMap<*const OperatorNode, (Rc, u64)>, - id_of_encoded_operator: HashMap, u64>, -} - -impl NodeToStructuralIdCache { - pub(crate) fn structural_id(&mut self, node: &Rc) -> u64 { - if let Some((_, key)) = self.id_of_node.get(&Rc::as_ptr(node)) { - return *key; - } - let children: HashMap<*const OperatorNode, u64> = node - .children() - .into_iter() - .map(|child| (Rc::as_ptr(child), self.structural_id(child))) - .collect(); - let operator = node.operator.map_children(|c| children[&Rc::as_ptr(c)]); - // MessagePack preserves non-finite floats and signed zero. HashMap - // compares the complete byte key, so a hash collision cannot merge - // different computations. Cloned/recomposed NaN literals retain identity. - let encoded = - rmp_serde::to_vec(&operator).expect("an IR operator over child IDs must serialize"); - let next = self.id_of_encoded_operator.len() as u64; - let key = *self.id_of_encoded_operator.entry(encoded).or_insert(next); - self.id_of_node - .insert(Rc::as_ptr(node), (Rc::clone(node), key)); - key - } -} - #[cfg(test)] mod tests { use super::*; use crate::stage0::generate_dag_of_logical_memo_groups; - #[test] - fn structural_keys_preserve_nonfinite_and_signed_zero_literals() { - let input = crate::test_support::lower_promql( - "m", - asap_types::ir::properties::AccuracyTarget::exact(), - ); - let literal = |value| { - OperatorNode::new_shared(asap_types::ir::Operator::NonASAP( - asap_types::ir::NonASAPOp::Project { - cols: vec![asap_types::ir::ProjectItem { - alias: Some("value".into()), - expr: asap_types::ir::ScalarExpr::literal_f64(value), - }], - qualifier: None, - child: Rc::clone(&input), - }, - )) - .unwrap() - }; - let mut structural_ids = NodeToStructuralIdCache::default(); - for (left, right) in [ - (f64::NAN, f64::INFINITY), - (f64::INFINITY, f64::NEG_INFINITY), - (0.0, -0.0), - ] { - assert_ne!( - structural_ids.structural_id(&literal(left)), - structural_ids.structural_id(&literal(right)) - ); - } - assert_eq!( - structural_ids.structural_id(&literal(1.0)), - structural_ids.structural_id(&literal(1.0)) - ); - let node = literal(f64::NAN); - assert_eq!( - structural_ids.structural_id(&node), - structural_ids.structural_id(&literal(f64::NAN)) - ); - } - fn populate_pass1( initial: DagOfLogicalMemoGroups, inventory: &PerComputationCandidates, @@ -616,8 +471,8 @@ mod tests { use asap_types::ir::properties::AccuracyTarget; use asap_types::ir::ASAPOp; + use std::collections::HashSet; - /// The cross-stage API rejects mismatched root identities without panicking. #[test] fn population_rejects_an_inventory_from_other_roots() { let roots = vec![QueryRoot::Operator(lower_promql( @@ -738,7 +593,8 @@ mod tests { /// The Stage 0 groups `candidate` reads, directly or through the /// groups Pass 1 added beneath them, in the order first reached. fn stage0_inputs(memo: &Memo, candidate: &CandidateForMemoGroup) -> Vec { - groups_read_through_added_groups(memo, &candidate.input_groups).stage0_groups + crate::added_groups::groups_read_through_added_groups(memo, &candidate.input_groups) + .stage0_groups } /// The summary algorithms a candidate's private nodes build. @@ -833,7 +689,7 @@ mod tests { }, )) .unwrap(); - let stripped = without_exact_guarantees_below(&reader); + let stripped = crate::operator_over_groups::without_exact_guarantees_below(&reader); assert_eq!(stripped.children()[0].guarantee, Some(guarantee)); } diff --git a/crates/logical-asap-optimization/src/pass1/per_computation_candidates.rs b/crates/logical-asap-optimization/src/pass1/per_computation_candidates.rs index 307ae1d89..81d764865 100644 --- a/crates/logical-asap-optimization/src/pass1/per_computation_candidates.rs +++ b/crates/logical-asap-optimization/src/pass1/per_computation_candidates.rs @@ -219,12 +219,11 @@ pub struct PerComputationCandidates { impl PerComputationCandidates { /// Whether this inventory walked `dag`, or a DAG built the same way from - /// the same roots: the same groups (by structural id, in the same order) + /// the same roots: the same groups (equal by Stage 0's equality, in the + /// same order) /// and each query's root over the same groups. Its targets' group ids /// then name `dag`'s groups. pub(crate) fn walked(&self, dag: &DagOfLogicalMemoGroups) -> bool { - use super::add_pass1_candidates::NodeToStructuralIdCache; - let mut structural_ids = NodeToStructuralIdCache::default(); let group_of: HashMap<*const OperatorNode, MemoGroupId> = self .logical_nodes .iter() @@ -239,8 +238,7 @@ impl PerComputationCandidates { .zip(self.logical_nodes.iter()) .all(|(group, node)| { Rc::ptr_eq(&group.logical_node, node) - || structural_ids.structural_id(&group.logical_node) - == structural_ids.structural_id(node) + || asap_types::stages::hash_consing::equal_nodes(&group.logical_node, node) }) && dag .query_roots diff --git a/crates/logical-asap-optimization/src/pass2/memo/rules/reuse.rs b/crates/logical-asap-optimization/src/pass2/memo/rules/reuse.rs index 15f8c6c32..9f3f333af 100644 --- a/crates/logical-asap-optimization/src/pass2/memo/rules/reuse.rs +++ b/crates/logical-asap-optimization/src/pass2/memo/rules/reuse.rs @@ -1,5 +1,7 @@ //! Shared boundary handling for producer-level reuse rules. Rule outputs retain //! canonical input identities after guarantee recomposition rebuilds their DAGs. +use crate::find_or_add_group::find_or_add_group; +use crate::operator_over_groups::{operator_over_groups, without_exact_guarantees_below}; use crate::pass2::memo::state_groups::{boundary_groups, Pass2RuleError}; use asap_types::ir::OperatorNode; use asap_types::stages::{CandidateForMemoGroup, Memo, OriginOfCandidateForMemoGroup}; @@ -113,13 +115,7 @@ pub(in crate::pass2::memo) fn append_producer( // The new group's id must fit a `MemoGroupId`. u32::try_from(memo.groups().len()).map_err(|_| Pass2RuleError::TooManyGroups)?; let node = Rc::new((*node).clone()); - let boundaries = memo - .groups() - .iter() - .enumerate() - .map(|(id, group)| (Rc::as_ptr(&group.logical_node), MemoGroupId(id as u32))) - .collect(); - let children = boundary_groups(&node, &boundaries); + let children = boundary_groups(memo, &node, |_| true); Ok(memo.push_group( MemoGroup { logical_node: Rc::clone(&node), @@ -137,6 +133,92 @@ pub(in crate::pass2::memo) fn append_producer( )) } +/// Whether `actual` has `expected`'s output: the same result kind, fields, +/// time index and unique keys, and a closed label set where `expected`'s is +/// closed (a projection may enumerate every output of an open schema). +pub(in crate::pass2::memo) fn same_output_contract( + expected: &OperatorNode, + actual: &OperatorNode, +) -> bool { + expected.result_kind == actual.result_kind + && expected.schema.fields == actual.schema.fields + && expected.schema.time_index == actual.schema.time_index + && expected.schema.unique_keys == actual.schema.unique_keys + && (!expected.schema.closed || actual.schema.closed) +} + +/// Rejects `node`, `original` rewritten by `origin`, when its output differs +/// from `original`'s ([`same_output_contract`]). +fn check_output_contract( + original: &CandidateForMemoGroup, + node: &OperatorNode, + origin: &OriginOfCandidateForMemoGroup, +) -> Result<(), Pass2RuleError> { + if !same_output_contract(&original.root_node, node) { + return Err(Pass2RuleError::ReuseRule(format!( + "reuse changed the reader output contract ({origin:?}): expected {:?}, got {:?}", + original.root_node.schema, node.schema + ))); + } + Ok(()) +} + +/// Adds to group `owner` the candidate `original` rewritten by `origin` to +/// `node`, as one operator over groups ([`operator_over_groups`]: every node +/// beneath `node` is found or added as a group made by `origin`, sized for +/// `original`'s budget parts). `false` when `owner` has an equal candidate +/// already ([`Memo::add_candidate`]), or when `node` reads `owner` +/// itself. +pub(in crate::pass2::memo) fn add_candidate_over_groups( + memo: &mut Memo, + owner: MemoGroupId, + original: &CandidateForMemoGroup, + node: Rc, + origin: OriginOfCandidateForMemoGroup, +) -> Result { + check_output_contract(original, &node, &origin)?; + let queries = memo.groups()[owner.0 as usize].queries.clone(); + let (root_node, input_groups) = operator_over_groups( + memo, + &node, + &queries, + &origin, + original.accuracy_budget_parts, + ); + if input_groups.contains(&owner) { + return Ok(false); + } + let mut candidate = original.clone(); + candidate.root_node = root_node; + candidate.input_groups = input_groups; + candidate.origin = origin; + let known = memo.candidates[owner.0 as usize].len(); + Ok(memo.add_candidate(owner, candidate).index == known) +} + +/// The group of `node`, a producer a rule factors out of its readers, made +/// by `origin` when no group's node equals it ([`find_or_add_group`]), its +/// exact guarantees removed first so it equals the groups of the exact nodes +/// it repeats. A producer holds no approximate state of its own, so its +/// candidate is sized for the whole budget (1 part). +pub(in crate::pass2::memo) fn find_or_add_producer( + memo: &mut Memo, + node: &Rc, + queries: &[usize], + origin: &OriginOfCandidateForMemoGroup, +) -> MemoGroupId { + let below = without_exact_guarantees_below(node); + let node = if below.guarantee.as_ref().is_some_and(|g| g.is_exact()) { + Rc::new(OperatorNode { + guarantee: None, + ..(*below).clone() + }) + } else { + below + }; + find_or_add_group(memo, &node, queries, origin, 1) +} + pub(in crate::pass2::memo) fn append_reader( memo: &mut Memo, owner: usize, @@ -144,25 +226,8 @@ pub(in crate::pass2::memo) fn append_reader( node: Rc, origin: OriginOfCandidateForMemoGroup, ) -> Result { - if original.root_node.schema.fields != node.schema.fields - || original.root_node.schema.time_index != node.schema.time_index - || original.root_node.schema.unique_keys != node.schema.unique_keys - || (original.root_node.schema.closed && !node.schema.closed) - || original.root_node.result_kind != node.result_kind - { - return Err(Pass2RuleError::ReuseRule(format!( - "reuse changed the reader output contract ({origin:?}): expected {:?}, got {:?}", - original.root_node.schema, node.schema - ))); - } - let boundaries = memo - .groups() - .iter() - .enumerate() - .filter(|(id, _)| *id != owner) - .map(|(id, group)| (Rc::as_ptr(&group.logical_node), MemoGroupId(id as u32))) - .collect(); - let children = boundary_groups(&node, &boundaries); + check_output_contract(original, &node, &origin)?; + let children = boundary_groups(memo, &node, |group| group.0 as usize != owner); if memo.candidates[owner].iter().any(|known| { known.input_groups == children && known.accuracy_budget_parts == original.accuracy_budget_parts diff --git a/crates/logical-asap-optimization/src/pass2/memo/rules/scan_labels.rs b/crates/logical-asap-optimization/src/pass2/memo/rules/scan_labels.rs index 1aba06944..4fdb5319d 100644 --- a/crates/logical-asap-optimization/src/pass2/memo/rules/scan_labels.rs +++ b/crates/logical-asap-optimization/src/pass2/memo/rules/scan_labels.rs @@ -7,6 +7,7 @@ use asap_accuracy::propagation::recompose_guarantees; use asap_types::ir::operator::Reduction; use asap_types::ir::{ASAPOp, NonASAPOp, OperatorNode, QueryRoot}; +use super::reuse::same_output_contract; use crate::added_groups::groups_read_through_added_groups; use crate::pass2::memo::state_groups::{ boundary_groups, build_down_to_stage0_groups, Pass2RuleError, @@ -116,15 +117,9 @@ pub(crate) fn add_scan_label_candidates(memo: &mut Memo) -> Result Result bool { - expected.result_kind == actual.result_kind - && expected.schema.fields == actual.schema.fields - && expected.schema.time_index == actual.schema.time_index - && expected.schema.unique_keys == actual.schema.unique_keys - // A projection may enumerate every output of an originally open schema. - && (!expected.schema.closed || actual.schema.closed) -} - fn is_source_chain(node: &OperatorNode) -> bool { match node.non_asap() { Some(NonASAPOp::Scan { .. }) => true, diff --git a/crates/logical-asap-optimization/src/pass2/memo/rules/shared_configuration_readers.rs b/crates/logical-asap-optimization/src/pass2/memo/rules/shared_configuration_readers.rs index 7746fdd32..abe49ab85 100644 --- a/crates/logical-asap-optimization/src/pass2/memo/rules/shared_configuration_readers.rs +++ b/crates/logical-asap-optimization/src/pass2/memo/rules/shared_configuration_readers.rs @@ -20,7 +20,7 @@ use crate::pass2::memo::shared_configurations::{ use crate::pass2::memo::state_groups::{ boundary_ptrs, build_of, same_range_input, BuildPlace, Pass2RuleError, }; -use asap_types::stages::{Memo, OriginOfCandidateForMemoGroup}; +use asap_types::stages::{Memo, MemoGroupId, OriginOfCandidateForMemoGroup}; /// `resized` (a rewrite of `original` with the same shape) with each node /// that is a child group's node in `original` put back, and the nodes above @@ -99,8 +99,6 @@ enum StateCovering { /// later ([`super::super::state_groups`]). struct SummaryStates { deduplicate_requests: bool, - /// Every candidate's graph signature by group, so no rewrite repeats one. - known: HashMap<(usize, String), Vec>>, rewrite_count: usize, /// The (normalized layout, configuration index) rewrites already made. requests: HashSet<(usize, usize)>, @@ -116,7 +114,6 @@ fn expand_configurations( ) -> Result { let mut states = SummaryStates { deduplicate_requests, - known: HashMap::new(), rewrite_count: 0, requests: HashSet::new(), }; @@ -190,14 +187,6 @@ impl SharedProducer for SummaryStates { } } } - for (group, candidates) in memo.candidates.iter().enumerate() { - for candidate in candidates { - self.known - .entry((group, graph_signature(&candidate.root_node)?)) - .or_default() - .push(Rc::clone(&candidate.root_node)); - } - } let mut layouts: Vec<(usize, usize, FlatDag)> = Vec::new(); for set in &mut sets { if spans_groups(set) { @@ -377,29 +366,24 @@ impl SharedProducer for SummaryStates { } }; // Resizing affects internal state types, never public output fields. - if original.root_node.schema.fields != node.schema.fields - || original.root_node.schema.time_index != node.schema.time_index - || original.root_node.schema.unique_keys != node.schema.unique_keys - || (original.root_node.schema.closed && !node.schema.closed) - { + if !super::reuse::same_output_contract(&original.root_node, &node) { return Err(Pass2RuleError::StateConfiguration( "state rewrite changed the reader output contract".into(), )); } - let previous = self - .known - .entry((reader.group, graph_signature(&node)?)) - .or_default(); - if previous.iter().any(|known| **known == *node) { + // A rewrite never reads the group it implements. + let owner = MemoGroupId(reader.group as u32); + if input_groups.contains(&owner) { return Ok(0); } - previous.push(Rc::clone(&node)); let mut candidate = original; candidate.root_node = node; candidate.input_groups = input_groups; candidate.accuracy_budget_parts = accuracy_budget_parts; - memo.candidates[reader.group].push(candidate); - Ok(1) + let known = memo.candidates[owner.0 as usize].len(); + Ok(usize::from( + memo.add_candidate(owner, candidate).index == known, + )) } } @@ -461,6 +445,7 @@ fn assign_normalized_layouts( Ok(()) } +#[cfg(test)] fn graph_signature(node: &Rc) -> Result { serde_json::to_string(&flatten(&[QueryRoot::Operator(Rc::clone(node))]).0) .map_err(|error| Pass2RuleError::StateConfiguration(error.to_string())) diff --git a/crates/logical-asap-optimization/src/pass2/memo/rules/summary_inputs.rs b/crates/logical-asap-optimization/src/pass2/memo/rules/summary_inputs.rs index ba1267965..87336af43 100644 --- a/crates/logical-asap-optimization/src/pass2/memo/rules/summary_inputs.rs +++ b/crates/logical-asap-optimization/src/pass2/memo/rules/summary_inputs.rs @@ -1,10 +1,10 @@ //! Expose private value projections and filters below summary builds as canonical inputs. //! This preserves their full expressions and schema, including opaque series //! identity, rather than assuming that a projection is an identity operation. -use super::reuse::{append_producer, append_reader, eligible, replace}; +use super::reuse::{add_candidate_over_groups, eligible, find_or_add_producer, replace}; use crate::pass2::memo::state_groups::{boundary_ptrs, nodes_within, Pass2RuleError}; use asap_types::ir::{ASAPOp, NonASAPOp, OperatorNode}; -use asap_types::stages::{Memo, OriginOfCandidateForMemoGroup}; +use asap_types::stages::{Memo, MemoGroupId, OriginOfCandidateForMemoGroup}; use std::collections::{HashMap, HashSet}; use std::rc::Rc; @@ -17,7 +17,7 @@ pub(crate) fn factor_summary_inputs(memo: &mut Memo) -> Result<(), Pass2RuleErro candidates .iter() .filter(|candidate| eligible(candidate)) - .map(move |candidate| (owner, candidate.clone())) + .map(move |candidate| (MemoGroupId(owner as u32), candidate.clone())) }) .collect(); for (owner, original) in originals { @@ -35,14 +35,14 @@ pub(crate) fn factor_summary_inputs(memo: &mut Memo) -> Result<(), Pass2RuleErro } } } - let queries = memo.groups()[owner].queries.clone(); + let queries = memo.groups()[owner.0 as usize].queries.clone(); let mut replacements = HashMap::new(); for input in inputs { - canonical_preprocessing(memo, &input, &boundaries, &queries, &mut replacements)?; + canonical_preprocessing(memo, &input, &boundaries, &queries, &mut replacements); } if !replacements.is_empty() { let reader = replace(&original.root_node, &replacements); - append_reader(memo, owner, &original, reader, original.origin.clone())?; + add_candidate_over_groups(memo, owner, &original, reader, original.origin.clone())?; } } Ok(()) @@ -54,47 +54,36 @@ fn canonical_preprocessing( boundaries: &HashSet<*const OperatorNode>, queries: &[usize], replacements: &mut HashMap<*const OperatorNode, Rc>, -) -> Result { - if replacements.contains_key(&Rc::as_ptr(node)) { - return Ok(true); - } - if boundaries.contains(&Rc::as_ptr(node)) { - // Logical TimeRange groups can remain distinct when their inputs - // differ structurally. The existing state-sharing rule nevertheless - // permits identical range descriptors over the same canonical source. - if let Some(group) = memo.groups().iter().find(|group| { - crate::pass2::memo::state_groups::same_range_input(&group.logical_node, node) - }) { - replacements.insert(Rc::as_ptr(node), Rc::clone(&group.logical_node)); - } - return Ok(true); +) -> bool { + if replacements.contains_key(&Rc::as_ptr(node)) || boundaries.contains(&Rc::as_ptr(node)) { + return true; } if !matches!( node.non_asap(), Some(NonASAPOp::Project { .. } | NonASAPOp::Filter { .. }) ) { - return Ok(false); + return false; } // Predicates/projections can reference subqueries. Every dependency must // be canonical, not only the row-stream child. for child in node.children() { - if !canonical_preprocessing(memo, child, boundaries, queries, replacements)? { - return Ok(false); + if !canonical_preprocessing(memo, child, boundaries, queries, replacements) { + return false; } } let canonical = replace(node, replacements); - let group = append_producer( + let group = find_or_add_producer( memo, - canonical, + &canonical, queries, - OriginOfCandidateForMemoGroup::SummaryInput, - // These operators introduce no approximate state. Their identity does - // not depend on how a downstream consumer divides its error budget. - 1, - )?; + // These operators introduce no approximate state, so the group does + // not depend on how a downstream consumer divides its error budget + // ([`find_or_add_producer`] sizes it for the whole budget). + &OriginOfCandidateForMemoGroup::SummaryInput, + ); replacements.insert( Rc::as_ptr(node), Rc::clone(&memo.groups()[group.0 as usize].logical_node), ); - Ok(true) + true } diff --git a/crates/logical-asap-optimization/src/pass2/memo/rules/summary_results.rs b/crates/logical-asap-optimization/src/pass2/memo/rules/summary_results.rs index 42327db2d..d911d966a 100644 --- a/crates/logical-asap-optimization/src/pass2/memo/rules/summary_results.rs +++ b/crates/logical-asap-optimization/src/pass2/memo/rules/summary_results.rs @@ -1,13 +1,15 @@ -//! Factor composite implementations at summary readouts. Each readout can then -//! choose a state configuration independently, without enumerating combinations -//! of all private states in the enclosing query. -use super::reuse::{append_producer, append_reader, eligible, replace}; +//! Factor composite implementations at summary readouts: a candidate with two +//! or more private builds and a readout of one of them gets a copy as one +//! operator over groups, so each readout is its own group and can choose a +//! state configuration independently, without enumerating combinations of +//! all private states in the enclosing query. +use super::reuse::{add_candidate_over_groups, eligible}; use crate::pass2::memo::state_groups::{ boundary_ptrs, direct_input_builds, nodes_within, Pass2RuleError, }; use asap_types::ir::{ASAPOp, OperatorNode}; -use asap_types::stages::{Memo, OriginOfCandidateForMemoGroup}; -use std::collections::{HashMap, HashSet}; +use asap_types::stages::{Memo, MemoGroupId, OriginOfCandidateForMemoGroup}; +use std::collections::HashSet; use std::rc::Rc; pub(crate) fn factor_summary_results(memo: &mut Memo) -> Result { @@ -19,7 +21,7 @@ pub(crate) fn factor_summary_results(memo: &mut Memo) -> Result Result = builds.iter().map(|(build, _)| Rc::as_ptr(build)).collect(); let boundaries = boundary_ptrs(memo, &original); - let mut readouts = Vec::new(); - for node in nodes_within(&original.root_node, &boundaries) { - if !Rc::ptr_eq(&node, &original.root_node) + // A readout of exactly one of its private builds. + let reads_one_build = |node: &Rc| { + !Rc::ptr_eq(node, &original.root_node) && matches!( node.asap(), Some(ASAPOp::SummaryEstimate { .. } | ASAPOp::FinalizeExactAccumulator { .. }) ) - { - let private_builds: HashSet<_> = OperatorNode::reachable(&node) + && OperatorNode::reachable(node) .iter() .map(Rc::as_ptr) .filter(|pointer| build_pointers.contains(pointer)) - .collect(); - if private_builds.len() == 1 { - readouts.push(Rc::clone(&node)); - } - } - } - if readouts.is_empty() { + .collect::>() + .len() + == 1 + }; + if !nodes_within(&original.root_node, &boundaries) + .iter() + .any(reads_one_build) + { continue; } - let queries = memo.groups()[owner].queries.clone(); - let mut replacements = HashMap::new(); - for readout in readouts { - let group = append_producer( - memo, - Rc::clone(&readout), - &queries, - OriginOfCandidateForMemoGroup::SummaryResult, - original.accuracy_budget_parts, - )?; - // The original reader's allocation was used to generate these - // parameters; lifting the boundary must not reset that provenance. - replacements.insert( - Rc::as_ptr(&readout), - Rc::clone(&memo.groups()[group.0 as usize].logical_node), - ); - } - let reader = replace(&original.root_node, &replacements); - added += usize::from(append_reader( + // Every node beneath the root becomes a group, so each readout is + // its own group and chooses its state configuration on its own. + added += usize::from(add_candidate_over_groups( memo, owner, &original, - reader, + Rc::clone(&original.root_node), OriginOfCandidateForMemoGroup::SummaryResult, )?); } diff --git a/crates/logical-asap-optimization/src/pass2/memo/state_groups.rs b/crates/logical-asap-optimization/src/pass2/memo/state_groups.rs index 44e238b19..44e1288b2 100644 --- a/crates/logical-asap-optimization/src/pass2/memo/state_groups.rs +++ b/crates/logical-asap-optimization/src/pass2/memo/state_groups.rs @@ -133,14 +133,11 @@ pub(crate) fn add_shared_summary_state_groups(memo: &mut Memo) -> Result<(), Pas // Only declared inputs and newly lifted states are boundaries. A // factored result group's representative can equal its candidate root; // treating every memo node as an input would create a self-alias here. - let boundaries = reader - .input_groups - .iter() - .chain(&state_groups) - .map(|id| (Rc::as_ptr(&memo.groups()[id.0 as usize].logical_node), *id)) - .collect(); + let declared = reader.input_groups.clone(); reader.root_node = super::rules::reuse::replace(&reader.root_node, &replacements); - reader.input_groups = boundary_groups(&reader.root_node, &boundaries); + reader.input_groups = boundary_groups(memo, &reader.root_node, |group| { + declared.contains(&group) || state_groups.contains(&group) + }); reader.origin = OriginOfCandidateForMemoGroup::SharedBuild; memo.candidates[group].push(reader); } @@ -356,9 +353,13 @@ pub(super) fn builds_beneath( builds } +/// The groups `node` reads: walking down from `node`, each node that is a +/// group's own node ([`Memo::group_whose_node_is`]) of a group +/// `is_boundary` accepts, in the order first reached, each once. pub(super) fn boundary_groups( + memo: &Memo, node: &Rc, - boundaries: &HashMap<*const OperatorNode, MemoGroupId>, + is_boundary: impl Fn(MemoGroupId) -> bool, ) -> Vec { let mut found = Vec::new(); let mut stack = vec![Rc::clone(node)]; @@ -367,7 +368,10 @@ pub(super) fn boundary_groups( if !seen.insert(Rc::as_ptr(&node)) { continue; } - if let Some(&group) = boundaries.get(&Rc::as_ptr(&node)) { + if let Some(group) = memo + .group_whose_node_is(&node) + .filter(|&group| is_boundary(group)) + { if !found.contains(&group) { found.push(group); } @@ -602,12 +606,6 @@ mod tests { tumbling_window_candidates, window_composition_pass2_candidates, }; - let boundaries: HashMap<_, _> = memo - .groups() - .iter() - .enumerate() - .map(|(id, group)| (Rc::as_ptr(&group.logical_node), MemoGroupId(id as u32))) - .collect(); // As local window composition does: Stage 0's groups only. for group in 0..memo.groups().len() { if !memo.is_stage0_group(MemoGroupId(group as u32)) { @@ -640,7 +638,7 @@ mod tests { let node = replace_node(node, chain.children()[0], &input); let mut tumbling_window = candidate; tumbling_window.root_node = Rc::clone(&node); - tumbling_window.input_groups = boundary_groups(&node, &boundaries); + tumbling_window.input_groups = boundary_groups(&memo, &node, |_| true); tumbling_window.origin = OriginOfCandidateForMemoGroup::TumblingWindows; memo.candidates[group].push(tumbling_window); } @@ -1308,9 +1306,12 @@ mod tests { } assert!(injected >= 2); let expanded = expand(memo).unwrap(); - assert!(expanded.state_groups.iter().any(|state| { - let group = &expanded.memo.groups()[state.0 as usize]; - group.queries == vec![0, 1] + // One build group both queries reach, over a canonical input group. + assert!(expanded.memo.groups().iter().any(|group| { + matches!( + group.logical_node.asap(), + Some(ASAPOp::SummaryAgg { .. } | ASAPOp::SummaryTumblingWindowChain { .. }) + ) && group.queries == vec![0, 1] && group.child_groups.iter().any(|input| { expanded.memo.candidates[input.0 as usize] .iter() diff --git a/crates/planner/src/run.rs b/crates/planner/src/run.rs index f1ec8b489..fa654910d 100644 --- a/crates/planner/src/run.rs +++ b/crates/planner/src/run.rs @@ -282,13 +282,10 @@ pub fn selected_chosen_candidates( continue; // a candidate's private node } let options = &node.candidates_for_memo_group; - let reads = |&&CandidateForMemoGroupId { group, index }: &&CandidateForMemoGroupId| { + let reads = |&&option: &&CandidateForMemoGroupId| { producer_groups.get(&node_id).is_none_or(|read| { read.iter().all(|&child| { - child == group - || memo.candidates[group.0 as usize][index] - .input_groups - .contains(&child) + child == option.group || memo.candidate(option).input_groups.contains(&child) }) }) }; diff --git a/crates/types/src/stages/candidates.rs b/crates/types/src/stages/candidates.rs index 80e4a6c13..e4ecc1943 100644 --- a/crates/types/src/stages/candidates.rs +++ b/crates/types/src/stages/candidates.rs @@ -39,6 +39,14 @@ struct MemoGroupLookup { group_with_equal_node: HashConsingTable, } +/// One implementation candidate: index `index` in group `group`'s +/// candidates ([`Memo::candidates`]). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub struct CandidateForMemoGroupId { + pub group: MemoGroupId, + pub index: usize, +} + /// One way to implement a group: its node as Stage 2 implements it, over /// child groups. #[derive(Debug, Clone)] @@ -122,6 +130,37 @@ impl Memo { &mut self.groups } + /// Adds `candidate` to `group`'s candidates and returns its id, or + /// returns the id of the candidate there already with the same input + /// groups, fused groups and accuracy budget parts and an equal node + /// ([`equal_nodes`](super::hash_consing::equal_nodes)): a candidate is + /// in a group once. + pub fn add_candidate( + &mut self, + group: MemoGroupId, + candidate: CandidateForMemoGroup, + ) -> CandidateForMemoGroupId { + let candidates = &mut self.candidates[group.0 as usize]; + let index = candidates + .iter() + .position(|known| { + known.input_groups == candidate.input_groups + && known.fused_groups == candidate.fused_groups + && known.accuracy_budget_parts == candidate.accuracy_budget_parts + && super::hash_consing::equal_nodes(&known.root_node, &candidate.root_node) + }) + .unwrap_or_else(|| { + candidates.push(candidate); + candidates.len() - 1 + }); + CandidateForMemoGroupId { group, index } + } + + /// The candidate `id` names. + pub fn candidate(&self, id: CandidateForMemoGroupId) -> &CandidateForMemoGroup { + &self.candidates[id.group.0 as usize][id.index] + } + /// Adds `group` with its `candidates` and returns its id. pub fn push_group( &mut self, diff --git a/crates/types/src/stages/hash_consing.rs b/crates/types/src/stages/hash_consing.rs index dcf16edfa..492227af1 100644 --- a/crates/types/src/stages/hash_consing.rs +++ b/crates/types/src/stages/hash_consing.rs @@ -162,6 +162,14 @@ fn same_node( && same_value(&own_fields(left), &own_fields(right)) } +/// Whether `left` and `right` are equal by Stage 0's equality ([`same_node`]), +/// whether or not sharing them is legal: children compared by pointer +/// first, so two operators over the same groups' nodes compare in `O(1)` +/// beyond their own fields. +pub fn equal_nodes(left: &OperatorNode, right: &OperatorNode) -> bool { + same_node(left, right, &mut NodePairToIsSameNodeCache::new()) +} + /// Whether two equal copies of `node` return the same rows for the same /// input rows, so they may be one node: an ASAP node, or a non-ASAP node /// [`NonASAPOp::is_shareable_when_equal`]. diff --git a/crates/types/src/stages/mod.rs b/crates/types/src/stages/mod.rs index 69e9f1a06..2111434e5 100644 --- a/crates/types/src/stages/mod.rs +++ b/crates/types/src/stages/mod.rs @@ -17,7 +17,9 @@ mod physical; mod pruning; mod realizer; -pub use candidates::{CandidateForMemoGroup, Memo, OriginOfCandidateForMemoGroup}; +pub use candidates::{ + CandidateForMemoGroup, CandidateForMemoGroupId, Memo, OriginOfCandidateForMemoGroup, +}; pub use logical_memo_candidates::LogicalASAPMemoCandidates; pub use memo::{queries_rooted_at, root_groups, DagOfLogicalMemoGroups, MemoGroup, MemoGroupId}; pub use per_computation_logical_memo_candidates::{ @@ -25,10 +27,10 @@ pub use per_computation_logical_memo_candidates::{ HowToComputeAggregateMeasure, PerComputationLogicalMemoCandidates, }; pub use physical::{ - output_references, CandidateForMemoGroupId, DagOfSelectedCandidates, - OperatorNodeOfCandidateForMemoGroup, OperatorNodeOfCandidateForMemoGroupId, - PhysicalASAPMemoCandidates, PhysicalCandidate, PhysicalCandidateId, PhysicalCandidateKey, - ProducerCandidates, ProducerSetInterner, ReaderConstraint, SelectedEdge, SelectedNodeId, + output_references, DagOfSelectedCandidates, OperatorNodeOfCandidateForMemoGroup, + OperatorNodeOfCandidateForMemoGroupId, PhysicalASAPMemoCandidates, PhysicalCandidate, + PhysicalCandidateId, PhysicalCandidateKey, ProducerCandidates, ProducerSetInterner, + ReaderConstraint, SelectedEdge, SelectedNodeId, }; #[cfg(any(test, feature = "test-support"))] pub use physical::{producer_storage_metrics, ProducerStorageMetrics}; diff --git a/crates/types/src/stages/physical/memo_candidates.rs b/crates/types/src/stages/physical/memo_candidates.rs index 1d500fc27..0bf2e7bc4 100644 --- a/crates/types/src/stages/physical/memo_candidates.rs +++ b/crates/types/src/stages/physical/memo_candidates.rs @@ -9,15 +9,7 @@ use super::producers::ProducerCandidates; use crate::ir::properties::ExecutionTiming; use crate::ir::{OperatorNode, QueryRoot}; use crate::physical::Materialization; -use crate::stages::{MemoGroupId, PrunedCandidate}; - -/// One implementation candidate: index `index` in group `group`'s -/// candidates ([`Memo::candidates`](crate::stages::Memo::candidates)). -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] -pub struct CandidateForMemoGroupId { - pub group: MemoGroupId, - pub index: usize, -} +use crate::stages::{CandidateForMemoGroupId, MemoGroupId, PrunedCandidate}; /// Index into /// [`PhysicalASAPMemoCandidates::operator_nodes_of_candidates_for_memo_groups`]. diff --git a/crates/types/src/stages/physical/mod.rs b/crates/types/src/stages/physical/mod.rs index 7f029925a..459c0666c 100644 --- a/crates/types/src/stages/physical/mod.rs +++ b/crates/types/src/stages/physical/mod.rs @@ -9,9 +9,8 @@ pub use graph::{ DagOfSelectedCandidates, PhysicalCandidateId, ReaderConstraint, SelectedEdge, SelectedNodeId, }; pub use memo_candidates::{ - output_references, CandidateForMemoGroupId, OperatorNodeOfCandidateForMemoGroup, - OperatorNodeOfCandidateForMemoGroupId, PhysicalASAPMemoCandidates, PhysicalCandidate, - PhysicalCandidateKey, + output_references, OperatorNodeOfCandidateForMemoGroup, OperatorNodeOfCandidateForMemoGroupId, + PhysicalASAPMemoCandidates, PhysicalCandidate, PhysicalCandidateKey, }; #[cfg(any(test, feature = "test-support"))] pub use producers::{producer_storage_metrics, ProducerStorageMetrics}; diff --git a/crates/types/src/stages/physical/producers.rs b/crates/types/src/stages/physical/producers.rs index 0e7547434..fb0d9ec4e 100644 --- a/crates/types/src/stages/physical/producers.rs +++ b/crates/types/src/stages/physical/producers.rs @@ -115,13 +115,13 @@ mod tests { fn shared_producer_storage_keeps_input_choices_independent() { use super::super::graph::{DagOfSelectedCandidates, SelectedEdge, SelectedNodeId}; use super::super::memo_candidates::{ - CandidateForMemoGroupId, OperatorNodeOfCandidateForMemoGroup, - OperatorNodeOfCandidateForMemoGroupId, PhysicalASAPMemoCandidates, PhysicalCandidate, + OperatorNodeOfCandidateForMemoGroup, OperatorNodeOfCandidateForMemoGroupId, + PhysicalASAPMemoCandidates, PhysicalCandidate, }; use crate::ir::properties::ExecutionTiming; use crate::ir::QueryRoot; use crate::physical::Materialization; - use crate::stages::MemoGroupId; + use crate::stages::{CandidateForMemoGroupId, MemoGroupId}; let mut interner = ProducerSetInterner::default(); let choices = interner.intern(vec![PhysicalCandidateId(0), PhysicalCandidateId(1)]); use crate::ir::scalar::expr_ir::ScalarValue;