Sharding

The previous lesson showed what adding a member to the cluster solves and what it does not: when a member is lost, service continues, but every member keeps carrying the entirety of the same data. Once the catalog grows and loan records no longer fit on a single member’s disk, replication has nothing left to offer; the data has to be split across members.

The name for this split is sharding, and it comes down to a single decision: choosing the field that determines which shard each document goes to. That field is called the shard key. The arithmetic of data-splitting strategies — whether the split is range-based or hash-based, how much data a rebalance moves — was measured in the Scaling the Data Layer course and is not repeated here. The question here is specific to the document model: the key is a document field, and choosing that field is an expensive decision to reverse.

The Key Comes From Inside the Document

In a relational table, the partitioning column is part of the schema and is present in every row. The document model offers no such guarantee: documents in the same collection can carry different sets of fields. This places two conditions on the shard key.

The key field must be present in every document. The router cannot send a document that lacks the field to any shard at all. In the library database this is not an abstract risk: some of the loan records migrated from the legacy system have no branch field at all.

The key’s uniqueness bounds the shard count. The smallest unit of sharding is a single key value; documents that share a value cannot be split apart. A field with six distinct values cannot be spread across more than six shards, and if one of those values carries half the documents, that pile never gets split. This is an upper bound that is independent of the run.

The third condition is the one this lesson measures: the key has to spread writes across shards while still letting frequent queries route to a single shard. If a query pins the key’s prefix, the router sends it to one shard; this is called a targeted query. If it does not, the router has to ask every shard; this is called a scatter-gather query. The difference between the two kinds is the number of shards touched, and it can be counted.

The Mechanism

NS16 — the distribution of loan records across branches is 45, 18, 14, 11, 8, and 4 units. Reason: the central branch carries the largest collection and the most members. NS17 — 2 percent of the migrated legacy records have no branch field. NS18 — query mix: the member query is 45 percent, the branch end-of-day listing 20 percent, the last seven days’ loans 15 percent, access by record id 15 percent, the overdue-loan scan 5 percent. All three assumptions have a linear effect; if the weights in the mix change, the weighted average below shifts in the same direction.

// shard/model.mjs — loan document generator and shard placement. The generator is
// handwritten and the seed is visible: every run produces the same document set. The
// shard count P is a parameter; range- vs. hash-based placement was measured in the
// Scaling the Data Layer course. Fixed here: every document routes by its KEY FIELD VALUE.
export const BRANCH = [["central", 45], ["kadikoy", 18], ["besiktas", 14], ["uskudar", 11],
  ["sisli", 8], ["adalar", 4]];
export const DAY = 180, MEMBER = 6000;

export function generate({ count = 20000, seed = 20260731, missingBranch = 0.02 } = {}) {
  let s = seed % 2147483647;
  const rand = () => (s = (s * 48271) % 2147483647) / 2147483647;
  const total = BRANCH.reduce((t, [, w]) => t + w, 0);
  const docs = [];
  for (let i = 1; i <= count; i += 1) {
    let x = rand() * total, branch = BRANCH[0][0];
    for (const [name, w] of BRANCH) { if (x < w) { branch = name; break; } x -= w; }
    const d = { loan_id: i, member_id: 1 + Math.floor(rand() * MEMBER),
      borrow_day: 1 + Math.floor(rand() * DAY), returned: rand() < 0.82 };
    if (rand() >= missingBranch) d.branch = branch;   // some legacy records have no branch field
    docs.push(d);
  }
  return docs;
}

const compare = (a, b) => {                   // total ordering over key tuples
  for (let i = 0; i < a.length; i += 1) if (a[i] !== b[i]) return a[i] < b[i] ? -1 : 1;
  return 0;
};
const hashValue = (m) => { let h = 2166136261; for (const c of m) h = Math.imul(h ^ c.charCodeAt(0), 16777619); return h >>> 0; };

