Designing a Distributed Search Engine

Full-text search lets users find documents by words rather than keys. This article designs a search service in the style of Elasticsearch or Solr for an index of one billion documents.

Requirements

Functional

• Index documents with text and structured fields; update and delete them.
• Keyword, phrase, boolean and filtered queries with ranked results.
• Faceted counts and pagination; typo tolerance and prefix autocomplete.

Non-functional

• 5,000 queries per second, p99 latency under 200 ms.
• 10,000 document writes per second, searchable within 1 to 2 seconds.
• 99.9 percent availability; survive node loss without losing data.
• Index of 1 billion documents averaging 5 KB.

Capacity estimation

Raw text is 1 billion times 5 KB, which is 5 TB. An inverted index with positions typically lands at 30 to 60 percent of source size, so about 3 TB, plus stored fields and doc values for sorting: say 8 TB per copy. With one replica, 16 TB. At 1 TB of index per node (to keep working sets in page cache), we need about 16 data nodes, and more for query load. Shards of 30 GB give around 270 primary shards.

High-level design

 Client --> Query coordinator (any node)
                |  scatter to one copy of each shard
        +-------+--------+---------+
        v       v        v         v
     Shard 1  Shard 2  Shard 3 ... Shard N   (primary or replica)
        |  local top-K by score
        +--------> gather, merge, fetch top documents
 Indexing: Client --> router(hash(doc_id)) --> primary shard --> replicas
 Cluster state: master nodes (Raft-like), shard allocation

Data model

The core structure is the inverted index: a term dictionary (sorted, stored as a finite state transducer) mapping each term to a postings list of document ids, term frequencies and optionally positions. Postings are delta-encoded and compressed in blocks (for example 128 ids) with skip pointers. Alongside sit doc values (columnar fields for sorting and aggregations) and a stored-fields store for returning the original document.

Key deep-dives

Analysis pipeline. Text passes through tokenization, lowercasing, stop-word handling and stemming (running to run). The same analyzer must run at query time. Different fields can use different analyzers; autocomplete uses edge n-grams.

Segments and near-real-time indexing. New documents go to an in-memory buffer and a transaction log. A refresh every second writes a small immutable segment that becomes searchable. Segments merge in the background into larger ones, which also physically removes deleted documents. Immutability allows lock-free reads and aggressive caching.

Sharding and replication. A document goes to shard hash(id) modulo shard count. Shard count is fixed at index creation, so plan for growth or use time-based indices (one per day) for logs. Each shard has replicas; queries round-robin across copies, increasing throughput and tolerating failures. The primary forwards writes to replicas.

Query execution. Scatter-gather: the coordinator sends the query to one copy of every shard; each returns its local top K (document id and score); the coordinator merges and then fetches the full documents for just the final page. Deep pagination is expensive because every shard must produce offset plus size results, so prefer search-after cursors.

Ranking. BM25 scores a term by frequency within the document (with saturation), inverse document frequency across the corpus and document length normalization. Real systems layer on freshness, popularity and a learning-to-rank model that re-scores the top few hundred candidates. A subtle issue: IDF is computed per shard, so tiny shards give skewed scores; a distributed frequency pre-phase or larger shards fixes this.

Typo tolerance. Fuzzy queries expand terms within an edit distance using Levenshtein automata over the term dictionary.

Failure modes and trade-offs

• Node loss: replicas are promoted and missing copies rebuilt by recovery from the transaction log and segment files; throttle recovery to protect query latency.
• Slow shard: one slow shard sets the tail latency. Use hedged requests to another replica and timeouts with partial results.
• Split brain: require a master quorum so only one partition accepts cluster state changes.
• Merge storms: heavy merging consumes IO and hurts queries; cap merge throughput.
• Eventual freshness: the one-second refresh interval is a trade of indexing throughput against visibility; raising it to 30 seconds can double bulk indexing speed.
• Search is not a database: keep the system of record elsewhere and reindex from it, enabling schema changes by building a new index and swapping an alias.

Key takeaways

• The inverted index plus immutable segments is the foundation: fast reads, simple concurrency, background merging.
• Scatter-gather across shards with a local top-K keeps network cost proportional to page size, not result count.
• Pick shard counts deliberately; they are hard to change, and tail latency follows the slowest shard.
• BM25 gets you good baseline relevance; learning-to-rank on top candidates improves it cheaply.
• Treat the index as a rebuildable derived store and keep the source of truth in a database.