Cross-Partition Operations: Paying for What the Split Took Away

A query that cannot be answered from one partition must be sent to several and merged. Three costs stack up, and the third is the one that surprises people.

Work multiplies. P partitions each do work; the coordinator merges. For a LIMIT 10 ORDER BY x across 50 partitions, each must return its top 10 — 500 rows fetched to produce 10.

Latency becomes a maximum, not a mean. The query finishes when the slowest partition answers. If each partition independently answers within its p99 latency, the probability that *all* P are fast is 0.99^P. At P = 50 that is 61%; at P = 100 it is 37%. So a 100-way fan-out experiences its per-partition p99 on nearly two thirds of queries — the median of the fan-out is roughly the p99 of a shard. This is Fan Out to 100 and the Component’s Tail Becomes the System’s Median and it is the single most important number in this section.

The result is not a snapshot. Each partition answered at a different moment. A cross-partition COUNT can double-count a row that moved between partitions, or miss one, unless the reads are taken at a common snapshot timestamp — which requires either a global clock (There Is No Global Clock) or an explicit coordination step.

The mitigations are the usual ones and they are all about *avoiding* the fan-out rather than speeding it up: choose the partition key to match the dominant query, maintain a differently-partitioned copy for the other access pattern (Materialized Views: A Read Model That Lags), or, when you must fan out, use hedging (Send a Second Request After p95 and Take Whichever Answers First) and per-partition deadlines (Pass the Remaining Budget Down, Not a Fresh One) so one slow shard cannot hold the whole query.

Joins. A join is cheap when both sides of the join key live in the same partition — co-partitioning. Choose the same partition key for orders and order_items and the join is local, forever. When they are not co-partitioned there are two strategies, both borrowed from parallel databases: broadcast the small side to every partition holding the large side (cheap when one side genuinely is small), or shuffle both sides by the join key so matching rows meet on the same node (The Shuffle Is the Job). Shuffle is correct and expensive; it moves data proportional to the input size, over the network, per query.

Aggregations decompose exactly when the function is associative and commutative. SUM, COUNT, MIN, MAX and AVG (as a sum/count pair) reduce partially at each partition and merge trivially — the fan-out cost is a handful of numbers per partition regardless of data size. That is why they feel free.

DISTINCT, COUNT(DISTINCT), MEDIAN and percentiles do not decompose. There is no partial result smaller than the data: to know the global distinct count you must, in principle, see every value. The practical resolutions are exact-but-expensive (shuffle by the value being counted so each distinct value lands in one place) or approximate-but-cheap (HyperLogLog for cardinality, t-digest or KLL for quantiles). Sketches are mergeable by construction, which restores the associative-merge property at the cost of a bounded error — this is precisely why they exist and why every large analytics system uses them.

Top-N is the interesting middle case: it decomposes, but only if each partition returns N candidates, which is exact for ORDER BY on a stored column and *not* exact when the ranking depends on a global quantity, such as a score normalised across the whole dataset.

You partition users by user_id and now need to look one up by email. The index has to live somewhere, and there are exactly two choices.

Local (per-partition) index. Each partition indexes its own rows. Writes are cheap and atomic — the index entry is written in the same partition, in the same transaction, as the row. Reads by the indexed field must ask *every* partition, because the email could be anywhere. This is the fan-out tax on every lookup.

Global (term-partitioned) index. The index is itself partitioned, by the indexed value. A lookup by email hits exactly one partition. But now writing a user touches two partitions — the row’s and the index term’s — so the write is a distributed operation with all that implies: it can partially fail, leaving an index entry with no row or a row with no index entry.

Almost every system resolves the write side by making the global index asynchronous: the write commits locally and the index is updated shortly after. That is a real weakening — there is a window in which a lookup by email does not find a user who exists, and after a delete, one that does not. It is usually the right trade, and it must be a decision rather than a surprise.

The framing worth keeping: a global secondary index is a materialized view partitioned differently from its source, and every property of Materialized Views: A Read Model That Lags — asynchronous update, bounded staleness, the need for reconciliation — applies to it.

Everything above trades latency and money. Two things trade correctness, and they need naming separately.

Atomicity across partitions requires an agreement protocol. Two-Phase Commit: Buying Atomicity With a Promise gives it to you and gives you its failure modes in the same box: a coordinator crash between prepare and commit leaves participants holding locks with no authority to release them (The Blocking Window: When 2PC Stops and Waits), which is an availability outage on those rows for as long as the coordinator is gone. Sagas: Trading Isolation for Availability avoid the blocking by giving up isolation: the steps commit independently and a failure is handled by compensation, which is a new action rather than an undo (A Refund Is Not a Rollback). The third option is the one to reach for first — choose the partition key so the transaction fits inside one partition. Entity groups, aggregate roots, and "partition by the thing transactions are scoped to" are all the same idea, and it is the cheapest correctness technique available in this domain.

Uniqueness across partitions is subtler and catches people out. If users are partitioned by user_id and emails must be unique, the uniqueness check cannot be done in the user’s partition — the conflicting email lives elsewhere. A unique index is therefore *a differently-partitioned structure*, and creating a user atomically with its email claim is inherently a two-partition operation. The standard resolution is a reservation protocol: claim the email in the email-partitioned store first with a conditional write, then create the user, then confirm — with a sweeper for abandoned claims. See Distributed Uniqueness: One Name, Many Shards for the full treatment; the point here is that it is a *partitioning* consequence, not a database feature you lost.

And when the operation spans services rather than partitions of one store, it is the same problem with worse tooling — Atomicity Stops at the Process Boundary.

Every technique above is a way of paying for a boundary you drew. The cheapest technique is to draw the boundary somewhere else.

Enumerate your operations. For each, ask: does it touch one partition or many? Then choose the partition key so that the *frequent* and the *correctness-critical* operations are single-partition, and only the rare, tolerant ones fan out. A system where 99% of operations are single-partition scales almost linearly; a system where 30% are cross-partition does not scale at all, because the cross-partition work grows with both traffic and partition count.

When two access patterns genuinely conflict — orders by customer *and* orders by product — the answer is usually not a cleverer key. It is two differently-partitioned copies of the data, kept in sync asynchronously, each serving one pattern. That is Materialized Views: A Read Model That Lags-shaped thinking, it costs storage and staleness, and it is nearly always cheaper than making every query fan out.