Scatter-gather query

Deep dive

How it works

Parts:

• coordinator — prunes, fans out, owns deadline and merge
• shard map — key range or hash slot to shard and replicas
• shard executor — runs the query locally, returns a partial
• partial — mergeable state: count, sum, min, max, top-k list, sketch
• merger — folds partials in arrival order
• completeness flag — which shards answered; a late shard degrades the result instead of failing it

Steps:

1. Prune: keep shards whose range or partition can match the predicate (key range, hash of an equality key, min/max statistics). No usable key means all shards.
2. Fan out in parallel with one absolute deadline and per-shard limit k (deadline propagation).
3. Each shard filters, aggregates and limits locally; it returns only its partial.
4. The coordinator merges each partial on arrival.
5. After about a shard's p95 latency, send a backup to another replica; first answer wins (hedged requests).
6. At the deadline return the merge with partial = true and the missing shards, or fail if completeness is required.
7. Finalize (avg = sum / count); fetch full rows for the final top-k only.

            query + k + deadline
 client --> coordinator --prune--> [shard 0] --partial-+
               ^    \------------> [shard 2] --partial-+--> merge
               |   (shards 1, 3 skipped)    backup after p95
               +---- result + partial flag <-- deadline

Merge rules:

• Top-k by a per-document score is exact: each global top-k member is in its shard's top-k.
• Top-k by aggregated count can miss a term just below the cut on every shard. Elasticsearch asks each shard for shard_size terms (default size × 1.5 + 10) and reports doc_count_error_upper_bound.
• Distinct counts and quantiles merge as sketches (HyperLogLog, quantile sketches).
• Page p needs the top p × size from every shard. Elasticsearch caps from + size at 10,000; search_after with a point in time avoids it (cursor pagination).

Code

std only; compiled and run.

use std::cmp::Reverse;
use std::collections::BinaryHeap;
use std::sync::atomic::{AtomicUsize, Ordering::Relaxed};
use std::sync::{mpsc, Arc};
use std::{thread, time::{Duration, Instant}};

#[derive(Clone, Copy)]
struct Query { lo: u64, hi: u64, k: usize }

// Mergeable partial: every field combines associatively and commutatively.
#[derive(Clone, Debug, PartialEq)]
struct Partial { count: u64, sum: i64, min: i64, max: i64, top: Vec<(i64, u64)> } // top = (score, key), best first

fn top_k(items: impl Iterator<Item = (i64, u64)>, k: usize) -> Vec<(i64, u64)> {
    let mut h = BinaryHeap::new(); // min-heap of the k best so far
    for it in items { h.push(Reverse(it)); if h.len() > k { h.pop(); } }
    h.into_sorted_vec().into_iter().map(|Reverse(x)| x).collect()
}

impl Partial {
    fn empty() -> Self { Partial { count: 0, sum: 0, min: i64::MAX, max: i64::MIN, top: vec![] } }
    fn scan(rows: &[(u64, i64)], q: &Query) -> Self {
        let hit = || rows.iter().filter(|r| (q.lo..q.hi).contains(&r.0));
        let mut p = Self::empty();
        for r in hit() { p.count += 1; p.sum += r.1; p.min = p.min.min(r.1); p.max = p.max.max(r.1); }
        p.top = top_k(hit().map(|r| (r.1, r.0)), q.k);
        p
    }
    fn merge(&mut self, o: &Partial, k: usize) {
        self.count += o.count; self.sum += o.sum;
        self.min = self.min.min(o.min); self.max = self.max.max(o.max);
        self.top = top_k(self.top.iter().chain(&o.top).copied(), k);
    }
}

struct Shard { lo: u64, hi: u64, rows: Vec<(u64, i64)>, delay: Duration, hits: AtomicUsize }
impl Shard {
    fn overlaps(&self, q: &Query) -> bool { self.lo < q.hi && q.lo < self.hi } // pruning
    fn run(&self, q: &Query) -> Partial {
        self.hits.fetch_add(1, Relaxed);
        thread::sleep(self.delay);
        Partial::scan(&self.rows, q)
    }
}

/// Returns (merged partial, partial_result flag, shards contacted).
fn scatter_gather(shards: &Arc<Vec<Shard>>, q: Query, wait: Duration) -> (Partial, bool, usize) {
    let targets: Vec<usize> = (0..shards.len()).filter(|&i| shards[i].overlaps(&q)).collect();
    let (tx, rx) = mpsc::channel();
    for &i in &targets {
        let (tx, shards) = (tx.clone(), Arc::clone(shards));
        thread::spawn(move || { let _ = tx.send(shards[i].run(&q)); });
    }
    drop(tx);
    let deadline = Instant::now() + wait;
    let (mut acc, mut got) = (Partial::empty(), 0);
    while got < targets.len() {
        match rx.recv_timeout(deadline.saturating_duration_since(Instant::now())) {
            Ok(p) => { acc.merge(&p, q.k); got += 1; } // merge in arrival order
            Err(_) => break, // deadline: stop waiting
        }
    }
    (acc, got < targets.len(), targets.len())
}

