Julien Aimonier-Davat
email: julien.aimonier-davat@univ-nantes.fr

Hala Skaf-Molli

Pascal Molli
email: pascal.molli@univ-nantes.fr

Arnaud Grall
email: arnaud.grall@univ-nantes.fr

Thomas Minier
email: thomas.minier@univ-nantes.fr

Online approximative SPARQL query processing for COUNT-DISTINCT queries with Web Preemption

Keywords: Semantic Web, SPARQL, Aggregate Queries, Web preemption, Public SPARQL Endpoints

Getting complete results when processing aggregate queries on public SPARQL endpoints is a challenge, mainly due to the application of quotas. Although Web preemption allows processing aggregate queries online, on preemptable SPARQL servers, data transfer is still very large when processing count-distinct aggregate queries. In this paper, it is shown that countdistinct aggregate queries can be approximated with low data transfer by extending the partial aggregation operator with Hyper-LogLog++ sketches. Experimental results demonstrate that the proposed approach outperforms existing approaches by orders of magnitude in terms of the amount of data transferred.

Introduction

Context and motivation: Processing SPARQL aggregate queries on public SPARQL endpoints is challenging, mainly due to the fair-use policies of public endpoints that stop queries before termination [START_REF] Hasnain | Sportal: profiling the content of public sparql endpoints[END_REF][START_REF] Soulet | Anytime large-scale analytics of linked open data[END_REF]. For instance, a SPARQL query that computes the number of distinct objects per class, for all available classes, cannot be executed online on Wikidata or DBPedia. On both SPARQL endpoints, the query hits the quotas. Consequently, no results are delivered. Related works: A common workaround for computing such queries relies on dataset dumps, but re-ingesting large dumps is very costly and timeconsuming. Approximate Query Processing is a wellknown approach for computing aggregations and can * Corresponding author. E-mail: hala.skaf@univ-nantes.fr. be updated to support fair-use policies [START_REF] Soulet | Anytime large-scale analytics of linked open data[END_REF], but requires to accept a trade-off between accuracy and response time. Restricted SPARQL servers such as TPF [START_REF] Verborgh | Triple pattern fragments: a low-cost knowledge graph interface for the web[END_REF], Web preemption [START_REF] Minier | Sage: Web preemption for public sparql query services[END_REF] or SmartKG [START_REF] Azzam | Smart-kg: hybrid shipping for sparql querying on the web[END_REF] ensure termination of a restricted set of SPARQL operations while preserving the responsiveness of the restricted server. Unfortunately, aggregate functions are not supported by the restricted servers. Processing aggregate queries requires materializing the query mappings on the client-side before computing aggregates locally. Even if the processing is guaranteed to terminate, the size of the data transfer may be prohibitive.

In a previous work [START_REF] Grall | Processing sparql aggregate queries with web preemption[END_REF], we demonstrated that a partial aggregation operator is preemptable. Computing partial aggregations on a preemptable server drastically reduces data transfer for most aggregate queries, while ensuring complete results. However, count-distinct aggregate queries still generate a large data transfer, even with a partial aggregation operator. Computing the exact cardinality of a set requires a data transfer proportional to the size of the set, which is impractical for very large datasets. Approach and Contributions: To improve the evaluation of count-distinct aggregate queries, the approach proposed in [START_REF] Grall | Processing sparql aggregate queries with web preemption[END_REF] is extended with HyperLogLog++ sketches. HyperLogLog++ is a probabilistic algorithm that can estimate the cardinality of large sets with a small amount of memory and strong guarantees on the error rate. As HLL++ supports the decomposability property of aggregate functions, it can be integrated into the partial aggregations framework promoted in [START_REF] Grall | Processing sparql aggregate queries with web preemption[END_REF]. Compared to related Approximated Query Approaches [START_REF] Soulet | Anytime large-scale analytics of linked open data[END_REF], this approach ensures to find all GroupKeys in a single pass with a pre-defined and bounded error rate for all values. The contributions of the paper are the following:

-An extension of the partial aggregation operator presented in [START_REF] Grall | Processing sparql aggregate queries with web preemption[END_REF]. This extension allows estimating the result of a count-distinct query with a bounded error rate. -Additional experimental results that compare the performance of the extended operator and the previous operator [START_REF] Grall | Processing sparql aggregate queries with web preemption[END_REF]. Experimental results demonstrate that relying on estimates does not improve execution time but significantly reduces data transfer for count-distinct queries, and in the general case, show that the proposed approach outperforms existing approaches used for processing aggregate queries.

The remainder of this paper is structured as follows. Section 2 summarizes related works. Section 3 introduces SPARQL aggregate queries and the Web preemption model. Section 4 presents the approach for processing aggregate queries using a preemptive SPARQL server. Section 5 introduces Hyper-LogLog++ and its integration in the partial aggregation operator. Section 6 presents the different algorithms used to implement the proposed approach. Section 7 presents our experimental results. Section 8 discusses the limitations of the current proposal. Finally, conclusions and future work are outlined in Section 9.

Related Works

Aggregate Queries on public SPARQL endpoints Public endpoints such as DBPedia or Wikidata allow executing any SPARQL aggregate queries. However, such queries are often long-running queries that require a lot of CPU and memory resources to terminate. To ensure stable and responsive services to the user community, public SPARQL endpoints set up quotas on the maximum number of results returned, execution time, and arrival rate. Consequently, many aggregate queries cannot be executed online, simply because they reach the quotas of the fair-use policies [START_REF] Buil-Aranda | Strategies for executing federated queries in sparql1. 1[END_REF][START_REF] Minier | Sage: Web preemption for public sparql query services[END_REF][START_REF] Soulet | Anytime large-scale analytics of linked open data[END_REF].

Use of dumps A common workaround for quota limitations relies on dumps of datasets. Datasets dumps have to be first re-ingested on local resources before executing aggregate queries [START_REF] Auer | Lodstats-an extensible framework for high-performance dataset analytics[END_REF][START_REF] Schätzle | S2rdf: Rdf querying with sparql on spark[END_REF]. As datasets become bigger and bigger, re-ingesting large datasets is very costly, time-consuming, and raises freshness issues. Re-ingesting data dumps can be profitable only if a high number of aggregate queries have to be executed. The purpose of this paper is to process aggregate queries online, i.e. without moving the data.

Decomposition of queries Another well-known approach to overcome quotas is to decompose a query into smaller subqueries that can be evaluated under quotas. Query results are then recombined on the client-side [START_REF] Buil-Aranda | Strategies for executing federated queries in sparql1. 1[END_REF]. Such a decomposition requires a smart client that performs the decomposition and recombines the intermediate results. However, ensuring that subqueries can be completed under quotas remains hard [START_REF] Buil-Aranda | Strategies for executing federated queries in sparql1. 1[END_REF].

Restricted SPARQL server approaches Restricted SPARQL servers such as TPF [START_REF] Verborgh | Triple pattern fragments: a low-cost knowledge graph interface for the web[END_REF], Web preemption [START_REF] Minier | Sage: Web preemption for public sparql query services[END_REF] or SmartKG [START_REF] Azzam | Smart-kg: hybrid shipping for sparql querying on the web[END_REF] ensure termination of a restricted set of SPARQL operations while preserving the responsiveness of the restricted SPARQL servers.

The Triple Pattern Fragments restricted server (TPF) [START_REF] Verborgh | Triple pattern fragments: a low-cost knowledge graph interface for the web[END_REF] only supports triple pattern queries but ensures termination. To avoid server congestion, query results are paginated so that a page of results can be obtained in bounded time (a few milliseconds in practice). Thus, the server does not need quotas to be fair. However, as the TPF server only processes triple pattern queries, joins and aggregates are evaluated on a smart TPF client. This requires transferring all the intermediate results from the server to the client to perform joins, and then computing aggregate functions locally. Such an evaluation leads to poor query execution performance.

Web preemption [START_REF] Minier | Sage: Web preemption for public sparql query services[END_REF] is another approach to process SPARQL queries on a public server without quota enforcement. Web preemption allows a Web server to suspend a running SPARQL query after a quantum of time and resume the next waiting query. Suspended queries are returned to users that can re-submit them to continue the execution for another quantum of time.

Web preemption provides a fair allocation of server resources, a better average query completion time, and a better time for first results. However, if Web preemption allows processing projections and joins on the server-side, aggregate functions are not supported by the restricted preemptable SPARQL server. Processing aggregate queries requires materializing mappings on the client-side before performing local aggregations. Therefore, the data transfer may be intensive, especially for aggregate queries.

In our previous work [START_REF] Grall | Processing sparql aggregate queries with web preemption[END_REF], we demonstrated that a preemptable server supports partial aggregations. Combined with a smart client that can merge partial aggregations, it is possible to compute any aggregate queries online and ensure complete results. Partial aggregations drastically reduce data transfer for almost all aggregate queries, except those using the distinct modifier. Indeed, counting the number of distinct elements in a multiset requires a data transfer proportional to the size of the multiset. Such an approach is not tractable for large datasets. This is especially a problem since queries that count the number of distinct elements are common queries for many useful statistics.

Approximate Query Processing Approximate query processing is a well-known approach to speed up the processing of aggregate queries. Different approaches provide different trade-offs among the accuracy, response time, space budget, and supported queries [START_REF] Li | Approximate query processing: What is new and where to go?[END_REF]. The sampling approach proposed in [START_REF] Soulet | Anytime large-scale analytics of linked open data[END_REF] aims to explore large federations of SPARQL endpoints while being compatible with SPARQL endpoint fair-use policies. Given an aggregate query, the approach ensures that results converge to exact results as more samples are collected. However, this approach does not detail how to handle count-distinct aggregate queries and how SPARQL endpoints can answer probe queries with high offsets without being interrupted by fair-use policies. Moreover, the number of samples we need to collect to ensure that the algorithm converges could be greater than the number of triples in the datasets. The error-bound is also hard to estimate during processing. This paper explores a different trade-off: using probabilistic data structures to approximate the result of a count-distinct query in a single pass, with strong guarantees on the error rate.

Count-distinct aggregate queries can be computed with probabilistic cardinality estimators [START_REF] Flajolet | Hyperloglog: the analysis of a near-optimal cardinality estimation algorithm[END_REF] such as HyperLogLog or Count-Min sketches. These algorithms approximate the number of distinct elements in a multiset with a bounded error rate and bounded memory. For instance, the HyperLogLog algorithm can estimate cardinalities greater than 10 9 with a typical error rate of 2%, using only 1.5 KBytes of memory. HyperLogLog and its variant HyperLogLog++ are implemented and used for cardinality estimation by Google, Redis, Amazon, etc. For more information on cardinality estimation algorithms, the reader can refer to the review [START_REF] Singh | Probabilistic data structures for big data analytics: A comprehensive review[END_REF]. In this paper, the mergeability property of HyperLogLog++ counters is used to extend the preemptable partial aggregation operator introduced in [START_REF] Grall | Processing sparql aggregate queries with web preemption[END_REF].

Preliminaries

SPARQL Aggregate Queries

This paper uses the semantics of aggregates as defined in [START_REF] Kaminski | Query nesting, assignment, and aggregation in sparql 1.1[END_REF]. The important definitions to understand the proposal are recalled here. According to [START_REF] Kaminski | Query nesting, assignment, and aggregation in sparql 1.1[END_REF][START_REF] Pérez | Semantics and complexity of sparql[END_REF][START_REF] Schmidt | Foundations of sparql query optimization[END_REF], let us consider three disjoint sets I (IRIs), L (literals) and B (blank nodes). Let T be the set of RDF terms such that T = I ∪ L ∪ B. An RDF triple (s, p, o) ∈ (I ∪ B) × I × T connects a subject s through a predicate p to an object o. An RDF graph G is a finite set of RDF triples. Let us assume the existence of an infinite set V of variables, disjoint with previous sets. A mapping µ from V to T is a partial function µ : V → T where the domain of µ, denoted dom(µ), is the subset of V where µ is defined. A SPARQL graph pattern expression P is defined recursively as follows: The SPARQL 1.1 language [START_REF] Steve | SPARQL 1.1 query language[END_REF] introduces new features for supporting aggregate queries: i) A collection of aggregate functions for computing values, like COUNT, SUM, MIN, MAX and AVG. ii) GROUP BY and HAVING. HAVING restricts the application of aggregate functions to groups of solutions satisfying certain conditions.

-A tuple from (I ∪ L ∪ V) × (I ∪ V) × (I ∪ L ∪ V) is a triple pattern. -If P1
Both groups and aggregates deal with lists of expressions E 1 , . . . , E n , which are evaluated to v-lists, i.e. lists of values in T ∪ {error}. More precisely, the evaluation of a list of expressions E = E 1 , . . . , E n , according to a mapping µ, is defined as:

E µ = E 1 µ , . . . , E n µ .
For simplicity, lists of expressions are restricted to lists of variables. According to [START_REF] Kaminski | Query nesting, assignment, and aggregation in sparql 1.1[END_REF] this restriction does not reduce the expressive power of aggregates. Every query that uses lists of expressions can be rewritten into a query where grouping is only allowed over lists of variables. Inspired by [START_REF] Kaminski | Query nesting, assignment, and aggregation in sparql 1.1[END_REF][START_REF] Steve | SPARQL 1.1 query language[END_REF], we formalized Group and Aggregate as follows.

Definition 1 (Group). A group is a construct G(E, P) with E a list of expressions and P a graph pattern. The evaluation G(E, P) G of a group G(E, P) over an RDF graph G produces a set of partial functions from v-list keys (called GroupKeys) to multisets of mappings as follows:

G(E, P) G = {GroupKey → {µ | µ ∈ P G , E µ = GroupKey}} Definition 2 (Aggregate).
An aggregate is a construct γ(F, f, P) with F a list of expressions, f an aggregate function and P a graph pattern. Let {k 1 → ω 1 , . . . , k n → ω n } be the set of partial functions produced by the evaluation of G(E, P) G over an RDF graph G where k 1 , . . . , k n are GroupKeys and ω 1 , . . . , ω n are multisets of mappings. The evaluation of γ(F, f, P) G maps each GroupKey to a single value as follows:

γ(F, f, P) G = {k i → f (Ω), Ω = { F µ | µ ∈ ω i }}
To illustrate, consider the query Q 1 of Figure 1b, which returns the total number of objects per class, for subjects connected to the object o 1 , through the predicate p 1 . P Q1 = { ?s :a ?c. ?s ?p ?o. ?s :p1 :o1.} is the graph pattern of Q 1 . According to Definition 1, we have: G( ?c ,P Q1 ) G1 = { :c3 → {:c3, :c1, :c2, :o1, :c3, :o1, }, :c1 → {:o1, :c3, :c1}, :c2 → {:o1, :c3, :c2}} where ?c is the list of expressions E used by the GROUP BY operator. As we can see, this query returns 3 different GroupKeys, i.e. :c1, :c2 and :c3. For simplicity, for each GroupKey, only the value of the variable ?o is represented as ?o is the only variable used by the COUNT function. Then, according to Definition 2, the query Q 1 is evaluated as:

γ( ?o ,COUNT, P Q1 ) G1 = { :c3 → 6, :c1 → 3, :c2 → 3}
where COUNT is the aggregate function f and ?o is the list of expressions F used by f .

Web preemption and SPARQL Aggregate queries

