catalog: query-aware statistics requests via ScanArgs / ScanResult

Adds an opt-in handshake that lets callers ask a `TableProvider` for
specific stats by name and receive only what the provider can answer
cheaply, instead of the all-or-nothing dense `Statistics` we have today.

## What's new

* `datafusion-common::stats::StatisticsRequest` — enum of stat kinds
  that mirror `Statistics` / `ColumnStatistics` (Min, Max, NullCount,
  DistinctCount, Sum, ByteSize, RowCount, TotalByteSize). `Hash + Eq`
  so it can key a `HashMap`.

* `datafusion-common::stats::StatisticsValue` — `Scalar(Precision<...>)
  | Distribution(Arc<dyn Any>) | Sketch(Arc<dyn Any>) | Absent`. Whether
  a value is exact or estimated travels in the `Precision` wrapper, not
  the variant.

* `ScanArgs::with_statistics_requests` / `statistics_requests()` — the
  caller's question.

* `ScanResult::with_statistics` / `statistics()` / `into_parts()` — the
  provider's answer, paired 1:1 with the requests slice.

* `PartitionedFile::satisfied_stats` — sparse,
  `Arc<HashMap<StatisticsRequest, StatisticsValue>>` for per-file
  answers. Memory scales with what was asked, not with table width.
  Providers that store stats out-of-band (Delta/Iceberg/Hudi manifests,
  Hive Metastore, custom catalogs) can populate this directly without
  rebuilding a full dense `Statistics`.

* `FilePruner` learns to consume the sparse map. Internally,
  `file_stats_pruning` is now `Box<dyn PruningStatistics + Send + Sync>`
  so we can dispatch between the existing `PrunableStatistics` (dense)
  and a new `SparseFilePruningStats` adapter (sparse). The sparse
  adapter looks up each `StatisticsRequest` directly in the map and
  materializes single-row arrays only for the columns the pruning
  predicate touches — no densify-then-throw-away.

* `ListingTable::scan_with_args` populates `ScanResult.statistics` from
  the merged dense `Statistics` it already computed when
  `args.statistics_requests()` is set and `collect_statistics=true`.
  When `collect_statistics=false` it returns `Absent` for everything
  (the contract is "answer what's free"). `DistinctCount`/`Sum`/
  `ByteSize` are likewise `Absent` for parquet — those aren't in
  thrift footers; layered helpers (or richer providers) can fill the
  gaps.

## Backwards compat

All additions are opt-in:

* `ScanArgs` / `ScanResult` gain new fields with `Default`-friendly
  initializers; existing callers that don't use the new builders see
  no change.
* `FilePruner`'s field-type change is internal (private field).
* The only minor source-level break is a new pub field on
  `PartitionedFile` (`satisfied_stats`). Callers using
  `PartitionedFile::new` / `From<ObjectMeta>` / the existing builders
  are unaffected. Direct struct literals — uncommon, none in-tree —
  need to add `satisfied_stats: None` (or use the new
  `with_satisfied_stats` builder).

## Tests

* `datafusion-common::stats::tests::statistics_request_is_hashable_keyable`
  — round-trip a `StatisticsRequest` through a `HashMap`.
* `datafusion-pruning::file_pruner::tests` — three tests demonstrating
  end-to-end pruning against a sparse-only `PartitionedFile` (`x > 100`
  prunes a `[10, 20]` file, `x > 15` doesn't, no stats at all → no
  pruner).

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

Author: adriangb

datafusion/catalog-listing/src/table.rs

@@ -23,7 +23,8 @@ use async_trait::async_trait;
 use datafusion_catalog::{ScanArgs, ScanResult, Session, TableProvider};
 use datafusion_common::stats::Precision;
 use datafusion_common::{
-    Constraints, SchemaExt, Statistics, internal_datafusion_err, plan_err, project_schema,
+    Constraints, ScalarValue, SchemaExt, Statistics, internal_datafusion_err, plan_err,
+    project_schema,
 };
 use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_groups::FileGroup;
@@ -39,6 +40,9 @@ use datafusion_execution::cache::cache_manager::FileStatisticsCache;
 use datafusion_execution::cache::cache_unit::DefaultFileStatisticsCache;
 use datafusion_expr::dml::InsertOp;
 use datafusion_expr::execution_props::ExecutionProps;
+use datafusion_expr::statistics::{
+    SatisfiedStatistics, StatisticsRequest, StatisticsValue,
+};
 use datafusion_expr::{Expr, TableProviderFilterPushDown, TableType};
 use datafusion_physical_expr::create_lex_ordering;
 use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory;
@@ -515,6 +519,39 @@ impl TableProvider for ListingTable {
             .list_files_for_scan(state, &partition_filters, statistic_file_limit)
             .await?;
 
+        // If the caller asked for per-file stats and we already have dense
+        // per-file `Statistics` (i.e. `collect_statistics=true` populated
+        // them during file listing), project them into a sparse
+        // `satisfied_stats` map keyed by request. This is "free" — the IO
+        // / decode cost was already paid; we're just surfacing what we
+        // have in the request-driven shape that downstream consumers
+        // (e.g. `FilePruner` via `SparseFilePruningStats`) prefer.
+        if !args.statistics_requests().is_empty() && self.options.collect_stat {
+            partitioned_file_lists = partitioned_file_lists
+                .into_iter()
+                .map(|fg| {
+                    let files = fg
+                        .into_inner()
+                        .into_iter()
+                        .map(|pf| {
+                            let sparse = pf.statistics.as_ref().map(|stats| {
+                                project_satisfied_stats(
+                                    args.statistics_requests(),
+                                    &self.table_schema,
+                                    stats,
+                                )
+                            });
+                            match sparse {
+                                Some(s) => pf.with_satisfied_stats(Arc::new(s)),
+                                None => pf,
+                            }
+                        })
+                        .collect();
+                    FileGroup::new(files)
+                })
+                .collect();
+        }
+
         // if no files need to be read, return an `EmptyExec`
         if partitioned_file_lists.is_empty() {
             let projected_schema = project_schema(&self.schema(), projection.as_ref())?;
@@ -688,6 +725,112 @@ impl TableProvider for ListingTable {
     }
 }
 
+/// Project a per-file dense [`Statistics`] into the sparse map shape
+/// that downstream consumers (e.g. [`FilePruner`] via
+/// [`SparseFilePruningStats`]) read. Only entries the request list asked
+/// for are present; non-`Absent` slots are kept verbatim (preserves the
+/// `Precision` already on the field). The dense `Statistics` itself is
+/// untouched.
+///
+/// `DistinctCount` / `Sum` / `ByteSize` will only be present if the
+/// underlying file format actually populated them on
+/// [`ColumnStatistics`] — for parquet, only `null_count`, `min_value`,
+/// `max_value`, and `num_rows` are typically available from thrift
+/// footers; the rest stay `Absent` (so the entry is omitted).
+///
+/// [`FilePruner`]: datafusion_pruning::FilePruner
+/// [`SparseFilePruningStats`]: datafusion_pruning::SparseFilePruningStats
+/// [`ColumnStatistics`]: datafusion_common::ColumnStatistics
+fn project_satisfied_stats(
+    requests: &[StatisticsRequest],
+    schema: &SchemaRef,
+    stats: &Statistics,
+) -> SatisfiedStatistics {
+    let mut out = SatisfiedStatistics::new();
+    for req in requests {
+        let value = match req {
+            StatisticsRequest::RowCount => precision_to_scalar(&stats.num_rows, |n| {
+                ScalarValue::UInt64(Some(n as u64))
+            }),
+            StatisticsRequest::TotalByteSize => {
+                precision_to_scalar(&stats.total_byte_size, |n| {
+                    ScalarValue::UInt64(Some(n as u64))
+                })
+            }
+            StatisticsRequest::Min(c) => {
+                column_scalar(schema, stats, &c.name, |cs| &cs.min_value)
+            }
+            StatisticsRequest::Max(c) => {
+                column_scalar(schema, stats, &c.name, |cs| &cs.max_value)
+            }
+            StatisticsRequest::Sum(c) => {
+                column_scalar(schema, stats, &c.name, |cs| &cs.sum_value)
+            }
+            StatisticsRequest::NullCount(c) => {
+                column_usize(schema, stats, &c.name, |cs| &cs.null_count)
+            }
+            StatisticsRequest::DistinctCount(c) => {
+                column_usize(schema, stats, &c.name, |cs| &cs.distinct_count)
+            }
+            StatisticsRequest::ByteSize(c) => {
+                column_usize(schema, stats, &c.name, |cs| &cs.byte_size)
+            }
+        };
+        // Skip `Absent` entries — the sparse map's invariant is "only
+        // entries the provider answered". Consumers infer `Absent` from
+        // a missing key.
+        if !matches!(value, StatisticsValue::Absent) {
+            out.insert(req.clone(), value);
+        }
+    }
+    out
+}
+
+fn precision_to_scalar(
+    p: &Precision<usize>,
+    map: impl Fn(usize) -> ScalarValue,
+) -> StatisticsValue {
+    match p {
+        Precision::Exact(n) => StatisticsValue::Scalar(Precision::Exact(map(*n))),
+        Precision::Inexact(n) => StatisticsValue::Scalar(Precision::Inexact(map(*n))),
+        Precision::Absent => StatisticsValue::Absent,
+    }
+}
+
+fn column_scalar(
+    schema: &SchemaRef,
+    stats: &Statistics,
+    name: &str,
+    pick: impl Fn(&datafusion_common::ColumnStatistics) -> &Precision<ScalarValue>,
+) -> StatisticsValue {
+    let Ok(idx) = schema.index_of(name) else {
+        return StatisticsValue::Absent;
+    };
+    let Some(cs) = stats.column_statistics.get(idx) else {
+        return StatisticsValue::Absent;
+    };
+    match pick(cs) {
+        Precision::Exact(v) => StatisticsValue::Scalar(Precision::Exact(v.clone())),
+        Precision::Inexact(v) => StatisticsValue::Scalar(Precision::Inexact(v.clone())),
+        Precision::Absent => StatisticsValue::Absent,
+    }
+}
+
+fn column_usize(
+    schema: &SchemaRef,
+    stats: &Statistics,
+    name: &str,
+    pick: impl Fn(&datafusion_common::ColumnStatistics) -> &Precision<usize>,
+) -> StatisticsValue {
+    let Ok(idx) = schema.index_of(name) else {
+        return StatisticsValue::Absent;
+    };
+    let Some(cs) = stats.column_statistics.get(idx) else {
+        return StatisticsValue::Absent;
+    };
+    precision_to_scalar(pick(cs), |n| ScalarValue::UInt64(Some(n as u64)))
+}
+
 impl ListingTable {
     /// Get the list of files for a scan as well as the file level statistics.
     /// The list is grouped to let the execution plan know how the files should
@@ -1049,4 +1192,78 @@ mod tests {
         let result = derive_common_ordering_from_files(&file_groups);
         assert_eq!(result, Some(ordering));
     }
+
+    #[test]
+    fn project_satisfied_stats_picks_only_requested_and_present() {
+        use arrow::datatypes::{DataType, Field, Schema};
+        use datafusion_common::{Column, ColumnStatistics};
+        use datafusion_expr::statistics::{StatisticsRequest, StatisticsValue};
+
+        let schema: SchemaRef = Arc::new(Schema::new(vec![
+            Field::new("a", DataType::Int64, true),
+            Field::new("b", DataType::Int64, true),
+        ]));
+
+        let mut stats = Statistics::new_unknown(&schema);
+        stats.num_rows = Precision::Exact(42);
+        // a: min/max known, null_count Absent
+        stats.column_statistics[0] = ColumnStatistics {
+            null_count: Precision::Absent,
+            max_value: Precision::Exact(ScalarValue::Int64(Some(10))),
+            min_value: Precision::Exact(ScalarValue::Int64(Some(1))),
+            sum_value: Precision::Absent,
+            distinct_count: Precision::Absent,
+            byte_size: Precision::Absent,
+        };
+        // b: only null_count known
+        stats.column_statistics[1] = ColumnStatistics {
+            null_count: Precision::Exact(3),
+            max_value: Precision::Absent,
+            min_value: Precision::Absent,
+            sum_value: Precision::Absent,
+            distinct_count: Precision::Absent,
+            byte_size: Precision::Absent,
+        };
+
+        let a = Column::from_name("a");
+        let b = Column::from_name("b");
+        let requests = vec![
+            StatisticsRequest::RowCount,
+            StatisticsRequest::Min(a.clone()),
+            StatisticsRequest::Max(a.clone()),
+            StatisticsRequest::NullCount(a.clone()), // Absent — should be omitted
+            StatisticsRequest::NullCount(b.clone()),
+            StatisticsRequest::Min(b.clone()), // Absent — should be omitted
+        ];
+
+        let out = project_satisfied_stats(&requests, &schema, &stats);
+
+        assert_eq!(out.len(), 4);
+        assert!(matches!(
+            out.get(&StatisticsRequest::RowCount),
+            Some(StatisticsValue::Scalar(Precision::Exact(
+                ScalarValue::UInt64(Some(42))
+            )))
+        ));
+        assert!(matches!(
+            out.get(&StatisticsRequest::Min(a.clone())),
+            Some(StatisticsValue::Scalar(Precision::Exact(
+                ScalarValue::Int64(Some(1))
+            )))
+        ));
+        assert!(matches!(
+            out.get(&StatisticsRequest::Max(a.clone())),
+            Some(StatisticsValue::Scalar(Precision::Exact(
+                ScalarValue::Int64(Some(10))
+            )))
+        ));
+        assert!(matches!(
+            out.get(&StatisticsRequest::NullCount(b.clone())),
+            Some(StatisticsValue::Scalar(Precision::Exact(
+                ScalarValue::UInt64(Some(3))
+            )))
+        ));
+        assert!(!out.contains_key(&StatisticsRequest::NullCount(a)));
+        assert!(!out.contains_key(&StatisticsRequest::Min(b)));
+    }
 }

datafusion/catalog/src/table.rs

@@ -26,6 +26,7 @@ use async_trait::async_trait;
 use datafusion_common::{Constraints, Statistics, not_impl_err};
 use datafusion_common::{Result, internal_err};
 use datafusion_expr::Expr;
+use datafusion_expr::statistics::StatisticsRequest;
 
 use datafusion_expr::dml::InsertOp;
 use datafusion_expr::{
@@ -406,6 +407,7 @@ pub struct ScanArgs<'a> {
     filters: Option<&'a [Expr]>,
     projection: Option<&'a [usize]>,
     limit: Option<usize>,
+    statistics_requests: &'a [StatisticsRequest],
 }
 
 impl<'a> ScanArgs<'a> {
@@ -467,6 +469,31 @@ impl<'a> ScanArgs<'a> {
     pub fn limit(&self) -> Option<usize> {
         self.limit
     }
+
+    /// Set a list of statistics the caller would like the provider to
+    /// answer if it can do so cheaply.
+    ///
+    /// Typical sources a provider may answer from:
+    /// * Parquet thrift footers (Min/Max/NullCount/RowCount, exact)
+    /// * An external metadata catalog (Delta/Iceberg/Hudi manifests,
+    ///   Hive-Metastore-style stats columns)
+    /// * Cached / materialized stats columns
+    ///
+    /// The provider returns its answers on
+    /// [`ScanResult::statistics`] paired 1:1 with `requests`. Anything
+    /// not answerable should come back as [`StatisticsValue::Absent`] —
+    /// the caller decides what to do with the gaps. The contract is
+    /// "answer what's free, leave the rest as `Absent`": providers MUST
+    /// NOT do expensive scans purely to satisfy these requests.
+    pub fn with_statistics_requests(mut self, requests: &'a [StatisticsRequest]) -> Self {
+        self.statistics_requests = requests;
+        self
+    }
+
+    /// Get the statistics requests. Empty if none were set.
+    pub fn statistics_requests(&self) -> &'a [StatisticsRequest] {
+        self.statistics_requests
+    }
 }
 
 /// Result of a table scan operation from [`TableProvider::scan_with_args`].

datafusion/datasource/src/mod.rs

@@ -58,6 +58,7 @@ use chrono::TimeZone;
 use datafusion_common::stats::Precision;
 use datafusion_common::{ColumnStatistics, Result, exec_datafusion_err};
 use datafusion_common::{ScalarValue, Statistics};
+use datafusion_expr::statistics::SatisfiedStatistics;
 use datafusion_physical_expr::LexOrdering;
 use futures::{Stream, StreamExt};
 use object_store::{GetOptions, GetRange, ObjectStore};
@@ -138,6 +139,18 @@ pub struct PartitionedFile {
     /// When set via [`Self::with_statistics`], partition column statistics are automatically
     /// computed from [`Self::partition_values`] with exact min/max/null_count/distinct_count.
     pub statistics: Option<Arc<Statistics>>,
+    /// Sparse, request-keyed stats answered by the provider for this file.
+    ///
+    /// Only entries for non-`Absent` answers are present, so memory scales
+    /// with the *count of stats actually requested* rather than the table's
+    /// column count. Used in tandem with — not in place of —
+    /// [`Self::statistics`]: existing consumers that read the dense
+    /// `Statistics` keep working; new consumers (e.g. `FilePruner` in
+    /// `datafusion-pruning`) prefer this sparse map when it's
+    /// populated. Providers that store stats out-of-band (Delta/Iceberg/Hudi
+    /// manifests, Hive Metastore, custom catalogs) can populate this
+    /// directly without rebuilding a full dense `Statistics`.
+    pub satisfied_stats: Option<Arc<SatisfiedStatistics>>,
     /// The known lexicographical ordering of the rows in this file, if any.
     ///
     /// This describes how the data within the file is sorted with respect to one or more
@@ -168,6 +181,7 @@ impl PartitionedFile {
             partition_values: vec![],
             range: None,
             statistics: None,
+            satisfied_stats: None,
             ordering: None,
             extensions: None,
             metadata_size_hint: None,
@@ -181,6 +195,7 @@ impl PartitionedFile {
             partition_values: vec![],
             range: None,
             statistics: None,
+            satisfied_stats: None,
             ordering: None,
             extensions: None,
             metadata_size_hint: None,
@@ -200,6 +215,7 @@ impl PartitionedFile {
             partition_values: vec![],
             range: Some(FileRange { start, end }),
             statistics: None,
+            satisfied_stats: None,
             ordering: None,
             extensions: None,
             metadata_size_hint: None,
@@ -328,6 +344,21 @@ impl PartitionedFile {
         self.ordering = ordering;
         self
     }
+
+    /// Attach a sparse map of provider-answered statistics for this file.
+    ///
+    /// Used by table providers that store per-file stats out-of-band
+    /// (Delta/Iceberg/Hudi manifests, Hive Metastore, custom catalogs)
+    /// and want to surface them without reconstructing a full dense
+    /// [`Statistics`]. Consumers (e.g. `FilePruner`) prefer this map
+    /// when [`Self::statistics`] is not set.
+    pub fn with_satisfied_stats(
+        mut self,
+        satisfied_stats: Arc<SatisfiedStatistics>,
+    ) -> Self {
+        self.satisfied_stats = Some(satisfied_stats);
+        self
+    }
 }
 
 impl From<ObjectMeta> for PartitionedFile {
@@ -337,6 +368,7 @@ impl From<ObjectMeta> for PartitionedFile {
             partition_values: vec![],
             range: None,
             statistics: None,
+            satisfied_stats: None,
             ordering: None,
             extensions: None,
             metadata_size_hint: None,
@@ -534,6 +566,7 @@ pub fn generate_test_files(num_files: usize, overlap_factor: f64) -> Vec<FileGro
                     byte_size: Precision::Absent,
                 }],
             })),
+            satisfied_stats: None,
             ordering: None,
             extensions: None,
             metadata_size_hint: None,