Fix the requirement

Author: Fokko

pyiceberg/table/__init__.py

@@ -243,6 +243,8 @@ class TableProperties:
     WRITE_PARTITION_SUMMARY_LIMIT_DEFAULT = 0
 
     DELETE_MODE = "write.delete.mode"
+    DELETE_MODE_COPY_ON_WRITE = "merge-on-read"
+    DELETE_MODE_MERGE_ON_READ = "copy-on-write"
 
     DEFAULT_NAME_MAPPING = "schema.name-mapping.default"
     FORMAT_VERSION = "format-version"
@@ -301,7 +303,13 @@ def _apply(self, updates: Tuple[TableUpdate, ...], requirements: Tuple[TableRequ
             requirement.validate(self.table_metadata)
 
         self._updates += updates
-        self._requirements += requirements
+
+        # For the requirements, it does not make sense to add a requirement more than once
+        # For example, you cannot assert that the current schema has two different IDs
+        existing_requirements = {type(requirement) for requirement in self._requirements}
+        for new_requirement in requirements:
+            if type(new_requirement) not in existing_requirements:
+                self._requirements = self._requirements + requirements
 
         self.table_metadata = update_table_metadata(self.table_metadata, updates)
 
@@ -461,8 +469,8 @@ def overwrite(
                     update_snapshot.append_data_file(data_file)
 
     def delete(self, delete_filter: BooleanExpression, snapshot_properties: Dict[str, str] = EMPTY_DICT) -> None:
-        if (mode := self.table_metadata.properties.get(TableProperties.DELETE_MODE)) and mode != 'copy-on-write':
-            warnings.warn("PyIceberg only supports copy on write")
+        if self.table_metadata.properties.get(TableProperties.DELETE_MODE, TableProperties.DELETE_MODE_COPY_ON_WRITE) == TableProperties.DELETE_MODE_MERGE_ON_READ:
+            raise NotImplementedError("Merge on read is not yet supported")
 
         with self.update_snapshot(snapshot_properties=snapshot_properties).delete() as delete_snapshot:
             delete_snapshot.delete_by_predicate(delete_filter)
@@ -2948,7 +2956,7 @@ def _commit(self) -> UpdatesAndRequirements:
             ),
             (
                 AssertTableUUID(uuid=self._transaction.table_metadata.table_uuid),
-                # AssertRefSnapshotId(snapshot_id=self._parent_snapshot_id, ref="main"),
+                AssertRefSnapshotId(snapshot_id=self._transaction.table_metadata.current_snapshot_id, ref="main"),
             ),
         )
 

tests/integration/test_rest_schema.py

@@ -2512,15 +2512,18 @@ def test_two_add_schemas_in_a_single_transaction(catalog: Catalog) -> None:
         ),
     )
 
-    with pytest.raises(CommitFailedException) as exc_info:
-        with tbl.transaction() as tr:
-            with tr.update_schema() as update:
-                update.add_column("bar", field_type=StringType())
-            with tr.update_schema() as update:
-                update.add_column("baz", field_type=StringType())
-
-    assert "CommitFailedException: Requirement failed: current schema changed: expected id 1 != 0" in str(exc_info.value)
+    with tbl.transaction() as tr:
+        with tr.update_schema() as update:
+            update.add_column("bar", field_type=StringType())
+        with tr.update_schema() as update:
+            update.add_column("baz", field_type=StringType())
 
+    assert tbl.schema().schema_id == 2
+    assert tbl.schema() == Schema(
+        NestedField(field_id=1, name="foo", field_type=StringType()),
+        NestedField(field_id=2, name="bar", field_type=StringType()),
+        NestedField(field_id=3, name="baz", field_type=StringType()),
+    )
 
 @pytest.mark.integration
 def test_create_table_integrity_after_fresh_assignment(catalog: Catalog) -> None: