Smart Intra-query Fault Tolerance for Massive Parallel Processing Databases

Typical MPP databases usually adopt a shared-nothing architecture [2], composed of one master node and n slave nodes. The master node is responsible for interacting with clients, managing the whole cluster and coordinating the query processing. Each of the n salve nodes is responsible for storing a partition of the data and performing query processing on its partition. Each slave node hosts d database instances, which will be referred to as segments subsequently. Figure 1 illustrates an example of this architecture, in which \({n = 4}\) and \({d = 2}\).

Most MPP databases provide fault tolerance at storage level. A mirror scheme, i.e., replication, is commonly used to ensure durability and availability of data. As Fig. 1 shows, each segment (primary segment) is allocated with a mirror (mirror segment) in another node. The master node detects node failures by monitoring heartbeats of slave nodes. If a slave node stops responding for a certain amount of time, known as a system delay time (normally around 1 min), the master will treat it as a failed node. Once a failure is detected, the corresponding mirror will be activated to replace the failed primary segment. Similarly, a standby works as the replication/mirror of the master node. Through the mirror scheme, the system’s availability can be significantly enhanced.

However, such a mirror scheme does not support intra-query fault tolerance automatically. When a node failure occurs, the running query’s state on the failed node will be lost. After the corresponding mirror is activated, the whole query has to be rerun. If it is a long running query, response to the client will be severely delayed. In the worst case, a query will run indefinitely, if the probability of failure is high.

Typical MPP databases model a query plan as a directed acyclic graph [6, 10]. Each vertex in this figure corresponds to an operator, which is regarded as the unit of query processing. Typical operators include scan, sort, aggregation, join, etc. A query plan is produced by the master and broadcasted to all segments, which will execute the same plan on their own data.

Operators are usually executed in pipelines in MPP databases [10]. Pipelining can effectively reduce data materialization during query processing. Intermediate results will be immediately passed to the succeeding operator to process as soon as they are generated by the preceding operator.

In an MPP database, data need to be exchanged among slave nodes during query processing. Communication points break a query plan into multiple stages. On each segment, a query plan will be carried out by s concurrent processes or threads (further referred to as slices), each responsible for a stage of the query plan.

Figure 2 shows the actual query plan for the 17th query (Q17) of TPC-H benchmark [11] generated by Greenplum. The query is defined in Fig. 3. It is a typical query plan on a segment, with 15 operators and 4 slices. In this example, 4 processes or threads per segment will run in parallel to execute the query. Figure 4 shows the situation where Greenplum executes the query using 6 pipelines. It shows the processing time of each operator. For instance, in P1, intermediate results produced by Op.1 will be passed to Op.4 as soon as they are generated. Thus, we could see that the two operators overlap in executing time.

Injection of checkpoints into a query plan may have two-sided effects. Firstly, it may break the pipeline. Secondly, extra cost will be introduced for materialization. To minimize performance penalty, SIFT’s checkpoint candidates must satisfy the following conditions.For instance, query operators in Greenplum include scan, limit, sort, aggregation, motion, hashjoin, nestloopjoin, sortjoin and etc. Among them, motion is a special type of operator responsible for exchanging data. According to the above rules, three kinds of operators can be candidates of checkpoints. They are the hash operator within hashjoin, sort, and aggregation. They are all endpoints of pipelines. Besides, they are all common operators in query plans.

Upon a failure, the master will replace failed segments with their mirrors and restart queries from their latest checkpoints. Figure 5 explains the effects of intra-query fault tolerance. It uses the same query as Fig. 3. As shown in Fig. 3, the query plan consists of 4 slices and 6 pipelines. In the plan, Operators 4, 5, 10, 11, 13, and 15 are checkpoint candidates. We assume that all 6 checkpoints are activated. Figure 5 shows a situation where a failure occurs after finishing checkpointing Op.4 and Op.5. During rerunning, Operators 1–5 can be skipped. The query execution can start from intermediate results of Op.4 and Op.5. As a result, Slice 2 does not need to be launched at all. Meanwhile, the work of Slice 1 and Slice 3 will be reduced significantly and the query execution will be accelerated. From this figure, we could see the time saved by intra-query fault tolerance directly. As another example, if a failure occurs after the checkpoint on Op.13, only Slice 4 will be executed during the rerun.

In a query plan, many operators can be candidates of checkpoints. However, it is not necessary to materialize them all, which may introduce unnecessary overhead to query processing. As long as a desired success rate can be achieved, checkpoints should be as fewer as possible. Thus, we need to select an optimal set of checkpoints that incur the minimum materialization cost.

Databases usually estimate the cost of a query plan during query optimization. Based on their cost estimation, we can estimate overhead of checkpointing as well as success rate. This will lead us to an optimal checkpointing plan. For instance, the query plan in Fig. 3 offers 6 checkpoint candidates, i.e.,{Op.4, Op.5, Op.10, Op.11, Op.13, Op.15}. A possible checkpointing plan is to activate any subsets of them, e.g., {Op.4, Op.11, Op.13, Op.15}. In total, there will be \({2^6}\) combinations or \({2^6}\) candidate checkpointing plans. For each candidate plan, we can use Monte Carlo method to estimate its success rate. Since the search space is exponential and Monte Carlo method is time-consuming, we employ some pruning methods to accelerate searching.

In general, SIFT’s checkpointing plan optimization consists of 3 steps:

The implementation on the master node includes the following four parts.

The implementation on the slave side mainly aims to enable checkpointing. That includes materializing their intermediate results on mirrors and reusing them after failures. Table 2 shows that the implementation of the checkpointing module constitute a large part of the changes on the whole system. The module of checkpointing was implemented on operators of hash, sort, and aggregation.

In the experiments, we used the TPC-H benchmark [11]. Its schema includes 8 tables. To show robustness of our strategy, we used two computer clusters of different scales, whose machine configurations are summarized in Table 3.In the two clusters, each slave hosts two primary and two mirror segments. Data of each table is partitioned horizontally into segments and distributed to all slave nodes. Among the 22 queries of the TPC-H benchmark, we chose 6 typical queries to carry out our experiments. They include the 2nd, 7th, 9th, 16th, 17th, and 21st queries(Q2, Q7, Q9, Q16, Q17, Q21).

Our experiments evaluated four systems:

Each of our experiments was performed for at least 20 times. Figure 10 shows the average success rates on the first three systems. Average success rates with 90% confidence intervals are shown in Fig. 10. And we can see from the figure, results are stable. For all the experiments, we set MTBF as 48h and time constraint T as 1.6 times of \({t_0}\)(\(1.6t_0\)). We ignored MTTR in the experiments, as it is much shorter than execution time of long queries. We can see that SIFT can promote the success rates significantly, especially for Q17 and Q21. These two queries’ original success rates are relatively low, which makes the effects of fault tolerance more significant. They also offer more candidates for checkpointing, which enables us to find better checkpointing plans. On both clusters, we obtained similar results.

For some queries, such as Q16, the improvements made by SIFT are less visible. The execution times of the queries are shorter. Thus, their original success rates are already high and near to 100%, and little space is left for improvements. If we assume that at least one failure will occur during the query execution, the conditional probability will become the ones in Fig. 11. In this case, the effect of intra-query fault tolerance is amplified. For instance, SIFT can raise the success rate of Q21 from 24.4% to 76.6% on cluster A, and from 50.1% to 97.6% on cluster B. Intra-query fault tolerance will be more effective if the clusters are less stable.

We chose 4 queries, Q7, Q9, Q17, and Q21, to carry out further experiments, in which we manually shut down a slave node to see the impact of failures. Two rounds of experiments were performed for each query. In each round, one failure occurred, but at a different stage. Figure 12 shows results on cluster A, and Fig. 13 shows results on cluster B. In the figures, \({qi_j}\) represents the jth round of test of the ith query; bt represents the time span between the start of the query and the failure; ct represents the time required to rerun a query when fault tolerance is enabled; \({et_c}\) presents the time required to rerun a query when fault tolerance is disabled, i.e., the time needed for rerun a query with one failed node. et represents the time required to execute a query without failure. Usually, \({et_c}\) is larger than et as there is one less node usable for executing the query. In the figure, the first bar represents the time a query required without failure. The second bar represents the time a query required if a failure occurs and fault tolerance is enabled. And the third bar represents the time a query required if a failure occurs and fault tolerance is disabled.

As Figs. 12 and 13 shows, \({bt + ct}\) is significantly smaller than \({bt + et_c}\). The former is the query time at the presence of one failure for SIFT. The latter is that value for the system without intra-query fault tolerance. This indicates that SIFT can effectively shorten the latency of query re-execution. Besides, we can see that ct drops as bt increases. This indicates that less work is needed to rerun a query when the failure is near to end of the query. This complies with our analysis in Sect. 1. That is, the later the failure occurs, the more work can be checkpointed. As the original system will waste what have already done before failures, the later the failure is, the more time it will waste. Thus, compared to original system, our approach could reduce the wasted time significantly while the failure is near to end of the query. As we can see, \({bt + ct}\) can be bigger than et. On the one hand, the work after the latest checkpoint is still wasted. On the other hand, after recovering, less slaves, i.e., less resources, are employed to execute the query, making the query processing slower. Figure 14 shows the ratio of time saved by intra-query fault tolerance, i.e., \({\frac{et_c-ct}{bt + et_c - et}}\). It represents benefits of using SIFT. As we can see, in most cases the ratio is more than 60%.

