Adaptive Continuous Query Reoptimization over Data Streams

A data stream is a series of massive unbounded tuples continuously generated at a rapid rate. Continuous queries for data streams should be processed continuously, so that a strict time constraint is required. In most previous research studies, in order to guarantee this constraint, the evaluation order of join predicates in a continuous query is optimized using a greedy strategy. However, because a greedy strategy traces only the first promising plan, it often finds a suboptimal plan. To reduce the possibility of producing a suboptimal plan, in this paper, we propose an improved scheme, k-Extended Greedy Algorithm (k-EGA), that simultaneously examines a set of promising plans and reoptimize an execution plan adaptively. The number of promising plans is flexibly controlled by a user-defined range variable. The scheme verifies the performance of the current plan periodically. If the plan is no longer efficient, a newly optimized plan is generated. The performance of the proposed scheme is verified through various experiments to identify its various characteristics.


Introduction
Many recent research studies of emerging applications often need to deal with a data stream, which is a massive unbounded sequence of data elements continuously generated at a rapid rate. Such applications can be web click monitoring, sensor data processing, network traffic analysis, telephone recording and multimedia data processing. In these applications, a continuous query is generally used to monitor the ongoing information on data streams. It is registered in advance and produces results continuously and repeatedly whenever new data elements of data streams are generated. Since new data elements can be rapidly generated and should be processed in real time, a continuous query should be evaluated as quickly as possible. For this, query optimization is essential. However, query optimization techniques used in a traditional database management system are not effective for data stream applications because of the unpredictable characteristics of data streams [1]. For this reason, many previous research studies [1], [4]- [6], [8], [9] are focused on devising a method of optimizing the execution of continuous queries. A multiway join query contains multiple join operations.
Finding the optimal execution sequence of join opera- tions in such a query is problematic since a join operation is very expensive. The execution plan of a multiway join query should be continuously reoptimized in data stream environment. For this purpose, properties such as the input rates of its operand data streams and the join selectivities of its join operations should be continuously monitored. However, such monitoring requires a huge runtime overhead. In [7], given a join operation, the domain sizes of its two join attributes are monitored to estimate its join selectivity. In [5], STREAM actually obtains the join selectivity of every join operation in every possible left-deep tree plan by additionally executing them for sampled tuples. Therefore, it can obtain more accurate statistics, but it must evaluate a number of plans in run time.
To estimate the current evaluation cost of a join operation, the past properties of the operation are used. Even though the properties of a join operation may be changed in runtime dynamically, they may not vary randomly. Instead, they tend to have valid ranges respectively. This means that the past properties of a join operation can be used to predict the current cost of the operation more accurately. For this purpose, the properties of join operations that are executed at least once are maintained in a table called cost Statistics.
In order to continuously reoptimize the query, the time to generate the optimized plan of a multiway join query is also an important factor. Consequently, most approaches employ a greedy optimization strategy with respect to either the size of the result [4] or join selectivity [5]. Even though the greedy strategy can generate its solution very quickly, it is apt to produce a suboptimal plan. This is because there is no backtracking mechanism to undo the intermediate order of join operations when the order turns out to be nonoptimal. However, employing a backtracking mechanism in data stream environments is also inadequate because of the time constraint as mentioned above.
In this paper, we propose a query optimization scheme called k-Extended Greedy Algorithm (k-EGA) to produce the optimal execution plan of a multiway join query. As mentioned above, the greedy strategy traces only one promising subplan, whereas k-EGA traces a set of promising subplans simultaneously. Intuitively the cardinality of the set affects the optimality of the algorithm as well as optimization time. As it increases, the possibility of generating a more optimal plan is increased, but the time to produce the plan is increased as well. To identify the effective set of promising subplans, given a user-defined range variable k, the cost of each subplan is represented as a range based on the aver-age cost and its standard deviation in the cost statistics table. Among the join operations of a multiway join query, the one with the smallest average cost is chosen first. If the range of the selected join operation overlaps with the remaining join operations, they are traced together. As k becomes larger, the range of each join operation becomes broader. This means that the number of traced subplans increases, thereby improving the optimality of the optimized plan of a query. However, the time to produce the optimized plan clearly increases as well. By flexibly controlling the number of simultaneously traced subplans, our proposed scheme provides a desirable trade-off between the optimality of an execution plan and the time to generate the plan. Depending on the current processing load, the range variable k can be flexibly adjusted.
The remainder of this paper is organized as follows: In Sect. 2, related work is presented. In Sect. 3, the join cost model is described on the basis of the cost statistics for a multiway join query. In Sect. 4, the proposed method that can find the optimal execution plan of a multiway join query is introduced. In Sect. 5, the performance of the proposed method is analyzed through a series of experiments. Finally, In Sect. 6, our conclusions are presented.

