Design Distributed Search: Inverted Indexes and Shards

A LIKE search, SQL's text pattern match, has to read every row when the pattern starts with a wildcard (a symbol that matches any text). Double the rows, and every single search does about double the work. Distributed search is the usual fix, and this guide builds it up from one index.

Picture a common setup. A classifieds marketplace lets people post things to sell or rent, such as used cycles, sofas and flats. A buyer types used cycle jayanagar and expects nearby listings in a blink. The setup is made up, but the problem is real.

When the site had 2 lakh (200,000) listings, the site's small team searched them with one SQL query using LIKE. It answered in about 50 milliseconds (made up). Now the site has 20 crore (200 million) listings, a thousand times more rows for every search to read.

At the evening peak, 3,000 people search every second (made up). The same query cannot keep up. The team sets three targets: answer in under 200 milliseconds, keep working when a machine dies, and show a new listing within a few seconds.

The fix has two steps. First, store the text so that finding a word is a lookup, not a scan. Second, split that structure across machines, because it no longer fits on one, and one machine cannot answer every search.

An inverted index is a map from each word to the list of documents that contain it. Instead of reading every listing to find a word, you look the word up and get the matching listings directly.

It is called inverted because it flips the normal direction. A listing holds its words. The index holds, for each word, its listings. In search, a document is any one thing you search over. Here, one document is one listing.

Three terms come up again and again. A term is one searchable word after cleaning, such as cycle. The dictionary is the list of all terms. A postings list is the list of document ids for one term, kept in sorted order.

Think of a property broker with a thick notebook. Each page has one feature at the top, such as 2BHK (two bedrooms, a hall and a kitchen) or near metro. Under it, he writes the number of every flat with that feature.

Map it part by part. A page heading is a term. The flat numbers under it are a postings list. The whole notebook is the inverted index, and each flat is a document. To answer a 2BHK near metro request, he opens two pages and reads out the flats on both.

Where the picture breaks: the broker scribbles numbers in any order and judges the flats by eye. A real index keeps each list sorted and compressed, and a separate ranking step scores every match.

Here are three postings lists from the team's index, each sorted by listing id:

• used: 101, 380, 455
• cycle: 101, 245, 380, 912
• jayanagar: 101, 102, 380, 777

For the query used cycle jayanagar, the engine walks the three lists side by side and keeps only ids found in all of them: 101 and 380. Because the lists are sorted, one pass is enough, with no going back.

You build an inverted index by splitting each document into words, cleaning those words, and adding the document's id to the postings list of every word. A search engine repeats this for every new or changed document.

The diagram shows one listing on its way into the index. Read it left to right, and notice that the listing id is what gets stored under each term, not the listing text.

1. 1

   Split the text

   Break the listing into tokens, which are the raw words. Used cycle, Jayanagar becomes three tokens: Used, cycle and Jayanagar.

2. 2

   Clean each word

   Lowercase it and strip commas and full stops. Many engines can also be set up to map close word forms together, so cycles and cycle become one term.

3. 3

   Add the id to each list

   For every term, add the listing's id to that term's postings list. Keep the list sorted by id.

4. 4

   Save a few extra details

   Store how often each word appears in the listing, and where. Ranking uses the counts, and exact phrase matching uses the positions.

Here is the idea in Python, small enough to read in a minute. It keeps postings as Python sets for clarity. Real engines keep sorted, compressed lists on disk.

from collections import defaultdict

def words(text):
    return [w.strip(".,").lower() for w in text.split()]

def build_index(listings):
    index = defaultdict(set)            # term -> ids of listings
    for lid, text in listings.items():
        for w in words(text):
            index[w].add(lid)
    return index

def search(index, query):
    lists = [index.get(w, set()) for w in words(query)]
    return set.intersection(*lists) if lists else set()

idx = build_index({101: "Used cycle, Jayanagar, good brakes",
                   102: "New cycle helmet, Jayanagar",
                   103: "Used cycle, Hebbal"})
print(sorted(idx["cycle"]))                   # [101, 102, 103]
print(search(idx, "used cycle jayanagar"))    # {101}

What just happened? Running it on Python 3.13 prints [101, 102, 103] and then {101}. The line index[w].add(lid) is the inversion itself: the word points to the listing, not the other way round.

The call to set.intersection keeps only listings that contain every query word. Listing 102 has no used, and 103 has no jayanagar, so only 101 survives.

Matching is only half of search. Ranking decides the order. BM25 is the default scoring method in Lucene, a search library, and in Elasticsearch, a widely used search engine built on it. It gives more weight to rare words and to words repeated in a document, with diminishing returns for each repeat.

BM25 also scores a match in a short listing above the same match in a long one. The marketplace then mixes in its own signals, such as distance and how recently the listing was posted.

Distributed search splits one search index into pieces called shards and spreads them across many machines. Each machine searches its own shard at the same time, and a coordinator merges their answers into one result list.

The team needs it for two reasons: size and load. The index for 20 crore listings comes to about 400 GB (made up), more than one machine should hold in memory. One machine also cannot answer 3,000 searches a second in under 200 milliseconds.

Two ideas fix this. A shard is one slice of the index, held by one machine. A replica is a full copy of a shard on another machine. Shards share out the size, and replicas share out the query load.

Back to the broker. One broker cannot cover a whole city, so the agency gives each locality its own broker and notebook. A head office takes each request, phones every broker, and picks the best flats from their short lists.

The 2003 paper on Google's web search cluster describes the same plan. Its index was divided into shards, each holding a randomly chosen subset of documents. A pool of machines served each shard, so one failed machine did not make any part of the index disappear.

Elasticsearch follows the same pattern. Its documentation says each shard is a self-contained Lucene index. Each document belongs to one primary shard, and a replica shard is a copy of a primary.

Split by document in almost every case, because each query then sends only short result lists over the network. Splitting by term sounds clever, but multi-word queries force machines to ship long postings lists to each other.

In document partitioning, each shard gets a set of whole documents and builds a full inverted index for them. The Stanford textbook on information retrieval calls this a local index and the more common implementation. Every query goes to every shard, and the results are merged.

In term partitioning, each shard owns a set of words and the full postings lists for those words. The textbook calls this a global index. A query visits only the shards that own its words.

In broker terms, term partitioning gives one broker every 2BHK page in the city and another every near-metro page. To answer one client, the first broker must send his whole list to the second. For popular features, those lists are huge.

The Google paper adds one more reason. Its shards did not need to talk to each other, so spreading one search over more shards gave a nearly linear speedup.

The word rarity row needs care. BM25 gives rare words more weight, but each shard only knows how rare a word is among its own listings. The textbook notes that statistics like this must cover the whole collection.

BM25 turns rarity into a weight called idf (inverse document frequency): the rarer the word, the higher. We tested it on 2 lakh made-up listings, using Lucene's idf formula.

A word common in Bengaluru listings got an idf of 3.42 overall and 3.40 to 3.43 on three randomly filled shards. With one shard per city, the same word got 2.97 to 5.37, so city shards would rank it very differently.

So how does a listing pick its shard? The team hashes the listing id. A hash function turns the id into a random-looking number, and the remainder after dividing by 16 picks one of 16 shards. Random spread keeps the shards about the same size.

Elasticsearch does the same by default. For a regular index, its 9.5.5 source hashes the document id, or a custom routing value if you give one, with a hash function called Murmur3.

A query runs in two rounds. First the coordinator sends it to one copy of every shard, and each shard returns only its top ids and scores. Then the coordinator merges those lists and fetches full details for the winners.

This pattern is often called scatter-gather: scatter the query out, then gather the answers back. Elasticsearch's documentation says every node can receive requests and coordinate them. Its default search type, query then fetch, runs these two rounds.

Follow one search for used cycle jayanagar through the team's design. Notice that shards send back short lists of ids, and full listings are read only for the final ten.

Why does each shard send its top 10, and not fewer? The global top 10 could all sit on one shard. Asking every shard for 10 means no winner is missed, as long as every shard scores the same way.

With random spread, asking for fewer usually works. In 500 test runs with random made-up scores over 16 shards, top 5 per shard still found the exact top 10 every time. When all ten winners sat on one shard, it missed five.

Deep pages cost more. For page 50, with 10 results per page, each shard must return its top 500, and the coordinator sorts 8,000 entries to show ten. By default, Elasticsearch rejects a search where from plus size, the skip count plus the page size, goes over 10,000 (index.max_result_window).

Here is the coordinator's gather step in Python. Each fake shard sleeps for a while and returns (score, listing_id) pairs. One shard is slow, and the coordinator stops waiting at 200 milliseconds.

import heapq, time
from concurrent.futures import ThreadPoolExecutor, wait

class FakeShard:                     # stands in for one shard copy
    def __init__(self, hits, delay):
        self.hits, self.delay = hits, delay
    def top_k(self, query, k):
        time.sleep(self.delay)       # pretend search + network time
        return heapq.nlargest(k, self.hits)

def search_all_shards(pool, shards, query, k=10, timeout=0.2):
    calls = [pool.submit(s.top_k, query, k) for s in shards]
    done, _late = wait(calls, timeout=timeout)   # stop waiting at 200 ms
    ok = [c for c in done if c.exception() is None]
    hits = [hit for c in ok for hit in c.result()]   # (score, id) pairs
    best = heapq.nlargest(k, hits)               # global top k by score
    return best, len(ok) < len(shards)           # True means partial

shards = [FakeShard([(9.1, 101), (4.0, 102)], delay=0.01),
          FakeShard([(7.5, 205), (6.2, 207)], delay=0.02),
          FakeShard([(9.8, 330)], delay=0.50)]   # the slow shard
with ThreadPoolExecutor(max_workers=8) as pool:
    best, partial = search_all_shards(pool, shards, "used cycle", k=3)
print(best)       # [(9.1, 101), (7.5, 205), (6.2, 207)]
print(partial)    # True

What just happened? It printed [(9.1, 101), (7.5, 205), (6.2, 207)] and True in 20 of 20 runs. The call to wait with a timeout returned before the slow shard answered, and heapq.nlargest picked the top three from the rest.

Look at what partial means. The slow shard held the best listing, scored 9.8, and the page never saw it. The True flag lets the page say results may be incomplete. Python cannot stop the late call, so it runs on and is ignored. The with block still waits for it on exit, so the program takes about half a second.

Beginners can stop here: you now know the index, the split and the query path. The rest is for engineers who run search in production. Elasticsearch details below come from its 9.5.5 source and docs as of October 2026, so pin your version.

A scatter-gather query is only as fast as the slowest shard it waits for. Pause and predict: if each shard is slow one time in 100, how often is a 16-shard query slow?

The answer is about 15 percent, because 1 minus 0.99 to the power 16 is 0.149. The Tail at Scale, by Dean and Barroso in 2013, works the same maths for 100 servers: 63 percent of requests become slow.

The paper warns that brief slowdowns, harmless in a moderate-size system, can come to dominate overall response time at large scale. More shards means more chances to wait on a bad one.

• Set a deadline for every query, and return what arrived in time, marked as partial.
• Send a hedged request: the paper's name for asking a second replica after a short delay and using whichever answers first.
• Pick the replica that has been answering fastest lately, not just the next one in turn.

Elasticsearch does the third by default, calling it adaptive replica selection. Its documentation lists the inputs: past response times from the coordinator, how long the node took on earlier searches, and the node's search queue size.

What the Elasticsearch timeout really limits

By default, an Elasticsearch search does not time out. It waits for every shard. This differs from the Python sketch above, where the coordinator owns one deadline.

The timeout parameter applies per shard, starting when the query phase begins on that shard. A shard that runs out of time returns the hits it collected, and the response says timed_out true. The documentation says it is not an overall search deadline.

Partial results are allowed by default: search.default_allow_partial_results is true in the 9.5.5 source. So one failed shard does not fail the whole search. Check the _shards section of every response, where failed counts the shards that returned nothing.

If your page must be complete, set allow_partial_search_results to false on the request, or the cluster default to false. Then a shard failure fails the whole search instead.

New listings become searchable through a pipeline that indexes each change within seconds. Inside each shard, the engine adds small new index pieces instead of rebuilding the whole index. Elasticsearch calls this near real-time search, not instant search.

When a seller posts a cycle, the marketplace saves the listing in its main database first. That database stays the source of truth. The search index is a copy shaped for fast lookup, and it can always be rebuilt from the database.

Follow a new listing from posting to search results. Notice that the database and the index are updated by different steps, joined by a queue.

One weak spot is the gap between saving the listing and sending the change event. If the service crashes in between, search never hears about it. The transactional outbox pattern fixes this by writing the event into the database in the same transaction.

