update parent snapshot when there are conflicts and change exception

Author: kaushiksrini

pyiceberg/table/update/snapshot.py

@@ -27,6 +27,7 @@
 
 from sortedcontainers import SortedList
 
+from pyiceberg.exceptions import CommitFailedException
 from pyiceberg.expressions import (
     AlwaysFalse,
     BooleanExpression,
@@ -259,9 +260,14 @@ def _commit(self) -> UpdatesAndRequirements:
         if not isinstance(self._transaction._table, StagedTable):
             starting_snapshot = self._transaction.table_metadata.current_snapshot()
             current_snapshot = self._transaction._table.refresh().metadata.current_snapshot()
+
             if starting_snapshot is not None and current_snapshot is not None:
                 self._validate(starting_snapshot, current_snapshot)
 
+            # If the current snapshot is not the same as the parent snapshot, update the parent snapshot id
+            if current_snapshot is not None and current_snapshot.snapshot_id != self._parent_snapshot_id:
+                self._parent_snapshot_id = current_snapshot.snapshot_id
+
         with write_manifest_list(
             format_version=self._transaction.table_metadata.format_version,
             output_file=self._io.new_output(manifest_list_file_path),
@@ -309,7 +315,7 @@ def _validate(self, starting_snapshot: Snapshot, current_snapshot: Snapshot) ->
             snapshot_operation = snapshot.summary.operation if snapshot.summary is not None else None
 
             if snapshot_operation not in allowed_operations[self._operation]:
-                raise ValueError(
+                raise CommitFailedException(
                     f"Operation {snapshot_operation} is not allowed when performing {self._operation}. "
                     "Check for overlaps or conflicts."
                 )

tests/integration/test_add_files.py

@@ -858,12 +858,15 @@ def test_conflict_delete_delete(
     spark: SparkSession, session_catalog: Catalog, arrow_table_with_null: pa.Table, format_version: int
 ) -> None:
     identifier = "default.test_conflict"
-    tbl1 = _create_table(session_catalog, identifier, format_version, [arrow_table_with_null])
+    tbl1 = _create_table(session_catalog, identifier, format_version, schema=arrow_table_with_null.schema)
+    tbl1.append(arrow_table_with_null)
     tbl2 = session_catalog.load_table(identifier)
 
     tbl1.delete("string == 'z'")
 
-    with pytest.raises(CommitFailedException, match="(branch main has changed: expected id ).*"):
+    with pytest.raises(
+        CommitFailedException, match="Operation .* is not allowed when performing .*. Check for overlaps or conflicts."
+    ):
         # tbl2 isn't aware of the commit by tbl1
         tbl2.delete("string == 'z'")
 
@@ -874,7 +877,8 @@ def test_conflict_delete_append(
     spark: SparkSession, session_catalog: Catalog, arrow_table_with_null: pa.Table, format_version: int
 ) -> None:
     identifier = "default.test_conflict"
-    tbl1 = _create_table(session_catalog, identifier, format_version, [arrow_table_with_null])
+    tbl1 = _create_table(session_catalog, identifier, format_version, schema=arrow_table_with_null.schema)
+    tbl1.append(arrow_table_with_null)
     tbl2 = session_catalog.load_table(identifier)
 
     # This is allowed
@@ -888,12 +892,15 @@ def test_conflict_append_delete(
     spark: SparkSession, session_catalog: Catalog, arrow_table_with_null: pa.Table, format_version: int
 ) -> None:
     identifier = "default.test_conflict"
-    tbl1 = _create_table(session_catalog, identifier, format_version, [arrow_table_with_null])
+    tbl1 = _create_table(session_catalog, identifier, format_version, schema=arrow_table_with_null.schema)
+    tbl1.append(arrow_table_with_null)
     tbl2 = session_catalog.load_table(identifier)
 
     tbl1.append(arrow_table_with_null)
 
-    with pytest.raises(CommitFailedException, match="(branch main has changed: expected id ).*"):
+    with pytest.raises(
+        CommitFailedException, match="Operation .* is not allowed when performing .*. Check for overlaps or conflicts."
+    ):
         # tbl2 isn't aware of the commit by tbl1
         tbl2.delete("string == 'z'")
 
@@ -904,7 +911,8 @@ def test_conflict_append_append(
     spark: SparkSession, session_catalog: Catalog, arrow_table_with_null: pa.Table, format_version: int
 ) -> None:
     identifier = "default.test_conflict"
-    tbl1 = _create_table(session_catalog, identifier, format_version, [arrow_table_with_null])
+    tbl1 = _create_table(session_catalog, identifier, format_version, schema=arrow_table_with_null.schema)
+    tbl1.append(arrow_table_with_null)
     tbl2 = session_catalog.load_table(identifier)
 
     tbl1.append(arrow_table_with_null)