Shard Geometry

May 11, 2025

Managing Search Latency made the product query cheap by baking the business score into a stored field and pushing the price anchor into a rescore window. All of it runs on top of the index’s physical layout, which has been the untouched default since day one.

products_semantic_v1 was created without a shard count, so OpenSearch filled in a single primary shard. Everything since has been running on it, embeddings and hybrid retrieval included.

For a while that’s fine. Shard geometry stays invisible while the catalog is small and becomes expensive to change right around the time it starts to matter. It looks like capacity planning. It also, less obviously, changes what scores come back.

One shard by default

A shard is a complete Lucene index with its own segments and caches. When a query arrives, every shard of the index runs it independently and produces its own ranked results, like a small search engine that only knows about its slice of the catalog.

OpenSearch creates indexes with index.number_of_shards: 1 unless told otherwise. One shard means no fan-out and nothing to merge, which is genuinely the right call for small indexes. It also means a single Lucene index holds the whole catalog, so one node does all the search work for that copy and a recovery has to move everything at once.

Fan-out

With more than one shard, a search becomes a scatter and gather. The coordinator picks a copy of each shard and sends the query. Every shard runs it against its own slice and answers with its top size candidates, then the coordinator merges those lists and fetches the winners.

  flowchart TD
    Q["<strong>Query</strong><br/><span class='mermaid-detail'>walnut record cabinet</span>"]
    C["<strong>Coordinator</strong>"]
    S0["<strong>Shard 0</strong><br/><span class='mermaid-detail'>top 10</span>"]
    S1["<strong>Shard 1</strong><br/><span class='mermaid-detail'>top 10</span>"]
    S2["<strong>Shard 2</strong><br/><span class='mermaid-detail'>top 10</span>"]
    M["<strong>Merge</strong><br/><span class='mermaid-detail'>30 candidates, top 10 out</span>"]

    Q --> C
    C --> S0
    C --> S1
    C --> S2
    S0 --> M
    S1 --> M
    S2 --> M

Every shard you add buys parallelism and charges overhead. The Elasticsearch sizing guidance puts it bluntly: searching a thousand 50 MB shards is substantially more expensive than searching a single 50 GB shard holding the same data.

Custom _routing can pin queries to one shard when traffic is tenant-shaped. A shared furniture catalog isn’t, so every search visits every shard, and the shard count decides how wide that fan is.

Too many shards

Each shard carries fixed costs whether or not it holds much data: heap for segment metadata and a slot in the cluster state. Each query pays per shard too, since a full query execution and a results queue exist for every one of them.

Oversharding gets you the overhead without the parallelism. Ten shards holding 6 GB each answer the same question as three shards holding 20 GB, but the coordinator merges ten lists instead of three, and the cluster babysits ten sets of segments. Small shards also warm slowly, because each one maintains its own caches over a sliver of the data.

Too few shards

In the other direction, a shard is the unit of recovery and rebalancing. When a node dies, its shards move as whole units, and a 60 GB shard takes a long time to copy while the cluster runs with reduced redundancy. Merges get heavier as segments grow, and a single query can’t spread one shard’s work across nodes. The shard is the ceiling on per-query parallelism.

The folk band is 10 to 50 GB per shard. Like most folk wisdom it encodes real trade-offs: big enough that fixed costs amortize, small enough that recovery and merges stay boring.

The sizing math

The catalog is at five million products now. The storage shape, following the embedding storage math:

5,000,000 products

text, metadata, _source     ~5 KB each        25 GB
vectors, 1,536 dims, FP32   6.1 KB each       31 GB
HNSW graph, m = 16          128 bytes each    0.6 GB
                                              ------
                                              ~57 GB

The vectors quietly became the majority of the index. That’s normal for an embedding-backed catalog and worth noticing, because it means shard sizing is now dominated by a field no shopper ever sees.

One shard at 57 GB sits past the band. Ten shards would be 5.7 GB apiece and pay fan-out for nothing. Three shards land around 19 GB each. I aim for the middle of the band rather than the top of it, because catalogs only grow.

PUT /products_semantic_v2
{
  "settings": {
    "index": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "knn": true
    }
  }
}

The mappings are identical to v1. Only the settings change.

After reindexing, _cat/shards shows the layout:

GET /_cat/shards/products_semantic?v&h=index,shard,prirep,store,node

index                 shard  prirep  store    node
products_semantic_v2  0      p       19.2gb   search-node-1
products_semantic_v2  0      r       19.2gb   search-node-3
products_semantic_v2  1      p       18.7gb   search-node-2
products_semantic_v2  1      r       18.7gb   search-node-1
products_semantic_v2  2      p       19gb     search-node-3
products_semantic_v2  2      r       19gb     search-node-2

Three primaries spread across three nodes, each with a replica on a different node. A broad query now runs on all three at once instead of one.

Resizing

Shard count is fixed at index creation. OpenSearch has _split and _shrink to multiply or divide the count, but both require blocking writes on the index first, which makes them tools for quiet hours rather than live catalogs.

For a catalog that can’t pause, the path is reindexing with zero downtime: create the new index with the geometry you want, reindex into it, and swap the alias. That discipline of always querying through the alias is exactly what makes a reshard a routine operation instead of a migration project.

Shards change scores

BM25 statistics are computed per shard. The default query_then_fetch scores documents with each shard’s local term and document frequencies, so a term’s IDF on shard 0 isn’t quite the IDF on shard 2. At five million randomly routed documents the shards are statistically near-identical and nobody can tell. The place it bites is small or skewed indexes, and the classic victim is the staging cluster holding two percent of production data: same code, same query, visibly different order.

dfs_query_then_fetch fixes the statistics by collecting global frequencies in an extra round trip:

GET /products_semantic/_search?search_type=dfs_query_then_fetch

It costs latency on every query, so treat it as a diagnostic. If result order changes meaningfully between the two search types, your shards are too small or your routing is skewed.

The vector branch has its own version. In the k-NN query, k is per shard: each shard searches its own HNSW graph k deep, then sends its top size to the coordinator, which keeps the best size of the merged pool. The hybrid query runs with k: 100. On one shard that’s a single graph searched 100 deep. On three shards it’s three graphs searched 100 deep each, with the coordinator picking ten from thirty. More shards examine more candidates in total, which costs more and can lift recall. Either way the page can change.

Which means a reshard is a ranking change that never touched the query body. The lexical scores shift a little and the vector candidate pool shifts more. RRF fuses whatever those branches now produce.

Replicas

Replicas are the setting people reach for when queries feel slow, and half the time it’s the wrong knob. A query uses one copy of each shard, so adding replicas doesn’t make any single query faster. What replicas buy is concurrency and survival: more copies to spread queries across, and a copy left standing when a node disappears.

Unlike shard count, the replica count is live and adjustable:

PUT /products_semantic/_settings
{
  "index": {
    "number_of_replicas": 2
  }
}

Storage multiplies accordingly: 57 GB of primaries becomes 171 GB on disk with two replicas. Primaries are the parallelism knob and replicas are the throughput knob. They look interchangeable on a whiteboard right up until a node fails or a sale starts.

Evaluation

A resize needs the same two checks as the latency work, and for once the relevance check is the less obvious one.

Measure latency under production-like concurrency, before and after, with attention on the p99. Fan-out moves both ends of the distribution: more shards can cut the p50 on broad queries while the extra merge work piles up at the tail.

Then re-run the judged-query loop. The previous section is the reason: per-shard BM25 statistics and per-shard k both moved, so nDCG@10 should land within noise of the old geometry, and a real gap usually points at the vector branch. Once the alias swaps, watch the online metrics with the same suspicion you’d give any ranking deploy.

The single-shard default was a fine place to start and a bad place to stay, and the move between those two states is cheapest while the index is still small. Pick the geometry deliberately and write down why. When the catalog outgrows it, treat the resize as what it actually is: a ranking change that happens to arrive in an ops ticket.