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

Introduces a new expression-based statistics API for pruning that
separates async data resolution from sync predicate evaluation.

- StatisticsSource trait: accepts &[Expr], returns Vec<Option<ArrayRef>>
- ResolvedStatistics: HashMap<Expr, ArrayRef> cache for pre-resolved stats
- PruningPredicate::evaluate(): sync evaluation against pre-resolved cache
- PruningPredicate::all_required_expressions(): exposes needed Expr list
- Blanket impl bridges existing PruningStatistics implementations
- prune() refactored to delegate through resolve_all_sync + evaluate

This enables async statistics sources (external metastores, runtime
sampling) while keeping the evaluation path synchronous for use in
Stream::poll_next() contexts like EarlyStoppingStream.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: adriangb

Cargo.lock

@@ -17,9 +17,13 @@ workspace = true
 
 [dependencies]
 arrow = { workspace = true }
+async-trait = { workspace = true }
 datafusion-common = { workspace = true, default-features = true }
 datafusion-datasource = { workspace = true }
+datafusion-expr = { workspace = true, default-features = true }
 datafusion-expr-common = { workspace = true, default-features = true }
+datafusion-functions-aggregate = { workspace = true, default-features = true }
+datafusion-functions-nested = { workspace = true, default-features = true }
 datafusion-physical-expr = { workspace = true }
 datafusion-physical-expr-common = { workspace = true }
 datafusion-physical-plan = { workspace = true }
@@ -30,3 +34,4 @@ datafusion-expr = { workspace = true }
 datafusion-functions-nested = { workspace = true }
 insta = { workspace = true }
 itertools = { workspace = true }
+tokio = { workspace = true }

datafusion/pruning/Cargo.toml

@@ -19,9 +19,11 @@
 
 mod file_pruner;
 mod pruning_predicate;
+mod statistics;
 
 pub use file_pruner::FilePruner;
 pub use pruning_predicate::{
     PredicateRewriter, PruningPredicate, PruningStatistics, RequiredColumns,
     UnhandledPredicateHook, build_pruning_predicate,
 };
+pub use statistics::{ResolvedStatistics, StatisticsSource};