You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/07/23 15:14:08 UTC
(paimon-rust) branch main updated: feat: add local disk cache (#600)
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 612f1a26 feat: add local disk cache (#600)
612f1a26 is described below
commit 612f1a2666b020b99b1d5cb3d863d8a002e9ce79
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Thu Jul 23 23:14:03 2026 +0800
feat: add local disk cache (#600)
---
crates/paimon/Cargo.toml | 2 +-
crates/paimon/src/catalog/filesystem.rs | 31 +-
crates/paimon/src/catalog/rest/rest_catalog.rs | 36 +-
.../paimon/src/catalog/rest/rest_token_file_io.rs | 57 +-
crates/paimon/src/common/options.rs | 30 +
crates/paimon/src/io/cache/disk.rs | 1240 ++++++++++++++++++++
crates/paimon/src/io/cache/file_type.rs | 165 +++
crates/paimon/src/io/cache/mod.rs | 529 +++++++++
crates/paimon/src/io/cache/reader.rs | 726 ++++++++++++
crates/paimon/src/io/file_io.rs | 601 +++++++++-
crates/paimon/src/io/mod.rs | 1 +
crates/paimon/src/table/rest_env.rs | 72 +-
docs/src/getting-started.md | 41 +
13 files changed, 3513 insertions(+), 18 deletions(-)
diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml
index 043f95f8..b84e7dcd 100644
--- a/crates/paimon/Cargo.toml
+++ b/crates/paimon/Cargo.toml
@@ -60,7 +60,7 @@ url = "2.5.2"
async-trait = "0.1.81"
bytes = "1.7.1"
bitflags = "2.6.0"
-tokio = { version = "1.39.2", features = ["macros"] }
+tokio = { version = "1.39.2", features = ["fs", "io-util", "macros", "sync", "time"] }
chrono = { version = "0.4.38", features = ["serde"] }
serde = { version = "1", features = ["derive"] }
serde_bytes = "0.11.15"
diff --git a/crates/paimon/src/catalog/filesystem.rs b/crates/paimon/src/catalog/filesystem.rs
index b34d4472..cf2f745a 100644
--- a/crates/paimon/src/catalog/filesystem.rs
+++ b/crates/paimon/src/catalog/filesystem.rs
@@ -24,6 +24,7 @@ use std::collections::HashMap;
use crate::catalog::{Catalog, Database, Identifier, DB_LOCATION_PROP, DB_SUFFIX};
use crate::common::{CatalogOptions, Options};
use crate::error::{ConfigInvalidSnafu, Error, Result};
+use crate::io::cache::create_local_cache;
use crate::io::FileIO;
use crate::spec::{Schema, TableSchema};
use crate::table::{SchemaManager, Table};
@@ -101,9 +102,13 @@ impl FileSystemCatalog {
message: format!("Missing required option: {}", CatalogOptions::WAREHOUSE),
})?;
- let file_io = FileIO::from_path(&warehouse)?
- .with_props(options.to_map().iter())
- .build()?;
+ let local_cache = create_local_cache(&options)?;
+ let mut file_io_builder =
+ FileIO::from_path(&warehouse)?.with_props(options.to_map().iter());
+ if let Some(local_cache) = local_cache {
+ file_io_builder = file_io_builder.with_local_cache(local_cache);
+ }
+ let file_io = file_io_builder.build()?;
Ok(Self { file_io, warehouse })
}
@@ -502,6 +507,26 @@ mod tests {
FileSystemCatalog::new(options).unwrap()
}
+ #[test]
+ fn test_filesystem_catalog_builds_local_cache_from_catalog_options() {
+ let warehouse = TempDir::new().unwrap();
+ let cache = TempDir::new().unwrap();
+ let mut options = Options::new();
+ options.set(
+ CatalogOptions::WAREHOUSE,
+ warehouse.path().to_string_lossy(),
+ );
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ options.set(
+ CatalogOptions::LOCAL_CACHE_DIR,
+ cache.path().to_string_lossy(),
+ );
+
+ let catalog = FileSystemCatalog::new(options).unwrap();
+
+ assert!(catalog.file_io().has_local_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 2057d3f5..7a19ba12 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -33,6 +33,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::spec::{Partition, Schema, SchemaChange};
use crate::table::{RESTEnv, Table};
use crate::Result;
@@ -52,6 +53,8 @@ pub struct RESTCatalog {
warehouse: String,
/// Whether data token is enabled for FileIO construction.
data_token_enabled: bool,
+ /// Catalog-scoped cache shared by all table FileIO instances.
+ local_cache: Option<Arc<LocalCache>>,
}
impl RESTCatalog {
@@ -70,7 +73,6 @@ impl RESTCatalog {
.ok_or_else(|| RestError::BadRequest {
message: format!("Missing required option: {}", CatalogOptions::WAREHOUSE),
})?;
-
let api = Arc::new(RESTApi::new(options.clone(), config_required).await?);
let data_token_enabled = api
@@ -80,12 +82,14 @@ impl RESTCatalog {
.unwrap_or(false);
let api_options = api.options().clone();
+ let local_cache = create_local_cache_with_namespace(&options, &api_options)?;
Ok(Self {
api,
options: api_options,
warehouse,
data_token_enabled,
+ local_cache,
})
}
@@ -104,6 +108,11 @@ impl RESTCatalog {
self.data_token_enabled
}
+ #[cfg(test)]
+ fn has_local_cache(&self) -> bool {
+ self.local_cache.is_some()
+ }
+
/// List databases with pagination.
pub async fn list_databases_paged(
&self,
@@ -201,6 +210,7 @@ impl Catalog for RESTCatalog {
self.api.clone(),
self.options.clone(),
self.data_token_enabled,
+ self.local_cache.clone(),
)
.await
}
@@ -531,3 +541,27 @@ where
}
})
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[tokio::test]
+ async fn test_rest_catalog_builds_local_cache_from_client_options() {
+ let cache = tempfile::tempdir().unwrap();
+ let mut options = Options::new();
+ options.set(CatalogOptions::URI, "http://localhost:1");
+ options.set(CatalogOptions::WAREHOUSE, "test-warehouse");
+ options.set("token.provider", "bear");
+ options.set("token", "test-token");
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ options.set(
+ CatalogOptions::LOCAL_CACHE_DIR,
+ cache.path().to_string_lossy(),
+ );
+
+ let catalog = RESTCatalog::new(options, false).await.unwrap();
+
+ assert!(catalog.has_local_cache());
+ }
+}
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 502a0d99..8e2accfc 100644
--- a/crates/paimon/src/catalog/rest/rest_token_file_io.rs
+++ b/crates/paimon/src/catalog/rest/rest_token_file_io.rs
@@ -22,6 +22,7 @@
//! and automatic refresh.
use std::collections::HashMap;
+use std::sync::Arc;
use tokio::sync::{OnceCell, RwLock};
@@ -29,6 +30,7 @@ use crate::api::rest_api::RESTApi;
use crate::api::rest_util::RESTUtil;
use crate::catalog::Identifier;
use crate::common::{CatalogOptions, Options};
+use crate::io::cache::LocalCache;
use crate::io::FileIO;
use crate::Result;
@@ -56,6 +58,8 @@ pub struct RESTTokenFileIO {
api: OnceCell<RESTApi>,
/// Cached token with RwLock for concurrent access.
token: RwLock<Option<RESTToken>>,
+ /// Catalog-scoped cache preserved across token-driven FileIO rebuilds.
+ local_cache: Option<Arc<LocalCache>>,
}
impl RESTTokenFileIO {
@@ -65,13 +69,20 @@ impl RESTTokenFileIO {
/// * `identifier` - Table identifier for token requests.
/// * `path` - Table path for FileIO construction.
/// * `catalog_options` - Catalog options for RESTApi and FileIO.
- pub fn new(identifier: Identifier, path: String, catalog_options: Options) -> Self {
+ /// * `local_cache` - Catalog-scoped local cache shared across FileIO rebuilds.
+ pub(crate) fn new(
+ identifier: Identifier,
+ path: String,
+ catalog_options: Options,
+ local_cache: Option<Arc<LocalCache>>,
+ ) -> Self {
Self {
identifier,
path,
catalog_options,
api: OnceCell::new(),
token: RwLock::new(None),
+ local_cache,
}
}
@@ -98,11 +109,18 @@ impl RESTTokenFileIO {
// Build FileIO with merged properties
let mut builder = FileIO::from_path(&self.path)?;
builder = builder.with_props(merged_props);
+ if let Some(local_cache) = &self.local_cache {
+ builder = builder.with_local_cache(local_cache.clone());
+ }
builder.build()
}
None => {
// No token available, build FileIO from path only
- FileIO::from_path(&self.path)?.build()
+ let mut builder = FileIO::from_path(&self.path)?;
+ if let Some(local_cache) = &self.local_cache {
+ builder = builder.with_local_cache(local_cache.clone());
+ }
+ builder.build()
}
}
}
@@ -191,3 +209,38 @@ impl RESTTokenFileIO {
merged
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::io::cache::create_local_cache;
+
+ #[tokio::test]
+ async fn test_token_file_io_keeps_catalog_local_cache() {
+ let cache_directory = tempfile::tempdir().unwrap();
+ let table_directory = tempfile::tempdir().unwrap();
+ let mut options = Options::new();
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ options.set(
+ CatalogOptions::LOCAL_CACHE_DIR,
+ cache_directory.path().to_string_lossy(),
+ );
+ let local_cache = create_local_cache(&options).unwrap();
+ let token_file_io = RESTTokenFileIO::new(
+ Identifier::new("database", "table"),
+ table_directory.path().to_string_lossy().into_owned(),
+ options,
+ local_cache,
+ );
+ let valid_until = std::time::SystemTime::now()
+ .duration_since(std::time::UNIX_EPOCH)
+ .unwrap()
+ .as_millis() as i64
+ + TOKEN_EXPIRATION_SAFE_TIME_MILLIS * 2;
+ *token_file_io.token.write().await = Some(RESTToken::new(HashMap::new(), valid_until));
+
+ let file_io = token_file_io.build_file_io().await.unwrap();
+
+ assert!(file_io.has_local_cache());
+ }
+}
diff --git a/crates/paimon/src/common/options.rs b/crates/paimon/src/common/options.rs
index 792c6017..5ad2147d 100644
--- a/crates/paimon/src/common/options.rs
+++ b/crates/paimon/src/common/options.rs
@@ -72,6 +72,21 @@ impl CatalogOptions {
/// DLF OSS endpoint override.
pub const DLF_OSS_ENDPOINT: &'static str = "dlf.oss-endpoint";
+
+ /// Whether to enable local block caching for file reads.
+ pub const LOCAL_CACHE_ENABLED: &'static str = "local-cache.enabled";
+
+ /// Directory for the local disk block cache.
+ pub const LOCAL_CACHE_DIR: &'static str = "local-cache.dir";
+
+ /// Maximum total encoded size of the local block cache.
+ pub const LOCAL_CACHE_MAX_SIZE: &'static str = "local-cache.max-size";
+
+ /// Block size used by the local cache.
+ pub const LOCAL_CACHE_BLOCK_SIZE: &'static str = "local-cache.block-size";
+
+ /// Comma-separated file types eligible for local caching.
+ pub const LOCAL_CACHE_WHITELIST: &'static str = "local-cache.whitelist";
}
/// Configuration options container.
@@ -214,4 +229,19 @@ mod tests {
assert_eq!(options1.get("key1"), Some(&"overwritten".to_string()));
assert_eq!(options1.get("key2"), Some(&"value2".to_string()));
}
+
+ #[test]
+ fn test_local_cache_catalog_option_keys() {
+ assert_eq!(CatalogOptions::LOCAL_CACHE_ENABLED, "local-cache.enabled");
+ assert_eq!(CatalogOptions::LOCAL_CACHE_DIR, "local-cache.dir");
+ assert_eq!(CatalogOptions::LOCAL_CACHE_MAX_SIZE, "local-cache.max-size");
+ assert_eq!(
+ CatalogOptions::LOCAL_CACHE_BLOCK_SIZE,
+ "local-cache.block-size"
+ );
+ assert_eq!(
+ CatalogOptions::LOCAL_CACHE_WHITELIST,
+ "local-cache.whitelist"
+ );
+ }
}
diff --git a/crates/paimon/src/io/cache/disk.rs b/crates/paimon/src/io/cache/disk.rs
new file mode 100644
index 00000000..e63a2b5d
--- /dev/null
+++ b/crates/paimon/src/io/cache/disk.rs
@@ -0,0 +1,1240 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use indexmap::IndexMap;
+use sha2::{Digest, Sha256};
+use std::collections::{HashMap, HashSet};
+use std::io::Read;
+use std::path::{Path, PathBuf};
+use std::sync::{Arc, Mutex, OnceLock, Weak};
+use tokio::io::AsyncWriteExt;
+
+const CACHE_MAGIC: &[u8; 8] = b"PAIMONLC";
+const CACHE_FORMAT_VERSION: u8 = 2;
+const FIXED_HEADER_LEN: usize = CACHE_MAGIC.len() + 1 + 4 + 4 + 8 + 8 + 8;
+const CHECKSUM_LEN: usize = 4;
+const MAX_CACHE_KEY_HEADER_LEN: usize = 1024 * 1024;
+
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub(super) struct BlockKey {
+ namespace: String,
+ path: String,
+ block_size: u64,
+ block_index: u64,
+}
+
+impl BlockKey {
+ #[cfg(test)]
+ pub(super) fn new(path: impl Into<String>, block_size: u64, block_index: u64) -> Self {
+ Self::with_namespace("", path, block_size, block_index)
+ }
+
+ pub(super) fn with_namespace(
+ namespace: impl Into<String>,
+ path: impl Into<String>,
+ block_size: u64,
+ block_index: u64,
+ ) -> Self {
+ Self {
+ namespace: namespace.into(),
+ path: path.into(),
+ block_size,
+ block_index,
+ }
+ }
+
+ pub(super) fn cache_relative_path(&self) -> PathBuf {
+ let mut digest = Sha256::new();
+ digest.update([CACHE_FORMAT_VERSION]);
+ digest.update((self.namespace.len() as u64).to_le_bytes());
+ digest.update(self.namespace.as_bytes());
+ digest.update((self.path.len() as u64).to_le_bytes());
+ digest.update(self.path.as_bytes());
+ digest.update(self.block_size.to_le_bytes());
+ digest.update(self.block_index.to_le_bytes());
+ let hex = hex::encode(digest.finalize());
+ PathBuf::from(&hex[..2]).join(hex)
+ }
+}
+
+#[derive(Debug)]
+pub(super) struct BlockDecodeError(&'static str);
+
+impl std::fmt::Display for BlockDecodeError {
+ fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ formatter.write_str(self.0)
+ }
+}
+
+#[derive(Debug)]
+pub(super) struct DiskCache {
+ root: PathBuf,
+ state: Mutex<CacheState>,
+ recovered: tokio::sync::OnceCell<()>,
+ in_flight: tokio::sync::Mutex<HashMap<BlockKey, Weak<tokio::sync::Mutex<()>>>>,
+ path_states: Mutex<HashMap<LogicalPath, Weak<PathCacheState>>>,
+ prefix_barrier: tokio::sync::RwLock<()>,
+}
+
+#[derive(Debug, Default)]
+struct CacheState {
+ entries: IndexMap<BlockKey, u64>,
+ paths: HashMap<LogicalPath, HashSet<BlockKey>>,
+ file_sizes: IndexMap<LogicalPath, u64>,
+ current_size: u64,
+ max_size: Option<u64>,
+}
+
+#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+struct LogicalPath {
+ namespace: String,
+ path: String,
+}
+
+#[derive(Debug)]
+struct PathCacheState {
+ generation: std::sync::atomic::AtomicU64,
+ publish_gate: tokio::sync::RwLock<()>,
+}
+
+#[derive(Clone)]
+pub(in crate::io) struct CacheReadToken {
+ generation: u64,
+ state: Arc<PathCacheState>,
+}
+
+impl CacheReadToken {
+ pub(super) fn is_current(&self) -> bool {
+ self.state
+ .generation
+ .load(std::sync::atomic::Ordering::SeqCst)
+ == self.generation
+ }
+
+ pub(super) async fn publish_guard(&self) -> tokio::sync::RwLockReadGuard<'_, ()> {
+ self.state.publish_gate.read().await
+ }
+}
+
+impl DiskCache {
+ #[cfg(test)]
+ pub(super) fn new(root: impl AsRef<Path>, max_size: Option<u64>) -> crate::Result<Self> {
+ let root = prepare_cache_root(root.as_ref())?;
+ Ok(Self::new_resolved(root, max_size))
+ }
+
+ pub(super) fn shared(
+ root: impl AsRef<Path>,
+ max_size: Option<u64>,
+ ) -> crate::Result<Arc<Self>> {
+ let root = prepare_cache_root(root.as_ref())?;
+ let mut registry = disk_cache_registry()
+ .lock()
+ .unwrap_or_else(|error| error.into_inner());
+ if registry.len() >= 1024 {
+ registry.retain(|_, cache| cache.strong_count() > 0);
+ }
+ if let Some(cache) = registry.get(&root).and_then(Weak::upgrade) {
+ cache.tighten_max_size(max_size);
+ return Ok(cache);
+ }
+
+ let cache = Arc::new(Self::new_resolved(root.clone(), max_size));
+ registry.insert(root, Arc::downgrade(&cache));
+ Ok(cache)
+ }
+
+ fn new_resolved(root: PathBuf, max_size: Option<u64>) -> Self {
+ Self {
+ root,
+ state: Mutex::new(CacheState {
+ max_size,
+ ..CacheState::default()
+ }),
+ recovered: tokio::sync::OnceCell::new(),
+ in_flight: tokio::sync::Mutex::new(HashMap::new()),
+ path_states: Mutex::new(HashMap::new()),
+ prefix_barrier: tokio::sync::RwLock::new(()),
+ }
+ }
+
+ fn tighten_max_size(&self, requested: Option<u64>) {
+ let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
+ state.max_size = match (state.max_size, requested) {
+ (None, requested) => requested,
+ (current, None) => current,
+ (Some(current), Some(requested)) => Some(current.min(requested)),
+ };
+ }
+
+ async fn ensure_recovered(&self) {
+ self.recovered
+ .get_or_init(|| async {
+ let root = self.root.clone();
+ let recovered = tokio::task::spawn_blocking(move || {
+ if let Err(error) = cleanup_temporary_files(&root) {
+ log::warn!(
+ "Failed to clean temporary files in local cache directory '{}': {error}",
+ root.display()
+ );
+ }
+ scan_existing_blocks(&root)
+ })
+ .await;
+ match recovered {
+ Ok(recovered) => {
+ let mut state = self
+ .state
+ .lock()
+ .unwrap_or_else(|error| error.into_inner());
+ for (key, encoded_size) in recovered.entries {
+ insert_state_entry(&mut state, key, encoded_size);
+ }
+ }
+ Err(error) => {
+ log::warn!(
+ "Failed to recover local cache directory '{}': {error}",
+ self.root.display()
+ );
+ }
+ }
+ })
+ .await;
+ self.evict_over_limit().await;
+ }
+
+ pub(super) fn read_token(&self, namespace: &str, path: &str) -> CacheReadToken {
+ let state = self.path_state(namespace, path);
+ CacheReadToken {
+ generation: state.generation.load(std::sync::atomic::Ordering::SeqCst),
+ state,
+ }
+ }
+
+ pub(super) async fn prefix_read_guard(&self) -> tokio::sync::RwLockReadGuard<'_, ()> {
+ self.prefix_barrier.read().await
+ }
+
+ fn path_state(&self, namespace: &str, path: &str) -> Arc<PathCacheState> {
+ let logical_path = LogicalPath {
+ namespace: namespace.to_string(),
+ path: path.to_string(),
+ };
+ let mut states = self
+ .path_states
+ .lock()
+ .unwrap_or_else(|error| error.into_inner());
+ if let Some(state) = states.get(&logical_path).and_then(Weak::upgrade) {
+ return state;
+ }
+ if states.len() >= 1024 {
+ states.retain(|_, state| state.strong_count() > 0);
+ }
+ let state = Arc::new(PathCacheState {
+ generation: std::sync::atomic::AtomicU64::new(0),
+ publish_gate: tokio::sync::RwLock::new(()),
+ });
+ states.insert(logical_path, Arc::downgrade(&state));
+ state
+ }
+
+ pub(super) async fn block_load_lock(&self, key: &BlockKey) -> Arc<tokio::sync::Mutex<()>> {
+ let mut in_flight = self.in_flight.lock().await;
+ if let Some(lock) = in_flight.get(key).and_then(Weak::upgrade) {
+ return lock;
+ }
+ if in_flight.len() >= 1024 {
+ in_flight.retain(|_, lock| lock.strong_count() > 0);
+ }
+ let lock = Arc::new(tokio::sync::Mutex::new(()));
+ in_flight.insert(key.clone(), Arc::downgrade(&lock));
+ lock
+ }
+
+ pub(super) async fn release_block_load_lock(
+ &self,
+ key: &BlockKey,
+ lock: &Arc<tokio::sync::Mutex<()>>,
+ ) {
+ let mut in_flight = self.in_flight.lock().await;
+ if Arc::strong_count(lock) == 1
+ && in_flight
+ .get(key)
+ .and_then(Weak::upgrade)
+ .is_some_and(|current| Arc::ptr_eq(¤t, lock))
+ {
+ in_flight.remove(key);
+ }
+ }
+
+ pub(super) fn file_size(&self, namespace: &str, path: &str) -> Option<u64> {
+ let logical_path = LogicalPath {
+ namespace: namespace.to_string(),
+ path: path.to_string(),
+ };
+ let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
+ let size = state.file_sizes.shift_remove(&logical_path)?;
+ state.file_sizes.insert(logical_path, size);
+ Some(size)
+ }
+
+ pub(super) fn put_file_size(&self, namespace: &str, path: &str, size: u64, capacity: usize) {
+ let logical_path = LogicalPath {
+ namespace: namespace.to_string(),
+ path: path.to_string(),
+ };
+ let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
+ state.file_sizes.shift_remove(&logical_path);
+ state.file_sizes.insert(logical_path, size);
+ while state.file_sizes.len() > capacity {
+ state.file_sizes.shift_remove_index(0);
+ }
+ }
+
+ pub(super) async fn get_block(&self, key: &BlockKey) -> Option<Bytes> {
+ self.ensure_recovered().await;
+ if !self.is_active(key) {
+ return None;
+ }
+ let path = self.root.join(key.cache_relative_path());
+ let encoded = match tokio::fs::read(&path).await {
+ Ok(encoded) => encoded,
+ Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+ self.forget_entry(key);
+ return None;
+ }
+ Err(error) => {
+ log::debug!(
+ "Failed to read local cache block '{}': {error}",
+ path.display()
+ );
+ return None;
+ }
+ };
+ match decode_block(key, &encoded) {
+ Ok(payload) if self.touch_entry(key) => Some(payload),
+ Ok(_) => None,
+ Err(error) => {
+ log::debug!(
+ "Discarding invalid local cache block '{}': {error}",
+ path.display()
+ );
+ self.forget_entry(key);
+ let _ = tokio::fs::remove_file(path).await;
+ None
+ }
+ }
+ }
+
+ pub(super) async fn put_block(&self, key: &BlockKey, payload: Bytes) {
+ self.ensure_recovered().await;
+ let encoded = encode_block(key, &payload);
+ let encoded_size = encoded.len() as u64;
+ if self
+ .state
+ .lock()
+ .unwrap_or_else(|error| error.into_inner())
+ .max_size
+ .is_some_and(|max_size| encoded_size > max_size)
+ {
+ return;
+ }
+
+ let path = self.root.join(key.cache_relative_path());
+ let Some(parent) = path.parent() else {
+ return;
+ };
+ if let Err(error) = tokio::fs::create_dir_all(parent).await {
+ log::debug!(
+ "Failed to create local cache shard '{}': {error}",
+ parent.display()
+ );
+ return;
+ }
+
+ let file_name = path
+ .file_name()
+ .map(|name| name.to_string_lossy())
+ .unwrap_or_default();
+ let temporary = parent.join(format!(".{file_name}.tmp.{}", uuid::Uuid::new_v4()));
+ let mut options = tokio::fs::OpenOptions::new();
+ options.write(true).create_new(true);
+ #[cfg(unix)]
+ {
+ options.mode(0o600);
+ }
+ let mut temporary_file = match options.open(&temporary).await {
+ Ok(file) => file,
+ Err(error) => {
+ log::debug!(
+ "Failed to create local cache temporary block '{}': {error}",
+ temporary.display()
+ );
+ return;
+ }
+ };
+ if let Err(error) = temporary_file.write_all(&encoded).await {
+ log::debug!(
+ "Failed to write local cache temporary block '{}': {error}",
+ temporary.display()
+ );
+ drop(temporary_file);
+ let _ = tokio::fs::remove_file(&temporary).await;
+ return;
+ }
+ drop(temporary_file);
+ if let Err(error) = tokio::fs::rename(&temporary, &path).await {
+ log::debug!(
+ "Failed to publish local cache block '{}': {error}",
+ path.display()
+ );
+ let _ = tokio::fs::remove_file(temporary).await;
+ return;
+ }
+ self.record_entry_and_evict(key.clone(), encoded_size).await;
+ }
+
+ pub(super) async fn invalidate_path(&self, namespace: &str, path: &str) {
+ self.ensure_recovered().await;
+ let _prefix_guard = self.prefix_barrier.read().await;
+ let path_state = self.path_state(namespace, path);
+ let _publish_guard = path_state.publish_gate.write().await;
+ path_state
+ .generation
+ .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+ self.invalidate_matching(|logical_path| {
+ logical_path.namespace == namespace && logical_path.path == path
+ })
+ .await;
+ }
+
+ pub(super) async fn remove_block(&self, key: &BlockKey) {
+ self.ensure_recovered().await;
+ self.forget_entry(key);
+ let path = self.root.join(key.cache_relative_path());
+ if let Err(error) = tokio::fs::remove_file(&path).await {
+ if error.kind() != std::io::ErrorKind::NotFound {
+ log::debug!(
+ "Failed to remove local cache block '{}': {error}",
+ path.display()
+ );
+ }
+ }
+ }
+
+ pub(super) async fn invalidate_prefix(&self, namespace: &str, prefix: &str) {
+ self.ensure_recovered().await;
+ let _prefix_guard = self.prefix_barrier.write().await;
+ let prefix = prefix.trim_end_matches('/');
+ let mut states = {
+ let states = self
+ .path_states
+ .lock()
+ .unwrap_or_else(|error| error.into_inner());
+ states
+ .iter()
+ .filter(|(path, _)| logical_path_matches_prefix(path, namespace, prefix))
+ .filter_map(|(path, state)| Weak::upgrade(state).map(|state| (path.clone(), state)))
+ .collect::<Vec<_>>()
+ };
+ states.sort_unstable_by(|left, right| left.0.cmp(&right.0));
+ let mut publish_guards = Vec::with_capacity(states.len());
+ for (_, state) in &states {
+ publish_guards.push(state.publish_gate.write().await);
+ state
+ .generation
+ .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+ }
+ self.invalidate_matching(|path| logical_path_matches_prefix(path, namespace, prefix))
+ .await;
+ drop(publish_guards);
+ }
+
+ async fn invalidate_matching(&self, matches: impl Fn(&LogicalPath) -> bool) {
+ let keys = {
+ let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
+ let keys = state
+ .paths
+ .keys()
+ .filter(|path| matches(path))
+ .filter_map(|path| state.paths.get(path))
+ .flat_map(|keys| keys.iter().cloned())
+ .collect::<Vec<_>>();
+ for key in &keys {
+ remove_state_entry(&mut state, key);
+ }
+ state.file_sizes.retain(|path, _| !matches(path));
+ keys
+ };
+ for key in keys {
+ let cache_path = self.root.join(key.cache_relative_path());
+ if let Err(error) = tokio::fs::remove_file(&cache_path).await {
+ if error.kind() != std::io::ErrorKind::NotFound {
+ log::debug!(
+ "Failed to invalidate local cache block '{}': {error}",
+ cache_path.display()
+ );
+ }
+ }
+ }
+ }
+
+ async fn record_entry_and_evict(&self, key: BlockKey, encoded_size: u64) {
+ let to_evict = {
+ let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
+ insert_state_entry(&mut state, key, encoded_size);
+ collect_evictions(&mut state)
+ };
+ self.remove_cache_files(to_evict).await;
+ }
+
+ async fn evict_over_limit(&self) {
+ let to_evict = {
+ let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
+ collect_evictions(&mut state)
+ };
+ self.remove_cache_files(to_evict).await;
+ }
+
+ async fn remove_cache_files(&self, keys: Vec<BlockKey>) {
+ for key in keys {
+ let path = self.root.join(key.cache_relative_path());
+ if let Err(error) = tokio::fs::remove_file(&path).await {
+ if error.kind() != std::io::ErrorKind::NotFound {
+ log::debug!(
+ "Failed to evict local cache block '{}': {error}",
+ path.display()
+ );
+ }
+ }
+ }
+ }
+
+ fn is_active(&self, key: &BlockKey) -> bool {
+ self.state
+ .lock()
+ .unwrap_or_else(|error| error.into_inner())
+ .entries
+ .contains_key(key)
+ }
+
+ fn touch_entry(&self, key: &BlockKey) -> bool {
+ let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
+ let Some(encoded_size) = state.entries.shift_remove(key) else {
+ return false;
+ };
+ state.entries.insert(key.clone(), encoded_size);
+ true
+ }
+
+ fn forget_entry(&self, key: &BlockKey) {
+ let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
+ remove_state_entry(&mut state, key);
+ }
+}
+
+fn disk_cache_registry() -> &'static Mutex<HashMap<PathBuf, Weak<DiskCache>>> {
+ static REGISTRY: OnceLock<Mutex<HashMap<PathBuf, Weak<DiskCache>>>> = OnceLock::new();
+ REGISTRY.get_or_init(|| Mutex::new(HashMap::new()))
+}
+
+fn prepare_cache_root(root: &Path) -> crate::Result<PathBuf> {
+ initialize_cache_directory(root)?;
+ std::fs::canonicalize(root).map_err(|error| crate::Error::ConfigInvalid {
+ message: format!(
+ "Failed to resolve local cache directory '{}': {error}",
+ root.display()
+ ),
+ })
+}
+
+fn collect_evictions(state: &mut CacheState) -> Vec<BlockKey> {
+ let mut to_evict = Vec::new();
+ if let Some(max_size) = state.max_size {
+ while state.current_size > max_size {
+ let Some((eldest, size)) = state.entries.shift_remove_index(0) else {
+ break;
+ };
+ state.current_size = state.current_size.saturating_sub(size);
+ remove_path_index_entry(state, &eldest);
+ to_evict.push(eldest);
+ }
+ }
+ to_evict
+}
+
+fn logical_path(key: &BlockKey) -> LogicalPath {
+ LogicalPath {
+ namespace: key.namespace.clone(),
+ path: key.path.clone(),
+ }
+}
+
+fn logical_path_matches_prefix(path: &LogicalPath, namespace: &str, prefix: &str) -> bool {
+ path.namespace == namespace
+ && (path.path == prefix
+ || path
+ .path
+ .strip_prefix(prefix)
+ .is_some_and(|suffix| suffix.starts_with('/')))
+}
+
+fn insert_state_entry(state: &mut CacheState, key: BlockKey, encoded_size: u64) {
+ if let Some(previous_size) = state.entries.shift_remove(&key) {
+ state.current_size = state.current_size.saturating_sub(previous_size);
+ }
+ state
+ .paths
+ .entry(logical_path(&key))
+ .or_default()
+ .insert(key.clone());
+ state.entries.insert(key, encoded_size);
+ state.current_size = state.current_size.saturating_add(encoded_size);
+}
+
+fn remove_state_entry(state: &mut CacheState, key: &BlockKey) {
+ if let Some(encoded_size) = state.entries.shift_remove(key) {
+ state.current_size = state.current_size.saturating_sub(encoded_size);
+ remove_path_index_entry(state, key);
+ }
+}
+
+fn remove_path_index_entry(state: &mut CacheState, key: &BlockKey) {
+ let path = logical_path(key);
+ if let Some(keys) = state.paths.get_mut(&path) {
+ keys.remove(key);
+ if keys.is_empty() {
+ state.paths.remove(&path);
+ }
+ }
+}
+
+fn initialize_cache_directory(root: &Path) -> crate::Result<()> {
+ if let Ok(metadata) = std::fs::symlink_metadata(root) {
+ if metadata.file_type().is_symlink() || !metadata.is_dir() {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Local cache path '{}' must be a directory and not a symlink",
+ root.display()
+ ),
+ });
+ }
+ } else {
+ let mut builder = std::fs::DirBuilder::new();
+ builder.recursive(true);
+ #[cfg(unix)]
+ {
+ use std::os::unix::fs::DirBuilderExt;
+ builder.mode(0o700);
+ }
+ builder
+ .create(root)
+ .map_err(|error| crate::Error::ConfigInvalid {
+ message: format!(
+ "Failed to initialize local cache directory '{}': {error}",
+ root.display()
+ ),
+ })?;
+ }
+ #[cfg(unix)]
+ {
+ use std::os::unix::fs::PermissionsExt;
+ std::fs::set_permissions(root, std::fs::Permissions::from_mode(0o700)).map_err(
+ |error| crate::Error::ConfigInvalid {
+ message: format!(
+ "Failed to secure local cache directory '{}': {error}",
+ root.display()
+ ),
+ },
+ )?;
+ }
+ Ok(())
+}
+
+fn cleanup_temporary_files(directory: &Path) -> std::io::Result<()> {
+ for shard in std::fs::read_dir(directory)? {
+ let shard = shard?;
+ let shard_name = shard.file_name();
+ let shard_name = shard_name.to_string_lossy();
+ if !shard.file_type()?.is_dir() || !is_lower_hex(&shard_name, 2) {
+ continue;
+ }
+ for entry in std::fs::read_dir(shard.path())? {
+ let entry = entry?;
+ let name = entry.file_name();
+ let name = name.to_string_lossy();
+ if entry.file_type()?.is_file() && is_cache_temporary_name(&name, &shard_name) {
+ std::fs::remove_file(entry.path())?;
+ }
+ }
+ }
+ Ok(())
+}
+
+fn scan_existing_blocks(root: &Path) -> CacheState {
+ let mut discovered = Vec::new();
+ collect_cache_files(root, &mut discovered);
+ discovered.sort_by_key(|(modified, _, _, _)| *modified);
+
+ let mut state = CacheState::default();
+ for (_, path, key, encoded_size) in discovered {
+ if root.join(key.cache_relative_path()) != path {
+ let _ = std::fs::remove_file(path);
+ continue;
+ }
+ insert_state_entry(&mut state, key, encoded_size);
+ }
+
+ state
+}
+
+fn collect_cache_files(
+ directory: &Path,
+ discovered: &mut Vec<(std::time::SystemTime, PathBuf, BlockKey, u64)>,
+) {
+ let shards = match std::fs::read_dir(directory) {
+ Ok(entries) => entries,
+ Err(error) => {
+ log::debug!(
+ "Failed to scan local cache directory '{}': {error}",
+ directory.display()
+ );
+ return;
+ }
+ };
+ for shard in shards.flatten() {
+ let shard_name = shard.file_name();
+ let shard_name = shard_name.to_string_lossy();
+ let file_type = match shard.file_type() {
+ Ok(file_type) => file_type,
+ Err(_) => continue,
+ };
+ if !file_type.is_dir() || !is_lower_hex(&shard_name, 2) {
+ continue;
+ }
+ let entries = match std::fs::read_dir(shard.path()) {
+ Ok(entries) => entries,
+ Err(_) => continue,
+ };
+ for entry in entries.flatten() {
+ let name = entry.file_name();
+ let name = name.to_string_lossy();
+ if !entry.file_type().is_ok_and(|file_type| file_type.is_file())
+ || !is_cache_block_name(&name, &shard_name)
+ {
+ continue;
+ }
+ collect_cache_file(&entry, discovered);
+ }
+ }
+}
+
+fn collect_cache_file(
+ entry: &std::fs::DirEntry,
+ discovered: &mut Vec<(std::time::SystemTime, PathBuf, BlockKey, u64)>,
+) {
+ let path = entry.path();
+ let metadata = match entry.metadata() {
+ Ok(metadata) => metadata,
+ Err(_) => return,
+ };
+ let key = match read_block_key_header(&path, metadata.len()) {
+ Ok(key) => key,
+ Err(error) => {
+ log::debug!(
+ "Discarding invalid local cache block '{}' during startup: {error}",
+ path.display()
+ );
+ let _ = std::fs::remove_file(path);
+ return;
+ }
+ };
+ discovered.push((
+ metadata
+ .modified()
+ .unwrap_or(std::time::SystemTime::UNIX_EPOCH),
+ path,
+ key,
+ metadata.len(),
+ ));
+}
+
+fn read_block_key_header(path: &Path, encoded_len: u64) -> Result<BlockKey, BlockDecodeError> {
+ let mut file =
+ std::fs::File::open(path).map_err(|_| BlockDecodeError("cache block cannot be opened"))?;
+ let mut fixed_header = [0_u8; FIXED_HEADER_LEN];
+ file.read_exact(&mut fixed_header)
+ .map_err(|_| BlockDecodeError("cache block header is truncated"))?;
+ let layout = decode_block_layout(&fixed_header, encoded_len)?;
+ let mut header = Vec::with_capacity(layout.header_len);
+ header.extend_from_slice(&fixed_header);
+ header.resize(layout.header_len, 0);
+ file.read_exact(&mut header[FIXED_HEADER_LEN..])
+ .map_err(|_| BlockDecodeError("cache block key header is truncated"))?;
+ decode_block_header(&header, encoded_len).map(|(key, _)| key)
+}
+
+fn is_lower_hex(value: &str, expected_len: usize) -> bool {
+ value.len() == expected_len
+ && value
+ .bytes()
+ .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
+}
+
+fn is_cache_block_name(name: &str, shard: &str) -> bool {
+ is_lower_hex(name, 64) && name.starts_with(shard)
+}
+
+fn is_cache_temporary_name(name: &str, shard: &str) -> bool {
+ let Some(name) = name.strip_prefix('.') else {
+ return false;
+ };
+ let Some((digest, suffix)) = name.split_once(".tmp.") else {
+ return false;
+ };
+ is_cache_block_name(digest, shard) && uuid::Uuid::parse_str(suffix).is_ok()
+}
+
+fn encode_block(key: &BlockKey, payload: &Bytes) -> Vec<u8> {
+ let namespace = key.namespace.as_bytes();
+ let path = key.path.as_bytes();
+ let mut encoded = Vec::with_capacity(
+ FIXED_HEADER_LEN + namespace.len() + path.len() + payload.len() + CHECKSUM_LEN,
+ );
+ encoded.extend_from_slice(CACHE_MAGIC);
+ encoded.push(CACHE_FORMAT_VERSION);
+ encoded.extend_from_slice(&(namespace.len() as u32).to_le_bytes());
+ encoded.extend_from_slice(&(path.len() as u32).to_le_bytes());
+ encoded.extend_from_slice(&key.block_size.to_le_bytes());
+ encoded.extend_from_slice(&key.block_index.to_le_bytes());
+ encoded.extend_from_slice(&(payload.len() as u64).to_le_bytes());
+ encoded.extend_from_slice(namespace);
+ encoded.extend_from_slice(path);
+ encoded.extend_from_slice(payload);
+ let checksum = crc32fast::hash(&encoded);
+ encoded.extend_from_slice(&checksum.to_le_bytes());
+ encoded
+}
+
+fn decode_block(key: &BlockKey, encoded: &[u8]) -> Result<Bytes, BlockDecodeError> {
+ let (decoded_key, payload) = decode_block_any(encoded)?;
+ if &decoded_key != key {
+ return Err(BlockDecodeError("cache block key does not match"));
+ }
+ Ok(payload)
+}
+
+fn decode_block_any(encoded: &[u8]) -> Result<(BlockKey, Bytes), BlockDecodeError> {
+ let (key, layout) = decode_block_header(encoded, encoded.len() as u64)?;
+ let payload_len = usize::try_from(layout.payload_len)
+ .map_err(|_| BlockDecodeError("cache block payload is too large"))?;
+ let payload_end = layout
+ .header_len
+ .checked_add(payload_len)
+ .ok_or(BlockDecodeError("cache block length overflows"))?;
+ let expected_checksum = u32::from_le_bytes(
+ encoded[payload_end..payload_end + CHECKSUM_LEN]
+ .try_into()
+ .unwrap(),
+ );
+ if crc32fast::hash(&encoded[..payload_end]) != expected_checksum {
+ return Err(BlockDecodeError("cache block checksum does not match"));
+ }
+
+ Ok((
+ key,
+ Bytes::copy_from_slice(&encoded[layout.header_len..payload_end]),
+ ))
+}
+
+#[derive(Clone, Copy, Debug)]
+struct BlockLayout {
+ namespace_len: usize,
+ path_len: usize,
+ block_size: u64,
+ block_index: u64,
+ payload_len: u64,
+ header_len: usize,
+}
+
+fn decode_block_layout(
+ fixed_header: &[u8],
+ encoded_len: u64,
+) -> Result<BlockLayout, BlockDecodeError> {
+ if fixed_header.len() < FIXED_HEADER_LEN {
+ return Err(BlockDecodeError("cache block header is truncated"));
+ }
+ if &fixed_header[..CACHE_MAGIC.len()] != CACHE_MAGIC {
+ return Err(BlockDecodeError("cache block magic does not match"));
+ }
+ if fixed_header[CACHE_MAGIC.len()] != CACHE_FORMAT_VERSION {
+ return Err(BlockDecodeError("cache block version is unsupported"));
+ }
+
+ let mut offset = CACHE_MAGIC.len() + 1;
+ let namespace_len = read_u32(fixed_header, &mut offset)? as usize;
+ let path_len = read_u32(fixed_header, &mut offset)? as usize;
+ let block_size = read_u64(fixed_header, &mut offset)?;
+ let block_index = read_u64(fixed_header, &mut offset)?;
+ let payload_len = read_u64(fixed_header, &mut offset)?;
+ let key_header_len = namespace_len
+ .checked_add(path_len)
+ .ok_or(BlockDecodeError("cache block key header length overflows"))?;
+ if key_header_len > MAX_CACHE_KEY_HEADER_LEN {
+ return Err(BlockDecodeError("cache block key header is too large"));
+ }
+ let header_len = offset
+ .checked_add(namespace_len)
+ .and_then(|length| length.checked_add(path_len))
+ .ok_or(BlockDecodeError("cache block length overflows"))?;
+ let expected_len = u64::try_from(header_len)
+ .ok()
+ .and_then(|length| length.checked_add(payload_len))
+ .and_then(|length| length.checked_add(CHECKSUM_LEN as u64))
+ .ok_or(BlockDecodeError("cache block length overflows"))?;
+ if expected_len != encoded_len {
+ return Err(BlockDecodeError("cache block length does not match"));
+ }
+ if block_size == 0 {
+ return Err(BlockDecodeError("cache block size is zero"));
+ }
+
+ Ok(BlockLayout {
+ namespace_len,
+ path_len,
+ block_size,
+ block_index,
+ payload_len,
+ header_len,
+ })
+}
+
+fn decode_block_header(
+ header: &[u8],
+ encoded_len: u64,
+) -> Result<(BlockKey, BlockLayout), BlockDecodeError> {
+ let layout = decode_block_layout(header, encoded_len)?;
+ if header.len() < layout.header_len {
+ return Err(BlockDecodeError("cache block key header is truncated"));
+ }
+
+ let namespace_start = FIXED_HEADER_LEN;
+ let namespace_end = namespace_start + layout.namespace_len;
+ let namespace = std::str::from_utf8(&header[namespace_start..namespace_end])
+ .map_err(|_| BlockDecodeError("cache block namespace is not UTF-8"))?;
+ let path_end = namespace_end + layout.path_len;
+ let path = std::str::from_utf8(&header[namespace_end..path_end])
+ .map_err(|_| BlockDecodeError("cache block path is not UTF-8"))?;
+
+ Ok((
+ BlockKey::with_namespace(namespace, path, layout.block_size, layout.block_index),
+ layout,
+ ))
+}
+
+fn read_u32(encoded: &[u8], offset: &mut usize) -> Result<u32, BlockDecodeError> {
+ let end = offset
+ .checked_add(4)
+ .ok_or(BlockDecodeError("cache block offset overflows"))?;
+ let value = encoded
+ .get(*offset..end)
+ .ok_or(BlockDecodeError("cache block header is truncated"))?;
+ *offset = end;
+ Ok(u32::from_le_bytes(value.try_into().unwrap()))
+}
+
+fn read_u64(encoded: &[u8], offset: &mut usize) -> Result<u64, BlockDecodeError> {
+ let end = offset
+ .checked_add(8)
+ .ok_or(BlockDecodeError("cache block offset overflows"))?;
+ let value = encoded
+ .get(*offset..end)
+ .ok_or(BlockDecodeError("cache block header is truncated"))?;
+ *offset = end;
+ Ok(u64::from_le_bytes(value.try_into().unwrap()))
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use std::sync::Arc;
+
+ #[test]
+ fn test_disk_block_codec_round_trip() {
+ let key = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 1024, 3);
+ let payload = Bytes::from_static(b"cached block");
+
+ let encoded = encode_block(&key, &payload);
+ let decoded = decode_block(&key, &encoded).unwrap();
+
+ assert_eq!(decoded, payload);
+ }
+
+ #[test]
+ fn test_disk_block_codec_rejects_crc_corruption() {
+ let key = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 1024, 3);
+ let mut encoded = encode_block(&key, &Bytes::from_static(b"cached block"));
+ let last_payload_byte = encoded.len() - CHECKSUM_LEN - 1;
+ encoded[last_payload_byte] ^= 0xff;
+
+ assert!(decode_block(&key, &encoded).is_err());
+ }
+
+ #[test]
+ fn test_disk_block_key_uses_sharded_sha256_path() {
+ let key = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 1024, 3);
+ let relative = key.cache_relative_path();
+ let components = relative
+ .iter()
+ .map(|component| component.to_string_lossy().into_owned())
+ .collect::<Vec<_>>();
+
+ assert_eq!(components.len(), 2);
+ assert_eq!(components[0].len(), 2);
+ assert_eq!(components[1].len(), 64);
+ assert!(components[1].starts_with(&components[0]));
+ assert_ne!(
+ relative,
+ BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 1024, 4).cache_relative_path()
+ );
+ }
+
+ #[tokio::test]
+ async fn test_disk_cache_persistence_survives_restart() {
+ let directory = tempfile::tempdir().unwrap();
+ let key = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 1024, 0);
+ let payload = Bytes::from_static(b"persistent block");
+
+ let cache = DiskCache::new(directory.path(), None).unwrap();
+ cache.put_block(&key, payload.clone()).await;
+ assert_eq!(cache.get_block(&key).await, Some(payload.clone()));
+ drop(cache);
+
+ let restarted = DiskCache::new(directory.path(), None).unwrap();
+ assert_eq!(restarted.get_block(&key).await, Some(payload));
+ }
+
+ #[tokio::test]
+ async fn test_disk_cache_shared_instances_enforce_one_size_limit() {
+ let directory = tempfile::tempdir().unwrap();
+ let first_key = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 4, 0);
+ let second_key = BlockKey::new("s3://bucket/table/snapshot/snapshot-2", 4, 0);
+ let payload = Bytes::from_static(b"data");
+ let one_block_size = encode_block(&first_key, &payload).len() as u64;
+
+ let first = DiskCache::shared(directory.path(), None).unwrap();
+ let second = DiskCache::shared(directory.path().join("."), Some(one_block_size)).unwrap();
+
+ assert!(Arc::ptr_eq(&first, &second));
+ first.put_block(&first_key, payload.clone()).await;
+ second.put_block(&second_key, payload.clone()).await;
+ assert_eq!(first.get_block(&first_key).await, None);
+ assert_eq!(second.get_block(&second_key).await, Some(payload));
+ }
+
+ #[tokio::test]
+ async fn test_disk_cache_restart_defers_crc_validation_until_first_hit() {
+ let directory = tempfile::tempdir().unwrap();
+ let key = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 4, 0);
+ let cache = DiskCache::new(directory.path(), None).unwrap();
+ cache.put_block(&key, Bytes::from_static(b"data")).await;
+ drop(cache);
+
+ let block_path = directory.path().join(key.cache_relative_path());
+ let mut encoded = std::fs::read(&block_path).unwrap();
+ let payload_offset = encoded.len() - CHECKSUM_LEN - 1;
+ encoded[payload_offset] ^= 0xff;
+ std::fs::write(&block_path, encoded).unwrap();
+
+ let restarted = DiskCache::new(directory.path(), None).unwrap();
+ restarted.ensure_recovered().await;
+ assert!(restarted.is_active(&key));
+
+ assert_eq!(restarted.get_block(&key).await, None);
+ assert!(!restarted.is_active(&key));
+ assert!(!block_path.exists());
+ }
+
+ #[tokio::test]
+ async fn test_disk_cache_does_not_read_unindexed_block_file() {
+ let directory = tempfile::tempdir().unwrap();
+ let key = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 4, 0);
+ let cache = DiskCache::new(directory.path(), None).unwrap();
+ cache.ensure_recovered().await;
+ let block_path = directory.path().join(key.cache_relative_path());
+ std::fs::create_dir_all(block_path.parent().unwrap()).unwrap();
+ std::fs::write(
+ &block_path,
+ encode_block(&key, &Bytes::from_static(b"data")),
+ )
+ .unwrap();
+
+ assert_eq!(cache.get_block(&key).await, None);
+ assert!(block_path.exists());
+ }
+
+ #[cfg(unix)]
+ #[tokio::test]
+ async fn test_disk_cache_uses_owner_only_permissions() {
+ use std::os::unix::fs::PermissionsExt;
+
+ let directory = tempfile::tempdir().unwrap();
+ let key = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 4, 0);
+ let cache = DiskCache::new(directory.path(), None).unwrap();
+ cache.put_block(&key, Bytes::from_static(b"data")).await;
+
+ assert_eq!(
+ std::fs::metadata(directory.path())
+ .unwrap()
+ .permissions()
+ .mode()
+ & 0o777,
+ 0o700
+ );
+ assert_eq!(
+ std::fs::metadata(directory.path().join(key.cache_relative_path()))
+ .unwrap()
+ .permissions()
+ .mode()
+ & 0o777,
+ 0o600
+ );
+ }
+
+ #[cfg(unix)]
+ #[test]
+ fn test_disk_cache_rejects_symlink_root() {
+ let directory = tempfile::tempdir().unwrap();
+ let target = tempfile::tempdir().unwrap();
+ let symlink = directory.path().join("cache-link");
+ std::os::unix::fs::symlink(target.path(), &symlink).unwrap();
+
+ assert!(DiskCache::new(&symlink, None).is_err());
+ }
+
+ #[tokio::test]
+ async fn test_disk_cache_persistence_removes_temporary_files_on_restart() {
+ let directory = tempfile::tempdir().unwrap();
+ let key = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 4, 0);
+ let relative = key.cache_relative_path();
+ let shard = directory.path().join(relative.parent().unwrap());
+ std::fs::create_dir_all(&shard).unwrap();
+ let digest = relative.file_name().unwrap().to_string_lossy();
+ let temporary = shard.join(format!(".{digest}.tmp.{}", uuid::Uuid::nil()));
+ std::fs::write(&temporary, b"partial").unwrap();
+
+ let cache = DiskCache::new(directory.path(), None).unwrap();
+ cache.ensure_recovered().await;
+
+ assert!(!temporary.exists());
+ }
+
+ #[tokio::test]
+ async fn test_disk_cache_lru_evicts_oldest_block_over_max_size() {
+ let directory = tempfile::tempdir().unwrap();
+ let first = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 4, 0);
+ let second = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 4, 1);
+ let payload = Bytes::from_static(b"data");
+ let one_block_size = encode_block(&first, &payload).len() as u64;
+ let cache = DiskCache::new(directory.path(), Some(one_block_size)).unwrap();
+
+ cache.put_block(&first, payload.clone()).await;
+ cache.put_block(&second, payload.clone()).await;
+
+ assert_eq!(cache.get_block(&first).await, None);
+ assert_eq!(cache.get_block(&second).await, Some(payload));
+ }
+
+ #[tokio::test]
+ async fn test_disk_cache_lru_hit_refreshes_access_order() {
+ let directory = tempfile::tempdir().unwrap();
+ let first = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 4, 0);
+ let second = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 4, 1);
+ let third = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 4, 2);
+ let payload = Bytes::from_static(b"data");
+ let two_blocks = 2 * encode_block(&first, &payload).len() as u64;
+ let cache = DiskCache::new(directory.path(), Some(two_blocks)).unwrap();
+
+ cache.put_block(&first, payload.clone()).await;
+ cache.put_block(&second, payload.clone()).await;
+ assert_eq!(cache.get_block(&first).await, Some(payload.clone()));
+ cache.put_block(&third, payload.clone()).await;
+
+ assert_eq!(cache.get_block(&second).await, None);
+ assert_eq!(cache.get_block(&first).await, Some(payload.clone()));
+ assert_eq!(cache.get_block(&third).await, Some(payload));
+ }
+
+ #[tokio::test]
+ async fn test_disk_cache_lru_evicts_over_limit_during_restart() {
+ let directory = tempfile::tempdir().unwrap();
+ let first = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 4, 0);
+ let second = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 4, 1);
+ let payload = Bytes::from_static(b"data");
+ let one_block_size = encode_block(&first, &payload).len() as u64;
+ let cache = DiskCache::new(directory.path(), None).unwrap();
+ cache.put_block(&first, payload.clone()).await;
+ cache.put_block(&second, payload).await;
+ drop(cache);
+
+ let restarted = DiskCache::new(directory.path(), Some(one_block_size)).unwrap();
+ restarted.ensure_recovered().await;
+
+ let block_count = std::fs::read_dir(directory.path())
+ .unwrap()
+ .map(|shard| std::fs::read_dir(shard.unwrap().path()).unwrap().count())
+ .sum::<usize>();
+ assert_eq!(block_count, 1);
+ }
+
+ #[tokio::test]
+ async fn test_disk_cache_lru_invalidates_all_blocks_for_path() {
+ let directory = tempfile::tempdir().unwrap();
+ let path = "s3://bucket/table/snapshot/snapshot-1";
+ let first = BlockKey::new(path, 4, 0);
+ let second = BlockKey::new(path, 4, 1);
+ let other = BlockKey::new("s3://bucket/table/snapshot/snapshot-2", 4, 0);
+ let payload = Bytes::from_static(b"data");
+ let cache = DiskCache::new(directory.path(), None).unwrap();
+ cache.put_block(&first, payload.clone()).await;
+ cache.put_block(&second, payload.clone()).await;
+ cache.put_block(&other, payload.clone()).await;
+
+ cache.invalidate_path("", path).await;
+
+ assert_eq!(cache.get_block(&first).await, None);
+ assert_eq!(cache.get_block(&second).await, None);
+ assert_eq!(cache.get_block(&other).await, Some(payload));
+ }
+
+ #[tokio::test]
+ async fn test_disk_cache_lru_invalidates_directory_prefix() {
+ let directory = tempfile::tempdir().unwrap();
+ let inside = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 4, 0);
+ let sibling = BlockKey::new("s3://bucket/table2/snapshot/snapshot-1", 4, 0);
+ let payload = Bytes::from_static(b"data");
+ let cache = DiskCache::new(directory.path(), None).unwrap();
+ cache.put_block(&inside, payload.clone()).await;
+ cache.put_block(&sibling, payload.clone()).await;
+
+ cache.invalidate_prefix("", "s3://bucket/table").await;
+
+ assert_eq!(cache.get_block(&inside).await, None);
+ assert_eq!(cache.get_block(&sibling).await, Some(payload));
+ }
+}
diff --git a/crates/paimon/src/io/cache/file_type.rs b/crates/paimon/src/io/cache/file_type.rs
new file mode 100644
index 00000000..2db0ae5a
--- /dev/null
+++ b/crates/paimon/src/io/cache/file_type.rs
@@ -0,0 +1,165 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashSet;
+
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+pub(super) enum FileType {
+ Meta,
+ Data,
+ BucketIndex,
+ GlobalIndex,
+ FileIndex,
+}
+
+impl FileType {
+ pub(super) fn classify(path: &str) -> Self {
+ let name = path.rsplit('/').next().unwrap_or(path);
+
+ if name.starts_with("snapshot-")
+ || name.starts_with("schema-")
+ || name.starts_with("stat-")
+ || name.starts_with("tag-")
+ || name.starts_with("consumer-")
+ || name.starts_with("service-")
+ || name.contains("manifest")
+ {
+ return Self::Meta;
+ }
+
+ if name.ends_with(".index") {
+ return if name.contains("global-index-") {
+ Self::GlobalIndex
+ } else {
+ Self::FileIndex
+ };
+ }
+
+ if name.starts_with("index-") {
+ return Self::BucketIndex;
+ }
+
+ Self::Data
+ }
+
+ pub(super) fn is_mutable(path: &str) -> bool {
+ let name = path.rsplit('/').next().unwrap_or(path);
+ matches!(name, "LATEST" | "EARLIEST")
+ || name.starts_with("tag-")
+ || name.starts_with("consumer-")
+ || name.starts_with("service-")
+ || name.ends_with(".tmp")
+ || name.contains(".tmp-")
+ || name.contains(".tmp.")
+ }
+
+ pub(super) fn parse_whitelist(value: &str) -> HashSet<Self> {
+ value
+ .split(',')
+ .filter_map(|name| match name.trim() {
+ "meta" => Some(Self::Meta),
+ "global-index" => Some(Self::GlobalIndex),
+ "bucket-index" => Some(Self::BucketIndex),
+ "data" => Some(Self::Data),
+ "file-index" => Some(Self::FileIndex),
+ "" => None,
+ unknown => {
+ log::warn!(
+ "Unknown local-cache.whitelist value '{}'; supported values are \
+ meta, global-index, bucket-index, data, file-index",
+ unknown
+ );
+ None
+ }
+ })
+ .collect()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_file_type_classifies_paimon_paths() {
+ let cases = [
+ ("s3://bucket/table/snapshot/snapshot-42", FileType::Meta),
+ ("s3://bucket/table/schema/schema-3", FileType::Meta),
+ ("s3://bucket/table/manifest/stat-1", FileType::Meta),
+ (
+ "s3://bucket/table/manifest/manifest-list-abc-0",
+ FileType::Meta,
+ ),
+ (
+ "s3://bucket/table/index/btree-global-index-abc.index",
+ FileType::GlobalIndex,
+ ),
+ (
+ "s3://bucket/table/index/vector-ivf-global-index-abc.index",
+ FileType::GlobalIndex,
+ ),
+ ("s3://bucket/table/index/index-abc-0", FileType::BucketIndex),
+ (
+ "s3://bucket/table/data/data-abc.parquet.index",
+ FileType::FileIndex,
+ ),
+ (
+ "s3://bucket/table/bucket-0/data-abc.parquet",
+ FileType::Data,
+ ),
+ ];
+
+ for (path, expected) in cases {
+ assert_eq!(FileType::classify(path), expected, "{path}");
+ }
+ }
+
+ #[test]
+ fn test_file_type_bypasses_mutable_paths() {
+ for path in [
+ "s3://bucket/table/snapshot/LATEST",
+ "s3://bucket/table/snapshot/EARLIEST",
+ "s3://bucket/table/tag/tag-production",
+ "s3://bucket/table/consumer/consumer-job",
+ "s3://bucket/table/service/service-api",
+ "s3://bucket/table/snapshot/.snapshot-1.123e4567-e89b-12d3-a456-426614174000.tmp",
+ "s3://bucket/table/snapshot/snapshot-1.tmp-123e4567-e89b-12d3-a456-426614174000",
+ ] {
+ assert!(FileType::is_mutable(path), "{path}");
+ }
+ assert!(!FileType::is_mutable(
+ "s3://bucket/table/snapshot/snapshot-1"
+ ));
+ }
+
+ #[test]
+ fn test_file_type_parses_whitelist() {
+ let whitelist =
+ FileType::parse_whitelist(" meta,global-index, bucket-index,data,file-index,unknown ");
+
+ assert_eq!(whitelist.len(), 5);
+ for file_type in [
+ FileType::Meta,
+ FileType::GlobalIndex,
+ FileType::BucketIndex,
+ FileType::Data,
+ FileType::FileIndex,
+ ] {
+ assert!(whitelist.contains(&file_type));
+ }
+ }
+}
diff --git a/crates/paimon/src/io/cache/mod.rs b/crates/paimon/src/io/cache/mod.rs
new file mode 100644
index 00000000..66792c53
--- /dev/null
+++ b/crates/paimon/src/io/cache/mod.rs
@@ -0,0 +1,529 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod disk;
+mod file_type;
+mod reader;
+
+use self::file_type::FileType;
+use crate::common::{CatalogOptions, Options};
+use sha2::{Digest, Sha256};
+use std::collections::HashSet;
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use disk::{BlockKey, CacheReadToken, DiskCache};
+pub(super) use reader::CachedFileReader;
+
+const CACHE_DIRECTORY_NAME: &str = "paimon-local-cache-v2";
+const DEFAULT_FILE_SIZE_CAPACITY: usize = 65_536;
+
+#[derive(Debug)]
+pub(crate) struct LocalCache {
+ disk: Arc<DiskCache>,
+ namespace: String,
+ block_size: u64,
+ whitelist: HashSet<FileType>,
+ file_size_capacity: usize,
+}
+
+impl LocalCache {
+ pub(super) fn new(config: LocalCacheConfig) -> crate::Result<Self> {
+ let file_size_capacity = config
+ .max_size
+ .map(|max_size| max_size / config.block_size)
+ .and_then(|capacity| usize::try_from(capacity).ok())
+ .unwrap_or(DEFAULT_FILE_SIZE_CAPACITY)
+ .clamp(1, DEFAULT_FILE_SIZE_CAPACITY);
+ Ok(Self {
+ disk: DiskCache::shared(config.dir.join(CACHE_DIRECTORY_NAME), config.max_size)?,
+ namespace: config.namespace,
+ block_size: config.block_size,
+ whitelist: config.whitelist,
+ file_size_capacity,
+ })
+ }
+
+ fn block_size(&self) -> u64 {
+ self.block_size
+ }
+
+ fn block_key(&self, path: &str, block_index: u64) -> BlockKey {
+ BlockKey::with_namespace(&self.namespace, path, self.block_size, block_index)
+ }
+
+ pub(super) fn is_cacheable(&self, path: &str) -> bool {
+ !FileType::is_mutable(path) && self.whitelist.contains(&FileType::classify(path))
+ }
+
+ async fn get_block(&self, key: &BlockKey, token: &CacheReadToken) -> Option<bytes::Bytes> {
+ let _prefix_guard = self.disk.prefix_read_guard().await;
+ let _publish_guard = token.publish_guard().await;
+ if !token.is_current() {
+ return None;
+ }
+ let payload = self.disk.get_block(key).await;
+ if token.is_current() {
+ payload
+ } else {
+ None
+ }
+ }
+
+ async fn put_block(&self, key: &BlockKey, payload: bytes::Bytes, token: &CacheReadToken) {
+ let _prefix_guard = self.disk.prefix_read_guard().await;
+ let _publish_guard = token.publish_guard().await;
+ if !token.is_current() {
+ return;
+ }
+ self.disk.put_block(key, payload).await;
+ }
+
+ async fn remove_block(&self, key: &BlockKey) {
+ self.disk.remove_block(key).await;
+ }
+
+ pub(super) fn read_token(&self, path: &str) -> CacheReadToken {
+ self.disk.read_token(&self.namespace, path)
+ }
+
+ async fn block_load_lock(&self, key: &BlockKey) -> Arc<tokio::sync::Mutex<()>> {
+ self.disk.block_load_lock(key).await
+ }
+
+ async fn release_block_load_lock(&self, key: &BlockKey, lock: &Arc<tokio::sync::Mutex<()>>) {
+ self.disk.release_block_load_lock(key, lock).await;
+ }
+
+ pub(super) async fn file_size(&self, path: &str, token: &CacheReadToken) -> Option<u64> {
+ let _prefix_guard = self.disk.prefix_read_guard().await;
+ let _publish_guard = token.publish_guard().await;
+ token
+ .is_current()
+ .then(|| self.disk.file_size(&self.namespace, path))
+ .flatten()
+ }
+
+ pub(super) async fn put_file_size(&self, path: &str, size: u64, token: &CacheReadToken) {
+ let _prefix_guard = self.disk.prefix_read_guard().await;
+ let _publish_guard = token.publish_guard().await;
+ if !token.is_current() {
+ return;
+ }
+ self.disk
+ .put_file_size(&self.namespace, path, size, self.file_size_capacity);
+ }
+
+ pub(super) async fn invalidate_path(&self, path: &str) {
+ self.disk.invalidate_path(&self.namespace, path).await;
+ }
+
+ pub(super) async fn invalidate_prefix(&self, prefix: &str) {
+ self.disk.invalidate_prefix(&self.namespace, prefix).await;
+ }
+}
+
+pub(crate) fn create_local_cache(options: &Options) -> crate::Result<Option<Arc<LocalCache>>> {
+ create_local_cache_with_namespace(options, options)
+}
+
+pub(crate) fn create_local_cache_with_namespace(
+ cache_options: &Options,
+ namespace_options: &Options,
+) -> crate::Result<Option<Arc<LocalCache>>> {
+ LocalCacheConfig::from_options_with_namespace(cache_options, namespace_options)?
+ .map(LocalCache::new)
+ .transpose()
+ .map(|cache| cache.map(Arc::new))
+}
+
+#[derive(Debug)]
+pub(crate) struct LocalCacheConfig {
+ dir: PathBuf,
+ namespace: String,
+ max_size: Option<u64>,
+ block_size: u64,
+ whitelist: HashSet<FileType>,
+}
+
+impl LocalCacheConfig {
+ #[cfg(test)]
+ pub(crate) fn from_options(options: &Options) -> crate::Result<Option<Self>> {
+ Self::from_options_with_namespace(options, options)
+ }
+
+ fn from_options_with_namespace(
+ options: &Options,
+ namespace_options: &Options,
+ ) -> crate::Result<Option<Self>> {
+ let enabled = match options
+ .get(CatalogOptions::LOCAL_CACHE_ENABLED)
+ .map(|value| value.trim())
+ {
+ None => false,
+ Some(value) if value.eq_ignore_ascii_case("true") => true,
+ Some(value) if value.eq_ignore_ascii_case("false") => false,
+ Some(value) => {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Invalid boolean for {}: '{}'",
+ CatalogOptions::LOCAL_CACHE_ENABLED,
+ value
+ ),
+ });
+ }
+ };
+ if !enabled {
+ return Ok(None);
+ }
+
+ let dir = options
+ .get(CatalogOptions::LOCAL_CACHE_DIR)
+ .filter(|value| !value.trim().is_empty())
+ .ok_or_else(|| crate::Error::ConfigInvalid {
+ message: format!(
+ "Missing required option: {}",
+ CatalogOptions::LOCAL_CACHE_DIR
+ ),
+ })?
+ .into();
+
+ let max_size = options
+ .get(CatalogOptions::LOCAL_CACHE_MAX_SIZE)
+ .map(|value| parse_memory_size(CatalogOptions::LOCAL_CACHE_MAX_SIZE, value))
+ .transpose()?;
+ let block_size = options
+ .get(CatalogOptions::LOCAL_CACHE_BLOCK_SIZE)
+ .map(|value| parse_memory_size(CatalogOptions::LOCAL_CACHE_BLOCK_SIZE, value))
+ .transpose()?
+ .unwrap_or(1024 * 1024);
+ if block_size == 0 {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "{} must be greater than zero",
+ CatalogOptions::LOCAL_CACHE_BLOCK_SIZE
+ ),
+ });
+ }
+ let whitelist = options
+ .get(CatalogOptions::LOCAL_CACHE_WHITELIST)
+ .map(String::as_str)
+ .unwrap_or("meta,global-index");
+
+ Ok(Some(Self {
+ dir,
+ namespace: catalog_namespace(namespace_options),
+ max_size,
+ block_size,
+ whitelist: FileType::parse_whitelist(whitelist),
+ }))
+ }
+}
+
+fn catalog_namespace(options: &Options) -> String {
+ let mut entries = options
+ .to_map()
+ .iter()
+ .filter(|(key, _)| !key.starts_with("local-cache."))
+ .collect::<Vec<_>>();
+ entries.sort_unstable_by(|left, right| left.0.cmp(right.0));
+ let mut digest = Sha256::new();
+ for (key, value) in entries {
+ digest.update((key.len() as u64).to_le_bytes());
+ digest.update(key.as_bytes());
+ digest.update((value.len() as u64).to_le_bytes());
+ digest.update(value.as_bytes());
+ }
+ hex::encode(digest.finalize())
+}
+
+fn parse_memory_size(key: &str, value: &str) -> crate::Result<u64> {
+ let compact = value
+ .chars()
+ .filter(|character| !character.is_ascii_whitespace())
+ .collect::<String>()
+ .to_ascii_lowercase();
+ let unit_start = compact
+ .find(|character: char| !character.is_ascii_digit())
+ .unwrap_or(compact.len());
+ let (number, unit) = compact.split_at(unit_start);
+ let number = number
+ .parse::<u64>()
+ .map_err(|_| crate::Error::ConfigInvalid {
+ message: format!("Invalid memory size for {key}: '{value}'"),
+ })?;
+ let multiplier = match unit {
+ "" | "b" => 1,
+ "k" | "kb" | "kib" => 1024,
+ "m" | "mb" | "mib" => 1024 * 1024,
+ "g" | "gb" | "gib" => 1024 * 1024 * 1024,
+ "t" | "tb" | "tib" => 1024_u64.pow(4),
+ _ => {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!("Invalid memory size for {key}: '{value}'"),
+ });
+ }
+ };
+ number
+ .checked_mul(multiplier)
+ .ok_or_else(|| crate::Error::ConfigInvalid {
+ message: format!("Memory size for {key} is too large: '{value}'"),
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_local_cache_config_disabled_by_default() {
+ assert!(LocalCacheConfig::from_options(&Options::new())
+ .unwrap()
+ .is_none());
+ }
+
+ #[test]
+ fn test_local_cache_config_requires_directory_when_enabled() {
+ let mut options = Options::new();
+ options.set(crate::common::CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+
+ let error = LocalCacheConfig::from_options(&options).unwrap_err();
+ assert!(matches!(error, crate::Error::ConfigInvalid { .. }));
+ assert!(error.to_string().contains("local-cache.dir"));
+ }
+
+ #[test]
+ fn test_local_cache_config_uses_disk_defaults() {
+ let mut options = Options::new();
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ options.set(CatalogOptions::LOCAL_CACHE_DIR, "/tmp/paimon-cache");
+
+ let config = LocalCacheConfig::from_options(&options).unwrap().unwrap();
+ assert_eq!(config.dir, std::path::Path::new("/tmp/paimon-cache"));
+ assert_eq!(config.max_size, None);
+ assert_eq!(config.block_size, 1024 * 1024);
+ assert_eq!(
+ config.whitelist,
+ std::collections::HashSet::from([FileType::Meta, FileType::GlobalIndex])
+ );
+ }
+
+ #[test]
+ fn test_local_cache_config_parses_custom_sizes() {
+ let mut options = Options::new();
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ options.set(CatalogOptions::LOCAL_CACHE_DIR, "/tmp/paimon-cache");
+ options.set(CatalogOptions::LOCAL_CACHE_MAX_SIZE, "2gb");
+ options.set(CatalogOptions::LOCAL_CACHE_BLOCK_SIZE, "64 kb");
+ options.set(CatalogOptions::LOCAL_CACHE_WHITELIST, "meta,data");
+
+ let config = LocalCacheConfig::from_options(&options).unwrap().unwrap();
+ assert_eq!(config.max_size, Some(2 * 1024 * 1024 * 1024));
+ assert_eq!(config.block_size, 64 * 1024);
+ assert_eq!(
+ config.whitelist,
+ std::collections::HashSet::from([FileType::Meta, FileType::Data])
+ );
+ }
+
+ #[test]
+ fn test_local_cache_namespace_uses_effective_catalog_options() {
+ let mut local_options = Options::new();
+ local_options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ local_options.set(CatalogOptions::LOCAL_CACHE_DIR, "/tmp/paimon-cache");
+ local_options.set(CatalogOptions::WAREHOUSE, "warehouse");
+ let mut effective_options = local_options.clone();
+ effective_options.set("s3.endpoint", "https://server-provided-endpoint");
+
+ let local_namespace = LocalCacheConfig::from_options(&local_options)
+ .unwrap()
+ .unwrap()
+ .namespace;
+ let effective_config =
+ LocalCacheConfig::from_options_with_namespace(&local_options, &effective_options)
+ .unwrap()
+ .unwrap();
+
+ assert_eq!(
+ effective_config.dir,
+ std::path::Path::new("/tmp/paimon-cache")
+ );
+ assert_ne!(effective_config.namespace, local_namespace);
+ assert_eq!(
+ effective_config.namespace,
+ catalog_namespace(&effective_options)
+ );
+ }
+
+ #[test]
+ fn test_local_cache_config_rejects_zero_block_size() {
+ let mut options = Options::new();
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ options.set(CatalogOptions::LOCAL_CACHE_DIR, "/tmp/paimon-cache");
+ options.set(CatalogOptions::LOCAL_CACHE_BLOCK_SIZE, "0");
+
+ let error = LocalCacheConfig::from_options(&options).unwrap_err();
+ assert!(matches!(error, crate::Error::ConfigInvalid { .. }));
+ assert!(error.to_string().contains("local-cache.block-size"));
+ }
+
+ #[test]
+ fn test_local_cache_config_rejects_invalid_enabled_value() {
+ let mut options = Options::new();
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "yes");
+
+ let error = LocalCacheConfig::from_options(&options).unwrap_err();
+ assert!(matches!(error, crate::Error::ConfigInvalid { .. }));
+ assert!(error.to_string().contains("local-cache.enabled"));
+ }
+
+ #[tokio::test]
+ async fn test_local_cache_file_size_is_removed_with_path_invalidation() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = LocalCache::new(LocalCacheConfig {
+ dir: directory.path().to_path_buf(),
+ namespace: "test".to_string(),
+ max_size: None,
+ block_size: 4,
+ whitelist: HashSet::from([FileType::Meta]),
+ })
+ .unwrap();
+ let path = "s3://bucket/table/snapshot/snapshot-1";
+ let token = cache.read_token(path);
+
+ assert_eq!(cache.file_size(path, &token).await, None);
+ cache.put_file_size(path, 42, &token).await;
+ assert_eq!(cache.file_size(path, &token).await, Some(42));
+ cache.invalidate_path(path).await;
+ assert_eq!(cache.file_size(path, &token).await, None);
+ }
+
+ #[tokio::test]
+ async fn test_local_cache_file_size_is_invalidated_across_shared_instances() {
+ let directory = tempfile::tempdir().unwrap();
+ let config = || LocalCacheConfig {
+ dir: directory.path().to_path_buf(),
+ namespace: "test".to_string(),
+ max_size: None,
+ block_size: 4,
+ whitelist: HashSet::from([FileType::Meta]),
+ };
+ let first = LocalCache::new(config()).unwrap();
+ let second = LocalCache::new(config()).unwrap();
+ let path = "s3://bucket/table/snapshot/snapshot-1";
+ let token = first.read_token(path);
+
+ first.put_file_size(path, 42, &token).await;
+ assert_eq!(second.file_size(path, &token).await, Some(42));
+ second.invalidate_path(path).await;
+ let current_token = first.read_token(path);
+ assert_eq!(first.file_size(path, ¤t_token).await, None);
+ }
+
+ #[tokio::test]
+ async fn test_stale_file_size_cannot_republish_after_invalidation() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = LocalCache::new(LocalCacheConfig {
+ dir: directory.path().to_path_buf(),
+ namespace: "test".to_string(),
+ max_size: None,
+ block_size: 4,
+ whitelist: HashSet::from([FileType::Meta]),
+ })
+ .unwrap();
+ let path = "s3://bucket/table/snapshot/snapshot-1";
+ let stale_token = cache.read_token(path);
+
+ assert_eq!(cache.file_size(path, &stale_token).await, None);
+ cache.invalidate_path(path).await;
+ cache.put_file_size(path, 42, &stale_token).await;
+
+ let current_token = cache.read_token(path);
+ assert_eq!(cache.file_size(path, ¤t_token).await, None);
+ }
+
+ #[test]
+ fn test_local_cache_uses_whitelist_and_bypasses_mutable_files() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = LocalCache::new(LocalCacheConfig {
+ dir: directory.path().to_path_buf(),
+ namespace: "test".to_string(),
+ max_size: None,
+ block_size: 4,
+ whitelist: HashSet::from([FileType::Meta]),
+ })
+ .unwrap();
+
+ assert!(cache.is_cacheable("s3://bucket/table/snapshot/snapshot-1"));
+ assert!(!cache.is_cacheable("s3://bucket/table/data/data-1.parquet"));
+ assert!(!cache.is_cacheable("s3://bucket/table/snapshot/LATEST"));
+ assert!(!cache.is_cacheable("s3://bucket/table/tag/tag-production"));
+ }
+
+ #[test]
+ fn test_local_cache_preserves_foreign_files_in_configured_directory() {
+ let directory = tempfile::tempdir().unwrap();
+ let foreign = directory.path().join("keep.txt");
+ let foreign_temporary = directory.path().join("foreign.tmp.data");
+ let nested_directory = directory.path().join("other-application");
+ let nested = nested_directory.join("keep.bin");
+ std::fs::create_dir(&nested_directory).unwrap();
+ std::fs::write(&foreign, b"foreign").unwrap();
+ std::fs::write(&foreign_temporary, b"foreign temporary").unwrap();
+ std::fs::write(&nested, b"nested foreign").unwrap();
+
+ LocalCache::new(LocalCacheConfig {
+ dir: directory.path().to_path_buf(),
+ namespace: "test".to_string(),
+ max_size: None,
+ block_size: 4,
+ whitelist: HashSet::from([FileType::Meta]),
+ })
+ .unwrap();
+
+ assert_eq!(std::fs::read(foreign).unwrap(), b"foreign");
+ assert_eq!(
+ std::fs::read(foreign_temporary).unwrap(),
+ b"foreign temporary"
+ );
+ assert_eq!(std::fs::read(nested).unwrap(), b"nested foreign");
+ }
+
+ #[tokio::test]
+ async fn test_local_cache_bounds_file_size_entries() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = LocalCache::new(LocalCacheConfig {
+ dir: directory.path().to_path_buf(),
+ namespace: "test".to_string(),
+ max_size: Some(8),
+ block_size: 4,
+ whitelist: HashSet::from([FileType::Meta]),
+ })
+ .unwrap();
+ let first_token = cache.read_token("snapshot-1");
+ let second_token = cache.read_token("snapshot-2");
+ let third_token = cache.read_token("snapshot-3");
+
+ cache.put_file_size("snapshot-1", 1, &first_token).await;
+ cache.put_file_size("snapshot-2", 2, &second_token).await;
+ cache.put_file_size("snapshot-3", 3, &third_token).await;
+
+ assert_eq!(cache.file_size("snapshot-1", &first_token).await, None);
+ assert_eq!(cache.file_size("snapshot-2", &second_token).await, Some(2));
+ assert_eq!(cache.file_size("snapshot-3", &third_token).await, Some(3));
+ }
+}
diff --git a/crates/paimon/src/io/cache/reader.rs b/crates/paimon/src/io/cache/reader.rs
new file mode 100644
index 00000000..40aeb728
--- /dev/null
+++ b/crates/paimon/src/io/cache/reader.rs
@@ -0,0 +1,726 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::ops::Range;
+use std::sync::Arc;
+
+use bytes::{Bytes, BytesMut};
+
+use crate::io::FileRead;
+
+use super::{CacheReadToken, LocalCache};
+
+pub(crate) struct CachedFileReader {
+ delegate: Arc<dyn FileRead>,
+ path: String,
+ file_size: u64,
+ cache: Arc<LocalCache>,
+ read_token: CacheReadToken,
+}
+
+impl CachedFileReader {
+ #[cfg(test)]
+ pub(crate) fn new(
+ delegate: Arc<dyn FileRead>,
+ path: impl Into<String>,
+ file_size: u64,
+ cache: Arc<LocalCache>,
+ ) -> Self {
+ let path = path.into();
+ let read_token = cache.read_token(&path);
+ Self::new_with_token(delegate, path, file_size, cache, read_token)
+ }
+
+ pub(in crate::io) fn new_with_token(
+ delegate: Arc<dyn FileRead>,
+ path: impl Into<String>,
+ file_size: u64,
+ cache: Arc<LocalCache>,
+ read_token: CacheReadToken,
+ ) -> Self {
+ Self {
+ delegate,
+ path: path.into(),
+ file_size,
+ cache,
+ read_token,
+ }
+ }
+
+ async fn read_block(&self, block_index: u64) -> crate::Result<Bytes> {
+ let block_size = self.cache.block_size();
+ let key = self.cache.block_key(&self.path, block_index);
+ let start = block_index * block_size;
+ let end = start.saturating_add(block_size).min(self.file_size);
+ let expected_len = usize::try_from(end - start).map_err(|_| crate::Error::DataInvalid {
+ message: format!("Cache block is too large for '{}'", self.path),
+ source: None,
+ })?;
+ if let Some(payload) = self.cache.get_block(&key, &self.read_token).await {
+ if payload.len() == expected_len {
+ return Ok(payload);
+ }
+ self.cache.remove_block(&key).await;
+ }
+
+ let load_lock = self.cache.block_load_lock(&key).await;
+ let load_guard = load_lock.lock().await;
+ let result = if let Some(payload) = self.cache.get_block(&key, &self.read_token).await {
+ if payload.len() == expected_len {
+ Ok(payload)
+ } else {
+ self.cache.remove_block(&key).await;
+ Err(crate::Error::DataInvalid {
+ message: format!(
+ "Cached block {} for '{}' has length {}, expected {}",
+ block_index,
+ self.path,
+ payload.len(),
+ expected_len
+ ),
+ source: None,
+ })
+ }
+ } else {
+ match self.delegate.read(start..end).await {
+ Ok(payload) if payload.len() == expected_len => {
+ self.cache
+ .put_block(&key, payload.clone(), &self.read_token)
+ .await;
+ Ok(payload)
+ }
+ Ok(payload) => Err(crate::Error::DataInvalid {
+ message: format!(
+ "Source block {} for '{}' has length {}, expected {}",
+ block_index,
+ self.path,
+ payload.len(),
+ expected_len
+ ),
+ source: None,
+ }),
+ Err(error) => Err(error),
+ }
+ };
+ drop(load_guard);
+ self.cache.release_block_load_lock(&key, &load_lock).await;
+ result
+ }
+
+ pub(crate) async fn read_full(&self) -> crate::Result<Bytes> {
+ if self.file_size == 0 {
+ return Ok(Bytes::new());
+ }
+ if let Some(payload) = self.read_cached_full().await {
+ return Ok(payload);
+ }
+
+ let first_key = self.cache.block_key(&self.path, 0);
+ let load_lock = self.cache.block_load_lock(&first_key).await;
+ let load_guard = load_lock.lock().await;
+ let result = if let Some(payload) = self.read_cached_full().await {
+ Ok(payload)
+ } else {
+ match self.delegate.read(0..self.file_size).await {
+ Ok(payload) if payload.len() as u64 == self.file_size => {
+ let chunk_size = usize::try_from(self.cache.block_size()).map_err(|_| {
+ crate::Error::DataInvalid {
+ message: format!("Cache block size is too large for '{}'", self.path),
+ source: None,
+ }
+ })?;
+ for (block_index, block) in payload.chunks(chunk_size).enumerate() {
+ let block_index =
+ u64::try_from(block_index).map_err(|_| crate::Error::DataInvalid {
+ message: format!("Too many cache blocks for '{}'", self.path),
+ source: None,
+ })?;
+ let key = self.cache.block_key(&self.path, block_index);
+ self.cache
+ .put_block(&key, Bytes::copy_from_slice(block), &self.read_token)
+ .await;
+ }
+ Ok(payload)
+ }
+ Ok(payload) => Err(crate::Error::DataInvalid {
+ message: format!(
+ "Source file '{}' has length {}, expected {}",
+ self.path,
+ payload.len(),
+ self.file_size
+ ),
+ source: None,
+ }),
+ Err(error) => Err(error),
+ }
+ };
+ drop(load_guard);
+ self.cache
+ .release_block_load_lock(&first_key, &load_lock)
+ .await;
+ result
+ }
+
+ async fn read_cached_full(&self) -> Option<Bytes> {
+ let block_size = self.cache.block_size();
+ let block_count = self.file_size.div_ceil(block_size);
+ let output_len = usize::try_from(self.file_size).ok()?;
+ let mut output = BytesMut::with_capacity(output_len);
+ for block_index in 0..block_count {
+ let key = self.cache.block_key(&self.path, block_index);
+ let payload = self.cache.get_block(&key, &self.read_token).await?;
+ let start = block_index * block_size;
+ let expected_len = (self.file_size - start).min(block_size) as usize;
+ if payload.len() != expected_len {
+ self.cache.remove_block(&key).await;
+ return None;
+ }
+ output.extend_from_slice(&payload);
+ }
+ Some(output.freeze())
+ }
+}
+
+#[async_trait::async_trait]
+impl FileRead for CachedFileReader {
+ async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
+ if range.start > range.end || range.end > self.file_size {
+ return self.delegate.read(range).await;
+ }
+ let end = range.end;
+ if range.start >= end {
+ return Ok(Bytes::new());
+ }
+
+ let block_size = self.cache.block_size();
+ let first_block = range.start / block_size;
+ let last_block = (end - 1) / block_size;
+ let output_len =
+ usize::try_from(end - range.start).map_err(|_| crate::Error::DataInvalid {
+ message: format!("File read range is too large for '{}'", self.path),
+ source: None,
+ })?;
+ let mut output = BytesMut::with_capacity(output_len);
+
+ for block_index in first_block..=last_block {
+ let block = self.read_block(block_index).await?;
+ let block_start = block_index * block_size;
+ let copy_start = range.start.max(block_start) - block_start;
+ let copy_end = end.min(block_start + block.len() as u64) - block_start;
+ output.extend_from_slice(&block[copy_start as usize..copy_end as usize]);
+ }
+
+ Ok(output.freeze())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::sync::atomic::{AtomicUsize, Ordering};
+
+ use super::super::{FileType, LocalCache, LocalCacheConfig};
+ use super::*;
+ use crate::common::{CatalogOptions, Options};
+ use crate::io::cache::create_local_cache;
+
+ #[derive(Debug)]
+ struct CountingReader {
+ data: Bytes,
+ reads: AtomicUsize,
+ }
+
+ #[async_trait::async_trait]
+ impl FileRead for CountingReader {
+ async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
+ self.reads.fetch_add(1, Ordering::SeqCst);
+ Ok(self.data.slice(range.start as usize..range.end as usize))
+ }
+ }
+
+ #[tokio::test]
+ async fn test_cached_range_reader_reads_unaligned_blocks_and_reuses_them() {
+ let directory = tempfile::tempdir().unwrap();
+ let delegate = Arc::new(CountingReader {
+ data: Bytes::from_static(b"abcdefghijkl"),
+ reads: AtomicUsize::new(0),
+ });
+ let cache = Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ dir: directory.path().to_path_buf(),
+ namespace: "test".to_string(),
+ max_size: None,
+ block_size: 4,
+ whitelist: std::collections::HashSet::from([FileType::Meta]),
+ })
+ .unwrap(),
+ );
+ let reader = CachedFileReader::new(
+ delegate.clone(),
+ "s3://bucket/table/snapshot/snapshot-1",
+ 12,
+ cache,
+ );
+
+ assert_eq!(
+ reader.read(2..10).await.unwrap(),
+ Bytes::from_static(b"cdefghij")
+ );
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 3);
+ assert_eq!(
+ reader.read(2..10).await.unwrap(),
+ Bytes::from_static(b"cdefghij")
+ );
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 3);
+ }
+
+ #[derive(Debug)]
+ struct SlowCountingReader {
+ data: Bytes,
+ reads: AtomicUsize,
+ }
+
+ #[async_trait::async_trait]
+ impl FileRead for SlowCountingReader {
+ async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
+ self.reads.fetch_add(1, Ordering::SeqCst);
+ tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+ Ok(self.data.slice(range.start as usize..range.end as usize))
+ }
+ }
+
+ #[tokio::test]
+ async fn test_cached_range_single_flight_reads_cold_block_once() {
+ let directory = tempfile::tempdir().unwrap();
+ let delegate = Arc::new(SlowCountingReader {
+ data: Bytes::from_static(b"abcdefgh"),
+ reads: AtomicUsize::new(0),
+ });
+ let cache = Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ dir: directory.path().to_path_buf(),
+ namespace: "test".to_string(),
+ max_size: None,
+ block_size: 4,
+ whitelist: std::collections::HashSet::from([FileType::Meta]),
+ })
+ .unwrap(),
+ );
+ let reader = CachedFileReader::new(
+ delegate.clone(),
+ "s3://bucket/table/snapshot/snapshot-1",
+ 8,
+ cache,
+ );
+
+ let (first, second) = tokio::join!(reader.read(0..4), reader.read(0..4));
+
+ assert_eq!(first.unwrap(), Bytes::from_static(b"abcd"));
+ assert_eq!(second.unwrap(), Bytes::from_static(b"abcd"));
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
+ }
+
+ #[tokio::test]
+ async fn test_cached_range_single_flight_is_shared_across_cache_instances() {
+ let directory = tempfile::tempdir().unwrap();
+ let delegate = Arc::new(SlowCountingReader {
+ data: Bytes::from_static(b"abcdefgh"),
+ reads: AtomicUsize::new(0),
+ });
+ let config = || LocalCacheConfig {
+ dir: directory.path().to_path_buf(),
+ namespace: "test".to_string(),
+ max_size: None,
+ block_size: 4,
+ whitelist: std::collections::HashSet::from([FileType::Meta]),
+ };
+ let first_reader = CachedFileReader::new(
+ delegate.clone(),
+ "s3://bucket/table/snapshot/snapshot-1",
+ 8,
+ Arc::new(LocalCache::new(config()).unwrap()),
+ );
+ let second_reader = CachedFileReader::new(
+ delegate.clone(),
+ "s3://bucket/table/snapshot/snapshot-1",
+ 8,
+ Arc::new(LocalCache::new(config()).unwrap()),
+ );
+
+ let (first, second) = tokio::join!(first_reader.read(0..4), second_reader.read(0..4));
+
+ assert_eq!(first.unwrap(), Bytes::from_static(b"abcd"));
+ assert_eq!(second.unwrap(), Bytes::from_static(b"abcd"));
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
+ }
+
+ #[derive(Debug)]
+ struct BlockingReader {
+ data: Bytes,
+ started: Arc<tokio::sync::Notify>,
+ release: Arc<tokio::sync::Notify>,
+ }
+
+ #[async_trait::async_trait]
+ impl FileRead for BlockingReader {
+ async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
+ self.started.notify_one();
+ self.release.notified().await;
+ Ok(self.data.slice(range.start as usize..range.end as usize))
+ }
+ }
+
+ #[tokio::test]
+ async fn test_in_flight_miss_does_not_republish_after_invalidation() {
+ let directory = tempfile::tempdir().unwrap();
+ let path = "s3://bucket/table/snapshot/snapshot-1";
+ let cache = Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ dir: directory.path().to_path_buf(),
+ namespace: "test".to_string(),
+ max_size: None,
+ block_size: 4,
+ whitelist: std::collections::HashSet::from([FileType::Meta]),
+ })
+ .unwrap(),
+ );
+ let started = Arc::new(tokio::sync::Notify::new());
+ let release = Arc::new(tokio::sync::Notify::new());
+ let old_reader = CachedFileReader::new(
+ Arc::new(BlockingReader {
+ data: Bytes::from_static(b"old!"),
+ started: started.clone(),
+ release: release.clone(),
+ }),
+ path,
+ 4,
+ cache.clone(),
+ );
+ let old_load = tokio::spawn(async move { old_reader.read(0..4).await });
+
+ started.notified().await;
+ cache.invalidate_path(path).await;
+ release.notify_one();
+ assert_eq!(
+ old_load.await.unwrap().unwrap(),
+ Bytes::from_static(b"old!")
+ );
+
+ let current_delegate = Arc::new(CountingReader {
+ data: Bytes::from_static(b"new!"),
+ reads: AtomicUsize::new(0),
+ });
+ let current_reader =
+ CachedFileReader::new(current_delegate.clone(), path, 4, cache.clone());
+ assert_eq!(
+ current_reader.read(0..4).await.unwrap(),
+ Bytes::from_static(b"new!")
+ );
+ assert_eq!(current_delegate.reads.load(Ordering::SeqCst), 1);
+ }
+
+ #[tokio::test]
+ async fn test_in_flight_miss_cannot_republish_across_shared_cache_instances() {
+ let directory = tempfile::tempdir().unwrap();
+ let path = "s3://bucket/table/snapshot/snapshot-1";
+ let cache_a = Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ dir: directory.path().to_path_buf(),
+ namespace: "test".to_string(),
+ max_size: None,
+ block_size: 4,
+ whitelist: std::collections::HashSet::from([FileType::Meta]),
+ })
+ .unwrap(),
+ );
+ let cache_b = Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ dir: directory.path().to_path_buf(),
+ namespace: "test".to_string(),
+ max_size: None,
+ block_size: 4,
+ whitelist: std::collections::HashSet::from([FileType::Meta]),
+ })
+ .unwrap(),
+ );
+ let started = Arc::new(tokio::sync::Notify::new());
+ let release = Arc::new(tokio::sync::Notify::new());
+ let old_reader = CachedFileReader::new(
+ Arc::new(BlockingReader {
+ data: Bytes::from_static(b"old!"),
+ started: started.clone(),
+ release: release.clone(),
+ }),
+ path,
+ 4,
+ cache_a,
+ );
+ let old_load = tokio::spawn(async move { old_reader.read(0..4).await });
+
+ started.notified().await;
+ cache_b.invalidate_path(path).await;
+ release.notify_one();
+ assert_eq!(
+ old_load.await.unwrap().unwrap(),
+ Bytes::from_static(b"old!")
+ );
+
+ let current_delegate = Arc::new(CountingReader {
+ data: Bytes::from_static(b"new!"),
+ reads: AtomicUsize::new(0),
+ });
+ let current_reader =
+ CachedFileReader::new(current_delegate.clone(), path, 4, cache_b.clone());
+ assert_eq!(
+ current_reader.read(0..4).await.unwrap(),
+ Bytes::from_static(b"new!")
+ );
+ assert_eq!(current_delegate.reads.load(Ordering::SeqCst), 1);
+ }
+
+ #[tokio::test]
+ async fn test_prefix_invalidation_cannot_hit_old_block_while_waiting_on_sibling() {
+ let directory = tempfile::tempdir().unwrap();
+ let prefix = "s3://bucket/table/snapshot";
+ let first_path = "s3://bucket/table/snapshot/snapshot-1";
+ let observed_path = "s3://bucket/table/snapshot/snapshot-2";
+ let blocked_path = "s3://bucket/table/snapshot/snapshot-3";
+ let cache = Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ dir: directory.path().to_path_buf(),
+ namespace: "test".to_string(),
+ max_size: None,
+ block_size: 4,
+ whitelist: std::collections::HashSet::from([FileType::Meta]),
+ })
+ .unwrap(),
+ );
+ let old_reader = CachedFileReader::new(
+ Arc::new(CountingReader {
+ data: Bytes::from_static(b"old!"),
+ reads: AtomicUsize::new(0),
+ }),
+ first_path,
+ 4,
+ cache.clone(),
+ );
+ assert_eq!(
+ old_reader.read(0..4).await.unwrap(),
+ Bytes::from_static(b"old!")
+ );
+ let warm_size_token = cache.read_token(first_path);
+ cache.put_file_size(first_path, 4, &warm_size_token).await;
+ drop(warm_size_token);
+ drop(old_reader);
+
+ let observed_token = cache.read_token(observed_path);
+ let blocked_token = cache.read_token(blocked_path);
+ let blocked_guard = blocked_token.publish_guard().await;
+ let invalidating_cache = cache.clone();
+ let invalidation =
+ tokio::spawn(async move { invalidating_cache.invalidate_prefix(prefix).await });
+ tokio::time::timeout(std::time::Duration::from_secs(1), async {
+ while observed_token.is_current() {
+ tokio::task::yield_now().await;
+ }
+ })
+ .await
+ .unwrap();
+
+ let current_delegate = Arc::new(CountingReader {
+ data: Bytes::from_static(b"new!"),
+ reads: AtomicUsize::new(0),
+ });
+ let current_reader =
+ CachedFileReader::new(current_delegate.clone(), first_path, 4, cache.clone());
+ let mut current_load = tokio::spawn(async move { current_reader.read(0..4).await });
+ let size_cache = cache.clone();
+ let mut current_size = tokio::spawn(async move {
+ let token = size_cache.read_token(first_path);
+ size_cache.file_size(first_path, &token).await
+ });
+
+ assert!(
+ tokio::time::timeout(std::time::Duration::from_millis(100), &mut current_load)
+ .await
+ .is_err(),
+ "cache read completed before prefix invalidation removed the old block"
+ );
+ assert!(
+ tokio::time::timeout(std::time::Duration::from_millis(100), &mut current_size)
+ .await
+ .is_err(),
+ "file-size lookup completed before prefix invalidation removed the old entry"
+ );
+ drop(blocked_guard);
+ invalidation.await.unwrap();
+ assert_eq!(
+ current_load.await.unwrap().unwrap(),
+ Bytes::from_static(b"new!")
+ );
+ assert_eq!(current_size.await.unwrap(), None);
+ assert_eq!(current_delegate.reads.load(Ordering::SeqCst), 1);
+ }
+
+ #[tokio::test]
+ async fn test_catalog_namespace_prevents_cross_catalog_cache_hits() {
+ let directory = tempfile::tempdir().unwrap();
+ let path = "s3://bucket/table/snapshot/snapshot-1";
+ let mut first_options = Options::new();
+ first_options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ first_options.set(
+ CatalogOptions::LOCAL_CACHE_DIR,
+ directory.path().to_string_lossy(),
+ );
+ first_options.set(CatalogOptions::URI, "https://catalog-a.example");
+ let first_cache = create_local_cache(&first_options).unwrap().unwrap();
+ let first_reader = CachedFileReader::new(
+ Arc::new(CountingReader {
+ data: Bytes::from_static(b"from"),
+ reads: AtomicUsize::new(0),
+ }),
+ path,
+ 4,
+ first_cache,
+ );
+ assert_eq!(
+ first_reader.read(0..4).await.unwrap(),
+ Bytes::from_static(b"from")
+ );
+
+ let mut second_options = first_options;
+ second_options.set(CatalogOptions::URI, "https://catalog-b.example");
+ let second_cache = create_local_cache(&second_options).unwrap().unwrap();
+ let second_delegate = Arc::new(CountingReader {
+ data: Bytes::from_static(b"new!"),
+ reads: AtomicUsize::new(0),
+ });
+ let second_reader = CachedFileReader::new(second_delegate.clone(), path, 4, second_cache);
+
+ assert_eq!(
+ second_reader.read(0..4).await.unwrap(),
+ Bytes::from_static(b"new!")
+ );
+ assert_eq!(second_delegate.reads.load(Ordering::SeqCst), 1);
+ }
+
+ #[derive(Debug)]
+ struct BoundedReader {
+ data: Bytes,
+ }
+
+ #[async_trait::async_trait]
+ impl FileRead for BoundedReader {
+ async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
+ if range.end > self.data.len() as u64 {
+ return Err(crate::Error::DataInvalid {
+ message: "range exceeds file size".to_string(),
+ source: None,
+ });
+ }
+ if range.start > range.end {
+ return Ok(Bytes::new());
+ }
+ Ok(self.data.slice(range.start as usize..range.end as usize))
+ }
+ }
+
+ #[tokio::test]
+ async fn test_cached_range_reader_preserves_delegate_boundary_semantics() {
+ let directory = tempfile::tempdir().unwrap();
+ let delegate = Arc::new(BoundedReader {
+ data: Bytes::from_static(b"abcde"),
+ });
+ let cache = Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ dir: directory.path().to_path_buf(),
+ namespace: "test".to_string(),
+ max_size: None,
+ block_size: 4,
+ whitelist: std::collections::HashSet::from([FileType::Meta]),
+ })
+ .unwrap(),
+ );
+ let reader = CachedFileReader::new(delegate, "snapshot-1", 5, cache);
+
+ assert!(reader.read(3..8).await.is_err());
+ let reversed_start = 4;
+ let reversed_end = 3;
+ assert_eq!(
+ reader.read(reversed_start..reversed_end).await.unwrap(),
+ Bytes::new()
+ );
+ }
+
+ #[derive(Debug)]
+ struct ShortReader;
+
+ #[async_trait::async_trait]
+ impl FileRead for ShortReader {
+ async fn read(&self, _range: Range<u64>) -> crate::Result<Bytes> {
+ Ok(Bytes::from_static(b"x"))
+ }
+ }
+
+ #[tokio::test]
+ async fn test_cached_range_reader_rejects_logically_short_block() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ dir: directory.path().to_path_buf(),
+ namespace: "test".to_string(),
+ max_size: None,
+ block_size: 4,
+ whitelist: std::collections::HashSet::from([FileType::Meta]),
+ })
+ .unwrap(),
+ );
+ let reader = CachedFileReader::new(Arc::new(ShortReader), "snapshot-1", 8, cache);
+
+ assert!(reader.read(2..4).await.is_err());
+ }
+
+ #[tokio::test]
+ async fn test_cached_full_reader_loads_source_once_then_hits_blocks() {
+ let directory = tempfile::tempdir().unwrap();
+ let delegate = Arc::new(CountingReader {
+ data: Bytes::from_static(b"abcdefghijkl"),
+ reads: AtomicUsize::new(0),
+ });
+ let cache = Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ dir: directory.path().to_path_buf(),
+ namespace: "test".to_string(),
+ max_size: None,
+ block_size: 4,
+ whitelist: std::collections::HashSet::from([FileType::Meta]),
+ })
+ .unwrap(),
+ );
+ let reader = CachedFileReader::new(delegate.clone(), "snapshot-1", 12, cache.clone());
+
+ assert_eq!(
+ reader.read_full().await.unwrap(),
+ Bytes::from_static(b"abcdefghijkl")
+ );
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
+ assert_eq!(
+ reader.read_full().await.unwrap(),
+ Bytes::from_static(b"abcdefghijkl")
+ );
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
+ }
+}
diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index 996d9acc..ee5b5711 100644
--- a/crates/paimon/src/io/file_io.rs
+++ b/crates/paimon/src/io/file_io.rs
@@ -17,8 +17,11 @@
use crate::error::*;
use std::collections::HashMap;
+use std::future::Future;
use std::ops::Range;
+use std::pin::Pin;
use std::sync::Arc;
+use std::task::{Context, Poll};
use std::time::SystemTime;
use bytes::Bytes;
@@ -29,14 +32,21 @@ use snafu::ResultExt;
use tokio_util::compat::FuturesAsyncWriteCompatExt;
use url::Url;
+use super::cache::{CachedFileReader, LocalCache};
use super::Storage;
#[derive(Clone, Debug)]
pub struct FileIO {
storage: Arc<Storage>,
+ cache: Option<Arc<LocalCache>>,
}
impl FileIO {
+ #[cfg(test)]
+ pub(crate) fn has_local_cache(&self) -> bool {
+ self.cache.is_some()
+ }
+
/// Try to infer file io scheme from path.
///
/// The input HashMap is paimon-java's [`Options`](https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/options/Options.java#L60)
@@ -79,10 +89,17 @@ impl FileIO {
/// Reference: <https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java#L76>
pub fn new_input(&self, path: &str) -> crate::Result<InputFile> {
let (op, relative_path) = self.storage.create(path)?;
+ let cache_path = cache_object_path(&op, relative_path.as_ref());
Ok(InputFile {
op,
path: path.to_string(),
relative_path: relative_path.into_owned(),
+ cache_path,
+ cache: self
+ .cache
+ .as_ref()
+ .filter(|cache| cache.is_cacheable(path))
+ .cloned(),
})
}
@@ -91,10 +108,17 @@ impl FileIO {
/// Reference: <https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java#L87>
pub fn new_output(&self, path: &str) -> Result<OutputFile> {
let (op, relative_path) = self.storage.create(path)?;
+ let cache_path = cache_object_path(&op, relative_path.as_ref());
Ok(OutputFile {
op,
path: path.to_string(),
relative_path: relative_path.into_owned(),
+ cache_path,
+ cache: self
+ .cache
+ .as_ref()
+ .filter(|cache| cache.is_cacheable(path))
+ .cloned(),
})
}
@@ -230,12 +254,16 @@ 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.storage.create(path)?;
+ let cache_path = cache_object_path(&op, relative_path.as_ref());
op.delete(relative_path.as_ref())
.await
.context(IoUnexpectedSnafu {
message: format!("Failed to delete file '{path}'"),
})?;
+ if let Some(cache) = self.cache.as_ref().filter(|cache| cache.is_cacheable(path)) {
+ cache.invalidate_path(&cache_path).await;
+ }
Ok(())
}
@@ -245,6 +273,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.storage.create(path)?;
+ let cache_path = cache_object_path(&op, relative_path.as_ref());
op.delete_with(relative_path.as_ref())
.recursive(true)
@@ -252,6 +281,9 @@ impl FileIO {
.context(IoUnexpectedSnafu {
message: format!("Failed to delete directory '{path}'"),
})?;
+ if let Some(cache) = &self.cache {
+ cache.invalidate_prefix(&cache_path).await;
+ }
Ok(())
}
@@ -288,7 +320,9 @@ 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.storage.create(src)?;
- let (_, relative_path_dst) = self.storage.create(dst)?;
+ let (op_dst, relative_path_dst) = self.storage.create(dst)?;
+ 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())
@@ -296,6 +330,10 @@ impl FileIO {
.context(IoUnexpectedSnafu {
message: format!("Failed to rename '{src}' to '{dst}'"),
})?;
+ if let Some(cache) = &self.cache {
+ cache.invalidate_prefix(&cache_path_src).await;
+ cache.invalidate_prefix(&cache_path_dst).await;
+ }
Ok(())
}
@@ -309,6 +347,17 @@ fn status_path(base_path: &str, entry_path: &str) -> String {
}
}
+fn cache_object_path(op: &Operator, relative_path: &str) -> String {
+ let info = op.info();
+ format!(
+ "{}\0{}\0{}\0{}",
+ info.scheme(),
+ info.name(),
+ info.root(),
+ relative_path.trim_start_matches('/')
+ )
+}
+
/// 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();
@@ -322,6 +371,7 @@ pub(crate) fn looks_like_windows_drive_path(path: &str) -> bool {
pub struct FileIOBuilder {
scheme_str: Option<String>,
props: HashMap<String, String>,
+ cache: Option<Arc<LocalCache>>,
}
impl FileIOBuilder {
@@ -329,6 +379,7 @@ impl FileIOBuilder {
Self {
scheme_str: Some(scheme_str.to_string()),
props: HashMap::default(),
+ cache: None,
}
}
@@ -350,10 +401,17 @@ impl FileIOBuilder {
self
}
+ pub(crate) fn with_local_cache(mut self, cache: Arc<LocalCache>) -> Self {
+ self.cache = Some(cache);
+ self
+ }
+
pub fn build(self) -> crate::Result<FileIO> {
+ let cache = self.cache.clone();
let storage = Storage::build(self)?;
Ok(FileIO {
storage: Arc::new(storage),
+ cache,
})
}
}
@@ -370,6 +428,21 @@ impl FileRead for opendal::Reader {
}
}
+enum InputFileReader {
+ Direct(opendal::Reader),
+ Cached(CachedFileReader),
+}
+
+#[async_trait::async_trait]
+impl FileRead for InputFileReader {
+ async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
+ match self {
+ Self::Direct(reader) => FileRead::read(reader, range).await,
+ Self::Cached(reader) => FileRead::read(reader, range).await,
+ }
+ }
+}
+
#[async_trait::async_trait]
pub trait FileWrite: Send + Unpin + 'static {
async fn write(&mut self, bs: Bytes) -> crate::Result<()>;
@@ -389,11 +462,82 @@ impl FileWrite for opendal::Writer {
}
}
+struct CacheInvalidatingWriter {
+ delegate: Box<dyn FileWrite>,
+ cache: Arc<LocalCache>,
+ path: String,
+}
+
+#[async_trait::async_trait]
+impl FileWrite for CacheInvalidatingWriter {
+ async fn write(&mut self, bs: Bytes) -> crate::Result<()> {
+ self.delegate.write(bs).await
+ }
+
+ async fn close(&mut self) -> crate::Result<()> {
+ self.delegate.close().await?;
+ self.cache.invalidate_path(&self.path).await;
+ Ok(())
+ }
+}
+
/// Async streaming writer trait for format-level writers (e.g. parquet).
pub trait AsyncFileWrite: tokio::io::AsyncWrite + Unpin + Send {}
impl<T: tokio::io::AsyncWrite + Unpin + Send> AsyncFileWrite for T {}
+struct CacheInvalidatingAsyncWriter {
+ delegate: Box<dyn AsyncFileWrite>,
+ cache: Arc<LocalCache>,
+ path: String,
+ delegate_shutdown: bool,
+ invalidation: Option<Pin<Box<dyn Future<Output = ()> + Send>>>,
+}
+
+impl tokio::io::AsyncWrite for CacheInvalidatingAsyncWriter {
+ fn poll_write(
+ mut self: Pin<&mut Self>,
+ context: &mut Context<'_>,
+ buffer: &[u8],
+ ) -> Poll<std::io::Result<usize>> {
+ Pin::new(&mut *self.delegate).poll_write(context, buffer)
+ }
+
+ fn poll_flush(
+ mut self: Pin<&mut Self>,
+ context: &mut Context<'_>,
+ ) -> Poll<std::io::Result<()>> {
+ Pin::new(&mut *self.delegate).poll_flush(context)
+ }
+
+ fn poll_shutdown(
+ mut self: Pin<&mut Self>,
+ context: &mut Context<'_>,
+ ) -> Poll<std::io::Result<()>> {
+ if !self.delegate_shutdown {
+ match Pin::new(&mut *self.delegate).poll_shutdown(context) {
+ Poll::Pending => return Poll::Pending,
+ Poll::Ready(Err(error)) => return Poll::Ready(Err(error)),
+ Poll::Ready(Ok(())) => {
+ self.delegate_shutdown = true;
+ let cache = self.cache.clone();
+ let path = self.path.clone();
+ self.invalidation =
+ Some(Box::pin(async move { cache.invalidate_path(&path).await }));
+ }
+ }
+ }
+
+ if let Some(invalidation) = &mut self.invalidation {
+ match invalidation.as_mut().poll(context) {
+ Poll::Pending => return Poll::Pending,
+ Poll::Ready(()) => self.invalidation = None,
+ }
+ }
+ Poll::Ready(Ok(()))
+ }
+}
+
#[derive(Clone, Debug)]
pub struct FileStatus {
pub size: u64,
@@ -409,6 +553,8 @@ pub struct InputFile {
/// The opendal-relative path (see [`FileIO::new_input`]); not necessarily a
/// suffix of `path`, since local paths are separator-normalized.
relative_path: String,
+ cache_path: String,
+ cache: Option<Arc<LocalCache>>,
}
impl InputFile {
@@ -434,11 +580,53 @@ impl InputFile {
}
pub async fn read(&self) -> crate::Result<Bytes> {
- Ok(self.op.read(&self.relative_path).await?.to_bytes())
+ let Some(cache) = &self.cache else {
+ return Ok(self.op.read(&self.relative_path).await?.to_bytes());
+ };
+ let read_token = cache.read_token(&self.cache_path);
+ let size = if let Some(size) = cache.file_size(&self.cache_path, &read_token).await {
+ size
+ } else {
+ let size = self.op.stat(&self.relative_path).await?.content_length();
+ cache
+ .put_file_size(&self.cache_path, size, &read_token)
+ .await;
+ size
+ };
+ let delegate = Arc::new(self.op.reader(&self.relative_path).await?);
+ CachedFileReader::new_with_token(
+ delegate,
+ &self.cache_path,
+ size,
+ cache.clone(),
+ read_token,
+ )
+ .read_full()
+ .await
}
pub async fn reader(&self) -> crate::Result<impl FileRead> {
- Ok(self.op.reader(&self.relative_path).await?)
+ let reader = self.op.reader(&self.relative_path).await?;
+ let Some(cache) = &self.cache else {
+ return Ok(InputFileReader::Direct(reader));
+ };
+ let read_token = cache.read_token(&self.cache_path);
+ let size = if let Some(size) = cache.file_size(&self.cache_path, &read_token).await {
+ size
+ } else {
+ let size = self.op.stat(&self.relative_path).await?.content_length();
+ cache
+ .put_file_size(&self.cache_path, size, &read_token)
+ .await;
+ size
+ };
+ Ok(InputFileReader::Cached(CachedFileReader::new_with_token(
+ Arc::new(reader),
+ &self.cache_path,
+ size,
+ cache.clone(),
+ read_token,
+ )))
}
}
@@ -449,6 +637,8 @@ pub struct OutputFile {
/// The opendal-relative path (see [`FileIO::new_output`]); not necessarily a
/// suffix of `path`, since local paths are separator-normalized.
relative_path: String,
+ cache_path: String,
+ cache: Option<Arc<LocalCache>>,
}
impl OutputFile {
@@ -461,10 +651,13 @@ impl OutputFile {
}
pub fn to_input_file(self) -> InputFile {
+ let cache = self.cache.filter(|cache| cache.is_cacheable(&self.path));
InputFile {
op: self.op,
path: self.path,
relative_path: self.relative_path,
+ cache_path: self.cache_path,
+ cache,
}
}
@@ -475,17 +668,35 @@ impl OutputFile {
}
pub async fn writer(&self) -> crate::Result<Box<dyn FileWrite>> {
- Ok(Box::new(self.opendal_writer().await?))
+ let writer: Box<dyn FileWrite> = Box::new(self.opendal_writer().await?);
+ let Some(cache) = &self.cache else {
+ return Ok(writer);
+ };
+ Ok(Box::new(CacheInvalidatingWriter {
+ delegate: writer,
+ cache: cache.clone(),
+ path: self.cache_path.clone(),
+ }))
}
/// Get an async streaming writer for format-level writes (e.g. parquet).
pub(crate) async fn async_writer(&self) -> crate::Result<Box<dyn AsyncFileWrite>> {
- Ok(Box::new(
+ let writer: Box<dyn AsyncFileWrite> = Box::new(
self.opendal_writer()
.await?
.into_futures_async_write()
.compat_write(),
- ))
+ );
+ let Some(cache) = &self.cache else {
+ return Ok(writer);
+ };
+ Ok(Box::new(CacheInvalidatingAsyncWriter {
+ delegate: writer,
+ cache: cache.clone(),
+ path: self.cache_path.clone(),
+ delegate_shutdown: false,
+ invalidation: None,
+ }))
}
async fn opendal_writer(&self) -> crate::Result<opendal::Writer> {
@@ -896,7 +1107,11 @@ mod object_storage_path_test {
#[cfg(test)]
mod input_output_test {
+ use std::sync::Arc;
+
use super::*;
+ use crate::common::{CatalogOptions, Options};
+ use crate::io::cache::{LocalCache, LocalCacheConfig};
use bytes::Bytes;
fn setup_memory_file_io() -> FileIO {
@@ -907,6 +1122,23 @@ mod input_output_test {
FileIOBuilder::new("file").build().unwrap()
}
+ fn setup_cached_fs_file_io(cache_directory: &std::path::Path) -> FileIO {
+ let mut options = Options::new();
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ options.set(
+ CatalogOptions::LOCAL_CACHE_DIR,
+ cache_directory.to_string_lossy(),
+ );
+ options.set(CatalogOptions::LOCAL_CACHE_BLOCK_SIZE, "4");
+ let cache = Arc::new(
+ LocalCache::new(LocalCacheConfig::from_options(&options).unwrap().unwrap()).unwrap(),
+ );
+ FileIOBuilder::new("file")
+ .with_local_cache(cache)
+ .build()
+ .unwrap()
+ }
+
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();
@@ -1009,4 +1241,361 @@ mod input_output_test {
let file_io = setup_fs_file_io();
common_test_input_file_partial_read(&file_io, "file:/tmp/test_file_read_fs").await;
}
+
+ #[cfg(not(windows))]
+ #[tokio::test]
+ async fn test_file_io_local_cache_serves_full_read_after_source_disappears() {
+ let source_directory = tempfile::tempdir().unwrap();
+ let cache_directory = tempfile::tempdir().unwrap();
+ let source_path = source_directory.path().join("snapshot-1");
+ std::fs::write(&source_path, b"cached metadata").unwrap();
+ let location = format!("file:{}", source_path.display());
+ let file_io = setup_cached_fs_file_io(cache_directory.path());
+
+ assert_eq!(
+ file_io.new_input(&location).unwrap().read().await.unwrap(),
+ Bytes::from_static(b"cached metadata")
+ );
+ std::fs::remove_file(&source_path).unwrap();
+ assert_eq!(
+ file_io.new_input(&location).unwrap().read().await.unwrap(),
+ Bytes::from_static(b"cached metadata")
+ );
+ }
+
+ #[cfg(not(windows))]
+ #[tokio::test]
+ async fn test_file_io_local_cache_serves_range_after_source_disappears() {
+ let source_directory = tempfile::tempdir().unwrap();
+ let cache_directory = tempfile::tempdir().unwrap();
+ let source_path = source_directory.path().join("snapshot-1");
+ std::fs::write(&source_path, b"cached metadata").unwrap();
+ let location = format!("file:{}", source_path.display());
+ let file_io = setup_cached_fs_file_io(cache_directory.path());
+
+ let reader = file_io
+ .new_input(&location)
+ .unwrap()
+ .reader()
+ .await
+ .unwrap();
+ assert_eq!(
+ reader.read(1..7).await.unwrap(),
+ Bytes::from_static(b"ached ")
+ );
+ std::fs::remove_file(&source_path).unwrap();
+ let reader = file_io
+ .new_input(&location)
+ .unwrap()
+ .reader()
+ .await
+ .unwrap();
+ assert_eq!(
+ reader.read(1..7).await.unwrap(),
+ Bytes::from_static(b"ached ")
+ );
+ }
+
+ #[cfg(not(windows))]
+ #[tokio::test]
+ async fn test_file_io_local_cache_invalidates_after_successful_write() {
+ let source_directory = tempfile::tempdir().unwrap();
+ let cache_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 file_io = setup_cached_fs_file_io(cache_directory.path());
+
+ assert_eq!(
+ file_io.new_input(&location).unwrap().read().await.unwrap(),
+ Bytes::from_static(b"old metadata")
+ );
+ file_io
+ .new_output(&location)
+ .unwrap()
+ .write(Bytes::from_static(b"new metadata"))
+ .await
+ .unwrap();
+ assert_eq!(
+ file_io.new_input(&location).unwrap().read().await.unwrap(),
+ Bytes::from_static(b"new metadata")
+ );
+ }
+
+ #[cfg(not(windows))]
+ #[tokio::test]
+ async fn test_file_io_local_cache_invalidates_equivalent_local_path_alias() {
+ let source_directory = tempfile::tempdir().unwrap();
+ let cache_directory = tempfile::tempdir().unwrap();
+ let source_path = source_directory.path().join("snapshot-1");
+ std::fs::write(&source_path, b"old metadata").unwrap();
+ let file_location = format!("file:{}", source_path.display());
+ let absolute_location = source_path.to_string_lossy();
+ let file_io = setup_cached_fs_file_io(cache_directory.path());
+
+ assert_eq!(
+ file_io
+ .new_input(&file_location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ Bytes::from_static(b"old metadata")
+ );
+ file_io
+ .new_output(absolute_location.as_ref())
+ .unwrap()
+ .write(Bytes::from_static(b"new metadata"))
+ .await
+ .unwrap();
+
+ assert_eq!(
+ file_io
+ .new_input(&file_location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ Bytes::from_static(b"new metadata")
+ );
+ }
+
+ #[cfg(not(windows))]
+ #[tokio::test]
+ async fn test_file_io_local_cache_invalidates_after_delete() {
+ let source_directory = tempfile::tempdir().unwrap();
+ let cache_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 file_io = setup_cached_fs_file_io(cache_directory.path());
+
+ assert_eq!(
+ file_io.new_input(&location).unwrap().read().await.unwrap(),
+ Bytes::from_static(b"old metadata")
+ );
+ file_io.delete_file(&location).await.unwrap();
+ std::fs::write(&source_path, b"new metadata").unwrap();
+ assert_eq!(
+ file_io.new_input(&location).unwrap().read().await.unwrap(),
+ Bytes::from_static(b"new metadata")
+ );
+ }
+
+ #[cfg(not(windows))]
+ #[tokio::test]
+ async fn test_file_io_local_cache_invalidates_after_delete_directory() {
+ let source_directory = tempfile::tempdir().unwrap();
+ let cache_directory = tempfile::tempdir().unwrap();
+ let snapshot_directory = source_directory.path().join("snapshot");
+ std::fs::create_dir(&snapshot_directory).unwrap();
+ let source_path = snapshot_directory.join("snapshot-1");
+ std::fs::write(&source_path, b"old metadata").unwrap();
+ let location = format!("file:{}", source_path.display());
+ let directory_location = format!("file:{}", snapshot_directory.display());
+ let file_io = setup_cached_fs_file_io(cache_directory.path());
+
+ assert_eq!(
+ file_io.new_input(&location).unwrap().read().await.unwrap(),
+ Bytes::from_static(b"old metadata")
+ );
+ file_io.delete_dir(&directory_location).await.unwrap();
+ std::fs::create_dir(&snapshot_directory).unwrap();
+ std::fs::write(&source_path, b"new metadata").unwrap();
+ assert_eq!(
+ file_io.new_input(&location).unwrap().read().await.unwrap(),
+ Bytes::from_static(b"new metadata")
+ );
+ }
+
+ #[cfg(not(windows))]
+ #[tokio::test]
+ async fn test_file_io_local_cache_invalidates_copy_target() {
+ let source_directory = tempfile::tempdir().unwrap();
+ let cache_directory = tempfile::tempdir().unwrap();
+ let source_path = source_directory.path().join("snapshot-1");
+ let target_path = source_directory.path().join("snapshot-2");
+ std::fs::write(&source_path, b"source value").unwrap();
+ std::fs::write(&target_path, b"stale target").unwrap();
+ let source_location = format!("file:{}", source_path.display());
+ let target_location = format!("file:{}", target_path.display());
+ let file_io = setup_cached_fs_file_io(cache_directory.path());
+
+ assert_eq!(
+ file_io
+ .new_input(&target_location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ Bytes::from_static(b"stale target")
+ );
+ file_io
+ .copy_file(&source_location, &target_location)
+ .await
+ .unwrap();
+
+ assert_eq!(
+ file_io
+ .new_input(&target_location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ Bytes::from_static(b"source value")
+ );
+ }
+
+ #[cfg(not(windows))]
+ #[tokio::test]
+ async fn test_file_io_local_cache_invalidates_source_and_target_after_rename() {
+ let source_directory = tempfile::tempdir().unwrap();
+ let cache_directory = tempfile::tempdir().unwrap();
+ let source_path = source_directory.path().join("snapshot-1");
+ let target_path = source_directory.path().join("snapshot-2");
+ std::fs::write(&source_path, b"source value").unwrap();
+ std::fs::write(&target_path, b"target value").unwrap();
+ let source_location = format!("file:{}", source_path.display());
+ let target_location = format!("file:{}", target_path.display());
+ let file_io = setup_cached_fs_file_io(cache_directory.path());
+
+ assert_eq!(
+ file_io
+ .new_input(&source_location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ Bytes::from_static(b"source value")
+ );
+ assert_eq!(
+ file_io
+ .new_input(&target_location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ Bytes::from_static(b"target value")
+ );
+ file_io
+ .rename(&source_location, &target_location)
+ .await
+ .unwrap();
+ std::fs::write(&source_path, b"new source!!").unwrap();
+
+ assert_eq!(
+ file_io
+ .new_input(&target_location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ Bytes::from_static(b"source value")
+ );
+ assert_eq!(
+ file_io
+ .new_input(&source_location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ Bytes::from_static(b"new source!!")
+ );
+ }
+
+ #[cfg(not(windows))]
+ #[tokio::test]
+ async fn test_file_io_local_cache_invalidates_directories_after_rename() {
+ let source_directory = tempfile::tempdir().unwrap();
+ let cache_directory = tempfile::tempdir().unwrap();
+ let old_directory = source_directory.path().join("old");
+ let new_directory = source_directory.path().join("new");
+ std::fs::create_dir(&old_directory).unwrap();
+ std::fs::create_dir(&new_directory).unwrap();
+ let old_snapshot = old_directory.join("snapshot-1");
+ let new_snapshot = new_directory.join("snapshot-1");
+ std::fs::write(&old_snapshot, b"old directory").unwrap();
+ std::fs::write(&new_snapshot, b"new directory").unwrap();
+ let old_directory_location = format!("file:{}", old_directory.display());
+ let new_directory_location = format!("file:{}", new_directory.display());
+ let old_snapshot_location = format!("file:{}", old_snapshot.display());
+ let new_snapshot_location = format!("file:{}", new_snapshot.display());
+ let file_io = setup_cached_fs_file_io(cache_directory.path());
+
+ assert_eq!(
+ file_io
+ .new_input(&old_snapshot_location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ Bytes::from_static(b"old directory")
+ );
+ assert_eq!(
+ file_io
+ .new_input(&new_snapshot_location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ Bytes::from_static(b"new directory")
+ );
+ std::fs::remove_dir_all(&new_directory).unwrap();
+ file_io
+ .rename(&old_directory_location, &new_directory_location)
+ .await
+ .unwrap();
+ std::fs::create_dir(&old_directory).unwrap();
+ std::fs::write(&old_snapshot, b"replacement!!").unwrap();
+
+ assert_eq!(
+ file_io
+ .new_input(&new_snapshot_location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ Bytes::from_static(b"old directory")
+ );
+ assert_eq!(
+ file_io
+ .new_input(&old_snapshot_location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ Bytes::from_static(b"replacement!!")
+ );
+ }
+
+ #[cfg(not(windows))]
+ #[tokio::test]
+ async fn test_file_io_local_cache_invalidates_after_streaming_write_shutdown() {
+ use tokio::io::AsyncWriteExt;
+
+ let source_directory = tempfile::tempdir().unwrap();
+ let cache_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 file_io = setup_cached_fs_file_io(cache_directory.path());
+
+ assert_eq!(
+ file_io.new_input(&location).unwrap().read().await.unwrap(),
+ Bytes::from_static(b"old metadata")
+ );
+ let mut writer = file_io
+ .new_output(&location)
+ .unwrap()
+ .async_writer()
+ .await
+ .unwrap();
+ writer.write_all(b"new metadata").await.unwrap();
+ writer.shutdown().await.unwrap();
+
+ assert_eq!(
+ file_io.new_input(&location).unwrap().read().await.unwrap(),
+ Bytes::from_static(b"new metadata")
+ );
+ }
}
diff --git a/crates/paimon/src/io/mod.rs b/crates/paimon/src/io/mod.rs
index a16e7ff7..5a075bcd 100644
--- a/crates/paimon/src/io/mod.rs
+++ b/crates/paimon/src/io/mod.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+pub(crate) mod cache;
mod file_io;
pub use file_io::*;
diff --git a/crates/paimon/src/table/rest_env.rs b/crates/paimon/src/table/rest_env.rs
index b7771001..4f2a9df7 100644
--- a/crates/paimon/src/table/rest_env.rs
+++ b/crates/paimon/src/table/rest_env.rs
@@ -22,6 +22,7 @@ use crate::api::rest_error::RestError;
use crate::catalog::{Identifier, RESTTokenFileIO};
use crate::common::Options;
use crate::error::Error;
+use crate::io::cache::LocalCache;
use crate::io::FileIO;
use crate::spec::{CoreOptions, TableSchema, PATH_OPTION};
use crate::table::snapshot_commit::{RESTSnapshotCommit, SnapshotCommit};
@@ -38,6 +39,7 @@ pub struct RESTEnv {
api: Arc<RESTApi>,
options: Options,
data_token_enabled: bool,
+ local_cache: Option<Arc<LocalCache>>,
}
impl std::fmt::Debug for RESTEnv {
@@ -51,12 +53,13 @@ impl std::fmt::Debug for RESTEnv {
impl RESTEnv {
/// Create a new RESTEnv.
- pub fn new(
+ pub(crate) fn new(
identifier: Identifier,
uuid: String,
api: Arc<RESTApi>,
options: Options,
data_token_enabled: bool,
+ local_cache: Option<Arc<LocalCache>>,
) -> Self {
Self {
identifier,
@@ -64,9 +67,15 @@ impl RESTEnv {
api,
options,
data_token_enabled,
+ local_cache,
}
}
+ #[cfg(test)]
+ fn has_local_cache(&self) -> bool {
+ self.local_cache.is_some()
+ }
+
/// Get the REST API client.
pub fn api(&self) -> &Arc<RESTApi> {
&self.api
@@ -84,6 +93,7 @@ impl RESTEnv {
self.api.clone(),
self.options.clone(),
self.data_token_enabled,
+ self.local_cache.clone(),
)
.await
}
@@ -94,6 +104,7 @@ impl RESTEnv {
api: Arc<RESTApi>,
options: Options,
data_token_enabled: bool,
+ local_cache: Option<Arc<LocalCache>>,
) -> Result<Table> {
let response = api
.get_table(identifier)
@@ -142,16 +153,31 @@ impl RESTEnv {
})?;
let file_io = if data_token_enabled && !is_external {
- RESTTokenFileIO::new(identifier.clone(), table_path.clone(), options.clone())
- .build_file_io()
- .await?
+ RESTTokenFileIO::new(
+ identifier.clone(),
+ table_path.clone(),
+ options.clone(),
+ local_cache.clone(),
+ )
+ .build_file_io()
+ .await?
} else {
let mut builder = FileIO::from_path(&table_path)?;
builder = builder.with_props(options.to_map());
+ if let Some(local_cache) = &local_cache {
+ builder = builder.with_local_cache(local_cache.clone());
+ }
builder.build()?
};
- let rest_env = RESTEnv::new(identifier.clone(), uuid, api, options, data_token_enabled);
+ let rest_env = RESTEnv::new(
+ identifier.clone(),
+ uuid,
+ api,
+ options,
+ data_token_enabled,
+ local_cache,
+ );
Ok(Table::new(
file_io,
@@ -187,3 +213,39 @@ fn map_rest_error_for_table(err: Error, identifier: &Identifier) -> Error {
other => other,
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::common::CatalogOptions;
+ use crate::io::cache::create_local_cache;
+
+ #[tokio::test]
+ async fn test_rest_env_clones_catalog_local_cache() {
+ let cache_directory = tempfile::tempdir().unwrap();
+ let mut options = Options::new();
+ options.set(CatalogOptions::URI, "http://localhost:1");
+ options.set(CatalogOptions::WAREHOUSE, "test-warehouse");
+ options.set(CatalogOptions::TOKEN_PROVIDER, "bear");
+ options.set(CatalogOptions::TOKEN, "test-token");
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ options.set(
+ CatalogOptions::LOCAL_CACHE_DIR,
+ cache_directory.path().to_string_lossy(),
+ );
+ let local_cache = create_local_cache(&options).unwrap();
+ let api = Arc::new(RESTApi::new(options.clone(), false).await.unwrap());
+
+ let rest_env = RESTEnv::new(
+ Identifier::new("database", "table"),
+ "uuid".to_string(),
+ api,
+ options,
+ false,
+ local_cache,
+ );
+
+ assert!(rest_env.has_local_cache());
+ assert!(rest_env.clone().has_local_cache());
+ }
+}
diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md
index 2d43fadc..92728d37 100644
--- a/docs/src/getting-started.md
+++ b/docs/src/getting-started.md
@@ -140,6 +140,47 @@ Supported metastore types:
| `filesystem` | Local or remote filesystem (default) |
| `rest` | REST catalog server |
+### Local Disk Cache
+
+Catalogs can cache immutable Paimon metadata and index file blocks on local
+disk. The cache works with both filesystem and REST catalogs:
+
+```rust
+let mut options = Options::new();
+options.set(CatalogOptions::WAREHOUSE, "s3://bucket/warehouse");
+options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+options.set(CatalogOptions::LOCAL_CACHE_DIR, "/var/cache/paimon/worker-1");
+options.set(CatalogOptions::LOCAL_CACHE_MAX_SIZE, "20 GiB");
+options.set(CatalogOptions::LOCAL_CACHE_BLOCK_SIZE, "1 MiB");
+options.set(
+ CatalogOptions::LOCAL_CACHE_WHITELIST,
+ "meta,global-index",
+);
+let catalog = CatalogFactory::create(options).await?;
+```
+
+| Option | Default | Description |
+|--------|---------|-------------|
+| `local-cache.enabled` | `false` | Enable catalog-scoped local block caching. |
+| `local-cache.dir` | none | Base cache directory; required when caching is enabled. Paimon stores entries in a private versioned child directory. |
+| `local-cache.max-size` | unlimited | Maximum encoded disk usage. Values accept byte units such as `512 MiB` or `20 GiB`. |
+| `local-cache.block-size` | `1 MiB` | Block size used for cached range reads. |
+| `local-cache.whitelist` | `meta,global-index` | Comma-separated eligible types: `meta`, `global-index`, `bucket-index`, `data`, and `file-index`. |
+
+The cache is disk-only and is reused after process restarts. Cache keys include
+a catalog-configuration fingerprint and the canonical storage object path, so
+catalogs can safely use the same base directory without reading one another's
+entries. Runtime cache read, write, validation, and eviction failures are
+fail-open: the original storage remains the source of truth. Paimon mutable
+markers and temporary files always bypass the cache; other eligible files rely
+on Paimon's immutable-file convention. Cache managers using the same canonical
+directory in one process share LRU and size accounting; if their configured
+limits differ, the smallest `local-cache.max-size` is used. Restart recovery
+runs on a blocking worker, reads only block headers and file metadata, and
+validates payload CRC lazily on the first hit. Use a separate `local-cache.dir`
+for each worker or process because processes do not share exact LRU or size
+accounting.
+
### Manage Databases
```rust