feat(delete-vector): [1/N] decode deletion-vector-v1 puffin blobs

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"

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 }

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,148 @@ 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 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<Self> {
+        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 = (&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 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}"
+                ),
+            ));
+        }
+
+        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?}"
+                ),
+            ));
+        }
+
+        // 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",
+            )
+            .with_source(e)
+        })?;
+
+        let mut bitmaps = Vec::new();
+        let mut last_key: Option<u32> = 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 })
+    }
 }
 
 // Ideally, we'd just wrap `roaring::RoaringTreemap`'s iterator, `roaring::treemap::Iter` here.
@@ -198,4 +349,148 @@ 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<u8> {
+        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<u8> {
+        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<Item = u64>) -> DeleteVector {
+        let mut dv = DeleteVector::default();
+        for pos in positions {
+            dv.insert(pos);
+        }
+        dv
+    }
+
+    fn sorted(dv: &DeleteVector) -> Vec<u64> {
+        let mut positions: Vec<u64> = 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<u8> {
+        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}");
+    }
 }