Datalog in Time and Space, Synchronously

Abstract

Motivated by recent developments of Datalog-based languages for highly distributed systems [9], in this paper we introduce a version of Datalog: speci cally tailored for distributed programming [2] in synchronous settings, along with its operational and declarative semantics.

Introduction

Nowadays we are living the \end of the Moore's law as we know it ": increasing performance cannot be achieved just by increasing hardware speed; instead, concurrent and parallel computation must be exploited. Thus, in the last few years new paradigms such as cloud computing, frameworks like MapReduce [ 7 ], and systems like NoSQL databases, are put at work to help programmers in developing scalable applications which exploit distributed systems. We are however aware that limitations still exist in the current literature concerning the formal models needed by the above discussed new trends. Our aim is then to develop new techniques that can be employed to solve the challenges of today's highly distributed systems.

We think that Datalog is in a very favorable position to provide an important contribution to the development of the aforementioned technologies: (i) its intrinsically (embarrassingly) parallel nature provides a rm basis to develop more complex languages for modern parallel and distributed applications [ 6, 5 ]; (ii) its solid logical foundation provides the theoretical background that permits to formally specify and analyze complex distributed systems [ 9, 3 ].

In particular, the role of our present work is to de ne what a synchronous distributed system is and to provide the semantics for such kind of systems for a version of Datalog speci cally tailored for distributed programming [ 2 ]. Albeit asynchronous systems are highly preferable today to synchronous systems, the latter are still very convenient because programs do not have to deal with much uncertainty, thus, once designed for such an ideal model, they can be re ned to work in more realistic settings. Our work, in particular, proceeds of our previous work [ 11 ], and is motivated by the following two use-cases: (i) active deductive databases [ 13 ] have been developed in centralized settings in which updates must be issued to a unique central database: our aim is to enhance this model, developing the semantics of synchronous networks of partitioned and/or replicated databases; (ii) in the Massive Parallel (MP) model introduced in [ 10 ], computation proceeds by steps that are performed in parallel by clusters of machines on which the same program is running. This model actually is a particular instantiation of a synchronous system. In addition, the MP model is strictly related to the MapReduce model (MR)[ 10 ]. As a consequence, by developing a semantics for synchronous systems we would be able to seamlessly embed in our framework both the MP and the MR computation models.

The paper is organized as follows: in Section 2 we introduce some preliminaries on DatalogS:T , i.e., Datalog: augmented with a notion of time and space. In Section 3 we describe what a synchronous distributed systems is, and give an idea of the operational and declarative semantics of DatalogS:T in such timing model. 1 2

Preliminaries

In the following we de ne a distributed system to be a non-empty nite set of fully and reliably connected node identi ers L = f1; : : : ; ng. A DatalogT:S program is composed by a set of rules, each having one of the following two forms: H(w; t) H(w; t0) (1) (2) time(t): time(t); succ(t; t0): where H, Bi and Sj , with 0 i n, 0 j m, are taken from the set relname of relation (names), and (w), (ui), (vj ) are tuples whose components (terms) can either be variables or constants. In addition, we have two \special" terms: a temporal term t (t0) called time-step identi er and ranging over the set of natural numbers N0, and a spatial term lj called location speci er and ranging over the set of node identi ers L. We assume that the time-step identi er term cannot appear in any of the vectors w, ui, vj , while we will see that the location speci er is allowed to appear in w. As usual, the left-hand part of a rule (H(w)) is referred to as the head, and the right-hand part as the body. The head of each rule is an atom, i.e. an occurrence of a predicate, while the Bi(ui; t) Sj (vj ; t) are literals, i.e., positive or negated atoms. If m = 0 the rule is local, distributed otherwise. Moreover, if n = m = 0 the rule is expressing a fact. We assume each rule to be safe, i.e., every variable occurring in a rule-head or in negative literal, must appear in at least one positive literal of the rule body. As a consequence, we assume all facts to be ground, i.e., containing only constants terms.

