Distributed Set Reachability
Sairam Gurajada
Max-Planck-Institute for Informatics
Campus E1.4
66123 Saarbrücken, Germany
gurajada@mpi-inf.mpg.de
Martin Theobald
University of Ulm
James-Franck-Ring O27
89069 Ulm, Germany
ABSTRACT
In this paper, we focus on the efficient and scalable process-
ing of set-reachability queries over a distributed, directed
data graph. A set-reachability query is a generalized form
of a reachability query, in which we consider two sets Sand
Tof source and target vertices, respectively, to be given as
the query. The result of a set-reachability query are all pairs
of source and target vertices (s, t), with s∈Sand t∈T,
where sis reachable to t(denoted as S;T). In case the
data graph is partitioned into multiple, edge- and vertex-
disjoint subgraphs (e.g., when distributed across multiple
compute nodes in a cluster), we refer to the resulting set-
reachability problem as distributed set reachability. The key
goal in processing a distributed set-reachability query over
a partitioned data graph both efficiently and in a scalable
manner is (1) to avoid redundant computations within the
local compute nodes as much as possible, (2) to partially
evaluate the local components of a set-reachability query
S;Tamong all compute nodes in parallel, and (3) to
minimize both the size and number of messages exchanged
among the compute nodes.
Distributed set reachability has a plethora of applications
in graph analytics and for query processing. The current
W3C recommendation for SPARQL 1.1, for example, intro-
duces a notion of labeled property paths which resolves to
processing a form of generalized graph-pattern queries with
set-reachability predicates. Moreover, analyzing dependen-
cies among social-network communities inherently involves
reachability checks between large sets of source and target
vertices. Our experiments confirm very significant perfor-
mance gains of our approach in comparison to state-of-the-
art graph engines such as Giraph++, and over a variety of
graph collections with up to 1.4 billion edges.
1. INTRODUCTION
1.1 Background & Motivation
The reachability problem in directed graphs is one of the
most fundamental problems in terms of both graph theory
Permission to make digital or hard copies of all or part of this work for personal or
classroom use is granted without fee provided that copies are not made or distributed
for profit or commercial advantage and that copies bear this notice and the full cita-
tion on the first page. Copyrights for components of this work owned by others than
ACM must be honored. Abstracting with credit is permitted. To copy otherwise, or re-
publish, to post on servers or to redistribute to lists, requires prior specific permission
SIGMOD’16, June 26-July 01, 2016, San Francisco, CA, USA
c
2016 ACM. ISBN 978-1-4503-3531-7/16/06. . . $15.00
DOI: http://dx.doi.org/10.1145/2882903.2915226
and applications: for a directed graph G= (V, E)with ver-
tices Vand edges E, given a source vertex s∈Vand a
target vertex t∈V, determine whether there is a consecu-
tive path of edges from sto tover E.
To avoid redundant computations, many graph applica-
tions in fact require an extension of this basic reachability
problem, where entire sets S,Tof source and target vertices,
respectively, need to be processed “at once”. The resulting
reachability problem, which we then coin set reachability,
aims to retrieve all pairs of source and target vertices (s, t),
with s∈Sand t∈T, where sis reachable to t. Moreover,
in case the data graph is partitioned into multiple, edge-
and vertex-disjoint subgraphs (e.g., when distributed across
multiple compute nodes in a cluster), we refer to the result-
ing set-reachability problem as distributed set reachability
(or “DSR” for short). The key goal in processing a DSR
query over a partitioned data graph both efficiently and in
a scalable manner is (1) to avoid redundant computations
within the local compute nodes as much as possible, (2) to
partially evaluate the local components of a set-reachability
query S;Tamong all compute nodes in parallel, and (3) to
minimize both the size and number of messages exchanged
among the compute nodes.
State-of-the-art indexing techniques for reachability que-
ries [12, 16, 18, 25, 28, 32, 33, 34, 36] are largely limited to
a centralized setting and thus address only point (1) of the
above objectives. As for (2) and (3), we are currently aware
of only one approach that specifically tackles the problem
of distributed reachability queries for single-source, single-
target queries [9]. For multi-source, multi-target (i.e., actual
set-) reachability queries, we are aware of just two central-
ized approaches [12, 30] that provide suitable indexing and
processing strategies. Both are based on a notion of equiv-
alence sets among graph vertices which effectively resolves
to a preprocessing and indexing step of the data graph to
predetermine these sets. Efficient centralized approaches,
however, are naturally limited to the main memory of a sin-
gle machine and usually do not consider a parallel—in this
case multi-threaded—execution of a reachability query.
Distributed graph engines, such as Google’s Pregel [22],
Berkeley’s GraphX [35] (based on Spark [37]), Apache Gi-
raph [1] and IBM’s very recent Giraph++ [31], on the other
hand, allow for the scalable processing of graph algorithms
over massive, distributed data graphs. All of these pro-
vide generic API’s for implementing various kinds of al-
gorithms, including multi-source, multi-target reachability
queries. However, a principal assumption we follow in this
work is that set-reachability queries are selective. That is,
for any given sets S,Tof source and target vertices, both
Sand Tare usually much smaller than V, while the set
of reachable pairs in turn usually is much smaller than the
cross-product S×T. Just like in relational approaches, an ef-
ficient processing of set-reachability queries thus calls for the
aforementioned usage of indexing strategies that take advan-
tage of the salient properties of the data graph. Graph in-
dexing and the processing of selective queries however breaks
the node-centric computing paradigm of Pregel and Giraph,
where major amounts of the graph are successively shuffled
through the network in each of the underlying MapReduce
iterations (the so-called “supersteps”).
Giraph++ is a very recent approach to overcome this overly
myopic form of node-centric computing, which led to a new
type of distributed graph processing that is coined graph-
centric computing in [31]. By exposing intra-node state in-
formation as well as the inter-node partition structure to the
local compute nodes, Giraph++ is a great step towards mak-
ing these graph computations more context-aware. How-
ever, index structures that specifically tackle the iterative
communication rounds required for the supersteps are diffi-
cult to accomplish even here, such that a direct implemen-
tation of a reachability query may still result in as many
iterations (and hence communication rounds) as the diame-
ter of the graph in the worst case.
Finally, the set-reachability problem has a plethora of
applications in graph analytics and query-processing tasks.
With its recent update, SPARQL 1.1, for example, under-
went a major revision in which the usage of labeled prop-
erty paths [2] allows a user to formulate transitive reachabil-
ity constraints among the query variables. Since both the
source and target variables of a property path may become
bound to multiple RDF constants at query processing time,
the processing of property paths in SPARQL 1.1 resolves
to processing set-reachability queries. Another interesting
application of set-reachability is community analysis in so-
cial networks. That is, given sets of source and target ver-
tices, each representing social-net users such as on Twitter
or Facebook, we may want to efficiently detect which com-
munities are densely connected. For example, consider two
communities—billionaires and non-profit organizations—, it
would be interesting to find the list of billionaires who are
also involved in philanthropic activities.
1.2 Contributions
We summarize the contributions of our work as follows.
•We formalize the problem of distributed set reachabili-
ty (DSR) over a partitioned, and hence distributed, di-
rected data graph. To our knowledge, our approach is
the first to specifically tackle this problem.
•We develop a graph-based index structure that allows
us to strictly restrict the communication protocol among
the compute nodes to a single round of message exchange
in order to resolve the results of any DSR query posed
against a given partitioning of the data graph. This
guarantee holds regardless of the properties of the data
graph (such as its diameter and partition structure) and
the properties of the query (such as the distribution of
source and target vertices among the graph partitions).
•Our indexing strategy allows for incremental updates of
the underlying data graph, with an efficient support for
vertex and edge insertions and a principle support for
respective deletions.
•Our approach is also extensible in the sense that any ex-
isting, centralized reachability index can be “plugged-in”
at the local compute nodes. We report the results of our
distributed approach in combination with a plain DFS
search [6], the MS-BFS approach of [30], and FERRARI
[28] as local search strategies.
•Moreover, we provide an extensive experimental evalu-
ation of our approach over a variety of both small and
large graphs and in comparison to different extensions of
Giraph++. We also investigate two application scenarios
of our approach for processing SPARQL 1.1 queries with
property paths and for detecting dependencies among
social-network communities.
2. PRELIMINARIES
We start by formally defining our data and query model.
This section also serves to establish the notation we will use
through the rest of the paper.
Definition 1. Adata graph G(V, E, L, φ)is a directed
graph consisting of vertices Vand edges E⊆V×V. Further,
we assume a unique label (i.e., an identifier) for each vertex
in the graph to be given by a bijective mapping φ:V→L
from vertices Vto labels L.
Given two vertices s, t ∈V, a path Pfrom sto tis denoted
by a consecutive set of edges {(s:= u0, u1),(u1, u2),...,
(un−1, un=: t)}such that each (ui, uj)∈E. Two vertices s
and tare called reachable, denoted as s;t, iff there exists
at least one path P⊆Efrom sto t.
Graph Partitioning. To scale a reachability query s;t
to a very large data graph, which may not necessarily fit
into the main-memory of a single compute node, we allow
the graph to be partitioned into kedge- and vertex-disjoint
subgraphs. Formally, a subgraph Gi(Vi, Ei, L, φ)of a data
graph G(V, E, L, φ)is a vertex-induced subgraph of G, such
that Vi⊆Vand Ei={(u, v)|u∈Vi, v ∈Viand (u, v)∈
E}. We refer to G={G1, G2,...,Gk}as the partitioning of
Gand to C(VC, EC, L, φ)as the cut of G, respectively. Here,
kdenotes the number of partitions of Gand each partition
Gi(including C) denotes a subgraph of G. Moreover, Cis
a subgraph of Gsuch that, for a given graph partitioning
G,EC={(u, v)|(u, v)∈E, u ∈Vi, v ∈Vjand i6=j}with
vertices u, v ∈VC, iff edge (u, v)∈EC.
Definition 2. Given a partitioning G={G1, G2,...,Gk}
and the respective cut C(VC, EC, L, φ)of a data graph G, a
DSR query, denoted as S;T, returns all pairs of source
and target vertices (s, t), with s∈Sand t∈T, where s;t.
Definition 3. For a given graph partitioning Gand im-
plied cut Cof a data graph G, we define the set of in-
boundaries Iifor partition Gias Ii={v|v∈Vi,∃(u, v)∈
EC, u ∈Vjand i6=j}, i.e., as the set of vertices in Githat
have an incoming edge from the cut C.
Conversely, we define the set of out-boundaries Oi=
{v|v∈Vi,∃(v, u)∈EC, u ∈Vjand i6=j}as the set of
vertices in Githat have an outgoing edge into the cut C.
Partitioning Function. We denote by ρ:V7→ N+
0the
partitioning function that determines to which of the com-
pute nodes (i.e., “slaves”) in a cluster architecture each graph
vertex v∈Vis distributed. Without loss of generality, and
to simplify notation for the following presentation, we as-
sume a simple partitioning strategy by distributing every
de
b
f
r
ac
gi
l
k
h
u
n
m
p
q
v
o
Slave 1
G1
Slave 2
G2
Slave 3
G3
(a) Graph G
f
b
e
c
g
hin
m
o
G1G2G3
(b) Cut C
Figure 1: (a) Graph Gwith partitions G={G1, G2, G3}and (b) respective cut C
vertex v∈Vito slave i(i.e., ρ(v) = ifor each v∈Vi). We
will thus refer to a graph partition and a slave interchange-
ably. To increase concurrent executions (e.g., when using
multi-threading at the local compute nodes), an “overpar-
titioning” strategy may be employed instead, by assigning
multiple graph partitions to each of the slaves.
Example 1. Figure 1(a) shows an example graph Gwith
three partitions G={G1, G2, G3}which are stored at three
slaves. Its corresponding cut Cis shown in Figure 1(b). In-
and out-boundaries are I1={f}, O1={b, e},I2={c, g, h},
O2={i}, and I3={m, n},O3={o}, respectively.
Distributed Reachability. Our approach for processing
DSR queries uses a similar setting as described by Fan et
al. [9]. In [9], a distributed (single-source, single-target)
reachability query s;tover a master-slave architecture is
evaluated as follows. The master receives s;tand commu-
nicates it to all slaves. At partition Gi, containing the source
s, a local evaluation of the reachability of sto each vertex
in the set of out-boundaries Oiis computed first. Similarly,
at partition Gjcontaining the target t, a local evaluation of
the reachability of each vertex in the set of in-boundaries Ij
is computed next. Additionally, a local computation of the
reachability between all in-boundaries Iiand out-boundaries
Oiis computed at each partition G1,...,Gkand hence at
all slaves i= 1..k in the compute cluster in parallel.1
The resulting local reachability information is encoded
into sets of Boolean formulas, where each such set repre-
sents the local connectivities between the in-boundaries Ii
(including the source s, if present) with the out-boundaries
Oi(including the target t, if present) at partition Gi. All
of these formulas are communicated back to a single master
node for the final evaluation. A query-specific global depen-
dency graph is constructed at this master node for s;t
using the Boolean formulas and the static cut C. A reach-
ability algorithm is then run over the dependency graph to
answer s;tvia substitution of the variables that are re-
versely connected to the target vertex t. The overall al-
gorithm provided in [9] can be implemented with a simple
communication protocol and, for example, be executed in a
single MapReduce iteration among all compute nodes.
Example 2. Consider the distributed reachability query
d;qover the graph partitioning shown in Figure 1. The
local evaluation at each partition results in the following
Boolean representation of partial reachability information:
{d=b∨e, f =b∨e}@G1,{c=i, g =i, h =i}@G2,
{m=q∨o, n =q∨o}@G3. By including the edges in the cut
C(Figure 1(b)), the global dependency graph (Figure 2(a))
is constructed at the master node to finally resolve d;q.
By running a reachability algorithm (such as backward DFS)
1Note that we follow a slightly different definition of in- and
out-boundaries than in [9]. However, the algorithm in [9]
directly translates to the one outlined above.
f
d
b
e
g
c
h
in
m
o
q
(a) Single Reachability
Master
f
d
a
b
e
g
c
h
ii
l
n
m
o
p
(b) Set Reachability
Master
Figure 2: Dependency graph as constructed in [9]
for a single reachability query (a) and an extension
to set reachability (b)
over the dependency graph, one can find that d;qis indeed
true (the red path in Figure 2(a)).
3. DISTRIBUTED SET REACHABILITY
3.1 Naïve Approach
Anaïve approach to extend the distributed reachability
problem to sets of vertices S,Twould be to simply invoke a
separate reachability query s;tfor every pair (s, t), with
s∈Sand t∈T. However, an obvious reason for the limited
efficiency of this approach, even for reasonably-sized sets S
and T, is that this approach does not exploit the assumption
we made earlier, namely that queries are selective, i.e., that
by far not all pairs in S×Tare reachable. Consequently, this
approach can also not reuse any intermediate computations
and thus likely performs many redundant computations.
3.2 DSR with a Dynamic Dependency Graph
An improved approach to extend the distributed reachabil-
ity algorithm provided in [9] to sets is as follows. Let S;T
be the query received at the master. First, we partition
S;Tinto subqueries S1;T1,S2;T2,. . . ,Sk;Tk,
where kis the number of graph partitions, such that each
Si⊆Viand Ti⊆Vicontains only vertices that are local to
partition Gi. Next, a local evaluation at each slave iinvolves
finding the reachability among all pairs of vertices from the
sets Si∪Iiand Oi∪Ti, respectively. These can again be
run in parallel across all slaves. The resulting reachability
information, again represented as sets of Boolean formulas,
is then communicated from all slaves to the master node for
the final evaluation. At the master node, the query-specific
dependency graph for the sets S,Tis constructed as de-
scribed in Section 2, and a local reachability algorithm is
then used at the master node to emit all reachable pairs
(s, t), with s∈Sand t∈T.
Example 3. Consider the DSR query S;Twith S=
{a, d, g}and T={l, p}over the cut Cshown in Figure 1(b).
The sets of Boolean formulas obtained after the local evalu-
ation at each slave are as follows: {a=b∨e, d =b∨e, f =
b∨e}@G1,{c=i, g =i∨l, h =i}@G2,{m=p∨o, n =
p∨o}@G3. At the master node, after evaluating S;T
over the global dependency graph shown in Figure 2(b), we
obtain the following reachable pairs of source and target ver-
tices: {(a, l),(a, p),(d, l),(d, p),(g, l),(g, p)}.
3.2.1 Discussion
Although the second algorithm provides a more viable so-
lution of the DSR problem than the naïve approach, it still
leaves a number of disadvantages that limit both its effi-
ciency and scalability.
•First, the query-dependent, global dependency graph is
generated “from scratch” for each query S;T, al-
though both the cut Cand the local reachability infor-
mation Ii;Oiamong the in- and out-boundaries at
each graph partition Giare in fact static.
•Second, the approach does not leverage any distributed
computation in its second step, as the final reachability
computation S;Tover the global dependency graph
is performed only by a single master node.
•Third, since the global dependency graph is generated
dynamically for each query S;T, a reachability index
for the static cut Cand the local Ii;Oicomponents
cannot be constructed, which restricts the final reacha-
bility computation to either a simple BFS or DFS strat-
egy over the dependency graph.
3.3 DSR with a Static Reachability Index
In our approach, instead of computing the global depen-
dency graph for each incoming query from scratch at the
master node, we precompute a partition-specific variant there-
of, called the “boundary graph”, only once and store this in
the form of a static reachability index at each slave. This
strategy provides multiple benefits. First, it avoids repeated
computations of the boundary graph for each query. Second,
since each slave has the complete reachability information
among the boundary vertices of all other slaves available,
finding the reachability of any two vertices (s, t)in the en-
tire data graph Gresolves to a local reachability computa-
tion at at most two slaves, which is irrespective of the di-
ameter of the graph and the distribution of the source and
target vertices of a set-reachability query (see Theorems 1
and 2). Additionally, an index can be built over the static
boundary graph to accelerate this processing. Third, storing
a (compacted version of the) boundary graph at each slave
allows for a fully distributed processing of a set-reachability
query and thus avoids the single-node bottleneck of previous
approaches. We next formally define how we generate the
boundary graph and its derived index structures.
3.3.1 Precomputed Index Structures
Boundary Graph. Aboundary graph is a directed graph
that represents the reachability information among the in-
and out-boundaries of all graph partitions G={G1,...,Gk}
with respect to a given cut C.
Definition 4. Let GB
i(VB
i, EB
i,L, φ)denote the boun-
dary graph we compute for partition Gi, such that the fol-
lowing holds:
•The vertices VB
i=Si=1..k Ii∪Oiconsist of the union of
all in- and out-boundaries of all partitions G1,...,Gk.
•There exists an edge (u, v)∈EB
i, iff
-(u, v)∈EC, or
-u∈Ijand v∈Oj, for j6=i, and u;v(i.e., uand
vare both located at another partition Gjand there
exists a path from uto vin Gj).
That is, the boundary graph for partition Gimerges the
static cut Cwith the static reachability information Ij;Oj
fb
e
g
c
h
in
m
o
(a) Non-optimized
Slave 1
GB
1
fb
eυ3ν2
υ2ν3υ4
ν4
−→
c
−→
h−→
m
−→
n
(b) Optimized
Slave 1
GB
1
Figure 3: Boundary graph GB
1for partition G1
among all the remaining graph partitions Gj(for i6=j) into
a new, precomputed graph GB
i. The resulting boundary
graphs are thus partition-specific.
Example 4. For our graph Gwith partitions G1,G2,G3
and respective cut Cas shown in Figure 1, the boundary
graph GB
1for partition G1is shown in Figure 3(a). Here,
the dashed edges refer to edges in the cut C, while the solid
edges denote the transitive reachability Ij;Oj(for j6= 1).
Complexity. The construction of the boundary graph re-
quires us to materialize the pairwise reachability Ii;Oi
among the in- and out-boundaries for each partition Gi. Us-
ing a simple BFS/DFS-based approach, the time complexity
of this computation is O((|Vi|+|Ei|)·|Ii|·|Oi|)per partition.
This can be further improved to O(1 · |Ii|·|Oi|)when using
a sophisticated, local reachability index for this operation.
On the other hand, the (worst-case) space complexity for
storing the boundary graph at partition iis O(Pk
j=1 |Ij| ·
|Oj|+|EC|), for j6=i. From this, one can deduce that
both the time and space complexity of the boundary graph
computation strongly depend on the amounts of in- and out-
boundaries we obtain from the cut C.
Min-k-Cut Partitioning. A standard approach to reduce
this number of boundary vertices is to reduce the number of
edges in the cut C, while trying to keep the sizes of the par-
titions G1,...,Gkbalanced. Although finding an optimal
such min-k-cut partitioning is a well-known NP-complete
problem [6], current graph libraries such as METIS [17]
are capable of achieving very good approximations even for
graphs with hundreds of millions of edges.
Equivalence Sets. Even for a given cut C, we can further
reduce the size of the boundary graph by grouping the in-
and out-boundary vertices into equivalence sets of vertices,
thus continuing the idea presented in [12] to a distributed
setting. Specifically, we achieve this by grouping the bound-
ary vertices into forward- and backward-equivalent sets ac-
cording to the following definition.
Definition 5. Two in-boundaries b1, b2are called for-
ward-equivalent with respect to subgraph Gi, i.e., b1≡fb2,
iff for any vertex v∈Vi−Iiand b1;v, it holds that b2;v.
Conversely, two out-boundaries b1, b2are called backward-
equivalent with respect to subgraph Gi, i.e., b1≡bb2, iff for
any vertex v∈Vi−Oiand v;b1, it holds that v;b2.
That is, once the forward- and backward-equivalent sets
of vertices are identified for each subgraph Gi, each such
set is replaced by a new in-virtual vertex υ(for a forward-
equivalent set) and a new out-virtual vertex ν(for a backward-
equivalent set), respectively.
Example 5. Following the above definition of equivalence
for the partitioning G={G1,...,G3}of Gshown in Fig-
ure 1(a), we can obtain the following, partition-specific equiv-
alence sets: {υ1={f}}@G1,{ν1={b, e}}@G1;{υ2=
de
b
f
r
a
υ3
υ2
ν2
ν3
υ4
ν4
−→
c
−→
h
−→
n
−→
m
Slave 1
GC
1
c
gi
l
k
h
u
ν1
υ1
υ4
ν4
←−
b
←−
e
←−
e
−→
n
−→
m
Slave 2
GC
2
n
m
p
q
v
o
ν3
υ3
υ1
ν1υ2ν2
−→
c, h
←−
b, e
Slave 3
GC
3
Figure 4: Final compound graphs GC
1,GC
2,GC
3constructed for graph Gwith cut Cof Figure 1
{c, h}, υ3={g}}@G2,{ν2={g}, ν3={i}}@G2;{υ4=
{m, n}}@G3,{ν4={o}}@G3.
Next, the in- and out-boundaries are redefined with re-
spect to the new virtual vertices. That is, Iicomprises of
all in-virtual vertices and Oicomprises of all out-virtual ver-
tices. The optimized boundary graph for partition G1is
shown in Figure 3(b). Note that we attach additional labels
to the cross-edges in the boundary graph to obtain a loss-
less representation of the boundary graph with respect to
the partitions G1,...,Gk. For example, the cross-edge (b, c)
is represented by connecting the vertex band the in-virtual
vertex υ2with the label −→
cto denote that bis connected to
only cin υ2. The forward arrow denotes that this connec-
tion is valid only for a forward exploration. This is required,
since vertices c,hare forward-equivalent, i.e., c≡fh, with
respect to partition G2only.
Computing Equivalence Sets. According to Definition 5,
two in-boundaries b1∈Iiand b2∈Iiare forward-equivalent
if they are reachable to exactly the same set of vertices in
Vi−Ii. To determine the sets of forward-equivalent bound-
aries, we need to (1) compute all reachable pairs from Ii
to Vi−Iiand then (2) group the vertices in Iiinto these
equivalence sets. For large sets Iiand Vi−Ii, this compu-
tation may be prohibitively expensive. To address (1) and
thus reduce the input that needs to be considered for (2),
we apply the following optimizations.
•b1,b2can only be forward-equivalent with respect to par-
tition Giif both belong to the same strongly connected
component (SCC) in Gi. We thus condense Giinto a
more compact DAG by computing the SCCs over Gi.
•Instead of considering all target vertices Vi−Ii, we con-
sider only the direct successors S(Ii)of Ii, and hence
S(Ii)−Ii, to check for forward-equivalence. The intu-
ition for considering only successors is that if two bound-
aries b1,b2are reachable to the same set of vertices in
S(Ii)−Ii, then b1,b2also are reachable to the same set
of vertices in Vi−Ii.
A similar construction then holds also for backward-equiva-
lence, except that predecessors P(Oi)are considered instead.
Example 6. Consider partition G3with in-boundary set
I3={m, n}to compute the sets of forward-equivalent ver-
tices in I3. In this case, this requires us to only verify
whether m≡fn, since m,nare the only in-boundaries in
I3. First, we run the SCC algorithm to condense G3into
the DAG G0
3. In this example, G0
3=G3, and we see that
m,ndo not belong to the same SCC. We then check their
forward-equivalence based on the sets of vertices in V3−I3
that are reachable from both mand n. To compute these
reachable sets of vertices, we consider only the direct suc-
cessors S(I3)−I3={p, v}instead of considering all of
V3−I3={p, o, q, v}. Thus, the reachable set of vertices
of both mand nis {p, v}, and hence we have m≡fn.
A compact algorithm for computing these equivalence sets
is depicted in Appendix 8.1.
Compound Graph. After compacting the partition-specific
boundary graphs GB
iby replacing both the forward- and
backward-equivalent sets of vertices with their in- and out-
virtual counterparts, we perform one more step to obtain our
final graph index for evaluating DSR queries. To do so, we
merge the partition-specific boundary graphs with the local
partitions into a compound graph GC
ifor each partition Gi.
These compound graphs will facilitate the processing of DSR
queries via a combination of local reachability computations
and a single filtering step among these local results.
Definition 6. Let GC
i(VC
i, EC
i,L, φ)denote the com-
pound graph we compute for partition Gi, such that the
following holds:
•The vertices VC
i=Vi∪VB
iconsist of the union of ver-
tices in the local subgraph Giand boundary graph GB
i.
•The edges EB
i=Ei∪EB
iconsist of the union of edges
in the local subgraph Giand boundary graph GB
i.
Figure 4 shows the compound graphs for the initial data
graph Gfrom Figure 1(a).
Forward- and Backward-Lists. Our last precomputa-
tion step consists of storing the forward- and backward-lists,
Fiand Bi, of boundaries which are non-local to each par-
tition Gi. These will serve for routing messages to only
those partitions Gjwhich are connected to Gi. Specifically,
the forward-list Fi=Sj6=i{υ|υis in-virtual vertex of Gj}
is the set of all vertices that are non-local to Giand are
in-virtual vertices of another partition Gj. Similarly, the
backward-list Bi=Sj6=i{ν|νis out-virtual vertex of Gj}
consists of all out-virtual vertices that are non-local to Gi.
For instance, for partition G1shown in Figure 4, we have
F1={υ2, υ3, υ4}and B1={ν2, ν3, ν4}.
3.3.2 Evaluating DSR Queries
Given these precomputed index structures, i.e., the com-
pound graphs GC
iand respective forward- and backward-
lists, Fiand Bi, evaluating a DSR query now becomes straight-
forward. We again begin with a discussion of the single-
source, single-target case and then explain how it generalizes
to the multi-source, multi-target case.
A. Single Reachability. Consider the reachability query
s;t. The algorithm for processing the query is shown in
Algorithm 1. Given a data graph Gwith partitioning G,
we evaluate the query as follows. If both sand tbelong to
same partition Gi, then the reachability s;tis confined
to only slave iwhich stores the compound graph GC
i. Since
the compound graph GC
iaugments each Giwith the global
reachability information among all boundary vertices, we
can safely evaluate the reachability of s;ton GC
iby call-
ing any centralized reachability algorithm via the function
localSetReachability(.) (Lines 11-13). A formal justification
for this is provided by the following theorem.
Theorem 1. Let s,tboth be local vertices of partition
i, i.e., s, t ∈Vi. Then the evaluation of the reachability
Algorithm 1: Distributed Reachability Processing
Input: Compound graphs: {GC
1, GC
2, ..., GC
k}, Query: s;t
Output: true/false
1Master:
2ranks:= ρ(s).i.e., s∈Viand Giis at Slave ρ(i)
3rankt:= ρ(t).i.e., t∈Vjand Gjis at Slave ρ(j)
4result := false
5foreach rank do
6result := result ∨compute(s, ranks, t, rankt)
.invokes parallel computations at all ranks
7return result
8Slave i:
9method compute(s,ranks,t,rankt):
10 rset := ∅
11 if ranks=iand rankt=ithen
.invoke local reachability evaluation
12 if localSetReachability({s},{t})6=∅then
13 return true
14 else if i=ranksthen
15 j:= rankt
16 Υj
s:= localSetReachability({s},Fj
i);
. F j
i⊆Fiis the set of in-virtual vertices local to j
17 rset[s] := Υj
s
18 sendMessage(j,rset)
19 return false
20 else if i=ranktthen
21 receiveMessage(i,rset)
22 Υi
s:= rset[s]
23 for υin Υi
sdo
24 b:= υ.rep . b is a member vertex in eqset υ
25 if localSetReachability({b},{t})6=∅then
26 return true
27 return false
s;tover graph Gcan be answered entirely locally over the
compound graph GC
iwithout requiring any message exchange
among the compute nodes.
Proof. See Appendix 8.2.1.
Example 7. Consider the query b;f. Both vertices b,
fare local to partition G1. By considering only the subgraph
G1, one cannot find that fis reachable from b. But by con-
sidering the whole graph G, we see that b;fis true via the
path b→c→i→n→p→o→f. However, using the local
compound graph GC
1(see Figure 4), we can indeed find that
b;fis true via the path b→υ2→ν3→υ4→ν4→f.
If, on the other hand, sand tare located at two dif-
ferent partitions Gi,Gj, with i6=j, the evaluation of a
reachability query works as follows (Lines 14-25). Starting
at partition Gi, we find the reachability from sto all the
forward-boundaries υ∈Fj
i⊆Fi(Line 15) which are lo-
cated at another partition Gj. Let Υj
s⊆Fibe the set of
in-virtual vertices located at partition Gj(and hence stored
by slave jas per our assumption) which are reachable from
s. The message rset[s]:=hs, Υj
siis then communicated to
slave j. At slave j, we consider each υ∈Υj
sand replace
it with any one of its members b, after which we evaluate
the reachability from bto the local target vertex t. If there
exists one such b∈υ∈Υj
swith b;t, we report that s;t
is true (Lines 22-25).
Theorem 2. Let s∈Viand t∈Vj, with i6=j. Then,
the evaluation of the reachability s;tcan be answered over
the two compound graphs GC
iand GC
jby using a single step
of message exchange from slave ito slave j.
Proof. See Appendix 8.2.2
Example 8. Consider the query a;q, where ais located
at partition G1and qis located at partition G3. At partition
G1, we compute the reachability from ato the single forward-
boundary {υ4}which is located at G3. From the compound
graph GC
1(shown in Figure 4), we have Υ3
a={υ4}since
a;υ4.Υ3
ais then communicated to slave 3. At slave 3, we
expand the actual vertices represented by the virtual vertex
υ4(say m, since m∈υ4) and find the reachability from m
to q. Since m;q, we thus find that a;qis true.
B. Set Reachability. An actual DSR query S;T, which
is received by the master node, is processed in our approach
as shown in Algorithm 2. First, S;Tis partitioned into
subqueries S1;T1,S2;T2,. . . ,Sk;Tk, where kis
again the number of graph partitions. The partitioning of
the query into these subqueries is determined such that each
source vertex si∈Siand target vertex ti∈Tiresides locally
at partition Gi(Line 2).
Step 1. (Lines 13-19) A local evaluation at partition Gi
involves processing the pairwise reachability among the ver-
tices from Sito Tiand from Sito Fiat all slaves i= 1..k
in parallel. This operation generates two types of reachable
pairs: (si, ti)and (si, υj). The first type denotes the reacha-
bility between both a local source si∈Siand a local target
ti∈Ti. The second type denotes the reachability between a
local source si∈Siand a forward-boundary υj∈Fi, which
is represented by an in-virtual vertex located at slave j.
Step 2. (Lines 21-32) The communication of the remotely
reachable pairs, each of the form (si, υj), is performed from
slave ito slave jamong all pairs of slaves i, j = 1..k in
parallel. In order to reduce the overhead of communicating
individual pairs, each slave buffers its partial reachability
information and communicates this buffer at once. Each
buffer sent from slave ito slave jis of the form {hsi,Υj
sii}
for all si∈Si. For easier processing, the messages received
at slave ifrom all other slaves are stored in an inverted
index Ii(Υi
∗, Li), where Υi
∗is the aggregated set of in-virtual
vertices (local to slave i). For each in-virtual vertex υ∈Υi
∗,
its aggregated non-local source set Sυ⊆Sis stored in Li.
That is, for s∈Sυand υ∈Υi
∗, we already know that s;υ.
Step 3. (Lines 34-39) A final local evaluation involves
processing the set reachability Υi
∗;Tifrom the in-virtual
vertices Υi
∗to the target sets Tiat all slaves i= 1..k in
parallel. For each in-virtual vertex υ∈Υi
∗and original
vertex brepresented by υ, we evaluate the reachability from
bto all targets t∈Ti. If b;tis true, then for each s∈Sυ,
we report that s;tis true.
Example 9. Consider again the graph Gwith partitions
G1,G2,G3in Figure 1(a). The respective compound graphs
GC
1,GC
2,GC
3are shown in Figure 4. Let S={d, l, p};T=
{a, k, q}be the DSR query received at the master node. The
query is partitioned into {d};{a},{l};{k},{p};{q}.
At partition G1, we find the set-reachability (Step 1) be-
tween {d},{υ2, υ3, υ4, a}, thus returning the reachable pairs
{(d, υ2),(d, υ3),(d, υ4),(d, a)}. We perform the same op-
eration in parallel at slaves 2and 3and communicate the
results to all other slaves (Step 2). At slave 1, we receive
Algorithm 2: Distributed Set-Reachability Processing
Input: Compound graphs {GC
1, GC
2, ..., GC
k}, Query: S;T
Output: R {(s, t)|s∈S, t ∈Tand s;t}
1Master:
2partition S, T into {(S1, T2),(S2, T2),...,(Sk, Tk)}
.where Si⊆Viand Ti⊆Vi
3result := ∅
4for i= 1 ...k do
5result := result ∪compute(Si,Ti)
6return result
7Slave i:
8method compute(Si, Ti):
9local_rset := ∅
10 remote_rset := ∅
11 result := ∅
12 // Step 1:
13 local_rset := localSetReachability(Si,Ti)
14 remote_rset := localSetReachability(Si,Fi)
15 for s in Sido
16 for t in local_rset[s]do
17 result := result ∪ {(s, t)}
18 for υ in remote_rset[s]do
19 Υj
s:= Υj
s∪υ
. υ is an in-virtual vertex of partition j
20 // Step 2:
21 for j= 1 to kdo
22 if j6=ithen
23 msg := ∅
24 for s in Sido
25 msg := msg ∪ {hs, Υj
si}
26 sendMessage(j, msg)
27 Ii(Υi
∗, Li) = ∅
28 for j= 1 to kdo
29 receiveMessage(j,msg)
30 for hs, Υi
siin msg do
31 for υ in Υi
sdo
32 Ii[υ] := Ii[υ]∪ {s}
33 // Step 3:
34 for υ in Υi
∗do
35 b:= υ.rep
36 local_rset := localSetReachability({b},Ti)
37 for s in Ii[υ]do
38 for t∈local_rset[b]do
39 result := result ∪ {(s, t)}
40 return result
the following reachability information: {(υ1,[l, p])}. Simi-
larly, at slave 2, we receive {(υ2,[d, p]),(υ3,[d, p])}; and at
slave 3, we receive {(υ4,[d, l])}. At the end of the local evalu-
ation from boundaries to the final targets (Step 3), by replac-
ing virtual vertices with each of their represented vertices (at
slave 1,υ1is replaced with f), the sets {(d, a),(l, a),(p, a)}@
G1,{(d, k),(l, k),(p, k)}@G2and {(d, q),(l, q),(p, q)}@G3of
reachable pairs are generated at the partitions.
Local Reachability Evaluation. Algorithms 1 and 2
both require partial reachability processing at each slave
via the function localSetReachability(.). For this, any cen-
tralized reachability index (see, e.g., [6, 28, 12, 36]) can be
plugged into our framework. We abstract this by calling
localSetReachability(.) in our algorithms whenever a local
(set-)reachability operation is invoked.
Forward vs. Backward Processing. Our above discus-
sion focused on starting from the source vertices and ending
at the target vertices. If there are less targets than sources,
one may also start from the target vertices and search back-
wards to the source vertices to arrive at the same results.
We therefore maintain both forward- and backward-lists, Fi
and Bi, to facilitate these two directions of searching.
3.3.3 Incremental Updates
Insertions. Insertions over the SCC-condensed compound
graphs GC
ican be implemented without storing the original
(i.e., uncondensed) compound graphs.
Let (u, v)denote a new edge that is to be inserted into
the graph G. First, assume both uand vbelong to the
same graph partition i. Further, if u,vbelong to the same
SCC, then adding (u, v)to Giwould not change the local
compound graph GC
i(nor any other) at all and thus can be
safely ignored. If, on the other hand, u,vbelong to two
different SCCs, then a series of update actions are required.
First, we add the new edge to the local compound graph GC
i
and locally recompute the SCCs and equivalence sets. Next,
new connections among the local in- and out-boundaries, Ii
and Oi, are communicated to all other partitions j(for j6=i)
as additional edges. These can be incrementally merged into
all the compound graphs GC
jby updating their SCCs as
well. Second, if uand vbelong to two different partitions
iand j, then this means we have a new edge in the cut
C, which however does not affect the reachability within
partitions iand j. Thus, (u, v)can directly be merged into
the distributed compound graphs as described above.
Let n,mdenote the number of vertices and edges in the
condensed compound graph GC
i, and let |Ii|,|Oi|be the
number of in- and out-boundaries for partition i, respec-
tively. By adding a local edge to partition i, a partial or full
recomputation of the connections among vertices from Iito
Oiis required. Thus the worst-case time complexity of this
step is O((n+m)· |Ii|·|Oi|), which is asymptotically opti-
mal [7]. The SCC recomputation at each compound graph
has a time complexity of O(n0+m0), where n0and m0are
the numbers of vertices and edges in the new GC
i’s.
Deletions. Deletions over the SCC-condensed compound
graphs GC
i, on the other hand, result in a decremental main-
tenance of the SCCs, which requires either storing the orig-
inal (i.e., uncondensed) compound graphs or organizing the
SCCs in a hierarchical manner [19]. In our implementa-
tion, we resort to storing the uncondensed compound graphs
along with the condensed compound graphs GC
i, albeit ap-
proaches like [19] may be employed for further optimizations.
A deletion of a local edge (u, v)in partition iis processed
over the condensed compound graph GC
ias follows. If the
vertices u,vbelong to the same SCC, then we expand this
SCC into its original edges and reconnect these edges to
the remaining SCCs in GC
i. Moreover, in case of deletions,
some of the existing boundaries may not be connected any-
more. We identify such pairs of boundaries and communi-
cate these to the other compute nodes. After receiving this
list of deleted boundary edges, we reconstruct the local com-
pound graphs GC
j(for j6=i) analogously to the insertion
case. If, on the other hand, the vertices u,vbelong to two
different SCCs, then we expand both of them.
Here, the worst-case time complexity to maintain the lo-
cal boundary edges is O((|Vi|+|Ei|)· |Ii|·|Oi|), which is
the same as for rebuilding the local boundary graphs (see
Section 3.3.1). The new compound graphs are condensed
via SCC computation, whose worst-case time complexity is
O(n0+m0), where n0and m0again are the numbers of ver-
tices and edges in the new GC
i’s.
4. EVALUATION
We next present a detailed empirical evaluation of our
proposed indexing and updating strategies for DSR queries.
Specifically, we compare the following approaches:
•our DSR approach with a static reachability index (coined
“DSR”), as described in Section 3.3;
•a naïve enumeration of all pairs of source and target ver-
tices (coined “DSR-Naïve”), as described in Section 3.1;
•a generalization of the algorithm described by Fan et
al. (coined “DSR-Fan”) [9] to sets of source and target
vertices, as described in Section 3.2;
•an implementation of DSR queries in Apache Giraph
(coined “Giraph” ) [1], as depicted in Appendix 8.4.1;
•an implementation of DSR queries in Giraph++2(coined
“Giraph++”) [31], as depicted in Appendix 8.4.2;
•an extended version of Giraph++ with equivalence sets
(coined “Giraph++wEq” ), as depicted in Appendix 8.4.3.
Further, for “DSR”, we report the results in combination
with the following local reachability indexes:
•a plain depth-first-search (DFS) strategy which requires
no additional index structures except for those described
in Section 3.3 (coined “DSR-DFS” );
•the multi-source-breath-first-search (MS-BFS) algorithm
described by Then et al. [30] which also requires no
additional index structures except for those described in
Section 3.3 (coined “DSR-MSBFS” );
•using FERRARI [28] as a local reachability index that is
generated on top of the compound graphs described in
Section 3.3 (coined “DSR-FERRARI” ).
Unless stated otherwise, we report the combination of our
DSR index with DFS as the default local search strategy.
Datasets. The list of graph collections we consider for
our evaluation is shown in Table 1. All the smaller graphs
(including the two Live Journal versions) are obtained from
the Stanford Snap3project. For our evaluation over larger
graphs, we used the real-world Freebase4and Twitter5snap-
shots. In addition, we also used the widely popular synthetic
LUBM RDF benchmark, which we generated using the UBA
1.76data generator.
Small Large
Graphs |V| |E| Graphs |V| |E|
Amazon 0.4M 3.3M LiveJ-68M 4.8M 68.9M
BerkStan 0.7M 7.6M Twitter-1.4B 41.7M 1,468.4M
Google 0.9M 5.1M Freebase-500M 97.3M 499.9M
NotreDame 0.3M 1.5M Freebase-1B 156.6M 999.9M
Stanford 0.3M 2.3M LUBM-500M 115.6M 500.0M
LiveJ-20M 2.5M 20.0M LUBM-1B 222.2M 961.4M
Table 1: Graph datasets and sizes
General Setup. We implemented the DSR variants in
C++ using GCC-4.7.2 with Boost-1.55 and compiled with
-O3 optimization. We used MPICH2-1.4.1 for communica-
tion among the compute nodes, using a cluster of 10 nodes
which are connected via a 10GBit LAN. Each node has
with HT enabled. Giraph and its variants are implemented
in Java, where we used Hadoop v0.20 for running Giraph [1].
Appendix 8.4 depicts our actual implementation of DSR
queries for the three Giraph variants.
2https://issues.apache.org/jira/browse/GIRAPH-818
3http://snap.stanford.edu
4http://freebase.com
5http://an.kaist.ac.kr/traces/WWW2010.html
6http://swat.cse.lehigh.edu/projects/lubm/
DSR DSR-Fan DSR-Naïve
Compound graph Dep.graph Dep.graph
Original DAG Size
Graphs (#edges) (#edges) (MB) (#edges) (#edges)
Amazon 1.0M 34.7K 206 622.3M 622.2M
BerkStan 2.1M 0.5M 383 2.2M 2.1M
Google 1.2M 0.2M 302 43.6M 43.6M
NotreDame 0.8M 68.9K 123 4.7M 4.7M
Stanford 0.8M 41.2K 122 1.2M 1.2M
LiveJ-20M 13.7M 1.0M 1,553 861.4M n/a
LiveJ-68M 44.1M 0.3M 928 n/a n/a
Freebase-1B 460.4M 241.6M 64,141 n/a n/a
Twitter-1.4B 1,285.0M 8.2M 20,053 n/a n/a
LUBM-1B 891.8M 891.3M 107,608 n/a n/a
Table 2: Index sizes for DSR variants
4.1 Efficiency
For this experiment, we considered several real-world data
graphs (both small and large) and the synthetic LUBM
graph. We fixed the compute cluster to 6 nodes (i.e., to
5 slaves and 1 master). We randomly selected 10 source and
10 target vertices from all datasets (except LUBM-1B) as
queries, thus resulting in 100 reachability comparisons. For
LUBM-1B, which is very sparsely connected, we randomly
chose 1,000 sources and 1,000 targets, of which only 131
pairs turned out to be reachable.
Table 2 shows the maximum (i.e., per node) uncondensed
(“Original”) and SCC-condensed (“DAG”) compound-graph
sizes as well as the total byte size (“Size”) for our DSR in-
dex in comparison to the dependency-graph sizes for DSR-
Fan and DSR-Naïve. In DSR-Fan, for a given DSR query
S;T, all of Sand Tare used “at once” to generate the
dependency graph. In DSR-Naïve, which generates the de-
pendency graph per (s, t)pair, the sizes represented in Ta-
ble 2 are the average dependency-graph sizes over 100 pairs.
SCC compression, which is not feasible for the dynamically
generated dependency graph, drastically reduces the sizes
of the compound graphs stored at each slave. For example,
for the Twitter-1.4B graph, which is highly connected, the
size of each compound graph stored at the slaves initially is
comparable to the size of the original graph. Applying SCC
compression condenses these graphs by a factor of about 150.
Also for LiveJ-68M, the SCC compression leads to a much
smaller DAG size than for LiveJ-20M, such that our query
times are actually lower for LiveJ-68M than for LiveJ-20M.
Table 3 shows our query-processing results. For both the
small and large graphs, our approach clearly demonstrates
efficiency improvements of several orders of magnitude when
compared to the three Giraph variants as well as to DSR-
Fan and DSR-Naïve. Even with a single round of commu-
nication, DSR-Fan and DSR-Naïve exhibit a considerable
overhead in generating the dynamic dependency graph for
each query. Specifically, we observed that for LiveJ-20M,
DSR-Fan generates a dependency graph of about 861 million
edges even when the data graph is partitioned by METIS
[17] in order to minimize the cut. Our DSR approach, which
benefits from the optimizations we apply when constructing
the compound graphs, avoids the repeated generation of a
large dependency graph at the master node and therefore is
able to achieve very significant performance gains over DSR-
Fan and DSR-Naïve. Giraph++ and Giraph++wEq, on the
other hand, perform better than the native Giraph imple-
mentation, as the former benefit from their local updates of
neighboring vertices. This drastically reduced the number of
supersteps required for processing a set-reachability query.
The equivalence-sets optimization for Giraph++wEq further
reduced the communication but only marginally improved
the query processing times.
Indexing Time Query Size Query Time
Graphs DSR |S| |T| DSR Giraph++ Giraph++wEq Giraph DSR-Fan DSR-Naïve
(a) Small Graphs (times in seconds)
Amazon 2.380 10 10 0.008 12.250 11.348 55.034 72.111 855.159
BerkStan 3.048 10 10 0.009 44.180 5.680 779.006 2.219 38.036
Google 3.194 10 10 0.060 60.154 11.426 53.614 25.210 114.078
NotreDame 1.089 10 10 0.057 11.085 12.320 94.787 1.800 50.598
Stanford 1.511 10 10 0.008 7.808 8.922 341.976 0.468 6.211
LiveJ-20M 44.536 10 10 0.227 19.888 19.262 28.075 521.569 n/a
(b) Large Graphs (times in seconds)
LiveJ-68M 144.981 10 10 0.090 64.728 61.940 93.253 n/a n/a
Freebase-1B 1,938.670 10 10 67.849 1,371.423 1,014.442 1,857.124 n/a n/a
Twitter-1.4B 6,963.730 10 10 1.119 3,065.483 3,046.450 n/a n/a n/a
LUBM-1B 2,083.190 1,000 1,000 1.340 146.864 142.142 154.407 n/a n/a
Table 3: Efficiency evaluation (indexing and query times) of DSR approaches for small and large graphs
4.2 Scalability
Next, we evaluated our approach in comparison to the
Giraph variants under both strong and weak scaling. We
dropped DSR-Fan and DSR-Naïve from these comparisons,
as they could not scale to the larger graphs anymore. We
considered LiveJ, Freebase, Twitter and the synthetic LUBM
graph for our scalability evaluation. We used METIS to par-
tition the graphs and distributed the partitions to up to 10
compute nodes (one of which used as master), and we con-
sidered 10 random source and 10 random target vertices as
queries. Figures 5(d)(h)(l)(p) also show the robustness of
our approach with respect to larger query sets.
Live Journal: Figures 5(a)-(d) depict the scalability eval-
uation of our approach and the Giraph variants for LiveJ-
68M. Figure 5(a) shows the results for a strong scaling. Here,
we can observe that DSR scales very well and performs sig-
nificantly better than the Giraph variants. We also observe
that Giraph++ and Giraph++wEq perform slightly better
than Giraph by leveraging the node locality and equivalence-
sets optimization, respectively. This observation is con-
firmed further by Figure 5(b), where Giraph communicates
about two orders of magnitude more messages compared to
Giraph++and Giraph++wEq. Figure 5(c) shows the weak
scalability for the 10 by 10 DSR queries.
Freebase: The scalability results for Freebase-1B are shown
in Figures 5(e)-(h). We can observe that our approach scales
well on average, even when the graph sometimes is rather un-
evenly partitioned as a result of using METIS. This uneven
partitioning also is the reason for the runtime increase from
7 to 8 slaves. By leveraging node locality in Giraph++ and
the equivalence-sets optimization in Giraph++wEq, both ap-
proaches continue to show performance gains over Giraph,
but this time with a more visible difference in the commu-
nication costs among the three variants as shown in Fig-
ure 5(f). Figure 5(g) shows the weak scalability.
Twitter: We next performed a similar scalability evalua-
tion over Twitter, consisting of more than 1.4 billion edges
and more than 41 million vertices. METIS this time resulted
in a very skewed partitioning, with one partition containing
almost half of the edges and almost one third of the edges be-
ing cut edges. This constituted a challenge for our approach,
because we compute the boundary graph along with the cut
edges. However, since vertices in Twitter are densely con-
nected, the resulting compound graphs at all slaves can very
well be condensed using SCC compression, which led to very
small graph indexes at the slaves (with a compression fac-
tor of more than 150). Without this optimization, Giraph
was able to load the Twitter graph but failed to process the
set-reachability query, returning an “out of memory” excep-
tion. The strong and weak scalability of our approach and
the Giraph variants are shown in Figures 5(i) and 5(k).
LUBM: As the final scalability experiment, we considered
the synthetic LUBM-1B dataset whose results are shown
Figures 5(m)-5(p). Most of the RDF-based LUBM graph
is acyclic and sparsely connected. Thus, our SCC conden-
sation for the compound graphs has very low effect on the
overall query processing. Figure 5(m) shows the strong scal-
ability of our approach versus the Giraph variants. Fig-
ure 5(n) shows the communication costs for different vari-
ants of Giraph and our approach. Again, our DSR approach,
which evaluates a set-reachability query in a single round of
communication, exchanges a very low amount of messages
compared to the iterative Giraph variants.
4.3 Updates
We considered the six smaller graphs plus the LiveJ-68M
dataset (see Table 1) for updates. We distinguish two prin-
cipal kinds of incremental update workloads, which we call
bulk updates and progressive updates, respectively.
•For bulk insertions, we start with 60% of randomly cho-
sen edges of the original graph and then increment the
graph by 5% of the remaining edges, until we reach the
original graph. For bulk deletions, we start with the
original graph and decrement the graph in 5% steps.
•For progressive insertions, we randomly pick x%(say,
5%) of edges from the original graph and measure the
time to insert these into an index built over the remain-
ing (100 −x)% (say, 95%) of edges. We increment xin
5% steps. For progressive deletions, we decrement the
original graph by a progressive amount of edges.
We used the same queries as described in Section 4.1 to
measure the effect of these update steps on the query times.
Insertions. Figures 6(a)(e) show the update and respective
query times for our bulk insertions. It can be observed that
the time needed for bulk insertions remains almost constant
for each 5% step. Query performance, which depends on
the final DAG size, however varied considerably with each
update. Next, we considered progressive insertions. From
Figure 6(b), it can be clearly seen that the update times are
only a fraction of the total rebuild time (see Table 2). Query
performance, shown in Figure 6(f), increased marginally at
each step, as expected, although also this depends on the
final DAG size as a result of the update operation.
Deletions. Deletions are generally more costly in our set-
ting and took almost the same time as building the index
from scratch (see Table 2) for both bulk and progressive
updates. Figures 6(c)(g) depict the update and respec-
tive query times for bulk deletions. While deletion times
show a downward trend, query times tend to increase as
the graphs become more sparsely connected, thus leading
to larger DAG sizes. This is especially visible for the LiveJ-
68M dataset. For the case of progressive deletions, as shown
2 3 4 5 6 7 8 9
102
104
Query Time (in ms)
DSR Giraph++wEq Giraph++ Giraph
2 3 4 5 6 7 8 9
101
104
107
Comm. Size (in KB)
2[20%] 3 4 5 6 7 8 9[90%]
103
104
105
Query Time (in ms)
10x10 50x50 100x100
102
104
Query Time (in ms)
DSR Giraph++wEq Giraph++ Giraph
3456789
105
106
Query Time (in ms)
3456789
100
105
Comm. Size (in KB)
2[20%] 3 4 5 6 7 8 9[90%]
102
104
106
Query Time (in ms)
10x10 50x50 100x100
102
103
Query Time (in ms)
3456789
103
105
107
Query Time (in ms)
3456789
102
104
106
Comm. Size (in KB)
2[20%] 3 4 5 6 7 8 9[90%]
105
106
Query Time (in ms)
10x10 50x50 100x100
103
105
107
Query Time (in ms)
x x x
3456789
103
104
105
Strong Scaling (#Slaves)
Query Time (in ms)
3456789
0
10
20
Communication Cost (#Slaves)
Comm. Size (in KB)
2[20%] 3 4 5 6 7 8 9[90%]
103
104
105
Weak Scaling (#Slaves[%Data])
Query Time (in ms)
1kx1k 5kx5k 10kx10k
103
104
105
Query Sizes (|S|x|T|)
Query Time (in ms)
(a) (b) (c) (d)
(e) (f) (g) (h)
(i) (j) (k) (l)
(m) (n) (o) (p)
Figure 5: Scalability evaluation for LiveJ-68M (a-d), Freebase-1B (e-h), Twitter-1.4B (i-l), LUBM-1B (m-p)
in Figures 6(d)(h), we observe similar trends in terms of up-
date and query times.
4.4 Parameters
A. Local Reachability Indexes. We next measured our
DSR approach in conjunction with three centralized strate-
gies. For all three cases, we condense the local compound
graphs via computing SCCs. DSR-DFS uses a standard DFS
strategy [6] for processing a DSR query, where no additional
index is built over the compound graphs. For each source
sand target tin the given DSR query, we perform a DFS
to evaluate the reachability s;t. MS-BFS [30] caches,
for each vertex vthat is visited during a graph traversal,
the reachability of vto all its targets. Thus, if vis another
source in the query, we avoid recomputing the reachability
and thus save a graph traversal. FERRARI [28] finally pro-
vides a tunable tradeoff between index size and query per-
formance. We set both the number of intervals per vertex
and the number of seed vertices to 1,000.
Figure 7 shows the effects on query performance when us-
ing different local search strategies. For this experiment, we
again considered 10 nodes of which one was the master node.
We used two real-world datasets, LiveJ-68M and Freebase-
1B, for this evaluation. We considered different query sizes
to demonstrate the strengths and weaknesses of the three
approaches. Figure 7(a) shows the results for LiveJ-68M.
We can observe that DSR-DFS takes longer compared to
the other two baselines as it requires one graph traversal (in
the worst case) for each source. DSR-FERRARI, with its
compact reachability index, demonstrates significant perfor-
mance gains over the other two baselines for different query
sizes. On the other hand, for large query sizes, the DSR-
MSBFS approach benefits from its memoization and less
redundant graph traversal and tends to close the gap to
FERRARI. The three strategies show similar trends also for
the larger Freebase-1B dataset (see Figure 7(b)).
B. Equivalence-Sets Optimization. By computing equiv-
alence sets among in- and out-boundaries, we are able to
reduce both the boundary-graph sizes as well as the num-
ber of reachability computations required per slave. Table 4
shows the benefits of this optimization. Figure 8 shows a
comparison for the query performance and communication
costs with and without equivalence sets in Giraph.
Query Time Boundary-Graph Sizes
(times in sec.) (#forward; #backward)
Graph Non-Opt. Opt. Non-Opt. Opt.
Amazon 0.101 0.008 900; 530 18; 5
BerkStan 0.157 0.110 20,750; 49,462 3,916; 4,981
Google 1.416 1.003 47,822; 98,955 3,759; 6,287
NotreDame 1.085 0.768 16,771; 6,899 2,481; 37
Stanford 0.061 0.038 5,411; 13,942 183; 475
Table 4: Equivalence-sets optimization in DSR
C. Partitioning Strategy. We next considered the ef-
fect of the partitioning strategy on the performance of our
approach. For this, we used two partitioning strategies: a
random hash-partitioning and METIS [17]. We used a clus-
ter of 6 nodes (one of which was dedicated as the master)
and evaluated the strategies using a set-reachability query
with 10 sources and 10 targets. Table 5 shows the perfor-
mance comparison of the two partitioning strategies over
several real-world graphs. It can be clearly observed that
the choice of the partitioning strategy influences the perfor-
mance of our approach. Hash partitioning (i.e., “random
60%65%70%75%80%85%90%95%
103
104
Update Time (in ms)
5%10%15%20%25%
103
104
Amazon Berkstan Google NotreDame Stanford LiveJ-20M LiveJ-68M
100%95%90%85%80%75%70%65%
103
104
105
5%10%15%20%25%
103
104
105
60%65%70%75%80%85%90%95%
101
102
Bulk Insertions
Query Time (in ms)
5%10%15%20%25%
101
102
Progressive Insertions
100%95%90%85%80%75%70%65%
101
102
Bulk Deletions
5%10%15%20%25%
101
102
Progressive Deletions
(a) (b) (c) (d)
(e) (f) (g) (h)
Figure 6: Update evaluation (both insertions and deletions) for various graph collections
10x10 100x100 1kx1k
101
102
Query Sizes (|S|x|T|)
Query Time (in ms)
DSR-DFS DSR-Ferrari DSR-MSBFS
10x10 100x100 1kx1k
103
104
105
Query Sizes (|S|x|T|)
Query Time (in ms)
(a) LiveJ-68M (b) Freebase-1B
Figure 7: Comparison of local reachability indexes
sharding”) usually results in a drastic increase of cut edges
and thus in a lower query performance. METIS partition-
ing in turn helps in minimizing this cut, which significantly
improves the query performance.
Partitioning
(query times in sec.)
Graph Hash METIS
Amazon 0.009 0.008
BerkStan 0.016 0.009
Google 0.330 0.060
Partitioning
(query times in sec.)
Graph Hash METIS
NotreDame 0.085 0.057
Stanford 0.009 0.008
LiveJ-20 0.524 0.227
LiveJ-68 0.188 0.090
Table 5: Impact of hash vs. METIS partitioning
4.5 Applications
A. SPARQL 1.1 with Property Paths. For this exper-
iment, we considered the LUBM-500M and Freebase-500M
datasets (both in RDF format). We augmented a distributed
RDF store [15] with our DSR approach by modifying its
query processor to handle property paths via our new index
structures. To evaluate the performance of our approach in
processing SPARQL 1.1 queries, we compared against the
commercial Virtuoso RDF store [8]. The results are shown
in Table 6, while the customized SPARQL queries we used
for this evaluation are depicted in Appendix 8.3
B. Social-Network Communities. As another DSR ap-
plication, we detected connectivities among communities in
a social network. This problem is a basic step in many
graph-analytics tasks. That is, given two communities C1
and C2together with a set of representative members for
each community S⊆C1,T⊆C2, find all pairs s,t, with
s∈Sand t∈T, such that s;t. We considered two social
network datasets, LiveJ-68M and Twitter-1.4B, for this ex-
periment. We employed the iterative community-detection
algorithm by Blondel et al. [3] to identify communities. We
then randomly picked two communities, and from each we
picked 10 to 1,000 members as representatives. We then
Amazon
BerkStan
Google
NotreDame
Stanford
LiveJ-20M
101
102
103
# Supersteps
Giraph++wEq Giraph++ Giraph
Amazon
BerkStan
Google
NotreDame
Stanford
LiveJ-20M
100
102
104
106
Comm. Size (in KB)
Figure 8: Equivalence-sets optimization in Giraph
ran our DSR approach to identify all reachable pairs among
these representatives. The results are shown in Table 7.
(a) LUBM-500M (query times in sec.) Geo.-
#Slaves L1 L2 L3 Mean
DSR 1 6.437 0.331 42.681 4.497
DSR 5 1.250 0.162 8.516 1.199
Virtuoso (cold) 1 10.050 12.624 57.776 19.425
Virtuoso (warm) 1 4.963 5.452 56.603 11.527
(b) Freebase-500M (query times in sec.) Geo.-
#Slaves F1 F2 F3 Mean
DSR 1 1.084 1.568 0.677 1.048
DSR 5 0.356 0.642 0.423 0.459
Virtuoso (cold) 1 6.590 4.112 13.809 7.206
Virtuoso (warm) 1 1.196 0.002 5.601 0.238
Table 6: SPARQL 1.1 queries with property paths
LiveJ-68M Twitter-1.4B
#Communities: 5,032 #Communities: 17,121
Query Size Query Time #Pairs Query Time #Pairs
(|S|x|T|) (in sec.) (in sec.)
10x10 0.065 81 1.339 63
100x100 0.164 8,184 2.476 8,526
1kx1k 0.717 784,947 10.175 712,725
Table 7: Community connectedness using DSR
4.6 Summary of Results
Our experiments confirm the significantly improved effi-
ciency (with a gain in query times of several orders of mag-
nitude) of our DSR index compared to iterative approaches
such as Apache Giraph and variants of [9]. Moreover, we are
also able to demonstrate the good update support of our in-
dex structure, which—in particular for insertions—behaves
much better in practice than suggested by the worst-case
bounds we provide in Section 3.3.3. We believe that in-
sertions are the much more likely use-case for managing
large, dynamic graphs (e.g., Twitter streams), while dele-
tions, which are costly to handle for any kind of graph-
compression technique, are much more uncommon in prac-
tice. Also, there is also hardly any support for updates in
the centralized approaches (such as [28]), which restricts our
local search strategy to a simple DFS or BFS in this case.
Further experiments demonstrate the robustness of our ap-
proach under different parameters and show its viability for
various large-scale graph-analytics tasks.
5. RELATED WORK DISCUSSION
Centralized Approaches. The reachability problem in
directed graphs is one of the most fundamental graph prob-
lems and thus has been tackled by a plethora of centralized
indexing techniques [5, 12, 16, 18, 25, 28, 32, 33, 34, 36] (just
to name a number of recent approaches). All of these aim to
find tradeoffs among query time and indexing space which,
for a directed graph G(V, E), are in between O(|V|+|E|)for
both query time and space consumption when no indexes are
used, and O(1) query time and O(|V|2)space consumption
when the transitive closure is fully materialized.
Recently, Gao et al. [12] proposed a suitable, but cen-
tralized indexing strategy, based on a notion of equivalence
sets of graph vertices that have the same reachability prop-
erties. [30], on the other hand, focused on the query-time
optimization of multi-source BFS searches. However, there
exist hardly any works so far on distributed reachability
queries [9]. Fan et al. [9] recently discussed distributed, but
single-source, single-target reachability algorithms for parti-
tioned graphs and also provide performance guarantees. For
a directed graph and given cut, [9] uses a partially iterative
and partially indexing-based evaluation to find the reacha-
bility for a given pair of vertices over a classical master-slave
architecture. In Section 3, we therefore provide a detailed
review of the techniques proposed in [9], while the query-
time processing we perform based on equivalence sets to a
large extent resembles also the techniques described in [12,
30] for a centralized setting. However, unlike in [12], we do
not enumerate the actual path sequences.
Distributed Graph Engines. Distributed graph engines
such as Pregel [22], GraphX [35], GraphLab [20], Trinity [29],
PowerGraph [21], Giraph [1] and Giraph++ [31] are either
based on MapReduce [22, 35, 21], or they implement their
own, proprietary communication protocols via Message Pass-
ing [13, 20, 29]. Giraph, for example, offers the sendMes-
sage(.) and compute(.) methods as generic API functions to
implement various kinds of graph algorithms (including BFS
and DFS). To implement a single-source, single-target reach-
ability query over a directed graph, each iteration over the
compute(.) method (as it is required for a single BFS/DFS
step), however, results in a new call of the Map function
or so-called “superstep”. Among two such supersteps, mes-
sages are communicated among all compute nodes, which
is a strategy that—due to the a-priori unknown amount of
iterations—usually does not permit for interactive query re-
sponse times. For multi-source, multi-target queries, on the
other hand, this approach scales well with the query size due
to the possibility to implement shared computations in the
compute(.) method.
A similar observation holds for GraphLab [21], Trinity [29]
and PowerGraph [13] which implement asynchronous pro-
tocols based on the Message Passing Interface (MPI) [11].
PowerGraph, for example, which is specifically tuned for
skewed graphs, implements a judiciously chosen schedule of
exchanged messages, but also here the worst-case amount
of iterations remains equal to the diameter of the graph.
The very recently proposed Giraph++ [31] framework (built
on top of Giraph) provides further optimizations by shifting
from a purely node-centric to a more graph-centric (“think
like a graph”) compute paradigm. All (local) messages among
the vertices within the same graph partition are performed
inside a superstep, while other messages are processed only
between two such supersteps. This significantly improves
the performance by minimizing the number of messages but
also gives a much higher degree of freedom in the implemen-
tation of various graph algorithms.
Thus, on the one hand, the generic abstraction layers of
these distributed graph engines make it difficult to exploit
shared computations and yet to fully benefit from the un-
derlying distribution scheme. On the other hand, these en-
gines generally do not support graph-indexing techniques
known from the centralized approaches, which could ideally
be employed to even completely avoid iterative communica-
tion rounds among the compute nodes. Since Giraph++ of-
fers the most flexible API among the aforedescribed engines,
we extensively compared our approach against two princi-
ple implementations of DSR queries in the very recent Gi-
raph++ framework (including one native Giraph version).
Processing SPARQL 1.1 Property Paths. Finally,
combining relational joins with reachability predicates in
SPARQL 1.1 inherently leads to a multi-source, multi-target
(i.e., set-) reachability problem. Very few works so far fo-
cused on the combined processing of relational joins with
additional graph-reachability predicates [4, 10, 14, 27]. In
earlier works, the bisimulation-based indexing of path ex-
pressions for XML trees [23] has been extended to RDF
graphs [24], but the latter did not yet consider property
paths. Likewise, [4] proposed an index structure that is
limited to DAGs obtained from XML/XLink(s). [26] fi-
nally investigated an initial approach to evaluate property
paths via MapReduce. Also here, we are aware of only one
RDF engine that is available for processing property paths
in SPARQL 1.1, namely Virtuoso [8], against which we com-
pared our approach experimentally.
6. CONCLUSIONS
We investigated a generalized form of the well-known reach-
ability problem in directed graphs, which (1) considers both
sets of source and target vertices as queries, and (2) allows
the underlying graph to be partitioned and hence be dis-
tributed across multiple compute nodes. The DSR problem
is a basic building block and thus has a plethora of appli-
cations in graph analytics and query-processing tasks. As
our core contribution, we presented an efficient and scalable
framework for processing DSR queries and also studied its
formal properties. By precomputing and materializing the
reachability information among vertices along the cut of a
partitioned data graph, our approach is guaranteed to re-
quire at most one round of communication among the com-
pute nodes to resolve any DSR query. Our approach exhibits
a very good support for incremental vertex and edge inser-
tions, while our current implementation resorts to just a ba-
sic support for respective deletions. Moreover, any state-of-
the-art centralized reachability index may be applied to the
local graph partitions to further accelerate query-processing
times. Our evaluation over both real-world and synthetic
graphs and in comparison to very recent iterative approaches
also empirically demonstrated the viability of our approach.
As a future work, we aim to explore more types of reachabil-
ity problems with additional length restrictions and regular
expressions over path labels as constraints.
7. REFERENCES
[1] http://giraph.apache.org/.
[2] http://www.w3.org/TR/sparql11-property-paths/.
[3] V. D. Blondel, J.-L. Guillaume, R. Lambiotte, and
E. Lefebvre. Fast unfolding of communities in large
networks. Journal of Statistical Mechanics: Theory
and Experiment, 2008(10):P10008, 2008.
[4] J. Cheng, J. X. Yu, and B. Ding. Cost-Based Query
Optimization for Multi Reachability Joins. In
DASFAA, pages 18–30, 2007.
[5] E. Cohen, E. Halperin, H. Kaplan, and U. Zwick.
Reachability and distance queries via 2-hop labels. In
SODA, pages 937–946, 2002.
[6] T. H. Cormen, C. E. Leiserson, R. L. Rivest, and
C. Stein. Introduction to Algorithms. The MIT Press,
3rd edition, 2009.
[7] C. Demetrescu and G. F. Italiano. Dynamic shortest
paths and transitive closure: Algorithmic techniques
and data structures. Journal of Discrete Algorithms,
4(3):353 – 383, 2006.
[8] O. Erling and I. Mikhailov. Virtuoso: RDF Support in
a Native RDBMS. In SWIM, pages 501–519, 2009.
[9] W. Fan, X. Wang, and Y. Wu. Performance
guarantees for distributed reachability queries.
PVLDB, 5(11):1304–1315, 2012.
[10] W. Fan, X. Wang, and Y. Wu. Answering graph
pattern queries using views. In ICDE, pages 184–195,
2014.
[11] T. M. Forum. MPI: A Message Passing Interface, 1993.
[12] S. Gao and K. Anyanwu. PrefixSolve: efficiently
solving multi-source multi-destination path queries on
RDF graphs by sharing suffix computations. In
WWW, pages 423–434, 2013.
[13] J. E. Gonzalez, Y. Low, H. Gu, D. Bickson, and
C. Guestrin. PowerGraph: Distributed Graph-Parallel
Computation on Natural Graphs. In OSDI, pages
17–30, 2012.
[14] A. Gubichev, S. J. Bedathur, and S. Seufert. Sparqling
Kleene: fast property paths in RDF-3X. In GRADES,
pages 14:1–14:7, 2013.
[15] S. Gurajada, S. Seufert, I. Miliaraki, and
M. Theobald. TriAD: a distributed shared-nothing
RDF engine based on asynchronous message passing.
In SIGMOD, pages 289–300, 2014.
[16] R. Jin, N. Ruan, Y. Xiang, and H. Wang. Path-tree:
An efficient reachability indexing scheme for large
directed graphs. TODS, 36(1):7, 2011.
[17] G. Karypis and V. Kumar. A Fast and High Quality
Multilevel Scheme for Partitioning Irregular Graphs.
SIAM J. Sci. Comput., 20(1):359–392, 1998.
[18] A. Kyrola, G. Blelloch, and C. Guestrin. GraphChi:
Large-Scale Graph Computation on Just a PC. In
OSDI, pages 31–46, 2012.
[19] J. Łącki. Improved deterministic algorithms for
decremental reachability and strongly connected
components. ACM Trans. Algorithms, 9(3):27:1–27:15,
2013.
[20] Y. Low, J. Gonzalez, A. Kyrola, D. Bickson,
C. Guestrin, and J. M. Hellerstein. GraphLab: A New
Framework For Parallel Machine Learning. In UAI,
pages 340–349, 2010.
[21] Y. Low, J. Gonzalez, A. Kyrola, D. Bickson,
C. Guestrin, and J. M. Hellerstein. Distributed
GraphLab: A Framework for Machine Learning in the
Cloud. PVLDB, 5(8):716–727, 2012.
[22] G. Malewicz, M. H. Austern, A. J. C. Bik, J. C.
Dehnert, I. Horn, N. Leiser, and G. Czajkowski.
Pregel: a system for large-scale graph processing. In
SIGMOD, pages 135–146, 2010.
[23] T. Milo and D. Suciu. Index structures for path
expressions. In ICDT, pages 277–295, 1999.
[24] F. Picalausa, Y. Luo, G. H. L. Fletcher, J. Hidders,
and S. Vansummeren. A structural approach to
indexing triples. In ESWC, pages 406–421, 2012.
[25] V. Prabhakaran, M. Wu, X. Weng, F. McSherry,
L. Zhou, and M. Haradasan. Managing large graphs
on multi-cores with graph awareness. In USENIX,
pages 41–52, 2012.
[26] M. Przyjaciel-Zablocki, A. Schätzle, T. Hornung, and
G. Lausen. RDFPath: Path query processing on large
RDF graphs with MapReduce. In ESWC, pages
50–64, 2011.
[27] M. Sarwat, S. Elnikety, Y. He, and M. F. Mokbel.
Horton+: A distributed system for processing
declarative reachability queries over partitioned
graphs. PVLDB, 6(14):1918–1929, 2013.
[28] S. Seufert, A. Anand, S. J. Bedathur, and G. Weikum.
FERRARI: flexible and efficient reachability range
assignment for graph indexing. In ICDE, pages
1009–1020, 2013.
[29] B. Shao, H. Wang, and Y. Li. Trinity: a distributed
graph engine on a memory cloud. In SIGMOD, pages
505–516, 2013.
[30] M. Then, M. Kaufmann, F. Chirigati, T. Hoang-Vu,
K. Pham, A. Kemper, T. Neumann, and H. T. Vo.
The more the merrier: Efficient multi-source graph
traversal. PVLDB, 8(4):449–460, 2014.
[31] Y. Tian, A. Balmin, S. A. Corsten, S. Tatikonda, and
J. McPherson. From "think like a vertex" to "think like
a graph". PVLDB, 7(3):193–204, 2013.
[32] S. Trißl and U. Leser. Fast and practical indexing and
querying of very large graphs. In SIGMOD, pages
845–856, 2007.
[33] S. J. van Schaik and O. de Moor. A memory efficient
reachability data structure through bit vector
compression. In SIGMOD, pages 913–924, 2011.
[34] R. R. Veloso, L. Cerf, W. M. Jr., and M. J. Zaki.
Reachability queries in very large graphs: A fast
refined online search approach. In EDBT, pages
511–522, 2014.
[35] R. S. Xin, J. E. Gonzalez, M. J. Franklin, and
I. Stoica. GraphX: a resilient distributed graph system
on Spark. In GRADES, 2013.
[36] H. Yildirim, V. Chaoji, and M. J. Zaki. GRAIL:
Scalable Reachability Index for Large Graphs.
PVLDB, 3(1-2):276–284, 2010.
[37] M. Zaharia, N. M. M. Chowdhury, M. Franklin,
S. Shenker, and I. Stoica. Spark: Cluster Computing
with Working Sets. Technical Report
UCB/EECS-2010-53, UC Berkeley, 2010.
8. APPENDIX
8.1 Computing Equivalence Sets
Algorithm 3 computes the forward-equivalent sets of ver-
tices in each graph partition Gias follows. Given a graph
partition Giwith in-boundaries Ii, the forward-equivalent
sets EQf
iare computed as follows. First, the graph is con-
densed into its DAG representation G0
iby computing the
strongly connected components of Gi. Next, the target
vertices S(Ii)−Iiare chosen as the successors of the in-
boundaries Ii. We define a Boolean array rep[] (for “rep-
resentative”), whose truth value for a given boundary bm
denotes whether a forward-equivalent set (i.e., an in-virtual
vertex) υis formed with any other boundary bl, for m>l.
The rep[] array is initially set to “true” for all boundaries. A
boundary blis equivalent to bm, iff either bland bmbelong
to same SCC (Lines 11-14) or have the same reachability
set rset (Lines 17-19). rset[j]for the jth boundary denotes
the set of vertices from S(Ii)−Iithat are reachable from
bj. The computed equivalence set υstarting at boundary bl,
where rep[bl]:=true, is added to EQf
iat the end of each itera-
tion (outer loop). With minor modifications from S(Ii)−Ii
to P(Oi)−Oi(thus using predecessors instead of succes-
sors), the algorithm can similarly be adapted to compute
the backward-equivalent sets EQb
iof out-boundaries.
8.2 Proofs of Theorems
8.2.1 Proof of Theorem 1
Proof. Let P={(s, u1),...,(um, u),(u, v),(v, vn),...,
(v1, t)}denote the set of edges along a path from source s
to target t. Then the following holds: edge (u, v)∈P, with
u∈Vi, v ∈Vj, can either be (1) a cut edge iff i6=j, (2) a
local edge in partition iiff i=j, or (3) a non-local edge with
respect to partition kiff i=jand i6=k.
Case A: Let all edges in Pbe either local to partition i
(1) or be a cut edge (2) among partitions i,j. Then, from
Definition 6 of the compound graph GC
i= (VC
i, EC
i), it
follows that P⊆EC
i. That is, the reachability s;tcan be
computed entirely locally at partition iusing EC
i.
Case B: Let (u, v)∈Psuch that (u, v)is a non-local edge
(3) to partition ibut a local edge (1) to another partition
j, with i6=j. That is, (u, v)∈EC
jbut (u, v)/∈EC
i. From
this, it follows that ∃p, q such that the edges (up−1, up),
(vq, vq−1)∈Pare cut edges (2), where up,vq∈Vjwith
1≤p≤mand 1≤q≤n. Next, we choose (up−1, up),
(vq, vq−1)∈Pas the edges with the largest indices of p, q for
which this property holds. This choice ensures that a path
from upto vqvia (u, v)resides entirely in partition j. Then,
vertex upforms an in-boundary while vertex vpforms an
out-boundary of partition j, and the edges of the sub-path
{(up−1, up),...,(um, u),(u, v),(v, vn),..., (vq, vq−1)} ⊆ P
reside in partition j. In this case, by the construction of
the boundary graph GB
i= (VB
i, EB
i), we added a reacha-
bility edge (up, vq)to EB
i(see Definition 4). This means,
in the optimized EB
i, we add an edge (υ, ν), where up∈υ
is an in-virtual vertex and vq∈νis an out-virtual ver-
tex. Since EB
i⊆EC
i(see Definition 6), there exists a path
{(s, u1),(u1, u2),...,(up−1, υ),(υ, ν),(ν, vq−1),...,(v1, t)}in
partition i, thus again ensuring that the reachability of s;t
can be computed entirely locally at partition iusing EC
i.
Algorithm 3: Computing Forward-Equivalent Sets
Input: Subgraph Gi, In-boundaries Ii
Output: Forward-equivalent sets EQf
i
1EQf
i:= ∅
2G0
i:= condense(Gi).graph condensation via SCC computation
3S(Ii) := successors(Ii,G0
i)
4rep[1..|Ii|] := true
5rset[k] := ∅
6for l= 1 ...|Ii|do
7if rep[l]then
8υ:= {bl}
9if rset[bl]=∅then
10 rset := localSetReachability({bl}, S(Ii)−Ii)
11 for m=l+ 1 ...|Ii|do
12 if scc(bl)=scc(bm)then
13 υ:= υ∪ {bm}
14 rep[m]:= false
15 else
16 rset := localSetReachability({bm}, S(Ii)−Ii)
17 if rset[bl]=rset[bm]then
18 υ:= υ∪ {bm}
19 rep[m]:= false
20 EQf
i:= EQf
i∪υ
8.2.2 Proof of Theorem 2
Proof. The proof is simple and leverages the result of
Theorem 1. Let P={(s, u1),(u1, u2),...,(up−1, up),...,
(un, t)}denote the set of edges along the path from source
sto target t, where s∈Vi, t ∈Vj. If i6=j, then there exists
an edge (up−1, up)∈Pwhich is a cut edge (1). That is,
up−1∈Vk,up∈Vjand k6=j. Next, we choose the largest
index psuch that the subpath {(up, up+1),...,(un, t)}re-
sides entirely at partition j. Since (up−1, up)is a cut edge
and up∈Vj,upforms an in-boundary of partition j. We
next choose the smallest qsuch that {(s, u1),...,(uq−1, uq)}
resides entirely at partition i. Note that in the optimized
boundary graph, we actually use virtual vertices instead of
the regular ones, which we omit here for simplicity.
Path Pcan be thus written as a concatenation of subpaths
P1={(s, u1), . . . , (uq−1, uq),P2= (uq, uq+1). . . (up−1, up},
and P3={(up, up+1),...,(un, ut)}. According to Theo-
rem 1, P1and P3can be computed entirely at partition iand
j, respectively. uqand upthus are an out- and in-boundary
of partition iand j, respectively, i.e., uq, up∈VC. As per
the construction of the local compound graphs (see Defini-
tion 6), P2can be evaluated at either partition ior j. Thus,
the reachability problem s;tis reduced to two reachability
problems: (a) s;upat partition iand (b) up;tat par-
tition j. To find such a up, we iterate over all in-boundaries
bof partition jresiding at partition i. We then compute
the reachability s;band communicate the reachable in-
boundaries to partition j. Thus, answering s;t, where
s∈Viand t∈Vj, with i6=j, requires a local processing at
two partitions i,jand involves only a single step of message
exchange from partition ito partition j.
8.3 SPARQL 1.1 Queries with Property Paths
A. LUBM Queries
@prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
@prefix ub: <http://www.lehigh.edu/∼zhp2/2004/0401/univ-bench.owl#>
L1: SELECT * WHERE { ?x rdf:type ub:ResearchGroup .
?x ub:subOrganizationOf* ?y. ?y rdf:type ub:University. }
L2: SELECT * WHERE { ?x rdf:type ub:FullProfessor. ?x ub:headOf ?d.
?d ub:subOrganizationOf* ?y. ?y rdf:type ub:University.}
L3: SELECT * WHERE { ?r1 rdf:type ub:ResearchGroup .
?r1 ub:subOrganizationOf* ?y. ?y rdf:type ub:University . ?r2 rdf:type
ub:ResearchGroup. ?r2 ub:subOrganizationOf* ?y.}
B. Freebase Queries
@prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
@prefix fb: <http://rdf.freebase.com/ns>
F1: SELECT * WHERE { ?p fb:people.person.place_of_birth ?city . ?city
fb:location.location.containedby* ?state. ?country fb:location.location.con-
tains ?state. }
F2: SELECT * WHERE { ?p fb:people.person.place_of_birth ?city . ?city
fb:location.location.containedby* ?state. ?country fb:location.location.con-
tains ?state. ?p fb:award.award _winner.awards_won ?prize.
?p rdf:type fb:government.us _president. }
F3: SELECT * WHERE {?p fb:award.award_winner.awards_won ?prize.
?prize rdf:type* ?z . ?z fb:award.award_honor.ceremony> ?c.
?p fb:people.person.sibling_s* ?p1. ?p1 fb:award.award_winner.awards_won>
?prize. }
8.4 Giraph Implementations of DSR Queries
8.4.1 Giraph
The following program code illustrates our implementa-
tion of DSR queries in Giraph. In superstep 0, all source
vertices are first added to a newSources array. Thus, the
function isSource(.) returns true if the vertex is a source.
In the subsequent supersteps, newSources represents the ad-
ditional sources from which the current vertex vis reachable.
If newSources is not empty, then we iteratively propagate
these sources to all neighbors of vertex v.
Distributed Set Reachability in Giraph
public void compute(Vertex v, Iterable m){
ArrayList<Integer> newSources = new ArrayList<Integer>();
if(getSuperStep() == 0){
if(isSource(v))
newSources.add(v.getId().get());
v.getValue().clearSources()
}else
for(IntWritable msg : m)
newSources.add(m.get());
newSources.removeAll(v.getValue().getSources());
if(newSources.size() > 0){
v.addSources(newSources);
for(Edge<IntWritable, NullWritable> e : v.getEdges()){
IntWritable nb = e.getTargetVertexId();
for(int src: newSources)
sendMessage(nb,new IntWritable(src)); } } }
}
8.4.2 Giraph++
Unlike in Graph, the Giraph++ API exposes the underly-
ing partitioning information along with each call of the com-
pute(.) function. The code for DSR processing is similar to
Giraph, except that the vertices that are local to the current
source vertices are directly updated using a centralized lo-
cal reachability computation via localProcess(.). After the
local processing, for each vertex we again communicate its
reachable list of vertices to the remote neighbors.
Distributed Set Reachability in Giraph++
public void compute(Partition p){
ArrayList<Integer> q_sources = new ArrayList<Integer>();
ArrayList<Integer> newSources = new ArrayList<Integer>();
if(getSuperStep() == 0){
if(isSource(v))
sources.add(v.getId().get();
}else{
MessageStore<IntWritable, IntWritable> mstore
= getCurrentMessageStore();
for(Vertex v : p.getVertices()){
if(mstore.hasMessagesForVertex(v.getId())){
newSources.clear();
for(IntWritable message :
mstore.getVertexMessages(v.getId()))
newSources.add(message.get());
newSources.removeAll(v.getValue().getSources());
if(newSources.size() > 0){
q_sources.add(v.getId());
v.getValue().addNewSources(newSources);
}}}
}
localProcess(p,q_sources);
for(Vertex v : p.getVertices()){
if(v.getValue().getNewSources().size() > 0){
for(Edge<IntWritable, NullWritable> edge
: v.getEdges()){
int nb = edge.getTargetVertexId().get();
if(!p.contains(nb)
for(int src : v.getValue().getNewSources())
sendMessage(nb,new IntWritable(src));
}
v.getValue().addSources(v.getValue().getNewSources());
v.getValue().getNewSources().clear();}
}
}
8.4.3 Giraph++wEq
The following code depicts our DSR implementation in
Giraph++wEq, including our proposed equivalence-sets op-
timization. We first compute equivalence sets in our DSR
system and prepare an adjacency graph as input to Giraph.
For each vertex vin the input graph, in addition to its ad-
jacent neighbors, we also add their equivalence sets (our
in-virtual vertices) as counterparts. This graph is loaded
into Giraph using a custom input reader. The below code
shows the DSR computation. The implementation shares a
major part of the code with the Giraph++implementation,
where the only difference lies in the communication of the
reachable sets of vertices in each superstep. After the local
processing, we iterate over each vertex and send its reach-
able list of sources to only the in-virtual vertices instead of
all neighbors.
Distributed Set Reachability in Giraph++wEq
public void compute(Partition p){
ArrayList<Integer> q_sources = new ArrayList<Integer>();
ArrayList<Integer> newSources = new ArrayList<Integer>();
if(getSuperStep() == 0){
if(isSource(v))
sources.add(v.getId().get();
}else{
MessageStore<IntWritable, IntWritable> mstore
= getCurrentMessageStore();
for(Vertex v : p.getVertices()){
if(mstore.hasMessagesForVertex(v.getId())){
newSources.clear();
for(IntWritable message :
mstore.getVertexMessages(v.getId()))
newSources.add(message.get());
newSources.removeAll(v.getValue().getSources());
if(newSources.size() > 0){
q_sources.add(v.getId());
v.getValue().addNewSources(newSources);
}}}
}
localProcess(p,q_sources);
for(Vertex v : p.getVertices()){
if(v.getValue().getNewSources().size() > 0){
for(int eq_nb : v.getEqList())
for(int src : v.getValue().getNewSources())
sendMessage(new IntWritable(eq_nb),new IntWritable(src));
v.getValue().addSources(v.getValue().getNewSources());
v.getValue().getNewSources().clear();}
}
}
