Under the Hood: Distributing Inverted Indexes Across Clusters
When an index grows to hundreds of gigabytes or billions of documents, it must be partitioned across multiple machines. In search systems, partitioning is called Sharding. Each Elasticsearch shard is a fully functional, self-contained Apache Lucene instance.
Document-Based Sharding: Scatter-Gather Flow
Document Partitioning vs Term Partitioning
| Strategy | Document Partitioning (Scatter-Gather) | Term Partitioning |
|---|---|---|
| Partition Logic | Documents are hashed to shards by DocID: shard = Murmur3(id) % N. Each shard contains all terms for its subset of documents. | Vocabulary is split across shards: Shard 1 holds "a"-"g", Shard 2 holds "h"-"p". |
| Ingest Speed | Fast & Local: Document is indexed entirely on a single shard with zero network coordination. | Slow: Indexing a single document requires network RPCs to scatter words across all shards. |
| Query Execution | Requires Scatter-Gather across all N shards. | Multi-term queries require network joins across term shards. |
Custom Routing for Multi-Tenancy: In SaaS platforms, querying every shard for a single customer is wasteful. Setting routing=tenant_id forces all documents belonging to tenant 42 into a single shard, converting a global scatter-gather into a single-shard targeted seek.
Worked example: sizing a cluster
You must index 2 billion documents averaging 1.5 KB of indexed data each: about 3 TB of primary data.
- Elastic's guidance is shards of roughly 10 to 50 GB. At about 40 GB per shard, 3 TB needs about 75 shards; choose 80 primaries.
- One replica each gives 160 shards, which spreads well over, say, 20 data nodes at 8 shards per node.
- Primary shard count is fixed at index creation, so plan for growth, or use time-based indices, rollover, or the split API.
Worked example: why fan-out hurts the tail
If each shard independently exceeds its own p99 latency 1 percent of the time, a query touching 80 shards waits on at least one slow shard with probability 1 minus 0.99 to the power 80, about 55 percent. So the per-shard p99 becomes roughly your median. Mitigations: fewer, larger shards; adaptive replica selection, which Elasticsearch enables by default; hedged requests; and routing to reduce fan-out.
Worked example: deep pagination
Request from=10000, size=10 over 80 shards. Each shard must return its top 10,010 hits, and the coordinator merges 800,800 entries to keep 10. Elasticsearch caps this with index.max_result_window, 10,000 by default. search_after avoids it by passing the last sort key, so each shard returns only 10.
Routing trade-offs
| Routing | Query fan-out | Risk |
|---|---|---|
| Hash of document ID, the default | All shards | Even load, but every query scatters |
| Custom routing by tenant ID | One shard per tenant | A huge tenant creates a hot shard; mitigate with routing partitions spanning several shards |
| Index per tenant or per time window | Only relevant indices | Too many small indices; cluster state overhead |
Scoring subtlety: each shard computes IDF from its own documents, so small or skewed shards can rank the same document differently. The DFS query-then-fetch mode gathers global term statistics first, at the cost of an extra round trip.