remove local parts which source replica doesnt have

ClickHouse · Aug 19, 2019 · 6991683 · 6991683
1 parent 8bbcecf
commit 6991683
Show file tree

Hide file tree

Showing 4 changed files with 33 additions and 5 deletions.
diff --git a/dbms/src/Storages/StorageReplicatedMergeTree.cpp b/dbms/src/Storages/StorageReplicatedMergeTree.cpp
@@ -1942,10 +1942,37 @@ void StorageReplicatedMergeTree::cloneReplica(const String & source_replica, Coo
     }
 
     /// Add to the queue jobs to receive all the active parts that the reference/master replica has.
-    Strings parts = zookeeper->getChildren(source_path + "/parts");
-    ActiveDataPartSet active_parts_set(format_version, parts);
+    Strings source_replica_parts = zookeeper->getChildren(source_path + "/parts");
+    ActiveDataPartSet active_parts_set(format_version, source_replica_parts);
 
     Strings active_parts = active_parts_set.getParts();
+
+    /// Remove local parts if source replica does not have them, because such parts will never be fetched by other replicas.
+    Strings local_parts_in_zk = zookeeper->getChildren(replica_path + "/parts");
+    Strings parts_to_remove_from_zk;
+    for (const auto & part : local_parts_in_zk)
+    {
+        if (active_parts_set.getContainingPart(part).empty())
+        {
+            queue.remove(zookeeper, part);
+            parts_to_remove_from_zk.emplace_back(part);
+            LOG_WARNING(log, "Source replica does not have part " << part << ". Removing it from ZooKeeper.");
+        }
+    }
+    tryRemovePartsFromZooKeeperWithRetries(parts_to_remove_from_zk);
+
+    auto local_active_parts = getDataParts();
+    DataPartsVector parts_to_remove_from_working_set;
+    for (const auto & part : local_active_parts)
+    {
+        if (active_parts_set.getContainingPart(part->name).empty())
+        {
+            parts_to_remove_from_working_set.emplace_back(part);
+            LOG_WARNING(log, "Source replica does not have part " << part->name << ". Removing it from working set.");
+        }
+    }
+    removePartsFromWorkingSet(parts_to_remove_from_working_set, true);
+
     for (const String & name : active_parts)
     {
         LogEntry log_entry;

diff --git a/...ent_parts_after_clone_replica/__init__.py → ...ent_parts_after_clone_replica/__init__.py b/...ent_parts_after_clone_replica/__init__.py → ...ent_parts_after_clone_replica/__init__.py
diff --git a/..._clone_replica/configs/remote_servers.xml → ..._clone_replica/configs/remote_servers.xml b/..._clone_replica/configs/remote_servers.xml → ..._clone_replica/configs/remote_servers.xml
diff --git a/...sistent_parts_after_clone_replica/test.py → ...sistent_parts_after_clone_replica/test.py b/...sistent_parts_after_clone_replica/test.py → ...sistent_parts_after_clone_replica/test.py
@@ -36,7 +36,8 @@ def start_cluster():
 def test_inconsistent_parts_if_drop_while_replica_not_active(start_cluster):
     with PartitionManager() as pm:
         # insert into all replicas
-        node1.query("INSERT INTO test_table VALUES ('2019-08-16', 100)")
+        for i in range(50):
+            node1.query("INSERT INTO test_table VALUES ('2019-08-16', {})".format(i))
         assert_eq_with_retry(node2, "SELECT count(*) FROM test_table", node1.query("SELECT count(*) FROM test_table"))
 
         # disable network on the first replica
@@ -49,8 +50,8 @@ def test_inconsistent_parts_if_drop_while_replica_not_active(start_cluster):
 
         # insert into the second replica
         # DROP_RANGE will be removed from the replication log and the first replica will be lost
-        for i in range(100):
-            node2.query("INSERT INTO test_table VALUES ('2019-08-16', {})".format(i))
+        for i in range(50):
+            node2.query("INSERT INTO test_table VALUES ('2019-08-16', {})".format(50 + i))
 
         # the first replica will be cloned from the second
         pm.heal_all()