# Density-optimized Intersection-free Mapping and Matrix Multiplication for
Join-Project Operations (extended version)

Zichun Huang, Shimin Chen  SKL of Computer Architecture, ICT, CASUniversity of
Chinese Academy of Sciences

###### Abstract.

A Join-Project operation is a join operation followed by a duplicate
eliminating projection operation. It is used in a large variety of
applications, including entity matching, set analytics, and graph analytics.
Previous work proposes a hybrid design that exploits the classical solution
(i.e., join and deduplication), and MM (matrix multiplication) to process the
sparse and the dense portions of the input data, respectively. However, we
observe three problems in the state-of-the-art solution: 1) The outputs of the
sparse and dense portions overlap, requiring an extra deduplication step; 2)
Its table-to-matrix transformation makes an over-simplified assumption of the
attribute values; and 3) There is a mismatch between the employed MM in BLAS
packages and the characteristics of the Join-Project operation.

In this paper, we propose DIM3, an optimized algorithm for the Join-Project
operation. To address 1), we propose an intersection-free partition method to
completely remove the final deduplication step. For 2), we develop an
optimized design for mapping attribute values to natural numbers. For 3), we
propose DenseEC and SparseBMM algorithms to exploit the structure of Join-
Project for better efficiency. Moreover, we extend DIM3 to consider partial
result caching and support Join-$op$ queries, including Join-Aggregate and MJP
(Multi-way Joins with Projection). Experimental results using both real-world
and synthetic data sets show that DIM3 outperforms previous Join-Project
solutions by a factor of 2.3$\times$-18$\times$. Compared to RDBMSs, DIM3
achieves orders of magnitude speedups.

††Our codes are available at https://github.com/schencoding/JoinProject-DIM3 .

##  1\. Introduction

A Join-Project operation is a join operation followed by a duplicate
eliminating projection operation (Amossen and Pagh, 2009). Given two tables
$R(x,y)$ and $S(z,y)$, the Join-Project operation can be written as follows:

(1) $\Pi_{x,z}(R(x,y)\Join_{y}S(z,y))$

It joins $R$ and $S$ with $y$ as the join key, then projects and deduplicates
$($$x$,$z$$)$ tuples. $\Pi$ denotes the duplicate eliminating projection.

_Example_ : In the HetRec2011 data set (Cantador et al., 2011), _U(userID,
bookID, tagID)_ table contains tags given by users to books that they read,
and _B(bookID, tagID, weight)_ table records books and their possible tags
with weights. The following SQL query recommends books to users based on the
tags recorded in users’ reading history:
 _SELECT DISTINCT U.userID, B.bookID_
 _FROM U, B_
 _WHERE U.tagID = B.tagID_
This Join-Project operation joins U and B on tagID, then projects and
deduplicates using SELECT DISTINCT. The results can be stored by the
application for quick user-specific recommendations.

The Join-Project operation is used in a large variety of applications (Deep et
al., 2020), including entity matching, set analytics, and graph analytics. The
above is an example of entity matching. Similar examples include finding users
who have seen the same movies in the MovieLens data set (Harper and Konstan,
2015), and discovering co-authors in the DBLP data set (Yang and Leskovec,
2015a). Moreover, if tuple $($$x$,$y$$)$ represents that set $x$ contains
element $y$, then the Join-Project operation using $y$ as the join key obtains
all the pairs of sets that intersect with each other. Furthermore, if we
interpret tuple $($$x$,$y$$)$ as an edge between two vertices $x$ and $y$ in a
graph, then the Join-Project operation can be used to compute all pairs of
vertices that are indirectly connected.

###  1.1. Previous Solutions

Classical Solution. The classical solution to compute the Join-Project in
RDBMSs is to first perform the join operation (Mishra and Eich, 1992; Kim et
al., 2009; Balkesen et al., 2013), then deduplicate the projected join results
using hash tables (Kocberber et al., 2013) or other types of indices (Graefe
and Kuno, 2011; Ho and Park, 2016; Kraska et al., 2018). The time complexity
of the classical solution is $\Theta(|R|+|S|+|OUT_{J}|)$, where $|R|$, $|S|$,
and $|OUT_{J}|$ denote the sizes of input table $R$, table $S$, and the join
results before deduplication, respectively. This cost is reasonable when the
number of duplicates is low. However, the solution is less efficient when
$|OUT_{J}|$ is much larger than the size $|OUT_{P}|$ of the final results
after deduplication. For example, $|OUT_{J}|$ is 3.7x as large as $|OUT_{P}|$
in the HetRec2011 example, while $|OUT_{J}|$ is 24x larger than $|OUT_{P}|$ in
the MovieLens data set. Let $|X|$, $|Y|$ and $|Z|$ denote the number of
distinct values in column $x$, $y$, and $z$ in Eqn 1, respectively. Consider
the case where $|X|$=$|Y|$=$|Z|$=$n$. $|OUT_{J}|$ can be $\mathcal{O}(n^{3})$
in the worst case, while $|OUT_{P}|$ is only $\mathcal{O}(n^{2})$. In other
words, the classical solution can spend a lot of time generating the
$|OUT_{J}|$ join results and then processing them to remove a large number of
duplicates.

Matrix Multiplication. Alternatively, the Join-Project operation can be
computed using MM. The basic idea is to represent tables $R(x,y)$ and $S(z,y)$
as two matrices $\mathbf{R}^{x\times y}$ and $\mathbf{S}^{y\times z}$.
Specifically, $\mathbf{R}_{x_{i},y_{k}}=1$ if and only if tuple
$($$x_{i}$,$y_{k}$$)$$\in R$ (similarly for $S$). Then the multiplication of
$\mathbf{R}^{x\times y}$ and $\mathbf{S}^{y\times z}$ gives matrix
$\mathbf{C}^{x\times z}$, where
$\mathbf{C}_{x_{i},z_{j}}=\sum_{k=1}^{|Y|}\mathbf{R}_{x_{i},y_{k}}\mathbf{S}_{y_{k},z_{j}}$.
A non-zero element $\mathbf{C}_{x_{i},z_{j}}>0$ in the matrix corresponds to a
tuple $($$x_{i}$,$z_{j}$$)$ in the final output of the Join-Project operation.
Compared to the classical solution, MM performs the join and the deduplication
together. There are efficient MM implementations in BLAS (Basic Linear Algebra
Subprograms) packages with advanced techniques (Saule et al., 2013; Dalton et
al., 2015; Kurzak et al., 2009). Moreover, there are sub-cubic MM algorithms
in theory. The best known is the Coppersmith-Winograd algorithm with
$\mathcal{O}(n^{2.373})$ complexity (Gall and Urrutia, 2018).

Hybrid Solution. Recent studies (Amossen and Pagh, 2009; Deep et al., 2020)
combine the classical solution and the MM solution based on the observation
that the classical solution performs better when the data is sparse, while MM
performs better when the data is dense. Amossen et al. (Amossen and Pagh,
2009) propose an algorithm to partition the data into dense and sparse parts
according to degrees of $x$, $y$, and $z$. Then dense MM is employed for the
dense parts and the classical solution is used for the remaining parts.

Figure 1. Breakdown of DHK run time.

Deep, Hu, and Koutris (Deep et al., 2020) correct errors in the cost analysis
of (Amossen and Pagh, 2009) and implement the algorithm for experimental
comparison. We call this state-of-the-art algorithm _DHK_. DHK has been shown
to significantly outperform the classical solution. However, the following
problems reduce the efficiency and practicality of DHK:

  * •

_Overlapping outputs between sparse and dense parts_ : The Join-Project
results computed from the sparse and from the dense parts can overlap.
Consequently, a final deduplication step is necessary to deduplicate the
overlapping results. Figure 1 shows the breakdown of the run time of DHK for
five data sets (cf. Section 5). For all these data sets, DHK partitions the
input data into dense and sparse parts. It performs the final deduplication
step, incurring significant overhead.

  * •

_Over-simplified assumption for table-to-matrix transformation_ : The DHK
implementation assumes that the input columns $x$, $y$, and $z$ contain
consecutive natural numbers starting from 0. Thus, it directly uses their
values as row or column ids in the matrices. However, in reality, database
attribute values are rarely natural numbers. A step is missing: mapping values
of input columns to consecutive natural numbers.

  * •

_Caveats of MM Implementation_ : DHK invokes MM in BLAS packages (e.g., Intel
MKL (Wang et al., 2014)) as a black box. On the one hand, BLAS packages
typically implement the $\mathcal{O}(n^{3})$ MM algorithm. It would be
interesting to investigate sub-cubic MM algorithms. On the other hand, the MM
invocation as a black box cannot exploit the characteristics of the Join-
Project operations for performance improvement.

###  1.2. Our Solution: DIM3

In this paper, we propose an efficient and practical Join-Project algorithm,
DIM3 (Density-optimized Intersection-free Mapping and Matrix Multiplication).
We address the above problems as follows:

  * •

_Intersection-free partitioning_ : We propose a novel partitioning method that
divides matrix $\mathbf{S}^{y\times z}$ into subsets of rows based on the
density of $z_{j}$ rows and then chooses different evaluation strategies for
the dense rows and the sparse rows. Since the results $($$x_{i}$,$z_{j}$$)$ of
the two parts have different $z_{j}$s, this method is guaranteed to be
intersection-free. Hence, DIM3 completely removes the final deduplication step
required by DHK.

  * •

_Optimized mapping design_ : We investigate the design of the mapping step in
DIM3. First, we identify cases (e.g., auto increment, dictionary encoding)
where columns contain roughly natural numbers and thus the mapping step can be
skipped. Second, we exploit the mapping of the shared join key columns to
perform a semi-join like optimization to discard tuples without matches.
Finally, we design a cache-optimized hash-based algorithm to efficiently
compute the mappings.

  * •

_Optimized MM algorithms_ : We obtain and evaluate an implementation of the
sub-cubic Strassen algorithm (Strassen, 1969) in a recent study (Benson and
Ballard, 2015). However, our results show that it cannot beat Intel MKL (cf.
Section 5.2). Therefore, we focus on $\mathcal{O}(n^{3})$ algorithms in DIM3.
For the dense rows, we observe that the computation
$\mathbf{C}_{x_{i},z_{j}}=\sum_{k=1}^{|Y|}\mathbf{R}_{x_{i},y_{k}}\mathbf{S}_{y_{k},z_{j}}$
can stop early as soon as there is a non-zero
$\mathbf{R}_{x_{i},y_{k}}\mathbf{S}_{y_{k},z_{j}}$. We design the _DenseEC_
(Dense MM with Early stopping Checking) algorithm. For the sparse rows, we
design the _SparseBMM_ (Sparse Boolean MM) algorithm that leverages the CSR
(Compressed Sparse Row) (Kepner and Gilbert, 2011) format of matrix
$\mathbf{S}^{y\times z}$ as a hash table on the join key $y$ with NO hash
conflicts. We also introduce a way to reduce the cost for initializing the
deduplication vector.

In addition to addressing the three problems, we extend the Join-Project
solution in the following two directions:

  * •

_Partial Result Caching_ : We observe that in real-world data sets, computing
results for different $z_{j}$ values often take widely different amounts of
time. This motivates us to investigate if caching results for a subset of
$z_{j}$ values is profitable. We consider caching either the original or the
complement set of $x$’s given a $z_{j}$. The caching decision for $z_{j}$ is
based on a score computed from the computation time and the required result
space. Our experiments show that partial result caching can effectively reduce
the Join-Project computation time with reasonable space cost for most data
sets.

  * •

_Support for Join-Op Queries_ : We discuss different types of relational
operations $op$, and investigate whether we can leverage the Join-Project
algorithm to support Join-$op$ queries. We study two interesting $op$s in
depth: 1) For Join-Aggregate queries (Hu and Yi, 2020), we show that DIM3 can
be applied with simple modifications; and 2) For MJP (Multi-Way Joins with
Projection) queries, a.k.a. conjunctive queries with projection (Deep et al.,
2021), we develop a dynamic programming algorithm to find the optimal query
plan that considers pushing down deduplication operations to after the join
operations.

###  1.3. Contributions

The main contributions of this paper is threefold. First, we propose DIM3 with
intersection-free partitioning, optimized mapping, and DenseEC and SparseBMM
algorithms to address the three problems in the state-of-the-art solution (cf.
Section 2). Second, we propose partial result caching for the Join-Project
algorithm (cf. Section 3) and generalize DIM3 to support Join-$op$ queries
(cf. Section 4). Third, we perform extensive experimental evaluation using
both real-world and synthetic data sets (cf. Section 5). Experimental results
show that DIM3 outperforms previous Join-Project solutions by a factor of
2.3$\times$-18$\times$. Compared to commercial and open-source RDBMSs, DIM3
achieves orders of magnitude speedups.

##  2\. DIM3 for Join-Project

In this section, we first overview the DIM3 algorithm in Section 2.1. Then, we
explain the components of the algorithm in detail. Specifically, we present
the mapping phase in Section 2.2, intersection-free partitioning in Section
2.3, SparseBMM in Section 2.4, and DenseEC in Section 2.5. We analyze the
algorithm and describe the strategy selection criteria in Section 2.6.
Finally, we analyze the time complexities of the solutions in Section 2.7.

###  2.1. Overview

The DIM3 algorithm is depicted in Figure 2 and listed in Algorithm 1. We
perform the Join-Project operation on two tables $R(x,y)$ and $S(z,y)$.
Without loss of generality, suppose $R$ is the larger table and $S$ is the
smaller table.

DIM3 begins by selecting either the classical or the hybrid solution for
evaluating the Join-Project operation. This allows completely avoiding the
mapping step, which can have significant cost for highly sparse data. DHK
(Deep et al., 2020) uses a rule-of-thumb condition to choose the classical
solution. In comparison, our strategy selection function $f_{1}$ makes better
decisions based on the estimated run times of the classical and the hybrid
solutions (cf. Section 2.6 and 5.2).

In the hybrid strategy, the first step is to map columns $x$, $y$, and $z$ to
consecutive natural numbers. In this way, an $($$x$, $y$$)$ tuple in $R$
(similarly $($$y$, $z$$)$ tuple in $S$) can be converted to an element at row
$x$ and column $y$ in matrix $\mathbf{R}^{x\times y}$. DHK (Deep et al., 2020)
makes the over-simplified assumption that the input columns contain
consecutive natural numbers. We delve into the design of the mapping step to
provide general-purpose support for all attribute types.

The second step converts table $R$ to CSR (Kepner and Gilbert, 2011) format,
then partitions $S$ to $S_{sparse}$ and $S_{dense}$. We propose an
intersection-free partition method so that the Join-Project results of
$S_{sparse}$ and $S_{dense}$ can be simply combined without the final
deduplication step required by DHK (Deep et al., 2020). The partition method
uses function $f_{2}$ to decide which $z$ row is dense. $f_{2}$ will be
described in Section 2.6.

The third step designs SparseBMM and DenseEC algorithms for processing the
sparse and dense data, respectively. SparseBMM computes the Join-Project
$Result_{sparse}$ on $R_{CSR}$ and $S_{sparse}$. DenseEC multiplies $R_{CSR}$
with $S_{dense}$ to obtain $Result_{dense}$. In DenseEC, function $f_{3}$ is
used to choose the best method for intersecting two bitmaps. The computation
of $f_{3}$ will be discussed in Section 2.6.

The final step is to merge $Result_{MM}$ and $Result_{EC}$ to obtain the final
results. Because of intersection-free partitioning, there is no need to
perform an extra deduplication step.

Figure 2. DIM3 for Join-Project.

1

Input: Table $R(x,y)$ and Table $S(z,y)$

Output: List of result tuples $(x,z)$

2

3Estimate $OUT_{J}$;

4 if  _$f_{1}(|R|,|S|,|OUT_{J}|)$ $>$ 0_ then /* Classical is better */

5 return Use classical solution;

6

7 Mapping column values to consecutive natural numbers;

8 $R_{CSR}$ $\leftarrow$ Create CSR for $R(x,y)$;

9 Intersection-free partition $S$ by $S.z$; /* dense if $f_{2}(z)$$>$0 */

10$S_{sparse}$ $\leftarrow$ Save the sparse part of $S$ as CSR;

11 $S_{dense}$ $\leftarrow$ Save the dense part of $S$ as Bitmap array;

12 $Result_{sparse}$ $\leftarrow$ SparseBMM($R_{CSR}$, $S_{sparse}$);

13 $Result_{dense}$ $\leftarrow$ DenseEC($R_{CSR}$, $S_{dense}$);

return $Result_{sparse}$ $\cup$ $Result_{dense}$;

Algorithm 1 DIM3

###  2.2. Mapping

The mapping step maps columns $x$, $y$, and $z$ to consecutive natural
numbers. This can be achieved with a baseline hash-based algorithm. Given a
column, the algorithm looks up the values of the column one by one in a hash
table. If a value $v$ does not exist in the hash table, it inserts ($v$, the
next consecutive number) into the hash table. As a result, every distinct
value in the column is assigned a natural number. We can repeat this algorithm
for $x$, $y$, and $z$.

This mapping algorithm can be costly. It creates three hash tables for $x$,
$y$, and $z$, and performs up to $2(|R|+|S|)$ hash table lookups and/or
inserts. Compared to the hash join of $R(x,y)$ and $S(z,y)$, which performs
$|R|+|S|$ hash table accesses, the mapping algorithm pays twice as much cost
for hash table visits. This can be a significant additional overhead for the
Join-Project algorithm when the join result size is not much larger than the
input sizes.

In the following, we consider three opportunities to optimize the baseline
algorithm. Then, we extend the mapping to support a wider range of Join-
Project operations.

Optimization 1: Skip Mapping. First of all, DIM3 chooses the classical
solution for highly sparse data sets as shown in Figure 2. This completely
avoids the mapping step. Second, it is possible to skip mapping for columns
that already contain natural numbers. There are two common cases in database
systems. (i) Columns declared with auto increment (e.g., _AUTO_INCREMENT_ in
Oracle, _IDENTITY_ in SQLServer and DB2, _SERIAL_ in PostgreSQL,
_AUTOINCREMENT_ in MySQL) contain consecutive natural numbers. (ii) String
columns can be encoded by dictionary encoding (Kanda et al., 2017) and stored
as natural numbers in database systems (e.g., SAP HANA and MonetDB).

Optimization 2: Reduce Computation with Join Key Mapping. The naïve way to map
$y$ is to use $R.y\cup S.y$ as the mapping input. We observe that only
$R.y\cap S.y$ contributes to the equality join results. Therefore, we can
employ a semi-join like idea, and optimize the mapping of $y$ as follows.
Since $S$ is the smaller table, we first compute the mapping of $S.y$ using a
hash table. Then we map tuples in $R$ using the same hash table. If $y_{k}$
$\in$ $R.y$ but $y_{k}$ $\notin$ $S.y$, then the corresponding $R$ tuple can
be safely discarded because it does not have any matches in $S$. Note that we
choose not to pay the cost of re-scanning $S$ to remove $S$ tuples with
$y_{k}$ $\in$ $S.y$ but $y_{k}$ $\notin$ $R.y$. As shown in Figure 2, the
first and third $R$ tuples are filtered out.

Optimization 3: Optimize Hash Table Performance. When a hash table is larger
than the CPU cache, hash table accesses result in expensive random memory
accesses with poor CPU cache behavior. A hash table visit may probe multiple
locations, and dereference pointers (e.g., in the case of chained hash table),
incurring significant overhead. Therefore, we employ the following designs to
improve the hash table performance in the mapping algorithm. First, we
estimate the hash table size for a column (e.g., based on statistics of the
number of distinct keys). If the size exceeds the last-level CPU cache, we
employ cache partitioning. We use the last $k$ bits of the hash value to
divide the data into $2^{k}$ partitions so that the hash table of each
partition fits into the last-level CPU cache. Then, we compute the mapping for
each partition. Second, we employ a linear probing hash table design to avoid
pointer dereference. We tune the number of slots and the maximum probing
distance to reduce the cost of hash table accesses. If no available slots are
found for a given column value, we employ a stash hash table (Flat_hash_map
(Skarupke, 2017) in our implementation) to store the overflow data.

Supporting Wider Range of Join-Project Operations. We can map not only single
attribute but also multiple attributes to consecutive numbers. For instance,

$\Pi_{a,b,c,d,e}(R(a,b,c,d)\Join_{c,d}S(c,d,e,f))\vspace{-0.03in}$

can be treated as

$\Pi_{x,z}(R(x,y)\Join_{y}S(z,y))\vspace{-0.03in}$

where $R.x=\\{a,b\\}$, $S.z=\\{c,d,e\\}$, and $R.y=S.y=\\{c,d\\}$. In this
way, we can support any combinations of join keys and projection columns,
including operations that contain the join key in the output.

###  2.3. Intersection-Free Partitioning

Figure 3. Comparing the partition methods in DHK (Deep et al., 2020) and DIM3.
(A tuple in a table is displayed as an edge in graphs and an element in
matrices. Red: dense, blue: sparse)

Figure 3 compares the partition methods of DHK (Deep et al., 2020) and DIM3.
DHK makes separate decisions on $x$, $y$ and $z$ according to their degrees
(i.e. the number of tuples of the same attribute value). An $($$x$, $y$$)$
tuple is added to the dense part only if both its $x$ and $y$ attributes are
considered as dense (shown as red color in the figure).

In comparison, we propose an intersection-free partition method111Previous
work proposes an intersection-free partition method in the context of query
enumeration algorithms (Deep et al., 2021). It divides the data into two sets
with roughly equal number of join results. The method optimizes enumeration
delays rather than end-to-end query run times. It is not directly applicable
to the Join-Project operation. as shown in Figure 3. It examines table $S$ and
ensure that all tuples with the same $S.z$ value are in the same partition.
Note that table $R$ will not be partitioned. Essentially, we partition matrix
$S$ by its columns. Our partitioning method has the following benefits.

First, the Join-Project results generated by the sparse and dense parts do not
intersect. Given a result $($$x$,$z$$)$, if $z$ is judged as dense, this
result must be generated by the dense part. Otherwise, it is generated by the
sparse part. In contrast, this property does not hold in DHK. As shown in
Figure 3, the result $($$x_{2}$, $z_{2}$$)$ is generated both in the dense
part by joining $($$x_{2}$, $y_{1}$$)$ and $($$y_{1}$, $z_{2}$$)$, and in the
sparse part by joining $($$x_{2}$, $y_{3}$$)$ and $($$y_{3}$, $z_{2}$$)$.
Therefore, while DHK must deduplicate the results from the two parts, DIM3 can
eliminate this final deduplication step.

Second, DIM3 may apply dense MM to more tuples. DHK considers a tuple as dense
only if both its attributes are judged as dense. In comparison, DIM3 makes the
partitioning decision based solely on $S.z$. Since its dense criteria tend to
be more flexible, DIM3 can employ dense MM under more scenarios, as
illustrated in Figure 3.

Third, the selection based on $z$ simplifies the Join-Project computation. In
DHK, it is costly (spatially) to record the per-tuple density decisions.
Therefore, DHK does not save them. Instead, when processing the sparse part,
DHK uses the degree thresholds to re-compute whether a tuple is dense and
should be skipped. In comparison, DIM3 avoids this complexity. As shown in
Figure 2, table $S$ is divided into the sparse and the dense parts based on
$S.z$. Hence, there is no need to re-evaluate the density criteria any more.

Finally, the cost for computing the partition decision in DIM3 is lower
compared to DHK. DHK makes $|R|+|S|$ decisions on all input tuples. In
comparison, DIM3 makes $|Z|$ decisions on $S.z$. The number of decisions to
make is much smaller. Consider the case where $|X|$=$|Y|$=$|Z|$=$n$. In the
worst case, the cost is $\mathcal{O}(n^{2})$ in DHK, but only $\mathcal{O}(n)$
in DIM3. Moreover, DHK performs binary search to determine the density
threshold, which incurs additional cost.

###  2.4. SparseBMM

The classical hash-based solution is often used to process the sparse part of
the data. We propose a _SparseBMM_ algorithm with two main optimization
techniques, as shown in Algorithm 2.

First, we observe that hash table accesses are often one main cost of the
classical hash-based computation. Interestingly, since the column values are
mapped to natural numbers, the CSR (Compressed Sparse Row) (Kepner and
Gilbert, 2011) format of matrix $\mathbf{S}^{y\times z}$ is essentially a hash
table on the join key $y$ with NO hash conflicts. The original CSR structure
consists of three arrays: $Val[]$, $Col[]$, and $RowPtr[]$. $Val[]$ and
$Col[]$ contain the value and the column index of non-zero elements in the
matrix, respectively. $RowPtr[]$ points to the row starts in $Val[]$ and
$Col[]$. In the case of Join-Project, $Val[]$ contains all 1’s and can be
omitted. Therefore, we have two arrays $Col[]$ and $RowPtr[]$. We employ
matrix $\mathbf{S}^{y\times z}$ as the hash table. Given $y_{k}$, we locate
all non-zero element $\mathbf{S}_{y_{k},z_{j}}$ by visiting entries
$Col[RowPtr[y_{k}]]$ .. $Col[RowPtr[y_{k}$+1]-1]. In fact, $RowPtr$ serves as
the hash bucket header, and $Col[RowPtr[y_{k}]]$ .. $Col[RowPtr[y_{k}$+1]-1]
contain the hash entries in bucket $y_{k}$. In this way, we avoid the hash
function computation and hash conflicts in common hash table designs.

Second, DHK performs deduplication using a $z$-vector. For each $x_{i}$, it
initializes the $z$-vector to all zeros. Then, it checks all $($$x_{i}$,
$y_{k}$$)$s to compute the join results. For every result $($$x_{i}$,
$z_{j}$$)$, DHK increments the corresponding element in the $z$-vector. Hence,
the non-zero elements in the $z$-vector indicate the deduplicated Join-Project
results. However, the initialization cost is $\mathcal{O}(|Z|)$, while the
number of non-zero $z$’s can be small for the sparse part of the data.
Consequently, the initialization of the $z$-vector is often a main cost of the
deduplication computation. We remove this per-$x$ initialization cost with a
monotonically increasing flag for different $x$. As shown in Algorithm 2, the
$SPA$ array is the $z$-vector. We initialize the $SPA$ array only once before
any computation. The $cur_{y}$ and $cur_{z}$ loops (Line 2–2) compute the join
results for the given $cur_{x}$. For a newly computed join result
$($$cur_{x}$, $cur_{z}$$)$, we set $SPA$[$cur_{z}$] to $cur_{x}$. If there are
multiple duplicate $($$cur_{x}$, $cur_{z}$$)$, $SPA$[$cur_{z}$] is set only
once for the first instance. In this way, for the next $cur_{x}+1$, the
previous content of $SPA$ is _automatically_ invalid. This saves the cost of
initializing $SPA$ in every $cur_{x}$ loop iteration.

1

Input: CSR-stored $R_{CSR}$ and CSR-stored $S_{sparse}$

Output: List of result tuples $(x,z)$

2

3SPA[0..$|Z|$-1]=$-1$;

4 for  _$cur_{x}\leftarrow 0$ to $|X|$_ do

5 foreach  _$cur_{y}$ related to $cur_{x}$ in $R_{CSR}$_ do

6 foreach  _$cur_{z}$ related to $cur_{y}$ in $S_{sparse}$_ do

7 if  _SPA[ $cur_{z}$]!=$cur_{x}$_ then

8 SPA[$cur_{z}$]=$cur_{x}$;

9 Result.append($($$cur_{x}$, $cur_{z}$$)$);

10

11

12

13

14 return Result;

Algorithm 2 SparseBMM (for sparse data).

We consider the time and space complexity of SparseBMM. Line 2 of Algorithm 2
runs $|OUT_{J}|$ times. Thus, the time complexity is
$\Theta(|R|+|S|+|OUT_{J}|)$. While this is the same as the classical hash-
based solution, SparseBMM reduces the constant factor, accelerating hash table
visits and deduplication. Moreover, SparseBMM requires $\Theta(|R|+|S|+|Z|)$
space if the final output is consumed by upper level operators. While hash-
based deduplication requires $\Theta(|OUT_{P}|)$ space for the hash table,
SparseBMM allocates only $\Theta(|Z|)$ space for $SPA$, which is often much
smaller than $\Theta(|OUT_{P}|)$.

###  2.5. DenseEC

To compute
$\mathbf{C}_{x_{i},z_{j}}=\sum_{k=1}^{|Y|}\mathbf{R}_{x_{i},y_{k}}\mathbf{S}_{y_{k},z_{j}}$,
standard dense MM enumerates all pairs of $\mathbf{R}_{x_{i},y_{k}}$ and
$\mathbf{S}_{y_{k},z_{j}}$. We observe that the computation can stop early as
soon as there is a non-zero
$\mathbf{R}_{x_{i},y_{k}}\mathbf{S}_{y_{k},z_{j}}$. Therefore, we propose a
DenseEC algorithm to leverage this observation, as listed in Algorithm 3.

In Algorithm 3, the first two _for_ -loops enumerate all the pairs of $R.x$
and $S.z$. Line 3–3 use one of two methods to check if there is any common $y$
between $Bitmap_{x}$ and $Bitmap_{z}$. The first method uses SIMD to compute
the bit-wise _AND_ of $Bitmap_{x}$ and $Bitmap_{z}$ (e.g., using
_mm256_testz_si256). The second method examines each $y$ related to $cur_{x}$
in $Bitmap_{z}$ using a random memory access. If any common $y$ is found, both
methods stop early. Generally speaking, the SIMD method performs better when
the $cur_{x}$ row has a large number of $y$. We determine which method to use
with function $f_{3}$, which will be discussed in Section 2.6.

Apart from the time saving due to early stopping, DenseEC saves memory space
compared to dense MM. DenseEC represents every element as a single bit rather
than a 4-byte integer or floating point value in BLAS packages. Moreover, a
dense MM invocation would require space to save the temporary output matrix.
In comparison, DenseEC never generates the output matrix.

1

Input: CSR-stored $R_{CSR}$ and bitmap array $S_{dense}$

Output: List of result tuples $(x,z)$

2

3for  _$cur_{x}\leftarrow 0$ to $|X|$_ do

4 $Bitmap_{x}$ $\leftarrow$ Save the $y$ in the $cur_{x}$ row as bitmap;

5 foreach  _$cur_{z}$ in $S_{dense}$_ do

6 if  _$f_{3}(m_{x},m_{z},|Y|)$ $>$0_ then /* SIMD is better */

7 if  _SIMD_AND( $Bitmap_{x}$,$Bitmap_{z}$) not all 0_ then

8 Result.append($($$cur_{x}$, $cur_{z}$$)$);

9

10

11 else

12 foreach  _$y$ related to $cur_{x}$ in $R_{CSR}$_ do

13 if  _the $y$th bit in $Bitmap_{z}$ == 1_ then

14 Result.append($($$cur_{x}$, $cur_{z}$$)$);

15 Break;

16

17

18

19

20

21return Result;

Algorithm 3 DenseEC (for dense data).

###  2.6. Evaluation Path Section Functions

We compute the three functions used in the DIM3 algorithm to select different
evaluation paths. Table 1 lists the symbols used in this subsection. All the
listed parameters can be measured in advance.

Table 1. Symbols used in Section 2.6. Symbol |  Description
---|---
$t_{seqR}$ |  time for sequential memory read
$t_{randR}$ |  time for random memory read
$t_{randRW}$ |  time for random memory read-modify-write
$t_{hash}$ |  time of a lookup or insertion to plain hash table
$t_{map}$ |  time to access the optimized hash table for mapping
$t_{ECs}$ |  time of a non-SIMD comparison in DenseEC
$t_{ECd}$ |  time of a SIMD comparison in DenseEC

(a) $f_{1}$ for strategy selection

(b) $f_{3}$ for selecting bitmap comparison method

(c) $f_{2}$ for deciding column density

Figure 4. Effectiveness of the three functions for choosing evaluation paths
in DIM3.

Strategy Selection ($f_{1}$). DHK chooses the classical solution using a rule-
of-thumb condition: $|OUT_{J}|\leq 20\cdot N$, where $N=|R|=|S|$. In
comparison, DIM3 determines whether to use the classical solution or the
hybrid solution with function $f_{1}$. As the classical solution is beneficial
only when the data is sparse, we compare mapping+SparseBMM and the classical
solution to compute $f_{1}$. When $f_{1}>0$, the classical solution is faster
and will be chosen.

$\displaystyle f_{1}(|R|,$
$\displaystyle|S|,|OUT_{J}|)=(T_{mapping}+T_{SparseBMM})-(T_{join}+T_{dedup})$
$\displaystyle=T_{mapping}+(T_{SparseBMM}-T_{join})-T_{dedup}$ (2)
$\displaystyle\approx 2(|R|+|S|)t_{map}+|OUT_{J}|t_{randRW}-|OUT_{J}|t_{hash}$

Here, $T_{mapping}=2(|R|+|S|)t_{map}$ from Section 2.2. To estimate
$T_{SparseBMM}-T_{join}$, we see that generating join results with CSR in
SparseBMM has smaller or similar cost compared to hash joins. Hence, the
difference is mainly the deduplication cost of SparseBMM. This is
$|OUT_{J}|*t_{randRW}$ because SparseBMM performs a random access to the $SPA$
array per join result. Finally, $T_{dedup}=|OUT_{J}|t_{hash}$ because the
classical solution performs a hash table access for deduplicating every join
output tuple.

Bitmap Comparison in DenseEC ($f_{3}$). Line 3 of Algorithm 3 uses function
$f_{3}$ to choose the SIMD or non-SIMD method for comparing $Bitmap_{x}$ and
$Bitmap_{z}$. The bitmaps have $|Y|$ bits. Suppose there are $m_{x}$ and
$m_{z}$ 1’s in $Bitmap_{x}$ and $Bitmap_{z}$, respectively.

In the non-SIMD method, the comparison stops as soon as a check hits a set bit
in $Bitmap_{z}$. The probability that the check hits a set bit is
$p_{s}=\frac{m_{z}}{|Y|}$. The number of checks follows a geometric
distribution with probability $p_{s}$, and the method performs at most $m_{x}$
checks. Hence, the expected number of checks is calculated as:

$Check_{nonsimd}=\sum_{i=1}^{m_{x}}i(1-p_{s})^{i-1}p_{s}+m_{x}(1-p_{s})^{m_{x}}=\tfrac{1-(1-p_{s})^{m_{x}}}{p_{s}}\vspace{-0.05in}$

In the SIMD method, every SIMD comparison checks 256 bits in the two bitmaps.
When the bits at the same position in the two bitmaps are both set, the
comparison returns true and the process stops. The probability that an SIMD
comparison returns true is $p_{d}=1-(1-\frac{m_{x}m_{z}}{|Y|^{2}})^{256}$. The
number of SIMD checks follows a geometric distribution with probability
$p_{d}$, and the method performs up to $\frac{|Y|}{256}$ checks. Hence, the
expected $Check_{simd}$= $\tfrac{1-(1-p_{d})^{\frac{|Y|}{256}}}{p_{d}}$.

Then, we can compute $f_{3}$ as follows. When $f_{3}>0$, the SIMD method is
preferred. Otherwise, the non-SIMD method is selected.

(3)
$f_{3}(m_{x},m_{z},|Y|)=Check_{nonsimd}t_{ECs}-Check_{simd}t_{ECd}\vspace{-0.03in}$

We can reduce the computation overhead of $f_{3}$ as follows. Note that
$m_{x}$ and $|Y|$ are constants in the for-loop at line 3 of Algorithm 3.
$f_{3}(m_{x},m_{z},|Y|)$ is monotonically increasing with $m_{z}$. Therefore,
an optimization is to use binary search to find the threshold value of
$m_{zt}$ so that $f_{3}(m_{x},m_{zt},|Y|)=0$. Then, we can select the SIMD
method if $m_{z}>m_{zt}$. Moreover, we find that multiple $R.x$’s can share
the same $m_{x}$. Since $m_{zt}$ is determined for a given $m_{x}$, we can
cache the pairs of $m_{x}$ and $m_{zt}$, then reuse the computed $m_{zt}$ to
avoid redundant binary searches. In this way, the time for computing all
$f_{3}$ thresholds is at most 1.15ms for the real-world data sets in Section
5, which is less than 0.5% of the total run time.

Dense vs. Sparse Partitions ($f_{2}$). At line 1 in DIM3 (Algorithm 1),
intersection-free partitioning uses function $f_{2}$ to determine if $S.z$
column is dense or sparse.

For column $z$, the number of join results can be estimated as
$\tfrac{m_{z}}{|S|}|OUT_{J}|$, where $m_{z}$ is the number of $S$ tuples in
column $z$. We denote this value as $OUT_{J,z}$. We estimate the cost of
processing column $z$ using either SparseBMM and denseEC.

The cost of SparseBMM for $z$ is computed as follows:

$T_{sparseBMM}=\tfrac{(2|X|+|R|)t_{seqR}+2|R|t_{randR}}{|Z|}+OUT_{J,z}(t_{seqR}+t_{randRW})\vspace{-0.01in}$

The first component computes the cost of the two for-loops at Line 2–3
amortized to one of $|Z|$ columns. The second component estimates the cost of
Line 4–7. For DenseEC, we know that the cost for each pair of $x$ and $z$ from
the above. Then we can sum this up to obtain $T_{denseEC}$=
$\sum_{i=1}^{|X|}min(Check_{nonsimd}t_{ECs},Check_{simd}t_{ECd})$.

Finally, we can compute $f_{2}$ as follows. When $f_{2}>0$, we consider column
$z$ as dense.

(4) $f_{2}=T_{sparseBMM}-T_{denseEC}\vspace{-0.03in}$

Effectiveness of the Functions. Figure 4(a) compares the measured running time
and the estimated running time of the classical solution and
mapping+SparseBMM. In this experiment, we generate random data with column
values in the range of 0 to $10^{7}$. We vary $|R|$=$|S|$ from 5 million to 60
million. The X-axis shows the join result size. The Y-axis reports the running
time. From the figure, we see that the estimated times show consistent trends
compared to the measured times. Function $f_{1}$ models the intersection of
the curves for the two algorithms. We see that the intersection points of the
measured and estimated time curves are very close, showing the effectiveness
of $f_{1}$ for strategy selection.

Figure 4(b) compares the measured running time and the estimated running time
of the SIMD and non-SIMD methods. From the figure, we see that (i) the
estimated values are always smaller than the measured values because certain
runtime overheads, such as the for-loop, are not taken into consideration;
(ii) the estimated curves show similar trends as the measured curves; and
(iii) the intersection points of the two sets of curves are very close.

Figure 4(c) compares the measured running time and the estimated running time
of SparseBMM and DenseEC. From the figure, we see that (i) the estimated
curves show similar trends as the measured curves; and (ii) the intersection
points of the two sets of curves are close. This demonstrates the
effectiveness of using function $f_{2}$ for intersection-free partitioning.

###  2.7. Time Complexity Analysis

Based on the above discussion in Section 2.6, we have the following:

$\displaystyle Complexity_{SparseBMM}$
$\displaystyle=\Theta(|R|+|S|+|OUT_{J}|)$ $\displaystyle
Complexity_{DenseEC_{nonsimd}}$
$\displaystyle=\Theta(\sum_{i=1}^{|X|}\sum_{j=1}^{|Z|}T_{Check\\_nonsimd}(x_{i},z_{j}))$
$\displaystyle Complexity_{DenseEC_{simd}}$
$\displaystyle=\Theta(\sum_{i=1}^{|X|}\sum_{j=1}^{|Z|}T_{Check\\_simd}(x_{i},z_{j}))$

Note that $T_{Check\\_nosimd}(x_{i},z_{j})$ and
$T_{Check\\_simd}(x_{i},z_{j})$ are the actual numbers of checks given
$(x_{i},z_{j})$, which are determined by the input data distribution.

Worst-Case Costs. We compute the worst-case costs. Suppose that input tables
$R$ and $S$ do not have duplicates. Then, for SparseBMM, it is clear that
$|OUT_{J}|<|X||Y||Z|$. For DenseEC, early stopping never happens in the worst
cases. Hence, $T_{Check\\_nonsimd}(x_{i},z_{j})=\Theta(m_{x_{i}})$ and
$T_{Check\\_simd}(x_{i},z_{j})=\Theta(\frac{|Y|}{SIMD\\_bit\\_length})$.

Therefore, the worst case time complexities are as follows:

$\displaystyle Complexity_{SparseBMM}$ $\displaystyle=\mathcal{O}(|X||Y||Z|)$
$\displaystyle Complexity_{DenseEC_{nonsimd}}$
$\displaystyle=\mathcal{O}(|R||Z|)$ $\displaystyle
Complexity_{DenseEC_{simd}}$
$\displaystyle=\mathcal{O}(\frac{|X||Y||Z|}{SIMD\\_bit\\_length})$

Average-Case Costs. It is difficult to compute the precise formula for
average-case costs because the time complexities are affected by the path
selection decisions and the density of the input tables. In the following, we
analyze the costs under a simplifying assumption of uniform density
distribution in the input tables.

We assume that the non-zeros in matrices $\mathbf{R}_{x,y}$ and
$\mathbf{S}_{y,z}$ are uniformly randomly distributed, and the densities
(probability) of non-zero values are $\alpha$ and $\beta$, respectively. Then,
we have $|OUT_{J}|=\alpha\beta|X||Y||Z|$, $m_{x}=\alpha|Y|$ and
$m_{z}=\beta|Y|$. We also assume that $|OUT_{J}|>max(|R|,|S|)$.

Since all the rows and columns of the input data are of the same density,
$f_{2}$ considers the entire $S$ as dense or sparse. Similarly, $f_{3}$
selects the same execution path for all pairs of $(x,z)$. That is, all data
will be processed by the same algorithm. We can use the expected numbers of
checks ($Check_{nonsimd}$ and $Check_{simd}$) computed in the previous
subsection to estimate $T_{Check\\_nosimd}$ and $T_{Check\\_simd}$. The above
formula can be rewritten as follows:

$\displaystyle Complexity_{SparseBMM}$
$\displaystyle=\Theta(\alpha\beta|X||Y||Z|)$ $\displaystyle
Complexity_{DenseEC_{nonsimd}}$
$\displaystyle=\Theta(\frac{1-(1-\beta)^{\alpha|Y|}}{\beta}|X||Z|)$
$\displaystyle Complexity_{DenseEC_{simd}}$
$\displaystyle=\Theta(\frac{1-(1-\alpha\beta)^{|Y|}}{1-(1-\alpha\beta)^{SIMD\\_bit\\_length}}|X||Z|)$

The classical hash solution shares the same complexity with SparseBMM (i.e.,
$\Theta(\alpha\beta|X||Y||Z|)$).

The complexity of dense MM is $\Theta(|X||Y||Z|\gamma^{\omega-3})$, where
$\gamma=min\\{|X|,|Y|,|Z|\\}$ (Deep et al., 2020). Though the best known
$\omega$ value can be 2.373 (Gall and Urrutia, 2018), these theoretical MM
algorithms are often considered impractical (Le Gall, 2012; Pegoraro et al.,
2020). And in the widely used packages for MM (e.g. Intel MKL), $\omega=3$.

Estimating the Costs of Two Special Cases. We consider two specific cases to
better understand the average-case costs. First, $\alpha=\beta=1$. That is,
the input matrices are full. In this very dense case, the time complexity of
SparseBMM is $\Theta(|X||Y||Z|)$, while the costs of DenseECnonsimd and
DenseECsimd are both $\Theta(|X||Z|)$. Thanks to the early stopping technique,
DenseEC is significantly faster than the other algorithms (including Dense
MM).

Second, $\alpha=\beta=\frac{1}{|Y|}$. This is a very sparse case. The
complexity of SparseBMM is $\Theta(\frac{|X||Z|}{|Y|})$. No-simd will be
chosen for the very sparse case. The complexity of DenseECnonsimd is
$\Theta(|X||Z|)$. Therefore, SparseBMM is the best solution.

##  3\. Partial Result Caching

(a) HetRec data set. (b) MV2 data set.

Figure 5. CDF of computation cost.

We observe that computing Join-Project results for different $z$ values can
take widely different amounts of time for real-world data sets. Figure 5
depicts the CDF (Cumulative Distribution Function) of computation cost for two
representative real-world data sets. In HetRec, the top 1% of $z$ values take
60% of the total time. In MV2, the top 10% of $z$ values contribute to 23% of
the total cost. This motivates us to study caching the results of a subset of
$z$ values to improve Join-Project performance. Note that our technique is
different from materialized views (Larson and Yang, 1985) or query result
cache (Larson et al., 2004), where the full query results are cached.

DIM3 with Cached Partial Results. We choose to cache results based on $z$
values because the intersection-free partitioning method divides table $S$
according to the $S.z$ values. Hence, it is easy to integrate the cached
partial results into the DIM3 algorithm.

Suppose $Z_{cached}$ is the subset of ${S.z}$, whose Join-Project results are
cached. That is, the cached partial results are $R_{cached}$ =
$\\{$$($$x,z$$)$$|$$($$x,z$$)$$\in\Pi_{x,z}(R(x,y)\Join_{y}S(z,y))\land z\in
Z_{cached}\\}$. In Figure 2, we can simply omit any $z\in Z_{cached}$ when
generating $S_{sparse}$ and $S_{dense}$, then keep the other steps of DIM3
unchanged. Finally, we output $R_{cached}$ in addition to the results computed
by DIM3.

Caching Score. Given a caching space budget $B$ for a Join-Project query, we
rank $z$ values for caching with the following score:

$CacheScore(z)=\tfrac{cost(z)}{space(z)}\vspace{-0.05in}$

$space(z)$ is the cache space required to store the $($$x$,$z$$)$ results for
the given $z$. The cache content is $($$z,size,x_{1},x_{2},...,x_{k}$$)$. When
$k$ is small, $size$=$k$¿0, and we cache the original results. When
$k$¿$0.5|X|$, we store the complement set of $x$’s to save space, and set
$size=k-|X|$¡0. Hence, $space(z)$=$2$+$min(k,|X|-k)$.

$cost(z)$ is the time for computing $($$x$,$z$$)$ results for the given $z$.
This is the benefit from caching $z$. Since the processing related to $z$ is
distributed into many inner loops in SparseBMM and DenseEC, we cannot directly
measure $cost(z)$. Instead, we collect statistics and estimate the cost as
follows:

$cost(z)=n_{sparse}(t_{seqR}+t_{randRW})+n_{simd}t_{ECd}+n_{nonsimd}t_{ECs}\vspace{-0.03in}$

where $n_{sparse}$ is the number of times that $z$ is checked in the inner
loop of the SparseBMM algorithm, $n_{simd}$ and $n_{nonsimd}$ are the number
of times that $z$ is encountered in the SIMD and non-SIMD part of the inner
loop of the DenseEC algorithm, respectively.

$CacheScore(z)$ shows the benefit per unit space for caching $z$. The higher
the $cost(z)$, the lower the $space(z)$, the more beneficial to cache $z$.
Interestingly, a $z$ value with high $cost(z)$ may produce a large number of
results. This actually leads to small $space(z)$ when the complement result
set is saved.

Practical Considerations. We discuss several practical issues for using the
partial result caching. First, database users can enable Join-Project caching
dynamically. We see in experiments that the statistics collection and
computation for $cost(z)$ and $space(z)$ do not incur significant overhead for
DIM3. Hence, after caching is enabled, the first run of DIM3 can compute
$CacheScore(z)$ and populate the partial result cache. Then, subsequent runs
of DIM3 on the same tables can leverage the cached results to improve
performance. Second, when the underlying tables are modified by
insert/delete/update, we can simply invalidate the cache and let the next run
of DIM3 to re-populate the result cache. Note that updating the cached results
(similar to the view maintenance problem (Blakeley et al., 1986)) is beyond
the scope of this work. Finally, we can use the cached result to support
filter predicates on $x$ and/or $z$. However, if there are filter predicates
on $y$, we have to run DIM3 without the cached result.

##  4\. Support for Join-Op Query Types

In the above, we focus on the Join-Project operation. Join-Project is a
special case of Join-$op$ queries, where $op$ is a common SQL operation. An
interesting question arises: Is it possible to apply DIM3 to Join-$op$ queries
in general? We consider $op$ other than projection in the following:

  * •

