Making parallel execution of structured query language statements fault-tolerant

• US 8,572,051 B1
• Filed: 08/08/2012
• Issued: 10/29/2013
• Est. Priority Date: 08/08/2012
• Status: Active Grant

• Current Assignee: Oracle International Corporation (Oracle Corporation)
• Original Assignee: Oracle International Corporation (Oracle Corporation)
• Inventors: Zhu, Yali, Cruanes, Thierry, Wood, Alan, Chen, Mengmeng
• Primary Examiner(s): GIRMA, ANTENEH B
• Application Number:
  US13/569,377
• Time in Patent Office: 447 Days
• Field of Search: None
• US Class Current: 707/687
• CPC Class Codes: G06F 16/24532 of parallel queries

Abstract

Parallel execution of SQL statements is made fault-tolerant. Specifically, during parallel query processing, a new row source is created that allows checkpoints to be stored at certain points in the query processing. If a failure occurs, then the query can be restarted from the most recent checkpoint (rather than from the very beginning of the query). The checkpoints are implemented at operation boundaries where rows need to be redistributed among processes. These are natural checkpoint locations because (1) each subsequent operation can proceed from the output of the previous operation, which is stored as a checkpoint, and (2) these operation completions tend to be points in the computation where the query state is minimized, thus minimizing checkpoint storage requirements.

Claims

1. A method comprising:

a particular consumer slave, which receives input rows from producer slaves during parallel processing of a query, processing said input rows as said particular consumer slave receives said input rows;

said particular consumer slave storing output rows, which said particular consumer slave produces by processing said input rows, into a buffer maintained in volatile memory;

determining that all the producer slaves from which the particular consumer slave receives said input rows have finished sending rows to the particular consumer slave;

in response to determining that all the producer slaves from which the particular consumer slave receives said input rows have finished sending rows to the particular consumer slave, storing a copy of rows from the buffer that were stored in the buffer by the particular consumer slave;

wherein said processing comprises performing one or more query operations from a single data flow operator that specifies a set of query operations that are executable on a set of consumer slaves, including the particular consumer slave, without redistributing rows processed by said set of consumer slaves to an other consumer slave to perform said set of one or more query operations; and

wherein said method is performed in a system in which output rows are stored into buffers by consumer slaves at outputs of data flow operators.

2. The method of claim 1, further comprising:

in response to detecting a fault during said parallel processing of said query, restoring query state based on information stored into a non-volatile storage; and

using said information stored into said non-volatile storage to re-perform operations of a particular data flow operator on which the producer slaves were working at a time that the fault was detected; and

resuming parallel processing of said query without re-performing operations of any data flow operator preceding said particular data flow operator in a query execution plan for the query.

3. The method of claim 1, further comprising:

in response to determining that each consumer slave of the set of consumer slaves has finished processing all input rows produced by all producer slaves from which said each consumer slave receives input rows, (1) causing each consumer slave in the set of consumer slaves to become a new producer slave working on the single data flow operator, and (2) causing each producer slave that had been producing rows for the single data flow operator to become a consumer slave that processes rows output by each new producer slave.

4. The method of claim 1, further comprising:

creating a new checkpoint for the single data flow operator by storing a copy of all rows produced by all consumer slaves working on the single data flow operator; and

in response to creating said new checkpoint, deleting, from a non-volatile storage, an old checkpoint that was formerly created for a previous data flow operator.

5. A method comprising:

creating a query execution plan for a query;

wherein creating the query execution plan comprises creating two or more data flow operators that each specify a set of query operations that are executable on a set of consumer slaves without redistributing rows processed by said set of consumer slaves to any other consumer slave to perform said set of one or more query operations;

selecting one or more outputs of the two or more data flow operators at which to place fault-tolerance buffers;

placing fault-tolerance buffers only at each of the said one or more outputs of the data flow operators without placing any fault-tolerance buffers at any operator output that is not a data flow operator output;

in response to detecting a fault during processing of the query, determining a most recent checkpoint stored to non-volatile storage from a fault-tolerance buffer of said fault-tolerance buffers;

restoring a query state based on information contained in said checkpoint;

sending, to a plurality of producer slaves, rows that are contained in said checkpoint; and

said producer slaves processing said rows and sending results of said processing to a plurality of consumer slaves that process said results while said producer slaves are processing said rows, thereby restarting execution said query execution plan from a point after completion of a last data flow operator at which a checkpoint was made without re-performing work performed for any data flow operators preceding or including said last data flow operator.

6. The method of claim 5, wherein the step of selecting the one or more outputs of the two or more data flow operators is based on an amount of work that will not need to be re-performed when a fault occurs in the query execution plan between said one or more outputs and a checkpoint established in the query execution plan previous to said one or more outputs.

7. The method of claim 5, wherein the step of selecting the one or more outputs of the two or more data flow operators is based on an amount of state that will need to be stored in order to establish a checkpoint at the one or more outputs of the two or more data flow operators.

8. The method of claim 5, further comprising:

two or more consumer slaves storing, into a particular fault-tolerance buffer established at an output of a particular data flow operator upon which the two or more consumer slaves are working, output rows that are products of processing that the two or more consumer slaves performed upon input rows that the two or more consumer slaves received from two or more producer slaves working on a data flow operator preceding the particular data flow operator in the query execution plan.

9. The method of claim 5, further comprising:

