Distributed Search

It is midnight, and a shopping app's Diwali sale has just opened. Four thousand shoppers a second type into the search box: "red running shoes", "phone under 15000". The search box runs WHERE title LIKE '%shoes%' on the product database, and every query reads all 50 million listings. Within two minutes the database's CPUs hit 100%, searches take ten seconds, and checkout, which shares the same database, starts timing out too. The busiest hour of the sale is spent on a spinner. What the team needed was a search index: a separate copy of the data, built only for finding things.

The database failed for a simple reason: a pattern that starts with a wildcard cannot use a normal index, so it scans every row and returns matches in no useful order. A search system solves three problems that a plain query does not. It finds documents containing the words quickly, it matches words written differently (Shoes, shoe, running and run), and it puts the best matches first. This chapter builds one for that shopping app.

Here is what shoppers expect, and what each expectation demands of the system.

Every row is a number or a property you can test, and together they shape the rest of the chapter.

Search has its own vocabulary. These terms appear throughout the chapter, so fix them now.

Two of these carry the whole design: an index is built from terms, and each term points to its postings.

Users also mix text with exact filters. Someone who types "shoes" and ticks "under ₹2,000" and "size 9" wants word matching plus precise conditions. So the index stores some fields, like price, size and brand, as plain values that can be filtered and sorted, not only as words.

Keep one picture in mind: documents go in at one end, get broken into terms, and are stored so that any term leads straight to the documents that contain it. The next page shows that structure.

The index at the back of a textbook maps a word to the pages it appears on. Search engines do the same. Instead of storing each document's words (document → terms), they store, for each term, the list of documents containing it (term → documents). This is the inverted index, and each list is a postings list.

To answer red shoes, the engine fetches two lists, red → 1, 3 and shoe → 1, 3, 4, and intersects them. Both lists are sorted by document id, so one pass with two pointers finds the common ids, 1 and 3, without ever touching document 2 or 4.

Real postings carry more than ids. They record how often the term appears in each document, for ranking, and where it appears, for phrase queries like "running shoes" where the words must sit side by side. Lists are kept sorted and compressed, often as gaps between ids (1, 3, 4 is stored as 1, 2, 1), because small numbers take fewer bytes.

The index on the previous page stored run and shoe, not "Running-Shoes". Before text enters the index it passes through an analyser: a fixed chain of steps that turns raw text into terms. The same chain must run on every query too, or the query's terms will not match the stored ones.

Each optional step helps some searches and hurts others.

Every choice trades recall, finding more matches, against precision, keeping them relevant. Languages differ too: Hindi, Tamil and other Indian languages need their own tokenisers, and many users type Hindi in Latin letters, such as "joote" for shoes, so an app used across India may index the same text with more than one analyser.

Finding matches is half the job. A query for shoes may match 3 million listings, and users look at the first ten. Ranking gives every match a score and returns the highest. The classic text signals rest on two simple ideas.

• Term frequency (TF): a listing that mentions trail three times is probably more about trail running than one that mentions it once.
• Inverse document frequency (IDF): rare words matter more, because far fewer documents contain them. In a catalogue of 1 million listings:

A match on trail counts about 50 times more than a match on the: rare words carry the meaning. A document's text score adds up TF × IDF over the query's terms. BM25, the default in many engines, refines this in two ways: each extra mention adds less than the one before, and long documents do not win just because they hold more words. Two settings control it: k1, often about 1.2, sets how quickly extra mentions stop counting, and b, often 0.75, sets how strongly long documents are pulled back.

Text match is rarely enough on its own. Good ranking mixes in freshness, popularity (clicks, sales, ratings), quality (complete listings, trusted sellers) and sometimes the user's history. Large systems rank in two stages: a cheap score picks a few hundred to a few thousand candidates, then a slower machine-learned model re-ranks only those, because running it on 3 million matches would blow the latency budget.

One index on one machine works until the sale. Fifty million listings and their index take a couple of hundred gigabytes, and the traffic reaches thousands of queries per second, too much for one machine. So the index is split into shards, and there are two ways to cut it.

Document partitioning is the usual choice. Sending every query to every shard sounds wasteful, but each shard searches only its slice, all of them work in parallel, the load stays even, and a new document touches exactly one shard. Documents are placed by hashing their id, so shards stay close in size and their word statistics, such as IDF, come out nearly the same.

