Distributed Indexing

"I am an inverted index, split across shards that have never met. Each of us swears it holds the whole corpus, and each of us is exactly one fifth right."

A Posting List Convinced It Is the Whole Vocabulary

The previous sections of this chapter took a raw web crawl and turned it into a clean, deduplicated, chunked corpus. That corpus is now a pile of text passages, each with an identifier, and a retrieval system that cannot find anything in it is useless. Retrieval has two complementary engines. The lexical engine matches the exact words of a query against the exact words of the documents and is unbeatable when the query contains a rare identifier, a product code, or a proper noun that no embedding model has memorized. The dense engine, built in Section 36.5 and sharded for search in Section 36.6, matches meaning rather than surface form and recovers documents that share no words with the query. A production RAG system runs both and fuses their scores. This section builds the lexical engine and the distributed index underneath it; the vector index of Chapter 25 is its structural twin.

1. The Inverted Index: Term to Posting List Beginner

A corpus stored the obvious way, as a list of documents, answers the wrong question. It can tell you what words document seven contains, but a search engine needs the reverse: given a word, which documents contain it. The inverted index is the data structure that answers that reverse question directly. For every distinct term in the vocabulary it stores a posting list, the sorted set of identifiers of the documents in which the term appears, usually annotated with the per-document term frequency and sometimes the token positions. The name "inverted" is literal: it inverts the document-to-terms map that the corpus gives you for free into the term-to-documents map that search requires.

The structure is small relative to the text it indexes because it stores integers, not words, and because the integers compress extraordinarily well, a property we exploit in Section 5. A query is answered by intersecting or unioning the posting lists of its terms and scoring the resulting candidates. Because each posting list is a sorted run of integers, the intersection of two terms is a linear merge, and a conjunctive query over a handful of terms touches only the documents that actually contain those terms, never the rest of the corpus. That selectivity is why a lexical index can serve a billion-document collection from a query that names three rare words in single-digit milliseconds.

2. Building the Index with MapReduce Intermediate

Constructing an inverted index over a few thousand documents is a tight loop on one machine. Constructing it over a few billion is the canonical MapReduce job, and it was in fact one of the motivating examples for which MapReduce was designed. The shape, shown in Figure 36.4.1, maps the indexing problem onto the same map, shuffle, reduce skeleton from Chapter 6. The map function reads one document, tokenizes it, and emits a key-value record for every token occurrence: the key is the term, the value is the document identifier (carrying the term frequency or position when needed). The framework's shuffle groups every record by key, so all occurrences of a given term, no matter which mapper saw them, arrive together at one reducer. The reduce function receives a term and the stream of document identifiers that contain it, sorts and deduplicates them, and writes the finished posting list.

This is exactly the structure that the chapter's narrative spine keeps returning to. The shuffle that groups $(\text{term}, \text{docID})$ pairs by term is the same shuffle that becomes the all-reduce of distributed training in Chapter 15; here it groups postings rather than gradients, but the cost model and the skew problem are identical. A term like "the" produces a posting list the size of the corpus and lands on a single reducer, the textbook hot-key skew that Chapter 7 teaches you to mitigate by partitioning hot keys or pruning stopwords before the emit. The finished posting lists are written to the distributed storage layer of Chapter 8, where they live as immutable segment files that the query path memory-maps.

3. Two Ways to Shard the Index Intermediate

A single inverted index over a web-scale corpus is far too large for one machine, so it must be partitioned across many. There are exactly two natural ways to cut it, and the choice is the single most consequential decision in the whole index design because it determines what every query does at run time. Figure 36.4.2 contrasts them.

