How to Shard a Large Vector Index Without Losing Recall

Here is how to shard a large vector index: hash-partition vectors by ID into N shards, build an independent HNSW or IVF index on each, send every query to all shards asking each for the full top-k, and merge by distance. Choose N so each shard’s index fits in RAM with headroom, as a multiple of your node count.

The page that starts this work is usually an OOM kill, or an index rebuild that no longer fits the maintenance window: the memory sizing arithmetic says one box is not enough. Sharding fixes capacity. It does not fix throughput, it worsens tail latency, and most engines lock the shard count at creation.

When do you actually need to shard a vector index?

Shard when one node cannot hold the index resident with operating headroom, or when ingest and index builds outgrow one machine. Do not shard to raise QPS; add replicas for that. Weaviate’s architecture docs say it directly: query throughput does not improve when adding more sharded nodes, while import speed does.

HNSW search has logarithmic complexity scaling with corpus size, so N small searches cost more total CPU than one big one: every shard still descends its own graph. Faiss’s GPU docs show the same shape. Replicating over 8 GPUs gives near linear speedup, 6 to 7x, while sharding yields “some speedup, but is more sub-linear.” That is Faiss project documentation, not an independent benchmark.

Compress first (see product quantization), consider a disk-resident index second, and shard when the compressed index still does not fit.

The metric that matters: slowest-shard latency

A scatter-gather query finishes when the slowest shard answers, so coordinator latency is the max over N shard latencies, not the mean. If each shard independently exceeds its own p99 on 1% of requests, the probability that a fanned-out query hits at least one slow shard is:

P(slow query) = 1 - (1 - 0.01) ** N

That works out to about 7.7% at 8 shards, 14.9% at 16 and 27.5% at 32, assuming independence. Dean and Barroso’s The Tail at Scale used the same arithmetic for a request that fans out to 100 servers: 63% of user requests take more than one second when each server is slow on one request in a hundred. At 8 shards, your shard p99 is roughly your coordinator p92.

Dashboard the gap between the worst per-shard p99 and coordinator p99, which is the fan-out tax, next to merged recall@k against exact search over the whole corpus. Mean shard latency hides the problem: adding shards can lower it while coordinator p99 climbs.

How many shards, and how big?

Use the smallest shard count that keeps each shard’s index resident in one node’s RAM with headroom, rounded up to a multiple of your node count and padded for the growth you expect before the next rebuild. Derive it from memory, never from QPS:

min_shards = ceil(index_bytes_per_copy / (node_ram_bytes * fill_target))
shards     = smallest multiple of node_count >= min_shards * growth_factor

Keep fill_target well below 1.0 so payload indexes, build headroom and the page cache still fit. Elastic warns that once the working set no longer fits in cache, latency climbs sharply. Per-vector graph cost scales with M, so tuning HNSW M changes index_bytes_per_copy before you touch the shard count.

Engines disagree on what a shard is:

• Qdrant. shard_number defaults to the node count at creation and cannot change without recreating the collection, except through resharding in Qdrant Cloud from v1.13.0, which the docs say is slow on large collections. Qdrant recommends at least 2 shards per node, and 12 for heavy growth because 12 spreads evenly over 1, 2, 3, 6 and 12 nodes.
• Weaviate. Objects are hashed by UUID, and each shard carries its own vector index, inverted indexes and object store. Single-tenant shard count is fixed at creation, and adding a node does not move existing shards; replica moves for rebalancing arrived in v1.32.
• Milvus. A collection gets one shard by default, and a shard is a data input channel hashed on primary key. The docs suggest one shard per 200 million entities, or one per 100 GB inserted. Shard count here is mostly an ingest decision.
• Elasticsearch. Each shard returns k results that the coordinator merges into the global top k; num_candidates is also per shard, defaulting to 1.5 times k, capped at 10,000. Each segment is its own HNSW graph, a hidden fan-out inside the shard.

Which partitioning scheme: hash, tenant, or semantic?

Use hash partitioning by default. It keeps shards the same size and the same temperature, and when every shard returns its own top-k, merged recall equals per-shard recall. Route by tenant only when every query carries the tenant filter. Treat semantic, cluster-based partitioning as an optimization for very large fleets where fan-out cost dominates.

The hash case is lossless at merge time for a simple reason. An item in the global top-k has fewer than k items closer to the query across the whole corpus, so it has fewer than k closer items in its own shard and must appear in that shard’s local top-k. The only error left is each shard’s ANN error. The corollary: never ask each shard for k divided by N.

Tenant routing removes fan-out for scoped queries. Qdrant’s custom sharding (sharding_method=models.ShardingMethod.CUSTOM, then shard_key_selector on writes and queries) maps tenants to shard keys, but physical shards are resource-heavy, so the docs ask for a low-cardinality key, and any operation without a shard key runs on all shards. For thousands of tenants, Milvus hashes a partition key into 16 partitions by default and requires the key in the search filter. Multitenancy support varies a lot by engine; the vector database selection guide compares them.

