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 525f78ff perf: skip source stat for complete shared-cache small files
(#1228)
525f78ff is described below
commit 525f78ffa096551ff090a893715de5a6ec1b0459
Author: XiaoHongbo <[email protected]>
AuthorDate: Tue Oct 27 16:09:13 2026 +0800
perf: skip source stat for complete shared-cache small files (#1228)
---
crates/paimon/src/io/cache/mod.rs | 17 ++++
crates/paimon/src/io/cache/small_file_tests.rs | 112 +++++++++++++++++++++++++
crates/paimon/src/io/file_io.rs | 3 +
3 files changed, 132 insertions(+)
diff --git a/crates/paimon/src/io/cache/mod.rs
b/crates/paimon/src/io/cache/mod.rs
index 57f15967..4fa39953 100644
--- a/crates/paimon/src/io/cache/mod.rs
+++ b/crates/paimon/src/io/cache/mod.rs
@@ -20,6 +20,8 @@ mod file_type;
mod memory;
mod priority_lru;
mod reader;
+#[cfg(test)]
+mod small_file_tests;
mod state;
use self::file_type::FileType;
@@ -229,6 +231,21 @@ impl LocalCache {
token
}
+ pub(super) async fn read_cached_small_file(
+ &self,
+ path: &str,
+ token: &CacheReadToken,
+ ) -> Option<bytes::Bytes> {
+ if !matches!(self.backend, CacheBackend::SharedDisk(_)) {
+ return None;
+ }
+ let block_size = usize::try_from(self.block_size).ok()?;
+ let key = self.block_key(path, 0);
+ let payload = self.get_block(&key, block_size, token).await?;
+ // Native fills publish complete blocks; a short first block is the
whole file.
+ (payload.len() < block_size).then_some(payload)
+ }
+
async fn shared_load_lock(&self, key: &BlockKey) -> Option<std::fs::File> {
match &self.backend {
CacheBackend::SharedDisk(disk) => disk.load_lock(key).await,
diff --git a/crates/paimon/src/io/cache/small_file_tests.rs
b/crates/paimon/src/io/cache/small_file_tests.rs
new file mode 100644
index 00000000..45efcd5d
--- /dev/null
+++ b/crates/paimon/src/io/cache/small_file_tests.rs
@@ -0,0 +1,112 @@
+// 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 super::{LocalCache, LocalCacheConfig};
+use crate::common::{CatalogOptions, Options};
+use crate::io::{FileIO, FileIOBuilder};
+use bytes::Bytes;
+use std::path::Path;
+use std::sync::Arc;
+
+fn shared_io(root: &Path) -> FileIO {
+ let mut options = Options::new();
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ options.set("local-cache.shared", "true");
+ options.set("local-cache.max-size", "16 mb");
+ options.set(
+ CatalogOptions::LOCAL_CACHE_DIR,
+ root.join("cache").to_string_lossy(),
+ );
+ options.set(CatalogOptions::LOCAL_CACHE_BLOCK_SIZE, "4");
+ let cache =
+
LocalCache::new(LocalCacheConfig::from_options(&options).unwrap().unwrap()).unwrap();
+ FileIOBuilder::new("file")
+ .with_local_cache(Arc::new(cache))
+ .build()
+ .unwrap()
+}
+
+fn file_path(path: &Path) -> String {
+ let normalized = path.to_string_lossy().replace('\\', "/");
+ if normalized.starts_with('/') {
+ format!("file:{normalized}")
+ } else {
+ format!("file:/{normalized}")
+ }
+}
+
+#[tokio::test]
+async fn short_first_block_is_a_complete_file() {
+ let root = tempfile::tempdir().unwrap();
+ let io = shared_io(root.path());
+ let path = root.path().join("snapshot-1");
+ std::fs::write(&path, b"abc").unwrap();
+ let input = io.new_input(&file_path(&path)).unwrap();
+ assert_eq!(input.read().await.unwrap().as_ref(), b"abc");
+ std::fs::rename(&path, root.path().join("offline")).unwrap();
+ // The cached complete file must not require a source stat.
+ assert_eq!(input.read().await.unwrap().as_ref(), b"abc");
+}
+
+#[tokio::test]
+async fn full_first_block_does_not_imply_eof() {
+ let root = tempfile::tempdir().unwrap();
+ let io = shared_io(root.path());
+ for data in [&b"abcd"[..], &b"abcdef"[..]] {
+ let path = root.path().join(format!("snapshot-{}", data.len()));
+ std::fs::write(&path, data).unwrap();
+ let input = io.new_input(&file_path(&path)).unwrap();
+ assert_eq!(input.read().await.unwrap().as_ref(), data);
+ std::fs::remove_file(&path).unwrap();
+ assert!(input.read().await.is_err());
+ }
+}
+
+#[tokio::test]
+async fn replacement_child() {
+ let Some(root) = std::env::var_os("PAIMON_SMALL_FILE_TEST_ROOT") else {
+ return;
+ };
+ let root = Path::new(&root);
+ shared_io(root)
+ .new_output(&file_path(&root.join("snapshot-1")))
+ .unwrap()
+ .write(Bytes::from_static(b"replacement"))
+ .await
+ .unwrap();
+}
+
+#[tokio::test]
+async fn another_process_invalidates_the_small_file() {
+ let root = tempfile::tempdir().unwrap();
+ let io = shared_io(root.path());
+ let path = root.path().join("snapshot-1");
+ std::fs::write(&path, b"abc").unwrap();
+ let input = io.new_input(&file_path(&path)).unwrap();
+ assert_eq!(input.read().await.unwrap().as_ref(), b"abc");
+ assert!(std::process::Command::new(std::env::current_exe().unwrap())
+ .args([
+ "--exact",
+ "io::cache::small_file_tests::replacement_child",
+ "--nocapture"
+ ])
+ .env("PAIMON_SMALL_FILE_TEST_ROOT", root.path())
+ .status()
+ .unwrap()
+ .success());
+ assert_eq!(input.read().await.unwrap().as_ref(), b"replacement");
+}
diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index 301294fa..14e481dd 100644
--- a/crates/paimon/src/io/file_io.rs
+++ b/crates/paimon/src/io/file_io.rs
@@ -1562,6 +1562,9 @@ impl InputFile {
return Ok(op.read(&relative_path).await?.to_bytes());
};
let read_token = cache.read_token(&cache_path);
+ if let Some(bytes) = cache.read_cached_small_file(&cache_path,
&read_token).await {
+ return Ok(bytes);
+ }
let size = if let Some(size) = cache.file_size(&cache_path,
&read_token).await {
size
} else {
