LevelHeaded: Making Worst-Case Optimal Joins Work in the Common Case
Christopher R. Aberger Andrew Lamb Kunle Olukotun Christopher Ré

empty



2126Streaming Graph Challenge: Stochastic Block Partition
    Edward Kao,
Vijay Gadepally,
Michael Hurley,
Michael Jones,
Jeremy Kepner,
Sanjeev Mohindra, 

Paul Monticciolo,
Albert Reuther,
Siddharth Samsi,
William Song,
Diane Staheli,
Steven Smith 

MIT Lincoln Laboratory, Lexington, MA
    Received: date / Accepted: date
=======================================================================================================================================================================================================================================


 
Pipelines combining SQL-style business intelligence (BI) queries
and linear algebra (LA) are becoming increasingly common in industry. As a result,
there
is a growing need to unify these workloads in a single
framework. Unfortunately, existing solutions either sacrifice the inherent
benefits of exclusively using a relational database (e.g. logical
and physical independence) or
incur orders of magnitude performance gaps compared to specialized engines (or both).
In this work we study applying a new type of query processing architecture to
standard BI and LA benchmarks.  To do this we present a new in-memory query 
processing engine called . uses worst-case optimal joins as its core
execution mechanism for both BI and LA queries.
With , we show how crucial optimizations for
BI and LA queries can be captured in
a worst-case optimal query architecture. Using these optimizations, outperforms other relational database engines (LogicBlox,
MonetDB, and HyPer) by orders of magnitude on standard LA 
benchmarks, while performing on average within 31%
of the best-of-breed BI (HyPer) and LA (Intel MKL) solutions on their own
benchmarks. Our results show that such a single query processing architecture is
capable of delivering competitive performance on both BI and
LA queries.


Finally, on an application which combines SQL-style querying and
linear algebra, we show that can outperform popular
solutions like MonetDB/SciKit-Learn, Spark, and Pandas/SciKit-Learn by up to two
orders of magnitude.
In this work we identify these techniques, validate their merit, and describe a novel but practical join processing architecture which uses them.







§ INTRODUCTION

The efficient processing of classic SQL-style workloads is no longer enough;
machine learning algorithms are being adopted at an explosive rate. In fact,
Intel projects that by 2020 the hardware cycles dedicated to machine
learning tasks will grow by 12x, resulting in more servers running
this than any other workload <cit.>. As a result, there is a
growing need
for query processing engines that
are efficient on (1) the SQL-style queries at the core of most
business intelligence workloads and (2) the linear algebra operations
at the core
of most machine learning algorithms. In this work, we explore whether a new
query processing architecture is capable of delivering competitive
performance in both cases.

An increasingly popular workflow combines business intelligence (BI) and linear algebra (LA) queries  
by executing SQL queries in a relational warehouse as a means to extract 
feature sets for machine learning models. 
Unsurprisingly, these SQL queries are similar to standard BI workloads: 
the data is de-normalized (via joins), filtered, and aggregated to form a single feature set 
<cit.>. Still, BI queries 
are best processed in a relational database
management system (RDBMS) and LA queries are best 
processed in a
LA package. As a
result, there has been a flurry of activity around building systems
capable of unifying both BI and LA
querying <cit.>. 
At a high-level, existing approaches fall into one of three classes:





  * Exclusively using a relational engine. There are many
inherent advantages to exclusively using a RDBMS to process both BI and LA queries.
Simplifying extract-transform-load 
(ETL), increasing usability, and leveraging well-known
optimizations 
 are just a few <cit.>. Although it is known that 
LA queries can be expressed using joins and aggregations, executing these queries 
via the pairwise join algorithms in standard RDBMSs is orders of magnitude slower 
than using a LA package (see <Ref>). 
Thus, others <cit.> have shown that a RDBMS must be modified
to compete on LA queries. 

  * Extending a linear algebra package. Linear algebra packages, like 
BLAS <cit.>
or LAPACK <cit.>, provide high-performance through low-level 
procedural interfaces and therefore lack the ability for high-level querying. To 
address this, array databases with high-level querying, like SciDB 
<cit.>, have been proposed. Unfortunately, array databases are 
highly specialized and are not designed for general BI querying. 
As a result, support for SQL-style BI querying 
<cit.> has recently been
combined with the LA support found in popular packages 
like Spark's MLlib <cit.> and Scikit-learn 
<cit.>. 
Still, these solutions lack many of the inherent benefits of a RDMBS, 
like a sophisticated (shared-memory) query optimizer or efficient
data structures (e.g. indexes) for query execution, and therefore achieve suboptimal 
performance on BI workloads (see <Ref>).

  * Combining a relational engine with a linear algebra package.
