Towards Fast Theta-join: A Prefiltering and Amalgamated Partitioning Approach

As one of the most useful online processing techniques, the theta-join operation has been utilized by many applications to fully excavate the relationships between data streams in various scenarios. As such, constant research efforts have been put to optimize its performance in the distributed environment, which is typically characterized by reducing the number of Cartesian products as much as possible. In this article, we design and implement a novel fast theta-join algorithm, called Prefap, by developing two distinct techniques - prefiltering and amalgamated partitioning-based on the state-of-the-art FastThetaJoin algorithm to optimize the efficiency of the theta-join operation. Firstly, we develop a prefiltering strategy before data streams are partitioned to reduce the amount of data to be involved and benefit a more fine-grained partitioning. Secondly, to avoid the data streams being partitioned in a coarse-grained isolated manner and improve the quality of the partition-level filtering, we introduce an amalgamated partitioning mechanism that can amalgamate the partitioning boundaries of two data streams to assist a fine-grained partitioning. With the integration of these two techniques into the existing FastThetaJoin algorithm, we design and implement a new framework to achieve a decreased number of Cartesian products and a higher theta-join efficiency. By comparing with existing algorithms, FastThetaJoin in particular, we evaluate the performance of Prefap on both synthetic and real data streams from two-way to multiway theta-join to demonstrate its superiority.


FIGURE 1
An illustrative example of a theta-join predicate  ⋈  (.B > .B), its input and its corresponding output. In this example, there are only 4 phone-laptop combinations that satisfy the price of the phone is more expensive than laptop.
To relate data streams together, the join operation 16,17 is one of the vital operations that is capable to detect scenarios that satisfy certain conditions. To find out data elements in two data streams that are equal, equi-join 18,19,20 should be used. As for non-equal relationships between data streams, the theta-join 21,22 comes to help. Theta-join, denoted as follows serves as a special kind of join operation that relates the attribute A of the data stream  and the attribute B of the data stream  using a non-equal theta condition among one of the following {<, ≤, >, ≥}. Cartesian products 23,24 between two data streams will be generated and data element pairs that satisfy the theta condition are picked out to form the theta-join results. For instance, as illustrated in Figure 1, the inputs of this two-way theta-join operation are data stream Phone (Stream ) and data stream Laptop (Stream ). Under the join predicate  ⋈  (.B > .B), all phone-laptop combinations that satisfy the price of the phone is higher than the price of the laptop will be returned as results. Therefore, 4 satisfied combinations are returned as output as shown in Figure 1. Moreover, the theta-join can also be generalised to work on multiple data streams. For instance, the 3-way theta-join predicate is expressed as follows which exploits all A, B, C attribute combinations that satisfy the A 1 B 2 C condition from three data streams ,  and  , respectively. When dealing with data streams that come in an online manner, the theta-join algorithm will receive the data stream in the form of a window, with the window size denotes the amount of data consisted in the window. Then, the theta-join operation will be performed between windows of data streams. As a powerful data analysis and processing tool 25 , theta-join can be widely utilised in broad applications, such as discerning on which day stock  performs worse than stock  in 2019, or whether the wind speed in July is faster than June 2020 in most of the days. The theta-join operation has also been implemented in popular database management systems such as Oracle database 26 , PostgreSql database 27 , etc. However, the theta-join's efficiency is heavily affected by the number of involved time-consuming Cartesian products 17,28 . Handling the theta-join in a careless way can result in the number of Cartesian products grows drastically, and will even lead to an exponential surge if multi-way theta-join is conducted. The unacceptably large number of Cartesian products also become a curse, especially in distributed environments 29,30,31 where the incurred I/O overhead and the communication cost 25 make it an influential factor that dramatically compromises the efficiency of the theta-join operation. Therefore, they should be reduced as much as possible. As a result, numerous efforts have been made by both industry and academia from different aspects to optimise the efficiency of the theta-join operation 32,33,34,35 .
Although the existing algorithms in the literature exhibited different merits in making the theta-join operation more efficient, they still suffered from some deficiencies in effective Cartesian product reduction and data stream partitioning, which more or less compromised the theta-join operation efficacy. Therefore, we still have rooms to further improve them in various ways. In this paper, we propose a novel fast online theta-join algorithm, called Prefap. Instead of performing filtering only after partitioning as in 32,33,34,35 , it makes sense to perform a pre-filtering based on the theta condition before the partitioning takes place as it can not only reduce the amount of data involved in the partitioning but also make the partitioning more fine-grained. Since our partitioning is based on the range boundaries calculated using the minimum and the maximum values of the data streams, the pre-filtering of the data streams is possible to condense the range between the minimum and the maximum value, hence making the partition more fine-grained. Moreover, this pre-filtering mechanism does not suffer from the severe overhead caused by the time-consuming sorting operation 36,37 . As opposed to 32,33,34 , the proposed algorithm requires neither inter-partition nor intra-partition ordering.
Furthermore, for all the aforementioned algorithms, the partitioning of two data streams are conducted in a coarse-grained isolated manner, i.e., the partitioning of data stream  has no impact on the partitioning of data stream , which may impair the effectiveness of the filtering mechanism after the partitioning. To overcome the drawback incurred by the isolated coarsegrained partitioning, the proposed algorithm introduces an amalgamated partitioning mechanism. With this mechanism, the data stream will be partitioned based on the amalgamated partitioning boundary that fuses the partitioning information of data streams. Hence, worthless Cartesian products between partitions that are deemed impossible to possess valid theta-join results will be avoided, and hence making the algorithm more efficient, which is an improvement to 32,33,34,35 .
To validate the effectiveness of the proposed method and to make it applicable, we integrate the pre-filtering strategy and the amalgamated partitioning mechanism to form a new theta-join processing framework. The framework uses FastThetaJoin 35 as a basis. However, the proposed Prefap framework makes contributions by performing pre-filtering and avoiding the isolated partitioning. With the Prefap framework, we can substantially boost the efficiency of the theta-join operation. The proposed Prefap framework is comprehensively evaluated on both synthetic and real data streams in distributed environments and compared with other algorithms, demonstrating its superior performance.
Therefore, in summary, the paper makes the following contributions: • We propose a pre-filtering strategy that can not only reduce the amount of data involved in the partitioning but also make the partitioning more fine-grained.
• We introduce the amalgamated partitioning mechanism that avoids the coarse-grained isolated partitioning and hence benefits the reduction of Cartesian products and makes the algorithm more efficient.
• We unify the proposed pre-filtering strategy and the amalgamated partitioning mechanism to form a holistic framework, called Prefap, that can boost the effectiveness of theta-join operations while avoiding the time-consuming burdens such as sorting.
• The Prefap framework is implemented and comprehensive empirical evaluations on both synthetic and real data streams are conducted to testify the superiority.
The remainders of the paper are organised as follows. Section 2 overviews some related works to show the distinct features of the proposed algorithm. The Prefap algorithm is introduced in Section 3. Section 4 presents the experimental results, which are also comprehensively analysed and discussed. The last section concludes the paper.

