Iztok Savnik
email: iztok.savnik@upr.si

Kiyoshi Nitta
email: knitta@yahoo-corp.jp

Algebra of RDF Graphs for Querying Large-Scale Distributed Triple-Store

Large-scale RDF graph databases stored in shared-nothing clusters require query processing engine that can effectively exploit highly parallel computation environment. We propose algebra of RDF graphs and its physical counterpart, physical algebra of RDF graphs, designed to implement queries as distributed dataflow programs that run on cluster of servers. Operations of algebra reflect the characteristic features of RDF graph data model while they are tied to the technology provided by relational query execution systems. Algebra of RDF graphs allows for the expression of pipelined and partitioned parallelism. Preliminary experimental results show that proposed algebra and architecture of query execution system scale well with large clusters of data servers.

Introduction

Recent development of graph-based semantic web shows the enormous interest of society to construct a detailed knowledge base (graph) including properties of categories from all popular areas of human activities. Knowledge bases such as Knowledge Graph, Wikidata, YAGO and Knowledge Vault currently include from 1000 up to 350.000 categories, up to 570 Mega instances of categories, up to 35.000 relationship types, and, up to 18 Giga relation instances [START_REF] Dong | Knowledge vault: A web-scale approach to probabilistic knowledge fusion[END_REF]. However, from many aspects existent knowledge graphs are in their infant stage-more systematic use of intelligent tools for extracting the knowledge from various data sources has just begun.

The need for triple-store systems capable to store and manage from Tera (10 12 ) towards Peta triples is obvious. The scalability of storage system and query processing system to this amount of data is currently possible by using large-scale distribution of data into shared-nothing clusters. Query execution system in such environment must be able to employ various types of parallelism to allow simultaneous execution of huge amount of queries and provide reasonable response time.

Distributed triple-store big3store is based on dataflow architecture of query processing. Each query is a tree of algebra operations that is dynamically mapped to the tree composed of processes interconnected by streams of graphs, i.e., sets of triples. The scheduler that maps query trees to set of processes balances the computation load among the servers of cluster.

Triple-store of big3store is distributed into columns that store replicas of partitions into rows-cluster data servers. Data distribution is achieved by means of semantic distribution function [START_REF] Savnik | Semantic partitioning method for very large rdf graphs[END_REF] that splits the triples on the basis of the relation of each particular triple to the taxonomy of RDF classes and properties.

Algebra of RDF graphs is an abstract model used for the implementation of query execution system. Algebra is defined using set semantics-inputs and outputs of operations are sets of graphs. We present the denotational semantics of algebra and its implementation in the form of physical algebra that is further mapped to sets of processes implementing algebra operations. The architecture of query execution system based on algebra allows for the use of pipelined and partitioned parallelism [START_REF] Dewitt | Parallel database systems: The future of high performance database processing[END_REF].

Programming environment of parallel programming language Erlang [START_REF] Armstrong | Programming Erlang: Software for a Concurrent World[END_REF] is used for the implementation of big3store. Erlang, together with database management system Mnesia that is tightly integrated with Erlang, may represent alternative data processing system for big data to Hadoop [START_REF] White | Hadoop: The Definitive Guide[END_REF]. Indeed, it provides simple and robust parallel programming environment allowing processes to be effectively used in cluster of servers, it incorporates mechanisms that allow for the implementation of reasonable level of fault-tolerance, and, it integrates low-level database system appropriate for telecommunication applications that includes key-value indexes comparable to those of Hadoop storage system.

The contributions of this paper are the following. The architecture of distributed query execution system for processing large-scale RDF graphs based on algebra of RDF graphs is proposed. Query processor uses left-deep query trees to implement pipelined parallelism of algebra operations. Furthermore, it employs semantic triple distribution function [START_REF] Savnik | Semantic partitioning method for very large rdf graphs[END_REF] to achieve highly flexible partitioned parallelism. Access methods for tripe-patterns that address large partitions of triple-base are distributed to larger number of data servers, while the queries that address small partitions are executed on a single server. Query processor uses affinity scheduling, i.e., two level scheduling that persists to allocate the same data servers for execution of algebra operations for particular user. Finally, it uses key-value indexes in a similar manner to Hadoop to access data triples and to implement index-based nested-loop join operations.

The rest of the paper is organized as follows. The following Section 2 presents algebras of RDF graphs closely related to big3store algebra. Section 3 gives formal definition of algebra of RDF graphs and describes its physical counterpart, physical algebra of RDF graphs. The architecture of big3store query execution system together with detailed description of algebra implementation, is presented in Section 4. Preliminary experimental results are described in Section 5. Finally, Section 6 gives some conclusions and presents further work.

Related work

Algebra of RDF graphs implemented in distributed triple-store big3store is based on relational algebra and technology of relational database management systems [START_REF] Graefe | Query evaluation techniques for large databases[END_REF][START_REF] Graefe | Dynamic query evaluation plans: Some course corrections?[END_REF]. Database algebras are by nature functional languages where inputs and outputs of algebra operations can be treated as input and output flows of database objects. Operations of database algebra can be combined to form graph structure where operations (nodes) are interconnected by flows of objects [START_REF] Dewitt | Parallel database systems: The future of high performance database processing[END_REF].

The design of big3store algebra of graphs follows the leading ideas of relational algebra [START_REF] Codd | A relational model of data for large shared data banks[END_REF] while we identified and incorporated in it the salient features of triple-store data model. Firstly, instead of access methods scanning relational tables we use triple-pattern based access method to triple-store that can use all possible indexes on SPO attributes. Secondly, the results of algebra operations are not relations-sets of tuplesbut sets of graphs. Consequently, operations select, project and join are adapted for graphs. Selection is based on expressions defined by means of graph nodes. Similarly, operation project eliminates graph edges. Finally, operation join is defined on graphs by introducing graph matching as join predicate.

Similar approach to definition of algebra for querying triple-stores are proposed by Angles and Gutierrez in [START_REF] Angles | The expressive power of sparql[END_REF]. Their formalization of SPARQL operations is based on mappings that follow semantics of triples. While their definition is tuned for studying expressive power of the language, our work is focused more on the implementation of algebra of graphs in shared-nothing cluster. They have shown in the paper that SPARQL have equivalent expressive power to non-recursive Datalog with negation that has, in turn, equivalent expressive to classical relational algebra.

Schmidt et.al. have proposed SPARQL algebra [START_REF] Schmidt | Foundations of sparql query optimization[END_REF] to be used as foundations for SPARQL query optimization. They have defined set-based semantics for SPARQL by introducing SPARQL set algebra including similar operations to our algebra of graphs. They have identified fragments of SPARQL together with their complexity classes. For instance, they have shown that OPTIONAL-free fragments of SPARQL are either NPcomplete or in PTIME. Furthermore, they have introduced algebraic equivalence rules that can be used for SPARQL query optimization, and, extensions of classical chase algorithm for optimization of AND queries.

Cyganiak proposed in [START_REF] Cyganiak | A relational algebra for sparql[END_REF] the use of relational algebra for SPARQL query processing. He presents the transformation from SPARQL into abstract relational algebra and shows differences between semantics of SPARQL and relational model. This approach allows for direct use of relational query optimization and query evaluation techniques for processing SPARQL queries. The transformation from relational algebra to SQL is defined. In comparison to Cyganiak's proposal, our approach focuses on distributed implementation of algebra of graphs while, in the similar manner, we use knowledge and technology gathered in area of relational systems for the implementation of triple-store database system.

Algebra of RDF graphs

Algebra of RDF graphs is a functional language defined on sets of RDF graphs. Inputs and outputs of algebra operations are sets of RDF graphs that are linked to other operations forming in this way a tree. As we will show later, algebra expressions i.e. trees of algebra operations are converted to trees of Erlang processes that can be located on different data servers.

Let us first define the basic terminology used in presentation. Let I be the set of URI-s, B the set of blanks and L be the set of literals. Let us also define sets S = I ∪ B, P = I, and

O = I ∪ B ∪ L.
RDF triple is a triple (s, p, o) ∈ S × P × O. RDF graph g ⊆ S × P × O is a set of triples. Set of all graphs will be denoted as G. We suppose the existence of a set of variables V and the set of terms

T = O ∪ V . Term t ∈ T is ground if t ∈ O.
We say that RDF graph g 1 is sub-graph of g 2 , denoted g 1 g 2 , if all triples in g 1 are also triples from g 2 .

Ground graphs and graph patterns

Triple pattern (s, p, o) ∈ (S ∪ V ) × (P ∪ V ) × (O ∪ V ) is a triple that can include variables as components. Graph pattern gp ⊆ (S ∪ V ) × (P ∪ V ) × (O ∪ V
) is a set of triple patterns, i.e., graph defined as set of triples that can include variables as components. The set of all graph patterns is in the sequel denoted as G P .

We will separate between ground and abstract entities. Ground triples are triples that include ground terms. Abstract triples, that can include variables, are triple patterns. Similarly, ground graphs are graphs that include triples composed of ground values, and, graph patterns represent abstract graphs that stand for a set of graphs from a given triple-store.

To be able to determine set of variables included in graph pattern gp we define function vars : G P → P(V ).

Matching of graphs.

Let us now define relationship "match", denoted as ∼, between graphs including graph patterns. Graphs g 1 and g 2 match, denoted g 1 ∼ g 2 , iff the following conditions hold.

1. Two terms t 1 , t 2 ∈ T match, written t 1 ∼ t 2 , if either t 1 and t 2 are ground and t 1 = t 2 , or, one of values is variable and the other is ground value. 2. Matching between two triples r 1 and r 2 exists, written r 1 ∼ r 2 , if all components of r 1 and r 2 match. 3. Graph g 1 matches graph g 2 , written g 1 ∼ g 2 , when there exists bijection alpha :

g 1 → g 2 so that each triple t 1 ∈ g 1 matches alpha(t 1 ) = t 2 ∈ g 2 .
Let gp be graph pattern. Function val

: V × G P × G → O maps variables v ∈ vars(gp), graph patterns gp ∈ G p and ground graphs g ∈ G that match gp to values o ∈ O. Let t 1 ∈ gp be triple that includes variable v, then val(v, gp, g) = o is component of triple alpha(t 1 ) = t 2 ∈ g that corresponds to v in gp.
Interpretation of graph pattern. Interpretation of graph pattern gp in database of triples storing graph db is a set of all sub-graphs g of db that match gp.

gp db = {g | g db ∧ g ∼ gp}
Special case of graph pattern is triple pattern tp where complete graph is one single triple that can include variables possibly in all three positions. Interpretation of tp is a set of all triples from db that match tp.

Triple patterns represent graph counterpart of relational access methods [START_REF] Carey | Storage management for objects in exodus[END_REF]. They are always the leafs of query tree. Implementation of query node for a given triple pattern can use SPO indexes to access ground triples.

Definition of algebra

Let us now present algebra of RDF graphs. We denote graph query as Q, triple pattern as T P , selection condition as C, condition operations as OP , sets of variables as SV , and, variables as V . Syntax of algebra is defined as follows.

Q ::= T P | select(Q, C) | project(Q, SV )| join(Q, Q) | union(Q, Q) | intsc(Q, Q) | diff (Q, Q) | leftjoin(Q, Q) T P ::= (S | V, P | V, O | V ) C ::= V OP V | V OP O | C ∧ C | C ∨ C | ¬ C OP ::= = | = | > | ≥ | < | ≤ SV ::= {V +} S ::= URI | Blank-Node P ::= URI O ::= URI | Blank-Node | Literal V ::= ?a .. ?z
We extend previously defined function vars to queries. Let (Q) be the set of all queries. The function vars : Q → P(V ) maps each query to the set of variables that are included in the query. Let us now present the denotational semantics of RDF algebra by defining the interpretation of each particular operation.

Access paths to database of triples storing graph db are defined using triple patterns

(t 1 , t 2 , t 3 ) where t 1 ∈ (S ∪ V ), t 2 ∈ (P ∪ V ) and t 3 ∈ (O ∪ V ). (t 1 , t 2 , t 3 ) db = { (s, p, o) | (s, p, o) db ∧ (s, p, o) ∼ (t 1 , t 2 , t 3 ) }
SPARQL operation FILTER is represented by means of operation select(q, C) where q is query and C is condition expression.

select(q, C) db = { g | g ∈ q db ∧ C(g) = true }
The evaluation of condition C on graph g is defined by the following rules. Value of C(g) is presented by cases of C structure.

-C =?a OP o, where ?a ∈ V and o ∈ O: if val (?a, q, g) OP o = true then C(g) = true, else f alse. -C =?a OP ?b, where ?a, ?b ∈ V : if val (?a, q, g) OP val (?b, q, g) = true then C(g) = true, else f alse.

-C = C 1 ∧ C 2 : if C 1 (g) = true and C 2 (g) = true then C(g) = true, else f alse. -C = C 1 ∨ C 2 : if C 1 (g) = true or C 2 (g) = true then C(g) = true, else f alse. -C = ¬C 1 : if C 1 (g) = f alse then C(g) = true, else f alse.
Operation project(q, s) projects graphs g ∈ q db to graphs composed of triples that include values of variables from set s. Let tr-vars : db × G P → P(V ) denote function that maps triples t ∈ q db and query q to set of variables vs ∈ P(V ) such that for each var ∈ vs value of variable var is a component of t.

project(q, s) db = { g 1 | g ∈ q db ∧ ∀t ∈ g(tr-vars(t, q) ⊆ s =⇒ t ∈ g 1 ) }

Operation join(q 1 , q 2 ) joins two sets of graphs that are interpretations of queries q 1 and q 2 . Let vs be a set of variables vars(q 1 ) ∩ vars(q 2 ). The result of join includes union of graphs g 1 ∈ q 1 db and g 2 ∈ q 2 db such that they agree in the values of all common variables from vs. Observe also that joining two graphs is obtained by making union of graph triples from both graphs. Semantics of operation join can be defined as follows.

join(q 1 , q 2 ) db = { g 1 ∪ g 2 | g 1 ∈ q 1 db ∧ g 2 ∈ q 2 db ∧ ∀v ∈ vs : val (v, q 1 , g 1 ) = val (v, q 2 , g 2 ) }
Set operations are defined in a usual way except that argument sets can include graphs that have heterogeneous structure. Union, intersection and difference of q 1 and q 2 is defined as union, intersection and difference of their interpretations q 1 db and q 2 db . Set operations of RDF algebra are defined as follows.

union(q 1 , q 2 ) db = { g | g ∈ q 1 db ∨ g ∈ q 2 db ) } intsc(q 1 , q 2 ) db = { g | g ∈ q 1 db ∧ q 2 db ) } diff (q 1 , q 2 ) db = { g | g ∈ q 1 db ∧ g ∈ q 2 db }
Finally, to implement SPARQL operation OPTION we define operation leftjoin(q 1 , q 2 ), that is, left outer join of two sets of graphs which are elements of the interpretations of queries q 1 and q 2 . For each pair g 1 ∈ q 1 db and g 2 ∈ q 2 db the result of leftjoin(q 1 , q 2 ) includes either g 1 ∪ g 2 in the case that g 1 can be joined with g 2 , or g 1 if g 1 can not be joined with g 2 . Let vs be a set of variables vars(q 1 ) ∩ vars(q 2 ). Operation leftjoin(q 1 , q 2 ) can be defined as follows.

