chore: update DataFusion dependencies to 53.0.0 (#1442)

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: linhr <5601366+linhr@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: lonless9 <167735979+lonless9@users.noreply.github.com>
Co-authored-by: XL Liang <xiaolong@lakesail.com>

Author: 4 people

Cargo.lock

@@ -142,33 +142,33 @@ prost-types = "0.14"
 # The `axum` version must match the one used in `tonic` (replace `RELEASE` with the release we are using):
 #   https://github.com/hyperium/tonic/blob/vRELEASE/tonic/Cargo.toml
 axum = "0.8.8"
-datafusion = { version = "52.1.0", features = ["serde", "avro", "sql"] }
-datafusion-common = { version = "52.1.0", features = ["object_store", "avro"] }
-datafusion-datasource = { version = "52.1.0" }
-datafusion-expr = { version = "52.1.0" }
-datafusion-expr-common = { version = "52.1.0" }
-datafusion-proto = { version = "52.1.0" }
-datafusion-functions = { version = "52.1.0" }
-datafusion-functions-nested = { version = "52.1.0" }
-datafusion-physical-expr = { version = "52.1.0" }
-datafusion-session = { version = "52.1.0" }
-datafusion-spark = { version = "52.1.0" }
+datafusion = { version = "53.0.0", features = ["serde", "avro", "sql"] }
+datafusion-common = { version = "53.0.0", features = ["object_store", "avro"] }
+datafusion-datasource = { version = "53.0.0" }
+datafusion-expr = { version = "53.0.0" }
+datafusion-expr-common = { version = "53.0.0" }
+datafusion-proto = { version = "53.0.0" }
+datafusion-functions = { version = "53.0.0" }
+datafusion-functions-nested = { version = "53.0.0" }
+datafusion-physical-expr = { version = "53.0.0" }
+datafusion-session = { version = "53.0.0" }
+datafusion-spark = { version = "53.0.0", features = ["core"] }
 # The `pyo3` version must match the one used in `arrow-pyarrow` (replace `RELEASE` with the release we are using):
 #   https://github.com/apache/arrow-rs/blob/RELEASE/arrow-pyarrow/Cargo.toml
-pyo3 = { version = "0.26.0", features = ["serde"] }
+pyo3 = { version = "0.28.0", features = ["serde"] }
 # Jiter has a dependency on pyo3, which needs to match the version used in Sail.
-# https://github.com/pydantic/jiter/blob/v0.11.1/Cargo.toml
-jiter = { version = "0.11.1" }
-arrow = { version = "57.1.0", features = ["chrono-tz"] }
-arrow-buffer = { version = "57.1.0" }
-arrow-schema = { version = "57.1.0", features = ["serde"] }
-arrow-flight = { version = "57.1.0" }
-arrow-pyarrow = { version = "57.1.0" }
-parquet = { version = "57.3.0" }
-serde_arrow = { version = "0.13.7", features = ["arrow-57"] }
+# https://github.com/pydantic/jiter/blob/v0.13.0/Cargo.toml
+jiter = { version = "0.13.0" }
+arrow = { version = "58.0.0", features = ["chrono-tz"] }
+arrow-buffer = { version = "58.0.0" }
+arrow-schema = { version = "58.0.0", features = ["serde"] }
+arrow-flight = { version = "58.0.0" }
+arrow-pyarrow = { version = "58.0.0" }
+parquet = { version = "58.0.0" }
+serde_arrow = { version = "0.14.0", features = ["arrow-58"] }
 # The `object_store` version must match the one used in DataFusion.
-object_store = { version = "0.12.4", features = ["aws", "gcp", "azure", "http"] }
-hdfs-native-object-store = "0.15.0"
+object_store = { version = "0.13.2", features = ["aws", "gcp", "azure", "http"] }
+hdfs-native-object-store = "0.16.0"
 
 ######
 # This is the end of the manually managed dependencies.

Cargo.toml

@@ -4,15 +4,14 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use datafusion::common::{Result as DataFusionResult, TableReference};
-use datafusion::execution::cache::cache_manager::ListFilesCache;
+use datafusion::execution::cache::cache_manager::{CachedFileList, ListFilesCache};
 use datafusion::execution::cache::{CacheAccessor, ListFilesEntry, TableScopedPath};
 use log::debug;
 use moka::sync::Cache;
-use object_store::path::Path;
 use object_store::ObjectMeta;
 
 pub struct MokaFileListingCache {
-    objects: Cache<TableScopedPath, Arc<Vec<ObjectMeta>>>,
+    objects: Cache<TableScopedPath, CachedFileList>,
     ttl: Option<Duration>,
     max_entries: Option<u64>,
 }
@@ -58,63 +57,17 @@ fn meta_heap_bytes(object_meta: &ObjectMeta) -> usize {
     size
 }
 
-impl CacheAccessor<TableScopedPath, Arc<Vec<ObjectMeta>>> for MokaFileListingCache {
-    type Extra = Option<Path>;
-
-    fn get(&self, k: &TableScopedPath) -> Option<Arc<Vec<ObjectMeta>>> {
-        self.get_with_extra(k, &None)
+impl CacheAccessor<TableScopedPath, CachedFileList> for MokaFileListingCache {
+    fn get(&self, k: &TableScopedPath) -> Option<CachedFileList> {
+        self.objects.get(k)
     }
 
-    fn get_with_extra(
-        &self,
-        k: &TableScopedPath,
-        prefix: &Self::Extra,
-    ) -> Option<Arc<Vec<ObjectMeta>>> {
-        let objects = self.objects.get(k)?;
-
-        let Some(prefix) = prefix else {
-            return Some(objects);
-        };
-
-        // Build full prefix: table_base/prefix
-        let table_base = &k.path;
-        let mut parts: Vec<_> = table_base.parts().collect();
-        parts.extend(prefix.parts());
-        let full_prefix = Path::from_iter(parts);
-        let full_prefix_str = full_prefix.as_ref();
-
-        let filtered = objects
-            .iter()
-            .filter(|meta| meta.location.as_ref().starts_with(full_prefix_str))
-            .cloned()
-            .collect::<Vec<_>>();
-
-        if filtered.is_empty() {
-            None
-        } else {
-            Some(Arc::new(filtered))
-        }
-    }
-
-    fn put(
-        &self,
-        key: &TableScopedPath,
-        value: Arc<Vec<ObjectMeta>>,
-    ) -> Option<Arc<Vec<ObjectMeta>>> {
+    fn put(&self, key: &TableScopedPath, value: CachedFileList) -> Option<CachedFileList> {
         self.objects.insert(key.clone(), value);
         None
     }
 
-    fn put_with_extra(
-        &self,
-        key: &TableScopedPath,
-        value: Arc<Vec<ObjectMeta>>,
-        _e: &Self::Extra,
-    ) -> Option<Arc<Vec<ObjectMeta>>> {
-        self.put(key, value)
-    }
-
-    fn remove(&self, k: &TableScopedPath) -> Option<Arc<Vec<ObjectMeta>>> {
+    fn remove(&self, k: &TableScopedPath) -> Option<CachedFileList> {
         self.objects.remove(k)
     }
 
@@ -157,14 +110,14 @@ impl ListFilesCache for MokaFileListingCache {
     fn list_entries(&self) -> HashMap<TableScopedPath, ListFilesEntry> {
         self.objects
             .iter()
-            .map(|(table_scoped_path, metas)| {
-                let metas = Arc::clone(&metas);
+            .map(|(table_scoped_path, cached)| {
+                let metas = Arc::clone(&cached.files);
                 let size_bytes = (metas.capacity() * size_of::<ObjectMeta>())
                     + metas.iter().map(meta_heap_bytes).sum::<usize>();
                 (
                     (*table_scoped_path).clone(),
                     ListFilesEntry {
-                        metas,
+                        metas: cached.clone(),
                         size_bytes,
                         // moka handles expiration; we don't have per-entry expiration time
                         expires: None,
@@ -200,7 +153,7 @@ mod tests {
     #[test]
     fn test_file_listing_cache() {
         let meta = ObjectMeta {
-            location: Path::from("test"),
+            location: object_store::path::Path::from("test"),
             last_modified: DateTime::parse_from_rfc3339("2022-09-27T22:36:00+02:00")
                 .unwrap()
                 .into(),
@@ -216,9 +169,9 @@ mod tests {
         };
         assert!(cache.get(&key).is_none());
 
-        cache.put(&key, vec![meta.clone()].into());
+        cache.put(&key, CachedFileList::new(vec![meta.clone()]));
         assert_eq!(
-            cache.get(&key).unwrap().first().unwrap().clone(),
+            cache.get(&key).unwrap().files.first().unwrap().clone(),
             meta.clone()
         );
     }

crates/sail-cache/src/file_listing_cache.rs

@@ -1,20 +1,18 @@
 use std::collections::HashMap;
-use std::sync::Arc;
 use std::time::Duration;
 
 use datafusion::execution::cache::cache_manager::{
-    FileMetadata, FileMetadataCache, FileMetadataCacheEntry,
+    CachedFileMetadataEntry, FileMetadataCache, FileMetadataCacheEntry,
 };
 use datafusion::execution::cache::CacheAccessor;
 use log::debug;
 use moka::policy::EvictionPolicy;
 use moka::sync::Cache;
 use object_store::path::Path;
-use object_store::ObjectMeta;
 
 pub struct MokaFileMetadataCache {
     size_limit: Option<u64>,
-    metadata: Cache<Path, (ObjectMeta, Arc<dyn FileMetadata>)>,
+    metadata: Cache<Path, CachedFileMetadataEntry>,
 }
 
 impl MokaFileMetadataCache {
@@ -34,11 +32,9 @@ impl MokaFileMetadataCache {
                 Self::NAME
             );
             builder = builder
-                .weigher(
-                    |_key: &Path, (_, meta): &(ObjectMeta, Arc<dyn FileMetadata>)| -> u32 {
-                        meta.memory_size() as u32
-                    },
-                )
+                .weigher(|_key: &Path, entry: &CachedFileMetadataEntry| -> u32 {
+                    entry.file_metadata.memory_size() as u32
+                })
                 .max_capacity(size_limit);
         } else {
             debug!("No size limit set for {}", Self::NAME);
@@ -65,67 +61,38 @@ impl FileMetadataCache for MokaFileMetadataCache {
     fn list_entries(&self) -> HashMap<Path, FileMetadataCacheEntry> {
         self.metadata
             .iter()
-            .map(|(path, (object_meta, meta))| {
+            .map(|(path, entry)| {
                 (
                     path.as_ref().clone(),
                     FileMetadataCacheEntry {
-                        object_meta,
-                        size_bytes: meta.memory_size(),
+                        object_meta: entry.meta.clone(),
+                        size_bytes: entry.file_metadata.memory_size(),
                         // TODO: get hits from the cache
                         hits: 0,
-                        extra: meta.extra_info(),
+                        extra: entry.file_metadata.extra_info(),
                     },
                 )
             })
             .collect()
     }
 }
 
-impl CacheAccessor<ObjectMeta, Arc<dyn FileMetadata>> for MokaFileMetadataCache {
-    type Extra = ObjectMeta;
-
-    fn get(&self, k: &ObjectMeta) -> Option<Arc<dyn FileMetadata>> {
-        self.metadata
-            .get(&k.location)
-            .and_then(|(extra, metadata)| {
-                if extra.size == k.size && extra.last_modified == k.last_modified {
-                    Some(Arc::clone(&metadata))
-                } else {
-                    None
-                }
-            })
-    }
-
-    fn get_with_extra(&self, k: &ObjectMeta, _e: &Self::Extra) -> Option<Arc<dyn FileMetadata>> {
-        self.get(k)
+impl CacheAccessor<Path, CachedFileMetadataEntry> for MokaFileMetadataCache {
+    fn get(&self, k: &Path) -> Option<CachedFileMetadataEntry> {
+        self.metadata.get(k)
     }
 
-    fn put(&self, key: &ObjectMeta, value: Arc<dyn FileMetadata>) -> Option<Arc<dyn FileMetadata>> {
-        self.metadata
-            .insert(key.location.clone(), (key.clone(), value));
+    fn put(&self, key: &Path, value: CachedFileMetadataEntry) -> Option<CachedFileMetadataEntry> {
+        self.metadata.insert(key.clone(), value);
         None
     }
 
-    fn put_with_extra(
-        &self,
-        key: &ObjectMeta,
-        value: Arc<dyn FileMetadata>,
-        _e: &Self::Extra,
-    ) -> Option<Arc<dyn FileMetadata>> {
-        self.put(key, value)
+    fn remove(&self, k: &Path) -> Option<CachedFileMetadataEntry> {
+        self.metadata.remove(k)
     }
 
-    fn remove(&self, k: &ObjectMeta) -> Option<Arc<dyn FileMetadata>> {
-        self.metadata
-            .remove(&k.location)
-            .map(|(_, metadata)| metadata)
-    }
-
-    fn contains_key(&self, k: &ObjectMeta) -> bool {
-        self.metadata
-            .get(&k.location)
-            .map(|(extra, _)| extra.size == k.size && extra.last_modified == k.last_modified)
-            .unwrap_or(false)
+    fn contains_key(&self, k: &Path) -> bool {
+        self.metadata.contains_key(k)
     }
 
     fn len(&self) -> usize {
@@ -184,47 +151,36 @@ mod tests {
             version: None,
         };
 
-        let metadata: Arc<dyn FileMetadata> = Arc::new(TestFileMetadata {
+        let file_metadata: Arc<dyn FileMetadata> = Arc::new(TestFileMetadata {
             metadata: "retrieved_metadata".to_owned(),
         });
+        let entry = CachedFileMetadataEntry::new(object_meta.clone(), Arc::clone(&file_metadata));
 
         let cache = MokaFileMetadataCache::new(None, None);
-        assert!(cache.get(&object_meta).is_none());
+        assert!(cache.get(&object_meta.location).is_none());
 
         // put
-        cache.put(&object_meta, metadata);
+        cache.put(&object_meta.location, entry.clone());
 
         // get and contains of a valid entry
-        assert!(cache.contains_key(&object_meta));
-        let value = cache.get(&object_meta);
+        assert!(cache.contains_key(&object_meta.location));
+        let value = cache.get(&object_meta.location);
         assert!(value.is_some());
-        let test_file_metadata = Arc::downcast::<TestFileMetadata>(value.unwrap());
-        assert!(test_file_metadata.is_ok());
-        assert_eq!(test_file_metadata.unwrap().metadata, "retrieved_metadata");
-
-        // file size changed
-        let mut object_meta2 = object_meta.clone();
-        object_meta2.size = 2048;
-        assert!(cache.get(&object_meta2).is_none());
-        assert!(!cache.contains_key(&object_meta2));
-
-        // file last_modified changed
-        let mut object_meta2 = object_meta.clone();
-        object_meta2.last_modified = DateTime::parse_from_rfc3339("2025-07-29T13:13:13+00:00")
-            .unwrap()
-            .into();
-        assert!(cache.get(&object_meta2).is_none());
-        assert!(!cache.contains_key(&object_meta2));
+        let cached_entry = value.unwrap();
+        assert!(cached_entry.is_valid_for(&object_meta));
+        let test_meta = Arc::downcast::<TestFileMetadata>(cached_entry.file_metadata);
+        assert!(test_meta.is_ok());
+        assert_eq!(test_meta.unwrap().metadata, "retrieved_metadata");
 
         // different file
         let mut object_meta2 = object_meta.clone();
         object_meta2.location = Path::from("test2");
-        assert!(cache.get(&object_meta2).is_none());
-        assert!(!cache.contains_key(&object_meta2));
+        assert!(cache.get(&object_meta2.location).is_none());
+        assert!(!cache.contains_key(&object_meta2.location));
 
         // remove
-        cache.remove(&object_meta);
-        assert!(cache.get(&object_meta).is_none());
-        assert!(!cache.contains_key(&object_meta));
+        cache.remove(&object_meta.location);
+        assert!(cache.get(&object_meta.location).is_none());
+        assert!(!cache.contains_key(&object_meta.location));
     }
 }