# Merging sorted integer vectors in C++

Given multiple sorted integer vectors, the goal is to merge them into a single vector and eliminate duplicate values. We've already achieved some improvements over the most simple approaches, but overall the operation is still very expensive and we're not sure if we're missing something, there must be a way to perform this efficiently. A simplified example is:

{ {0, 3, 7, 12, 19, 28}, {2, 3, 12, 17}, {7, 40} }
// should be merged into
{0, 2, 3, 7, 12, 17, 19, 28, 40}


The approaches that we tried are:

1. std::set - collect values and eliminate duplicates
2. std::inplace_merge - essentially do the merge part of the merge-sort algorithm
3. std::vector<bool> - set bits on positions indicated by the integer values, then collect the indices where those bits were true

The following code listing contains the three approaches that we have tried, with randomly generated data that is representative of our cases. Likewise, the following link contains the same code where the benchmark may be evaluated: http://quick-bench.com/0SNjMyrkqm2N1ugjJb2OwZ5zVWc

// Google benchmark library
#include <benchmark/benchmark.h>

#include <algorithm>
#include <iostream>
#include <numeric>
#include <random>
#include <set>
#include <vector>

template<typename T>
class RndGen {
public:
RndGen(T start, T end) : mersenne_engine{rnd_device()},
dist{start, end} {
}

T operator()() {
return dist(mersenne_engine);
}

private:
std::random_device rnd_device;
std::mt19937 mersenne_engine;
std::uniform_int_distribution<T> dist;

};

template<typename T>
std::vector<T> gen_random_range(std::size_t count, T start, T end) {
std::vector<T> vec(count);

RndGen<T> generator{start, end};
auto gen = [&generator]() {
return generator();
};

std::generate(std::begin(vec), std::end(vec), gen);
std::sort(vec.begin(), vec.end());

return vec;
}

template<typename T>
std::vector<T> reversed(const std::vector<T>& src) {
auto result = src;
std::reverse(result.begin(), result.end());
return result;
}

std::vector<std::vector<int> > gen_ranges(const std::vector<std::size_t> range_sizes) {
std::vector<std::vector<int> > ranges;
ranges.reserve(range_sizes.size());
for (std::size_t i = 0ul; i < range_sizes.size(); ++i) {
ranges.push_back(gen_random_range(range_sizes[i], 0, static_cast<int>(range_sizes[i])));
}
return ranges;
}

struct TestData {
std::size_t range_count;
std::vector<std::size_t> range_sizes;
std::size_t max_possible_size;
std::vector<std::vector<int> > ranges;

explicit TestData(std::size_t range_count_) : range_count{range_count_},
range_sizes{reversed(gen_random_range(range_count, 5000ul, 35000ul))},
max_possible_size{std::accumulate(range_sizes.begin(), range_sizes.end(), 0ul)},
ranges{gen_ranges(range_sizes)} {
}

};

static const TestData& get_test_data() {
static TestData test_data{4ul  /* RndGen<std::size_t>{3ul, 6ul} (); */};
return test_data;
}

void dump_results(const std::vector<int>& result) {
for (auto v : result) {
std::cout << v << ", ";
}
std::cout << std::endl;
}

/*
* Generate a vector of int vectors, where these sub-vector are all already sorted in ascending order, e.g.:
*
* { {0, 3, 7, 12, 19, 28}, {2, 3, 12, 17}, {7, 40} }
*
* The size of these sub-vectors in the above example are representative of the actual data,
* as the first vector will always be the largest, with smaller vectors following it.
* Usually the first vector has around 25000 elements, the second vector around 13000, and the third vector 7000.
*
* The goal is to merge these lists into one sorted-contiguous chunk, and eliminate duplicates. For the above example it would be:
*
* {0, 2, 3, 7, 12, 17, 19, 28, 40}
*
*/

static void merge_with_set(benchmark::State& state) {
const auto& test_data = get_test_data();

std::set<int> sorted_distinct_numbers;
for (auto _ : state) {
sorted_distinct_numbers.clear();
for (const auto& r : test_data.ranges) {
sorted_distinct_numbers.insert(r.begin(), r.end());
}

std::vector<int> result{sorted_distinct_numbers.begin(), sorted_distinct_numbers.end()};

// return dump_results(result);

benchmark::DoNotOptimize(result);
}
}

BENCHMARK(merge_with_set);

static void merge_with_inplace_merge(benchmark::State& state) {
const auto& test_data = get_test_data();

std::vector<std::size_t> chunk_sizes;
chunk_sizes.resize(test_data.ranges.size());

for (auto _ : state) {

std::vector<int> result;
result.reserve(test_data.max_possible_size);

for (std::size_t i = 0ul; i < test_data.ranges.size(); ++i) {
result.insert(result.end(), test_data.ranges[i].begin(), test_data.ranges[i].end());
chunk_sizes[i] = result.size();
}

const auto begin = result.begin();
const auto end = result.end();
auto start = end;
// Chunks towards the end are smaller than at the beginning, thus for our data
// it's actually faster to merge those first
for (auto it = chunk_sizes.rbegin(); it != chunk_sizes.rend(); ++it) {
auto middle = start;
start = begin + *it;
std::inplace_merge(start, middle, end);
}
std::inplace_merge(begin, start, end);

// Eliminate duplicates
result.erase(std::unique(result.begin(), result.end()), result.end());

// return dump_results(result);

benchmark::DoNotOptimize(result);
}
}

