Clear updates/requirements after commit (#1961)

# Rationale for this change

Resolves #1946

# Are these changes tested?

Yes, using a test that used to fail before :)

# Are there any user-facing changes?

<!-- In the case of user-facing changes, please add the changelog label.
-->

Author: Fokko

pyiceberg/table/__init__.py

@@ -292,8 +292,6 @@ def _apply(self, updates: Tuple[TableUpdate, ...], requirements: Tuple[TableRequ
 
         if self._autocommit:
             self.commit_transaction()
-            self._updates = ()
-            self._requirements = ()
 
         return self
 
@@ -937,13 +935,15 @@ def commit_transaction(self) -> Table:
                 updates=self._updates,
                 requirements=self._requirements,
             )
-            return self._table
-        else:
-            return self._table
+
+        self._updates = ()
+        self._requirements = ()
+
+        return self._table
 
 
 class CreateTableTransaction(Transaction):
-    """A transaction that involves the creation of a a new table."""
+    """A transaction that involves the creation of a new table."""
 
     def _initial_changes(self, table_metadata: TableMetadata) -> None:
         """Set the initial changes that can reconstruct the initial table metadata when creating the CreateTableTransaction."""
@@ -988,11 +988,15 @@ def commit_transaction(self) -> Table:
         Returns:
             The table with the updates applied.
         """
-        self._requirements = (AssertCreate(),)
-        self._table._do_commit(  # pylint: disable=W0212
-            updates=self._updates,
-            requirements=self._requirements,
-        )
+        if len(self._updates) > 0:
+            self._table._do_commit(  # pylint: disable=W0212
+                updates=self._updates,
+                requirements=(AssertCreate(),),
+            )
+
+        self._updates = ()
+        self._requirements = ()
+
         return self._table
 
 

tests/integration/test_writes/test_writes.py

@@ -1802,6 +1802,23 @@ def test_write_optional_list(session_catalog: Catalog) -> None:
     assert len(session_catalog.load_table(identifier).scan().to_arrow()) == 4
 
 
+@pytest.mark.integration
+@pytest.mark.parametrize("format_version", [1, 2])
+def test_double_commit_transaction(
+    spark: SparkSession, session_catalog: Catalog, arrow_table_with_null: pa.Table, format_version: int
+) -> None:
+    identifier = "default.arrow_data_files"
+    tbl = _create_table(session_catalog, identifier, {"format-version": format_version}, [])
+
+    assert len(tbl.metadata.metadata_log) == 0
+
+    with tbl.transaction() as tx:
+        tx.append(arrow_table_with_null)
+        tx.commit_transaction()
+
+    assert len(tbl.metadata.metadata_log) == 1
+
+
 @pytest.mark.integration
 @pytest.mark.parametrize("format_version", [1, 2])
 def test_evolve_and_write(