Optimizing SPARQL queries over decentralized knowledge graphs

2.1. Client-server architectures

SPARQL endpoints are Web services providing an HTTP interface that accepts SPARQL queries and remain some of the most popular interfaces for querying RDF data on the Web. However, studies [12,67] have found that such endpoints are often unavailable and experience downtime.

Linked Data Fragment (LDF) interfaces, such as Triple Pattern Fragments (TPF) [68], attempt to increase the availability of the server by shifting some of the query processing load towards the client while the server only processes requests with low time complexity. For instance, TPF servers only process individual triple patterns while the TPF clients process joins and other expensive operations. Today, several TPF clients exist that rely on either a greedy algorithm [68], a metadata based strategy [36], or star-shaped query decomposition combined with adaptive query processing techniques [1] to determine the join order of the triple patterns in a query. However, while in all these approaches the server can handle more concurrent requests in comparison to SPARQL endpoints without becoming unresponsive, TPF naturally incurs a large network overhead when processing queries since intermediate bindings from previously evaluated triple patterns are transferred along with subsequently evaluated triple patterns to limit the amount of intermediate results, one by one. Furthermore, studies found that the performance of TPF is heavily affected by the type of triple pattern (i.e., the position of variables in the triple pattern) [34] and the shape of the query [50,51].

Several different systems have since been proposed to lower the network overhead. For instance, Bindings-Restricted TPF (brTPF) [32] bulks bindings from previously evaluated triple patterns such that multiple bindings can be attached to a single request. While this reduces the number of requests made for a triple pattern, it still incurs a somewhat large data transfer overhead, since each request still evaluates a single triple pattern. hybridSE [49] combines a brTPF server with a SPARQL endpoint and takes advantage of the strengths of each approach; subqueries with large numbers of intermediate results are sent to the SPARQL endpoint to overcome the limitations posed by LDF systems. However, hybridSE often answers complex queries using the SPARQL endpoint and is thus vulnerable to server failure.

To further limit the network overhead, Star Pattern Fragments (SPF) [3] clients send conjunctive subqueries in the shape of stars (star patterns) to the server and process more complex patterns locally on the client. Such conjunctive subqueries can be processed relatively efficiently by the server [58], which results in the transfer of significantly fewer intermediate results than in systems like TPF and brTPF. On the other hand, Smart-KG [14] ships predicate-family partitions (i.e., characteristic sets) to the client and processes the entire query locally; however, triple patterns with infrequent predicate values (according to a certain threshold) are sent to and evaluated by the server. While this takes advantage of the distributed resources that the clients possess, Smart-KG often ends up transferring excessive amounts of data unnecessarily since entire partitions of a dataset are transferred regardless of any bindings from previously evaluated star patterns. WiseKG [13] combines SPF and Smart-KG and uses a cost model to determine which strategy (SPF or Smart-KG) is the most cost-effective to process a given star-shaped subquery. Like SPF and Smart-KG, WiseKG processes more complex patterns on the client. Nevertheless, all the aforementioned LDF approaches rely on a centralized server or a fixed set of servers that are subject to failure.

Lastly, different from LDF approaches, SaGe [48] decreases the load on the server by suspending queries after a fixed time quantum to prevent long-running queries from exhausting server resources; the queries can then be restarted by making a new request to the server. However, SaGe processes entire, and possibly complex, queries on the server, and as stated above, such servers are subject to failure.

2.2. Federated systems

Federated systems enable answering queries over data spread out across multiple independent SPARQL endpoints [2,18,26,40,64] or LDF servers [33] offering access to different datasets. While such approaches spread out query processing over several servers, lowering the load on each individual server, they sometimes generate suboptimal query execution plans that increase the number of intermediate results and the load on individual servers [43]. As such, several approaches [30,38,52,53,62,66] have attempted to optimize federated queries in different ways. For instance, [64] builds an index over time by remembering which endpoints in the federation can provide answers to which triple patterns. Furthermore, [53] decomposes queries into subqueries that can be evaluated by a single endpoint. While [53] uses a similar query decomposition strategy as Lothbrok, they target federations over SPARQL endpoints, and as previously mentioned, such endpoints suffer from availability issues. On the other hand, [52,62] estimate the selectivity of joins to produce more efficient join plans. For instance, [52] uses characteristic sets [55] and pairs [28] to index the data in the federation and combines this with Dynamic Programming (DP) to optimize query execution plans. Furthermore, [33] proposes an interface for processing federated queries over heterogeneous LDF interfaces. To achieve this, the query optimizer is adapted to the characteristics of the different interfaces as well as the locality of the data, i.e., knowledge of which nodes hold which data. Inspired by these approaches, Lothbrok fragments knowledge graphs based on characteristic sets and uses a similar cardinality estimation technique to optimize join plans in consideration of data locality in the network.

2.3. Peer-to-peer systems

Peer-to-Peer (P2P) systems [4,6,17,24,44,45,47] tackle the availability issue from a different perspective: by removing the central point of failure completely and replicating the data across multiple nodes in a P2P network, they can ensure the data remains available even if the original node that uploaded the data fails. As such, they consist of a set of nodes (often resource limited) that act both as servers and clients, maintaining a limited local datastore. The structure of the network, i.e., connections between the nodes, as well as data placement (data allocation), varies from system to system. For instance, some systems [17,44,45] enforce data placement by applying a structured overlay over the network, such as Dynamic Hash Tables (DHTs) [46]. On the other hand, Piqnic [4] imposes no structure on top the network; nodes are connected randomly to a set of neighbors that are shuffled periodically with another node’s neighbors to increase the degree of joinability between the fragments of neighboring nodes. Lastly, ColChain [6] extends Piqnic and divides the entire network into smaller communities of nodes that collaborate on keeping certain data available and up-to-date. By applying community-based ledgers of updates and relying on a consensus protocol within a community, ColChain lets users actively participate in keeping the data up-to-date.

Each P2P system has different ways of processing queries. For instance, due to the lack of global knowledge over the network, basic P2P systems have to flood the network with requests for a given horizon to increase the likelyhood of receiving complete query results. To counteract this, distributed indexes [5,20,66] like Prefix-Partitioned Bloom Filter (PPBF) indexes [5] determine which nodes may include relevant data for a given query and thus allow the system to prune nodes from consideration during query optimization. Yet, the aforementioned systems still experience a significant overhead partly caused by inaccurate cardinality estimations, query optimization that does not consider the locality of data, as well as data fragmentation that splits up closely related data. For instance, Piqnic and ColChain both use a predicate-based fragmentation strategy that creates a fragment for each predicate. This, together with the replication and allocation strategy used, means that data relevant to a single query is distributed over a significant number of fragments and nodes.

However, while an approach that maximizes the degree to which entire queries can be processed by one node can lower the communication overhead, distributing some of the query processing load across multiple nodes is equally important when optimizing queries in a decentralized context [7] to avoid overloading individual nodes. As such, Lothbrok introduces a query optimization technique that distributes the processing of subqueries to nodes in the network based on data locality and fragment compatibility, while the characteristic set fragmentation technique allows entire star-shaped subqueries to be processed on the same node.

3.1. Peer-to-peer

In its simplest form, an unstructured P2P system consists of a set of interconnected nodes that all maintain a local datastore managing a set of (partial) knowledge graphs, where each node maintains a local view over the network, i.e., a set of neighboring nodes and nodes reachable from those neighbors within a certain number of steps (also known as hops), called the horizon of a node.

Formally, we define a P2P network N as a set of interconnected nodes N = { n 1 , … , n n } where each node maintains a local datastore and a local view over the network. The data uploaded to a node in N is replicated throughout the network. Furthermore, in line with previous work [5,6], each node maintains a distributed index describing the knowledge graphs reachable within its horizon. A node n is defined as follows:

While maintaining the structure of the network is important for P2P systems, it is not relevant for the data and query processing techniques that this paper is focusing on. As such, we do not go into detail on network topology, data replication and allocation, and periodic shuffles. Instead, we refer the interested reader to related work such as [4,6] for more details. In the following, we define data fragmentation and introduce a running example.

In line with previous work [4,6], and to avoid having to replicate large knowledge graphs throughout the network, Lothbrok divides knowledge graphs into smaller disjoint fragments, i.e., partial knowledge graphs, which can be replicated more easily. Fragments can be obtained using a fragmentation function. A fragmentation function is a function that, given a knowledge graph, returns a set of disjoint fragments, and is formally defined as follows:

Different fragmentation functions can have different granularities. For instance, the most coarse-granular fragmentation function is F C G ( G ) = { G } , i.e., the fragmentation function does not split up the original knowledge graph. ColChain [6] as well as Piqnic [4] use a predicate-based fragmentation function for G , i.e., F P ( G ) = { { ( s ′ , p ′ , o ′ ) ∣ ( s ′ , p ′ , o ′ ) ∈ G ∧ p ′ = p } ∣ ∃ s , o : ( s , p , o ) ∈ G } , which creates a fragment for each unique predicate in G . Lothbrok uses a fragmentation function based on characteristic sets [55] (i.e., predicate families) that is detailed in Section 4.2.

The fragments created by the fragmentation function are replicated and allocated at multiple nodes in the network to ensure availability in case the original provider of the knowledge graph becomes unavailable and to enable load balancing. The replication and allocation factor are parameters of the underlying network; for instance, in Piqnic [4], fragments are replicated and allocated across the node’s neighbors, and nodes index all fragments available within a certain horizon. On the other hand, ColChain [6] replicates and allocates fragments at nodes that participate within the same communities. Since this paper focuses on data fragmentation and query optimization, we omit details on data replication and allocation and refer the interested reader to related work [4,6] for details.

OPEN IN VIEWER

Fig. 2.

(a) Example of an unstructured P2P network N = { n 1 , … , n 5 } and (b) architecture of a single node n 5 that indexes data within a horizon of 2 nodes.

Consider, as a running example, the unstructured P2P network in Fig. 2(a) consisting of five nodes ( N = { n 1 , … , n 5 } ) that replicate a total of five fragments ( f 1 , … , f 5 ). In this example, each node maintains a set of two neighbors and each fragment is replicated across two nodes. For instance, node n 5 has { n 2 , n 4 } as its set of neighbors, and replicates the fragments { f 2 , f 4 , f 5 } in its local datastore. While the running example is based on an unstructured network, such as the one presented in [4], Lothbrok could be adapted to more structured setups, such as the one presented in [6].

To speed up query processing performance, systems like Piqnic [4] and ColChain [6] use distributed indexes [5,20] to efficiently identify nodes holding relevant data for a given SPARQL query. The indexes capture information about the fragments stored locally at the node itself as well as information about fragments that can be accessed via its neighbors.

A distributed index, as defined in [5,6], is a structure that maps the triple patterns in a query to nodes that hold relevant fragments to those triple patterns. In line with [5,6], we thus define distributed indexes as follows.

Given a node n, n’s distributed index is denoted I n . Given the definition of a distributed index, we define a node mapping as a mapping from a triple pattern t in a BGP P to a set of nodes that contain relevant fragments to t, as follows:

To build the index for a node’s local view over the network, nodes share partial indexes, i.e., partial mappings, for the fragments that they have access to, called index slices. An index slice for a fragment is a partial mapping from triple patterns to the fragments that contain relevant triples to the triple patterns, as well as a mapping from the fragment to the nodes that replicate it, and is defined as follows:

Index slices for the fragments that a node has access to are combined into a distributed index for that particular node using the ⊕ operator.1

