studyDDIAChapter 7 · Sharding

A Local Secondary Index Reads From 8 Shards; a Global One From 1 — but Pays It Back on Writes

A used-cars database sharded by listing id, with secondary indexes on color and make. The read color=red touches 8 shards on a local (document-partitioned) index but 1 shard on a global (term-partitioned) index — then a single-car write touches 1 shard locally and 3 globally. You don't remove the cost; you choose where it lands.


Concept

A used-cars database is sharded by listing id (id % 8), with secondary indexes on color and make. You want color = red. Build the secondary index as a local (document-partitioned) index — each shard indexes only its own cars — and the read must scatter to all 8 shards and gather, because a red car can sit on any shard. Build it as a global (term-partitioned) index — the index is sharded by the value, so the term "red" lives on exactly one shard — and the same read touches 1 shard. It looks like the global index simply wins. Then you write one car: the local index touches 1 shard (the car's primary shard), the global index touches 3 (the primary shard, plus the color-term shard, plus the make-term shard). The cost didn't disappear — you moved it. Read-cheap/write-expensive vs write-cheap/read-expensive; you pick which.

Provenance

Designing Data-Intensive Applications, 2nd ed. (Kleppmann & Riccomini, 2025), Chapter 7 — Sharding, §"Sharding and Secondary Indexes", §"Local Secondary Indexes (Partitioning Secondary Indexes by Document)", §"Global Secondary Indexes (Partitioning Secondary Indexes by Term)":

The advantage of a global (term-partitioned) index over a local (document-partitioned) index is that reads are more efficient: rather than doing scatter/gather over all shards, a client only needs to make a request to the shard containing the term that it wants. However, the downside of a global index is that writes are slower and more complicated, because a write to a single document may now affect multiple shards of the index.

This exercise makes both halves of that sentence into numbers you measure: the read fan-out (8 vs 1) and the write fan-out (1 vs 3), on the book's own used-cars example.

Prerequisites

The result is a cost-conservation argument about where a query touches the cluster. These help; each line notes where it shows up.

  1. Secondary indexes and postings lists — a secondary index maps a value (color=red) to the list of record ids that have it (its postings list). The primary key is the listing id; color and make are secondary. It shows up as the color -> [ids] and make -> [ids] dicts both index types build.
  2. Scatter/gather over shards — when a query's filter isn't the partition key, the coordinator must fan the query out to every shard and merge the replies. It shows up as the local read (and local AND-query) touching all 8 shards.
  3. Tail-latency amplification — when one request waits on many backends in parallel, the slowest backend sets your latency, so the chance of a slow request grows with fan-out. It shows up in Step 4: an 8-way scatter is slow 1 − 0.99^8 ≈ 7.7% of the time even though each shard is slow only 1%.
Where to learn the prerequisites Secondary indexes, local vs global (#1): DDIA §"Sharding and Secondary Indexes"; the same chapter's §"Local Secondary Indexes" and §"Global Secondary Indexes" spell out the two partitionings. Scatter/gather and tail latency (#2, #3): DDIA Ch. 2 §"Describing Performance" — the tail-latency-amplification discussion (waiting on the slowest of N parallel calls) is the engine behind why the local read's fan-out hurts. If only one is new, make it #3 — tail-latency amplification is why "read from 8 shards" is worse than 8× the work; it's the reason a global index exists at all.
Environment these numbers came from Deterministic, so the numbers reproduce exactly. Captured on macOS 26.5.2 (Darwin 25.5.0), arm64, Python 3.15.0a8, standard library only — no Docker, no deps. Stable hashing: term→shard uses int(md5(term).hexdigest(), 16) % 8, never Python's per-process-salted hash(). Fixed workload: 10,000 cars, ids 0..9999, colors/makes assigned with random.Random(42); 8 shards.

Mental model

Same 10,000 cars, same two secondary indexes, same query color=red — only the index partitioning differs.

LOCAL (document-partitioned)GLOBAL (term-partitioned)
How it's shardedindex co-located with the records: each shard indexes only its own carsindex sharded by the value: term "red" lives on md5("red") % 8
READ color=redscatter/gather all 8 shards (matches are scattered)1 shard (the term owner)
READ color=red AND make=Hondascatter/gather all 8 shards2 shards (the two term owners), then intersect
WRITE one car1 shard (its primary shard; local index updates in place)up to 3 shards (primary + color-term + make-term)
Good whenwrites are hot, reads rare or always carry the partition keyreads are hot, especially single-value equality reads

The record's primary shard is chosen by its id; a secondary value can appear on any shard. Those are two different partitionings of the same data, and no single index can make both the read and the write single-shard. The whole thing is one script, code/local_vs_global_index.py.

Setup

cd study/ddia/ch07
python3 code/local_vs_global_index.py

Do not read the output yet — make each prediction first.


Step 1 · single-value READ: color = red

Predict. The cars are sharded by listing id, and roughly a tenth of 10,000 cars are red. For color=red, how many shards must a local index query, and how many must a global index query? (There are 8 shards.)

How many shards does each index query?
STEP 1 -- single-value READ: color = red ==================================================================== LOCAL (document-partitioned) index: red cars physically live on shards: [0, 1, 2, 3, 4, 5, 6, 7] the query key is 'red', NOT a listing id -> cannot know which shards hold matches -> must scatter to ALL 8 shards, gather. shards queried = 8 red cars found = 980 GLOBAL (term-partitioned) index: term 'red' is owned by shard md5('red') % 8 = 4 query goes to that ONE shard, reads its postings list. shards queried = 1 (shard 4) red cars found = 980

8 vs 1 — same 980 red cars. Red cars land on every shard, so a local index has no way to know which shards have matches; it must ask all 8. The global index put every red car's id on shard 4, so one request answers the whole query.

Step 2 · WRITE one new car

Predict. Now insert one car (id=10000, color=red, make=Honda). Its primary shard is 10000 % 8 = 0. How many shards does the local index write touch? How many does the global index write touch?

How many shards does the write touch?
STEP 2 -- WRITE one new listing ==================================================================== new car id=10000 color=red make=Honda its primary shard (id % 8) = 0 LOCAL index: the record and its index entries are co-located, so the write lands entirely on the primary shard. shards updated = {0} (count = 1) GLOBAL index: record body -> primary shard 0 color term 'red' -> color-term shard 4 make term 'Honda' -> make-term shard 2 shards updated = [0, 2, 4] (count = 3)

1 vs 3 — the read cost came back as a write cost. The local index sits on the car's own shard, so one write updates the record and its index entries. The global index scatters this one car's facts across the cluster: its body on shard 0, its "red" onto shard 4, its "Honda" onto shard 2 — three shards for one insert.

Step 3 · AND query: color = red AND make = Honda

Predict. Now color=red AND make=Honda. The global index owns "red" on one shard and "Honda" on another. How many shards does each index type query, and do they agree on the count?

How many shards does the AND query touch?
STEP 3 -- AND query: color = red AND make = Honda ==================================================================== LOCAL index: still no partition key in the filter -> scatter to ALL 8 shards, each intersects its own two postings lists, coordinator unions. shards queried = 8 matches found = 119 GLOBAL index: color term 'red' -> shard 4 make term 'Honda' -> shard 2 fetch the two postings lists, intersect them. shards queried = 2 ([2, 4]) matches found = 119

8 vs 2 — still 119 matches either way. The local index has no partition key in the filter, so it's back to all 8 shards. The global index fetches just the two relevant term shards (4 and 2) and intersects the postings lists — a 2-shard read for a 2-term query.

Step 4 · tail-latency amplification of the scatter read

Predict. Suppose each shard is slow 1% of the time, independently. A global single-term read waits on 1 shard. A local scatter read waits on all 8. What fraction of local reads are slow?

What fraction of local scatter reads are slow?
STEP 4 -- tail-latency amplification of the local scatter read ==================================================================== suppose any single shard is slow 1% of the time (independently). a GLOBAL single-term read waits on 1 shard -> slow 1.00% of reads a LOCAL scatter read waits on ALL 8 -> slow 0.0773 of reads = 1 - (1 - 0.01)^8 = 0.0773 (~7.7%) the slowest of 8 shards sets the latency (DDIA Ch. 2).

~7.7%, not 1%. A scatter read is only as fast as its slowest shard, so the probability of hitting a slow shard compounds with fan-out: 1 − 0.99^8 ≈ 0.077. The local index doesn't just do more work — it amplifies the tail. This is the Ch. 2 effect, arriving through the back door of the index design.


What you should see

Why

Start from the two partitionings. A record's primary shard is a function of its id (id % 8). A secondary valuecolor=red — is a property of the record's contents, and red records were assigned ids all over the place, so they land on every shard. These are two independent ways of slicing the same 10,000 rows, and that independence is the whole story.

A local index is organized the same way the records are: it lives on the record's own shard and indexes only that shard's cars. That makes writes cheap — inserting a car updates one shard's records and one shard's index, atomically, in one place (Step 2: 1 shard). But a read for color=red is asking a question the id-partitioning can't answer: "which shards hold red cars?" is unknowable without looking, because red is scattered by id. So the coordinator must scatter/gather across all 8 (Step 1). A global index is organized by the secondary value instead: it collects every red id onto md5("red") % 8, so the single-value read goes to exactly one shard (Step 1: 1 shard). But now one record's terms live apart — its "red" on shard 4, its "Honda" on shard 2, its body on shard 0 — so writing that one car must touch all three (Step 2: 3 shards).

This is a conservation law, not an implementation quirk. You cannot make both the read and the write single-shard, because "group by id" and "group by value" are different partitionings, and an index can only be physically laid out one way. Choosing local vs global is choosing which operation pays: co-locate the index with the records (cheap writes, scattered reads) or co-locate it with the values (cheap reads, scattered writes).

And the scattered read is worse than it looks. Step 4 is the Ch. 2 tail-latency point: an 8-way scatter completes only when its slowest shard replies, so if each shard is independently slow 1% of the time, the read is slow 1 − 0.99^8 ≈ 7.7% of the time. Fan-out multiplies your exposure to the tail; a 1-shard read never does.

The boundary — a local index is single-shard too, if the query carries the partition key The fan-out isn't intrinsic to local indexes; it's the price of querying by a value alone. Add the partition key to the filter — color=red AND listing_id=8324 (or "red cars in this seller's shard", if you shard by seller) — and the coordinator knows exactly which shard to ask, so the local index also hits one shard with no scatter. That's why the right question isn't "local or global?" but "will my hot reads know their partition key?" If yes, local wins on both axes; if no (a global filter like color=red across the whole dataset), the global index buys back the read at the cost of the write. Global indexes are also usually updated asynchronously, so a just-written term may not be visible on its term shard immediately — the write fan-out is often eventually consistent, trading the latency for staleness.

Go deeper

Sources: DDIA 2e, Ch. 7, §"Sharding and Secondary Indexes", §"Local Secondary Indexes (Partitioning Secondary Indexes by Document)", §"Global Secondary Indexes (Partitioning Secondary Indexes by Term)" · Ch. 2, §"Describing Performance" (tail-latency amplification).