Request routing and query processing in a sharded database

• US 10,331,634 B2
• Filed: 09/16/2016
• Issued: 06/25/2019
• Est. Priority Date: 10/07/2015
• Status: Active Grant

• Current Assignee: Oracle International Corporation (Oracle Corporation)
• Original Assignee: Oracle International Corporation (Oracle Corporation)
• Inventors: Hu, Wei-Ming, Dilman, Mark, Novak, Leonid, Taranov, Ilya, Zeng, Zheng, de Lavarene, Jean, Hegde, Vidya
• Primary Examiner(s): Nguyen, Kim T
• Application Number:
  US15/268,108
• Publication Number:
  US 20170103094A1
• Time in Patent Office: 1,012 Days
• Field of Search: 707755, 707756
• CPC Class Codes: G06F 16/213 with details for schema evolution support
• CPC Class Codes: G06F 16/217 Database tuning G06F16/2282 takes precedence; database performance monitoring G06F11/3409
• CPC Class Codes: G06F 16/221 Column-oriented storage; Management thereof
• CPC Class Codes: G06F 16/2272 Management thereof
• CPC Class Codes: G06F 16/2282 Tablespace storage structures; Management thereof
• CPC Class Codes: G06F 16/2455 Query execution
• CPC Class Codes: G06F 16/2471 Distributed queries
• CPC Class Codes: G06F 16/248 Presentation of query results
• CPC Class Codes: G06F 16/252 between a Database Management System and a front-end application
• 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

Techniques are provided for creating, organizing, and maintaining a sharded database. A sharded database can be created using user-defined sharding, system-managed sharding, or composite sharding. The sharded database is implemented with relational database techniques. The techniques described can be used to route database requests and process queries in a sharded database.

Claims

1. A method, comprising:

storing, by a shard coordinator, a plurality of map entries;

wherein each map entry, of the plurality of map entries, maps a distinct sharding key value set to a shard of a plurality of shards of a sharded database;

wherein at least one shard of the plurality of shards is mapped to a sharding key value set that comprises non-contiguous sharding key values;

wherein each shard of the plurality of shards comprises one or more database servers;

receiving, at the shard coordinator, a first query from a client application, wherein the first query comprises a first sharding key value;

in response to receiving the first query, performing;

identifying a particular map entry, of the plurality of map entries, that corresponds to a sharding key value set that includes the first sharding key value;

using the particular map entry to identify a target shard, from the plurality of shards, that corresponds to the sharding key value set that includes the first sharding key value;

generating a first result for the first query by using the target shard that corresponds to the sharding key value set that includes the first sharding key value; and

sending, to the client application, a response to the first query that is based on the first result;

wherein generating the first result for the first query and sending the response to the first query are performed by the target shard.

2. The method of claim 1 further comprising:

redirecting, by the shard coordinator, the client application to the target shard; and

receiving, at the target shard, the first query.

3. The method of claim 2, wherein sending the response to the first query to the client application is performed directly by the target shard without involving the shard coordinator.

4. The method of claim 1, wherein:

the respective shard to which each map entry maps a sharding key value set is the shard, of the plurality of shards, that includes a chunk associated with the sharding key value set; and

generating the first result for the first query comprises generating the first result for the first query by using the chunk, at the target shard, that corresponds to the sharding key value set that includes the first sharding key value.

5. The method of claim 4, wherein the method further comprises:

sending, by the target shard, a subset of the plurality of map entries to the client application for local caching, wherein the subset of the plurality of map entries maps a particular sharding key value set to the target shard.

6. The method of claim 5, further comprising:

receiving, by the target shard, a second query from a client application, wherein the second query comprises a second sharding key value;

wherein the second query was sent directly from the client application to the target shard in response to looking up the subset of the plurality of map entries in a local cache available to the client application; and

wherein the second query was sent directly from the client application to the target shard without involving the shard coordinator.

7. The method of claim 6, further comprising:

generating a second result for the second query by using a chunk, at the target shard, that corresponds to a sharding key value set that includes the second sharding key value; and

sending, to the client application, a response to the second query that is based on the second result.

8. The method of claim 4, further comprising:

in response to receiving the first query, coordinating, by the shard coordinator, a processing of the first query on the target shard; and

wherein the sending, by the target shard to the client application, a response to the first query that is based on the first result, is performed by sending the response from the target shard to the shard coordinator and sending the response to the client application from the shard coordinator;

wherein the response sent to the shard coordinator by the target shard includes the first result.

9. The method of claim 8, wherein coordinating, by the shard coordinator, the processing of the first query on the target shard comprises:

sending the first query, from the shard coordinator to the target shard;

executing the first query on the target shard by using the chunk, at the target shard, that corresponds to the sharding key value set that includes the first sharding key value to generate the first result;

and

aggregating, by the shard coordinator, the first result and a second result into an aggregated result;

wherein the second result was received by the shard coordinator from a second shard; and

wherein sending the response to the client application by the shard coordinator includes the aggregated result.

10. The method of claim 8, wherein coordinating, by the shard coordinator, the processing of the first query on the target shard comprises:

sending a first predicate of the first query from the shard coordinator to the target shard;

executing the first predicate on the target shard by using the chunk, at the target shard, that corresponds to the sharding key value set that includes the first sharding key value to generate the first result;

and

executing, by the shard coordinator, the first query on the first result and a second result to generate a final result;

wherein the second result was received by the shard coordinator from a second shard; and

wherein the sending of the response to the client application by the shard coordinator includes the final result.

11. The method of claim 1, wherein the first query is written in a query language.

12. The method of claim 11, wherein the query language is a Structured Query Language (SQL).

13. The method of claim 1, wherein each shard of the plurality of shards does not share a processor, a memory, or disk storage with another shard of the plurality of shards.

14. The method of claim 1, further comprising:

for each map entry of the plurality of map entries, mapping the distinct sharding key value set to a shard of a plurality of shards of a sharded database based, at least in part, on a consistent hash function applied to a sharding key.

15. The method of claim 1, wherein the shard coordinator is executing external to both the client application and the sharded database.

16. A method comprising:

storing, by a shard coordinator, a plurality of map entries;

wherein each map entry, of the plurality of map entries, maps a distinct sharding key value set to a shard of a plurality of shards of a sharded database;

wherein each shard of the plurality of shards comprises one or more database servers;

receiving, at the shard coordinator, a first query from a client application, wherein the first query comprises a first sharding key value;

in response to receiving the first query, performing;

identifying a particular map entry, of the plurality of map entries, that corresponds to the sharding key value set that includes the first sharding key value;

using the particular map entry to identify a target shard from the plurality of shards;

generating a first result for the first query by using the target shard that corresponds to the sharding key value set that includes the first sharding key value;

sending, to the client application, a response to the first query that is based on the first result; and

for each map entry of the plurality of map entries, mapping the distinct sharding key value set to the shard of a plurality of shards of the sharded database based, at least in part, on user-defined instructions;

wherein the user-defined instructions include a mapping criterion, wherein the mapping criterion specifies either;

which specific ranges map to which specific shards, or

which specific individual sharding key values map to which specific shards.

17. One or more non-transitory computer-readable media storing instructions, wherein the instructions include:

instructions which, when executed by one or more hardware processors, cause storing, by a shard coordinator, a plurality of map entries;

wherein each map entry, of the plurality of map entries, maps a distinct sharding key value set to a shard of a plurality of shards of a sharded database;

wherein at least one shard of the plurality of shards is mapped to a sharding key value set that comprises non-contiguous sharding key values;

wherein each shard of the plurality of shards comprises one or more database servers;

instructions which, when executed by one or more hardware processors, cause receiving, at the shard coordinator, a first query from a client application, wherein the first query comprises a first sharding key value;

instructions which, when executed by one or more hardware processors, cause, in response to receiving the first query, identifying a particular map entry, of the plurality of map entries, that corresponds to a sharding key value set that includes the first sharding key value;

instructions which, when executed by one or more hardware processors, cause, in response to receiving the first query, using the particular map entry to identify a target shard, from the plurality of shards, that corresponds to the sharding key value set that includes the first sharding key value;

instructions which, when executed by one or more hardware processors, cause, in response to receiving the first query, generating a first result for the first query by using the target shard that corresponds to the sharding key value set that includes the first sharding key value; and

instructions which, when executed by one or more hardware processors, cause, in response to receiving the first query, sending, to the client application, a response to the first query that is based on the first result;

wherein generating the first result for the first query and sending the response to the first query are performed by the target shard.

18. The one or more non-transitory computer-readable media of claim 17, further comprising:

instructions which, when executed by one or more hardware processors, cause redirecting, by the shard coordinator, the client application to the target shard; and

instructions which, when executed by one or more hardware processors, cause receiving, at the target shard, the first query.

19. The one or more non-transitory computer-readable media of claim 18, wherein the instructions which, when executed by one or more hardware processors, cause sending the response to the first query to the client application are executed directly by the target shard without involving the shard coordinator.

20. The one or more non-transitory computer-readable media of claim 17, wherein:

the respective shard to which each map entry maps a sharding key value set is the shard, of the plurality of shards, that includes a chunk associated with the sharding key value set; and

the instructions for generating the first result for the first query comprising instructions which, when executed by one or more hardware processors, cause generating the first result for the first query by using the chunk, at the target shard, that corresponds to the sharding key value set that includes the first sharding key value.

21. The one or more non-transitory computer-readable media of claim 20, further comprising:

instructions which, when executed by one or more hardware processors, cause sending, by the target shard, a subset of the plurality of map entries to the client application for local caching, wherein the subset of the plurality of map entries maps a particular sharding key value set to the target shard.

22. The one or more non-transitory computer-readable media of claim 21, further comprising:

instructions which, when executed by one or more hardware processors, cause receiving, by the target shard, a second query from a client application, wherein the second query comprises a second sharding key value;

wherein the second query was sent directly from the client application to the target shard in response to looking up the subset of the plurality of map entries in a local cache available to the client application; and

wherein the second query was sent directly from the client application to the target shard without involving the shard coordinator.

23. The one or more non-transitory computer-readable media of claim 22, further comprising:

instructions which, when executed by one or more hardware processors, cause generating a second result for the second query by using the chunk, at the target shard, that corresponds to the sharding key value set that includes the second sharding key value; and

instructions which, when executed by one or more hardware processors, cause sending, to the client application, a response to the second query that is based on the second result.

24. The one or more non-transitory computer-readable media of claim 20, further comprising instructions which, when executed by one or more hardware processors, cause in response to receiving the first query, coordinating, by the shard coordinator, a processing of the first query on the target shard;

and wherein the sending, by the target shard to the client application, a response to the first query that is based on the first result is performed by sending the response from the target shard to the shard coordinator;

wherein the response sent to the shard coordinator by the target shard includes the first result.

25. The one or more non-transitory computer-readable media of claim 24, further comprising:

instructions which, when executed by one or more hardware processors, cause sending the first query, from the shard coordinator to the target shard;

instructions which, when executed by one or more hardware processors, cause executing the first query on the target shard by using the chunk, at the target shard, that corresponds to the sharding key value set that includes the first sharding key value to generate the first result;

and

instructions which, when executed by one or more hardware processors, cause aggregating, by the shard coordinator, the first result and a second result into an aggregated result;

wherein the second result was received by the shard coordinator from a second shard; and

wherein sending the response to the client application by the sharding coordinator includes the aggregated result.

26. The one or more non-transitory computer-readable media of claim 24, further comprising:

instructions which, when executed by one or more hardware processors, cause sending a first predicate of the first query, from the shard coordinator to the target shard;

instructions which, when executed by one or more hardware processors, cause executing the first predicate on the target shard by using the chunk, at the target shard, that corresponds to the sharding key value set that includes the first sharding key value to generate the first result;

instructions which, when executed by one or more hardware processors, cause executing, by the shard coordinator, the first query on the first result and a second result to generate a final result;

wherein the second result was received by the shard coordinator from a second shard; and

wherein the sending of the response to the client application by the shard coordinator includes the final result.

27. The one or more non-transitory computer-readable media of claim 17, wherein the first query is written in a query language.

28. The one or more non-transitory computer-readable media of claim 27, wherein the query language is a Structured Query Language (SQL).

29. The one or more non-transitory computer-readable media of claim 17, wherein each shard of the plurality of shards does not share a processor, a memory, or disk storage with another shard of the plurality of shards.

30. The one or more non-transitory computer-readable media of claim 17, further comprising:

instructions which, when executed by one or more hardware processors, cause, for each map entry of the plurality of map entries, mapping the distinct sharding key value set to a shard of a plurality of shards of a sharded database based, at least in part, on a consistent hash function applied to a sharding key.

31. The one or more non-transitory computer-readable media of claim 17, wherein the shard coordinator is external to both the client application and the sharded database.

32. One or more non-transitory computer-readable media storing instructions which, when executed by one or more processors, cause:

storing, by a shard coordinator, a plurality of map entries;

wherein each map entry, of the plurality of map entries, maps a distinct sharding key value set to a shard of a plurality of shards of a sharded database;

wherein each shard of the plurality of shards comprises one or more database servers;

receiving, at the shard coordinator, a first query from a client application, wherein the first query comprises a first sharding key value;

in response to receiving the first query, performing;

identifying a particular map entry, of the plurality of map entries, that corresponds to the sharding key value set that includes the first sharding key value;

using the particular map entry to identify a target shard from the plurality of shards;

generating a first result for the first query by using the target shard that corresponds to the sharding key value set that includes the first sharding key value; and

sending, to the client application, a response to the first query that is based on the first result;

for each map entry of the plurality of map entries, mapping the distinct sharding key value set to a shard of a plurality of shards of a sharded database based, at least in part, on user-defined instructions;

wherein the user-defined instructions include a mapping criterion, wherein the mapping criterion specifies either;

which specific ranges map to which specific shards, or

which specific individual sharding key values map to which specific shards.