filter iceberg_type across all SqlCatalog table operations

Author: avison9

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,29 +378,33 @@ 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(
                             IcebergTables.catalog_name == self.name,
                             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,21 +447,25 @@ 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(
                                 IcebergTables.catalog_name == self.name,
                                 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)

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)