Web preemption [START_REF] Minier | Sage: Web preemption for public sparql query services[END_REF] is the capacity of a web server to suspend a running SPARQL query after a fixed quantum of time and resume the next waiting query. When suspending a query Q, a preemptable server saves the internal state of all operators of Q in a saved plan Q s that is sent to the client. The client can continue the execution of Q by sending Q s back to the server. When reading Q s , the server restarts the query Q from where it has been stopped. As a preemptable server can restart queries from where they have been stopped and makes a progress at each quantum, it eventually delivers complete results after a bounded number of quanta.

However, Web preemption comes with overheads. The time taken by the suspend and resume opera-

:C3 Ω = Ω ∪ ω 1 q2 ω 2 :C3 :C1 :C1 :O1 Ω = Ω ∪ ω 2 q3 ω 3 :C1 :C3 :C1 :C1 Ω = Ω ∪ ω 3 q4 ω 4 :C3 :O1 :C3 :C3 Ω = Ω ∪ ω 4 q5 ω 5 :C3 :C2 :C2 :O1 Ω = Ω ∪ ω 5 q6 ω 6 :C2 :C3 :C2 :C2 Ω = Ω ∪ ω 6 γ({?O}, COUNT, Ω) {:C3 → 6, :C1 → 3, :C2 → 3} ω 1 next ω 2 next ω 3 next ω 4 next ω 5 next ω 6 : Done Q1 ' : SELECT ?c ?o WHERE {
?s : a ?c ; ?p ?o ; : p1 : o1 } Fig. 2. Evaluation of Q 1 on G 1 with regular Web preemption [START_REF] Minier | Sage: Web preemption for public sparql query services[END_REF] tions represents the overhead in time of a preemptable server. The size of Q s represents the overhead in space of a preemptable server and may be transferred over the network each time a query is suspended by the server. To be tractable, a preemptable server has to minimize these overheads.

For this purpose, a preemptable server implements only SPARQL operators that can be saved and resumed in constant time, i.e. preemptable operators. Based on the definition of tuple-at-a-time and full-relation operators [START_REF] Garcia-Molina | Database Systems: The Complete Book[END_REF], SPARQL operators can be classified into two groups: mapping-at-a-time and full-mappings operators.

Mapping-at-a-time operators such as SCAN, JOIN, UNION or BIND are implemented on the server. As they just need to manage one mapping at a time [START_REF] Garcia-Molina | Database Systems: The Complete Book[END_REF], these operators can be saved and resumed in constant time 1 . Queries that can be evaluated using only mapping-at-a-time operators are supported by the preemptable server.

On the other hand, full-mappings operators require "seeing all or most of the mappings in memory at once" [START_REF] Garcia-Molina | Database Systems: The Complete Book[END_REF]. Consequently, they cannot be saved and resumed in constant time and are implemented on the client. For example, the ORDER BY is a full-mappings operator as it needs to materialize all the mappings before sorting them. Note that using a full-mappings operator does not mean that the query is not preemptable. In some cases, the server can process a query that uses a full-mappings operator in such a way that the query is preemptable. For example, a query with an ORDER BY can be evaluated without materializing all the mappings if the server has the required sorted indexes. For queries that use both an ORDER BY and a LIMIT k operators, the server can replace them by a TOP k operator that is preemptable. However, in the general case, queries that use a full-mappings operator are not preemptable. To evaluate such a query, the client must decompose the query into a set of subqueries supported by the server, evaluate each subquery and recombine intermediate results to produce the query results. Such a decomposition can be extremely costly in terms of data transfer, number of calls to the server, and execution time.

Unfortunately, aggregate queries require a serverside operator that belongs to the full-mappings operators. Consequently, there is no support on the server and aggregate queries must be decomposed.

Figure 2 illustrates how Web preemption processes the query Q 1 of Figure 1b over the dataset D 1 . First, the smart client sends the BGP of Q 1 to the server, i.e. the query Q 1 = SELECT ?c ?o WHERE { ?s :a ?c; ?p ?o; :p1 :o1 }. Let us suppose that the query Q 1 requires six quanta to complete. At the end of each quantum q i , the client receives the mappings ω i and asks for the next results (next link). Then, when all mappings are obtained, the smart client computes γ( ?o , COUNT, i ω i ). As a result, to compute the three mappings {:c3 → 6, :c1 → 3, :c2 → 3}, the server transferred 6 + 3 + 3 = 12 mappings to the client.

In a more general way, to evaluate γ(F, f, Ω) G , the smart client first asks a preemptable web server to evaluate P G = Ω. Then the server transfers incrementally Ω, and finally, the client evaluates γ(F, f, Ω) locally. The main problem with this evaluation is that the size of Ω is usually much bigger than the size of γ(F, f, Ω).

Reducing data transfer requires reducing | P G | which is impossible without deteriorating the completeness of the answer. Therefore, the only way to reduce data transfer when processing aggregate queries is to process the aggregates on the preemptable server. However, the operator we need to evaluate SPARQL aggregates is a full-mappings operator, as it requires to materialize | P G |, hence it cannot be suspended and resumed in constant time.

Problem Statement: Define a preemptable aggregation operator γ such that the complexity in time and space of suspending and resuming γ is bounded in constant time2 .

Computing Partial Aggregations with Web Preemption

To build a preemptable evaluator for SPARQL aggregates, the presented approach relies on two key ideas: (i) First, Web preemption naturally creates a partition of mappings over time. Thanks to the decomposability of aggregate functions [START_REF] Yan | Eager aggregation and lazy aggregation[END_REF], partial aggregations can be computed server-side on each partition of mappings and recombined on the client. (ii) Second, to control the size of partial aggregations, the size of the quantum can be adjusted for aggregate queries.

In the following, the decomposability property of aggregate functions is presented, as well as how the property is used in the context of Web preemption.

COUNT SUM MIN MAX AVG f1 COUNT SUM MIN MAX SaC v v v + v min(v, v ) max(v, v ) v ⊕ v h Id (x, y) → x/y
(b) Aggregate functions with the DISTINCT modifier SPARQL Aggregate functions

COUNT D SUM D AVG D COUNT D f 1 CT HLL add v v v ∪ v HLL merge h COUNT SUM AVG HLL count

Decomposable aggregate functions

Traditionally, the decomposability property of aggregate functions [START_REF] Yan | Eager aggregation and lazy aggregation[END_REF] ensures the correctness of the distributed computation of aggregate functions [START_REF] Jesus | A survey of distributed data aggregation algorithms[END_REF]. This property is adapted for SPARQL aggregate queries in Definition 3.

Definition 3 (Decomposable aggregation function).

An aggregate function f that used a list of expressions F is decomposable if for all non-empty multisets of solution mappings Ω 1 and Ω 2 , there exists a (merge) operator , a function h and an aggregate function f 1 such that:

γ(F, f, Ω 1 Ω 2 ) = {GroupKey → h(v 1 v 2 ) | GroupKey → v 1 ∈ γ(F, f 1 , Ω 1 ), GroupKey → v 2 ∈ γ(F, f 1 , Ω 2 )}
In Definition 3, denotes the multiset union as defined in [START_REF] Kaminski | Query nesting, assignment, and aggregation in sparql 1.1[END_REF]. Abusing the notation, we use a multiset of solution mappings Ω instead of the graph pattern P in Definition 2. Table 1 gives the decomposition of all SPARQL aggregate functions, where Id denotes the identity function and ⊕ is the point-wise sum of pairs, i.e. (x 1 , y 1 ) ⊕ (x 2 , y 2 ) = (x 1 + x 2 , y 1 + y 2 ).

To illustrate, consider the function f = COUNT with F = ?c and an aggregate query γ(F, f, the COUNT function can be merged using an arithmetic addition operation, i.e. {:c1 → 2 5 = 2 + 5 = 7}.

Ω 1 Ω 2 ) such as γ(F, f, Ω 1 ) = {:c1 → 2} and γ(F, f, Ω 2 ) = {:c1 → 5}.
Decomposing SUM, COUNT, MIN and MAX is relatively simple, as partial aggregation results only need to be merged to produce the final query results. However, decomposing AVG and aggregate functions that use the DISTINCT modifier are more complex. Two auxiliary aggregate functions have been introduced, called SaC (SUM-and-COUNT) and CT (Collect), respectively. The SaC function collects the information required to compute an average, while the CT function collects a set of distinct values. They are defined as follows: SaC(X) = SUM(X), COUNT(X) and CT(X) is the base set of X as defined in sec-tion 3. For instance, the aggregate function of the query

