feat: support partition prune api

Cargo.toml

@@ -34,6 +34,8 @@ homepage = "https://github.com/apache/hudi-rs"
 repository = "https://github.com/apache/hudi-rs"
 
 [workspace.dependencies]
+log = "^0.4"
+
 # arrow
 arrow = { version = "= 52.2.0", features = ["pyarrow"] }
 arrow-arith = { version = "= 52.2.0" }

crates/core/Cargo.toml

@@ -63,9 +63,9 @@ tokio = { workspace = true }
 # datafusion
 datafusion = { workspace = true, optional = true }
 datafusion-expr = { workspace = true, optional = true }
-datafusion-common = { workspace = true }
+datafusion-common = { workspace = true, optional = true }
 datafusion-physical-expr = { workspace = true, optional = true }
-log = "0.4.22"
+log = { workspace = true}
 
 [dev-dependencies]
 hudi-tests = { path = "../tests" }

crates/core/src/config/table.rs

@@ -130,8 +130,8 @@ impl ConfigParser for HudiTableConfig {
         match self {
             Self::DatabaseName => Some(HudiConfigValue::String("default".to_string())),
             Self::DropsPartitionFields => Some(HudiConfigValue::Boolean(false)),
-            Self::PopulatesMetaFields => Some(HudiConfigValue::Boolean(true)),
             Self::PartitionFields => Some(HudiConfigValue::List(vec!["".to_string()])),
+            Self::PopulatesMetaFields => Some(HudiConfigValue::Boolean(true)),
             _ => None,
         }
     }

crates/core/src/table/fs_view.rs

@@ -47,7 +47,7 @@
         configs: Arc<HudiConfigs>,
     ) -> Result<Self> {
         let storage = Storage::new(base_url, &storage_options)?;
         let partition_to_file_groups = Arc::new(DashMap::new());
         Ok(FileSystemView {
             configs,
             storage,
@@ -169,7 +169,7 @@
         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_mut_as_of(timestamp) {
@@ -180,7 +180,6 @@
                 }
             }
         }
-        assert!(!self.partition_to_file_groups.is_empty());
         Ok(file_slices)
     }
 

crates/core/src/table/mod.rs

@@ -442,6 +442,7 @@ impl Table {
         }
     }
 
+    #[cfg(test)]
     pub fn reset(&mut self) {
         self.file_system_view.reset()
     }