feat(planner): compose candidate logical ASAP DAGs from Pass 1 per-computation candidates

[zzylol]

Why

The planner-layering design doc defines the output of Stage 1 (logical ASAP-aware optimization) as candidate logical ASAP DAGs: each is one logical ASAP DAG for the whole workload. Pass 1 (#561, merged) only produces per-computation candidates (ComputationChoices per ComputationTarget). It builds no DAG, so nothing after Pass 1 has a whole-workload candidate to work on, and there is no way to look at what the frontends and Stage 1 produce.

What

• compose_logical_candidate(inventory, choice) in asap-aware-mapping/src/per_computation_candidates.rs: Algorithm 1's SUBSTITUTE. It takes one Pass 1 per-computation candidate per target and builds one whole-workload candidate logical ASAP DAG.
• Every node of a composed candidate gets a theoretical guarantee from the built-in model.
• Summary candidates are kept for combinations of approximate values (comparisons now, as arithmetic before).
• Top-K item identity for composed summaries.
• FinalizeExactAccumulator output naming fix.
• stage_pipeline devtool, which writes stage0_query_language_frontends and stage1_logical_asap_optimization documents (read by its devtools test; the viewer support is feat(dag-viewer): redesign the viewer around the stage pipeline #574), plus the regenerated tools/dag-viewer/examples/planner-layering-example1.json fixture.
• Moved here from Pass 1: single-measure filtered aggregates #608: the rule that a measure filter becomes SummaryAgg.filter.

How

choice[i] indexes targets[i].choices.candidates. The targets (measures) of one aggregate are grouped. An aggregate whose chosen candidates are all Exact is kept as is. Otherwise each measure is substituted on its own over the aggregate's input:

• ExactAccumulator: SummaryAgg with the exact family → FinalizeExactAccumulator.
• Summary: SummaryAgg → SummaryEstimate.
• AvgAsSumCount: the finalized sum and count accumulators, divided. For PromQL this is a vector BinaryOp; for SQL the two states are joined on the group keys (NULL keys match) and divided, with CASE count = 0 THEN NULL for a filtered or global avg.
• Exact (or a measure that is not a target): a single-measure Aggregate.

Then:

• Multi-measure targets: the measures are joined back on the group keys (NULL keys match, as grouping treats them).
• Column names: the substituted output keeps the aggregate's column names (quantile_0_99 no longer becomes quantile, nor avg become sum). A renaming projection keeps its child's guarantee.
• Measure filters become SummaryAgg.filter, so a group with no matching row stays. ASAPOp schemas mark a filtered quantile and a filtered SUM/MIN/MAX as nullable, and the exact Aggregate now marks a filtered quantile nullable too, so every candidate has the same nullability. Subqueries inside a measure filter or HAVING are substituted like any other target.
• Key-only HAVING (BelowAggregate) is pushed below the state as a Filter, only when every consumer applies it. Its key columns are renumbered to the input's (ScalarExpr::map_columns), and the consumer Filter it came from is dropped. It is not a SummaryAgg.filter, which would keep the dropped groups.
• Value HAVING (AboveOutput) stays a Filter above the output; its guarantee is None when the measure is approximate (exact candidates stay exact).
• Sizing: Pass 1 summaries are unsized. Composition sizes each summary for its own measure's accuracy requirement. This is the post-Pass-2 sizing step with no sharing, because Pass 2 (ASAP-aware CSE) is not implemented yet.
• Coverage: summary coverage is derived from the subtree.
• Guarantees (composed_guarantee): exact over exact inputs. A SummaryEstimate over exact input gets the sized family's bound, except for Top-K (a heap sketch bounds item frequencies, not membership) and for Count-Min over weights not proven non-negative, which get None. Anything computed from an approximate value, including value HAVING and any predicate at any depth, gets None. Composing guarantees through operators is Compose accuracy guarantees through arithmetic and comparisons over approximate values #669.
• Combination candidates (replacement.rs): is_supported_exact_binary now accepts comparisons (Compare, CompareBool) as well as arithmetic, so quantile(0.99, …) > quantile(0.5, …) keeps its summary candidates. Certified inputs are not enough for automatic selection, since a difference or ratio of them can be unbounded. Such candidates stay, but are selected automatically only with output-level evidence: a root guarantee that meets the target, or the deployment's empirical output-level accuracy model. The plan-selection part is feat(planner): physical ASAP-aware optimization and plan selection for candidate logical ASAP DAGs #576.
• Top-K item identity: a child that carries the PromQL series identity ranks that column. A closed PromQL child (for example topk(3, sum by (job) (rate(m[5m])))) ranks the tuple of its labels outside the partition, as the legacy current-series and counter rules do.
• FinalizeExactAccumulator named its output state, so a consumer reading the sample value (for example sum by (job) over a finalized per-series rate) found no value column. It now takes the name of the aggregate it realizes.
• stage_pipeline (crates/devtools/src/bin/stage_pipeline.rs) writes an asap-stage-pipeline/v1 document. stage0_query_language_frontends is the frontends' workload DAG. stage1_logical_asap_optimization holds one composed candidate per choice of per-computation candidate for every target, in enumeration order. Every combination is enumerated (no enumeration cap); the document shows the first --max-displayed candidates (default 64) plus the selected one, with combinations and capped (whether some are not shown). A candidate that cannot be composed is recorded with its error, and the run continues.

Key code interfaces added / modified

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

/// Why Pass 1 or composition rejected its input.
pub enum PerComputationCandidateError {
    /// Deriving a node's schema failed: malformed input DAG.
    Structure(#[from] SchemaDerivationError),
    /* ... existing variants ... */
    /// `choice` has the wrong length or an index past a target's candidate
    /// list: bad input from the caller.
    InvalidChoice,               // new: `choice` does not name one listed candidate per target
    /// The chosen combination needs a substitution composition cannot build
    /// yet; the message names the unsupported shape.
    Unsupported(&'static str),   // new: a substitution composition cannot build
}

/// Where a target's `HAVING` is applied during composition.
pub enum HavingPlacement {
    /// Key-only predicate every consumer applies; pushed below the aggregate
    /// as a row predicate.
    BelowAggregate,   // now applied as a `Filter` on the aggregate's input (drops the group)
    /// Predicate on aggregate values; stays a `Filter` above the
    /// finalize/estimate output, so approximate candidates get `guarantee = None`.
    AboveOutput,
}

/// Builds one whole-workload logical candidate by substituting the chosen
/// candidate for every target; callers enumerate `choice` vectors over the inventory.
// new: Algorithm 1's SUBSTITUTE; choice[i] indexes inventory.targets[i].choices.candidates.
// Sizes each summary for its own measure and sets a guarantee on every node.
pub fn compose_logical_candidate<Id: Clone>(
    inventory: &PerComputationCandidates<Id>,
    choice: &[usize],
) -> Result<Vec<(Id, QueryRoot)>, PerComputationCandidateError>;

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

/// The `ColumnRef` naming a schema field (qualified when it has a table);
/// now shared with composition.
pub(crate) fn column_ref(column: &Field) -> ColumnRef;   // was private
/// What a summary consumes per input row for an intent (one column, or the
/// whole tuple for multi-column `COUNT(DISTINCT)`); errors if a tuple leg is missing.
pub(crate) fn summarised_input(                           // was private
    intent: &AggIntent,
    child_schema: &Schema,
) -> Result<SummaryInputExpr, RealizationError>;

/// Whether a binary operator over summary operands keeps its candidates.
fn is_supported_exact_binary(root: &OperatorNode) -> bool;   // now also Compare / CompareBool, not only Arithmetic

crates/types/src/ir/scalar.rs

impl<C> ScalarExpr<C> {
    /// Rebuilds the expression with `f` applied to every operator node it
    /// reads, e.g. to turn `Rc` references into node ids.
    pub fn map_operator_refs<D>(&self, f: &mut impl FnMut(&C) -> D) -> ScalarExpr<D>;   // now delegates to map_parts
    /// Returns the expression with every column reference renumbered by
    /// `column`; operator nodes it reads keep their own column scope.
    // new: renumbers column references; used to push key-only HAVING below the state
    pub fn map_columns(&self, column: &mut impl FnMut(ColumnId) -> ColumnId) -> ScalarExpr<C>
    where
        C: Clone;
}

crates/types/src/ir/asap.rs, non_asap.rs

// Signatures unchanged; schema derivation changes:
// - SummaryEstimate of a Quantile over a filtered SummaryAgg: output field nullable
// - FinalizeExactAccumulator: keeps the SummaryAgg filter (nullable SUM/MIN/MAX) and takes the realized aggregate's column name
// - Aggregate: a filtered Quantile measure is nullable

crates/devtools/src/bin/stage_pipeline.rs (new binary)

# Writes the first two planner stages of a workload as JSON for tools/dag-viewer.
#   --example planner-layering-1  built-in example workload (instead of --promql)
#   --promql <query>              a PromQL query; repeatable
#   --epsilon / --delta           accuracy for every --promql query; omitted means exact; --delta needs --epsilon
#   --interval-ms <u64>           source cadence PromQL needs
#   --max-candidates <n>          cap on listed logical candidates
#   --out <file>                  output path; required
stage_pipeline (--example planner-layering-1 | --promql <query>... [--epsilon <f64> --delta <f64>] [--interval-ms <u64>]) [--max-candidates <n>] --out <file>
# --max-candidates defaults to 64; --interval-ms defaults to 15000

{
  // Document format tag.
  "format": "asap-stage-pipeline/v1",
  // The input queries with their language, text and requirements.
  "workload": { "queries": [{ "id": "q1", "language": "promql", "text": "...", "requirements": { "accuracy": "exact" } }] },
  // The frontends' workload DAG, one root per query.
  "language_specific_frontends": { "dag": { /* FlatDag */ } },
  // Pass 1 whole-workload candidates, one per choice combination, in enumeration order.
  "logical_asap_optimization": {
    // Total number of combinations (product of per-target candidate counts).
    "combinations": 4,
    // True when combinations exceed --max-candidates and the list is truncated.
    "capped": false,
    // Each candidate has a `dag`, or an `error` when composition could not build it.
    "candidates": [{ "id": "L1", "label": "...", "dag": { /* FlatDag */ } }, { "id": "L2", "label": "...", "error": "..." }]
  }
}

Usage:

let inventory = generate_per_computation_candidates(roots.into_iter().enumerate().collect())?;
let choice = vec![0; inventory.targets.len()];   // the exact option of every target
let candidate: Vec<(usize, QueryRoot)> = compose_logical_candidate(&inventory, &choice)?;

Before this PR

Pass 1 lists per-computation candidates for each target, and nothing turns them into candidate logical ASAP DAGs. A comparison of approximate values loses its summary candidates. A finalized exact accumulator's column is called state. Top-K over rows with series identity, or over a closed PromQL child, has no usable summary item. Nothing shows the frontends' DAG or the Stage 1 output.

After this PR

A choice of one per-computation candidate per target composes into a whole-workload candidate logical ASAP DAG, except for a few shapes that return Unsupported and are recorded with their error by stage_pipeline (Top-K with no label outside its partition or without, a multi-measure aggregate with without keys, a non-Float64 per-series avg, and SQL count(*) over rows with no column known to be non-null). Composition never returns an invalid DAG: every composed root passes validate_structure, otherwise the choice returns an error. A composed candidate has the queries' output column names, types and nullability, a sized summary per measure, and a theoretical guarantee on every node. stage_pipeline writes the frontends' DAG and these candidates. Stage 2 (physical ASAP-aware optimization) and Stage 3 (plan selection) consume the candidates in #576.

Evidence

• crates/integration-tests/tests/stage1_logical_asap_optimization.rs: every candidate of a set of SQL and PromQL queries composes into a valid DAG with the query's output columns (names, types, nullability), including filtered quantiles; subqueries in measure filters are substituted; SQL avg executes like the original (NULL and '' groups stay separate). Count states: sql_avg_counts_its_own_input, sql_count_star_counts_a_non_null_column, sql_count_star_without_a_non_null_column_is_unsupported, sql_count_of_a_nullable_column_counts_its_values; HAVING subqueries: own_having_subquery_is_substituted. Top-K estimates carry no frequency bound while a unit-count CMS estimate keeps its bound. Top-K over a closed PromQL child ranks its labels.
• crates/asap-aware-mapping/tests/per_computation_candidates.rs: substitution of the chosen target, per-measure sizing, filters and HAVING, avg as sum/count, a multi-measure join, and guarantees.
• crates/planner/tests/summary_sharing.rs: combined_vector_quantiles_keep_unselected_summary_candidates.
• crates/devtools/tests/stage_pipeline.rs: the Example 1 document is valid and the committed fixture is current, the display cap is recorded and keeps the selection, and an uncomposable candidate is recorded with its error. Example 1 has three candidates (one exact option per target).

Verification

• cargo fmt --all --check: OK
• cargo clippy --workspace --all-targets -- -D warnings: OK
• cargo test --workspace: 1591 passed, 0 failed, 2 ignored
• dag-viewer test_render: OK
• verify_metricsql_external_consumer.sh: OK

Architectural decisions

• Composition lives next to Pass 1 in per_computation_candidates.rs. Its helpers are named after Algorithm 1's SUBSTITUTE.
• Sizing happens during composition, per measure, as the post-Pass-2 sizing step. When Pass 2 sharing lands, sizing moves after it, so a shared summary takes the strictest requirement among its consumers.
• A measure filter is realized as SummaryAgg.filter. A key-only HAVING becomes a Filter below the state only when every consumer applies it. A value HAVING stays above, with guarantee None.
• Guarantees on composed candidates come only from the built-in model. Where it gives none, the candidate is kept and plan selection decides, never this stage.

Later changes: #577 removed the enumeration cap (--max-candidates) in favor of the display cap --max-displayed; #674 renamed the document keys and test file by stage number and name (stage0_query_language_frontends, stage1_logical_asap_optimization, stage1_logical_asap_optimization.rs) and made composition part of the planner pipeline (run_stage1_to_stage3 / Stage1ToStage3Pass, #675 names); #675 renamed the candidate label to display_name.

Limitations and follow-up

• Pass 2 (ASAP-aware CSE) sharing, then move sizing after it.
• feat(planner): physical ASAP-aware optimization and plan selection for candidate logical ASAP DAGs #576: Stage 2 (physical ASAP-aware optimization) and Stage 3 (plan selection), including the empirical output-level accuracy model.
• Count states read a column that is non-null on every counted row: avg counts its own input, SQL count(col) counts col under its IS NOT NULL filter, SQL count(*) counts a non-nullable column (preferring Float64; a non-Float64 pick composes but fails physical compile, since the runtime update accepts only Float64). With no such column (e.g. over a derived table whose columns are all nullable), SQL count(*) is Unsupported. Follow-up: count a projected literal 1.0 column under the SummaryAgg, which removes the column search. Filtered summary updates (e.g. count(v) over a nullable v) compose but have no native runtime implementation yet.
• Compose accuracy guarantees through arithmetic and comparisons over approximate values #669: composing guarantees through operators.
• Pass 1: push down filters and HAVING across levels to create more candidates (and reuse DataFusion's filter push-down) #664: push filter and HAVING predicates down through more than one level (only the Filter directly above an aggregate is handled here).

Stack: tracker #528 · main → #575 → #576 → #577 → …

Human review — do not complete with an agent

• The MVP boundary is correct.
• New conceptual layers or public interfaces are necessary.
• The before/after description matches the intended product behavior.
• Human reviewer:
• Decision and rationale: