# NOCAP: Near-Optimal Correlation-Aware Partitioning Joins

Zichen Zhu<EMAIL_ADDRESS>Boston UniversityBostonMAUSA ,  Xiao Hu
<EMAIL_ADDRESS>University of WaterlooWaterlooOntarioCanada and  Manos
Athanassoulis<EMAIL_ADDRESS>Boston UniversityBostonMAUSA

###### Abstract.

Storage-based joins are still commonly used today because the memory budget
does not always scale with the data size. One of the many join algorithms
developed that has been widely deployed and proven to be efficient is the
Hybrid Hash Join (HHJ), which is designed to exploit any available memory to
maximize the data that is joined directly in memory. However, HHJ cannot fully
exploit detailed knowledge of the join attribute correlation distribution.

In this paper, we show that given a correlation skew in the join attributes,
HHJ partitions data in a suboptimal way. To do that, we derive the optimal
partitioning using a new cost-based analysis of partitioning-based joins that
is tailored for primary key - foreign key (PK-FK) joins, one of the most
common join types. This optimal partitioning strategy has a high memory cost,
thus, we further derive an approximate algorithm that has tunable memory cost
and leads to near-optimal results. Our algorithm, termed NOCAP (Near-Optimal
Correlation-Aware Partitioning) join, outperforms the state-of-the-art for
skewed correlations by up to $30\%$, and the textbook Grace Hash Join by up to
$4\times$. Further, for a limited memory budget, NOCAP outperforms HHJ by up
to 10%, even for uniform correlation. Overall, NOCAP dominates state-of-the-
art algorithms and mimics the best algorithm for a memory budget varying from
below $\sqrt{\|\text{relation}\|}$ to more than $\|\text{relation}\|$.

Storage-based Join, Partitioning Join, Dynamic Hybrid Hash Join

††ccs: Information systems Join algorithms††ccs: Theory of computation
Database query processing and optimization (theory)

##  1\. Introduction

Joins are ubiquitous in database management systems (DBMS). Further, _primary
key - foreign key_ (PK-FK) equi-joins are the most common type of joins. For
example, all the queries of industry-grade benchmarks like TPC-H (TPC, 2021)
and most of Join Order Benchmark (JOB) (Leis et al., 2015) are PK-FK equi-
joins. Recent research has focused on optimizing in-memory equi-joins (Boncz
et al., 1999; Blanas et al., 2011; Wassenberg and Sanders, 2011; Balkesen et
al., 2013b; Barber et al., 2014; Liu and Blanas, 2015; Schuh et al., 2016;
Bandle et al., 2021), however, as the memory prices scale slower than storage
(McCallum, 2022), the available memory might not always be sufficient to store
both tables simultaneously, thus requiring a classical storage-based join
(Ramakrishnan and Gehrke, 2002). This is common in a shared resource setting,
like multiple colocated databases or virtual database instances deployed on
the same physical cloud server (Jennings and Stadler, 2015; Buyya et al.,
2019; Berenberg and Calder, 2021). Besides, in edge computing, memory is also
limited, which is further exacerbated when other memory-demanding services are
running (He et al., 2020). In several other data-intensive use cases like
Internet-of-things, 5G communications, and autonomous vehicles (Cisco, 2018;
Gartner, 2017), memory might also be constrained. Finally, there are two main
reasons a workload might need to use storage-based joins: (1) workloads
consisting of queries with multiple joins, and (2) workloads with a high
number of concurrent queries. In both cases, the available memory and
computing resources must be shared among all concurrently executed join
operators.

Figure 1. A conceptual graph on the I/O cost of the state-of-the-art DHH, OCAP
and NOCAP, assuming $\|R\|\leq\|S\|$. $F$ is the fudge factor indicating the
space amplification factor between the in-memory hash table and the stored raw
data).

Storage-based Joins. When executing a join, storage-based join algorithms are
used when the available memory is not enough to hold the hash table for the
smaller relation. Traditional storage-based join algorithms include Nested
Block Join (NBJ), Sort Merge Join (SMJ), Grace Hash Join (GHJ), Simple Hash
Join (SMJ), and Hybrid Hash Join (HHJ). Overall, HHJ, which acts as a blend of
Simple Hash Join and Grace Hash Join, is considered the state-of-the-art
approach (Kitsuregawa et al., 1983; DeWitt et al., 1984; Shapiro, 1986), and
is extensively used in existing database engines (e.g., MySQL (MySQL, 2023),
PostgreSQL (PostgreSQL, 2023), AsterixDB (Alsubaiee et al., 2014)). HHJ
uniformly distributes records from the two input relations to a number of
partitions via hashing the join key and ensures that, when possible, one or
more partitions remain in memory and are joined on the fly without writing
them to disk. The remaining disk-resident partitions are joined using a
classical hash join approach to produce the final result. Unlike GHJ and SMJ,
HHJ uses the available memory budget judiciously and thus achieves a lower I/O
cost. Existing relational database engines (e.g., PostgreSQL (PostgreSQL,
2023) and AsterixDB (Alsubaiee et al., 2014; Kim et al., 2020)) often
implement a variant of HHJ, Dynamic Hybrid Hash join (DHH) that dynamically
decides which partitions should stay in memory during partitioning.

The Challenge: Exploit Skew. For PK-FK joins, we describe the number of
matching keys per PK using a distribution, which we also refer to as join
attribute correlation, or simply join correlation. In turn, join correlation
can be uniform (all PKs have the same number of FK matches) or skewed (some
PKs have more matches than others). Although there are many skew optimizations
for joins (DeWitt et al., 1992; Hua and Lee, 1991; Kitsuregawa et al., 1989;
Nakayama et al., 1988; Li et al., 2002), a potential detailed knowledge of the
join attribute correlation and its skew is not fully exploited since existing
techniques use heuristic rules to design the caching and partitioning
strategy. As such, while these heuristics may work well in some scenarios, in
general, they lead to suboptimal I/O cost under an arbitrary join correlation.
For example, HHJ can be optimized by prioritizing entries with high-frequency
keys when building the in-memory hash table (PostgreSQL, [n. d.]). However,
practical deployments use limited information about the join attribute
correlation and typically employ a fixed threshold for building an in-memory
hash table for keys with high skew (e.g., 2% of available memory). As a
result, prior work does not systematically quantify the benefit of such
approaches, nor does it offer a detailed analysis of how close to optimal they
might be. In fact, due to the exponential search space, there is no previous
literature that accurately reveals the relationship between the optimal I/O
cost and the join correlation, thus leaving a large unexplored space for
studying the caching and partitioning strategy.

Key Idea: Optimal Correlation-Aware Partitioning. To study the optimality of
different partitioning-based join algorithms, we model a general partitioning
strategy that allows for arbitrary partitioning schemes (not necessarily based
on a specific hash function). We assume that the relations we join have a PK-
FK relationship, and we develop an algorithm for Optimal Correlation-Aware
Partitioning (OCAP) that allows us to compare any partitioning strategy with
the optimal partitioning, given the join correlation. Our analysis reveals
that the state of the art is suboptimal in the entire spectrum of available
memory budget (varied from $\sqrt{\|\text{relation}\|}$ to
$\|\text{relation}\|$), leading up to $60\%$ more I/Os than strictly needed as
shown in Figure 1, where the black lines are the state of the art, and the red
lines are the optimal number of I/Os. The OCAP is constructed by modeling the
PK-FK join cost as an optimization problem and then proving the consecutive
theorem which establishes the basis of finding the optimal cost within
polynomial time complexity. We propose a dynamic programming algorithm that
finds the optimal solution in quadratic time complexity, $O(n^{2}\cdot
m^{2})$, and a set of pruning techniques that further reduce this cost to
$O\left({\left(n^{2}\cdot\log m\right)}/m\right)$ (where $n$ is the number of
records of the smaller relation and $m$ is the memory budget in pages). OCAP
has a large memory footprint as it assumes that the detailed information of
the join attributes correlation is readily available and, thus, can be only
applied for offline analysis. Further, note that OCAP models the cost using
the number of records as a proxy to I/Os, and is used as a building block of a
practical algorithm that we discuss next. We rely on OCAP to identify the
headroom for improvement from the state of the art.

The Solution: Near-Optimal Correlation-Aware Partitioning Join. In order to
build a practical join algorithm with a tunable memory budget, we approximate
the optimal partitioning provided by OCAP with Near-Optimal Correlation-Aware
Partitioning (NOCAP) algorithm. NOCAP enforces a strict memory budget and
splits the available memory between buffering partitions and caching
information regarding skew keys in join correlation. In Figure 1, we construct
a conceptual graph to compare our solution and the state-of-the-art storage-
based join methods, such as Dynamic Hybrid Hash (DHH). As shown, DHH cannot
fully adapt to the correlation and thus results in higher I/O cost than OCAP
when the correlation is skewed. Our approximate algorithm, NOCAP, uses the
same input from DHH and achieves near-optimal I/O cost compared with OCAP.
Further, while DHH is able to exploit the higher skew to reduce its I/O cost,
the headroom for improvement for high skew is higher than for low skew, which
is largely achieved by our approach. Overall, NOCAP is a practical join
algorithm that is mostly beneficial compared to the state of the art, and
offers its maximum benefit for a high skew in the join attribute correlation.

Contributions. In summary, our contributions are as follows:

  * •

We build a new cost model for partitioning-based PK-FK join algorithms, and
propose optimal correlation-aware partitioning (OCAP) based on dynamic
programming (§3.1), assuming that the join correlation is known. OCAP’s time
complexity is $O\left(n^{2}\cdot\log m/m\right)$ and space complexity is
$O(n)$, where $n$ is the input size in tuples and $m$ is the memory budget in
pages (§3.2).

  * •

We design an approximate correlation-aware partitioning (NOCAP) algorithm
based on OCAP. NOCAP uses partial correlation information (the same
information used by the state-of-the-art skew optimized join algorithms) and
achieves near-optimal performance while respecting memory budget constraints
(§4).

  * •

We thoroughly examine the performance of our algorithm by comparing it against
GHJ, SMJ, and DHH. We identify that the headroom for improvement is much
higher for skewed join correlations by comparing the I/O cost of DHH against
the optimal. Further, we show that NOCAP can reach near-optimal I/O cost and
thus lower latency under different correlation skew and memory budget,
compared to the state of the art (§5). Overall, NOCAP dominates the state-of-
the-art and offers performance benefits of up to 30% when compared against DHH
and up to $4\times$ when compared against the textbook GHJ.

##  2\. Previous Storage-based Joins

We first review four classical storage-based join methods (DeWitt et al.,
1984; Haas et al., 1997; Jahangiri et al., 2022; Ramakrishnan and Gehrke,
2002) (see Table 1), and then present more details of Dynamic Hybrid Hash
(DHH), a variant of the state-of-the-art HHJ.

###  2.1. Classical Storage-based Joins

Nested Block Join (NBJ). NBJ partially loads the smaller relation $R$ (in
chunks equal to the available memory) in the form of an in-memory hash table
and then scans the larger relation $S$ once _per chunk_ to produce the join
output for the partial data. This process is repeated multiple times for the
smaller relation until the entire relation is scanned. As such, the larger
relation is scanned for as many times as the number of chunks in the smaller
relation.

Sort Merge Join (SMJ). SMJ works by sorting both input tables by the join
attribute using external sorting and applying $M$-way ($M\leq B-1$) merge sort
to produce the join result. During the external sorting process, if the number
of total runs is less than $B-1$, the last sort phase can be combined together
with the multi-way join phase to avoid repetitive reads and writes
(Ramakrishnan and Gehrke, 2002).

Grace Hash Join (GHJ). GHJ uniformly distributes records from the two input
relations to a number of partitions (at most $B-1$) via hashing the join key,
and the corresponding partitions are joined then. Specifically, when the
smaller partition fits in memory, we simply store it in memory (typically as a
hash table) and then scan the larger partition to produce the output .

Table 1. Estimated Cost for NBJ, GHJ, and SMJ for $R\Join S$. $\|R\|$ and $\|S\|$ are the number of pages of $R$ and $S$. Assume $\|R\|\leq\|S\|$. #chunks is the number of passes for scanning $S$. #pa-runs is the number of times to partition $R$ and $S$ until the smaller partition fits in memory. #s-passes is the number of partially sorted passes of $R$ and $S$, until the number of the total sorted runs is $\leq B-1$ ($B$ is the total number of memory available). $\mu$ and $\tau$ indicate the write/read asymmetry, where $RW$, $SW$, and $SR$ denote the latency per I/O for random write, sequential write and sequential read respectively. Approach | Normalized #I/O | Notation
---|---|---
$\textsf{NBJ}(R,S)$ | $\|R\|+\\#\text{chunks}\cdot\|S\|$ | None
$\textsf{GHJ}(R,S)$ | $\left(1+\\#\text{pa-runs}\cdot\left(1+\mu\right)\right)\cdot\left(\|R\|+\|S\|\right)$ | $\mu\stackrel{{\scriptstyle\text{def}}}{{=}}RW/SR$
$\textsf{SMJ}(R,S)$ | $\left(1+\\#\text{s-passes}\cdot\left(1+\tau\right)\right)\cdot\left(\|R\|+\|S\|\right)$ | $\tau\stackrel{{\scriptstyle\text{def}}}{{=}}SW/SR$

Hybrid Hash Join (HHJ). HHJ is a variant of GHJ that allows one or more
partitions to stay in memory without being spilled to disk, if the space is
sufficient. When partitioning the second relation, we can directly probe in-
memory partitions and generate the join output for those in-memory partitions.
The remaining keys of the second relation are partitioned to disk, and we
execute the same probing phase as in GHJ to join the on-disk partitions. When
the memory budget is lower than $\sqrt{\|R\|\cdot F}$ (where $\|R\|$ is the
size of the smaller relation in pages, and $F$ is the fudge factor that stands
for the space amplification factor between the in-memory hash table and the
stored raw data), HHJ downgrades into GHJ because it will not be feasible to
maintain a partition in memory while having enough buffers for the remaining
partitions and the output.

Figure 2. An illustration of partitioning in DHH. Partitions in red are staged
in memory when $R$ is being partitioned, while the remaining ones in blue are
written to disk.

1 $m_{\textsf{DHH}}\leftarrow\max\left\\{20,\left\lceil\frac{\|R\|\cdot
F-B}{B-1}\right\rceil\right\\}$;

2 Initialize a length-$m_{\textsf{DHH}}$ array POB as $\\{0,0,...,0\\}$;

3 Initialize a length-$m_{\textsf{DHH}}$ partitioning PartPages;

4 foreach  _$r\in R$_ do

5 $\textsf{PartID}\leftarrow h_{\textsf{split}}(r.\textsf{key})$;

6 if  _${\textsf{POB}}[{\textsf{PartID}}]=0$_ then

7 if
_$\sum\limits_{i=1}^{m_{\textsf{DHH}}}\textsf{PartPages}[i].\textsf{size}()=B-2-m_{\textsf{DHH}}$_
then

8 Flush an arbitrary partition $j$ to disk;

9 ${\textsf{POB}}[j]=1$;

10

11 else append $r$ to $\textsf{PartPages}[\textsf{PartID}]$;

12

13 else append $r$ to the in-disk partition PartID;

14

15foreach  _$r\in\textsf{PartPages}$_ do add $r$ to
$HT_{\textsf{mem}}[h_{\textsf{probe}}(r.\textsf{key})]$;

16 return $HT_{\textsf{mem}},{\textsf{POB}}$;

Algorithm 1 $\textsc{DHH-Primary}(R,HS_{\textsf{mem}},B)$

1 foreach  _$s\in S$_ do

2 if  _$h_{\textsf{probe}}(s.\textsf{key})\in HT_{\textsf{mem}}$_ then

3 if  _$s.\textsf{key}$ is in records
$HT_{\textsf{mem}}[h_{\textsf{probe}}(s.\textsf{key})]$_ then

4 join $s$ with matched records and output;

5

6

7 else

8 if  _${\textsf{POB}}[h_{\textsf{split}}(s.\textsf{key})]=1$_ then

9 append $s$ to the in-disk partition PartID;

10

11

12

Algorithm 2 $\textsc{DHH-Foreign}(S,HT_{\textsf{mem}},{\textsf{POB}}$)

###  2.2. Dynamic Hybrid Hash Join

Here we review DHH, the state-of-the-art implementation of HHJ, as well as its
skew optimization. While there are many different implementations of DHH
(Jahangiri et al., 2022), the core idea is the same across all of them: decide
which partitions to stay in memory or on disk.

Framework. We visualize one implementation (AsterixDB (Alsubaiee et al., 2014;
Kim et al., 2020)) of DHH in Figure 2. Compared to HHJ, DHH exhibits higher
robustness by dynamically deciding which partitions should be spilled to disk.
Specifically, every partition is initially staged in memory, and when needed,
a staged partition will be selected to be written to disk, and it will free up
its pages for new incoming records. Typically, the largest partition will be
selected, however, the victim selection policy may vary in different systems.
A bit vector (“Page Out Bits”, short as POB, in Figure 2) is maintained to
record which partitions have been written out at the end of the partitioning
phase of $R$. Another hash function is applied to build a large in-memory hash
table using all staged partitions. It is worth noting that, when no partitions
can be staged in the memory (i.e., all the partitions are spilled out to
disk), DHH naturally downgrades into GHJ. When partitioning the relation $S$,
DHH applies the partitioning hash function first and checks POB for every
record. If it is set, DHH directly probes the in-memory hash table and
generate the join result. Otherwise, DHH moves the joined record to the
corresponding output page and flushes it to the disk if it is full. Finally,
DHH performs exactly the same steps as the traditional hash join to probe all
the disk-resident partitions. The partitioning and probing of DHH is presented
in detail in Algorithms 1 and 2.

Number of Partitions. The number of partitions in DHH (noted by
$m_{\textsf{DHH}}$) is a key parameter that largely determines join
performance. A large $m_{\textsf{DHH}}$ requires more output buffer pages
reserved for each partition, and, thus, less memory is left for memory-
resident partitions. On the other hand, small $m_{\textsf{DHH}}$ renders the
size of each partition very large, which makes it harder to stage in memory.
No previous work has formally investigated the optimal number of partitions
for arbitrary correlation skew, but past work on HHJ employs some heuristic
rules. For example, an experimental study (Jahangiri et al., 2022) recommends
that 20 is the minimum number of partitions when we do not have sufficient
information about the join correlation. In addition, if we restrict the number
of memory-resident partitions to be 1, we can set the number of partitions
$m_{\textsf{DHH}}$ as $m_{\textsf{DHH}}=\left\lceil\frac{\|R\|\cdot
F-B}{B-1}\right\rceil$ (Shapiro, 1986), where $\|R\|$ is the size of relation
$R$ in pages, $F$ is the fudge factor for the hash table, and $B$ is the total
given memory budget in pages. Integrating this equation with the above tuning
guideline, we configure
$m_{\textsf{DHH}}=\max\left\\{\left\lceil\frac{\|R\|\cdot
F-B}{B-1}\right\rceil,20\right\\}$.

Heuristic Skew Optimization In Practical Deployments. For ease of notation, we
assume that $R$ is the relation that holds the primary key and is smaller in
size (dimension table) and that $S$ is the relation that holds the foreign key
and is larger in size (fact table). In PK-FK joins, the join correlation can
be non-uniform. Histojoin (Cutt and Lawrence, 2009) proposes to cache in
memory the Most Common Values (MCVs, maintained by DBMS) to help reducing the
I/O cost. Specifically, if the memory budget is enough, Histojoin first
partitions the dimension table $R$ and caches entries with high-frequency keys
in a small hash table in memory. When it partitions the fact table $S$, if
there are any keys matched in the small hash table, it directly joins them and
output. That way, Histojoin avoids writing out many entries with the same key
from the fact table and reading them back into memory during the probing
phase. However, it is unclear how much frequency should be treated as high to
qualify for this optimization and how large the hash table for the skewed keys
should be within the available memory budget. In one implementation of
Histojoin (Cutt et al., 2008), it limits the memory usage to 2% for
constructing the hash table specifically for skewed keys. This restriction
ensures that the specific hash table fits in memory. Likewise, in PostgreSQL
(PostgreSQL, [n. d.]), when the skew optimization knob is enabled and MCVs are
present in the system cache (thus not occupying the user-specified memory),
PostgreSQL follows a similar approach by adding an extra trigger to build the
hash table for skewed keys. By default, PostgreSQL also allocates a fixed
budget of 2% memory for the hash table dedicated to handling skewed keys, but
this allocation only occurs when the total frequency of skewed keys is more
than $0.01\cdot n_{S}$ ($n_{S}$ is the number of keys in relation $S$). In
fact, PostgreSQL implements a general version of Histojoin by specifying the
frequency threshold. However, existing skew optimizations in both PostgreSQL
and Histojoin rely on fixed thresholds (i.e., memory budget for skew keys, and
frequency threshold), which apparently cannot be adapted to different memory
budget and different join skew. For example, as shown in Figure 1, state-of-
the-art DHH are not enough to achieve ideal I/O cost, and they are further
away from the optimal for higher skew.

##  3\. Optimal Correlation-Aware Partitioning Join

Table 2. Summary of our notation. Notation | Definitions (Explanations)
---|---
$\|R\|$ $(\|S\|)$ | #pages of relation $R$ ($S$)
$n_{R}$ $(n_{S})$ | #records from relation $R$ ($S$)
$P_{j}$ | the $j$-th partition ($P_{j}=R_{j}$)
$|P_{j}|$ | #records in $j$-th partition
$\|R_{j}\|$ $(\|S_{j}\|)$ | #pages of a partition $R_{j}$ ($S_{j}$)
$\mathbb{P}$ | a Boolean matrix for partitioning assignation
$f:i\rightarrow j$ | returns $j$ for $i$ so that $\mathbb{P}_{i,j}=1$
$\mathcal{N}_{f}(i)$ | the partition where the $i$-th records belongs
$n$ |  $n=n_{R}$ in the partitioning context
$m$ | #partitions
$B$ | #pages of the total available buffer
$F$ | the fudge factor of the hash table
CT | the correlation table of keys and their frequency
$b_{R}$ ($b_{S}$) | #records from relation $R$ ($S$) per page
$c_{R}$ ($c_{S}$) | #records from relation $R$ ($S$) per chunk in NBJ

We now discuss the theoretical limit of an I/O-optimal partitioning-based join
algorithm when the correlation skewness information is known in advance and
can be accessed for free. We assume that $R$ is the relation that holds the
primary key (dimension table) and $S$ is the relation that holds the foreign
key (fact table). More specifically, we model the correlation between two
input relations with respect to the join attributes as a _correlation table_
CT, where $\textsf{CT}[i]$ is the number of records in $S$ matching the $i$-th
record in $R$. Our goal is to find which keys should be cached in memory, and
how the rest of the keys should be partitioned in order to minimize the total
I/Os of the join execution with an arbitrary memory budget.

We start with an easy case when no records can be cached during the partition
phase (§3.1) and turn to the general case when keys can be cached (§3.2). We
note again that the memory used for storing CT as well as the optimal
partitioning does not compete with the available memory budget $B$, which is
indeed unrealistic. This is why we consider it as a theoretically I/O-optimal
algorithm, since it can help us understand the lower bound of any feasible
partitioning but cannot be converted into a practical algorithm. More proofs
and details can be found in the appendix.

###  3.1. Optimal Partitioning Without Caching

We first build the cost model for partitioning assuming no cached records
during the partitioning (§3.1.1), then present the main theorem for
characterizing the optimal partitioning (§3.1.2) and find the optimal
partitioning efficiently via dynamic programming (§3.1.3).

####  3.1.1. Partitioning as An Integer Program

A partitioning $\mathbb{P}$ is an assignment of $n$ records from $R$ to $m$
partitions. As we use at least one page to stream the input relation and the
rest of the pages as output buffers for each partition, $m\leq B-1$. More
specific constraints on $m$ will be determined in Section 3.2 and Section 4.1.
Once we know how to partition $R$ (the relation with primary key), we can
apply the same partition strategy to $S$.

Partitioning. We encode a partitioning $\mathbb{P}$ as a Boolean matrix of
size $n\times m$, where $\mathbb{P}_{i,j}=1$ indicates that the $i$-th record
belongs to the $j$-th partition. If the $i$-th record from $R$ does not have a
match in $S$, i.e., $\textsf{CT}[i]=0$, we will not assign it to any
partition. We can preprocess the input to filter out these records so that
each of remaining records belongs to exactly one partition, although it is
unrealistic in practice. In this way, a partitioning $\mathbb{P}$ can be
compressed as a mapping function $f:[n]\to[m]$ from the index of record to the
index of partition in $\mathbb{P}$, such that $i$-th record from relation $R$
is assigned to $f(i)$-th partition in $\mathbb{P}$. Let
$P_{j}=\\{i\in[n]:f(i)=j\\}$ be the set of indexes of records from $R$
assigned to the $j$-th partition in $\mathbb{P}$. We can then derive the
number of pages for $R_{j}$ and $S_{j}$ as
$\|R_{j}\|=\left\lceil|P_{j}|/b_{R}\right\rceil$, $\|S_{j}\|=\lceil\sum_{i\in
P_{j}}\textsf{CT}[i]/b_{S}\rceil$, where $b_{R}$ (resp. $b_{S}$) is the number
of records per page for $R$ (resp. $S$). Moreover, we use
$\mathcal{N}_{f}(i)=\\{i^{\prime}\in[n]:f(i^{\prime})=f(i)\\}$ to denote the
records from $S$ that fall into the same partition with the $i$-th record in
$R$. We summarize frequently-used notations in Table 2.

Cost Function. Given a partitioning $\mathbb{P}$, it remains to join
partitions in a pair-wise manner. Reusing the cost model in Table 1, the cost
of computing the join results for a pair of partitions $(R_{j},S_{j})$ is
$\min\\{\textsf{NBJ}(R_{j},S_{j}),\textsf{SMJ}(R_{j},S_{j}),\textsf{GHJ}(R_{j},S_{j})\\}$,
by picking the cheapest one among NBJ, SMJ, and GHJ. We omit DHH here for
simplicity, which does not lead to major changes of the optimal partitioning,
since in most cases NBJ is the most efficient one when taking read/write
asymmetry into consideration (write is generally much slower than read in
modern SSDs (Papon and Athanassoulis, 2021b, a)). For ease of illustration, we
assume only NBJ is always applied in the pair-wise join, i.e., the _join cost_
induced by partitioning $\mathbb{P}$ is
$\textsf{Join}(\mathbb{P},m)=\sum_{j=1}^{m}\textsf{NBJ}(R_{j},S_{j})$. It can
be proved that most of the following results still apply when alll of NBJ,
SMJ, and GHJ can be used for the partition-wise joins as long as we always
pick the one with minimum cost.

Recall the cost function of NBJ in Table 1. For simplicity, we assume
$\|R_{j}\|\leq\|S_{j}\|$ for each $j\in[m]$. The other case with
$\|R_{j}\|\geq\|S_{j}\|$ can be discussed similarly. By setting
#chunks$=\lceil\|R_{j}\|/((B-2)/F)\rceil$, $\|R_{j}\|=|P_{j}|/b_{R}$ and
$\|S_{j}\|=\lceil\sum_{i\in P_{j}}\textsf{CT}[i]/b_{S}\rceil$, we obtain the
cost function as follows:

$\begin{split}\textsf{Join}(\mathbb{P},m)=\|R\|+\sum\limits_{j=1}^{m}\left\lceil\frac{|P_{j}|}{c_{R}}\right\rceil\cdot\left\lceil\sum\limits_{i\in
P_{j}}\frac{\textsf{CT}[i]}{b_{S}}\right\rceil&\leq\|R\|+m+\frac{n_{R}}{c_{R}}+\sum\limits_{j=1}^{m}\left\lceil\frac{|P_{j}|}{c_{R}}\right\rceil\cdot\sum\limits_{i\in
P_{j}}\frac{\textsf{CT}[i]}{b_{S}}\\\
&=\|R\|+m+\frac{n_{R}}{c_{R}}+\frac{1}{b_{S}}\cdot\sum\limits_{i=1}^{n}\textsf{CT}[i]\cdot\left\lceil\frac{|\mathcal{N}_{f}(i)|}{c_{R}}\right\rceil\end{split}$

where $c_{R}=\lfloor b_{R}\cdot(B-2)/F\rfloor$ is a constant that denotes the
number of records per chunk in relation $R$, and $\mathbb{I}$ is the indicator
function. Note that the I/O cost of read and write both relations is the same
for all partitioning strategies.

Integer Program. Given the number of partitions $m$, finding a partitioning
$\mathbb{P}$ for $R$ (containing the primary key) of $n$ records with minimum
cost can be captured by the following integer program:

$\displaystyle\min_{\mathbb{P}}$
$\displaystyle\quad\textsf{Join}(\mathbb{P},m)$ subject to
$\displaystyle\quad\sum\limits_{j=1}^{m}\mathbb{P}_{i,j}=1,\forall i\in[n]$
$\displaystyle\quad m+1\leq B$
$\displaystyle\quad\mathbb{P}_{i,j}\in\\{0,1\\},\forall i\in[n],\forall
j\in[m]$

However, solving such an integer program takes exponentially large time by
exhaustively searching all possible partitionings. We next characterize the
optimal partitioning with some nice properties.

####  3.1.2. Main Theorem

For simplicity, we assume that CT is _sorted in ascending order_ , i.e., if
$i_{1}\leq i_{2}$, then $\textsf{CT}[i_{1}]\leq\textsf{CT}[i_{2}]$. With
sorted CT, we present our main theorem for characterizing the optimal
partitioning $\mathbb{P}$ with three critical properties. Firstly, each
partition of $\mathbb{P}$ contains consecutive records lying on the sorted CT
array. Then, we can define a partial ordering on the partition in
$\mathbb{P}$. Secondly, the sizes of all partitions are in a weakly descending
order. At last, the size of each partition except the first one (or the
largest one) is divisible by $c_{R}$.

###### Theorem 3.1 (Main Theorem).

Given an arbitrary sorted CT array, there is an optimal partitioning
$\mathbb{P}=\langle P_{1},P_{2},\cdots,P_{m}\rangle$ such that

  * •

(consecutive property) for any $i_{1}\leq i_{2}$, if $i_{1}\in P_{j}$ and
$i_{2}\in P_{j}$, then $i\in P_{j}$ holds for any $i\in[i_{1},i_{2}]$;

  * •

(weakly-ordered property)
$\left\lceil\frac{|P_{1}|}{c_{R}}\right\rceil\geq\left\lceil\frac{|P_{2}|}{c_{R}}\right\rceil\geq\cdots\geq\left\lceil\frac{|P_{m}|}{c_{R}}\right\rceil$;

  * •

(divisible property) for any $2\leq i\leq m$, $|P_{i}|$ is divisible by
$c_{R}$.

where $P_{j}$ is the set of records assigned to the $j$-th partition of
$\mathbb{P}$, i.e., $P_{j}=\\{i\in[n]:f(i)=j\\}$.

###### Lemma 3.2 (Swap Lemma).

For a partitioning $\mathbb{P}$, if there exists a pair of indexes
$i_{1},i_{2}$ with $\textsf{CT}[i_{1}]>\textsf{CT}[i_{2}]$ and
$|\mathcal{N}_{f}(i_{1})|\geq|\mathcal{N}_{f}(i_{2})|$, it is feasible to find
a new partitioning $\mathbb{P}^{\prime}$ by swapping records at $i_{1}$ and
$i_{2}$, such that
$\textsf{Join}(\mathbb{P})\geq\textsf{Join}(\mathbb{P}^{\prime})$.

Figure 3. An illustration of records from one partition $P_{j}$ lying in a
non-consecutive way in the sorted CT.

To prove the main theorem, the high-level idea is to transform an arbitrary
partitioning $\mathbb{P}$ into $\mathbb{P}^{\prime}$ with desired properties,
while keeping the join cost of $\mathbb{P}^{\prime}$ always smaller (or at
least no larger) than that of $\mathbb{P}$. The essence is a swap procedure,
as described by Lemma 3.2, which can reassign records in a way such records
with larger CT values always appear with fewer records in a partition. Its
proof directly follows the definition of join cost. Next, we will show how to
transform an arbitrary partitioning $\mathbb{P}$ step by step.

Consecutive Property. Given an arbitrary partitioning $\mathbb{P}$, we first
show how to transform it into $\mathbb{P}^{\prime}$ such that
$\mathbb{P}^{\prime}$ satisfies the consecutive property while
$\textsf{Join}(\mathbb{P}^{\prime})\leq\textsf{Join}(\mathbb{P})$. As
described in Algorithm 3, we always identify a partition whose records are not
consecutive on the sorted CT, say $P_{j}$, and apply the swap procedure to it.
Let $i^{-}_{j}$ and $i^{+}_{j}$ be the minimum and maximum index of records
from $P_{j}$ lying on CT. Let $S^{1}_{j},S^{2}_{j},\cdots,S^{\ell}_{j}$ be the
longest consecutive records whose indexes fall into $[i^{-}_{j},i^{+}_{j}]$ in
CT, and belong to the same partition $P_{j^{\prime}}$ for some $j^{\prime}\neq
j$ ($S^{\ell_{1}}_{j}$ and $S^{\ell_{2}}_{j}$ may belong to different
partitions). We denote $S_{j}=S^{1}_{j}\cup S^{2}_{j}\cup\cdots\cup
S^{\ell}_{j}$ as the set of all records that make $P_{j}$ non-consecutive with
respect to the sorted CT (Figure 3).

1 while  _there exists $P_{j}\in\mathbb{P}$ that is non-consecutive_ do

2 while  _true_ do

3
$\textsf{left},\textsf{right}\leftarrow\min_{i\in[n]:f(i)=j}i,\max_{i\in[n]:f(i)=j}i$;

4 $S_{j}\leftarrow\\{i\in[n]:\textsf{left}\leq i\leq\textsf{right},f(i)\neq
j\\}$;

5 if  _$S_{j}=\emptyset$_ then break;

6 $\ell\leftarrow\min_{i\in S_{j}}i$;

7 if  _$|\mathcal{N}_{f}(\ell)| >|P_{j}|$_ then

8 $\textsc{swap}(f(\ell),f(\textsf{left}))$,
$\textsf{left}\leftarrow\textsf{left}+1$;

9

10 else $\textsc{swap}(f(\ell),f(\textsf{right}))$,
$\textsf{right}\leftarrow\textsf{right}-1$;

11

12

13return $\mathbb{P}$;

Algorithm 3 SWAP$(\mathbb{P})$

Algorithm 3 maintains two pointers left and right to indicate the smallest and
largest indexes of records to be swapped. Consider the smallest index $\ell\in
S_{j}$. If $|\mathcal{N}_{f}(i)|>|P_{j}|$, we swap $\ell$ and left by putting
the record at $i$ into $P_{f(\textsf{left})}$ and the record at left into
$P_{f(\ell)}$ ($P_{f(\ell)}=\mathcal{N}_{f}(\ell)$ by definition). Otherwise,
we swap $\ell$ and right similarly. We repeat the procedure above until all
partitions become consecutive on the sorted CT. It can be seen that the
resulted partitioning from $\textsc{SWAP}(\mathbb{P})$ is “consecutive”, and
the cost does not increase in this process, thanks to the swap lemma.

1 while  _there exists $P_{j^{\prime}}\precsim P_{j}\in\mathbb{P}$ s.t.
$\left\lceil\frac{|P_{j^{\prime}}|}{c_{R}}\right\rceil<\left\lceil\frac{|P_{j}|}{c_{R}}\right\rceil$
_ do

2 $\ell,\ell^{\prime}\leftarrow\min_{i\in P_{j}}i,\min_{i\in
P_{j^{\prime}}}i$;

3 for  _$i\leftarrow 0$ to $|P_{j^{\prime}}|$_ do
swap$(f(\ell+i),f(\ell^{\prime}+i))$;

4 $\mathbb{P}\leftarrow\textsc{SWAP}(\mathbb{P})$;

5

Algorithm 4 ORDER$(\mathbb{P})$

Weakly-Ordered Property. Given a partitioning $\mathbb{P}$ with consecutive
property, we next show how to transform it into $\mathbb{P}^{\prime}$ such
that $\mathbb{P}^{\prime}$ satisfies both consecutive and weakly-ordered
properties. As each partition in $\mathbb{P}$ contains consecutive records in
the sorted CT, we can define a partial ordering $\precsim$ on the partitions
in $\mathbb{P}$. For simplicity, we assume that $P_{1}\precsim
P_{2}\precsim\cdots\precsim P_{m}$. As described in Algorithm 4, we always
identify a pair of partitions that are not “weakly-ordered”, i.e.,
$P_{j^{\prime}}\precsim P_{j}$ but
$\left\lceil|P_{j^{\prime}}|/c_{R}\right\rceil<\left\lceil|P_{j}|/c_{R}\right\rceil$.
In this case, we can apply Lemma 3.2 to swap records between $P_{j^{\prime}}$
and $P_{j}$. This swap would turn $P_{j}$ into non-consecutive on the sorted
CT. If this happens, we further invoke Algorithm 3 to transform $P_{j}$ into a
consecutive one; moreover, $P_{j^{\prime}}$ will be put after $P_{j}$ in the
sorted CT. We will repeatedly apply the procedure above until no unordered
pair of partitions exists in $\mathbb{P}$. In fact, a strongly-ordered
property also holds by replacing $\left\lceil|P_{j}|/c_{R}\right\rceil$ with
$|P_{j}|$. We keeps the weakly-ordered version here so that it does not
conflict with the following divisible property.

Divisible Property. Given a partitioning $\mathbb{P}$ satisfying the
consecutive and weakly-ordered property, we next show how to transform it into
$\mathbb{P}^{\prime}$ such that $\mathbb{P}^{\prime}$ also satisfies divisible
property as follows. We start checking the $m$-th partition $P_{m}$. If
$|P_{m}|$ is divisible by $c_{R}$, we just skip it. Otherwise, we move records
from the previous partition $P_{m-1}$ to $P_{m}$ until $|P_{m}|$ is divisible
by $c_{R}$. After done with $P_{m}$, we move to $P_{m-1}$. We continue this
procedure until $|P_{m}|,|P_{m-1}|,\cdots,|P_{2}|$ are all divisible by
$c_{R}$. This fine-grained movement between adjacent partitions naturally
preserves the consecutive property as well as the join cost.

####  3.1.3. Dynamic Programming

With Theorem 3.1, we can now reduce the complexity of finding the optimal
partitioning in a brute-force way by resorting to dynamic programming, and
searching candidate partitionings with consecutive, weakly-ordered, and
divisible properties.

Formulation by Consecutive Property. We define a sub-problem parameterized by
$(i,j)$: finding the optimal partitioning for the first $i$ records in the
sorted CT using $j$ partitions. The join cost of this sub-problem
parameterized by $(i,j)$ is denoted as $V[i][j]$. Recall that the optimal
solution of our integer program in Section 3.1.1 is exactly $V[n][m]$. As
described by Algorithm 5, we can compute $V$ in a bottom-up fashion. More
specifically, we use $V[i][j].\textsf{cost}$ to store the cost of an optimal
partitioning for the sub-problem parameterized by $(i,j)$ and
$V[i][j].\textsf{LastPar}$ stores the starting position of the last partition
correspondingly. We also define $\textsf{CalCost}(s,e)$ as the per-partition
join $\textsf{PPJ}(R_{j^{\prime}},S_{j^{\prime}})$ (omitting the term
$\|R_{j^{\prime}}\|$ in $\textsf{NBJ}(R_{j^{\prime}},S_{j^{\prime}})$ for a
pair of partition $R_{j^{\prime}}$ and $S_{j^{\prime}}$, if we let
$P_{j^{\prime}}=[s,e]$). This can be easily computed using the pre-computed
prefix sum:

(1)
$\begin{split}\textsf{CalCost}(s,e)=\left(\sum_{i=1}^{e}\textsf{CT}[i]-\sum_{i=1}^{s-1}\textsf{CT}[i]\right)\cdot&\Bigg{\lceil}\frac{e-s+1}{c_{R}}\Bigg{\rceil}\end{split}$

The essence of Algorithm 5 is the following recurrence formula:

$V[i][j].\textsf{cost}=\min_{0\leq k\leq
i-1}\left\\{V[k][j-1].\textsf{cost}+\textsf{CalCost}(k+1,i)\right\\}$

which iteratively search the starting position of the last partition in
$[0,i-1]$. Also, we can backtrack the index of the last partition to produce
the mapping function $f$ from each index to its associated partition, as
described by Algorithm 6.

1 Initialize $V$ of sizes $n\times m$;

2 for  _$i\leftarrow 1$ to $n$_ do

3 $V[i][1].\textsf{cost}\leftarrow\textsf{CalCost}(1,i)$,
$V[i][1].\textsf{LastPar}\leftarrow 1$;

4

5for  _ $i\leftarrow 2$ to $n$_ do

6 for  _ $j\leftarrow 2$ to $\min(m,i)$_ do

7 $V[i][j].\textsf{cost}\leftarrow+\infty$;

8 for  _ $k\leftarrow 0$ to $i-1$_ do

9 $\textsf{tmp}\leftarrow V[k][j-1].\textsf{cost}+\textsf{CalCost}(k+1,i)$;

10 if  _${\textsf{tmp}} <V[i][j].\textsf{cost}$_ then

11 $V[i][j].\textsf{cost}\leftarrow\textsf{tmp}$;

12 $V[i][j].\textsf{LastPar}\leftarrow k+1$;

13

14

15

16

17return $V$;

Algorithm 5 Partition$(\textsf{CT},n,m)$

1 $\textsf{LastPar}\leftarrow n$;

2 for  _$j\leftarrow 0$ to $m-1$_ do

3 for  _$i\leftarrow V[\textsf{LastPar}][m-j].\textsf{LastPar}$ to LastPar_ do
$f(i)\leftarrow m-j$;

4 $\textsf{LastPar}\leftarrow V[\textsf{LastPar}][m-j].\textsf{LastPar}-1$;

5

6return $F$;

Algorithm 6 $\textsc{GetCut}(V,n,m)$

The time complexity of Algorithm 5 is $O(m\cdot n^{2})$, which is dominated by
the three for-loops, and the space complexity is $O(m\cdot n)$ for storing
$V$. This significantly improves the brute-force enumerating method, whose
time complexity is as large as $O(m^{n})$.

Speedup by Weakly-Ordered Property. Combing the weakly-ordered property in
Theorem 3.1 with the pigeonhole principle, for any sub-problem parameterized
by $(i,j)$, there exists an optimal partitioning such that: (i) the “largest”
partition contains at least the first
$\left\lfloor\frac{i-1}{j}\right\rfloor+1-c_{R}$ records in the sorted CT, and
(ii) the “smallest” partition contains at most the last
$\left\lfloor\frac{i}{j}\right\rfloor+c_{R}$ records in the sorted CT. Note
that the “largest” (“smallest”) partition is not essentially the largest
(smallest) in terms of the actual partition size. It is possible that
$P_{j^{\prime}}\precsim P_{j^{\prime}+1}$ and
$\left\lceil|P_{j^{\prime}}|/c_{R}\right\rceil=\left\lceil|P_{j+1}|/c_{R}\right\rceil$
but $|P_{j^{\prime}}|\leq|P_{j^{\prime}+1}|$. However, even if
$|P_{j^{\prime}}|\leq|P_{j^{\prime}+1}|$, the difference cannot be larger than
$c_{R}$ from the weakly-ordered property. As such, the “largest” partition for
the sub-problem parameterized by $(i,j)$ cannot be smaller than
$\left\lfloor\frac{i-1}{j}\right\rfloor+1-c_{R}$. Similar analysis can be
applied to the upper bound of the size of the “smallest” partition.
Synthesizing above analysis, we can get two tighter bounds on $k$ in line 8:

  * •

$k\geq\max\\{(i-c_{R})\cdot(1-\frac{1}{j}),0\\}$. The partition $P_{j}$
contains records from $\textsf{CT}[k+1:i]$, and $|P_{j}|$ should be no larger
than the last (“smallest”) partition in $V[k][j-1]$. The last (“smallest”)
partition for the sub-problem parameterized by $(k,j-1)$ is at most
$\left\lfloor\frac{k}{j-1}\right\rfloor+c_{R}$. Hence,
$i-k\leq\left\lfloor\frac{k}{j-1}\right\rfloor+c_{R}$.

  * •

$k\leq\max\\{i-\left\lfloor\frac{n-i-1}{m-j}\right\rfloor-1+c_{R},1\\}$. The
partition $P_{j}$ contains records from $\textsf{CT}[k+1:i]$, and it should be
no smaller than the “largest” partition of remaining records in
$\textsf{CT}[i+1:n]$. Recall that, we still need to put the remaining $n-i$
records into $m-j$ partitions. The largest partition of the sub-problem
parameterized by $(n-i,m-j)$ contains at least
$\left\lfloor\frac{n-i-1}{m-j}\right\rfloor+1-c_{R}$ records. Hence,
$i-k\geq\left\lfloor\frac{n-i-1}{m-j}\right\rfloor+1-c_{R}$.

After changing the range of the third loop (line 8, Algorithm 5) to
$[\max\\{(i-c_{R})\cdot(1-\frac{1}{j}),0\\},\max\\{i-\left\lfloor\frac{n-i-1}{m-j}\right\rfloor-1+c_{R},1\\}]$,
we can skip the calculation of $V[i][j]$ when
$(i-c_{R})\cdot(1-\frac{1}{j})>i-\left\lfloor\frac{n-i-1}{m-j}\right\rfloor-1+c_{R}$.
This pruning reduces the time complexity to $O(n^{2}\log m)$.

Speedup by Divisible Property. To utilize the divisible property, we initially
put the first $(n\mod c_{R})$ records in the first partition, and then change
the step size (line 5 and line 8) of Algorithm 5 to $c_{R}$. This way, we can
shrink the size of $V$ to $\left(\left\lceil
n/c_{R}\right\rceil+1\right)\times m$. Recall that $m$ can be as large as
$B-1$, $F$ is a constant larger than 1, and thus $c_{R}=\left\lfloor
b_{R}\cdot(B-2)/F\right\rfloor\geq m$. Therefore, the time complexity can be
reduced to $O\left(\left(n/c_{R}\right)^{2}\log
m\right)=O\left(\frac{n^{2}\log m}{m^{2}}\right)$ and the space complexity to
$O\left(n/c_{R}\cdot m\right)=O(n)$.

(a) Uniform

(b) Zipfian

Figure 4. A comparison of the number of passes required for the outer relation
$S$ between the partitioning adopted by DHH/GHJ and our optimal partitioning,
assuming $B\leq\sqrt{\|R\|\cdot F}$ (DHH without skew optimization downgrades
to GHJ), $n_{R}=1M$, $n_{S}=8M$, $F=1.02$ and $B=320$ pages. Moreover, the
record size of $R$ is $1KB$, thus $R$ occupies 250K pages. In (a), random
(uniform) partitioning (GHJ), the partition size is around $250K/319\approx
784$ pages which require $\lceil 783/(318\times F)\rceil=3$ passes to scan
every partition of $S$. In contrast, the optimal partitioning allows the
number of passes to vary from 2 to 6, which follows a non-monotonically
decreasing trend as CT value increases. When we have a skewed correlation (b),
a similar pattern is observed. Further, the optimal partitioning naturally
excludes records with $\textsf{CT}[i]=0$.

###  3.2. Optimal Partitioning With Caching

We move to the general case when records can be cached. In this case, some
records stay in memory during the partition phase, so as to avoid repetitive
reads/writes, while remaining records still go through the partitioning phase
as Section 3.1. A natural question arises: which records should stay in memory
and which records should go to partition? As we can see in DHH (Figure 2), the
hash function $h_{\textsf{split}}$, together with POB actually categorizes all
the join keys by their indexes into two sets, memory-resident keys as
$K_{\textsf{mem}}\subseteq[n]$ and disk-resident keys as
$K_{\textsf{disk}}=[n]-K_{\textsf{mem}}$. Then, it suffices to find an optimal
partitioning (without caching) for $K_{\textsf{disk}}$, which degenerates to
our problem in Section 3.1. Together, we derive a general join cost function
$\textsf{Join}(K_{\textsf{disk}},\mathbb{P},m)$ as:

$\left\lceil\sum\limits_{i\in
K_{\textsf{disk}}}\frac{\textsf{CT}[i]}{b_{S}}\cdot\lceil\frac{|\mathcal{N}_{f}(i)|}{c_{R}}\rceil\right\rceil+\left(1+\mu\right)\cdot\left\lceil\frac{|K_{\textsf{disk}}|}{b_{R}}\right\rceil+\mu\cdot\sum\limits_{i\in
K_{\textsf{disk}}}\left\lceil\frac{\textsf{CT}[i]}{b_{S}}\right\rceil$

where $b_{R}$ (resp. $b_{S}$) represents the number of records per page for
relation $R$ (resp. $S$), and $\mu$ is the ratio between random write and
sequential read. Compared to the original cost function
$\textsf{Join}(\mathbb{P},m)$, the new cost function
$\textsf{Join}(K_{\textsf{disk}},\mathbb{P},m)$ involves two additional terms
–
$\mu\cdot(\left\lceil|K_{\textsf{disk}}|/{b_{R}}\right\rceil+\left\lceil\sum_{i\in
K_{\textsf{disk}}}\textsf{CT}[i]/{b_{S}}\right\rceil)$ as the cost of
partitioning both relations, and
$\left\lceil|K_{\textsf{disk}}|/{b_{R}}\right\rceil$ as the cost of loading
$R$ in the probing phase. Together, we have the following integer program:

$\begin{split}\min_{K_{\textsf{disk}},\mathbb{P}}&\quad\textsf{Join}(K_{\textsf{disk}},\mathbb{P},m)\\\
\textrm{subject to}&\quad\left\lceil\frac{n-|K_{\textsf{disk}}|}{b_{R}}\cdot
F\right\rceil+m+2\leq B,\\\
&\quad\sum\limits_{j=1}^{m}\mathbb{P}_{i,j}=1,\forall i\in[n]\\\
&\quad\mathbb{P}_{i,j}\in\\{0,1\\},\forall i\in[n],j\in[m]\end{split}$

where the first constraint limits the size of the memory-resident hash table,
as $2$ pages are reserved for input and join output, and $m$ pages for the
output buffer for each disk-resident partition.

1 $\textsf{tmp}\leftarrow+\infty$, $f\leftarrow\emptyset$;

2 for  _$k\leftarrow 0$ to $c_{R}$_ do

3
$V\leftarrow\textbf{{Partition}}(\textsf{CT}[1:n-k],n-k,B-2-\lceil\frac{k\cdot
F}{b_{R}}\rceil)$;

4
$c_{\textsf{probe}}\leftarrow\lceil\frac{n-k}{b_{R}}\rceil+V[n-k][B-2-\lceil\frac{k\cdot
F}{b_{R}}\rceil].\textsf{cost}$;

5
$c_{\textsf{part}}\leftarrow\mu\cdot(\lceil\frac{n-k}{b_{R}}\rceil+\lceil\frac{1}{b_{S}}\cdot\sum_{i=1}^{n-k}\textsf{CT}[i]\rceil)$;

6 if  _$\textsf{tmp} <c_{\textsf{probe}}+c_{\textsf{part}}$_ then

7 $\textsf{tmp}\leftarrow c_{\textsf{probe}}+c_{\textsf{part}}$;

8 $f\leftarrow\textsc{GetCut}(V,n-k,B-2-\lceil\frac{k\cdot F}{b_{R}}\rceil)$;

9

10

11return $f$;

Algorithm 7 OCAP$(\textsf{CT},n,B)$

The integer program above can be interpreted by first fixing the
$K_{\textsf{disk}}$ (i.e., which keys/records should be spilled out to disk),
and then finding the optimal partitioning for $K_{\textsf{disk}}$. Our
Algorithm OCAP can be invoked as a primitive for given $K_{\textsf{disk}}$. It
remains to identify the optimal $K_{\textsf{disk}}$ (i.e., which records
should be spilled out to disk) that leads to the overall minimum cost.
Although there is an exponentially large number of candidates for
$K_{\textsf{disk}}$, it is easy to see that records with low CT value should
be spilled out to disk, while records with high CT values should be kept in
memory. In other words, the consecutive property of Theorem 3.1 still applies
for this hybrid partitioning – we can achieve the optimal cost by caching the
top-$k$ records from CT in memory, if restricting the size of
$K_{\textsf{mem}}$ to be $k$, where $k<c_{R}$ since at most $c_{R}$ records
can remain in memory. By extending Algorithm 5 to a hybrid version, we finally
come to the OCAP in Algorithm 7.

Complexity Analysis. As the number of records that can remain in memory is at
most $c_{R}=\lfloor b_{R}\cdot(B-2)/F\rfloor$, we run Algorithm 5 at most
$c_{R}$ times. Recall that the complexity of Algorithm 5 can be reduced to
$O(n^{2}\log m/c_{R}^{2})$, the time complexity for OCAP is now $O(n^{2}\log
m/c_{R})=O(n^{2}\log m/m)$. The space complexity is $O(n)$.

##  4\. Our Practical Algorithm

In practice, it is impossible to know the exact correlation between input
relations for each key in advance; instead, a much smaller set of high-
frequency keys are collected for skew optimization, such as the Most Common
Values (MCVs) in PostgreSQL (PostgreSQL, [n. d.]). PostgreSQL supports skew
optimization by assigning 2% memory to build a hash table for skewed keys if
their total frequency is larger than 1% of the outer relation, but this
heuristic uses fixed thresholds to trigger skew optimization, which may be
sub-optimal for arbitrary correlation skewness. Hence, we present a practical
algorithm built upon the theoretically I/O-optimal algorithm in the previous
section. Instead of relying on the whole correlation CT table, our practical
algorithm only needs the same amount of MCVs as PostgreSQL and generates a
hybrid partitioning that can be adaptive to arbitrary correlation skewness,
and constrained memory budget.

###  4.1. Hybrid Partitioning

Let $K$ be the set of skew keys tracked from MCVs with their CT values known.
Guided by the consecutive property and weakly-ordered property in Theorem 3.1,
it is always efficient to keep as many skew keys as possible in memory if
possible, so that we can avoid reading the corresponding records in $S$
repeatedly. For the rest of keys not tracked by MCVs, we consider their
frequency as low and can be tackled well by the traditional hashing method
GHJ/DHH.

Framework. In a hybrid partitioning, we design a hash set $HS_{\textsf{mem}}$
to store a subset of skew key (denoted as $K_{\textsf{mem}}\subseteq K$) that
will be used to build the in-memory hash table $HT_{\textsf{mem}}$, a hash map
$f_{\textsf{disk}}$ to store a subset of skew keys (denoted as
$K_{\textsf{disk}}\subseteq K$) with their designated partition identifiers,
and the rest of skew keys (if there exists any) together with non-skew keys
(denoted as $K_{\textsf{rest}}$) will be uniformly partitioned by GHJ/DHH with
a certain memory budget $m_{\textsf{rest}}$ (in pages). With
$HS_{\textsf{mem}}$, $f_{\textsf{disk}}$ and $m_{\textsf{rest}}$ in hand, we
can partition records in $R$ using Algorithm 8. For each record in $R$, if its
key is in $HS_{\textsf{mem}}$, we put the entire record in the in-memory hash
table $HT_{\textsf{mem}}$ (the hash set $HS_{\textsf{mem}}$ only stores the
key); if its key belongs to $f_{\textsf{disk}}$, we put it into the partition
specified by $f_{\textsf{disk}}$, which will be spilled to disk later;
otherwise, it will be assigned by GHJ/DHH along with the rest of the records
using the $m_{\textsf{rest}}$ pages reserved. We can partition records in $S$
similarly using Algorithm 9. This hybrid partitioning workflow is visualized
in Figure 5.

Figure 5. An illustration of the hybrid partitioning workflow of NOCAP.
Partitions $P_{4}$ and $P_{7}$ are spilled to disk when invoking DHH, while
$P_{1}$ and $P_{2}$ are written to disk due to designated partitioning by
$f_{disk}$. Other records in $R$ either have keys from $K_{\textsf{mem}}$ or
stage in memory when invoking DHH, which will be inserted to the in-memory
hash table $HT_{\textsf{mem}}$. Figure 6. The memory breakdown when
partitioning $R$.

1 $R^{\prime}\leftarrow\\{r\in R:r.\textsf{key}\notin
HS_{\textsf{mem}},r.\textsf{key}\notin f_{\textsf{disk}}\\}$; $\triangleright$
Logically construct $R^{\prime}$ without instantiation;

2 foreach  _$r\in R-R^{\prime}$_ do

3 if  _$r.\textsf{key}\in HS_{\textsf{mem}}$_ then add $r$ to
$HT_{\textsf{mem}}[h_{\textsf{probe}}(r.\textsf{key})]$;

4 else assign $r$ to disk-resident partition
$f_{\textsf{disk}}(r.\textsf{key})$;

5

6$(HT^{\prime}_{\textsf{mem}},{\textsf{POB}})\leftarrow$ DHH-
Primary$(R^{\prime},m_{\textsf{rest}})$;

return $(HT^{\prime}_{\textsf{mem}}\cup HT_{\textsf{mem}},{\textsf{POB}})$

Algorithm 8 $\textsc{Hybrid-Partition-
Primary}(R,HS_{\textsf{mem}},f_{\textsf{disk}},m_{\textsf{rest}})$

1 $S^{\prime}\leftarrow\\{s\in S:s.\textsf{key}\notin f_{\textsf{disk}}\\}$;
$\triangleright$ Logically construct $S^{\prime}$ without instantiation;

2 foreach  _$s\in S^{\prime}$_ do

3 if  _$s.\textsf{key}\in f_{disk}$_ then

4 assign $s$ to disk-resident partition $f_{\textsf{disk}}(s.\textsf{key})$

5

6DHH-Foreign$(S-S^{\prime},HT_{\textsf{mem}},{\textsf{POB}})$;

Algorithm 9 $\textsc{Hybrid-Partition-
Foreign}(S,HT_{\textsf{mem}},f_{\textsf{disk}},{\textsf{POB}})$

Regarding whether to use GHJ or DHH, we note the following difference. When
the available memory is large, say
$\lceil|K_{\textsf{rest}}|/c_{R}\rceil<m_{\textsf{rest}}$, we simply invoke
DHH to partition the rest of keys $K_{\textsf{rest}}$ using
$m_{\textsf{rest}}$ pages. When the available memory is small, as discussed in
Section 2.2, the DHH just downgrades to the GHJ.

Enforcing Memory Constraints. State-of-the-art systems (e.g., MySQL and
PostgreSQL) typically assign to each join operator a user-defined memory
budget. The default memory limit for the join operator in MySQL is 256 KB
(MySQL, 2021) and the one in PostgreSQL is 4 MB (PostgreSQL, 2022). Assuming
the available buffer is in total $B$ pages, we need to ensure that Algorithm 8
uses no more than $B$ pages. We present a memory breakdown as illustrated in
Figure 6.

  * •

$B_{HS}(K_{\textsf{mem}})$:#pages for the hash set $HS_{\textsf{mem}}$ with
keys in $K_{\textsf{mem}}$;

  * •

$B_{HT}(K_{\textsf{mem}})$:#pages for the hash table $HT_{\textsf{mem}}$ with
records whose keys are in $K_{\textsf{mem}}$;

  * •

$B_{f}(K_{\textsf{disk}})$: #pages for the hash map $f_{\textsf{disk}}$ with
keys in $K_{\textsf{disk}}$;

  * •

$m_{\textsf{disk}}$: #pages as write buffer for all the on-disk partitions
specified in $f_{\textsf{disk}}$ (which is the same as #partitions specified
in $f_{\textsf{disk}}$);

  * •

$m_{\textsf{rest}}$: #pages to partition records whose keys are in
$K_{\textsf{rest}}$;

Moreover, we denote $ks$ as the size of each key (in bytes), and $ps$ as the
size of each page (in bytes). We can roughly have
$B_{HT}(K_{\textsf{mem}})=\lceil b_{R}\cdot|K_{\textsf{mem}}|/F\rceil$,
$B_{HS}(|K_{\textsf{mem}}|)=\lceil ks\cdot|K_{\textsf{mem}}|/(F\cdot
ps)\rceil$, and
$B_{f}(|K_{\textsf{disk}}|)=\lceil(ks+4)\cdot|K_{\textsf{disk}}|/(F\cdot
ps)\rceil$, where a 4-byte integer is used to store the partition identifier
in the hash map $f_{\textsf{disk}}$. At last, we come to the memory constraint
subject to $B$ (the total number of pages):

$B_{HS}(|K_{\textsf{mem}}|)+B_{HT}(|K_{\textsf{mem}}|)+B_{f}(|K_{\textsf{disk}}|)+m_{\textsf{disk}}+m_{\textsf{rest}}\leq
B-2$

Cost function. Combining all these above together, we come to the cost
function and integer programming as follows:

$\displaystyle\min_{K_{\textsf{mem}},K_{\textsf{disk}},\mathbb{P}_{\textsf{disk}}}\
\ $ $\displaystyle\
\textsf{Join}(K_{\textsf{disk}},\mathbb{P}_{\textsf{disk}},m_{\textsf{disk}})+g_{\textsf{DHH}}(|K_{\textsf{disk}}|,m_{\textsf{rest}})$
s.t. $\displaystyle\
B_{HS}(|K_{\textsf{mem}}|)+B_{HT}(|K_{\textsf{mem}}|)+B_{f}(|K_{\textsf{disk}}|)\leq
B-2-m_{\textsf{disk}}-m_{\textsf{rest}},$ $\displaystyle 0\leq
m_{\textsf{disk}},0\leq m_{\textsf{rest}}\leq B-2,$ $\displaystyle
K_{\textsf{mem}}\cap K_{\textsf{disk}}=\emptyset,K_{\textsf{mem}}\cup
K_{\textsf{disk}}\subseteq K$
$\displaystyle\sum\limits_{j=1}^{m}\mathbb{P}_{i,j}=1,\forall i\in[n]$
$\displaystyle\mathbb{P}_{i,j}\in\\{0,1\\},\forall i\in[n],\forall j\in[m]$

In the objective function, the first term
$\textsf{Join}(K_{\textsf{disk}},f_{\textsf{disk}},m_{\textsf{disk}})$ is the
cost of partitioning $K_{\textsf{disk}}$ into $m_{\textsf{disk}}$ partitions,
and the second term $g_{\textsf{DHH}}(|K_{\textsf{rest}}|,m_{\textsf{rest}})$
is the estimated cost of partitioning $K_{\textsf{rest}}$ with
$m_{\textsf{rest}}$ pages using either GHJ or DHH, depending on how large
$m_{\textsf{rest}}$ is, as mentioned earlier. We will discuss
$g_{\textsf{DHH}}(|K_{\textsf{disk}}|,m_{\textsf{rest}})$ in Section 4.2. In
both estimation models of GHJ or DHH, we need to know the total number of
records from $S$ whose key falls into $K_{\textsf{rest}}$, which can be
estimated by $n_{S}-\sum_{i\in K_{\textsf{mem}}\cup
K_{\textsf{disk}}}\textsf{CT}[i]$, since the CT values of keys in $K$ are
known from MCVs.

To find out the best combination of $K_{\textsf{mem}}$, $K_{\textsf{rest}}$
and $f_{\textsf{disk}}$ (the optimal value of $m_{\textsf{rest}}$ can be
calculated using Equation (4.1)), we iterate all possible combinations of
$|K_{\textsf{mem}}|$, $|K_{\textsf{disk}}|$ and $f_{\textsf{disk}}$, and pick
the one with the minimum estimated cost, as described by Algorithm 10. Then,
implied by the consecutive property and weakly-ordered property in Theorem 3.1
(i.e., keys with higher frequency should be kept in memory or in a small
partition with higher priority), we always pick the top-$|K_{\textsf{mem}}|$
keys from $K$ as $K_{\textsf{mem}}$ and the next top-$|K_{\textsf{disk}}|$
keys from $K$ as $K_{\textsf{disk}}$.

1 $c_{opt}\leftarrow+\infty$, $k_{\textsf{mem}}\leftarrow 0$,
$k_{\textsf{disk}}\leftarrow 0$, $f_{\textsf{disk}}\leftarrow\emptyset$;

2 for  _$i_{1}\leftarrow 0$ to $\min(k,c_{R})$_ do

3
$V\leftarrow\mathbf{\textsc{Partition}}(\textsf{CT}[i_{1}+1:k],k-i_{1},\left\lceil\frac{k-i_{1}}{c_{R}}\right\rceil)$;

4 for  _$i_{2}\leftarrow 0$ to $\min(k,c_{R})-i_{1}$_ do

5 for  _$j\leftarrow\min(i_{2},1)$ to $\lceil\frac{i_{2}}{c_{R}}\rceil$_ do

6 $m_{\textsf{rest}}\leftarrow
B-2-B_{HT}(i_{1})-B_{HS}(i_{1})-B_{f}(i_{2})-j$;

7 $c_{\textsf{probe}}\leftarrow V[i_{2}][j].\textsf{cost}$;

8
$c_{\textsf{part}}\leftarrow\mu\cdot\left(\lceil\frac{i_{2}}{b_{R}}\rceil+\lceil\frac{1}{b_{S}}\cdot\sum_{j=i_{1}+1}^{j=i_{2}}\textsf{CT}[j]\rceil\right)$;

9 $c_{\textsf{{rest}}}\leftarrow
g_{\textsf{DHH}}(i_{2}+1,n,m_{\textsf{rest}})$;

10 $c_{\textsf{tmp}}\leftarrow
c_{\textsf{probe}}+c_{\textsf{part}}+c_{\textsf{rest}}$;

11 if  _$c_{\textsf{tmp}} <c_{\textsf{opt}}$_ then

12 $c_{opt}\leftarrow c_{\textsf{tmp}}$, $k_{\textsf{mem}}\leftarrow
i_{1},k_{\textsf{disk}}\leftarrow i_{2}$;

13 $f_{\textsf{disk}}\leftarrow\mathbf{\textsc{GetCut}}(V,i_{2},j)$;

14

15

16

17

18return $k_{\textsf{mem}}$, $k_{\textsf{disk}}$, $f_{\textsf{disk}}$;

Algorithm 10 $\textsc{NOCAP}(\textsf{CT},k,n,m)$

###  4.2. Optimizations with Rounded Hash

In our implementation, we also design rounded hash (RH) in partitioning that
can further reduce the I/O cost of scanning the outer relation in the per-
partition join. This is motivated by the divisible property in Theorem 3.1,
that the sizes of most partitions should be divisible by $c_{R}$ in the
optimal partitioning. To understand why this insight may help, let us consider
an example in Figure 7.

We note that uniform partitioning namely assigns records using a uniform hash
function $\textsf{PartID}=\textsf{hash}(\textsf{key})\mod m$, based on the
modular function with respect to $m$ (the number of partitions), which is
denoted as plain hash (PH). To reduce the extra I/O cost using non-uniform
partitioning, our RH uses an additional modular function with respect to
$\left\lceil n/c_{R}\right\rceil$:

(2) $\textsf{PartID}=\left(\textsf{hash}(\textsf{key})\mod\left\lceil
n/c_{R}\right\rceil\right)\mod m$

where $c_{R}=\left\lfloor b_{R}\cdot(B-2)/F\right\rfloor$ is the chunk size of
records. Intuitively, our RH maximizes the number of partitions with
$\left\lfloor\left\lceil n/{c_{R}}\right\rceil/(B-1)\right\rfloor$ chunks. In
addition to reducing unnecessary passes, RH also merge small partitions to
avoid fragmentation when partitioning (Kitsuregawa et al., 1989).

Figure 7. An illustration of the uniform and non-uniform partitioning on the
inner relation $R$, with a 5-page buffer (with 4 pages for partitioning $R$
and 1 page for streaming the input). In NBJ, the maximum #chunks loaded at a
time is 3-page long (with 1 page for streaming $S$ and $1$ page for the output
assuming $F=1.0$). Uniform partitioning leads to 4 partitions with each having
$18/4=4.5$ pages worth of data. One needs to read $S$ twice for each
partition. As a comparison, rounded hashing with $n/{c_{R}}=6$ and $m=4$
generates 2 partitions with each having 6 pages, and 2 partitions with each
having 3 page worth of data. In that way, partitions 1 and 2 need two passes
on the corresponding partitions in $S$, while partitions 3 and 4 only need a
single pass.

Parametric Optimization. Since we rely on a hash function to partition
records, partitions 1 and 2 in Figure 7 could occasionally “overflow” and grow
larger than six pages due to randomness. In a skewed workload, overflowed
partitions could result in higher cost compared to the cost of uniform
partitioning. In fact, a more robust scheme in Figure 7 is to assign 2.5 pages
to partition 4 and evenly distribute 15.5 pages to partitions 1, 2, and 3.
Formally, we replace $c_{R}$ with $c_{R}^{*}=\beta\cdot c_{R}$ in Equation
(2), where $\beta$ should be very close to 1 in the range $(0,1]$ (we fix
$\beta=0.95$ in our implementation).

Cost Estimation. To get a rough estimation of how many I/Os are used by RH, we
first build the I/O model for PH-based partitioning. We actually consider a
generalized problem by estimating the per-partition cost for partitioning
records lying on the CT array from index $s$ to index $e$ into $m$ partitions
using PH, noted by $g_{PH}(s,e,m)$:

$g_{\textsf{PH}}(s,e,m)=\sum_{j=1}^{m}\mathbb{E}\left[\left\lceil\frac{|P_{j}|}{c_{R}}\right\rceil\right]\cdot\sum\limits_{i\in
P_{j}}\textsf{CT}[i]=\left\lceil\frac{e-s+1}{m\cdot
c_{R}}\right\rceil\cdot\sum\limits_{i=s}^{e}\textsf{CT}[i]$

where each $\|P_{j}\|$ is approximated as a Poisson distribution with
$\lambda=\frac{e-s+1}{m}$. When estimating the cost of RH we need to know how
much proportion (noted by $\gamma$) of data from $S$ is scanned with one fewer
pass ($1-\gamma$ of data is scanned with one more pass):

$\gamma=\left(\left(\left\lceil\frac{e-s+1}{c_{R}^{*}}\right\rceil\text{ mod
}m\right)\cdot\left\lfloor\frac{e-s+1}{m\cdot c_{R}^{*}}\right\rfloor\cdot
c_{R}^{*}\right)/(e-s+1)$

As such, the normalized number of rounded passes is:

$\\#\textsf{rounded}\\_\textsf{passes}(s,e)=\gamma\cdot\left\lfloor\frac{e-s+1}{m\cdot
c_{R}^{*}}\right\rfloor+(1-\gamma)\cdot\left\lceil\frac{e-s+1}{m\cdot
c_{R}^{*}}\right\rceil$

We then have our cost model for rounded hash $g_{\textsf{RH}}$:

(3)
$g_{\textsf{RH}}(s,e,m)=\\#\textsf{rounded}\\_\textsf{passes}(s,e)\cdot\sum\limits_{i=s}^{e}\textsf{CT}[i]$

The cost of our original problem is captured by $g_{\textsf{RH}}(1,n,m)$.

Overestimation using Chernoff Bound. When the filling percent for each
partition using PH reaches the predefined threshold $\beta$,
($\frac{e-s+1}{m}>\beta\cdot t\cdot c_{R}$, where $t$ is the largest positive
integer such that $t\cdot c_{R}>\frac{e-s+1}{m}$), we disable RH. This is to
avoid additional passes that are triggered by occasional overflow in the
random process. We also need to take into account the possible overflow in the
cost model. In fact, every element is distributed independently and
identically using the same hash function, and thus we can apply Chernoff bound
to overestimate the probability. For an arbitrary partition, we define
$X=\sum_{i=s}^{e}X_{i}$ where $X_{i}=1$ (i.e., $i$-th element is assigned to
this partition) with probability $\frac{1}{m}$, and $X_{i}=0$ with probability
$1-\frac{1}{m}$. Then $\mathbb{E}[X]=\frac{e-s+1}{m}$. From Chernoff bound,
$\Pr[X>t\cdot
c_{R}]<\left(\frac{e^{\sigma}}{(1+\sigma)^{1+\sigma}}\right)^{\mathbb{E}[X]}$,
where $\sigma=\frac{t\cdot c_{R}\cdot m}{e-s+1}-1$. We thus let
$1-\gamma=\left(\frac{e^{\sigma}}{(1+\sigma)^{1+\sigma}}\right)^{\mathbb{E}[X]}$
and set:

$\\#\textsf{rounded}\\_\textsf{passes}(s,e)=\gamma\cdot\Bigg{\lceil}\frac{e-s+1}{m\cdot
c_{R}^{*}}\Bigg{\rceil}+(1-\gamma)\cdot\left(\Bigg{\lceil}\frac{e-s+1}{m\cdot
c_{R}^{*}}\Bigg{\rceil}+1\right)$

We reuse Equation (3) to estimate the I/O cost.

##  5\. Experimental Analysis

(a) Zipf ($\alpha$=$1.3$)

(b) Zipf ($\alpha$=$1.0$)

(c) Zipf ($\alpha$=$0.7$)

(d) Uniform

(e) Zipf ($\alpha$=$1.3$) [w/o sync]

(f) Zipf ($\alpha$=$1.0$) [w/o sync]

(g) Zipf ($\alpha$=$0.7$) [w/o sync]

(h) Uniform [w/o sync]

(i) Zipf ($\alpha=1.3$) [w/ sync]

(j) Zipf ($\alpha$=$1.0$) [w/ sync]

(k) Zipf ($\alpha$=$0.7$) [w/ sync]

(l) Uniform [w/ sync]

Figure 8. While state-of-the-art skew optimization in DHH helps reduce the I/O
cost, NOCAP can better exploit the correlation skew to achieve even lower I/O
cost and latency. The benefit of NOCAP is more pronounced with skew
correlation.

We now present experimental results for our practical method.

Approaches Compared. We compare NOCAP with Grace Hash Join (GHJ), Sort-Merge
Join (SMJ), and Dynamic Hybrid Hash join (DHH). For all the partitioning-based
methods (GHJ, DHH, and ours), we apply a light optimizer that picks the most
efficient algorithm according to Table 1 in the partition-wise join. We also
augment GHJ by allowing it to fall back to NBJ if the latter has a lower cost.
For DHH, we follow prior approaches that use fixed thresholds to trigger skew
optimization (2% of the available memory is used for an in-memory hash table
for skew keys if the sum of their frequency is larger than 2% of the relation
size). We also compare Histojoin (by setting the trigger frequency threshold
as zero) as one of our baselines. All the approaches are implemented in C++,
compiled with gcc 10.1.0, in a CentOS Linux with kernel 4.18.0. For NOCAP,
DHH, and Histojoin, we assume top $k=50K$ frequently matching keys are tracked
(i.e., 5% of the keys when $n=1M$). In PostgreSQL’s implementation of DHH, the
frequent keys are stored using a small amount of system cache, which does not
consume the user-defined working memory budget unless they are inserted into
the in-memory hash table. Similarly, in our prototype, the top-$k$ keys are
given. Due to the pruning techniques (§3.1.3), computing the partitioning
scheme (Alg. 10) with $k=50K$ takes less than one second, so we omit its
discussion.

