- some copier work needs to be improved

1. clickhouse create destionation table just once if success
2. support settings_push.replication_alter_partitions_sync =1 at attach partition 
3. ...
const ClusterPtr & cluster,
        const String & query,
        const Settings & current_settings,
        PoolMode pool_mode,
        ClusterExecutionMode execution_mode,
        UInt64 max_successful_executions_per_shard) const
{
    auto num_shards = cluster->getShardsInfo().size();
    std::vector<UInt64> per_shard_num_successful_replicas(num_shards, 0);
    ParserQuery p_query(query.data() + query.size());
    ASTPtr query_ast = parseQuery(p_query, query, current_settings.max_query_size, current_settings.max_parser_depth);
    
    if (execution_mode == ClusterExecutionMode::ON_EACH_NODE)
        max_successful_executions_per_shard = 0;
    std::atomic<size_t> origin_replicas_number = 0;
    
    auto do_for_shard = [&] (UInt64 shard_index, Settings shard_settings)
    {
        setThreadName("QueryForShard");
        const Cluster::ShardInfo & shard = cluster->getShardsInfo().at(shard_index);
        UInt64 & num_successful_executions = per_shard_num_successful_replicas.at(shard_index);
        num_successful_executions = 0;
        auto increment_and_check_exit = [&] () -> bool
        {
            ++num_successful_executions;
            return max_successful_executions_per_shard && num_successful_executions >= max_successful_executions_per_shard;
        };
        UInt64 num_replicas = cluster->getShardsAddresses().at(shard_index).size();
        origin_replicas_number += num_replicas;
        UInt64 num_local_replicas = shard.getLocalNodeCount();