From 9d1f6ed07b4cfcef4799a5087c12b2b040fc3f57 Mon Sep 17 00:00:00 2001 From: Michael Taranov Date: Tue, 22 Sep 2026 10:32:49 +0300 Subject: [PATCH 1/3] perf: cache Iceberg FileIO per executor instead of building one per task Every native Iceberg scan or write task built its own FileIO, and since iceberg-rust creates the storage client lazily per instance, every task also opened and tore down its own client: an S3 client, or with hdfs-native a NameNode session. Spark's Hadoop layer shares one client per executor JVM. load_file_io now serves clones from a per-executor cache and only builds on a miss. The key is everything that shapes the client: access mode, catalog name, the full reference path, and the whole catalog property bag. The path is in the key because the S3 access bridge is constructed with the bucket and path of the reference location, so a bridge built for one table must not serve another. memory:/// is never cached, the map clears past 64 entries, and release_runtime clears it with the runtime. Closes #6105 Co-Authored-By: Claude Fable 5.1 --- native/core/src/execution/jni_api.rs | 1 + .../src/execution/operators/iceberg_common.rs | 122 +++++++++++++++++- native/core/src/execution/operators/mod.rs | 1 + 3 files changed, 122 insertions(+), 2 deletions(-) diff --git a/native/core/src/execution/jni_api.rs b/native/core/src/execution/jni_api.rs index 7652428410e..3d9ae7561cc 100644 --- a/native/core/src/execution/jni_api.rs +++ b/native/core/src/execution/jni_api.rs @@ -376,6 +376,7 @@ pub fn get_runtime() -> Handle { /// Must not be called from within the runtime's own worker threads, otherwise the shutdown /// would deadlock/panic. pub fn release_runtime() { + crate::execution::operators::clear_file_io_cache(); let runtime = TOKIO_RUNTIME.lock().take(); if let Some(runtime) = runtime { runtime.shutdown_timeout(Duration::from_secs(3)); diff --git a/native/core/src/execution/operators/iceberg_common.rs b/native/core/src/execution/operators/iceberg_common.rs index 509e3b1f986..d035eaf5aed 100644 --- a/native/core/src/execution/operators/iceberg_common.rs +++ b/native/core/src/execution/operators/iceberg_common.rs @@ -18,11 +18,12 @@ //! Helpers shared between the Iceberg scan and Iceberg write operators. use std::collections::HashMap; -use std::sync::Arc; +use std::sync::{Arc, LazyLock}; use datafusion::common::DataFusionError; use iceberg::io::{FileIO, FileIOBuilder, StorageFactory}; use iceberg_storage_opendal::{CustomAwsCredentialLoader, OpenDalStorageFactory}; +use parking_lot::Mutex; use crate::cloud::s3::credential_bridge::{AccessMode, CometS3CredentialBridge}; use crate::parquet::objectstore::s3_blob_fs_support::{ @@ -96,12 +97,89 @@ pub(crate) fn storage_factory_for( } } +#[derive(Debug, PartialEq, Eq, Hash)] +struct FileIoCacheKey { + access_mode: u8, + catalog_name: String, + /// The full path: the S3 access bridge is scoped to the exact path it was built for. + reference_path: String, + properties: Vec<(String, String)>, +} + +impl FileIoCacheKey { + /// `None` for `memory:///`, whose namespace must stay private to its task. + fn new( + catalog_properties: &HashMap, + reference_path: &str, + catalog_name: &str, + access_mode: AccessMode, + ) -> Option { + if scheme_of(reference_path) == "memory" { + return None; + } + let mut properties: Vec<(String, String)> = catalog_properties + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(); + properties.sort(); + Some(Self { + access_mode: access_mode as u8, + catalog_name: catalog_name.to_string(), + reference_path: reference_path.to_string(), + properties, + }) + } +} + +const FILE_IO_CACHE_CAPACITY: usize = 64; + +/// Shared per executor so tasks reuse one storage client instead of each building its own. +static FILE_IO_CACHE: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + +pub fn clear_file_io_cache() { + FILE_IO_CACHE.lock().clear(); +} + +pub(crate) fn load_file_io( + catalog_properties: &HashMap, + reference_path: &str, + catalog_name: &str, + access_mode: AccessMode, +) -> Result { + let key = FileIoCacheKey::new( + catalog_properties, + reference_path, + catalog_name, + access_mode, + ); + if let Some(key) = &key { + if let Some(file_io) = FILE_IO_CACHE.lock().get(key) { + return Ok(file_io.clone()); + } + } + let file_io = build_file_io( + catalog_properties, + reference_path, + catalog_name, + access_mode, + )?; + if let Some(key) = key { + let mut cache = FILE_IO_CACHE.lock(); + if cache.len() >= FILE_IO_CACHE_CAPACITY { + cache.clear(); + } + cache.insert(key, file_io.clone()); + } + Ok(file_io) +} + /// Build a `FileIO` whose storage scheme is inferred from `reference_path` and whose properties /// come from the catalog. The reference path is the metadata location for reads or the data /// location for writes — anything that carries the right URI scheme. `catalog_name` is the /// credential dispatch key and `access_mode` is the access intent forwarded to the S3 credential /// bridge, so the write path can request write-capable credentials. -pub(crate) fn load_file_io( +fn build_file_io( catalog_properties: &HashMap, reference_path: &str, catalog_name: &str, @@ -239,6 +317,46 @@ fn is_s3_family_scheme(scheme: &str, catalog_properties: &HashMap Result<(), String> { storage_factory_for(path, &HashMap::new(), "test_cat", mode) .map(|_| ()) diff --git a/native/core/src/execution/operators/mod.rs b/native/core/src/execution/operators/mod.rs index d09b0b4fb37..cd91365703f 100644 --- a/native/core/src/execution/operators/mod.rs +++ b/native/core/src/execution/operators/mod.rs @@ -35,6 +35,7 @@ pub use expand::ExpandExec; mod explode; pub use explode::ExplodeExec; mod iceberg_common; +pub use iceberg_common::clear_file_io_cache; mod iceberg_partition_path; mod iceberg_scan; mod iceberg_write; From 841cbb0b89b30be9bd6cb71d19ce874e1d1af8c2 Mon Sep 17 00:00:00 2001 From: Michael Taranov Date: Tue, 22 Sep 2026 11:23:38 +0300 Subject: [PATCH 2/3] perf: evict FileIO cache entries one at a time and test the hit path Review of #6106 found that the bound cleared the whole cache, including the entry a running query was using, and dropped every FileIO while holding the lock, where the last clone of an S3 access bridge releases JNI global refs. The cache is now a small LRU that evicts one entry per insert, and evicted or replaced FileIOs are dropped after the lock is released. The builder is injected into cached_file_io so tests can count builds: a repeated identical load builds once, a load without a key builds every time, memory:/// never enters the cache, and eviction removes the least recently used entry. Co-Authored-By: Claude Fable 5.1 --- .../src/execution/operators/iceberg_common.rs | 220 +++++++++++++----- 1 file changed, 168 insertions(+), 52 deletions(-) diff --git a/native/core/src/execution/operators/iceberg_common.rs b/native/core/src/execution/operators/iceberg_common.rs index d035eaf5aed..8d76049c525 100644 --- a/native/core/src/execution/operators/iceberg_common.rs +++ b/native/core/src/execution/operators/iceberg_common.rs @@ -17,7 +17,7 @@ //! Helpers shared between the Iceberg scan and Iceberg write operators. -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::sync::{Arc, LazyLock}; use datafusion::common::DataFusionError; @@ -97,7 +97,7 @@ pub(crate) fn storage_factory_for( } } -#[derive(Debug, PartialEq, Eq, Hash)] +#[derive(Clone, Debug, PartialEq, Eq, Hash)] struct FileIoCacheKey { access_mode: u8, catalog_name: String, @@ -133,12 +133,74 @@ impl FileIoCacheKey { const FILE_IO_CACHE_CAPACITY: usize = 64; +/// Least recently used entries are evicted first. +struct FileIoCache { + entries: HashMap, + order: VecDeque, + capacity: usize, +} + +impl FileIoCache { + fn new(capacity: usize) -> Self { + Self { + entries: HashMap::new(), + order: VecDeque::new(), + capacity, + } + } + + fn get(&mut self, key: &FileIoCacheKey) -> Option { + let file_io = self.entries.get(key)?.clone(); + if let Some(pos) = self.order.iter().position(|k| k == key) { + let recent = self.order.remove(pos)?; + self.order.push_back(recent); + } + Some(file_io) + } + + /// Returns the replaced or evicted `FileIO` so the caller can drop it outside the lock. + fn insert(&mut self, key: FileIoCacheKey, file_io: FileIO) -> Option { + if let Some(previous) = self.entries.insert(key.clone(), file_io) { + return Some(previous); + } + self.order.push_back(key); + if self.entries.len() > self.capacity { + let oldest = self.order.pop_front()?; + return self.entries.remove(&oldest); + } + None + } +} + /// Shared per executor so tasks reuse one storage client instead of each building its own. -static FILE_IO_CACHE: LazyLock>> = - LazyLock::new(|| Mutex::new(HashMap::new())); +static FILE_IO_CACHE: LazyLock> = + LazyLock::new(|| Mutex::new(FileIoCache::new(FILE_IO_CACHE_CAPACITY))); pub fn clear_file_io_cache() { - FILE_IO_CACHE.lock().clear(); + let dropped: Vec = { + let mut cache = FILE_IO_CACHE.lock(); + cache.order.clear(); + cache.entries.drain().map(|(_, file_io)| file_io).collect() + }; + drop(dropped); +} + +fn cached_file_io( + cache: &Mutex, + key: Option, + build: impl FnOnce() -> Result, +) -> Result { + let Some(key) = key else { + return build(); + }; + if let Some(file_io) = cache.lock().get(&key) { + return Ok(file_io); + } + let file_io = build()?; + // Dropped after the lock is released: the last clone of a FileIO releases JNI global refs. + let evicted = cache.lock().insert(key, file_io.clone()); + drop(evicted); + Ok(file_io) } pub(crate) fn load_file_io( @@ -147,31 +209,23 @@ pub(crate) fn load_file_io( catalog_name: &str, access_mode: AccessMode, ) -> Result { - let key = FileIoCacheKey::new( - catalog_properties, - reference_path, - catalog_name, - access_mode, - ); - if let Some(key) = &key { - if let Some(file_io) = FILE_IO_CACHE.lock().get(key) { - return Ok(file_io.clone()); - } - } - let file_io = build_file_io( - catalog_properties, - reference_path, - catalog_name, - access_mode, - )?; - if let Some(key) = key { - let mut cache = FILE_IO_CACHE.lock(); - if cache.len() >= FILE_IO_CACHE_CAPACITY { - cache.clear(); - } - cache.insert(key, file_io.clone()); - } - Ok(file_io) + cached_file_io( + &FILE_IO_CACHE, + FileIoCacheKey::new( + catalog_properties, + reference_path, + catalog_name, + access_mode, + ), + || { + build_file_io( + catalog_properties, + reference_path, + catalog_name, + access_mode, + ) + }, + ) } /// Build a `FileIO` whose storage scheme is inferred from `reference_path` and whose properties @@ -317,46 +371,108 @@ fn is_s3_family_scheme(scheme: &str, catalog_properties: &HashMap FileIO { + build_file_io( + &HashMap::new(), + "file:///tmp/warehouse", + "", + AccessMode::Read, + ) + .unwrap() + } + #[test] - fn load_file_io_is_cached_per_storage_client_configuration() { + fn cache_key_separates_client_configurations() { let props = HashMap::from([("s3.region".to_string(), "eu-west-1".to_string())]); - let catalog = "file_io_cache_test"; let table = "s3://bucket/warehouse/db/t"; - let key = |mode| FileIoCacheKey::new(&props, table, catalog, mode).unwrap(); - { - let mut cache = FILE_IO_CACHE.lock(); - cache.remove(&key(AccessMode::Read)); - cache.remove(&key(AccessMode::Write)); - } - - load_file_io(&props, table, catalog, AccessMode::Read).unwrap(); - assert!(FILE_IO_CACHE.lock().contains_key(&key(AccessMode::Read))); - assert!(!FILE_IO_CACHE.lock().contains_key(&key(AccessMode::Write))); + let key = |mode| FileIoCacheKey::new(&props, table, "cat", mode); + assert_ne!(key(AccessMode::Read), key(AccessMode::Write)); assert_ne!( + key(AccessMode::Read), FileIoCacheKey::new( &props, "s3://bucket/warehouse/db/other", - catalog, + "cat", AccessMode::Read - ), - Some(key(AccessMode::Read)) + ) + ); + assert_ne!( + key(AccessMode::Read), + FileIoCacheKey::new(&props, table, "other_cat", AccessMode::Read) ); - - load_file_io(&props, table, catalog, AccessMode::Write).unwrap(); - assert!(FILE_IO_CACHE.lock().contains_key(&key(AccessMode::Write))); - let mut moved = props.clone(); moved.insert("s3.endpoint".to_string(), "http://minio:9000".to_string()); assert_ne!( - FileIoCacheKey::new(&moved, table, catalog, AccessMode::Read), - Some(key(AccessMode::Read)) + key(AccessMode::Read), + FileIoCacheKey::new(&moved, table, "cat", AccessMode::Read) ); - assert!( FileIoCacheKey::new(&HashMap::new(), "memory:///", "", AccessMode::Write).is_none() ); } + #[test] + fn cached_file_io_builds_once_per_key_and_always_without_a_key() { + let cache = Mutex::new(FileIoCache::new(4)); + let key = FileIoCacheKey::new( + &HashMap::new(), + "file:///tmp/warehouse", + "", + AccessMode::Read, + ); + let mut builds = 0; + for _ in 0..2 { + cached_file_io(&cache, key.clone(), || { + builds += 1; + Ok(local_file_io()) + }) + .unwrap(); + } + assert_eq!(builds, 1); + for _ in 0..2 { + cached_file_io(&cache, None, || { + builds += 1; + Ok(local_file_io()) + }) + .unwrap(); + } + assert_eq!(builds, 3); + assert_eq!(cache.lock().entries.len(), 1); + } + + #[test] + fn load_file_io_does_not_cache_memory() { + load_file_io(&HashMap::new(), "memory:///", "", AccessMode::Write).unwrap(); + assert!(FILE_IO_CACHE + .lock() + .entries + .keys() + .all(|k| scheme_of(&k.reference_path) != "memory")); + } + + #[test] + fn cache_evicts_the_least_recently_used_entry() { + let mut cache = FileIoCache::new(2); + let key = |name: &str| { + FileIoCacheKey::new( + &HashMap::new(), + &format!("file:///{name}"), + "", + AccessMode::Read, + ) + .unwrap() + }; + assert!(cache.insert(key("a"), local_file_io()).is_none()); + assert!(cache.insert(key("b"), local_file_io()).is_none()); + assert!(cache.get(&key("a")).is_some()); + assert!(cache.insert(key("c"), local_file_io()).is_some()); + assert!(cache.get(&key("b")).is_none()); + assert!(cache.get(&key("a")).is_some()); + assert!(cache.get(&key("c")).is_some()); + assert!(cache.insert(key("c"), local_file_io()).is_some()); + assert_eq!(cache.entries.len(), 2); + } + fn factory_result(path: &str, mode: AccessMode) -> Result<(), String> { storage_factory_for(path, &HashMap::new(), "test_cat", mode) .map(|_| ()) From 36ffdc1f57043043b0e6cc16e7b33096693a97b8 Mon Sep 17 00:00:00 2001 From: Michael Taranov Date: Tue, 22 Sep 2026 12:12:45 +0300 Subject: [PATCH 3/3] perf: do not cache a FileIO whose S3 access provider failed to initialise A read whose configured S3 access provider fails to initialise falls back to opendal's default chain. Caching that FileIO made the fallback sticky for every later task on the executor. storage_factory_for and build_file_io now report whether the build is cacheable, and a degraded build is returned without being inserted so the next task retries. Also splits the S3 operator cache follow-up into #6109 and updates two comments that still described per-task lifetimes. Co-Authored-By: Claude Fable 5.1 --- native/core/src/cloud/s3/credential_bridge.rs | 4 +- .../src/execution/operators/iceberg_common.rs | 96 ++++++++++++------- 2 files changed, 65 insertions(+), 35 deletions(-) diff --git a/native/core/src/cloud/s3/credential_bridge.rs b/native/core/src/cloud/s3/credential_bridge.rs index 9fc5b562cbc..603f6c5f256 100644 --- a/native/core/src/cloud/s3/credential_bridge.rs +++ b/native/core/src/cloud/s3/credential_bridge.rs @@ -45,8 +45,8 @@ use std::time::Duration; /// executor from holding a stale credential for the entire job lifetime. const DEFAULT_EXPIRY_WHEN_UNKNOWN: Duration = Duration::from_secs(300); -/// Once-per-process latch for the "missing expiry" warning. Bridges are per-scan, so a per-bridge -/// latch would re-log on every scan. +/// Once-per-process latch for the "missing expiry" warning. Bridges live as long as their entry in +/// the executor's FileIO cache, so a per-bridge latch would re-log for every new configuration. static WARNED_MISSING_EXPIRY: OnceCell<()> = OnceCell::new(); /// Access intent forwarded to the Java SPI. Ordinal must match the JVM `CometS3AccessMode` enum. diff --git a/native/core/src/execution/operators/iceberg_common.rs b/native/core/src/execution/operators/iceberg_common.rs index 8d76049c525..d56f776833f 100644 --- a/native/core/src/execution/operators/iceberg_common.rs +++ b/native/core/src/execution/operators/iceberg_common.rs @@ -49,24 +49,27 @@ const STORAGE_PROPERTY_PREFIXES: &[&str] = &["s3.", "gcs.", "adls.", "client."]; /// cleanly instead of failing at execution. Changing the arms below means updating /// `CometScanRule.icebergReadableSchemes` (reads) and /// `CometIcebergNativeWrite.SupportedStorageSchemes` (writes). +/// The bool is false when a read fell back to opendal's default chain because the configured S3 +/// access provider failed to initialise; such a `FileIO` must not be cached, so the next task +/// retries. pub(crate) fn storage_factory_for( path: &str, catalog_properties: &HashMap, catalog_name: &str, access_mode: AccessMode, -) -> Result, DataFusionError> { +) -> Result<(Arc, bool), DataFusionError> { let scheme = scheme_of(path); match scheme { - "file" => Ok(Arc::new(OpenDalStorageFactory::Fs)), - "memory" => Ok(Arc::new(OpenDalStorageFactory::Memory)), - "gs" => Ok(Arc::new(OpenDalStorageFactory::Gcs)), + "file" => Ok((Arc::new(OpenDalStorageFactory::Fs), true)), + "memory" => Ok((Arc::new(OpenDalStorageFactory::Memory), true)), + "gs" => Ok((Arc::new(OpenDalStorageFactory::Gcs), true)), // Reads keep the OSS backend they have always had (CometScanRule admits `oss` scan // locations through HadoopFileIO). Writes fail closed: Comet does not forward `oss.*` // properties into the FileIO and no test covers the write path, so OSS-specific // endpoint/credential configuration could silently be dropped. The JVM write gate // already declines `oss` locations; this is the native-side backstop. "oss" => match access_mode { - AccessMode::Read => Ok(Arc::new(OpenDalStorageFactory::Oss)), + AccessMode::Read => Ok((Arc::new(OpenDalStorageFactory::Oss), true)), AccessMode::Write => Err(DataFusionError::Execution( "OSS is not supported for native Iceberg writes (oss.* properties are not \ forwarded to the native FileIO)" @@ -79,17 +82,19 @@ pub(crate) fn storage_factory_for( // promotes a HOSTLESS `blob:///bucket/key` into the host at the open boundary -- see // s3_blob_fs_support for why that never touches the recorded delete-matching string. s if is_s3_family_scheme(s, catalog_properties) => { - let customized_credential_load = + let (customized_credential_load, cacheable) = build_s3_credential_loader(path, catalog_properties, catalog_name, access_mode)?; - if is_s3_compliant_alias_scheme(s, catalog_properties) { - Ok(Arc::new(BlobHostPromotingS3StorageFactory::new( - customized_credential_load, - ))) - } else { - Ok(Arc::new(OpenDalStorageFactory::S3 { - customized_credential_load, - })) - } + let factory: Arc = + if is_s3_compliant_alias_scheme(s, catalog_properties) { + Arc::new(BlobHostPromotingS3StorageFactory::new( + customized_credential_load, + )) + } else { + Arc::new(OpenDalStorageFactory::S3 { + customized_credential_load, + }) + }; + Ok((factory, cacheable)) } _ => Err(DataFusionError::Execution(format!( "Unsupported storage scheme: {scheme}" @@ -172,7 +177,8 @@ impl FileIoCache { } } -/// Shared per executor so tasks reuse one storage client instead of each building its own. +/// Shared per executor so tasks reuse one FileIO: its factory, parsed config and access bridge, +/// plus the storage client where the backend caches operators. static FILE_IO_CACHE: LazyLock> = LazyLock::new(|| Mutex::new(FileIoCache::new(FILE_IO_CACHE_CAPACITY))); @@ -188,18 +194,20 @@ pub fn clear_file_io_cache() { fn cached_file_io( cache: &Mutex, key: Option, - build: impl FnOnce() -> Result, + build: impl FnOnce() -> Result<(FileIO, bool), DataFusionError>, ) -> Result { let Some(key) = key else { - return build(); + return Ok(build()?.0); }; if let Some(file_io) = cache.lock().get(&key) { return Ok(file_io); } - let file_io = build()?; - // Dropped after the lock is released: the last clone of a FileIO releases JNI global refs. - let evicted = cache.lock().insert(key, file_io.clone()); - drop(evicted); + let (file_io, cacheable) = build()?; + if cacheable { + // Dropped after the lock is released: the last clone of a FileIO releases JNI global refs. + let evicted = cache.lock().insert(key, file_io.clone()); + drop(evicted); + } Ok(file_io) } @@ -238,8 +246,8 @@ fn build_file_io( reference_path: &str, catalog_name: &str, access_mode: AccessMode, -) -> Result { - let factory = storage_factory_for( +) -> Result<(FileIO, bool), DataFusionError> { + let (factory, cacheable) = storage_factory_for( reference_path, catalog_properties, catalog_name, @@ -276,7 +284,7 @@ fn build_file_io( file_io_builder = file_io_builder.with_prop("s3.region", "us-east-1"); } - Ok(file_io_builder.build()) + Ok((file_io_builder.build(), cacheable)) } /// Wires the configured Comet credential provider into opendal's S3 service. `Ok(None)` means no @@ -290,19 +298,19 @@ fn build_s3_credential_loader( catalog_properties: &HashMap, catalog_name: &str, access_mode: AccessMode, -) -> Result, DataFusionError> { +) -> Result<(Option, bool), DataFusionError> { let Ok(url) = url::Url::parse(reference_path) else { - return Ok(None); + return Ok((None, true)); }; let Some(bucket) = url.host_str() else { - return Ok(None); + return Ok((None, true)); }; let Some(provider_class) = catalog_properties .get(ICEBERG_PROVIDER_CLASS_PROPERTY) .map(|s| s.trim()) .filter(|s| !s.is_empty()) else { - return Ok(None); + return Ok((None, true)); }; // Fall back to the bucket when the table has no catalog identity (e.g. HadoopTables loaded by // raw path). @@ -320,7 +328,7 @@ fn build_s3_credential_loader( catalog_properties, ); match bridge { - Ok(b) => Ok(Some(CustomAwsCredentialLoader::new(b))), + Ok(b) => Ok((Some(CustomAwsCredentialLoader::new(b)), true)), Err(e) => match access_mode { AccessMode::Write => Err(DataFusionError::Execution(format!( "Configured S3 credential provider {provider_class} failed to initialize: {e}; \ @@ -331,7 +339,7 @@ fn build_s3_credential_loader( "Failed to initialize CometS3CredentialBridge for {provider_class}: {e}; \ falling back to default opendal credential chain" ); - Ok(None) + Ok((None, false)) } }, } @@ -379,6 +387,7 @@ mod tests { AccessMode::Read, ) .unwrap() + .0 } #[test] @@ -424,7 +433,7 @@ mod tests { for _ in 0..2 { cached_file_io(&cache, key.clone(), || { builds += 1; - Ok(local_file_io()) + Ok((local_file_io(), true)) }) .unwrap(); } @@ -432,7 +441,7 @@ mod tests { for _ in 0..2 { cached_file_io(&cache, None, || { builds += 1; - Ok(local_file_io()) + Ok((local_file_io(), true)) }) .unwrap(); } @@ -440,6 +449,27 @@ mod tests { assert_eq!(cache.lock().entries.len(), 1); } + #[test] + fn cached_file_io_does_not_cache_a_degraded_build() { + let cache = Mutex::new(FileIoCache::new(4)); + let key = FileIoCacheKey::new( + &HashMap::new(), + "file:///tmp/warehouse", + "", + AccessMode::Read, + ); + let mut builds = 0; + for _ in 0..2 { + cached_file_io(&cache, key.clone(), || { + builds += 1; + Ok((local_file_io(), false)) + }) + .unwrap(); + } + assert_eq!(builds, 2); + assert!(cache.lock().entries.is_empty()); + } + #[test] fn load_file_io_does_not_cache_memory() { load_file_io(&HashMap::new(), "memory:///", "", AccessMode::Write).unwrap();