Merge pull request ClickHouse#87239 from ClickHouse/backport/25.8/87043

Backport ClickHouse#87043 to 25.8: Fix lightweight updates on cluster

Author: clickhouse-gh[bot]

src/Interpreters/InterpreterUpdateQuery.cpp

@@ -18,6 +18,7 @@
 #include <Storages/MutationCommands.h>
 #include <Core/Settings.h>
 #include <Core/ServerSettings.h>
+#include <Interpreters/executeDDLQueryOnCluster.h>
 
 
 namespace DB
@@ -73,14 +74,24 @@ BlockIO InterpreterUpdateQuery::execute()
     if (!settings[Setting::enable_lightweight_update])
         throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "Lightweight updates are not allowed. Set 'enable_lightweight_update = 1' to allow them");
 
+    FunctionNameNormalizer::visit(query_ptr.get());
+    auto & update_query = query_ptr->as<ASTUpdateQuery &>();
+
+    AccessRightsElements required_access;
+    required_access.emplace_back(AccessType::ALTER_UPDATE, update_query.getDatabase(), update_query.getTable());
+
+    if (!update_query.cluster.empty())
+    {
+        DDLQueryOnClusterParams params;
+        params.access_to_check = std::move(required_access);
+        return executeDDLQueryOnCluster(query_ptr, getContext(), params);
+    }
+
     if (getContext()->getGlobalContext()->getServerSettings()[ServerSetting::disable_insertion_and_mutation])
         throw Exception(ErrorCodes::QUERY_IS_PROHIBITED, "Update queries are prohibited");
 
-    FunctionNameNormalizer::visit(query_ptr.get());
-    auto & update_query = query_ptr->as<ASTUpdateQuery &>();
+    getContext()->checkAccess(required_access);
     auto table_id = getContext()->resolveStorageID(update_query, Context::ResolveOrdinary);
-
-    getContext()->checkAccess(AccessType::ALTER_UPDATE, table_id);
     update_query.setDatabase(table_id.database_name);
 
     /// First check table storage for validations.

tests/integration/test_lightweight_updates/configs/remote_servers.xml

@@ -0,0 +1,18 @@
+<clickhouse>
+    <remote_servers>
+        <test_cluster>
+            <shard>
+                <replica>
+                    <host>node1</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+            <shard>
+                <replica>
+                    <host>node2</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+        </test_cluster>
+    </remote_servers>
+</clickhouse>

tests/integration/test_lightweight_updates/test.py

@@ -9,11 +9,13 @@
 node1 = cluster.add_instance(
     "node1",
     with_zookeeper=True,
+    main_configs=["configs/remote_servers.xml"],
 )
 
 node2 = cluster.add_instance(
     "node2",
     with_zookeeper=True,
+    main_configs=["configs/remote_servers.xml"],
 )
 
 node3 = cluster.add_instance(
@@ -183,3 +185,45 @@ def test_lwu_upgrade(started_cluster, table_engine):
         )
         == "100000\n"
     )
+
+
+def test_lwu_on_cluster(started_cluster):
+    node1.query("DROP TABLE IF EXISTS t_lwu_on_cluster")
+    node2.query("DROP TABLE IF EXISTS t_lwu_on_cluster")
+
+    create_query = """
+    CREATE TABLE t_lwu_on_cluster
+    (
+        `id` UInt32,
+        `value` String,
+    )
+    ENGINE = MergeTree
+    ORDER BY id
+    SETTINGS enable_block_number_column = 1, enable_block_offset_column = 1
+"""
+
+    node1.query(create_query)
+    node2.query(create_query)
+
+    node1.query(
+        "INSERT INTO t_lwu_on_cluster SELECT number, '' FROM numbers(10000) WHERE number % 4 != 0"
+    )
+    node2.query(
+        "INSERT INTO t_lwu_on_cluster SELECT number, '' FROM numbers(10000) WHERE number % 4 != 1"
+    )
+
+    assert (
+        node1.query(
+            "SELECT count() from remote(test_cluster, currentDatabase(), t_lwu_on_cluster)"
+        )
+        == "15000\n"
+    )
+    node1.query(
+        "UPDATE t_lwu_on_cluster ON CLUSTER test_cluster SET value = 'updated' WHERE id >= 2000 AND id < 3000"
+    )
+    assert (
+        node1.query(
+            "SELECT count() from remote(test_cluster, currentDatabase(), t_lwu_on_cluster) WHERE value = 'updated'"
+        )
+        == "1500\n"
+    )