Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions sqlmesh/core/config/connection.py
Original file line number Diff line number Diff line change
Expand Up @@ -2391,7 +2391,7 @@ def _static_connection_kwargs(self) -> t.Dict[str, t.Any]:
# https://clickhouse.com/docs/en/operations/settings/settings
# - below are set to align with dbt-clickhouse
# - https://github.com/ClickHouse/dbt-clickhouse/blob/44d26308ea6a3c8ead25c280164aa88191f05f47/dbt/adapters/clickhouse/dbclient.py#L77
settings = self.connection_settings or {}
settings = dict(self.connection_settings or {})
# mutations_sync = 2: "The query waits for all mutations [ALTER statements] to complete on all replicas (if they exist)"
settings["mutations_sync"] = "2"
# insert_distributed_sync = 1: "INSERT operation succeeds only after all the data is saved on all shards"
Expand All @@ -2405,7 +2405,8 @@ def _static_connection_kwargs(self) -> t.Dict[str, t.Any]:
# - "INSERT succeeds only when ClickHouse manages to correctly write data to the insert_quorum of replicas during
# the insert_quorum_timeout"
# - "use majority number (number_of_replicas / 2 + 1) as quorum number"
settings["insert_quorum"] = "auto"
# - a value set by the user in connection_settings takes precedence
settings.setdefault("insert_quorum", "auto")

return {
"compress": compress,
Expand Down
27 changes: 27 additions & 0 deletions tests/core/test_connection_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -1287,6 +1287,33 @@ def test_clickhouse(make_config):
assert not config3._static_connection_kwargs["compress"]


def test_clickhouse_insert_quorum_user_setting(make_config):
user_settings = {"insert_quorum": "0"}
config = make_config(
type="clickhouse",
host="localhost",
username="default",
password="default",
cluster="default",
connection_settings=user_settings,
)

# A user-provided insert_quorum is respected in cluster mode
assert config._static_connection_kwargs["insert_quorum"] == "0"
# The user's dict is not mutated by building the connection kwargs
assert user_settings == {"insert_quorum": "0"}

# Without a user value, cluster mode still defaults to "auto"
config_default = make_config(
type="clickhouse",
host="localhost",
username="default",
password="default",
cluster="default",
)
assert config_default._static_connection_kwargs["insert_quorum"] == "auto"


def test_athena(make_config):
config = make_config(type="athena", work_group="primary")
assert isinstance(config, AthenaConnectionConfig)
Expand Down