RELATED WORK AND RESEARCH OPPORTUNITY
As the scale of data streams keeps growing rapidly 7,8,9 , how to efficiently utilise the theta-join to process data streams becomes a crucial problem, and thereby attracting numerous attentions from both industry and academic communities. In this section, we first overview some past research efforts on improving the efficiency of the theta-join operation, and then identify their deficiencies to show our research opportunities.

FIGURE 2
Illustration of the number of Cartesian products involved in each algorithm. The theta condition in this example is  > . The data stream  is placed vertically and the data stream  is placed horizontally. The thick coloured border indicates the boundary of Cartesian products that need to be performed, while the thin coloured border indicates the partitioning boundary.

FIGURE 3
The bar chart presents the number of Cartesian products performed by each algorithm in the example in Figure 2. The rightmost red bar, as well as the red dotted line, indicate the number of theta-join results, i.e., no matter how the theta-join algorithm is optimised, this is the lower bound of the number of Cartesian products that the algorithm needs to perform and the algorithm cannot perform better than this.

Range-based Method
Dewitt et al., 32 proposed a Range-based Method (RBM) to carry out the theta-join operation in distributed environments. The joint attributes in data streams were firstly sorted, then two data streams were partitioned where the partitioning boundaries were characterised by sampled ranges, and Cartesian products were performed without any filtering being performed.
For example, in Figure 2(a), if we set the number of partitions to 3, the Range-based Method will sample the 4 th and the 7 th value from the sorted data stream , which are 3 and 6, respectively. Then the data stream  is partitioned based on the ranges as follows: (−∞,3), [3,6) and [6,+∞). The same partitioning is applied for data stream  as well.
However, the Range-based Method suffered from several disadvantages, which made it extremely time-consuming. Firstly, sorting data streams is very inefficient 36,37 , especially when the size of the data stream is large. Secondly, the Range-based Method does not apply any kind of filtering, hence, entire data streams will participate in the Cartesian products, resulting in a huge number of Cartesian products being performed. Finally, the Range-based Method suffers a lot from data skewness. For instance, [1,1,1,1,1,1,2,2,3] will be partitioned into (−∞,1), [1,2) and [2,+∞), which could lead to severely imbalanced workloads between partitions, and is especially hurtful in distributed environments.

Randomised Method
To address the issues faced by the Range-based Method, Okcan et al., 33 put forward an algorithm, called One-Bucket-Theta (OBT), which partitioned the data stream as evenly as possible, and then randomly distributed the partitioned blocks in the distributed environment.
For instance, in Figure 2(b), the One-Bucket-Theta algorithm first sorts the data streams. Then, take the number of partitions to be 3 as an example, instead of partitioning the data stream based on ranges, the One-Bucket-Theta partitions the 1 st → 3 rd , 4 th → 6 th , and 7 th → 9 th data elements into 3 partitions, respectively, resulting in a relatively more even partitioning scheme in practice.
Furthermore, the randomised distribution of partitions to processes in the distributed environment further balances the workloads among processes and avoids the severe load imbalance. However, the One-Bucket-Theta still suffers from the time complexity brought by its sorting operation. Also, the lack of filtering strategy makes entire data streams being involved when performing the Cartesian products, which significantly impairs its efficiency.

Filtering Method
In order to improve the efficiency of the theta-join by reducing the number of Cartesian products, the Cross Filter Strategy (CFS) proposed by Liu et al., 34 performed stream-level filtering after the data stream partitioning to eliminate data elements that are unlikely to form valid theta-join results based on the theta condition.
As shown in Figure 2(c), under the ">" theta condition, the Cartesian products will not be performed between the entire data stream  and partition [9,12] of data stream . Since the maximum value 9 of data stream  equals to the minimum value 9 in partition [9,12] of data stream , hence there is no way for the Cartesian products between data stream  and range [9,12] of data stream  to possess valid ">" theta-join results, and hence partition [9,12] of data stream  is eliminated by the stream-level filtering strategy. Following the same principle, the stream-level filtering can also be applied to other similar theta conditions as in {≥, <, ≤}. However, this stream-level filtering is coarse-grained and fails to remove unnecessary Cartesian products as much as possible. The stream-level filtering finds out that partition [0,3) of data stream  can form valid theta-join results with partition [0,4) of data stream  under the ">" theta condition, and hence partition [0,3) of data stream  could not be eliminated and the Cartesian products between it and the entire filtered data stream  will be performed, even though the Cartesian products between [0, 3) and [4,9) are totally redundant.
To further improve the efficiency of the theta-join operation in terms of reducing the number of Cartesian products, Hu et al., 35 presented the FastThetaJoin (FTJ) algorithm. In contrast to performing filtering at the stream-level, the FastThetaJoin algorithm utilised the partition-level filtering strategy, which compared all the partition pairs from both data streams and avoided performing the Cartesian products between the partitions that are unable to form any valid theta-join results based on the theta condition as shown in Figure 2(d). However, the lack of the pre-filtering mechanism before partitioning not only incurred all data elements in two data streams to be involved in partitioning, irrespective of whether they are able to form valid theta-join results or not, but also made the partitioning more coarse-grained, and hence compromised the overall performance.
Furthermore, in the FastThetaJoin algorithm, the partitioning of data streams was conducted in an isolated manner. Failing to partition the data streams collaboratively made the FastThetaJoin highly laborious as the coarse-grained isolated data stream partitioning is incapable to remove some worthless data elements in some partitions. In terms of auxiliary procedures, the FastThetaJoin algorithm adopted the re-partitioning on oversized partitions so that the workloads would be well-balanced and the method will be more distributed-environment-friendly. The Cartesian products were performed between the remaining partitions of two data streams.

Research Opportunity
Despite constant optimisation efforts being made, there is still room to extend and improve the efficiency of the theta-join operation, hence it brings us the motivation of the proposed framework. As shown in Figure 2(d), performing the filtering strategy after partitioning ends up with performing the Cartesian products between 0 in data stream  and the partition [0,4) of data stream , which is unnecessary under the ">" theta condition. Hence, a pre-filtering strategy that filters the data streams before the partitioning will be beneficial to reduce the amount of data that needs to be partitioned, and it can also make the partitioning become more fine-grained as the pre-filtering shortens the ranges of the data streams.
Moreover, the isolated partitioning of data streams can be substituted with the amalgamated partitioning mechanism. Under the isolated manner with the theta condition as ">", the Cartesian products between partition [3,6) and [4,8) will be conducted simply because of a single valid theta-join result which is 5 > 4, but the Cartesian products between [3,4] and [4,8) are completely unnecessary. To avoid these pointless Cartesian products incurred by the isolated partitioning, the partitioning information of two data streams will be amalgamated to form an amalgamated partitioning scheme so that the partition [3,6) will be split to filter more unnecessary Cartesian products that possess no valid theta-join result.
Hence, by introducing the pre-filtering strategy and the amalgamated partitioning mechanism, we can integrate them into a unified framework to effectively reduce the amount of unnecessary Cartesian products while balancing the workload in the distributed environment and avoiding the time-consuming sorting operation. Therefore, we can substantially boost the efficiency of the theta-join operation.
As illustrated in Figure 3, the bar chart presents the number of Cartesian products that are performed by each algorithm in the example in Figure 2. The rightmost red bar, as well as the red dotted line, indicate the number of theta-join results, i.e., no matter how the theta-join algorithm is optimised, it is the lower bound of the number of Cartesian products that needs to be performed. The proposed Prefap algorithm achieves the lowest number of Cartesian products in this example and is very close to the optimal lower bound, which indicates that the research direction of the Prefap algorithm is promising.

