You have to provide the following information whenever possible.

**Describe the bug**

Case ：Two clickhouse instances A and B，they are two replicas of the same shard, quickly perform many "alter table replace partition" operations on the same table on A, and finally the data in the two instances A and B are inconsistent.

The code version is 20.8.12.2

I debugged the code and found that this function “ReplicatedMergeTreeQueue::removePartProducingOpsInRange” deleted LogEntry by mistake

This line of code "checkReplaceRangeCanBeRemoved(part_info, *it, current)"

For example, if variable "current" is "queue-0000000002",  the content is

format version: 4
create_time: 2021-06-02 15:01:51
source replica: replica01
block_id:
REPLACE_RANGE
drop_range_name: 68a6cbc22e808d5c3665a492584ec053_0_0_4294967295
from_database: erp_pa_dw_stream
from_table: dim_dept_info_d_whole_cdy1_4762f914c8ded772e6e742f43565e5d4_20210528_tmp_local
source_parts: []
new_parts: []
part_checksums: []
columns_version: -1

variable "*it" is "queue-0000000003",  the content is

format version: 4
create_time: 2021-06-02 15:01:51
source replica: replica01
block_id:
REPLACE_RANGE
drop_range_name: cc04c226c1f9ac46fa63e73d7f6ec8f9_0_0_4294967295
from_database: erp_pa_dw_stream
from_table: dim_dept_info_d_whole_cdy1_358f28df0a09f4176940da62e12bdf63_20210528_tmp_local
source_parts: []
new_parts: []
part_checksums: []
columns_version: -1

"checkReplaceRangeCanBeRemoved" will return true, queue-0000000003 will be removed and not be executed.

queue-0000000002 and queue-0000000003 delete different partitions


**Does it reproduce on recent release?**

I'm not sure if the latest release has this problem, but the logic of this code has not changed

{
    std::lock_guard lock(state_mutex);
    if (isNotCoveredByFuturePartsImpl(entry.znode_name, part_name, reject_reason, lock))
    {
        CurrentlyExecuting::setActualPartName(entry, part_name, *this);
        return true;
    }
    return false;
}
bool ReplicatedMergeTreeQueue::shouldExecuteLogEntry(
    const LogEntry & entry,
    String & out_postpone_reason,
    MergeTreeDataMergerMutator & merger_mutator,
    MergeTreeData & data,
    std::lock_guard<std::mutex> & state_lock) const
{
    
    
    for (const String & new_part_name : entry.getVirtualPartNames(format_version))
    {
        if (!isNotCoveredByFuturePartsImpl(entry.znode_name, new_part_name, out_postpone_reason, state_lock))
            return false;
    }
    
    if ((entry.type == LogEntry::GET_PART || entry.type == LogEntry::ATTACH_PART)
        && !storage.canExecuteFetch(entry, out_postpone_reason))
    {
        
        
        return false;