Sarah Chlyah

On the Optimization of Iterative Programming with Distributed Data Collections

Keywords: CCS Concepts:, Theory of computation → Algebraic semantics, • Computing methodologies → Distributed computing methodologies, • Software and its engineering → Recursion fixpoint operator, distributed data, rewrite rules, optimization

Big data programming frameworks are becoming increasingly important for the development of applications for which performance and scalability are critical. In those complex frameworks, optimizing code by hand is hard and time-consuming, making automated optimization particularly necessary. In order to automate optimization, a prerequisite is to find suitable abstractions to represent programs; for instance, algebras based on monads or monoids to represent distributed data collections. Currently, however, such algebras do not represent recursive programs in a way which allows for analyzing or rewriting them. In this paper, we extend a monoid algebra with a fixpoint operator for representing recursion as a first class citizen and show how it enables new optimizations.

Experiments with the Spark platform illustrate performance gains brought by these systematic optimizations.

INTRODUCTION

With the proliferation of large scale datasets of various data structures (such as graphs, collections, documents, trees, etc.) and in various domains (such as knowledge representation, social networks, transportation, biology, etc.), the need for efficiently extracting information from these datasets becomes increasingly important. This requires the development of methods for effectively distributing both data and computations so as to enable scalability and improve perfomance.

Efforts to address these challenges over the past few years have led to various systems such as MapReduce [START_REF] Dean | MapReduce: Simplified Data Processing on Large Clusters[END_REF],

Dryad [START_REF] Isard | Dryad: Distributed Data-Parallel Programs from Sequential Building Blocks[END_REF], Spark [START_REF] Zaharia | Apache Spark: a unified engine for big data processing[END_REF], Flink [START_REF] Carbone | Apache Flink™: Stream and Batch Processing in a Single Engine[END_REF]. While these systems can handle large amounts of data and allow users to write a broad range of applications, writing efficient applications is nevertheless not trivial. Let us consider for instance the problem of finding the shortest paths in a large scale graph. We could write the Spark/Scala program in Fig 1 to solve it. The shortestPaths() function takes as input a graph R of weighted edges (src, dst, weight) and returns the shortest paths between each pair of nodes in the graph. The loop (in lines 6 to 14 of Fig 1) computes all the paths in the graph and their lengths; to get new paths, edges from the graph get appended to the paths found in the previous iteration using the join operation. Then reduceByKey operation is used to keep the shortest paths. Spark performs the join and distinct operations by transferring the datasets (arguments of the operations) across the workers so as to ensure that records having the same key are in the same partition for join, and that no record is repeated across the cluster for distinct. Hence, for optimizing such programs, the programmer needs to take this data exchange into account as well as other factors like the amount of data processed by each worker and its memory capacity, the network overhead incurred by shuffles, etc. One optimization that can be done to reduce data exchange in this program is to assign each worker a part of the graph and make it compute the paths in the graph that start from its own part. This optimization leads to the following program (Fig 2 .) which is not straightforward to write, less readable, and requires .join(R.map({ case (z,t,l2) => (z,(t,l2)) }))

.map({case (_,((x,l1),(t,l2))) => (x,t,l1+l2) })

.subtract(ret) ret = ret.union(X).distinct() new_cnt = ret.count() } while (new_cnt > cnt) ret.map({case (x,y,l) => ((x,y),l)}).reduceByKey(min) the programmer to give his own local version of dataset operators (such as join) that are going to be used to perform the local computations on each worker.

Another possible optimization is to put the reduceByKey operation inside the loop to keep only the shortest paths at each iteration because each subpath of a shortest path is necessarily a shortest path. More generally, finding such program rewritings can be hard. First, it requires guessing which program parts affect performance the most and could potentially be rewritten more efficiently. Second, assessing that the rewriting performs better can hardly be determined without experiments. During such experiments, the programmer might rewrite the program possibly several times, because he has limited clues of which combination of rewritings actually improves performance.

One approach to this problem is to offer the user a Domain Specific Language (DSL) to query the data. A DSL is a high level language that is specialized in a particular application domain, and that can be called from within a general purpose language. Queries in this DSL would be translated to an intermediate representation (e.g. an algebra) so that they can be optimized automatically. The idea is to relieve users from having to worry about optimization in the distributed setting, so that they can focus only on formulating domain-specific queries in a declarative manner. A notoriously successful example of this approach is the SQL language and its associated Relational Algebra. This success is due to the level of abstraction provided by the declarative syntax of SQL as well as the extensively studied optimizations provided by Relational Algebra. In RA, data is modelled as relations made of rows and columns. This means that in order to express complex computations on more complex data like nested collections, using a formalism based on the relational model requires flattening the data and using ad-hoc solutions for supporting user defined functions (UDFs). This means that: (1) At the language level, we could have a query language expressed on a flat data model which causes impedance mismatch issues. It is the term that is used to refer to the problems that arise when the data model of the high level language is different from that of the general purpose host language. Specifically, more complex user defined data (data defined by the user in the general purpose programming language) has to be flattened to match the tabular data model of the DSL. In addition, the DSL provides limited support for complex data processing (data transformation, iteration, aggregation, etc.). In order to perform custom transformations on data, one could use language extensions like PL/SQL which, in addition to exacerbating the impedance mismatch problem, requires user expertise and provides only limited optimizations. Alternatively, the user could perform data transformations on the query results in the programming language, which increases roundtrips between the program and the database and does not allow for holistic program optimization. ( 2) At the algebraic level, a number of additional joins are introduced to go from hierarchical to flat types and vice versa which has an impact on performance. Additionally, arguments to second order operations are treated as black box functions which means that they cannot be analyzed and transformed to make automatic optimizations.

It is then important to investigate intermediate representations for expressing and optimizing queries that manipulate data in their native format. As argued by Meijer in [START_REF] Meijer | A Co-relational Model of Data for Large Shared Data Banks[END_REF], establishing and standardizing a formal background for the noSQL market, which now contains multiple separate systems and solutions, is necessary for its economic growth as it was the case for the SQL market thanks to the introduction of RA. The author considers that an algebra based on monads is a suitable formalism for this purpose. Studying intermediate representations that allow for expressing operations on data in their native format would also pave the way for optimizing subsets of general purpose languages and embedded DSLs that do not suffer from impedance mismatch problems. In the context of big data applications, considered algebras must be able to capture distributed programs on big data platforms and provide the appropriate primitives to allow for their optimization. One example of optimizations is to push computations as close as possible to where data reside. When programming with big data frameworks, data is usually split into partitions and both data partitions and computations are distributed to several machines. These partitions are processed in parallel and intermediate results coming from different machines are combined, so that a unique final result is obtained, regardless of how data was split initially. This imposes a few constraints on computations that combine intermediate results.

Typically, functions used as aggregators must be associative. For this reason, we consider that the monoid algebra is a suitable algebraic foundation for taking this constraint into account at its core. It provides operations that are monoid homomorphisms, which means that they can be broken down to the application of an associative operator. This associativity implies that parts of the computation can actually be performed in parallel and combined to get the final result.

A significant class of big data programs are iterative or recursive in nature (PageRank, k-means, shortest-path, reachability, etc.). Recursion is also a very important feature for graph querying as it enables to navigate through the graph and express traversal queries such as paths of arbitrary length [START_REF] Jachiet | On the Optimization of Recursive Relational Queries: Application to Graph Queries[END_REF][START_REF] Libkin | Regular Path Queries on Graphs with Data[END_REF][START_REF] Reutter | Regular Queries on Graph Databases[END_REF]. Iterations and recursions can be implemented with loops. Depending on the nature of the computations performed inside a loop, the loop might be evaluated in a distributed manner or not. Furthermore, certain loops that can be distributed might be evaluated in several ways (global loop on the driver1 , parallel loops on the workers, or a nested combination of the latter). The way loops are evaluated in a distributed setting often has a great impact on the overall program execution cost. Obviously, the task of identifying which loops of an entire program can be reorganized into more efficient distributed variants is challenging. This often constitutes a major obstacle for automatic program optimization. In the algebraic formalism, having a recursion operator makes it possible to express recursion while abstracting away from how it is executed.

The goal of this work is to introduce a gain in automation of distributed program transformation towards more efficient variants. We focus especially on recursive programs (that compute a fixpoint). For this purpose, we propose an algebra capable of capturing the basic operations of distributed computations that occur in big data frameworks, and that makes it possible to express rewriting rules that rearrange the basic operations so as to optimize the program.

We build on the monoid algebra introduced in [START_REF] Fegaras | An algebra for distributed Big Data analytics[END_REF][START_REF] Fegaras | Compile-Time Code Generation for Embedded Data-Intensive Query Languages[END_REF] that we extend with an operator for expressing recursion. This monoid algebra is able to model a subset of a programming language L (for instance Scala), that expresses computations on distributed platforms (for instance Spark).

Contributions. Our contributions are the following:

(1) An extension of the monoid algebra with a fixpoint operator. This enables the expression of iteration in a more functional way than an imperative loop and makes it possible to define new rewriting rules;

(2) New optimization rules for terms using this fixpoint operator:

• We show that under reasonable conditions, this fixpoint can be considered as a monoid homomorphism, and can thus be evaluated by parallel loops with one final merge rather than by a global loop requiring network overhead after each iteration; • We also present new rewriting rules with criteria to push filters through a recursive term, for filtering inside a fixpoint before a join, and for pushing aggregations into recursive terms;

• Finally, we present experimental evidence that these new rules generate significantly more efficient programs.

THE 𝜇-MONOIDS ALGEBRA

In this section, we describe a core calculus, which we call 𝜇-monoids, intended to model a subset of a programming language L (e.g. Scala2 ) that is used for computations on a big data framework (through an API provided by the framework). 𝜇-monoids aims at being as general as possible, while focusing on formalizing computations subject to optimization. It is based upon the monoid algebra of Fegaras [START_REF] Fegaras | An algebra for distributed Big Data analytics[END_REF]. Dataset manipulations are captured as algebraic operations, and specific operations on elements of those datasets are captured as functional expressions that are passed as arguments to some of the algebraic operations. In 𝜇-monoids, we formalize some of those functional constructs, specifically the ones that we need to analyse in the algebraic expressions. For example, some optimization rules need to analyse the pattern and body of flatmap expressions in order to check whether the optimization can take place.

Making explicit only the shapes that are interesting for the analysis enables to abstract from the specific programming language L that we optimize. This way, constructs of L other than those which we model explicitly are represented as constants 𝑐, as they are going to be left to L's compiler to typecheck and evaluate. We only assume that every constant 𝑐 has a type type(𝑐) which is either a basic type or a function type, and that, when its type is 𝑡 1 → 𝑡 2 , it can be applied to any argument of type 𝑡 1 to yield results of type 𝑡 2 .

We first describe the data model we consider, then in Sec. 2.2 we recall the main definitions of the monoid algebra proposed by Fegaras [START_REF] Fegaras | An algebra for distributed Big Data analytics[END_REF]. We then introduce a general notion of aggregation function in Sec. 2.3 and our addition to the monoid algebra, the fixpoint operator 𝜇, in Sec. 2.4. Then in Sec. 2.5 we define the syntax of our own core calculus, and, in Sec. 2.6, a minimal type system for it. We then proceed to give a denotational semantics for our specific constructs in Sec. 2.7 and discuss evaluation of expressions in Sec. 2.9.

2.1 Data model: distributed collections of data 2.1.1 Collection monoids. We are interested in programs which work on distributed datasets of an homogenous type.

Such a dataset consists in a number of records, which are all values of the same type, and we assume a cluster of networked machines where each machine stores some of the records.

Different abstraction levels are possible for such a distributed dataset. At the programming level, we usually want to abstract away from the partitioning, i. e. we consider two states of the storage as representing the same data if they contain the same records, regardless of the number of machines and of which machine holds which records. That way, the program is reasonably independent from the structure of the cluster it will be run on. We may or may not want to abstract away from the order in which the records are stored, and we may or may not want to abstract away from the number of times the same record appears. Depending of the abstraction level, we thus can see the dataset as a list, a bag, or a set of records. We regroup finite lists, finite bags and finite sets under the generic term of collections.

Notation. Given a data type 𝑡, and for Coll a sort of collection, i. e. one of List, Bag or FSet, we write Coll[𝑡] for the set of collections of the sort Coll containing values of type 𝑡.

Let us now recall the algebraic definition of a monoid: Definition 2.1. A monoid is a triple (𝑆, ⊗, 𝑒) where 𝑆 is a set, ⊗ an associative binary operation on 𝑆, and 𝑒 a neutral element for ⊗, i. e. such that:

∀𝑥, 𝑦, 𝑧 ∈ 𝑆 𝑥 ⊗ (𝑦 ⊗ 𝑧) = (𝑥 ⊗ 𝑦) ⊗ 𝑧 ∀𝑥 ∈ 𝑆 𝑥 ⊗ 𝑒 =𝑒 = 𝑒 ⊗ 𝑥
As noticed by Fegaras [START_REF] Fegaras | An algebra for distributed Big Data analytics[END_REF], our three sorts of collections are particularly useful for representing distributed data because they each have the algebraic structure of a monoid, where the neutral element is the empty collection and the associative operator is respectively list union (i. e. concatenation) ++, bag union ⊎ and set union ∪. Associativity means that the whole collection can be seen as the union of the subcollections stored on the different machines without specifying an order in which to apply the union operator.

The three sorts of collection monoids, as Fegaras terms them, can be related with equivalence relations, reflecting the fact that they represent different abstraction levels for the same data. To formalize this, we recall the algebraic definitions of congruence and quotient monoid: Definition 2.2 (Congruence). Let (𝐴, ⊗, 𝑒) be a monoid. Let ∼ be an equivalence relation on 𝐴. We say that ∼ is a congruence on the monoid if it is compatible with ⊗, i. e. if:

∀𝑎, 𝑏, 𝑎 ′ , 𝑏 ′ 𝑎 ∼ 𝑎 ′ ∧ 𝑏 ∼ 𝑏 ′ ⇒ 𝑎 ⊗ 𝑏 ∼ 𝑎 ′ ⊗ 𝑏 ′
Definition 2.3 (Quotient monoid). Let (𝐴, ⊗, 𝑒) be a monoid and ∼ a congruence on it. For 𝑎 ∈ 𝐴, let â = {𝑏 ∈ 𝐴 | 𝑏 ∼ 𝑎}, the equivalence class of 𝑎. Let  = { â | 𝑎 ∈ 𝐴} be the set of all equivalence classes, and let ⊗ be the operation on  defined by â ⊗ b def = 𝑎 ⊗ 𝑏. This is well-defined because ∼ is compatible with ⊗, so that the result is the same independently of the particular choice of 𝑎 and 𝑏 in their equivalence class.

Then ( Â, ⊗, ê) is a monoid, termed the quotient monoid of (𝐴, ⊗, 𝑒) by ∼ and noted (𝐴, ⊗, 𝑒)/∼. An important feature of Fegaras' monoid algebra, and of 𝜇-monoids, is that all algebraic operations are defined in a way which is agnostic to distribution. So, although we introduce the distinction between Bag 𝑙 [𝑡] and Bag 𝑑 [𝑡] in order to prevent nesting distributed bags, we will use the notation Bag[𝑡] to represent a bag which may or may not be distributed when both are possible and it does not affect the semantics.

