Blog

Distribution: Shards, Replicas and Consistency

Written by

What happens when one machine isn’t enough. Shards split the data, replicas copy it, and the consistency factors decide what “written” and “read” mean when a node is missing.

Everything so far has assumed one Qdrant. This part is what changes when there are several.

Two separate problems get solved by two separate mechanisms, and conflating them is the most common confusion in this area. Sharding is for data that doesn’t fit or load that one node can’t serve. Replication is for surviving a node that stops responding. You can want either without the other.

Qdrant 1.19.1. The per-collection defaults below were read from a fresh collection.

Try this first

A three-node cluster, replication_factor=2. One node loses power mid-write.

Answer three questions:

  1. Can the cluster still serve reads?
  2. Can it still accept writes?
  3. If a write was acknowledged a second before the failure, is it still there?

The third one is the interesting one, and the honest answer is “it depends on a setting you chose earlier”.

Shards

A collection is divided into shards, and each shard lives on a node. A point’s shard is chosen by hashing its id, so points spread roughly evenly.

{
  "shard_number": 1,
  "replication_factor": 1,
  "write_consistency_factor": 1
}

Those are the defaults — a single shard, a single copy. That’s a single-node collection, and for a great many workloads it is the right answer for a long time.

A query goes to every shard, each searches its own portion, and the results are merged. So sharding gives you:

  • Capacity. Ten shards across ten nodes means each holds a tenth of the vectors, so a collection can exceed one machine’s memory.
  • Parallelism. Each shard searches concurrently, on its own CPU.

And it costs you:

  • Fan-out. Every query touches every shard, so the query’s latency is the slowest shard’s latency. More shards means more chances that one of them is having a bad moment. This is the tail-latency problem that fan-out always brings, and it is the main reason not to shard more than you need.
  • A number you can’t easily change. Shard count is set at creation. Qdrant supports resharding, and it involves moving a lot of data, so it is an operation rather than a setting.

Pick shard count for your target size, not your current one, and remember that more shards is not better — it’s more fan-out.

Replicas

replication_factor is how many copies of each shard exist, on different nodes.

client.create_collection(
    "docs",
    vectors_config=models.VectorParams(size=384, distance=models.Distance.COSINE),
    shard_number=6,
    replication_factor=2,
    write_consistency_factor=1,
)

With replication_factor=2, every shard exists twice. Lose a node and every shard it held is still available elsewhere. Replicas also serve reads, so they add read throughput as well as resilience.

The cost is arithmetic: the collection occupies replication_factor times the storage and memory. A factor of 2 doubles your hardware bill. That is the price of surviving a node, and it’s worth stating plainly because it’s often a surprise.

The consistency factors

Here is where the real decisions are.

write_consistency_factor

How many replicas must confirm a write before it is acknowledged. Default 1.

With replication_factor=2 and write_consistency_factor=1, a write is acknowledged as soon as one replica has it. Fast, available — and if that node dies before the second copy catches up, the write can be lost.

Raise it to 2 and both replicas must confirm. Now an acknowledged write survives one node failing. In exchange, a write fails if fewer than two replicas are reachable — you have traded availability for durability, which is the trade, not a bug.

That is the answer to the third opening question: whether an acknowledged write survived depends on whether you required more than one replica to confirm it.

Read consistency

Set per query, deciding how many replicas must agree before a result is returned.

hits = client.query_points(
    "docs", query=vector, limit=10,
    consistency=models.ReadConsistencyType.MAJORITY,
).points

The options trade latency against the risk of reading stale data:

  • all — every replica must respond and agree. Slowest, most consistent, least available.
  • majority — more than half must agree.
  • quorum — a quorum must respond.
  • a number — that many replicas.
  • default — one replica answers. Fastest, and can serve data that replica hasn’t caught up on.

A replica that was briefly unreachable and is catching up will happily answer a default read with data that is behind. Whether that matters depends entirely on your product: for document search it usually doesn’t, and for “did my write land” it very much does.

Three nodes, two shards, replication factor 2 node A shard 1 shard 2 node B shard 1 shard 2 node C shard 1 shard 2

Schematic — no timings. What differs between the two scenarios is a single setting, and what it decides is whether an acknowledged write can disappear with a node, or whether writes stop instead. There is no setting that gives both.

What you’re actually choosing

Put the two factors together and the shape is familiar to anyone who has configured a distributed database:

you want set you give up
writes never blocked by a slow node write_consistency_factor=1 an acknowledged write can be lost with a node
acknowledged writes survive a node loss write_consistency_factor ≥ 2 writes fail when too few replicas are reachable
fastest reads consistency=default reads can be stale
reads that reflect acknowledged writes consistency=majority or higher latency, and failures when replicas are down

There is no configuration that gives you all four, and that isn’t a limitation of Qdrant. It’s the same constraint every replicated system faces: when a network partition means some nodes can’t be reached, you either refuse the operation or proceed without them, and those are different products.

