network N consists of a universe U and a total function rfactsP
that maps each node of N to a subset of factsb from facts(U).
Here, facts(U ) denotes the set of all possible facts over U. A
node k is responsible for a fact f (under policy P) if f ∈ rfactsP (k).
For an instance I and a k ∈ N, let loc-instP,I (k) denote I ∩ rfactsP
(k), that is, the set of facts in I for which node k is responsible.
We refer to a given database instance I as the global instance
and to loc-instP,I (k) as the local instance on node k.
The result [Q, P](I) of the distributed evaluation in one
round of a query Q on an instance I under a distribution policy
P is defined as the union of the results of Q evaluated over
every local instance. Formally,
Example 5. 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),
from Example 2. 1. Consider a network Ne consisting of two
nodes {k1, k2}. Let P1 = ({a, b, c}, rfactsP) be the distribution
policy that assigns all Like-facts to both nodes k1 and k2, and
every fact Dislike(d1, d2) to node k1 when d1 = d2 and to node k2
otherwise. Then,
loc-instP1, Ie (k
1) =
{Like (a, b), Like(b, a), Like(b, c), Dislike(a, a)}
and
loc-instP1, Ie (k
2) =
{Like(a, b), Like(b, a), Like(b, c), Dislike(c, a)}.
Furthermore,
[Qe, P1](Ie) =
Qe (loc-instP1, Ie (k
1)) ∪ Qe (loc-instP1, Ie (k
2)),
which is just {H(a, b)} ∪ {H(a, c)}.
We get [Qe, P2](Ie) = 0 / for the distribution policy P2 that
assigns all Like-facts to node k1 and all Dislike-facts to
node k
2.
Now we can define parallel-correctness.
Definition 5. 2. A query Q is parallel-correct on instance I
under distribution policy P if Q(I) = [Q, P](I).
Q is parallel-correct under distribution policy P = (U, rfactsP), if
it is parallel-correct on all instances I ⊆ facts(U).
We note that parallel-correctness is the combination of
• parallel-soundness: [Q, P](I) ⊆ Q(I), and
• parallel-completeness: Q(I) ⊆ [Q, P](I).
The technique in Example 4. 1 can be generalized to arbitrary conjunctive queries and was first introduced in the
context of MapReduce by Afrati and Ullman1 as the Shares
algorithm. The values αx, αy, and αz are called shares (hence,
the name) and the work of Afrati and Ullman focuses on
computing optimal values for the shares minimizing the
total load (as a measure for the communication cost).
Beame et al.
4, 5 show that the method underlying Example 4. 1
is essentially communication optimal for full conjunctive
queries Q. Assuming that the sizes of all relations are equal
to m and under the assumption that there is no skew, the
maximum load per server is bounded by O (m/p1/t*) with high
probability. Here, t depends on the structure of Q and corresponds to the optimal fractional edge packing (which for
Q2 is t = 3/2). The algorithm is referred to as HyperCube in
Refs.
4, 5 Additionally, the bound is tight over all one-round
MPC algorithms, indicating that HyperCube is a fundamental algorithm.
Chu et al.
7 provide an empirical study of HyperCube (in
combination with a worst-case optimal algorithm for sequential evaluation16, 22) for complex join queries, and establish,
among other things, that HyperCube performs well for join
queries with large intermediate results. However, HyperCube
can perform badly on queries with small output.
5. PARALLEL-CORRECTNESS
In the remainder of this paper, we present a framework for
reasoning about data partitioning for generic one-round
algorithms for the evaluation of queries under arbitrary distribution policies. We recall from the introduction that such
algorithms consist of a distribution phase (where data is
repartitioned or reshuffled over the servers) followed by a
computation phase where each server evaluates the query
at hand over the local data. In particular, generic one-round
algorithms are one-round MPC algorithms where every server
in the computation phase evaluates the same given query.
When such algorithms are used in a multi-query setting, there is room for optimization. We recall that the
HyperCube algorithm requires a reshuffling of the base data
for every separate query. As the amount of communication
induced by a reshuffling of the data can be huge, it is relevant to detect when the reshuffle step can be avoided and
the current distribution of the data can be reused to evaluate another query. Here, parallel-correctness and parallel-correctness transfer become relevant static analysis tasks.
We study parallel-correctness in this section and parallel-correctness transfer in Section 6.
Before we can address the parallel-correctness problem
in detail, we first need to fix our model and our notation.
A characteristic of the HyperCube algorithm is that it
reshuffles data on the granularity of facts and assigns each
fact in isolation (i.e., independent of the presence or absence
of any other facts) to a subset of the servers. This means that
the HyperCube reshuffling is independent of the current
distribution of the data and can therefore be applied locally
at every server. We therefore define distribution policies as
arbitrary mappings linking facts to servers.
Following the MPC model, a network N is a nonempty finite
set of node names. A distribution policy P = (U, rfactsP) for a
b We mention that for HyperCube distributions, the view is reversed: facts
are assigned to nodes. However, both views are essentially equivalent and we
will freely adopt the view that fits best for the purpose at hand.