harden virtual catalog injection and config validation

Author: mday-io

sqlmesh/core/config/connection.py

@@ -2120,6 +2120,15 @@ class ClickhouseConnectionConfig(ConnectionConfig):
 
     _engine_import_validator = _get_engine_import_validator("clickhouse_connect", "clickhouse")
 
+    @field_validator("virtual_catalog")
+    def validate_virtual_catalog(cls, v: t.Optional[str]) -> t.Optional[str]:
+        if v is not None and not v.strip():
+            raise ConfigError(
+                "virtual_catalog cannot be an empty string. "
+                "Omit the field to use the default synthetic prefix (__<gateway_name>__)."
+            )
+        return v
+
     @property
     def _connection_kwargs_keys(self) -> t.Set[str]:
         kwargs = {

sqlmesh/core/context.py

@@ -487,12 +487,7 @@ def engine_adapter(self) -> EngineAdapter:
     @property
     def snapshot_evaluator(self) -> SnapshotEvaluator:
         if not self._snapshot_evaluator:
-            # Ensure virtual catalog injection (via default_catalog_per_gateway) has run before
-            # cloning adapters with with_settings(). Adapters that support virtual catalogs (e.g.
-            # ClickHouse alongside catalog-aware gateways) mutate _default_catalog during
-            # get_default_catalog_per_gateway. with_settings() forwards _default_catalog to the
-            # clone, so the mutation must happen first or the clones will miss the virtual catalog.
-            self.default_catalog_per_gateway  # noqa: B018
+            self._ensure_virtual_catalog_injection()
             self._snapshot_evaluator = SnapshotEvaluator(
                 {
                     gateway: adapter.with_settings(execute_log_level=logging.INFO)
@@ -503,6 +498,15 @@ def snapshot_evaluator(self) -> SnapshotEvaluator:
             )
         return self._snapshot_evaluator
 
+    def _ensure_virtual_catalog_injection(self) -> None:
+        """Ensure virtual catalog injection has run before adapters are cloned for SnapshotEvaluator.
+
+        Injection is a side effect of get_default_catalog_per_gateway. In normal usage it fires
+        earlier (default_catalog is accessed during model loading), but this guard covers the edge
+        case where snapshot_evaluator is accessed directly on a fresh context before any model ops.
+        """
+        _ = self.default_catalog_per_gateway
+
     def execution_context(
         self,
         deployability_index: t.Optional[DeployabilityIndex] = None,

sqlmesh/core/engine_adapter/clickhouse.py

@@ -45,10 +45,9 @@ class ClickhouseEngineAdapter(EngineAdapterWithIndexSupport, LogicalMergeMixin):
 
     @property
     def catalog_support(self) -> CatalogSupport:
-        # When a virtual catalog has been injected via inject_virtual_catalog() (to align
-        # nesting levels with catalog-aware gateways in the same project), treat ClickHouse as
-        # SINGLE_CATALOG_ONLY so the set_catalog decorator strips the virtual catalog from DDL
-        # expressions instead of raising UnsupportedCatalogOperationError.
+        # This property is intentionally dynamic: it transitions from UNSUPPORTED to
+        # SINGLE_CATALOG_ONLY after inject_virtual_catalog() sets _default_catalog. Callers must
+        # not cache the result — always read it live so they see the post-injection state.
         if self._default_catalog:
             return CatalogSupport.SINGLE_CATALOG_ONLY
         return CatalogSupport.UNSUPPORTED

tests/core/engine_adapter/test_clickhouse.py

@@ -1474,3 +1474,19 @@ def test_clickhouse_connection_config_virtual_catalog_extra_engine_config():
         host="localhost", username="user", virtual_catalog="my_catalog"
     )
     assert config._extra_engine_config.get("virtual_catalog") == "my_catalog"
+
+
+def test_clickhouse_connection_config_virtual_catalog_empty_string_rejected():
+    """virtual_catalog: "" is a footgun — the empty string propagates to _default_catalog,
+    which is falsy, so catalog_support stays UNSUPPORTED and the nesting error persists.
+    Reject it at config parse time with a clear message."""
+    import pytest
+
+    from sqlmesh.core.config.connection import ClickhouseConnectionConfig
+    from sqlmesh.utils.errors import ConfigError
+
+    with pytest.raises(ConfigError, match="virtual_catalog cannot be an empty string"):
+        ClickhouseConnectionConfig(host="localhost", username="user", virtual_catalog="")
+
+    with pytest.raises(ConfigError, match="virtual_catalog cannot be an empty string"):
+        ClickhouseConnectionConfig(host="localhost", username="user", virtual_catalog="   ")

tests/core/test_context.py

@@ -541,6 +541,23 @@ def test_multi_gateway_clickhouse_custom_virtual_catalog(tmp_path: Path, mocker)
     assert ch_adapter.catalog_support == CatalogSupport.SINGLE_CATALOG_ONLY
 
 
+def test_snapshot_evaluator_calls_ensure_virtual_catalog_injection(mocker):
+    """snapshot_evaluator must call _ensure_virtual_catalog_injection before cloning adapters.
+
+    This guards the edge case where snapshot_evaluator is the first property accessed on a fresh
+    context — before default_catalog fires during model loading — and ensures virtual catalog
+    injection still happens even in that order.
+    """
+    ctx = Context(config=Config())
+    ctx._snapshot_evaluator = None  # force re-initialization
+
+    inject_spy = mocker.patch.object(ctx, "_ensure_virtual_catalog_injection")
+
+    _ = ctx.snapshot_evaluator
+
+    inject_spy.assert_called_once()
+
+
 def test_plan_execution_time():
     context = Context(config=Config())
     context.upsert_model(