(hudi-rs) branch main updated: fix(hfile): count checksum chunks over header + data (#836)

[yihua]

This is an automated email from the ASF dual-hosted git repository.

yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git


The following commit(s) were added to refs/heads/main by this push:
     new 90b2be3c fix(hfile): count checksum chunks over header + data (#836)
90b2be3c is described below

commit 90b2be3c6ee82151624c2693844b64387cbf1b24
Author: Lin Liu <141371752+linliu-c...@users.noreply.github.com>
AuthorDate: Wed Oct 28 14:11:13 2026 -0700

    fix(hfile): count checksum chunks over header + data (#836)
    
    Co-authored-by: Y Ethan Guo <ethan.guoyi...@gmail.com>
---
 crates/core/src/hfile/block.rs | 55 +++++++++++++++++++++++++++++++++---------
 1 file changed, 43 insertions(+), 12 deletions(-)

diff --git a/crates/core/src/hfile/block.rs b/crates/core/src/hfile/block.rs
index 113b5b93..67849305 100644
--- a/crates/core/src/hfile/block.rs
+++ b/crates/core/src/hfile/block.rs
@@ -42,9 +42,6 @@ const HEADER_SIZE_NO_CHECKSUM: usize = MAGIC_LENGTH + 2 * 
SIZEOF_INT32 + SIZEOF_
 /// - 4 bytes: on-disk data size with header
 pub const BLOCK_HEADER_SIZE: usize = HEADER_SIZE_NO_CHECKSUM + SIZEOF_BYTE + 2 
* SIZEOF_INT32;
 
-/// Checksum size (each checksum is 4 bytes)
-const CHECKSUM_SIZE: usize = SIZEOF_INT32;
-
 /// Parsed block header information.
 #[derive(Debug, Clone)]
 #[allow(dead_code)]
@@ -100,13 +97,6 @@ impl BlockHeader {
         })
     }
 
-    /// Calculate the number of checksum bytes.
-    pub fn checksum_bytes_count(&self) -> usize {
-        let on_disk_with_header = BLOCK_HEADER_SIZE + 
self.on_disk_size_without_header;
-        let num_chunks = on_disk_with_header.div_ceil(self.bytes_per_checksum);
-        num_chunks * CHECKSUM_SIZE
-    }
-
     /// Returns the total on-disk size including header.
     pub fn on_disk_size_with_header(&self) -> usize {
         BLOCK_HEADER_SIZE + self.on_disk_size_without_header
@@ -128,8 +118,7 @@ impl HFileBlock {
 
         // Extract the compressed data (between header and checksum)
         let data_start = BLOCK_HEADER_SIZE;
-        let data_end =
-            data_start + header.on_disk_size_without_header - 
header.checksum_bytes_count();
+        let data_end = header.on_disk_data_size_with_header;
 
         // For uncompressed blocks, on_disk_size == uncompressed_size
         // For compressed blocks, we need to decompress
@@ -338,6 +327,48 @@ mod tests {
         assert_eq!(header.block_type, HFileBlockType::Data);
     }
 
+    #[test]
+    fn test_parse_gz_block_at_checksum_chunk_boundary() {
+        use flate2::{Compression, write::GzEncoder};
+        use std::io::Write;
+
+        let bytes_per_checksum = 16384usize;
+        let checksum_size = SIZEOF_INT32;
+        let (input, compressed) = (16000..16384)
+            .find_map(|len| {
+                let input = vec![b'x'; len];
+                let mut encoder = GzEncoder::new(Vec::new(), 
Compression::none());
+                encoder.write_all(&input).unwrap();
+                let compressed = encoder.finish().unwrap();
+                let data_size = BLOCK_HEADER_SIZE + compressed.len();
+                (data_size == bytes_per_checksum - 1).then_some((input, 
compressed))
+            })
+            .expect("a stored GZIP block one byte below the checksum 
boundary");
+        let data_size = BLOCK_HEADER_SIZE + compressed.len();
+        let num_chunks = data_size.div_ceil(bytes_per_checksum);
+        let checksum_bytes = num_chunks * checksum_size;
+        assert!(num_chunks < (data_size + 
checksum_bytes).div_ceil(bytes_per_checksum));
+        eprintln!(
+            "synthetic GZIP boundary: N={data_size}, C={}, 
chunks={num_chunks}, checksum_bytes={checksum_bytes}",
+            compressed.len()
+        );
+
+        let mut bytes = Vec::new();
+        bytes.extend_from_slice(b"DATABLK*");
+        bytes.extend_from_slice(&((compressed.len() + checksum_bytes) as 
i32).to_be_bytes());
+        bytes.extend_from_slice(&(input.len() as i32).to_be_bytes());
+        bytes.extend_from_slice(&(-1i64).to_be_bytes());
+        bytes.push(1);
+        bytes.extend_from_slice(&(bytes_per_checksum as i32).to_be_bytes());
+        bytes.extend_from_slice(&(data_size as i32).to_be_bytes());
+        assert_eq!(bytes.len(), BLOCK_HEADER_SIZE);
+        bytes.extend_from_slice(&compressed);
+        bytes.resize(bytes.len() + checksum_bytes, 0);
+
+        let block = HFileBlock::parse(&bytes, CompressionCodec::Gzip).unwrap();
+        assert_eq!(block.data, input);
+    }
+
     #[test]
     fn test_block_index_entry_ordering() {
         let key1 = Key::from_bytes(vec![0, 3, b'a', b'a', b'a']);