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.
Web search engine / at a glance
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.
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 ask
They say
What 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.
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.
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.
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.
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.
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.
query1seen it?2on a miss3fan out to all4posting lists5and the fresh onehydrate top 1067write once8publish9UserQuery front endparse, spellcheckResult cachetop queriesRoot aggregatorfan out, mergeLeaf shardsx1,000, by documentSegmentsimmutable filesDocument storetitles, snippetsCrawlera page of its ownIndexerbuilds segmentsMergercompacts segmentsFresh indexminutes old
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:
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.
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.
On a miss the root aggregator takes over. It owns the deadline.
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.
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.
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.
Indexing runs completely separately, on its own schedule, feeding on whatever the
crawler produced.
The indexer writes segments once and never edits them.
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?
This is the question, and both answers sound reasonable until you count round trips.
Index partitioning
1full query2full query345QueryRootfans out to allShard 1docs 0-10MShard 2docs 10-20MMerge top 10from each
1full query2full query345QueryRootfans out to allShard 1docs 0-10MShard 2docs 10-20MMerge top 10from each
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.
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.
Result cachehead queriesTier 1best 1% of docsTier 2the restRerank top 1,000expensive model
Result cachehead queriesTier 1best 1% of docsTier 2the restRerank top 1,000expensive model
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.
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.
12write once3load and serve4CrawlerDocument storeIndexerbuilds segmentsSegmentsimmutableServing shardsx1,000
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.
fan out to allQueriesRootfan outShardsx1,000Ranking
fan out to allQueriesRootfan outShardsx1,000Ranking
Healthy. Cache absorbs the head, tier one answers most misses, fan out is fast and the p99 sits comfortably inside budget.
Read every step as text
10k qps · Healthy. Cache absorbs the head, tier one answers most misses, fan out is fast and the p99 sits comfortably inside budget.
one slow shard · Degraded. One machine out of a thousand starts running slow, perhaps mid compaction. Because every query touches every shard, that one machine is now in the critical path of every single query and the whole p99 moves.
50k qps · Down. At peak, fan out multiplies. 50,000 queries a second across 1,000 shards is 50 million shard requests a second, and the root is now doing more connection management than search. Fan out cost is superlinear in a way raw query rate hides.
hedged and tiered · Healthy. Tier one answers most queries against a much smaller fan out. Hedged requests remove the slow shard from the tail by asking a replica. And at the deadline the root returns what it has rather than waiting, so one bad machine costs a result nobody notices instead of a slow page everybody notices.
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.
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.