Robust query processing for linked data fragments

3.1.Federated SPARQL query processing

A variety of cost models [9,12,23–25] and adaptive query processing approaches [2,20] have been proposed to employ efficient federated SPARQL query processing. DARQ [24] implements a cost model to reduce the amount of data transferred and the number of transmissions. The cost estimation functions for nested loop joins and bind joins rely on join cardinality estimations derived from statistical information from the service descriptions of the federation members. The optimizer uses Iterative Dynamic Programming (IDP) to obtain an efficient query plan. The federated engine SPLENDID [12] also implements a cost model to devise efficient plans. The cost model incorporates the network communication based on the estimated number of tuples to be transferred for either a bind or a hash join operator. Join cardinalities are estimated using precomputed indices based on VoID descriptions. The optimizer uses Dynamic Programming (DP) to find the cost-optimal plan. The cost model in SemaGrow [9] incorporates the costs for querying endpoints, transferring tuples, and processing them locally. The cardinality estimations for computing these costs rely on detailed statistics including the number of distinct subjects, predicates, and objects for a given triple pattern. The query plan optimizer uses DP to enumerate the space of possible join plans and prunes inferior plans. The cost function of Odyssey [23] is only based on the cardinalities of intermediate results to favor plans that produce fewer intermediate results. The cardinalities of intermediate results are estimated using characteristics sets statistics of the federation members, and DP is applied to enumerate alternative join orders for the subexpressions. CostFed’s [25] cost model also relies on detailed data summaries to estimate the cardinalities of subexpressions. The cost model is tailored to physical query plans with symmetric hash joins and bind joins. The optimizer follows a greedy-heuristic to incrementally build sub-plans with minimal cardinalities.

Federated SPARQL query processing approaches that focus on runtime adaptivity include ANAPSID [2] and ADERIS [20]. ANAPSID [2] is a federated SPARQL query engine that adapts to data availability and runtime conditions of endpoints to hide delays from users. The engine implements a query planner based on adaptive sampling and two non-blocking join operators that aim to produce query results even in the case that SPARQL endpoints get blocked. The ADERIS system [20] focuses on adaptive join ordering for federated SPARQL queries. The system relies on basic statistics (predicates per source) for an initial decomposition of the query. In contrast to a static query plan, ADERIS adapts the join order during query execution using a cost model that uses the cardinalities and selectivities determined during runtime.

Similar to our work, existing federated querying approaches [9,12,23–25] implement cost models for comparing alternative query plans. The cost models typically combine local processing and network communication cost and consider different join operators. However, they rely on fine-grained cardinality estimations that are based on detailed statistics about the data sources. Moreover, the optimizers are able to employ different plan enumeration approaches, such a DP, since typically the search space for federated plans is smaller as they work with subexpressions which are typically composed of several triple patterns and other operators. Similar to the Polymorphic Join Operators, ANAPSID [2] also focuses on intra-operator adaptivity but the join operators adapt to delayed or bursty data traffic rather than cardinality estimation errors. In contrast to our work, ADERIS [20] implements inter-operator adaptivity and the cost model relies on cardinality and selectivity statistics produced during the query execution.

3.2.Linked data fragments and clients

Linked Data Fragments (LDF) are interfaces for accessing and querying RDF graphs on the Web [28]. A central difference between these interfaces is their expressivity and the metadata they provide [15,17], resulting in different client-side query processing approaches for LDF interfaces. The original Triple Pattern Fragment (TPF) client [28] supports the evaluation of SPARQL queries over TPF servers and implements a heuristics to process the query which tries to minimize the number of requests. The TPF client implements non-blocking operators and the join order is given by sorting the triple patterns according to their cardinality. The triple pattern with the smallest estimated number of matches is evaluated and the resulting solution mappings are used to instantiate variables in the remaining triple patterns. This procedure is executed continuously during runtime until all triple patterns have been evaluated. Comunica [27] is a modular query engine that supports SPARQL query evaluation over heterogeneous interfaces including TPF servers. The client is embedded in the Comunica framework which aims to provide a research platform for SPARQL query evaluation to support the development of modular, web-based query engines. Comunica currently supports two heuristic-based configurations. The sort configuration sorts all triple patterns according to the metadata and joins them in that order similar to [28]. The smallest configuration does not sort the entire BGP but selects the triple pattern with the smallest estimated count on every recursive evaluation call. The network of Linked Data Eddies (nLDE) [1] is an adaptive client-side SPARQL query engine over TPF servers. The query optimizer in nLDE builds star-shaped groups (SSG) and joins the triple patterns by ascending cardinality. The optimizer places either symmetric hash join or nested loop join operators to minimize the expected number of requests that need to be performed. Furthermore, nLDE realizes adaptivity by adjusting the routing of result tuples during the query execution according to changing runtime conditions and data transfer rates.

