Range query capacity allocation

• US 10,866,970 B1
• Filed: 09/15/2016
• Issued: 12/15/2020
• Est. Priority Date: 05/20/2013
• Status: Active Grant

• Current Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Original Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Inventors: Xiao, Wei, Filipe, Miguel Mascarenhas, Swift, Bjorn Patrick, Muniswamy-Reddy, Kiran-Kumar, Marshall, Stuart Henry Seelye, Rath, Timothy Andrew, Blackman, Rande A.
• Primary Examiner(s): Willoughby, Alicia M
• Application Number:
  US15/266,981
• Time in Patent Office: 1,552 Days
• CPC Class Codes: G06F 16/27 Replication, distribution or synchronisation of data between databases or within a distributed database system; Distributed database system architectures therefor
• CPC Class Codes: G06F 16/278 Data partitioning, e.g. horizontal or vertical partitioning

Abstract

Distributed database management systems may perform range queries over the leading portion of a primary key. Non-random distribution of data may improve performance related to the processing of range queries, but may tend to cause workload to be concentrated on particular partitions. Groups of partitions may be expanded and collapsed based on detection of disproportionate workload. Disproportionate write workload may be distributed among a group of partitions that can subsequently be queried using a federated approach. Disproportionate read workload may be distributed among a group of read-only replicated partitions.

Claims

1. A computer-implemented method, comprising:

hosting a database on a plurality of groups of computing nodes, wherein;

individual data items in a table in the database are assigned to respective ones of a plurality of portions of the database according to respective primary key values of the individual data items, and

individual ones of the portions of the database are hosted by respective ones of the groups;

determining whether a first rate of read accesses to a first portion of the database exceeds a threshold value, the first portion hosted by a group in the plurality of groups;

in response to a determination that the first rate of read accesses exceeds the threshold value;

adding another computing node to the group as a read-only member of the group that does not accept requests to store new data items to the first portion; and

replicating the first portion of the database from one computing node in the group to the other computing node in the group;

subsequent to the replication, maintaining consistency of a committed state of the first portion of the database between the one computing node and the other computing node, including propagating data in the first portion from the one computing node to the other computing node; and

servicing at least one read request of a plurality of read requests to the group by using the replicated first portion of the database stored by the other computing node so that read workload targeted to the group is shared by the one computing node and the other computing node.

2. The computer-implemented method of claim 1, wherein the group is a quorum group that includes voting members that vote to determine whether individual write operations are committed, and wherein the other computing node is a non-voting member of the quorum group.

3. The computer-implemented method of claim 1, wherein the replicating of the first portion is performed as part of a chain replication from another non-voting computing node in the group.

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

determining that one or more utilization metrics for a second portion of the database on a second computing node in the group indicates that the second computing node is underutilized; and

responsive to the determination that the second computing node is underutilized, removing the second computing node from the group.

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

determining that a utilization metric for a second portion of the database hosted on a second computing node of the group is excessive based at least in part on an average value of the utilization metric for the group, wherein the utilization metric is distinct from a rate of read accesses; and

replicating the second portion of the database in response to the determination that the utilization metric for a second portion exceeds the average value.

6. The computer-implemented method of claim 1, wherein the one computing node hosts data items in the table whose primary key values fall within a range, and replicating the first portion on the other computing node in the group comprises replicating a subset of the data items whose primary key values fall within a more limited range of the range on the other computing node.

7. The computer-implemented method of claim 6, wherein replicating the first portion to another computing node in the group comprises:

locating a split point in the range of the table based at least in part on the first rate of read accesses, wherein the subset of data items to be replicated on the other node on one side of the split point.

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

determining that an access response time of accesses to a second portion of the database exceeds a second threshold value;

replicating the second portion of the database in response to the access response time of accesses exceeding the second threshold value; and

servicing at least one request to the database by using the replicated second portion of the database.

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

determining, based at least in part on a predictive model, that a predicted value of a utilization metric of a second portion of the database will exceed one or more workload criteria;

replicating the second portion of the database in response to the determination that the predicted value of a utilization metric of the second portion of the database will exceed the workload criteria; and