To preserve the benefits of using a RDBMS on BI queries, while also
avoiding their pairwise join algorithms on LA queries, 
others (e.g. Oracle's UT_NLA <cit.> and MonetDB/NumPy <cit.>)
have integrated a RDBMS with a LA package. 
Still, these approaches
just tack on an external LA package to a RDBMS—they do not fully
integrate it. Therefore, users are forced to write low-level code wrapping
the LA package and the LA computation is a black box to
the query optimizer. Even worse, this integration, while straightforward on
dense data, is complicated (and largely unsolved) on sparse data because a RDBMS 
and a LA package use fundamentally different 
data layouts here (e.g. column store versus
compressed sparse row).



Therefore, existing approaches either (1) sacrifice the inherent benefits of using
only an
RDBMS to process both classes of queries, (2) are unable to process both classes
of
queries, or (3) incur orders of magnitude performance gaps relative to the best 
single-class approach.



In this work we study an alternative approach to building a RDBMS
for both BI and LA querying. In particular, we
study using worst-case optimal join (WCOJ) algorithms <cit.> as the
mechanism to unify these query workloads. To do this, we present a new in-memory query 
processing engine called . uses a novel WCOJ query architecture to optimize,
plan,  and 
execute both BI and LA queries. In contrast 
to previous WCOJ engines <cit.>, 
is designed for and evaluated on more than just graph
queries. As such, is the first WCOJ engine to present an evaluation on 
both BI and LA queries. In contrast to other query engines 
<cit.>, 
competes with both LA packages and RDBMSs
on their own benchmarks (see <Ref>).

However, designing a new query processing engine that
is efficient on both BI and LA queries is a
challenging task. The recently proposed WCOJs at the
core of this
new query processing architecture are most effective on graph queries 
where they have an asymptotic advantage over traditional
pairwise join algorithms.
In contrast, pairwise
join algorithms are well-understood, and have the
advantage of
45+ years of proven constant factor optimizations for BI
workloads
<cit.>.
Further, LA queries, which also have the benefit of
decades of optimizations 
<cit.>, are
typically
not a good match for the relational model. Therefore, it is not at all
obvious whether they are good match for this new type of relational query
processing architecture.

Despite these challenges, we found that the unification of three new
techniques in a WCOJ architecture could enable it 
 to deliver competitive performance on both 
BI and LA queries. The techniques we leverage are (1) a new mechanism to translate general
SQL queries to WCOJ query plans, (2) a new cost-based query optimizer for 
WCOJ's, and (3) a simple storage engine optimizer for . It was
not at all obvious how to identify and unify these techniques in a WCOJ framework and,
as such, we are the first to do so. The three core techniques of in more detail are:


 

  * SQL to GHDs: Pushing Down Selections and Attribute
Elimination. Neither the theoretical literature <cit.> nor the query compilation
techniques for WCOJs <cit.> maps directly to all
SQL features. In we implement a practical extension of these
techniques that enables us to capture more general query workloads
as well as the classic query optimizations of pushing down selections and attribute 
elimination. Besides providing up to 4x speedup on BI 
queries, a core artifact of our attribute elimination implementation is that it 
enables to target BLAS packages on dense LA queries
at little to no execution cost. This is because it is challenging to outperform BLAS 
packages, like Intel MKL <cit.> on sparse data and is usually not 
possible[Intel MKL often gets peak hardware efficiency on dense 
LA <cit.>.] on dense data. Therefore, 
leverages attribute elimination to opaquely call Intel MKL on dense LA 
queries while executing sparse LA queries as pure aggregate-join queries (entirely in 
).

  * Cost-Based Optimizer: Attribute Ordering.
WCOJ query optimizers need to select an attribute order <cit.> in 
a similar manner to how traditional query optimizers select a join order 
<cit.>. With , we present a 
cost-based optimizer to select a WCOJ attribute order for the 
first time. We highlight that our optimizer follows heuristics
that are different from what conventional wisdom from pairwise join optimizers 
suggests (i.e. highest cardinality first) and describe how to leverage these 
heuristics to provide an accurate cost-estimate for a WCOJ algorithm.  
Using this cost-estimate, we validate that 's 
cost-based optimizer selects attribute orders than can be up to
8815x faster than attribute orders that previous WCOJ engines 
<cit.> might select.

  * Group By Tradeoffs: Battling Skew on Group By. Finally, 
we study the classic tradeoffs around executing s  
in the presence of skew. Although these tradeoffs
are well understood for BI queries, they have yet to be applied to this new
style of query processing engine and are also essential for LA queries 
(see <Ref>). We describe 's  
optimizer that exploits these tradeoffs and show that it provides
up to a 875x and 185x speedup over a naive implementation on BI 
and LA queries respectively.




We evaluate on standard BI and
LA benchmarks: seven TPC-H queries[The TPC-H queries
are run without the  clause.] and four (two sparse, two dense) 
LA kernels. These benchmarks are de-facto standards
for relational query processing and LA engines. Thus, each engine
we compare to is designed to process one of these benchmarks efficiently
using specific optimizations that enable high-performance on one type 
of benchmark, but not necessarily the other.
Therefore, although these engines are the state-of-the-art solutions within a
benchmark, they are unable to remain competitive across benchmarks.
For example, HyPeR delivers high performance on BI queries, but is not 
competitive on most LA workloads; similarly, Intel's Math Kernel Library (MKL) 
<cit.> delivers high performance on LA queries, but does not provide 
support for BI querying. In contrast, is
designed to be generic, maintaining efficiency across the queries in both 
benchmarks.



  
*Contribution Summary  This paper introduces the engine and
demonstrates that its novel architecture can compete on standard BI and LA 
benchmarks. We show
that outperforms other relational engines by at least an order of
magnitude
on LA queries, while remaining on average within 31% of 
best-of-the-breed solutions on both BI and LA queries.

Our contributions and outline are as follows. 
 
  * In <Ref>, we describe the architecture. 
    In particular, we describe how 
's query and data model, which is different from that of previous WCOJ engines <cit.>, preserves the theoretical benefits of a WCOJ architecture 
and enables it to 
efficiently process BI and LA queries.
	
  * In
<Ref> we present the
core (logical and physical) optimizations that we unify in a
WCOJ query architecture for the first time. We show that these optimizations 
provide up to a three orders of magnitude performance 
speedup and are necessary to compete on BI and LA queries. In more detail, we present the
classic query optimizations of pushing down selections
and attribute elimination in <Ref>, a cost-based
optimizer for selecting an attribute order for a WCOJ algorithm in <Ref>,
and, finally, a simple query execution optimizer for s in <Ref>.    

  * In <Ref> we show that can outperform other
relational
engines by an order of magnitude on standard BI and LA benchmarks 
while remaining on average within 31% of a best-of-breed solution within
each benchmark. For the first time, this evaluation validates that a WCOJ architecture can
compete on BI and LA workloads. We argue that the inherent benefits of 
such a unified (relational) design has the potential to outweigh its
minor performance overhead. 

We believe represents a first step in unifying relational algebra and
machine learning in a single engine. As such, we extend in <Ref> to explore some of the potential benefits of 
this approach on a full
application. We show here that is up to an order of magnitude 
faster than the popular (unified) solutions of MonetDB/Scikit-learn, Pandas/Scikit-learn, 
and Spark on a workload that combines SQL-style querying and a machine
learning algorithm. We hope adds to the debate surrounding the
design of unified querying systems.




 §.§ Related Work



extends previous work in worst-case optimal join processing (and
LogicBlox), relational data processing, and linear algebra processing.



  
*EmptyHeaded
is designed around many of the techniques presented in the engine 
<cit.>, but with many fundamental differences.
First, was only evaluated in the graph and 
resource description framework (RDF) domains. As such, the design is unable to support BI
workloads that contain many attributes. The limitations in the design stem
from its storage model and query model which was specifically designed for graph and
RDF workloads. In addition, the design only
supported equality selections and 's on indexed attributes. 
Finally, the design of 
was optimized for only sparse data and achieved suboptimal performance on
workloads with dense data. is designed to fix these flaws by leveraging
new techniques (<Ref>)
that eliminate these limitations, 
while preserving the theoretical benefits of such a design.



  
*LogicBlox

LogicBlox <cit.> is a full featured commercial database engine built around similar
worst case optimal join <cit.> and query compilation <cit.> algorithms. Still, a systematic evaluation of the LogicBlox
engine on BI and LA workloads is yet to be presented. 
From our conversations with LogicBlox, we learned that they 
often avoid using a WCOJ algorithm on these workloads in favor of more traditional
approaches to join processing. In contrast, always uses a WCOJ algorithm  
(even on queries not approaching the worst-case). Further, LogicBlox uses a query
optimizer that has similar benefits to 's (see 
<Ref>), but does so using custom algorithms <cit.>. 
In contrast, uses a generalization of these algorithms <cit.> 
that maps to well-known techniques <cit.>. Finally,
LogicBlox stores its data in manner that resembles a row store, whereas
resembles column store.



  
*Relational Processing A significant amount of work has focused on 
bringing LA to relational data processing engines. Some have
suggested treating LA objects as first class citizens in a column
store <cit.>. Others, such as Oracle's UTL_NLA <cit.>
and MonetDB with embedded Python <cit.> allows users to call LA 
packages through user defined functions. Still, the relational optimizers in these
approaches do not see, and therefore are incapable of optimizing, the 
LA routines. Even worse, these packages
are cumbersome to use and place significant burden on the user to make low-level
library calls. Finally, the SimSQL project <cit.> suggests that relational
engines can be modified in straightforward ways to accommodate LA
workloads. Our goals are similar to the SimSQL, but explored with different
mechanics. SimSQL studied the necessary modifications for a classic
database architecture to support LA queries and
was only evaluated on LA queries. In , we evaluate a new,
theoretically advanced, query processing architecture on both BI and LA 
workloads. Additionally, SimSQL was evaluated
in the distributed setting and therefore compared to higher-level baselines. 
is evaluated in a shared memory setting against Intel MKL and HyPer 
<cit.>. Other high performance
in-memory databases, like HyPer, focused on classic OLTP and OLAP
workloads and were not designed with other workloads in mind.



  
*Linear Algebra Processing Researchers have long studied how to
implement high-performance LA kernels. Intel 
MKL <cit.> represents the culmination of this work on Intel CPUs. 
Unsurprisingly, researchers have shown  <cit.> that it requires
tedious low-level code optimizations to come near the performance of a package
such as Intel MKL. As a result, processing these queries in a traditional RDBMS 
(using relational operators) is at least an order of magnitude slower than using 
such packages (see <Ref>). In response,
researchers have released array databases, like SciDB and TileDB
<cit.>, which provide high-level LA querying, often by 
wrapping lower-level libraries. 
In contrast, our goal is not to
design an entirely different and specialized engine for these workloads, but 
rather to design a single (relational) engine that processes multiple classes 
of queries efficiently.






§ LEVELHEADED ARCHITECTURE







In this section we overview the data model, storage engine, query compiler, 
and join algorithm at
the core of the architecture. This overview presents the 
preliminaries necessary to understand the optimizations presented in <Ref>. ingests structured data from delimited
files on disk and has a Python front-end that accepts Pandas dataframes. 
The query language is a subset of the SQL 2008 syntax 
with some standard limitations that we
detail in <Ref>. The input and output of each query is
a table which we describe in <Ref>. The SQL queries
are translated to generalized hypertree decompositions (GHD) which we describe
in
<Ref>. Finally, code is generated from the selected GHD using
a WCOJ algorithm which we describe in <Ref>. The entire process is shown in <Ref>. 



 §.§ Data Model


The data model is relational with some minor restrictions. A core
aspect of 's data model is that attributes are classified as either keys
or annotations via a user-defined schema. Keys in correspond to primary or foreign
keys and are the only attributes which can partake in a join. Keys cannot be
aggregated. Annotations are all other attributes and can be aggregated. 
Both keys and annotations support filter predicates and
 operations. In its current implementation, supports
equality () filters on keys and range (,,
) filters on annotations. This represents the
existing implementation, not fundamental restrictions. 's current
implementation supports
attributes with types of , , , 
, and . In many
regards is similar to a key-value store combined with the relational
model. is not unique in this regard, and commercial databases like
Google's Mesa <cit.> and Spanner <cit.> follow 
a similar (if not identical) model.



 §.§ Storage Engine


's data model is tightly coupled with how it stores relations. All key 
attributes from a relation are stored in a trie, which serves as the only 
physical index in . In the trie, each level is composed of sets of 
dictionary encoded (unsigned integer) values. As is 
standard <cit.>, stores dense sets using 
a bitset and sparse sets using unsigned integers. Annotations are stored
in separate buffers attached to the trie. supports multiple
annotations, and each can be reached from any level of the trie (core 
differences from ). An 
example trie is shown in <Ref> where the dimensions 
 and 
 are keys and  
 is an annotation.
tries can be thought of as
an index on key attributes or a materialized view of the input table.



 §.§ Query Compiler



We briefly overview the theoretical foundation of 's query compiler by
describing the essential details.



  
*Representing Queries as Hypergraphs
Like <cit.>, a query is represented using a 
hypergraph H = (V, E), where V is a
set of vertices (attributes) and each hyperedge e ∈ E 
(relation) is a subset of V. A join query is
represented as a subgraph of H. Example hypergraphs are shown in 
<Ref>. In <Ref> we describe
how SQL is translated to hypergraphs in more detail.



  
*Generalized Hypertree Decompositions

The query compiler uses generalized hypertree decompositions (GHDs) 
<cit.> to represent query plans. It is useful to think of 
GHDs as an analog of relational algebra for a WCOJ algorithm. Formally, given a 
hypergraph H = (V_H, E_H), a GHD is a tree T = (V_T, E_T) and a mapping 
χ : V_T → 2^V_H that associates each node of the tree with a 
subset of vertices in the hypergraph.  The asymptotic
runtime of a plan is bound by the fractional hypertree width (FHW) of the GHD 
<cit.> and, as is standard <cit.>,  
chooses a GHD with the lowest FHW to select a query plan that matches the best 
worst-case guarantee. 



 
  * Each edge of the hypergraph e ∈ E_H must be a subset of the vertices in one of the nodes of the tree, i.e. there exists a tree node t ∈ V_T such that e ⊆χ(t).

  * For each attribute v ∈ V_H, the tree nodes it is a member of, {t | v ∈χ(t)}, must form a connected subtree. This is called the running intersection property.


GHDs are designed to capture the degree of cyclicity in a query. We formalize this idea by defining the fractional hypertree width (FHW) of a GHD. Each node t of a GHD defines a subgraph (V', E'). That is, V' is χ(t) and E' is the set of edges that are subsets of χ(t).

To calculate the width of this node we define a linear program with one variable x_e for each hyperedge e ∈ E', with the constraints

    ∑_e ∈ E': e ∋ v x_e ≥ 1, ∀ v ∈ V'

and objective function ∑_e ∈ E' x_e. Consider a vertex v. The
constraint requires that the sum of the edges the vertex is contained in is at
least 1. The objective minimizes the sum of all the edges. The FHW of a GHD is
then defined as the maximum width of all of the nodes V_T. The asymptotic
runtime of a plan is bound by the FHW of the GHD and by choosing the GHD with
the lowest FHW, chooses the theoretically optimal plan.




  
*Capturing Aggregate-Join Queries

Aggregate-join queries are common in both BI and LA workloads.
To capture aggregate-join queries, uses the AJAR framework 
<cit.>. Despite that others <cit.>
have argued that custom algorithms are necessary to capture such queries,
AJAR extends the theoretical results of 
GHDs to queries with aggregations. AJAR does this by associating each tuple 
in a one-to-one mapping with an annotation. Aggregated annotations are 
members of a commutative semiring, which is equipped with product and 
sum operators that satisfy a set of properties (identity and annihilation, 
associativity, commutativity, and distributivity). Therefore, when relations are joined, 
the annotations on each relation are multiplied together to form the annotation 
on the result. Aggregations are expressed by an aggregation ordering 
α = (α_1, ⊕_1), (α_2, ⊕_2), … of attributes and
operators. 

