Design Search System

What are we designing? A full-text search system over 10B documents (web pages, product listings, code, logs). A user types a query and gets a ranked list of matching documents in under 200 ms, even when the corpus is petabytes. The system must index new documents within seconds (near real-time) and rank by relevance.

The defining challenges are (1) the inverted index — the core data structure that maps terms to postings lists (which documents contain them, in which positions), (2) sharding — how to split 10B documents across 100 nodes such that a query can be answered without scanning the whole corpus, and (3) ranking — given a set of matching documents, how to put the most relevant ones first using TF-IDF, BM25, and page-rank-style signals.

Functional requirements.

• A document is indexed (title, body, metadata) within seconds of being published.
• A user can search by arbitrary text query and get ranked results.
• Search supports phrase queries ("distributed systems"), boolean operators, filters (date, language), and facets (counts by category).
• Optional: typo tolerance (fuzzy search), autocomplete, personalization.

Non-functional requirements.

• Query latency: p99 < 200 ms for 10B docs across 100 shards.
• Indexing latency: < 5 s from publish to searchable (NRT — near real time).
• Freshness: indexed documents visible within seconds.
• Throughput: 10K queries/sec avg, 100K/sec peak.
• Availability: 99.9% (degraded results acceptable during partial outages).

Non-goals. No ML-based semantic search (separate vector-DB layer — see below), no real-time personalization per query (cached signals only).

Capacity estimation.

Corpus. 10B documents, avg 10 KB each = 100 TB of source text. Postings list (term -> [doc_id, position]) is typically 50-100% of source size with positions; without positions, much smaller. Estimate 50 TB of index data.

Shards. A single shard can hold ~50 GB of index comfortably (Lucene segment merge cost, JVM heap for caches). 50 TB / 50 GB = 1000 shards. Replicate 2x = 2000 shard instances across ~100 nodes (each node holds ~20 shards).

Query throughput. 10K QPS. Each query touches ~20 shards (one shard per primary, plus 1 replica for HA) = 200K shard queries/sec. Each shard query scans at most ~10K postings and ranks via BM25 — ~5 ms per shard, parallelized. With fan-out, total query latency = max shard latency + merge ~ 50-100 ms. Good.

Indexing throughput. 10K docs/sec avg. Each doc requires: analyze tokens, update postings lists, write to segment. Single shard ~1K docs/sec. We have 1000 shards, so total ~1M docs/sec capacity — plenty.

Memory. Hot postings lists (top-K most frequent terms) cached in JVM heap. Caches ~30% hit rate on long-tail queries; ~80% hit rate on top 1000 terms.

APIs.

POST /v1/index/:index_name/_doc
  { id, title, body, metadata: {lang, date, ...} }
  -> { _id, _version, result: "created" }

POST /v1/index/:index_name/_search
  { query: { match: { body: "distributed systems design" } },
    filter: { range: { date: { gte: "2024-01-01" } } },
    facets: ["category"],
    size: 20 }
  -> { hits: { total, hits: [{ _id, _score, _source }] } }

POST /v1/index/:index_name/_refresh    -- make recent docs visible

Elasticsearch's REST API is the de-facto standard. The _refresh endpoint forces Lucene to reopen its searcher, making newly indexed docs visible (NRT cycle is ~1s by default).

Data model: the inverted index.

The inverted index is a map: term -> postings list. A posting is (doc_id, term_frequency, positions[]).

Inverted index for corpus of 3 docs:
  Doc1: "the cat sat on the mat"
  Doc2: "the dog sat on the log"
  Doc3: "the cat chased the dog"

Term   -> Postings
-----
the    -> [(1,2,[0,5]), (2,2,[0,5]), (3,2,[0,4])]
cat    -> [(1,1,[1]), (3,1,[1])]
sat    -> [(1,1,[2]), (2,1,[2])]
on     -> [(1,1,[3]), (2,1,[3])]
mat    -> [(1,1,[6])]
dog    -> [(2,1,[5]), (3,1,[4])]
log    -> [(2,1,[6])]
chased -> [(3,1,[2])]

Query "cat sat" finds postings for both cat and sat, intersects doc_ids ({1} ∩ {1,2} = {1}), and ranks by BM25 score on combined term frequencies.