###  5.1. Sensitivity Analysis

We now experimental results with synthetic data produced by our workload
generator that allows us to customize the distribution of matching keys
between two input relation to test the robustness of different join methods.

Experimental Setup. We use our in-house server, which is equipped with 375GB
memory in total and two Intel Xeon Gold 6230 2.1GHz processors, each having 20
cores with virtualization enabled. For our storage, we use a 350GB PCIe P4510
SSD with direct I/O enabled. We can change the read/write asymmetry by turning
the `O_SYNC` flag on and off. When `O_SYNC` is on, every write is ensured to
flush to disk before the write is completed, and thus has higher asymmetry
($\mu_{sync}=3.3$, $\tau_{sync}=3.2$). When it is off, we have lower asymmetry
$(\mu_{no\\_sync}=1.28$, $\tau_{no\\_sync}=1.2)$. By default, sync I/O is off
to accelerate joins.

Experimental Methodology. We first experiment with a synthetic workload, which
contains two tables $R$ and $S$ with $n_{R}=1M$ and $n_{S}=8M$. The record
size is $1$KB for both $R$ and $S$, and thus we have $\|R\|=250K$ pages and
$\|S\|=2M$ pages (the page size is fixed as $4$KB in all the experiments). In
this experiment, we vary the buffer size from
$0.5\cdot\sqrt{F\cdot\|R\|}\approx 256$ pages to $\|R\|=250K$ pages. We use
#I/Os (reads + writes) and latency as two metrics, and when comparing #I/Os,
we also run OCAP (§3.2) to plot the optimal (lower bound) #I/Os . Further, we
run our experiments under uniform correlation and Zipfian correlation (short
as Zipf) with $\alpha=0.7,1.0,1.3$.

NOCAP Dominates for any Correlation Skew and any Memory Budget. For all the
join methods, #I/Os decreases as the buffer size increases, as shown Figures
8(a)-8(d). We also observe that NOCAP achieves the lowest #I/Os (near-optimal)
among all the join algorithms for any correlation skew and any memory budget.
When we compare latency from Figures 8(e)-8(h), we also conclude that GHJ and
SMJ have a clear gap while they have nearly the same #I/Os regardless of the
join correlation. This is because random reads in SMJ are $1.2\times$ slower
than sequential reads in GHJ. While two traditional join methods have no
optimization for skewed join correlations, DHH, Histojoin, and NOCAP can take
advantage of correlation skew to alleviate #I/Os. However, DHH and Histojoin
cannot fully exploit the data skew because it limits the space for high-
frequency keys with a fixed threshold ($2\%$ of the total memory budget).
Compared to DHH and Histojoin, NOCAP normally has fewer #I/Os because it
freely decides the size of the in-memory hash table for frequent keys without
any predefined thresholds. Figures 8(f)-8(h) show that the latency gap between
NOCAP and DHH (Histojoin) is more pronounced when we have medium skew
($\alpha=1$ or $\alpha=0.7$), compared to the uniform correlation. However,
when the correlation is highly skewed ($\alpha=1.3$), the small hash table (2%
of the available memory) in DHH is enough to capture the most frequent keys
and, hence, issue near-optimal #I/Os (Figure 8(a)). As a result, the benefit
of NOCAP for $\alpha=1.3$ (Figure 8(e)) is smaller than for $\alpha=1$ (Figure
8(f)). Nevertheless, NOCAP still issues 30% fewer #I/Os when the memory budget
is small ($2^{8}$ pages).