Using AJAR, picks a query plan with the best worst-case guarantee 
by going through three phases. 
First, it breaks the input query into characteristic
hypergraphs, which are subgraphs of the input that can be decomposed to optimal 
GHDs. Second, for each characteristic hypergraph all possible
decompositions are enumerated, and a decomposition that minimizes FHW is chosen.
Finally, the chosen GHDs are combined to form an optimal GHD. To avoid
unnecessary intermediate results, compresses 
all final GHDs with a FHW of 1 into a single 
node, as the query plans here are always equivalent to running just a WCOJ algorithm. The GHDs for TPC-H 
query 5 and matrix multiplication are shown in <Ref>.





 §.§ Join Algorithm



The generic WCOJ algorithm <cit.> shown in <Ref> is the
computational kernel for all queries in the engine. This algorithm is
used in each node of the GHD-based query plan. The generic WCOJ 
algorithm can be asymptotically better than any pairwise join algorithm <cit.>. 
Intuitively, the generic WCOJ algorithm joins 
attributes in a multiway fashion as opposed to the traditional approach of
joining relations in a pairwise fashion. This means that
queries in are processed one attribute equivalence class at a time,
and the core operation to process an equivalence class is a set intersection. 
Selecting the order that these attributes are processed in is
the focus of <Ref>. An unrolling of this algorithm on
TPC-H query 5 and matrix multiplication is shown in <Ref>.




§ SQL TO GHDS



The query compilation techniques presented in
<Ref> are able to capture a wide
range of domains, including linear algebra, message passing, and graph queries
<cit.>.
However, most of this work has been theoretical and none of the
current literature demonstrates how to capture general SQL-style queries 
in such a framework. In this
section we show how to extend this theoretical work to more complex queries. 
In particular we describe
how translates generic SQL queries (which could not) to hypergraphs in <Ref>. Using this translation, we show how the 
well-known optimization of attribute elimination is captured both logically and physically in . 
In <Ref> we describe the manner in which selects a GHD
and optimizes it by pushing down selections.
We argue that with these extensions represents the first practical 
implementation of these techniques capable of capturing general SQL 
workloads.



 §.§ SQL to Hypergraph Translation



We demonstrate how certain features of SQL, such as complex
expressions inside aggregation functions, can be translated to operations on annotated
relations using a series of simple rules to construct query hypergraphs.
The rules uses to translate a 
SQL query to a hypergraph H = (V, E) and an aggregation ordering α are as follows:
 
  * The set V of vertices in the hypergraph contains all of the key columns in
the SQL query. The set of hyperedges E is each relation in the SQL query.  All attributes that appear in an equi-join condition are mapped to the same attribute in V. 

  * All key attributes that do not appear in the output of the query must be in
the aggregation ordering α.  

  * If only the columns of a single relation appear
inside of an aggregation function, the expression inside of the aggregation function is the annotation of that relation. If none of the relation's columns appear inside an aggregation function, the relation's annotation is the identity element. If the inner expression of an aggregation function touches multiple relations, those relations are constrained to be in the same node of the GHD where the expression is the output annotation. 

  * The rules above do not capture annotations which are not aggregated, so 
these annotations are added to a meta data container M that associates 
them to the hyperedge (relation) from which they originate.




Consider how these rules capture TPC-H query 5 from <Ref>
in a query hypergraph.

By Rule 1, the equality join in this query is captured in the set of 
vertices (V) and hyperedges (E) shown in the hypergraph in 
<Ref>. The columns  and
 must be mapped to the same vertex in “custkey" ∈ V.
 Similarly, the columns
 and  are mapped to the vertex “orderkey",
the columns  and  are mapped to the vertex
“suppkey", the columns , , and
 are mapped to the vertex “nationkey", the columns
 and  are mapped to the vertex
“regionkey".

By Rule 2, a valid aggregation ordering is:

    α = [regionkey, nationkey, suppkey, custkey, orderkey]

with the aggregation operator Σ (the order is irrelevant here).

To apply Rule 3, consider the expression inside the aggregation function, 
. Only columns on the  
table are involved in this expression, so the annotations on the  
table will be this expression for each tuple. The  and 
 tables do not have any columns in aggregation expressions, so 
they are annotated with the identity element.

By Rule 4, the hypergraph does not capture the 
attributes , , or  but our metadata 
container M does. M here is the following: 
{ ↔ ,  ↔ ,  ↔ }.

Thus, the final aggregate-join query is:



    ∑_regionkey, nationkey, suppkey, custkey, orderkey
        supplier(custkey, suppkey) ⋈ orders(custkey, orderkey) 
       ⋈ nation(nationkey, regionkey) ⋈ region(regionkey) 
       ⋈ lineitem(orderkey, suppkey)




Each (orderkey, suppkey) tuple on relation lineitem is annotated with
the value of  ,
and the
tuples on all other relations are annotated with the identity element.




*Attribute Elimination
The rules above only add the attributes that are used in the query to the hypergraph.
Although this elimination of unused attributes is obvious, ensuring this physically in the trie data structure was non-trivial. To do this we ensured that any number
of levels from the trie can be used during query execution.
This means that annotations can be reached individually from any level of the
trie. Further, the annotations are all stored in individual data buffers 
(like a column store)
to ensure that they can be loaded in isolation. These fundamental differences with
, enable to support attribute elimination both logically and
physically. This is essential on dense LA kernels because it enables 
to call BLAS routines by storing a single dense 
annotation in flat (BLAS compatible) buffer.



 §.§ GHD Optimization



After applying the rules in <Ref>, a GHD is selected 
using the process described in <Ref>. Still, needs a way 
to select among multiple GHDs that the theory cannot distinguish. In this 
section we explain how adapts the theoretical definition of GHDs to both
select and produce practical query plans.



  
*Choosing Among GHDs with the same FHW 
For many queries, multiple GHDs have the same 
FHW. Therefore, a practical implementation must 
also include principled methods to choose between query plans 
with the same worst-case guarantee. Fortunately, there are three intuitive characteristics 
of  GHD-based query plans that makes this choice relatively simple (and cheap). 
The first is that the smaller a GHD is (in terms of number of nodes and height), the quicker
it can be executed (less generated code). The second is that fewer intermediate
results (shared vertices between nodes) results in faster execution time. Finally,
the lower selection constraints appear in a query plan corresponds to how early
work is eliminated in the query plan. Therefore,
uses the following order of heuristics to choose between GHDs with the same
FHW:

 
  * Minimize |V_T| (number of nodes in the tree).

  * Minimize the depth (longest path from root to leaf).

  * Minimize the number of shared vertices
 between nodes.

  * Maximize the depth of selections.