PREFAP FRAMEWORK
In this section, we will introduce the proposed Prefap approach in terms of its framework and workflow, followed by a detailed explanation of each constituting component, i.e., the pre-filtering strategy, the amalgamated partitioning scheme, as well as some auxiliary steps to complete the theta-join operation. Note that in this section and the corresponding pseudocode, we explain the theta-join operation on two data streams for the sake of simplicity. However, it is not difficult to extend our approach to multi-way theta-joins. The symbols and acronyms used and their corresponding interpretations are given in Table 1.

Prefap Workflow
The framework of the proposed Prefap is illustrated in Figure 4, together with its workflow as follows: Step 1 -Pre-filtering Strategy: Firstly, the proposed pre-filtering strategy will be employed to filter out data elements in two data streams based on the theta condition, so that the workloads in later steps can be lowered and the partitioning will be more fine-grained (as in Section 3.2 and Algorithm 1). This step will be completed by the Filtering Unit of the Prefap framework.
Step 2 -Amalgamated Partitioning Mechanism: Then, it comes to the amalgamated partitioning mechanism. The partitioning boundaries of data streams are calculated, amalgamated, and the amalgamated partitioning boundaries will be used to produce fine-grained partitions to benefit the partition-level filtering (as in Section 3.3 and Algorithm 2). This step will be completed by the Partitioning Unit of the Prefap framework.
Step 3 -Auxiliary Procedures: Finally, after completing the developed pre-filtering strategy and the amalgamated partitioning mechanism, some auxiliary procedures adopted in FastThetaJoin are followed to complete the theta-join operation for the final output results: Step 3.1 -Oversized Partition Re-partitioning: To balance the workload among processes under the distributed environment, oversized partitions will be re-partitioned to balance the workload as much as possible. This step will be completed by the Partitioning Unit of the Prefap framework.

2.
Step 3.2 -Partition-level Filtering: The partitions will then be filtered again based on the theta condition to avoid unnecessary Cartesian products. This step will be completed by the Filtering Unit of the Prefap framework.

FIGURE 4
The Prefap framework. In the Filtering Unit of the Prefap framework, the data streams will firstly be pre-filtered based on the theta condition to get rid of unnecessary data elements that are deemed not possible to form any valid theta-join results. Then, their corresponding partitioning boundaries will be calculated by the Partitioning Unit, followed by the amalgamated partitioning procedure, in which the partitioning boundaries are amalgamated. The resulted data streams are partitioned based on the amalgamated partitioning boundaries, and those oversized partitions are re-partitioned to achieve load balancing in the distributed environment. After that, the partitions will be handled by the Filtering Unit and the partition-level filtering is performed. Finally, Cartesian products are conducted and the theta-join results are retrieved based on the theta condition. Note that the diagram shows how two data streams are processed by the Prefap framework for the purpose of illustration only, the framework can be extended to work on multi-way data stream theta-join as well. For three-way theta-join, two data streams will firstly be processed by the Prefap framework, the result produced will then be joined with the third data stream to yield the final result.

Step 3.3 -Cartesian Products and Theta-join Results:
Finally, the Cartesian products are performed and the theta-join results are retrieved based on the theta condition. All auxiliary procedures will be presented in Section 3.4 and Algorithm 3.

