From ec1af7ee09df16210dd2875c4abb576a462fb4fe Mon Sep 17 00:00:00 2001
From: zzylol <50204836+zzylol@users.noreply.github.com>
Date: Thu, 5 Nov 2026 05:08:46 +0000
Subject: [PATCH 1/2] refactor(stage1): top-k reuse and finer-aggregate reads
 as one operator over groups

Rules 4 (top-k reuse) and 5 (finer aggregate) add their reader copies
through `add_candidate_over_groups`, so the `Limit` of a top-k read and
the compensating `Filter` of a finer-aggregate read are groups. Top-k's
sort and ranking producers are found or added with Stage 0's equality
(`find_or_add_producer`); its exact-only input groups through
`Memo::find_or_add_group`, their exact guarantee keeping them apart from
the group they copy.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K1M6br9PL4SDeVS1XpezKJ
---
 .../src/pass2/memo/rules/finer_aggregate.rs   |  4 +--
 .../src/pass2/memo/rules/topk.rs              | 31 ++++++++++---------
 2 files changed, 18 insertions(+), 17 deletions(-)

diff --git a/crates/logical-asap-optimization/src/pass2/memo/rules/finer_aggregate.rs b/crates/logical-asap-optimization/src/pass2/memo/rules/finer_aggregate.rs
index 64da2407d..dbc710605 100644
--- a/crates/logical-asap-optimization/src/pass2/memo/rules/finer_aggregate.rs
+++ b/crates/logical-asap-optimization/src/pass2/memo/rules/finer_aggregate.rs
@@ -1,7 +1,7 @@
 //! Read a SQL aggregate from a finer aggregate's result
 //! ([`match_finer`](crate::pass2::finer_aggregate::match_finer)): filter the finer
 //! rows, then merge them into the query's groups with each measure's rollup.
-use super::reuse::append_reader;
+use super::reuse::add_candidate_over_groups;
 use crate::pass2::finer_aggregate::match_finer;
 use crate::pass2::memo::state_groups::Pass2RuleError;
 use asap_types::ir::operator::{AggIntent, Reduction};
@@ -167,7 +167,7 @@ fn add_read(memo: &mut Memo, owner: usize, finer: usize) -> Result<usize, Pass2R
         return Ok(0);
     }
     let reader = Rc::new(OperatorNode::with_schema(reader.operator.clone(), schema));