Per-shard storage (Lucene segment files):

• .tii / .tip — term dictionary (FST, prefix-compressed)
• .doc / .pos — postings lists (doc IDs and positions, delta-encoded + bit-packed)
• .fdt / .fdx — stored fields (the original document content for retrieval)
• .cfs — compound segment file (single file for many sub-files, fewer file handles)

Sharding strategy. Elasticsearch defaults to shard-by-document: each primary shard holds a random ~1/N subset of documents. A query is broadcast to all primaries (scatter-gather), each returns top-K, the coordinator merges. This is simple and balanced — but every query touches every shard.

Deep dive: the inverted index and analyzer.

The analyzer pipeline. Before a document is indexed, its text fields pass through an analyzer:

1. Tokenizer — split text into tokens (e.g. StandardTokenizer on Unicode whitespace + punctuation).
2. Lowercase filter — case folding ("Cat" -> "cat").
3. Stop word filter — drop "the", "a", "of" for English.
4. Stemmer filter — Porter stemmer reduces "running" -> "run", "mice" -> "mouse".
5. Synonym filter (optional) — "car" -> "automobile".

The same analyzer runs at query time, so query and document tokens live in the same space. Mismatched analyzers are the #1 bug in search systems ("I searched for 'running' but got nothing because docs were stemmed to 'run' and my query wasn't").

The postings list encoding. A naive postings list of doc IDs would be huge. Lucene encodes postings as delta-encoded varints: [5, 8, 12] becomes [5, +3, +4] and each delta is varint-encoded (smaller deltas = 1 byte). Block-level encodings (PFor, PFOR-DELTA) compress 128-posting blocks with bit-packing. This is why Lucene indexes are smaller than the source text despite containing positional info.

The term dictionary (.tii). The term dictionary maps term -> offset in postings file. Stored as a finite-state transducer (FST) — a prefix-compressed DAG. Memory-resident (a few MB per shard). Lookups are O(length of term). This is the magic that lets Lucene find any term in microseconds.

Segment immutability. Lucene writes segments — each segment is a write-once inverted index. New documents create a new segment. Deleted documents are marked in a bitset (liveDocs). Periodically segments are merged (smaller -> larger) to reclaim space from deletes and improve query efficiency. This is why deletes are 'soft' until the next merge.

Deep dive: sharding, ranking, and NRT.

Shard by document (Elasticsearch default). Random hash of doc_id to shard. Pros: simple, balanced, every shard is independent. Cons: every query fans out to all N shards. With N=1000 shards and 10K QPS, that's 10M shard queries/sec. Fine — but query latency = slowest shard's response time, so the tail is bounded by the slowest shard's GC pause.