// Shard placement: key values are sorted and split into P windows with an equal-doc-count target.
// A single key value cannot be split — the smallest unit of sharding is one key value.
export function place(docs, fields, P, hashed = false) {
  const key = (d) => fields.map((f) => d[f]);
  const complete = docs.filter((d) => fields.every((f) => d[f] !== undefined));
  const counts = new Map();
  for (const d of complete) {
    const k = JSON.stringify(hashed ? [hashValue(JSON.stringify(key(d)))] : key(d));
    counts.set(k, (counts.get(k) ?? 0) + 1);
  }
  const sorted = [...counts.entries()].sort((a, b) => compare(JSON.parse(a[0]), JSON.parse(b[0])));
  const target = complete.length / P;
  const placement = new Map();
  let p = 0, cumulative = 0;
  for (const [k, n] of sorted) {
    placement.set(k, p);
    cumulative += n;
    if (cumulative >= target * (p + 1) && p < P - 1) p += 1;
  }
  const shardOf = (d) => placement.get(JSON.stringify(hashed ? [hashValue(JSON.stringify(key(d)))] : key(d)));
  const distribution = Array(P).fill(0);
  for (const d of complete) distribution[shardOf(d)] += 1;
  return { shardOf, distribution, complete, missing: docs.length - complete.length, distinct: counts.size };
}

// A query can be routed if it pins a PREFIX of the key (targeted query); if it does not, the
// router has to go to every shard (scatter-gather query). Under hash placement a range clause
// cannot be used, because adjacent values do not land on adjacent shards.
export function touched(query, fields, hashed, y, P) {
  const usable = [];
  for (const f of fields) {
    if (query.eq && f in query.eq) { usable.push({ field: f, type: "eq" }); continue; }
    if (query.range && query.range.field === f && hashed === false) usable.push({ field: f, type: "range" });
    break;
  }
  if (usable.length === 0) return P;
  const matches = (d) => usable.every((u) => u.type === "eq"
    ? d[u.field] === query.eq[u.field]
    : d[u.field] >= query.range.min && d[u.field] <= query.range.max);
  return new Set(y.complete.filter(matches).map(y.shardOf)).size;
}

// shard/measure.mjs — five shard-key candidates: placement balance, hot shard, touched shards
import { generate, place, touched, BRANCH, DAY, MEMBER } from "./model.mjs";

const P = 4, COUNT = 20000;
const docs = generate({ count: COUNT });
const CANDIDATE = [["branch", ["branch"], false], ["borrow_day", ["borrow_day"], false],
  ["hash(borrow_day)", ["borrow_day"], true], ["member_id", ["member_id"], false],
  ["{branch,member_id}", ["branch", "member_id"], false]];
const QUERY = [
  ["S1 member's open loans", 45, (i) => ({ eq: { member_id: 1 + (i * 137) % MEMBER } })],
  ["S2 branch end-of-day listing", 20, (i) => ({ eq: { branch: BRANCH[i % BRANCH.length][0], borrow_day: 1 + (i * 31) % DAY } })],
  ["S3 last 7 days' loans", 15, () => ({ range: { field: "borrow_day", min: DAY - 6, max: DAY } })],
  ["S4 loan record by id", 15, (i) => ({ eq: { loan_id: 1 + (i * 911) % COUNT } })],
  ["S5 overdue-loan scan", 5, () => ({ eq: {} })],
];
const s = (x, n) => String(x).padStart(n);
const recent = docs.filter((d) => d.borrow_day > DAY - 7);
console.log(`${COUNT} loan documents, ${P} shards, ${BRANCH.length} branches, ${MEMBER} members, ${DAY} days. Seed 20260731.`);
console.log(`taken in the last 7 days: ${recent.length} documents\n`);
console.log("shard key            | distinct vals | unroutable | shard distribution      | busiest/avg | last-7-day busiest shard");
console.log("--------------------|---------------|------------|-------------------------|-------------|-------------------------");
const placement = new Map();
for (const [name, fields, hashed] of CANDIDATE) {
  const y = place(docs, fields, P, hashed);
  placement.set(name, y);
  const fullest = Math.max(...y.distribution) / (y.complete.length / P);
  const rd = Array(P).fill(0);
  for (const d of recent) if (fields.every((f) => d[f] !== undefined)) rd[y.shardOf(d)] += 1;
  const share = Math.max(...rd) / rd.reduce((t, x) => t + x, 0);
  console.log(`${name.padEnd(19)} | ${s(y.distinct, 12)} | ${s(y.missing, 17)} | ` +
    `${y.distribution.map((x) => s(x, 5)).join(" ").padEnd(23)} | ${s(fullest.toFixed(2), 11)} | ` +
    `${s("%" + (100 * share).toFixed(1), 23)}`);
}

