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 603ecdec [rust] Cache parsed Parquet metadata across readers (#929)
603ecdec is described below
commit 603ecdec20195a8071aa28404112f7fc94d3f2d4
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Sep 28 20:18:24 2026 +0800
[rust] Cache parsed Parquet metadata across readers (#929)
---
crates/paimon/src/arrow/format/metadata_cache.rs | 412 +++++++++++++++++++++
crates/paimon/src/arrow/format/mod.rs | 1 +
crates/paimon/src/arrow/format/parquet.rs | 377 +++++++++++++++++--
crates/paimon/src/catalog/rest/rest_catalog.rs | 8 +
.../paimon/src/catalog/rest/rest_token_file_io.rs | 18 +-
crates/paimon/src/common/options.rs | 8 +
crates/paimon/src/io/file_io.rs | 280 +++++++++++++-
crates/paimon/src/table/data_file_reader.rs | 8 +
crates/paimon/src/table/rest_env.rs | 72 +++-
crates/paimon/src/table/snapshot_manager.rs | 12 +-
.../paimon/src/table/table_commit/parity_tests.rs | 13 +-
.../src/table/table_commit/recovery_tests.rs | 13 +-
12 files changed, 1170 insertions(+), 52 deletions(-)
diff --git a/crates/paimon/src/arrow/format/metadata_cache.rs
b/crates/paimon/src/arrow/format/metadata_cache.rs
new file mode 100644
index 00000000..5bb140ad
--- /dev/null
+++ b/crates/paimon/src/arrow/format/metadata_cache.rs
@@ -0,0 +1,412 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use lru::LruCache;
+use std::future::Future;
+use std::hash::Hash;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
+use tokio::sync::OnceCell;
+
+struct Entry<V> {
+ value: OnceCell<Arc<V>>,
+ base_weight: usize,
+ weight: AtomicUsize,
+ users: AtomicUsize,
+}
+
+struct State<K, V> {
+ entries: LruCache<K, Arc<Entry<V>>>,
+ weight: usize,
+}
+
+pub(super) struct FileMetadataCache<K, V> {
+ max_bytes: usize,
+ max_entries: usize,
+ state: Mutex<State<K, V>>,
+}
+
+struct EntryGuard<'a, K, V>
+where
+ K: Clone + Eq + Hash,
+{
+ cache: &'a FileMetadataCache<K, V>,
+ key: K,
+ entry: Arc<Entry<V>>,
+}
+
+impl<'a, K, V> EntryGuard<'a, K, V>
+where
+ K: Clone + Eq + Hash,
+{
+ fn new(cache: &'a FileMetadataCache<K, V>, key: K, entry: Arc<Entry<V>>)
-> Self {
+ Self { cache, key, entry }
+ }
+}
+
+impl<K, V> Drop for EntryGuard<'_, K, V>
+where
+ K: Clone + Eq + Hash,
+{
+ fn drop(&mut self) {
+ self.cache.release(&self.key, &self.entry);
+ }
+}
+
+impl<K, V> FileMetadataCache<K, V>
+where
+ K: Clone + Eq + Hash,
+{
+ pub(super) fn new(max_bytes: usize, max_entries: usize) -> Self {
+ Self {
+ max_bytes,
+ max_entries,
+ state: Mutex::new(State {
+ entries: LruCache::unbounded(),
+ weight: 0,
+ }),
+ }
+ }
+
+ pub(super) fn entry_weight(key_heap_bytes: usize, value_weight: usize) ->
usize {
+ std::mem::size_of::<K>()
+ .saturating_add(std::mem::size_of::<Entry<V>>())
+ .saturating_add(key_heap_bytes)
+ .saturating_add(value_weight)
+ }
+
+ pub(super) async fn get_or_try_insert_with<E, F, Fut, W>(
+ &self,
+ key: Option<K>,
+ key_heap_bytes: usize,
+ load: F,
+ value_weight: W,
+ ) -> Result<Arc<V>, E>
+ where
+ F: FnOnce() -> Fut,
+ Fut: Future<Output = Result<Arc<V>, E>>,
+ W: FnOnce(&V) -> usize,
+ {
+ let Some(key) = key else {
+ return load().await;
+ };
+ let base_weight = Self::entry_weight(key_heap_bytes, 0);
+ if self.max_bytes == 0 || base_weight > self.max_bytes {
+ return load().await;
+ }
+
+ let entry = {
+ let mut state = self.state.lock().unwrap();
+ let entry = if let Some(entry) = state.entries.get(&key) {
+ Arc::clone(entry)
+ } else {
+ let entry = Arc::new(Entry {
+ value: OnceCell::new(),
+ base_weight,
+ weight: AtomicUsize::new(base_weight),
+ users: AtomicUsize::new(0),
+ });
+ state.weight = state.weight.saturating_add(base_weight);
+ state.entries.put(key.clone(), Arc::clone(&entry));
+ entry
+ };
+ entry.users.fetch_add(1, Ordering::Relaxed);
+ entry
+ };
+ let _guard = EntryGuard::new(self, key.clone(), Arc::clone(&entry));
+
+ let value = entry.value.get_or_try_init(load).await.map(Arc::clone)?;
+ if entry.weight.load(Ordering::Relaxed) != entry.base_weight {
+ return Ok(value);
+ }
+ let loaded_weight = entry
+ .base_weight
+ .saturating_add(value_weight(value.as_ref()).max(1));
+ let mut state = self.state.lock().unwrap();
+ if state
+ .entries
+ .peek(&key)
+ .is_some_and(|cached| Arc::ptr_eq(cached, &entry))
+ && entry.weight.load(Ordering::Relaxed) == entry.base_weight
+ {
+ if loaded_weight > self.max_bytes {
+ state.entries.pop(&key);
+ state.weight = state.weight.saturating_sub(entry.base_weight);
+ return Ok(value);
+ }
+ entry.weight.store(loaded_weight, Ordering::Relaxed);
+ state.weight = state
+ .weight
+ .saturating_sub(entry.base_weight)
+ .saturating_add(loaded_weight);
+ state.entries.promote(&key);
+ self.evict(&mut state);
+ }
+ Ok(value)
+ }
+
+ fn release(&self, key: &K, entry: &Arc<Entry<V>>) {
+ let mut state = self.state.lock().unwrap();
+ let users = entry.users.fetch_sub(1, Ordering::Relaxed);
+ debug_assert!(users > 0);
+ if users == 1
+ && entry.value.get().is_none()
+ && state
+ .entries
+ .peek(key)
+ .is_some_and(|cached| Arc::ptr_eq(cached, entry))
+ {
+ state.entries.pop(key);
+ state.weight = state
+ .weight
+ .saturating_sub(entry.weight.load(Ordering::Relaxed));
+ }
+ }
+
+ fn evict(&self, state: &mut State<K, V>) {
+ while state.weight > self.max_bytes || state.entries.len() >
self.max_entries {
+ let Some(key) = state
+ .entries
+ .iter()
+ .rev()
+ .find(|(_, entry)| entry.weight.load(Ordering::Relaxed) !=
entry.base_weight)
+ .map(|(key, _)| key.clone())
+ else {
+ // In-flight loads remain addressable so callers still
coalesce.
+ break;
+ };
+ let Some(entry) = state.entries.pop(&key) else {
+ state.weight = 0;
+ break;
+ };
+ state.weight = state
+ .weight
+ .saturating_sub(entry.weight.load(Ordering::Relaxed));
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use std::convert::Infallible;
+ use std::future::pending;
+ use std::time::Duration;
+
+ #[tokio::test]
+ async fn evicts_least_recently_used_entry_by_count() {
+ let cache = FileMetadataCache::<String, usize>::new(usize::MAX, 2);
+ let loads = AtomicUsize::new(0);
+
+ for key in ["first", "second", "third", "first"] {
+ cache
+ .get_or_try_insert_with(
+ Some(key.to_string()),
+ key.len(),
+ || async {
+ loads.fetch_add(1, Ordering::Relaxed);
+ Ok::<_, Infallible>(Arc::new(1))
+ },
+ |_| 1,
+ )
+ .await
+ .unwrap();
+ }
+
+ assert_eq!(loads.load(Ordering::Relaxed), 4);
+ }
+
+ #[tokio::test]
+ async fn cache_hit_does_not_recalculate_weight() {
+ let cache = FileMetadataCache::<String, usize>::new(1024, 2);
+ let loads = AtomicUsize::new(0);
+ let weights = AtomicUsize::new(0);
+
+ for _ in 0..2 {
+ cache
+ .get_or_try_insert_with(
+ Some("shared".to_string()),
+ 6,
+ || async {
+ loads.fetch_add(1, Ordering::Relaxed);
+ Ok::<_, Infallible>(Arc::new(7))
+ },
+ |_| {
+ weights.fetch_add(1, Ordering::Relaxed);
+ 7
+ },
+ )
+ .await
+ .unwrap();
+ }
+
+ assert_eq!(loads.load(Ordering::Relaxed), 1);
+ assert_eq!(weights.load(Ordering::Relaxed), 1);
+ }
+
+ #[tokio::test]
+ async fn removes_cancelled_load() {
+ let cache = Arc::new(FileMetadataCache::<String, usize>::new(1024, 2));
+ let task_cache = Arc::clone(&cache);
+ let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+
+ let task = tokio::spawn(async move {
+ task_cache
+ .get_or_try_insert_with(
+ Some("cancelled".to_string()),
+ 9,
+ || async move {
+ started_tx.send(()).unwrap();
+ pending::<Result<Arc<usize>, Infallible>>().await
+ },
+ |_| 1,
+ )
+ .await
+ });
+ started_rx.await.unwrap();
+ task.abort();
+ assert!(task.await.unwrap_err().is_cancelled());
+
+ let state = cache.state.lock().unwrap();
+ assert!(state.entries.is_empty());
+ assert_eq!(state.weight, 0);
+ }
+
+ #[tokio::test]
+ async fn waiter_caches_value_after_initializer_is_cancelled() {
+ let cache = Arc::new(FileMetadataCache::<String, usize>::new(1024, 2));
+ let loads = Arc::new(AtomicUsize::new(0));
+ let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+
+ let first_cache = Arc::clone(&cache);
+ let first_loads = Arc::clone(&loads);
+ let first = tokio::spawn(async move {
+ first_cache
+ .get_or_try_insert_with(
+ Some("shared".to_string()),
+ 6,
+ || async move {
+ first_loads.fetch_add(1, Ordering::Relaxed);
+ started_tx.send(()).unwrap();
+ pending::<Result<Arc<usize>, Infallible>>().await
+ },
+ |_| 1,
+ )
+ .await
+ });
+ started_rx.await.unwrap();
+
+ let second_cache = Arc::clone(&cache);
+ let second_loads = Arc::clone(&loads);
+ let (waiting_tx, waiting_rx) = tokio::sync::oneshot::channel();
+ let second = tokio::spawn(async move {
+ waiting_tx.send(()).unwrap();
+ second_cache
+ .get_or_try_insert_with(
+ Some("shared".to_string()),
+ 6,
+ || async move {
+ second_loads.fetch_add(1, Ordering::Relaxed);
+ Ok::<_, Infallible>(Arc::new(7))
+ },
+ |_| 1,
+ )
+ .await
+ });
+ waiting_rx.await.unwrap();
+ tokio::time::timeout(Duration::from_secs(1), async {
+ loop {
+ let users = cache
+ .state
+ .lock()
+ .unwrap()
+ .entries
+ .iter()
+ .next()
+ .unwrap()
+ .1
+ .users
+ .load(Ordering::Relaxed);
+ if users == 2 {
+ break;
+ }
+ tokio::task::yield_now().await;
+ }
+ })
+ .await
+ .unwrap();
+
+ first.abort();
+ assert!(first.await.unwrap_err().is_cancelled());
+ assert_eq!(*second.await.unwrap().unwrap(), 7);
+
+ let third_loads = Arc::clone(&loads);
+ let value = cache
+ .get_or_try_insert_with(
+ Some("shared".to_string()),
+ 6,
+ || async move {
+ third_loads.fetch_add(1, Ordering::Relaxed);
+ Ok::<_, Infallible>(Arc::new(8))
+ },
+ |_| 1,
+ )
+ .await
+ .unwrap();
+
+ assert_eq!(*value, 7);
+ assert_eq!(loads.load(Ordering::Relaxed), 2);
+ }
+
+ #[tokio::test]
+ async fn oversized_load_does_not_evict_cached_entries() {
+ let cache = FileMetadataCache::<String, usize>::new(512, 2);
+ let small_loads = AtomicUsize::new(0);
+ let oversized_loads = AtomicUsize::new(0);
+
+ for _ in 0..2 {
+ cache
+ .get_or_try_insert_with(
+ Some("small".to_string()),
+ 5,
+ || async {
+ small_loads.fetch_add(1, Ordering::Relaxed);
+ Ok::<_, Infallible>(Arc::new(1))
+ },
+ |value| *value,
+ )
+ .await
+ .unwrap();
+
+ cache
+ .get_or_try_insert_with(
+ Some("oversized".to_string()),
+ 9,
+ || async {
+ oversized_loads.fetch_add(1, Ordering::Relaxed);
+ Ok::<_, Infallible>(Arc::new(1024))
+ },
+ |value| *value,
+ )
+ .await
+ .unwrap();
+ }
+
+ assert_eq!(small_loads.load(Ordering::Relaxed), 1);
+ assert_eq!(oversized_loads.load(Ordering::Relaxed), 2);
+ }
+}
diff --git a/crates/paimon/src/arrow/format/mod.rs
b/crates/paimon/src/arrow/format/mod.rs
index a9b46f92..0e0d00bd 100644
--- a/crates/paimon/src/arrow/format/mod.rs
+++ b/crates/paimon/src/arrow/format/mod.rs
@@ -18,6 +18,7 @@
mod avro;
mod avro_write;
pub(crate) mod blob;
+mod metadata_cache;
mod mosaic;
mod mosaic_write;
mod orc;
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index 441fe049..5ce567bf 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use super::metadata_cache::FileMetadataCache;
use super::shredding::PhysicalFormatWriterFactory;
use super::{
timestamp_millis_schema, FilePredicates, FormatFileReader,
FormatFileWriter, FormatWriteResult,
@@ -60,6 +61,7 @@ use tokio::sync::{mpsc, oneshot};
pub(crate) struct ParquetFormatReader {
read_budget: Option<Arc<ReadBudget>>,
page_index_enabled: bool,
+ metadata_cache_enabled: bool,
}
impl Default for ParquetFormatReader {
@@ -67,6 +69,7 @@ impl Default for ParquetFormatReader {
Self {
read_budget: None,
page_index_enabled: true,
+ metadata_cache_enabled: true,
}
}
}
@@ -76,10 +79,12 @@ impl ParquetFormatReader {
options: &HashMap<String, String>,
read_budget: Option<Arc<ReadBudget>>,
) -> crate::Result<Self> {
+ let core_options = CoreOptions::new(options);
Ok(Self {
read_budget,
- page_index_enabled: crate::spec::CoreOptions::new(options)
- .parquet_filter_column_index_enabled()?,
+ page_index_enabled:
core_options.parquet_filter_column_index_enabled()?,
+ // Format-table files may be replaced in place.
+ metadata_cache_enabled: !core_options.is_format_table(),
})
}
@@ -607,7 +612,8 @@ impl FormatFileReader for ParquetFormatReader {
row_selection: Option<Vec<RowRange>>,
) -> crate::Result<ArrowRecordBatchStream> {
let shared_reader: Arc<dyn FileRead> = reader.into();
- let arrow_file_reader = ArrowFileReader::new(file_size,
Arc::clone(&shared_reader));
+ let arrow_file_reader = ArrowFileReader::new(file_size,
Arc::clone(&shared_reader))
+ .with_metadata_cache_enabled(self.metadata_cache_enabled);
let empty_predicates = Vec::new();
let (preds, file_fields): (&[Predicate], &[DataField]) = match
predicates {
@@ -2462,6 +2468,70 @@ fn build_row_ranges_selection(
// ArrowFileReader — async Parquet IO adapter
// ---------------------------------------------------------------------------
+const PARQUET_METADATA_CACHE_MAX_ENTRIES: usize = 4096;
+const PARQUET_METADATA_CACHE_MIN_ENTRY_BYTES: usize = 8 * 1024;
+
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+struct ParquetMetadataCacheKey {
+ file: String,
+ size: u64,
+ column_index: u8,
+ offset_index: u8,
+}
+
+struct ParquetMetadataCache {
+ inner: FileMetadataCache<ParquetMetadataCacheKey, ParquetMetaData>,
+}
+
+impl ParquetMetadataCache {
+ fn new(max_bytes: usize) -> Self {
+ Self {
+ inner: FileMetadataCache::new(max_bytes,
PARQUET_METADATA_CACHE_MAX_ENTRIES),
+ }
+ }
+
+ async fn load(
+ &self,
+ reader: &mut ArrowFileReader,
+ options: Option<&ArrowReaderOptions>,
+ ) -> 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());
+ let key = reader
+ .r
+ .cache_key()
+ .filter(|key| !key.is_empty())
+ .map(|file| ParquetMetadataCacheKey {
+ file: file.to_string(),
+ // Paimon data files are immutable; size separates supported
replacements.
+ size: reader.file_size,
+ column_index: page_index_policy_tag(column_index),
+ offset_index: page_index_policy_tag(offset_index),
+ });
+ 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,
+ || reader.load_metadata(options),
+ |metadata| {
+ metadata
+ .memory_size()
+ .max(PARQUET_METADATA_CACHE_MIN_ENTRY_BYTES)
+ },
+ )
+ .await
+ }
+}
+
+fn page_index_policy_tag(policy: PageIndexPolicy) -> u8 {
+ match policy {
+ PageIndexPolicy::Skip => 0,
+ PageIndexPolicy::Optional => 1,
+ PageIndexPolicy::Required => 2,
+ }
+}
+
/// ArrowFileReader is a wrapper around a FileRead that impls parquets
AsyncFileReader.
///
/// # TODO
@@ -2475,6 +2545,7 @@ fn build_row_ranges_selection(
struct ArrowFileReader {
file_size: u64,
r: Arc<dyn FileRead>,
+ metadata_cache_enabled: bool,
}
/// coalesce threshold: 1 MiB.
@@ -2494,10 +2565,10 @@ pub(crate) async fn read_row_count(
reader: Box<dyn FileRead>,
file_size: u64,
) -> crate::Result<i64> {
- let reader = ArrowFileReader::new(file_size, Arc::from(reader));
- let metadata = ParquetMetaDataReader::new()
- .with_prefetch_hint(Some(METADATA_SIZE_HINT))
- .load_and_finish(reader, file_size)
+ let mut reader =
+ ArrowFileReader::new(file_size,
Arc::from(reader)).with_metadata_cache_enabled(false);
+ let metadata = reader
+ .get_metadata(None)
.await
.map_err(|error| Error::UnexpectedError {
message: format!("Failed to read the Parquet footer: {error}"),
@@ -2508,7 +2579,16 @@ pub(crate) async fn read_row_count(
impl ArrowFileReader {
fn new(file_size: u64, r: Arc<dyn FileRead>) -> Self {
- Self { file_size, r }
+ Self {
+ file_size,
+ r,
+ metadata_cache_enabled: true,
+ }
+ }
+
+ fn with_metadata_cache_enabled(mut self, enabled: bool) -> Self {
+ self.metadata_cache_enabled = enabled;
+ self
}
fn read_bytes(&mut self, range: Range<u64>) -> BoxFuture<'_,
parquet::errors::Result<Bytes>> {
@@ -2517,6 +2597,26 @@ impl ArrowFileReader {
parquet::errors::ParquetError::External(err_msg.into())
}))
}
+
+ async fn load_metadata(
+ &mut self,
+ options: Option<&ArrowReaderOptions>,
+ ) -> parquet::errors::Result<Arc<ParquetMetaData>> {
+ let metadata_opts = options.map(|o| o.metadata_options().clone());
+ let column_index_policy = options.map(|o| o.column_index_policy());
+ let offset_index_policy = options.map(|o| o.offset_index_policy());
+ let file_size = self.file_size;
+ let mut reader = ParquetMetaDataReader::new()
+ .with_prefetch_hint(Some(METADATA_SIZE_HINT))
+ .with_metadata_options(metadata_opts);
+ if let Some(policy) = column_index_policy {
+ reader = reader.with_column_index_policy(policy);
+ }
+ if let Some(policy) = offset_index_policy {
+ reader = reader.with_offset_index_policy(policy);
+ }
+ Ok(Arc::new(reader.load_and_finish(self, file_size).await?))
+ }
}
impl MetadataFetch for ArrowFileReader {
@@ -2646,28 +2746,24 @@ impl AsyncFileReader for ArrowFileReader {
&mut self,
options: Option<&ArrowReaderOptions>,
) -> BoxFuture<'_, parquet::errors::Result<Arc<ParquetMetaData>>> {
- let metadata_opts = options.map(|o| o.metadata_options().clone());
- // The page-index policies live on `ArrowReaderOptions` directly, not
- // inside `metadata_options`, so they must be forwarded explicitly (the
- // upstream default `AsyncFileReader::get_metadata` does the same).
- // Without this, `with_page_index_policy` would silently no-op here and
- // no page index would ever be loaded.
- let column_index_policy = options.map(|o| o.column_index_policy());
- let offset_index_policy = options.map(|o| o.offset_index_policy());
- let prefetch_hint = Some(METADATA_SIZE_HINT);
+ let options = options.cloned();
+ let metadata_cache_enabled = self.metadata_cache_enabled;
Box::pin(async move {
- let file_size = self.file_size;
- let mut reader = ParquetMetaDataReader::new()
- .with_prefetch_hint(prefetch_hint)
- .with_metadata_options(metadata_opts);
- if let Some(policy) = column_index_policy {
- reader = reader.with_column_index_policy(policy);
+ if !metadata_cache_enabled {
+ return self.load_metadata(options.as_ref()).await;
}
- if let Some(policy) = offset_index_policy {
- reader = reader.with_offset_index_policy(policy);
+ let Some(context) =
self.r.file_format_metadata_cache().and_then(|cache| {
+
cache.downcast_ref::<crate::io::FileFormatMetadataCacheContext>()
+ }) else {
+ return self.load_metadata(options.as_ref()).await;
+ };
+ if context.max_bytes() == 0 {
+ return self.load_metadata(options.as_ref()).await;
}
- let metadata = reader.load_and_finish(self, file_size).await?;
- Ok(Arc::new(metadata))
+ context
+ .get_or_init(ParquetMetadataCache::new)
+ .load(self, options.as_ref())
+ .await
})
}
}
@@ -2808,6 +2904,7 @@ mod tests {
use arrow_buffer::{NullBuffer, OffsetBuffer, ScalarBuffer};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema
as ArrowSchema};
use futures::{StreamExt, TryStreamExt};
+ use parquet::arrow::async_reader::AsyncFileReader;
use parquet::basic::{Compression, GzipLevel, ZstdLevel};
use parquet::file::properties::{EnabledStatistics, WriterProperties};
use parquet::file::statistics::Statistics as ParquetStatistics;
@@ -4632,6 +4729,8 @@ mod tests {
ranges: Arc<std::sync::Mutex<Vec<std::ops::Range<u64>>>>,
resident_bytes: Arc<AtomicUsize>,
peak_resident_bytes: Arc<AtomicUsize>,
+ cache_key: Option<Arc<str>>,
+ metadata_cache: Option<Arc<crate::io::FileFormatMetadataCacheContext>>,
}
struct TrackedReadBuffer {
@@ -4659,9 +4758,44 @@ mod tests {
ranges: Arc::new(std::sync::Mutex::new(Vec::new())),
resident_bytes: Arc::new(AtomicUsize::new(0)),
peak_resident_bytes: Arc::new(AtomicUsize::new(0)),
+ cache_key: None,
+ metadata_cache: None,
}
}
+ fn with_cache_key(mut self, cache_key: &str) -> Self {
+ self.cache_key = Some(Arc::from(cache_key));
+ if self.metadata_cache.is_none() {
+ self.metadata_cache = Some(
+
crate::io::FileFormatMetadataCacheContext::from_props(&HashMap::new()).unwrap(),
+ );
+ }
+ self
+ }
+
+ fn with_metadata_cache_max_bytes(mut self, max_bytes: usize) -> Self {
+ self.metadata_cache = Some(
+
crate::io::FileFormatMetadataCacheContext::from_props(&HashMap::from([(
+
crate::common::CatalogOptions::FILE_FORMAT_METADATA_CACHE_MAX_SIZE.to_string(),
+ max_bytes.to_string(),
+ )]))
+ .unwrap(),
+ );
+ self
+ }
+
+ fn with_metadata_cache(
+ mut self,
+ cache: Arc<crate::io::FileFormatMetadataCacheContext>,
+ ) -> Self {
+ self.metadata_cache = Some(cache);
+ self
+ }
+
+ fn read_count(&self) -> usize {
+ self.ranges.lock().unwrap().len()
+ }
+
fn bytes_read(&self) -> u64 {
self.ranges
.lock()
@@ -4695,6 +4829,195 @@ mod tests {
resident_bytes: Arc::clone(&self.resident_bytes),
}))
}
+
+ fn cache_key(&self) -> Option<&str> {
+ self.cache_key.as_deref()
+ }
+
+ fn file_format_metadata_cache(&self) -> Option<&(dyn std::any::Any +
Send + Sync)> {
+ self.metadata_cache
+ .as_ref()
+ .map(|cache| cache.as_ref() as &(dyn std::any::Any + Send +
Sync))
+ }
+ }
+
+ #[tokio::test]
+ async fn parquet_metadata_cache_reuses_metadata_across_readers() {
+ let data = Bytes::from(
+ write_multi_row_group_parquet(8, 64, EnabledStatistics::Chunk,
false).await,
+ );
+ let tracker =
TrackingFileRead::new(data.clone()).with_cache_key("storage-a\0data.parquet");
+
+ for _ in 0..10 {
+ let mut reader =
+ super::ArrowFileReader::new(data.len() as u64,
Arc::new(tracker.clone()));
+ reader.get_metadata(None).await.unwrap();
+ }
+
+ assert_eq!(tracker.read_count(), 1);
+ }
+
+ #[cfg(feature = "storage-fs")]
+ #[tokio::test]
+ async fn parquet_metadata_cache_reuses_metadata_across_file_io_instances()
{
+ let data = Bytes::from(
+ write_multi_row_group_parquet(8, 64, EnabledStatistics::Chunk,
false).await,
+ );
+ let directory = tempfile::tempdir().unwrap();
+ let path = directory.path().join("data.parquet");
+ std::fs::write(&path, &data).unwrap();
+ let path = path.to_string_lossy().into_owned();
+ let context =
+
crate::io::FileFormatMetadataCacheContext::from_props(&HashMap::new()).unwrap();
+ let first = crate::io::FileIO::from_path(&path)
+ .unwrap()
+ .with_file_format_metadata_cache(context.clone())
+ .build()
+ .unwrap();
+ let second = crate::io::FileIO::from_path(&path)
+ .unwrap()
+ .with_file_format_metadata_cache(context)
+ .build()
+ .unwrap();
+
+ let mut first_reader = super::ArrowFileReader::new(
+ data.len() as u64,
+ Arc::new(first.new_input(&path).unwrap().reader().await.unwrap()),
+ );
+ let first_metadata = first_reader.get_metadata(None).await.unwrap();
+ let mut second_reader = super::ArrowFileReader::new(
+ data.len() as u64,
+ Arc::new(second.new_input(&path).unwrap().reader().await.unwrap()),
+ );
+ let second_metadata = second_reader.get_metadata(None).await.unwrap();
+
+ assert!(Arc::ptr_eq(&first_metadata, &second_metadata));
+ }
+
+ #[tokio::test]
+ async fn parquet_metadata_cache_can_be_bypassed() {
+ let data = Bytes::from(
+ write_multi_row_group_parquet(8, 64, EnabledStatistics::Chunk,
false).await,
+ );
+ let tracker =
TrackingFileRead::new(data.clone()).with_cache_key("storage-a\0data.parquet");
+ let unidentified = TrackingFileRead::new(data.clone());
+
+ for _ in 0..2 {
+ let mut reader =
+ super::ArrowFileReader::new(data.len() as u64,
Arc::new(tracker.clone()))
+ .with_metadata_cache_enabled(false);
+ reader.get_metadata(None).await.unwrap();
+ let mut reader =
+ super::ArrowFileReader::new(data.len() as u64,
Arc::new(unidentified.clone()));
+ reader.get_metadata(None).await.unwrap();
+ }
+
+ assert_eq!(tracker.read_count(), 2);
+ assert_eq!(unidentified.read_count(), 2);
+ }
+
+ #[tokio::test]
+ async fn format_table_parquet_metadata_bypasses_cache() {
+ let data = Bytes::from(
+ write_multi_row_group_parquet(8, 64, EnabledStatistics::Chunk,
false).await,
+ );
+ let tracker =
TrackingFileRead::new(data.clone()).with_cache_key("format-table.parquet");
+ let reader = super::ParquetFormatReader::with_options(
+ &HashMap::from([("type".to_string(), "format-table".to_string())]),
+ None,
+ )
+ .unwrap();
+
+ for _ in 0..2 {
+ let mut file_reader =
+ super::ArrowFileReader::new(data.len() as u64,
Arc::new(tracker.clone()));
+ file_reader.metadata_cache_enabled = reader.metadata_cache_enabled;
+ file_reader.get_metadata(None).await.unwrap();
+ }
+
+ assert_eq!(tracker.read_count(), 2);
+ }
+
+ #[tokio::test]
+ async fn parquet_row_count_bypasses_metadata_cache() {
+ let data = Bytes::from(
+ write_multi_row_group_parquet(8, 64, EnabledStatistics::Chunk,
false).await,
+ );
+ let tracker =
TrackingFileRead::new(data.clone()).with_cache_key("row-count.parquet");
+
+ for _ in 0..2 {
+ super::read_row_count(Box::new(tracker.clone()), data.len() as u64)
+ .await
+ .unwrap();
+ }
+
+ assert_eq!(tracker.read_count(), 2);
+ }
+
+ #[tokio::test]
+ async fn parquet_metadata_cache_zero_does_not_affect_other_contexts() {
+ let data = Bytes::from(
+ write_multi_row_group_parquet(8, 64, EnabledStatistics::Chunk,
false).await,
+ );
+ let tracker =
TrackingFileRead::new(data.clone()).with_cache_key("storage-a\0data.parquet");
+
+ let mut reader = super::ArrowFileReader::new(data.len() as u64,
Arc::new(tracker.clone()));
+ reader.get_metadata(None).await.unwrap();
+
+ let disabled = tracker.clone().with_metadata_cache_max_bytes(0);
+ let mut reader = super::ArrowFileReader::new(data.len() as u64,
Arc::new(disabled));
+ reader.get_metadata(None).await.unwrap();
+
+ let mut reader = super::ArrowFileReader::new(data.len() as u64,
Arc::new(tracker.clone()));
+ reader.get_metadata(None).await.unwrap();
+
+ assert_eq!(tracker.read_count(), 2);
+ }
+
+ #[tokio::test]
+ async fn parquet_metadata_cache_coalesces_concurrent_loads() {
+ let data = Bytes::from(
+ write_multi_row_group_parquet(8, 64, EnabledStatistics::Chunk,
false).await,
+ );
+ let tracker =
TrackingFileRead::new(data.clone()).with_cache_key("storage-a\0data.parquet");
+ let loads = (0..10).map(|_| {
+ let tracker = tracker.clone();
+ let data = data.clone();
+ async move {
+ let mut reader = super::ArrowFileReader::new(data.len() as
u64, Arc::new(tracker));
+ reader.get_metadata(None).await
+ }
+ });
+
+ futures::future::try_join_all(loads).await.unwrap();
+
+ assert_eq!(tracker.read_count(), 1);
+ }
+
+ #[tokio::test]
+ async fn parquet_metadata_cache_separates_storage_and_page_index_policy() {
+ let data = Bytes::from(
+ write_multi_row_group_parquet(8, 64, EnabledStatistics::Chunk,
false).await,
+ );
+ let cache =
crate::io::FileFormatMetadataCacheContext::from_props(&HashMap::new()).unwrap();
+ let first = TrackingFileRead::new(data.clone())
+ .with_cache_key("storage-a\0data.parquet")
+ .with_metadata_cache(cache.clone());
+ let second = TrackingFileRead::new(data.clone())
+ .with_cache_key("storage-b\0data.parquet")
+ .with_metadata_cache(cache);
+
+ let mut reader = super::ArrowFileReader::new(data.len() as u64,
Arc::new(first.clone()));
+ reader.get_metadata(None).await.unwrap();
+ let options =
+
super::ArrowReaderOptions::new().with_offset_index_policy(PageIndexPolicy::Optional);
+ let mut reader = super::ArrowFileReader::new(data.len() as u64,
Arc::new(first.clone()));
+ reader.get_metadata(Some(&options)).await.unwrap();
+ let mut reader = super::ArrowFileReader::new(data.len() as u64,
Arc::new(second.clone()));
+ reader.get_metadata(None).await.unwrap();
+
+ assert_eq!(first.read_count(), 2);
+ assert_eq!(second.read_count(), 1);
}
#[tokio::test]
diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs
b/crates/paimon/src/catalog/rest/rest_catalog.rs
index a7e410cb..de42003d 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -37,6 +37,7 @@ use crate::catalog::{
use crate::common::{CatalogOptions, Options};
use crate::error::Error;
use crate::io::cache::{create_local_cache_with_namespace, LocalCache};
+use crate::io::FileFormatMetadataCacheContext;
use crate::spec::{Partition, PartitionStatistics, Schema, SchemaChange};
use crate::table::{RESTEnv, Table};
use crate::Result;
@@ -60,6 +61,7 @@ pub struct RESTCatalog {
data_token_enabled: bool,
/// Catalog-scoped cache shared by all table FileIO instances.
local_cache: Option<Arc<LocalCache>>,
+ file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
}
impl RESTCatalog {
@@ -88,6 +90,8 @@ impl RESTCatalog {
let api_options = api.options().clone();
let local_cache = create_local_cache_with_namespace(&options,
&api_options)?;
+ let file_format_metadata_cache =
+ FileFormatMetadataCacheContext::from_props(api_options.to_map())?;
Ok(Self {
api,
@@ -95,6 +99,7 @@ impl RESTCatalog {
warehouse,
data_token_enabled,
local_cache,
+ file_format_metadata_cache,
})
}
@@ -280,6 +285,7 @@ impl Catalog for RESTCatalog {
self.options.clone(),
self.data_token_enabled,
self.local_cache.clone(),
+ self.file_format_metadata_cache.clone(),
)
.await
}
@@ -297,6 +303,7 @@ impl Catalog for RESTCatalog {
self.options.clone(),
self.data_token_enabled,
self.local_cache.clone(),
+ self.file_format_metadata_cache.clone(),
)
.await
.map(crate::catalog::LoadedTable::Object);
@@ -317,6 +324,7 @@ impl Catalog for RESTCatalog {
self.options.clone(),
self.data_token_enabled,
self.local_cache.clone(),
+ self.file_format_metadata_cache.clone(),
)
.await
.map(|table| crate::catalog::LoadedTable::Paimon(Box::new(table)))
diff --git a/crates/paimon/src/catalog/rest/rest_token_file_io.rs
b/crates/paimon/src/catalog/rest/rest_token_file_io.rs
index d7d17f78..7f106011 100644
--- a/crates/paimon/src/catalog/rest/rest_token_file_io.rs
+++ b/crates/paimon/src/catalog/rest/rest_token_file_io.rs
@@ -31,7 +31,7 @@ use crate::api::rest_util::RESTUtil;
use crate::catalog::Identifier;
use crate::common::{CatalogOptions, Options};
use crate::io::cache::LocalCache;
-use crate::io::{FileIO, FileIOProvider};
+use crate::io::{FileFormatMetadataCacheContext, FileIO, FileIOProvider};
use crate::Result;
use super::rest_token::RESTToken;
@@ -55,6 +55,7 @@ pub struct RESTTokenFileIO {
state: RwLock<Option<TokenState>>,
refresh_lock: Mutex<()>,
local_cache: Option<Arc<LocalCache>>,
+ file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
}
impl std::fmt::Debug for RESTTokenFileIO {
@@ -73,6 +74,7 @@ impl RESTTokenFileIO {
catalog_options: Options,
api: Arc<RESTApi>,
local_cache: Option<Arc<LocalCache>>,
+ file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
) -> Self {
Self {
identifier,
@@ -82,6 +84,7 @@ impl RESTTokenFileIO {
state: RwLock::new(None),
refresh_lock: Mutex::new(()),
local_cache,
+ file_format_metadata_cache,
}
}
@@ -121,6 +124,7 @@ impl RESTTokenFileIO {
fn build_static_file_io(&self, token: &RESTToken) -> Result<FileIO> {
let merged_props =
RESTUtil::merge(Some(self.catalog_options.to_map()), Some(&token.token));
let mut builder =
FileIO::from_path(&self.path)?.with_props(merged_props);
+ builder =
builder.with_file_format_metadata_cache(self.file_format_metadata_cache.clone());
if let Some(local_cache) = &self.local_cache {
builder = builder.with_local_cache(local_cache.clone());
}
@@ -229,18 +233,24 @@ mod tests {
(options, api, requests, server)
}
+ fn metadata_cache(options: &Options) ->
Arc<FileFormatMetadataCacheContext> {
+ FileFormatMetadataCacheContext::from_props(options.to_map()).unwrap()
+ }
+
#[tokio::test]
async fn test_token_file_io_keeps_catalog_local_cache() {
let table_directory = tempfile::tempdir().unwrap();
let (mut options, api, _, server) = token_api().await;
options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
let local_cache = create_local_cache(&options).unwrap();
+ let metadata_cache = metadata_cache(&options);
let token_file_io = Arc::new(RESTTokenFileIO::new(
Identifier::new("database", "table"),
table_directory.path().to_string_lossy().into_owned(),
options,
api,
local_cache,
+ metadata_cache,
));
let file_io = token_file_io.build_file_io().await.unwrap();
@@ -254,12 +264,14 @@ mod tests {
let table_directory = tempfile::tempdir().unwrap();
let file_path = table_directory.path().join("data");
let (options, api, requests, server) = token_api().await;
+ let metadata_cache = metadata_cache(&options);
let token_file_io = Arc::new(RESTTokenFileIO::new(
Identifier::new("database", "table"),
table_directory.path().to_string_lossy().into_owned(),
options,
api,
None,
+ metadata_cache,
));
let file_io = token_file_io.build_file_io().await.unwrap();
@@ -301,12 +313,14 @@ mod tests {
let file_path = table_directory.path().join("blob");
std::fs::write(&file_path, b"abcdefghij").unwrap();
let (options, api, requests, server) = token_api().await;
+ let metadata_cache = metadata_cache(&options);
let token_file_io = Arc::new(RESTTokenFileIO::new(
Identifier::new("database", "table"),
table_directory.path().to_string_lossy().into_owned(),
options,
api,
None,
+ metadata_cache,
));
let file_io = token_file_io.build_file_io().await.unwrap();
@@ -329,12 +343,14 @@ mod tests {
let file_path = table_directory.path().join("blob");
std::fs::write(&file_path, b"abcdefghij").unwrap();
let (options, api, requests, server) = token_api().await;
+ let metadata_cache = metadata_cache(&options);
let token_file_io = Arc::new(RESTTokenFileIO::new(
Identifier::new("database", "table"),
table_directory.path().to_string_lossy().into_owned(),
options,
api,
None,
+ metadata_cache,
));
let file_io = token_file_io.build_file_io().await.unwrap();
diff --git a/crates/paimon/src/common/options.rs
b/crates/paimon/src/common/options.rs
index 5bb8c121..0110b80c 100644
--- a/crates/paimon/src/common/options.rs
+++ b/crates/paimon/src/common/options.rs
@@ -90,6 +90,10 @@ impl CatalogOptions {
/// Comma-separated file types eligible for local caching.
pub const LOCAL_CACHE_WHITELIST: &'static str = "local-cache.whitelist";
+
+ /// Maximum process-local cache size for parsed file-format metadata.
+ pub const FILE_FORMAT_METADATA_CACHE_MAX_SIZE: &'static str =
+ "file-format.metadata-cache.max-size";
}
/// Configuration options container.
@@ -280,6 +284,10 @@ mod tests {
CatalogOptions::LOCAL_CACHE_WHITELIST,
"local-cache.whitelist"
);
+ assert_eq!(
+ CatalogOptions::FILE_FORMAT_METADATA_CACHE_MAX_SIZE,
+ "file-format.metadata-cache.max-size"
+ );
}
#[test]
diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index 8d4b966a..7dba9fe8 100644
--- a/crates/paimon/src/io/file_io.rs
+++ b/crates/paimon/src/io/file_io.rs
@@ -15,13 +15,15 @@
// specific language governing permissions and limitations
// under the License.
+use crate::common::options::{parse_memory_size, CatalogOptions};
use crate::error::*;
+use std::any::Any;
use std::collections::HashMap;
use std::future::Future;
use std::ops::Range;
use std::pin::Pin;
use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::Arc;
+use std::sync::{Arc, OnceLock};
use std::task::{Context, Poll};
use std::time::SystemTime;
@@ -92,6 +94,7 @@ pub struct FileIO {
backend: FileIOBackend,
cache: Option<Arc<LocalCache>>,
cache_namespace: usize,
+ file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
}
static NEXT_FILE_IO_CACHE_NAMESPACE: AtomicUsize = AtomicUsize::new(1);
@@ -100,11 +103,72 @@ fn next_file_io_cache_namespace() -> usize {
NEXT_FILE_IO_CACHE_NAMESPACE.fetch_add(1, Ordering::Relaxed)
}
+pub(crate) const DEFAULT_FILE_FORMAT_METADATA_CACHE_MAX_BYTES: usize = 50 *
1024 * 1024;
+
+pub(crate) struct FileFormatMetadataCacheContext {
+ max_bytes: usize,
+ cache: OnceLock<Arc<dyn Any + Send + Sync>>,
+}
+
+impl std::fmt::Debug for FileFormatMetadataCacheContext {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("FileFormatMetadataCacheContext")
+ .field("max_bytes", &self.max_bytes)
+ .finish_non_exhaustive()
+ }
+}
+
+impl FileFormatMetadataCacheContext {
+ pub(crate) fn from_props(props: &HashMap<String, String>) ->
crate::Result<Arc<Self>> {
+ let max_bytes = props
+ .get(CatalogOptions::FILE_FORMAT_METADATA_CACHE_MAX_SIZE)
+ .map(|value| {
+ parse_memory_size(value)
+ .ok()
+ .and_then(|bytes| usize::try_from(bytes).ok())
+ .ok_or_else(|| Error::ConfigInvalid {
+ message: format!(
+ "Invalid value for {}: {value}",
+ CatalogOptions::FILE_FORMAT_METADATA_CACHE_MAX_SIZE
+ ),
+ })
+ })
+ .transpose()?
+ .unwrap_or(DEFAULT_FILE_FORMAT_METADATA_CACHE_MAX_BYTES);
+ Ok(Arc::new(Self {
+ max_bytes,
+ cache: OnceLock::new(),
+ }))
+ }
+
+ pub(crate) fn max_bytes(&self) -> usize {
+ self.max_bytes
+ }
+
+ 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>),
+ );
+ match cache.downcast::<T>() {
+ Ok(cache) => cache,
+ Err(_) => panic!("file-format metadata cache type mismatch"),
+ }
+ }
+}
+
impl std::fmt::Debug for FileIO {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("FileIO")
.field("backend", &self.backend)
.field("cache", &self.cache)
+ .field(
+ "file_format_metadata_cache_max_bytes",
+ &self.file_format_metadata_cache.max_bytes(),
+ )
.finish()
}
}
@@ -138,6 +202,11 @@ impl FileIO {
self
}
+ #[cfg(test)]
+ pub(crate) fn file_format_metadata_cache(&self) ->
Arc<FileFormatMetadataCacheContext> {
+ Arc::clone(&self.file_format_metadata_cache)
+ }
+
pub(crate) fn create_static(&self, path: &str) -> crate::Result<(Operator,
String)> {
let FileIOBackend::Storage(storage) = &self.backend else {
return Err(Error::IoUnsupported {
@@ -221,6 +290,7 @@ impl FileIO {
source: self.file_source(path)?,
path: path.to_string(),
cache_namespace: self.cache_namespace,
+ file_format_metadata_cache:
Arc::clone(&self.file_format_metadata_cache),
cache: self
.cache
.as_ref()
@@ -238,6 +308,7 @@ impl FileIO {
source: self.file_source(path)?,
path: path.to_string(),
cache_namespace: self.cache_namespace,
+ file_format_metadata_cache:
Arc::clone(&self.file_format_metadata_cache),
cache: self
.cache
.as_ref()
@@ -617,6 +688,7 @@ pub struct FileIOBuilder {
scheme_str: Option<String>,
props: HashMap<String, String>,
cache: Option<Arc<LocalCache>>,
+ file_format_metadata_cache: Option<Arc<FileFormatMetadataCacheContext>>,
operator: Option<Operator>,
provider: Option<Arc<dyn FileIOProvider>>,
}
@@ -627,6 +699,7 @@ impl FileIOBuilder {
scheme_str: Some(scheme_str.to_string()),
props: HashMap::default(),
cache: None,
+ file_format_metadata_cache: None,
operator: None,
provider: None,
}
@@ -681,8 +754,21 @@ impl FileIOBuilder {
self
}
+ pub(crate) fn with_file_format_metadata_cache(
+ mut self,
+ cache: Arc<FileFormatMetadataCacheContext>,
+ ) -> Self {
+ self.file_format_metadata_cache = Some(cache);
+ self
+ }
+
pub fn build(mut self) -> crate::Result<FileIO> {
let cache = self.cache.clone();
+ let file_format_metadata_cache = self
+ .file_format_metadata_cache
+ .clone()
+ .map(Ok)
+ .unwrap_or_else(||
FileFormatMetadataCacheContext::from_props(&self.props))?;
let backend = if let Some(provider) = self.provider.take() {
if self.operator.is_some() {
return Err(Error::ConfigInvalid {
@@ -697,6 +783,7 @@ impl FileIOBuilder {
backend,
cache,
cache_namespace: next_file_io_cache_namespace(),
+ file_format_metadata_cache,
})
}
}
@@ -709,6 +796,16 @@ pub trait FileRead: Send + Sync + Unpin + 'static {
fn cache_namespace(&self) -> Option<usize> {
None
}
+
+ /// Stable identity of an immutable file within one storage context.
+ fn cache_key(&self) -> Option<&str> {
+ None
+ }
+
+ #[doc(hidden)]
+ fn file_format_metadata_cache(&self) -> Option<&(dyn Any + Send + Sync)> {
+ None
+ }
}
#[async_trait::async_trait]
@@ -719,22 +816,56 @@ impl FileRead for opendal::Reader {
}
enum InputFileReader {
- Direct(opendal::Reader, usize),
- Cached(CachedFileReader, usize),
+ Direct {
+ reader: opendal::Reader,
+ cache_namespace: usize,
+ cache_key: String,
+ file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
+ },
+ Cached {
+ reader: CachedFileReader,
+ cache_namespace: usize,
+ cache_key: String,
+ file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
+ },
}
#[async_trait::async_trait]
impl FileRead for InputFileReader {
async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
match self {
- Self::Direct(reader, _) => FileRead::read(reader, range).await,
- Self::Cached(reader, _) => FileRead::read(reader, range).await,
+ Self::Direct { reader, .. } => FileRead::read(reader, range).await,
+ Self::Cached { reader, .. } => FileRead::read(reader, range).await,
+ }
+ }
+
+ fn cache_key(&self) -> Option<&str> {
+ match self {
+ Self::Direct { cache_key, .. } | Self::Cached { cache_key, .. } =>
Some(cache_key),
+ }
+ }
+
+ fn file_format_metadata_cache(&self) -> Option<&(dyn Any + Send + Sync)> {
+ match self {
+ Self::Direct {
+ file_format_metadata_cache,
+ ..
+ }
+ | Self::Cached {
+ file_format_metadata_cache,
+ ..
+ } => Some(file_format_metadata_cache.as_ref()),
}
}
fn cache_namespace(&self) -> Option<usize> {
Some(match self {
- Self::Direct(_, namespace) | Self::Cached(_, namespace) =>
*namespace,
+ Self::Direct {
+ cache_namespace, ..
+ }
+ | Self::Cached {
+ cache_namespace, ..
+ } => *cache_namespace,
})
}
}
@@ -874,6 +1005,7 @@ pub struct InputFile {
source: FileSource,
path: String,
cache_namespace: usize,
+ file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
cache: Option<Arc<LocalCache>>,
}
@@ -924,7 +1056,12 @@ impl InputFile {
let (op, relative_path, cache_path) =
self.source.resolve(&self.path).await?;
let reader = op.reader(&relative_path).await?;
let Some(cache) = &self.cache else {
- return Ok(InputFileReader::Direct(reader, self.cache_namespace));
+ return Ok(InputFileReader::Direct {
+ reader,
+ cache_namespace: self.cache_namespace,
+ cache_key: cache_path,
+ file_format_metadata_cache:
Arc::clone(&self.file_format_metadata_cache),
+ });
};
let read_token = cache.read_token(&cache_path);
let size = if let Some(size) = cache.file_size(&cache_path,
&read_token).await {
@@ -934,16 +1071,18 @@ impl InputFile {
cache.put_file_size(&cache_path, size, &read_token).await;
size
};
- Ok(InputFileReader::Cached(
- CachedFileReader::new_with_token(
+ Ok(InputFileReader::Cached {
+ reader: CachedFileReader::new_with_token(
Arc::new(reader),
&cache_path,
size,
cache.clone(),
read_token,
),
- self.cache_namespace,
- ))
+ cache_namespace: self.cache_namespace,
+ cache_key: cache_path,
+ file_format_metadata_cache:
Arc::clone(&self.file_format_metadata_cache),
+ })
}
}
@@ -952,6 +1091,7 @@ pub struct OutputFile {
source: FileSource,
path: String,
cache_namespace: usize,
+ file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
cache: Option<Arc<LocalCache>>,
}
@@ -971,6 +1111,7 @@ impl OutputFile {
source: self.source,
path: self.path,
cache_namespace: self.cache_namespace,
+ file_format_metadata_cache: self.file_format_metadata_cache,
cache,
}
}
@@ -1736,6 +1877,123 @@ mod input_output_test {
common_test_input_file_partial_read(&file_io,
"memory:/test_file_part_read_mem").await;
}
+ #[tokio::test]
+ async fn test_file_format_metadata_cache_is_scoped_to_file_io_context() {
+ let path = "memory:/cache-key.parquet";
+ let first = setup_memory_file_io();
+ first
+ .new_output(path)
+ .unwrap()
+ .write(Bytes::from_static(b"first"))
+ .await
+ .unwrap();
+ let first_reader =
first.new_input(path).unwrap().reader().await.unwrap();
+ let first_key = first_reader.cache_key().unwrap().to_string();
+ let first_cache = first_reader
+ .file_format_metadata_cache()
+ .unwrap()
+ .downcast_ref::<FileFormatMetadataCacheContext>()
+ .unwrap();
+ let clone_reader = first
+ .clone()
+ .new_input(path)
+ .unwrap()
+ .reader()
+ .await
+ .unwrap();
+ let clone_key = clone_reader.cache_key().unwrap().to_string();
+ let clone_cache = clone_reader
+ .file_format_metadata_cache()
+ .unwrap()
+ .downcast_ref::<FileFormatMetadataCacheContext>()
+ .unwrap();
+
+ let second = setup_memory_file_io();
+ second
+ .new_output(path)
+ .unwrap()
+ .write(Bytes::from_static(b"second"))
+ .await
+ .unwrap();
+ let second_reader =
second.new_input(path).unwrap().reader().await.unwrap();
+ let second_key = second_reader.cache_key().unwrap().to_string();
+ let second_cache = second_reader
+ .file_format_metadata_cache()
+ .unwrap()
+ .downcast_ref::<FileFormatMetadataCacheContext>()
+ .unwrap();
+
+ assert_eq!(first_key, clone_key);
+ assert_ne!(first_key, second_key);
+ assert!(std::ptr::eq(first_cache, clone_cache));
+ assert!(!std::ptr::eq(first_cache, second_cache));
+ }
+
+ #[tokio::test]
+ async fn test_file_format_metadata_cache_size_reaches_file_reader() {
+ let path = "memory:/metadata-cache-size.parquet";
+ let default = setup_memory_file_io();
+ default
+ .new_output(path)
+ .unwrap()
+ .write(Bytes::from_static(b"data"))
+ .await
+ .unwrap();
+ assert_eq!(
+ default
+ .new_input(path)
+ .unwrap()
+ .reader()
+ .await
+ .unwrap()
+ .file_format_metadata_cache()
+ .unwrap()
+ .downcast_ref::<FileFormatMetadataCacheContext>()
+ .unwrap()
+ .max_bytes(),
+ DEFAULT_FILE_FORMAT_METADATA_CACHE_MAX_BYTES
+ );
+
+ let configured = FileIOBuilder::new("memory")
+ .with_prop(CatalogOptions::FILE_FORMAT_METADATA_CACHE_MAX_SIZE, "1
mb")
+ .build()
+ .unwrap();
+ configured
+ .new_output(path)
+ .unwrap()
+ .write(Bytes::from_static(b"data"))
+ .await
+ .unwrap();
+ assert_eq!(
+ configured
+ .new_input(path)
+ .unwrap()
+ .reader()
+ .await
+ .unwrap()
+ .file_format_metadata_cache()
+ .unwrap()
+ .downcast_ref::<FileFormatMetadataCacheContext>()
+ .unwrap()
+ .max_bytes(),
+ 1024 * 1024
+ );
+ }
+
+ #[test]
+ fn test_file_format_metadata_cache_size_rejects_invalid_value() {
+ let error = FileIOBuilder::new("memory")
+ .with_prop(
+ CatalogOptions::FILE_FORMAT_METADATA_CACHE_MAX_SIZE,
+ "invalid",
+ )
+ .build()
+ .unwrap_err();
+ assert!(error
+ .to_string()
+ .contains(CatalogOptions::FILE_FORMAT_METADATA_CACHE_MAX_SIZE));
+ }
+
#[tokio::test]
async fn test_output_file_write_and_read_fs() {
let file_io = setup_fs_file_io();
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 009baf49..e608927c 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -111,6 +111,14 @@ impl FileRead for TimedFileRead {
fn cache_namespace(&self) -> Option<usize> {
self.inner.cache_namespace()
}
+
+ fn cache_key(&self) -> Option<&str> {
+ self.inner.cache_key()
+ }
+
+ fn file_format_metadata_cache(&self) -> Option<&(dyn std::any::Any + Send
+ Sync)> {
+ self.inner.file_format_metadata_cache()
+ }
}
/// Reads data files through their format-specific readers.
diff --git a/crates/paimon/src/table/rest_env.rs
b/crates/paimon/src/table/rest_env.rs
index af810f08..f518f0e6 100644
--- a/crates/paimon/src/table/rest_env.rs
+++ b/crates/paimon/src/table/rest_env.rs
@@ -23,7 +23,7 @@ use crate::catalog::{Identifier, RESTTokenFileIO};
use crate::common::{CatalogOptions, Options};
use crate::error::Error;
use crate::io::cache::{create_local_cache_with_namespace, LocalCache};
-use crate::io::FileIO;
+use crate::io::{FileFormatMetadataCacheContext, FileIO};
use crate::spec::{CoreOptions, TableSchema, PATH_OPTION};
use crate::table::snapshot_commit::{RESTSnapshotCommit, SnapshotCommit};
use crate::table::{ObjectTable, Table};
@@ -53,6 +53,8 @@ impl Table {
.map(|v| v.eq_ignore_ascii_case("true"))
.unwrap_or(false);
let local_cache = create_local_cache_with_namespace(&rest_options,
api.options())?;
+ let file_format_metadata_cache =
+
FileFormatMetadataCacheContext::from_props(api.options().to_map())?;
RESTEnv::build_table(
&identifier,
response,
@@ -60,6 +62,7 @@ impl Table {
rest_options,
data_token_enabled,
local_cache,
+ file_format_metadata_cache,
)
.await
}
@@ -75,6 +78,7 @@ pub struct RESTEnv {
options: Options,
data_token_enabled: bool,
local_cache: Option<Arc<LocalCache>>,
+ file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
}
impl std::fmt::Debug for RESTEnv {
@@ -95,6 +99,7 @@ impl RESTEnv {
options: Options,
data_token_enabled: bool,
local_cache: Option<Arc<LocalCache>>,
+ file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
) -> Self {
Self {
identifier,
@@ -103,6 +108,7 @@ impl RESTEnv {
options,
data_token_enabled,
local_cache,
+ file_format_metadata_cache,
}
}
@@ -134,6 +140,7 @@ impl RESTEnv {
self.options.clone(),
self.data_token_enabled,
self.local_cache.clone(),
+ self.file_format_metadata_cache.clone(),
)
.await
}
@@ -145,6 +152,7 @@ impl RESTEnv {
options: Options,
data_token_enabled: bool,
local_cache: Option<Arc<LocalCache>>,
+ file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
) -> Result<Table> {
let response = Self::fetch_table_response(identifier, &api).await?;
Self::build_table(
@@ -154,6 +162,7 @@ impl RESTEnv {
options,
data_token_enabled,
local_cache,
+ file_format_metadata_cache,
)
.await
}
@@ -177,6 +186,7 @@ impl RESTEnv {
options: Options,
data_token_enabled: bool,
local_cache: Option<Arc<LocalCache>>,
+ file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
) -> Result<Table> {
let identifier = response_identifier(identifier, &response)?;
let schema = response.schema.ok_or_else(|| Error::DataInvalid {
@@ -240,9 +250,9 @@ impl RESTEnv {
&table_path,
api.clone(),
&options,
- data_token_enabled,
- is_external,
+ data_token_enabled && !is_external,
local_cache.clone(),
+ file_format_metadata_cache.clone(),
)
.await?;
@@ -253,6 +263,7 @@ impl RESTEnv {
options,
data_token_enabled,
local_cache,
+ file_format_metadata_cache,
);
let parsed_identifier = identifier.parsed_object_name()?;
let branch = parsed_identifier.branch_or_default().to_string();
@@ -277,6 +288,7 @@ impl RESTEnv {
options: Options,
data_token_enabled: bool,
local_cache: Option<Arc<LocalCache>>,
+ file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
) -> Result<ObjectTable> {
let identifier = response_identifier(identifier, &response)?;
let schema = response.schema.ok_or_else(|| Error::DataInvalid {
@@ -318,9 +330,9 @@ impl RESTEnv {
&object_path,
api,
&options,
- data_token_enabled,
- is_external,
+ data_token_enabled && !is_external,
local_cache,
+ file_format_metadata_cache,
)
.await?;
@@ -332,23 +344,25 @@ impl RESTEnv {
path: &str,
api: Arc<RESTApi>,
options: &Options,
- data_token_enabled: bool,
- is_external: bool,
+ use_data_token: bool,
local_cache: Option<Arc<LocalCache>>,
+ file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
) -> Result<FileIO> {
- if data_token_enabled && !is_external {
+ if use_data_token {
return Arc::new(RESTTokenFileIO::new(
identifier.clone(),
path.to_string(),
options.clone(),
api,
local_cache,
+ file_format_metadata_cache,
))
.build_file_io()
.await;
}
let mut builder =
FileIO::from_path(path)?.with_props(options.to_map());
+ builder =
builder.with_file_format_metadata_cache(file_format_metadata_cache);
if let Some(local_cache) = local_cache {
builder = builder.with_local_cache(local_cache);
}
@@ -488,6 +502,7 @@ mod tests {
);
let local_cache = create_local_cache(&options).unwrap();
let api = Arc::new(RESTApi::new(options.clone(),
false).await.unwrap());
+ let metadata_cache =
FileFormatMetadataCacheContext::from_props(options.to_map()).unwrap();
let rest_env = RESTEnv::new(
Identifier::new("database", "table"),
@@ -496,9 +511,50 @@ mod tests {
options,
false,
local_cache,
+ metadata_cache,
);
assert!(rest_env.has_local_cache());
assert!(rest_env.clone().has_local_cache());
}
+
+ #[tokio::test]
+ async fn test_rest_table_loads_share_catalog_metadata_cache() {
+ let mut options = Options::new();
+ options.set(CatalogOptions::URI, "http://localhost:1";);
+ options.set(CatalogOptions::WAREHOUSE, "test-warehouse");
+ options.set(CatalogOptions::TOKEN_PROVIDER, "bear");
+ options.set(CatalogOptions::TOKEN, "test-token");
+ let api = Arc::new(RESTApi::new(options.clone(),
false).await.unwrap());
+ let metadata_cache =
FileFormatMetadataCacheContext::from_props(options.to_map()).unwrap();
+ let identifier = Identifier::new("database", "table");
+
+ let first = RESTEnv::build_file_io(
+ &identifier,
+ "memory:/table",
+ api.clone(),
+ &options,
+ false,
+ None,
+ metadata_cache.clone(),
+ )
+ .await
+ .unwrap();
+ let second = RESTEnv::build_file_io(
+ &identifier,
+ "memory:/table",
+ api,
+ &options,
+ false,
+ None,
+ metadata_cache,
+ )
+ .await
+ .unwrap();
+
+ assert!(Arc::ptr_eq(
+ &first.file_format_metadata_cache(),
+ &second.file_format_metadata_cache()
+ ));
+ }
}
diff --git a/crates/paimon/src/table/snapshot_manager.rs
b/crates/paimon/src/table/snapshot_manager.rs
index dbc76397..fc1b681e 100644
--- a/crates/paimon/src/table/snapshot_manager.rs
+++ b/crates/paimon/src/table/snapshot_manager.rs
@@ -586,7 +586,17 @@ mod tests {
let server = tokio::spawn(async move { axum::serve(listener,
app).await.unwrap() });
let api = Arc::new(RESTApi::new(options.clone(),
false).await.unwrap());
let id = Identifier::new("database", "table");
- let env = crate::table::RESTEnv::new(id.clone(), "uuid".into(), api,
options, false, None);
+ let metadata_cache =
+
crate::io::FileFormatMetadataCacheContext::from_props(options.to_map()).unwrap();
+ let env = crate::table::RESTEnv::new(
+ id.clone(),
+ "uuid".into(),
+ api,
+ options,
+ false,
+ None,
+ metadata_cache,
+ );
let (io, manager) = setup("/rest-table").await;
manager.commit_snapshot(&test_snapshot(2)).await.unwrap();
let schema = TableSchema::new(
diff --git a/crates/paimon/src/table/table_commit/parity_tests.rs
b/crates/paimon/src/table/table_commit/parity_tests.rs
index bdd9c4d2..3c2ecbac 100644
--- a/crates/paimon/src/table/table_commit/parity_tests.rs
+++ b/crates/paimon/src/table/table_commit/parity_tests.rs
@@ -614,8 +614,17 @@ async fn
rest_commit_uses_catalog_snapshot_schema_and_retry_identity() {
let server = tokio::spawn(async move { axum::serve(listener,
app).await.unwrap() });
let api = Arc::new(RESTApi::new(options.clone(), false).await.unwrap());
let identifier = Identifier::new("database", "table");
- let env =
- crate::table::RESTEnv::new(identifier.clone(), "uuid".into(), api,
options, false, None);
+ let metadata_cache =
+
crate::io::FileFormatMetadataCacheContext::from_props(options.to_map()).unwrap();
+ let env = crate::table::RESTEnv::new(
+ identifier.clone(),
+ "uuid".into(),
+ api,
+ options,
+ false,
+ None,
+ metadata_cache,
+ );
let table = Table::new(
io.clone(),
identifier,
diff --git a/crates/paimon/src/table/table_commit/recovery_tests.rs
b/crates/paimon/src/table/table_commit/recovery_tests.rs
index b202f100..ffaaae51 100644
--- a/crates/paimon/src/table/table_commit/recovery_tests.rs
+++ b/crates/paimon/src/table/table_commit/recovery_tests.rs
@@ -546,8 +546,17 @@ async fn
rest_delete_writer_pins_catalog_snapshot_and_preserves_vectors() {
let server = tokio::spawn(async move { axum::serve(listener,
app).await.unwrap() });
let api = Arc::new(RESTApi::new(options.clone(), false).await.unwrap());
let identifier = Identifier::new("database", "table");
- let env =
- crate::table::RESTEnv::new(identifier.clone(), "uuid".into(), api,
options, false, None);
+ let metadata_cache =
+
crate::io::FileFormatMetadataCacheContext::from_props(options.to_map()).unwrap();
+ let env = crate::table::RESTEnv::new(
+ identifier.clone(),
+ "uuid".into(),
+ api,
+ options,
+ false,
+ None,
+ metadata_cache,
+ );
let table = Table::new(io.clone(), identifier, path.into(), schema,
Some(env));
let mut commit = TableCommit::new(table.clone(), "rest-writer".into());
commit.commit_min_retry_wait_ms = 0;
