mbutrovich commented on code in PR #2414:
URL: https://github.com/apache/iceberg-rust/pull/2414#discussion_r3554889960
##########
crates/iceberg/src/delete_vector.rs:
##########
@@ -198,4 +292,212 @@ mod tests {
let res = dv.insert_positions(&positions);
assert!(res.is_err());
}
+
+ // ----- deletion-vector-v1 blob deserialization -----
+
+ fn serialize_blob_for_test(dv: &DeleteVector) -> Vec<u8> {
+ let mut roaring = Vec::with_capacity(dv.inner.serialized_size());
+ dv.inner.serialize_into(&mut roaring).unwrap();
+
+ let bitmap_data_len = DV_MAGIC_SIZE + roaring.len();
+ let mut blob = Vec::with_capacity(DV_LENGTH_FIELD_SIZE +
bitmap_data_len + DV_CRC_SIZE);
+ blob.extend_from_slice(&(bitmap_data_len as u32).to_be_bytes());
+ blob.extend_from_slice(&DV_BLOB_MAGIC);
+ blob.extend_from_slice(&roaring);
+ let crc = crc32fast::hash(&blob[DV_LENGTH_FIELD_SIZE..]);
+ blob.extend_from_slice(&crc.to_be_bytes());
+ blob
+ }
+
+ fn dv_from(positions: impl IntoIterator<Item = u64>) -> DeleteVector {
+ let mut dv = DeleteVector::default();
+ for p in positions {
+ dv.insert(p);
+ }
+ dv
+ }
+
+ fn assert_positions(dv: &DeleteVector, expected: &[u64]) {
+ let mut got: Vec<u64> = dv.iter().collect();
+ got.sort_unstable();
+ let mut want = expected.to_vec();
+ want.sort_unstable();
+ assert_eq!(got, want);
+ }
+
+ #[test]
+ fn dv_blob_roundtrip_empty() {
+ let dv = DeleteVector::default();
+ let bytes = serialize_blob_for_test(&dv);
+ let parsed = DeleteVector::deserialize_blob(&bytes, Some(0)).unwrap();
+ assert_eq!(parsed.len(), 0);
+ assert_positions(&parsed, &[]);
+ }
+
+ #[test]
+ fn dv_blob_roundtrip_small() {
+ let dv = dv_from([0, 5, 100]);
+ let bytes = serialize_blob_for_test(&dv);
+ let parsed = DeleteVector::deserialize_blob(&bytes, Some(3)).unwrap();
+ assert_positions(&parsed, &[0, 5, 100]);
+ }
+
+ #[test]
+ fn dv_blob_roundtrip_64bit_keys() {
+ // Positions span multiple inner 32-bit bitmaps.
+ let positions = [1u64, 1u64 << 33, (1u64 << 33) + 5];
+ let dv = dv_from(positions);
+ let bytes = serialize_blob_for_test(&dv);
+ let parsed = DeleteVector::deserialize_blob(&bytes, Some(3)).unwrap();
+ assert_positions(&parsed, &positions);
+ }
+
+ #[test]
+ fn dv_blob_roundtrip_dense_range() {
+ let dv = dv_from(0..10_000);
+ let bytes = serialize_blob_for_test(&dv);
+ let parsed = DeleteVector::deserialize_blob(&bytes,
Some(10_000)).unwrap();
+ assert_eq!(parsed.len(), 10_000);
+ }
+
+ #[test]
+ fn dv_blob_cardinality_mismatch_errors() {
+ let bytes = serialize_blob_for_test(&dv_from([0, 5, 100]));
+ let err = DeleteVector::deserialize_blob(&bytes,
Some(99)).unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(
+ err.message().contains("cardinality mismatch"),
+ "unexpected error: {err}"
+ );
+ }
+
+ #[test]
+ fn dv_blob_cardinality_skip_when_none() {
+ let bytes = serialize_blob_for_test(&dv_from([0, 5, 100]));
+ let parsed = DeleteVector::deserialize_blob(&bytes, None).unwrap();
+ assert_positions(&parsed, &[0, 5, 100]);
+ }
+
+ #[test]
+ fn dv_blob_rejects_short_buffer() {
+ let err = DeleteVector::deserialize_blob(&[0u8; 11],
None).unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(
+ err.message().contains("too short"),
+ "unexpected error: {err}"
+ );
+ }
+
+ #[test]
+ fn dv_blob_rejects_wrong_magic() {
+ let mut bytes = serialize_blob_for_test(&dv_from([0]));
+ bytes[DV_LENGTH_FIELD_SIZE] ^= 0xFF;
+ let err = DeleteVector::deserialize_blob(&bytes, None).unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(
+ err.message().contains("bad magic"),
+ "unexpected error: {err}"
+ );
+ }
+
+ #[test]
+ fn dv_blob_rejects_bad_crc() {
+ let mut bytes = serialize_blob_for_test(&dv_from([0, 1, 2]));
+ // Corrupt a roaring payload byte; roaring may also reject before CRC
check.
+ let mid = DV_LENGTH_FIELD_SIZE + DV_MAGIC_SIZE + 1;
+ bytes[mid] ^= 0xFF;
+ let err = DeleteVector::deserialize_blob(&bytes, None).unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ let msg = err.message();
+ assert!(
+ msg.contains("CRC mismatch") || msg.contains("roaring"),
+ "expected CRC or roaring failure, got: {msg}"
+ );
+ }
+
+ #[test]
+ fn dv_blob_rejects_length_mismatch() {
+ let mut bytes = serialize_blob_for_test(&dv_from([0]));
+ let original_len = u32::from_be_bytes(bytes[0..4].try_into().unwrap());
+ bytes[0..4].copy_from_slice(&(original_len - 1).to_be_bytes());
+ let err = DeleteVector::deserialize_blob(&bytes, None).unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(
+ err.message().contains("length mismatch"),
+ "unexpected error: {err}"
+ );
+ }
+
+ // ----- golden fixtures (see testdata/deletes/README.md) -----
+
+ fn fixture_path(name: &str) -> std::path::PathBuf {
+ std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+ .join("testdata")
+ .join("deletes")
+ .join(name)
+ }
+
+ fn read_fixture(name: &str) -> Vec<u8> {
+ let path = fixture_path(name);
+ std::fs::read(&path).unwrap_or_else(|e| panic!("missing fixture {}:
{e}", path.display()))
+ }
+
+ #[test]
+ fn dv_blob_golden_empty() {
+ let bytes = read_fixture("empty.bin");
+ let dv = DeleteVector::deserialize_blob(&bytes, Some(0)).unwrap();
+ assert_eq!(dv.len(), 0);
+ }
+
+ #[test]
+ fn dv_blob_golden_single_position() {
+ let bytes = read_fixture("single-position.bin");
+ let dv = DeleteVector::deserialize_blob(&bytes, Some(1)).unwrap();
+ assert_positions(&dv, &[0]);
+ }
+
+ #[test]
+ fn dv_blob_golden_small_bitmap() {
+ let bytes = read_fixture("small-bitmap.bin");
+ let dv = DeleteVector::deserialize_blob(&bytes, Some(4)).unwrap();
+ assert_positions(&dv, &[0, 1, 100, 1000]);
+ }
+
+ #[test]
+ fn dv_blob_golden_spanning_keys() {
+ let bytes = read_fixture("spanning-keys.bin");
+ let dv = DeleteVector::deserialize_blob(&bytes, Some(4)).unwrap();
+ assert_positions(&dv, &[0, 1u64 << 33, (1u64 << 33) + 5, 1u64 << 34]);
+ }
+
+ #[test]
+ fn dv_blob_golden_dense_range() {
+ let bytes = read_fixture("dense-range.bin");
+ let dv = DeleteVector::deserialize_blob(&bytes, Some(10_000)).unwrap();
+ assert_eq!(dv.len(), 10_000);
+ assert!(dv.iter().any(|p| p == 0));
+ assert!(dv.iter().any(|p| p == 9_999));
+ }
+
+ /// Regenerate the committed golden fixtures (see testdata README).
+ #[test]
+ #[ignore]
+ fn dv_blob_regenerate_golden_fixtures() {
Review Comment:
This regenerates the fixtures with `serialize_blob_for_test`, which encodes
through roaring's `serialize_into`. So the committed fixtures are produced by
this crate, not by Iceberg-Java, and the `dv_blob_golden_*` tests validate the
codec against its own encoder rather than against a reference. The PR
description says the fixtures come from the Java reference writer, but this
test overwrites them with roaring-encoded bytes.
It also leaves a real gap. Java runs `runLengthEncode` (`runOptimize`) on
each 32-bit bitmap before serializing, so production DVs use RUN containers.
roaring's `serialize_into` does not run-optimize, so none of these fixtures
decode a RUN container, which is the case most likely to break cross-engine
reads.
Commit at least one fixture produced by Iceberg-Java, ideally a run-heavy
one like the dense range, document the exact command that generated it in the
README, and either drop this regenerate test or gate it so it cannot silently
replace the reference bytes. #2678 already committed Java-verified DV fixtures
under `testdata/puffin` that can be reused here.
##########
crates/iceberg/src/delete_vector.rs:
##########
@@ -68,6 +80,88 @@ impl DeleteVector {
pub fn len(&self) -> u64 {
self.inner.len()
}
+
+ #[allow(dead_code)]
+ pub fn deserialize_blob(blob_bytes: &[u8], expected_cardinality:
Option<u64>) -> Result<Self> {
Review Comment:
Add a doc comment on `deserialize_blob`, matching the style of
`insert_positions` above it. State the layout it expects
(`[length][magic][roaring][crc32]`), document that `expected_cardinality` is an
optional cross-check against the manifest `record_count` and is skipped when
`None`, and list the error conditions, with a link to the Puffin spec
`deletion-vector-v1` section. That keeps the byte-format contract next to the
code and settles where the cardinality check lives.
##########
crates/iceberg/src/delete_vector.rs:
##########
@@ -68,6 +80,88 @@ impl DeleteVector {
pub fn len(&self) -> u64 {
self.inner.len()
}
+
+ #[allow(dead_code)]
+ pub fn deserialize_blob(blob_bytes: &[u8], expected_cardinality:
Option<u64>) -> Result<Self> {
+ if blob_bytes.len() < DV_MIN_BLOB_SIZE {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "deletion-vector-v1 blob too short: {} bytes (minimum
{DV_MIN_BLOB_SIZE})",
+ blob_bytes.len()
+ ),
+ ));
+ }
+
+ // 2GB cap matches Java's signed-i32 length field.
+ if blob_bytes.len() > i32::MAX as usize {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "deletion-vector-v1 blob exceeds 2GB: {} bytes",
+ blob_bytes.len()
+ ),
+ ));
+ }
+
+ let declared_len =
+
u32::from_be_bytes(blob_bytes[0..DV_LENGTH_FIELD_SIZE].try_into().unwrap()) as
usize;
+ let expected_len = blob_bytes.len() - DV_LENGTH_FIELD_SIZE -
DV_CRC_SIZE;
+ if declared_len != expected_len {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "deletion-vector-v1 length mismatch: declared
{declared_len}, actual {expected_len}"
+ ),
+ ));
+ }
+
+ let magic_end = DV_LENGTH_FIELD_SIZE + DV_MAGIC_SIZE;
+ let magic = &blob_bytes[DV_LENGTH_FIELD_SIZE..magic_end];
+ if magic != DV_BLOB_MAGIC {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "deletion-vector-v1 bad magic: {magic:02X?}, expected
{DV_BLOB_MAGIC:02X?}"
+ ),
+ ));
+ }
+
+ let roaring_end = blob_bytes.len() - DV_CRC_SIZE;
+ let mut cursor = Cursor::new(&blob_bytes[magic_end..roaring_end]);
+ let inner = RoaringTreemap::deserialize_from(&mut cursor).map_err(|e| {
Review Comment:
Check the CRC before deserializing the roaring payload. Right now the
payload is parsed here and the CRC is verified at line 141, so a corrupted blob
fails with an opaque roaring parse error instead of a clear CRC mismatch, and
`dv_blob_rejects_bad_crc` having to accept either error is the symptom. The CRC
is a single cheap pass over `blob_bytes[4..len-4]`, so verifying it first gives
a deterministic error and keeps corrupt bytes out of the parser.
##########
crates/iceberg/src/delete_vector.rs:
##########
@@ -23,6 +24,17 @@ use roaring::treemap::BitmapIter;
use crate::{Error, ErrorKind, Result};
+#[allow(dead_code)]
Review Comment:
These constants and `deserialize_blob` are all `#[allow(dead_code)]` because
nothing calls them in this PR. Add a reference to the read tracking issue #2792
next to them, and remove the allows when the loader wiring lands, so they do
not become permanent.
##########
crates/iceberg/src/delete_vector.rs:
##########
@@ -68,6 +80,88 @@ impl DeleteVector {
pub fn len(&self) -> u64 {
self.inner.len()
}
+
+ #[allow(dead_code)]
+ pub fn deserialize_blob(blob_bytes: &[u8], expected_cardinality:
Option<u64>) -> Result<Self> {
Review Comment:
`deserialize_blob` reads a full `deletion-vector-v1` blob. #2678 and #2681
use `from_puffin_blob` / `from_serialized_bytes` for the same operation, and
Iceberg-Java names it `deserialize` on the index type. Pick one name now so it
does not get renamed when these land together.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
