Elasticsearch in depth

Elasticsearch is a cluster of nodes that splits each index into shards, keeps a primary and replicas of every shard, and answers a search by asking one copy of every shard and merging the answers. Above Lucene’s segments sit an elected master that owns the metadata, a replication protocol built on sequence numbers, and aggregations that trade exactness for speed.

This is part 5 of the search engines series. Part 1 mapped the kinds of search, part 2 built the inverted index, part 3 put it on disk and part 4 chose the engine; here we open up the cluster itself. How search engines work gives the one-page tour; this page goes deeper, with Elasticsearch 9.5 (current docs, October 2026) and Bazaar’s 50 million products.

Picture it first: the district land-records office

A district’s land records are split across four registers by a fixed rule on the survey number, each in a different tehsil office with a certified copy in another, so a flood loses nothing.

A new sale deed goes to the office holding the main register for that survey number. The clerk numbers the entry, sends it to the copy, and says “registered” only when the copy has it too. When a lawyer asks for every plot owned by a Sharma, the front-desk clerk sends a peon to all four offices and merges their best matches.

The collectorate keeps the master list of which office holds which register and which copies are up to date. Three deputy collectors run it; any two signatures decide.

Map it back: registers are shards, the main register is the primary, the certified copy a replica, the fixed rule is routing, entry numbers are sequence numbers, the front-desk clerk is the coordinating node, the collectorate is the elected master keeping the cluster state, and two of three signatures is the quorum.

Cluster anatomy: nodes, roles and the master

Every Elasticsearch process is a node, and node.roles decide its jobs (node roles): master-eligible nodes may be elected to manage the cluster; data nodes (data_content, data_hot … data_frozen) hold shards, index and search; ingest, ml and transform nodes run pipelines and jobs; a coordinating-only node (empty roles) just routes requests and merges results, as every node does for requests it receives.

The elected master owns the cluster state: settings and mappings, the routing table of which shard copy sits on which node, and each shard’s set of in-sync allocation IDs. It allocates shards and publishes every new cluster state to all nodes. The work is light, but a master starved by searches stalls the cluster, so beyond a handful of nodes run three dedicated master-eligible nodes, with about 1 GB of heap per 3,000 indices (size your shards).

Elections and quorums. Every decision, electing a master or committing a cluster state, needs more than half of the voting configuration of master-eligible nodes (voting configurations). Three voters survive one failure, and a partition leaves a majority on at most one side, so two masters can never both commit. With four master-eligible nodes, one is left out of the vote so a 2–2 split still leaves one side working. You name the first voters once, in cluster.initial_master_nodes, when a new cluster bootstraps. It is the majority logic of consensus and leader election.

Before 7.0 you set discovery.zen.minimum_master_nodes to a majority by hand; get it wrong and two masters could form, split brain, losing writes. The 7.0 rewrite removed the setting, manages the voting configuration itself and elects a master in well under a second.

Indices, mappings, aliases and data streams

An index is a set of documents with one mapping, the schema for how each field is indexed. What matters most:

• text or keyword. text is analysed into terms for full-text matching (part 2); keyword keeps the exact value, with doc values for filters, sorting and facets. title is text; brand is a keyword.
• Multi-fields. One value indexed two ways, such as title plus a title.raw keyword for sorting. Each sub-field costs disk and counts as a field.
• Dynamic mapping. By default an unseen string becomes text plus a .keyword sub-field (dynamic field mapping). Let sellers send free-form attributes and you get thousands of fields: a mapping explosion. Every object, field and multi-field counts towards index.mapping.total_fields.limit, 1,000 by default (mapping limits), and every node holds mappings in heap. Use dynamic: strict and one flattened field for open-ended attributes.

You can add fields, but not change an existing field’s type or analyser, because every segment was written with the old one. That needs a reindex, so Bazaar’s search API only ever names an alias:

POST _aliases
{
  "actions": [
    { "remove": { "index": "products_v3", "alias": "products" } },
    { "add":    { "index": "products_v4", "alias": "products" } }
  ]
}

Both actions apply in one cluster-state update, so queries switch at once; rollback is the reverse call.

Data streams suit append-only, timestamped data such as Bazaar’s logs. You write to one name, logs-bazaar; behind it sit hidden backing indices (.ds-logs-bazaar-2026.10.27-000042), and writes go to the newest. Rollover starts a new one when the current index is big or old enough, so retention deletes whole indices, not billions of documents.

Shards and routing

Each index is split into primary shards, each a complete Lucene index with index.number_of_replicas copies (default 1). The primary count (default 1) is fixed at creation (general index settings) because it is in the routing formula, a plain modulo for indices created since 9.4.0 (_routing field):

shard = hash(_routing) % number_of_shards      # _routing is the document _id by default

Older indices keep a legacy formula built on the now-deprecated index.number_of_routing_shards, which also capped how far they could split.

Why you can’t just add a primary. Going from 4 to 5 primaries, a document stays put only if h % 4 == h % 5, true for 4 of every 20 hash values, so 80% of documents move. Consistent hashing avoids that; Elasticsearch offers three exits, each on a write-blocked source:

• Split into a multiple (4 → 8): segments are hard-linked, then each new shard deletes documents that aren’t its own.
• Shrink into a factor (8 → 4), after gathering a copy of every shard on one node.
• Reindex into any count and swap the alias: slowest, most flexible, the usual answer.

Replicas, by contrast, are dynamic: Bazaar goes from 1 to 2 before a sale for read capacity.

Sizing Bazaar’s shards

Elastic’s guidance is 10–50 GB and under 200 million documents per shard, at most 1,000 non-frozen shards per node, and one search thread per shard. Bazaar’s catalogue is about 100 GB per copy (part 3):

• 1 primary of 100 GB: a lost copy takes 100 × 1,024 ÷ 40 MiB/s = 2,560 s, about 43 minutes, to rebuild at the default recovery rate, and each search gets one thread.
• 40 primaries of 2.5 GB: every search fans out to 40 shard tasks, paying per-shard overhead for nothing.
• 4 primaries of 25 GB, 12.5 million documents each: inside the guidance, four threads per search, 8 copies on 4 nodes as in Fig. 1. Bazaar’s choice.

Custom routing

With routing=seller_42, a seller’s products share one shard, so “my listings” searches 1 shard instead of 4. You pay three ways: a mega-seller makes a hot shard, _id is unique only per shard, and every get, update and delete must repeat the routing value (index.routing_partition_size spreads each value over a few shards). Shopper search spans all sellers, so Bazaar keeps _id routing; database sharding covers the same trade-offs.

The write path: primary-backup replication

Elasticsearch uses a primary-backup model based on Microsoft’s PacificA paper (reading and writing documents). Follow a price update for P-48213 during a sale, with two replicas (Fig. 2):

1. Coordinating stage. The receiving node routes it: hash("P-48213") % 4 = 0, shard 0, primary on data-1.
2. Primary stage. The primary validates and indexes it, assigns the next sequence number in the current primary term, and appends it to its translog (part 3).
3. Replica stage. It forwards the operation in parallel to every copy in the in-sync set, kept by the master in the cluster state.
4. Acknowledge. Only when every in-sync replica has applied it does the primary answer.

A replica fails. The primary must never acknowledge a write an in-sync copy lacks, so it first asks the master to drop that copy from the in-sync set and waits for confirmation; the master also starts a fresh copy elsewhere. A copy outside the set can never be promoted, so nothing acknowledged is lost.

The primary fails. In-flight writes wait, up to a minute by default, while the master promotes an in-sync replica and increments the shard’s primary term. An old primary merely cut off by a partition or GC pause may accept writes briefly, but replicas reject its stale term. Meanwhile a reader can see a never-acknowledged write, a dirty read; pinging the master every second keeps that window short.

Why two copies are enough. A quorum store needs three copies to survive one failure (quorums and leaderless replication). Here the quorum lives in the master tier, which decides who is in sync, so two copies survive losing either. You pay in latency: every write waits for the slowest in-sync copy (data replication).

wait_for_active_shards defaults to 1, the primary alone. Raising it stops writes from starting while copies are missing, but it is a pre-check, not a promise about how many copies receive them (index API).

Sequence numbers, checkpoints and recovery

Elasticsearch 6.0 added sequence numbers after rebuilding replication, as Elastic’s 2017 post describes. Every operation carries a primary term, a generation of primaries stored in the cluster state, and a sequence number that orders operations within the shard. A higher term always wins, so a stale primary’s operations can be spotted and discarded.

Operations are forwarded concurrently, so a replica can apply 1,042 before 1,041. Each copy therefore tracks a local checkpoint, the highest sequence number below which it has everything, and the primary computes the global checkpoint, their minimum, sending it with later operations: in Fig. 2, min(1,042, 1,042, 1,040) = 1,040. Below it every in-sync copy has every operation, so after a failover the new primary reconciles replicas only above it instead of comparing whole shards.

Peer recovery uses the same marker: a returning replica replays everything above its last global checkpoint. Deletes can’t be replayed from Lucene, so recent ones are kept as soft deletes, held by a shard history retention lease per copy that expires after 12 hours by default; after that, recovery copies segment files.

For Bazaar: data-3 reboots for a kernel patch and returns in 6 minutes. Replicas are rebuilt elsewhere only after index.unassigned.node_left.delayed_timeout (1 minute by default; raise it above your patch window). At 30,000,000 changes ÷ 86,400 s ≈ 347 a second, about 87 per shard, six minutes is 87 × 360 ≈ 31,000 operations to replay: seconds, instead of 11 minutes copying 25 GB.

Concurrency control for clients

A read returns _seq_no and _primary_term; send them back as if_seq_no and if_primary_term, and the write succeeds only if nobody changed the document since, otherwise 409 Conflict (optimistic concurrency control). Use it for read-modify-write, such as a merchandiser editing pinned keywords (optimistic vs pessimistic locking).

Bazaar’s indexer uses part 4’s external versioning instead: version_type=external with the database row’s version, accepted only if higher than the stored one. A replayed Kafka message gets a 409, which the indexer skips: an idempotent consumer for free.

The search path in depth

Here is what the coordinating node does with Bazaar’s request for page 3 of “cotton kurta”, 20 hits a page (Fig. 3).

Pick a copy of each shard. Adaptive replica selection (search shard routing) prefers nodes with the best recent response times, service times and search queues, so a node in a GC pause gets less traffic. preference=<session id> pins a shopper to the same copies; otherwise ties, broken by per-copy Lucene doc IDs, can reorder between page loads.

Query phase. Each shard returns its top from + size = 40 + 20 = 60 (doc ID, score) pairs plus partial aggregations. Totals are exact only up to track_total_hits, 10,000 by default.

Merge, then fetch. The coordinator merges 4 × 60 = 240 entries, keeps ranks 41–60, and fetches _source only from the shards owning those 20.

Hence index.max_result_window caps from + size at 10,000. Deeper, page by keys, as in offset vs keyset pagination:

search_after needs a unique tiebreaker; a PIT adds _shard_doc for you (paginate search results). Bazaar caps its UI at 50 pages and exports with search_after and a PIT.

Scores are computed per shard

Each shard computes BM25’s N, document frequencies and average length from its own documents (part 2). With 12.5 million randomly routed products per shard, “kurta” in 100,300 titles on one shard and 99,700 on another gives IDFs of 4.825 and 4.831, about 0.1% apart. Route by seller, and on a 2,000-product shard where 1,500 titles say “kurta”, IDF = ln(1 + 500.5 ÷ 1,500.5) ≈ 0.29, a seventeenth of the rest. search_type=dfs_query_then_fetch first gathers global term statistics: consistent scores for an extra round trip. Use it on small or skewed indices.

Filters, caches and fielddata

Put clauses that needn’t score (in_stock, category, price ranges) in a bool filter: no scoring, and cacheable in the node query cache, up to 10% of heap, per segment, for segments with at least 10,000 documents and 3% of the shard; plain term queries are never cached (query cache). The shard request cache (1% of heap) keeps whole size: 0 responses such as aggregations until the shard refreshes (request cache): useless on a catalogue refreshing every second, ideal for last week’s logs. Aggregate on keyword doc values; text fields would need heap-hungry fielddata, off by default.

Aggregations and their approximations

Shards compute partial aggregations from doc values and the coordinator reduces them. Sums and maximums merge exactly; three common aggregations don’t.

Terms: top N per shard. A terms aggregation takes each shard’s top shard_size terms, by default size × 1.5 + 10 (terms aggregation), sums them and keeps the top size. A term just below one shard’s cut is undercounted. Fig. 4 forces size and shard_size to 3 on Bazaar’s brand facet to show it.

Elasticsearch bounds the damage. A shard that didn’t report a brand holds at most its smallest reported count, so with show_term_doc_count_error each bucket gets an upper bound: 100 for Rangrez (missing from shard 0), 85 for Noor, 0 for Kesar. The response-level doc_count_error_upper_bound, 100 + 85 + 70 = 255, bounds any brand that missed the list.

Fixes, cheapest first: keep the default shard_size, here 3 × 1.5 + 10 = 14, which catches all four brands; raise it for long-tailed fields; route by the facet key so each term lives on one shard; or walk every term exactly with a composite aggregation. Never sort by ascending count; use rare_terms.

Cardinality: HyperLogLog++. The cardinality aggregation avoids per-shard hash sets with HyperLogLog++: near-exact below precision_threshold (default 3,000, maximum 40,000), at about threshold × 8 bytes, 24 KB by default, and 1–6% error in Elastic’s tests even at a threshold of 100.

Percentiles: t-digest. Percentiles use a t-digest with compression 100: at most 2,000 nodes of about 32 bytes, about 64 KB. Accuracy is proportional to q(1 − q), so p99 is more accurate than the median.

Tuning relevance

BM25 (part 2) is only the first-stage scorer. Product search shapes it in four layers, cheapest first.

1. Field boosts. A multi_match over title^3, brand^2 and description counts a title match three times a description match.
2. BM25 parameters per field. Similarity settings take k1 (default 1.2) and b (default 0.75). A lower b suits short, similar-length titles; a lower k1 saturates repeated words sooner, blunting keyword stuffing.
3. Business signals. function_score can multiply BM25 by field_value_factor on 30-day sales with log1p, meaning log₁₀(1 + sales) (function score), plus a gauss decay on delivery days. Part 2’s kurta A (3.83, 10 sales) becomes 3.83 × log₁₀ 11 = 3.99; kurta B (3.58, 2,000 sales) becomes 3.58 × log₁₀ 2,001 = 11.82. Popularity now outweighs text threefold; cap it with max_boost or boost_mode: sum if that’s not what you want.
4. Rescoring. A rescore re-ranks only each shard’s top window_size hits (default 10) with a costlier query or a learning-to-rank model, so a window of 200 treats 4 × 200 = 800 documents, not 200,000 matches.

Judge every change against rated queries first; search end to end covers that loop.

Vectors and hybrid search in 9.x

A dense_vector field holds up to 4,096 dimensions (dense vector). Lucene builds an HNSW graph per segment, so kNN walks every segment’s graph and merges rebuild graphs (part 3). Defaults moved fast: int8 in 9.0; binary quantisation (bbq_hnsw) for 384+ dimensions since 9.1; since 9.4, under an Enterprise licence, bbq_disk, which clusters vectors with hierarchical k-means and reads them from disk instead of a graph in memory.

A knn search examines num_candidates per shard (default 1.5 × k, at most 10,000) and returns each shard’s top k for the coordinator to merge (knn query): Bazaar’s k = 50 with 150 candidates examines 4 × 150 = 600 and keeps 50. An rrf retriever fuses the BM25 and kNN lists by rank, rank_constant 60 (reciprocal rank fusion). Memory is part 4’s arithmetic: 42.4 GB (int8) or 8.7 GB (BBQ) per copy for 50 million 768-dimension vectors; see vector databases for alternatives.

Data tiers and lifecycle

Bazaar’s catalogue is content and stays on the content tier. Its logs are time series, about 1 TB a day (part 4), losing value with age. Data tiers match hardware to that: hot SSD nodes take writes and most searches; warm nodes are cheaper; cold holds fully mounted searchable snapshots, backed by object storage so no replica is needed, roughly halving disk; frozen holds partially mounted snapshots with only a local cache, up to 20 times more data per node than warm, slower on cache misses.

Index lifecycle management (ILM) moves each backing index along (Fig. 5). Bazaar rolls over at 50 GB per primary shard, Elastic’s recommended ceiling (rollover), or after a day; with 5 primaries that is 250 GB, a new index about every 6 hours. Ages count from rollover: warm at 2 days (force-merged to one segment, as part 3 advised), cold at 7, frozen at 30, delete at 90.

All 90 days hot with a replica is 90 × 1 TB × 2 = 180 TB of SSD. Tiered, local disk holds 2 × 2 + 5 × 2 + 23 × 1 = 37 TB plus frozen caches, with the oldest 60 TB in object storage. You gain a fivefold cut; you pay with slower old-data searches, a snapshot repository, and an Enterprise licence for searchable snapshots. Data stream lifecycle is the simpler alternative and can move indices to frozen since 9.5; design a distributed logging system builds on this.

Going deeper: failure modes and the numbers that warn you

• Split brain is impossible since 7.0, but availability needs a majority: never run exactly two master-eligible nodes.
• Yellow health means some replicas are unassigned; red means a primary is, so part of an index can’t be searched or written.
• Flood stage. At 85% disk a node gets no new shards, at 90% shards move away, and at 95% every index with a shard there turns read-only until usage drops below 90% (disk watermarks). Alert well before 85%.
• Heap and GC pressure. Keep the heap at most 50% of RAM and under the compressed-oops threshold, about 26 GB (JVM settings). Huge terms sizes, deep pages, fielddata and thousands of shards fill it, and long pauses make nodes drop out.
• Oversharding and hot shards. Daily 5-shard indices kept a year are 1,825 small shards; roll over by size. A mega-seller under custom routing makes a hot node, like hot keys in a cache.
• Mapping explosion and deep pages. Dynamic fields hit the 1,000-field limit and documents bounce; a scraper at from=9,980 costs 40,000 sorted entries a page.
• Bulk sizing and 429s. Benchmark from 100 documents per bulk, doubling until throughput plateaus, under a couple of tens of megabytes (indexing speed). A 429 means queues are full: back off with jitter (retries and backoff).

Cheat sheet

How to say it in the interview

“Bazaar’s catalogue is one index behind an alias: 4 primaries of about 25 GB, one replica each, on 4 data nodes, with 3 dedicated masters elected by majority, so a partition can’t create two masters. Documents route by hash of the ID modulo 4, so to change the shard count I reindex and swap the alias. The primary assigns each write a sequence number and waits for every in-sync replica; the indexer sends external versions so replays can’t overwrite newer data. A search takes the top from + size from one copy of each shard, merges, and fetches only the winners; deep exports use search_after with a point in time. Facet counts are top-N per shard, so I keep shard_size generous. Logs go to a data stream with rollover through hot, warm, cold and frozen tiers.”

What happens if the node holding a primary dies mid-write? The write waits, up to a minute by default, while the master promotes an in-sync replica and bumps the primary term, then retries on the new primary. Every acknowledged write is on every in-sync copy, so none is lost.

Why are two copies enough here when quorum stores need three? The master tier, itself a majority quorum, decides which copies are in sync, and a replica that misses a write is removed before that write is acknowledged. The price: every write waits for the slowest copy.

Why might a facet count be wrong, and how would you know? Each shard sends only its top shard_size terms, so a term just below one shard’s cut is undercounted. doc_count_error_upper_bound bounds the error; raise shard_size or route by the facet key.

How would you go from 4 to 6 primaries? Not with split, which needs a multiple. Build products_v5 with 6 primaries from a database snapshot, replay Kafka from the snapshot’s position, compare results, swap the alias and keep the old index for rollback.

A node restarts for patching. Does it copy 25 GB back? Not if it returns within the delayed-allocation timeout while retention leases hold history: it replays only what it missed, about 31,000 operations for Bazaar.

Where to go next

• Next in the series: How Google Search works: crawling, indexing and ranking at web scale.
• Earlier parts: Types of search, Search data structures, Disk I/O and durability for search and Choosing a search engine.
• Elasticsearch: the short version, what interviewers expect you to know.
• Search end to end: the whole Bazaar pipeline from CDC to the ranking funnel.
• Data replication and Consensus and leader election: the theory behind primaries and masters.
• Design product search for an online store: this cluster in an interview.

References and further reading

1. Elasticsearch documentation: Reading and writing documentselastic.coThe PacificA-style primary-backup model, in-sync copies kept by the master, the one-minute wait for a new primary, dirty reads and why two copies suffice.
2. Elasticsearch documentation: Voting configurationselastic.coMajority decisions over the voting configuration, why an even number of master-eligible nodes drops one voter, and cluster bootstrapping.
3. A new era for cluster coordination in Elasticsearch (Elastic blog, 2019)elastic.coThe 7.0 rewrite: minimum_master_nodes removed, elections in well under a second, voting configuration managed automatically.
4. Elasticsearch sequence IDs: what, why and how (Elastic blog, 2017)elastic.coPrimary terms, sequence numbers, local and global checkpoints, and operation-based replica recovery from 6.0.
5. Elasticsearch documentation: History retention settingselastic.coSoft deletes, shard history retention leases, their 12-hour default expiry and the fallback to copying files.
6. Elasticsearch documentation: _routing fieldelastic.cohash(_routing) % number_of_shards for indices created in 9.4.0 and later, the legacy routing-shards formula, custom routing and partitions.
7. Elasticsearch documentation: Size your shardselastic.co10–50 GB and 200 million documents per shard, one thread per shard per search, 1,000 non-frozen shards per node, 1 GB of master heap per 3,000 indices.
8. Elasticsearch documentation: General index settingselastic.coDefaults for number_of_shards, number_of_replicas and max_result_window, and the deprecation of number_of_routing_shards in 9.4.
9. Elasticsearch documentation: Terms aggregationelastic.coshard_size = size × 1.5 + 10, sum_other_doc_count, doc_count_error_upper_bound and show_term_doc_count_error.
10. Elasticsearch documentation: Data tierselastic.coContent, hot, warm, cold and frozen tiers, and the storage savings of fully and partially mounted searchable snapshots.