feat: thread statistics requests into ScanArgs

Add a `statistics_requests` field to `ScanArgs` (with
`with_statistics_requests` / `statistics_requests` accessors) and have
the physical planner thread `TableScan::statistics_requests` into it.

This completes the request-side path: a custom optimizer rule annotates
`TableScan`, and the request reaches a custom `TableProvider` in
`scan_with_args`. DataFusion's own providers ignore the field; the
default `ScanArgs` value is an empty slice.

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,27 @@ impl<'a> ScanArgs<'a> {
     pub fn limit(&self) -> Option<usize> {
         self.limit
     }
+
+    /// Set the statistics the caller would like the provider to answer for
+    /// this scan, if it can do so cheaply.
+    ///
+    /// Providers read these via [`Self::statistics_requests()`]; anything a
+    /// provider cannot answer cheaply it simply ignores. DataFusion's own
+    /// `TableProvider`s ignore this field — it exists so a request can be
+    /// threaded from a custom optimizer rule (which annotates
+    /// `TableScan::statistics_requests`) through to a custom provider.
+    pub fn with_statistics_requests(
+        mut self,
+        statistics_requests: &'a [StatisticsRequest],
+    ) -> Self {
+        self.statistics_requests = statistics_requests;
+        self
+    }
+
+    /// Get the statistics requests for the scan. 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/core/src/physical_planner.rs

@@ -645,6 +645,7 @@ impl DefaultPhysicalPlanner {
                     filters,
                     fetch,
                     projected_schema,
+                    statistics_requests,
                     ..
                 } = scan;
 
@@ -657,7 +658,8 @@ impl DefaultPhysicalPlanner {
                     let opts = ScanArgs::default()
                         .with_projection(projection.as_deref())
                         .with_filters(Some(&filters_vec))
-                        .with_limit(*fetch);
+                        .with_limit(*fetch)
+                        .with_statistics_requests(statistics_requests);
                     let res = source.scan_with_args(session_state, opts).await?;
                     Arc::clone(res.plan())
                 } else {