LearnHLDDesign a web search engine

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.

Step 1: Understand the problem

Search is the question where scope discipline earns the most points, because the full problem is a decade of work and the interviewer knows it.

You askThey sayWhat it settles
How big is the corpus, and how many queries a second?Ten billion documents, tens of thousands of queries a second.That you cannot look at the corpus. Everything on this page exists to avoid touching all but a tiny fraction of it per query.
What is the latency budget?200ms at p99, end to end.A fan out to a thousand shards where the slowest one decides your latency. It is also what makes deadlines a query parameter rather than a constant.
How fresh does the index have to be?News in minutes, everything else in hours.Two indexing paths, not one. A small fast path for recent documents and a slow bulk path for the rest, merged at query time.
Are we building ranking, or retrieval?Retrieval, with simple scoring. Ranking quality is a separate team.The biggest cut on the page. Say it early: you are designing the system that finds candidate documents fast, and leaving what order they finally appear in to a layer above.
Do queries need phrases and boolean operators?Phrases yes, complex boolean no.Position lists in the index, which are expensive enough to store separately from the posting lists.
Personalised results?No.Results are cacheable. Without personalisation the same query returns the same top ten for everyone, and a cache absorbs a large share of traffic that would otherwise fan out.

What you are building

In scope
  • 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.
The numbers you commit to
  • 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.
Cut, and say so out loud
  • 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.

Back of the envelope

Index size
10 billion
500
3
Total postings10B x 500 = 5.0 trillion
Index size compressed5.0T x 3B = 15 TB
Machines to hold it in memory at 512GB30
Postings scanned for a two word queryup to hundreds of millions
30 machines
just to hold the index in memory, before serving a single query

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.

Step 2: Propose the high level design

The API

GET/v1/search?q=distributed+consensus&limit=10&cursor=...
returns 200 { results: [...], took: 84, complete: true }
Why: The response says whether every shard answered. A caller that needs completeness, such as an internal evaluation job, can retry, while a user facing page ignores it. Hiding partial results makes debugging a slow shard impossible.
GET/v1/search?q=...&deadline_ms=120
returns 200 with whatever finished in time
Why: The deadline is a parameter, not a constant. A page load wants 120ms and a background quality evaluation can afford 5 seconds, and the same fan out serves both by returning what it has when the clock runs out.
POST/v1/documents
{ "url": "...", "content": "...", "fetchedAt": "..." }
returns 202 { segmentEta: "~90s" }
Why: Indexing is asynchronous and the caller is told roughly when the document will be searchable. Promising immediate visibility would mean writing into a live compressed index, which is exactly what the segment design avoids.

The inverted index

Flip the obvious structure. Instead of a document pointing to its words, a word points to its documents.

posting listone entry per term, sorted by document id
termstringPKAfter tokenising, lowercasing and stemming.
doc_freqintHow many documents contain it. Drives the IDF part of scoring and query planning.
postingsdelta encoded listAscending document ids stored as gaps, so a dense list costs a byte or two per entry.
positionsoptional listWhere in the document. Needed for phrase queries and expensive to store, so often kept in a separate structure.
skip pointerssparse indexEvery few hundred entries, so intersecting a rare term with a common one can jump rather than scan.
Sample row
"quorum" | 84,102 | [17, +4, +23, +2, +981, ...] | ... | every 128th entry
Sorted order is what makes intersection cheap: two sorted lists intersect in a single linear pass, and skip pointers turn that into something closer to a binary search when one list is far shorter than the other. Nearly every optimisation in query serving depends on that ordering.

The whole system on one whiteboard

Two systems live in Figure 1 and they meet in exactly one place. Indexing turns documents into immutable segment files. Serving loads those files and answers queries. Neither one can slow the other down, which is the property you are buying.

Figure 1. A query fans out to every shard and comes back in 200ms. Indexing runs on its own clock and hands over immutable segments. The only thing crossing between them is a file.

Walking Figure 1:

  1. The query is parsed, corrected and expanded at the front end. This is cheap and it is where a bad query gets turned into a good one.
  2. Popular queries never go further. Without personalisation the top ten for “weather” is the same for everyone, and a result cache absorbs a large slice of traffic.
  3. On a miss the root aggregator takes over. It owns the deadline.
  4. It asks every leaf shard, because the index is partitioned by document and any shard might hold a match. A thousand parallel requests, and the slowest one is your latency.
  5. Each leaf intersects posting lists inside its own segments and returns its best few candidates, not its matches. Sending back thousands of document ids would drown the root in network traffic.
  6. The root merges a thousand short lists, keeps the top ten, and only then fetches titles and snippets. Hydrating before merging would mean fetching a thousand times more documents than you show.
  7. Indexing runs completely separately, on its own schedule, feeding on whatever the crawler produced.
  8. The indexer writes segments once and never edits them.
  9. A merger compacts small segments into large ones in the background, and publishing a new segment set is how updates reach serving. Rolling back is pointing at yesterday’s set.
