fix(clickhouse): support multi-gateway projects with catalog-aware engines

ClickHouse's UNSUPPORTED catalog_support caused 2-level FQNs that broke
sqlglot MappingSchema when mixed with 3-level FQNs from Trino or other
catalog-aware gateways. Fix: auto-inject a virtual catalog (gateway name)
for UNSUPPORTED adapters when catalog-aware peers exist, then strip it
before any SQL reaches ClickHouse.

Signed-off-by: mday-io <mdaytn@gmail.com>

Author: mday-io

sqlmesh/core/config/scheduler.py

@@ -138,9 +138,25 @@ def create_plan_evaluator(self, context: GenericContext) -> PlanEvaluator:
 
     def get_default_catalog_per_gateway(self, context: GenericContext) -> t.Dict[str, str]:
         default_catalogs_per_gateway: t.Dict[str, str] = {}
+        unsupported_gateways = []
+
         for gateway, adapter in context.engine_adapters.items():
-            if catalog := adapter.default_catalog:
+            if adapter.catalog_support.is_unsupported:
+                unsupported_gateways.append((gateway, adapter))
+            elif catalog := adapter.default_catalog:
                 default_catalogs_per_gateway[gateway] = catalog
+
+        # When catalog-aware gateways exist, assign the gateway name as a virtual catalog for
+        # catalog-unsupported gateways that opt in (e.g. ClickHouse) so that all models in the
+        # project have a uniform 3-level FQN and the MappingSchema nesting level check passes.
+        # Only adapters that explicitly return True from supports_virtual_catalog() are mutated;
+        # other UNSUPPORTED adapters are left unchanged to avoid silent breakage.
+        if default_catalogs_per_gateway and unsupported_gateways:
+            for gateway, adapter in unsupported_gateways:
+                if adapter.supports_virtual_catalog():
+                    adapter.inject_virtual_catalog(gateway)
+                    default_catalogs_per_gateway[gateway] = gateway
+
         return default_catalogs_per_gateway
 
 

sqlmesh/core/context.py

