ClickHouse cluster mode runs CREATE ... AS SELECT on every replica #6008

Summary

In self-hosted ClickHouse cluster mode (cluster: set on the gateway, non-Cloud), SQLMesh materializes full-refresh / CTAS models with a single CREATE TABLE … AS SELECT … ON CLUSTER <cluster>. Because the statement is ON CLUSTER, every replica executes the whole statement — SELECT and INSERT included. On an N-replica cluster this causes:

1. N× the source reads/compute for one result (each replica independently re-scans the same upstream source).
2. Deadlock/failure under insert_quorum > 0: the N concurrent quorum inserts never reach a clean quorum, so each blocks for insert_quorum_timeout (default 600s) and then fails with Code 319 UNKNOWN_STATUS_OF_INSERT, rolling the new table back to 0 rows.
3. The long CTAS occupies the distributed-DDL queue for its whole duration, head-of-line-blocking every other ON CLUSTER operation cluster-wide.

ClickHouse Cloud mode already avoids this: it creates the table EMPTY (schema only) then runs a plain single-node INSERT … SELECT that ReplicatedMergeTree replicates. Self-hosted cluster mode should do the same.

Root cause

In sqlmesh/core/engine_adapter/clickhouse.py, ClickhouseEngineAdapter._create_table, both the empty-CTAS split and the follow-up single-node insert are gated on engine_run_mode.is_cloud:

empty_ctas=(self.engine_run_mode.is_cloud and expression is not None),
...
if (self.engine_run_mode.is_cloud and table_kind != "VIEW" and expression and not (<limit 0>)):
    self._insert_append_query(table_name, expression, ...)

So cluster mode falls through to a full CREATE … AS SELECT ON CLUSTER.

Suggested fix

Broaden those two gates to is_cloud or is_cluster, so cluster mode also does empty-CREATE + single-node INSERT … SELECT and lets ReplicatedMergeTree replicate the parts. (A single writer is also fine under insert_quorum > 0 — it just waits for one peer to confirm, the normal path.)

Verified with a local monkeypatch broadening those gates: the build then did NewPart on one replica + DownloadPart on the others, produced the correct row count consistent across all replicas, read the source once, and no longer hit Code 319.

Possibly related: #5785.

Environment

• SQLMesh 0.236.1
• Self-hosted ClickHouse cluster, 1 shard × 4 replicas, ReplicatedMergeTree
• Gateway: type: clickhouse with cluster: set
• Server-side insert_quorum = 2, insert_quorum_timeout = 600000

Reproduction

1. Gateway in cluster mode (cluster: set), models stored as ReplicatedMergeTree, server-side insert_quorum >= 2.
2. Run a full-refresh model whose query reads from an external source.
3. Observe via clusterAllReplicas('…','system.processes'): the SELECT pipeline runs on all replicas; after insert_quorum_timeout the CREATE fails with UNKNOWN_STATUS_OF_INSERT (code 319) and the table is left empty.