diff --git a/pyiceberg/catalog/sql.py b/pyiceberg/catalog/sql.py index fb0d9b98e9..abe31194b8 100644 --- a/pyiceberg/catalog/sql.py +++ b/pyiceberg/catalog/sql.py @@ -24,6 +24,7 @@ ) from sqlalchemy import ( + ColumnElement, String, create_engine, delete, @@ -84,6 +85,7 @@ DEFAULT_ECHO_VALUE = "false" DEFAULT_POOL_PRE_PING_VALUE = "false" DEFAULT_INIT_CATALOG_TABLES = "true" +ICEBERG_TABLE_TYPE = "TABLE" class SqlCatalogBaseTable(MappedAsDataclass, DeclarativeBase): @@ -209,9 +211,16 @@ def _create_table_row(self, namespace: str, table_name: str, metadata_location: "previous_metadata_location": None, } if self._schema_version == "v1": - row["iceberg_type"] = "TABLE" + row["iceberg_type"] = ICEBERG_TABLE_TYPE return row + def _iceberg_type_filter(self) -> ColumnElement[bool] | None: + # Excludes non-table rows (e.g. views written by iceberg-java or iceberg-rust) from table + # lookups. None on v0 schemas, where the iceberg_type column doesn't exist to filter on. + if self._schema_version != "v1": + return None + return (IcebergTables.iceberg_type == ICEBERG_TABLE_TYPE) | (IcebergTables.iceberg_type.is_(None)) + def _convert_orm_to_iceberg(self, orm_table: IcebergTables) -> Table: # Check for expected properties. if not (metadata_location := orm_table.metadata_location): @@ -349,6 +358,8 @@ def load_table(self, identifier: str | Identifier) -> Table: IcebergTables.table_namespace == namespace, IcebergTables.table_name == table_name, ) + if (type_filter := self._iceberg_type_filter()) is not None: + stmt = stmt.where(type_filter) result = session.scalar(stmt) if result: return self._convert_orm_to_iceberg(result) @@ -367,20 +378,22 @@ def drop_table(self, identifier: str | Identifier) -> None: namespace_tuple = Catalog.namespace_from(identifier) namespace = Catalog.namespace_to_string(namespace_tuple) table_name = Catalog.table_name_from(identifier) + type_filter = self._iceberg_type_filter() with Session(self.engine) as session: if self.engine.dialect.supports_sane_rowcount: - res = session.execute( - delete(IcebergTables).where( - IcebergTables.catalog_name == self.name, - IcebergTables.table_namespace == namespace, - IcebergTables.table_name == table_name, - ) + stmt = delete(IcebergTables).where( + IcebergTables.catalog_name == self.name, + IcebergTables.table_namespace == namespace, + IcebergTables.table_name == table_name, ) + if type_filter is not None: + stmt = stmt.where(type_filter) + res = session.execute(stmt) if res.rowcount < 1: raise NoSuchTableError(f"Table does not exist: {namespace}.{table_name}") else: try: - tbl = ( + query = ( session.query(IcebergTables) .with_for_update(of=IcebergTables) .filter( @@ -388,8 +401,10 @@ def drop_table(self, identifier: str | Identifier) -> None: IcebergTables.table_namespace == namespace, IcebergTables.table_name == table_name, ) - .one() ) + if type_filter is not None: + query = query.filter(type_filter) + tbl = query.one() session.delete(tbl) except NoResultFound as e: raise NoSuchTableError(f"Table does not exist: {namespace}.{table_name}") from e @@ -419,6 +434,7 @@ def rename_table(self, from_identifier: str | Identifier, to_identifier: str | I to_table_name = Catalog.table_name_from(to_identifier) if not self.namespace_exists(to_namespace): raise NoSuchNamespaceError(f"Namespace does not exist: {to_namespace}") + type_filter = self._iceberg_type_filter() with Session(self.engine) as session: try: if self.engine.dialect.supports_sane_rowcount: @@ -431,12 +447,14 @@ def rename_table(self, from_identifier: str | Identifier, to_identifier: str | I ) .values(table_namespace=to_namespace, table_name=to_table_name) ) + if type_filter is not None: + stmt = stmt.where(type_filter) result = session.execute(stmt) if result.rowcount < 1: raise NoSuchTableError(f"Table does not exist: {from_table_name}") else: try: - tbl = ( + query = ( session.query(IcebergTables) .with_for_update(of=IcebergTables) .filter( @@ -444,8 +462,10 @@ def rename_table(self, from_identifier: str | Identifier, to_identifier: str | I IcebergTables.table_namespace == from_namespace, IcebergTables.table_name == from_table_name, ) - .one() ) + if type_filter is not None: + query = query.filter(type_filter) + tbl = query.one() tbl.table_namespace = to_namespace tbl.table_name = to_table_name except NoResultFound as e: @@ -656,9 +676,8 @@ def list_tables(self, namespace: str | Identifier) -> list[Identifier]: IcebergTables.table_namespace == namespace, ) - # Filter out views only if schema_version is v1 - if self._schema_version == "v1": - stmt = stmt.where((IcebergTables.iceberg_type == "TABLE") | (IcebergTables.iceberg_type.is_(None))) + if (type_filter := self._iceberg_type_filter()) is not None: + stmt = stmt.where(type_filter) with Session(self.engine) as session: result = session.scalars(stmt) diff --git a/tests/catalog/test_sql.py b/tests/catalog/test_sql.py index 1eabc56a9e..da9b3397fd 100644 --- a/tests/catalog/test_sql.py +++ b/tests/catalog/test_sql.py @@ -33,6 +33,7 @@ ) from pyiceberg.exceptions import ( NoSuchPropertyException, + NoSuchTableError, TableAlreadyExistsError, ) from pyiceberg.schema import Schema @@ -333,6 +334,60 @@ def test_list_tables_filters_by_iceberg_type(warehouse: Path) -> None: assert "some_view" not in tables +def _insert_view_row(catalog: SqlCatalog, namespace: str, table_name: str) -> None: + # Simulates a view row written by iceberg-java or iceberg-rust, which SqlCatalog table + # operations must not treat as a table (issue #3337). + with catalog.engine.connect() as conn: + conn.execute( + text( + "INSERT INTO iceberg_tables " + "(catalog_name, table_namespace, table_name, metadata_location, previous_metadata_location, iceberg_type) " + "VALUES ('test', :namespace, :table_name, 's3://fake/metadata.json', NULL, 'VIEW')" + ), + {"namespace": namespace, "table_name": table_name}, + ) + conn.commit() + + +def test_load_table_ignores_view_rows(warehouse: Path) -> None: + catalog = SqlCatalog(name="test", uri="sqlite:///:memory:", warehouse=f"file://{warehouse}") + catalog.create_namespace("ns") + _insert_view_row(catalog, "ns", "a_view") + + with pytest.raises(NoSuchTableError): + catalog.load_table(("ns", "a_view")) + + +def test_drop_table_ignores_view_rows(warehouse: Path) -> None: + catalog = SqlCatalog(name="test", uri="sqlite:///:memory:", warehouse=f"file://{warehouse}") + catalog.create_namespace("ns") + _insert_view_row(catalog, "ns", "a_view") + + with pytest.raises(NoSuchTableError): + catalog.drop_table(("ns", "a_view")) + + # The view row must not have been deleted. + 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_rename_table_ignores_view_rows(warehouse: Path) -> None: + catalog = SqlCatalog(name="test", uri="sqlite:///:memory:", warehouse=f"file://{warehouse}") + catalog.create_namespace("ns") + _insert_view_row(catalog, "ns", "a_view") + + with pytest.raises(NoSuchTableError): + catalog.rename_table(("ns", "a_view"), ("ns", "renamed_view")) + + # The view row must not have been renamed. + 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}/test-v1-migrate" engine = _create_v0_db(uri)