From 2fcc2f03902960c089bda3b9b0e0f9ed6d697555 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Thu, 9 Jul 2026 17:50:05 -0400 Subject: [PATCH 1/5] feat(delete-vector): decode deletion-vector-v1 puffin blobs Add DeleteVector::deserialize to parse a deletion-vector-v1 blob ([length][magic][roaring][crc32], portable 64-bit roaring) into a DeleteVector. Verifies the length prefix, CRC-32, and magic before decoding, and leaves cardinality validation to the caller that holds the delete file's record_count. Tests cover round-trips (including a run-optimized bitmap that exercises the RUN-container decode path) and the length, magic, and CRC error paths. Refs #2792. --- Cargo.lock | 1 + Cargo.toml | 1 + crates/iceberg/Cargo.toml | 1 + crates/iceberg/src/delete_vector.rs | 192 ++++++++++++++++++++++++++++ 4 files changed, 195 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index 8b4c87422a..7857814a28 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3773,6 +3773,7 @@ dependencies = [ "bimap", "bytes", "chrono", + "crc32fast", "derive_builder", "expect-test", "fastnum", diff --git a/Cargo.toml b/Cargo.toml index 141f1f2c3a..dc0dea563f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -76,6 +76,7 @@ bytes = "1.11" cfg-if = "1" chrono = "0.4.41" clap = { version = "4.5.48", features = ["derive", "cargo"] } +crc32fast = "1" dashmap = "6" datafusion = "54.0.0" datafusion-cli = "54.0.0" diff --git a/crates/iceberg/Cargo.toml b/crates/iceberg/Cargo.toml index c33445c64c..733779d741 100644 --- a/crates/iceberg/Cargo.toml +++ b/crates/iceberg/Cargo.toml @@ -52,6 +52,7 @@ base64 = { workspace = true } bimap = { workspace = true } bytes = { workspace = true } chrono = { workspace = true } +crc32fast = { workspace = true } derive_builder = { workspace = true } expect-test = { workspace = true } fastnum = { workspace = true } diff --git a/crates/iceberg/src/delete_vector.rs b/crates/iceberg/src/delete_vector.rs index df8a10193c..14a950b60e 100644 --- a/crates/iceberg/src/delete_vector.rs +++ b/crates/iceberg/src/delete_vector.rs @@ -23,6 +23,14 @@ use roaring::treemap::BitmapIter; use crate::{Error, ErrorKind, Result}; +/// Magic bytes prefixing a serialized `deletion-vector-v1` bitmap, per the Iceberg Puffin spec. +/// Iceberg-Java stores these as the little-endian int 1681511377 (0x6439D3D1). +const DV_MAGIC: [u8; 4] = [0xD1, 0xD3, 0x39, 0x64]; +const DV_LENGTH_PREFIX_BYTES: usize = 4; +const DV_MAGIC_BYTES: usize = 4; +const DV_CRC_BYTES: usize = 4; +const DV_MIN_BLOB_BYTES: usize = DV_LENGTH_PREFIX_BYTES + DV_MAGIC_BYTES + DV_CRC_BYTES; + #[derive(Debug, Default)] pub struct DeleteVector { inner: RoaringTreemap, @@ -68,6 +76,87 @@ impl DeleteVector { pub fn len(&self) -> u64 { self.inner.len() } + + /// Parses a `deletion-vector-v1` Puffin blob into a `DeleteVector`. + /// + /// The layout, defined by the Iceberg Puffin spec and matching Iceberg-Java's + /// `BitmapPositionDeleteIndex`, is: + /// + /// ```text + /// [length: u32 big-endian][magic: D1 D3 39 64][vector][crc: u32 big-endian] + /// ``` + /// + /// `length` counts the magic and vector bytes (not itself or the CRC). The CRC-32 is + /// computed over the magic and vector. `vector` is a roaring bitmap in the portable + /// 64-bit format read by [`RoaringTreemap::deserialize_from`]. + /// + /// Cardinality is not checked here. The caller validates the decoded length against the + /// delete file's `record_count`, where the manifest metadata is available. + /// + /// # Errors + /// + /// Returns [`ErrorKind::DataInvalid`] if the blob is shorter than the minimum, the length + /// prefix or CRC does not match, the magic is wrong, or the roaring payload fails to decode. + // Consumed by the scan delete loader once the deletion-vector read path is wired up. + #[allow(dead_code)] + pub fn deserialize(blob: &[u8]) -> Result { + if blob.len() < DV_MIN_BLOB_BYTES { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion-vector-v1 blob is {} bytes, shorter than the {DV_MIN_BLOB_BYTES}-byte minimum", + blob.len() + ), + )); + } + + // The magic and vector, i.e. the bytes covered by both the length prefix and the CRC. + let body = &blob[DV_LENGTH_PREFIX_BYTES..blob.len() - DV_CRC_BYTES]; + + let declared_len = + u32::from_be_bytes(blob[..DV_LENGTH_PREFIX_BYTES].try_into().unwrap()) as usize; + if declared_len != body.len() { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion-vector-v1 length prefix is {declared_len}, expected {}", + body.len() + ), + )); + } + + // Verify the CRC before interpreting any bytes so a corrupt blob yields a single clear + // error rather than an opaque roaring decode failure. + let stored_crc = + u32::from_be_bytes(blob[blob.len() - DV_CRC_BYTES..].try_into().unwrap()); + let computed_crc = crc32fast::hash(body); + if computed_crc != stored_crc { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion-vector-v1 CRC mismatch: computed {computed_crc:#010x}, stored {stored_crc:#010x}" + ), + )); + } + + let (magic, vector) = body.split_at(DV_MAGIC_BYTES); + if magic != DV_MAGIC { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("deletion-vector-v1 magic mismatch: {magic:02x?}, expected {DV_MAGIC:02x?}"), + )); + } + + let inner = RoaringTreemap::deserialize_from(vector).map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + "failed to decode deletion-vector-v1 roaring payload", + ) + .with_source(e) + })?; + + Ok(DeleteVector { inner }) + } } // Ideally, we'd just wrap `roaring::RoaringTreemap`'s iterator, `roaring::treemap::Iter` here. @@ -198,4 +287,107 @@ mod tests { let res = dv.insert_positions(&positions); assert!(res.is_err()); } + + // Reproduces Iceberg-Java's `deletion-vector-v1` framing so tests can round-trip through + // `deserialize` without a Java writer. Cross-implementation golden fixtures produced by + // Iceberg-Java are tracked separately; this only checks that our decode matches our encode. + fn encode_dv_blob(dv: &DeleteVector) -> Vec { + let mut vector = Vec::with_capacity(dv.inner.serialized_size()); + dv.inner.serialize_into(&mut vector).unwrap(); + + let body_len = DV_MAGIC_BYTES + vector.len(); + let mut blob = Vec::with_capacity(DV_LENGTH_PREFIX_BYTES + body_len + DV_CRC_BYTES); + blob.extend_from_slice(&(body_len as u32).to_be_bytes()); + blob.extend_from_slice(&DV_MAGIC); + blob.extend_from_slice(&vector); + let crc = crc32fast::hash(&blob[DV_LENGTH_PREFIX_BYTES..]); + blob.extend_from_slice(&crc.to_be_bytes()); + blob + } + + fn dv_of(positions: impl IntoIterator) -> DeleteVector { + let mut dv = DeleteVector::default(); + for pos in positions { + dv.insert(pos); + } + dv + } + + fn sorted(dv: &DeleteVector) -> Vec { + let mut positions: Vec = dv.iter().collect(); + positions.sort_unstable(); + positions + } + + #[test] + fn deserialize_roundtrip_empty() { + let blob = encode_dv_blob(&DeleteVector::default()); + assert_eq!(DeleteVector::deserialize(&blob).unwrap().len(), 0); + } + + #[test] + fn deserialize_roundtrip_small() { + let positions = [0u64, 5, 100, 1000]; + let dv = DeleteVector::deserialize(&encode_dv_blob(&dv_of(positions))).unwrap(); + assert_eq!(sorted(&dv), positions); + } + + #[test] + fn deserialize_roundtrip_spanning_64bit_keys() { + let positions = [1u64, 1 << 33, (1 << 33) + 5, 1 << 34]; + let dv = DeleteVector::deserialize(&encode_dv_blob(&dv_of(positions))).unwrap(); + assert_eq!(sorted(&dv), positions); + } + + // Java run-optimizes every deletion vector before writing, so real blobs carry RUN + // containers, which use the SERIAL_COOKIE roaring layout. Force that layout so decode + // exercises the run-container path rather than only array and bitmap containers. + #[test] + fn deserialize_roundtrip_run_optimized() { + let mut dv = dv_of(0..10_000); + assert!( + dv.inner.optimize(), + "expected a dense range to run-length encode" + ); + let decoded = DeleteVector::deserialize(&encode_dv_blob(&dv)).unwrap(); + assert_eq!(decoded.len(), 10_000); + assert_eq!(sorted(&decoded).first(), Some(&0)); + assert_eq!(sorted(&decoded).last(), Some(&9_999)); + } + + #[test] + fn deserialize_rejects_short_blob() { + let err = DeleteVector::deserialize(&[0u8; DV_MIN_BLOB_BYTES - 1]).unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + } + + #[test] + fn deserialize_rejects_bad_magic() { + let mut blob = encode_dv_blob(&dv_of([1])); + blob[DV_LENGTH_PREFIX_BYTES] ^= 0xFF; + // Recompute the CRC so the magic check, not the CRC check, is what fails. + let end = blob.len() - DV_CRC_BYTES; + let crc = crc32fast::hash(&blob[DV_LENGTH_PREFIX_BYTES..end]); + blob[end..].copy_from_slice(&crc.to_be_bytes()); + let err = DeleteVector::deserialize(&blob).unwrap_err(); + assert!(err.message().contains("magic mismatch"), "got: {err}"); + } + + #[test] + fn deserialize_rejects_bad_crc() { + let mut blob = encode_dv_blob(&dv_of([1, 2, 3])); + let end = blob.len() - DV_CRC_BYTES; + blob[end] ^= 0xFF; + let err = DeleteVector::deserialize(&blob).unwrap_err(); + assert!(err.message().contains("CRC mismatch"), "got: {err}"); + } + + #[test] + fn deserialize_rejects_length_prefix_mismatch() { + let mut blob = encode_dv_blob(&dv_of([1])); + let declared = u32::from_be_bytes(blob[..DV_LENGTH_PREFIX_BYTES].try_into().unwrap()); + blob[..DV_LENGTH_PREFIX_BYTES].copy_from_slice(&(declared + 1).to_be_bytes()); + let err = DeleteVector::deserialize(&blob).unwrap_err(); + assert!(err.message().contains("length prefix"), "got: {err}"); + } } From ed8020614d9169d858c505d9e0fd1437ed0deab4 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 21 Jul 2026 13:44:06 -0400 Subject: [PATCH 2/5] cargo fmt --- crates/iceberg/src/delete_vector.rs | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/crates/iceberg/src/delete_vector.rs b/crates/iceberg/src/delete_vector.rs index 14a950b60e..4b78908e6d 100644 --- a/crates/iceberg/src/delete_vector.rs +++ b/crates/iceberg/src/delete_vector.rs @@ -127,8 +127,7 @@ impl DeleteVector { // Verify the CRC before interpreting any bytes so a corrupt blob yields a single clear // error rather than an opaque roaring decode failure. - let stored_crc = - u32::from_be_bytes(blob[blob.len() - DV_CRC_BYTES..].try_into().unwrap()); + let stored_crc = u32::from_be_bytes(blob[blob.len() - DV_CRC_BYTES..].try_into().unwrap()); let computed_crc = crc32fast::hash(body); if computed_crc != stored_crc { return Err(Error::new( @@ -143,7 +142,9 @@ impl DeleteVector { if magic != DV_MAGIC { return Err(Error::new( ErrorKind::DataInvalid, - format!("deletion-vector-v1 magic mismatch: {magic:02x?}, expected {DV_MAGIC:02x?}"), + format!( + "deletion-vector-v1 magic mismatch: {magic:02x?}, expected {DV_MAGIC:02x?}" + ), )); } From 23df07b298a089b69865c04c77070425e9b93bbd Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 21 Jul 2026 15:55:46 -0400 Subject: [PATCH 3/5] address PR feedback. renamed tests to be consistent, and sorted decoded bytes once. --- crates/iceberg/src/delete_vector.rs | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/crates/iceberg/src/delete_vector.rs b/crates/iceberg/src/delete_vector.rs index 4b78908e6d..650fd98899 100644 --- a/crates/iceberg/src/delete_vector.rs +++ b/crates/iceberg/src/delete_vector.rs @@ -321,20 +321,20 @@ mod tests { } #[test] - fn deserialize_roundtrip_empty() { + fn test_deserialize_roundtrip_empty() { let blob = encode_dv_blob(&DeleteVector::default()); assert_eq!(DeleteVector::deserialize(&blob).unwrap().len(), 0); } #[test] - fn deserialize_roundtrip_small() { + fn test_deserialize_roundtrip_small() { let positions = [0u64, 5, 100, 1000]; let dv = DeleteVector::deserialize(&encode_dv_blob(&dv_of(positions))).unwrap(); assert_eq!(sorted(&dv), positions); } #[test] - fn deserialize_roundtrip_spanning_64bit_keys() { + fn test_deserialize_roundtrip_spanning_64bit_keys() { let positions = [1u64, 1 << 33, (1 << 33) + 5, 1 << 34]; let dv = DeleteVector::deserialize(&encode_dv_blob(&dv_of(positions))).unwrap(); assert_eq!(sorted(&dv), positions); @@ -344,7 +344,7 @@ mod tests { // containers, which use the SERIAL_COOKIE roaring layout. Force that layout so decode // exercises the run-container path rather than only array and bitmap containers. #[test] - fn deserialize_roundtrip_run_optimized() { + fn test_deserialize_roundtrip_run_optimized() { let mut dv = dv_of(0..10_000); assert!( dv.inner.optimize(), @@ -352,18 +352,19 @@ mod tests { ); let decoded = DeleteVector::deserialize(&encode_dv_blob(&dv)).unwrap(); assert_eq!(decoded.len(), 10_000); - assert_eq!(sorted(&decoded).first(), Some(&0)); - assert_eq!(sorted(&decoded).last(), Some(&9_999)); + let positions = sorted(&decoded); + assert_eq!(positions.first(), Some(&0)); + assert_eq!(positions.last(), Some(&9_999)); } #[test] - fn deserialize_rejects_short_blob() { + fn test_deserialize_rejects_short_blob() { let err = DeleteVector::deserialize(&[0u8; DV_MIN_BLOB_BYTES - 1]).unwrap_err(); assert_eq!(err.kind(), ErrorKind::DataInvalid); } #[test] - fn deserialize_rejects_bad_magic() { + fn test_deserialize_rejects_bad_magic() { let mut blob = encode_dv_blob(&dv_of([1])); blob[DV_LENGTH_PREFIX_BYTES] ^= 0xFF; // Recompute the CRC so the magic check, not the CRC check, is what fails. @@ -375,7 +376,7 @@ mod tests { } #[test] - fn deserialize_rejects_bad_crc() { + fn test_deserialize_rejects_bad_crc() { let mut blob = encode_dv_blob(&dv_of([1, 2, 3])); let end = blob.len() - DV_CRC_BYTES; blob[end] ^= 0xFF; @@ -384,7 +385,7 @@ mod tests { } #[test] - fn deserialize_rejects_length_prefix_mismatch() { + fn test_deserialize_rejects_length_prefix_mismatch() { let mut blob = encode_dv_blob(&dv_of([1])); let declared = u32::from_be_bytes(blob[..DV_LENGTH_PREFIX_BYTES].try_into().unwrap()); blob[..DV_LENGTH_PREFIX_BYTES].copy_from_slice(&(declared + 1).to_be_bytes()); From 8c2e5202eb46e663f956d0d53e75f70c97e31656 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Mon, 17 Aug 2026 12:50:22 -0400 Subject: [PATCH 4/5] address PR feedback --- crates/iceberg/src/delete_vector.rs | 127 +++++++++++++++++++++++++--- 1 file changed, 114 insertions(+), 13 deletions(-) diff --git a/crates/iceberg/src/delete_vector.rs b/crates/iceberg/src/delete_vector.rs index 650fd98899..67312de6a1 100644 --- a/crates/iceberg/src/delete_vector.rs +++ b/crates/iceberg/src/delete_vector.rs @@ -17,9 +17,10 @@ use std::ops::BitOrAssign; -use roaring::RoaringTreemap; +use bytes::Buf; use roaring::bitmap::Iter; use roaring::treemap::BitmapIter; +use roaring::{RoaringBitmap, RoaringTreemap}; use crate::{Error, ErrorKind, Result}; @@ -87,8 +88,9 @@ impl DeleteVector { /// ``` /// /// `length` counts the magic and vector bytes (not itself or the CRC). The CRC-32 is - /// computed over the magic and vector. `vector` is a roaring bitmap in the portable - /// 64-bit format read by [`RoaringTreemap::deserialize_from`]. + /// computed over the magic and vector. `vector` is a roaring bitmap in the portable 64-bit + /// format: a directory of 32-bit key / 32-bit roaring bitmap pairs, ordered by unsigned + /// comparison of the keys, one bitmap per key. /// /// Cardinality is not checked here. The caller validates the decoded length against the /// delete file's `record_count`, where the manifest metadata is available. @@ -96,7 +98,8 @@ impl DeleteVector { /// # Errors /// /// Returns [`ErrorKind::DataInvalid`] if the blob is shorter than the minimum, the length - /// prefix or CRC does not match, the magic is wrong, or the roaring payload fails to decode. + /// prefix or CRC does not match, the magic is wrong, the roaring directory's keys are not + /// ordered by unsigned comparison, or the roaring payload fails to decode. // Consumed by the scan delete loader once the deletion-vector read path is wired up. #[allow(dead_code)] pub fn deserialize(blob: &[u8]) -> Result { @@ -113,8 +116,15 @@ impl DeleteVector { // The magic and vector, i.e. the bytes covered by both the length prefix and the CRC. let body = &blob[DV_LENGTH_PREFIX_BYTES..blob.len() - DV_CRC_BYTES]; - let declared_len = - u32::from_be_bytes(blob[..DV_LENGTH_PREFIX_BYTES].try_into().unwrap()) as usize; + let declared_len = (&blob[..DV_LENGTH_PREFIX_BYTES]) + .try_get_u32() + .map_err(|e| { + Error::new( + ErrorKind::Unexpected, + "failed to read the deletion-vector-v1 length prefix", + ) + .with_source(e) + })? as usize; if declared_len != body.len() { return Err(Error::new( ErrorKind::DataInvalid, @@ -127,7 +137,15 @@ impl DeleteVector { // Verify the CRC before interpreting any bytes so a corrupt blob yields a single clear // error rather than an opaque roaring decode failure. - let stored_crc = u32::from_be_bytes(blob[blob.len() - DV_CRC_BYTES..].try_into().unwrap()); + let stored_crc = (&blob[blob.len() - DV_CRC_BYTES..]) + .try_get_u32() + .map_err(|e| { + Error::new( + ErrorKind::Unexpected, + "failed to read the deletion-vector-v1 CRC", + ) + .with_source(e) + })?; let computed_crc = crc32fast::hash(body); if computed_crc != stored_crc { return Err(Error::new( @@ -148,7 +166,14 @@ impl DeleteVector { )); } - let inner = RoaringTreemap::deserialize_from(vector).map_err(|e| { + // The Puffin spec defines the roaring directory as the bitmaps "ordered by unsigned + // comparison of the 32-bit keys", with one bitmap per key. Walk it ourselves (rather than + // `RoaringTreemap::deserialize_from`, which stores keys in a `BTreeMap` via a plain insert + // and would silently accept a stream with duplicate or out-of-order keys, discarding the + // earlier bitmap on a duplicate) so a non-conformant blob is rejected instead of decoded + // into a value that doesn't match what was actually written. + let mut reader = vector; + let bitmap_count = reader.try_get_u64_le().map_err(|e| { Error::new( ErrorKind::DataInvalid, "failed to decode deletion-vector-v1 roaring payload", @@ -156,6 +181,42 @@ impl DeleteVector { .with_source(e) })?; + let mut bitmaps = Vec::new(); + let mut last_key: Option = None; + for _ in 0..bitmap_count { + let key = reader.try_get_u32_le().map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + "failed to decode deletion-vector-v1 roaring payload", + ) + .with_source(e) + })?; + if let Some(last) = last_key + && key <= last + { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion-vector-v1 roaring keys must be ordered by unsigned comparison, got key {key} after {last}" + ), + )); + } + last_key = Some(key); + + let bitmap = RoaringBitmap::deserialize_from(&mut reader).map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + "failed to decode deletion-vector-v1 roaring payload", + ) + .with_source(e) + })?; + bitmaps.push((key, bitmap)); + } + + // `bitmaps` is already sorted by key, but `roaring` has no constructor that accepts + // pre-sorted pairs without re-sorting them; revisit if that changes upstream. + let inner = RoaringTreemap::from_bitmaps(bitmaps); + Ok(DeleteVector { inner }) } } @@ -292,20 +353,23 @@ mod tests { // Reproduces Iceberg-Java's `deletion-vector-v1` framing so tests can round-trip through // `deserialize` without a Java writer. Cross-implementation golden fixtures produced by // Iceberg-Java are tracked separately; this only checks that our decode matches our encode. - fn encode_dv_blob(dv: &DeleteVector) -> Vec { - let mut vector = Vec::with_capacity(dv.inner.serialized_size()); - dv.inner.serialize_into(&mut vector).unwrap(); - + fn frame_dv_blob(vector: &[u8]) -> Vec { let body_len = DV_MAGIC_BYTES + vector.len(); let mut blob = Vec::with_capacity(DV_LENGTH_PREFIX_BYTES + body_len + DV_CRC_BYTES); blob.extend_from_slice(&(body_len as u32).to_be_bytes()); blob.extend_from_slice(&DV_MAGIC); - blob.extend_from_slice(&vector); + blob.extend_from_slice(vector); let crc = crc32fast::hash(&blob[DV_LENGTH_PREFIX_BYTES..]); blob.extend_from_slice(&crc.to_be_bytes()); blob } + fn encode_dv_blob(dv: &DeleteVector) -> Vec { + let mut vector = Vec::with_capacity(dv.inner.serialized_size()); + dv.inner.serialize_into(&mut vector).unwrap(); + frame_dv_blob(&vector) + } + fn dv_of(positions: impl IntoIterator) -> DeleteVector { let mut dv = DeleteVector::default(); for pos in positions { @@ -392,4 +456,41 @@ mod tests { let err = DeleteVector::deserialize(&blob).unwrap_err(); assert!(err.message().contains("length prefix"), "got: {err}"); } + + // Crafts a raw roaring treemap directory (bitmap count header + key/bitmap entries) so tests + // can exercise key-ordering violations that `dv_of`/`encode_dv_blob` can never produce, since + // `DeleteVector::insert` always keeps keys unique and ascending. + fn raw_roaring_vector(entries: &[(u32, &[u32])]) -> Vec { + let mut vector = Vec::new(); + vector.extend_from_slice(&(entries.len() as u64).to_le_bytes()); + for (key, positions) in entries { + let mut bitmap = RoaringBitmap::new(); + for &pos in *positions { + bitmap.insert(pos); + } + vector.extend_from_slice(&key.to_le_bytes()); + bitmap.serialize_into(&mut vector).unwrap(); + } + vector + } + + // `RoaringTreemap::deserialize_from` stores keys in a `BTreeMap` via a plain insert, so + // without our own ordering check, decoding this would silently keep only the second bitmap + // for key 5 (position 2) and drop the first (position 1). + #[test] + fn test_deserialize_rejects_duplicate_keys() { + let vector = raw_roaring_vector(&[(5, &[1]), (5, &[2])]); + let err = DeleteVector::deserialize(&frame_dv_blob(&vector)).unwrap_err(); + assert!(err.message().contains("unsigned comparison"), "got: {err}"); + } + + // The Puffin spec requires the roaring directory's keys to be "ordered by unsigned + // comparison"; a stream with unique but out-of-order keys is not a conformant blob even + // though `BTreeMap` would happily reorder it into a correct-looking result. + #[test] + fn test_deserialize_rejects_out_of_order_keys() { + let vector = raw_roaring_vector(&[(5, &[1]), (3, &[2])]); + let err = DeleteVector::deserialize(&frame_dv_blob(&vector)).unwrap_err(); + assert!(err.message().contains("unsigned comparison"), "got: {err}"); + } } From 83e037a7b8e8eef1841c217a48bcfac0360721a6 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 18 Aug 2026 11:41:21 -0400 Subject: [PATCH 5/5] address PR feedback --- crates/iceberg/src/delete_vector.rs | 207 ++++++++++++++++------------ 1 file changed, 120 insertions(+), 87 deletions(-) diff --git a/crates/iceberg/src/delete_vector.rs b/crates/iceberg/src/delete_vector.rs index 67312de6a1..cfcc29700c 100644 --- a/crates/iceberg/src/delete_vector.rs +++ b/crates/iceberg/src/delete_vector.rs @@ -98,8 +98,9 @@ impl DeleteVector { /// # Errors /// /// Returns [`ErrorKind::DataInvalid`] if the blob is shorter than the minimum, the length - /// prefix or CRC does not match, the magic is wrong, the roaring directory's keys are not - /// ordered by unsigned comparison, or the roaring payload fails to decode. + /// prefix or CRC does not match, the magic is wrong, the roaring bitmap count exceeds the + /// portable format's maximum, the roaring directory's keys are not ordered by unsigned + /// comparison, or the roaring payload fails to decode. // Consumed by the scan delete loader once the deletion-vector read path is wired up. #[allow(dead_code)] pub fn deserialize(blob: &[u8]) -> Result { @@ -115,110 +116,132 @@ impl DeleteVector { // The magic and vector, i.e. the bytes covered by both the length prefix and the CRC. let body = &blob[DV_LENGTH_PREFIX_BYTES..blob.len() - DV_CRC_BYTES]; - - let declared_len = (&blob[..DV_LENGTH_PREFIX_BYTES]) - .try_get_u32() - .map_err(|e| { - Error::new( - ErrorKind::Unexpected, - "failed to read the deletion-vector-v1 length prefix", - ) - .with_source(e) - })? as usize; - if declared_len != body.len() { - return Err(Error::new( - ErrorKind::DataInvalid, - format!( - "deletion-vector-v1 length prefix is {declared_len}, expected {}", - body.len() - ), - )); - } + verify_length_prefix(&blob[..DV_LENGTH_PREFIX_BYTES], body)?; // Verify the CRC before interpreting any bytes so a corrupt blob yields a single clear // error rather than an opaque roaring decode failure. - let stored_crc = (&blob[blob.len() - DV_CRC_BYTES..]) - .try_get_u32() - .map_err(|e| { - Error::new( - ErrorKind::Unexpected, - "failed to read the deletion-vector-v1 CRC", - ) - .with_source(e) - })?; - let computed_crc = crc32fast::hash(body); - if computed_crc != stored_crc { - return Err(Error::new( - ErrorKind::DataInvalid, - format!( - "deletion-vector-v1 CRC mismatch: computed {computed_crc:#010x}, stored {stored_crc:#010x}" - ), - )); - } + verify_crc(body, &blob[blob.len() - DV_CRC_BYTES..])?; let (magic, vector) = body.split_at(DV_MAGIC_BYTES); - if magic != DV_MAGIC { + verify_magic(magic)?; + + let inner = decode_roaring_directory(vector)?; + + Ok(DeleteVector { inner }) + } +} + +fn verify_length_prefix(mut prefix: &[u8], body: &[u8]) -> Result<()> { + let declared_len = prefix.try_get_u32().map_err(|e| { + Error::new( + ErrorKind::Unexpected, + "failed to read the deletion-vector-v1 length prefix", + ) + .with_source(e) + })? as usize; + if declared_len != body.len() { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion-vector-v1 length prefix is {declared_len}, expected {}", + body.len() + ), + )); + } + Ok(()) +} + +fn verify_crc(body: &[u8], mut crc_bytes: &[u8]) -> Result<()> { + let stored_crc = crc_bytes.try_get_u32().map_err(|e| { + Error::new( + ErrorKind::Unexpected, + "failed to read the deletion-vector-v1 CRC", + ) + .with_source(e) + })?; + let computed_crc = crc32fast::hash(body); + if computed_crc != stored_crc { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion-vector-v1 CRC mismatch: computed {computed_crc:#010x}, stored {stored_crc:#010x}" + ), + )); + } + Ok(()) +} + +fn verify_magic(magic: &[u8]) -> Result<()> { + if magic != DV_MAGIC { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("deletion-vector-v1 magic mismatch: {magic:02x?}, expected {DV_MAGIC:02x?}"), + )); + } + Ok(()) +} + +// The Puffin spec defines the roaring directory as the bitmaps "ordered by unsigned comparison +// of the 32-bit keys", with one bitmap per key. Walk it ourselves (rather than +// `RoaringTreemap::deserialize_from`, which stores keys in a `BTreeMap` via a plain insert and +// would silently accept a stream with duplicate or out-of-order keys, discarding the earlier +// bitmap on a duplicate) so a non-conformant blob is rejected instead of decoded into a value +// that doesn't match what was actually written. +fn decode_roaring_directory(mut reader: &[u8]) -> Result { + let bitmap_count = reader.try_get_u64_le().map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + "failed to decode deletion-vector-v1 roaring payload", + ) + .with_source(e) + })?; + // The roaring portable format restricts the bitmap count to [0, 2^32 - 1] (it is stored as a + // u64 with the upper 32 bits reserved as zero padding). + if bitmap_count > u32::MAX as u64 { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion-vector-v1 roaring bitmap count {bitmap_count} exceeds the {}-key maximum", + u32::MAX + ), + )); + } + + let mut bitmaps = Vec::new(); + let mut last_key: Option = None; + for _ in 0..bitmap_count { + let key = reader.try_get_u32_le().map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + "failed to decode deletion-vector-v1 roaring payload", + ) + .with_source(e) + })?; + if let Some(last) = last_key + && key <= last + { return Err(Error::new( ErrorKind::DataInvalid, format!( - "deletion-vector-v1 magic mismatch: {magic:02x?}, expected {DV_MAGIC:02x?}" + "deletion-vector-v1 roaring keys must be ordered by unsigned comparison, got key {key} after {last}" ), )); } + last_key = Some(key); - // The Puffin spec defines the roaring directory as the bitmaps "ordered by unsigned - // comparison of the 32-bit keys", with one bitmap per key. Walk it ourselves (rather than - // `RoaringTreemap::deserialize_from`, which stores keys in a `BTreeMap` via a plain insert - // and would silently accept a stream with duplicate or out-of-order keys, discarding the - // earlier bitmap on a duplicate) so a non-conformant blob is rejected instead of decoded - // into a value that doesn't match what was actually written. - let mut reader = vector; - let bitmap_count = reader.try_get_u64_le().map_err(|e| { + let bitmap = RoaringBitmap::deserialize_from(&mut reader).map_err(|e| { Error::new( ErrorKind::DataInvalid, "failed to decode deletion-vector-v1 roaring payload", ) .with_source(e) })?; - - let mut bitmaps = Vec::new(); - let mut last_key: Option = None; - for _ in 0..bitmap_count { - let key = reader.try_get_u32_le().map_err(|e| { - Error::new( - ErrorKind::DataInvalid, - "failed to decode deletion-vector-v1 roaring payload", - ) - .with_source(e) - })?; - if let Some(last) = last_key - && key <= last - { - return Err(Error::new( - ErrorKind::DataInvalid, - format!( - "deletion-vector-v1 roaring keys must be ordered by unsigned comparison, got key {key} after {last}" - ), - )); - } - last_key = Some(key); - - let bitmap = RoaringBitmap::deserialize_from(&mut reader).map_err(|e| { - Error::new( - ErrorKind::DataInvalid, - "failed to decode deletion-vector-v1 roaring payload", - ) - .with_source(e) - })?; - bitmaps.push((key, bitmap)); - } - - // `bitmaps` is already sorted by key, but `roaring` has no constructor that accepts - // pre-sorted pairs without re-sorting them; revisit if that changes upstream. - let inner = RoaringTreemap::from_bitmaps(bitmaps); - - Ok(DeleteVector { inner }) + bitmaps.push((key, bitmap)); } + + // `bitmaps` is already sorted by key, but `roaring` has no constructor that accepts + // pre-sorted pairs without re-sorting them; revisit if that changes upstream. + Ok(RoaringTreemap::from_bitmaps(bitmaps)) } // Ideally, we'd just wrap `roaring::RoaringTreemap`'s iterator, `roaring::treemap::Iter` here. @@ -493,4 +516,14 @@ mod tests { let err = DeleteVector::deserialize(&frame_dv_blob(&vector)).unwrap_err(); assert!(err.message().contains("unsigned comparison"), "got: {err}"); } + + // The roaring portable format stores the bitmap count as a u64 with the upper 32 bits + // reserved as zero padding, restricting it to [0, 2^32 - 1]; a value above that is not a + // conformant blob, regardless of whether any key/bitmap entries follow. + #[test] + fn test_deserialize_rejects_bitmap_count_overflow() { + let vector = (u32::MAX as u64 + 1).to_le_bytes().to_vec(); + let err = DeleteVector::deserialize(&frame_dv_blob(&vector)).unwrap_err(); + assert!(err.message().contains("exceeds the"), "got: {err}"); + } }