Function list_with_cache

datafusion/datasource/src/url.rs:379–433

Lists files with cache support, using prefix-aware lookups. # Arguments `ctx` - The session context `store` - The object store to list from `table_base_path` - The table's base path (the stable cache key) `prefix` - Optional prefix relative to table base for filtering results # Cache Behavior: The cache key is always `table_base_path`. When a prefix-filtered listing is requested via `prefix`, th

(
    ctx: &'b dyn Session,
    store: &'b dyn ObjectStore,
    table_ref: Option<&TableReference>,
    table_base_path: &Path,
    prefix: Option<&Path>,
)

/// On cache miss, the full table is always listed and cached, ensuring
/// subsequent prefix queries can be served from cache.
async fn list_with_cache<'b>(
    ctx: &'b dyn Session,
    store: &'b dyn ObjectStore,
    table_ref: Option<&TableReference>,
    table_base_path: &Path,
    prefix: Option<&Path>,
) -> Result<BoxStream<'b, Result<ObjectMeta>>> {
    // Build the full listing path (table_base + prefix)
    let full_prefix = match prefix {
        Some(p) => {
            let mut parts: Vec<_> = table_base_path.parts().collect();
            parts.extend(p.parts());
            Path::from_iter(parts)
        }
        None => table_base_path.clone(),
    };

    match ctx.runtime_env().cache_manager.get_list_files_cache() {
        None => Ok(store
            .list(Some(&full_prefix))
            .map(|res| res.map_err(|e| DataFusionError::ObjectStore(Box::new(e))))
            .boxed()),
        Some(cache) => {
            // 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 - get returns CachedFileList
            let vec = if let Some(cached) = cache.get(&table_scoped_base_path) {
                debug!("Hit list files cache");
                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 mut vec = store
                    .list(Some(table_base_path))
                    .try_collect::<Vec<ObjectMeta>>()
                    .await?;
                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(Arc::unwrap_or_clone(vec).into_iter().map(Ok))
                    .boxed(),
            )
        }
    }
}

/// Creates a file URL from a potentially relative filesystem path
#[cfg(not(target_arch = "wasm32"))]