Fault Tolerant Distributed Query Processing Using Query Operator Motion

• US 20140310258A1
• Filed: 04/15/2013
• Published: 10/16/2014
• Est. Priority Date: 04/15/2013
• Status: Active Grant

• Current Assignee: Vmware LLC (Broadcom, Inc.)
• Original Assignee: VMware, Inc. (Broadcom, Inc.)
• Inventors: Tian, Feng
• Granted Patent: US 9,659,057 B2
• Time in Patent Office: Days
• US Class Current: 707/718
• CPC Class Codes: G06F 16/24532 of parallel queries
• CPC Class Codes: G06F 16/2471 Distributed queries

Abstract

Techniques for enabling fault tolerant distributed query processing are provided. In one embodiment, a first processing node in a plurality of processing nodes can execute an instance of a query operator, where the query operator is part of a query plan for a database query that is processed in a distributed manner by the plurality of processing nodes. The first processing node can further generate a snapshot of the instance of the query operator, where the snapshot includes state information usable for moving execution of the query operator from the first processing node to another processing node after a failure of the first processing node. The first processing node can then save the snapshot to a distributed data store.

Claims

1. A method for enabling fault tolerant distributed query processing, the method comprising:

executing, by a first processing node in a plurality of processing nodes, a first instance of a first query operator, the first query operator being part of a query plan for a database query that is processed in a distributed manner by the plurality of processing nodes;

generating, by the first processing node, a snapshot of the first instance of the first query operator, the snapshot including state information usable for moving execution of the first query operator from the first processing node to another processing node after a failure of the first processing node; and

saving, by the first processing node, the snapshot to a distributed data store.

2. The method of claim 1 further comprising:

transmitting a result of executing the first instance of the first query operator to a second processing node in the plurality of processing nodes, the second processing node executing an instance of a second query operator in the query plan that is a parent of the first query operator.

3. The method of claim 2 wherein the transmitting is performed without storing the result in the distributed data store.

4. The method of claim 1 further comprising, by a management server associated with the plurality of processing nodes:

detecting, during the processing of the query plan by the plurality of processing nodes, a failure of the first processing node; and

allocating a second processing node to replace the first processing node, the second processing node being configured to execute a second instance of the first query operator.

5. The method of claim 4 further comprising, by the second processing node:

loading the snapshot of the first instance of the first query operator from the distributed data store; and

initiating execution of the second instance of the first query operator from a starting point identified by the state information included in the snapshot.

6. The method of claim 4 further comprising, by the management server:

determining one or more instances of other query operators in the query plan that are affected by the failure of the first processing node, the one or more instances running on one or more other processing nodes in the plurality of processing nodes; and

causing the one or more instances of the other query operators to restart execution from a previous snapshot.

7. The method of claim 1 wherein:

if the first query operator is a select or scan operator, the state information comprises an offset into an input file or database table;

if the first query operator is a hash join operator, the state information comprises a hash table of entries from a first database table and an offset into a second database table;

if the first query operator is a summation operator, the state information comprises a running total; and

if the first query operator is a sort operator, the state information comprises a sort buffer.

8. A non-transitory computer readable storage medium having stored thereon computer software executable by a first processing node in a plurality of processing nodes, the computer software embodying a method comprising:

executing a first instance of a first query operator, the first query operator being part of a query plan for a database query that is processed in a distributed manner by the plurality of processing nodes;

generating a snapshot of the first instance of the first query operator, the snapshot including state information usable for moving execution of the first query operator from the first processing node to another processing node after a failure of the first processing node; and

saving the snapshot to a distributed data store.

9. The non-transitory computer readable storage medium of claim 8 wherein the method further comprises:

transmitting a result of executing the first instance of the first query operator to a second processing node in the plurality of processing nodes, the second processing node executing an instance of a second query operator in the query plan that is a parent of the first query operator.

10. The non-transitory computer readable storage medium of claim 9 wherein the transmitting is performed without storing the result in the distributed data store.

11. The non-transitory computer readable storage medium of claim 8 wherein a management server associated with the plurality of processing nodes is configured to:

detect, during the processing of the query plan by the plurality of processing nodes, a failure of the first processing node; and

allocating a second processing node to replace the first processing node, the second processing node being configured to execute a second instance of the first query operator.

12. The non-transitory computer readable storage medium of claim 11 wherein the second computer system is configured to:

load the snapshot of the first instance of the first query operator from the distributed data store; and

initiate execution of the second instance of the first query operator from a starting point identified by the state information included in the snapshot.

13. The non-transitory computer readable storage medium of claim 11 wherein the management server is further configured to:

determine one or more instances of other query operators in the query plan that are affected by the failure of the first processing node, the one or more instances running on one or more other processing nodes in the plurality of processing nodes; and

cause the one or more instances of the other query operators to restart execution from a previous snapshot.

14. The non-transitory computer readable storage medium of claim 8 wherein:

if the first query operator is a select or scan operator, the state information comprises an offset into an input file or database table;

if the first query operator is a hash join operator, the state information comprises a hash table of entries from a first database table and an offset into a second database table;

if the first query operator is a summation operator, the state information comprises a running total; and

if the first query operator is a sort operator, the state information comprises a sort buffer.

15. A computer system comprising:

a processor; and

a non-volatile data store having stored thereon program code which, when executed by the processor, causes the processor to;

execute a first instance of a first query operator, the first query operator being part of a query plan for a database query that is processed in a distributed manner by a plurality of computer systems;

generate a snapshot of the first instance of the first query operator, the snapshot including state information usable for moving execution of the first query operator from the computer system to another computer system after a failure of the computer system; and

save the snapshot to a distributed data store.

16. The system of claim 15 where the program code further causes the processor to:

transmit a result of executing the first instance of the first query operator to a second computer system in the plurality of computer systems, the second computer system executing an instance of a second query operator in the query plan that is a parent of the first query operator.

17. The system of claim 16 wherein the transmitting is performed without storing the result in the distributed data store.

18. The system of claim 15 wherein a management server associated with the plurality of computer systems is configured to:

detect, during the processing of the query plan by the plurality of computer systems, a failure of the computer system; and

allocate a second computer system to replace the computer system, the second computer system being configured to execute a second instance of the first query operator.

19. The system of claim 18 wherein the second computer system is configured to:

load the snapshot of the first instance of the first query operator from the distributed data store; and

initiate execution of the second instance of the first query operator from a starting point identified by the state information included in the snapshot.

20. The system of claim 18 wherein the management server is further configured to:

determine one or more instances of other query operators in the query plan that are affected by the failure of the computer system, the one or more instances running on one or more other computer systems in the plurality of computer systems; and

cause the one or more instances of the other query operators to restart execution from a previous snapshot.

21. The system of claim 15 wherein:

if the first query operator is a select or scan operator, the state information comprises an offset into an input file or database table;

if the first query operator is a hash join operator, the state information comprises a hash table of entries from a first database table and an offset into a second database table;

if the first query operator is a summation operator, the state information comprises a running total; and

if the first query operator is a sort operator, the state information comprises a sort buffer.