Sharding Patterns

The previous lesson split a table into partitions, and every partition lived inside the same database. Distributing partitions across separate machines is one step beyond the same idea, and at that step its nature changes.

Sharding (sharding) is the horizontal splitting of data across more than one independent database instance. Every shard carries its own processes, its own buffer pool, its own log, and its own lock manager. Every structure established in the Engine Architecture topic exists once per shard, and the shards are not aware of each other’s existence. This is the gain: write load, memory, and disk are genuinely divided. This is also the cost: three abilities that are free within a single engine for one table — joining, uniqueness, and transactions — stop being free across shards.

This lesson counts those three costs. Measurements are models built with node; they are not a real distributed database. What the model can count is relationships — which choice multiplies which cost by how much — not absolute values.

What the Shard Key Imposes on the Application

The shard key (shard key) is the column that decides which shard a row falls into. The library data has two candidates, and the choice determines which query stays cheap.

In a setup sharded by member ID, all of a member’s loan records sit in a single shard. The question “this member’s open loans” goes to a single shard; the query is no different from one against a single-machine setup. In the same setup, the answer to “who has borrowed this book” is scattered across every shard.

In a setup sharded by book ID, the relationship reverses. Which choice is correct does not come from the data but from the workload’s question distribution: whichever key the most frequent, lowest-latency question filters on is the shard key.

This decision does not resemble a reversible index decision on a single-machine setup. An index can be dropped and rebuilt; changing the shard key means moving all of the data. This is the first thing sharding imposes on the application: query shape determines data layout, and changing it later is expensive.

The Cross-Shard Query

A query that does not filter on the shard key is sent to every shard, a partial result comes back from each, and a coordinator merges them. This pattern is called scatter-gather (scatter-gather), and its cost runs against intuition: even if the total work performed stays the same, response time gets worse, because the query has to wait for the slowest shard.

cat > cross_shard.mjs <<'JS'
// Model: for loan records distributed across 8 shards, the number of shards touched by
// each query type and the response time. Not a real distributed database; what is
// counted is the relationship between shards touched and queueing delay.
const SHARDS = 8, QUERIES = 20000;
// Reproducible pseudo-random generator (linear congruential).
let state = 20240728;
const random = () => (state = (state * 1103515245 + 12345) % 2147483648) / 2147483648;

// A single shard's response time: 2 ms base plus weighted queueing. One percent chance of being slow.
function shardTime() {
  const r = random();
  return r < 0.99 ? 2 + 3 * r : 40 + 60 * r;
}
function measure(shardsTouched) {
  const times = [];
  for (let i = 0; i < QUERIES; i++) {
    let slowest = 0;
    for (let p = 0; p < shardsTouched; p++) slowest = Math.max(slowest, shardTime());
    times.push(slowest + 1);            // 1 ms coordination share
  }
  times.sort((a, b) => a - b);
  const pct = (q) => times[Math.floor(QUERIES * q)];
  return { median: pct(0.5), p95: pct(0.95), p99: pct(0.99), worst: times[QUERIES - 1] };
}
console.log('shards touched     median      p95      p99   worst');
for (const d of [1, 2, 4, 8]) {
  const s = measure(d);
  console.log(String(d).padStart(14),
              s.median.toFixed(1).padStart(9), s.p95.toFixed(1).padStart(8),
              s.p99.toFixed(1).padStart(8), s.worst.toFixed(1).padStart(9));
}
console.log('\n— query mix: does the shard key apply —');
const mix = [
  ['a members loans (shard key applies)',   1],
  ['a books loan history (no key)',         SHARDS],
  ['monthly report (no key)',               SHARDS],
];
for (const [label, d] of mix) {
  const s = measure(d);
  console.log(label.padEnd(38), 'shards', String(d).padStart(2),
              '| median', s.median.toFixed(1).padStart(5),
              '| p99', s.p99.toFixed(1).padStart(6), 'ms');
}
JS
node cross_shard.mjs

shards touched     median      p95      p99   worst
             1       4.5      5.9      6.0     101.0
             2       5.1      5.9    100.6     101.0
             4       5.5      6.0    100.8     101.0
             8       5.8    100.5    100.9     101.0

— query mix: does the shard key apply —
a members loans (shard key applies)    shards  1 | median   4.5 | p99    6.0 ms
a books loan history (no key)          shards  8 | median   5.8 | p99  100.9 ms
monthly report (no key)                shards  8 | median   5.8 | p99  100.9 ms

