From 40ee8d62a23e2d551e0eaa8bafddb2d6c04d6991 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 2 Nov 2026 22:33:15 -0500 Subject: [PATCH 1/2] =?UTF-8?q?refactor(stage1):=20move=20Pass=201's=20mem?= =?UTF-8?q?o=20population=20into=20pass1;=20forbid=20pass1=20=E2=86=92=20p?= =?UTF-8?q?ass2=20imports=20(#1096)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `pass2/memo/groups.rs` (`try_populate_pass1`, `NodeToStructuralIdCache`) and `pass2/memo/candidates.rs` (`CandidatePopulationError`) are Pass 1 code. They move into `pass1/memo_population.rs`, so Pass 1's production code imports nothing from Pass 2. `tests/pass_layering.rs` enforces the direction; Pass 2 → Pass 1 remains (#1191). Co-Authored-By: Claude Opus 5.5 --- crates/logical-asap-optimization/src/lib.rs | 2 +- crates/logical-asap-optimization/src/pass1.rs | 2 ++ .../src/pass1/logical_memo_candidates.rs | 4 +-- .../groups.rs => pass1/memo_population.rs} | 21 +++++++---- .../src/pass1/per_computation_candidates.rs | 2 +- .../src/pass2/memo.rs | 4 --- .../src/pass2/memo/candidates.rs | 15 -------- .../src/pass2/memo/rules/rollup.rs | 2 +- .../rules/shared_configuration_readers.rs | 2 +- .../src/pass2/memo/state_groups.rs | 2 +- .../src/pass2/memo/window_candidates.rs | 2 +- .../src/test_support.rs | 2 +- .../tests/pass_layering.rs | 36 +++++++++++++++++++ .../design_docs/proposals/planner-layering.md | 4 +-- 14 files changed, 64 insertions(+), 36 deletions(-) rename crates/logical-asap-optimization/src/{pass2/memo/groups.rs => pass1/memo_population.rs} (98%) delete mode 100644 crates/logical-asap-optimization/src/pass2/memo/candidates.rs create mode 100644 crates/logical-asap-optimization/tests/pass_layering.rs diff --git a/crates/logical-asap-optimization/src/lib.rs b/crates/logical-asap-optimization/src/lib.rs index bee42d27d..213790d6d 100644 --- a/crates/logical-asap-optimization/src/lib.rs +++ b/crates/logical-asap-optimization/src/lib.rs @@ -88,8 +88,8 @@ pub use pass1::{ PerComputationLogicalMemoCandidatesError, }; // Pass 2: its entry and its error. +pub use pass1::memo_population::CandidatePopulationError; pub use pass2::logical_memo_candidates::generate_logical_asap_memo_candidates; -pub use pass2::memo::candidates::CandidatePopulationError; pub use pass2::memo::state_groups::Pass2RuleError; // The CandidateLogicalASAPDAGs a memo stands for (developer tools' // reference pricing). diff --git a/crates/logical-asap-optimization/src/pass1.rs b/crates/logical-asap-optimization/src/pass1.rs index f08d5e695..2f115053b 100644 --- a/crates/logical-asap-optimization/src/pass1.rs +++ b/crates/logical-asap-optimization/src/pass1.rs @@ -7,6 +7,8 @@ /// Stage 1, Pass 1's entry and output: the per-computation candidates on /// the memo groups ([`PerComputationLogicalMemoCandidates`](asap_types::stages::PerComputationLogicalMemoCandidates)). pub(crate) mod logical_memo_candidates; +/// Pass 1's candidates on Stage 0's memo groups (`try_populate_pass1`). +pub(crate) mod memo_population; /// Stage 1, Pass 1's computation targets, their candidates and their /// composition. No execution timing is assigned: that is a Stage 2 /// materialization decision. Its target list diff --git a/crates/logical-asap-optimization/src/pass1/logical_memo_candidates.rs b/crates/logical-asap-optimization/src/pass1/logical_memo_candidates.rs index f045fa778..1d20e5e79 100644 --- a/crates/logical-asap-optimization/src/pass1/logical_memo_candidates.rs +++ b/crates/logical-asap-optimization/src/pass1/logical_memo_candidates.rs @@ -13,12 +13,12 @@ use asap_accuracy::sizing::{ }; use asap_accuracy::{AccuracyEvidenceProvider, AccuracyModel}; +use super::memo_population::try_populate_pass1; +use super::memo_population::CandidatePopulationError; use crate::deployment_capabilities::SupportedSummary; use crate::pass1::per_computation_candidates::{ generate_per_computation_candidates, PerComputationCandidateError, }; -use crate::pass2::memo::candidates::CandidatePopulationError; -use crate::pass2::memo::groups::try_populate_pass1; use asap_types::stages::{ DagOfLogicalMemoGroups, Memo, PerComputationLogicalMemoCandidates, PrunedCandidate, }; diff --git a/crates/logical-asap-optimization/src/pass2/memo/groups.rs b/crates/logical-asap-optimization/src/pass1/memo_population.rs similarity index 98% rename from crates/logical-asap-optimization/src/pass2/memo/groups.rs rename to crates/logical-asap-optimization/src/pass1/memo_population.rs index 55c339479..b3b411f1a 100644 --- a/crates/logical-asap-optimization/src/pass2/memo/groups.rs +++ b/crates/logical-asap-optimization/src/pass1/memo_population.rs @@ -11,7 +11,7 @@ //! //! A Pass 1 target's aggregate group gets one candidate per combination of //! its measures' Pass 1 candidates: the aggregate alone composed by -//! [`crate::pass1::per_computation_candidates::compose_logical_candidate`] with that choice, every target beneath it +//! [`super::per_computation_candidates::compose_logical_candidate`] with that choice, every target beneath it //! exact. Its local nodes are the composed nodes above the first node //! equal to one of the group's descendant groups, which are its children. //! The local nodes read each child group through its node (the same @@ -38,18 +38,28 @@ use asap_types::stages::HowToComputeAggregateMeasure; use std::collections::{HashMap, HashSet}; use std::rc::Rc; -use crate::pass1::per_computation_candidates::{ +use super::per_computation_candidates::{ compose_logical_candidate_with_accuracy_budget_parts, PerComputationCandidates, }; use asap_types::ir::schema::FieldDataType; use asap_types::ir::{ASAPOp, OperatorNode, QueryRoot}; -use super::candidates::CandidatePopulationError; use asap_types::stages::{ CandidateForMemoGroup, DagOfLogicalMemoGroups, Memo, MemoGroup, MemoGroupId, OriginOfCandidateForMemoGroup, }; +/// A Pass 1 inventory does not belong to the initial logical memo. +#[derive(Debug, thiserror::Error, PartialEq, Eq)] +pub enum CandidatePopulationError { + #[error("Pass 1 root dependencies differ from the initial logical memo")] + RootDependenciesChanged, + #[error("Pass 1 target {target} is not a valid measure of an initial aggregate group")] + InvalidTarget { target: usize }, + #[error("Pass 1 target {target} references missing absorbed target {absorbed}")] + InvalidAbsorption { target: usize, absorbed: usize }, +} + /// Populate implementation candidates without changing initial group identities. /// The inventory must have walked `initial`'s groups /// (`PerComputationCandidates::walked`); its targets name them by id. @@ -308,8 +318,7 @@ fn compose_alone( targets: &[usize], choice: &[usize], accuracy_budget_parts: usize, -) -> Result, crate::pass1::per_computation_candidates::PerComputationCandidateError> -{ +) -> Result, super::per_computation_candidates::PerComputationCandidateError> { let mut kept: Vec = targets.to_vec(); for &t in targets { if let Some((_, Some(absorbs))) = per_computation.targets[t].computation_for(choice[t]) { @@ -770,7 +779,7 @@ mod tests { } /// The sketch summaries of the first query of `queries` composed with - /// every target at its `algorithm` summary, by [`crate::pass1::per_computation_candidates::compose_logical_candidate`]: + /// every target at its `algorithm` summary, by [`super::per_computation_candidates::compose_logical_candidate`]: /// as the exhaustive search sizes them. fn composed_sketches( queries: &[&str], 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 f1ef5319a..e18050e22 100644 --- a/crates/logical-asap-optimization/src/pass1/per_computation_candidates.rs +++ b/crates/logical-asap-optimization/src/pass1/per_computation_candidates.rs @@ -239,7 +239,7 @@ impl PerComputationCandidates { /// 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 crate::pass2::memo::NodeToStructuralIdCache; + use super::memo_population::NodeToStructuralIdCache; let mut structural_ids = NodeToStructuralIdCache::default(); let group_of: HashMap<*const OperatorNode, MemoGroupId> = self .logical_nodes diff --git a/crates/logical-asap-optimization/src/pass2/memo.rs b/crates/logical-asap-optimization/src/pass2/memo.rs index f41be0ddf..c9bc66a80 100644 --- a/crates/logical-asap-optimization/src/pass2/memo.rs +++ b/crates/logical-asap-optimization/src/pass2/memo.rs @@ -1,12 +1,8 @@ //! The logical memo's candidates, rules and enumeration, over the memo //! groups Stage 0 builds (`crate::stage0`). -/// Logical implementation candidates populated after initial identity construction. -pub(crate) mod candidates; /// The candidate DAGs a memo stands for: choices and their composition. pub(crate) mod enumeration; -pub(crate) mod groups; -pub(crate) use groups::NodeToStructuralIdCache; pub(crate) mod rules; /// Shared finite state configuration enumeration. pub(crate) mod shared_configurations; diff --git a/crates/logical-asap-optimization/src/pass2/memo/candidates.rs b/crates/logical-asap-optimization/src/pass2/memo/candidates.rs deleted file mode 100644 index 5712ec8ba..000000000 --- a/crates/logical-asap-optimization/src/pass2/memo/candidates.rs +++ /dev/null @@ -1,15 +0,0 @@ -//! Logical implementation candidates attached to stable memo groups. -//! -//! Rule expansion owns no pricing or solver state. Shared computations are -//! explicit state groups connected through candidate child dependencies. - -/// A Pass 1 inventory does not belong to the initial logical memo. -#[derive(Debug, thiserror::Error, PartialEq, Eq)] -pub enum CandidatePopulationError { - #[error("Pass 1 root dependencies differ from the initial logical memo")] - RootDependenciesChanged, - #[error("Pass 1 target {target} is not a valid measure of an initial aggregate group")] - InvalidTarget { target: usize }, - #[error("Pass 1 target {target} references missing absorbed target {absorbed}")] - InvalidAbsorption { target: usize, absorbed: usize }, -} diff --git a/crates/logical-asap-optimization/src/pass2/memo/rules/rollup.rs b/crates/logical-asap-optimization/src/pass2/memo/rules/rollup.rs index af01314ef..1f684e04f 100644 --- a/crates/logical-asap-optimization/src/pass2/memo/rules/rollup.rs +++ b/crates/logical-asap-optimization/src/pass2/memo/rules/rollup.rs @@ -313,8 +313,8 @@ pub(crate) fn add_window_rollup_candidates(memo: &mut Memo) -> Result bool { #[cfg(test)] mod tests { use super::*; + use crate::pass1::memo_population::try_populate_pass1; use crate::pass1::per_computation_candidates::generate_per_computation_candidates; - use crate::pass2::memo::groups::try_populate_pass1; use crate::stage0::generate_dag_of_logical_memo_groups; use crate::test_support::{entry_with, lower_promql}; use asap_types::ir::properties::AccuracyTarget; 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 bb945bd6c..4cee7ca54 100644 --- a/crates/logical-asap-optimization/src/pass2/memo/state_groups.rs +++ b/crates/logical-asap-optimization/src/pass2/memo/state_groups.rs @@ -442,8 +442,8 @@ fn replace_node( #[cfg(test)] mod tests { use super::*; + use crate::pass1::memo_population::try_populate_pass1; use crate::pass1::per_computation_candidates::generate_per_computation_candidates; - use crate::pass2::memo::groups::try_populate_pass1; use crate::stage0::generate_dag_of_logical_memo_groups; use asap_types::ir::properties::AccuracyTarget; use asap_types::ir::QueryRoot; diff --git a/crates/logical-asap-optimization/src/pass2/memo/window_candidates.rs b/crates/logical-asap-optimization/src/pass2/memo/window_candidates.rs index 6f981102a..468de54d5 100644 --- a/crates/logical-asap-optimization/src/pass2/memo/window_candidates.rs +++ b/crates/logical-asap-optimization/src/pass2/memo/window_candidates.rs @@ -252,8 +252,8 @@ fn candidate_signature(candidate: &CandidateForMemoGroup) -> Result`s; `ComputationTarget` is Pass 1's working type, in `asap_types::stages`), the one entry: inside it, the computation targets (walking the memo groups), the summaries the deployment does not support and those no query can use pruned, the candidates attached to the groups (`try_populate_pass1`), then the accuracy-scoped sizes | `pass1/logical_memo_candidates.rs`; `pass1/per_computation_candidates.rs`; `pass2/memo/groups.rs` | +| Pass 1 | `generate_per_computation_logical_memo_candidates` → `PerComputationLogicalMemoCandidates` (its `Memo` and its `PrunedCandidate`s; `ComputationTarget` is Pass 1's working type, in `asap_types::stages`), the one entry: inside it, the computation targets (walking the memo groups), the summaries the deployment does not support and those no query can use pruned, the candidates attached to the groups (`try_populate_pass1`), then the accuracy-scoped sizes | `pass1/logical_memo_candidates.rs`; `pass1/per_computation_candidates.rs`; `pass1/memo_population.rs` | | Memo types | `DagOfLogicalMemoGroups`, `PerComputationLogicalMemoCandidates`, `HowToComputeAggregateMeasure`, Pass 1's working types (`ComputationTarget`, `FusedComputation`, `HavingEntry`, `HavingPlacement`), `Memo`, `CandidateForMemoGroup`, `OriginOfCandidateForMemoGroup`, `LogicalASAPMemoCandidates`, `CandidateGenerationTimeout`, `root_groups`, `queries_rooted_at` | `crates/types/src/stages/` (`asap_types::stages`) | | Pass 2 | `generate_logical_asap_memo_candidates` → `LogicalASAPMemoCandidates`, from Pass 1's output. It calls in this order: 1. window composition for every legal tumbling-window width (`add_local_window_composition`); 2–9. the rules in `pass2/memo/rules/`: scan widening, summary inputs, top-k reuse, finer aggregate, rollup, window rollup, summary results, state configurations; 10. `add_shared_summary_state_groups`, the shared builds | `pass2/logical_memo_candidates.rs` (`LogicalASAPMemoCandidates` itself: `asap_types::stages`); `pass2/memo/window_candidates.rs`, `pass2/memo/rules/`, `pass2/memo/state_groups.rs`, `pass2/memo/shared_configurations.rs` | | Memo to candidate DAGs | `enumerate`, `build_logical_asap_dags_from_chosen_candidates`, `ChosenCandidatePerMemoGroup`, `AllChosenCandidatesPerMemoGroup`, `ErrorBuildingLogicalASAPDagsFromChosenCandidates` | `pass2/memo/enumeration.rs` (re-exported at the crate root) | From c85e82e726bacd71f83f9ca92c0bb9b8ded43745 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 2 Nov 2026 22:36:27 -0500 Subject: [PATCH 2/2] refactor(stage1): one same_family, boundary_ptrs/nodes_within helpers, dead Pass 2 code (#1096) - The two identical `same_family` copies use `stage0::cse::same_value`; `asap_aware_cse`'s differently-meant one is `same_build_family`. - `boundary_ptrs` and `nodes_within` replace 4 copies of a candidate's boundary set and 3 copies of the walk above it (same visit order). - Delete `Pass2RuleError::InvalidReaderBoundary` (never constructed) and the test-only `may_change_rollup_candidate`. Co-Authored-By: Claude Opus 5.5 --- .../src/pass2/asap_aware_cse.rs | 34 ++++--------------- .../rules/shared_configuration_readers.rs | 19 ++++------- .../src/pass2/memo/rules/summary_inputs.rs | 16 ++------- .../src/pass2/memo/rules/summary_results.rs | 18 +++------- .../src/pass2/memo/rules/topk.rs | 27 ++++----------- .../src/pass2/memo/shared_configurations.rs | 17 +++------- .../src/pass2/memo/state_groups.rs | 33 ++++++++++++++++-- .../src/stage0/cse.rs | 5 +-- 8 files changed, 66 insertions(+), 103 deletions(-) diff --git a/crates/logical-asap-optimization/src/pass2/asap_aware_cse.rs b/crates/logical-asap-optimization/src/pass2/asap_aware_cse.rs index 681f06973..c402d1f33 100644 --- a/crates/logical-asap-optimization/src/pass2/asap_aware_cse.rs +++ b/crates/logical-asap-optimization/src/pass2/asap_aware_cse.rs @@ -359,7 +359,7 @@ fn collect_children_first( /// Whether builds of families `a` and `b` can be one build: one sketch /// algorithm and grouping at any sizes, or one exact accumulator. -fn same_family(a: &FieldDataType, b: &FieldDataType) -> bool { +fn same_build_family(a: &FieldDataType, b: &FieldDataType) -> bool { match (a, b) { (FieldDataType::Sketch(a, grouping), FieldDataType::Sketch(b, grouping_b)) => { a.algorithm() == b.algorithm() && grouping == grouping_b @@ -962,7 +962,7 @@ fn rolls_up_into(member: &TumblingWindowCandidate, host: &TumblingWindowCandidat && by.len() < host_by.len() && by.keys().iter().all(|key| host_by.keys().contains(key)) && FamilyCapability::merges_groups(family) - && same_family(family, host_family) + && same_build_family(family, host_family) && (Rc::ptr_eq(below, host_below) || below == host_below) && input == host_input && grouping == host_grouping @@ -1004,7 +1004,7 @@ fn same_build_over_one_input(a: &TumblingWindowCandidate, b: &TumblingWindowCand .collect() }; names(&a.build) == names(&b.build) - && same_family(family, family_b) + && same_build_family(family, family_b) && (Rc::ptr_eq(below, below_b) || below == below_b) && input == input_b && reduction == reduction_b @@ -1402,21 +1402,6 @@ pub(crate) fn rollup_candidate_for( rollup_candidates_of(roots, Some(coarse)).into_iter().next() } -/// Whether adding `root` to a candidate's roots may change the rollup -/// candidate of its coarse build `coarse`: `root` reaches `coarse` itself -/// or a build that could answer it ([`finer_keys`]). Otherwise `root` adds -/// no reader of `coarse` or of any build that could answer it (a reader -/// reaches what it reads), no such build, and leaves the walk order of the -/// other nodes as it was, so [`rollup_candidate_for`] is unchanged. -#[cfg(test)] -fn may_change_rollup_candidate(coarse: &Rc, root: &Rc) -> bool { - OperatorNode::reachable(root).iter().any(|node| { - Rc::ptr_eq(node, coarse) - || finer_keys(coarse, node).is_some() - || filtered_finer(coarse, node).is_some() - }) -} - /// [`rollup_candidates`], over the coarse builds that are `only` when set. fn rollup_candidates_of( roots: &[QueryRoot], @@ -1695,7 +1680,7 @@ fn finer_keys(coarse: &OperatorNode, fine: &OperatorNode) -> Option { // A filter indexes its input's columns: equal only over equal rows. (a, b) => a == b && (Rc::ptr_eq(child, fine_child) || child == fine_child), }; - (same_family(family, fine_family) + (same_build_family(family, fine_family) && input == fine_input && same_filter && fine_keys.len() > coarse_keys.len() @@ -1733,7 +1718,7 @@ fn filtered_finer(coarse: &OperatorNode, fine: &OperatorNode) -> Option<(usize, }; if by.is_without() || fine_by.is_without() - || !same_family(family, fine_family) + || !same_build_family(family, fine_family) || input != fine_input { return None; @@ -3122,8 +3107,7 @@ mod tests { } /// One coarse build's rollup candidate, computed alone, is the one - /// [`rollup_candidates`] lists for it; a query reaching no build that - /// could answer it may not change it, one reaching a finer build may. + /// [`rollup_candidates`] lists for it, and needs a finer build to exist. #[test] fn one_coarse_builds_rollup_candidate_is_the_listed_one() { let roots = exact(&[ @@ -3139,12 +3123,6 @@ mod tests { } let coarse = &builds(&roots[0])[0]; assert!(rollup_candidate_for(&roots[..1], Rc::as_ptr(coarse)).is_none()); - assert!(may_change_rollup_candidate(coarse, operator(&roots[1]))); - let other_metric = exact(&["sum by (job, instance) (other)"]); - assert!(!may_change_rollup_candidate( - coarse, - operator(&other_metric[0]) - )); } /// `sum by (job)` is answered from `sum by (job, instance)`, which the 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 a903220b8..5d45eac0d 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 @@ -1,5 +1,6 @@ //! Expand compatible readers at finite shared sizes and tumbling-window horizons before //! interning concrete states. Local candidates remain available. +use crate::stage0::cse::same_value; use std::collections::{HashMap, HashSet}; use std::rc::Rc; @@ -15,7 +16,9 @@ use crate::pass2::memo::shared_configurations::{ shared_size, state_size_configurations, tumbling_window_configurations, TumblingWindowConfiguration, TumblingWindowRequirement, }; -use crate::pass2::memo::state_groups::{same_range_input, simple_build, Pass2RuleError}; +use crate::pass2::memo::state_groups::{ + boundary_ptrs, same_range_input, simple_build, Pass2RuleError, +}; use asap_types::stages::Memo; /// `resized` (a rewrite of `original` with the same shape) with each node @@ -293,11 +296,7 @@ impl SharedProducer for SummaryStates { with_resized_build(&roots, &need.build, family.clone()) } }; - let boundaries: HashSet<_> = original - .input_groups - .iter() - .map(|group| Rc::as_ptr(&memo.groups[group.0 as usize].logical_node)) - .collect(); + let boundaries = boundary_ptrs(memo, &original); let roots = result.map_err(|error| Pass2RuleError::StateConfiguration(error.to_string()))?; let [QueryRoot::Operator(node)] = roots.as_slice() else { @@ -432,7 +431,7 @@ fn tumbling_window_requirement(reader: &StateNeed) -> Option bool { - shared_size(family, reader).is_some_and(|joined| same_family(&joined, family)) + shared_size(family, reader).is_some_and(|joined| same_value(&joined, family)) } /// The shares of the budget a build of `family` is sized for: the most of @@ -446,10 +445,6 @@ fn covered_parts(family: &FieldDataType, set: &[Reader]) -> usize { .unwrap_or(1) } -fn same_family(left: &FieldDataType, right: &FieldDataType) -> bool { - left == right && serde_json::to_string(left).ok() == serde_json::to_string(right).ok() -} - fn compatible(left: &StateNeed, right: &StateNeed) -> bool { if shared_size(&left.family, &right.family).is_none() { return false; @@ -484,7 +479,7 @@ fn compatible(left: &StateNeed, right: &StateNeed) -> bool { } let mut schema = reader.build.schema.clone(); for field in &mut schema.fields { - if same_family(&field.dtype, &reader.family) { + if same_value(&field.dtype, &reader.family) { field.dtype = left.family.clone(); } } 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 06e1cbd04..e2bb0e84f 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 @@ -2,7 +2,7 @@ //! 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 crate::pass2::memo::state_groups::Pass2RuleError; +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 std::collections::{HashMap, HashSet}; @@ -21,18 +21,9 @@ pub(crate) fn factor_summary_inputs(memo: &mut Memo) -> Result<(), Pass2RuleErro }) .collect(); for (owner, original) in originals { - let boundaries: HashSet<_> = original - .input_groups - .iter() - .map(|group| Rc::as_ptr(&memo.groups[group.0 as usize].logical_node)) - .collect(); - let mut pending = vec![Rc::clone(&original.root_node)]; - let mut seen = HashSet::new(); + let boundaries = boundary_ptrs(memo, &original); let mut inputs = Vec::new(); - while let Some(node) = pending.pop() { - if boundaries.contains(&Rc::as_ptr(&node)) || !seen.insert(Rc::as_ptr(&node)) { - continue; - } + for node in nodes_within(&original.root_node, &boundaries) { if matches!( node.asap(), Some(ASAPOp::SummaryAgg { .. } | ASAPOp::SummaryTumblingWindowChain { .. }) @@ -43,7 +34,6 @@ pub(crate) fn factor_summary_inputs(memo: &mut Memo) -> Result<(), Pass2RuleErro } } } - pending.extend(node.children().into_iter().cloned()); } let queries = memo.groups[owner].queries.clone(); let mut replacements = HashMap::new(); 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 58137290e..49d84f991 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 @@ -2,7 +2,9 @@ //! 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}; -use crate::pass2::memo::state_groups::{direct_input_builds, Pass2RuleError}; +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}; @@ -28,18 +30,9 @@ pub(crate) fn factor_summary_results(memo: &mut Memo) -> Result = builds.iter().map(|(build, _)| Rc::as_ptr(build)).collect(); - let boundaries: HashSet<_> = original - .input_groups - .iter() - .map(|group| Rc::as_ptr(&memo.groups[group.0 as usize].logical_node)) - .collect(); - let mut pending = vec![Rc::clone(&original.root_node)]; - let mut seen = HashSet::new(); + let boundaries = boundary_ptrs(memo, &original); let mut readouts = Vec::new(); - while let Some(node) = pending.pop() { - if boundaries.contains(&Rc::as_ptr(&node)) || !seen.insert(Rc::as_ptr(&node)) { - continue; - } + for node in nodes_within(&original.root_node, &boundaries) { if !Rc::ptr_eq(&node, &original.root_node) && matches!( node.asap(), @@ -55,7 +48,6 @@ pub(crate) fn factor_summary_results(memo: &mut Memo) -> Result Result { @@ -206,25 +206,10 @@ fn exact_input_group( if !eligible(&candidate) { continue; } - let boundaries: HashSet<_> = candidate - .input_groups - .iter() - .map(|id| Rc::as_ptr(&memo.groups[id.0 as usize].logical_node)) - .collect(); - let mut pending = vec![Rc::clone(&candidate.root_node)]; - let mut seen = HashSet::new(); - let mut exact = true; - while let Some(node) = pending.pop() { - if boundaries.contains(&Rc::as_ptr(&node)) || !seen.insert(Rc::as_ptr(&node)) { - continue; - } - if matches!(node.asap(), Some(ASAPOp::SummaryAgg {family, ..} | ASAPOp::SummaryTumblingWindowChain {family, ..}) if !matches!(family, FieldDataType::ExactAggregate(..))) - { - exact = false; - break; - } - pending.extend(node.children().into_iter().cloned()); - } + let boundaries = boundary_ptrs(memo, &candidate); + let mut exact = !nodes_within(&candidate.root_node, &boundaries).iter().any(|node| { + matches!(node.asap(), Some(ASAPOp::SummaryAgg {family, ..} | ASAPOp::SummaryTumblingWindowChain {family, ..}) if !matches!(family, FieldDataType::ExactAggregate(..))) + }); if !exact { continue; } diff --git a/crates/logical-asap-optimization/src/pass2/memo/shared_configurations.rs b/crates/logical-asap-optimization/src/pass2/memo/shared_configurations.rs index aa5682be1..9eb61a2db 100644 --- a/crates/logical-asap-optimization/src/pass2/memo/shared_configurations.rs +++ b/crates/logical-asap-optimization/src/pass2/memo/shared_configurations.rs @@ -1,4 +1,5 @@ //! Finite state configurations shared by memo rules and candidate selection. +use crate::stage0::cse::same_value; use asap_accuracy::estimators::size_for_all_readers; use asap_types::ir::schema::{ sketch::{SketchKind, SketchParams}, @@ -44,7 +45,7 @@ impl TumblingWindowConfiguration { .is_multiple_of(self.tumbling_window_width_ms) && reader.window_ms <= self.longest_window_ms && shared_size(&self.family, &reader.family) - .is_some_and(|joined| same_family(&joined, &self.family)) + .is_some_and(|joined| same_value(&joined, &self.family)) } } /// Every positive integer-millisecond divisor of a grid, in ascending order. @@ -92,7 +93,7 @@ pub(crate) fn tumbling_window_configurations( // The sizes: the readers' own, closed under the shared size. let mut sizes: Vec = Vec::new(); for reader in &eligible { - if !sizes.iter().any(|known| same_family(known, &reader.family)) { + if !sizes.iter().any(|known| same_value(known, &reader.family)) { sizes.push(reader.family.clone()); } } @@ -100,7 +101,7 @@ pub(crate) fn tumbling_window_configurations( while i < sizes.len() { for j in 0..i { if let Some(joined) = shared_size(&sizes[i], &sizes[j]) { - if !sizes.iter().any(|known| same_family(known, &joined)) { + if !sizes.iter().any(|known| same_value(known, &joined)) { sizes.push(joined); } } @@ -126,9 +127,7 @@ pub(crate) fn tumbling_window_configurations( })) }); if window == Some(longest_window_ms) - && size - .flatten() - .is_some_and(|size| same_family(&size, family)) + && size.flatten().is_some_and(|size| same_value(&size, family)) { configurations.push(configuration); } @@ -198,12 +197,6 @@ pub(crate) fn shared_size(a: &FieldDataType, b: &FieldDataType) -> Option bool { - left == right && serde_json::to_string(left).ok() == serde_json::to_string(right).ok() -} - #[cfg(test)] mod tests { use super::*; 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 4cee7ca54..1d3645026 100644 --- a/crates/logical-asap-optimization/src/pass2/memo/state_groups.rs +++ b/crates/logical-asap-optimization/src/pass2/memo/state_groups.rs @@ -27,8 +27,6 @@ pub enum Pass2RuleError { CyclicDependencies, #[error("explicit summary-state groups have already been introduced")] AlreadyExpanded, - #[error("a previously discovered reader no longer exposes its summary boundary")] - InvalidReaderBoundary, #[error("scan-label candidate expansion: {0}")] ScanLabelWidening(String), #[error("state configuration expansion: {0}")] @@ -340,6 +338,37 @@ pub(super) fn boundary_groups( found } +/// The nodes of `candidate`'s input groups, where its own nodes end. +pub(super) fn boundary_ptrs( + memo: &Memo, + candidate: &CandidateForMemoGroup, +) -> HashSet<*const OperatorNode> { + candidate + .input_groups + .iter() + .map(|group| Rc::as_ptr(&memo.groups[group.0 as usize].logical_node)) + .collect() +} + +/// `root` and the nodes beneath it down to `boundaries` (excluded), each +/// once, in depth-first order (last child first). +pub(super) fn nodes_within( + root: &Rc, + boundaries: &HashSet<*const OperatorNode>, +) -> Vec> { + let mut pending = vec![Rc::clone(root)]; + let mut seen = HashSet::new(); + let mut nodes = Vec::new(); + while let Some(node) = pending.pop() { + if boundaries.contains(&Rc::as_ptr(&node)) || !seen.insert(Rc::as_ptr(&node)) { + continue; + } + pending.extend(node.children().into_iter().cloned()); + nodes.push(node); + } + nodes +} + // Hashing narrows comparisons only; full strict equality below remains the // authority, including probability metadata and floating-point corner cases. #[derive(Hash, PartialEq, Eq)] diff --git a/crates/logical-asap-optimization/src/stage0/cse.rs b/crates/logical-asap-optimization/src/stage0/cse.rs index 4ffc86930..6f209a4e6 100644 --- a/crates/logical-asap-optimization/src/stage0/cse.rs +++ b/crates/logical-asap-optimization/src/stage0/cse.rs @@ -131,8 +131,9 @@ pub(crate) fn structural_hash(node: &OperatorNode, cache: &mut NodeToStructuralH /// Numeric `PartialEq` alone conflates signed zeros. The serialized check is /// additional evidence, never a replacement for typed equality (JSON maps /// non-finite floats to `null`). Used for the guarantee, whose bounds are -/// floats a shared node must preserve bit-for-bit. -fn same_value(left: &T, right: &T) -> bool { +/// floats a shared node must preserve bit-for-bit, and for summary families, +/// whose `SketchParams` equality omits the recorded error and probability. +pub(crate) 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,