Shards and Replicas

The previous lesson held shard count at four and pulled a fixed number of candidates from each shard. Both numbers quietly assumed something: that the top ten candidates a shard sends are truly that shard’s ten best documents. But a shard computes its score using only the document frequency it can see. In a four-shard cluster each shard sees a quarter of the corpus; in a sixteen-shard cluster, a sixteenth. As a term thins out, its inverse document frequency drifts, and what a shard calls “best” changes. This lesson raises shard count from 1 to 16 and counts how far the same four catalog queries’ top ten result drifts, then measures replica count as a separate decision.

Setup

The baseline is fixed: the single-shard index’s top ten result — the only placement that sees the whole corpus’s statistics. Two accuracy measures are used: shared document count and the average rank shift of shared documents. The cost columns are collected candidates, merge steps, and total index bytes.

The cluster is again modeled in process: shards are separate inverted indexes in the same process, and there is no network. What is measured is not time but request counts and countable structure sizes.

CO3 — documents are distributed to shards by id remainder, so the split is content-independent and balanced. This is the best case: in a corpus split by topic, each shard’s word distribution would be more skewed and the drift below would grow larger. CO4 — a query goes to only one copy of each shard, and copies carry the same documents; replica lag is taken as zero.

// cluster/corpus.mjs — library catalog corpus and inverted index (same generator and seed as the
// previous lesson; this lesson carries only the parts it uses). The index holds a dictionary:
// term -> posting list; a posting here is a document entry (document id + term frequency).
const COMMON = ["book", "author", "work", "text", "chapter", "edition", "page", "language", "volume", "publication"];
const SPECIAL = {
  child: ["tale", "illustrated", "school", "play", "animal", "cartoon"], story: ["short", "narrative", "collection", "diary", "loneliness", "village"],
  novel: ["hero", "borough", "generation", "house", "journey", "letter"], history: ["empire", "document", "archive", "war", "century", "chronicle"],
  travel: ["sea", "road", "map", "city", "harbor", "ship"], poetry: ["verse", "meter", "image", "sound", "silence", "rhyme"],
  essay: ["thought", "critique", "reading", "time", "note", "conversation"], science: ["measurement", "experiment", "theory", "data", "observation", "equation"],
};
const RARE = ["lighthouse", "well", "silk", "tower", "garden", "snow", "island", "grove", "stone", "sycamore", "swallow", "blacksmith", "compass", "amber"];
const TOPIC = Object.keys(SPECIAL);

export function corpus({ count = 4000, seed = 20260731 } = {}) {
  let s = seed % 2147483647;
  const r = () => (s = (s * 48271) % 2147483647) / 2147483647;
  const pick = (a) => a[Math.floor(r() * a.length)], docs = [];
  for (let i = 1; i <= count; i += 1) {
    const topic = pick(TOPIC), sp = SPECIAL[topic], terms = [pick(sp), pick(RARE)];
    if (r() < 0.5) terms.push(pick(COMMON));
    for (let j = 0, n = 10 + Math.floor(r() * 7); j < n; j += 1)
      terms.push(r() < 0.45 ? pick(COMMON) : r() < 0.85 ? pick(sp) : pick(RARE));
    docs.push({ id: i, topic, year: 1990 + Math.floor(r() * 36), text: [terms[0], terms[1], topic, ...terms.slice(2)].join(" ") });
  }
  return docs;
}

export function buildIndex(docs) {
  const postings = new Map(), lengths = new Map();
  let bytes = 0;
  for (const d of docs) {
    const t = d.text.split(" "), counts = new Map();
    for (const x of t) counts.set(x, (counts.get(x) ?? 0) + 1);
    lengths.set(d.id, t.length);
    for (const [x, n] of counts) {
      if (!postings.has(x)) { postings.set(x, []); bytes += x.length + 4; }   // dictionary entry
      postings.get(x).push([d.id, n]); bytes += 8;
    }
  }
  const avg = [...lengths.values()].reduce((a, b) => a + b, 0) / (docs.length || 1);
  return { postings, lengths, N: docs.length, avg, bytes };
}

// Scoring: term frequency, inverse document frequency, and document length. In LOCAL scoring df
// and average length are read only from this index; if global is given, the whole corpus's
// statistics are used instead.
export function search(ix, terms, k, global = null) {
  const N = global ? global.N : ix.N, avg = global ? global.avg : ix.avg, scores = new Map();
  let scanned = 0;
  for (const t of terms) {
    const g = ix.postings.get(t) ?? [];
    const df = global ? global.df.get(t) ?? 0 : g.length;
    const idf = Math.log(1 + (N - df + 0.5) / (df + 0.5));
    for (const [id, tf] of g) {
      scanned += 1;
      const norm = tf + 1.2 * (0.25 + 0.75 * ix.lengths.get(id) / avg);
      scores.set(id, (scores.get(id) ?? 0) + idf * tf * 2.2 / norm);
    }
  }
  const sorted = [...scores].sort((a, b) => b[1] - a[1] || a[0] - b[0]).slice(0, k);
  return { candidates: sorted.map(([id, p]) => ({ id, p })), scanned, matched: scores.size };
}

// k-way merge: at each step the heads of the shard lists are compared, the highest score is
// taken. The step count is the real comparison count and is machine-independent.
export function merge(lists, k) {
  const pos = lists.map(() => 0), result = [];
  let steps = 0, best = 0;
  while (result.length < k && best >= 0) {
    best = -1;
    for (let i = 0; i < lists.length; i += 1) {
      const a = lists[i][pos[i]], b = best < 0 ? null : lists[best][pos[best]];
      if (!a) continue;
      steps += 1;
      if (!b || a.p > b.p || (a.p === b.p && a.id < b.id)) best = i;
    }
    if (best >= 0) { result.push(lists[best][pos[best]]); pos[best] += 1; }
  }
  return { result, steps };
}

// cluster/shards.mjs — the effect of shard count and candidate depth on the top ten result. The
// cluster is modeled IN-PROCESS: shards are separate inverted indexes in the same process. The
// baseline is the single-shard index's top ten result; every placement is compared against it by
// shared documents and rank shift.
import { corpus, buildIndex, search, merge } from "./corpus.mjs";

const K = 10, QUERIES = [["short", "story", "loneliness"], ["sea", "harbor", "ship"],
  ["tale", "illustrated", "child"], ["archive", "document", "century"]];
const pad = (x, n) => String(x).padStart(n);
const shard = (b, p) => Array.from({ length: p }, (_, i) => buildIndex(b.filter((d) => d.id % p === i)));

// Global statistics: document frequency is summed across every shard. Using this costs one extra
// round — the coordinating node first sums df, then has the shards score with it (a two-round query).
function global(ix) {
  const df = new Map();
  let N = 0, len = 0;
  for (const d of ix) {
    N += d.N; len += d.avg * d.N;
    for (const [t, g] of d.postings) df.set(t, (df.get(t) ?? 0) + g.length);
  }
  return { df, N, avg: len / N };
}

// One query: each shard returns KP local candidates, the coordinating node merges. If glo is
// given, shards score with global statistics instead (a two-round query).
function ask(ix, terms, KP, glo = null) {
  const lists = ix.map((d) => search(d, terms, KP, glo).candidates);
  const b = merge(lists, K);
  return { top: b.result.map((x) => x.id), steps: b.steps, candidates: lists.reduce((t, l) => t + l.length, 0) };
}

const docs = corpus(), single = buildIndex(docs);
const baseline = QUERIES.map((q) => search(single, q, K).candidates.map((x) => x.id));
const compare = (a, i) => {                     // shared documents and total rank shift vs. baseline
  let shared = 0, shift = 0;
  a.forEach((id, j) => { const k = baseline[i].indexOf(id); if (k >= 0) { shared += 1; shift += Math.abs(k - j); } });
  return { shared, shift };
};
const average = (f) => { const t = QUERIES.map(f); return t.reduce((a, b) => a + b, 0) / t.length; };

