Can someone give me a high level overview of how sharding is implemented?

[Ricardo Amendoeira]

Hello,

I want to implement a new type of sharding (by 2dsphere coordinates) to MongoDB as part of my thesis.

I'm having a hard time figuring out all the components I need to update.

This is my understanding so far:

- mongos is stateless and only acts as a bridge to config server and shards (maybe it needs a shard  map for inserts/queries?)

- shards are somewhat unaware that they just have part of a collection. they don't need to know about sharding.

- config server is what handles balancing the chunks between the shards and manages all the required information for that

Am I wrong on any of these?

Furthermore, I'm still a bit lost on the entire protocol for sharding a collection. This is my understanding:

0. The necessary indexes for the collection must exist already

1. Mongos tells config server to enable sharding on the collection (I don't know why this is separated from shardCollection but there must be a reason)

2. Mongos tells config server to shard a collection with a certain key. I already found s/commands/cluster_shard_collection_cmd.cpp but I'm not sure where this is answered on the config server side, since I'm still having issues debugging mongos and haven't started analysing mongod.

3. The config server starts the balancing round, if needed, and somehow (I have no idea) the chunks are now split between the shards. Who is managing inserts/queries? Is it mongos or the config server?

And that's my current very rough understanding.

If someone could give a general overview of this process and maybe what files/classes are involved in making this work I would be extremely grateful since most of the relevant files I have found use lots of mongodb specific abstractions that make it very time consuming to actually figure out what's going on.

I know a debugger would be extremely helpful but I'm having issues with that (I already created another thread for that).

Thank you so much if you help me :)

Ricardo Amendoeira

[Andy Schwerin]

On Fri, Jan 20, 2017 at 9:03 AM Ricardo Amendoeira <ricardofili...@gmail.com> wrote:

Hello,

I want to implement a new type of sharding (by 2dsphere coordinates) to MongoDB as part of my thesis.

I'm having a hard time figuring out all the components I need to update.

This is my understanding so far:

- mongos is stateless and only acts as a bridge to config server and shards (maybe it needs a shard  map for inserts/queries?)

Correct. Mongos acts primarily as an operation router, though it also does some query execution work, mostly around merging results from shards.

 

- shards are somewhat unaware that they just have part of a collection. they don't need to know about sharding.

This isn't entirely true. As of the 3.4 release and the current development head (master branch), all shard nodes are aware that they are members of a sharded cluster. More importantly, the protocol for routing operations requires each shard to know what range of shard keys it owns. This serves two purposes. The most important is that the routing protocol uses optimistic concurrency control similar to an mvcc type system -- shards have to know what key ranges they own so that when they encounter a router (mongos) with a stale routing table, they can detect it and inform the router that it needs an update.

Additionally, shards need to know what key ranges they own because they lazily delete the documents corresponding to key ranges that they have donated to another shard. As such, queries need to have an additional filter attached to exclude documents in unowned but present key ranges (orphan documents).

 

- config server is what handles balancing the chunks between the shards and manages all the required information for that

That is correct, beginning with MongoDB 3.4 and its current development branch (master).

 

Am I wrong on any of these?

Furthermore, I'm still a bit lost on the entire protocol for sharding a collection. This is my understanding:

0. The necessary indexes for the collection must exist already

1. Mongos tells config server to enable sharding on the collection (I don't know why this is separated from shardCollection but there must be a reason)

You must enable sharding on the *database* containing the collection. This is bookkeeping, and could maybe have been implicit when you shard your first collection in the database, but that's not what we chose to implement.

 

2. Mongos tells config server to shard a collection with a certain key. I already found s/commands/cluster_shard_collection_cmd.cpp but I'm not sure where this is answered on the config server side, since I'm still having issues debugging mongos and haven't started analysing mongod.

 

3. The config server starts the balancing round, if needed, and somehow (I have no idea) the chunks are now split between the shards. Who is managing inserts/queries? Is it mongos or the config server?

Migrations are initiated by the config servers, but the work is all done by the shards involved. In the master branch or 3.4 branch, you want to look at src/mongo/db/s/migration_*_manager.{h,cpp}. The chunk splitting process (taking one key range, and dividing it in two) continues to be driven by the mongos nodes, though it is our plan to eventually move it to the shards owning the key ranges.

 

And that's my current very rough understanding.

If someone could give a general overview of this process and maybe what files/classes are involved in making this work I would be extremely grateful since most of the relevant files I have found use lots of mongodb specific abstractions that make it very time consuming to actually figure out what's going on.

If your goal is to partition by 2D-sphere coordinates, you'll have to do relatively little work with any of this code, though.  The problem won't be managing where chunks live, at least initially. Instead, you'll need to change how the router decides what shards are relevant to a query. I believe this already uses the query execution subsystem, which has logic to extract 2dsphere key ranges from geojson shapes described in queries. Getting that right, and figuring out how to disable the check that prohibits 2dsphere indexes as shard indexes seems like it might be a substantial fraction of the work.

 

I know a debugger would be extremely helpful but I'm having issues with that (I already created another thread for that).

Thank you so much if you help me :)

