#!/usr/bin/env python3 """ DDIA Ch. 7 (Sharding) - Local vs Global secondary indexes. A used-cars database is sharded by listing id (id % N_SHARDS). We build two kinds of secondary index on `color` and `make`, and measure WHERE the cost lands: LOCAL (document-partitioned): each shard indexes only its own cars. READ color=red -> must query ALL N shards (scatter/gather). WRITE one car -> touch 1 shard (its primary shard). GLOBAL (term-partitioned): the index is sharded BY the secondary value, so the term "red" lives on exactly one shard. READ color=red -> query 1 shard (the term owner). WRITE one car -> touch up to 3 shards (primary + color-term + make-term). Deterministic: stable md5 hashing + fixed RNG seed, so numbers reproduce exactly. Python 3 stdlib only. """ import hashlib import random from collections import defaultdict N_SHARDS = 8 N_CARS = 10_000 COLORS = ["red", "blue", "black", "white", "silver", "gray", "green", "yellow", "orange", "brown"] MAKES = ["Toyota", "Honda", "Ford", "BMW", "Audi", "Tesla", "Nissan", "Kia"] def md5_int(s: str) -> int: """Stable string hash (never use Python's built-in hash() on strings: it is salted per-process and would not reproduce).""" return int(hashlib.md5(s.encode()).hexdigest(), 16) def primary_shard(car_id: int) -> int: """Records are sharded by their listing id.""" return car_id % N_SHARDS def term_shard(term: str) -> int: """A global (term-partitioned) index shards by the secondary VALUE.""" return md5_int(term) % N_SHARDS def build_cars(): rng = random.Random(42) cars = {} for cid in range(N_CARS): cars[cid] = { "color": rng.choice(COLORS), "make": rng.choice(MAKES), } return cars def build_local_index(cars): """document-partitioned: shard -> field -> value -> [ids], over that shard's cars only.""" idx = {s: {"color": defaultdict(list), "make": defaultdict(list)} for s in range(N_SHARDS)} for cid, car in cars.items(): s = primary_shard(cid) idx[s]["color"][car["color"]].append(cid) idx[s]["make"][car["make"]].append(cid) return idx def build_global_index(cars): """term-partitioned: field -> value -> [ids], with each value living on term_shard(value).""" color_idx = defaultdict(list) make_idx = defaultdict(list) for cid, car in cars.items(): color_idx[car["color"]].append(cid) make_idx[car["make"]].append(cid) return color_idx, make_idx def rule(): print("=" * 68) def main(): cars = build_cars() local = build_local_index(cars) color_global, make_global = build_global_index(cars) rule() print("SETUP") rule() print(f" {N_CARS} used-car listings, ids 0..{N_CARS - 1}") print(f" sharded by listing id across {N_SHARDS} shards (id % {N_SHARDS})") print(f" secondary indexes on: color ({len(COLORS)} values), make ({len(MAKES)} values)") print() # ---- Step 1: READ color=red ----------------------------------------- rule() print("STEP 1 -- single-value READ: color = red") rule() # LOCAL: which shards actually hold a red car? local_shards_with_red = [s for s in range(N_SHARDS) if local[s]["color"].get("red")] local_red_count = sum(len(local[s]["color"].get("red", [])) for s in range(N_SHARDS)) print(" LOCAL (document-partitioned) index:") print(f" red cars physically live on shards: {local_shards_with_red}") print(f" the query key is 'red', NOT a listing id -> cannot know which") print(f" shards hold matches -> must scatter to ALL {N_SHARDS} shards, gather.") print(f" shards queried = {N_SHARDS}") print(f" red cars found = {local_red_count}") print() # GLOBAL: term "red" lives on one shard. red_shard = term_shard("red") global_red_count = len(color_global["red"]) print(" GLOBAL (term-partitioned) index:") print(f" term 'red' is owned by shard md5('red') % {N_SHARDS} = {red_shard}") print(f" query goes to that ONE shard, reads its postings list.") print(f" shards queried = 1 (shard {red_shard})") print(f" red cars found = {global_red_count}") print() assert local_red_count == global_red_count, "both indexes must return the same set" # ---- Step 2: WRITE one new car -------------------------------------- rule() print("STEP 2 -- WRITE one new listing") rule() new_id = N_CARS # 10000 new_color = "red" new_make = "Honda" p = primary_shard(new_id) print(f" new car id={new_id} color={new_color} make={new_make}") print(f" its primary shard (id % {N_SHARDS}) = {p}") print() print(" LOCAL index:") print(f" the record and its index entries are co-located, so the write") print(f" lands entirely on the primary shard.") print(f" shards updated = {{{p}}} (count = 1)") print() color_term_shard = term_shard(new_color) make_term_shard = term_shard(new_make) global_write_shards = {p, color_term_shard, make_term_shard} print(" GLOBAL index:") print(f" record body -> primary shard {p}") print(f" color term '{new_color}' -> color-term shard {color_term_shard}") print(f" make term '{new_make}' -> make-term shard {make_term_shard}") print(f" shards updated = {sorted(global_write_shards)} (count = {len(global_write_shards)})") print() # ---- Step 3: AND query ---------------------------------------------- rule() print("STEP 3 -- AND query: color = red AND make = Honda") rule() print(" LOCAL index:") print(f" still no partition key in the filter -> scatter to ALL {N_SHARDS} shards,") print(f" each intersects its own two postings lists, coordinator unions.") print(f" shards queried = {N_SHARDS}") local_and = sum( len(set(local[s]["color"].get("red", [])) & set(local[s]["make"].get("Honda", []))) for s in range(N_SHARDS) ) print(f" matches found = {local_and}") print() make_shard = term_shard("Honda") print(" GLOBAL index:") print(f" color term 'red' -> shard {red_shard}") print(f" make term 'Honda' -> shard {make_shard}") global_and_shards = {red_shard, make_shard} print(f" fetch the two postings lists, intersect them.") print(f" shards queried = {len(global_and_shards)} ({sorted(global_and_shards)})") global_and = len(set(color_global["red"]) & set(make_global["Honda"])) print(f" matches found = {global_and}") print() assert local_and == global_and # ---- Step 4: tail-latency amplification ----------------------------- rule() print("STEP 4 -- tail-latency amplification of the local scatter read") rule() p_slow = 0.01 p_read_slow = 1 - (1 - p_slow) ** N_SHARDS print(f" suppose any single shard is slow {p_slow:.0%} of the time (independently).") print(f" a GLOBAL single-term read waits on 1 shard -> slow {p_slow:.2%} of reads") print(f" a LOCAL scatter read waits on ALL {N_SHARDS} -> slow {p_read_slow:.4f} of reads") print(f" = 1 - (1 - {p_slow})^{N_SHARDS} = {p_read_slow:.4f} (~{p_read_slow:.1%})") print(f" the slowest of {N_SHARDS} shards sets the latency (DDIA Ch. 2).") print() rule() print("SUMMARY") rule() print(f" READ color=red : LOCAL {N_SHARDS} shards vs GLOBAL 1 shard") print(f" WRITE one car : LOCAL 1 shard vs GLOBAL {len(global_write_shards)} shards") print(f" AND red+Honda : LOCAL {N_SHARDS} shards vs GLOBAL {len(global_and_shards)} shards") print(f" local scatter read is slow ~{p_read_slow:.1%} of the time vs {p_slow:.0%} for global") if __name__ == "__main__": main()