---
title: 'Node Roles'
source: 'https://academia.sh/en/courses/search-engines/node-roles'
course: 'Search Engines and Text Retrieval'
language: en
updated: '2026-08-23T07:00:50+00:00'
license: 'CC BY-SA 4.0'
---

# Node Roles

Moving search from a single process to a cluster: separating the data, coordinating, and master node roles, running the same three hundred queries under two role placements and counting the postings each node reads and the merge steps it performs, showing that as shard count grows the data work splits while the coordinating work does not, and measuring how placing the master role on a loaded data node delays cluster state and changes the returned document set and the order of the top five results.

The previous topic ran every query in a single process over a single index: the query
evaluator read the posting lists directly, and the returned order was the scorer's own
order. As the catalog grows, this arrangement breaks in two places — the index no longer
fits in one machine's memory, and a single process cannot keep up with the incoming query
rate. Once the index spreads across multiple machines, the query itself does not change;
what changes is who answers it, who reads the posting lists, and who merges the returned
candidates. This lesson splits those three tasks into three roles and counts what changes as
the roles are distributed over the cluster.

## Three Roles

A **data node** carries one shard: that shard's inverted index lives in its memory. When a
query arrives it scans only its own shard and returns the top candidates from its own local
order.

A **coordinating node** receives the query, sends one request to each shard, merges the
returned candidate lists, and produces the final list; it reads no posting lists itself.

A **master node** takes no part in the query path at all. What it carries is cluster state:
which shard sits on which node, which mapping is in effect, where shards get assigned when a
node is added. It is the only node that can propagate a new version of that state to the
cluster.

The setup below models these three roles **in process**: the nodes are objects in the same
process, and there is no network or message delay. Because it is a model, what it measures is
not time but the work each node does — postings read and merge steps performed, both
independent of the run.

**CO1 — the query mix consists of four catalog questions and holds an equal share across 300
rounds.**
**CO2 — applying one cluster state change costs 500 work units, and indexing runs at one
document per 20 work units.** Both assumptions have a linear effect: if the values change,
the delay and rejected-document count below change in the same proportion.