Ricardo Amendoeira

--
You received this message because you are subscribed to the Google Groups "mongodb-dev" group.
To unsubscribe from this group and stop receiving emails from it, send an email to mongodb-dev...@googlegroups.com.
To post to this group, send email to mongo...@googlegroups.com.
Visit this group at https://groups.google.com/group/mongodb-dev.
To view this discussion on the web visit https://groups.google.com/d/msgid/mongodb-dev/f43cf7f8-f901-455a-90fd-a1a9301b5ea1%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Ricardo Amendoeira]

First of all, thank you for taking the time to help me :)
If you don't mind, I have some followup questions:

On Friday, January 20, 2017 at 3:22:10 PM UTC, Andy Schwerin wrote:

Hello,

I want to implement a new type of sharding (by 2dsphere coordinates) to MongoDB as part of my thesis.

I'm having a hard time figuring out all the components I need to update.

This is my understanding so far:

- mongos is stateless and only acts as a bridge to config server and shards (maybe it needs a shard  map for inserts/queries?)

Correct. Mongos acts primarily as an operation router, though it also does some query execution work, mostly around merging results from shards.

So does Mongos also need to be updated to understand new types of sharding keys? I'm guessing yes from your next answer, since mongos has a routing table.

 

- shards are somewhat unaware that they just have part of a collection. they don't need to know about sharding.

This isn't entirely true. As of the 3.4 release and the current development head (master branch), all shard nodes are aware that they are members of a sharded cluster. More importantly, the protocol for routing operations requires each shard to know what range of shard keys it owns. This serves two purposes. The most important is that the routing protocol uses optimistic concurrency control similar to an mvcc type system -- shards have to know what key ranges they own so that when they encounter a router (mongos) with a stale routing table, they can detect it and inform the router that it needs an update.

Additionally, shards need to know what key ranges they own because they lazily delete the documents corresponding to key ranges that they have donated to another shard. As such, queries need to have an additional filter attached to exclude documents in unowned but present key ranges (orphan documents). 

This makes me curious, the donation process is entirely controlled by the config server, correct? If so, what's the reason that makes this more complicated then the config server telling each shard what ranges to mark as lazy delete? 

 

- config server is what handles balancing the chunks between the shards and manages all the required information for that

That is correct, beginning with MongoDB 3.4 and its current development branch (master).

 

Am I wrong on any of these?

Furthermore, I'm still a bit lost on the entire protocol for sharding a collection. This is my understanding:

0. The necessary indexes for the collection must exist already

1. Mongos tells config server to enable sharding on the collection (I don't know why this is separated from shardCollection but there must be a reason)

You must enable sharding on the *database* containing the collection. This is bookkeeping, and could maybe have been implicit when you shard your first collection in the database, but that's not what we chose to implement.

 

 

2. Mongos tells config server to shard a collection with a certain key. I already found s/commands/cluster_shard_collection_cmd.cpp but I'm not sure where this is answered on the config server side, since I'm still having issues debugging mongos and haven't started analysing mongod.

 

3. The config server starts the balancing round, if needed, and somehow (I have no idea) the chunks are now split between the shards. Who is managing inserts/queries? Is it mongos or the config server?

Migrations are initiated by the config servers, but the work is all done by the shards involved. In the master branch or 3.4 branch, you want to look at src/mongo/db/s/migration_*_manager.{h,cpp}. The chunk splitting process (taking one key range, and dividing it in two) continues to be driven by the mongos nodes, though it is our plan to eventually move it to the shards owning the key ranges.

Thank you for pointing me to the right place, you probably saved me quite a lot of time! Is there a resource with mongodb's file naming and organization conventions?

 

 

And that's my current very rough understanding.

If someone could give a general overview of this process and maybe what files/classes are involved in making this work I would be extremely grateful since most of the relevant files I have found use lots of mongodb specific abstractions that make it very time consuming to actually figure out what's going on.

If your goal is to partition by 2D-sphere coordinates, you'll have to do relatively little work with any of this code, though.  The problem won't be managing where chunks live, at least initially. Instead, you'll need to change how the router decides what shards are relevant to a query. I believe this already uses the query execution subsystem, which has logic to extract 2dsphere key ranges from geojson shapes described in queries. Getting that right, and figuring out how to disable the check that prohibits 2dsphere indexes as shard indexes seems like it might be a substantial fraction of the work.

 

I was already able to change the code that checks if a sharding key is valid to also accept 2dsphere indexes, is does the router decide what shards are relevant to a query based on the aforementioned  migration_*_manager files or is that only for the balancer?

Again, thanks for taking the time,

Ricardo Amendoeira

[Andy Schwerin]

On Sun, Jan 22, 2017 at 5:38 PM Ricardo Amendoeira <ricardofili...@gmail.com> wrote:

First of all, thank you for taking the time to help me :)
If you don't mind, I have some followup questions:

On Friday, January 20, 2017 at 3:22:10 PM UTC, Andy Schwerin wrote:

Hello,

