apache
diff --git a/‎datafusion/catalog-listing/src/table.rs‎
Lines changed: 110 additions & 3 deletions b/‎datafusion/catalog-listing/src/table.rs‎
Lines changed: 110 additions & 3 deletions
diff --git a/‎datafusion/catalog/src/table.rs‎
Lines changed: 62 additions & 2 deletions b/‎datafusion/catalog/src/table.rs‎
Lines changed: 62 additions & 2 deletions
diff --git a/‎datafusion/common/src/stats.rs‎
Lines changed: 109 additions & 0 deletions b/‎datafusion/common/src/stats.rs‎
Lines changed: 109 additions & 0 deletions
@@ -21,9 +21,10 @@ use crate::{ListingOptions, ListingTableConfig};
 use arrow::datatypes::{Field, Schema, SchemaBuilder, SchemaRef};
 use async_trait::async_trait;
 use datafusion_catalog::{ScanArgs, ScanResult, Session, TableProvider};
-use datafusion_common::stats::Precision;
+use datafusion_common::stats::{Precision, StatisticsRequest, StatisticsValue};
 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;
@@ -515,6 +516,15 @@ impl TableProvider for ListingTable {
             .list_files_for_scan(state, &partition_filters, statistic_file_limit)
             .await?;
 
+        // Snapshot the merged stats if the caller asked for cheap-stats
+        // answers — `statistics` is moved into the FileScanConfig builder
+        // below, so we need a snapshot to read from afterwards. Skipped
+        // entirely when no `statistics_requests` were passed.
+        let stats_snapshot_for_requests = args
+            .statistics_requests()
+            .filter(|_| self.options.collect_stat)
+            .map(|_| statistics.clone());
+
         // 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())?;
@@ -583,7 +593,24 @@ impl TableProvider for ListingTable {
             )
             .await?;
 
-        Ok(ScanResult::new(plan))
+        // Answer any requested stats from the table-level metadata we
+        // already touched. Anything not derivable from the dense
+        // `Statistics` we computed comes back as `Absent`. Skipped
+        // entirely when the caller didn't ask. We also skip when
+        // `collect_statistics=false` — the contract is "answer what's
+        // free", and computing stats here just to populate this map
+        // would violate that.
+        let stats_responses = match (
+            args.statistics_requests(),
+            stats_snapshot_for_requests.as_ref(),
+        ) {
+            (Some(reqs), Some(stats)) => {
+                answer_statistics_requests(reqs, &self.table_schema, stats)
+            }
+            _ => Vec::new(),
+        };
+
+        Ok(ScanResult::new(plan).with_statistics(stats_responses))
     }
 
     fn supports_filters_pushdown(
@@ -688,6 +715,86 @@ impl TableProvider for ListingTable {
     }
 }
 
+/// Answer a batch of [`StatisticsRequest`]s from a pre-computed
+/// table-level [`Statistics`]. Used by `ListingTable::scan_with_args`
+/// when the caller passes `args.with_statistics_requests(...)`.
+///
+/// Returns a vector aligned 1:1 with `requests`. Anything not derivable
+/// from the merged dense `Statistics` (typically: `DistinctCount`,
+/// `Sum`, `ByteSize`, since parquet thrift footers don't carry those)
+/// returns [`StatisticsValue::Absent`]; the caller decides whether to
+/// fill the gaps via another mechanism.
+fn answer_statistics_requests(
+    requests: &[StatisticsRequest],
+    schema: &SchemaRef,
+    stats: &Statistics,
+) -> Vec<StatisticsValue> {
+    let mut out = Vec::with_capacity(requests.len());
+    for req in requests {
+        out.push(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(col) => column_scalar(schema, stats, &col.name, |cs| &cs.min_value),
+            StatisticsRequest::Max(col) => column_scalar(schema, stats, &col.name, |cs| &cs.max_value),
+            StatisticsRequest::Sum(col) => column_scalar(schema, stats, &col.name, |cs| &cs.sum_value),
+            StatisticsRequest::NullCount(col) => column_usize(schema, stats, &col.name, |cs| &cs.null_count),
+            StatisticsRequest::DistinctCount(col) => column_usize(schema, stats, &col.name, |cs| &cs.distinct_count),
+            StatisticsRequest::ByteSize(col) => column_usize(schema, stats, &col.name, |cs| &cs.byte_size),
+        });
+    }
+    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
 
@@ -23,6 +23,7 @@ use std::sync::Arc;
 use crate::session::Session;
 use arrow::datatypes::SchemaRef;
 use async_trait::async_trait;
+use datafusion_common::stats::{StatisticsRequest, StatisticsValue};
 use datafusion_common::{Constraints, Statistics, not_impl_err};
 use datafusion_common::{Result, internal_err};
 use datafusion_expr::Expr;
@@ -406,6 +407,7 @@ pub struct ScanArgs<'a> {
     filters: Option<&'a [Expr]>,
     projection: Option<&'a [usize]>,
     limit: Option<usize>,
