diff --git a/docs/integrations/engines/clickhouse.md b/docs/integrations/engines/clickhouse.md index 4c2aab6e78..aa6ac2bb7c 100644 --- a/docs/integrations/engines/clickhouse.md +++ b/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 | diff --git a/sqlmesh/core/config/connection.py b/sqlmesh/core/config/connection.py index 73fe1b9300..3c73c1f9a7 100644 --- a/sqlmesh/core/config/connection.py +++ b/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 diff --git a/tests/core/test_connection_config.py b/tests/core/test_connection_config.py index c506d401d9..163ca1aa5b 100644 --- a/tests/core/test_connection_config.py +++ b/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)