I want to implement a new type of sharding (by 2dsphere coordinates) to MongoDB as part of my thesis.

I'm having a hard time figuring out all the components I need to update.

This is my understanding so far:

- mongos is stateless and only acts as a bridge to config server and shards (maybe it needs a shard  map for inserts/queries?)

Correct. Mongos acts primarily as an operation router, though it also does some query execution work, mostly around merging results from shards.

So does Mongos also need to be updated to understand new types of sharding keys? I'm guessing yes from your next answer, since mongos has a routing table.

Yes. Probably you'll need to focus primarily on the logic that "targets" operations. If you don't worry about locality at first, you can ignore the balancer and chunk splitter.

 

 

- shards are somewhat unaware that they just have part of a collection. they don't need to know about sharding.

This isn't entirely true. As of the 3.4 release and the current development head (master branch), all shard nodes are aware that they are members of a sharded cluster. More importantly, the protocol for routing operations requires each shard to know what range of shard keys it owns. This serves two purposes. The most important is that the routing protocol uses optimistic concurrency control similar to an mvcc type system -- shards have to know what key ranges they own so that when they encounter a router (mongos) with a stale routing table, they can detect it and inform the router that it needs an update.

Additionally, shards need to know what key ranges they own because they lazily delete the documents corresponding to key ranges that they have donated to another shard. As such, queries need to have an additional filter attached to exclude documents in unowned but present key ranges (orphan documents). 

This makes me curious, the donation process is entirely controlled by the config server, correct? If so, what's the reason that makes this more complicated then the config server telling each shard what ranges to mark as lazy delete? 

No, this is not completely correct. The config server stores the data, and tells the shards to move the chunks, but the actual moving is done by the shards, and the shards tell the config server when and how to update the metadata.

In any event, the shards need to know when their ownership of chunk ranges changes, and what chunk ranges running queries expect them to own, so that they can decide when to perform the orphan data cleanup.

 

 

- config server is what handles balancing the chunks between the shards and manages all the required information for that

That is correct, beginning with MongoDB 3.4 and its current development branch (master).

 

Am I wrong on any of these?

Furthermore, I'm still a bit lost on the entire protocol for sharding a collection. This is my understanding:

0. The necessary indexes for the collection must exist already

1. Mongos tells config server to enable sharding on the collection (I don't know why this is separated from shardCollection but there must be a reason)

You must enable sharding on the *database* containing the collection. This is bookkeeping, and could maybe have been implicit when you shard your first collection in the database, but that's not what we chose to implement.

 

 

2. Mongos tells config server to shard a collection with a certain key. I already found s/commands/cluster_shard_collection_cmd.cpp but I'm not sure where this is answered on the config server side, since I'm still having issues debugging mongos and haven't started analysing mongod.

 

3. The config server starts the balancing round, if needed, and somehow (I have no idea) the chunks are now split between the shards. Who is managing inserts/queries? Is it mongos or the config server?

Migrations are initiated by the config servers, but the work is all done by the shards involved. In the master branch or 3.4 branch, you want to look at src/mongo/db/s/migration_*_manager.{h,cpp}. The chunk splitting process (taking one key range, and dividing it in two) continues to be driven by the mongos nodes, though it is our plan to eventually move it to the shards owning the key ranges.

Thank you for pointing me to the right place, you probably saved me quite a lot of time! Is there a resource with mongodb's file naming and organization conventions?

No, not really. Newer files tend to be named for the type they implement, and header files are implemented by correspondingly named cpp files. A lot of older code still exists that violates the first convention, and there's not a strong convention for naming files that hold free functions.

 

 

 

And that's my current very rough understanding.

If someone could give a general overview of this process and maybe what files/classes are involved in making this work I would be extremely grateful since most of the relevant files I have found use lots of mongodb specific abstractions that make it very time consuming to actually figure out what's going on.

If your goal is to partition by 2D-sphere coordinates, you'll have to do relatively little work with any of this code, though.  The problem won't be managing where chunks live, at least initially. Instead, you'll need to change how the router decides what shards are relevant to a query. I believe this already uses the query execution subsystem, which has logic to extract 2dsphere key ranges from geojson shapes described in queries. Getting that right, and figuring out how to disable the check that prohibits 2dsphere indexes as shard indexes seems like it might be a substantial fraction of the work.

 

I was already able to change the code that checks if a sharding key is valid to also accept 2dsphere indexes, is does the router decide what shards are relevant to a query based on the aforementioned  migration_*_manager files or is that only for the balancer?

Again, thanks for taking the time,

Ricardo Amendoeira

--

You received this message because you are subscribed to the Google Groups "mongodb-dev" group.
To unsubscribe from this group and stop receiving emails from it, send an email to mongodb-dev...@googlegroups.com.
To post to this group, send email to mongo...@googlegroups.com.
Visit this group at https://groups.google.com/group/mongodb-dev.

To view this discussion on the web visit https://groups.google.com/d/msgid/mongodb-dev/563db178-4b0e-4901-9d1d-36f185688a7b%40googlegroups.com.