[zzylol]

Part of #1203 (PR 2a-1), tracked in #1207.

Merge unit. This PR merges into main only together with the rest of #1207's merge unit (steps 1a–6, #1207 "Order"): interim items below are deleted by a later PR in the unit.

Why

#1203 decided "one operator per candidate, no private nodes": every intermediate result of a way to compute is its own memo group, as in Cascades. Today a Pass 1 candidate's root_node holds a private chain, e.g. Project ← SummaryEstimate ← SummaryAgg Kll(k=519) over the range group, so two queries' equal KLL builds are two private nodes that Pass 2's shared-build step must find and factor afterwards. This PR is the Stage 1 start of 2a: Pass 1 stops making private chains.

What

• Pass 1 finds or adds a memo group for every node beneath a composed candidate's root (find_or_add_group, Stage 1), with Stage 0's node equality (one HashConsingTable shared by Stage 0, the memo and the shared summary-state step). Each Pass 1 candidate is its top operator over groups.
  • Example (quantile_over_time(0.95, latency[1m]) and quantile_over_time(0.99, latency[1m]) / quantile_over_time(0.50, latency[1m]), max_error 0.01): Project ← SummaryEstimate ← SummaryAgg Kll(k=519) is now three groups, and the queries' k=519 builds are one group (G6 below).
• Pass 1 sizes summaries for their consumers' failure probability before the lookup, so each group holds its sized node; it then removes the exact guarantee composition attaches to exact nodes (see "Equality").
• The Pass 2 rules that matched Pass 1's private builds find them as groups instead (smallest set needed for coherence, see "Why not Pass 1 alone").
• Plans unchanged: every corpus_snapshots and sketch_plan_snapshots snapshot keeps its selected plan and cost; Stage 2's physical-candidate count is the same or lower on every snapshot.

Equality

find_or_add_group and Memo::group_with_equal_node use Stage 0's equality (asap_types::stages::hash_consing: operator, schema, result kind, guarantee, timing, materialization and children, gated by shareable_when_equal). Two consequences:

• Stage 0's nodes carry no guarantee, while composition gives every exact node it rebuilds the exact guarantee ("exact over exact inputs"). Without a fix that difference alone split a composed Scan/TimeRange from Stage 0's group (duplicate G6: Scan, G7: TimeRange 60s in the example). Pass 1 therefore removes only exact guarantees below a composed root before the lookup (without_exact_guarantees_below, one function); an approximate guarantee stays part of equality, so builds of equal k but another δ or guarantee stay different groups (test equal_k_builds_with_another_failure_probability_or_guarantee_are_other_groups).
• Rule 9 rebuilds a resized build over the build group's own inputs (restored by node identity), so its resized builds also match existing groups.
• Duplicate children: a candidate's input_groups are by child position, like a group's child_groups: input i is child i's group, so x / x lists x's group twice (over_child_groups).

Why not Pass 1 alone

Switching Pass 1 alone broke 12 integration tests and 23 unit tests: Pass 2 rules no longer found Pass 1's builds, so rollups, finer-aggregate state merges, scan-widened shared KLL builds, shared UnivMon at the joint size and shared builds at a covering size disappeared from the memo. So this PR also makes the smallest rule changes that keep those candidates:

Rule	Change
1 local window composition	Rewrites only Stage 0's groups (Memo::is_stage0_group: the groups Stage 0 produced from the queries' DAG), through the added groups; Pass 1's intermediate groups are not rewritten on their own.
2 scan widening	Same: widens a Stage 0 group's candidate whose only build (build_down_to_stage0_groups) reads only Stage 0 source groups, reached through added groups.
6 rollup	Also takes a candidate that reads a build group directly (build_of → BuildPlace::InputGroup); the rollup lands on that consumer (the estimate or finalize group).
9 state configurations	Also takes a consumer of a build group (BuildPlace::InputGroup): the resized build's group is found or added (origin SharedBuild: rule 9 has no origin of its own), and the copied consumer reads it.
10 shared summary state	Groups equal private builds with the shared HashConsingTable; a private build equal to an existing group's node is read from that group, whose own candidate is raised to the consumers' most budget parts.
precondition	The "already expanded" check is deleted: Pass 1 now creates summary-state groups.

The rules still emit chains (tumbling windows, widened scans, rollups, configuration consumers, SharedBuild consumers); switching each to one operator over groups is 2a-2..k.

New abstractions not in #1207 or the doc: decisions (2026-11-05)

Per #1207's interface policy, these names are not defined by #1207, #1203 or section 3 of memo-groups-properties-requirements.md. #1207's steps 1a–6 merge into main as one merge unit (#1207, Order), so an interim abstraction may stay in this PR only when a later PR of the unit deletes it; none reaches main. Rows marked Pending approval are not deleted by any PR in the unit and need the user's decision.

