COUNT(DISTINCT) cross-bucket combine

Part of the aggregate pushdown effort — see #6765. Depends on #6762 and #6763.

Goal

Enable COUNT(DISTINCT col) pushdown across mixed-bucket plans. After issues #6762 and #6763 land, COUNT(DISTINCT) is unsupported (the rule refuses to push it). This issue adds it.

Why this is its own issue

COUNT(DISTINCT) partial state is not a scalar. Two disjoint fragments may share distinct values, so per-fragment partials cannot be added — they have to be unioned and then counted. The combine semantics are different from COUNT/SUM/MIN/MAX, and the partial-state representation is a set (or sketch), not a number.

This is also where the bitmap index pays off most directly: for a fragment with a bitmap index on the target column, the distinct count is exactly the cardinality of { v : popcount(posting[v] AND effective_mask) > 0 }, with no column reads.

Design

Two choices for the partial-state representation, and a recommended path:

Option A — emit the distinct-values set as partial state

Each branch emits a partial state shaped like List<col_type> (the set of distinct values it observed). The final combine unions across branches and returns the cardinality of the union.

• Pros: correct under any mix of buckets. Slots into DataFusion's existing two-stage COUNT(DISTINCT) execution, which already produces and consumes set-shaped partials.
• Cons: partial-state size is proportional to distinct cardinality. For high-cardinality columns over many fragments this can be large.

Option B — restrict to all-(a)+(b)

Refuse the rewrite if any fragment falls into bucket (c). When all fragments are in (a) or (b), IndexedAggregateExec does its own internal cross-fragment dictionary union and emits a final scalar; bucket (b)'s literal is similarly pre-computed at plan time.

• Pros: simple, no new partial-state shape.
• Cons: any single non-indexed fragment kills the pushdown for the whole query.

Recommendation

Start with Option A. The set-shaped partial state matches DataFusion's existing COUNT(DISTINCT) two-stage execution, so the final combine should require no changes — branches (b) and (c) already emit the right shape from existing code. The only new work is teaching IndexedAggregateExec to emit it.

If the partial-state size becomes a concern for high-cardinality columns, a future issue can introduce a sketch-based approximate variant gated by a separate aggregate function (e.g., approx_count_distinct) rather than swapping the semantics of exact COUNT(DISTINCT).

Work

1. Add a COUNT(DISTINCT) evaluator in IndexedAggregateExec:
   • Requires a bitmap index on the target column.
   • For each value v in the column's dictionary, check popcount(posting[v] AND effective_mask AND NOT deletion_mask) > 0; collect the vs that pass into a List<col_type> array.
   • Emit one row of List<col_type> per fragment (or merged across fragments — must match AggregateExec(Partial)'s output shape for COUNT(DISTINCT)).
2. Update the optimizer rule (issue #6763) to allow COUNT(DISTINCT col) when every fragment in bucket (a) has a bitmap index on col. Fragments without the index route to (c) and use the existing scan-based partial.
3. Verify the final combine works unchanged across all three branches.

Test plan

• SELECT COUNT(DISTINCT col) FROM t over a single bitmap-indexed fragment → bucket (a), result matches unoptimized.
• Same query, multi-fragment, all indexed → bucket (a) only, result correct (distinct values that appear in multiple fragments counted once).
• Same query, multi-fragment, partial index coverage → mixed buckets (a)+(c), result correct.
• SELECT COUNT(DISTINCT col) FROM t WHERE filter with a bitmap-indexed col and an index-evaluable filter → bucket (a), result correct.
• NULL handling: column with nulls — verify nulls are excluded from the distinct count (matching SQL semantics).
• Deletion vector: distinct value that appears only in deleted rows is not counted.
• High-cardinality fragment (~100K distinct values) — sanity check on partial-state encoding/decoding size and correctness.
• Mixed aggregates: SELECT COUNT(*), COUNT(DISTINCT col) FROM t WHERE x = 5 where both can be pushed — single IndexedAggregateExec emits both partial states, final combine produces both scalars.

Acceptance criteria

• IndexedAggregateExec supports COUNT(DISTINCT col) for bitmap-indexed col.
• Optimizer rule routes COUNT(DISTINCT) correctly across mixed buckets.
• All tests above pass.
• Final combine is unchanged from upstream DataFusion (the value of going through DataFusion's existing two-stage COUNT(DISTINCT) shape).

Non-goals

• Approximate distinct count (approx_count_distinct, HLL). Separate future issue.
• COUNT(DISTINCT) over a non-bitmap-indexed column via index. Falls to (c) — covered by existing scan path.
• COUNT(DISTINCT a, b) (multi-column distinct). Separate future issue.

Read IndexedAggregateExec and the optimizer rule from issue #6763 first, then compare their partial output with DataFusion's existing AggregateExec(Partial) COUNT(DISTINCT) shape. Run the listed single- and multi-fragment, mixed-bucket, filter, NULL, deletion-vector, high-cardinality, and mixed-aggregate cases; done means correct exact counts with the final combine unchanged.