diff --git a/sqlmesh/core/config/connection.py b/sqlmesh/core/config/connection.py
index 7a65e87d72..731121876c 100644
--- a/sqlmesh/core/config/connection.py
+++ b/sqlmesh/core/config/connection.py
@@ -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"
@@ -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,
diff --git a/tests/core/test_connection_config.py b/tests/core/test_connection_config.py
index c506d401d9..21c2ae0b1e 100644
--- a/tests/core/test_connection_config.py
+++ b/tests/core/test_connection_config.py
@@ -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)