In document-partitioned indexing, the more common choice and the one nearly every production engine defaults to, each shard holds a complete, self-contained inverted index over a disjoint slice of the documents. Adding documents means adding shards; the vocabulary is replicated, but each shard's posting lists cover only its own documents. The cost lands at query time: because any query term might match documents in any shard, every query must be broadcast to all shards, each shard computes its local top-$k$, and a coordinator merges the partial results. The end-to-end latency is therefore governed by the slowest shard, $$ T_{\text{query}} \;=\; T_{\text{coordinate}} \;+\; \max_{1 \le s \le S} T_s \;+\; T_{\text{merge}}, $$ where $S$ is the number of shards and $T_s$ is shard $s$'s local search time. The fan-out makes a single straggler set the tail latency for the whole query, the same straggler problem that Chapter 3 models, which is why production systems hedge by querying a replica when a shard runs slow.

In term-partitioned indexing, each shard owns the complete posting lists for a slice of the vocabulary. A query routes only to the shards that own its terms, so a three-term query touches at most three shards rather than all $S$, which is attractive for very long queries against very many shards. The catch is that joining the posting lists, the intersection that actually answers the query, requires shipping entire posting lists across the network to one node, and a common term's posting list can be gigabytes. Term partitioning also concentrates load on whichever shards own popular terms and makes per-document updates touch many shards at once. For these reasons web search engines overwhelmingly choose document partitioning and pay the fan-out, accepting broad-but-shallow work per shard over the network-heavy joins of the term-partitioned design. This is the same index-sharding trade-off that Chapter 25 revisits for vector indexes, where document partitioning reappears as partitioning the vectors across shards.

4. Scoring with BM25 Intermediate

Finding the documents that contain the query terms is only half of retrieval; ranking them is the other half. The lexical engine ranks with BM25, the probabilistic scoring function that has anchored keyword search for decades and remains a strikingly strong baseline. BM25 scores a document $D$ against a query $Q$ by summing, over the query terms, a saturating term-frequency contribution weighted by how rare the term is across the corpus, $$ \text{BM25}(Q, D) \;=\; \sum_{t \in Q} \text{IDF}(t)\,\cdot\,\frac{f(t, D)\,(k_1 + 1)}{f(t, D) + k_1\,\bigl(1 - b + b\,\frac{|D|}{\text{avgdl}}\bigr)}, $$ where $f(t, D)$ is the frequency of term $t$ in document $D$, $|D|$ is the document length, $\text{avgdl}$ is the average document length, and $k_1$ and $b$ are tuning constants (typically $k_1 \approx 1.5$, $b \approx 0.75$). The inverse document frequency rewards rare terms, $$ \text{IDF}(t) \;=\; \ln\!\left( \frac{N - n(t) + 0.5}{n(t) + 0.5} + 1 \right), $$ with $N$ the number of documents and $n(t)$ the number containing $t$, which is exactly the length of $t$'s posting list. Two properties make BM25 the right fit for a distributed index. The term-frequency term saturates, so a document that repeats a word fifty times does not outscore one that uses it five times by a factor of ten, and the length normalization through $b$ keeps long documents from winning by sheer size.

The distributed subtlety hides in $\text{IDF}$. Under document partitioning, each shard knows $n(t)$ only over its own documents, not over the whole corpus, so a naively local IDF differs from the global one. Production systems either accept the approximation, since per-shard ranking errors mostly wash out in the merge, or maintain a small replicated table of global document frequencies that every shard reads. The runnable demo below makes this concrete: it builds an inverted index over a handful of documents, scores a BM25 query against the global index, then splits the index into two document-partitioned shards and shows that the per-shard scores, computed with local statistics, differ from the global ones yet still surface the right document after the merge.

import math, re
from collections import defaultdict

DOCS = {
    0: "distributed indexing splits the inverted index across many shards",
    1: "an inverted index maps each term to a posting list of documents",
    2: "bm25 ranks documents by term frequency and inverse document frequency",
    3: "vector search complements lexical retrieval for hybrid rag pipelines",
    4: "the mapreduce shuffle groups term postings to build the inverted index",
}

def tokenize(text):
    return re.findall(r"[a-z0-9]+", text.lower())

