perf(core): reuse immutable ROW sidecar file sizes

crates/paimon/src/io/file_io.rs

@@ -17,12 +17,14 @@
 
 use crate::common::options::{parse_memory_size, CatalogOptions};
 use crate::error::*;
+use lru::LruCache;
 use std::any::Any;
 use std::collections::HashMap;
 use std::future::Future;
+use std::num::NonZeroUsize;
 use std::ops::Range;
 use std::pin::Pin;
-use std::sync::{Arc, OnceLock};
+use std::sync::{Arc, Mutex, OnceLock};
 use std::task::{Context, Poll};
 use std::time::SystemTime;
 
@@ -113,10 +115,13 @@ pub struct FileIO {
 
 pub(crate) const DEFAULT_FILE_FORMAT_METADATA_CACHE_MAX_BYTES: usize = 50 * 1024 * 1024;
 pub(crate) const DEFAULT_BLOB_INDEX_CACHE_MAX_BYTES: usize = 64 * 1024 * 1024;
+const ROW_SIDECAR_SIZE_CACHE_MAX_ENTRIES: usize = 4096;
 
 pub(crate) struct FileFormatMetadataCacheContext {
     max_bytes: usize,
     cache: OnceLock<Arc<dyn Any + Send + Sync>>,
+    process_id: u32,
+    row_sidecar_sizes: OnceLock<Mutex<LruCache<String, u64>>>,
 }
 
 impl std::fmt::Debug for FileFormatMetadataCacheContext {
@@ -147,13 +152,68 @@ impl FileFormatMetadataCacheContext {
         Ok(Arc::new(Self {
             max_bytes,
             cache: OnceLock::new(),
+            process_id: std::process::id(),
+            row_sidecar_sizes: OnceLock::new(),
         }))
     }
 
     pub(crate) fn max_bytes(&self) -> usize {
         self.max_bytes
     }
 
+    fn row_sidecar_size(&self, key: &str) -> Option<u64> {
+        if self.max_bytes == 0 || self.process_id != std::process::id() {
+            return None;
+        }
+        self.row_sidecar_sizes
+            .get()?
+            .lock()
+            .unwrap()
+            .get(key)
+            .copied()
+    }
+
+    fn cache_row_sidecar_size(&self, key: String, size: u64) {
+        if self.max_bytes == 0 || self.process_id != std::process::id() {
+            return;
+        }
+        self.row_sidecar_sizes
+            .get_or_init(|| {
+                Mutex::new(LruCache::new(
+                    NonZeroUsize::new(ROW_SIDECAR_SIZE_CACHE_MAX_ENTRIES).unwrap(),
+                ))
+            })
+            .lock()
+            .unwrap()
+            .put(key, size);
+    }
+
+    fn invalidate_row_sidecar_size(&self, key: &str) {
+        if self.process_id == std::process::id() {
+            if let Some(sizes) = self.row_sidecar_sizes.get() {
+                sizes.lock().unwrap().pop(key);
+            }
+        }
+    }
+
+    fn invalidate_row_sidecar_prefix(&self, prefix: &str) {
+        if self.process_id != std::process::id() {
+            return;
+        }
+        let Some(sizes) = self.row_sidecar_sizes.get() else {
+            return;
+        };
+        let mut sizes = sizes.lock().unwrap();
+        let keys: Vec<_> = sizes
+            .iter()
+            .filter(|(key, _)| key.starts_with(prefix))
+            .map(|(key, _)| key.clone())
+            .collect();
+        for key in keys {
+            sizes.pop(&key);
+        }
+    }
+
     pub(crate) fn get_or_init<T>(&self, init: impl FnOnce(usize) -> T) -> Arc<T>
     where
         T: Send + Sync + 'static,
@@ -496,6 +556,33 @@ impl FileIO {
         })
     }
 
+    /// Return the size of an immutable ROW sidecar without repeating remote
+    /// status requests for subsequent reads through the same cache context.
+    pub(crate) async fn immutable_row_sidecar_size(&self, path: &str) -> Result<u64> {
+        let (op, relative_path, cache_key) = self.create_with_cache_path(path).await?;
+        if let Some(size) = cache_key
+            .as_deref()
+            .and_then(|key| self.file_format_metadata_cache.row_sidecar_size(key))
+        {
+            return Ok(size);
+        }
+        let size = op
+            .stat(relative_path.as_ref())
+            .await
+            .map_err(|error| {
+                Error::from_opendal_with_context(
+                    error,
+                    format!("Failed to get file status for '{path}'"),
+                )
+            })?
+            .content_length();
+        if let Some(key) = cache_key {
+            self.file_format_metadata_cache
+                .cache_row_sidecar_size(key, size);
+        }
+        Ok(size)
+    }
+
     /// List the statuses of the files/directories in the given path if the path is a directory.
     ///
     /// References: <https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java#L105>
