[zzylol]

Merge unit. This is PR 3 of #1207's merge unit (PRs 0–9, with 0b). It is stacked on PR 2b (#1233, refactor/1207-lineage 58549f2: column lineage replaces table qualifiers), which is stacked on PR 2 (#1230), PR 1 (#1227), PR 0b (#1229) and PR 0 (#1226). The PRs merge into main together, after PR 9. This PR fills in Pass 1, which PR 0 left as todo!(). Planning now gets through Pass 1 and stops at the first todo!() of Pass 2 (PR 4); see "Test status".

Approval. All interface items were approved by the user on 2026-11-07, on condition that existing structures are reused instead of new definitions wherever possible, with no over-abstraction or duplication: B1 plain exact beside the accumulator (decision 9), B2 key-only HAVING pushed down per consumer Filter group (decision 10), B3 integration helpers todo! until PR 6 (decision 12), B4 the crate-internal top-k fns and the generic any_measure_filtered, B5 the test fixtures, B6 pub topk::ranked_column_of_group, B7 ComputationTarget/FusedComputation private to Pass 1, B8 FusedComputation::absorbed_groups, B9 no Pass1Error. The head 61f585a applies that: see "Simplifications (2026-11-07)", "Deleted" and "Restack onto PR 2b 58549f2 (2026-11-09)".

Why

#1207's Order, row "3. Pass 1": Pass 1 adds each group's initial candidates (exact, exact accumulator, summary with one candidate per θ, fused) to the memo, with every intermediate result its own group (#1203: one operator per candidate, no private nodes). It also re-creates the Pass 1 test coverage that PR 0b deleted, with tests that build their input Memo through the Memo methods, so no stage crate depends on another, even in tests.

PR 0 left Pass 1 stopping at a private stub logical_node_trees(memo): target discovery walked node trees, and candidates came from a whole-workload node-tree composition (compose_logical_candidate) sized per budget share. Neither has a place in the memo model (concepts doc §3: "Everything that was computed on operator-node trees is computed on groups").

What

• Target discovery over memo groups. computation_targets(memo, supported_summaries) reads memo.groups(): each group's expression and group_properties. It finds one ComputationTarget per measure of each Aggregate group with a realization besides plain exact execution, once however many queries read it. Consumers are the group's parents. The logical_node_trees stub is deleted.

• Each way to compute a group becomes candidates, through Memo::find_or_add_group and Memo::add_candidate (add_pass1_candidates):

  • exact: every group's own expression is its first candidate (no θ);
  • exact accumulator: a state group SummaryAgg[ExactAggregate(kind)] (one candidate, no θ) and a FinalizeExactAccumulator over it. avg finalizes its sum and count states and divides them: a PromQL BinaryOp, or for SQL a Join group and a Project;
  • summary: one state group SummaryAgg[Sketch(algorithm)] per supported sketch algorithm, with one build candidate per θ in parameter_list(algorithm), all with the group's own expression (θ only in LogicalProperties::parameters). A separate readout group SummaryEstimate(state) is read by the result group;
  • fused builds: the whole-expression top-k (a heap state group over the rows below the absorbed sums) and the nested distributive merge: sum of sum, min of min, max of max (concepts doc §3, "Fused groups"), one exact Sum, Min or Max build over the rows below the inner aggregate, added as a candidate of the outer aggregate's own accumulator state group (Logical memo candidate as (one operator over groups, logical properties θ) with a requirement per input; no private nodes #1203 "fused builds"). One code path serves the three (distributive(intent) gives the accumulator kind). Their builds list the absorbed groups in fused_groups;
  • multi-measure aggregates (SQL): one single-measure Aggregate group per measure, each with its own candidates, joined back on the group keys by one candidate of the aggregate's group;
  • HAVING, read from the memo (no recorded copy): the aggregate's own having, split by key_only_having into below/above; and each parent group whose expression is Filter { child: aggregate_group, pred } with a key-only pred, which gets the candidates that push it below the aggregate (a Filter group over the input rows, which the states read; result group = that Filter group). A value HAVING stays a Filter above each readout.

• Every intermediate result is its own group; each candidate is one operator over input groups. A candidate must output its group's columns (add_candidate rejects it otherwise), so a readout whose column names differ from the group's (quantile for quantile_0_9) is its own group, read by a renaming Project candidate. A heap top-k's Float64 estimate is cast to the exact top-k's column type (a SQL count ranks Int64).

• No input requirement, no accuracy floor. Every candidate's input_requirements are LogicalRequirements::default(), one per input group. Pass 2 (PR 4) gives each readout's input its required_accuracy; a top-k readout's heap_size ≥ k is part of that rule (concepts doc §2, "One rule for every algorithm"). The heap floor Pass 1 set before (input_requirements, heap_floor) is deleted.

• A way Pass 1 cannot build is not offered (Unsupported, e.g. a SQL count(*) with no non-null column): the group keeps its other candidates. A candidate that breaks the node rules or the group's output is a bug and fails Pass 1 (PerComputationCandidateError::Candidate).

• pass1_memo(memo, supported_summaries) no longer takes the queries' accuracy targets, the accuracy model, evidence or data. They served only without_summaries_no_query_can_use, which composed whole-workload node trees to prune summaries no query could use; that pruning is query-sensitive (§4: "query-sensitive pruning follows logical generation and reachability analysis") and moves to Pass 2's dead-candidate elimination (PR 4).