+    statistics_requests: Option<&'a [StatisticsRequest]>,
 }
 
 impl<'a> ScanArgs<'a> {
@@ -467,22 +469,68 @@ 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: Option<&'a [StatisticsRequest]>,
+    ) -> Self {
+        self.statistics_requests = requests;
+        self
+    }
+
+    /// Get the statistics requests, if any.
+    pub fn statistics_requests(&self) -> Option<&'a [StatisticsRequest]> {
+        self.statistics_requests
+    }
 }
 
 /// Result of a table scan operation from [`TableProvider::scan_with_args`].
 #[derive(Debug, Clone)]
 pub struct ScanResult {
     /// The ExecutionPlan to run.
     plan: Arc<dyn ExecutionPlan>,
+    /// Responses paired 1:1 with `ScanArgs::statistics_requests`. Empty
+    /// if no requests were made or the provider declined to answer any.
+    statistics: Vec<StatisticsValue>,
 }
 
 impl ScanResult {
-    /// Create a new `ScanResult` with the given execution plan.
+    /// Create a new `ScanResult` with the given execution plan and no
+    /// statistics responses.
     ///
     /// # Arguments
     /// * `plan` - The execution plan that will perform the table scan
     pub fn new(plan: Arc<dyn ExecutionPlan>) -> Self {
-        Self { plan }
+        Self {
+            plan,
+            statistics: Vec::new(),
+        }
+    }
+
+    /// Attach statistics responses to this result.
+    ///
+    /// The vector MUST have the same length and ordering as the
+    /// `ScanArgs::statistics_requests` slice that was passed in. Use
+    /// [`StatisticsValue::Absent`] for individual requests the provider
+    /// can't answer; the caller pairs requests with responses by index.
+    pub fn with_statistics(mut self, statistics: Vec<StatisticsValue>) -> Self {
+        self.statistics = statistics;
+        self
     }
 
     /// Get a reference to the execution plan for this scan result.
@@ -493,13 +541,25 @@ impl ScanResult {
         &self.plan
     }
 
+    /// Get the statistics responses, paired with the requests sent in
+    /// via [`ScanArgs::with_statistics_requests`]. Empty if no requests
+    /// were made.
+    pub fn statistics(&self) -> &[StatisticsValue] {
+        &self.statistics
+    }
+
     /// Consume this ScanResult and return the execution plan.
     ///
     /// Returns the owned [`ExecutionPlan`] that will perform
     /// the actual table scanning and data retrieval.
     pub fn into_inner(self) -> Arc<dyn ExecutionPlan> {
         self.plan
     }
+
+    /// Consume this ScanResult and return the (plan, statistics) pair.
+    pub fn into_parts(self) -> (Arc<dyn ExecutionPlan>, Vec<StatisticsValue>) {
+        (self.plan, self.statistics)
+    }
 }
 
 impl From<Arc<dyn ExecutionPlan>> for ScanResult {
 
@@ -1172,13 +1172,122 @@ impl ColumnStatistics {
     }
 }
 
+// ---------------------------------------------------------------------------
+// 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.
+// ---------------------------------------------------------------------------
+
+/// What stat does the caller want?
+///
+/// Each variant maps onto a field of [`Statistics`] / [`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`. Mirrors [`ColumnStatistics::min_value`].
+    Min(crate::Column),
+    /// Largest non-null value of `column`. Mirrors [`ColumnStatistics::max_value`].
+    Max(crate::Column),
+    /// Number of NULLs in `column`. Mirrors [`ColumnStatistics::null_count`].
+    NullCount(crate::Column),
+    /// Number of distinct values in `column` (exact or estimated).
+    /// Mirrors [`ColumnStatistics::distinct_count`].
+    DistinctCount(crate::Column),
+    /// Sum of values in `column` (numerics, widened per
+    /// [`ColumnStatistics::sum_value`]).
+    Sum(crate::Column),
+    /// Encoded/output byte size of `column`. Mirrors
+    /// [`ColumnStatistics::byte_size`].
+    ByteSize(crate::Column),
+    /// Number of rows in the container (table / file). Mirrors
+    /// [`Statistics::num_rows`].
+    RowCount,
+    /// Total byte size of the container's output. Mirrors
+    /// [`Statistics::total_byte_size`].
+    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` /
+/// `ApproxDistinct` return `Scalar(Precision::*(ScalarValue::UInt64(...)))`.
+///
+/// `Distribution` and `Sketch` are reserved for future use (histograms,
+/// theta/HLL sketches, raw sketch state for downstream merging) — the
+/// optimizer doesn't consume them today, but the type carries an opaque
+/// payload so providers can populate them ahead of optimizer support.
+#[derive(Debug, Clone)]
+pub enum StatisticsValue {
+    /// A single scalar value.
+    Scalar(Precision<ScalarValue>),
+    /// A distribution / histogram. Carries an opaque payload for now.
+    Distribution(std::sync::Arc<dyn std::any::Any + Send + Sync>),
+    /// A sketch (HLL, theta, KLL, …). Carries an opaque payload for now.
+    Sketch(std::sync::Arc<dyn std::any::Any + Send + Sync>),
+    /// Provider can't (or won't) answer this request. The caller decides
+    /// whether to fall back to another mechanism (re-route to a different
+    /// stats source, run a sampled scan, give up, …).
+    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))
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::assert_contains;
     use arrow::datatypes::Field;
     use std::sync::Arc;
 
+    #[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).
+        use crate::Column;
+        use std::collections::HashMap;
+        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:?}"),
+        }
+    }
+
     #[test]
     fn test_get_value() {
         let exact_precision = Precision::Exact(42);