c++ - 多数元素-数组的一部分

Question

我有一个用整数填充的数组。我的工作是为数组的任何部分快速找到多数元素，我需要这样做... log n time ，不是线性的，但是事先我可以花一些时间准备数组。

例如:

1 5 2 7 7 7 8 4 6

并查询:
[4, 7]返回 7 [4, 8]返回 7 [1, 2]返回 0(不包含多数元素)，依此类推...

我需要为每个查询提供答案，如果可能的话，它需要快速执行。

为了准备，我可以使用 O(n log n)时间

Answer 1

O(log n)查询和O(n log n)预处理/空间可以通过查找并使用具有以下属性的多数间隔来实现:

对于输入数组中的每个值，可能会有一个或几个多数时间间隔(或者如果具有这些值的元素太稀疏，则可能没有任何间隔；我们不需要长度为1的多数时间间隔，因为它们可能仅对以下查询间隔有用)尺寸1，最好将其作为特殊情况处理)。

如果查询间隔完全位于这些多数时间间隔之一内，则对应的值可以作为此查询间隔的多数元素。

如果没有完全包含查询间隔的多数间隔，则对应的值不能成为此查询间隔的多数元素。

输入数组的每个元素都被O(log n)多数间隔覆盖。

换句话说，多数间隔的唯一目的是为任何查询间隔提供O(log n)多数元素候选。

该算法使用以下 数据结构:

输入数组(map<Value, vector<Position>>)中每个值的位置列表。另外，这里也可以使用unordered_map来提高性能(但我们需要提取所有键并对它们进行排序，以便按正确的顺序填充结构3)。

每个值的多数数间隔列表(vector<Interval>)。

用于处理查询的数据结构(vector<small_map<Value, Data>>)。其中Data包含两个来自结构＃1的适当 vector 的索引，指向具有给定值的元素的下一个/上一个位置。 更新:由于使用@justhalf，最好在Data中存储具有给定值的元素的累积频率。 small_map可以实现为成对的排序 vector -预处理将追加已排序顺序的元素，并且查询仅将small_map用于线性搜索。

预处理:

扫描输入数组并将当前位置插入结构＃1中的适当 vector 。

对结构＃1中的每个 vector 执行步骤3 .. 4。

将位置列表转换为多数数间隔列表。请参阅下面的详细信息。

对于大多数间隔之一覆盖的输入数组的每个索引，将数据插入结构3的适当元素:具有该值(或该值的累积频率)的值和前一/后一元素的位置。

查询:

如果查询间隔长度为1，则返回源数组的相应元素。

对于查询间隔的起点，获取第三结构的 vector 的相应元素。对于该映射的每个元素，请执行步骤3。与该映射并行地扫描与查询间隔结束点相对应的映射的所有元素，以允许步骤3的O(1)复杂度(而不是O(log log n))。

如果与查询间隔终点相对应的映射包含匹配值，则计算s3[stop][value].prev - s3[start][value].next + 1。如果大于查询间隔的一半，则返回值。如果使用累积频率代替下一个/上一个索引，请计算s3[stop+1][value].freq - s3[start][value].freq。

如果在步骤3上找不到任何内容，则返回“Nothing”。

算法的主要部分是从位置列表中获取多数区间:

为列表中的每个位置分配权重:number_of_matching_values_to_the_left - number_of_nonmatching_values_to_the_left。

仅将权重严格降低的顺序(贪心地)过滤到“前缀”数组:for (auto x: positions) if (x < prefix.back()) prefix.push_back(x);。

仅将权重严格按递增顺序(贪婪，向后)过滤到“后缀”数组:reverse(positions); for (auto x: positions) if (x > suffix.back()) suffix.push_back(x);。

