Add pool_pre_ping param to SQLCatalog (#886)

* Add pool_pre_ping param to SQLCatalog and fix echo parsing logic

* Remove print statements

* Add unit tests for catalog params

* PR review changes

* Slight code optimization

* Adopting code simplification from PR review

* Convert parameterized tests to for loop

Author: cccs-eric

mkdocs/docs/configuration.md

@@ -222,6 +222,12 @@ catalog:
     uri: sqlite:////tmp/pyiceberg.db
 ```
 
+| Key           | Example                                                      | Default | Description                                                                                                                                                                                    |
+| ------------- | ------------------------------------------------------------ | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| uri           | postgresql+psycopg2://username:password@localhost/mydatabase |         | SQLAlchemy backend URL for the catalog database (see [documentation for URL format](https://docs.sqlalchemy.org/en/20/core/engines.html#backend-specific-urls))                                |
+| echo          | true                                                         | false   | SQLAlchemy engine [echo param](https://docs.sqlalchemy.org/en/20/core/engines.html#sqlalchemy.create_engine.params.echo) to log all statements to the default log handler                      |
+| pool_pre_ping | true                                                         | false   | SQLAlchemy engine [pool_pre_ping param](https://docs.sqlalchemy.org/en/20/core/engines.html#sqlalchemy.create_engine.params.pool_pre_ping) to test connections for liveness upon each checkout |
+
 ## Hive Catalog
 
 ```yaml

pyiceberg/catalog/sql.py

@@ -64,10 +64,14 @@
 from pyiceberg.table.metadata import new_table_metadata
 from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
 from pyiceberg.typedef import EMPTY_DICT, Identifier, Properties
+from pyiceberg.types import strtobool
 
 if TYPE_CHECKING:
     import pyarrow as pa
 
+DEFAULT_ECHO_VALUE = "false"
+DEFAULT_POOL_PRE_PING_VALUE = "false"
+
 
 class SqlCatalogBaseTable(MappedAsDataclass, DeclarativeBase):
     pass
@@ -110,8 +114,12 @@ def __init__(self, name: str, **properties: str):
 
         if not (uri_prop := self.properties.get("uri")):
             raise NoSuchPropertyException("SQL connection URI is required")
-        echo = bool(self.properties.get("echo", False))
-        self.engine = create_engine(uri_prop, echo=echo)
+
+        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))
+
+        self.engine = create_engine(uri_prop, echo=echo, pool_pre_ping=pool_pre_ping)
 
         self._ensure_tables_exist()
 

tests/catalog/test_sql.py

@@ -28,7 +28,7 @@
 from pyiceberg.catalog import (
     Catalog,
 )
-from pyiceberg.catalog.sql import SqlCatalog
+from pyiceberg.catalog.sql import DEFAULT_ECHO_VALUE, DEFAULT_POOL_PRE_PING_VALUE, SqlCatalog
 from pyiceberg.exceptions import (
     CommitFailedException,
     NamespaceAlreadyExistsError,
@@ -52,7 +52,7 @@
 )
 from pyiceberg.transforms import IdentityTransform
 from pyiceberg.typedef import Identifier
-from pyiceberg.types import IntegerType
+from pyiceberg.types import IntegerType, strtobool
 
 
 @pytest.fixture(scope="module")
@@ -168,6 +168,49 @@ def test_creation_with_unsupported_uri(catalog_name: str) -> None:
         SqlCatalog(catalog_name, uri="unsupported:xxx")
 
 
+def test_creation_with_echo_parameter(catalog_name: str, warehouse: Path) -> None:
+    # echo_param, expected_echo_value
+    test_cases = [(None, strtobool(DEFAULT_ECHO_VALUE)), ("debug", "debug"), ("true", True), ("false", False)]
+
+    for echo_param, expected_echo_value in test_cases:
+        props = {
+            "uri": f"sqlite:////{warehouse}/sql-catalog.db",
+            "warehouse": f"file://{warehouse}",
+        }
+        # None is for default value
+        if echo_param is not None:
+            props["echo"] = echo_param
+        catalog = SqlCatalog(catalog_name, **props)
+        assert catalog.engine._echo == expected_echo_value, (
+            f"Assertion failed: expected echo value {expected_echo_value}, "
+            f"but got {catalog.engine._echo}. For echo_param={echo_param}"
+        )
+
+
+def test_creation_with_pool_pre_ping_parameter(catalog_name: str, warehouse: Path) -> None:
+    # pool_pre_ping_param, expected_pool_pre_ping_value
+    test_cases = [
+        (None, strtobool(DEFAULT_POOL_PRE_PING_VALUE)),
+        ("true", True),
+        ("false", False),
+    ]
+
+    for pool_pre_ping_param, expected_pool_pre_ping_value in test_cases:
+        props = {
+            "uri": f"sqlite:////{warehouse}/sql-catalog.db",
+            "warehouse": f"file://{warehouse}",
+        }
+        # None is for default value
+        if pool_pre_ping_param is not None:
+            props["pool_pre_ping"] = pool_pre_ping_param
+
+        catalog = SqlCatalog(catalog_name, **props)
+        assert catalog.engine.pool._pre_ping == expected_pool_pre_ping_value, (
+            f"Assertion failed: expected pool_pre_ping value {expected_pool_pre_ping_value}, "
+            f"but got {catalog.engine.pool._pre_ping}. For pool_pre_ping_param={pool_pre_ping_param}"
+        )
+
+
 @pytest.mark.parametrize(
     "catalog",
     [