Q = γ( ?o , COUNT D , Ω 1 Ω 2 ) is decomposed as Q = COUNT(γ( ?o , CT, Ω 1 ) ∪ γ( ?o , CT, Ω 2 )).

Partial aggregation with Web preemption

Using a preemptive web server, the evaluation of a graph pattern P over G naturally creates a partition of mappings ω 1 , ..., ω n over time, where ω i is produced during the quantum q i . Intuitively, a partial aggregation A i , formalized in Definition 4, is obtained by applying an aggregate function on the partition of mappings ω i .

Definition 4 (Partial aggregation). Let F be a list of expressions, f an aggregate function and ω i ⊆ P G such that P G = i=n i=1 ω i where n is the number of quanta required to complete the evaluation of P over the RDF graph G. A partial aggregation A i is defined as

A i = γ(F, f, ω i ).
Because a partial aggregation operates on ω i , partial aggregations can be implemented on the server-side as a mapping-at-a-time operator. Suspending the evaluation of aggregate queries using partial aggregations does not require to materialize intermediate results on the server. Finally, to process a SPARQL aggregate query, the smart client computes γ(F, f, P)

G = h(A 1 A 2 • • • A n ).
Figure 3a illustrates how a smart client computes Q 1 over D 1 using partial aggregations. Suppose that Q 1 is executed over six quanta q 1 , . . . , q 6 that produce two mappings each. At each quantum q i , two new mappings are produced in ω i and the partial aggregation A i = γ( ?o , COUNT, ω i ) is sent to the client. The client merges all A i thanks to the operator and then produces the final results by applying h. Figure 3b describes the execution of Q 2 with partial aggregations under the same conditions. As we can see, the DISTINCT modifier requires to transfer more data, however a reduction in data transfer is still observable compared with transferring all ω i for q 1 , q 2 , q 3 , q 4 , q 5 and q 6 .

The duration of the quantum seriously impacts query processing using partial aggregations. Suppose that instead of six quanta of two mappings in Figure 3a, the server requires twelve quanta that produce one mapping each, therefore partial aggregations are useless. If the server requires two quanta that produce six mappings each, then only two partial aggregations A 1 = {:c3 → 3, :c1 → 3} and A 2 = {:c3 → 3, :c2 → 3} are sent to the client and data transfer is reduced. If the quantum is infinite, then the whole aggregation is produced on the server-side, and data transfer is optimal. Overall, for a SPARQL aggregate query, the larger the quantum, the smaller the data transfer and execution time.

Count-Distinct SPARQL Aggregate Queries

Count-distinct aggregate queries count the number of distinct elements in the multisets obtained after grouping. Query Q 2 of Figure 1c is an example of a count-distinct aggregate query.

As illustrated in Figure 3b, processing count-distinct aggregate queries requires transferring all elements from the server to the client before counting them. Moreover, these elements could be transferred several times if the query is processed over several quanta. For example, :o1 and :c3 for the GroupKey :c3 in Figure 3b. Consequently, computing an exact count for a GroupKey requires an amount of memory, and thus data transfer, proportional to the cardinality of the multiset of the GroupKey. Such a data transfer is prohibitive and does not scale to large datasets.

To address this issue, we propose to estimate the number of distinct elements in a multiset rather than computing the exact count. Several probabilistic algorithms have been proposed [START_REF] Estan | Bitmap algorithms for counting active flows on high speed links[END_REF][START_REF] Flajolet | Hyperloglog: the analysis of a near-optimal cardinality estimation algorithm[END_REF][START_REF] Whang | A lineartime probabilistic counting algorithm for database applications[END_REF] to estimate large cardinalities with a bounded memory. According to [START_REF] Heule | Hyperloglog in practice: Algorithmic engineering of a state of the art cardinality estimation algorithm[END_REF], the LinearCounting algorithm [START_REF] Whang | A lineartime probabilistic counting algorithm for database applications[END_REF] achieves good accuracy, regardless of the cardinality. However, this algorithm is not attractive for large cardinalities, as it requires too much memory for an accurate estimate. Compared to the LinearCounting algorithm, the HyperLogLog (HLL) algorithm [START_REF] Flajolet | Hyperloglog: the analysis of a near-optimal cardinality estimation algorithm[END_REF] is efficient for large cardinalities, both in terms of space complexity and accuracy. For instance, HLL can estimate cardinalities greater than 10 9 with a typical error rate of 2%, using only 1.5KBytes of memory. However, HLL fails to estimate the cardinality of small sets. Moreover, the HLL algorithm is not memory efficient. No matter if the cardinality to be estimated is small, it uses the maximum amount of memory specified by the user, e.g. 1.5KBytes for an error rate of 2%.

In the context of SPARQL aggregate queries, a good estimator must be accurate on both small and large cardinalities, and adapt its memory usage to cardinality. Indeed, aggregate queries deal with GroupKeys that may have millions of distinct values as well as just a few. To fit these criteria, we use HyperLogLog++ (HLL++) [START_REF] Heule | Hyperloglog in practice: Algorithmic engineering of a state of the art cardinality estimation algorithm[END_REF], an adaptive counting algorithm that combines the HLL and the LinearCounting algorithms.

Because the LinearCounting algorithm is more efficient for small cardinalities than HLL, HLL++ relies on it to estimate small cardinalities, and automatically switches to HLL for larger cardinalities. Finally, HLL++ supports the decomposability property of aggregate functions. Consequently, it can be used to extend the partial aggregation operator proposed in [START_REF] Grall | Processing sparql aggregate queries with web preemption[END_REF].

A smart client merging partial aggregations based on HLL++ can now compute the number of distinct elements with a bounded error rate and bounded data transfer for each GroupKey.

HyperLogLog++

HLL++ is a probabilistic data structure that behaves like a set with two main operations:

1. HLL add for adding a new element e to the set. 2. HLL count for estimating the cardinality of the set with a fixed error rate .

The payload of a HLL++ set H is an array R of m registers noted R[1], ..., R[m] where is the error rate. According to [START_REF] Flajolet | Hyperloglog: the analysis of a near-optimal cardinality estimation algorithm[END_REF][START_REF] Heule | Hyperloglog in practice: Algorithmic engineering of a state of the art cardinality estimation algorithm[END_REF], m is equal to (1.04/ ) 2 , which is the number of registers required to ensure an error rate . To add an element e into H , e is first mapped to a 64 bit hash value h(e). The first p = log 2 (m) bits of h(e) represents the index i of R to update. The number of leading zeros k located just after the first p bits are stored in

R[i], if k > R[i].
To compute the cardinality of H , HLL++ relies both on the HyperLogLog and the LinearCounting algorithms. It first uses HLL to estimate the cardinality of H . If the estimated cardinality is greater than a threshold defined in [START_REF] Flajolet | Hyperloglog: the analysis of a near-optimal cardinality estimation algorithm[END_REF], HLL++ uses the HyperLogLog algorithm, otherwise the LinearCounting algorithm is used. To estimate the cardinality of H , the HyperLogLog algorithm relies on the idea that, in a uniformly distributed multiset of 64 bit hash values, long runs of leading zeros are less likely and indicate a larger cardinality. Based on this observation, if the maximum number of leading zeros k is known, a good estimation of the number of distinct values is 2 k+1 . Because a single measurement has a large variability, HLL divides the elements into m registers and then computes the cardinality estimate as the average of the m registers. This technique known as stochastic averaging operates as a variance reduction device [START_REF] Flajolet | Hyperloglog: the analysis of a near-optimal cardinality estimation algorithm[END_REF], which increases the quality of estimates.

