[milindsrivastava1997]

Why

Part of #1096 (Stage 1 duplicate and dead code). Stacked on #1192; review only the top commit.

What

• same_family: two identical copies (pass2/memo/shared_configurations.rs, pass2/memo/rules/shared_configuration_readers.rs) are replaced by stage0::cse::same_value (typed equality plus serialized equality). The third copy, in asap_aware_cse.rs, means "can be one build" (same algorithm and grouping at any size), so it is renamed same_build_family.
• Helpers in pass2/memo/state_groups.rs:
  • boundary_ptrs(memo, candidate): the nodes of a candidate's input groups. Replaces 4 copies (shared_configuration_readers, summary_inputs, summary_results, topk).
  • nodes_within(root, boundaries): the candidate's own nodes, depth-first, each once. Replaces 3 copies of the walk (summary_inputs, summary_results, topk). direct_input_builds keeps its own walk: it keys boundaries by group.
• Dead code: Pass2RuleError::InvalidReaderBoundary (never constructed) and the test-only may_change_rollup_candidate, along with the test assertions that exercised only that helper.

How

The helpers visit nodes in the same order as the loops they replace, so the candidate order doesn't change. topk's walk used to stop at the first approximate build; it now collects the nodes and then checks them with any. The result is the same, though the walk is no longer cut short.

same_value returns false if serialization fails, while the old same_family returned true when both sides failed. FieldDataType always serializes, so this has no observable effect.

Key code interfaces added / modified

Crate-private only:

pub(super) fn boundary_ptrs(memo: &Memo, candidate: &CandidateForMemoGroup) -> HashSet<*const OperatorNode>;
pub(super) fn nodes_within(root: &Rc<OperatorNode>, boundaries: &HashSet<*const OperatorNode>) -> Vec<Rc<OperatorNode>>;
pub(crate) fn same_value<T: PartialEq + Serialize>(left: &T, right: &T) -> bool; // was private to stage0::cse

Pass2RuleError loses InvalidReaderBoundary (public enum; nothing constructed or matched it).

Major algorithm changes

None.

Before this PR

Each of summary_inputs, summary_results and topk had its own copy of this:

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();
while let Some(node) = pending.pop() {
    if boundaries.contains(&Rc::as_ptr(&node)) || !seen.insert(Rc::as_ptr(&node)) { continue; }
    /* rule-specific */
    pending.extend(node.children().into_iter().cloned());
}

After this PR

let boundaries = boundary_ptrs(memo, &original);
for node in nodes_within(&original.root_node, &boundaries) { /* rule-specific */ }

Snapshot changes

None expected. The corpus snapshots weren't run locally (they're heavy on this machine), so CI is the check.

Verification

• cargo test -p asap-logical-asap-optimization --all-targets: 207 passed, 1 ignored; pass_layering 1 passed.
• cargo test -p asap-integration-tests --test integration -- memo_enumeration reuse_rules stage1_logical_asap_optimization sharing_issue_regressions rollup_controlled tumbling_window_merge finer_aggregate_reuse operator_sharing: 76 passed, 21 ignored.
• cargo clippy -p asap-logical-asap-optimization --all-targets: 0 warnings. cargo fmt: clean.
• Pre-commit hooks were skipped locally (memory-heavy on this machine); relying on CI.

🤖 Generated with Claude Code

[milindsrivastava1997]

Automated code review (Claude Code); findings are unverified.

• crates/logical-asap-optimization/src/pass2/memo/shared_configurations.rs:185: Behaviour change: stage0::cse::same_value returns false when serialization fails, whereas the old same_family compared to_string(..).ok() == to_string(..).ok() and returned true when both sides failed. For such a type, serves_size / TumblingWindowConfiguration::serves now return false and the sizes dedup in tumbling_window_configurations_before can push duplicates. Not a pure de-duplication.
• crates/logical-asap-optimization/src/pass2/memo/rules/topk.rs:293: exact_input_group now collects every node within the boundaries into a Vec before calling .any(); the old loop short-circuited at the first non-exact summary. That is extra walking and allocation on a deadline-guarded path, repeated on every recursive call. An early-exit walk would keep the old cost.
• crates/logical-asap-optimization/src/pass2/memo/state_groups.rs:391: nodes_within has no check_generation_deadline, but the inlined loops it replaces in factor_summary_inputs / factor_summary_results checked the deadline on every popped node. On a large DAG the whole traversal now runs before any deadline check.
• crates/logical-asap-optimization/src/pass2/memo/state_groups.rs:338: Partial de-duplication: boundary_groups and direct_input_builds still hand-roll the same pending/seen DFS, and boundary-map construction is still inline (~line 183, and in tests ~1531, ~1610). These can drift from nodes_within; child order already differs (boundary_groups pushes children reversed).