More expressive LDF interfaces include brTPF and smart-KG. Bindings-restricted Triple Pattern Fragments (brTPF) [14] are an extension of the TPF interface that allows for evaluating a given triple pattern with a sequence of bindings to enable more efficient bind join strategies. Given a triple pattern and sequence of bindings, the brTPF server instantiates the variables of the triple pattern with the bindings and evaluates them over the RDF graph to return the matching triples. The authors propose a heuristic-based client that builds left-deep query plans which aims to reduce the number of requests and data transferred by leveraging bind joins. Smart-KG [6] is a hybrid shipping approach which aims to balance the load between clients and servers when evaluating SPARQL queries over remote sources. The smart-KG server extends the TPF interface by providing access to compressed partitions of the graph. These partitions are based on the concept of predicate families to support the evaluation of star-shaped subexpression over the partition at the client. The smart-KG client determines which subexpressions are evaluated locally over the shipped partitions and which triple patterns should be evaluated at the server.

Finally, SaGe [22] is a query engine that supports Web preemption by combining a preemptable server and a corresponding smart client. The server supports the fragment of SPARQL which can be evaluated in a preemptable fashion. The client decomposes a query such that the resulting subexpressions can be answered by the server and it also handles the preemptable execution of the query. As a result, the evaluation of the subexpressions is carried out at the server using a heuristic-based query planner that builds left-deep plans with index loop joins.

Different from the existing LDF clients, our query planner relies on a cost model, a robustness measure, and IDP to devise efficient query plans. Specifically, we focus on the TPF interface to showcase the effectiveness of our approach, however, it can be extended to support additional LDF interfaces. For instance, by extending the probe function in our cost-model to also support brTPF with several bindings per request. In addition, more expressive LDF interfaces and their clients may also benefit from our robustness measure. For example, to devise efficient and robust query plans in the smart-KG client or the SaGe server. While existing clients implement adaptivity during runtime [1,27,28], in contrast to the Polymorphic Join Operators, these adaptive approaches do not adjust the join strategy in response to estimation errors but focus on changing the join order and tuple routing during execution.

3.3.Robust query processing in relational databases

In the realm of relational databases, various approaches address uncertainties in the statistics and parameters used in cost models. Wiener et al. [29] consider different types of robustness including (i) query optimizer robustness as the ability of the optimizer to choose good plans under unexpected conditions, and (ii) query execution robustness as the efficient execution of a given plan under different runtime conditions. Yin et al. [32] focus on the former by investigating robust query optimization methods which are robust with respect to estimation errors. Their classification of such methods includes Robust Plan Selection, which comprises approaches that select a “robust” plan which is less sensitive to estimation errors over the “optimal” plan. These approaches, for example, use probability density functions for cardinality estimations instead of single-point values [7] or define cardinality estimation intervals where the size of the intervals indicate the uncertainty of the optimizer [8]. Wolf et al. [30] propose cardinality-based and selectivity-based robustness metrics for query plans. The core idea is computing the cost of a query plan as a function of the cardinality and selectivity estimations at all edges in the plan. The robustness metrics are computed based on the slope and area under the resulting cost function.

The CROP query planner can be considered a Robust Plan Selection approach. In contrast to [7] and [8], CROP has to rely on coarse grained statistics that do not allow for computing selectivity or cardinality estimation probabilities. Similar to [30], our measure computes the robustness of query plans and selects a robust plan from a selection of the cheapest plans.