Related Work
There are two common techniques for processing a join operator. The first technique, involving a pipelined processing tree (PPT) [17], uses a multiway nested loop to generate the result of a multiway join query without producing any intermediate join result. The second technique uses a binary linear processing tree (BLPT) [17] that gives intermediate join results. A representative research on the first technique is a multi-stream join algorithm called MJoin [1]. It produces the result of a multiway join query without any tree-structured plan. In [4], a greedy-based optimization method is proposed to determine the evaluation order of each operand of an MJoin operator.
STREAM [5] uses the A-Greedy algorithm to adaptively determine the evaluation order of an MJoin operator to minimize the processing cost of a multiway join query whose join selectivities vary unpredictably over time [14], [15]. The ongoing change in the join selectivity of every possible execution order is continuously monitored to adaptively change the evaluation order of each operand in an MJoin operator. In [18], A-Caching focuses specifically on the problem of the adaptive placement and removal of caches in an MJoin operator to optimize join performance. It provides algorithms for selecting caches, monitoring their cost and benefits in current conditions, allocating memory to caches, and adapting as conditions change.
An eddy [11] accomplishes very fine-grained adaptivity without employing any fixed execution plan. Its query execution model routes input and intermediate result tuples among query operators by making independent routing decisions for each of the tuples to reflect the dynamically changing runtime selectivity of each query condition. Based on the eddy query processing framework, a continuously adaptive, continuous query (CACQ) [8] is implemented. It provides a grouped-filter index to simultaneously evaluate multiple selection predicates and a space-efficient generalization of double-pipelined hash joins called SteMs [16]. The execution order of multiple join operations in a multiway join query is dynamically determined by a ticket-based routing policy that executes a lottery between query operators. The winning chances of a particular operator are proportional to the number of tickets it owns. The operator with a higher selectivity gets more tickets, which means the operator will receive more tuples early through the eddy [11].
In [7], as a modified architecture of the original eddy that uses SteMs, a query operator called a STAIR (Storage, Transformation and Access for Intermediate Results) is introduced to perform a join operation. A STAIR holds the runtime state of join operations. It employs the second technique, BLPT. Even though it solves the drawbacks of the original eddy, they address the problems caused by allowing the query state to be modified and migrated across STAIRs during query execution. STAIRs also solve the problem in which the ability of an eddy to adapt is constrained by the state that accumulates in the query operators. On the basis of the cumulative selectivity of every subplan of join operations, a greedy optimization strategy is used to determine the execution plan of STAIRs dynamically.

Cost Statistics
To define the cost of a binary join operation, we use a unittime-based cost model [13] that defines the evaluation cost of a binary join operation to be the processing cost of all incoming tuples in a unit time. In the execution of a binary linear processing tree, the result of one binary join operation should be materialized because it can be one of the operand for the next binary join operation. Consequently, the result size of a binary join operation affects the overall evaluation cost of a multiway join query, so that the unit-time-based cost model is modified for a multiway join query in this paper. The terms used in the modified cost model are described in Table 1. A join operation R S should be executed when a new tuple arrives from any of its two operand streams R and S . Accordingly, the evaluation cost C i (R S ) of a binary join operation R S for newly arriving tuples in unit time should be symmetrically defined as follows.
The generating cost of the result of a binary join operation, C s (R S ) is proportional to the size of the result and is defined as.
Therefore the total evaluation cost of a binary join operation on data streams R and S becomes A multiway join query can be evaluated using a number of different execution plans. All combinations of the join operations that can appear in the plans are maintained in cost statistics. A row in cost statistics is formally defined as follows.
) and σ(C(J q m )) denote the average cost of the J q m and its standard deviation. The term count(J q m ) denotes how many times the join operation J q has been executed before. If count(J q m ) = 0, the join operation J q m has not been executed before, so that its current statistics are estimated.
When a join operation is not executed, the cost of the join operation is estimated using Eq. (3), and the estimated selectivity is reflected to its statistics in the CS table. When the join operation is part of the optimized execution plan, its real selectivity is obtained while executing the plan and is reflected onto the CS table. If this is the first real selectivity of the join operation, its past estimated statistics are deleted and its statistics are replaced with the real selectivity. In addition, its count field is set to 1. The statistics of a join operation in the CS table represent how the cost of the join operation fluctuates. If the cost of a join operation fluctuated widely in the past, its standard deviation becomes large. This is a simple three-join operation between the streams R, S , T and U. Table 2 shows the cost statistics table of the query Q when the execution plan (R S ) (T U), has been executed three times and the other plan (((S T ) R) U) has been executed five times before. The values of μ and σ are merely example values.

Subplan and Set of Traced Subplans (STSs)
A complete execution plan for a multiway join query with n join operations is a binary linear processing tree that determines the execution order of n join operations. The objective of query optimization is to find the execution order with the smallest cost. The query can be evaluated using a number of different execution plans. To denote part of an execution plan, the term m-subplan is defined formally in Definition 2. The evaluation cost of each subplan is defined by the accumulated cost of all the join operations in the subplan. Since the cost of an individual join operation is maintained by the average cost and its standard deviation in the CS table, the cost of the subplan is also represented in the same manner as follows:

Definition 3. (Cost of m-subplan)
Given a multiway join query q with n join operations J q = {J 1 , . . . J n }, its child plan sp m+1 is sp m ∪ {J e }, whose cost is obtained as follows: i) The average cost of sp m+1 , μ(sp m+1 ) = μ(sp m ) + μ(J e ). ii) Suppose that sp m and J e are independent, the standard deviation of sp m+1 , denoted by σ(sp m+1 ), is σ(sp m ) 2 + σ(J e ) 2 + 2cov(sp m , J e ). cov(sp m , J e ) is a covariance, which is a measure of how much two variables, σ(sp m ) and σ(J e ), change together. Since sp m−1 and J e are independent, cov(sp m , J e ) = 0, so that The effective range of an m-subplan is defined by a range variable k in Definition 4. The minimum value of the range, i.e., μ(sp m ) − k*σ(sp m ), is denoted byĈ min (sp m ) and the maximum value of the range, i.e., μ(sp m ) + kxσ(sp m ), isĈ max (sp m ).
As mentioned above, k-EGA simultaneously traces a set of promising subplans, which is defined by a set of traced subplans (STS) as follows.
The standard deviation of the cost of a subplan represents how much the cost of the subplan fluctuates. When the value is small, the cost is almost the same in the past, which indicates the cost can accurately represent the current cost of the subplan. In contrast, when the value is large, the cost estimation based on the average cost is not accurate enough. Our scheme traces not only the subplan with the smallest average cost, but also those subplans whose effective ranges overlap with the subplan together. The subplan whose cost cannot be estimated exactly is most likely to be traced because the effective ranges of the subplan are wide. Consequently, the subplan is examined more closely. This is because one of the subplans may lead to a better plan.
Intuitively, the range variable k affects the cardinality of its STS m . The greater the value is, the wider the effective range of each subplan becomes. Consequently, the size of STS m increases as well. Furthermore, it means that more promising subplans are traced simultaneously to find out the final optimized execution plan. Therefore, the optimality of the final plan would be better than the one obtained by the general greedy method. On the other hand, the time to generate the final plan may be increased. The proposed method provides a good trade-off between the optimality of the final plan and the time to generate the final plan since they can be flexibly controlled by the range variable k. Figure 1 illustrates how a range variable k plays a role in controlling the cardinality of STS. Suppose that there are four m-subplans {sp m 1 , sp m 2 , sp m 3 , sp m 4 } and that they are the candidate m-subplans for a multiway join query with n join operations (m ≤ n). To determine the STS m of the query, the subplan with the smallest average cost is chosen first as in the greedy optimization strategy. Consequently, the subplan sp m 3 is selected first. The effective range of each of these four subplans is calculated. Suppose that the range variable k is set to 1, then the effective range of sp m 3 be-

