Design a web search engine

I would build an inverted index in immutable segments and split it by page across about 1,000 shards, each with three replicas. Fetched pages arrive on a feed. The builder tokenises them, writes compressed posting lists to the shard that owns the page, and stores titles, text and signals in a doc store. A small fresh tier, flushed every few minutes, makes news searchable in 10 minutes.

Deleted pages sit on a per-shard list until a merge removes them. A query goes to the query service, which checks a result cache that answers about 40% of traffic. On a miss, it sends the query to one replica of every shard with a 150 ms deadline, and sends a hedged copy to another replica if a shard is slow. Every shard returns its best 20 by a cheap score.

The service merges 20,000 into 200, and the re-ranker applies a costly model to those only, which takes 20 ms instead of 1,000 seconds. I chose document partitioning over term partitioning because writes stay local and load stays even. The price is fan-out, so tail latency is the thing to watch.

These numbers are assumptions for the exercise, not facts about a real system:

• The index holds 20 billion pages with 300 distinct words each, and one posting takes about 2 bytes once compressed.
• A shard holds 20 million pages and has 3 replicas, each on its own machine.
• 1 billion queries a day, peak at twice the average, and 40% of queries are answered from the result cache.
• A shard answers in about 5 ms, but 1% of its answers take over 50 ms. The costly ranking model takes 0.1 ms per page. A typical query matches 10 million pages.

The arithmetic:

Index size and shards

20B × 300 words × 2 bytes = 12 TB; 20B ÷ 20M pages = 1,000 shards

12 GB a shard, 3,000 machines with 3 replicas

Queries that reach the shards

1B ÷ 86,400 s ≈ 11,600/s; × 2 peak ≈ 23,000; × 60% not cached

≈ 14,000/s

Load on each machine

14,000/s × 1,000 shards = 14M shard requests/s; ÷ 3,000 machines

≈ 4,700/s each, about 5 ms of work apiece

Chance a query meets a slow shard

1 − 0.99^1000

≈ 99.996%, so nearly every query waits for a slow answer

Cost of the costly model

10M matches × 0.1 ms = 1,000 s; or 200 candidates × 0.1 ms

1,000 s for all matches, 20 ms for the top 200

An inverted index maps each word to a posting list: the ids of the pages that contain it, in ascending order, with how often the word appears and where. To answer 'cheap running shoes', walk the lists for the three words together and keep the ids that appear in all of them. Because the lists are sorted, this is a merge, and skip pointers let it jump over long gaps. This page-id order is the one the full lists always keep, and the ranking section adds a separate short list for early stopping.

Compress the lists. Store the gap between each id and the one before it, as small numbers in a variable-length code. That brings a posting to about 2 bytes, which is what makes 12 TB possible instead of several times more.

Build in batches. Map each page to records of word, page id and count. Sort and group them by word, and write the sorted result as a segment. Segments are immutable: they are written once and never edited, so they can be memory-mapped and read without locks. Each shard builds the segments for its own pages, so a build is 1,000 independent jobs.

A full rebuild of 20 billion pages takes a long time, so freshness needs a second path. Keep a fresh tier next to the base index: a small in-memory index of recent pages that is flushed into a small segment every few minutes. A query searches the base segments and the fresh ones, then merges. News pages reach the fresh tier within the 10-minute target. Everything else waits for the next base build.

Updates and deletes. A changed page gets a new id in the fresh tier, and its old id goes onto a deleted list for that shard. Every query drops ids on that list. A background job merges small segments into larger ones, as a log-structured store does, and drops deleted pages as it goes. Without merging, the number of segments would grow, and each query would have to open all of them.

Publishing. A shard replica loads a new set of segments listed in a versioned manifest and switches to it in one step. Roll the change through the replicas one at a time, so the shard never loses its capacity.

There are two ways to cut an inverted index.

• Document-partitioned. Each shard holds the whole index for a subset of pages, covering every word. A query goes to every shard, and each ranks only its own pages.
• Term-partitioned. Each shard holds the complete posting lists for a subset of words. A query goes only to the shards that own its words.

Term partitioning looks efficient, since a query touches two or three shards instead of a thousand. The costs are heavy. Intersecting a long list for a common word with another list means moving millions of ids across the network. A popular word makes one shard hot. And indexing one page writes to up to 300 shards, one for each of its words. So at this scale, I would choose document partitioning. A page lives in one shard, writes are local, load is even, and losing a machine hides only a slice of the pages.

The price is fan-out: 14 million shard requests a second, and a latency set by the slowest shard. If each shard is slow 1% of the time, then 1 − 0.99^1000, about 99.996%, of queries meet at least one slow answer. Three tools fix this:

• Deadlines. The query service waits at most about 150 ms and then answers with the shards that replied, noting the coverage. Missing a few percent of one query's pages rarely changes the top ten.
• Hedged requests. If a shard has not answered in about 30 ms, send the same request to another replica and take the first reply. It adds a few percent more load and removes most of the long tail.
• Smart replica choice. Send each request to the least busy replica, and avoid one that is merging segments.

Scores from different shards must be comparable. Word rarity depends on how many pages in the whole index contain a word, so compute these counts across all shards and give each shard the same table. Otherwise a word that is rare in one small shard would look more important there than it is.

To cut work, split the index into tiers. Put pages with the best quality scores in a small first tier of shards. Query it first, and go to the larger lower tier only when the first yields too few good results.

Scoring every match with the best model is out of the question. At 0.1 ms a page, 10 million matches would take 1,000 seconds. So ranking runs in two phases, each tuned for a different job.

Phase 1: cheap and wide. Inside each shard, score pages while walking the lists. Use a simple formula based on how often the query words appear in a page, how rare they are, and the length of the page, plus a static quality score computed in advance for each page. Keep the best 20 in a small heap. The full lists stay in page-id order for intersection. For early stopping, keep beside them a short champion list for each word, holding its best few thousand pages ordered by quality and word count. Scan that list first, and stop once the remaining entries cannot beat the heap's lowest score. Fall back to the full lists only if too few pages match. This phase must have high recall: the right answer must land in the top 200 even if it is not first.

Phase 2: costly and narrow. The query service merges the shard lists into the best 200 and sends them to the re-ranker. It loads richer signals for each page in one batch from the doc store: title and heading matches, how the page is linked to, freshness, language, past clicks. A learned model then orders them. At 0.1 ms each, 200 candidates take 20 ms. Pick the number 200 by testing offline: how often does the page that the full model would put first sit within the first 200 of the cheap score?

Then build snippets by choosing the passages of the stored text that contain the query words.

Caching. A small set of popular queries makes up a large share of traffic, so a result cache absorbs about 40%. The key is the cleaned query, the language, the region, the page number and the index version, so a new base index starts a fresh set of entries. Use a short time to live, a minute or two, for queries about news. Allow hours for stable ones. Let one request rebuild an expired popular entry while the others wait, so a burst does not hit the shards at once.

Keep a cache of hot posting lists in memory on each shard too. Lists for common words are read by almost every query.