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 2073ce55 feat(core): protect local metadata and global index cache
blocks (#1184)
2073ce55 is described below
commit 2073ce554b44d8ef72a437ce6d6ad4ea0df673e7
Author: XiaoHongbo <[email protected]>
AuthorDate: Tue Oct 20 21:30:27 2026 +0800
feat(core): protect local metadata and global index cache blocks (#1184)
---
crates/paimon/src/common/options.rs | 8 ++
crates/paimon/src/io/cache/disk.rs | 153 +++++++++++++++++++++++------
crates/paimon/src/io/cache/memory.rs | 108 +++++++++++++++++---
crates/paimon/src/io/cache/mod.rs | 62 ++++++++++--
crates/paimon/src/io/cache/priority_lru.rs | 145 +++++++++++++++++++++++++++
crates/paimon/src/io/cache/reader.rs | 40 ++++++++
docs/src/getting-started.md | 4 +
7 files changed, 472 insertions(+), 48 deletions(-)
diff --git a/crates/paimon/src/common/options.rs
b/crates/paimon/src/common/options.rs
index 86391377..c75f1954 100644
--- a/crates/paimon/src/common/options.rs
+++ b/crates/paimon/src/common/options.rs
@@ -85,6 +85,10 @@ impl CatalogOptions {
/// Maximum total size of the local block cache.
pub const LOCAL_CACHE_MAX_SIZE: &'static str = "local-cache.max-size";
+ /// Fraction of the local cache protected for metadata and global indexes.
+ pub const LOCAL_CACHE_HIGH_PRIORITY_POOL_RATIO: &'static str =
+ "local-cache.high-priority-pool-ratio";
+
/// Block size used by the local cache.
pub const LOCAL_CACHE_BLOCK_SIZE: &'static str = "local-cache.block-size";
@@ -293,6 +297,10 @@ mod tests {
assert_eq!(CatalogOptions::LOCAL_CACHE_ENABLED, "local-cache.enabled");
assert_eq!(CatalogOptions::LOCAL_CACHE_DIR, "local-cache.dir");
assert_eq!(CatalogOptions::LOCAL_CACHE_MAX_SIZE,
"local-cache.max-size");
+ assert_eq!(
+ CatalogOptions::LOCAL_CACHE_HIGH_PRIORITY_POOL_RATIO,
+ "local-cache.high-priority-pool-ratio"
+ );
assert_eq!(
CatalogOptions::LOCAL_CACHE_BLOCK_SIZE,
"local-cache.block-size"
diff --git a/crates/paimon/src/io/cache/disk.rs
b/crates/paimon/src/io/cache/disk.rs
index 72cb84f9..b731e303 100644
--- a/crates/paimon/src/io/cache/disk.rs
+++ b/crates/paimon/src/io/cache/disk.rs
@@ -16,7 +16,6 @@
// under the License.
use bytes::Bytes;
-use indexmap::IndexMap;
use sha2::{Digest, Sha256};
use std::collections::{HashMap, HashSet};
use std::io::Read;
@@ -24,6 +23,9 @@ use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, OnceLock, Weak};
use tokio::io::AsyncWriteExt;
+#[cfg(test)]
+use super::priority_lru::DEFAULT_HIGH_PRIORITY_POOL_RATIO;
+use super::priority_lru::{protected_limit, PriorityLru};
use super::state::{BlockKey, CacheCoordinator, LogicalPath};
const CACHE_MAGIC: &[u8; 8] = b"PAIMONLC";
@@ -64,24 +66,47 @@ pub(super) struct DiskCache {
coordinator: Arc<CacheCoordinator>,
}
-#[derive(Debug, Default)]
+#[derive(Debug)]
struct CacheState {
- entries: IndexMap<BlockKey, u64>,
+ entries: PriorityLru<u64>,
paths: HashMap<LogicalPath, HashSet<BlockKey>>,
current_size: u64,
max_size: Option<u64>,
+ high_priority_ratio: f64,
+}
+
+impl CacheState {
+ fn new(max_size: Option<u64>, high_priority_ratio: f64) -> Self {
+ Self {
+ entries: PriorityLru::new(protected_limit(max_size,
high_priority_ratio)),
+ paths: HashMap::new(),
+ current_size: 0,
+ max_size,
+ high_priority_ratio,
+ }
+ }
}
impl DiskCache {
#[cfg(test)]
pub(super) fn new(root: impl AsRef<Path>, max_size: Option<u64>) ->
crate::Result<Self> {
+ Self::new_with_ratio(root, max_size, DEFAULT_HIGH_PRIORITY_POOL_RATIO)
+ }
+
+ #[cfg(test)]
+ pub(super) fn new_with_ratio(
+ root: impl AsRef<Path>,
+ max_size: Option<u64>,
+ ratio: f64,
+ ) -> crate::Result<Self> {
let root = prepare_cache_root(root.as_ref())?;
- Ok(Self::new_resolved(root, max_size))
+ Ok(Self::new_resolved(root, max_size, ratio))
}
pub(super) fn shared(
root: impl AsRef<Path>,
max_size: Option<u64>,
+ ratio: f64,
) -> crate::Result<Arc<Self>> {
let root = prepare_cache_root(root.as_ref())?;
let mut registry = disk_cache_registry()
@@ -91,34 +116,39 @@ impl DiskCache {
registry.retain(|_, cache| cache.strong_count() > 0);
}
if let Some(cache) = registry.get(&root).and_then(Weak::upgrade) {
- cache.tighten_max_size(max_size);
+ cache.tighten_max_size(max_size, ratio)?;
return Ok(cache);
}
- let cache = Arc::new(Self::new_resolved(root.clone(), max_size));
+ let cache = Arc::new(Self::new_resolved(root.clone(), max_size,
ratio));
registry.insert(root, Arc::downgrade(&cache));
Ok(cache)
}
- fn new_resolved(root: PathBuf, max_size: Option<u64>) -> Self {
+ fn new_resolved(root: PathBuf, max_size: Option<u64>, ratio: f64) -> Self {
Self {
root,
- state: Mutex::new(CacheState {
- max_size,
- ..CacheState::default()
- }),
+ state: Mutex::new(CacheState::new(max_size, ratio)),
recovered: tokio::sync::OnceCell::new(),
coordinator: Arc::new(CacheCoordinator::default()),
}
}
- fn tighten_max_size(&self, requested: Option<u64>) {
+ fn tighten_max_size(&self, requested: Option<u64>, ratio: f64) ->
crate::Result<()> {
let mut state = self.state.lock().unwrap_or_else(|error|
error.into_inner());
+ if state.high_priority_ratio != ratio {
+ return Err(crate::Error::ConfigInvalid {
+ message: "Local caches using the same directory must use the
same local-cache.high-priority-pool-ratio".into(),
+ });
+ }
state.max_size = match (state.max_size, requested) {
(None, requested) => requested,
(current, None) => current,
(Some(current), Some(requested)) => Some(current.min(requested)),
};
+ let limit = protected_limit(state.max_size, ratio);
+ state.entries.set_protected_limit(limit);
+ Ok(())
}
pub(super) async fn ensure_recovered(&self) {
@@ -141,7 +171,7 @@ impl DiskCache {
.state
.lock()
.unwrap_or_else(|error| error.into_inner());
- for (key, encoded_size) in recovered.entries {
+ for (key, encoded_size) in recovered {
insert_state_entry(&mut state, key, encoded_size);
}
}
@@ -368,11 +398,7 @@ impl DiskCache {
fn touch_entry(&self, key: &BlockKey) -> bool {
let mut state = self.state.lock().unwrap_or_else(|error|
error.into_inner());
- let Some(encoded_size) = state.entries.shift_remove(key) else {
- return false;
- };
- state.entries.insert(key.clone(), encoded_size);
- true
+ state.entries.touch(key).is_some()
}
fn forget_entry(&self, key: &BlockKey) {
@@ -400,7 +426,7 @@ fn collect_evictions(state: &mut CacheState) ->
Vec<BlockKey> {
let mut to_evict = Vec::new();
if let Some(max_size) = state.max_size {
while state.current_size > max_size {
- let Some((eldest, size)) = state.entries.shift_remove_index(0)
else {
+ let Some((eldest, _, size)) = state.entries.pop_lru() else {
break;
};
state.current_size = state.current_size.saturating_sub(size);
@@ -425,7 +451,10 @@ fn logical_path_matches_prefix(path: &LogicalPath,
namespace: &str, prefix: &str
}
fn insert_state_entry(state: &mut CacheState, key: BlockKey, encoded_size:
u64) {
- if let Some(previous_size) = state.entries.shift_remove(&key) {
+ if let Some((_, previous_size)) = state
+ .entries
+ .insert(key.clone(), encoded_size, encoded_size)
+ {
state.current_size = state.current_size.saturating_sub(previous_size);
}
state
@@ -433,12 +462,11 @@ fn insert_state_entry(state: &mut CacheState, key:
BlockKey, encoded_size: u64)
.entry(logical_path(&key))
.or_default()
.insert(key.clone());
- state.entries.insert(key, encoded_size);
state.current_size = state.current_size.saturating_add(encoded_size);
}
fn remove_state_entry(state: &mut CacheState, key: &BlockKey) {
- if let Some(encoded_size) = state.entries.shift_remove(key) {
+ if let Some((_, encoded_size)) = state.entries.remove(key) {
state.current_size = state.current_size.saturating_sub(encoded_size);
remove_path_index_entry(state, key);
}
@@ -516,21 +544,21 @@ fn cleanup_temporary_files(directory: &Path) ->
std::io::Result<()> {
Ok(())
}
-fn scan_existing_blocks(root: &Path) -> CacheState {
+fn scan_existing_blocks(root: &Path) -> Vec<(BlockKey, u64)> {
let mut discovered = Vec::new();
collect_cache_files(root, &mut discovered);
discovered.sort_by_key(|(modified, _, _, _)| *modified);
- let mut state = CacheState::default();
+ let mut entries = Vec::with_capacity(discovered.len());
for (_, path, key, encoded_size) in discovered {
if root.join(key.cache_relative_path()) != path {
let _ = std::fs::remove_file(path);
continue;
}
- insert_state_entry(&mut state, key, encoded_size);
+ entries.push((key, encoded_size));
}
- state
+ entries
}
fn collect_cache_files(
@@ -872,8 +900,14 @@ mod tests {
let payload = Bytes::from_static(b"data");
let one_block_size = encode_block(&first_key, &payload).len() as u64;
- let first = DiskCache::shared(directory.path(), None).unwrap();
- let second = DiskCache::shared(directory.path().join("."),
Some(one_block_size)).unwrap();
+ let first =
+ DiskCache::shared(directory.path(), None,
DEFAULT_HIGH_PRIORITY_POOL_RATIO).unwrap();
+ let second = DiskCache::shared(
+ directory.path().join("."),
+ Some(one_block_size),
+ DEFAULT_HIGH_PRIORITY_POOL_RATIO,
+ )
+ .unwrap();
assert!(Arc::ptr_eq(&first, &second));
first.put_block(&first_key, payload.clone()).await;
@@ -992,6 +1026,69 @@ mod tests {
assert_eq!(cache.get_block(&second).await, Some(payload));
}
+ #[tokio::test]
+ async fn test_disk_cache_protects_global_index_from_data_scan() {
+ let directory = tempfile::tempdir().unwrap();
+ let high = BlockKey::new("global-index-1.index", 4, 0);
+ let payload = Bytes::from_static(b"data");
+ let one_block_size = encode_block(&high, &payload).len() as u64;
+ let cache = DiskCache::new(directory.path(), Some(4 *
one_block_size)).unwrap();
+ cache.put_block(&high, payload.clone()).await;
+ for index in 1..=4 {
+ cache
+ .put_block(
+ &BlockKey::new(format!("data-{index}"), 4, 0),
+ payload.clone(),
+ )
+ .await;
+ }
+
+ assert_eq!(cache.get_block(&high).await, Some(payload));
+ assert_eq!(cache.get_block(&BlockKey::new("data-1", 4, 0)).await,
None);
+ }
+
+ #[tokio::test]
+ async fn test_disk_cache_recovery_preserves_priority() {
+ let directory = tempfile::tempdir().unwrap();
+ let high = BlockKey::new("global-index-1.index", 4, 0);
+ let payload = Bytes::from_static(b"data");
+ let one_high_block = encode_block(&high, &payload).len() as u64;
+ let initial = DiskCache::new(directory.path(), None).unwrap();
+ initial.put_block(&high, payload.clone()).await;
+ for index in 1..=4 {
+ initial
+ .put_block(
+ &BlockKey::new(format!("data-{index}"), 4, 0),
+ payload.clone(),
+ )
+ .await;
+ }
+ drop(initial);
+
+ let recovered = DiskCache::new(directory.path(), Some(4 *
one_high_block)).unwrap();
+ recovered.ensure_recovered().await;
+ assert_eq!(recovered.get_block(&high).await, Some(payload));
+ let mut data_misses = 0;
+ for index in 1..=4 {
+ if recovered
+ .get_block(&BlockKey::new(format!("data-{index}"), 4, 0))
+ .await
+ .is_none()
+ {
+ data_misses += 1;
+ }
+ }
+ assert!(data_misses >= 1);
+ }
+
+ #[test]
+ fn test_disk_cache_shared_directory_rejects_different_priority_ratio() {
+ let directory = tempfile::tempdir().unwrap();
+ let _first = DiskCache::shared(directory.path(), Some(1024),
0.25).unwrap();
+ let error = DiskCache::shared(directory.path(), Some(1024),
0.5).unwrap_err();
+ assert!(matches!(error, crate::Error::ConfigInvalid { .. }));
+ }
+
#[tokio::test]
async fn test_disk_cache_lru_hit_refreshes_access_order() {
let directory = tempfile::tempdir().unwrap();
diff --git a/crates/paimon/src/io/cache/memory.rs
b/crates/paimon/src/io/cache/memory.rs
index c50cd91a..a57790d1 100644
--- a/crates/paimon/src/io/cache/memory.rs
+++ b/crates/paimon/src/io/cache/memory.rs
@@ -16,9 +16,11 @@
// under the License.
use bytes::Bytes;
-use lru::LruCache;
use std::sync::Mutex;
+#[cfg(test)]
+use super::priority_lru::DEFAULT_HIGH_PRIORITY_POOL_RATIO;
+use super::priority_lru::{protected_limit, PriorityLru};
use super::state::BlockKey;
#[derive(Debug)]
@@ -28,16 +30,21 @@ pub(super) struct MemoryCache {
#[derive(Debug)]
struct MemoryState {
- entries: LruCache<BlockKey, Bytes>,
+ entries: PriorityLru<Bytes>,
current_size: u64,
max_size: Option<u64>,
}
impl MemoryCache {
+ #[cfg(test)]
pub(super) fn new(max_size: Option<u64>) -> Self {
+ Self::with_ratio(max_size, DEFAULT_HIGH_PRIORITY_POOL_RATIO)
+ }
+
+ pub(super) fn with_ratio(max_size: Option<u64>, ratio: f64) -> Self {
Self {
state: Mutex::new(MemoryState {
- entries: LruCache::unbounded(),
+ entries: PriorityLru::new(protected_limit(max_size, ratio)),
current_size: 0,
max_size,
}),
@@ -46,7 +53,7 @@ impl MemoryCache {
pub(super) fn get_block(&self, key: &BlockKey) -> Option<Bytes> {
let mut state = self.state.lock().unwrap_or_else(|error|
error.into_inner());
- state.entries.get(key).cloned()
+ state.entries.touch(key).cloned()
}
pub(super) fn put_block(&self, key: &BlockKey, payload: Bytes) {
@@ -58,25 +65,25 @@ impl MemoryCache {
{
return;
}
- if let Some(previous) = state.entries.put(key.clone(), payload) {
- state.current_size =
state.current_size.saturating_sub(previous.len() as u64);
+ if let Some((_, previous_size)) = state.entries.insert(key.clone(),
payload, payload_size) {
+ state.current_size =
state.current_size.saturating_sub(previous_size);
}
state.current_size = state.current_size.saturating_add(payload_size);
while state
.max_size
.is_some_and(|max_size| state.current_size > max_size)
{
- let Some((_, payload)) = state.entries.pop_lru() else {
+ let Some((_, _, size)) = state.entries.pop_lru() else {
break;
};
- state.current_size =
state.current_size.saturating_sub(payload.len() as u64);
+ state.current_size = state.current_size.saturating_sub(size);
}
}
pub(super) fn remove_block(&self, key: &BlockKey) {
let mut state = self.state.lock().unwrap_or_else(|error|
error.into_inner());
- if let Some(payload) = state.entries.pop(key) {
- state.current_size =
state.current_size.saturating_sub(payload.len() as u64);
+ if let Some((_, size)) = state.entries.remove(key) {
+ state.current_size = state.current_size.saturating_sub(size);
}
}
@@ -93,14 +100,13 @@ impl MemoryCache {
let mut state = self.state.lock().unwrap_or_else(|error|
error.into_inner());
let keys = state
.entries
- .iter()
- .map(|(key, _)| key)
+ .keys()
.filter(|key| matches(key))
.cloned()
.collect::<Vec<_>>();
for key in keys {
- if let Some(payload) = state.entries.pop(&key) {
- state.current_size =
state.current_size.saturating_sub(payload.len() as u64);
+ if let Some((_, size)) = state.entries.remove(&key) {
+ state.current_size = state.current_size.saturating_sub(size);
}
}
}
@@ -136,4 +142,78 @@ mod tests {
assert_eq!(cache.get_block(&key), None);
}
+
+ #[test]
+ fn test_memory_cache_protects_metadata_from_data_scan() {
+ let cache = MemoryCache::new(Some(16));
+ let metadata = BlockKey::new("snapshot-1", 4, 0);
+ cache.put_block(&metadata, Bytes::from_static(b"meta"));
+ for index in 1..=4 {
+ cache.put_block(
+ &BlockKey::new(format!("data-{index}"), 4, 0),
+ Bytes::from_static(b"data"),
+ );
+ }
+
+ assert_eq!(
+ cache.get_block(&metadata),
+ Some(Bytes::from_static(b"meta"))
+ );
+ assert_eq!(cache.get_block(&BlockKey::new("data-1", 4, 0)), None);
+ }
+
+ #[test]
+ fn test_memory_cache_high_priority_uses_idle_capacity() {
+ let cache = MemoryCache::new(Some(16));
+ for index in 1..=5 {
+ cache.put_block(
+ &BlockKey::new(format!("snapshot-{index}"), 4, 0),
+ Bytes::from_static(b"meta"),
+ );
+ }
+
+ assert_eq!(cache.get_block(&BlockKey::new("snapshot-1", 4, 0)), None);
+ for index in 2..=5 {
+ assert_eq!(
+ cache.get_block(&BlockKey::new(format!("snapshot-{index}"), 4,
0)),
+ Some(Bytes::from_static(b"meta"))
+ );
+ }
+ }
+
+ #[test]
+ fn test_memory_cache_zero_ratio_restores_single_lru() {
+ let cache = MemoryCache::with_ratio(Some(16), 0.0);
+ let metadata = BlockKey::new("snapshot-1", 4, 0);
+ cache.put_block(&metadata, Bytes::from_static(b"meta"));
+ for index in 1..=4 {
+ cache.put_block(
+ &BlockKey::new(format!("data-{index}"), 4, 0),
+ Bytes::from_static(b"data"),
+ );
+ }
+ assert_eq!(cache.get_block(&metadata), None);
+ }
+
+ #[test]
+ fn test_memory_cache_high_priority_overflow_competes_with_data() {
+ let cache = MemoryCache::new(Some(16));
+ for index in 1..=2 {
+ cache.put_block(
+ &BlockKey::new(format!("snapshot-{index}"), 4, 0),
+ Bytes::from_static(b"meta"),
+ );
+ }
+ for index in 1..=3 {
+ cache.put_block(
+ &BlockKey::new(format!("data-{index}"), 4, 0),
+ Bytes::from_static(b"data"),
+ );
+ }
+ assert_eq!(cache.get_block(&BlockKey::new("snapshot-1", 4, 0)), None);
+ assert_eq!(
+ cache.get_block(&BlockKey::new("snapshot-2", 4, 0)),
+ Some(Bytes::from_static(b"meta"))
+ );
+ }
}
diff --git a/crates/paimon/src/io/cache/mod.rs
b/crates/paimon/src/io/cache/mod.rs
index 74c636d0..42eb32af 100644
--- a/crates/paimon/src/io/cache/mod.rs
+++ b/crates/paimon/src/io/cache/mod.rs
@@ -18,11 +18,13 @@
mod disk;
mod file_type;
mod memory;
+mod priority_lru;
mod reader;
mod state;
use self::file_type::FileType;
use self::memory::MemoryCache;
+use self::priority_lru::DEFAULT_HIGH_PRIORITY_POOL_RATIO;
use self::state::{BlockKey, CacheCoordinator, CacheReadToken};
use crate::common::{CatalogOptions, Options};
use crate::io::FileBlockCache;
@@ -75,7 +77,12 @@ impl LocalCache {
})
}
+ #[cfg(test)]
pub(super) fn new(config: LocalCacheConfig) -> crate::Result<Self> {
+ Self::new_with_ratio(config, DEFAULT_HIGH_PRIORITY_POOL_RATIO)
+ }
+
+ fn new_with_ratio(config: LocalCacheConfig, ratio: f64) ->
crate::Result<Self> {
let file_size_capacity = config
.max_size
.map(|max_size| max_size / config.block_size)
@@ -83,12 +90,12 @@ impl LocalCache {
.unwrap_or(DEFAULT_FILE_SIZE_CAPACITY)
.clamp(1, DEFAULT_FILE_SIZE_CAPACITY);
let (backend, coordinator) = if let Some(dir) = config.dir {
- let disk = DiskCache::shared(dir.join(CACHE_DIRECTORY_NAME),
config.max_size)?;
+ let disk = DiskCache::shared(dir.join(CACHE_DIRECTORY_NAME),
config.max_size, ratio)?;
let coordinator = disk.coordinator();
(CacheBackend::Disk(disk), coordinator)
} else {
(
- CacheBackend::Memory(MemoryCache::new(config.max_size)),
+ CacheBackend::Memory(MemoryCache::with_ratio(config.max_size,
ratio)),
Arc::new(CacheCoordinator::default()),
)
};
@@ -234,10 +241,34 @@ pub(crate) fn create_local_cache_with_namespace(
cache_options: &Options,
namespace_options: &Options,
) -> crate::Result<Option<Arc<LocalCache>>> {
- LocalCacheConfig::from_options_with_namespace(cache_options,
namespace_options)?
- .map(LocalCache::new)
- .transpose()
- .map(|cache| cache.map(Arc::new))
+ let Some(config) =
+ LocalCacheConfig::from_options_with_namespace(cache_options,
namespace_options)?
+ else {
+ return Ok(None);
+ };
+ let ratio = parse_high_priority_ratio(cache_options)?;
+ LocalCache::new_with_ratio(config, ratio)
+ .map(Arc::new)
+ .map(Some)
+}
+
+fn parse_high_priority_ratio(options: &Options) -> crate::Result<f64> {
+ let key = CatalogOptions::LOCAL_CACHE_HIGH_PRIORITY_POOL_RATIO;
+ let Some(value) = options.get(key) else {
+ return Ok(DEFAULT_HIGH_PRIORITY_POOL_RATIO);
+ };
+ let ratio = value
+ .trim()
+ .parse::<f64>()
+ .map_err(|_| crate::Error::ConfigInvalid {
+ message: format!("{key} must be a number in [0, 1)"),
+ })?;
+ if !(0.0..1.0).contains(&ratio) {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!("{key} must be in [0, 1)"),
+ });
+ }
+ Ok(ratio)
}
#[derive(Debug)]
@@ -369,6 +400,7 @@ mod tests {
assert_eq!(config.dir, None);
assert_eq!(config.max_size, None);
assert_eq!(config.block_size, 1024 * 1024);
+ assert_eq!(parse_high_priority_ratio(&options).unwrap(), 0.25);
assert_eq!(
config.whitelist,
HashSet::from([FileType::Meta, FileType::GlobalIndex])
@@ -467,6 +499,24 @@ mod tests {
assert!(error.to_string().contains("local-cache.enabled"));
}
+ #[test]
+ fn test_local_cache_high_priority_ratio_validation() {
+ let mut options = Options::new();
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ for (value, expected) in [("0", 0.0), ("0.75", 0.75)] {
+ options.set(CatalogOptions::LOCAL_CACHE_HIGH_PRIORITY_POOL_RATIO,
value);
+ assert_eq!(parse_high_priority_ratio(&options).unwrap(), expected);
+ }
+ for value in ["-0.1", "1", "NaN", "inf", "garbage"] {
+ options.set(CatalogOptions::LOCAL_CACHE_HIGH_PRIORITY_POOL_RATIO,
value);
+ let error = create_local_cache(&options).unwrap_err();
+ assert!(matches!(error, crate::Error::ConfigInvalid { .. }));
+ assert!(error
+ .to_string()
+
.contains(CatalogOptions::LOCAL_CACHE_HIGH_PRIORITY_POOL_RATIO));
+ }
+ }
+
#[tokio::test]
async fn test_local_cache_file_size_is_removed_with_path_invalidation() {
let directory = tempfile::tempdir().unwrap();
diff --git a/crates/paimon/src/io/cache/priority_lru.rs
b/crates/paimon/src/io/cache/priority_lru.rs
new file mode 100644
index 00000000..bf498428
--- /dev/null
+++ b/crates/paimon/src/io/cache/priority_lru.rs
@@ -0,0 +1,145 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use lru::LruCache;
+
+use super::file_type::FileType;
+use super::state::BlockKey;
+
+pub(super) const DEFAULT_HIGH_PRIORITY_POOL_RATIO: f64 = 0.25;
+
+pub(super) fn protected_limit(max_size: Option<u64>, ratio: f64) -> u64 {
+ if ratio == 0.0 {
+ 0
+ } else {
+ max_size
+ .map(|size| (size as f64 * ratio) as u64)
+ .unwrap_or(u64::MAX)
+ }
+}
+
+#[derive(Debug)]
+struct Entry<V> {
+ value: V,
+ size: u64,
+}
+
+/// Newest high-priority blocks occupy a protected LRU; excess blocks compete
+/// with ordinary blocks in the regular LRU. Either class may use idle
capacity.
+#[derive(Debug)]
+pub(super) struct PriorityLru<V> {
+ protected: LruCache<BlockKey, Entry<V>>,
+ regular: LruCache<BlockKey, Entry<V>>,
+ protected_size: u64,
+ protected_limit: u64,
+}
+
+impl<V> PriorityLru<V> {
+ pub(super) fn new(protected_limit: u64) -> Self {
+ Self {
+ protected: LruCache::unbounded(),
+ regular: LruCache::unbounded(),
+ protected_size: 0,
+ protected_limit,
+ }
+ }
+
+ pub(super) fn set_protected_limit(&mut self, limit: u64) {
+ self.protected_limit = limit;
+ self.demote_excess();
+ }
+
+ pub(super) fn contains_key(&self, key: &BlockKey) -> bool {
+ self.protected.contains(key) || self.regular.contains(key)
+ }
+
+ pub(super) fn keys(&self) -> impl Iterator<Item = &BlockKey> {
+ self.protected
+ .iter()
+ .map(|(key, _)| key)
+ .chain(self.regular.iter().map(|(key, _)| key))
+ }
+
+ pub(super) fn touch(&mut self, key: &BlockKey) -> Option<&V> {
+ if self.protected.contains(key) {
+ return self.protected.get(key).map(|entry| &entry.value);
+ }
+ if !self.regular.contains(key) {
+ return None;
+ }
+ if self.protected_limit > 0 && Self::high_priority(key) {
+ let entry = self.regular.pop(key)?;
+ self.protected_size =
self.protected_size.saturating_add(entry.size);
+ self.protected.put(key.clone(), entry);
+ self.demote_excess();
+ if self.protected.contains(key) {
+ self.protected.get(key).map(|entry| &entry.value)
+ } else {
+ self.regular.get(key).map(|entry| &entry.value)
+ }
+ } else {
+ self.regular.get(key).map(|entry| &entry.value)
+ }
+ }
+
+ pub(super) fn insert(&mut self, key: BlockKey, value: V, size: u64) ->
Option<(V, u64)> {
+ let previous = self.remove(&key);
+ let entry = Entry { value, size };
+ if self.protected_limit > 0 && Self::high_priority(&key) {
+ self.protected_size = self.protected_size.saturating_add(size);
+ self.protected.put(key, entry);
+ self.demote_excess();
+ } else {
+ self.regular.put(key, entry);
+ }
+ previous
+ }
+
+ pub(super) fn remove(&mut self, key: &BlockKey) -> Option<(V, u64)> {
+ if let Some(entry) = self.protected.pop(key) {
+ self.protected_size =
self.protected_size.saturating_sub(entry.size);
+ return Some((entry.value, entry.size));
+ }
+ self.regular.pop(key).map(|entry| (entry.value, entry.size))
+ }
+
+ pub(super) fn pop_lru(&mut self) -> Option<(BlockKey, V, u64)> {
+ if let Some((key, entry)) = self.regular.pop_lru() {
+ return Some((key, entry.value, entry.size));
+ }
+ let (key, entry) = self.protected.pop_lru()?;
+ self.protected_size = self.protected_size.saturating_sub(entry.size);
+ Some((key, entry.value, entry.size))
+ }
+
+ fn demote_excess(&mut self) {
+ while self.protected_size > self.protected_limit {
+ let Some((key, entry)) = self.protected.pop_lru() else {
+ break;
+ };
+ self.protected_size =
self.protected_size.saturating_sub(entry.size);
+ self.regular.put(key, entry);
+ }
+ }
+
+ fn high_priority(key: &BlockKey) -> bool {
+ matches!(
+ FileType::classify(&key.path),
+ FileType::Meta | FileType::GlobalIndex
+ )
+ }
+}
diff --git a/crates/paimon/src/io/cache/reader.rs
b/crates/paimon/src/io/cache/reader.rs
index 240b163d..edb00a0a 100644
--- a/crates/paimon/src/io/cache/reader.rs
+++ b/crates/paimon/src/io/cache/reader.rs
@@ -297,6 +297,46 @@ mod tests {
assert_eq!(delegate.reads.load(Ordering::SeqCst), 3);
}
+ #[tokio::test]
+ async fn test_data_reads_do_not_evict_cached_snapshot_block() {
+ let delegate = Arc::new(CountingReader {
+ data: Bytes::from_static(b"data"),
+ reads: AtomicUsize::new(0),
+ });
+ let cache = Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ dir: None,
+ namespace: "test".to_string(),
+ max_size: Some(16),
+ block_size: 4,
+ whitelist: std::collections::HashSet::from([
+ FileType::Meta,
+ FileType::StructuredData,
+ ]),
+ })
+ .unwrap(),
+ );
+ let snapshot = CachedFileReader::new(
+ delegate.clone(),
+ "s3://bucket/table/snapshot/snapshot-1",
+ 4,
+ cache.clone(),
+ );
+ snapshot.read(0..4).await.unwrap();
+ for index in 1..=4 {
+ let reader = CachedFileReader::new(
+ delegate.clone(),
+ format!("s3://bucket/table/data/data-{index}.parquet"),
+ 4,
+ cache.clone(),
+ );
+ reader.read(0..4).await.unwrap();
+ }
+ snapshot.read(0..4).await.unwrap();
+
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 5);
+ }
+
#[derive(Debug)]
struct SlowCountingReader {
data: Bytes,
diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md
index 49fca6a1..8c99f8c2 100644
--- a/docs/src/getting-started.md
+++ b/docs/src/getting-started.md
@@ -402,6 +402,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.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.
|
| `local-cache.whitelist` | `meta,global-index` | Comma-separated eligible
types: `meta`, `global-index`, `bucket-index`, `data`, `structured-data`, and
`file-index`, or `*` for all of them. |
@@ -409,6 +410,9 @@ Use
`local-cache.whitelist=meta,global-index,structured-data` to cache data
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.
+
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
fingerprint and the canonical storage object path, so catalogs can safely use
