SQL Catalog: Filter on iceberg_type in commit_table (#3337)

pyiceberg/catalog/sql.py

@@ -517,6 +517,7 @@ def commit_table(
         with Session(self.engine) as session:
             if current_table:
                 # table exists, update it
+                type_filter = self._iceberg_type_filter()
                 if self.engine.dialect.supports_sane_rowcount:
                     stmt = (
                         update(IcebergTables)
@@ -531,12 +532,14 @@ def commit_table(
                             previous_metadata_location=current_table.metadata_location,
                         )
                     )
+                    if type_filter is not None:
+                        stmt = stmt.where(type_filter)
                     result = session.execute(stmt)
                     if result.rowcount < 1:
                         raise CommitFailedException(f"Table has been updated by another process: {namespace}.{table_name}")
                 else:
                     try:
-                        tbl = (
+                        query = (
                             session.query(IcebergTables)
                             .with_for_update(of=IcebergTables)
                             .filter(
@@ -545,8 +548,10 @@ def commit_table(
                                 IcebergTables.table_name == table_name,
                                 IcebergTables.metadata_location == current_table.metadata_location,
                             )
-                            .one()
                         )
+                        if type_filter is not None:
+                            query = query.filter(type_filter)
+                        tbl = query.one()
                         tbl.metadata_location = updated_staged_table.metadata_location
                         tbl.previous_metadata_location = current_table.metadata_location
                     except NoResultFound as e:

tests/catalog/test_sql.py

@@ -18,6 +18,7 @@
 from collections.abc import Generator
 from pathlib import Path
 from typing import cast
+from unittest.mock import patch
 
 import pytest
 from sqlalchemy import Engine, create_engine, inspect, text
@@ -32,6 +33,7 @@
     SqlCatalogBaseTable,
 )
 from pyiceberg.exceptions import (
+    CommitFailedException,
     NoSuchPropertyException,
     NoSuchTableError,
     TableAlreadyExistsError,
@@ -388,6 +390,37 @@ def test_rename_table_ignores_view_rows(warehouse: Path) -> None:
     assert row[0] == "VIEW"
 
 
+def test_commit_table_ignores_view_rows(warehouse: Path) -> None:
+    catalog = SqlCatalog(
+        name="test", uri=f"sqlite:///{warehouse.as_posix()}/test_commit_view.db", warehouse=f"file://{warehouse}"
+    )
+    catalog.create_namespace("ns")
+    schema = Schema(NestedField(1, "id", StringType(), required=True))
+    tbl = catalog.create_table(("ns", "a_table"), schema=schema)
+
+    tx = tbl.transaction()
+    tx.set_properties({"key": "val"})
+    updates = tuple(tx._updates)
+    requirements = tuple(tx._requirements)
+
+    # Tamper the table row into a VIEW (simulating concurrent change to VIEW after load_table)
+    with catalog.engine.connect() as conn:
+        conn.execute(text("UPDATE iceberg_tables SET iceberg_type = 'VIEW' WHERE table_name = 'a_table'"))
+        conn.commit()
+
+    # When load_table returns the pre-loaded table, commit_table must fail in the SQL UPDATE
+    # due to type_filter and not overwrite the VIEW row.
+    with patch.object(catalog, "load_table", return_value=tbl):
+        with pytest.raises(CommitFailedException, match="Table has been updated by another process: ns.a_table"):
+            catalog.commit_table(tbl, requirements=requirements, updates=updates)
+
+    # The view row must remain untouched with iceberg_type == 'VIEW'
+    with catalog.engine.connect() as conn:
+        row = conn.execute(text("SELECT iceberg_type FROM iceberg_tables WHERE table_name = 'a_table'")).fetchone()
+    assert row is not None
+    assert row[0] == "VIEW"
+
+
 def test_migration_to_v1_with_property_set(warehouse: Path) -> None:
     uri = f"sqlite:///{warehouse.as_posix()}/test-v1-migrate"
     engine = _create_v0_db(uri)