Besides robust query planning, a variety of adaptive query processing approaches [10,13] have been proposed to support query execution robustness. Similar to our work, operator replacement considers approaches, where physical operators can be replaced with a logically equivalent operator at runtime [13]. Operator replacement typically occurs due to mid-query re-optimization [10]. For example, progressive query optimization (POP) [21] adapts to cardinality estimation errors by re-optimizing the query plans potentially leading to operator replacement. To this end, POP determines validity ranges of sub-plan cardinalities that trigger query plan re-optimization if the actual cardinalities violate these ranges. Materialized views and corresponding checkpoints allow the re-optimized query plans to reuse intermediate results.

Fig. 2.

Overview of CROP.

Similarly, Rio [8] adapts to cardinality estimation errors by (i) considering the robustness of query plans during query optimization based on bounding-boxes for estimations, and (ii) creating switchable plans during the optimization phase that can be used as alternatives plans in response to estimation errors at runtime.

The proposed Polymorphic Join Operators can be considered as dynamic operator replacement. However, in contrast to [21] and [8], the Polymorphic Join Operators do not require a re-optimization of the query plan or precomputed switchable plans. Moreover, the Polymorphic Join Operators independently decide whether they adapt their join strategy solely based on runtime conditions without detailed statistics for validity ranges or bounding-boxes. This makes our proposed solution suitable for query execution over LDFs, where the engine has access only to coarse-grained statistics.

4.1.Preliminaries

The foundation of this work is the Resource Description Framework (RDF). Consider the three pairwise disjoint sets of Internationalized Resource Identifiers (IRIs) I, blank nodes B, or literals L. An RDF term is an element in I∪B∪L and an RDF triple is a 3-tuple of RDF terms: t=(s,p,o)∈(I∪B)×I×(I∪B∪L), with s the subject, p the predicate, and o the object of the triple. A finite set of RDF triples is called an RDF graph G and the universe of RDF graphs is denoted by G. SPARQL is the recommended query language for RDF which allows to construct queries by replacing RDF terms with variables. Following the notation introduced by Schmidt et al. [26], let V be the set of variables disjoint from I, B, and L.

Furthermore, we denote the universe of SPARQL expression as P. A basic graph pattern (BGP) P is either a triple pattern or an expression of the form P=(P1ANDP2), where P1 and P2 are either a conjunctive expression (And) or a triple pattern. By ⟦P⟧G, we denote the evaluation of a SPARQL expression P over an RDF graph G. We denote the number of triple patterns in a BGP P by |P|. We assume set semantics for SPARQL as defined by Schmidt et al [26] and thus, the evaluation of an expression yields a set of solution mappings Ω={μ1,…,μn}, where a solution mapping is a partial function μ:V→IBL, mapping variables to RDF terms. The set of variables for which μ is defined is called the domain of a solution mapping dom(μ)⊂V. Moreover, we define a function vars:P→V that maps an expression to the set of variables in the expression. Finally, given a BGP P and a solution mapping μ, we denote μ(P) replacing all variables ?x∈dom(μ)∩vars(P) in P by μ(?x).

In the remainder of this work, we focus on query plans for BGPs to be evaluated over Linked Data Fragment (LDF) servers. An LDF server is a Web interface to access and query RDF graphs. We identify an LDF server by its IRI c∈I. Similar to [5] and [17], we defined a function ep:I→G that maps the IRI of an LDF server to the (default) graph available at the server. Moreover, we denote the type of interface of an LDF server int(c).

The central components of query plans for evaluating BGPs over LDF servers are its access operators. An access operator evaluates a SPARQL expression over a given LDF server by performing the necessary HTTP requests to obtain all solution mappings according to the interface.

A join query plan defines the join order, the physical join operators, and the access operator for evaluating a BGP P in a tree structure.55

For query plan T, we denote the number of leaves as |T|=|A(T)|. The number of join operations is then given as |T|−1. The longest path from the root of a (sub) query plan T to any access operator is denoted as height(T) and for an access operator A, we have height(A)=0. Moreover, the estimated number of solution mappings obtained by evaluating T is denoted card(T). Note that we distinguish algebraic join operators () and physical join operators (⋈). In addition, we indicate the join algorithm of a physical join operator by its subscript. For example, a bind join operator is denoted by ⋈BJ. In this work, we consider a physical join operator to be correct, if it adheres to the set semantics defined in [26].

Furthermore, we define a function expr(T) that maps a query plan T to its algebraic structure. This function traverses the tree structure of the query plan and replaces the access operators by their algebra expression and physical operators by the corresponding algebra operators as defined in Definition 4.4. Hence, expr(T) allows us to compare different plans and to determine whether they are algebraically equivalent.

