Move file source projected statistics into config

friendlymatthew · friendlymatthew · commit 2b92b1dd1707 · 2025-08-20T14:24:17.000-04:00
diff --git a/datafusion/core/src/datasource/file_format/mod.rs b/datafusion/core/src/datasource/file_format/mod.rs
@@ -89,7 +89,7 @@ pub(crate) mod test_util {
                     format.file_source(),
                 )
                 .with_file_groups(file_groups)
-                .with_statistics(statistics)
+                .with_file_source_projected_statistics(statistics)
                 .with_projection(projection)
                 .with_limit(limit)
                 .build(),
diff --git a/datafusion/core/src/datasource/listing/table.rs b/datafusion/core/src/datasource/listing/table.rs
@@ -1247,7 +1247,7 @@ impl TableProvider for ListingTable {
                 )
                 .with_file_groups(partitioned_file_lists)
                 .with_constraints(self.constraints.clone())
-                .with_statistics(statistics)
+                .with_file_source_projected_statistics(statistics)
                 .with_projection(projection.cloned())
                 .with_limit(limit)
                 .with_output_ordering(output_ordering)
diff --git a/datafusion/core/src/datasource/physical_plan/arrow_file.rs b/datafusion/core/src/datasource/physical_plan/arrow_file.rs
@@ -26,7 +26,6 @@ use datafusion_datasource::schema_adapter::SchemaAdapterFactory;
 use arrow::buffer::Buffer;
 use arrow::datatypes::SchemaRef;
 use arrow_ipc::reader::FileDecoder;
-use datafusion_common::Statistics;
 use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_scan_config::FileScanConfig;
 use datafusion_datasource::PartitionedFile;
@@ -41,7 +40,6 @@ use object_store::{GetOptions, GetRange, GetResultPayload, ObjectStore};
 #[derive(Clone, Default)]
 pub struct ArrowSource {
     metrics: ExecutionPlanMetricsSet,
-    projected_statistics: Option<Statistics>,
     schema_adapter_factory: Option<Arc<dyn SchemaAdapterFactory>>,
 }
 
@@ -75,11 +73,6 @@ impl FileSource for ArrowSource {
     fn with_schema(&self, _schema: SchemaRef) -> Arc<dyn FileSource> {
         Arc::new(Self { ..self.clone() })
     }
-    fn with_statistics(&self, statistics: Statistics) -> Arc<dyn FileSource> {
-        let mut conf = self.clone();
-        conf.projected_statistics = Some(statistics);
-        Arc::new(conf)
-    }
 
     fn with_projection(&self, _config: &FileScanConfig) -> Arc<dyn FileSource> {
         Arc::new(Self { ..self.clone() })
@@ -89,13 +82,6 @@ impl FileSource for ArrowSource {
         &self.metrics
     }
 
-    fn file_source_statistics(&self) -> Result<Statistics> {
-        let statistics = &self.projected_statistics;
-        Ok(statistics
-            .clone()
-            .expect("projected_statistics must be set"))
-    }
-
     fn file_type(&self) -> &str {
         "arrow"
     }
diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown/util.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown/util.rs
@@ -167,25 +167,10 @@ impl FileSource for TestSource {
         })
     }
 
-    fn with_statistics(&self, statistics: Statistics) -> Arc<dyn FileSource> {
-        Arc::new(TestSource {
-            statistics: Some(statistics),
-            ..self.clone()
-        })
-    }
-
     fn metrics(&self) -> &ExecutionPlanMetricsSet {
         &self.metrics
     }
 
-    fn file_source_statistics(&self) -> Result<Statistics> {
-        Ok(self
-            .statistics
-            .as_ref()
-            .expect("statistics not set")
-            .clone())
-    }
-
     fn file_type(&self) -> &str {
         "test"
     }
diff --git a/datafusion/core/tests/physical_optimizer/test_utils.rs b/datafusion/core/tests/physical_optimizer/test_utils.rs
@@ -131,13 +131,13 @@ pub(crate) fn parquet_exec_with_stats(file_size: u64) -> Arc<DataSourceExec> {
         Arc::new(ParquetSource::new(Default::default())),
     )
     .with_file(PartitionedFile::new("x".to_string(), file_size))
