System and methods for distributed database query engines

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;

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, 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 at least one of the worker nodes is further configured to divide the respective partial query into subordinate partial queries; and

a cache configured to:

cache data associated with previous database queries for the database;

invalidate entries in the cached data that have been changed; and

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

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 at least a portion of data related to the respective partial query that is stored on at least one data node of the distributed storage cluster and to generate at least one 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 the 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 gateway server is further configured to identify a straggling worker node, 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, 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 the cache is further configured to retrieve 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.

7. 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.

8. 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.

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

10. 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.

11. 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.

12. 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.

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

14. 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.

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

16. 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 based on at least one of quantity or location information of input file blocks of the query, and assigning the plurality of subordinate partial queries to some of the plurality of worker nodes; and

configuring a cache to cache data associated with previous database queries for the database and to purge entries in the cached data from the cache that have not been queried for a specified time period.

17. The method of claim 16 , further comprising:

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 respective worker node runs; and

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

18. The method of claim 16 , 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 includes a status of a partial query processing by the straggling worker node that is below a threshold value.

19. The method of claim 16 , further comprising

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.

20. 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;

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.