(a) Uniform

(b) Zipf ($\alpha$=$1.0$)

Figure 9. When the memory is limited, NOCAP can even reach 10% speedup with
uniform workload.

(c) Uniform

(d) Zipf ($\alpha$=$0.7$)

Figure 10. With noisy MCVs, both DHH and NOCAP have similar performance to
Figures 8(h) and 8(g).

(a) $B=2MB$

(b) $B=32MB$

Figure 11. DHH requires careful tuning to achieve its best performance, which
is still slower than NOCAP.

(a) $\sigma_{S}$=$0.488,SF$=$10$

(b) $\sigma_{S}$=$0.63,SF$=$10$

(c) $\sigma_{S}$=$0.488,SF$=$50$

(d) $\sigma_{S}$=$0.63,SF$=$50$

Figure 12. TPC-H: NOCAP leads to higher speedup when more data from $S$ are
joined in a query.

DHH Cannot Adapt to Different Memory Budget. The skew optimization in DHH
relies on two thresholds (memory budget and minimum frequency) and thus it
cannot easily adapt to a different memory budget. To verify this, we compare
DHH with NOCAP under a given Zipfian correlation ($\alpha=0.7$) using as
memory budgets $2$MB and $32$MB. Figures 11(a) and 11(b) show the the
percentage of reduced #I/Os using NOCAP based on DHH when we vary the two
thresholds used by DHH. When $B=2$MB, the best DHH (the darkest cell) does not
trigger skew optimization by setting a high frequency trigger threshold or the
memory budget as zero. In contrast, when $B=32$MB, the best DHH assigns 8%
memory to build the hash table for skewed keys. Note that even though DHH can
achieve close-to-NOCAP performance by varying these two thresholds, this
relies on well-tuned parameters, which may not be feasible when the workload
and the buffer size changes.

NOCAP Outperforms DHH Even for Uniform Correlation with Small Memory Budget.
We also examine the speedup when we have a lower memory budget. Although
Figures 8(l) and 8(h) show that there _seems_ no difference between DHH and
NOCAP when the workload is uniform, we actually capture a larger difference
after we narrow down the buffer range (128$\sim$512 pages). As shown in Figure
9(a), NOCAP can even achieve up to 15% and 10% speedup when there are 480 and
352 pages. The step-wise pattern of DHH and GHJ comes from random (uniform)
partitioning. When the memory is smaller than $\sqrt{\|R\|\cdot F}$, uniform
partitioning easily makes each partition larger than the chunk size, as
illustrated in Figure 7. On the contrary, rounded hash allows some partitions
to be larger so that other partitions have fewer chunks and thus require fewer
passes to complete the partition-wise joins. When the correlation becomes more
skewed, NOCAP can be more than 30% faster than DHH under limited memory, as
shown in Figure 9(b).

DHH and NOCAP Both Exhibit Robustness for Noisy MCVs. To examine how noisy CT
can affect DHH and NOCAP, we add Gaussian noise to the CT values, with the
average noise as 0 (the value of the average noise has no impact over MCVs)
and $\sigma=n_{S}/n_{R}$. We reuse most of experimental settings so that
$n_{S}/n_{R}=8$ and thus
$\textsf{CT}_{noise}[i]\in[\textsf{CT}[i]-8,\textsf{CT}[i]+8]$ with
probability 68%. We redo the experiments with uniform and Zipfian jon
correlation ($\alpha=0.7$), and the results are shown in Figure 10. We do not
observe a significant difference between the results with noisy CT values and
the original results in Figures 8. The rational behind this is that keys with
high frequency are still likely to be prioritized (cached) during hybrid
partitioning even after we add the Gaussian noise, thus having very similar
performance to the results without noise.

###  5.2. Experiments with TPC-H, JCC-H, and JOB

We now experiment with TPC-H (TPC, 2021), JCC-H (Boncz et al., 2017), and JOB
(Leis et al., 2015), using DHH as our main competitor.

Experimental Setup. We experiment with a modified TPC-H Q12. Q12 selects data
from table `lineitem`, and joins them with `orders` on
`l_orderskey=o_orderskey`, followed by an aggregation. As we have already
seen, DHH has only a minor difference from NOCAP when the correlation is
uniform. Here, we focus on a skew correlation in TPC-H data. To achieve this,
we modify the TPC-H dbgen codebase – all the keys are classified into _hot_ or
_cold_ keys, and the frequency of hot keys and cold keys, respectively,
follows two different uniform distributions, which controls the overall skew
and the average matching keys (Chatterjee et al., 2022; Bruck et al., 2006).
In this experiment, we focus on a skew correlation where 0.5% of the
`o_orderkey` keys match 500 `lineitem` records on average and the rest keys
only match 1.5 `lineitem` records on average.

(a) JCC-H (Tuned Skew)

(b) JCC-H (Original Skew)

(c) JOB (cast $\bowtie$ title)

(d) JOB (cast $\bowtie$ name)

Figure 13. While DHH can perform as close as NOCAP, NOCAP is more adaptive
when the workload varies.

We further remove the filtering condition on `l_shipmode` and `l_receiptdate`
so that the size of the filtered `lineitem` is larger than `orders` (since we
focus on the case when $\|R\|<\|S\|$). We then have two conditions left,
`l_receiptdate`¡`l_shipdate` and `l_shipdate`¡`l_commitdate`, which has
separately selectivity $0.488$ and $0.63$. We choose one of filter conditions
to vary the selectivity. We run our experiments with these two selectivity and
two scale factors ($SF$=$10$, $SF$=$50$), as shown in Figure 12. In these
experiments, we employ storage-optimized AWS instances (i3.4xlarge) and we
turn off `O_SYNC`. The device asymmetry is $\mu=1.2$ and $\tau=1.14$.

NOCAP Accelerates TPC-H Joins. We observe that the difference in the total
latency between NOCAP and DHH is not so large as we see in previous
experiments on skew correlation in Figure 12(a). In fact, we find that the
proportion of time spent on I/Os is lower in the TPC-H experiment due to extra
aggregations in Q12. In earlier experiments (e.g., Figures 8(f) and 8(g)), the
time spent on CPU is 10$\sim$20 seconds out of a few minutes (the total
latency). In contrast, we observe that in both Figures 12(a) and 12(b), the
proportion of time spent on I/Os is less than 85%. This explains why the
speedup of NOCAP diminishes when we have more CPU-intensive operations (e.g.,
aggregation) in a query. In addition, when we adjust the selectivity so that
more data from `lineitem` are involved in the join, we observe a higher
benefit in Figure 12(b). We have similar observation when we move to a larger
data set (SF=50), as shown in Figures 12(c) and 12(d). In fact, in both cases,
when we have larger `lineitem` data involved in the join, NOCAP usually leads
to higher speedup.

NOCAP Outperforms DHH for JCC-H and JOB. To further validate the effectiveness
of NOCAP, we experiment with JCC-H (Join Cross Correlation) and JOB (Join
Order Benchmark). JCC-H is built on top of TPC-H with adding join skew, so we
reuse our TPC-H experimental setup (running the modified Q12 with $SF=10$ and
$\sigma_{S}=0.488$). In the original JCC-H dataset, the majority (i.e.,
$300K\cdot SF$) of `lineitem` records only join with 5 `orders` records. To
vary the skew, we tune the code by allowing around $5100\cdot SF$ `orders`
records to match 600 `lineitem` records on average. In addition to JCC-H, we
also compare DHH and NOCAP by executing a PK-FK join between two tables from
the JOB dataset. Specifically, `cast_info` stores the cast info between movies
and actors, which are respectively stored in tables `title` and `name`. The
number of movies per actor is highly skewed where the top 50 actors match 0.6%
records in `cast_info`, while the number of actors per movie is less skewed
where the top 50 movies match less than 0.1% records in `cast_info` (“top” in
terms of the number of occurrences in `cast_info`). As we can observe in
Figures 13a-d, when the correlation is extremely skewed (i.e., the original
JCC-H and `cast_info` $\bowtie$ `name`), DHH with fixed thresholds can
identify the skew and exploit it to achieve close-to-NOCAP performance.
However, when it comes to medium skewed (i.e., tuned JCC-H and `cast_info`
$\bowtie$ `title`), DHH with fixed thresholds results in suboptimal
performance compared to NOCAP.

##  6\. Discussion

General (Many-to-Many) Joins. In many-to-many joins, we have two CT lists that
store the frequency per key in each relation, and thus we need to create a new
cost function (with two CT lists) to use as the main objective function.
Despite that, we can still reuse our dynamic programming algorithm 5 and
replace CalCost by considering extra CT values, however, the error bound is no
longer guaranteed since our main theorem does not hold. In practice, even this
crude approximation of the optimal partitioning for general joins is worth
exploring since the CPU cost of enumerating partitioning schemes is very low,
and it covers the design space of DHH with variable values for its two
thresholds. As a result, NOCAP is still expected to outperform DHH for general
joins. We leave the complete formulation for many-to-many joins as future
work.

On-the-fly Sampling. On-the-fly sampling can give us more accurate CT values
because it takes into account the applied predicate while MCVs rely only on
the estimated selectivity. However, relying on sampling without knowledge of
MCVs makes the partitioning operation not pipelineable with the scan/select
operator. This is because partitioning relies on the exact join correlation,
but in case of sampling we need one complete pass first to acquire this
information. In this case, we may downgrade NOCAP into DHH to avoid the
additional pass. Another approach proposed by Flow-join (Rödiger et al., 2016)
is to employ a streaming sketch to obtain an approximate distribution during
partitioning. However, this requires scanning $S$ first to build the
histogram, which conflicts with partitioning $R$ first in the hybrid join. On
the other hand, if we already have MCVs (note that it is acceptable to be
noisy), sampling on-the-fly to enable sideways information passing (SIP) (Orr
et al., 2019) can be helpful. Specifically, we can sample $R$ on-the-fly
during partitioning and build a Bloom Filter, which is later used when
partitioning/scanning $S$. The additional Bloom Filter makes the estimated CT
more accurate, and thus the partitioning generated by NOCAP can get closer to
expected #I/O cost. However, adding a BF in the memory also introduces new
trade-offs between performance and memory consumption.

##  7\. Related Work

In-Memory Joins. When the memory is enough to store the entire relations, in-
memory join algorithms can be applied (Shatdal et al., 1994; Liu and Blanas,
2015; Schuh et al., 2016; Pohl et al., 2020; Bandle et al., 2021; Barber et
al., 2014). Similar to storage-based joins, in-memory join algorithms can be
classified as hash-based and sort-based (e.g., MPSM (Albutiu et al., 2012) and
MWAY (Balkesen et al., 2013a)). Hash joins can be further classified into
partitioned joins (e.g., parallel radix join (Balkesen et al., 2013a)) and
non-partitioned joins (e.g., CHT (Balkesen et al., 2013b)). GPUs can also be
applied to accelerate in-memory joins (Rui et al., 2015; Guo and Chen, 2019;
Lutz et al., 2020). Although storage-based joins focus on #I/Os, when two
partitions both fit in memory after the partitioning phase, NOCAP can still
benefit from optimized in-memory joins.

