refactor(stage1): one same_family, boundary_ptrs/nodes_within helpers, dead Pass 2 code (#1096)

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).

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

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,
 };

crates/logical-asap-optimization/src/pass2/memo/groups.rs → 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<Vec<QueryRoot>, crate::pass1::per_computation_candidates::PerComputationCandidateError>
-{
+) -> Result<Vec<QueryRoot>, super::per_computation_candidates::PerComputationCandidateError> {
     let mut kept: Vec<usize> = 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],

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

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<OperatorNode>, root: &Rc<OperatorNode>) -> 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<usize> {
         // 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

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;

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<usize, Pas
 
 #[cfg(test)]
 mod tests {
+    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::{lower_promql, sharing_rules_and_shared_state_groups};
     use asap_types::ir::properties::AccuracyTarget;

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<TumblingWindowRequi
 /// Whether a build of `family` serves a reader of `reader`: the size both
 /// share is `family` itself.
 fn serves_size(family: &FieldDataType, reader: &FieldDataType) -> 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<StateNeed>]) -> 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();
             }
         }
@@ -503,8 +498,8 @@ fn compatible(left: &StateNeed, right: &StateNeed) -> 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;

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();

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<usize, Pass2Rule
         }
         let build_pointers: HashSet<_> =
             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<usize, Pass2Rule
                     readouts.push(Rc::clone(&node));
                 }
             }
-            pending.extend(node.children().into_iter().cloned());
         }
         if readouts.is_empty() {
             continue;