Semantic partitioning groups similar vectors and searches only the shards a query likely needs, the approach of Pyramid. Gottesbüren et al. pair balanced graph partitioning with learned routing and report up to 2.14x higher QPS at 90% recall@10 than the best competing partitioner on billion-scale datasets; that is the authors’ own measurement, not an independent replication. The bill: a router that skips the right shard drops neighbors outright, popular regions become hot shards, and partitions drift as the corpus changes.

Wiring it up

On a managed engine, sharding is collection config. Qdrant with 12 logical shards and 2 replicas:

from qdrant_client import QdrantClient, models

client = QdrantClient(url="http://qdrant-0.internal:6333")

client.create_collection(
    collection_name="chunks_v3",
    vectors_config=models.VectorParams(size=768, distance=models.Distance.COSINE),
    shard_number=12,
    replication_factor=2,  # 24 physical shards, each replica resident in RAM
)

Self-managed shards (Faiss, hnswlib or pgvector per node) need a coordinator that asks every shard for the full k and records per-shard latency:

import asyncio
import heapq
from prometheus_client import Counter, Histogram

K = 10
SHARD_TIMEOUT_S = 0.25

SHARD_LATENCY = Histogram(
    "vector_shard_search_seconds", "Per-shard ANN search latency", ["shard"]
)
SHARD_DROPPED = Counter(
    "vector_shard_dropped_total", "Shards missing from a merged result", ["shard"]
)

async def search_one(shard_id, client, query_vec, k):
    with SHARD_LATENCY.labels(shard=shard_id).time():
        # returns [(distance, doc_id), ...]; lower distance is closer
        return await client.search(query_vec, k=k)

async def sharded_search(shards, query_vec, k=K):
    # Every shard returns the full k, never k // len(shards)
    calls = [
        asyncio.wait_for(search_one(sid, c, query_vec, k), SHARD_TIMEOUT_S)
        for sid, c in shards.items()
    ]
    results = await asyncio.gather(*calls, return_exceptions=True)
    hits = []
    for sid, res in zip(shards, results):
        if isinstance(res, Exception):
            SHARD_DROPPED.labels(shard=sid).inc()
            continue
        hits.extend(res)
    return heapq.nsmallest(k, hits)

For Faiss IVF shards in one process, faiss.IndexShards does the merge, with two traps from the Faiss FAQ: index_shards.nprobe = 123 in Python silently does nothing (loop over sub-indexes or use ParameterSpace), and strongly compressed indexes trained separately produce distances that are hard to compare. Train one quantizer on a representative sample and share it.

What you’ll see

Healthy hash sharding looks boring: near-identical per-shard vector counts, per-shard p99 lines stacked together, coordinator p99 a little above the worst shard, and merged recall@10 matching the single-node number for the same index config. Benchmark recall@k with ground truth from exact search over the full corpus, never per shard.

The bad shapes are diagnostic:

• One shard’s p99 peels away. A hot tenant or semantic hot spot, a node mid-compaction, or an Elasticsearch shard accumulating segments.
• Coordinator p99 rises with each shard added while per-shard p99 falls. Pure fan-out tax: you sharded past what memory required, when replicas were the right tool.
• Merged recall drops right after the change while per-shard recall is fine. Per-shard k below global k, separately trained quantizers, or a low num_candidates, which Elasticsearch applies per shard.
• Memory skew after scale-out. New Weaviate nodes stay empty until you move shard replicas onto them.

Caveats

Shard count is a one-way door on most engines, so overshard modestly rather than wildly; every physical shard carries its own index structures and overhead.

A timed-out shard means silently lower recall: k results drawn from N-1 shards. Count the drops, alert on their rate, and decide per endpoint whether degraded beats failed. Dean and Barroso’s hedged requests help: send a backup to a replica only once the first request outlives the 95th-percentile expected latency, which they report adds about 5% load.

Selective filters interact badly with hash sharding, because each shard runs its own filtered traversal and the slowest one sets latency; if the filter is always present, route on it instead. Label metrics by shard, which is bounded, and never by tenant or query, which is not.

FAQ

should i shard or replicate my vector database

Shard when the index will not fit in one node’s RAM; replicate when it fits but one node cannot serve the query rate or you need failover. Sharding adds capacity and fan-out latency. Replication adds throughput and availability but multiplies memory, because every replica keeps a full copy of its shard resident.

does sharding a vector index reduce recall

No, not if every shard returns the full top-k and results are merged by a comparable distance. Any true global neighbor is also in its own shard’s local top-k, so merging loses nothing beyond per-shard ANN error. Recall drops when shards return k divided by N, use separately trained quantizers, or a router skips shards.

can you change the shard count of a vector collection later

Usually not in place. Qdrant fixes shard_number at creation except for resharding in Qdrant Cloud from v1.13.0, which is slow on large collections, and Weaviate fixes the shard count of single-tenant collections at creation. The portable path is a new collection with the target count, dual writes, a backfill, then a read cutover.