-    Ok(usize::from(append_reader(
+    Ok(usize::from(add_candidate_over_groups(
         memo,
         owner,
         &original,
diff --git a/crates/logical-asap-optimization/src/pass2/memo/rules/topk.rs b/crates/logical-asap-optimization/src/pass2/memo/rules/topk.rs
index 6f7c502f8..0be8b198d 100644
--- a/crates/logical-asap-optimization/src/pass2/memo/rules/topk.rs
+++ b/crates/logical-asap-optimization/src/pass2/memo/rules/topk.rs
@@ -1,6 +1,6 @@
 //! Exact ranking reuse with explicit exact-input groups. No global selection
 //! flags are required: only exact implementations can supply these inputs.
-use super::reuse::{append_producer, append_reader, eligible, replace};
+use super::reuse::{add_candidate_over_groups, eligible, find_or_add_producer, replace};
 use super::shared_producer::{add_shared_producer_candidates, Reader, SharedProducer};
 use crate::pass2::asap_aware_cse::topk_reuse_candidates;
 use crate::pass2::memo::state_groups::{boundary_ptrs, nodes_within, Pass2RuleError};
@@ -123,25 +123,25 @@ impl SharedProducer for TopKRankings {
         let sort = covering.ranking.children()[0]
             .with_new_children(|_| Rc::clone(&input))
             .map_err(|error| Pass2RuleError::ReuseRule(error.to_string()))?;
-        let sort_group = append_producer(
+        let sort_group = find_or_add_producer(
             memo,
-            Rc::new(sort),
+            &Rc::new(sort),
             &queries,
-            OriginOfCandidateForMemoGroup::TopKReuse,
+            &OriginOfCandidateForMemoGroup::TopKReuse,
             1,
-        )?;
+        );
         let sort = Rc::clone(&memo.groups()[sort_group.0 as usize].logical_node);
         let ranking = covering
             .ranking
             .with_new_children(|_| Rc::clone(&sort))
             .map_err(|error| Pass2RuleError::ReuseRule(error.to_string()))?;
-        let ranking_group = append_producer(
+        let ranking_group = find_or_add_producer(
             memo,
-            Rc::new(ranking),
+            &Rc::new(ranking),
             &queries,
-            OriginOfCandidateForMemoGroup::TopKReuse,
+            &OriginOfCandidateForMemoGroup::TopKReuse,
             1,
-        )?;
+        );
         let ranking = Rc::clone(&memo.groups()[ranking_group.0 as usize].logical_node);
         let owner = reader.group;
         let logical = Rc::clone(&memo.groups()[owner].logical_node);
@@ -176,7 +176,7 @@ impl SharedProducer for TopKRankings {
             child: limited,
         }))
         .map_err(|error| Pass2RuleError::ReuseRule(error.to_string()))?;
-        Ok(usize::from(append_reader(
+        Ok(usize::from(add_candidate_over_groups(
             memo,
             owner,
             &original,
@@ -251,13 +251,14 @@ fn exact_input_group(
         AccuracyGuarantee::exact("only exact implementations and exact inputs"),
     )));
     let queries = memo.groups()[group.0 as usize].queries.clone();
-    let id = append_producer(
-        memo,
-        node,
+    // The exact guarantee is the group's identity: it keeps the group apart
+    // from `group`, whose candidates may be approximate.
+    let id = memo.find_or_add_group(
+        &node,
         &queries,
-        OriginOfCandidateForMemoGroup::TopKReuse,
+        &OriginOfCandidateForMemoGroup::TopKReuse,
         1,
-    )?;
+    );
     for candidate in &mut candidates {
         candidate.fused_groups = Vec::new();
     }

From a81ad4e7ac273e37d8390e27e045f69a5ebd5620 Mon Sep 17 00:00:00 2001
From: zzylol <50204836+zzylol@users.noreply.github.com>
Date: Thu, 5 Nov 2026 16:00:47 +0000
Subject: [PATCH 2/2] refactor(stage1): review fixes for #1220: merge an exact
 input group's candidates when found

`exact_input_group` replaced the candidates of the group `find_or_add_group`
returned, also when it found an existing group. Now the exact candidates
replace the own candidate only of a group it added; a found group keeps its
candidates and gets the new ones through `push_candidate_if_new`. Top-k and
finer-aggregate reads follow #1218's interface changes (`MemoGroupId` owner,
`find_or_add_producer` without budget parts).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K1M6br9PL4SDeVS1XpezKJ
---
 .../src/pass2/memo/rules/finer_aggregate.rs   |  4 ++--
 .../src/pass2/memo/rules/topk.rs              | 20 ++++++++++++++-----
 2 files changed, 17 insertions(+), 7 deletions(-)

diff --git a/crates/logical-asap-optimization/src/pass2/memo/rules/finer_aggregate.rs b/crates/logical-asap-optimization/src/pass2/memo/rules/finer_aggregate.rs
index dbc710605..2712061ee 100644
--- a/crates/logical-asap-optimization/src/pass2/memo/rules/finer_aggregate.rs
+++ b/crates/logical-asap-optimization/src/pass2/memo/rules/finer_aggregate.rs
@@ -6,7 +6,7 @@ use crate::pass2::finer_aggregate::match_finer;
 use crate::pass2::memo::state_groups::Pass2RuleError;
 use asap_types::ir::operator::{AggIntent, Reduction};
 use asap_types::ir::{NonASAPOp, Operator, OperatorNode, OperatorResultKind};
-use asap_types::stages::{Memo, OriginOfCandidateForMemoGroup};
+use asap_types::stages::{Memo, MemoGroupId, OriginOfCandidateForMemoGroup};
 use std::collections::HashMap;
 use std::rc::Rc;
 
@@ -169,7 +169,7 @@ fn add_read(memo: &mut Memo, owner: usize, finer: usize) -> Result<usize, Pass2R
     let reader = Rc::new(OperatorNode::with_schema(reader.operator.clone(), schema));
     Ok(usize::from(add_candidate_over_groups(
         memo,
-        owner,
+        MemoGroupId(owner as u32),
         &original,
         reader,
         OriginOfCandidateForMemoGroup::Rollup,
diff --git a/crates/logical-asap-optimization/src/pass2/memo/rules/topk.rs b/crates/logical-asap-optimization/src/pass2/memo/rules/topk.rs
index 0be8b198d..1c1fe1b95 100644
--- a/crates/logical-asap-optimization/src/pass2/memo/rules/topk.rs
+++ b/crates/logical-asap-optimization/src/pass2/memo/rules/topk.rs
@@ -2,6 +2,7 @@
 //! flags are required: only exact implementations can supply these inputs.
 use super::reuse::{add_candidate_over_groups, eligible, find_or_add_producer, replace};
 use super::shared_producer::{add_shared_producer_candidates, Reader, SharedProducer};
+use crate::find_or_add_group::find_or_add_group;
 use crate::pass2::asap_aware_cse::topk_reuse_candidates;
 use crate::pass2::memo::state_groups::{boundary_ptrs, nodes_within, Pass2RuleError};
 use asap_types::ir::operator::{AggIntent, Reduction};
@@ -128,7 +129,6 @@ impl SharedProducer for TopKRankings {
             &Rc::new(sort),
             &queries,
             &OriginOfCandidateForMemoGroup::TopKReuse,
-            1,
         );
         let sort = Rc::clone(&memo.groups()[sort_group.0 as usize].logical_node);
         let ranking = covering
@@ -140,7 +140,6 @@ impl SharedProducer for TopKRankings {
             &Rc::new(ranking),
             &queries,
             &OriginOfCandidateForMemoGroup::TopKReuse,
-            1,
         );
         let ranking = Rc::clone(&memo.groups()[ranking_group.0 as usize].logical_node);
         let owner = reader.group;
@@ -178,7 +177,7 @@ impl SharedProducer for TopKRankings {
         .map_err(|error| Pass2RuleError::ReuseRule(error.to_string()))?;
         Ok(usize::from(add_candidate_over_groups(
             memo,
-            owner,
+            MemoGroupId(owner as u32),
             &original,
             reader,
             OriginOfCandidateForMemoGroup::TopKReuse,
@@ -253,7 +252,9 @@ fn exact_input_group(
     let queries = memo.groups()[group.0 as usize].queries.clone();
     // The exact guarantee is the group's identity: it keeps the group apart
     // from `group`, whose candidates may be approximate.
-    let id = memo.find_or_add_group(
+    let groups_before = memo.groups().len();
+    let id = find_or_add_group(
+        memo,
         &node,
         &queries,
         &OriginOfCandidateForMemoGroup::TopKReuse,
@@ -262,7 +263,16 @@ fn exact_input_group(
     for candidate in &mut candidates {
         candidate.fused_groups = Vec::new();
     }
-    memo.candidates[id.0 as usize] = candidates;
+    if id.0 as usize >= groups_before {
+        // Added: the exact candidates replace the group's own candidate.
+        memo.candidates[id.0 as usize] = candidates;
+    } else {
+        // Found (another group's exact input equals it): keep its
+        // candidates and add the new ones.
+        for candidate in candidates {
+            memo.add_candidate(id, candidate);
+        }
+    }
     known.insert(group, Some(id));
     Ok(Some(id))
 }