Delivering superior expressive power over RDBMS, while maintaining competitive performance, has represented the main goal and technical challenge for deductive database research since its inception forty years ago. Significant progress toward this ambitious goal is being achieved by the DeALS system through the parallel bottom-up evaluation of logic programs, including recursive programs with monotonic aggregates, on a shared-memory multicore machine. In DeALS, a program is represented as an AND/OR tree, where the parallel evaluation instantiates multiple copies of the same AND/OR tree that access the tables in the database concurrently. Synchronization methods such as locks are used to ensure the correctness of the evaluation. We describe a technique which finds an efficient hash partitioning strategy of the tables that minimizes the use of locks during the evaluation. Experimental results demonstrate the effectiveness of the proposed technique - DeALS achieves competitive performance on non-recursive programs compared with commercial RDBMSs and superior performance on recursive programs compared with other existing systems.
MOHAN YANG, ALEXANDER SHKAPSKY, CARLO ZANIOLO
There has been much research on improving the performance of Datalog systems
through parallel bottom-up evaluation. Previous studies focused on the message
passing model where processors communicate with each other by exchanging
messages. This includes both strategies for programs that can be evaluated without any
communication
The key to achieving a good performance in the shared-memory model is to
use as little synchronization as possible in the program evaluation. In this paper,
we present the technique used by the Deductive Application Language System
(DeALS)1 under development at UCLA, which extends the LDL++ technology
The rest of this paper is organized as follows. We introduce the concept of lockfree programs in Section 2. We describe how DeALS tries to find a lock-free evaluation plan in Section 3. An overview of DeALS is presented in Section 4. We report experimental results in Section 5. Related work is discussed in Section 6. The paper concludes in Section 7.
Let arc be a base relation that represents the edges of a directed graph. The transitive closure tc of arc is a derived relation that contains all the pairs (X; Y) where there is a path from X to Y in the graph. The following program is used to compute tc.
r1:1 : tc(X; Y) <- arc(X; Y): r1:2 : tc(X; Y) <- tc(X; Z); arc(Z; Y): The bottom-up evaluation of this program works as follows. The exit rule r1:1 is evaluated first. A tuple (X; Y) is added to tc for each tuple (X; Y) in arc. Then the left-linear recursive rule r1:2 is evaluated. For each tuple (X; Z) in tc, a tuple (X; Y) is added to tc for each tuple of the form (Z; Y) in arc.5 r1:2 is repeatedly evaluated until tc does not change between two successive evaluations of r1:2.
Now we describe how to parallelize the bottom-up evaluation on a shared-memory
machine with n processors. The parallel evaluation strategy presented in this
paper uses the same workflow as the sequential bottom-up evaluation to ensure the
correctness of evaluation, while the parallelism is achieved through the parallel
evaluation of each single rule (including the parallel pipelined evaluation of all the
rules which support its goal as described in Section 3). As shown in our
experiments, our strategy is able to achieve a reasonable speedup for a data intensive
application to the point that we do not need to explore rules level and components
level parallelism
We divide each relation into n partitions and we use the relation name with a superscript i to denote the i-th partition of the relation. For each partition, we use a lock to ensure the atomicity of each write operation if multiple write operations can occur concurrently. Let h be a hash function that maps a vertex to an integer between 1 to n. Both arc and tc are partitioned by the first column, i.e., h(X) = i for each (X; Y) in arci and h(X) = i for each (X; Y) in tci. Assuming that there are one coordinator and n workers, the parallel evaluation proceeds as follows. (1) The i-th worker evaluates r1:1 by adding a tuple (X; Y) to tc for each tuple (X; Y) in arci. (2) Once all workers finish Step (1), the coordinator notifies each worker to start
Step (3). (3) For each tuple (X; Z) in tci, the i-th worker looks for tuples in the form (Z; Y) in arc and adds a tuple (X; Y) to tc. (4) Once all workers finish Step (3), the coordinator checks if the evaluation for tc is completed. If so, the computation terminates; otherwise the computation starts from Step (3).
In Step (1) and Step (3), each worker performs its task on one processor while the coordinator waits. Step (2) and Step (4) serve as synchronization barriers.
In Step (1), the i-th worker only writes to tci since every tuple (X; Y) it adds to tc is taken from arci where h(X) = i. In Step (3), the i-th worker reads from tci and only writes to tci since every tuple (X; Y) it adds to tc is derived from a tuple (X; Z) in tci and a tuple (Z; Y) in arc where h(X) = i. There is no need for locks since tci is only accessed by the i-th worker during evaluation. The above evaluation plan is called a lock-free plan where no lock is needed, and a program which has a lock-free plan is called a lock-free program.
A key factor that enables a lock-free plan is the partitioning strategy. If we keep the current partitioning for arc but instead partition tc by its second column, then every worker could write to tci in Step (3), and an exclusive lock (x-lock) is needed to ensure every write operation to tci is atomic. For some programs, it is possible that a worker reads from a partition of a relation while the same partition is being modified by another worker. In this case, a readers-writer lock (rw-lock) is needed to ensure that every write operation is atomic, read operations will not get a tuple that is partially updated, and concurrent read operations are allowed when the partition is not being modified by other workers.
Another key factor is the selection of hash functions. It is possible to get a
lockfree plan even if we use different hash functions for arc and tc. However, in this
paper we focus on the case where every relation is partitioned using the same hash
function h defined as
t
X g(xi)
i=1
h(x1;
; xt) =
mod n;
(2)
where the input to h is a tuple of any arity t, g is a hash function with a range
no less than n, and P can be replaced with any commutative function. Using a
commutative function to combine the hash values of h allows for more parallelism
than an arbitrary function, such as concatenation (jj) where h(x1; ; xt) becomes
g(x1)jj jjg(xt)
Discriminating Sets and Parallel Program Evaluation. Consider the evaluation of a stratifiable program P that consists of l rules r1; ; rl, where each rule ri is in the form pi;0 <- pi;1; ; pi;mi . Here pi;0 is a predicate that serves as the head of ri, mi is the number of predicates in the body of ri, and each pi;j is a predicate in the body for j = 1; ; mi. Assume that a predicate p is associated with a relation of the same name and arity as the predicate, and we use R(p) to denote the relation that stores all tuples corresponding to facts about p in memory. A discriminating set of a (non-nullary) relation R is a non-empty subset of columns in R. Given a discriminating set of a relation, we divide the relation into n partitions by the hash values of the columns that belong to the discriminating set.
Let Pj be the program to be executed by the j-th worker. For each rule ri, we add the following rule in Pj : j ri : pi;0 <- pi;1; ; pi;mi ; h(pi;ti [Xi]) = j; (3) where pi;ti is a predicate selected from the mi predicates in the body for a ti between 1 to mi, Xi is a selected discriminating set of R(pi;ti ), pi;ti [Xi] denotes a tuple of arity jXij by retrieving the arguments in pi;ti whose positions belong to Xi, and the last predicate h(pi;ti [Xi]) = j means a tuple from the j-th partition of R(pi;ti ) is accessed in every successful derivation of rij . We select the same pi;ti and Xi for j = 1; ; n. We also select a discriminating set, denoted by X(R), for each derived relation R. These discriminating sets can be arbitrarily selected as long as there is a unique discriminating set for each derived relation. We might have selected several different discriminating sets of the same base relation which correspond to different ways of partitioning the relation. This relation is preprocessed before the evaluation so that it can be efficiently accessed for every partitioning.
In general, the parallel evaluation proceeds as follows. (1) The coordinator determines the first rule to be evaluated, say ri, and instructs the workers to start Step (2). (2) All n workers become active with the j-th worker evaluating rij . (3) After all workers finish, the coordinator checks if the evaluation for P is completed. If not, it determines the next rule to be evaluated and Step (2) repeats.
The parallel evaluation scheme described above is a special case of the substitution
partitioned scheme
The corresponding Pj in the lock-free plan for the program in (1) is shown as follows.
r4:1 : tc(X; Y) <- arc(X; Y); h(X) = j: r4:2 : tc(X; Y) <- tc(X; Z); arc(Z; Y); h(X) = j:
For each worker, there are still many different ways to evaluate r4:2 since the
three predicates in the body can be evaluated in different orders. For a specific
order, a bound/free adornment
The program in (5) shows an adorned version of the program in Example 1. The predicates in r5:2 are reordered to reflect the order in which they are evaluated. In the evaluation of r5:2, the j-th worker evaluates tcff(X; Z); h(X) = j by reading from the j-th partition of tc. Then it evaluates arcbf(Z; Y) where Z is bound. So it finds a tuple (Z; Y) from arc with the given Z, and adds (X; Y) to tc if it succeeds. r5:1 : tcff(X; Y) <- arcff(X; Y); h(X) = j: r5:2 : tcff(X; Y) <- tcff(X; Z); h(X) = j; arcbf(Z; Y): attendf(X) <- organizerf(X); h(X) = j: attendf(Y) <- cntfriendsff(Y; N); h(Y) = j; N 3:
In DeALS, a program is represented as an AND/OR tree
The attend program below finds all the people who will attend the party. A person
will attend the party if he/she is an organizer, or he/she has at least three friends
who will attend the party. Here, mcount is the monotonic and continuous count
aggregate
cntfriendsff(Y; mcount<X>) <- friendff(X; Y); h(Y) = j; attendb(X): relations when it is necessary — if 1) it is the root; or 2) it is an aggregation; or 3) the optimizer determines that the cost of computing the tuples when they are needed is much higher than the cost of materializing the relation in memory.6
A program is transformed into an AND/OR tree such that the root represents the query and the children of each AND node follow the same order as the corresponding predicates in the corresponding rule. The AND/OR tree is then adorned for bottomup evaluation. An OR node is an entry node if 1) it is a leaf, and 2) it is the first R-node among its siblings, and 3) each of its ancestor OR node does not have a left-sibling (i.e., a sibling that appears before the current node) that has an R-node descendant or a W-node descendant.
A non-linear formulation of the transitive closure program is shown as follows. tc(X; Y) <- arc(X; Y): tc(X; Y) <- tc(X; Z); tc(Z; Y): (8) The corresponding AND/OR tree is shown in Fig. 1. The text description inside a node indicates the type and ID of the node, e.g., “OR-1” indicates the root is an OR node with ID 1. OR-4, OR-5 and OR-6 are R-nodes and OR-1 is a W-node. OR-4 and OR-5 are entry nodes in this program.
OR-1 tc(X, Y) tc(X, Y) AND-2
AND-3 tc(X, Y) arc(X, Y) OR-4 tc(X, Z) OR-5
OR-6 tc(Z, Y)
In DeALS, the j-th worker evaluates an entry node by instantiating variables with constants from the j-th partition of the corresponding relation. For the remaining R-nodes, each worker has full access to all partitions of the corresponding relations. This strategy ensures that the evaluation is divided into n disjoint parts; otherwise if each worker evaluates an entry node by instantiating variables with constants from all partitions of the corresponding relation, redundant work will be performed. 1 Determining the Discriminating Sets. Now we describe the read/write analysis on an adorned AND/OR tree that determines the type of lock needed for each derived relation. The analysis performs a depth-first traversal on the AND/OR tree that simulates the actual evaluation to check each read or write operation performed by the j-th worker. For each node encountered during the traversal: 6 Currently this is done manually where a user can force the materialization of a relation by an annotation in the program. (1) if it is an entry node, set it as the current entry node; if it reads from a derived relation, for each W-node that is an ancestor of the current node,7 determine if the j-th worker only writes to the j-th partition of R(pw) by checking if pe[Xj ] = pw[Xk]8, where pe and pw are the predicates associated with the entry node and W-node, respectively, and Xj and Xk are the corresponding discriminating sets; (2) if it is an R-node that reads from a derived relation, determine if the j-th worker only reads from the j-th partition of R(pr) by checking if Xk B and pe[Xj ] = pr[Xk], where pe and pr are the predicates associated with the current entry node and R-node, respectively, Xj and Xk are the corresponding discriminating sets, and B is the set of positions for bound arguments in the R-node.
We use the following discriminating set equations (DSE) obtained through a
read/write analysis to find the best possible discriminating sets for a given program.
A DSE consists of three types of equations:
(1) pe[Xj ] = pw[Xk] for each entry node in Case (1) of the read/write analysis;
(2) Xk B, pe[Xj ] = pr[Xk] for each R-node in Case (2) of the read/write analysis;
(3) ; ( X f1; ; arity(R)g for each X appearing in the above two types of
equations, where arity(R) is the arity of the relation R associated with X.
The target is to find an optimal solution to the DSE which is an assignment to the
variables that satisfies all the equations of Type (3) and maximizes the number of
satisfied equations of Type (1) and Type (2). DeALS finds an optimal solution by
enumerating all possible assignments, which is feasible for most recursive programs
studied in the literature since the number of variables is very small. DSE is very
similar to the idea of generalized pivoting where a system of equations is obtained
from the rules and an exact solution is required
Finally, the discriminating sets are determined as follows.
(1) Obtain the DSE by performing a read/write analysis on the AND/OR tree.
(2) Find an optimal solution to the DSE.
(3) Determine the type of lock for each derived relation by a read/write analysis
on the AND/OR tree with the selected discriminating sets.
7 The set of ancestor W-nodes of an entry node can be efficiently obtained using a stack during
the traversal
The DSE for the AND/OR tree in Fig. 1 is shown is shown below. arc(X; Y)[X1] = tc(X; Y)[X2] tc(X; Z)[X2] = tc(X; Y)[X2] X2 f1g tc(X; Z)[X2] = tc(Z; Y)[X2] ; ( X1 ; ( X2 f1; 2g f1; 2g (9) The optimal solution is X1 = X2 = f1g that only violates the fourth equation. The result of a read/write analysis determines that an rw-lock is needed for tc in the evaluation of the non-linear recursive rule.
We assume the program does not contain aggregates and arithmetic expressions in the above procedure. If the program does contain these constructs, the same procedure is applicable if we ignore all the arguments which are either aggregates or arithmetic expressions. The only exception is that the evaluation of a W-node always requires locks if it contains an aggregate with no group by arguments.
4 DeALS DeALS is a Datalog system under development at UCLA. It has a Java-based compiler and a sequential Java-based interpreter that allows users to develop and debug their applications. The new parallel evaluation module targeted for sharedmemory multicore machines is implemented as a separate module in the system — the compiler compiles a program into an AND/OR tree and then the parallel module determines the parallel evaluation plan using the technique presented in this paper and generates a corresponding C++ program. We implemented the database objects (index and storage), base classes for each kind of node in the AND/OR tree and common functions. The generated program contains the definition of tuples and relations, and the actual implementation of the AND/OR tree based on the base classes. It is compiled into the final executable by invoking the Visual C++ Compiler that comes with Visual Studio 2013 (v120) on a Windows machine or GCC 4.9.2 on a Linux machine. The thread implementation provided by the Microsoft Windows runtime library is used on a Windows machine to evaluate the query in parallel, while Pthreads is used on a Linux machine.
In this section, we report some experimental results on evaluating both non-recursive and recursive programs. The test machine has four AMD Opteron 6376 CPUs (16 cores per CPU) and 256GB memory (configured into eight NUMA regions). The operating system is Ubuntu Linux 12.04 LTS. All execution times are calculated by taking the average of five runs of the same experiment. 1.143 1
DeALS
VectorWise 2.0.1
SQL Server 2014
7.755
Exp-I: Non-recursive programs — TPC-H benchmark. The benchmark
contains 22 (non-recursive) SQL queries over a database of eight tables. The data types
involved in the queries are integer, decimal, string and date. We implemented all the
22 queries following the query plans described in
Exp-II: Recursive programs. We test the performance of DeALS on three classical recursive queries — tc (program in (1)), attend (program in Example 3) and sg (same generation) as shown below: sg(X; Y) <- anc(P; X); anc(P; Y); X 6= Y: sg(X; Y) <- anc(A; X); sg(A; B); anc(B; Y): (10) The test datasets contain synthetic graphs and real world graphs. 9 COUNT(DISTINCT) is replaced with COUNT in query16. ORDER BY and LIMIT are ignored in our program. The evaluation time will not change significantly if we add these constructs since most queries return very few results except query3 and query10. 10 TPC-H Result on Dell PowerEdge R720, http://www.tpc.org/3282. 11 TPC-H Result on Cisco UCS C460 M4 Server, http://www.tpc.org/3311. .02 .02 27 17 26 2 .
6 tree-11
1) tree-11 is a randomly generated tree of depth 11 where the out degree of a non-leaf vertex is a random number between 2 to 6. 2) grid-150 is a 151 151 square grid. 3) gnp-10K is a random graph (Erdős-Rényi model) of 10,000 vertices generated by connecting vertices randomly such that the average out-degree of a vertex is 10.
1) patent is the US patent citation graph12. Each vertex represents a patent, and each edge represents a citation between two patents. It has 3,774,769 vertices and 16,518,948 edges. 2) wiki is the Wikipedia knowledge graph. Each vertex represents an entity in the Wikipedia, and each edge represents an appearance of an entity in another entity’s infobox. It has 3,165,181 vertices and 23,190,820 edges.
The experiments on tc (sg) evaluation use each of these synthetic graphs as arc (anc). The experiments on attend evaluation use each real world graphs as friend, while organizer contains all the vertices in the graph whose in-degrees are zero.
We compare DeALS with LogicBlox 4.1.9
The parallel evaluation strategy proposed in this paper uses a simple hash-based
data partitioning strategy. Various data partitioning strategies for parallel
bottomup evaluation have been studied in
In this paper, we presented the technique used in DeALS for parallel
bottomup evaluation on shared-memory multicore machines. The technique is simple and
applicable to a wide range of non-recursive and recursive programs. DeALS is
able to achieve competitive performance on non-recursive programs compared with
RDBMSs and superior performance on recursive programs compared with other
existing systems, by adding a parallel evaluation module based on this technique.
However, there is still a clear performance gap between DeALS and the hand
written optimal programs, such as the SSC12 algorithm for transitive closure
This work was supported by NSF grants IIS 1218471 and IIS 1118107. We would like to thank the reviewers and Matteo Interlandi for their comments. We thank LogicBlox especially Martin Bravenboer, Dung Nguyen, and Yannis Smaragdakis, for their assistance with the LogicBlox comparison.