The proposition follows from the algebraic equivalences defined by Schmidt et al. [26] and the fact that all expressions are evaluated over the same graph ep(c). Finally, T(P) denotes a query plan for a SPARQL expression P, that is expr(T)=P.

4.2.Cost model

We now present our cost model to estimate the cost of evaluating a query plan for a basic graph pattern over an LDF server. In principle, the cost for evaluating a query plan in a decentralized scenario are determined by (i) the server cost for processing the requests (i.e., evaluating the expression) on the server, (ii) the network cost for transporting requests and responses over the network, and (iii) the client cost, for processing the tuples of the response. Specifically, determining the server and network costs is challenging in practice since they are influenced by a large number of factors, such as the server load or potential network delays. Therefore, we use the number of requests that need to be performed by an access operator as a proxy for server and network costs. Consequently, for the sake of comparability of the individual requests of the access operator and because any LDF interface (except data dumps) allows for evaluating triple patterns, we assume the subexpressions in the access operators to be triple patterns, i.e., SEi=tpi, ∀(SEi,c)∈A(T). Moreover, we only consider query plans which are evaluated over a single LDF server c. The number of requests that an access operator needs to perform depends on the specific LDF server. For example, TPF servers require several requests when the number of resulting tuples exceeds the page size of the server, while (in principle) a single request suffices when accessing SPARQL endpoints. Furthermore, other LDF servers, such as brTPF servers, may require fewer requests than TPF servers when probing tuples for a triple pattern because they support probing multiple bindings [14].

In the following, we focus on the request cost for access operators for TPF servers and for the physical join operators Bind Join (BJ) and symmetric Hash Join (HJ). Given a query plan T for a conjunctive query the cost of evaluating T over LDF server c is computed as

4.3.Query plan robustness

Query planning approaches benefit from accurate join cardinality estimations to determine a suitable join order and to properly place physical operators such that the execution time of the query plan is minimized. However, estimating the join cardinalities is a challenging task, especially in the case that only basic statistics about the data are available. Addressing this challenge, we propose a robustness measure in order to determine how strongly the costs of a query plan are affected by potential cardinality estimations errors. To this end, our robustness measure compares the best-case cost of a query plan to its average-case cost. The average-case cost of a query plan is computed by using different cardinality estimation functions in the cost model to cover alternative join cardinalities. The resulting cost for each estimation function and the same query plan can be aggregated to an average cost value. Thus, a robust query plan is a plan in which the best-case cost only slightly differs from the average-case cost.

That is, the cost ratio robustness (crr) of a plan is the ratio between the cost in the best-case cost∗ and the cost in the average-case cost‾. A higher ratio indicates a more robust query plan because its expected average-case costs are not as strongly affected by changes in the cardinality estimations with respect to its best-case cost. Note that while we focus on query plans for basic graph patterns in this work, our robustness measure can be applied to query plans for other expressions as well.

We extend the definition of the cost function from Section 4.2 to capture the average-case cost of a query plan by including the cardinality estimation functions applied to each join operator. Let O={o1,…,on−1} be the set of (binary) join operators for a query plan T (|T|=n). Let E=[e1,…,en−1] be a vector of estimation functions with ei the cardinality estimation function applied at join operator oi. For the join operator oi, the cardinality estimation function ei:N02→N0 maps the cardinalities of the sub-plans a=card(T1) and b=card(T2) to an estimated join cardinality value. In practice, the values of a cardinality estimation function are bound within zero and the cross-product of a and b: ei(a,b)∈[0,a·b]. We then denote the cost for a query plan T computed using the cardinality estimation functions given by E as costE(T).

In other words, at every join operator in the query plan, we use the minimum cardinality of the sub-plans to estimate the join cardinality. This is identical to the estimations used in our cost model. Note that while in principle the cardinality for very selective joins can even be lower than the minimum, it still provides an optimistic estimation for computing the best case cost. The computation of the average-case cost requires applying different combinations of such estimation functions at the join operators.

By applying a variety of cardinality estimation functions, different join selectivities can be reflected in the average-case cost. We empirically tested different sets of estimation functions in F and found that the following functions yield suitable estimations for computing the average-case cost: F={f1,f2,f3,f4} with

