This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new dc049cb9 perf(core): cache ROW sidecar size and reader metadata (#1196)
dc049cb9 is described below
commit dc049cb906d056a7350b3e76f450df0ed32371e0
Author: XiaoHongbo <[email protected]>
AuthorDate: Sat Oct 24 10:24:37 2026 +0800
perf(core): cache ROW sidecar size and reader metadata (#1196)
---
crates/paimon/src/arrow/format/mod.rs | 2 +-
crates/paimon/src/arrow/format/row.rs | 418 +++++++++++++++++++++++++++------
crates/paimon/src/io/file_io.rs | 320 ++++++++++++++++++++++++-
crates/paimon/src/table/row_sidecar.rs | 4 +-
4 files changed, 670 insertions(+), 74 deletions(-)
diff --git a/crates/paimon/src/arrow/format/mod.rs
b/crates/paimon/src/arrow/format/mod.rs
index 3f12a841..3de32c7e 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 ff53b619..3294fdf6 100644
--- a/crates/paimon/src/arrow/format/row.rs
+++ b/crates/paimon/src/arrow/format/row.rs
@@ -20,13 +20,14 @@
//! 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::spec::{DataField, DataType, IntType, VarBinaryType, VariantType};
+use crate::io::{FileFormatMetadataCacheContext, FileRead, FileWrite,
OutputFile};
+use crate::spec::{CoreOptions, DataField, DataType, IntType, VarBinaryType,
VariantType};
use crate::table::{ArrowRecordBatchStream, RowRange};
use crate::Error;
use arrow_array::builder::{
@@ -48,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<String, String>) -> 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<dyn FileWrite>,
@@ -71,6 +92,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<RowMetadataCacheKey, RowFileMetadata>,
+}
+
+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<Arc<RowFileMetadata>> {
+ 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::<RowFileMetadata>()
+ .saturating_add(metadata.index.heap_bytes())
+ },
+ )
+ .await
+ }
+}
impl RowFormatWriter {
pub(crate) async fn new(
@@ -254,56 +326,19 @@ 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 cache = if self.metadata_cache_enabled {
+ reader
+ .file_format_metadata_cache()
+ .and_then(|cache|
cache.downcast_ref::<FileFormatMetadataCacheContext>())
+ .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?,
+ };
+ 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 +348,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 +390,62 @@ impl FormatFileReader for RowFormatReader {
}
}
+async fn load_row_file_metadata(
+ reader: &dyn FileRead,
+ file_size: u64,
+) -> crate::Result<Arc<RowFileMetadata>> {
+ 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<u8>,
@@ -1980,6 +2072,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::<i64>())
+ }
+
fn new(
block_compressed_sizes: Vec<i64>,
block_uncompressed_sizes: Vec<i64>,
@@ -2468,6 +2569,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;
@@ -2563,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,
@@ -2641,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,
@@ -2730,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,
@@ -2803,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,
@@ -2863,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,
@@ -2944,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,
@@ -3048,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,
@@ -3212,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,
@@ -3391,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,
@@ -3581,12 +3683,196 @@ mod tests {
.is_err());
}
+ struct CountingFileRead {
+ bytes: Bytes,
+ key: Option<String>,
+ context: Arc<FileFormatMetadataCacheContext>,
+ reads: Arc<AtomicUsize>,
+ }
+
+ #[async_trait::async_trait]
+ impl FileRead for CountingFileRead {
+ async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
+ 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<FileFormatMetadataCacheContext>,
+ key: Option<&str>,
+ reads: Arc<AtomicUsize>,
+ ) {
+ let bytes = java_fixture_bytes();
+ let batches = RowFormatReader::default()
+ .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::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
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);
+ }
+
+ 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::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+ let values = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .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,
@@ -3937,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,
diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index 9369cb74..301294fa 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,17 @@ 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;
+// Keep the combined parsed Parquet and ROW metadata below the configured
limit.
+const ROW_METADATA_CACHE_BUDGET_DIVISOR: usize = 4;
+const ROW_SIDECAR_SIZE_CACHE_MAX_ENTRIES: usize = 4096;
pub(crate) struct FileFormatMetadataCacheContext {
max_bytes: usize,
cache: OnceLock<Arc<dyn Any + Send + Sync>>,
+ row_metadata_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,6 +156,9 @@ impl FileFormatMetadataCacheContext {
Ok(Arc::new(Self {
max_bytes,
cache: OnceLock::new(),
+ row_metadata_cache: OnceLock::new(),
+ process_id: std::process::id(),
+ row_sidecar_sizes: OnceLock::new(),
}))
}
@@ -154,19 +166,100 @@ impl FileFormatMetadataCacheContext {
self.max_bytes
}
+ fn row_metadata_max_bytes(&self) -> usize {
+ (self.max_bytes /
ROW_METADATA_CACHE_BUDGET_DIVISOR).min(ROW_METADATA_CACHE_MAX_BYTES)
+ }
+
+ fn parquet_metadata_max_bytes(&self) -> usize {
+ self.max_bytes - self.row_metadata_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,
{
- let cache = Arc::clone(
- self.cache
- .get_or_init(|| Arc::new(init(self.max_bytes)) as Arc<dyn Any
+ Send + Sync>),
- );
+ let cache = Arc::clone(self.cache.get_or_init(|| {
+ Arc::new(init(self.parquet_metadata_max_bytes())) as Arc<dyn Any +
Send + Sync>
+ }));
match cache.downcast::<T>() {
Ok(cache) => cache,
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<T>(
+ &self,
+ init: impl FnOnce(usize) -> T,
+ ) -> Option<Arc<T>>
+ 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.row_metadata_max_bytes())) as Arc<dyn Any +
Send + Sync>
+ }));
+ match cache.downcast::<T>() {
+ Ok(cache) => Some(cache),
+ Err(_) => panic!("ROW metadata cache type mismatch"),
+ }
+ }
}
pub(crate) struct BlobIndexCacheContext {
@@ -509,6 +602,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>
@@ -664,6 +784,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,
@@ -693,6 +817,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?;
}
@@ -1611,6 +1739,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),
@@ -1623,6 +1752,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 {
@@ -1663,6 +1793,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())
}
@@ -1919,6 +2055,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)));
@@ -2184,6 +2321,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(
@@ -2687,6 +2890,113 @@ 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());
+ }
+
+ #[test]
+ fn row_and_parquet_metadata_share_configured_budget() {
+ let context =
FileFormatMetadataCacheContext::from_props(&HashMap::from([(
+ CatalogOptions::FILE_FORMAT_METADATA_CACHE_MAX_SIZE.to_string(),
+ "64 kb".to_string(),
+ )]))
+ .unwrap();
+ let parquet_budget = *context.get_or_init(|bytes| bytes);
+ let row_budget = *context.row_metadata_get_or_init(|bytes|
bytes).unwrap();
+ assert_eq!(parquet_budget, 48 * 1024);
+ assert_eq!(row_budget, 16 * 1024);
+ assert_eq!(parquet_budget + row_budget, context.max_bytes());
+
+ let default =
FileFormatMetadataCacheContext::from_props(&HashMap::new()).unwrap();
+ assert_eq!(
+ default.parquet_metadata_max_bytes() +
default.row_metadata_max_bytes(),
+ DEFAULT_FILE_FORMAT_METADATA_CACHE_MAX_BYTES
+ );
+ }
+
+ #[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 624ca6ef..a337b9c9 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))
}