leftjoin(q 1 , q 2 ) db = { g | g 1 ∈ q 1 db ∧ g 2 ∈ q 2 db ∧ ((is-join(g 1 , g 2 ) ∧ g = g 1 ∪ g 2 ) ∨ (¬is-join(g 1 , g 2 ) ∧ g = g 1 )) },
where is-join(g 1 , g 2 ) is defined as

is-join(g 1 , g 2 ) = ∀v ∈ vs : val(v, q 1 , g 1 ) = val(v, q 2 , g 2 )

Physical algebra of RDF graphs

The design of physical algebra of RDF graphs follows the ideas used for implementation of relational algebra in the frame of relational database management systems [START_REF] Graefe | Query evaluation techniques for large databases[END_REF][START_REF] Graefe | Dynamic query evaluation plans: Some course corrections?[END_REF]. Previously presented operations of RDF algebra are converted into three physical operations: physical access method (AM) AM, physical join denoted join, and, physical set operations union, diff and intsc.

All physical operations now include besides the functionality of their logical counterparts also the functionality of operations select and project operations. Each physical operation therefore includes also select list and project list. There are more reasons for folding more logical operations into single physical operation.

Firstly, it makes sense to perform selection of triples immediately after data needed for selection is available. For instance, immediately after obtaining triples by means of a given triple-pattern access method, they are filtered using selection conditions.

For similar reason operation project is performed as soon as possible. Immediately after some triple in a result RDF graph is not useful, it is dropped. For instance, after using particular triple for performing join operation, it can be omitted from result graph, if of course it is not needed as the result of query, or, for some other operation higher in the query.

The above two rules resemble "pushing" selections and projections down towards the leafs of query tree in relational database systems. Secondly, the reasons for folding selections and projection into AM and join are: 1) the possibility to use join reordering algorithm for query optimization, and, 2) the possibility to implement left-deep as well as bushy query trees [START_REF] Graefe | Query evaluation techniques for large databases[END_REF]-both of them have operations AM as leafs and operations join as the inner nodes of query trees. big3store is currently using left-deep query trees. An example of left-deep query tree with 3 join operations and 4 AM operations is given in Figure 3.3(a). The most important advantage of using left-deep trees is the pipeline that is formed by physical join operations. The results of retrieving graphs from outer query node of join operation is used for index-based access to the inner query node. The graph that is constructed as the result of join operation is then sent to the parent query node i.e. join query node. Consequently, there is no need to store intermediate results during query evaluation.

Triples related to some class with very large number of instances are, by using of semantic distribution algorithm, distributed to more data servers. Therefore, physical operation AM, defined using some triple-pattern, may be executed on number of data servers. Indeed, it is desirable that triple-pattern based operations AM, that tackle large number of triples, are distributed to more servers. The number of servers depends on the size of targeted set of triples. Left-deep query trees can therefore have multiple AM query nodes as presented in Figure 3.

3(b).

Storing and querying huge volumes of data efficiently is currently possible by using shared-nothing cluster architecture [START_REF] Stonebraker | The case for shared nothing[END_REF]. Efficient data servers with huge amount of RAM and disk storage are available as inexpensive commodity hardware. This allows heavy distribution and replication of data as well as massive distribution of query processing on servers forming very large clusters.

Big3store is a data-flow system [START_REF] Babu | Massively parallel databases and mapreduce systems[END_REF] where triple-store is composed of an array of data servers arranged into columns and rows. The complete triple-store is partitioned and distributed into columns based on semantic information attached to triples via triplestore schema. Each column stores a partition of triple store that is replicated to the column rows. Rows of the column therefore contain replicas of triple-store partitions assigned to columns.

While triple store partitioning affects significantly the performance of query executions, it is not the focus of this paper. Detailed presentation of big3store partitioning algorithm is given in [START_REF] Savnik | Semantic partitioning method for very large rdf graphs[END_REF]. Let us here present only some important ideas that have guided the design of triple-store (graph) partitioning.

Hash-based partitioning can not be employed for storing huge triple datasets that are expected to grow significantly in the following decade. Splitting data into a large number of partitions based on hashing can increase significantly the communication traffic among the data servers, especially, when large number of transactions is executed in parallel.

Big3store uses semantic distribution algorithms to partition triple-store into chunks that are suitable for distribution and that are related to a set of schema entities which serve as the key for distribution. Since distribution is based on rich taxonomy of classes spanning more then ten hierarchical levels we can achieve well-defined distribution in the sense that triples defined for classes including large number of instances are split into larger number of chunks. Triples defined for a class that has small number of instances is stored in one chunk. Query distribution must follow data distribution: larger the class of triples addressed by query, larger the number of columns where query will be executed. Figure 4 shows a cluster composed of two types of servers: front servers represented as the nodes of plane A, and data servers represented as the nodes of plane B. Data servers are configured into columns labeled from (a) to (f). A complete database is distributed to columns such that each column stores a portion of the complete database.

The portion of the database stored in a column is replicated into rows labeled from 1 to 5. The number of rows for a particular column is determined dynamically based on the query workload for each particular column. The heavier the load on a given column, larger the number of row data servers chosen for replication. The particular row used for executing a query is selected dynamically based on the current load of servers in a column.

Architecture of query execution system

Erlang programming environment [START_REF] Armstrong | Programming Erlang: Software for a Concurrent World[END_REF] is used for the implementation of big3store as an alternative to Hadoop-like systems [START_REF] White | Hadoop: The Definitive Guide[END_REF]. It provides remarkably simple and effective parallel programming model based on lightweight processes. Erlang processes use "shared nothing" philosophy where the communication among processes is realized solely by means of synchronous and asynchronous messages. State modules b3s_state and node_state are used for efficient sharing of big3store configuration data structures as well as for storing and querying current state of system, such as for instance number of processes running at each particular data server.

Each data server runs one instance of Erlang Mnesia database system that serves as local triple-store. Triple-store is realized by means of a single table triple_store that is accompanied with 6 indexes for all combination of SPO attributes. Mnesia provides transaction-based access to local triple-store through module db_interface. However, since db_interface provides only very simple cursor based access to a single table, local triple-store can be easily replaced by other database engine, and, even file-based access to RDF triples.

Module triple_distributor implements various schema-based algorithms for the distribution of triple-store into a set of cluster columns [START_REF] Savnik | Semantic partitioning method for very large rdf graphs[END_REF].

Session processes are implemented in module session. They serve as user-interface for interaction with users, initiate creation of query-tree processes, control the execution of query tree, and collect the results of query execution. One session can spawn many query trees in parallel.

Module query-tree implements query tree processes that run on front-servers. The main task of query-tree process is to prepare, schedule and initiate the execution of query in the form of query tree composed of query-node processes interconnected by means of streams. Therefore, each query-tree process controls one or more query node processes that constitute query. This is presented in more detail in Sub-section 4.2.

Physical algebra operations AM, join, union, intsc and diff are implemented in query-node modules. Each physical algebra operation is realized as independent Erlang query-node process that runs on one of data servers. All operations are implemented as state machines executing particular protocol: access method to local triplestore, indexed nested-loop join algorithm, or, particular set operation. Sub-sections 4.4 and 4.5 give more detailed description of operations AM and join.

Query-tree process

Query-tree module implements processes that serve as front-end of query tree represented as tree of inter-connected processes running on array of servers. Query is received from session process in the form of a list of triple-patterns augmented with projections and selections as presented in previous section.