Pre-filtering Strategy
As opposed to all the aforementioned algorithms that directly perform the partitioning without any pre-filtering, a pre-filtering strategy is applied in the Prefap framework to eliminate certain amounts of unnecessary data involved in the operation. This is performed in the Partitioning Unit and the the pseudocode of this step is given in Algorithm 1. Specifically, the pre-filtering strategy scans and filters the two data streams according to the theta condition to eliminate the data elements that are deemed impossible to produce valid theta-join results before the partitioning is performed. Also, unlike some previous methods, the proposed pre-filtering strategy does not require data stream sorting, which severely hurts the efficiency of the theta-join operation. For example, as illustrated in Figure 2(e), given the theta condition is ">", any value in data stream  that is less than or equal to the minimum value of data stream  is safe to be removed as it is not possible to form valid theta-join results with any value in data stream . The similar mechanism applies for data stream  as well. Therefore, we can safely eliminate any value in data stream  that is greater than or equal to the maximum value of data stream , as is shown in line 2-3 in Algorithm 1. The prefiltering mechanism works similarly for other theta conditions as presented in Algorithm 1. As indicated in Figure 2(e), upon applying the pre-filtering strategy, the first row and the last three columns will be directly filtered out and not involved in later  ← .remove(≤  ) 10: else 11:  ← .remove(>  ) 12:  ← .remove(<  ) 13: end if 14: return ,  processing steps. Hence the pre-filtering strategy can greatly reduce the amount of data that needs to be processed and improve the efficiency of the theta-join operation.
Meanwhile, since the partitioning boundaries being used in the subsequent processing are calculated based on the minimum and maximum values of data streams, the use of the pre-filtering strategy to filter useless data is likely to reduce the span between the maximum and the minimum values, and therefore making the partitioning more fine-grained and benefiting the partition-level filtering performed later.  (3) 3: .sort() 6: // Partition data streams using the amalgamated partitioning boundary 7: ← Partition based on 8: ← .filter( .size() != 0) 9: return , ∈ {, }

Amalgamated Partitioning Mechanism
The data streams are partitioned in the Partitioning Unit based on the range defined by the partitioning boundaries in the course of the theta-join operation as shown in line 3 in Algorithm 2. The partitioning boundaries of data streams will be calculated after the pre-filtering strategy is executed with each partition having a span, which is denoted as sp and is calculated as follows: where denotes the number of partitions. As such, in the example as shown in Figure 2(e), given that data streams are partitioned into three partitions, data stream  has a partition span equal to 8 3 and will be partitioned into the following three partitions: [1,3.67), [3.67,6.33) and [6.33,9], and the same partitioning boundary calculations can also be applied to data stream  as indicated in line 2-3 in Algorithm 2.
For all aforementioned algorithms, two data streams are partitioned separately based on their respective partitioning boundaries after they are obtained. As such there is no interference between the partitions of each stream, which implies the partitioning is accomplished in an isolated manner. Clearly, the isolated partitioning lacks the notion of collaborative partitioning information of the data streams, and thus damages the efficacy of the partition-level filtering. Take Figure 2(e) as an example, for the coarse-grained partitioning, partition [1,3.67) of data stream  and partition [2.67, 5.33) of data stream  will be produced separately and the Cartesian products are performed between them under the ">" theta condition since valid thetajoin results can be available between these two partitions. However, not all Cartesian products between these two partitions are necessary, say, the Cartesian products between [1,2.67] and [3.67,5.33) are completely useless as they are judged to be impossible to possess any valid theta-join results based on the ">" theta condition. Therefore, the coarse partitions produced by the coarse-grained isolated partitioning strategy will impair the efficacy of the partition-level filtering, which could result in more Cartesian products than necessary, and thus seriously hinder the efficiency of the theta-join operation.
Given the drawback caused by this isolated coarse-grained partitioning strategy, we consider an amalgamated partitioning mechanism as shown in Algorithm 2, which is useful to address this issue. To make the partitioning more fine-grained, after calculating the partitioning boundaries, we amalgamate the partitioning boundaries of data streams to form the amalgamated partitioning boundaries as presented in line 5 in Algorithm 2. By fusing the partitioning information of data streams, the aforementioned drawback can be circumvented. Therefore, the effectiveness of the partition-level filtering is improved, which would lead to the reduction of the number of Cartesian products to be conducted.

Auxiliary Procedures
To obtain the theta-join results, some auxiliary procedures as shown in Algorithm 3 that are adopted in the FastThetaJoin algorithm are exploited to follow up the proposed approaches in the Prefap framework, which are detailed as follows: 1) Oversized Partition Re-partitioning: By following the common design scheme in 35 , we design the framework that can repartition any oversized partitions to balance the workload as much as possible in the distributed environment as shown in line 2-9 in Algorithm 3. Specifically, any partition with its size larger than the average partition size is regarded as an oversized partition. Once a partition is judged as an oversized partition, it will be re-partitioned into a number of sub-partitions defined as: and the pseudocode of this process is shown in line 3-8 in Algorithm 3. The oversized partition re-partitioning is completed by the Partitioning Unit of the Prefap framework. Together with the fine-grained amalgamated partitioning, the load balancing of the framework can be improved.
2) Partition-level Filtering: Once both data streams are partitioned, the Partitioning Unit of the Prefap framework will filter the partitions based on the theta condition as is presented in line 11-23 in Algorithm 3, so that the Cartesian products between the partitions that possess no valid theta-join results will not be performed.
In the example illustrated in Figure 2(e), even though partition [1,2.67) from data stream  and partition [3.67,5.33) from data stream  are both produced, the Cartesian products between these two partitions will not be performed under the ">" theta condition, because even the maximum value in the former partition is smaller than the minimum value in the latter partition, which is exactly line 13-14 in Algorithm 3. Hence, by applying the partition-level filtering based on the theta condition, unnecessary Cartesian products are eliminated, leading to a more efficient algorithm.

