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

[adriangb]

@kumarUjjawal i'm curious if you'd be interested in this since you've been working on statistics?

[kumarUjjawal]

@kumarUjjawal i'm curious if you'd be interested in this since you've been working on statistics?

Thanks @adriangb I will take a look

[kumarUjjawal]

Just leaving some preliminary thoughts, let me know what you think:

PruningPredicate uses row_count as total rows in null_count != row_count (all-null guard), but all_required_expressions()encodes row count as count(*) FILTER (WHERE column IS NOT NULL).

A custom StatisticsSource can return non-null count for that expression, and then the all-null guard becomes wrong and can prune valid containers.
Current blanket impl hides this by mapping that expression to row_counts(), but external implementations are exposed to this contract mismatch.

We can either:

• make the row-count expression unambiguous for total rows, or
• clearly document/enforce that this specific filtered count expression is an alias for total row_counts.

[kumarUjjawal]

I think it silently truncates on length mismatch; If it returns a wrong-length vector, we silently lose mappings and continue with null fallback.