Design Google Search: Inverted Indexes, Sharding, Scatter-Gather and Ranking

Listen

Pin the scope. A text query returns ten ranked results with a title, URL and snippet, with paging and language and region filters. Ordinary pages should be searchable within days of being crawled, news within minutes. Server-side p ninety-nine must stay under about half a second, and the service can never be down. Ads, images, maps and instant answers are out. We will assume the crawler from Design Problems twenty-four feeds us pages.

Ask scale

Assume ten billion queries a day: a hundred thousand a second, three hundred thousand at peak, with a cache absorbing maybe forty percent. Fifty billion pages of text is five hundred terabytes, about a hundred and fifty as a compressed index. Keep the best ten percent in a fast tier: fifteen terabytes in memory across a hundred shards. At two thousand queries a second per server, each shard needs about ninety replicas. Thousands of machines, by design.

assume 10B queries/day
  = 100,000 qps avg, 300,000 peak
cache hits 40% -> 180,000 qps to index

50B pages x 10 KB text = 500 TB
inverted index, compressed = 150 TB
tier 1 (best 10% of pages) = 15 TB
  100 shards x 150 GB in RAM
one server ~2,000 qps -> 90 replicas
  = about 9,000 machines for tier 1

Define API

Publicly there is one endpoint: a GET with the query, the offset, the interface language and the region, returning results plus a spelling suggestion. The interesting calls are internal. Every index shard exposes search, which takes parsed terms and filters and returns its local top fifty document ids with scores. Doc servers turn a document id and the query terms into a title, URL and highlighted snippet.

GET /search?q=solar+panel+cost
     &start=10&hl=en&gl=in
200 { results: [{ url, title, snippet }],
      spelling: null, took_ms: 180 }

internal:
shard.search(terms, filters, k=50)
  -> [(doc_id, score)]
docserver.snippet(doc_id, terms)
  -> title, url, highlighted text

Data model

The core structure is the inverted index: for each term, a posting list of the documents containing it, with term frequency and positions for phrase matching. Store doc ids as gaps from the previous one and compress them with variable-length encoding, which shrinks lists dramatically. A forward store maps each doc id to its URL, title, text and static quality signals. None of this lives in a general database: it is immutable segment files, built offline and swapped in, much like Lucene segments.

inverted index (per shard):
  "solar" -> [d3, d9, d42, ...]
  posting = doc_id gap, term freq,
    positions; delta + varint coded
forward / doc store:
  doc_id -> url, title, text, lang
  static signals: link score, spam

immutable segment files, rebuilt
  and swapped; no row updates

High-level design

The offline half. The crawler stores parsed pages in the doc store. A batch job extracts the link graph and computes link-based importance scores. The indexer assigns each document to a shard, builds that shard's inverted index segments with the scores baked in, and ships new segments to every replica, which load them and swap. This half is all throughput: big batch jobs measured in hours.

The online half. The front end parses the query, fixes spelling and checks the results cache. On a miss, a root aggregator sends the query to one replica of every shard in parallel. Each returns its local best fifty. The root merges them, sends the top few hundred to an expensive re-ranker, then asks doc servers for snippets of the final ten. This half is all latency: a few hundred milliseconds, end to end.

Deep dive: the index

Answering a three-word query means intersecting three sorted posting lists. Walk the shortest list first and use skip pointers to leap ahead in the long ones, so you never scan the whole list for a common word. A neat trick: order postings by a static quality score instead of doc id. Then the best candidates come first, and a shard can stop early once it is sure no later document can beat its current top fifty.

solar: 3, 9, 42, 77, 120, 300, ...
panel: 9, 15, 42, 120, 512, ...
cost:  1, 9, 42, 99, 120, 640, ...

walk the shortest list first
skip pointers jump ahead in big lists
  -> 9, 42, 120 contain all three

order postings by static score:
  stop early once top-k is certain

Deep dive: building

Because we partition by document, each shard can build its own index from its own documents, with no global shuffle. Rebuild or merge segments every few hours. That is too slow for news, so add a fresh tier: a small in-memory index per shard, updated within minutes from the crawler's priority stream. Queries search both and merge, with the newer version of a page winning. Deleted or changed pages get tombstones until the next merge drops them.

batch: shard = hash(doc_id) mod 100
  each shard builds its own segments
  rebuild or merge every few hours

