Support loading custom catalog impl

Author: Jack Ye

mkdocs/docs/configuration.md

@@ -139,7 +139,13 @@ For the FileIO there are several configuration options available:
 
 ## Catalogs
 
-PyIceberg currently has native support for REST, SQL, Hive, Glue and DynamoDB.
+PyIceberg currently has native catalog type support for REST, SQL, Hive, Glue and DynamoDB.
+You can also set the catalog implementation explicitly:
+
+| Key             | Example                      | Description                                                                                      |
+| --------------- | ---------------------------- | ------------------------------------------------------------------------------------------------ |
+| type            | rest                         | Type of catalog, one of `rest`, `sql`, `hive`, `glue`, `dymamodb`. Default to `rest`             |
+| py-catalog-impl | mypackage.mymodule.MyCatalog | Sets the catalog explicitly to an implementation, and will fail explicitly if it can't be loaded |
 
 There are three ways to pass in configuration:
 
@@ -379,6 +385,18 @@ catalog:
 
 <!-- prettier-ignore-end -->
 
+### Custom Catalog Implementations
+
+If you want to load any custom catalog implementation, you can set catalog configurations like the following:
+
+```yaml
+catalog:
+  default:
+    py-catalog-impl: mypackage.mymodule.MyCatalog
+    custom-key1: value1
+    custom-key2: value2
+```
+
 ## Unified AWS Credentials
 
 You can explicitly set the AWS credentials for both Glue/DynamoDB Catalog and S3 FileIO by configuring `client.*` properties. For example:

pyiceberg/catalog/__init__.py

@@ -17,6 +17,7 @@
 
 from __future__ import annotations
 
+import importlib
 import logging
 import re
 import uuid
@@ -77,6 +78,7 @@
 
 TOKEN = "token"
 TYPE = "type"
+PY_CATALOG_IMPL = "py-catalog-impl"
 ICEBERG = "iceberg"
 TABLE_TYPE = "table_type"
 WAREHOUSE_LOCATION = "warehouse"
@@ -230,6 +232,13 @@ def load_catalog(name: Optional[str] = None, **properties: Optional[str]) -> Cat
     env = _ENV_CONFIG.get_catalog_config(name)
     conf: RecursiveDict = merge_config(env or {}, cast(RecursiveDict, properties))
 
+    if catalog_impl := properties.get(PY_CATALOG_IMPL):
+        if catalog := _import_catalog(name, catalog_impl, properties):
+            logger.info("Loaded Catalog: %s", catalog_impl)
+            return catalog
+        else:
+            raise ValueError(f"Could not initialize Catalog: {catalog_impl}")
+
     catalog_type: Optional[CatalogType]
     provided_catalog_type = conf.get(TYPE)
 
@@ -283,6 +292,20 @@ def delete_data_files(io: FileIO, manifests_to_delete: List[ManifestFile]) -> No
                 deleted_files[path] = True
 
 
+def _import_catalog(name: str, catalog_impl: str, properties: Properties) -> Optional[Catalog]:
+    try:
+        path_parts = catalog_impl.split(".")
+        if len(path_parts) < 2:
+            raise ValueError(f"py-catalog-impl should be full path (module.CustomCatalog), got: {catalog_impl}")
+        module_name, class_name = ".".join(path_parts[:-1]), path_parts[-1]
+        module = importlib.import_module(module_name)
+        class_ = getattr(module, class_name)
+        return class_(name, **properties)
+    except ModuleNotFoundError:
+        logger.warning("Could not initialize Catalog: %s", catalog_impl)
+        return None
+
+
 @dataclass
 class PropertiesUpdateSummary:
     removed: List[str]

tests/catalog/test_base.py

@@ -32,11 +32,7 @@
 from pydantic_core import ValidationError
 from pytest_lazyfixture import lazy_fixture
 
-from pyiceberg.catalog import (
-    Catalog,
-    MetastoreCatalog,
-    PropertiesUpdateSummary,
-)
+from pyiceberg.catalog import PY_CATALOG_IMPL, Catalog, MetastoreCatalog, PropertiesUpdateSummary, load_catalog
 from pyiceberg.exceptions import (
     NamespaceAlreadyExistsError,
     NamespaceNotEmptyError,
@@ -295,6 +291,20 @@ def given_catalog_has_a_table(
     )
 
 
+def test_load_catalog_impl_not_full_path() -> None:
+    with pytest.raises(ValueError) as exc_info:
+        load_catalog("catalog", **{PY_CATALOG_IMPL: "CustomCatalog"})
+
+    assert "py-catalog-impl should be full path (module.CustomCatalog), got: CustomCatalog" in str(exc_info.value)
+
+
+def test_load_catalog_impl_does_not_exist() -> None:
+    with pytest.raises(ValueError) as exc_info:
+        load_catalog("catalog", **{PY_CATALOG_IMPL: "pyiceberg.does.not.exist.Catalog"})
+
+    assert "Could not initialize Catalog: pyiceberg.does.not.exist.Catalog" in str(exc_info.value)
+
+
 def test_namespace_from_tuple() -> None:
     # Given
     identifier = ("com", "organization", "department", "my_table")

tests/catalog/test_sql.py

@@ -27,6 +27,7 @@
 
 from pyiceberg.catalog import (
     Catalog,
+    load_catalog,
 )
 from pyiceberg.catalog.sql import DEFAULT_ECHO_VALUE, DEFAULT_POOL_PRE_PING_VALUE, SqlCatalog
 from pyiceberg.exceptions import (
@@ -210,6 +211,18 @@ def test_creation_with_pool_pre_ping_parameter(catalog_name: str, warehouse: Pat
         )
 
 
+def test_creation_from_impl(catalog_name: str, warehouse: Path) -> None:
+    assert isinstance(
+        load_catalog(
+            catalog_name,
+            py_catalog_impl="pyiceberg.catalog.sql.SqlCatalog",
+            uri=f"sqlite:////{warehouse}/sql-catalog.db",
+            warehouse=f"file://{warehouse}",
+        ),
+        SqlCatalog,
+    )
+
+
 @pytest.mark.parametrize(
     "catalog",
     [