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 d2462936 [rust] Replace BLOB index entry limit with a memory budget
(#998)
d2462936 is described below
commit d246293694af2899732bb403e37d550582d28e01
Author: XiaoHongbo <[email protected]>
AuthorDate: Sat Oct 3 21:34:11 2026 +0800
[rust] Replace BLOB index entry limit with a memory budget (#998)
---
Cargo.lock | 1 +
bindings/c/Cargo.toml | 1 +
bindings/c/src/error.rs | 37 +
crates/paimon/src/arrow/format/blob.rs | 989 +++++++++++++++++++--
crates/paimon/src/arrow/format/metadata_cache.rs | 34 +-
crates/paimon/src/catalog/filesystem.rs | 20 +-
crates/paimon/src/catalog/rest/rest_catalog.rs | 15 +-
.../paimon/src/catalog/rest/rest_token_file_io.rs | 40 +-
crates/paimon/src/common/options.rs | 7 +
crates/paimon/src/error.rs | 30 +
crates/paimon/src/io/file_io.rs | 539 +++++++++--
crates/paimon/src/io/file_io/provider_tests.rs | 33 +
crates/paimon/src/io/storage_hdfs.rs | 42 +-
crates/paimon/src/table/data_file_reader.rs | 22 +-
crates/paimon/src/table/mod.rs | 5 +-
crates/paimon/src/table/rest_env.rs | 49 +-
crates/paimon/src/table/snapshot_manager.rs | 5 +-
.../paimon/src/table/table_commit/parity_tests.rs | 5 +-
.../src/table/table_commit/recovery_tests.rs | 5 +-
docs/src/getting-started.md | 6 +
20 files changed, 1676 insertions(+), 209 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index d9d01852..3347b5bb 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4739,6 +4739,7 @@ dependencies = [
"async-trait",
"bytes",
"futures",
+ "opendal-core",
"paimon",
"paimon-vindex-core",
"serde_json",
diff --git a/bindings/c/Cargo.toml b/bindings/c/Cargo.toml
index 916605b7..a62c2c36 100644
--- a/bindings/c/Cargo.toml
+++ b/bindings/c/Cargo.toml
@@ -48,6 +48,7 @@ async-trait = "0.1.81"
bytes = "1.7.1"
[dev-dependencies]
+opendal = { package = "opendal-core", version = "0.58.0" }
# Test-only: the vector-search integration tests build a real primary-key
vindex
# IVF-flat ANN segment fixture in-process. Versions match crates/paimon.
paimon-vindex-core = "0.5.0"
diff --git a/bindings/c/src/error.rs b/bindings/c/src/error.rs
index e6fdf426..26322967 100644
--- a/bindings/c/src/error.rs
+++ b/bindings/c/src/error.rs
@@ -65,6 +65,7 @@ impl paimon_error {
| paimon::Error::DataInvalid { .. }
| paimon::Error::IdentifierInvalid { .. } =>
PaimonErrorCode::InvalidInput,
paimon::Error::IoUnexpected { .. } => PaimonErrorCode::IoError,
+ _ if e.is_io_unexpected() => PaimonErrorCode::IoError,
paimon::Error::ResourceExhausted { .. } =>
PaimonErrorCode::ResourceExhausted,
_ => PaimonErrorCode::Unexpected,
};
@@ -123,3 +124,39 @@ pub fn check_non_null<T>(ptr: *const T, name: &str) ->
Result<(), *mut paimon_er
Ok(())
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn wrapped_blob_io_failure_keeps_c_io_error_code() {
+ let io_error = || paimon::Error::IoUnexpected {
+ message: "BLOB index read failed".to_string(),
+ source: Box::new(opendal::Error::new(
+ opendal::ErrorKind::Unexpected,
+ "storage failure",
+ )),
+ };
+ let wrapped = paimon::Error::UnexpectedError {
+ message: "shared BLOB index load failed".to_string(),
+ source: Some(Box::new(std::io::Error::other(io_error()))),
+ };
+ for error in [io_error(), wrapped] {
+ let error = paimon_error::from_paimon(error);
+ unsafe {
+ assert_eq!((*error).code, PaimonErrorCode::IoError as i32);
+ paimon_error_free(error);
+ }
+ }
+
+ let other = paimon_error::from_paimon(paimon::Error::UnexpectedError {
+ message: "not an I/O error".to_string(),
+ source: Some(Box::new(std::io::Error::other("unrelated"))),
+ });
+ unsafe {
+ assert_eq!((*other).code, PaimonErrorCode::Unexpected as i32);
+ paimon_error_free(other);
+ }
+ }
+}
diff --git a/crates/paimon/src/arrow/format/blob.rs
b/crates/paimon/src/arrow/format/blob.rs
index fff61b82..c1da6905 100644
--- a/crates/paimon/src/arrow/format/blob.rs
+++ b/crates/paimon/src/arrow/format/blob.rs
@@ -15,9 +15,10 @@
// specific language governing permissions and limitations
// under the License.
+use super::metadata_cache::FileMetadataCache;
use super::{FilePredicates, FormatFileReader, FormatFileWriter,
FormatWriteResult};
use crate::arrow::build_target_arrow_schema;
-use crate::io::{FileRead, FileWrite};
+use crate::io::{BlobIndexCacheContext, FileRead, FileWrite};
use crate::spec::{BlobDescriptor, DataField, DataType};
use crate::table::{ArrowRecordBatchStream, RowRange};
use crate::Error;
@@ -33,7 +34,6 @@ use async_stream::try_stream;
use async_trait::async_trait;
use bytes::Bytes;
use futures::{StreamExt, TryStreamExt};
-use lru::LruCache;
use std::ops::Range;
use std::sync::Arc;
@@ -93,7 +93,7 @@ impl IndexedBlobReader {
blob_parallelism: usize,
) -> crate::Result<Self> {
debug_assert!(blob_parallelism > 0);
- let index = BlobFileIndex::load_cached(reader.as_ref(), file_size,
&file_path).await?;
+ let index = BlobFileIndex::load_cached(reader.as_ref(),
file_size).await?;
Ok(Self {
reader,
index,
@@ -163,20 +163,9 @@ pub(crate) enum BlobReadValue {
const BLOB_FOOTER_SIZE: u64 = 5;
const BLOB_FORMAT_VERSION: u8 = 1;
-const BLOB_INDEX_CACHE_CAPACITY: usize = 16;
-#[derive(Clone, Debug, Eq, Hash, PartialEq)]
-struct BlobIndexCacheKey {
- namespace: usize,
- file_path: String,
-}
-
-static BLOB_INDEX_CACHE: std::sync::LazyLock<
- std::sync::Mutex<LruCache<BlobIndexCacheKey, Arc<BlobFileIndex>>>,
-> = std::sync::LazyLock::new(|| {
- std::sync::Mutex::new(LruCache::new(
- std::num::NonZeroUsize::new(BLOB_INDEX_CACHE_CAPACITY).unwrap(),
- ))
-});
+const BLOB_INDEX_CACHE_CONTAINER_OVERHEAD: usize =
std::mem::size_of::<String>()
+ + std::mem::size_of::<Arc<BlobFileIndex>>()
+ + 4 * std::mem::size_of::<usize>();
const BLOB_MAGIC_NUMBER: i32 = 1481511375;
const BLOB_MAGIC_NUMBER_BYTES: [u8; 4] = BLOB_MAGIC_NUMBER.to_le_bytes();
const BLOB_INLINE_HEADER_SIZE: u64 = 4;
@@ -1688,36 +1677,71 @@ async fn read_blob_range(reader: &dyn FileRead, range:
Range<u64>) -> crate::Res
Ok(bytes)
}
-impl BlobFileIndex {
- async fn load_cached(
- reader: &dyn FileRead,
- file_size: u64,
- file_path: &str,
- ) -> crate::Result<Arc<Self>> {
- let cache_key = reader
- .cache_namespace()
- .filter(|_| !file_path.is_empty())
- .map(|namespace| BlobIndexCacheKey {
- namespace,
- file_path: file_path.to_string(),
- });
- if let Some(cache_key) = &cache_key {
- let mut cache = BLOB_INDEX_CACHE
- .lock()
- .unwrap_or_else(|error| error.into_inner());
- if let Some(index) = cache.get(cache_key) {
- return Ok(index.clone());
- }
+type BlobIndexLoadResult = Result<Arc<BlobFileIndex>, Arc<Error>>;
+
+struct BlobIndexCache {
+ entries: FileMetadataCache<String, BlobIndexLoadResult>,
+}
+
+impl BlobIndexCache {
+ fn new(max_bytes: usize) -> Self {
+ Self {
+ entries: FileMetadataCache::new(max_bytes, usize::MAX),
}
+ }
+}
- let index = Arc::new(Self::load(reader, file_size).await?);
- if let Some(cache_key) = cache_key {
- BLOB_INDEX_CACHE
- .lock()
- .unwrap_or_else(|error| error.into_inner())
- .put(cache_key, index.clone());
+impl BlobFileIndex {
+ async fn load_cached(reader: &dyn FileRead, file_size: u64) ->
crate::Result<Arc<Self>> {
+ let Some(context) = reader
+ .blob_index_cache()
+ .and_then(|cache| cache.downcast_ref::<BlobIndexCacheContext>())
+ else {
+ return Ok(Arc::new(Self::load(reader, file_size).await?));
+ };
+ if !context.belongs_to_current_process() {
+ return Ok(Arc::new(Self::load(reader, file_size).await?));
+ }
+ let cache = context.get_or_init(BlobIndexCache::new);
+ let cache_key = reader.cache_key().map(ToOwned::to_owned);
+ let key_heap_bytes = cache_key.as_ref().map_or(0, String::capacity);
+ let load = cache
+ .entries
+ .get_or_try_insert_with_admission(
+ cache_key,
+ key_heap_bytes,
+ || async {
+ Ok::<_, std::convert::Infallible>(Arc::new(
+ Self::load(reader, file_size)
+ .await
+ .map(Arc::new)
+ .map_err(Arc::new),
+ ))
+ },
+ |load| {
+ load.as_ref()
+ .ok()
+ .map(|index| index.estimated_cache_bytes())
+ },
+ )
+ .await
+ .unwrap();
+ match load.as_ref() {
+ Ok(index) => Ok(Arc::clone(index)),
+ Err(error) => Err(clone_blob_index_error(error)),
}
- Ok(index)
+ }
+
+ fn estimated_cache_bytes(&self) -> usize {
+ std::mem::size_of::<Self>()
+ .saturating_add(
+ self.entries
+ .capacity()
+ .saturating_mul(std::mem::size_of::<BlobEntry>()),
+ )
+ // Account for the LRU node, hash slot and Arc allocation which are
+ // not represented by the decoded index itself.
+ .saturating_add(BLOB_INDEX_CACHE_CONTAINER_OVERHEAD)
}
async fn load(reader: &dyn FileRead, file_size: u64) ->
crate::Result<Self> {
@@ -1785,6 +1809,45 @@ impl BlobFileIndex {
}
}
+#[derive(Debug)]
+struct SharedBlobIndexError(Arc<Error>);
+
+impl std::fmt::Display for SharedBlobIndexError {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ std::fmt::Display::fmt(self.0.as_ref(), f)
+ }
+}
+
+impl std::error::Error for SharedBlobIndexError {
+ fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+ Some(self.0.as_ref())
+ }
+}
+
+fn clone_blob_index_error(error: &Arc<Error>) -> Error {
+ let source = || {
+ Some(Box::new(SharedBlobIndexError(Arc::clone(error)))
+ as Box<dyn std::error::Error + Send + Sync>)
+ };
+ match error.as_ref() {
+ Error::DataInvalid { message, .. } => Error::DataInvalid {
+ message: message.clone(),
+ source: source(),
+ },
+ Error::Unsupported { message } => Error::Unsupported {
+ message: message.clone(),
+ },
+ Error::UnexpectedError { message, .. } => Error::UnexpectedError {
+ message: message.clone(),
+ source: source(),
+ },
+ _ => Error::UnexpectedError {
+ message: error.to_string(),
+ source: source(),
+ },
+ }
+}
+
#[derive(Debug, Clone)]
enum BlobEntry {
Value(Range<u64>),
@@ -2333,11 +2396,35 @@ fn encode_varint(value: i64, out: &mut Vec<u8>) {
mod tests {
use super::*;
use crate::btree::test_util::BytesFileRead;
- use crate::io::{FileIO, FileIOBuilder};
+ use crate::common::CatalogOptions;
+ #[cfg(feature = "storage-oss")]
+ use crate::io::FileIOCacheContext;
+ #[cfg(feature = "storage-oss")]
+ use crate::io::FileIOProvider;
+ use crate::io::{BlobIndexCacheContext, FileIO, FileIOBuilder};
use crate::spec::{ArrayType, BlobType, MapType, VarCharType};
use arrow_array::Array;
+ #[cfg(feature = "storage-oss")]
+ use axum::{
+ body::Body,
+ extract::State,
+ http::{
+ header::{CONTENT_LENGTH, CONTENT_RANGE, RANGE},
+ HeaderMap, Response, StatusCode,
+ },
+ routing::get,
+ Router,
+ };
use bytes::Bytes;
use futures::TryStreamExt;
+ #[cfg(feature = "storage-oss")]
+ use opendal::{Configurator, HttpTransporter, OperationContext, Operator};
+ #[cfg(feature = "storage-oss")]
+ use opendal_http_transport_reqwest::ReqwestTransport;
+ #[cfg(feature = "storage-oss")]
+ use opendal_service_oss::OssConfig;
+ #[cfg(feature = "storage-oss")]
+ use std::collections::HashMap;
use std::mem::size_of;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Mutex;
@@ -2387,6 +2474,120 @@ mod tests {
assert!(matches!(error, Error::ProcessForkUnsupported { .. }));
}
+ #[cfg(target_os = "linux")]
+ struct ForkBoundBlobRead {
+ parent_pid: u32,
+ cache: Arc<BlobIndexCacheContext>,
+ started: Arc<tokio::sync::Notify>,
+ }
+
+ #[cfg(target_os = "linux")]
+ #[async_trait::async_trait]
+ impl FileRead for ForkBoundBlobRead {
+ async fn read(&self, _range: Range<u64>) -> crate::Result<Bytes> {
+ if std::process::id() != self.parent_pid {
+ return Err(Error::ProcessForkUnsupported {
+ message: crate::error::JINDO_FORK_ERROR.to_string(),
+ });
+ }
+ self.started.notify_one();
+ std::future::pending().await
+ }
+
+ fn cache_key(&self) -> Option<&str> {
+ Some("fork-pending-blob")
+ }
+
+ fn blob_index_cache(&self) -> Option<&(dyn std::any::Any + Send +
Sync)> {
+ Some(self.cache.as_ref())
+ }
+ }
+
+ #[cfg(target_os = "linux")]
+ #[tokio::test(flavor = "current_thread")]
+ async fn blob_index_cache_does_not_join_parent_load_after_fork() {
+ let cache = blob_index_cache("64 MiB");
+ let started = Arc::new(tokio::sync::Notify::new());
+ let parent_pid = std::process::id();
+ let parent_reader = ForkBoundBlobRead {
+ parent_pid,
+ cache: Arc::clone(&cache),
+ started: Arc::clone(&started),
+ };
+ let parent_load =
+ tokio::spawn(async move {
BlobFileIndex::load_cached(&parent_reader, 5).await });
+ tokio::time::timeout(Duration::from_secs(2), started.notified())
+ .await
+ .expect("parent load did not start");
+
+ let child_pid = unsafe { libc::fork() };
+ assert!(child_pid >= 0, "fork failed");
+ if child_pid == 0 {
+ let passed = std::thread::spawn(move || {
+ let runtime = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ runtime.block_on(async {
+ let reader = ForkBoundBlobRead {
+ parent_pid,
+ cache,
+ started,
+ };
+ matches!(
+ tokio::time::timeout(
+ Duration::from_millis(500),
+ BlobFileIndex::load_cached(&reader, 5),
+ )
+ .await,
+ Ok(Err(Error::ProcessForkUnsupported { .. }))
+ )
+ })
+ })
+ .join()
+ .unwrap_or(false);
+ unsafe { libc::_exit(i32::from(!passed)) };
+ }
+
+ let (status, timed_out) = tokio::task::spawn_blocking(move || {
+ let deadline = std::time::Instant::now() + Duration::from_secs(3);
+ loop {
+ let mut status = 0;
+ let waited = unsafe { libc::waitpid(child_pid, &mut status,
libc::WNOHANG) };
+ if waited == child_pid {
+ return Ok::<_, std::io::Error>((status, false));
+ }
+ if waited < 0 {
+ let error = std::io::Error::last_os_error();
+ if error.kind() != std::io::ErrorKind::Interrupted {
+ return Err(error);
+ }
+ }
+ if std::time::Instant::now() >= deadline {
+ unsafe { libc::kill(child_pid, libc::SIGKILL) };
+ loop {
+ let waited = unsafe { libc::waitpid(child_pid, &mut
status, 0) };
+ if waited == child_pid {
+ return Ok((status, true));
+ }
+ let error = std::io::Error::last_os_error();
+ if error.kind() != std::io::ErrorKind::Interrupted {
+ return Err(error);
+ }
+ }
+ }
+ std::thread::sleep(Duration::from_millis(10));
+ }
+ })
+ .await
+ .unwrap()
+ .unwrap();
+ parent_load.abort();
+ assert!(!timed_out, "child hung on inherited BLOB cache state");
+ assert!(libc::WIFEXITED(status));
+ assert_eq!(libc::WEXITSTATUS(status), 0);
+ }
+
#[tokio::test]
async fn test_index_tail_prefetch_boundaries() {
for rows in [0, 1, 4090, 4091, 4092, 9000] {
@@ -2630,10 +2831,11 @@ mod tests {
async fn test_blob_reader_reuses_cached_index() {
let file_path = "file:///blob-index-cache-test/data.blob";
let file_bytes = load_blob_fixture("blob-basic.blob");
- let first =
-
TrackingFileRead::new(Bytes::from(file_bytes.clone())).with_cache_namespace(usize::MAX);
- let second =
-
TrackingFileRead::new(Bytes::from(file_bytes.clone())).with_cache_namespace(usize::MAX);
+ let cache = blob_index_cache("64 MiB");
+ let first = TrackingFileRead::new(Bytes::from(file_bytes.clone()))
+ .with_blob_index_cache(file_path, cache.clone());
+ let second = TrackingFileRead::new(Bytes::from(file_bytes.clone()))
+ .with_blob_index_cache(file_path, cache);
let first_reader = IndexedBlobReader::open(
Box::new(first.clone()),
@@ -2657,6 +2859,234 @@ mod tests {
assert!(second.ranges().is_empty());
}
+ #[tokio::test]
+ async fn test_blob_index_cache_can_be_disabled() {
+ let file_path = "file:///blob-index-cache-disabled/data.blob";
+ let file_bytes = load_blob_fixture("blob-basic.blob");
+ let cache = blob_index_cache("0");
+ let first = TrackingFileRead::new(Bytes::from(file_bytes.clone()))
+ .with_blob_index_cache(file_path, cache.clone());
+ let second = TrackingFileRead::new(Bytes::from(file_bytes.clone()))
+ .with_blob_index_cache(file_path, cache);
+
+ for reader in [first.clone(), second.clone()] {
+ IndexedBlobReader::open(
+ Box::new(reader),
+ file_bytes.len() as u64,
+ file_path.to_string(),
+ true,
+ )
+ .await
+ .unwrap();
+ }
+
+ assert_eq!(first.ranges().len(), 1);
+ assert_eq!(second.ranges().len(), 1);
+ }
+
+ #[tokio::test]
+ async fn test_blob_index_cache_coalesces_concurrent_loads() {
+ let file_path = "file:///blob-index-cache-concurrent/data.blob";
+ let file_bytes = load_blob_fixture("blob-basic.blob");
+ let cache = blob_index_cache("64 MiB");
+ let tracking = TrackingFileRead::new(Bytes::from(file_bytes.clone()))
+ .with_blob_index_cache(file_path, cache);
+
+ let open = |reader: TrackingFileRead| {
+ IndexedBlobReader::open(
+ Box::new(reader),
+ file_bytes.len() as u64,
+ file_path.to_string(),
+ true,
+ )
+ };
+ let (first, second) = tokio::join!(open(tracking.clone()),
open(tracking.clone()));
+
+ assert_eq!(first.unwrap().num_rows(), second.unwrap().num_rows());
+ assert_eq!(tracking.ranges().len(), 1);
+ }
+
+ #[tokio::test]
+ async fn test_blob_index_cache_coalesces_failure_then_retries() {
+ let file_path = "file:///blob-index-cache-failure/data.blob";
+ let file_bytes = load_blob_fixture("blob-basic.blob");
+ let cache = blob_index_cache("64 MiB");
+ let failing = TrackingFileRead::new(Bytes::from(file_bytes.clone()))
+ .with_blob_index_cache(file_path, cache.clone())
+ .with_failure();
+
+ let open = |reader: TrackingFileRead| {
+ IndexedBlobReader::open(
+ Box::new(reader),
+ file_bytes.len() as u64,
+ file_path.to_string(),
+ true,
+ )
+ };
+ let failures = futures::future::join_all((0..8).map(|_|
open(failing.clone()))).await;
+
+ assert!(failures.iter().all(Result::is_err));
+ for failure in failures {
+ let Error::UnexpectedError {
+ source: Some(source),
+ ..
+ } = failure.err().unwrap()
+ else {
+ panic!("cached failure lost its source");
+ };
+ let shared =
source.downcast_ref::<SharedBlobIndexError>().unwrap();
+ assert!(std::error::Error::source(shared)
+ .unwrap()
+ .downcast_ref::<Error>()
+ .is_some());
+ assert!(matches!(
+ shared.0.as_ref(),
+ Error::UnexpectedError {
+ source: Some(_),
+ ..
+ }
+ ));
+ }
+ assert_eq!(failing.ranges().len(), 1);
+
+ let retry = TrackingFileRead::new(Bytes::from(file_bytes.clone()))
+ .with_blob_index_cache(file_path, cache);
+ assert_eq!(open(retry.clone()).await.unwrap().num_rows(), 4);
+ assert_eq!(retry.ranges().len(), 1);
+ }
+
+ #[test]
+ fn test_blob_index_cache_preserves_direct_error_cause() {
+ let original = Arc::new(Error::IoUnexpected {
+ message: "injected storage failure".to_string(),
+ source: Box::new(opendal::Error::new(
+ opendal::ErrorKind::Unsupported,
+ "injected native failure",
+ )),
+ });
+ let Error::UnexpectedError {
+ source: Some(source),
+ ..
+ } = clone_blob_index_error(&original)
+ else {
+ panic!("cached failure lost its cause");
+ };
+ let cause = std::error::Error::source(source.as_ref())
+ .unwrap()
+ .downcast_ref::<Error>()
+ .unwrap();
+ assert!(std::ptr::eq(cause, original.as_ref()));
+ }
+
+ #[tokio::test]
+ async fn test_blob_index_cache_preserves_io_error_category() {
+ let file_size = load_blob_fixture("blob-basic.blob").len() as u64;
+ for budget in ["64 MiB", "0"] {
+ let path =
format!("file:///blob-index-io-error/{budget}/data.blob");
+ let reader = TrackingFileRead::new(Bytes::new())
+ .with_blob_index_cache(&path, blob_index_cache(budget))
+ .with_io_failure();
+
+ let direct = BlobFileIndex::load(&reader,
file_size).await.unwrap_err();
+ assert!(matches!(direct, Error::IoUnexpected { .. }));
+ for _ in 0..2 {
+ let cached = IndexedBlobReader::open(
+ Box::new(reader.clone()),
+ file_size,
+ path.clone(),
+ true,
+ )
+ .await
+ .err()
+ .expect("BLOB index load should fail");
+ assert!(cached.is_io_unexpected(), "budget={budget}:
{cached:?}");
+ }
+ assert_eq!(reader.ranges().len(), 3);
+ }
+ }
+
+ #[tokio::test]
+ async fn test_blob_index_cache_evicts_by_decoded_bytes() {
+ let first_key = "memory:/blob-index-cache-bytes/first.blob";
+ let second_key = "memory:/blob-index-cache-bytes/other.blob";
+ assert_eq!(first_key.len(), second_key.len());
+ let file_bytes = blob_test_utils::build_blob_file_bytes(&[None, None]);
+ let max_bytes = blob_index_cache_entry_weight(first_key, 2);
+ let cache = blob_index_cache(&max_bytes.to_string());
+ let first = TrackingFileRead::new(Bytes::from(file_bytes.clone()))
+ .with_blob_index_cache(first_key, cache.clone());
+ let second = TrackingFileRead::new(Bytes::from(file_bytes.clone()))
+ .with_blob_index_cache(second_key, cache.clone());
+ let first_again =
TrackingFileRead::new(Bytes::from(file_bytes.clone()))
+ .with_blob_index_cache(first_key, cache);
+
+ for (reader, key) in [
+ (first.clone(), first_key),
+ (second.clone(), second_key),
+ (first_again.clone(), first_key),
+ ] {
+ IndexedBlobReader::open(
+ Box::new(reader),
+ file_bytes.len() as u64,
+ key.to_string(),
+ true,
+ )
+ .await
+ .unwrap();
+ }
+
+ assert_eq!(first.ranges().len(), 1);
+ assert_eq!(second.ranges().len(), 1);
+ assert_eq!(first_again.ranges().len(), 1);
+ }
+
+ #[tokio::test]
+ async fn test_oversized_blob_index_does_not_evict_cached_index() {
+ let small_key = "memory:/blob-index-cache-size/small.blob";
+ let large_key = "memory:/blob-index-cache-size/large.blob";
+ assert_eq!(small_key.len(), large_key.len());
+ let small_bytes = blob_test_utils::build_blob_file_bytes(&[None]);
+ let large_values = vec![None; 100];
+ let large_bytes =
blob_test_utils::build_blob_file_bytes(&large_values);
+ let max_bytes = blob_index_cache_entry_weight(small_key, 1);
+ let cache = blob_index_cache(&max_bytes.to_string());
+ let small = TrackingFileRead::new(Bytes::from(small_bytes.clone()))
+ .with_blob_index_cache(small_key, cache.clone());
+ let large = TrackingFileRead::new(Bytes::from(large_bytes.clone()))
+ .with_blob_index_cache(large_key, cache.clone());
+ let small_again =
TrackingFileRead::new(Bytes::from(small_bytes.clone()))
+ .with_blob_index_cache(small_key, cache);
+
+ IndexedBlobReader::open(
+ Box::new(small.clone()),
+ small_bytes.len() as u64,
+ small_key.to_string(),
+ true,
+ )
+ .await
+ .unwrap();
+ IndexedBlobReader::open(
+ Box::new(large.clone()),
+ large_bytes.len() as u64,
+ large_key.to_string(),
+ true,
+ )
+ .await
+ .unwrap();
+ IndexedBlobReader::open(
+ Box::new(small_again.clone()),
+ small_bytes.len() as u64,
+ small_key.to_string(),
+ true,
+ )
+ .await
+ .unwrap();
+
+ assert_eq!(small.ranges().len(), 1);
+ assert_eq!(large.ranges().len(), 1);
+ assert!(small_again.ranges().is_empty());
+ }
+
#[tokio::test]
async fn test_blob_index_cache_isolated_by_file_io() {
let path = "memory:///blob-index-cache-namespace/data.blob";
@@ -2680,32 +3110,338 @@ mod tests {
.await
.unwrap();
- let first_namespace = first_io
+ assert!(!Arc::ptr_eq(
+ &first_io.blob_index_cache(),
+ &second_io.blob_index_cache()
+ ));
+
+ assert_eq!(
+ read_scalar_blob_file(&first_io, path).await,
+ vec![None, Some(value.to_vec())]
+ );
+ assert_eq!(
+ read_scalar_blob_file(&second_io, path).await,
+ vec![Some(value.to_vec()), None]
+ );
+ }
+
+ #[tokio::test]
+ #[cfg(feature = "storage-oss")]
+ async fn test_blob_index_cache_isolated_by_storage_endpoint() {
+ let value = b"value";
+ let first_bytes = Bytes::from(blob_test_utils::build_blob_file_bytes(&[
+ None,
+ Some(value.as_slice()),
+ ]));
+ let second_bytes =
Bytes::from(blob_test_utils::build_blob_file_bytes(&[
+ Some(value.as_slice()),
+ None,
+ ]));
+ assert_eq!(first_bytes.len(), second_bytes.len());
+ let first_endpoint = serve_blob_file(first_bytes.clone()).await;
+ let second_endpoint = serve_blob_file(second_bytes.clone()).await;
+ let cache_context =
FileIOCacheContext::from_props(&HashMap::new()).unwrap();
+
+ let first_io = blob_oss_file_io(&first_endpoint,
cache_context.clone());
+ let second_io = blob_oss_file_io(&second_endpoint, cache_context);
+ let path = "/data.blob";
+
+ let first = IndexedBlobReader::open(
+
Box::new(first_io.new_input(path).unwrap().reader().await.unwrap()),
+ first_bytes.len() as u64,
+ path.to_string(),
+ false,
+ )
+ .await
+ .unwrap();
+ let second = IndexedBlobReader::open(
+
Box::new(second_io.new_input(path).unwrap().reader().await.unwrap()),
+ second_bytes.len() as u64,
+ path.to_string(),
+ false,
+ )
+ .await
+ .unwrap();
+
+ assert_eq!(second.num_rows(), 2);
+ assert!(matches!(
+ first.read_positions(&[0]).await.unwrap().as_slice(),
+ [BlobReadValue::Null]
+ ));
+ assert!(matches!(
+ second.read_positions(&[0]).await.unwrap().as_slice(),
+ [BlobReadValue::Value(bytes)] if bytes.as_ref() == value
+ ));
+ }
+
+ #[tokio::test]
+ #[cfg(feature = "storage-azdls")]
+ async fn test_blob_index_cache_isolated_by_azure_account() {
+ let first_path =
"abfss://[email protected]/data.blob";
+ let second_path =
"abfss://[email protected]/data.blob";
+ let file_io = FileIO::from_path(first_path)
+ .unwrap()
+ .with_prop("azure.account-key", "account-key")
+ .build()
+ .unwrap();
+ let first_key = file_io
+ .new_input(first_path)
+ .unwrap()
+ .reader()
+ .await
+ .unwrap()
+ .cache_key()
+ .unwrap()
+ .to_string();
+ let second_key = file_io
+ .new_input(second_path)
+ .unwrap()
+ .reader()
+ .await
+ .unwrap()
+ .cache_key()
+ .unwrap()
+ .to_string();
+ assert_ne!(first_key, second_key);
+
+ let value = b"value";
+ let first_bytes = blob_test_utils::build_blob_file_bytes(&[None,
Some(value)]);
+ let second_bytes =
blob_test_utils::build_blob_file_bytes(&[Some(value), None]);
+ let cache = blob_index_cache("64 MiB");
+ let first = TrackingFileRead::new(Bytes::from(first_bytes.clone()))
+ .with_blob_index_cache(first_key, cache.clone());
+ let second = TrackingFileRead::new(Bytes::from(second_bytes.clone()))
+ .with_blob_index_cache(second_key, cache);
+ let first = IndexedBlobReader::open(
+ Box::new(first),
+ first_bytes.len() as u64,
+ first_path.to_string(),
+ false,
+ )
+ .await
+ .unwrap();
+ let second = IndexedBlobReader::open(
+ Box::new(second),
+ second_bytes.len() as u64,
+ second_path.to_string(),
+ false,
+ )
+ .await
+ .unwrap();
+ assert!(matches!(
+ first.read_positions(&[0]).await.unwrap().as_slice(),
+ [BlobReadValue::Null]
+ ));
+ assert!(matches!(
+ second.read_positions(&[0]).await.unwrap().as_slice(),
+ [BlobReadValue::Value(bytes)] if bytes.as_ref() == value
+ ));
+ }
+
+ #[tokio::test]
+ #[cfg(feature = "storage-hdfs")]
+ async fn test_blob_index_cache_isolated_by_hdfs_name_node() {
+ let path = "hdfs://logical-cluster/table/data.blob";
+ let first_io = FileIO::from_path(path)
+ .unwrap()
+ .with_prop("hdfs.name-node", "hdfs://cluster-a:8020")
+ .build()
+ .unwrap();
+ let second_io = FileIO::from_path(path)
+ .unwrap()
+ .with_prop("hdfs.name-node", "hdfs://cluster-b:8020")
+ .with_cache_context(first_io.cache_context())
+ .build()
+ .unwrap();
+ let first_key = first_io
.new_input(path)
.unwrap()
.reader()
.await
.unwrap()
- .cache_namespace();
- let second_namespace = second_io
+ .cache_key()
+ .unwrap()
+ .to_string();
+ let second_key = second_io
.new_input(path)
.unwrap()
.reader()
.await
.unwrap()
- .cache_namespace();
- assert_ne!(first_namespace, second_namespace);
+ .cache_key()
+ .unwrap()
+ .to_string();
+ assert_ne!(first_key, second_key);
- assert_eq!(
- read_scalar_blob_file(&first_io, path).await,
- vec![None, Some(value.to_vec())]
+ let value = b"value";
+ let first_bytes = blob_test_utils::build_blob_file_bytes(&[None,
Some(value)]);
+ let second_bytes =
blob_test_utils::build_blob_file_bytes(&[Some(value), None]);
+ assert_eq!(first_bytes.len(), second_bytes.len());
+ let cache = first_io.blob_index_cache();
+ let first = TrackingFileRead::new(Bytes::from(first_bytes.clone()))
+ .with_blob_index_cache(first_key, cache.clone());
+ let second = TrackingFileRead::new(Bytes::from(second_bytes.clone()))
+ .with_blob_index_cache(second_key, cache);
+ let first = IndexedBlobReader::open(
+ Box::new(first),
+ first_bytes.len() as u64,
+ path.to_string(),
+ false,
+ )
+ .await
+ .unwrap();
+ let second = IndexedBlobReader::open(
+ Box::new(second),
+ second_bytes.len() as u64,
+ path.to_string(),
+ false,
+ )
+ .await
+ .unwrap();
+ assert!(matches!(
+ first.read_positions(&[0]).await.unwrap().as_slice(),
+ [BlobReadValue::Null]
+ ));
+ assert!(
+ matches!(second.read_positions(&[0]).await.unwrap().as_slice(),
[BlobReadValue::Value(bytes)] if bytes.as_ref() == value)
);
- assert_eq!(
- read_scalar_blob_file(&second_io, path).await,
- vec![Some(value.to_vec()), None]
+ }
+
+ #[derive(Debug)]
+ #[cfg(feature = "storage-oss")]
+ struct FixedBlobProvider(Operator);
+
+ #[async_trait]
+ #[cfg(feature = "storage-oss")]
+ impl FileIOProvider for FixedBlobProvider {
+ async fn create(&self, _path: &str) -> crate::Result<(Operator,
String)> {
+ Ok((self.0.clone(), "data.blob".to_string()))
+ }
+ }
+
+ #[derive(Debug)]
+ #[cfg(feature = "storage-oss")]
+ struct RoutedBlobProvider {
+ first: Operator,
+ second: Operator,
+ }
+
+ #[async_trait]
+ #[cfg(feature = "storage-oss")]
+ impl FileIOProvider for RoutedBlobProvider {
+ async fn create(&self, path: &str) -> crate::Result<(Operator,
String)> {
+ let op = if path.starts_with("oss://first/") {
+ &self.first
+ } else {
+ &self.second
+ };
+ Ok((op.clone(), "data.blob".to_string()))
+ }
+ }
+
+ #[tokio::test]
+ #[cfg(feature = "storage-oss")]
+ async fn test_blob_index_cache_bypassed_for_opaque_provider_routes() {
+ let value = b"value";
+ let first_bytes =
Bytes::from(blob_test_utils::build_blob_file_bytes(&[None, Some(value)]));
+ let second_bytes =
+ Bytes::from(blob_test_utils::build_blob_file_bytes(&[Some(value),
None]));
+ assert_eq!(first_bytes.len(), second_bytes.len());
+ let first_endpoint = serve_blob_file(first_bytes.clone()).await;
+ let second_endpoint = serve_blob_file(second_bytes.clone()).await;
+ let io = FileIOBuilder::new("fs")
+ .with_provider(Arc::new(RoutedBlobProvider {
+ first: blob_oss_operator(&first_endpoint),
+ second: blob_oss_operator(&second_endpoint),
+ }))
+ .build()
+ .unwrap();
+ let first_path = "oss://first/data.blob";
+ let second_path = "oss://second/data.blob";
+ let first_reader =
io.new_input(first_path).unwrap().reader().await.unwrap();
+ let second_reader =
io.new_input(second_path).unwrap().reader().await.unwrap();
+ assert_eq!(first_reader.cache_key(), None);
+ assert_eq!(second_reader.cache_key(), None);
+ let first = IndexedBlobReader::open(
+ Box::new(first_reader),
+ first_bytes.len() as u64,
+ first_path.to_string(),
+ false,
+ )
+ .await
+ .unwrap();
+ let second = IndexedBlobReader::open(
+ Box::new(second_reader),
+ second_bytes.len() as u64,
+ second_path.to_string(),
+ false,
+ )
+ .await
+ .unwrap();
+ assert!(matches!(
+ first.read_positions(&[0]).await.unwrap().as_slice(),
+ [BlobReadValue::Null]
+ ));
+ assert!(
+ matches!(second.read_positions(&[0]).await.unwrap().as_slice(),
[BlobReadValue::Value(bytes)] if bytes.as_ref() == value)
);
}
+ #[tokio::test]
+ #[cfg(feature = "storage-oss")]
+ async fn test_blob_index_cache_isolated_after_provider_replacement() {
+ let value = b"value";
+ let first_bytes = Bytes::from(blob_test_utils::build_blob_file_bytes(&[
+ None,
+ Some(value.as_slice()),
+ ]));
+ let second_bytes =
Bytes::from(blob_test_utils::build_blob_file_bytes(&[
+ Some(value.as_slice()),
+ None,
+ ]));
+ let first_endpoint = serve_blob_file(first_bytes.clone()).await;
+ let second_endpoint = serve_blob_file(second_bytes.clone()).await;
+ let cache_context =
FileIOCacheContext::from_props(&HashMap::new()).unwrap();
+ let original = FileIOBuilder::new("fs")
+ .with_cache_context(cache_context)
+ .build()
+ .unwrap()
+ .with_provider(Arc::new(FixedBlobProvider(blob_oss_operator(
+ &first_endpoint,
+ ))));
+ let replacement =
+ original
+ .clone()
+ .with_provider(Arc::new(FixedBlobProvider(blob_oss_operator(
+ &second_endpoint,
+ ))));
+ let path = "oss://bucket/data.blob";
+ let first = IndexedBlobReader::open(
+
Box::new(original.new_input(path).unwrap().reader().await.unwrap()),
+ first_bytes.len() as u64,
+ path.to_string(),
+ false,
+ )
+ .await
+ .unwrap();
+ let second = IndexedBlobReader::open(
+
Box::new(replacement.new_input(path).unwrap().reader().await.unwrap()),
+ second_bytes.len() as u64,
+ path.to_string(),
+ false,
+ )
+ .await
+ .unwrap();
+ assert!(matches!(
+ first.read_positions(&[0]).await.unwrap().as_slice(),
+ [BlobReadValue::Null]
+ ));
+ assert!(matches!(
+ second.read_positions(&[0]).await.unwrap().as_slice(),
+ [BlobReadValue::Value(bytes)] if bytes.as_ref() == value
+ ));
+ }
+
#[tokio::test]
async fn test_blob_array_reader_reads_java_fixture() {
let read_fields = vec![DataField::new(
@@ -4072,10 +4808,92 @@ mod tests {
std::fs::read(&path).unwrap_or_else(|e| panic!("Failed to read {path}:
{e}"))
}
+ fn blob_index_cache(max_size: &str) -> Arc<BlobIndexCacheContext> {
+ BlobIndexCacheContext::from_props(&std::collections::HashMap::from([(
+ CatalogOptions::BLOB_INDEX_CACHE_MAX_SIZE.to_string(),
+ max_size.to_string(),
+ )]))
+ .unwrap()
+ }
+
+ fn blob_index_cache_entry_weight(key: &str, rows: usize) -> usize {
+ let index = BlobFileIndex {
+ entries: vec![BlobEntry::Null; rows],
+ };
+ FileMetadataCache::<String, BlobFileIndex>::entry_weight(
+ key.len(),
+ index.estimated_cache_bytes(),
+ )
+ }
+
+ #[cfg(feature = "storage-oss")]
+ async fn serve_blob_file(bytes: Bytes) -> String {
+ async fn get_blob(State(bytes): State<Bytes>, headers: HeaderMap) ->
Response<Body> {
+ let range = headers
+ .get(RANGE)
+ .and_then(|value| value.to_str().ok())
+ .and_then(|value| value.strip_prefix("bytes="))
+ .and_then(|value| value.split_once('-'))
+ .and_then(|(start, end)| {
+ Some((start.parse::<usize>().ok()?,
end.parse::<usize>().ok()?))
+ });
+ let Some((start, end)) = range else {
+ return Response::builder()
+ .status(StatusCode::OK)
+ .header(CONTENT_LENGTH, bytes.len())
+ .body(Body::from(bytes))
+ .unwrap();
+ };
+ let body = bytes.slice(start..=end);
+ Response::builder()
+ .status(StatusCode::PARTIAL_CONTENT)
+ .header(CONTENT_LENGTH, body.len())
+ .header(
+ CONTENT_RANGE,
+ format!("bytes {start}-{end}/{}", bytes.len()),
+ )
+ .body(Body::from(body))
+ .unwrap()
+ }
+
+ let app = Router::new().fallback(get(get_blob)).with_state(bytes);
+ let listener =
tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let address = listener.local_addr().unwrap();
+ tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+ format!("http://{address}";)
+ }
+
+ #[cfg(feature = "storage-oss")]
+ fn blob_oss_file_io(endpoint: &str, cache_context: FileIOCacheContext) ->
FileIO {
+ FileIOBuilder::new("fs")
+ .with_prop("fs.oss.endpoint", endpoint)
+ .with_fs_operator(blob_oss_operator(endpoint))
+ .with_cache_context(cache_context)
+ .build()
+ .unwrap()
+ }
+
+ #[cfg(feature = "storage-oss")]
+ fn blob_oss_operator(endpoint: &str) -> Operator {
+ let mut config = OssConfig::default();
+ config.endpoint = Some(endpoint.to_string());
+ config.addressing_style = Some("path".to_string());
+ config.skip_signature = true;
+ Operator::new(config.into_builder().bucket("bucket"))
+ .unwrap()
+ .with_context(
+ OperationContext::new()
+
.with_http_transport(HttpTransporter::new(ReqwestTransport::default())),
+ )
+ }
+
#[derive(Clone)]
struct TrackingFileRead {
bytes: Bytes,
- cache_namespace: Option<usize>,
+ cache_key: Option<String>,
+ blob_index_cache: Option<Arc<BlobIndexCacheContext>>,
+ fail: bool,
+ io_fail: bool,
in_flight: Arc<AtomicUsize>,
max_in_flight: Arc<AtomicUsize>,
ranges: Arc<Mutex<Vec<Range<u64>>>>,
@@ -4085,15 +4903,33 @@ mod tests {
fn new(bytes: Bytes) -> Self {
Self {
bytes,
- cache_namespace: None,
+ cache_key: None,
+ blob_index_cache: None,
+ fail: false,
+ io_fail: false,
in_flight: Arc::new(AtomicUsize::new(0)),
max_in_flight: Arc::new(AtomicUsize::new(0)),
ranges: Arc::new(Mutex::new(Vec::new())),
}
}
- fn with_cache_namespace(mut self, cache_namespace: usize) -> Self {
- self.cache_namespace = Some(cache_namespace);
+ fn with_blob_index_cache(
+ mut self,
+ cache_key: impl Into<String>,
+ cache: Arc<BlobIndexCacheContext>,
+ ) -> Self {
+ self.cache_key = Some(cache_key.into());
+ self.blob_index_cache = Some(cache);
+ self
+ }
+
+ fn with_failure(mut self) -> Self {
+ self.fail = true;
+ self
+ }
+
+ fn with_io_failure(mut self) -> Self {
+ self.io_fail = true;
self
}
@@ -4114,11 +4950,32 @@ mod tests {
self.max_in_flight.fetch_max(in_flight, Ordering::SeqCst);
tokio::time::sleep(Duration::from_millis(10)).await;
self.in_flight.fetch_sub(1, Ordering::SeqCst);
+ if self.io_fail {
+ return Err(Error::IoUnexpected {
+ message: "injected BLOB index I/O failure".to_string(),
+ source: Box::new(opendal::Error::new(
+ opendal::ErrorKind::Unexpected,
+ "injected storage failure",
+ )),
+ });
+ }
+ if self.fail {
+ return Err(Error::UnexpectedError {
+ message: "injected BLOB index read failure".to_string(),
+ source: Some(Box::new(std::io::Error::other("injected
source"))),
+ });
+ }
Ok(self.bytes.slice(range.start as usize..range.end as usize))
}
- fn cache_namespace(&self) -> Option<usize> {
- self.cache_namespace
+ fn cache_key(&self) -> Option<&str> {
+ self.cache_key.as_deref()
+ }
+
+ fn blob_index_cache(&self) -> Option<&(dyn std::any::Any + Send +
Sync)> {
+ self.blob_index_cache
+ .as_deref()
+ .map(|cache| cache as &(dyn std::any::Any + Send + Sync))
}
}
diff --git a/crates/paimon/src/arrow/format/metadata_cache.rs
b/crates/paimon/src/arrow/format/metadata_cache.rs
index 5bb140ad..94a0c45c 100644
--- a/crates/paimon/src/arrow/format/metadata_cache.rs
+++ b/crates/paimon/src/arrow/format/metadata_cache.rs
@@ -100,6 +100,24 @@ where
F: FnOnce() -> Fut,
Fut: Future<Output = Result<Arc<V>, E>>,
W: FnOnce(&V) -> usize,
+ {
+ self.get_or_try_insert_with_admission(key, key_heap_bytes, load,
|value| {
+ Some(value_weight(value))
+ })
+ .await
+ }
+
+ pub(super) async fn get_or_try_insert_with_admission<E, F, Fut, W>(
+ &self,
+ key: Option<K>,
+ key_heap_bytes: usize,
+ load: F,
+ admitted_weight: W,
+ ) -> Result<Arc<V>, E>
+ where
+ F: FnOnce() -> Fut,
+ Fut: Future<Output = Result<Arc<V>, E>>,
+ W: FnOnce(&V) -> Option<usize>,
{
let Some(key) = key else {
return load().await;
@@ -133,9 +151,19 @@ where
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 Some(value_weight) = admitted_weight(value.as_ref()) else {
+ let mut state = self.state.lock().unwrap();
+ if state
+ .entries
+ .peek(&key)
+ .is_some_and(|cached| Arc::ptr_eq(cached, &entry))
+ {
+ state.entries.pop(&key);
+ state.weight = state.weight.saturating_sub(entry.base_weight);
+ }
+ return Ok(value);
+ };
+ let loaded_weight =
entry.base_weight.saturating_add(value_weight.max(1));
let mut state = self.state.lock().unwrap();
if state
.entries
diff --git a/crates/paimon/src/catalog/filesystem.rs
b/crates/paimon/src/catalog/filesystem.rs
index 4d48de2a..a95a8244 100644
--- a/crates/paimon/src/catalog/filesystem.rs
+++ b/crates/paimon/src/catalog/filesystem.rs
@@ -136,7 +136,9 @@ impl FileSystemCatalog {
}
fn build_file_io(&self, path: &str) -> Result<FileIO> {
- let mut builder =
FileIO::from_path(path)?.with_props(self.options.to_map().iter());
+ let mut builder = FileIO::from_path(path)?
+ .with_props(self.options.to_map().iter())
+ .with_cache_context(self.file_io.cache_context());
if let Some(local_cache) = &self.local_cache {
builder = builder.with_local_cache(local_cache.clone());
}
@@ -820,6 +822,7 @@ fn fill_table_name(err: Error, identifier: &Identifier) ->
Error {
#[cfg(not(windows))]
mod tests {
use super::*;
+ use std::sync::Arc;
use tempfile::TempDir;
/// Returns a temp dir guard (keep alive for test duration) and a catalog
using it as warehouse.
@@ -853,6 +856,21 @@ mod tests {
assert!(catalog.file_io().has_local_cache());
}
+ #[test]
+ fn test_filesystem_catalog_derived_file_io_shares_metadata_caches() {
+ let catalog = create_memory_catalog();
+ let derived = catalog.build_file_io("memory:/objects").unwrap();
+
+ assert!(Arc::ptr_eq(
+ &catalog.file_io().file_format_metadata_cache(),
+ &derived.file_format_metadata_cache()
+ ));
+ assert!(Arc::ptr_eq(
+ &catalog.file_io().blob_index_cache(),
+ &derived.blob_index_cache()
+ ));
+ }
+
fn testing_schema() -> Schema {
Schema::builder()
.column(
diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs
b/crates/paimon/src/catalog/rest/rest_catalog.rs
index 67807f0e..1097f799 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -37,7 +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::io::FileIOCacheContext;
use crate::spec::{Partition, PartitionStatistics, Schema, SchemaChange};
use crate::table::{RESTEnv, Table};
use crate::Result;
@@ -61,7 +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>,
+ file_io_cache: FileIOCacheContext,
}
impl RESTCatalog {
@@ -90,8 +90,7 @@ 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())?;
+ let file_io_cache =
FileIOCacheContext::from_props(api_options.to_map())?;
Ok(Self {
api,
@@ -99,7 +98,7 @@ impl RESTCatalog {
warehouse,
data_token_enabled,
local_cache,
- file_format_metadata_cache,
+ file_io_cache,
})
}
@@ -285,7 +284,7 @@ impl Catalog for RESTCatalog {
self.options.clone(),
self.data_token_enabled,
self.local_cache.clone(),
- self.file_format_metadata_cache.clone(),
+ self.file_io_cache.clone(),
)
.await
}
@@ -303,7 +302,7 @@ impl Catalog for RESTCatalog {
self.options.clone(),
self.data_token_enabled,
self.local_cache.clone(),
- self.file_format_metadata_cache.clone(),
+ self.file_io_cache.clone(),
)
.await
.map(crate::catalog::LoadedTable::Object);
@@ -324,7 +323,7 @@ impl Catalog for RESTCatalog {
self.options.clone(),
self.data_token_enabled,
self.local_cache.clone(),
- self.file_format_metadata_cache.clone(),
+ self.file_io_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 7f106011..446a56ea 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::{FileFormatMetadataCacheContext, FileIO, FileIOProvider};
+use crate::io::{FileIO, FileIOCacheContext, FileIOProvider};
use crate::Result;
use super::rest_token::RESTToken;
@@ -55,7 +55,7 @@ pub struct RESTTokenFileIO {
state: RwLock<Option<TokenState>>,
refresh_lock: Mutex<()>,
local_cache: Option<Arc<LocalCache>>,
- file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
+ file_io_cache: FileIOCacheContext,
}
impl std::fmt::Debug for RESTTokenFileIO {
@@ -74,7 +74,7 @@ impl RESTTokenFileIO {
catalog_options: Options,
api: Arc<RESTApi>,
local_cache: Option<Arc<LocalCache>>,
- file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
+ file_io_cache: FileIOCacheContext,
) -> Self {
Self {
identifier,
@@ -84,7 +84,7 @@ impl RESTTokenFileIO {
state: RwLock::new(None),
refresh_lock: Mutex::new(()),
local_cache,
- file_format_metadata_cache,
+ file_io_cache,
}
}
@@ -124,7 +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());
+ builder = builder.with_cache_context(self.file_io_cache.clone());
if let Some(local_cache) = &self.local_cache {
builder = builder.with_local_cache(local_cache.clone());
}
@@ -175,6 +175,16 @@ impl FileIOProvider for RESTTokenFileIO {
async fn create(&self, path: &str) -> Result<(opendal::Operator, String)> {
self.current_file_io().await?.create_static(path)
}
+
+ async fn create_with_cache_namespace(
+ &self,
+ path: &str,
+ ) -> Result<(opendal::Operator, String, Option<String>)> {
+ let file_io = self.current_file_io().await?;
+ let cache_namespace = file_io.cache_namespace_for_path(path)?;
+ let (op, relative_path) = file_io.create_static(path)?;
+ Ok((op, relative_path, Some(cache_namespace)))
+ }
}
#[cfg(test)]
@@ -233,8 +243,8 @@ mod tests {
(options, api, requests, server)
}
- fn metadata_cache(options: &Options) ->
Arc<FileFormatMetadataCacheContext> {
- FileFormatMetadataCacheContext::from_props(options.to_map()).unwrap()
+ fn file_io_cache(options: &Options) -> FileIOCacheContext {
+ FileIOCacheContext::from_props(options.to_map()).unwrap()
}
#[tokio::test]
@@ -243,14 +253,14 @@ mod tests {
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 file_io_cache = file_io_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,
+ file_io_cache,
));
let file_io = token_file_io.build_file_io().await.unwrap();
@@ -264,14 +274,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 file_io_cache = file_io_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,
+ file_io_cache,
));
let file_io = token_file_io.build_file_io().await.unwrap();
@@ -313,14 +323,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 file_io_cache = file_io_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,
+ file_io_cache,
));
let file_io = token_file_io.build_file_io().await.unwrap();
@@ -343,14 +353,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 file_io_cache = file_io_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,
+ file_io_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 0110b80c..3eed389e 100644
--- a/crates/paimon/src/common/options.rs
+++ b/crates/paimon/src/common/options.rs
@@ -94,6 +94,9 @@ impl CatalogOptions {
/// 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";
+
+ /// Maximum catalog-local cache size for decoded BLOB indexes.
+ pub const BLOB_INDEX_CACHE_MAX_SIZE: &'static str =
"cache.blob-index.max-size";
}
/// Configuration options container.
@@ -288,6 +291,10 @@ mod tests {
CatalogOptions::FILE_FORMAT_METADATA_CACHE_MAX_SIZE,
"file-format.metadata-cache.max-size"
);
+ assert_eq!(
+ CatalogOptions::BLOB_INDEX_CACHE_MAX_SIZE,
+ "cache.blob-index.max-size"
+ );
}
#[test]
diff --git a/crates/paimon/src/error.rs b/crates/paimon/src/error.rs
index 47cd9e6e..aa53a6dc 100644
--- a/crates/paimon/src/error.rs
+++ b/crates/paimon/src/error.rs
@@ -179,6 +179,36 @@ impl Error {
_ => false,
}
}
+
+ /// Whether a shared BLOB index failure wraps an I/O error.
+ #[doc(hidden)]
+ pub fn is_io_unexpected(&self) -> bool {
+ match self {
+ Error::IoUnexpected { .. } => true,
+ Error::UnexpectedError {
+ source: Some(source),
+ ..
+ } => error_chain_contains_io_unexpected(source.as_ref()),
+ _ => false,
+ }
+ }
+}
+
+fn error_chain_contains_io_unexpected(error: &(dyn std::error::Error +
'static)) -> bool {
+ if let Some(error) = error.downcast_ref::<Error>() {
+ return error.is_io_unexpected();
+ }
+ if let Some(error) = error.downcast_ref::<std::io::Error>() {
+ if error
+ .get_ref()
+ .is_some_and(|source| error_chain_contains_io_unexpected(source))
+ {
+ return true;
+ }
+ }
+ error
+ .source()
+ .is_some_and(error_chain_contains_io_unexpected)
}
fn is_jindo_fork_error(error: &opendal::Error) -> bool {
diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index 1f370b5a..e81af113 100644
--- a/crates/paimon/src/io/file_io.rs
+++ b/crates/paimon/src/io/file_io.rs
@@ -22,7 +22,6 @@ 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, OnceLock};
use std::task::{Context, Poll};
use std::time::SystemTime;
@@ -33,6 +32,7 @@ use futures::stream::BoxStream;
use futures::{StreamExt, TryStreamExt};
use opendal::raw::{normalize_path, normalize_root};
use opendal::Operator;
+use sha2::{Digest, Sha256};
use tokio_util::compat::FuturesAsyncWriteCompatExt;
use url::Url;
@@ -68,8 +68,8 @@ pub trait FileBlockCache: std::fmt::Debug + Send + Sync +
'static {
/// Already-open readers and writers retain their operator, so its backend must
/// refresh credentials internally if those handles need to outlive
credentials.
///
-/// Custom services must report distinct `(scheme, name, root)` storage
identities
-/// for different namespaces, since these identities are used by the file
cache.
+/// Caching is disabled unless `create_with_cache_namespace` supplies a stable
+/// identity for the concrete storage target.
#[async_trait::async_trait]
pub trait FileIOProvider: std::fmt::Debug + Send + Sync + 'static {
/// Return an operator and its relative path, preserving literal object
keys.
@@ -80,6 +80,16 @@ pub trait FileIOProvider: std::fmt::Debug + Send + Sync +
'static {
/// Object paths that OpenDAL would trim or collapse are rejected by
FileIO.
/// Rename requires both paths to resolve to the same shared service
instance.
async fn create(&self, path: &str) -> crate::Result<(Operator, String)>;
+
+ /// Return a stable namespace for the resolved storage target. `None`
+ /// disables caching; distinct endpoints must use distinct namespaces.
+ 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, None))
+ }
}
#[derive(Clone, Debug)]
@@ -91,18 +101,14 @@ enum FileIOBackend {
#[derive(Clone)]
pub struct FileIO {
backend: FileIOBackend,
+ cache_namespace: Arc<str>,
cache: Option<Arc<LocalCache>>,
- cache_namespace: usize,
file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
-}
-
-static NEXT_FILE_IO_CACHE_NAMESPACE: AtomicUsize = AtomicUsize::new(1);
-
-fn next_file_io_cache_namespace() -> usize {
- NEXT_FILE_IO_CACHE_NAMESPACE.fetch_add(1, Ordering::Relaxed)
+ blob_index_cache: Arc<BlobIndexCacheContext>,
}
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;
pub(crate) struct FileFormatMetadataCacheContext {
max_bytes: usize,
@@ -159,6 +165,82 @@ impl FileFormatMetadataCacheContext {
}
}
+pub(crate) struct BlobIndexCacheContext {
+ max_bytes: usize,
+ process_id: u32,
+ cache: OnceLock<Arc<dyn Any + Send + Sync>>,
+}
+
+impl std::fmt::Debug for BlobIndexCacheContext {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("BlobIndexCacheContext")
+ .field("max_bytes", &self.max_bytes)
+ .finish_non_exhaustive()
+ }
+}
+
+impl BlobIndexCacheContext {
+ pub(crate) fn from_props(props: &HashMap<String, String>) ->
crate::Result<Arc<Self>> {
+ let max_bytes = props
+ .get(CatalogOptions::BLOB_INDEX_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::BLOB_INDEX_CACHE_MAX_SIZE
+ ),
+ })
+ })
+ .transpose()?
+ .unwrap_or(DEFAULT_BLOB_INDEX_CACHE_MAX_BYTES);
+ Ok(Arc::new(Self {
+ max_bytes,
+ process_id: std::process::id(),
+ cache: OnceLock::new(),
+ }))
+ }
+
+ pub(crate) fn belongs_to_current_process(&self) -> bool {
+ self.process_id == std::process::id()
+ }
+
+ 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!("BLOB index cache type mismatch"),
+ }
+ }
+}
+
+#[derive(Clone, Debug)]
+pub(crate) struct FileIOCacheContext {
+ file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
+ blob_index_cache: Arc<BlobIndexCacheContext>,
+}
+
+impl FileIOCacheContext {
+ pub(crate) fn from_props(props: &HashMap<String, String>) ->
crate::Result<Self> {
+ Ok(Self {
+ file_format_metadata_cache:
FileFormatMetadataCacheContext::from_props(props)?,
+ blob_index_cache: BlobIndexCacheContext::from_props(props)?,
+ })
+ }
+}
+
impl std::fmt::Debug for FileIO {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("FileIO")
@@ -168,6 +250,10 @@ impl std::fmt::Debug for FileIO {
"file_format_metadata_cache_max_bytes",
&self.file_format_metadata_cache.max_bytes(),
)
+ .field(
+ "blob_index_cache_max_bytes",
+ &self.blob_index_cache.max_bytes(),
+ )
.finish()
}
}
@@ -191,21 +277,31 @@ impl FileIO {
Ok(self)
}
- /// Replace the storage backend with a provider, retaining the file cache.
+ /// Replace the storage backend with a provider. Caching requires the
provider
+ /// to supply a stable storage namespace for each resolved path.
///
/// Resolution is deferred to async operations, including for file handles
/// subsequently created by [`Self::new_input`] and [`Self::new_output`].
pub fn with_provider(mut self, provider: Arc<dyn FileIOProvider>) -> Self {
self.backend = FileIOBackend::Provider(provider);
- self.cache_namespace = next_file_io_cache_namespace();
self
}
- #[cfg(test)]
pub(crate) fn file_format_metadata_cache(&self) ->
Arc<FileFormatMetadataCacheContext> {
Arc::clone(&self.file_format_metadata_cache)
}
+ pub(crate) fn blob_index_cache(&self) -> Arc<BlobIndexCacheContext> {
+ Arc::clone(&self.blob_index_cache)
+ }
+
+ pub(crate) fn cache_context(&self) -> FileIOCacheContext {
+ FileIOCacheContext {
+ file_format_metadata_cache: self.file_format_metadata_cache(),
+ blob_index_cache: self.blob_index_cache(),
+ }
+ }
+
pub(crate) fn create_static(&self, path: &str) -> crate::Result<(Operator,
String)> {
let FileIOBackend::Storage(storage) = &self.backend else {
return Err(Error::IoUnsupported {
@@ -223,12 +319,37 @@ impl FileIO {
}
}
+ async fn create_with_cache_path(
+ &self,
+ path: &str,
+ ) -> crate::Result<(Operator, String, Option<String>)> {
+ let (op, relative_path, namespace) = match &self.backend {
+ FileIOBackend::Provider(provider) => {
+ let (op, relative_path, namespace) =
+ resolve_provider_with_cache_namespace(provider.as_ref(),
path).await?;
+ (op, relative_path, namespace)
+ }
+ FileIOBackend::Storage(_) => {
+ let (op, relative_path) = self.create_static(path)?;
+ (
+ op,
+ relative_path,
+ Some(self.cache_namespace_for_path(path)?),
+ )
+ }
+ };
+ let cache_path =
+ namespace.map(|namespace| cache_object_path(&namespace, &op,
&relative_path));
+ Ok((op, relative_path, cache_path))
+ }
+
fn file_source(&self, path: &str) -> crate::Result<FileSource> {
match &self.backend {
FileIOBackend::Provider(provider) =>
Ok(FileSource::Provider(provider.clone())),
FileIOBackend::Storage(_) => {
let (op, relative_path) = self.create_static(path)?;
- let cache_path = cache_object_path(&op, &relative_path);
+ let namespace = self.cache_namespace_for_path(path)?;
+ let cache_path = cache_object_path(&namespace, &op,
&relative_path);
Ok(FileSource::Static {
op,
relative_path,
@@ -238,6 +359,24 @@ impl FileIO {
}
}
+ pub(crate) fn cache_namespace_for_path(&self, _path: &str) ->
crate::Result<String> {
+ #[cfg(feature = "storage-azdls")]
+ if let FileIOBackend::Storage(storage) = &self.backend {
+ if let Storage::Azdls { config, .. } = storage.as_ref() {
+ let identity = super::azdls_operator_cache_key(config, _path)?;
+ return Ok(format!("{}\0{identity}", self.cache_namespace));
+ }
+ }
+ #[cfg(feature = "storage-hdfs")]
+ if let FileIOBackend::Storage(storage) = &self.backend {
+ if let Storage::Hdfs { config, .. } = storage.as_ref() {
+ let identity = super::hdfs_effective_name_node(config, _path)?;
+ return Ok(format!("{}\0{identity}", self.cache_namespace));
+ }
+ }
+ Ok(self.cache_namespace.to_string())
+ }
+
#[cfg(test)]
pub(crate) fn has_local_cache(&self) -> bool {
self.cache.is_some()
@@ -288,8 +427,8 @@ impl FileIO {
Ok(InputFile {
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),
+ blob_index_cache: Arc::clone(&self.blob_index_cache),
cache: self
.cache
.as_ref()
@@ -306,8 +445,8 @@ impl FileIO {
Ok(OutputFile {
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),
+ blob_index_cache: Arc::clone(&self.blob_index_cache),
cache: self
.cache
.as_ref()
@@ -488,13 +627,15 @@ impl FileIO {
///
/// Reference:
<https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java#L139>
pub async fn delete_file(&self, path: &str) -> Result<()> {
- let (op, relative_path) = self.create(path).await?;
- let cache_path = cache_object_path(&op, relative_path.as_ref());
+ let (op, relative_path, cache_path) =
self.create_with_cache_path(path).await?;
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(cache) = self.cache.as_ref().filter(|cache|
cache.is_cacheable(path)) {
+ if let (Some(cache), Some(cache_path)) = (
+ self.cache.as_ref().filter(|cache| cache.is_cacheable(path)),
+ cache_path,
+ ) {
cache.invalidate_path(&cache_path).await;
}
@@ -505,8 +646,7 @@ impl FileIO {
///
/// Reference:
<https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java#L139>
pub async fn delete_dir(&self, path: &str) -> Result<()> {
- let (op, relative_path) = self.create(path).await?;
- let cache_path = cache_object_path(&op, relative_path.as_ref());
+ let (op, relative_path, cache_path) =
self.create_with_cache_path(path).await?;
op.delete_with(relative_path.as_ref())
.recursive(true)
@@ -517,7 +657,7 @@ impl FileIO {
format!("Failed to delete directory '{path}'"),
)
})?;
- if let Some(cache) = &self.cache {
+ if let (Some(cache), Some(cache_path)) = (&self.cache, cache_path) {
cache.invalidate_prefix(&cache_path).await;
}
@@ -586,8 +726,8 @@ impl FileIO {
///
/// Reference:
<https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java#L159>
pub async fn rename(&self, src: &str, dst: &str) -> Result<()> {
- let (op_src, relative_path_src) = self.create(src).await?;
- let (op_dst, relative_path_dst) = self.create(dst).await?;
+ let (op_src, relative_path_src, cache_path_src) =
self.create_with_cache_path(src).await?;
+ let (op_dst, relative_path_dst, cache_path_dst) =
self.create_with_cache_path(dst).await?;
if matches!(self.backend, FileIOBackend::Provider(_))
&& !Arc::ptr_eq(op_src.service(), op_dst.service())
{
@@ -596,9 +736,6 @@ impl FileIO {
.to_string(),
});
}
- let cache_path_src = cache_object_path(&op_src,
relative_path_src.as_ref());
- let cache_path_dst = cache_object_path(&op_dst,
relative_path_dst.as_ref());
-
op_src
.rename(relative_path_src.as_ref(), relative_path_dst.as_ref())
.await
@@ -609,8 +746,12 @@ impl FileIO {
)
})?;
if let Some(cache) = &self.cache {
- cache.invalidate_prefix(&cache_path_src).await;
- cache.invalidate_prefix(&cache_path_dst).await;
+ if let Some(cache_path) = cache_path_src {
+ cache.invalidate_prefix(&cache_path).await;
+ }
+ if let Some(cache_path) = cache_path_dst {
+ cache.invalidate_prefix(&cache_path).await;
+ }
}
Ok(())
@@ -626,6 +767,15 @@ async fn resolve_provider(
Ok((op, relative_path))
}
+async fn resolve_provider_with_cache_namespace(
+ provider: &dyn FileIOProvider,
+ path: &str,
+) -> crate::Result<(Operator, String, Option<String>)> {
+ let (op, relative_path, cache_namespace) =
provider.create_with_cache_namespace(path).await?;
+ validate_provider_path(path, &relative_path)?;
+ Ok((op, relative_path, cache_namespace))
+}
+
fn validate_provider_path(path: &str, relative_path: &str) -> Result<()> {
// Filesystem paths retain their existing separator normalization. Object
// keys must not silently resolve to another object through OpenDAL's path
@@ -676,10 +826,11 @@ fn status_path(base_path: &str, entry_path: &str) ->
String {
}
}
-fn cache_object_path(op: &Operator, relative_path: &str) -> String {
+fn cache_object_path(namespace: &str, op: &Operator, relative_path: &str) ->
String {
let info = op.info();
format!(
- "{}\0{}\0{}\0{}",
+ "{}\0{}\0{}\0{}\0{}",
+ namespace,
info.scheme(),
info.name(),
info.root(),
@@ -687,6 +838,29 @@ fn cache_object_path(op: &Operator, relative_path: &str)
-> String {
)
}
+fn storage_cache_namespace(props: &HashMap<String, String>) -> Arc<str> {
+ // Credentials may rotate while the storage namespace stays unchanged.
+ // Endpoints identify storage without coupling cache reuse to credentials.
+ // The complete cache key already includes the operator's canonical scheme;
+ // hashing the builder scheme here would split equivalent storage aliases.
+ let mut endpoints = props
+ .iter()
+ .filter(|(key, _)| key.to_ascii_lowercase().ends_with("endpoint"))
+ .collect::<Vec<_>>();
+ endpoints.sort_unstable_by_key(|(key, _)| *key);
+
+ let mut digest = Sha256::new();
+ for (key, value) in endpoints {
+ let key = key.to_ascii_lowercase();
+ let value = value.trim_end_matches('/');
+ digest.update((key.len() as u64).to_le_bytes());
+ digest.update(key);
+ digest.update((value.len() as u64).to_le_bytes());
+ digest.update(value);
+ }
+ Arc::from(hex::encode(digest.finalize()))
+}
+
/// Whether `path` begins with a Windows drive specifier such as `C:\` or
`C:/`.
pub(crate) fn looks_like_windows_drive_path(path: &str) -> bool {
let bytes = path.as_bytes();
@@ -702,6 +876,7 @@ pub struct FileIOBuilder {
props: HashMap<String, String>,
cache: Option<Arc<LocalCache>>,
file_format_metadata_cache: Option<Arc<FileFormatMetadataCacheContext>>,
+ blob_index_cache: Option<Arc<BlobIndexCacheContext>>,
operator: Option<Operator>,
provider: Option<Arc<dyn FileIOProvider>>,
}
@@ -713,6 +888,7 @@ impl FileIOBuilder {
props: HashMap::default(),
cache: None,
file_format_metadata_cache: None,
+ blob_index_cache: None,
operator: None,
provider: None,
}
@@ -767,6 +943,7 @@ impl FileIOBuilder {
self
}
+ #[cfg(test)]
pub(crate) fn with_file_format_metadata_cache(
mut self,
cache: Arc<FileFormatMetadataCacheContext>,
@@ -775,13 +952,25 @@ impl FileIOBuilder {
self
}
+ pub(crate) fn with_cache_context(mut self, cache: FileIOCacheContext) ->
Self {
+ self.file_format_metadata_cache =
Some(cache.file_format_metadata_cache);
+ self.blob_index_cache = Some(cache.blob_index_cache);
+ self
+ }
+
pub fn build(mut self) -> crate::Result<FileIO> {
let cache = self.cache.clone();
+ let cache_namespace = storage_cache_namespace(&self.props);
let file_format_metadata_cache = self
.file_format_metadata_cache
.clone()
.map(Ok)
.unwrap_or_else(||
FileFormatMetadataCacheContext::from_props(&self.props))?;
+ let blob_index_cache = self
+ .blob_index_cache
+ .clone()
+ .map(Ok)
+ .unwrap_or_else(||
BlobIndexCacheContext::from_props(&self.props))?;
let backend = if let Some(provider) = self.provider.take() {
if self.operator.is_some() {
return Err(Error::ConfigInvalid {
@@ -794,9 +983,10 @@ impl FileIOBuilder {
};
Ok(FileIO {
backend,
+ cache_namespace,
cache,
- cache_namespace: next_file_io_cache_namespace(),
file_format_metadata_cache,
+ blob_index_cache,
})
}
}
@@ -806,6 +996,7 @@ pub trait FileRead: Send + Sync + Unpin + 'static {
async fn read(&self, range: Range<u64>) -> crate::Result<Bytes>;
#[doc(hidden)]
+ #[deprecated(note = "BLOB index caching no longer uses numeric FileIO
namespaces")]
fn cache_namespace(&self) -> Option<usize> {
None
}
@@ -819,6 +1010,11 @@ pub trait FileRead: Send + Sync + Unpin + 'static {
fn file_format_metadata_cache(&self) -> Option<&(dyn Any + Send + Sync)> {
None
}
+
+ #[doc(hidden)]
+ fn blob_index_cache(&self) -> Option<&(dyn Any + Send + Sync)> {
+ None
+ }
}
#[async_trait::async_trait]
@@ -831,15 +1027,15 @@ impl FileRead for opendal::Reader {
enum InputFileReader {
Direct {
reader: opendal::Reader,
- cache_namespace: usize,
- cache_key: String,
+ cache_key: Option<String>,
file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
+ blob_index_cache: Arc<BlobIndexCacheContext>,
},
Cached {
reader: CachedFileReader,
- cache_namespace: usize,
cache_key: String,
file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
+ blob_index_cache: Arc<BlobIndexCacheContext>,
},
}
@@ -854,7 +1050,8 @@ impl FileRead for InputFileReader {
fn cache_key(&self) -> Option<&str> {
match self {
- Self::Direct { cache_key, .. } | Self::Cached { cache_key, .. } =>
Some(cache_key),
+ Self::Direct { cache_key, .. } => cache_key.as_deref(),
+ Self::Cached { cache_key, .. } => Some(cache_key),
}
}
@@ -871,15 +1068,15 @@ impl FileRead for InputFileReader {
}
}
- fn cache_namespace(&self) -> Option<usize> {
- Some(match self {
+ fn blob_index_cache(&self) -> Option<&(dyn Any + Send + Sync)> {
+ match self {
Self::Direct {
- cache_namespace, ..
+ blob_index_cache, ..
}
| Self::Cached {
- cache_namespace, ..
- } => *cache_namespace,
- })
+ blob_index_cache, ..
+ } => Some(blob_index_cache.as_ref()),
+ }
}
}
@@ -997,18 +1194,20 @@ enum FileSource {
}
impl FileSource {
- async fn resolve(&self, path: &str) -> crate::Result<(Operator, String,
String)> {
+ async fn resolve(&self, path: &str) -> crate::Result<(Operator, String,
Option<String>)> {
match self {
Self::Provider(provider) => {
- let (op, relative_path) = resolve_provider(provider.as_ref(),
path).await?;
- let cache_path = cache_object_path(&op, &relative_path);
+ let (op, relative_path, resolved_namespace) =
+ resolve_provider_with_cache_namespace(provider.as_ref(),
path).await?;
+ let cache_path = resolved_namespace
+ .map(|namespace| cache_object_path(&namespace, &op,
&relative_path));
Ok((op, relative_path, cache_path))
}
Self::Static {
op,
relative_path,
cache_path,
- } => Ok((op.clone(), relative_path.clone(), cache_path.clone())),
+ } => Ok((op.clone(), relative_path.clone(),
Some(cache_path.clone()))),
}
}
}
@@ -1017,8 +1216,8 @@ impl FileSource {
pub struct InputFile {
source: FileSource,
path: String,
- cache_namespace: usize,
file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
+ blob_index_cache: Arc<BlobIndexCacheContext>,
cache: Option<Arc<LocalCache>>,
}
@@ -1048,7 +1247,7 @@ impl InputFile {
pub async fn read(&self) -> crate::Result<Bytes> {
let (op, relative_path, cache_path) =
self.source.resolve(&self.path).await?;
- let Some(cache) = &self.cache else {
+ let (Some(cache), Some(cache_path)) = (&self.cache, cache_path) else {
return Ok(op.read(&relative_path).await?.to_bytes());
};
let read_token = cache.read_token(&cache_path);
@@ -1068,33 +1267,33 @@ impl InputFile {
pub async fn reader(&self) -> crate::Result<impl FileRead> {
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 {
+ let (Some(cache), Some(cache_path)) = (&self.cache,
cache_path.as_ref()) else {
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),
+ blob_index_cache: Arc::clone(&self.blob_index_cache),
});
};
- let read_token = cache.read_token(&cache_path);
- let size = if let Some(size) = cache.file_size(&cache_path,
&read_token).await {
+ let read_token = cache.read_token(cache_path);
+ let size = if let Some(size) = cache.file_size(cache_path,
&read_token).await {
size
} else {
let size = op.stat(&relative_path).await?.content_length();
- cache.put_file_size(&cache_path, size, &read_token).await;
+ cache.put_file_size(cache_path, size, &read_token).await;
size
};
Ok(InputFileReader::Cached {
reader: CachedFileReader::new_with_token(
Arc::new(reader),
- &cache_path,
+ cache_path.as_str(),
size,
cache.clone(),
read_token,
),
- cache_namespace: self.cache_namespace,
- cache_key: cache_path,
+ cache_key: cache_path.clone(),
file_format_metadata_cache:
Arc::clone(&self.file_format_metadata_cache),
+ blob_index_cache: Arc::clone(&self.blob_index_cache),
})
}
}
@@ -1103,8 +1302,8 @@ impl InputFile {
pub struct OutputFile {
source: FileSource,
path: String,
- cache_namespace: usize,
file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
+ blob_index_cache: Arc<BlobIndexCacheContext>,
cache: Option<Arc<LocalCache>>,
}
@@ -1123,8 +1322,8 @@ impl OutputFile {
InputFile {
source: self.source,
path: self.path,
- cache_namespace: self.cache_namespace,
file_format_metadata_cache: self.file_format_metadata_cache,
+ blob_index_cache: self.blob_index_cache,
cache,
}
}
@@ -1142,7 +1341,7 @@ impl OutputFile {
.chunk(8 * 1024 * 1024)
.await?,
);
- let Some(cache) = &self.cache else {
+ let (Some(cache), Some(cache_path)) = (&self.cache, cache_path) else {
return Ok(writer);
};
Ok(Box::new(CacheInvalidatingWriter {
@@ -1163,7 +1362,7 @@ impl OutputFile {
.into_futures_async_write()
.compat_write(),
);
- let Some(cache) = &self.cache else {
+ let (Some(cache), Some(cache_path)) = (&self.cache, cache_path) else {
return Ok(writer);
};
Ok(Box::new(CacheInvalidatingAsyncWriter {
@@ -1870,6 +2069,76 @@ mod input_output_test {
.unwrap()
}
+ #[test]
+ fn test_storage_cache_namespace_tracks_endpoint_not_credentials() {
+ let first = HashMap::from([
+ (
+ "s3.endpoint".to_string(),
+ "https://first.example".to_string(),
+ ),
+ ("s3.access-key".to_string(), "first-key".to_string()),
+ ]);
+ let rotated = HashMap::from([
+ (
+ "s3.endpoint".to_string(),
+ "https://first.example".to_string(),
+ ),
+ ("s3.access-key".to_string(), "rotated-key".to_string()),
+ ]);
+ let second = HashMap::from([(
+ "s3.endpoint".to_string(),
+ "https://second.example".to_string(),
+ )]);
+
+ assert_eq!(
+ storage_cache_namespace(&first),
+ storage_cache_namespace(&rotated)
+ );
+ assert_ne!(
+ storage_cache_namespace(&first),
+ storage_cache_namespace(&second)
+ );
+ }
+
+ #[cfg(all(feature = "storage-fs", feature = "storage-memory"))]
+ #[test]
+ fn test_cache_object_path_distinguishes_storage_schemes() {
+ let namespace = storage_cache_namespace(&HashMap::new());
+ let (fs_op, _) = setup_fs_file_io()
+ .create_static("file:/snapshot-1")
+ .unwrap();
+ let (memory_op, _) = setup_memory_file_io()
+ .create_static("memory:/snapshot-1")
+ .unwrap();
+ assert_ne!(
+ cache_object_path(&namespace, &fs_op, "snapshot-1"),
+ cache_object_path(&namespace, &memory_op, "snapshot-1")
+ );
+ }
+
+ #[cfg(feature = "storage-s3")]
+ #[test]
+ fn test_storage_cache_keys_match_for_s3_scheme_aliases() {
+ let mut expected_key = None;
+ for scheme in ["s3", "s3a", "S3", "S3A"] {
+ let file_io = FileIOBuilder::new(scheme)
+ .with_prop("s3.endpoint", "https://s3.example";)
+ .with_prop("s3.region", "us-east-1")
+ .build()
+ .unwrap();
+ let FileSource::Static { cache_path, .. } =
+ file_io.file_source("s3://bucket/snapshot-1").unwrap()
+ else {
+ panic!("expected built-in storage");
+ };
+ if let Some(expected_key) = &expected_key {
+ assert_eq!(&cache_path, expected_key, "{scheme}");
+ } else {
+ expected_key = Some(cache_path);
+ }
+ }
+ }
+
async fn common_test_output_file_write_and_read(file_io: &FileIO, path:
&str) {
let output = file_io.new_output(path).unwrap();
let mut writer = output.writer().await.unwrap();
@@ -2052,6 +2321,57 @@ mod input_output_test {
);
}
+ #[tokio::test]
+ async fn test_blob_index_cache_size_reaches_file_reader() {
+ let path = "memory:/blob-index-cache-size.blob";
+ 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()
+ .blob_index_cache()
+ .unwrap()
+ .downcast_ref::<BlobIndexCacheContext>()
+ .unwrap()
+ .max_bytes(),
+ DEFAULT_BLOB_INDEX_CACHE_MAX_BYTES
+ );
+
+ let configured = FileIOBuilder::new("memory")
+ .with_prop(CatalogOptions::BLOB_INDEX_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()
+ .blob_index_cache()
+ .unwrap()
+ .downcast_ref::<BlobIndexCacheContext>()
+ .unwrap()
+ .max_bytes(),
+ 1024 * 1024
+ );
+ }
+
#[test]
fn test_file_format_metadata_cache_size_rejects_invalid_value() {
let error = FileIOBuilder::new("memory")
@@ -2066,6 +2386,17 @@ mod input_output_test {
.contains(CatalogOptions::FILE_FORMAT_METADATA_CACHE_MAX_SIZE));
}
+ #[test]
+ fn test_blob_index_cache_size_rejects_invalid_value() {
+ let error = FileIOBuilder::new("memory")
+ .with_prop(CatalogOptions::BLOB_INDEX_CACHE_MAX_SIZE, "invalid")
+ .build()
+ .unwrap_err();
+ assert!(error
+ .to_string()
+ .contains(CatalogOptions::BLOB_INDEX_CACHE_MAX_SIZE));
+ }
+
#[tokio::test]
async fn test_output_file_write_and_read_fs() {
let file_io = setup_fs_file_io();
@@ -2208,6 +2539,94 @@ mod input_output_test {
);
}
+ #[cfg(not(windows))]
+ #[tokio::test]
+ async fn
test_file_io_external_cache_invalidates_across_local_scheme_aliases() {
+ #[derive(Debug, Default)]
+ struct TestBlockCache(std::sync::Mutex<HashMap<(String, u64), Bytes>>);
+
+ #[async_trait::async_trait]
+ impl FileBlockCache for TestBlockCache {
+ async fn get(&self, path: &str, range: Range<u64>) ->
Option<Bytes> {
+ self.0
+ .lock()
+ .unwrap()
+ .get(&(path.to_string(), range.start))
+ .filter(|data| data.len() as u64 == range.end -
range.start)
+ .cloned()
+ }
+
+ async fn put(&self, path: &str, offset: u64, data: Bytes) {
+ self.0
+ .lock()
+ .unwrap()
+ .insert((path.to_string(), offset), data);
+ }
+
+ async fn invalidate_path(&self, path: &str) {
+ self.0.lock().unwrap().retain(|(key, _), _| key != path);
+ }
+
+ async fn invalidate_prefix(&self, prefix: &str) {
+ self.0
+ .lock()
+ .unwrap()
+ .retain(|(key, _), _| !key.starts_with(prefix));
+ }
+ }
+
+ for reader_scheme in ["", "file", "fs", "FILE", "FS"] {
+ for writer_scheme in ["", "file", "fs", "FILE", "FS"] {
+ let source_directory = tempfile::tempdir().unwrap();
+ let source_path = source_directory.path().join("snapshot-1");
+ std::fs::write(&source_path, b"old metadata").unwrap();
+ let location = format!("file:{}", source_path.display());
+ let cache = Arc::new(TestBlockCache::default());
+ let reader_io = FileIOBuilder::new(reader_scheme)
+ .build()
+ .unwrap()
+ .with_file_block_cache(cache.clone(), 64, "meta")
+ .unwrap();
+ let writer_io = FileIOBuilder::new(writer_scheme)
+ .build()
+ .unwrap()
+ .with_file_block_cache(cache.clone(), 64, "meta")
+ .unwrap();
+ assert_eq!(
+ reader_io
+ .new_input(&location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ Bytes::from_static(b"old metadata")
+ );
+ assert_eq!(cache.0.lock().unwrap().len(), 1);
+ writer_io
+ .new_output(&location)
+ .unwrap()
+ .write(Bytes::from_static(b"new metadata"))
+ .await
+ .unwrap();
+ assert!(
+ cache.0.lock().unwrap().is_empty(),
+ "reader={reader_scheme:?}, writer={writer_scheme:?}"
+ );
+ assert_eq!(std::fs::read(&source_path).unwrap(), b"new
metadata");
+ assert_eq!(
+ reader_io
+ .new_input(&location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ Bytes::from_static(b"new metadata"),
+ "reader={reader_scheme:?}, writer={writer_scheme:?}"
+ );
+ }
+ }
+ }
+
#[cfg(not(windows))]
#[tokio::test]
async fn test_file_io_local_cache_invalidates_after_delete() {
diff --git a/crates/paimon/src/io/file_io/provider_tests.rs
b/crates/paimon/src/io/file_io/provider_tests.rs
index 270dfcf9..cf949fd0 100644
--- a/crates/paimon/src/io/file_io/provider_tests.rs
+++ b/crates/paimon/src/io/file_io/provider_tests.rs
@@ -24,6 +24,7 @@ struct PrefixProvider {
routes: Vec<(String, Operator)>,
calls: AtomicUsize,
reject: AtomicBool,
+ cacheable: AtomicBool,
}
#[async_trait::async_trait]
@@ -43,6 +44,18 @@ impl FileIOProvider for PrefixProvider {
}
RejectingProvider.create(path).await
}
+
+ async fn create_with_cache_namespace(
+ &self,
+ path: &str,
+ ) -> Result<(Operator, String, Option<String>)> {
+ let (op, relative) = self.create(path).await?;
+ let namespace = self
+ .cacheable
+ .load(Ordering::SeqCst)
+ .then(|| format!("{:p}", Arc::as_ptr(op.service())));
+ Ok((op, relative, namespace))
+ }
}
fn memory_operator() -> Operator {
@@ -54,6 +67,7 @@ fn provider_io(routes: Vec<(String, Operator)>) -> (FileIO,
Arc<PrefixProvider>)
routes,
calls: AtomicUsize::new(0),
reject: AtomicBool::new(false),
+ cacheable: AtomicBool::new(false),
});
let io = FileIOBuilder::new("application-storage")
.with_provider(provider.clone())
@@ -392,6 +406,24 @@ fn with_memory_cache(mut io: FileIO) -> FileIO {
io
}
+#[tokio::test]
+async fn provider_without_namespace_bypasses_caches() {
+ let op = memory_operator();
+ let (io, _) = provider_io(vec![("s3://bucket/".to_string(), op.clone())]);
+ let io = with_memory_cache(io);
+ let path = "s3://bucket/table/snapshot/snapshot-1";
+ let input = io.new_input(path).unwrap();
+ op.write("table/snapshot/snapshot-1", "first")
+ .await
+ .unwrap();
+ assert_eq!(input.reader().await.unwrap().cache_key(), None);
+ assert_eq!(input.read().await.unwrap(), "first");
+ op.write("table/snapshot/snapshot-1", "later")
+ .await
+ .unwrap();
+ assert_eq!(input.read().await.unwrap(), "later");
+}
+
#[tokio::test]
async fn provider_cache_isolates_buckets_and_invalidates_aliases() {
use tokio::io::AsyncWriteExt;
@@ -403,6 +435,7 @@ async fn
provider_cache_isolates_buckets_and_invalidates_aliases() {
("s3a://first/".to_string(), first.clone()),
("oss://second/".to_string(), second.clone()),
]);
+ provider.cacheable.store(true, Ordering::SeqCst);
let io = with_memory_cache(io);
let key = "table/snapshot/snapshot-1";
let a = io.new_input(&format!("s3://first/{key}")).unwrap();
diff --git a/crates/paimon/src/io/storage_hdfs.rs
b/crates/paimon/src/io/storage_hdfs.rs
index 8f40aea1..ab6fcbef 100644
--- a/crates/paimon/src/io/storage_hdfs.rs
+++ b/crates/paimon/src/io/storage_hdfs.rs
@@ -76,25 +76,27 @@ pub(crate) fn hdfs_config_parse(props: HashMap<String,
String>) -> Result<HdfsNa
///
/// Example path: "hdfs://namenode:8020/warehouse/db/table"
pub(crate) fn hdfs_config_build(cfg: &HdfsNativeConfig, path: &str) ->
Result<Operator> {
- let url = Url::parse(path).map_err(|_| Error::ConfigInvalid {
- message: format!("Invalid HDFS url: {path}"),
- })?;
-
let mut cfg = cfg.clone();
-
- if cfg.name_node.is_none() {
- let host = url.host_str().ok_or_else(|| Error::ConfigInvalid {
- message: format!("Invalid HDFS url: {path}, missing name node
host"),
- })?;
- let port_part = url.port().map(|p|
format!(":{p}")).unwrap_or_default();
- cfg.name_node = Some(format!("hdfs://{host}{port_part}"));
- }
-
+ cfg.name_node = Some(hdfs_effective_name_node(&cfg, path)?);
cfg.root = Some("/".to_string());
Ok(Operator::from_config(cfg)?)
}
+pub(crate) fn hdfs_effective_name_node(cfg: &HdfsNativeConfig, path: &str) ->
Result<String> {
+ let url = Url::parse(path).map_err(|_| Error::ConfigInvalid {
+ message: format!("Invalid HDFS url: {path}"),
+ })?;
+ if let Some(name_node) = &cfg.name_node {
+ return Ok(name_node.clone());
+ }
+ let host = url.host_str().ok_or_else(|| Error::ConfigInvalid {
+ message: format!("Invalid HDFS url: {path}, missing name node host"),
+ })?;
+ let port_part = url.port().map(|p| format!(":{p}")).unwrap_or_default();
+ Ok(format!("hdfs://{host}{port_part}"))
+}
+
#[cfg(test)]
#[allow(deprecated)]
mod tests {
@@ -147,6 +149,20 @@ mod tests {
assert_eq!(op.info().scheme().to_string(), "hdfs-native");
}
+ #[test]
+ fn test_hdfs_effective_name_node() {
+ let mut cfg = HdfsNativeConfig::default();
+ assert_eq!(
+ hdfs_effective_name_node(&cfg,
"hdfs://cluster-a:8020/table/data.blob").unwrap(),
+ "hdfs://cluster-a:8020"
+ );
+ cfg.name_node = Some("hdfs://nameservice-b".to_string());
+ assert_eq!(
+ hdfs_effective_name_node(&cfg,
"hdfs://logical-name/table/data.blob").unwrap(),
+ "hdfs://nameservice-b"
+ );
+ }
+
#[test]
fn test_hdfs_config_build_invalid_url() {
let cfg = HdfsNativeConfig::default();
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index e608927c..3347bb12 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -108,10 +108,6 @@ impl FileRead for TimedFileRead {
result
}
- fn cache_namespace(&self) -> Option<usize> {
- self.inner.cache_namespace()
- }
-
fn cache_key(&self) -> Option<&str> {
self.inner.cache_key()
}
@@ -119,6 +115,10 @@ impl FileRead for TimedFileRead {
fn file_format_metadata_cache(&self) -> Option<&(dyn std::any::Any + Send
+ Sync)> {
self.inner.file_format_metadata_cache()
}
+
+ fn blob_index_cache(&self) -> Option<&(dyn std::any::Any + Send + Sync)> {
+ self.inner.blob_index_cache()
+ }
}
/// Reads data files through their format-specific readers.
@@ -2083,18 +2083,24 @@ mod tests {
}
#[tokio::test]
- async fn timed_file_read_preserves_cache_namespace() {
+ async fn timed_file_read_preserves_cache_context() {
let file_io = FileIOBuilder::new("memory").build().unwrap();
let input = file_io.new_input("memory:/timed-file").unwrap();
let reader = input.reader().await.unwrap();
- let cache_namespace = reader.cache_namespace();
+ let blob_index_cache = file_io.blob_index_cache();
let timed = TimedFileRead {
inner: Box::new(reader),
timing: Arc::new(DataFileReadTiming::default()),
};
- assert!(cache_namespace.is_some());
- assert_eq!(timed.cache_namespace(), cache_namespace);
+ assert!(std::ptr::eq(
+ timed
+ .blob_index_cache()
+ .unwrap()
+ .downcast_ref::<crate::io::BlobIndexCacheContext>()
+ .unwrap(),
+ blob_index_cache.as_ref()
+ ));
}
#[test]
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index d9e958b6..f3e02e50 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -870,8 +870,7 @@ pub(crate) async fn rest_query_auth_table() -> Table {
options.set("token.provider", "bear");
options.set("token", "test_token");
let api = std::sync::Arc::new(RESTApi::new(options.clone(),
false).await.unwrap());
- let metadata_cache =
-
crate::io::FileFormatMetadataCacheContext::from_props(options.to_map()).unwrap();
+ let file_io_cache =
crate::io::FileIOCacheContext::from_props(options.to_map()).unwrap();
let table = query_auth_table();
Table {
rest_env: Some(RESTEnv::new(
@@ -881,7 +880,7 @@ pub(crate) async fn rest_query_auth_table() -> Table {
options,
false,
None,
- metadata_cache,
+ file_io_cache,
)),
..table
}
diff --git a/crates/paimon/src/table/rest_env.rs
b/crates/paimon/src/table/rest_env.rs
index 7dfda992..cf3ad2a6 100644
--- a/crates/paimon/src/table/rest_env.rs
+++ b/crates/paimon/src/table/rest_env.rs
@@ -24,7 +24,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::{FileFormatMetadataCacheContext, FileIO};
+use crate::io::{FileIO, FileIOCacheContext};
use crate::spec::{CoreOptions, TableSchema, PATH_OPTION};
use crate::table::snapshot_commit::{RESTSnapshotCommit, SnapshotCommit};
use crate::table::{ObjectTable, Table};
@@ -54,8 +54,7 @@ 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())?;
+ let file_io_cache =
FileIOCacheContext::from_props(api.options().to_map())?;
RESTEnv::build_table(
&identifier,
response,
@@ -63,7 +62,7 @@ impl Table {
rest_options,
data_token_enabled,
local_cache,
- file_format_metadata_cache,
+ file_io_cache,
)
.await
}
@@ -79,7 +78,7 @@ pub struct RESTEnv {
options: Options,
data_token_enabled: bool,
local_cache: Option<Arc<LocalCache>>,
- file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
+ file_io_cache: FileIOCacheContext,
}
impl std::fmt::Debug for RESTEnv {
@@ -100,7 +99,7 @@ impl RESTEnv {
options: Options,
data_token_enabled: bool,
local_cache: Option<Arc<LocalCache>>,
- file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
+ file_io_cache: FileIOCacheContext,
) -> Self {
Self {
identifier,
@@ -109,7 +108,7 @@ impl RESTEnv {
options,
data_token_enabled,
local_cache,
- file_format_metadata_cache,
+ file_io_cache,
}
}
@@ -201,7 +200,7 @@ impl RESTEnv {
self.options.clone(),
self.data_token_enabled,
self.local_cache.clone(),
- self.file_format_metadata_cache.clone(),
+ self.file_io_cache.clone(),
)
.await
}
@@ -213,7 +212,7 @@ impl RESTEnv {
options: Options,
data_token_enabled: bool,
local_cache: Option<Arc<LocalCache>>,
- file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
+ file_io_cache: FileIOCacheContext,
) -> Result<Table> {
let response = Self::fetch_table_response(identifier, &api).await?;
Self::build_table(
@@ -223,7 +222,7 @@ impl RESTEnv {
options,
data_token_enabled,
local_cache,
- file_format_metadata_cache,
+ file_io_cache,
)
.await
}
@@ -245,7 +244,7 @@ impl RESTEnv {
options: Options,
data_token_enabled: bool,
local_cache: Option<Arc<LocalCache>>,
- file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
+ file_io_cache: FileIOCacheContext,
) -> Result<Table> {
let identifier = response_identifier(identifier, &response)?;
let schema = response.schema.ok_or_else(|| Error::DataInvalid {
@@ -311,7 +310,7 @@ impl RESTEnv {
&options,
data_token_enabled && !is_external,
local_cache.clone(),
- file_format_metadata_cache.clone(),
+ file_io_cache.clone(),
)
.await?;
@@ -322,7 +321,7 @@ impl RESTEnv {
options,
data_token_enabled,
local_cache,
- file_format_metadata_cache,
+ file_io_cache,
);
let parsed_identifier = identifier.parsed_object_name()?;
let branch = parsed_identifier.branch_or_default().to_string();
@@ -348,7 +347,7 @@ impl RESTEnv {
options: Options,
data_token_enabled: bool,
local_cache: Option<Arc<LocalCache>>,
- file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
+ file_io_cache: FileIOCacheContext,
) -> Result<ObjectTable> {
let identifier = response_identifier(identifier, &response)?;
let schema = response.schema.ok_or_else(|| Error::DataInvalid {
@@ -392,7 +391,7 @@ impl RESTEnv {
&options,
data_token_enabled && !is_external,
local_cache,
- file_format_metadata_cache,
+ file_io_cache,
)
.await?;
@@ -406,7 +405,7 @@ impl RESTEnv {
options: &Options,
use_data_token: bool,
local_cache: Option<Arc<LocalCache>>,
- file_format_metadata_cache: Arc<FileFormatMetadataCacheContext>,
+ file_io_cache: FileIOCacheContext,
) -> Result<FileIO> {
if use_data_token {
return Arc::new(RESTTokenFileIO::new(
@@ -415,14 +414,14 @@ impl RESTEnv {
options.clone(),
api,
local_cache,
- file_format_metadata_cache,
+ file_io_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);
+ builder = builder.with_cache_context(file_io_cache);
if let Some(local_cache) = local_cache {
builder = builder.with_local_cache(local_cache);
}
@@ -562,7 +561,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 file_io_cache =
FileIOCacheContext::from_props(options.to_map()).unwrap();
let rest_env = RESTEnv::new(
Identifier::new("database", "table"),
@@ -571,7 +570,7 @@ mod tests {
options,
false,
local_cache,
- metadata_cache,
+ file_io_cache,
);
assert!(rest_env.has_local_cache());
@@ -586,7 +585,7 @@ mod tests {
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 file_io_cache =
FileIOCacheContext::from_props(options.to_map()).unwrap();
let identifier = Identifier::new("database", "table");
let first = RESTEnv::build_file_io(
@@ -596,7 +595,7 @@ mod tests {
&options,
false,
None,
- metadata_cache.clone(),
+ file_io_cache.clone(),
)
.await
.unwrap();
@@ -607,7 +606,7 @@ mod tests {
&options,
false,
None,
- metadata_cache,
+ file_io_cache,
)
.await
.unwrap();
@@ -616,5 +615,9 @@ mod tests {
&first.file_format_metadata_cache(),
&second.file_format_metadata_cache()
));
+ assert!(Arc::ptr_eq(
+ &first.blob_index_cache(),
+ &second.blob_index_cache()
+ ));
}
}
diff --git a/crates/paimon/src/table/snapshot_manager.rs
b/crates/paimon/src/table/snapshot_manager.rs
index fc1b681e..1ca5344d 100644
--- a/crates/paimon/src/table/snapshot_manager.rs
+++ b/crates/paimon/src/table/snapshot_manager.rs
@@ -586,8 +586,7 @@ 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 metadata_cache =
-
crate::io::FileFormatMetadataCacheContext::from_props(options.to_map()).unwrap();
+ let file_io_cache =
crate::io::FileIOCacheContext::from_props(options.to_map()).unwrap();
let env = crate::table::RESTEnv::new(
id.clone(),
"uuid".into(),
@@ -595,7 +594,7 @@ mod tests {
options,
false,
None,
- metadata_cache,
+ file_io_cache,
);
let (io, manager) = setup("/rest-table").await;
manager.commit_snapshot(&test_snapshot(2)).await.unwrap();
diff --git a/crates/paimon/src/table/table_commit/parity_tests.rs
b/crates/paimon/src/table/table_commit/parity_tests.rs
index 3c2ecbac..e4aa339c 100644
--- a/crates/paimon/src/table/table_commit/parity_tests.rs
+++ b/crates/paimon/src/table/table_commit/parity_tests.rs
@@ -614,8 +614,7 @@ 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 metadata_cache =
-
crate::io::FileFormatMetadataCacheContext::from_props(options.to_map()).unwrap();
+ let file_io_cache =
crate::io::FileIOCacheContext::from_props(options.to_map()).unwrap();
let env = crate::table::RESTEnv::new(
identifier.clone(),
"uuid".into(),
@@ -623,7 +622,7 @@ async fn
rest_commit_uses_catalog_snapshot_schema_and_retry_identity() {
options,
false,
None,
- metadata_cache,
+ file_io_cache,
);
let table = Table::new(
io.clone(),
diff --git a/crates/paimon/src/table/table_commit/recovery_tests.rs
b/crates/paimon/src/table/table_commit/recovery_tests.rs
index ffaaae51..32b5d3ec 100644
--- a/crates/paimon/src/table/table_commit/recovery_tests.rs
+++ b/crates/paimon/src/table/table_commit/recovery_tests.rs
@@ -546,8 +546,7 @@ 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 metadata_cache =
-
crate::io::FileFormatMetadataCacheContext::from_props(options.to_map()).unwrap();
+ let file_io_cache =
crate::io::FileIOCacheContext::from_props(options.to_map()).unwrap();
let env = crate::table::RESTEnv::new(
identifier.clone(),
"uuid".into(),
@@ -555,7 +554,7 @@ async fn
rest_delete_writer_pins_catalog_snapshot_and_preserves_vectors() {
options,
false,
None,
- metadata_cache,
+ file_io_cache,
);
let table = Table::new(io.clone(), identifier, path.into(), schema,
Some(env));
let mut commit = TableCommit::new(table.clone(), "rest-writer".into());
diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md
index 0993d198..2d05ec11 100644
--- a/docs/src/getting-started.md
+++ b/docs/src/getting-started.md
@@ -362,6 +362,12 @@ blocking worker, reads only block headers and file
metadata, and validates
payload CRC lazily on the first hit. Use a separate `local-cache.dir` for each
worker or process because processes do not share exact LRU or size accounting.
+### BLOB Index Cache
+
+Decoded `.blob` indexes are cached per catalog. Set
+`cache.blob-index.max-size` to control the budget (default `64 MiB`; `0`
+disables caching). BLOB payloads are not cached.
+
### Manage Databases
```rust
