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.