fresh tier: news and changed pages
  small in-memory index per shard,
  updated within minutes
  query both, merge, newest wins

deletions: tombstone list per segment

Deep dive: sharding

There are two ways to split an inverted index. By term, each shard owns some words, so a query touches only a few shards. But popular terms overload their shard, and intersecting means shipping enormous posting lists across the network. By document, each shard owns a slice of pages with a complete mini index. Every query hits every shard, but load is perfectly even and each shard returns only fifty ids. Web search uses document partitioning, and you add replicas for throughput.

Deep dive: tail latency

Here is the subtle part. If each shard is slow one time in a hundred, and every query waits for a hundred shards, then about sixty-three percent of queries hit at least one slow shard. Two fixes. Hedging: if a shard has not answered by its ninety-fifth percentile time, send the same request to another replica and take whichever answers first. Deadlines: at a fixed time, merge whatever has arrived. Missing one shard out of a hundred barely changes the top ten.

one shard p99 = 50 ms
100 shards: 1 - 0.99^100 = 63% of
  queries wait on a slow shard

hedge: resend to another replica
  after p95 (e.g. 30 ms), take first
deadline: answer at 150 ms with
  99 of 100 shards; quality dips
  slightly, user never notices
tree: root -> 10 mids -> 10 leaves

Deep dive: ranking

Ranking is a funnel. On each shard, a cheap score combining text relevance, such as BM25, with the page's static quality picks the local top fifty from possibly millions of matches. The root merges those into a few hundred. Only then does an expensive learned model run, using hundreds of features like freshness, locale and how users engaged with results before. Finally, limit results per site for diversity, and build snippets only for the ten you show.

stage 1, per shard (cheap):
  BM25 text match + static score
  millions of matches -> top 50
stage 2, root (merge):
  100 shards x 50 -> top 500
stage 3, re-ranker (expensive):
  learned model, hundreds of features
  freshness, locale, click signals
  -> top 10, then dedup by site
snippets: only for the final 10

Deep dive: link signals

The classic static signal is PageRank, which was published by the founders of Google in the late nineties. A page is important if important pages link to it, with each page splitting its vote across its outgoing links, plus a small random-jump term. Compute it by iterating over the whole link graph until the scores settle, as an offline batch job. Anchor text is a second gift from links: the words other sites use to link to a page describe it well.

rank(p) = 0.15 / N
  + 0.85 x sum over pages q linking p
        of rank(q) / outlinks(q)

iterate over the whole link graph
  until scores settle (about 50 rounds)
batch job, days old is fine

also: anchor text of incoming links
  describes the target page

Deep dive: query side

The front end does query understanding once, before the fan-out. Normalize case and accents, correct spelling using what other people typed and then retyped, and expand terms with stems and synonyms like cost and price. Then check a results cache keyed by normalized query, language, region and page. Popular queries repeat constantly, so even a cache with a few minutes of lifetime saves a large fraction of the expensive fan-out.

Trade-offs

Failures. A dead replica is invisible because the root simply picks another. If a whole shard is slow, answer without it and track the partial-results rate. A bad index push is the real danger, so canary new segments on a few replicas, compare quality metrics, then roll forward. Each region holds a complete copy of the index, so if one region fails, traffic shifts to another. When a hot query's cache entry expires, let only one request recompute it.

Name the trade-offs. Freshness costs money, so keep the fresh tier small and the big tier batch built. Tiers and early termination trade a little recall for a lot of latency. Replicas add queries per second, while more shards let you index more pages or make each lookup faster. And any public signal, like links, gets gamed, so spam detection is a permanent part of ranking.

Monitor latency percentiles, the partial-results rate, cache hit rate and the age of each index tier. Quality needs its own signals: how often people click the first result, how often they bounce straight back, and human raters scoring sampled queries. Ten times more pages means more shards and deeper tiers. Ten times more queries means more replicas, a larger cache and more regions. The architecture is the same; only the multipliers change.

Interview summary

In the interview: three hundred thousand queries a second at peak over a hundred and fifty terabyte index. A compressed inverted index, partitioned by document and heavily replicated. A batch tier plus a fresh tier. Scatter-gather from a root aggregator, with hedged requests and deadlines for tail latency. A ranking funnel from cheap text and static scores to a learned re-ranker. Link scores computed offline, a results cache, and a full copy per region.