Shard by term (alternative). Partition the term space (e.g. hash(term) mod N). A query for "distributed systems" only hits the shards that own those terms — usually 1-2 shards instead of all 1000. Pros: dramatic query fan-out reduction (only relevant shards). Cons: term frequency is wildly skewed — "the" appears in 99% of docs, "supercalifragilistic" in 0.001%. The shard owning common terms becomes a hotspot. Used by some specialized systems (Google's old index used term-sharding); rare in practice for general-purpose search because of the skew.

Ranking: TF-IDF vs BM25. Classical TF-IDF: score = tf * idf, where idf = log(N/df) (df = number of docs containing the term). Penalizes common terms, rewards rare-term matches. BM25 is the modern default: similar but with saturation (tf is capped by a k1 parameter so a term appearing 100 times in one doc doesn't dominate). BM25 also rewards shorter docs (b parameter) — all else equal, a 100-word doc matching is more relevant than a 10000-word doc matching.

PageRank and beyond. Pure text relevance isn't enough; a spam page repeating keywords would score high. Production systems layer signals: PageRank-style link authority, freshness (recency boost for news), personalization (user's history), and ML rerankers (LambdaMART, BERT-based). All these are computed offline / asynchronously and merged at query time.

Near Real Time (NRT). Lucene writes new docs to an in-memory buffer; every refresh_interval (default 1s) it flushes to a new segment and reopens the searcher. The new docs become visible. Every translog_flush (default 30 min or 5K docs) it fsyncs the segment to disk (durability). The 1s refresh is what makes ES 'near real time' rather than real time — and the source of the classic 'why isn't my doc searchable?' interview question.

Scatter-gather query. Coordinator sends query to one replica of each shard (load-balanced). Each shard runs the query locally, returns top-K (just doc_id + score; source fields fetched lazily from coordinator if top-K). Coordinator merges with a heap of size K. With K=20 and 1000 shards, the merge is 20K entries — trivial.

Bottlenecks and failure modes.

• Hot term. A query for a popular term ("python") fans out to all shards, each loads a huge postings list. Mitigation: cache top-K postings for hot terms in JVM heap (LRU); use filter cache for repeated boolean filters.

• JVM GC pauses. Lucene relies on JVM; large heap = long stop-the-world pauses = query tail latency spikes. Mitigation: keep heap < 32 GB (avoids compressed-oops break); use off-heap for postings (Lucene is mostly off-heap); use ZGC or Shenandoah for sub-ms pauses.

• Slow shard (tail latency). Scatter-gather latency = slowest shard. One shard with a 500ms GC pause kills p99. Mitigation: query replicas in parallel and take first response (hedged requests); per-shard query timeout with partial-result return.

• Reindex on mapping change. Changing a field's analyzer or type requires reindexing all docs (expensive). Mitigation: use aliases; reindex to a new index in the background; swap alias when done.

• Segment merge storm. Heavy indexing creates many small segments; merges consume CPU and IO. Mitigation: tune index.merge.policy (tiered merge, max_merged_segment); backpressure on indexing throughput.

• Cluster split-brain. Network partition between master and data nodes. Mitigation: require quorum of master-eligible nodes; use Elasticsearch's 7.x coordination (no more split-brain since they removed Zen1).

• Field data out-of-memory. Sorting on a text field loads all values into heap (fielddata). Mitigation: use doc_values (on-disk column store) by default since Lucene 4 — no heap pressure.

Scaling strategy and trade-offs.

Shard count. Pick shard count up front — too few = unbalanced, too many = overhead. Rule of thumb: shard size 30-50 GB. Re-sharding requires reindexing (expensive). Plan for 2x growth.

Replicas. Each primary shard has 1-2 replicas. Reads go to any replica (load balancing). Writes go to primary then replicated. Replica also provides HA — lose a node, no data loss.

Hot-warm-cold architecture. Time-series indices (logs): recent = hot (SSD, high CPU), older = warm (HDD, less CPU), ancient = cold (object storage, searchable snapshots). Saves cost massively.

Index per time period. For logs: one index per day. Old indices can be deleted (drop the whole index, no GC). For product search: one index, sharded.

Trade-offs made explicit.

• We chose shard-by-document — gained simplicity and balanced load, lost query fan-out efficiency (every query hits every shard).
• We chose 1s NRT refresh — gained near-real-time freshness, lost indexing throughput (segment creation has overhead) and strict consistency.
• We chose BM25 over learned rankers — gained interpretability and zero training data, lost personalization (would require ML reranker).
• We chose Lucene segments (immutable) — gained lock-free concurrent reads, lost cheap deletes (soft until merge).
• We chose Elasticsearch-managed replicas — gained simplicity, lost the ability to do cross-datacenter synchronous writes (ES replication is async; cross-DC sync writes would murder latency).

Mental model. Think of Design Search System as a deliberate boundary in a system. The boundary exists because something becomes harder to manage when everything is done in one place: latency, scale, failure isolation, consistency, cost, or team ownership. The useful question is not “what does Design Search System mean?” but “what pressure makes this boundary worth introducing, and what new failure mode does it create?”

Numerical lens: write down traffic, payload size, read/write ratio, peak multiplier and durability target before choosing a component. The numbers should justify the architecture.

Answer this without notes: When would you choose Design Search System, and when would you intentionally avoid it? Mention at least one bottleneck it addresses, one failure mode it introduces, and one alternative. Then quantify the workload you are designing for.

Numerical sanity check

Back-of-the-envelope reasoning beats fake precision. State your traffic, payload, concurrency and growth assumptions explicitly, then calculate enough to know whether the current architecture is orders of magnitude away from the target.