- 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. ...
ASTPtr query_select_ast = get_select_query(task_shard.table_read_shard, "*",  true, inject_fault ? "1" : "");
        LOG_DEBUG(log, "Executing SELECT query and pull from {} : {}", task_shard.getDescription(), queryToString(query_select_ast));
        ASTPtr query_insert_ast;
        {
            String query;
            query += "INSERT INTO " + getQuotedTable(split_table_for_current_piece) + " VALUES ";
            ParserQuery p_query(query.data() + query.size());
            const auto & settings = context.getSettingsRef();
            query_insert_ast = parseQuery(p_query, query, settings.max_query_size, settings.max_parser_depth);
            LOG_DEBUG(log, "Executing INSERT query: {}", query);
        }
        try
        {
            std::unique_ptr<Context> context_select = std::make_unique<Context>(context);
            context_select->setSettings(task_cluster->settings_pull);
            std::unique_ptr<Context> context_insert = std::make_unique<Context>(context);
            context_insert->setSettings(task_cluster->settings_push);
            
            BlockInputStreamPtr input;
            BlockOutputStreamPtr output;
            {
                BlockIO io_select = InterpreterFactory::get(query_select_ast, *context_select)->execute();
                BlockIO io_insert = InterpreterFactory::get(query_insert_ast, *context_insert)->execute();
                input = io_select.getInputStream();
                output = io_insert.out;
            }
            
            std::future<Coordination::ExistsResponse> future_is_dirty_checker;
            Stopwatch watch(CLOCK_MONOTONIC_COARSE);
            constexpr UInt64 check_period_milliseconds = 500;