IndexedAggregateExec: physical operator for index-driven aggregates

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

Goal

Introduce a new physical ExecutionPlan node that computes partial aggregate state for a set of fragments by probing their indices at execute time. The operator is the execute-time engine for bucket (a) of the optimizer rule (next issue) — it answers questions like "given this filter and this aggregate, here is the partial state for these fragments" without scanning column data.

This issue covers the operator only. The optimizer rule that produces this node is a separate issue and is not required to land this one — the operator is testable by direct construction.

Interface

pub struct IndexedAggregateExec {
    filter: Option<Arc<dyn PhysicalExpr>>,
    aggregates: Vec<Arc<AggregateFunctionExpr>>,
    fragments: Vec<FragmentPushdownPlan>,
    /// Partial-state schema (must match what AggregateExec(Partial) would produce
    /// for `aggregates`, so a downstream AggregateExec(Final) can consume it).
    schema: SchemaRef,
    /// Standard DataFusion exec plumbing.
    properties: PlanProperties,
}

pub struct FragmentPushdownPlan {
    pub fragment_id: u64,
    /// Pre-resolved by the optimizer rule from manifest metadata.
    pub index_bindings: Vec<IndexBinding>,
    pub deletion_vector: Option<DeletionVectorHandle>,
}

pub struct IndexBinding {
    pub column: String,
    pub index_id: Uuid,
    pub kind: IndexKind,
}

execute(partition, ctx) returns one batch per fragment (or one combined batch — implementation choice, doesn't affect correctness since a final combine sits above it).

Per-fragment evaluation

For each fragment:

1. Open the indices named in index_bindings. (This is execute-time I/O — fine.)
2. Evaluate filter against the indices to produce a filter_mask: RoaringBitmap of matching rows.
3. Compose the deletion vector: effective_mask = filter_mask AND NOT deletion_mask.
4. For each aggregate, compute its partial state from effective_mask and any auxiliary index state (e.g., per-value posting lists for the aggregate's column).
5. Emit a batch row whose schema matches AggregateExec(Partial)'s output for the same aggregates.

The aggregate-to-index dispatch should live behind a trait so additional index/aggregate combinations can be added incrementally:

trait IndexAggregateEvaluator {
    fn supports(agg: &AggregateFunctionExpr, kind: IndexKind) -> bool;
    fn evaluate(
        &self,
        agg: &AggregateFunctionExpr,
        effective_mask: &RoaringBitmap,
        index: &dyn IndexHandle,
    ) -> Result<ScalarValue>;
}

Initial scope

Land the operator with support for exactly these combinations. Everything else returns an error from the evaluator and is the optimizer rule's job to avoid.

Aggregate	Index	Notes
COUNT(*)	any	popcount(effective_mask). The "index" only needs to evaluate filter.
COUNT(col)	bitmap on col	popcount(effective_mask AND NOT null_posting[col]).

COUNT(DISTINCT), MIN, MAX, SUM, and grouped aggregates are explicitly out of scope here. They will land in follow-up issues — COUNT(DISTINCT) is the next one (issue INSERT_LINK_TO_3).

Output schema

For each aggregate in aggregates, the output schema column is whatever AggregateFunctionExpr::state_fields() reports. For COUNT that's a single Int64. The operator must not invent its own schema — it has to match what the partial-aggregate path produces, so the final combine works unchanged.

Test plan

Direct construction tests in the same crate (no optimizer rule needed):

• COUNT(*) with no filter — verifies effective_mask defaults to all rows, deletion vector composed correctly.
• COUNT(*) WHERE col = v against a bitmap-indexed column — verifies posting list lookup and popcount.
• COUNT(*) over a fragment with a deletion vector covering some matching rows — verifies the AND NOT deletion_mask composition.
• COUNT(col) where col has nulls — verifies null-posting subtraction.
• Multi-fragment input — verifies one row of partials per fragment (or a single merged row, depending on implementation), and that a downstream AggregateExec(Final) produces the expected scalar.
• Output schema equals AggregateExec(Partial)'s schema for the same aggregate list — assert via direct schema comparison.
• Unsupported (aggregate, index) combination → operator construction or first execute() returns a clear error.

Acceptance criteria

• New IndexedAggregateExec node in lance-datafusion (or wherever Lance's physical operator extensions live).
• Tests above all pass.
• No optimizer rule changes — the operator is unreachable in normal queries until issue INSERT_LINK_TO_2 lands.
• Memory: uses RoaringBitmap, not HashSet<u32>, for row sets (per AGENTS.md).

Non-goals

• Plan-time fragment classification (issue INSERT_LINK_TO_2).
• COUNT(DISTINCT) (issue INSERT_LINK_TO_3).
• Performance benchmarks (separate follow-up; needs realistic datasets).
• Wiring up additional index kinds beyond bitmap (separate follow-ups per kind).

Start in the lance-datafusion physical operator extensions and inspect DataFusion's AggregateExec partial and final schemas, index APIs, RoaringBitmap usage, and AGENTS.md. Add direct-construction tests for the listed COUNT cases, deletion vectors, null postings, multiple fragments, schema equality, and unsupported combinations; done means all tests pass without optimizer changes.