Everyone builds search the same way the first time: put the documents in a database, and run SELECT * FROM docs WHERE body LIKE '%term%'. It works on a thousand rows. Then the interviewer makes it the real problem: billions of documents, a query that matches a few keywords has to come back ranked by relevance in under 50ms, documents must be searchable within a second of being written, and the whole thing has to survive machines dying without losing data or going dark.

LIKE '%term%' cannot do any of that. It is a full scan - it reads every row for every query, it has no notion of relevance, and it falls over at a few million rows. A full-text search engine inverts the problem completely. Instead of “for this query, scan every document,” it precomputes “for this term, here is the exact list of documents that contain it.” That one data structure - the inverted index - is the whole game. Get the inverted index right (how it is built, sharded, scored, kept fresh, and replicated) and everything else is routing and plumbing. Get it wrong and you either scan forever, return garbage rankings, or lose data when a node dies.

Let me do it properly.

Functional Requirements (FR)

In scope:

  • Index documents. Given a document (an id plus a set of fields - title, body, tags, timestamps, numbers), tokenize the text and add it to the index so it becomes searchable. Documents arrive continuously and must be searchable quickly (near-real-time, not a nightly batch).
  • Full-text query. Given a query string of one or more terms, return the documents that match, ranked by relevance (the most relevant first), with pagination. Support the common query shapes: boolean AND/OR, phrase (“exact sequence”), and prefix.
  • Relevance ranking. Results are not just “matches” - they are ordered so the best match is on top. A term that is rare across the corpus but frequent in a document should rank that document higher. This scoring (BM25-class) is mandatory, not optional.
  • Filtering and faceting. Combine the full-text match with structured filters (price < 1000, category = "shoes", created_at > last week) and return facet counts (how many results per category). Search is rarely pure text.
  • Updates and deletes. Documents change and get deleted; the index must reflect that. Reindex on update, remove on delete, without rebuilding the world.
  • Near-real-time. A document written now is searchable within ~1 second. Not instant (that is expensive), not an hour (that is useless).

Explicitly out of scope (say this to control scope):

  • Web crawling / data ingestion source. We index documents handed to us via an API. Where they come from (a crawler, a database CDC stream, app writes) is upstream.
  • Semantic / vector search. Dense-vector embedding search (ANN over embeddings) is a different index type (HNSW/IVF). We build the classic lexical inverted index; we note where a vector tier bolts on.
  • Query understanding / NLP. Spell correction, synonym expansion, and intent classification are query-rewrite layers that sit in front of the engine. We take the analyzed query terms as the contract.
  • Learning-to-rank / ML reranking. We do BM25-class lexical scoring. A learned reranker (a model that re-scores the top-K) is a downstream stage we mention but do not build.

The one decision that drives everything: a search engine is a read-optimized system built around a precomputed inverted index, where the scarce resource is query latency over a sharded index, not write throughput. You pay a lot at write time (tokenize, invert, merge segments) to make reads cheap (look up a term, intersect posting lists, score top-K). So the design is dominated by the inverted index - how it is structured, sharded across machines, scored, kept near-real-time fresh, and replicated for fault tolerance.

Non-Functional Requirements (NFR)

  • Scale: target ~10B documents, average ~1 KB of indexable text each. Read-heavy: on the order of 100K queries/sec at peak, with write/index throughput of ~50K-100K documents/sec (continuous ingestion plus updates). The index itself is hundreds of terabytes and must live across hundreds of nodes.
  • Latency (the hard target): p99 query latency < 50ms for a typical multi-term query over the full corpus, including scatter-gather across shards and ranking. This is the number the whole design serves. Index latency (write to searchable) target < 1s.
  • Availability: 99.9%+ for the query path. Search going down is a visible, revenue-affecting outage. The system must keep serving reads even when individual nodes fail, which forces replication.
  • Consistency: search is eventually consistent and that is acceptable. A document being searchable ~1s after write, or a just-deleted doc showing up for a moment, is fine. We do not promise read-your-writes on the search path. The source of truth lives elsewhere; the index is a derived, rebuildable view - but losing it means an expensive full reindex, so we still replicate it.
  • Durability: the index is derived and could be rebuilt from the source of truth, but rebuilding 100s of TB takes hours to days, so we treat the index as needing high durability - a translog (write-ahead log) per shard plus replica copies, so a single node loss never forces a reindex and never loses an acknowledged write.
  • Freshness vs cost: near-real-time (~1s) refresh is a deliberate cost trade. True real-time (every write instantly visible) would mean no batching and crippling write amplification. ~1s refresh batches writes into segments cheaply.

Back-of-the-Envelope Estimation (BoE)

Let me ground the design in numbers before drawing boxes.

Corpus and index size:

Documents:                 10,000,000,000  (10B)
Avg indexable text/doc:    ~1 KB
Raw text corpus         = 10B * 1 KB = 10 TB of raw text.

The inverted index is NOT the raw text. It stores, per term,
a posting list of doc ids (+ positions + frequencies).
Rule of thumb: a positional inverted index is ~30-50% of the
raw text size after compression (delta + variable-byte encoding).

Inverted index size  ≈ 10 TB * 0.4 ≈ 4 TB  (postings)
Plus stored fields / doc values (for filtering, sorting,
  returning source): another ~10-20 TB.
Total on-disk footprint: ~15-25 TB per replica copy.
With 1 primary + 2 replicas: ~45-75 TB of storage total.

Sharding math (how many shards/nodes):

Keep each shard to a manageable size for fast queries and fast
recovery: target ~30-50 GB of index per shard.

Index postings ~4 TB / 40 GB per shard ≈ 100 primary shards.
With full stored fields (~20 TB) at 40 GB/shard ≈ 500 shards.
Call it ~200-500 primary shards, each replicated x2.

Nodes: if a node holds ~5-10 shards (primary+replica mix) at
~40 GB each + RAM for caches, you need on the order of
100-300 nodes. The exact split depends on query fan-out cost.

Query load (reads):

Peak queries/sec:          100,000 QPS
Each query scatters to ALL primary-or-replica shards it must
  touch. For a corpus-wide text query that is every shard group.
Fan-out factor:            ~200 shard-queries per user query
  (one per shard group, hitting one replica of each).

Effective internal shard-queries/sec = 100K * 200 = 20,000,000
  shard-level query executions per second across the cluster.
Per node (say 200 nodes): ~100K shard-queries/sec/node.

This is why per-shard query execution must be microsecond-to-
low-millisecond: a single user query's 50ms budget is spent
across a 200-way scatter-gather, and the slowest shard sets p99.

Write/index load:

Index rate:                ~50,000 docs/sec sustained
Each doc: tokenize ~1 KB (~150 terms), update posting lists.
Writes are batched into in-memory buffers, flushed to immutable
  segments every ~1s (the refresh interval).

Translog (WAL) write:      50K docs/sec * ~1 KB = ~50 MB/sec
  durable append per the whole cluster, spread across shards.
Segment merges run continuously in the background (the real
  write-amplification cost; covered in the deep dive).

Cache sizing:

Two caches matter most:
1. Filesystem/page cache for hot postings + doc values.
   Aim to keep the "hot" working set of the index in RAM.
   If 20% of 4 TB postings is hot: ~800 GB of RAM across the
   cluster dedicated to page cache -> ~4-8 GB/node just for that.
2. Query result / request cache for popular identical queries
   (filters especially). A few GB/node, high hit rate on
   head queries (the top ~1% of queries are a huge share of traffic).

The takeaway: this is a read-amplified, fan-out system. One user query becomes hundreds of parallel shard executions, and the 50ms p99 is a budget split across all of them plus the gather-and-merge. Every design choice - shard count, replica placement, scoring cost, caching - exists to keep that fan-out fast and the slowest shard fast.

High-Level Design (HLD)

A search engine is two paths that meet at the inverted index: an index path (documents in, inverted index built and kept fresh) and a query path (query in, scatter to shards, score, gather, rank, return). The index is partitioned into shards (for scale) and each shard has replicas (for availability and read throughput). A coordinator routes queries and documents; it holds no index data itself.

        WRITE PATH                                 READ PATH
   ┌──────────────────┐                      ┌──────────────────┐
   │  Index API       │                      │   Query API      │
   │ (docs in)        │                      │ (search in)      │
   └────────┬─────────┘                      └────────┬─────────┘
            │                                          │
   ┌────────▼───────────────────────────────────────────▼─────────┐
   │                    COORDINATOR NODE                            │
   │  - route doc to shard = hash(doc_id) % N  (write)              │
   │  - analyze query, scatter to one replica of EACH shard (read) │
   │  - gather partial top-K, merge, re-rank, paginate (read)      │
   └───────┬───────────────────────────────────────────┬──────────┘
           │ index doc                        scatter   │ query
           │                                  (fan-out)  │
  ┌────────▼───────────┐   ┌───────────────┐   ┌────────▼─────────┐
  │   SHARD 1 group    │   │  SHARD 2 grp  │...│  SHARD N group   │
  │ ┌───────┐┌───────┐ │   │               │   │                  │
  │ │PRIMARY││REPLICA│ │   │  PRIMARY      │   │  PRIMARY         │
  │ │       ││  x2   │ │   │  REPLICA x2   │   │  REPLICA x2      │
  │ └───┬───┘└───────┘ │   └───────────────┘   └──────────────────┘
  └─────┼──────────────┘
        │ inside one shard (a Lucene-style index):
   ┌────▼──────────────────────────────────────────────┐
   │  In-memory buffer ──(refresh ~1s)──► Segment (immutable)
   │  Translog (WAL, durable)                           │
   │  Segments: seg_0 seg_1 seg_2 ... ──(merge)──► seg_big
   │  Each segment = its own inverted index + doc values│
   └────────────────────────────────────────────────────┘

   ┌──────────────────────────────────────────────────────────────┐
   │ Cluster control plane (master-eligible nodes, Raft/Zen-style):│
   │  shard allocation, replica placement, node health, failover   │
   └──────────────────────────────────────────────────────────────┘