Query of type qt_query() presented to query tree process as parameter of message start is converted into tree data structure stored as process dictionary entry. First element of list representing qt_query() is triple-pattern of the lower leftmost query node. Last element of list is triple-pattern of the upper rightmost query node. All other triple-patterns are placed as inner query nodes in order between lower leftmost and upper rightmost.

Query-tree process analyzes the query, computes all components of query node processes to be started, determines cluster columns associated to each query node, and, schedules the rows of columns to be employed for running each particular query node of query tree.

Query-tree process determines the location of each query node in terms of column and row in array (cluster) of servers. Each query node is executed on location determined by query tree process. Firstly, the column of query node is computed by using distribution function that translates triple-patterns to columns in array of servers. Secondly, rows in given columns are scheduled dynamically based on current load of servers in columns.

We use two types of scheduling of column rows to query nodes. First type of scheduling is random assignment of rows to query nodes. The second method used for scheduling is bookkeeping the execution of each particular query node on particular server. Bookkeeping is realized by means of local node_state process. Besides bookkeeping node_state provides a function that selects row server with least load.

Both types of scheduling resemble affinity scheduling where we tend to select the same servers for the same session. The benefits of assigning the same servers (rows) in columns for same session is primarily in utilizing cache of local database management system Mnesia. Experiments are currently under way to present the benefits of affinity scheduling in terms of execution speed.

Triple-pattern query node

Triple-pattern (abbr. TP) query node is implemented as Erlang gen-process. It realizes access method at local triple-store implemented as Mnesia table. Access method is defined by means of triple-pattern, and, it can use index based access to triple-store.

TP query node is implemented as state machine. Input and output messages trigger coroutines that comprise protocol. The states of TP query node are: active, db_access, eos, and inactive. Message start initializes TP query node process and moves state to active. Message eval starts with evaluation and moves state to read_db.

After obtaining triples from local Mnesia database, TP query node process checks them against selection list. The selected triples are sent to parent of query node by using data messages. Protocol requires that each data message is sent to parent process only after receiving empty message from parent process. Therefore, protocol can control the number of data/empty messages that comprise stream. Subsequent messages empty retain state read_db.

After end_of_stream is obtained from function accessing triple-store, state moves to eos. Message eval can be received multiple times if in state active or eos. Finally, stop message puts TP query node to state inactive.

Join query node process

Join query node is implemented as independent Erlang gen-process. Join query node is a state-machine realizing protocol that has incoming and outcoming messages. Each message is implemented as coroutine.

Join query node state-machine has the following states: active, wait_next_outer, wait_next_inner, eos and inactive. Message start initializes the main data structures of join query node and sets state of protocol to active. Message eval start the evaluation of join query node by sending message eval to all children, and, moves state to wait_next_outer. After this, state alternates between wait_next_outer and wait_next_inner. State moves to eos after end of outer streams is detected.

Join query node implements join method which is a variant of indexed nested-loop join algorithm. However, it can have multiple outer query nodes as well as multiple inner query nodes. Since we suppose that every local triple-store indexes triple table on all possible subsets of SPO, all join variables are supported by indexes.

Algorithm of join method is defined as follows. Each graph obtained from outer query nodes causes initialization of inner query nodes by means of message eval. Initialization of inner query nodes uses the values of join variables obtained from outer graph. Only those graphs are retrieved from inner query nodes that match previously obtained outer graph. Each outer and inner graphs are merged into one graph which is tested against selection list and projected using project list of given query node. If selected then resulting graph is sent to parent query node.

Fault tolerance

Erlang programming environment provides the tools for the construction of fail-safe process hierarchies by means of Erlang supervision processes [START_REF] Armstrong | Programming Erlang: Software for a Concurrent World[END_REF]. Important process state data structures are circulating among supervision and supervised processes. The mechanism is integrated into message sending/receiving protocol. Each message received by process A includes the current state of A that was stored by its supervision process. After completing the task, process A returns its new state as a function result. In this way, a supervision process always has up-to-date state of all processes that it manages. In the case that process failure is detected by supervision process it can be restarted using the last state. Furthermore, supervised processes can form various types of structures with specific behavior.

Experimental results

As a preliminary study, the execution time of benchmark queries in big3store are compared with the execution time obtained with Virtuoso [START_REF] Erling | Rdf support in the virtuoso dbms[END_REF].

Benchmark environment comprises six server machines. All of them have the same physical specifications. Each server has two 2.9 GHz Xeon E5-2960 CPU and 256 GB of RAM. One Erlang interpreter process was invoked on each server. Benchmark configuration uses one server as front server and the other five servers as data servers.

Virtuoso was installed on one of the servers that were used to execute big3store. Let us first describe benchmark queries presented in Figure 5. The first group of queries are simple queries that produce small number of intermediate and final results. Query Q1 finds all triples having property <startedOnDate>. It returns 9 triples from YAGO2s. Query Q2 finds all sets (graphs) of triples sharing the same subject that has <startedOnDate> and <endedOnDate> properties in the graph. It returns 1 triple. Query Q3 finds graphs that describe Japanese computer scientists that have created a programming language. Query Q4 returns the creation dates of all things classified as wordnet_language that were created by Ericsson.

Query Q5 compares <Slovenia> and <Japan> by using the same predicate in triple patterns. While it is similar to query Q2, query Q3 returns 241,596 graphs. Query causes large number of intermediate results that are transferred as messages among data servers. Queries Q6, Q7, and Q8 correspond to YAGO queries B1, A1, and B2 from [START_REF] Neumann | Rdf-3x: a risc-style engine for rdf[END_REF], respectively. Because YAGO and YAGO2s [START_REF] Hoffart | Yago2: A spatially and temporally enhanced knowledge base from wikipedia[END_REF] have different schema structures, queries were rewritten to have similar meaning.

Query Q6 returns pairs of actors that were playing in the same film and live in the same city in England. Query Q7 returns graphs describing scientists that were born in a city in Switzerland, and have academic advisor who was born in a city in Germany. Query Q8 returns all married couples that were born in the same city.

Query Q9 was constructed to test circular queries. While current version of query Q9 is specific and executes fast, a circular query can be constructed by removing the first two triple patterns of <Tim_Burton> and <Johnny_Depp>.

Let us now give some comments on comparison presented in Table 1. System big3store executed queries Q1, Q2, Q3 and Q4 faster than Virtuoso. One reason for this is that Mnesia copies complete database in main memory, if it is possible.

It is also apparent that queries that do not produce a lot of traffic execute in big3store much faster that queries that produce a lot of traffic among the servers. There are more reasons for this. Firstly, we currently do not use any data compression, so data is stored in raw form. Secondly, streams are implemented by sending one message for one graph.

The improved version of big3store will map IRIs to integers to optimize storage and transfer speed. Furthermore, the speed of stream transfer will be improved by packing more graphs into bundles that will serve as unit of transfer.

Another reason for slow performance of some queries is in the implementation of cursors in Mnesia. Index-based access to table always returns all results in one package. Consequently, there is almost no parallelism in the execution of queries. The improved version of big3store will replace Mnesia with BerkeleyDB.

Conclusions

Algebra of RDF graphs and its implementation on shared-nothing clusters is presented. Algebra is described by first defining denotational semantics of abstract algebra. Physical algebra corresponding to its abstract counterpart is based on technology of relational and parallel database systems. The architecture of distributed query processing system based on the presented algebra is described. Finally, some preliminary experimental results are discussed.