in response to a particular consumer slave, of a plurality of consumer slaves working on a first data flow operator, receiving an end-of-file message from each of a plurality of producer slaves working on a second data flow operator that precedes the first data flow operator in the query execution plan, the particular consumer slave causing rows, which the particular consumer slave stored into a particular fault-tolerance buffer associated specifically with the first data flow operator and not with the second data flow operator, to be transferred from the particular fault-tolerance buffer onto non-volatile storage.

10. A non-transitory computer-readable medium storing instructions which, when executed by one or more processors, cause performance of a method comprising steps of:

a particular consumer slave, which receives input rows from producer slaves during parallel processing of a query, processing said input rows as said particular consumer slave receives said input rows;

said particular consumer slave storing output rows, which said particular consumer slave produces by processing said input rows, into a buffer maintained in volatile memory;

determining that all the producer slaves from which the particular consumer slave receives said input rows have finished sending rows to the particular consumer slave;

in response to determining that all the producer slaves from which the particular consumer slave receives said input rows have finished sending rows to the particular consumer slave, storing a copy of rows from the buffer that were stored in the buffer by the particular consumer slave;

wherein said processing comprises performing one or more query operations from a single data flow operator that specifies a set of query operations that are executable on a set of consumer slaves, including the particular consumer slave, without redistributing rows processed by said set of consumer slaves to any other consumer slave to perform said set of one or more query operations; and

wherein said method is performed in a system in which output rows are stored into buffers by consumer slaves at outputs of data flow operators.

11. The non-transitory computer-readable medium of claim 10, said steps further comprising:

in response to detecting a fault during said parallel processing of said query, restoring query state based on information stored into a non-volatile storage; and

using said information stored into said non-volatile storage to re-perform operations of a particular data flow operator on which the producer slaves were working at a time that the fault was detected; and

resuming parallel processing of said query without re-performing operations of any data flow operator preceding said particular data flow operator in a query execution plan for the query.

12. The non-transitory computer-readable medium of claim 10, said steps further comprising:

in response to determining that each consumer slave of the set of consumer slaves has finished processing all input rows produced by all producer slaves from which said each consumer slave receives input rows, (1) causing each consumer slave in the set of consumer slaves to become a new producer slave working on the single data flow operator, and (2) causing each producer slave that had been producing rows for the single data flow operator to become a consumer slave that processes rows output by each new producer slave.

13. The non-transitory computer-readable medium of claim 10, said steps further comprising:

creating a new checkpoint for the single data flow operator by storing a copy of all rows produced by all consumer slaves working on the single data flow operator; and

in response to creating said new checkpoint, deleting, from a non-volatile storage, an old checkpoint that was formerly created for a previous data flow operator.

14. A non-transitory computer-readable medium storing instructions which, when executed by one or more processors, cause performance of a method comprising steps of:

creating a query execution plan for a query;

wherein creating the query execution plan comprises creating two or more data flow operators that each specify a set of query operations that are executable on a set of consumer slaves without redistributing rows processed by said set of consumer slaves to any other consumer slave to perform said set of one or more query operations;

selecting one or more outputs of the two or more data flow operators at which to place fault-tolerance buffers;

placing fault-tolerance buffers only at each of the said one or more outputs of the data flow operators without placing any fault-tolerance buffers at any operator output that is not a data flow operator output;

in response to detecting a fault during processing of the query, determining a most recent checkpoint stored to non-volatile storage from a fault-tolerance buffer of said fault-tolerance buffers;

restoring a query state based on information contained in said checkpoint;

sending, to a plurality of producer slaves, rows that are contained in said checkpoint; and

said producer slaves processing said rows and sending results of said processing to a plurality of consumer slaves that process said results while said producer slaves are processing said rows, thereby restarting execution said query execution plan from a point after completion of a last data flow operator at which a checkpoint was made without re-performing work performed for an data flow a operators preceding or including said last data flow operator.

15. The non-transitory computer-readable medium of claim 14, wherein the step of selecting the one or more outputs of the two or more data flow operators is based on an amount of work that will not need to be re-performed when a fault occurs in the query execution plan between said one or more outputs and a checkpoint established in the query execution plan previous to said one or more outputs.

16. The non-transitory computer-readable medium of claim 14, wherein the step of selecting the one or more outputs of the two or more data flow operators is based on an amount of state that will need to be stored in order to establish a checkpoint at the one or more outputs of the two or more data flow operators.

17. The non-transitory computer-readable medium of claim 14, said steps further comprising:

two or more consumer slaves storing, into a particular fault-tolerance buffer established at an output of a particular data flow operator upon which the two or more consumer slaves are working, output rows that are products of processing that the two or more consumer slaves performed upon input rows that the two or more consumer slaves received from two or more producer slaves working on a data flow operator preceding the particular data flow operator in the query execution plan.

18. The non-transitory computer-readable medium of claim 14, said steps further comprising:

in response to a particular consumer slave, of a plurality of consumer slaves working on a first data flow operator, receiving an end-of-file message from each of a plurality of producer slaves working on a second data flow operator that precedes the first data flow operator in the query execution plan, the particular consumer slave causing rows, which the particular consumer slave stored into a particular fault-tolerance buffer associated specifically with the first data flow operator and not with the second data flow operator, to be transferred from the particular fault-tolerance buffer onto non-volatile storage.