While the definition of distributed indexes allows for several different types of indexes, the index slices used in Piqnic [4] and ColChain [6] correspond to Prefix-Partitioned Bloom Filters (PPBFs) [5], which extend regular Bloom filters [15]. Given a set S of IRIs, a Bloom filter B for S is a tuple B = ( b ˆ , H ) where b ˆ is a bitvector of size m and H is a set of k hash functions [5]. Each hash function in H maps the elements from S (i.e., IRIs) to a position in b ˆ ; these positions are thus set to 1 whereas the positions not mapped to by a function in H are 0. In other words, index slices in [5] represent the set of entities in a fragment as bitvectors following the approach described above. Looking up whether an element e is in S using the Bloom filter for S is done by hashing e using the hash functions in H and checking the value of each position in b ˆ . If at least one of those positions is set to 0, it is certain that e ∉ S . However, if all corresponding bits are set to 1, it is not certain that e ∈ S , since it could be a false positive caused by hash collisions, i.e., different values are mapped to the same positions in the underlying bitvector. In this case, we say that e may be in S , denoted e ∃ S .

To check the compatibility of two fragments relevant for conjunctive triple patterns, we check whether or not they produce any join results. To do this, we could check whether or not the intersection of the bitvectors describing the subjects and objects of the fragments is empty (i.e., if they have some IRI in common). Given two Bloom filters B 1 = ( b ˆ 1 , H ) and B 2 = ( b ˆ 2 , H ) , the intersection of B 1 and B 2 is approximated by the logic AND operation between b ˆ 1 and b ˆ 2 , B 1 ∩ B 2 ≈ b ˆ 1 & b ˆ 2 .

To avoid exceedingly large bitvectors, PPBFs partition the bitvector based on the prefix of the IRIs. The prefix of an IRI u corresponds to the IRI of the namespace of u.2

Consider the example where the IRI dbr:Copenhagen is inserted into a PPBF, visualized in Fig. 3(a). In this case, the IRI is matched to the prefix dbr, and the name Copenhagen is hashed using each hash function in the PPBF; each corresponding bit in the bitvector for the dbr prefix is thus set to 1.

OPEN IN VIEWER

Fig. 3.

Example of (a) inserting an IRI into a PPBF B 1 P and (b) intersection between two PPBFs B 1 P ∩ B 2 P  [5].

Like for regular Bloom filters, we say that an IRI i with prefix p and name n may be in a PPBF B P , denoted i ∃ B P , if and only if all positions given by h ( n ) such that h ∈ H are set to 1 in the bitvector θ ( p ) . PPBFs are used by Piqnic and ColChain to prune non-overlapping fragments of joining triple patterns from the query execution plan (i.e., the match ( P , I ) function in Definition 7). This is done by finding the intersection of the two PPBFs to check whether or not they overlap; if the intersection of the two PPBFs is empty, the corresponding fragments do not produce any join results. The PPBF intersection is defined in [5] as follows.

Consider the example intersection visualized in Fig. 3(b). As described above, the intersection of two PPBFs is the bitwise AND operation on the bitvectors for the prefixes that B 1 P and B 2 P have in common. In this example, B 2 P does not have a bitvector with the prefix dbp, thus this partition is omitted from the intersection. Similarly, the bitvector partition with the dbo prefix is omitted. Since both PPBFs have bitvectors for the dbr prefix, the resulting PPBF has one partition for the dbr prefix that is the result of the bitwise AND operation between the two corresponding partitions in B 1 P and B 2 P .

Furthermore, given a partitioned bitvector B and b ˆ ∈ B . B ˆ , let t ( b ˆ ) be a function that returns the number of bits in b ˆ that are set. Then, the estimated cardinality of a partitioned bitvector B , denoted card P ( B ) , is the sum of the estimated cardinality for all bitvector partitions in B . B ˆ  [5,56] and is formally defined as follows: (2) card P ( B ) = ∑ b ˆ ∈ B . B ˆ ln ( 1 − t ( b ˆ ) / | b ˆ | ) | B . H | · ln ( 1 − 1 / | b ˆ | )

Consider, for instance, a PPBF B P such that | B P . H | = 5 and that | b ˆ | = 20000 for all b ˆ ∈ B P . B ˆ with two partitions, dbr and dbp. Since the partitioned bitvector has two partitions, obtaining the estimated cardinality for B P is the sum of estimating the cardinality of both prefix partitions. Let the number of set bits in the bitvector for the dbr prefix be 736 and the number of set bits in the bitvector for the dbp prefix be 249. Then, the estimated cardinality using Equation (2) is: card P ( B P ) = ln ( 1 − 736 / 20000 ) 5 · ln ( 1 − 1 / 20000 ) + ln ( 1 − 249 / 20000 ) 5 · ln ( 1 − 1 / 20000 ) ≈ − 0.0375 − 0.00025 + − 0.0125 − 0.00025 ≈ 150 + 50 ≈ 200

4.1. Design and overview

Lothbrok introduces three contributions, that altogether decrease the communication overhead and in doing so increases query processing performance. First, Lothbrok creates fragments based on characteristic sets such that entire star patterns can be answered by a single fragment. This is beneficial since, as we discussed in Section 1, such star patterns are relatively efficiently processed by the nodes [58] and reduce the communication overhead. The characteristic set of a subject value (entity) is the set of predicates that occur in triples with that subject. As such, Lothbrok creates one fragment per unique characteristic set and each fragment thus contains all the triples with the subjects that match the characteristic set of the fragment. Consider, for instance, the example network in Fig. 2 and query Q shown in Fig. 4(a). Figure 4(b) shows the characteristic sets of each fragment in the network. Using this fragmentation method, each fragment can provide answers to entire star patterns; for instance, P 3 ∈ S ( Q ) can be processed over just f 5 , since it is the only fragment containing triples with both predicates present in P 3 . The formal definition of the fragmentation approach is presented in Section 4.2.

OPEN IN VIEWER

Fig. 4.

(a) Example SPARQL query Q and (b) corresponding characteristic sets in the example network.

Second, to accommodate processing entire star patterns over individual fragments, and to encode structural information that can be used for cardinality estimation and locality awareness, Lothbrok introduces a novel indexing scheme, called Semantically Partitioned Bloom Filter (SPBF) Indexes, that builds upon the Prefix-Partitioned Bloom Filter (PPBF) indexes presented in [5]. In particular, SPBFs partition the bitvectors based on the IRI’s position in the fragment, i.e., whether it is a subject, predicate, or object. For instance, in the running example, the SPBF for f 5 contains a partition encoding all the subjects with the characteristic set {dbo:publisher,dbo:language}, as well as partitions encoding all the objects in f 5 that occur in a triple with each predicate. The formal definition of SPBF indexes is discussed in Section 4.3.

Third, Lothbrok proposes a query optimization technique that takes advantage of the fragmentation based on characteristic sets and the SPBF indexes to estimate cardinalities and consider data locality while optimizing the query execution plan. First, Lothbrok builds a compatibility graph using the SPBF indexes that describes, for a given query, which fragments are compatible with one another for each join in the query (i.e., which fragments may produce results for the joins). In other words, the nodes in a compatibility graph denote the relevant fragments, and the edges denote which fragments may produce join results with one another for the given query. Then, Lothbrok builds a query execution plan using a Dynamic Programming (DP) algorithm that considers the compatibility of fragments in the compatibility graph and the locality of the fragments in the index. The query execution plan built by the query optimizer follows a left-deep approach that uses the bindings obtained from previously evaluated subplans as input (filter) when processing joins. Furthermore, the plan obtained from this step includes all the relevant fragments (i.e., only non-relevant fragments are pruned).

OPEN IN VIEWER

Fig. 5.

Flow diagram of the contributions of Lothbrok (extended from Fig. 1).

Figure 5 shows an extended version of Fig. 1 using the principles explained in Section 3.1 and the contributions explained above. The query optimizer thus contains three sequential steps; (1) fragment selection, (2) compatibility graph, and (3) query planning, that altogether computes the query execution plan for a given query.

Notice that for star patterns P with a large number of solution mappings, [ [ P ] ] G could become computationally expensive to enumerate. However, as shown in our experimental evaluation in Section 7, Lothbrok actually decreases the number of intermediate results to be enumerated by up to two orders of magnitude compared to triple pattern-based query executors using the optimization techniques explained above. Furthermore, we apply pagination of large result sets, which related studies [32,68] have already shown can effectively limit the effects of a large number of solution mappings, even for star pattern-based query execution [3].

In the remainder of this section, we detail data fragmentation (Section 4.2) and indexing (Section 4.3) in Lothbrok. Section 5 details the query optimization approach used by Lothbrok.

As discussed in Section 1, star-shaped subqueries can be processed relatively efficiently over a fragment [58], thus they can also help achieving a better balance between reducing the communication overhead and distributing the query processing load [3,13,14]. To facilitate processing such star patterns on single nodes, we propose to fragment the uploaded knowledge graphs based on characteristic sets [13,14,55]. This is the first contribution as explained in Section 4.1, and corresponds to the CS Fragments step in Fig. 5. Formally, a characteristic set is defined as follows:

In other words, the characteristic set of a subject is the set of predicates (i.e., predicate combination) used to describe the subject, i.e., that occur in the same triples as the subject. For instance, if the triples (dbr:Denmark,dbo:capital,dbr:Copenhagen) and (dbr:Denmark,dbo:currency,dbr:Danish_Krone) are the only ones with subject dbr:Denmark, then this subject is described by the characteristic set {dbo:capital,dbo:currency}.

Characteristic sets were first introduced in [55], used for cardinality estimation and, in extension of that, join ordering. WiseKG [13] and Smart-KG [14] used the notion of characteristic sets for fragmentation of knowledge graphs in LDF systems to balance the query load between clients and servers. In this paper, we use characteristic set based fragments as an alternative to the purely predicate-based fragmentation used by for example Piqnic. We define the characteristic set based fragmentation function as follows:

That is, the characteristic set fragmentation function creates a fragment for each characteristic set in the knowledge graph. In the characteristic sets shown in Fig. 4(b), f 4 thus contains all triples of all subjects that are described by the characteristic set {dbo:capital,dbo:currency}.

Depending on the complexity of the knowledge graph, however, using fragmentation purely based on characteristic sets can quickly lead to an unwieldy number of fragments. In our experimental evaluation in Section 7, fragmenting the data from LargeRDFBench [61] using Equation (3) led to 181,859 distinct fragments most of which contain very few subjects. Consider, for instance, in the running example, the situation where the following five characteristic sets are found in the uploaded knowledge graph; for illustration purposes we have extended the notation with the number of subjects covered by each characteristic set: C S 1 = ( { dbo : nationality , dbo : author , dbo : deathDate } , 500 ) C S 2 = ( { dbo : nationality , dbo : author } , 550 ) C S 3 = ( { dbo : publisher , dbo : language } , 1000 ) C S 4 = ( { dbo : nationality , dbo : author , dbo : language } , 2 ) C S 5 = ( { dbo : nationality } , 1 )

The fragments in the above example are skewed with a significant difference between the largests and smallest fragments. For instance, a separate fragment is created for C S 4 even though it does not carry very much information because it describes only two subjects. As documented in previous studies [28,52,55], similar approaches using DP to optimize the join order often struggle when presented with a large number of sources since they generally have to compare the cost of each possible combination of the sources. Concretely, in our case, the DP algorithm presented in Section 5 has polynomial time complexity with the number of relevant fragments; in the worst case, the DP algorithm has to check compatibility between each pair of relevant fragments. This means that as the number of relevant fragments increases, the potential number of compatibility checks increases polynomially as well. Clearly, this could affect lookup time during query optimization.

To partially address the data skew issue, we merge fragments with infrequent characteristic sets into fragments with more frequent characteristic sets, similar to [55]. While such an approach potentially has the tradeoff that some of the information in the merged fragments is lost, which could in rare cases lead to incomplete query results, we did not encounter such incomplete results in our experimental evaluation (Section 7). After fragmenting datasets using Equation (3), we iteratively merge the fragments with the lowest number of distinct subject values into other fragments until the total number of fragments is below a threshold, or all fragments with fewer subject values than a threshold have been merged, using the following strategy. In our experiments (Section 7), we computed two sets of fragments for each dataset; one where the total number of fragments matched the number of predicate-based fragments, and one where all fragments with fewer than 50 subjects (determined empirically based on the data used in our experiments) were merged.

