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/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,

datafusion/expr-common/src/statistics.rs

@@ -1632,3 +1632,121 @@ mod tests {
         all_ops.into_iter().collect()
     }
 }
+
+// ---------------------------------------------------------------------------
+// Query-aware statistics request / response.
+//
+// A small extension to the existing `Statistics` model: instead of
+// "give me everything you have for every column", a caller can ask for
+// a specific list of stats by name. Providers that have something
+// cheap to offer (parquet thrift footers, an external catalog, cached
+// metadata) answer the entries they can; everything else comes back
+// `Absent`. Callers (optimizer rules, layered helpers, etc.) decide
+// what to do with the gaps.
+//
+// See `TableProvider::scan_with_args`
+// (`ScanArgs::with_statistics_requests` / `ScanResult::statistics`)
+// for the table-level handshake, and `PartitionedFile::satisfied_stats`
+// for the per-file one.
+// ---------------------------------------------------------------------------
+
+use datafusion_common::Column;
+use datafusion_common::stats::Precision;
+
+/// What stat does the caller want?
+///
+/// Each variant maps onto a field of
+/// [`datafusion_common::Statistics`] / [`datafusion_common::ColumnStatistics`]
+/// so providers that already populate one can answer the other
+/// trivially. The companion [`StatisticsValue`] is paired 1:1 with
+/// the request in the response. Whether a value is exact or estimated
+/// is encoded in the returned [`Precision`] wrapper, not in the
+/// request kind itself — `DistinctCount` covers both an exact distinct
+/// count from a metadata catalog and an HLL-style estimate from a
+/// sampled scan.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub enum StatisticsRequest {
+    /// Smallest non-null value of `column`.
+    Min(Column),
+    /// Largest non-null value of `column`.
+    Max(Column),
+    /// Number of NULLs in `column`.
+    NullCount(Column),
+    /// Number of distinct values in `column` (exact or estimated).
+    DistinctCount(Column),
+    /// Sum of values in `column` (numerics, widened per
+    /// `ColumnStatistics::sum_value`).
+    Sum(Column),
+    /// Encoded/output byte size of `column`.
+    ByteSize(Column),
+    /// Number of rows in the container (table / file).
+    RowCount,
+    /// Total byte size of the container's output.
+    TotalByteSize,
+}
+
+/// Response value paired 1:1 with an inbound [`StatisticsRequest`].
+///
+/// Variants are intentionally schema-agnostic: a provider answering
+/// `Min(c)` returns `Scalar(Precision::Exact(ScalarValue::...))` with
+/// the column's natural type; `RowCount` / `NullCount` / `DistinctCount`
+/// return `Scalar(Precision::*(ScalarValue::UInt64(...)))`. The
+/// `Distribution` variant carries a richly-typed [`Distribution`] for
+/// providers that have one to surface.
+#[derive(Debug, Clone)]
+pub enum StatisticsValue {
+    /// A single scalar value.
+    Scalar(Precision<ScalarValue>),
+    /// A typed probability distribution. Boxed so the enum stays small —
+    /// `Distribution` is significantly larger than the other variants.
+    Distribution(Box<Distribution>),
+    /// Provider can't (or won't) answer this request. The caller
+    /// decides whether to fall back to another mechanism.
+    Absent,
+}
+
+impl StatisticsValue {
+    /// Convenience: an `Exact` scalar response.
+    pub fn exact(value: ScalarValue) -> Self {
+        Self::Scalar(Precision::Exact(value))
+    }
+    /// Convenience: an `Inexact` scalar response.
+    pub fn inexact(value: ScalarValue) -> Self {
+        Self::Scalar(Precision::Inexact(value))
+    }
+}
+
+/// Sparse map of stats answers, keyed by request. Used as the storage for
+/// per-file answers (see `PartitionedFile::satisfied_stats`) and as the
+/// internal representation for adapters that consume request-driven
+/// stats. Only entries the provider actually answered are present, so
+/// memory scales with what was asked rather than with table width.
+pub type SatisfiedStatistics =
+    std::collections::HashMap<StatisticsRequest, StatisticsValue>;
+
+#[cfg(test)]
+mod stats_request_tests {
+    use super::*;
+    use std::collections::HashMap;
+
+    #[test]
+    fn statistics_request_is_hashable_keyable() {
+        // Sanity: two equal `StatisticsRequest`s hash equal and round-trip
+        // through a HashMap, so they can be used as keys (e.g. for the
+        // sparse `PartitionedFile::satisfied_stats` map).
+        let r1 = StatisticsRequest::Min(Column::new_unqualified("c"));
+        let r2 = StatisticsRequest::Min(Column::new_unqualified("c"));
+        assert_eq!(r1, r2);
+        let mut map: HashMap<StatisticsRequest, StatisticsValue> = HashMap::new();
+        map.insert(
+            r1.clone(),
+            StatisticsValue::exact(ScalarValue::Int64(Some(7))),
+        );
+        match map.get(&r2) {
+            Some(StatisticsValue::Scalar(Precision::Exact(ScalarValue::Int64(
+                Some(7),
+            )))) => {}
+            other => panic!("unexpected lookup: {other:?}"),
+        }
+    }
+}