fix(clickhouse): complete virtual-catalog stripping across all overrides

Address reviewer feedback on PR #5826:

- Add _strip_virtual_catalog helper to ClickhouseEngineAdapter; apply to
  delete_from, insert_overwrite_by_partition, and alter_table overrides
  that previously bypassed both the @set_catalog decorator and the
  create_schema manual strip — preventing __gateway__ prefix leaking
  into DELETE/INSERT OVERWRITE/ALTER TABLE SQL sent to ClickHouse
- Deduplicate catalog-stripping logic in create_schema via the helper
- Rename inject_virtual_catalog param catalog→gateway in base class and
  align docstring (adapter decides the final catalog name)
- Reject virtual_catalog values containing '.' to prevent 4-level FQN
- Add unit tests verifying SQL output is free of the virtual catalog
  prefix for each newly covered method
- Add integration test exercising get_default_catalog_per_gateway +
  FQN uniformity + create_schema stripping end-to-end
- Fix clickhouse.md trailing newline

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

Author: mday-io

docs/integrations/engines/clickhouse.md

@@ -486,4 +486,4 @@ With this configuration, ClickHouse models will appear as `ch_virtual.mydb.mytab
 | `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     |
-| `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     |
+| `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

@@ -2127,6 +2127,10 @@ def validate_virtual_catalog(cls, v: t.Optional[str]) -> t.Optional[str]:
                 "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

sqlmesh/core/engine_adapter/base.py

