Tracking: Aggregate pushdown

Motivation

Lance currently supports filter pushdown: filters can be evaluated by scalar/bitmap/inverted indices before a column scan, so the scan only materializes matching rows. We do not have an analogous capability for aggregates. A query like:

SELECT COUNT(*)            FROM t WHERE x = 5;
SELECT COUNT(DISTINCT col) FROM t;
SELECT MIN(y)              FROM t WHERE x BETWEEN 10 AND 20;

today scans the column(s) and runs a standard partial/final aggregate, even when the answer is fully derivable from index state or per-fragment statistics. The opportunity is significant — Lance's bitmap index stores per-value posting lists, which can answer exact COUNT(DISTINCT) and filtered COUNT(*) without touching column data; per-fragment stats can answer MIN/MAX/COUNT(*) on fragments where the filter is decisive.

This issue tracks the cross-cutting work to introduce aggregate pushdown. It does not deliver code on its own — see sub-issues for individual work items.

Design overview

We model aggregate pushdown as a physical optimizer rule that splits a query's fragments into three buckets and rewrites the plan accordingly:

        AggregateExec(Final)
              │
         UnionExec
        ┌─────┼─────────────────────────────────────────────┐
        │     │                                             │
 IndexedAggregateExec    ProjectionExec(literals)    AggregateExec(Partial)
   (filter, aggs,             over                          │
    fragments_a)         PlaceholderRowExec          FilterExec / LanceScan
                                                     (fragments_c)
        (a)                     (b)                          (c)

• (a) Index-answerable fragments. Filter touches only indexed columns and the aggregate is computable from index state. Probe happens at execute time inside IndexedAggregateExec, not at plan time.
• (b) Stats-answerable fragments. Filter is decisive against per-fragment column stats (Fully Matching → contribute row count / min / max / null count; Not Matching → contribute zeros). Literal projection is computed at plan time from manifest stats.
• (c) Residual fragments. Anything that falls through gets scanned and aggregated normally.

The final AggregateExec combines partials from all three branches. All three branches must produce partials in the same state schema (e.g., Int64 for COUNT, (sum, count) for AVG).

Why three branches, not one

Bailing the whole query when a single fragment can't be pushed (Iceberg's choice in apache/iceberg#6622) leaves significant performance on the table. Lance fragments are independently indexed; partial pushdown per fragment is the natural fit.

Why defer index access to execute time

partition_statistics is called during physical planning. If the rule itself opened indices to compute exact counts, that cost would be paid for every query that touches a Lance table — including queries with no aggregates that just need a cardinality estimate for join planning. We therefore distinguish:

• Plan-time information: manifest metadata only (index existence per fragment, per-fragment column stats, deletion vector presence). No index I/O.
• Execute-time information: actual index probes, posting list intersections, popcounts.

The optimizer rule baked plan classifies fragments at plan time using only manifest metadata; index probes happen inside IndexedAggregateExec::execute.

Constraints

1. Filter must be fully evaluable by the index for any fragment going into bucket (a). A partial filter evaluation produces a candidate set, not a final one; that breaks aggregate correctness.
2. Deletion vectors must compose into every index-derived count: popcount(posting AND filter_mask AND NOT deletion_mask). The rule must not push down for a fragment whose deletion vector cannot be composed in.
3. Partial-state schema invariant: branches (a), (b), (c) all emit the same state schema for a given aggregate, because they feed a shared final combine.
4. No regression in plan time: classification logic must use only manifest metadata, no index reads.

Phases

• lance-format/lance#6762 — IndexedAggregateExec operator (the execute-time engine for bucket (a))
• lance-format/lance#6763 — Physical optimizer rule (bucket classification, plan rewrite)
• lance-format/lance#6764 — COUNT(DISTINCT) cross-bucket combine (set-shaped partial states)

Follow-up work — not scoped here — will add support for additional index types (btree → MIN/MAX, inverted index → text aggregates), GROUP BY pushdown, SUM (requires extending column stats), and additional aggregates.

Out of scope (for this tracking issue)

• GROUP BY pushdown. Different plan shape; needs its own design.
• SUM/AVG from stats — requires extending ColumnStatistics with a sum field, which is upstream-affecting.
• Cross-index queries (e.g., filter on column A using bitmap, aggregate on column B using inverted index, on the same fragment).

Background research

A taxonomy of how other engines handle this lives in this gist (Postgres MinMaxAggPath, DataFusion AggregateStatistics, Iceberg PR apache/iceberg#6622, SQL Server columnstore aggregate pushdown, Druid bitmap groupby, Snowflake micro-partition metadata). Key takeaways that shaped the design above:

• Postgres rewrites MIN/MAX queries into ordered index-scans with LIMIT 1 — different mode from what Lance needs but worth noting for future btree work.
• DataFusion's AggregateStatistics rule is metadata-only and bails on any filter — we need to do better.
• Iceberg's all-or-nothing pushdown is the cleanest reference implementation but trades performance for simplicity.
• Druid's bitmap-based filtered groupBy is the closest analogue to what bucket (a) does in Lance.

Start with the design overview and the linked sub-issues for IndexedAggregateExec, the physical optimizer rule, and COUNT(DISTINCT) cross-bucket combining. The tracking issue is complete when those scoped aggregate-pushdown work items are implemented while preserving deletion-vector handling, shared partial-state schemas, and metadata-only plan classification.