From e20eab93cd9521f9a2aab15f2ea3a78047dae3bb Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Tue, 20 Oct 2026 07:24:10 -0700 Subject: [PATCH 1/2] [core] Add opt-in node-shared disk block cache --- crates/paimon/src/catalog/factory.rs | 59 +- .../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 | 152 +++- crates/paimon/src/io/cache/reader.rs | 18 +- crates/paimon/src/io/cache/shared.rs | 827 ++++++++++++++++++ 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 | 13 +- 11 files changed, 1429 insertions(+), 34 deletions(-) create mode 100644 crates/paimon/src/io/cache/shared.rs diff --git a/crates/paimon/src/catalog/factory.rs b/crates/paimon/src/catalog/factory.rs index c531ccd5d..81ee70593 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> { - let catalog = FileSystemCatalog::new(options)?; + async fn create_filesystem_catalog(options: Options) -> Result> { + 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 de88f8d6f..661f6d1a3 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 c75f1954f..eeefb7d07 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 b731e3030..3b217e04e 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 42eb32afa..83d53d87b 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), + SharedDisk(SharedDiskCache), External(Arc), } @@ -89,7 +91,22 @@ 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( + 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(), + })?, + )?), + 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 +152,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 +176,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 +191,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 { + match &self.backend { + CacheBackend::SharedDisk(disk) => disk.load_lock(key).await, + _ => None, + } } async fn block_load_lock(&self, key: &BlockKey) -> Arc> { @@ -187,6 +224,10 @@ impl LocalCache { } pub(super) async fn file_size(&self, path: &str, token: &CacheReadToken) -> Option { + // 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 +237,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 +249,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 +257,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 +276,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> { + 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 +346,34 @@ fn parse_high_priority_ratio(options: &Options) -> crate::Result { Ok(ratio) } +pub(crate) async fn create_local_cache_with_namespace_async( + cache_options: &Options, + namespace_options: &Options, +) -> crate::Result>> { + 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, namespace: String, max_size: Option, @@ -338,6 +439,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 +635,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 +649,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 +675,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 +711,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 +720,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 +733,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 +744,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 +774,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 +795,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 edb00a0af..f57b8f7cc 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, @@ -360,6 +363,7 @@ mod tests { }); let cache = Arc::new( LocalCache::new(LocalCacheConfig { + shared: false, dir: None, namespace: "test".to_string(), max_size: None, @@ -390,6 +394,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 +442,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 +466,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 +492,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 +503,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 +527,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 +555,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 +622,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 +702,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 +737,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 +759,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 000000000..552018fb9 --- /dev/null +++ b/crates/paimon/src/io/cache/shared.rs @@ -0,0 +1,827 @@ +// 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, +} + +fn open_lock(path: &Path) -> io::Result { + 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 { + 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 { + 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 { + 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 { + pub(crate) fn new(root: PathBuf, max_size: u64) -> crate::Result { + 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), + } + 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)?; + 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)?; + finish_usage(&root, used)?; + } + Ok(()) + }; + initialize().map_err(|e| crate::Error::ConfigInvalid { + message: e.to_string(), + })?; + Ok(Self { root, max_size }) + } + + pub(crate) fn generation(&self) -> Option { + let generation = read_generation(&self.root)?; + mutation_cleared(&self.root).then_some(generation) + } + + pub(crate) async fn load_lock(&self, key: &BlockKey) -> Option { + // 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) -> Option { + 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) { + 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 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)? + } else { + match std::fs::read_to_string(root.join("usage")) + .ok() + .and_then(|s| s.parse::().ok()) + { + Some(used) => used, + None => recover_usage(&root, max_size)?, + } + }; + // 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), + )?; + 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 { + 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 { + 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. + let state = scan_existing_blocks(root); + for key in state.entries.keys() { + let _ = std::fs::remove_file(root.join(key.cache_relative_path())); + } + cleanup_temporary_files(root)?; + finish_usage(root, scan_existing_blocks(root).current_size) +} + +fn clean_usage(root: &Path, max_size: u64) -> Option { + if root.join("dirty").try_exists().ok()? { + return None; + } + let used = std::fs::read_to_string(root.join("usage")) + .ok()? + .parse::() + .ok()?; + (used <= max_size).then_some(used) +} + +fn recover_usage(root: &Path, limit: u64) -> io::Result { + // 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 = scan_existing_blocks(root); + state.max_size = Some(limit); + 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 = 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) -> crate::Result { + 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; + } + let state = scan_existing_blocks(directory.path()); + assert!(state.current_size <= 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 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, scan_existing_blocks(directory.path()).current_size); + 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 7625186b2..363bfdc75 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, generation: u64, state: Arc, } @@ -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 41cc2adbc..ab0934272 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, cache: Arc, 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, path: String, delegate_shutdown: bool, - invalidation: Option + Send>>>, + invalidation: Option> + 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 9c4ef802a..c55bfd704 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 8c99f8c24..76d6a45c7 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. | @@ -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 From c2e89b116df245e077ff183e7549f755f8822fdf Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Tue, 20 Oct 2026 07:38:30 -0700 Subject: [PATCH 2/2] fix(io): preserve shared cache priority after rebase --- crates/paimon/src/io/cache/mod.rs | 3 +- crates/paimon/src/io/cache/reader.rs | 1 + crates/paimon/src/io/cache/shared.rs | 89 +++++++++++++++++++++++----- docs/src/getting-started.md | 2 +- 4 files changed, 79 insertions(+), 16 deletions(-) diff --git a/crates/paimon/src/io/cache/mod.rs b/crates/paimon/src/io/cache/mod.rs index 83d53d87b..701f305be 100644 --- a/crates/paimon/src/io/cache/mod.rs +++ b/crates/paimon/src/io/cache/mod.rs @@ -93,7 +93,7 @@ impl LocalCache { .clamp(1, DEFAULT_FILE_SIZE_CAPACITY); let (backend, coordinator) = if config.shared { ( - CacheBackend::SharedDisk(SharedDiskCache::new( + CacheBackend::SharedDisk(SharedDiskCache::new_with_ratio( config .dir .ok_or_else(|| crate::Error::ConfigInvalid { @@ -103,6 +103,7 @@ impl LocalCache { config.max_size.ok_or_else(|| crate::Error::ConfigInvalid { message: "Shared local cache requires local-cache.max-size".into(), })?, + ratio, )?), Arc::new(CacheCoordinator::default()), ) diff --git a/crates/paimon/src/io/cache/reader.rs b/crates/paimon/src/io/cache/reader.rs index f57b8f7cc..bcb83ddd8 100644 --- a/crates/paimon/src/io/cache/reader.rs +++ b/crates/paimon/src/io/cache/reader.rs @@ -308,6 +308,7 @@ mod tests { }); let cache = Arc::new( LocalCache::new(LocalCacheConfig { + shared: false, dir: None, namespace: "test".to_string(), max_size: Some(16), diff --git a/crates/paimon/src/io/cache/shared.rs b/crates/paimon/src/io/cache/shared.rs index 552018fb9..0ace0c0ea 100644 --- a/crates/paimon/src/io/cache/shared.rs +++ b/crates/paimon/src/io/cache/shared.rs @@ -30,6 +30,7 @@ const MUTATION_MARKER: &str = "mutation-pending"; pub(crate) struct SharedDiskCache { root: PathBuf, max_size: u64, + high_priority_ratio: f64, } fn open_lock(path: &Path) -> io::Result { @@ -143,7 +144,16 @@ impl SharedInvalidationGuard { } impl SharedDiskCache { + #[cfg(test)] pub(crate) fn new(root: PathBuf, max_size: u64) -> crate::Result { + 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 { let root = prepare_cache_root(&root)?; let initialize = || -> io::Result<()> { let gate = open_lock(&root.join("directory.lock"))?; @@ -161,11 +171,25 @@ impl SharedDiskCache { } 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)?; + let used = recover_usage(&root, 0, high_priority_ratio)?; rotate_generation(&root)?; finish_usage(&root, used)?; } @@ -173,7 +197,7 @@ impl SharedDiskCache { // 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)?; + let used = recover_usage(&root, max_size, high_priority_ratio)?; finish_usage(&root, used)?; } Ok(()) @@ -181,7 +205,11 @@ impl SharedDiskCache { initialize().map_err(|e| crate::Error::ConfigInvalid { message: e.to_string(), })?; - Ok(Self { root, max_size }) + Ok(Self { + root, + max_size, + high_priority_ratio, + }) } pub(crate) fn generation(&self) -> Option { @@ -240,6 +268,7 @@ impl SharedDiskCache { 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) @@ -253,14 +282,14 @@ impl SharedDiskCache { } let path = root.join(key.cache_relative_path()); let mut used = if root.join("dirty").exists() { - recover_usage(&root, max_size)? + recover_usage(&root, max_size, high_priority_ratio)? } else { match std::fs::read_to_string(root.join("usage")) .ok() .and_then(|s| s.parse::().ok()) { Some(used) => used, - None => recover_usage(&root, max_size)?, + None => recover_usage(&root, max_size, high_priority_ratio)?, } }; // Keep the marker until publication and accounting both complete. @@ -276,6 +305,7 @@ impl SharedDiskCache { 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); } @@ -388,12 +418,19 @@ fn start_invalidation(root: &Path) -> io::Result<()> { fn finish_invalidation(root: &Path) -> io::Result<()> { // This intentionally invalidates the whole shared directory. - let state = scan_existing_blocks(root); - for key in state.entries.keys() { + 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, scan_existing_blocks(root).current_size) + 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 { @@ -407,14 +444,16 @@ fn clean_usage(root: &Path, max_size: u64) -> Option { (used <= max_size).then_some(used) } -fn recover_usage(root: &Path, limit: u64) -> io::Result { +fn recover_usage(root: &Path, limit: u64, high_priority_ratio: f64) -> io::Result { // 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 = scan_existing_blocks(root); - state.max_size = Some(limit); + 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()))?; } @@ -675,8 +714,7 @@ mod tests { key.block_index = index; cache.put(&key, Bytes::from(vec![42; 64]), &token).await; } - let state = scan_existing_blocks(directory.path()); - assert!(state.current_size <= 512); + assert!(existing_usage(directory.path()) <= 512); let newest = BlockKey::with_namespace("catalog", "data.parquet", 64, 9); let file = directory .path() @@ -685,6 +723,29 @@ mod tests { 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(); @@ -702,7 +763,7 @@ mod tests { .unwrap() .parse() .unwrap(); - assert_eq!(used, scan_existing_blocks(directory.path()).current_size); + assert_eq!(used, existing_usage(directory.path())); assert!(used <= 512); } diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md index 76d6a45c7..dcfa30e65 100644 --- a/docs/src/getting-started.md +++ b/docs/src/getting-started.md @@ -412,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