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

Author: hedger9487

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

@@ -32,6 +32,7 @@
     SqlCatalogBaseTable,
 )
 from pyiceberg.exceptions import (
+    CommitFailedException,
     NoSuchPropertyException,
     NoSuchTableError,
     TableAlreadyExistsError,
@@ -388,6 +389,31 @@ 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_view"), schema=schema)
+
+    # Tamper the table row into a VIEW (simulating external writer)
+    with catalog.engine.connect() as conn:
+        conn.execute(text("UPDATE iceberg_tables SET iceberg_type = 'VIEW' WHERE table_name = 'a_view'"))
+        conn.commit()
+
+    # Attempting to commit table updates must fail and not modify the VIEW row
+    with pytest.raises(CommitFailedException):
+        with tbl.update_schema() as update:
+            update.add_column("new_col", StringType())
+
+    # 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_view'")).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)