System Design: Designing a Distributed Search Engine (Elasticsearch)

Designing a search engine capable of parsing billions of text documents and returning highly relevant results in under 50 milliseconds is an extraordinary engineering challenge. Traditional relational databases using standard LIKE %keyword% queries fail immediately at scale because they require sequential table scans that saturate disk I/O.

To achieve sub-second search speeds across massive datasets, we must design a highly distributed search cluster leveraging an Inverted Index storage model, combined with intelligent partition routing, document sharding, and Lucene segment optimizations.

This case study deconstructs the architecture of a distributed search engine modeled on the core internals of Elasticsearch.

1. Requirements & Core Constraints

A production-grade distributed search engine must balance fast, real-time ingestion with ultra-low latency queries.

Functional Requirements

Document Ingestion: The system must accept text documents (JSON structures) and index them within seconds.
Full-Text Querying: Users must be able to perform keyword searches, including multi-term, prefix, and wildcard queries.
Fuzzy & Synonym Matching: The query engine must handle misspelled words and synonym mapping on the fly.
Document Scoring: Search results must be dynamically ranked based on a scoring algorithm (e.g. TF-IDF or BM25).

Non-Functional Requirements & SLAs

Sub-50ms Search Latency: The search coordinator must compile and return ranked query results within 50 milliseconds at P99.
Real-time Ingestion Delay: Uploaded documents must be indexed and searchable within 1 second of submission (Near Real-Time).
Linear Scalability: The system must scale horizontally to store 10 Billion documents and process 50,000 search queries per second (QPS).
Fault Tolerance: The cluster must survive the crash of any data node without data loss or downtime.

Back-of-the-Envelope Estimates

Let's calculate the hardware and network scaling bounds for a search cluster holding 10 Billion documents:

Ingestion and QPS Throughput: Assume a daily ingestion rate of 100 Million new documents. $$\text{Average Ingestion Rate} = \frac{100,000,000 \text{ documents}}{86,400 \text{ seconds}} \approx 1,157 \text{ documents/sec}$$ $$\text{Peak Ingestion Rate} = 1,157 \times 2 = 2,314 \text{ documents/sec}$$ If the platform handles 50,000 search QPS at peak: $$\text{Query Volume} = 50,000 \text{ QPS}$$
Cluster Storage Sizing: Assume an average document size of 2 Kilobytes (including title, body text, metadata fields). $$\text{Raw Document Storage} = 10,000,000,000 \times 2 \text{ KB} = 20 \text{ Terabytes}$$ Building an inverted index (containing token posting lists, position vectors, and term frequencies) adds approximately 40% metadata overhead: $$\text{Total Storage on Disk} = 20 \text{ TB} \times 1.4 = 28 \text{ Terabytes}$$ To ensure high availability and prevent data loss, we enforce a 1 primary + 1 replica sharding policy (2x multiplier): $$\text{Total Cluster NVMe Sizing} = 28 \text{ TB} \times 2 = 56 \text{ Terabytes}$$ Dividing this across 50 data nodes gives approximately 1.12 Terabytes of active NVMe storage per node, which is well within standard limits.

2. API Design & Core Contracts

Elasticsearch exposes RESTful HTTP interfaces. Below are the primary JSON schemas defining document ingestion and complex query resolution.

Ingest Document API

Adds a new document to the specified search index.

POST /api/v1/indices/{index_name}/documents

Request Payload (JSON):

{
  "document_id": "doc_88321049",
  "fields": {
    "title": "Scalable Systems: Designing Distributed Caches",
    "author": "Jane Dev",
    "content": "Consistent hashing rings are critical for routing cached objects without key disruption...",
    "publish_year": 2026
  }
}

Response Payload (JSON):

{
  "index_name": "engineering_articles",
  "document_id": "doc_88321049",
  "result": "CREATED",
  "primary_shard": 4,
  "took_ms": 12
}

Search Query API

Executes a multi-term search query with text scoring and filters.

POST /api/v1/indices/{index_name}/search

Request Payload (JSON):

{
  "query": {
    "match": {
      "content": {
        "text": "consistent hashing",
        "fuzziness": "AUTO"
      }
    }
  },
  "filter": {
    "term": {
      "publish_year": 2026
    }
  },
  "pagination": {
    "from": 0,
    "size": 10
  }
}