There is one exception. When every query stays inside one tenant, such as a seller searching their own 30,000 orders in a seller dashboard, route documents by tenant id instead, so each query visits one shard rather than all of them. A few huge tenants would break the balance, so give any tenant above a size limit several shards of its own and scatter only across those.

Splitting decides how the data is cut; sizing decides how many machines hold it. Think of shards as shelves and replicas as shop counters: more stock needs more shelves, more customers need more counters.

A shard should be small enough that its index sits in memory or on fast SSD, often tens of gigabytes, because a query that waits on slow disk misses its budget. Each shard then gets replicas, full copies on other machines, and any copy can answer any query for that shard. For the shopping app, under stated assumptions:

• Data: 50 M listings × 2 KB of text ≈ 100 GB; assume the index is about the same size.
• Shards: at about 33 GB each, 3 shards, the three you will see in the next scene.
• Traffic: a peak of 4,000 queries a second, and every query visits every shard. If one copy handles about 500 a second, each shard needs 8 copies.
• Failures: spread copies over three zones. To lose a zone at peak and keep 8, run 12 copies, 4 per zone: 36 shard copies in all.

Shards set how much data you can hold; replicas set how many queries you can serve. Adding shards adds no query capacity, because every query still visits every shard, and adding replicas adds no room for data. The advice flips at the extremes: hundreds of tiny shards make each query pay per-shard overhead and lengthen the tail, so keep the shard count as low as memory allows. How does one query use all those copies? That is the next page.

A query first reaches a coordinator, a service that knows where every shard lives from a shard map kept in a coordination service (the coordination chapter). It sends the query to one copy of every shard at once: the scatter. Each shard scores its own matches and sends back only its best 10. Collecting those replies is the gather: the coordinator merges the short lists and keeps the overall top 10.

Sending only 10 per shard is safe, because the overall top 10 must be among the shards' top 10s: with 3 shards the coordinator merges 30 candidates, not millions. It is also why deep pages are expensive: page 500 (results 4,991 to 5,000) forces every shard to send its top 5,000.

A query is only as fast as its slowest shard, because the overall top 10 is unknown until every shard answers. Replicas do double duty: a hedge to a second copy cuts the slow tail, and the same copy takes over when a shard dies. The next page puts numbers on both.

The scene had three shards. Real clusters have 20 or 100, and waiting for all of them is like a group trip that leaves only when the last friend arrives: the more friends, the more often someone is late. If each shard copy is slow on 1 query in 100, the chance that at least one of n shards is slow is 1 − 0.99ⁿ: about 10% for 10 shards, 18% for 20, 40% for 50 and 63% for 100. Rare slowness on single machines becomes common slowness for users.

A hedged request attacks this directly. The coordinator waits about as long as a shard's p99, say 50 ms, and if a copy has not answered, it sends the same query to another copy and takes the first reply. Suppose each copy is slow 1% of the time, independently. A shard now stays slow only if both copies are: 0.01 × 0.01 = 1 in 10,000. Across 50 shards, 1 − 0.9999⁵⁰ ≈ 0.5% of queries meet a slow shard, down from 40%. The cost is small: only the 1% of requests that pass 50 ms get a second copy, so about 1% more load.

The advice flips when slowness is not independent. A query that is heavy on every copy, or a cluster that is overloaded everywhere, gains nothing from a hedge and gets extra load at the worst moment, so cap hedges at a few percent of traffic. Behind the hedge sits a deadline: after, say, 150 ms, the coordinator returns whatever has arrived and marks the result partial.

A fast cluster is useless if its data is old. There are two ways to get documents into the index, and many real systems use both.

Batch builds give a clean, tested index; incremental updates give freshness. Incremental indexing usually works in segments. New documents collect in a small in-memory buffer, and every second or so the buffer is written out as a small, immutable segment that searches can see. A query searches every segment of a shard, and a background job merges small segments into bigger ones so the count stays low.

Because segments never change once written, an update is a delete plus a re-add. The old copy is marked deleted in a small bitmap and dropped for good when its segment is merged. This keeps reads simple and makes segments easy to cache and to copy to replicas.