Distributed Equi-Joins. In a distributed environment, the efficiency of equi-
joins also relies on load balance and communication cost. When the join
correlation distribution is skewed, uniform partitioning may overwhelm some
workers by assigning excessive work to do (Walton et al., 1991). More
specifically, when systems like Spark (Zaharia et al., 2012) rely on in-memory
computations, the overall performance drops when the transmitted data for a
machine do not fit in memory. Several approaches (Xu et al., 2008; Li et al.,
2018; Beame et al., 2014; Rödiger et al., 2016) are proposed to tackle the
data (and thus, workload) skew for distributed equi-joins.

##  8\. Conclusion

In this paper, we propose a new cost model for partitioning-based PK-FK joins
that allows us to find the optimal partitioning strategy assuming accurate
knowledge of the join correlation. Using this optimal partitioning strategy,
we show that the state-of-the-art Dynamic Hybrid Hash Join (DHH) yields sub-
optimal performance since it does not fully exploit the available memory and
the correlation skew. To address DHH’s limitations, we develop a practical
near-optimal partitioning-based join with variable memory requirements (NOCAP)
which is based on OCAP. We show that NOCAP dominates the state-of-the-art for
any available memory budget and for any join correlation, leading to up to 30%
improvement.

###### Acknowledgements.

This material is based upon work supported by the National Science Foundation
under Grant No. IIS-2144547, a Facebook Faculty Research Award, and a Meta
Gift.

## References

  * (1)
  * Albutiu et al. (2012) Martina-Cezara Albutiu, Alfons Kemper, and Thomas Neumann. 2012\.  Massively Parallel Sort-Merge Joins in Main Memory Multi-Core Database Systems.  _Proceedings of the VLDB Endowment_ 5, 10 (2012), 1064–1075.  https://doi.org/10.14778/2336664.2336678
  * Alsubaiee et al. (2014) Sattam Alsubaiee, Yasser Altowim, Hotham Altwaijry, Alexander Behm, Vinayak R. Borkar, Yingyi Bu, Michael J. Carey, Inci Cetindil, Madhusudan Cheelangi, Khurram Faraaz, Eugenia Gabrielova, Raman Grover, Zachary Heilbron, Young-Seok Kim, Chen Li, Guangqiang Li, Ji Mahn Ok, Nicola Onose, Pouria Pirzadeh, Vassilis J. Tsotras, Rares Vernica, Jian Wen, and Till Westmann. 2014.  AsterixDB: A Scalable, Open Source BDMS.  _Proceedings of the VLDB Endowment_ 7, 14 (2014), 1905–1916.  https://doi.org/10.14778/2733085.2733096
  * Balkesen et al. (2013a) Cagri Balkesen, Gustavo Alonso, Jens Teubner, and M Tamer Özsu. 2013a.  Multi-Core, Main-Memory Joins: Sort vs. Hash Revisited.  _Proceedings of the VLDB Endowment_ 7, 1 (2013), 85–96.  https://doi.org/10.14778/2732219.2732227
  * Balkesen et al. (2013b) Cagri Balkesen, Jens Teubner, Gustavo Alonso, and M. Tamer Ozsu. 2013b.  Main-memory hash joins on multi-core CPUs: Tuning to the underlying hardware. In _Proceedings of the IEEE International Conference on Data Engineering (ICDE)_. 362–373.  https://doi.org/10.1109/ICDE.2013.6544839
  * Bandle et al. (2021) Maximilian Bandle, Jana Giceva, and Thomas Neumann. 2021\.  To Partition, or Not to Partition, That is the Join Question in a Real System. In _Proceedings of the ACM SIGMOD International Conference on Management of Data_. 168–180.  https://doi.org/10.1145/3448016.3452831
  * Barber et al. (2014) Ronald Barber, Guy M Lohman, Ippokratis Pandis, Vijayshankar Raman, Richard Sidle, Gopi K Attaluri, Naresh Chainani, Sam Lightstone, and David Sharpe. 2014.  Memory-Efficient Hash Joins.  _Proceedings of the VLDB Endowment_ 8, 4 (2014), 353–364.  https://doi.org/10.14778/2735496.2735499
  * Beame et al. (2014) Paul Beame, Paraschos Koutris, and Dan Suciu. 2014\.  Skew in parallel query processing. In _Proceedings of the 33rd ACM SIGMOD-SIGACT-SIGART Symposium on Principles of Database Systems, PODS’14, Snowbird, UT, USA, June 22-27, 2014_. 212–223.  https://doi.org/10.1145/2594538.2594558
  * Berenberg and Calder (2021) Anna Berenberg and Brad Calder. 2021.  Deployment Archetypes for Cloud Applications.  _CoRR_ abs/2105.0 (2021).  https://arxiv.org/abs/2105.00560
  * Blanas et al. (2011) Spyros Blanas, Yinan Li, and Jignesh M Patel. 2011.  Design and evaluation of main memory hash join algorithms for multi-core CPUs. In _Proceedings of the ACM SIGMOD International Conference on Management of Data_. 37–48.  https://doi.org/10.1145/1989323.1989328
  * Boncz et al. (2017) Peter A Boncz, Angelos-Christos G Anadiotis, and Steffen Kläbe. 2017.  JCC-H: Adding Join Crossing Correlations with Skew to TPC-H. In _Performance Evaluation and Benchmarking for the Analytics Era - 9th TPC Technology Conference, TPCTC 2017, Munich, Germany, August 28, 2017, Revised Selected Papers_ _(Lecture Notes in Computer Science, Vol. 10661)_. 103–119.  https://doi.org/10.1007/978-3-319-72401-0_8
  * Boncz et al. (1999) Peter A. Boncz, Stefan Manegold, and Martin L. Kersten. 1999\.  Database architecture optimized for the new bottleneck: Memory access. In _Proceedings of the International Conference on Very Large Data Bases (VLDB)_. 54–65.  http://www.vldb.org/conf/1999/P5.pdf
  * Bruck et al. (2006) Jehoshua Bruck, Jie Gao, and Anxiao Jiang. 2006.  Weighted Bloom filter. In _Proceedings of the IEEE International Symposium on Information Theory (ISIT)_. 2304–2308.  https://doi.org/10.1109/ISIT.2006.261978
  * Buyya et al. (2019) Rajkumar Buyya, Satish Narayana Srirama, Giuliano Casale, Rodrigo N Calheiros, Yogesh Simmhan, Blesson Varghese, Erol Gelenbe, Bahman Javadi, Luis Miguel Vaquero, Marco A S Netto, Adel Nadjaran Toosi, Maria Alejandra Rodriguez, Ignacio Mart$\backslash$’$\backslash$in Llorente, Sabrina De Capitani di Vimercati, Pierangela Samarati, Dejan S Milojicic, Carlos A Varela, Rami Bahsoon, Marcos Dias de Assunção, Omer F Rana, Wanlei Zhou, Hai Jin, Wolfgang Gentzsch, Albert Y Zomaya, and Haiying Shen. 2019\.  A Manifesto for Future Generation Cloud Computing: Research Directions for the Next Decade.  _Comput. Surveys_ 51, 5 (2019), 105:1—-105:38.  https://doi.org/10.1145/3241737
  * Chatterjee et al. (2022) Subarna Chatterjee, Meena Jagadeesan, Wilson Qin, and Stratos Idreos. 2022.  Cosine: A Cloud-Cost Optimized Self-Designing Key-Value Storage Engine. In _In Proceedings of the Very Large Databases Endowment_.  https://doi.org/10.14778/3485450.3485461
  * Cisco (2018) Cisco. 2018.  Cisco Global Cloud Index: Forecast and Methodology, 2016–2021.  _White Paper_ (2018).  https://tinyurl.com/CiscoGlobalCloud2018
  * Cutt and Lawrence (2009) Bryce Cutt and Ramon Lawrence. 2009.  Using intrinsic data skew to improve hash join performance.  _Inf. Syst._ 34, 6 (2009), 493–510.  https://doi.org/10.1016/j.is.2009.02.003
  * Cutt et al. (2008) Bryce Cutt, Ramon Lawrence, and Tolley Joshua. 2008\.  Discussion and Implementation of Histojoin in PostgreSQL<EMAIL_ADDRESS>(2008).
  * DeWitt et al. (1984) David J. DeWitt, Randy H. Katz, Frank Olken, Leonard D. Shapiro, Michael Stonebraker, and David A. Wood. 1984.  Implementation Techniques for Main Memory Database Systems. In _Proceedings of the ACM SIGMOD International Conference on Management of Data_. 1–8.  http://dl.acm.org/citation.cfm?id=602259.602261
  * DeWitt et al. (1992) David J DeWitt, Jeffrey F Naughton, Donovan A Schneider, S Seshadri, Wei Li, Dengfeng Gao, Richard T Snodgrass, Kien A Hua, and Chiang Lee. 1992.  Practical Skew Handling in Parallel Joins. In _Proceedings of the International Conference on Very Large Data Bases (VLDB)_. 525–535.  https://doi.org/10.1145/564691.564711
  * Gartner (2017) Gartner. 2017.  Gartner Says 8.4 Billion Connected “Things” Will Be in Use in 2017, Up 31 Percent From 2016.  https://tinyurl.com/Gartner2020.
  * Guo and Chen (2019) Chengxin Guo and Hong Chen. 2019.  In-Memory Join Algorithms on GPUs for Large-Data. In _21st IEEE International Conference on High Performance Computing and Communications; 17th IEEE International Conference on Smart City; 5th IEEE International Conference on Data Science and Systems, HPCC/SmartCity/DSS 2019, Zhangjiajie, China, August 10-12_. 1060–1067.  https://doi.org/10.1109/HPCC/SmartCity/DSS.2019.00151
  * Haas et al. (1997) Laura M Haas, Michael J Carey, Miron Livny, and Amit Shukla. 1997\.  Seeking the Truth About ad hoc Join Costs.  _The VLDB Journal_ 6, 3 (1997), 241–256.  https://doi.org/10.1007/s007780050043
  * He et al. (2020) Shuo He, Xinchen Lyu, Wei Ni, Hui Tian, Ren Ping Liu, and Ekram Hossain. 2020\.  Virtual Service Placement for Edge Computing Under Finite Memory and Bandwidth.  _IEEE Trans. Commun._ 68, 12 (2020), 7702–7718.  https://doi.org/10.1109/TCOMM.2020.3022692
  * Hua and Lee (1991) Kien A Hua and Chiang Lee. 1991.  Handling Data Skew in Multiprocessor Database Computers Using Partition Tuning. In _Proceedings of the International Conference on Very Large Data Bases (VLDB)_. 525–535.  http://www.vldb.org/conf/1991/P525.PDF
  * Jahangiri et al. (2022) Shiva Jahangiri, Michael J Carey, and Johann-Christoph Freytag. 2022\.  Design Trade-offs for a Robust Dynamic Hybrid Hash Join.  _Proceedings of the VLDB Endowment_ 15, 10 (2022), 2257–2269.  https://www.vldb.org/pvldb/vol15/p2257-jahangiri.pdf
  * Jennings and Stadler (2015) Brendan Jennings and Rolf Stadler. 2015.  Resource Management in Clouds: Survey and Research Challenges.  _J. Netw. Syst. Manag._ 23, 3 (2015), 567–619.  https://doi.org/10.1007/s10922-014-9307-7
  * Kim et al. (2020) Taewoo Kim, Alexander Behm, Michael Blow, Vinayak Borkar, Yingyi Bu, Michael J. Carey, Murtadha Hubail, Shiva Jahangiri, Jianfeng Jia, Chen Li, Chen Luo, Ian Maxon, and Pouria Pirzadeh. 2020.  Robust and efficient memory management in Apache AsterixDB.  _Software - Practice and Experience_ 50, 7 (2020), 1114–1151.  https://doi.org/10.1002/spe.2799
  * Kitsuregawa et al. (1989) Masaru Kitsuregawa, Masaya Nakayama, and Mikio Takagi. 1989\.  The Effect of Bucket Size Tuning in the Dynamic Hybrid GRACE Hash Join Method. In _Proceedings of the International Conference on Very Large Data Bases (VLDB)_. 257–266.  http://www.vldb.org/conf/1989/P257.PDF
  * Kitsuregawa et al. (1983) Masaru Kitsuregawa, Hidehiko Tanaka, and Tohru Moto-Oka. 1983\.  Application of Hash to Data Base Machine and Its Architecture.  _New Generation Computing_ 1, 1 (1983), 63–74.
  * Leis et al. (2015) Viktor Leis, Andrey Gubichev, Atanas Mirchev, Peter A Boncz, Alfons Kemper, and Thomas Neumann. 2015\.  How Good Are Query Optimizers, Really?  _Proceedings of the VLDB Endowment_ 9, 3 (2015), 204–215.  https://doi.org/10.14778/2850583.2850594
  * Li et al. (2018) Rundong Li, Mirek Riedewald, and Xinyan Deng. 2018\.  Submodularity of Distributed Join Computation. In _Proceedings of the ACM SIGMOD International Conference on Management of Data_. 1237–1252.  https://doi.org/10.1145/3183713.3183728
  * Li et al. (2002) Wei Li, Dengfeng Gao, and Richard T Snodgrass. 2002\.  Skew handling techniques in sort-merge join. In _Proceedings of the ACM SIGMOD International Conference on Management of Data_. 169–180.  https://doi.org/10.1145/564691.564711
  * Liu and Blanas (2015) Feilong Liu and Spyros Blanas. 2015.  Forecasting the cost of processing multi-join queries via hashing for main-memory databases (Extended version).  _CoRR_ abs/1507.0 (2015).  http://arxiv.org/abs/1507.03049
  * Lutz et al. (2020) Clemens Lutz, Sebastian Breßand Steffen Zeuch, Tilmann Rabl, and Volker Markl. 2020.  Pump Up the Volume: Processing Large Data on GPUs with Fast Interconnects. In _Proceedings of the ACM SIGMOD International Conference on Management of Data_. 1633–1649.  https://doi.org/10.1145/3318464.3389705
  * McCallum (2022) John C. McCallum. 2022\.  Historical Cost of Computer Memory and Storage.  _https://jcmit.net/mem2015.htm_ (2022).
  * MySQL (2021) MySQL. 2021.  MySQL System Variables.  _https://dev.mysql.com/doc/refman/8.0/en/server-system-variables.html_ (2021).
  * MySQL (2023) MySQL. 2023.  MySQL.  _https://www.mysql.com/_ (2023).
  * Nakayama et al. (1988) Masaya Nakayama, Masaru Kitsuregawa, and Mikio Takagi. 1988\.  Hash-Partitioned Join Method Using Dynamic Destaging Strategy. In _Proceedings of the International Conference on Very Large Data Bases (VLDB)_. 468–478.  http://www.vldb.org/conf/1988/P468.PDF
  * Orr et al. (2019) Laurel J Orr, Srikanth Kandula, and Surajit Chaudhuri. 2019\.  Pushing Data-Induced Predicates Through Joins in Big-Data Clusters.  _Proceedings of the VLDB Endowment_ 13, 3 (2019), 252–265.  https://doi.org/10.14778/3368289.3368292
  * Papon and Athanassoulis (2021a) Tarikul Islam Papon and Manos Athanassoulis. 2021a.  A Parametric I/O Model for Modern Storage Devices. In _Proceedings of the International Workshop on Data Management on New Hardware (DAMON)_.
  * Papon and Athanassoulis (2021b) Tarikul Islam Papon and Manos Athanassoulis. 2021b.  The Need for a New I/O Model. In _Proceedings of the Biennial Conference on Innovative Data Systems Research (CIDR)_.
  * Pohl et al. (2020) Constantin Pohl, Kai-Uwe Sattler, and Goetz Graefe. 2020\.  Joins on high-bandwidth memory: a new level in the memory hierarchy.  _The VLDB Journal_ 29, 2-3 (2020), 797–817.  https://doi.org/10.1007/s00778-019-00546-z
  * PostgreSQL ([n. d.]) PostgreSQL. [n. d.].  Skew Optimization in PostgreSQL.  _https://github.com/postgres/postgres/blob/master/src/include/executor/hashjoin.h#L84_ ([n. d.]).
  * PostgreSQL (2022) PostgreSQL. 2022\.  PostgreSQL Resource Consumption.  _https://www.postgresql.org/docs/13/runtime-config-resource.html_ (2022).
  * PostgreSQL (2023) PostgreSQL. 2023\.  PostgreSQL: The World’s Most Advanced Open Source Relational Database.  _https://www.postgresql.org_ (2023).
  * Ramakrishnan and Gehrke (2002) Raghu Ramakrishnan and Johannes Gehrke. 2002.  _Database Management Systems_.  McGraw-Hill Higher Education, 3rd edition.
  * Rödiger et al. (2016) Wolf Rödiger, Sam Idicula, Alfons Kemper, and Thomas Neumann. 2016.  Flow-Join: Adaptive skew handling for distributed joins over high-speed networks. In _Proceedings of the IEEE International Conference on Data Engineering (ICDE)_. {IEEE} Computer Society, 1194–1205.  https://doi.org/10.1109/ICDE.2016.7498324
  * Rui et al. (2015) Ran Rui, Hao Li, and Yi-Cheng Tu. 2015.  Join algorithms on GPUs: A revisit after seven years. In _Proceedings of the IEEE International Conference on Big Data (BigData)_. 2541–2550.  https://doi.org/10.1109/BigData.2015.7364051
  * Schuh et al. (2016) Stefan Schuh, Xiao Chen, and Jens Dittrich. 2016.  An Experimental Comparison of Thirteen Relational Equi-Joins in Main Memory. In _Proceedings of the ACM SIGMOD International Conference on Management of Data_. 1961–1976.  https://doi.org/10.1145/2882903.2882917
  * Shapiro (1986) Leonard D Shapiro. 1986\.  Join Processing in Database Systems with Large Main Memories.  _ACM Trans. Database Syst._ 11, 3 (1986), 239–264.  https://doi.org/10.1145/6314.6315
  * Shatdal et al. (1994) Ambuj Shatdal, Chander Kant, and Jeffrey F. Naughton. 1994\.  Cache Conscious Algorithms for Relational Query Processing. In _Proceedings of the International Conference on Very Large Data Bases (VLDB)_. 510–521.  http://dl.acm.org/citation.cfm?id=645920.758363
  * TPC (2021) TPC. 2021.  TPC-H benchmark.  _http://www.tpc.org/tpch/_ (2021).
  * Walton et al. (1991) Christopher B Walton, Alfred G Dale, and Roy M Jenevein. 1991\.  A Taxonomy and Performance Model of Data Skew Effects in Parallel Joins. In _Proceedings of the International Conference on Very Large Data Bases (VLDB)_. 537–548.  http://www.vldb.org/conf/1991/P537.PDF
  * Wassenberg and Sanders (2011) Jan Wassenberg and Peter Sanders. 2011.  Engineering a Multi-core Radix Sort. In _Proceedings of the International Conference on Parallel Processing (EuroPar)_. 160–169.  https://doi.org/10.1007/978-3-642-23397-5 _16
  * Xu et al. (2008) Yu Xu, Pekka Kostamaa, Xin Zhou, and Liang Chen. 2008\.  Handling data skew in parallel joins in shared-nothing systems. In _Proceedings of the ACM SIGMOD International Conference on Management of Data_. 1043–1052.  https://doi.org/10.1145/1376616.1376720
  * Zaharia et al. (2012) Matei Zaharia, Mosharaf Chowdhury, Tathagata Das, Ankur Dave, Justin Ma, Murphy McCauly, Michael J Franklin, Scott Shenker, and Ion Stoica. 2012.  Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing. In _Proceedings of the USENIX Symposium on Networked Systems Design and Implementation (NSDI)_. 15–28.  https://www.usenix.org/conference/nsdi12/technical-sessions/presentation/zaharia