Overhead of checkpointing is inevitable. We tested all the four systems in Sect. 5.1 on cluster A, and the first two on cluster B, and recorded their query times. For the system, simulated-XDB, we measured the runtime when it attempted to achieve similar success rates as ours. Each of our experiments was performed for at least 10 times. We treat query time on original Greenplum system as 1.0, and normalize the query times of other systems based on it. This helps us see the differences. The average normalized execution time with 90% confidence intervals is shown in Fig. 15. As we can see, for most of the queries, the overheads of SIFT is visible. However, it is low most of the time. On average, the overhead introduced by SIFT is about 10% of the total execution time on cluster A and 8% on cluster B. As we mentioned before, intermediate results need to be materialized locally and remotely. And it is the main part of the overhead. In other words, the size of intermediate results impact the overhead significantly. This explains the higher overheads for some queries. In the architecture of Greenplum, mirrors and primaries reside on the same physical machines (Sect. 2.1). Thus, backups on mirrors will interfere with the original query execution and introduce significant overhead. However, this overhead could be easily reduced if we backup intermediate results somewhere else.

The results also indicate that it is not necessary to activate all candidate checkpoints, as it does not always benefits performance or success rates. In other words, optimized checkpoint selection can both reduce the overhead and improve the success rate. Table 4 shows the total number of candidate checkpoints and the number of checkpoints selected by SIFT. In our tests, as we regard success rate as our goal, materializing all candidates is usually not an ideal strategy. For instance, for Q2, if we do not perform checkpoint selection, the success rate is 94.86%. If we perform checkpoint selection, the success rate can rise to 99.08%. When the execution time is shortened, the probability of run-time failure can be naturally reduced (Sect. 2.3.2).

The experiment results on the simulated-XDB show that not all operators are good candidates for checkpointing. If a checkpoint breaks the query processing pipeline, it usually introduces a significant overhead. In this case, the system needs to materialize much more intermediate results, making the query time much longer. As we can see, SIFT remarkably outperforms XDB in quite a number of queries, as it never intends to break pipelines.

In this set of experiments, we evaluated how different parameters affect the performance of SIFT. Firstly, we consider the influence of the time constraint T. Figure 16 shows the success rates when T is set as \({T = 1.3t_0}\), \({T = 1.4t_0}\), \({T = 1.5t_0}\), \({T = 1.6t_0}\) and \({T = 1.8t_0}\). MTBF was still set as 48h. As expected, the larger the T, the higher the success rate. And for all of them, there is an increase on success rate compared with the original system. And Fig. 17 shows the conditional probability under the condition of at least one failure with different given time. For the queries Q2, Q9, Q17 and Q21, the success rates of system applying SIFT under the condition of \({T = 1.3t_0}\) are even higher than those of original Greenplum under the condition of \({T = 1.6t_0}\).

Another set of experiments were conducted to study the influence of MTBF, in which T was set to \({1.6t_0}\). The results can be found in Fig. 18. We can see that the drop of MTBF will increase the probability of failure and thus magnify the fault tolerance effect of SIFT. In other words, the benefits of SIFT will be more significant if MTBF is smaller. On the other hand, MTBF will drop as the scale of the cluster increases. Thus, SIFT is supposed to be more effective on larger clusters. Figure 19 shows how the conditional probability of at least one failure varies with MTBF . We can see that, on original system, the conditional probability will converge to 60%, as T is set to \({1.6t_0}\). For the system implementing SIFT, that value can be significantly improved. For instance, on the queries Q2, Q17, and Q21, it could approach 100%.

The results of Q21 on the two clusters are shown in Figs. 20 and 21 respectively. In the figures, \(S\_SR\_C\) means the success rate on the cluster C for the system S, and \(S\_CP\_C\) means the conditional probability with at least one failure on the cluster C for the system S. These results lead us to the same conclusion.

Although I/O operations are often seen as a performance bottleneck, CPU operations cannot be ignored. Compression can effectively reduce I/O operations. However, it comes with a price of extra CPU cost. In order to understand the influence of compression, we designed a simple query plan shown in Fig. 22. It comprises two hashjoin operators, Op.3 and Op.8, which we chose as checkpoints. On each segment, the average sizes of hash tables of Op.3 and Op.8 are 194MB and 757.5MB respectively. To speedup the checkpointing process, we perform compression on the hash tables. The compression ratios are 2.8 and 3.48 respectively. In other words, the data size after compression is about \({30\%}\) of the original.

