Better sort algorithm for pre-sorted chunks

user_1310875 0
Viewed 2608

I have a very large list of elements (few millions) which are locally sorted in chunks of variable size.

a b c b i h a j k l e f g a b l...

I know the size, starting and ending position of every chunk in advance.

[a b c] [b i h m] [a j k l] [e f g] [a] [b l] ...

Is there a faster way to sort the list using an algorithm that can make use of the boundary information?

sorting algorithm
6 Answers

I have thought about the problem and came out with following generic solution in C++. 

template<typename T>
struct Triplet
{
    T elm;
    size_t listIdx;
    size_t elmIdx;

    Triplet(T e, size_t ldx, size_t eIdx) : elm(e), listIdx(ldx), elmIdx(eIdx) {}

    bool operator<(const Triplet &trp) const {
        return elm < trp.elm;
    }
    bool operator>(const Triplet &trp) const {
        return elm > trp.elm;
    }

};

template<typename T>
vector<T> mergeSortedLists(const vector<vector<T>> &vec)
{
    vector<T> ret;
    priority_queue<Triplet<T>, vector<Triplet<T>>, std::greater<Triplet<T>>> pq;
    for (size_t i = 0; i < vec.size(); i++) {
        if (vec[i].size() > 0) {
            pq.push({ vec[i][0], i, 0 });
        }
    }
    while (!pq.empty()) {
        Triplet<T> trp = pq.top();
        pq.pop();
        ret.push_back(trp.elm);
        if (trp.elmIdx < vec[trp.listIdx].size() - 1) {
            pq.push({ vec[trp.listIdx][trp.elmIdx + 1], trp.listIdx, trp.elmIdx + 1 });
        }
    }
    return ret;
}

The idea is simple: create a min-heap (in this case priority queue with the smallest element on top). First fill the heap with firsts elements of each list, represented as triplets: the element, the index of the list containing the element ind the index of the element inside of the list. Pop the top element, put it into solution vector, advance the element index one position and push the new triplet into the heap.

The algorithm works with any number of presorted lists, and the lists must not be of the same size. Here are some examples of usage:

int main()
{
    vector<vector<int>> vint = { {1,3,5}, {2,4,6}, {3,5,7} };
    vector<int> resInt = mergeSortedLists(vint);
    vector<int> combinedInt = { 1,3,5,2,4,6,3,5,7 };
    std::sort(combinedInt.begin(), combinedInt.end());
    assert(resInt == combinedInt);

    vector<vector<string>> vstr = { {"abc"}, {"adef"}, {"ccfgt"}, {"aaabb"} };
    vector<string> resStr = mergeSortedLists(vstr);
    vector<string> combinedStr = { "abc", "adef", "ccfgt", "aaabb" };
    std::sort(combinedStr.begin(), combinedStr.end());
    assert(resStr == combinedStr);

    vector<vector<char>> vchar = { {'a', 'b', 'c'}, {'d', 'e' }, {'b', 'd', 'f'}, {'e','f','g','h'} };
    vector<char> resChar = mergeSortedLists(vchar);
    vector<char> combinedChar = { 'a', 'b', 'c', 'd', 'e' , 'b', 'd', 'f','e','f', 'g', 'h' };
    std::sort(combinedChar.begin(), combinedChar.end());
    assert(resChar == combinedChar);
}

The algorithm runs faster than "brute force" putting all elements into a single list and sorting them. Let's assume we have m lists, every list of the size n. So, the "brute force" algorithm run time would be 

O(m * n) + O(m * n * log(m * n)) = O(m * n * log(m * n))

The run time of the algorithm above is 

O (m * n * log(m))

Here, the size of the heap is always <= m. O (m * n * log(m)) for pushing and popping elements from the heap.

The algorithm can be slightly improved as follows:

template<typename T>
vector<T> mergeSortedLists(const vector<vector<T>> &vec)
{
    vector<T> ret;
    priority_queue<Triplet<T>, vector<Triplet<T>>, std::greater<Triplet<T>>> pq;
    for (size_t i = 0; i < vec.size(); i++) {
        if (vec[i].size() > 0) {
            pq.push({ vec[i][0], i, 0 });
        }
    }
    while (!pq.empty()) {
        Triplet<T> trp = pq.top();
        pq.pop();
        ret.push_back(trp.elm);
        while ((trp.elmIdx < vec[trp.listIdx].size() - 1) && (vec[trp.listIdx][trp.elmIdx + 1] == ret.back())) {
            ret.push_back(vec[trp.listIdx][trp.elmIdx + 1]);
            trp.elmIdx++;
        }
        if (trp.elmIdx < vec[trp.listIdx].size() - 1) {
            pq.push({ vec[trp.listIdx][trp.elmIdx + 1], trp.listIdx, trp.elmIdx + 1 });
        }
    }
    return ret;
}

My measurements on a Windows machine (compiled with MS Visual Studio) show following results:

10 x 100000 (10 presorted vectors each of size 100000 ):
Brute Force algorithm: 0.0407124 s
My algorithm: 0.030523 s

50 x 100000 :
Brute Force algorithm: 0.237032 s
My algorithm: 0.217482 s

100 x 100000:
Brute Force algorithm: 0.422206 s
My algorithm: 0.356308 s

In all cases I filled vectors with random integers between 0 and 100000 and used exactly the same presorted lists for both algorithms.

user_60939530
Related