From dbfc09a6c5f4fd6c16a1cabce89f4a9ce7b6066d Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Sat, 31 Oct 2026 17:53:07 +0800 Subject: [PATCH 1/2] feat(core): support local-cache extension exclusions Add optional extension exclusions to the local block cache so data files can remain cacheable without caching BLOB bodies. Preserve write and delete invalidation for excluded files, and cover memory, disk and shared-disk reads with regression tests. Closes #971 Signed-off-by: QuakeWang --- crates/paimon/src/common/options.rs | 3 + crates/paimon/src/io/cache/mod.rs | 40 ++- crates/paimon/src/io/cache/reader.rs | 17 ++ crates/paimon/src/io/cache/shared.rs | 1 + crates/paimon/src/io/file_io.rs | 6 +- .../tests/local_cache_exclusions_test.rs | 269 ++++++++++++++++++ docs/src/getting-started.md | 12 + 7 files changed, 345 insertions(+), 3 deletions(-) create mode 100644 crates/paimon/tests/local_cache_exclusions_test.rs diff --git a/crates/paimon/src/common/options.rs b/crates/paimon/src/common/options.rs index 28eab750c..01082ec46 100644 --- a/crates/paimon/src/common/options.rs +++ b/crates/paimon/src/common/options.rs @@ -97,6 +97,9 @@ impl CatalogOptions { /// Comma-separated file types eligible for local caching. pub const LOCAL_CACHE_WHITELIST: &'static str = "local-cache.whitelist"; + /// Comma-separated file extensions excluded from local caching (empty by default). + pub const LOCAL_CACHE_EXCLUDE_EXTENSIONS: &'static str = "local-cache.exclude-extensions"; + /// Maximum process-local cache size for parsed file-format metadata. pub const FILE_FORMAT_METADATA_CACHE_MAX_SIZE: &'static str = "file-format.metadata-cache.max-size"; diff --git a/crates/paimon/src/io/cache/mod.rs b/crates/paimon/src/io/cache/mod.rs index eb0fbf4de..0bc7cfd8d 100644 --- a/crates/paimon/src/io/cache/mod.rs +++ b/crates/paimon/src/io/cache/mod.rs @@ -51,6 +51,7 @@ pub(crate) struct LocalCache { block_size: u64, whitelist: HashSet, file_size_capacity: usize, + excluded_extensions: HashSet, // Bound queued and active background fills. fill_slots: Arc, // Generated lookup reads can outlive the executor. Keep temporary disk @@ -84,6 +85,7 @@ impl LocalCache { block_size, whitelist: FileType::parse_whitelist(whitelist), file_size_capacity: DEFAULT_FILE_SIZE_CAPACITY, + excluded_extensions: HashSet::new(), fill_slots: Arc::new(tokio::sync::Semaphore::new(8)), _temporary_directory: None, }) @@ -134,6 +136,7 @@ impl LocalCache { block_size: config.block_size, whitelist: config.whitelist, file_size_capacity, + excluded_extensions: config.excluded_extensions, fill_slots: Arc::new(tokio::sync::Semaphore::new(8)), _temporary_directory: None, }) @@ -152,6 +155,7 @@ impl LocalCache { let mut cache = Self::new_with_ratio( LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: String::new(), max_size, @@ -188,10 +192,28 @@ impl LocalCache { BlockKey::with_namespace(&self.namespace, path, self.block_size, block_index) } - pub(super) fn is_cacheable(&self, path: &str) -> bool { + /// Extension exclusions must not disable invalidation of existing shared blocks. + pub(super) fn is_whitelisted(&self, path: &str) -> bool { !FileType::is_mutable(path) && self.whitelist.contains(&FileType::classify(path)) } + pub(super) fn is_cacheable(&self, path: &str) -> bool { + if !self.is_whitelisted(path) { + return false; + } + if self.excluded_extensions.is_empty() { + return true; + } + let name = path.rsplit('/').next().unwrap_or(path); + !std::path::Path::new(name) + .extension() + .and_then(|extension| extension.to_str()) + .is_some_and(|extension| { + self.excluded_extensions + .contains(&extension.to_ascii_lowercase()) + }) + } + async fn get_block( &self, key: &BlockKey, @@ -450,6 +472,7 @@ pub(crate) async fn create_local_cache_with_namespace_async( #[derive(Debug)] pub(crate) struct LocalCacheConfig { shared: bool, + excluded_extensions: HashSet, dir: Option, namespace: String, max_size: Option, @@ -533,6 +556,15 @@ impl LocalCacheConfig { max_size, block_size, whitelist: FileType::parse_whitelist(whitelist), + excluded_extensions: options + .get(CatalogOptions::LOCAL_CACHE_EXCLUDE_EXTENSIONS) + .map(String::as_str) + .unwrap_or_default() + .split(',') + .map(|extension| extension.trim().trim_start_matches('.')) + .filter(|extension| !extension.is_empty()) + .map(str::to_ascii_lowercase) + .collect(), })) } } @@ -712,6 +744,7 @@ mod tests { let directory = tempfile::tempdir().unwrap(); let cache = LocalCache::new(LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".to_string(), max_size: None, @@ -774,6 +807,7 @@ mod tests { let directory = tempfile::tempdir().unwrap(); let config = || LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".to_string(), max_size: None, @@ -797,6 +831,7 @@ mod tests { let directory = tempfile::tempdir().unwrap(); let config = || LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".to_string(), max_size: None, @@ -821,6 +856,7 @@ mod tests { let directory = tempfile::tempdir().unwrap(); let cache = LocalCache::new(LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".to_string(), max_size: None, @@ -851,6 +887,7 @@ mod tests { LocalCache::new(LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".to_string(), max_size: None, @@ -872,6 +909,7 @@ mod tests { let directory = tempfile::tempdir().unwrap(); let cache = LocalCache::new(LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".to_string(), max_size: Some(8), diff --git a/crates/paimon/src/io/cache/reader.rs b/crates/paimon/src/io/cache/reader.rs index cafb398ac..b9886d3e1 100644 --- a/crates/paimon/src/io/cache/reader.rs +++ b/crates/paimon/src/io/cache/reader.rs @@ -393,6 +393,7 @@ mod tests { Arc::new( LocalCache::new(LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: None, namespace: "test".into(), max_size: None, @@ -472,6 +473,7 @@ mod tests { let cache = Arc::new( LocalCache::new(LocalCacheConfig { shared: true, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".into(), max_size: Some(16 * 1024 * 1024), @@ -567,6 +569,7 @@ mod tests { let cache = Arc::new( LocalCache::new(LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: None, namespace: "test".to_string(), max_size: None, @@ -603,6 +606,7 @@ mod tests { let cache = Arc::new( LocalCache::new(LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: None, namespace: "test".to_string(), max_size: Some(16), @@ -657,6 +661,7 @@ mod tests { let cache = Arc::new( LocalCache::new(LocalCacheConfig { shared: true, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".into(), max_size: Some(128 * 1024 * 1024), @@ -729,6 +734,7 @@ mod tests { let cache = Arc::new( LocalCache::new(LocalCacheConfig { shared: true, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".into(), max_size: Some(4096), @@ -785,6 +791,7 @@ mod tests { let cache = Arc::new( LocalCache::new(LocalCacheConfig { shared: true, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".into(), max_size: Some(4096), @@ -841,6 +848,7 @@ mod tests { let cache = Arc::new( LocalCache::new(LocalCacheConfig { shared: true, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".into(), max_size: Some(4096), @@ -882,6 +890,7 @@ mod tests { let cache = Arc::new( LocalCache::new(LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: None, namespace: "test".to_string(), max_size: None, @@ -913,6 +922,7 @@ mod tests { }); let config = || LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".to_string(), max_size: None, @@ -961,6 +971,7 @@ mod tests { let cache = Arc::new( LocalCache::new(LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: None, namespace: "test".to_string(), max_size: None, @@ -1011,6 +1022,7 @@ mod tests { let cache_a = Arc::new( LocalCache::new(LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".to_string(), max_size: None, @@ -1022,6 +1034,7 @@ mod tests { let cache_b = Arc::new( LocalCache::new(LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".to_string(), max_size: None, @@ -1074,6 +1087,7 @@ mod tests { let blocked_path = "s3://bucket/table/snapshot/snapshot-3"; let config = || LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".to_string(), max_size: None, @@ -1221,6 +1235,7 @@ mod tests { let cache = Arc::new( LocalCache::new(LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".to_string(), max_size: None, @@ -1256,6 +1271,7 @@ mod tests { let cache = Arc::new( LocalCache::new(LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: Some(directory.path().to_path_buf()), namespace: "test".to_string(), max_size: None, @@ -1278,6 +1294,7 @@ mod tests { let cache = Arc::new( LocalCache::new(LocalCacheConfig { shared: false, + excluded_extensions: Default::default(), dir: None, namespace: "test".to_string(), max_size: None, diff --git a/crates/paimon/src/io/cache/shared.rs b/crates/paimon/src/io/cache/shared.rs index 4ced74492..910335c14 100644 --- a/crates/paimon/src/io/cache/shared.rs +++ b/crates/paimon/src/io/cache/shared.rs @@ -747,6 +747,7 @@ mod tests { let local = Arc::new( LocalCache::new(LocalCacheConfig { shared: true, + excluded_extensions: Default::default(), dir: Some(root.clone()), namespace: "catalog".into(), max_size: Some(4096), diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs index dec7b603a..4956d34c1 100644 --- a/crates/paimon/src/io/file_io.rs +++ b/crates/paimon/src/io/file_io.rs @@ -661,7 +661,7 @@ impl FileIO { cache: self .cache .as_ref() - .filter(|cache| cache.is_cacheable(path)) + .filter(|cache| cache.is_whitelisted(path)) .cloned(), }) } @@ -875,7 +875,9 @@ impl FileIO { .invalidate_row_sidecar_size(cache_path); } if let (Some(cache), Some(cache_path)) = ( - self.cache.as_ref().filter(|cache| cache.is_cacheable(path)), + self.cache + .as_ref() + .filter(|cache| cache.is_whitelisted(path)), cache_path, ) { cache.invalidate_path(&cache_path).await?; diff --git a/crates/paimon/tests/local_cache_exclusions_test.rs b/crates/paimon/tests/local_cache_exclusions_test.rs new file mode 100644 index 000000000..f9b26af19 --- /dev/null +++ b/crates/paimon/tests/local_cache_exclusions_test.rs @@ -0,0 +1,269 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#![cfg(feature = "storage-fs")] + +use paimon::catalog::FileSystemCatalog; +use paimon::io::FileRead; +use paimon::{CatalogOptions, Error, Options}; + +async fn assert_cache_selection( + whitelist: &str, + excluded_extensions: Option<&str>, + files: &[(&str, bool)], +) { + let mut mismatches = Vec::new(); + for backend in ["memory", "disk", "shared"] { + for range_read in [false, true] { + let directory = tempfile::tempdir().unwrap(); + let warehouse = directory.path().join("warehouse"); + std::fs::create_dir(&warehouse).unwrap(); + let mut options = Options::new(); + options.set(CatalogOptions::WAREHOUSE, warehouse.to_string_lossy()); + options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true"); + options.set(CatalogOptions::LOCAL_CACHE_BLOCK_SIZE, "4"); + options.set(CatalogOptions::LOCAL_CACHE_WHITELIST, whitelist); + if backend != "memory" { + options.set( + CatalogOptions::LOCAL_CACHE_DIR, + directory.path().join("cache").to_string_lossy(), + ); + } + if backend == "shared" { + options.set(CatalogOptions::LOCAL_CACHE_SHARED, "true"); + options.set(CatalogOptions::LOCAL_CACHE_MAX_SIZE, "1 MiB"); + } + if let Some(extensions) = excluded_extensions { + options.set(CatalogOptions::LOCAL_CACHE_EXCLUDE_EXTENSIONS, extensions); + } + let catalog = FileSystemCatalog::new(options).unwrap(); + for &(name, expected_cached) in files { + let path = warehouse.join(name); + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + // Only block-cache routing is under test, not format decoding. + std::fs::write(&path, b"abcdefgh").unwrap(); + let input = catalog.file_io().new_input(path.to_str().unwrap()).unwrap(); + let expected: &[u8] = if range_read { b"bcdefg" } else { b"abcdefgh" }; + let cold = if range_read { + input.reader().await.unwrap().read(1..7).await.unwrap() + } else { + input.read().await.unwrap() + }; + assert_eq!(cold.as_ref(), expected); + + // Bypass FileIO invalidation. Keep the source available for shared-cache stat. + std::fs::write(&path, b"12345678").unwrap(); + let replacement: &[u8] = if range_read { b"234567" } else { b"12345678" }; + let warm = if range_read { + async { input.reader().await?.read(1..7).await }.await + } else { + input.read().await + }; + let warm = warm.unwrap(); + assert!(warm.as_ref() == expected || warm.as_ref() == replacement); + let cached = warm.as_ref() == expected; + if cached != expected_cached { + mismatches.push(format!( + "backend={backend}, range_read={range_read}, whitelist={whitelist}, \ + excluded_extensions={excluded_extensions:?}, file={name}: \ + expected cached={expected_cached}, actual cached={cached}" + )); + } + } + } + } + assert!(mismatches.is_empty(), "{}", mismatches.join("\n")); +} + +#[tokio::test] +async fn test_missing_or_empty_exclusions_preserve_data_cache() { + for exclusions in [None, Some(""), Some(" , , ")] { + assert_cache_selection( + "meta,global-index,data", + exclusions, + &[("data.parquet", true), ("data.blob", true)], + ) + .await; + } +} + +#[tokio::test] +async fn test_structured_data_cache_excludes_dedicated_files() { + assert_cache_selection( + "meta,global-index,structured-data", + None, + &[ + ("data.parquet", true), + ("data.unknown", true), + ("data.blob", false), + ("data.video", false), + ("data.vector.vortex", false), + ], + ) + .await; +} + +#[tokio::test] +async fn test_issue_971_excluded_blob_bypasses_cache() { + assert_cache_selection( + "meta,global-index,data", + Some("blob"), + &[("data.parquet", true), ("data.blob", false)], + ) + .await; +} + +#[tokio::test] +async fn test_exclusions_normalize_entries_and_match_only_the_last_extension() { + assert_cache_selection( + "*", + Some(" blob, .VIDEO, , BLOB "), + &[ + ("data.blob", false), + ("data.BLOB", false), + ("data.video", false), + ("data.VIDEO", false), + ("data.parquet.blob", false), + ("data.blob.gz", true), + ("data.blob.index", true), + ("data.blobber", true), + ("dir.blob/data.parquet", true), + ("blob", true), + (".blob", true), + ], + ) + .await; +} + +#[tokio::test] +async fn test_exclusions_override_whitelisted_metadata_and_indexes() { + assert_cache_selection( + "*", + Some("blob,index,gz"), + &[ + ("snapshot-1.blob", false), + ("global-index-1.index", false), + ("data.parquet.index", false), + ("data.csv.gz", false), + ("manifest-1.avro", true), + ("snapshot-1", true), + ("data.csv", true), + ], + ) + .await; +} + +#[tokio::test] +async fn test_exclusions_do_not_expand_the_whitelist() { + assert_cache_selection( + "meta", + Some("blob"), + &[ + ("snapshot-1", true), + ("data.parquet", false), + ("data.blob", false), + ("global-index-1.index", false), + ], + ) + .await; + assert_cache_selection( + "structured-data", + Some("parquet"), + &[ + ("data.parquet", false), + ("data.unknown", true), + ("data.blob", false), + ("snapshot-1", false), + ], + ) + .await; +} + +#[tokio::test] +async fn test_exclusions_bypass_existing_blocks_and_preserve_invalidation() { + for shared in [false, true] { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("data.blob"); + std::fs::write(&path, b"abcdefgh").unwrap(); + let mut options = Options::new(); + options.set( + CatalogOptions::WAREHOUSE, + directory.path().to_string_lossy(), + ); + options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true"); + options.set(CatalogOptions::LOCAL_CACHE_WHITELIST, "data"); + options.set(CatalogOptions::LOCAL_CACHE_BLOCK_SIZE, "4"); + options.set(CatalogOptions::LOCAL_CACHE_MAX_SIZE, "1 MiB"); + options.set(CatalogOptions::LOCAL_CACHE_SHARED, shared.to_string()); + options.set( + CatalogOptions::LOCAL_CACHE_DIR, + directory.path().join("cache").to_string_lossy(), + ); + let catalog = FileSystemCatalog::new(options.clone()).unwrap(); + let cached_input = catalog.file_io().new_input(path.to_str().unwrap()).unwrap(); + assert_eq!(cached_input.read().await.unwrap().as_ref(), b"abcdefgh"); + std::fs::write(&path, b"12345678").unwrap(); + assert_eq!(cached_input.read().await.unwrap().as_ref(), b"abcdefgh"); + + options.set(CatalogOptions::LOCAL_CACHE_EXCLUDE_EXTENSIONS, "blob"); + let excluded_catalog = FileSystemCatalog::new(options).unwrap(); + let input = excluded_catalog + .file_io() + .new_input(path.to_str().unwrap()) + .unwrap(); + assert_eq!(input.read().await.unwrap().as_ref(), b"12345678"); + assert_eq!( + input + .reader() + .await + .unwrap() + .read(1..7) + .await + .unwrap() + .as_ref(), + b"234567" + ); + + let output = excluded_catalog + .file_io() + .new_output(path.to_str().unwrap()) + .unwrap(); + output + .write(bytes::Bytes::from_static(b"updated!")) + .await + .unwrap(); + assert_eq!(cached_input.read().await.unwrap().as_ref(), b"updated!"); + std::fs::write(&path, b"changed!").unwrap(); + assert_eq!( + output.to_input_file().read().await.unwrap().as_ref(), + b"changed!" + ); + + excluded_catalog + .file_io() + .delete_file(path.to_str().unwrap()) + .await + .unwrap(); + assert!(matches!( + cached_input.read().await, + Err(Error::IoUnexpected { source, .. }) + if source.kind() == opendal::ErrorKind::NotFound + )); + std::fs::write(&path, b"recreate").unwrap(); + assert_eq!(cached_input.read().await.unwrap().as_ref(), b"recreate"); + } +} diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md index 5bcce1a87..bba456ce7 100644 --- a/docs/src/getting-started.md +++ b/docs/src/getting-started.md @@ -445,11 +445,23 @@ let catalog = CatalogFactory::create(options).await?; | `local-cache.high-priority-pool-ratio` | `0.25` | Fraction of a finite cache budget protected for `meta` and `global-index` blocks. Use `0` for a single LRU. | | `local-cache.block-size` | `1 MiB` | Block size used for cached range reads. | | `local-cache.whitelist` | `meta,global-index` | Comma-separated eligible types: `meta`, `global-index`, `bucket-index`, `data`, `structured-data`, and `file-index`, or `*` for all of them. | +| `local-cache.exclude-extensions` | empty | Comma-separated extensions to exclude from local block caching, overriding every whitelist selection, including `*`, metadata, and indexes. | Use `local-cache.whitelist=meta,global-index,structured-data` to cache data blocks without dedicated BLOB, video, or vector-store files. `data` and `*` retain their broader behavior. +To exclude only BLOB files while keeping other data eligible, use +`local-cache.whitelist=meta,global-index,data` with +`local-cache.exclude-extensions=blob`. Extensions match the last extension of +the file name, ignoring ASCII case and an optional leading dot in each entry. +Surrounding whitespace and empty entries are ignored. For example, `blob, .VIDEO` +excludes `.blob` and `.video` files; `gz` excludes `data.csv.gz`, while `csv` +does not. Directory names do not participate in matching. An omitted or empty +list preserves existing behavior. This option only filters source-file block +caching; it does not change parsed metadata caches or generated lookup files. +Extension exclusions do not suppress write or delete cache invalidation. + The protected and regular LRUs share one total budget; either can use idle capacity. Catalogs or processes sharing a disk cache must use the same ratio. From 54ae5187f1ea7b6189fa4980c4662c271fa6dafb Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Sat, 31 Oct 2026 19:59:02 +0800 Subject: [PATCH 2/2] test(core): normalize local-cache test paths on Windows Use slash-separated paths for FileIO cache classification while retaining native paths for direct filesystem operations. Keep all exclusion and invalidation assertions enabled on Windows. Signed-off-by: QuakeWang --- .../tests/local_cache_exclusions_test.rs | 19 ++++++++----------- 1 file changed, 8 insertions(+), 11 deletions(-) diff --git a/crates/paimon/tests/local_cache_exclusions_test.rs b/crates/paimon/tests/local_cache_exclusions_test.rs index f9b26af19..c8c115298 100644 --- a/crates/paimon/tests/local_cache_exclusions_test.rs +++ b/crates/paimon/tests/local_cache_exclusions_test.rs @@ -56,7 +56,9 @@ async fn assert_cache_selection( std::fs::create_dir_all(path.parent().unwrap()).unwrap(); // Only block-cache routing is under test, not format decoding. std::fs::write(&path, b"abcdefgh").unwrap(); - let input = catalog.file_io().new_input(path.to_str().unwrap()).unwrap(); + // Cache classification uses slash-separated FileIO paths on every platform. + let io_path = path.to_str().unwrap().replace('\\', "/"); + let input = catalog.file_io().new_input(&io_path).unwrap(); let expected: &[u8] = if range_read { b"bcdefg" } else { b"abcdefgh" }; let cold = if range_read { input.reader().await.unwrap().read(1..7).await.unwrap() @@ -198,6 +200,7 @@ async fn test_exclusions_bypass_existing_blocks_and_preserve_invalidation() { for shared in [false, true] { let directory = tempfile::tempdir().unwrap(); let path = directory.path().join("data.blob"); + let io_path = path.to_str().unwrap().replace('\\', "/"); std::fs::write(&path, b"abcdefgh").unwrap(); let mut options = Options::new(); options.set( @@ -214,17 +217,14 @@ async fn test_exclusions_bypass_existing_blocks_and_preserve_invalidation() { directory.path().join("cache").to_string_lossy(), ); let catalog = FileSystemCatalog::new(options.clone()).unwrap(); - let cached_input = catalog.file_io().new_input(path.to_str().unwrap()).unwrap(); + let cached_input = catalog.file_io().new_input(&io_path).unwrap(); assert_eq!(cached_input.read().await.unwrap().as_ref(), b"abcdefgh"); std::fs::write(&path, b"12345678").unwrap(); assert_eq!(cached_input.read().await.unwrap().as_ref(), b"abcdefgh"); options.set(CatalogOptions::LOCAL_CACHE_EXCLUDE_EXTENSIONS, "blob"); let excluded_catalog = FileSystemCatalog::new(options).unwrap(); - let input = excluded_catalog - .file_io() - .new_input(path.to_str().unwrap()) - .unwrap(); + let input = excluded_catalog.file_io().new_input(&io_path).unwrap(); assert_eq!(input.read().await.unwrap().as_ref(), b"12345678"); assert_eq!( input @@ -238,10 +238,7 @@ async fn test_exclusions_bypass_existing_blocks_and_preserve_invalidation() { b"234567" ); - let output = excluded_catalog - .file_io() - .new_output(path.to_str().unwrap()) - .unwrap(); + let output = excluded_catalog.file_io().new_output(&io_path).unwrap(); output .write(bytes::Bytes::from_static(b"updated!")) .await @@ -255,7 +252,7 @@ async fn test_exclusions_bypass_existing_blocks_and_preserve_invalidation() { excluded_catalog .file_io() - .delete_file(path.to_str().unwrap()) + .delete_file(&io_path) .await .unwrap(); assert!(matches!(