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

Signed-off-by: mday-io <mdaytn@gmail.com>
Signed-off-by: Michael Day <michael.day@cloudkitchens.com>

Author: mday-io

docs/integrations/engines/clickhouse.md

@@ -420,6 +420,54 @@ If a model has many records in each partition, you may see additional performanc
 
     Choose a model's time partitioning granularity based on the characteristics of the data it will process, making sure the total number of partitions is 1000 or fewer.
 
+## Multi-gateway setup
+
+ClickHouse does not have a catalog concept — its fully-qualified table names are two-level (`database.table`), not three-level (`catalog.database.table`).
+
+When a SQLMesh project uses ClickHouse alongside a catalog-aware gateway such as Trino or BigQuery, the two gateway types produce FQNs with different nesting depths. SQLMesh's internal schema tracking requires uniform nesting, so it assigns a **virtual catalog** to ClickHouse models at load time.
+
+### How the virtual catalog works
+
+- SQLMesh automatically detects the nesting mismatch and injects a virtual catalog into each ClickHouse adapter when a catalog-aware gateway is also present.
+- ClickHouse models will appear with three-level FQNs in `sqlmesh plan` output and logs — for example, `__ch_prod__.mydb.mytable` for a gateway named `ch_prod`.
+- The virtual catalog prefix is **never sent to ClickHouse**. It is stripped from every DDL and DML statement before execution.
+- When ClickHouse is the only gateway in a project, no virtual catalog is assigned and models remain two-level.
+
+### Adding a second gateway to an existing ClickHouse-only project
+
+!!! warning "Re-materialization required"
+    Adding a catalog-aware gateway (such as Trino or BigQuery) to a project that previously used ClickHouse as the only gateway triggers a **full re-materialization of every ClickHouse model** on the next `sqlmesh apply`. Plan for this before making the change.
+
+If your project previously used ClickHouse as the only gateway, your models were fingerprinted with 2-level FQNs (`db.table`). Adding a catalog-aware gateway causes all ClickHouse models to be treated as new versions (their FQNs change to `__{gateway_name}__.db.table`):
+
+- `FULL` models are recreated once — cost is proportional to the size of each table.
+- `INCREMENTAL_BY_TIME_RANGE` models require a **full historical backfill** from the model's configured start date.
+- The old 2-level model names appear as **Removed** in the plan and will be cleaned up after the environment TTL expires.
+
+This is a one-time cost at the transition point and does not recur. There is no way to skip it — `--forward-only` does not apply because SQLMesh treats the 3-level names as new models, not modified ones.
+
+### Virtual catalog naming
+
+By default, the virtual catalog name is derived from **the gateway name you chose in your config**, wrapped in double underscores — for example, a gateway named `clickhouse` produces `__clickhouse__`, and a gateway named `ch_prod` produces `__ch_prod__`. The double-underscore wrapping makes it visually clear that this is an internal SQLMesh concept, not a real ClickHouse object.
+
+You can override the default name by setting `virtual_catalog` in your ClickHouse connection configuration:
+
+```yaml
+gateways:
+  clickhouse:
+    connection:
+      type: clickhouse
+      host: my-clickhouse-host
+      username: default
+      virtual_catalog: ch_virtual  # optional; defaults to __{gateway_name}__ (e.g. __clickhouse__)
+  trino:
+    connection:
+      type: trino
+      ...
+```
+
+With this configuration, ClickHouse models will appear as `ch_virtual.mydb.mytable` in plan output instead of `__clickhouse__.mydb.mytable`.
+
 ## Local/Built-in Scheduler
 
 **Engine Adapter Type**: `clickhouse`
