Distributed Search Engine

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:

L5 · 100M docs · ES cluster + RF=1 · $10K/mo · 6 months:

L6 · 1B docs · 33 shards + hedged reads · $60K/mo · 18 months:

L7 · Google 130T pages · custom index + neural rerank · $10B+/yr:

Why every 10× breaks the architecture

1. 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.

1. 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.

1. 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

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.

1. 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.

1. 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?

The CTO's one sentence — 'we need search that scales' — is 90% ambiguity. Your job as engineer #4 is to translate it into 8 quantified requirements that force the architecture. Miss even one and you'll design the wrong system.

Functional requirements

FR-1: Query mix. Not all search is BM25 keyword lookup. Real search workloads are heterogeneous:
- Keyword search (BM25): "noise cancelling headphones" — 60% of queries typically. Latency budget: 50ms p99. Cheap per query (~5ms per shard).
- Filtered aggregation (facets): "headphones, price 100-200, brand Bose, rating>=4" — 25% of queries. Latency: 100-200ms p99. Moderate cost (aggregation over posting lists).
- Vector kNN (semantic): "songs that feel like a rainy day" — 10% and growing. Latency: 100-200ms p99. Expensive (HNSW graph traversal, cache-cold cost).
- Autocomplete / suggest: "no..." → "noise cancelling..." — 5% of queries. Latency: <20ms — user typing speed. Extremely latency-sensitive.

Each of these has different indexing structures, memory footprints, and scaling curves. A search platform that only handles keyword BM25 fails when the ML team wants semantic search.

FR-2: Freshness requirements. Different data has different freshness tolerance:
- E-commerce catalog: 1s visibility is fine. Standard refresh_interval=1s.
- Product inventory: near-realtime — 30s stale on inventory shows "in stock" for sold-out items. Use write-through cache in front of index.
- Logs: 30s+ is fine — throughput matters, refresh_interval=30s is 5-10x faster indexing.
- Social feed: 100-500ms — set refresh_interval=200ms with dedicated ingest nodes.

Refresh interval directly trades throughput for freshness. A 1s refresh is Elasticsearch's default. Lowering to 100ms doubles CPU; raising to 30s makes bulk indexing 5-10x faster.

FR-3: Autocomplete. If you promise autocomplete <20ms p99, you cannot use the same cluster as your main search — the cache-cold cost of autocomplete kills its own latency budget. Options: (a) dedicated autocomplete cluster with different sharding, (b) prefix trie in-memory (Redis + custom trie), (c) autocomplete-as-a-service (Algolia Places, AWS Kendra). Reference: Elastic completion suggester docs.

FR-4: Filtered + faceted queries. Facets are the #1 killer of coord memory. A filter like "brand=Bose" is cheap. An aggregation like "top 100 brands by count in current query results" allocates 100 counters × N shards × M concurrent queries. Set search.max_buckets=10000 cluster-wide or you WILL take down coord nodes during a traffic spike.

Non-functional requirements

NFR-1: p99 latency. Search is user-facing. p99 = 100ms is the "feels instant" threshold. p99 = 500ms is "annoyingly slow." p99 > 1s is "search is broken." Every architecture decision — shard sizing, hedged requests, coord role, GC tuning — is downstream of this budget.

NFR-2: Availability. 99.95% for user-facing (23 min/mo). Fully-degrade path: on cluster failure, serve featured items + cached top-100 queries from Redis instead of showing broken search. This is the pattern Netflix and Amazon use.

NFR-3: Cost bound at scale. At 2B documents + 20K QPS peak, the options are roughly:
- Algolia: $200K-$1M+/year depending on QPS. Non-linear cost trajectory.
- Managed Elastic Cloud: $180-260K/yr for a 20-node r6i.4xlarge cluster equivalent. Linear cost.
- Self-hosted OpenSearch on EC2: $45-90K/yr compute + 1-2 FTEs ($400K-$800K). Best cost at scale but requires ops.
- AWS OpenSearch Service: ~$120-200K/yr — middle ground, less flexibility than self-hosted but no JVM tuning.

The right answer depends on team maturity. 3-engineer team → managed. 6+ engineer team with prior ES experience → self-hosted.

NFR-4: Ops complexity budget. Elasticsearch requires 15-30% of one FTE at steady state (shard rebalancing, JVM GC tuning, upgrades, snapshot validation). Under incidents (hot shard, coord OOM, cluster-yellow), can consume 100% of the team for 2-3 days. Under-budgeting ops complexity is the #1 reason search projects fail.

Shard sizing math

This is the single most consequential architectural decision at L4-L7. The formula:

```
target_shards = corpus_size / 30 GB
= 2 TB / 30 GB
= 67 primary shards
```

Then decide replication factor. RF=2 (1 primary + 1 replica) = 134 total shards. Distribute across 12-20 data nodes. Each node holds ~7-11 shards @ 30 GB = 210-330 GB per node.

Why 30 GB? Below 30 GB: too many shards → coord overhead + heap pressure per shard. Above 30 GB: single-shard query CPU becomes bottleneck (Lucene uses one core per shard per query). 30 GB is empirically the sweet spot, documented in Elastic's official sizing guide.

Getting shard sizing wrong is the #1 reason clusters fail at scale — either too many small shards (heap pressure) or too few large shards (hot shards on write skew).