console.log(`${docs.length} documents, seed 20260731, ${single.postings.size} terms. Single-shard index: ${single.bytes} bytes.`);
console.log(`Baseline: single-shard index's top ${K} result, averaged over four catalog queries.\n`);
console.log("shards | index bytes | candidates | merge steps | depth 10: shared / shift | depth 50 | global df round");
console.log("-------|-------------|------------|-------------|---------------------------|----------|----------------");
for (const P of [1, 2, 4, 8, 16]) {
  const ix = shard(docs, P), glo = global(ix);
  const bytes = ix.reduce((t, d) => t + d.bytes, 0);
  const d10 = average((q, i) => compare(ask(ix, q, 10).top, i).shared);
  const k10 = average((q, i) => { const x = compare(ask(ix, q, 10).top, i); return x.shift / x.shared; });
  const d50 = average((q, i) => compare(ask(ix, q, 50).top, i).shared);
  const dglo = average((q, i) => compare(ask(ix, q, 10, glo).top, i).shared);
  const r = ask(ix, QUERIES[0], 10);
  console.log(`${pad(P, 6)} | ${pad(bytes, 11)} | ${pad(r.candidates, 10)} | ${pad(r.steps, 11)} | ` +
    `${pad(d10.toFixed(2) + " / " + k10.toFixed(2), 25)} | ${pad(d50.toFixed(2), 8)} | ${pad(dglo.toFixed(2), 15)}`);
}

console.log("\ntop 10 result (query: short story loneliness):");
console.log(`${"single shard".padEnd(23)}: ${baseline[0].join(" ")}`);
for (const P of [4, 16]) {
  const ix = shard(docs, P);
  console.log(`${(P + " shards, depth 10").padEnd(23)}: ${ask(ix, QUERIES[0], 10).top.join(" ")}`);
  console.log(`${(P + " shards, global df").padEnd(23)}: ${ask(ix, QUERIES[0], 10, global(ix)).top.join(" ")}`);
}

// Replica: R+1 copies of each shard sit on separate nodes. A query goes to ONE copy of each
// shard; replica count does not change query width, it changes index bytes and node loss.
const P = 4, ix = shard(docs, P), bytes = ix.reduce((t, d) => t + d.bytes, 0);
console.log("\nreplica count (4 shards, one node lost):");
console.log("replicas | nodes | total index bytes | nodes touched | unreachable shards | matched | remaining in top 10");
for (const R of [0, 1, 2]) {
  const lost = R === 0 ? 1 : 0;                  // R=0: one node loss takes an entire shard down
  const remaining = ix.filter((_, i) => i >= lost);
  const matched = remaining.reduce((t, d) => t + search(d, QUERIES[0], K).matched, 0);
  const top = ask(remaining, QUERIES[0], 10).top;
  console.log(`${pad(R, 8)} | ${pad(P * (R + 1), 5)} | ${pad(bytes * (R + 1), 18)} | ${pad(P, 13)} | ` +
    `${pad(lost, 19)} | ${pad(matched, 7)} | ${pad(top.filter((x) => baseline[0].includes(x)).length + "/10", 19)}`);
}
console.log(`\nrun-independent: collected candidates are P x depth, merge steps are ${K} x P, index bytes` +
  ` scale by (R+1). A single-round query costs P requests, the global df round costs 2P. The` +
  ` number of nodes a query touches does not change with replica count, only with shard count.`);

4000 documents, seed 20260731, 80 terms. Single-shard index: 382832 bytes.
Baseline: single-shard index's top 10 result, averaged over four catalog queries.

shards | index bytes | candidates | merge steps | depth 10: shared / shift | depth 50 | global df round
-------|-------------|------------|-------------|---------------------------|----------|----------------
     1 |      382832 |         10 |          10 |              10.00 / 0.00 |    10.00 |           10.00
     2 |      383656 |         20 |          20 |               8.50 / 1.28 |     8.50 |           10.00
     4 |      385304 |         40 |          40 |               6.25 / 2.50 |     6.25 |           10.00
     8 |      388600 |         80 |          80 |               4.75 / 2.04 |     4.75 |           10.00
    16 |      395192 |        160 |         160 |               4.75 / 4.01 |     4.75 |           10.00

top 10 result (query: short story loneliness):
single shard           : 3159 3544 2689 2362 3370 701 2448 502 2181 2320
4 shards, depth 10     : 2689 701 2181 3159 1777 549 3544 2362 3370 565
4 shards, global df    : 3159 3544 2689 2362 3370 701 2448 502 2181 2320
16 shards, depth 10    : 2689 1777 3265 2323 2180 2529 3544 502 2181 1150
16 shards, global df   : 3159 3544 2689 2362 3370 701 2448 502 2181 2320

