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 c1e5810d feat: share parsed Parquet metadata across read contexts
(#1223)
c1e5810d is described below
commit c1e5810d2a15b429137db43eb4e5141fef226284
Author: XiaoHongbo <[email protected]>
AuthorDate: Tue Oct 27 16:10:54 2026 +0800
feat: share parsed Parquet metadata across read contexts (#1223)
---
crates/paimon/src/arrow/format/parquet.rs | 279 +++++++++++++++++++++++++++-
crates/paimon/src/common/options.rs | 3 +
crates/paimon/src/io/cache/reader.rs | 4 +
crates/paimon/src/io/file_io.rs | 251 ++++++++++++++++++++++++-
crates/paimon/src/table/data_file_reader.rs | 7 +
docs/src/getting-started.md | 32 ++--
6 files changed, 553 insertions(+), 23 deletions(-)
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index aa2df217..69fdeafd 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -2902,6 +2902,7 @@ impl ParquetMetadataCache {
&self,
reader: &mut ArrowFileReader,
options: Option<&ArrowReaderOptions>,
+ shared: bool,
) -> parquet::errors::Result<Arc<ParquetMetaData>> {
let column_index = options.map_or(PageIndexPolicy::Skip, |o|
o.column_index_policy());
let offset_index = options.map_or(PageIndexPolicy::Skip, |o|
o.offset_index_policy());
@@ -2917,18 +2918,34 @@ impl ParquetMetadataCache {
offset_index: page_index_policy_tag(offset_index),
});
let key_heap_bytes = key.as_ref().map_or(0, |key| key.file.capacity());
- self.inner
+ let mut loaded = false;
+ let metadata = self
+ .inner
.get_or_try_insert_with(
key,
key_heap_bytes,
- || reader.load_metadata(options),
+ || {
+ loaded = true;
+ reader.load_metadata(options)
+ },
|metadata| {
metadata
.memory_size()
.max(PARQUET_METADATA_CACHE_MIN_ENTRY_BYTES)
},
)
- .await
+ .await?;
+ if shared
+ && !loaded
+ && !reader
+ .r
+ .check_shared_metadata_access()
+ .await
+ .map_err(|error|
parquet::errors::ParquetError::External(Box::new(error)))?
+ {
+ return reader.load_metadata(options).await;
+ }
+ Ok(metadata)
}
}
@@ -3165,9 +3182,10 @@ impl AsyncFileReader for ArrowFileReader {
if context.max_bytes() == 0 {
return self.load_metadata(options.as_ref()).await;
}
+ let shared = context.parquet_metadata_is_shared();
context
.get_or_init(ParquetMetadataCache::new)
- .load(self, options.as_ref())
+ .load(self, options.as_ref(), shared)
.await
})
}
@@ -5724,6 +5742,11 @@ mod tests {
#[async_trait::async_trait]
impl crate::io::FileRead for TrackingFileRead {
+ async fn check_shared_metadata_access(&self) -> crate::Result<bool> {
+ self.read(0..1).await?;
+ Ok(true)
+ }
+
async fn read(&self, range: std::ops::Range<u64>) ->
crate::Result<Bytes> {
self.ranges.lock().unwrap().push(range.clone());
// Count each source allocation until its last slice is dropped,
not slice lengths.
@@ -5769,6 +5792,114 @@ mod tests {
assert_eq!(tracker.read_count(), 1);
}
+ #[tokio::test]
+ async fn parquet_metadata_cache_shares_independently_created_contexts() {
+ if !crate::io::run_isolated_metadata_cache_test(
+
"arrow::format::parquet::tests::parquet_metadata_cache_shares_independently_created_contexts",
+ ) {
+ return;
+ }
+ let data = Bytes::from(
+ write_multi_row_group_parquet(8, 64, EnabledStatistics::Chunk,
false).await,
+ );
+ let props = HashMap::from([(
+
crate::common::CatalogOptions::FILE_FORMAT_METADATA_CACHE_SHARED.to_string(),
+ "true".to_string(),
+ )]);
+ let mut trackers = Vec::new();
+ let mut metadata = Vec::new();
+ for _ in 0..40 {
+ let context =
crate::io::FileFormatMetadataCacheContext::from_props(&props).unwrap();
+ let tracker = TrackingFileRead::new(data.clone())
+ .with_cache_key("process-sharing-test\0data.parquet")
+ .with_metadata_cache(context);
+ let mut reader =
+ super::ArrowFileReader::new(data.len() as u64,
Arc::new(tracker.clone()));
+ metadata.push(reader.get_metadata(None).await.unwrap());
+ trackers.push(tracker);
+ }
+ assert_eq!(
+ trackers
+ .iter()
+ .map(TrackingFileRead::read_count)
+ .sum::<usize>(),
+ 40
+ );
+ assert!(trackers
+ .iter()
+ .skip(1)
+ .all(|reader| reader.bytes_read() == 1));
+ assert!(metadata
+ .iter()
+ .all(|value| Arc::ptr_eq(value, &metadata[0])));
+ }
+
+ #[tokio::test]
+ async fn shared_metadata_rejects_an_unauthorized_reader() {
+ if !crate::io::run_isolated_metadata_cache_test(
+
"arrow::format::parquet::tests::shared_metadata_rejects_an_unauthorized_reader",
+ ) {
+ return;
+ }
+ struct DeniedReader(TrackingFileRead, bool, Arc<AtomicUsize>);
+ #[async_trait::async_trait]
+ impl crate::io::FileRead for DeniedReader {
+ async fn read(&self, _: std::ops::Range<u64>) ->
crate::Result<Bytes> {
+ Err(crate::Error::from(opendal::Error::new(
+ opendal::ErrorKind::PermissionDenied,
+ "denied",
+ )))
+ }
+ async fn check_shared_metadata_access(&self) ->
crate::Result<bool> {
+ self.2.fetch_add(1, AtomicOrdering::SeqCst);
+ if !self.1 {
+ return Ok(false);
+ }
+ self.read(0..1).await?;
+ Ok(true)
+ }
+ fn cache_key(&self) -> Option<&str> {
+ self.0.cache_key()
+ }
+ fn file_format_metadata_cache(&self) -> Option<&(dyn std::any::Any
+ Send + Sync)> {
+ self.0.file_format_metadata_cache()
+ }
+ }
+ let data = Bytes::from(
+ write_multi_row_group_parquet(8, 64, EnabledStatistics::Chunk,
false).await,
+ );
+ let props = HashMap::from([(
+
crate::common::CatalogOptions::FILE_FORMAT_METADATA_CACHE_SHARED.to_string(),
+ "true".to_string(),
+ )]);
+ let first = TrackingFileRead::new(data.clone())
+ .with_cache_key("authorization-test/data.parquet")
+ .with_metadata_cache(
+
crate::io::FileFormatMetadataCacheContext::from_props(&props).unwrap(),
+ );
+ let mut allowed = super::ArrowFileReader::new(data.len() as u64,
Arc::new(first));
+ allowed.get_metadata(None).await.unwrap();
+ let second = TrackingFileRead::new(data.clone())
+ .with_cache_key("authorization-test/data.parquet")
+ .with_metadata_cache(
+
crate::io::FileFormatMetadataCacheContext::from_props(&props).unwrap(),
+ );
+ for supports_check in [true, false] {
+ let checks = Arc::new(AtomicUsize::new(0));
+ let mut denied = super::ArrowFileReader::new(
+ data.len() as u64,
+ Arc::new(DeniedReader(second.clone(), supports_check,
checks.clone())),
+ );
+ assert!(denied
+ .get_metadata(None)
+ .await
+ .unwrap_err()
+ .to_string()
+ .contains("denied"));
+ assert_eq!(checks.load(AtomicOrdering::SeqCst), 1);
+ }
+ }
+
#[cfg(feature = "storage-fs")]
#[tokio::test]
async fn parquet_metadata_cache_reuses_metadata_across_file_io_instances()
{
@@ -5806,6 +5937,146 @@ mod tests {
assert!(Arc::ptr_eq(&first_metadata, &second_metadata));
}
+ #[cfg(feature = "storage-memory")]
+ #[tokio::test]
+ async fn shared_metadata_separates_recycled_memory_storage_identities() {
+ if !crate::io::run_isolated_metadata_cache_test(
+
"arrow::format::parquet::tests::shared_metadata_separates_recycled_memory_storage_identities",
+ ) {
+ return;
+ }
+ use crate::io::FileRead;
+
+ struct RecycledMemoryReader {
+ inner: Arc<dyn FileRead>,
+ key: String,
+ }
+ impl RecycledMemoryReader {
+ fn new(inner: impl FileRead) -> Self {
+ let mut parts =
inner.cache_key().unwrap().split('\0').collect::<Vec<_>>();
+ assert_eq!(parts.len(), 5);
+ assert_eq!(parts[1], "memory");
+ // Simulate OpenDAL reusing an address-based service name. Keep
+ // the real FileIO namespace, root and path; do not rely on
malloc.
+ parts[2] = "recycled-memory-address";
+ let key = parts.join("\0");
+ Self {
+ inner: Arc::new(inner),
+ key,
+ }
+ }
+ }
+ #[async_trait::async_trait]
+ impl FileRead for RecycledMemoryReader {
+ async fn read(&self, range: std::ops::Range<u64>) ->
crate::Result<Bytes> {
+ self.inner.read(range).await
+ }
+ async fn check_shared_metadata_access(&self) ->
crate::Result<bool> {
+ self.inner.check_shared_metadata_access().await
+ }
+ fn cache_key(&self) -> Option<&str> {
+ Some(&self.key)
+ }
+ fn file_format_metadata_cache(&self) -> Option<&(dyn std::any::Any
+ Send + Sync)> {
+ self.inner.file_format_metadata_cache()
+ }
+ }
+ fn parquet_data(name: &str, value: i32) -> Bytes {
+ let schema = Arc::new(ArrowSchema::new(vec![ArrowField::new(
+ name,
+ ArrowDataType::Int32,
+ false,
+ )]));
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![Arc::new(Int32Array::from(vec![value; 4]))],
+ )
+ .unwrap();
+ let mut data = Vec::new();
+ let mut writer = parquet::arrow::ArrowWriter::try_new(&mut data,
schema, None).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ Bytes::from(data)
+ }
+
+ let first_data = parquet_data("first", 1);
+ let second_data = parquet_data("other", 9);
+ assert_eq!(first_data.len(), second_data.len());
+ let props = HashMap::from([(
+
crate::common::CatalogOptions::FILE_FORMAT_METADATA_CACHE_SHARED.to_string(),
+ "true".to_string(),
+ )]);
+ let _keep_pool_alive =
+
crate::io::FileFormatMetadataCacheContext::from_props(&props).unwrap();
+ let path = "memory:/recycled-identity.parquet";
+ let first = FileIOBuilder::new("memory")
+ .with_props(props.clone())
+ .build()
+ .unwrap();
+ first
+ .new_output(path)
+ .unwrap()
+ .write(first_data.clone())
+ .await
+ .unwrap();
+ let mut reader = super::ArrowFileReader::new(
+ first_data.len() as u64,
+ Arc::new(RecycledMemoryReader::new(
+ first.new_input(path).unwrap().reader().await.unwrap(),
+ )),
+ );
+ let first_metadata = reader.get_metadata(None).await.unwrap();
+ assert_eq!(
+ first_metadata
+ .file_metadata()
+ .schema_descr()
+ .column(0)
+ .name(),
+ "first"
+ );
+ let clone = first.clone();
+ let mut cloned_reader = super::ArrowFileReader::new(
+ first_data.len() as u64,
+ Arc::new(RecycledMemoryReader::new(
+ clone.new_input(path).unwrap().reader().await.unwrap(),
+ )),
+ );
+ assert!(Arc::ptr_eq(
+ &first_metadata,
+ &cloned_reader.get_metadata(None).await.unwrap()
+ ));
+ drop(cloned_reader);
+ drop(clone);
+ drop(reader);
+ drop(first);
+
+ let second = FileIOBuilder::new("memory")
+ .with_props(props)
+ .build()
+ .unwrap();
+ second
+ .new_output(path)
+ .unwrap()
+ .write(second_data.clone())
+ .await
+ .unwrap();
+ let mut reader = super::ArrowFileReader::new(
+ second_data.len() as u64,
+ Arc::new(RecycledMemoryReader::new(
+ second.new_input(path).unwrap().reader().await.unwrap(),
+ )),
+ );
+ let metadata = reader.get_metadata(None).await.unwrap();
+ assert_eq!(
+ metadata.file_metadata().schema_descr().column(0).name(),
+ "other"
+ );
+ match metadata.row_group(0).column(0).statistics().unwrap() {
+ ParquetStatistics::Int32(stats) => assert_eq!(stats.min_opt(),
Some(&9)),
+ _ => panic!("expected Int32 statistics"),
+ }
+ }
+
#[tokio::test]
async fn parquet_metadata_cache_can_be_bypassed() {
let data = Bytes::from(
diff --git a/crates/paimon/src/common/options.rs
b/crates/paimon/src/common/options.rs
index eeefb7d0..28eab750 100644
--- a/crates/paimon/src/common/options.rs
+++ b/crates/paimon/src/common/options.rs
@@ -101,6 +101,9 @@ impl CatalogOptions {
pub const FILE_FORMAT_METADATA_CACHE_MAX_SIZE: &'static str =
"file-format.metadata-cache.max-size";
+ /// Share parsed Parquet metadata across contexts in the same process
(default: false).
+ pub const FILE_FORMAT_METADATA_CACHE_SHARED: &'static str =
"file-format.metadata-cache.shared";
+
/// Maximum catalog-local cache size for decoded BLOB indexes.
pub const BLOB_INDEX_CACHE_MAX_SIZE: &'static str =
"cache.blob-index.max-size";
}
diff --git a/crates/paimon/src/io/cache/reader.rs
b/crates/paimon/src/io/cache/reader.rs
index d60099f7..a4a083d4 100644
--- a/crates/paimon/src/io/cache/reader.rs
+++ b/crates/paimon/src/io/cache/reader.rs
@@ -333,6 +333,10 @@ impl CachedFileReader {
#[async_trait::async_trait]
impl FileRead for CachedFileReader {
+ async fn check_shared_metadata_access(&self) -> crate::Result<bool> {
+ self.delegate.check_shared_metadata_access().await
+ }
+
async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
if range.start > range.end || range.end > self.file_size {
return self.delegate.read(range).await;
diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index 3e889450..acf9a2f4 100644
--- a/crates/paimon/src/io/file_io.rs
+++ b/crates/paimon/src/io/file_io.rs
@@ -120,11 +120,53 @@ const ROW_METADATA_CACHE_MAX_BYTES: usize = 32 * 1024 *
1024;
const ROW_METADATA_CACHE_BUDGET_DIVISOR: usize = 4;
const ROW_SIDECAR_SIZE_CACHE_MAX_ENTRIES: usize = 4096;
+type ParquetMetadataSlot = OnceLock<Arc<dyn Any + Send + Sync>>;
+type ParquetMetadataRegistry = HashMap<(u32, usize),
std::sync::Weak<ParquetMetadataSlot>>;
+static PARQUET_METADATA_SLOTS: Mutex<Option<ParquetMetadataRegistry>> =
Mutex::new(None);
+
+fn shared_parquet_metadata_slot(max_bytes: usize) -> Arc<ParquetMetadataSlot> {
+ // Sharing is best-effort; never wait on a registry lock inherited by fork.
+ let Ok(mut registry) = PARQUET_METADATA_SLOTS.try_lock() else {
+ return Arc::new(OnceLock::new());
+ };
+ let slots = registry.get_or_insert_with(HashMap::new);
+ slots.retain(|_, slot| slot.strong_count() > 0);
+ let key = (std::process::id(), max_bytes);
+ if let Some(slot) = slots.get(&key).and_then(std::sync::Weak::upgrade) {
+ return slot;
+ }
+ let slot = Arc::new(OnceLock::new());
+ slots.insert(key, Arc::downgrade(&slot));
+ slot
+}
+
+#[cfg(test)]
+pub(crate) fn run_isolated_metadata_cache_test(test_name: &str) -> bool {
+ const CHILD: &str = "PAIMON_TEST_ISOLATED_METADATA_CACHE";
+ if std::env::var(CHILD).as_deref() == Ok(test_name) {
+ return true;
+ }
+ // Registry contention legitimately disables sharing. Tests requiring a
+ // shared slot must not race unrelated tests constructing cache contexts.
+ let status = std::process::Command::new(std::env::current_exe().unwrap())
+ .args(["--exact", test_name, "--test-threads=1"])
+ .env(CHILD, test_name)
+ .status()
+ .unwrap();
+ assert!(
+ status.success(),
+ "isolated test {test_name} failed: {status}"
+ );
+ false
+}
+
pub(crate) struct FileFormatMetadataCacheContext {
max_bytes: usize,
- cache: OnceLock<Arc<dyn Any + Send + Sync>>,
+ shared: bool,
+ cache: Arc<ParquetMetadataSlot>,
row_metadata_cache: OnceLock<Arc<dyn Any + Send + Sync>>,
process_id: u32,
+ memory_identity: uuid::Uuid,
row_sidecar_sizes: OnceLock<Mutex<LruCache<String, u64>>>,
}
@@ -153,11 +195,30 @@ impl FileFormatMetadataCacheContext {
})
.transpose()?
.unwrap_or(DEFAULT_FILE_FORMAT_METADATA_CACHE_MAX_BYTES);
+ let shared = props
+ .get(CatalogOptions::FILE_FORMAT_METADATA_CACHE_SHARED)
+ .map(|value| {
+ value.parse::<bool>().map_err(|_| Error::ConfigInvalid {
+ message: format!(
+ "Invalid value for {}: {value}",
+ CatalogOptions::FILE_FORMAT_METADATA_CACHE_SHARED
+ ),
+ })
+ })
+ .transpose()?
+ .unwrap_or(false);
+ let cache = if shared && max_bytes > 0 {
+ shared_parquet_metadata_slot(max_bytes)
+ } else {
+ Arc::new(OnceLock::new())
+ };
Ok(Arc::new(Self {
max_bytes,
- cache: OnceLock::new(),
+ shared,
+ cache,
row_metadata_cache: OnceLock::new(),
process_id: std::process::id(),
+ memory_identity: uuid::Uuid::new_v4(),
row_sidecar_sizes: OnceLock::new(),
}))
}
@@ -231,6 +292,9 @@ impl FileFormatMetadataCacheContext {
where
T: Send + Sync + 'static,
{
+ if self.process_id != std::process::id() {
+ return Arc::new(init(self.parquet_metadata_max_bytes()));
+ }
let cache = Arc::clone(self.cache.get_or_init(|| {
Arc::new(init(self.parquet_metadata_max_bytes())) as Arc<dyn Any +
Send + Sync>
}));
@@ -240,6 +304,10 @@ impl FileFormatMetadataCacheContext {
}
}
+ pub(crate) fn parquet_metadata_is_shared(&self) -> bool {
+ self.shared && self.process_id == std::process::id()
+ }
+
/// 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>(
@@ -1176,6 +1244,14 @@ impl FileIOBuilder {
pub trait FileRead: Send + Sync + Unpin + 'static {
async fn read(&self, range: Range<u64>) -> crate::Result<Bytes>;
+ /// Check current access before reusing metadata from another context.
+ /// Implementations must bypass data caches and propagate storage errors.
+ /// Returning false disables cross-context metadata reuse for this reader.
+ #[doc(hidden)]
+ async fn check_shared_metadata_access(&self) -> crate::Result<bool> {
+ Ok(false)
+ }
+
#[doc(hidden)]
#[deprecated(note = "BLOB index caching no longer uses numeric FileIO
namespaces")]
fn cache_namespace(&self) -> Option<usize> {
@@ -1200,6 +1276,11 @@ pub trait FileRead: Send + Sync + Unpin + 'static {
#[async_trait::async_trait]
impl FileRead for opendal::Reader {
+ async fn check_shared_metadata_access(&self) -> crate::Result<bool> {
+ self.read(0..1).await?;
+ Ok(true)
+ }
+
async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
Ok(opendal::Reader::read(self, range).await?.to_bytes())
}
@@ -1222,6 +1303,13 @@ enum InputFileReader {
#[async_trait::async_trait]
impl FileRead for InputFileReader {
+ async fn check_shared_metadata_access(&self) -> crate::Result<bool> {
+ match self {
+ Self::Direct { reader, .. } =>
reader.check_shared_metadata_access().await,
+ Self::Cached { reader, .. } =>
reader.check_shared_metadata_access().await,
+ }
+ }
+
async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
match self {
Self::Direct { reader, .. } => FileRead::read(reader, range).await,
@@ -1589,11 +1677,21 @@ impl InputFile {
async fn reader_with_size_hint(&self, size: Option<u64>) ->
crate::Result<impl FileRead> {
let (op, relative_path, cache_path) =
self.source.resolve(&self.path).await?;
+ let cache_key = cache_path.as_ref().map(|path| {
+ if op.info().scheme() == "memory" {
+ // OpenDAL's address-based Memory name can be recycled. Keep
+ // parsed metadata context-local without changing block-cache
+ // keys or invalidation for operators shared by several
FileIOs.
+ format!("{}:{path}",
self.file_format_metadata_cache.memory_identity)
+ } else {
+ path.clone()
+ }
+ });
let reader = op.reader(&relative_path).await?;
let (Some(cache), Some(cache_path)) = (&self.cache,
cache_path.as_ref()) else {
return Ok(InputFileReader::Direct {
reader,
- cache_key: cache_path,
+ cache_key,
file_format_metadata_cache:
Arc::clone(&self.file_format_metadata_cache),
blob_index_cache: Arc::clone(&self.blob_index_cache),
});
@@ -1616,7 +1714,7 @@ impl InputFile {
cache.clone(),
read_token,
),
- cache_key: cache_path.clone(),
+ cache_key: cache_key.unwrap(),
file_format_metadata_cache:
Arc::clone(&self.file_format_metadata_cache),
blob_index_cache: Arc::clone(&self.blob_index_cache),
})
@@ -2693,6 +2791,38 @@ mod input_output_test {
);
}
+ #[cfg(feature = "storage-memory")]
+ #[tokio::test]
+ async fn test_memory_metadata_keys_are_context_scoped() {
+ let first = setup_memory_file_io();
+ let path = "memory:/metadata.parquet";
+ let first_reader =
first.new_input(path).unwrap().reader().await.unwrap();
+ let clone = first.clone();
+ let cloned_reader =
clone.new_input(path).unwrap().reader().await.unwrap();
+ assert_eq!(first_reader.cache_key(), cloned_reader.cache_key());
+ let second = setup_memory_file_io();
+ let second_reader =
second.new_input(path).unwrap().reader().await.unwrap();
+ assert_ne!(first_reader.cache_key(), second_reader.cache_key());
+
+ let operator =
Operator::from_config(opendal::services::MemoryConfig::default()).unwrap();
+ let first = FileIOBuilder::new("file")
+ .with_fs_operator(operator.clone())
+ .build()
+ .unwrap();
+ let second = FileIOBuilder::new("file")
+ .with_fs_operator(operator)
+ .build()
+ .unwrap();
+ let path = "file:/metadata.parquet";
+ let first_reader =
first.new_input(path).unwrap().reader().await.unwrap();
+ let second_reader =
second.new_input(path).unwrap().reader().await.unwrap();
+ assert_ne!(first_reader.cache_key(), second_reader.cache_key());
+ assert_eq!(
+ first.create_with_cache_path(path).await.unwrap().2,
+ second.create_with_cache_path(path).await.unwrap().2,
+ );
+ }
+
#[cfg(all(feature = "storage-fs", feature = "storage-memory"))]
#[test]
fn test_cache_object_path_distinguishes_storage_schemes() {
@@ -3172,6 +3302,119 @@ mod input_output_test {
.contains(CatalogOptions::FILE_FORMAT_METADATA_CACHE_MAX_SIZE));
}
+ #[test]
+ fn test_shared_metadata_cache_budget_lifetime_and_row_isolation() {
+ if !run_isolated_metadata_cache_test(
+
"io::file_io::input_output_test::test_shared_metadata_cache_budget_lifetime_and_row_isolation",
+ ) {
+ return;
+ }
+ let mut props = HashMap::from([
+ (
+ CatalogOptions::FILE_FORMAT_METADATA_CACHE_SHARED.to_string(),
+ "true".to_string(),
+ ),
+ (
+
CatalogOptions::FILE_FORMAT_METADATA_CACHE_MAX_SIZE.to_string(),
+ "123456".to_string(),
+ ),
+ ]);
+ let first =
FileFormatMetadataCacheContext::from_props(&props).unwrap();
+ let second =
FileFormatMetadataCacheContext::from_props(&props).unwrap();
+ assert!(Arc::ptr_eq(&first.cache, &second.cache));
+ let row_first = first.row_metadata_get_or_init(|_| 1usize).unwrap();
+ let row_second = second.row_metadata_get_or_init(|_| 2usize).unwrap();
+ assert!(!Arc::ptr_eq(&row_first, &row_second));
+ props.insert(
+ CatalogOptions::FILE_FORMAT_METADATA_CACHE_MAX_SIZE.to_string(),
+ "0".to_string(),
+ );
+ let zero = FileFormatMetadataCacheContext::from_props(&props).unwrap();
+ assert!(!Arc::ptr_eq(&first.cache, &zero.cache));
+ let weak = Arc::downgrade(&first.cache);
+ drop(first);
+ drop(second);
+ assert!(weak.upgrade().is_none());
+ }
+
+ #[test]
+ fn test_shared_metadata_cache_falls_back_under_registry_contention() {
+ if !run_isolated_metadata_cache_test(
+
"io::file_io::input_output_test::test_shared_metadata_cache_falls_back_under_registry_contention",
+ ) {
+ return;
+ }
+ let props = HashMap::from([(
+ CatalogOptions::FILE_FORMAT_METADATA_CACHE_SHARED.to_string(),
+ "true".to_string(),
+ )]);
+ let shared =
FileFormatMetadataCacheContext::from_props(&props).unwrap();
+ let guard = PARQUET_METADATA_SLOTS.lock().unwrap();
+ let (tx, rx) = std::sync::mpsc::channel();
+ let concurrent_props = props.clone();
+ let constructor = std::thread::spawn(move || {
+ tx.send(FileFormatMetadataCacheContext::from_props(
+ &concurrent_props,
+ ))
+ .unwrap();
+ });
+ let fallback = rx.recv_timeout(std::time::Duration::from_secs(5));
+ // Release the gate before joining, even if construction incorrectly
waited.
+ drop(guard);
+ constructor.join().unwrap();
+ let fallback = fallback
+ .expect("cache construction must not wait for the registry lock")
+ .unwrap();
+ assert!(!Arc::ptr_eq(&shared.cache, &fallback.cache));
+ let available =
FileFormatMetadataCacheContext::from_props(&props).unwrap();
+ assert!(Arc::ptr_eq(&shared.cache, &available.cache));
+ }
+
+ #[test]
+ fn test_shared_metadata_cache_rejects_invalid_flag() {
+ let props = HashMap::from([(
+ CatalogOptions::FILE_FORMAT_METADATA_CACHE_SHARED.to_string(),
+ "invalid".to_string(),
+ )]);
+ assert!(FileFormatMetadataCacheContext::from_props(&props).is_err());
+ }
+
+ #[cfg(unix)]
+ #[test]
+ fn test_shared_metadata_registry_fork_with_locked_registry() {
+ if !run_isolated_metadata_cache_test(
+
"io::file_io::input_output_test::test_shared_metadata_registry_fork_with_locked_registry",
+ ) {
+ return;
+ }
+ // The child must not wait for a lock whose owning thread stayed in
the parent.
+ let guard = PARQUET_METADATA_SLOTS.lock().unwrap();
+ unsafe {
+ let pid = libc::fork();
+ assert!(pid >= 0);
+ if pid == 0 {
+ libc::alarm(3);
+ let slot = shared_parquet_metadata_slot(123456);
+ libc::_exit(if slot.get().is_none() { 0 } else { 1 });
+ }
+ let mut status = 0;
+ assert_eq!(libc::waitpid(pid, &mut status, 0), pid);
+ assert!(libc::WIFEXITED(status));
+ assert_eq!(libc::WEXITSTATUS(status), 0);
+ }
+ drop(guard);
+ }
+
+ #[test]
+ fn test_metadata_cache_bypasses_inherited_context() {
+ let mut context =
FileFormatMetadataCacheContext::from_props(&HashMap::new()).unwrap();
+ let before = context.get_or_init(|_| 1usize);
+ Arc::get_mut(&mut context).unwrap().process_id = 0;
+ let after = context.get_or_init(|_| 2usize);
+ assert_eq!(*after, 2);
+ assert!(!Arc::ptr_eq(&before, &after));
+ }
+
#[test]
fn test_blob_index_cache_size_rejects_invalid_value() {
let error = FileIOBuilder::new("memory")
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 7c47cfcd..47209872 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -106,6 +106,13 @@ struct TimedFileRead {
#[async_trait::async_trait]
impl FileRead for TimedFileRead {
+ async fn check_shared_metadata_access(&self) -> crate::Result<bool> {
+ let start = Instant::now();
+ let result = self.inner.check_shared_metadata_access().await;
+ self.timing.add_file_read(start.elapsed());
+ result
+ }
+
async fn read(&self, range: Range<u64>) -> crate::Result<bytes::Bytes> {
let start = Instant::now();
let result = self.inner.read(range).await;
diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md
index b6c8ff9b..c75554f2 100644
--- a/docs/src/getting-started.md
+++ b/docs/src/getting-started.md
@@ -426,21 +426,23 @@ share LRU, size accounting, and invalidation state; if
their configured limits
differ, the smallest `local-cache.max-size` is used. Restart recovery runs on a
blocking worker, reads only block headers and file metadata, and validates
payload CRC lazily on the first hit. Without shared mode, use a separate
-`local-cache.dir` per worker or process.
-
-For sharing, set `local-cache.enabled=true` and `local-cache.shared=true` in
-same-UID workers with the same node-local directory and capacity. NFS,
-cross-node sharing, and catalogs inherited across `fork` are unsupported.
-Invalidation clears the shared cache; if it fails after a storage mutation,
-do not reuse the affected path until invalidation succeeds.
-A canceled or failed rename bypasses this cache; after confirming storage
state,
-use a new cache directory to resume sharing.
-
-Shared range reads allow eight background fills per cache instance for blocks
-up to 1 MiB; larger blocks, whole-file reads, or saturated fills stay
synchronous.
-Concurrent whole-file and range reads may download overlapping data.
-Allow disk headroom beyond `local-cache.max-size` for up to eight staged writes
-per cache instance; this staging limit is not node-wide.
+`local-cache.dir` per process.
+
+To share disk blocks, set `local-cache.enabled=true`,
`local-cache.shared=true`,
+`local-cache.dir`, and `local-cache.max-size` consistently in same-UID
processes
+on one node. NFS, cross-node sharing, and catalog reuse after `fork` are
+unsupported. After failed invalidation or rename, verify storage state and use
+a fresh cache directory before resuming sharing. Allow disk headroom for up to
+eight staged 1-MiB writes per cache instance beyond `local-cache.max-size`.
+
+### Parsed Parquet Metadata Cache
+
+Set `file-format.metadata-cache.shared=true` to reuse parsed Parquet metadata
+across read contexts in one process (default `false`). Only equal budgets share
+a pool; this is not a process-wide memory limit. Shared hits perform an
uncached
+one-byte storage-access check; readers without that check reload metadata.
+ROW metadata and BLOB indexes remain context-local. This is independent of
+`local-cache.shared`, which shares disk blocks across processes.
### BLOB Index Cache
