1"""Chapter 15 practice (Sharding):
2- range, hash and directory routers (Level 3)
3- logical shards: reshard by moving whole buckets instead of rehashing every key
4- scatter-gather for a query that doesn't include the shard key
5- the 'shard by date' hot-shard bug (Level 4)
6Standard library only. Run: python3 ch15_sharding.py"""
7import bisect
8import hashlib
9from collections import Counter
10
11
12def h(key):
13 return int(hashlib.md5(str(key).encode()).hexdigest(), 16)
14
15
16class RangeRouter:
17 """Shard i holds keys from bounds[i-1] up to (not including) bounds[i]."""
18
19 def __init__(self, bounds, shards):
20 self.bounds, self.shards = bounds, shards
21
22 def shard_for(self, key):
23 return self.shards[bisect.bisect_right(self.bounds, key)]
24
25
26class HashRouter:
27 def __init__(self, shards):
28 self.shards = shards
29
30 def shard_for(self, key):
31 return self.shards[h(key) % len(self.shards)]
32
33
34class LogicalShardRouter:
35 """Many small logical shards (fixed forever) mapped to a few physical servers.
36 Growing = moving some logical shards to a new server; keys never re-hash."""
37
38 def __init__(self, n_logical, servers):
39 self.n_logical = n_logical
40 self.owner = {ls: servers[ls % len(servers)] for ls in range(n_logical)}
41
42 def logical(self, key):
43 return h(key) % self.n_logical
44
45 def shard_for(self, key):
46 return self.owner[self.logical(key)]
47
48 def add_server(self, new_server):
49 servers = sorted(set(self.owner.values())) + [new_server]
50 target = self.n_logical // len(servers)
51 moved = 0
52 per_server = Counter(self.owner.values())
53 for ls in range(self.n_logical):
54 old = self.owner[ls]
55 if per_server[old] > target and per_server[new_server] < target:
56 self.owner[ls] = new_server
57 per_server[old] -= 1
58 per_server[new_server] += 1
59 moved += 1
60 return moved
61
62
63class DirectoryRouter:
64 """A lookup table: tenant -> shard. Flexible, but the table must be stored and fast."""
65
66 def __init__(self):
67 self.table = {}
68
69 def assign(self, tenant, shard):
70 self.table[tenant] = shard
71
72 def shard_for(self, tenant):
73 return self.table[tenant]
74
75
76def scatter_gather(shards_data, predicate):
77 """Ask every shard, then combine. Needed when the query lacks the shard key."""
78 return sum(1 for rows in shards_data.values() for r in rows if predicate(r))
79
80
81def run_checks():
82 shards = ["s0", "s1", "s2", "s3"]
83
84 # Range: user IDs 0-249,999 -> s0, 250,000-499,999 -> s1 ...
85 rr = RangeRouter([250_000, 500_000, 750_000], shards)
86 assert rr.shard_for(10) == "s0" and rr.shard_for(600_000) == "s2" and rr.shard_for(999_999) == "s3"
87
88 # Hash: 100,000 users spread roughly evenly.
89 hr = HashRouter(shards)
90 load = Counter(hr.shard_for(u) for u in range(100_000))
91 assert all(23_000 < c < 27_000 for c in load.values()), load
92
93 # Resharding 4 -> 5 servers.
94 keys = range(100_000)
95 before = {k: HashRouter(shards).shard_for(k) for k in keys}
96 after = {k: HashRouter(shards + ["s4"]).shard_for(k) for k in keys}
97 moved_mod = sum(before[k] != after[k] for k in keys) / len(keys)
98 assert moved_mod > 0.75 # about 80% of keys move
99
100 ls = LogicalShardRouter(1024, shards)
101 before_ls = {k: ls.shard_for(k) for k in keys}
102 moved_logical = ls.add_server("s4")
103 after_ls = {k: ls.shard_for(k) for k in keys}
104 moved_keys = sum(before_ls[k] != after_ls[k] for k in keys) / len(keys)
105 assert 0.17 < moved_keys < 0.23 and moved_logical == 204 # only ~1/5 moves, as whole buckets
106
107 # Directory: a big tenant gets its own shard.
108 dr = DirectoryRouter()
109 dr.assign("small-shop", "s0")
110 dr.assign("huge-publisher", "s3")
111 assert dr.shard_for("huge-publisher") == "s3"
112
113 # Scatter-gather: "how many orders over 2,000?" needs every shard.
114 data = {s: [] for s in shards}
115 for oid in range(1000):
116 data[hr.shard_for(oid)].append({"order": oid, "total": oid * 3})
117 assert scatter_gather(data, lambda r: r["total"] > 2000) == 333
118
119 # Level 4: sharding orders by day sends ALL of today's writes to one shard.
120 days = list(range(400)) # day numbers
121 by_day = RangeRouter([100, 200, 300], shards)
122 todays = Counter(by_day.shard_for(399) for _ in range(10_000))
123 assert len(todays) == 1 # one hot shard
124 by_user = Counter(HashRouter(shards).shard_for(f"user-{i}") for i in range(10_000))
125 assert len(by_user) == 4 and max(by_user.values()) < 2_800 # the fix spreads them
126
127 print(f"All checks pass. Grow 4->5: plain hash moves {moved_mod:.0%} of keys; "
128 f"logical shards move {moved_keys:.0%} ({moved_logical} of 1,024 buckets).")
129
130
131if __name__ == "__main__":
132 run_checks()