abnobdoss commented on code in PR #3624:
URL: https://github.com/apache/iceberg-python/pull/3624#discussion_r3610944866
##########
pyiceberg/manifest.py:
##########
Review Comment:
Per the comment on `_wrap_data_file`: we likely need to go through all the
uses of `DEFAULT_READ_VERSION`, including this one. Reading a v3 manifest list
here drops `first_row_id`, which is what makes a round trip fail.
`fetch_manifest_entry` has the same issue for the new data file fields.
##########
pyiceberg/manifest.py:
##########
@@ -1419,19 +1508,101 @@ def prepare_manifest(self, manifest_file:
ManifestFile) -> ManifestFile:
return wrapped_manifest_file
+class ManifestListWriterV3(ManifestListWriterV2):
+ """Writes V3 manifest lists, assigning `first_row_id` to data manifests.
+
+ Follows the spec's First Row ID Assignment: existing `first_row_id` values
are
+ preserved, delete manifests are never assigned one, and data manifests
without a
+ `first_row_id` are assigned a running value starting at the snapshot's
+ `first-row-id`, advanced by each assigned manifest's existing and added
row counts.
+ """
+
+ _next_row_id: int
+
+ def __init__(
+ self,
+ output_file: OutputFile,
+ snapshot_id: int,
+ parent_snapshot_id: int | None,
+ sequence_number: int,
+ compression: AvroCompressionCodec,
+ first_row_id: int,
+ ):
+ super().__init__(output_file, snapshot_id, parent_snapshot_id,
sequence_number, compression)
+ self._format_version = 3
+ self._meta = {
+ **self._meta,
+ "first-row-id": str(first_row_id),
+ "format-version": "3",
+ }
+ self._next_row_id = first_row_id
+
+ def __enter__(self) -> ManifestListWriter:
+ """Open the writer for writing, using the V3 record layout so
`first_row_id` is written."""
+ self._writer = AvroOutputFile[ManifestFile](
+ output_file=self._output_file,
+ record_schema=MANIFEST_LIST_FILE_SCHEMAS[3],
+ file_schema=MANIFEST_LIST_FILE_SCHEMAS[3],
+ schema_name="manifest_file",
+ metadata=self._meta,
+ )
+ self._writer.__enter__()
+ return self
+
+ @property
+ def next_row_id(self) -> int:
+ """The row ID after the last assigned one; the table's `next-row-id`
after this snapshot."""
+ return self._next_row_id
+
+ def _wrap(self, manifest_file: ManifestFile) -> ManifestFile:
+ """Rebind a manifest file to the V3 record layout.
+
+ Records not created with an explicit layout are bound to
+ MANIFEST_LIST_FILE_SCHEMAS[DEFAULT_READ_VERSION] (the from_args
default),
+ so that is the layout to zip the positional data against.
+ """
+ if len(manifest_file._data) >=
len(MANIFEST_LIST_FILE_SCHEMAS[3].fields):
+ return copy(manifest_file)
Review Comment:
`copy(manifest_file)` doesn't seem to work in its current form: the copy and
the original share the same underlying `_data` list, so writing to one writes
to both.
##########
pyiceberg/manifest.py:
##########
@@ -1284,18 +1295,96 @@ def prepare_entry(self, entry: ManifestEntry) ->
ManifestEntry:
return entry
+class ManifestWriterV3(ManifestWriterV2):
+ """Writes V3 manifest files.
+
+ The writer inherits the V2 sequence-number semantics; the V3 manifest entry
+ schema additionally carries `first_row_id`, `referenced_data_file`,
+ `content_offset` and `content_size_in_bytes` on the data file struct.
+
+ An optional `first_row_id` can be provided when rewriting a manifest whose
+ `first_row_id` is already known; it is carried into the produced manifest
file
+ so the manifest list writer preserves it instead of assigning a new one.
For
+ new manifests it is None and assigned when writing the manifest list.
+ """
+
+ _first_row_id: int | None
+
+ def __init__(
+ self,
+ spec: PartitionSpec,
+ schema: Schema,
+ output_file: OutputFile,
+ snapshot_id: int,
+ avro_compression: AvroCompressionCodec,
+ first_row_id: int | None = None,
+ ):
+ super().__init__(spec, schema, output_file, snapshot_id,
avro_compression)
+ self._first_row_id = first_row_id
+
+ @property
+ def version(self) -> TableVersion:
+ return 3
+
+ def to_manifest_file(self) -> ManifestFile:
+ """Return the manifest file, bound to the V3 layout and carrying
`first_row_id`."""
+ manifest_file = super().to_manifest_file()
+ args = {
+ field.name: value
+ for field, value in
zip(MANIFEST_LIST_FILE_SCHEMAS[DEFAULT_READ_VERSION].fields,
manifest_file._data, strict=True)
+ }
+ return ManifestFile.from_args(_table_format_version=3,
first_row_id=self._first_row_id, **args)
+
+ def new_writer(self) -> AvroOutputFile[ManifestEntry]:
+ # Use the V3 record layout so the V3-only data file fields are written
+ return AvroOutputFile[ManifestEntry](
+ output_file=self._output_file,
+ file_schema=self._with_partition(3),
+ record_schema=self._with_partition(3),
+ schema_name="manifest_entry",
+ metadata=self._meta,
+ )
+
+ def _wrap_data_file(self, data_file: DataFile) -> DataFile:
+ """Rebind a data file to the V3 record layout."""
+ if len(data_file._data) >= len(DATA_FILE_TYPE[3].fields):
+ return data_file
+ args = {
+ field.name: value for field, value in
zip(DATA_FILE_TYPE[DEFAULT_READ_VERSION].fields, data_file._data, strict=True)
Review Comment:
Should this be `DEFAULT_READ_VERSION`? It assumes any record that isn't v3
is v2, but `DATA_FILE_TYPE` has 15 fields for v1, 16 for v2 and 20 for v3, so a
v1-bound `DataFile` would fail the strict zip with a confusing error.
--
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]
