I am reading clickhouse mutation code. In StorageMergeTree->selectPartsToMutate() function, there is a check 
`if (current_ast_elements + commands_size >= max_ast_elements)
                break;`

I think if there are many mutations then the check result will be true, so that it will skip the remaining mutations.
But the new part's mutation is set to the last one in mutation map, and it does not apply the remaining mutations actually.
`new_part_info.mutation = current_mutations_by_version.rbegin()->first;`

Then the new part will miss the remaining mutations in future mutation job.

I think it's wrong here. Am I right?

If it is right, I will submit a PR 

*
      * The problem is that the numbers (`sequential` node) of the queue elements in `log` and in `queue` do not match.
      * (And the numbers of the same log element for different replicas do not match in the `queue`.)
      */
    /** First, you need to wait until replica takes `queue` element from the `log` to its queue,
      *  if it has not been done already (see the `pullLogsToQueue` function).
      *
      * To do this, check its node `log_pointer` - the maximum number of the element taken from `log` + 1.
      */
    bool waiting_itself = replica == replica_name;
    
    bool wait_for_inactive = wait_for_inactive_timeout != 0;
    
    bool check_timeout = wait_for_inactive_timeout > 0;
    Stopwatch time_waiting;
    const auto & stop_waiting = [&]()
    {
        bool stop_waiting_itself = waiting_itself && partial_shutdown_called;
        bool timeout_exceeded = check_timeout && wait_for_inactive_timeout < time_waiting.elapsedSeconds();
        bool stop_waiting_inactive = (!wait_for_inactive || timeout_exceeded)
            && !getZooKeeper()->exists(fs::path(table_zookeeper_path) / "replicas" / replica / "is_active");
        return is_dropped || stop_waiting_itself || stop_waiting_inactive;
    };
    
    constexpr auto event_wait_timeout_ms = 3000;
    LOG_DEBUG(log, "Waiting for {} to process log entry", replica);
    if (startsWith(entry.znode_name, "log-"))
    {
        
        UInt64 log_index = parse<UInt64>(entry.znode_name.substr(entry.znode_name.size() - 10));