- 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. ...
else
        {
            LOG_DEBUG(log, "Creating clean state clock");
            zookeeper->create(piece_is_dirty_flag_path, host_id, zkutil::CreateMode::Persistent);
        }
    };
    
    auto get_select_query = [&] (const DatabaseAndTableName & from_table, const String & fields, bool enable_splitting, String limit = "")
    {
        String query;
        query += "SELECT " + fields + " FROM " + getQuotedTable(from_table);
        if (enable_splitting && experimental_use_sample_offset)
            query += " SAMPLE 1/" + toString(number_of_splits) + " OFFSET " + toString(current_piece_number) + "/" + toString(number_of_splits);
        
        query += " WHERE (" + queryToString(task_table.engine_push_partition_key_ast) + " = (" + task_partition.name + " AS partition_key))";
        if (enable_splitting && !experimental_use_sample_offset)
            query += " AND ( cityHash64(" + primary_key_comma_separated + ") %" + toString(number_of_splits) + " = " + toString(current_piece_number) + " )";
        if (!task_table.where_condition_str.empty())
            query += " AND (" + task_table.where_condition_str + ")";
        if (!limit.empty())
            query += " LIMIT " + limit;
        ParserQuery p_query(query.data() + query.size());
        const auto & settings = context.getSettingsRef();
        return parseQuery(p_query, query, settings.max_query_size, settings.max_parser_depth);
    };
    
    auto worker_node_holder = createTaskWorkerNodeAndWaitIfNeed(zookeeper, current_task_piece_status_path, is_unprioritized_task);
    LOG_DEBUG(log, "Processing {}", current_task_piece_status_path);
    const String piece_status_path = partition_piece.getPartitionPieceShardsPath();
    CleanStateClock clean_state_clock(zookeeper, piece_is_dirty_flag_path, piece_is_dirty_cleaned_path);