diff --git a/Cargo.lock b/Cargo.lock index c9cb4ed232..38804f6d6d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3780,6 +3780,7 @@ dependencies = [ "bimap", "bytes", "chrono", + "crc32fast", "derive_builder", "expect-test", "fastnum", diff --git a/Cargo.toml b/Cargo.toml index 8e69861a35..aec84da8eb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -77,6 +77,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.1.0" datafusion-cli = "54.1.0" diff --git a/crates/iceberg/Cargo.toml b/crates/iceberg/Cargo.toml index 66eadfb7ec..9219c728a5 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..cfcc29700c 100644 --- a/crates/iceberg/src/delete_vector.rs +++ b/crates/iceberg/src/delete_vector.rs @@ -17,12 +17,21 @@ 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}; +/// 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 +77,171 @@ 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: 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. + /// + /// # 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 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 { + 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]; + 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. + verify_crc(body, &blob[blob.len() - DV_CRC_BYTES..])?; + + let (magic, vector) = body.split_at(DV_MAGIC_BYTES); + 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 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. + Ok(RoaringTreemap::from_bitmaps(bitmaps)) } // Ideally, we'd just wrap `roaring::RoaringTreemap`'s iterator, `roaring::treemap::Iter` here. @@ -198,4 +372,158 @@ 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 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); + 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 { + dv.insert(pos); + } + dv + } + + fn sorted(dv: &DeleteVector) -> Vec { + let mut positions: Vec = dv.iter().collect(); + positions.sort_unstable(); + positions + } + + #[test] + fn test_deserialize_roundtrip_empty() { + let blob = encode_dv_blob(&DeleteVector::default()); + assert_eq!(DeleteVector::deserialize(&blob).unwrap().len(), 0); + } + + #[test] + 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 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); + } + + // 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 test_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); + let positions = sorted(&decoded); + assert_eq!(positions.first(), Some(&0)); + assert_eq!(positions.last(), Some(&9_999)); + } + + #[test] + 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 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. + 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 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; + let err = DeleteVector::deserialize(&blob).unwrap_err(); + assert!(err.message().contains("CRC mismatch"), "got: {err}"); + } + + #[test] + 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()); + 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}"); + } + + // 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}"); + } }