D , ω i ) Client Merge(Ω, ω i ) q1 ω 1 :C3 :O1 :C3 :C3 :C3 → {R[8] → 2, R[13] → 2} :C3 → {R[8] → 2, R[13] → 2} q2 ω 2 :C3 :C1 :C1 :O1 :C3 → {R[12] → 1} :C1 → {R[13] → 2} :C3 → {R[8] → 2, R[13] → 2, R[12] → 1} :C1 → {R[13] → 2} q3 ω 3 :C1 :C3 :C1 :C1 :C1 → {R[8] → 2, R[12] → 1} :C3 → {R[8] → 2, R[13] → 2, R[12] → 1} :C1 → {R[8] → 2, R[13] → 2, R[12] → 1} q4 ω 4 :C3 :O1 :C3 :C3 :C3 → {R[8] → 2, R[13] → 2} :C3 → {R[8] → 2, R[13] → 2, R[12] → 1} :C1 → {R[8] → 2, R[13] → 2, R[12] → 1} q5 ω 5 :C3 :C2 :C2 :O1 :C3 → {R[11] → 2} :C2 → {R[13] → 2} :C3 → {R[8] → 2, R[11] → 2, R[12] → 1, R[13] → 2} :C1 → {R[8] → 2, R[13] → 2, R[12] → 1} :C2 → {R[13] → 2} q6 ω 6 :C2 :C3 :C2 :C2 :C2 → {R[8] → 2, R[11] → 2} :C3 → {R[8] → 2, R[11] → 2, R[12] → 1, R[13] → 2} :C1 → {R[8] → 2, R[13] → 2, R[12] → 1} :C2 → {R[8] → 2, R[11] → 2,
On its side, the LinearCounting algorithm relies on the fraction of empty registers (R[i] = 0) to estimate the cardinality of H . According to [START_REF] Whang | A lineartime probabilistic counting algorithm for database applications[END_REF], an estimate of the cardinality of H is given by the equation -m × ln(V) where V is the number of empty registers divides by the total number of registers m.

To illustrate how HLL++ works, consider the example of Figure 3b where the GroupKey :c3 is incrementally filled with elements :o1, :c3, :c1, :o1, :c3 and :c2 to finally obtain 4 distinct elements. In Figure 4, the same elements are added to an HLL++ set H 0.26 c3 where the error rate = 26% and the number of registers m = (1.04/0.26) 2 = 16. Each element is mapped to a 64 bit hash value. The first p = log 2 (16) = 4 bits are represented in blue and used for identifying the register R[i] to update. The number of leading zeros k after the first p = 4 bits are highlighted in red and used to update

R[i], if k > R[i].
Once all the elements are inserted in H 0.26 c3 , HLL++ estimates the number of distinct elements for the GroupKey :c3 using either the HLL or the LinearCounting algorithm. According to [START_REF] Heule | Hyperloglog in practice: Algorithmic engineering of a state of the art cardinality estimation algorithm[END_REF], in our example, the LinearCounting algorithm is used and the cardinality is estimated as -16 × ln(12/16) ≈ 4.60.

Partial Aggregations and HLL++

To estimate the number of distinct elements in a multiset, with a fixed error rate , we introduced a new aggregate function COUNT D (cf Table 1). To follow the partial aggregations model, COUNT D has to provide an aggregate function f 1 , a merge operator and a function h as defined in Definition 3. Functions f 1 , and h of COUNT τ D are respectively mapped to HLL add , HLL merge and HLL count , where HLL merge merges two HLL++ sets H 1 and H 2 of m registers into a new HLL++ set H 3 such as

H 3 .R[i] = max(H 1 .R[i], H 2 .R[i]) for i ∈ 1..m.
Figure 5a illustrates how a smart client computes Q 2 over D 1 with a fixed error rate = 26% using COUNT 0.26 D . At each quantum q i , two new mappings are produced in ω i . For each GroupKey in ω i , the server creates a HLL++ set. During the first quan-tum, two mappings with two different objects :o1 and :c3 are produced for the GroupKey :c3. Using the HLL 0.26 add operation, :o1 and :c3 are assigned to registers 13 and 8, respectively. Both R [START_REF] Li | Approximate query processing: What is new and where to go?[END_REF] and R [START_REF] Grall | Processing sparql aggregate queries with web preemption[END_REF] are updated from 0 to 2 because both :o1 and :c3 hash values have two leading zeros. At the end of the quantum, registers are sent to the client.

Compared to the HyperLogLog algorithm, HLL++ does not necessarily send all the registers to the client. To fit the memory efficiency criteria, HLL++ can store the array R using either a sparse or a full representation [START_REF] Heule | Hyperloglog in practice: Algorithmic engineering of a state of the art cardinality estimation algorithm[END_REF]. The sparse representation is used when most of the registers are empty and avoid transferring all registers to the client. Thus, for an error rate of 2%, the 1.5KBytes per GroupKey is just a worst-case space complexity for HLL++. Typically, for small sets, the data transfer will be at most equivalent to transferring the sets.

To go back to our example, when the client receives the registers, it uses the HLL 0.26 merge operation to merge the incoming registers with the local ones. The client repeats the same process for all quanta until the query complete. When the query completes, the client uses the HLL 0.26 count operation to estimate the number of distinct elements per GroupKey.

The duration of the quantum has a significant impact on the data transfer. Long quanta reduce data transfer as HLL++ sets are better used. However, long quanta are also likely to gather many GroupKeys that require each to store a HLL++ set. Even if HLL++ is efficient in terms of space and can adapt its memory usage, gathering many GroupKeys may exhaust the memory of the server. This issue is already pointed out as the many-distinct count problem [START_REF] Ting | Approximate distinct counts for billions of datasets[END_REF]. In the context of Web preemption, this issue can be avoided by limiting the memory dedicated to the aggregation results, so that a quantum only deals with a bounded number of GroupKeys. Once the limit is reached, even if the quantum is not exhausted, the query is suspended and partial results are returned to the client. Of course, such an approach just moves the many-distinct count problem to the client. However, the client memory is not a shared resource.

Implementing Decomposable Aggregate Functions

Algorithm 1 presents the general algorithm to compute partial aggregates on a preemptable server. To evaluate an aggregate query Q, the algorithm starts by

9 foreach (F, f, v) ∈ A do 10 Ω ← γ(F, f, {µ}) ; µ[v] ← Ω[ E µ ]
11 Function Save(): ComputeAggregates(E, A, µc)

Algorithm 3: Merge two sets of solution mappings, Y into X

Input: E: list of expressions used by the GROUP BY A: set of 3-tuple (F, f, v) where f is an aggregate function, F a list of expressions and v a variable to bind the result of f X, Y: two sets of solution mappings 1 Procedure Merge(E, A, X, Y):

2 foreach µ ∈ X do 3 foreach µ ∈ Y with E µ = E µ do 4 foreach (F, f, v) ∈ A do 5 if f ∈ {COUNT, SUM} then 6 µ[v] ← µ[v] + µ [v] 7 else if f = SaC then 8 µ[v] ← µ[v] ⊕ µ [v] 9 else if f = MIN then 10 µ[v] ← Min(µ[v], µ [v]) 11 else if f MAX then 12 µ[v] ← Max(µ[v], µ [v]) 13 else if f = COUNT D then 14 µ[v] ← HLL merge (µ[v], µ [v]) 15 else 16 µ[v] ← µ[v] ∪ µ [v] 17 foreach µ ∈ Y, µ ∈ X, E µ = E µ do 18 X ← X ∪ {µ }
building the physical plan of Q, i.e. a pipeline of preemptable iterators (Lines 2-5), that will be consumed by Algorithm 1. Algorithm 2 defines a new preemptable aggregates iterator for computing aggregate functions. When the GetNext() method is called, the new iterator consumes a solution mapping µ from its predecessor (Line 3), and computes aggregate functions on µ (Line 5). As aggregate functions are computed one mapping at a time, this iterator is preemptable, i.e. it can be saved and resumed in constant time. During a quantum, Algorithm 1 consumes solution mappings from the pipeline of iterators, and merges them into Ω (Lines 10-18), using the Merge operation defined in Algorithm 3.

The Merge(E,A,X,Y) operation merges two sets of solution mappings X and Y. For each µ ∈ X, it finds a µ ∈ Y that has the same GroupKey as µ (Line 3). Then, Algorithm 3 iterates over all aggregation results in µ (Lines 4-16) to merge them with their equivalent in µ, using the different merge operators defined in Table 1.

When the quantum is exhausted, the server waits for the non-interruptible section of Algorithm 1 to complete. Thus, no solution mappings are discarded, and the merge operation is guaranteed to be applied to every solution mapping. The non-interruptible section can block the program for at most the time to merge a single solution mapping in Ω, which can be done in constant time. Then, Algorithm 1 suspends the query Q, and sends the suspended query Q s and the partial aggregates Ω to the client (Lines 23-25).

To avoid the many-count distinct problem, i.e. exhaust the server memory, the size of Ω is bounded to a predefined constant pageS ize. If the size of Ω exceeds pageS ize, Algorithm 1 stops computing new aggregation results (Line 11) and the query is suspended.

The evaluation of SPARQL aggregates on the server requires defining different parameters: the duration of a quantum, the maximum space allocated to the aggregation results, and the error rate when the COUNT function is used. Defining these parameters is left to the server administrator. Q: SPARQL aggregate query S : SAGE server URL

1 Function EvalQuery(Q, S ):

2 E ← GROUP BY expressions of Q 3 A ← Aggregate functions of Q 4 Q ← DecomposeQuery(Q) 5 Ω ← ∅ 6 repeat 7 (Ω , Q ) ← Evaluate Q at S 8 Merge(E, A, Ω, Ω ) 9 until Q = nil 10
ComputeAggregates(Ω, A)

11
return Ω

12 Procedure ComputeAggregates(Ω, A):

13 foreach µ ∈ Ω do 14 foreach (F, f, v) ∈ A do 15 if f = AVG then 16 (s, c) ← µ[v] ; µ[v] ← s/c 17 else if f = SUMD then 18 µ[v] ← S U M(µ[v]) 19 else if f = AVGD then 20 µ[v] ← AVG(µ[v]) 21 else if f = COUNTD then 22 µ[v] ← |µ[v]| 23 else if f = COUNT D then 24 µ[v] = HLL count (µ[v])
As the server computes only partial aggregates, it relies on the client to compute SPARQL aggregates, as shown in Algorithm 4. To execute a SPARQL aggregate query Q, the client first decomposes Q into Q to replace the AVG aggregate function and the DISTINCT modifier as described in Section 4.2. Then, the client submits Q to the SAGE server S , and follows the next links sent by S to fetch and merge all query results, following the Web preemption model (Lines 6-9). Finally, the client transforms the set of partial aggregation results returned by the server to produce the final aggregation results (Line 10). For each solution mapping µ ∈ Ω, the client applies the appropriate h function for each of the aggregate functions, as defined in Table 1.

Experimental Study

The purpose of the experimental study is to answer the following questions: (1) What is the data transfer reduction obtained with partial aggregations? (2) What is the speed up obtained with partial aggregations? (3) What is the impact of quantum on data transfer and execution time? (4) Does estimating the result of count-distinct queries reduce data transfer? (5) Does the observed error rate matches the theoretically guarantees provided by the HLL++ algorithm?

The partial aggregations approach has been implemented as an extension of the SAGE query engine 3 . The SAGE server has been extended with the new operator described in Algorithm 2. Python SAGE-agg and SAGE-approx clients have been extended with Algorithm 4. SAGE-agg uses the COUNT D function to compute count-distinct queries, while SAGE-approx uses the COUNT D function. The source code of the experimental study as well as all configuration files are available in the project repository at https://github. com/JulienDavat/sage-agg-experiments.

Dataset and Queries: The workload (SP) used in the experimental study is composed of 18 SPARQL aggregate queries extracted from the SPORTAL queries [START_REF] Hasnain | Sportal: profiling the content of public sparql endpoints[END_REF] (queries without ASK and FILTER). Most of the extracted queries use the DISTINCT modifier. SPORTAL queries are challenging because they aim to build VoID descriptions of RDF datasets 4 . As reported in [START_REF] Hasnain | Sportal: profiling the content of public sparql endpoints[END_REF], most of the queries cannot complete over the DBpedia public server because of the quotas. Moreover, as depicted in Figure 6, the SPORTAL queries return GroupKeys with different numbers of distinct values; from one to several million on the DBPedia dataset. Having different number of distinct values is important to demon-Fig. 6. Number of GroupKeys for the SPORTAL queries on DBPedia according to the number of distinct values strate that HLL++ is accurate for both small and large cardinalities when the COUNT D function is used.

To study the impact of the DISTINCT modifier on the aggregate queries execution, a new workload, denoted SP-ND, is defined by removing the DISTINCT modifier from the queries of SP.

Both the SP and the SP-ND workloads are run on synthetic and real-world datasets. For the synthetic datasets, the Berlin SPARQL Benchmark (BSBM) is used to generate three datasets of increasing size: BSBM-10, BSBM-100 and BSBM-1k. For the realworld dataset, a fragment of DBpedia v3.5.1 is used. The statistics of each dataset is detailed in Table 2.

Approaches:

The following approaches are compared:

-SaGe: corresponds to the SAGE query engine as defined in [START_REF] Minier | Sage: Web preemption for public sparql query services[END_REF]. The SAGE server is configured with a maximum page size of results of 10 000 mappings. The data are stored in a SQLite database, with Btree indexes on (SPO), (POS) and (OSP).

-SaGe-agg: is configured as SaGe. The size of the aggregation results on the server (MaxAggS ize) is limited to 10MBytes.

-SaGe-approx: is configured as SAGE-agg. To compute count-distinct queries, SAGE-approx uses an error rate = 2%.

-TPF: corresponds to the TPF query engine [START_REF] Verborgh | Triple pattern fragments: a low-cost knowledge graph interface for the web[END_REF]. The TPF server is configured with a page size of 10000 mappings and without Web caches. Data are stored using the HDT format. The TPF smart client is Comunica [START_REF] Taelman | Comunica: a modular sparql query engine for the web[END_REF] (v1.9.4).

-Virtuoso: corresponds to the Virtuoso SPARQL endpoint [START_REF] Erling | Rdf support in the virtuoso dbms[END_REF] (v7.2.4). Virtuoso is configured without quotas and with a a single thread so that Virtuoso delivers complete results and can be fairly compared with other engines.

Servers configurations:

All experiments have been run on the Google Cloud Platform, on a n2-highmem-4 machine with 4 vCPU, 32 GBytes of RAM and a SSD local disk of 375 GBytes.

Experimental results

Data transfer and execution time over BSBM datasets

Figure 7 presents the data transfer and the execution time over BSBM-10, BSBM-100 and BSBM-1k. In this experiment, the SAGE server is configured with a time quantum of 150ms. The plots on the left detail the results for the SP workload, while the plots on the right detail the results for the SP-ND workload.

As expected, Virtuoso without quota performs the best in terms of data transfer and execution time. On the other hand, TPF offers the worst performance as it does not support projections nor joins on the serverside. As a result, TPF transfers a large number of intermediate results and sends many HTTP requests to the server, which has a significant impact on query execution time. Although both SAGE and TPF evaluate SPARQL aggregate queries on the client-side, SAGE delivers better performance than TPF because it supports projections and joins on the server.

Compared to SAGE, SAGE-agg and SAGE-approx drastically reduce data transfer but do not improve the execution time, because partial aggregations do not increase the scanning speed on the disk. When comparing the performance of SAGE-agg and SAGE-approx on the two workloads, we can observe that query processing without the distinct modifier (on the right) is much more efficient in terms of data transfer than with the distinct modifier (on the left).

Without the distinct modifier, SAGE-agg and SAGEapprox are equivalent and transfer only one number per GroupKey, per quantum. Consequently, they can achieve performances that are close to Virtuoso. Note that if the data transfer for Virtuoso is a bit larger than SAGE-agg and SAGE-approx, it is only because of the output format used by the different endpoints. In the best case, SAGE-agg and SAGE-approx can only be as good as Virtuoso.

For queries that use the distinct modifier, SAGE-agg has to transfer all terms observed during a quantum. The only optimization that can be done to reduce data transfer is to remove the duplicates observed during the same quantum. However, those observed during different quanta cannot be removed. Compared to SAGEagg, SAGE-approx significantly improves the evaluation of count-distinct queries in terms of data transfer. For each GroupKey, the HLL++ algorithm transfers at most its m registers (integers). For an error rate of 2%, HLL++ uses m = 4096 registers, which represents a worst-case data transfer of 1.5KBytes [START_REF] Heule | Hyperloglog in practice: Algorithmic engineering of a state of the art cardinality estimation algorithm[END_REF]. For GroupKeys that return a very large number of different terms, 1.5KBytes is not much compared to what it would cost to send all the terms to the client. For GroupKeys that return a small number of different terms, HLL++ transfers only the used registers. For instance, if a GroupKey returns 10 different terms, HLL++ will only transfer at most 10 registers.

Data transfer and execution time over DBPedia

To confirm the results observed on the synthetic datasets, we ran the SP workload on a fragment of DBpedia, using both SAGE-agg, SAGE-approx and Virtuoso. The quantum for SAGE-agg and SAGE-approx has been set to 30 seconds. The results are shown in Figure 8, where the queries (Q2, Q3, Q4, Q5, Q7, Q8, Q9, Q10, Q12, Q13, Q15, Q16) labeled in blue are the ones that use the distinct modifier.

As expected, Virtuoso delivers the best performance in terms of data transfer and execution time. In terms of execution time, the differences between Virtuoso and both SAGE-agg and SAGE-approx are mainly due to a lack of query optimizations in the SAGE-agg and SAGE-approx implementations; no projection push-Fig. 8. Performance obtained in terms of execution time and data transfer with the SP workload on the DBpedia dataset down, no merge-joins, etc. In terms of data transfer, Virtuoso is optimal as it computes the full aggregations on the server-side and transfers only the final results. Compared to Virtuoso, SAGE-agg and SAGE-approx perform only partial aggregations on the server-side. Nevertheless, Virtuoso cannot ensure that all queries terminate under quotas. The red dotted line in Figure 8 corresponds to a quota of 60s. As we can see, queries Q5, Q6, Q8, Q10, Q12, Q13, Q14, Q15, Q16, Q17 and Q18 do not terminate, i.e. two third of the queries are interrupted after 60s and return no results.

Compared to Virtuoso, the SAGE server does not interrupt queries. The queries are just suspended after a time quantum and resumed later. Consequently, both SAGE-agg and SAGE-approx ensure termination of all queries. Finally, as expected, SAGE-approx dras- Error rates over DBpedia SAGE-approx approximates the result of countdistinct queries and hence, there is a potential for error. To ensure that the theoretical guarantees on the error rate holds in practice, we measured the error rate for each GroupKey returned by the queries of the SP workload on DBpedia. To compute the error rate, we used SaGe-agg as the ground-truth. In Figure 9, GroupKeys are grouped according to the number of distinct values returned, and the average error rate is computed for each group. As expected, although HLL is a very powerful approximate algorithm on large cardinalities, it fails on small cardinalities. Compared to HLL, HLL++ is a good estimator to estimate the result of SPARQL count-distinct queries. By adapting the algorithm used to compute the estimate according to the cardinalities, HLL++ achieves an error rate lower than 2% for both small and large cardinalities.

Impact of time quantum

To study the impact of the quantum on data transfer and query execution time, the two workloads have been run with different time quantum. Figure 10 reports the results of running SAGE, SAGE-agg, SAGEapprox and Virtuoso with a quantum of 75ms, 150ms, 1.5sec and 15sec on BSBM-1k. The plots on the left detail the results for the SP workload and on the right the SP-ND workload.

As we can see, increasing the quantum does not significantly improve the execution time. The speed of scans does not change whatever the value of the quantum. However, increasing the quantum reduces the data transfer for SAGE-agg and SAGE-approx on both workloads. Indeed, increasing the quantum allows a better use of partial aggregations. The less often a query is interrupted, the less likely it is to transfer the same GroupKeys multiple times. That is why Fig. 10. Quantum impact on the execution of SP (left) and SP-ND (right) workloads on the BSBM-1k dataset there is a significant drop between 150ms and 1.5sec in Figure 10. With a quantum of 1.5sec, queries are interrupted 10 times less often than with a quantum of 150ms. Between 75ms and 150ms, queries are only interrupted half as often, consequently, the improvement is not as important as between 150ms and 1.5sec. Finally, with a quantum of 15sec, data transfer is optimal as all queries terminate between 1.5 and 15 seconds.

Finally, we can observe that SAGE-agg is less impacted by the quantum duration than SAGE-approx. Even if higher quanta allow to deduplicate more terms, the number of elements transferred by SAGE-agg remains important and dominates the data transfer.

Discussion

The results show that using probabilistic data structures to compute count-distinct queries significantly reduces data transfer. However, the current implementation still has poor performance in terms of execution time, which limits its application to very large knowledge graphs such as Wikidata or DBpedia. As mentioned in the experimental study, these performance issues are due to a lack of query optimizations on the SAGE server. The simple application of state-of-art optimization techniques, including filter and projection push-down, aggregate push-down or merge-joins should significantly improve performance.

Moreover, the current approach only proposes to improve the evaluation of count-distinct queries. To evaluate avg-distinct and sum-distinct queries, the server still has to transfer all the elements to the client. Un-fortunately, to the best of our knowledge, there is currently no probabilistic data structure that allows estimating a distinct sum.

Finally, to avoid the many-count distinct problem, we currently rely on Web preemption. By limiting the memory dedicated to the aggregation results, we ensure that a quantum only processes a limited number of GroupKeys. Such a solution has several drawbacks. First, it prevents us from using large quantum. Indeed, queries that return a large number of GroupKeys will reach the memory limit before reaching the end of the quantum. As a result, the HLL++ sets will be less well utilized, queries will require more quanta to complete, which means more HTTP calls, more data transfer and therefore worse execution time. Secondly, it just shifts the problem on the client-side. To address the manycount distinct problem, different approaches [START_REF] Ting | Approximate distinct counts for billions of datasets[END_REF][START_REF] Xiao | Hyper-compact virtual estimators for big network data based on register sharing[END_REF] propose to make many HLL sketches share the same registers. By sharing registers, the server could deal with more GroupKeys before exhausting its memory, but none of these approaches propose solutions to handle HLL++ sketches. However, as HLL++ sketches rely both on HLL and the LinearCounting algorithm, it could be possible to adapt the HLL++ algorithm so that HLL registers are shared between different HLL++ sketches.

Conclusion and Future Works

In this paper, we have extended the partial aggregation operator presented in [START_REF] Grall | Processing sparql aggregate queries with web preemption[END_REF] in order to improve the evaluation of count-distinct aggregate queries. We have shown how the decomposability property of the HyperLogLog++ algorithm can be used to integrate HyperLogLog++ sketches in our framework. We have demonstrated experimentally that using Hyper-LogLog++ sketches drastically reduce data transfer for SPARQL count-distinct queries. Compared to related approaches, the presented solution ensures that all GroupKeys are discovered in a single pass with strong guarantees on the error rate.

The next step is to extend this approach to handle large knowledge graphs. One way to scale up is to parallelize the evaluation of SPARQL aggregate queries. Currently, Web preemption does not support intraquery parallelization techniques. Defining how to suspend and resume parallel scans is clearly part of our research agenda. Finally, addressing the many-count distinct problem on the server-side could reduce the data transfer, and the memory consumption on both the server and the client.

Fig. 1 .

 1 Fig. 1. Aggregate queries Q1 and Q 2 over G 1

  SELECT ?c (COUNT( ? o ) as ?z ) WHERE { ?s : a ?c ; ?p ?o ; : p1 : o1 } GROUP BY ?c

1 a

 1 SCAN can be resumed in O(log(|D|)) with B-Tree indexes on SPO, POS and OSP, where |D| is the size of the dataset D.

:Fig. 3 .

 3 Fig. 3. Evaluation of Q 1 and Q 2 on the RDF graph G 1 with a partial aggregation operator.

  aggregates A i γ( ?O , COUNT 0.26

15 if µc = nil then 16

 1516

Algorithm 4 :

 4 Client-Side evaluation of partial aggregatesInput:

Fig. 7 .

 7 Fig. 7. Data Transfer and execution time for BSBM-10, BSBM-100 and BSBM-1k, when running the SP (left) and SP-ND (right) workloads Evaluation Metrics: Presented results correspond to the average of three successive executions of the query workloads. (i) Data transfer: is the number of bytes transferred to the client when evaluating a query. (ii) Execution time: is the time between the start of the query and the production of the final results by the client. (iii) Error rate: is defined as the difference between the real cardinality c and the estimated cardinality ĉ : (1 -(min(c, ĉ)/max(c, ĉ)) × 100

Fig. 9 .

 9 Fig. 9. Average error rate for the GroupKeys of the SP workload queries on DBpedia according to the number of distinct values

  Ω , card Ω ), where S Ω is the base set of mappings and card Ω is the multiplicity function which assigns a cardinality to each element of S Ω . For simplicity, µ ∈ S Ω is often written µ ∈ Ω.

	and P2 are graph patterns, then expressions (P1 AND P2), (P1 OPT P2) and (P1 UNION P2) are graph patterns (a conjunctive graph pattern, an optional graph pattern and an union graph pattern, respectively). -If P is a graph pattern and R is a SPARQL built-in condition, then the expression (P FILTER R) is a graph pattern (a filter graph pattern). The evaluation of a graph pattern P over an RDF graph G denoted by P G produces a multiset of solution mappings Ω = (S :s1 :p1 :o1 . :s1 :a :c2, :c3. :s2 :p1 :o1 . :s2 :a :c1, :c3. (a) RDF Graph G1 (COUNT( ? o ) AS ?z ) SELECT ?c WHERE { ?s : a ?c . ?s ?p ?o . ?s : p1 : o1 } GROUP BY ?c SELECT ?c (COUNT( DISTINCT ( ? o ) ) AS ?z ) WHERE { ?s : a ?c . ?s ?p ?o . ?s : p1 : o1 } GROUP BY ?c (b) SPARQL query Q1 (c) SPARQL query Q2

Table 1

 1

	Decomposition of SPARQL aggregate functions with and without
	the DISTINCT modifier
	(a) Aggregate functions without the DISTINCT modifier
	SPARQL Aggregate functions

  The intermediate aggregation results for

			SELECT ?c (COUNT( ? o ) AS ?z )
		WHERE {
			?s : a ?c . ?s ?p ?o . ?s : p1 : o1
			} GROUP BY ?c
	quantum	Server ?C ?O	Partial aggregates A i γ( ?O , COUNT, ω i )		Client Merge(Ω, ω i )
	q1 ω 1	:C3 :C3	:O1 :C3	:C3 → 2	next	:C3 → 2
	q2	:C3	:C1	:C3 → 1		:C3 → 3
	ω 2 q3 ω 3 q4 ω 4	:C1 :C1 :C1 :C3 :C3	:O1 :C3 :C1 :O1 :C3	:C1 → 1 :C1 → 2 :C3 → 2	next next	:C1 → 1 :C3 → 3 :C1 → 3 :C1 → 3 :C3 → 5
	q5 ω 5	:C3 :C2	:C2 :O1	:C2 → 1 :C3 → 1	next	:C2 → 1 :C1 → 3 :C3 → 6
	q6 ω 6	:C2 :C2	:C2 :C3	:C1 → 2	next	:C2 → 3 :C1 → 3 :C3 → 6
				(a) Evaluation of Q1
	quantum	?C Server ?O		Partial aggregates Ai

γ( ?O , CT, ωi) Client Merge(Ω, ωi)

  Fig.[START_REF] Erling | Rdf support in the virtuoso dbms[END_REF]. Approximate count-distinct for the GroupKey :c3 of query Q2, on the RDF graph G 1 , using HLL++ with an error rate of 26%

		object	64-bit hash		register value
		:o1	110100100 56		R[1]		0
						//
		:c3	100000100 56	max	R[8]		2
	time	:c1	110001000 56		R[9] R[10]		0 0
		:o1	110100100 56		R[11]		2
		:c3	100000100 56		R[12]		1
		:c2	101100100 56	max	R[13]	//	2
					R[16]		0
				ĉ = -16 × ln( 12 16 ) ≈ 4.60

Algorithm 1 :

 1 Server-Side evaluation of partial aggregates

		Require:
		quantum: Duration of a quantum
		pageS ize: Maximum size of a result page
		Input: Q: SPARQL aggregate query
	1 Function EvalQuery(Q):
	2	if Q is suspended query then
	3	iterator ← Resume(Q)
	4	else
	5	iterator ← ParseQuery(Q)
	9	try:
	10	EvalQuantum( quantum ):
	11	repeat
	12	µ ← iterator.GetNext()
	13	non interruptible
	14	if µ = nil then
	15	Merge(E, A, Ω, {µ})
	16	µ ← nil
	17	else done ← T rue
	18	until done ∨ S ize(Ω)	pageS ize
	19	catch QuantumExhausted:
	6	return µc
	7	else return nil

6

E ← GROUP BY expressions of

Q 7 A ← Aggregate functions of Q 8 Ω ← ∅ ; done ← False 20 if µ = nil then 21 Merge(E, A, Ω, {µ})

22 finally: 23 if done then Qs ← nil 24 else Qs ← S uspend(iterator) 25 return (Ω, Qs) Algorithm 2: Server-Side Preemptable SPARQL Aggregates Iterator Require: Ip: predecessor in the pipeline of iterators E: list of expressions used by the GROUP BY A: set of 3-tuple (F, f, v) where f is an aggregate function, F a list of expressions and v a variable to bind the result of f Data: µc: the last element read from Ip 1 Function GetNext():

2 if ∀(F, f, v), v ∈ dom(µc) then return µc 3 µc ← Ip.GetNext() 4 if µc = nil then

5

ComputeAggregates(E, A, µc)

8 Procedure ComputeAggregates(E, A, µ):

Table 2

 2 Statistics of RDF datasets used in the experimental study

	RDF Dataset	# Triples	# Subjects	# Predicates	# Objects	# Classes
	BSBM-10	4 987	614	40	1 920	11
	BSBM-100	40 177	4 174	40	11 012	22
	BSBM-1k	371 911	36433	40	86202	103
	DBpedia 3.5.1	100M	2 835 701	35 168	26 840 695	342

In this paper, for simplicity, only aggregate queries with Basic Graph Patterns and no OPTIONAL clauses are considered

https://sage.univ-nantes.fr

https://www.w3.org/TR/void/

Acknowledgements

This work is supported by the ANR DeKaloG (Decentralized Knowledge Graphs) project, ANR-19-CE23-0014 -AAPG2019-program CE23.