Design a web search engine
Ten billion documents. A query arrives, and in under two hundred milliseconds you have to find every document containing several words, score all of them, and return the best ten.
You cannot look at ten billion documents in two hundred milliseconds. You cannot even stream them off disk. Everything in this design exists to avoid looking at almost all of the corpus, and the two structures that make that possible are an inverted index and a tiered serving stack.
You shard the index across 1,000 machines. Would you split it by document or by term? Name the query cost of each.
What you are building
- Answer a multi word query from a corpus of billions of documents. Boolean intersection first, then relevance ranking.
- Return the top ten by relevance in under 200ms. End to end, at p99, including ranking.
- Ingest new and updated documents continuously. Fresh content within minutes, not a nightly rebuild.
- Serve tens of thousands of queries per second. With a heavy head of repeated popular queries.
- 10 billion documents, average 500 useful terms each.
- 50,000 queries per second at peak.
- p99 under 200ms including ranking.
- A machine failure must not lose documents from results.
- Crawling. It feeds this and has its own page.
- The ranking model itself. Assume a scorer exists and focus on how the system feeds and budgets it.
- Query understanding, spelling correction and synonyms. Real, large, and separable.
- Personalisation, which changes the caching story enough to deserve its own discussion.
The numbers
Three bytes per posting is not an accident, it is the whole reason this is affordable. Document ids are stored as deltas between consecutive ids and then variable byte encoded, so common terms with dense posting lists compress far below a raw 8 byte id. Doubling that number doubles your fleet.
The inverted index
Flip the obvious structure. Instead of a document pointing to its words, a word points to its documents.
| term | string | PK | After tokenising, lowercasing and stemming. |
| doc_freq | int | How many documents contain it. Drives the IDF part of scoring and query planning. | |
| postings | delta encoded list | Ascending document ids stored as gaps, so a dense list costs a byte or two per entry. | |
| positions | optional list | Where in the document. Needed for phrase queries and expensive to store, so often kept in a separate structure. | |
| skip pointers | sparse index | Every few hundred entries, so intersecting a rare term with a common one can jump rather than scan. |
Deep dive one: how to shard 100TB of index
This is the question, and both answers sound reasonable until you count round trips.
Sharding by document wins, and it wins for a reason worth stating precisely: it keeps the intersection local. Term sharding minimises the number of machines contacted and maximises the bytes moved, and in this workload bytes moved is what costs you.
Deep dive two: the two hundred millisecond budget
A thousand shards is a thousand chances to be slow. If each shard is fast 99% of the time, the odds that all thousand are fast on any given query are effectively zero. Tail latency does not average out under fan out, it compounds.
Two more techniques matter and both are worth naming out loud.
Hedged requests. If a shard has not replied by the time 95% of shards have, send the same request to a replica and take whichever answers first. It costs a few percent extra load and it removes most of the tail, because a slow response is usually an unlucky machine rather than a hard query.
Serve incomplete results. At the deadline, return what you have. A result set missing two shards out of a thousand is almost always identical in its top ten, and a user will never know. Waiting for stragglers to be complete is the wrong trade when the ranking is approximate anyway.
“What if a shard is down?” Each shard is replicated several times, so the root picks a healthy replica. If every replica of a shard is unavailable, the query returns results from the other 999 and records the incompleteness. Refusing to answer because one thousandth of the corpus is unreachable is worse for the user than a very slightly worse result set.
Deep dive three: keeping the index fresh
Posting lists are sorted, compressed and packed. Inserting one document into the middle of a compressed sorted list is close to the worst operation you could ask for.
So do not. Build small immutable index segments continuously, search all of them, and merge them in the background. New documents land in a small in-memory segment that is searched alongside the large ones, and deletes are a bitmap of removed document ids applied at query time rather than a rewrite.
This is the same log structured idea as an LSM tree, applied to an index rather than a key value store, and it is why searching a large corpus means intersecting across several segments and merging results.
Because segments are immutable, a serving shard never modifies anything it reads. That makes concurrency trivial, makes replication a file copy, and means a rollback is pointing at an older segment set.
Break it
The API
{ "url": "...", "content": "...", "fetchedAt": "..." }Trade-offs
| Choice | What you gain | What you pay | Pick it when |
|---|---|---|---|
| Shard by document | Intersection stays local, load is even without tuning, and one round trip answers the query. | Every query touches every shard, so tail latency compounds and fan out cost grows with cluster size. | Essentially always for full text search. Term sharding loses on network bytes. |
| Tiered index | Most queries are answered against a small fraction of the corpus, cutting both latency and fan out. | Two indexes to maintain, and a quality decision about what belongs in tier one. | Any corpus where document quality varies enormously, which is the whole web. |
| Hedged requests | Removes most of the tail caused by unlucky machines rather than hard queries. | A few percent extra load, and duplicate work that must be safe to discard. | Any large fan out where p99 matters more than raw throughput. |
| Immutable segments | No in place mutation, so replication is a file copy and rollback is pointing at older files. | Queries intersect across several segments, and background merging consumes real resources. | Any index that must stay fresh without being rebuilt wholesale. |
Interview replay
Checkpoint
1. Why is term sharding worse than document sharding despite touching fewer machines per query?
2. Each shard is slow 1% of the time and a query fans out to 1,000 shards. What is the practical consequence?
3. Why are index segments immutable rather than updated in place?
The core structure is an inverted index with delta encoded posting lists, sharded by document rather than by term, because a multi word query has to intersect lists and term sharding would mean shipping hundreds of megabytes across the network per query while document sharding keeps the intersection local and the load naturally even. The cost of that choice is that every query touches every shard, so tail latency compounds instead of averaging, and I would handle that with a tiered index so most queries only search a small high quality subset, hedged requests to a replica once most shards have answered, and returning at the deadline with whatever arrived rather than waiting. Retrieval is deliberately cheap and returns about a thousand candidates, and only then does the expensive ranking model run. Freshness comes from small immutable segments merged in the background, with deletes as a bitmap applied at query time, so nothing is ever rewritten in place and replication is a file copy.
