Design Google Search
Search 50 billion pages at 100 k queries a second: the ten best results with snippets in under 200 ms, and news within minutes.
Last updated 2026-10-26. Difficulty: hard. Patterns: inverted-index, scatter-gather, tail-latency, ranking, search. Reported at Google, Microsoft, Amazon, Meta, LinkedIn, Apple.
The interviewer asks, the candidate answers and draws, and you press Next. Skip to any stage or key moment, and ask the AI Mentor anything along the way.
Or sit it in the style of Google · Microsoft · Amazon · Meta, run as that company’s round is reported to go.
The short answer
How to design Google Search in a system design interview, in five steps:
- Scope it. Return the ten best results for a query, query-dependent snippets, understand the query, phrases and operators, fresh content within minutes, index the crawled web, removals take effect quickly.
- Set the targets. Latency: p99 < 200 ms server side; Query rate: 100 k/s peak · ~4 B/day; Corpus: 50 B documents · ~125 TB index; Freshness: fetched news searchable < 1 min.
- Do the numbers. Query rate: ~100 k/s peak · ~4 B/day; Index size: ~2.5 KB/doc · ~125 TB; Tier-1 shard size: 10 M docs · ~25 GB in RAM.
- Draw the design. Search front end, Query understanding, Result cache, Root server, Aggregators, Tier-1 leaves, Tier-2 leaves, Real-time index, Re-ranker.
- Go deep where it is hard.
- Inside a shard: the inverted index → Batch-built immutable shards: delta + varint posting lists, doc ids in static-rank order.
- Document- vs term-partitioned index → Document-partitioned: each shard indexes 1/1,000 of pages, all terms.
- Tail latency at a 1,000-way fan-out → Hedged requests at the leaf p95, a hard deadline with partial results, replicas, and an aggregator tree.
- Two-phase ranking → Cheap score on leaves (BM25F + proximity + static rank), learned re-ranker on the merged top 500.
- Tiered index: the best pages first → Two tiers by static rank: 10 B in RAM, 40 B on SSD, tier 2 only when tier 1 will come up short.
- Freshness: a real-time tier and an honest cache → A small real-time tier merged at query time, freshness-classed TTLs, and burst-based invalidation.
Functional requirements
- Return the ten best results for a query. Ranked by relevance and quality, each with title, URL and snippet. Later pages exist but are rare, so the design optimises page one.
- Query-dependent snippets. The lines under each result show the query's words in context, so they are generated per query from stored page text, and only for the ten results shown.
- Understand the query. The same tokenisation as the index, spelling correction ("did you mean"), stemming and synonyms, language and region. Most bad results start as a badly understood query.
- Phrases and operators. Quotes, site: and minus terms. Exact phrases need word positions in the index, which roughly doubles its size.
- Fresh content within minutes. A news article fetched a minute ago should rank for the query people are typing about it. The main index is rebuilt daily, so freshness needs its own path.
- Index the crawled web. Turn the crawler's stored pages and link graph into a sharded inverted index: parse, keep one page per duplicate cluster, compute link-based quality (PageRank) and anchor text, invert, compress, ship.
- Removals take effect quickly. Dead pages, spam and legal takedowns must leave results, and cached result pages, within minutes to hours, not at the next weekly build.
- Out of scope. Crawling (the Web Crawler question) and autocomplete (the Typeahead question) are existing systems we consume. Also out: ads, image and video verticals, knowledge panels, personalisation beyond language and region, and training the ranking model.
Non-functional requirements
- Latency (p99 < 200 ms server side). Budget: ~15 ms front end and query understanding, ~10 ms canary, ~70 ms fan-out with a 60 ms leaf deadline, ~30 ms re-rank, ~20 ms snippets, ~15 ms render, ~40 ms reserved for a tier-2 fall-through. This drives the hardest tradeoff: at a 1,000-way fan-out, waiting for every leaf makes the slowest leaf everyone's latency.
- Query rate (100 k/s peak · ~4 B/day). Read-only from the user's side. About half is absorbed by the result cache; the other half fans out to every index shard.
- Corpus (50 B documents · ~125 TB index). After the crawler's duplicate collapse. A thousand times what one machine holds in memory, which forces the partitioning decision.
- Freshness (fetched news searchable < 1 min). "Minutes after publication" end to end also depends on the crawler finding the page. The batch index lags by a day (tier 1) or a week (tier 2); the real-time tier covers the gap.
- Availability (99.99 %). Degrade by recall, never by errors: losing one shard drops 0.1 % of documents from a query, which nobody can see. An error page or a two-second page is seen by everyone.
- Relevance (NDCG offline + A/B online). Ranking quality is a requirement with a number, checked on human-rated queries and live experiments. A change that is faster but worse does not ship.
- Cost (RAM only where it earns it). Everything in memory at 15 replicas would be ~1.9 PB of index in RAM. Tiering and the result cache bound the cost per query; this is the second hardest tradeoff after latency.
Back-of-envelope estimates
- Query rate: ~100 k/s peak · ~4 B/day. Peak 100 k/s is about twice the average, so ~50 k/s × 86 400 s ≈ 4.3 B queries/day. All of them are reads; documents arrive from the crawler at a far lower rate and in batches.
- Index size: ~2.5 KB/doc · ~125 TB. A page has ~1 000 words and ~400 distinct terms after boilerplate removal: 400 postings. With delta + varint a posting with positions is ~5 B (vs ~20 B raw), so 2 KB of postings plus ~0.5 KB of per-document scoring data. 50 B docs × 2.5 KB ≈ 125 TB. It grows with the corpus, not with time, because it is rebuilt rather than appended.
- Tier-1 shard size: 10 M docs · ~25 GB in RAM. Tier 1 is the best 10 B documents over 1 000 shards: 10 M docs per shard × 2.5 KB ≈ 25 GB. Small enough for a leaf to hold two copies in memory, which is what lets a new version load beside the live one.
- Queries reaching the index: ~50 k/s · ~55 M leaf requests/s. A result-cache hit rate of ~50 % (Jeff Dean's 2009 WSDM talk gives 30–60 % for Google's cache servers) leaves 100 k × 0.5 = 50 k/s at peak. Each fans out to ~1 100 leaves (1 000 tier 1 plus 100 real-time): ~55 M leaf requests a second.
- Tier-1 leaf machines: ~15 replicas/shard · ~15 k leaves. A leaf spends ~4 ms of CPU per query; 32 cores at 60 % ≈ 5 k queries/s per replica. 50 k/s ÷ 5 k = 10 replicas per shard for throughput, ~15 once three regions are each sized to absorb a lost one. 1 000 shards × 15 ≈ 15 000 machines: the fan-out, not the corpus, sets the fleet size.
- Queries that wait on a slow leaf: ~100 %. If each leaf exceeds its p99 (40 ms) 1 % of the time, all 1 000 are fast with probability 0.99^1000 ≈ 0.004 %. Almost every query waits on some leaf's tail, so without hedging and deadlines the leaf p99.9 becomes the query median.
- Real-time tier: ~170 M docs · ~1.7 TB. ~1 k fresh docs/s from the crawler's priority lanes × 48 h (172 800 s) ≈ 170 M docs. Updatable in-memory segments compress less, ~10 KB/doc, so ~1.7 TB over 100 shards of ~17 GB. Under 0.4 % of the corpus: freshness is a small index, not a fast big one.
- Snippet reads and doc store: ~500 k reads/s · ~200 TB. 10 snippets × 50 k queries/s = 500 k doc reads/s. Making snippets for all 500 re-ranked candidates would be 25 M/s, 50× more, which is why they come last. The doc store is 50 B docs × ~4 KB of compressed text ≈ 200 TB, three replicas on SSD.
Components
- Browser / app: The search box and results page. Sends the submitted query (suggestions while typing come from the separate Typeahead system) and reports clicks, the main online signal of whether ranking is good. Results are the same for everyone with the same language and region, which is what makes them cacheable.
- Search front end (owns the 200 ms budget): Terminates TLS in the nearest region, calls query understanding, checks the result cache and on a miss hands the query to a root server. Holds the small hot-term table from the real-time tier, which decides a query's freshness class and whether a cached page is still valid. Sets one deadline that every downstream call inherits, renders the page, and logs impressions and clicks.
- Query understanding (spelling · intent · term stats): Normalises and tokenises the query with exactly the analyser the indexer used, corrects spelling from what users retype, adds stems and synonyms, and detects language. Returns each term's global document frequency, which gives the root its IDF and predicts whether the query is rare. Also tags freshness intent from history: do users of this query click recent results?
- Result cache (query → page · ~50 % hits): Finished result pages keyed by normalised query, language, region and SafeSearch: ~100 M entries, ~500 GB per region. Query popularity is Zipf-like, so the head is cheap to cache, and the ~15 % of daily queries that are new cap the hit rate. TTL follows the freshness class (60 s for news-like queries, hours otherwise), and an entry created before a burst on one of its terms is treated as a miss.
- Root server (scatter, merge, re-rank): Runs one query: canary leaves first, then the fan-out to tier 1 and the real-time tier (and tier 2 when needed), merge, dedupe by URL fingerprint, re-rank, snippets. Holds the query deadline and decides whether shard coverage is good enough to serve. Stateless; any root can serve any query.
- Aggregators (~25 per query, 40 leaves each): The middle of the serving tree. Each sends the query to 40 leaves, hedges the slow ones, merges 400 results into its top 50, and returns at the deadline with whatever it has plus a count of shards answered. Keeps any one machine from gathering 1,000 responses and puts hedging decisions next to the leaves.
- Tier-1 leaves (1,000 shards × 15 replicas · RAM): Each holds one shard: ~10 M of the best documents, ~25 GB of compressed postings and per-document data in memory. Intersects posting lists, scores matches with BM25 over fields plus proximity and static rank, prunes with WAND-style bounds, and returns its top 10 with features. Documents are placed by URL hash, so every shard does the same work for every query.
- Tier-2 leaves (1,000 shards × 6 replicas · SSD): The long tail: the other ~40 B documents, ~100 GB per shard on local NVMe SSD, rebuilt weekly. Searched in parallel when a query looks rare, or afterwards when tier 1 returns too few good results: ~15 % of index queries. Rare terms have short posting lists, so SSD is fast enough for exactly the queries that get here.
- Real-time index (~48 h · 100 shards · in memory): Indexes fresh documents from the stream into small in-memory segments that become searchable every ~5 s and merge in the background. Queried on every cache miss beside tier 1; when both return a page, the newer copy wins. Counts terms in fresh documents per 5-minute window and publishes the bursting ones as hot terms.
- Re-ranker (learned model · 500 candidates): Orders the merged candidates with a learned model over a few hundred features: the leaves' text scores, anchor and click signals, freshness, and the similarity of query and document embeddings. About 50 µs a document, affordable for 500 and absurd for millions. Trained offline on human ratings and clicks; a new model ships only after it wins an A/B test.
- Doc servers (text by URL fingerprint · 200 TB): The stored text, title and URL of every indexed page (~4 KB compressed), sharded by URL fingerprint and replicated three times on SSD. Builds query-dependent snippets for the final ten results only. Written by the batch build and by the fresh stream, so new pages get snippets too.
- Crawler (existing) (content store + crawl log): The Web Crawler design, taken as given: it fetches pages politely, collapses duplicates, and stores pages with a log of every fetch and the extracted link graph. Its priority lanes (news sites, pinged sitemaps) also publish to the fresh stream. Everything in this design starts from its output.
- Fresh doc stream (Kafka · key = URL fingerprint): Recently fetched pages from the crawler's priority lanes, ~1 k documents a second. Keyed by URL fingerprint so successive versions of one page stay in order. The real-time index and the doc servers consume it independently, so a slow consumer delays only itself.
- Batch indexer (MapReduce · tier 1 daily): Parses pages, computes PageRank and anchor text over the link graph, assigns documents to shards by URL hash, to tiers by static rank and to doc ids in rank order, then inverts and compresses postings in a MapReduce-style job. Rebuilds tier 1 daily and tier 2 weekly. Validates every build against the live version before it ships.
- Index versions (immutable shard files · DFS): Versioned, immutable shard files on a distributed file system: postings, term dictionary, per-document data and global term statistics. Leaves load a new version beside the old one and swap. Rollback is pointing at yesterday's files, never a rebuild under pressure.
User flows
- A query the cache has not seen. The main read path. One query becomes about 1,100 leaf requests plus aggregators, a re-ranker and ten doc servers, and must come back within 200 ms at p99. Each stage does expensive work on fewer documents than the last: millions matched, 10 per leaf, 500 re-ranked, 10 snippets.
- The browser sends the query to the nearest front end. DNS and anycast route to the closest region. The front end owns the 200 ms budget and stamps a deadline that every downstream call inherits, so work stops as soon as its answer can no longer be used. It logs the impression now and the click later, for ranking training and cache statistics.
- Query understanding normalises the text, fixes spelling, and returns term statistics and a freshness intent. Tokenisation must match indexing exactly (Unicode normalisation, case folding, stemming), or query terms never meet their postings. Spelling corrections come from query logs: what people retyped after a miss. Global document frequencies give the root its IDF and predict whether the query is rare.
- The front end looks the normalised query up in the result cache and misses. The key is a hash of normalised terms, language, region, SafeSearch and page number. About half of all queries hit and return the finished page in ~20 ms without touching the index; this one is in the long tail.
- The root sends the query to two canary leaves, then to the whole tier-1 tree and the real-time tier at once. Canaries first: a query that crashes leaves (a parser bug, a pathological term) takes down two machines instead of a thousand, a technique Dean and Barroso describe in "The Tail at Scale". Global IDF travels with the query so every shard scores on the same scale. About 25 aggregators fan out to 40 leaves each, so no machine gathers 1,000 responses.
- Each leaf finds the matching documents in its shard, scores them cheaply, and returns its top 10. Doc ids are assigned in descending static rank, so lists sorted by id are also sorted by quality. The leaf intersects from the rarest list using skip pointers, scores matches with BM25 over title, body, anchor text and URL plus proximity and static rank, and uses WAND-style upper bounds to skip documents that cannot beat its current tenth. It scores thousands of documents rather than millions, ~4 ms of CPU.
- Aggregators merge their leaves' lists; the root merges those with the real-time hits and keeps the best 500. Each aggregator keeps 50 of its 400 and the real-time tier contributes its best 100, so the root keeps 500 of ~1,350 by cheap score. Duplicates collapse on URL fingerprint and a real-time copy beats the batch copy because it is newer. A leaf that misses the deadline costs 0.1 % of the documents for this query, not the query.
- The re-ranker orders the 500 candidates with a learned model, and the root picks the final ten. Gradient-boosted trees or a small network over a few hundred features: the leaf's text scores, anchor matches, the document's click history, freshness, and query-document embedding similarity. ~50 µs a document, ~25 ms for 500. Then rules: at most two results per host, near-duplicates collapsed, removals filtered.
- Doc servers build query-dependent snippets for the ten winners. Each doc server reads the page's stored text and picks the window with the most query terms closest together, then highlights them. Ten documents instead of 500 is 500 k reads a second instead of 25 M. A doc server that misses its 20 ms deadline is replaced by the page's stored description.
- The front end renders the page, caches it with a TTL set by freshness class, and returns it. Low-freshness queries are cached for hours, bounded by the index version, with jitter so entries never expire together; fresh-class queries for 60 s. The entry stores its terms and creation time, so a later burst on any of its terms invalidates it on read.
- Building and shipping the daily index. The batch write path. The crawler's pages become 1,000 immutable tier-1 shards every day (tier 2 weekly) through a MapReduce-style job, then replace the live index one replica group at a time. The build may be slow; the swap must never be risky.
- The batch indexer reads the crawler's stored pages and link graph. The crawler's content store and crawl log are its output and our input. Parsing extracts main text, title, headings, outlinks with their anchor text and language, and keeps one page per near-duplicate cluster the crawler identified.
- Link analysis computes PageRank and attaches anchor text to the pages it points to. PageRank (Brin and Page, 1998) is a few dozen iterations over ~50 B pages and ~1 T links in a graph-processing job. Anchor text often describes a page better than the page itself, so it is indexed as a field of the target. With spam and quality scores this gives every page a static rank: its quality before anyone types a query.
- Pages are assigned to shards by URL hash, to tiers by static rank, and to doc ids in rank order. shard = hash(url_fp) mod 1,000 spreads pages, and therefore work, evenly. The top 10 B by static rank go to tier 1 and the other 40 B to tier 2. Within a shard doc id 0 is the best page, so every posting list is in quality order and a leaf can stop early.
- Mappers tokenise each page and emit postings; one reducer per shard writes compressed posting lists. The shuffle groups by shard and sorts by term, then doc id. Each list stores gaps between doc ids in variable-length bytes (group varint, which Dean's 2009 WSDM talk describes, decodes faster than plain varint), with skip pointers every 128 postings, shrinking a posting with positions from ~20 bytes to ~5. A failed reduce task is retried alone, because every shard is an independent partition.
- The build writes immutable shard files and the document text for snippets. A tier-1 shard is ~25 GB: postings, term dictionary, per-document data and global term statistics. Files are versioned and never modified, so rollback is pointing at the previous version. The doc store gets each page's text, title and URL (~4 KB compressed) keyed by URL fingerprint.
- Validation compares the new build with the live one before any leaf loads it. Per-shard document counts and sizes against yesterday (a shard 30 % smaller is a bug, not news), then a replay of sampled queries against both versions comparing top-10 overlap and NDCG on a rated query set. A failed build simply does not ship: yesterday's index keeps serving and the real-time tier keeps the newer pages.
- Leaves load the new shard beside the live one, warm it, and swap, one replica group at a time. A tier-1 leaf has memory for two 25 GB copies: it loads the new one while serving the old, warms it with replayed queries, flips a pointer, and frees the old. One of 15 replica groups goes first and is watched for latency, coverage and click metrics before the rest follow. Mixed versions during the rollout are harmless, because a page lives in the same shard in both.
- Breaking news, searchable within a minute. An earthquake story is fetched at 14:02 and ranks for "earthquake" by 14:03, although the batch index was built last night and "earthquake" is suddenly the hottest key in the cache. Freshness needs its own small index and a cache that knows which entries went stale.
- The crawler fetches the article and publishes it to the fresh-document stream. Only the crawler's priority lanes publish here (news sites, sitemaps with a recent lastmod, pinged URLs): ~1 k documents a second, a tiny fraction of the crawl. Keyed by URL fingerprint so later versions of the same page stay ordered.
- Real-time leaves index the article in memory and make it searchable within seconds. The same analyser as the batch build writes into small in-memory segments that are opened for search every ~5 s and merged in the background, an LSM-like design. With no PageRank yet, static quality comes from the host's history and early links. The tier keeps ~48 h, ~170 M documents over 100 shards; anything older is in the batch index.
- The article's text is also written to the doc store so it can get a snippet. A result without a snippet is a bad result. A separate consumer writes it keyed by URL fingerprint, so a newer version of a page replaces the snippet source and a slow doc store never delays indexing.
- The real-time tier sees "earthquake" bursting and publishes it to the front ends as a hot term. Per-term counts in fresh documents over 5-minute windows, compared with each term's own baseline, so a term that is always common never looks like news. A hot term makes any query containing it fresh-class: a 60 s cache TTL and a recency boost in ranking.
- Searches for "earthquake" jump to 20 k/s; the cached page from before the burst is treated as a miss. The cached page from 13:50 has no news in it. Instead of finding every affected key, the front end compares an entry's created_at with the burst time of each of its terms on read. Invalidation by comparison costs nothing to issue and is exact for the queries that are actually asked.
- One request per key goes to the index; the rest wait for it or are served the current page. Single-flight per cache key: the first miss computes the page and concurrent requests wait up to ~100 ms for it. With a 60 s TTL and stale-while-revalidate, the busiest query in the world costs the index about one query a minute per region. The recency boost lets the 14:02 article outrank a high-PageRank page about earthquakes in general.
- Tonight's build includes the article, and the real-time tier drops it once that version is live. The real-time tier keeps everything newer than the live batch version's cutoff, which is why it holds 48 h for a daily build: a late or failed build must not make yesterday's news vanish. During the overlap the root dedupes by URL fingerprint and keeps the newer copy.
- One slow leaf among a thousand. At a 1,000-way fan-out something is always slow: a leaf in a garbage-collection pause, a machine being drained, a failing disk. Waiting for everyone would make the slowest leaf's tail every query's latency, so the design never waits for everyone.
- The aggregator sends each shard's request to the replica with the fewest outstanding requests. Each shard has ~5 replicas per region. Least-outstanding-requests (or the better of two random picks) steers away from busy machines before anything has gone wrong, which is the cheapest tail-latency fix there is.
- Replica 3 is stuck in a 150 ms pause; after 20 ms with no answer, the aggregator sends a backup to another replica. Hedging at about the leaf's p95 sends a backup for ~5 % of requests, and the loser is cancelled on the first answer. Hedges are capped at 5 % per aggregator so they cannot snowball when every leaf is slow. "The Tail at Scale" reports that hedging after 10 ms cut a 100-server BigTable fan-out's p99.9 from 1,800 ms to 74 ms for 2 % more requests.
- A replica machine dies; its in-flight requests fail fast and are retried on a sibling. A reset connection is an immediate signal, unlike a slow leaf, so the retry lands within milliseconds and well inside the deadline. Health checks drop the machine from every aggregator's list within seconds, and its siblings absorb the load because each runs at ~60 % CPU.
- At 60 ms the aggregator stops waiting and returns what it has, saying which shards are missing. With document partitioning, a missing shard removes 0.1 % of the documents from this one query, and the chance that one of its documents belonged in the top ten is about 1 %. That is invisible; a two-second page is not. The coverage count lets the root and monitoring see it.
- The root checks coverage; below a floor it retries the query in another region instead of serving a quietly bad page. Coverage is normally 99.9 % or more. If a rack or a replica group is out and it falls below ~97 %, results get noticeably worse, so the front end resends the query to the next region and an alert fires. Silent partial results must never become the normal state.
- A replica that is persistently slower than its siblings is put on probation. Latency-induced probation, another technique from "The Tail at Scale": the replica gets only shadow requests until its latency matches its siblings again. Slow machines are usually a noisy neighbour, a failing disk or a stuck background task, and none of those recover faster under live load.
- A rare query falls through to tier 2. Most queries are answered entirely from the best 10 B documents in memory. A query for an obscure part number needs the long tail, which lives on SSD in tier 2 and is searched only when tier 1 will come up short.
- Query understanding returns tiny document frequencies, so the root predicts tier 1 will be thin. For "TPS54331 errata" the rarest term appears in a few thousand pages in the whole corpus. When the expected number of tier-1 matches, from global document frequencies, is under ~100, the root queries both tiers in parallel rather than in sequence, which keeps the query inside its budget.
- The root fans out to tier 1 and tier 2 through the aggregators at the same time. Tier 2 has the same 1,000-way layout with 40 M documents per shard, ~100 GB on local NVMe. Rare terms have short posting lists, so a tier-2 leaf reads a few hundred KB and answers in ~20 to 30 ms: the long tail is cheap exactly for the queries that need it.
- For a query not predicted rare, tier 2 is consulted only when tier 1 returns too few good candidates. The trigger is fewer than ~50 merged candidates above a quality threshold. This sequential path costs ~40 ms, which the latency budget reserves, and happens for a few percent of queries. Under overload it is the first work to be shed.
- Candidates from both tiers are merged on one scale and re-ranked together. Both tiers use the same scoring function and the same global IDF, so their cheap scores compare directly. A tier-2 page can still win: static rank decided where it is stored, not whether it may rank for a query only it answers.
- Snippets come from the same doc store, and the page is cached like any other. Rare queries rarely repeat, so caching them is cheap insurance rather than a win: they are why the hit rate tops out around 50 to 60 %. Google has said for years that about 15 % of each day's queries have never been seen before.
Deep dives
Inside a shard: the inverted index
What exactly is stored in an index shard, and how do you build tens of terabytes of it every day?
A leaf has to find, among 10 M documents, the few that contain every query term, in a few milliseconds. Scanning text is out of the question; the structure has to be term to documents, and it has to fit in RAM. At ~400 distinct terms per page, 50 B pages is 20 trillion postings. At ~20 bytes each with positions that is 400 TB, too big to hold in memory once, let alone fifteen times.
Two properties make it tractable. Posting lists sorted by doc id become small gaps that compress to a byte or two. And between builds nothing in the main index needs to change in place: a changed page is a new version, and the real-time tier holds it until the next build. An index built offline and never modified can be compressed hard, laid out for reading, and swapped atomically.
- Batch-built immutable shards: delta + varint posting lists, doc ids in static-rank order chosen
- A Lucene or Elasticsearch cluster with per-document updates situational: under a few billion documents, or a vertical index (products, jobs, a site search) where updating any document within a second matters more than the last 4× of compression
- Posting lists in a distributed key-value store keyed by term rejected
- Incremental in-place updates to the main index (Percolator-style) situational: the whole corpus, not only news, must be minutes fresh, and the team can afford transactional incremental processing
The answer: Each tier-1 shard holds ~10 M pages chosen by hash(URL) and is rebuilt from scratch daily by a MapReduce-style job: map tokenises each page and emits ((shard, term), (doc id, frequency, positions)); the shuffle groups by shard and sorts by term and doc id; each reducer writes one shard's posting lists as delta-encoded doc ids in variable-length bytes, with skip pointers every 128 postings. Group varint, described in Jeff Dean's 2009 WSDM talk, packs four integers behind one tag byte and decodes faster than plain varint, which matters because decoding is a large share of leaf CPU. Doc ids are assigned in descending static rank, so the best pages sit at the front of every list and pruning can stop early. Beside the postings the shard stores a term dictionary, per-document scoring data (static rank, field lengths, ~100 quantised features, an embedding, the URL fingerprint) and global term statistics, all as one immutable, versioned artefact of ~25 GB. Changes between builds ride the real-time tier and a tombstone list. Tier 2 uses the same format on SSD and rebuilds weekly.
Walk me through how a two-term AND query uses the posting lists.
Start from the rarer term's list, because it bounds the result. For each doc id in it, advance the other list to the first id at or after it, using skip pointers to jump 128 postings at a time instead of decoding each one. A match is scored; a miss moves on. With "jaguar" in 50 k documents of a shard and "speed" in 1 M, the work is about 50 k skip-and-compare steps, not a million decodes. WAND pruning on top skips documents whose best possible score cannot reach the current top 10.
Why assign doc ids by static rank rather than crawl order?
Two wins at once. Delta compression needs ids sorted, and rank order is as good a sort as any. And because the best pages come first in every list, a leaf that has ten good matches can stop once the remaining documents' upper-bound scores cannot beat them, which cuts work on common terms by an order of magnitude. Crawl or URL order compresses slightly better, since pages from one site cluster, but gives up early termination.
A page is taken down by court order an hour after tonight's build. What happens?
Its URL fingerprint goes onto a removal list that every leaf, doc server and root reloads every minute, and the root filters final results against it, so the page is gone within minutes without rebuilding anything. The front end applies the same list to cache hits, so a cached page cannot bring it back. The list also feeds the next build as a blocklist, so a re-crawl does not reintroduce it.
How long does the daily build take, and what if it fails halfway?
Hours on a few thousand machines; the shuffle for tier 1 is tens of TB. Each shard is an independent reduce partition, so a failed task is retried on its own, which is the normal MapReduce behaviour. If the whole build is late or fails validation, serving is unaffected: yesterday's version keeps serving and the real-time tier, which keeps 48 hours, covers the extra day.
How do you support exact phrase queries?
Positions in the postings. Intersect the terms' lists as usual, then check that the candidate's positions are consecutive. Positions are about half the index size, which is the price of phrases and proximity scoring. For very frequent phrases like "new york", indexing the word pair as its own term trades a little index size for much faster lookups.
Document- vs term-partitioned index
Every query fanning out to all 1,000 shards sounds wasteful. Why not partition by term, so a three-word query touches three shards?
There are two natural ways to split an index. By document: each shard is a complete small index over a thousandth of the pages, and every query visits every shard. By term: each shard owns the complete posting lists for some terms, and a query visits only the shards that own its terms. At 50 k queries a second reaching the index, document partitioning means 50 M leaf requests a second; term partitioning means about 150 k.
The cost that matters is work and balance, not request count. Posting lists follow Zipf's law: "new" or "the" appear in billions of pages, and a multi-term query has to bring its lists together somewhere. Under document partitioning that happens in local memory on every shard in parallel. Under term partitioning it means shipping lists between machines, and the machines that own popular terms carry far more load than the rest.
- Document-partitioned: each shard indexes 1/1,000 of pages, all terms chosen
- Term-partitioned: each shard owns all postings for a range of terms rejected
- Selective search: cluster pages by topic and route each query to a few clusters situational: the cost of the long tail dominates and some recall loss is acceptable, for example inside a low tier or an archive
- Replicate the whole index on every machine rejected
The answer: Partition by document: shard = hash(URL fingerprint) mod 1,000 in each tier, so every shard holds a random sample of pages and does the same work for every query. Each leaf intersects, scores and keeps its local top 10 independently with no cross-shard traffic, using global IDF from query understanding that the root sends with the query, so scores compare across shards. The price is the 1,000-way fan-out, which is paid down with an aggregator tree (25 × 40), replicas, hedging and deadlines, and halved by the result cache. Jeff Dean's 2009 WSDM talk describes Google serving the same way: document-partitioned index shards, each replicated, behind a tree of parent servers. The costs of document partitioning are latency and overhead, which engineering can bound; the costs of term partitioning are data movement and imbalance, which it cannot.
Are scores from different shards actually comparable?
Only if they use the same statistics. BM25 weights each term by IDF, and a shard's local document frequency differs from the global one, a lot for rare terms. So every shard scores with global document frequencies, computed by the build and sent with the query. With 10 M random pages per shard the numbers are close for common terms, but for a term in 3,000 pages the difference between a shard holding 1 and one holding 7 would otherwise reorder results.
Each leaf returns only its top 10. Could you miss something that belongs in the global top 10?
No, if scoring is the same everywhere: every member of the global top 10 is also in the top 10 of its own shard. For the re-ranker's 500 the question is whether one shard holds more than 10 of them; with pages placed at random a shard expects half of one, and holding 11 is vanishingly rare. Aggregators keep 50 of their 400 for the same reason: each expects 20 of the 500.
Didn't the original Google partition by term?
In a sense: Brin and Page's 1998 prototype stored its index in barrels, each holding a range of word ids. It indexed about 24 M pages on a few machines, where moving posting lists around was cheap. At tens of billions of pages and thousands of machines, data movement dominates, and Dean's later talks describe document-partitioned shards with replicas.
"the" appears in nearly every page. What does a leaf do with it?
Keep it, because phrases like "the who" need it, but it barely affects scoring because its IDF is near zero. With doc ids in rank order and an upper bound on what "the" can contribute, WAND finds the top 10 among the first few thousand postings and stops. For phrases, frequent word pairs indexed as their own terms avoid walking the full list at all.
The corpus grows 30 %. How do you add capacity?
Let documents per shard grow while they fit the memory budget, which needs no topology change. When they no longer fit, raise the shard count at the next build: shards are rebuilt from scratch, so resharding is a build parameter, not a live migration. More shards widen the fan-out, so the tree gets a few more aggregators and the tail-latency defences matter a little more.
Tail latency at a 1,000-way fan-out
Your leaves have a p99 of 40 ms and a query waits for 1,000 of them. What is the query's latency, and what do you do about it?
If each leaf is slower than 40 ms 1 % of the time, the chance that all 1,000 are fast is 0.99^1000, about 0.004 %. So essentially every query waits on at least one leaf's tail, and the query median becomes something like the leaf p99.9. Leaves are slow for reasons unrelated to the query: garbage collection, background compaction, a noisy neighbour, a machine being drained.
Two facts give a way out. Slowness is mostly transient and specific to one machine, so a second replica asked a little later is very likely fast. And with document partitioning a missing shard costs a sliver of recall rather than the answer, so the query does not have to wait for everyone. Dean and Barroso's "The Tail at Scale" (2013) collects the techniques that follow from these two facts.
- Hedged requests at the leaf p95, a hard deadline with partial results, replicas, and an aggregator tree chosen
- Wait for every leaf rejected
- Send every request to two replicas and take the first situational: a small fan-out where capacity is cheap relative to latency, such as the two canary leaves
- Make leaves faster: over-provision until the leaf p99.9 meets the budget situational: as part of the answer (keep leaves near 60 % CPU, isolate them from batch work), never as the whole answer
The answer: Each shard has ~5 replicas per region and aggregators send each request to the replica with the fewest outstanding requests. If a leaf has not answered in 20 ms, about its p95, the aggregator sends a backup to another replica and cancels the loser on the first answer, with hedges capped at 5 % of requests so they cannot snowball under load. At 60 ms the aggregator returns what it has with a count of shards answered; the root serves when coverage is at least ~97 % and otherwise the front end retries in another region. Every query goes to one or two canary leaves before the full fan-out, so a query that crashes leaves takes down two machines, not a thousand. Replicas that are persistently slow go on probation (shadow traffic only) until they recover. "The Tail at Scale" reports that hedging after 10 ms cut the p99.9 of a 100-server BigTable fan-out from 1,800 ms to 74 ms for 2 % more requests; the same idea is what lets a 1,000-way fan-out fit inside 200 ms.
Why hedge at 20 ms rather than immediately, or at 40 ms?
Immediately is duplicating every request, which doubles leaf load. At 40 ms, the p99, the backup starts too late to finish before the 60 ms deadline. Around the p95 only 5 % of requests get a backup, and one that starts at 20 ms on a healthy replica finishes by about 30 ms. The threshold should follow the measured leaf distribution rather than be a constant someone typed in.
During a traffic spike every leaf slows down. Don't hedges make it worse?
They would: everything crosses the threshold at once and load jumps when there is no headroom. So hedges are capped per aggregator, and past the cap the aggregator stops hedging and lets the deadline do its job. Tied requests, queued on two replicas where whichever starts first cancels the other, avoid duplicate work when queues are long. Under real overload, shed by value: tier-2 fall-through first, then the re-ranker falls back to the cheap order, and only then reject queries.
What is a query of death, and how do canaries help?
A query that crashes or hangs the leaf process, such as a pathological term or a parser bug. Every leaf runs the same code on the same input, so a full fan-out kills every shard at once, and each retry does it again. Sending the query to one or two leaves first and fanning out only if they answer confines the damage to two machines, and the root then blocks the query. It costs one leaf round trip, ~10 ms, and Dean and Barroso note that Google's large fan-out search systems use canaries on every request for this safety.
How do you know partial results are not quietly hurting quality?
Measure coverage on every query and alert when the share of queries below 99.5 % coverage rises. Run a small sample of queries with no deadline in the background and compare their top 10 with what was served. If coverage falls because of one rack or a bad build, it shows as a step change long before anyone notices worse results.
A region loses a third of its leaf machines. What happens?
Replicas absorb it first: each shard keeps about three of its five, now at higher CPU, so latency rises and hedges hit their cap. A shard that loses every replica in the region costs 0.1 % coverage and the page still serves. When the load exceeds what is left, global load balancing moves traffic to the other two regions, each sized to take half the global peak, which is why there are ~15 replicas per shard rather than the 10 throughput needs.
Two-phase ranking
Why not run your best ranking model on every document that matches?
A common query matches millions of pages across the index. The learned model that produces good rankings uses a few hundred features and costs ~50 µs a document. Running it on 10 M matches is 500 CPU-seconds per query; at 50 k queries a second that is 25 M cores. Even BM25 over every match is too much without pruning.
Relevance is very unevenly distributed: almost every match is obviously worse than the best few hundred, and a cheap score is enough to tell which. So ranking is a funnel: cheap scoring with pruning where the data lives (the leaves), the expensive model on a few hundred survivors at the root, and the expensive per-result work (snippets) on the final ten.
- Cheap score on leaves (BM25F + proximity + static rank), learned re-ranker on the merged top 500 chosen
- Learned model on every leaf over all matches rejected
- BM25 and PageRank only, no learned stage rejected
- A third stage: a neural model over query and passage text on the top 50 situational: the quality gain on hard queries justifies accelerators on the hot path; a reasonable extension inside the same 30 ms
The answer: Phase 1 on each leaf: pruned retrieval over rank-ordered postings, scored with BM25 over fields (title, body, anchor text, URL), term proximity and static rank (PageRank, spam and quality scores). Each leaf returns its top 10 with ~100 quantised features (~200 bytes), aggregators keep their top 50, and the root keeps the best 500 by phase-1 score. Phase 2 at the root: a learned model, gradient-boosted trees or a small network, over a few hundred features, adding the document's click history, freshness and the similarity between a query embedding and document embeddings stored at index time, in ~25 ms for 500 candidates. Then rules: at most two results per host, near-duplicates collapsed, removals filtered. Snippets are generated last, for the final ten only. The model is trained offline on human relevance ratings and position-corrected clicks, and ships only after it wins an online experiment.
How do you know a new ranking model is better?
Two gates. Offline: NDCG@10 on queries with human relevance ratings, broken down by slice (navigational, rare, fresh, each language) so an average gain cannot hide a regression. Online: an A/B test on a small share of traffic measuring clicks on top results, long clicks against quick returns to the results page, abandonment and reformulation. A model ships when it wins online, not when it wins offline.
The best page is 600th by cheap score and never reaches the re-ranker. How would you detect and fix that?
Detect it by sampling queries and re-ranking a much deeper candidate list offline, say 5,000 instead of 500, then measuring how often the deep list's top 10 contains pages the 500 missed: that is phase 1's recall loss. Fix it by moving the missing signal into phase 1 (often anchor text or a cheap embedding score), or by raising the candidate count for the query classes where it pays, such as long queries.
Why return features with the candidates instead of letting the re-ranker fetch them?
Many features exist only on the leaf: field matches, positions and proximity come from the postings it just walked. Recomputing them at the root would mean shipping documents or a second round trip to 1,000 shards. Two hundred bytes for each of 10,000 candidates is ~2 MB per query, spread across 25 aggregators, which is cheaper than an extra round trip in a 200 ms budget.
Clicks are your training signal, but users click what you put on top. Isn't that circular?
Yes, that is position bias. Estimate how likely each position is to be clicked regardless of relevance, from occasional result-swapping experiments, and weight clicks by the inverse. Use pairwise signals too: a click on result 3 after skipping 1 and 2 says 3 beat them. Human ratings anchor the loop so it cannot drift into ranking whatever it already ranks.
How does PageRank actually enter the score?
As a static, query-independent feature computed offline from the link graph: a page is important if important pages link to it, iterated until the scores settle. It decides which tier a page lives in and its doc id order, and it is a phase-1 feature that breaks ties between textually similar pages. On its own it would rank the biggest sites first for everything; it is a prior, not a relevance score.
Tiered index: the best pages first
Most queries are answered by the best few billion pages. Why pay to search all 50 billion every time?
Holding all 125 TB in RAM at 15 replicas is ~1.9 PB of index in memory, and every query would walk five times as much posting data per shard as it does over 10 M documents. Yet for most queries the top 10 come from a small, high-quality part of the web: popular sites, well-linked pages, recent news. The long tail matters only when a query is rare enough that the good pages do not answer it.
Static rank, which the build computes anyway, predicts well which pages can ever reach a top 10. Splitting by it lets the hot part of the index have RAM and many replicas, and the cold part SSD and few. The risk is a query whose best answer is in the cold part, so the design needs a reliable, measured rule for when to look there.
- Two tiers by static rank: 10 B in RAM, 40 B on SSD, tier 2 only when tier 1 will come up short chosen
- One tier, everything in RAM rejected
- One tier, everything on SSD situational: a much lower query rate, or an archive or research index where 100 ms at the leaf is fine
- Early termination inside one tier only (rank-ordered postings) situational: always, inside each tier: it complements tiering rather than replacing it
The answer: Tier 1 holds the top ~10 B pages by static rank (PageRank, quality and spam scores, plus a guaranteed place for each site's home page and most-visited pages) in RAM across 1,000 shards × ~15 replicas, rebuilt daily. Tier 2 holds the other ~40 B on local NVMe SSD across 1,000 shards × ~6 replicas, rebuilt weekly. The root queries tier 2 in parallel when global document frequencies predict fewer than ~100 tier-1 matches, and afterwards when tier 1 returns fewer than ~50 candidates above a quality threshold; together that is ~15 % of index queries. Both tiers score on the same scale, so candidates merge and re-rank together. Tier membership is recomputed every build as static rank changes, and the real-time tier covers new pages until then. The fall-through rule is tuned per query class against a shadow sample that always searches both tiers.
Tier 1 returned ten decent results, so you skipped tier 2. What if the best answer was there?
That is the cost of the design, and it is measured: a small sample of queries always searches both tiers in shadow, and I count how often a tier-2 page would have made the final top 10 when the rule skipped it. If that rate is meaningful for a class of query, say long technical queries or a smaller language, I lower that class's threshold. The rule is a per-class policy tuned on data, and its miss rate is a tracked quality metric.
How does a page move from tier 2 to tier 1?
At build time, when its static rank rises: new links, more clicks, a site becoming popular. Tier membership is recomputed every build, so promotion takes at most a week for tier 2's schedule. A new page that suddenly matters is covered sooner by the real-time tier, which accepts documents regardless of tier.
Why two tiers and not four?
Each boundary is a decision that can be wrong and a sequential step that costs latency. Two captures the big split: memory against SSD, often useful against rarely useful. A third archival tier on cheap disk, for pages not seen in years and only for exact-match queries, becomes worth it if the corpus grows to hundreds of billions.
Doesn't SSD hurt latency for the queries that reach tier 2?
Less than it seems. Queries get there because their terms are rare, and rare terms have short posting lists: a few hundred KB per shard instead of tens of MB, a handful of SSD reads at ~100 µs each. A tier-2 leaf answers in 20 to 30 ms. A common-term query would be slow on SSD, but those are exactly the queries tier 1 answers.
Freshness: a real-time tier and an honest cache
An earthquake happened four minutes ago. Your index was built last night and half your answers come from a cache. How does the news show up?
Two parts of the design work against freshness. The batch index is a day old by construction, because re-inverting 10 B pages takes hours. And the result cache, which halves index load, stores pages that were right when they were cached. For "python sort list" a six-hour-old page is perfect; for "earthquake" during an earthquake it is useless, and that query may be the busiest in the world for the next hour.
Fresh content is a small fraction of the corpus: ~1 k documents a second, ~170 M over two days, against 50 B. And only some queries need it, which the system can detect from bursts of new documents and searches. So freshness can be a small, fast, separate index plus a cache that decides per query how stale it may be, instead of making the whole system fast.
- A small real-time tier merged at query time, freshness-classed TTLs, and burst-based invalidation chosen
- Rebuild the main index more often rejected
- Incremental updates to the main index situational: the goal is freshness for the whole corpus, not just news; Google's 2010 Percolator paper describes moving its indexing to incremental processing
- A short TTL for every cached query, or no cache rejected
The answer: The crawler's priority lanes publish fresh pages to a stream keyed by URL fingerprint; real-time leaves index them into in-memory segments searchable every ~5 s, with static quality estimated from the host and early links. The tier keeps everything newer than the live batch version's cutoff (~48 h for a daily build), every cache miss queries it beside tier 1, and when both have a page the newer copy wins. It also counts terms in fresh pages per 5-minute window against each term's baseline and pushes bursting terms to the front ends. A query is fresh-class if it contains a hot term or its history says users want recent results: such pages are cached for 60 s with stale-while-revalidate and get a recency boost in ranking, while others are cached for hours, bounded by the index version, with jitter. A cache entry created before a burst on any of its terms is treated as a miss, and single-flight per key means the busiest query costs the index about one request a minute per region.
A fresh article has no links yet. How does it outrank an authoritative page from last year?
For a fresh-class query recency is a strong feature, and the article inherits trust from its site, which has a long history. The re-ranker learns from clicks how much recency matters per query class, so "earthquake" during an earthquake weights the last hour heavily and "earthquake magnitude scale" does not. Early signals, such as a link from the site's home page within minutes, stand in for PageRank until the batch build computes it.
The single-flight request for a hot key fails. Do 20,000 waiters all hit the index?
No. Waiters fall back to the stale page, which for a hot key almost always exists, and the next attempt starts after a short backoff. Stale-while-revalidate refreshes the hottest keys before they expire, so a failure costs a few seconds of freshness, not load. Each front end also keeps hot keys in local memory, so the cache cluster never sees 20 k requests a second on one key either.
Why not just invalidate every cached query containing "earthquake"?
Because the cache cannot find them cheaply: keys are hashes, and a term-to-keys index over 100 M entries is a second large system to keep consistent. Checking at read time is free: each entry stores its terms and creation time, and the hot-term table has a few thousand rows. Entries nobody asks for again simply expire.
An article is updated with a correction. Which version do users see?
The newest. The stream is keyed by URL fingerprint, so versions arrive in order and the real-time leaf replaces its copy; the doc store overwrites the snippet source; and at the root the real-time copy beats the batch copy of the same page. A cached page with the old snippet lives at most 60 s for a fresh-class query.
How do you measure freshness?
End to end, from fetch time to the first time a page is served in results, using canary pages on our own sites that publish known content on a schedule. Also the age of the newest result for fresh-class queries and the cache's stale-serve rate. If the real-time tier falls behind, its consumer lag on the stream is the first number to move.
The theory behind it
- Tail latency and hedged requests: an interactive playground. Fan one request out to 100 backends and watch a 1% stall become the typical case, then hedge slow calls and see what it costs in load.
- Inverted index and search ranking: an interactive playground. Search sixteen reviews through an inverted index: see posting lists, AND and OR queries, TF-IDF ranking and why common words cost the most.
- Web search engine architecture: How a Google-style web search engine fits together: crawling, document processing, building inverted indexes, document-partitioned versus term-partitioned indexes, tiered indexes, query serving with scatter-gather, ranking signals including links, freshness, caching and the numbers involved.
- Search, indexing and autocomplete: Inverted indexes, analysis and ranking, prefix search with tries, filtering by permissions, index freshness, and when vector search actually belongs in the answer.
- Search ranking and relevance: How search results are ordered: retrieval then ranking, text relevance with BM25, business and personal signals, learning to rank, semantic and hybrid search, query understanding, evaluation with NDCG and click metrics, and serving ranking within a latency budget.