fix load slice stats

crates/core/src/table/fs_view.rs

@@ -167,15 +167,19 @@ impl FileSystemView {
                 self.partition_to_file_groups.insert(pair.0, pair.1);
             });
         }
-        for fgs in self.partition_to_file_groups.iter() {
-            let fgs_ref = fgs.value();
+        for mut fgs in self.partition_to_file_groups.iter_mut() {
+            let fgs_ref = fgs.value_mut();
             for fg in fgs_ref {
                 if excluding_file_groups.contains(fg) {
                     continue;
                 }
-                if let Some(fsl) = fg.get_file_slice_as_of(timestamp) {
+                if let Some(fsl) = fg.get_file_slice_mut_as_of(timestamp) {
                     // TODO: pass ref instead of copying
-                    file_slices.push(fsl.clone());
+                    fsl.load_stats(&self.storage)
+                        .await
+                        .expect("Fail to load file slice stats.");
+                    let immut_fsl: &FileSlice = fsl;
+                    file_slices.push(immut_fsl.clone());
                 }
             }
         }
@@ -299,6 +303,9 @@ mod tests {
             .iter()
             .map(|fsl| fsl.file_group_id())
             .collect::<Vec<_>>();
-        assert_eq!(fg_ids, vec!["a079bdb3-731c-4894-b855-abfcd6921007-0"])
+        assert_eq!(fg_ids, vec!["a079bdb3-731c-4894-b855-abfcd6921007-0"]);
+        for fsl in file_slices.iter() {
+            assert_eq!(fsl.base_file.stats.as_ref().unwrap().num_records, 4);
+        }
     }
 }

crates/core/src/table/mod.rs

@@ -246,10 +246,6 @@ impl Table {
 
     async fn get_file_slices_as_of(&self, timestamp: &str) -> Result<Vec<FileSlice>> {
         let excludes = self.timeline.get_replaced_file_groups().await?;
-        self.file_system_view
-            .load_file_slices_stats_as_of(timestamp, &excludes)
-            .await
-            .context("Fail to load file slice stats.")?;
         self.file_system_view
             .get_file_slices_as_of(
                 timestamp,