Response Payload (JSON):

{
  "took_ms": 18,
  "timed_out": false,
  "hits": {
    "total": 1420,
    "max_score": 3.842,
    "documents": [
      {
        "document_id": "doc_88321049",
        "score": 3.842,
        "source": {
          "title": "Scalable Systems: Designing Distributed Caches",
          "author": "Jane Dev"
        }
      }
    ]
  }
}

3. High-Level Design (HLD)

A distributed search engine avoids a single centralized master database model. The design is separated into distinct server roles to handle request routing, index writes, and consensus management.

Cluster Topology and Search Ingestion

The diagram below details the operational node layout and data replication paths of a 3-node primary/replica cluster layout.

graph TD
    Client([Application Client]) -->|1. Submit Search Request| Coordinator[Coordinator Node]
    
    subgraph Search Engine Cluster
        Coordinator -->|2. Route Query in Parallel| Shard0_P[Shard 0 Primary Node]
        Coordinator -->|2. Route Query in Parallel| Shard1_P[Shard 1 Primary Node]
        
        Shard0_P -->|3. Sync Write| Shard0_R[Shard 0 Replica Node]
        Shard1_P -->|3. Sync Write| Shard1_R[Shard 1 Replica Node]
    end
    
    subgraph Master Coordination
        MasterNode[Dedicated Master Node] -.->|Manages Cluster State Map| Shard0_P
        MasterNode -.->|Manages Cluster State Map| Shard1_P
    end
    
    Shard0_P -->|4. Return Local Hits & Scores| Coordinator
    Shard1_P -->|4. Return Local Hits & Scores| Coordinator
    Coordinator -->|5. Merge, Sort & Score| Coordinator
    Coordinator -->|6. Return Global Top-K Documents| Client

The Document Tokenization and Ingestion Pipeline

When a raw text document is ingested, it must pass through an analyzer pipeline before entering the inverted index.

graph LR
    Doc[Raw Text Document] -->|1. Ingest| CharFilter[Character Filter: Strip HTML]
    CharFilter -->|2. Tokenize| Tokenizer[Tokenizer: Split by Whitespace]
    Tokenizer -->|3. Filter| Lowercase[Lowercase Filter: Map to Lowercase]
    Lowercase -->|4. Stem| Stemmer[Porter Stemmer: running -> run]
    Stemmer -->|5. Write| PostingList[(Inverted Index Posting Lists)]

4. Low-Level Design (LLD) & Data Models

Operating at peak speed requires understanding how inverted indices map tokens to documents. Standard relational databases store maps of Document -> Words. An Inverted Index reverses this layout, mapping Word -> List of Document IDs (called a Posting List).

Inverted Index Logical Layout

Suppose we ingest two documents:

Doc 1: "Consistent hashing rings are fast"
Doc 2: "Fast caches use hashing"

The analyzer converts the documents into posting lists with term frequencies:

+------------+--------------------+----------------------------+
| Token      | Posting List       | Term Frequencies           |
+------------+--------------------+----------------------------+
| consistent | [Doc 1]            | {Doc 1: 1}                 |
| hashing    | [Doc 1, Doc 2]     | {Doc 1: 1, Doc 2: 1}       |
| ring       | [Doc 1]            | {Doc 1: 1}                 |
| fast       | [Doc 1, Doc 2]     | {Doc 1: 1, Doc 2: 1}       |
| cache      | [Doc 2]            | {Doc 2: 1}                 |
+------------+--------------------+----------------------------+

Compilable Python Implementation: Inverted Index & TF-IDF Searcher

The query execution engine parses multi-word searches, intersects posting lists, and calculates dynamic matching scores using Term Frequency-Inverse Document Frequency (TF-IDF) calculations. term frequency (how often a term appears in a document) is balanced against inverse document frequency (how rare the term is across the entire index).

import math
import re
from typing import List, Dict, Set, Tuple

