diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs index ab0934272..93b8107b3 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: OnceLock>>, } impl std::fmt::Debug for FileFormatMetadataCacheContext { @@ -147,6 +152,8 @@ impl FileFormatMetadataCacheContext { Ok(Arc::new(Self { max_bytes, cache: OnceLock::new(), + process_id: std::process::id(), + row_sidecar_sizes: OnceLock::new(), })) } @@ -154,6 +161,59 @@ 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 + .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(&self, init: impl FnOnce(usize) -> T) -> Arc 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 { + 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 +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, fork_error: Option<&'static str>, + stat_response: Option<(Arc, 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)> { + 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"; 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)) }