@@ -487,6 +487,12 @@ 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._snapshot_evaluator = SnapshotEvaluator(
                 {
                     gateway: adapter.with_settings(execute_log_level=logging.INFO)

sqlmesh/core/engine_adapter/base.py

@@ -218,6 +218,29 @@ def comments_enabled(self) -> bool:
     def catalog_support(self) -> CatalogSupport:
         return CatalogSupport.UNSUPPORTED
 
+    def supports_virtual_catalog(self) -> bool:
+        """Return True if this adapter can accept a virtual catalog for multi-gateway nesting alignment.
+
+        When a project mixes catalog-aware gateways (e.g. DuckDB) with catalog-unsupported gateways
+        (e.g. ClickHouse), all adapters need a uniform 3-level FQN so MappingSchema nesting stays
+        consistent. Adapters that return True here opt in to receiving an injected virtual catalog
+        via inject_virtual_catalog(), which causes the set_catalog decorator to strip the catalog
+        from DDL expressions rather than raising UnsupportedCatalogOperationError.
+        """
+        return False
+
+    def inject_virtual_catalog(self, catalog: str) -> None:
+        """Inject a virtual catalog name for multi-gateway nesting alignment.
+
+        Only call this on adapters that return True from supports_virtual_catalog(). After
+        injection, catalog_support should return SINGLE_CATALOG_ONLY so the set_catalog decorator
+        strips the virtual catalog from DDL expressions instead of raising an error.
+        """
+        raise NotImplementedError(
+            f"{self.dialect} does not support virtual catalog injection. "
+            "Override supports_virtual_catalog() to return True and implement inject_virtual_catalog()."
+        )
+
     @cached_property
     def schema_differ(self) -> SchemaDiffer:
         return SchemaDiffer(

sqlmesh/core/engine_adapter/clickhouse.py

@@ -8,6 +8,7 @@
 from sqlmesh.core.engine_adapter.mixins import LogicalMergeMixin
 from sqlmesh.core.engine_adapter.base import EngineAdapterWithIndexSupport
 from sqlmesh.core.engine_adapter.shared import (
+    CatalogSupport,
     DataObject,
     DataObjectType,
     EngineRunMode,
@@ -42,6 +43,22 @@ class ClickhouseEngineAdapter(EngineAdapterWithIndexSupport, LogicalMergeMixin):
     DEFAULT_TABLE_ENGINE = "MergeTree"
     ORDER_BY_TABLE_ENGINE_REGEX = "^.*?MergeTree.*$"
 
+    @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.
+        if self._default_catalog:
+            return CatalogSupport.SINGLE_CATALOG_ONLY
+        return CatalogSupport.UNSUPPORTED
+
+    def supports_virtual_catalog(self) -> bool:
+        return True
+
+    def inject_virtual_catalog(self, catalog: str) -> None:
+        self._default_catalog = catalog
+
     @property
     def engine_run_mode(self) -> EngineRunMode:
         if self._extra_config.get("cloud_mode"):
@@ -172,10 +189,28 @@ def create_schema(
 
         Clickhouse has a two-level naming scheme [database].[table].
         """
+        from sqlmesh.utils.errors import SQLMeshError
+
         properties_copy = properties.copy()
         if self.engine_run_mode.is_cluster:
             properties_copy.append(exp.OnCluster(this=exp.to_identifier(self.cluster)))
 
+        # ClickHouse does not support catalogs. When a virtual catalog has been injected
+        # (self._default_catalog is set), strip it from the schema name. This mirrors the
+        # SINGLE_CATALOG_ONLY branch in the set_catalog decorator, which does not apply here
+        # because this override is not wrapped by @set_catalog().
+        if self._default_catalog:
+            schema_exp = to_schema(schema_name)
+            catalog_name = schema_exp.catalog
+            if catalog_name:
+                if catalog_name != self._default_catalog:
+                    raise SQLMeshError(
+                        f"clickhouse requires that all catalog operations be against a single catalog: "
+                        f"{self._default_catalog}. Provided catalog: {catalog_name}"
+                    )
+                schema_exp.set("catalog", None)
+                schema_name = schema_exp
+
         # can't call super() because it will try to set a catalog
         return self._create_schema(
             schema_name=schema_name,

tests/core/engine_adapter/test_clickhouse.py

@@ -1407,3 +1407,40 @@ def test_exchange_tables(
         'RENAME TABLE "table2" TO "table1"',
         'DROP TABLE IF EXISTS "__temp_table1_abcd"',
     ]
+
+
+def test_virtual_catalog_ddl_stripping(make_mocked_engine_adapter: t.Callable):
+    """After inject_virtual_catalog(), create_schema() with the virtual catalog prefix must strip
+    the catalog and execute without raising, and with a wrong catalog must raise SQLMeshError."""
+    from sqlmesh.utils.errors import SQLMeshError
+
+    adapter = make_mocked_engine_adapter(ClickhouseEngineAdapter)
+
+    assert adapter.supports_virtual_catalog() is True
+    adapter.inject_virtual_catalog("clickhouse_gw")
+
+    # catalog_support must switch to SINGLE_CATALOG_ONLY after injection
+    from sqlmesh.core.engine_adapter.shared import CatalogSupport
+
+    assert adapter.catalog_support == CatalogSupport.SINGLE_CATALOG_ONLY
+    assert adapter._default_catalog == "clickhouse_gw"
+
+    # create_schema with the virtual catalog prefix must strip the catalog and not raise
+    adapter.create_schema("clickhouse_gw.mydb")
+    assert to_sql_calls(adapter) == ['CREATE DATABASE IF NOT EXISTS "mydb"']
+
+    # create_schema with a wrong catalog must raise SQLMeshError
+    with pytest.raises(SQLMeshError, match="clickhouse_gw"):
+        adapter.create_schema("wrong_catalog.mydb")
+
+
+def test_supports_virtual_catalog_returns_true():
+    """ClickhouseEngineAdapter.supports_virtual_catalog() must return True without any connection."""
+    from unittest.mock import MagicMock
+
+    adapter = ClickhouseEngineAdapter(
+        lambda *a, **k: MagicMock(),
+        dialect="clickhouse",
+    )
+    assert adapter.supports_virtual_catalog() is True
+    assert adapter._default_catalog is None

tests/core/test_context.py

@@ -399,6 +399,114 @@ def test_multiple_gateways(tmp_path: Path):
     assert context.dag._sorted == ['"db"."staging"."stg_model"', '"db"."main"."final_model"']
 
 
+def test_multi_gateway_catalog_aware_and_unsupported(tmp_path: Path, mocker):
+    """ClickHouse (catalog UNSUPPORTED) alongside DuckDB (catalog FULL_SUPPORT) must not raise a
+    nesting-level SchemaError when models are loaded.
+
+    Expected behaviour after the fix:
+    - get_default_catalog_per_gateway assigns the gateway name as a virtual catalog for
+      catalog-unsupported gateways when catalog-aware gateways are present.
+    - ClickHouse models end up with a 3-level FQN so the MappingSchema nesting is uniform.
+    - The virtual catalog is stripped from DDL expressions (not raised as an error) because the
+      adapter's catalog_support flips to SINGLE_CATALOG_ONLY when _default_catalog is set.
+    """
+
+    from sqlmesh.core.config.scheduler import BuiltInSchedulerConfig
+    from sqlmesh.core.engine_adapter.clickhouse import ClickhouseEngineAdapter
+    from sqlmesh.core.engine_adapter.duckdb import DuckDBEngineAdapter
+    from sqlmesh.core.engine_adapter.shared import CatalogSupport
+
+    db_path = str(tmp_path / "db.db")
+
+    # Build a real DuckDB adapter for the primary gateway.
+    duck_adapter = DuckDBEngineAdapter(
+        lambda *a, **k: __import__("duckdb").connect(db_path),
+        dialect="duckdb",
+    )
+
+    # Build a minimal ClickHouse adapter stub — no real connection needed.
+    ch_adapter = ClickhouseEngineAdapter(
+        lambda *a, **k: mocker.NonCallableMock(),
+        dialect="clickhouse",
+    )
+
+    # Simulate the context's engine_adapters dict and call the scheduler directly.
+    engine_adapters = {
+        "duckdb_gw": duck_adapter,
+        "clickhouse_gw": ch_adapter,
+    }
+
+    ctx_mock = mocker.MagicMock()
+    ctx_mock.engine_adapters = engine_adapters
+
+    scheduler = BuiltInSchedulerConfig()
+    catalog_per_gw = scheduler.get_default_catalog_per_gateway(ctx_mock)
+
+    # DuckDB gateway must have a real catalog entry.
+    assert "duckdb_gw" in catalog_per_gw
+    # DuckDB's default catalog is the database filename without extension.
+    assert catalog_per_gw["duckdb_gw"] == "db"
+    # ClickHouse gateway must now also have a virtual catalog equal to its gateway name.
+    assert "clickhouse_gw" in catalog_per_gw
+    assert catalog_per_gw["clickhouse_gw"] == "clickhouse_gw"
+
+    # The ClickHouse adapter's _default_catalog must be mutated to the virtual catalog name.
+    assert ch_adapter._default_catalog == "clickhouse_gw"
+
+    # The adapter's catalog_support must now be SINGLE_CATALOG_ONLY (not UNSUPPORTED),
+    # so that the set_catalog decorator strips the virtual catalog instead of raising.
+    assert ch_adapter.catalog_support == CatalogSupport.SINGLE_CATALOG_ONLY
+
+    # Loading models for both gateways must not raise a SchemaError.
+    duckdb_model = load_sql_based_model(
+        parse("MODEL(name main.duckdb_tbl, kind FULL, gateway duckdb_gw);\nSELECT 1 AS col"),
+        default_catalog="db",
+    )
+    ch_model = load_sql_based_model(
+        parse("MODEL(name mydb.ch_tbl, kind FULL, gateway clickhouse_gw);\nSELECT 1 AS col"),
+        default_catalog="clickhouse_gw",
+    )
+
+    # Both models must have 3-level FQNs so MappingSchema nesting is uniform.
+    assert duckdb_model.fqn.count(".") == 2, (
+        f"Expected 3-level FQN for duckdb model, got: {duckdb_model.fqn}"
+    )
+    assert ch_model.fqn.count(".") == 2, f"Expected 3-level FQN for ch model, got: {ch_model.fqn}"
+
+    # Both models loaded into the same MappingSchema must not raise a nesting SchemaError.
+    from sqlglot.schema import MappingSchema
+
+    schema = MappingSchema(normalize=False)
+    schema.add_table(duckdb_model.fqn, duckdb_model.columns_to_types or {})
+    schema.add_table(ch_model.fqn, ch_model.columns_to_types or {})
+
+
+def test_single_gateway_clickhouse_no_virtual_catalog(mocker):
+    """When ClickHouse is the only gateway (no catalog-aware peer), it must NOT receive a virtual
+    catalog.  Models remain 2-level and catalog_support stays UNSUPPORTED."""
+    from sqlmesh.core.config.scheduler import BuiltInSchedulerConfig
+    from sqlmesh.core.engine_adapter.clickhouse import ClickhouseEngineAdapter
+    from sqlmesh.core.engine_adapter.shared import CatalogSupport
+
+    ch_adapter = ClickhouseEngineAdapter(
+        lambda *a, **k: mocker.NonCallableMock(),
+        dialect="clickhouse",
+    )
+
+    ctx_mock = mocker.MagicMock()
+    ctx_mock.engine_adapters = {"clickhouse_gw": ch_adapter}
+
+    scheduler = BuiltInSchedulerConfig()
+    catalog_per_gw = scheduler.get_default_catalog_per_gateway(ctx_mock)
+
+    # With only a catalog-unsupported gateway there must be no entry at all.
+    assert "clickhouse_gw" not in catalog_per_gw
+
+    # The adapter must remain unchanged — no virtual catalog injected.
+    assert ch_adapter._default_catalog is None
+    assert ch_adapter.catalog_support == CatalogSupport.UNSUPPORTED
+
+
 def test_plan_execution_time():
     context = Context(config=Config())
     context.upsert_model(