console.log("\nshards touched (per query, 1 = targeted, 4 = scatter-gather):");
console.log("shard key            |  S1  |  S2  |  S3  |  S4  |  S5  | weighted average");
console.log("--------------------|------|------|------|------|------|-----------------");
for (const [name, fields, hashed] of CANDIDATE) {
  const y = placement.get(name);
  const cell = [], weighted = [];
  for (const [, weight, example] of QUERY) {
    let t = 0;
    for (let i = 0; i < 12; i += 1) t += touched(example(i), fields, hashed, y, P);
    cell.push(t / 12);
    weighted.push((t / 12) * weight);
  }
  const avg = weighted.reduce((a, b) => a + b, 0) / QUERY.reduce((a, b) => a + b[1], 0);
  console.log(`${name.padEnd(19)} | ${cell.map((x) => s(x.toFixed(2), 4)).join(" | ")} | ${s(avg.toFixed(2), 16)}`);
}
console.log("\nquery mix: " + QUERY.map(([a, w]) => `${a.slice(0, 2)} %${w}`).join(", "));
console.log(`independent of the run: the number of distinct values of a key is the upper bound` +
  ` on shard count (${BRANCH.length} for branch).`);

20000 loan documents, 4 shards, 6 branches, 6000 members, 180 days. Seed 20260731.
taken in the last 7 days: 782 documents

shard key            | distinct vals | unroutable | shard distribution      | busiest/avg | last-7-day busiest shard
--------------------|---------------|------------|-------------------------|-------------|-------------------------
branch              |            6 |               406 | 12328  3489  1597  2180 |        2.52 |                   %63.8
borrow_day          |          180 |                 0 |  5022  5031  5031  4916 |        1.01 |                  %100.0
hash(borrow_day)    |          180 |                 0 |  5034  4973  5002  4991 |        1.01 |                   %59.6
member_id           |         5788 |                 0 |  5000  5000  5002  4998 |        1.00 |                   %27.0
{branch,member_id}  |        13443 |               406 |  4899  4898  4899  4898 |        1.00 |                   %25.8

shards touched (per query, 1 = targeted, 4 = scatter-gather):
shard key            |  S1  |  S2  |  S3  |  S4  |  S5  | weighted average
--------------------|------|------|------|------|------|-----------------
branch              | 4.00 | 1.00 | 4.00 | 4.00 | 4.00 |             3.40
borrow_day          | 4.00 | 1.00 | 1.00 | 4.00 | 4.00 |             2.95
hash(borrow_day)    | 4.00 | 1.00 | 4.00 | 4.00 | 4.00 |             3.40
member_id           | 1.00 | 4.00 | 4.00 | 4.00 | 4.00 |             2.65
{branch,member_id}  | 4.00 | 1.50 | 4.00 | 4.00 | 4.00 |             3.50

query mix: S1 %45, S2 %20, S3 %15, S4 %15, S5 %5
independent of the run: the number of distinct values of a key is the upper bound on shard count (6 for branch).

Placement Balance and Hot Shards

The most intuitive candidate gives the worst balance. Splitting loan records by branch looks like it matches the operational structure; the measurement says something else. When six branches are squeezed into four shards, the shards end up with 12,328, 3,489, 1,597, and 2,180 documents: the busiest shard holds 2.52 times the average. The reason is uniqueness — the smallest grain a shard can carry is one branch, and the central branch alone produces 45 percent of all loan records. Adding more shards does not fix the imbalance either, because six distinct values are the upper bound for six shards. The row’s second number is a warning of its own: 406 documents carry no branch field and cannot be routed to any shard.

