feat(pruning): add StatisticsSource trait with two-phase resolve/evaluate API

[adriangb]

Summary

• Introduces StatisticsSource trait: an expression-based async statistics API that accepts &[Expr] and returns Vec<Option<ArrayRef>>
• Adds ResolvedStatistics: a HashMap<Expr, ArrayRef> cache that separates async data resolution from sync predicate evaluation
• Adds PruningPredicate::evaluate(): sync evaluation against pre-resolved stats cache
• Blanket impl bridges all existing PruningStatistics implementations automatically
• Refactors prune() to delegate through resolve_all_sync() + evaluate(), validating the two-phase pattern end-to-end

Design

The core idea is a two-phase resolve/evaluate split:

1. Resolve (async): PruningPredicate::all_required_expressions() exposes what stats are needed as Vec<Expr>. The caller passes these to StatisticsSource::expression_statistics(), which returns arrays packaged into a ResolvedStatistics cache.
2. Evaluate (sync): PruningPredicate::evaluate(&ResolvedStatistics) looks up each required expression in the cache, null-fills missing entries (conservative — won't prune), builds a RecordBatch, and evaluates the predicate.

This keeps the evaluation path synchronous for Stream::poll_next() contexts like EarlyStoppingStream, while allowing the resolution step to be async.

Future work

Struct field pruning (#21003)

Because StatisticsSource accepts arbitrary Expr, a custom implementation can handle expressions like min(get_field(struct_col, 'field')) by resolving nested Parquet column statistics directly. The blanket impl on PruningStatistics returns None for these (it only handles flat Expr::Column args), but a Parquet-aware StatisticsSource impl can override this. No further API changes needed — the expression language is already rich enough.

Async statistics sources

The async StatisticsSource trait enables use cases like querying an external metastore for statistics or sampling data at runtime. The two-phase pattern means callers resolve once (async) and evaluate many times (sync), which works well for dynamic filter scenarios where the predicate changes but the underlying data statistics don't.

Cardinality estimation

StatisticsSource could sit on ExecutionPlan nodes via a method like partition_expression_statistics(&[Expr]), delegating through DataSourceExec → FileScanConfig → FileSource → format-specific impl. This would enable queries like approx_count_distinct(col) for join optimization.

There is work in progress to add NDV statistics to Parquet but this could unlock things like extracting stats from sampled data.

Test plan

• All 82 existing pruning tests pass unchanged
• 16 new tests covering: resolve helpers (min/max/count/InList/NOT IN), ResolvedStatistics cache, evaluate-matches-prune equivalence, missing cache entries → conservative keep
• Zero clippy warnings
• datafusion-datasource-parquet compiles unchanged

🤖 Generated with Claude Code