NAÏVE, CLIENT-SIDE SHARDING WITH ONLINE ADDITION OF SHARDS

• US 20150006482A1
• Filed: 03/26/2014
• Published: 01/01/2015
• Est. Priority Date: 06/28/2013
• Status: Active Grant

• Current Assignee: Oracle International Corporation (Oracle Corporation)
• Original Assignee: Oracle International Corporation (Oracle Corporation)
• Inventors: Hardy, Alexandre, Tilak, Omkar
• Granted Patent: US 9,619,545 B2
• Time in Patent Office: Days
• US Class Current: 707/613
• CPC Class Codes: G06F 16/2365 Ensuring data consistency and integrity
• CPC Class Codes: G06F 16/273 Asynchronous replication or reconciliation
• CPC Class Codes: G06F 16/278 Data partitioning, e.g. horizontal or vertical partitioning

Abstract

Multiple clients can be enabled to perform operations relative to data items in a shard system asynchronously to each other without the use by those clients of exclusive locks. A rebalancing event, in which data items are redistributed automatically among a set of shards due to a modification of the quantity of shards in the system, can be performed without the use of exclusive locks by clients. Clients can continue to perform operations relative to at least some of the data items in the shard system even while rebalancing processes are redistributing at least some of the data items asynchronously during a system-wide rebalancing event. All of these benefits can be obtained without sacrificing data consistency within the shard system.

Claims

1. A computer-implemented method comprising:

determining that a quantity of shards in a multi-shard system has changed;

in response to determining that the quantity of shards has changed, transitioning a client from a normal state, in which the client performs a particular type of operation relative to data items stored in the system in a first manner, to a rebalancing state, in which the client performs the particular type of operation relative to data items stored in the system in a second manner that differs from the first manner and without the client acquiring exclusive locks relative to any data items;

while the client is in the rebalancing state, determining, for one or more particular data items, destination shards that are separate from source shards on which the one or more particular data items were stored prior to the client'"'"'s transition to the rebalancing state; and

moving the one or more data items from the source shards to the destination shards while the client is in the rebalancing state.

2. The computer-implemented method of claim 1, further comprising:

while the one or more data items are being moved from the source shards to the destination shards, the client performing, asynchronously to the moving of the one or more data items, and relative to a data item, an operation of the particular type in the second manner without acquiring an exclusive lock.

3. The computer-implemented method of claim 1, further comprising:

determining the destination shards for the one or more particular data items by hashing primary keys of the data items based at least in part on the quantity of shards in the multi-shard system following the shard quantity change; and

in response to determining that the one or more data items have been moved from the source shards to the destination shards, transitioning the client from the rebalancing state to the normal state.

4. The computer-implemented method of claim 1, further comprising:

while the one or more data items are being moved from the source shards to the destination shards, the client performing an add type of operation in the second manner relative to a first data item having a primary key at least in part by;

determining, based at least in part on a quantity of shards in the multi-shard system preceding the shard quantity change, a particular source shard on which the first data item would have been added prior to the shard quantity change;

determining, based at least in part on a quantity of shards in the multi-shard system following the shard quantity change, a particular destination shard on which the first data item is to be added after the shard quantity change;

determining that no data item on the source shard has the first data item'"'"'s primary key;

determining that no data item on the destination shard has both a true tombstone attribute value and the first data item'"'"'s primary key; and

inserting the first data item into the destination shard.

5. The computer-implemented method of claim 1, further comprising:

while the one or more data items are being moved from the source shards to the destination shards, the client performing an add type of operation in the second manner relative to a first data item having a primary key at least in part by;

determining, based at least in part on a quantity of shards in the multi-shard system preceding the shard quantity change, a particular source shard on which the first data item would have been added prior to the shard quantity change;

determining, based at least in part on a quantity of shards in the multi-shard system following the shard quantity change, a particular destination shard on which the first data item is to be added after the shard quantity change;

determining that no data item on the source shard has the first data item'"'"'s primary key;

determining that a second data item on the destination shard has both a true tombstone attribute value and the first data item'"'"'s primary key;

assigning, to attributes of the second data item, attribute values of the first data item; and

setting the second data item'"'"'s tombstone attribute value to false.

6. The computer-implemented method of claim 1, further comprising:

while the one or more data items are being moved from the source shards to the destination shards, the client performing an update type of operation in the second manner relative to a first data item having a primary key at least in part by;

determining, based at least in part on a quantity of shards in the multi-shard system preceding the shard quantity change, a particular source shard on which the first data item would have been located prior to the shard quantity change;

determining, based at least in part on a quantity of shards in the multi-shard system following the shard quantity change, a particular destination shard on which the first data item is to be located after the shard quantity change;

determining that no data item on the source shard has the first data item'"'"'s primary key;

determining that a second data item on the destination shard has both a false tombstone attribute value and the first data item'"'"'s primary key; and

assigning, to attributes of the second data item, attribute values of the first data item.

7. The computer-implemented method of claim 1, further comprising:

while the one or more data items are being moved from the source shards to the destination shards, the client performing an update type of operation in the second manner relative to a first data item having a primary key at least in part by;

determining, based at least in part on a quantity of shards in the multi-shard system preceding the shard quantity change, a particular source shard on which the first data item would have been located prior to the shard quantity change;

determining, based at least in part on a quantity of shards in the multi-shard system following the shard quantity change, a particular destination shard on which the first data item is to be located after the shard quantity change;

determining that a second data item on the source shard has both the first data item'"'"'s primary key and a false tombstone attribute value;

determining that a third data item on the destination shard has the first data item'"'"'s primary key;

assigning, to attributes of the third data item, attribute values of the first data item; and

setting a version number of the third data item to a value different from a version number of the second data item.

8. The computer-implemented method of claim 1, further comprising:

while the one or more data items are being moved from the source shards to the destination shards, the client performing an update type of operation in the second manner relative to a first data item having a primary key at least in part by;

determining, based at least in part on a quantity of shards in the multi-shard system preceding the shard quantity change, a particular source shard on which the first data item would have been located prior to the shard quantity change;

determining, based at least in part on a quantity of shards in the multi-shard system following the shard quantity change, a particular destination shard on which the first data item is to be located after the shard quantity change;

determining that a second data item on the source shard has both the first data item'"'"'s primary key and a false tombstone attribute value;

determining no data item on the destination shard has the first data item'"'"'s primary key; and

inserting the first data item into the destination shard with a version number that differs from a version number of the second data item.

9. The computer-implemented method of claim 1, further comprising:

while the one or more data items are being moved from the source shards to the destination shards, the client performing a delete type of operation in the second manner relative to a first data item having a primary key at least in part by;

determining, based at least in part on a quantity of shards in the multi-shard system preceding the shard quantity change, a particular source shard on which the first data item would have been located prior to the shard quantity change;

determining, based at least in part on a quantity of shards in the multi-shard system following the shard quantity change, a particular destination shard on which the first data item is to be located after the shard quantity change;

determining that a second data item on the source shard has the first data item'"'"'s primary key;

upserting the first data item into the destination shard with a version number that differs from a version number of the second data item; and

setting a tombstone attribute value of the first data item on the destination shard to true.

10. The computer-implemented method of claim 1, further comprising:

while the one or more data items are being moved from the source shards to the destination shards, the client performing a delete type of operation in the second manner relative to a first data item having a primary key at least in part by;

determining, based at least in part on a quantity of shards in the multi-shard system preceding the shard quantity change, a particular source shard on which the first data item would have been located prior to the shard quantity change;

determining, based at least in part on a quantity of shards in the multi-shard system following the shard quantity change, a particular destination shard on which the first data item is to be located after the shard quantity change;

determining that no data item on the source shard has the first data item'"'"'s primary key;

determining that a second data item on the destination shard has the first data item'"'"'s primary key; and

setting a tombstone attribute value of the second data item to true.

11. The computer-implemented method of claim 1, further comprising:

while the one or more data items are being moved from the source shards to the destination shards, the client performing a get type of operation in the second manner relative to a primary key at least in part by;

determining, based at least in part on a quantity of shards in the multi-shard system preceding the shard quantity change, a particular source shard on which a data item having the primary key would have been located prior to the shard quantity change;

determining, based at least in part on a quantity of shards in the multi-shard system following the shard quantity change, a particular destination shard on which a data item having the primary key is to be located after the shard quantity change;

determining that a first data item on the destination shard has the primary key and a false tombstone attribute value; and

reading attribute values of the first data item.

12. The computer-implemented method of claim 1, further comprising:

while the one or more data items are being moved from the source shards to the destination shards, the client performing a get type of operation in the second manner relative to a primary key at least in part by;

determining, based at least in part on a quantity of shards in the multi-shard system preceding the shard quantity change, a particular source shard on which a data item having the primary key would have been located prior to the shard quantity change;

determining, based at least in part on a quantity of shards in the multi-shard system following the shard quantity change, a particular destination shard on which a data item having the primary key is to be located after the shard quantity change;

determining that no data item on the destination shard has the primary key;

determining that a first data item on the source shard has the primary key and a false tombstone attribute value; and

reading attribute values of the first data item.

13. The computer-implemented method of claim 1, further comprising:

while the one or more data items are being moved from the source shards to the destination shards, the client performing a get type of operation in the second manner relative to a primary key at least in part by;

determining, based at least in part on a quantity of shards in the multi-shard system preceding the shard quantity change, a particular source shard on which a data item having the primary key would have been located prior to the shard quantity change;

determining, based at least in part on a quantity of shards in the multi-shard system following the shard quantity change, a particular destination shard on which a data item having the primary key is to be located after the shard quantity change;

determining, at a first moment in time, that no data item on the destination shard has the primary key;

determining that no data item on the source shard has the primary key;

determining, at a second moment in time, that a first data item on the destination shard has the primary key and a false tombstone attribute value; and

reading attribute values of the first data item.

