feat(clickhouse): allow dynamic schema (#32610)

superset/db_engine_specs/clickhouse.py

@@ -20,6 +20,7 @@ import logging
 import re
 from datetime import datetime
 from typing import Any, cast, TYPE_CHECKING
+from urllib import parse
 
 from flask import current_app
 from flask_babel import gettext as __
@@ -267,6 +268,8 @@ class ClickHouseConnectEngineSpec(BasicParametersMixin, ClickHouseEngineSpec):
     parameters_schema = ClickHouseParametersSchema()
     encryption_parameters = {"secure": "true"}
 
+    supports_dynamic_schema = True
+
     @classmethod
     def get_dbapi_exception_mapping(cls) -> dict[type[Exception], type[Exception]]:
         return {}
@@ -414,3 +417,15 @@ class ClickHouseConnectEngineSpec(BasicParametersMixin, ClickHouseEngineSpec):
         :return: Conditionally mutated label
         """
         return f"{label}_{md5_sha_from_str(label)[:6]}"
+
+    @classmethod
+    def adjust_engine_params(
+        cls,
+        uri: URL,
+        connect_args: dict[str, Any],
+        catalog: str | None = None,
+        schema: str | None = None,
+    ) -> tuple[URL, dict[str, Any]]:
+        if schema:
+            uri = uri.set(database=parse.quote(schema, safe=""))
+        return uri, connect_args

tests/unit_tests/db_engine_specs/test_clickhouse.py

@@ -20,6 +20,7 @@ from typing import Any, Optional
 from unittest.mock import Mock
 
 import pytest
+from sqlalchemy.engine.url import make_url
 from sqlalchemy.types import (
     Boolean,
     Date,
@@ -225,3 +226,26 @@ def test_connect_make_label_compatible(column_name: str, expected_result: str) -
 
     label = spec.make_label_compatible(column_name)
     assert label == expected_result
+
+
+@pytest.mark.parametrize(
+    "schema, expected_result",
+    [
+        (None, "clickhousedb+connect://localhost:443/__default__"),
+        (
+            "new_schema",
+            "clickhousedb+connect://localhost:443/new_schema",
+        ),
+    ],
+)
+def test_adjust_engine_params_fully_qualified(
+    schema: str, expected_result: str
+) -> None:
+    from superset.db_engine_specs.clickhouse import (
+        ClickHouseConnectEngineSpec as spec,  # noqa: N813
+    )
+
+    url = make_url("clickhousedb+connect://localhost:443/__default__")
+
+    uri = spec.adjust_engine_params(url, {}, None, schema)[0]
+    assert str(uri) == expected_result