diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 8d79472af5..03fcf5fd00 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -794,7 +794,7 @@ def delete( preserve_row_filter = _expression_to_complementary_pyarrow(bound_delete_filter, self.table_metadata.schema()) file_scan = self._scan(row_filter=delete_filter, case_sensitive=case_sensitive) - if branch is not None: + if branch in self.table_metadata.refs: file_scan = file_scan.use_ref(branch) files = file_scan.plan_files() diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index 8feebb9901..764653e92f 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/pyiceberg/table/update/snapshot.py @@ -108,7 +108,8 @@ 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). + head: The branch HEAD snapshot after metadata refresh, or the current snapshot if the branch does not exist + (inclusive upper bound). """ base: Snapshot | None @@ -117,7 +118,7 @@ class CommitWindow: @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) + head = metadata.snapshot_by_name(branch) or metadata.current_snapshot() 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}") @@ -197,8 +198,9 @@ def _validate_target_branch(self, branch: str | None) -> str | None: return branch def _current_branch_head_id(self) -> int | None: - """Return the snapshot id at the head of the target branch, or None if the branch has no snapshot.""" - snapshot = self._transaction.table_metadata.snapshot_by_name(self._target_branch) + """Return the snapshot id at the head of the target branch, or the current snapshot id if the branch does not exist.""" + table_metadata = self._transaction.table_metadata + snapshot = table_metadata.snapshot_by_name(self._target_branch) or table_metadata.current_snapshot() return snapshot.snapshot_id if snapshot else None def append_data_file(self, data_file: DataFile) -> _SnapshotProducer[U]: @@ -782,7 +784,9 @@ def _existing_manifests(self) -> list[ManifestFile]: existing_files = [] manifest_evaluators: dict[int, Callable[[ManifestFile], bool]] = KeyDefaultDict(self._build_manifest_evaluator) - if snapshot := self._transaction.table_metadata.snapshot_by_name(name=self._target_branch): + if self._parent_snapshot_id is not None and ( + snapshot := self._transaction.table_metadata.snapshot_by_id(self._parent_snapshot_id) + ): for manifest_file in snapshot.manifests(io=self._io): # Manifest does not contain rows that match the files to delete partitions if not manifest_evaluators[manifest_file.partition_spec_id](manifest_file): @@ -922,7 +926,11 @@ def overwrite(self, commit_uuid: uuid.UUID | None = None) -> _OverwriteFiles: return _OverwriteFiles( commit_uuid=commit_uuid, operation=Operation.OVERWRITE - if self._transaction.table_metadata.snapshot_by_name(name=self._branch) is not None + if ( + self._transaction.table_metadata.snapshot_by_name(name=self._branch) + or self._transaction.table_metadata.current_snapshot() + ) + is not None else Operation.APPEND, transaction=self._transaction, io=self._io, diff --git a/tests/catalog/test_catalog_behaviors.py b/tests/catalog/test_catalog_behaviors.py index 1148ca8527..fa79292fa1 100644 --- a/tests/catalog/test_catalog_behaviors.py +++ b/tests/catalog/test_catalog_behaviors.py @@ -1409,3 +1409,37 @@ def test_record_batch_reader_schema_mismatch_writes_no_files(catalog: Catalog) - # _append_snapshot_producer opens. assert tbl.metadata.current_snapshot() is None assert len(tbl.scan().to_arrow()) == 0 + + +def test_append_to_missing_branch_starts_from_main(catalog: Catalog) -> None: + catalog.create_namespace("default") + identifier = f"default.append_to_missing_branch_{catalog.name}" + tbl = catalog.create_table(identifier=identifier, schema=pa.schema([("x", pa.int64())])) + tbl.append(pa.table({"x": [1, 2]})) + main_snapshot_id = tbl.metadata.current_snapshot_id + + tbl.append(pa.table({"x": [3]}), branch="b") + + # Like Java's SnapshotUtil.latestSnapshot, a branch that does not exist starts from the current snapshot of main + branch_snapshot = tbl.snapshot_by_name("b") + assert branch_snapshot is not None + assert branch_snapshot.parent_snapshot_id == main_snapshot_id + assert sorted(tbl.scan().use_ref("b").to_arrow()["x"].to_pylist()) == [1, 2, 3] + + +def test_delete_on_missing_branch_starts_from_main(catalog: Catalog) -> None: + catalog.create_namespace("default") + identifier = f"default.delete_on_missing_branch_{catalog.name}" + tbl = catalog.create_table(identifier=identifier, schema=pa.schema([("x", pa.int64())])) + tbl.append(pa.table({"x": [1, 2]})) + tbl.append(pa.table({"x": [5]})) + + # x == 1 matches only part of the file [1, 2], so that file is rewritten to [2] and the file [5] is kept + tbl.delete("x == 1", branch="b") + + branch_snapshot = tbl.snapshot_by_name("b") + assert branch_snapshot is not None + assert sorted(tbl.scan().use_ref("b").to_arrow()["x"].to_pylist()) == [2, 5] + # The rewrite removes a data file of main, so it is an overwrite and not an append + assert branch_snapshot.summary is not None + assert branch_snapshot.summary.operation == Operation.OVERWRITE diff --git a/tests/table/test_commit_retry.py b/tests/table/test_commit_retry.py index ab95457e23..2dc8d7d457 100644 --- a/tests/table/test_commit_retry.py +++ b/tests/table/test_commit_retry.py @@ -957,6 +957,59 @@ def test_commit_retry_delete_on_non_main_branch(catalog: Catalog) -> None: assert sorted(result.column("x").to_pylist()) == [2, 3, 4, 5, 6, 7, 8, 9] +def test_commit_retry_after_branch_removed_starts_from_main(catalog: Catalog) -> None: + """Verify that a retried append to a branch removed concurrently starts from the current snapshot of main.""" + catalog.create_namespace("default") + catalog.create_table("default.branch_removed_retry_test", schema=_test_schema()) + + import pyarrow as pa + + tbl = catalog.load_table("default.branch_removed_retry_test") + tbl.append(pa.table({"x": [1, 2]})) + assert tbl.metadata.current_snapshot_id is not None + tbl.manage_snapshots().create_branch(snapshot_id=tbl.metadata.current_snapshot_id, branch_name="test-branch").commit() + tbl.append(pa.table({"x": [3]})) + + tbl1 = catalog.load_table("default.branch_removed_retry_test") + tbl2 = catalog.load_table("default.branch_removed_retry_test") + + tbl1.manage_snapshots().remove_branch("test-branch").commit() + + # tbl2 still sees test-branch, so the first attempt fails and the retry finds the branch missing + tbl2.append(pa.table({"x": [4]}), branch="test-branch") + + refreshed = catalog.load_table("default.branch_removed_retry_test") + branch_snapshot = refreshed.snapshot_by_name("test-branch") + assert branch_snapshot is not None + # The parent is the current snapshot of main ([1, 2] + [3]), not the removed head of test-branch ([1, 2]) + assert branch_snapshot.parent_snapshot_id == refreshed.metadata.current_snapshot_id + assert sorted(refreshed.scan(snapshot_id=branch_snapshot.snapshot_id).to_arrow().column("x").to_pylist()) == [1, 2, 3, 4] + + +def test_delete_retried_after_branch_removed_validates_against_main(catalog: Catalog) -> None: + """Verify that a delete retried on main after its branch was removed concurrently validates the changes on main.""" + catalog.create_namespace("default") + catalog.create_table("default.branch_removed_delete_test", schema=_test_schema()) + + import pyarrow as pa + + tbl = catalog.load_table("default.branch_removed_delete_test") + tbl.append(pa.table({"x": [1, 2]})) + assert tbl.metadata.current_snapshot_id is not None + tbl.manage_snapshots().create_branch(snapshot_id=tbl.metadata.current_snapshot_id, branch_name="test-branch").commit() + + tbl1 = catalog.load_table("default.branch_removed_delete_test") + tbl2 = catalog.load_table("default.branch_removed_delete_test") + + tbl1.manage_snapshots().remove_branch("test-branch").commit() + # A concurrent append to main adds a row that matches the delete filter + catalog.load_table("default.branch_removed_delete_test").append(pa.table({"x": [1]})) + + # tbl2 still sees test-branch, so the retry starts from main, and the default serializable isolation rejects the new row + with pytest.raises(ValidationException, match="Added data files were found matching the filter"): + tbl2.delete("x <= 2", branch="test-branch") + + def test_negative_num_retries_still_commits(catalog: Catalog) -> None: """A negative commit.retry.num-retries is clamped to 0, so one attempt still runs (matches Java).