replica count (4 shards, one node lost):
replicas | nodes | total index bytes | nodes touched | unreachable shards | matched | remaining in top 10
       0 |     4 |             385304 |             4 |                   1 |     379 |                6/10
       1 |     8 |             770608 |             4 |                   0 |     515 |                7/10
       2 |    12 |            1155912 |             4 |                   0 |     515 |                7/10

run-independent: collected candidates are P x depth, merge steps are 10 x P, index bytes scale by (R+1). A single-round query costs P requests, the global df round costs 2P. The number of nodes a query touches does not change with replica count, only with shard count.

Shard Count Corrupts the Top Ten

The first table’s cost columns behave as expected: collected candidates and merge steps grow linearly with shard count, from 10 to 160. Total index bytes rise from 382,832 to 395,192, a 3.2 percent increase; the source is not the posting lists but the dictionary — the same 80 terms are repeated in every shard.

The real number is in the accuracy columns. At two shards, the top ten shares an average of 8.5 documents with the baseline; at four, 6.25; at eight and sixteen, 4.75. At sixteen shards, more than half the top ten result is absent from the baseline entirely, and the average rank shift of the shared documents rises from 1.28 to 4.01.

The lists show this directly. The single-shard index orders 3159, 3544, 2689 first; the same query under four shards returns 2689, 701, 2181 — the baseline’s first result, 3159, drops to fourth place, and 1777, 549, and 565, none in the baseline, enter the list. At sixteen shards, 3159 drops out of the top ten entirely and only four documents remain shared with the baseline. No query changed, no document was deleted; the only thing that changed is how many documents each shard can see.

Depth Does Not Fix It, an Extra Round Does

The obvious way to close the gap is to pull more candidates from each shard. The table refutes this: raising candidate depth from 10 to 50 leaves the shared document count unchanged (8.50 / 6.25 / 4.75 / 4.75). The error is not coming from truncation: each shard sorts its list by its own local score, and the merger trusts those scores. What is wrong is not the length of the ranking but the ranking itself.

What fixes it is in the last column: the coordinating node first sums per-term document frequency across every shard, then has the shards score using that global statistic. Shared document count returns to 10.00 at every shard count; the four- and sixteen-shard lists are identical to the single-shard index’s. The cost is that the query goes from one round to two, from P requests to 2P.

Replica Count Preserves the Set, Not the Order

The second table measures replicas as a separate decision. In a replica-free, four-shard cluster, losing one node takes an entire shard down with it: matched documents fall from 515 to 379 — 136 catalog records never appear to the query — and shared top ten documents drop from 7 to 6. Adding one replica means the same node loss changes nothing.

The cost is index bytes, and it is linear: from 385,304 to 770,608, and to 1,155,912 at two replicas. The number of nodes a query touches stays at 4 across all three rows — adding a replica does not widen the query.

The last column staying at 7/10 is this lesson’s distinction: replicas brought back the lost documents, not the wrong order. That three-document gap comes from local scoring, and only a global-statistics round closes it. Redundancy is a durability decision; sharding, at the same time, is a scoring decision.

Summary

• As shard count rose from 1 to 16, collected candidates and merge steps rose from 10 to 160, and total index bytes rose from 382,832 to 395,192; the increase comes not from the posting lists but from the dictionary repeated in every shard.
• The top ten’s shared document count with the baseline fell to 6.25 at four shards and 4.75 at sixteen; the shared documents’ average rank shift rose to 4.01.
• Raising candidate depth from 10 to 50 changed accuracy not at all: the error does not come from truncation but from per-shard local document frequency.
• A global document-frequency round returned the top ten to 10.00 at every shard count and made the lists identical to the single-shard index’s; the cost is the query rising from P requests to 2P.
• In a replica-free cluster, one node loss dropped matched documents from 515 to 379; one replica prevented this entirely, doubled index bytes, and did not change the number of nodes touched. Replicas bring back the lost set, not the corrupted order: the top ten’s shared document count stayed at 7/10.

Next Step

This lesson measured replicas as a durability decision and assumed replicas carry the same documents; that holds while the cluster is healthy. When the network splits in two, replicas can no longer see each other and each half can accept writes on its own: documents enter the same index from two places, and the two halves’ posting lists diverge. The next lesson splits a five-replica index into a three-node half and a two-node half, toggles the majority requirement, and counts two things — how many of the indexing operations accepted in the minority are lost once the split is repaired, and how many documents each half’s query returns short of the full set while the split lasts.