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..c8c115298 --- /dev/null +++ b/crates/paimon/tests/local_cache_exclusions_test.rs @@ -0,0 +1,266 @@ +// 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(); + // 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() + } 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"); + let io_path = path.to_str().unwrap().replace('\\', "/"); + 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(&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(); + 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(&io_path).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(&io_path) + .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.