servicing at least one request to the database by using the replicated second portion of the database.

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

monitoring accesses to the first portion of the database, wherein the determining of whether the first rate of read accesses to the first portion of exceeds the threshold value is performed based at least in part on the monitoring.

11. The computer-implemented method of claim 1, further comprising generating a hotspot notification in response to the first rate of read accesses exceeding the threshold value.

12. The computer-implemented method of claim 1, further comprising selecting the replicated first portion of the database to service the read request from among a plurality of replicas of the first portion, wherein the selection is based at least in part on a random selection or a round-robin selection.

13. The computer-implemented method of claim 1, further comprising selecting the replicated first portion of the database to service the read request based at least in part on a probability that a requested data item is present in the replicated first portion.

14. A system, comprising:

one or more computing devices with respective processors and memories storing program instructions to;

host a database on a plurality of groups of computing nodes, wherein;

individual data items in a table in the database are assigned to respective ones of a plurality of portions of the database according to respective primary key values of the individual data items, and

individual ones of the portions of the database are hosted by respective ones of the groups;

determine whether a first rate of read accesses to a first portion of the database exceeds a threshold value, the first portion hosted by a group in the plurality of groups;

in response to a determination that the first rate of read accesses exceeds the threshold value;

add another computing node to the group as a read-only member of the group that does not accept requests to store new data items to the first portion; and

replicate the first portion of the database from one computing node in the group to the other computing node in the group;

subsequent to the replication, propagate data in the first portion from the one computing node to the other computing node to maintain consistency of a committed state of the first portion of the database between the one computing node and the other computing node; and

service at least one read request of a plurality of read requests to the group by using the replicated first portion of the database stored by the other computing node so that read workload targeted to the group is shared by the one computing node and the other computing node.

15. The system of claim 14, wherein the group is a quorum group that includes voting members that vote to determine whether individual write operations are committed, and wherein the other computing node is a non-voting member of the quorum group.

16. The system of claim 14, wherein the one or more computing devices are to:

determine that a utilization metric for a second portion of the database hosted on a second computing node of the group is excessive based at least in part on one or more workload criteria, wherein the utilization metric is distinct from a rate of read accesses; and

replicate the second portion of the database in response to the determination that the utilization metric for a second portion is excessive.

17. The system of claim 16, wherein to determine that the utilization metric is excessive based at least in part on one or more workload criteria, the one or more computing devices to determine that the utilization metric for the second portion of the database hosted on a second computing node of the group is excessive based at least in part on an average value of the utilization metric for the group.

18. One or more non-transitory computer-accessible storage media storing program instructions that when executed on or across one or more processors causes the one or more processors to:

determine whether a first rate of read accesses to a first portion of a database exceeds a threshold value, wherein;

individual data items in a table in the database are assigned to respective ones of a plurality of portions of the database according to respective primary key values of the individual data items, and

the first portion is hosted by one of a plurality of groups of computing nodes hosting respective ones of the portions of the database;

in response to a determination that the first rate of read accesses exceeds the threshold value;

add another computing node to the group as a read-only member of the group that does not accept requests to store new data items to the first portion; and

replicate the first portion of the database from one computing node in the group to the other computing node in the group;

subsequent to the replication, propagate data in the first portion from the one computing node to the other computing node to maintain consistency of a committed state of the first portion of the database between the one computing node and the other computing node; and

service at least one read request of a plurality of read requests to the group by using the replicated first portion of the database stored by the other computing node so that read workload targeted to the group is shared by the one computing node and the other computing node.

19. The non-transitory computer-accessible storage media of claim 18, wherein the group is a quorum group that includes voting members that vote to determine whether individual write operations are committed, and wherein the other computing node is a non-voting member of the quorum group.

20. The non-transitory computer-accessible storage media of claim 18, wherein the one computing node hosts data items in the table whose primary key values fall within a range, and to replicate the first portion on another computing node in the group, the program instructions when executed on the one or more processors causes the one or more processors to replicate a subset of the data items whose primary key values fall within a more limited range of the range on the other computing node.