Validate against full history when starting from an empty table

A writer that started on a table with no snapshots treated the commit
window as empty and skipped validation. If another writer landed the first
snapshot concurrently, its rows could be dropped with no error. Treat a None
base with a live head as a full-history validation, matching Java.

Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

Author: lawofcycles

pyiceberg/table/update/snapshot.py

@@ -118,8 +118,13 @@ def resolve(cls, metadata: TableMetadata, base_id: int | None, branch: str | Non
         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
+        """Return True if no concurrent commits occurred (validation can be skipped).
+
+        A None base means the operation started on a table with no snapshots. If a head
+        exists, another writer landed the first snapshot concurrently, so the window is not
+        empty and validation must run against the whole history.
+        """
+        return self.head is None or (self.base is not None and self.base.snapshot_id == self.head.snapshot_id)
 
 
 class _SnapshotProducer(UpdateTableMetadata[U], Generic[U]):
@@ -476,7 +481,7 @@ def _validate_concurrency(self) -> None:
         catalog_head = self._commit_window.head
         starting_snapshot = self._commit_window.base
 
-        if catalog_head is None or starting_snapshot is None:
+        if catalog_head is None:
             return
 
         table = self._transaction._table

pyiceberg/table/update/validate.py

@@ -40,19 +40,20 @@
 
 def _validation_history(
     table: Table,
-    from_snapshot: Snapshot,
+    from_snapshot: Snapshot | None,
     to_snapshot: Snapshot,
     matching_operations: set[Operation],
     manifest_content_filter: ManifestContent,
 ) -> tuple[list[ManifestFile], set[int]]:
     """Return newly added manifests and snapshot IDs between the starting snapshot and parent snapshot.
 
     Walks from to_snapshot backwards towards from_snapshot, collecting manifests from
-    snapshots whose operations match. from_snapshot is excluded from results.
+    snapshots whose operations match. from_snapshot is excluded from results. A None
+    from_snapshot walks the entire history of to_snapshot down to the root.
 
     Args:
         table: Table to get the history from
-        from_snapshot: Snapshot where the walk stops (exclusive)
+        from_snapshot: Snapshot where the walk stops (exclusive), or None to walk the whole history
         to_snapshot: Snapshot where the walk starts
         matching_operations: Operations to match on
         manifest_content_filter: Manifest content type to filter
@@ -63,15 +64,15 @@ def _validation_history(
     Returns:
         List of manifest files and set of snapshots ID's matching conditions
     """
-    if from_snapshot.snapshot_id == to_snapshot.snapshot_id:
+    if from_snapshot is not None and from_snapshot.snapshot_id == to_snapshot.snapshot_id:
         return [], set()
 
     manifests_files: list[ManifestFile] = []
     snapshots: set[int] = set()
 
     last_snapshot = None
     for snapshot in ancestors_between(from_snapshot, to_snapshot, table.metadata):
-        if snapshot.snapshot_id == from_snapshot.snapshot_id:
+        if from_snapshot is not None and snapshot.snapshot_id == from_snapshot.snapshot_id:
             last_snapshot = snapshot
             break
         last_snapshot = snapshot
@@ -90,7 +91,7 @@ def _validation_history(
             ]
         )
 
-    if last_snapshot is None or last_snapshot.snapshot_id != from_snapshot.snapshot_id:
+    if from_snapshot is not None and (last_snapshot is None or last_snapshot.snapshot_id != from_snapshot.snapshot_id):
         raise ValidationException("No matching snapshot found.")
 
     return manifests_files, snapshots
@@ -216,9 +217,6 @@ def _added_data_files(
     Returns:
         Iterator of manifest entries for added data files matching the conditions
     """
-    if parent_snapshot is None:
-        return
-
     manifests, snapshot_ids = _validation_history(
         table,
         parent_snapshot,
@@ -252,7 +250,7 @@ def _added_delete_files(
     Returns:
         DeleteFileIndex
     """
-    if parent_snapshot is None or table.format_version < 2:
+    if table.format_version < 2:
         return DeleteFileIndex()
 
     manifests, snapshot_ids = _validation_history(
@@ -352,7 +350,7 @@ def _validate_no_new_deletes_for_data_files(
         parent_snapshot: Ending snapshot on the branch being validated
     """
     # If there is no current state, or no files has been added
-    if parent_snapshot is None or table.format_version < 2:
+    if table.format_version < 2:
         return
 
     deletes = _added_delete_files(table, starting_snapshot, data_filter, None, parent_snapshot)

tests/table/test_commit_retry.py

@@ -902,3 +902,35 @@ def test_negative_num_retries_still_commits(catalog: Catalog) -> None:
     table.append(pa.table({"x": [1]}))
 
     assert catalog.load_table("default.negative_retries_test").scan().to_arrow().to_pylist() == [{"x": 1}]
+
+
+def test_writer_that_started_on_an_empty_table_still_validates(catalog: Catalog) -> None:
+    """A writer that started on an empty table must still validate against a concurrently-landed first snapshot.
+
+    When the starting snapshot is None (empty table), the commit window must not be treated as empty
+    if a head exists. Otherwise a concurrent first snapshot bypasses validation and its rows can be
+    deleted with no error.
+    """
+    import pyarrow as pa
+
+    catalog.create_namespace("default")
+    schema = Schema(NestedField(1, "x", LongType(), required=False))
+    catalog.create_table(
+        "default.empty_start_validate",
+        schema=schema,
+        properties={
+            TableProperties.COMMIT_MIN_RETRY_WAIT_MS: "1",
+            TableProperties.COMMIT_MAX_RETRY_WAIT_MS: "2",
+        },
+    )
+
+    stale = catalog.load_table("default.empty_start_validate")
+    tx = stale.transaction()
+    with pytest.warns(UserWarning):  # the delete matches nothing on an empty table
+        tx.overwrite(pa.table({"x": [2]}), overwrite_filter="x == 1")
+
+    # The first ever snapshot lands concurrently, with a row matching the filter.
+    catalog.load_table("default.empty_start_validate").append(pa.table({"x": [1]}))
+
+    with pytest.raises(ValidationException):
+        tx.commit_transaction()