fn main() {
    let rows: Vec<(u64, i64)> = (0..8000).map(|k| (k, (k * 2654435761 % 100_003) as i64 - 50_000)).collect();
    let mk = |slow: usize| Arc::new((0..8usize).map(|i| Shard {
        lo: i as u64 * 1000, hi: i as u64 * 1000 + 1000, rows: rows[i * 1000..][..1000].to_vec(),
        delay: Duration::from_millis(if i == slow { 300 } else { 2 }), hits: AtomicUsize::new(0),
    }).collect::<Vec<_>>());
    let (wait, q) = (Duration::from_millis(100), Query { lo: 0, hi: 8000, k: 10 });

    let shards = mk(99); // merged == single node, top-k == full sort
    let (m, partial, n) = scatter_gather(&shards, q, wait);
    assert!(!partial && n == 8 && m == Partial::scan(&rows, &q));
    let mut all: Vec<_> = rows.iter().map(|r| (r.1, r.0)).collect();
    all.sort_by(|a, b| b.cmp(a));
    assert_eq!(m.top, all[..10]);
    let parts: Vec<Partial> = shards.iter().map(|sh| sh.run(&q)).collect();
    let mut rev = Partial::empty();
    for p in parts.iter().rev() { rev.merge(p, q.k); }
    assert_eq!(rev, m);

    let shards = mk(99); // keys 1500..2500 touch shards 1 and 2 only
    let q2 = Query { lo: 1500, hi: 2500, k: 5 };
    let (m2, _, n2) = scatter_gather(&shards, q2, wait);
    assert!(n2 == 2 && m2 == Partial::scan(&rows, &q2));
    assert!([0, 3, 4, 5, 6, 7].iter().all(|&i| shards[i].hits.load(Relaxed) == 0));

    let shards = mk(3); // shard 3 needs 300 ms, deadline 100 ms: partial result of the other 7
    let (m3, partial, _) = scatter_gather(&shards, q, wait);
    let others: Vec<_> = rows.iter().filter(|r| r.0 / 1000 != 3).copied().collect();
    assert!(partial && m3 == Partial::scan(&others, &q));
    println!("ok: partial result covers {} of {} rows", m3.count, rows.len());
}

Complexity and constants

• n requests per query after pruning, each with a round trip.
• Merge O(n·k log k) with the heap; a k-way merge of sorted lists O(n·k log n).
• P(slow query) = 1 − (1 − p)^n for slow probability p per shard.

Efficiency

• Cost: measured on an Apple M4 Pro, one thread, release build. Merging one partial: 380-460 ns at k = 10, 4.0-4.7 µs at k = 100 (heap rebuilt from 2k items); 64 shards: 24 and 295 µs. Thread-per-shard fan-out with no work: p50 69 µs (4 shards), 206 µs (16), 845 µs (64), mostly thread spawn. Real queries are bound by network and slowest shard.
• Levers:
  • shard pruning — skips whole requests (Elasticsearch pre_filter_shard_size, Vitess vindexes, ClickHouse optimize_skip_unused_shards) — predicate has the shard key
  • partials, not rows — bytes scale with groups and k (ClickHouse sends aggregate states) — aggregations
  • small per-shard k — merge and wire cost ∝ k — top-k queries
  • hedged request after ~p95 — simulated, hedge at 3 ms, 1 % of shard requests at 50 ms instead of about 2 ms: p99 over 4/16/64 shards falls from 50 ms to 5.3/5.4/5.5 ms for about 1 % extra requests — idempotent reads with replicas
  • aggregator tree (Thanos Queriers stack), compressed partials — relieve coordinator fan-in — hundreds of shards
• Naive → standard → state of the art:
  • sequential queries, raw rows to the client · n × shard latency
  • parallel fan-out, pruning, partial aggregates, deadline · max shard latency
  • plus hedged requests, partial flag, aggregator tree, cursors · tail near shard p95

Straggler numbers are simulated (sampled latencies, no network); p99.9 stayed at 50 ms for 16 and 64 shards.

Implementation notes

• Put an absolute deadline in every sub-request; cancel the loser of a hedge.
• Batch per node; reuse connections.
• Break top-k ties by a unique key, so merge order cannot change the result.
• Cap in-flight requests (backpressure).
• Report skipped and failed shards in every response.

Pitfalls

• Replicas at different versions make paging skip or repeat rows; pin a snapshot.
• Clients ignoring the partial flag treat partial results as complete (Elasticsearch allow_partial_search_results defaults to true).
• Unbudgeted hedging adds load during overload; cap it at a few percent (circuit breaker).

Variants and alternatives

In production

Elasticsearch and OpenSearch search all shards of an index and merge hits. Vitess VTGate scatters to all shards when no vindex narrows the query. ClickHouse Distributed tables send intermediate aggregate states. Druid Brokers consolidate data-server results. Thanos Querier returns partial data with a warning (--query.partial-response). Trino plans show Aggregate(PARTIAL) stages.

Implementations

Sources

• Jeffrey Dean, Luiz André Barroso, The Tail at Scale, Communications of the ACM 56(2), 2013, pp. 74-80. paper
• Elastic, Terms aggregation. docs
• Elastic, Paginate search results. docs
• Elastic, Search API. docs
• ClickHouse, Distributed table engine. docs
• Vitess, Vindexes. docs
• Apache Druid, Broker. docs
• Thanos, Query. docs
• Trino, EXPLAIN. docs