Index path - one document, end to end:

  1. A document hits the Index API and reaches a coordinator. The coordinator computes the target shard: shard = hash(doc_id) % num_primary_shards (document routing - covered in the deep dive).
  2. The write goes to that shard’s primary. The primary first appends the operation to its translog (a durable write-ahead log) so the write survives a crash, then adds the document to an in-memory buffer (an indexing buffer that is not yet searchable).
  3. The primary forwards the operation to its replicas in parallel. Once the required copies ack (primary + enough replicas), the write is acknowledged to the client. This is what makes an acked write durable and highly available.
  4. On the refresh interval (~1s), the in-memory buffer is flushed into a new immutable segment - a small, self-contained inverted index. The moment the segment is opened for search, its documents become searchable. This ~1s batching is the “near” in near-real-time.
  5. In the background, many small segments are merged into fewer, larger segments (segment merging - the write-amplification cost), and the translog is periodically flushed/truncated once its operations are safely in segments (a flush/commit).

Query path - one search, end to end:

  1. A query hits the Query API and a coordinator. The coordinator analyzes the query (same tokenizer/analyzer used at index time: lowercase, stem, remove stopwords) into terms, and parses filters.
  2. The coordinator scatters the query to one replica of every shard group (it only needs one copy of each shard’s data). This is the fan-out - a corpus-wide query touches every shard.
  3. Each shard, in parallel, runs the query against its local inverted index: look up the posting list for each term, intersect/union them (boolean logic), apply filters, score each matching doc with BM25, and return its local top-K (just doc ids + scores, not full documents). This is the query phase.
  4. The coordinator gathers the partial top-K lists from all shards and merges them into a global top-K by score. Now it knows the final K doc ids.
  5. The coordinator issues a fetch phase: ask the owning shards for the full stored source of just those K winning docs (and any highlighting). This two-phase “query then fetch” avoids shipping full documents from every shard - only the final winners are materialized.
  6. The coordinator assembles the ranked page and returns it. Facets/aggregations are computed per-shard and combined in the same gather step.

Failure handling: if a shard replica is slow or down during scatter, the coordinator routes that shard’s query to another replica (adaptive replica selection); if a primary dies, the control plane promotes a replica to primary (failover). Because the translog + replicas hold every acked write, no acknowledged data is lost.

The architectural insight: the inverted index is partitioned for scale and replicated for speed and safety, and every query is a parallel scatter-gather whose latency is set by the slowest shard. Coordinators are stateless routers; the intelligence (scoring, posting-list intersection, near-real-time segments) lives inside each shard.

Component Deep Dive

Four hard parts: (1) the inverted index itself and how a query executes against it fast, (2) sharding and the scatter-gather that keeps corpus-wide queries under 50ms, (3) near-real-time ingestion via immutable segments, merging, and the translog, and (4) relevance scoring (BM25) in a distributed index. I walk each from naive to scalable.

1. The Inverted Index and Fast Query Execution

The inverted index answers “which documents contain this term?” instantly, instead of scanning documents to ask “does this doc contain this term?” Everything hinges on it.

Approach A: Scan the documents (the naive instinct)

Store documents, and for each query scan them checking for the terms (LIKE '%term%', or a forward index of doc -> terms).

query "distributed systems":
  for each of 10,000,000,000 docs:
      if "distributed" in doc and "systems" in doc: match

Where it breaks - immediately and fatally: this is O(corpus) per query. At 10B docs you read the entire 10 TB corpus for every single query, and you run 100K queries/sec. That is physically impossible - the disk and CPU cost is thousands of times your budget. A forward index (doc -> its terms) does not help a term lookup; you still visit every doc. The access pattern is backwards.

Approach B: The inverted index

Precompute the mapping the other way: term -> sorted list of doc ids that contain it (the posting list). Built once at index time, read cheaply at query time.

