Refactor cache APIs to support ordering information (apache#19597)

## Which issue does this PR close?

Part of apache#19433

## Rationale for this change

In preparation for ordering inference from Parquet metadata, the cache
system needs refactoring to:
1. Support storing ordering information alongside statistics
2. Simplify the `CacheAccessor` trait by removing the `Extra` associated
type and `*_with_extra` methods
3. Move validation logic into typed wrapper structs with explicit
`is_valid_for()` methods

## What changes are included in this PR?

### Simplify `CacheAccessor` trait

**Before:**
```rust
pub trait CacheAccessor<K, V>: Send + Sync {
    type Extra: Clone;
    fn get(&self, k: &K) -> Option<V>;
    fn get_with_extra(&self, k: &K, e: &Self::Extra) -> Option<V>;
    fn put(&self, key: &K, value: V) -> Option<V>;
    fn put_with_extra(&self, key: &K, value: V, e: &Self::Extra) -> Option<V>;
    // ... other methods
}
```

**After:**
```rust
pub trait CacheAccessor<K, V>: Send + Sync {
    fn get(&self, key: &K) -> Option<V>;
    fn put(&self, key: &K, value: V) -> Option<V>;
    // ... other methods (no Extra type, no *_with_extra methods)
}
```

### Introduce typed wrapper structs for cached values

Instead of passing validation metadata separately via `Extra`, embed it
in the cached value type:

- **`CachedFileMetadata`** - contains `meta: ObjectMeta`, `statistics:
Arc<Statistics>`, `ordering: Option<LexOrdering>`
- **`CachedFileList`** - contains `files: Arc<Vec<ObjectMeta>>` with
`filter_by_prefix()` helper
- **`CachedFileMetadataEntry`** - contains `meta: ObjectMeta`,
`file_metadata: Arc<dyn FileMetadata>`

Each wrapper has an `is_valid_for(&ObjectMeta)` method that checks if
the cached entry is still valid (size and last_modified match).

### New validation pattern

The typical usage pattern changes from:
```rust
// Before: validation hidden in get_with_extra
if let Some(stats) = cache.get_with_extra(&path, &object_meta) {
    // use stats
}
```

To:
```rust
// After: explicit validation
if let Some(cached) = cache.get(&path) {
    if cached.is_valid_for(&object_meta) {
        // use cached.statistics
    }
}
```

### Add ordering support

- `CachedFileMetadata` has new `ordering: Option<LexOrdering>` field
- `FileStatisticsCacheEntry` has new `has_ordering: bool` field for
introspection

## Are these changes tested?

Yes, existing cache tests pass plus new tests for ordering support.

## Are there any user-facing changes?

Breaking change to cache traits. Users with custom cache implementations
will need to:
1. Update `CacheAccessor` impl to remove `Extra` type and `*_with_extra`
methods
2. Update cached value types to the new wrappers (`CachedFileMetadata`,
etc.)
3. Update callsites to use the new validation pattern with
`is_valid_for()`

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

Author: adriangb

Cargo.lock

@@ -831,14 +831,14 @@ impl TableFunctionImpl for ListFilesCacheFunc {
                         .map(|t| t.duration_since(now).as_millis() as i64),
                 );
 
-                for meta in entry.metas.iter() {
+                for meta in entry.metas.files.iter() {
                     file_path_arr.push(meta.location.to_string());
                     file_modified_arr.push(meta.last_modified.timestamp_millis());
                     file_size_bytes_arr.push(meta.size);
                     etag_arr.push(meta.e_tag.clone());
                     version_arr.push(meta.version.clone());
                 }
-                current_offset += entry.metas.len() as i32;
+                current_offset += entry.metas.files.len() as i32;
                 offsets.push(current_offset);
             }
         }

datafusion-cli/src/functions.rs

@@ -710,31 +710,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/catalog-listing/src/table.rs

@@ -31,7 +31,9 @@ use datafusion_common::stats::Precision;
 use datafusion_common::{
     ColumnStatistics, DataFusionError, Result, ScalarValue, Statistics,
 };
-use datafusion_execution::cache::cache_manager::{FileMetadata, FileMetadataCache};
+use datafusion_execution::cache::cache_manager::{
+    CachedFileMetadataEntry, FileMetadata, FileMetadataCache,
+};
 use datafusion_functions_aggregate_common::min_max::{MaxAccumulator, MinAccumulator};
 use datafusion_physical_plan::Accumulator;
 use log::debug;
@@ -125,19 +127,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 =
@@ -163,8 +161,11 @@ impl<'a> DFParquetMetadata<'a> {
 
         if cache_metadata && let Some(file_metadata_cache) = file_metadata_cache {
             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-parquet/src/metadata.rs

@@ -19,6 +19,7 @@ use std::sync::Arc;
 
 use datafusion_common::{DataFusionError, Result, TableReference};
 use datafusion_execution::cache::TableScopedPath;
+use datafusion_execution::cache::cache_manager::CachedFileList;
 use datafusion_execution::object_store::ObjectStoreUrl;
 use datafusion_session::Session;
 
@@ -396,42 +397,35 @@ 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();
+            // Build the filter prefix (only Some if prefix was requested)
+            let filter_prefix = prefix.is_some().then(|| full_prefix.clone());
 
             let table_scoped_base_path = TableScopedPath {
                 table: table_ref.cloned(),
                 path: table_base_path.clone(),
             };
 
-            // Try cache lookup with optional prefix filter
-            let vec = if let Some(res) =
-                cache.get_with_extra(&table_scoped_base_path, &prefix_filter)
-            {
+            // Try cache lookup - get returns CachedFileList
+            let vec = if let Some(cached) = cache.get(&table_scoped_base_path) {
                 debug!("Hit list files cache");
-                res.as_ref().clone()
+                cached.files_matching_prefix(&filter_prefix)
             } else {
                 // Cache miss - always list and cache the full table
                 // This ensures we have complete data for future prefix queries
-                let vec = store
+                let mut vec = store
                     .list(Some(table_base_path))
                     .try_collect::<Vec<ObjectMeta>>()
                     .await?;
-                cache.put(&table_scoped_base_path, Arc::new(vec.clone()));
-
-                // If a prefix filter was requested, apply it to the results
-                if prefix.is_some() {
-                    let full_prefix_str = full_prefix.as_ref();
-                    vec.into_iter()
-                        .filter(|meta| {
-                            meta.location.as_ref().starts_with(full_prefix_str)
-                        })
-                        .collect()
-                } else {
-                    vec
-                }
+                vec.shrink_to_fit(); // Right-size before caching
+                let cached: CachedFileList = vec.into();
+                let result = cached.files_matching_prefix(&filter_prefix);
+                cache.put(&table_scoped_base_path, cached);
+                result
             };
-            Ok(futures::stream::iter(vec.into_iter().map(Ok)).boxed())
+            Ok(
+                futures::stream::iter(Arc::unwrap_or_clone(vec).into_iter().map(Ok))
+                    .boxed(),
+            )
         }
     }
 }
@@ -531,6 +525,7 @@ mod tests {
     use std::any::Any;
     use std::collections::HashMap;
     use std::ops::Range;
+    use std::sync::Arc;
     use tempfile::tempdir;
 
     #[test]

datafusion/datasource/src/url.rs

@@ -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"] }