amogh-jahagirdar commented on code in PR #3478:
URL: https://github.com/apache/iceberg-python/pull/3478#discussion_r3591278366
##########
pyiceberg/table/deletion_vector.py:
##########
@@ -77,17 +89,103 @@ def to_vector(self) -> "pa.ChunkedArray":
return self._bitmaps_to_chunked_array(self._bitmaps)
-def _extract_vector_payload(blob_payload: bytes) -> bytes:
- """Strip deletion-vector-v1 blob framing: length(4 big-endian) + DV
magic(4) ... CRC(4 big-endian)."""
- length_prefix = int.from_bytes(blob_payload[0:4], "big")
- return blob_payload[8 : 4 + length_prefix]
+def _deserialize_dv_blob(blob: bytes, record_count: int | None = None) ->
list[BitMap]:
+ # The DV blob encoding matches Iceberg Java's BitmapPositionDeleteIndex:
+ # 4-byte big-endian bitmap-data length, 4-byte little-endian magic number,
+ # portable Roaring bitmap data, and 4-byte big-endian CRC-32.
+ if len(blob) < _DV_BLOB_MIN_SIZE_BYTES:
+ raise ValueError(f"Invalid deletion vector blob length: {len(blob)}")
+
+ bitmap_data_length = _DV_BLOB_LENGTH.unpack_from(blob)[0]
+ expected_bitmap_data_length = len(blob) - _DV_BLOB_LENGTH.size -
_DV_BLOB_CRC.size
+ if bitmap_data_length != expected_bitmap_data_length:
+ raise ValueError(f"Invalid bitmap data length: {bitmap_data_length},
expected {expected_bitmap_data_length}")
+
+ bitmap_data_offset = _DV_BLOB_LENGTH.size
+ crc_offset = bitmap_data_offset + bitmap_data_length
+ bitmap_data = blob[bitmap_data_offset:crc_offset]
+
+ magic_number = _DV_BLOB_MAGIC.unpack_from(bitmap_data)[0]
+ if magic_number != _DV_BLOB_MAGIC_NUMBER:
+ raise ValueError(f"Invalid magic number: {magic_number}, expected
{_DV_BLOB_MAGIC_NUMBER}")
+
+ checksum = zlib.crc32(bitmap_data) & 0xFFFFFFFF
+ expected_checksum = _DV_BLOB_CRC.unpack_from(blob, crc_offset)[0]
+ if checksum != expected_checksum:
+ raise ValueError("Invalid CRC")
+
+ bitmaps =
DeletionVector._deserialize_bitmap(bitmap_data[_DV_BLOB_MAGIC.size :])
+ if record_count is not None:
+ cardinality = sum(len(bitmap) for bitmap in bitmaps)
+ if cardinality != record_count:
+ raise ValueError(f"Invalid cardinality: {cardinality}, expected
{record_count}")
+
+ return bitmaps
+
+
+def _validate_deletion_vector_content(data_file: "DataFile") -> None:
+ content_offset = data_file.content_offset
+ content_size_in_bytes = data_file.content_size_in_bytes
+ referenced_data_file = data_file.referenced_data_file
+
+ if content_offset is None:
+ raise ValueError(f"Invalid deletion vector, content offset is missing:
{data_file.file_path}")
+ if content_size_in_bytes is None:
+ raise ValueError(f"Invalid deletion vector, content size is missing:
{data_file.file_path}")
+ if content_offset < 0:
+ raise ValueError(f"Invalid deletion vector, content offset cannot be
negative: {content_offset}")
+ if content_size_in_bytes < 0:
+ raise ValueError(f"Invalid deletion vector, content size cannot be
negative: {content_size_in_bytes}")
+ if content_size_in_bytes > _MAX_DELETION_VECTOR_CONTENT_SIZE:
+ raise ValueError(f"Cannot read deletion vector larger than 2GB:
{content_size_in_bytes}")
+ if referenced_data_file is None:
+ raise ValueError(f"Invalid deletion vector, referenced data file is
missing: {data_file.file_path}")
+
+
+def has_deletion_vector_content_reference(data_file: "DataFile") -> bool:
+ return (
+ data_file.content_offset is not None
+ or data_file.content_size_in_bytes is not None
+ or data_file.referenced_data_file is not None
+ )
+
+
+def _read_deletion_vector(io: "FileIO", data_file: "DataFile") ->
DeletionVector:
+ _validate_deletion_vector_content(data_file)
+
+ content_offset = data_file.content_offset
+ content_size_in_bytes = data_file.content_size_in_bytes
+ referenced_data_file = data_file.referenced_data_file
+ assert content_offset is not None
+ assert content_size_in_bytes is not None
+ assert referenced_data_file is not None
+
+ with io.new_input(data_file.file_path).open() as fi:
+ fi.seek(content_offset)
+ payload = fi.read(content_size_in_bytes)
+
+ if len(payload) != content_size_in_bytes:
+ raise ValueError(f"Could not read deletion vector, expected
{content_size_in_bytes} bytes, got {len(payload)}")
+
+ return DeletionVector(
+ referenced_data_file=referenced_data_file,
+ bitmaps=_deserialize_dv_blob(payload, data_file.record_count),
+ )
+
+
+def read_deletion_vectors(io: "FileIO", data_file: "DataFile") ->
list[DeletionVector]:
+ if has_deletion_vector_content_reference(data_file):
+ return [_read_deletion_vector(io, data_file)]
+
+ with io.new_input(data_file.file_path).open() as fi:
+ return deletion_vectors_from_puffin_file(PuffinFile(fi.read()))
Review Comment:
See
https://github.com/apache/iceberg-python/pull/3478#pullrequestreview-4708727926
for more details.
--
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]