First, for infrequent fragments f 1 with characteristic set C S 1 where there exists another fragment f 2 with a more frequent characteristic set C S 2 , such that C S 1 ⊆ C S 2 , we merge f 1 into f 2 by adding the triples of f 1 to f 2 ; if there are multiple candidates for f 2 , we select the one with the smallest set of predicates, since that fragment has the fewest additional predicates. In the example above, for instance, since C S 5 ⊆ C S 2 , we merge C S 5 into C S 2 by adding the triples from f 5 to f 2 .

Second, the remaining fragments f, i.e., ones that cannot directly be merged into any frequent fragments, are split into a set of disjoint fragments, { f 1 , … , f n } , such that each f i ∈ { f 1 , … , f n } can be merged into other fragments with more frequent characteristic sets using the first step. This is done by continuously selecting the largest possible set of predicates that can be merged into other fragments, until no predicates are left. In the (rare) case that some predicates do not occur in any frequent fragments, we store the (small) fragments containing those predicates separately; however, this never happened in our experimental evaluation in Section 7. For instance, in the example above, since C S 4 is not a subset of any frequent characteristic set, we have to split f 4 into smaller fragments. As such, we first create a fragment f 4 ′ with the characteristic set {dbo:nationality,dbo:author}, since this is the largest subset of C S 4 that can be merged into other fragments (either f 1 or f 2 ). Then, we create a fragment f 4 ″ with the characteristic set {dbo:language}, since this is the only predicate left in the original fragment, and merge f 4 ′ with f 2 and f 4 ″ with f 3 according to the first step above.

The steps above are sequential, i.e., all fragments that can be merged into other fragments without splitting are merged according to the first step, whereas the remaining infrequent fragments afterwards have to be split and merged according to the second step. In the example above, we end up with the following fragments: C S 1 = ( { dbo : nationality , dbo : author , dbo : deathDate } , 500 ) C S 2 = ( { dbo : nationality , dbo : author } , 553 ) C S 3 = ( { dbo : publisher , dbo : language } , 1002 )

Clearly, the data skew caused by the fragmentation approach is affected by the structuredness and heterogeneity of the datasets. In Lothbrok, we logically expect well-structured datasets to perform well, while unstructured datasets should lead to a large number of fragments. This is also what we see in our experimental evaluation in Section 7, where a few very unstructured datasets in LargeRDFBench [61] (e.g., the DBPedia subset) caused a large number of fragments some of which have very similar characteristic sets, whereas more structured datasets (e.g., LinkedTCGA-E) resulted in fewer fragments (Table 3) with less similar characteristic sets. Furthermore, we were able to decrease the number of fragments in our experiments by up to two orders of magnitude. For LargeRDFBench specifically, we decreased the number of fragments from 181,859 to 2,160, which significantly reduces the number of compatibility checks in the DP algorithm as well; as shown in our experimental evaluation (Section 7), using the merging procedure presented above, we were able to achieve significantly improved performance compared to triple-pattern-based query processors.

Since the merging procedure described above has already been reasoned and documented by previous studies [28,52,55], and our experimental results (Section 7) are in line with those studies, we will not provide an in-depth analysis of the benefits and tradeoffs of the merging procedure. It is, however, a topic for future work. Furthermore, a complete analysis of the effects of graph complexity measures on the different fragmentation approaches and on the data skew incurred by each fragmentation approach, is a topic for future work.

4.3. Semantically partitioned bloom filter indexes

The updated fragmentation function described in Section 4.2 often results in fragments that contain characteristic sets with several predicates. However, as described in Section 3.2, the PPBF indexes from [5] encode the set of entities in a fragment without any regard for the position of the entity in the fragment or the connection between the subjects, predicates, and objects.

Hence, we propose a novel indexing schema called Semantically Partitioned Bloom Filters (SPBFs), which builds upon PPBF as baseline. This is the second contribution of Lothbrok described in Section 4.1. Specifically, SPBFs encode the subject values in a single prefix-partitioned bitvector, while there is one prefix-partitioned bitvector for each predicate in the fragment that encodes the objects occurring in triples with that predicate. This structural change in the index lets us do two things: (1) by checking the overlap of the partitioned bitvectors that correspond to the position of the join, we can more accurately determine whether or not fragments produce join results with one another, and (2) we can maintain the link between the subjects, predicates, and objects. As explained in Section 4.1, the SPBF indexes are the second contribution of Lothbrok, and corresponds to SPBF Index in Fig. 5. The change in indexing structure requires adjustments in the following query optimization procedure. This procedure is outlined in Fig. 5 and detailed in Section 5, and involves source selection based on the compatibility of the fragments and Dynamic Programming (DP).

Note that since Lothbrok fragments and indexes data based on characteristic sets, the query optimizer described in Section 5 decomposes queries into star patterns. The following description therefore does not mirror exactly the definitions from Section 3.2 [5], since SPBF indexes have to match entire star patterns to fragments rather than triple patterns.

Formally, an SPBF is defined as follows:

Given a fragment f, B S ( f ) describes the SPBF for f. For instance, in the running example (Fig. 4), the SPBF for f 2 , B S ( f 2 ) , contains a prefix-partitioned bitvector encoding all the subject values in f 2 , B S ( f 2 ) . B s , as well as a separate prefix-partitioned bitvector for each predicate encoding the object values for those predicates, i.e., the partition B S ( f 2 ) . Φ ( dbo:author) that describes the objects that are connected with the dbo:author predicate, and so on. The SPBF for f 2 is visualized in Fig. 6(b); Fig. 6 further visualizes the SPBFs for each fragment in the running example (Fig. 4).

OPEN IN VIEWER

Fig. 6.

SPBFs of the fragments in the running example from Fig. 4.

Similarly to PPBFs (Section 3.2), we say that an IRI i at position ρ ∈ { s , p , o } may be in an SPBF B S , denoted i ∃ ρ B S , if and only if i ∃ B S . B s if ρ = s , ∃ p ∈ B S . P : i ∃ B S . Φ ( p ) if ρ = o , or i ∈ B S . P if ρ = p . For instance, dbo : nationality ∃ p B S ( f 2 ) means that the IRI dbo:nationalitymay be in f 2 on the predicate position. Furthermore, B p ( B S ) denotes a function that computes and returns the prefix-partitioned bitvector that contains all predicates in B S . P .

