Aggregate pushdown optimizer rule

Part of the aggregate pushdown effort — see lance-format/lance#6765. Depends on lance-format/lance#6762.

Goal

Add a physical optimizer rule that recognizes scalar-aggregate query shapes over Lance scans and rewrites them into the three-branch plan from the parent issue. The rule decides bucket membership for each fragment using only manifest metadata — no index I/O at plan time.

Pattern matched

AggregateExec(Final, group=[])
  └─ [single-child layers]
       └─ AggregateExec(Partial, group=[], filter=None)
            └─ [optional FilterExec]
                 └─ LanceScanExec

The rule rejects (does nothing) if:

• The aggregate has GROUP BY (covered by a future issue).
• The partial aggregate has its own FILTER (WHERE …) per-aggregate filter.
• The scan node is not a Lance scan.
• The filter (if present) is not fully decomposable against indexed/statted columns (see classification below).
• Any aggregate in the list is not supported by IndexedAggregateExec or by literal-from-stats.

The rule must run before DataFusion's built-in AggregateStatistics rule, since this rule subsumes the unfiltered case and AggregateStatistics would otherwise fire first and break the pattern.

Bucket classification

For each fragment of the Lance scan, classify it into (a), (b), or (c) using only manifest metadata:

for fragment in scan.fragments():
    let stats_verdict = evaluate_filter_against_stats(filter, fragment.column_stats);
        // returns: Fully_matching | Not_matching | Unknown

    let index_supports_filter = filter.columns().all(|c|
        fragment.has_compatible_index(c, filter.predicate_on(c))
    );
    let index_supports_aggs = aggregates.iter().all(|a|
        a.is_count_star() || fragment.has_compatible_index_for(a)
    );

    match (stats_verdict, index_supports_filter && index_supports_aggs) {
        (Fully_matching | Not_matching, _) => bucket B,    // stats are decisive
        (Unknown, true)                    => bucket A,    // index can do it
        (Unknown, false)                   => bucket C,    // fall back to scan
    }

evaluate_filter_against_stats is the same kind of analysis DataFusion's PruningPredicate does for row-group pruning. Reuse it if practical.

Plan rewrite

For each non-empty bucket, build a branch:

• (a) → an IndexedAggregateExec configured with the bucket's fragments, the filter, and the aggregate list. Per-fragment IndexBindings are resolved here from the manifest.
• (b) → a ProjectionExec of literals over PlaceholderRowExec. Literal values are computed at plan time:
  • COUNT(*): sum of num_rows over Fully_matching fragments (plus zero from Not_matching).
  • COUNT(col): sum of (num_rows − null_count[col]) over Fully_matching, capped at column-level resolution.
  • MIN(col)/MAX(col): min/max of min_value[col] / max_value[col] over Fully_matching fragments. (Out of initial scope per issue lance-format/lance#6762; the rule must not route MIN/MAX into (a) until a follow-up adds support.)
• (c) → a scan-and-partial-aggregate over the remaining fragments. Reuse existing partial-agg infrastructure; just restrict the scan's fragment list.

Combine with UnionExec, place the existing AggregateExec(Final) on top.

Edge cases — produce the smallest valid plan, do not emit empty branches:

• Only one bucket non-empty → no Union, just that branch + Final.
• (a) and (b) both empty → don't rewrite. Original plan stands.
• Bucket (b) needs no children but bucket (b) doesn't exist when all fragments are Unknown → that's fine, just don't emit it.

Constraints

• No index I/O at plan time. The rule reads only the manifest. Verified by a test that fails if the rule opens any index file.
• Partial-state schema invariant. All branches emit the same per-aggregate state shape; assert this at rule construction time.
• Deletion vector composition is the operator's responsibility, but the rule must record deletion_vector in each FragmentPushdownPlan so the operator can apply it. For bucket (b), if a Fully_matching fragment has a deletion vector that is not accounted for by the num_rows stat, the rule must downgrade that fragment to (c). Need to confirm whether num_rows is the live count or the original count.
• Filter splitting. If the filter is p1 AND p2 where p1 is index-evaluable and p2 is not, the current scope does not split — the whole fragment falls to (c). Splitting is a follow-up.

Initial scope

To match IndexedAggregateExec's initial scope (issue lance-format/lance#6762):

• COUNT(*) and COUNT(col), no DISTINCT.
• Bitmap index for filter evaluation; any other index kind for filter evaluation routes that fragment to (c).
• Single base relation.

COUNT(DISTINCT), MIN/MAX, SUM, multi-aggregate-mixed-support queries, and grouped aggregates are out of scope. They land in follow-up issues.

Test plan

• SELECT COUNT(*) FROM t with a single fragment, no filter → bucket (b), Union not emitted, plan is Final ← Projection(lit(N)).
• SELECT COUNT(*) FROM t WHERE x = 5 against a bitmap-indexed column, single fragment → bucket (a), plan is Final ← IndexedAggregateExec.
• Same query, multi-fragment, where some fragments lack the index → mixed buckets (a)+(c), plan has both branches under a Union, final result equals the unoptimized result.
• Query with a filter that doesn't match an index and stats are inconclusive → no rewrite, original plan returns.
• Query with GROUP BY → no rewrite.
• Query with COUNT(DISTINCT) → no rewrite (issue INSERT_LINK_TO_3 enables this).
• Query with MIN/MAX or SUM → no rewrite (future issues enable this).
• Plan-time I/O test: instrument the index reader and assert zero reads during physical planning.
• Deletion vector regression: fragment with deletes is classified into (c) (or (a) if the operator handles it) — never into (b) silently.
• Schema invariant: rule construction panics or returns an error if branches would produce mismatched partial-state schemas.
• Result-equality: for each rewrite case, the optimized plan's result equals the unoptimized plan's result on the same data.

Acceptance criteria

• New rule registered in Lance's physical optimizer pipeline before AggregateStatistics.
• All tests above pass.
• Result-equality tests cover at least: single fragment (b)-only, single fragment (a)-only, multi-fragment (a)+(c), multi-fragment (b)+(c).
• No measurable regression in plan time on a benchmark query with no aggregates (the rule should noop quickly for non-matching plans).

Non-goals

• COUNT(DISTINCT) cross-bucket combine (issue INSERT_LINK_TO_3).
• Filter splitting (future).
• Additional index types (future).
• Grouped aggregate pushdown (future).

Start with the physical optimizer pipeline and the existing IndexedAggregateExec, AggregateStatistics, FragmentPushdownPlan, and partial-aggregate infrastructure; trace how Lance scan fragments, manifest metadata, indexes, and deletion vectors are represented. Done means the new rule is registered before AggregateStatistics, handles the listed bucket and no-rewrite cases, performs no plan-time index I/O, and passes the specified result-equality, schema, deletion-vector, and planning tests.