Let

Fegaras' monoid algebra

In this section, we recall briefly the main definitions from Fegaras' monoid algebra [START_REF] Fegaras | An algebra for distributed Big Data analytics[END_REF], upon which our work is based.

The monoid algebra is based on the three sorts of collection monoids described in 2.1.1 and on collection homomorphisms.

We first recall the definition of a monoid homomorphism: Definition 2.4. Let (𝐴, ⊗, 𝑒) and (𝐵, ⊙, 𝜀) be two monoids. A monoid homomorphism from 𝐴 to 𝐵 is a function ℎ from 𝐴 to 𝐵 such that:

ℎ(𝑒) = 𝜀 and ∀𝑥, 𝑦 ∈ 𝐴 ℎ(𝑥 ⊗ 𝑦) = ℎ(𝑥) ⊙ (𝑦).
Collection homomorphisms are now defined using the following universal property enjoyed by the collection monoids (which are free structures in the algebraic sense):

Property 1 (universal property of collection monoids). For Coll a collection monoid, let U Coll be the corresponding singleton construction function. Let (𝐴, ⊗, 𝑒) be a monoid which satisfies all the algebraic laws of Coll (i. e. commutativity for bags, and commutativity and idempotence for sets). Let 𝑓 : 𝑡 → 𝐴 be a function.

Then there exists a unique monoid homomorphism 𝐻 ⊗ 𝑓 :

Coll[𝑡] → 𝐴 such that: 𝐻 ⊗ 𝑓 (U Coll (𝑥)) = 𝑓 (𝑥).
This collection homomorphism applies the function 𝑓 to all the elements of the input collection and combines all the results together with the operation ⊗, yielding a single element of 𝐴.

Fegaras' monoid algebra comprises a number of collection homomorphisms, all defined in the form 𝐻 ⊗ 𝑓 for appropriate 𝑓 s and ⊗s. We refer the reader to [START_REF] Fegaras | An algebra for distributed Big Data analytics[END_REF] for the detail. In the present work, we use a slightly different set. Namely:

• we do not consider orderBy because we concentrate on bags rather than lists;

• we use reduceByKey rather than groupBy (together with the other operations, they lead to the same expressivity);

• we add the join operation. Even though it can be expressed in terms of coGroup, this operation is useful for us to have as a primitive because it is an homomorphism from each of its two arguments separately, whereas coGroup is only a binary homomorphism (an homomorphism from the product monoid of its two arguments).

Our set of primitive operations is thus the following, here presented in a way adapted to our default abstraction level of bags:

• flmap(𝑓 , 𝑋 ), with 𝑓 : 𝑡 1 → Bag 𝑙 [𝑡 2 ] and 𝑋 : Bag[𝑡 1 ] is the flatmap operation: it applies 𝑓 to each element of 𝑋 and merges all the results into a single bag using bag union ⊎. This operation is a monoid homomorphism from Bag[𝑡 1 ] to Bag[𝑡 2 ], so that if 𝑋 is distributed it can be run separately on each local subcollection without any data exchange. Note the restriction that 𝑓 is not allowed to return distributed bags. It makes the flatmap operator less general than the mathematical function it represents but reflects what we have in distributed data frameworks.

• reduce(⊕, 𝑒 ⊕ , 𝑋 ), with 𝑋 : Bag[𝑡] and (𝑡, ⊕, 𝑒 ⊕ ) a commutative monoid, reduces the input dataset by combining all its elements with ⊕. For example: reduce(+, 0, {{1, 4, 6}}) = 11. This operation is a monoid homomorphism from Bag[𝑡] to (𝑡, ⊕, 𝑒 ⊕ ), so that if 𝑋 is distributed it can be run separately on each local subcollection before combining all the local results once.

• reduceByKey(⊕, 𝑋 ), with 𝑋 : Bag[𝑡 1 × 𝑡 2 ] and ⊕ an associative and commutative binary operation on 𝑡 2 , takes as argument a bag of elements in the form (𝑘, 𝑣) (key-value pair) and combines all values 𝑣 having the same key 𝑘 into a single one using the ⊕ operator. For example: reduceByKey(+, {{(1, 2), (1, 4), (2, 2), (2, 1), (1, 3)}}) = {{(1, 9), (2, 3)}}}. This operation is a monoid homomorphism which can again be run separately on each local subcollection before combining the results.

• join(𝑋, 𝑌 ), with 𝑋 :

Bag[𝑡 1 × 𝑡 2 ] and 𝑌 : Bag[𝑡 1 × 𝑡 3 ],
is the join-by-key operation: it takes two collections of elements of the form (𝑘, 𝑣) and (𝑘, 𝑤), and returns a collection of elements of the form (𝑘, (𝑣, 𝑤)), one for each pair (𝑣, 𝑤) of values having the same key 𝑘. If a key appears 𝑛 times in one input dataset and 𝑚 times in the other, it appears 𝑛𝑚 times in the result. It is a monoid homomorphism from each of its arguments to Bag[𝑡 1 × (𝑡 2 × 𝑡 3 )], so that if any of the input bags is distributed it can be run separately on each local subcollection for that one.

Note that, algebraically, the join operation can be written with flatmaps (this is a feature of all homomorphisms from bags to bags); however, if both inputs are distributed then this is not possible in 𝜇-monoids without violating the restriction on the functional argument of flatmap, which justifies including join as a primitive.

• cogroup(𝑋, 𝑌 ), with 𝑋 : Bag[𝑡 1 × 𝑡 2 ] and 𝑌 : Bag[𝑡 1 × 𝑡 3 ], takes two collections of elements of the form (𝑘, 𝑣) and (𝑘, 𝑤) and returns a collection of elements of the form (𝑘, (𝑉 ,𝑊 )) where 𝑉 and 𝑊 are the sets of v values and w values having the same key 𝑘. This operation is an homomorphism from the product monoid

Bag[𝑡 1 × 𝑡 2 ] × Bag[𝑡 1 × 𝑡 3 ] to Bag[𝑡 1 × (Bag 𝑙 [𝑡 2 ] × Bag 𝑙 [𝑡 3 ])].
Additionnally, Fegaras' monoid algebra includes a repeat operation, which is not an homomorphism. In contrast, the 𝜇 operation that we introduce is an homomorphism under certain conditions, as explained in Section 2.4. Before we get there, we introduce our notion of aggregation function.

Equivalence relations and aggregation functions

It is quite common in practice that a programmer is only interested in the set of values of a dataset, not in potential duplicates the bag representing the storage may contain. So this programmer will work with bags up to ∼ idem (see Sec. 2.1.1). We can notice that each equivalence class of bags has a canonical representant: the bag where each element appears only once. Let distinct : Bag[𝑡] → Bag[𝑡] be the function which removes duplicates, returning this canonical representant. This function is useful, but costly to compute in a distributed context, since duplicates can occur across different machines and eliminating them thus involves a lot of communication over the network. Therefore, it should not be used all the time but only when necessary: bags with duplicates can be used in intermediary computation steps, where we tolerate redundant information temporarily.

Sometimes, the programmer is not even interested in the whole set of values, but only in more synthetic information about the dataset. For example, in the shortest path problem: if we have a dataset containing paths together with their length and this dataset contains two paths from 𝑎 to 𝑏 with different lengths, then the longer path is irrelevant and can be considered redundant even though it is not the same value as the other one. It is useful to also think of this situation in terms of an equivalence relation: two bags are equivalent for this purpose iff they contain exactly the same shortest paths. Then the canonical representant of an equivalence class is the bag with no duplicates which does not contain any non-shortest path. We can also see that the function 𝛿 which removes non-shortest paths (and duplicates) from a dataset has features analogous to distinct, as we will detail below. We regroup such functions under the term aggregation functions. Definition 2.5 (aggregation function). We call aggregation function any function 𝛿 : Bag[𝑡] → Bag[𝑡] with the following properties:

• 𝛿 ({{}}) = {{}} • ∀𝑎, 𝑏 𝛿 (𝑎 ⊎ 𝑏) = 𝛿 (𝛿 (𝑎) ⊎ 𝛿 (𝑏))
Note that these two properties also imply that 𝛿 is idempotent.

Remark that our definition excludes some functions which could be considered aggregators in a more general sense, e. g. functions computing an average. Definition 2.6. The equivalence relation associated with an aggregation function, ∼ 𝛿 , is defined by:

𝑎 ∼ 𝛿 𝑏 def ⇔ 𝛿 (𝑎) = 𝛿 (𝑏)
Lemma 2.7. Let 𝛿 be an aggregation function, then ∼ 𝛿 is compatible with the monoid operation ⊎, i. e. we have:

∀𝑎, 𝑏, 𝑎 ′ , 𝑏 ′ 𝑎 ∼ 𝛿 𝑎 ′ ∧ 𝑏 ∼ 𝛿 𝑏 ′ ⇒ 𝑎 ⊎ 𝑏 ∼ 𝛿 𝑎 ′ ⊎ 𝑏 ′ Proof. We have 𝛿 (𝑎 ⊎ 𝑏) = 𝛿 (𝛿 (𝑎) ⊎ 𝛿 (𝑏)) = 𝛿 (𝛿 (𝑎 ′ ) ⊎ 𝛿 (𝑏 ′ )) = 𝛿 (𝑎 ′ ⊎ 𝑏 ′ ). □ Definition 2.8.
Let 𝛿 be an aggregation function, we define the binary operation ⊗ 𝛿 on bags as follows: 1) is true. Let 𝑎 and 𝑏 be any bags. We have 𝜑 (𝑎) ⊗ 𝛿 𝜑 (𝑏) = 𝛿 (𝜑 (𝑎) ⊎ 𝜑 (𝑏)) = 𝛿 (𝜑 (𝑎 ⊎ 𝑏)) (because 𝜑 is a homomorphism). We also have 𝑎 ⊎ 𝑏 ∼ 𝛿 𝛿 (𝑎 ⊎ 𝑏), by definition of ∼ 𝛿 since 𝛿 is idempotent. Therefore, using (1), 𝜑 (𝑎 ⊎ 𝑏) ∼ 𝛿 𝜑 (𝛿 (𝑎 ⊎ 𝑏)), and this last term is 𝜑 (𝑎 ⊗ 𝛿 𝑏); hence (2). Assume (2) is true. Let 𝑎 be any bag, by taking for 𝑏 the empty bag and using the definitions, (2) yields 𝛿 (𝜑 (𝛿 (𝑎⊎{{}}))) = 𝛿 (𝜑 (𝑎) ⊎ 𝜑 ({{}})). Since 𝜑 is an homomorphism we have 𝜑 ({{}}) = {{}}, thus 𝛿 (𝜑 (𝛿 (𝑎))) = 𝛿 (𝜑 (𝑎)); hence (3). Assume (3) is true. Let 𝑎 and 𝑏 such that 𝑎 ∼ 𝛿 𝑏. Using (3) and the definition of ∼ 𝛿 , we have 𝛿 (𝜑 (𝑎)) = 𝛿 (𝜑 (𝛿 (𝑎))) = 𝛿 (𝜑 (𝛿 (𝑏))) = 𝛿 (𝜑 (𝑏)); hence (1). □ Definition 2.12. We say that 𝜑 is compatible with 𝛿 if any of these properties is true. Note that (2) can also be formulated as:

𝑎 ⊗ 𝛿 𝑏 def = 𝛿 (𝑎 ⊎ 𝑏) Lemma 2.
(1) ∀𝑎, 𝑏 𝑎 ∼ 𝛿 𝑏 ⇒ 𝜑 (𝑎) ∼ 𝛿 𝜑 (𝑏) (2) ∀𝑎, 𝑏 𝜑 (𝑎 ⊗ 𝛿 𝑏) ∼ 𝛿 𝜑 (𝑎) ⊗ 𝛿 𝜑 (𝑏) (3) 𝛿 • 𝜑 • 𝛿 = 𝛿 • 𝜑 Proof. Assume (
𝛿 • 𝜑 is a monoid homomorphism from 𝑀 𝛿 to 𝑀 𝛿 .
This definition is strongly related to the premappability condition in Datalog [START_REF] Zaniolo | Fixpoint semantics and optimization of recursive Datalog programs with aggregates[END_REF], as made more apparent by property

(3).

Property 2. distinct is compatible with all homomorphisms 𝜑 :

Bag[𝑡] → Bag[𝑡].
Proof. In the following, we write 𝑚 • 𝑋 , where 𝑋 is a bag, to denote 𝑋 combined 𝑚 times with itself using ⊎.

Let 𝐴 be a finite bag of values of type 𝑡. Let 𝑎 1 , ..., 𝑎 𝑛 be the distinct values it contains and 𝑚 1 , ..., 𝑚 𝑛 the number of times each one appears in the bag. Since 𝜑 is an homomorphism, we have

𝜑 (𝐴) = 1≤𝑖 ≤𝑛 𝑚 𝑖 • 𝜑 ({{𝑎 𝑖 }}). Then, distinct(𝜑 (𝐴)) = 1≤𝑖 ≤𝑛 distinct(𝜑 ({{𝑎 𝑖 }})): this is independent from the 𝑚 𝑖 (since ∪ is idempotent). Thus, distinct(𝜑 (𝐴)) = distinct(𝜑 (distinct(𝐴))). □ 2.

The 𝜇 operation

Our purpose is to extend the monoid algebra (as defined in 2.2) with an operator for expressing iteration which allows effective optimisations when working with a distributed dataset (note that this does not preclude more general loops outside our algebra).

As a first idea, consider the following type of iteration. Let 𝜑 : Bag[𝑡] → Bag[𝑡] be a monoid homomorphism. We start with a bag 𝑅, then:

(1) at each iteration, 𝜑 is executed on the result of the previous iteration;

(2) the results of all iterations are accumulated into a single bag;

(3) it ends when 𝜑 adds nothing to the results; then the bag of all results is returned.

Algebraically, this amounts to computing Therefore, it makes sense to periodically remove duplicates. However, it may not be necessary to remove them globally (which is costly as it involves network communication), as we will detail in Section 3.4. More generally, we can add to the loop, as a parameter, an aggregation function 𝛿 to be run at each iteration step. Our general iteration operator 𝜇 is thus: computes the following sequences:

• 𝑅 0 = 𝛿 (𝑅) • 𝑆 0 = 𝑅 0 • 𝑅 𝑛+1 = 𝛿 (𝜑 (𝑅 𝑛 )) • 𝑆 𝑛+1 = 𝑆 𝑛 ⊗ 𝛿 𝑅 𝑛+1
until it reaches an 𝑁 such that 𝑆 𝑁 +1 = 𝑆 𝑁 ; then it returns 𝑆 𝑁 .

Note that the first idea discussed before is the particular case where 𝛿 is the identity function. Also note that the requirement that 𝛿 and 𝜑 are compatible is automatically true when 𝛿 is either the identity function or distinct.

In the following, we consider distinct the default aggregation function and write 𝜇 (𝑅, 𝜑) as a shortcut for 𝜇 distinct (𝑅, 𝜑).

Our idea is that programmers would usually not write 𝜇 terms with a different 𝛿 themselves, but they can be obtained through rewriting and used for optimization (Sec. 3.3).

Syntax of 𝜇-monoids

The syntax of expressions is formally defined as follows:

𝜋 ::= 𝑎 | 𝐶 (𝜋 1 , 𝜋 2 , ..., 𝜋 𝑛 )
pattern: variable, constructor pattern To this, we add the following as syntactic sugar:

𝑒 ::= 𝑐 | 𝑎 | {{𝑒 }} expression: constant, variable, singleton | 𝜆 ⟨𝜋 1 → 𝑒 1 | ••• | 𝜋 𝑛 → 𝑒 𝑛 ⟩
• (𝑒 1 , ..., 𝑒 𝑛 ) with no constructor is an abbreviation for: Tuple(𝑒 1 , ..., 𝑒 𝑛 )

• if 𝑒 then 𝑒 1 else 𝑒 2 is an abbreviation for: 𝜆 ⟨True → 𝑒 1 | False → 𝑒 2 ⟩ 𝑒, i. e. a particular case of pattern-matching against the two constant constructors True and False representing Boolean values. • groupby(𝑒) is an abbreviation for: reduceByKey(⊎, flmap(𝜆 ⟨(𝑘, 𝑣) → (𝑘, {{𝑣 }})⟩, 𝑒))

• Constants 𝑐 can also represent functions (defined in the language L). We consider operators such as the bag union operator ⊎ as constant functions of two arguments and use the infix notation as syntactic sugar.

• To make examples more readable, we use the let name = 𝑒 1 in 𝑒 2 syntax with the usual meaning.

Example:

let appendToWords = 𝜆 ⟨𝑋 → flmap(𝜆 ⟨𝑥 → flmap(𝜆 ⟨𝑐 → if contains 𝑥 𝑐 then {{ }} else {{𝑥 + 𝑐 }}⟩, 𝐶) ⟩, 𝑋 ) ⟩ in 𝜇 (𝐶, appendToWords)
This expression computes the set of all possible words with no repeated letters that can be formed from a set of characters C. We assume that + and contains are defined in L: + appends its second argument to the first and contains checks whether the first argument is contained in the second argument. appendToWords thus returns a new set of words from a given set of words 𝑋 by appending to each of the words in 𝑋 each letter in 𝐶 whenever it was not already present.

The iteration operator computes the following, where we consider 𝐶 = {{𝑎, 𝑏, 𝑐}} -the fixpoint is reached in 3 steps:

𝑅 0 = 𝐶 𝑆 0 = 𝐶 𝑅 1 = distinct(𝜑 (𝐶)) = {{𝑎𝑏, 𝑎𝑐, 𝑏𝑎, 𝑏𝑐, 𝑐𝑎, 𝑐𝑏 }} 𝑆 1 = 𝐶 ∪ 𝑅 1 = {{𝑎𝑏,
𝑅 3 = distinct(𝜑 (𝑅 2 )) = {{ }} 𝑆 3 = 𝑆 2 ∪ 𝑅 3 = 𝑆 2

Well-typed terms

We define typing rules for algebraic terms, in order to exclude meaningless terms. In these rules, we use type environments Γ which bind variables to types. An environment contains at most one binding for a given variable. We combine them in two different ways:

• Γ ∪ Γ ′ is only defined if Γ and Γ ′ have no variable in common, and is the union of all bindings in Γ and Γ ′ ;

• Γ + Γ ′ is defined by taking all bindings in Γ ′ plus all bindings in Γ for variables not appearing in Γ ′ . In other words, if a variable appears in both, the binding in Γ ′ overrides the one in Γ.

Definition 2.14 (matching). We first define the environment obtained by matching a data type to a pattern by the following:

match(𝑎, 𝑡 ) → 𝑎 : 𝑡 ∀𝑖 match(𝜋 𝑖 , 𝑡 𝑖 ) → Γ 𝑖 match(𝐶 (𝜋 1 , ..., 𝜋 𝑛 ), 𝐶 [𝑡 1 , ..., 𝑡 𝑛 ]) → Γ 1 ∪ ••• ∪ Γ 𝑛
If, according to these rules, there is no Γ such that match(𝜋, 𝑡) → Γ holds, we say that pattern 𝜋 is incompatible with type 𝑡. Note that, with our conditions, a pattern containing several occurrences of the same variable is not compatible with any type and hence cannot appear in a well-typed term, as the typing rules will show.

In order to type functions with pattern-matching, we define the following operation for combining sum types:

Definition 2.15. The operation + on sum types is defined recursively as follows. Let 𝑡 be a sum type and 𝐶 a constructor not appearing in 𝑡, then:

𝑡 + (𝑡 ′ 1 | | ••• | | 𝑡 ′ 𝑚 ) = (𝑡 + 𝑡 ′ 1 ) + (𝑡 ′ 2 | | ••• | | 𝑡 ′ 𝑚 ) 𝑡 + 𝐶 [𝑡 1 , ..., 𝑡 𝑛 ] = 𝑡 | | 𝐶 [𝑡 1 , ..., 𝑡 𝑛 ] (𝑡 | | 𝐶 [𝑡 1 , ..., 𝑡 𝑛 ]) + 𝐶 [𝑡 ′ 1 , ..., 𝑡 ′ 𝑛 ] = 𝑡 | | 𝐶 [𝑡 1 + 𝑡 ′ 1 , ..., 𝑡 𝑛 + 𝑡 ′ 𝑛 ]
If 𝑡 is not a sum type, we define 𝑡 + 𝑡 = 𝑡. The type 𝑡 + 𝑡 ′ is not defined if 𝑡 ≠ 𝑡 ′ and 𝑡 or 𝑡 ′ is not a sum type, or if they have constructors in common with incompatible type parameters, i. e. type parameters which cannot themselves be combined with +.

Definition 2.16 (Subtyping). We define subtyping as follows (it is nontrivial only for sum types):

𝑡 <: 𝑡 ′ def ⇔ 𝑡 + 𝑡 ′ = 𝑡 ′
Definition 2.17 (Well-typed terms). A term 𝑒 is well-typed in a given environment Γ iff Γ ⊢ 𝑒 : 𝑡 for some type 𝑡, as judged by the relation defined in Figure 3. In these rules, 𝑇 represents one of Bag 𝑙 or Bag 𝑑 .

Γ (𝑎) = 𝑡 Γ ⊢ 𝑎 : 𝑡 type(𝑐) = 𝑡 Γ ⊢ 𝑐 : 𝑡 Fig. 3. Typing judgements.
Note that these rules do not give a way to infer the parameter type of a 𝜆 expression in general; we assume some mechanism for that in the language L.

2.6.1 Additional restrictions. In addition to the constraints imposed by our type system, some operations require their operands to fulfill certain criteria in order to be well-defined:

• reduce(𝑓 , 𝑧, 𝐴) and reduceByKey(𝑓 , 𝐴): 𝑓 is associative and commutative, and 𝑧 is a neutral element for 𝑓 .

• 𝜇 𝛿 (𝑅, 𝜑): 𝜑 is a monoid homomorphism, 𝛿 is an aggregation function, and they are compatible.

The user needs to provide terms that satisfy these criteria since they cannot be verified statically in general. However, regarding the homomorphism criterion for 𝜑, even though we cannot check statically whether an arbitrary function is a monoid homomorphism, we can identify a subset of functions that can be statically checked. It is the set of terms 𝜑 of the form 𝜆 ⟨𝑋 → h(𝑋 )⟩ where h(𝑋 ) is defined as follows: where 𝑘𝑒𝑦𝑠 (𝐴) = distinct( {𝑘 | (𝑘, 𝑎) ∈ 𝐴 }); and ∪ is distinct union of bags.

Fig. 4. Denotational semantics cannot be expressed as a join. Thus, it makes sense to check statically whether the term provided by the programmer is of that form and issue a warning if it is not.

𝜇-monoids denotational semantics

Figure 4 gives the denotational semantics of the main algebraic operations. It assumes all terms are well-typed and satisfy the additional restrictions mentioned in Sec. 2.6.1. Each closed term has a denotation in the domain corresponding to its type, with the additional possible denotation 𝜔 which belongs to all types and represents a computation which does not terminate. Any of those operations returns 𝜔 when applied to 𝜔.

These operations, except 𝜇, are monoid homomorphisms [START_REF] Fegaras | An algebra for distributed Big Data analytics[END_REF], as discussed in 2.4. We can check that they are still monoid homomorphisms if we add 𝜔 to all the monoids as an absorbing element. So 𝑆 𝑛+1 depends only on 𝑆 𝑛 , making the definition in Fig. 4 consistent (if an 𝑁 is reached such that 𝑆 𝑁 +1 = 𝑆 𝑁 then the sequence becomes stationary) and meaning that 𝜇 𝛿 (𝑅, 𝜑) is a fixpoint4 of Ψ.

Properties of

We now prove the following theorem, which is crucial for optimizing distributed fixpoint computations:

Theorem 2.18. 𝜇 𝛿 (•, 𝜑) is a monoid homomorphism from Bag[𝑡] ∪ {𝜔 } to 𝑀 𝛿 ∪ {𝜔 }. Proof. 𝜇 𝛿 ({{}}, 𝜑) = {{}} is immediate. Let 𝑅 0 = 𝑅 1 ⊎ 𝑅 2 .
For all 𝑛 and for 𝑖 in 0, 1, 2, we write

𝑆 𝑖 𝑛 = 𝛿 𝑘 ≤𝑛 (𝛿 • 𝜑) 𝑘 (𝛿 (𝑅 𝑖 )). Since 𝛿 is a homomorphism from bags to 𝑀 𝛿 and 𝛿 • 𝜑 is a homomorphism from 𝑀 𝛿 to 𝑀 𝛿 , we have 𝑆 0 𝑛 = 𝑆 1 𝑛 ⊗ 𝛿 𝑆 2
𝑛 for all 𝑛. Thus, if (𝑆 1 𝑛 ) and (𝑆 2 𝑛 ) both become stationary at some point, say 𝑁 1 and 𝑁 2 , then (𝑆 0 𝑛 ) becomes stationary at 𝑚𝑎𝑥 (𝑁 1 , 𝑁 2 ) and we do have

𝜇 𝛿 (𝑅 0 , 𝜑) = 𝜇 𝛿 (𝑅 1 , 𝜑) ⊗ 𝛿 𝜇 𝛿 (𝑅 2 , 𝜑). □ 2.

Examples

We present in this section examples of recursive programs expressed in 𝜇-monoids.

Transitive closure (TC). where 𝑅 is a dataset of tuples (source, destination) representing the edges of a graph.

This expressions computes the entire transitive closure of the input graph 𝑅.

The sub-expression join(flmap(reverse_edges, 𝑋 ), 𝑅) joins a path from 𝑋 with a path from 𝑅 when the target node of the first path corresponds to the start node of the second path. This intermediary node is the join key, thus the join constructs pairs of the form (𝑚𝑖𝑑, (𝑠𝑟𝑐, 𝑑𝑒𝑠𝑡)) where 𝑚𝑖𝑑 is the intermediary node, which must then be dropped by another flatmap. So, at each iteration, the paths in 𝑋 obtained in the last iteration get appended with edges from 𝑅 whenever possible. The computation ends when no new paths are found.

Shortest path (SP). where 𝑅 is a dataset of tuples ((source, destination), weight) representing the weighted edges of a graph.

The expression computes the shortest path between each pair of nodes in the input graph 𝑅. New paths are computed by performing a transitive closure while summing the lengths of the joined paths. Finally, the reduceByKey operation keeps the shortest paths between each pair of nodes. where R is a set of routes between two cities, represented as pairs of cities. Each city City(𝑛, 𝑙) has a name 𝑛 and a list of landmarks 𝑙 and each landmark Landmark(𝑛, 𝑟 ) has a rating 𝑟 . bestRated(𝑙 1 , 𝑙 2 ) is a function that returns the best set of landmarks based on its ratings.

The fixpoint all_paths computes the set of landmarks that can be visited for each possible path between each two cities. The final term then computes the best path between Paris and Geneva.

Movie Recommendations.

let users_who_like = 𝜆 ⟨𝑥 → flmap(𝜆 ⟨User(𝑢, 𝑏𝑚) → if 𝑥 ∈ 𝑏𝑚 then 𝑏𝑚 else {{ }}⟩, 𝑈 ) ⟩ in 𝜇 (𝑆, 𝜆 ⟨𝑋 → flmap(users_who_like, 𝑋 ) ⟩)
where 𝑈 is a set of users, each user User(𝑢, 𝑏𝑚) has a set of best movies 𝑏𝑚.

The query computes a set of recommended movies by starting from a set of movies 𝑆 and by adding the best movies of a user if one of his best movies is in the set of recommended movies until no new movie is added.

2.9 Evaluation of expressions 2.9.1 Local execution.

Pattern matching and function application. The result of matching a value against a pattern is either a set of pattern variable assignments or ⊥. It is defined as follows:

m(𝑣, 𝑎) = {𝑎 ↦ → 𝑣 } m(𝐶 (𝑣 1 , ..., 𝑣 𝑛 ), 𝐶 (𝜋 1 , ..., 𝜋 𝑛 )) = m(𝑣 1 , 𝜋 1 ) ∪ ••• ∪ m(𝑣 𝑛 , 𝜋 𝑛 ) m(𝐶 (•••), 𝐶 ′ (•••)) = ⊥ if 𝐶 ≠ 𝐶 ′
where we extend ∪ so that ⊥ ∪ 𝑆 = ⊥.

A lambda expression 𝑓 = 𝜆 ⟨𝜋 1 → 𝑒 1 | ••• | 𝜋 𝑛 → 𝑒 𝑛 ⟩
contains a number of patterns together with return expressions.

When this lambda expression is applied on an argument 𝑣 (𝑓 𝑣), the argument is matched against the patterns in order, until the result of the match is not ⊥. Let 𝑖 be the smallest index such that m(𝑣, 𝜋 𝑖 ) = 𝑆 ≠ ⊥, the result of the application is obtained by substituting the free pattern variables in 𝑒 𝑖 according to the assignments in 𝑆. This means that in our algebra, all operators apart from flatmap, or join when one of the parameters is a local bag, need to send data across the network (for executing the non-local version of their monoid operator). The execution of these non-local operators depends on the distributed platform. Spark for example performs shuffling to redistribute the data across partitions for the computation of certain of its operations like cogroup and groupByKey.

Monoid

OPTIMIZATIONS

In this section, we propose new optimization rules for terms with fixpoints, and describe when and how they apply.

The purposes of the rules are (i) to identify which basic operations within an algebraic term can be rearranged and under which conditions, and (ii) to describe how new terms are produced or evaluated after transformation.

We first give the intuition behind each optimization rule before zooming on each of them to formally describe when they apply. The four new optimization rules are:

• PF is a rewrite rule of the form: 𝐹 (𝜇 (𝑅, 𝜑)) -→ 𝜇 (𝐹 (𝑅), 𝜑) it aims at pushing a filter F inside a fixpoint, whenever this is possible. A filter is a function which keeps only some elements of a dataset based on their values; we define it formally in Sec. 3.1.1.

• PJ is a rewrite rule of the form: join(𝐴, 𝜇 (𝑅, 𝜑)) -→ join(𝐴, 𝜇 (𝐹 𝐴 (𝑅), 𝜑)) it aims at inserting a filter 𝐹 𝐴 inside a fixpoint before a join is performed. It is inspired by the semi-join found in relational databases, and tailored for 𝜇-monoids.

• PA is a rewrite rule of the form:

𝛿 (𝜇 (𝑅, 𝜑)) -→ 𝜇 𝛿 (𝑅, 𝜑)
It aims at pushing an aggregation function 𝛿 inside a fixpoint, transforming a simple fixpoint into a fixpoint with aggregation. This rule requires 𝛿 to be compatible with 𝜑; it is inspired from the premappability condition in Datalog [START_REF] Zaniolo | Fixpoint semantics and optimization of recursive Datalog programs with aggregates[END_REF].

• an optimization rule P dist that determines how a fixpoint term is evaluated in a distributed manner by choosing among two possible execution plans.

Pushing filter inside a fixpoint (PF)

3.1.1 Filter depending on a single pattern variable. Definition 3.1 (filter). We call filter a function of the form:

𝜆 ⟨𝐷 → flmap(𝜆 ⟨𝜋 → if 𝑐 (𝑎) then {{𝜋 }} else {{}}⟩, 𝐷)⟩
where 𝜋 is a pattern containing the variable 𝑎 and 𝑐 (𝑎) is a Boolean condition depending on the value of 𝑎.

Such a function returns the dataset 𝐷 filtered by retaining only the elements whose value for 𝑎 (as determined by pattern-matching that element with 𝜋) satisfies 𝑐 (𝑎). The elements are unmodified, so the result is a subcollection of 𝐷.

In the following, we consider a filter 𝐹 with 𝜋 and 𝑎 defined as above, and we denote by 𝜋 𝑎 the function that matches an element against 𝜋 and returns the value of 𝑎 (𝜋 𝑎 = 𝜆 ⟨𝜋 → 𝑎⟩). For instance, 𝜋 𝑎 ((1, (5, 6))) = 5 for 𝜋 = (𝑥, (𝑎, 𝑦)).

Let us consider a dataset 𝐷. In terms of denotational semantics, with the notations above, we have

𝐹 (𝐷) = {𝑑 ∈ 𝐷 | 𝑐 (𝜋 𝑎 (𝑑))}.
The PF rule. This rule consists in transforming an expression of the form 𝐹 (𝜇 (𝑅, 𝜑)) to an expression of the form 𝜇 (𝐹 (𝑅), 𝜑), where F is a filter.

In the second form, the filter is pushed before the fixpoint operation. In other words, the constant part R is filtered first before applying the fixpoint on it. We now present sufficient conditions for the two terms to be equivalent.

PF condition. Let (C) be the following condition:

∀𝑟 ∈ 𝑅 ∀𝑠 ∈ 𝜑 ({{𝑟 }}) 𝜋 𝑎 (𝑟 ) = 𝜋 𝑎 (𝑠)
Intuitively, this condition means that the operation 𝜑 does not change the part of its input data that corresponds to 𝑎 in the pattern 𝜋, which is the part used in the filter; so for each record in the fixpoint that does not pass the filter, the record in R that has originated it does not pass the filter and the other way round. That is why we can just filter R in the first place.

Let 𝐴 = flmap(𝜆 ⟨𝜋 → if 𝑐 (𝑎) then {{𝜋 }} else {{}}⟩, 𝜇 (𝑅, 𝜑)). We prove that if (C) is satisfied, then 𝐴 = 𝜇 (𝐹 (𝑅), 𝜑). To prove this, we use the following property of fixpoints where 𝛿 = distinct: Lemma 3.2. ∀𝑎 ∈ 𝜇 (𝑅, 𝜑) 𝑎 ∈ 𝜑 (𝑛) ({{𝑟 }}) for some 𝑟 ∈ 𝑅 and 𝑛 ∈ 𝑁 Proof. We have:

𝜇 (𝑅, 𝜑) = 𝑛 ∈N 𝜑 (𝑛) (𝑅) = 𝑛 ∈N 𝜑 ( 𝑟 ∈𝑅 {{𝑟 }}) = 𝑛 ∈N ( 𝑟 ∈𝑅 𝜑 ({{𝑟 }})) = 𝑛 ∈N ( 𝑟 ∈𝑅 𝜑 ({{𝑟 }}))

□

Using the above lemma and condition (C), we have:

( * ) ∀𝑠 ∈ 𝜇 (𝑅, 𝜑) ∃𝑟 ∈ 𝑅 𝜋 𝑎 (𝑟 ) = 𝜋 𝑎 (𝑠)
We now prove 𝐴 = 𝜇 (𝐹 (𝑅), 𝜑) by proving the two inclusions:

( Verifying the condition (C) using type inference. We will start by explaining the intuition behind this before going into the details.

For the condition (C) to hold, we need to make sure that the part of the data extracted by 𝜋 𝑎 is not modified by 𝜑.

For this, our solution is inspired by the idea that the type of a parametric polymorphic function tells us information about its behaviour [START_REF] Wadler | Theorems for Free![END_REF]: for example, if 𝑓 is a polymorphic function whose argument contains exactly one value of the undetermined type 𝛼 and whose result must also contain a value of type 𝛼, then the 𝛼 value in the result is necessarily the one in the argument (𝑓 :

𝛼 → 𝛼 ⇒ ∀𝑥 𝑓 (𝑥) = 𝑥).
This reasoning can also be used for a more complex input type 𝐶 (𝛼) that contains a polymorphic type 𝛼. For instance:

𝐶 (𝛼) = 𝐴(𝐵(𝛼), 𝐷) is such a type given that 𝐴,𝐵 and 𝐷 are type constuctors. So our goal is, given that 𝜑 takes as input a bag of elements of type C, to find an appropriate polymorphic type 𝐶 (𝛼) that will be used for type checking 𝜑. In practice, we translate the 𝜑 operation to a Scala function that takes a polymorphic input type and use the Scala type inference system [START_REF] Odersky | An overview of the Scala programming language[END_REF] to get the output type 5 . 𝐶 (𝛼) should be built in such a way that the position of 𝛼 in 𝐶 (𝛼) is the same as the position of 𝑎 in 𝜋. Such a type is possible to build because the type 𝐶 matches the pattern 𝜋, otherwise the filtered term would not be type correct. Finally, if the output type also contains the type 𝛼 and has the same position as 𝑎 in 𝜋 then we can show that the condition (C) holds. Note that we do not need a full-fledged parametricity theorem for this: we only use the fact that the Scala type system has singleton types for all values.

Building 𝐶 (𝛼). Types are made from type constructors and basic types, and patterns are made from type constructors and pattern variables. So we can represent their structures using trees. In the following we sometimes refer to types by the trees representing them.

Definition 3.3 (path).

We define the path to the node labelled 𝑛 in the tree 𝑇 denoted path(𝑛,𝑇 ) by the ordered sequence Seq(𝑎 𝑖 ) where 𝑎 𝑖 is the next child arity of the 𝑖th visited node to reach 𝑛 from the root of the tree. A node in a tree can be identified by its path.

Let us consider the function replace 𝛼 (𝑝,𝑇 ) that, given a path 𝑝 and a type 𝑇 returns a polymorphic type 𝑇 (𝛼) 2)) so 𝜋 𝑎 (𝑠) = 𝜋 𝑎 (𝑟 ) hence (C).

that
3.1.2 Filters depending on multiple variables. We showed that (C): ∀𝑟 ∈ 𝑅 ∀𝑠 ∈ 𝜑 ({{𝑟 }}) 𝜋 𝑎 (𝑟 ) = 𝜋 𝑎 (𝑠) is sufficient for pushing the filter in a fixpoint when the filter condition depends on 𝑎. We can easily show that when the filter depends on a set of pattern variables 𝑉 , the sufficient condition becomes: ∀𝑟 ∈ 𝑅 ∀𝑠 ∈ 𝜑 ({{𝑟 }}) ∀𝑣 ∈ 𝑉 𝜋 𝑣 (𝑟 ) = 𝜋 𝑣 (𝑠). So, if one of the variables in 𝑉 does not satisfy the condition the filter would not be pushed. However, we can do better by trying to split the condition 𝑐 to two conditions 𝑐 1 and 𝑐 2 , such that 𝑐 = 𝑐 1 ∧ 𝑐 2 and 𝑐 1 depends only on the subset of variables that satisfies the condition (this splitting technique is used in [START_REF] Fegaras | Compile-Time Code Generation for Embedded Data-Intensive Query Languages[END_REF] to push filters in a cogroup or a groupby). If such a split is found, the filter flmap(𝜆 ⟨𝜋 → if 𝑐 then {{𝜋 }} else {{}}⟩, 𝑅) can be rewritten as flmap(𝜆 ⟨𝜋 → if 𝑐 2 then {{𝜋 }} else {{}}⟩, flmap(𝜆 ⟨𝜋 → if 𝑐 1 then {{𝜋 }} else {{}}⟩, 𝑅)). The inner filter can then be pushed.

Filtering inside a fixpoint before a join (PJ)

Let us consider the expression: join(𝐴, 𝐵), where 𝐴 is a constant and 𝐵 = 𝜇 (𝑅, 𝜑). After the execution of the fixpoint, the result is going to be joined with 𝐴, so only elements of this result sharing the same keys with 𝐴 are going to be kept.

So in order to optimize this term, we want to push a filter that keeps only the elements having a key in 𝐴. This way, elements not sharing keys with 𝐴 are going to be removed before applying the fixpoint operation on them.

(1) we show that 𝑗𝑜𝑖𝑛(𝐴, 𝐵) = 𝑗𝑜𝑖𝑛(𝐴, 𝐹 𝐴 (𝐵)), where 𝐹 𝐴 (𝐵) = {(𝑘, 𝑣) | (𝑘, 𝑣) ∈ 𝐵, ∃𝑤 (𝑘, 𝑤) ∈ 𝐴}:

We have 𝑗𝑜𝑖𝑛(𝐴, 𝐵) = {(𝑘, (𝑥, 𝑦)) | (𝑘, 𝑥) ∈ 𝐴, (𝑘, 𝑦) ∈ 𝐵}. So (𝑘, (𝑥, 𝑦)) ∈ 𝑗𝑜𝑖𝑛(𝐴, 𝐵) ⇔ (𝑘, 𝑥) ∈ 𝐴 ∧ (𝑘, 𝑦) ∈ 𝐵 ⇔ (𝑘, 𝑥) ∈ 𝐴 ∧ ((𝑘, 𝑦) ∈ 𝐵 ∧ ∃𝑤 (𝑘, 𝑤) ∈ 𝐴) ⇔ (𝑘, 𝑥) ∈ 𝐴 ∧ (𝑘, 𝑦) ∈ 𝐹 𝐴 (𝐵) ⇔ (𝑘, (𝑥, 𝑦)) ∈ 𝑗𝑜𝑖𝑛(𝐴, 𝐹 𝐴 (𝐵))).
(2) we show that 𝐹 𝐴 (𝐵) is a filter on B. This filter can be pushed when the criteria on pushing filters is fulfilled:

We can show that 𝐹 𝐴 (𝐵) = flmap(𝜆 ⟨(𝑘, 𝑣) → if 𝑐 (𝑘) then {{(𝑘, 𝑣)}} else {{}}⟩, 𝐵) where 𝑐 (𝑘) is the boolean expression that corresponds to the predicate ∃𝑤 (𝑘, 𝑤) ∈ 𝐴. This expression can be: 𝑐 (𝑘) = reduce(∨, 𝑒 ∨ , flmap(𝜆 ⟨(𝑘 ′ , 𝑎) → 𝑘 == 𝑘 ′ ⟩, 𝐴)). Which means that in case 𝜑 fulfills the criteria for pushing filters we will have 𝐹 𝐴 (𝐵) = 𝐹 𝐴 (𝜇 (𝑅, 𝜑)) = 𝜇 (𝐹 𝐴 (𝑅), 𝜑).

(3) we show as well that 𝐹 𝐴 (𝐵) = 𝐶 where:

𝐶 = flmap(𝜆 ⟨(𝑘, (𝑠 𝑥 , 𝑠 𝑦 )) → if 𝑠 𝑦 ≠ {{}} then flmap(𝜆 ⟨𝑥 → {{(𝑘, 𝑥)}}⟩, 𝑠 𝑥 ) else {{}}⟩, cogroup(𝐵, 𝐴))
We have: 𝐶 = (𝑘,(𝑠 𝑥 ,𝑠 𝑦 )) ∈cogroup(𝐵,𝐴) 𝑥 ∈𝑠 𝑥 (if 𝑠 𝑦 ≠ {{}} then {{(𝑘, (𝑥, 𝑦))}} else {{}})

(1) Let 𝑒 ∈ 𝐶. So ∃(𝑘, (𝑠 𝑥 , 𝑠 𝑦 ))) ∈ cogroup(𝐵, 𝐴) such that ∃𝑥 ∈ 𝑠 𝑥 𝑒 = (𝑘, 𝑥) and 𝑠 𝑦 ≠ {{}}.