The model assumes every shard has a one-percent chance of responding slowly. Median time is barely touched by this assumption: 4.5 ms at one shard, 5.8 ms at eight shards. The tail collapses instead. The 99th percentile is 6.0 ms at one shard and 100.9 ms at eight shards — a seventeenfold increase.

The arithmetic behind it is direct. If a single shard’s chance of being slow is one percent, the chance that at least one of eight shards is slow is $1 - 0.99^8 \approx 0.077$, that is, 7.7 percent. Scatter-gather turns every shard’s rare flaw into the query’s frequent flaw.

From this comes the most commonly misunderstood side of sharding: sharding grows throughput, it does not shrink a single query’s latency. A measurement that looks only at average times never sees this degradation; seeing it requires looking at the tail percentiles.

Joining, Uniqueness, and Transactions

Three abilities change in nature at the shard boundary.

Joining. Loan records sharded by member ID can be joined, within a shard, with a member table sharded by the same key; the two tables’ related rows sit on the same machine. This is called co-location (co-location), and it is the only way for a join to stay cheap in a sharded design. The book table cannot be co-located with loan records, because a loan record cannot belong to two different keys at once. The common solution is to copy small, rarely changing tables to every shard; in the library example the branch list is such a table.

Uniqueness. In a single engine, a uniqueness constraint is enforced with an index. No such index spans shards: each shard sees only its own rows, so uniqueness can only be given within a shard. A field requiring global uniqueness — a member’s email address, for instance — either becomes the shard key or is kept in a separate lookup table. Auto-incrementing identifiers fail for the same reason; in sharded setups, identifiers are drawn not from a central source but from a generator that either embeds the shard number or is wide enough that collision probability is negligible.

Transactions. The atomicity defined in the ACID Properties lesson relies on a single engine’s log. A transaction writing to two shards at once writes to two separate logs, and getting the two to commit together requires a separate protocol: the coordinator first sends a prepare command to every shard, and commits only if it receives approval from all of them. This protocol’s cost is not the extra round trips but the possibility of hanging — if the coordinator fails between prepare and commit, the shards hold their locks and wait. This is why the golden rule of a sharded design is: a transaction must stay within a single shard. If it cannot, the design has chosen either the wrong shard key or the wrong transaction boundary.

Rebalancing

Shard count does not stay fixed. When a new shard is added, the key-to-shard mapping changes, and every changed mapping means data has to move. How much moves depends on how the mapping was built, and this can be computed.

cat > rebalance.mjs <<'JS'
// Model: the fraction of rows that must move when the shard count changes. Not a real
// rebalancing procedure; what is counted is only the mapping's effect on move cost.
const MEMBERS = 120000, BUCKETS = 4096;
const M = 2654435761n;
const hash = (x) => Number(((BigInt(x) * M) % 4294967296n) >> 8n);

// Loans per member: a few members hold many loans.
const member = [];
{
  let total = 0;
  const weight = [];
  for (let i = 1; i <= MEMBERS; i++) { const a = 1 / Math.pow(i, 0.8); weight.push(a); total += a; }
  for (let i = 0; i < MEMBERS; i++)
    member.push({ id: i + 1, loans: Math.max(1, Math.round(weight[i] / total * 2000000)) });
}
const totalRows = member.reduce((t, m) => t + m.loans, 0);

