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 e4ec3a45 [core] Add opt-in node-shared disk block cache (#1175)
e4ec3a45 is described below
commit e4ec3a45e229ab856cbae59a4733b9fad3e5f76e
Author: XiaoHongbo <[email protected]>
AuthorDate: Tue Oct 20 22:43:23 2026 +0800
[core] Add opt-in node-shared disk block cache (#1175)
---
crates/paimon/src/catalog/factory.rs | 59 +-
crates/paimon/src/catalog/rest/rest_catalog.rs | 55 +-
crates/paimon/src/common/options.rs | 2 +
crates/paimon/src/io/cache/disk.rs | 3 +
crates/paimon/src/io/cache/mod.rs | 153 ++++-
crates/paimon/src/io/cache/reader.rs | 19 +-
crates/paimon/src/io/cache/shared.rs | 888 +++++++++++++++++++++++++
crates/paimon/src/io/cache/state.rs | 2 +
crates/paimon/src/io/file_io.rs | 327 ++++++++-
crates/paimon/src/table/rest_env.rs | 5 +-
docs/src/getting-started.md | 15 +-
11 files changed, 1493 insertions(+), 35 deletions(-)
diff --git a/crates/paimon/src/catalog/factory.rs
b/crates/paimon/src/catalog/factory.rs
index c531ccd5..81ee7059 100644
--- a/crates/paimon/src/catalog/factory.rs
+++ b/crates/paimon/src/catalog/factory.rs
@@ -89,7 +89,7 @@ impl CatalogFactory {
.unwrap_or(METASTORE_FILESYSTEM);
match metastore {
- METASTORE_FILESYSTEM => Self::create_filesystem_catalog(options),
+ METASTORE_FILESYSTEM =>
Self::create_filesystem_catalog(options).await,
METASTORE_REST => Self::create_rest_catalog(options).await,
_ => ConfigInvalidSnafu {
message: format!(
@@ -101,8 +101,13 @@ impl CatalogFactory {
}
/// Create a filesystem catalog.
- fn create_filesystem_catalog(options: Options) -> Result<Arc<dyn Catalog>>
{
- let catalog = FileSystemCatalog::new(options)?;
+ async fn create_filesystem_catalog(options: Options) -> Result<Arc<dyn
Catalog>> {
+ let catalog = tokio::task::spawn_blocking(move ||
FileSystemCatalog::new(options))
+ .await
+ .map_err(|error| crate::Error::UnexpectedError {
+ message: "Filesystem catalog initialization task
failed".into(),
+ source: Some(Box::new(error)),
+ })??;
Ok(Arc::new(catalog))
}
@@ -140,6 +145,54 @@ mod tests {
assert!(result.is_ok());
}
+ #[test]
+ fn test_shared_cache_factory_init_does_not_block_current_thread_runtime() {
+ use std::fs::File;
+ use std::sync::mpsc;
+ use std::time::Duration;
+
+ let directory = tempfile::tempdir().unwrap();
+ let mut options = Options::new();
+ options.set(
+ CatalogOptions::WAREHOUSE,
+ directory.path().join("warehouse").to_str().unwrap(),
+ );
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ options.set(CatalogOptions::LOCAL_CACHE_SHARED, "true");
+ options.set(CatalogOptions::LOCAL_CACHE_MAX_SIZE, "1 MiB");
+ options.set(
+ CatalogOptions::LOCAL_CACHE_DIR,
+ directory.path().to_str().unwrap(),
+ );
+ drop(FileSystemCatalog::new(options.clone()).unwrap());
+ let gate = File::open(
+ directory
+ .path()
+ .join("paimon-shared-local-cache-v1/directory.lock"),
+ )
+ .unwrap();
+ gate.lock().unwrap();
+
+ let (tx, rx) = mpsc::channel();
+ let catalog_thread = std::thread::spawn(move || {
+ let runtime = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ runtime.block_on(async move {
+ tokio::spawn(async move {
+ tokio::time::sleep(Duration::from_millis(50)).await;
+ tx.send(()).unwrap();
+ });
+ CatalogFactory::create(options).await.unwrap();
+ });
+ });
+ let runtime_progressed =
rx.recv_timeout(Duration::from_secs(2)).is_ok();
+ drop(gate);
+ catalog_thread.join().unwrap();
+ assert!(runtime_progressed, "catalog factory blocked the runtime");
+ }
+
#[tokio::test]
async fn test_missing_warehouse_option() {
let options = Options::new();
diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs
b/crates/paimon/src/catalog/rest/rest_catalog.rs
index de88f8d6..661f6d1a 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -36,7 +36,7 @@ use crate::catalog::{
};
use crate::common::{CatalogOptions, Options};
use crate::error::Error;
-use crate::io::cache::{create_local_cache_with_namespace, LocalCache};
+use crate::io::cache::{create_local_cache_with_namespace_async, LocalCache};
use crate::io::FileIOCacheContext;
use crate::spec::{Partition, PartitionStatistics, Schema, SchemaChange};
use crate::table::{RESTEnv, Table};
@@ -90,7 +90,7 @@ impl RESTCatalog {
.unwrap_or(false);
let api_options = api.options().clone();
- let local_cache = create_local_cache_with_namespace(&options,
&api_options)?;
+ let local_cache = create_local_cache_with_namespace_async(&options,
&api_options).await?;
let file_io_cache =
FileIOCacheContext::from_props(api_options.to_map())?;
let fuse_resolver =
Arc::new(super::FusePathResolver::new(&api_options));
@@ -1229,6 +1229,57 @@ mod tests {
assert!(catalog.has_local_cache());
}
+ #[test]
+ fn test_shared_cache_catalog_init_does_not_block_current_thread_runtime() {
+ use std::fs::File;
+ use std::sync::mpsc;
+ use std::time::Duration;
+
+ let directory = 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_SHARED, "true");
+ options.set(CatalogOptions::LOCAL_CACHE_MAX_SIZE, "1 MiB");
+ options.set(
+ CatalogOptions::LOCAL_CACHE_DIR,
+ directory.path().to_str().unwrap(),
+ );
+ drop(crate::io::cache::create_local_cache_with_namespace(&options,
&options).unwrap());
+ let gate = File::open(
+ directory
+ .path()
+ .join("paimon-shared-local-cache-v1/directory.lock"),
+ )
+ .unwrap();
+ gate.lock().unwrap();
+
+ let (tx, rx) = mpsc::channel();
+ let catalog_thread = std::thread::spawn(move || {
+ let runtime = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ runtime.block_on(async move {
+ tokio::spawn(async move {
+ tokio::time::sleep(Duration::from_millis(50)).await;
+ tx.send(()).unwrap();
+ });
+ RESTCatalog::new(options, false).await.unwrap();
+ });
+ });
+ let runtime_progressed =
rx.recv_timeout(Duration::from_secs(2)).is_ok();
+ drop(gate);
+ catalog_thread.join().unwrap();
+ assert!(
+ runtime_progressed,
+ "catalog initialization blocked the runtime"
+ );
+ }
+
#[test]
fn test_tag_error_mapping() {
let identifier = Identifier::new("db", "table");
diff --git a/crates/paimon/src/common/options.rs
b/crates/paimon/src/common/options.rs
index c75f1954..eeefb7d0 100644
--- a/crates/paimon/src/common/options.rs
+++ b/crates/paimon/src/common/options.rs
@@ -81,6 +81,8 @@ impl CatalogOptions {
/// Directory for the local disk block cache. If unset, an in-memory cache
is used.
pub const LOCAL_CACHE_DIR: &'static str = "local-cache.dir";
+ /// Share a node-local disk cache between cooperating processes.
+ pub const LOCAL_CACHE_SHARED: &'static str = "local-cache.shared";
/// Maximum total size of the local block cache.
pub const LOCAL_CACHE_MAX_SIZE: &'static str = "local-cache.max-size";
diff --git a/crates/paimon/src/io/cache/disk.rs
b/crates/paimon/src/io/cache/disk.rs
index b731e303..3b217e04 100644
--- a/crates/paimon/src/io/cache/disk.rs
+++ b/crates/paimon/src/io/cache/disk.rs
@@ -28,6 +28,9 @@ use super::priority_lru::DEFAULT_HIGH_PRIORITY_POOL_RATIO;
use super::priority_lru::{protected_limit, PriorityLru};
use super::state::{BlockKey, CacheCoordinator, LogicalPath};
+#[path = "shared.rs"]
+pub(super) mod shared;
+
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;
diff --git a/crates/paimon/src/io/cache/mod.rs
b/crates/paimon/src/io/cache/mod.rs
index 42eb32af..701f305b 100644
--- a/crates/paimon/src/io/cache/mod.rs
+++ b/crates/paimon/src/io/cache/mod.rs
@@ -33,6 +33,7 @@ use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
+use disk::shared::SharedDiskCache;
use disk::DiskCache;
pub(super) use reader::CachedFileReader;
@@ -53,6 +54,7 @@ pub(crate) struct LocalCache {
enum CacheBackend {
Memory(MemoryCache),
Disk(Arc<DiskCache>),
+ SharedDisk(SharedDiskCache),
External(Arc<dyn FileBlockCache>),
}
@@ -89,7 +91,23 @@ impl LocalCache {
.and_then(|capacity| usize::try_from(capacity).ok())
.unwrap_or(DEFAULT_FILE_SIZE_CAPACITY)
.clamp(1, DEFAULT_FILE_SIZE_CAPACITY);
- let (backend, coordinator) = if let Some(dir) = config.dir {
+ let (backend, coordinator) = if config.shared {
+ (
+ CacheBackend::SharedDisk(SharedDiskCache::new_with_ratio(
+ config
+ .dir
+ .ok_or_else(|| crate::Error::ConfigInvalid {
+ message: "Shared local cache requires
local-cache.dir".into(),
+ })?
+ .join("paimon-shared-local-cache-v1"),
+ config.max_size.ok_or_else(|| crate::Error::ConfigInvalid {
+ message: "Shared local cache requires
local-cache.max-size".into(),
+ })?,
+ ratio,
+ )?),
+ Arc::new(CacheCoordinator::default()),
+ )
+ } else if let Some(dir) = config.dir {
let disk = DiskCache::shared(dir.join(CACHE_DIRECTORY_NAME),
config.max_size, ratio)?;
let coordinator = disk.coordinator();
(CacheBackend::Disk(disk), coordinator)
@@ -135,6 +153,7 @@ impl LocalCache {
let payload = match &self.backend {
CacheBackend::Memory(memory) => memory.get_block(key),
CacheBackend::Disk(disk) => disk.get_block(key).await,
+ CacheBackend::SharedDisk(disk) => disk.get(key,
&token.shared_generation).await,
CacheBackend::External(cache) => {
let start = key.block_index.checked_mul(key.block_size)?;
let length = u64::try_from(expected_len).ok()?;
@@ -158,6 +177,9 @@ impl LocalCache {
match &self.backend {
CacheBackend::Memory(memory) => memory.put_block(key, payload),
CacheBackend::Disk(disk) => disk.put_block(key, payload).await,
+ CacheBackend::SharedDisk(disk) => {
+ disk.put(key, payload, &token.shared_generation).await
+ }
CacheBackend::External(cache) => {
if let Some(offset) =
key.block_index.checked_mul(key.block_size) {
cache.put(&key.path, offset, payload).await;
@@ -170,12 +192,28 @@ impl LocalCache {
match &self.backend {
CacheBackend::Memory(memory) => memory.remove_block(key),
CacheBackend::Disk(disk) => disk.remove_block(key).await,
+ CacheBackend::SharedDisk(disk) => {
+ if let Err(error) = disk.invalidate().await {
+ log::debug!("Failed to evict shared cache block: {error}");
+ }
+ }
CacheBackend::External(cache) =>
cache.invalidate_path(&key.path).await,
}
}
pub(super) fn read_token(&self, path: &str) -> CacheReadToken {
- self.coordinator.read_token(&self.namespace, path)
+ let mut token = self.coordinator.read_token(&self.namespace, path);
+ if let CacheBackend::SharedDisk(disk) = &self.backend {
+ token.shared_generation = disk.generation();
+ }
+ token
+ }
+
+ async fn shared_load_lock(&self, key: &BlockKey) -> Option<std::fs::File> {
+ match &self.backend {
+ CacheBackend::SharedDisk(disk) => disk.load_lock(key).await,
+ _ => None,
+ }
}
async fn block_load_lock(&self, key: &BlockKey) ->
Arc<tokio::sync::Mutex<()>> {
@@ -187,6 +225,10 @@ impl LocalCache {
}
pub(super) async fn file_size(&self, path: &str, token: &CacheReadToken)
-> Option<u64> {
+ // Process-local file sizes cannot observe another process's
invalidation.
+ if matches!(self.backend, CacheBackend::SharedDisk(_)) {
+ return None;
+ }
let _prefix_guard = self.coordinator.prefix_read_guard().await;
let _publish_guard = token.publish_guard().await;
token
@@ -196,6 +238,9 @@ impl LocalCache {
}
pub(super) async fn put_file_size(&self, path: &str, size: u64, token:
&CacheReadToken) {
+ if matches!(self.backend, CacheBackend::SharedDisk(_)) {
+ return;
+ }
let _prefix_guard = self.coordinator.prefix_read_guard().await;
let _publish_guard = token.publish_guard().await;
if !token.is_current() {
@@ -205,7 +250,7 @@ impl LocalCache {
.put_file_size(&self.namespace, path, size,
self.file_size_capacity);
}
- pub(super) async fn invalidate_path(&self, path: &str) {
+ pub(super) async fn invalidate_path(&self, path: &str) ->
crate::Result<()> {
let _guard = self
.coordinator
.begin_path_invalidation(&self.namespace, path)
@@ -213,11 +258,15 @@ impl LocalCache {
match &self.backend {
CacheBackend::Memory(memory) =>
memory.invalidate_path(&self.namespace, path),
CacheBackend::Disk(disk) => disk.invalidate_path(&self.namespace,
path).await,
+ CacheBackend::SharedDisk(disk) => {
+ disk.invalidate().await.map_err(invalidation_error)?
+ }
CacheBackend::External(cache) => cache.invalidate_path(path).await,
}
+ Ok(())
}
- pub(super) async fn invalidate_prefix(&self, prefix: &str) {
+ pub(super) async fn invalidate_prefix(&self, prefix: &str) ->
crate::Result<()> {
if let CacheBackend::Disk(disk) = &self.backend {
disk.ensure_recovered().await;
}
@@ -228,8 +277,35 @@ impl LocalCache {
match &self.backend {
CacheBackend::Memory(memory) =>
memory.invalidate_prefix(&self.namespace, prefix),
CacheBackend::Disk(disk) =>
disk.invalidate_prefix(&self.namespace, prefix).await,
+ CacheBackend::SharedDisk(disk) => {
+ disk.invalidate().await.map_err(invalidation_error)?
+ }
CacheBackend::External(cache) =>
cache.invalidate_prefix(prefix).await,
}
+ Ok(())
+ }
+
+ pub(in crate::io) async fn begin_shared_invalidation(
+ &self,
+ ) -> crate::Result<Option<disk::shared::SharedInvalidationGuard>> {
+ match &self.backend {
+ CacheBackend::SharedDisk(disk) => {
+ disk.begin_invalidation().await.map(Some).map_err(|error| {
+ crate::Error::UnexpectedError {
+ message: "Failed to establish shared cache mutation
fence".into(),
+ source: Some(Box::new(error)),
+ }
+ })
+ }
+ _ => Ok(None),
+ }
+ }
+}
+
+fn invalidation_error(error: std::io::Error) -> crate::Error {
+ crate::Error::UnexpectedError {
+ message: "Shared cache invalidation failed; storage may already have
been modified".into(),
+ source: Some(Box::new(error)),
}
}
@@ -271,8 +347,34 @@ fn parse_high_priority_ratio(options: &Options) ->
crate::Result<f64> {
Ok(ratio)
}
+pub(crate) async fn create_local_cache_with_namespace_async(
+ cache_options: &Options,
+ namespace_options: &Options,
+) -> crate::Result<Option<Arc<LocalCache>>> {
+ let Some(config) =
+ LocalCacheConfig::from_options_with_namespace(cache_options,
namespace_options)?
+ else {
+ return Ok(None);
+ };
+ let ratio = parse_high_priority_ratio(cache_options)?;
+ if config.shared {
+ tokio::task::spawn_blocking(move || LocalCache::new_with_ratio(config,
ratio).map(Arc::new))
+ .await
+ .map_err(|error| crate::Error::UnexpectedError {
+ message: "Shared local cache initialization task
failed".into(),
+ source: Some(Box::new(error)),
+ })?
+ .map(Some)
+ } else {
+ LocalCache::new_with_ratio(config, ratio)
+ .map(Arc::new)
+ .map(Some)
+ }
+}
+
#[derive(Debug)]
pub(crate) struct LocalCacheConfig {
+ shared: bool,
dir: Option<PathBuf>,
namespace: String,
max_size: Option<u64>,
@@ -338,6 +440,19 @@ impl LocalCacheConfig {
.map(String::as_str)
.unwrap_or("meta,global-index");
Ok(Some(Self {
+ shared: match options
+ .get(CatalogOptions::LOCAL_CACHE_SHARED)
+ .map(|v| v.trim())
+ {
+ None => false,
+ Some(v) if v.eq_ignore_ascii_case("true") => true,
+ Some(v) if v.eq_ignore_ascii_case("false") => false,
+ Some(_) => {
+ return Err(crate::Error::ConfigInvalid {
+ message: "Invalid boolean for
local-cache.shared".into(),
+ })
+ }
+ },
dir,
namespace: catalog_namespace(namespace_options),
max_size,
@@ -521,6 +636,7 @@ mod tests {
async fn test_local_cache_file_size_is_removed_with_path_invalidation() {
let directory = tempfile::tempdir().unwrap();
let cache = LocalCache::new(LocalCacheConfig {
+ shared: false,
dir: Some(directory.path().to_path_buf()),
namespace: "test".to_string(),
max_size: None,
@@ -534,7 +650,7 @@ mod tests {
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;
+ cache.invalidate_path(path).await.unwrap();
assert_eq!(cache.file_size(path, &token).await, None);
}
@@ -560,10 +676,29 @@ mod tests {
}
}
+ #[test]
+ fn test_shared_cache_requires_directory_and_explicit_capacity() {
+ let directory = tempfile::tempdir().unwrap();
+ let mut options = Options::default();
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ options.set(CatalogOptions::LOCAL_CACHE_SHARED, "true");
+ assert!(create_local_cache(&options).is_err());
+ options.set(
+ CatalogOptions::LOCAL_CACHE_DIR,
+ directory.path().to_str().unwrap(),
+ );
+ assert!(create_local_cache(&options).is_err());
+ options.set(CatalogOptions::LOCAL_CACHE_MAX_SIZE, "1 MiB");
+ assert!(create_local_cache(&options).unwrap().is_some());
+ options.set(CatalogOptions::LOCAL_CACHE_SHARED, "invalid");
+ assert!(create_local_cache(&options).is_err());
+ }
+
#[tokio::test]
async fn
test_local_cache_file_size_is_invalidated_across_shared_instances() {
let directory = tempfile::tempdir().unwrap();
let config = || LocalCacheConfig {
+ shared: false,
dir: Some(directory.path().to_path_buf()),
namespace: "test".to_string(),
max_size: None,
@@ -577,7 +712,7 @@ mod tests {
first.put_file_size(path, 42, &token).await;
assert_eq!(second.file_size(path, &token).await, Some(42));
- second.invalidate_path(path).await;
+ second.invalidate_path(path).await.unwrap();
let current_token = first.read_token(path);
assert_eq!(first.file_size(path, ¤t_token).await, None);
}
@@ -586,6 +721,7 @@ mod tests {
async fn test_stale_file_size_cannot_republish_after_invalidation() {
let directory = tempfile::tempdir().unwrap();
let config = || LocalCacheConfig {
+ shared: false,
dir: Some(directory.path().to_path_buf()),
namespace: "test".to_string(),
max_size: None,
@@ -598,7 +734,7 @@ mod tests {
let stale_token = first.read_token(path);
assert_eq!(first.file_size(path, &stale_token).await, None);
- second.invalidate_path(path).await;
+ second.invalidate_path(path).await.unwrap();
first.put_file_size(path, 42, &stale_token).await;
let current_token = first.read_token(path);
@@ -609,6 +745,7 @@ mod tests {
fn test_local_cache_uses_whitelist_and_bypasses_mutable_files() {
let directory = tempfile::tempdir().unwrap();
let cache = LocalCache::new(LocalCacheConfig {
+ shared: false,
dir: Some(directory.path().to_path_buf()),
namespace: "test".to_string(),
max_size: None,
@@ -638,6 +775,7 @@ mod tests {
std::fs::write(&nested, b"nested foreign").unwrap();
LocalCache::new(LocalCacheConfig {
+ shared: false,
dir: Some(directory.path().to_path_buf()),
namespace: "test".to_string(),
max_size: None,
@@ -658,6 +796,7 @@ mod tests {
async fn test_local_cache_bounds_file_size_entries() {
let directory = tempfile::tempdir().unwrap();
let cache = LocalCache::new(LocalCacheConfig {
+ shared: false,
dir: Some(directory.path().to_path_buf()),
namespace: "test".to_string(),
max_size: Some(8),
diff --git a/crates/paimon/src/io/cache/reader.rs
b/crates/paimon/src/io/cache/reader.rs
index edb00a0a..bcb83ddd 100644
--- a/crates/paimon/src/io/cache/reader.rs
+++ b/crates/paimon/src/io/cache/reader.rs
@@ -83,6 +83,7 @@ impl CachedFileReader {
let load_lock = self.cache.block_load_lock(&key).await;
let load_guard = load_lock.lock().await;
+ let _shared_guard = self.cache.shared_load_lock(&key).await;
let result = if let Some(payload) = self
.cache
.get_block(&key, expected_len, &self.read_token)
@@ -140,6 +141,7 @@ impl CachedFileReader {
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 _shared_guard = self.cache.shared_load_lock(&first_key).await;
let result = if let Some(payload) = self.read_cached_full().await {
Ok(payload)
} else {
@@ -270,6 +272,7 @@ mod tests {
});
let cache = Arc::new(
LocalCache::new(LocalCacheConfig {
+ shared: false,
dir: None,
namespace: "test".to_string(),
max_size: None,
@@ -305,6 +308,7 @@ mod tests {
});
let cache = Arc::new(
LocalCache::new(LocalCacheConfig {
+ shared: false,
dir: None,
namespace: "test".to_string(),
max_size: Some(16),
@@ -360,6 +364,7 @@ mod tests {
});
let cache = Arc::new(
LocalCache::new(LocalCacheConfig {
+ shared: false,
dir: None,
namespace: "test".to_string(),
max_size: None,
@@ -390,6 +395,7 @@ mod tests {
reads: AtomicUsize::new(0),
});
let config = || LocalCacheConfig {
+ shared: false,
dir: Some(directory.path().to_path_buf()),
namespace: "test".to_string(),
max_size: None,
@@ -437,6 +443,7 @@ mod tests {
let path = "s3://bucket/table/snapshot/snapshot-1";
let cache = Arc::new(
LocalCache::new(LocalCacheConfig {
+ shared: false,
dir: None,
namespace: "test".to_string(),
max_size: None,
@@ -460,7 +467,7 @@ mod tests {
let old_load = tokio::spawn(async move { old_reader.read(0..4).await
});
started.notified().await;
- cache.invalidate_path(path).await;
+ cache.invalidate_path(path).await.unwrap();
release.notify_one();
assert_eq!(
old_load.await.unwrap().unwrap(),
@@ -486,6 +493,7 @@ mod tests {
let path = "s3://bucket/table/snapshot/snapshot-1";
let cache_a = Arc::new(
LocalCache::new(LocalCacheConfig {
+ shared: false,
dir: Some(directory.path().to_path_buf()),
namespace: "test".to_string(),
max_size: None,
@@ -496,6 +504,7 @@ mod tests {
);
let cache_b = Arc::new(
LocalCache::new(LocalCacheConfig {
+ shared: false,
dir: Some(directory.path().to_path_buf()),
namespace: "test".to_string(),
max_size: None,
@@ -519,7 +528,7 @@ mod tests {
let old_load = tokio::spawn(async move { old_reader.read(0..4).await
});
started.notified().await;
- cache_b.invalidate_path(path).await;
+ cache_b.invalidate_path(path).await.unwrap();
release.notify_one();
assert_eq!(
old_load.await.unwrap().unwrap(),
@@ -547,6 +556,7 @@ mod tests {
let observed_path = "s3://bucket/table/snapshot/snapshot-2";
let blocked_path = "s3://bucket/table/snapshot/snapshot-3";
let config = || LocalCacheConfig {
+ shared: false,
dir: Some(directory.path().to_path_buf()),
namespace: "test".to_string(),
max_size: None,
@@ -613,7 +623,7 @@ mod tests {
"file-size lookup completed before prefix invalidation removed the
old entry"
);
drop(blocked_guard);
- invalidation.await.unwrap();
+ invalidation.await.unwrap().unwrap();
assert_eq!(
current_load.await.unwrap().unwrap(),
Bytes::from_static(b"new!")
@@ -693,6 +703,7 @@ mod tests {
});
let cache = Arc::new(
LocalCache::new(LocalCacheConfig {
+ shared: false,
dir: Some(directory.path().to_path_buf()),
namespace: "test".to_string(),
max_size: None,
@@ -727,6 +738,7 @@ mod tests {
let directory = tempfile::tempdir().unwrap();
let cache = Arc::new(
LocalCache::new(LocalCacheConfig {
+ shared: false,
dir: Some(directory.path().to_path_buf()),
namespace: "test".to_string(),
max_size: None,
@@ -748,6 +760,7 @@ mod tests {
});
let cache = Arc::new(
LocalCache::new(LocalCacheConfig {
+ shared: false,
dir: None,
namespace: "test".to_string(),
max_size: None,
diff --git a/crates/paimon/src/io/cache/shared.rs
b/crates/paimon/src/io/cache/shared.rs
new file mode 100644
index 00000000..0ace0c0e
--- /dev/null
+++ b/crates/paimon/src/io/cache/shared.rs
@@ -0,0 +1,888 @@
+// 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.
+
+//! Opt-in node-local cache. Block files, not a process-local index, are
authoritative.
+//! A shared directory lock protects cache hits, while an exclusive lock
serializes
+//! publication, eviction, and invalidation. File-striped locks coalesce
downloads.
+
+use super::*;
+use std::fs::{File, OpenOptions};
+use std::io::{self, Write};
+use std::time::Duration;
+
+const MUTATION_MARKER: &str = "mutation-pending";
+
+#[derive(Debug)]
+pub(crate) struct SharedDiskCache {
+ root: PathBuf,
+ max_size: u64,
+ high_priority_ratio: f64,
+}
+
+fn open_lock(path: &Path) -> io::Result<File> {
+ let mut options = OpenOptions::new();
+ options.read(true).write(true).create(true).truncate(false);
+ #[cfg(unix)]
+ {
+ use std::os::unix::fs::OpenOptionsExt;
+ options.mode(0o600);
+ }
+ options.open(path)
+}
+
+/// Try-lock polling avoids occupying a blocking thread while another process
+/// performs remote I/O. Dropping the file (including cancellation/exit)
releases it.
+async fn lock(path: PathBuf) -> Option<File> {
+ let file = tokio::task::spawn_blocking(move || open_lock(&path))
+ .await
+ .ok()?
+ .ok()?;
+ let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
+ loop {
+ match file.try_lock() {
+ Ok(()) => return Some(file),
+ Err(std::fs::TryLockError::WouldBlock) => {
+ if tokio::time::Instant::now() >= deadline {
+ return None;
+ }
+ tokio::time::sleep(Duration::from_millis(10)).await;
+ }
+ Err(std::fs::TryLockError::Error(error)) => {
+ log::debug!("Shared cache lock unavailable: {error}");
+ return None;
+ }
+ }
+ }
+}
+
+async fn read_lock(path: PathBuf) -> Option<File> {
+ let file = tokio::task::spawn_blocking(move || open_lock(&path))
+ .await
+ .ok()?
+ .ok()?;
+ let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
+ loop {
+ match File::try_lock_shared(&file) {
+ Ok(()) => return Some(file),
+ Err(std::fs::TryLockError::WouldBlock) => {
+ if tokio::time::Instant::now() >= deadline {
+ return None;
+ }
+ tokio::time::sleep(Duration::from_millis(10)).await;
+ }
+ Err(std::fs::TryLockError::Error(error)) => {
+ log::debug!("Shared cache read lock unavailable: {error}");
+ return None;
+ }
+ }
+ }
+}
+
+/// Mutating storage after a cache invalidation timeout can leave cached data
+/// reachable with no safe way to retry the mutation. Mutation fencing
therefore
+/// waits for the exclusive gate instead of falling back after the normal
timeout.
+async fn required_lock(path: PathBuf) -> io::Result<File> {
+ let file = tokio::task::spawn_blocking(move || open_lock(&path))
+ .await
+ .map_err(io::Error::other)??;
+ loop {
+ match file.try_lock() {
+ Ok(()) => return Ok(file),
+ Err(std::fs::TryLockError::WouldBlock) => {
+ tokio::time::sleep(Duration::from_millis(10)).await;
+ }
+ Err(std::fs::TryLockError::Error(error)) => return Err(error),
+ }
+ }
+}
+
+#[derive(Debug)]
+pub(in crate::io) struct SharedInvalidationGuard {
+ root: PathBuf,
+ gate: File,
+ clear_marker: bool,
+}
+
+impl SharedInvalidationGuard {
+ pub(in crate::io) async fn finish(self) {
+ let Self {
+ root,
+ gate,
+ clear_marker,
+ } = self;
+ let result = tokio::task::spawn_blocking(move || {
+ let _gate = gate;
+ finish_invalidation(&root)?;
+ if clear_marker {
+ std::fs::remove_file(root.join(MUTATION_MARKER))?;
+ }
+ Ok::<(), io::Error>(())
+ })
+ .await;
+ // Cleanup failure leaves the mutation marker in place, so uncertain
+ // blocks remain inaccessible.
+ match result {
+ Ok(Ok(())) => (),
+ Ok(Err(error)) => log::debug!("Shared cache invalidation cleanup
failed: {error}"),
+ Err(error) => log::debug!("Shared cache invalidation task failed:
{error}"),
+ }
+ }
+}
+
+impl SharedDiskCache {
+ #[cfg(test)]
+ pub(crate) fn new(root: PathBuf, max_size: u64) -> crate::Result<Self> {
+ Self::new_with_ratio(root, max_size, DEFAULT_HIGH_PRIORITY_POOL_RATIO)
+ }
+
+ pub(crate) fn new_with_ratio(
+ root: PathBuf,
+ max_size: u64,
+ high_priority_ratio: f64,
+ ) -> crate::Result<Self> {
+ let root = prepare_cache_root(&root)?;
+ let initialize = || -> io::Result<()> {
+ let gate = open_lock(&root.join("directory.lock"))?;
+ gate.lock()?;
+ let limit_path = root.join("capacity");
+ match std::fs::read_to_string(&limit_path) {
+ Ok(value) if value == max_size.to_string() => (),
+ Ok(_) => {
+ return Err(io::Error::other(
+ "Shared cache participants must use the same
local-cache.max-size",
+ ))
+ }
+ Err(e) if e.kind() == io::ErrorKind::NotFound => {
+ std::fs::write(limit_path, max_size.to_string())?;
+ }
+ Err(e) => return Err(e),
+ }
+ let ratio_path = root.join("high-priority-pool-ratio");
+ let ratio_bits = high_priority_ratio.to_bits().to_string();
+ match std::fs::read_to_string(&ratio_path) {
+ Ok(value) if value == ratio_bits => (),
+ Ok(_) => {
+ return Err(io::Error::other(
+ "Shared cache participants must use the same
local-cache.high-priority-pool-ratio",
+ ))
+ }
+ Err(e) if e.kind() == io::ErrorKind::NotFound => {
+ std::fs::write(ratio_path, ratio_bits)?;
+ }
+ Err(e) => return Err(e),
+ }
+ if read_generation(&root).is_none() {
+ // No block can be trusted without the epoch used to encode its
+ // key. Clear unknown blocks before publishing a fresh epoch.
+ std::fs::write(root.join("dirty"), b"")?;
+ let used = recover_usage(&root, 0, high_priority_ratio)?;
+ rotate_generation(&root)?;
+ finish_usage(&root, used)?;
+ }
+ // Clean participants share persisted accounting; do not rescan
every
+ // block whenever another worker starts.
+ if clean_usage(&root, max_size).is_none() {
+ std::fs::write(root.join("dirty"), b"")?;
+ let used = recover_usage(&root, max_size,
high_priority_ratio)?;
+ finish_usage(&root, used)?;
+ }
+ Ok(())
+ };
+ initialize().map_err(|e| crate::Error::ConfigInvalid {
+ message: e.to_string(),
+ })?;
+ Ok(Self {
+ root,
+ max_size,
+ high_priority_ratio,
+ })
+ }
+
+ pub(crate) fn generation(&self) -> Option<String> {
+ let generation = read_generation(&self.root)?;
+ mutation_cleared(&self.root).then_some(generation)
+ }
+
+ pub(crate) async fn load_lock(&self, key: &BlockKey) -> Option<File> {
+ // All blocks of one file use the same stripe, also coordinating
read_full.
+ let stripe =
+ BlockKey::with_namespace(&key.namespace, &key.path, 0,
0).cache_relative_path();
+ lock(self.root.join(format!(
+ "load-{}.lock",
+ stripe.components().next()?.as_os_str().to_string_lossy()
+ )))
+ .await
+ }
+
+ pub(crate) async fn get(&self, key: &BlockKey, generation:
&Option<String>) -> Option<Bytes> {
+ if generation.is_none() {
+ return None;
+ }
+ let gate = read_lock(self.root.join("directory.lock")).await?;
+ let root = self.root.clone();
+ let key = generation_key(key, generation.as_ref()?);
+ let generation = generation.clone();
+ tokio::task::spawn_blocking(move || {
+ let _gate = gate;
+ if !mutation_cleared(&root)
+ || std::fs::read_to_string(root.join("generation")).ok() !=
generation
+ {
+ return None;
+ }
+ let path = root.join(key.cache_relative_path());
+ let data = std::fs::read(&path).ok()?;
+ let payload = decode_block(&key, &data).ok()?;
+ // Persist recency so eviction observes hits from every process.
+ if let Ok(file) = File::open(path) {
+ let _ = file.set_modified(std::time::SystemTime::now());
+ }
+ Some(payload)
+ })
+ .await
+ .ok()
+ .flatten()
+ }
+
+ pub(crate) async fn put(&self, key: &BlockKey, payload: Bytes, generation:
&Option<String>) {
+ if generation.is_none() {
+ return;
+ }
+ let Some(gate) = lock(self.root.join("directory.lock")).await else {
+ return;
+ };
+ let root = self.root.clone();
+ let key = generation_key(key, generation.as_ref().unwrap());
+ let generation = generation.clone();
+ let max_size = self.max_size;
+ let high_priority_ratio = self.high_priority_ratio;
+ let result = tokio::task::spawn_blocking(move || -> io::Result<()> {
+ let _gate = gate;
+ if !mutation_cleared(&root)
+ || std::fs::read_to_string(root.join("generation")).ok() !=
generation
+ {
+ return Ok(());
+ }
+ let encoded = encode_block(&key, &payload);
+ if encoded.len() as u64 > max_size {
+ return Ok(());
+ }
+ let path = root.join(key.cache_relative_path());
+ let mut used = if root.join("dirty").exists() {
+ recover_usage(&root, max_size, high_priority_ratio)?
+ } else {
+ match std::fs::read_to_string(root.join("usage"))
+ .ok()
+ .and_then(|s| s.parse::<u64>().ok())
+ {
+ Some(used) => used,
+ None => recover_usage(&root, max_size,
high_priority_ratio)?,
+ }
+ };
+ // Keep the marker until publication and accounting both complete.
+ // A killed process leaves it for the next writer to recover.
+ std::fs::write(root.join("dirty"), b"")?;
+ let mut previous = std::fs::metadata(&path).map(|m|
m.len()).unwrap_or(0);
+ if used
+ .saturating_sub(previous)
+ .saturating_add(encoded.len() as u64)
+ > max_size
+ {
+ // Evict in batches, not one full directory scan per new block.
+ used = recover_usage(
+ &root,
+ (max_size - encoded.len() as u64).min(max_size - max_size
/ 10),
+ high_priority_ratio,
+ )?;
+ previous = std::fs::metadata(&path).map(|m|
m.len()).unwrap_or(0);
+ }
+ let parent = path.parent().expect("cache shard");
+ std::fs::create_dir_all(parent)?;
+ let temporary = parent.join(format!(
+ ".{}.tmp.{}",
+ path.file_name().unwrap().to_string_lossy(),
+ uuid::Uuid::new_v4()
+ ));
+ let mut options = OpenOptions::new();
+ options.write(true).create_new(true);
+ #[cfg(unix)]
+ {
+ use std::os::unix::fs::OpenOptionsExt;
+ options.mode(0o600);
+ }
+ let publish = (|| {
+ let mut file = options.open(&temporary)?;
+ file.write_all(&encoded)?;
+ drop(file);
+ std::fs::rename(&temporary, &path)
+ })();
+ if publish.is_err() {
+ let _ = std::fs::remove_file(&temporary);
+ }
+ publish?;
+ finish_usage(
+ &root,
+ used.saturating_sub(previous)
+ .saturating_add(encoded.len() as u64),
+ )
+ })
+ .await;
+ if !matches!(result, Ok(Ok(()))) {
+ log::debug!("Shared cache publication failed");
+ }
+ }
+
+ pub(crate) async fn invalidate(&self) -> io::Result<()> {
+ let gate = lock(self.root.join("directory.lock"))
+ .await
+ .ok_or_else(|| {
+ io::Error::other("Shared cache invalidation could not acquire
directory lock")
+ })?;
+ let root = self.root.clone();
+ tokio::task::spawn_blocking(move || -> io::Result<()> {
+ let _gate = gate;
+ start_invalidation(&root)?;
+ finish_invalidation(&root)
+ })
+ .await
+ .map_err(io::Error::other)?
+ }
+
+ pub(in crate::io) async fn begin_invalidation(&self) ->
io::Result<SharedInvalidationGuard> {
+ let gate = required_lock(self.root.join("directory.lock")).await?;
+ let root = self.root.clone();
+ tokio::task::spawn_blocking(move || {
+ // Unlike the accounting dirty marker, this survives cancellation.
+ let clear_marker = match OpenOptions::new()
+ .write(true)
+ .create_new(true)
+ .open(root.join(MUTATION_MARKER))
+ {
+ Ok(_) => true,
+ Err(error) if error.kind() == io::ErrorKind::AlreadyExists =>
false,
+ Err(error) => return Err(error),
+ };
+ start_invalidation(&root)?;
+ Ok(SharedInvalidationGuard {
+ root,
+ gate,
+ clear_marker,
+ })
+ })
+ .await
+ .map_err(io::Error::other)?
+ }
+}
+
+fn read_generation(root: &Path) -> Option<String> {
+ let value = std::fs::read_to_string(root.join("generation")).ok()?;
+ uuid::Uuid::parse_str(&value).ok()?;
+ Some(value)
+}
+
+fn mutation_cleared(root: &Path) -> bool {
+ matches!(
+ std::fs::symlink_metadata(root.join(MUTATION_MARKER)),
+ Err(error) if error.kind() == io::ErrorKind::NotFound
+ )
+}
+
+// The mutation and accounting helpers below are only called while holding the
+// exclusive directory lock.
+fn rotate_generation(root: &Path) -> io::Result<()> {
+ let temporary = root.join("generation.tmp");
+ std::fs::write(&temporary, uuid::Uuid::new_v4().to_string())?;
+ std::fs::rename(temporary, root.join("generation"))
+}
+
+fn start_invalidation(root: &Path) -> io::Result<()> {
+ // Keep recovery enabled until block cleanup and accounting both complete.
+ std::fs::write(root.join("dirty"), b"")?;
+ // Epoch is part of each block's key. Rotation makes every old block
+ // unreachable even if cleanup fails or this process exits midway.
+ rotate_generation(root)
+}
+
+fn finish_invalidation(root: &Path) -> io::Result<()> {
+ // This intentionally invalidates the whole shared directory.
+ for (key, _) in scan_existing_blocks(root) {
+ let _ = std::fs::remove_file(root.join(key.cache_relative_path()));
+ }
+ cleanup_temporary_files(root)?;
+ finish_usage(root, existing_usage(root))
+}
+
+fn existing_usage(root: &Path) -> u64 {
+ scan_existing_blocks(root)
+ .iter()
+ .fold(0u64, |used, (_, encoded_size)| {
+ used.saturating_add(*encoded_size)
+ })
+}
+
+fn clean_usage(root: &Path, max_size: u64) -> Option<u64> {
+ if root.join("dirty").try_exists().ok()? {
+ return None;
+ }
+ let used = std::fs::read_to_string(root.join("usage"))
+ .ok()?
+ .parse::<u64>()
+ .ok()?;
+ (used <= max_size).then_some(used)
+}
+
+fn recover_usage(root: &Path, limit: u64, high_priority_ratio: f64) ->
io::Result<u64> {
+ // Publication holds the directory lock, so every temporary block found
+ // here belongs to an interrupted publisher. Remove it before clearing
dirty.
+ cleanup_temporary_files(root)?;
+ #[cfg(test)]
+ RECOVERY_SCANS.with(|count| count.set(count.get() + 1));
+ let mut state = CacheState::new(Some(limit), high_priority_ratio);
+ for (key, encoded_size) in scan_existing_blocks(root) {
+ insert_state_entry(&mut state, key, encoded_size);
+ }
+ for key in collect_evictions(&mut state) {
+ std::fs::remove_file(root.join(key.cache_relative_path()))?;
+ }
+ Ok(state.current_size)
+}
+
+#[cfg(test)]
+thread_local! {
+ static RECOVERY_SCANS: std::cell::Cell<usize> = const {
std::cell::Cell::new(0) };
+}
+
+fn finish_usage(root: &Path, used: u64) -> io::Result<()> {
+ std::fs::write(root.join("usage"), used.to_string())?;
+ match std::fs::remove_file(root.join("dirty")) {
+ Ok(()) => Ok(()),
+ Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),
+ Err(e) => Err(e),
+ }
+}
+
+fn generation_key(key: &BlockKey, generation: &str) -> BlockKey {
+ BlockKey::with_namespace(
+ format!("{}:{generation}", key.namespace),
+ &key.path,
+ key.block_size,
+ key.block_index,
+ )
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::io::FileRead;
+ use std::process::{Command, Stdio};
+
+ const CHILD: &str = "io::cache::disk::shared::tests::shared_cache_child";
+
+ #[derive(Debug)]
+ struct Source(PathBuf);
+
+ #[async_trait::async_trait]
+ impl crate::io::FileRead for Source {
+ async fn read(&self, range: std::ops::Range<u64>) ->
crate::Result<Bytes> {
+ OpenOptions::new()
+ .create(true)
+ .append(true)
+ .open(&self.0)
+ .unwrap()
+ .write_all(b"read\n")
+ .unwrap();
+ tokio::time::sleep(Duration::from_millis(50)).await;
+ Ok(Bytes::from_static(b"payload").slice(range.start as
usize..range.end as usize))
+ }
+ }
+
+ fn key() -> BlockKey {
+ BlockKey::with_namespace("catalog", "data.parquet", 64, 0)
+ }
+
+ fn child(root: &Path, mode: &str) -> Command {
+ let mut command = Command::new(std::env::current_exe().unwrap());
+ command
+ .args(["--exact", CHILD, "--nocapture"])
+ .env("PAIMON_SHARED_TEST_ROOT", root)
+ .env("PAIMON_SHARED_TEST_MODE", mode)
+ .stdin(Stdio::null());
+ command
+ }
+
+ // A separate executable process, not another cache instance in this
process.
+ #[tokio::test]
+ async fn shared_cache_child() {
+ let Some(root) = std::env::var_os("PAIMON_SHARED_TEST_ROOT") else {
+ return;
+ };
+ let root = PathBuf::from(root);
+ let mode = std::env::var("PAIMON_SHARED_TEST_MODE").unwrap();
+ if mode == "startup" {
+ SharedDiskCache::new(root, 1024 * 1024).unwrap();
+ assert_eq!(RECOVERY_SCANS.with(|count| count.get()), 0);
+ return;
+ }
+ let cache = SharedDiskCache::new(root.clone(), 4096).unwrap();
+ let token = cache.generation();
+ match std::env::var("PAIMON_SHARED_TEST_MODE").unwrap().as_str() {
+ "populate" => {
+ let _guard = cache.load_lock(&key()).await.unwrap();
+ if cache.get(&key(), &token).await.is_none() {
+ let mut source = OpenOptions::new()
+ .create(true)
+ .append(true)
+ .open(root.join("source-reads"))
+ .unwrap();
+ source.write_all(b"read\n").unwrap();
+ tokio::time::sleep(Duration::from_millis(50)).await;
+ cache
+ .put(&key(), Bytes::from_static(b"payload"), &token)
+ .await;
+ }
+ assert_eq!(cache.get(&key(), &token).await.unwrap(),
"payload");
+ }
+ "hold" => {
+ let _guard = cache.load_lock(&key()).await.unwrap();
+ std::fs::write(root.join("locked"), b"ready").unwrap();
+ std::future::pending::<()>().await;
+ }
+ "invalidate" => cache.invalidate().await.unwrap(),
+ "reader" => {
+ use crate::io::cache::{CachedFileReader, FileType, LocalCache,
LocalCacheConfig};
+ let local = Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ shared: true,
+ dir: Some(root.clone()),
+ namespace: "catalog".into(),
+ max_size: Some(4096),
+ block_size: 64,
+ whitelist:
std::collections::HashSet::from([FileType::StructuredData]),
+ })
+ .unwrap(),
+ );
+ let reader = CachedFileReader::new(
+ Arc::new(Source(root.join("source-reads"))),
+ "s3://bucket/table/data.parquet",
+ 7,
+ local,
+ );
+ assert_eq!(reader.read(1..4).await.unwrap(), "ayl");
+ assert_eq!(reader.read_full().await.unwrap(), "payload");
+ }
+ other => panic!("Unknown test mode: {other}"),
+ }
+ }
+
+ #[test]
+ fn processes_coalesce_downloads() {
+ let directory = tempfile::tempdir().unwrap();
+ let mut children: Vec<_> = (0..4)
+ .map(|_| child(directory.path(), "populate").spawn().unwrap())
+ .collect();
+ for process in &mut children {
+ assert!(process.wait().unwrap().success());
+ }
+ assert_eq!(
+
std::fs::read_to_string(directory.path().join("source-reads")).unwrap(),
+ "read\n"
+ );
+ }
+
+ #[test]
+ fn cached_file_reader_reuses_blocks_across_processes() {
+ let directory = tempfile::tempdir().unwrap();
+ let mut children: Vec<_> = (0..4)
+ .map(|_| child(directory.path(), "reader").spawn().unwrap())
+ .collect();
+ for process in &mut children {
+ assert!(process.wait().unwrap().success());
+ }
+ assert_eq!(
+
std::fs::read_to_string(directory.path().join("source-reads")).unwrap(),
+ "read\n"
+ );
+ }
+
+ #[tokio::test]
+ async fn another_process_invalidates_old_tokens() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = SharedDiskCache::new(directory.path().to_path_buf(),
4096).unwrap();
+ let old = cache.generation();
+ cache.put(&key(), Bytes::from_static(b"old"), &old).await;
+ assert!(child(directory.path(), "invalidate")
+ .status()
+ .unwrap()
+ .success());
+ cache.put(&key(), Bytes::from_static(b"late"), &old).await;
+ assert!(cache.get(&key(), &old).await.is_none());
+ assert!(cache.get(&key(), &cache.generation()).await.is_none());
+ }
+
+ #[tokio::test]
+ async fn mutation_does_not_issue_publishable_generation() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = SharedDiskCache::new(directory.path().to_path_buf(),
4096).unwrap();
+ let before = cache.generation();
+
+ let mutation = cache.begin_invalidation().await.unwrap();
+ let during = cache.generation();
+ assert!(during.is_none());
+ mutation.finish().await;
+
+ cache.put(&key(), Bytes::from_static(b"old"), &before).await;
+ cache.put(&key(), Bytes::from_static(b"old"), &during).await;
+ let after = cache.generation();
+ assert_ne!(after, before);
+ assert!(cache.get(&key(), &after).await.is_none());
+ cache.put(&key(), Bytes::from_static(b"new"), &after).await;
+ assert_eq!(cache.get(&key(), &after).await.unwrap(), "new");
+
+ let cancelled = cache.begin_invalidation().await.unwrap();
+ drop(cancelled);
+ assert!(cache.generation().is_none());
+ let uncertain = read_generation(directory.path());
+ cache
+ .put(&key(), Bytes::from_static(b"stale"), &uncertain)
+ .await;
+ assert!(cache.generation().is_none());
+ assert!(cache.get(&key(), &uncertain).await.is_none());
+ let reopened = SharedDiskCache::new(directory.path().to_path_buf(),
4096).unwrap();
+ assert!(reopened.generation().is_none());
+ reopened.begin_invalidation().await.unwrap().finish().await;
+ assert!(reopened.generation().is_none());
+ }
+
+ #[tokio::test]
+ async fn cache_hits_can_share_the_directory_read_lock() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = SharedDiskCache::new(directory.path().to_path_buf(),
4096).unwrap();
+ let token = cache.generation();
+ cache
+ .put(&key(), Bytes::from_static(b"payload"), &token)
+ .await;
+ let gate =
open_lock(&directory.path().join("directory.lock")).unwrap();
+ File::lock_shared(&gate).unwrap();
+
+ let payload = tokio::time::timeout(Duration::from_millis(500),
cache.get(&key(), &token))
+ .await
+ .expect("cache hits must not serialize behind another reader");
+ assert_eq!(payload.unwrap(), "payload");
+ }
+
+ #[tokio::test]
+ async fn killed_downloader_releases_lock() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = SharedDiskCache::new(directory.path().to_path_buf(),
4096).unwrap();
+ let mut process = child(directory.path(), "hold").spawn().unwrap();
+ let ready = tokio::time::timeout(Duration::from_secs(10), async {
+ while !directory.path().join("locked").exists() {
+ tokio::time::sleep(Duration::from_millis(10)).await;
+ }
+ })
+ .await;
+ process.kill().unwrap();
+ process.wait().unwrap();
+ ready.unwrap();
+ let guard = tokio::time::timeout(Duration::from_secs(2),
cache.load_lock(&key()))
+ .await
+ .unwrap();
+ assert!(guard.is_some());
+ }
+
+ #[tokio::test]
+ async fn capacity_and_corruption() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = SharedDiskCache::new(directory.path().to_path_buf(),
512).unwrap();
+ assert!(SharedDiskCache::new(directory.path().to_path_buf(),
1024).is_err());
+ let token = cache.generation();
+ for index in 0..10 {
+ let mut key = key();
+ key.block_index = index;
+ cache.put(&key, Bytes::from(vec![42; 64]), &token).await;
+ }
+ assert!(existing_usage(directory.path()) <= 512);
+ let newest = BlockKey::with_namespace("catalog", "data.parquet", 64,
9);
+ let file = directory
+ .path()
+ .join(generation_key(&newest,
token.as_ref().unwrap()).cache_relative_path());
+ std::fs::write(file, b"corrupt").unwrap();
+ assert!(cache.get(&newest, &token).await.is_none());
+ }
+
+ #[tokio::test]
+ async fn shared_cache_preserves_metadata_priority() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache =
+ SharedDiskCache::new_with_ratio(directory.path().to_path_buf(),
2048, 0.25).unwrap();
+ assert!(
+ SharedDiskCache::new_with_ratio(directory.path().to_path_buf(),
2048, 0.5).is_err()
+ );
+ let token = cache.generation();
+ let metadata = BlockKey::with_namespace("catalog", "snapshot-1", 64,
0);
+ cache.put(&metadata, Bytes::from(vec![1; 64]), &token).await;
+ for index in 0..20 {
+ let mut data = key();
+ data.block_index = index;
+ cache.put(&data, Bytes::from(vec![2; 64]), &token).await;
+ }
+ assert_eq!(
+ cache.get(&metadata, &token).await,
+ Some(Bytes::from(vec![1; 64]))
+ );
+ assert!(existing_usage(directory.path()) <= 2048);
+ }
+
+ #[tokio::test]
+ async fn interrupted_accounting_recovers_before_next_publication() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = SharedDiskCache::new(directory.path().to_path_buf(),
512).unwrap();
+ let token = cache.generation();
+ cache.put(&key(), Bytes::from(vec![42; 64]), &token).await;
+ // Simulate exit after publishing a block, before committing its size.
+ std::fs::write(directory.path().join("usage"), b"0").unwrap();
+ std::fs::write(directory.path().join("dirty"), b"").unwrap();
+ let mut second = key();
+ second.block_index = 1;
+ cache.put(&second, Bytes::from(vec![43; 64]), &token).await;
+ assert!(!directory.path().join("dirty").exists());
+ let used: u64 = std::fs::read_to_string(directory.path().join("usage"))
+ .unwrap()
+ .parse()
+ .unwrap();
+ assert_eq!(used, existing_usage(directory.path()));
+ assert!(used <= 512);
+ }
+
+ #[tokio::test]
+ async fn interrupted_temporary_blocks_are_removed_before_clean_startup() {
+ for invalidate in [false, true] {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = SharedDiskCache::new(directory.path().to_path_buf(),
1024 * 1024).unwrap();
+ let token = cache.generation();
+ let block = directory
+ .path()
+ .join(generation_key(&key(),
token.as_ref().unwrap()).cache_relative_path());
+ std::fs::create_dir_all(block.parent().unwrap()).unwrap();
+ let temporary = block.parent().unwrap().join(format!(
+ ".{}.tmp.{}",
+ block.file_name().unwrap().to_string_lossy(),
+ uuid::Uuid::new_v4()
+ ));
+ // Simulate a publisher exiting after writing a partial temporary
block.
+ std::fs::write(&temporary, b"partial block").unwrap();
+ std::fs::write(directory.path().join("dirty"), b"").unwrap();
+ if invalidate {
+ cache.invalidate().await.unwrap();
+ } else {
+ cache
+ .put(&key(), Bytes::from_static(b"payload"), &token)
+ .await;
+ assert_eq!(cache.get(&key(), &token).await.unwrap(),
"payload");
+ }
+ assert!(!temporary.exists());
+ assert!(clean_usage(directory.path(), 1024 * 1024).is_some());
+ assert!(child(directory.path(), "startup")
+ .status()
+ .unwrap()
+ .success());
+ assert!(!temporary.exists());
+ }
+ }
+
+ #[test]
+ fn clean_workers_do_not_scan_blocks_on_startup() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = SharedDiskCache::new(directory.path().to_path_buf(), 1024
* 1024).unwrap();
+ let generation = cache.generation().unwrap();
+ let mut used = 0;
+ for index in 0..2000 {
+ let key = generation_key(
+ &BlockKey::with_namespace("catalog", "data.parquet", 64,
index),
+ &generation,
+ );
+ let path = directory.path().join(key.cache_relative_path());
+ std::fs::create_dir_all(path.parent().unwrap()).unwrap();
+ let bytes = encode_block(&key, &Bytes::from_static(b"payload"));
+ used += bytes.len() as u64;
+ std::fs::write(path, bytes).unwrap();
+ }
+ finish_usage(directory.path(), used).unwrap();
+ let mut children: Vec<_> = (0..8)
+ .map(|_| child(directory.path(), "startup").spawn().unwrap())
+ .collect();
+ for process in &mut children {
+ assert!(process.wait().unwrap().success());
+ }
+ }
+
+ #[test]
+ fn startup_recovers_dirty_invalid_and_oversized_accounting() {
+ let directory = tempfile::tempdir().unwrap();
+ SharedDiskCache::new(directory.path().to_path_buf(), 4096).unwrap();
+ for usage in ["invalid", "4097", "0"] {
+ std::fs::write(directory.path().join("usage"), usage).unwrap();
+ if usage == "0" {
+ std::fs::write(directory.path().join("dirty"), b"").unwrap();
+ }
+ let before = RECOVERY_SCANS.with(|count| count.get());
+ SharedDiskCache::new(directory.path().to_path_buf(),
4096).unwrap();
+ assert_eq!(RECOVERY_SCANS.with(|count| count.get()), before + 1);
+ assert_eq!(clean_usage(directory.path(), 4096), Some(0));
+ }
+ }
+
+ #[tokio::test]
+ async fn startup_replaces_malformed_generation_and_unknown_blocks() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = SharedDiskCache::new(directory.path().to_path_buf(),
4096).unwrap();
+ let old_generation = cache.generation();
+ cache
+ .put(&key(), Bytes::from_static(b"old"), &old_generation)
+ .await;
+ let old_block = directory
+ .path()
+ .join(generation_key(&key(),
old_generation.as_ref().unwrap()).cache_relative_path());
+ assert!(old_block.exists());
+ std::fs::write(directory.path().join("generation"),
b"truncated").unwrap();
+
+ let repaired = SharedDiskCache::new(directory.path().to_path_buf(),
4096).unwrap();
+ let new_generation = repaired.generation();
+ assert!(new_generation.is_some());
+ assert_ne!(new_generation, old_generation);
+ assert!(!old_block.exists());
+ assert_eq!(clean_usage(directory.path(), 4096), Some(0));
+
+ repaired
+ .put(&key(), Bytes::from_static(b"new"), &new_generation)
+ .await;
+ assert_eq!(repaired.get(&key(), &new_generation).await.unwrap(),
"new");
+ }
+
+ #[tokio::test]
+ async fn invalidation_reports_lock_timeout_and_filesystem_failure() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = SharedDiskCache::new(directory.path().to_path_buf(),
4096).unwrap();
+ let gate =
open_lock(&directory.path().join("directory.lock")).unwrap();
+ gate.lock().unwrap();
+ assert!(cache.invalidate().await.is_err());
+ drop(gate);
+ std::fs::create_dir(directory.path().join("generation.tmp")).unwrap();
+ assert!(cache.invalidate().await.is_err());
+ std::fs::remove_dir(directory.path().join("generation.tmp")).unwrap();
+ cache.invalidate().await.unwrap();
+ }
+}
diff --git a/crates/paimon/src/io/cache/state.rs
b/crates/paimon/src/io/cache/state.rs
index 7625186b..363bfdc7 100644
--- a/crates/paimon/src/io/cache/state.rs
+++ b/crates/paimon/src/io/cache/state.rs
@@ -94,6 +94,7 @@ struct PathCacheState {
#[derive(Clone)]
pub(in crate::io) struct CacheReadToken {
+ pub(super) shared_generation: Option<String>,
generation: u64,
state: Arc<PathCacheState>,
}
@@ -134,6 +135,7 @@ impl CacheCoordinator {
pub(super) fn read_token(&self, namespace: &str, path: &str) ->
CacheReadToken {
let state = self.path_state(namespace, path);
CacheReadToken {
+ shared_generation: None,
generation:
state.generation.load(std::sync::atomic::Ordering::SeqCst),
state,
}
diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index 41cc2adb..ab093427 100644
--- a/crates/paimon/src/io/file_io.rs
+++ b/crates/paimon/src/io/file_io.rs
@@ -655,7 +655,7 @@ impl FileIO {
self.cache.as_ref().filter(|cache| cache.is_cacheable(path)),
cache_path,
) {
- cache.invalidate_path(&cache_path).await;
+ cache.invalidate_path(&cache_path).await?;
}
Ok(())
@@ -681,7 +681,7 @@ impl FileIO {
)
})?;
if let (Some(cache), Some(cache_path)) = (&self.cache, cache_path) {
- cache.invalidate_prefix(&cache_path).await;
+ cache.invalidate_prefix(&cache_path).await?;
}
Ok(())
@@ -759,7 +759,16 @@ impl FileIO {
.to_string(),
});
}
- op_src
+ let shared_invalidation = if cache_path_src.is_some() ||
cache_path_dst.is_some() {
+ match &self.cache {
+ Some(cache) => cache.begin_shared_invalidation().await?,
+ None => None,
+ }
+ } else {
+ None
+ };
+ let shared_cache_fenced = shared_invalidation.is_some();
+ let rename_result = op_src
.rename(relative_path_src.as_ref(), relative_path_dst.as_ref())
.await
.map_err(|error| {
@@ -767,13 +776,21 @@ impl FileIO {
error,
format!("Failed to rename '{src}' to '{dst}'"),
)
- })?;
- if let Some(cache) = &self.cache {
- if let Some(cache_path) = cache_path_src {
- cache.invalidate_prefix(&cache_path).await;
- }
- if let Some(cache_path) = cache_path_dst {
- cache.invalidate_prefix(&cache_path).await;
+ });
+ // A failed rename may still complete remotely. Leave the shared cache
+ // fenced rather than publish data read while its outcome is uncertain.
+ rename_result?;
+ if let Some(invalidation) = shared_invalidation {
+ invalidation.finish().await;
+ }
+ if !shared_cache_fenced {
+ if let Some(cache) = &self.cache {
+ if let Some(cache_path) = cache_path_src {
+ cache.invalidate_prefix(&cache_path).await?;
+ }
+ if let Some(cache_path) = cache_path_dst {
+ cache.invalidate_prefix(&cache_path).await?;
+ }
}
}
@@ -1132,6 +1149,7 @@ struct CacheInvalidatingWriter {
delegate: Box<dyn FileWrite>,
cache: Arc<LocalCache>,
path: String,
+ delegate_closed: bool,
}
/// OpenDAL's local position writer retains its last write until close. Use a
@@ -1245,13 +1263,16 @@ impl FileWrite for CacheInvalidatingWriter {
async fn flush(&mut self) -> crate::Result<()> {
self.delegate.flush().await?;
- self.cache.invalidate_path(&self.path).await;
+ self.cache.invalidate_path(&self.path).await?;
Ok(())
}
async fn close(&mut self) -> crate::Result<()> {
- self.delegate.close().await?;
- self.cache.invalidate_path(&self.path).await;
+ if !self.delegate_closed {
+ self.delegate.close().await?;
+ self.delegate_closed = true;
+ }
+ self.cache.invalidate_path(&self.path).await?;
Ok(())
}
}
@@ -1266,7 +1287,7 @@ struct CacheInvalidatingAsyncWriter {
cache: Arc<LocalCache>,
path: String,
delegate_shutdown: bool,
- invalidation: Option<Pin<Box<dyn Future<Output = ()> + Send>>>,
+ invalidation: Option<Pin<Box<dyn Future<Output = crate::Result<()>> +
Send>>>,
}
impl tokio::io::AsyncWrite for CacheInvalidatingAsyncWriter {
@@ -1306,7 +1327,16 @@ impl tokio::io::AsyncWrite for
CacheInvalidatingAsyncWriter {
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(())) => self.invalidation = None,
+ Poll::Ready(Err(error)) => {
+ // A retry must retry invalidation, not report successful
+ // shutdown merely because the delegate is already closed.
+ let cache = self.cache.clone();
+ let path = self.path.clone();
+ self.invalidation =
+ Some(Box::pin(async move {
cache.invalidate_path(&path).await }));
+ return Poll::Ready(Err(std::io::Error::other(error)));
+ }
}
}
Poll::Ready(Ok(()))
@@ -1514,6 +1544,7 @@ impl OutputFile {
delegate: writer,
cache: cache.clone(),
path: cache_path,
+ delegate_closed: false,
})
}
@@ -3022,6 +3053,272 @@ mod input_output_test {
);
}
+ #[cfg(not(windows))]
+ #[tokio::test]
+ async fn test_shared_cache_delete_timeout_and_recreate() {
+ use tokio::io::AsyncWriteExt;
+ let source = tempfile::tempdir().unwrap();
+ let directory = tempfile::tempdir().unwrap();
+ let path = source.path().join("snapshot-1");
+ std::fs::write(&path, b"old metadata").unwrap();
+ let location = format!("file:{}", path.display());
+ let mut options = Options::new();
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ options.set(CatalogOptions::LOCAL_CACHE_SHARED, "true");
+ options.set(
+ CatalogOptions::LOCAL_CACHE_DIR,
+ directory.path().to_string_lossy(),
+ );
+ options.set(CatalogOptions::LOCAL_CACHE_MAX_SIZE, "1 MiB");
+ let cache = Arc::new(
+
LocalCache::new(LocalCacheConfig::from_options(&options).unwrap().unwrap()).unwrap(),
+ );
+ let file_io = FileIOBuilder::new("file")
+ .with_local_cache(cache)
+ .build()
+ .unwrap();
+ assert_eq!(
+ file_io.new_input(&location).unwrap().read().await.unwrap(),
+ "old metadata"
+ );
+ let gate = std::fs::OpenOptions::new()
+ .read(true)
+ .write(true)
+ .open(
+ directory
+ .path()
+ .join("paimon-shared-local-cache-v1/directory.lock"),
+ )
+ .unwrap();
+ gate.lock().unwrap();
+ let error = file_io.delete_file(&location).await.unwrap_err();
+ assert!(error.to_string().contains("invalidation failed"));
+ drop(gate);
+ // The underlying delete may already have succeeded. Retrying the
+ // idempotent delete must complete invalidation before path reuse.
+ file_io.delete_file(&location).await.unwrap();
+ std::fs::write(&path, b"new metadata").unwrap();
+ assert_eq!(
+ file_io.new_input(&location).unwrap().read().await.unwrap(),
+ "new metadata"
+ );
+ let mut writer = file_io
+ .new_output(&location)
+ .unwrap()
+ .async_writer()
+ .await
+ .unwrap();
+ writer.write_all(b"replacement!").await.unwrap();
+ let gate = std::fs::OpenOptions::new()
+ .read(true)
+ .write(true)
+ .open(
+ directory
+ .path()
+ .join("paimon-shared-local-cache-v1/directory.lock"),
+ )
+ .unwrap();
+ gate.lock().unwrap();
+ assert!(writer.shutdown().await.is_err());
+ // A second shutdown cannot silently succeed while invalidation still
fails.
+ assert!(writer.shutdown().await.is_err());
+ drop(gate);
+ writer.shutdown().await.unwrap();
+ assert_eq!(
+ file_io.new_input(&location).unwrap().read().await.unwrap(),
+ "replacement!"
+ );
+
+ let gate = std::fs::OpenOptions::new()
+ .read(true)
+ .write(true)
+ .open(
+ directory
+ .path()
+ .join("paimon-shared-local-cache-v1/directory.lock"),
+ )
+ .unwrap();
+ gate.lock().unwrap();
+ let directory_location = format!("file:{}", source.path().display());
+ assert!(file_io.delete_dir(&directory_location).await.is_err());
+ drop(gate);
+ file_io.delete_dir(&directory_location).await.unwrap();
+ std::fs::create_dir_all(source.path()).unwrap();
+ std::fs::write(&path, b"recreated!!!").unwrap();
+ assert_eq!(
+ file_io.new_input(&location).unwrap().read().await.unwrap(),
+ "recreated!!!"
+ );
+ }
+
+ #[cfg(not(windows))]
+ #[tokio::test]
+ async fn test_shared_cache_regular_close_retries_invalidation() {
+ let source = tempfile::tempdir().unwrap();
+ let directory = tempfile::tempdir().unwrap();
+ let path = source.path().join("snapshot-1");
+ std::fs::write(&path, b"old").unwrap();
+ let location = format!("file:{}", path.display());
+ let mut options = Options::new();
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ options.set(CatalogOptions::LOCAL_CACHE_SHARED, "true");
+ options.set(
+ CatalogOptions::LOCAL_CACHE_DIR,
+ directory.path().to_string_lossy(),
+ );
+ options.set(CatalogOptions::LOCAL_CACHE_MAX_SIZE, "1 MiB");
+ let cache = Arc::new(
+
LocalCache::new(LocalCacheConfig::from_options(&options).unwrap().unwrap()).unwrap(),
+ );
+ let file_io = FileIOBuilder::new("file")
+ .with_local_cache(cache)
+ .build()
+ .unwrap();
+ assert_eq!(
+ file_io.new_input(&location).unwrap().read().await.unwrap(),
+ "old"
+ );
+ let mut writer = file_io
+ .new_output(&location)
+ .unwrap()
+ .writer()
+ .await
+ .unwrap();
+ writer.write(Bytes::from_static(b"new")).await.unwrap();
+ let gate = std::fs::OpenOptions::new()
+ .read(true)
+ .write(true)
+ .open(
+ directory
+ .path()
+ .join("paimon-shared-local-cache-v1/directory.lock"),
+ )
+ .unwrap();
+ gate.lock().unwrap();
+ assert!(writer.close().await.is_err());
+ assert!(writer.close().await.is_err());
+ drop(gate);
+ writer.close().await.unwrap();
+ assert_eq!(
+ file_io.new_input(&location).unwrap().read().await.unwrap(),
+ "new"
+ );
+ }
+
+ #[cfg(not(windows))]
+ #[tokio::test]
+ async fn test_shared_cache_rename_waits_for_invalidation_fence() {
+ 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 mut options = Options::new();
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ options.set(CatalogOptions::LOCAL_CACHE_SHARED, "true");
+ options.set(
+ CatalogOptions::LOCAL_CACHE_DIR,
+ cache_directory.path().to_string_lossy(),
+ );
+ options.set(CatalogOptions::LOCAL_CACHE_MAX_SIZE, "1 MiB");
+ let cache = Arc::new(
+
LocalCache::new(LocalCacheConfig::from_options(&options).unwrap().unwrap()).unwrap(),
+ );
+ let file_io = FileIOBuilder::new("file")
+ .with_local_cache(cache)
+ .build()
+ .unwrap();
+ assert_eq!(
+ file_io
+ .new_input(&target_location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ "target value"
+ );
+
+ let gate = std::fs::OpenOptions::new()
+ .read(true)
+ .write(true)
+ .open(
+ cache_directory
+ .path()
+ .join("paimon-shared-local-cache-v1/directory.lock"),
+ )
+ .unwrap();
+ gate.lock().unwrap();
+ let rename_io = file_io.clone();
+ let rename_source = source_location.clone();
+ let rename_target = target_location.clone();
+ let rename =
+ tokio::spawn(async move { rename_io.rename(&rename_source,
&rename_target).await });
+ tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+ assert!(!rename.is_finished());
+ assert_eq!(std::fs::read(&source_path).unwrap(), b"source value");
+ assert_eq!(std::fs::read(&target_path).unwrap(), b"target value");
+
+ drop(gate);
+ tokio::time::timeout(std::time::Duration::from_secs(5), rename)
+ .await
+ .unwrap()
+ .unwrap()
+ .unwrap();
+ assert!(!source_path.exists());
+ assert_eq!(std::fs::read(&target_path).unwrap(), b"source value");
+ assert_eq!(
+ file_io
+ .new_input(&target_location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ "source value"
+ );
+
+ let missing_source = format!("file:{}",
source_directory.path().join("missing").display());
+ assert!(file_io
+ .rename(&missing_source, &target_location)
+ .await
+ .is_err());
+ assert!(cache_directory
+ .path()
+ .join("paimon-shared-local-cache-v1/mutation-pending")
+ .exists());
+ std::fs::write(&target_path, b"newer value").unwrap();
+ assert_eq!(
+ file_io
+ .new_input(&target_location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ "newer value"
+ );
+ let next_path = source_directory.path().join("snapshot-3");
+ let next_location = format!("file:{}", next_path.display());
+ file_io
+ .rename(&target_location, &next_location)
+ .await
+ .unwrap();
+ assert_eq!(
+ file_io
+ .new_input(&next_location)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ "newer value"
+ );
+ assert!(cache_directory
+ .path()
+ .join("paimon-shared-local-cache-v1/mutation-pending")
+ .exists());
+ }
+
#[cfg(not(windows))]
#[tokio::test]
async fn test_file_io_local_cache_invalidates_after_delete_directory() {
diff --git a/crates/paimon/src/table/rest_env.rs
b/crates/paimon/src/table/rest_env.rs
index 9c4ef802..c55bfd70 100644
--- a/crates/paimon/src/table/rest_env.rs
+++ b/crates/paimon/src/table/rest_env.rs
@@ -23,7 +23,7 @@ use crate::api::GetTableResponse;
use crate::catalog::{FusePathResolver, Identifier, RESTToken, RESTTokenFileIO};
use crate::common::{CatalogOptions, Options};
use crate::error::Error;
-use crate::io::cache::{create_local_cache_with_namespace, LocalCache};
+use crate::io::cache::{create_local_cache_with_namespace_async, LocalCache};
use crate::io::{FileIO, FileIOCacheContext};
use crate::spec::{CoreOptions, TableSchema, PATH_OPTION};
use crate::table::snapshot_commit::{RESTSnapshotCommit, SnapshotCommit};
@@ -64,7 +64,8 @@ impl Table {
.get(CatalogOptions::DATA_TOKEN_ENABLED)
.map(|v| v.eq_ignore_ascii_case("true"))
.unwrap_or(false);
- let local_cache = create_local_cache_with_namespace(&rest_options,
api.options())?;
+ let local_cache =
+ create_local_cache_with_namespace_async(&rest_options,
api.options()).await?;
let file_io_cache =
FileIOCacheContext::from_props(api.options().to_map())?;
let fuse_resolver = Arc::new(FusePathResolver::new(api.options()));
RESTEnv::build_table_with_token(
diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md
index 8c99f8c2..dcfa30e6 100644
--- a/docs/src/getting-started.md
+++ b/docs/src/getting-started.md
@@ -401,6 +401,7 @@ let catalog = CatalogFactory::create(options).await?;
|--------|---------|-------------|
| `local-cache.enabled` | `false` | Enable catalog-scoped local block caching.
|
| `local-cache.dir` | none | Optional base directory. When set, Paimon uses a
persistent disk cache in a private versioned child directory; otherwise it uses
memory. |
+| `local-cache.shared` | `false` | Share disk blocks across processes on one
node; requires `local-cache.dir` and `local-cache.max-size`. |
| `local-cache.max-size` | unlimited | Maximum cache size. Memory caches count
payload bytes; disk caches count encoded bytes. Values accept byte units such
as `512 MiB` or `20 GiB`. |
| `local-cache.high-priority-pool-ratio` | `0.25` | Fraction of a finite cache
budget protected for `meta` and `global-index` blocks. Use `0` for a single
LRU. |
| `local-cache.block-size` | `1 MiB` | Block size used for cached range reads.
|
@@ -411,7 +412,7 @@ blocks without dedicated BLOB, video, or vector-store
files. `data` and `*`
retain their broader behavior.
The protected and regular LRUs share one total budget; either can use idle
-capacity. Catalogs sharing a disk cache within one process must use the same
ratio.
+capacity. Catalogs or processes sharing a disk cache must use the same ratio.
Each catalog owns its in-memory cache for the catalog's lifetime. Disk caches
are reused after process restarts. Cache keys include a catalog-configuration
@@ -424,8 +425,16 @@ convention. Cache managers using the same canonical
directory in one process
share LRU, size accounting, and invalidation state; if their configured limits
differ, the smallest `local-cache.max-size` is used. Restart recovery runs on a
blocking worker, reads only block headers and file metadata, and validates
-payload CRC lazily on the first hit. Use a separate `local-cache.dir` for each
-worker or process because processes do not share exact LRU or size accounting.
+payload CRC lazily on the first hit. Without shared mode, use a separate
+`local-cache.dir` per worker or process.
+
+For sharing, set `local-cache.enabled=true` and `local-cache.shared=true` in
+same-UID workers with the same node-local directory and capacity. NFS,
+cross-node sharing, and catalogs inherited across `fork` are unsupported.
+Invalidation clears the shared cache; if it fails after a storage mutation,
+do not reuse the affected path until invalidation succeeds.
+A canceled or failed rename bypasses this cache; after confirming storage
state,
+use a new cache directory to resume sharing.
### BLOB Index Cache
