enhance: Remove redundant statistics from FileScanConfig

Signed-off-by: Alan Tang <[email protected]>

Author: Standing-Man

datafusion/datasource/src/file_scan_config.rs

@@ -148,9 +148,6 @@ pub struct FileScanConfig {
     pub file_groups: Vec<Vec<PartitionedFile>>,
     /// Table constraints
     pub constraints: Constraints,
-    /// Estimated overall statistics of the files, taking `filters` into account.
-    /// Defaults to [`Statistics::new_unknown`].
-    pub statistics: Statistics,
     /// Columns on which to project the data. Indexes that are higher than the
     /// number of columns of `file_schema` refer to `table_partition_cols`.
     pub projection: Option<Vec<usize>>,
@@ -319,7 +316,6 @@ impl FileScanConfig {
             file_schema,
             file_groups: vec![],
             constraints: Constraints::empty(),
-            statistics,
             projection: None,
             limit: None,
             table_partition_cols: vec![],
@@ -328,14 +324,15 @@ impl FileScanConfig {
             new_lines_in_values: false,
             file_source: Arc::clone(&file_source),
         };
+        config = config.with_statistics(statistics);
 
-        config = config.with_source(Arc::clone(&file_source));
         config
     }
 
     /// Set the file source
     pub fn with_source(mut self, file_source: Arc<dyn FileSource>) -> Self {
-        self.file_source = file_source.with_statistics(self.statistics.clone());
+        self.file_source =
+            file_source.with_statistics(Statistics::new_unknown(&self.file_schema));
         self
     }
 
@@ -347,7 +344,6 @@ impl FileScanConfig {
 
     /// Set the statistics of the files
     pub fn with_statistics(mut self, statistics: Statistics) -> Self {
-        self.statistics = statistics.clone();
         self.file_source = self.file_source.with_statistics(statistics);
         self
     }
@@ -362,10 +358,7 @@ impl FileScanConfig {
     }
 
     fn projected_stats(&self) -> Statistics {
-        let statistics = self
-            .file_source
-            .statistics()
-            .unwrap_or(self.statistics.clone());
+        let statistics = self.file_source.statistics().unwrap();
 
         let table_cols_stats = self
             .projection_indices()
@@ -498,7 +491,7 @@ impl FileScanConfig {
             return (
                 Arc::clone(&self.file_schema),
                 self.constraints.clone(),
-                self.statistics.clone(),
+                self.file_source.statistics().unwrap().clone(),
                 self.output_ordering.clone(),
             );
         }
@@ -641,7 +634,11 @@ impl Debug for FileScanConfig {
         write!(f, "FileScanConfig {{")?;
         write!(f, "object_store_url={:?}, ", self.object_store_url)?;
 
-        write!(f, "statistics={:?}, ", self.statistics)?;
+        write!(
+            f,
+            "statistics={:?}, ",
+            self.file_source.statistics().unwrap()
+        )?;
 
         DisplayAs::fmt_as(self, DisplayFormatType::Verbose, f)?;
         write!(f, "}}")

datafusion/proto/src/physical_plan/to_proto.rs

@@ -507,7 +507,7 @@ pub fn serialize_file_scan_config(
 
     Ok(protobuf::FileScanExecConf {
         file_groups,
-        statistics: Some((&conf.statistics).into()),
+        statistics: Some((&conf.file_source.statistics().unwrap()).into()),
         limit: conf.limit.map(|l| protobuf::ScanLimit { limit: l as u32 }),
         projection: conf
             .projection

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -741,22 +741,22 @@ fn roundtrip_parquet_exec_with_pruning_predicate() -> Result<()> {
     let source = Arc::new(
         ParquetSource::new(options).with_predicate(Arc::clone(&file_schema), predicate),
     );
+    let statistics = Statistics {
+        num_rows: Precision::Inexact(100),
+        total_byte_size: Precision::Inexact(1024),
+        column_statistics: Statistics::unknown_column(&Arc::new(Schema::new(vec![
+            Field::new("col", DataType::Utf8, false),
+        ]))),
+    };
 
-    let scan_config = FileScanConfig {
+    let mut scan_config = FileScanConfig {
         object_store_url: ObjectStoreUrl::local_filesystem(),
         file_schema,
         file_groups: vec![vec![PartitionedFile::new(
             "/path/to/file.parquet".to_string(),
             1024,
         )]],
         constraints: Constraints::empty(),
-        statistics: Statistics {
-            num_rows: Precision::Inexact(100),
-            total_byte_size: Precision::Inexact(1024),
-            column_statistics: Statistics::unknown_column(&Arc::new(Schema::new(vec![
-                Field::new("col", DataType::Utf8, false),
-            ]))),
-        },
         projection: None,
         limit: None,
         table_partition_cols: vec![],
@@ -765,6 +765,7 @@ fn roundtrip_parquet_exec_with_pruning_predicate() -> Result<()> {
         new_lines_in_values: false,
         file_source: source,
     };
+    scan_config = scan_config.with_statistics(statistics);
 
     roundtrip_test(scan_config.build())
 }
@@ -806,21 +807,23 @@ fn roundtrip_parquet_exec_with_custom_predicate_expr() -> Result<()> {
             .with_predicate(Arc::clone(&file_schema), custom_predicate_expr),
     );
 
-    let scan_config = FileScanConfig {
+    let statistics = Statistics {
+        num_rows: Precision::Inexact(100),
+        total_byte_size: Precision::Inexact(1024),
+        column_statistics: Statistics::unknown_column(&Arc::new(Schema::new(vec![
+            Field::new("col", DataType::Utf8, false),
+        ]))),
+    };
+
+    let mut scan_config = FileScanConfig {
         object_store_url: ObjectStoreUrl::local_filesystem(),
         file_schema,
         file_groups: vec![vec![PartitionedFile::new(
             "/path/to/file.parquet".to_string(),
             1024,
         )]],
         constraints: Constraints::empty(),
-        statistics: Statistics {
-            num_rows: Precision::Inexact(100),
-            total_byte_size: Precision::Inexact(1024),
-            column_statistics: Statistics::unknown_column(&Arc::new(Schema::new(vec![
-                Field::new("col", DataType::Utf8, false),
-            ]))),
-        },
+
         projection: None,
         limit: None,
         table_partition_cols: vec![],
@@ -829,6 +832,7 @@ fn roundtrip_parquet_exec_with_custom_predicate_expr() -> Result<()> {
         new_lines_in_values: false,
         file_source: source,
     };
+    scan_config = scan_config.with_statistics(statistics);
 
     #[derive(Debug, Clone, Eq)]
     struct CustomPredicateExpr {
@@ -1616,7 +1620,6 @@ async fn roundtrip_projection_source() -> Result<()> {
             1024,
         )]],
         constraints: Constraints::empty(),
-        statistics,
         file_schema: schema.clone(),
         projection: Some(vec![0, 1, 2]),
         limit: None,