Figure 23 shows the time consumption of different stages of query processing. \({tc_i}\) represents time required by compression in Slice i. \({td_i}\) represents time required by decompression in Slice i. \({IO_i}\) represents time for persisting the intermediate results before compression. \({IO^{comp}_{i}}\) represents persistence time after compression. In the figure, the first bar measures the total time for materializing intermediate results with compression for Slice2, the second are measures that without compression. The other two bars represent those values of Slice3. We can see that compression is useful. Its effect is especially significant for Slice 2, in which a half of the time can be saved. Figure 24 shows the total execution time of the query in the four tests. It shows that compression can speedup query processing.

Traditional relational databases [16‐18] implement fault tolerance mainly through replication [19] and checkpointing at storage level. They usually do not take intra-query fault tolerance into consideration as queries on them usually could finish in seconds or minutes most of the time. There is little need for intra-query fault tolerance.

Well-known MPP DB products include Teradata [1], Greenplum [2], Vertica [3], as well as some SQL-on-Hadoop systems such as Impala [4] and HAWQ [5]. They all support fault tolerance at the storage level as the traditional databases do. To the best of our knowledge, none of them support intra-query fault tolerance.

The known work on intra-query fault tolerance includes FTOpt [10] and XDB [6]. XDB supports cost-based fault tolerance by choosing operators in a query plan to materialize. Its goal is to find a materialization configuration that would lead to minimum expected running time. However, XDB has little consideration about pipelines of query processing. It is prone to break pipelines of the original query and impair its performance. Besides, XDB is more complex to implement, as it needs to make every operators possible to materialize, which leads to re-implementation of all the query operators. We have shown in our experimental evaluation, SIFT tends to be more efficient as it always tries to preserve the query processing pipelines.

FTOpt considers pipelines and presents an extensible and heterogeneous fault tolerance framework. It also considers the differences between operators and applies different strategies to handle them. However, FTOpt assumes that data transmission over the network among operators is always order preserving. This makes FTOpt impractical for most MPP databases, which normally do not guarantee order preserving data transmission.

Neither XDB nor FTOpt was implemented in real MPP databases. In contrast, the design of SIFT emphasizes practicality. Our goal is a light weighted fault tolerance mechanism that can be applied to most real MPP databases. Besides, SIFT aims to ensure success rate, so that it adopts a unique approach to optimize checkpointing plans.

Since Google published the work of MapReduce [20] and GFS [21], open-source data processing platforms such as Hadoop [7] have become popular. Multiple stacks of systems have been built on top of them. Some SQL-On-Hadoop implement intra-query fault tolerance by utilizing the fault tolerance function of Hadoop [7]. However they usually suffer from too much overhead. The work of [22] implemented pipeline within the Hadoop MapReduce framework. Experiments showed that its overhead of materialization is high. Osprey [23] applies the thought of MapReduce to MPP systems. It splits a analytical query over a star schema into short sub-queries. Its fault tolerance builds upon the data replication scheme of chained declustering [24]. However, it does not consider query success rate, making it difficult to achieve a good trade-off between fault tolerance and performance.

Microsoft has introduced a general-purpose, distributed execution engine, Dryad [25]. It executes queries over what is a communication graph. It allows data between operators to either be pipelined or materialized. However, it requires manual configuration and pushes the work of optimization to the user of the system. Bubble Execution [26] is a new query processing framework for interactive workloads at cloud scale introduced by Microsoft. It balances cost-based query optimization, fault tolerance, optimal resource management, and execution orchestration. It divides a query execution graph into a collection of query sub-graphs (bubbles), which is unit of scheduling and fault tolerance. Inside a bubble, tasks are connected via pipe channels. And between bubbles, tasks are connected via recoverable channels. While for SIFT, the unit of fault tolerance is divided by checkpoint operators, which are breaking operators.

Many stream processing systems implement fault tolerance [27‐33] by redundant processing, checkpointing, and remote logging. They often allow weaker recovery guarantees in exchange for improved performance, but it is not allowed by databases. Some systems, such as Flink, provide exactly-once semantics. However, as a stream processing systems, they usually don’t allow the existence of breaking operators, such as hashjoin and sort. The system proposed in [30] aims at find a materialization configuration with minimized overhead. But it considers only one failure.

In summary, most big data platforms, such as Hadoop, Spark and Flink, provide fault tolerance support at the data processing level. OLAP engines over these platforms can take advantage of it. While the existing OLAP engines still face performance issues, they can hopefully be improved as more optimization techniques are introduced. Nevertheless, SIFT is designed for traditional MPP databases rather than big data platforms.