A separate relay then publishes it, for example by tailing the database's transaction log, its record of every committed change.

Segments, refresh and merges

Inside each shard, Lucene does not rewrite one giant index for every new listing. New documents collect in an in-memory buffer and are written out as a new segment, which works like a small inverted index of its own. A search reads all of a shard's segments.

Writing and opening a new segment for search is called a refresh. Elasticsearch's documentation says it refreshes every second by default, but only on indices that received a search in the last 30 seconds.

Small segments pile up, so the engine merges them into bigger ones. Segments are never edited, so a deleted listing is only marked as deleted. An edit is a delete plus a fresh add, per Lucene's IndexWriter, and merges finally drop the old copies.

The search-idle catch

The 30-second rule is per shard, and the 9.5.5 source checks several things before skipping a refresh. The two you control: no index.refresh_interval set by you, and no search or get request for index.search.idle.after, 30 seconds by default.

Such a shard skips its scheduled refreshes. The next search that hits it, with changes pending, triggers a refresh and waits for it. So the first search after a quiet spell can be slower than the rest.

This is great for bulk loading at night, and surprising for a low-traffic index with a strict latency target. Setting index.refresh_interval explicitly, even to 1s, turns the idle skip off.

You keep it fast and available with enough replicas for the query rate, a deadline on every shard call, and a shard count chosen with room to grow. Each answers a specific failure.

Replicas for load and for survival

Beginners often miss this. With document partitioning, every query visits every shard. So each shard must handle the full 3,000 searches a second, not a sixteenth of it.

Say one copy of a shard handles about 1,500 searches a second (made up). The team keeps three copies of each shard: one primary and two replicas. Together they handle 4,500 a second. With one copy down, the other two cover exactly 3,000, with nothing to spare. Busy copies queue requests and slow the tail, so the team adds a fourth for peak hours.

Elasticsearch creates one replica per primary by default, and you can change the replica count at any time. When a machine dies, searches go to the remaining copies.

Choosing the number of shards

Pick the shard count early. In Elasticsearch, the number of primary shards is fixed when you create the index. Changing it later means the split or shrink APIs, which build a new index, or a full reindex.

Elastic's sizing guide suggests shards between 10 GB and 50 GB, with under 200 million documents each. At 400 GB, 16 shards of about 25 GB sit in that range and leave room for the data to double.

The team avoids going much higher. Every shard costs memory and CPU, and every query visits every shard, so a thousand tiny shards are slower to search than a few big ones.

Hot spots and routing

Some searches only ever need one slice of the data. Elasticsearch lets you attach a routing value, such as a city, when you index a document. All documents with that value land on one shard, and a search can target only that shard.

The team tested routing by city and backed away. Bengaluru brings 60 percent of the searches (made up), so the Bengaluru shard would be overloaded while others sat idle. They kept random spread and added a city filter to every query.

If you do route, Elasticsearch's index.routing_partition_size setting spreads each routing value over a few shards instead of one. Its 9.5.5 source adds an offset from the document id's hash to pick among them.

Watch the right numbers

• The 99th percentile response time, meaning the time within which 99 of every 100 searches finish, for each shard and overall.
• The share of searches with timed_out true or a non-zero _shards.failed count.
• Index lag: the seconds between a listing change and that change showing in search.
• Shard sizes and disk use, so one shard does not quietly outgrow the rest.

Start from one inverted index on one machine, then show exactly why you split it and how a query flows. Interviewers listen for the trade-offs, not for product names.

1. 1

   Clarify scale and freshness

   Ask how many documents, how many searches a second, the latency target and how fast new data must appear. For the marketplace: 20 crore listings, 3,000 a second, 200 ms.

2. 2

   Explain the inverted index

   Each term points to a sorted list of document ids. Mention cleaning the words, and the separate ranking step.

3. 3

   Estimate the size

   Do rough maths on index size and query rate out loud. Use it to justify the number of shards and replicas.

4. 4

   Choose document partitioning

   Shard by hashed document id. Explain why term partitioning ships too much data, and pick a shard count with room to grow.

5. 5

   Walk through one query

   Coordinator, one copy per shard, top k from each, merge by score, then fetch only the winners.

6. 6

   Cover tail latency and freshness

   Show that 16 shards turn rare slow calls into common slow queries. Then cover deadlines, partial results, replica choice, the change queue and the refresh.