TERM DICTIONARY            POSTING LISTS (sorted doc ids + data)
  "distributed"  ───────►  [ (doc 5, tf=3, pos=[2,40,77]),
                             (doc 12, tf=1, pos=[9]),
                             (doc 88, tf=2, pos=[1,55]), ... ]
  "systems"      ───────►  [ (doc 5, tf=2, pos=[3,78]),
                             (doc 41, tf=1, pos=[12]),
                             (doc 88, tf=1, pos=[56]), ... ]
  "scale"        ───────►  [ (doc 12, ...), (doc 90, ...), ... ]
  • Term dictionary: the sorted set of all unique terms, with a pointer to each term’s posting list. Stored compactly (an FST - finite state transducer - gives prefix-compressed, fast term lookup and enables prefix/wildcard queries).
  • Posting list: for each term, the sorted list of doc ids containing it, plus per-doc term frequency (tf, how many times the term appears - needed for scoring) and optionally positions (for phrase queries like “distributed systems” where order matters).
  • Query execution for distributed AND systems: fetch both posting lists (already sorted by doc id) and intersect them with a merge-style walk - advance the two cursors in lockstep, emit doc ids present in both. Intersecting two sorted lists is O(len of shorter list) with skip pointers, not O(corpus). OR is a union; NOT is a difference; phrase match checks that positions are adjacent.

This is the whole trick: a query touches only the posting lists of its terms, which are a tiny fraction of the corpus, and intersects pre-sorted lists linearly.

Making it fast - the details that matter at 50ms:

  • Compression. Doc ids in a posting list are sorted, so store deltas (gaps) not absolute ids, then variable-byte or Frame-of-Reference encode them. Deltas are small integers that compress hugely, shrinking the index (the 4 TB number) and, more importantly, letting more of it sit in RAM/page cache so intersection avoids disk.
  • Skip lists / skip pointers. To intersect a short list with a very long one, you do not want to walk every entry of the long list. Skip pointers let you jump ahead to the next doc id >= target, turning intersection into something closer to O(shorter list * log).
  • Block-max WAND (the key latency trick). You rarely need all matches - you need the top-K by score. Block-max WAND stores the max possible score per block of a posting list, so the query can skip entire blocks that cannot beat the current K-th best score. This is what lets a term matching 100M docs return the top 10 without scoring all 100M - it is the difference between “score everything” and “score only what can win.”

Approach C: What breaks next - one index is too big for one machine

A single inverted index over 10B docs is ~20 TB and cannot live on one node, nor could one node serve 100K QPS against it. That forces partitioning - which is deep dive #2. The inverted index is correct; it just has to be cut into shards.

Index approachPer-query costRanked?Scales to 10BVerdict
Scan docs / forward indexO(corpus)nonoRejected
Inverted index (1 machine)O(posting lists)yes (with tf)no (too big)Right structure, wrong scale
Sharded inverted index + block-max WANDO(posting) / N shardsyesyesThe design

2. Sharding and Scatter-Gather Under 50ms

The index does not fit on one node, so we cut it into shards, each a complete, independent inverted index over a subset of documents. The hard question is how to partition and how a query reassembles results without blowing the latency budget.

Partitioning: document-partitioned vs term-partitioned

There are two ways to split an inverted index, and the choice is fundamental.

  • Term partitioning (partition by term): shard A holds the full posting lists for terms a-m, shard B for n-z. A single-term query hits one shard - great. But a multi-term query distributed AND systems needs posting lists from different shards, so you must ship a giant posting list across the network to intersect, or do a distributed join. Posting lists for common terms are enormous (hundreds of MB). This creates massive network traffic and brutal hot spots (the shard owning a popular term is swamped). Rejected for general full-text search.
  • Document partitioning (partition by document): each shard holds a complete inverted index over its own slice of documents. shard = hash(doc_id) % N. A query runs fully and independently on every shard (each has all terms for its docs), and results merge. This is what Elasticsearch/Solr do. Downside: every query must fan out to all shards (because any shard might hold a matching doc). We accept the fan-out because each shard’s work is small and parallel, and it avoids shipping posting lists.

Document partitioning wins because it keeps each shard’s query local and self-contained - the expensive part (intersecting posting lists) never crosses the network.

Document-partitioned (chosen):
  Doc 5  -> shard 2    Each shard = full inverted index over ITS docs.
  Doc 6  -> shard 0    Query runs completely on every shard in parallel,
  Doc 7  -> shard 1    returns local top-K. Coordinator merges.
  shard = hash(doc_id) % num_shards

The scatter-gather, and why the slowest shard owns your p99

