address comments

Author: xudong963

datafusion/datasource/src/file_groups.rs

@@ -420,9 +420,14 @@ impl FileGroup {
         self.files.push(file);
     }
 
-    /// Get the statistics for this group
-    pub fn statistics_ref(&self) -> Option<&Statistics> {
-        self.statistics.as_deref()
+    /// Get the specific file statistics for the given index
+    /// If the index is None, return the `FileGroup` statistics
+    pub fn file_statistics(&self, index: Option<usize>) -> Option<&Statistics> {
+        if let Some(index) = index {
+            self.files.get(index).and_then(|f| f.statistics.as_deref())
+        } else {
+            self.statistics.as_deref()
+        }
     }
 
     /// Partition the list of files into `n` groups

datafusion/datasource/src/source.rs

@@ -195,7 +195,7 @@ impl ExecutionPlan for DataSourceExec {
                 self.data_source.as_any().downcast_ref::<FileScanConfig>()
             {
                 if let Some(file_group) = file_config.file_groups.get(partition) {
-                    if let Some(stat) = file_group.statistics_ref() {
+                    if let Some(stat) = file_group.file_statistics(None) {
                         statistics = stat.clone();
                     }
                 }

datafusion/datasource/src/statistics.rs

@@ -476,7 +476,7 @@ pub fn compute_all_files_statistics(
     // Then summary statistics across all file groups
     let file_groups_statistics = file_groups_with_stats
         .iter()
-        .filter_map(|file_group| file_group.statistics_ref());
+        .filter_map(|file_group| file_group.file_statistics(None));
 
     let mut statistics =
         Statistics::try_merge_iter(file_groups_statistics, &table_schema)?;

datafusion/physical-plan/src/execution_plan.rs

@@ -435,25 +435,18 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
     /// (the default), not an error.
     /// If `partition` is `None`, it returns statistics for the entire plan.
     fn partition_statistics(&self, partition: Option<usize>) -> Result<Statistics> {
-        match partition {
-            Some(idx) => {
-                // Validate partition index
-                let partition_count = self.properties().partitioning.partition_count();
-                if idx >= partition_count {
-                    return internal_err!(
-                        "Invalid partition index: {}, the partition count is {}",
-                        idx,
-                        partition_count
-                    );
-                }
-                // Default implementation: return unknown statistics for the specific partition
-                Ok(Statistics::new_unknown(&self.schema()))
-            }
-            None => {
-                // Return unknown statistics for the entire plan
-                Ok(Statistics::new_unknown(&self.schema()))
+        if let Some(idx) = partition {
+            // Validate partition index
+            let partition_count = self.properties().partitioning.partition_count();
+            if idx >= partition_count {
+                return internal_err!(
+                    "Invalid partition index: {}, the partition count is {}",
+                    idx,
+                    partition_count
+                );
             }
         }
+        Ok(Statistics::new_unknown(&self.schema()))
     }
 
     /// Returns `true` if a limit can be safely pushed down through this