@@ -446,4 +494,5 @@ If a model has many records in each partition, you may see additional performanc
 | `server_host_name`        | The ClickHouse server hostname as identified by the CN or SNI of its TLS certificate. Set this to avoid SSL errors when connecting through a proxy or tunnel with a different hostname.                                                                                         | string |    N     |
 | `tls_mode`                | Controls advanced TLS behavior. proxy and strict do not invoke ClickHouse mutual TLS connection, but do send client cert and key. mutual assumes ClickHouse mutual TLS auth with a client certificate.                                                                          | string |    N     |
 | `connection_settings`     | Additional [connection settings](https://clickhouse.com/docs/integrations/python#settings-argument)                                                                                                                                                                             |  dict  |    N     |
-| `connection_pool_options` | Additional [options](https://clickhouse.com/docs/integrations/python#customizing-the-http-connection-pool)                                                                                                                                         for the HTTP connection pool |  dict  |    N     |
+| `connection_pool_options` | Additional [options](https://clickhouse.com/docs/integrations/python#customizing-the-http-connection-pool)                                                                                                                                         for the HTTP connection pool |  dict  |    N     |
+| `virtual_catalog`         | Override the virtual catalog name used when ClickHouse runs alongside a catalog-aware gateway (e.g. Trino). Defaults to `__{gateway_name}__`. See [Multi-gateway setup](#multi-gateway-setup) for details.                                                                      | string |    N     |

sqlmesh/core/config/connection.py

@@ -2086,6 +2086,7 @@ class ClickhouseConnectionConfig(ConnectionConfig):
     password: t.Optional[str] = None
     port: t.Optional[int] = None
     cluster: t.Optional[str] = None
+    virtual_catalog: t.Optional[str] = None
     connect_timeout: int = 10
     send_receive_timeout: int = 300
     query_limit: int = 0
@@ -2120,6 +2121,19 @@ 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>__)."
+            )
+        if v is not None and "." in v:
+            raise ConfigError(
+                f"virtual_catalog must be a single identifier with no dots (got: {v!r})"
+            )
+        return v
+
     @property
     def _connection_kwargs_keys(self) -> t.Set[str]:
         kwargs = {
@@ -2181,7 +2195,11 @@ def cloud_mode(self) -> bool:
 
     @property
     def _extra_engine_config(self) -> t.Dict[str, t.Any]:
-        return {"cluster": self.cluster, "cloud_mode": self.cloud_mode}
+        return {
+            "cluster": self.cluster,
+            "cloud_mode": self.cloud_mode,
+            "virtual_catalog": self.virtual_catalog,
+        }
 
     @property
     def _static_connection_kwargs(self) -> t.Dict[str, t.Any]:

sqlmesh/core/config/scheduler.py

@@ -138,9 +138,29 @@ 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)
+                    # Read the actual virtual catalog name back from the adapter — it may differ
+                    # from the gateway name if the user configured a custom virtual_catalog value.
+                    # inject_virtual_catalog() always sets _default_catalog so default_catalog
+                    # cannot return None at this point.
+                    default_catalogs_per_gateway[gateway] = adapter.default_catalog  # type: ignore[assignment]
+
         return default_catalogs_per_gateway
 
 

sqlmesh/core/context.py

@@ -492,6 +492,7 @@ def engine_adapter(self) -> EngineAdapter:
     @property
     def snapshot_evaluator(self) -> SnapshotEvaluator:
         if not self._snapshot_evaluator:
+            self._ensure_virtual_catalog_injection()
             self._snapshot_evaluator = SnapshotEvaluator(
                 {
                     gateway: adapter.with_settings(execute_log_level=logging.INFO)
@@ -502,6 +503,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,
@@ -1440,6 +1450,8 @@ def plan(
 
         plan = plan_builder.build()
 
+        self._warn_if_virtual_catalog_rematerialization(plan)
+
         if no_auto_categorization or plan.uncategorized:
             # Prompts are required if the auto categorization is disabled
             # or if there are any uncategorized snapshots in the plan
@@ -2746,6 +2758,61 @@ def _run_plan_tests(self, skip_tests: bool = False) -> t.Optional[ModelTextTestR
             return result
         return None
 
+    def _warn_if_virtual_catalog_rematerialization(self, plan: "Plan") -> None:
+        """Warn when ClickHouse models appear as new snapshots solely because a virtual catalog
+        prefix was added to their FQNs after a catalog-aware gateway joined the project.
+
+        This situation causes every previously-applied ClickHouse model to be treated as brand-new
+        by SQLMesh, triggering full re-materialization and historical backfills. Emitting a warning
+        before the plan is displayed gives users a chance to understand the cost before applying.
+        """
+        from sqlglot import exp
+
+        # Collect the set of old 2-level snapshot names from the current environment so we can
+        # detect which new 3-level names are renames rather than genuinely new models.
+        old_names: t.Set[str] = set()
+        for s_id in plan.context_diff.removed_snapshots:
+            old_names.add(s_id.name)
+        for name in plan.context_diff.snapshots_by_name:
+            old_names.add(name)
+
+        affected: t.List[t.Tuple[str, str]] = []  # (new_3level_name, old_2level_name)
+
+        for gateway, adapter in self.engine_adapters.items():
+            if not adapter.supports_virtual_catalog() or not adapter._default_catalog:
+                continue
+            virtual_catalog = adapter._default_catalog
+
+            for snapshot in plan.new_snapshots:
+                table = exp.to_table(snapshot.name)
+                if table.catalog != virtual_catalog:
+                    continue
+                # Reconstruct the 2-level name that would have been used before injection.
+                old_name = f"{table.db}.{table.name}"
+                if old_name in old_names:
+                    affected.append((snapshot.name, old_name))
+
+        if not affected:
+            return
+
+        max_display = 10
+        model_lines = "\n".join(
+            f"  - {new_name}  (was: {old_name})" for new_name, old_name in affected[:max_display]
+        )
+        if len(affected) > max_display:
+            model_lines += f"\n  ... and {len(affected) - max_display} more"
+
+        self.console.log_warning(
+            "ClickHouse models are being re-materialized due to virtual catalog FQN change.\n\n"
+            "The following ClickHouse models appear as new because their fully-qualified\n"
+            "names changed from 2-level (db.table) to 3-level (__gateway__.db.table):\n\n"
+            f"{model_lines}\n\n"
+            "FULL models will be recreated once. INCREMENTAL_BY_TIME_RANGE models will\n"
+            "require a full historical backfill from their configured start date.\n\n"
+            "This is a one-time cost when first adding a catalog-aware gateway to an\n"
+            "existing ClickHouse project. To proceed, run `sqlmesh apply`."
+        )
+
     @property
     def _model_tables(self) -> t.Dict[str, str]:
         """Mapping of model name to physical table name.

sqlmesh/core/engine_adapter/base.py

@@ -223,6 +223,31 @@ 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, gateway: str) -> None:
+        """Inject a gateway name to configure the adapter's virtual catalog.
+
+        The adapter determines the final catalog name from the gateway name (e.g. ClickHouse
+        wraps it as __{gateway}__). 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(