feat(core): support local-cache extension exclusions

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<String, String>,
 ) -> Result<FileIO, *mut paimon_error> {
     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),
     )
 }

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());

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";

crates/paimon/src/io/cache/mod.rs

@@ -51,6 +51,7 @@ pub(crate) struct LocalCache {
     block_size: u64,
     whitelist: HashSet<FileType>,
     file_size_capacity: usize,
+    excluded_extensions: HashSet<String>,
     // Bound queued and active background fills.
     fill_slots: Arc<tokio::sync::Semaphore>,
     // Generated lookup reads can outlive the executor. Keep temporary disk
@@ -71,6 +72,7 @@ impl LocalCache {
         cache: Arc<dyn FileBlockCache>,
         block_size: u64,
         whitelist: &str,
+        excluded_extensions: &str,
     ) -> crate::Result<Self> {
         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,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 +473,7 @@ pub(crate) async fn create_local_cache_with_namespace_async(
 #[derive(Debug)]
 pub(crate) struct LocalCacheConfig {
     shared: bool,
+    excluded_extensions: HashSet<String>,
     dir: Option<PathBuf>,
     namespace: String,
     max_size: Option<u64>,
@@ -533,10 +557,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<String> {
+    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 +751,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 +814,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 +838,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 +863,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 +894,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 +916,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),