The rationale for selecting these functions is to capture different join selectivities based on the cardinality estimations of the sub-plans to be joined. The function f1 captures high join selectivities (i.e., the join produces few solution mappings), while f2 and f3 capture medium and f4 low join selectivities. Moreover, we observed that for subject-object (s-o) and object-object (o-o) joins, the cardinalities were more frequently misestimated with the optimistic cardinality estimation function, while for all other types of joins, such as in star-shaped groups, it provided adequate estimations. Similar observations about the challenge in estimating join cardinalities for s-o and o-o joins have been reported in other works as well [31]. Therefore, we only consider alternative cardinality estimation function ei for a join operator oi, if the join performed at oi is either of type s-o or o-o. Thus, for s-s joins, we estimate their cost using only the best-case cost estimation (Definition 4.6). For s-o and o-o join, the alternative cost estimation functions in F are used to compute the average case cost as defined in 4.7.

4.4.Query plan optimizer

After presenting our cost model and robustness measure, we now present our optimizer that combines both aspects. Introducing the concept of robustness for query plans in addition to their cost yields two major questions: (i) in which cases should a more robust plan be chosen over the cheapest plan, and (ii) which alternative robust plan should be chosen instead of the cheapest plan? To this end, we propose a query plan optimizer that combines both aspects. Its parameters allow for defining the sensitivity of when a robust plan should be selected and also which alternative plan should be chosen over the cheapest plan. The query plan optimizer follows three main steps:

Algorithm 1:

CROP query plan optimizer

The query plan optimizer is detailed in Algorithm 1. Given a BGP P, the query planner determines the best query plan T∗ for P. The input parameters are the block size k∈[2,∞) and the number of top t∈N cheapest plans for the IDP algorithm. Moreover, the planner relies on a robustness threshold ρ∈[0,1] and a cost threshold γ∈[0,1]. The first step is to obtain a selection of alternative query plans using IDP. We adapted the original “IDP1−standard−bestPlan” algorithm presented by Kossmann and Stocker [19] in the following way. Identical to the original algorithm, we only consider select-project-join queries, i.e. basic graph patterns, and each triple pattern tpi∈P is considered a relation in the algorithm. Given a subset of triple patterns S⊂P, the original algorithm considers the single optimal plan for S according to the cost model in optPlan(S) by applying the prunePlans function to the potential candidate plans. However, as we do want to obtain alternative plans, we keep the top t cheapest plans for S in optPlan(S) for |S|>2. When joining two triple patterns (|S|=2), we always choose the physical join operator with the lowest cost. We follow this strategy for the following reasons. First, we expect accurate cost estimations for joining two triple patterns as the join estimation error impact is low in the base case. Second, by considering fewer alternatives that are likely to be discarded later in the algorithm, we can reduce the number of plans to be explored with IDP without loosing viable alternatives.

Given the set of t candidate query plans T from the IDP, the overall cheapest plan T∗ is determined (Line 3). If the cheapest plan is considered robust enough according to its cost ratio robustness crr(T∗) and the robustness threshold ρ, it becomes the final plan and is returned (Line 11). However, if the plan is not robust enough with respect to ρ and there are alternative plans to choose from (Line 4), the query plan optimizer tries to obtain a more robust alternative plan. First, the planner considers the set of plans R which are above the robustness threshold as potential alternatives. If no such plans exist, it considers all alternative plans except the cheapest plan (Line 7). If the ratio of best-case cost of the cheapest plan T∗ to the best-case cost of the alternative plan R∗ is higher than the cost threshold γ, the alternative plan R∗ is selected as the final plan T∗. For instance, for ρ=0.1 and γ=0.2, a robust plan is chosen over the cheapest plan if (i) for the cheapest plan T∗, the average-case cost cost‾(T∗) is 10 times higher than the best-case cost cost∗(T∗) and (ii) for the alternative robust plan R∗, the best-case cost cost∗(R∗) is no more than 5 times (1/γ) higher than best-case cost of the cheapest plan cost∗(T∗). Hence, smaller robustness threshold values lead to selecting alternative plans when the cheapest plan is less robust, and smaller cost threshold values lead to less restriction on the alternative robust plan with respect to its cost. The combination of both parameters allows for exploring alternative robust plans (ρ) but does not require to choose them at any cost (γ) and therefore, the performance degradation risk [32] is limited. Finally, we investigate the time complexity of the proposed optimizer.

