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