一起扫描“前缀”和“后缀”数组，找到从每个“前缀”元素到“后缀”数组中对应位置以及从每个“后缀”元素到“前缀”数组中对应位置的间隔。 (如果所有“后缀”元素的权重均小于给定“前缀”元素的权重，或者它们的位置不在其右侧，则不会生成间隔；如果没有“后缀”元素的权重与给定“前缀”元素的权重完全相同，获取权重较大的最近的“后缀”元素，并在此权重差向右扩展间隔。

合并重叠的间隔。

该算法保证多数间隔的属性1 .. 3。至于属性＃4，我可以想象覆盖最大多数数间隔的某些元素的唯一方法是: 11111111222233455666677777777。此处 4间隔覆盖了元素 2 * log n，因此似乎可以满足此属性。请参阅本文结尾处有关此属性的更多正式证明。

示例:

对于输入数组“0 1 2 0 0 1 1 0”，将生成以下位置列表:

value  positions
    0  0 3 4 7
    1  1 5 6
    2  2

值 0的位置将获得以下属性:

weights:    0:1 3:0 4:1 7:0
prefix:     0:1 3:0          (strictly decreasing)
suffix:     4:1 7:0          (strictly increasing when scanning backwards)
intervals:  0->4 3->7 4->0 7->3
merged intervals: 0-7

值 1的位置将获得以下属性:

weights:    1:0  5:-2  6:-1
prefix:     1:0  5:-2
suffix:     1:0  6:-1
intervals:  1->none 5->6+1 6->5-1 1->none
merged intervals: 4-7

查询数据结构:

positions value next prev
        0     0    0    x
     1..2     0    1    0
        3     0    1    1
        4     0    2    2
        4     1    1    x
        5     0    3    2
    ...

查询[0,4]:

prev[4][0]-next[0][0]+1=2-0+1=3
query size=5
3>2.5, returned result 0

查询[2,5]:

prev[5][0]-next[2][0]+1=2-1+1=2
query size=4
2=2, returned result "none"

注意，没有尝试检查元素“1”，因为它的多数间隔不包括这些间隔中的任何一个。

属性＃4的证明:

多数区间的构造方式是，严格地说，其所有元素的1/3以上具有相应的值。对于诸如 any*(m-1) value*m any*m的子数组(例如 01234444456789)，此比率最接近1/3。

为了使这一证明更加明显，我们可以将每个间隔表示为2D点:水平轴代表每个可能的起点，垂直轴代表每个可能的终点(请参见下图)。

所有有效间隔都位于对角线上或上方。白色矩形表示覆盖某个数组元素的所有间隔(在其右下角表示为单位大小的间隔)。

让我们用共享相同右下角的大小为1、2、4、8、16 ...的正方形覆盖这个白色矩形。这会将白色区域划分为类似于黄色区域的O(log n)区域(以及大小为1的单个正方形包含大小为1的单个间隔，该算法将忽略该间隔)。

让我们计算在黄色区域中可以放置多少个多数间隔。一个间隔(位于最接近对角线的角落)占距对角线最远的间隔的元素的1/4(并且该最大间隔包含属于黄色区域中任何间隔的所有元素)。这意味着最小间隔至少包含整个黄色区域可用的1/12值。因此，如果我们尝试在黄色区域放置12个间隔，则没有足够的元素用于不同的值。因此，黄色区域不能包含超过11个多数票区间。白色矩形不能包含超过 11 * log n个多数时间间隔。证明已完成。
11 * log n被高估了。就像我之前说的，很难想象除了覆盖某些元素的 2 * log n多数时间间隔之外。甚至该值也远大于覆盖多数间隔的平均数。

C++ 11实现。 在 ideone或此处查看:

#include <iostream>
#include <vector>
#include <map>
#include <algorithm>
#include <functional>
#include <random>

constexpr int SrcSize = 1000000;
constexpr int NQueries = 100000;

using src_vec_t = std::vector<int>;
using index_vec_t = std::vector<int>;
using weight_vec_t = std::vector<int>;
using pair_vec_t = std::vector<std::pair<int, int>>;
using index_map_t = std::map<int, index_vec_t>;
using interval_t = std::pair<int, int>;
using interval_vec_t = std::vector<interval_t>;
using small_map_t = std::vector<std::pair<int, int>>;
using query_vec_t = std::vector<small_map_t>;

constexpr int None = -1;
constexpr int Junk = -2;

src_vec_t generate_e()
{ // good query length = 3
    src_vec_t src;
    std::random_device rd;
    std::default_random_engine eng{rd()};
    auto exp = std::bind(std::exponential_distribution<>{0.4}, eng);

    for (int i = 0; i < SrcSize; ++i)
    {
        int x = exp();
        src.push_back(x);
        //std::cout << x << ' ';
    }

    return src;
}

src_vec_t generate_ep()
{ // good query length = 500
    src_vec_t src;
    std::random_device rd;
    std::default_random_engine eng{rd()};
    auto exp = std::bind(std::exponential_distribution<>{0.4}, eng);
    auto poisson = std::bind(std::poisson_distribution<int>{100}, eng);

    while (int(src.size()) < SrcSize)
    {
        int x = exp();
        int n = poisson();

        for (int i = 0; i < n; ++i)
        {
            src.push_back(x);
            //std::cout << x << ' ';
        }
    }

    return src;
}

src_vec_t generate()
{
    //return generate_e();
    return generate_ep();
}

int trivial(const src_vec_t& src, interval_t qi)
{
    int count = 0;
    int majorityElement = 0; // will be assigned before use for valid args

    for (int i = qi.first; i <= qi.second; ++i)
    {
        if (count == 0)
            majorityElement = src[i];

        if (src[i] == majorityElement) 
           ++count;
        else 
           --count;
    }

    count = 0;
    for (int i = qi.first; i <= qi.second; ++i)
    {
        if (src[i] == majorityElement)
            count++;
    }

    if (2 * count > qi.second + 1 - qi.first)
        return majorityElement;
    else
        return None;
}

index_map_t sort_ind(const src_vec_t& src)
{
    int ind = 0;
    index_map_t im;

    for (auto x: src)
        im[x].push_back(ind++);

    return im;
}

weight_vec_t get_weights(const index_vec_t& indexes)
{
    weight_vec_t weights;

    for (int i = 0; i != int(indexes.size()); ++i)
        weights.push_back(2 * i - indexes[i]);

    return weights;
}

pair_vec_t get_prefix(const index_vec_t& indexes, const weight_vec_t& weights)
{
    pair_vec_t prefix;

    for (int i = 0; i != int(indexes.size()); ++i)
        if (prefix.empty() || weights[i] < prefix.back().second)
            prefix.emplace_back(indexes[i], weights[i]);

    return prefix;
}

pair_vec_t get_suffix(const index_vec_t& indexes, const weight_vec_t& weights)
{
    pair_vec_t suffix;

    for (int i = indexes.size() - 1; i >= 0; --i)
        if (suffix.empty() || weights[i] > suffix.back().second)
            suffix.emplace_back(indexes[i], weights[i]);

    std::reverse(suffix.begin(), suffix.end());
    return suffix;
}

interval_vec_t get_intervals(const pair_vec_t& prefix, const pair_vec_t& suffix)
{
    interval_vec_t intervals;
    int prev_suffix_index = 0; // will be assigned before use for correct args
    int prev_suffix_weight = 0; // same assumptions

    for (int ind_pref = 0, ind_suff = 0; ind_pref != int(prefix.size());)
    {
        auto i_pref = prefix[ind_pref].first;
        auto w_pref = prefix[ind_pref].second;

        if (ind_suff != int(suffix.size()))
        {
            auto i_suff = suffix[ind_suff].first;
            auto w_suff = suffix[ind_suff].second;

            if (w_pref <= w_suff)
            {
                auto beg = std::max(0, i_pref + w_pref - w_suff);

                if (i_pref < i_suff)
                    intervals.emplace_back(beg, i_suff + 1);

                if (w_pref == w_suff)
                    ++ind_pref;

                ++ind_suff;
                prev_suffix_index = i_suff;
                prev_suffix_weight = w_suff;
                continue;
            }
        }

        // ind_suff out of bounds or w_pref > w_suff:
        auto end = prev_suffix_index + prev_suffix_weight - w_pref + 1;
        // end may be out-of-bounds; that's OK if overflow is not possible
        intervals.emplace_back(i_pref, end);
        ++ind_pref;
    }

    return intervals;
}

interval_vec_t merge(const interval_vec_t& from)
{
    using endpoints_t = std::vector<std::pair<int, bool>>;
    endpoints_t ep(2 * from.size());

    std::transform(from.begin(), from.end(), ep.begin(),
                   [](interval_t x){ return std::make_pair(x.first, true); });

    std::transform(from.begin(), from.end(), ep.begin() + from.size(),
                   [](interval_t x){ return std::make_pair(x.second, false); });

    std::sort(ep.begin(), ep.end());

    interval_vec_t to;
    int start; // will be assigned before use for correct args
    int overlaps = 0;

    for (auto& x: ep)
    {
        if (x.second) // begin
        {
            if (overlaps++ == 0)
                start = x.first;
        }
        else // end
        {
            if (--overlaps == 0)
                to.emplace_back(start, x.first);
        }
    }

    return to;
}

interval_vec_t get_intervals(const index_vec_t& indexes)
{
    auto weights = get_weights(indexes);
    auto prefix = get_prefix(indexes, weights);
    auto suffix = get_suffix(indexes, weights);
    auto intervals = get_intervals(prefix, suffix);
    return merge(intervals);
}

void update_qv(
    query_vec_t& qv,
    int value,
    const interval_vec_t& intervals,
    const index_vec_t& iv)
{
    int iv_ind = 0;
    int qv_ind = 0;
    int accum = 0;

    for (auto& interval: intervals)
    {
        int i_begin = interval.first;
        int i_end = std::min<int>(interval.second, qv.size() - 1);

        while (iv[iv_ind] < i_begin)
        {
            ++accum;
            ++iv_ind;
        }

        qv_ind = std::max(qv_ind, i_begin);

        while (qv_ind <= i_end)
        {
            qv[qv_ind].emplace_back(value, accum);

            if (iv[iv_ind] == qv_ind)
            {
                ++accum;
                ++iv_ind;
            }

            ++qv_ind;
        }
    }
}

void print_preprocess_stat(const index_map_t& im, const query_vec_t& qv)
{
    double sum_coverage = 0.;
    int max_coverage = 0;

    for (auto& x: qv)
    {
        sum_coverage += x.size();
        max_coverage = std::max<int>(max_coverage, x.size());
    }

    std::cout << "             size = " << qv.size() - 1 << '\n';
    std::cout << "           values = " << im.size() << '\n';
    std::cout << "     max coverage = " << max_coverage << '\n';
    std::cout << "     avg coverage = " << sum_coverage / qv.size() << '\n';
}

query_vec_t preprocess(const src_vec_t& src)
{
    query_vec_t qv(src.size() + 1);
    auto im = sort_ind(src);

    for (auto& val: im)
    {
        auto intervals = get_intervals(val.second);
        update_qv(qv, val.first, intervals, val.second);
    }

    print_preprocess_stat(im, qv);
    return qv;
}

int do_query(const src_vec_t& src, const query_vec_t& qv, interval_t qi)
{
    if (qi.first == qi.second)
        return src[qi.first];

    auto b = qv[qi.first].begin();
    auto e = qv[qi.second + 1].begin();

    while (b != qv[qi.first].end() && e != qv[qi.second + 1].end())
    {
        if (b->first < e->first)
        {
            ++b;
        }
        else if (e->first < b->first)
        {
            ++e;
        }
        else // if (e->first == b->first)
        {
            // hope this doesn't overflow
            if (2 * (e->second - b->second) > qi.second + 1 - qi.first)
                return b->first;

            ++b;
            ++e;
        }
    }

    return None;
}

int main()
{
    std::random_device rd;
    std::default_random_engine eng{rd()};
    auto poisson = std::bind(std::poisson_distribution<int>{500}, eng);
    int majority = 0;
    int nonzero = 0;
    int failed = 0;

    auto src = generate();
    auto qv = preprocess(src);

    for (int i = 0; i < NQueries; ++i)
    {
        int size = poisson();
        auto ud = std::uniform_int_distribution<int>(0, src.size() - size - 1);
        int start = ud(eng);
        int stop = start + size;
        auto res1 = do_query(src, qv, {start, stop});
        auto res2 = trivial(src, {start, stop});
        //std::cout << size << ": " << res1 << ' ' << res2 << '\n';

        if (res2 != res1)
            ++failed;

        if (res2 != None)
        {
            ++majority;

            if (res2 != 0)
                ++nonzero;
        }
    }

    std::cout << "majority elements = " << 100. * majority / NQueries << "%\n";
    std::cout << " nonzero elements = " << 100. * nonzero / NQueries << "%\n";
    std::cout << "          queries = " << NQueries << '\n';
    std::cout << "           failed = " << failed << '\n';

    return 0;
}

相关工作:

正如 other answer to this question中指出的那样，还有其他工作已经解决了这个问题: "Range majority in constant time and linear space" by S. Durocher, M. He, I Munro, P.K. Nicholson, M. Skala。

对于查询时间:用 O(1)代替 O(log n)以及对于空间:用 O(n)代替 O(n log n)，本文提出的算法具有更好的渐进复杂性。

更好的空间复杂度允许该算法处理更大的数据集(与该答案中提出的算法相比)。预处理数据所需的内存更少，更规则的数据访问模式更可能使此算法更快地预处理数据。但是查询时间并非易事...

假设我们从论文中获得了最适合算法的输入数据:n = 1000000000(很难想象，在2013年，具有超过10..30 GB内存的系统)。

此答案中提出的算法需要为每个查询处理多达120个(或2个查询边界* 2 * log n)元素。但是它执行非常简单的操作，类似于线性搜索。并且它顺序地访问两个连续的内存区域，因此对缓存友好。

本文中的算法需要为每个查询执行多达20个操作(或2个查询边界* 5个候选对象* 2个小波树级别)。这少了六倍。但是每个操作都比较复杂。每个用于位计数器的简洁表示的查询本身都包含一个线性搜索(这意味着20个线性搜索而不是一个)。最糟糕的是，每个这样的操作都应访问几个独立的内存区域(除非查询大小，因此四倍大小非常小)，因此查询是不友好的缓存。这意味着每个查询(虽然是一个恒定时间的操作)都非常慢，可能比这里提出的算法要慢。如果我们减小输入数组的大小，则增加此处提出的算法的机会。

本文算法的实际缺点是小波树和简洁的位计数器实现。从头开始实现它们可能会非常耗时。使用预先存在的实现并不总是很方便。