@@ -651,6 +738,10 @@ impl FileIO {
         op.delete(relative_path.as_ref()).await.map_err(|error| {
             Error::from_opendal_with_context(error, format!("Failed to delete file '{path}'"))
         })?;
+        if let Some(ref cache_path) = cache_path {
+            self.file_format_metadata_cache
+                .invalidate_row_sidecar_size(cache_path);
+        }
         if let (Some(cache), Some(cache_path)) = (
             self.cache.as_ref().filter(|cache| cache.is_cacheable(path)),
             cache_path,
@@ -680,6 +771,10 @@ impl FileIO {
                     format!("Failed to delete directory '{path}'"),
                 )
             })?;
+        if let Some(ref cache_path) = cache_path {
+            self.file_format_metadata_cache
+                .invalidate_row_sidecar_prefix(cache_path);
+        }
         if let (Some(cache), Some(cache_path)) = (&self.cache, cache_path) {
             cache.invalidate_prefix(&cache_path).await?;
         }
@@ -1598,6 +1693,7 @@ mod file_action_test {
             let service: Servicer = Arc::new(CountingListService {
                 pulls: Arc::clone(&self.pulls),
                 fork_error: None,
+                stat_response: None,
             });
             Ok((
                 Operator::from_parts(OperationContext::default(), service),
@@ -1610,6 +1706,7 @@ mod file_action_test {
     struct CountingListService {
         pulls: Arc<AtomicUsize>,
         fork_error: Option<&'static str>,
+        stat_response: Option<(Arc<AtomicUsize>, u64)>,
     }
 
     impl Service for CountingListService {
@@ -1650,6 +1747,12 @@ mod file_action_test {
             if let Some(message) = self.fork_error {
                 return Err(fork_test_error(message));
             }
+            if let Some((calls, size)) = &self.stat_response {
+                calls.fetch_add(1, AtomicOrdering::SeqCst);
+                return Ok(RpStat::new(
+                    Metadata::new(EntryMode::FILE).with_content_length(*size),
+                ));
+            }
             Err(unsupported_test_operation())
         }
 
@@ -1906,6 +2009,7 @@ mod file_action_test {
             let service: Servicer = Arc::new(CountingListService {
                 pulls: Arc::new(AtomicUsize::new(0)),
                 fork_error: Some(message),
+                stat_response: None,
             });
             let operator = Operator::from_parts(OperationContext::default(), service);
             let file_io = setup_memory_file_io().with_provider(Arc::new(FixedProvider(operator)));
@@ -2171,6 +2275,72 @@ mod file_action_test {
         assert!(dir.path().join("t1").exists());
         assert!(!dir.path().join("t1 ").exists());
     }
+
+    #[tokio::test]
+    async fn test_immutable_row_sidecar_size_cache_avoids_repeated_stat() {
+        #[derive(Debug)]
+        struct CountingStatProvider(Operator);
+
+        #[async_trait::async_trait]
+        impl FileIOProvider for CountingStatProvider {
+            async fn create(&self, path: &str) -> crate::Result<(Operator, String)> {
+                Ok((
+                    self.0.clone(),
+                    path.strip_prefix("counting:/").unwrap().to_string(),
+                ))
+            }
+
+            async fn create_with_cache_namespace(
+                &self,
+                path: &str,
+            ) -> crate::Result<(Operator, String, Option<String>)> {
+                let (op, relative_path) = self.create(path).await?;
+                Ok((op, relative_path, Some("counting-stat".to_string())))
+            }
+        }
+
+        let calls = Arc::new(AtomicUsize::new(0));
+        let service: Servicer = Arc::new(CountingListService {
+            pulls: Arc::new(AtomicUsize::new(0)),
+            fork_error: None,
+            stat_response: Some((Arc::clone(&calls), 123)),
+        });
+        let provider = Arc::new(CountingStatProvider(Operator::from_parts(
+            OperationContext::default(),
+            service,
+        )));
+        let path = "counting:/objects/sidecar.row";
+
+        let cached = FileIOBuilder::new("memory")
+            .with_provider(provider.clone())
+            .build()
+            .unwrap();
+        assert_eq!(cached.immutable_row_sidecar_size(path).await.unwrap(), 123);
+        assert_eq!(
+            cached
+                .clone()
+                .immutable_row_sidecar_size(path)
+                .await
+                .unwrap(),
+            123
+        );
+        assert_eq!(calls.load(AtomicOrdering::SeqCst), 1);
+
+        let uncached = FileIOBuilder::new("memory")
+            .with_provider(provider)
+            .with_prop(CatalogOptions::FILE_FORMAT_METADATA_CACHE_MAX_SIZE, "0")
+            .build()
+            .unwrap();
+        assert_eq!(
+            uncached.immutable_row_sidecar_size(path).await.unwrap(),
+            123
+        );
+        assert_eq!(
+            uncached.immutable_row_sidecar_size(path).await.unwrap(),
+            123
+        );
+        assert_eq!(calls.load(AtomicOrdering::SeqCst), 3);
+    }
 }
 
 #[cfg(all(
@@ -2674,6 +2844,84 @@ mod input_output_test {
         assert!(!std::ptr::eq(first_cache, second_cache));
     }
 
+    #[tokio::test]
+    async fn test_immutable_row_sidecar_size_cache_is_scoped_and_disabled_with_metadata_cache() {
+        let path = "memory:/row-sidecar-size-cache.row";
+        let first = setup_memory_file_io();
+        assert!(first
+            .file_format_metadata_cache
+            .row_sidecar_sizes
+            .get()
+            .is_none());
+        first
+            .new_output(path)
+            .unwrap()
+            .write(Bytes::from_static(b"first"))
+            .await
+            .unwrap();
+        assert!(first
+            .file_format_metadata_cache
+            .row_sidecar_sizes
+            .get()
+            .is_none());
+        assert_eq!(first.immutable_row_sidecar_size(path).await.unwrap(), 5);
+        assert!(first
+            .file_format_metadata_cache
+            .row_sidecar_sizes
+            .get()
+            .is_some());
+
+        // The same context reuses the size. A different FileIO with the same
+        // literal path has its own storage target and must not reuse it.
+        assert_eq!(
+            first
+                .clone()
+                .immutable_row_sidecar_size(path)
+                .await
+                .unwrap(),
+            5
+        );
+        assert!(setup_memory_file_io()
+            .immutable_row_sidecar_size(path)
+            .await
+            .is_err());
+        first.delete_file(path).await.unwrap();
+        assert!(first.immutable_row_sidecar_size(path).await.is_err());
+
+        let nested = "memory:/row-sidecar-size-prefix/sidecar.row";
+        first
+            .new_output(nested)
+            .unwrap()
+            .write(Bytes::from_static(b"nested"))
+            .await
+            .unwrap();
+        assert_eq!(first.immutable_row_sidecar_size(nested).await.unwrap(), 6);
+        first
+            .delete_dir("memory:/row-sidecar-size-prefix")
+            .await
+            .unwrap();
+        assert!(first.immutable_row_sidecar_size(nested).await.is_err());
+
+        let uncached = FileIOBuilder::new("memory")
+            .with_prop(CatalogOptions::FILE_FORMAT_METADATA_CACHE_MAX_SIZE, "0")
+            .build()
+            .unwrap();
+        uncached
+            .new_output(path)
+            .unwrap()
+            .write(Bytes::from_static(b"data"))
+            .await
+            .unwrap();
+        assert_eq!(uncached.immutable_row_sidecar_size(path).await.unwrap(), 4);
+        uncached.delete_file(path).await.unwrap();
+        assert!(uncached.immutable_row_sidecar_size(path).await.is_err());
+        assert!(uncached
+            .file_format_metadata_cache
+            .row_sidecar_sizes
+            .get()
+            .is_none());
+    }
+
     #[tokio::test]
     async fn test_file_format_metadata_cache_size_reaches_file_reader() {
         let path = "memory:/metadata-cache-size.parquet";

crates/paimon/src/table/row_sidecar.rs

@@ -242,8 +242,8 @@ pub(super) async fn read_target(
 ) -> Result<(String, u64)> {
     if let Some(name) = selected_sidecar(file, local_ranges, options)? {
         let path = file.aligned_file_path(bucket_path, name);
-        let status = io.get_status(&path).await?;
-        return Ok((path, status.size));
+        let size = io.immutable_row_sidecar_size(&path).await?;
+        return Ok((path, size));
     }
     Ok((file.data_file_path(bucket_path), file.file_size as u64))
 }