5.1.Polymorphic bind join

The Polymorphic Bind Join (PBJ) is a physical join operator that can switch its join strategy during query execution from a bind join to a hash join strategy. We denote the PBJ by ⋈PBJ and given a query plan T=T1⋈T2, the PBJ can be placed when T2 is an access operator in the query plan.

The PBJ operator is outlined in Algorithm 2. The operator receives a stream of tuples from the evaluation of the sub-plan T1 as Ω1 (Line 2). The end of the stream is indicated by an end-of-file (EOF) tuple. The operator receives and processes the tuples in an asynchronous, non-blocking fashion.77 In its first phase (Line 5 to 12), the operator follows a bind join strategy and produces the resulting tuples to its output (Line 7). The operator keeps track of the number of probed tuples cnt, which is used to decide whether it should adapt its join strategy. The switch function in Line 9 determines whether the PBJ will switch from the bind join to the hash join strategy. The challenge in deciding whether the strategy should be switched is the fact that the operator does not certainly know the number of tuples that are still remaining from eval(T1) until the EOF is received. In our case, the operator assumes that a bind join strategy was chosen by the planner as it expected that the requests cost for probing all tuples in eval(T1) would be lower than for a hash join operator. To this end, the operator determines whether to switch its join strategy according to the number of probed tuples (cnt) and the access cost for T2 if it was to switch to a hash join:

Adding the parameter λ∈(0,∞) allows for setting the sensitivity of the operator. Lower λ values indicate a higher sensitivity as the operator would decide earlier to switch its join strategy. Furthermore, λ could be set according to the height of the operator in the plan.

In the case that the operator decides to switch its join strategy (Line 9), it terminates the loop of the bind join. Thereafter, the operator sets up two hash tables (Line 14 and 15) and starts evaluating T2 and process the tuples from the resulting stream. In the second phase (Line 18 to 26), the operator implements a non-blocking, symmetric hash join. It produces the join tuples by inserting and probing the remaining tuples it receives from eval(T1) and all tuples from eval(T2) in the proper hash tables. The operator finishes after receiving the EOF from both inputs.

Algorithm 2:

Polymorphic bind join

Parameters for the PBJ The PBJ follows a simple heuristic-based decision rule to decide whether it should switch its join strategy. This is due to the fact that it cannot assess how many tuples in eval(T1) are still remaining to be received. If the bind join strategy of the operator is sub-optimal with respect to the number of requests, the operator should switch to the hash join strategy after probing the first tuple. Since the operator cannot know whether this is the case, it follows the decision rule in switchPBJ to determine the number of tuples (cnt) that it probes in the bind join phase, before switching to the hash join phase. There is a maximum value for cnt for which switching to the hash join reduces the number of requests in comparison to not switching. We can split the tuples received from eval(T1) into two disjoint subsets Ω1=Ω1BJ∪Ω1HJ, with Ω1BJ the tuples that are received and probed during the bind join phase and Ω1HJ the tuples received during the hash join phase. The operator can reduce the number of requests if λ·acc(T2) is set such that the operator switches with cnt=|Ω1BJ| and

In other words, the PBJ decides correctly if it switches to a hash join operation as long as the access requests for T2 in the hash join are lower than probing the remaining tuples Ω1HJ in the bind join phase. This will be illustrated in the following example.

Correctness of the PBJ We show that the Polymorphic Bind Join operator produces correct solution mappings according to the SPARQL set semantics [26].

The intuition of the operator’s correctness is as follows. If the operator does not switch its join strategy, it operates as a regular bind join operator producing correct results. In the case that it switches from the bind join strategy to the hash join strategy, there are still tuples from eval(T1) that have not been considered in the join. These remaining tuples will be processed in the hash join phase (Line 19) and inserted into the hash table H1. The remaining results of the join are produced in the hash join phase: (i) either when the tuples are probed in the hash table H2 of eval(T2) (Line 22), or (ii) when the tuples from eval(T2) are probed against the tuples in H1 (Line 26). In order to avoid spurious duplicates, the tuples of eval(T1) that are probed during the bind join phase are not considered again in the hash join phase. The formal proof of Theorem 5.1 is presented in the following.