class InvertedIndexSearcher:
    def __init__(self):
        # Inverted index structure: Map{Token -> List{DocumentID}}
        self.index: Dict[str, Set[str]] = {}
        # Stores raw documents: Map{DocumentID -> RawText}
        self.documents: Dict[str, str] = {}
        # Stores pre-calculated term frequencies for each document: Map{DocumentID -> Map{Token -> Frequency}}
        self.term_frequencies: Dict[str, Dict[str, int]] = {}

    def _tokenize(self, text: str) -> List[str]:
        """
        Splits text, converts to lowercase, and strips non-alphanumeric marks.
        """
        clean_text = re.sub(r"[^\w\s]", "", text.lower())
        return clean_text.split()

    def add_document(self, doc_id: str, content: str):
        """
        Tokenizes and indexes a document.
        """
        self.documents[doc_id] = content
        tokens = self._tokenize(content)
        
        # Calculate term frequencies for this document
        self.term_frequencies[doc_id] = {}
        for token in tokens:
            self.term_frequencies[doc_id][token] = self.term_frequencies[doc_id].get(token, 0) + 1
            
            # Append document ID to inverted index posting list
            if token not in self.index:
                self.index[token] = set()
            self.index[token].add(doc_id)

    def _calculate_idf(self, token: str) -> float:
        """
        Calculates the Inverse Document Frequency (IDF) for a term.
        """
        total_docs = len(self.documents)
        matching_docs = len(self.index.get(token, []))
        
        if matching_docs == 0:
            return 0.0
        
        # Standard IDF logarithm formula
        return math.log(1.0 + (total_docs / matching_docs))

    def search(self, query: str) -> List[Tuple[str, float]]:
        """
        Searches the inverted index for terms, calculates TF-IDF scores, and ranks results.
        """
        query_tokens = self._tokenize(query)
        if not query_tokens:
            return []

        doc_scores: Dict[str, float] = {}

        for token in query_tokens:
            posting_list = self.index.get(token, set())
            idf = self._calculate_idf(token)
            
            # Score each document in the token's posting list
            for doc_id in posting_list:
                # Term Frequency (TF) = term count in doc / total tokens in doc
                tf_count = self.term_frequencies[doc_id].get(token, 0)
                total_tokens = len(self._tokenize(self.documents[doc_id]))
                tf = tf_count / total_tokens
                
                # TF-IDF Score calculation
                tf_idf_score = tf * idf
                doc_scores[doc_id] = doc_scores.get(doc_id, 0.0) + tf_idf_score

        # Sort documents by calculated score in descending order
        return sorted(doc_scores.items(), key=lambda item: item[1], reverse=True)

# Local Verification Test Suite
if __name__ == "__main__":
    search_engine = InvertedIndexSearcher()
    
    # Ingest document corpus
    search_engine.add_document("doc_1", "Consistent hashing rings are fast and secure for web cache clusters.")
    search_engine.add_document("doc_2", "Web caches use simple hashing algorithms to lookup active data.")
    search_engine.add_document("doc_3", "Distributed databases rely on Paxos consensus to coordinate locks.")

    # Execute search query
    query_string = "consistent cache hashing"
    results = search_engine.search(query_string)
    
    print(f"Executing Query: '{query_string}'")
    print("---------------------------------------------")
    for doc_id, score in results:
        print(f"Document ID: {doc_id} | Matching Score: {score:.5f} | Snippet: '{search_engine.documents[doc_id][:55]}...'")
    
    # Verification check: doc_1 should score highest as it contains all search words
    if results[0][0] == "doc_1":
        print("\nTest Result: VERIFIED. TF-IDF Inverted Index Ranking Executed Successfully.")

5. Scaling Challenges & Bottlenecks

Operating a search cluster at high volume presents critical architectural bottlenecks.

Inverted Index Segment Merging & Write Amplification

Inside each shard, incoming documents are not written directly to the primary index file on disk. Doing so would saturate random write capacity. Instead, incoming writes commit to an in-memory buffer (Index Buffer) and a persistent transaction log (Translog).

Every 1 second, the buffer flushes to create a new immutable search segment on disk (Lucene Segment). Having thousands of small segments degrades search latency because the query engine must scan every segment to find matches.

The Solution: We run an asynchronous Segment Merge Engine in the background. This worker reads multiple small, immutable segments, compiles them into a single, large, sorted segment, and deletes the old segments. This balances ingestion speed with search performance but introduces write amplification. We mitigate this by using SSD-direct I/O block structures and limiting maximum segment sizes to 5 Gigabytes.

Split-Brain Prevention in Master Nodes