Users spot stale search quickly: a sold-out phone still listed, a price that changed an hour ago, a deleted post still showing. Freshness comes from a pipeline that follows the database closely.

• Capture every change: put each insert, update and delete on a queue, either from the application or by reading the database's own change log (change data capture). As the message queue chapter showed, the queue absorbs bursts.
• Apply changes in order per document: tag each change with the document's version number, and have the indexer ignore anything older than what the index already holds. A late, out-of-order message then cannot undo a newer one.
• Refresh often, not instantly: making new segments visible every second is cheap. Doing it on every single write would create thousands of tiny segments.
• Separate fast fields from slow ones: price and stock change every few minutes, titles rarely. Some systems keep fast-changing values in a separate store that ranking reads at query time, instead of re-indexing the whole document.

When the analyser or the document shape changes, incremental updates are not enough, because every old document was analysed the old way. Build a new index next to the live one from the database, replay the changes that arrived meanwhile, compare results, then point the search alias (a name that refers to one index) at the new index. Rolling back means pointing it back.

Keyword search fails when the words differ but the meaning matches. A shopper who types "shoes for standing all day" wants the listings that say "cushioned sole" and "arch support", yet none of them contains the word "standing". Vector search handles this. A machine-learned model turns each text into an embedding, a list of a few hundred numbers, often 768, placed so that texts with similar meanings get nearby lists.

Every document is embedded once, at index time. Each query is embedded when it arrives, and the engine returns the documents whose vectors sit closest to it. Comparing the query with all 50 million vectors would take about 38 billion multiplications (50 M × 768), far too slow at thousands of queries a second.

So engines build an approximate nearest neighbour (ANN) index. HNSW links each vector to a few dozen neighbours in a layered graph and walks greedily towards the query. IVF groups the vectors into thousands of clusters and searches only the few clusters nearest the query.

Approximate means some true matches are missed. Recall is the share of the true top 10 that the index returns. Every ANN index has a knob that trades speed for recall: how many graph nodes HNSW explores, or how many clusters IVF opens. Turn it up and recall rises towards 100% while each query slows; a common target is about 95%, measured on real queries.

Here is what those vectors cost in memory for the shopping app.

Vectors are memory-hungry: about 3 KB each at full precision, and ANN indexes are fastest when the vectors sit in RAM. Storing each number in 1 byte cuts memory to a quarter at a small cost in recall; many engines win that back by re-scoring the best few hundred candidates with the full vectors kept on SSD.

Vectors are weak where keywords are strong. A shopper who pastes a model number or a seller's product code wants that exact item, and an embedding may happily return a similar but wrong one. Brand names, sizes and rare words behave the same way. So most systems run both searches side by side and merge the two lists: hybrid retrieval.

Merging is harder than adding scores, because a BM25 score and a vector similarity live on different scales. Reciprocal rank fusion (RRF) ignores the scores and uses only positions: a document earns 1 ÷ (60 + its rank) from each list it appears in, and the totals decide the order. A listing ranked 1st by keywords and 3rd by meaning scores 1/61 + 1/63 ≈ 0.032, ahead of one ranked 1st in a single list (1/61 ≈ 0.016).

A slower re-ranking model can then reorder the top few hundred, as in two-stage ranking. Three things are harder with vectors than with an inverted index.

• Filters: "under ₹2,000, size 9" is a cheap postings intersection in a keyword index, but an ANN graph has no such shortcut. Filtering after the vector search can leave 2 results out of 10, so engines filter during the walk or fetch extra candidates, such as 200 to return 10.
• Freshness: a new listing must pass through the embedding model before it can be found, and inserting into an HNSW graph costs more than adding postings. Changing the model means re-embedding all 50 M listings and swapping the alias, as with an analyser change.
• Sharding: shard by document, as before. Each shard holds an ANN index over its slice and returns its nearest 10, and the coordinator merges by similarity. Each shard's answer is approximate, so the merged one is too.

Slow shards are the best-known failure, but not the only one. These are the others that appear as traffic and data grow.

Most fixes either keep the load even across shards or keep one heavy job from starving the queries. A machine that dies outright is the easy case, because replicas already cover it.

Search turns up inside many designs, and you rarely build the engine: you decide how data reaches it, how it is split and how queries use it.