Fix ClickHouse cluster CTAS execution

sqlmesh/core/engine_adapter/clickhouse.py

@@ -536,7 +536,7 @@ def _create_table(
     ) -> None:
         """Creates a table in the database.
 
-        Clickhouse Cloud requires doing CTAS in two steps.
+        ClickHouse Cloud and cluster modes require doing CTAS in two steps.
 
         First, we add the `EMPTY` property to the CTAS call to create a table with the proper
         schema, then insert the data with the CTAS query.
@@ -567,16 +567,19 @@ def _create_table(
             table_description,
             column_descriptions,
             table_kind,
-            empty_ctas=(self.engine_run_mode.is_cloud and expression is not None),
+            empty_ctas=(
+                (self.engine_run_mode.is_cloud or self.engine_run_mode.is_cluster)
+                and expression is not None
+            ),
             track_rows_processed=track_rows_processed,
             **kwargs,
         )
 
-        # execute the second INSERT step if on cloud and creating a table
+        # execute the second INSERT step if on cloud or cluster and creating a table
         # - Additional clause is to avoid clickhouse-connect HTTP client bug where CTAS LIMIT 0
         #     returns a success code but malformed response
         if (
-            self.engine_run_mode.is_cloud
+            (self.engine_run_mode.is_cloud or self.engine_run_mode.is_cluster)
             and table_kind != "VIEW"
             and expression
             and not (

tests/core/engine_adapter/test_clickhouse.py

@@ -101,6 +101,33 @@ def test_create_table(adapter: ClickhouseEngineAdapter, mocker):
     ]
 
 
+def test_ctas_in_cluster_mode_uses_empty_ctas_then_single_insert(
+    adapter: ClickhouseEngineAdapter, mocker: MockerFixture
+):
+    """A distributed CREATE must not execute its SELECT once per cluster node."""
+    mocker.patch.object(
+        ClickhouseEngineAdapter,
+        "cluster",
+        new_callable=mocker.PropertyMock(return_value="default"),
+    )
+    mocker.patch.object(
+        ClickhouseEngineAdapter,
+        "engine_run_mode",
+        new_callable=mocker.PropertyMock(return_value=EngineRunMode.CLUSTER),
+    )
+
+    adapter.ctas(
+        "foo",
+        parse_one("SELECT 1 AS a", dialect=adapter.dialect),
+        {"a": exp.DataType.build("Int8", dialect=adapter.dialect)},
+    )
+
+    create_sql, insert_sql = to_sql_calls(adapter)
+    assert 'ON CLUSTER "default"' in create_sql
+    assert " EMPTY AS SELECT" in create_sql
+    assert insert_sql.startswith('INSERT INTO "foo" ("a") SELECT')
+
+
 def test_rename_table(adapter: ClickhouseEngineAdapter, mocker):
     mocker.patch.object(
         ClickhouseEngineAdapter,