Add flag to allow disabling creation of catalog tables (#1155)

* Remove unnecessary _ensure_tables_exist method as this is already the default behavior of Metadata.create_all()

* Added tests for creating Catalog tables when no, some or all tables already exist

* add init_catalog_tables flag to SQLCatalog

* add _ensure_tables_exists back until postgres integration tests completed

* fixed tests, added flag to docs

Author: isc-psulin

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                                                                                                                                                                                    |

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,10 +124,12 @@ 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._ensure_tables_exist()
+        if init_catalog_tables:
+            self._ensure_tables_exist()
 
     def _ensure_tables_exist(self) -> None:
         with Session(self.engine) as session:

tests/catalog/test_sql.py

@@ -17,19 +17,26 @@
 
 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, create_engine, inspect
 from sqlalchemy.exc import ArgumentError, IntegrityError
 
 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,
+    IcebergTables,
+    SqlCatalog,
+    SqlCatalogBaseTable,
+)
 from pyiceberg.exceptions import (
     CommitFailedException,
     NamespaceAlreadyExistsError,
@@ -54,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:
@@ -132,6 +141,16 @@ def catalog_sqlite(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog,
     catalog.destroy_tables()
 
 
+@pytest.fixture(scope="module")
+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")
 def catalog_sqlite_without_rowcount(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog, None, None]:
     props = {
@@ -225,6 +244,69 @@ def test_creation_from_impl(catalog_name: str, warehouse: Path) -> None:
     )
 
 
+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)
+
+    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")
+
+
+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, catalog_uri: str) -> SqlCatalog:
+    catalog = load_catalog(
+        catalog_name,
+        type="sql",
+        uri=catalog_uri,
+        init_catalog_tables="true",
+    )
+
+    return cast(SqlCatalog, 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, catalog_uri: str) -> None:
+    confirm_no_tables_exist(alchemy_engine)
+
+    # Create one table
+    inspector = inspect(alchemy_engine)
+    IcebergTables.__table__.create(bind=alchemy_engine)
+    assert IcebergTables.__tablename__ in [t for t in inspector.get_table_names() if t in CATALOG_TABLES]
+
+    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, catalog_uri: str) -> None:
+    confirm_no_tables_exist(alchemy_engine)
+
+    # Create all tables
+    inspector = inspect(alchemy_engine)
+    SqlCatalogBaseTable.metadata.create_all(bind=alchemy_engine)
+    for c in CATALOG_TABLES:
+        assert c in [t for t in inspector.get_table_names() if t in CATALOG_TABLES]
+
+    catalog = load_catalog_for_catalog_table_creation(catalog_name=catalog_name, catalog_uri=catalog_uri)
+    confirm_all_tables_exist(catalog)
+
+
 @pytest.mark.parametrize(
     "catalog",
     [