To adapt the general definition of distributed indexes from Section 3.2 [5] to the characteristic set fragmentation and star-shaped query decomposition of Lothbrok, in the following, we change the definition of the I n . ν ( t ) function from Definition 6 to map entire star patterns to the relevant fragments rather than triple patterns. To do this, we formally define the relevantFragment ( P , f ) function as a binary function that, for a given star pattern P and fragment f, determines whether f is a relevant fragment to P, as follows.

Consider, as an example, the star pattern P 1 in the example query in Fig. 4(a) and the SPBFs for the fragments in Fig. 6. In this example, relevantFragment ( P 1 , f 1 ) = true , since both dbo : nationality ∃ p B S ( f 1 ) and dbo : author ∃ p B S ( f 1 ) , and so we say the f 1 is a relevant fragment to P 1 .

Note, that fragment relevance in Lothbrok is a binary decision, relevantFragment ( P , f ) returns either true or false, and is not based on a relevance score. However, extending our work to consider relevance scores could be interesting future work.

Given the definition of the relevantFragment ( P , f ) function above, we define an SPBF index as follows:

Consider again the running example in Fig. 4 and the example network and fragment distribution in Fig. 2(a). In this case, given the SPBF index I n 1 S that comprises the SPBFs from all the fragments in Fig. 6 (given that n 1 has a horizon of 2 hops), then I n 1 S . υ ( P 1 ) = { f 1 , f 2 } , since f 1 and f 2 both contain partitions with the dbo:nationality and dbo:author predicates. Furthermore, in this example, I n 1 S . η ( f 1 ) = { n 2 , n 4 } , since they are the nodes replicating f 1 in their local datastore.

Since Lothbrok, like Piqnic and ColChain, builds partial indexes, i.e., slices (cf. Section 3.2), for each fragment that are combined to form the node’s distributed index, we define an SPBF index slice similar to Definition 8 as follows:

In the running example, for instance, the SPBF slice for f 1 , s f 1 S , is the SPBF visualized in Fig. 6(a), and s f 1 S . υ ′ ( P 1 ) = { f 1 } , since relevantFragment ( P 1 , f 1 ) = true , as explained above. In other words, the SPBF slice describing a particular fragment is the SPBF obtained from the respective fragment. The function s S ( f ) finds the SPBF slice describing f.

Since the SPBF indexes presented in this section are more complex than the ones presented in [5], combined with the more complex query optimization technique outlined in Section 5, we expect a slightly increased query optimization overhead compared to existing approaches [5]. However, this overhead should be compensated with the increased query execution efficiency that is partially obtained from the usage of the SPBF indexes. In fact, this is also what our experimental evaluation in Section 7 shows. Nevertheless, a deeper analysis of this tradeoff using even more diverse real-world datasets and queries is part of our future work.

Like the fragmentation approach (Section 4.2), the indexes are structurally affected by graph complexity measures. Unstructured datasets can lead to skewed partitions where some partitioned bitvectors encode a large number of values. Nevertheless, a complete analysis of the effect of the graph complexity on the indexing approach is a topic for future work. In Section 5, we detail how SPBF indexes are used to optimize queries using cardinality estimations and the locality of the data.

5.1. Fragment and source selection

In order to prune fragments that do not contribute to the query result, as well as to determine subqueries that can be processed in parallel, Lothbrok builds a compatibility graph (Fig. 5), describing which of the relevant fragments are compatible for the given query, i.e., which fragments produce join results with one another for each join in the query. Specifically, two fragments are said to be compatible for a given query if the intersection of the corresponding SPBF partitions is non-empty, i.e., if the sets of entities represented by these partitions could have some common elements.

As an example, consider again the query Q and fragments from the running example in Fig. 4. In this case, { f 1 , f 2 } are the relevant fragments for P 1 and { f 3 , f 4 } are the relevant fragments for P 2 . Since P 1 and P 2 join on the ?country variable, we can check the compatibility of each combination of fragments for those star patterns, by checking the overlap of the partitioned bitvector on the object position for the dbo:nationality predicate for P 1 and the partitioned bitvector on the subject position for P 2 , since those are the positions of ?country in P 1 and P 2 . Figure 7 visualizes this computation for two of the fragment combinations. For instance, in Fig. 7(a), we see that { f 1 , f 4 } are compatible, since the intersection of the corresponding partitions is non-empty. On the other hand, { f 1 , f 3 } (Fig. 7(b)) are not compatible, since the intersection of the corresponding SPBF partitions is empty.

OPEN IN VIEWER

Fig. 7.

Checking compatibility of fragments for the join P 1 ⋈ P 2 in the example query Q (Fig. 2).

In other words, a compatibility graph is an undirected graph where nodes are the relevant fragments and edges describe the compatible ones. Structurally, a compatibility graph is thus defined as follows:

Consider again the running example in Fig. 4, and the example compatibility graph in Fig. 8(g). As we saw in Fig. 7, f 1 and f 3 are not compatible since they do not produce any join results for the example query, meaning the compatibility graph in Fig. 8(g) has no edge between those two fragments.

OPEN IN VIEWER

Algorithm 1

Compute the compatibility graph of a BGP over an SPBF index

In the following, we detail how Algorithm 1 computes the compatibility graph by going through the algorithm showing a step-by-step example of how the compatibility graph is built in the running example in Fig. 4 (visualized in Fig. 8). Recall the function B S ( f ) that returns the SPBF for a fragment f, and let vars ( P ) be a function that returns all the variables in a star pattern P. Furthermore, given an SPBF B S , a star pattern P, and a variable v, let B ( B S , P , v ) denote a function that returns (assuming v can only occur once in P) B S . B s if v is the subject in P, B S . Φ ( p ) if v is the object with predicate p, i.e., ( s , p , v ) ∈ P , or B p ( B S ) if v is a predicate in P. Algorithm 1 defines the G C ( P , I S ) function in lines 1–16 that computes a compatibility graph given a BGP P and SPBF index I S .

