[zzylol]

Why

A query is often computable from a finer query the workload already computes. Example: SUM(value) … WHERE city = 'NYC' is computable from SUM(value) … GROUP BY city, host. Today the planner offers this reuse in only two fixed shapes:

• filtered_group.rs handles one global quantile, one finer key and one extra key = literal.
• the rollup rule needs both builds over the same rows.

Every other filtered or coarser variant builds its own scan and aggregate. #950 designs one rule for all of them, following view matching for aggregation queries (SIGMOD 2001 §3.3; Calcite SubstitutionVisitor.unifyAggregates / SqlAggFunction.getRollup).

What

A query aggregate gets a candidate that reads a finer aggregate through a compensating filter (the query's extra conjuncts on finer keys) and, when its groups are coarser, a rollup:

• Value path (SQL aggregates, any implementation of the finer group). The finer result passes through Filter, then Aggregate:
  • When every dropped key is pinned by an equality (regroup not needed), it reads with MIN per group. This works for any measure, including quantiles, and keeps SQL's one NULL row for an absent group.
  • Otherwise it rolls the values up: SUM of sums or counts (counts only into grouped queries), MIN, MAX.
• State path (sketch builds). The finer states pass through Filter, then SummaryMerge { group_by }, then the query's estimate. For example, a p90 over city IN ('NYC','SF') is computed by merging the NYC and SF KLL states of a per-city build.
• IR. A Filter over a state whose predicate reads only plain (key) columns now stays State.

filtered_group.rs is deleted; its case is the pinned-key case of the value path.

How

• match_finer (finer_aggregate.rs) normalizes both aggregates' input rows into a spine (time ranges + scan without predicates) and a conjunct set. It requires:

  • equal spines;
  • finer conjuncts ⊆ query conjuncts;
  • every remaining conjunct reads only finer group-by columns (so filtering finer groups equals filtering rows);
  • query keys ⊆ finer keys.

  It returns the compensating filter over the finer keys, the query keys' positions, and whether a rollup is needed (some dropped key is not pinned).

• Value path: the memo rule memo/rules/finer_aggregate.rs buckets SQL aggregate groups by their measures and builds the reader for each matching pair. The reader is a candidate only when its schema equals the query's.

• State path: in asap_aware_cse.rs, filtered_finer uses the same match_finer when the builds' rows differ. The sketch rollup then regroups over Filter(states), and resizing for all readers keeps the filter.

• Guarantees:

  • selected_group_guarantee now covers any unique key whose columns are group keys or pinned, not just one key.
  • rolled_up_guarantee reads through the state selection.
  • Summary coverage maps a state selection to the same build over filtered input rows.

Deviations from the #950 design.

• add_rollup_candidates stays, so a same-rows SQL rollup can come from two rules: the value path here and the exact-accumulator rollup. Plan selection prices both.
• The value rule buckets groups by their measures only, not by source and window (the filter tree's first level only); match_finer then checks source and window.
• The compensating filter accepts any expression that reads only finer key columns (including OR), not only column-literal comparisons; such an expression is constant within a finer group.
• A same-level read applies only to aggregates with one row per group: a top-k or count-values measure is never read from a finer aggregate.

Execution of the state path.

• A node whose output is state (a key filter, a projection of the state column) carries summary state in the execution-data-state checks (timing.rs), and a key filter passes states of any family.
• The executor compiles a state filter's predicate over the plain columns only (CompiledExpression::compile_over_states); other operators still refuse state inputs.
• A global rollup of selected states keeps SQL's one row over an empty selection (built_from_relation looks through the filter).
• The executor evaluates SQL x [NOT] IN (…). Before, every IN was an "unsupported scalar expression", in a Filter or a Scan predicate alike, so each state-path plan for example C's city IN ('NYC','SF') failed to compile.
• IN is vectorized: the batch path lowers x [NOT] IN (v1, …) to [NOT] (x = v1 OR …), which has the same three-valued result, so a filter with IN runs as Arrow kernels. A NULL literal item keeps it on the row path, as it does for =. An empty list still validates its operand.
• An expression the vector path cannot lower is evaluated per input with NULL in the columns it does not read. A state filter accepts NULL there as well as a state. Only state columns get NULL; a filter over states can never read one, so this cannot hide a schema mismatch.
• PromQL =~ is not affected: it lowers to CompareOpKind::Regex, not IN.

Key code interfaces added / modified

crates/logical-asap-optimization/src/finer_aggregate.rs (new, crate-private)

/// How a query aggregate is computed from a finer one: keep the finer rows that pass
/// `compensating_filter`, then, when `needs_rollup`, merge them into the query's groups.
pub(crate) struct FinerAggregateMatch {                     // new
    /// The query's conjuncts the finer aggregate lacks, over the finer group-by (column i = i-th finer key).
    pub compensating_filter: Option<Predicate>,              // new
    /// Each query group key, as an index into the finer group-by.
    pub query_keys: Vec<usize>,                              // new
    /// A dropped finer key is not pinned by an equality, so finer groups merge.
    pub needs_rollup: bool,                                  // new
}
/// Whether the query aggregate (rows, keys) is computable from the finer one, and how.
pub(crate) fn match_finer(query_rows: &OperatorNode, query_keys: &[usize],
    finer_rows: &OperatorNode, finer_keys: &[usize]) -> Option<FinerAggregateMatch>; // new
/// The columns `pred` pins to one literal each (`c = v`, `c IN (v)`).
pub(crate) fn pinned_columns(pred: &ScalarExpr) -> Vec<usize>; // new

crates/logical-asap-optimization/src/memo/rules/finer_aggregate.rs (new; replaces memo/rules/filtered_group.rs)

/// Adds to each SQL aggregate group a read of every finer aggregate group that computes it.
pub(in crate::memo) fn add_finer_aggregate_candidates(memo: &mut CandidateMemo,
    deadline: Option<Instant>) -> Result<usize, StateGroupExpansionError>; // new; was add_filtered_group_candidates (removed)

crates/logical-asap-optimization/src/asap_aware_cse.rs

/// Finer key count and compensating filter over the finer states when `fine` answers `coarse` from more rows.
fn filtered_finer(coarse: &OperatorNode, fine: &OperatorNode) -> Option<(usize, Predicate)>; // new

crates/types/src/ir/operator/asap.rs · non_asap.rs · summary_coverage.rs

/// The states a `Filter` over state selects from; `None` for any other node.
pub fn selected_from(node: &OperatorNode) -> Option<&Rc<OperatorNode>>; // new
/// Whether `Filter(pred)` over `child` keeps whole states (child is State, pred reads only plain columns).
fn selects_states(child: &OperatorNode, pred: &Predicate) -> bool;      // new (private)
/// The build that computes exactly the states such a filter keeps.
fn selected_build(node: &OperatorNode) -> Result<Option<Rc<OperatorNode>>, CoverageError>; // new (private)

NonASAPOp::output_kind: Filter over State reading only plain columns → State (was Relation); validate_inputs accepts it.

crates/executor/src/expressions/planner.rs

impl CompiledExpression {
    /// `expression` compiled over `input` whose other columns may hold summary state:
    /// a filter keeping whole states by their keys. Bad input when it reads a state column.
    pub fn compile_over_states(expression: &PlannedExpr, input: &SchemaRef) -> Result<Self, Error>; // new
}

Usage (example B, end to end):

// The user's workload: two SQL queries every 10 s over the last 5 minutes.
let queries = [
    // The finer query: per (city, host) sums.
    "SELECT city, host, SUM(value) FROM metrics WHERE ts > now() - INTERVAL '5' MINUTE GROUP BY city, host",
    // The filtered query: the NYC sum.
    "SELECT SUM(value) FROM metrics WHERE ts > now() - INTERVAL '5' MINUTE AND city = 'NYC'",
];
// Stage 1 memo expansion (expand_concrete_summary_state_groups) calls
// add_finer_aggregate_candidates. match_finer(NYC rows, [], all rows, [city, host]) returns
// compensating_filter = Some(key0 = 'NYC'), query_keys = [], needs_rollup = true.
// The filtered query's aggregate group gains a Rollup candidate:
//   Aggregate(by [], SUM(col 2)) ▸ Filter(col 0 = 'NYC') ▸ <finer group's result>
// Plan selection (Stage 3) picks it: the whole workload costs 0.0272527/s versus
// 0.0272326/s for the finer query alone. The filtered query alone costs 0.0054229/s.

(From finer_aggregate_reuse::filtered_sum_costs_little_beyond_the_finer_sums.)

Major algorithm changes

Finer-aggregate reuse (Pass 2 rollup)

• Problem. The workload above planned the filtered sum with its own scan and aggregate. filtered_group only took one quantile with one key. A KLL p90 over city IN ('NYC','SF') could not use a per-city KLL.
• Before.
  1. add_filtered_group_candidates handled one global quantile, one finer key and one extra key = literal scan conjunct.
  2. add_rollup_candidates paired builds over identical rows.
• After.
  1. match_finer runs per pair of SQL aggregates with equal measures, which are bucketed.
  2. The value reader is Filter then Aggregate with MIN (when the query's key set or pinned keys cover the finer unique key) or the per-measure rollup.
  3. In the sketch rollup, when no same-rows finer build exists, the finer build with the most keys that filtered_finer accepts is merged through Filter(states).
• Correctness.
  • A conjunct that reads only finer group-by columns is constant within a finer group, so filtering groups equals filtering rows.
  • MIN over at most one row is the identity; for a global query over no rows it gives SQL's one NULL row.
  • A global COUNT is never read from a sum (0 vs NULL).
  • Merging selected disjoint group states is the family's group merge.
  • Two aggregates with the same key set and no compensation are one computation, not a read of each other (a read both ways would be a memo cycle).
  • Examples A, B, the grouped rollup and C execute the original and the reuse candidate on the same rows (empty input, absent group, NULL samples) and compare. For C the original is the query's own KLL read, so both sides are the same family.
• Complexity. The value rule is O(Σ bucket²) over SQL aggregate groups (was O(G²) for filtered_group). The state path adds one match_finer per (coarse, build) pair that is not the same rows.
• Assumptions and limits.
  • Measures must be equal.
  • No AVG, no HAVING, no per-measure filters.
  • Conjuncts may read only finer keys and no subqueries.
  • The window dimension is not covered.
  • Sketch builds must have no build filter.

Before this PR

• Planning. SUM(value) … WHERE city = 'NYC' next to SUM(value) … GROUP BY city, host built its own scan and aggregate: 0.0054229/s on its own.
• Not reused at all:
  • a global quantile pinned by two keys;
  • a grouped query keeping several groups;
  • a sketch over city IN (…).
• IR. A Filter over states was rejected.

After this PR

• Planning. The filtered sum reads the finer sums. Together the two queries cost 0.0272527/s, versus 0.0272326/s for the finer query alone.
• Newly reused:
  • Example A: two pinned keys read one finer quantile row.
  • Grouped filtered rollup: SUM and MAX by city over city = 'NYC' OR city = 'SF'.
  • Example C: the filtered KLL states merge. Planned jointly, example C's optimum drops from 0.2284651/s (main) to 0.1963020/s: q1 keeps the NYC and SF states of a KLL by city built over q0's query-time rows. The search proves that optimum only with a longer seed round; see "Example C and the joint search" below.
• IR. A Filter that reads only key columns keeps summary states as states.

Verification

Failing first. On origin/main with only the new test file, cargo test -p asap-integration-tests --test finer_aggregate_reuse:

• fail: pinned_keys_read_one_finer_quantile_row, filtered_sum_rolls_up_finer_sums, filtered_grouped_aggregates_roll_up_finer_values, filtered_kll_states_merge, filtered_sum_costs_little_beyond_the_finer_sums;
• pass, as expected for negatives: the three negative tests.

On this branch:

Suite	Result
finer_aggregate_reuse	10 passed, incl. the example C joint cost (0.2284651/s together, no more than the separate plans); filtered_kll_states_merge also checks the IN read against the OR form, and filtered_grouped_aggregates_roll_up_finer_values runs IN on the value path
asap-types --lib	211 passed
asap-types --test summary_coverage	33 passed, incl. new a_key_filter_over_states_selects_states
asap-types --test pane_chain	14 passed
asap-logical-asap-optimization --lib	229 passed (rebased on main 6a22ec49)
integration sharing_issue_regressions	12 passed
integration sketch_rollup	3 passed
integration sketch_rollup_planning	3 passed
integration rollup_controlled	2 passed
integration production_rollup_workloads	4 passed
integration reuse_rules	7 passed
integration identical_exact_queries	2 passed
integration per_computation_candidates_sql	9 passed
integration filtered_aggregates	7 passed
asap-executor (lib and tests)	lib 164 passed (1 ignored, as on main), incl. new in_list_follows_sql_three_valued_logic and an_in_list_is_vectorized_with_row_values (batch result equals the row result for IN/NOT IN, NULL items and an empty list); all test targets pass
integration joint_search_equivalence	14 passed
integration kll_pane_execution / sql_to_physical / materialization_e2e	2 / 1 / 4 passed
cargo fmt --all -- --check	clean

IN support, failing first. On the branch rebased on main, before the executor change:

• filtered_kll_states_merge with the query as written (city IN ('NYC','SF'), previously rewritten to OR because the executor had no IN) fails to compile its filter.
• With IN evaluated but the per-input path still passing NULL into the state check, it fails with expression input differs from its bound schema.
• Both pass at this head.

Review fixes, failing first (each fails at 7faad60, passes at this head):

• permuted_group_by_is_not_a_finer_read: CyclicDependencies before.
• filtered_kll_states_merge now executes the merged-state read: IllegalChildDataState { edge: "SummaryMerge.children" … } before; then the Filter edge and the executor rejected state input.
• The state-path tie-break, the one-row-measure check and the single replacement per build have no separate failing test; the code is covered by the suites above.
• The range-filter case that must lose a single-group bound is filtered_group_preserves_sql_empty_and_null_semantics (Gt on the key).

Test runs for this head were on one 56-core host; main and branch cost comparisons ran there too.

The #923 cost test is timing-sensitive.

• When it ran concurrently with another planner build, exact_coarse_sum_reuses_fine_sum failed its 0.00004/s margin on both origin/main (0.0566/s) and this branch.
• Run alone, both give 0.0066551/s (base) and 0.0066901/s (related), and the test passes. The whole file passes on both main and this head (12/12) on an idle host.

Guarantee check. sharing_issue_regressions::filtered_group_preserves_sql_empty_and_null_semantics gains a case: a filter that also reads the estimate (IsNotNull(col 1)) must not keep the selected group's rank bound.

• Fails without the key-only check in selected_group_guarantee.
• Passes with it.

Workspace clippy and full-workspace tests were not run locally.

Example C and the joint search

Workload: SELECT city, p90 … GROUP BY city with SELECT p90 … WHERE city IN ('NYC','SF'), both every 10 s over 5 minutes.

Run (same host)	Grouped alone	Filtered alone	Together (default 30 s budget)	Together, seed round uncapped
origin/main	0.1760462/s	0.0524189/s	0.2284651/s	0.2284651/s (Optimal after 18 evaluations, 0.9 s)
this branch, before the plan-selection fix (#973)	0.1760462/s	0.0524189/s	0.3320576/s	0.1963020/s (Optimal after 152 evaluations, 228 s; first found at evaluation 16, 9.2 s)
this branch on main with #973	0.1760462/s	0.0524189/s	0.2284651/s (seed-joint-plan-8)	—

The branch's optimum: q0 exact at query time (0.1760); q1 reads q0's query-time rows (shared time range), builds a KLL by city, keeps NYC and SF, merges and estimates.

Cause, from instrumented runs of the seed round:

• Lower bounds. lower_bounds.rs prices query-time Scan, TimeRange, Aggregate and KLL builds at 0, but an ingestion Scan at 0.0038686/s. Every plan that runs q1 at query time therefore has bound 0, and the MILP proposes those first. The bound gives the proposals no useful order.
• Rejections. The MILP has no accuracy or compile constraints, and each rejection cuts one query closure. q0 reading any KLL state fails its whole-result target: no distinct_values_per_label is declared, and that evidence exists only in Stage 3. q0's own Pass 1 KLL candidates fail the same way on main. Before the IN fix, 40 further proposals failed to compile.
• Size. The state path enlarges the seed round's MILP from 58 candidates / 191 variables / 290 constraints to 120 / 1547 / 3807, and each solve from about 32 ms to 0.4–1.4 s. The seed round is capped at min(total/4, 5 s) and 32 evaluations (route.rs), so it ends at 0.3321/s. Later pane rounds return the same. A larger total budget does not help.

None of the new candidates can be dropped in Stage 1: the q0 KLL readers depend on Stage 3 evidence, and the q1 state reads are the optimum. The fix belongs to plan selection. #973 (merged) evaluates the queries' separate plans combined when the joint search ends above them, so the default run now returns 0.2284651/s, no worse than main, and the example C cost test (filtered_kll_states_cost_no_more_than_separate_plans) runs un-ignored. Reaching 0.1963/s within the budget needs the candidate-level rejection cuts of #964 (measured 0.1963 with them) and the lower bounds of #963.

Closes #950

🤖 Generated with Claude Code