fix(clickhouse): strip virtual catalog from view sources

sqlmesh/core/engine_adapter/clickhouse.py

@@ -595,6 +595,48 @@ def _create_table(
                 target_columns_to_types or self.columns(table_name),
             )
 
+    def create_view(
+        self,
+        view_name: TableName,
+        query_or_df: QueryOrDF,
+        target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
+        replace: bool = True,
+        materialized: bool = False,
+        materialized_properties: t.Optional[t.Dict[str, t.Any]] = None,
+        table_description: t.Optional[str] = None,
+        column_descriptions: t.Optional[t.Dict[str, str]] = None,
+        view_properties: t.Optional[t.Dict[str, exp.Expr]] = None,
+        source_columns: t.Optional[t.List[str]] = None,
+        **create_kwargs: t.Any,
+    ) -> None:
+        if self._default_catalog and isinstance(query_or_df, exp.Query):
+            from sqlmesh.utils.errors import SQLMeshError
+
+            query_or_df = query_or_df.copy()
+            for table in query_or_df.find_all(exp.Table):
+                if not table.catalog:
+                    continue
+                if table.catalog != self._default_catalog:
+                    raise SQLMeshError(
+                        f"{self.dialect} requires that all catalog operations be against a single "
+                        f"catalog: {self._default_catalog}. Provided catalog: {table.catalog}"
+                    )
+                table.set("catalog", None)
+
+        super().create_view(
+            view_name,
+            query_or_df,
+            target_columns_to_types=target_columns_to_types,
+            replace=replace,
+            materialized=materialized,
+            materialized_properties=materialized_properties,
+            table_description=table_description,
+            column_descriptions=column_descriptions,
+            view_properties=view_properties,
+            source_columns=source_columns,
+            **create_kwargs,
+        )
+
     def _strip_virtual_catalog(self, name: "TableName") -> exp.Table:
         """Strip the virtual catalog prefix from a table name if present.
 

tests/core/engine_adapter/test_clickhouse.py

@@ -1594,3 +1594,46 @@ def test_virtual_catalog_stripped_in_alter_table(make_mocked_engine_adapter: t.C
     assert "mydb" in sql_calls[0]
     assert "my_table" in sql_calls[0]
     assert "ALTER TABLE" in sql_calls[0]
+
+
+def test_virtual_catalog_stripped_from_create_view_source(
+    make_mocked_engine_adapter: t.Callable,
+):
+    adapter = make_mocked_engine_adapter(
+        ClickhouseEngineAdapter,
+        cluster="my_cluster",
+    )
+    adapter.inject_virtual_catalog("clickhouse_gw")
+    query = parse_one("SELECT * FROM __clickhouse_gw__.my_db.my_db__connection_test__1234567890")
+
+    adapter.create_view(
+        "__clickhouse_gw__.my_db.connection_test__dev",
+        query,
+    )
+
+    assert to_sql_calls(adapter) == [
+        'CREATE OR REPLACE VIEW "my_db"."connection_test__dev" '
+        'ON CLUSTER "my_cluster" AS SELECT * FROM '
+        '"my_db"."my_db__connection_test__1234567890"'
+    ]
+    assert query.sql() == (
+        "SELECT * FROM __clickhouse_gw__.my_db.my_db__connection_test__1234567890"
+    )
+
+
+def test_create_view_source_rejects_unexpected_virtual_catalog(
+    make_mocked_engine_adapter: t.Callable,
+):
+    from sqlmesh.utils.errors import SQLMeshError
+
+    adapter = make_mocked_engine_adapter(ClickhouseEngineAdapter)
+    adapter.inject_virtual_catalog("clickhouse_gw")
+
+    with pytest.raises(
+        SQLMeshError,
+        match="Provided catalog: unexpected_catalog",
+    ):
+        adapter.create_view(
+            "__clickhouse_gw__.my_db.connection_test__dev",
+            parse_one("SELECT * FROM unexpected_catalog.my_db.physical_view"),
+        )