_Group-by Aggregation_ : The group-by operation inherently removes duplicates.
A join followed by a group-by aggregation operation, which we call Join-
Aggregate for simplicity, implicitly performs a Join-Project operation. We
describe how to extend DIM3 to support Join-Aggregate in Section 4.1.

  * •

_Join_ : Multiple join operations are followed by a duplicate eliminating
projection. One strategy to evaluate this type of queries is to employ Join-
Project to deduplicate the results of the last join operation. But can we do
better? We consider how and when to push the deduplication down in the query
plan of MJP in Section 4.2.

  * •

_Selection or Sorting_ : As Join-Selection or Join-Sorting do not require
deduplication, there is no need to employ Join-Project for these types of
queries in general. However, in special cases where there are too many
duplicates, an alternative evaluation strategy can be more efficient. We
compute Join-Project and keep the duplicate count for each generated join
result tuple. Then, the selection or sorting operation processes the much
smaller, deduplicated join result, thereby achieving better performance.
Finally, we output the correct number of duplicates based on the per-tuple
duplicate counts.

  * •

_Intersection/Difference/Union_ : First, the computation of intersection is
similar to a join operation. Hence, for Join-Intersection, we can employ MJP.
Second, set difference can be evaluated as a left-outer join followed by
deduplication. We can modify DIM3 to compute Outer-Join-Project for Join-
Difference. To support outer-joins, DIM3 can be extended with a bitmap for $x$
($z$). It sets a bit if the corresponding $x$ row ($z$ column) has generated
join results. In this way, the modified DIM3 can compute $x$ ($z$) with no
matches for outer-joins. Finally, Join-Union requires the deduplication of the
outputs from two joins. We can push the deduplication operation down in the
query plan, and apply similar considerations as in Section 4.2.

###  4.1. Join-Aggregate

Our DIM3 algorithm can be applied to Join-Aggregate operations with slight
modifications. Without loss of generality, we divide Join-Aggregate operations
into two categories based on the group by attributes: i) group-by attributes
are from both tables, and ii) group-by attributes come from one table.

Group-by Attributes from Both Tables. For instance, given tables $R(x,y,v)$
and $S(z,y,u)$, we want to compute

${}_{x,z}\mathcal{G}_{aggregate(f(R.v,S.v))}(R(x,y,v)\Join_{y}S(z,y,v))\vspace{-0.02in}$

This task is similar to the original Join-Project operation. The main
difference is that it computes $aggregate(f(R.v,S.u))$ on the join results
with the same $($$x$, $z$$)$ rather than deduplicating the results.

We modify DIM3 to support this task. In SparseBMM, we change the $SPA$ array
to contain the aggregate for each $cur_{z}$. $SPA$ is initialized in each
outer-loop iteration. Line 5–7 is modified to accumulate the aggregate for
group $($$cur_{x}$, $cur_{z}$$)$. In DenseEC, bitmaps cannot support the
aggregates. Therefore, we use a plain dense MM, while computing aggregates for
each pair of $($$cur_{x}$, $cur_{z}$$)$.

Group-by Attributes from One Table. Given tables $R(x,y)$ and $S(z,y)$, we
want to compute the following query. If the group-by attribute is from $S$, we
switch table $R$ and $S$.

${}_{x}\mathcal{G}_{aggr(z)}(R(x,y)\Join_{y}S(z,y))\vspace{-0.02in}$

We can rewrite the query as follows:

${}_{x}\mathcal{G}_{aggr"(z^{\prime})}(R(x,y)\Join_{y}(_{y}\mathcal{G}_{aggr^{\prime}(z)}(S(z,y))))\vspace{-0.02in}$

We first compute the group-by aggregate on $S$ with $y$ as the group-by key.
If $aggr$ is $sum$, $min$, or $max$, then $aggr^{\prime}$ and $aggr"$ are the
same. For $count$, $aggr^{\prime}$ is $count$ and $aggr"$ is $sum$. For $avg$,
the $aggr^{\prime}$ consists of both $sum$ and $count$. Then $aggr"$
accumulates the two components, and finally computes a division to obtain the
$avg$. We denote the resulting table as $SG(y,z^{\prime})$, where $z^{\prime}$
is the intermediate aggregate value(s) for $y$. Note that $|SG|=|Z|$ is often
much smaller than $|S|$. $SG$ is highly sparse because there is no duplicate
$y$’s in the table. Therefore, we employ the classical hash-based algorithm to
join $R$ and $SG$ then compute the final group-by aggregates.

###  4.2. MJP (Multi-Way Joins with Projection)

So far, we have been focusing on Join-Project on _two_ tables. In this
subsection, we study deduplication on the results of joining _multiple_
tables. This is also called conjunctive queries with projection (Deep et al.,
2021). For example, the line join projection with $n$ tables is expressed as:

$\Pi_{x_{1},x_{n+1}}(R_{1}(x_{1},x_{2})\Join_{x_{2}}R_{2}(x_{2},x_{3})\Join_{x_{3}}...\Join_{x_{n}}R_{n}(x_{n},x_{n+1}))$

Figure 6 illustrates the query plan tree for evaluating a MJP query. The
deduplication operation ($\Pi$) can be pushed down to after each join
operation. While it incurs extra overhead, deduplication reduces the
intermediate result size, thereby reducing the cost of subsequent join
operations. There are two baseline execution plans. The first plan computes
all the joins and then deduplicate the final join results. The second plan
deduplicates the results immediately after each join. However, it is easy to
construct cases to show neither plan is optimal. Therefore, we need to
judiciously place the deduplication operations into the query plan tree.

In the following, we develop a DP (Dynamic Programming) algorithm to find the
optimal query plan. To limit the scope of our investigation, we make the
following assumptions.

  1. (1)

_The join order is given by the query optimizer._ We focus on the
deduplication placement problem. For ease of presentation, we number the
tables in the join order as $R_{1}$, …, $R_{k}$. The problem of optimizing
both join order and deduplication placement is beyond the scope of this work.

  2. (2)

_The query plan is in the form of a left-deep tree(Schneider and DeWitt, 1990;
Ioannidis and Kang, 1991)._ Left-deep or right-deep trees are widely used in
RDBMSs to process multi-way joins. If a solution produces left-deep trees, it
is easy to modify it to support right-deep trees.

  3. (3)

_Estimated $|OUT_{Ji}|$ and $|OUT_{Pi}|$, where $i$=2,…,$k$, are available_.
(Please refer to recent work on cardinality esitmation for details (Qiu et
al., 2021; Cai et al., 2019; Hertzschuch et al., 2021; Leis et al., 2017).)
Here, $|OUT_{Ji}|$ = $|R_{1}\Join\cdots\Join R_{i}|$ denotes the size of the
intermediate results after joining the first $i$ tables without deduplication.
$|OUT_{Pi}|$ = $|\Pi(OUT_{Ji})|$ denotes the size of the intermediate results
after deduplication. For simplicity, we define $|OUT_{P1}|$ = $|OUT_{J1}|$ =
$|R_{1}|$.

  4. (4)

_Adding deduplication to the $i$-th join reduces the final join result by a
factor of $\tfrac{|OUT_{Pi}|}{|OUT_{Ji}|}$_. That is,
$|\Pi(R_{1}\Join\cdots\Join R_{i})\Join\cdots\Join
R_{k}|\approx\tfrac{|OUT_{Pi}|}{|OUT_{Ji}|}|OUT_{Jk}|$. This is reasonable
since the join input size to the $(i+1)$-th join is reduced by a factor of
$\tfrac{|OUT_{Pi}|}{|OUT_{Ji}|}$. This assumption simplifies the estimation of
the final result size after inserting a deduplication operation. More precise
estimation requires collecting more statistics, and may incur much higher
cost.

Figure 6. The deduplication placement problem.

We use $DP_{i}$ to denote the optimal time for computing
$\Pi(R_{1}\Join\cdots\Join R_{i})$. We observe that if we already add
deduplication to the $i$-th join, the deduplicated result $OUT_{Pi}$ does not
change if more deduplication operations are added to the joins before the
$i$-th join. Therefore, we have the following equations.

$\displaystyle DP_{i}=\begin{split}\left\\{\begin{array}[]{ll}0,&i=1\\\
\min_{j=1}^{i-1}(DP_{j}+\sum_{h=j+1}^{i-1}JoinCost_{h}+JPCost_{i}),&i>1\end{array}\right.\end{split}$

The formula inside $min$ computes the case where a deduplication is added to
the $j$-th join and there is no deduplication from the $(j+1)$-th to
$(i-1)$-th join. The total cost of this case has three components: (i)
$DP_{j}$, which is the optimal time for computing $OUT_{Pj}$; (ii) the cost of
the subsequent joins, i.e., the $(j+1)$-th to $(i-1)$-th join; and (iii) the
cost of the final join project operation. Note that when $j$=1, there is no
deduplication before the $i$-th join.

Based on assumption (3), we estimate the input table size for component (ii)
and (iii). Specifically, $JoinCost_{h}$ is the cost of joining
$\Pi(R_{1}\Join\cdots\Join R_{j})\Join\cdots\Join R_{h-1}$ and $R_{h}$. The
size of the former is estimated as
$\tfrac{|OUT_{Pj}|}{|OUT_{Jj}|}|OUT_{Jh-1}|$. Moreover, $JPCost_{i}$ is the
cost of the Join-Project operation at the $i$-th join. The sizes of the two
input tables to the Join-Project operation are
$\tfrac{|OUT_{Pj}|}{|OUT_{Jj}|}|OUT_{Ji-1}|$ and $|R_{i}|$. The intermediate
join result in the Join-Project operation can be estimated as
$\tfrac{|OUT_{Pj}|}{|OUT_{Jj}|}|OUT_{Ji}|$ and the final output size is
$|OUT_{Pi}|$. Given the sizes of the input tables, the intermediate join
result, and the final output, we can use the formulas in Section 2.6 to
estimate the cost of the Join-Project operation.

For a MJP query on $n$ tables, the path to get $DP_{n}$ in the DP process
gives the optimal deduplication placement. The time complexity of this
algorithm is $\Theta(n^{3})$. As $n$ is often not large, the overhead of this
DP algorithm is small.

##  5\. Performance Evaluation

In this section, we evaluate the performance of our proposed solutions using
both real-world and synthetic data sets.

###  5.1. Experimental Setup

Machine Configuration. All experiments are performed on a machine with Intel
Core i7-9700 CPU (3.00GHz, Turbo Boost 4.70 GHz, 8 cores/8 threads, 12MB last
level cache) and 32 GB RAM, running Ubuntu 18.04.5 LTS with Linux 5.4.0-81
kernel. All code is written in C/C++ and compiled with _g++_ 7.5.0 using
_–std=c++11_ , _-O3_ , and _-mavx_ flags. MKL (Intel Math Kernel Library) is
used in DHK. By default, we run single-threaded experiments. In the parallel
experiments, we use OpenMP for parallelization.

Datasets. We use six real-world datasets with different input size,
$|OUT_{J}|$, and $|OUT_{P}|$ in our experiments, as shown in Table 2. Amazon
(Leskovec et al., 2007) data set records the frequently co-purchased products
on Amazon website. The Join-Project queries help to find potential products
that can be co-purchased. Slashdot (Leskovec et al., 2009) is a technology-
related news website. The data set contains friend/foe links between users of
Slashdot. The query computes the indirectly connected pairs of users. The data
set and the query of HetRec (Cantador et al., 2011) follow the motivating
example in Section 1. MV1 and MV2 are two MovieLens (Harper and Konstan, 2015)
data sets. They contain user ratings for movies. Similarly, Jokes (Goldberg et
al., 2001) contains user ratings for jokes. The Join-Project queries on
MovieLens and Jokes data sets find users that have rated the same objects.
Friendster (Yang and Leskovec, 2015b) is a social network dataset that
contains the retweet relationship between users. We use this data set to study
MJP, which computes the multi-hop connections between users.

We also conduct experiments on the TPC-H data set with SF=10. A representative
query joins _LineItem_ and _Orders_ , then projects on _CUSTKEY_ and _SUPPKEY_
to attain the purchase relationship between customers and suppliers. However,
since this join is a primary-foreign key join, $|OUT_{J}|$ is similar to the
size of _LineItem_. Both DIM3 and DHK choose the classical solution, showing
the same performance. Hence, we do not study TPC-H further.

In contrast, $|OUT_{J}|$ and $|OUT_{P}|$ in the other data sets are much
larger than their input sizes. The higher the $pnz$, the denser the dataset,
and thus the more intermediate join results. As we do not assume any special
sort order in the input tables of Join-Project, we randomly shuffle the data
sets before the experiments.

Table 2. Real-world data sets used in experiments.

Data set | $|R|$ | $|S|$ | $pnz(R)$ | $pnz(S)$ | $|OUT_{J}|$ | $|OUT_{P}|$
---|---|---|---|---|---|---
Amazon | 1.2M | 1.2M | 0.0018% | 0.0018% | 14M | 11M
Slashdot | 905K | 905K | 0.02% | 0.02% | 118M | 81M
HetRec | 487K | 438K | 0.02% | 0.03% | 125M | 34M
MV1 | 500K | 500K | 2.2% | 2.2% | 204M | 29M
MV2 | 1M | 1M | 4.5% | 4.5% | 816M | 35M
Jokes | 617K | 617K | 25% | 25% | 10B | 622M
Friendster | 1.8B | $4.2\times 10^{-7}$ | – | –

Note: $pnz(M)$ is the percentage of non-zero elements in matrix $M$.

Solutions to Compare. We compare two categories of solutions: stand-alone
Join-Project algorithms, and full-fledged RDBMSs.

We compare the following stand-alone implementations. (1) _Classical_ : The
classical solution performs the Radix join (Balkesen et al., 2013) then a
hash-based deduplication using a flat_hash_map (Skarupke, 2017). (2) _MKL_ :
The state-of-the-art dense MM in Intel MKL is used to evaluate Join-Project.
We implement the baseline mapping algorithm as described in Section 2.2. (3)
_DHK_ : We obtained the DHK code from the authors of the DHK paper (Deep et
al., 2020). Algorithm 3 in DHK chooses the classical solution when
$|OUT_{J}|\leq 20\cdot N$, where $N=|R|=|S|$. We find this feature is missing
in the DHK code, and implement the feature. The code assumes the input
contains consecutive natural numbers. Hence, we add the baseline mapping
algorithm. The code leaves the final deduplication unimplemented. Thus, we add
a deduplication step that checks every result $($$x$,$z$$)$ from the sparse
part against the matrix $\mathbf{C}^{x\times z}$ computed in the dense part.
(4) _DIM 3_: Our proposed solution follows the description in Section 2. We
also study the individual components of DIM3. For the MM component, we compare
DenseEC, SparseBMM, MM in MKL, and a sub-cubic MM (Benson and Ballard, 2015)
based on the Strassen algorithm (Strassen, 1969).

In addition to the stand-alone algorithms, we compare DIM3 with four full-
fledged RDBMSs. (5) _DBMSX_ : one of the best performing commercial RDBMSs.
(6) _PostgreSQL_ version 13.3 and (7) _MariaDB_ version 10.5.11: two popular
open-source RDBMSs. (8) _MonetDB_ version 11.39.17: a representative
analytical main memory RDBMS. We set the configuration parameters of the
RDBMSs to ensure that they make full use of the memory. For each experiment,
we run the same query on the target data set twice. The first run warms up the
RDBMSs so that the input tables are loaded into main memory. Then we measure
the performance of the second run.

Our comparison between _DIM 3_ and RDBMSs is meaningful. One concern is that
RDBMSs pay additional cost, including socket communication, SQL parsing, query
optimization. (Note that we compute a final count to avoid returning a large
number of query results from RDBMSs.) We quantify this additional cost by
measuring the execution time of the same queries on a very small data set. The
result is 1–3ms. In comparison, our reported run times on RDBMSs are from 7s
to about 1 hour. Hence, the extra cost of 1–3ms is negligible. Our reported
run times indeed correspond to the cost of Join-Project query processing in
RDBMSs.

Unless otherwise noted, we focus on single-threaded performance in the
experiments. The scalability results report multi-threaded performance to show
that our proposed algorithms are amenable to parallelization. Partial result
caching is disable by default. Every experiment is run five times. We report
the average of the five runs.

###  5.2. Evaluation for Join-Project Operations

(a) Comparison with stand-alone Join-Project algorithms.

(b) Comparison with RDBMSs.

Figure 7.  Join-Project on real-world datasets. (By default, the algorithms
count then discard the result tuples. Subscript ’M’ denotes a version of an
algorithm that materializes the result tuples in memory. ’X’ shows when
MKL/Classical runs out of memory and fails. We label speedups of DIM3 over DHK
in (a), and over the best RDBMS solution in (b).)

Performance of Stand-alone Solutions on Real-World Data Sets. Figure 7(a)
compares DIM3 with stand-alone Join-Project algorithms on real-world data
sets. We see that DIM3 achieves the best performance among all stand-alone
algorithms with or without result materialization. Compared to _DHK_ , the
state-of-the-art algorithm, DIM3 achieves 2.3$\times$-18$\times$ improvements.
For all the six data sets, DIM3 chooses the hybrid strategy. In contrast, for
Amazon, DHK chooses the classical solution. The speedup of DIM3 over DHK for
Amazon shows that our strategy selection function $f_{1}$ is more accurate
than the rule-of-thumb condition of DHK.

Moreover, DIM3 has smallest memory footprints among all algorithms, as shown
in Table 3. We use ’ _/usr/bin/time -v_ ’ to measure the peak memory usage
(i.e., maximum resident set size). Typically, the final Join-Project results
will be consumed by upper-level operators, and therefore we do not allocate
space for storing the final results. Compared to _Classical_ , when the data
sets are dense, DIM3 saves 80%–99% memory because it chooses the hybrid
strategy and saves the space required by hash-based deduplication in
_Classical_. The memory usage of _Gemm(MKL)_ is mainly determined by the
matrix sizes. MKL runs out of memory for the two sparsest data sets (i.e.,
Amazon and Slashdot). When the datasets are dense, DIM3 saves 68%–99% memory
of _Gemm(MKL)_ because DenseEC saves the space of the output matrix, and uses
one bit per element rather than 4-byte integers in MKL. As for _DHK_ , one
main source of its memory usage is the input and output matrices for dense MM
invocation. In comparison, DenseEC significantly saves this memory.

Comparison with RDBMSs on Real-World Data Sets. Figure 7(b) compares DIM3 with
RDBMSs. We see that DIM3 outperforms all the RDBMSs. For data sets with
$|OUT_{J}|\gg|R|+|S|$, DIM3 achieves one to three orders of magnitudes of
speedups.

We examine the query plans generated by the RDBMSs and see that they all
essentially employ the _Classical_ solution, i.e., a join followed by a
deduplication of the intermediate results. Among the RDBMSs, MonetDB is the
best when memory is sufficient. However, its performance drops sharply when
the required space exceeds the memory size (for MV2 and Jokes). In the case of
Jokes, all the RDBMSs fail to perform the deduplication operation entirely in
memory. The queries take much longer time because of the disk I/Os for storing
and accessing the intermediate results. We stop the queries when they took
longer than 1 hour. The figure reports the lower bound (i.e., 3600s) for
RDBMSs on the Jokes data set.

As RDBMSs implement the classical solution, we focus on the comparison with
stand-alone solutions in the rest of this subsection.

Table 3. Memory usage of stand-alone Join-Project algorithms. | Amazon | Slashdot | HetRec | MV1 | MV2 | Jokes
---|---|---|---|---|---|---
DIM3 | 86MB | 64MB | 32MB | 39MB | 72MB | 47MB
DHK | 426MB | 70MB | 168MB | 68MB | 132MB | 2.4GB
Gemm(MKL) |  $>$32GB |  $>$32GB | 7.1GB | 211MB | 227MB | 2.4GB
Classical | 426MB | 3.0GB | 1.5GB | 789MB | 1.5GB | 24GB

(a) Performance with 8 threads. (b) Scalability on the HetRec data set. (c)
Scalability on the MV1 data set.

Figure 8. Multi-threaded performance.

Scalability. We implement the multi-threaded DIM3 with OpenMP by modifying
around 20 lines of code. Specifically, we parallelize the outer _for_ -loop in
SparseBMM and DenseEC. Since every outer loop iteration focuses on a specific
$x$, there is no dependence or contention among the execution of different
outer-loop iterations. While more involved changes may better parallelize the
algorithm, we find that such simple modifications can already achieve
promising results.

We compare DIM3 with _DHK_ using 8 threads in Figure 8(a). We see that DIM3
achieves 2.6$\times$-20$\times$ speedups over DHK. Figure 8(b)–(c) show the
scalability of the two algorithms. The Y-axis reports the speedup compared to
the single-threaded execution of the same algorithm. We see that DIM3 shows
good scalability as the number of threads increases from 1 to 8.

(a) Varying $n$ ($|X|=|Y|=|Z|=n$).

(b) Zipf on $R.x$ and $S.z$.

(c) Zipf on $R.y$ and $S.y$.

(d) R-mat.

(e) Varying selectivity of $R.x$ and $S.z$.

(f) Zoom in to bottom left of (e).

Figure 9. Sensitivity analysis of DIM3.

Sensitivity Analysis with Synthetic Data Sets. We use synthetic data sets to
investigate the Join-Project performance for a wide range of situations. The
default parameters in these experiments are as follows. All input columns are
generated with uniform distributions. $|R|=|S|=10^{6}$, and
$|X|=|Y|=|Z|=n=10^{4}$. We study the performance impact varying $n$, data
skews, and selectivity in Figure 9. Overall, we see that DIM3 performs the
best in all the cases.

_(1) Varying $n$_. Figure 9(a) varies $n$, the number of distinct $x$/$y$/$z$,
while keeping the number of non-zero elements (i.e. $|R|$ and $|S|$) fixed. As
$n$ increases, the input data sets become more and more sparse, and
$|OUT_{J}|$ decreases from $10^{9}$ to $10^{6}$. _Classical_ runs faster for
more sparse data sets. _Gemm(MKL)_ sees a sharp increase of run time as the
matrix dimensions (i.e., $n$) increase. DIM3 wisely selects evaluation
strategies based on data density. It employs DenseEC for all data when
$n$¡$10^{4}$, SparseBMM when $n\in[10^{4},4\times 10^{5}]$, and the classical
solution when $n$$>$$4\times 10^{5}$. In this way, DIM3 achieves the best
performance in all cases.

_DHK_ has good performance in the middle range in Figure 9(a) and switches to
Classical in the higher range. However, its performance is worse than
_Gemm(MKL)_ in the lower range. To understand the reason, we look into the
_DHK_ code and find that _DHK_ restricts the degree thresholds (i.e.,
$\Delta_{1}$ and $\Delta_{2}$ (Deep et al., 2020)) to have
$\Delta_{2}=\tfrac{|R|\Delta_{1}}{|OUT_{P}|}$ in order to reduce the threshold
search space. Unfortunately, this narrows the range of considered thresholds.
For the lower range when the data is very dense, _DHK_ fails to put all data
into the dense part, and thus performs worse than _Gemm(MKL)_. In comparison,
DIM3 does not suffer from this problems. DIM3 makes the partition decision
based on $z$, which is simple and less error-prone to compute.

_(2) Varying data skews_. In Figure 9(b), we generate $R.x$ and $S.z$ using
the Zipf distribution, and vary the parameter $\alpha$ from 0 to 1. As $R.x$
and $S.z$ become more skewed, $|OUT_{J}|$ stays around $10^{8}$, while the
deduplicated result size $|OUT_{P}|$ decreases from $6\times 10^{7}$ to
$1.4\times 10^{7}$. DIM3 partitions the more skewed $z$’s into the dense part
as $\alpha$ increases, and achieves a speedup of $\sim$1.5$\times$ compared to
_DHK_.

In Figure 9(c), we generate the join attribute, $R.y$ and $S.y$, using Zipf
distribution, and vary $\alpha$ from 0 to 1. As $R.y$ and $S.y$ become more
skewed, the number of intermediate join results ($|OUT_{J}|$) increases
sharply from $10^{8}$ to $4.3\times 10^{9}$, and the data sets become
increasingly dense. _Classical_ runs slower as the data sets become denser.
_Gemm(MKL)_ sees a flat curve because the matrix dimensions are the same.
_DHK_ suffers from the threshold search space problem for very dense data
sets. DIM3 achieves the best performance in all cases. There is a dip in the
DIM3 curve at $\alpha$=1 because early stopping effectively reduces DenseEC
computation in this case.

In Figure 9(d), we use R-mat (Chakrabarti et al., 2004) to generate skew data
so that the two columns in a table are correlated. We use the R-mat parameters
in Graph500 (Murphy et al., 2010) (i.e., $a=0.57$, $b=0.19$, $c=0.19$, and
$d=0.05$). We generate a directed graph with $n$ vertices and $|R|$ edges. The
Join-Project finds 2-hop paths in the graph. As R-mat requires $n$ to be a
power of 2, we set $n=2^{14}$, which is close to the default $10^{4}$. We vary
$|R|$ from $10^{6}$ to $10^{7}$. From Figure 9(d), we see that DIM3 performs
significantly better than the other algorithms. As $|R|$ increases, the data
set becomes increasingly dense. _Classical_ , _Gemm(MKL)_ , and _DHK_ curves
show similar trends as in Figure 9(c).

_(3) Varying selectivity_. In Figure 9(e), we consider the cases where there
are filtering predicates on both $R.x$ and $S.z$. We vary their selectivity at
the same time from 0 to 1. Figure 9(f) zooms in to the bottom left of the
Figure 9(e). As the selectivity decreases, both the input sizes and the output
size decrease. Hence, all the algorithms run faster. From the figure, we see
that DIM3 performs the best in all cases. Note that when the selectivity is
very small (less than 0.03), DIM3 switches to the classical solution. When the
selectivity is less or equal than 0.2, _DHK_ switches to the classical
solution.

(a) Mapping step varying $n$.

(b) Mapping step varying $|R|=|S|$.

(c) DenseEC varying $n$.

(d) SparseBMM varying $n$.

Figure 10. Effectiveness of DIM3 Components.

Effectiveness of DIM3 Components. In the following, we use synthetic data sets
with the same default parameters as the above.

_(1) Mapping step_. We compare our optimized mapping solution with the
baseline using std::unordered_map, and an improved baseline using
flat_hash_map. In Figure 10(a), we fix $|R|$=$|S|$=$10^{6}$ while varying $n$
from $10^{3}$ to $10^{6}$. The hash table size increases as $n$. When
$n<4\times 10^{5}$, the hash table fits into the L3 cache, and thus we do not
perform cache partitioning. Our solution has similar performance compared to
flat_hash_map. When $n\geq 4\times 10^{5}$, we use cache partitioning to
reduce expensive CPU cache misses, thereby significantly out-performing
flat_hash_map. As for std::unordered_map, it is much slower because of chained
hashing.

In Figure 10(b), we fix $n$=$10^{6}$ while varying $|R|$=$|S|$ from $10^{6}$
to $10^{7}$. The size of the hash table is fixed. The number of hash visits
increases. Hence, we see the increasing trends for all curves. Overall, our
optimized mapping solution performs the best.

Moreover, we compare DIM3 (with optimized mapping) and DIM3 with baseline
std::unordered_map for the Amazon data set. The mapping step takes 54.9% of
the Join-Project run time for DIM3 with baseline mapping. Replacing the
baseline mapping with optimized mapping, DIM3 achieves an improvement of 1.5x.

_(2) DenseEC_. In Figure 10(c), we vary $n$ from 1000 to 6000, where the data
sets are relatively dense and thus dense MM makes sense. We compare DenseEC
with dense MM implementations.

We investigate sub-cubic fast MM algorithms. While the best known
$\mathcal{O}(n^{2.373})$ algorithm (Gall and Urrutia, 2018) is considered
impractical because of its huge constant factors (Le Gall, 2012; Pegoraro et
al., 2020), recent work aims to make the Strassen algorithm practical (Benson
and Ballard, 2015). We obtain and evaluate this _STRASSEN_ implementation
(Benson and Ballard, 2015). Our initial results show that it is slower than
Intel MKL. One potential reason is that the original _STRASSEN_ supports only
floating point MM, while the MKL run uses integer MM. Since the _STRASSEN_
code calls BLAS as the underlying MM sub-routine, it is straight-forward to
modify the code to call MKL’s integer MM. The resulting integer _STRASSEN_
indeed runs slightly faster, but it is still slower than MKL. Figure 10(c)
shows the performance of the improved integer _STRASSEN_.

Overall, DenseEC out-performs dense MM implementations because early stopping
can effectively reduce computation cost.

_(3) SparseBMM_. In Figure 10(d), we vary $n$ from $10^{3}$ to $10^{6}$. The
generated data sets become more and more sparse, and therefore sparse MM and
_Classical_ are more suitable. We compare SparseBMM with _Classical_ and MKL’s
sparse MM. From the figure, we see that SparseBMM performs the best.

We also plot the SparseBMM curve in Figure 10(c). The crossing point of
SparseBMM and DenseEC is around 4000. This shows why DIM3 chooses between
DenseEC and SparseBMM with function $f_{2}$.

###  5.3. Evaluation for Partial Result Caching

(a) Run time on real-world data sets.

(b) Space cost on real-world data sets.

(c) Run time on Zipf data.

(d) Run time on R-mat data.

Figure 11. Partial result caching. ($\star$ shows the point where cached
result size $=$ input data size.)

We evaluate partial result caching for DIM3 using both real-world and
synthetic data sets, as shown in Figure 11. The synthetic data sets use the
same default parameters as in Section 5.2. Our solution retrieves cached
partial results and performs JoinProject between $R$ and
$\\{$$($$z,y$$)$$|$$($$z,y$$)$$\in S\land z\notin Z_{cached}\\}$. The X-axis
varies the percentage of $z$ values that are cached. The Y-axis reports either
the run time normalized to DIM3 without caching, or the cached result size
normalized to the size of the total JoinProject results.

Figure 11(a) and 11(b) show the run time and the cached result size on real-
world data sets, respectively. For Slashdot and HetRec, the computation time
distributions for $z$ values are very skewed. The space cost is relatively
high because most results are cached by their original values. Caching a
relatively small percentage (e.g., 20%) of $z$ values can significantly
speedup DIM3. On the other hand, the denser data sets (i.e., MV1, MV2, and
Jokes) see less skewed computation time distributions for $z$ values. As the
number of results for the same $z$ is often large, our technique to cache the
complement set of result vectors can drastically reduce the cached result
size. For example, the space at 100% (i.e., for caching all results) is only
12%, 6%, and less than 1% of the total result sizes for MV1, MV2, and Jokes,
respectively. Thus, we can use a small amount of cache space to achieve
significant speedups. Overall, if we cache either 20% of $z$ values or up to
the input data size, partial result caching achieves 3.3x–20x improvements
over DIM3 without caching.

Figure 11(c) shows the run time on synthetic data sets where $R.x$ and $S.z$
follow Zipf distribution. When Zipf parameter $\alpha=0$, the data is
uniformly distributed, and the benefits of caching is low. As $\alpha$
increases from 0.25 to 1, the computation time distribution for $z$ values
becomes more skewed and caching is more helpful. Caching only 10% of $z$
values reduces the run time by 13% to 57%.

Figure 11(d) shows the run time on R-mat data varying the input table size
$|R|$ from $10^{6}$ to $10^{7}$. If we limit the cache space to the size of
the input table, partial result caching reduces the run time of DIM3 by
25%–67%.

###  5.4. Evaluation for Join-OP Query Types

Join-Aggregate. Figure 12(a) compares the Join-Aggregate performance of DIM3
with RDBMSs and the stand-alone hash-based solution on MV1, MV2, and Jokes. We
focus on the more complex case where the group-by attributes are from both
tables. As described in Section 4.1, if the group-by attributes are from one
table, we can rewrite the query and employ the classical solution for
efficient evaluation. We compute aggregates on the ratings for the MovieLens
and Jokes data sets. (We omit Slashdot and HetRec in this set of experiments
because they do not have natural value columns to express the aggregation.) We
see that DIM3 performs the best among all solutions. Compared with the hash-
based solution, DIM3 obtains a speedup of 18–24$\times$. Compared with the
fastest RDBMS solution, DIM3 obtains a speedup of as least 30$\times$.

MJP. We evaluate the effectiveness of our proposed DP algorithm. We compare
three DIM3 variants with RDBMSs: 1) DIM3 _(DP)_ runs DP to find the optimal
query plan; 2) DIM3 _(EagerDedup)_ places a deduplication after every join;
and 3) DIM3 _(LazyDedup)_ performs only one deduplication after all the joins.

