From 41aaa67cce5265f60fa9bd050af74cf5df0858e6 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Wed, 21 Oct 2026 13:24:25 -0700 Subject: [PATCH 1/3] perf(core): reuse immutable ROW sidecar file sizes --- crates/paimon/src/io/file_io.rs | 139 ++++++++++++++++++++++++- crates/paimon/src/table/row_sidecar.rs | 4 +- 2 files changed, 140 insertions(+), 3 deletions(-) diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs index ab0934272..cc9e24e63 100644 --- a/crates/paimon/src/io/file_io.rs +++ b/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>, + process_id: u32, + row_sidecar_sizes: Mutex>, } impl std::fmt::Debug for FileFormatMetadataCacheContext { @@ -147,6 +152,10 @@ impl FileFormatMetadataCacheContext { Ok(Arc::new(Self { max_bytes, cache: OnceLock::new(), + process_id: std::process::id(), + row_sidecar_sizes: Mutex::new(LruCache::new( + NonZeroUsize::new(ROW_SIDECAR_SIZE_CACHE_MAX_ENTRIES).unwrap(), + )), })) } @@ -154,6 +163,41 @@ impl FileFormatMetadataCacheContext { self.max_bytes } + fn row_sidecar_size(&self, key: &str) -> Option { + if self.max_bytes == 0 || self.process_id != std::process::id() { + return None; + } + self.row_sidecar_sizes.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.lock().unwrap().put(key, size); + } + + fn invalidate_row_sidecar_size(&self, key: &str) { + if self.process_id == std::process::id() { + self.row_sidecar_sizes.lock().unwrap().pop(key); + } + } + + fn invalidate_row_sidecar_prefix(&self, prefix: &str) { + if self.process_id != std::process::id() { + return; + } + let mut sizes = self.row_sidecar_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(&self, init: impl FnOnce(usize) -> T) -> Arc where T: Send + Sync + 'static, @@ -496,6 +540,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 { + 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: @@ -651,6 +722,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 +755,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?; } @@ -2674,6 +2753,64 @@ 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(); + first + .new_output(path) + .unwrap() + .write(Bytes::from_static(b"first")) + .await + .unwrap(); + assert_eq!(first.immutable_row_sidecar_size(path).await.unwrap(), 5); + + // 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()); + } + #[tokio::test] async fn test_file_format_metadata_cache_size_reaches_file_reader() { let path = "memory:/metadata-cache-size.parquet"; diff --git a/crates/paimon/src/table/row_sidecar.rs b/crates/paimon/src/table/row_sidecar.rs index 624ca6ef7..a337b9c9d 100644 --- a/crates/paimon/src/table/row_sidecar.rs +++ b/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)) } From 624d140855a7e46b89cffaf65aad609f5e060c23 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Wed, 21 Oct 2026 16:19:51 -0700 Subject: [PATCH 2/3] test(core): count ROW sidecar stat cache hits --- crates/paimon/src/io/file_io.rs | 75 +++++++++++++++++++++++++++++++++ 1 file changed, 75 insertions(+) diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs index cc9e24e63..31f06d4a1 100644 --- a/crates/paimon/src/io/file_io.rs +++ b/crates/paimon/src/io/file_io.rs @@ -1677,6 +1677,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), @@ -1689,6 +1690,7 @@ mod file_action_test { struct CountingListService { pulls: Arc, fork_error: Option<&'static str>, + stat_response: Option<(Arc, u64)>, } impl Service for CountingListService { @@ -1729,6 +1731,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()) } @@ -1985,6 +1993,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))); @@ -2250,6 +2259,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)> { + 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( From 22c509fad0940cd8f5fbed0cbbb631d6edb39383 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Wed, 21 Oct 2026 19:35:11 -0700 Subject: [PATCH 3/3] perf(core): initialize ROW sidecar size cache on first use --- crates/paimon/src/io/file_io.rs | 52 ++++++++++++++++++++++++++++----- 1 file changed, 44 insertions(+), 8 deletions(-) diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs index 31f06d4a1..93b8107b3 100644 --- a/crates/paimon/src/io/file_io.rs +++ b/crates/paimon/src/io/file_io.rs @@ -121,7 +121,7 @@ pub(crate) struct FileFormatMetadataCacheContext { max_bytes: usize, cache: OnceLock>, process_id: u32, - row_sidecar_sizes: Mutex>, + row_sidecar_sizes: OnceLock>>, } impl std::fmt::Debug for FileFormatMetadataCacheContext { @@ -153,9 +153,7 @@ impl FileFormatMetadataCacheContext { max_bytes, cache: OnceLock::new(), process_id: std::process::id(), - row_sidecar_sizes: Mutex::new(LruCache::new( - NonZeroUsize::new(ROW_SIDECAR_SIZE_CACHE_MAX_ENTRIES).unwrap(), - )), + row_sidecar_sizes: OnceLock::new(), })) } @@ -167,19 +165,34 @@ impl FileFormatMetadataCacheContext { if self.max_bytes == 0 || self.process_id != std::process::id() { return None; } - self.row_sidecar_sizes.lock().unwrap().get(key).copied() + 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.lock().unwrap().put(key, size); + 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() { - self.row_sidecar_sizes.lock().unwrap().pop(key); + if let Some(sizes) = self.row_sidecar_sizes.get() { + sizes.lock().unwrap().pop(key); + } } } @@ -187,7 +200,10 @@ impl FileFormatMetadataCacheContext { if self.process_id != std::process::id() { return; } - let mut sizes = self.row_sidecar_sizes.lock().unwrap(); + 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)) @@ -2832,13 +2848,28 @@ mod input_output_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. @@ -2884,6 +2915,11 @@ mod input_output_test { 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]