refactor(stage1): scan widening as one operator over groups; one exact-guarantee strip (#1203 2a-5)

crates/logical-asap-optimization/src/operator_over_groups.rs

@@ -16,7 +16,8 @@ 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
+/// ([`without_exact_guarantees`]; its own guarantee is the candidate's and
+/// is kept), 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
@@ -29,24 +30,36 @@ pub(crate) fn operator_over_groups(
     origin: &OriginOfCandidateForMemoGroup,
     accuracy_budget_parts: usize,
 ) -> (Rc<OperatorNode>, Vec<MemoGroupId>) {
-    let root = without_exact_guarantees_below(root);
+    let stripped = without_exact_guarantees(root);
+    let root = if stripped
+        .children()
+        .iter()
+        .zip(root.children())
+        .all(|(a, b)| Rc::ptr_eq(a, b))
+    {
+        Rc::clone(root)
+    } else {
+        Rc::new(OperatorNode {
+            guarantee: root.guarantee.clone(),
+            ..(*stripped).clone()
+        })
+    };
     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
+/// `node` with every exact guarantee removed, its own included. 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<OperatorNode>) -> Rc<OperatorNode> {
+/// compares guarantees). An approximate guarantee is kept, so builds that
+/// differ only in their failure probability or guarantee stay different
+/// groups.
+pub(crate) fn without_exact_guarantees(node: &Rc<OperatorNode>) -> Rc<OperatorNode> {
     fn without_exact(
         node: &Rc<OperatorNode>,
         done: &mut HashMap<*const OperatorNode, Rc<OperatorNode>>,
@@ -76,20 +89,5 @@ pub(crate) fn without_exact_guarantees_below(root: &Rc<OperatorNode>) -> Rc<Oper
         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()
-    })
+    without_exact(node, &mut HashMap::new())
 }

crates/logical-asap-optimization/src/pass1/add_pass1_candidates.rs

@@ -689,7 +689,7 @@ mod tests {
             },
         ))
         .unwrap();
-        let stripped = crate::operator_over_groups::without_exact_guarantees_below(&reader);
+        let stripped = crate::operator_over_groups::without_exact_guarantees(&reader);
         assert_eq!(stripped.children()[0].guarantee, Some(guarantee));
     }
 

crates/logical-asap-optimization/src/pass2/memo/rules/reuse.rs

@@ -2,7 +2,7 @@
 //! ([`eligible`]), and how they add what they derive, as one operator over
 //! groups ([`add_candidate_over_groups`], [`find_or_add_producer`]).
 use crate::find_or_add_group::find_or_add_group;
-use crate::operator_over_groups::{operator_over_groups, without_exact_guarantees_below};
+use crate::operator_over_groups::{operator_over_groups, without_exact_guarantees};
 use crate::pass2::memo::state_groups::Pass2RuleError;
 use asap_types::ir::OperatorNode;
 use asap_types::stages::MemoGroupId;
@@ -88,16 +88,7 @@ pub(in crate::pass2::memo) fn find_or_add_producer(
     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)