Although most of the queries in this paper are single-node
GHDs, on the two node TPC-H query 5, using these rules to select a GHD
results in a 3x performance advantage over a GHD 
(with the same FHW) that violates the rules above. 
We explain how selections are pushed down below joins in GHDs next. 



  
*Pushing Down Selections Below Joins
Our mechanism of selecting a GHD ensures that selections appear as low as 
possible in a GHD, but we still need a mechanism to push down selections down 
below joins in our chosen GHD. Motivated by how does this 
<cit.>, the query 
compiler pushes down selections below joins by:

 
  * Taking as input an optimal (with respect to FHW) GHD T = (V_T, E_T) 
and a set of selections σ_a_i applied to attributes a_i.

  * For each σ_a_i: Let e_i be the edge that contains the vertex 
derived from a_i or that has the meta data M associated with a_i. 
Let t_i be the GHD node in V_T associated with e_i. If t_i contains more
than one hyperedge, create a new GHD node t'_i that contains only e_i, and
make t'_i a child of t_i.


This means that pushes down selections by creating new GHD
nodes with only the selection constraints under the original GHD
nodes. On TPC-H query 5 this results in the GHD
shown in <Ref> which executes 1.8x faster than 
the GHD without this optimization.





§ COST-BASED OPTIMIZER



