Distributed Indexing and Searching

In one line: this is the first complete distributed design, and it introduces the merger — the component that makes fan-out queries work. It also introduces colocation, which the next lesson tears out.

The architecture

Indexing

• The crawler collects the document set.
• The cluster manager splits documents into N partitions based on data size and node availability. It monitors node health via periodic heartbeats. A hashing function assigns documents to partitions.
• The cluster manager runs indexing algorithms across all N partitions simultaneously on the N nodes. It creates "tiny" inverted indexes, stored locally — N smaller indexes rather than a single large index.

A hashing function assigns documents — and here hashing is the right call

The Blob Store chapter deliberately avoided hashing, range-partitioning by path instead, because its dominant query was a prefix scan. Here hashing is correct, and the contrast is instructive.

Search has no locality requirement between documents. Nobody asks for "all documents from partition 3." Every query goes to every node anyway (Lesson 7), so co-locating related documents would buy nothing.

What you do want is even distribution — equal-sized partitions so no node becomes the straggler that sets query latency. Hashing gives exactly that, with no lookup table to maintain.

Hash when you need balance and have no locality requirement; range-partition when you scan ranges. Two chapters, opposite answers, same rule applied to different access patterns.

Heartbeats are how the cluster manager knows anything

"Monitors node health via periodic heartbeats" is one clause carrying a lot of weight — it is the only mechanism by which the manager learns that a commodity node has died.

That matters because Lesson 5 chose commodity hardware, which fails routinely. Without failure detection, the manager would assign partitions to dead nodes and queries would hang rather than fail.

Heartbeats also drive the fault tolerance in Lesson 11's MapReduce: the cluster manager reschedules tasks when a node fails, which it can only do if it notices. Same mechanism as message-queue design's internal cluster manager listening for node heartbeats.

Searching

• The system executes parallel searches on all N nodes.
• Each node returns a list of mappings for the query terms. The merger aggregates these lists.
• The merger sorts the documents in the aggregated mapping list based on term frequency.
• The sorted results are returned to the user.

The merger is the component that makes fan-out usable

Without a merger, a fan-out query returns N separate result lists and the client has to reconcile them. The merger turns N partial answers into one ranked answer, and it is where several things happen at once:

Aggregation — collect what each node returned. Global ranking — each node ranked within its own slice; the merger produces the ordering across slices. Truncation — Lesson 2's response is 80 suggestions, so the merger discards almost everything it receives.

Two properties worth naming. The merger is a fan-in point, so it sees N responses per query and can become a bottleneck at high N — real systems use a tree of mergers rather than one. And it is where the tail latency from Lesson 7 bites: the merger cannot finish until the slowest node replies, which is why production systems often return after 99 of 100 nodes have answered rather than waiting.

Note also that sorting on term frequency alone is a simplification. Lesson 4 argued relevance needs more signals; frequency is the one the index actually stores.

Page 1,000 is far more expensive than page 1

Scatter-gather has a cost that grows with page depth, and it surprises people because paging feels like it should be cheap.

The reason is that no shard knows its own results' global rank. To guarantee the true 10,000th document is found, every shard must offer its own top 10,010 — because in the worst case they all came from one shard. So the coordinator handles shards × offset candidates to return ten.

The fix is a cursor rather than an offset: remember the sort key of the last result seen and ask each shard for the next ten after this value. Each shard then returns ten, the coordinator merges N×10, and cost is flat regardless of depth.

The trade is that a cursor only walks forward — you cannot jump to page 1,000 — which is why search products offer "next" and not a page-number strip once results run deep. That is a product constraint falling directly out of the architecture, and worth naming as such.

Making a query cheaper the second time

Separating filtering from scoring is what makes caching possible. A filter has a binary answer and no relevance component, so the set of documents matching published = true, language = en is identical across users and can be cached as a bitmap and reused. Scores cannot be reused that way, because they depend on the query.

That is why search APIs distinguish the two clause types rather than treating everything as a query: the distinction is a caching boundary, not a syntax preference.

Colocation

Note: we use colocation, meaning both searching and indexing are performed on the same nodes. This handles large datasets by working on smaller partitions.

Colocation is introduced here and removed two lessons later — track why

Running indexing and search on the same nodes has an obvious appeal: the index is already there, so a search needs no network transfer to reach it. Locality for free.

Lesson 10 explains why it fails anyway — resource contention between two heavy workloads, and no independent scaling. Worth flagging now so the reversal reads as a deliberate refinement rather than a contradiction.

It is also a useful pattern to recognize in interviews: the first design takes the locally optimal choice, and the second gives it up once you look at how the two workloads actually behave. Colocation optimizes data movement; separation optimizes resource isolation. At this scale, isolation wins.

Multi-data-center deployment

The proposed design can be replicated across multiple data centers worldwide to serve users globally, which:

Geographic replication is unusually cheap for search — because the index is read-only

Replicating a database across data centers is hard: writes must be ordered, conflicts resolved, consistency chosen. Replicating a search index is almost trivial by comparison.

The index is immutable and read-only to the search path. Nothing a searcher does modifies it. So a remote data center just needs a copy of a file, and there are no writes to reconcile and no conflicts possible.

Two things make it work. Lesson 6's offline indexing, so replication happens off the critical path. And Lesson 12's observation that search results are allowed to be slightly stale, so a data center running yesterday's index is degraded rather than broken.

Read-only data is the easiest thing in distributed systems to replicate — which is why global search deployment is a solved problem while global databases are not.

Interview signal by level

Next: making it survive failures.