Merge pull request ClickHouse#77774 from ClickHouse/notach

Fix refreshable materialized view not working on newly added replicas

Author: al13n321

src/Storages/MaterializedView/RefreshTask.cpp

@@ -63,7 +63,7 @@ namespace ErrorCodes
 }
 
 RefreshTask::RefreshTask(
-    StorageMaterializedView * view_, ContextPtr context, const DB::ASTRefreshStrategy & strategy, bool attach, bool coordinated, bool empty)
+    StorageMaterializedView * view_, ContextPtr context, const DB::ASTRefreshStrategy & strategy, bool /* attach */, bool coordinated, bool empty)
     : log(getLogger("RefreshTask"))
     , view(view_)
     , refresh_schedule(strategy)
@@ -88,29 +88,22 @@ RefreshTask::RefreshTask(
 
         auto zookeeper = context->getZooKeeper();
         String replica_path = coordination.path + "/replicas/" + coordination.replica_name;
+        bool replica_path_existed = zookeeper->exists(replica_path);
 
-        if (server_settings[ServerSetting::disable_insertion_and_mutation])
-            coordination.read_only = true;
-
-        if (attach)
-        {
-            /// Check that this replica is registered in keeper.
-            if (!zookeeper->exists(replica_path))
-            {
-                LOG_ERROR(log, "Attaching refreshable materialized view {} as read-only because znode {} is missing", view->getStorageID().getFullTableName(), replica_path);
-                coordination.read_only = true;
-            }
-        }
-        else
+        /// Create znodes even if it's ATTACH query. This seems weird, possibly incorrect, but
+        /// currently both DatabaseReplicated and DatabaseShared seem to require this behavior.
+        if (!replica_path_existed)
         {
             zookeeper->createAncestors(coordination.path);
-            /// Create coordination znodes if they don't exist. Register this replica, throwing if already exists.
             Coordination::Requests ops;
             ops.emplace_back(zkutil::makeCreateRequest(coordination.path, coordination.root_znode.toString(), zkutil::CreateMode::Persistent, true));
             ops.emplace_back(zkutil::makeCreateRequest(coordination.path + "/replicas", "", zkutil::CreateMode::Persistent, true));
             ops.emplace_back(zkutil::makeCreateRequest(replica_path, "", zkutil::CreateMode::Persistent));
             zookeeper->multi(ops);
         }
+
+        if (server_settings[ServerSetting::disable_insertion_and_mutation])
+            coordination.read_only = true;
     }
 }
 

tests/integration/test_refreshable_mv/test.py

@@ -338,6 +338,30 @@ def test_refresh_vs_shutdown_smoke(started_cluster):
     node1.query("drop database re sync")
     node2.query("drop database re sync")
 
+def test_adding_replica(started_cluster):
+    node1.query(
+        "create database re engine = Replicated('/test/re', 'shard1', 'r1');"
+        "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);"
+        "system wait view re.a")
+    assert node1.query("select * from re.a order by all") == "0\n10\n"
+    assert node1.query("select last_refresh_replica from system.view_refreshes") == "1\n"
+
+    r = node2.query(
+        "create database re engine = Replicated('/test/re', 'shard1', 'r2');"
+        "system sync database replica re")
+    assert node2.query("select * from re.a order by all") == "0\n10\n"
+
+    node1.query("system stop view re.a")
+    r = node2.query(
+        "system wait view re.a;"
+        "system refresh view re.a;"
+        "system wait view re.a;"
+        "select last_refresh_replica from system.view_refreshes")
+    assert r == "2\n"
+
+    node1.query("drop database re sync")
+    node2.query("drop database re sync")
+
 def test_replicated_db_startup_race(started_cluster):
     for node in nodes:
         node.query(