• Schema derivation fix (types, its own commit 246110d). One rule for a top-k's ranked rows, exact or a heap's readout: one row per (partition, item) when the item identity is complete (ranked_rows_unique_keys, item_identity_is_complete, exact_topk_unique_keys in ir/schema/topk.rs). Before, the exact top-k's group derived no unique key over per-series rows while the heap readout's Project carried (job, series identity) from topk_schema_of_build, so topk by (job) (10, sum_over_time(m[1m]))'s heap readout failed add_candidate's exact check. This PR is the first whose strict check exercises it; PRs 2/2b are not rewritten.

• Series identity. A readout over per-series rows keeps the $promql_series_identity column the planner realizes before Stage 0 (with_promql_series_identity), so its renaming Project outputs the group's exact (closed) schema. A group whose schema is open (identity not realized) gets no renamed readout: a Project lists its columns and would drop the unlisted labels.

• Simplifications (2026-11-07), each removing a definition that duplicated memo data or another function:

  Removed	Replaced by	Why
  HavingEntry, HavingPlacement, ComputationTarget::having_filters (types, + re-export)	the aggregate's having and its parent Filter groups, read from the memo; private has_having(memo, group) for the three "has a HAVING" checks	a copy of memo data
  ComputationTarget::measure_filter (types)	Aggregate.filters[measure_index]; its two uses were already covered by whole_expression_input and realize_keyed_additive_summary_input, which refuse filtered measures	a copy of memo data
  Pass 1's ranked_column(memo, group) (a second copy of types' Project/Filter/readout walk)	asap_types::ir::schema::topk::ranked_column_of_group(memo, group), types' one walk over its internal MemoGroupAsInput; types' generic ranked_column becomes pub(crate)	one walk
  non_asap(memo, group) (cloned the operator)	expression(memo, group), the one accessor beside schema(memo, group)	near-duplicate accessor
  nested_sum_input, add_nested_sum_targets	nested_distributive_input, add_nested_distributive_targets over distributive(intent)	one path for sum/min/max
  exact_accumulator_build, finalized_with, the finalized closure	accumulator_build + accumulator_state(memo, kind, intent, measure, read)	three builders of the same SummaryAgg
  ComputationTarget, FusedComputation in asap_types::stages (+ re-exports)	the same structs, crate-private in Pass 1 (B7)	only Pass 1 uses them
  FusedComputation::absorbed_targets (indices into the target list), AggregateWays::all_targets	absorbed_groups: Vec<MemoGroupId>, used directly as fused_groups (B8)	the index-to-group mapping existed only to produce the groups
  Pass1Error (one variant)	pass1_memo returns PerComputationCandidateError (B9); planner run_planning uses ?, devtools run_stages_until maps the error once (its second, unreachable error branch removed)	a wrapper with nothing to add
  test fixtures aggregate_group_with, range/instant, table_group, divide	one aggregate_group(.., having), one selector(.., kind, ..); the single-use ones inlined	near-duplicates / single use

