executor: filtered aggregates and summary builds

[zzylol]

Stack: Wave 3 chain: #605 → #607 → #608 → #609 → #610

Rebased on main d4869a7 (DF 54).

Stack: … → #605 → this PR → (Pass 1 filtered aggregates, next)

Why

Part of #509 and #580 (J, filtered aggregates); Q47 (a). The executor could not run any filtered aggregate. A per-measure filter on Aggregate and a SummaryAgg.filter were both rejected at binding, so even the exact plan for COUNT(*) FILTER (WHERE …) failed. This also covered COUNT(y) over a nullable y, which the SQL frontend lowers as a count filtered by y IS NOT NULL. #605 (item "Not in this PR") left filtered counts to this work.

What

events(g, x) with rows a: 1, -1, 2; b: -3, 0; c: 5:

query (exact, through plan_stages + executor)	Before this PR	After this PR
SELECT g, COUNT(*) FILTER (WHERE x > 0) … GROUP BY g	filtered aggregate has no native implementation	a 2, b 0, c 1
SELECT g, SUM(x) FILTER (WHERE x > 0) … GROUP BY g	same error	a 3.0, b NULL, c 5.0
SELECT g, COUNT(y) … GROUP BY g (nullable y)	same error	non-NULL y per group, b 0
a SummaryAgg with filter (exact Count, KLL)	filtered summary update has no native implementation	builds from the matching rows only, keeping every group

Empty groups follow SQL. A group is formed from all of its rows, and the filter only decides which rows update the measure. A group with no matching row therefore still appears. COUNT (and distinct count) reads 0. SUM, AVG, MIN, MAX and a quantile read NULL. HAVING is unaffected: the SQL frontend lowers it to a Filter above the aggregate. PromQL never emits per-measure filters, since label matchers are scan predicates.

How

• Operators: Aggregate gains per-measure filters (with_measure_filters). A filtered SUM/AVG/MIN/MAX output becomes nullable. The three summary builds (per-group, keyed, shared Hydra) gain an optional row filter (with_row_filter). A row passes only when the predicate is true; NULL drops it, as in SQL. Both fields are omitted from serialization when absent, so existing plans are unchanged. Deserialization rebuilds and checks them.
• Binding: physical_planner compiles the predicates against the input schema. The unfiltered SummaryAgg lowering moved, unchanged, into summary_build so that the filter is applied in one place. Ingestion-time precompute still rejects a filtered SummaryAgg, because its raw-sample input has no SQL columns to evaluate the filter against.
• Empty sketches: AggregateCore::is_empty (KLL, DDSketch). A sketch evaluation with a nullable result reads NULL for an empty state. An empty KLL quantile is now an error, not 0.0, as DDSketch's already was.
• Types: a filtered Quantile measure is declared nullable, matching SUM.

Tests

These failed on #605 with the errors above:

• filtered_aggregates::exact_filtered_count_keeps_groups_without_matches, exact_filtered_sum_is_null_for_groups_without_matches, exact_count_of_nullable_column_skips_nulls: SQL → plan_stages → executor, with a group that has no matching row.
• filtered_aggregates::hand_filtered_exact_count_compiles_and_executes, hand_filtered_kll_compiles_and_executes: Pass 1's unfiltered Count / KLL alternatives with SummaryAgg.filter set by hand, compiled and executed.

New executor tests:

• physical_semantics::measure_filters_keep_groups_without_matching_rows: filtered and unfiltered measures side by side, a NULL predicate, nullability, and a serde round trip.
• physical_semantics::filtered_summary_build_keeps_empty_groups: a filtered KLL build, where the empty group reads NULL, plus a serde round trip.

Gates

• cargo fmt --all --check, cargo clippy --workspace --all-targets --all-features -- -D warnings: pass.
• cargo test --workspace: 1,649 passed, 23 ignored (feat(planner): Pass 1 offers HydraCms for grouped approximate counts #605: 1,642 / 23; +7).
• Viewer tests (tools/dag-viewer): 29 ran, 6 skipped.

Not in this PR

• Pass 1 still rejects filtered aggregates. The next PR enables single-measure ones.
• Filtered precompute (ingestion-time) builds.

Refs #509, #580, #605.

🤖 Generated with Claude Code