Scatter-Gather Fan-Out and Tail Latency Multiplication

A request that touches one partition has the latency distribution of one server. A request that touches fifty has the distribution of the maximum of fifty samples from that same distribution. The shift is not a small penalty at the mean; it is a transformation of the shape of the curve, and it is the reason a cluster can look healthy in aggregate metrics while every user-facing operation times out.

The maximum of many samples

Suppose each partition’s response time is a random variable with median m and a tail such that one request in a hundred exceeds 3m. For a single-partition request, 1 per cent of requests exceed 3m. For a request that must gather from P independent partitions, the probability that at least one exceeds 3m is 1 − 0.99^P. At P = 50 that is 0.395: nearly forty per cent of requests exceed 3m, even though only one per cent of the underlying partition responses do.

The median of the fan-out request is where the probability of all P partitions being below it equals one half, so each partition must be below its own 0.5^(1/P) quantile. At P = 50 that is the 98.6th percentile of a single partition. The median of the fan-out is the 98.6th percentile of the single-node distribution. A cluster whose 99th percentile is three times its median will show a fan-out median of roughly three times its median.

This arithmetic is independent of the implementation. It follows from the maximum of independent samples, and it applies to any request that must wait for all of its parts. The only ways out are to reduce P, to make the tail shorter, or to stop waiting for all parts.

Where the tail comes from

The single-node tail is not noise. It has identifiable causes, and each responds to a different mitigation.

Queueing. A server with utilization ρ has a queueing delay that grows as 1/(1−ρ). At 50 per cent utilization the mean queue is one service time; at 90 per cent it is nine; at 99 per cent it is ninety-nine. The tail is dominated by whatever interval had the highest utilization, which is usually a burst rather than a steady state. Reducing P reduces the chance of hitting a busy server on any given request, and spreading load evenly reduces the probability that any server is at high utilization.

Garbage collection and compaction. A runtime pause or a storage-engine compaction holds the server’s attention for a period measured in milliseconds to hundreds of milliseconds. A single event affects many requests, producing a correlated spike rather than an independent tail. Correlated spikes break the independence assumption behind the maximum calculation: if all P partitions pause at once, the fan-out does not wait for one slow server, it waits for all of them.

Contention on shared resources. A lock held by a long operation, a connection pool exhausted by a burst, a page fault on a cold buffer. These produce a tail that depends on the request mix rather than on the server’s average load.

Network. Packet loss triggers a retransmission timeout, and that timer is measured in fractions of a second to whole seconds, far above the service times involved. A single lost packet on one partition’s response converts that response from a millisecond into hundreds of milliseconds. At a loss rate of 0.1 per cent and a request spanning 50 partitions, the probability that at least one response loses a packet is about 5 per cent, which is a tail of its own.

The last cause is the one most often overlooked, because the loss rate is invisible in application metrics and the delay appears as “one slow node” rather than “one lost packet”.

Reducing the fan-out

The cheapest improvement is to touch fewer partitions. That is a property of the data layout, not of the query engine.

A hash-partitioned store makes every range query a full fan-out, which is the cost described in Hash Partitioning and the Range Query It Destroys. A range-partitioned store touches only the partitions whose intervals intersect the query, which for a narrow range is one. The choice of partition function therefore sets the fan-out, and no query-level optimization recovers it.

A secondary index can narrow the fan-out for non-key predicates, but only if the index itself is partitioned in a way that lets the query identify the relevant partitions. An index that is hashed on the indexed value scatters the matching keys across every partition, so the query must consult every index partition and then fetch from every data partition, which doubles the fan-out. The design of index partitioning is covered in Local and Global Secondary Indexes Under Partitioning.

A covering index avoids the second hop. If the index contains the columns the query needs, the query is answered from the index alone and never fetches from the data partitions. The fan-out is then over index partitions only, which is one stage instead of two.

Hedged requests

When the fan-out cannot be reduced, the next lever is to stop waiting for the slowest response when a faster one is available. A hedged request sends a duplicate of the request to a second replica after a delay, and takes whichever response arrives first.

import asyncio

async def hedged(fetch, primary, backup, delay):
    first = asyncio.create_task(fetch(primary))
    done, pending = await asyncio.wait({first}, timeout=delay)
    if not done:                       # primary slower than the hedge delay
        pending.add(asyncio.create_task(fetch(backup)))
        done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
    for t in pending:
        t.cancel()                     # the loser's request is abandoned, not awaited
    return done.pop().result()

