From dbfc09a6c5f4fd6c16a1cabce89f4a9ce7b6066d Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Sat, 31 Oct 2026 17:53:07 +0800 Subject: [PATCH 1/3] 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/3] 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!( From c089978b57ffa7e8d55c23f22673f7121d69d55b Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Tue, 3 Nov 2026 17:05:09 +0800 Subject: [PATCH 3/3] fix(core): apply extension exclusions to external caches Add an opt-in Rust cache attachment method and pass extension exclusions from C options without changing existing Rust or C signatures. Reuse the local-cache parser and cover external cache bypass, default compatibility, and write/delete invalidation in Rust and C tests. Signed-off-by: QuakeWang --- bindings/c/src/file_io.rs | 28 ++++-- bindings/c/src/tests.rs | 81 ++++++++++++++- crates/paimon/src/io/cache/mod.rs | 27 +++-- crates/paimon/src/io/file_io.rs | 25 ++++- .../tests/local_cache_exclusions_test.rs | 98 ++++++++++++++++--- docs/src/c-binding.md | 7 ++ docs/src/getting-started.md | 4 + 7 files changed, 236 insertions(+), 34 deletions(-) diff --git a/bindings/c/src/file_io.rs b/bindings/c/src/file_io.rs index 7f0d4901b..795375450 100644 --- a/bindings/c/src/file_io.rs +++ b/bindings/c/src/file_io.rs @@ -22,6 +22,7 @@ use std::sync::Arc; use bytes::Bytes; use paimon::io::{FileBlockCache, FileIO}; +use paimon::CatalogOptions; use crate::error::{paimon_error, validate_cstr, PaimonErrorCode}; use crate::result::paimon_result_file_io_new; @@ -169,11 +170,9 @@ unsafe fn parse_options( unsafe fn create_file_io( path: *const c_char, - options: *const paimon_option, - options_len: usize, + options: HashMap, ) -> Result { let path = validate_cstr(path, "path")?; - let options = parse_options(options, options_len)?; FileIO::from_path(path) .and_then(|builder| builder.with_props(options).build()) .map_err(paimon_error::from_paimon) @@ -201,7 +200,9 @@ pub unsafe extern "C" fn paimon_file_io_create( options: *const paimon_option, options_len: usize, ) -> paimon_result_file_io_new { - file_io_result(create_file_io(path, options, options_len)) + file_io_result( + parse_options(options, options_len).and_then(|options| create_file_io(path, options)), + ) } /// Create a reusable FileIO backed by a caller-managed block cache. @@ -210,6 +211,8 @@ pub unsafe extern "C" fn paimon_file_io_create( /// `whitelist` may be null to use `meta,global-index`. Once validation and /// storage construction succeed, Rust owns `callbacks->context` and invokes /// `destroy` exactly once after the last derived FileIO/table is dropped. +/// `options` may include `local-cache.exclude-extensions` to bypass cache reads +/// and fills for selected extensions while retaining write/delete invalidation. #[no_mangle] pub unsafe extern "C" fn paimon_file_io_create_with_cache_v1( path: *const c_char, @@ -240,7 +243,15 @@ pub unsafe extern "C" fn paimon_file_io_create_with_cache_v1( Err(error) => return file_io_result(Err(error)), } }; - let file_io = match create_file_io(path, options, options_len) { + let options = match parse_options(options, options_len) { + Ok(options) => options, + Err(error) => return file_io_result(Err(error)), + }; + let excluded_extensions = options + .get(CatalogOptions::LOCAL_CACHE_EXCLUDE_EXTENSIONS) + .cloned() + .unwrap_or_default(); + let file_io = match create_file_io(path, options) { Ok(file_io) => file_io, Err(error) => return file_io_result(Err(error)), }; @@ -255,7 +266,12 @@ pub unsafe extern "C" fn paimon_file_io_create_with_cache_v1( }; file_io_result( file_io - .with_file_block_cache(cache, block_size, &whitelist) + .with_file_block_cache_and_exclusions( + cache, + block_size, + &whitelist, + &excluded_extensions, + ) .map_err(paimon_error::from_paimon), ) } diff --git a/bindings/c/src/tests.rs b/bindings/c/src/tests.rs index 108078d4e..de46c773a 100644 --- a/bindings/c/src/tests.rs +++ b/bindings/c/src/tests.rs @@ -41,7 +41,7 @@ use arrow_array::ffi::{FFI_ArrowArray, FFI_ArrowSchema}; use arrow_array::{Array, Int32Array, RecordBatch, StringArray, StructArray}; use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema}; use paimon::catalog::Identifier; -use paimon::io::FileIOBuilder; +use paimon::io::{FileIOBuilder, FileRead}; use paimon::spec::{ BlobDescriptor, CommitKind, DataType, IntType, Schema, TableSchema, VarCharType, }; @@ -576,6 +576,85 @@ fn test_file_io_external_cache_miss_populates_and_hit_reuses_blocks() { } } +#[test] +fn test_file_io_external_cache_extension_exclusions() { + for exclusions in [None, Some(""), Some(" , "), Some(" .BLOB, video, ")] { + for name in ["data.blob", "data.parquet", "snapshot-1.blob"] { + for range_read in [false, true] { + let state = Arc::new(TestFileCacheState::default()); + let callbacks = test_file_cache_callbacks(state.clone()); + let root = CString::new("memory:/ffi_cache_exclusions").unwrap(); + let whitelist = CString::new("meta,data").unwrap(); + let key = + CString::new(paimon::CatalogOptions::LOCAL_CACHE_EXCLUDE_EXTENSIONS).unwrap(); + let value = exclusions.map(|value| CString::new(value).unwrap()); + let options: Vec<_> = value + .iter() + .map(|value| paimon_option { + key: key.as_ptr(), + value: value.as_ptr(), + }) + .collect(); + let excluded = exclusions == Some(" .BLOB, video, ") && name.ends_with(".blob"); + + unsafe { + let result = paimon_file_io_create_with_cache_v1( + root.as_ptr(), + options.as_ptr(), + options.len(), + &callbacks, + 4, + whitelist.as_ptr(), + ); + assert!(result.error.is_null()); + let file_io = file_io_ref(result.file_io); + let path = format!("memory:/ffi_cache_exclusions/{name}"); + crate::runtime().block_on(async { + let output = file_io.new_output(&path).unwrap(); + output + .write(bytes::Bytes::from_static(b"abcdefgh")) + .await + .unwrap(); + let input = file_io.new_input(&path).unwrap(); + for _ in 0..2 { + let actual = if range_read { + input.reader().await.unwrap().read(1..7).await.unwrap() + } else { + input.read().await.unwrap() + }; + let expected: &[u8] = if range_read { b"bcdefg" } else { b"abcdefgh" }; + assert_eq!(actual.as_ref(), expected); + } + assert_eq!( + state.puts.load(Ordering::SeqCst), + if excluded { 0 } else { 2 }, + "exclusions={exclusions:?}, name={name}, range_read={range_read}" + ); + assert_eq!( + state.hits.load(Ordering::SeqCst), + if excluded { 0 } else { 2 } + ); + if excluded { + assert_eq!(state.misses.load(Ordering::SeqCst), 0); + } + assert_eq!(state.invalidations.load(Ordering::SeqCst), 1); + output + .write(bytes::Bytes::from_static(b"updated!")) + .await + .unwrap(); + assert_eq!(input.read().await.unwrap().as_ref(), b"updated!"); + assert_eq!(state.invalidations.load(Ordering::SeqCst), 2); + file_io.delete_file(&path).await.unwrap(); + assert_eq!(state.invalidations.load(Ordering::SeqCst), 3); + }); + paimon_file_io_free(result.file_io); + assert_eq!(state.destroys.load(Ordering::SeqCst), 1); + } + } + } + } +} + #[test] fn test_table_with_file_io_retains_cache_context_until_table_is_freed() { let state = Arc::new(TestFileCacheState::default()); diff --git a/crates/paimon/src/io/cache/mod.rs b/crates/paimon/src/io/cache/mod.rs index 0bc7cfd8d..96def4f0d 100644 --- a/crates/paimon/src/io/cache/mod.rs +++ b/crates/paimon/src/io/cache/mod.rs @@ -72,6 +72,7 @@ impl LocalCache { cache: Arc, block_size: u64, whitelist: &str, + excluded_extensions: &str, ) -> crate::Result { if block_size == 0 { return Err(crate::Error::ConfigInvalid { @@ -85,7 +86,7 @@ impl LocalCache { block_size, whitelist: FileType::parse_whitelist(whitelist), file_size_capacity: DEFAULT_FILE_SIZE_CAPACITY, - excluded_extensions: HashSet::new(), + excluded_extensions: parse_excluded_extensions(excluded_extensions), fill_slots: Arc::new(tokio::sync::Semaphore::new(8)), _temporary_directory: None, }) @@ -556,19 +557,25 @@ 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(), + excluded_extensions: parse_excluded_extensions( + options + .get(CatalogOptions::LOCAL_CACHE_EXCLUDE_EXTENSIONS) + .map(String::as_str) + .unwrap_or_default(), + ), })) } } +fn parse_excluded_extensions(value: &str) -> HashSet { + value + .split(',') + .map(|extension| extension.trim().trim_start_matches('.')) + .filter(|extension| !extension.is_empty()) + .map(str::to_ascii_lowercase) + .collect() +} + fn catalog_namespace(options: &Options) -> String { let mut entries = options .to_map() diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs index 4956d34c1..b9ed4d8b1 100644 --- a/crates/paimon/src/io/file_io.rs +++ b/crates/paimon/src/io/file_io.rs @@ -444,14 +444,37 @@ impl FileIO { /// `whitelist` uses the same comma-separated values as /// `local-cache.whitelist`: `meta`, `global-index`, `bucket-index`, `data`, /// `structured-data`, and `file-index`, or `*` for all of them. + /// No extensions are excluded. Use [`Self::with_file_block_cache_and_exclusions`] + /// to exclude selected file extensions. pub fn with_file_block_cache( + self, + cache: Arc, + block_size: u64, + whitelist: &str, + ) -> crate::Result { + self.with_file_block_cache_and_exclusions(cache, block_size, whitelist, "") + } + + /// Attach an externally managed block cache with extension exclusions. + /// + /// `block_size` and `whitelist` have the same meaning as in + /// [`Self::with_file_block_cache`]. `excluded_extensions` uses the + /// `local-cache.exclude-extensions` rules: comma-separated last extensions, + /// ignoring ASCII case, surrounding whitespace, optional leading dots, and + /// empty entries. An empty list preserves the whitelist behavior. + /// Exclusions bypass cache reads and fills, but retain write/delete invalidation. + pub fn with_file_block_cache_and_exclusions( mut self, cache: Arc, block_size: u64, whitelist: &str, + excluded_extensions: &str, ) -> crate::Result { self.cache = Some(Arc::new(LocalCache::external( - cache, block_size, whitelist, + cache, + block_size, + whitelist, + excluded_extensions, )?)); Ok(self) } diff --git a/crates/paimon/tests/local_cache_exclusions_test.rs b/crates/paimon/tests/local_cache_exclusions_test.rs index c8c115298..3395a91d8 100644 --- a/crates/paimon/tests/local_cache_exclusions_test.rs +++ b/crates/paimon/tests/local_cache_exclusions_test.rs @@ -17,27 +17,68 @@ #![cfg(feature = "storage-fs")] +use std::collections::HashMap; +use std::ops::Range; +use std::sync::{Arc, Mutex}; + +use bytes::Bytes; use paimon::catalog::FileSystemCatalog; -use paimon::io::FileRead; +use paimon::io::{FileBlockCache, FileRead}; use paimon::{CatalogOptions, Error, Options}; +#[derive(Debug, Default)] +struct TestBlockCache(Mutex>); + +#[async_trait::async_trait] +impl FileBlockCache for TestBlockCache { + async fn get(&self, path: &str, range: Range) -> Option { + self.0 + .lock() + .unwrap() + .get(&(path.to_string(), range.start)) + .filter(|data| data.len() as u64 == range.end - range.start) + .cloned() + } + + async fn put(&self, path: &str, offset: u64, data: Bytes) { + self.0 + .lock() + .unwrap() + .insert((path.to_string(), offset), data); + } + + async fn invalidate_path(&self, path: &str) { + self.0.lock().unwrap().retain(|(key, _), _| key != path); + } + + async fn invalidate_prefix(&self, prefix: &str) { + self.0 + .lock() + .unwrap() + .retain(|(key, _), _| !key.starts_with(prefix)); + } +} + 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 backend in ["memory", "disk", "shared", "external"] { 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_ENABLED, + (backend != "external").to_string(), + ); options.set(CatalogOptions::LOCAL_CACHE_BLOCK_SIZE, "4"); options.set(CatalogOptions::LOCAL_CACHE_WHITELIST, whitelist); - if backend != "memory" { + if matches!(backend, "disk" | "shared") { options.set( CatalogOptions::LOCAL_CACHE_DIR, directory.path().join("cache").to_string_lossy(), @@ -51,6 +92,16 @@ async fn assert_cache_selection( options.set(CatalogOptions::LOCAL_CACHE_EXCLUDE_EXTENSIONS, extensions); } let catalog = FileSystemCatalog::new(options).unwrap(); + let mut file_io = catalog.file_io().clone(); + if backend == "external" { + let cache = Arc::new(TestBlockCache::default()); + file_io = match excluded_extensions { + Some(exclusions) => file_io + .with_file_block_cache_and_exclusions(cache, 4, whitelist, exclusions), + None => file_io.with_file_block_cache(cache, 4, whitelist), + } + .unwrap(); + } for &(name, expected_cached) in files { let path = warehouse.join(name); std::fs::create_dir_all(path.parent().unwrap()).unwrap(); @@ -58,7 +109,7 @@ async fn assert_cache_selection( std::fs::write(&path, b"abcdefgh").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 input = 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() @@ -197,7 +248,7 @@ async fn test_exclusions_do_not_expand_the_whitelist() { #[tokio::test] async fn test_exclusions_bypass_existing_blocks_and_preserve_invalidation() { - for shared in [false, true] { + for backend in ["disk", "shared", "external"] { let directory = tempfile::tempdir().unwrap(); let path = directory.path().join("data.blob"); let io_path = path.to_str().unwrap().replace('\\', "/"); @@ -207,24 +258,43 @@ async fn test_exclusions_bypass_existing_blocks_and_preserve_invalidation() { CatalogOptions::WAREHOUSE, directory.path().to_string_lossy(), ); - options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true"); + options.set( + CatalogOptions::LOCAL_CACHE_ENABLED, + (backend != "external").to_string(), + ); 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_SHARED, + (backend == "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(&io_path).unwrap(); + let external_cache = Arc::new(TestBlockCache::default()); + let mut file_io = catalog.file_io().clone(); + if backend == "external" { + file_io = file_io + .with_file_block_cache(external_cache.clone(), 4, "data") + .unwrap(); + } + let cached_input = 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(&io_path).unwrap(); + let mut excluded_io = excluded_catalog.file_io().clone(); + if backend == "external" { + excluded_io = excluded_io + .with_file_block_cache_and_exclusions(external_cache, 4, "data", "blob") + .unwrap(); + } + let input = excluded_io.new_input(&io_path).unwrap(); assert_eq!(input.read().await.unwrap().as_ref(), b"12345678"); assert_eq!( input @@ -238,7 +308,7 @@ async fn test_exclusions_bypass_existing_blocks_and_preserve_invalidation() { b"234567" ); - let output = excluded_catalog.file_io().new_output(&io_path).unwrap(); + let output = excluded_io.new_output(&io_path).unwrap(); output .write(bytes::Bytes::from_static(b"updated!")) .await @@ -250,11 +320,7 @@ async fn test_exclusions_bypass_existing_blocks_and_preserve_invalidation() { b"changed!" ); - excluded_catalog - .file_io() - .delete_file(&io_path) - .await - .unwrap(); + excluded_io.delete_file(&io_path).await.unwrap(); assert!(matches!( cached_input.read().await, Err(Error::IoUnexpected { source, .. }) diff --git a/docs/src/c-binding.md b/docs/src/c-binding.md index b32a739ea..ea65a250c 100644 --- a/docs/src/c-binding.md +++ b/docs/src/c-binding.md @@ -216,6 +216,13 @@ CHECK_RESULT(table_result); paimon_file_io_free(io_result.file_io); ``` +To cache data while excluding BLOB bodies, pass `meta,global-index,data` as +`whitelist` and include `local-cache.exclude-extensions=blob` in `storage_options`. +The exclusion uses the same extension matching rules as the catalog local cache +and does not require `local-cache.enabled=true`. An omitted or empty list keeps +the existing behavior. Excluded files bypass `get` and `put`, but writes and +deletes still invoke invalidation callbacks for whitelisted files. + `get` receives an output buffer with exactly `length` writable bytes. It returns the number of bytes copied, `-1` for a miss, or any other negative value for a fail-open error. A nonnegative value different from `length` is also treated as diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md index bba456ce7..6908ee22a 100644 --- a/docs/src/getting-started.md +++ b/docs/src/getting-started.md @@ -462,6 +462,10 @@ 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. +For caller-managed Rust block caches, use +`FileIO::with_file_block_cache_and_exclusions` to supply the same exclusion list. +The existing `with_file_block_cache` method continues to use an empty list. + 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.