From e993bf751c9aee535938c6acb3e1f2a156ba2ead Mon Sep 17 00:00:00 2001 From: hedger9487 Date: Tue, 25 Aug 2026 11:05:44 +0800 Subject: [PATCH 1/2] SQL Catalog: Filter on iceberg_type in commit_table (#3337) --- pyiceberg/catalog/sql.py | 9 +++++++-- tests/catalog/test_sql.py | 26 ++++++++++++++++++++++++++ 2 files changed, 33 insertions(+), 2 deletions(-) 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..df96ad2f95 100644 --- a/tests/catalog/test_sql.py +++ b/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) From 00fbae1250cd15aa091564d33afba73e3fd6538f Mon Sep 17 00:00:00 2001 From: hedger9487 Date: Wed, 26 Aug 2026 10:10:46 +0800 Subject: [PATCH 2/2] Test: simulate concurrent change to verify commit_table SQL UPDATE type_filter regression --- tests/catalog/test_sql.py | 23 +++++++++++++++-------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/tests/catalog/test_sql.py b/tests/catalog/test_sql.py index df96ad2f95..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 @@ -395,21 +396,27 @@ def test_commit_table_ignores_view_rows(warehouse: Path) -> None: ) catalog.create_namespace("ns") schema = Schema(NestedField(1, "id", StringType(), required=True)) - tbl = catalog.create_table(("ns", "a_view"), schema=schema) + tbl = catalog.create_table(("ns", "a_table"), schema=schema) - # Tamper the table row into a VIEW (simulating external writer) + 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_view'")) + conn.execute(text("UPDATE iceberg_tables SET iceberg_type = 'VIEW' WHERE table_name = 'a_table'")) 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()) + # 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_view'")).fetchone() + 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"