Abstraction	What it is for	Decision
private Memo::groups + Memo::new, groups(), groups_mut() (test-support), push_group, private MemoGroupLookup	index every added group for hash-consing	Interim, deleted by 2b (Memo::find_or_add_group(Operator<MemoGroupId>), θ out of the key)
Memo::group_with_equal_node, Memo::add_queries, Memo::rebuild_parents	lookup half of find_or_add_group; keep the derived queries / parents	Interim, deleted by 2b (internals of Memo::find_or_add_group)
Memo::group_whose_node_is	pointer lookup for walks of private nodes	Interim, deleted by 2a-final (no private nodes)
Memo::is_stage0_group + stage0_group_count; added_groups.rs	tell Stage 0's groups from groups Pass 1/2 add	Interim, deleted by 2a-final
crate fns find_or_add_group (node tree), add_group, over_child_groups	find or add a node tree's groups	Interim, deleted by 2b
own_candidate_sizing.rs (size_own_candidate_for, find_or_add_sized_group), the parts argument, AddPass1CandidatesError::Sizing	raise an added group's accuracy_budget_parts	Interim, deleted by #1207 step 6
BuildPlace::{Private, InputGroup}, build_of (was build_read_by), build_down_to_stage0_groups, builds_beneath, input_group_build	where a candidate reads its one summary build	Interim, deleted by 2a-final
BuildPlace::BeneathGroups, StateNeed::place	rule 9's whole-candidate rewrite	Interim, deleted by 2b (rule 9 deleted, #1203 decision 3)
NodeToStructuralIdCache (pre-existing)	Pass 1's own structural identity	Deleted in #1218 (one node equality)
asap_types::stages::hash_consing (HashConsingTable, same_value)	the storage of find_or_add_group; Stage 0's equality	Approved by the user 2026-11-05
renames memo_population → add_pass1_candidates, try_populate_pass1 → try_add_pass1_candidates, CandidatePopulationError → AddPass1CandidatesError, check_unexpanded_memo → check_group_references_valid_and_acyclic	names say what they do	Approved by the user 2026-11-05 (check_group_references_valid_and_acyclic renamed again at the user's request: it checks that group references are in range and consistent and that the memo has no cycle)
existing names with "reader": shared_configuration_readers.rs, BuildReaders	rule 9; step 10's shared builds	Interim, deleted by 2b (rule 9) and 2a-final (step 10)

How

1. find_or_add_group(memo, node, queries, origin, accuracy_budget_parts) -> MemoGroupId (Stage 1, crate-private) returns the existing group whose node equals node (Stage 0's equality), after adding queries to it and every group below it; otherwise it adds a group for node over its children's groups (over_child_groups, which finds or adds each child the same way) and returns that group. It never copies a group. Memo knows nothing of accuracy: Stage 1's find_or_add_sized_group (rule 9, step 10) raises the budget parts of a group that already existed (size_own_candidate_for).
2. Memo.groups is private; every group is added through Memo::new(dag) / push_group, which index it eagerly, so group_whose_node_is (the same Rc) and group_with_equal_node (Stage 0's equality) are &self lookups. A lookup does not keep the node alive: the hash cache holds only the nodes added as groups and their descendants.
3. Pass 1 (add_candidates_to_initial_groups / target_candidates): compose → size (size_for_failure_probability) → remove exact guarantees below the root → over_child_groups → the candidate is the root's operator over those groups; the added groups it reads through are raised to its budget parts. At the end rebuild_parents.
4. Query reach and parents are Memo::add_queries and Memo::rebuild_parents, used by find_or_add_group, Pass 1 and step 10.

Representation / rules / enumeration

• (a) What is stored after this PR. CandidateForMemoGroup { fused_groups, root_node, input_groups, origin, accuracy_budget_parts } is unchanged. Every Pass 1 candidate's root_node is one operator whose children are its input_groups' logical nodes. Pass 2 candidates may still be chains. Memo's groups are private behind groups() and push_group, with a lookup by node (address and Stage 0 equality).
  • CandidateExpression { operator: Operator<MemoGroupId> } is not added yet: Pass 2 rules still emit chains, and Stage 2 reads each candidate's root_node (including its own schema and guarantee). It replaces root_node + input_groups in the PR that deletes private nodes.
  • root_node and input_groups stay: Pass 2 rules and Stage 2's catalog/inventory.rs read them.
• (b) How rules extend the memo. Pass 1 finds or adds intermediates through find_or_add_group. Pass 2 rules are unchanged except the rows in the table above.
• (c) Enumeration. Unchanged: pass2/memo/enumeration.rs and Stage 3 (crates/plan-selection) are not touched by this diff. Stage 2's private-node path and Stage 3's ResultSlot::PrivateNode are still used by the Pass 2 chains; their deletion waits for Physical properties: replace Materialization with timing × lifetime (None / OneEvaluation / AcrossEvaluations) and a schedule for shareable results, with r^phys and α^phys #1200 PR 1.

Key code interfaces added / modified

crates/types/src/stages/candidates.rs

/// The memo (stays).
pub struct Memo {
    /// The groups, by id; read through `groups()`. // was pub
    groups: Vec<MemoGroup>,
    /// Each query's root over groups. // unchanged
    pub query_roots: Vec<QueryRoot<MemoGroupId>>,
    /// Each group's candidates, by group index. // unchanged
    pub candidates: Vec<Vec<CandidateForMemoGroup>>,
    /// Every group by its node's address and by Stage 0's equality. // new, private
    group_lookup: MemoGroupLookup,
    /// How many groups Stage 0 produced from the queries' DAG (ids 0..n). // new, private
    stage0_group_count: usize,
}

impl Memo {
    /// The memo of Stage 0's groups and query roots, no candidates yet. // new
    pub fn new(dag: DagOfLogicalMemoGroups) -> Self;
    /// The groups, by id. // new (was the pub field)
    pub fn groups(&self) -> &[MemoGroup];
    /// The groups to change in a test; not looked up again. // new, test-support only
    pub fn groups_mut(&mut self) -> &mut [MemoGroup];
    /// Adds a group with its candidates, indexed; its id. // new
    pub fn push_group(&mut self, group: MemoGroup, candidates: Vec<CandidateForMemoGroup>) -> MemoGroupId;
    /// The group whose logical node is this `Rc`. // new
    pub fn group_whose_node_is(&self, node: &Rc<OperatorNode>) -> Option<MemoGroupId>;
    /// The group whose node equals `node`: by address, else by Stage 0's equality. // new
    pub fn group_with_equal_node(&self, node: &Rc<OperatorNode>) -> Option<MemoGroupId>;
    /// Whether `group` is one Stage 0 produced from the queries' DAG. // new
    pub fn is_stage0_group(&self, group: MemoGroupId) -> bool;
    /// Adds queries to a group and every group below it. // new
    pub fn add_queries(&mut self, group: MemoGroupId, queries: &[usize]);
    /// Sets every group's parents to the groups that read it. // new
    pub fn rebuild_parents(&mut self);
}

/// Where a candidate came from (stays).
pub enum OriginOfCandidateForMemoGroup {
    /// Pass 1; now also the operator of an intermediate result Pass 1 added as a group. // doc changed
    Pass1,
    /// One build serving several reader groups; also rule 9's resized build groups. // unchanged
    SharedBuild,
    /// A summary readout factored from a composite implementation. // unchanged
    SummaryResult,
    /// Canonical value preprocessing feeding a summary build. // unchanged
    SummaryInput,
    /// Tumbling windows. // unchanged
    TumblingWindows,
    /// A coarse group regrouping a finer group. // unchanged
    Rollup,
    /// A top-k read from another group's ranking. // unchanged
    TopKReuse,
    /// Scans widened to a shared label set. // unchanged
    WidenedScans,
}

crates/types/src/stages/hash_consing.rs (moved from logical-asap-optimization/src/stage0/cse.rs)

/// Hash-consing table: nodes equal by Stage 0's rule find one value. // new (was Stage 0's private InternTable)
pub struct HashConsingTable<V> { /* private */ }
impl<V: Clone> HashConsingTable<V> {
    /// The value of an added node equal to `node`, when sharing it is legal;
    /// caches nothing for `node`. // new
    pub fn find(&self, node: &Rc<OperatorNode>) -> Option<V>;
    /// Adds `node` with `value`. // new
    pub fn insert(&mut self, node: Rc<OperatorNode>, value: V);
}
/// Typed and serialized equality. // moved, was pub(crate)
pub fn same_value<T: PartialEq + serde::Serialize>(left: &T, right: &T) -> bool;
// structural_hash, NodeToStructuralHashCache, shareable_when_equal: moved, private.

crates/logical-asap-optimization/src/find_or_add_group.rs (crate-internal; Stage 1 only, so not on Memo)

/// Returns the existing group whose node equals `node`, else adds one over
/// its children's groups; never copies a group. // new
pub(crate) fn find_or_add_group(memo: &mut Memo, node: &Rc<OperatorNode>, queries: &[usize],
    origin: &OriginOfCandidateForMemoGroup, accuracy_budget_parts: usize) -> MemoGroupId;
/// Adds a group for a node no group's node equals. // new
pub(crate) fn add_group(memo: &mut Memo, node: &Rc<OperatorNode>, queries: &[usize],
    origin: &OriginOfCandidateForMemoGroup, accuracy_budget_parts: usize) -> MemoGroupId;
/// `node`'s operator over its children's found or added groups, and those
/// groups by child position. Used by `find_or_add_group` and Pass 1. // new
pub(crate) fn over_child_groups(memo: &mut Memo, node: &Rc<OperatorNode>, queries: &[usize],
    origin: &OriginOfCandidateForMemoGroup, accuracy_budget_parts: usize)
    -> (Rc<OperatorNode>, Vec<MemoGroupId>);

crates/logical-asap-optimization/src/own_candidate_sizing.rs (crate-internal)

/// Raises a group's own candidate to at least these budget parts. // new
pub(crate) fn size_own_candidate_for(memo: &mut Memo, group: MemoGroupId, accuracy_budget_parts: usize);
/// `find_or_add_group`, raising the parts of a group that already existed; one lookup. // new
pub(crate) fn find_or_add_sized_group(memo: &mut Memo, node: &Rc<OperatorNode>, queries: &[usize],
    origin: &OriginOfCandidateForMemoGroup, accuracy_budget_parts: usize) -> MemoGroupId;

crates/logical-asap-optimization/src/added_groups.rs (interim Stage 1 helpers, re-exported through test_support)

/// The groups some inputs read through the groups added beneath Stage 0's groups. // new
pub struct GroupsReadThroughAddedGroups {
    pub added_groups: Vec<MemoGroupId>,  // new
    pub stage0_groups: Vec<MemoGroupId>, // new
}
/// What inputs read through added groups, stopping at Stage 0's groups (`Memo::is_stage0_group`). // new
pub fn groups_read_through_added_groups(memo: &Memo, inputs: &[MemoGroupId])
    -> GroupsReadThroughAddedGroups;

crates/logical-asap-optimization/src/pass1/add_pass1_candidates.rs (was memo_population.rs)

/// Why adding Pass 1's candidates to the memo failed. // was CandidatePopulationError
pub enum AddPass1CandidatesError {
    /// The inventory walked other roots (bad input). // unchanged
    RootDependenciesChanged,
    /// A target names no measure of an initial aggregate group (bad input). // unchanged
    InvalidTarget { target: usize },
    /// A fusion absorbs a missing target (bad input). // unchanged
    InvalidAbsorption { target: usize, absorbed: usize },
    /// A candidate's summaries, resized for its readers' failure probability,
    /// did not rebuild (internal bug). // new, was PerComputationLogicalMemoCandidatesError::Sizing
    Sizing(SchemaDerivationError),
}

/// Adds Pass 1's candidates to the memo of `initial`'s groups. // was try_populate_pass1
pub fn try_add_pass1_candidates(
    initial: DagOfLogicalMemoGroups,
    per_computation: &PerComputationCandidates,
    accuracy_targets: &[AccuracyTarget], // new
    data: &DataWorkload,                 // new
) -> Result<Memo, AddPass1CandidatesError>;

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

/// Why Pass 1 failed (stays).
pub enum PerComputationLogicalMemoCandidatesError {
    /// A query's DAG cannot be planned. // unchanged
    PerComputation(PerComputationCandidateError),
    /// Adding Pass 1's candidates to the memo failed. // was Population(CandidatePopulationError)
    AddPass1Candidates(AddPass1CandidatesError),
    // Sizing(SchemaDerivationError) // removed: now AddPass1CandidatesError::Sizing
}

crates/logical-asap-optimization/src/pass2/memo/state_groups.rs (crate-internal)

/// Why Pass 2's rules failed (stays).
pub enum Pass2RuleError {
    MismatchedGroups,      // unchanged (internal bug)
    InvalidGroupReference, // unchanged (internal bug)
    TooManyGroups,         // unchanged (unsatisfiable)
    CyclicDependencies,    // unchanged (internal bug)
    // AlreadyExpanded     // removed: Pass 1 now creates summary-state groups
    ScanLabelWidening(String), // unchanged (internal bug)
    StateConfiguration(String), // unchanged (internal bug)
    ReuseRule(String),     // unchanged (internal bug)
}
/// Rejects bad references and cycles. // was check_unexpanded_memo
pub(crate) fn check_group_references_valid_and_acyclic(memo: &Memo) -> Result<(), Pass2RuleError>;
/// Where a candidate reads its one summary build. // new
pub(super) enum BuildPlace {
    Private,                  // one of its private nodes
    InputGroup(MemoGroupId),  // an input group its root reads directly (a build Pass 1 added)
    BeneathGroups,            // beneath the groups it reads, down to Stage 0's groups
}
/// The one summary build a candidate reads, and where: its one private build,
/// else its one `SummaryAgg` input group, else the only build down to Stage 0's
/// groups. The build finder of rules 6, 7 and 9 (was `simple_build`,
/// `input_group_build`). // new
pub(super) fn build_of(memo: &Memo, candidate: &CandidateForMemoGroup)
    -> Option<(Rc<OperatorNode>, BuildPlace)>;
/// The only summary build down to Stage 0's groups, with the Stage 0 groups
/// it reads. Scan widening's lookup and `build_of`'s last look. // new (was scan_labels::build_down_to_workload_groups)
pub(super) fn build_down_to_stage0_groups(memo: &Memo, candidate: &CandidateForMemoGroup)
    -> Option<(Rc<OperatorNode>, Vec<MemoGroupId>)>;
/// The summary builds beneath a root down to the nodes `boundary` gives a
/// group, with the boundary groups each reads. One walk for step 10
/// (`direct_input_builds`) and `build_of`. // new
pub(super) fn builds_beneath(root: &Rc<OperatorNode>,
    boundary: &dyn Fn(&Rc<OperatorNode>) -> Option<MemoGroupId>)
    -> Vec<(Rc<OperatorNode>, Option<Vec<MemoGroupId>>)>;

Usage. One asap_planner run on the example workload (kll.json: the two queries above, max_error 0.01, failure probability 0.001, every 10 s, latency 1000 series every 15 s):

$ asap_planner --workload kll.json --run-until stage3 --format text
# Stage 0 groups G0 Scan, G1 TimeRange 60s, G2..G4 the three quantiles, G5 the ratio.
# Pass 1 (`try_add_pass1_candidates`): each summary candidate is one operator;
# `find_or_add_group` adds G6..G12 and finds G1 for the builds' input.
G2:   Aggregate quantile(q=0.95)                       <- G1
    [0]  Pass1    Aggregate quantile(q=0.95)  inputs=[G1]
    [1]  Pass1    Project                     inputs=[G7]
G3:   Aggregate quantile(q=0.99)                       <- G1
    [1]  Pass1    Project                     inputs=[G8]
    [2]  Pass1    Project                     inputs=[G10] accuracy_budget_parts=2
G6:   SummaryAgg Kll(failure_probability=1e-6 k=519)   <- G1     # one build for G7, G8, G11
G7:   SummaryEstimate Quantile(q=0.95)                 <- G6
G8:   SummaryEstimate Quantile(q=0.99)                 <- G6
G9:   SummaryAgg Kll(failure_probability=5e-7 k=1089)  <- G1     # the half-budget build
G10:  SummaryEstimate Quantile(q=0.99)                 <- G9
# Stage 3 selects the KLL for q1 (G2:c1 over G7 over G6) and exact for q2:
selected choice: G0:c0 G1:c0 G2:c1 G3:c0 G4:c0 G5:c0 G6:c0 G7:c0
cost_per_second: 0.004380

Major algorithm changes

Pass 1 candidate population.

• Problem: a Pass 1 candidate was a private chain down to the first node equal to one of its group's descendants (over_groups); equal builds of different queries were separate private nodes.
• Before: compose → keep the chain above the descendant groups (over_groups) → after population, size every candidate's chain.
• After: compose → size the composed root → remove exact guarantees below it → find or add a group for each child of the root (recursively) → the candidate is the root's operator over those groups.
• Correctness: each added group's node is the composed (sized) node with only exact guarantee annotations removed, so a candidate composed back from its groups is the same plan. Groups are equal by Stage 0's equality; an approximate guarantee or another δ keeps groups apart. A group found by a consumer sized for more budget parts takes the larger parts. Snapshots unchanged; memo groups and candidates of the KLL example and the scan-widening example identical to the first head of this PR (368d34c); tests below.
• Complexity: a lookup is one structural hash (memoized per node) and one equality check per bucket entry. More groups: on the ratio-of-quantiles memo example, Pass 1 has 25 groups / 33 candidates (was 7 / 15) and Pass 2's last step 265 groups / 1237 candidates (was 255 / 1243). Planning time unchanged (below).
• Limits: Pass 2 rules still emit chains (2a-2..k).

Shared summary-state step (10).

• Before: fingerprint buckets (BuildFingerprint) plus same_concrete_build; a new SharedBuild group for each build two or more candidates make.
• After: the shared HashConsingTable groups equal private builds; a build equal to an existing group's node reads that group (its own candidate raised to the consumers' most budget parts); otherwise a group is added through find_or_add_sized_group. The step runs once.

Before this PR

Pass 1 candidates held private operator chains; Pass 2 found their builds inside the chains (rules 6, 9, 10) and factored equal builds into SharedBuild groups at the end, with its own build equality.

After this PR

Pass 1 candidates are one operator over groups, equal Pass 1 builds are one group, and one node equality (Stage 0's) decides group identity everywhere. Pass 2's own chains and Stage 2/3's private-node path are unchanged.

Deleted

• pass1/memo_population.rs (renamed add_pass1_candidates.rs): over_groups, reach, pass1_candidates; try_populate_pass1 / CandidatePopulationError (renamed).
• pass1/logical_memo_candidates.rs: the memo-wide size_for_failure_probability (sizing moved before the lookup) and PerComputationLogicalMemoCandidatesError::Sizing; variant Population renamed AddPass1Candidates.
• stage0/cse.rs: InternTable, structural_hash, same_node, same_value, own_fields, NodeToStructuralHashCache (moved to asap_types::stages::hash_consing).
• pass2/memo/state_groups.rs: BuildFingerprint, local_fingerprint, build_fingerprint, same_concrete_build, add_shared_summary_state_groups_with_index (and its indexed-vs-linear test), the hand-written parent rebuild and query propagation (now Memo methods), Pass2RuleError::AlreadyExpanded; check_unexpanded_memo renamed check_group_references_valid_and_acyclic.
• window_candidates.rs: group_of_node (now Memo::group_whose_node_is); scan_labels.rs: its pointer-to-group map and its own build walk (now Memo::group_whose_node_is / groups_read_through_added_groups / state_groups::builds_beneath, the walk direct_input_builds also uses), and its use of simple_build.
• pass1/add_pass1_candidates.rs: over_found_or_added_groups (now over_child_groups, which find_or_add_group uses too).
• Second review: state_groups::simple_build and input_group_build (now build_of), scan_labels::build_down_to_workload_groups (now state_groups::build_down_to_stage0_groups, shared with build_of); added_groups::workload_groups (now Memo::is_stage0_group); Memo::find_or_add_group / Memo::over_child_groups (now Stage 1 functions); the copied own-fields tuple of hash_consing::looked_up_structural_hash (now structural_hash_with); devtools stage0_memo_groups (inlined).
• GroupFoundOrAdded (find_or_add_group returns the group; find_or_add_sized_group raises parts); Memo::new's separate groups / query roots / candidates arguments (now the Stage 0 DAG) and Pass 1's overwrite of memo.candidates; the RefCell around the memo's hash-consing table.
• Step 10's "already added" check and its test shared_summary_state_step_is_idempotent (step 10 runs once).
• stage0/cse.rs's four structural_hash tests, moved to hash_consing.rs.
• Tests: state_groups::multiple_private_states_can_share_only_the_common_producer (Pass 1 has no private states) replaced by multiple_states_share_only_the_common_state; the added_reader_candidates counter of the state-group fixture.

Plan cost, Stage 2 candidates and planning time

• corpus_snapshots + sketch_plan_snapshots: no snapshot changed (UPDATE_SNAPSHOTS=1 left every file unchanged), so no selected cost changed.
• Stage 2 physical candidates and asap_planner --run-until stage3 wall time per snapshot input, main (82165ae) vs this branch, release builds on one host, best of two runs:

snapshot	Stage 2 candidates, main	this branch	seconds, main	this branch
o11y_bench/q02	248	248	0.11	0.12
o11y_bench/q05	498	498	0.23	0.24
o11y_bench/q06	251	251	0.11	0.11
o11y_bench/q11	7	7	0.01	0.02
o11y_bench/q12	2	2	0.01	0.01
o11y_bench/q18	737	737	0.29	0.29
o11y_bench/q19	495	495	0.25	0.26
o11y_bench/q20	4	4	0.01	0.01
o11y_bench/q21	368	368	0.11	0.12
o11y_bench/q22	247	247	0.10	0.10
o11y_bench/q23	247	247	0.10	0.10
o11y_bench/q25	495	495	0.26	0.26
o11y_bench/q26	247	247	0.10	0.11
o11y_bench/q27	4	4	0.01	0.01
sketch/shared-quantiles-1h-mvp	1801	1801	0.88	0.88
sketch/nested-windows-15m-1h-mvp	2281	2281	1.23	1.22
sketch/ratio-of-quantiles-1h-mvp	1802	1802	2.49	2.50
sketch/quantile-single-mvp	201	201	0.08	0.08
sketch/quantile-grouped-mvp	124	108	0.08	0.06
sketch/topk-rate-guarantees-ignored-mvp	87	87	0.05	0.05
sketch/fuzz-issue903-raw-retained	135	135	0.07	0.07
sketch/random-mixed-seed-862-w02-q2-q7-mvp	152	136	0.11	0.09
sketch/random-mixed-seed-862-w01-q3-q8-mvp	242	226	0.12	0.10
sketch/random-mixed-seed-862-long-w01-q3-mvp	88	72	0.10	0.09
total			6.93	6.90

Same or lower everywhere (#1203's requirement); the four lower inputs lose 16 candidates each because equal builds are one group. Stage 2 candidates are the Cn: lines of asap_planner --run-until stage2 --format text.

Second review fixes

• One build finder. state_groups::build_of(memo, candidate) -> Option<(build, BuildPlace)> replaces simple_build and input_group_build. It looks in the same order the callers did and says where it found the build:

  1. the candidate's one private build;
  2. else its one SummaryAgg input group, which its root reads directly;
  3. else the only build down to Stage 0's groups.

  Each rule keeps the places it handled before: rule 9 and rollups take Private and InputGroup, window rollups Private. Scan widening's lookup (build_down_to_workload_groups) moves to state_groups::build_down_to_stage0_groups, which is also build_of's third look.

  • Not done: scan widening on build_of. It changed plans. On avg by (job) (avg_over_time(latency[5m])) with avg by (job, region) and avg by (region), a tumbling-window candidate has a private chain plus a SummaryAgg over a private Project. build_of's first look counts only the chain, a build over input groups, and finds one build. Scan widening needs the only build and skips the candidate. On scan widening's lookup, Pass 2 kept 4 widened-scan candidates; on build_of it made 244. So scan widening keeps its own lookup, now shared.

• Stage 0 groups computed once. Memo stores how many groups Stage 0 produced (Memo::new), so Memo::is_stage0_group(id) is id < n. added_groups::workload_groups (a walk from the query roots, redone per rule and per Pass 1 candidate) is deleted. Before deleting it, every Pass 1 run in the tests below checked that the walk and id < n give the same set.

• Stage 1-only helpers off Memo. find_or_add_group and over_child_groups call each other, so both move to logical-asap-optimization/src/find_or_add_group.rs as crate-private functions. find_or_add_sized_group looks the node up once (it used to look it up, then call find_or_add_group, which looked it up again).

• Hash consing. looked_up_structural_hash reused structural_hash's own-fields tuple by copying it. Both now call structural_hash_with(node, child_hash).

• devtools. stage0_memo_groups is inlined (Memo::new(dag)).

• Plans unchanged: snapshots and both memo texts byte-identical to b080c96, and the Pass 2 memo text of 35 other workloads (see Verification).

Notation (formulation → code)

Formulation (#1203, #1201, #1207)	Code name in this PR
memo group $G$	MemoGroup, id MemoGroupId
candidate $v = (e_v, p_v)$	CandidateForMemoGroup
expression $e_v = \operatorname{op}v(G{v,1}, \dots, G_{v,n})$	root_node (one operator for Pass 1 candidates) + input_groups (by child position)
finding or adding the group of a node	find_or_add_group, add_group, over_child_groups
query $q$	the queries indices; Memo::add_queries

Concepts not in the formulation

Interim names with no formulation term, for the user to name or drop:

• Stage 0 groups (Memo::is_stage0_group, stage0_group_count, GroupsReadThroughAddedGroups::stage0_groups): the groups Stage 0 produced from the queries' DAG. Rules 1 and 2 still rewrite only these.
• Added groups (GroupsReadThroughAddedGroups::added_groups, groups_read_through_added_groups): the groups Pass 1 and Pass 2 add beneath them.
• Where a build is read (BuildPlace::{Private, InputGroup, BeneathGroups}): private nodes go away with 2a (one operator per candidate). The enum lasts until rule 9 and the rollups match over groups only.
• Accuracy budget parts (accuracy_budget_parts, size_own_candidate_for, find_or_add_sized_group): sizing for a share of a budget. 2b deletes it (Logical memo candidate as (one operator over groups, logical properties θ) with a requirement per input; no private nodes #1203: no budget shares).
• Origin (OriginOfCandidateForMemoGroup): which rule made a candidate.

Second review verification (8182edf)

• cargo test -p asap-types --lib: 146 passed. cargo test -p asap-logical-asap-optimization --lib: 198 passed.
• cargo test -p asap-integration-tests --test integration -- memo_enumeration reuse_rules operator_sharing sharing_issue_regressions finer_aggregate_reuse planner_layering_example having_shared_aggregate stage1_logical_asap_optimization joint_search_equivalence sketch_rollup rollup_controlled production_rollup_workloads: 154 passed, 23 ignored (pre-existing). An earlier run used a temporary check that workload_groups(memo) equals {id < stage0_group_count}.
• cargo test -p asap-devtools --lib: 18 passed. --test devtools -- asap_planner:: asap_planner_acceptance:: memo_pricing::: 56 passed, 1 ignored.
• UPDATE_SNAPSHOTS=1 cargo test --release -p asap-devtools --test corpus_snapshots -- --include-ignored: 9 passed, no file changed.
• asap_planner --run-until stage1_pass2 --format text on the KLL and scan-widening examples: byte-identical to b080c96.
• asap_planner --run-until stage1_pass2 --format text: byte-identical to the pre-review head b080c96 on 35 workloads besides the two examples: the reviewer's 20 workloads, six rollup / scan-widening / avg workloads, five earlier finer-aggregate, rollup, top-k and scan-KLL examples, and random-mixed-seed-862 workloads 01–04.
• cargo fmt --all; cargo clippy --workspace --all-targets -- -D warnings: clean.

Consumer naming (0c6d8b1, 2026-11-05)

• build_read_by is build_of; prose says consumer for the candidate that reads an input and query for the evaluation's sense (the concepts doc's terms). No behaviour change.
• cargo fmt --all -- --check; cargo clippy --workspace --all-targets -- -D warnings: clean. cargo test -p asap-logical-asap-optimization --lib: 198 passed.

Check renamed (2cd0aaa, 2026-11-05)

check_memo_well_formed → check_group_references_valid_and_acyclic (the user's request). cargo clippy --workspace --all-targets -- -D warnings: clean; cargo test -p asap-logical-asap-optimization --lib: 198 passed.

Viewer

viewer_contract tests pass on this branch. The viewer document is not updated for the new group shape (decided: viewer ignored for now).

Verification

• cargo test -p asap-types: 257 passed (the four structural_hash tests moved here from stage0/cse.rs).
• cargo test -p asap-logical-asap-optimization --lib: 198 passed (new: equal_k_builds_with_another_failure_probability_or_guarantee_are_other_groups, a_private_build_equal_to_a_group_raises_its_budget_parts).
• cargo test -p asap-physical-asap-optimization: 29 passed.
• cargo test -p asap-plan-selection --lib: 142 passed.
• cargo test -p asap-devtools --lib: 18 passed (planning_stages_text::prints_every_stage_of_a_repeating_rate_query: same plan and cost as main, the selected choice now also names the two groups Pass 1 adds).
• cargo test -p asap-integration-tests --test integration -- memo_enumeration reuse_rules operator_sharing sharing_issue_regressions finer_aggregate_reuse planner_layering_example having_shared_aggregate: 117 passed, 23 ignored (pre-existing).
• cargo test -p asap-devtools --test devtools -- asap_planner:: asap_planner_acceptance:: memo_pricing:: viewer_contract::: 58 passed, 1 ignored (pre-existing).
• UPDATE_SNAPSHOTS=1 cargo test --release -p asap-devtools --test corpus_snapshots -- --include-ignored: 9 passed, no snapshot file changed.
• cargo fmt --all; cargo clippy --workspace --all-targets -- -D warnings: clean.
• GitHub CI on b080c96: test, format-and-lint and Check PR title pass.
• Memo shape: asap_planner --run-until stage1_pass2 --format text on the KLL example and on a scan-widening example (avg by (job) / avg by (region) of quantiles over latency[5m]) gives the same groups and candidates as 368d34c; on b080c96 the output is byte-identical to af2a90d's for both examples.
• Test updates: expectations that counted Pass 1 groups or looked for rollups/shared builds on the query's root group now count the added groups and look at the consumer group (memo_enumeration, asap_planner::memo_section_*, finer_aggregate_reuse::rollups, sharing_issue_regressions::sketch_rollup_*, planner_layering_example2 memo choices, planning::select_rollup (used by sketch_rollup_planning: a shared plan now also chooses each query's summary candidate above its consumer), test_support::shared_summary_state_groups).

Left for 2a-2..k

• Switch each Pass 2 rule to emit one operator over groups through find_or_add_group: tumbling windows (rule 1), scan widening (2), summary inputs (3), top-k (4), rollups (6, 7), summary results (8), state configurations (9), and the shared summary-state step (10). With them, delete their own group-adding code and equalities: scan_labels::intern_inputs / same_source_node, reuse::append_producer / reuse::same_node, step 10's SharedBuild groups, SummaryResult, direct_input_builds, same_range_input, and Memo::is_stage0_group / groups_read_through_added_groups once no rule needs to tell added groups apart.
• Pass 1 still compares nodes by NodeToStructuralIdCache in three places: the composed root against the group's own node and the candidate dedupe key (add_pass1_candidates.rs), and PerComputationCandidates::walked. That id deliberately leaves out the schema, guarantee, timing and materialization composition derives (a composed root carries the exact guarantee and may close the label set where Stage 0's node has neither), so Stage 0's equality would not match there. Moving them to one equality is left for this series.
• Then add CandidateExpression { operator: Operator<MemoGroupId> } in place of root_node + input_groups, and delete Stage 2's private nodes and Stage 3's ResultSlot::PrivateNode (after Physical properties: replace Materialization with timing × lifetime (None / OneEvaluation / AcrossEvaluations) and a schedule for shareable results, with r^phys and α^phys #1200 PR 1).
• Stage 0 still fills queries with its own walk; it builds Vec<MemoGroup> before a Memo exists, so it does not use Memo::add_queries yet.

[zzylol]

/// The group of node, hash-consed as Stage 0 builds groups; adds it (and
/// its children's groups) when none exists. Used by Pass 1 and rule 9.
pub fn intern(
&mut self,
node: &Rc,
queries: &[usize],
origin: &OriginOfCandidateForMemoGroup,
accuracy_budget_parts: usize,
) -> MemoGroupId; // new
If this refer to "reference" for a group, instead of copy an existing group, can you rename?

[zzylol]

what does population in

/// Why Pass 1 failed (stays).
pub enum PerComputationLogicalMemoCandidatesError {
    /// A query's DAG cannot be planned. // unchanged
    PerComputation(PerComputationCandidateError),
    /// Bad targets, or a candidate's summaries did not size. // doc changed
    Population(CandidatePopulationError),
    // Sizing(SchemaDerivationError) // removed: now CandidatePopulationError::Sizing
}

mean?

[zzylol]

Replies to the two comments above. Both changes land in the next push, with the independent-review fixes.

1. Memo::intern: yes, it gives back a reference to a group; it never copies one.

• It returns the id of the existing group whose node equals node, or adds a new group (and groups for its children) when none matches.
• It will be renamed find_or_add_group, and the doc comment will say this explicitly.
• Separately, the review found that its equality is looser than Stage 0's (structural id only). It will use Stage 0's comparison, so "equal" means the same thing in both places.

2. "Population": it meant "adding Pass 1's candidates to the memo failed". In this codebase "population" also means the set of series (maintained_population), so the name misleads. Renames:

Before	After
CandidatePopulationError	AddPass1CandidatesError
PerComputationLogicalMemoCandidatesError::Population	PerComputationLogicalMemoCandidatesError::AddPass1Candidates
try_populate_pass1	try_add_pass1_candidates
memo_population.rs	add_pass1_candidates.rs

What each variant means:

• RootDependenciesChanged: Pass 1 changed which groups a root depends on.
• InvalidTarget: a target is not a valid measure of an initial aggregate group.
• InvalidAbsorption: a target names an absorbed target that does not exist.
• Sizing: a candidate's summaries, resized for its readers' failure probability, did not rebuild.

[zzylol]

Closed: replaced by #1207's new order (interfaces first, then one PR per stage), starting with PR 0 #1226. The code on this branch is reused in PRs 1–8; the branch is kept.