qzyu999 commented on code in PR #3320:
URL: https://github.com/apache/iceberg-python/pull/3320#discussion_r3449830133
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -353,11 +396,116 @@ def new_manifest_output(self) -> OutputFile:
location_provider = self._transaction._table.location_provider()
file_name =
_new_manifest_file_name(num=next(self._manifest_num_counter),
commit_uuid=self.commit_uuid)
file_path = location_provider.new_metadata_location(file_name)
+ self._written_manifests.append(file_path)
return self._io.new_output(file_path)
def fetch_manifest_entry(self, manifest: ManifestFile, discard_deleted:
bool = True) -> list[ManifestEntry]:
return manifest.fetch_manifest_entry(io=self._io,
discard_deleted=discard_deleted)
+ def commit(self) -> None:
+ self._transaction._register_snapshot_producer(self)
+ self._transaction._apply(*self._commit())
Review Comment:
**Nit**: `commit()` duplicates the base class implementation rather than
delegating via `super()`.
Consider:
```python
def commit(self) -> None:
self._transaction._register_snapshot_producer(self)
super().commit()
```
This preserves the override's additional behavior (producer registration)
while keeping the base class as the single source of truth for the commit
mechanics. If `UpdateTableMetadata.commit()` ever gains additional logic
(logging, hooks, etc.), this override will pick it up automatically rather than
silently diverging.
##########
pyiceberg/table/__init__.py:
##########
Review Comment:
**Nit (1/3)** - Proper typing for `_snapshot_producers`
`_snapshot_producers` and `_register_snapshot_producer` (introduced in this
PR) currently use `Any` as their type. This means `mypy` and IDEs treat
producer objects as completely untyped, no error if you call a nonexistent
method, no autocompletion, no verification that the right object type is being
stored.
We can't do a normal from `pyiceberg.table.update.snapshot import
_SnapshotProducer` at the top of the file because that module already imports
from `pyiceberg.table`, and it would create a circular import at runtime.
However, there's already a `TYPE_CHECKING` block here for exactly this
situation. Imports inside this block are only seen by type checkers (`mypy`,
`Pyright`, IDEs) so they're completely skipped at runtime, therefore no
circular import occurs.
Add at the end of this block:
```python
from pyiceberg.table.update.snapshot import _SnapshotProducer
```
This makes `_SnapshotProducer` available as a type annotation (for 2/3 and
3/3 below) without affecting runtime behavior.
##########
pyiceberg/table/__init__.py:
##########
@@ -265,6 +285,10 @@ def _stage(
return self
+ def _register_snapshot_producer(self, producer: Any) -> None:
Review Comment:
**Nit (3/3)**: Same reasoning, type the parameter so callers passing the
wrong type are caught:
```python
def _register_snapshot_producer(self, producer: _SnapshotProducer[Any]) ->
None:
```
This completes the type chain: `_SnapshotProducer.commit()` calls
`self._transaction._register_snapshot_producer(self)`, with this annotation,
`mypy` confirms that `self` (a `_SnapshotProducer)` satisfies the parameter
type. If someone accidentally tried to register a non-producer object, it would
be flagged immediately.
##########
pyiceberg/table/__init__.py:
##########
@@ -223,6 +242,7 @@ def __init__(self, table: Table, autocommit: bool = False):
self._autocommit = autocommit
self._updates = ()
self._requirements = ()
+ self._snapshot_producers: list[Any] = []
Review Comment:
**Nit (2/3)**: With the `TYPE_CHECKING` import from (1/3), this can be
properly annotated:
```python
self._snapshot_producers: list[_SnapshotProducer[Any]] = []
```
This matters since the retry loop in `_rebuild_snapshot_updates()` calls
`producer._refresh_for_retry()`, `producer._validate_concurrency()`,
`producer._clean_all_uncommitted()` etc. on items from this list. With
`list[Any]`, `mypy` treats those calls as untyped, and it won't catch a
misspelled method name, a wrong argument, or a missing attribute. With
`list[_SnapshotProducer[Any]]`, all those calls are statically verified against
the actual class definition.
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -91,19 +97,49 @@ def _new_manifest_list_file_name(snapshot_id: int, attempt:
int, commit_uuid: uu
return f"snap-{snapshot_id}-{attempt}-{commit_uuid}.avro"
+@dataclass(frozen=True)
+class CommitWindow:
+ """Tracks the commit range to validate against during retry.
+
+ base: The snapshot when the operation began (exclusive lower bound, fixed
across retries).
+ head: The branch HEAD snapshot after metadata refresh (inclusive upper
bound).
+ """
+
+ base: Snapshot | None
+ head: Snapshot | None
+
+ @classmethod
+ def resolve(cls, metadata: TableMetadata, base_id: int | None, branch: str
| None) -> CommitWindow:
+ """Resolve a CommitWindow from metadata, starting snapshot ID, and
target branch."""
+ head = metadata.snapshot_by_name(branch)
+ base = metadata.snapshot_by_id(base_id) if base_id is not None else
None
+ if base_id is not None and base is None:
+ raise ValidationException(f"Cannot find starting snapshot
{base_id}")
+ return cls(base=base, head=head)
+
+ def is_empty(self) -> bool:
+ """Return True if no concurrent commits occurred (validation can be
skipped)."""
+ return self.head is None or self.base is None or self.base.snapshot_id
== self.head.snapshot_id
+
+
class _SnapshotProducer(UpdateTableMetadata[U], Generic[U]):
commit_uuid: uuid.UUID
_io: FileIO
_operation: Operation
_snapshot_id: int
_parent_snapshot_id: int | None
+ _starting_snapshot_id: int | None
_added_data_files: list[DataFile]
_manifest_num_counter: itertools.count[int]
_deleted_data_files: set[DataFile]
_compression: AvroCompressionCodec
_target_branch: str | None
_predicate: BooleanExpression
_case_sensitive: bool
+ _commit_window: CommitWindow | None
+ _written_manifests: list[str]
+ _uncommitted_manifests: list[str]
+ _written_manifest_lists: list[str]
Review Comment:
**Nit**: `_isolation_level_property` should be declared as a class-level
type annotation alongside the other instance variable declarations. The rest of
`_SnapshotProducer's` fields follow this pattern of declaring the attribute's
type at the class body level for discoverability and static analysis, then
assigning in `__init__`. Missing it here breaks the convention and makes the
field invisible when scanning the class interface.
Currently, it's only set in the `__init__` as follows:
```python
self._isolation_level_property: str =
TableProperties.WRITE_DELETE_ISOLATION_LEVEL
```
we should also add it above in the class-level annotations block:
```python
_isolation_level_property: str
```
--
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]
