# DDIA Ch. 7: a local secondary index reads from 8 shards; a global one reads from 1 — but pays it back on writes ## 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 Deterministic, so the numbers reproduce exactly: - 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 sharded | index co-located with the records: each shard indexes only *its own* cars | index sharded *by the value*: term "red" lives on `md5("red") % 8` | | READ `color=red` | scatter/gather **all 8 shards** (matches are scattered) | **1 shard** (the term owner) | | READ `color=red AND make=Honda` | scatter/gather **all 8 shards** | **2 shards** (the two term owners), then intersect | | WRITE one car | **1 shard** (its primary shard; local index updates in place) | **up to 3 shards** (primary + color-term + make-term) | | Good when | writes are hot, reads rare or always carry the partition key | reads 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. ## Steps ### 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.) **Observe.** ``` 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? **Observe.** ``` 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? **Observe.** ``` 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? **Observe.** ``` 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 - READ `color=red`: **LOCAL 8 shards** vs **GLOBAL 1 shard**; both return **980** red cars. - WRITE one car: **LOCAL 1 shard** ({0}) vs **GLOBAL 3 shards** ({0, 2, 4}). - AND `color=red AND make=Honda`: **LOCAL 8 shards** vs **GLOBAL 2 shards** ({2, 4}); both return **119**. - With per-shard slow probability 1%, the 8-way local scatter read is slow **~7.7%** of the time vs **1%** for the global single-term read. ## Why Start from the two partitionings. A record's *primary* shard is a function of its **id** (`id % 8`). A secondary *value* — `color=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 1. **Add the partition key.** Extend the script with a `color=red AND id=8324` query and confirm the local index drops from 8 shards to 1 — the boundary condition where local's fan-out vanishes. 2. **Make the terms collide.** Pick a make whose `md5(make) % 8` equals red's shard (4) — e.g. try each make — so the global write touches only 2 shards and the AND-query only 1. Watch the write/read costs shift with the hashing. 3. **Grow the cluster.** Bump `N_SHARDS` to 32 and re-measure: the global read stays at 1 shard while the local scatter grows to 32 and its tail-slow probability climbs to `1 − 0.99^32 ≈ 27%`. The gap widens with cluster size — which is exactly when global indexes start to pay off.