5.2.Polymorphic hash join

We now introduce the Polymorphic Hash Join (PHJ), which is the complement of the PBJ. The PHJ enables adaptivity by switching its join strategy from a hash join to a bind join during query execution. We denote the PHJ by ⋈PHJ and it can be placed in a query plan T=T1⋈PHJT2 when T2 is an access operator in the plan. The core idea of the operator is that it decides to switch the join strategy after receiving the last tuple from the sub-plan T1. At this point, the operator estimates whether probing all tuples received from T1 would require fewer requests than continuing with the hash join by obtaining the remaining tuples from T2.

Algorithm 3:

Polymorphic hash join

The PHJ operator is outlined in Algorithm 3. In the first phase, the operator follows a non-blocking, symmetric hash join strategy as long as the operator receives tuples from T1 (Line 9 to 23). During the hash join phase, the operator builds two hash tables H1 and H2 by inserting the tuples obtained from T1 and T2. Tuples obtained from T1 are then probed in H2 and vice versa. Moreover, the operator keeps track of the set of solution mapping O that it produces (Line 14 and 22). As soon as the operator receives the EOF tuple from T1 which indicates that all tuples from eval(T1) have been received and processed, the operator determines whether it is convenient to switch the join strategy. The decision to switch is determined by the switch function. The function determines the number of remaining requests with the hash join strategy and compares it to an estimation of the requests for probing all tuples received from T1 using a bind join. The number of remaining hash join requests accΔ is determined by the total number of requests for obtaining the tuples from T2 and subtracting the number of requests which have already been performed. The operator then estimates whether instantiating all tuples received from T1 and requesting the resulting expressions would require fewer requests than to continue with the hash join:

The parameter ε∈(0,∞) allows for weighting the bind join requests and we will show that we can guarantee that the PBJ minimizes the total number of requests during its execution if the parameter ε is set correctly. For a TPF server c with page size pc, the remaining number of hash join requests is determined as

In the case that the PHJ decides to switch to the bind join, it terminates the hash join phase (Line 16). Thereafter, in the bind join phase, the operator probes all tuples from eval(T1) as they have been inserted in the hash table H1 (Line 25 to 28). The operator does not know whether there are remaining tuples from eval(T2) that are compatible with a solution mapping from eval(T1), which has not been inserted into H2 during the hash join phase. Therefore, it probes all tuples obtained from eval(T1) in the inner plan T2 (Line 26) during the bind join phase. Following set semantics, the operator does not produce duplicate tuples as it determines whether a tuple has been produced already (Line 27) before producing it.

Optimality condition for the PHJ For the PHJ there is an optimal parameter ε∗, for which the operator minimizes the total number of requests during its execution.

In other words, if ε∗ is equal to the average number of requests to be performed for probing a tuple from eval(T1) in T2, the switch function in the operator will evaluate to True only if switching requires fewer requests. The proposition follows from the following observations. First, we decide whether to change the strategy once we have received all tuples from the sub-plan P1 and therefore, we have that cnt1=|eval(T1)| and consequently, the left side in the decision rule of the switch function is ε·|eval(T1)|. Therefore, the switch function evaluates to True with ε∗, if we have

If we assume that all requests are equal in terms of response time, we have that ε⩾1 because probing a solution mapping requires at least one request. However, different types of requests may yield different response times [18] and, therefore, values for ε below 1 may also be feasible. Determining ε∗ requires knowing the number of requests for probing each solution mapping from T1 which is unlikely to be known in practice. Nonetheless, when setting the parameter ε it should be still considered that (i) depending on the expression SE2, probing an instantiation of SE2 may require several requests and (ii) a request for probing may require less time than a request of the hash join due to fewer intermediate results to be transferred.

Correctness of the PHJ We now show that the Polymorphic Hind Join operator produces correct solution mappings according to the SPARQL set semantics [26].

The intuition of the operator’s correctness is as follows. If the operator does not switch its join strategy, it operates as a regular hash join operator producing a correct result set. The operator considers switching from the hash join strategy to the bind join strategy when all tuples from eval(T1) have been received and inserted into H1 (Line 16). The hash join strategy is executed only when there are tuples from eval(T2) that have not been received yet (Line 24). In this case, all tuples in H1 are probed in T2 in the bind join phase to produce all results of the join. To avoid spurious duplicates, the operator checks in the bind join phase if a result has been produced during the hash join phase (Line 27). The proof of Theorem 5.2 is as follows.