Most vector search workloads should be at the top-left: fast writes, fast reads, tolerating a small staleness window — because a search index is usually a derived view of data that lives somewhere else, and can be rebuilt. If yours is a system of record, that reasoning doesn’t apply and you should be paying for consistency.

Explain it like I’m ten

Imagine a library so big it won’t fit in one building.

Sharding is splitting the books across several buildings, alphabetically-ish. Each building holds a fraction, so the whole library can be much bigger than one building. When someone asks for “books about volcanoes” you have to ask every building, and you can’t answer until the slowest one replies — so more buildings means more waiting on whoever is having a bad day.

Replication is keeping a second copy of each building’s books in a different building. Now if one burns down, nothing is lost, and twice as many people can read at once. It also costs you twice the shelves.

The consistency factors are the rules for what counts as done. If a donated book counts as “in the library” the moment one building has written it down, that’s fast — and if that building burns down tonight, the book was never really there. If you insist two buildings write it down first, the book is safe, and on a day when one building is closed you can’t accept donations at all.

Where the analogy breaks: a building that burns down stays gone. A node usually comes back a few minutes later, having missed everything, and starts confidently answering questions with what it knew before it left. That’s the staleness problem, and it has no fire-safety equivalent.

The precise version

A collection of N points is partitioned into S shards by a hash of the point id, so each holds roughly N/S points. Search is a scatter-gather over all S: total work is Θ(N) as before, but wall-clock latency is max over shards, so the p99 of a fan-out query is governed by the tail of the per-shard distribution, which worsens as S rises even when mean per-shard latency falls.

Each shard has R replicas. Storage and memory are R·N·(vector + graph + payload) — the factor is exact.

Let W = write_consistency_factor and r the replicas a read consults. A write is acknowledged once W replicas confirm; a read consults r. The classic condition for a read to observe every acknowledged write is W + r > R. With the defaults W = 1, r = 1, R = 2, that is 2 > 2 — false — so a default read may miss a recently acknowledged write, by design.

Setting W = 2 and r = 1 gives 3 > 2, and reads then see acknowledged writes; the cost is that writes require both replicas, so the collection stops accepting writes when one is unreachable. This is the availability/consistency trade under partition, and no setting evades it.

Trade-offs

More shards against tail latency. Capacity and parallelism, against a query that waits for the slowest of S shards. Shard for the size you’re going to be, not for the sake of it.

Replication factor against cost. Each additional copy multiplies storage and memory exactly. Factor 2 is the usual minimum for surviving a node; factor 3 buys surviving one while another is being replaced.

Write consistency against write availability. Requiring more confirmations means acknowledged writes survive failures, and writes fail when replicas are missing.

Read consistency against read latency. Consulting more replicas costs latency and removes staleness. Most search workloads should accept staleness.

Distributing at all against not. A single node with enough RAM is dramatically simpler to operate. Quantization (Part 5) and mmap (Part 10) both push the point where you’re forced to distribute a long way out.

Common mistakes

Sharding for speed on a collection that fits in one node. You’ve added fan-out and tail latency to something that was fine. Shard when you must.

Confusing shards with replicas. Shards split; replicas copy. Ten shards is not redundancy — lose a node and you’ve lost a tenth of your data.

replication_factor=2 with write_consistency_factor=1, expecting no data loss. A write acknowledged by one replica that then dies is gone. If you want durability against node loss, you have to require it.

Forgetting replication multiplies memory. Factor 2 doubles the hardware. It’s exact arithmetic and it belongs in the capacity plan.

Reading with default consistency and expecting read-your-writes. A lagging replica answers with what it has. W + r > R is the condition, and the defaults don’t satisfy it.

Treating shard count as a setting. Changing it means moving data. Pick it for your target scale.

Ignoring that every query touches every shard. A slow node doesn’t slow a fraction of your queries. It slows all of them.

Interview questions

1. What’s the difference between a shard and a replica?

Answer A shard is a partition — each holds a distinct subset of points, chosen by hashing the point id. Shards give capacity and parallelism. A replica is a copy of a shard on another node. Replicas give fault tolerance and read throughput. They solve different problems and you can want either alone. Ten shards with no replication is not redundant: lose a node and a tenth of your data is unavailable. **Follow-up:** what does each cost? Sharding costs fan-out, so query latency becomes the slowest shard’s. Replication costs a multiple of your storage and memory, exactly.

2. Why does query latency get worse as you add shards?

Answer Because a query scatters to every shard and cannot return until all have replied. The latency is the maximum over shards, not the average. Each shard does less work, so the mean falls — but you’ve taken more samples from the latency distribution, so the chance that at least one is slow rises. That makes the p99 of a fan-out query worse as shard count grows, even while per-shard work shrinks. **Follow-up:** what does that imply operationally? One degraded node affects every query, not a fraction of them. Fan-out spreads a local problem globally.

