Parallel Binary Search(並列二分探索)による効率的なクエリ処理

並列二分探索の基本概念

通常の二分探索は単一の値に対して適用されますが、複数のクエリに対して個別に二分探索を行うと計算量が爆発します。並列二分探索(Parallel Binary Search / 整体二分)は、複数のクエリを同時に処理することでこの問題を解決するオフラインアルゴリズムです。

核心的なアイデアは、すべてのクエリの探索範囲を値域 [low, high] とし、定義域 [L, R](操作の時間軸)に対して再帰的に分割していくことです。各再帰レベルで中間値 mid を用いてクエリを分類し、値域を半分に絞り込みます。

静的区間k番目の要素

配列内の区間 [l, r] におけるk番目に小さい値を求める問題を考えます。永続セグメント木を用いる解法もありますが、並列二分探索では以下のように実装できます。

値域に対して二分探索を行い、各 mid に対してFenwick Tree(BIT)を用いて「値が mid 以下の要素が区間内にいくつあるか」を効率的に計算します。クエリは結果に応じて左右の部分列に分割され、再帰的に処理されます。

#include <bits/stdc++.h>
using namespace std;

struct Operation {
    int type;      // 0: 要素追加, 1: クエリ
    int left, right;
    int param;     // 追加: 値, クエリ: k
    int idx;       // クエリ番号または位置
};

struct Fenwick {
    vector<int> bit;
    int n;
    Fenwick(int n = 0) { init(n); }
    void init(int n_) {
        n = n_;
        bit.assign(n + 1, 0);
    }
    void add(int pos, int delta) {
        for (; pos <= n; pos += pos & -pos) bit[pos] += delta;
    }
    int sum(int pos) {
        int res = 0;
        for (; pos > 0; pos -= pos & -pos) res += bit[pos];
        return res;
    }
    int rangeSum(int l, int r) { return sum(r) - sum(l - 1); }
};

vector<Operation> ops, bufLeft, bufRight;
vector<int> answers;
Fenwick fw;

void parallelBS(int valLow, int valHigh, int opL, int opR) {
    if (opL > opR) return;
    if (valLow == valHigh) {
        for (int i = opL; i <= opR; ++i)
            if (ops[i].type == 1) answers[ops[i].idx] = valLow;
        return;
    }
    
    int mid = (valLow + valHigh) >> 1;
    bufLeft.clear(); bufRight.clear();
    
    for (int i = opL; i <= opR; ++i) {
        if (ops[i].type == 0) {  // 要素追加
            if (ops[i].param <= mid) {
                fw.add(ops[i].idx, 1);
                bufLeft.push_back(ops[i]);
            } else {
                bufRight.push_back(ops[i]);
            }
        } else {  // クエリ
            int cnt = fw.rangeSum(ops[i].left, ops[i].right);
            if (ops[i].param <= cnt) {
                bufLeft.push_back(ops[i]);
            } else {
                ops[i].param -= cnt;  // 左側の数を引く
                bufRight.push_back(ops[i]);
            }
        }
    }
    
    // BITを復元
    for (auto &op : bufLeft)
        if (op.type == 0) fw.add(op.idx, -1);
    
    // 配列を再構成
    int split = opL;
    for (auto &op : bufLeft) ops[split++] = op;
    int midPos = split;
    for (auto &op : bufRight) ops[split++] = op;
    
    parallelBS(valLow, mid, opL, midPos - 1);
    parallelBS(mid + 1, valHigh, midPos, opR);
}

int main() {
    ios::sync_with_stdio(false);
    cin.tie(nullptr);
    
    int n, q;
    cin >> n >> q;
    
    int minVal = INT_MAX, maxVal = INT_MIN;
    ops.reserve(n + q);
    
    for (int i = 1; i <= n; ++i) {
        int x; cin >> x;
        minVal = min(minVal, x);
        maxVal = max(maxVal, x);
        ops.push_back({0, -1, -1, x, i});  // 初期要素
    }
    
    int queryCnt = 0;
    for (int i = 0; i < q; ++i) {
        int l, r, k; cin >> l >> r >> k;
        ops.push_back({1, l, r, k, queryCnt++});
    }
    
    fw.init(n);
    answers.resize(queryCnt);
    parallelBS(minVal, maxVal, 0, (int)ops.size() - 1);
    
    for (int x : answers) cout << x << '\n';
    return 0;
}

