From 0e99544a7385a9fc5af21ae8a80faa86a2eb5362 Mon Sep 17 00:00:00 2001 From: KaiqiJinWow Date: Mon, 6 Jul 2026 17:29:46 +0000 Subject: [PATCH 1/6] Support range-based reads for deletion vectors --- pyiceberg/io/pyarrow.py | 11 +- pyiceberg/table/__init__.py | 9 +- pyiceberg/table/delete_file.py | 96 ++++++++++++++++ pyiceberg/table/delete_file_index.py | 7 +- pyiceberg/table/deletion_vector.py | 106 ++++++++++++++++- tests/io/test_pyarrow.py | 158 ++++++++++++++++++++++++++ tests/table/test_delete_file_index.py | 64 ++++++++++- tests/table/test_deletion_vector.py | 75 +++++++++++- 8 files changed, 503 insertions(+), 23 deletions(-) create mode 100644 pyiceberg/table/delete_file.py diff --git a/pyiceberg/io/pyarrow.py b/pyiceberg/io/pyarrow.py index c36f1639d9..fd286270c2 100644 --- a/pyiceberg/io/pyarrow.py +++ b/pyiceberg/io/pyarrow.py @@ -147,11 +147,11 @@ visit_with_partner, ) from pyiceberg.table import DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE, TableProperties -from pyiceberg.table.deletion_vector import deletion_vectors_from_puffin_file +from pyiceberg.table.delete_file import DeleteFileSet +from pyiceberg.table.deletion_vector import read_deletion_vectors from pyiceberg.table.locations import load_location_provider from pyiceberg.table.metadata import TableMetadata from pyiceberg.table.name_mapping import NameMapping, apply_name_mapping -from pyiceberg.table.puffin import PuffinFile from pyiceberg.transforms import IdentityTransform, TruncateTransform from pyiceberg.typedef import EMPTY_DICT, Properties, Record, TableVersion from pyiceberg.types import ( @@ -1149,10 +1149,7 @@ def _read_deletes(io: FileIO, data_file: DataFile) -> dict[str, pa.ChunkedArray] for path in table.column("file_path").unique() } elif data_file.file_format == FileFormat.PUFFIN: - with io.new_input(data_file.file_path).open() as fi: - payload = fi.read() - - return {dv.referenced_data_file: dv.to_vector() for dv in deletion_vectors_from_puffin_file(PuffinFile(payload))} + return {dv.referenced_data_file: dv.to_vector() for dv in read_deletion_vectors(io, data_file)} else: raise ValueError(f"Delete file format not supported: {data_file.file_format}") @@ -1721,7 +1718,7 @@ def _task_to_record_batches( def _read_all_delete_files(io: FileIO, tasks: Iterable[FileScanTask]) -> dict[str, list[ChunkedArray]]: deletes_per_file: dict[str, list[ChunkedArray]] = {} - unique_deletes = set(itertools.chain.from_iterable([task.delete_files for task in tasks])) + unique_deletes = DeleteFileSet(itertools.chain.from_iterable(task.delete_files for task in tasks)) if len(unique_deletes) > 0: executor = ExecutorFactory.get_or_create() deletes_per_files: Iterator[dict[str, ChunkedArray]] = executor.map( diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 9624eac981..2d01585514 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -48,6 +48,7 @@ from pyiceberg.manifest import DataFile, DataFileContent, ManifestContent, ManifestEntry, ManifestEntryStatus, ManifestFile from pyiceberg.partitioning import PARTITION_FIELD_ID_START, UNPARTITIONED_PARTITION_SPEC, PartitionKey, PartitionSpec from pyiceberg.schema import Schema +from pyiceberg.table.delete_file import DeleteFileSet from pyiceberg.table.delete_file_index import DeleteFileIndex from pyiceberg.table.inspect import InspectTable from pyiceberg.table.locations import LocationProvider, load_location_provider @@ -2232,17 +2233,17 @@ class FileScanTask(ScanTask): """Task representing a data file and its corresponding delete files.""" file: DataFile - delete_files: set[DataFile] + delete_files: DeleteFileSet residual: BooleanExpression def __init__( self, data_file: DataFile, - delete_files: set[DataFile] | None = None, + delete_files: Iterable[DataFile] | None = None, residual: BooleanExpression = ALWAYS_TRUE, ) -> None: self.file = data_file - self.delete_files = delete_files or set() + self.delete_files = DeleteFileSet(delete_files if delete_files is not None else []) self.residual = residual @staticmethod @@ -2266,7 +2267,7 @@ def from_rest_response( data_file = _rest_file_to_data_file(rest_task.data_file) - resolved_deletes: set[DataFile] = set() + resolved_deletes = DeleteFileSet() if rest_task.delete_file_references: for idx in rest_task.delete_file_references: delete_file = delete_files[idx] diff --git a/pyiceberg/table/delete_file.py b/pyiceberg/table/delete_file.py new file mode 100644 index 0000000000..681b0eef9f --- /dev/null +++ b/pyiceberg/table/delete_file.py @@ -0,0 +1,96 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +from collections.abc import Iterable, Iterator, MutableSet +from dataclasses import dataclass +from typing import Any + +from pyiceberg.manifest import DataFile + + +@dataclass(frozen=True, slots=True) +class DeleteFileKey: + """Identity of a delete file, including its referenced content range.""" + + file_path: str + content_offset: int | None + content_size_in_bytes: int | None + + @classmethod + def from_file(cls, delete_file: DataFile) -> DeleteFileKey: + """Create a key from a delete file.""" + return cls( + file_path=delete_file.file_path, + content_offset=delete_file.content_offset, + content_size_in_bytes=delete_file.content_size_in_bytes, + ) + + +class DeleteFileSet(MutableSet[DataFile]): + """Set-like delete-file collection keyed by location and content range.""" + + _files: dict[DeleteFileKey, DataFile] + + def __init__(self, delete_files: Iterable[DataFile] = ()) -> None: + self._files = {} + for delete_file in delete_files: + self.add(delete_file) + + def __contains__(self, delete_file: object) -> bool: + """Return whether the delete file is present.""" + return isinstance(delete_file, DataFile) and DeleteFileKey.from_file(delete_file) in self._files + + def __iter__(self) -> Iterator[DataFile]: + """Return an iterator over delete files.""" + return iter(self._files.values()) + + def __len__(self) -> int: + """Return the number of delete files.""" + return len(self._files) + + def add(self, delete_file: DataFile) -> None: + self._files.setdefault(DeleteFileKey.from_file(delete_file), delete_file) + + def discard(self, delete_file: DataFile) -> None: + self._files.pop(DeleteFileKey.from_file(delete_file), None) + + def update(self, delete_files: Iterable[DataFile]) -> None: + for delete_file in delete_files: + self.add(delete_file) + + def __repr__(self) -> str: + """Return a string representation of the delete file set.""" + return f"{type(self).__name__}({list(self)!r})" + + def __eq__(self, other: Any) -> bool: + """Compare delete file sets by delete file identity.""" + if isinstance(other, DeleteFileSet): + return self._files.keys() == other._files.keys() + + if not isinstance(other, Iterable): + return False + + other_keys: set[DeleteFileKey] = set() + other_count = 0 + for delete_file in other: + if not isinstance(delete_file, DataFile): + return False + other_keys.add(DeleteFileKey.from_file(delete_file)) + other_count += 1 + + return len(other_keys) == other_count and set(self._files) == other_keys diff --git a/pyiceberg/table/delete_file_index.py b/pyiceberg/table/delete_file_index.py index 3f513aabe5..b5904a9491 100644 --- a/pyiceberg/table/delete_file_index.py +++ b/pyiceberg/table/delete_file_index.py @@ -21,6 +21,7 @@ from pyiceberg.expressions import EqualTo from pyiceberg.expressions.visitors import _InclusiveMetricsEvaluator from pyiceberg.manifest import INITIAL_SEQUENCE_NUMBER, POSITIONAL_DELETE_SCHEMA, DataFile, ManifestEntry +from pyiceberg.table.delete_file import DeleteFileSet from pyiceberg.typedef import Record PATH_FIELD_ID = 2147483546 @@ -125,11 +126,11 @@ def add_delete_file(self, manifest_entry: ManifestEntry, partition_key: Record | deletes = self._by_partition.setdefault(key, PositionDeletes()) deletes.add(delete_file, seq) - def for_data_file(self, seq_num: int, data_file: DataFile, partition_key: Record | None = None) -> set[DataFile]: + def for_data_file(self, seq_num: int, data_file: DataFile, partition_key: Record | None = None) -> DeleteFileSet: if self.is_empty(): - return set() + return DeleteFileSet() - deletes: set[DataFile] = set() + deletes = DeleteFileSet() spec_id = data_file.spec_id or 0 key = _partition_key(spec_id, partition_key) diff --git a/pyiceberg/table/deletion_vector.py b/pyiceberg/table/deletion_vector.py index 88fb3daf73..7094087404 100644 --- a/pyiceberg/table/deletion_vector.py +++ b/pyiceberg/table/deletion_vector.py @@ -15,7 +15,9 @@ # specific language governing permissions and limitations # under the License. import math -from typing import TYPE_CHECKING +import struct +import zlib +from typing import TYPE_CHECKING, cast from pyroaring import BitMap, FrozenBitMap @@ -24,9 +26,19 @@ if TYPE_CHECKING: import pyarrow as pa + from pyiceberg.io import FileIO + from pyiceberg.manifest import DataFile + EMPTY_BITMAP = FrozenBitMap() MAX_JAVA_SIGNED = int(math.pow(2, 31)) - 1 PROPERTY_REFERENCED_DATA_FILE = "referenced-data-file" +_MAX_DELETION_VECTOR_CONTENT_SIZE = 2**31 - 1 +_DV_BLOB_LENGTH = struct.Struct(">I") +_DV_BLOB_MAGIC = struct.Struct("I") +_DV_BLOB_MAGIC_NUMBER = 1681511377 +_ROARING_BITMAP_COUNT_SIZE_BYTES = 8 +_DV_BLOB_MIN_SIZE_BYTES = _DV_BLOB_LENGTH.size + _DV_BLOB_MAGIC.size + _ROARING_BITMAP_COUNT_SIZE_BYTES + _DV_BLOB_CRC.size class DeletionVector: @@ -77,17 +89,99 @@ 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(dv: "DataFile") -> None: + content_offset = dv.content_offset + content_size_in_bytes = dv.content_size_in_bytes + referenced_data_file = dv.referenced_data_file + + if content_offset is None: + raise ValueError(f"Invalid deletion vector, content offset is missing: {dv.file_path}") + if content_size_in_bytes is None: + raise ValueError(f"Invalid deletion vector, content size is missing: {dv.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: {dv.file_path}") + + +def has_deletion_vector_content_reference(dv: "DataFile") -> bool: + """Return whether a deletion vector is described by manifest content-range metadata.""" + return dv.content_offset is not None or dv.content_size_in_bytes is not None or dv.referenced_data_file is not None + + +def _read_deletion_vector(io: "FileIO", dv: "DataFile") -> DeletionVector: + _validate_deletion_vector_content(dv) + + content_offset = cast(int, dv.content_offset) + content_size_in_bytes = cast(int, dv.content_size_in_bytes) + referenced_data_file = cast(str, dv.referenced_data_file) + + with io.new_input(dv.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, dv.record_count), + ) + + +def read_deletion_vectors(io: "FileIO", dv: "DataFile") -> list[DeletionVector]: + """Read deletion vectors from a delete file or its manifest content range.""" + if has_deletion_vector_content_reference(dv): + return [_read_deletion_vector(io, dv)] + + with io.new_input(dv.file_path).open() as fi: + return deletion_vectors_from_puffin_file(PuffinFile(fi.read())) def deletion_vectors_from_puffin_file(puffin_file: PuffinFile) -> list[DeletionVector]: + """Read all deletion vectors stored in a Puffin file.""" return [ DeletionVector( referenced_data_file=blob.properties[PROPERTY_REFERENCED_DATA_FILE], - bitmaps=DeletionVector._deserialize_bitmap(_extract_vector_payload(puffin_file.get_blob_payload(blob))), + bitmaps=_deserialize_dv_blob(puffin_file.get_blob_payload(blob)), ) for blob in puffin_file.footer.blobs ] diff --git a/tests/io/test_pyarrow.py b/tests/io/test_pyarrow.py index b31c18949b..460cfd4eaf 100644 --- a/tests/io/test_pyarrow.py +++ b/tests/io/test_pyarrow.py @@ -15,12 +15,15 @@ # specific language governing permissions and limitations # under the License. # pylint: disable=protected-access,unused-argument,redefined-outer-name +import json import logging import os +import struct import sys import tempfile import uuid import warnings +import zlib from collections.abc import Iterator from datetime import date, datetime, timezone from pathlib import Path @@ -36,6 +39,7 @@ import pytest from packaging import version from pyarrow.fs import AwsDefaultS3RetryStrategy, FileType, LocalFileSystem, S3FileSystem +from pyroaring import BitMap from pyiceberg.exceptions import ResolveError from pyiceberg.expressions import ( @@ -75,6 +79,7 @@ _ConvertToArrowSchema, _determine_partitions, _primitive_to_physical, + _read_all_delete_files, _read_deletes, _task_to_record_batches, _to_requested_schema, @@ -91,8 +96,14 @@ from pyiceberg.partitioning import PartitionField, PartitionSpec from pyiceberg.schema import Schema, make_compatible_name, visit from pyiceberg.table import FileScanTask, TableProperties, WriteTask +from pyiceberg.table.deletion_vector import ( + _DV_BLOB_MAGIC_NUMBER, + PROPERTY_REFERENCED_DATA_FILE, + deletion_vectors_from_puffin_file, +) from pyiceberg.table.metadata import TableMetadataV2 from pyiceberg.table.name_mapping import create_mapping_from_schema +from pyiceberg.table.puffin import MAGIC_BYTES, PuffinFile from pyiceberg.transforms import HourTransform, IdentityTransform from pyiceberg.typedef import UTF8, Properties, Record, TableVersion from pyiceberg.types import ( @@ -1822,6 +1833,153 @@ def test_read_deletes(deletes_file: str, request: pytest.FixtureRequest) -> None assert list(deletes.values())[0] == pa.chunked_array([[1, 3, 5]]) +def _deletion_vector_bitmap_payload(values: list[int] | None = None) -> bytes: + return (1).to_bytes(8, byteorder="little") + (0).to_bytes(4, byteorder="little") + BitMap(values or [1, 3, 5]).serialize() + + +def _deletion_vector_blob(bitmap_payload: bytes) -> bytes: + bitmap_data = struct.pack("I", len(bitmap_data)) + bitmap_data + struct.pack(">I", zlib.crc32(bitmap_data) & 0xFFFFFFFF) + + +def _deletion_vector_puffin_payload(referenced_data_file: str) -> bytes: + dv_blob = _deletion_vector_blob(_deletion_vector_bitmap_payload()) + footer_payload = json.dumps( + { + "blobs": [ + { + "type": "deletion-vector-v1", + "fields": [2147483546], + "snapshot-id": 1, + "sequence-number": 1, + "offset": len(MAGIC_BYTES) + 4, + "length": len(dv_blob), + "properties": {PROPERTY_REFERENCED_DATA_FILE: referenced_data_file}, + } + ], + "properties": {}, + } + ).encode() + + return ( + MAGIC_BYTES + + b"\x00\x00\x00\x00" + + dv_blob + + footer_payload + + len(footer_payload).to_bytes(4, byteorder="little") + + b"\x00\x00\x00\x00" + + MAGIC_BYTES + ) + + +def test_deletion_vectors_from_puffin_file(tmp_path: Path) -> None: + referenced_data_file = f"{tmp_path}/data.parquet" + puffin_payload = _deletion_vector_puffin_payload(referenced_data_file) + delete_file_path = f"{tmp_path}/deletes.puffin" + + with open(delete_file_path, "wb") as f: + f.write(puffin_payload) + + with open(delete_file_path, "rb") as f: + deletion_vectors = deletion_vectors_from_puffin_file(PuffinFile(f.read())) + + assert {dv.referenced_data_file: dv.to_vector() for dv in deletion_vectors} == { + referenced_data_file: pa.chunked_array([[1, 3, 5]]) + } + + +def test_read_deletes_from_whole_puffin_file(tmp_path: Path) -> None: + referenced_data_file = f"{tmp_path}/data.parquet" + delete_file_path = f"{tmp_path}/deletes.puffin" + + with open(delete_file_path, "wb") as f: + f.write(_deletion_vector_puffin_payload(referenced_data_file)) + + deletes = _read_deletes( + PyArrowFileIO(), + DataFile.from_args( + _table_format_version=3, + content=DataFileContent.POSITION_DELETES, + file_path=delete_file_path, + file_format=FileFormat.PUFFIN, + record_count=3, + ), + ) + + assert deletes == {referenced_data_file: pa.chunked_array([[1, 3, 5]])} + + +def test_read_deletion_vector_blob_from_content_range(tmp_path: Path) -> None: + referenced_data_file = f"{tmp_path}/data.parquet" + dv_blob = _deletion_vector_blob(_deletion_vector_bitmap_payload()) + prefix = b"\x01not-a-puffin-file" + delete_file_path = f"{tmp_path}/deletes.bin" + + with open(delete_file_path, "wb") as f: + f.write(prefix + dv_blob + b"trailing-bytes") + + deletes = _read_deletes( + PyArrowFileIO(), + DataFile.from_args( + _table_format_version=3, + content=DataFileContent.POSITION_DELETES, + file_path=delete_file_path, + file_format=FileFormat.PUFFIN, + record_count=3, + referenced_data_file=referenced_data_file, + content_offset=len(prefix), + content_size_in_bytes=len(dv_blob), + ), + ) + + assert deletes == {referenced_data_file: pa.chunked_array([[1, 3, 5]])} + + +def test_read_all_delete_files_keeps_multiple_dv_content_ranges_for_same_path(tmp_path: Path) -> None: + referenced_data_file = f"{tmp_path}/data.parquet" + first_dv_blob = _deletion_vector_blob(_deletion_vector_bitmap_payload([1, 3])) + second_dv_blob = _deletion_vector_blob(_deletion_vector_bitmap_payload([5])) + delete_file_path = f"{tmp_path}/deletes.bin" + + with open(delete_file_path, "wb") as f: + f.write(first_dv_blob + second_dv_blob) + + first_dv = DataFile.from_args( + _table_format_version=3, + content=DataFileContent.POSITION_DELETES, + file_path=delete_file_path, + file_format=FileFormat.PUFFIN, + record_count=2, + referenced_data_file=referenced_data_file, + content_offset=0, + content_size_in_bytes=len(first_dv_blob), + ) + second_dv = DataFile.from_args( + _table_format_version=3, + content=DataFileContent.POSITION_DELETES, + file_path=delete_file_path, + file_format=FileFormat.PUFFIN, + record_count=1, + referenced_data_file=referenced_data_file, + content_offset=len(first_dv_blob), + content_size_in_bytes=len(second_dv_blob), + ) + data_file = DataFile.from_args( + content=DataFileContent.DATA, + file_path=referenced_data_file, + file_format=FileFormat.PARQUET, + record_count=10, + file_size_in_bytes=100, + ) + + deletes = _read_all_delete_files( + PyArrowFileIO(), + [FileScanTask(data_file=data_file, delete_files=[first_dv, second_dv])], + ) + + assert sorted(delete.to_pylist() for delete in deletes[referenced_data_file]) == [[1, 3], [5]] + + def test_delete(deletes_file: str, request: pytest.FixtureRequest, table_schema_simple: Schema) -> None: # Determine file format from the file extension file_format = FileFormat.PARQUET if deletes_file.endswith(".parquet") else FileFormat.ORC diff --git a/tests/table/test_delete_file_index.py b/tests/table/test_delete_file_index.py index 09dd9ac81b..6a02244363 100644 --- a/tests/table/test_delete_file_index.py +++ b/tests/table/test_delete_file_index.py @@ -17,6 +17,7 @@ import pytest from pyiceberg.manifest import DataFile, DataFileContent, FileFormat, ManifestEntry, ManifestEntryStatus +from pyiceberg.table.delete_file import DeleteFileSet from pyiceberg.table.delete_file_index import PATH_FIELD_ID, DeleteFileIndex, PositionDeletes from pyiceberg.typedef import Record @@ -65,17 +66,26 @@ def _create_partition_delete(sequence_number: int = 1, spec_id: int = 0, partiti def _create_deletion_vector( - sequence_number: int = 1, file_path: str = "s3://bucket/data.parquet", spec_id: int = 0 + sequence_number: int = 1, + file_path: str = "s3://bucket/data.parquet", + spec_id: int = 0, + delete_file_path: str | None = None, + content_offset: int | None = None, + content_size_in_bytes: int | None = None, ) -> ManifestEntry: delete_file = DataFile.from_args( + _table_format_version=3, content=DataFileContent.POSITION_DELETES, - file_path=f"s3://bucket/deletion-vector-{sequence_number}.puffin", + file_path=delete_file_path or f"s3://bucket/deletion-vector-{sequence_number}.puffin", file_format=FileFormat.PUFFIN, partition=Record(), record_count=10, file_size_in_bytes=100, lower_bounds={PATH_FIELD_ID: file_path.encode()}, upper_bounds={PATH_FIELD_ID: file_path.encode()}, + referenced_data_file=file_path, + content_offset=content_offset, + content_size_in_bytes=content_size_in_bytes, ) delete_file._spec_id = spec_id return ManifestEntry.from_args(status=ManifestEntryStatus.ADDED, sequence_number=sequence_number, data_file=delete_file) @@ -161,6 +171,56 @@ def test_dvs_treated_as_position_deletes() -> None: assert all(d.content == DataFileContent.POSITION_DELETES for d in result) +def test_delete_file_set_uses_content_range_identity() -> None: + shared_file_path = "s3://bucket/deletion-vectors.bin" + first_dv = _create_deletion_vector( + sequence_number=2, + delete_file_path=shared_file_path, + content_offset=4, + content_size_in_bytes=10, + ).data_file + second_dv = _create_deletion_vector( + sequence_number=3, + delete_file_path=shared_file_path, + content_offset=40, + content_size_in_bytes=12, + ).data_file + + assert first_dv == second_dv + assert len(DeleteFileSet([first_dv, second_dv])) == 2 + + +def test_dvs_with_same_file_path_and_different_content_ranges_are_not_deduped() -> None: + index = DeleteFileIndex() + shared_file_path = "s3://bucket/deletion-vectors.bin" + + index.add_delete_file( + _create_deletion_vector( + sequence_number=2, + delete_file_path=shared_file_path, + content_offset=4, + content_size_in_bytes=10, + ) + ) + index.add_delete_file( + _create_deletion_vector( + sequence_number=3, + delete_file_path=shared_file_path, + content_offset=40, + content_size_in_bytes=12, + ) + ) + + data_file = _create_data_file() + result = index.for_data_file(1, data_file) + + assert len(result) == 2 + assert {(dv.file_path, dv.content_offset, dv.content_size_in_bytes) for dv in result} == { + (shared_file_path, 4, 10), + (shared_file_path, 40, 12), + } + + def test_cannot_add_after_indexing() -> None: group = PositionDeletes() group.add(_create_positional_delete(sequence_number=1).data_file, 1) diff --git a/tests/table/test_deletion_vector.py b/tests/table/test_deletion_vector.py index 788216f8b3..3f4e8f7cfe 100644 --- a/tests/table/test_deletion_vector.py +++ b/tests/table/test_deletion_vector.py @@ -14,12 +14,20 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +import struct +import zlib from os import path import pytest from pyroaring import BitMap -from pyiceberg.table.deletion_vector import DeletionVector +from pyiceberg.manifest import DataFile, DataFileContent, FileFormat +from pyiceberg.table.deletion_vector import ( + _DV_BLOB_MAGIC_NUMBER, + DeletionVector, + _deserialize_dv_blob, + has_deletion_vector_content_reference, +) def _open_file(file: str) -> bytes: @@ -28,6 +36,71 @@ def _open_file(file: str) -> bytes: return f.read() +def _dv_blob(bitmap_payload: bytes) -> bytes: + bitmap_data = struct.pack("I", len(bitmap_data)) + bitmap_data + struct.pack(">I", zlib.crc32(bitmap_data) & 0xFFFFFFFF) + + +def _bitmap_payload() -> bytes: + return (1).to_bytes(8, byteorder="little") + (0).to_bytes(4, byteorder="little") + BitMap([1, 3, 5]).serialize() + + +def test_deserialize_deletion_vector_blob() -> None: + actual = _deserialize_dv_blob(_dv_blob(_bitmap_payload()), record_count=3) + + assert actual == [BitMap([1, 3, 5])] + + +def test_deserialize_deletion_vector_blob_invalid_length() -> None: + with pytest.raises(ValueError, match="Invalid bitmap data length"): + _deserialize_dv_blob(_dv_blob(_bitmap_payload())[:-1]) + + +def test_deserialize_deletion_vector_blob_invalid_magic() -> None: + bitmap_data = struct.pack("I", len(bitmap_data)) + bitmap_data + struct.pack(">I", zlib.crc32(bitmap_data) & 0xFFFFFFFF) + + with pytest.raises(ValueError, match="Invalid magic number"): + _deserialize_dv_blob(blob) + + +def test_deserialize_deletion_vector_blob_invalid_crc() -> None: + blob = bytearray(_dv_blob(_bitmap_payload())) + blob[-1] ^= 1 + + with pytest.raises(ValueError, match="Invalid CRC"): + _deserialize_dv_blob(bytes(blob)) + + +def test_deserialize_deletion_vector_blob_invalid_cardinality() -> None: + with pytest.raises(ValueError, match="Invalid cardinality"): + _deserialize_dv_blob(_dv_blob(_bitmap_payload()), record_count=4) + + +def _data_file( + content_offset: int | None = None, + content_size_in_bytes: int | None = None, + referenced_data_file: str | None = None, +) -> DataFile: + return DataFile.from_args( + _table_format_version=3, + content=DataFileContent.POSITION_DELETES, + file_path="deletes.puffin", + file_format=FileFormat.PUFFIN, + record_count=1, + content_offset=content_offset, + content_size_in_bytes=content_size_in_bytes, + referenced_data_file=referenced_data_file, + ) + + +def test_has_deletion_vector_content_reference() -> None: + assert not has_deletion_vector_content_reference(_data_file()) + assert has_deletion_vector_content_reference(_data_file(content_offset=0)) + assert has_deletion_vector_content_reference(_data_file(content_size_in_bytes=1)) + assert has_deletion_vector_content_reference(_data_file(referenced_data_file="data.parquet")) + + def test_map_empty() -> None: puffin = _open_file("64mapempty.bin") From e71c8d14162eb04320eea9c5bb55d0dea5bb9dbc Mon Sep 17 00:00:00 2001 From: KaiqiJinWow Date: Wed, 30 Sep 2026 16:49:34 +0000 Subject: [PATCH 2/6] Test external single and packed deletion vector fixtures --- tests/io/test_pyarrow.py | 42 ++++++++++++++++++ .../deletion_vector/v1/deletion-vector.bin | Bin 0 -> 47 bytes .../v1/packed-deletion-vectors.bin | Bin 0 -> 283 bytes tests/table/test_deletion_vector.py | 25 +++++++++++ 4 files changed, 67 insertions(+) create mode 100644 tests/table/deletion_vector/v1/deletion-vector.bin create mode 100644 tests/table/deletion_vector/v1/packed-deletion-vectors.bin diff --git a/tests/io/test_pyarrow.py b/tests/io/test_pyarrow.py index 460cfd4eaf..b10d1a8e70 100644 --- a/tests/io/test_pyarrow.py +++ b/tests/io/test_pyarrow.py @@ -1980,6 +1980,48 @@ def test_read_all_delete_files_keeps_multiple_dv_content_ranges_for_same_path(tm assert sorted(delete.to_pylist() for delete in deletes[referenced_data_file]) == [[1, 3], [5]] +def test_read_all_delete_files_from_external_packed_deletion_vectors() -> None: + """Read all DV ranges from an externally generated .bin file without losing entries to deduplication.""" + fixture_path = Path(__file__).parents[1] / "table" / "deletion_vector" / "v1" / "packed-deletion-vectors.bin" + # (content_offset, content_size_in_bytes, expected deleted row positions) + # Ranges copied from the fixture's V3 manifest; each DV references a different ten-row data file. + content_ranges = [ + (1, 46, [1, 4, 7]), + (47, 48, [0, 3, 6, 9]), + (95, 42, [0]), + (137, 50, [1, 3, 5, 7, 9]), + (187, 52, [0, 2, 4, 6, 8, 9]), + (239, 44, [0, 2]), + ] + tasks = [] + expected_deletes = {} + for index, (content_offset, content_size_in_bytes, expected_positions) in enumerate(content_ranges): + referenced_data_file = f"data-{index}.parquet" + dv = DataFile.from_args( + _table_format_version=3, + content=DataFileContent.POSITION_DELETES, + file_path=str(fixture_path), + file_format=FileFormat.PUFFIN, + record_count=len(expected_positions), + content_offset=content_offset, + content_size_in_bytes=content_size_in_bytes, + referenced_data_file=referenced_data_file, + ) + data_file = DataFile.from_args( + content=DataFileContent.DATA, + file_path=referenced_data_file, + file_format=FileFormat.PARQUET, + record_count=10, + file_size_in_bytes=100, + ) + tasks.append(FileScanTask(data_file=data_file, delete_files=[dv])) + expected_deletes[referenced_data_file] = [pa.chunked_array([expected_positions])] + + deletes = _read_all_delete_files(PyArrowFileIO(), tasks) + + assert deletes == expected_deletes + + def test_delete(deletes_file: str, request: pytest.FixtureRequest, table_schema_simple: Schema) -> None: # Determine file format from the file extension file_format = FileFormat.PARQUET if deletes_file.endswith(".parquet") else FileFormat.ORC diff --git a/tests/table/deletion_vector/v1/deletion-vector.bin b/tests/table/deletion_vector/v1/deletion-vector.bin new file mode 100644 index 0000000000000000000000000000000000000000..1366c998089300b3486b1b09eba04891b75013c2 GIT binary patch literal 47 pcmZQ%U|>+Yc-b+Yc-bc21BvK6RO3%5!Zs5_V$Sb;9#WZ3*$VINS34sIQ+P_vnUZf9fQ Y0P6YbGba|PM+LVYM!24nlMnU+0RAo?$p8QV literal 0 HcmV?d00001 diff --git a/tests/table/test_deletion_vector.py b/tests/table/test_deletion_vector.py index 3f4e8f7cfe..9fbf00d273 100644 --- a/tests/table/test_deletion_vector.py +++ b/tests/table/test_deletion_vector.py @@ -17,16 +17,19 @@ import struct import zlib from os import path +from pathlib import Path import pytest from pyroaring import BitMap +from pyiceberg.io.pyarrow import PyArrowFileIO from pyiceberg.manifest import DataFile, DataFileContent, FileFormat from pyiceberg.table.deletion_vector import ( _DV_BLOB_MAGIC_NUMBER, DeletionVector, _deserialize_dv_blob, has_deletion_vector_content_reference, + read_deletion_vectors, ) @@ -101,6 +104,28 @@ def test_has_deletion_vector_content_reference() -> None: assert has_deletion_vector_content_reference(_data_file(referenced_data_file="data.parquet")) +def test_read_external_deletion_vector_content_range() -> None: + """Read an externally generated Iceberg DV blob from a .bin content range.""" + fixture_path = Path(__file__).parent / "deletion_vector" / "v1" / "deletion-vector.bin" + # The V3 manifest points past the one-byte file prefix to the complete DV blob. + dv = DataFile.from_args( + _table_format_version=3, + content=DataFileContent.POSITION_DELETES, + file_path=str(fixture_path), + file_format=FileFormat.PUFFIN, + record_count=3, + content_offset=1, + content_size_in_bytes=46, + referenced_data_file="data.parquet", + ) + + deletion_vectors = read_deletion_vectors(PyArrowFileIO(), dv) + + assert len(deletion_vectors) == 1 + assert deletion_vectors[0].referenced_data_file == "data.parquet" + assert deletion_vectors[0].to_vector().to_pylist() == [1, 4, 5] + + def test_map_empty() -> None: puffin = _open_file("64mapempty.bin") From e8d6fe9fd2a20a9cac718ca39a8be5ef974e8ab1 Mon Sep 17 00:00:00 2001 From: KaiqiJinWow Date: Wed, 7 Oct 2026 20:47:29 +0000 Subject: [PATCH 3/6] Make delete file sets read-only and clarify DV validation --- pyiceberg/table/__init__.py | 4 ++-- pyiceberg/table/delete_file.py | 30 +++++++++---------------- pyiceberg/table/delete_file_index.py | 8 +++---- pyiceberg/table/deletion_vector.py | 6 ++++- tests/io/test_pyarrow.py | 32 +++++++++++++++++++++++++++ tests/table/test_delete_file_index.py | 29 ++++++++++++++++++++++++ 6 files changed, 82 insertions(+), 27 deletions(-) diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 2d01585514..a4836e8237 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -2267,13 +2267,13 @@ def from_rest_response( data_file = _rest_file_to_data_file(rest_task.data_file) - resolved_deletes = DeleteFileSet() + resolved_deletes: list[DataFile] = [] if rest_task.delete_file_references: for idx in rest_task.delete_file_references: delete_file = delete_files[idx] if isinstance(delete_file, RESTEqualityDeleteFile): raise NotImplementedError(f"PyIceberg does not yet support equality deletes: {delete_file.file_path}") - resolved_deletes.add(_rest_file_to_data_file(delete_file)) + resolved_deletes.append(_rest_file_to_data_file(delete_file)) return FileScanTask( data_file=data_file, diff --git a/pyiceberg/table/delete_file.py b/pyiceberg/table/delete_file.py index 681b0eef9f..af4f8f9ccb 100644 --- a/pyiceberg/table/delete_file.py +++ b/pyiceberg/table/delete_file.py @@ -16,9 +16,8 @@ # under the License. from __future__ import annotations -from collections.abc import Iterable, Iterator, MutableSet +from collections.abc import Iterable, Iterator, Set from dataclasses import dataclass -from typing import Any from pyiceberg.manifest import DataFile @@ -41,15 +40,15 @@ def from_file(cls, delete_file: DataFile) -> DeleteFileKey: ) -class DeleteFileSet(MutableSet[DataFile]): - """Set-like delete-file collection keyed by location and content range.""" +class DeleteFileSet(Set[DataFile]): + """Read-only delete-file collection keyed by location and content range.""" _files: dict[DeleteFileKey, DataFile] def __init__(self, delete_files: Iterable[DataFile] = ()) -> None: self._files = {} for delete_file in delete_files: - self.add(delete_file) + self._files.setdefault(DeleteFileKey.from_file(delete_file), delete_file) def __contains__(self, delete_file: object) -> bool: """Return whether the delete file is present.""" @@ -63,34 +62,25 @@ def __len__(self) -> int: """Return the number of delete files.""" return len(self._files) - def add(self, delete_file: DataFile) -> None: - self._files.setdefault(DeleteFileKey.from_file(delete_file), delete_file) - - def discard(self, delete_file: DataFile) -> None: - self._files.pop(DeleteFileKey.from_file(delete_file), None) - - def update(self, delete_files: Iterable[DataFile]) -> None: - for delete_file in delete_files: - self.add(delete_file) - def __repr__(self) -> str: """Return a string representation of the delete file set.""" return f"{type(self).__name__}({list(self)!r})" - def __eq__(self, other: Any) -> bool: + def __eq__(self, other: object) -> bool: """Compare delete file sets by delete file identity.""" if isinstance(other, DeleteFileSet): return self._files.keys() == other._files.keys() - if not isinstance(other, Iterable): + if not isinstance(other, Set): + return NotImplemented + + if len(self) != len(other): return False other_keys: set[DeleteFileKey] = set() - other_count = 0 for delete_file in other: if not isinstance(delete_file, DataFile): return False other_keys.add(DeleteFileKey.from_file(delete_file)) - other_count += 1 - return len(other_keys) == other_count and set(self._files) == other_keys + return self._files.keys() == other_keys diff --git a/pyiceberg/table/delete_file_index.py b/pyiceberg/table/delete_file_index.py index b5904a9491..beb073439e 100644 --- a/pyiceberg/table/delete_file_index.py +++ b/pyiceberg/table/delete_file_index.py @@ -130,7 +130,7 @@ def for_data_file(self, seq_num: int, data_file: DataFile, partition_key: Record if self.is_empty(): return DeleteFileSet() - deletes = DeleteFileSet() + deletes: list[DataFile] = [] spec_id = data_file.spec_id or 0 key = _partition_key(spec_id, partition_key) @@ -138,13 +138,13 @@ def for_data_file(self, seq_num: int, data_file: DataFile, partition_key: Record if partition_deletes: for delete_file in partition_deletes.filter_by_seq(seq_num): if _applies_to_data_file(delete_file, data_file): - deletes.add(delete_file) + deletes.append(delete_file) path_deletes = self._by_path.get(data_file.file_path) if path_deletes: - deletes.update(path_deletes.filter_by_seq(seq_num)) + deletes.extend(path_deletes.filter_by_seq(seq_num)) - return deletes + return DeleteFileSet(deletes) def referenced_delete_files(self) -> list[DataFile]: data_files: list[DataFile] = [] diff --git a/pyiceberg/table/deletion_vector.py b/pyiceberg/table/deletion_vector.py index 7094087404..3c33d7c737 100644 --- a/pyiceberg/table/deletion_vector.py +++ b/pyiceberg/table/deletion_vector.py @@ -143,7 +143,11 @@ def _validate_deletion_vector_content(dv: "DataFile") -> None: def has_deletion_vector_content_reference(dv: "DataFile") -> bool: - """Return whether a deletion vector is described by manifest content-range metadata.""" + """Return whether any manifest content-reference field is present. + + Partial references must be validated as range reads. Whole-Puffin reads + are only used when all content-reference fields are absent. + """ return dv.content_offset is not None or dv.content_size_in_bytes is not None or dv.referenced_data_file is not None diff --git a/tests/io/test_pyarrow.py b/tests/io/test_pyarrow.py index b10d1a8e70..34163043cd 100644 --- a/tests/io/test_pyarrow.py +++ b/tests/io/test_pyarrow.py @@ -1909,6 +1909,38 @@ def test_read_deletes_from_whole_puffin_file(tmp_path: Path) -> None: assert deletes == {referenced_data_file: pa.chunked_array([[1, 3, 5]])} +@pytest.mark.parametrize( + ("missing_field", "error"), + [ + ("content_offset", "content offset is missing"), + ("content_size_in_bytes", "content size is missing"), + ("referenced_data_file", "referenced data file is missing"), + ], +) +def test_read_deletes_rejects_partial_content_reference(tmp_path: Path, missing_field: str, error: str) -> None: + referenced_data_file = f"{tmp_path}/data.parquet" + delete_file_path = f"{tmp_path}/deletes.puffin" + puffin_payload = _deletion_vector_puffin_payload(referenced_data_file) + blob = PuffinFile(puffin_payload).footer.blobs[0] + + with open(delete_file_path, "wb") as f: + f.write(puffin_payload) + + dv = DataFile.from_args( + _table_format_version=3, + content=DataFileContent.POSITION_DELETES, + file_path=delete_file_path, + file_format=FileFormat.PUFFIN, + record_count=3, + content_offset=blob.offset if missing_field != "content_offset" else None, + content_size_in_bytes=blob.length if missing_field != "content_size_in_bytes" else None, + referenced_data_file=referenced_data_file if missing_field != "referenced_data_file" else None, + ) + + with pytest.raises(ValueError, match=error): + _read_deletes(PyArrowFileIO(), dv) + + def test_read_deletion_vector_blob_from_content_range(tmp_path: Path) -> None: referenced_data_file = f"{tmp_path}/data.parquet" dv_blob = _deletion_vector_blob(_deletion_vector_bitmap_payload()) diff --git a/tests/table/test_delete_file_index.py b/tests/table/test_delete_file_index.py index 6a02244363..614d94947f 100644 --- a/tests/table/test_delete_file_index.py +++ b/tests/table/test_delete_file_index.py @@ -14,6 +14,8 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +from collections.abc import Callable, Iterable, Set + import pytest from pyiceberg.manifest import DataFile, DataFileContent, FileFormat, ManifestEntry, ManifestEntryStatus @@ -190,6 +192,33 @@ def test_delete_file_set_uses_content_range_identity() -> None: assert len(DeleteFileSet([first_dv, second_dv])) == 2 +@pytest.mark.parametrize("collection_type", [DeleteFileSet, set, frozenset]) +@pytest.mark.parametrize(("content_offset", "content_size_in_bytes", "equal"), [(4, 10, True), (40, 10, False), (4, 12, False)]) +def test_delete_file_set_equality( + collection_type: Callable[[Iterable[DataFile]], Set[DataFile]], content_offset: int, content_size_in_bytes: int, equal: bool +) -> None: + dv = _create_deletion_vector(content_offset=4, content_size_in_bytes=10).data_file + other_dv = _create_deletion_vector(content_offset=content_offset, content_size_in_bytes=content_size_in_bytes).data_file + deletes = DeleteFileSet([dv]) + other = collection_type([other_dv]) + + assert (deletes == other) is equal + assert (other == deletes) is equal + assert (deletes != other) is not equal + + +def test_delete_file_set_does_not_compare_with_non_sets() -> None: + dv = _create_deletion_vector(content_offset=4, content_size_in_bytes=10).data_file + deletes = DeleteFileSet([dv]) + iterator = iter([dv]) + + assert deletes != [dv] + assert deletes != (dv,) + assert deletes != iterator + assert next(iterator) is dv + assert DeleteFileSet() != [] + + def test_dvs_with_same_file_path_and_different_content_ranges_are_not_deduped() -> None: index = DeleteFileIndex() shared_file_path = "s3://bucket/deletion-vectors.bin" From 5c05f03430322b3ec200faf7ccf7004a2d70e8c8 Mon Sep 17 00:00:00 2001 From: KaiqiJinWow Date: Thu, 8 Oct 2026 21:17:08 +0000 Subject: [PATCH 4/6] Keep DV range identity internal to scan aggregation Restore native scan task sets by routing deletion vectors to their referenced data files. Preserve REST range references and cover local and REST scans of packed DV fixtures. --- pyiceberg/io/pyarrow.py | 19 ++- pyiceberg/table/__init__.py | 39 ++++-- pyiceberg/table/delete_file.py | 86 ------------- pyiceberg/table/delete_file_index.py | 16 +-- tests/catalog/test_scan_planning_models.py | 75 +++++++++++ tests/io/test_pyarrow.py | 142 +++++++++++++++++---- tests/table/test_delete_file_index.py | 65 ++-------- 7 files changed, 258 insertions(+), 184 deletions(-) delete mode 100644 pyiceberg/table/delete_file.py diff --git a/pyiceberg/io/pyarrow.py b/pyiceberg/io/pyarrow.py index fd286270c2..2b437f25f3 100644 --- a/pyiceberg/io/pyarrow.py +++ b/pyiceberg/io/pyarrow.py @@ -147,7 +147,6 @@ visit_with_partner, ) from pyiceberg.table import DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE, TableProperties -from pyiceberg.table.delete_file import DeleteFileSet from pyiceberg.table.deletion_vector import read_deletion_vectors from pyiceberg.table.locations import load_location_provider from pyiceberg.table.metadata import TableMetadata @@ -1716,14 +1715,28 @@ def _task_to_record_batches( ) +@dataclass(frozen=True, slots=True) +class _DeleteFileKey: + """Identity of a delete file or a referenced DV range.""" + + file_path: str + content_offset: int | None + content_size_in_bytes: int | None + + def _read_all_delete_files(io: FileIO, tasks: Iterable[FileScanTask]) -> dict[str, list[ChunkedArray]]: deletes_per_file: dict[str, list[ChunkedArray]] = {} - unique_deletes = DeleteFileSet(itertools.chain.from_iterable(task.delete_files for task in tasks)) + # DataFile equality uses only the path; DVs from different tasks may share a file. + unique_deletes: dict[_DeleteFileKey, DataFile] = {} + for task in tasks: + for delete_file in task.delete_files: + key = _DeleteFileKey(delete_file.file_path, delete_file.content_offset, delete_file.content_size_in_bytes) + unique_deletes.setdefault(key, delete_file) if len(unique_deletes) > 0: executor = ExecutorFactory.get_or_create() deletes_per_files: Iterator[dict[str, ChunkedArray]] = executor.map( lambda args: _read_deletes(*args), - [(io, delete_file) for delete_file in unique_deletes], + [(io, delete_file) for delete_file in unique_deletes.values()], ) for delete in deletes_per_files: for file, arr in delete.items(): diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index a4836e8237..d4f47678bf 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -45,10 +45,17 @@ manifest_evaluator, ) from pyiceberg.io import FileIO, load_file_io -from pyiceberg.manifest import DataFile, DataFileContent, ManifestContent, ManifestEntry, ManifestEntryStatus, ManifestFile +from pyiceberg.manifest import ( + DataFile, + DataFileContent, + FileFormat, + ManifestContent, + ManifestEntry, + ManifestEntryStatus, + ManifestFile, +) from pyiceberg.partitioning import PARTITION_FIELD_ID_START, UNPARTITIONED_PARTITION_SPEC, PartitionKey, PartitionSpec from pyiceberg.schema import Schema -from pyiceberg.table.delete_file import DeleteFileSet from pyiceberg.table.delete_file_index import DeleteFileIndex from pyiceberg.table.inspect import InspectTable from pyiceberg.table.locations import LocationProvider, load_location_provider @@ -2233,17 +2240,17 @@ class FileScanTask(ScanTask): """Task representing a data file and its corresponding delete files.""" file: DataFile - delete_files: DeleteFileSet + delete_files: set[DataFile] residual: BooleanExpression def __init__( self, data_file: DataFile, - delete_files: Iterable[DataFile] | None = None, + delete_files: set[DataFile] | None = None, residual: BooleanExpression = ALWAYS_TRUE, ) -> None: self.file = data_file - self.delete_files = DeleteFileSet(delete_files if delete_files is not None else []) + self.delete_files = delete_files or set() self.residual = residual @staticmethod @@ -2267,13 +2274,13 @@ def from_rest_response( data_file = _rest_file_to_data_file(rest_task.data_file) - resolved_deletes: list[DataFile] = [] + resolved_deletes: set[DataFile] = set() if rest_task.delete_file_references: for idx in rest_task.delete_file_references: delete_file = delete_files[idx] if isinstance(delete_file, RESTEqualityDeleteFile): raise NotImplementedError(f"PyIceberg does not yet support equality deletes: {delete_file.file_path}") - resolved_deletes.append(_rest_file_to_data_file(delete_file)) + resolved_deletes.add(_rest_file_to_data_file(delete_file, default_referenced_data_file=data_file.file_path)) return FileScanTask( data_file=data_file, @@ -2282,9 +2289,9 @@ def from_rest_response( ) -def _rest_file_to_data_file(rest_file: RESTContentFile) -> DataFile: +def _rest_file_to_data_file(rest_file: RESTContentFile, default_referenced_data_file: str | None = None) -> DataFile: """Convert a REST content file to a manifest DataFile.""" - from pyiceberg.catalog.rest.scan_planning import RESTDataFile + from pyiceberg.catalog.rest.scan_planning import RESTDataFile, RESTPositionDeleteFile if isinstance(rest_file, RESTDataFile): column_sizes = rest_file.column_sizes.to_dict() if rest_file.column_sizes else None @@ -2297,6 +2304,17 @@ def _rest_file_to_data_file(rest_file: RESTContentFile) -> DataFile: null_value_counts = None nan_value_counts = None + referenced_data_file = None + if isinstance(rest_file, RESTPositionDeleteFile): + referenced_data_file = rest_file.referenced_data_file + # Infer the DV target from its task only for range reads, preserving whole-Puffin fallback. + if ( + referenced_data_file is None + and rest_file.file_format == FileFormat.PUFFIN + and (rest_file.content_offset is not None or rest_file.content_size_in_bytes is not None) + ): + referenced_data_file = default_referenced_data_file + data_file = DataFile.from_args( content=DataFileContent.from_rest_type(rest_file.content), file_path=rest_file.file_path, @@ -2310,6 +2328,9 @@ def _rest_file_to_data_file(rest_file: RESTContentFile) -> DataFile: nan_value_counts=nan_value_counts, split_offsets=rest_file.split_offsets, sort_order_id=rest_file.sort_order_id, + referenced_data_file=referenced_data_file, + content_offset=rest_file.content_offset if isinstance(rest_file, RESTPositionDeleteFile) else None, + content_size_in_bytes=rest_file.content_size_in_bytes if isinstance(rest_file, RESTPositionDeleteFile) else None, ) data_file.spec_id = rest_file.spec_id return data_file diff --git a/pyiceberg/table/delete_file.py b/pyiceberg/table/delete_file.py deleted file mode 100644 index af4f8f9ccb..0000000000 --- a/pyiceberg/table/delete_file.py +++ /dev/null @@ -1,86 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. -from __future__ import annotations - -from collections.abc import Iterable, Iterator, Set -from dataclasses import dataclass - -from pyiceberg.manifest import DataFile - - -@dataclass(frozen=True, slots=True) -class DeleteFileKey: - """Identity of a delete file, including its referenced content range.""" - - file_path: str - content_offset: int | None - content_size_in_bytes: int | None - - @classmethod - def from_file(cls, delete_file: DataFile) -> DeleteFileKey: - """Create a key from a delete file.""" - return cls( - file_path=delete_file.file_path, - content_offset=delete_file.content_offset, - content_size_in_bytes=delete_file.content_size_in_bytes, - ) - - -class DeleteFileSet(Set[DataFile]): - """Read-only delete-file collection keyed by location and content range.""" - - _files: dict[DeleteFileKey, DataFile] - - def __init__(self, delete_files: Iterable[DataFile] = ()) -> None: - self._files = {} - for delete_file in delete_files: - self._files.setdefault(DeleteFileKey.from_file(delete_file), delete_file) - - def __contains__(self, delete_file: object) -> bool: - """Return whether the delete file is present.""" - return isinstance(delete_file, DataFile) and DeleteFileKey.from_file(delete_file) in self._files - - def __iter__(self) -> Iterator[DataFile]: - """Return an iterator over delete files.""" - return iter(self._files.values()) - - def __len__(self) -> int: - """Return the number of delete files.""" - return len(self._files) - - def __repr__(self) -> str: - """Return a string representation of the delete file set.""" - return f"{type(self).__name__}({list(self)!r})" - - def __eq__(self, other: object) -> bool: - """Compare delete file sets by delete file identity.""" - if isinstance(other, DeleteFileSet): - return self._files.keys() == other._files.keys() - - if not isinstance(other, Set): - return NotImplemented - - if len(self) != len(other): - return False - - other_keys: set[DeleteFileKey] = set() - for delete_file in other: - if not isinstance(delete_file, DataFile): - return False - other_keys.add(DeleteFileKey.from_file(delete_file)) - - return self._files.keys() == other_keys diff --git a/pyiceberg/table/delete_file_index.py b/pyiceberg/table/delete_file_index.py index beb073439e..91365d6ea2 100644 --- a/pyiceberg/table/delete_file_index.py +++ b/pyiceberg/table/delete_file_index.py @@ -21,7 +21,6 @@ from pyiceberg.expressions import EqualTo from pyiceberg.expressions.visitors import _InclusiveMetricsEvaluator from pyiceberg.manifest import INITIAL_SEQUENCE_NUMBER, POSITIONAL_DELETE_SCHEMA, DataFile, ManifestEntry -from pyiceberg.table.delete_file import DeleteFileSet from pyiceberg.typedef import Record PATH_FIELD_ID = 2147483546 @@ -116,7 +115,7 @@ def is_empty(self) -> bool: def add_delete_file(self, manifest_entry: ManifestEntry, partition_key: Record | None = None) -> None: delete_file = manifest_entry.data_file seq = manifest_entry.sequence_number or INITIAL_SEQUENCE_NUMBER - target_path = _referenced_data_file_path(delete_file) + target_path = delete_file.referenced_data_file or _referenced_data_file_path(delete_file) if target_path: deletes = self._by_path.setdefault(target_path, PositionDeletes()) @@ -126,11 +125,12 @@ def add_delete_file(self, manifest_entry: ManifestEntry, partition_key: Record | deletes = self._by_partition.setdefault(key, PositionDeletes()) deletes.add(delete_file, seq) - def for_data_file(self, seq_num: int, data_file: DataFile, partition_key: Record | None = None) -> DeleteFileSet: + def for_data_file(self, seq_num: int, data_file: DataFile, partition_key: Record | None = None) -> set[DataFile]: if self.is_empty(): - return DeleteFileSet() + return set() - deletes: list[DataFile] = [] + # DVs are indexed by target path; a valid snapshot has at most one DV per data file. + deletes: set[DataFile] = set() spec_id = data_file.spec_id or 0 key = _partition_key(spec_id, partition_key) @@ -138,13 +138,13 @@ def for_data_file(self, seq_num: int, data_file: DataFile, partition_key: Record if partition_deletes: for delete_file in partition_deletes.filter_by_seq(seq_num): if _applies_to_data_file(delete_file, data_file): - deletes.append(delete_file) + deletes.add(delete_file) path_deletes = self._by_path.get(data_file.file_path) if path_deletes: - deletes.extend(path_deletes.filter_by_seq(seq_num)) + deletes.update(path_deletes.filter_by_seq(seq_num)) - return DeleteFileSet(deletes) + return deletes def referenced_delete_files(self) -> list[DataFile]: data_files: list[DataFile] = [] diff --git a/tests/catalog/test_scan_planning_models.py b/tests/catalog/test_scan_planning_models.py index caf571d322..f88e24e92f 100644 --- a/tests/catalog/test_scan_planning_models.py +++ b/tests/catalog/test_scan_planning_models.py @@ -40,6 +40,7 @@ ) from pyiceberg.expressions import AlwaysTrue, EqualTo, Reference from pyiceberg.manifest import FileFormat +from pyiceberg.table import FileScanTask TEST_URI = "https://iceberg-test-catalog/" @@ -526,6 +527,80 @@ def test_plan_scan_with_delete_files(rest_scan_catalog: RestCatalog, requests_mo assert len(tasks[0].delete_files) == 1 +@pytest.mark.parametrize("include_referenced_data_file", [True, False]) +def test_plan_scan_with_shared_dv_file( + rest_scan_catalog: RestCatalog, requests_mock: Mocker, include_referenced_data_file: bool +) -> None: + data_files = [_rest_data_file(file_path=f"s3://bucket/data-{index}.parquet") for index in range(2)] + delete_files = [ + { + **_rest_position_delete_file( + file_path="s3://bucket/deletes.bin", file_format="puffin", file_size_in_bytes=400, content_offset=offset + ), + **({"referenced-data-file": data_file["file-path"]} if include_referenced_data_file else {}), + } + for data_file, offset in zip(data_files, [0, 200], strict=True) + ] + requests_mock.post( + f"{TEST_URI}v1/namespaces/db/tables/tbl/plan", + json={ + "status": "completed", + "delete-files": delete_files, + "file-scan-tasks": [ + {"data-file": data_file, "delete-file-references": [index, index]} for index, data_file in enumerate(data_files) + ], + }, + ) + + tasks = rest_scan_catalog.plan_scan(("db", "tbl"), PlanTableScanRequest()) + + assert len(tasks) == 2 + assert all(len(task.delete_files) == 1 for task in tasks) + deletes = [dv for task in tasks for dv in task.delete_files] + assert len(deletes) == 2 + assert {(dv.referenced_data_file, dv.content_offset, dv.content_size_in_bytes) for dv in deletes} == { + ("s3://bucket/data-0.parquet", 0, 200), + ("s3://bucket/data-1.parquet", 200, 200), + } + + +@pytest.mark.parametrize("missing_field", ["content-offset", "content-size-in-bytes"]) +@pytest.mark.parametrize("include_referenced_data_file", [True, False]) +def test_rest_task_preserves_partial_dv_reference(missing_field: str, include_referenced_data_file: bool) -> None: + fields = { + **_rest_position_delete_file(file_format="puffin", content_offset=0), + **({"referenced-data-file": "s3://bucket/data.parquet"} if include_referenced_data_file else {}), + } + fields.pop(missing_field) + task = FileScanTask.from_rest_response( + RESTFileScanTask.model_validate( + {"data-file": _rest_data_file(file_path="s3://bucket/data.parquet"), "delete-file-references": [0]} + ), + [RESTPositionDeleteFile.model_validate(fields)], + ) + dv = next(iter(task.delete_files)) + + assert dv.referenced_data_file == "s3://bucket/data.parquet" + assert dv.content_offset == fields.get("content-offset") + assert dv.content_size_in_bytes == fields.get("content-size-in-bytes") + + +@pytest.mark.parametrize("file_format", ["puffin", "parquet"]) +def test_rest_task_preserves_whole_delete_file_reference(file_format: str) -> None: + fields = _rest_position_delete_file(file_format=file_format) + fields.pop("content-offset") + fields.pop("content-size-in-bytes") + task = FileScanTask.from_rest_response( + RESTFileScanTask.model_validate({"data-file": _rest_data_file(), "delete-file-references": [0]}), + [RESTPositionDeleteFile.model_validate(fields)], + ) + dv = next(iter(task.delete_files)) + + assert dv.referenced_data_file is None + assert dv.content_offset is None + assert dv.content_size_in_bytes is None + + def test_plan_scan_async_poll_completes( rest_scan_catalog: RestCatalog, requests_mock: Mocker, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/io/test_pyarrow.py b/tests/io/test_pyarrow.py index 34163043cd..ee2e0be2b4 100644 --- a/tests/io/test_pyarrow.py +++ b/tests/io/test_pyarrow.py @@ -27,7 +27,7 @@ from collections.abc import Iterator from datetime import date, datetime, timezone from pathlib import Path -from typing import Any +from typing import Any, Literal from unittest.mock import MagicMock, patch from uuid import uuid4 @@ -41,6 +41,7 @@ from pyarrow.fs import AwsDefaultS3RetryStrategy, FileType, LocalFileSystem, S3FileSystem from pyroaring import BitMap +from pyiceberg.catalog.rest.scan_planning import RESTDataFile, RESTDeleteFile, RESTFileScanTask, RESTPositionDeleteFile from pyiceberg.exceptions import ResolveError from pyiceberg.expressions import ( AlwaysFalse, @@ -92,16 +93,17 @@ schema_to_pyarrow, write_file, ) -from pyiceberg.manifest import DataFile, DataFileContent, FileFormat +from pyiceberg.manifest import DataFile, DataFileContent, FileFormat, ManifestEntry, ManifestEntryStatus from pyiceberg.partitioning import PartitionField, PartitionSpec from pyiceberg.schema import Schema, make_compatible_name, visit from pyiceberg.table import FileScanTask, TableProperties, WriteTask +from pyiceberg.table.delete_file_index import DeleteFileIndex from pyiceberg.table.deletion_vector import ( _DV_BLOB_MAGIC_NUMBER, PROPERTY_REFERENCED_DATA_FILE, deletion_vectors_from_puffin_file, ) -from pyiceberg.table.metadata import TableMetadataV2 +from pyiceberg.table.metadata import TableMetadataV2, TableMetadataV3 from pyiceberg.table.name_mapping import create_mapping_from_schema from pyiceberg.table.puffin import MAGIC_BYTES, PuffinFile from pyiceberg.transforms import HourTransform, IdentityTransform @@ -1968,7 +1970,7 @@ def test_read_deletion_vector_blob_from_content_range(tmp_path: Path) -> None: def test_read_all_delete_files_keeps_multiple_dv_content_ranges_for_same_path(tmp_path: Path) -> None: - referenced_data_file = f"{tmp_path}/data.parquet" + referenced_data_files = [f"{tmp_path}/data-{index}.parquet" for index in range(2)] first_dv_blob = _deletion_vector_blob(_deletion_vector_bitmap_payload([1, 3])) second_dv_blob = _deletion_vector_blob(_deletion_vector_bitmap_payload([5])) delete_file_path = f"{tmp_path}/deletes.bin" @@ -1982,7 +1984,7 @@ def test_read_all_delete_files_keeps_multiple_dv_content_ranges_for_same_path(tm file_path=delete_file_path, file_format=FileFormat.PUFFIN, record_count=2, - referenced_data_file=referenced_data_file, + referenced_data_file=referenced_data_files[0], content_offset=0, content_size_in_bytes=len(first_dv_blob), ) @@ -1992,28 +1994,41 @@ def test_read_all_delete_files_keeps_multiple_dv_content_ranges_for_same_path(tm file_path=delete_file_path, file_format=FileFormat.PUFFIN, record_count=1, - referenced_data_file=referenced_data_file, + referenced_data_file=referenced_data_files[1], content_offset=len(first_dv_blob), content_size_in_bytes=len(second_dv_blob), ) - data_file = DataFile.from_args( - content=DataFileContent.DATA, - file_path=referenced_data_file, - file_format=FileFormat.PARQUET, - record_count=10, - file_size_in_bytes=100, - ) + data_files = [ + DataFile.from_args( + content=DataFileContent.DATA, + file_path=referenced_data_file, + file_format=FileFormat.PARQUET, + record_count=10, + file_size_in_bytes=100, + ) + for referenced_data_file in referenced_data_files + ] deletes = _read_all_delete_files( PyArrowFileIO(), - [FileScanTask(data_file=data_file, delete_files=[first_dv, second_dv])], + [ + FileScanTask(data_file=data_file, delete_files={dv}) + for data_file, dv in zip(data_files, [first_dv, second_dv], strict=True) + ], ) - assert sorted(delete.to_pylist() for delete in deletes[referenced_data_file]) == [[1, 3], [5]] + assert deletes == { + referenced_data_files[0]: [pa.chunked_array([[1, 3]])], + referenced_data_files[1]: [pa.chunked_array([[5]])], + } -def test_read_all_delete_files_from_external_packed_deletion_vectors() -> None: - """Read all DV ranges from an externally generated .bin file without losing entries to deduplication.""" +@pytest.mark.parametrize("selected_indices", [(0, 1, 2, 3, 4, 5), (0, 3, 5)]) +@pytest.mark.parametrize("planning_mode", ["local", "rest"]) +def test_read_all_delete_files_from_external_packed_deletion_vectors( + tmp_path: Path, selected_indices: tuple[int, ...], planning_mode: Literal["local", "rest"] +) -> None: + """Apply selected DV ranges from an externally generated .bin file to their data files.""" fixture_path = Path(__file__).parents[1] / "table" / "deletion_vector" / "v1" / "packed-deletion-vectors.bin" # (content_offset, content_size_in_bytes, expected deleted row positions) # Ranges copied from the fixture's V3 manifest; each DV references a different ten-row data file. @@ -2025,34 +2040,113 @@ def test_read_all_delete_files_from_external_packed_deletion_vectors() -> None: (187, 52, [0, 2, 4, 6, 8, 9]), (239, 44, [0, 2]), ] - tasks = [] - expected_deletes = {} + schema = Schema(NestedField(1, "id", LongType(), required=True)) + data_files: list[DataFile] = [] + delete_files: list[DataFile] = [] + expected_deletes: dict[str, list[pa.ChunkedArray]] = {} + expected_rows: list[int] = [] for index, (content_offset, content_size_in_bytes, expected_positions) in enumerate(content_ranges): - referenced_data_file = f"data-{index}.parquet" + data_path = tmp_path / f"data-{index}.parquet" + referenced_data_file = str(data_path) + rows = list(range(index * 10, (index + 1) * 10)) + pq.write_table(pa.table({"id": rows}, schema=schema_to_pyarrow(schema)), data_path) dv = DataFile.from_args( _table_format_version=3, content=DataFileContent.POSITION_DELETES, file_path=str(fixture_path), file_format=FileFormat.PUFFIN, + partition=Record(), record_count=len(expected_positions), + file_size_in_bytes=fixture_path.stat().st_size, content_offset=content_offset, content_size_in_bytes=content_size_in_bytes, referenced_data_file=referenced_data_file, ) + dv.spec_id = 0 data_file = DataFile.from_args( content=DataFileContent.DATA, file_path=referenced_data_file, file_format=FileFormat.PARQUET, + partition=Record(), record_count=10, - file_size_in_bytes=100, + file_size_in_bytes=data_path.stat().st_size, + first_row_id=index * 10, ) - tasks.append(FileScanTask(data_file=data_file, delete_files=[dv])) - expected_deletes[referenced_data_file] = [pa.chunked_array([expected_positions])] + data_file.spec_id = 0 + data_files.append(data_file) + delete_files.append(dv) + if index in selected_indices: + expected_deletes[referenced_data_file] = [pa.chunked_array([expected_positions])] + expected_rows.extend(row for position, row in enumerate(rows) if position not in expected_positions) + + if planning_mode == "local": + delete_index = DeleteFileIndex() + for dv in delete_files: + delete_index.add_delete_file( + ManifestEntry.from_args(status=ManifestEntryStatus.ADDED, sequence_number=2, data_file=dv) + ) + tasks = [ + FileScanTask(data_file=data_files[index], delete_files=delete_index.for_data_file(1, data_files[index])) + for index in selected_indices + ] + else: + rest_deletes: list[RESTDeleteFile] = [ + RESTPositionDeleteFile( + spec_id=0, + file_path=dv.file_path, + file_format=dv.file_format, + file_size_in_bytes=dv.file_size_in_bytes, + record_count=dv.record_count, + content_offset=dv.content_offset, + content_size_in_bytes=dv.content_size_in_bytes, + ) + for dv in delete_files + ] + tasks = [ + FileScanTask.from_rest_response( + RESTFileScanTask( + data_file=RESTDataFile( + spec_id=0, + file_path=data_files[index].file_path, + file_format=FileFormat.PARQUET, + file_size_in_bytes=data_files[index].file_size_in_bytes, + record_count=10, + ), + delete_file_references=[index, index], + ), + rest_deletes, + ) + for index in selected_indices + ] + + for task, index in zip(tasks, selected_indices, strict=True): + assert isinstance(task.delete_files, set) + assert len(task.delete_files) == 1 + dv = next(iter(task.delete_files)) + assert dv.referenced_data_file == task.file.file_path + assert (dv.content_offset, dv.content_size_in_bytes) == content_ranges[index][:2] - deletes = _read_all_delete_files(PyArrowFileIO(), tasks) + # Repeated tasks must not cause a DV to be loaded more than once. + deletes = _read_all_delete_files(PyArrowFileIO(), iter(tasks + tasks[::-1])) assert deletes == expected_deletes + result = ArrowScan( + table_metadata=TableMetadataV3( + location=str(tmp_path), + last_column_id=1, + schemas=[schema], + current_schema_id=schema.schema_id, + partition_specs=[PartitionSpec()], + next_row_id=60, + ), + io=PyArrowFileIO(), + projected_schema=schema, + row_filter=AlwaysTrue(), + ).to_table(tasks) + + assert sorted(result.column("id").to_pylist()) == expected_rows + def test_delete(deletes_file: str, request: pytest.FixtureRequest, table_schema_simple: Schema) -> None: # Determine file format from the file extension diff --git a/tests/table/test_delete_file_index.py b/tests/table/test_delete_file_index.py index 614d94947f..6385e091cb 100644 --- a/tests/table/test_delete_file_index.py +++ b/tests/table/test_delete_file_index.py @@ -14,12 +14,9 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -from collections.abc import Callable, Iterable, Set - import pytest from pyiceberg.manifest import DataFile, DataFileContent, FileFormat, ManifestEntry, ManifestEntryStatus -from pyiceberg.table.delete_file import DeleteFileSet from pyiceberg.table.delete_file_index import PATH_FIELD_ID, DeleteFileIndex, PositionDeletes from pyiceberg.typedef import Record @@ -83,8 +80,6 @@ def _create_deletion_vector( partition=Record(), record_count=10, file_size_in_bytes=100, - lower_bounds={PATH_FIELD_ID: file_path.encode()}, - upper_bounds={PATH_FIELD_ID: file_path.encode()}, referenced_data_file=file_path, content_offset=content_offset, content_size_in_bytes=content_size_in_bytes, @@ -173,59 +168,14 @@ def test_dvs_treated_as_position_deletes() -> None: assert all(d.content == DataFileContent.POSITION_DELETES for d in result) -def test_delete_file_set_uses_content_range_identity() -> None: - shared_file_path = "s3://bucket/deletion-vectors.bin" - first_dv = _create_deletion_vector( - sequence_number=2, - delete_file_path=shared_file_path, - content_offset=4, - content_size_in_bytes=10, - ).data_file - second_dv = _create_deletion_vector( - sequence_number=3, - delete_file_path=shared_file_path, - content_offset=40, - content_size_in_bytes=12, - ).data_file - - assert first_dv == second_dv - assert len(DeleteFileSet([first_dv, second_dv])) == 2 - - -@pytest.mark.parametrize("collection_type", [DeleteFileSet, set, frozenset]) -@pytest.mark.parametrize(("content_offset", "content_size_in_bytes", "equal"), [(4, 10, True), (40, 10, False), (4, 12, False)]) -def test_delete_file_set_equality( - collection_type: Callable[[Iterable[DataFile]], Set[DataFile]], content_offset: int, content_size_in_bytes: int, equal: bool -) -> None: - dv = _create_deletion_vector(content_offset=4, content_size_in_bytes=10).data_file - other_dv = _create_deletion_vector(content_offset=content_offset, content_size_in_bytes=content_size_in_bytes).data_file - deletes = DeleteFileSet([dv]) - other = collection_type([other_dv]) - - assert (deletes == other) is equal - assert (other == deletes) is equal - assert (deletes != other) is not equal - - -def test_delete_file_set_does_not_compare_with_non_sets() -> None: - dv = _create_deletion_vector(content_offset=4, content_size_in_bytes=10).data_file - deletes = DeleteFileSet([dv]) - iterator = iter([dv]) - - assert deletes != [dv] - assert deletes != (dv,) - assert deletes != iterator - assert next(iterator) is dv - assert DeleteFileSet() != [] - - -def test_dvs_with_same_file_path_and_different_content_ranges_are_not_deduped() -> None: +def test_dvs_with_same_file_path_are_matched_by_referenced_data_file() -> None: index = DeleteFileIndex() shared_file_path = "s3://bucket/deletion-vectors.bin" index.add_delete_file( _create_deletion_vector( sequence_number=2, + file_path="s3://bucket/a.parquet", delete_file_path=shared_file_path, content_offset=4, content_size_in_bytes=10, @@ -234,14 +184,21 @@ def test_dvs_with_same_file_path_and_different_content_ranges_are_not_deduped() index.add_delete_file( _create_deletion_vector( sequence_number=3, + file_path="s3://bucket/b.parquet", delete_file_path=shared_file_path, content_offset=40, content_size_in_bytes=12, ) ) - data_file = _create_data_file() - result = index.for_data_file(1, data_file) + first_deletes = index.for_data_file(1, _create_data_file("s3://bucket/a.parquet")) + second_deletes = index.for_data_file(1, _create_data_file("s3://bucket/b.parquet")) + assert len(first_deletes) == len(second_deletes) == 1 + assert {dv.referenced_data_file for dv in first_deletes} == {"s3://bucket/a.parquet"} + assert {dv.referenced_data_file for dv in second_deletes} == {"s3://bucket/b.parquet"} + assert not index.for_data_file(1, _create_data_file("s3://bucket/unrelated.parquet")) + assert isinstance(first_deletes, set) + result = [*first_deletes, *second_deletes] assert len(result) == 2 assert {(dv.file_path, dv.content_offset, dv.content_size_in_bytes) for dv in result} == { From 8b4ee5c3f68c215fd9d7133651c48a6a36f8fb54 Mon Sep 17 00:00:00 2001 From: KaiqiJinWow Date: Mon, 12 Oct 2026 17:17:26 +0000 Subject: [PATCH 5/6] Preserve validated DV field types and keep helper internal --- pyiceberg/table/deletion_vector.py | 19 ++++++++----------- tests/table/test_deletion_vector.py | 10 +++++----- 2 files changed, 13 insertions(+), 16 deletions(-) diff --git a/pyiceberg/table/deletion_vector.py b/pyiceberg/table/deletion_vector.py index 3c33d7c737..9c56af24a2 100644 --- a/pyiceberg/table/deletion_vector.py +++ b/pyiceberg/table/deletion_vector.py @@ -17,7 +17,7 @@ import math import struct import zlib -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING from pyroaring import BitMap, FrozenBitMap @@ -32,7 +32,6 @@ EMPTY_BITMAP = FrozenBitMap() MAX_JAVA_SIGNED = int(math.pow(2, 31)) - 1 PROPERTY_REFERENCED_DATA_FILE = "referenced-data-file" -_MAX_DELETION_VECTOR_CONTENT_SIZE = 2**31 - 1 _DV_BLOB_LENGTH = struct.Struct(">I") _DV_BLOB_MAGIC = struct.Struct("I") @@ -123,7 +122,7 @@ def _deserialize_dv_blob(blob: bytes, record_count: int | None = None) -> list[B return bitmaps -def _validate_deletion_vector_content(dv: "DataFile") -> None: +def _validate_deletion_vector_content(dv: "DataFile") -> tuple[int, int, str]: content_offset = dv.content_offset content_size_in_bytes = dv.content_size_in_bytes referenced_data_file = dv.referenced_data_file @@ -136,13 +135,15 @@ def _validate_deletion_vector_content(dv: "DataFile") -> None: 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: + if content_size_in_bytes > MAX_JAVA_SIGNED: 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: {dv.file_path}") + return content_offset, content_size_in_bytes, referenced_data_file -def has_deletion_vector_content_reference(dv: "DataFile") -> bool: + +def _has_deletion_vector_content_reference(dv: "DataFile") -> bool: """Return whether any manifest content-reference field is present. Partial references must be validated as range reads. Whole-Puffin reads @@ -152,11 +153,7 @@ def has_deletion_vector_content_reference(dv: "DataFile") -> bool: def _read_deletion_vector(io: "FileIO", dv: "DataFile") -> DeletionVector: - _validate_deletion_vector_content(dv) - - content_offset = cast(int, dv.content_offset) - content_size_in_bytes = cast(int, dv.content_size_in_bytes) - referenced_data_file = cast(str, dv.referenced_data_file) + content_offset, content_size_in_bytes, referenced_data_file = _validate_deletion_vector_content(dv) with io.new_input(dv.file_path).open() as fi: fi.seek(content_offset) @@ -173,7 +170,7 @@ def _read_deletion_vector(io: "FileIO", dv: "DataFile") -> DeletionVector: def read_deletion_vectors(io: "FileIO", dv: "DataFile") -> list[DeletionVector]: """Read deletion vectors from a delete file or its manifest content range.""" - if has_deletion_vector_content_reference(dv): + if _has_deletion_vector_content_reference(dv): return [_read_deletion_vector(io, dv)] with io.new_input(dv.file_path).open() as fi: diff --git a/tests/table/test_deletion_vector.py b/tests/table/test_deletion_vector.py index 9fbf00d273..4b38ea8627 100644 --- a/tests/table/test_deletion_vector.py +++ b/tests/table/test_deletion_vector.py @@ -28,7 +28,7 @@ _DV_BLOB_MAGIC_NUMBER, DeletionVector, _deserialize_dv_blob, - has_deletion_vector_content_reference, + _has_deletion_vector_content_reference, read_deletion_vectors, ) @@ -98,10 +98,10 @@ def _data_file( def test_has_deletion_vector_content_reference() -> None: - assert not has_deletion_vector_content_reference(_data_file()) - assert has_deletion_vector_content_reference(_data_file(content_offset=0)) - assert has_deletion_vector_content_reference(_data_file(content_size_in_bytes=1)) - assert has_deletion_vector_content_reference(_data_file(referenced_data_file="data.parquet")) + assert not _has_deletion_vector_content_reference(_data_file()) + assert _has_deletion_vector_content_reference(_data_file(content_offset=0)) + assert _has_deletion_vector_content_reference(_data_file(content_size_in_bytes=1)) + assert _has_deletion_vector_content_reference(_data_file(referenced_data_file="data.parquet")) def test_read_external_deletion_vector_content_range() -> None: From 5fa7e3bdb605ac0505bf0bce7417a570a8045506 Mon Sep 17 00:00:00 2001 From: KaiqiJinWow Date: Tue, 13 Oct 2026 23:29:58 +0000 Subject: [PATCH 6/6] Retrigger CI