sungwy commented on code in PR #3474:
URL: https://github.com/apache/iceberg-python/pull/3474#discussion_r3484237330
##########
pyiceberg/table/puffin.py:
##########
@@ -75,3 +80,73 @@ def to_vector(self) -> dict[str, "pa.ChunkedArray"]:
from pyiceberg.table.deletion_vector import
deletion_vectors_from_puffin_file # local import avoids the cycle
return {dv.referenced_data_file: dv.to_vector() for dv in
deletion_vectors_from_puffin_file(self)}
+
+
+@dataclass(frozen=True)
+class PuffinBlob:
+ """A blob to write into a Puffin file: its metadata and serialized
payload."""
+
+ metadata: PuffinBlobMetadata
+ payload: bytes
+
+
+class PuffinWriter:
+ """Assembles a Puffin file from blobs and writes it to an output file.
+
+ This writer is format-level and blob-agnostic: callers supply
already-serialized blobs
+ (for example via DeletionVector.to_blob()). Use it as a context manager;
the file is
+ written on exit, after which its size is available via len(output_file).
+ """
+
+ closed: bool
+ _output_file: OutputFile
+ _blobs: list[PuffinBlob]
+ _created_by: str
+
+ def __init__(self, output_file: OutputFile, created_by: str | None = None)
-> None:
+ self.closed = False
+ self._output_file = output_file
+ self._blobs = []
+ self._created_by = created_by if created_by is not None else
f"PyIceberg version {__version__}"
+
+ def __enter__(self) -> "PuffinWriter":
+ """Open the writer."""
+ return self
+
+ def __exit__(
+ self,
+ exc_type: type[BaseException] | None,
+ exc_value: BaseException | None,
+ traceback: TracebackType | None,
+ ) -> None:
+ """Assemble the Puffin file and write it to the output file."""
+ self.closed = True
Review Comment:
nit (non-blocking): since orphaned files get reaped by the maintenance jobs
anyway, this is fine as-is, but `__exit__` will still assemble and write a
complete file even when the `with` body raised, which does unnecessary I/O on
the way out. Maybe short-circuit on failure so we don't write a half-populated
file? Something like:
```suggestion
self.closed = True
if exc_type is not None:
return
```
WDYT? 🙂
##########
pyiceberg/table/deletion_vector.py:
##########
@@ -37,6 +44,21 @@ def __init__(self, referenced_data_file: str, bitmaps:
list[BitMap]) -> None:
self.referenced_data_file = referenced_data_file
self._bitmaps = bitmaps
+ @classmethod
+ def from_positions(cls, referenced_data_file: str, positions:
Iterable[int]) -> "DeletionVector":
+ bitmaps_by_key: dict[int, BitMap] = {}
+ for position in positions:
+ if position < 0:
+ raise ValueError(f"Invalid position: {position}, positions
must be non-negative")
Review Comment:
nit: `from_positions` guards `position < 0` but not the upper end, so a
wildly out-of-range position blows up on the `range(max(...) + 1)` gap-fill
(OOM) instead of erroring cleanly. Java's `validatePosition` caps it at
`MAX_POSITION`. Do we want to mirror that?
```suggestion
if position < 0 or position > MAX_POSITION:
raise ValueError(f"Invalid position: {position}, must be
between 0 and {MAX_POSITION}")
```
with `MAX_POSITION = ((MAX_JAVA_SIGNED - 1) << 32) | 0x80000000` (= Java's
`toPosition(Integer.MAX_VALUE - 1, Integer.MIN_VALUE)`).
##########
pyiceberg/table/deletion_vector.py:
##########
@@ -67,6 +89,21 @@ def _deserialize_bitmap(pl: bytes) -> list[BitMap]:
return bitmaps
+ @staticmethod
+ def _serialize_bitmap(bitmaps: list[BitMap]) -> bytes:
+ # Counterpart of _deserialize_bitmap: number of bitmaps (8 bytes,
little-endian), then for each
+ # non-empty bitmap in ascending key order its key (4 bytes,
little-endian) and serialized payload.
+ non_empty = [(key, bitmap) for key, bitmap in enumerate(bitmaps) if
len(bitmap) > 0]
+
+ with io.BytesIO() as out:
+ out.write(len(non_empty).to_bytes(8, "little"))
+ for key, bitmap in non_empty:
+ if key > MAX_JAVA_SIGNED:
+ raise ValueError(f"Key {key} is too large, max
{MAX_JAVA_SIGNED} to maintain compatibility with Java impl")
+ out.write(key.to_bytes(4, "little"))
+ out.write(bitmap.serialize())
Review Comment:
One gap I see from Java: Java's `BitmapPositionDeleteIndex` calls
`runLengthEncode()` on the bitmap before serializing and we don't. For
run-heavy deletes (e.g. dropping a contiguous range) that's a big difference —
I measured a contiguous 100k-position DV at ~16KB vs ~37 bytes with RLE, and
the run-optimized bytes still round-trip through `_deserialize_bitmap`. Could
we match Java and run-optimize first?
```suggestion
bitmap.run_optimize()
out.write(bitmap.serialize())
```
--
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]