3) Cartesian Products and Theta-join Results:
After completing the pre-filtering, amalgamated partitioning, oversized partition re-partitioning, and the partition-level filtering, the remaining Cartesian products are highly refined, and then the Cartesian products will be distributed to different processes to output the final theta-join results based on the theta condition as shown in line 24-28 in Algorithm 3.

EMPIRICAL STUDIES
To validate the effectiveness of the Prefap algorithm, comprehensive evaluations are performed on both synthetic and real data streams from two-way to multi-way theta-join operations. We compare our algorithm against several theta-join algorithms, including the state-of-the-art algorithm FastThetaJoin (FTJ) 35 and several well-known and widely used algorithms, such as Range-based Method (RBM) 32 , One-Bucket Theta (OBT) 33 , and Cross Filter Strategy (CFS) 34 .

Number of Cartesian products The number of Cartesian products that the algorithm performs Elapsed time
The total time elapsed, measured in milliseconds (ms) Load balancing ratio (In) The maximum input load among processes divided by the average input load among processes Load balancing ratio (Out) The maximum number of theta-join results among processes divided by the average number of theta-join results among processes

Experimental Setup and Data Streams
We follow the Prefap workflow described in Section 3.1 to conduct the experiments where the number of partitions is set to be 10 and the window size to be 1000. The evaluation metrics we use to evaluate the performance of the algorithms are listed in Table 2 with their corresponding interpretations. To fully testify the effectiveness of the proposed algorithm, the theta-join on two-way data streams and multi-way data streams are performed to validate that the proposed algorithm scales well. The hypothesis testings are then conducted to demonstrate that the Prefap algorithm achieves a statistically significant performance improvement compared with FTJ. The ablation studies are also conducted to show the efficacy of the algorithm and reveal the importance and necessity of each component introduced in the Prefap framework. Finally, the number of partitions and the window size are adjusted to show the superior performance of the Prefap algorithm in various settings. We use both synthetic and real data streams to testify the effectiveness of the proposed algorithm. For the synthetic datasets, randomly generated Uniform and Normal data streams are exploited to represent some typical non-skewed data streams, while the randomly generated Zipf data stream acts as a representative of a typical skewed data stream. As for the real datasets, by following the method in 34 , we use the Clouds dataset provided by the U.S. Department of Energy in our experiments where the real-time wind speed measured in metre per second ( ∕ ) of different months in 2000 are leveraged to form different data streams. Additionally, the stock market price data streams provided by Yahoo! Finance 38 are also utilised, in which the realtime high price of stocks of different companies between 2010 and 2020 are used to form different data streams. The detailed parameter settings of different synthetic data streams will be given when they are used in the following subsections.
In terms of the experimental infrastructure, a server equipped with Intel Core i7 7600U CPU and 32GB of memory is utilised. The CPU has two cores with 4 processors, so that the Cartesian products involved in the Prefap framework can be computed in a distributed manner. ; The performance comparisons between algorithms when performing theta-join on synthetic 2-way Uniform data streams. The theta condition is  ≤ . The stream size of both data streams are 1000. The Uniform data stream  fluctuates in range [20,50], while the Uniform data stream  is in range [10,40].

2-way Data Stream Theta-join
The performances of theta-join algorithms for two-way uniform data streams, zipf data streams, clouds data streams, and stock price data streams are illustrated in Figure 5 to 8, respectively. Note that the detailed data stream and distribution configurations (including stream size, distribution parameters, join attribute used, condition) have been given in the corresponding image captions.
The results show that, in all cases, even for the highly skewed data streams such as the zipf data streams, the Prefap algorithm performs better than the state-of-the-art method FTJ in terms of efficiency, and significantly outperforms all other algorithms. More specifically, the Prefap algorithm achieves 34.7%, 19.4%, 3.0% and 37.7% reductions on the number of performed Cartesian products compared with FTJ with respect to 4 different kinds of data streams, respectively. The significant reduction of the number of Cartesian products yields better efficiency, as indicated by the shortest elapsed time performance in all cases.
The red dotted lines in Figure 5 (b) to 8(b) indicate the number of theta-join results, i.e., no matter how the algorithm is optimised, it must perform at least this number of Cartesian products to yield the complete result. As shown in the results, the Prefap algorithm significantly minimises the gap between the number of Cartesian products it performs and the optimal case. Compared with FTJ, it clearly indicates an effective performance boost and thus demonstrates that the cooperation between the pre-filtering strategy and the amalgamated partitioning mechanism contributes positively towards reducing the redundancy in Cartesian products, which in turn significantly boosts the performance.
In terms of the load balancing as illustrated in Figure 5 to 8, compared with the FTJ, the Prefap algorithm performs better for both input and output load balancing ratios, indicated by the red dotted line in sub-figure (d). Hence, it justifies that with the collaboration between the fine-grained amalgamated partitioning scheme and the oversized partition re-partitioning mechanism, the Prefap algorithm performs well in the distributed environment by distributing workloads in a relatively even way.

