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/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/generated.rs b/crates/paimon/src/io/cache/generated.rs index 3bb5845c8..968ecde55 100644 --- a/crates/paimon/src/io/cache/generated.rs +++ b/crates/paimon/src/io/cache/generated.rs @@ -192,7 +192,7 @@ pub(crate) fn colliding_cached_source( == head_key.components().next() }) .unwrap(); - assert!(cache.is_cacheable(&path)); + assert!(cache.is_whitelisted(&path) && !cache.excludes_extension(&path)); assert!(data.len() as u64 <= cache.generated_block_size()); Arc::new(super::reader::CachedFileReader::new( Arc::new(Source(data.clone())), diff --git a/crates/paimon/src/io/cache/mod.rs b/crates/paimon/src/io/cache/mod.rs index eb0fbf4de..415c64a89 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 @@ -71,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 { @@ -84,6 +86,7 @@ impl LocalCache { block_size, whitelist: FileType::parse_whitelist(whitelist), file_size_capacity: DEFAULT_FILE_SIZE_CAPACITY, + excluded_extensions: parse_excluded_extensions(excluded_extensions), fill_slots: Arc::new(tokio::sync::Semaphore::new(8)), _temporary_directory: None, }) @@ -134,6 +137,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 +156,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 +193,26 @@ 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 excludes_extension(&self, relative_path: &str) -> bool { + if self.excluded_extensions.is_empty() { + return false; + } + let path = opendal::raw::normalize_path(relative_path); + 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 +471,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,10 +555,25 @@ impl LocalCacheConfig { max_size, block_size, whitelist: FileType::parse_whitelist(whitelist), + 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() @@ -712,6 +749,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, @@ -735,8 +773,8 @@ mod tests { options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true"); options.set(CatalogOptions::LOCAL_CACHE_WHITELIST, "structured-data"); let cache = create_local_cache(&options).unwrap().unwrap(); - assert!(cache.is_cacheable("s3://bucket/data.parquet")); - assert!(cache.is_cacheable("s3://bucket/data.unknown")); + assert!(cache.is_whitelisted("s3://bucket/data.parquet")); + assert!(cache.is_whitelisted("s3://bucket/data.unknown")); for path in [ "s3://bucket/data.parquet.tmp", "s3://bucket/.data.parquet.123e4567-e89b-12d3-a456-426614174000.tmp", @@ -747,7 +785,7 @@ mod tests { "s3://bucket/data.video", "s3://bucket/data.vector.vortex", ] { - assert!(!cache.is_cacheable(path), "{path}"); + assert!(!cache.is_whitelisted(path), "{path}"); } } @@ -774,6 +812,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 +836,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 +861,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, @@ -829,12 +870,12 @@ mod tests { }) .unwrap(); - assert!(cache.is_cacheable("s3://bucket/table/snapshot/snapshot-1")); - assert!(!cache.is_cacheable("s3://bucket/table/data/data-1.parquet")); - assert!(!cache.is_cacheable("s3://bucket/table/snapshot/LATEST")); - assert!(!cache.is_cacheable("s3://bucket/table/tag/tag-production")); - assert!(cache.is_cacheable("s3://bucket/table/changelog/changelog-1")); - assert!(!cache.is_cacheable("s3://bucket/table/dt=1/_SUCCESS")); + assert!(cache.is_whitelisted("s3://bucket/table/snapshot/snapshot-1")); + assert!(!cache.is_whitelisted("s3://bucket/table/data/data-1.parquet")); + assert!(!cache.is_whitelisted("s3://bucket/table/snapshot/LATEST")); + assert!(!cache.is_whitelisted("s3://bucket/table/tag/tag-production")); + assert!(cache.is_whitelisted("s3://bucket/table/changelog/changelog-1")); + assert!(!cache.is_whitelisted("s3://bucket/table/dt=1/_SUCCESS")); } #[test] @@ -851,6 +892,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 +914,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..df7bf2144 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) } @@ -643,7 +666,7 @@ impl FileIO { cache: self .cache .as_ref() - .filter(|cache| cache.is_cacheable(path)) + .filter(|cache| cache.is_whitelisted(path)) .cloned(), }) } @@ -661,7 +684,7 @@ impl FileIO { cache: self .cache .as_ref() - .filter(|cache| cache.is_cacheable(path)) + .filter(|cache| cache.is_whitelisted(path)) .cloned(), }) } @@ -875,7 +898,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?; @@ -1773,6 +1798,12 @@ pub struct InputFile { } impl InputFile { + fn cache_for_read(&self, relative_path: &str) -> Option<&Arc> { + self.cache + .as_ref() + .filter(|cache| !cache.excludes_extension(relative_path)) + } + pub fn location(&self) -> &str { &self.path } @@ -1798,7 +1829,8 @@ impl InputFile { pub async fn read(&self) -> crate::Result { let (op, relative_path, cache_path) = self.source.resolve(&self.path).await?; - let (Some(cache), Some(cache_path)) = (&self.cache, cache_path) else { + let (Some(cache), Some(cache_path)) = (self.cache_for_read(&relative_path), cache_path) + else { return Ok(op.read(&relative_path).await?.to_bytes()); }; let read_token = cache.read_token(&cache_path); @@ -1840,7 +1872,9 @@ impl InputFile { } }); let reader = op.reader(&relative_path).await?; - let (Some(cache), Some(cache_path)) = (&self.cache, cache_path.as_ref()) else { + let (Some(cache), Some(cache_path)) = + (self.cache_for_read(&relative_path), cache_path.as_ref()) + else { return Ok(InputFileReader::Direct { reader, cache_key, @@ -1893,13 +1927,12 @@ impl OutputFile { } pub fn to_input_file(self) -> InputFile { - let cache = self.cache.filter(|cache| cache.is_cacheable(&self.path)); InputFile { source: self.source, path: self.path, file_format_metadata_cache: self.file_format_metadata_cache, blob_index_cache: self.blob_index_cache, - cache, + cache: self.cache, } } diff --git a/crates/paimon/src/io/file_io/provider_tests.rs b/crates/paimon/src/io/file_io/provider_tests.rs index cf949fd09..3c671a927 100644 --- a/crates/paimon/src/io/file_io/provider_tests.rs +++ b/crates/paimon/src/io/file_io/provider_tests.rs @@ -266,6 +266,39 @@ impl FileIOProvider for FixedPathProvider { async fn create(&self, _path: &str) -> Result<(Operator, String)> { Ok((self.op.clone(), self.relative.clone())) } + + async fn create_with_cache_namespace( + &self, + path: &str, + ) -> Result<(Operator, String, Option)> { + let (op, relative) = self.create(path).await?; + Ok((op, relative, Some("fixed".to_string()))) + } +} + +#[tokio::test] +async fn provider_cache_excludes_the_resolved_file_extension() { + use crate::io::cache::create_local_cache; + use crate::{CatalogOptions, Options}; + + let op = memory_operator(); + let mut options = Options::new(); + options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true"); + options.set(CatalogOptions::LOCAL_CACHE_WHITELIST, "data"); + options.set(CatalogOptions::LOCAL_CACHE_EXCLUDE_EXTENSIONS, "blob"); + let io = FileIOBuilder::new("unused") + .with_provider(Arc::new(FixedPathProvider { + op: op.clone(), + relative: "payload.blob".to_string(), + })) + .with_local_cache(create_local_cache(&options).unwrap().unwrap()) + .build() + .unwrap(); + let input = io.new_input("s3://bucket/logical.parquet").unwrap(); + for value in ["first", "later"] { + op.write("payload.blob", value).await.unwrap(); + assert_eq!(input.read().await.unwrap(), value); + } } #[tokio::test] 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..a823f909b --- /dev/null +++ b/crates/paimon/tests/local_cache_exclusions_test.rs @@ -0,0 +1,371 @@ +// 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 std::collections::HashMap; +use std::ops::Range; +use std::sync::{Arc, Mutex}; + +use bytes::Bytes; +use paimon::catalog::FileSystemCatalog; +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", "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, + (backend != "external").to_string(), + ); + options.set(CatalogOptions::LOCAL_CACHE_BLOCK_SIZE, "4"); + options.set(CatalogOptions::LOCAL_CACHE_WHITELIST, whitelist); + if matches!(backend, "disk" | "shared") { + 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(); + 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(); + // 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 = 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_excluded_extension_uses_the_opened_filesystem_path() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("payload.blob"); + let canonical = path.to_str().unwrap().replace('\\', "/"); + let alias = format!("{canonical} "); + 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_EXCLUDE_EXTENSIONS, "blob"); + let catalog = FileSystemCatalog::new(options).unwrap(); + let io = catalog.file_io(); + let canonical_input = io.new_input(&canonical).unwrap(); + let alias_input = io.new_input(&alias).unwrap(); + let output_input = io.new_output(&alias).unwrap().to_input_file(); + + for (full, range) in [(b"abcdefgh", b"bcdefg"), (b"12345678", b"234567")] { + std::fs::write(&path, full).unwrap(); + for input in [&canonical_input, &alias_input, &output_input] { + assert_eq!(input.read().await.unwrap().as_ref(), full); + assert_eq!( + input + .reader() + .await + .unwrap() + .read(1..7) + .await + .unwrap() + .as_ref(), + range + ); + } + } +} + +#[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 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('\\', "/"); + 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, + (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, + (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 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 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 + .reader() + .await + .unwrap() + .read(1..7) + .await + .unwrap() + .as_ref(), + b"234567" + ); + + let output = excluded_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_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/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 5bcce1a87..dc0839ee1 100644 --- a/docs/src/getting-started.md +++ b/docs/src/getting-started.md @@ -445,11 +445,29 @@ 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 normalized storage path that FileIO reads, ignoring ASCII case and an +optional leading dot in each entry. For example, a filesystem path ending in +`payload.blob ` opens `payload.blob` and is also excluded. +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. + +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.