• Deleted (replaced by the memo candidates): compose_logical_candidate, compose, compose_roots, rewrite, substitute, substitute_measure, with_names_of, whole_expression, nested_sum, Sizing, ChosenComputation, Chosen, PerComputationCandidates (with combinations, absorbed_choice_varies, without_unsupported_summaries, without_summaries_no_query_can_use), output_can_meet_target, sources_read, absorbed_targets, generate_per_computation_candidates, logical_node_trees, ComputationTarget::{choice_count, computation_for}, PerComputationCandidateError::InvalidChoice, the Pass 1 feature test-support and its test_support module; asap_accuracy::test_support::{estimate_has_bound, estimate_bound_over_raw_data_samples, local_guarantee} (their only callers were Pass 1's tests, which now live in asap-accuracy); the executor test helper computation_candidates (no caller) and the executor's dev-dependencies on Stage 0 and Pass 1.

Decisions and pending items

#	Item	Decision	Status
1	Accuracy requirements and floors	Pass 1 sets none (2026-11-08 redesign: θ = structure only; LogicalRequirements { required_accuracy } on readout inputs, set by Pass 2). The top-k heap floor moved out too: heap_size ≥ k is checked in Pass 2's α^log rule.	PR 4 (user, 2026-11-06 / 11-08)
2	Metric rule	$\alpha^{\log}_{u,c,i}$ per (producer, consumer input) edge: Pass 1 offers KLL and DDSketch readouts to every quantile group, and edge building drops KLL → B's ratio.	PR 4 (user, 2026-11-06)
3	Dropping θ whose error alone exceeds every reaching query's target (§2 level 2)	The pruning cap $\max_q \varepsilon_q / \lambda_{q,c,i}$ needs the queries reaching each edge (§4: query-sensitive pruning follows logical generation). Pass 1 keeps the whole list.	PR 4 (user, 2026-11-06)
4	fused_groups	Main's meaning (concepts doc §3, "Fused groups"): a fused candidate computes, as one operator, a sub-DAG pattern of several operators in one appearance; fused_groups lists the groups of the operators it replaces. The whole-expression heap's builds list the absorbed sums' groups; the nested distributive merge's build lists the inner aggregate's group. The doc's rules top-k, sum∘sum, min∘min and max∘max exist (min∘min and max∘max added 2026-11-07 on the sum∘sum path); Hydra is not (doc: "possibly").	approved 2026-11-06; min/max 2026-11-07
5	Pass 1 reads groups directly	memo.groups()[id]: expression and group_properties. nested_distributive_input, whole_expression_input, reads_only_time_series, measure_input, realize_keyed_additive_summary_input, summed_per_series_samples, ranked_value take (&Memo, MemoGroupId). MemoGroupAsInput stays types-internal; the ranked column comes from types' ranked_column_of_group. A private raw_data_samples_of_metric(memo, group) walks the groups to the metric's scan (accuracy's node version and node-tree bind_guarantees are deleted in PR 4). SummaryCoverage::derive_from_inputs is already pub(crate) in PR 2b: no change.	approved 2026-11-06
6	add_candidate's output check	Exact (==), as PR 2 defines it. Qualifiers no longer exist (PR 2b); closed stays compared (see What, "Series identity").	per PR 2 / 2b
7	pass1_memo(memo, supported_summaries)	Drops accuracy_targets, accuracy, evidence and data: they served only the whole-workload pruning that PR 4's caps replace.	approved 2026-11-06
8	PerComputationCandidateError::{Candidate, LogicalProperties}; ComputationTarget over groups	As listed in Key interfaces. HavingEntry is deleted and ComputationTarget is Pass 1-private (2026-11-07, B7).	approved 2026-11-06
9	Plain exact candidate beside the accumulator	Every group offers its own expression; an aggregate with an accumulator form offers both (old Pass 1 listed only the accumulator). Stage 3 picks the cheaper.	approved 2026-11-07 (B1)
10	HAVING push-down	A key-only consumer Filter's predicate goes below in the Filter group's own candidates, whatever the aggregate's other consumers read (old Pass 1 pushed it only when every consumer applied it, #1089, because a candidate replaced the aggregate for all of them). Read from the memo, no recorded entry.	approved 2026-11-07 (B2)
11	Nested distributive merge placement	A candidate of the outer aggregate's accumulator state group, reading the raw rows (#1203, "fused builds").	per #1203
12	Tests outside Pass 1 that composed node trees	Helpers that returned composed node trees to execute (planning::explicit_local_candidates, planning::explicit_summary_candidate, avg_as_sum_and_count) todo!("#1207 PR 6: ..."): realizing a plan from the memo is PR 6's realizer. Interim inside the merge unit.	approved 2026-11-07 (B3)
13	Schema derivation fix (exact top-k unique key)	See What.	approved 2026-11-06 (coordinator, as its own commit in this PR)

New abstractions not in #1207 or the doc (approved 2026-11-07, B4–B6)

Item	Where	Why
pub(crate) item_identity_is_complete, ranked_rows_unique_keys, exact_topk_unique_keys; private RankedSource, ranked_source (the walk ranked_column already did)	types/src/ir/schema/topk.rs	The one rule for ranked rows' unique key (decision 13). B4.
pub ranked_column_of_group(memo, group); ranked_column<I> from pub to pub(crate) (no caller outside types)	types/src/ir/schema/topk.rs	Pass 1's only way to the one ranked-column walk without exposing MemoGroupAsInput. The pub surface count is unchanged (one pub fn replaces another). B6.
any_measure_filtered<C> (was over node predicates)	types/src/ir/operator/non_asap.rs	Reads a group's measure filters. B4.
private Pass 1 helpers expression(memo, group), schema(memo, group), raw_data_samples_of_metric(memo, group)	realization.rs	Read groups directly (decision 5).
crate-private ComputationTarget, FusedComputation (moved from types, B7); private computation_targets, add_pass1_candidates, has_having, having_filter_groups, distributive, nested_distributive_input, add_nested_distributive_targets, AggregateWays, SingleMeasure, add_aggregate_candidates, add_fused_candidate, measure_output, accumulator_state, accumulator_build, add_output_candidate, renaming_project, add_group, add_state_group, add_candidate, unless_unsupported, without_unsupported_summaries (free fn)	per_computation_candidates.rs	The implementation. AggregateWays is kept: it is built twice (the aggregate group, each key-only Filter group, by struct update) and read by two functions; as arguments it would be seven per call. SingleMeasure is kept: five fields read by five functions.
test helpers: metric_schema, scan_operator, group, scan_group, aggregate_group, query (test_fixtures.rs); memo_printout, label, candidates, groups_where, state_group, selector, compare, filter, quantile, sum, topk, pushed_below, candidates_for, algorithms (Pass 1 tests); schema_of (types topk.rs test); pass1, pass1_over, having_pushed_below (per_computation_candidates_sql.rs); stage0, pass1, root_group, aggregate_groups, candidates_of_aggregates, sketch_builds, avg_candidate_groups, pass1_sql (stage1_logical_asap_optimization.rs)	tests	Fixtures. B5.

How

1. computation_targets walks memo.groups(). For each Aggregate expression it lists the ways per measure (computation_candidates_for_intent, unchanged), then detects fused computations over groups (add_whole_expression_targets, add_nested_distributive_targets, close_absorptions; neither applies where has_having), prunes a PromQL top-k's own heaps, and drops unsupported algorithms.
2. add_pass1_candidates adds every Stage 0 group's own expression as a candidate, then, per aggregate group (its own having split by key_only_having) and per Stage 0 Filter group directly above it with a key-only predicate, add_aggregate_candidates builds each way bottom-up: add_state_group (θ list or one exact build), add_group for each intermediate (the operator is also its group's candidate), and add_output_candidate for the top operator: added to the result group, or, when its column names differ, added as its own group and read through renaming_project.
3. Readouts and finalizers over groups whose names differ are renamed by a Project; an open group gets none.
4. Every candidate goes through add_candidate (this crate) → LogicalProperties::new (θ in the list, none on exact operators) → empty input requirements → Memo::add_candidate.

Restack onto PR 2b 58549f2 (2026-11-09)

Squashed to one commit on PR 2b's new head (PR 0's 2026-11-08/09 interfaces). Adapted:

Change	Why
SummaryAgg { summary_type: FieldDataType::Sketch(algorithm), .. }; no grouping, no GroupingStrategy	PR 0 names
computation_candidates_for_intent(intent) -> ComputationChoices (no Result); summaries offered when sketch_statistic(intent) is Some	AggIntent measures carry no accuracy (N1); PerComputationCandidateError::InvalidAccuracy is gone (PR 0)
input_requirements and heap_floor deleted; add_candidate sets vec![LogicalRequirements::default(); inputs]	no floors in Pass 1 (restack plan, 2026-11-08/09): Pass 2 owns every readout requirement, heap_size ≥ k included
LogicalProperties::new(&expression, parameters) (2 args)	PR 1
test fixture query builds a MemoQuery { root, accuracy_target: None, values_per_evaluation: None }; integration helper stage0(root) calls memo_of_queries(&[root], &[None], &DataWorkload::default()); lower_sql(sql, catalog)	PR 0/2 signatures
names: FAMILY_PREFERENCE → ALGORITHM_PREFERENCE; family variables → algorithm (a SketchAlgorithm) or summary_type (a FieldDataType); tests a_dropped_readout_drops_the_algorithm, preference_order_names_each_algorithm_once, topk_offers_both_heap_algorithms	terminology: family = FieldDataType variant, summary type = family + algorithm
tests: asap_input_and_invalid_accuracy_are_rejected → asap_input_is_rejected; aggregate_without_realization_is_not_a_target uses stddev; only_a_topk_readout_has_a_parameter_floor → pass1_sets_no_input_requirement; summary_candidates_are_unsized deleted (no accuracy on intents: nothing to vary)	follows the above
per_computation_candidates_sql.rs: a_sql_fold_over_estimates_leaves_its_row_count_unbound and reads_ddsketch not restored (PR 0 deleted explicit_summary_candidate); ColumnRef::Qualified arms dropped from the heap-update helper	PR 0 / PR 2b
planning::explicit_local_candidates, explicit_summary_candidate (B3 todo!s) are gone: PR 0 deleted them; only avg_as_sum_and_count stays todo!("#1207 PR 6 …")	PR 0
SourceColumn, ColumnLineage, column_lineage(), ColumnIdentity: Pass 1 uses none (it compares output schemas only), so no import changes	PR 2b watch-outs checked