Multi-way Data Stream Theta-join
To further verify that the Prefap algorithm scales well when processing multi-way data streams, comprehensive experiments are conducted on both synthetic and real multi-way data streams. As shown in Figure 9(a), the uniform multi-way data streams represent a typical example of the non-skewed data streams. While as illustrated in Figure 10(a), the New York Stock Exchange (NYSE) stock prices of three randomly selected companies between 2017 and 2020 are skewed and are therefore capable of fully testifying the effectiveness of the proposed algorithm. Note that the detailed data stream configurations have been given in the corresponding image captions.
; In Figure 9 to 10, sub-figure (b) and (d) present the evaluation results of the number of Cartesian products and the elapsed time, respectively. As the gaps between the Prefap algorithms and other algorithms are so significant, the sub-figure (b) and (d) have been zoomed in as shown in sub-figure (c) and (e), respectively. As we can see, under both synthetic and real multi-way data streams, the Prefap algorithm outperforms the FTJ algorithm by a large margin, achieving a 24.1% and a 7.7% decrease in the number of Cartesian products in two cases, respectively. Hence, it is also natural to observe that the elapsed time in both cases are reduced accordingly, hence the algorithm becomes more efficient. Furthermore, the Prefap algorithm attains a near-optimal performance. Given the number of theta-join results, i.e. the minimum number of Cartesian products required to be performed that is indicated by the red dotted line in sub-figure (b) and (c) in Figure 9 to 10, the Prefap algorithm only performs 0.026% and 1.7% more Cartesian products compared with the optimal scenario, indicating excellent theta-join performance.
Therefore, the superior performance of the Prefap algorithm in various cases strongly testifies that the collaboration of the pre-filtering strategy and the amalgamated partitioning mechanism is effective in improving the theta-join performance.

Significance Tests
To further validate that the improvement achieved by Prefap over the state-of-the-art FTJ algorithm is statistically significant, three representative data stream settings, i.e., theta-join between 2-way zipf data streams, 2-way cloud data streams, and multiway stock price data streams are performed 30 times and the T-tests are employed to verify the significance of the performance boost using 0.05 as the significance threshold. Note that the detailed data stream configuration has been given in the image caption.
As shown in Figure 11, 4 bars I, II, III, IV in sub-figure (a) and (b) indicate the − log( _ ) of the one-sided T-test on 4 different evaluation metrics as presented in Table 2 for two 2-way theta-join settings. The − log( _ ) results of the multi-way theta-join are presented in sub-figure(c), in which the bars correspond to the − log( _ ) results of the differences between Prefap and FTJ on the number of Cartesian products and the elapsed time, respectively. In Figure 11, the red dotted lines indicate − log(0.05), which is the significance threshold of the one-sided T-test. If the bar is higher than the red dotted line, it indicates the acceptance of the alternate hypothesis, which is Metric < Metric . And the higher the bar is, the more significant the performance improvement produced by the Prefap algorithm is.
From Figure 11, we can clearly observe that all bars in all cases are significantly higher than the red dotted line, which indicates that in all data stream settings, the Prefap algorithm achieves a lower evaluation metric value than the FTJ algorithm, i.e., less number of Cartesian products are performed, less amount of elapsed time is required, and the load balancing ratio of both input , ,, )7-DQG35 ()$3  D   ORJ3YDOXH   , ,, . If the bar is higher than the red dotted line, it indicates that the alternate hypothesis is accepted, i.e., the Prefap algorithm outperforms the FastThetaJoin algorithm on that metric. The higher the bar is, the more significant the performance improvement produced by the Prefap algorithm is. and output is closer to 1. The heights of the bars are much higher than the red dotted line, which indicate superior performances with high statistical confidence. Hence, statistically, the T-test results verify the superiority of the Prefap algorithm, and the performance enhancement is statistically significant. The statistically testifiable superior performance also demonstrates that the Prefap framework can collaboratively contribute towards better theta-join efficiency and excellent load balancing in the distributed environment, the benefit brings by the Prefap framework is statistically significant.

