Clarify virtual catalog

Author: mday-io

docs/integrations/engines/clickhouse.md

@@ -420,6 +420,45 @@ 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
+
+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 for the first time causes all ClickHouse models to be treated as new versions (their FQNs change to `__{gateway_name}__.db.table`), triggering a full re-materialization on the next `sqlmesh apply`. This is a one-time cost.
+
+### 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 +485,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

@@ -2085,6 +2085,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
@@ -2180,7 +2181,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

@@ -155,7 +155,11 @@ def get_default_catalog_per_gateway(self, context: GenericContext) -> t.Dict[str
             for gateway, adapter in unsupported_gateways:
                 if adapter.supports_virtual_catalog():
                     adapter.inject_virtual_catalog(gateway)
-                    default_catalogs_per_gateway[gateway] = 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/engine_adapter/clickhouse.py

@@ -56,8 +56,9 @@ def catalog_support(self) -> CatalogSupport:
     def supports_virtual_catalog(self) -> bool:
         return True
 
-    def inject_virtual_catalog(self, catalog: str) -> None:
-        self._default_catalog = catalog
+    def inject_virtual_catalog(self, gateway: str) -> None:
+        configured = self._extra_config.get("virtual_catalog")
+        self._default_catalog = f"__{gateway}__" if configured is None else configured
 
     @property
     def engine_run_mode(self) -> EngineRunMode:

tests/core/engine_adapter/test_clickhouse.py

@@ -1423,14 +1423,15 @@ def test_virtual_catalog_ddl_stripping(make_mocked_engine_adapter: t.Callable):
     from sqlmesh.core.engine_adapter.shared import CatalogSupport
 
     assert adapter.catalog_support == CatalogSupport.SINGLE_CATALOG_ONLY
-    assert adapter._default_catalog == "clickhouse_gw"
+    # The default synthetic virtual catalog wraps the gateway name in double underscores.
+    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")
+    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"):
+    with pytest.raises(SQLMeshError, match="__clickhouse_gw__"):
         adapter.create_schema("wrong_catalog.mydb")
 
 
@@ -1444,3 +1445,32 @@ def test_supports_virtual_catalog_returns_true():
     )
     assert adapter.supports_virtual_catalog() is True
     assert adapter._default_catalog is None
+
+
+def test_inject_virtual_catalog_uses_custom_config(make_mocked_engine_adapter: t.Callable):
+    """When virtual_catalog is set in _extra_config, inject_virtual_catalog uses that value
+    instead of the synthetic __gateway_name__ default."""
+    adapter = make_mocked_engine_adapter(
+        ClickhouseEngineAdapter,
+        virtual_catalog="my_custom_catalog",
+    )
+
+    adapter.inject_virtual_catalog("clickhouse_gw")
+
+    # The user-configured value must take precedence over the synthetic default.
+    assert adapter._default_catalog == "my_custom_catalog"
+
+    from sqlmesh.core.engine_adapter.shared import CatalogSupport
+
+    assert adapter.catalog_support == CatalogSupport.SINGLE_CATALOG_ONLY
+
+
+def test_clickhouse_connection_config_virtual_catalog_extra_engine_config():
+    """virtual_catalog set on ClickhouseConnectionConfig must appear in _extra_engine_config
+    so that the value reaches the adapter's _extra_config dict."""
+    from sqlmesh.core.config.connection import ClickhouseConnectionConfig
+
+    config = ClickhouseConnectionConfig(
+        host="localhost", username="user", virtual_catalog="my_catalog"
+    )
+    assert config._extra_engine_config.get("virtual_catalog") == "my_catalog"

tests/core/test_context.py

@@ -446,12 +446,12 @@ def test_multi_gateway_catalog_aware_and_unsupported(tmp_path: Path, mocker):
     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.
+    # ClickHouse gateway must now also have a virtual catalog wrapped in double underscores.
     assert "clickhouse_gw" in catalog_per_gw
-    assert catalog_per_gw["clickhouse_gw"] == "clickhouse_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 ClickHouse adapter's _default_catalog must be mutated to the synthetic virtual catalog.
+    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.
@@ -464,7 +464,7 @@ def test_multi_gateway_catalog_aware_and_unsupported(tmp_path: Path, mocker):
     )
     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",
+        default_catalog="__clickhouse_gw__",
     )
 
     # Both models must have 3-level FQNs so MappingSchema nesting is uniform.
@@ -507,6 +507,40 @@ def test_single_gateway_clickhouse_no_virtual_catalog(mocker):
     assert ch_adapter.catalog_support == CatalogSupport.UNSUPPORTED
 
 
+def test_multi_gateway_clickhouse_custom_virtual_catalog(tmp_path: Path, mocker):
+    """When virtual_catalog is configured on the ClickHouse connection, that value is used as the
+    virtual catalog instead of the synthetic __gateway_name__ default."""
+    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")
+
+    duck_adapter = DuckDBEngineAdapter(
+        lambda *a, **k: __import__("duckdb").connect(db_path),
+        dialect="duckdb",
+    )
+
+    # Pass virtual_catalog via _extra_config (the same path used by ClickhouseConnectionConfig).
+    ch_adapter = ClickhouseEngineAdapter(
+        lambda *a, **k: mocker.NonCallableMock(),
+        dialect="clickhouse",
+        virtual_catalog="my_custom_catalog",
+    )
+
+    ctx_mock = mocker.MagicMock()
+    ctx_mock.engine_adapters = {"duckdb_gw": duck_adapter, "clickhouse_gw": ch_adapter}
+
+    scheduler = BuiltInSchedulerConfig()
+    catalog_per_gw = scheduler.get_default_catalog_per_gateway(ctx_mock)
+
+    # The configured virtual_catalog value must be used, not __clickhouse_gw__.
+    assert catalog_per_gw["clickhouse_gw"] == "my_custom_catalog"
+    assert ch_adapter._default_catalog == "my_custom_catalog"
+    assert ch_adapter.catalog_support == CatalogSupport.SINGLE_CATALOG_ONLY
+
+
 def test_plan_execution_time():
     context = Context(config=Config())
     context.upsert_model(