(paimon-rust) branch main updated: fix(io): flush disk cache blocks before publishing (#836)

[lzljs3620320]

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 4fafa12c fix(io): flush disk cache blocks before publishing (#836)
4fafa12c is described below

commit 4fafa12c2776e5cc0fa24bb3b28c3cb422e60bd3
Author: QuakeWang <45645138+quakew...@users.noreply.github.com>
AuthorDate: Thu Sep 17 16:08:14 2026 +0800

    fix(io): flush disk cache blocks before publishing (#836)
---
 crates/paimon/src/io/cache/disk.rs | 13 ++++++++++++-
 1 file changed, 12 insertions(+), 1 deletion(-)

diff --git a/crates/paimon/src/io/cache/disk.rs 
b/crates/paimon/src/io/cache/disk.rs
index 771282ee..72cb84f9 100644
--- a/crates/paimon/src/io/cache/disk.rs
+++ b/crates/paimon/src/io/cache/disk.rs
@@ -243,7 +243,13 @@ impl DiskCache {
                 return;
             }
         };
-        if let Err(error) = temporary_file.write_all(&encoded).await {
+        let write_result = async {
+            temporary_file.write_all(&encoded).await?;
+            // Tokio may return from write_all before the blocking write 
completes.
+            temporary_file.flush().await
+        }
+        .await;
+        if let Err(error) = write_result {
             log::debug!(
                 "Failed to write local cache temporary block '{}': {error}",
                 temporary.display()
@@ -846,6 +852,11 @@ mod tests {
 
         let cache = DiskCache::new(directory.path(), None).unwrap();
         cache.put_block(&key, payload.clone()).await;
+        // Check publication without scheduling another Tokio filesystem 
operation.
+        assert_eq!(
+            
std::fs::read(directory.path().join(key.cache_relative_path())).unwrap(),
+            encode_block(&key, &payload)
+        );
         assert_eq!(cache.get_block(&key).await, Some(payload.clone()));
         drop(cache);