From 6b5cce16382ae728199582cee9565ef928fca6cf Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Tue, 15 Sep 2026 13:32:01 +0800 Subject: [PATCH] fix(io): flush disk cache blocks before publishing Wait for background writes to finish before renaming cache blocks, and clean up temporary files on write or flush errors. Verify that published blocks are immediately readable before testing cache restart recovery. Signed-off-by: QuakeWang --- 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 771282ee3..72cb84f9e 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);