We have (𝑘, (𝑠 𝑥 , 𝑠 𝑦 )) ∈ cogroup(𝐵, 𝐴), so 𝑠 𝑥 = {𝑣 | (𝑘, 𝑣) ∈ 𝐵}, which means that (𝑘, 𝑥) ∈ 𝐵 because 𝑥 ∈ 𝑠 𝑥 .

And 𝑠 𝑦 ≠ {{}} means that ∃𝑤 (𝑘, 𝑤) ∈ 𝐴, so 𝑒 = (𝑘, 𝑥) ∈ 𝐹 𝐴 (𝐵).

(2) Let (𝑘, 𝑥) ∈ 𝐹 𝐴 (𝐵). We have (𝑘, 𝑥) ∈ 𝐵 and ∃𝑤 (𝑘, 𝑤) ∈ 𝐴. So 𝑘 ∈ keys(𝐴) ∪ keys(𝐵).

Let 𝑠 𝑥 = {𝑣 | (𝑘, 𝑣) ∈ 𝐵} and 𝑠 𝑦 = {𝑣 | (𝑘, 𝑦) ∈ 𝐴}, so (𝑘, (𝑠 𝑥 , 𝑠 𝑦 )) ∈ cogroup(𝐵, 𝐴).
Since 𝑥 ∈ 𝑠 𝑥 and 𝑠 𝑦 ≠ {{}} (because ∃𝑤 (𝑘, 𝑤) ∈ 𝐴), then (𝑘, 𝑥) ∈ 𝐶.

Pushing aggregation into a fixpoint (PA)

The PA rule. consists in rewriting a term of the form 𝛿 (𝜇 (𝑅, 𝜑)) to a term of the form 𝜇 𝛿 (𝑅, 𝜑). It requires that 𝛿 is an aggregation function and compatible with 𝜑.