function modMove(oldCount, newCount) {
  let moved = 0;
  for (const m of member) if (hash(m.id) % oldCount !== hash(m.id) % newCount) moved += m.loans;
  return moved;
}
// Virtual bucket: the key always falls into a fixed number of buckets, and buckets are assigned to shards.
function bucketAssignment(P) {
  const assignment = new Array(BUCKETS);
  for (let b = 0; b < BUCKETS; b++) assignment[b] = b % P;
  return assignment;
}
function bucketMove(oldCount, newCount) {
  const oldAssignment = bucketAssignment(oldCount);
  const target = Math.floor(BUCKETS / newCount);
  const count = new Array(newCount).fill(0);
  const newAssignment = new Array(BUCKETS).fill(-1);
  for (let b = 0; b < BUCKETS; b++) {                 // buckets that can stay in place come first
    const p = oldAssignment[b];
    if (p < newCount && count[p] < target) { newAssignment[b] = p; count[p]++; }
  }
  let p = 0;
  for (let b = 0; b < BUCKETS; b++) {                 // remaining buckets go to the shards below target
    if (newAssignment[b] !== -1) continue;
    while (p < newCount - 1 && count[p] >= target) p++;
    newAssignment[b] = p; count[p]++;
  }
  let moved = 0;
  for (const m of member) {
    const b = hash(m.id) % BUCKETS;
    if (oldAssignment[b] !== newAssignment[b]) moved += m.loans;
  }
  return moved;
}
console.log('total loans in model:', totalRows);
console.log('transition       direct modulo      virtual bucket');
for (const [o, n] of [[8, 9], [8, 12], [8, 16]]) {
  const a = modMove(o, n), b = bucketMove(o, n);
  const pct = (x) => (100 * x / totalRows).toFixed(1) + '%';
  console.log(`${o} → ${n}`.padEnd(12),
              (a.toLocaleString('en-US') + ' (' + pct(a) + ')').padStart(21),
              (b.toLocaleString('en-US') + ' (' + pct(b) + ')').padStart(18));
}
JS
node rebalance.mjs

total loans in model: 2001046
transition       direct modulo      virtual bucket
8 → 9            1,793,865 (89.6%)    228,270 (11.4%)
8 → 12           1,337,553 (66.8%)    675,360 (33.8%)
8 → 16           1,006,912 (50.3%)    982,390 (49.1%)

Direct modulo — hashing the key and dividing by the shard count, then taking the remainder — moved 89.6 percent of records when going from eight to nine shards. Adding a single shard requires nearly all the data to shift, because when the divisor changes, the remainder changes for nearly every key too.

The virtual bucket (virtual bucket) approach inserts a layer in between: the key is first reduced to a fixed number of buckets (here, 4,096), and buckets are assigned to shards. When shard count changes, the key-to-bucket mapping never changes; only the bucket-to-shard assignment is updated, and only the data of buckets handed to the new shard moves. Going from eight to nine, the moved ratio came out at 11.4 percent — very close to the theoretical lower bound of 1/9 ≈ 11.1 percent.

The third row carries a warning. Going from eight to sixteen shards, the two methods come out equal (50.3 percent versus 49.1 percent), because doubling the shard count is direct modulo’s luckiest case: every shard splits exactly in two. This makes doubling a valid growth strategy — but only for as long as it can keep doubling. When capacity planning is forced into that pattern, the cost is provisioning capacity that is not yet needed, paid for in advance.

The moved ratio itself is not the final cost either. During the move, two copies live side by side for a while, reads must be answerable from either location, and writes to buckets in flight either have to be paused or written to both places at once. This is why rebalancing is a capability measured not by shard count but by whether the move can be carried out without interruption.

When Sharding Is Not Yet Needed

Sharding carries every cost counted above into the application’s code. This is why the last question is whether the ways of avoiding it have run out. The three approaches measured in this course usually arrive first: the right indexes cut read cost by orders of magnitude, partitioning turned removing old data into a fixed cost, and read load can be spread out through replication.

The one factor that truly makes sharding necessary is write load, and only once it exceeds a single engine’s log and checkpoint capacity. That is the measure the decision is made by — not the measure of “the table is big.”

Summary

• Sharding distributes data across independent database instances; the shard key decides which query stays within a single shard, and changing it later means moving all the data.
• The scatter-gather pattern barely affects median time but wrecks the tail: in the model, the 99th percentile that was 6.0 ms at one shard became 100.9 ms at eight shards.
• Joining stays cheap only across co-located tables; uniqueness can only be enforced within a shard; a transaction writing to two shards needs a separate protocol and carries the risk of hanging.
• With direct modulo, going from eight to nine shards moved 89.6 percent of records; with a virtual bucket layer, the same transition dropped to 11.4 percent.
• The factor that makes sharding necessary is write load exceeding a single engine’s capacity; indexing, partitioning, and replication remain the cheaper paths up to that threshold.

Next Step

This topic has completed how data is organized and how it is read: index types, column order, covering, maintenance, partitioning, and sharding. Together with the Engine Architecture topic, where data sits and how it is found are now established — the system is operational. Being operational and being administered are different things. The next topic opens with that distinction, and its first question is the oldest one: who can access this data, and what can they do? The application at the loan desk needs to open a loan record; it does not need to delete a member’s record. How that distinction is built at the privilege level is the next lesson’s subject.