Design a Search Engine (Google)

End-to-End Query Execution Flow with Timing Budget

To satisfy an external SLO of < 200 ms p99, execution is structured across parallel stages with an internal target budget of ~110 ms, leaving approximately 90 ms of headroom for network jitter, queueing, cache misses, serialization, speculative hedging, and tail latency:

• Step 1: Query Understanding (10 ms): Performs tokenization, spell checking, query expansion such as expanding "NYC" into "New York City", and query intent classification.
• Step 2: Scatter to Index Shards (5 ms network): The stateless Query Coordinator broadcasts the parsed query to 100 document-sharded Tier-0 index servers in parallel, enforcing a 150 ms deadline before skipping non-responsive nodes.
• Step 3: Per-Shard Processing (25 ms): Each shard intersects compressed posting lists using two-pointer merges over sorted doc ID lists, computes initial BM25 scores, and selects local top 100 candidates into a min-heap.
• Step 4: Gather and Merge (5 ms): The coordinator collects up to 10,000 local candidates from all shards and runs a K-way max-heap merge to extract the global top 1,000 documents.
• Step 5: ML Reranking (50 ms): Evaluates the top 1,000 candidates across hundreds of signals including CTR, freshness, domain authority, and PageRank using LambdaMART, supplemented by BERT cross-encoders for the top 100 to 200 candidates to select the final top 10.
• Step 6: Snippet Generation (10 ms): Extracts high-density matching passages from the document store for the final top 10 ranked documents, highlights query terms, and truncates snippets cleanly.
• Step 7: Response Assembly (5 ms): Injects related search recommendations and dynamic knowledge cards before serializing the final JSON payload to the client.

Timing Summary: 10 + 5 + 25 + 5 + 50 + 10 + 5 = 110 ms internal execution target, comfortably within the < 200 ms p99 external SLO with ~90 ms reserved for queueing and tail-latency hedging.

Scatter-Gather: Index Shard Coordination

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

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 to 5 times its median (p50) latency due to garbage collection pauses, background indexing contention, or physical hardware variance.
• 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 an index shard. If that replica does not respond within 30 ms, dispatch an identical speculative request to a secondary replica. The coordinator accepts whichever replica responds first and cancels the outstanding request. This technique reduces tail latency substantially while incurring only an approximate 5% increase in total throughput load.
2. Speculative Execution and Partial Returns (Graceful Degradation): If 95 out of 100 shards return results within 100 ms, complete the query using those 95 responses rather than waiting for stragglers. This strategy deliberately trades candidate recall for predictable tail latency. It should only be enabled when the product explicitly accepts approximate retrieval under heavy load, because an omitted shard could theoretically hold a top-ranked result.
3. Tiered Index Execution: Segregate documents into Tier 0 (the top 1B popular documents), Tier 1 (extending cumulative coverage to the top 10B documents total by adding the next ~9B pages), and Tier 2 (encompassing the full 100B-page corpus by storing the remaining ~90B long-tail pages), with each tier running as an independent document-sharded, replicated cluster. Queries evaluate Tier 0 first; only if Tier 0 yields insufficient candidate volume or relevance does execution cascade to Tier 1 or Tier 2, avoiding massive coordination fan-out for the vast majority of searches.

Shard Routing Trade-off: Document vs. Term Sharding

• Document Sharding (Google Choice): Each shard stores the complete inverted index dictionary for an assigned subset of documents. Although the query coordinator must broadcast every query across all shards in the active tier (a fixed fan-out of 100 RPCs), per-shard processing is uniform, predictable, and simple to operate. Hot-term load is addressed by pre-warming specialized in-memory hot-term posting structures and caching posting fragments on coordinators rather than restructuring shard ownership.
• Term Sharding: Shards are partitioned by lexicographical term ranges (for example, Shard A stores terms beginning with A through C). While a search for "system design" contacts only the shards owning 's' and 'd' requiring 2 RPCs, intersecting posting lists across distributed nodes creates heavy network traffic, and popular search terms produce severe hot-shard bottlenecks.

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 subsets, the same term would yield divergent IDF weights across shards, rendering local BM25 scores globally incomparable.
• Solution: Global IDF Precomputation during Offline Index Build (MapReduce)
  1. Count total documents N across all index shards.
  2. Calculate document frequency n(q) for each term across all index shards.
  3. Compute the absolute global IDF value for each term.
  4. Persist these global IDF values directly into each shard's term dictionary.
• At query time, every shard evaluates documents against identical global IDF constants, ensuring BM25 scores remain directly comparable across shards so the K-way merge produces a mathematically correct global top-K list.

K-Way Merge Algorithm

With 100 shards returning their local Top-100 results sorted by score descending, the coordinator must merge 10,000 candidates to select the top 1,000 for ML reranking:

1. Initialize a Max-Heap of size 100, containing one entry representing the top result from each shard.
2. Insert the highest scoring candidate from each of the 100 shards into the heap.
3. Pop the maximum item from the heap, which represents the highest-ranked global result.
4. Push the next highest scoring result from that same shard into the heap.
5. Repeat steps 3 and 4 until the target global top 1,000 candidates are fully extracted for the ML reranking stage.

This algorithm executes in O(K x log(num_shards)) time, requiring only approximately 7,000 comparisons (O(1000 x log(100))) to merge 10,000 candidates.

• Tie-Breaking: When text relevance scores are identical, static PageRank authority scores serve as a deterministic secondary sort key.
• Near-Duplicate Deduplication: Because document sharding ensures each document resides on exactly one shard, identical document duplicates cannot occur. For near-duplicate content published across different URLs, SimHash fingerprints computed during offline crawling allow ML reranking workers to filter redundant pages.