A query whose predicate is not the partition key cannot be routed, so it goes to all N shards and waits for the last answer. Both halves of that sentence cost something: the fleet's request rate is multiplied by N, and the query's latency is the maximum of N shard latencies rather than one of them. The partition count is the same dial for both, and for how many consumers can read in parallel.
Core ideas
Two costs from one fan-out, plus a third consequence of the partition count that is easy to forget because it lives on a different team's diagram.
Both are true and only one of them is usually said. Each shard does 1/N of the scan, which is why the query is fast at all; and the fleet receives N requests where it used to receive one. At 24 shards and 400 queries a second that is 9 600 shard requests a second against the 400 a routed query would leave — and it is the second number that sizes connection pools, thread pools and every per-request cost in the system.
This is the maximum of “Stragglers” in a different costume. A shard whose own p99 is 90 ms gives a 24-way query a p99 of 250 ms, because the query is fast only if all 24 were fast. The lab computes the query's distribution exactly from the per-shard one and also samples it from a seeded run, and the two agree at 250 ms.
Raising N reduces per-shard work, raises the fleet's request rate, worsens the latency tail, and raises the ceiling on how many consumers can read the data in parallel — that ceiling is N itself, because a partition is the unit of consumption. Choosing N for any one of those in isolation is how a system ends up with a tail nobody chose.
The material
Why some queries cannot be routed, what the fan-out does to the request rate and to the tail, and what else the partition count is quietly deciding.
A scatter-gather query is one whose predicate does not determine the shard, so the router sends it to all N shards (scatter) and combines the partial results (gather). A routed query names the partition key and goes to one shard. The request amplification of a scatter-gather is N: shard requests per client query.
Which queries are unroutable is decided by the partitioning function and nothing else. Partition by user and “everything for user 91” is routed while “everything created yesterday” is a scatter; partition by day and the two swap. Every partitioning choice makes one family of queries cheap and the rest of them scatters, which is why the choice belongs with the access patterns rather than with the data model.
The amplification is the figure that surprises people, because the sentence “we sharded, so each node does less work” is true and is not the whole of it. Each node does less work per query; the node receives N times as many queries, so a fleet serving 400 client queries a second is handling 9 600 requests a second. Everything that is priced per request — a connection, a lock acquisition, a log line, a tracing span — is multiplied by 24.
At two shards the query's p99 is already 250 ms — the worst outcome in the per-shard distribution — because 0.99² = 0.9801, which is below the 0.99 a p99 needs. By sixteen shards the p90 has reached 250 ms too. The per-shard distribution never changed; the query is simply asking more shards to all be fast at once, and “Tail Amplification under Fan-out” is the same calculation stated in general.
A partition is also the unit of parallel consumption. If a downstream reader wants to process this dataset with more than N workers, it cannot: worker N + 1 has no partition to read, and at N = 24 the ceiling is 24 regardless of how much hardware is available. So the partition count that was chosen for storage and load has also set a limit on a pipeline nobody in that conversation was thinking about, and raising it later means the rehash of “Rehashing When N Changes”.
The direct fix is to make the query routable, by partitioning on the attribute it filters on — which usually means another copy of the data partitioned differently, and that is exactly the global secondary index of the next lesson. Failing that, the levers are the ones from “Stragglers”: narrow the per-shard tail, accept a partial result, or hedge. What does not help is raising N, which makes both costs worse.
It is worth writing the two numbers down together whenever a scatter-gather is proposed, because they trade against each other in a way that a single figure hides. 9 600 requests a second is a capacity question and 250 ms is a latency question; a fleet can be sized for the first and still fail the second, and a tail fix that adds hedged requests makes the first worse.
Interactive
N requests per query, and a tail that is the maximum of N
Method
Two numbers, always both. A scatter-gather that has been costed only for capacity or only for latency has been costed for neither.
A query is unroutable when its predicate does not determine the partition key. If it does determine it and the router is not using it, that is a routing bug and none of this arithmetic applies.
N × rate is what the shards receive, and it is the number that sizes connection pools, per-request overheads and anything billed per call. Write the routed rate beside it so the multiple is visible.
P(query ≤ t) = P(shard ≤ t)ᴺ. Take the p99 and compare it with one shard's p99. At the lab's numbers those are 250 ms and 90 ms, and the gap is the whole cost of the fan-out on the latency side.
Consumer parallelism is capped at N, and the rehash cost of changing N later is N/(N+1) of the dataset under mod assignment. Both belong in the same decision, because both are hard to revisit once the data is placed.
The request rate first, because it is a multiplication; then the tail, because it is not.
Two shards. The p99 of a fan-out query reaches the worst per-shard outcome at N = 2 with this distribution, which is worth holding on to: the tail cost of scatter-gather is not something that arrives at large fleet sizes, it arrives immediately.
For a faded rehearsal, keep 24 shards and narrow the per-shard tail to 10:80, 25:16, 90:4 — the 250 ms outcome removed entirely. The supplied first move is the new per-shard p99, which is still 90 ms. Compute the query's p99 at N = 24, say how much the fleet's request rate changed, and then state which of the two costs of scatter-gather that change addressed and which it left exactly where it was.
You should be able to compute the amplified request rate and the query's tail from a per-shard distribution, say at what shard count the tail reached the worst per-shard outcome, and name what else the partition count has decided.
Something to change about this lesson? Write it down, tick it off when it is handled, and export the list when you want to act on it. Kept in this browser only: nothing is sent anywhere, and nobody is notified.
No recommendations yet for this lesson.