Refactor cache APIs to support ordering information

Refactor the cache system to support storing both statistics and ordering
information together, in preparation for ordering inference from Parquet
metadata.

Changes to cache_manager.rs:
- Add `CachedFileMetadata` struct with `meta`, `statistics`, and `ordering` fields
- Refactor `FileStatisticsCache` trait to use `CachedFileMetadata` and Path keys
- Add `has_ordering` field to `FileStatisticsCacheEntry`
- Add `CachedFileList` for list files cache
- Refactor `FileMetadataCache` trait to use `CachedFileMetadataEntry` and Path keys

Changes to cache implementations:
- Update `DefaultFileStatisticsCache` to use new trait methods
- Update `DefaultFilesMetadataCache` to use new trait methods
- Simplify list files cache implementation

Changes to callsites:
- Update `ListingTable::do_collect_statistics` to use new cache API
- Update `DFParquetMetadata::fetch_metadata` to use new cache API
- Update `ListingTableUrl` to use new cache API

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Author: adriangb

datafusion/catalog-listing/src/table.rs

@@ -705,31 +705,42 @@ impl ListingTable {
         store: &Arc<dyn ObjectStore>,
         part_file: &PartitionedFile,
     ) -> datafusion_common::Result<Arc<Statistics>> {
-        match self
+        use datafusion_execution::cache::cache_manager::CachedFileMetadata;
+
+        // Check cache first
+        if let Some(cached) = self
             .collected_statistics
-            .get_with_extra(&part_file.object_meta.location, &part_file.object_meta)
+            .get(&part_file.object_meta.location)
         {
-            Some(statistics) => Ok(statistics),
-            None => {
-                let statistics = self
-                    .options
-                    .format
-                    .infer_stats(
-                        ctx,
-                        store,
-                        Arc::clone(&self.file_schema),
-                        &part_file.object_meta,
-                    )
-                    .await?;
-                let statistics = Arc::new(statistics);
-                self.collected_statistics.put_with_extra(
-                    &part_file.object_meta.location,
-                    Arc::clone(&statistics),
-                    &part_file.object_meta,
-                );
-                Ok(statistics)
+            // Validate that cached entry is still valid
+            if cached.is_valid_for(&part_file.object_meta) {
+                return Ok(cached.statistics);
             }
         }
+
+        // Cache miss or invalid - infer statistics
+        let statistics = self
+            .options
+            .format
+            .infer_stats(
+                ctx,
+                store,
+                Arc::clone(&self.file_schema),
+                &part_file.object_meta,
+            )
+            .await?;
+        let statistics = Arc::new(statistics);
+
+        // Store in cache
+        self.collected_statistics.put(
+            &part_file.object_meta.location,
+            CachedFileMetadata::new(
+                part_file.object_meta.clone(),
+                Arc::clone(&statistics),
+                None, // No ordering information in this PR
+            ),
+        );
+        Ok(statistics)
     }
 }
 

datafusion/datasource-parquet/src/metadata.rs

@@ -125,19 +125,15 @@ impl<'a> DFParquetMetadata<'a> {
             !cfg!(feature = "parquet_encryption") || decryption_properties.is_none();
 
         if cache_metadata
-            && let Some(parquet_metadata) = file_metadata_cache
-                .as_ref()
-                .and_then(|file_metadata_cache| file_metadata_cache.get(object_meta))
-                .and_then(|file_metadata| {
-                    file_metadata
-                        .as_any()
-                        .downcast_ref::<CachedParquetMetaData>()
-                        .map(|cached_parquet_metadata| {
-                            Arc::clone(cached_parquet_metadata.parquet_metadata())
-                        })
-                })
+            && let Some(file_metadata_cache) = file_metadata_cache.as_ref()
+            && let Some(cached) = file_metadata_cache.get(&object_meta.location)
+            && cached.is_valid_for(object_meta)
+            && let Some(cached_parquet) = cached
+                .file_metadata
+                .as_any()
+                .downcast_ref::<CachedParquetMetaData>()
         {
-            return Ok(parquet_metadata);
+            return Ok(Arc::clone(cached_parquet.parquet_metadata()));
         }
 
         let mut reader =
@@ -162,9 +158,13 @@ impl<'a> DFParquetMetadata<'a> {
         );
 
         if cache_metadata && let Some(file_metadata_cache) = file_metadata_cache {
+            use datafusion_execution::cache::cache_manager::CachedFileMetadataEntry;
             file_metadata_cache.put(
-                object_meta,
-                Arc::new(CachedParquetMetaData::new(Arc::clone(&metadata))),
+                &object_meta.location,
+                CachedFileMetadataEntry::new(
+                    (*object_meta).clone(),
+                    Arc::new(CachedParquetMetaData::new(Arc::clone(&metadata))),
+                ),
             );
         }
 

datafusion/datasource/src/url.rs

@@ -15,9 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::sync::Arc;
-
 use datafusion_common::{DataFusionError, Result};
+use datafusion_execution::cache::cache_manager::CachedFileList;
 use datafusion_execution::object_store::ObjectStoreUrl;
 use datafusion_session::Session;
 
@@ -364,23 +363,31 @@ async fn list_with_cache<'b>(
             .map(|res| res.map_err(|e| DataFusionError::ObjectStore(Box::new(e))))
             .boxed()),
         Some(cache) => {
-            // Convert prefix to Option<Path> for cache lookup
-            let prefix_filter = prefix.cloned();
-
-            // Try cache lookup with optional prefix filter
-            let vec = if let Some(res) =
-                cache.get_with_extra(table_base_path, &prefix_filter)
-            {
+            // Try cache lookup
+            let vec = if let Some(cached) = cache.get(table_base_path) {
                 debug!("Hit list files cache");
-                res.as_ref().clone()
+                // Cache hit - apply prefix filter if needed
+                if prefix.is_some() {
+                    let full_prefix_str = full_prefix.as_ref();
+                    cached
+                        .files
+                        .iter()
+                        .filter(|meta| {
+                            meta.location.as_ref().starts_with(full_prefix_str)
+                        })
+                        .cloned()
+                        .collect()
+                } else {
+                    cached.files.as_ref().clone()
+                }
             } else {
                 // Cache miss - always list and cache the full table
                 // This ensures we have complete data for future prefix queries
                 let vec = store
                     .list(Some(table_base_path))
                     .try_collect::<Vec<ObjectMeta>>()
                     .await?;
-                cache.put(table_base_path, Arc::new(vec.clone()));
+                cache.put(table_base_path, CachedFileList::new(vec.clone()));
 
                 // If a prefix filter was requested, apply it to the results
                 if prefix.is_some() {
@@ -494,6 +501,7 @@ mod tests {
     use std::any::Any;
     use std::collections::HashMap;
     use std::ops::Range;
+    use std::sync::Arc;
     use tempfile::tempdir;
 
     #[test]

datafusion/execution/Cargo.toml

@@ -55,6 +55,7 @@ chrono = { workspace = true }
 dashmap = { workspace = true }
 datafusion-common = { workspace = true, default-features = false }
 datafusion-expr = { workspace = true, default-features = false }
+datafusion-physical-expr-common = { workspace = true, default-features = false }
 futures = { workspace = true }
 log = { workspace = true }
 object_store = { workspace = true, features = ["fs"] }
@@ -66,3 +67,4 @@ url = { workspace = true }
 
 [dev-dependencies]
 insta = { workspace = true }
+datafusion-physical-expr = { workspace = true, default-features = false }