Merge pull request #13670 from ClickHouse/backport/20.4/13626

tavplubix · web-flow · commit 1171630a39cd · 2020-08-13T20:00:11.000+03:00
Backport #13626 to 20.4: Fix deadlock and logical error in replication queue
diff --git a/src/Storages/MergeTree/ReplicatedMergeTreeLogEntry.h b/src/Storages/MergeTree/ReplicatedMergeTreeLogEntry.h
@@ -161,6 +161,7 @@ struct ReplicatedMergeTreeLogEntryData
 
     /// Access under queue_mutex, see ReplicatedMergeTreeQueue.
     bool currently_executing = false;    /// Whether the action is executing now.
+    bool removed_by_other_entry = false;
     /// These several fields are informational only (for viewing by the user using system tables).
     /// Access under queue_mutex, see ReplicatedMergeTreeQueue.
     size_t num_tries = 0;                 /// The number of attempts to perform the action (since the server started, including the running one).
diff --git a/src/Storages/MergeTree/ReplicatedMergeTreeQueue.cpp b/src/Storages/MergeTree/ReplicatedMergeTreeQueue.cpp
@@ -343,47 +343,60 @@ void ReplicatedMergeTreeQueue::updateTimesInZooKeeper(
 
 void ReplicatedMergeTreeQueue::removeProcessedEntry(zkutil::ZooKeeperPtr zookeeper, LogEntryPtr & entry)
 {
-    auto code = zookeeper->tryRemove(replica_path + "/queue/" + entry->znode_name);
-
-    if (code)
-        LOG_ERROR(log, "Couldn't remove " << replica_path << "/queue/" << entry->znode_name << ": "
-            << zkutil::ZooKeeper::error2string(code) << ". This shouldn't happen often.");
-
     std::optional<time_t> min_unprocessed_insert_time_changed;
     std::optional<time_t> max_processed_insert_time_changed;
 
     bool found = false;
+    bool need_remove_from_zk = true;
     size_t queue_size = 0;
 
     {
         std::unique_lock lock(state_mutex);
-
-        /// Remove the job from the queue in the RAM.
-        /// You can not just refer to a pre-saved iterator, because someone else might be able to delete the task.
-        /// Why do we view the queue from the end?
-        ///  - because the task for execution first is moved to the end of the queue, so that in case of failure it remains at the end.
-        for (Queue::iterator it = queue.end(); it != queue.begin();)
+        if (entry->removed_by_other_entry)
+        {
+            need_remove_from_zk = false;
+            queue_size = queue.size();
+        }
+        else
         {
-            --it;
-            if (*it == entry)
+            /// Remove the job from the queue in the RAM.
+            /// You can not just refer to a pre-saved iterator, because someone else might be able to delete the task.
+            /// Why do we view the queue from the end?
+            ///  - because the task for execution first is moved to the end of the queue, so that in case of failure it remains at the end.
+            for (Queue::iterator it = queue.end(); it != queue.begin();)
             {
-                found = true;
-                updateStateOnQueueEntryRemoval(
-                    entry, /* is_successful = */ true,
-                    min_unprocessed_insert_time_changed, max_processed_insert_time_changed, lock);
+                --it;
 
-                queue.erase(it);
-                queue_size = queue.size();
-                break;
+                if (*it == entry)
+                {
+                    found = true;
+                    updateStateOnQueueEntryRemoval(
+                            entry, /* is_successful = */ true,
+                            min_unprocessed_insert_time_changed, max_processed_insert_time_changed, lock);
+
+                    queue.erase(it);
+                    queue_size = queue.size();
+                    break;
+                }
             }
         }
     }
 
-    if (!found)
+    if (!found && need_remove_from_zk)
         throw Exception("Can't find " + entry->znode_name + " in the memory queue. It is a bug", ErrorCodes::LOGICAL_ERROR);
 
     notifySubscribers(queue_size);
 
+
+    if (!need_remove_from_zk)
+        return;
+
+    auto code = zookeeper->tryRemove(replica_path + "/queue/" + entry->znode_name);
+
+    if (code)
+        LOG_ERROR(log, "Couldn't remove " << replica_path << "/queue/" << entry->znode_name << ": "
+            << zkutil::ZooKeeper::error2string(code) << ". This shouldn't happen often.");
+
     updateTimesInZooKeeper(zookeeper, min_unprocessed_insert_time_changed, max_processed_insert_time_changed);
 }
 
@@ -845,13 +858,18 @@ void ReplicatedMergeTreeQueue::removePartProducingOpsInRange(
     const MergeTreePartInfo & part_info,
     const ReplicatedMergeTreeLogEntryData & current)
 {
+    /// TODO is it possible to simplify it?
     Queue to_wait;
     size_t removed_entries = 0;
     std::optional<time_t> min_unprocessed_insert_time_changed;
     std::optional<time_t> max_processed_insert_time_changed;
 
     /// Remove operations with parts, contained in the range to be deleted, from the queue.
     std::unique_lock lock(state_mutex);
+
+    [[maybe_unused]] bool called_from_alter_query_directly = current.replace_range_entry && current.replace_range_entry->columns_version < 0;
+    assert(currently_executing_drop_or_replace_range || called_from_alter_query_directly);
+
     for (Queue::iterator it = queue.begin(); it != queue.end();)
     {
         auto type = (*it)->type;
@@ -863,13 +881,18 @@ void ReplicatedMergeTreeQueue::removePartProducingOpsInRange(
             if ((*it)->currently_executing)
                 to_wait.push_back(*it);
             auto code = zookeeper->tryRemove(replica_path + "/queue/" + (*it)->znode_name);
+            /// FIXME it's probably unsafe to remove entries non-atomically
+            /// when this method called directly from alter query (not from replication queue task),
+            /// because entries will be lost if ALTER fails.
             if (code)
                 LOG_INFO(log, "Couldn't remove " << replica_path + "/queue/" + (*it)->znode_name << ": "
                     << zkutil::ZooKeeper::error2string(code));
 
             updateStateOnQueueEntryRemoval(
                 *it, /* is_successful = */ false,
                 min_unprocessed_insert_time_changed, max_processed_insert_time_changed, lock);
+
+            (*it)->removed_by_other_entry = true;
             queue.erase(it++);
             ++removed_entries;
         }
@@ -1105,6 +1128,16 @@ bool ReplicatedMergeTreeQueue::shouldExecuteLogEntry(
         }
     }
 
+    if (entry.type == LogEntry::DROP_RANGE || entry.type == LogEntry::REPLACE_RANGE)
+    {
+        /// DROP_RANGE and REPLACE_RANGE entries remove other entries, which produce parts in the range.
+        /// If such part producing operations are currently executing, then DROP/REPLACE RANGE wait them to finish.
+        /// Deadlock is possible if multiple DROP/REPLACE RANGE entries are executing in parallel and wait each other.
+        /// See also removePartProducingOpsInRange(...) and ReplicatedMergeTreeQueue::CurrentlyExecuting.
+        if (currently_executing_drop_or_replace_range)
+            return false;
+    }
+
     return true;
 }
 
@@ -1135,6 +1168,11 @@ Int64 ReplicatedMergeTreeQueue::getCurrentMutationVersion(const String & partiti
 ReplicatedMergeTreeQueue::CurrentlyExecuting::CurrentlyExecuting(const ReplicatedMergeTreeQueue::LogEntryPtr & entry_, ReplicatedMergeTreeQueue & queue_)
     : entry(entry_), queue(queue_)
 {
+    if (entry->type == ReplicatedMergeTreeLogEntry::DROP_RANGE || entry->type == ReplicatedMergeTreeLogEntry::REPLACE_RANGE)
+    {
+        assert(!queue.currently_executing_drop_or_replace_range);
+        queue.currently_executing_drop_or_replace_range = true;
+    }
     entry->currently_executing = true;
     ++entry->num_tries;
     entry->last_attempt_time = time(nullptr);
@@ -1168,6 +1206,11 @@ ReplicatedMergeTreeQueue::CurrentlyExecuting::~CurrentlyExecuting()
 {
     std::lock_guard lock(queue.state_mutex);
 
+    if (entry->type == ReplicatedMergeTreeLogEntry::DROP_RANGE || entry->type == ReplicatedMergeTreeLogEntry::REPLACE_RANGE)
+    {
+        assert(queue.currently_executing_drop_or_replace_range);
+        queue.currently_executing_drop_or_replace_range = false;
+    }
     entry->currently_executing = false;
     entry->execution_complete.notify_all();
 
diff --git a/src/Storages/MergeTree/ReplicatedMergeTreeQueue.h b/src/Storages/MergeTree/ReplicatedMergeTreeQueue.h
@@ -87,6 +87,9 @@ class ReplicatedMergeTreeQueue
     /// Index of the first log entry that we didn't see yet.
     Int64 log_pointer = 0;
 
+    /// Avoid parallel execution of queue enties, which may remove other entries from the queue.
+    bool currently_executing_drop_or_replace_range = false;
+
     /** What will be the set of active parts after executing all log entries up to log_pointer.
       * Used to determine which merges can be assigned (see ReplicatedMergeTreeMergePredicate)
       */
diff --git a/tests/queries/0_stateless/01035_concurrent_move_partition_from_table_zookeeper.sh b/tests/queries/0_stateless/01035_concurrent_move_partition_from_table_zookeeper.sh
@@ -43,18 +43,28 @@ function thread4()
     done
 }
 
+function thread5()
+{
+    while true;
+    do
+        $CLICKHOUSE_CLIENT --query="ALTER TABLE src MOVE PARTITION 1 TO TABLE dst;" --query_id=query5
+    done
+}
+
 # https://stackoverflow.com/questions/9954794/execute-a-shell-function-with-timeout
 export -f thread1;
 export -f thread2;
 export -f thread3;
 export -f thread4;
+export -f thread5;
 
 TIMEOUT=30
 
 timeout $TIMEOUT bash -c thread1 2> /dev/null &
 timeout $TIMEOUT bash -c thread2 2> /dev/null &
 timeout $TIMEOUT bash -c thread3 2> /dev/null &
 timeout $TIMEOUT bash -c thread4 2> /dev/null &
+timeout $TIMEOUT bash -c thread5 2> /dev/null &
 
 wait
 