We have a list of tasks that remain to be completed. Among the most important are: distributed implementation of mapping from strings (URIs) to integers and its inverse mapping, more deep study of the effects of structure and distribution of query trees to the execution speed, experimental study that will give more insight into interrelations between data and query distribution, and, improving the communication speed among cluster servers by packing triples into bundles.

Fig. 1 .

 1 Fig. 1. (a) Left-deep query tree (b) Left-deep query tree with multiple AM operations

Fig. 2 .

 2 Fig. 2. Configuration of servers for a particular query

Fig. 3 .

 3 Fig. 3. Architecture of big3store query executor

Fig. 4 .

 4 Fig. 4. Benchmark queries

Table 1 .

 1 Benchmark Results (in seconds)

	Query big3store Virtuoso
	Q1	0.015	0.149
	Q2	0.086	0.133
	Q3	0.033	0.159
	Q4	0.009	0.608
	Q5 95.594	0.054
	Q6	3.652	0.262
	Q7	7.549	0.279
	Q8 23.512	0.182
	Q9 104.364	0.558

Acknowledgments

We thank Andreas Schmidt for valuable suggestions on draft version of this paper. This work was supported by the Slovenian Research Agency and the ICT Programme of the EC under PlanetData (ICT-NoE-257641).


Katerina Zamani

Angelos Charalambidis

Stasinos Konstantopoulos

Maria Dagioglou
email: mdagiogl@iit.demokritos.gr

Vangelis Karkaletsis
email: vangelis@iit.demokritos.gr

A Peer-to-Peer Protocol and System Architecture for Privacy-Preserving Statistical Analysis

Keywords: Privacy-Preserving statistical analysis, Secure summation protocol, Statistical processing of health records

The insights gained by the large-scale analysis of healthrelated data can have an enormous impact in public health and medical research, but access to such personal and sensitive data poses serious privacy implications for the data provider and a heavy data security and administrative burden on the data consumer. In this paper we present an architecture that fills the gap between the statistical tools ubiquitously used in medical research on the one hand, and privacy-preserving data mining methods on the other. This architecture foresees the primitive instructions needed to re-implement the elementary statistical methods so that they only access data via a privacy-preserving protocol. The advantage is that more complex analysis and visualisation tools that are built upon these elementary methods can remain unaffected. Furthermore, we introduce RASSP, a secure summation protocol that implements the primitive instructions foreseen by the architecture. An open-source reference implementation of this architecture is provided for the R language. We use these results to argue that the tension between medical research and privacy requirements can be technically alleviated and we outline a research plan towards a system that covers further requirements on computation efficiency and on the trust that the medical researcher can place on the statistical results obtained by it.

Introduction

The insights gained by the large-scale analysis of health-related data can have an enormous impact in public health and medical research, but access to such personal and sensitive data poses serious privacy implications for the data provider and a heavy data security and administrative burden on the data consumer. The discussion on what exactly it means to not disclose private data [START_REF] Clifton | Defining privacy for data mining[END_REF] and the discussion on policies for balancing between scientific advancement and privacy [START_REF] Horvitz | Data, privacy, and the greater good[END_REF] are very relevant, but should be complemented by the equally relevant discussion of whether there is tension at all between data privacy and data-driven research. In other words, it is not straightforward if private data can be insulated from medical research workflows without compromising either.

As anonymization has been repeatedly proven to be inadequate [START_REF] Ohm | Broken promises of privacy: Responding to the surprising failure of anonymization[END_REF], attention has turned to research in cryptography and distributed computation. These fields can provide methods for computing aggregates and statistics without revealing the specific data values involved in the computation, offering a much stronger guarantee of privacy than anonymization. However, from the perspective of the data mining practitioners and the medical researchers there is still a residue of functionality missing between their workflows over anonymized data and what is technically possible to achieve without accessing specific datapoints. Naturally, part of the workflow involves browsing data in order to formulate a hypothesis, and cannot possibly be performed over anything else but experimental data specifically collected and licensed to be shared. The scope of our discussion is, therefore, necessarily restricted to the data and processing required to empirically validate an already formulated hypothesis over a larger dataset than what can reasonably be made available to research.

To make this more concrete, we will assume use cases from ambient assisted living (AAL) environments. AAL covers a wide range of concepts, hardware and software products, and services that facilitate better, healthier and safer life outside formal health-care institutions. These environments emphasise the automatic collection of health data in one's own environment and the secure sharing of such data with medical care providers. In such a system, health data is shared between the following entities:

-The AAL agent that is the data management component of the AAL environment. The AAL agent has unrestricted access to its user's sensitive data.

The management and security of the data held by the AAL agent is primarily within the scope of network security. -The health-care provider that needs access to sensitive data of a small set of individuals on a need-to-know basis, depending on the medical condition that necessitates the monitoring of each individual. The management and security of the data held by the health-care provider is primarily within the scope of network security and access control. -The medical researcher that needs access to aggregate values computed over the sensitive data of a large set of individuals, but does not need to know any specific individual's data. It is the data transfer protocols between this agent and the AAL agents that are within the scope of the work described here.

In the remainder of this paper, we first present the main approaches to privacy-preserving computation and discuss what requirements from our use case are not covered by the state of the art (Section 2). We then proceed to present a system architecture that exposes privacy-preserving computation functionality to tools (such as R) that are commonly used in medical research workflows (Section 3). We then present our peer-to-peer protocol that implements this privacy-preserving computation functionality based on the homomorphic property of composite secret sharing schemes (Section 4). We finally conclude and discuss future research direction (Section 5).

Related Work

We see in the literature three major approaches to privacy-preserving computation: differential privacy, homomorphic encryption, and secure multiparty computation. Differential privacy is based on the property that a result of a statistical value can be approximated even if random noise has been added to the data. Homomorphic encryption supports computations over cipher-texts, so that the result can be obtained without decrypting individual datapoints. Finally, secure multiparty computation is based on communication protocols between the agents to collaboratively compute a function over their private values without revealing the actual values.

Differential privacy preserves privacy by perturbing the datasets with randomized noise, such as symmetric exponential (Laplace) noise or with a use of a Geometric Distribution [START_REF] Shi | Privacy-preserving aggregation of time-series data[END_REF]. When the perturbed datasets are used in statistical analysis, knowledge of the distribution parameters of the noise applied allows approximating the analysis outcomes over the unperturbed data, but does not allow recovering any of the individual datapoints. To name an example, the PINQ data analysis platform [START_REF] Mcsherry | Privacy integrated queries: An extensible platform for privacy-preserving data analysis[END_REF] creates a differential privacy layer between the raw data and data analysis software. PINQ supplies the analyst with a set of transformations in operations like Where, Select, GroupBy and Join, in order to apply them to the data-set before applying operations for differential-privacy aggregations.

What should be noted about differential privacy is that it provides approximations and is only applicable where this is tolerated and where the datasets are large enough to allow for this approximation to be accurate enough for its purpose. In the the analysis for medical data, it is often the case that datasets are not large enough to give tolerable error margins or that outliers can lead to important insights and should be highlighted rather than smoothed out.

The second major strain of privacy-aware computation protocols is based on homomorphic cryptosystems, cryptographic mechanisms with the property that certain operators (such as addition) can be computed directly within the encrypted space without requiring that the individual operands can be decrypted. One of the most prominent homomorphic cryptosystems is Paillier's cryptosystem [START_REF] Paillier | Public-key cryptosystems based on composite degree residuosity classes[END_REF], which allows computing the cipher of the sum of two numbers given the ciphers of these numbers. Paillier's cryptosystem requires that all numbers are encrypted using derivatives of the public part of a master key; these derivatives are such that they cannot decrypt the cipher of other derivative keys, but the master key can decrypt the cipher of the sum. This algorithmic basis can be extended to provide further numerical and categorical operators beyond summation; for example Kissner and Song [START_REF] Kissner | Privacy-preserving set operations[END_REF] proposed an extension that supports union, intersection and element reduction.