@@ -229,12 +229,14 @@ def supports_virtual_catalog(self) -> bool:
         """
         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.
+    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. "

sqlmesh/core/engine_adapter/clickhouse.py

@@ -208,8 +208,7 @@ def create_schema(
                         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
+                schema_name = self._strip_virtual_catalog(schema_exp)
 
         # can't call super() because it will try to set a catalog
         return self._create_schema(
@@ -481,6 +480,7 @@ def insert_overwrite_by_partition(
         target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
         source_columns: t.Optional[t.List[str]] = None,
     ) -> None:
+        table_name = self._strip_virtual_catalog(table_name)
         source_queries, target_columns_to_types = self._get_source_queries_and_columns_to_types(
             query_or_df,
             target_columns_to_types,
@@ -595,6 +595,20 @@ def _create_table(
                 target_columns_to_types or self.columns(table_name),
             )
 
+    def _strip_virtual_catalog(self, name: "TableName") -> exp.Table:
+        """Strip the virtual catalog prefix from a table name if present.
+
+        When a virtual catalog has been injected, ClickHouse table names carry a
+        synthetic catalog prefix (e.g. ``__gw__``) so they match the 3-level FQN
+        depth of catalog-aware peers.  This helper removes that prefix before any
+        SQL is sent to the wire, since ClickHouse only supports a two-level
+        ``[database].[table]`` naming scheme.
+        """
+        table = exp.to_table(name)
+        if self._default_catalog and table.catalog == self._default_catalog:
+            table.set("catalog", None)
+        return table
+
     def _exchange_tables(
         self,
         old_table_name: TableName,
@@ -633,7 +647,7 @@ def _rename_table(
         self.execute(f"RENAME TABLE {old_table_sql} TO {new_table_sql}{self._on_cluster_sql()}")
 
     def delete_from(self, table_name: TableName, where: t.Union[str, exp.Expr]) -> None:
-        delete_expr = exp.delete(table_name, where)
+        delete_expr = exp.delete(self._strip_virtual_catalog(table_name), where)
         if self.engine_run_mode.is_cluster:
             delete_expr.set("cluster", exp.OnCluster(this=exp.to_identifier(self.cluster)))
         self.execute(delete_expr)
@@ -649,6 +663,9 @@ def alter_table(
             for alter_expression in [
                 x.expression if isinstance(x, TableAlterOperation) else x for x in alter_expressions
             ]:
+                if self._default_catalog and isinstance(alter_expression.this, exp.Table):
+                    if alter_expression.this.catalog == self._default_catalog:
+                        alter_expression.this.set("catalog", None)
                 if self.engine_run_mode.is_cluster:
                     alter_expression.set(
                         "cluster", exp.OnCluster(this=exp.to_identifier(self.cluster))

tests/core/engine_adapter/test_clickhouse.py

@@ -1490,3 +1490,88 @@ def test_clickhouse_connection_config_virtual_catalog_empty_string_rejected():
 
     with pytest.raises(ConfigError, match="virtual_catalog cannot be an empty string"):
         ClickhouseConnectionConfig(host="localhost", username="user", virtual_catalog="   ")
+
+
+def test_virtual_catalog_stripped_in_delete_from(make_mocked_engine_adapter: t.Callable):
+    """delete_from() must strip the virtual catalog prefix before building the DELETE expression."""
+    adapter = make_mocked_engine_adapter(ClickhouseEngineAdapter)
+    adapter.inject_virtual_catalog("ch_gw")
+    assert adapter._default_catalog == "__ch_gw__"
+
+    adapter.delete_from("__ch_gw__.mydb.my_table", "a = 1")
+
+    sql_calls = to_sql_calls(adapter)
+    assert len(sql_calls) == 1
+    assert "__ch_gw__" not in sql_calls[0]
+    assert "mydb" in sql_calls[0]
+    assert "my_table" in sql_calls[0]
+    assert "DELETE FROM" in sql_calls[0]
+
+
+def test_virtual_catalog_stripped_in_insert_overwrite_by_partition(
+    make_mocked_engine_adapter: t.Callable, mocker: MockerFixture
+):
+    """insert_overwrite_by_partition() must strip the virtual catalog prefix before any SQL is sent."""
+    adapter = make_mocked_engine_adapter(ClickhouseEngineAdapter)
+    adapter.inject_virtual_catalog("ch_gw")
+    assert adapter._default_catalog == "__ch_gw__"
+
+    # Patch _insert_overwrite_by_condition so we can inspect how table_name was passed.
+    overwrite_mock = mocker.patch.object(adapter, "_insert_overwrite_by_condition")
+    # Also patch _get_source_queries_and_columns_to_types to avoid needing a real query.
+    source_query_mock = mocker.patch.object(
+        adapter,
+        "_get_source_queries_and_columns_to_types",
+        return_value=([], {}),
+    )
+
+    adapter.insert_overwrite_by_partition(
+        "__ch_gw__.mydb.my_table",
+        parse_one("SELECT 1 AS col"),
+        partitioned_by=[exp.column("ds")],
+    )
+
+    # The table_name passed to _insert_overwrite_by_condition must not contain the virtual catalog.
+    assert overwrite_mock.called
+    table_name_arg = overwrite_mock.call_args[0][0]
+    table_name_sql = (
+        table_name_arg.sql("clickhouse")
+        if isinstance(table_name_arg, exp.Expression)
+        else str(table_name_arg)
+    )
+    assert "__ch_gw__" not in table_name_sql
+
+    # The target_table passed to _get_source_queries_and_columns_to_types must also be stripped.
+    assert source_query_mock.called
+    target_table_kwarg = source_query_mock.call_args[1].get(
+        "target_table",
+        source_query_mock.call_args[0][2] if len(source_query_mock.call_args[0]) > 2 else None,
+    )
+    if target_table_kwarg is not None:
+        target_sql = (
+            target_table_kwarg.sql("clickhouse")
+            if isinstance(target_table_kwarg, exp.Expression)
+            else str(target_table_kwarg)
+        )
+        assert "__ch_gw__" not in target_sql
+
+
+def test_virtual_catalog_stripped_in_alter_table(make_mocked_engine_adapter: t.Callable):
+    """alter_table() must strip the virtual catalog prefix from each ALTER TABLE statement."""
+    adapter = make_mocked_engine_adapter(ClickhouseEngineAdapter)
+    adapter.inject_virtual_catalog("ch_gw")
+    assert adapter._default_catalog == "__ch_gw__"
+
+    alter_expr = exp.Alter(
+        this=exp.to_table("__ch_gw__.mydb.my_table"),
+        kind="TABLE",
+        actions=[exp.Drop(this=exp.to_column("col_a"), kind="COLUMN")],
+    )
+    adapter.alter_table([alter_expr])
+
+    sql_calls = to_sql_calls(adapter)
+    assert len(sql_calls) == 1
+    assert "__ch_gw__" not in sql_calls[0]
+    assert "mydb" in sql_calls[0]
+    assert "my_table" in sql_calls[0]
+    assert "ALTER TABLE" in sql_calls[0]

tests/core/test_context.py

@@ -468,10 +468,13 @@ def test_multi_gateway_catalog_aware_and_unsupported(tmp_path: Path, mocker):
     )
 
     # Both models must have 3-level FQNs so MappingSchema nesting is uniform.
+    # count(".") == 2 means 3 parts (catalog.db.table), i.e. a 3-level FQN.
     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}"
+    assert ch_model.fqn.count(".") == 2, (
+        f"Expected 3-level FQN for ch model, got: {ch_model.fqn}"
+    )  # 3 parts = 2 dots
 
     # Both models loaded into the same MappingSchema must not raise a nesting SchemaError.
     from sqlglot.schema import MappingSchema
@@ -558,6 +561,122 @@ def test_snapshot_evaluator_calls_ensure_virtual_catalog_injection(mocker):
     inject_spy.assert_called_once()
 
 
+@pytest.mark.fast
+def test_multi_gateway_virtual_catalog_create_schema_strips_prefix(tmp_path: Path, mocker):
+    """Integration test: create_schema with a 3-level virtual-catalog FQN must strip the synthetic
+    catalog prefix before sending DDL to ClickHouse.
+
+    Flow exercised:
+    1. get_default_catalog_per_gateway detects a catalog-aware gateway (DuckDB) alongside
+       a catalog-unsupported gateway (ClickHouse) and calls inject_virtual_catalog().
+    2. The ClickHouse adapter's _default_catalog is set to ``__clickhouse_gw__``.
+    3. A ClickHouse model loaded with that virtual catalog gets a 3-level FQN.
+    4. When create_schema is called with the 3-level schema name the virtual catalog prefix
+       is stripped, so the SQL that reaches the wire uses only a 2-level name.
+    5. The DuckDB adapter's _default_catalog is NOT set to a synthetic value.
+    """
+    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",
+    )
+
+    # ClickHouse adapter with a mocked connection — no real server needed.
+    ch_adapter = ClickhouseEngineAdapter(
+        lambda *a, **k: mocker.NonCallableMock(),
+        dialect="clickhouse",
+    )
+
+    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)
+
+    # --- Phase 1: virtual catalog injection assertions ---
+
+    # DuckDB gateway must carry a real catalog entry.
+    assert "duckdb_gw" in catalog_per_gw
+
+    # ClickHouse gateway must receive the synthetic ``__clickhouse_gw__`` virtual catalog.
+    assert catalog_per_gw["clickhouse_gw"] == "__clickhouse_gw__"
+
+    # The ClickHouse adapter's _default_catalog must be mutated.
+    assert ch_adapter._default_catalog == "__clickhouse_gw__"
+
+    # catalog_support must flip to SINGLE_CATALOG_ONLY so the set_catalog decorator strips
+    # the virtual catalog instead of raising when DDL is executed.
+    assert ch_adapter.catalog_support == CatalogSupport.SINGLE_CATALOG_ONLY
+
+    # DuckDB adapter must be untouched — it already has real catalog support.
+    assert duck_adapter._default_catalog != "__duckdb_gw__"
+
+    # --- Phase 2: FQN uniformity ---
+
+    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__",
+    )
+    duckdb_model = load_sql_based_model(
+        parse("MODEL(name main.duckdb_tbl, kind FULL, gateway duckdb_gw);\nSELECT 1 AS col"),
+        default_catalog=catalog_per_gw["duckdb_gw"],
+    )
+
+    # Both models must have 3-level FQNs (catalog.db.table → 2 dots) so MappingSchema nesting
+    # is uniform and does not raise a SchemaError.
+    assert ch_model.fqn.count(".") == 2, (
+        f"Expected 3-level FQN for ClickHouse model, got: {ch_model.fqn}"
+    )
+    assert duckdb_model.fqn.count(".") == 2, (
+        f"Expected 3-level FQN for DuckDB model, got: {duckdb_model.fqn}"
+    )
+
+    from sqlglot.schema import MappingSchema
+
+    schema = MappingSchema(normalize=False)
+    schema.add_table(ch_model.fqn, ch_model.columns_to_types or {})
+    schema.add_table(duckdb_model.fqn, duckdb_model.columns_to_types or {})
+
+    # --- Phase 3: create_schema strips the virtual catalog prefix ---
+
+    # Spy on _create_schema to inspect what schema name is passed after stripping.
+    create_schema_calls: t.List[str] = []
+
+    def _capture_create_schema(
+        schema_name,
+        ignore_if_exists,
+        warn_on_error,
+        properties,
+        kind,
+    ):
+        create_schema_calls.append(
+            schema_name.sql(dialect="clickhouse")
+            if hasattr(schema_name, "sql")
+            else str(schema_name)
+        )
+
+    mocker.patch.object(ch_adapter, "_create_schema", side_effect=_capture_create_schema)
+
+    # Call create_schema with the 3-level virtual-catalog-prefixed schema name.
+    ch_adapter.create_schema("__clickhouse_gw__.mydb")
+
+    assert len(create_schema_calls) == 1, "Expected exactly one _create_schema call"
+    passed_schema = create_schema_calls[0]
+    # The virtual catalog prefix must NOT appear in the SQL sent to the wire.
+    assert "__clickhouse_gw__" not in passed_schema, (
+        f"Virtual catalog prefix should be stripped before reaching _create_schema, got: {passed_schema!r}"
+    )
+
+
 def test_plan_execution_time():
     context = Context(config=Config())
     context.upsert_model(