BENCHMARK(merge_with_inplace_merge);

static void merge_with_vector_of_bool(benchmark::State& state) {
const auto& test_data = get_test_data();

std::vector<bool> sorted_distinct_numbers;
sorted_distinct_numbers.resize(test_data.max_possible_size);

for (auto _ : state) {
std::fill(sorted_distinct_numbers.begin(), sorted_distinct_numbers.end(), false);
for (const auto& r : test_data.ranges) {
for (const auto value : r) {
sorted_distinct_numbers[value] = true;
}
}

std::vector<int> result;
result.reserve(test_data.max_possible_size);
for (std::size_t i = 0ul; i < sorted_distinct_numbers.size(); ++i) {
if (sorted_distinct_numbers[i] == true) {
result.push_back(static_cast<int>(i));
}
}

// return dump_results(result);

benchmark::DoNotOptimize(result);
}
}

BENCHMARK(merge_with_vector_of_bool);

BENCHMARK_MAIN();


The std::vector<bool> approach is the fastest of all, though still a very expensive operation.

EDIT: Based on @vvtoan's comment, the std::vector<char> approach is the fastest currently, about 15-20% faster than std::vector<bool>, though the fundamental approach hasn't changed, and overall the operation is still expensive.

EDIT2: The integers generated for the test data really are between 0 and max_possible_size, this is not by mistake. When I said randomly generated data that is representative of our cases, I really meant the data is accurate, and the existing implementations are correct for exploiting these properties.

• The code is missing the required header numeric, #include <numeric>. – pacmaninbw Nov 1 '19 at 14:21
• Apologies and thanks, edited the code to include it. – andrean Nov 1 '19 at 14:33
• Leaving aside the matter of algorithms, an obvious improvement would be using a vector<int> instead of vector<bool>. – vvotan Nov 1 '19 at 15:14
• Stating the constraints would help. How many arrays are we talking about? How large they are? What is the typical distribution of their sizes? Do you have to do it in place? – vnp Nov 3 '19 at 18:45
• @vnp: The random test data that is being generated in the provided code is representative of the actual data being used. On average, there are 3-6 arrays of integers, with the first array always being the largest, and each subsequent one is smaller than the previous. The largest array has around 25000 integers, the next one 15000, then 10000, and it keeps decreasing by a couple thousand with each next array. In-place merge is not mandatory. The only goal is to make the solution faster than the existing ones shown in the code. – andrean Nov 3 '19 at 20:53

The functions cheat their benchmarks by allocating their auxiliary storage outside of the timing loop. That's not reasonable unless you can arrange for the "real life" versions to have their own (per-thread?) long-lived auxiliary storage.