##  Appendix A Complexity Analysis for Monotonicity-based Pruining

Based on two pruning inequalities (i.e., Pruning 1 and Pruning 2) from weakly-
ordered property, the range for $k$ is at most
$\frac{i}{j}-\left\lfloor\frac{n-i-1}{m-j}\right\rfloor-1+(2-\frac{1}{j})\cdot
c_{R}$. If
$\frac{i}{j}-\left\lfloor\frac{n-i-1}{m-j}\right\rfloor-1+(2-\frac{1}{j})\cdot
c_{R}>\frac{i}{j}$, we can even skip the specific $j$ (line 7 in Algorithm 5)
because we do not have to iterate any $k$ in this case. Therefore, the
complexity for the third loop (line $7\sim 11$) is

(4)
$\begin{split}&\max\left\\{\frac{i}{j}-\left\lfloor\frac{n-i-1}{m-j}\right\rfloor-1+\left(2-\frac{1}{j}\right)\cdot
c_{R},0\right\\}\leq\max\left\\{\frac{i}{j}-\left\lfloor\frac{n-i-1}{m-j}\right\rfloor-1,0\right\\}+\left(2-\frac{1}{j}\right)\cdot
c_{R}\end{split}$

Summing up the complexity for all possible $i$ and $j$, we have:

(5)
$\begin{split}\sum\limits_{i=1}^{n}\sum\limits_{j=2}^{m-1}\max\left\\{\frac{i}{j}-\left\lfloor\frac{n-i-1}{m-j}\right\rfloor-1,0\right\\}\leq&\sum\limits_{i=1}^{n}\sum\limits_{j=2}^{m-1}\max\left\\{\frac{i}{j}-\frac{n-i}{m-j},0\right\\}=\sum\limits_{j=2}^{m-1}\sum\limits_{i=\frac{j\cdot
n}{m}}^{n}\frac{i\cdot m-n\cdot j}{j\cdot(m-j)}\\\
=&\sum\limits_{j=2}^{m-1}\frac{\frac{n\cdot j+n\cdot
m}{2}\cdot\left(n-\frac{n\cdot j}{m}+1\right)-n\cdot j\left(n-\frac{n\cdot
j}{m}+1\right)}{j\cdot(m-j)}\\\
=&\sum\limits_{j=2}^{m-1}\frac{\frac{n\cdot(m-j)}{2}\cdot\left(n-\frac{n\cdot
j}{m}+1\right)}{j\cdot(m-j)}=\sum\limits_{j=2}^{m-1}\frac{n\cdot\left(n-\frac{n\cdot
j}{m}+1\right)}{2j}\\\
=&\frac{1}{2}\left(\sum\limits_{j=2}^{m-1}n^{2}\cdot\left(\frac{1}{j}-\frac{1}{m}\right)+\sum\limits_{j=2}^{m-1}\frac{n}{j}\right)=O(n^{2}\log
m)\\\
\sum\limits_{i=1}^{n}\sum\limits_{j=2}^{m-1}\left(2-\frac{1}{j}\right)\cdot
c_{R}\leq\sum\limits_{i=1}^{n}2\cdot(m-2)\cdot c_{R}&=O(nm^{2})\end{split}$

We typically consider $c_{R}\propto m(m<<n)$ in our paper (the number of
buffer pages is much smaller than the number of input records to be
partitioned, otherwise we can simply put all the records from $R$ into buffer
to execute NBJ without partitioning), and thus the computation complexity is
dominated by $O(n^{2}\log m)$.

##  Appendix B Proof of Weakly-Ordered and Divisible Property

Weakly-Ordered Property. The weakly-ordered property can be formalized using
Corollary B.1.

###### Corollary B.1.0.

There exists an optimal partitioning scheme $P_{opt}$ where all the partitions
contain consecutive elements and the sequence
$\left\lceil\frac{\|\mathcal{N}(i)\|}{c_{R}}\right\rceil$ is in decreasing
order.

In fact, we can have a stronger Corollary B.2 as follows.

###### Corollary B.2.0.

There exists an optimal partitioning scheme $P$ where all the partitions
contain consecutive elements w.r.t. their $CT$ value, and the partition size
sequence $\\{\|\mathcal{N}(i)\|\\}$ is in decreasing (monotonic non-
increasing) order.

###### Proof.

We use the optimal partitioning scheme derived from the consecutive property.
Suppose that we find two adjacent partitions that are not in the non-
decreasing order, as shown in Figure 14. We can apply Lemma 3.2 multiple times
until we have the larger partition is completely placed before the smaller
one. We can repeat this process for every two adjacent partitions until all
the partitions are decreasingly sorted by their size. Note that, we always
ensure that the join cost does not increase by Lemma 3.2 and thus the final
partitioning scheme is still optimal. The proof of Corollary B.1 follows a
similar procedure as above. Proof completes.

Figure 14. When there are two adjacent partitions which do not follow the
decreasing order, we can always apply Lemma 3.2 to swap elements. For example,
in the first step, we have $i_{3}>i_{2}-1$ and
$\|\mathcal{N}(i_{3})\|>\|\mathcal{N}(i_{2}-1)\|$, and thus
$\left\lceil\frac{\|\mathcal{N}(i_{3})\|}{c_{R}}\right\rceil\geq\left\lceil\frac{\|\mathcal{N}(i_{2}-1)\|}{c_{R}}\right\rceil$,
we can swap $i_{3}$ and $i_{2}-1$. We continue swapping elements until the
indexes of all the elements in the larger partition are smaller than the other
partition.

∎

Divisible Property. We prove the divisible property using Corollary B.3.

###### Corollary B.3.0.

There exists an optimal partitioning $P_{opt}$ where the size of all
partitions except the largest partition, is divisible by $c_{R}$.

Note that Corollaries B.3 and B.2 may give us two different partitioning
strategies that have the same – optimal – cost. In other words, an optimal
solution might satisfy either one of the two or both. For example, consider
$n=c_{R}\cdot(B-2)+r\;(0<r<c_{R})$. According to Corollary B.2, we will group
the last $r$ records to form the smallest partition and the remaining records
form $B-2$ partitions. However, The consecutive property shows that we need to
group the first $r$ records together. Even though we may end up having a
different partitioning scheme, the optimal cost remains the same, due to the
consecutive property, Corollary B.3, and B.2. This explains why we use the
weakly-ordered property (Corollary B.1) instead of the ordered one (Corollary
B.2) in our main paper.

Figure 15. When $\|\mathcal{P}_{j}\|$ is not divisible by $c_{R}$, we can move
elements from $\mathcal{P}_{j-1}$ to $\mathcal{P}_{j}$ until
$\left\lceil\frac{\|\mathcal{P}_{j-1}\|}{c_{R}}\right\rceil<\left\lceil\frac{\|\mathcal{P}_{j}\|}{c_{R}}\right\rceil$
or $\|\mathcal{P}_{j}\|$ is divisible by $c_{R}$. In the above example, we
note the partition containing $[i_{2}+1,i_{3}]$ as $\mathcal{P}_{j}$ and the
previous partition containing $[i_{1},i_{2}]$ as $\mathcal{P}_{j-1}$. Assuming
$c_{R}=5$, the above process stops when $\|\mathcal{P}_{j}\|=k\cdot 5)$ where
$k\in\mathbf{N}$.

###### Proof.

We aim to find a process that can obtain another optimal solution which
satisfies both Corollaries B.3 and B.1. We note
$\mathcal{P}_{1},\mathcal{P}_{2},...,\mathcal{P}_{m}$ as the partition
sequence obtained from Corollary B.1. As such, these partitions are sorted by
$\left\lceil\frac{\|\mathcal{P}_{j}\|}{c_{R}}\right\rceil$ in a descending
order. Now we select the largest $j(j\in[m])$ where $\|\mathcal{P}_{j}\|$ is
not divisible by $c_{R}$. If $j=1$, we do nothing as the partitioning scheme
already satisfies Corollary B.3. Otherwise, we move adjacent elements from
$\mathcal{P}_{j-1}$ to $\mathcal{P}_{j}$ until
$\left\lceil\frac{\|\mathcal{P}_{j-1}\|}{c_{R}}\right\rceil<\left\lceil\frac{\|\mathcal{P}_{j}\|}{c_{R}}\right\rceil$
or $\|\mathcal{P}_{j}\|$ is a multiple of $c_{R}$. When
$\left\lceil\frac{\|\mathcal{P}_{j-1}\|}{c_{R}}\right\rceil<\left\lceil\frac{\|\mathcal{P}_{j}\|}{c_{R}}\right\rceil$,
we swap elements according to the process mentioned in the proof of Corollary
B.2 so that the decreasing order of
$\left\lceil\frac{\|\mathcal{P}_{j}\|}{c_{R}}\right\rceil$ is maintained. In
other words, we obtain two consecutive partitions
$\mathcal{P}^{new}_{j-1},\mathcal{P}^{new}_{j}$ and we have
$\|\mathcal{P}^{new}_{j-1}\|=\|\mathcal{P}^{old}_{j}\|,\|\mathcal{P}^{new}_{j}\|=\|\mathcal{P}^{old}_{j-1}\|$.
We can repeat this process until the largest $j$ of $\|\mathcal{P}_{j}\|$ can
be divisible by $c_{R}$. Note that, when
$\left\lceil\frac{\|\mathcal{P}_{j-1}\|}{c_{R}}\right\rceil<\left\lceil\frac{\|\mathcal{P}_{j}\|}{c_{R}}\right\rceil$,
after we swap elements between $\mathcal{P}_{j-1}$ and $\mathcal{P}_{j}$,
$\mathcal{P}_{j-1}$ will be selected to add elements from $\mathcal{P}_{j-2}$.
The above process already ensures that the final partition contains
consecutive elements and the descending order elements based on their CT
value. In this process, when we move elements to $\mathcal{P}_{j}$,
$\left\lceil\frac{\|\mathcal{P}_{j}\|}{c_{R}}\right\rceil$ does not change
since $\|\mathcal{P}_{j}\|$ is not divisible by $c_{R}$. Therefore, after we
move one element $i$ to $\mathcal{P}_{j}$, all the elements from
$\mathcal{P}_{j}$ contribute the same cost and all the elements (except $i$)
from $\mathcal{P}_{j-1}$ have equivalent or even lower cost since
$\|\mathcal{P}_{j-1}\|$ decreases. For element $i$, the cost cannot increase
since we maintain the descending order of
$\left\lceil\frac{\|\mathcal{P}_{j}\|}{c_{R}}\right\rceil$. Specifically, when
we move element $i$ from $\mathcal{P}_{j-1}$ to $\mathcal{P}_{j}$,
$\left\lceil\frac{\|\mathcal{N}(i)\|}{c_{R}}\right\rceil$ must decrease or
remain the same. Proof completes. ∎