Figure 8 shows how Algorithm 1 builds the compatibility graph for query Q in Fig. 4(a). In the following, we go through each intermediate step of the algorithm, describing the intermediate compatibility graphs built in the process. First, the G C function selects the star pattern in S ( P ) with the lowest estimated cardinality in line 2 (cardinality estimation is detailed in Section 5.2). Assume in the running example, that P 2 is the star pattern with the lowest estimated cardinality (Section 5.2), and that it is therefore selected in line 2 as the first star pattern. Furthermore, assume that f 4 is only compatible with f 1 and f 3 is only compatible with f 2 .

Then, the relevant fragments for the selected star pattern are found using the I S . υ function from the SPBF index (Definition 15) and iterated over in the for loop in lines 5–8; for each of these fragments, the function calls the buildBranch ( P , I S , f , P ′ , P ϵ ) function in lines 17–30 that builds the (sub)graph starting from the current fragment. In the example, the loop in lines 5–8 iterates over { f 3 , f 4 } , since these are the fragments relevant for P 2 .

The buildBranch ( P , I S , f , P ′ , P ϵ ) function defines a recursive function that builds a sub-graph starting from a specific fragment and star pattern. In the first iteration in the running example (i.e., for f 3 ), buildBranch is called with P = P 1 ∪ P 3 , f = f 3 , and P ′ = P 2 as parameters. First, if P does not contain any star patterns that join with P ′ , i.e., if P ′ is the outer-most star pattern in the join tree or for a Cartesian product, the function returns the compatibility graph just containing f without any edges (lines 18–19). In the example, since P 1 joins with P 2 , the algorithm does not execute the statement in line 19.

Instead, the for loop in lines 21–29 iterates through the star patterns P ″ ∈ P that join with P ′ , i.e., star patterns that have at least one variable in common. For each fragment f ′ relevant for P ″ (again found using the SPBF index), the function checks the compatibility of f and f ′ for each join variable v in line 24, i.e., whether or not f and f ′ may produce join results for each join variable, by intersecting the corresponding partitioned bitvectors in B S ( f ) and B S ( f ′ ) . If the fragments may produce join results, a recursive call is made in line 25 with P = P − P ″ , f = f ′ , and P ′ = P ″ as parameters. In the example, the for loop in line 21 has only one iteration for P ″ = P 1 , i.e., the only star pattern in S ( P ) that joins with P 2 . Hence, the for loop in line 24 checks the compatibility of each fragment relevant for P 1 ( f 1 and f 2 ) with f 3 (since f = f 3 in this call to the function). Since f 2 is compatible with f 3 (cf. the join cardinalities in Table 1), a recursive call is made in line 25 with P = P 3 , f = f 2 , and P ′ = P 1 .

OPEN IN VIEWER

Fig. 8.

Recursively building the compatibility graph for the query in Fig. 4(a) by applying Algorithm 1 resulting in G C ( Q , I n 1 S ) . Yellow nodes denote the fragments relevant for P 2 , blue nodes the fragments relevant for P 1 , and the green nodes the fragments relevant for P 3 .

Since P 3 joins with P 1 , the for loop in line 24 checks the compatibility of f 5 and f 2 and makes another recursive call to the function in line 25 with P = ∅ , f = f 5 , and P ′ = P 3 . In this iteration of the function, P is empty, thus the graph ( { f 5 } , ∅ ) is returned in line 19. This graph is visualized in Fig. 8(a) and contains only f 5 with no edges. Since this compatibility graph is non-empty, it is added to the output graph in lines 26–28 together with f 2 (since f = f 2 in this iteration of buildBranch) and the edge between f 5 and f 2 . This graph is visualized in Fig. 8(b) and returned by the current iteration of the buildBranch function. Upon receiving the graph in Fig. 8(b), the function adds f 3 (since f = f 3 in the current iteration) and an edge between f 2 and f 3 in lines 26–28, resulting in the compatibility graph shown on Fig. 8(c) that is returned in line 30.

In the next iteration of the for loop in line 5, the buildBranch is called with P = P 1 ∪ P 3 , f = f 4 , and P ′ = P 2 . Following the same procedure as described above for f 3 , we first build the subgraph containing only f 5 shown in Fig. 8(d). Then, f 1 is added to the graph along with an edge between f 1 and f 5 (since they produce join results), resulting in the subgraph shown in Fig. 8(e). Next, f 4 is added along with an edge between f 4 and f 1 , resulting in the compatibility graph for f 4 shown in Fig. 8(f). After merging this in lines 7–8 with the compatibility graph in Fig. 8(c), the resulting compatibility graph can be seen in Fig. 8(g).

The if statement in lines 9–15 ensures that subqueries with star patterns that do not join (i.e., in the case of Cartesian products) are included in the compatibility graph. This is done by keeping track of the considered star patterns in P using the accumulator P ϵ defined in line 3 and updated in line 29. The example query contains no Cartesian products and so the compatibility graph on Fig. 8(g) is returned by the algorithm.

The output of Algorithm 1 in the example is the compatibility graph shown in Fig. 8(g), specifying that f 1 is compatible with { f 4 , f 5 } and f 2 is compatible with { f 3 , f 5 } .

Algorithm 1 and the definition of SPBF indexes (Definition 15) ensure that pruned fragments do not contribute to the query result, i.e., that our pruning method does not miss any potential results. This follows from the theorem:

A high-level sketch of the proof of Theorem 1 follows. Algorithm 1 only prunes fragments when the condition in line 24 is false. Furthermore, this condition is only false when the bitvectors for the two fragments do not overlap; if there is any kind of overlap, even on a single bit, the condition is true. If the two fragments contain a common value, then by definition this value is mapped to the same positions in the corresponding bitvectors, and they will overlap at least on those bits. Hence, Theorem 1 holds, and we do not miss any potential results.

