diff --git a/pyiceberg/catalog/sql.py b/pyiceberg/catalog/sql.py index abe31194b8..0dd401a2e1 100644 --- a/pyiceberg/catalog/sql.py +++ b/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: diff --git a/tests/catalog/test_sql.py b/tests/catalog/test_sql.py index 6c1767f711..f819f5f464 100644 --- a/tests/catalog/test_sql.py +++ b/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)