It is correct thanks to the following lemma: Proof. Let (𝑆 𝑛 ) and (𝑆 ′ 𝑛 ) be the 𝑆 sequences corresponding respectively to the two fixpoints; thus we have 𝑆 𝑛+1 = 𝑅 ⊎ 𝜑 (𝑆 𝑛 ) and 𝑆 ′ 𝑛+1 = 𝑅 ⊗ 𝛿 𝜑 (𝑆 ′ 𝑛 ) = 𝛿 (𝑅 ⊎ 𝜑 (𝑆 ′ 𝑛 )). We prove by induction on 𝑛 that 𝛿 (𝑆 𝑛 ) = 𝑆 ′ 𝑛 for any 𝑛: for 𝑛 = 0 we have 𝛿 (𝑆 0 ) = 𝛿 (𝑅) = 𝑆 ′ 0 . Assume 𝑆 ′ 𝑛 = 𝛿 (𝑆 𝑛 ), we have:

𝛿 (𝑆 𝑛+1 ) = 𝛿 (𝑅 ⊎ 𝜑 (𝑆 𝑛 )) = 𝛿 (𝛿 (𝑅) ⊎ 𝛿 (𝜑 (𝑆 𝑛 ))) (𝛿 is an aggregation function) = 𝛿 (𝛿 (𝑅) ⊎ 𝛿 (𝜑 (𝛿 (𝑆 𝑛 ))) (𝛿 • 𝜑 = 𝛿 • 𝜑 • 𝛿) = 𝛿 (𝛿 (𝑅) ⊎ 𝛿 (𝜑 (𝑆 ′ 𝑛 ))) (induction hypothesis) = 𝛿 (𝑅 ⊎ 𝜑 (𝑆 ′ 𝑛 )) = 𝑆 ′ 𝑛+1 (𝛿 is an aggregation function)
Then the result propagates to the fixpoint since we assumed that 𝜇 (𝑅, 𝜑) ≠ 𝜔. □

Applying this optimization on the expression of the SP example (Sec.2.8) means that only the shortest paths are kept at each iteration of the fixpoint so we avoid computing all possible paths before keeping only the shortest ones at the end. The requirement of compatibility of 𝛿 with 𝜑 means that the application of 𝛿 first before the 𝜑 operation does not impact the result compared to when it is applied once at the end. For instance, if we are computing the shortest paths between 𝑎 and 𝑏, we look for all paths between 𝑎 and 𝑐, append them to paths from 𝑐 to 𝑏, then keep the shortest ones.

Alternatively, we could start by keeping only the shortest paths between 𝑎 and 𝑐 then append them to paths between 𝑐 and 𝑏 without altering results. At present, we do not have a method for statically checking this constraint. So, in practice, we require an annotation from the programmer on the aggregation operations that verify the necessary constraints.

However, we can list common known aggregation functions : reduceByKey(𝑓 , •), filters (see Def. 3.1), mainly.

Distribution of the fixpoint operations (P dist )

As explained in 2.9.1, the fixpoint operation is computed locally using a loop (defined by R 𝑖𝑛𝑖𝑡 , R 𝑙𝑜𝑜𝑝 and R 𝑠𝑡𝑜𝑝 ). To evaluate the fixpoint in a distributed setting, we could simply write a loop that distributes the computation of the operation that is performed at each iteration (𝑆 ⊗ 𝛿 𝜑 (𝑅)) among the workers. We call this execution plan P 1 . P 1 performs 𝛿 at each iteration on the whole intermediary distributed bag 𝑆 to compute 𝑆 ⊗ 𝛿 𝜑 (𝑅), which in most cases (i.

e. unless 𝛿 is the identity function) requires synchronisation and data transfer between workers at each iteration. In the TC example (Sec. 2.8), this plan amounts to appending, at each iteration, all currently found paths from all partitions with the graph edges R.

Alternatively

𝜇 𝛿 (𝑅 1 |𝑅 2 |..., 𝜑) 𝑅𝜇 𝛿 (𝛿 (𝑅 1 ), 𝜑; 𝛿 (𝑅 1 )) ⊗ 𝑛𝑙 𝛿 𝑅𝜇 𝛿 (𝛿 (𝑅 2 ), 𝜑; 𝛿 (𝑅 2 )) ⊗ 𝑛𝑙 𝛿 ...
Then each 𝑅𝜇 𝛿 (𝛿 (𝑅 𝑖 ), 𝜑; 𝛿 (𝑅 𝑖 )) is going to be evaluated by R 𝑙𝑜𝑜𝑝 and R 𝑠𝑡𝑜𝑝 as they are fixpoints on local bags. This execution plan, that we name P 2 , will avoid doing non-local set unions or aggregations between all partitions at each iteration of the fixpoint. Instead, the fixpoint is executed locally on each partition on a part of the input, after which the aggregate ⊗ 𝑛𝑙 𝛿 is computed once to gather results. In our example, this amounts to computing, on each partition 𝑖, all paths in the graph starting from nodes in 𝑅 𝑖 ; the result is then the union of all obtained paths. This reduction in data transfers can lead to a significant improvement of performance, since the size of data transfers over the network is a determining factor of the performance of distributed applications.

The optimization rule P dist uses the plan P 2 instead of P 1 for evaluating fixpoints.

3.4.1 Avoiding ∪ 𝑛𝑙 in P 2 . In the common case where 𝛿 is distinct, P 2 can be optimized further by repartitioning the data in the cluster in such a way that every result of the fixpoint appears in one partition only. When that is the case, it is sufficient to perform a bag union rather than a set union that removes duplicates from across the cluster. If we know that there is a part in the input that does not get modified by 𝜑, we can repartition the data on this part of the input (no two different partitions have the same value for this part), so the result of the fixpoint is also going to be repartitioned in the same way. We formalize this optimization in the following way:

Let 𝜋 a pattern that matches the input of 𝜑 and 𝑎 a pattern variable in 𝜋. We consider the following propositions:

(𝐶 𝑎 ) : ∀𝑟 ∈ 𝑅 ∀𝑠 ∈ 𝜑 ({{𝑟 }}) 𝜋 𝑎 (𝑟 ) = 𝜋 𝑎 (𝑠) (𝑃 𝑎 ) : ∀𝑖 ≠ 𝑗 ∀𝑥 ∈ 𝑅 𝑖 ∀𝑦 ∈ 𝑅 𝑗 𝜋 𝑎 (𝑥) ≠ 𝜋 𝑎 (𝑦) Lemma 3.5.
If there exists a pattern variable 𝑎 that verifies (𝐶 𝑎 ), then:

𝑃 𝑎 ⇒ ∀𝑖 ≠ 𝑗 𝜇 (𝑅 𝑖 , 𝜑) ∩ 𝜇 (𝑅 𝑗 , 𝜑) = ∅
Proof. Let us suppose there exist a pattern variable 𝑎 for which (𝐶 𝑎 ) is verified, and let us suppose (𝑃 𝑎 ). Let 𝑅 𝑖 and 𝑅 𝑗 partitions of 𝑅 such that 𝑖 ≠ 𝑗. (𝐶 𝑎 ) implies ∀𝑠 ∈ 𝜇 (𝑅, 𝜑) ∃𝑟 ∈ 𝑅 𝜋 𝑎 (𝑟 ) = 𝜋 𝑎 (𝑠) because of Lemma 3.2. Which means that for any 𝑥 ∈ 𝜇 (𝑅 𝑖 , 𝜑) and 𝑦 ∈ 𝜇 (𝑅 𝑗 , 𝜑), ∃𝑟 𝑖 ∈ 𝑅 𝑖 ∃𝑟 𝑗 ∈ 𝑅 𝑗 𝜋 𝑎 (𝑟 𝑖 ) = 𝜋 𝑎 (𝑥) and 𝜋 𝑎 (𝑟 𝑗 ) = 𝜋 𝑎 (𝑦). We have

𝜋 𝑎 (𝑟 𝑖 ) ≠ 𝜋 𝑎 (𝑟 𝑗 ) because (𝑃 𝑎 ), so 𝑥 ≠ 𝑦. Hence ∀𝑖 ≠ 𝑗 𝜇 (𝑅 𝑖 , 𝜑) ∩ 𝜇 (𝑅 𝑗 , 𝜑) = ∅. □ This means that 𝜇 (𝑅 1 ∪ 𝜑 (𝑅 1 ), 𝜑) ∪ 𝑛𝑙 𝜇 (𝑅 2 ∪ 𝜑 (𝑅 2 ), 𝜑) ∪ 𝑛𝑙 ... = 𝜇 (𝑅 1 ∪ 𝜑 (𝑅 1 ), 𝜑) | 𝜇 (𝑅 2 ∪ 𝜑 (𝑅 2 ), 𝜑) |...
The pattern variable 𝑎 that verifies (𝐶 𝑎 ) can be found by using the technique explained in Section 3.1.1. We explore every node 𝑛 in 𝐶 (Bag[C] is the input type of 𝜑) starting from the root of 𝐶 and we build 𝐶 (𝛼) = replace 𝛼 (path(𝑛, 𝐶), 𝐶)

until we find a node that verifies 𝜑 : 𝐶 (𝛼) → 𝐶 (𝛼).

If such 𝑎 is found, we repartition the data according to (𝑃 𝑎 ) by using the API provided by the big data platform on which the code is executed, given that 𝑎 can be extracted from the input data using pattern matching.

Effects of the rules on performance

In this section, we discuss the impacts of the rules and the conditions under which they produce terms that are more efficient in practice. The verification of these conditions is outside of the scope of this paper. Techniques that estimate the size of algebraic expressions such as those found in [START_REF] Lawal | A Cost Estimation Technique for Recursive Relational Algebra[END_REF] can be used to perform such verifications.

PF effects.

Rule PF is a logical optimization rule in the sense that the term it produces is always more efficient than the initial term. Indeed, a filter reduces the size of intermediate data. The application of PF thus reduces data transfers. Operators are also executed faster on smaller data. The application of PF can thus only improve performance. PJ can improve local complexity. The reason is that the additional cogroup is evaluated only once, whereas the pushed filter makes 𝑅 (the first argument of the fixpoint 𝜇 (𝑅, 𝜑)) smaller. Therefore, in general, each iteration of the fixpoint is executed faster as it deals with increasingly less data (each value removed from the initial bag would have generated more additional values with each iteration). The final join with the result of the fixpoint also executes faster because its size is reduced prior to the join. In the worst case (the filter does not remove any result), the additional cogroup does not change the worst case complexity of the computation (join and cogroup have the same worst case complexity 𝑂 (𝑛 2 )).

To analyse the impact of the rule on non-local data transfers, we need to estimate and compare the size of the transfers incurred by the terms: 𝑗𝑜𝑖𝑛(𝐴, 𝜇 (𝑅, 𝜑)) and 𝑗𝑜𝑖𝑛(𝐴, 𝜇 (𝐹 𝐴 (𝑅), 𝜑)) (obtained after applying the rule). As mentioned in Section 2.9.2, all our algebraic operators apart from flatmap trigger non-local transfers. Let 𝑠𝑖𝑧𝑒 (𝑡) be the size of the result obtained by evaluating 𝑡, 𝑠𝑖𝑧𝑒 𝑡 (𝑡) the size of transfers incurred by the evaluation of 𝑡, and 𝑁 the number of partitions (parallel tasks) in the cluster. We consider the following:

• Repartitioning a dataset 𝐴 by key requires all 𝐴 to be tranferred across the network (each element of 𝐴 has to be in the partition corresponding to its key). This means that size 𝑡 (groupby(𝑡)) = 𝑠𝑖𝑧𝑒 (𝑡), and size 𝑡 (cogroup(𝑡 1 , 𝑡 2 )) = size 𝑡 (join(𝑡 1 , 𝑡 2 )) = size(𝑡 1 ) + size(𝑡 2 ). • size 𝑡 (distinct(𝐴)) = 𝑁 ×size(𝐴) because all 𝐴 has to be seen by each partition so that duplicates can be removed globally. This means that size 𝑡 (𝜇 (𝑅, 𝜑)) = 𝑁 × size(𝜇 (𝑅, 𝜑)).

Let 𝑆 1 = size 𝑡 (join(𝐴, 𝜇 (𝑅, 𝜑))) and 𝑆 2 = size 𝑡 (join(𝐴, 𝜇 (𝐹 𝐴 (𝑅), 𝜑))). So we have: 𝑆 1 = size(𝐴) + (𝑁 + 1) × size(𝜇 (𝑅, 𝜑)), here the result of the fixpoint is sent twice: the first time to compute the fixpoint and the second time to compute the join between A and the fixpoint result. 𝑆 2 = 2 × size(𝐴) + (𝑁 + 1) × size(𝜇 (𝐹 𝐴 (𝑅), 𝜑)) + size(𝑅), here 𝐹 𝐴 (𝑅) requires making a cogroup between 𝐴 and 𝑅 which incurs an additional transfer of their sizes. On the other hand, only a filtered fixpoint result is sent.

In order to determine if PJ improves data transfers we need to compare 𝑆 1 and 𝑆 2 , which amounts to comparing the following quantities: (𝑁 + 1) × size(𝜇 (𝑅, 𝜑)) and (𝑁 + 1) × size(𝜇 (𝐹 𝐴 (𝑅), 𝜑)) + size(𝐴) + size(𝑅). In other words, PJ improves data transfers when the data removed from the fixpoint result (by pushing the filter into it) makes up for the sizes of 𝐴 and 𝑅 that are transferred to compute the additional cogroup. This is likely to be the case since the data obtained at each iteration of the fixpoint (including R) is filtered.

3.5.3 PA effects. The PA rule applies the aggregation function 𝛿 on the fixpoint's intermediate results instead of once at the end. Whenever 𝛿 reduces the size of these results, the fixpoint operation deals with less data at each iteration (which also generally reduces the number of iterations). For example, if we are computing the shortest paths, applying the rule would mean that we are only going to deal with the shortest paths at each step instead of the entirety of possible paths. This can also lead to the termination of the program in case the graph has cycles (note that the programs are semanticaly equivalent but the evaluation of the first does not terminate). Additionally, when P dist is applied, PA can only reduce the size of the data transferred across the network because 𝛿 is executed locally and reduces the sizes of the local fixpoints. The efficiency of the two plans that distribute the fixpoint depends on two aspects. First, for a term 𝜇 (𝑅, 𝜑) to be evaluated on a plateform like Spark, the collections referenced in 𝜑 have to be available locally in each worker so that it can compute the fixpoint locally. For instance, if 𝜑 = 𝑗𝑜𝑖𝑛(𝑋, 𝑆) then 𝑆 and 𝑋 (at each iteration) are both referenced by 𝜑. This is a limitation of plan P 2 : when those datasets become too large to be handled by one worker, P 1 is more appropriate. Second, a factor that determines the efficiency of P 2 (and impacts the size of the iteration results 𝑋 ) is the number of parallel tasks that execute the program. In Spark, this corresponds to the number of partitions. Increasing the number of partitions increases the parallelization and reduces the load on each worker because the local fixpoints start from smaller constant parts. For a term 𝜇 (𝑅, 𝜑), it is thus possible to regulate the load on the workers by splitting 𝑅 into smaller 𝑅 𝑖 , resulting in smaller tasks on more partitions. The ideal number of partitions is the smallest one that makes all workers busy for the same time period, and for which the size of the task remains suitable for the capacity of each worker. Increasing the number of partitions further would only increase the overhead of scheduling. Thus, estimation of an appropriate number of partitions for P 2 would ideally be based on an estimated size of the constant part, the size of intermediate data produced by the fixpoint and the workers memory capacity. In the experiments we present below, we use a simple heuristic to determine the number of partitions: 4 times the total number of cores of the cluster.

EXPERIMENTAL RESULTS

Methodology. We experiment the 𝜇-monoids approach in the context of the Spark platform [START_REF] Zaharia | Apache Spark: a unified engine for big data processing[END_REF].

We evaluate Spark programs generated from optimized 𝜇-monoids expressions, and compare their performance with the state-of-the-art implementations Emma [START_REF] Alexandrov | Implicit Parallelism through Deep Language Embedding[END_REF] and DIQL [START_REF] Fegaras | Compile-Time Code Generation for Embedded Data-Intensive Query Languages[END_REF], which are Domain Specific Languages (more detail about them in Section 5). The authors of Emma showed that their approach outperforms earlier works in [START_REF] Alexandrov | Implicit Parallelism through Deep Language Embedding[END_REF]. DIQL is a DSL built on monoid algebra (of which the 𝜇-monoids algebra is an extension). Comparing against DIQL shows the interest of having a first-class fixpoint operator in the monoid algebra. Algorithms. The algorithms considered in these experiments are: TC, SP, Flights, Path Planning, and Movie Recommendations presented in the examples (Section 2.8). In addition, we evaluate two variants of TC and SP: TC filter and SP filter, where we compute the paths starting from a subset of 2000 nodes randomly chosen in the input graph.

Systems. 𝜇-monoids is evaluated against other systems on the algorithms mentioned above. 𝜇-monoids programs are generated from the 𝜇-monoids terms expressing these algorithms (see Examples Section 2.8) and by systematically applying the rules PF, PJ, PA, P dist (of Section 3). We evaluate these programs by comparing their execution times against the following programs:

• DIQL: The algorithms have been expressed using DIQL [START_REF] Fegaras | Compile-Time Code Generation for Embedded Data-Intensive Query Languages[END_REF] queries. In particular, the fixpoint operation is expressed in terms of the more generic repeat operator of the DIQL language. We have written the queries in such a way that they compute the fixpoint more efficiently using the algorithm mentioned in 2.9.1. The DIQL system is available at [? ].

• Emma: We used the example provided by Emma authors [START_REF] Volker | Emma is a quotation-based Scala DSL for scalable data analysis[END_REF] to compute the TC queries, and we wrote modified versions to compute the SP and the path planning examples. The Emma system is available at [START_REF] Volker | Emma is a quotation-based Scala DSL for scalable data analysis[END_REF].

• 𝜇-monoids-no-PA: 𝜇-monoids without the application of PA to assess the impact of the PA rule.

• 𝜇-monoids-no-P dist : 𝜇-monoids without the application of P dist to assess the impact of the P dist rule.

All these systems run on the same experimental setup presented above and on the same Spark plateform.

We have also written the DIQL queries in such a way they apply PF. Such a pre-filtering was not possible for Emma because the programs perform a non linear fixpoint. Trying to write a linear version leads to an exception in the execution. We were not able to write an Emma program that computes movie recommendations. Iterating over a users own movies leads to an exception.

Datasets. We use two kinds of datasets:

• Real world graphs of different sizes, presented in Table 1, including a knowledge graph (the Yago [START_REF]YAGO: A high-quality knowledge base[END_REF] dataset 6 ), a social network graph (Facebook), and a scientific collaborations network (DBLP) taken from [START_REF] Leskovec | SNAP: Stanford Large Network Dataset Collection[END_REF].

• Synthetic graphs shown in Table 1, generated using the Erdos Renyi algorithm that, given an integer 𝑛 and a probability 𝑝, generates a graph of 𝑛 vertices in which two vertices are connected by an edge with a probability 𝑝. rnd_n_p denotes such a synthetic graph, whereas rnd_n_p_W denotes a rnd_n_p graph with edges weighted randomly (between 0 and 5).

Other synthetic graphs are:

-flight_n_p: where edges are taken from rnd_n_p with random depart and arrival times and duration assigned to them. -c_n_p: serialized object RDD files representing paths between cities. It is also generated from rnd_n_p, each city has been assigned up to 10 random landmarks. -u_n: serialized object RDD files of 𝑛 users, each assigned up to 15 random movies. We observe that the programs generated by 𝜇-monoids almost systematically outperform the other program versions and always outperform the DIQL and Emma systems.

Dataset

Comparison between 𝜇-monoids and 𝜇-monoids-no-P dist shows the impact of the P dist rule. It can be noticed that the speedup achieved by this rule increases with the data size. The only case where 𝜇-monoids is slower than 𝜇-monoidsno-P dist is on the 𝑟𝑛𝑑_10𝑘_0.01 dataset in Fig. 7. This is due to the graph topology as this dataset is the densest graph tested. This means that the size of the intermediate results generated during the fixpoint computation is large, which puts more strain on the parallel tasks. The fixed number of parallel tasks (Sec. 3.5) is too small. We observe in practice that, in this case, increasing the number of partitions improves the performance of the system.

In the SP, SP filter and path planning programs, both P dist and PA are applied in 𝜇-monoids. We notice that the speedup of 𝜇-monoids in comparison to DIQL and Emma is even more important in these cases. Comparison between 𝜇-monoids and 𝜇-monoids-no-PA shows the impact of the PA rule alone. It can also be observed that applying PA and not P dist (𝜇-monoids-no-P dist ) can be faster than applying P dist and not PA (𝜇-monoids-no-PA) and the other way round depending on the cases. It is the combination of these two rules that leads to the best performances.

This experimental comparison shows the benefit of the plan that distributes the fixpoint. It also highlights the benefits of the approach that synthesises code: generating programs that are not natural for a programmer to write, like the distributed loop to compute the fixpoint.

RELATED WORKS

The idea of using an intermediate representation (or an algebra) for representing user queries and performing automatic optimization originates from the work of Codd [START_REF] Edgar F Codd | A relational model of data for large shared data banks[END_REF]. He proposed the idea of a separation between the internal representation (physical storage) from the logical representation of data. The idea is to offer a level of abstraction to represent data and operate upon it via a universal language which is independent from implementation details and possible changes to how data is physically stored and retrieved. This insight led to the relational algebra being widely adopted by database systems and extensively studied in database research. It also led to the standard SQL language.

SQL is a Domain Specific Language (DSL) that is called from within a general purpose language (also called the host language). It gets translated to a relational algebra term that gets optimized then translated to a physical execution plan.

Work in [START_REF] Gábor | Handling Iterations in Distributed Dataflow Systems[END_REF] surveys state of the art approaches for handling iterations in distributed systems and classifies them to different categories. Among these categories are the relational algebra approach mentioned earlier, and the functional approach that offer higher-order functions for specifying control flow (such as loops). The present work belongs to this latter category. In this section we review related literature along two lines of work: works that are based on the relational model, and works that are based on more generic data models.

Works based on the relational formalism

The relational model is based on n-ary relations to represent entities and relationships between them. An n-ary relation is a set of rows. Each row consists of a tuple of n records of atomic types. To operate on these relations, [START_REF] Edgar F Codd | A relational model of data for large shared data banks[END_REF] proposed the relational algebra. Relational algebra offers operations on relations such as projection, selection, and join, as well as a number of rewrite rules that aim to optimize expressions regardless of their initial shape. Another prevalent formalism based on the relational model is Datalog [START_REF] Maier | Datalog: Concepts, History, and Outlook[END_REF]. A Datalog program is composed of rules that infer new facts from previously known facts. Facts are expressed as predicates depending on a fixed number of variables. They can thus be seen as relations. Optimization techniques such as Magic Sets [START_REF] Bancilhon | Magic Sets and Other Strange Ways to Implement Logic Programs (Extended Abstract)[END_REF][START_REF] Saccà | On the Implementation of a Simple Class of Logic Queries for Databases[END_REF] or Demand transformation [START_REF] Tuncay | More efficient datalog queries: subsumptive tabling beats magic sets[END_REF] are proposed to optimize Datalog programs.

Regarding the ability to express recursion, a number of formalisms that extend RA with a recursive operator have been proposed [START_REF] Chlyah | Alpha: an extension of relational algebra to express a class of recursive queries[END_REF][START_REF] Aho | Universality of Data Retrieval Languages[END_REF]. The algebra proposed in [START_REF] Jachiet | On the Optimization of Recursive Relational Queries: Application to Graph Queries[END_REF] provides more optimizations of recursion than previous works on RA and recursive Datalog. However, it is limited to the centralized setting and to relational algebra. Regarding distribution, the Spark SQL [START_REF] Armbrust | Spark SQL: Relational Data Processing in Spark[END_REF] library enables the user to write SQL queries and process relational data using datasets or dataframes. Queries are optimized using the Spark Catalyst engine and executed in a distributed way on the Spark platform. BigDatalog [START_REF] Shkapsky | Big Data Analytics with Datalog Queries on Spark[END_REF] is a system that studies the distribution of Datalog programs on Spark. Compared to previous distributed Datalog systems such as Socialite [START_REF] Seo | Distributed Socialite: A Datalog-Based Language for Large-Scale Graph Analysis[END_REF] and Myria [START_REF] Wang | Asynchronous and Fault-Tolerant Recursive Datalog Evaluation in Shared-Nothing Engines[END_REF], it achieves better performances. The BigDatalog system uses the Datalog GPS technique [START_REF] Seib | Parallelizing Datalog Programs by Generalized Pivoting[END_REF] that analyses Datalog rules to identify decomposable Datalog programs and determine how to distribute data and computations. These ideas are tied to Datalog and are not applicable to other formalisms. In contrast, the present work proposes a new distribution method designed for a more generic algebra.

Works based on more generic formalisms

In the relational model, data consists of relations that are sets of tuples of atomic values. A more generic data model are collections of arbitrary homogeneous types. In Sec. 2.1.1 we discussed the 3 types of data structures which we call collections: lists, bags, and sets. They are, along with other data structures, part of what is called the Boom hierarchy of types [START_REF] Bunkenburg | The Boom Hierarchy[END_REF]. The author of this paper presents data structures of that hierarchy as free algebras. A data structure value can either be empty [ ], a singleton [𝑎], or a combination of two values using a binary operator 𝑐1 ++ 𝑐2. ++ can obey to a combination of four algebraic laws: unit (it has a unit element), associativity, commutativity, and idempotence.

Different combinations of laws lead to different types of structures. [START_REF] Bunkenburg | The Boom Hierarchy[END_REF] defines 16 types of data structures for all possible combinations of these laws. For instance, tree is a data structure where only the first law is satisfied: it is not associative (and therefore not a monoid). Lists are the data structure from the hierarchy obeying only unit and associativity. Bags are obtained when we add commutativity, and sets when we add idempotence. These three structures, which are monoids, are called collection monoids in the work of [START_REF] Fegaras | An algebra for distributed Big Data analytics[END_REF]. It is on this basic notion this work builds the monoid algebra.

Collections can also be seen as a particular case of Algebraic Data Types which constitutes the basic notion of the Emma language approach [START_REF] Alexandrov | Representations and Optimizations for Embedded Parallel Dataflow Languages[END_REF]. Both of these approaches propose an algebra for distributed collections. We review these works in more detail below.

It was shown in [START_REF] Buneman | Principles of Programming with Complex Objects and Collection Types[END_REF] that monads can be used to generalize nested relational algebra to different collection types and complex data. Wadler also explored and developed monad comprehensions [START_REF] Wadler | Comprehending Monads[END_REF] and ringad comprehensions [START_REF] Gibbons | Comprehending Ringads -For Phil Wadler, on the Occasion of his 60th Birthday[END_REF],

inspired by the early works on list comprehensions [START_REF] Simon | The Implementation of Functional Programming Languages[END_REF][START_REF] Turner | Recursion Equations as a Programming Language[END_REF]. These ideas were used in LINQ [START_REF] Meijer | LINQ: reconciling object, relations and XML in the .NET framework[END_REF], Ferry [START_REF] Grust | FERRY: Database-Supported Program Execution[END_REF], and Emma [START_REF] Alexandrov | Implicit Parallelism through Deep Language Embedding[END_REF] which are comprehensions-based programming languages. To be evaluated, LINQ and Ferry queries get translated into an intermediate form that can be executed on relational database systems supporting SQL. As they target relational database systems, the set of host language expressions that can be used in query clauses like selection and projection is restricted. In addition, they do not analyse comprehensions to make optimizations. Emma [START_REF] Alexandrov | Implicit Parallelism through Deep Language Embedding[END_REF] is a comprehensions-based language similar in spirit, but which rather targets JVM-based parallel dataflow engines (such as Spark and Flink).

DryadLinq [START_REF] Yu | DryadLINQ: A System for General-Purpose Distributed Data-Parallel Computing Using a High-Level Language[END_REF] proposes a system that distributes LINQ queries on the Dryad platform [START_REF] Isard | Dryad: Distributed Data-Parallel Programs from Sequential Building Blocks[END_REF]. In these works recursion is expressed using host language loops that are not modelled in the algebra, hence with no optimisation provided. The idea of using monoids and monoid homomorphisms for modeling computations with data collections originates from the works found in [START_REF] Tannen | Structural Recursion as a Query Language[END_REF][START_REF] Tannen | Data Structures and Data Types for Object-Oriented Databases[END_REF]. Fegaras proposed a monoid comprehension calculus [START_REF] Fegaras | Optimizing object queries using an effective calculus[END_REF] which later evolved in the monoid algebra presented in [START_REF] Fegaras | An algebra for distributed Big Data analytics[END_REF]. It proposes an algebra based on monoid homomorphisms therefore with parallelism at its core: a homomorphic operation 𝐻 on a collection is defined as the application of 𝐻 on each subpart of the collection, results are then gathered using an associative operator. Distributed collections are modelled using the union representation of bags, and collection elements can be of any type defined in the host language. The algebra allows for defining second order operators such as flatmap and reduce that can take a UDF written in the host language as an argument. The present work further builds on this approach and proposes a generic criteria (Sec. 3.1.1) using the host language type checking system that examines those UDFs in order to determine whether the PF optimization can take place. The monoid algebra [START_REF] Fegaras | An algebra for distributed Big Data analytics[END_REF] has a repeat operator, however no optimization is provided for this operator. The authors of [START_REF] Fegaras | Compile-Time Code Generation for Embedded Data-Intensive Query Languages[END_REF] designed DIQL (a DSL that translates to the monoid algebra). Using reflection of the host language (Scala in this work) and quotations, queries of this DSL can be compiled and type checked seamlessly with the rest of the host language code.

In fact, Emma uses the same approach as well. This approach offers more optimization opportunities than approaches like the Spark and Flink API. As argued in [START_REF] Alexandrov | Representations and Optimizations for Embedded Parallel Dataflow Languages[END_REF], even though these APIs offer a DSL that is well integrated with the host language and allow for expressing general purpose computations, they suffer from the difficulty of automatically optimizing programs. This is due to the limited program context available in the intermediate representation of the DSLs. For instance, arguments to second order operations are treated as black box functions which means that they cannot be analyzed and transformed to make automatic optimizations.

CONCLUSION

We propose to extend the monoid algebra with a fixpoint operator that models recursion. The extended 𝜇-monoids algebra is suitable for modeling recursive computations with distributed data collections such as the ones found in big data frameworks. The major interest of the "𝜇" fixpoint operator is that, under prerequisites that are often met in practice, it can be considered as a monoid homomorphism and thus can be evaluated by parallel loops with one final merge rather than by a global loop requiring network overhead after each iteration.

We also propose rewriting rules for optimizing fixpoint terms: we show when and how filters can be pushed into fixpoints. In particular, we find a sufficient condition on the repeatedly evaluated term (𝜑) regardless of its shape, and we present a method using polymorphic types and a type system such as Scala's to check whether this condition holds.

We also propose a rule to prefilter a fixpoint before a join. The third rule allows for pushing aggregation functions inside a fixpoint.

Experiments suggest that: (i) Spark programs generated by the systematic application of these optimizations can be radically different from -and less intuitive -than the input ones written by the programmer; (ii) generated programs can be significantly more efficient. This illustrates the interest of developing optimizing compilers for programming with big data frameworks.

  Authors' addresses: Sarah Chlyah, Nils Gesbert, Pierre Genevès, Nabil Layaïda, Tyrex team, INRIA, 655 avenue de l'Europe, 38330 Montbonnot, France. 1 def shortestPaths(R:RDD[(Int,Int,Int)]) = { var ret = R var X: RDD[(Int, Int, Int)] = R var new_cnt = ret.count() var cnt = new_cnt do { cnt = new_cnt X = X.map({case (x,y,l1) => (y,(x,l1)) })

Fig. 1 .Fig. 2 .

 12 Fig. 1. Shortest paths program.

9 .

 9 Let 𝛿 (Bag[𝑡]) be the image of 𝛿. Then (𝛿 (Bag[𝑡]), ⊗ 𝛿 , {{}}) is a monoid, noted 𝑀 𝛿 , isomorphic to the quotient monoid Bag[𝑡]/∼ 𝛿 , and 𝛿 is a monoid homomorphism: Bag[𝑡] → 𝑀 𝛿 . Proof. Since 𝛿 is idempotent, we can consider 𝛿 (𝑎) the canonical representant of the equivalence class of 𝑎; then we have an isomorphism between the equivalence classes (the monoid Bag[𝑡]/∼ 𝛿 ) and their canonical representants (the monoid 𝑀 𝛿 ). □ Example 2.10. The function distinct : Bag[𝑡] → Bag[𝑡] which removes duplicates from a bag is an aggregation function; ∼ distinct is the relation ∼ idem ; ⊗ distinct is distinct union of bags ∪; and 𝑀 distinct is the monoid of bags with no duplicates, isomorphic to FSet[𝑡].

Finally, in order

  to work up to equivalence relations, we need the notion of compatibility between an homomorphism 𝜑 from bags to bags and an aggregation function 𝛿: Lemma 2.11. Let 𝜑 : Bag[𝑡] → Bag[𝑡] be a monoid homomorphism and 𝛿 : Bag[𝑡] → Bag[𝑡] an aggregation function. The three following properties are equivalent:

Definition 2 .

 2 13 (𝜇 operator). Let 𝜑 : Bag[𝑡] → Bag[𝑡] be a monoid homomorphism, 𝛿 : Bag[𝑡] → Bag[𝑡] an aggregation function, and 𝑅 : Bag[𝑡] a dataset. Assume 𝜑 and 𝛿 are compatible (Def. 2.12). The operation 𝜇 𝛿 (𝑅, 𝜑)

  h(𝑋 ) ::= 𝑋 | flmap(𝑓 , h(𝑋 )) 𝑋 does not appear in 𝑓 | join(h(𝑋 ), 𝐴) 𝑋 does not appear in 𝐴 | join(𝐴, h(𝑋 )) 𝑋 does not appear in 𝐴 This set of terms is in fact quite general: indeed, we know from algebra that homomorphisms from Bag[𝑡] to Bag[𝑡] are in one-to-one correspondance with functions from 𝑡 to Bag[𝑡], via the general flatmap operation 3 . In our case, flatmap has a restriction relative to distributed bags, which is why we also have join; so the only monoid homomorphisms which cannot be written in the form 𝜆 ⟨𝑋 → h(𝑋 )⟩ are functions which manipulate distributed bags in a way which flmap(𝑓 , 𝐴) = 𝑎∈𝐴 𝑓 (𝑎) reduce( ⊗, 𝑒 ⊗ , 𝐴) = 𝑎∈𝐴 𝑓 (𝑎) reduceByKey( ⊗, 𝐴) = {{ (𝑘, (𝑘,𝑣) ∈𝐴 𝑣) | 𝑘 ∈ keys(𝐴) }} cogroup(𝐴, 𝐵) = {{ (𝑘, ( {{𝑣 | (𝑘, 𝑣) ∈ 𝐴 }}, {{𝑤 | (𝑘, 𝑤) ∈ 𝐵 }})) | 𝑘 ∈ keys(𝐴) ∪ keys(𝐵) }} join(𝐴, 𝐵) = {{ (𝑘, (𝑣, 𝑤)) | (𝑘, 𝑣) ∈ 𝐴 ∧ (𝑘, 𝑤) ∈ 𝐵 }} 𝜇 𝛿 (𝑅, 𝜑) = 𝑆 𝑁 if there exists N such that 𝑆 𝑁 +1 = 𝑆 𝑁 𝜔 otherwise where 𝑆 𝑛 = 𝛿 𝑘 ≤𝑛 (𝛿 • 𝜑) 𝑘 (𝛿 (𝑅))

  𝜇. Recall that 𝛿 and 𝜑 being compatible means that 𝛿 • 𝜑 is a monoid homomorphism: 𝑀 𝛿 → 𝑀 𝛿 . Thus we have: (𝛿 • 𝜑)( 𝛿 𝑘 ≤𝑛 (𝛿 • 𝜑) 𝑘 (𝛿 (𝑅))) = 𝛿 𝑘 ≤𝑛 (𝛿 • 𝜑) 𝑘+1 (𝛿 (𝑅)). The only term missing to obtain 𝑆 𝑛+1 on the right is (𝛿 • 𝜑) 0 (𝛿 (𝑅)), i. e. 𝛿 (𝑅). So we have, for any 𝑛: 𝑆 𝑛+1 = 𝛿 (𝑅) ⊗ 𝛿 (𝛿 • 𝜑 (𝑆 𝑛 )). In other words, if we use the definitions to 'clean up' superfluous 𝛿s: 𝑆 𝑛+1 = Ψ(𝑆 𝑛 ) with Ψ = 𝑋 ↦ → 𝛿 (𝑅 ⊎ 𝜑 (𝑋 )).

  let reverse_edges = 𝜆 ⟨(𝑎, 𝑏) → {{ (𝑏, 𝑎) }}⟩ in let drop_mid = 𝜆 ⟨(𝑚𝑖𝑑, (𝑠𝑟𝑐, 𝑑𝑒𝑠𝑡 )) → {{ (𝑠𝑟𝑐, 𝑑𝑒𝑠𝑡 ) }}⟩ in 𝜇 (𝑅, 𝜆 ⟨𝑋 → flmap(drop_mid, join(flmap(reverse_edges, 𝑋 ), 𝑅)) ⟩)

  let key_dest = 𝜆 ⟨( (𝑠𝑟𝑐, 𝑑𝑒𝑠𝑡 ), 𝑤) → {{ (𝑑𝑠𝑡, (𝑠𝑟𝑐, 𝑤)) }}⟩ in let key_src = 𝜆 ⟨( (𝑠𝑟𝑐, 𝑑𝑒𝑠𝑡 ), 𝑤) → {{ (𝑠𝑟𝑐, (𝑑𝑒𝑠𝑡, 𝑤)) }}⟩ in let combine = 𝜆 ⟨(𝑚𝑖𝑑, ( (𝑠𝑟𝑐, 𝑤 1 ), (𝑑𝑒𝑠𝑡, 𝑤 2 ))) → {{ ( (𝑠𝑟𝑐, 𝑑𝑒𝑠𝑡 ), 𝑤 1 + 𝑤 2 ) }}⟩ in let all_paths = 𝜇 (𝑅, 𝜆 ⟨𝑋 → flmap(combine, join(flmap(key_dest, 𝑋 ), flmap(key_src, 𝑅))) ⟩) in reduceByKey(min, all_paths)

  Flights. let corr_possible = 𝜆 ⟨(corr, (Flight(dtime1, atime1, dep1, dest1, dur1), Flight(dtime2, atime2, dep2, dest2, dur2))) → if atime1 < dtime2 then {{Flight(dtime1, atime2, dep1, dest2, dur1 + dur2) }} else {{ }}⟩ in let key_dest = 𝜆 ⟨Flight(dtime, atime, dep, dest, dur) → {{ (dest, Flight(dtime, atime, dep, dest, dur)) }}⟩ in let key_dep = 𝜆 ⟨Flight(dtime, atime, dep, dest, dur) → {{ (dep, Flight(dtime, atime, dep, dest, dur)) }}⟩ in 𝜇 (𝑅, 𝜆 ⟨𝑋 → flmap(corr_possible, join(flmap(key_dest, 𝑋 ), flmap(key_dep, 𝑅))) ⟩) where 𝑅 is a dataset of direct flights. Flight(dtime, atime, dep, dest, dur) is a flight object with a departure time dtime, arrival time atime, departure location dep, destination dest and duration dur. At each iteration, the fixpoint expression computes new flights by joining the flights obtained at the previous iteration with the flights dataset, in such a way that two flights produce a new flight if the first flight arrives before the second flight departs, and the first flight destination airport is the second's flight departure airport. The computation stops when no more new non-direct flights can be deduced. Path planning. let paths = 𝜆 ⟨(City(𝑛 1 , 𝑙 1 ), City(𝑛 2 , 𝑙 2 )) → {{ (Path(𝑛 1 , 𝑛 2 ), 𝑙 1 ++ 𝑙 2 ) }}⟩ in let key_name_dep = 𝜆 ⟨(Path(𝑠, 𝑑), 𝑙) → {{ (𝑠, (Path(𝑠, 𝑑), 𝑙)) }}⟩ in let key_name_dest = 𝜆 ⟨(Path(𝑠, 𝑑), 𝑙) → {{ (𝑑, (Path(𝑠, 𝑑), 𝑙)) }}⟩ in let combine = 𝜆 ⟨(𝑘, ( (Path(𝑠 1 , 𝑑 1 ), 𝑙 1 ), (Path(𝑠 2 , 𝑑 2 ), 𝑙 2 ))) → {{ (Path(𝑠 1 , 𝑑 2 ), 𝑙 1 ++ 𝑙 2 ) }}⟩ in let all_paths = 𝜇 (flmap(paths, 𝑅), 𝜆 ⟨𝑋 → flmap(combine, join(flmap(key_name_dest, 𝑋 ), flmap(key_name_dep, flmap(paths, 𝑅)))) ⟩) in flmap(𝜆 ⟨(Path(𝑠, 𝑑), 𝑙) → if 𝑠 = "Paris" and 𝑑 = "Geneva" then {{ (Path(𝑠, 𝑑), 𝑙) }} else {{ }}⟩, reduceByKey(bestRated, flmap(𝜆 ⟨(𝑠, 𝑑, 𝑙) → ( (𝑠, 𝑑), 𝑙)) ⟩, all_paths)))

  1) 𝜇 (𝐹 (𝑅), 𝜑) ⊂ 𝐴: 𝐹 (𝑅) ⊂ 𝑅 ⇒ 𝜇 (𝐹 (𝑅), 𝜑) ⊂ 𝜇 (𝑅, 𝜑) (because 𝜇 (𝑅, 𝜑) = 𝜇 (𝐹 (𝑅) ⊎ 𝑅 ′ , 𝜑) = 𝜇 (𝐹 (𝑅), 𝜑) ∪ 𝜇 (𝑅 ′ , 𝜑) and 𝜇 (𝑅, 𝜑) does not contain duplicates) Let 𝑠 ∈ 𝜇 (𝐹 (𝑅), 𝜑) ∃𝑟 ∈ 𝐹 (𝑅) 𝜋 𝑎 (𝑠) = 𝜋 𝑎 (𝑟 ) ( * ) So 𝑐 (𝜋 𝑎 (𝑠)) = 𝑐 (𝜋 𝑎 (𝑟 )) = true (because 𝑟 ∈ 𝐹 (𝑅)) So 𝑠 ∈ 𝜇 (𝑅, 𝜑) and 𝑐 (𝜋 𝑎 (𝑠)) is true, then 𝑠 ∈ 𝐴 (2) 𝐴 ⊂ 𝜇 (𝐹 (𝑅), 𝜑): Let 𝑠 ∈ 𝐴. We have: 𝑠 ∈ 𝜇 (𝑅, 𝜑) and 𝑐 (𝜋 𝑎 (𝑠)) = true So ∃𝑟 ∈ 𝑅 ∃𝑛 ∈ N 𝜋 𝑎 (𝑠) = 𝜋 𝑎 (𝑟 ) (because ( * ) and 𝑠 ∈ 𝜑 (𝑛) ({{𝑟 }})) So 𝑐 (𝜋 𝑎 (𝑟 )) = 𝑐 (𝜋 𝑎 (𝑠)) = true So 𝑟 ∈ 𝐹 (𝑅), which means distinct(𝜑 (𝑛) ({{𝑟 }})) ⊂ 𝜇 (𝐹 (𝑅), 𝜑), so 𝑠 ∈ 𝜇 (𝐹 (𝑅), 𝜑).

Lemma 3 . 4 .

 34 Let 𝜑 be a monoid homomorphism: Bag[𝑡] → Bag[𝑡], 𝛿 an aggregation function: Bag[𝑡] → Bag[𝑡] compatible with 𝜑, and 𝑅 : Bag[𝑡] a dataset. Assume 𝜇 (𝑅, 𝜑) ≠ 𝜔 (i. e. the computation terminates). Then 𝛿 (𝜇 (𝑅, 𝜑)) = 𝜇 𝛿 (𝑅, 𝜑).

3. 5 . 2

 52 PJ effects. The rule PJ introduces an additional cogroup to compute the filter being pushed in the fixpoint (as detailed in Sec. 3.2). The cost of evaluating a term depends on two important aspects: the size of non-local data transfers it generates, and the local complexity of the term (i.e. the time needed for executing its local operations).

3. 5

 5 .4 P dist effects. Application of P dist can drastically decrease data transfers across the network. As explained in Sec. 3.4, plan P 2 avoids transferring intermediate results during fixpoint iterations or even entirely (if a data partioning that verifies the criteria presented in 3.4.1 exists).

  Experimental setup. Experiments have been conducted on a Spark cluster composed of 5 machines (hence using 5 workers, one on each machine, and the driver on one of them). Each machine has 128 Go and is configured to use 40 GB of RAM, 2 Intel Xeon E5-2630 v4 CPUs (2.20 GHz, 20 cores each) and 66 TB of 7200 RPM hard disk drives, running Spark 2.2.3 and Hadoop 2.8.4 inside Debian-based Docker containers.

Fig. 5 .

 5 Fig. 5. TC running times.

Fig. 6 .Fig. 7 .

 67 Fig. 6. SP running times.

Fig. 8 .Fig. 9 .

 89 Fig. 8. SP filter running times.

Fig. 10 .

 10 Fig. 10. Movie recommendations and Flights running times.

  ∼ comm be the congruence on (List[𝑡], ++, [ ]) generated by commutativity, i. e. the smallest congruence such that: ∀𝑎, 𝑏 𝑎 ++ 𝑏 ∼ comm 𝑏 ++ 𝑎. This relation relates all lists containing exactly the same elements, with the same multiplicity, in any order. So a bag can be seen as an equivalence class of lists for ∼ comm , meaning that (Bag[𝑡], ⊎, {{}}) Values and types of 𝜇-monoids. In order to enable algebraic datatypes, we assume an infinite set of constructors 𝐶 which can be applied to any number of values. We assume this set contains the special constructors True, False and Tuple for which we will define some syntactic sugar. [𝑡 𝑙 , ..., 𝑡 𝑙 ]...𝐶 𝑛 [𝑡 𝑙 , ..., 𝑡 𝑙 ] and can be deconstructed by pattern-matching.We also define product types 𝑡 1 × ••• × 𝑡 𝑛 as syntactic sugar for Tuple[𝑡 1 , ..., 𝑡 𝑛 ], i. e. a particular case of constructor

	is the quotient monoid (List[𝑡], ++, [ ])/∼ comm .
	Similarly, let ∼ idem be the congruence on bags generated by idempotence (𝑎 ⊎ 𝑎 ∼ idem 𝑎): it relates all bags
	containing the same elements, regardless of their multiplicity, and we have that (FSet[𝑡], ∪, ∅) is the quotient monoid
	(Bag[𝑡], ⊎, {{}})/∼ idem .
	In this work, we choose bags as the default base abstraction level, since there is no canonical ordering of the machines
	in the cluster; but this can be adapted to work with lists. So from now on, we consider that a dataset is a distributed
	bag. We now define the formal syntax of our data model before developing further how we can sometimes work up to
	equivalence relations if, e. g., we are in fact interested in sets and not bags.
	2.1.2 The syntax of considered data values is defined as follows:
		𝑣 ::= 𝑐		constant
		|	𝐶 (𝑣 1 , 𝑣 2 , ..., 𝑣 𝑛 ) 𝑛-ary constructor
		|	{{𝑣 1 , ..., 𝑣 𝑛 }}	bag
	As mentioned previously (Sec. 2), a constant 𝑐 can be any value from the language L (in particular any function) that is
	not explicitly defined in our syntax.
	We define the following syntax for types:
	𝑡 𝑙 ::=			local type	𝑡 ::=	type
		B		basic type		|	𝑡 𝑙
	|	𝐶 1 [𝑡 𝑙 , ..., 𝑡 𝑙 ] | | ••• | | 𝐶 𝑛 [𝑡 𝑙 , ..., 𝑡 𝑙 ]	sum type		|	Bag 𝑑 [𝑡 𝑙 ]	distributed bag type
	|	Bag 𝑙 [𝑡 𝑙 ]		local bag type	|	𝑡 → 𝑡	function type
	where B represents any arbitrary basic type (i.e., considered as a constant atomic type in our formalism).
	In sum types, all constructors have to be different and their order is irrelevant. They represent values which can
	belong to any of the case types 𝐶 1 type.
	For a given type 𝑡, we denote by Bag 𝑙 [𝑡] the type of a local bag and by Bag 𝑑 [𝑡] the type of a distributed bag of
	values of type 𝑡. Notice that we can have distributed bags of any data type 𝑡 including local bags, which allows us to
	have nested collections. We allow data distribution only at the top level though (distributed bags cannot be nested).

  𝑎𝑐, 𝑏𝑎, 𝑏𝑐, 𝑐𝑎, 𝑐𝑏, 𝑎, 𝑏, 𝑐 }} 𝑅 2 = {{𝑎𝑏𝑐, 𝑎𝑐𝑏, 𝑏𝑎𝑐, 𝑏𝑐𝑎, 𝑐𝑎𝑏, 𝑐𝑏𝑎 }} 𝑆 2 = {{𝑎𝑏𝑐, 𝑎𝑐𝑏, 𝑏𝑎𝑐, 𝑏𝑐𝑎, 𝑐𝑎𝑏, 𝑐𝑏𝑎, 𝑎𝑏, 𝑎𝑐, 𝑏𝑎, 𝑏𝑐, 𝑐𝑎, 𝑎, 𝑏, 𝑐 }}

  homomorphisms. The definition of algebraic operations as monoid homomorphism suggests that they can be evaluated in the following way: if 𝜑 is a homomorphism from Bag[𝑡] to (𝑡 ′ , 𝑒, ⊗), 𝜑 ({{𝑣 1 ...𝑣 𝑛 }}) 𝜑 ({{𝑣 1 }}) ⊗ 𝜑 ({{𝑣 2 }}) ⊗ ... ⊗ 𝜑 ({{𝑣 𝑛 }}). As monoid operators are associative, parts of an expression in the form 𝑒 1 ⊗ 𝑒 2 ⊗ ... ⊗ 𝑒 𝑛 can be evaluated in any order and in parallel. Distributed execution. We consider in a distributed setting that distributed bags are partitioned. Distributed data is noted in the following way: 𝑅 = 𝑅 1 |𝑅 2 |...|𝑅 𝑝 , meaning that R is split into 𝑝 partitions stored on 𝑝 machines. We can write a new slightly different version of the rule described above for evaluating partitioned data: • 𝜑 (𝑅 1 |𝑅 2 |...|𝑅 𝑝 ) 𝜑 (𝑅 1 ) ⊗ 𝑛𝑙 𝜑 (𝑅 2 ) ⊗ 𝑛𝑙 ... ⊗ 𝑛𝑙 𝜑 (𝑅 𝑝 ) if 𝜑 is an homomorphism from bags to (𝑀, 𝑒, ⊗), where ⊗ 𝑛𝑙 is the non-local version of ⊗. Applying this non-local operation means that data transfers are required.

	Fixpoint operator. The fixpoint operator can be evaluated as a loop, as described in Def. 2.13. We can summarise it
	with the following reduction rules, where 𝑅𝜇 represents a running 𝜇 computation and has the bag which accumulates
	the results as an additional parameter:
			𝑆 ⊗ 𝛿 𝜑 (𝑅) = 𝑆		𝑆 ⊗ 𝛿 𝜑 (𝑅) ≠ 𝑆
	R 𝑖𝑛𝑖𝑡 𝜇 𝛿 (𝑅, 𝜑)	𝑅𝜇 𝛿 (𝛿 (𝑅), 𝜑; 𝛿 (𝑅))	R 𝑠𝑡𝑜𝑝	R 𝑙𝑜𝑜𝑝
			𝑅𝜇 𝛿 (𝑅, 𝜑; 𝑆)	𝑆	𝑅𝜇 𝛿 (𝑅, 𝜑; 𝑆)	𝑅𝜇 𝛿 (𝛿 (𝜑 (𝑅)), 𝜑; 𝑆 ⊗ 𝛿 𝜑 (𝑅))
	2.9.2