```js
// cluster/corpus.mjs — library catalog corpus and inverted index. The generator is self-written,
// the seed is visible: every run produces the same 4000 records. 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; }
      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/roles.mjs — the work master, data, and coordinating nodes do per query. The cluster is
// modeled IN-PROCESS: nodes are objects in the same process, there is no network. What is measured
// is not time but the work each node does — the postings read and the merge steps taken, both
// independent of the run.
import { corpus, buildIndex, search, merge } from "./corpus.mjs";

const P = 4, KP = 10, K = 10, ROUNDS = 300, STATE_WORK = 500, DOC_WORK = 20, NEW = 800;
const 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)));

// One query round: the coordinating node goes to every shard, collects KP local candidates per
// shard, and merges them.
function query(index, terms, coord, work) {
  const lists = [];
  for (let i = 0; i < index.length; i += 1) {
    const r = search(index[i], terms, KP);
    work[`d${i}`] = (work[`d${i}`] ?? 0) + r.scanned;
    lists.push(r.candidates);
  }
  const b = merge(lists, K);
  work[coord] = (work[coord] ?? 0) + b.steps;
  return { result: b.result, candidates: lists.reduce((t, l) => t + l.length, 0), steps: b.steps };
}

const docs = corpus(), whole = buildIndex(docs), index = shard(docs, P), load = {};
console.log(`${docs.length} documents, ${P} shards, ${KP} candidates per shard, final list ${K}. Seed 20260731.`);
console.log(`single-shard index: ${whole.postings.size} terms, ${whole.bytes} bytes. Query rounds: ${ROUNDS},`);
console.log(`candidates reaching the coordinating node: ${ROUNDS * P * KP}.\n`);
console.log("placement       | node  | roles                     | work units | share");
console.log("-----------------|-------|---------------------------|-----------|------");
for (const separate of [false, true]) {
  const work = {}, coord = separate ? "e0" : "d0";
  for (let t = 0; t < ROUNDS; t += 1) query(index, QUERIES[t % QUERIES.length], coord, work);
  const nodes = separate ? ["a0", "e0", "d0", "d1", "d2", "d3"] : ["d0", "d1", "d2", "d3"];
  const role = (n) => n === "a0" ? "master" : n === "e0" ? "coordinating"
    : (n === "d0" && !separate) ? "master+data+coordinating" : "data";
  const total = nodes.reduce((t, n) => t + (work[n] ?? 0), 0);
  load[separate ? "separate" : "combined"] = work;
  for (const n of nodes)
    console.log(`${(separate ? "roles separate" : "roles combined").padEnd(16)} | ${n.padEnd(5)} | ${role(n).padEnd(25)} | ` +
      `${pad(work[n] ?? 0, 9)} | ${pad("%" + (100 * (work[n] ?? 0) / total).toFixed(1), 5)}`);
}

console.log("\ncoordinating node work per query (same corpus, shard count varies):");
console.log("shards | candidates collected | merge steps | coordinating share | entries scanned per data node");
for (const p of [1, 2, 4, 8, 16]) {
  const work = {}, r = query(shard(docs, p), QUERIES[0], "e0", work);
  const data = Object.entries(work).filter(([n]) => n !== "e0").reduce((t, [, v]) => t + v, 0);
  console.log(`${pad(p, 6)} | ${pad(r.candidates, 21)} | ${pad(r.steps, 11)} | ` +
    `${pad("%" + (100 * r.steps / (data + r.steps)).toFixed(1), 19)} | ${pad(Math.round(data / p), 30)}`);
}

// The master node takes no part in the query path; it propagates cluster state. If a fifth shard
// is requested after query 150, and the master role is placed on a loaded data node, that work
// falls behind that node's query queue. Until the assignment is applied, documents routed to the
// new shard are rejected.
console.log(`\nmaster node placement: after query 150, a fifth shard is requested, ${NEW} new records are waiting`);
console.log("placement       | master's queue | state delay | rejected | indexed docs | matched | top 5 results");
const wide = corpus({ count: 4000 + NEW });
for (const separate of [false, true]) {
  const queue = separate ? 0 : Math.round(load.combined.d0 * (ROUNDS - 150) / ROUNDS);
  const delay = queue + STATE_WORK;
  const rejected = Math.min(NEW, Math.floor(delay / DOC_WORK));
  const kept = wide.filter((d) => d.id <= 4000 || d.id > 4000 + rejected);
  const ix = shard(kept, rejected === NEW ? P : P + 1), work = {};
  const last = query(ix, QUERIES[0], "e0", work);
  const matched = ix.reduce((t, x) => t + search(x, QUERIES[0], K).matched, 0);
  console.log(`${(separate ? "roles separate" : "roles combined").padEnd(16)} | ${pad(queue, 14)} | ${pad(delay, 11)} | ` +
    `${pad(rejected, 8)} | ${pad(kept.length, 12)} | ${pad(matched, 7)} | ${last.result.slice(0, 5).map((x) => x.id).join(" ")}`);
}
console.log(`\nrun-independent: the coordinating node collects P x ${KP} candidates per query. As shards` +
  ` are added, per-data-node work is divided; the coordinating node's work is not divided.`);
```

```
4000 documents, 4 shards, 10 candidates per shard, final list 10. Seed 20260731.
single-shard index: 80 terms, 382832 bytes. Query rounds: 300,
candidates reaching the coordinating node: 12000.

placement       | node  | roles                     | work units | share
-----------------|-------|---------------------------|-----------|------
roles combined   | d0    | master+data+coordinating  |     94875 | %27.4
roles combined   | d1    | data                      |     78375 | %22.6
roles combined   | d2    | data                      |     89100 | %25.7
roles combined   | d3    | data                      |     84150 | %24.3
roles separate   | a0    | master                    |         0 |  %0.0
roles separate   | e0    | coordinating              |     12000 |  %3.5
roles separate   | d0    | data                      |     82875 | %23.9
roles separate   | d1    | data                      |     78375 | %22.6
roles separate   | d2    | data                      |     89100 | %25.7
roles separate   | d3    | data                      |     84150 | %24.3

coordinating node work per query (same corpus, shard count varies):
shards | candidates collected | merge steps | coordinating share | entries scanned per data node
     1 |                    10 |          10 |                %0.8 |                           1238
     2 |                    20 |          20 |                %1.6 |                            619
     4 |                    40 |          40 |                %3.1 |                            310
     8 |                    80 |          80 |                %6.1 |                            155
    16 |                   160 |         160 |               %11.4 |                             77

