feat(clickhouse): add cloud_mode connection option

docs/integrations/engines/clickhouse.md

@@ -59,6 +59,8 @@ ClickHouse Cloud automates ClickHouse's cluster controls, which sometimes constr
 
 Aside from those constraints, ClickHouse Cloud mode is similar to single server mode - you run standard SQL commands/queries, and ClickHouse Cloud executes them.
 
+SQLMesh enables this mode automatically when your connection's `host` contains `clickhouse.cloud`. Some self-hosted deployments have the same constraints without a ClickHouse Cloud host name - the [replicated database engine](https://clickhouse.com/docs/en/engines/database-engines/replicated), for example, also cannot create a table with a `SELECT` command in one step. Set the `cloud_mode` connection option to `true` to enable the mode for those deployments, or to `false` to disable it on a ClickHouse Cloud host.
+
 ## Permissions
 
 In the default SQLMesh configuration, users must have sufficient permissions to create new ClickHouse databases.
@@ -480,6 +482,7 @@ With this configuration, ClickHouse models will appear as `ch_virtual.mydb.mytab
 | `password`                | ClickHouse user password                                                                                                                                                                                                                                                        | string |    N     |
 | `port`                    | The ClickHouse HTTP or HTTPS port (Default: `8123`)                                                                                                                                                                                                                             |  int   |    N     |
 | `cluster`                 | ClickHouse cluster name                                                                                                                                                                                                                                                         | string |    N     |
+| `cloud_mode`              | Force [ClickHouse Cloud mode](#clickhouse-cloud-mode) on or off. Defaults to detecting it from the host name, so most projects do not need to set this. Set to `true` for self-hosted deployments that share Cloud's constraints, such as those using the replicated database engine. |  bool  |    N     |
 | `connect_timeout`         | Connection timeout in seconds (Default: `10`)                                                                                                                                                                                                                                   |  int   |    N     |
 | `send_receive_timeout`    | Send/receive timeout in seconds (Default: `300`)                                                                                                                                                                                                                                |  int   |    N     |
 | `query_limit`             | Query result limit (Default: `0` - no limit)                                                                                                                                                                                                                                    |  int   |    N     |

sqlmesh/core/config/connection.py

@@ -2240,6 +2240,7 @@ class ClickhouseConnectionConfig(ConnectionConfig):
     password: t.Optional[str] = None
     port: t.Optional[int] = None
     cluster: t.Optional[str] = None
+    cloud_mode: t.Optional[bool] = None
     virtual_catalog: t.Optional[str] = None
     connect_timeout: int = 10
     send_receive_timeout: int = 300
@@ -2345,14 +2346,24 @@ def _connection_factory(self) -> t.Callable:
         return partial(connect, pool_mgr=pool_mgr)
 
     @property
-    def cloud_mode(self) -> bool:
+    def _resolved_cloud_mode(self) -> bool:
+        """Whether to use ClickHouse Cloud mode.
+
+        An explicit `cloud_mode` setting always wins. When it is unset we fall back to
+        detecting ClickHouse Cloud from the host name, which preserves the behavior for
+        existing configurations. The setting lets self-hosted deployments that share
+        Cloud's constraints (e.g. the replicated database engine, which also cannot run
+        `CREATE TABLE ... AS SELECT`) opt in without renaming their host.
+        """
+        if self.cloud_mode is not None:
+            return self.cloud_mode
         return "clickhouse.cloud" in self.host
 
     @property
     def _extra_engine_config(self) -> t.Dict[str, t.Any]:
         return {
             "cluster": self.cluster,
-            "cloud_mode": self.cloud_mode,
+            "cloud_mode": self._resolved_cloud_mode,
             "virtual_catalog": self.virtual_catalog,
         }
 
@@ -2376,7 +2387,7 @@ def _static_connection_kwargs(self) -> t.Dict[str, t.Any]:
         settings["mutations_sync"] = "2"
         #  insert_distributed_sync = 1: "INSERT operation succeeds only after all the data is saved on all shards"
         settings["insert_distributed_sync"] = "1"
-        if self.cluster or self.cloud_mode:
+        if self.cluster or self._resolved_cloud_mode:
             # database_replicated_enforce_synchronous_settings = 1:
             #   - "Enforces synchronous waiting for some queries"
             #   - https://github.com/ClickHouse/ClickHouse/blob/ccaa8d03a9351efc16625340268b9caffa8a22ba/src/Core/Settings.h#L709

tests/core/test_connection_config.py

@@ -29,6 +29,7 @@
     _connection_config_validator,
     _get_engine_import_validator,
 )
+from sqlmesh.core.engine_adapter.shared import EngineRunMode
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.pydantic import PydanticModel
 
@@ -1287,6 +1288,65 @@ def test_clickhouse(make_config):
     assert not config3._static_connection_kwargs["compress"]
 
 
+def test_clickhouse_cloud_mode(make_config):
+    """Cloud mode falls back to host detection but can be set explicitly.
+
+    These assertions deliberately go through `_extra_engine_config` and the adapter's
+    run mode rather than reading the `cloud_mode` field back. The field only matters
+    insofar as it reaches the adapter and selects the two-step CTAS that Cloud requires,
+    so asserting on the field alone would still pass if that wiring were broken.
+    """
+
+    def run_mode(**kwargs) -> EngineRunMode:
+        config = make_config(type="clickhouse", username="default", **kwargs)
+        assert isinstance(config, ClickhouseConnectionConfig)
+        adapter = config.create_engine_adapter()
+        # this dict is the contract between the connection config and the adapter
+        assert config._extra_engine_config["cloud_mode"] is adapter.engine_run_mode.is_cloud
+        return adapter.engine_run_mode
+
+    # An existing Cloud project that has never set `cloud_mode` must keep working.
+    assert run_mode(host="foo.clickhouse.cloud").is_cloud
+
+    # A self-hosted host must not be silently treated as Cloud.
+    assert run_mode(host="localhost").is_standalone
+
+    # Self-hosted deployments that share Cloud's CTAS constraint can opt in.
+    assert run_mode(host="localhost", cloud_mode=True).is_cloud
+
+    # An explicit setting wins over the host name in both directions.
+    assert run_mode(host="foo.clickhouse.cloud", cloud_mode=False).is_standalone
+    assert run_mode(host="localhost", cloud_mode=False).is_standalone
+
+    # `cluster` enables the replication settings independently of cloud mode, so opting
+    # out of cloud mode must not drop them for a cluster deployment.
+    config = make_config(
+        type="clickhouse",
+        host="foo.clickhouse.cloud",
+        username="default",
+        cluster="cluster1",
+        cloud_mode=False,
+    )
+    assert config._static_connection_kwargs["insert_quorum"] == "auto"
+    assert (
+        config._static_connection_kwargs["database_replicated_enforce_synchronous_settings"] == "1"
+    )
+
+    # Cloud mode on its own must enable them too, with no cluster configured.
+    cloud_only = make_config(
+        type="clickhouse", host="localhost", username="default", cloud_mode=True
+    )
+    assert cloud_only._static_connection_kwargs["insert_quorum"] == "auto"
+    assert (
+        cloud_only._static_connection_kwargs["database_replicated_enforce_synchronous_settings"]
+        == "1"
+    )
+
+    # ...and a standalone deployment with cloud mode off should not get them at all.
+    standalone = make_config(type="clickhouse", host="localhost", username="default")
+    assert "insert_quorum" not in standalone._static_connection_kwargs
+
+
 def test_athena(make_config):
     config = make_config(type="athena", work_group="primary")
     assert isinstance(config, AthenaConnectionConfig)