- 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. ...
copyData(*input, *output, cancel_check, update_stats);
            
            if (future_is_dirty_checker.valid())
                future_is_dirty_checker.get();
            if (inject_fault)
                throw Exception("Copy fault injection is activated", ErrorCodes::UNFINISHED);
        }
        catch (...)
        {
            tryLogCurrentException(log, "An error occurred during copying, partition will be marked as dirty");
            create_is_dirty_node(clean_state_clock);
            return TaskStatus::Error;
        }
    }
    LOG_INFO(log, "Partition {} piece {} copied. But not moved to original destination table.", task_partition.name, toString(current_piece_number));
    
    try
    {
        auto create_query_push_ast = rewriteCreateQueryStorage(task_shard.current_pull_table_create_query,
                                                               task_table.table_push, task_table.engine_push_ast);
        auto & create = create_query_push_ast->as<ASTCreateQuery &>();
        create.if_not_exists = true;
        InterpreterCreateQuery::prepareOnClusterQuery(create, context, task_table.cluster_push_name);
        String query = queryToString(create_query_push_ast);
        LOG_DEBUG(log, "Create destination tables. Query: {}", query);
        UInt64 shards = executeQueryOnCluster(task_table.cluster_push, query, task_cluster->settings_push, PoolMode::GET_MANY);
        LOG_DEBUG(log, "Destination tables {} have been created on {} shards of {}", getQuotedTable(task_table.table_push), shards, task_table.cluster_push->getShardCount());
    }
    catch (...)