After a GHD-based query plan is produced using the process described in 
<Ref>, needs to select an attribute
order for the WCOJ algorithm. Similar to the classic query optimization problem  
of selecting a join  
order <cit.>, WCOJ attribute ordering can result in 
orders of magnitude performance differences on the same query. Unfortunately, the known 
techniques for estimating the cost of join orders are designed for Selinger-style <cit.>
query optimizers using pairwise join algorithms—not a GHD-based query optimizer
with a WCOJ algorithm. In this section 
we present the first cost-based optimizer for the generic
WCOJ algorithm and validate the crucial observations it uses to derive its cost 
estimate. We show that this optimizer selects attribute orders 
that provide up to a three orders of magnitude speedup over ones that could 
select. 




  
*Optimizer Overview 's cost-based optimizer 
selects a key attribute order
for each node in a GHD-based query plan. As is standard <cit.>, 
requires that 
materialized key attributes appear before those that are projected away (with one 
important exception described in <Ref>) and that materialized attributes always adhere to some global ordering (e.g. if attribute `a' is before attribute `b' in one GHD node order, then 'a' must be before 'b' in all GHD orders). To assign an attribute order to each GHD node, 's cost-based optimizer: traverses the GHD in a top-down fashion,
considers all attribute orders[We
remind the reader that the number of attributes considered here is only the number 
of key (joined) attributes inside of a GHD node which is typically small.] adhering to the previously described criteria at each node, and selects the attribute order with the lowest cost estimate.

For each order, a cost estimate is computed 
based on two characteristics of the generic WCOJ algorithm: (1) the algorithm 
processes one attribute at a time and (2) set intersection is the bottleneck 
operator. As such, assigns a set intersection cost () and a cardinality
weight (<Ref>) to each key attribute (or vertex) in the 
hypergraph of a GHD node. Using this, the cost estimate for a given a key attribute (or hypergraph vertex) order 
[v_0,v_1,...,v_|V|] is:

    cost = ∑_i=0^|V|((v_i) ×weight(v_i))

The remainder of this section discusses how the s 
(<Ref>) and weights (<Ref>) are derived.



 §.§ Intersection Cost


The bottleneck of the generic WCOJ algorithm is set intersection operations. 
In this section, we describe how to derive a simple cost estimate, called 
, for the set intersections in the generic WCOJ algorithm.



  §.§.§ Cost Metric


Recall that the sets in tries are stored using an unsigned integer 
layout () if they are sparse and a bitset layout 
() if they are dense, a design inherited from . Thus, the 
intersection algorithm used is different depending on the data layout of the 
sets. These different 
layouts have a large impact on the set intersection performance,
even with similar cardinality sets. For example, 
<Ref>, shows that a  ∩  is roughly 
50x faster than a  ∩  with the same cardinality sets. 
Therefore, uses the results from 
<Ref> to assign the following 
s:


( ∩ ) = 1, 
( ∩ ) = 10, 

( ∩ ) = 50 




Unfortunately, it is too expensive for the query compiler to check (or track)
the layout of each set during query compilation—set layouts 
are chosen dynamically during data ingestion and a single trie can have millions
of sets. To address this uses <Ref>.



The sets in the first level of a trie are typically dense and therefore
represented as a bitset. The sets of any other level of a trie are typically 
sparse (unless the relation is completely dense) and therefore represented 
using the unsigned integer layout.



Empirical Validation: Consider a 
trie for the TPC-H  relation where the trie levels
correspond to the key attributes 
[,,,] 
in that order. At scale factor 10, each level of this trie has the following number
of  and  sets:
  



    
  * 1st level() = {0  sets, 1  set}
    
  * 2nd level() = {14999914  sets, 86  sets}
    
  * 3rd level() = {59984817  sets, 0  sets}
    
  * 4th level() = {59986042  sets, 0  sets}
  



Thus, given a key attribute order [v_0,...,v_|V|] (where each 
v_i ∈ V), the optimizer assigns an  
to each v_i, in order of appearance, using the following method which leverages
<Ref>:

 
  * For each edge e_j with node v_i, assign l(e_j) (where 
    l=layout), to either  or . 
    As a reminder, edges are relations and
    vertices are attributes. Thus, for each relation this assignment guesses one 
    data layout for all of the relation's v_i sets. 
    If e_j has been assigned with a previous vertex v_k where k < i, 
    l(e_j) =
    (not the first trie level), 
    otherwise 
    l(e_j) =. 
	
  * Compute the cost of intersecting the v_i attribute from each edge (relation) e_j.
	For a vertex with two edges, the pairwise  is used. 
    For a vertex with N edges, where 
	N > 2, the  is the sum of pairwise s where 
    the  sets are always processed first. For example,
    when N=3 and l(e_0) ≤ l(e_1) ≤ l(e_2) where  < , =  ( l(e_0) ∩l(e_1)) +  (l(l(e_0) ∩l(e_1)) ∩l(e_2)). Note, 
    = l( ∩ ). 



Consider the attribute order [,,,] for 
one of the GHD nodes in TPC-H query 5 (see <Ref>). 
The  vertex is assigned an  of 1 as it is 
classified with  intersections. The 
vertex is assigned an  of 10, classified with 
 intersections. The  
vertex is assigned an  of 11, classified with 
 intersections. Finally, 
the  vertex is assigned an 
 of 50, classified with  
intersections. 


Finally, in the special case of a completely dense relation, the LevelHeaded 
optimizer assigns an  of 0 because no set intersection is necessary in this 
case. This is essential to estimate the cost of LA queries properly.



  §.§.§ Relaxing the Materialized Attributes First Rule


An interesting aspect of the intersection cost metric is that the cheapest key 
attribute order could have materialized key attributes come after 
those which are projected away.[We remind the reader that later
(or after) in the attribute order corresponds to a lower level of the tries.] To
support such key attribute orders, the execution engine must be able to combine children 
(in the trie) of projected away key attributes using a set union or  
(to materialize the result sets). Unfortunately, it is
difficult to design an efficient data structure to union more than one
level of a trie (materialized key attribute) in parallel (e.g. use a massive 2-dimensional
buffer or incur expensive merging costs). Therefore, kept its design simple 
and never considered relaxing the
rule that materialized attributes must appear before those which are
projected away. In we relax this rule by allowing
1-attribute unions (see <Ref>) on keys when it can lower 
the .

Within a GHD node, relaxes the materialized attributes first rule under 
the following conditions:
 
  * The last attribute is projected away.

  * The second to last attribute is materialized.

  * The  is improved by swapping the two attributes. 

These conditions ensure that 1-attribute union will only be introduced when
the  can be lowered. 




 Consider the sparse matrix multiplication query and its unrolling
 of the generic WCOJ algorithm  for an attribute order of 
  shown in <Ref>. This attribute order has a
 cost 50  assigned to the  attribute.

Now consider an attribute
order  . Here a cost 10  is assigned to  and the 
unrolling of the generic WCOJ algorithm (where () denotes accesses to keys and [] denotes access 
to annotations) is the following:


for $̄i ∈π_iM1dos_j ←∅for $̄k ∈π_kM1(i) ∩π_kM2 do

for $̄j ∈π_jM2(k)dos_j[j]   +=   π_v1M1[i,k] * π_v2M2[k,j]out(i) ←s_jThis lower cost attribute order 
recovers the same loop ordering as Intel MKL on sparse matrix multiplication 
<cit.> and its bottleneck is the 
 operation that unions  values and sums   values. 
This is in contrast to the standard bottleneck operation of a set intersection as in the 
 order. In <Ref>
we discuss the tradeoffs around implementing this union add or  
operation. <Ref> shows that this order is essential to run 
sparse matrix multiplication as a join query without running out of memory.


 §.§ Weights


Like classic query optimizers,  also tracks the cardinality of 
each relation as this influences the s
for the generic WCOJ algorithm. <Ref> shows the 
unsurprising fact that larger cardinality sets result in longer intersection 
times. To take this into account when computing a cost estimate 
assigns weights to each vertex using <Ref>, which directly
contradicts conventional wisdom from pairwise optimizers:


The highest cardinality attributes should be processed first in the generic 
WCOJ algorithm. This enables these attributes to partake in
fewer intersections (outermost loops) and ensures that they are a higher 
trie levels (more likely 's with lower 
s). 


Empirical Validation: We show in <Ref> 
that the generic WCOJ can run over 70x faster on TPC-H query 5 when 
the high cardinality  attribute 
is first in the attribute order instead of last.
's 
goal when assigning weights is to follow <Ref> by 
assigning high cardinality attributes heavier weights so that they appear 
earlier in the attribute order. To do this,  assigns a cardinality score
to each queried relation and uses this to weight to each attribute. We describe
this in more detail next.



  
*Score maintains a cardinality score for each 
relation in a query which is just the relation's cardinality relative to the 
highest cardinality relation in the query. The score (out of 100) for 
a relationr_iis:score = ceiling(|r_i|/|r_heavy|×100)wherer_heavyis the highest cardinality relation in the query.


  
*Weight To assign a weight to each vertex  uses the highest score edge (or 
relation) with the vertex when a high selectivity (equality) constraint is present, otherwise  takes the lowest score edge (or relation).  
The intuition for 
using the highest score edge (or relation) with a high 
selectivity constraint is that this relation represents the amount of work that
could be filtered (or eliminated) at this vertex (or attribute).
The intuition for otherwise taking the lowest score edge (or relation) is that the 
output cardinality of an intersection is at most the size of the smallest set. 

Consider TPC-H Q5 at scale factor 10. The cardinality score for each relation 
here is:


score(lineitem) = 100, score(orders) = 26, score(customer)= 3, 
 
score(region) = 1, score(supplier) = 1, score(nation) = 1 


The weight for each vertex is (region is equality selected):


weight(orderkey) = min(26,100), weight(custkey) = min(3,26)  

weight(suppkey) = min(1,100), weight(nationkey) = min(1,1,3) 

weight(regionkey) = max(1,1) 


These weights are then used to derive the cost estimates shown in <Ref>.



§ GROUP BY TRADEOFFS


Designing a skew-resistant  operator requires careful 
consideration of the tradeoffs associated with its implementation. In this section we present these tradeoffs in the context
of . Although many s are automatically captured in 's 
tries, two classes of s (that were not supported in ) require 
an additional implementation in : 
 (1) a  to union 
keys in the attribute orders from <Ref> and (2) a  
on annotations. In this section we present the tradeoffs around two implementations 
for each type of  and describe a simple optimizer that automatically exploits these tradeoffs to 
select the implementation used during execution. In <Ref>, we show that this  optimizer 
provides up to a 875x and 185x speedup over using a single  implementation
on BI and LA queries respectively. 



  
*GROUP BY Key An important artifact of the key attribute orders from
<Ref> is that a projected away attribute can appear 
before a materialized attribute in . In this case a  
on a key attribute is needed to union the result. 
  selects between two implementations for this: (1) a hash map that 
 upserts (key, value) pairs as they are encountered and (2) a bitset 
 that unions (OR's) keys and a dense array to hold the values.
<Ref> shows that the performance of these implementations 
depends greatly on the density of the 
output key attribute set. The bitset performs better when the output density is high and the
hash map performs better when the output density is low. To predict the output 
set's density  leverages a simple
observation: the output density is correlated in a 
1-1 manner with the density of 
projected away attribute (set being looped over). As such,  uses the
density of the projected away attribute to predict the 
output set's density and select the implementation used.



  
*GROUP BY Annotation supports SQL queries with 
operations on annotations or on both annotations and keys.
To support this in parallel,  selects between two implementations of a 
parallel hash map: (1) libcuckoo <cit.> and (2) a per-thread
instance of the C++ standard library unordered map. In the context of a ,
the crucial operation for these hash maps is an upsert and we found that each implementation
worked best under different conditions. <Ref>
show that when the output key size is small (many collisions), the per-thread 
implementation  
outperforms libcuckoo by up to 4x. In contrast, libuckoo outperforms 
the per-thread approach by an order of magnitude when the output size
is large (few collisions).  Unfortunately, predicting output cardinalities is 
a difficult problem <cit.>, so  leverages a general 
observation to avoid this when selecting between these two approaches: 
libcuckoo is at worst close (<2x) to the 
performance of the 
per-thread approach when the hash-map key tuple is wide (>3 values), and the 
per-thread  approach is at worst close (<2x) to the performance of libcuckoo 
when the hash map key is small (<3 values). This trend is shown in 
<Ref>. Thus, the  optimizer 
selects
a per-thread hash map when the hash map key is a tuple of three elements or less 
and 
libcuckoo otherwise.





§ EXPERIMENTS


We compare  to state-of-the-art relational database management 
engines and LA packages on standard BI and LA benchmark queries. 
We show that  is able to compete within 2.5x of these 
engines, while sometimes outperforming them, and that the techniques from 
in <Ref> can provide
up to a three orders of magnitude speedup. This validates that 
a WCOJ architecture is a practical solution for both BI and LA queries.



 §.§ Setup

We describe the experimental setup for all experiments.



  
*Environment is a shared memory engine that runs and is evaluated on a single node server. 
As such, we ran all experiments on a single machine with a total of 56 cores 
on four Intel Xeon E7-4850 v3 CPUs and 1 TB of RAM. For all engines, we chose 
buffer and heap sizes that were at least an order of magnitude larger than the 
dataset to avoid garbage collection and swapping data out to disk.



  
*Relational Comparisons We compare to HyPer, MonetDB, and LogicBlox 
on all queries to highlight the performance of other relational 
databases. Unlike , these engines are unable to compete within an 
order of magnitude of the best approaches on BI and LA queries. We compare to 
HyPer v0.5.0 <cit.> as HyPer is a state-of-the-art in-memory 
RDBMS design. We also compare to the MonetDB Dec2016-SP5 release. MonetDB
is a popular columnar store database engine and is a widely used 
baseline <cit.>. Finally, we compare to LogicBlox v4.4.5 as 
LogicBlox is 
the first general purpose commercial engine to provide similar worst-case 
optimal join guarantees <cit.>. Our setup of LogicBlox  
was aided by a LogicBlox engineer. Because of this, we know that LogicBlox 
does not use a WCOJ algorithm for many of the join queries in the 
TPC-H benchmark.
HyPer, MonetDB, and LogicBlox are 
full-featured commercial strength systems (support transactions, etc.) and 
therefore incur inefficiencies that  does not.



  
*Linear Algebra Package Comparison We use Intel MKL v121.3.0.109 as the 
specialized linear algebra baseline. This is the best baseline for LA performance 
on Intel CPUs (as we use in this paper). Others <cit.> have shown 
that it takes considerable effort and tedious low-level optimizations to approach 
the performance of such libraries on LA queries.



  
*Omitted Comparisons We omit a comparison to array databases like SciDB 
<cit.> as these engines call BLAS or LAPACK libraries (like Intel MKL) on the
LA queries that we present in this paper. Therefore, Intel MKL was the 
proper baseline for our LA comparisons. Additionally, SciDB is not 
designed to process TPC-H queries.



  
*Metrics  For end-to-end performance, we measure the wall-clock time 
for each system to complete each query. We repeat each measurement seven times, 
eliminate the lowest and the highest value, and report the average. This 
measurement excludes the time used for outputting the result, data statistics 
collection, and index creation for all engines. We omit the data loading and 
query compilation for all systems except for LogicBlox and MonetDB where this 
is not possible. The query compilation time for these queries, especially at large
scale factors, is negligible to the execution time. Still, we found that 's 
unoptimized query compilation process performed within 2x of HyPer's.
To minimize unavoidable differences with disk-based engines 
(LogicBlox and MonetDB) we place each database in the tmpfs in-memory 
file system and collect hot runs back-to-back. Between measurements for the 
in-memory engines (HyPer and Intel MKL), we wipe the caches and 
re-load the data to avoid the use  of intermediate results.



 §.§ Experimental Results


We show that  can compete within 2x of HyPer on seven TPC-H queries and 
within 2.5x of Intel MKL on four LA queries, while outperforming
MonetDB and LogicBlox by up to two orders of magnitude.



  §.§.§ Business Intelligence


On seven queries from the TPC-H benchmark we show that  can compete 
within 2x of HyPer while outperforming 
MonetDB by up to an order of magnitude and LogicBlox by up to two orders 
of magnitude.



  
*Datasets We run the TPC-H queries at scale factors 1, 10, 
and 100. We stopped at TPC-H 100 as in-memory engines, such as HyPer, 
often use 2-3x more memory than the size of the input database during loading—therefore 
approaching the memory limit of our machine. For reference, on TPC-H query 1 HyPer uses
161GB of memory whereas  uses 25GB (only loading the data it needs from disk).

Additionally, scale factor 100 was the largest one where we could guarantee that
the buffer space for disk-based engines was an order of 
magnitude large than the data.


  
*Queries We choose TPC-H queries 1, 3, 5, 6, 8, 9 and 10 to 
benchmark, as these queries exercise the core operations 
of BI querying and also containing interesting join 
patterns (except 1 and 6). TPC-H queries 1 and 6 do not contain a join and 
demonstrate that although  is designed for join queries, it can also 
compete on scan queries. 



  
*Discussion In <Ref> we show that  can outperform 
MonetDB by up to 80x and LogicBlox by up to 270x while remaining 
within 1.88x of the highly optimized HyPer database. Unsurprisingly, the 
queries where  is the farthest off the performance of the HyPer engine is 
TPC-H queries 1 and 8. Here the output cardinality is small and the runtime
is dominated by the  operation.
As described in <Ref>,  leverages existing hash map 
implementations 
to perform these s, 
and a custom built (NUMA aware) hash map would likely close this performance gap. 
On queries 3 and 9, where the output cardinality is larger (and closer to 
worst-case),   is able to compete within 11% of Hyper and sometimes 
outperforms it. In , we also tested additional optimizations
like indexing annotations with selection
constraints and pipelining intermediate results between GHD nodes. We found that 
these optimizations could 
provide up to a 2x performance increase on queries like TPC-H query 8, but
did not consistently outweigh their added complexity.



  §.§.§ Linear Algebra


We show that  can compete within 2.5x of Intel MKL while outperforming 
HyPer and LogicBlox more than 18x and 587x respectively on 
LA benchmarks.



  
*Datasets We evaluate linear algebra queries on three dense 
matrices and three sparse matrices. The first sparse matrix dataset we 
use is the Harbor dataset, which is a 3D CFD model of the Charleston Harbor
<cit.>. The Harbor dataset is a sparse matrix 
<cit.> that contains 46,835 rows and columns and 
2,329,092 nonzeros. The second sparse matrix dataset we use is the 
HV15R dataset, which is a CFD matrix of a 3D engine fan <cit.>.
The HV15R matrix contains 2,017,169 rows and columns and 283,073,458 nonzeros 
and is a large, non-graph, sparse matrix. 
The final sparse matrix dataset we use is the nlpkkt240 dataset with 27,993,600 
rows and columns and 401,232,976 nonzeros <cit.>. The 
nlpkkt240 dataset is a symmetric indefinite KKT matrix.
For dense matrices, we use synthetic matrices with dimensions 
of 8192x8192 (8192), 12288x12288 (12288), and 16384x16384 (16384).



  
*Queries We run matrix dense vector multiplication and matrix 
multiplication queries on both sparse (SMV,SMM) and dense (DMV,DMM) matrices. 
These queries were chosen because they are simple to express using joins 
and aggregations in SQL and are the core operations for 
most machine learning algorithms. Further, 
Intel MKL is specifically designed to process these queries and, as a result, 
achieves the largest speedups over using a RDBMS here. 
Thus, these queries represent the most 
challenging LA baseline queries possible.  For both SMM and DMM we
 multiply the matrix by itself, as is standard for 
benchmarking <cit.>.



  
*Discussion<Ref> shows that  is able to compete
within 2.44x of Intel MKL on both sparse and dense LA queries. 
On dense data,  uses the attribute elimination optimization from 
<Ref> to store dense annotations in single buffers
that are BLAS compatible and code generates to Intel MKL. Still, MKL  
produces only the output annotation, not the key values, so
 incurs a minor performance penalty (<2%) for producing the key values.
On sparse data,  is able to compete with MKL when executing these LA queries
as pure aggregate-join queries. To do this, the attribute order 
and  optimizations from <Ref>
were essential. 
Although we tested more sophisticated optimizations, like cache 
blocking (by adding additional levels to the trie), we found that the performance
benefit of these optimizations did not consistently outweigh their added complexity.
In contrast, other relational designs fall flat on these LA queries. 
Namely, HyPer usually runs out of memory on the matrix multiplication query and, 
on the queries which it does complete, is often an order of magnitude slower 
than Intel MKL. Similarly, LogicBlox and MonetDB are at least an 
order of magnitude slower than Intel MKL on these queries.



 §.§ Micro-Benchmarking Results

We break down the performance impact of each optimization presented in
<Ref>.




  
*Attribute Elimination<Ref> show
that attribute elimination can enable up to a 4.82x performance advantage
on the TPC-H queries and up to a 500x performance advantage on 
dense LA queries. Attribute elimination is crucial 
on most on TPC-H queries, as these queries typically touch a small number of 
attributes from schemas with many attributes. Unsurprisingly, 
<Ref> shows that attribute elimination provides the largest 
benefit on the scan TPC-H queries (1 and 6) because it allows  to scan less 
data. On dense LA queries,  calls Intel MKL with little overhead because
attribute elimination enables us to store each 
dense annotation in a BLAS acceptable buffer. As <Ref> shows, 
this yields up to a 500x speedup over processing these queries purely
in . 



  
*Selections As shown in <Ref>, pushing down 
selections can provide up to 2.67x performance advantage 
on the TPC-H queries we benchmark.
Although pushing down selections is generally useful, on
TPC-H Q10 it actually hurts execution time. 
This is because this optimization adds additional GHD nodes (intermediate 
results) to our query plans (something pipelining would fix). Still, 
the performance impact is small on this query (<12%) and on average 
this optimization provided a 90% performance gain across TPC-H queries.



  
*Attribute Order
 As shown in <Ref>, the cost-based attribute ordering optimizer 
 presented in <Ref> can enable up to a 8815x performance 
 advantage on TPC-H queries and enables  to run sparse matrix multiplication
 as a join query without running out of memory.
<Ref> show the difference between the 
best-cost and the worst-cost attribute orders. 
The most interesting queries here are TPC-H query 5 and TPC-H query 8. On TPC-H
query 5, it is essential that
the high cardinality  attribute appears first. On TPC-H query
8, it is essential that the  attribute, which
was connected to an equality selection, appears first. The process of assigning
weights to the intersection costs in <Ref> ensures that orders 
satisfying these constraints are chosen. Finally, the cost-based 
attribute ordering optimizer is also crucial on sparse matrix multiplication. 
Here it is essential that the lower cost attribute order, with a projected 
away attribute before one that is materialized, is selected. This
order not only prevents a high cost intersection, but eliminates the 
computation and materialization of annotation values do not participate 
in the output (due to sparsity).



  
*GROUP BY Finally, we show in <Ref> 
that 's  optimizers provide up to a
875x performance advantage on TPC-H queries and up to a 185x 
performance advantage on LA queries. The  annotation optimizer 
provides a 875x speedup on TPC-H query 1 because  is the 
bottleneck operation in this query and our optimizer properly selects to use a 
per-thread hash map instead of libcuckoo here. The  key optimizer 
provides a 185x speedup on SMM with the nlp240 dataset because
it correctly predicts that the many of the output key attribute sets are
sparse. As such 's optimizer chooses to use a standard hash map to produce 
these sparse sets because using a bitset is highly inefficient (due to the 
amount of memory it wastes). This optimizer making the opposite choice provides 
a 3x performance advantage on SMM with the HV15R dataset. 




§ EXTENSION
 

We extend  to show that such a unified query processing architecture
could enable faster end-to-end applications. 
To do this, we add
the ability for  to process workloads that combine SQL queries and 
full machine learning algorithms (as described in <cit.>). With this, 
we show on a full-fledged application that  can be an order of magnitude 
faster than the popular solutions 
of Spark v2.0.0, 
MonetDB Dec2016[ Development build
with embedded Python <cit.>.]/Scikit-learn v0.17.1, and Pandas v0.18.1/Scikit-learn v0.17.1
(using the same experimental setting as <Ref>).



  
*Application We run a voter classification application 
<cit.> that joins and filters two tables to create a single feature
set which is then used to train a logistic regression model for 
five iterations. This application is a pipeline 
workload that consists of three pipeline phases: (1) a SQL-processing phase, (2) a feature
engineering phase where categorical variables are encoded, and (3) a 
machine learning phase. The dataset <cit.> consists of two tables: (1) one with
information such as gender and age for  
7,503,555 voters and
(2) one with information about the 2,751 precincts that the voters were registered in.



  
*Performance<Ref> shows that 
outperforms
Spark, MonetDB, and Pandas on the voter classification application by up to 
an order of magnitude. This is largely due to 's optimized shared-memory
SQL processing and ability to minimize data transformations between 
the SQL and training phase. To expand on the cost of data transformations a bit further, 
in <Ref> we show the cost of 
converting from a column store to the compressed sparse row  
format used by most sparse library packages. This transformation is not 
necessary in  as it always uses a single, trie-based data structure. 
As a result, <Ref> shows that up to 41 SMV queries can be
run in  in the time that it takes for a 
column store to convert the data to a BLAS compatible format.  
Similarly, on the voter classification application LevelHeaded avoids expensive 
data transformations (in the encoding phase) by using its trie-based data 
structure for all phases.




§ CONCLUSIONS


This paper introduced the  engine and
demonstrated that a query architecture built around WCOJs can compete on 
standard BI and LA benchmarks. We showed that  outperforms other relational 
engines by at least an order of magnitude on LA queries, while remaining 
on average within 31% of best-of-the-breed solutions on BI and LA 
benchmark queries. Our results are promising and suggest that 
such a query architecture could serve as the foundation for future 
unified query engines. 


Acknowledgments:  We thank LogicBlox for their helpful conversations 
and assistance in setting up our comparisons.



plain