No new pub item. Self-review (over-abstraction, duplication, dead code, names): expression(memo, g)/schema(memo, g) are the only group accessors (no copy elsewhere); raw_data_samples_of_metric has one Pass 1 copy (accuracy's node copy goes in PR 4); no unused helpers (clippy -D warnings clean).

Key code interfaces added / modified

stage1-pass1-per-computation-candidates/src/logical_memo_candidates.rs

/// Stage 1, Pass 1's entry: adds each group's initial candidates to Stage 0's memo. Stays.
pub fn pass1_memo(
    memo: Memo,
    supported_summaries: Option<&[SupportedSummary]>,
) -> Result<Memo, PerComputationCandidateError>;
// was (memo, accuracy_targets, accuracy, evidence, data, supported_summaries) -> Result<Memo, Pass1Error>
// removed: Pass1Error (a one-variant wrapper of PerComputationCandidateError) (B9)

stage1-pass1-per-computation-candidates/src/per_computation_candidates.rs

/// Why Pass 1 could not add a group's candidates. Stays.
pub enum PerComputationCandidateError {
    /// A group's expression breaks the node rules: bad input. // unchanged
    Structure(SchemaDerivationError),
    /// A Stage 0 group is already an ASAP operator: bad input. // unchanged
    ContainsAsap,
    // InvalidAccuracy — removed in PR 0 (measures carry no accuracy target)
    /// `Memo::add_candidate` rejected a candidate Pass 1 built: internal bug. // new
    Candidate(CandidateError),
    /// A candidate's θ or window does not fit its operator: internal bug. // new
    LogicalProperties(LogicalPropertiesError),
    /// A way Pass 1 cannot build; it is not offered (never returned by `pass1_memo`). // unchanged
    Unsupported(&'static str),
    // InvalidChoice — removed (no whole-workload choices)
}
// removed: pub mod test_support { compose_logical_candidate, generate_per_computation_candidates }, feature `test-support`

types/src/stages/candidates.rs

/// One candidate of a memo group. Stays; only `fused_groups`' doc changes.
pub struct CandidateForMemoGroup {
    /// $e_v$. // unchanged
    pub expression: CandidateExpression,
    /// $p^{\log}_v$. // unchanged
    pub logical_properties: LogicalProperties,
    /// $r^{\log}_{c,i}$, one per input group. // unchanged
    pub input_requirements: Vec<LogicalRequirements>,
    /// The groups of the operators a fused candidate replaces (one appearance;
    /// concepts doc §3 "Fused groups"); empty otherwise. // doc was "groups it also implements"
    pub fused_groups: Vec<MemoGroupId>,
    /// Which rule added it. // unchanged
    pub origin: OriginOfCandidateForMemoGroup,
}

types/src/ir/schema/topk.rs (schema derivation fix; the one ranked-column walk)

/// [`ranked_column`] of memo group `group`: the column a top-k reading the group ranks it by. // new (B6, approved 2026-11-07)
pub fn ranked_column_of_group(memo: &Memo, group: MemoGroupId) -> Result<ColumnId, ResolveError>;
/// The column a top-k ranks `child` by. // was `pub`; no caller outside types
pub(crate) fn ranked_column<I: OperatorInput>(child: &I) -> Result<ColumnId, ResolveError>;
/// Whether ranked rows' item columns identify each item completely: the input is closed,
/// or the items include the realized PromQL series identity. // new
pub(crate) fn item_identity_is_complete(input: &Schema, item_columns: impl IntoIterator<Item = ColumnId>) -> bool;
/// One row per (partition, item): every column but the ranked value. Used by the exact
/// top-k and by `topk_schema_of_build` (the heap readout). // new
pub(crate) fn ranked_rows_unique_keys(columns: usize, value: ColumnId) -> Vec<Vec<ColumnId>>;
/// The exact top-k's unique keys: `ranked_rows_unique_keys` when its input has one row per
/// item with a complete identity, else the ranked projection's own keys. // new
pub(crate) fn exact_topk_unique_keys<I: OperatorInput>(child: &I, partitions: usize, output: &Schema) -> Vec<Vec<ColumnId>>;

types/src/stages/per_computation_logical_memo_candidates.rs keeps only HowToComputeAggregateMeasure (devtools reads it). ComputationTarget and FusedComputation moved, crate-private, to stage1-pass1-per-computation-candidates/src/per_computation_candidates.rs (B7):

// removed from asap_types::stages (and its re-exports): ComputationTarget, FusedComputation
/// One measure of an aggregate group with a realization besides exact execution.
pub(crate) struct ComputationTarget { // was pub, in asap_types::stages
    /// The aggregate's group. // unchanged
    pub aggregate_group: MemoGroupId,
    /// The measure's index. // unchanged
    pub measure_index: usize,
    /// Its ways to compute, the exact option first. // unchanged
    pub ways_to_compute: Vec<HowToComputeAggregateMeasure>,
    /// The fused computation with the targets beneath. // unchanged
    pub fused_computation: Option<FusedComputation>,
    // removed: measure_filter (= Aggregate.filters[measure_index]),
    //          having_filters (= Aggregate.having + parent Filter groups, read from the memo)
}
// removed: ComputationTarget::choice_count, ComputationTarget::computation_for
// removed: HavingEntry, HavingPlacement (and their re-export from asap_types::stages)
/// A target's measure computed with the targets beneath it, as one operator: a
/// whole-expression top-k heap, or a nested `sum`/`min`/`max` merge. Stays; doc only.
pub(crate) struct FusedComputation { // was pub, in asap_types::stages
    /// The groups of the operators it replaces, listed in its builds' `fused_groups` (B8).
    pub absorbed_groups: Vec<MemoGroupId>, // was absorbed_targets: Vec<usize> (target indices)
    /// `Summary(heap algorithm)` or `ExactAccumulator([Sum | Min | Max])`. // was [Sum] only
    pub ways_to_compute: Vec<HowToComputeAggregateMeasure>,
}

types/src/ir/operator/non_asap.rs, accuracy/src/test_support.rs

/// Whether any measure has its own filter. Stays.
pub fn any_measure_filtered<C>(filters: &[Option<Predicate<C>>]) -> bool; // was (&[Option<Predicate>])
// removed from asap_accuracy::test_support (only Pass 1's moved tests called them):
//   estimate_has_bound, estimate_bound_over_raw_data_samples, local_guarantee

Usage (from running_example_gives_one_build_per_theta_and_separate_readouts, run at this head):

// Stage 0's memo, built as Stage 0 builds it: `latency` (instant selector),
// p95 / p99 / p50 aggregates over it, and B's ratio p99 / p50.
let mut memo = Memo::default();
// `latency` with its series identity realized (doc G1)
let g1 = selector(&mut memo, TimeRangeKind::Instant, "latency", &[]);
let g3 = aggregate_group(&mut memo, g1, Reduction::by(vec![]), vec![quantile(0.95)], None); // query A
let g4 = aggregate_group(&mut memo, g1, Reduction::by(vec![]), vec![quantile(0.99)], None);
let g5 = aggregate_group(&mut memo, g1, Reduction::by(vec![]), vec![quantile(0.5)], None);
let g6 = group(&mut memo, Operator::NonASAP(NonASAPOp::BinaryOp { lhs: g4, rhs: g5, /* ÷ */ .. })); // query B
query(&mut memo, g3); // memo.add_query(MemoQuery { root, accuracy_target: None, values_per_evaluation: None })
query(&mut memo, g6);
// Pass 1 (the facade calls it after Stage 0 with the deployment's supported summaries).
let memo = pass1_memo(memo, None)?;
// KLL's state group: 7 build candidates, θ = k ∈ {26, 52, 132, 269, 547, 1404, 2863};
// DDSketch's: 7, α ∈ {0.1 … 0.001}; G3/G4/G5: own aggregate + one readout per algorithm;
// G6: its own BinaryOp. Pass 2 (PR 4) adds reuse candidates and the edges, where the
// metric rule drops KLL → B's ratio; Stage 3 picks one producer per input.

Major algorithm changes

Pass 1 candidate generation.

• Problem. On PR 2, Pass 1 stops at todo!(). Before PR 0, it built one whole-workload node tree per choice and sized each sketch for an equal budget share; the result was copied into private node chains.
• Before / after. Before: targets over node trees → compose_logical_candidate(choice) → node tree per choice → sizing per accuracy_budget_parts → private chains. After: targets over groups → per way, groups bottom-up through find_or_add_group, one build candidate per listed θ, one operator per candidate.
• Correctness. Every candidate passes the node rules and outputs its group's columns (add_candidate); the ported frontend tests (stage1_logical_asap_optimization::{sql,promql}_candidates_output_the_query_columns) run every SQL and PromQL query of the old composition test through Pass 1.
• Complexity. Linear in the targets × ways; each summary way adds |parameter_list(algorithm)| (4–280: KLL/DDSketch 7, HLL/KMV 4, CMS/Count Sketch 28, UnivMon 40, heap variants 280) build candidates to one shared state group, instead of one node tree per whole-workload combination.
• Limits. No accuracy pruning (PR 4). Hydra algorithms have empty lists (Planner: error models and parameter lists for Hydra (and other unsized families), so Pass 1 can offer them #1228) and are not offered.

Before / After

The doc's running example (concepts doc §1): query A = p95 of latency, query B = p99 / p50 of it (running_example_gives_one_build_per_theta_and_separate_readouts). Before (PR 2): pass1_memo panics at todo!("#1207 PR 3: discover targets over the memo groups, not node trees"). After, the memo (printed by the test):

G0: Scan
  [0] Scan inputs=[]
G1: TimeRange
  [0] TimeRange inputs=[G0]
G2: Aggregate[Quantile]
  [0] Aggregate[Quantile] inputs=[G1]
  [1] Project inputs=[G7]
  [2] Project inputs=[G9]
G3: Aggregate[Quantile]
  [0] Aggregate[Quantile] inputs=[G1]
  [1] Project inputs=[G10]
  [2] Project inputs=[G11]
G4: Aggregate[Quantile]
  [0] Aggregate[Quantile] inputs=[G1]
  [1] Project inputs=[G12]
  [2] Project inputs=[G13]
G5: BinaryOp
  [0] BinaryOp inputs=[G3, G4]
G6: SummaryAgg[Sketch(Kll)]
  [0] SummaryAgg[Sketch(Kll)] inputs=[G1] θ=Kll { k: 26 }
  [1] SummaryAgg[Sketch(Kll)] inputs=[G1] θ=Kll { k: 52 }
  [2] SummaryAgg[Sketch(Kll)] inputs=[G1] θ=Kll { k: 132 }
  [3] SummaryAgg[Sketch(Kll)] inputs=[G1] θ=Kll { k: 269 }
  [4] SummaryAgg[Sketch(Kll)] inputs=[G1] θ=Kll { k: 547 }
  [5] SummaryAgg[Sketch(Kll)] inputs=[G1] θ=Kll { k: 1404 }
  [6] SummaryAgg[Sketch(Kll)] inputs=[G1] θ=Kll { k: 2863 }
G7: SummaryEstimate[Quantile { q: 0.95 }]
  [0] SummaryEstimate[Quantile { q: 0.95 }] inputs=[G6]
G8: SummaryAgg[Sketch(DDSketch)]
  [0] SummaryAgg[Sketch(DDSketch)] inputs=[G1] θ=DDSketch { alpha: 0.1 }
  [1] SummaryAgg[Sketch(DDSketch)] inputs=[G1] θ=DDSketch { alpha: 0.05 }
  [2] SummaryAgg[Sketch(DDSketch)] inputs=[G1] θ=DDSketch { alpha: 0.02 }
  [3] SummaryAgg[Sketch(DDSketch)] inputs=[G1] θ=DDSketch { alpha: 0.01 }
  [4] SummaryAgg[Sketch(DDSketch)] inputs=[G1] θ=DDSketch { alpha: 0.005 }
  [5] SummaryAgg[Sketch(DDSketch)] inputs=[G1] θ=DDSketch { alpha: 0.002 }
  [6] SummaryAgg[Sketch(DDSketch)] inputs=[G1] θ=DDSketch { alpha: 0.001 }
G9: SummaryEstimate[Quantile { q: 0.95 }]
  [0] SummaryEstimate[Quantile { q: 0.95 }] inputs=[G8]
G10: SummaryEstimate[Quantile { q: 0.99 }]
  [0] SummaryEstimate[Quantile { q: 0.99 }] inputs=[G6]
G11: SummaryEstimate[Quantile { q: 0.99 }]
  [0] SummaryEstimate[Quantile { q: 0.99 }] inputs=[G8]
G12: SummaryEstimate[Quantile { q: 0.5 }]
  [0] SummaryEstimate[Quantile { q: 0.5 }] inputs=[G6]
G13: SummaryEstimate[Quantile { q: 0.5 }]
  [0] SummaryEstimate[Quantile { q: 0.5 }] inputs=[G8]

Mapping to the doc's table: G1 here is the doc's G1 (latency, via Scan G0 and TimeRange G1); G6 = doc G2 (KLL state, one build per θ of KLL's list; the doc's k = 200/345 are illustrative); G8 = doc G2′ (DDSketch); G2/G3/G4 = doc G3/G4/G5, each with its exact aggregate and a renaming Project over a readout group per algorithm (G7/G9, G10/G11, G12/G13); G5 = doc G6. The doc shows G4/G5 reading only G2′: that is the metric rule, which drops the KLL → ratio edges in PR 4 (decision 2).

Verification

cargo check, cargo clippy --workspace --all-targets --all-features -- -D warnings and cargo fmt --all --check are clean.

Pass 1's tests (stage1-pass1-per-computation-candidates/src/per_computation_candidates_tests.rs, 28 tests; plus 4 in realization.rs, keyed_heap_input_refuses_a_filtered_count now over a memo). Each builds its memo through Memo::find_or_add_group and Memo::add_query (test_fixtures.rs); no other stage crate is a dependency, even in tests. PR 0b's list (#1229), by category:

Category	Test	Verifies
running example	running_example_gives_one_build_per_theta_and_separate_readouts	KLL and DDSketch state groups hold one build per θ of their list, with the group's own expression; p95/p99/p50 each read either state through a readout group; the ratio keeps its own expression
series identity	per_series_readouts_keep_the_series_identity	a quantile_over_time readout's Project outputs the per-series aggregate's exact closed schema, mapping the estimate's $promql_series_identity column onto the group's
	an_open_per_series_aggregate_offers_no_renamed_readout	without a realized identity (open schema) no renamed readout is offered: it would drop unlisted labels
target discovery	multi_measure_aggregate_has_one_target_per_measure	each measure of a multi-measure aggregate is a target
	aggregate_without_realization_is_not_a_target	a stddev (no accumulator, no summary) is no target and gets no candidate beyond its own expression
	an_aggregate_read_by_several_roots_is_one_target	a group read by two roots is one target
	asap_input_is_rejected	ASAP groups fail at the boundary
	unsupported_summaries_are_not_offered	the deployment's unsupported algorithms add no group
candidate composition	every_group_offers_its_own_expression	every Stage 0 group's first candidate is its expression, no θ
	exact_accumulator_is_a_state_group_and_a_finalize	an accumulator is a one-candidate state group read by a finalize
	multi_measure_candidate_joins_single_measure_groups	a SQL multi-measure aggregate joins single-measure groups, each with its own candidates
	pass1_sets_no_input_requirement	every candidate, a top-k readout's included, has one empty LogicalRequirements per input group
HAVING placement	key_only_having_is_pushed_below_into_the_filter_group	a key-only consumer Filter gets candidates over filtered input rows; the aggregate keeps its own
	value_having_stays_above	a value predicate pushes nothing below
	global_aggregate_having_stays_above	a global aggregate's constant HAVING stays above
	own_value_having_filters_each_readout	the aggregate's own value having is a Filter over each readout
fused heaps / whole-expression top-k / raw-data signs	whole_expression_topk_heaps_read_the_raw_samples	the heaps read the range's raw samples keyed by job, with RawDataSamplesOfMetric, one build per θ, fused_groups = the absorbed sums
ranked items	count_ranked_topk_counts_with_unit_weight	a count-ranked top-k uses unit, non-negative weights
	ranked_item_is_the_keys_outside_the_partition	the top-k's partition keys stay out of the ranked item
nested distributive merges	nested_distributive_aggregate_merges_into_the_outer_state_group	for sum∘sum, min∘min (new) and max∘max (new): the merged build is a candidate of the outer aggregate's state group of that kind, over the raw rows, fused_groups = the inner aggregate
	nested_merge_needs_the_same_distributive_aggregate	a sum or a min over a max gets no merge
stream norms	update_weights_follow_the_measure	a Count-Min count adds 1 per row; a sum's state reads the raw value column
intents	cardinality_keeps_…, frequency_statistics_keep_universal_choices, topk_keeps_both_specialized_heap_choices, approximate_count_lists_its_accumulator_and_summaries, a_count_offers_no_count_sketch	the ways listed per intent (unchanged tests)

Schema derivation fix (types/src/ir/schema/topk.rs): exact_topk_and_heap_readout_derive_the_same_unique_keys — for topk by (job) (10, sum_over_time(m[1m])) over realized series identity, the exact top-k's group and a heap's ranked readout derive the same non-empty unique key.

Guarantee composition is no longer Pass 1's: candidates carry θ, and each build's ε follows from θ (PR 1's parameter_list); composition to the query root is error_propagation (PR 4) and constraint (K) (PR 8). The Count-Min bound test that lived in Pass 1 but tests asap-accuracy (count_min_estimate_has_no_bound_over_an_open_requirement) moved to accuracy/src/propagation.rs; its sibling …_is_gated_on_the_metric was deleted by PR 0 (its local_guarantee is gone).

Ported to the memo (integration-tests): per_computation_candidates_sql.rs (the four HAVING tests, the three heap-update tests, over Pass 1's memo of SQL queries) and stage1_logical_asap_optimization.rs (its promql() helper realizes series identity, as the planner does): sql_candidates_output_the_query_columns and promql_candidates_output_the_query_columns (were *_candidates_compose: every query of the old composition test goes through Pass 1 with every candidate's output checked by add_candidate), topk_offers_both_heap_algorithms (was topk_estimates_carry_no_frequency_bound; the guarantee half moves to PR 8), topk_over_an_aggregate_ranks_its_labels, subquery_in_a_measure_filter_is_an_input_group_of_the_state (was …_is_substituted), filtered_sql_avg_is_null_without_matching_rows, promql_avg_divides_sum_and_count_vectors, sql_avg_keeps_its_explicit_output_name, sql_count_star_without_a_non_null_column_is_only_exact (was …_is_unsupported), sql_count_of_a_nullable_column_counts_its_values.

Deleted: devtools/src/candidate_labels.rs connected_labels_describe_selected_states and connected_labels_name_actual_shared_state (they labelled composed node trees; tools are PR 9; PR 0 already deletes them on the new base). PR 0 deleted stage1_logical_asap_optimization.rs (P11: composition tests); this PR adds back its memo-based version and registers it in tests/main.rs.

Test status

cargo test -p <crate> --all-features --no-fail-fast, per crate, at 61f585a (logs: per-crate, compared by test name with PR 2b 58549f2). No test fails inside Pass 1; planning stops at Pass 2's first todo!() (PR 4: "rule 1, chain groups and the merges over them").

Crate	PR 2b (58549f2)	PR 3 (61f585a)	Where the failures stop now
asap-stage1-pass1-per-computation-candidates	11 / 0 / 0	32 / 0 / 0	—
asap-accuracy	60 / 0 / 0	61 / 0 / 0	+1 test moved from Pass 1
asap-types	199 / 0 / 0	200 / 0 / 0	+1 (exact_topk_and_heap_readout_derive_the_same_unique_keys)
asap-stage0-memo	15 / 0 / 0	15 / 0 / 0	—
asap-planner	8 / 18 / 1	8 / 18 / 1	18 at PR 4 (were at PR 3)
asap-devtools	37 / 56 / 3	37 / 56 / 3	45 at PR 4 (were 48 at PR 3), 9 at PR 9, 2 in asap_planner subprocesses
asap-integration-tests	83 / 51 / 4	100 / 50 / 4	43 at PR 4; 3 at PR 6 (view retention); 3 at PR 6 (avg_as_sum_and_count, new tests); 1 as on PR 2b (materialized_views)
asap-executor	210 / 63 / 1	210 / 63 / 1	unchanged (PR 6 θ; the PR 3 one now at PR 4)
asap-cost	31 / 1 / 0	31 / 1 / 0	unchanged (PR 6)
asap-frontend-promql / -sql / -common / -metricsql, asap-sql-function-catalog, metricsql_common, asap-stage3-plan-selection, Pass 2, Stage 2	218 / 175 / 49 / 15 / 10 / 3 / 3 / 0 / 0, no failures	same	—
Total	1127 / 189 / 9	1167 / 188 / 9

By test name against PR 2b: no test that passes on PR 2b fails here. 46 newly passing: 4 SQL HAVING tests (failed at PR 3's todo! on 2b) and 42 tests absent on 2b (24 Pass 1, 13 stage1_logical_asap_optimization, 3 SQL heap-update, 1 types, 1 accuracy; renamed and moved ones included). Three new tests fail at an interim todo!("#1207 PR 6: realize the sum-and-count plan of avg from Pass 1's memo") (B3): sql_avg_counts_its_own_input, sql_avg_from_sum_and_count_executes_as_avg, sql_avg_of_an_int64_column_keeps_the_exact_columns (absent on 2b). Six 2b tests are gone by name: renamed (a_dropped_readout_drops_the_family, preference_order_names_each_family_once), moved (count_min_estimate_has_no_bound_over_an_open_requirement → accuracy; a_count_offers_no_count_sketch, approximate_count_keeps_one_exact_accumulator_and_universal_choices → Pass 1's test module, the latter as approximate_count_lists_its_accumulator_and_summaries), deleted (summary_candidates_are_unsized).

crate_graph (in asap-integration-tests) passes. cargo fmt --all --check and cargo clippy --workspace --all-features --all-targets -- -D warnings are clean.

Done means

• Code pushed (head 61f585a, one commit; backup backup/1207-pass1-2c6d99336).
• Body current with the head.
• Tests re-measured at the final head (table above).
• Stale-name grep clean over the diff and this body (HavingEntry, HavingPlacement, having_filters, measure_filter, nested_sum, non_asap(memo, ColumnIdentity, Pass1Error, absorbed_targets, all_targets, ComputationTarget/FusedComputation in types, reader, kept, MILP row, σ/Schedule, EvaluationSchedule, logical_node, child_groups, accuracy_budget_parts, operators_kept_in_definition, group_as_input, BuiltOrInput, qualifier): no hit in the code; in the body only the names of deleted items (logical_node_trees, accuracy_budget_parts, HavingEntry, HavingPlacement, having_filters, measure_filter, nested_sum*, non_asap, Pass1Error, absorbed_targets, all_targets, qualifier in PR 2b's summary) and SQL/PromQL data rows (row/rows).
• Rebased onto PR 2b (refactor/1207-lineage 58549f2).