Although data providers are perfectly protected from their peers, the main weakness of homomorphic systems is the trust that must be placed on the entity that issues the master key [START_REF] Kerschbaum | Privacy-preserving computation[END_REF]. The typical summation protocol based on Paillier's cryptosystem has a master agent issue a master key and a number of data agents that exchange their encrypted values between them in order to send a total encrypted summation back to the master agent. Privacy from the master agent is only guaranteed by the fact the master agent only receives the cipher of the end-result. If the master agent colludes with one malicious data agent, they can use the private part of the master key to reveal the private value of the victim agent, the data agent that passes its encrypted data to the malicious agent.

To lift the requirement that the master agent must be trusted, Shi et al. [START_REF] Shi | Privacy-preserving aggregation of time-series data[END_REF] proposed a framework that can compute statistics on medical data with the use of an untrusted data aggregator, by encrypting values that can be decrypted with the sum of multiple cipher-texts under different user keys. Shi et al. propose a method where each agent encrypts periodically its data with its respective private key. The data of every agent includes its private value combined with white noise. The untrusted aggregator receives all the encrypted values from the agents and decrypts them with its private key and with the use of a correlation between the private keys of all agents and a specific hash function, that is based on the time series. The algorithm needs an initial trusted setup phase, which does not allow agents to join or leave the system dynamically. The proposed protocol is based on differential privacy and as an implication the resulted statistic is an approximation of the real one, which may cause problems in medical data. Moreover, as authors report, in order for their approach to work efficiently, the plain text space should be small.

There are many studies that combine their secure mechanism with the use of a trusted third party that works as the aggregator. In trusted third party protocols, there is an external trusted party which receives the private data of the agents and computes a function by using them. Hanmanthu et al. [START_REF] Hanmanthu | Third party privacy preserving protocol for perturbation based classification of vertically fragmented data bases[END_REF] propose an enhanced protocol that combines a technique which perturbs distributed data with the use of a third party. Specifically, they define a protocol for constructing a Naive Bayes classifier. In this protocol, each agent encrypts its perturbed data with its private key and sends it to a trusted third party. The trusted third party decrypts this data with the public key of the respective agent and constructs a perturbed Naive Bayes Classifier. Moreover, there are some studies that combine secure multiparty computation (SMC) systems with a trusted third party. Generally speaking, an SMC system deals with the computation of any function with any input in a distributed network, where the involved agents can learn only the total result and their own input. Thus, a common strategy to ensure trustworthiness is the use of a trusted third party. Ajmani et al. [START_REF] Ajmani | A trusted third-party computation service[END_REF] present TEP, a trusted third party computation service that maintains generality. TEP offers flexibility because it fits in many SMC applications to guarantee privacy. However, this type of mechanism requires the existence of a trusted third party, so is inherently weaker than purely peer-to-peer networks.

Nevertheless, Sheikh et al. [START_REF] Sheikh | Privacy preserving k secure sum protocol[END_REF] proposed a SMC system that applies a secure summation protocol without the use of a trusted third party. The proposed protocol focuses on the increased computation complexity to avoid hacking. Each agent splits its data to a fixed number of segments and promotes a single segment to the next agent at each iteration. As an extension Sheikh et al. [START_REF] Sheikh | Privacy preserving k secure sum protocol[END_REF] define a master agent, which sets a random number during the initialization. Despite the fact that this protocol does not utilize a third trusted party, it is weak because if two neighbour agents collude, they can reveal the data value of the middle agent. Moreover, this technique imposes a considerable overhead in the communication between the agents.

Many recent research studies focus on privacy preserving on vertical and on horizontal partition of data. Our approach is oriented to horizontally distributed data, as each AAL agent keeps a private database with its values and each database contains the same set of attributes. Specifically, Karr et al. [START_REF] Karr | Secure regression on distributed databases[END_REF] propose a secure computation of linear regression for horizontally partitioned data without the use of a trusted third party. This is achieved by converting the linear regression equation to a summation form, where the quantities of each summation involve attribute values of the same agency. To protect data from the scope of the source and the values, they propose a SMC secure sum computation protocol. During the initialization of the protocol, a master agent adds his private value with a random number, that he previously produced, and forwards the summed value to the next agent. Each agent receives the aggregated value from the previous agent and forwards it to the neighbor agent, after the addition of his private value. The total summation result is returned back to the master agent, which removes his random number. This protocol is weak mostly because a private value of an agent can be revealed by the collusion of his neighbors. Also due to the circular mode of the algorithm, it can not be parallelised.

The study of Molina et al. [START_REF] Molina | Hiccups: health information collaborative collection using privacy and security[END_REF] is closer to our approach. Specifically, they propose an application of homomorphic encryption to compute basic statistics on aggregated medical data which also guarantee the privacy of the medical data. Their SMC protocol preserves the privacy between the caregivers, where each one computes statistics for their corresponding patients. This is achieved with a double encryption, each one depending on a different public key -the public key of researcher and the public key of a caregiver chosen randomly to work as the aggregator. This approach can be mapped well in distributed systems because each caregiver can work in parallel to compute aggregates of their patients' data. However, privacy is relatively weak as the researcher and the aggregator can collude to reveal the plaintexts of each caregiver. Moreover, doubly homomorphic encryption schemes are not fully explored to define which statistics can be determined.

Privacy-Preserving Statistical Analysis

In this section we introduce our system architecture, and show how elementary statistical analysis methods can be implemented within this architecture in a way that essencially preserves the API of their convential implementation. As a showcase, we assume the R language implementation of the t-test and show how the same interface can be implemented within our privacy-preserving architecture instead of by directly accessing data matrices. As the architecture assumes the existence of a privacy-preseving summation protocol to access the private data, we also discuss what characteristics are required from this protocol.

System Architecture

The system architecture can be perceived as a stack of three layers and each layer depends on the functionality provided from the layer at the lower stage. The upper layer, called the Medical Researcher's interface, accepts from the medical researcher the method with the initial parameters to be executed by the system. The purpose of this interface is to provide a familiar environment to the researcher and therefore in our current implementation this layer is developed in the R language. The initial parameters are transformed appropriately in order to be passed to the next layer, which is the Compilation Layer . At that stage, the high-level parameters and commands of the statistical method are transformed into low-level instruction for accessing the private databases of the agents. An instruction represents an aggregation over a selection of data. Currently, the aggregation operation is summation. However, the aggregations that are on one hand feasible by the system and on the other hand safe for preserving privacy depend on the secure protocol used. These instructions will be eventually evaluated by the lowest layer of the architecture, the Privacy Protocol Layer . Figure 1 depicts the system architecture and the information exchanged between the layers.

The Medical Researcher's interface The interface is developed in the R language since it offers a variety of plotting and analysis tools, while in parallel it is a familiar environment for statisticians. The researchers can execute the statistical method through the R environment by importing the secure statistics library. The purpose of this library is to expose high-level statistic methods (e.g. linear regression, t-test) as R functions.

The secure statistics library receive the same arguments as the conventional statistics functions in R. The only difference is that the data arguments are not matrices of values, but the parameters needed to make a distributed computation. The results of the statistics functions are, then, identical to those of the respective conventional functions over the same data.

