fix ci, add files_matching_prefix

Author: adriangb

datafusion-cli/src/functions.rs

@@ -811,14 +811,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.cached_file_list.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.cached_file_list.files.len() as i32;
                 offsets.push(current_offset);
             }
         }

datafusion/datasource/src/url.rs

@@ -15,6 +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;
@@ -369,7 +371,7 @@ async fn list_with_cache<'b>(
             // Try cache lookup
             let vec = if let Some(cached) = cache.get(table_base_path) {
                 debug!("Hit list files cache");
-                cached.filter_by_prefix(&filter_prefix)
+                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
@@ -378,11 +380,14 @@ async fn list_with_cache<'b>(
                     .try_collect::<Vec<ObjectMeta>>()
                     .await?;
                 let cached = CachedFileList::new(vec);
-                let result = cached.filter_by_prefix(&filter_prefix);
+                let result = cached.files_matching_prefix(&filter_prefix);
                 cache.put(table_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(),
+            )
         }
     }
 }

datafusion/execution/src/cache/cache_manager.rs

@@ -128,7 +128,7 @@ impl CachedFileList {
     }
 
     /// Filter the files by prefix.
-    pub fn filter_by_prefix(&self, prefix: &Option<Path>) -> Vec<ObjectMeta> {
+    fn filter_by_prefix(&self, prefix: &Option<Path>) -> Vec<ObjectMeta> {
         match prefix {
             Some(prefix) => self
                 .files
@@ -139,6 +139,17 @@ impl CachedFileList {
             None => self.files.as_ref().clone(),
         }
     }
+
+    /// Returns files matching the given prefix.
+    ///
+    /// When prefix is `None`, returns a clone of the `Arc` (no data copy).
+    /// When filtering is needed, returns a new `Arc` with filtered results (clones each matching [`ObjectMeta`]).
+    pub fn files_matching_prefix(&self, prefix: &Option<Path>) -> Arc<Vec<ObjectMeta>> {
+        match prefix {
+            None => Arc::clone(&self.files),
+            Some(p) => Arc::new(self.filter_by_prefix(&Some(p.clone()))),
+        }
+    }
 }
 
 /// Cache for storing the [`ObjectMeta`]s that result from listing a path