5.3.Summary

We introduced a new class of adaptive join operators that are able to switch their join strategy during the execution of a query plan. In particular, we presented two instances from this class: the Polymorphic Bind Join (PBJ) and the Polymorphic Hash Join (PHJ). These operators allow for switching their join strategy from a bind join to a hash join and vice versa. While we presented PBJ and PHJ operators probe a single tuple in the bind join phase, both operators can easily be extended to handle several bindings per request (binding block size > 1). Moreover, the proposed operators can be further optimized by leveraging the query planner and the properties of the LDF interface. The query planner could determine the sensitivity of individual operators according to estimated cardinalities and probabilities of estimation errors. In addition, the operators could leverage the properties of the underlying LDF interface. For example, in the case that the solution mappings are sorted (e.g., for a TPF server with an HDT backend), the operators could use the order to reduce the number of requests after switching the join strategy. In addition, the PHJ could be extended in two ways. First, instead of keeping all produced tuples in the set O to avoid producing duplicate solution mappings, the operator could also check during the bind join phase, whether the tuples are in the hash table H2. If this is the case, they have been processed in the hash join phase already and the corresponding solution mapping has been produced.88 Second, the PHJ could track response times and estimate the number of requests for probing each tuple during its execution to properly set the ε parameter. In this work, we do not consider these optimizations in the implementation of the operators as we aim to evaluate the concept of switching the join strategy in a more general scenario and for a variety of query planning approaches.

6.1.Experimental setup

Datasets and queries As the basis for our evaluation, we focus on two different datasets and corresponding benchmark queries that have also been used in previous evaluation for LDF clients [1,6,14,22,27,28]. First, we use a synthetic RDF graph and benchmark queries from the Waterloo SPARQL Diversity Test Suite (WatDiv) [4]. Specifically, we generated a dataset with scale factor = 100 and the corresponding default queries with query-count = 5 resulting in a total of 88 distinct queries from query categories, namely, 25 linear queries (L), 35 star queries (S), 25 snowflake-shaped queries (F) and 3 complex queries (C). The second benchmark is taken from the evaluation of nLDE99 and is based on the real-world RDF graph of DBpedia 2014 [1]. The benchmark consists of two subsets of queries. The first subset, Benchmark 1 (BM 1), consists of 20 non-selective queries composed of basic graph patterns with up to 14 triple patterns that produce a large number of intermediate results. The second subset, Benchmark 2 (BM 2), consists of 25 more selective queries from the domains Historical, Life Science, Movies, Music, and Sports. The number of queries per category as well as statistics on the number of answers and number of triple patterns for the queries of benchmarks is provided in Table 2. In addition, to showcase the benefits of combining the cost model with robustness in the query plan optimizer on different RDF graphs, we designed an additional test-set with 10 queries for 3 RDF graphs that include either a s-o and or an o-o join and 3-4 triple patterns. We used the RDF graphs DBpedia 2014, GeoNames 2012, and DBLP 2017 for which we obtained the HDT files from the RDF-HDT website.1010

Implementation We implemented CROP1111 on top of the nLDE client⁴, which is implemented in Python 2.7.13. We implemented our cost model, robustness measure, and query plan optimizer, such that the resulting physical query plans can be executed using nLDE. We used the default of 2 eddy operators and do not consider the routing adaptivity features of nLDE, that is, we select no routing policy for the execution. We deployed the TPF server with an HDT backend [11] using the Server.js v2.2.31212 implementation. We set the number of workers for the TPF server to 5. All experiments were executed on a Debian Jessie 64 bit machine with CPU: 2× Intel(R) Xeon(R) CPU E5-2670 2.60 GHz (16 physical cores), and 256 GB RAM. The queries were executed three times in all experiments. The timeout was set to 900 seconds for the experiments to set the parameters for CROP (δ, k, γ, and ρ). For all the remaining experiments, we set a runtime timeout of 600 seconds.

Evaluation metrics We consider the following query execution metrics:

We provide all results of our experimental study in our supplemental material on Zenodo.1313