k-Extended Greedy Algorithm
The overall steps of k-EGA are very similar to those in the greedy optimization strategy that determines the execution order of one of unordered join operations in each optimization step. Likewise, given a multiway join query q with n join operations J q = {J 1 , . . . J n }, k-EGA determines STS 1 in the first step and then generates its child subplans STS 2 . This procedure is continued until STS n is found. Among the n-subplans in STS n , that with the minimum evaluation cost sp n min (sp n min ∈ STS n ) is the final optimized plan. Given STS m = {sp m 1 , sp m 2 , .....sp m k }, STS m+1 is generated. For each m-subplan sp m i (1 ≤ i ≤ k), one additional join operation J which is not part of sp m i is added to sp m i to generate a candidate (m + 1)-subplan. Although there are (n − m) different (m + 1)-subplans for each sp m i , only those (m + 1)-subplans that satisfy Definition 5 are inserted into STS m+1 . In order to determine sp m+1 min , all the subplans generated by STS m should be scanned after they are generated. However, maintaining all the child subplans of STS m is a runtime overhead. sp m+1 min can be incrementally identified by tracing each of m-subplans in STS m . When sp m min is changed, the (m + 1)subplans already inserted into STS m+1 are re-examined. The pseudo-code of these steps is shown in Fig. 3. The following example shows how k-EGA finds the optimal plan of a multiway join query.
Example 2) Suppose that there are the SQL-like query Q in example 1 and the cost statistics in Table 2. Let k be 1.  erated by sp 1 1 . Likewise sp 2 3 and sp 2 4 are generated by sp 1 2 . The effective ranges of the subplans are presented in Fig. 2 (b). As is shown in Fig. 2 (b) is {J 1 , J 4 , J 8 } and sp 3 2 is {J 2 , J 6 , J 9 }. μ(sp 3 1 ) is lower than μ(sp 3 2 ), so that the final query plan becomes sp 3 The standard deviation of a subplan presents how much the subplan fluctuates. The small standard deviation means there is little change in the cost of the subplan. On the other hand, the large standard deviation means the cost of the subplan fluctuated widely in the past, and it is difficult to determine the current cost of the plan. Also, if any subplans fluctuate widely, the following subplan is also likely to be done because the result of the parent can be the input of the child. Therefore, k-EGA traces these subplans more deeply and intensively and this can cause a more optimal plan. (By Definitions 4 and 5, the subplans which have a large standard deviation are likely to be included in promising subplans.) As mentioned above, the number of simultaneously traced subplans is flexibly controlled. Actually, the capability of the k-EGA is from the general greedy optimization to exhausted optimization. If the range variable k is set to 0, only one subplan is traced since the minimum and maximum values of effective ranges of all subplans become the same by Definition 4. Consequently, it produces the same plan as the general greedy optimization. In contrast, if the range variable k is sufficiently large, so that all the m-subplans can be included in STS m , its resulting plan is the completely optimized plan.

