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

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"

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

@@ -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,88 @@ 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<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 =
+            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 +288,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<u8> {
+        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<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 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}");
+    }
 }