The Compilation Layer This layer is responsible for the communication between the two other layers. Specifically, it translates the arguments of the secure statistic to a suitable format, thus it defines the appropriate data that are going to be used for the statistic computation. Moreover, it converts the simple statistic equations to a set of summations; a compatible format to achieve the secure The Aggregation Protocol This layer executes the privacy protocol between the AAL agents, To deal with the concurrent computation of each instruction, we model our agents as actors. Each actor makes the appropriate computations with respect to the given instruction and its private data. These computations can easily be done since every AAL agent controls its corresponding health records. After the computation of the value, which represents the initial secret, the privacy protocol is executed. The protocol may involve all the actors to work collaboratively in order to compute the aggregation of their secrets without revealing the actual secrets to each other or the agent requesting the aggregation. The aggregated result is collected a designated actor. The selection of such actor is irrelevant and can be done randomly. Our proposed implementation for this layer is presented in more detail in Section 4.

Example We will use a simple example to better demonstrate the proposed system. Suppose that a medical researcher needs to run a t-test to assess whether the means of two groups are statistically different from each other, that is to compute t in Eq. 1:

t = X -Ȳ |X| -1 σ 2 X + |Y | -1 σ 2 Y (1)
where X and Y are the datapoints of the two groups, X and Ȳ are their means, |X| and |Y | are cardinalities, and σ 2 X and σ 2 Y their variances. Assume, for instance, that a researcher wants to test the effect of medicine M 1 (Group 1) and medicine M 2 (Group 2) on blood pressure, with the further restriction that participants in Group 2 should be above 65 years old. A workflow using the R language would be:

-Select from a database the instances that match Group 1 criteria and store them in variable X -Select from a database the instances that match Group 2 criteria and store them in variable Y -Decide on the conditions of the T-test, such as the confidence level and alternation, and store them in variable C -Pass X, Y, C as arguments to an implementation of t-test Our architecture allows this workflow to remain essentially unaffected, except for the contents of X and Y . Instead of holding actual data arrays these now contain a representation of the Group 1 and Group 2 criteria, so that the selection can be executed in distributed manner. Using this representation, a privacy-aware implementation of t-test can produce the exact same result as the conventional implementation, except without ever accessing any individual data.

This representation declares a list of dependent variables and a list of eligibility criteria of the sample groups, as a set of (variable, operator, value) tuples.

In our example, we assign to X and Y the criteria that we would have used to assign to them a value array if we had full access to the data:

-X = [("medicine", =, "M 1 ")] -Y = [("medicine", =, "M 2 "), ("age", >, "65")]

The compilation layer converts the t-test implementation into a set of instructions. Recall that each instruction is an aggregation over the private data

Function Definition add(C)

i si(C), where si(C) is the secret value of the i-th AAL agent if condition C is satisfied, 0 otherwise add 2 (C, k) i (si(C) + k) 2 , where k is a constant and si(C) is same as above cnt(C) i ci(C), where ci(C) is 1 if the i-th AAL agent satisfies condition C, 0 otherwise. Table 1. Characteristic instructions provided by the RASSP Protocol.

of each agent, under the given selection restrictions. Table 1 defines the instructions needed to implement the t-test (Eq. 1), which is then implemented using the following pseudo-code:

1. X = [("medicine", =, "M 1 ")];
X is a representation of the secret values of all AAL agents where medicine M 1 is used. 2. Y = [("medicine", =, "M 1 "), ("age", >, "65")];

Y is a representation of the secret values of all AAL agents where medicine M 2 is used and age is above 65. This uses the values above to calculate means. 6. σ 2 X = add 2 (X, -X); σ 2 Y = add 2 (Y, -Ȳ ); This uses the values above to calculate variances.

7. T = X -Ȳ /sqroot σ 2 X /C 1 + σ 2 Y /C 2 ;
Each instruction is executed with the use of the secure summation protocol, obtaining the aggregate values specified in the instruction without obtaining the values themselves. From the perspective of the R interface user, the t-test functions operate as if they had been passed the actual value matrices as parameters.

Reference Implementation

The system architecture that is described in Section 3.1 is implemented by the open source project at https://bitbucket.org/dataengineering/rassp The project's source code is organized in three modules, each one implementing one of the layers in our architecture:

proto implements the aggregation protocol stats is the implementation of statistical analysis primitives over an aggregation protocol, and implements the compilation layer -RStats implements the R interface for the medical researcher over the compilation layer.

To execute the example immediatelly above using our implementation, the medical researcher executes the following code in the R interface:

# Describe the two groups in GroupStat structures: group1 <-GroupStat(list(c("med","=","A"))) group2 <-GroupStat(list(c("med","=","B"), c("age",">","65"))) # Set dependent variables and groups in a Parameters structure: p <-Parameters(list("bloodPr"), list(group1, group2)) # Execute the normal t-test using the Parameters structure p: ttest(p, varEq=TRUE)

What is important to note in the example is that our implementation of the ttest() function presents an interface identical to the standard R implementation of the t-test. The underlying difference is that the Parameters structure does not point to actual data matrices but to instances of our GroupStat structure, which hold the information needed by the compilation layer in order to distibute the computation to the participating nodes.

Discussion

The proposed system architecture assumes that:

-The statistical analysis that is to be carried out can be implemented using the set of aggregation instructions provided by the aggregation protocol. In other words the algorithm should not depend on individual data points. -A summation protocol exists that guarantees privacy.

The first assumption holds, since the most commonly used class of data mining algorithms can be expressed as an iteration of summation expressions [START_REF] Kearns | Efficient noise-tolerant learning from statistical queries[END_REF]. If needed, categorical operators can be implemented based on summation [START_REF] Kissner | Privacy-preserving set operations[END_REF].

Regarding the second assumption, we will now proceed to discuss the summation protocols that can be used in our architecture and, in Section 4, present the protocol we use in our reference implementation of the architecture.

Most of the related studies guarantee their privacy by utilising encryption or differential privacy techniques. These approaches do not fit in our problem, because we deal with medical history data that are distributed across AAL agents. In homomorphic techniques, a master agent shares a public key with the rest of the agents, in order to encrypt their data, and keeps a private key for the final decryption. Such a mechanism is privately weak in the case of collaborative computations, because if the medical researcher (master agent) and one AAL agent collude, they can learn another AAL agent's private value. This makes the technical protocol weak, as it places a heavy burden on non-technical policies and protocols to guarantee the integrity of the medical researcher. Since our main aim is to alleviate the need for non-technical policies and protocols and to make it easier for medical researchers to run statistics over datapoint they are not meant to access directly, homomorphism encryption does not cover our requirements.

In addition, differential privacy is also not applicable, from both the perspective of the medical researcher as well as from that of the AAL agent. From the perspective of the medical researcher, differential privacy computes approximations, which can be a problem as discussed in Section 2 above. From the perspective of the AAL agent, the secret value can be approximated by its repeated querying, since a different perturbation of the real secret needs to be computed for each query. The AAL agent cannot produce a single perturbed value and use this for all queries, since it needs to be re-computed to follow the distribution parameters requested by the medical researcher. This might be less of a problem in time-series data (such as power grid data or traffic data), but can result in substantial information leaking in static historical data, such as health records. [START_REF] Clifton | Defining privacy for data mining[END_REF] The Secure Summation Protocol

Background

Secret sharing schemes divide a secret into many shares which can be distributed to n mutually suspicious agents. The initial secret can be revealed if any k of these n agents combine their shares. We will call such schemes, (k, n)-threshold schemes. If such a scheme also possesses the homomorphism property, then multiple secrets can be combined by direct computation only on the shares. Such schemes are usually called composite secret sharing schemes [START_REF] Benaloh | Secret sharing homomorphisms: Keeping shares of a secret secret[END_REF].