14. The computer-implemented method of claim 1, wherein moving the one or more data items from the source shards to the destination shards comprises:

incrementing a system-wide version number; and

for each first data item having a version number that is less than the system-wide version number, determining, based at least in part on a quantity of shards in the multi-shard system following the shard quantity change, a first destination shard on which the first data item is to be located after the shard quantity change;

for each second data item that (a) has a version number that is less than the system-wide version number and (b) is located on a source shard that differs from a second destination shard on which the second data item is to be located after the shard quantity change, inserting, into the second destination shard, a copy of the second data item having the system-wide version number;

for each third data item that (a) has a version number that is less than the system-wide version number and (b) is already located on a third destination shard on which the third data item is to be located after the shard quantity change, changing a version number of the third data item to the system-wide version number.

15. The computer-implemented method of claim 1, wherein moving the one or more data items from the source shards to the destination shards comprises incrementing a system-wide version number, and further comprising:

determining a set of queries that are pending against one or more shards in the multi-shard system as of a time that the movement of the one or more data items completes;

waiting for all queries in the set of queries to finish;

after all queries in the set of queries have finished, removing, from all shards in the multi-shard system, all data item copies having version number attribute values that differ from the system-wide version number; and

after all queries in the set of queries have finished, removing, from all shards in the multi-shard system, all data item copies having true tombstone attribute values.

16. The computer-implemented method of claim 1, further comprising:

while the one or more data items are being moved from the source shards to the destination shards, and for each particular shard in the multi-shard system, populating a preliminary result queue for that particular shard with data items that both (a) are located on that particular shard and (b) satisfy specified query criteria;

for each particular shard'"'"'s preliminary result queue, sorting data items in that shard'"'"'s preliminary result queue based at least in part on primary keys of the data item in that shard'"'"'s preliminary result queue;

until every shard'"'"'s preliminary result queue is empty, repeatedly performing operations comprising;

selecting a first data item from a set comprising data items currently located at tops of preliminary result queues of all of the shards in the multi-shard system,

adding, to a final result set, each first data item that does not have a true tombstone attribute value, and

removing, from every shard'"'"'s preliminary result queue, all data item copies having a primary key that matches a primary key of the first data item; and

after every shard'"'"'s preliminary result queue is empty, returning data items in the final result set as query result.

17. The computer-implemented method of claim 16, wherein selecting the first data item from the set comprising data items currently located at tops of preliminary result queues of all of the shards in the multi-shard system comprises:

selecting, from the set, a subset of data items having a smallest primary key of primary keys of data items in the set; and

selecting, as the first data item, a data item having a largest version number attribute value of version number attribute values of data item in the subset.

18. The computer-implemented method of claim 1, further comprising:

while the one or more data items are being moved from the source shards to the destination shards, the client attempting and failing to perform a delete type of operation in the second manner relative to a first data item having a primary key at least in part by;

determining, based at least in part on a quantity of shards in the multi-shard system preceding the shard quantity change, a particular source shard on which the first data item would have been located prior to the shard quantity change;

determining, based at least in part on a quantity of shards in the multi-shard system following the shard quantity change, a particular destination shard on which the first data item is to be located after the shard quantity change;

determining that no data item on the source shard has the first data item'"'"'s primary key;

determining that no data item on the destination shard has the first data item'"'"'s primary key; and

generating data that indicates that the delete operation failed due to the first data item not existing.

19. A system comprising:

a plurality of database shards that store data items;

a plurality of clients that are configured to transition, in response to a change in a quantity of the plurality of database shards, from a normal state, in which each client of the plurality of clients is configured to perform a particular type of operation relative to the data items in a first manner, to a rebalancing state, in which each client of the plurality of clients is configured to perform the particular type of operation relative to the data items in a second manner that differs from the first manner and without the client acquiring exclusive locks relative to any data items; and

at least one computing device configured to (a) determine, while the plurality of clients are in the rebalancing state, and for one or more particular data items, destination shards that are separate from source shards on which the one or more particular data items were stored while each client of the plurality of clients was in the normal state, and (b) move the one or more data items from the source shards to the destination shards while each client of the plurality of clients is in the rebalancing state.

20. A computer-readable storage memory storing processor-executable instructions comprising:

instructions to cause one or more processors to determine that a quantity of shards in a multi-shard system has changed;

instructions to cause one or more processors to transition a client, in response to a determination that the quantity of shards has changed, from a normal state, in which the client performs a particular type of operation relative to data items stored in the system in a first manner, to a rebalancing state, in which the client performs the particular type of operation relative to data items stored in the system in a second manner that differs from the first manner and without the client acquiring exclusive locks relative to any data items;

instructions to cause one or more processors to determine, while the client is in the rebalancing state, and for one or more particular data items, destination shards that are separate from source shards on which the one or more particular data items were stored prior to the client'"'"'s transition to the rebalancing state; and

instructions to cause one or more processors to move the one or more data items from the source shards to the destination shards while the client is in the rebalancing state.