3. replication_factor=2, write_consistency_factor=1. A node dies right after a write is acknowledged. What happens?

Answer The write may be lost. It was acknowledged once a single replica had it; if that replica was the one that died before the second copy caught up, nothing else ever saw it. That’s the documented meaning of the default, not a failure. To make an acknowledged write survive one node failing, you need `write_consistency_factor=2` — and then writes fail whenever fewer than two replicas are reachable. **Follow-up:** which would you choose for a search index? Usually the fast one, because the index is a derived view rebuildable from a source of truth. For a system of record, the other.

4. What is read consistency for?

Answer It decides how many replicas must be consulted and agree before a read returns. `default` asks one, which is fastest and can return data from a replica that is behind. `majority`, `quorum`, `all` or an explicit number consult more, costing latency and removing staleness. The condition worth remembering is `W + r > R`: with write consistency 1, read consistency 1 and replication factor 2, that’s 2 > 2, which is false — so reads are not guaranteed to see acknowledged writes. **Follow-up:** when does staleness actually matter? “Did my document get indexed” flows, and anything where a user writes and immediately reads back.

5. How would you size a cluster for 100 million vectors at 768 dimensions?

Answer Start with the arithmetic: 100,000,000 × 768 × 4 bytes is about 286 GiB of raw vectors, before graph, payloads or replication. Multiply by `replication_factor`, so factor 2 is around 572 GiB. Then reduce it before adding hardware: scalar quantization takes the vector term to a quarter, and keeping originals on disk for rescoring takes them off the hot path entirely. Shard count follows from what’s left divided by per-node memory, and I’d want it as low as that allows, because of fan-out. **Follow-up:** what would you measure before committing? Recall under quantization on your own vectors (Part 4), because the whole plan depends on the compressed configuration being good enough.

6. A node was unreachable for two minutes and came back. What’s the risk?

Answer It missed every write during that window and, until it catches up, can answer reads with stale data if read consistency is `default`. The cluster looks healthy and a fraction of reads are quietly wrong. If `write_consistency_factor` was 1, there is a second risk: writes acknowledged by only the other replica during the window are the only copies, so a failure there before catch-up loses them. **Follow-up:** how would you avoid serving from it? Raise read consistency so a single lagging replica cannot answer alone, at the cost of latency.

7. Can you change the shard count later?

Answer Not as a simple setting. Shard count is chosen at creation and changing it means resharding — moving a large fraction of points between nodes, because the shard assignment is a hash of the id. So pick for your target scale. And if you get it wrong badly enough, the alias procedure from [Part 11](/qdrant-snapshots-backup-aliases/) is available: build a new collection with the right shard count alongside and swap. **Follow-up:** what’s the cost of over-sharding early? Fan-out latency you pay on every query from day one, for capacity you don’t need yet.

8. Should a vector search index be strongly consistent?

Answer Usually not. A search index is typically a derived view of data that lives in a system of record, and it can be rebuilt. Trading a small staleness window for availability and latency is the right call, and it’s why the defaults are what they are. The exceptions are real, though: if the vectors *are* the system of record, or a user writes and immediately reads back, or deletions must take effect immediately for compliance reasons — then you pay for consistency. **Follow-up:** what would you do about deletions for compliance specifically? Not rely on eventual propagation. Verify the delete took effect with a consistent read, and remember ([Part 10](/qdrant-storage-memory-mmap/)) that a deleted point is flagged before its space is reclaimed.

Sources

  • Qdrant documentation, https://qdrant.tech/documentation/guides/distributed_deployment/ — shards, replication factor, write consistency factor, read consistency and resharding.
  • The per-collection defaults shard_number: 1, replication_factor: 1 and write_consistency_factor: 1 were read from a fresh collection on Qdrant 1.19.1.
  • The W + r > R condition is the standard quorum-overlap requirement from replicated storage systems; it is stated here in Qdrant’s terms rather than quoted.
  • The memory arithmetic follows Part 5.

What to remember

Shards split data for capacity and parallelism, and cost you fan-out: every query waits for the slowest shard. Replicas copy data for fault tolerance and read throughput, and cost you an exact multiple of your storage.

The consistency factors are where the real decision lives. write_consistency_factor decides whether an acknowledged write survives a node failure, and read consistency decides whether a read can be served by a replica that is behind. The defaults — one and one — are fast and neither of those guarantees.

For most vector search that’s the right trade, because the index is a derived view you can rebuild. Make sure that’s true of yours before you rely on it.

Shards are for size, replicas are for survival, and the consistency factors decide what “written” means. Nothing about that is specific to vectors, and everything about it will surprise you at 3am if you never chose it deliberately.

How useful was this post?

Click on a heart to rate it!

Average rating 0 / 5. Vote count: 0

No votes so far! Be the first to rate this post.