In Section 4.2 we have described how Lothbrok fragments knowledge graphs based on characteristic sets. Furthermore, in Section 4.3 we described how SPBF indexes connect the objects in a fragment to the predicates they occur in triples with. Since the SPBF of a fragment includes partitioned bitvectors describing the subjects and objects (Definition 15), we can estimate the number of values within these partitioned bitvectors and use those estimations to obtain cardinality estimations in a similar way as [52,55].

Note that the cardinality estimation technique presented in this section is used by the Dynamic Programming (DP) algorithm (Section 5.3) to find the cheapest costing query execution plan including all the relevant fragments, and is not used to rank relevant fragments according to the cardinalities. Recall the card P ( B ) function from Equation (2). Table 1 then shows the estimated cardinalities of each partitioned bitvector in the running example.

To estimate the cardinality of star-shaped subqueries, we utilize the fact that the subjects are described by a single partitioned bitvector. For a star-shaped subquery asking for the set of unique subject values described by a given set of predicates (i.e., queries with the DISTINCT keyword), the cardinality can be estimated as the sum of the number of subjects in each fragment that includes all the predicates in the star-shaped subquery. For instance, the cardinality of P 1 in the query in Fig. 4(a) is the number of distinct subject values in f 1 and f 2 .

Given a star pattern P and a fragment f, the cardinality of P over f, assuming that f is a relevant fragment for P, is the number of values in the partitioned bitvector on the subject position in B S ( f ) , and is formally defined as: (4) card D ( P , f ) = card P ( B S ( f ) . B s )

For queries not including the DISTINCT keyword, we need to account for duplicates by considering, on average, the number of triples for each non-variable predicate value in P that each subject value is associated with. Given a star pattern P and fragment f, let preds ( P ) denote the non-variable predicate values in P (in the case of a variable on the predicate position in P, we consider the average number of predicate occurences in the characteristic set). The cardinality of P is thus estimated as follows [52,55]: (5) card S ( P , f ) = card D ( P , f ) · ∏ p i ∈ preds ( P ) card P ( B S ( f ) . Φ ( p i ) ) card P ( B S ( f ) . B s )

Henceforth, we will refer to the more generalized function card rather than card D and card S to be equivalent to card D for queries with the DISTINCT modifier and card S for queries without. Using Equations (4) or (5), the cardinality of a star pattern P over a node n’s SPBF index is, for all queries (both with and without the DISTINCT keyword), the aggregated cardinality over each relevant fragment to P, and is formally defined as follows: (6) card n ( P ) = ∑ f ∈ I n S . η ( P ) card ( P , f )

OPEN IN VIEWER

Fig. 9.

Estimating the cardinality of P 1 with the DISTINCT modifier as the number of subjects in f 1 and f 2 found using equation (4).

Consider, for instance, in the running example, the star-shaped BGP P 1 in Fig. 4(a) and the estimated cardinalities of the partitioned bitvectors for each fragment in Table 1. Assume in this case that the DISTINCT keyword is given in the query. Then, card n 1 ( P 1 ) is computed as the aggregated estimation of subject values in f 1 and f 2 , card n 1 ( P 1 ) = 1000 + 2000 = 3000 . This is visualized in Fig. 9.

OPEN IN VIEWER

Fig. 10.

Estimating the cardinality of P 1 without the DISTINCT modifier. Outlines show which bitvector each value is computed from.

If, instead, the DISTINCT keyword was not included in the query, the cardinality card n 1 ( P 1 ) is, for each relevant fragment ( f 1 and f 2 ), the number of subject values within the fragment multiplied with the average number of triples with each predicate p i ∈ preds ( P 1 ) that each subject value is associated with, card n 1 ( P 1 ) = 1000 · ( 5000 / 1000 ) · ( 1000 / 1000 ) + 2000 · ( 3000 / 2000 ) · ( 2000 / 2000 ) = 5000 + 3000 = 8000 . Figure 10 visualizes the above computations and shows which bitvector each value is computed from.

Until now, the cardinality estimations presented in this section are useful for estimating the cardinality of individual star patterns in a query [52,55]. However, to estimate the cardinality of arbitrary BGPs, [28] introduced characteristic pairs that describe the connections between IRIs described by different characteristic sets. In our case, however, we rely on the SPBFs of the relevant fragments to compute characteristic pairs without storing additional information; by intersecting the partitioned bitvectors on the positions corresponding to the join variable, we can estimate the selectivity of a given join and use that to estimate the cardinality of the join.

To achieve this, we first extend the framework for cardinality estimation described above to enable cardinality estimation of an entire query execution plan. This is straightforward for Cartesian products, unions, and selections; for Cartesian products it is the multiplication of the cardinality of the operands, for unions it is the sum of the cardinality of the operands, and for selections it is the cardinality of the star pattern over a specific fragment defined in Equations (4) and (5). Given the reasoning above, we define the cardinality of a query execution plan Π, card ( Π ) , covering all types of Π, as follows: (7) card ( Π ) = card ( Π 1 ) · card ( Π 2 ) , if Π = Π 1 × n Π 2 card ( Π 1 ) + card ( Π 2 ) , if Π = Π 1 ∪ Π 2 card ( P , f ) , if Π = [ [ P ] ] f n card ⋈ ( Π 1 ⋈ n Π 2 ) , if Π = Π 1 ⋈ n Π 2

To compute the cardinality of any join Π = Π 1 ⋈ n Π 2 (e.g., including joins between a BGP with multiple star patterns and a star pattern), we consider two cases: (1) where Π 2 is a union Π 2 = Π 2 ′ ∪ Π 2 ″ , and (2) where Π 2 is a selection Π 2 = [ [ P ] ] f n 1 . The cardinality of the join can thus be estimated using the following formula: (8) card ⋈ ( Π 1 ⋈ n Π 2 ) = card ⋈ ( Π 1 ⋈ n Π 2 ′ ) + card ⋈ ( Π 1 ⋈ n Π 2 ″ ) , if Π 2 = Π 2 ′ ∪ Π 2 ″ card ⋈ ( Π 1 , P , f ) , if Π 2 = [ [ P ] ] f n 1