A balanced placement is not enough on its own. The borrow_day key splits documents almost perfectly — the busiest shard holds 1.01 times the average. The last column refutes this row: 100 percent of the 782 loans taken in the last seven days fall onto a single shard. When a monotonically increasing field is the key, past data spreads evenly, but every new write piles onto the shard carrying the most recent window. This is called a hot shard, and its only visible symptom in the store is that one member queues writes while the cluster’s other three sit idle.

Hash placement trades the hot shard for something else. Hashing the same field brings the hot shard’s share down from 100 percent to 59.6 percent. The reason it does not drop to 25 percent is, again, uniqueness: when seven days’ seven distinct values spread across four shards, some shards end up with two days. The cost sits in the second table, and it is large.

The candidate with the highest uniqueness gives the best balance. Under the member_id key, the distribution is 5,000, 5,000, 5,002, 4,998, and the last-seven-days share is 27 percent — close to the 25 percent that is ideal for four shards. The compound key gives the same balance but brings back the 406 unroutable documents, because branch sits in its prefix.

Targeted and Scatter-Gather Queries

The second table runs the same query mix across all five candidates, and it reverses the decision that balance alone would suggest.

The member_id key gives the best result on the weighted average, 2.65 shards, because the mix’s heaviest query — the member query — pins the key exactly: 1.00 shard. Under the same key, the branch end-of-day listing rises to 4.00 — because it never pins the key at all, it goes to all four shards at once.

The compound key {branch,member_id} is this lesson’s most instructive row. Its placement balance is perfect (1.00) and the branch query drops to 1.50 shards; against that, the member query rises to 4.00. The reason is that a compound key works left to right: the member query pins the member_id field but does not pin the branch field in the prefix, so the router cannot tell which shard to go to. Its weighted average, 3.50, is the worst of the five candidates. Adding a compound key does not narrow access — it tightens the prefix condition.

Comparing borrow_day against its hash placement shows the trade-off in a single column. Hash placement brought the hot shard down from 100 percent to 59.6 percent; in exchange, the last-seven-days query rose from 1.00 to 4.00 shards, and the weighted average climbed from 2.95 to 3.40. Because hash placement does not put adjacent values on adjacent shards, the range query cannot be routed. The thing that spreads writes and the thing that collects reads come into direct conflict here.

The last line restates the bound that is independent of the run: the number of distinct values a key has is the upper bound on shard count. This bound does not depend on the machine, the data, or the configuration.

Summary

• The shard key is a document field, and a document that lacks the field cannot be routed: under the branch-based key, 406 of the 20,000 documents could not go to any shard.
• The number of distinct values a key has is the upper bound on shard count; in the collection split across six branches, the busiest shard held 2.52 times the average, and adding more shards does not fix this.
• A balanced placement does not prevent a hot shard: under the borrow_day key, placement was balanced to within 1.01 times even as 100 percent of the 782 loans from the last seven days fell onto a single shard.
• Hash placement brought the hot shard’s share down to 59.6 percent but raised the range query from 1.00 shard to 4.00 shards; the weighted average climbed from 2.95 to 3.40.
• A compound key depends on the prefix condition: the {branch,member_id} placement balanced perfectly, but the member query, which does not pin its prefix, scattered across 4.00 shards and came out worst on the weighted average, 3.50.
• In this mix, the best key was member_id: 1.00 times balance, a 27 percent hot share, and 2.65 weighted shards. The choice is made not by looking at balance but at the mix’s heaviest query.

Next Step

This lesson resolved which shard a document goes to, but the shard itself is the previous lesson’s replica set: every shard is made up of more than one member, and which of those members has to have seen a write is still open. Both lessons left the same gap — in the first lesson, 6 accepted loan records were rolled back, and how to prevent that was never said. How many members have to acknowledge a write for it to count as “successful” is a setting; in the same way, which member a read is answered from and with what verification is also a setting. The next lesson names these two settings and runs the same loan scenario under three of them: the number of turns an acknowledgement takes to arrive, the number of writes lost at the moment of failure, and how the version seen on a read changes.