From 59206fb92887045759ce02026866721d4df20872 Mon Sep 17 00:00:00 2001 From: isc-patrick Date: Tue, 10 Sep 2024 09:56:59 -0400 Subject: [PATCH 1/5] Remove unnecessary _ensure_tables_exist method as this is already the default behavior of Metadata.create_all() --- pyiceberg/catalog/sql.py | 18 +++--------------- 1 file changed, 3 insertions(+), 15 deletions(-) diff --git a/pyiceberg/catalog/sql.py b/pyiceberg/catalog/sql.py index 78883e35bb..40595e5684 100644 --- a/pyiceberg/catalog/sql.py +++ b/pyiceberg/catalog/sql.py @@ -33,7 +33,7 @@ union, update, ) -from sqlalchemy.exc import IntegrityError, NoResultFound, OperationalError, ProgrammingError +from sqlalchemy.exc import IntegrityError, NoResultFound from sqlalchemy.orm import ( DeclarativeBase, Mapped, @@ -126,20 +126,8 @@ def __init__(self, name: str, **properties: str): self.engine = create_engine(uri_prop, echo=echo, pool_pre_ping=pool_pre_ping) - self._ensure_tables_exist() - - def _ensure_tables_exist(self) -> None: - with Session(self.engine) as session: - for table in [IcebergTables, IcebergNamespaceProperties]: - stmt = select(1).select_from(table) - try: - session.scalar(stmt) - except ( - OperationalError, - ProgrammingError, - ): # sqlalchemy returns OperationalError in case of sqlite and ProgrammingError with postgres. - self.create_tables() - return + # Only creates tables that do not exist + self.create_tables() def create_tables(self) -> None: SqlCatalogBaseTable.metadata.create_all(self.engine) From 00efac1bb001e683bcd97bfe7a5bb1df245abed2 Mon Sep 17 00:00:00 2001 From: isc-patrick Date: Thu, 12 Sep 2024 07:55:59 -0400 Subject: [PATCH 2/5] Added tests for creating Catalog tables when no, some or all tables already exist --- tests/catalog/test_sql.py | 100 +++++++++++++++++++++++++++++++++++++- 1 file changed, 99 insertions(+), 1 deletion(-) diff --git a/tests/catalog/test_sql.py b/tests/catalog/test_sql.py index e6c9a5b01b..01e2c3fd2b 100644 --- a/tests/catalog/test_sql.py +++ b/tests/catalog/test_sql.py @@ -24,12 +24,22 @@ from pydantic_core import ValidationError from pytest_lazyfixture import lazy_fixture from sqlalchemy.exc import ArgumentError, IntegrityError +from sqlalchemy import create_engine, inspect, Engine, Inspector from pyiceberg.catalog import ( Catalog, load_catalog, + ) -from pyiceberg.catalog.sql import DEFAULT_ECHO_VALUE, DEFAULT_POOL_PRE_PING_VALUE, SqlCatalog +from pyiceberg.catalog.sql import ( + DEFAULT_ECHO_VALUE, + DEFAULT_POOL_PRE_PING_VALUE, + SqlCatalog, + IcebergTables, + IcebergNamespaceProperties, + SqlCatalogBaseTable +) + from pyiceberg.exceptions import ( CommitFailedException, NamespaceAlreadyExistsError, @@ -131,6 +141,9 @@ def catalog_sqlite(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog, yield catalog catalog.destroy_tables() +@pytest.fixture(scope="function") +def alchemy_engine(): + return create_engine("sqlite:///:memory:") @pytest.fixture(scope="module") def catalog_sqlite_without_rowcount(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog, None, None]: @@ -225,6 +238,91 @@ def test_creation_from_impl(catalog_name: str, warehouse: Path) -> None: ) +def confirm_no_tables_exist(alchemy_engine: Engine): + + inspector = inspect(alchemy_engine) + catalog_tables = [c.__tablename__ for c in SqlCatalogBaseTable.__subclasses__()] + any_table_exists = any(t for t in inspector.get_table_names() if t in catalog_tables) + if any_table_exists: + pytest.raises(TableAlreadyExistsError, "Tables exist, but should not have been created yet") + + return inspector, catalog_tables + + +def confirm_all_tables_exist(inspector: Inspector, catalog_tables: list, catalog: SqlCatalog): + + # Make sure both tables exist after catalog created + all_tables_exists = all([t for t in inspector.get_table_names() if t in catalog_tables]) + assert all_tables_exists, "Tables should have been created" + assert isinstance(catalog, SqlCatalog), "Catalog should be a SQLCatalog" + + +def test_creation_when_no_tables_exist(alchemy_engine: Engine, catalog_name: str, warehouse: Path) -> None: + + # Make sure none of the tables exist + inspector, catalog_tables = confirm_no_tables_exist(alchemy_engine) + + # Create catalog + catalog = load_catalog( + catalog_name, + **{ + "type": "sql", + "uri": "sqlite:///:memory:", + "warehouse": f"file://{warehouse}", + }, + ) + + # Make sure both tables exist after catalog created + confirm_all_tables_exist(inspector, catalog_tables, catalog) + + +def test_creation_when_one_tables_exists(alchemy_engine: Engine, catalog_name: str, warehouse: Path) -> None: + + # Make sure none of the tables exist + inspector, catalog_tables = confirm_no_tables_exist(alchemy_engine) + + # Create one table + IcebergTables.__table__.create(bind = alchemy_engine) + inspector.clear_cache() + assert IcebergTables.__tablename__ in [t for t in inspector.get_table_names() if t in catalog_tables] + + catalog = load_catalog( + catalog_name, + **{ + "type": "sql", + "uri": "sqlite:///:memory:", + "warehouse": f"file://{warehouse}", + }, + ) + + # Make sure both tables exist after catalog created + confirm_all_tables_exist(inspector, catalog_tables, catalog) + + +def test_creation_when_all_tables_exists(alchemy_engine: Engine, catalog_name: str, warehouse: Path) -> None: + + # Make sure none of the tables exist + inspector, catalog_tables = confirm_no_tables_exist(alchemy_engine) + + # Create all tables + SqlCatalogBaseTable.metadata.create_all(bind = alchemy_engine) + inspector.clear_cache() + for c in catalog_tables: + assert c in [t for t in inspector.get_table_names() if t in catalog_tables] + + catalog = load_catalog( + catalog_name, + **{ + "type": "sql", + "uri": "sqlite:///:memory:", + "warehouse": f"file://{warehouse}", + }, + ) + + # Make sure both tables exist after catalog created + confirm_all_tables_exist(inspector, catalog_tables, catalog) + + @pytest.mark.parametrize( "catalog", [ From 58f9e538d51d4397aeeae69ea557d41492277c66 Mon Sep 17 00:00:00 2001 From: isc-patrick Date: Mon, 16 Sep 2024 12:12:37 -0400 Subject: [PATCH 3/5] add init_catalog_tables flag to SQLCatalog --- pyiceberg/catalog/sql.py | 7 ++- tests/catalog/test_sql.py | 119 +++++++++++++++++++------------------- 2 files changed, 65 insertions(+), 61 deletions(-) diff --git a/pyiceberg/catalog/sql.py b/pyiceberg/catalog/sql.py index 40595e5684..c9884681d0 100644 --- a/pyiceberg/catalog/sql.py +++ b/pyiceberg/catalog/sql.py @@ -76,6 +76,7 @@ DEFAULT_ECHO_VALUE = "false" DEFAULT_POOL_PRE_PING_VALUE = "false" +DEFAULT_INIT_CATALOG_TABLES = "true" class SqlCatalogBaseTable(MappedAsDataclass, DeclarativeBase): @@ -123,11 +124,13 @@ 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) - # Only creates tables that do not exist - self.create_tables() + if init_catalog_tables: + # Only creates tables that do not exist + self.create_tables() def create_tables(self) -> None: SqlCatalogBaseTable.metadata.create_all(self.engine) diff --git a/tests/catalog/test_sql.py b/tests/catalog/test_sql.py index 01e2c3fd2b..32b9b74888 100644 --- a/tests/catalog/test_sql.py +++ b/tests/catalog/test_sql.py @@ -23,23 +23,20 @@ import pytest from pydantic_core import ValidationError from pytest_lazyfixture import lazy_fixture +from sqlalchemy import Engine, Inspector, create_engine, inspect from sqlalchemy.exc import ArgumentError, IntegrityError -from sqlalchemy import create_engine, inspect, Engine, Inspector from pyiceberg.catalog import ( Catalog, load_catalog, - ) from pyiceberg.catalog.sql import ( - DEFAULT_ECHO_VALUE, - DEFAULT_POOL_PRE_PING_VALUE, - SqlCatalog, + DEFAULT_ECHO_VALUE, + DEFAULT_POOL_PRE_PING_VALUE, IcebergTables, - IcebergNamespaceProperties, - SqlCatalogBaseTable + SqlCatalog, + SqlCatalogBaseTable, ) - from pyiceberg.exceptions import ( CommitFailedException, NamespaceAlreadyExistsError, @@ -141,9 +138,11 @@ def catalog_sqlite(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog, yield catalog catalog.destroy_tables() -@pytest.fixture(scope="function") -def alchemy_engine(): - return create_engine("sqlite:///:memory:") + +@pytest.fixture(scope="module") +def alchemy_engine(warehouse: Path) -> Engine: + return create_engine(f"sqlite:////{warehouse}/sql-catalog.db") + @pytest.fixture(scope="module") def catalog_sqlite_without_rowcount(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog, None, None]: @@ -238,87 +237,89 @@ def test_creation_from_impl(catalog_name: str, warehouse: Path) -> None: ) -def confirm_no_tables_exist(alchemy_engine: Engine): - +def confirm_no_tables_exist(alchemy_engine: Engine) -> list[str]: inspector = inspect(alchemy_engine) - catalog_tables = [c.__tablename__ for c in SqlCatalogBaseTable.__subclasses__()] + for c in SqlCatalogBaseTable.__subclasses__(): + if inspector.has_table(c.__tablename__): + c.__table__.drop(alchemy_engine) + + catalog_tables = [c.__tablename__ for c in SqlCatalogBaseTable.__subclasses__()] any_table_exists = any(t for t in inspector.get_table_names() if t in catalog_tables) if any_table_exists: pytest.raises(TableAlreadyExistsError, "Tables exist, but should not have been created yet") - - return inspector, catalog_tables + return catalog_tables -def confirm_all_tables_exist(inspector: Inspector, catalog_tables: list, catalog: SqlCatalog): +def confirm_all_tables_exist(inspector: Inspector, catalog_tables: list[str], catalog: Catalog) -> None: # Make sure both tables exist after catalog created - all_tables_exists = all([t for t in inspector.get_table_names() if t in catalog_tables]) - assert all_tables_exists, "Tables should have been created" + all_tables_exists = True # all([False for t in catalog_tables if t not in inspector.get_table_names() ]) + for t in catalog_tables: + if t not in inspector.get_table_names(): + all_tables_exists = False + assert isinstance(catalog, SqlCatalog), "Catalog should be a SQLCatalog" + assert all_tables_exists, "Tables should have been created" + + +def load_catalog_for_catalog_table_creation(catalog_name: str, warehouse: Path) -> Catalog: + catalog = load_catalog( + catalog_name, + type="sql", + uri=f"sqlite:////{warehouse}/sql-catalog.db", + warehouse=f"file://{warehouse}", + init_catalog_tables="true", + ) + + return catalog def test_creation_when_no_tables_exist(alchemy_engine: Engine, catalog_name: str, warehouse: Path) -> None: - # Make sure none of the tables exist - inspector, catalog_tables = confirm_no_tables_exist(alchemy_engine) - + catalog_tables = confirm_no_tables_exist(alchemy_engine) + # Create catalog - catalog = load_catalog( - catalog_name, - **{ - "type": "sql", - "uri": "sqlite:///:memory:", - "warehouse": f"file://{warehouse}", - }, - ) - + catalog = load_catalog_for_catalog_table_creation(catalog_name=catalog_name, warehouse=warehouse) + inspector = inspect(alchemy_engine) + # Make sure both tables exist after catalog created confirm_all_tables_exist(inspector, catalog_tables, catalog) def test_creation_when_one_tables_exists(alchemy_engine: Engine, catalog_name: str, warehouse: Path) -> None: - # Make sure none of the tables exist - inspector, catalog_tables = confirm_no_tables_exist(alchemy_engine) - + catalog_tables = confirm_no_tables_exist(alchemy_engine) + + inspector = inspect(alchemy_engine) + # Create one table - IcebergTables.__table__.create(bind = alchemy_engine) + IcebergTables.__table__.create(bind=alchemy_engine) inspector.clear_cache() assert IcebergTables.__tablename__ in [t for t in inspector.get_table_names() if t in catalog_tables] - - catalog = load_catalog( - catalog_name, - **{ - "type": "sql", - "uri": "sqlite:///:memory:", - "warehouse": f"file://{warehouse}", - }, - ) - + + # Create catalog + catalog = load_catalog_for_catalog_table_creation(catalog_name=catalog_name, warehouse=warehouse) + inspector.clear_cache() + # Make sure both tables exist after catalog created confirm_all_tables_exist(inspector, catalog_tables, catalog) def test_creation_when_all_tables_exists(alchemy_engine: Engine, catalog_name: str, warehouse: Path) -> None: - # Make sure none of the tables exist - inspector, catalog_tables = confirm_no_tables_exist(alchemy_engine) - + catalog_tables = confirm_no_tables_exist(alchemy_engine) + + inspector = inspect(alchemy_engine) + # Create all tables - SqlCatalogBaseTable.metadata.create_all(bind = alchemy_engine) + SqlCatalogBaseTable.metadata.create_all(bind=alchemy_engine) inspector.clear_cache() for c in catalog_tables: assert c in [t for t in inspector.get_table_names() if t in catalog_tables] - - catalog = load_catalog( - catalog_name, - **{ - "type": "sql", - "uri": "sqlite:///:memory:", - "warehouse": f"file://{warehouse}", - }, - ) - + + # Create catalog + catalog = load_catalog_for_catalog_table_creation(catalog_name=catalog_name, warehouse=warehouse) + # Make sure both tables exist after catalog created confirm_all_tables_exist(inspector, catalog_tables, catalog) From c0451751489b587646b6f2f0d52c28d6ae6cead4 Mon Sep 17 00:00:00 2001 From: isc-patrick Date: Mon, 16 Sep 2024 14:02:05 -0400 Subject: [PATCH 4/5] add _ensure_tables_exists back until postgres integration tests completed --- pyiceberg/catalog/sql.py | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/pyiceberg/catalog/sql.py b/pyiceberg/catalog/sql.py index c9884681d0..65e5e6bd1e 100644 --- a/pyiceberg/catalog/sql.py +++ b/pyiceberg/catalog/sql.py @@ -33,7 +33,7 @@ union, update, ) -from sqlalchemy.exc import IntegrityError, NoResultFound +from sqlalchemy.exc import IntegrityError, NoResultFound, OperationalError, ProgrammingError from sqlalchemy.orm import ( DeclarativeBase, Mapped, @@ -129,8 +129,20 @@ def __init__(self, name: str, **properties: str): self.engine = create_engine(uri_prop, echo=echo, pool_pre_ping=pool_pre_ping) if init_catalog_tables: - # Only creates tables that do not exist - self.create_tables() + self._ensure_tables_exist() + + def _ensure_tables_exist(self) -> None: + with Session(self.engine) as session: + for table in [IcebergTables, IcebergNamespaceProperties]: + stmt = select(1).select_from(table) + try: + session.scalar(stmt) + except ( + OperationalError, + ProgrammingError, + ): # sqlalchemy returns OperationalError in case of sqlite and ProgrammingError with postgres. + self.create_tables() + return def create_tables(self) -> None: SqlCatalogBaseTable.metadata.create_all(self.engine) From 11d9f3ac748932584fc126158abe5427eebf8deb Mon Sep 17 00:00:00 2001 From: isc-patrick Date: Tue, 17 Sep 2024 10:53:26 -0400 Subject: [PATCH 5/5] fixed tests, added flag to docs --- mkdocs/docs/configuration.md | 4 +- tests/catalog/test_sql.py | 91 +++++++++++++++--------------------- 2 files changed, 40 insertions(+), 55 deletions(-) diff --git a/mkdocs/docs/configuration.md b/mkdocs/docs/configuration.md index 8495437f23..1269e34a83 100644 --- a/mkdocs/docs/configuration.md +++ b/mkdocs/docs/configuration.md @@ -231,7 +231,7 @@ catalog: ### SQL Catalog -The SQL catalog requires a database for its backend. PyIceberg supports PostgreSQL and SQLite through psycopg2. The database connection has to be configured using the `uri` property. See SQLAlchemy's [documentation for URL format](https://docs.sqlalchemy.org/en/20/core/engines.html#backend-specific-urls): +The SQL catalog requires a database for its backend. PyIceberg supports PostgreSQL and SQLite through psycopg2. The database connection has to be configured using the `uri` property. The init_catalog_tables is optional and defaults to True. If it is set to False, the catalog tables will not be created when the SQLCatalog is initialized. See SQLAlchemy's [documentation for URL format](https://docs.sqlalchemy.org/en/20/core/engines.html#backend-specific-urls): For PostgreSQL: @@ -240,6 +240,7 @@ catalog: default: type: sql uri: postgresql+psycopg2://username:password@localhost/mydatabase + init_catalog_tables: false ``` In the case of SQLite: @@ -256,6 +257,7 @@ catalog: default: type: sql uri: sqlite:////tmp/pyiceberg.db + init_catalog_tables: false ``` | Key | Example | Default | Description | diff --git a/tests/catalog/test_sql.py b/tests/catalog/test_sql.py index 32b9b74888..d3815fec04 100644 --- a/tests/catalog/test_sql.py +++ b/tests/catalog/test_sql.py @@ -17,13 +17,13 @@ import os from pathlib import Path -from typing import Any, Generator, List +from typing import Any, Generator, List, cast import pyarrow as pa import pytest from pydantic_core import ValidationError from pytest_lazyfixture import lazy_fixture -from sqlalchemy import Engine, Inspector, create_engine, inspect +from sqlalchemy import Engine, create_engine, inspect from sqlalchemy.exc import ArgumentError, IntegrityError from pyiceberg.catalog import ( @@ -61,6 +61,8 @@ from pyiceberg.typedef import Identifier from pyiceberg.types import IntegerType, strtobool +CATALOG_TABLES = [c.__tablename__ for c in SqlCatalogBaseTable.__subclasses__()] + @pytest.fixture(scope="module") def catalog_name() -> str: @@ -140,8 +142,13 @@ def catalog_sqlite(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog, @pytest.fixture(scope="module") -def alchemy_engine(warehouse: Path) -> Engine: - return create_engine(f"sqlite:////{warehouse}/sql-catalog.db") +def catalog_uri(warehouse: Path) -> str: + return f"sqlite:////{warehouse}/sql-catalog.db" + + +@pytest.fixture(scope="module") +def alchemy_engine(catalog_uri: str) -> Engine: + return create_engine(catalog_uri) @pytest.fixture(scope="module") @@ -237,91 +244,67 @@ def test_creation_from_impl(catalog_name: str, warehouse: Path) -> None: ) -def confirm_no_tables_exist(alchemy_engine: Engine) -> list[str]: +def confirm_no_tables_exist(alchemy_engine: Engine) -> None: inspector = inspect(alchemy_engine) for c in SqlCatalogBaseTable.__subclasses__(): if inspector.has_table(c.__tablename__): c.__table__.drop(alchemy_engine) - catalog_tables = [c.__tablename__ for c in SqlCatalogBaseTable.__subclasses__()] - any_table_exists = any(t for t in inspector.get_table_names() if t in catalog_tables) + any_table_exists = any(t for t in inspector.get_table_names() if t in CATALOG_TABLES) if any_table_exists: pytest.raises(TableAlreadyExistsError, "Tables exist, but should not have been created yet") - return catalog_tables - -def confirm_all_tables_exist(inspector: Inspector, catalog_tables: list[str], catalog: Catalog) -> None: - # Make sure both tables exist after catalog created - all_tables_exists = True # all([False for t in catalog_tables if t not in inspector.get_table_names() ]) - for t in catalog_tables: - if t not in inspector.get_table_names(): +def confirm_all_tables_exist(catalog: SqlCatalog) -> None: + all_tables_exists = True + for t in CATALOG_TABLES: + if t not in inspect(catalog.engine).get_table_names(): all_tables_exists = False assert isinstance(catalog, SqlCatalog), "Catalog should be a SQLCatalog" assert all_tables_exists, "Tables should have been created" -def load_catalog_for_catalog_table_creation(catalog_name: str, warehouse: Path) -> Catalog: +def load_catalog_for_catalog_table_creation(catalog_name: str, catalog_uri: str) -> SqlCatalog: catalog = load_catalog( catalog_name, type="sql", - uri=f"sqlite:////{warehouse}/sql-catalog.db", - warehouse=f"file://{warehouse}", + uri=catalog_uri, init_catalog_tables="true", ) - return catalog - - -def test_creation_when_no_tables_exist(alchemy_engine: Engine, catalog_name: str, warehouse: Path) -> None: - # Make sure none of the tables exist - catalog_tables = confirm_no_tables_exist(alchemy_engine) - - # Create catalog - catalog = load_catalog_for_catalog_table_creation(catalog_name=catalog_name, warehouse=warehouse) - inspector = inspect(alchemy_engine) + return cast(SqlCatalog, catalog) - # Make sure both tables exist after catalog created - confirm_all_tables_exist(inspector, catalog_tables, catalog) +def test_creation_when_no_tables_exist(alchemy_engine: Engine, catalog_name: str, catalog_uri: str) -> None: + confirm_no_tables_exist(alchemy_engine) + catalog = load_catalog_for_catalog_table_creation(catalog_name=catalog_name, catalog_uri=catalog_uri) + confirm_all_tables_exist(catalog) -def test_creation_when_one_tables_exists(alchemy_engine: Engine, catalog_name: str, warehouse: Path) -> None: - # Make sure none of the tables exist - catalog_tables = confirm_no_tables_exist(alchemy_engine) - inspector = inspect(alchemy_engine) +def test_creation_when_one_tables_exists(alchemy_engine: Engine, catalog_name: str, catalog_uri: str) -> None: + confirm_no_tables_exist(alchemy_engine) # Create one table + inspector = inspect(alchemy_engine) IcebergTables.__table__.create(bind=alchemy_engine) - inspector.clear_cache() - assert IcebergTables.__tablename__ in [t for t in inspector.get_table_names() if t in catalog_tables] - - # Create catalog - catalog = load_catalog_for_catalog_table_creation(catalog_name=catalog_name, warehouse=warehouse) - inspector.clear_cache() + assert IcebergTables.__tablename__ in [t for t in inspector.get_table_names() if t in CATALOG_TABLES] - # Make sure both tables exist after catalog created - confirm_all_tables_exist(inspector, catalog_tables, catalog) + catalog = load_catalog_for_catalog_table_creation(catalog_name=catalog_name, catalog_uri=catalog_uri) + confirm_all_tables_exist(catalog) -def test_creation_when_all_tables_exists(alchemy_engine: Engine, catalog_name: str, warehouse: Path) -> None: - # Make sure none of the tables exist - catalog_tables = confirm_no_tables_exist(alchemy_engine) - - inspector = inspect(alchemy_engine) +def test_creation_when_all_tables_exists(alchemy_engine: Engine, catalog_name: str, catalog_uri: str) -> None: + confirm_no_tables_exist(alchemy_engine) # Create all tables + inspector = inspect(alchemy_engine) SqlCatalogBaseTable.metadata.create_all(bind=alchemy_engine) - inspector.clear_cache() - for c in catalog_tables: - assert c in [t for t in inspector.get_table_names() if t in catalog_tables] - - # Create catalog - catalog = load_catalog_for_catalog_table_creation(catalog_name=catalog_name, warehouse=warehouse) + for c in CATALOG_TABLES: + assert c in [t for t in inspector.get_table_names() if t in CATALOG_TABLES] - # Make sure both tables exist after catalog created - confirm_all_tables_exist(inspector, catalog_tables, catalog) + catalog = load_catalog_for_catalog_table_creation(catalog_name=catalog_name, catalog_uri=catalog_uri) + confirm_all_tables_exist(catalog) @pytest.mark.parametrize(