More specifically, assume n mutually suspicious agents and each agent holds a secret s i . The desired computation is combination into a super-secret s under an operation ⊕, namely s = s 1 ⊕ • • • ⊕ s n . Using a secret sharing scheme each s i can be split into k shares d i1 , . . . , d i k such that given a known function F I it is the case that:

s i = F I (d i1 , . . . d i k )
We will say the (k, n) threshold scheme has the (⊕, ⊗)-homomorphism property if whenever s = F I (d 1 , . . . , d k ) and s = F I (d 1 , . . . d k ) then

s ⊕ s = F I (d 1 ⊗ d 1 , . . . , d k ⊗ d k )
The composition of the shares d 1 , d 1 yield a super-share d 1 ⊗ d 1 . In other words, the (⊕, ⊗)-homomorphism property implies that the composition of the shares under the operator ⊗ are shares of the composition under the operator ⊕.

Overall, the advantage of having a composite secret sharing scheme is that secret cannot be obtained, only if k or more agents collude and combine their sub-shares. In addition, this protocol is suitable to our approach from the AAL agent's point of view, because it does not use a trusted third party or depends on cryptographic assumptions, while at the same time it is k-secure. This approach represents a secure summation protocol that can easily be applied to collaborative agent systems.

Based on this mathematical foundation, we will now proceed to present the RASSP protocol, a (+, +)-homophorphic composite secret sharing scheme.

The RASSP Protocol

Assume that we have n AAL agents, where each one has its private value v i , i ∈ [1..n]. Each AAL produces random breakdown of v i into n terms r ij , j = 1..n such that v i = n j=1 r ij . These terms are computed by first producing n -1 random terms r ij , j = 1..i -1, i + 1..n and then setting

r ii = v i - j∈[1..n]-{i} r ij
The r ij terms are called sub-shares and are (except for r ii ) shared with the rest of the AAL agents, one per agent. In this manner, each AAL agent shares n -1 values and receives n-1 values from the rest of the AAL agents. The super-share Y i for each agent is defined as:

Y i = r ii + k∈[1..n]-{i} r ki (2)
Notice how the super-share of AAL agent i is the sum of the sub-share that it has not shared and of all the sub-shares that it has received from the other AAL agents. Finally, we define a function F I as the sum of the super-shares:

F I (Y 1 , . . . , Y n ) = n i=1 Y i (3)
It is straightforward to verify that F I (Y 1 , . . . , Y n ) is equal to the sum of all secrets. It is also straightforward to verify that only random numbers and obscured data values are shared between AAL agents and between AAL agents the researcher. Notice also that only if (n -1) AAL agents collude to merge their sub-shares can the private value of the n-th agent be revealed. Therefore, our system guarantees (n -1)-security. Figure 2 gives an example of the above description for a system of three AAL agents. In this example House1 has the private value v 1 and produces three numbers: r 11 , r 12 , r 13 . Then, it shares r 12 and r 13 with House2 and House3, keeping r 11 hidden. House1 receives two numbers (r 21 , r 31 ) from the other AAL agents. In then shares the computed Y 1 , so that F I can be computed by summing all Y i . F I (Y 1 , Y 2 , Y 3 ) computes the sum of all three AAL agents' secret values.

The described secure summation protocol is suitable for computing medical statistics and preserve privacy at the same time. The only constraint is that House1 v1 -------------------------r11 , r12, r13 House2 v2 -------------------------r21 , r22, r23 House3 v3 ------------------------ the resulted outcome is a sum of the private values, thus the statistic equations should be converted in a summation form. The summation form results in accurate values and not approximations, while simultaneously it can easily be parallelised [START_REF] Chu | Map-reduce for machine learning on multicore[END_REF]. Besides, medical researchers typically use descriptive statistics which utilise numerical descriptors such as mean and standard deviation. These descriptors can easily be converted into a summation form, thus they can be computed by our system.

Conclusions and Further Work

With this paper we experiment with privacy-preserving data mining that is accessed not through specialized APIs and tools, but through statistical analysis tools that are ubiquitous in data-driven research, such as the R language and its statistical analysis libraries. In this manner, we are targeting the uptake of our privacy-preservation infrastructure by the medical research community, as the discussion around privacy preservation is mute if the data cannot be efficiently and effectively used to achieve the medical research purpose. Specifically, our first contribution is our architecture and its reference implementation. This architecture foresees the primitive instructions needed to re-implement the elementary statistical methods so that they only access data via a privacy-preserving protocol. The overall gist is that function arguments remain the same, except for substituting data matrices with a specification of how to select the data that each AAL agent will contribute to the distributed computation. The advantage is that more complex analysis and visualisation tools that are built upon these elementary methods can remain unaffected by replacing direct access to data with access via privacy-preserving protocols.

A further contribution is our review of secure multiparty computation, differential privacy, and homomorphic encryption approaches to justify and explain assuming the former as the most appropriate basis for our personal health record use case. Finally, we introduce RASSP, a secure summation protocol that computes sums within Benaloh's composite secret sharing framework for secure multiparty computation.

More complex instructions will be implemented as iterations of the primitive sum operator. As this is bound to reduce the run-time efficiency of the system, our next steps will be to integrate distributed computation concepts in order to parallelise the computation. Chu et al. [START_REF] Chu | Map-reduce for machine learning on multicore[END_REF] propose using the map-reduce framework to execute a variety of statistics, where the summation form of their equations facilitates the distribution of their calculations. This approach will be mapped to our architecture to improve the run-time efficiency of the system. A further optimization step will be to execute simultaneously instructions when there is no dependence between them. To achieve this, we will transfer from the programming languages and distributed computation literature optimization methods that can decide about the most efficient execution plan for a given program with multiple calls to the primitive instructions. We will also need to extend the current API of these instructions in order to allow multiple requests to be made to the distributed AAL agents with one network transaction.

Another relevent on-going discussion in the community is the involvement of humans in the data mining process itself [START_REF] Holzinger | Interactive machine learning for health informatics: When do we need the human-in-the-loop?[END_REF] and when acting upon data mining results [START_REF] Kieseberg | A tamperproof audit and control system for the doctor in the loop[END_REF]. Given the responsibility of the medical practitioner when using data to make medical decisions, the uptake of our-and, in fact, any-privacy preserving protocol depends on the data consumers' ability to apply checks and bounds to the values that are allowed to participate in the computation. In this context, a more ambitious goal is to extend the RASSP protocol so that the medical researcher can specify what value ranges of the secret variables are reasonable or useful and to have this range restriction guaranteed without having to trust the AAL agents. To achieve this, we will experiment with variations of the current RASSP protocol where the AAL agents can to some extend also check whether their peers are (erroneously or maliciously) sharing sub-shares that add up to out-of-range values. Since such AAL agent behaviour can corrupt the result or obscure useful outliers, it undermines the trust that the researcher places on the result. Our goal is to devise a system of cross-checks that makes it unlikely that out-of-range values contribute to the sum undetected, but without compromising the privacy-preserving nature of the protocol.

Fig. 1 .

 1 Fig. 1. The system's architecture

3 .

 3 N 1 = add(X); N 2 = add(Y ); N 1 is the sum of the secret values X and N 2 is the sum of the secret values Y . 4. C 1 = cnt(X); C 2 = cnt(Y ); C 1 is the number of AAL agents with non-zero values in X and C 2 is the number of AAL agents with non-zero values in Y . 5. X = N 1 /C 1 ; Ȳ = N 2 /C 2 ;

Fig. 2 .

 2 Fig. 2. The RASSP secure summation protocol.

Acknowledgements

The work described here was carried out in the context of the RADIO project. This project has received funding from the European Union's Horizon 2020 research and innovation programme under grant agreement No 643892. For more details, please visit the RADIO Web site http://www.radio-project.eu