From 41aaa67cce5265f60fa9bd050af74cf5df0858e6 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Wed, 21 Oct 2026 13:24:25 -0700 Subject: [PATCH 1/5] 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/5] 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 ad1c20759077ef4b5590ca3a4d63ef912f7cc88a Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Wed, 21 Oct 2026 17:05:47 -0700 Subject: [PATCH 3/5] perf(core): cache ROW footer and block index metadata by default --- crates/paimon/src/arrow/format/row.rs | 275 +++++++++++++++++++++----- crates/paimon/src/io/file_io.rs | 36 ++++ 2 files changed, 258 insertions(+), 53 deletions(-) diff --git a/crates/paimon/src/arrow/format/row.rs b/crates/paimon/src/arrow/format/row.rs index ff53b619f..b984ddd3d 100644 --- a/crates/paimon/src/arrow/format/row.rs +++ b/crates/paimon/src/arrow/format/row.rs @@ -20,12 +20,13 @@ //! Layout reference: //! `org.apache.paimon.format.row.RowFormatWriter` in paimon-java. +use super::metadata_cache::FileMetadataCache; use super::{FilePredicates, FormatFileReader, FormatFileWriter, FormatWriteResult}; use crate::arrow::{ arrow_to_paimon_type, build_target_arrow_schema, is_variant_arrow_fields, paimon_type_to_arrow, variant_arrow_type, }; -use crate::io::{FileRead, FileWrite, OutputFile}; +use crate::io::{FileFormatMetadataCacheContext, FileRead, FileWrite, OutputFile}; use crate::spec::{DataField, DataType, IntType, VarBinaryType, VariantType}; use crate::table::{ArrowRecordBatchStream, RowRange}; use crate::Error; @@ -71,6 +72,57 @@ const ROW_BLOCK_READ_CONCURRENCY: usize = 8; const FOOTER_SIZE: u64 = 32; const ROW_MAGIC: i32 = 0x524F5753; const ROW_VERSION: u8 = 1; +const ROW_METADATA_CACHE_MAX_ENTRIES: usize = 4096; + +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +struct RowMetadataCacheKey { + file: String, + size: u64, +} + +struct RowFileMetadata { + index: RowBlockIndex, + total_rows: usize, +} + +struct RowMetadataCache { + inner: FileMetadataCache, +} + +impl RowMetadataCache { + fn new(max_bytes: usize) -> Self { + Self { + inner: FileMetadataCache::new(max_bytes, ROW_METADATA_CACHE_MAX_ENTRIES), + } + } + + async fn load( + &self, + reader: &dyn FileRead, + file_size: u64, + ) -> crate::Result> { + let key = reader + .cache_key() + .filter(|key| !key.is_empty()) + .map(|file| RowMetadataCacheKey { + file: file.to_string(), + // ROW sidecars are immutable; size separates supported replacements. + size: file_size, + }); + let key_heap_bytes = key.as_ref().map_or(0, |key| key.file.capacity()); + self.inner + .get_or_try_insert_with( + key, + key_heap_bytes, + || load_row_file_metadata(reader, file_size), + |metadata| { + std::mem::size_of::() + .saturating_add(metadata.index.heap_bytes()) + }, + ) + .await + } +} impl RowFormatWriter { pub(crate) async fn new( @@ -254,56 +306,15 @@ impl FormatFileReader for RowFormatReader { }); } - let footer_start = file_size - FOOTER_SIZE; - let footer_bytes = reader.read(footer_start..file_size).await?; - let footer = RowFileFooter::from_bytes(footer_bytes.as_ref())?; - if footer.index_offset < 0 || footer.index_length < 0 { - return Err(Error::DataInvalid { - message: format!( - "Invalid .row footer index offset/length: {}/{}", - footer.index_offset, footer.index_length - ), - source: None, - }); - } - let index_start = footer.index_offset as u64; - let index_end = index_start - .checked_add(footer.index_length as u64) - .ok_or_else(|| Error::DataInvalid { - message: ".row index range overflows u64".to_string(), - source: None, - })?; - if index_end > footer_start { - return Err(Error::DataInvalid { - message: format!( - ".row index range [{index_start}, {index_end}) exceeds footer start {footer_start}" - ), - source: None, - }); - } - let index_bytes = reader.read(index_start..index_end).await?; - let index = RowBlockIndex::from_bytes(index_bytes.as_ref())?; - let footer_block_count = - usize::try_from(footer.block_count).map_err(|e| Error::DataInvalid { - message: format!("Invalid .row footer block count {}", footer.block_count), - source: Some(Box::new(e)), - })?; - if index.block_count() != footer_block_count { - return Err(Error::DataInvalid { - message: format!( - ".row footer block count {} does not match index block count {}", - footer.block_count, - index.block_count() - ), - source: None, - }); - } - let total_rows = - usize::try_from(footer.total_row_count).map_err(|e| Error::DataInvalid { - message: format!("Invalid .row total row count {}", footer.total_row_count), - source: Some(Box::new(e)), - })?; - index.validate_for_file(total_rows, index_start)?; + let metadata = match reader + .file_format_metadata_cache() + .and_then(|cache| cache.downcast_ref::()) + .and_then(|context| context.row_metadata_get_or_init(RowMetadataCache::new)) + { + Some(cache) => cache.load(reader.as_ref(), file_size).await?, + None => load_row_file_metadata(reader.as_ref(), file_size).await?, + }; + let total_rows = metadata.total_rows; validate_row_selection(total_rows, row_selection.as_deref())?; // Public callers may supply unordered or overlapping ranges. Normalize // once so every block can seek directly to its overlapping window. @@ -313,15 +324,16 @@ impl FormatFileReader for RowFormatReader { let row_type = read_fields.to_vec(); let batch_size = batch_size.unwrap_or(DEFAULT_BATCH_SIZE); - let blocks_to_read = blocks_to_read(&index, total_rows, row_selection.as_deref()); + let blocks_to_read = blocks_to_read(&metadata.index, total_rows, row_selection.as_deref()); let predicates = predicates.map(|fp| FilePredicates { predicates: fp.predicates.clone(), row_filter_factory: None, file_fields: fp.file_fields.clone(), }); Ok(try_stream! { + let index = &metadata.index; let mut blocks = futures::stream::iter(blocks_to_read.into_iter().map(|block_idx| { - read_row_block(reader.as_ref(), &index, block_idx) + read_row_block(reader.as_ref(), index, block_idx) })) .buffered(ROW_BLOCK_READ_CONCURRENCY); @@ -354,6 +366,62 @@ impl FormatFileReader for RowFormatReader { } } +async fn load_row_file_metadata( + reader: &dyn FileRead, + file_size: u64, +) -> crate::Result> { + let footer_start = file_size - FOOTER_SIZE; + let footer_bytes = reader.read(footer_start..file_size).await?; + let footer = RowFileFooter::from_bytes(footer_bytes.as_ref())?; + if footer.index_offset < 0 || footer.index_length < 0 { + return Err(Error::DataInvalid { + message: format!( + "Invalid .row footer index offset/length: {}/{}", + footer.index_offset, footer.index_length + ), + source: None, + }); + } + let index_start = footer.index_offset as u64; + let index_end = index_start + .checked_add(footer.index_length as u64) + .ok_or_else(|| Error::DataInvalid { + message: ".row index range overflows u64".to_string(), + source: None, + })?; + if index_end > footer_start { + return Err(Error::DataInvalid { + message: format!( + ".row index range [{index_start}, {index_end}) exceeds footer start {footer_start}" + ), + source: None, + }); + } + let index_bytes = reader.read(index_start..index_end).await?; + let index = RowBlockIndex::from_bytes(index_bytes.as_ref())?; + let footer_block_count = + usize::try_from(footer.block_count).map_err(|e| Error::DataInvalid { + message: format!("Invalid .row footer block count {}", footer.block_count), + source: Some(Box::new(e)), + })?; + if index.block_count() != footer_block_count { + return Err(Error::DataInvalid { + message: format!( + ".row footer block count {} does not match index block count {}", + footer.block_count, + index.block_count() + ), + source: None, + }); + } + let total_rows = usize::try_from(footer.total_row_count).map_err(|e| Error::DataInvalid { + message: format!("Invalid .row total row count {}", footer.total_row_count), + source: Some(Box::new(e)), + })?; + index.validate_for_file(total_rows, index_start)?; + Ok(Arc::new(RowFileMetadata { index, total_rows })) +} + struct RowBlockPayload { block_idx: usize, data: Vec, @@ -1980,6 +2048,15 @@ struct RowBlockIndex { } impl RowBlockIndex { + fn heap_bytes(&self) -> usize { + self.block_offsets + .capacity() + .saturating_add(self.block_compressed_sizes.capacity()) + .saturating_add(self.block_uncompressed_sizes.capacity()) + .saturating_add(self.block_row_starts.capacity()) + .saturating_mul(std::mem::size_of::()) + } + fn new( block_compressed_sizes: Vec, block_uncompressed_sizes: Vec, @@ -2468,6 +2545,7 @@ mod tests { }; use crate::variant::GenericVariant; use futures::TryStreamExt; + use std::collections::HashMap; use std::ops::Range; use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; @@ -3581,6 +3659,97 @@ mod tests { .is_err()); } + struct CountingFileRead { + bytes: Bytes, + key: Option, + context: Arc, + reads: Arc, + } + + #[async_trait::async_trait] + impl FileRead for CountingFileRead { + async fn read(&self, range: Range) -> crate::Result { + self.reads.fetch_add(1, Ordering::SeqCst); + Ok(self.bytes.slice(range.start as usize..range.end as usize)) + } + + fn cache_key(&self) -> Option<&str> { + self.key.as_deref() + } + + fn file_format_metadata_cache(&self) -> Option<&(dyn std::any::Any + Send + Sync)> { + Some(self.context.as_ref()) + } + } + + async fn read_counted_fixture( + context: Arc, + key: Option<&str>, + reads: Arc, + ) { + let bytes = java_fixture_bytes(); + let batches = RowFormatReader + .read_batch_stream( + Box::new(CountingFileRead { + bytes: bytes.clone(), + key: key.map(str::to_string), + context, + reads, + }), + bytes.len() as u64, + &java_fixture_fields(), + None, + Some(8), + None, + ) + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::(), 2); + } + + #[tokio::test] + async fn row_reader_caches_footer_and_index_by_file_io_context() { + let enabled = FileFormatMetadataCacheContext::from_props(&HashMap::new()).unwrap(); + let reads = Arc::new(AtomicUsize::new(0)); + let key = Some("storage-A/data.row"); + read_counted_fixture(enabled.clone(), key, reads.clone()).await; + assert_eq!(reads.load(Ordering::SeqCst), 3); // footer, index, block + read_counted_fixture(enabled.clone(), key, reads.clone()).await; + assert_eq!(reads.load(Ordering::SeqCst), 4); // block only + + let other = FileFormatMetadataCacheContext::from_props(&HashMap::new()).unwrap(); + let other_reads = Arc::new(AtomicUsize::new(0)); + read_counted_fixture(other, key, other_reads.clone()).await; + assert_eq!(other_reads.load(Ordering::SeqCst), 3); + + let other_key_reads = Arc::new(AtomicUsize::new(0)); + read_counted_fixture( + enabled.clone(), + Some("storage-B/data.row"), + other_key_reads.clone(), + ) + .await; + assert_eq!(other_key_reads.load(Ordering::SeqCst), 3); + + let unkeyed_reads = Arc::new(AtomicUsize::new(0)); + read_counted_fixture(enabled.clone(), None, unkeyed_reads.clone()).await; + read_counted_fixture(enabled, None, unkeyed_reads.clone()).await; + assert_eq!(unkeyed_reads.load(Ordering::SeqCst), 6); + + let disabled = FileFormatMetadataCacheContext::from_props(&HashMap::from([( + crate::common::options::CatalogOptions::FILE_FORMAT_METADATA_CACHE_MAX_SIZE.to_string(), + "0".to_string(), + )])) + .unwrap(); + let disabled_reads = Arc::new(AtomicUsize::new(0)); + read_counted_fixture(disabled.clone(), key, disabled_reads.clone()).await; + read_counted_fixture(disabled, key, disabled_reads.clone()).await; + assert_eq!(disabled_reads.load(Ordering::SeqCst), 6); + } + #[tokio::test] async fn row_reader_reads_java_generated_fixture() { let fields = java_fixture_fields(); diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs index ab0934272..aed6d533b 100644 --- a/crates/paimon/src/io/file_io.rs +++ b/crates/paimon/src/io/file_io.rs @@ -113,10 +113,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_METADATA_CACHE_MAX_BYTES: usize = 32 * 1024 * 1024; pub(crate) struct FileFormatMetadataCacheContext { max_bytes: usize, cache: OnceLock>, + row_metadata_cache: OnceLock>, + process_id: u32, } impl std::fmt::Debug for FileFormatMetadataCacheContext { @@ -147,6 +150,8 @@ impl FileFormatMetadataCacheContext { Ok(Arc::new(Self { max_bytes, cache: OnceLock::new(), + row_metadata_cache: OnceLock::new(), + process_id: std::process::id(), })) } @@ -167,6 +172,28 @@ impl FileFormatMetadataCacheContext { Err(_) => panic!("file-format metadata cache type mismatch"), } } + + /// ROW footer and block index metadata has its own typed cache. Keep it + /// separate from the Parquet cache, which also uses `get_or_init`. + pub(crate) fn row_metadata_get_or_init( + &self, + init: impl FnOnce(usize) -> T, + ) -> Option> + where + T: Send + Sync + 'static, + { + if self.max_bytes == 0 || self.process_id != std::process::id() { + return None; + } + let cache = Arc::clone(self.row_metadata_cache.get_or_init(|| { + Arc::new(init(self.max_bytes.min(ROW_METADATA_CACHE_MAX_BYTES))) + as Arc + })); + match cache.downcast::() { + Ok(cache) => Some(cache), + Err(_) => panic!("ROW metadata cache type mismatch"), + } + } } pub(crate) struct BlobIndexCacheContext { @@ -2674,6 +2701,15 @@ mod input_output_test { assert!(!std::ptr::eq(first_cache, second_cache)); } + #[test] + fn row_metadata_cache_has_separate_type_and_bypasses_inherited_fork_state() { + let mut context = FileFormatMetadataCacheContext::from_props(&HashMap::new()).unwrap(); + assert_eq!(*context.get_or_init(|_| 7u8), 7); + assert_eq!(*context.row_metadata_get_or_init(|_| 11u16).unwrap(), 11); + Arc::get_mut(&mut context).unwrap().process_id = 0; + assert!(context.row_metadata_get_or_init(|_| 13u16).is_none()); + } + #[tokio::test] async fn test_file_format_metadata_cache_size_reaches_file_reader() { let path = "memory:/metadata-cache-size.parquet"; From 22c509fad0940cd8f5fbed0cbbb631d6edb39383 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Wed, 21 Oct 2026 19:35:11 -0700 Subject: [PATCH 4/5] 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] From 60a80468bf9b38a9e95621add276e6ae7218d28a Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Thu, 22 Oct 2026 07:23:17 -0700 Subject: [PATCH 5/5] [fix] Bypass ROW metadata cache for format tables --- crates/paimon/src/arrow/format/mod.rs | 2 +- crates/paimon/src/arrow/format/row.rs | 155 ++++++++++++++++++++++---- 2 files changed, 137 insertions(+), 20 deletions(-) diff --git a/crates/paimon/src/arrow/format/mod.rs b/crates/paimon/src/arrow/format/mod.rs index 3f12a8410..3de32c7e9 100644 --- a/crates/paimon/src/arrow/format/mod.rs +++ b/crates/paimon/src/arrow/format/mod.rs @@ -411,7 +411,7 @@ pub(crate) fn create_format_reader_with_budget( } else if lower.ends_with(".avro") { Box::new(avro::AvroFormatReader) } else if lower.ends_with(".row") { - Box::new(row::RowFormatReader) + Box::new(row::RowFormatReader::with_options(table_options)) } else if lower.ends_with(".mosaic") { Box::new(mosaic::MosaicFormatReader::with_prefetch(mosaic_prefetch)) } else { diff --git a/crates/paimon/src/arrow/format/row.rs b/crates/paimon/src/arrow/format/row.rs index b984ddd3d..3294fdf68 100644 --- a/crates/paimon/src/arrow/format/row.rs +++ b/crates/paimon/src/arrow/format/row.rs @@ -27,7 +27,7 @@ use crate::arrow::{ variant_arrow_type, }; use crate::io::{FileFormatMetadataCacheContext, FileRead, FileWrite, OutputFile}; -use crate::spec::{DataField, DataType, IntType, VarBinaryType, VariantType}; +use crate::spec::{CoreOptions, DataField, DataType, IntType, VarBinaryType, VariantType}; use crate::table::{ArrowRecordBatchStream, RowRange}; use crate::Error; use arrow_array::builder::{ @@ -49,9 +49,29 @@ use async_stream::try_stream; use async_trait::async_trait; use bytes::Bytes; use futures::StreamExt; +use std::collections::HashMap; use std::sync::Arc; -pub(crate) struct RowFormatReader; +pub(crate) struct RowFormatReader { + metadata_cache_enabled: bool, +} + +impl Default for RowFormatReader { + fn default() -> Self { + Self { + metadata_cache_enabled: true, + } + } +} + +impl RowFormatReader { + pub(crate) fn with_options(options: &HashMap) -> Self { + Self { + // Format-table files may be replaced in place, even with the same size. + metadata_cache_enabled: !CoreOptions::new(options).is_format_table(), + } + } +} pub(crate) struct RowFormatWriter { writer: Box, @@ -306,11 +326,15 @@ impl FormatFileReader for RowFormatReader { }); } - let metadata = match reader - .file_format_metadata_cache() - .and_then(|cache| cache.downcast_ref::()) - .and_then(|context| context.row_metadata_get_or_init(RowMetadataCache::new)) - { + let cache = if self.metadata_cache_enabled { + reader + .file_format_metadata_cache() + .and_then(|cache| cache.downcast_ref::()) + .and_then(|context| context.row_metadata_get_or_init(RowMetadataCache::new)) + } else { + None + }; + let metadata = match cache { Some(cache) => cache.load(reader.as_ref(), file_size).await?, None => load_row_file_metadata(reader.as_ref(), file_size).await?, }; @@ -2641,7 +2665,7 @@ mod tests { let input = file_io.new_input(path).unwrap(); let bytes = input.read().await.unwrap(); - let batches = RowFormatReader + let batches = RowFormatReader::default() .read_batch_stream( Box::new(BytesFileRead(bytes.clone())), bytes.len() as u64, @@ -2719,7 +2743,7 @@ mod tests { let input = file_io.new_input(path).unwrap(); let bytes = input.read().await.unwrap(); // read_fields is the FULL physical schema, as DataFileReader passes for `.row`. - let batches = RowFormatReader + let batches = RowFormatReader::default() .read_batch_stream( Box::new(BytesFileRead(bytes.clone())), bytes.len() as u64, @@ -2808,7 +2832,7 @@ mod tests { let input = file_io.new_input(path).unwrap(); let bytes = input.read().await.unwrap(); - let batches = RowFormatReader + let batches = RowFormatReader::default() .read_batch_stream( Box::new(BytesFileRead(bytes.clone())), bytes.len() as u64, @@ -2881,7 +2905,7 @@ mod tests { let input = file_io.new_input(path).unwrap(); let bytes = input.read().await.unwrap(); - let actual = RowFormatReader + let actual = RowFormatReader::default() .read_batch_stream( Box::new(BytesFileRead(bytes.clone())), bytes.len() as u64, @@ -2941,7 +2965,7 @@ mod tests { Box::new(writer).close().await.unwrap(); let bytes = file_io.new_input(path).unwrap().read().await.unwrap(); - let batches = RowFormatReader + let batches = RowFormatReader::default() .read_batch_stream( Box::new(BytesFileRead(bytes.clone())), bytes.len() as u64, @@ -3022,7 +3046,7 @@ mod tests { Box::new(writer).close().await.unwrap(); let bytes = file_io.new_input(path).unwrap().read().await.unwrap(); - let batches = RowFormatReader + let batches = RowFormatReader::default() .read_batch_stream( Box::new(BytesFileRead(bytes.clone())), bytes.len() as u64, @@ -3126,7 +3150,7 @@ mod tests { Box::new(writer).close().await.unwrap(); let bytes = file_io.new_input(path).unwrap().read().await.unwrap(); - let batches = RowFormatReader + let batches = RowFormatReader::default() .read_batch_stream( Box::new(BytesFileRead(bytes.clone())), bytes.len() as u64, @@ -3290,7 +3314,7 @@ mod tests { let bytes = file_io.new_input(path).unwrap().read().await.unwrap(); let reader = TrackingFileRead::new(bytes.clone()); - let batches = RowFormatReader + let batches = RowFormatReader::default() .read_batch_stream( Box::new(reader.clone()), bytes.len() as u64, @@ -3469,7 +3493,7 @@ mod tests { Box::new(writer).close().await.unwrap(); let bytes = file_io.new_input(path).unwrap().read().await.unwrap(); - let batches = RowFormatReader + let batches = RowFormatReader::default() .read_batch_stream( Box::new(BytesFileRead(bytes.clone())), bytes.len() as u64, @@ -3688,7 +3712,7 @@ mod tests { reads: Arc, ) { let bytes = java_fixture_bytes(); - let batches = RowFormatReader + let batches = RowFormatReader::default() .read_batch_stream( Box::new(CountingFileRead { bytes: bytes.clone(), @@ -3750,12 +3774,105 @@ mod tests { assert_eq!(disabled_reads.load(Ordering::SeqCst), 6); } + fn one_string_row_file(value: &str) -> Bytes { + let fields = vec![DataField::new( + 0, + "value".to_string(), + DataType::VarChar(VarCharType::new(256).unwrap()), + )]; + let schema = build_target_arrow_schema(&fields).unwrap(); + let batch = + RecordBatch::try_new(schema, vec![Arc::new(StringArray::from(vec![value]))]).unwrap(); + let mut block_writer = RowBlockWriter::new(); + block_writer.write_row(&batch, 0, &fields).unwrap(); + let block = block_writer.finish(); + let compressed = zstd::bulk::compress(&block, 1).unwrap(); + let index = RowBlockIndex::new( + vec![compressed.len() as i64], + vec![block.len() as i64], + vec![0], + ) + .unwrap() + .to_bytes() + .unwrap(); + let footer = RowFileFooter { + total_row_count: 1, + block_count: 1, + index_offset: compressed.len() as i64, + index_length: index.len() as i32, + }; + let mut file = compressed; + file.extend_from_slice(&index); + file.extend_from_slice(&footer.to_bytes()); + Bytes::from(file) + } + + #[tokio::test] + async fn format_table_row_metadata_bypasses_cache_after_same_size_replacement() { + let old_value = "a".repeat(100); + let new_value = "a".repeat(101); + let old_file = one_string_row_file(&old_value); + let new_file = one_string_row_file(&new_value); + assert_eq!(old_file.len(), new_file.len()); + + let fields = vec![DataField::new( + 0, + "value".to_string(), + DataType::VarChar(VarCharType::new(256).unwrap()), + )]; + let context = FileFormatMetadataCacheContext::from_props(&HashMap::new()).unwrap(); + let reads = Arc::new(AtomicUsize::new(0)); + for (file, expected) in [(old_file, old_value), (new_file, new_value)] { + let configured = super::super::create_format_reader_with_budget( + "data.row", + false, + super::super::FormatReadFields { + data_schema: &fields, + projected: &fields, + }, + &HashMap::from([("type".to_string(), "format-table".to_string())]), + None, + super::super::blob::DEFAULT_BLOB_READ_PARALLELISM, + super::super::MosaicPrefetchOptions::default(), + ) + .unwrap(); + let batches = configured + .reader + .read_batch_stream( + Box::new(CountingFileRead { + bytes: file.clone(), + key: Some("same-path/data.row".to_string()), + context: context.clone(), + reads: reads.clone(), + }), + file.len() as u64, + &fields, + None, + Some(8), + None, + ) + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + assert_eq!(batches.len(), 1); + let values = batches[0] + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(values.value(0), expected); + } + assert_eq!(reads.load(Ordering::SeqCst), 6); // footer, index, block each time + } + #[tokio::test] async fn row_reader_reads_java_generated_fixture() { let fields = java_fixture_fields(); let bytes = java_fixture_bytes(); let file_size = bytes.len() as u64; - let batches = RowFormatReader + let batches = RowFormatReader::default() .read_batch_stream( Box::new(BytesFileRead(bytes.clone())), file_size, @@ -4106,7 +4223,7 @@ mod tests { .await .unwrap(); - let batches = RowFormatReader + let batches = RowFormatReader::default() .read_batch_stream( Box::new(BytesFileRead(bytes.clone())), bytes.len() as u64,