- 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. ...
TaskShard & task_shard, const String & partition_quoted_name)
{
    createShardInternalTables(timeouts, task_shard, false);
    TaskTable & task_table = task_shard.task_table;
    std::string query = "SELECT 1 FROM " + getQuotedTable(task_shard.table_read_shard)
                        + " WHERE (" + queryToString(task_table.engine_push_partition_key_ast) +
                        " = (" + partition_quoted_name + " AS partition_key))";
    if (!task_table.where_condition_str.empty())
        query += " AND (" + task_table.where_condition_str + ")";
    query += " LIMIT 1";
    LOG_DEBUG(log, "Checking shard {} for partition {} existence, executing query: {}", task_shard.getDescription(), partition_quoted_name, query);
    ParserQuery parser_query(query.data() + query.size());
const auto & settings = context.getSettingsRef();
    ASTPtr query_ast = parseQuery(parser_query, query, settings.max_query_size, settings.max_parser_depth);
    Context local_context = context;
    local_context.setSettings(task_cluster->settings_pull);
    return InterpreterFactory::get(query_ast, local_context)->execute().getInputStream()->read().rows() != 0;
}
bool ClusterCopier::checkPresentPartitionPiecesOnCurrentShard(const ConnectionTimeouts & timeouts,
                           TaskShard & task_shard, const String & partition_quoted_name, size_t current_piece_number)
{
    createShardInternalTables(timeouts, task_shard, false);
    TaskTable & task_table = task_shard.task_table;
    const size_t number_of_splits = task_table.number_of_splits;