Experiments
In this section, we provide the experimental results of the proposed algorithm. All the algorithms are implemented in C, and all the experiments are executed on a Pentium 4 CPU 2.66 GHz system with 1 GB RAM. The system runs Linux with a 2.4.5 kernel and gcc 3.3.2. To validate the cost model proposed in this paper, a multiway join query with 2 join operations, A B and B C, is experimented on for a synthetic dataset containing tuples of integer attributes. The domain size of each join attribute is 100. The value of each attribute ranges over [0..99] and is randomly selected. The input rates of the operand data streams A, B and C are 10 (tuples/s), 20 (tuples/s) and 50 (tuples/s), respectively. The cost of the query is compared with the actual execution time of the query by varying either the input rate of the operand A and the domain size of the join attribute of A B. Figure 4 shows that the cost is closely proportional to the actual execution time.
To analyze the performance of the k-EGA, a synthetic dataset is generated. For a multiway join query, the sizes of a sliding window and a hash table are fixed. The rates of the operand data streams are uniformly chosen from 10 tuples/s to 1000 tuples/s and the domain size of the attribute of each join operation is also randomly varied from 1 to 1000. The value of each join attribute is randomly selected from 1 to the domain size of the join attribute. Figure 5 shows the effects of a range variable k as the number of operand data streams for a multiway join query is increased. If the range variable k is set to 0, k-EGA is the same as the routing policy for a STAIR [7], which routes an incoming tuple to the join operation with the lowest selectivity. Intuitively, the execution time should increases as the number of join operations in a query increases. However, as k is increased, the execution time decreases since the final plan produced by the k-EGA becomes more efficient. In Figs. 6 and 7, a multiway join query with six join operations is experimented on for the nine synthetic datasets in Table 3 to compare the performance of k-EGA. The input rate and domain size of an operand data stream are randomly selected within the predefined range. The join selectivity of a join operation is determined by the reciprocal of the domain size of the attribute in each join predicate. In Table 3, the term a±b means that a value is randomly generated from the range [a − b, a + b]. For example, the input rates of the cost in DS 5 are randomly generated from the range [50,150] and the domain size of each join attribute is randomly generated from the range [500, 1500].
Given the k for a multiway join query, let Cost k denote the cost of the execution plan generated by k-EGA and Cost opt denote the cost of the optimal plan. To observe the performance of k-EGA with respect to the range variable k, we define the relative effectiveness E(k) of k-EGA. It is defined as.
The smaller the E(k) is, the closer to the optimal plan the execution plan generated by k-EGA is. In Fig. 6, E(k) decreases as k increases even if the optimality of k-EGA decreases as the range of the input rates becomes wider. As shown in Fig. 7, E(k) decreases as k logarithmically increases even if the optimality of our algorithm decreases as the range of the domain size increases.
To reoptimize the current execution plan continuously, it is important to monitor its cost periodically. If the plan is no longer efficient, k-EGA generates a newly optimized plan. Given the current plan p i , let Cost lastest (p i ) denote the cost of p i when the plan was originally generated. On the other hand, let Cost cur (p i ) denote the cost of p i at present. A newly optimized plan should be generated by k-EGA when the following equation is satisfied.
A cost allowable ratio δ is a user-defined parameter that denotes the allowable ratio of the current cost relatively to the latest optimized cost. As δ is set to be smaller, k-EGA changes its optimized plan more sensitively. Figure 8 shows how k-EGA adaptively reoptimizes its     execution plan for a seven-way join query. To simulate the dynamic change in the characteristics of input data streams, the tuples of the three datasets D 1 , D 2 and D 3 in Table 4   Table 4 Sub-datasets for Fig. 8.
are processed in sequence. The y-axis in Fig. 8 shows the cost per 1000 input tuples. The term "Optimal for D 1 " denotes the static optimal plan for the dataset D 1 only. When the tuples in D 2 start being processed, the cost of the optimal plan for D 1 sharply increases since the current evaluation sequence is no longer optimal to D 2 . However, k-EGA (k = 1) reoptimizes to generate a newly adjusted plan, and its execution plan cost remains to be close to the cost of executing the optimal plan. Figure 9 shows the trade-off between the optimality of the final plan generated by k-EGA and the time to generate the final plan. When k < 2, the cost of the execution plan optimized by k-EGA markedly decreases while the optimization time almost linearly increases. However, when k is large, the time to generate the final plan exponentially increases while the cost of the plan almost does not decrease. This means that when k becomes larger than a specific threshold (in this experiment, it is 2), the overhead of tracing multiple subplans simultaneously becomes greater than the profits gain obtained by finding a better plan.

Conclusions
In this paper, to optimize a multiway join query in a data stream environment, the K-Extended Greedy Algorithm (k-EGA) is proposed. The proposed algorithm traces a set of promising subplans simultaneously to reduce the possibility of producing a suboptimal plan. We provide a method of controlling the number of simultaneously traced subplans flexibly with respect to the range variable k. As a result, it is possible to adaptively control the number of subplans by changing the range variable k depending on the current processing load. Experiment results show how efficiently k-EGA can optimize a multiway join query in various stream environments. k-EGA can provide a good trade-off between the optimality of a generated execution plan and its optimization time. In addition, the plan generated by an MJoin operator cannot be utilized in a multiple query operation since the join orders of all operand streams are different. However, it is easy to extract common join operations in a number of plans generated by k-EGA. Most of the continuous queries are registered in advanced and should be processed continuously, so multiple query optimization is very inportant in a data stream environment.