Distributed Search Engine
Sharded inverted index + query fan-out.
The scenario
Sharded inverted index at Google/Elasticsearch scale — where 30 GB shards, JVM heap, and Tail-at-Scale hedging converge
Same startup, same engineer #4. Eighteenth Monday.
Your CTO drops by. "Every product we've built now needs search. E-commerce catalog. Slack messages. Instagram hashtags. Ticket events. Time to understand how distributed search ACTUALLY works — inverted indexes, scatter-gather queries, shard sizing, JVM heap tuning. Ship a design in 12 weeks."
She pauses. "For context — Google indexes 130+ TRILLION web pages. Elasticsearch runs at Uber for 8 PB of logs. Netflix runs Elasticsearch for 40 PB of operational metrics. Meta runs Unicorn (their in-house search) for the social graph at Facebook. If you don't understand shard sizing + heap tuning + hedged requests + inverted index mechanics, you'll design search that dies under real workloads."
Here's the paradigm shift:
Distributed Cache was in-memory, ephemeral. Distributed Database was structured records + tunable consistency. Kafka was ordered logs.
Distributed Search is fundamentally different:
- The primary data structure is the inverted index — not a B-tree, not a hash, not a document. Term → posting list of doc_ids. Reference: Lucene inverted index deep-dive.
- Every query is scatter-gather — coordinator fans out to N shards, each shard executes locally, coordinator merges. P99 latency = tail of N parallel shard queries.
- The bottleneck is single-shard CPU — Lucene uses ONE CORE per shard per query. Oversized shards create hot spots. Undersized shards create JVM heap pressure. The sweet spot is 30 GB per shard.
- BM25 relevance scoring is math-intensive — every document scored per query. This is why search is expensive per-QPS compared to KV lookups.
Doug Cutting's answer (Lucene creator) was to build the inverted-index data structure with segment-based storage + FST + compression. Shay Banon's answer (Elasticsearch creator) was to wrap Lucene with distributed sharding + REST + monitoring. Yonik Seeley's answer was Solr. All three are variants of the same core idea. Reference: Doug Cutting on Lucene history.
Google's answer is proprietary but leaks about it via research papers — Anatomy of a Large-Scale Hypertextual Web Search Engine (Brin & Page 1998) is the foundational paper. Reference: Google's PageRank patent.
The real 2024 numbers
- Google Search: 130+ trillion pages indexed (Google's How Search Works), 8.5B searches/day (~100K QPS avg)
- Elasticsearch shard size sweet spot: 10-50 GB, target ~30 GB (Elastic official docs)
- JVM heap max: 30 GB per node — compressed oops boundary; exceeding this doubles memory for the same data
- Shard overhead: 50-100 MB heap per shard — practical max ~200 primary shards/node for search workloads
- BM25 latency: ~1-10ms per shard for typical query at 200M docs/shard
- Uber Elasticsearch: 8 PB of logs (Uber Engineering blog)
- Netflix Atlas: 40 PB operational metrics via Cassandra-backed time-series, but Elasticsearch for log search
- Elasticsearch default replication: RF=1 (1 primary + 1 replica per shard)
Interview soundbite: "Distributed search is 4 primitives: (1) inverted index in Lucene with FST for terms + posting lists for doc_ids, (2) sharded hash-by-doc_id with target 30 GB per shard, (3) scatter-gather query across N shards with tail-at-scale hedging, (4) JVM heap capped at 30 GB per node (compressed oops). Elasticsearch is Lucene + REST + distribution. Uber runs 8 PB of logs on it. Naming these 4 primitives + shard sizing + Dean & Barroso Tail-at-Scale hedging signals L6+ preparation."
The whole journey at a glance
Every 10× in dataset size surfaces different bottlenecks:
text═══════════ DISTRIBUTED SEARCH ACROSS 4 SCALES ═══════════ L4 (10M docs) L5 (100M docs) L6 (1B docs) L7 (Google 130T pages) Single Elasticsrch ES cluster + RF=1 Hedged reads + tiering Custom index + BERT 12 weeks · $500/mo 6 months · $10K/mo 18 months · $60K/mo ongoing · $10B+/yr ┌────────┐ ┌────────┐ ┌── Client apps ─────┐ ┌── Client apps ──────┐ │ App │ │ App │ │ 100+ services │ │ Search UI + APIs │ │ pods │ │ pods │ └─┬──┬──┬──┬─────────┘ └──┬──┬──┬──┬──────────┘ └───┬────┘ └───┬────┘ │ │ │ │ │ │ │ │ │ │ ┌─▼──▼──▼──▼─────┐ ┌───▼──▼──▼──▼──────────┐ ┌──▼───┐ ┌──▼──┐ │ Coordinator │ │Google's serving stack │ │ ES │ │ ES │ │ + hedged reqs │ │+ neural re-ranker │ │ 1 │ │clust│ │ + caching │ │+ BERT / MUM models │ │ node │ │ 3 │ └───┬──────┬─────┘ └──┬──┬──┬──┬───────────┘ │ 5 GB │ │nodes│ │ │ │ │ │ │ └──┬───┘ │RF=1 │ ┌───▼──────▼───┐ ┌──▼──▼──▼──▼──────────┐ │ └──┬──┘ │ ES cluster │ │Google/Bing scale: │ │ │ │ 20 nodes │ │10B+ docs indexed │ │ │ │ ~33 shards │ │Custom index format │ │ │ │ RF=2 = 66 │ │Multi-tier retrieval │ │ │ │ total │ │+ BM25 first stage │ │ │ │ m5.2xlarge │ │+ BERT/MUM re-rank │ │ │ │ 30 GB heap │ │+ PageRank │ │ ┌───▼──┐ │ ~30 GB/shard│ └──┬───────────────────┘ │ │Kafka │ └───┬──────────┘ │ │ │ pipe │ ┌───▼──────────┐ ┌──▼─────────────────────┐ │ │line │ │Multi-region │ │Multi-region custom │ │ │for │ │ILM tiering: │ │index + neural retrieval│ ┌──▼───┐ │index │ │hot/warm/cold │ │+ Bigtable metadata │ │Postgr│ │refresh │+ time-based │ │+ Kafka for realtime │ │(sourc│ └──────┘ │partitioning │ │ index updates │ │e of │ │+ Kafka for │ │+ SLO burn-rate alerts │ │truth)│ │ realtime │ └──────────────────────┘ └──────┘ │ indexing │ └──────────────┘ Bottleneck Bottleneck Bottleneck Bottleneck Single node Cross-shard Hot shard from viral Global multi-region. memory limit. aggregation + query pattern. Hedged Neural re-ranking = 5-10 GB index hot shard risk. requests + shard-level compute-bound. Custom fits on 1 box. result caching. silicon for inference. Chapter 5 Chapters 6+6.5 Chapter 7+7.5 Chapter 8 walks walks through walks through hedged walks through Google/ through Elasticsearch requests + Tail-at-Scale Bing scale, BM25 vs L4 MVP cluster + RF=2 + shard sizing math BERT, and custom silicon + 30 GB target for neural retrieval Key insight: Distributed search is INVERTED INDEX + SCATTER-GATHER. Sharding on doc_id gives even distribution but fan-out cost. Tail-at-Scale hedging (Dean & Barroso 2013) is the L6 answer to slow-shard problem. JVM heap 30 GB max is a hard boundary (compressed oops). If you name 4 primitives (inverted index + 30 GB shard + hedged requests + JVM heap 30 GB) you're L6+.
The same 4 tiers as clean architecture diagrams
L4 · 10M docs · Single Elasticsearch node · $500/mo · 12 weeks:
flowchart TD
W([App pods]) -->|REST query| ES[Elasticsearch 1 node<br/>m5.xlarge<br/>~5 GB index]
W -.->|cache miss| DB[(Postgres source-of-truth)]
classDef n fill:#dbeafe,stroke:#2563eb,color:#1e3a8a
class ES,DB nL5 · 100M docs · ES cluster + RF=1 · $10K/mo · 6 months:
flowchart TD
W([App pods]) -->|REST query| CO[Coordinator node<br/>parses query + routes]
CO --> S1[Shard 1 primary]
CO --> S2[Shard 2 primary]
CO --> S3[Shard 3 primary]
S1 --> R1[Replica 1]
S2 --> R2[Replica 2]
S3 --> R3[Replica 3]
KAFKA[Kafka index refresh pipeline] --> CO
classDef n fill:#dbeafe,stroke:#2563eb,color:#1e3a8a
classDef m fill:#fef3c7,stroke:#d97706,color:#78350f
class CO n
class S1,S2,S3,R1,R2,R3,KAFKA mL6 · 1B docs · 33 shards + hedged reads · $60K/mo · 18 months:
flowchart TD
W([100+ services]) -->|REST query| L1[Client-side query cache<br/>+ hedged requests]
L1 --> CO[Coordinator<br/>+ scatter-gather<br/>+ shard-level cache]
CO --> SHARDS[33 primary shards<br/>+ 33 replicas<br/>30 GB each<br/>m5.2xlarge · 30 GB heap]
SHARDS --> ILM[ILM tiering<br/>hot/warm/cold<br/>time-based partitioning]
KAFKA[Kafka realtime index] --> CO
classDef n fill:#dbeafe,stroke:#2563eb,color:#1e3a8a
classDef m fill:#fef3c7,stroke:#d97706,color:#78350f
class L1,CO n
class SHARDS,ILM,KAFKA mL7 · Google 130T pages · custom index + neural rerank · $10B+/yr:
flowchart TD
W([Google Search UI]) -->|HTTPS| CDN[Google edge PoPs<br/>1000+ globally]
CDN --> API[Serving stack]
API --> BM25[BM25 first-stage retrieval<br/>~top-1000 candidates from 10B+ docs]
BM25 --> NEURAL[Neural re-ranker<br/>BERT / MUM models<br/>on TPU + custom silicon]
NEURAL --> PR[PageRank signals<br/>+ knowledge graph]
API --> BT[(Bigtable<br/>index metadata<br/>Kafka for realtime)]
classDef n fill:#dbeafe,stroke:#2563eb,color:#1e3a8a
classDef m fill:#fef3c7,stroke:#d97706,color:#78350f
classDef ml fill:#dcfce7,stroke:#16a34a,color:#14532d
class CDN,API n
class BT m
class BM25,NEURAL,PR mlWhy every 10× breaks the architecture
- Shard sizing is the L5 signal. 10-50 GB per shard, target 30 GB. Too small = JVM heap pressure from overhead (50-100 MB heap per shard). Too large = Lucene single-core-per-shard hot spot. Reference: Elastic official sizing guide.
- JVM heap max 30 GB per node is a hard boundary. Compressed oops stop working above 30 GB, doubling memory usage for the same data. This is why Elasticsearch nodes are typically 64 GB RAM (30 GB heap + 34 GB OS page cache for Lucene mmap). Reference: Elastic JVM heap docs.
- Hedged requests are the L6 tail-at-scale answer. Scatter-gather to N shards means P99 latency = tail of N parallel queries. Send a duplicate request to a second replica after 50ms if the first hasn't responded. Reference: Dean & Barroso "The Tail at Scale" 2013 CACM paper — invented at Google.
The 3 senior insights before we start Chapter 1
- Elasticsearch is Lucene + REST + distribution. Lucene is the actual inverted-index library, 20+ years old. Elasticsearch (2010) and Solr (2004) both wrap Lucene. Knowing this distinguishes L5 from L4 candidates. Reference: Lucene project history.
- BM25 + neural re-ranking is the industry pattern. BM25 for cheap first-stage retrieval (top 1000 candidates), then neural model (BERT, ColBERT, cross-encoder) to re-rank top 20. Google, Bing, Meta, Netflix all use this pattern. Naming BM25 + cross-encoder re-ranking signals L6+ awareness. Reference: Nogueira & Cho "Passage Re-ranking with BERT" 2019.
- Tail-at-Scale hedging is the L6+ scatter-gather primitive. The Dean & Barroso 2013 CACM paper is required reading. If you can't cite "Tail at Scale" you'll fail the L6 signal for search. Reference: Dean & Barroso 2013.
Chapter map for the journey ahead
- Chapter 1 — Requirements (full-text, faceted, autocomplete, real-time index)
- Chapter 2 — Capacity estimation (Google 130T pages, Uber 8 PB logs)
- Chapter 3 — API design (query, filter, aggregate, suggest)
- Chapter 4 — Data model (inverted index, doc store, facet index)
- Chapter 4.5 — Lucene deep-dive: FST for terms, posting lists for doc_ids
- Chapter 5 — L4 MVP: Single Elasticsearch node. Works to 10M docs
- Chapter 6 — L5: Elasticsearch cluster + RF=1 + 30 GB shard sizing
- Chapter 6.5 — BM25 relevance scoring + facets + aggregations
- Chapter 7 — L6: Hedged requests + shard-level caching + ILM tiering
- Chapter 7.5 — Tail at Scale deep-dive: hedged reads + speculative retries
- Chapter 8 — L7: BM25 + neural re-ranking + custom silicon
- Chapter 9 — Failure modes: hot shard, JVM heap pressure, index bloat, slow rebalance
- Chapter 10 — Trade-off matrix (Elasticsearch vs Solr vs Vespa vs Meilisearch vs Google Vertex AI Search)
- Chapter 11 — Interview masterclass: 45-min mock, questions to ask
- Chapter 12 — Defense: the 20 hardest interview questions on search
Ready? Chapter 1 next: what did the CTO actually ask for?
Distributed search is 4 primitives: (1) inverted index in Lucene with FST for terms + posting lists for doc_ids, (2) sharded hash-by-doc_id with target 30 GB per shard (10-50 GB range), (3) scatter-gather query across N shards with tail-at-scale hedging (Dean & Barroso 2013 CACM), (4) JVM heap capped at 30 GB per node (compressed oops boundary). Elasticsearch is Lucene + REST + distribution. Uber runs 8 PB of logs on it. Google indexes 130+ trillion pages. BM25 first-stage + neural re-ranking (BERT/cross-encoder) is the industry pattern. Naming 4 primitives + Tail at Scale + BM25+re-ranking signals L6+/L7 preparation.
- Why is Elasticsearch shard sweet spot 30 GB?
- Why is JVM heap capped at 30 GB (compressed oops)?
- What is Tail-at-Scale and how do hedged requests solve it?
- What's the difference between Lucene, Elasticsearch, and Solr?
- Why is BM25 + neural re-ranking the industry pattern for L6+ search?
Every concept below has its own interactive, animated page in the Learning Tracks section. Read them any time you want to go deeper than the mentor prose above — they're the reusable foundation this chapter is built on.
The algorithm behind Elasticsearch shard placement — hash(doc_id) mod primary_shard_count determines destination.
Kafka is the canonical index-refresh pipeline — every doc change publishes to Kafka, Elasticsearch consumers keep indices in sync.
Chapter 1 next: what did the CTO actually ask for? Full-text, faceted, autocomplete, real-time index — each has functional and non-functional requirements. Get these wrong and you'll design the wrong system.