System and method for distributed database query engines

• US 9,361,344 B2
• Filed: 06/02/2015
• Issued: 06/07/2016
• Est. Priority Date: 01/07/2013
• Status: Active Grant

• Current Assignee: Meta Platforms, Inc. (f/k/a Facebook, Inc.)
• Original Assignee: Meta Platforms, Inc. (f/k/a Facebook, Inc.)
• Inventors: Murthy, Raghotham, Goel, Rajat
• Primary Examiner(s): Perveen, Rehana
• Assistant Examiner(s): Khong, Alexander
• Application Number:
  US14/728,966
• Publication Number:
  US 20150261831A1
• Time in Patent Office: 371 Days
• Field of Search: 707/770
• US Class Current: 1/1
• CPC Class Codes: G06F 16/2358 Change logging, detection, and notification replication G06F16/27
• CPC Class Codes: G06F 16/24539 using cached or materialised query results
• CPC Class Codes: G06F 16/24552 Database cache management
• CPC Class Codes: G06F 16/2471 Distributed queries
• CPC Class Codes: G06F 16/951 Indexing; Web crawling techniques
• CPC Class Codes: G06F 16/9535 Search customisation based on user profiles and personalisation

Abstract

Techniques for a system capable of performing low-latency database query processing are disclosed herein. The system includes a gateway server and a plurality of worker nodes. The gateway server is configured to divide a database query, for a database containing data stored in a distributed storage cluster having a plurality of data nodes, into a plurality of partial queries and construct a query result based on a plurality of intermediate results. Each worker node of the plurality of worker nodes is configured to process a respective partial query of the plurality of partial queries by scanning data related to the respective partial query that stored on at least one data node of the distributed storage cluster and generate an intermediate result of the plurality of intermediate results that is stored in a memory of that worker node.

Claims

1. A system, comprising:

a gateway server configured to generate a plurality of partial queries from a database query for a database containing data stored in a distributed storage cluster that has a plurality of data nodes, and to construct a query result based on a plurality of intermediate results; and

a plurality of worker nodes, the worker nodes being separate from the data nodes, wherein each worker node of the plurality of worker nodes is configured to process a respective partial query of the plurality of partial queries by scanning data related to the respective partial query and stored on at least one data node of the distributed storage cluster, and wherein each worker node of the plurality of worker nodes is further configured to generate an intermediate result of the plurality of intermediate results that is stored in a memory of that worker node;

wherein the gateway server is further configured to identify a straggling worker node and further divide a partial query that is assigned to the straggling worker node into a plurality of subordinate partial queries and assign the plurality of subordinate partial queries to some of the plurality of worker nodes, and wherein the partial query is divided into the subordinate partial queries based on quantity and location information of input file blocks of the query.

2. The system of claim 1, wherein each worker node of the plurality of worker nodes is further configured to process the respective partial query of the plurality of partial queries by scanning a portion of the data related to the respective partial query that is stored on the at least one data node of the distributed storage cluster and to generate an approximate intermediate result that is stored in the memory of that worker node.

3. The system of claim 2, wherein the gateway server is further configured to construct an approximate query result based on at least one approximate intermediate result.

4. The system of claim 1, wherein the gateway server is further configured to construct an approximate query result based on a portion of the plurality of intermediate results.

5. The system of claim 1, wherein the straggling worker node is a worker node that either fails to report a rate of progress to the gateway server or reports the rate of progress below a specified value after a specified time period to the gateway server.

6. The system of claim 1, wherein each worker node of the plurality of the worker nodes is a service running a respective data node within the distributed storage cluster.

7. The system of claim 1, further comprising:

a metadata cache configured to cache table level metadata of the database and file level metadata of the distributed storage cluster.

8. The system of claim 7, wherein the metadata cache is configured to retain cached metadata from a previous database query for the database query.

9. The system of claim 1, wherein each worker node of the plurality of the worker nodes periodically sends heartbeat messages to the gateway server to report status of a partial query processing by that worker node.

10. The system of claim 1, wherein the gateway server is further configured to receive an instruction from a client device to return an approximate query result or terminate a processing of the database query.

11. The system of claim 1, wherein the gateway server is further configured to instruct the worker nodes to immediately return approximate intermediate results, and to return an approximate query result based on the approximate intermediate results to a client device.

12. The system of claim 1, wherein the database query includes a request for an approximate query result.

13. The system of claim 1, wherein the query result is accompanied by an indication of a portion of related data stored in the data nodes that has been scanned for the query result.

14. The system of claim 1, wherein the database is a Hive data warehouse system and the distributed storage cluster is a Hadoop cluster.

15. A method, comprising:

receiving a database query from a client device, for a database containing data stored in a distributed storage cluster having a plurality of cluster nodes;

dividing the database query into a plurality of partial queries;

sending each of the partial queries to a respective worker node of a plurality of worker nodes, wherein each worker node is a service running on a memory of a cluster node of the distributed storage cluster;

identifying a straggling worker node, dividing a partial query that is assigned to the straggling worker node into a plurality of subordinate partial queries, and assigning the plurality of subordinate partial queries to some of the plurality of worker nodes;

retrieving a plurality of intermediate results for the partial queries from the worker nodes, wherein each intermediate result is processed by a respective worker node of the worker nodes by scanning related data stored in a cluster node on which the perspective worker node runs; and

generating a query result based on the plurality of intermediate results;

,

wherein the partial query is divided into the subordinate partial queries based on quantity and location information of input file blocks of the query.

16. The method of claim 15, wherein the step of identifying comprises:

identifying a straggling worker node by monitoring heartbeat messages that the worker nodes periodically send, wherein the straggling worker node is identified when heartbeat messages from the straggling worker node are not received for a predetermined time period, or when a heartbeat message from the straggling worker node is received and the heartbeat message including a number representing a status of a partial query processing by the straggling worker node that is below a threshold value.

17. The method of claim 15, further comprising:

caching data associated with previous database queries for the database in a cache;

retrieving a real-time feed of audit logs of the database to invalidate entries in the cached data stored in the cache that have been changed by the previous database queries; and

purging entries in the cached data from the cache that have not been queried for a specified time period.