Your answer

Step 4 asks all 1,000 shards on every query. The alternative is to shard by term, so a query only touches the shards holding its words. Which would you pick, and what does the other one cost?

Step 3: Design deep dive

How to shard 100TB of index

This is the question, and both answers sound reasonable until you count round trips.

Index partitioning
Every shard holds a complete index over its own slice of documents, so every shard can answer the whole query locally and return its own top ten. One round trip, perfect parallelism, and each shard scores documents it holds entirely in memory. The cost is that every query touches every shard, so a thousand shards means a thousand requests per query and your p99 is the slowest of a thousand machines. This is what everyone actually runs.

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.

Shard load
13
12
13
13
12
13
13
13
0
1
2
3
4
5
6
7
Shard 2 carries 12.6% (even would be 12.5%)
Every shard answers every query over its own document slice, so load is inherently even regardless of what people search for. Even distribution with no tuning is a large part of why this scheme survives.

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.

1/4 Try the result cache first. Query popularity is extremely head heavy, so a modest cache absorbs a large share of traffic outright. This is also the only layer that can return in single digit milliseconds.

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.

The follow up you will get

“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.

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.

Figure 2. Indexing and serving are separate systems that meet only through immutable segments.

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

Query serving
10k qps
10k qpsone slow shard50k qpshedged and tiered
Healthy. Cache absorbs the head, tier one answers most misses, fan out is fast and the p99 sits comfortably inside budget.

Trade-offs

ChoiceWhat you gainWhat you payPick it when
Shard by documentIntersection 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 indexMost 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 requestsRemoves 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 segmentsNo 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

Interviewer
Index by document or by term across your thousand machines?
The core decision, and both answers can be argued badly.
You
By document. Term sharding touches fewer machines per query, which sounds better, but a multi word query has to intersect posting lists and a common term has hundreds of millions of entries, so you end up shipping hundreds of megabytes across the network per query. Document sharding keeps the intersection local: every shard answers the whole query over its own slice and returns its own top ten. It also spreads load evenly for free, whereas term frequency follows a power law and skews badly.
Names the metric that decides it, bytes moved rather than machines contacted, and adds the skew argument as a second independent reason.
Interviewer
Every query hits a thousand machines. What does that do to your p99?
The tail latency question, which is really about whether you have thought about fan out.
You
It makes the p99 of the query roughly the worst case of a thousand machines rather than the typical case of one. If each shard is slow one time in a hundred, essentially every query hits at least one slow shard. So I would hedge: once most shards have replied, send the outstanding ones to a replica and take the first answer. And at the deadline I return what I have. Missing two shards out of a thousand almost never changes the top ten.
The observation that tail compounds rather than averages under fan out is the senior insight in this whole question.
Interviewer
Returning incomplete results seems dangerous. Justify it.
Pushing to see whether the previous answer was considered.
You
The ranking is already an approximation, so completeness was never exact to begin with. The choice is between a page that is 200ms slower and a top ten that is very occasionally missing a result that was unlikely to be in it. I would still expose completeness in the response, so an internal evaluation job can demand the full set and a slow shard is visible rather than silently degrading quality.
Grounds the trade in what the product actually guarantees, then keeps the signal rather than hiding it.
Interviewer
How do you add a new document without rebuilding the index?
Freshness, and the answer should mention immutability.
You
I do not modify anything. New documents go into a small new segment, and a query searches all segments and merges. A background process compacts small segments into larger ones. Deletes are a bitmap of removed ids applied at query time rather than a rewrite. It is the same idea as an LSM tree, and immutability is what makes replication a file copy and rollback a matter of pointing at older segments.
Connects it to a structure already covered and names three separate benefits of immutability.
Interviewer
Where would you spend effort first if the p99 were 400ms?
Open ended prioritisation, which is what separates levels here.
You
I would measure where the time goes before changing anything, because the three candidates have very different fixes. If it is the tail of fan out, hedging and deadlines. If it is ranking, cut the candidate set or make the model cheaper. If it is a cache miss rate, that is the cheapest win of the three. I would not guess between them, since fan out tail and model cost look identical from the outside and the fixes have nothing in common.
Refuses to optimise without measurement and names why the guess would be unreliable. Honest and specific.

Checkpoint

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?

Worth memorising
  • 10 billion documents, 200ms at p99. You can never touch more than a sliver of the corpus.
  • Shard by document, not by term. Every query asks all 1,000 shards and the slowest one is your latency.
  • Leaves return their best few candidates, not their matches. Otherwise the root drowns in network traffic.
  • Segments are immutable. A rollback is pointing at yesterday’s set of files.
Say this in 60 seconds

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.

IndGeek provides solutions in the software field, and is a hub for ultimate Tech Knowledge.