The delay is the tuning parameter and it should sit near the high end of the normal latency distribution, typically the 90th to 95th percentile of a single replica. Set it too low and the second request is sent for most requests, doubling load without benefit. Set it too high and the hedge arrives after the original would have completed.

Hedging converts the maximum over P into something closer to the maximum over P of a minimum of two samples, which shifts the whole curve left. The load cost is bounded by the fraction of requests that exceed the delay, which for a delay at the 95th percentile is 5 per cent extra requests. That is a favourable trade: a 5 per cent load increase buys a reduction in the fan-out tail that is much larger than 5 per cent, because the tail of the maximum is dominated by the slowest samples.

Hedging requires that the operation be idempotent, or that duplicate execution be harmless. Reads are naturally safe. Writes are not, and hedging a write means the write may be applied twice, which is safe only with an idempotency key or a conditional write. The general problem of duplicate application is not a partitioning problem, but a partitioned write path that hedges must solve it.

Hedging also interacts badly with the cause of the tail. If the tail is caused by a server that is queueing, hedging adds load to that server and lengthens its queue, which makes the tail worse for other requests. The hedge helps when the tail is random (a lost packet, a scheduled pause) and hurts when it is load-induced. Distinguishing the two at runtime is the hard part; a common policy is to hedge only when the target replica’s queue depth is below a threshold.

Partial results and deadlines

The third lever is to accept an incomplete answer. A query that gathers from 50 partitions can return results from 49 and flag the missing one, which is correct for a search and wrong for a sum.

The mechanism that makes this expressible is a deadline propagated with the request. The coordinator sets an absolute deadline, and every partition must return what it has by that deadline rather than running to completion. A partition that cannot finish returns a partial result and a marker, and the coordinator decides whether the aggregate is acceptable.

import asyncio

async def gather(fetch, parts, timeout):
    tasks = {asyncio.create_task(fetch(p, timeout)) for p in parts}
    done, pending = await asyncio.wait(tasks, timeout=timeout)
    for t in pending:
        t.cancel()
    results = [t.result() for t in done if not t.exception()]
    missing = [t for t in tasks if t not in done or t.exception()]
    return results, missing

The decision of acceptability belongs to the operation, not the transport. A count of matching documents can tolerate a missing partition if the count is reported as approximate. A sum of account balances cannot. The distinction has to be encoded in the request, and it is the same distinction that appears in quorum reads, where a read may return a stale value if it contacts fewer replicas than the write quorum.

There is a subtlety with partial results and correctness: a partition that times out may still be executing and may later apply a write. A coordinator that returned a partial result and then reports a failure to the client has told the client the operation did not happen, while the operation may yet complete on the missing partition. For reads this is harmless; for writes it is the standard “in-doubt” problem, and it requires the write to be idempotent or to be resolved by a later read.

Aggregation strategies

When all partitions must respond, the coordinator’s merge strategy determines how much of the tail is exposed.

Collect all, then merge. The coordinator waits for every response before producing anything. The latency is the maximum, and the memory is the sum. This is the default and the worst.

Streaming merge. The coordinator merges responses as they arrive and emits results incrementally. For a top-k query the client sees the first results before the slowest partition has answered, which improves perceived latency without changing the completion time. The final answer is still the maximum, but the time to first useful byte is the minimum.

Early termination. For a top-k query, once k results are known and every outstanding partition’s local maximum is below the current k-th value, the query can stop. This requires each partition to report a bound with its partial results, and it works well when the top-k values are concentrated in a few partitions. When they are not, the query degenerates to collect-all.

Hierarchical aggregation. Partitions report to intermediate aggregators, which report to the coordinator. The depth is log P, and each level’s maximum is over fewer samples, but each level adds a round trip. For a fan-out of 1000, a tree of depth 3 with fan-out 10 turns a 1000-way maximum into three 10-way maxima plus three round trips. The trade is favourable when the single-hop latency is high and the per-node tail is long.

The choice among these is driven by whether the answer can be produced incrementally and whether a bound is available. Both are properties of the query, not of the transport.

What the metrics hide

An aggregate latency metric over a fan-out operation reports the fan-out latency, which already includes the maximum. The per-partition metrics report the single-node distribution. The gap between them is the multiplication factor, and it is invisible unless the two are compared.

The comparison is worth making because the multiplication factor is a function of P, and P changes as the cluster grows. A query that fans out to 10 partitions on a small cluster and 100 on a large one has the same per-partition latency and a much worse end-to-end latency. The cluster’s growth silently degrades the query, and no single-node metric moves. This is one of the reasons partition counts should be kept coarse and the fan-out bounded by design rather than by accident.