def build_index(docs):
    """Return postings: term -> {docID: term_freq}, plus per-doc lengths."""
    postings, doc_len = defaultdict(dict), {}
    for did, text in docs.items():
        toks = tokenize(text)
        doc_len[did] = len(toks)
        tf = defaultdict(int)
        for t in toks:
            tf[t] += 1
        for t, f in tf.items():
            postings[t][did] = f                 # term -> (docID, frequency)
    return postings, doc_len

def bm25_scores(query, postings, doc_len, k1=1.5, b=0.75):
    N = len(doc_len)
    avgdl = sum(doc_len.values()) / N
    scores = defaultdict(float)
    for term in tokenize(query):
        plist = postings.get(term)
        if not plist:
            continue
        df = len(plist)                          # n(t) = posting-list length
        idf = math.log((N - df + 0.5) / (df + 0.5) + 1.0)
        for did, f in plist.items():
            denom = f + k1 * (1 - b + b * doc_len[did] / avgdl)
            scores[did] += idf * (f * (k1 + 1)) / denom
    return sorted(scores.items(), key=lambda kv: kv[1], reverse=True)

postings, doc_len = build_index(DOCS)
print("=== Inverted index (term -> posting list) ===")
for term in ["inverted", "index", "bm25", "rag"]:
    print(f"  {term:9s} -> {dict(postings.get(term, {}))}")

query = "inverted index for rag"
print(f"\n=== BM25 ranking for query: '{query}' ===")
for rank, (did, score) in enumerate(bm25_scores(query, postings, doc_len), 1):
    print(f"  #{rank}  doc {did}  score={score:.4f}  | {DOCS[did]}")

# Document-partitioned sharding: each shard owns a disjoint slice of docs
# and builds its OWN local index. A query must fan out to every shard.
shard_docs = [{d: DOCS[d] for d in (0, 1, 2)}, {d: DOCS[d] for d in (3, 4)}]
shard_index = [build_index(sd) for sd in shard_docs]
print("\n=== Document-partitioned sharding (2 shards) ===")
for s, (p, dl) in enumerate(shard_index):
    print(f"  shard {s}: docs {sorted(dl)}  | {len(p)} distinct terms")

print(f"\n=== Fan-out query '{query}' across shards, then merge ===")
merged = []
for s, (p, dl) in enumerate(shard_index):
    local = bm25_scores(query, p, dl)
    if local:
        print(f"  shard {s} local top: doc {local[0][0]} score={local[0][1]:.4f}")
    merged.extend(local)
merged.sort(key=lambda kv: kv[1], reverse=True)
print("  merged top doc:", merged[0][0] if merged else None)

=== Inverted index (term -> posting list) ===
  inverted  -> {0: 1, 1: 1, 4: 1}
  index     -> {0: 1, 1: 1, 4: 1}
  bm25      -> {2: 1}
  rag       -> {3: 1}

=== BM25 ranking for query: 'inverted index for rag' ===
  #1  doc 3  score=2.9276  | vector search complements lexical retrieval for hybrid rag pipelines
  #2  doc 0  score=1.1383  | distributed indexing splits the inverted index across many shards
  #3  doc 4  score=1.0412  | the mapreduce shuffle groups term postings to build the inverted index
  #4  doc 1  score=0.9987  | an inverted index maps each term to a posting list of documents

=== Document-partitioned sharding (2 shards) ===
  shard 0: docs [0, 1, 2]  | 26 distinct terms
  shard 1: docs [3, 4]  | 19 distinct terms

=== Fan-out query 'inverted index for rag' across shards, then merge ===
  shard 0 local top: doc 0 score=0.9980
  shard 1 local top: doc 3 score=1.4516
  merged top doc: 3

from opensearchpy import OpenSearch, helpers

