feat(planner): enumerate local logical alternatives without execution timing

[zzylol]

Rebased on main d4869a7 (DF 54).

Now sits on #543 (Phase C). The base is now stack/528-08-cleanup, and this PR is C1 of the Phase C stack: #575 Stage 1 workload candidates, #576 Stage 2/3, #577 Example 1 acceptance. A follow-up commit calls the promoted asap_frontend_promql::lower_promql_query_workload. #543 removed the legacy realizations_for_intent/QueryExpr path, so the mentions of it below describe the code before #543.

Problem: Pass 1 has no logical-only, unranked candidate list over the unified IR

#509 §1 "Logical ASAP-aware optimization", Pass 1: Local candidate generation, says:

For each eligible sub-DAG, Pass 1 identifies its computation semantics, applies rewrite rules, and generates every candidate that is not provably unable to meet its accuracy requirement.

#509 §"Stages and their decisions" also says that only stage 3 (plan selection) may discard a valid candidate, and that it does so with the deployment's empirical cost and accuracy models. Pass 1 gets only the logical DAGs, the accuracy requirements, time_selection and the repetition interval. It gets no cost model.

Its candidate table includes the exact option for every computation:

Original computation	Local candidates (#509)
Sum(x) by (g)	Exact grouped sum
TopK(k, x) by (g)	Exact sort and limit per group, Count-Min Sketch with a top-k heap per group, Hydra over all groups
Distinct(x)	Exact distinct, a specialized distinct summary, UnivMon
Quantile(x, window)	Exact quantile, KLL over the requested window

Before this PR, the only enumeration is the legacy replacement::realizations_for_intent(intent, cost_model). It works on the legacy QueryExpr graph, not on OperatorNode/QueryRoot from #511. It also breaks the Pass 1 contract in three ways:

1. It needs a cost model and ranks with it. The list is "exhaustive and ranked (most-preferred first) via cost_model". Ranking is a stage 3 decision.

2. It drops exact execution for approximate requests. For an approximate request it returns only the sketches. Count also gets its exact accumulator, but no intent gets PassThrough (the original sub-DAG run exactly):

   Quantile{q=0.99, ε=0.05, δ=0.01}  →  [Kll, DDSketch]                      (ranked)
   Count{ε=0.05, δ=0.01}             →  [Cms, CountSketch, UnivMon, ExactAggregate{Count}]
   

   So "Exact quantile" from the table above is missing. Stage 3 can never choose it, even when it would be cheaper.

3. It gives an exact request an approximate choice. Count{Exact} returns [ExactAggregate{Count}, Sketch(UnivMon)].

This PR covers the per-aggregate part of Pass 1 on the unified IR: find every single-measure aggregate reachable from the workload roots, and list its exact and summary choices without ranking. It does not build replacement sub-DAGs, apply rewrite rules (for example recognizing Entropy/L2 forms), produce Hydra (a grouping-level choice), split multi-measure aggregates, or do any Pass 2 sharing.

Proposed method

The new module asap_aware_mapping::logical_candidates runs in Pass 1. It takes the named roots from a frontend (for example #540's lower_promql_query_workload) and returns an inventory. It does not change the roots.

enumerate_local_logical_candidates(roots):

1. Calls QueryRoot::validate_structure() on every root. A structural error is returned as LogicalCandidateError::Structure.
2. Finds the operator nodes each root reads. For QueryRoot::Operator(node) that is node. For QueryRoot::Scalar(expr) it is expr.operator_refs() (for example the plan inside scalar(...) or a SQL scalar subquery).
3. Walks every node reachable from them with OperatorNode::reachable. That walk also follows operator nodes referenced from scalar expressions inside operators, because OperatorNode::children() includes them.
4. Visits each node once. One HashSet of Rc pointers is shared across all roots, so a producer read by two roots becomes one target.
5. Rejects any node with timing.is_some() (AssignedTiming). Execution timing is assigned in physical planning (docs: propose workload-wide planning, summary sharing, and materialization #509 stage 2), so it must not be present yet.
6. For each NonASAPOp::Aggregate with exactly one measure, calls local_realizations_for_intent and stores a LocalLogicalTarget { target, alternatives }. An aggregate with several measures is skipped and stays as it is in the roots.

local_realizations_for_intent(intent) builds the list in a fixed order:

1. Realization::PassThrough first, always. This is exact execution of the original sub-DAG.
2. Realization::ExactAggregate { kind, params } when the intent has an exact mergeable accumulator: Count, Sum, Min, Max, Rate, IRate, Increase. Count gets it for both exact and approximate targets.
3. If the intent carries an accuracy target (accuracy_target(intent) is Some) and that target is not Exact:
   • Resolve (epsilon, delta) with the existing accuracy_budget. Epsilon(e) uses DEFAULT_DELTA = 0.01.
   • Reject the target unless epsilon is finite and > 0 and delta is finite and in (0, 1) (InvalidAccuracy).
   • Add one Realization::Sketch per algorithm in summary_candidates(intent), in catalog order. Each is sized with default_size_params(algorithm, intent, epsilon, delta).

The function takes no cost model, accuracy model, runtime capabilities or storage policy. The sketch sizes are nominal dimensions from the built-in sizing contracts. They do not certify that a deployment meets the accuracy target; stage 3 checks that. The list order has no preference meaning.

The legacy ranked path stays for the existing pipeline until the planner cutover. This PR adds the new entry point beside it.

Key code interfaces

crates/asap-aware-mapping/src/logical_candidates.rs (new; pub mod logical_candidates in lib.rs):

/// All local realizations of one single-measure aggregate.
#[derive(Debug, Clone)]
pub struct LocalLogicalTarget {
    pub target: Rc<OperatorNode>,
    pub alternatives: Vec<Realization>,
}

/// Compact Pass 1 inventory; roots and nested producer dependencies are retained.
#[derive(Debug, Clone)]
pub struct LocalLogicalCandidates<Id> {
    pub roots: Vec<(Id, QueryRoot)>,
    pub targets: Vec<LocalLogicalTarget>,
}

#[derive(Debug, Error)]
pub enum LogicalCandidateError {
    Structure(#[from] SchemaDerivationError),
    AssignedTiming,
    InvalidAccuracy,
}

pub fn local_realizations_for_intent(
    intent: &AggIntent,
) -> Result<Vec<Realization>, LogicalCandidateError>;

pub fn enumerate_local_logical_candidates<Id>(
    roots: Vec<(Id, QueryRoot)>,
) -> Result<LocalLogicalCandidates<Id>, LogicalCandidateError>;

The alternatives reuse the existing Realization enum from replacement.rs unchanged. This PR produces only these variants:

pub enum Realization {
    ExactAggregate { kind: ExactKind, params: ExactParams },
    Sketch(SketchKind),
    PassThrough,
    // Sample / Wavelet / StatModel exist but are never produced here.
}

Usage, from the frontend to the inventory:

let roots = asap_frontend_promql::unified::lower_promql_query_workload(&workload, 0)?;
let candidates = enumerate_local_logical_candidates(roots.into_iter().enumerate().collect())?;
for t in &candidates.targets {
    // t.target: the original Aggregate node; t.alternatives: unranked choices
}

Fields

LocalLogicalTarget

Field	Type	Meaning
target	Rc<OperatorNode>	The original NonASAPOp::Aggregate node, the same Rc as in the roots (pointer-equal, not a copy). It keeps the source, grouping (reduction), filters, input expressions and evaluation context. timing is None.
alternatives	Vec<Realization>	Unranked local choices for its single measure. Never empty. The first entry is always PassThrough.

LocalLogicalCandidates<Id>

Field	Type	Meaning
Id	type parameter	Caller's name for a root (query index, query string, …). Not interpreted.
roots	Vec<(Id, QueryRoot)>	The input roots, returned unchanged and in input order. They still contain multi-measure aggregates and every non-aggregate operator.
targets	Vec<LocalLogicalTarget>	One entry per distinct single-measure aggregate reachable from any root, in discovery order (roots in order, nodes parent-before-child). No two entries share a node pointer.

LogicalCandidateError

Variant	When
Structure(SchemaDerivationError)	QueryRoot::validate_structure() failed for a root.
AssignedTiming	A reachable node already has timing: Some(_). Pass 1 input must have no execution timing assigned; timing is a Stage 2 materialization decision.
InvalidAccuracy	The intent's target is approximate and the resolved epsilon is not finite or <= 0, or delta is not finite or not in (0, 1).

local_realizations_for_intent

Parameter / result	Meaning
intent: &AggIntent	One aggregate measure. Its accuracy target (for Quantile, Cardinality, FrequencyL2, FrequencyEntropy, Count, TopK) decides whether sketches are added.
Ok(Vec<Realization>)	PassThrough, then the exact accumulator if any, then sketches in summary_candidates order.

enumerate_local_logical_candidates

Parameter / result	Meaning
roots: Vec<(Id, QueryRoot)>	Named workload roots from a frontend. Taken by value and returned in roots.
Ok(LocalLogicalCandidates<Id>)	The inventory described above.

Realization variants produced here

Variant	Meaning here
PassThrough	Execute the original sub-DAG exactly. Always present.
ExactAggregate { kind, params }	An exact mergeable accumulator. kind: ExactKind is one of Count, Sum, Min, Max, Rate, IRate, Increase. params: ExactParams is the matching variant with the same name (exact accumulators have no tuning parameters).
Sketch(SketchKind)	One sketch algorithm with nominal size parameters. Built with SketchKind::new(algorithm, default_size_params(...)). SketchKind::algorithm() returns the SketchAlgorithm.

Examples

End to end: #509 Example 1's rate query

Input (from tests/logical_candidates.rs, promql_lowering_reaches_local_candidates_without_execution_timing): a PromQL workload with one entry, sum by (job) (rate(http_requests_total[1m])), accuracy EpsilonDelta { epsilon: 0.05, delta: 0.01 }, ingestion interval 1 s.

1. feat(promql): lower queries to unified operator and scalar IR #540's lower_promql_query_workload produces one operator root:

   Aggregate{ by (job), [Sum] }
   └── Aggregate{ PerEntity, [Rate] }
       └── TimeRange{ 1m, Range }
           └── Scan{ http_requests_total }
   

2. enumerate_local_logical_candidates(vec![(0, root)]) validates the root, walks the four nodes once each, and finds no assigned timing.

3. Both aggregates have one measure, so both become targets. Neither Sum nor Rate carries an accuracy target, so no sketch is added even though the query is approximate:

   Target	alternatives
   Aggregate{by (job), [Sum]}	PassThrough, ExactAggregate{Sum}
   Aggregate{PerEntity, [Rate]}	PassThrough, ExactAggregate{Rate}

The test asserts that some target offers ExactAggregate { kind: Rate } and that every target still has timing: None. No cost model is passed anywhere.

Shared producer read by two roots

scalar_root_producers_are_discovered_once: one Cardinality aggregate over flows is used both as QueryRoot::Scalar(ScalarExpr::ScalarSubquery(producer)) and as QueryRoot::Operator(producer). The result has 2 roots and 1 target, and Rc::ptr_eq(&targets[0].target, &producer) holds. Both roots still export with compile_logical_asap_query(..).validate(), and the producer has no timing and no guarantee.

Per-intent results

From tests/logical_candidates.rs and the unit test in logical_candidates.rs. "approx" means EpsilonDelta { epsilon: 0.05, delta: 0.01 }.

Intent	Accuracy	Result
Count	approx	PassThrough, ExactAggregate{Count}, Cms, CountSketch, UnivMon (unit test checks PassThrough, exact Count and UnivMon)
Count	Exact	PassThrough, ExactAggregate{Count}; no sketch
Cardinality{cols: [0]}	approx	PassThrough, Hll, Theta, Kmv, UnivMon (#509 Example 2)
Cardinality{cols: [0, 1]}	approx	PassThrough, Hll, Theta, Kmv; no UnivMon for a tuple
FrequencyL2 / FrequencyEntropy	approx	PassThrough, UnivMon
Quantile{q: 0.99}	approx	PassThrough, Kll, DDSketch
Quantile{q: 0.99}	Exact	exactly [PassThrough]
TopK{k: 10}	approx	PassThrough, CmsWithHeap, CountSketchWithHeap
Sum, Min, Max, Rate, IRate, Increase	none	PassThrough, matching ExactAggregate
Avg, StdDev, HistogramQuantile, Extension, …	—	[PassThrough]
Count	Epsilon(NaN)	Err(InvalidAccuracy)
Count	EpsilonDelta{epsilon: 0.1, delta: 0.0}	Err(InvalidAccuracy)
any root with a node whose timing = Some(QueryTime)	—	Err(AssignedTiming)
Aggregate with two measures	—	not a target; left intact in roots

Compared with the legacy path

Request	Legacy realizations_for_intent	This PR
Count, approx	ranked sketches, then ExactAggregate{Count}; no PassThrough	PassThrough, ExactAggregate{Count}, sketches in catalog order
Quantile, approx	ranked Kll/DDSketch only	PassThrough, Kll, DDSketch
Count, Exact	ExactAggregate{Count}, UnivMon	PassThrough, ExactAggregate{Count}
Extension	cost_model.realize_extension(..)	PassThrough only

Out of scope

• Building or validating replacement sub-DAGs (SummaryAgg/SummaryEstimate nodes). The descriptors are not executable plans, and taking the first one is not a selection policy.
• Rewrite rules, Hydra and other grouping-level choices, multi-measure splitting, and extra choices for opaque Extension intents.
• Pass 2 sharing (docs: propose workload-wide planning, summary sharing, and materialization #509 §Pass 2), physical planning, cost/accuracy assessment and selection.
• Cutting the production pipeline over from the legacy ranked search. That happens in a later PR (refactor(planner): plan one shared unified workload DAG #542).

API notes and boundaries: docs/develop_docs/local-logical-candidates.md.

Stack and validation

Revised logical foundation 5/5 · Previous: #540 · Subsequent physical scopes pending reorganization · Tracker: #528

Order: #567 → #560 → #537 → #539 → #540 → #561.

• Reproduced the legacy omission of exact execution for approximate count. The new entry point keeps the exact choices and UnivMon.
• Frontend-to-inventory coverage for planner-layering Example 1's rate query; cardinality, frequency and TopK catalogs; exact and tuple restrictions; scalar producer discovery and export with no execution timing assigned.
• Complete first-five tip: 1,961 workspace tests/doctests pass; formatting and workspace/all-target/all-feature Clippy with warnings denied pass.

🤖 Generated with Claude Code

[zzylol]

Parked as draft until Phase C (see #528). Stage 1 Pass 1 local alternatives belong to the #509 end-to-end work, which comes after finishing #511 (Phase A) and the #572 reorganization (Phase B). It will move into the logical-optimizer crate then.

🤖 Generated with Claude Code