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 1f1a47c0 [core] Coalesce consecutive local cache misses (#1218)
1f1a47c0 is described below
commit 1f1a47c0101b3629188a72d0e7207fe7d2cc80a3
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Oct 25 23:09:18 2026 +0800
[core] Coalesce consecutive local cache misses (#1218)
---
crates/paimon/src/io/cache/mod.rs | 7 +
crates/paimon/src/io/cache/reader.rs | 410 ++++++++++++++++++++++++++++++++++-
crates/paimon/src/io/cache/shared.rs | 52 ++++-
3 files changed, 459 insertions(+), 10 deletions(-)
diff --git a/crates/paimon/src/io/cache/mod.rs
b/crates/paimon/src/io/cache/mod.rs
index 5891f283..57f15967 100644
--- a/crates/paimon/src/io/cache/mod.rs
+++ b/crates/paimon/src/io/cache/mod.rs
@@ -236,6 +236,13 @@ impl LocalCache {
}
}
+ async fn shared_load_locks(&self, keys: &[BlockKey]) -> Vec<std::fs::File>
{
+ match &self.backend {
+ CacheBackend::SharedDisk(disk) => disk.load_locks(keys).await,
+ _ => Vec::new(),
+ }
+ }
+
async fn block_load_lock(&self, key: &BlockKey) ->
Arc<tokio::sync::Mutex<()>> {
self.coordinator.block_load_lock(key).await
}
diff --git a/crates/paimon/src/io/cache/reader.rs
b/crates/paimon/src/io/cache/reader.rs
index 35781cc5..d60099f7 100644
--- a/crates/paimon/src/io/cache/reader.rs
+++ b/crates/paimon/src/io/cache/reader.rs
@@ -145,6 +145,114 @@ impl CachedFileReader {
result
}
+ async fn read_blocks(&self, first: u64, last: u64) ->
crate::Result<Vec<Bytes>> {
+ if first == last {
+ return Ok(vec![self.read_block(first).await?]);
+ }
+ let size = self.cache.block_size();
+ let mut blocks = Vec::new();
+ let mut locks = Vec::new();
+ // Acquire missing-block locks in offset order, including overlapping
reads.
+ for index in first..=last {
+ let key = self.cache.block_key(&self.path, index);
+ let len = (self.file_size - index * size).min(size) as usize;
+ let mut block = self.cache.get_block(&key, len,
&self.read_token).await;
+ if block.as_ref().is_some_and(|b| b.len() != len) {
+ self.cache.remove_block(&key).await;
+ block = None;
+ }
+ if block.is_none() {
+ let lock = self.cache.block_load_lock(&key).await;
+ let guard = lock.clone().lock_owned().await;
+ block = self.cache.get_block(&key, len,
&self.read_token).await;
+ if block.as_ref().is_some_and(|b| b.len() != len) {
+ self.cache.remove_block(&key).await;
+ block = None;
+ }
+ locks.push(Some((key, lock, guard)));
+ } else {
+ locks.push(None);
+ }
+ blocks.push(block);
+ }
+ // Try shared stripes after local locks; collisions must not stall a
batch.
+ let keys: Vec<_> = locks
+ .iter()
+ .flatten()
+ .map(|(key, _, _)| key.clone())
+ .collect();
+ let shared_guards =
Arc::new(self.cache.shared_load_locks(&keys).await);
+ for (index, entry) in locks.iter().enumerate() {
+ if let Some((key, _, _)) = entry {
+ let len = (self.file_size - (first + index as u64) *
size).min(size) as usize;
+ let block = self.cache.get_block(key, len,
&self.read_token).await;
+ if block.as_ref().is_some_and(|b| b.len() != len) {
+ self.cache.remove_block(key).await;
+ blocks[index] = None;
+ } else {
+ blocks[index] = block;
+ }
+ }
+ }
+ let result = async {
+ let mut i = 0;
+ while i < blocks.len() {
+ if blocks[i].is_some() {
+ i += 1;
+ continue;
+ }
+ let begin = i;
+ while i < blocks.len() && blocks[i].is_none() {
+ i += 1;
+ }
+ let start = (first + begin as u64) * size;
+ let end = (first + i as
u64).saturating_mul(size).min(self.file_size);
+ let payload = self.delegate.read(start..end).await?;
+ if payload.len() as u64 != end - start {
+ return Err(crate::Error::DataInvalid {
+ message: "Truncated cache range read".to_string(),
+ source: None,
+ });
+ }
+ for (offset, slot) in blocks[begin..i].iter_mut().enumerate() {
+ let lo = offset * size as usize;
+ let hi = (lo + size as usize).min(payload.len());
+ // Each cached block owns only its bytes, not the entire
fetched range.
+ let block = Bytes::copy_from_slice(&payload[lo..hi]);
+ let key = self
+ .cache
+ .block_key(&self.path, first + (begin + offset) as
u64);
+ if let Some(fill_slot) = self.cache.background_fill_slot()
{
+ let (_, lock, guard) = locks[begin +
offset].take().unwrap();
+ let shared_guards = shared_guards.clone();
+ let cache = self.cache.clone();
+ let token = self.read_token.clone();
+ let fill_block = block.clone();
+ tokio::spawn(async move {
+ let _slot = fill_slot;
+ let _shared_guards = shared_guards;
+ cache.put_block(&key, fill_block, &token).await;
+ drop(guard);
+ cache.release_block_load_lock(&key, &lock).await;
+ });
+ } else {
+ self.cache
+ .put_block(&key, block.clone(), &self.read_token)
+ .await;
+ }
+ *slot = Some(block);
+ }
+ }
+ Ok(blocks.into_iter().map(Option::unwrap).collect())
+ }
+ .await;
+ for (key, lock, guard) in locks.into_iter().flatten() {
+ drop(guard);
+ self.cache.release_block_load_lock(&key, &lock).await;
+ }
+ result
+ }
+
pub(crate) async fn read_full(&self) -> crate::Result<Bytes> {
if self.file_size == 0 {
return Ok(Bytes::new());
@@ -244,12 +352,18 @@ impl FileRead for CachedFileReader {
})?;
let mut output = BytesMut::with_capacity(output_len);
- for block_index in first_block..=last_block {
- let block = self.read_block(block_index).await?;
- let block_start = block_index * block_size;
- let copy_start = range.start.max(block_start) - block_start;
- let copy_end = end.min(block_start + block.len() as u64) -
block_start;
- output.extend_from_slice(&block[copy_start as usize..copy_end as
usize]);
+ // Bound both temporary bytes and the number of held block locks.
+ let batch_size = (8 * 1024 * 1024 / block_size).clamp(1, 32);
+ let mut first = first_block;
+ while first <= last_block {
+ let last = first.saturating_add(batch_size - 1).min(last_block);
+ for (offset, block) in self.read_blocks(first,
last).await?.into_iter().enumerate() {
+ let block_start = (first + offset as u64) * block_size;
+ let copy_start = range.start.max(block_start) - block_start;
+ let copy_end = end.min(block_start + block.len() as u64) -
block_start;
+ output.extend_from_slice(&block[copy_start as usize..copy_end
as usize]);
+ }
+ first = last + 1;
}
Ok(output.freeze())
@@ -271,6 +385,167 @@ mod tests {
reads: AtomicUsize,
}
+ fn test_cache() -> Arc<LocalCache> {
+ Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ shared: false,
+ dir: None,
+ namespace: "test".into(),
+ max_size: None,
+ block_size: 4,
+ whitelist: std::collections::HashSet::from([FileType::Meta]),
+ })
+ .unwrap(),
+ )
+ }
+
+ #[tokio::test]
+ async fn test_coalesced_misses_preserve_warm_holes_and_partial_tail() {
+ let delegate = Arc::new(CountingReader {
+ data: Bytes::from_static(b"abcdefghijklmn"),
+ reads: AtomicUsize::new(0),
+ });
+ let reader = CachedFileReader::new(delegate.clone(), "snapshot-1", 14,
test_cache());
+ assert_eq!(
+ reader.read(4..8).await.unwrap(),
+ Bytes::from_static(b"efgh")
+ );
+ assert_eq!(
+ reader.read(1..14).await.unwrap(),
+ Bytes::from_static(b"bcdefghijklmn")
+ );
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 3);
+ reader.read(0..14).await.unwrap();
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 3);
+ }
+
+ #[tokio::test]
+ async fn test_coalesced_misses_bound_batch_size() {
+ let delegate = Arc::new(CountingReader {
+ data: Bytes::from(vec![42; 160]),
+ reads: AtomicUsize::new(0),
+ });
+ let reader = CachedFileReader::new(delegate.clone(), "snapshot-1",
160, test_cache());
+ assert_eq!(reader.read(0..160).await.unwrap(), delegate.data);
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 2);
+ }
+
+ #[tokio::test]
+ async fn test_coalesced_overlapping_reads_share_missing_blocks() {
+ let delegate = Arc::new(SlowCountingReader {
+ data: Bytes::from_static(b"abcdefghijkl"),
+ reads: AtomicUsize::new(0),
+ });
+ let reader = CachedFileReader::new(delegate.clone(), "snapshot-1", 12,
test_cache());
+ let (a, b) = tokio::time::timeout(std::time::Duration::from_secs(2),
async {
+ tokio::join!(reader.read(0..12), reader.read(4..12))
+ })
+ .await
+ .unwrap();
+ assert_eq!(a.unwrap(), Bytes::from_static(b"abcdefghijkl"));
+ assert_eq!(b.unwrap(), Bytes::from_static(b"efghijkl"));
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
+ }
+
+ #[tokio::test]
+ async fn test_coalesced_truncated_read_releases_locks() {
+ let reader = CachedFileReader::new(Arc::new(ShortReader),
"snapshot-1", 12, test_cache());
+ for _ in 0..2 {
+ assert!(
+ tokio::time::timeout(std::time::Duration::from_secs(2),
reader.read(0..12))
+ .await
+ .unwrap()
+ .is_err()
+ );
+ }
+ }
+
+ #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+ async fn coalesced_shared_stripe_collisions_do_not_wait_on_self() {
+ let directory = tempfile::tempdir().unwrap();
+ let block_size = 256 * 1024;
+ let data = Bytes::from(vec![42; 32 * block_size]);
+ let cache = Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ shared: true,
+ dir: Some(directory.path().to_path_buf()),
+ namespace: "test".into(),
+ max_size: Some(16 * 1024 * 1024),
+ block_size: block_size as u64,
+ whitelist: std::collections::HashSet::from([FileType::Meta]),
+ })
+ .unwrap(),
+ );
+ for (a, b) in [(15, 27), (20, 28), (16, 31)] {
+ let a = cache.block_key("snapshot-1", a).cache_relative_path();
+ let b = cache.block_key("snapshot-1", b).cache_relative_path();
+ assert_eq!(a.components().next(), b.components().next());
+ }
+ let delegate = Arc::new(CountingReader {
+ data: data.clone(),
+ reads: AtomicUsize::new(0),
+ });
+ let reader = CachedFileReader::new(
+ delegate.clone(),
+ "snapshot-1",
+ data.len() as u64,
+ cache.clone(),
+ );
+ let result = tokio::time::timeout(
+ std::time::Duration::from_secs(2),
+ reader.read(0..data.len() as u64),
+ )
+ .await
+ .expect("colliding stripes must not wait for their own lock timeout")
+ .unwrap();
+ assert_eq!(result, data);
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
+ let _fills = tokio::time::timeout(
+ std::time::Duration::from_secs(2),
+ cache.fill_slots.clone().acquire_many_owned(8),
+ )
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(reader.read(0..data.len() as u64).await.unwrap(), data);
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
+ }
+
+ #[tokio::test]
+ async fn test_coalesced_miss_does_not_republish_after_invalidation() {
+ let cache = test_cache();
+ let started = Arc::new(tokio::sync::Notify::new());
+ let release = Arc::new(tokio::sync::Notify::new());
+ let old = CachedFileReader::new(
+ Arc::new(BlockingReader {
+ data: Bytes::from_static(b"old!old!old!"),
+ started: started.clone(),
+ release: release.clone(),
+ }),
+ "snapshot-1",
+ 12,
+ cache.clone(),
+ );
+ let loading = tokio::spawn(async move { old.read(0..12).await });
+ started.notified().await;
+ cache.invalidate_path("snapshot-1").await.unwrap();
+ release.notify_one();
+ assert_eq!(
+ loading.await.unwrap().unwrap(),
+ Bytes::from_static(b"old!old!old!")
+ );
+ let delegate = Arc::new(CountingReader {
+ data: Bytes::from_static(b"new!new!new!"),
+ reads: AtomicUsize::new(0),
+ });
+ let current = CachedFileReader::new(delegate.clone(), "snapshot-1",
12, cache);
+ assert_eq!(
+ current.read(0..12).await.unwrap(),
+ Bytes::from_static(b"new!new!new!")
+ );
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
+ }
+
#[async_trait::async_trait]
impl FileRead for CountingReader {
async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
@@ -307,12 +582,12 @@ mod tests {
reader.read(2..10).await.unwrap(),
Bytes::from_static(b"cdefghij")
);
- assert_eq!(delegate.reads.load(Ordering::SeqCst), 3);
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
assert_eq!(
reader.read(2..10).await.unwrap(),
Bytes::from_static(b"cdefghij")
);
- assert_eq!(delegate.reads.load(Ordering::SeqCst), 3);
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
}
#[tokio::test]
@@ -356,6 +631,69 @@ mod tests {
assert_eq!(delegate.reads.load(Ordering::SeqCst), 5);
}
+ #[derive(Debug)]
+ struct ConcurrentReader {
+ barrier: tokio::sync::Barrier,
+ reads: AtomicUsize,
+ }
+
+ #[async_trait::async_trait]
+ impl FileRead for ConcurrentReader {
+ async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
+ self.reads.fetch_add(1, Ordering::SeqCst);
+ self.barrier.wait().await;
+ Ok(Bytes::from(vec![42; (range.end - range.start) as usize]))
+ }
+ }
+
+ #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+ async fn coalesced_disjoint_shared_reads_remain_concurrent() {
+ let directory = tempfile::tempdir().unwrap();
+ let size = 8 * 1024 * 1024;
+ let cache = Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ shared: true,
+ dir: Some(directory.path().to_path_buf()),
+ namespace: "test".into(),
+ max_size: Some(128 * 1024 * 1024),
+ block_size: 256 * 1024,
+ whitelist: std::collections::HashSet::from([FileType::Meta]),
+ })
+ .unwrap(),
+ );
+ let source = Arc::new(ConcurrentReader {
+ barrier: tokio::sync::Barrier::new(8),
+ reads: AtomicUsize::new(0),
+ });
+ let reader = Arc::new(CachedFileReader::new(
+ source.clone(),
+ "snapshot-1",
+ 8 * size,
+ cache.clone(),
+ ));
+ let reads = (0..8).map(|i| {
+ let reader = reader.clone();
+ async move {
+ let data = reader.read(i * size..(i + 1) *
size).await.unwrap();
+ assert_eq!(data.len(), size as usize);
+ assert!(data.iter().all(|byte| *byte == 42));
+ }
+ });
+ tokio::time::timeout(
+ std::time::Duration::from_secs(2),
+ futures::future::join_all(reads),
+ )
+ .await
+ .expect("all eight disjoint source reads must overlap");
+ assert_eq!(source.reads.load(Ordering::SeqCst), 8);
+ let _fills = cache
+ .fill_slots
+ .clone()
+ .acquire_many_owned(8)
+ .await
+ .unwrap();
+ }
+
#[derive(Debug)]
struct SlowCountingReader {
data: Bytes,
@@ -427,6 +765,62 @@ mod tests {
drop(permits);
}
+ #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+ async fn coalesced_shared_fill_preserves_single_flight() {
+ let directory = tempfile::tempdir().unwrap();
+ let delegate = Arc::new(CountingReader {
+ data: Bytes::from_static(b"abcdefgh"),
+ reads: AtomicUsize::new(0),
+ });
+ let cache = Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ shared: true,
+ dir: Some(directory.path().to_path_buf()),
+ namespace: "test".into(),
+ max_size: Some(4096),
+ block_size: 4,
+ whitelist: std::collections::HashSet::from([FileType::Meta]),
+ })
+ .unwrap(),
+ );
+ let reader = CachedFileReader::new(delegate.clone(), "data.parquet",
8, cache.clone());
+ 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_eq!(
+ tokio::time::timeout(std::time::Duration::from_millis(500),
reader.read(0..8))
+ .await
+ .unwrap()
+ .unwrap(),
+ "abcdefgh"
+ );
+ assert_eq!(cache.fill_slots.available_permits(), 6);
+ assert!(
+ tokio::time::timeout(std::time::Duration::from_millis(50),
reader.read(0..8))
+ .await
+ .is_err()
+ );
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
+ drop(gate);
+ assert_eq!(reader.read(0..8).await.unwrap(), "abcdefgh");
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
+ let permits = cache
+ .fill_slots
+ .clone()
+ .acquire_many_owned(8)
+ .await
+ .unwrap();
+ assert!(cache.background_fill_slot().is_none());
+ drop(permits);
+ }
+
#[test]
fn shared_fill_does_not_block_reads_on_another_runtime() {
let directory = tempfile::tempdir().unwrap();
diff --git a/crates/paimon/src/io/cache/shared.rs
b/crates/paimon/src/io/cache/shared.rs
index f5b3dcdc..56db9c2a 100644
--- a/crates/paimon/src/io/cache/shared.rs
+++ b/crates/paimon/src/io/cache/shared.rs
@@ -305,12 +305,38 @@ impl SharedDiskCache {
pub(crate) async fn load_lock(&self, key: &BlockKey) -> Option<File> {
// Best-effort block coalescing; read_full locks only the first block.
// Publication safety relies on the directory lock and generation
checks.
+ lock(self.load_lock_path(key)?).await
+ }
+
+ pub(crate) async fn load_locks(&self, keys: &[BlockKey]) -> Vec<File> {
+ let mut paths: Vec<_> = keys
+ .iter()
+ .filter_map(|key| self.load_lock_path(key))
+ .collect();
+ paths.sort_unstable();
+ paths.dedup();
+ // Busy stripes may belong to unrelated blocks; never wait while
holding a batch.
+ tokio::task::spawn_blocking(move || {
+ let mut guards = Vec::new();
+ for path in paths {
+ if let Ok(file) = open_lock(&path) {
+ if file.try_lock().is_ok() {
+ guards.push(file);
+ }
+ }
+ }
+ guards
+ })
+ .await
+ .unwrap_or_default()
+ }
+
+ fn load_lock_path(&self, key: &BlockKey) -> Option<PathBuf> {
let stripe = key.cache_relative_path();
- lock(self.root.join(format!(
+ Some(self.root.join(format!(
"load-{}.lock",
stripe.components().next()?.as_os_str().to_string_lossy()
)))
- .await
}
pub(crate) async fn get(&self, key: &BlockKey, generation:
&Option<String>) -> Option<Bytes> {
@@ -1093,6 +1119,28 @@ mod tests {
drop(other);
}
+ #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+ async fn batch_download_locks_deduplicate_and_skip_busy_stripes() {
+ let directory = tempfile::tempdir().unwrap();
+ let first = SharedDiskCache::new(directory.path().to_path_buf(), 1024
* 1024).unwrap();
+ let second = SharedDiskCache::new(directory.path().to_path_buf(), 1024
* 1024).unwrap();
+ let a = BlockKey::with_namespace("test", "snapshot-1", 256 * 1024, 15);
+ let b = BlockKey::with_namespace("test", "snapshot-1", 256 * 1024, 20);
+ let collision = BlockKey::with_namespace("test", "snapshot-1", 256 *
1024, 27);
+ assert_eq!(first.load_lock_path(&a), first.load_lock_path(&collision));
+ assert_ne!(first.load_lock_path(&a), first.load_lock_path(&b));
+ let forward = [a.clone(), b.clone(), collision.clone()];
+ let reverse = [collision, b, a];
+ let guards = first.load_locks(&forward).await;
+ assert_eq!(guards.len(), 2);
+ let contended = tokio::time::timeout(Duration::from_secs(1),
second.load_locks(&reverse))
+ .await
+ .expect("busy stripes must not block another batch");
+ assert!(contended.is_empty());
+ drop(guards);
+ assert_eq!(second.load_locks(&reverse).await.len(), 2);
+ }
+
#[tokio::test]
async fn interrupted_temporary_blocks_are_removed_before_clean_startup() {
for invalidate in [false, true] {
