---
title: 'Shards and Replicas'
source: 'https://academia.sh/en/courses/search-engines/shards-and-replicas'
course: 'Search Engines and Text Retrieval'
language: en
updated: '2026-08-23T07:00:51+00:00'
license: 'CC BY-SA 4.0'
---

# Shards and Replicas

Showing that shard count is a scoring decision: splitting the same corpus into one, two, four, eight, and sixteen shards and comparing each placement's top ten result with the single-shard index's top ten by shared documents and rank shift, measuring that raising candidate depth does not fix the drift, counting that a global document-frequency round restores the top ten exactly and costs one extra round, and separately measuring how replica count changes index bytes and the returned document set under node loss.

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.**

```js
// 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 };
}
```

```js
// 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.