master node placement: after query 150, a fifth shard is requested, 800 new records are waiting
placement       | master's queue | state delay | rejected | indexed docs | matched | top 5 results
roles combined   |          47438 |       47938 |      800 |         4000 |     515 | 2689 701 2181 3159 1777
roles separate   |              0 |         500 |       25 |         4775 |     624 | 3159 3544 2689 701 2362

run-independent: the coordinating node collects P x 10 candidates per query. As shards are added, per-data-node work is divided; the coordinating node's work is not divided.
```

## Where Per-Query Work Falls

The first table runs the same 300 queries under two role placements and **returns the same
result**: since both placements read the same shard indexes, the returned document set and
order are identical either way. What changes is where the work falls.

With combined roles, queries all land at one entry point, d0; d0 both scans its own shard and
does the entire merge. It carries 94,875 work units, 27.4 percent of the cluster's total,
while the lightest node, d1, carries 22.6 percent. The 12,000-unit gap between them is the
coordinating work itself; once the roles are split, those units move to e0 and d0 drops to
82,875 units.

In the second row group, the master node's share comes out at **0.0 percent**: across 300
queries, a0 read not one posting and made not one comparison. The master node's load grows
not with query count but with **how often cluster state changes**.

## Coordinating Work Does Not Split

The second table splits the same corpus into 1, 2, 4, 8, and 16 shards and runs a single
query. Two columns move in opposite directions. Postings scanned per data node drop from
1238 to 77 — dividing exactly by shard count, since the same posting lists are spread across
the shards. The coordinating node's collected candidates and merge steps, by contrast, rise
from 10 to 160: every shard must send its own top ten candidates, and the merger compares
every list head for every result.

The coordinating share rising from 0.8 percent to 11.4 percent is the sum of these two
curves: adding shards divides the cheap side and gathers the expensive side onto a single
node.

## Master Node Placement Changes the Cluster

The third table asks for a fifth shard to be added to the cluster after query 150; 800 new
catalog records are waiting to be routed to that shard. When the master role is loaded onto
d0, the state change falls behind that node's query queue: 47,438 units of accumulated query
work, then 500 units of state work. At CO2's indexing rate, that means all 800 records are
rejected, and the corpus stays at 4000 documents. With the master role on its own node, the
delay is 500 units, rejected records are 25, and the index grows to 4775 documents.

The difference shows up in the query result. The same query — records about a short story
touching on loneliness — matches 515 documents under combined placement and 624 under
separate placement: a gap of 109 documents. The top five results differ too. Under combined
placement the order is 2689, 701, 2181, 3159, 1777; under separate placement it is 3159,
3544, 2689, 701, 2362. Three documents are shared, but none at the same rank, and two list
members change entirely. Role placement is not a query setting; it nonetheless changes the
returned set and its order, because it determines which documents could even enter the
index.

## Summary

- The three roles map to three separate tasks: the data node reads posting lists, the
  coordinating node collects and merges candidates, and the master node propagates cluster
  state and takes no part in the query path.
- Splitting the roles did not change the returned set or order; across 300 queries it moved
  12,000 work units from d0 to the coordinating node, and the heaviest node's share fell from
  27.4 percent to 25.7 percent. In the same run the master node's share stayed at 0.0
  percent: its load grows not with query count but with state changes.
- As shard count rose from 1 to 16, postings scanned per data node fell from 1238 to 77,
  while the coordinating node's candidates and merge steps rose from 10 to 160, its share
  from 0.8 percent to 11.4 percent.
- Loading the master role onto a busy data node delayed the state change by 47,938 units and
  rejected all 800 records; on its own node the delay was 500 units and 25 records were
  rejected. That delay carried into the query: matched documents were 515 versus 624, and the
  order of the top five changed entirely.

## Next Step

This lesson held shard count fixed and pulled a fixed number of candidates from each shard;
whether either number was well chosen was never questioned. A heavier question is still
open: each shard picks its own top ten candidates using **its own local statistics**, and the
merger trusts those local scores. As shard count grows, the document frequency each shard
measures drifts further from the global value, and the candidates it selects change. The
next lesson raises shard count from 1 to 16 and counts how far the top ten results drift from
the single-shard index's top ten, then adds replica count's effect on index size and on node
loss.