How Queries Route Across Shards
In a sharded MongoDB cluster, no client ever talks to a shard directly — every read and write passes through mongos, a lightweight, stateless query router that decides which shard or shards actually hold the data you’re asking for. Whether your query gets routed to exactly one shard or broadcast to every shard in the cluster is one of the biggest performance levers in a sharded deployment, and it depends almost entirely on whether your query filter includes the collection’s shard key. This lesson explains exactly how mongos makes that routing decision, what the config servers store to make it possible, and how to use explain() to prove whether your queries are being targeted or scattered.
Overview: How Query Routing Works
A sharded collection is split into chunks — contiguous ranges of shard key values — and each chunk lives on exactly one shard at a time. The cluster’s three moving pieces are: the shards themselves (each a replica set holding a slice of the data), a small set of config servers (also a replica set) that store the cluster’s metadata — which chunk ranges live on which shard — and one or more mongos routers that applications actually connect to.
mongos does not store data and does not run the query planner against real documents. Instead, it keeps a cached copy of the routing table (the chunk-to-shard map) pulled from the config servers, and uses it to figure out, for a given query filter, which shard(s) could possibly hold matching documents. If the filter pins down the shard key to an exact value (or, for a compound shard key, an exact value for a usable prefix of the fields), mongos can compute the specific chunk range the value falls into and send the query to only the one or two shards that own it. This is called a targeted (or single-shard) query.
If the filter says nothing about the shard key at all, mongos has no information to narrow the search — the matching documents could be on any shard — so it broadcasts the query to every shard in the cluster. This is a scatter-gather query. Each shard’s mongod runs the query locally against its own data (using its own indexes, exactly like an unsharded query would), and mongos merges the individual cursors back into one result stream for the client. If the query also has a sort(), mongos has to perform a merge-sort across all the per-shard streams rather than trusting a single sorted stream, and a limit() can’t be safely applied on each shard alone until enough sorted results have come back from all of them.
The shard key’s type matters here too: ranged (the default) and hashed shard keys behave very differently for routing. With a ranged shard key, chunks are contiguous ranges of the actual field value, so both equality queries and range queries ($gt/$lt) on that field can be targeted to a contiguous set of shards. With a hashed shard key, MongoDB stores the hash of the field’s value, which scatters even sequential input values across the hash space — great for write distribution, but it means only exact-equality queries on the hashed field can be targeted; a range query on a hashed shard key always scatter-gathers, because a range of original values maps to unpredictable, non-contiguous hash chunks.
For a compound shard key like { customerId: 1, orderDate: 1 }, mongos can target using a query only if it supplies at least an equality condition on the leading field, customerId. A query that filters only on orderDate (the second field) gives mongos no way to narrow the chunk range, even though orderDate is technically part of the shard key — it still scatter-gathers. This is the same left-to-right prefix rule you already know from compound indexes, applied at the cluster-routing level instead of the single-node index level.
Aggregation pipelines route the same way: if the first $match stage filters on the shard key, mongos pushes the pipeline down to only the relevant shards. Stages that need a global view — $group, $sort, an unbounded $lookup — run on each shard first as much as possible, then a merging step runs on mongos (or a chosen shard) to combine the partial results, which is why putting $match as early as possible in a sharded pipeline matters even more than in an unsharded one.
Syntax
Query routing itself isn’t something you call directly — it happens automatically based on your query filter and the collection’s shard key. What you do call are the tools used to inspect and verify routing decisions:
| Command | Purpose |
|---|---|
db.collection.find(filter).explain() |
Shows the winning plan, including a shards array when run against a sharded collection — one entry per shard the query actually touched. |
sh.status() |
Prints an overview of the cluster: shards, sharded databases/collections, and chunk counts per shard. |
db.collection.getShardDistribution() |
Prints per-shard document counts, data size, and chunk counts for one sharded collection. |
sh.shardCollection(namespace, shardKey) |
Shards a collection on the given key document; required once before a collection’s data is distributed across shards. |
Examples
Example 1: Sharding a collection with a compound shard key
use ecommerce
sh.shardCollection("ecommerce.orders", { customerId: 1, orderDate: 1 });
Output:
{
collectionsharded: 'ecommerce.orders',
ok: 1
}
This shards the orders collection on the compound key { customerId: 1, orderDate: 1 }. Because customerId is high-cardinality and matches the app’s most common access pattern (“get this customer’s orders”), most real queries can supply an equality match on it and get targeted.
Example 2: A targeted query
db.orders.find({
customerId: 48213,
orderDate: { $gte: ISODate("2026-01-01") }
}).explain();
Output:
{
queryPlanner: {
winningPlan: {
stage: 'SINGLE_SHARD',
shards: [
{
shardName: 'shard0002',
connectionString: 'shard0002/mongo-s2a:27018,mongo-s2b:27018',
winningPlan: {
stage: 'FETCH',
inputStage: {
stage: 'IXSCAN',
keyPattern: { customerId: 1, orderDate: 1 }
}
}
}
]
}
}
}
The shards array contains exactly one entry, and the top-level stage is SINGLE_SHARD. Because the query supplies an equality value for customerId (the leading shard key field), mongos computed exactly which chunk — and therefore which shard — owns that customer’s data, and sent the query only there. The shard itself then used a normal IXSCAN against its local copy of the shard key index.
Example 3: A scatter-gather query
db.orders.find({ status: "shipped" }).explain();
Output:
{
queryPlanner: {
winningPlan: {
stage: 'SHARD_MERGE',
shards: [
{ shardName: 'shard0000', winningPlan: { stage: 'COLLSCAN' } },
{ shardName: 'shard0001', winningPlan: { stage: 'COLLSCAN' } },
{ shardName: 'shard0002', winningPlan: { stage: 'COLLSCAN' } }
]
}
}
}
This query filters on status, which is not the shard key, so mongos has no way to know which shard(s) hold matching documents. The top-level stage becomes SHARD_MERGE, and the shards array lists all three shards — every one of them ran a full COLLSCAN locally, and mongos merged the results. On a large collection this is both a full scan and a fan-out, which is why non-shard-key access patterns need their own plan, usually a supporting index on every shard rather than reliance on routing alone.
Example 4: Checking chunk and data distribution
db.orders.getShardDistribution();
Output:
Shard shard0000 at shard0000/mongo-s0a:27018,mongo-s0b:27018
data: 341MiB docs: 128460 chunks: 4
estimated data per chunk: 85MiB
estimated docs per chunk: 32115
Shard shard0001 at shard0001/mongo-s1a:27018,mongo-s1b:27018
data: 356MiB docs: 133920 chunks: 4
estimated data per chunk: 89MiB
estimated docs per chunk: 33480
Shard shard0002 at shard0002/mongo-s2a:27018,mongo-s2b:27018
data: 349MiB docs: 131100 chunks: 4
estimated data per chunk: 87MiB
estimated docs per chunk: 32775
Totals
data: 1046MiB docs: 393480 chunks: 12
Shard shard0000 contains 32.6% data, 32.65% docs in cluster
Shard shard0001 contains 34.03% data, 34.03% docs in cluster
Shard shard0002 contains 33.36% data, 33.31% docs in cluster
Roughly even percentages across shards, as shown here, indicate the shard key and balancer are doing their job. If one shard’s percentage were dramatically higher than the others, that would point to an uneven or monotonically-increasing shard key creating a hot shard.
How It Works Step by Step
mongosreceives the query from the driver and parses the filter document.- It consults its cached routing table (refreshed from the config server replica set) for that collection’s shard key and current chunk map.
- It compares the filter to the shard key. An equality on the full key (or a usable left-to-right prefix for a compound key) resolves to one or a few specific chunks and therefore one or a few shards — a targeted query. Anything less specific resolves to “could be anywhere,” and every shard is targeted.
mongossends the query to each targeted shard’s primary (or a secondary, depending on read preference).- Each shard’s
mongodruns its own query planner completely independently — it decidesIXSCANvsCOLLSCANusing only its local indexes, with no awareness that it’s part of a bigger cluster. mongoscollects the cursor(s) from every targeted shard. For a single targeted shard it mostly just passes results through. For multiple shards it merges the streams — performing an actual merge-sort if asort()was requested, and only applyinglimit()/skip()after that merge is safely resolved.- If a shard reports that its local chunk metadata version doesn’t match what
mongosexpected (aStaleConfigerror, typically right after the balancer moved a chunk),mongosrefreshes its routing table from the config servers and silently retries — this is normal, transient behavior, not a bug.
Common Mistakes
Mistake 1: Querying by a field that isn’t the shard key and expecting it to be fast at scale.
// status is not part of the shard key { customerId: 1, orderDate: 1 }
db.orders.find({ status: "shipped" });
As shown in Example 3, this scatter-gathers to every shard, and if there’s no supporting index on status, every one of those shards runs a full COLLSCAN. The fix isn’t to change the shard key — that would hurt the more common customer-lookup pattern — it’s to add a normal secondary index on every shard. That still scatter-gathers across shards (routing can’t avoid that without shard-key information), but each shard now does an efficient IXSCAN instead of a full scan:
db.orders.createIndex({ status: 1, orderDate: -1 });
Mistake 2: Comparing a shard key of type ObjectId against a plain string.
// customerId came from a URL param, so it's a string, not an ObjectId
const customerIdParam = "648f1b2e9a1c4d3f5b6a7c8d";
db.orders.find({ customerId: customerIdParam });
If the shard key’s actual BSON type is ObjectId, a string never matches an ObjectId value — MongoDB compares by type first. Beyond just returning zero documents, this can quietly defeat routing too: mongos may still compute a target based on the string’s position in the ranged or hashed space, sending the query to a shard that doesn’t contain the matching value at all, and you get an empty result with no error. Always convert first:
const customerIdParam = "648f1b2e9a1c4d3f5b6a7c8d";
db.orders.find({ customerId: new ObjectId(customerIdParam) });
Mistake 3: Choosing a hashed shard key and then relying on range queries to be fast.
// orderDate is hashed for even write distribution
db.orders.find({
orderDate: { $gte: ISODate("2026-01-01"), $lt: ISODate("2026-02-01") }
});
Hashing destroys the original value’s ordering, so a range on a hashed field can never be targeted — this always scatter-gathers, no matter how narrow the range looks. If range queries on a field are common, don’t hash it; use a ranged shard key instead, possibly compound with a high-cardinality equality field first to still avoid hot chunks.
Best Practices
- Choose a shard key that matches your application’s dominant query pattern’s equality filter — routing is only fast when queries can supply it.
- Run
explain()against sharded collections and check the size of theshardsarray and whether the top-level stage isSINGLE_SHARDversusSHARD_MERGE. - For fields you must query on that aren’t the shard key, create supporting indexes on every shard — it won’t make routing targeted, but it turns a per-shard
COLLSCANinto a per-shardIXSCAN. - Use a ranged shard key (not hashed) when range queries on that field matter; use hashed only when you need even write distribution and will only ever query it by equality.
- For compound shard keys, put the field your queries can always supply an equality value for first, so a query prefix keeps routing targeted.
- Convert client-supplied identifiers to the shard key’s actual BSON type (e.g.
new ObjectId(idString)) before querying — a type mismatch can silently return nothing and mis-target the query. - Treat occasional
StaleConfigretries in logs as normal after balancer chunk moves; investigate only if they’re constant, which usually points to balancer thrashing from a poor shard key. - Periodically check
getShardDistribution()orsh.status()to catch an emerging hot shard before it becomes a capacity problem.
Practice Exercises
- Given
orderssharded on{ customerId: 1, orderDate: 1 }, write a query you expectmongosto target to a single shard, and a second query on the same collection you expect to scatter-gather. Runexplain()on both and compare the length of theshardsarray. - Write a query that filters only on
orderDate(the second shard key field, nocustomerId). Predict whether it will be targeted, then verify withexplain()and explain in your own words why the prefix rule applies here. - Run
db.orders.getShardDistribution()against a sharded collection and identify whether any shard is holding a disproportionate share of the data — what shard key property would cause that?
Summary
mongosis a stateless router that decides which shard(s) a query goes to, using a cached routing table sourced from the config servers.- A query that supplies an equality value for the shard key (or a left-to-right prefix of a compound shard key) is targeted to one or a few shards; anything else is scatter-gathered to all shards.
- Ranged shard keys support both equality and range targeting; hashed shard keys only support equality targeting, since hashing destroys the original ordering.
- Each shard runs its own independent query planner (
IXSCANvsCOLLSCAN) against only its local data — routing and local plan selection are separate decisions. explain()‘sshardsarray is the ground truth for whether a query was targeted — always check it rather than assuming.- Comparing a shard key to the wrong BSON type (like a string vs an
ObjectId) can silently break both matching and routing.