• 𝜑 (𝑅 1 |𝑅 2 |...|𝑅 𝑝 ) 𝜑 (𝑅 1 )|𝜑 (𝑅 2 )|...|𝜑 (𝑅 𝑝 ) if 𝜑 is an homomorphism from bags to bags (partitioning does not have to change)

  is obtained by replacing in 𝑇 the node at path 𝑝 and its children by a node labelled 𝛼. Let us now consider 𝐶 (𝛼) = 𝑟𝑒𝑝𝑙𝑎𝑐𝑒 𝛼 (𝑝𝑎𝑡ℎ(𝑎, 𝜋), 𝐶), where Bag[𝐶] is the input type of 𝜑. Note that this path makes sense in 𝐶 because 𝐶 matches 𝜋 (see Appendix 2.6).With 𝐶 (𝛼) built this way, we have the following:

				(1)
	𝑒 : 𝐶 (𝛼) ⇒ 𝜋 𝑎 (𝑒) : 𝛼		(2)
	For example:
	Tuple	Tuple		Tuple
	a b c
			𝛼	String Int
	Tuple	String	Int
	Int Int
	𝜋	𝐶		𝐶 (𝛼)

𝑒 : 𝐶 and 𝜋 𝑎 (𝑒) : 𝛼 ⇒ 𝑒 : 𝐶 (𝛼) We show that if 𝜑 : Bag[𝐶 (𝛼)] → Bag[𝐶 (𝛼)] then the condition (C) is verified: Let 𝑟 ∈ 𝑅 and let us take 𝛼 = {{𝜋 𝑎 (𝑟 )}} which is the singleton type containing the value 𝜋 𝑎 (𝑟 ). Since 𝜋 𝑎 (𝑟 ) : 𝛼 and 𝑟 : 𝐶, we have 𝑟 : 𝐶 (𝛼) according to (1). We also have 𝜑 ({{𝑟 }}) : Bag[𝐶 (𝛼)] because {{𝑟 }} : Bag[𝐶 (𝛼)] and 𝜑 : Bag[𝐶 (𝛼)] → Bag[𝐶 (𝛼)] which means that ∀𝑠 ∈ 𝜑 ({{𝑟 }}) 𝑠 : 𝐶 (𝛼). So 𝜋 𝑎 (𝑠) : 𝛼 (according to (

  , if we use the fact that 𝜇 𝛿 (𝑅, 𝜑) is a monoid homomorphism, then we can replace R 𝑖𝑛𝑖𝑡 with the following distributed version (recall that 𝑅 1 |𝑅 2 |... denotes a distributed bag split across different partitions 𝑅 𝑖 . ⊗ 𝑛𝑙 𝛿 denotes the non-local version of ⊗ 𝛿 ):

Table 1 .

 1 Synthetic and real graphs used in experiments.

		Edges Nodes	TC size
	rnd_10k_0.001	50,119 10,000	5,718,306
	rnd_20k_0.001	199,871 20,000	81,732,096	Dataset	Edges	Nodes
	rnd_30k_0.001	450,904 30,000 255,097,974	Yago 62,643,951 42,832,856
	rnd_10k_0.005	249,791 10,000	39,113,982	Facebook	88,234	4,039
	rnd_10k_0.01	499,486 10,000	45,098,336	DBLP	1,049,866	317,080
	rnd_40k_0.001	799,961 40,000 531,677,274
	rnd_50k_0.001 1,250,922 50,000 906,630,823

In Big Data frameworks such as Spark, the driver is the process that creates tasks and sends them to be executed in parallel by worker nodes.

Major Bigdata frameworks like Spark and Flink provide a Scala API and are implemented in Scala which makes Scala a suitable language for our work. Scala also provides reflection which allows generic Scala constructs to be part of the algebra as we will explain later.

Γ ⊢ 𝑒 1 : 𝑡 → Bag 𝑙 [𝑡 ′ ] Γ ⊢ 𝑒 2 : 𝑇 [𝑡 ] Γ ⊢ flmap(𝑒 1 , 𝑒 2 ) : 𝑇 [𝑡 ′ ] Γ ⊢ 𝑒 1 : 𝑡 → 𝑡 → 𝑡 Γ ⊢ 𝑒 2 : 𝑡 Γ ⊢ 𝑒 3 : 𝑇 [𝑡 ] Γ ⊢ reduce(𝑒 1 , 𝑒 2 , 𝑒 3 ) : 𝑡 Γ ⊢ 𝑒 1 : 𝑡 ′ → 𝑡 ′ → 𝑡 ′ Γ ⊢ 𝑒 2 : 𝑇 [𝑡 × 𝑡 ′ ] Γ ⊢ reduceByKey(𝑒 1 , 𝑒 2 ) : 𝑇 [𝑡 × 𝑡 ′ ] Γ ⊢ 𝑒 1 : 𝑇 1 [𝑡 × 𝑡 1 ] Γ ⊢ 𝑒 2 : 𝑇 2 [𝑡 × 𝑡 2 ] 𝑇 3 = (if 𝑇 1 = 𝑇 2 then 𝑇 1 else Bag 𝑑 ) Γ ⊢ cogroup(𝑒 1 , 𝑒 2 ) : 𝑇 3 [𝑡 × Bag 𝑙 [𝑡 1 ] × Bag 𝑙 [𝑡 2 ] ] Γ ⊢ 𝑒 1 : 𝑇 1 [𝑡 × 𝑡 1 ] Γ ⊢ 𝑒 2 : 𝑇 2 [𝑡 × 𝑡 2 ] 𝑇 3 = (if 𝑇 1 = 𝑇 2 then 𝑇 1 else Bag 𝑑 ) Γ ⊢ join(𝑒 1 , 𝑒 2 ) : 𝑇 3 [𝑡 × (𝑡 1 × 𝑡 2 ) ] Γ ⊢ 𝑒 1 : 𝑇 [𝑡 ] Γ ⊢ 𝑒 2 : 𝑇 [𝑡 ] → 𝑇 [𝑡 ] Γ ⊢ 𝑒 : 𝑇 [𝑡 ] → 𝑇 [𝑡 ] Γ ⊢ 𝜇 𝑒 (𝑒 1 , 𝑒 2 ) : 𝑇 [𝑡 ] ∀𝑖 Γ ⊢ 𝑒 𝑖 : 𝑡 𝑖 Γ ⊢ 𝐶 (𝑒 1 , 𝑒 2 , ..., 𝑒 𝑛 ) : 𝐶 [𝑡 1 , 𝑡 2 , ..., 𝑡 𝑛 ] Γ ⊢ 𝑒 : 𝑡 Γ ⊢ {{𝑒 }} : Bag 𝑙 [𝑡 ] 𝑡 ′ 1 + ••• + 𝑡 ′ 𝑛 = 𝑡 ′ match(𝜋 𝑖 , 𝑡 ′ 𝑖 ) → Γ ′ 𝑖 Γ + Γ ′ 𝑖 ⊢ 𝑒 𝑖 : 𝑡 𝑖 𝑡 1 + ••• + 𝑡 𝑛 = 𝑡 Γ ⊢ 𝜆 ⟨𝜋 1 → 𝑒 1 | ••• | 𝜋 𝑛 → 𝑒 𝑛 ⟩ : 𝑡 ′ → 𝑡 Γ ⊢ 𝑒 1 : 𝑡 1 → 𝑡 ′ Γ ⊢ 𝑒 2 : 𝑡 2 𝑡 2 <: 𝑡 1 Γ ⊢ 𝑒 1 𝑒 2 : 𝑡 ′

This is due to the universal property of Bag[𝑡 ], which is the free commutative monoid on 𝑡 .

The least fixpoint, if we define an appropriate ordering relation on 𝑀 𝛿 , e. g. set inclusion in the standard case where 𝛿 = distinct.

We consider it a more practical solution than implementing our own type inference system supporting polymorphism.

We use a cleaned version of the real world dataset Yago 2s[START_REF]YAGO: A high-quality knowledge base[END_REF], that we have preprocessed in order to remove duplicate RDF[START_REF] Cyganiak | RDF 1.1 Concepts and Abstract Syntax[END_REF] triples (of the form <source, label, target>) and keep only triples with existing and valid identifiers. After preprocessing, we obtain a table of Yago facts with 83 predicates and 62,643,951 rows (graph edges). For this dataset, transitive closures are computed for the isLocatedIn edge label.