Make file_source_statistics infallible

Author: friendlymatthew

datafusion/core/tests/physical_optimizer/test_utils.rs

@@ -135,11 +135,7 @@ pub(crate) fn parquet_exec_with_stats(file_size: u64) -> Arc<DataSourceExec> {
     .build();
 
     assert_eq!(
-        config
-            .file_source
-            .file_source_statistics(&config)
-            .unwrap()
-            .num_rows,
+        config.file_source.file_source_statistics(&config).num_rows,
         Precision::Inexact(10000)
     );
     DataSourceExec::from_data_source(config)

datafusion/datasource-parquet/src/source.rs

@@ -584,10 +584,7 @@ impl FileSource for ParquetSource {
         &self.metrics
     }
 
-    fn file_source_statistics(
-        &self,
-        config: &FileScanConfig,
-    ) -> datafusion_common::Result<Statistics> {
+    fn file_source_statistics(&self, config: &FileScanConfig) -> Statistics {
         let statistics = config.file_source_projected_statistics.clone();
 
         // When filters are pushed down, we have no way of knowing the exact statistics.
@@ -596,10 +593,10 @@ impl FileSource for ParquetSource {
         // Because filter pushdown may happen dynamically as long as there is a predicate
         // if we have *any* predicate applied, we can't guarantee the statistics are exact.
         if self.predicate().is_some() {
-            Ok(statistics.to_inexact())
-        } else {
-            Ok(statistics)
+            return statistics.to_inexact();
         }
+
+        statistics
     }
 
     fn file_type(&self) -> &str {

datafusion/datasource/src/file.rs

@@ -63,8 +63,8 @@ pub trait FileSource: Send + Sync {
     /// Return execution plan metrics
     fn metrics(&self) -> &ExecutionPlanMetricsSet;
     /// Return projected statistics
-    fn file_source_statistics(&self, config: &FileScanConfig) -> Result<Statistics> {
-        Ok(config.file_source_projected_statistics.clone())
+    fn file_source_statistics(&self, config: &FileScanConfig) -> Statistics {
+        config.file_source_projected_statistics.clone()
     }
     /// String representation of file source such as "csv", "json", "parquet"
     fn file_type(&self) -> &str;

datafusion/datasource/src/file_scan_config.rs

@@ -691,7 +691,7 @@ impl FileScanConfig {
     }
 
     pub fn projected_stats(&self) -> Statistics {
-        let statistics = self.file_source.file_source_statistics(self).unwrap();
+        let statistics = self.file_source.file_source_statistics(self);
 
         let table_cols_stats = self
             .projection_indices()
@@ -758,10 +758,7 @@ impl FileScanConfig {
             return (
                 Arc::clone(&self.file_schema),
                 self.constraints.clone(),
-                self.file_source
-                    .file_source_statistics(self)
-                    .unwrap()
-                    .clone(),
+                self.file_source.file_source_statistics(self).clone(),
                 self.output_ordering.clone(),
             );
         }
@@ -993,7 +990,7 @@ impl Debug for FileScanConfig {
         write!(
             f,
             "statistics={:?}, ",
-            self.file_source.file_source_statistics(self).unwrap()
+            self.file_source.file_source_statistics(self)
         )?;
 
         DisplayAs::fmt_as(self, DisplayFormatType::Verbose, f)?;
@@ -1584,7 +1581,7 @@ mod tests {
             to_partition_cols(partition_cols.clone()),
         );
 
-        let source_statistics = conf.file_source.file_source_statistics(&conf).unwrap();
+        let source_statistics = conf.file_source.file_source_statistics(&conf);
         let conf_stats = conf.data_source_statistics().unwrap();
 
         // projection should be reflected in the file source statistics
@@ -2218,34 +2215,27 @@ mod tests {
 
         // Verify statistics are set to unknown
         assert_eq!(
-            config
-                .file_source
-                .file_source_statistics(&config)
-                .unwrap()
-                .num_rows,
+            config.file_source.file_source_statistics(&config).num_rows,
             Precision::Absent
         );
         assert_eq!(
             config
                 .file_source
                 .file_source_statistics(&config)
-                .unwrap()
                 .total_byte_size,
             Precision::Absent
         );
         assert_eq!(
             config
                 .file_source
                 .file_source_statistics(&config)
-                .unwrap()
                 .column_statistics
                 .len(),
             file_schema.fields().len()
         );
         for stat in config
             .file_source
             .file_source_statistics(&config)
-            .unwrap()
             .column_statistics
         {
             assert_eq!(stat.distinct_count, Precision::Absent);

datafusion/proto/src/physical_plan/to_proto.rs

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