A (temporal-spatial ) database schema RT S is de ned as the set of relation names R 2 relname plus two pairs of built-in relations: succ (of arity two), and time (of arity one) ranging over time-step identi ers, and id and all, both of arity one and ranging over the set of location speci ers. The interpretation of succ(t; t0) is t0 = t + 1, while time(t) is used to bind the time-step of a tuple to a precise time t. As for the interpretation of id and all, an occurrence of the rst is in each node and stores only that node id, while the second contains all the nodes 1 The formal basis of this work, along with many examples, is developed in [ 12 ]. belonging to the network, i.e., L. The intensional part of the database schema, namely idb(RT S ; ), is the subset of the database schema RT S containing all relations that appear in at least one rule-head in , while with edb(RT S ; ) we denote the extensional database. In addition, with sdb(RT S ; ) { disjoined from both edb and idb { we refer to the set of distributed shared relations (DSR), which contains the relations represented as Sj in eq.s (1, 2) above, and are the only ones allowed to contain location speci er terms. As the name highlights, DSR's are shared among multiples nodes simultaneously, and are the means by which nodes are able to interact among themselves; therefore a mechanism for deciding which node is responsible for which tuple must be developed. The location speci er term exactly serves at this scope: it is used to identify which node is responsible for which portion of data. From the above de nitions and eq.s (1, 2) follows that Bi 2 edb(RT S ; ) [ idb(RT S ; ), Sj 2 sdb(RT S ; ) and nally H 2 idb(RT S ; ) [ sdb(RT S ; ), with the meaning that H can specify an intensional relation or a distributed relation. In the latter case, we assume w to contain also the location speci er term. In particular, in case of rules of the type of (1) the location speci er term in w must be the local node id, while for rules of type (2) can be any node identi er in L, so that, with inductive rules, we are able to perform communication among nodes. Then, given a time-step value t 2 N0, a (temporal-spatial ) database instance is a nite set of facts I[t] over the relations of RT S . Similarly, a (temporal-spatial ) relation instance IR I[t] is a set of facts de ned over R, with R 2 RT S . In the following we will refer to Init as the nite initial database instance of the schema.

In DatalogT:S , we consider tuples to be immutable { once instantiated they cannot be retracted nor changed { and ephemeral by default, i.e., a tuple is valid only for its assigned time-step. Hence, we use rules in the form of (1) to derive new facts given the information currently available at that point in time. Adopting the same notation of [ 2 ] we call this type of rules deductive. Nevertheless, in DatalogT:S mutable tuples can be emulated using time and explicitly stating how tuples evolve with the progress of time. In fact, if a tuple, let's say R(a; b; t) is valid at time t, by employing inductive rules [ 2 ] in the form of eq. (2), we specify a new immutable version which will be valid at time-step t + 1. If de ned by means of inductive rules, tuples from ephemeral become persistent : once derived, for example at time-step t, they will eventually last for every t0 t.

As example, we use the program depicted in Listing 1.1 where we employ an edb relation link containing tuples in the form (x,y,t) to specify the existence of a directed link between a source node x and a destination node y at time t. In addition, we employ a distributed shared relation path to de ne the transitive closure of the link relation. r1: r2: r3: r4: Computation is performed simultaneously on multiple distributed nodes. Communication among nodes is achieved through rules r4 which speci es for example that node a knows that a path from node a to a node c exists if it knows that there is a path from a to another node b and this last node knows that a path to c exists. Rule r4 contains relations stored at di erent locations, therefore implicitly assuming tuples to be exchanged among nodes. 3

Semantics for Synchronous Systems

The operational semantics of Datalog for distributed systems is classically based on the notion of relational transducer and network of relational transducers [ 3 ]. We follow the same approach. Given a DatalogT:S program de ned over a relational schema RT S , we de ne the database, memory, and dsr schemas, respectively, as Rdb = edb(RT S ; ), Rmem = idb(RT S ; ), and Rdsr = sdb(RT S ; ). A transducer schema R is a tuple (Rdb, Rmem, Rdsr, Rtime, Rsys) where Rtime contains just the unary relation time, and the system schema Rsys contains the two unary relations id, and all. A transducer state for R is a database instance T over (Rdb, Rmem, Rdsr, Rtime), and a DatalogT:S -relational transducer is a program T de ned over R. Finally, a transducer con guration is a tuple (L; ) where L is the set of nodes de ning the relation all, and is the location identi er of the node.

Initially a relational transducer T is assumed as loaded with the initial instance Tdb = Init, time contains the tuple (0), and all the other relations are empty. In addition, we assume to have a (possibly in nite) input tape containing an ordered sequence of consecutive natural numbers starting from 0. This sequence will be used as input for a relational transducer in order to provide the clock driving the computation. Why we provide time-steps in this way will be more clear in Section 3.3, where we de ne synchronous transducer networks. Then, given as input a set of input tuples Tin de ned over Rdsr, together with a con guration and the next time value taken from the input tape, the relational transducer will transit to the next state and output a set of output tuples To0ut. Thus, a local transition is denoted as T, Tin t;=(L);i) T0, To0ut, where t 2 N is taken from the input tape, Tin, To0ut are instances of Rdsr, and (i; L) is a con guration. Now, if we denote with Mded the perfect model of the deductive rules in T computed by employing the so called temporal strati ed semantics [ 2, 12 ] to control negation, Mind the model derived from Mded by ring all the inductive rules, and analogously Mcom is derived from Mded by ring all the communication rules { i.e., inductive rules of the type of eq. (2) where the head relation is a DSR { then T0, To0ut are transducer states such that:

Distributed Computation

At any point in time each node is in some particular local state incapsulating all the information of interest the node possesses. For convenience, we de ne the local state si of a node i 2 L as the pair (Ii; n) where Ii = Ti[n], being Ti[n] a transducer state for node i at time-step n. We de ne the global state g of a distributed system as a tuple (se; s1; :::; sn) where si is node i's state and se is the environment local state. We can consider the environment as a \special" node storing all the information external to all nodes. We de ne how global states may change over time through the notion of run, which binds time values to global states. If we assume the set of time values to be isomorphic to the natural numbers we can de ne the function : N ! G where G = fSe S1 ::: Sng, Si is the set of possible local states for node i 2 L , and Se is the set of possible states for the environment. If (t) = (se; s1; :::; sn) is the global state at time t, we de ne e(t) = se and i(t) = si, for i 2 L. We want to note here that the time t and the notion of time-step n incapsulated in programs are two di erent entities. In fact, while the rst one is an external time used to reason about the evolution of global states, the second is de nitely an internal (relative) perspective that each node has about the passing of time. A system may have many possible runs, indicating all the possible ways the global state can evolve. In order to capture this, we de ne a system S as a set of runs. 3.2

Synchronous Systems

For a system to be synchronous it must satisfy the following properties: S1 a global clock is de ned and is accessible by every node S2 the relative di erence between the time-step of any two nodes is bounded S3 updates to remote sdb relations arrive at destination at most after a certain bounded physical time delay [ 8 ] A synchronous system Ssync is therefore a set of runs ful lling the above conditions. The rst property can be expressed in our framework by linking the time-step identi er of each node with the external time, i.e., by assuming the time tape to be globally accessible by every node. Thus, by de nition, every local state i(t) = (Ii; n) will have now t = n. In this con guration, the second property is implemented by assuming that programs proceed in rounds, and that each round, operationally speaking, lasts enough to permit each node computation to reach the x-point. In order to express the third property, we assume to be amply lower than the amount of (physical) time spent between the end of one round and the start of the consecutive one. 3.3

Synchronous Transducer Networks

We have already de ned how local states evolve by de ning what a relational transducer is. Now that we have de ned synchronous systems, it remains to specify how properties S1 - S3 can be enforced during the evolution of global states. To accomplish this, we employ a synchronous transducer network [ 3 ]: a synchronous DatalogT:S -transducer network is a tuple (L; ; T ; T e) where is a function mapping each element i 2 L to a transducer state Ti. We assume the environment to be a \special" relational transducer. In particular, we consider the transducer de ning the environment T e as having an empty program, and the schema composed only by dsr relations with Redsr = Si2L Ridsr. Hence, we consider the environment as registering the sdb facts oating in the network and not received yet. As can be noticed, in our de nition we assume each node i 2 L to have the same transducer T , while the only thing that we allow to change from node to node is its instance. Then, a state N of a transducer network (L; ; T ; T e) is a tuple (Te; T1; : : : ; Tn) where for each i 2 L the i-th element is the related relational transducer state (i) = Ti such that Tdb = Initi.

Let a transducer network initial state be a state where each time relation contains the value 0, and except Rdb all the remaining relations are empty. A global transition is denoted by N =)t N0, where N and N0 are transducer network states, and t 2 N is denoting the input from the time tape specifying what will be the next round. Then, a global transition is de ned such that: { Tin = Te[t] { 8i 2 L, 9Ti s. t. (i) = Ti, (Ti, Tiin, t;=(L);i) T0i, To0iut) is a local transition for node i, where Tiin denotes the set of facts in Tin for node i 2 L { T0e = Si2L To0iut Informally, during a global transition all the transducers composing the network instantaneously make a local transition taking as input the associated sdb -tuples' output of the t 1 round. In addition we assume that one global transition, in order to satisfy property S3, can start only after that a certain amount of physical time has elapsed after the end of the previous transition. If we consider a global state at round t to be de ned as (t) = (se; s1; : : : ; sn) with si = (Ii; t) and Ii = Ti[t] where Ti[t] denotes the local DatalogT:S -transducer state for node i 2 L [ e at time-step t, the de nition of global transition speci es how global states evolve in synchronous settings because it satis es conditions S1 - S3. Given a DatalogT:S program and an initial instance Init, its operational semantics in synchronous settings is completely determined by the synchronous system Ssync de ning its evolution. 3.4

Model-Theoretic Semantics

Starting from a synchronous transducer network (L; ; T ; T e), we can develop a restricted form of DatalogT:S -relational transducer, namely DatalogT: -relational transducer, which is able to simulate the behavior of such a network in centralized settings. Thus, given a transducer network (L; ; T ; T e) with schema R = (Rdb, Rmem, Rdsr, Rtime, Rsys), its centralized version is a DatalogT: relational transducer de ned as follows (where we use the apex c to denote the centralized variant): Rc = (Rdb, Rcmem, Rtime) where Rcmem = Rmem[ Rydsr { where with Rydsr we denote Rdsr restricted over the temporal database { and T c is a DatalogT: -relational transducer, i.e., the restricted version of T ranging over Rc. Informally, the centralized version of a synchronous transducer network is composed by the same program T de ning the network but where each sdb relation is now an idb relations. Due to this restriction, the environment and the system relations are no more necessary. We are now able to derive the model-theoretic semantics of synchronous transducer networks.

Theorem 1 Let Init be a nite initial database instance, (L; ; T ; T e) a transducer network, and Mc the perfect model of the centralized version of (L; ; T ; T e). Then:

Mc = ( [ i(0); : : : ; [

i(n)) i2L i2L (3) 4 Pushed by recent trends of highly distributed systems, we tried to give an intuition of the semantics of a distributed version of Datalog: in synchronous settings. We introduced a new type of Relational Transducer [ 1 ] and Transducer network [ 3 ] tailored for synchronous distributed systems, and sketched the model-theoretic semantics of distributed Datalog programs for such systems.

References

1. S. Abiteboul , V. Vianu , B. Fordham , and Y. Yesha . Relational transducers for electronic commerce . In Proceedings of the seventeenth symposium on Principles of database systems , PODS' 98 , 179 { 187 . 1998 .

2. P. Alvaro , W. R. Marczak , N. Conway , J. M. Hellerstein , et al. Dedalus: Datalog in time and space . In de Moor et al. Datalog Reloaded , 262 { 281 . 2010

3. T. J. Ameloot , F. Neven , and J. Van den Bussche. Relational transducers for declarative networking . In Proceedings of the thirtieth symposium on Principles of database systems , PODS' 11 , 283 { 292 . 2011 .

4. P. Barcelo and R. Pichler, editors. Datalog in Academia and Industry - Second International Workshop, Datalog 2.0, Vienna, Austria, September 11-13 , 2012 . Proceedings, volume 7494 of Lecture Notes in Computer Science. Springer, 2012 .

5. Bloom Language http://boom.cs.berkeley.edu/index.html.

6. Cascalog Libray https://github.com/nathanmarz/cascalog.

7. J. Dean and S. Ghemawat . Mapreduce: simpli ed data processing on large clusters. Commun. ACM , 51 ( 1 ): 107 { 113 , Jan . 2008 .

8. Fagin , R. , Halpern , J.Y. , Moses , Y. , Vardi , M.Y.: Reasoning About Knowledge . MIT Press, Cambridge, MA, USA. 2003 .

9. J. M. Hellerstein . The declarative imperative: experiences and conjectures in distributed logic . SIGMOD Rec ., 39 :5{ 19 , September 2010 .

10. P. Koutris , D. Suciu . Parallel evaluation of conjunctive queries . In Proceedings of the thirtieth symposium on Principles of database systems , PODS'11 . 2011 .

11. M. Interlandi . Reasoning about knowledge in distributed systems using datalog . In Barcelo and Pichler [4] , pages 99 { 110 .

12. M. Interlandi , L. Tanca , S. Bergamaschi: Datalog in Time and Space . Synchoronusly. http://www.dbgroup.unimo.it/TechnicalReport/interlandi2013. pdf, Febbraury 2013 . Technical Report.

13. C. Zaniolo . Advanced database systems . Morgan Kaufmann series in data management systems. Morgan Kaufmann Publishers. 1997 .