動的(更新付き)区間k番目の要素

要素の更新が発生する場合、更新操作を「削除+挿入」のペアとして扱います。値の変更を old_value の削除と new_value の挿入に分解することで、同じ並列二分探索の枠組みが適用可能です。

#include <bits/stdc++.h>
using namespace std;

struct Cmd {
    int kind;      // 0: 挿入, 1: 削除, 2: クエリ
    int l, r, v, id;
};

struct BIT {
    vector<int> t; int n;
    BIT(int n = 0) { init(n); }
    void init(int n_) { n = n_; t.assign(n + 2, 0); }
    void modify(int p, int d) { for (; p <= n; p += p & -p) t[p] += d; }
    int prefix(int p) { int s = 0; for (; p; p -= p & -p) s += t[p]; return s; }
    int query(int l, int r) { return prefix(r) - prefix(l - 1); }
};

vector<Cmd> seq, tmpL, tmpR;
vector<int> curVal;      // 現在の値
vector<int> res;
BIT bit;

void solve(int lo, int hi, int L, int R) {
    if (L > R) return;
    if (lo == hi) {
        for (int i = L; i <= R; ++i)
            if (seq[i].kind == 2) res[seq[i].id] = lo;
        return;
    }
    int mid = (lo + hi) >> 1;
    tmpL.clear(); tmpR.clear();
    
    for (int i = L; i <= R; ++i) {
        if (seq[i].kind == 2) {  // クエリ
            int cnt = bit.query(seq[i].l, seq[i].r);
            if (seq[i].v <= cnt) tmpL.push_back(seq[i]);
            else { seq[i].v -= cnt; tmpR.push_back(seq[i]); }
        } else {  // 挿入または削除
            if (seq[i].v <= mid) {
                bit.modify(seq[i].id, seq[i].kind == 0 ? 1 : -1);
                tmpL.push_back(seq[i]);
            } else tmpR.push_back(seq[i]);
        }
    }
    
    // ロールバック
    for (auto &c : tmpL)
        if (c.kind != 2) bit.modify(c.id, c.kind == 0 ? -1 : 1);
    
    int m = L;
    for (auto &c : tmpL) seq[m++] = c;
    int split = m;
    for (auto &c : tmpR) seq[m++] = c;
    
    solve(lo, mid, L, split - 1);
    solve(mid + 1, hi, split, R);
}

int main() {
    ios::sync_with_stdio(false); cin.tie(nullptr);
    
    int n, m; cin >> n >> m;
    curVal.resize(n + 1);
    int mn = INT_MAX, mx = INT_MIN;
    
    for (int i = 1; i <= n; ++i) {
        cin >> curVal[i];
        mn = min(mn, curVal[i]);
        mx = max(mx, curVal[i]);
        seq.push_back({0, -1, -1, curVal[i], i});
    }
    
    int qid = 0;
    for (int i = 0; i < m; ++i) {
        char op; cin >> op;
        if (op == 'C') {  // 変更
            int pos, newVal; cin >> pos >> newVal;
            mn = min(mn, newVal); mx = max(mx, newVal);
            seq.push_back({1, -1, -1, curVal[pos], pos});  // 削除
            seq.push_back({0, -1, -1, newVal, pos});      // 挿入
            curVal[pos] = newVal;
        } else {  // クエリ
            int l, r, k; cin >> l >> r >> k;
            seq.push_back({2, l, r, k, qid++});
        }
    }
    
    bit.init(n);
    res.resize(qid);
    solve(mn, mx, 0, (int)seq.size() - 1);
    
    for (int i = 0; i < qid; ++i) cout << res[i] << '\n';
    return 0;
}

計算量と考察

並列二分探索の時間計算量は O((N + Q) log V log N) となります。ここで N は要素数、Q はクエリ数、V は値域の大きさです。log V は値域に対する二分探索の深さ、log N はBIT操作のコストに対応します。

永続セグメント木による解法(O((N + Q) log N))と比較すると漸近的には劣りますが、定数倍が軽く実装も比較的シンプルであるため、競技プログラミングで有用です。また、オンライン処理が不要な場面では強力な選択肢となります。

タグ: parallel-binary-search offline-algorithm fenwick-tree order-statistics divide-and-conquer

8月8日 04:16 投稿