From 736cc5c08598f71e74a3e631462ce65393926bdd Mon Sep 17 00:00:00 2001 From: mday-io Date: Wed, 5 Aug 2026 16:45:14 -0400 Subject: [PATCH] fix(clickhouse): strip virtual catalog from view sources Signed-off-by: mday-io --- sqlmesh/core/engine_adapter/clickhouse.py | 42 +++++++++++++++++++ tests/core/engine_adapter/test_clickhouse.py | 43 ++++++++++++++++++++ 2 files changed, 85 insertions(+) diff --git a/sqlmesh/core/engine_adapter/clickhouse.py b/sqlmesh/core/engine_adapter/clickhouse.py index 7bbb7d9870..d1f67e0564 100644 --- a/sqlmesh/core/engine_adapter/clickhouse.py +++ b/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. diff --git a/tests/core/engine_adapter/test_clickhouse.py b/tests/core/engine_adapter/test_clickhouse.py index fcac3bba94..a3dfe0fdda 100644 --- a/tests/core/engine_adapter/test_clickhouse.py +++ b/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"), + )