Design a Search Engine (Google)

End-to-End Query Execution Flow with Timing Budget

To serve search queries within an ultra-strict < 200 ms total budget, Google structures execution across parallel stages with fine-grained millisecond caps:

• Step 1: Query Understanding (10 ms): Tokenization, spell-checking, synonym query expansion (e.g. "NYC" → "New York City"), and intent classification.
• Step 2: Scatter to Index Shards (5 ms network): Stateless Query Coordinator broadcasts query to 100 document-sharded Tier-0 servers in parallel. Enforces a 150 ms deadline before skipping tardy nodes.
• Step 3: Per-Shard Processing (25 ms): Each shard performs compressed posting-list intersection (using fast two-pointer merges on sorted lists) + initial BM25 local scoring and local Top-100 heap selection.
• Step 4: Gather + Merge (5 ms): Coordinator receives up to 10,000 local candidates and performs a fast K-Way max-heap merge to fetch the global top 1,000.
• Step 5: ML Re-ranking (50 ms): Evaluates the top 1,000 candidates across hundreds of features (CTR, freshness, domain authority, PageRank) using LambdaMART + heavy BERT cross-encoders for the top 200.
• Step 6: Snippet Generation (10 ms): Fetches high-density passages for the final top 10 documents, highlights matching terms, and truncates snippets.
• Step 7: Response Assembly (5 ms): Inject related search queries, dynamic knowledge cards, and serialize the JSON payload.

Timing Summary: 10 + 5 + 25 + 5 + 50 + 10 + 5 = 110 ms total, well within target limits.

Scatter-Gather: Index Shard Coordination

Stateless query coordinators scatter queries to all document shards in parallel, presenting a major tail-latency issue.

The Tail Latency (Slow Shard) Problem:

• With 100 index shards, the aggregate query latency is bound by the slowest single shard.
• At 100 shards, the p99 latency of any single shard is approximately 3-5 times its median (p50) latency (due to garbage collection pauses, background indexing contention, or physical hardware issues).
• The likelihood of the entire query being slowed down by at least one laggy shard scales exponentially:
  P(at least one shard is slow) = 1 - (1 - 0.01)¹⁰⁰ = 63%
• Without software-level mitigations, 63% of all user search queries would suffer from high tail latency.

Technical Solutions:

1. Hedged Requests (⭐ Recommended): Send the query to the primary replica of a shard. If it does not respond within 30 ms, send an identical speculative request to the secondary replica. Take the result from whichever replica responds first and cancel the other. This reduces tail latency drastically while adding only a small (~5%) increase in overall throughput load.
2. Speculative Execution / Partial Returns: If 95 out of 100 shards have returned within 100 ms, complete the query using only those 95 shards. Missing a minor 5% of candidate documents is visually and qualitatively invisible to the user.
3. Index Tiers: Split the index into Tier 0 (viral/popular 1B documents), Tier 1 (10B documents), and Tier 2 (100B documents). Only fallback to search Tier 1/2 if Tier 0 yields insufficient results, avoiding massive coordination fan-outs.

Shard Routing Trade-off: Document vs. Term Sharding

• Document Sharding (Google's choice): Each shard stores the entire inverted index dictionary for a subset of documents. The query coordinator must broadcast queries to all shards (fixed 100 RPCs), but execution is highly uniform and simple.
• Term Sharding: Shards are partitioned by lexicographical term ranges (e.g., Shard A stores terms A-C). While a query only requires querying a few shards (e.g., query "system design" needs shards for 's' and 'd' → 2 RPCs), intersecting lists across nodes is incredibly network-expensive, and popular terms create massive load imbalances.

Result Merging Across Document Shards

To accurately merge Top-100 scores returned by independent document shards, we must solve BM25 score comparability and execute sorting efficiently.

The Global Score Comparability Problem:

• The BM25 formula depends heavily on **Inverse Document Frequency (IDF)**:
  IDF(q) = log((N - n(q) + 0.5) / (n(q) + 0.5))
  Where N is the total documents in the corpus and n(q) is the count of documents containing the term.
• If shards computed IDF using only their local document subset, the same term would carry different IDF values across shards, making the calculated BM25 scores incomparable.
• Solution: Global IDF Precomputation during Offline Index Build (MapReduce)
  1. Count total documents N across ALL shards.
  2. Count document frequency n(q) for each term across ALL shards.
  3. Compute the absolute global IDF value for each term.
  4. Store these IDF values in each shard's term dictionary.
• At query time, every shard uses the exact same global IDF value, making BM25 scores globally comparable and ensuring the K-way merge produces the mathematically correct global Top-K.

K-Way Merge Algorithm:

With 100 shards returning their local Top-100 results sorted by score descending, the coordinator must merge 10,000 candidates:

1. Create a Max-Heap of size 100 (containing one entry per shard).
2. Insert the first (highest scoring) result from each of the 100 shards into the heap.
3. Pop the maximum item from the heap → this is the global #1 result.
4. Push the next highest scoring result from that popped shard into the heap.
5. Repeat steps 3-4 until the global top-K (e.g., top 1,000) results are fully extracted.

This algorithm runs in extremely fast O(K × log(num_shards)) time, executing only around 7,000 comparisons (O(1000 × log(100))) to merge 10,000 candidates.

• Tie-Breaking: If text relevance scores are identical, static PageRank authority scores are used as a secondary sort.
• Near-Duplicate Deduping: Since document-sharding guarantees a document resides on exactly one shard, duplicates are impossible. For near-duplicates across different URLs, SimHash values are calculated on page text offline, and near-duplicates are filtered out in the ML re-ranking stage.