You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by fo...@apache.org on 2024/07/04 21:00:06 UTC
(iceberg-python) branch main updated: Add `pool_pre_ping` param to SQLCatalog (#886)
This is an automated email from the ASF dual-hosted git repository.
fokko pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
The following commit(s) were added to refs/heads/main by this push:
new 2a8b2605 Add `pool_pre_ping` param to SQLCatalog (#886)
2a8b2605 is described below
commit 2a8b26052a781a21a1ad5af0f6e198a3491755af
Author: Eric L (CCCS) <er...@cyber.gc.ca>
AuthorDate: Thu Jul 4 17:00:00 2024 -0400
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
---
mkdocs/docs/configuration.md | 6 ++++++
pyiceberg/catalog/sql.py | 12 +++++++++--
tests/catalog/test_sql.py | 47 ++++++++++++++++++++++++++++++++++++++++++--
3 files changed, 61 insertions(+), 4 deletions(-)
diff --git a/mkdocs/docs/configuration.md b/mkdocs/docs/configuration.md
index f583e773..cf22f8e6 100644
--- a/mkdocs/docs/configuration.md
+++ b/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
diff --git a/pyiceberg/catalog/sql.py b/pyiceberg/catalog/sql.py
index 039c7ddb..a8a36032 100644
--- a/pyiceberg/catalog/sql.py
+++ b/pyiceberg/catalog/sql.py
@@ -64,10 +64,14 @@ from pyiceberg.table import CommitTableRequest, CommitTableResponse, Table
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 @@ class SqlCatalog(MetastoreCatalog):
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()
diff --git a/tests/catalog/test_sql.py b/tests/catalog/test_sql.py
index a32cd047..bb8fcea3 100644
--- a/tests/catalog/test_sql.py
+++ b/tests/catalog/test_sql.py
@@ -28,7 +28,7 @@ from sqlalchemy.exc import ArgumentError, IntegrityError
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.table.sorting import (
)
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",
[