In this set of experiments, we use the Friendster data set (Yang and Leskovec,
2015b) and compute multi-hop connections between users. Since the original
Friendster data set is large and very sparse, as shown in Table 2, we
construct subsets of the data set using a parameter $range$. Given a $range$,
we filter out any tuples whose attributes are both larger than $range$. From
the remaining tuples, we randomly extracted 10 tables, each with $10^{6}$
tuples, then perform a 10-way join with projection. We set $range$= $5\times
10^{5}$, $7\times 10^{5}$ or $9\times 10^{5}$. Note that the larger the
$range$, the sparser the input tables. For the three sub data sets, the final
output size $|OUT_{P}|$ is 46M, 18M, and 8M, and DIM3 _(DP)_ decides to place
4, 3, and 2 deduplication operations, respectively.

As shown in Figure 12(b), all DIM3 variants run faster than RDBMSs because of
the better Join-Project performance of DIM3. Compared with the hash-based
solution with EagerDedup strategy, DIM3(DP) obtains a speedup of
4.9–7.4$\times$. When $range$= $5\times 10^{5}$, DIM3 _(DP)_ gains a 10x
speedup compared to _LazyDedup_. When $range$= $7\times 10^{5}$, _DP_ are 11%
and 39% better than _EagerDedup_ and _LazyDedup_ , respectively. When $range$=
$9\times 10^{5}$, _DP_ is 1.9x as fast as _EagerDedup_. However, _DP_ is
slightly (i.e., 5%) slower than _LazyDedup_. In this case, _DP_ places a
deduplication after the first join. The hope is that the amount of computation
in subsequent joins will be proportionally reduced, but the actual reduction
is not as significant. Overall, we see that DIM3 _(DP)_ performs the best
among the three variants. This confirms the effectiveness of the DP algorithm
for finding the optimal plan for evaluating MJP.

(a) Join-Aggregate.
(${}_{x,z}\mathcal{G}_{avg(|R.v-S.v|)}(R(x,y,v)\Join_{y}S(z,y,v))$)

(b) MJP.

Figure 12. Evaluation for Join-$op$ query types.(_Hash_ : stand-alone
implementation using flat_hash_map for join, aggregation, and deduplication.
’X’: _Hash_ runs out of memory and fails.)

##  6\. Conclusion and Future Work

In this paper, we propose DIM3 that combines intersection-free partitioning,
optimized mapping, and DenseEC and SparseBMM algorithms to improve the state-
of-the-art DHK solution. Moreover, we investigate partial result caching and
extend DIM3 to efficiently compute Join-$op$ queries. Our results show that
DIM3 is a promising solution for the widely used Join-Project operation.

Several promising future directions remain to be explored. The first is out-
of-core computation. When the input table $R$ and $S$ are too large to fit in
the allocated memory, one way is to perform I/O partitioning for $R$ and $S$
according to $R.x$ and $S.z$, respectively. Then, we load each pair of $R.x$
partition and $S.z$ partition into memory and employ DIM3 to compute the
results. An alternative way is to partition the tables according to the join
key $y$. However, a final deduplication step is necessary because the
intermediate results from different partitions may contain duplicates.

The second direction is to study caching for multiple queries when partial
result caching is enabled. In addition to traditional considerations, such as
query access statistics, the cache space allocated to a Join-Project query
becomes a tunable parameter. This adds a new dimension in the design of the
caching strategy.

Last but not least, it is interesting to exploit new hardware to accelerate
database operations. For example, GPUs (Graphics Processing Units) (Breß et
al., 2014) and TPUs (Tensor Core Units) (Hu et al., 2021) can significantly
accelerate matrix multiplication in Join-Project. NVM (Non-Volatile Memory)
provides an interesting combination of persistence, capacity, and performance,
which can be used to speed up queries with huge memory consumption (van Renen
et al., 2018).

