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 

{
        /
        Field partition_key_value = evaluateConstantExpression(partition_ast.value, local_context).first;
        if (partition_key_value.getType() != Field::Types::Tuple)
            throw Exception(ErrorCodes::INVALID_PARTITION_VALUE,
                            "Expected tuple for complex partition key, got {}", partition_key_value.getTypeName());
        const Tuple & tuple = partition_key_value.get<Tuple>();
        if (tuple.size() != fields_count)
            throw Exception(ErrorCodes::LOGICAL_ERROR,
                            "Wrong number of fields in the partition expression: {}, must be: {}", tuple.size(), fields_count);
        for (size_t i = 0; i < fields_count; ++i)
            partition_row[i] = convertFieldToTypeOrThrow(tuple[i], *key_sample_block.getByPosition(i).type);
    }
    MergeTreePartition partition(std::move(partition_row));
    String partition_id = partition.getID(*this);
    {
        auto data_parts_lock = lockParts();
        DataPartPtr existing_part_in_partition = getAnyPartInPartition(partition_id, data_parts_lock);
        if (existing_part_in_partition && existing_part_in_partition->partition.value != partition.value)
        {
            WriteBufferFromOwnString buf;
            partition.serializeText(*this, buf, FormatSettings{});
            throw Exception(ErrorCodes::LOGICAL_ERROR, "Parsed partition value: {} "
                            "doesn't match partition value for an existing part with the same partition ID: {}",
                            buf.str(), existing_part_in_partition->name);
        }
    }
    return partition_id;
}
std::unordered_set<String> MergeTreeData::getPartitionIDsFromQuery(const ASTs & asts, ContextPtr local_context) const