feat(planner): Example 1 completion — identical-expression CSE, whole-expression top-k heaps, counter-sample weights, per-second plan-selection cost

[zzylol]

Why

#509's Example 1 could not run end to end on the planning stages:

• Stage 1 had no Pass 2, so two queries over the same input could never share it.
• Q2's heap sketches ranked an exact per-series sum instead of absorbing it, and their weights were read from the sample-value column, which is wrong for SQL keys.
• Count-Min could never be admitted, because nothing proved counter samples non-negative.
• Plan selection priced one evaluation of the whole workload, so queries with different recurrences could not be compared, and kept state had no price.

This PR completes Example 1 (consolidation group G4) under the user's decisions on these points.

What

• Stage 1 Pass 2, identical-expression rule. One candidate per non-empty subset of a candidate's sharing groups that gives a distinct candidate. The independent candidate is kept. The variants are ordinary Stage 2/3 candidates, and nothing is capped in the library.
• Whole-expression top-k heaps. A top-k over a per-item sum or count gets Count-Min + heap and CountSketch + heap candidates that read the inner aggregate's input and absorb it. They replace the heaps over the exact sum, so Example 1's Q2 has exactly three candidates.
• Counter samples as an open requirement. Pass 1 records that a heap's weights are raw samples of metric M (or plain sums of them), without reading DataWorkload. Plan selection confirms it from the new DataWorkload.metric_types; unconfirmed counts as unmet.
  • The same rule binds feat(planner): compose accuracy guarantees through arithmetic over approximate values (#669, part a) #687's sign statistic for sums of counter samples, and gates a Count-Min estimate's bound.
  • The selected plan carries the confirmed proof for every consumer.
• Top-K heap weights (fix). A heap is weighted by the column the top-k ranks, so SQL with an Int64 key, or a key named value, plans correctly.
• Per-second plan-selection cost.
  • Each query's recurrence is an input, and every term is a cost per second.
  • Memory is charged only as kept state (bytes × seconds); fix(planner): price state by its size; size KLL for δ #682's per-computation peak-memory charge is removed.
  • ResourceCalibration gains horizon_s and cost_per_retained_byte_second.
• Devtool limit. stage_pipeline --max-combinations (default 10^4) counts Stage 1 candidates, Pass 2's included, while enumerating, and stops with an error past the limit. Exit 3 / server 422 are unchanged.

How

• Pass 2 (asap_aware_cse). ir::cse interning identifies the sharing groups. Sharing a set of groups also shares every group it leaves as a single node. The distinct candidates are therefore the closed sets of that closure, and they are enumerated with Next Closure (at most one closure per step and group) instead of trying all 2^g subsets. They are emitted in the subsets' binary-counter order.
• Stage 1 enumeration is a lazy iterator (stage1_logical_asap_optimization), and run_stage2_and_stage3 runs over what a caller enumerated. The devtool counts as it consumes the iterator; the library collects all candidates.
• Whole-expression candidates (ComputationCandidate::WholeExpression) reuse the keyed-additive rule, which moves to realization.rs along with ranked_column. The outer partition keys are re-indexed onto the heap's input. Enumeration and combinations() skip choices that differ only in an absorbed target.
• Requirements as symbols. The open weight requirement is WeightDomain::SamplesOfMetric { metric }. The sign of M's samples is the statistic values_non_negative[M], which gates a bound as a factor. Stage 3 binds it only to 1, when M is declared a counter, and leaves it unbound otherwise.
• Cost (stage3_plan_selection module docs):
  • A NotMaterialized node is charged per consumer at the sum of their rates.
  • A MatAtQuery node is charged once at the combined rate of distinct intervals.
  • A MatAtIngestion node is charged per second of arriving rows.
  • One-off, scheduled and unknown queries are spread over horizon_s (3600 s).
  • Kept state is twice its bytes at ingestion and once at query time.
  • Totals within 1e-9 relative tie, and a tie keeps the earlier candidate.

Key code interfaces added / modified

crates/asap-aware-mapping/src/asap_aware_cse.rs (new), lib.rs

/// Stage 1, Pass 2: ASAP-aware CSE (the identical-expression rule only).
pub mod asap_aware_cse; // new: Pass 2 identical-expression rule

/// Two or more identical nodes that are legal to share, in children-first walk order.
pub struct SharingGroup { pub nodes: Vec<Rc<OperatorNode>> }   // from ir::cse interning
/// One Pass 2 candidate: the input roots with the listed sharing groups each made one node.
pub struct IdenticalExpressionVariant {
    /// Indices into the `sharing_groups` result of the groups this variant shares.
    pub shared_groups: Vec<usize>,
    /// The candidate's query roots with those groups shared.
    pub roots: Vec<QueryRoot>,
}

/// The sharing groups of a whole-workload candidate's roots, children before parents.
pub fn sharing_groups(roots: &[QueryRoot]) -> Vec<SharingGroup>;
/// One variant per distinct closed subset of groups (Next Closure).
/// Empty when nothing is identical; the caller keeps the unshared candidate beside them.
pub fn identical_expression_variants(roots: &[QueryRoot], groups: &[SharingGroup]) -> Vec<IdenticalExpressionVariant>;

crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs, pass/mod.rs

/// One Stage 1 candidate in enumeration order, before Stage 2 numbers it.
pub struct Stage1Candidate {                       // new: one enumerated Stage 1 candidate
    /// Human-readable name, used only for display.
    pub display_name: String,
    /// The Pass 1 choice per target; a Pass 2 candidate has its Pass 1 candidate's.
    pub choice: Vec<usize>,
    /// The composed query roots, or why the choice did not compose.
    pub roots: Result<Vec<QueryRoot>, PerComputationCandidateError>,
    /// For a Pass 2 candidate: its Pass 1 candidate's enumeration position and the
    /// shared groups (operator kinds); `None` for a Pass 1 candidate.
    pub pass2_of: Option<(usize, Vec<String>)>,
}
/// A candidate Pass 2 added: `id` is `from_logical` with each group in `shared` made one node.
pub struct IdenticalExpressionCandidate { pub id: String, pub from_logical: String, pub shared: Vec<String> } // new

/// The output of Stages 1 to 3 for one workload. The type stays; it gains two fields.
pub struct Stage1ToStage3Run<Id> {
    /// Stage 1, Pass 1: the per-computation candidates, with the roots they read.
    pub stage1_pass1_per_computation: PerComputationCandidates<Id>,          // unchanged
    /// Stage 1, Pass 1: how many whole-workload candidates compose; Pass 2 adds to them.
    pub combinations: usize,                                                 // unchanged
    /// Stage 1, Pass 2: the candidates the identical-expression rule added, in enumeration order.
    pub stage1_pass2_identical_expression: Vec<IdenticalExpressionCandidate>, // new
    /// Stage 1: every whole-workload candidate, each Pass 1 one followed by its Pass 2 ones.
    pub stage1_logical_asap_optimization: Vec<LogicalCandidate>,             // unchanged
    /// Stage 2: one physical candidate `P<n>` per implemented `L<n>`.
    pub stage2_physical_asap_optimization: Vec<PhysicalCandidate>,           // unchanged
    /// Stage 2: the Stage 1 candidates it could not implement, with reasons.
    pub stage2_pruned: Vec<PrunedCandidate>,                                 // unchanged
    /// Stage 3: plan selection over the Stage 2 candidates, or why it failed.
    pub stage3_plan_selection: Result<Selection, SelectionError>,            // unchanged
    /// Stage 3: the selected candidate's logical roots with confirmed weight requirements;
    /// `None` when selection failed.
    pub stage3_selected_logical: Option<Vec<QueryRoot>>,                     // new
}
impl<Id> Stage1ToStage3Run<Id> {
    /// The selected logical plan, with the weight requirements Stage 3 confirmed made proofs.
    pub fn selected_logical(&self) -> Option<&[QueryRoot]>; // now the confirmed plan
}

/// Lazy, uncapped Stage 1 enumeration; a caller that limits candidates stops consuming it.
// new: lazy Stage 1 enumeration, Pass 1 choices each followed by their Pass 2 variants
pub fn stage1_logical_asap_optimization<Id: Clone>(
    per_computation: &PerComputationCandidates<Id>,
) -> impl Iterator<Item = Stage1Candidate> + '_;
/// Runs Stage 2 and Stage 3 over `stage1` (all candidates, or the ones a caller
/// enumerated); candidate `n` becomes `L<n>`, implemented as `P<n>`.
// new: Stage 2/3 over caller-enumerated candidates
pub fn run_stage2_and_stage3<Id: Clone + std::fmt::Debug>(
    per_computation: PerComputationCandidates<Id>,
    stage1: Vec<Stage1Candidate>,
    targets: &[Option<AccuracyTarget>],
    recurrence: &[QueryRecurrence],
    data: &DataWorkload,
    models: PlanningModels<'_>,
) -> Stage1ToStage3Run<Id>;
/// Runs Stages 1 to 3 over the frontends' roots; `recurrence[i]` is how often root `i`
/// is evaluated. Only a Pass 1 failure is an `Err`.
pub fn run_stage1_to_stage3<Id: Clone + std::fmt::Debug>(
    roots: Vec<(Id, QueryRoot)>,
    targets: &[Option<AccuracyTarget>],
    recurrence: &[QueryRecurrence],                // new parameter
    data: &DataWorkload,
    models: PlanningModels<'_>,
) -> Result<Stage1ToStage3Run<Id>, PerComputationCandidateError>;
/// As `run_stage1_to_stage3`, from an existing Pass 1 inventory, so a caller can
/// inspect it before enumerating every Stage 1 candidate.
pub fn run_stage1_to_stage3_from_pass1<Id: Clone + std::fmt::Debug>(
    per_computation: PerComputationCandidates<Id>,
    targets: &[Option<AccuracyTarget>],
    recurrence: &[QueryRecurrence],                // new parameter
    data: &DataWorkload,
    models: PlanningModels<'_>,
) -> Stage1ToStage3Run<Id>;

crates/asap-aware-mapping/src/per_computation_candidates.rs

/// One Pass 1 candidate for a computation (one measure of an aggregate).
/// The type stays; it gains `WholeExpression`.
pub enum ComputationCandidate {
    /// Plain exact execution of the original sub-DAG, when no exact accumulator form applies.
    Exact,                                                         // unchanged
    /// An exact mergeable accumulator state.
    ExactAccumulator { kind: ExactKind, params: ExactParams },     // unchanged
    /// An unsized summary family, sized later for the measure's accuracy.
    Summary(SketchAlgorithm),                                      // unchanged
    /// `avg` rewritten as exact `sum` and `count` accumulator states.
    AvgAsSumCount,                                                 // unchanged
    /// Top-k as one unsized heap summary over the input of the per-item sum/count beneath;
    /// replaces the same-family heap over the exact per-item result.
    WholeExpression(SketchAlgorithm), // new: whole-expression Count-Min/CountSketch + heap absorbing the inner sum/count
}

/// One measure of an aggregate with at least one realization besides exact execution.
/// The type stays; it gains `absorbs` and `samples_of`.
pub struct ComputationTarget {
    /// The aggregate node this target realizes.
    pub target: Rc<OperatorNode>,     // unchanged
    /// Index into the aggregate's measures.
    pub measure: usize,               // unchanged
    /// The candidates for this measure, exact first.
    pub choices: ComputationChoices,  // unchanged
    /// The measure's own filter; only rows that pass it update the state (`SummaryAgg.filter`).
    pub filter: Option<Predicate>,    // unchanged
    /// The aggregate's own `having`, then one entry per consumer `Filter` directly above it.
    pub having: Vec<HavingEntry>,     // unchanged
    /// The target directly beneath that this target's `WholeExpression` candidates
    /// absorb (not computed when chosen); `None` otherwise.
    pub absorbs: Option<usize>,       // new: the target a WholeExpression candidate absorbs
    /// The metric whose raw samples (or plain sums) are every value the aggregate reads;
    /// `None` otherwise. Heap weights from it carry `WeightDomain::SamplesOfMetric`.
    pub samples_of: Option<String>,   // new: the metric whose raw samples feed the aggregate
}

impl<Id> PerComputationCandidates<Id> {
    /// Whether `choice` picks a whole-expression candidate and a non-first candidate for
    /// the target it absorbs (a duplicate of the first), so enumeration skips it.
    // new; combinations() now excludes choices differing only in an absorbed target
    pub fn absorbed_choice_varies(&self, choice: &[usize]) -> bool;
}

crates/asap-aware-mapping/src/stage3_plan_selection.rs, analytical_cost.rs

/// Unit of every Stage 3 cost: cost per second of wall time (one unit is one
/// CPU-millisecond under `illustrative_calibration`).
pub const COST_UNIT: &str = "cost_per_second";                              // was "cpu_ms_per_workload_evaluation"
/// Label naming the cost model and statistics the costs come from.
pub const COST_SOURCE: &str = "analytical-cost-v2 (illustrative statistics)"; // was v1

/// Stage 3: rejects candidates that miss a query's accuracy target, prices the rest
/// per second and selects the cheapest (first on ties).
pub fn select_plan(
    cands: &[PhysicalCandidate],
    targets: &[Option<AccuracyTarget>],
    recurrence: &[QueryRecurrence],   // new: prices per second from each query's recurrence
    data: &DataWorkload,
    models: PlanningModels<'_>,
) -> Result<Selection, SelectionError>;
/// Coefficients used until a deployment is calibrated: 1 ns CPU per op, 0.1 ns per
/// scanned byte, 1 GB kept for a second = 125 CPU-ms, one-off work spread over an hour.
pub fn illustrative_calibration() -> ResourceCalibration;                              // new
/// Answers an open `SamplesOfMetric` requirement from `data`'s declared metric types:
/// proven for a declared counter, unknown otherwise.
pub fn confirmed_weights(domain: &WeightDomain, data: &DataWorkload) -> WeightDomain;   // new
/// `roots` with every open weight requirement replaced by its answer: the form of a
/// selected plan the executor runs.
pub fn with_confirmed_weights(roots: &[QueryRoot], data: &DataWorkload) -> Vec<QueryRoot>; // new

/// Stage 3's result. The type stays; it gains `plan`.
pub struct Selection {
    /// The selected physical candidate's id, e.g. "P1".
    pub selected: String,                          // unchanged
    /// The selected candidate's roots with every confirmed weight requirement made a proof: the plan to run.
    pub plan: Vec<QueryRoot>,                      // new
    /// Each valid candidate's cost, by id.
    pub costs: BTreeMap<String, CandidateCost>,    // unchanged
    /// Every other candidate: invalid with the reason, or valid but costlier/tied.
    pub rejected: Vec<Rejection>,                  // unchanged
}
/// One node's price. The type stays; it gains `per_second`.
pub struct NodeCost {
    /// Price per second: one computation's price × `multiplier` × `per_second`.
    pub cost: f64,                                 // unchanged (now per second)
    /// The consumers it is computed for; 1 when materialized.
    pub multiplier: u64,                           // unchanged
    /// Computations per second, from the readers' recurrences.
    pub per_second: f64,                           // new
    /// A human-readable note, e.g. "scan 3000000 samples".
    pub detail: String,                            // unchanged
}
/// Why plan selection produced no selection. The type stays.
pub enum SelectionError {
    /// A candidate's root count differs from the targets or recurrences given;
    /// a caller bug (mismatched inputs).
    TargetCount { candidate: String, roots: usize, targets: usize, recurrences: usize },  // recurrences new
    /// Pricing one node failed; an internal bug or an uncostable operator.
    Cost { candidate: String, node: PhysicalASAPNodeId, error: AnalyticalCostError },     // unchanged
    /// Every candidate was rejected; an unsatisfiable request, with each rejection.
    NoValidCandidate(Vec<Rejection>),                                                     // unchanged
}

// analytical_cost.rs
/// Conversion from physical dimensions to one deployment-specific cost objective.
/// The type stays; it gains two fields.
pub struct ResourceCalibration {
    /// Price of one CPU operation.
    pub cost_per_cpu_op: f64,                 // unchanged
    /// Price of one scanned byte.
    pub cost_per_scan_byte: f64,              // unchanged
    /// Price of one byte held by a computation.
    pub cost_per_retained_byte: f64,          // unchanged
    /// Price of state kept across evaluations, per byte per second; default 0.
    #[serde(default)]
    pub cost_per_retained_byte_second: f64,   // new: price of state kept across evaluations
    /// Seconds over which one-off, scheduled and unknown queries are spread;
    /// default `DEFAULT_HORIZON_S`.
    #[serde(default = "default_horizon_s")]
    pub horizon_s: f64,                       // new: spread for one-off, scheduled and unknown queries
    /// Names the calibration in cost sources.
    pub version: String,                      // unchanged
}
/// Default `horizon_s`: one hour.
pub const DEFAULT_HORIZON_S: f64 = 3_600.0;   // new

crates/asap-aware-mapping/src/accuracy/evidence.rs, composition.rs

/// Statistics a propagation rule may consult; every field defaults to unknown.
/// The type stays; it gains `sign_statistic`.
pub struct PropagationStats {
    /// Certified ranges for a division's true numerator and denominator.
    pub division_operand_domains: Option<[OperandDomain; 2]>,   // unchanged
    /// Provenance of the supplied evidence.
    pub evidence_provenance: Vec<GuaranteeSource>,              // unchanged
    /// Whether every input value is non-negative.
    pub values_non_negative: Option<bool>,                      // unchanged
    /// Names the statistic an unknown sign is gated on when the values are one metric's
    /// raw samples or sums; plan selection binds it to 1 for a declared counter. Default `None`.
    pub sign_statistic: Option<String>, // new: names the values_non_negative[M] statistic gating a relative bound
    /// Input rows an exact aggregation folds.
    pub input_row_count: Option<u64>,                           // unchanged
    /// The statistic an unknown row count is named by (default `input_row_count`).
    pub row_count_statistic: Option<&'static str>,              // unchanged
    /// Key-frequency distribution evidence.
    pub data_distribution: Option<DataDistribution>,            // unchanged
    /// Lower confidence bound of the k-th selected top-k item.
    pub topk_selected_lower_bound: Option<f64>,                 // unchanged
    /// Greatest upper confidence bound among excluded top-k items.
    pub topk_excluded_upper_bound: Option<f64>,                 // unchanged
    /// Union-bound failure probability of the top-k margin intervals.
    pub topk_interval_failure_probability: Option<f64>,         // unchanged
    /// Hydra shared-grid collision error.
    pub hydra_shared_grid_collision_bound: Option<f64>,         // unchanged
    /// Failure probability of the Hydra shared-grid term.
    pub hydra_shared_grid_failure_probability: Option<f64>,     // unchanged
}
/// The `values_non_negative[metric]` statistic name for samples of `metric`.
pub(crate) fn samples_sign_statistic(metric: &str) -> String;                    // new
/// The metric a `samples_sign_statistic` name refers to; `None` if it is not one.
pub(crate) fn parse_samples_sign_statistic(statistic: &str) -> Option<&str>;     // new

crates/types/src/post_asap/sketch.rs, workload.rs, ir/schema_support.rs

/// What is known about the sign of a summary update's weights. The type stays; it gains `SamplesOfMetric`.
pub enum WeightDomain {
    /// Nothing proven; the weights may be negative. The default.
    UnknownOrSigned,                                // unchanged
    /// The weights are proven non-negative, by `proof`.
    NonNegative { proof: NonNegativeWeightProof },  // unchanged
    /// Open requirement: weights are raw samples (or plain sums) of `metric`; plan
    /// selection confirms it, and unconfirmed it counts as not proven.
    SamplesOfMetric { metric: String },   // new: open requirement recorded by Pass 1
}
impl WeightDomain {
    /// Answers an open `SamplesOfMetric`: `CounterSamples` proof if `is_counter`, else
    /// `UnknownOrSigned`; any other domain is returned unchanged.
    pub fn confirmed(&self, is_counter: impl Fn(&str) -> bool) -> WeightDomain; // new
}
/// Why weights are non-negative. The type stays; it gains `CounterSamples`.
pub enum NonNegativeWeightProof {
    /// Every weight is 1 (a count).
    UnitCount,                     // unchanged
    /// The weights are a reset-aware counter derivative (`rate`, `increase`).
    ResetAwareCounterDerivative,   // unchanged
    /// The weights are raw samples of a metric the data workload declares a counter.
    CounterSamples,                // new
}

/// The data workload: statistics and declarations about the data being queried.
/// The type stays; it gains `metric_types`.
pub struct DataWorkload {
    /// Whether data keeps arriving or is at rest.
    pub arrival: DataArrival,                                          // unchanged
    /// Cadence at which each PromQL source supplies a sample.
    pub data_ingestion_interval: Evidence<DurationMs>,                 // unchanged
    /// Total ingested volume.
    pub ingestion_volume: Evidence<u64>,                               // unchanged
    /// Rows per second arriving.
    pub ingestion_rate: Evidence<Rate>,                                // unchanged
    /// Input rows (series) per evaluation.
    pub input_cardinality: Evidence<u64>,                              // unchanged
    /// Key-frequency distribution.
    pub distribution: Evidence<DataDistribution>,                      // unchanged
    /// Per-source L1/L2 norm bounds for Count-Min / Count Sketch errors.
    pub stream_norms: Evidence<BTreeMap<String, StreamNorms>>,         // unchanged
    /// Per-source bound on distinct items per evaluation (classic HLL's failure probability).
    pub max_distinct_per_evaluation: Evidence<BTreeMap<String, u64>>,  // unchanged
    /// Declared Prometheus type of each metric by name; never inferred from a name.
    /// Default empty (an undeclared metric's samples may take any value).
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub metric_types: BTreeMap<String, MetricType>, // new: optional in workload JSON
}
/// The Prometheus type of a metric's samples: `Counter` (only increases or resets to
/// zero, so every sample is non-negative) or `Gauge` (may take any value).
#[serde(rename_all = "snake_case")]
pub enum MetricType { Counter, Gauge } // new

/// Resolves a PromQL root to rows carrying the series identity before candidate search.
// now adds a (series identity, timestamp) unique key so Pass 2 may share the node
pub fn with_promql_series_identity(root: &Rc<OperatorNode>) -> Result<Rc<OperatorNode>, String>;

crates/devtools/src/planning_stages.rs, bin/stage_pipeline.rs

/// Runs the planning stages over roots lowered from any query language; query `i` has
/// target `targets[i]` and recurrence `recurrence[i]`.
pub fn run_planning_stages_over(
    roots: Vec<QueryRoot>,
    targets: &[AccuracyTarget],
    recurrence: &[QueryRecurrence],   // new parameter
    data: &DataWorkload,
    max_combinations: Option<usize>,  // now a Stage 1 candidate limit, Pass 2 included
) -> Result<PlanningStagesRun, PlanningStagesError>;

/// Why a planning-stages run produced nothing. The type stays; one variant is replaced.
pub enum PlanningStagesError {
    /// Pass 1 failed: there are no candidates, only the frontends' roots.
    Pass1(String),                       // unchanged
    /// Stage 1 enumerated more candidates than the caller's limit; enumeration stopped and
    /// nothing was planned. An unsatisfiable request under that limit, not a bug.
    TooManyCandidates { limit: usize }, // was TooManyCombinations { combinations, limit }
    /// The frontends' roots or a composed candidate failed validation; an internal bug.
    Validation(String),                  // unchanged
}
// CLI flag: the most Stage 1 candidates a run enumerates; past it the run writes no document.
// stage_pipeline --max-combinations: now a Stage 1 candidate limit (default 10^4); exit 3 unchanged

Usage, end to end (checked with a scratch test over Example 1 at 80fc72e1): the library path a caller such as stage_pipeline takes, from the user's workload to the selected plan.

// What the user writes: Example 1's two panels, both every 10 s over the last minute.
//   Q1 "sum by (job) (rate(http_requests_total[1m]))", exact
//   Q2 "topk by (job) (10, sum_over_time(http_requests_total[1m]))", ε = 0.01, δ = 0.001
// plus data_workload.metric_types = { "http_requests_total": Counter } (new in this PR).
// `workload` is read below for the queries, their targets and recurrences, and the data facts.
let workload = example1();

// `roots`: the PromQL frontend's DAG, one root per query (Stage 0); the input to Pass 1.
let roots = lower_promql_roots(&workload)?;
// `targets`: each query's accuracy target [Exact, EpsilonDelta { 0.01, 0.001 }]; Stage 3 checks guarantees against it.
let targets: Vec<_> = workload.query_workload.entries().map(|e| Some(e.requirements.accuracy.target())).collect();
// `recurrence`: each query's demand, [FixedInterval(10 s); 2]; Stage 3 turns it into cost per second (new).
let recurrence: Vec<_> = workload.query_workload.entries().map(|e| e.recurrence).collect();
// `data`: the data workload; Stage 3 prices rows from it and confirms counter samples from `metric_types`.
let data = workload.data_workload.clone().unwrap_or_default();

// `per_computation`: Stage 1 / Pass 1's candidates per computation. Q2's top-k offers exact,
// whole-expression CMS+heap and whole-expression CountSketch+heap (new: the heaps absorb the inner
// per-job sum and record their weights as WeightDomain::SamplesOfMetric { metric: "http_requests_total" }).
// 3 whole-workload combinations; it is the input to the Stage 1 enumeration.
let per_computation = generate_per_computation_candidates(roots.iter().cloned().enumerate().collect())?;

// `stage1`: Stage 1's lazy enumeration, collected: each Pass 1 combination followed by its Pass 2
// identical-expression candidates (new), 3 × (independent, shared Scan #1, shared Scan #1 + TimeRange #2)
// = L1..L9. A caller with a limit (stage_pipeline --max-combinations) stops consuming the iterator there.
let stage1: Vec<Stage1Candidate> = stage1_logical_asap_optimization(&per_computation).collect();

// `run`: Stage 2 implements L<n> as P<n>; Stage 3 binds the counter requirement from `data`,
// checks every target, prices every valid candidate in cost per second, and selects the cheapest.
let run = run_stage2_and_stage3(per_computation, stage1, &targets, &recurrence, &data, PlanningModels::builtin());

// `selection`: what the user gets, the selected plan and every other candidate's verdict.
let selection = run.stage3_plan_selection.as_ref()?;
// selection.selected == "P1"   all exact, 7.94021 cost_per_second
// P2, P3 (shared scan, shared scan and range): valid, "tied with the selected candidate P1 at 7.940 cost_per_second; lost on enumeration order"
// P4–P6 (CMS+heap): weights confirmed non-negative from Counter, rejected only because "q2: the output has no guarantee, …"
//   (without the Counter declaration: "q2: CmsWithHeap needs non-negative update weights, and these are not proven non-negative …")
// P7–P9 (CountSketch+heap): "q2: the output has no guarantee, …"

// `plan`: the selected candidate's two logical roots with every weight requirement Stage 3 confirmed
// turned into a proof; this is what the executor is handed, so a selected Count-Min heap is accepted.
let plan = run.selected_logical();

Before this PR

• Example 1. Stage 1 had 3 candidates and no sharing. Q2's heaps ranked an exact per-series sum, and Count-Min was always rejected on its weights.
• SQL top-k. With an Int64 key the heap candidates failed to compose; with a key named value the heap was weighted by the key.
• Costs were per workload evaluation, ignoring recurrence. A computation was charged for the memory it held while running, and kept state was not priced per second.
• Devtool limit. stage_pipeline refused workloads by their Pass 1 combination count only.

After this PR

• Example 1 has 9 Stage 1 candidates: 3 Pass 1 × (independent, shared scan, shared scan and range).
  • Q2 offers exact, whole-expression Count-Min + heap and whole-expression CountSketch + heap.
  • With http_requests_total declared a counter, Count-Min is rejected only because a top-k output has no guarantee; without the declaration it is rejected on its weights.
  • The shared-input plans tie with the independent one, which is kept.
• Example 3a has 972 candidates.
• Library callers of run_stage1_to_stage3 get Selection::plan and selected_logical() with confirmed weights, so the executor accepts a selected Count-Min heap.
• Evidence: the selected plan and total per example, measured at the previous head of this PR and at this head.

Example	Before (previous head)	After
1	P1, exact Q2: 8.105	P1, exact Q2: 7.940
2	P1, all exact: 13.440	P1, all exact: 8.640 (tied with its shared-input plans)
3a	P433, KLL for Q1–Q2, exact for Q3–Q5: 21725.736	P485, KLL for all five: 21724.827; all-exact costs about 16% more
3b	P1, exact: 2.720	P2, KLL: 2.100, tied with DDSketch; exact 2.400

Totals are cost per second under the illustrative calibration. Screenshots: not applicable.

Verification

• Failing test first for each fix and decision:
  • SQL Int64 key and key named value.
  • Kept-state-only memory.
  • Sign binding for declared counters, and the gate opening only at 1.
  • Candidate limit including Pass 2.
  • The confirmed plan reaching every consumer.
  • An open requirement not certifying Count-Min.
  • 20 nested sharing groups enumerated quickly.
• Brute-force equivalence of the Pass 2 enumeration on small cases, and a label/roots agreement test.
• Example acceptance tests:
  • Example 1: the Pass 2 tests are un-ignored, the accumulator count and Count-Min reasons are updated, and a tie-tolerance test is added.
  • Example 3: the Pass 2 test is un-ignored.
  • The whole-expression heaps are now checked: the raw-sample precompute test requires both heap families, and the numeric-regression test is un-ignored.
• Fixtures and docs. The viewer fixtures are regenerated. The Example 3 acceptance spec and the stage viewer guide state the new selections.
• Full CI gate on the remote runner at d542c30: fmt and clippy --all-features --locked -D warnings pass, and nextest over the workspace passes 1650 tests (46 skipped).

Limitations and follow-ups

• The pricing is illustrative until cost model 2 (Plan selection cost models: usage over time (cost model 1), peak capacity (cost model 2), own-cluster trade-offs (cost model 3) #683): query-time sketches carry no memory charge, and grouped sketch builds are not charged per-group hashing or initialization.
• select_plan uses illustrative_calibration(); a pluggable calibration is G9.
• evaluation_rate ignores alignment and phases, and one-off queries use the largest invocation count.
• Example 3b's KLL and DDSketch plans tie; KLL wins on enumeration order. Example 3a has similar ties.
• Whole-expression heaps need an open schema (PromQL only).
• The design doc's Example 1 says 2 input choices; the code gives 3, per Algorithm 2 line 31. This and the Example 1 acceptance doc's accumulator count and Count-Min reason are noted on Update the docs after the G1–G11 consolidation groups land #677.
• weighted_topk_binding stays ignored on the heap estimate's output shape (fix(ir): top-k sketch readouts return selected rows; Example 1 expectations follow planner output #579).
• The default-limit devtool test enumerates 10^4 candidates (about 90 s in a debug build).

Supersedes #588, #589, #640, #593 and #594, which are left open.