###### Acknowledgements.

This work is partially supported by Natural Science Foundation of China
(62172390). Shimin Chen is the corresponding author.

## References

  * (1)
  * Amossen and Pagh (2009) Rasmus Resen Amossen and Rasmus Pagh. 2009.  Faster join-projects and sparse matrix multiplications. In _Proceedings of the 12th International Conference on Database Theory_. 121–126.
  * Balkesen et al. (2013) Cagri Balkesen, Gustavo Alonso, Jens Teubner, and M Tamer Özsu. 2013.  Multi-core, main-memory joins: Sort vs. hash revisited.  _Proceedings of the VLDB Endowment_ 7, 1 (2013), 85–96.
  * Benson and Ballard (2015) Austin R Benson and Grey Ballard. 2015.  A framework for practical parallel fast matrix multiplication.  _ACM SIGPLAN Notices_ 50, 8 (2015), 42–53.
  * Blakeley et al. (1986) José A. Blakeley, Per-Åke Larson, and Frank Wm. Tompa. 1986.  Efficiently Updating Materialized Views. In _Proceedings of the 1986 ACM SIGMOD International Conference on Management of Data, Washington, DC, USA, May 28-30, 1986_. ACM Press, 61–71.
  * Breß et al. (2014) Sebastian Breß, Max Heimel, Norbert Siegmund, Ladjel Bellatreche, and Gunter Saake. 2014\.  Gpu-accelerated database systems: Survey and open challenges.  In _Transactions on Large-Scale Data-and Knowledge-Centered Systems XV_. Springer, 1–35.
  * Cai et al. (2019) Walter Cai, Magdalena Balazinska, and Dan Suciu. 2019\.  Pessimistic cardinality estimation: Tighter upper bounds for intermediate join cardinalities. In _Proceedings of the 2019 International Conference on Management of Data_. 18–35.
  * Cantador et al. (2011) Iván Cantador, Peter Brusilovsky, and Tsvi Kuflik. 2011\.  Second workshop on information heterogeneity and fusion in recommender systems (HetRec2011). In _Proceedings of the fifth ACM conference on Recommender systems_. 387–388.
  * Chakrabarti et al. (2004) Deepayan Chakrabarti, Yiping Zhan, and Christos Faloutsos. 2004\.  R-MAT: A recursive model for graph mining. In _Proceedings of the 2004 SIAM International Conference on Data Mining_. SIAM, 442–446.
  * Dalton et al. (2015) Steven Dalton, Luke Olson, and Nathan Bell. 2015\.  Optimizing sparse matrix—matrix multiplication for the gpu.  _ACM Transactions on Mathematical Software (TOMS)_ 41, 4 (2015), 1–20.
  * Deep et al. (2020) Shaleen Deep, Xiao Hu, and Paraschos Koutris. 2020.  Fast join project query evaluation using matrix multiplication. In _Proceedings of the 2020 ACM SIGMOD International Conference on Management of Data_. 1213–1223.
  * Deep et al. (2021) Shaleen Deep, Xiao Hu, and Paraschos Koutris. 2021.  Enumeration Algorithms for Conjunctive Queries with Projection. In _24th International Conference on Database Theory, ICDT 2021, March 23-26, 2021, Nicosia, Cyprus_ _(LIPIcs)_ , Ke Yi and Zhewei Wei (Eds.), Vol. 186. Schloss Dagstuhl - Leibniz-Zentrum für Informatik, 14:1–14:17.
  * Gall and Urrutia (2018) François Le Gall and Florent Urrutia. 2018.  Improved rectangular matrix multiplication using powers of the Coppersmith-Winograd tensor. In _Proceedings of the Twenty-Ninth Annual ACM-SIAM Symposium on Discrete Algorithms_. SIAM, 1029–1046.
  * Goldberg et al. (2001) Ken Goldberg, Theresa Roeder, Dhruv Gupta, and Chris Perkins. 2001.  Eigentaste: A constant time collaborative filtering algorithm.  _information retrieval_ 4, 2 (2001), 133–151.
  * Graefe and Kuno (2011) Goetz Graefe and Harumi Kuno. 2011.  Modern B-tree techniques. In _2011 IEEE 27th International Conference on Data Engineering_. IEEE, 1370–1373.
  * Harper and Konstan (2015) F Maxwell Harper and Joseph A Konstan. 2015.  The movielens datasets: History and context.  _Acm transactions on interactive intelligent systems (tiis)_ 5, 4 (2015), 1–19.
  * Hertzschuch et al. (2021) Axel Hertzschuch, Claudio Hartmann, Dirk Habich, and Wolfgang Lehner. 2021.  Simplicity Done Right for Join Ordering.. In _CIDR_.
  * Ho and Park (2016) VanPhi Ho and Dong-Joo Park. 2016.  A survey of the-state-of-the-art b-tree index on flash memory.  _International Journal of Software Engineering and Its Applications_ 10, 4 (2016), 173–188.
  * Hu and Yi (2020) Xiao Hu and Ke Yi. 2020\.  Parallel Algorithms for Sparse Matrix Multiplication and Join-Aggregate Queries. In _Proceedings of the 39th ACM SIGMOD-SIGACT-SIGAI Symposium on Principles of Database Systems_. 411–425.
  * Hu et al. (2021) Yu-Ching Hu, Yuliang Li, and Hung-Wei Tseng. 2021.  TCUDB: Accelerating Database with Tensor Processors.  _arXiv preprint arXiv:2112.07552_ (2021).
  * Ioannidis and Kang (1991) Yannis E Ioannidis and Younkyung Cha Kang. 1991.  Left-deep vs. bushy trees: An analysis of strategy spaces and its implications for query optimization. In _Proceedings of the 1991 ACM SIGMOD international conference on Management of data_. 168–177.
  * Kanda et al. (2017) Shunsuke Kanda, Kazuhiro Morita, and Masao Fuketa. 2017\.  Practical string dictionary compression using string dictionary encoding. In _2017 International Conference on Big Data Innovations and Applications (Innovate-Data)_. IEEE, 1–8.
  * Kepner and Gilbert (2011) Jeremy Kepner and John Gilbert. 2011.  _Graph algorithms in the language of linear algebra_.  SIAM.
  * Kim et al. (2009) Changkyu Kim, Tim Kaldewey, Victor W Lee, Eric Sedlar, Anthony D Nguyen, Nadathur Satish, Jatin Chhugani, Andrea Di Blas, and Pradeep Dubey. 2009.  Sort vs. hash revisited: Fast join implementation on modern multi-core CPUs.  _Proceedings of the VLDB Endowment_ 2, 2 (2009), 1378–1389.
  * Kocberber et al. (2013) Onur Kocberber, Boris Grot, Javier Picorel, Babak Falsafi, Kevin Lim, and Parthasarathy Ranganathan. 2013.  Meet the walkers accelerating index traversals for in-memory databases. In _2013 46th Annual IEEE/ACM International Symposium on Microarchitecture (MICRO)_. IEEE, 468–479.
  * Kraska et al. (2018) Tim Kraska, Alex Beutel, Ed H Chi, Jeffrey Dean, and Neoklis Polyzotis. 2018.  The case for learned index structures. In _Proceedings of the 2018 International Conference on Management of Data_. 489–504.
  * Kurzak et al. (2009) Jakub Kurzak, Wesley Alvaro, and Jack Dongarra. 2009\.  Optimizing matrix multiplication for a short-vector SIMD architecture–CELL processor.  _Parallel Comput._ 35, 3 (2009), 138–150.
  * Larson et al. (2004) Per-Åke Larson, Jonathan Goldstein, and Jingren Zhou. 2004\.  MTCache: Transparent Mid-Tier Database Caching in SQL Server. In _Proceedings of the 20th International Conference on Data Engineering, ICDE 2004, 30 March - 2 April 2004, Boston, MA, USA_. IEEE Computer Society, 177–188.
  * Larson and Yang (1985) Per-Åke Larson and H. Z. Yang. 1985.  Computing Queries from Derived Relations. In _VLDB’85, Proceedings of 11th International Conference on Very Large Data Bases, August 21-23, 1985, Stockholm, Sweden_ , Alain Pirotte and Yannis Vassiliou (Eds.). Morgan Kaufmann, 259–269.
  * Le Gall (2012) François Le Gall. 2012\.  Faster algorithms for rectangular matrix multiplication. In _2012 IEEE 53rd annual symposium on foundations of computer science_. IEEE, 514–523.
  * Leis et al. (2017) Viktor Leis, Bernhard Radke, Andrey Gubichev, Alfons Kemper, and Thomas Neumann. 2017.  Cardinality Estimation Done Right: Index-Based Join Sampling.. In _Cidr_.
  * Leskovec et al. (2007) Jure Leskovec, Lada A. Adamic, and Bernardo A. Huberman. 2007\.  The dynamics of viral marketing.  _ACM Trans. Web_ 1, 1 (2007), 5.  https://doi.org/10.1145/1232722.1232727
  * Leskovec et al. (2009) Jure Leskovec, Kevin J Lang, Anirban Dasgupta, and Michael W Mahoney. 2009.  Community structure in large networks: Natural cluster sizes and the absence of large well-defined clusters.  _Internet Mathematics_ 6, 1 (2009), 29–123.
  * Mishra and Eich (1992) Priti Mishra and Margaret H Eich. 1992.  Join processing in relational databases.  _ACM Computing Surveys (CSUR)_ 24, 1 (1992), 63–113.
  * Murphy et al. (2010) Richard C Murphy, Kyle B Wheeler, Brian W Barrett, and James A Ang. 2010.  Introducing the graph 500.  _Cray Users Group (CUG)_ 19 (2010), 45–74.
  * Pegoraro et al. (2020) Marco Pegoraro, Merih Seran Uysal, and Wil MP van der Aalst. 2020.  Efficient time and space representation of uncertain event data.  _Algorithms_ 13, 11 (2020), 285.
  * Qiu et al. (2021) Yuan Qiu, Yilei Wang, Ke Yi, Feifei Li, Bin Wu, and Chaoqun Zhan. 2021.  Weighted Distinct Sampling: Cardinality Estimation for SPJ Queries. In _Proceedings of the 2021 International Conference on Management of Data_. 1465–1477.
  * Saule et al. (2013) Erik Saule, Kamer Kaya, and Ümit V Çatalyürek. 2013\.  Performance evaluation of sparse matrix multiplication kernels on intel xeon phi. In _International Conference on Parallel Processing and Applied Mathematics_. Springer, 559–570.
  * Schneider and DeWitt (1990) Donovan A Schneider and David J DeWitt. 1990.  _Tradeoffs in processing complex join queries via hashing in multiprocessor database machines_.  University of Wisconsin-Madison. Computer Sciences Department.
  * Skarupke (2017) Malte Skarupke. 2017\.  I Wrote The Fastest Hashtable.  https://probablydance.com/2017/02/26/i-wrote-the-fastest-hashtable/.
  * Strassen (1969) Volker Strassen. 1969\.  Gaussian elimination is not optimal.  _Numer. Math._ 13 (1969), 354–356.
  * van Renen et al. (2018) Alexander van Renen, Viktor Leis, Alfons Kemper, Thomas Neumann, Takushi Hashida, Kazuichi Oe, Yoshiyasu Doi, Lilian Harada, and Mitsuru Sato. 2018\.  Managing non-volatile memory in database systems. In _Proceedings of the 2018 International Conference on Management of Data_. 1541–1555.
  * Wang et al. (2014) Endong Wang, Qing Zhang, Bo Shen, Guangyong Zhang, Xiaowei Lu, Qing Wu, and Yajuan Wang. 2014.  Intel math kernel library.  In _High-Performance Computing on the Intel® Xeon Phi™_. Springer, 167–188.
  * Yang and Leskovec (2015a) Jaewon Yang and Jure Leskovec. 2015a.  Defining and evaluating network communities based on ground-truth.  _Knowledge and Information Systems_ 42, 1 (2015), 181–213.
  * Yang and Leskovec (2015b) Jaewon Yang and Jure Leskovec. 2015b.  Defining and evaluating network communities based on ground-truth.  _Knowledge and Information Systems_ 42, 1 (2015), 181–213.