Merge pull request ClickHouse#78277 from XuJia0210/feature_refreshable_mv_ignore_RO

bugfix: should not schedule RefreshMV task to read_only node

Author: al13n321

src/Storages/MaterializedView/RefreshTask.cpp

@@ -42,6 +42,7 @@ namespace ServerSetting
 {
     extern const ServerSettingsString default_replica_name;
     extern const ServerSettingsString default_replica_path;
+    extern const ServerSettingsBool disable_insertion_and_mutation;
 }
 
 namespace RefreshSetting
@@ -87,6 +88,10 @@ RefreshTask::RefreshTask(
 
         auto zookeeper = context->getZooKeeper();
         String replica_path = coordination.path + "/replicas/" + coordination.replica_name;
+
+        if (server_settings[ServerSetting::disable_insertion_and_mutation])
+            coordination.read_only = true;
+
         if (attach)
         {
             /// Check that this replica is registered in keeper.

tests/integration/test_refreshable_mv/configs/read_only.xml

@@ -0,0 +1,3 @@
+<clickhouse>
+    <disable_insertion_and_mutation>true</disable_insertion_and_mutation>
+</clickhouse>

tests/integration/test_refreshable_mv/test.py

@@ -29,6 +29,15 @@
     keeper_required_feature_flags=["multi_read", "create_if_not_exists"],
     macros={"shard": "shard1", "replica": "2"},
 )
+
+reading_node = cluster.add_instance(
+    "reading_node",
+    main_configs=["configs/read_only.xml"],
+    user_configs=["configs/users.xml"],
+    with_zookeeper=True,
+    keeper_required_feature_flags=["multi_read", "create_if_not_exists"],
+    macros={"shard": "shard1", "replica": "3"},
+)
 nodes = [node1, node2]
 
 
@@ -223,6 +232,60 @@ def test_refreshable_mv_in_system_db(started_cluster):
 
     node1.query("drop table system.a")
 
+def test_refreshable_mv_in_read_only_node(started_cluster):
+    # writable node
+    node1.query(
+        "create database re engine = Replicated('/test/re', 'shard1', '{replica}');"
+    )
+
+    # read_only node
+    reading_node.query(
+        "create database re engine = Replicated('/test/re', 'shard1', '{replica}');"
+    )
+
+    # disable view sync on writable node, see if there's RefreshTask on read_only node
+    node1.query(
+        "system stop view sync"
+    )
+
+    # clear text_log ensure all logs are related to this test
+    reading_node.query(
+        "system flush logs;"
+        "truncate table system.text_log;"
+    )
+
+    # this MV will be replicated to read_only node
+    node1.query(
+        "create materialized view re.a refresh every 1 second (x Int64) engine ReplicatedMergeTree order by x as select number*10 as x from numbers(2)"
+    )
+
+    # refresh the view manually
+    reading_node.query("system refresh view re.a")
+
+    # slepp 3 seconds to make sure the view is refreshed
+    reading_node.query("select sleep(3)")
+
+    # check if there's RefreshTask on read_only node
+    reading_node.query("system flush logs")
+    assert reading_node.query("select count() from system.text_log where message like '%QUERY_IS_PROHIBITED%'") == "0\n"
+    assert reading_node.query("select count() from system.view_refreshes where exception != ''") == "0\n"
+
+    # start sync and chek refresh task works well on node1
+    node1.query("system start view sync")
+    node1.query("system refresh view re.a")
+    assert_eq_with_retry(
+        node1,
+        "select * from re.a order by x",
+        "0\n10\n",
+    )
+    assert_eq_with_retry(
+        node1,
+        "select count() from system.view_refreshes where exception = '' and last_refresh_replica = '1'",
+        "1\n",
+    )
+
+    reading_node.query("drop database re sync")
+    node1.query("drop database re sync")
 
 def test_refresh_vs_shutdown_smoke(started_cluster):
     for node in nodes: