Request routing and query processing in a sharded database

• US 11,204,900 B2
• Filed: 05/03/2019
• Issued: 12/21/2021
• 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:
  US16/402,864
• Publication Number:
  US 20190258613A1
• Time in Patent Office: 963 Days
• Field of Search: None
• 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:

performing a first operation that involves data that resides in both a first shard and a second shard of a sharded database by;

causing a client application to send a first request to the first shard to cause the first shard to perform a first portion of the first operation that involves data in the first shard, and

causing the client application to send a second request to the second shard to cause the second shard to perform a second portion of the first operation that involves data in the second shard;

wherein;

the data in the first shard involved in the first operation corresponds to a first data criterion;

the data in the second shard involved in the first operation corresponds to a second data criterion; and

the first data criterion is different than the second data criterion;

performing a second operation that involves data that resides in both the first shard and the second shard of the sharded database by;

causing the client application to send a third request to a shard coordinator; and

in response to the third request, the shard coordinator causing;

the first shard to perform a first portion of the second operation that involves data in the first shard, and

the second shard to perform a second portion of the second operation that involves data in the second shard.

2. The method of claim 1, wherein the shard coordinator is executing, at least in part, within the first shard.

3. The method of claim 1, wherein the shard coordinator is executing, at least in part, within the second shard.

4. The method of claim 1, wherein the shard coordinator is executing external to the first shard, the second shard, and the client application.

5. The method of claim 1, wherein:

the method further comprises storing, by the shard coordinator, a plurality of map entries;

each map entry, of the plurality of map entries, maps a distinct sharding key value set to a plurality of shards including the first shard and the second shard;

the third request comprises a first sharding key value that is mapped to the first shard and a second sharding key value that is mapped to the second shard;

using the plurality of map entries to identify data in the first shard corresponding to the first sharding key value;

using the plurality of map entries to identify data in the second shard corresponding to the second sharding key value.

6. The method of claim 5, wherein for each map entry of the plurality of map entries, mapping the distinct sharding key value set to a shard of the plurality of shards comprises applying a consistent hash function to a sharding key value and mapping a result of the consistent hash function to the shard.

7. The method of claim 5, wherein:

the method further comprises sending, by the shard coordinator, 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 a shard;

the third request comprising a first sharding key value that is mapped to the first shard and a second sharding key value that is mapped to the second shard is generated from the locally cached map entries at the client application.

8. The method of claim 5, wherein:

the first portion of the second operation comprises generating a first result for the first portion of the second operation;

the second portion of the second operation comprises generating a second result for the second portion of the second operation;

the method further comprises;

sending, to the client application, the first result;

sending, to the client application, the second result.

9. The method of claim 8, wherein:

the second operation comprises the sending, to the client application, both the first result and the second result concurrently;

the second operation is performed only when both the first result and the second result are generated.

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

performing a first operation that involves data that resides in both a first shard and a second shard of a sharded database by;

causing a client application to send a first request to the first shard to cause the first shard to perform a first portion of the first operation that involves data in the first shard, and

causing the client application to send a second request to the second shard to cause the second shard to perform a second portion of the first operation that involves data in the second shard;

wherein;

the data in the first shard involved in the first operation corresponds to a first data criterion;

the data in the second shard involved in the first operation corresponds to a second data criterion; and

the first data criterion is different than the second data criterion;

performing a second operation that involves data that resides in both the first shard and the second shard of the sharded database by;

causing the client application to send a third request to a shard coordinator; and

in response to the third request, the shard coordinator causing;

the first shard to perform a first portion of the second operation that involves data in the first shard, and

the second shard to perform a second portion of the second operation that involves data in the second shard.

11. The one or more non-transitory computer-readable media of claim 10, wherein the shard coordinator is executing, at least in part, within the first shard.

12. The one or more non-transitory computer-readable media of claim 10, wherein the shard coordinator is executing, at least in part, within the second shard.

13. The one or more non-transitory computer-readable media of claim 10, wherein the shard coordinator is executing external to the first shard, the second shard, and the client application.

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

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

each map entry, of the plurality of map entries, maps a distinct sharding key value set to a plurality of shards including the first shard and the second shard;

the third request comprises a first sharding key value that is mapped to the first shard and a second sharding key value that is mapped to the second shard;

using the plurality of map entries to identify data in the first shard corresponding to the first sharding key value;

using the plurality of map entries to identify data in the second shard corresponding to the second sharding key value.

15. The one or more non-transitory computer-readable media of claim 14, wherein for each map entry of the plurality of map entries, mapping the distinct sharding key value set to a shard of the plurality of shards comprises applying a consistent hash function to a sharding key value and mapping a result of the consistent hash function to the shard.

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

the instructions, when executed by the one or more processors, further cause sending, by the shard coordinator, 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 a shard;

the third request comprising a first sharding key value that is mapped to the first shard and a second sharding key value that is mapped to the second shard is generated from the locally cached map entries at the client application.

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

the first portion of the second operation comprises generating a first result for the first portion of the second operation;

the second portion of the second operation comprises generating a second result for the second portion of the second operation;

the instructions, when executed by the one or more processors, further cause;

sending, to the client application, the first result;

sending, to the client application, the second result.

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

the second operation comprises the sending, to the client application, both the first result and the second result concurrently;

the second operation is performed only when both the first result and the second result are generated.