I think there's a problem with the vector<bool> version that's masked by the test data (which always keeps the element values between 0 and test_data.max_possible_size. Perhaps that's how the real-life input data are naturally distributed, but the constraint needs to be clearly specified. For a more general case, we would need to arrange bool storage from the min to max values of all the inputs (and memory exhaustion becomes much more likely).

I did manage to improve the inplace-merge version by using a heap to identify the current lowest iterator among all inputs, for about 20% speedup:

#include <queue>

template<typename Container>
using QueueItem = std::pair<typename Container::const_iterator,
typename Container::const_iterator>;

static void merge_with_heap_merge(benchmark::State& state)
{
const auto& test_data = get_test_data();

for (auto _ : state) {

auto compare = [](auto a, auto b) { return *(a.first) < *(b.first); };
std::priority_queue<QueueItem<std::vector<int>>,
std::vector<QueueItem<std::vector<int>>>,
decltype(compare)>
heap(compare);

std::vector<int> result;
result.reserve(test_data.max_possible_size);

for (auto const& r: test_data.ranges) {
heap.emplace(r.begin(), r.end());
}

while (!heap.empty()) {
auto item = heap.top();
heap.pop();
if (!result.empty() && result.back() != *(item.first))
result.push_back(*(item.first));
if (++item.first != item.second) {
heap.emplace(item);
}
}

// return dump_results(result);

benchmark::DoNotOptimize(result);
}
}


The speedup comes primarily from avoiding repeated copying, and in particular, the final deduplication pass.

## Style and other issues

Equality comparisons with boolean constants are usually redundant:

            if (sorted_distinct_numbers[i] == true)


can be simply if (sorted_distinct_numbers[i])

gen_random_range() doesn't need an lvalue RndGen, so we can simplify a lot:

template<typename T>
std::vector<T> gen_random_range(std::size_t count, T start, T end) {
std::vector<T> vec(count);
std::generate(vec.begin(), vec.end(), RndGen<T>{start, end});
std::sort(vec.begin(), vec.end());

return vec;
}


The RndGen doesn't need to store a random device:

template<typename T>
class RndGen {
public:
RndGen(T start, T end)
: mersenne_engine{std::random_device{}()},
dist{start, end}
{
}

T operator()()
{
return dist(mersenne_engine);
}

private:
std::mt19937 mersenne_engine;
std::uniform_int_distribution<T> dist;
};


If we make reversed() accept argument by value, that means we don't need to explicitly copy it:

template<typename T>
std::vector<T> reversed(std::vector<T> v) {
std::reverse(v.begin(), v.end());
return v;
}

• Thanks for answering, I've extended the question with a confirmation about the test input data. Regarding the auxiliary storage, I consider it not cheating since they do not affect the results of the algorithm, and such optimizations are allowed / encouraged if they can help in achieving better results. – andrean Nov 5 '19 at 13:32
• The point wasn't about affecting the results; it was about affecting the timing. I think it's more honest to include the allocation within the loop, so that its time is measured as part of the run, as I have done in the code I show here. – Toby Speight Nov 5 '19 at 13:43
• With the input distribution that you have, I think you're unlikely to improve on the std::vector<char> mapping. I did try parallelising the population of this bit map, but for these data, the thread creation overheads appear to be a net loss. – Toby Speight Nov 5 '19 at 14:14
• All right, thanks for looking into it, I'll let the question sit a bit more since the bounty ends in 7 days, but if no other, better answers come in, I'll grant it to you. – andrean Nov 5 '19 at 14:54

It's not clear what benefit the code is receiving by using the Google Benchmark Library, there are other ways to measure the elapsed time of each of the merges. C++ provides time measurement functionality.

    std::chrono::time_point<std::chrono::system_clock> start, end;
start = std::chrono::system_clock::now();
// execute merge here;
end = std::chrono::system_clock::now();

std::chrono::duration<double> elapsed_seconds = end-start;
std::time_t end_time = std::chrono::system_clock::to_time_t(end);


The function static const TestData& get_test_data() violates the Single Responsibility Principle because it is responsible for constructing the TestData struct as well as returning it. This may lead to a one time performance hit while the struct is being created.

It might be better if main() or a sub-function that main calls constructed the test data once and then passed the test data into each of the separate merge functions.

Creating the result vector with the sum of all the sizes of the contributory vectors will decrease or eliminate the number of memory allocations necessary when adding the data from the test vectors to the result.

Profiling each of the merge algorithms will help find bottle necks that you may be able to optimize out.

• Sorry if I wasn't clear enough in the question, and thanks for taking the time to look at it and write an answer. The reason google benchmark code is included is to help potential reviewers easily benchmark the approaches. I am not looking for a code review about whether the code is clean or elegant, I agree it could be cleaner, but it gets through the point of the approaches simply enough. At the moment I'm interested in only making it as fast as possible. Of course i have profiled it many times already, the approaches shown are the result of such profiling / tuning iterations over it. – andrean Nov 1 '19 at 16:57
• One additional note, the creation of the struct is outside the benchmark loop, so the one-time hit's penalty won't be taken into account when measuring the performance. The reason it's a function local static is to use the same test data in all benchmarks within a single benchmarking session, since the data is randomly generated. – andrean Nov 1 '19 at 17:00

If your real data has indeed the same pattern as the test data, I think using a vector<> is a reasonable solution. You can probably try micro-optimizing it. I quickly tried some loop-unrolling and shortcuts, which allowed me to improve 3x over the original vector version. Sorry for the code quality, this is just to illustrate the concept (http://quick-bench.com/c3tnWIXA_CDEcKWzOepIR91XoV0):

template <class C> static void add(C &result, uint64_t val, uint64_t offs, int i){
const auto v = val & (0xFFull << (offs*8));
if(v!=0) result.push_back(i*8 + offs);
}

static  void merge_with_vector_of_int2(benchmark::State& state) {
const auto& test_data = get_test_data();

std::vector<char> sorted_distinct_numbers;
sorted_distinct_numbers.resize(test_data.max_possible_size);

for (auto _ : state) {
std::fill(sorted_distinct_numbers.begin(), sorted_distinct_numbers.end(), false);
for (const auto& r : test_data.ranges) {
for (const auto value : r) {
sorted_distinct_numbers[value] = 1;
}
}

std::vector<int> result;
result.reserve(test_data.max_possible_size);
int i = 0;
char* pSrc = &sorted_distinct_numbers;
for (; i < sorted_distinct_numbers.size() / 8; ++i) {
uint64_t val = *((uint64_t*)(pSrc + i * 8));
if(val == 0){
continue;
}
int64_t lo = val & 0xFFFFFFFFull;
int64_t hi = val & (0xFFFFFFFFull << 32ull);
if(lo!=0){
}
if(hi!=0){
}
}

for (int j = i*8+1; j < sorted_distinct_numbers.size(); ++j) {
if (sorted_distinct_numbers[j] == 1) {
result.push_back(j);
}
}

//return dump_results(result);
//return result;
benchmark::DoNotOptimize(result);
}
}