Sharding

The router hashes _id and sends the document to the process that owns it. Use it when one machine cannot hold the data.

Leave [sharding] commented and every document stays in this process. Uncomment it when one machine cannot hold the collection or the writes. The router does not store the documents. A shard is a normal Kova process. The router speaks the MongoDB protocol to the address in shards[].client.

[sharding]
enabled = true
role = "router"
shards = [
  { id = 1, client = "127.0.0.1:27101" },
  { id = 2, client = "127.0.0.1:27102" },
]

Each shard has its own config, its own port, and its own data directory. role = "shard" on those processes is a normal server. The example ports stay off 27018, so the single server you already run is left alone.

Where a command goes

The router hashes _id onto a ring of 128 virtual nodes per shard.

• An insert with no _id receives one on the router, then goes to one shard.
• A find, count, update, or delete whose filter is one _id goes to that shard.
• The same commands without a single _id visit every shard. The router merges the batches or sums n.
• A single-document update or delete that is not one _id stops at the first shard that changes a row.
• create, drop, createIndexes, and dropDatabase visit every shard, so the collection and its indexes exist wherever a document lands.
• listCollections and listDatabases merge the names.
• SQL on the router loads each named collection from every shard and runs the statement there.
• An aggregate stage other than match, project, or unwind is refused on the router. A group inside each shard is not the group of the whole collection.
• mapReduce visits every shard and the batches are concatenated.
• A transaction is refused on the router.
• An update of admin.counters is sent to every shard. See Autoincrement for why each shard has its own last.

Adding a shard

The shard list is fixed for data already written. Adding a shard changes the ring and leaves existing documents where they were stored. A find by an _id that now hashes to the new shard misses that document. A find with no _id still walks every shard and finds it. Plan the shard list before the collection is large, or move documents yourself.

Inside each shard

Each shard is a full Kova process. It still has four local workers, and a collection name still hashes to one of them. All documents of one collection on that process live on one worker. Sharding is what splits the same collection across processes, by document _id.

A shard may enable [replication] on its own. Voters of that set are copies of that piece. They are not other shards. Losing a shard that has no copy silences that piece. The other shards still answer. Restarting the process on the same data directory brings the piece back. If the disk is gone and there is no copy, that piece is gone.

A dump pointed at the router reads every shard, and fails with no partial file if one shard is down. A dump pointed at one shard reads that piece. See Backup and restore.