+    find_or_add_group(memo, &without_exact_guarantees(node), queries, origin, 1)
 }
 
 pub(in crate::pass2::memo) fn replace(

crates/logical-asap-optimization/src/pass2/memo/rules/scan_labels.rs

@@ -1,19 +1,16 @@
 //! Scan-label widening for explicit memo state sharing. Reuses the existing
 //! Pass 2 widening rule; only its source-chain boundaries are interned here.
-use std::collections::{HashMap, HashSet};
 use std::rc::Rc;
 
 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 super::reuse::{add_candidate_over_groups, 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,
-};
-use asap_types::stages::{CandidateForMemoGroup, Memo, OriginOfCandidateForMemoGroup};
-use asap_types::stages::{MemoGroup, MemoGroupId};
+use crate::operator_over_groups::without_exact_guarantees;
+use crate::pass2::memo::state_groups::{build_down_to_stage0_groups, Pass2RuleError};
+use asap_types::stages::{Memo, MemoGroupId, OriginOfCandidateForMemoGroup};
 
 /// Preserve local implementations and append readers over union-label inputs.
 /// Parent/query reachability is rebuilt by the enclosing state-group expansion.
@@ -83,53 +80,49 @@ pub(crate) fn add_scan_label_candidates(memo: &mut Memo) -> Result<usize, Pass2R
     let unwidened = recompose_guarantees(&roots)
         .map_err(|error| Pass2RuleError::ScanLabelWidening(error.to_string()))?;
     let mut added = 0;
-    let mut canonical = HashMap::new();
     for (((group, index), root), unwidened) in owners.into_iter().zip(widened).zip(unwidened) {
         let (QueryRoot::Operator(node), QueryRoot::Operator(unwidened)) = (root, unwidened) else {
             unreachable!("operator readers")
         };
         let original = memo.candidates[group][index].clone();
-        if *node == *unwidened || !same_output_contract(&original.root_node, &node) {
+        if *node == *unwidened {
             continue;
         }
-        let queries = memo.groups()[group].queries.clone();
-        let mut node = intern_inputs(&node, memo, &queries, &mut canonical)?;
-        if memo.group_whose_node_is(&node).is_some() {
-            // Interning an equivalent range/source can return another group's
-            // canonical value. Keep a concrete reader instead of a root alias.
-            node =
-                OperatorNode::new_shared(asap_types::ir::Operator::NonASAP(NonASAPOp::Project {
-                    cols: node
-                        .schema
-                        .fields
-                        .iter()
-                        .enumerate()
-                        .map(|(column, field)| asap_types::ir::ProjectItem {
-                            alias: Some(field.name.clone()),
-                            expr: asap_types::ir::ScalarExpr::Column(column),
-                        })
-                        .collect(),
-                    qualifier: None,
-                    child: node,
-                }))
-                .map_err(|error| Pass2RuleError::ScanLabelWidening(error.to_string()))?;
-            if !same_output_contract(&original.root_node, &node) {
-                continue;
-            }
-        }
-        let mut reader = original;
-        reader.root_node = node;
-        reader.input_groups = boundary_groups(memo, &reader.root_node, |_| true);
-        reader.origin = OriginOfCandidateForMemoGroup::WidenedScans;
-        if memo.candidates[group].iter().any(|known| {
-            known.origin == OriginOfCandidateForMemoGroup::WidenedScans
-                && known.input_groups == reader.input_groups
-                && *known.root_node == *reader.root_node
-        }) {
+        // A widened source chain equal to a group's node is read through a
+        // projection instead of aliasing that group.
+        let node = if memo
+            .group_with_equal_node(&without_exact_guarantees(&node))
+            .is_some()
+        {
+            OperatorNode::new_shared(asap_types::ir::Operator::NonASAP(NonASAPOp::Project {
+                cols: node
+                    .schema
+                    .fields
+                    .iter()
+                    .enumerate()
+                    .map(|(column, field)| asap_types::ir::ProjectItem {
+                        alias: Some(field.name.clone()),
+                        expr: asap_types::ir::ScalarExpr::Column(column),
+                    })
+                    .collect(),
+                qualifier: None,
+                child: node,
+            }))
+            .map_err(|error| Pass2RuleError::ScanLabelWidening(error.to_string()))?
+        } else {
+            node
+        };
+        // A reader whose output the widening changed is not added.
+        if !same_output_contract(&original.root_node, &node) {
             continue;
         }
-        memo.candidates[group].push(reader);
-        added += 1;
+        added += usize::from(add_candidate_over_groups(
+            memo,
+            MemoGroupId(group as u32),
+            &original,
+            node,
+            OriginOfCandidateForMemoGroup::WidenedScans,
+        )?);
     }
     Ok(added)
 }
@@ -143,85 +136,3 @@ fn is_source_chain(node: &OperatorNode) -> bool {
         _ => false,
     }
 }
-
-fn same_source_node(left: &OperatorNode, right: &OperatorNode) -> bool {
-    // These operators preserve exact source rows; recomposition's exact
-    // guarantee annotation does not change their logical/execution contract.
-    left.operator.map_children(|_| ()) == right.operator.map_children(|_| ())
-        && left.schema == right.schema
-        && left.result_kind == right.result_kind
-        && left.timing == right.timing
-        && left.materialization == right.materialization
-        && left
-            .children()
-            .iter()
-            .zip(right.children())
-            .all(|(a, b)| Rc::ptr_eq(a, b))
-        && left.children().len() == right.children().len()
-}
-
-fn intern_inputs(
-    node: &Rc<OperatorNode>,
-    memo: &mut Memo,
-    queries: &[usize],
-    done: &mut HashMap<*const OperatorNode, Rc<OperatorNode>>,
-) -> Result<Rc<OperatorNode>, Pass2RuleError> {
-    if let Some(known) = done.get(&Rc::as_ptr(node)) {
-        return Ok(Rc::clone(known));
-    }
-    let mut children = HashMap::new();
-    for child in node.children() {
-        children.insert(
-            Rc::as_ptr(child),
-            intern_inputs(child, memo, queries, done)?,
-        );
-    }
-    let rewritten = Rc::new(OperatorNode {
-        operator: node
-            .operator
-            .map_children(|child| Rc::clone(&children[&Rc::as_ptr(child)])),
-        ..(**node).clone()
-    });
-    let result = if is_source_chain(&rewritten) {
-        if let Some(existing) = memo
-            .groups()
-            .iter()
-            .find(|group| same_source_node(&group.logical_node, &rewritten))
-        {
-            Rc::clone(&existing.logical_node)
-        } else {
-            // The new group's id must fit a `MemoGroupId`.
-            u32::try_from(memo.groups().len()).map_err(|_| Pass2RuleError::TooManyGroups)?;
-            let mut seen = HashSet::new();
-            let inputs: Vec<_> = rewritten
-                .children()
-                .iter()
-                .map(|child| {
-                    memo.group_whose_node_is(child)
-                        .expect("source-chain children were interned")
-                })
-                .filter(|input| seen.insert(*input))
-                .collect();
-            memo.push_group(
-                MemoGroup {
-                    logical_node: Rc::clone(&rewritten),
-                    child_groups: inputs.clone(),
-                    parents: Vec::new(),
-                    queries: queries.to_vec(),
-                },
-                vec![CandidateForMemoGroup {
-                    fused_groups: Vec::new(),
-                    root_node: Rc::clone(&rewritten),
-                    input_groups: inputs,
-                    origin: OriginOfCandidateForMemoGroup::WidenedScans,
-                    accuracy_budget_parts: 1,
-                }],
-            );
-            rewritten
-        }
-    } else {
-        rewritten
-    };
-    done.insert(Rc::as_ptr(node), Rc::clone(&result));
-    Ok(result)
-}