If a network partition isolates our cluster, one partition might elect a new master node while the original master is still running in the other partition. This leads to dual conflicting master nodes (split-brain), which quickly corrupts cluster routing maps.

The Solution: We enforce a minimum master node quorum constraint. The cluster state map can only be updated if a majority quorum of master-eligible nodes are present. In a cluster of $N$ master-eligible nodes, the minimum consensus quorum is defined as: $$\text{Quorum Bound} = \text{floor}\left(\frac{N}{2}\right) + 1$$ This mathematically guarantees that only a single partition can elect or act as a master node during network cuts.

6. Technical Trade-offs & Compromises

Balancing ingestion indexing speeds with query response latency requires deliberate design tradeoffs.

Ingestion Latency vs. Search Latency

The Tradeoff: We prioritize search speeds over immediate document freshness. When a document is ingested, it is not instantly committed to the global multi-segment disk index. Instead, it is written to the in-memory Lucene buffer.
Operational Strategy: This "Near Real-Time" (NRT) design compromises by introducing a 1-second synchronization window. We sacrifice immediate transaction consistency to avoid blocking edge search requests, keeping search latencies sub-50ms.

CAP Posture: CP vs. AP

Posturing: Elasticsearch operates as a CP (Consistency / Partition-tolerance) system for index metadata state, but acts as an AP (Availability / Partition-tolerance) system for document querying. Under network partitions, we allow readers to fetch slightly stale data from isolated replicas, but block new index creation updates unless a majority quorum is present.

7. Failure Scenarios & Operational Resiliency

Distributed search clusters are defined by continuous hardware failures and node drops.

Data Node Crash and Shard Rebalancing

If a server containing Shard 2 Primary crashes, the active dedicated cluster Master Node detects the failure (via missed node heartbeats).

Mitigation: The Master Node instantly promotes the Shard 2 Replica (located on a healthy data node) to Primary. The Coordinator Node updates its routing map, preventing search failures. In the background, the Master Node commands the remaining data nodes to copy the promoted Shard 2 Primary data and build a new replica elsewhere, maintaining high availability.

Mitigation of "Thundering Herd" on Shard Searches

If a highly popular keyword is queried, a coordinator node might fire thousands of sub-requests to all data shards concurrently, saturating data node execution pools.

Mitigation: We implement Request Coalescing at the coordinator layer. If multiple identical search queries arrive within a 10ms window, the coordinator executes only a single request to the backend shards and broadcasts the compiled response to all waiting clients.

8. Candidate Verbal Script

Mock Interview Sequence

Interviewer: How does a distributed search engine like Elasticsearch locate documents matching a search term in milliseconds across billions of records? Explain the sharding and query execution path.

Candidate: "The system achieves this by utilizing two core architectural concepts: an Inverted Index and a decoupled sharding model.

When a document is ingested, it is routed to a specific primary shard using a hash of its ID:

shard = hash(document_id) % total_shards

Inside each shard, raw text is tokenized, filtered, and mapped to an inverted index that associates individual terms with lists of document IDs (posting lists).

When a search query like 'consistent caching' arrives, the request lands on a Coordinator Node. The Coordinator splits the query into tokens ('consistent', 'caching') and routes the sub-queries to all active shards (or their replicas) in parallel.

Each shard queries its local, immutable Lucene segments. It retrieves the matching posting lists, intersects them, and calculates local TF-IDF or BM25 relevance scores. Each shard returns only its top-K local matches (e.g., the top 100 documents) back to the Coordinator Node.

Once the Coordinator receives responses from all shards, it merges the document lists, performs a global sort and re-scoring, and retrieves the full source fields for the top-10 global documents. This two-phase query-then-fetch execution path prevents routing massive volumes of document data over the network, keeping search latencies sub-50ms even at petabyte scale."

Interviewer: What is the purpose of BM25 scoring compared to standard TF-IDF?

Candidate: "Standard TF-IDF scoring has a major limitation: term frequency (TF) scoring scales linearly. If a term appears 100 times in a document, it receives 100 times the score of a single appearance, which often skews results.

BM25 resolves this by introducing Term Frequency Saturation. As term frequency rises, the score contribution follows an asymptotic curve, capping its influence. BM25 also accounts for document length normalization, penalizing long documents where terms appear occasionally while boosting concise documents where the term represents a larger share of the content."