Merging sorted integer vectors in C++

Question

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.

Answer 1

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;
}

Answer 2

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.

Answer 3

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[0];
        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){
              add(result, val, 0, i);
              add(result, val, 1, i);
              add(result, val, 2, i);
              add(result, val, 3, i);
            }
            if(hi!=0){
              add(result, val, 4, i);
              add(result, val, 5, i);
              add(result, val, 6, i);
              add(result, val, 7, i);
            }
        }

        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);
    }
}