A corpus-wide query fans out to all N shard groups (one replica each), each returns its local top-K, and the coordinator merges. The total latency is max(shard latencies) + merge. With a 200-way fan-out, p99 is governed by the slowest of 200 parallel requests - the classic tail-latency-amplification problem. Fixes:

  • Keep shards small and numerous enough to parallelize, but not so many that fan-out overhead dominates. More shards = each shard faster but more fan-out coordination; fewer shards = less coordination but each shard slower. The ~30-50 GB/shard target balances this. Over-sharding (thousands of tiny shards) makes the coordinator’s gather the bottleneck.
  • Adaptive replica selection. Each shard has multiple replicas; the coordinator sends each shard’s sub-query to the replica that is currently fastest (lowest observed latency / least loaded), not round-robin. This dodges a slow node dragging the whole query.
  • Request-level caching and query short-circuiting. Cache results of identical filter/query combinations (especially filters, which are boolean and highly cacheable). Block-max WAND (deep dive #1) short-circuits each shard’s work so it only scores docs that can make the top-K.
  • Routing to prune fan-out when possible. If documents are routed by a known key (e.g. all of a tenant’s docs on one shard via a custom routing value), a tenant-scoped query can hit one shard instead of all N. When the query carries such a routing key, fan-out collapses. This is a major optimization for multi-tenant search.

Two-phase query-then-fetch (do not ship full docs from every shard)

If every shard returned full documents for its top-K, a 200-way fan-out with K=10 would ship 2000 full documents when you only need 10. Instead:

  • Query phase: each shard returns only (doc_id, score) for its local top-K. Cheap to ship.
  • Coordinator merges to the global top-K doc ids.
  • Fetch phase: the coordinator asks only the owning shards for the full _source of the final K winners (plus highlighting). Only 10 full docs move.

This two-phase split is why deep pagination is expensive (page 1000 means every shard must return its top 10,000 so the coordinator can find the right slice - search_after/cursor pagination fixes this by carrying the last sort value instead of an offset).

3. Near-Real-Time Ingestion: Immutable Segments, Merging, and the Translog

Documents arrive at 50K/sec and must be searchable within ~1s, without rebuilding the index and without losing acked writes on a crash. The naive approaches all fail; the solution is immutable segments plus a write-ahead log.

Approach A: Mutate the inverted index in place (naive)

When a doc arrives, insert its terms into the existing posting lists directly.

Where it breaks: posting lists are compressed, sorted, and packed for read speed. Inserting a doc id into the middle of a compressed, sorted list means decompressing, inserting, re-compressing, and rewriting - for every term in the doc, on every write. Concurrent reads would need locking against these in-place mutations, killing query throughput. A read-optimized compressed structure is hostile to in-place writes. This is a non-starter at 50K writes/sec.

Approach B: Immutable segments + refresh (the core idea)

Never mutate. Buffer new docs in memory, and periodically flush the buffer into a brand-new, immutable mini-index called a segment.

new docs ─► in-memory indexing buffer
                  │  every ~1s (refresh)
                  ▼
            new immutable SEGMENT (its own tiny inverted index)
                  │  opened for search immediately
                  ▼
   searchable index = union of ALL segments: seg0 ∪ seg1 ∪ seg2 ...
   A query runs against every segment and merges results.
  • Immutability buys concurrency. A segment, once written, never changes. Readers need no locks - they just open a consistent set of segments. Writers only ever append a new segment. This decouples readers from writers completely, which is how you sustain 100K QPS and 50K writes/sec at once.
  • Refresh = near-real-time. Every ~1s the buffer becomes a new searchable segment. That ~1s interval is the freshness/cost knob: smaller interval = fresher but more tiny segments and more overhead; larger = cheaper but staler. ~1s is the usual sweet spot. A doc is searchable as soon as the next refresh opens its segment.
  • Updates and deletes without mutation. You cannot edit a segment, so a delete just records the doc id in a per-segment “deleted” bitmap (a tombstone); queries filter out tombstoned docs, and the space is reclaimed later during merge. An update is a delete of the old doc + an index of the new version (into a new segment). Deletes are logical until a merge physically drops them.

The problem Approach B creates: segment explosion

A new segment every second means thousands of tiny segments pile up. A query has to execute against every segment and merge, so query cost grows with segment count - search slowly grinds to a halt. Too many open files, too much per-segment overhead.

Approach B-fixed: background segment merging

A background process continuously merges many small segments into fewer large ones (a tiered merge policy: merge segments of similar size).

seg0 seg1 seg2 seg3 (small)  ──merge──►  seg_A (medium)
seg_A seg_B seg_C            ──merge──►  seg_BIG (large)
During merge: deleted docs (tombstones) are physically dropped,
posting lists re-compressed, the merged segment swapped in,
old segments deleted. Reads keep using old segments until swap.
  • Merging is the real write-amplification cost. Data gets rewritten multiple times as it climbs the size tiers. This CPU and I/O is the price of keeping segment count low for fast reads. Merge throttling keeps it from starving query traffic.
  • Merges reclaim deletes. Tombstoned docs vanish during merge, which is when deletes actually free space.

Durability: the translog (write-ahead log)

A refresh makes docs searchable, but segments only hit durable storage on a periodic flush (commit). A crash between flushes would lose buffered docs. So every write is first appended to a per-shard translog (a durable WAL) before being buffered:

  • Write order: append to translog (fsync per request or per interval) -> add to in-memory buffer -> ack. If the node crashes, on restart it replays the translog to rebuild the un-flushed buffer. No acked write is lost even though it was not yet in a committed segment.
  • Flush/commit: periodically, the buffered segments are fsynced to disk (a Lucene commit) and the translog is truncated up to that point. Flush is expensive, so it is infrequent relative to refresh. Refresh (searchability, ~1s, cheap) and flush (durability, ~30min or by translog size, expensive) are deliberately decoupled.

So three separate clocks run per shard: refresh (visibility, ~1s), merge (keep segment count low, continuous background), and flush (durability commit + translog truncation, periodic). Confusing them is the classic mistake.

4. Relevance Scoring (BM25) in a Distributed Index

Matching is binary; ranking is the product. Users want the best matches first, and “best” means a relevance score. Getting scoring right - and getting it right across shards - is subtle.

Approach A: Count matching terms (naive)

Rank by how many query terms a doc contains, or raw term frequency.

Where it breaks: raw counts are dominated by (a) common words - a doc with “the” 50 times beats a doc with the rare, meaningful query term once - and (b) long documents, which contain more of everything. Ranking by raw frequency surfaces long, keyword-stuffed junk. You need to weight by how informative a term is and normalize for document length.

Approach B: TF-IDF, then BM25 (the standard)

The insight has two halves:

  • TF (term frequency): a term appearing more often in a document makes that document more relevant to it - but with diminishing returns (the 10th occurrence adds less than the 2nd).
  • IDF (inverse document frequency): a term that is rare across the whole corpus is more discriminating than a common one. “distributed” appearing in 0.1% of docs is a strong signal; “the” appearing in 99% of docs is nearly worthless. IDF weights rare terms up and common terms down.

BM25 is the refined, industry-standard scoring function combining these with saturation and length normalization:

score(doc, query) = Σ  IDF(term) *  ( tf * (k1 + 1) )
                   terms            ---------------------------------
                                    tf + k1 * (1 - b + b * (dl / avgdl))

  tf    = term frequency of the term in the doc
  dl    = length of the doc;  avgdl = average doc length in the corpus
  IDF   = log( (N - n + 0.5) / (n + 0.5) + 1 )
          N = total docs, n = docs containing the term
  k1 (~1.2) = tf saturation knob (caps the benefit of repetition)
  b  (~0.75) = length normalization strength
  • k1 saturates tf: so the 50th “the” does not dominate - repetition has diminishing returns.
  • b normalizes length: so a long document is not unfairly favored just for being long.
  • IDF is computed from corpus-wide stats (N total docs and n = docs containing the term). This is the distributed wrinkle.

The distributed scoring problem: IDF is a global statistic

Here is the trap. BM25 needs N (total docs) and n (docs containing the term) - global numbers. But each shard only knows its own document frequencies. If every shard computes IDF from its local stats, the same term gets a different IDF on different shards, so scores are not comparable across shards, and the merged ranking is subtly wrong (a doc can outrank a better one just because it lives on a shard where the term looked rarer).

Fixes, in order of cost/accuracy:

  • Accept local IDF (default, usually fine). With random hash(doc_id) document distribution and reasonably large shards, term statistics are roughly uniform across shards, so local IDF approximates global IDF well enough. This is what engines do by default - it is cheap and the error is tiny at scale. The randomness of the shard key is what makes this acceptable.
  • Distributed frequency search (dfs_query_then_fetch). For correctness-sensitive cases, add a pre-phase: the coordinator first asks all shards for their local n and N per query term, sums them into true global stats, then sends those global stats with the scatter so every shard scores with identical, correct IDF. Costs one extra round trip - buy it only when ranking precision matters (small indices, skewed distributions).
  • Score, then rerank. Use BM25 to get a good top-K cheaply across shards, then apply a more expensive learned reranker (a model using more features) on just the merged top few hundred at the coordinator. Lexical scoring for recall + a learned model for final precision is the modern production shape (we noted the reranker is out of scope to build, but this is where it attaches).

Scoring also composes with filters: a filter clause (e.g. category = shoes) is a boolean yes/no that does not contribute to the score and is highly cacheable, while the query clause (the text match) is what gets scored. Keeping filters out of scoring makes them cacheable bitsets and keeps scoring focused on relevance.

API Design & Data Schema

A search engine exposes a document (index) API, a search (query) API, and an admin API. Internally it is a set of specialized stores.

Index / document API

PUT  /index/{index}/_doc/{id}          # index or replace a document
     { "title": "...", "body": "...", "tags": [...],
       "price": 999, "created_at": "2026-11-14T..." }
     -> routed to shard = hash(id) % num_shards, written to primary+replicas
     -> { "result": "created", "_shard": 7, "_version": 1 }

POST /index/{index}/_bulk              # batch index (the real ingestion path)
     ndjson: {action}\n{doc}\n{action}\n{doc}...   # thousands/request

DELETE /index/{index}/_doc/{id}        # tombstone the doc (logical delete)

POST /index/{index}/_refresh           # force a refresh (make buffered docs
                                       # searchable now) - rarely needed

Search / query API

POST /index/{index}/_search
  {
    "query": {                         # the scored full-text part
      "match": { "body": "distributed systems" }
    },
    "filter": [                        # unscored, cacheable constraints
      { "range": { "price": { "lt": 1000 } } },
      { "term":  { "category": "shoes" } }
    ],
    "sort": [ "_score", { "created_at": "desc" } ],
    "from": 0, "size": 10,             # pagination (use search_after for deep)
    "aggs": {                          # facets
      "by_category": { "terms": { "field": "category" } }
    },
    "search_after": [ 1234.5, "doc_988" ]   # cursor for deep pagination
  }
  -> {
       "took_ms": 23,
       "total": 48213,
       "hits": [ { "_id": "...", "_score": 14.2, "_source": {...} }, ... ],
       "aggregations": { "by_category": { "buckets": [...] } }
     }

Admin API

PUT  /index/{index}                    # create index: shard count, replicas,
     { "settings": { "number_of_shards": 200,
                     "number_of_replicas": 2,
                     "refresh_interval": "1s" },
       "mappings": { "body": {"type":"text","analyzer":"english"},
                     "price": {"type":"integer"} } }

GET  /_cluster/health                  # shard allocation, node status
POST /index/{index}/_forcemerge        # force segment merge (off-peak)

Data model - specialized stores, not a single DB

The index is not stored in a relational database. It is a set of structures each tuned for one access pattern. SQL/NoSQL is the wrong framing here - a search engine is its own storage engine (Lucene-class), and it sits alongside a separate source-of-truth datastore.

1. Per-shard inverted index (the term dictionary + posting lists):

Term dictionary (FST, sorted terms -> posting pointer):
  term            STRING   (sorted, prefix-compressed in an FST)
  doc_freq        INT      # n: how many docs contain the term (for IDF)
  posting_ptr     OFFSET   # where this term's posting list lives

Posting list (per term, on disk, compressed):
  [ delta-encoded doc_ids | term_freq per doc | positions (optional)
    | skip pointers | block-max scores ]
  - doc_ids: delta + variable-byte encoded, sorted ascending
  - stored once per segment; a shard = many segments

2. Doc values (columnar, per field - for sort/filter/aggregate):

doc_values   (column-oriented, per field, per segment)
  doc_id -> field_value     # e.g. price, created_at, category
  - stored column-wise so sorting/filtering/faceting on a field
    reads one contiguous column, not full documents
  - this is what makes range filters and aggregations fast

3. Stored fields / _source (row store, for the fetch phase):

stored_fields
  doc_id -> compressed original JSON (_source)
  - read ONLY in the fetch phase for the final top-K winners
  - compressed in blocks; not touched during the query/score phase

4. Translog (per-shard write-ahead log, durability):

translog
  seq_no       INT64    # monotonically increasing per shard
  operation    ENUM     # index | delete
  doc_id       STRING
  payload      BLOB     # the doc (for replay)
  - append-only, fsynced; replayed on crash recovery
  - truncated up to the last flush/commit point

5. Cluster metadata (on master-eligible nodes, Raft-replicated):

cluster_state
  index -> { num_shards, num_replicas, mappings, settings }
  shard -> { primary_node, [replica_nodes], state }   # routing table
  nodes -> { id, address, roles, health }
  - small, strongly consistent (Raft/Zen), the source of truth for
    WHERE each shard lives; consulted to route reads/writes

Why these choices: posting lists are a read-optimized compressed inverted structure (the search engine’s own format), not rows - a relational DB cannot do top-K term intersection at this speed. Doc values are columnar because sort/filter/aggregate touch one field across many docs. Stored fields are a row store read only for final winners. The translog is a WAL for durability. Cluster metadata is small and strongly consistent, so it lives in a Raft-backed store separate from the (eventually consistent, huge) index data. The source of truth for the documents themselves lives in a separate primary datastore (a database); the search index is a derived, rebuildable projection of it fed by bulk-index or CDC.

Bottlenecks & Scaling

Where it breaks first, in order, and the fix for each:

1. Full scan per query (breaks before anything else). LIKE/forward scan is O(corpus) per query and dies at a few million docs. Fix: the inverted index - term -> sorted posting list - turns a query into intersecting a few short sorted lists, plus block-max WAND to return top-K without scoring every match. This is the foundational fix everything else builds on.

2. Index too big for one node. 10B docs = ~20 TB; one machine cannot hold or serve it. Fix: document-partitioned sharding (hash(doc_id) % N), each shard a complete independent inverted index. Scales linearly by adding shards/nodes. Term partitioning is rejected because multi-term queries would ship giant posting lists across the network.

3. Tail latency from scatter-gather. A 200-way fan-out makes p99 the max of 200 parallel shard queries; one slow shard ruins the query. Fix: adaptive replica selection (send each shard’s sub-query to its fastest replica), right-sized shards (~30-50 GB, enough parallelism without fan-out overhead), query-then-fetch (ship only doc_id+score in the query phase, materialize only final K), and routing keys to collapse fan-out to one shard when the query is tenant-scoped.

4. In-place index mutation (write path). Editing compressed posting lists per write is impossible at 50K writes/sec and blocks reads. Fix: immutable segments + refresh - buffer writes, flush to new immutable segments every ~1s (near-real-time, lock-free reads), with deletes as tombstones and updates as delete+reindex.

5. Segment explosion. A new segment per second means thousands of segments and queries that must hit all of them. Fix: background tiered merging into fewer large segments (also physically drops tombstoned deletes). Throttle merges so they do not starve queries. This is the deliberate write-amplification cost paid to keep reads fast.

6. Durability vs a crash between flushes. Refresh makes docs searchable but not yet committed to disk; a crash loses them. Fix: a per-shard translog (WAL) fsynced on each write and replayed on restart; periodic flush/commit fsyncs segments and truncates the translog. Refresh, merge, and flush run on independent schedules.

7. Node failure / availability. A node holding a primary dies; without copies, that shard’s data and writes are gone and queries over it fail. Fix: replica shards - each primary has N replicas on different nodes. Writes go primary -> replicas and ack after enough copies; reads use any replica. On primary loss, the control plane promotes a replica to primary (failover). Translog + replicas mean no acked write is lost. This is the fault-tolerant segment replication the problem centers on.

8. Hot shards / skew. A celebrity tenant or a viral document set overloads one shard; or deep pagination (from=100000) forces every shard to return 100K hits. Fix: random hash(doc_id) routing spreads load evenly; for known skew, use custom routing or split the hot index. Replace offset pagination with search_after (cursor on the last sort value) so deep pages do not force every shard to over-return.

9. Wrong rankings across shards (global IDF). Local per-shard IDF makes scores incomparable across shards. Fix: rely on random doc distribution making local IDF ≈ global IDF (default, cheap), or run dfs_query_then_fetch to collect true global term stats in a pre-phase when precision matters. Optionally rerank the merged top-K with a learned model.

10. Head-query load and repeated filters. The top 1% of queries and common filters repeat constantly. Fix: request cache for identical query/filter combos, filter caches (cacheable bitsets since filters are unscored boolean), and the OS page cache keeping hot posting lists and doc values in RAM. Size RAM so the hot working set never touches disk.

Wrap-Up

The trade-offs that define this design:

  • Pay at write time to make reads cheap. We tokenize, invert, compress, and merge segments continuously so that a query touches only a few short posting lists and scores only the docs that can win (block-max WAND). The whole system is read-optimized because search is read-amplified: one query becomes hundreds of parallel shard executions.
  • Document-partition, not term-partition. We shard by hash(doc_id) so each shard is a complete, self-contained inverted index and the expensive posting-list intersection stays local. We accept fan-out to all shards rather than shipping giant posting lists across the network. The slowest shard owns p99, so we fight tail latency with adaptive replica selection and right-sized shards.
  • Immutable segments over in-place mutation. Never edit the index; append new immutable segments every ~1s and merge in the background. This buys lock-free concurrent reads during heavy writes, near-real-time freshness, and logical deletes via tombstones - at the cost of merge write-amplification and ~1s staleness, both deliberate.
  • Replicate for both speed and survival. Replica shards serve reads (throughput) and stand by for failover (availability), while a per-shard translog guarantees no acked write is lost between flushes. Durability comes from WAL + replicas, not from treating the derived index as irreplaceable.
  • Relevance from corpus statistics, carefully distributed. BM25 weights rare terms up (IDF), saturates repetition (k1), and normalizes length (b). IDF is a global statistic, so we lean on random sharding to approximate it locally and reach for distributed frequency stats only when precision demands it.

One-line summary: a document-sharded inverted index with block-max WAND execution returns BM25-ranked top-K in a parallel scatter-gather under 50ms, immutable segments plus a translog give near-real-time ingestion without losing writes, background merging keeps reads fast, and replica shards with primary failover make the whole thing fault-tolerant at billions of documents - trading write-time cost and ~1s freshness for cheap, ranked, highly available reads.