diff --git a/pyiceberg/catalog/sql.py b/pyiceberg/catalog/sql.py index 87446bd58b..abe31194b8 100644 --- a/pyiceberg/catalog/sql.py +++ b/pyiceberg/catalog/sql.py @@ -17,16 +17,20 @@ from __future__ import annotations +import logging from typing import ( TYPE_CHECKING, + Any, ) from sqlalchemy import ( + ColumnElement, String, create_engine, delete, insert, select, + text, union, update, ) @@ -76,9 +80,12 @@ if TYPE_CHECKING: import pyarrow as pa +logger = logging.getLogger(__name__) + DEFAULT_ECHO_VALUE = "false" DEFAULT_POOL_PRE_PING_VALUE = "false" DEFAULT_INIT_CATALOG_TABLES = "true" +ICEBERG_TABLE_TYPE = "TABLE" class SqlCatalogBaseTable(MappedAsDataclass, DeclarativeBase): @@ -93,6 +100,7 @@ class IcebergTables(SqlCatalogBaseTable): table_name: Mapped[str] = mapped_column(String(255), nullable=False, primary_key=True) metadata_location: Mapped[str | None] = mapped_column(String(1000), nullable=True) previous_metadata_location: Mapped[str | None] = mapped_column(String(1000), nullable=True) + iceberg_type: Mapped[str | None] = mapped_column(String(5), nullable=True, deferred=True, init=False, default=None) class IcebergNamespaceProperties(SqlCatalogBaseTable): @@ -127,25 +135,66 @@ def __init__(self, name: str, **properties: str): echo_str = str(self.properties.get("echo", DEFAULT_ECHO_VALUE)).lower() echo = strtobool(echo_str) if echo_str != "debug" else "debug" pool_pre_ping = strtobool(self.properties.get("pool_pre_ping", DEFAULT_POOL_PRE_PING_VALUE)) - init_catalog_tables = strtobool(self.properties.get("init_catalog_tables", DEFAULT_INIT_CATALOG_TABLES)) self.engine = create_engine(uri_prop, echo=echo, pool_pre_ping=pool_pre_ping) + self._init_catalog() + + def _init_catalog(self) -> None: + """Detect schema state, create tables if absent, and migrate V0 to V1 when requested. + + Rules: + - New catalogs always get v1 schema with view support. + - Existing v0 catalogs are not migrated unless schema_version='v1' is explicitly set. + """ + # Get schema version from properties, ok if None + schema_version_prop = self.properties.get("schema_version", "v0") + if schema_version_prop not in ("v0", "v1"): + raise ValueError(f"Invalid schema_version property: '{schema_version_prop}'. Must be 'v0' or 'v1'.") - if init_catalog_tables: - self._ensure_tables_exist() + # Determine if catalog tables should be created if absent + init_catalog_tables = strtobool(self.properties.get("init_catalog_tables", DEFAULT_INIT_CATALOG_TABLES)) - def _ensure_tables_exist(self) -> None: with Session(self.engine) as session: - for table in [IcebergTables, IcebergNamespaceProperties]: - stmt = select(1).select_from(table) + # 1. Check if catalog tables exist; create if absent + tables_absent = False + for tbl in [IcebergTables, IcebergNamespaceProperties]: try: - session.scalar(stmt) - except ( - OperationalError, - ProgrammingError, - ): # sqlalchemy returns OperationalError in case of sqlite and ProgrammingError with postgres. + session.execute(select(1).select_from(tbl)) + except (OperationalError, ProgrammingError): + session.rollback() + tables_absent = True + break + # Tables are created with the v1 schema. + if tables_absent: + if init_catalog_tables: self.create_tables() - return + self._schema_version = "v1" + return + + # 2. Tables exist at this point, so detect the schema version with a probe. + try: + stmt = text("SELECT iceberg_type FROM iceberg_tables LIMIT 0") + session.execute(stmt) + self._schema_version = "v1" + return + except (OperationalError, ProgrammingError): + session.rollback() + + # 3. We only get here if tables exist and iceberg_type column is missing. + if schema_version_prop == "v1": + try: + stmt = text(f"ALTER TABLE {IcebergTables.__tablename__} ADD COLUMN iceberg_type VARCHAR(5)") + session.execute(stmt) + session.commit() + except (OperationalError, ProgrammingError): + session.rollback() + self._schema_version = "v1" + else: + logger.warning( + "SqlCatalog detected a v0 schema (iceberg_type column missing). " + "The catalog will operate in v0 mode. To migrate, set schema_version='v1'." + ) + self._schema_version = "v0" def create_tables(self) -> None: SqlCatalogBaseTable.metadata.create_all(self.engine) @@ -153,6 +202,25 @@ def create_tables(self) -> None: def destroy_tables(self) -> None: SqlCatalogBaseTable.metadata.drop_all(self.engine) + def _create_table_row(self, namespace: str, table_name: str, metadata_location: str | None) -> dict[str, Any]: + row: dict[str, Any] = { + "catalog_name": self.name, + "table_namespace": namespace, + "table_name": table_name, + "metadata_location": metadata_location, + "previous_metadata_location": None, + } + if self._schema_version == "v1": + 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): @@ -224,15 +292,7 @@ def create_table( with Session(self.engine) as session: try: - session.add( - IcebergTables( - catalog_name=self.name, - table_namespace=namespace, - table_name=table_name, - metadata_location=metadata_location, - previous_metadata_location=None, - ) - ) + session.execute(insert(IcebergTables).values(**self._create_table_row(namespace, table_name, metadata_location))) session.commit() except IntegrityError as e: raise TableAlreadyExistsError(f"Table {namespace}.{table_name} already exists") from e @@ -266,15 +326,7 @@ def register_table(self, identifier: str | Identifier, metadata_location: str, o with Session(self.engine) as session: try: - session.add( - IcebergTables( - catalog_name=self.name, - table_namespace=namespace, - table_name=table_name, - metadata_location=metadata_location, - previous_metadata_location=None, - ) - ) + session.execute(insert(IcebergTables).values(**self._create_table_row(namespace, table_name, metadata_location))) session.commit() except IntegrityError as e: raise TableAlreadyExistsError(f"Table {namespace}.{table_name} already exists") from e @@ -306,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) @@ -324,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( @@ -345,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 @@ -376,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: @@ -388,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( @@ -401,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: @@ -492,15 +555,8 @@ def commit_table( else: # table does not exist, create it try: - session.add( - IcebergTables( - catalog_name=self.name, - table_namespace=namespace, - table_name=table_name, - metadata_location=updated_staged_table.metadata_location, - previous_metadata_location=None, - ) - ) + row = self._create_table_row(namespace, table_name, updated_staged_table.metadata_location) + session.execute(insert(IcebergTables).values(**row)) session.commit() except IntegrityError as e: raise TableAlreadyExistsError(f"Table {namespace}.{table_name} already exists") from e @@ -615,7 +671,14 @@ def list_tables(self, namespace: str | Identifier) -> list[Identifier]: raise NoSuchNamespaceError(f"Namespace does not exist: {namespace}") namespace = Catalog.namespace_to_string(namespace) - stmt = select(IcebergTables).where(IcebergTables.catalog_name == self.name, IcebergTables.table_namespace == namespace) + stmt = select(IcebergTables).where( + IcebergTables.catalog_name == self.name, + IcebergTables.table_namespace == namespace, + ) + + 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) return [(Catalog.identifier_to_tuple(table.table_namespace) + (table.table_name,)) for table in result] diff --git a/tests/catalog/test_sql.py b/tests/catalog/test_sql.py index f6846195fe..da9b3397fd 100644 --- a/tests/catalog/test_sql.py +++ b/tests/catalog/test_sql.py @@ -20,7 +20,7 @@ from typing import cast import pytest -from sqlalchemy import Engine, create_engine, inspect +from sqlalchemy import Engine, create_engine, inspect, text from sqlalchemy.exc import ArgumentError from pyiceberg.catalog import load_catalog @@ -33,6 +33,7 @@ ) from pyiceberg.exceptions import ( NoSuchPropertyException, + NoSuchTableError, TableAlreadyExistsError, ) from pyiceberg.schema import Schema @@ -261,3 +262,257 @@ def test_sql_catalog_multiple_close_calls(self, catalog_sqlite: SqlCatalog) -> N # Second close should not raise an exception catalog_sqlite.close() + + +def get_columns(engine: Engine) -> set[str]: + return {c["name"] for c in inspect(engine).get_columns("iceberg_tables")} + + +def test_adds_iceberg_type_column_to_old_schema(warehouse: Path) -> None: + uri = f"sqlite:////{warehouse}/test-migration-add-col" + engine = _create_v0_db(uri) + + # Verify the column does not exist in the old schema + assert "iceberg_type" not in get_columns(engine) + + # Load the catalog with schema_version='v1' to trigger migration + catalog = SqlCatalog( + name="test", + uri=uri, + warehouse=f"file://{warehouse}", + init_catalog_tables="false", + schema_version="v1", + ) + assert "iceberg_type" in get_columns(catalog.engine) + + +def test_idempotent_when_column_already_exists(warehouse: Path) -> None: + catalog = SqlCatalog( + name="test", + uri="sqlite:///:memory:", + warehouse=f"file://{warehouse}", + ) + assert "iceberg_type" in get_columns(catalog.engine) + + # Verify initialization is idempotent when column already exists + catalog._init_catalog() + assert "iceberg_type" in get_columns(catalog.engine) + + +def test_list_tables_filters_by_iceberg_type(warehouse: Path) -> None: + catalog = SqlCatalog( + name="test", + uri="sqlite:///:memory:", + warehouse=f"file://{warehouse}", + ) + catalog.create_namespace("ns") + schema = Schema(NestedField(1, "id", StringType(), required=True)) + catalog.create_table(("ns", "table_V1"), schema) + + # Insert a legacy-schema row (iceberg_type IS NULL); should appear in list_tables. + 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', 'ns', 'table_V0', NULL, NULL, NULL)" + ) + ) + # Insert a non-TABLE row; should NOT appear in list_tables. + conn.execute( + text( + "INSERT INTO iceberg_tables " + "(catalog_name, table_namespace, table_name, metadata_location, previous_metadata_location, iceberg_type) " + "VALUES ('test', 'ns', 'some_view', NULL, NULL, 'VIEW')" + ) + ) + conn.commit() + + tables = [t[-1] for t in catalog.list_tables("ns")] + assert "table_V1" in tables + assert "table_V0" in tables + 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) + + assert "iceberg_type" not in get_columns(engine) + + catalog = SqlCatalog( + name="test", + uri=uri, + warehouse=f"file://{warehouse}", + init_catalog_tables="false", + schema_version="v1", + ) + assert "iceberg_type" in get_columns(catalog.engine) + + +def test_invalid_schema_version_raises(warehouse: Path) -> None: + with pytest.raises(ValueError, match="Invalid schema_version"): + SqlCatalog( + name="test", + uri="sqlite:///:memory:", + warehouse=f"file://{warehouse}", + schema_version="invalid", + ) + + +def test_list_tables_works_on_v0_schema(warehouse: Path) -> None: + """list_tables should work on V0 schemas without iceberg_type column.""" + uri = f"sqlite:////{warehouse}/test-v0-list" + _create_v0_db(uri) + + catalog = SqlCatalog( + name="test", + uri=uri, + warehouse=f"file://{warehouse}", + init_catalog_tables="false", + ) + + # Create namespace directly + with catalog.engine.connect() as conn: + conn.execute( + text( + "INSERT INTO iceberg_namespace_properties " + "(catalog_name, namespace, property_key, property_value) " + "VALUES ('test', 'ns', 'exists', 'true')" + ) + ) + # Insert rows (all are tables on V0, no iceberg_type column to distinguish) + conn.execute( + text( + "INSERT INTO iceberg_tables " + "(catalog_name, table_namespace, table_name, metadata_location, previous_metadata_location) " + "VALUES ('test', 'ns', 'table1', 'loc1', NULL)" + ) + ) + conn.execute( + text( + "INSERT INTO iceberg_tables " + "(catalog_name, table_namespace, table_name, metadata_location, previous_metadata_location) " + "VALUES ('test', 'ns', 'table2', 'loc2', NULL)" + ) + ) + conn.commit() + + # list_tables should work on V0 (no iceberg_type filtering) + tables = [t[-1] for t in catalog.list_tables("ns")] + assert "table1" in tables + assert "table2" in tables + + +def _create_v0_db(uri: str) -> Engine: + """Create a SQLite DB with V0 schema (no iceberg_type column). Returns the engine.""" + engine = create_engine(uri) + with engine.connect() as conn: + conn.execute( + text( + "CREATE TABLE iceberg_tables (" + " catalog_name VARCHAR(255) NOT NULL," + " table_namespace VARCHAR(255) NOT NULL," + " table_name VARCHAR(255) NOT NULL," + " metadata_location VARCHAR(1000)," + " previous_metadata_location VARCHAR(1000)," + " PRIMARY KEY (catalog_name, table_namespace, table_name)" + ")" + ) + ) + conn.execute( + text( + "CREATE TABLE iceberg_namespace_properties (" + " catalog_name VARCHAR(255) NOT NULL," + " namespace VARCHAR(255) NOT NULL," + " property_key VARCHAR(255) NOT NULL," + " property_value VARCHAR(1000) NOT NULL," + " PRIMARY KEY (catalog_name, namespace, property_key)" + ")" + ) + ) + conn.commit() + return engine + + +def _make_v0_catalog(uri: str, warehouse: Path) -> SqlCatalog: + _create_v0_db(uri) + return SqlCatalog("test", uri=uri, warehouse=f"file://{warehouse}", init_catalog_tables="false") + + +def test_namespace_exists_on_v0_schema(warehouse: Path) -> None: + """namespace_exists should not fail on V0 schema (no iceberg_type column).""" + catalog = _make_v0_catalog(f"sqlite:////{warehouse}/test-v0-ns-exists", warehouse) + catalog.create_namespace("ns") + assert catalog.namespace_exists("ns") + assert not catalog.namespace_exists("missing") + + +def test_create_and_load_table_on_v0_schema(warehouse: Path) -> None: + """create_table and load_table should work on V0 schema without iceberg_type column.""" + catalog = _make_v0_catalog(f"sqlite:////{warehouse}/test-v0-create", warehouse) + catalog.create_namespace("ns") + schema = Schema(NestedField(1, "id", StringType(), required=True)) + tbl = catalog.create_table(("ns", "tbl"), schema) + assert tbl.name() == ("ns", "tbl") + + # load_table must work without iceberg_type in the SELECT + loaded = catalog.load_table(("ns", "tbl")) + assert loaded.name() == ("ns", "tbl") + + # iceberg_type column must still be absent from the DB (no migration occurred) + assert "iceberg_type" not in get_columns(catalog.engine)