-    .with_statistics(statistics)
+    .with_file_source_projected_statistics(statistics)
     .build();
 
     assert_eq!(
         config
             .file_source
-            .file_source_statistics()
+            .file_source_statistics(&config)
             .unwrap()
             .num_rows,
         Precision::Inexact(10000)
diff --git a/datafusion/datasource-avro/src/source.rs b/datafusion/datasource-avro/src/source.rs
@@ -24,7 +24,6 @@ use crate::avro_to_arrow::Reader as AvroReader;
 
 use arrow::datatypes::SchemaRef;
 use datafusion_common::error::Result;
-use datafusion_common::Statistics;
 use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_scan_config::FileScanConfig;
 use datafusion_datasource::file_stream::FileOpener;
@@ -41,7 +40,6 @@ pub struct AvroSource {
     batch_size: Option<usize>,
     projection: Option<Vec<String>>,
     metrics: ExecutionPlanMetricsSet,
-    projected_statistics: Option<Statistics>,
     schema_adapter_factory: Option<Arc<dyn SchemaAdapterFactory>>,
 }
 
@@ -89,11 +87,6 @@ impl FileSource for AvroSource {
         conf.schema = Some(schema);
         Arc::new(conf)
     }
-    fn with_statistics(&self, statistics: Statistics) -> Arc<dyn FileSource> {
-        let mut conf = self.clone();
-        conf.projected_statistics = Some(statistics);
-        Arc::new(conf)
-    }
 
     fn with_projection(&self, config: &FileScanConfig) -> Arc<dyn FileSource> {
         let mut conf = self.clone();
@@ -105,13 +98,6 @@ impl FileSource for AvroSource {
         &self.metrics
     }
 
-    fn file_source_statistics(&self) -> Result<Statistics> {
-        let statistics = &self.projected_statistics;
-        Ok(statistics
-            .clone()
-            .expect("projected_statistics must be set"))
-    }
-
     fn file_type(&self) -> &str {
         "avro"
     }
diff --git a/datafusion/datasource-csv/src/source.rs b/datafusion/datasource-csv/src/source.rs
@@ -35,7 +35,7 @@ use datafusion_datasource::{
 
 use arrow::csv;
 use arrow::datatypes::SchemaRef;
-use datafusion_common::{DataFusionError, Result, Statistics};
+use datafusion_common::{DataFusionError, Result};
 use datafusion_common_runtime::JoinSet;
 use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_scan_config::FileScanConfig;
@@ -92,7 +92,6 @@ pub struct CsvSource {
     escape: Option<u8>,
     comment: Option<u8>,
     metrics: ExecutionPlanMetricsSet,
-    projected_statistics: Option<Statistics>,
     schema_adapter_factory: Option<Arc<dyn SchemaAdapterFactory>>,
 }
 
@@ -251,12 +250,6 @@ impl FileSource for CsvSource {
         Arc::new(conf)
     }
 
-    fn with_statistics(&self, statistics: Statistics) -> Arc<dyn FileSource> {
-        let mut conf = self.clone();
-        conf.projected_statistics = Some(statistics);
-        Arc::new(conf)
-    }
-
     fn with_projection(&self, config: &FileScanConfig) -> Arc<dyn FileSource> {
         let mut conf = self.clone();
         conf.file_projection = config.file_column_projection_indices();
@@ -266,12 +259,7 @@ impl FileSource for CsvSource {
     fn metrics(&self) -> &ExecutionPlanMetricsSet {
         &self.metrics
     }
-    fn file_source_statistics(&self) -> Result<Statistics> {
-        let statistics = &self.projected_statistics;
-        Ok(statistics
-            .clone()
-            .expect("projected_statistics must be set"))
-    }
+
     fn file_type(&self) -> &str {
         "csv"
     }
diff --git a/datafusion/datasource-json/src/source.rs b/datafusion/datasource-json/src/source.rs
@@ -38,7 +38,6 @@ use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
 
 use arrow::json::ReaderBuilder;
 use arrow::{datatypes::SchemaRef, json};
-use datafusion_common::Statistics;
 use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_scan_config::FileScanConfig;
 use datafusion_execution::TaskContext;
@@ -79,7 +78,6 @@ impl JsonOpener {
 pub struct JsonSource {
     batch_size: Option<usize>,
     metrics: ExecutionPlanMetricsSet,
-    projected_statistics: Option<Statistics>,
     schema_adapter_factory: Option<Arc<dyn SchemaAdapterFactory>>,
 }
 
@@ -126,11 +124,6 @@ impl FileSource for JsonSource {
     fn with_schema(&self, _schema: SchemaRef) -> Arc<dyn FileSource> {
         Arc::new(Self { ..self.clone() })
     }
-    fn with_statistics(&self, statistics: Statistics) -> Arc<dyn FileSource> {
-        let mut conf = self.clone();
-        conf.projected_statistics = Some(statistics);
-        Arc::new(conf)
-    }
 
     fn with_projection(&self, _config: &FileScanConfig) -> Arc<dyn FileSource> {
         Arc::new(Self { ..self.clone() })
@@ -140,13 +133,6 @@ impl FileSource for JsonSource {
         &self.metrics
     }
 
-    fn file_source_statistics(&self) -> Result<Statistics> {
-        let statistics = &self.projected_statistics;
-        Ok(statistics
-            .clone()
-            .expect("projected_statistics must be set to call"))
-    }
-
     fn file_type(&self) -> &str {
         "json"
     }
diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs
@@ -285,7 +285,6 @@ pub struct ParquetSource {
     pub(crate) batch_size: Option<usize>,
     /// Optional hint for the size of the parquet metadata
     pub(crate) metadata_size_hint: Option<usize>,
-    pub(crate) projected_statistics: Option<Statistics>,
     #[cfg(feature = "parquet_encryption")]
     pub(crate) encryption_factory: Option<Arc<dyn EncryptionFactory>>,
 }
@@ -600,12 +599,6 @@ impl FileSource for ParquetSource {
         })
     }
 
-    fn with_statistics(&self, statistics: Statistics) -> Arc<dyn FileSource> {
-        let mut conf = self.clone();
-        conf.projected_statistics = Some(statistics);
-        Arc::new(conf)
-    }
-
     fn with_projection(&self, _config: &FileScanConfig) -> Arc<dyn FileSource> {
         Arc::new(Self { ..self.clone() })
     }
@@ -614,11 +607,12 @@ impl FileSource for ParquetSource {
         &self.metrics
     }
 
-    fn file_source_statistics(&self) -> datafusion_common::Result<Statistics> {
-        let statistics = &self.projected_statistics;
-        let statistics = statistics
-            .clone()
-            .expect("projected_statistics must be set");
+    fn file_source_statistics(
+        &self,
+        config: &FileScanConfig,
+    ) -> datafusion_common::Result<Statistics> {
+        let statistics = config.file_source_projected_statistics.clone();
+
         // When filters are pushed down, we have no way of knowing the exact statistics.
         // Note that pruning predicate is also a kind of filter pushdown.
         // (bloom filters use `pruning_predicate` too).
diff --git a/datafusion/datasource/src/file.rs b/datafusion/datasource/src/file.rs
@@ -67,12 +67,12 @@ pub trait FileSource: Send + Sync {
     fn with_schema(&self, schema: SchemaRef) -> Arc<dyn FileSource>;
     /// Initialize new instance with projection information
     fn with_projection(&self, config: &FileScanConfig) -> Arc<dyn FileSource>;
-    /// Initialize new instance with projected statistics
-    fn with_statistics(&self, statistics: Statistics) -> Arc<dyn FileSource>;
     /// Return execution plan metrics
     fn metrics(&self) -> &ExecutionPlanMetricsSet;
     /// Return projected statistics
-    fn file_source_statistics(&self) -> Result<Statistics>;
+    fn file_source_statistics(&self, config: &FileScanConfig) -> Result<Statistics> {
+        Ok(config.file_source_projected_statistics.clone())
+    }
     /// String representation of file source such as "csv", "json", "parquet"
     fn file_type(&self) -> &str;
     /// Format FileType specific information
diff --git a/datafusion/datasource/src/file_scan_config.rs b/datafusion/datasource/src/file_scan_config.rs
diff --git a/datafusion/datasource/src/test_util.rs b/datafusion/datasource/src/test_util.rs
diff --git a/datafusion/proto/src/physical_plan/from_proto.rs b/datafusion/proto/src/physical_plan/from_proto.rs
diff --git a/datafusion/proto/src/physical_plan/to_proto.rs b/datafusion/proto/src/physical_plan/to_proto.rs
diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs

Original file line number	Diff line number	Diff line change
`@@ -89,7 +89,7 @@ pub(crate) mod test_util {`
`89`	`89`	`format.file_source(),`
`90`	`90`	`)`
`91`	`91`	`.with_file_groups(file_groups)`
`92`		`- .with_statistics(statistics)`
	`92`	`+ .with_file_source_projected_statistics(statistics)`
`93`	`93`	`.with_projection(projection)`
`94`	`94`	`.with_limit(limit)`
`95`	`95`	`.build(),`
Original file line number	Diff line number	Diff line change
`@@ -1247,7 +1247,7 @@ impl TableProvider for ListingTable {`
`1247`	`1247`	`)`
`1248`	`1248`	`.with_file_groups(partitioned_file_lists)`
`1249`	`1249`	`.with_constraints(self.constraints.clone())`
`1250`		`- .with_statistics(statistics)`
	`1250`	`+ .with_file_source_projected_statistics(statistics)`
`1251`	`1251`	`.with_projection(projection.cloned())`
`1252`	`1252`	`.with_limit(limit)`
`1253`	`1253`	`.with_output_ordering(output_ordering)`