client = OpenSearch("https://localhost:9200")
client.indices.create("corpus", body={               # document-partitioned by default
    "settings": {"number_of_shards": 40, "number_of_replicas": 1},
    "mappings": {"properties": {"text": {"type": "text",
        "similarity": "BM25"}}}})                     # BM25 is the built-in default

helpers.bulk(client, ({"_index": "corpus", "_id": i, "text": t}
                      for i, t in DOCS.items()))      # map+shuffle+reduce, hidden

hits = client.search(index="corpus", body={           # scatter, score, merge top-k
    "query": {"match": {"text": "inverted index for rag"}}})["hits"]["hits"]

5. Compressing the Posting Lists Advanced

A web-scale inverted index holds hundreds of billions of postings, and stored as raw 32-bit integers it would not fit in cluster memory, where it must live for queries to be fast. Compression is therefore not an optimization but a requirement, and posting lists compress beautifully because they are sorted. The standard move is delta encoding: instead of storing the document identifiers $[17, 23, 24, 90, \dots]$, store the gaps between consecutive identifiers $[17, 6, 1, 66, \dots]$. Gaps are smaller than absolute identifiers, especially for common terms whose postings are dense, and small integers encode in few bits. A frequent term that appears in every tenth document has gaps clustered around ten, which a variable-length code stores in a handful of bits each rather than a full word.

The gaps are then packed with a variable-length integer code. Classic schemes include variable-byte encoding and the Elias gamma and delta codes; modern engines favor SIMD-friendly block codes such as PForDelta and the Roaring bitmap layout, which decode billions of integers per second by processing whole blocks at once. To bound the worst case, note that a posting list for a term appearing in $n(t)$ of $N$ documents, delta-encoded, needs on the order of $$ \text{bits}(t) \;\approx\; n(t)\,\Bigl\lceil \log_2\!\frac{N}{n(t)} \Bigr\rceil $$ bits, a quantity that grows with how common the term is yet shrinks per posting as the gaps tighten. Real indexes reach well under one byte per posting. Compression also interacts with the query path: block codes let the engine skip whole blocks of a posting list during an intersection without decoding them, turning a long merge into a series of jumps, the skip-list optimization that keeps conjunctive queries fast. The compressed segments are the immutable files that the storage layer of Chapter 8 memory-maps into each shard's address space.

6. Hybrid Retrieval: Lexical Meets Dense Intermediate

The lexical index built in this section and the dense vector index built in Section 36.5 have complementary failure modes. BM25 is precise on exact terms, rare identifiers, and proper nouns but is blind to synonymy: a query for "automobile" misses a document that only ever says "car". Dense retrieval matches meaning and bridges that gap but can drift on rare tokens and exact strings that no embedding has learned. A hybrid retriever runs both and fuses their results, and the fusion is what feeds the RAG generator its context. The simplest robust fusion is reciprocal rank fusion, which combines the two ranked lists by summing a rank-based score, $$ \text{RRF}(d) \;=\; \sum_{r \in \{\text{lex}, \text{dense}\}} \frac{1}{k + \text{rank}_r(d)}, $$ where $\text{rank}_r(d)$ is document $d$'s position in retriever $r$'s ranking and $k$ is a small constant (commonly 60) that damps the influence of low ranks. Because RRF uses ranks rather than raw scores, it sidesteps the awkward problem that BM25 scores and cosine similarities live on incomparable scales, the same reason the local-versus-global IDF mismatch in Output 36.4.1 stops mattering once results are ranked.

Architecturally the two indexes mirror each other: both are document-partitioned across shards, both fan a query out to all shards, and both merge per-shard top-$k$ lists at a coordinator, which is why a single retrieval service can host both and fuse them at the merge step. The lexical half owns recall on exact matches; the dense half owns recall on meaning; the generator downstream sees one fused, deduplicated context window. With this section the chapter has built the lexical retriever; Section 36.5 builds its dense twin and Section 36.6 shards it for billion-vector search, after which the two streams join.