Web-scale RDF datasets are increasingly processed using distributed RDF data stores built on top of a cluster of shared-nothing servers. Such systems critically rely on their data partitioning scheme and query answering scheme, the goal of which is to facilitate correct and e cient query processing. Existing data partitioning schemes are commonly based on hashing or graph partitioning techniques. The latter techniques split a dataset in a way that minimises the number of connections between the resulting subsets, thus reducing the need for communication between servers; however, to facilitate e cient query answering, considerable duplication of data at the intersection between subsets is often needed. Building upon the known graph partitioning approaches, in this paper we present a novel data partitioning scheme that employs minimal duplication and keeps track of the connections between partition elements; moreover, we propose a query answering scheme that uses this additional information to correctly answer all queries. We show experimentally that, on certain well-known RDF benchmarks, our data partitioning scheme often allows more answers to be retrieved without distributed computation than the known schemes, and we show that our query answering scheme can e ciently answer many queries.
While the exibility of the RDF data model o ers many advantages, e cient
management of large RDF datasets remains an open research topic. RDF data
management systems can be conceived as single-machine systems constructed
using techniques originating from relational databases [
The two main challenges in developing a distributed RDF system are (i) how to split up the data across multiple servers (i.e., data partitioning), and (ii) how to answer queries in a distributed environment (i.e., distributed query answering). These two challenges are closely connected: knowledge about data partitioning is relevant for query answering, and knowledge of typical query structure can be used to inform the data partitioning scheme. Nevertheless, one can investigate independently from query answering the extent to which a speci c data partitioning scheme reduces the need for distributed processing; for example, one can identify the percentage of the query answers that can be computed locally|that is, by evaluating queries on each server independently.
Data partitioning via hashing is a well-known partitioning scheme from the
database community, and it has been applied in several RDF systems [
These approaches can be augmented by duplicating data on partition
boundaries [
In this paper we focus mainly on the e ects of our data partitioning scheme,
which we experimentally show to be very promising on the LUBM [
We also experimentally evaluate our query answering scheme, which we show can e ectively answer many queries from the LUBM and SP2B benchmarks with little or no distributed processing. On some queries, however, our scheme becomes impractical as it requires the computation of a large number of partial matches (see Section 4 for details)|a drawback we look to overcome in our future work. These results are preliminary in the absence of a complete distributed system that would allow us to measure query processing times; instead, we measure the amount of work (in a sense that we make precise in Section 4) involved in distributed query processing. 2 2.1
RDF Let R be a set of resources. An RDF term is either a variable or a resource from R; a term is ground if it is not a variable. An RDF atom A is an expression of the form hs; p; oi where s, p, and o are RDF terms; the vocabulary of A is de ned as voc(A) = fs; p; og; var(A) is the set of all variables in voc(A); atom A is ground if var(A) = ;; a triple is a ground atom; and an RDF graph G is a set of triples. The vocabulary of G is de ned as voc(G) = SA2G voc(A). As in the SPARQL query language, all variables in this paper start with a question mark. A variable assignment (or just assignment ) is a partial mapping of variables to resources. For r a resource, let (r) = r; for A = hs; p; oi an RDF atom, let (A) = h (s); (p); (o)i; and for S a set of atoms, let (S) = SA2S (A). The domain dom( ) of is the set of variables that is de ned on; and the range rng( ) of is rng( ) = f (x) j x 2 dom( )g. An RDF conjunctive query Q is an expression of the form (1), where each Ai, 1 i m, is an RDF atom.
Q = A1 ^ : : : ^ Am By a slight abuse of notation, we often identify Q with the set of its atoms. The vocabulary and the set of variables of Q are de ned as follows.
[ 1 i m voc(Q) = voc(Ai) var(Q) =
[ 1 i m var(Ai) We sometimes write queries using the SPARQL syntax. An answer to a query Q over an RDF graph G is an assignment such that dom( ) = var(Q) and (Q) G; and ans(Q; G) is the set of all answers to Q over G. Note that our de nitions do not support variable projection. 2.2
A partition of an RDF graph G is an n-tuple of RDF graphs G = (G1; : : : ; Gn) such that G G1 [ : : : [ Gn. Each Gi is called a partition element, and n is the (1) (2) size of G; an n-partition is a partition of size n. One might expect a partition to satisfy G = G1 [ : : : [ Gn, but our relaxed condition allows us to capture our ideas in Section 3. Furthermore, we allow triples to be duplicated across partition elements|that is, we do not require Gi \ Gj = ; for i 6= j. A partitioning scheme is a process that, given an RDF graph G, produces a partition G. Given an RDF conjunctive query Q, an answer 2 ans(Q; G) is local for Q and G if some 1 i n exists such that 2 ans(Q; Gi); otherwise, is non-local for Q and G. When Q and G are clear, we simply call local or non-local. Since non-local answers span partition elements, they are more expensive to compute if each partition element is stored on a separate server; hence, the main aim of a partitioning scheme is to maximise the number of local answers to common queries. The quality of a partition G for a set of queries Q is de ned as the ratio of local answers to all answers for all of the queries in Q on G. We often use this term informally, in which case we consider partition quality with respect to an unspeci ed set of queries that can be considered typical.
Allowing duplication of triples in partition elements can improve partition quality: given a non-local answer to a query Q, answer becomes local if we add (Q) to some partition element. However, duplication also increases storage overhead, and in the limit it can result in each partition element containing a complete copy of G. Hence, another aim of a partitioning scheme is to achieve a suitable balance between triple duplication and partition quality.
A distributed query answering scheme, or just a query answering scheme, is a process that, given a partition G and a query Q, returns a set of assignments ans(Q; G) such that ans(Q; G) = ans(Q; G). If the shape of Q guarantees that all answers are local, one can evaluate Q against each element of G and take the union of all answers; otherwise, additional work is required to identify nonlocal answers or to detect that no such answers exist. Query answering schemes di er mainly in how they handle the latter case: answer pieces are spread across multiple partition elements and they must be retrieved and joined together. We now have two clear goals for a query answering scheme: the rst goal is to ensure correctness|that is, that ans(Q; G) = ans(Q; G)|and the second goal is to minimise the amount of work required to construct non-local answers. 2.3
Now we present a brief overview of partitioning schemes and query answering schemes known in the literature.
Hashing is the simplest and most common data partitioning scheme [
Graph-based approaches exploit the graph structure of RDF data. In
particular, one can use min-cut graph partitioning software such as METIS [
One can combine an arbitrary data partitioning scheme with n-hop duplication to increase the proportion of local answers. Given an RDF graph G and a subgraph H G, the n-hop expansion Hn of H with respect to G is de ned recursively as follows: H0 = H and, for each 1 i n,
Hi = Hi 1 [ fhs; p; oi j hs; p; oi 2 G and fs; og \ voc(Hi 1) 6= ;g:
(3)
While n-hop partitioning can considerably improve partition quality [
Most data partitioning schemes are paired with a speci c query answering
scheme. Due to lack of space, we cannot present all such approaches in detail.
Many of them have been implemented using MapReduce [
We now present our novel data partitioning scheme. Similar to [
The quality of partitions critically depends on the structure of the data and the anticipated query workload. Although application speci c, we found the following assumptions to be common to a large number of applications. Assumption 1. Subject{subject joins are common.
Assumption 2. Queries often constrain variables to elements of classes|that is, they often contain atoms of the form h?x; rdf :type; classi.
Assumption 3. Joins involving resources representing classes are uncommon| that is, queries rarely contain atoms h?x1; rdf :type; ?yi ^ h?x2; rdf :type; ?yi. Assumption 4. Joins on resources that are literals are uncommon|that is, if a query contains atoms h?x1; :R; ?yi ^ h?x2; :S ; ?yi, it is unlikely that a query answer will map variable ?y to a literal.
Assumption 5. The number of schema triples in G is small, so all schema triples can be replicated in each partition element.
Although n-hop duplication can increase partition quality [
Given an RDF graph G, let be a distinguished wildcard resource such that 62 voc(G). Now let V voc(G) be a subset of the vocabulary of G. Given a resource r, let [r]V = r if r 2 V and [r]V = otherwise. Moreover, given an RDF atom A = hs; p; oi, let [A]V = h[s]V ; [p]V ; [o]V i. Finally, given an RDF graph G, let [G]V be the RDF graph de ned by [G]V = f[A]V j A 2 Gg. In the rest of this section we formalise our data partitioning scheme, and in Section 4 we show how to use the wildcard resource to answer queries.
Instead of partitioning triples directly, we partition the vocabulary of the graph and use the result to construct a partition of the triples. More precisely, to construct an n-partition of G we rst partition voc(G) into n subsets V1; : : : ; Vn, and then we use these to construct a partition G = ([G]V1 ; : : : ; [G]Vn ) of G. To ensure that G is a valid partition, we must select V1; : : : ; Vn such that G [G]V1 [ : : : [ [G]Vn (4) holds. To achieve this, we rst partition the vocabulary voc(G) into n disjoint sets V10; : : : ; Vn0 , and then we extend these sets so that condition (4) is satis ed. This extension allows resources to be duplicated in multiple partition elements, which in turn means triples can also be duplicated. Typically, the duplicated triples are those that are on, or near, the border between partition elements. Since resource is not contained in voc(G), we use the subset relation in condition (4), rather than a more intuitive equality relation. Furthermore, our approach ensures that, for each partition element Gi = [G]Vi and each triple hs; p; oi 2 G, if fs; p; og voc(Gi), then hs; p; oi 2 Gi holds. This property is not satis ed in previously known partitioning schemes, but it increases partition quality. We formalise these ideas using the following steps.
Step 1. Compute the undirected graph G0 by removing from G all schema triples and triples containing class and literal resources (i.e., all triples of the form hs; rdf :type; oi and hs; p; `i with ` a literal), and by treating each remaining triple hs; p; oi as an undirected edge connecting s and o. Step 2. Partition the nodes of G0 into n disjoint sets using min-cut graph partitioning (e.g., using METIS), and let V10; : : : ; Vn0 be the resulting vocabularies. Step 3. Extend each Vi0 to Vi? = Vi0 [ fr j r occurs in a schema triple in Gg. Step 4. Extend each Vi? to Vi = V ?
i [ fo j hs; p; oi 2 G and s 2 Vi?g.
Step 5. Calculate [G]Vi for each Vi, and set G = f[G]V1 ; : : : ; [G]Vn g.
In Step 1, we takes into account Assumption 5 that schema triples can be replicated in each partition element. Furthermore, in line with our Assumptions 3 and 4 on our query workload, we do not expect triples to participate in joins on classes and literals; thus, we remove such triples in Step 1 so that the min-cut graph partitioning algorithm does not attempt to place resources connected via class or literal resources into the same partition element.
In order to satisfy (4), we must ensure that, for each triple A 2 G, some Vi exists such that voc(A) 2 Vi. Thus, we must reintroduce the triples from G that correspond to edges in G0 that were `cut' during min-cut partitioning, as well as triples removed in Step 1. Thus, in Step 3 we introduce all resources occurring in the schema (including all classes and properties) into all partition elements; note that this ensures an e cient evaluation of queries mentioned in Assumption 2. Finally, since we assume that subject{subject joins are common (cf. Assumption 1), in Step 4 we reintroduce the missing triples into the partition element that contains the triple subject.
Partition element [G]Vi is the core owner of a resource r if r 2 Vi0. Note that, if [G]Vi is the core owner of a resource r, then [G]Vi contains all triples in which r occurs in the subject position. Hence, if Q is a star query in which variable ?x participates in subject{subject joins, then all answers in which ?x is mapped to r can be obtained by evaluating Q in [G]Vi ; in other words, all answers to star queries are local, which is in line with our Aim 1. 3.3
To make our scheme clear, we present an extended example. Let G be the RDF graph containing the following eight triples, shown schematically in Figure 1a.
G = f ha; R; bi; hb; R; ci; hb; R; di; hd; R; f i; he; R; di;
hf; R; ai; hf; R; ei; hb; rdf :type; si; he; rdf :type; tig To produce a 2-partition of G, in Step 1 we rst remove all triples containing class and literal resources; in our example, we remove triples hb; rdf :type; si and hd; rdf :type; ti. In Step 2 we then apply min-cut graph partitioning to the resulting graph to split the resources into two sets while minimising the number of cut edges; let us assume that this produces the following vocabularies: V10 = fa; b; cg
V20 = fd; e; f g In Steps 3 and 4 we then extend these vocabularies so that each partition element that is a core owner of a subject also contains all triples with that subject, and (5) (6) (a) Example Graph G
(b) Partition Element [G]V1 (c) Partition Element [G]V2 so that each partition element includes all class and property resources; in our example, this produces the following vocabularies:
V1 = fa; b; c; d; s; t; R; rdf :typeg
V2 = fa; d; e; f; s; t; R; rdf :typeg (7)
Due to this step, nodes a, d, s and t as duplicated in V1 and V2; we explain using node d why this is necessary. Graph [G]V1 must contain all triples whose subject is in V10; thus, since hb; R; di is in G and b is in V10 , we must add d to V1. Finally, we construct [G]V1 and [G]V2 as shown in Figures 1b and 1c. 4
Although there is considerable variation in the details, existing query answering
schemes, such as [
Q = h?x; rdf :type; si ^ h?x; R; ?yi ^ h?z; R; ?xi The data partitioning scheme critically governs the rst step. For example, if the data partitioning scheme guarantees that subject{subject joins can be evaluated locally, then the query must be broken up into pieces each of which involves only subject{subject joins; thus, query Q will be broken into the following pieces: Q1 = h?x; rdf :type; si ^ h?x; R; ?yi
Q2 = h?z; R; ?xi If the data partitioning scheme employs n-hop duplication, one can break the query into pieces that involve joins with n hops; however, as we show in Section 5, duplication can incur a considerable storage overhead.
The main drawback of such approaches is that they do not take advantage of local answers. For example, answer = f?x 7! b; ?y 7! d; ?z 7! bg is local with (8) (9) respect to the partition [G]V1 shown in Figure 1b, but it would not be retrieved by evaluating Q on [G]V1 directly; instead, one must evaluate Q1 and Q2 on [G]V1 and then join the results. This can be problematical since evaluating query pieces might be less e cient than evaluating the entire query at once.
Our query answering scheme uses a completely di erent approach. Roughly speaking, we rst evaluate each query in each partition element independently, thus retrieving all local answers without any partial query evaluation or communication between the servers. However, in this step we may also retrieve answers containing the wildcard resource, each of which represents a potential match of the query across partition elements, so we join such answers to obtain all answers to the query. In this way, we restrict communication and partial query evaluation to (possible) non-local answers, rather than all answers. 4.1
To formalise our query answering scheme, we rst introduce some notation. Let V be a vocabulary not containing . For Q a conjunctive query of the form (1), let [Q]V = [A1]V ^ : : : ^ [Am]V . Furthermore, for a variable assignment, let [ ]V be the variable assignment such that dom([ ]V ) = dom( ) and, for each variable x 2 dom( ), we have [ ]V (x) = (x) if (x) 2 V , and [ ]V (x) = if (x) 62 V .
In the rest of this section, we x an arbitrary conjunctive query Q of the form (1) with m atoms, an arbitrary RDF graph G, and an arbitrary partition G = ([G]V1 ; : : : ; [G]Vn ) of G. To evaluate Q in G, we rst evaluate [Q]Vi in [G]Vi for each 1 i n. Note that query Q may contain resources not contained in Vi; therefore, in each partition element [G]Vi we evaluate [Q]Vi , rather than Q. We then join all answers obtained in the previous step, while assuming that resource matches any other resource. We formalise the join procedure as follows. De nition 1. A variable assignment is a join of assignments 1 and 2, written = 1 ./ 2, if dom( ) = dom( 1) = dom( 2) and, for each x 2 dom( ), (i) 1(x) = 2(x) implies (x) = 1(x) = 2(x), and (ii) 1(x) 6= 2(x) implies 1(x) = and (x) = 2(x), or 2(x) = and (x) = 1(x).
An assignment can be an answer to Q on G only if it instantiates all atoms of Q, which we formalise as follows.
De nition 2. Let be a variable assignment with dom( ) = var(Q). The set of valid atoms of Q under is de ned as val (Q) = fj j 62 voc( (Aj ))g. Moreover, is valid for Q if jval (Q)j = m.
Furthermore, when evaluating [Q]Vi in [G]Vi , we can ignore any variable assignment 1 2 ans([Q]Vi ; [G]Vi ) that is redundant according to the following de nition. Intuitively, 1 is redundant if, for each 2 ans(Q; G) that `extends' 1 (i.e., such that 1 = [ ]Vi ), there exists a variable assignment 2 obtained by evaluating [Q]Vj in some partition element [G]Vj such that extends 2, and the set of atoms of Q fully instantiated by 2 strictly includes the set of atoms fully instantiated by 1. Note that this includes the case where no 2 ans(Q; G) extends 1. This idea is formally captured using the following de nition. De nition 3. Consider arbitrary 1 i n and 1 2 ans([Q]Vi ; [G]Vi ). Assignment 1 is redundant for Q and i if, for each assignment 2 ans(Q; G) such that 1 = [ ]Vi , there exist 1 j n and an assignment 2 2 ans([Q]Vj ; [G]Vj ) such that 2 = [ ]Vj and val 1 ([Q]Vi ) ( val 2 ([Q]Vj ). If such 1 is neither valid for Q nor redundant for Q and i, then 1 is a partial match of Q in [G]Vi .
As a simple consequence of De nition 3, note that the number of nonredundant answers in each partition element is at most equal to the number of non-local query answers. Theorem 1 captures the essence of our query answering scheme. Intuitively, it says that each answer to Q on G is obtained either by evaluating [Q]Vi on some partition element [G]Vi (i.e., it is a local answer), or by joining non-valid and non-redundant assignments 1; : : : ; n obtained by evaluating [Q]Vi on [G]Vi that instantiate all atoms of Q.
Theorem 1. For a variable assignment , we have 2 ans(Q; G) if and only if 1. is valid for Q and 2 ans([Q]Vi ; [G]Vi ) for some 1 i n, or 2. variable assignments 1; : : : ; n exist such that (a) for each 1 i n, either dom( i) = var(Q) and rng( i) = f g, or we have i 2 ans([Q]Vi ; [G]Vi ) and i is a partial match of Q in [G]Vi , (b) for each 1 j m, some 1 k n exists such that j 2 val k ([Q]Vk ), and (c) = 1 ./ : : : ./ n.
Proof. ()) Assume that 2 ans(Q; G). The claim holds trivially if satis es (1), so assume that does not satisfy (1). For each 1 i n, let i = [ ]Vi , and let i be such that dom( i) = var(Q) and rng( i) = f g if i is redundant for Q and i = i otherwise. We next show that each i satis es (2a){(2c).
(2a) Consider an arbitrary 1 i n. The claim holds trivially if i is redundant for Q and i, so we assume that i = i is not redundant for Q and i. Since we assume that 62 ans(Q; G), assignment i is not valid for Q. For each 1 j m, since (Aj) 2 G, we clearly have [ (Aj)]Vi 2 [G]Vi ; furthermore, we have [ (Aj)]Vi = i([Aj]Vi ), so i([Aj]Vi ) 2 [G]Vi holds. Consequently, we have i 2 ans([Q]Vi ; [G]Vi ), as required.
(2b) Consider an arbitrary 1 j m; then, 2 ans(Q; G) clearly implies (Aj) 2 G. Since G Si[G]Vi , some 1 i n exists such that i(Aj) 2 [G]Vi , so clearly j 2 val i ([Q]Vi ). Now choose 1 k n such that val k ([Q]Vk ) is a largest set satisfying val i ([Q]Vi ) val k ([Q]Vk ). Since val k ([Q]Vk ) is largest, such k is not redundant for Q and k, so k = k; but then, j 2 val k ([Q]Vk ), as required.
(2c) For each variable x 2 dom( ), some 1 i n exists such that (x) 2 Vi; hence, it is obvious that = 1 ./ : : : ./ n holds. We next show that we can successively replace in this equation each i that is redundant for Q and i with i. To this end, choose an arbitrary i that is redundant for Q and i, and choose an arbitrary 1 j n such that val i ([Q]Vi ) val j ([Q]Vj ) and j is not redundant for Q and j; clearly, we have [ i]Vj ./ j = j. Now consider an arbitrary variable x 2 dom( i) such that i(x) 6= and i(x) 6= j(x). Since val i ([Q]Vi ) val j ([Q]Vj ) holds, variable x occurs in Q only in atoms A` such that 2 voc( i(A`)), so ` 62 val i ([Q]Vi ). But then, by (2b), some 1 k n exists such that k(x) = i(x) and k is not redundant for Q and k. But then, = 1 ./ : : : i 1 ./ i ./ i+1 ./ n clearly holds. We can iteratively replace in this equation each i that is redundant for Q and i with i without a ecting the equality, as required.
(() Assume that (1) is true for some ; then ([Aj]Vi ) = (Aj) for each 1 j m, so clearly 2 ans(Q; G). Assume now that (2a){(2c) are true, and consider an arbitrary 1 j m. By (2a) and (2b), some 1 i n exists such that i([Aj]Vi ) 2 [G]Vi and 62 voc( i([Aj]Vi )). But then, i([Aj]Vi ) 2 G; furthermore, by (2c), i(x) = (x) for each variable x 2 Vj, so (Aj) 2 G. Consequently, 2 ans(Q; G). tu
Hence, the answers to a query Q over G can be computed as follows. First, each i-th server computes ans([Q]Vi ; [G]Vi ) in parallel, and it immediately returns all answers that are valid for Q. Second, the server identi es a subset Pi ans([Q]Vi ; [G]Vi ) of partial matches of Q in [G]Vi , and it also extends Pi with assignment such that dom( ) = var(Q) and rng( ) = f g. Third, all servers communicate Pi to one designated server, which then computes the join P1 ./ : : : ./ Pn and returns each result that instantiates all atoms of Q. Our query answering scheme thus requires distributed computation only for answers spanning partition boundaries.
Checking whether some 1 2 ans([Q]Vi ; [G]Vi ) is redundant for Q and i requires one to consider each 2 ans(Q; G), which is clearly impractical. Thus, in this section we present an approximate redundancy check. Note that Theorem 1 holds even if some i in Condition (2a) is redundant, so using an approximate check is safe from the correctness point of view.
Our optimisation is based on the fact that our data partitioning scheme ensures that answers to subject{subject joins are always local. Hence, if, for some , each `star' in Q contains an atom that is not valid for , then is redundant. This is captured formally in Proposition 1; its proof is trivial, so we omit it for the sake of brevity.
Proposition 1. Consider an arbitrary 1 i n and an arbitrary variable assignment 2 ans([Q]Vi ; [G]Vi ). Then, is redundant for Q and i if, for each term s 6= occurring in ([Q]Vi ) in a subject position, an atom A 2 Q exists such that s occurs in the subject position of ([A]Vi ) and 2 voc( ([A]Vi )). 4.3
Practical applicability of our approach depends critically on e ective removal of redundant answers. As we show in Section 5, the optimisation from Proposition 1 is e ective on some, but not on all queries. The latter is often the case for long chain queries (i.e., queries of the form hx0; R1; x1i ^ : : : ^ hxn 1; Rn; xni): in each partition element, the wildcard resource typically has a large fan-out and fan-in, and it also occurs in triples of the form h ; Ri; i, which can give rise to a large number of answers that are not redundant.
As a possible remedy, we shall explore the possibility of adapting the
approach presented in [
In this section we experimentally evaluate our approach using the Lehigh
University Benchmark (LUBM) and the SPARQL Performance Benchmark (SP2B).
Each test dataset was split into a partition of size 20, and we used the queries
available in the respective benchmarks. This size was chosen to make it directly
comparable to related works such as [
For each G = (G1; : : : ; G20) of a test dataset G, we calculated (i) the percentage of local answers to test queries, (ii) the storage overhead|that is, the percentage jG1j+:::+jG20j jGj , and (iii) the number of partial matches to test jGj queries, according to Proposition 1. While experiment (i) determines how many non-local answers must be constructed, experiment (iii) provides us with an indication of how much work is required for this construction. This is critical because, in order to ensure the completeness of query answers, all partial matches in all partition elements must be computed and joined together.
We compared our approach with subject-based hash partitioning (written
Hash) as in [
The Lehigh University Benchmark (LUBM) [
?w ub:undergraduateDegreeFrom ?z . ?w ub:advisor ?x } Some LUBM queries have non-empty results only if the data is extended according to the axioms from the LUBM ontology; however, since distributed reasoning is out of scope of this paper, we rewrote the test queries into unions of conjunctive queries in order to take the ontology axioms into account.
The SPARQL Performance Benchmark (SP2B) [
Table 1 shows the percentage of local answers for each LUBM query. RDFox and SHAPE were able to answer all queries completely, which is in part due to the modular nature of the data; however, hashing performs poorly on all queries. Table 2 shows the results for SP2B. Again, hashing performs very poorly. Furthermore, both RDFox and SHAPE handled queries 4 and 6 well; however, RDFox signi cantly outperformed SHAPE on queries 5, 7, and 8.
One can intuitively understand these results as follows. Hashing by subject,
although e ective for star queries, performs very poorly for other types of query:
in most cases, it provides almost no local answers. Thus, hashing is likely to be a
poor data partitioning scheme for applications with diverse query loads. SHAPE
considerably improves hashing, to the extent that only two benchmark queries
are problematic. However, by partitioning the data based on its structure, one
can further improve the overall performance: our approach is weakest on query
Q5 from SP2B, but it still provides a high percentage of local answers.
As we have already discussed, the percentage of local answers can be increased
using n-hop duplication, but at the expense of storage overhead. For example,
with 2-hop duplication, the approach by [
Table 3 shows the overhead for all partitioning schemes and data sets we considered in our experiments. Hashing clearly incurs no overhead; moreover, although SHAPE incurs a considerable overhead, that can be acceptable for some applications. Our partitioning scheme, however, exhibits a negligible overhead. Intuitively, this is due to the fact that min-cut graph partitioning tries to minimise the number of cut edges, which leads to a small level of duplication. We evaluated each test query on each partition element, and we discarded all valid assignments and some redundant assignments (according to Proposition 1). For each query, we computed the mean, minimum, maximum, and the sum of the numbers of partial matches across all partition elements.
On LUBM, queries 2, 8, 11 and 12 had no partial matches, so they can be evaluated fully locally without the need for any distributed processing. Queries 9 and c had 6 and 11, respectively, partial matches in total, so the necessary distributed processing is negligible.
On SP2B, evaluating queries 4 and 8 on all partition elements did not nish within an hour, producing very large numbers of partial matches. Since the number of partial matches in each partition element is bounded by the number of non-local answers and the latter is small (cf. Table 2), this result shows that Proposition 1 is not very e cient in identifying redundant answers for queries 4 and 8. Table 4 summarises the results for the remaining queries; in order to better understand these numbers, the table also shows the numbers of total and non-local answers. For queries 5 and 7, the numbers of partial matches are much smaller than the numbers of non-local answers, suggesting that joining the partial matches should be practically feasible. In contrast, the number of partial Query Q5 Q6 Q7 matches to query 6 is orders of magnitude larger than the number of non-local answers, suggesting that joining the partial answers might be di cult.
To summarise, our approach produces no or few partial matches on many types of query, but it runs into problems with long chain queries such as SP2B query 8. We shall try to improve on this using the ideas outlined in Section 4.3. 6
We have presented a new scheme for partitioning RDF data across a cluster
of shared-nothing servers. Our main goal is to minimise the number of
connections between partition elements so as to ensure that most answers to typical
queries are local (i.e., they can be obtained by evaluating the query locally in
all partition elements). We encode in each partition element links to other
partition elements, and we use this information in a novel query answering scheme
to correctly compute all answers to queries. Unlike existing systems, our query
answering scheme retrieves all local answers by simply evaluating the query in
each partition element, and it uses the encoded links to reduce the need for
distributed processing. We have shown that, on the LUBM and SP2B benchmarks,
test queries have more local answers under our data partitioning scheme than
with subject-based hashing or semantic partitioning [
We see two main challenges for our future work. First, we shall try to adapt
the graph exploration technique by [
Acknowledgements Our work was supported by a doctoral grant by Roke Manor Research Ltd, an EPSRC doctoral training grant, and the EPSRC project MaSI3.