Ablation Study
The ablation study is conducted by evaluating several variants of the Prefap framework: (1) Prefap with the pre-filtering being removed; (2) Prefap with the amalgamated partitioning being turned off; and (3) Prefap with both the pre-filtering and the amalgamated partitioning being ablated.
Two-way and multi-way stock price data streams of randomly selected companies are used as the representative tasks on which the theta-join is performed by using both the full Prefap algorithm and its ablated variants. The results are presented in Table 3 and 4 for two-way and multi-way theta-join tasks, respectively. Note that the detailed data stream configuration is also provided in table captions. We can observe that the full Prefap algorithm outperforms all its variants by a large margin, which indicates that any one of these components in the Prefap framework plays an indispensable role and brings benefits to reduce the number of Cartesian products being performed and enhance the overall efficiency. Among all these components, the amalgamated partitioning scheme brings the highest amount of Cartesian product reductions in two tasks, which is 16.8% and 6.3%, respectively. Correspondingly, the elapsed time drops by 17.5% and 6.2%, respectively. This further validates the importance of amalgamating the partitioning scheme and the benefits of avoiding coarse-grained isolated partitioning. As indicated in Table  3 and 4, although the pre-filtering strategy attains relatively less amount of performance gain, which is 1.0% and 0.2% in the two-way and multi-way task, respectively, however, the collaboration of both the pre-filtering and the amalgamated partitioning yields a tremendous decrease of Cartesian products of 29.1% and 12.9%, respectively, for these two tasks. Therefore, by collaborating the pre-filtering strategy with the amalgamated partitioning mechanism in the Prefap framework, it is highly effective in improving the efficiency of the theta-join operation. As such, the promising results successfully verify the effectiveness of the Prefap algorithm.

Algorithm Efficiency
Number of Partitions: To verify the effectiveness of the Prefap algorithm when working with different number of partitions, the performance of the Prefap algorithm with different number of partitions when processing two randomly selected tasks are presented in Figure 12 and 13. The detailed data stream configuration has been given in the corresponding image caption. For both tasks, the number of Cartesian products decreases with the increase of the number of partitions, thanks to the more fine-grained partitioning effect brought by a larger number of partitions, as it benefits the filtering strategy and makes it more effective. In all partition settings, the Prefap algorithm outperforms the FTJ algorithm by significantly reducing the number of Cartesian products being performed, and the elapsed time is reduced accordingly. This further demonstrates the superiority of the Prefap algorithm with different number of partitions. Also, when the number of partitions is raised, the input and output load balancing ratio are improved marginally and are relatively close to 1, which indicates a relatively even workload distribution in the distributed environment achieved by the Prefap algorithm.  Window Sizes: The performance of the Prefap is also evaluated when the window size is varied. The performance on two randomly selected tasks, i.e., 2-way uniform data streams and 2-way stock price data streams, are shown in Figure 14 and 15, respectively. The detailed data stream configuration has been given in the corresponding image caption.
According to the experimental results, the number of Cartesian products being conducted and the elapsed time grow relatively proportionally with the increase of the window size, which demonstrates that the Prefap algorithm scales well in terms of window size. In all window size settings, the Prefap algorithm attains the lowest number of Cartesian products compared with its counterparts, and is relatively close to the optimal case marked by the red dotted line in sub-figure (a). The excellent performance of Cartesian product reduction benefits the efficiency of the algorithm, as indicated by the lowest elapsed time achieved by the Prefap algorithm. Meanwhile, from sub-figure (c) and (d) of Figure 14 and 15, the Prefap algorithm attains a superior performance, compared with the FTJ algorithm, on both the input and the output load balancing ratio in all window size settings. Hence, it further validates the excellent scalability of the Prefap algorithm in terms of window size.

CONCLUSION
In this paper, we propose the Prefap algorithm to enhance the performance of the theta-join operation. Compared with the Fast-ThetaJoin algorithm, the pre-filtering strategy is applied to filter data elements in data streams that are deemed not possible to produce any valid theta-join results. The pre-filtering not only reduces the amount of data and hence lessens the workload, but also makes the partitioning more fine-grained to benefit further filtering. Then, during partitioning, the amalgamated partitioning scheme is employed to amalgamate the partitioning of two data streams, so that the performance degradation of the partition-level filtering caused by the coarse-grained isolated partitioning is avoided. By collaborating these mechanisms with the oversized partition re-partitioning strategy, as well as the partition-level filtering mechanism based on the theta condition, it forms our proposed Prefap framework and it becomes more efficient when performing the theta-join operation. Comprehensive experiments and analyses are conducted to demonstrate the superiority of the performance against the recently published FastThetaJoin and several well-known theta-join algorithms.