94 COMMUNICATIONS OF THE ACM | MARCH 2017 | VOL. 60 | NO. 3
guarantees that the given (next) query can be evaluated correctly without reshuffling. To this end, we formalize the following decision problems:
Parallel-Correctness: Given a distribution policy and a query,
can we be sure that the corresponding generic one-round algorithm will always compute the query result
correctly—no matter the actual data?
Parallel-Correctness Transfer: Given two queries Q and Q′,
can we infer from the fact that Q is computed correctly
under the current distribution policy, that Q′ is computed correctly as well?
We say that parallel-correctness transfers from Q to Q′,
denoted , when Q′ is parallel-correct under every
distribution policy for which Q is parallel-correct. Parallel-correctness transfer is particularly relevant in a setting of
automatic data partitioning where an optimizer tries to
automatically partition the data across multiple nodes to
achieve overall optimal performance for a specific workload of queries (see, e.g., Refs.
15, 18). Indeed, when parallel-correctness transfers from a query Q to a set of queries
S, then any distribution policy under which Q is parallel-correct can be picked to evaluate all queries in S without
reshuffling the data.
We focus in this paper on conjunctive queries and first
study the parallel-correctness problem. We give a characterization of parallel-correctness: a distribution policy is
parallel-correct for a query, if and only if for every minimal
valuation of the query there is a node in the network to which
the distribution assigns all facts required by that valuation.
This criterion immediately yieldsa a upper bound for
parallel-correctness, for various representations of distribution policies. It turns out that this is essentially optimal,
because the problem is actually -complete. These results
also hold in the presence of union and inequalities. When
negation is added, deciding parallel-correctness might
involve counterexample databases of exponential size. More
specifically, in the presence of negation deciding parallel-correctness is coNEXPTIME-complete. The latter result is
related to the new result that query containment for conjunctive queries with negation is coNEXPTIME-complete,
as well.
For parallel-correctness transfer we also first provide a
semantical characterization in terms of a (value-based) containment condition for minimal valuations of Q′ and Q (Proposition
6. 4). Deciding transferability of parallel-correctness for conjunctive queries is -complete, again even in the presence of
unions and inequalities. We emphasize that the implied exponential time algorithm for parallel-correctness transfer does
not rule out practical applicability because the running time
is exponential in the size of the queries and not in the size of
a database.
Outline. In Section 2, we introduce the necessary preliminaries regarding databases and conjunctive queries. In Section 3,
we discuss the MPC model. In Section 4, we exemplify the
HyperCube algorithm. In Sections 5 and 6, we explore parallel-correctness and parallel-correctness transfer. We present concluding remarks together with direction for further research
in Section 7.
2. CONJUNCTIVE QUERIES
In this article, a (database) instance I is a finite set of facts
of the form R(a1, . . ., an), where R is an n-ary relation symbol
from a given database schema and each ai is an element
from some given infinite domain dom.
A conjunctive query (CQ) Q is an expression of the form
H(x) ← R1(y1), . . . , Rm( ym),
A valuation for a CQ Q maps its variables to values, that
is, it is a function V: vars(Q) → dom. We refer to V (bodyQ) as
the facts required by V. A valuation V satisfies Q on instance
I if all facts required by V are in I. In that case, V derives the
fact V (headQ). The result of Q on instance I, denoted Q(I), is
defined as the set of facts that can be derived by satisfying
valuations for Q on I. We denote the class of all CQs by CQ.
Example 2. 1. Let Ie be the example database instance
{Like(a, b), Like(b, a), Like(b, c), Dislike(a, a), Dislike(c, a)},
and Qe be the example CQ
H (x1, x3) ← Like(x1, x2), Like(x2, x3), Dislike(x3, x1).
ThenV1 = {x1 a, x2 b, x3 a} and V2 = {x1 a, x2 b, x3
c} are the only satisfying valuations. Consequently, Qe(Ie) =
{H(a, a), H(a, c)}.
3. MPC MODEL
The MPC model was introduced by Koutris and Suciu12 to study
the parallel complexity of conjunctive queries. It is motivated
by query processing on big data that is typically performed on
a shared-nothing parallel architecture where data is stored on
a large number of servers interconnected by a fast network. In
the MPC model, computation is performed by p servers con-
nected by a complete network of private channels. Examples
of such systems include Pig,
17 Hive,
20 Dremel,
13 and Spark.
23
The computation proceeds in rounds where each round con-
sists of two distinct phases:
• Communication Phase: The servers exchange data by
communicating with all other servers.
• Computation Phase: Each server performs only local
computation (on its local data).
The number of rounds then corresponds to the number of syn-
chronization barriers that an algorithm requires. The input
data is initially partitioned among the p servers and every
aIn this article, we refer to standard complexity classes like the famous
class NP, two classes from the second and third level of the polynomial hier-
archy, P2p and P3p, respectively, and the exponential time analogon of coNP,
coNEXPTIME. More information can be found in any textbook on computa-
tional complexity, for example, see Ref.
4