You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mrql.apache.org by fe...@apache.org on 2014/03/13 15:24:46 UTC
[21/26] MRQL-32: Refactoring directory structure for Eclipse
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/core/src/main/java/org/apache/mrql/MethodInfo.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/MethodInfo.java b/core/src/main/java/org/apache/mrql/MethodInfo.java
new file mode 100644
index 0000000..2dc9ad1
--- /dev/null
+++ b/core/src/main/java/org/apache/mrql/MethodInfo.java
@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.mrql;
+
+import java.lang.reflect.Method;
+import org.apache.mrql.gen.*;
+
+
+/** class for storing Java method information */
+final public class MethodInfo implements Comparable<MethodInfo> {
+ public String name;
+ public Trees signature;
+ public Method method;
+
+ MethodInfo ( String n, Trees s, Method m ) {
+ name = n;
+ signature = s;
+ method = m;
+ }
+
+ public int compareTo ( MethodInfo x ) {
+ int c = name.compareTo(x.name);
+ if (c != 0)
+ return c;
+ if (signature.length() < x.signature.length())
+ return -1;
+ if (signature.length() > x.signature.length())
+ return 1;
+ // handles overloading: more specific method signatures first
+ for ( int i = 1; i < signature.length(); i++ ) {
+ int ct = TypeInference.compare_types(signature.nth(i),x.signature.nth(i));
+ if (ct != 0)
+ return ct;
+ };
+ return TypeInference.compare_types(signature.nth(0),x.signature.nth(0));
+ }
+
+ public boolean equals ( Object x ) {
+ return name.equals(((MethodInfo)x).name)
+ && signature.equals(((MethodInfo)x).signature);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/core/src/main/java/org/apache/mrql/Normalization.gen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/Normalization.gen b/core/src/main/java/org/apache/mrql/Normalization.gen
new file mode 100644
index 0000000..37d628e
--- /dev/null
+++ b/core/src/main/java/org/apache/mrql/Normalization.gen
@@ -0,0 +1,406 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.mrql;
+
+import org.apache.mrql.gen.*;
+
+
+/** normalize algebraic expressions to more efficient forms using heuristic rules */
+public class Normalization extends Translator {
+
+ /** given that pattern=e, find the bindings of the pattern variables */
+ static Trees bind_pattern ( Tree pattern, Tree e ) {
+ Trees args = #[];
+ match pattern {
+ case tuple(...pl):
+ int i = 0;
+ for ( Tree p: pl ) {
+ args = args.append(bind_pattern(p,#<nth(`e,`i)>));
+ i++;
+ }
+ case record(...bl):
+ Trees attrs = #[];
+ for ( Tree b: bl )
+ match b {
+ case bind(`n,`p):
+ args = args.append(bind_pattern(p,#<project(`e,`n)>));
+ if (attrs.member(n))
+ error("Duplicate record attribute name: "+n);
+ attrs = attrs.append(n);
+ };
+ case typed(`p,`t):
+ args = bind_pattern(p,#<typed(`e,`t)>);
+ case list(...pl):
+ int i = 0;
+ for ( Tree p: pl ) {
+ args = args.append(bind_pattern(p,#<index(`e,`i)>));
+ i++;
+ };
+ args = args.append(#<call(eq,call(count,`e),`i)>);
+ case call(`c,...s):
+ Tree ci = data_constructors.lookup(c.toString());
+ if (ci == null)
+ error("Undefined data constructor: "+c);
+ match ci {
+ case `dname(`n,`tp):
+ args = args.append(#<call(eq,union_tag(`e),`n)>);
+ args = args.append(bind_pattern(s.length() == 1 ? s.head() : #<tuple(...s)>,
+ #<typed(union_value(`e),`tp)>));
+ };
+ case any: ;
+ case `v:
+ if (!v.is_variable()) // constant in pattern
+ args = #[call(eq,`e,`v)];
+ else if (st.lookup(v.toString()) != null // repeated pattern variable
+ && !(e.is_variable() && st.lookup(v.toString()).is_variable())) // exception
+ args = #[call(eq,`e,`(st.lookup(v.toString())))];
+ else st.insert(v.toString(),e); // new pattern variable
+ };
+ return args;
+ }
+
+ private static Tree make_tuple ( Trees pl ) {
+ if (pl.length() == 1)
+ return pl.head();
+ return #<tuple(...pl)>;
+ }
+
+ /** remove group-bys and order-bys from the MRQL queries */
+ static Tree remove_groupby ( Tree e ) {
+ Tree ret = #<error>;
+ match e {
+ case select(distinct,`u,from(...bl),where(`c),groupby(...gl),orderby(...ol)):
+ ret = #<select(none,tuple(`u,`u),from(...bl),where(`c),groupby(...gl),orderby(...ol))>;
+ ret = #<cmap(lambda(tuple(key,group),list(key)),groupBy(`ret))>;
+ return remove_groupby(ret);
+ case select(none,`u,from(...bl),where(`c),groupby(),orderby()):
+ return remove_groupby(#<select(`u,from(...bl),where(`c))>);
+ case select(none,`u,from(...bl),where(`c),groupby(...gl),orderby(`l,...ol)):
+ Tree tol = make_tuple(ol);
+ ret = #<cmap(lambda(tuple(key,group),group),
+ orderBy(select(none,tuple(`tol,`u),
+ from(...bl),
+ where(`c),groupby(...gl),orderby())))>;
+ return (l.equals(#<none>))
+ ? remove_groupby(ret)
+ : #<range(`(remove_groupby(ret)),0,`l)>;
+ case select(none,`u,from(...bl),where(`c),groupby(`h,...gl),orderby()):
+ Trees pl = #[];
+ Trees ul = #[];
+ Trees ql = #[];
+ for ( Tree b: bl )
+ match b {
+ case bind(`p,`d):
+ pl = pl.append(p);
+ };
+ Trees pvs = #[];
+ for ( Tree g: gl )
+ match g {
+ case bind(`p,`d):
+ ql = ql.append(p);
+ ul = ul.append(d);
+ pvs = pvs.append(pattern_variables(p));
+ };
+ Tree tql = make_tuple(ql);
+ Tree tul = make_tuple(ul);
+ Tree tpl = make_tuple(pl);
+ Trees xl = #[];
+ Trees partl = #[];
+ for ( Tree x: pattern_variables(#<tuple(...pl)>) )
+ if (!pvs.member(x)) {
+ partl = partl.append(#<bind(`x,`x)>);
+ match rename(#<select(`x,from(bind(`tpl,group)),where(true))>) {
+ case select(`hd,`binds,...):
+ xl = xl.append(#<bind(`x,bag(select(`hd,`binds,where(true))))>);
+ }
+ };
+ match rename(#<select(record(...partl),from(bind(`tpl,group)),where(true))>) {
+ case select(`hd,`binds,...):
+ xl = xl.cons(#<bind(partition,bag(select(`hd,`binds,where(true))))>);
+ }
+ tpl = subst(#<any>,#<0>,tpl);
+ ret = #<select(`u,from(bind(tuple(`tql,group),
+ groupBy(select(tuple(`tul,`tpl),from(...bl),where(`c)))),
+ ...xl),where(`h))>;
+ return remove_groupby(ret);
+ case intersect(`x,`y):
+ return remove_groupby(#<select(x,from(bind(x,`x),bind(y,`y)),
+ where(call(eq,x,y)))>);
+ case except(`x,`y):
+ return remove_groupby(#<select(x,from(bind(x,`x)),
+ where(call(not,call(exists,select(y,from(bind(y,`y)),
+ where(call(eq,x,y)))))))>);
+ case member(`x,`y):
+ return remove_groupby(#<call(exists,select(y,from(bind(y,`y)),
+ where(call(eq,y,`x))))>);
+ case call(gen,`min,`max,`size):
+ return #<gen(`(remove_groupby(min)),`(remove_groupby(max)),`(remove_groupby(size)))>;
+ case call(avg,`s):
+ return remove_groupby(#<call(avg_value,call(avg_aggr,`s))>);
+ case call(`f,...al):
+ Tree macro = global_macros.lookup(f.toString());
+ if (macro == null)
+ fail;
+ match macro {
+ case macro(params(...pl),`body):
+ Tree b = rename(remove_groupby(body));
+ if (pl.length() != al.length())
+ fail;
+ for ( ; !pl.is_empty(); pl = pl.tail(), al = al.tail() )
+ b = subst(pl.head(),remove_groupby(al.head()),b);
+ return b;
+ }
+ case call(`f,...al):
+ if (#[cmap,join,mapReduce,mapReduce2,groupBy,orderBy,tuple,bag,list,set].member(f))
+ return remove_groupby(#<`(f.toString())(...al)>);
+ else fail
+ case project(`x,`a):
+ return #<project(`(remove_groupby(x)),`a)>;
+ case `f(...al):
+ Trees bl = #[];
+ for ( Tree a: al )
+ bl = bl.append(remove_groupby(a));
+ return #<`f(...bl)>;
+ case `v:
+ if (v.is_variable()) {
+ ret = global_vars.lookup(v.toString());
+ if (ret == null)
+ return v;
+ else if (!v.equals(ret))
+ return remove_groupby(ret);
+ }
+ };
+ return e;
+ }
+
+ private static Tree make_and ( Trees tests ) {
+ if (tests.is_empty())
+ return #<true>;
+ Tree e = tests.head();
+ for ( Tree t: tests.tail() )
+ e = #<call(and,`e,`t)>;
+ return e;
+ }
+
+ private static Trees rename_list ( Trees al ) {
+ Trees bl = #[];
+ for ( Tree a: al )
+ bl = bl.append(rename(a));
+ return bl;
+ }
+
+ /** compile away patterns and rename local variables of an MRQL expression e with unique names */
+ static Tree rename ( Tree e ) {
+ Tree ret = #<error>;
+ match e {
+ case `v:
+ if (!v.is_variable())
+ fail;
+ ret = st.lookup(v.toString());
+ if (ret==null)
+ return v;
+ else return ret;
+ case select(`u,from(...bl),where(`c)):
+ st.begin_scope();
+ Trees binds = #[];
+ Trees tests = #[];
+ for ( Tree b: bl )
+ match b {
+ case bind(`p,`d):
+ Tree x = new_var();
+ binds = binds.append(#<bind(`x,`(rename(d)))>);
+ tests = tests.append(bind_pattern(p,x));
+ };
+ c = make_and(tests.cons(c));
+ ret = #<select(`(rename(u)),
+ from(...binds),
+ where(`(rename(c))))>;
+ st.end_scope();
+ return ret;
+ case lambda(`p,`b):
+ st.begin_scope();
+ Tree nv = new_var();
+ if (!bind_pattern(p,nv).is_empty())
+ error("Lambda patterns must be irrefutable: "+print_query(e));
+ ret = #<lambda(`nv,`(rename(b)))>;
+ st.end_scope();
+ return ret;
+ case function(tuple(...params),`outp,`body):
+ st.begin_scope();
+ Trees ps = #[];
+ Trees vs = #[];
+ for ( Tree p: params )
+ match p {
+ case `bind(`v,`tp):
+ Tree nv = new_var();
+ if (vs.member(v))
+ error("Duplicate function parameters: "+print_query(e));
+ vs = vs.append(v);
+ ps = ps.append(#<`bind(`nv,`tp)>);
+ st.insert(v.toString(),nv);
+ };
+ ret = #<function(tuple(...ps),`outp,`(rename(body)))>;
+ st.end_scope();
+ return ret;
+ case let(`p,`u,`b):
+ Tree ne = rename(u);
+ st.begin_scope();
+ Tree nv = new_var();
+ if (!bind_pattern(p,nv).is_empty())
+ error("Let patterns must be irrefutable: "+print_query(e));
+ ret = #<let(`nv,`ne,`(rename(b)))>;
+ st.end_scope();
+ return ret;
+ case case(`u,...cs):
+ Trees rs = cs.reverse();
+ Tree nu = rename(u);
+ match rs.head() {
+ case case(`p,`b):
+ Trees conds = bind_pattern(p,nu);
+ if (!conds.is_empty())
+ error("Non-exhaustive case "+print_query(p)+" in "+print_query(e));
+ ret = b;
+ };
+ for ( Tree c: rs.tail() )
+ match c {
+ case case(`p,`b):
+ Trees conds = bind_pattern(p,nu);
+ if (!conds.is_empty())
+ ret = #<if(`(make_and(conds)),`b,`ret)>;
+ else error("Unreachable case "+print_query(p)+" in "+print_query(e));
+ };
+ return rename(ret);
+ case project(`u,`a):
+ return #<project(`(rename(u)),`a)>;
+ case bind(`a,`u):
+ return #<bind(`a,`(rename(u)))>;
+ case loop(lambda(tuple(...vs),`b),`s,`n):
+ return #<loop(lambda(tuple(...vs),`(rename(b))),`(rename(s)),`n)>;
+ case `f(...al):
+ Trees bl = rename_list(al);
+ return #<`f(...bl)>;
+ };
+ return e;
+ }
+
+ private static Trees has_existential ( Tree e ) {
+ match e {
+ case call(and(`x,`y)):
+ Trees xs = has_existential(x);
+ Trees ys = has_existential(y);
+ return #[call(and(`(xs.head()),`(ys.head())),...(xs.tail()),...(ys.tail()))];
+ case call(exists,select(...)):
+ return #[true,`e];
+ case call(not,call(all,select(...l))):
+ return #[true,call(exists,select(...l))];
+ };
+ return #[`e];
+ }
+
+ /** normalize algebraic expressions to more efficient forms using heuristic rules */
+ public static Tree normalize ( Tree e ) {
+ match e {
+ case select(`u,from(),where(true)):
+ return normalize(#<bag(`u)>);
+ case select(`u,from(),where(`p)):
+ return normalize(#<if(`p,bag(`u),bag())>);
+ case select(`u,from(bind(`v,`d)),where(true)):
+ if (u.equals(v))
+ return normalize(d);
+ else fail
+ case select(`u,from(...bl,bind(`v,select(`iu,from(...ibl),where(`ic))),...al),where(`c)):
+ return normalize(#<select(`u,from(...bl,...ibl,bind(`v,bag(`iu)),...al),
+ where(call(and,`c,`ic)))>);
+ case select(`u,from(...bl,bind(`v,bag(`d)),...al),`c):
+ if (!is_pure(d) && occurences(v,#<f(`c,`u,...al)>) > 1) // duplicated side-effects
+ fail;
+ return normalize(#<select(`(subst(v,d,u)),
+ from(...bl,...(subst_list(v,d,al))),
+ `(subst(v,d,c)))>);
+ case select(`u,from(...bl),where(`c)):
+ Trees es = has_existential(c);
+ if (es.length() <= 1)
+ fail;
+ Trees binds = bl;
+ Trees preds = #[`(es.head())];
+ for ( Tree x: es.tail() )
+ match x {
+ case call(exists,select(`p,from(...bl2),where(`c2))):
+ preds = preds.cons(p).cons(c2);
+ binds = binds.append(bl2);
+ };
+ return normalize(#<select(`u,from(...binds),where(`(make_and(preds))))>);
+ case let_bind(`v,`x,`y):
+ return #<let(`v,`(normalize(x)),`(normalize(y)))>;
+ case call(eq,tuple(...l),`x):
+ Tree pl = #<true>;
+ int i = 0;
+ for ( Tree y: l ) {
+ pl = #<call(and,`pl,call(eq,`y,nth(`x,`i)))>;
+ i++;
+ };
+ return normalize(pl);
+ case call(eq,`x,tuple(...l)):
+ Tree pl = #<true>;
+ int i = 0;
+ for (Tree y: l) {
+ pl = #<call(and,`pl,call(eq,nth(`x,`i),`y))>;
+ i++;
+ };
+ return normalize(pl);
+ case call(and,true,`u): return normalize(u);
+ case call(and,`u,true): return normalize(u);
+ case call(and,false,`u): return #<false>;
+ case call(and,`u,false): return #<false>;
+ case call(or,true,`u): return #<true>;
+ case call(or,`u,true): return #<true>;
+ case call(or,false,`u): return normalize(u);
+ case call(or,`u,false): return normalize(u);
+ case call(not,true): return #<false>;
+ case call(not,false): return #<true>;
+ case if(true,`e1,`e2): return normalize(e1);
+ case if(false,`e1,`e2): return normalize(e2);
+ case nth(tuple(...al),`n):
+ if (!n.is_long())
+ fail;
+ int i = (int)n.longValue();
+ if ( i >= 0 && i < al.length() )
+ return normalize(al.nth(i));
+ case project(record(...bl),`a):
+ for ( Tree b: bl )
+ match b {
+ case bind(`v,`u): if (v.equals(a)) return normalize(u);
+ };
+ error("Wrong projection: "+print_query(e));
+ case `f(...al):
+ Trees bl = #[];
+ for ( Tree a: al )
+ bl = bl.append(normalize(a));
+ return #<`f(...bl)>;
+ };
+ return e;
+ }
+
+ /** normalize algebraic expressions to more efficient forms using heuristic rules */
+ public static Tree normalize_all ( Tree e ) {
+ Tree ne = normalize(e);
+ if (e.equals(ne))
+ return e;
+ else return normalize(ne);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/core/src/main/java/org/apache/mrql/ParsedDataSource.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/ParsedDataSource.java b/core/src/main/java/org/apache/mrql/ParsedDataSource.java
new file mode 100644
index 0000000..e000405
--- /dev/null
+++ b/core/src/main/java/org/apache/mrql/ParsedDataSource.java
@@ -0,0 +1,62 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.mrql;
+
+import org.apache.mrql.gen.*;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.conf.Configuration;
+
+
+/** A data source for a text HDFS file along with the parser to parse it */
+public class ParsedDataSource extends DataSource {
+ public Class<? extends Parser> parser;
+ public Trees args;
+
+ ParsedDataSource ( int source_num,
+ String path,
+ Class<? extends Parser> parser,
+ Trees args,
+ Configuration conf ) {
+ super(source_num,path,Evaluator.evaluator.parsedInputFormat(),conf);
+ this.parser = parser;
+ this.args = args;
+ }
+
+ ParsedDataSource ( String path,
+ Class<? extends Parser> parser,
+ Trees args,
+ Configuration conf ) {
+ super(-1,path,Evaluator.evaluator.parsedInputFormat(),conf);
+ this.parser = parser;
+ this.args = args;
+ }
+
+ public String toString () {
+ try {
+ String pn = "";
+ for ( String k: DataSource.parserDirectory.keySet() )
+ if (DataSource.parserDirectory.get(k).equals(parser))
+ pn = k;
+ return "Text"+separator+source_num+separator+pn+separator+path
+ +separator+(new Node("args",args)).toString();
+ } catch (Exception e) {
+ throw new Error(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/core/src/main/java/org/apache/mrql/Parser.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/Parser.java b/core/src/main/java/org/apache/mrql/Parser.java
new file mode 100644
index 0000000..c1728f1
--- /dev/null
+++ b/core/src/main/java/org/apache/mrql/Parser.java
@@ -0,0 +1,32 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.mrql;
+
+import org.apache.mrql.gen.*;
+import org.apache.hadoop.fs.FSDataInputStream;
+
+
+/** for a new text data source, you must implement a new parser */
+interface Parser {
+ public void initialize ( Trees args );
+ public Tree type ();
+ public void open ( String file );
+ public void open ( FSDataInputStream fsin, long start, long end );
+ public String slice ();
+ public Bag parse ( String s );
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/core/src/main/java/org/apache/mrql/Plan.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/Plan.java b/core/src/main/java/org/apache/mrql/Plan.java
new file mode 100644
index 0000000..c4951bf
--- /dev/null
+++ b/core/src/main/java/org/apache/mrql/Plan.java
@@ -0,0 +1,362 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.mrql;
+
+import org.apache.mrql.gen.*;
+import java.io.*;
+import java.util.Random;
+import java.util.ArrayList;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.SequenceFile.Sorter;
+
+
+/** A physical plan (a superclass for both MapReduce, BSP, and Spark plans) */
+public class Plan {
+ public static Configuration conf;
+ static ArrayList<String> temporary_paths = new ArrayList<String>();
+ private static Random random_generator = new Random();
+ final static int max_input_files = 100;
+
+ /** generate a new path name in HDFS to store intermediate results */
+ public static String new_path ( Configuration conf ) throws IOException {
+ String dir = (Config.local_mode)
+ ? ((Config.tmpDirectory == null) ? "/tmp/mrql" : Config.tmpDirectory)
+ : "mrql";
+ Path p;
+ do {
+ p = new Path(dir+"/mrql"+random_generator.nextInt(1000000));
+ } while (p.getFileSystem(conf).exists(p));
+ String path = p.toString();
+ temporary_paths.add(path);
+ DataSource.dataSourceDirectory.distribute(conf);
+ return path;
+ }
+
+ /** remove all temporary files */
+ public static void clean () throws IOException {
+ for (String p: temporary_paths)
+ try {
+ Path path = new Path(p);
+ path.getFileSystem(conf).delete(path,true);
+ } catch (Exception ex) {
+ FileSystem.getLocal(conf).delete(new Path(p),true);
+ };
+ temporary_paths.clear();
+ DataSource.dataSourceDirectory.clear();
+ }
+
+ /** return the data set size in bytes */
+ public final static long size ( DataSet s ) {
+ return s.size(conf);
+ }
+
+ /** the cache that holds all local data in memory */
+ static Tuple cache;
+
+ /** return the cache element at location loc */
+ public static synchronized MRData getCache ( int loc ) {
+ return cache.get(loc);
+ }
+
+ /** set the cache element at location loc to value and return ret */
+ public static synchronized MRData setCache ( int loc, MRData value, MRData ret ) {
+ if (value instanceof Bag)
+ ((Bag)value).materialize();
+ cache.set(loc,value);
+ return ret;
+ }
+
+ /** put the jar file that contains the compiled MR functional parameters into the TaskTracker classpath */
+ final static void distribute_compiled_arguments ( Configuration conf ) {
+ try {
+ if (!Config.compile_functional_arguments)
+ return;
+ Path local_path = new Path("file://"+Compiler.jar_path);
+ if (Config.spark_mode)
+ conf.set("mrql.jar.path",local_path.toString());
+ else {
+ // distribute the jar file with the compiled arguments to all clients
+ Path hdfs_path = new Path("mrql-tmp/class"+random_generator.nextInt(1000000)+".jar");
+ FileSystem fs = hdfs_path.getFileSystem(conf);
+ fs.copyFromLocalFile(false,true,local_path,hdfs_path);
+ temporary_paths.add(hdfs_path.toString());
+ conf.set("mrql.jar.path",hdfs_path.toString());
+ }
+ } catch (Exception ex) {
+ throw new Error(ex);
+ }
+ }
+
+ /** retrieve the compiled functional argument of code */
+ final static Function functional_argument ( Configuration conf, Tree code ) {
+ Node n = (Node)code;
+ if (n.name().equals("compiled"))
+ try {
+ // if the clent has not received the jar file with the compiled arguments, copy the file from HDFS
+ if (Compiler.jar_path == null) {
+ Path hdfs_path = new Path(conf.get("mrql.jar.path"));
+ String local_path = Compiler.tmp_dir+"/mrql_args_"+random_generator.nextInt(1000000)+".jar";
+ FileSystem fs = hdfs_path.getFileSystem(conf);
+ fs.copyToLocalFile(false,hdfs_path,new Path("file://"+local_path));
+ Compiler.jar_path = local_path;
+ };
+ return Compiler.compiled(conf.getClassLoader(),n.children().nth(0).toString());
+ } catch (Exception ex) {
+ System.err.println("*** Warning: Unable to retrieve the compiled lambda: "+code);
+ return ((Lambda) Interpreter.evalE(n.children().nth(1))).lambda();
+ }
+ else if (code.equals(Interpreter.identity_mapper))
+ return new Function () {
+ public MRData eval ( final MRData x ) { return new Bag(x); }
+ };
+ else return ((Lambda) Interpreter.evalE(code)).lambda();
+ }
+
+ /** comparator for MRData keys */
+ public final static class MRContainerKeyComparator implements RawComparator<MRContainer> {
+ int[] container_size;
+
+ public MRContainerKeyComparator () {
+ container_size = new int[1];
+ }
+
+ final public int compare ( byte[] x, int xs, int xl, byte[] y, int ys, int yl ) {
+ return MRContainer.compare(x,xs,xl,y,ys,yl,container_size);
+ }
+
+ final public int compare ( MRContainer x, MRContainer y ) {
+ return x.compareTo(y);
+ }
+ }
+
+ /** The source physical operator for binary files */
+ public final static DataSet binarySource ( int source_num, String file ) {
+ return new DataSet(new BinaryDataSource(source_num,file,conf),0,0);
+ }
+
+ /** The source physical operator for binary files */
+ public final static DataSet binarySource ( String file ) {
+ return new DataSet(new BinaryDataSource(-1,file,conf),0,0);
+ }
+
+ /** splits the range min..max into multiple ranges, one for each mapper */
+ public final static DataSet generator ( int source_num, long min, long max, long split_length ) throws Exception {
+ if (min > max)
+ throw new Error("Wrong range: "+min+"..."+max);
+ if (split_length < 1)
+ if (Config.bsp_mode)
+ split_length = (max-min)/Config.nodes+1;
+ else split_length = Config.range_split_size;
+ DataSet ds = new DataSet(0,0);
+ long i = min;
+ while (i+split_length <= max) {
+ String file = new_path(conf);
+ Path path = new Path(file);
+ SequenceFile.Writer writer
+ = SequenceFile.createWriter(path.getFileSystem(conf),conf,path,
+ MRContainer.class,MRContainer.class,
+ SequenceFile.CompressionType.NONE);
+ writer.append(new MRContainer(new MR_long(i)),
+ new MRContainer(new Tuple(new MR_long(i),new MR_long(split_length))));
+ writer.close();
+ ds.source.add(new GeneratorDataSource(source_num,file,conf));
+ i += split_length;
+ };
+ if (i <= max) {
+ String file = new_path(conf);
+ Path path = new Path(file);
+ SequenceFile.Writer writer
+ = SequenceFile.createWriter(path.getFileSystem(conf),conf,path,
+ MRContainer.class,MRContainer.class,
+ SequenceFile.CompressionType.NONE);
+ writer.append(new MRContainer(new MR_long(i)),
+ new MRContainer(new Tuple(new MR_long(i),new MR_long(max-i+1))));
+ writer.close();
+ ds.source.add(new GeneratorDataSource(source_num,file,conf));
+ };
+ return ds;
+ }
+
+ /** splits the range min..max into multiple ranges, one for each mapper */
+ public final static DataSet generator ( long min, long max, long split_length ) throws Exception {
+ return generator(-1,min,max,split_length);
+ }
+
+ /** The source physical operator for parsing text files */
+ public final static DataSet parsedSource ( int source_num, Class<? extends Parser> parser, String file, Trees args ) {
+ return new DataSet(new ParsedDataSource(source_num,file,parser,args,conf),0,0);
+ }
+
+ /** The source physical operator for parsing text files */
+ public final static DataSet parsedSource ( Class<? extends Parser> parser, String file, Trees args ) {
+ return new DataSet(new ParsedDataSource(file,parser,args,conf),0,0);
+ }
+
+ /** merge the sorted files of the data source */
+ public final static Bag merge ( final DataSource s ) throws Exception {
+ Path path = new Path(s.path);
+ final FileSystem fs = path.getFileSystem(conf);
+ final FileStatus[] ds
+ = fs.listStatus(path,
+ new PathFilter () {
+ public boolean accept ( Path path ) {
+ return !path.getName().startsWith("_");
+ }
+ });
+ int dl = ds.length;
+ if (dl == 0)
+ return new Bag();
+ Path[] paths = new Path[dl];
+ for ( int i = 0; i < dl; i++ )
+ paths[i] = ds[i].getPath();
+ if (dl > Config.max_merged_streams) {
+ if (Config.trace)
+ System.err.println("*** Merging "+dl+" files");
+ Path out_path = new Path(new_path(conf));
+ SequenceFile.Sorter sorter
+ = new SequenceFile.Sorter(fs,new MRContainerKeyComparator(),
+ MRContainer.class,MRContainer.class,conf);
+ sorter.merge(paths,out_path);
+ paths = new Path[1];
+ paths[0] = out_path;
+ };
+ final int n = paths.length;
+ SequenceFile.Reader[] sreaders = new SequenceFile.Reader[n];
+ for ( int i = 0; i < n; i++ )
+ sreaders[i] = new SequenceFile.Reader(fs,paths[i],conf);
+ final SequenceFile.Reader[] readers = sreaders;
+ final MRContainer[] keys_ = new MRContainer[n];
+ final MRContainer[] values_ = new MRContainer[n];
+ for ( int i = 0; i < n; i++ ) {
+ keys_[i] = new MRContainer();
+ values_[i] = new MRContainer();
+ };
+ return new Bag(new BagIterator () {
+ int min = 0;
+ boolean first = true;
+ final MRContainer[] keys = keys_;
+ final MRContainer[] values = values_;
+ final MRContainer key = new MRContainer();
+ final MRContainer value = new MRContainer();
+ public boolean hasNext () {
+ if (first)
+ try {
+ first = false;
+ for ( int i = 0; i < n; i++ )
+ if (readers[i].next(key,value)) {
+ keys[i].set(key.data());
+ values[i].set(value.data());
+ } else {
+ keys[i] = null;
+ readers[i].close();
+ }
+ } catch (IOException e) {
+ throw new Error("Cannot merge values from an intermediate result");
+ };
+ min = -1;
+ for ( int i = 0; i < n; i++ )
+ if (keys[i] != null && min < 0)
+ min = i;
+ else if (keys[i] != null && keys[i].compareTo(keys[min]) < 0)
+ min = i;
+ return min >= 0;
+ }
+ public MRData next () {
+ try {
+ MRData res = values[min].data();
+ if (readers[min].next(key,value)) {
+ keys[min].set(key.data());
+ values[min].set(value.data());
+ } else {
+ keys[min] = null;
+ readers[min].close();
+ };
+ return res;
+ } catch (IOException e) {
+ throw new Error("Cannot merge values from an intermediate result");
+ }
+ }
+ });
+ }
+
+ /** The collect physical operator */
+ public final static Bag collect ( final DataSet x, boolean strip ) throws Exception {
+ return Evaluator.evaluator.parsedInputFormat().newInstance().collect(x,strip);
+ }
+
+ /** The collect physical operator */
+ public final static Bag collect ( final DataSet x ) throws Exception {
+ return collect(x,true);
+ }
+
+ /** the DataSet union physical operator */
+ public final static DataSet merge ( final DataSet x, final DataSet y ) throws IOException {
+ DataSet res = x;
+ res.source.addAll(y.source);
+ return res;
+ }
+
+ final static MR_long counter_key = new MR_long(0);
+ final static MRContainer counter_container = new MRContainer(counter_key);
+ final static MRContainer value_container = new MRContainer(new MR_int(0));
+
+ /** The cache operator that dumps a bag into an HDFS file */
+ public final static DataSet fileCache ( Bag s ) throws IOException {
+ String newpath = new_path(conf);
+ Path path = new Path(newpath);
+ FileSystem fs = path.getFileSystem(conf);
+ SequenceFile.Writer writer
+ = new SequenceFile.Writer(fs,conf,path,
+ MRContainer.class,MRContainer.class);
+ long i = 0;
+ for ( MRData e: s ) {
+ counter_key.set(i++);
+ value_container.set(e);
+ writer.append(counter_container,value_container);
+ };
+ writer.close();
+ return new DataSet(new BinaryDataSource(0,newpath,conf),0,0);
+ }
+
+ /** for dumped data to a file, return the MRQL type of the data */
+ public final static Tree get_type ( String file ) {
+ try {
+ Path path = new Path(file);
+ FileSystem fs = path.getFileSystem(conf);
+ BufferedReader ftp = new BufferedReader(new InputStreamReader(fs.open(path.suffix(".type"))));
+ String s[] = ftp.readLine().split("@");
+ ftp.close();
+ if (s.length != 2 )
+ return null;
+ if (!s[0].equals("2"))
+ throw new Error("The binary file has been created in java mode and cannot be read in hadoop mode");
+ return Tree.parse(s[1]);
+ } catch (Exception e) {
+ return null;
+ }
+ }
+
+ /** create a new PrintStream from the file */
+ final static PrintStream print_stream ( String file ) throws Exception {
+ Path path = new Path(file);
+ FileSystem fs = path.getFileSystem(conf);
+ return new PrintStream(fs.create(path));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/core/src/main/java/org/apache/mrql/PlanGeneration.gen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/PlanGeneration.gen b/core/src/main/java/org/apache/mrql/PlanGeneration.gen
new file mode 100644
index 0000000..59b8cb8
--- /dev/null
+++ b/core/src/main/java/org/apache/mrql/PlanGeneration.gen
@@ -0,0 +1,716 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.mrql;
+
+import org.apache.mrql.gen.*;
+import java.util.*;
+import java.io.*;
+
+
+/** Generate a physical plan from an algebraic expression */
+final public class PlanGeneration extends AlgebraicOptimization {
+
+ /** extract the combiner from the reducer in a MapReduce plan */
+ static class Aggregates {
+ public static Trees maps = #[];
+ public static Trees combines = #[];
+ public static Trees reduces = #[];
+ public static boolean can_use_combiner = true;
+ private static SymbolTable st = new SymbolTable();
+
+ private static void clear () {
+ maps = #[];
+ combines = #[];
+ reduces = #[];
+ can_use_combiner = true;
+ }
+
+ /** extends the combines, maps, and reduces lists with new entries */
+ private static int union_aggegates ( Tree reduce, Tree map, Tree combine ) {
+ Tree m = simplify_all(map);
+ Tree c = simplify_all(combine);
+ Tree rd = simplify_all(reduce);
+ int i = 0;
+ for ( Trees r = reduces; !r.is_empty(); r = r.tail(), i++ )
+ if (alpha_equivalent(rd,r.head()))
+ return i;
+ maps = maps.append(m);
+ reduces = reduces.append(rd);
+ combines = combines.append(subst(#<-1>,#<`i>,c));
+ return i;
+ }
+
+ /** Generate the MR combiner from the MR reducer.
+ * Find the aggregation calls (eg, call(avg,_)) in the reducer
+ * @param e the body of the reducer
+ * @param map the map function
+ * @param mvar the variable of the map function
+ * @param rvar the variable of the reducer function
+ * @return the combiner, but also sets the combines, maps, and reduces lists
+ */
+ private static Tree derive_combiner ( Tree e, Tree map, Tree mvar, Tree rvar ) {
+ Tree gvar = #<nth(`rvar,1)>;
+ match e {
+ case call(`f,`u):
+ for ( Tree monoid: monoids ) // system & user-defined aggregations
+ match monoid {
+ case `nm(`mtp,`plus,`zero,`unit):
+ if (nm.equals(f.toString())) {
+ match TypeInference.type_inference2(u) {
+ case `S(`tp):
+ if (TypeInference.unify(tp,mtp) == null)
+ continue;
+ case _: throw new Error("Unexpected aggregation: "+e);
+ };
+ match u {
+ case cmap(`m,`v):
+ if (!v.equals(gvar) || occurences(rvar,m) > 0)
+ fail;
+ Tree ev = new_var();
+ Tree nv = new_var();
+ Tree mv = new_var();
+ int i = union_aggegates(e,
+ #<aggregate(lambda(`ev,apply(`plus,tuple(nth(`ev,0),apply(`unit,nth(`ev,1))))),
+ `zero,cmap(`m,cmap(lambda(x,bag(nth(x,1))),apply(`map,`mvar))))>,
+ #<aggregate(lambda(`nv,apply(`plus,tuple(nth(`nv,0),
+ nth(nth(`nv,1),-1)))),
+ `zero,`gvar)>);
+ return simplify_all(#<aggregate(lambda(`mv,apply(`plus,tuple(nth(`mv,0),
+ nth(nth(`mv,1),`i)))),
+ `zero,`gvar)>);
+ case `v:
+ if (!v.equals(gvar))
+ fail;
+ Tree ev = new_var();
+ Tree nv = new_var();
+ Tree mv = new_var();
+ int i = union_aggegates(e,
+ #<aggregate(lambda(`ev,apply(`plus,tuple(nth(`ev,0),apply(`unit,nth(`ev,1))))),
+ `zero,cmap(lambda(x,bag(nth(x,1))),apply(`map,`mvar)))>,
+ #<aggregate(lambda(`nv,apply(`plus,tuple(nth(`nv,0),nth(nth(`nv,1),-1)))),
+ `zero,`gvar)>);
+ return simplify_all(#<aggregate(lambda(`mv,apply(`plus,tuple(nth(`mv,0),
+ nth(nth(`mv,1),`i)))),
+ `zero,`gvar)>);
+ }
+ }
+ };
+ fail
+ case nth(`v,0):
+ if (v.is_variable())
+ return e;
+ else fail
+ case `f(...al):
+ Trees rs = #[];
+ for ( Tree a: al )
+ rs = rs.append(derive_combiner(a,map,mvar,rvar));
+ return #<`f(...rs)>;
+ };
+ if (#<nth(`e,1)>.equals(gvar))
+ Aggregates.can_use_combiner = false;
+ return e;
+ }
+ }
+
+ /** how many times e accesses the bag x? if it's more than one, it can't be streamed */
+ private static int number_of_accesses ( Tree x, Tree e ) {
+ if (e.equals(x))
+ return 1;
+ match e {
+ case cmap(`m,`s):
+ return number_of_accesses(x,m)*10+number_of_accesses(x,s);
+ case map(`m,`s):
+ return number_of_accesses(x,m)*10+number_of_accesses(x,s);
+ case filter(`p,`m,`s):
+ return number_of_accesses(x,p)*10+number_of_accesses(x,m)*10+number_of_accesses(x,s);
+ case `f(...r):
+ int i = 0;
+ for ( Tree z: r )
+ i += number_of_accesses(x,z);
+ return i;
+ };
+ return 0;
+ }
+
+ /** can we process the second arg of the MapReducer reducer (a bag) as a stream? */
+ public static boolean streamed_MapReduce_reducer ( Tree x ) {
+ match x {
+ case lambda(`v,`b):
+ return number_of_accesses(#<nth(`v,1)>,b) <= 1;
+ case compiled(_,lambda(`v,`b)):
+ return number_of_accesses(#<nth(`v,1)>,b) <= 1;
+ };
+ return false;
+ }
+
+ /** can we process the first arg of the MapReducer2 reducer (a bag) as a stream? */
+ public static boolean streamed_MapReduce2_reducer ( Tree x ) {
+ match x {
+ case lambda(`v,`b):
+ return number_of_accesses(#<nth(`v,0)>,b) <= 1;
+ case compiled(_,lambda(`v,`b)):
+ return number_of_accesses(#<nth(`v,0)>,b) <= 1;
+ };
+ return false;
+ }
+
+ /** true if e returns a dataset stored in HDFS */
+ public static boolean is_dataset_expr ( Tree e ) {
+ match TypeInference.type_inference2(e) {
+ case `T(_):
+ if (is_persistent_collection(T))
+ return true;
+ };
+ return false;
+ }
+
+ /** compile an algebraic form to a algebraic plan
+ * @param e the algebraic form
+ * @return the algebraic plan
+ */
+ public static Tree makePlan ( Tree e ) {
+ match e {
+ // combine groupBy with Join (experimental)
+ case mapReduce(lambda(`vm,`bm),lambda(`vr,`br),`s,`o):
+ if (!Config.use_combiner || !is_dataset_expr(s))
+ fail;
+ Tree splan = makePlan(s);
+ match splan {
+ case MapReduce2(lambda(`mvx,bag(tuple(`kx,`mx))),
+ lambda(`mvy,bag(tuple(`ky,`my))),
+ lambda(`v,cmap(lambda(`x,cmap(lambda(`y,bag(tuple(tuple(`gx,`gy),`b))),
+ nth(`vx,1))),
+ nth(`vy,0))),
+ `X,`Y,`o2):
+ if (!vx.equals(v) || !vy.equals(v) || !mx.equals(mvx) || !my.equals(mvy))
+ fail;
+ Tree gxx = gx;
+ Tree gyy = gy;
+ if (free_variables(gx,#[]).equals(#[`y]) && free_variables(gy,#[]).equals(#[`x])) {
+ gxx = gy;
+ gyy = gx;
+ } else if (!free_variables(gx,#[]).equals(#[`x]) || !free_variables(gy,#[]).equals(#[`y]))
+ fail;
+ Aggregates.clear();
+ Tree nv = new_var();
+ match TypeInference.type_inference(bm) {
+ case `S(`tp):
+ if (!is_collection(S))
+ fail;
+ type_env.insert(nv.toString(),tp);
+ };
+ Tree rd = Aggregates.derive_combiner(br,#<lambda(`vm,`bm)>,vm,vr);
+ if (Aggregates.reduces.is_empty())
+ fail;
+ Tree vr2 = new_var();
+ Tree m = simplify_all(#<lambda(`vm,cmap(lambda(`nv,bag(tuple(nth(`nv,0),
+ tuple(...(Aggregates.maps))))),`bm))>);
+ Tree c = subst(vr,vr2,#<bag(tuple(...(Aggregates.combines)))>);
+ c = simplify_all(#<lambda(`vr2,`c)>);
+ Tree r = simplify_all(#<lambda(`vr,`rd)>);
+ Tree mtp = TypeInference.type_inference(#<bag(tuple(...(Aggregates.maps)))>);
+ Tree rtp = #<tuple(`(TypeInference.type_inference(#<nth(`vr,0)>)),`mtp)>;
+ type_env.insert(vr.toString(),rtp);
+ type_env.insert(vr2.toString(),rtp);
+ TypeInference.type_inference(m);
+ TypeInference.type_inference(c);
+ TypeInference.type_inference(r);
+ Tree combiner = makePlan(c);
+ Tree reducer = makePlan(r);
+ Tree nm = simplify_all(subst(x,#<nth(`vx,0)>,
+ subst(y,#<nth(`vy,1)>,
+ #<cmap(lambda(x,bag(nth(x,1))),
+ cmap(`m,bag(tuple(tuple(`gx,`gy),`b))))>)));
+ type_env.insert(v.toString(),TypeInference.type_inference(#<tuple(`mvx,`mvy)>));
+ nm = makePlan(nm);
+ return #<GroupByJoin(lambda(`mvx,`kx),
+ lambda(`mvy,`ky),
+ lambda(`x,`gxx),
+ lambda(`y,`gyy),
+ lambda(`v,`nm),
+ `combiner,
+ `reducer,
+ `X,`Y,`o)>;
+ };
+ fail
+ // extract the mapReduce combiner
+ case mapReduce(lambda(`vm,`bm),lambda(`vr,`br),`s,`o):
+ if (!Config.use_combiner || !is_dataset_expr(s))
+ fail;
+ Tree splan = makePlan(s);
+ Aggregates.clear();
+ Tree nv = new_var();
+ match TypeInference.type_inference(bm) {
+ case `S(`tp):
+ if (!is_collection(S))
+ fail;
+ type_env.insert(nv.toString(),tp);
+ };
+ Tree rd = Aggregates.derive_combiner(br,#<lambda(`vm,`bm)>,vm,vr);
+ if (!Aggregates.can_use_combiner || Aggregates.reduces.is_empty())
+ fail;
+ Tree vr2 = new_var();
+ Tree m = simplify_all(#<lambda(`vm,cmap(lambda(`nv,bag(tuple(nth(`nv,0),
+ tuple(...(Aggregates.maps))))),`bm))>);
+ Tree c = subst(vr,vr2,#<bag(tuple(...(Aggregates.combines)))>);
+ c = simplify_all(#<lambda(`vr2,`c)>);
+ Tree r = simplify_all(#<lambda(`vr,`rd)>);
+ Tree mtp = TypeInference.type_inference(#<bag(tuple(...(Aggregates.maps)))>);
+ Tree rtp = #<tuple(`(TypeInference.type_inference(#<nth(`vr,0)>)),`mtp)>;
+ type_env.insert(vr.toString(),rtp);
+ type_env.insert(vr2.toString(),rtp);
+ TypeInference.type_inference(m);
+ TypeInference.type_inference(c);
+ TypeInference.type_inference(r);
+ Tree combiner = makePlan(c);
+ Tree reducer = makePlan(r);
+ match splan {
+ // if the MapCombineReduce input is a join, push the combiner to the join
+ case MapReduce2(`mx,`my,lambda(`rv,`rb),`x,`y,`o2):
+ Tree nr = makePlan(simplify_all(#<lambda(`rv,cmap(`m,`rb))>));
+ return #<MapReduce(lambda(`vm,bag(`vm)),`reducer,
+ MapCombineReduce2(`mx,`my,`combiner,`nr,`x,`y,`o2),`o)>;
+ case `input:
+ return #<MapCombineReduce(`(makePlan(m)),`combiner,`reducer,`input,`o)>;
+ };
+ fail
+ case mapReduce(`m,`r,`s,`o):
+ if (is_dataset_expr(s))
+ return #<MapReduce(`(makePlan(m)),
+ `(makePlan(r)),
+ `(makePlan(s)),`o)>;
+ else fail
+ case mapReduce2(`mx,`my,`r,`x,`y,`o):
+ if (is_dataset_expr(x) && is_dataset_expr(y) && streamed_MapReduce2_reducer(r))
+ return #<MapReduce2(`(makePlan(mx)),
+ `(makePlan(my)),
+ `(makePlan(r)),
+ `(makePlan(x)),
+ `(makePlan(y)),`o)>;
+ else fail
+ case mapReduce2(`mx,`my,lambda(`v,`b),`x,`y,`o):
+ if (!is_dataset_expr(x) || !is_dataset_expr(y))
+ fail;
+ // if the join reducer is not streaming, switch the inputs
+ Tree nv = new_var();
+ Tree nr = subst(#<nth(`v,0)>,#<nth(`nv,1)>,
+ subst(#<nth(`v,1)>,#<nth(`nv,0)>,b));
+ nr = #<lambda(`nv,`nr)>;
+ type_env.insert(nv.toString(),TypeInference.type_inference(#<tuple(nth(`v,1),nth(`v,0))>));
+ return #<MapReduce2(`(makePlan(my)),
+ `(makePlan(mx)),
+ `(makePlan(nr)),
+ `(makePlan(y)),
+ `(makePlan(x)),`o)>;
+ case crossProduct(`mx,`my,`r,`x,`y):
+ if (is_dataset_expr(x) && is_dataset_expr(y))
+ return #<CrossProduct(`(makePlan(mx)),
+ `(makePlan(my)),
+ `(makePlan(r)),
+ `(makePlan(x)),
+ `(makePlan(y)))>;
+ else fail
+ case cmap(`m,`s):
+ if (is_dataset_expr(s))
+ return #<cMap(`(makePlan(m)),
+ `(makePlan(s)))>;
+ else fail
+ case cmap(lambda(`v,if(`p,`T(`u),`S())),`s):
+ if (false && is_collection(T) && is_collection(S))
+ return makePlan(#<filter(lambda(`v,`p),lambda(`v,`u),`s)>);
+ else fail
+ case call(source,binary,`file,`tp):
+ return #<BinarySource(`file,`tp)>;
+ case call(source,gen,`f,`len,`ulen):
+ return #<SequenceSource(`(makePlan(f)),`(makePlan(len)),
+ `(makePlan(ulen)))>;
+ case call(source,`parser,`file,...args):
+ Trees el = #[];
+ for ( Tree a: args )
+ el = el.append(makePlan(a));
+ return #<ParsedSource(`parser,`(makePlan(file)),...el)>;
+ case type(`x): return e;
+ case gen(`min,`max,`size):
+ return #<Generator(`(makePlan(min)),`(makePlan(max)),`(makePlan(size)))>;
+ case repeat(lambda(`v,`b),`s,`n):
+ if (!is_dataset_expr(s))
+ fail;
+ repeat_variables = repeat_variables.cons(v);
+ return #<Repeat(lambda(`v,`(makePlan(b))),`(makePlan(s)),
+ `(makePlan(n)))>;
+ case repeat(lambda(`v,`b),`s):
+ if (!is_dataset_expr(s))
+ fail;
+ repeat_variables = repeat_variables.cons(v);
+ return #<Repeat(lambda(`v,`(makePlan(b))),`(makePlan(s)),`(Integer.MAX_VALUE))>;
+ case closure(lambda(`v,`b),`s,`n):
+ if (!is_dataset_expr(s))
+ fail;
+ repeat_variables = repeat_variables.cons(v);
+ return #<Closure(lambda(`v,`(makePlan(b))),`(makePlan(s)),
+ `(makePlan(n)))>;
+ case closure(lambda(`v,`b),`s):
+ if (!is_dataset_expr(s))
+ fail;
+ repeat_variables = repeat_variables.cons(v);
+ return #<Closure(lambda(`v,`(makePlan(b))),`(makePlan(s)),`(Integer.MAX_VALUE))>;
+ case loop(lambda(tuple(...vs),`b),`s,`n):
+ repeat_variables = repeat_variables.append(vs);
+ return #<Loop(lambda(tuple(...vs),`(makePlan(b))),`(makePlan(s)),`(makePlan(n)))>;
+ case record(...bl):
+ Trees el = #[];
+ for ( Tree b: bl )
+ match b {
+ case bind(_,`c):
+ el = el.append(c);
+ };
+ return makePlan(#<tuple(...el)>);
+ case project(`x,`a):
+ Tree tp = TypeInference.type_inference(x);
+ match tp {
+ case XML:
+ return makePlan(#<call(XMLchildren,`(new StringLeaf(a.toString())),`x)>);
+ case `S(XML):
+ if (is_collection(S))
+ return makePlan(#<call(XMLchildren,`(new StringLeaf(a.toString())),`x)>);
+ };
+ match TypeInference.expand(tp) {
+ case record(...bl):
+ int i = 0;
+ for ( Tree b: bl )
+ match b {
+ case bind(`c,_):
+ if (a.equals(c))
+ return makePlan(#<nth(`x,`i)>);
+ else fail
+ case _: i++;
+ };
+ case union(...tl):
+ for ( Tree t: tl )
+ match TypeInference.expand(t) {
+ case _(record(...bl)):
+ int i = 0;
+ for ( Tree b: bl )
+ match b {
+ case bind(`c,_):
+ if (a.equals(c))
+ return makePlan(#<nth(union_value(`x),`i)>);
+ else fail
+ case _: i++;
+ };
+ case _(bag(tuple(string,`tv))):
+ return #<map_index(`(makePlan(#<union_value(`x)>)),
+ `(new StringLeaf(a.toString())))>;
+ case `tt: error("wrong projection: "+e+" ("+tt+")");
+ };
+ case `S(`ttp):
+ if (!is_collection(S))
+ fail;
+ match TypeInference.expand(ttp) {
+ case tuple(string,`tv):
+ return #<map_index(`(makePlan(x)),
+ `(new StringLeaf(a.toString())))>;
+ case record(...bl):
+ Tree nv = new_var();
+ int i = 0;
+ for ( Tree b: bl )
+ match b {
+ case bind(`c,_):
+ if (a.equals(c))
+ return makePlan(#<cmap(lambda(`nv,nth(`nv,`i)),`x)>);
+ else fail
+ case _: i++;
+ };
+ case union(...tl):
+ Tree nv = new_var();
+ int j = 0;
+ for ( Tree t: tl ) {
+ match t {
+ case `c(record(...bl)):
+ int i = 0;
+ for ( Tree b: bl )
+ match b {
+ case bind(`w,_):
+ if (a.equals(w))
+ return makePlan(#<cmap(lambda(`nv,if(call(eq,union_tag(`nv),`j),
+ bag(nth(union_value(`nv),`i)),
+ bag())),`x)>);
+ else fail
+ case _: i++;
+ };
+ case `c(bag(tuple(string,`tv))):
+ return makePlan(#<cmap(lambda(`nv,if(call(eq,union_tag(`nv),`j),
+ bag(map_index(union_value(`nv),
+ `(new StringLeaf(a.toString())))),
+ bag())),`x)>);
+ case `tt: error("wrong projection: "+e+" ("+tt+")");
+ };
+ j++;
+ }
+ };
+ case `t: error("wrong projection: "+e+" ("+t+")");
+ };
+ case typed(`x,`tp):
+ if (tp.is_variable() && !tp.equals(#<string>) && MRContainer.type_code(tp.toString()) >= 0)
+ return makePlan(#<call(coerce,`x,
+ `(MRContainer.type_code(tp.toString())))>);
+ else fail
+ case index(`x,`n):
+ match TypeInference.type_inference2(x) {
+ case bag(tuple(`kt,`vt)):
+ if (TypeInference.type_inference2(n).equals(kt))
+ return #<map_index(`(makePlan(x)),
+ `(makePlan(n)))>;
+ case Bag(tuple(`kt,`vt)):
+ if (TypeInference.type_inference2(n).equals(kt))
+ return #<map_index(`(makePlan(x)),
+ `(makePlan(n)))>;
+ case union(...tl):
+ for ( Tree t: tl )
+ match TypeInference.expand(t) {
+ case _(bag(tuple(`kt,`vt))):
+ if (TypeInference.type_inference2(n).equals(kt))
+ return #<map_index(`(makePlan(#<union_value(`x)>)),
+ `(makePlan(n)))>;
+ case _(list(`tp)):
+ return #<index(`(makePlan(#<union_value(`x)>)),
+ `(makePlan(n)))>;
+ case `tt: error("wrong indexing: "+e+" ("+tt+")");
+ };
+ };
+ return #<index(`(makePlan(x)),
+ `(makePlan(n)))>;
+ case call(count,mapReduce(`m,lambda(`vr,`br),`X,`o)):
+ Tree nv = new_var();
+ type_env.insert(nv.toString(),TypeInference.type_inference(vr));
+ Tree nr = simplify(#<lambda(`nv,bag(call(count,`(subst(vr,nv,br)))))>);
+ Tree plan = #<call(sum,mapReduce(`m,`nr,`X,false))>;
+ return makePlan(plan);
+ case call(`f,`u):
+ for ( Tree monoid: monoids )
+ match monoid {
+ case `aggr(`mtp,`plus,`zero,`unit):
+ if (aggr.equals(f.toString())) {
+ Tree plan = makePlan(u);
+ Tree nx = new_var();
+ Tree np = new_var();
+ Tree na = new_var();
+ Tree tp = TypeInference.type_inference(e);
+ type_env.insert(np.toString(),#<tuple(`tp,`tp)>);
+ match TypeInference.type_inference(u) {
+ case `T(`t):
+ if (TypeInference.unify(mtp,t) == null)
+ continue;
+ type_env.insert(na.toString(),#<tuple(`tp,`t)>);
+ type_env.insert(nx.toString(),t);
+ };
+ plus = makePlan(simplify_all(#<lambda(`np,apply(`plus,`np))>));
+ Tree acc = makePlan(simplify_all(#<lambda(`na,apply(`plus,tuple(nth(`na,0),
+ apply(`unit,nth(`na,1)))))>));
+ zero = makePlan((f.equals(#<avg>)) ? zero : #<typed(`zero,`tp)>);
+ match plan {
+ case MapCombineReduce(`m,`c,`r,`s,_):
+ plan = #<MapAggregateReduce(`m,`r,`acc,`zero,`s,false)>;
+ case MapReduce(`m,`r,`s,_):
+ plan = #<MapAggregateReduce(`m,`r,`acc,`zero,`s,false)>;
+ case MapReduce2(`mx,`my,`r,`x,`y,_):
+ plan = #<MapAggregateReduce2(`mx,`my,`r,`acc,`zero,`x,`y,false)>;
+ case MapJoin(`kx,`ky,`r,`x,`y):
+ plan = #<MapAggregateJoin(`kx,`ky,`r,`acc,`zero,`x,`y)>;
+ case CrossProduct(`mx,`my,`r,`x,`y):
+ plan = #<CrossAggregateProduct(`mx,`my,`r,`acc,`zero,`x,`y)>;
+ case cMap(`m,`s):
+ plan = #<AggregateMap(`m,`acc,`zero,`s)>;
+ case _:
+ if (is_dataset_expr(u))
+ plan = #<AggregateMap(lambda(`nx,bag(`nx)),`acc,`zero,`plan)>;
+ else return #<aggregate(`acc,`zero,`plan)>;
+ };
+ if (is_dataset_expr(u))
+ return #<Aggregate(`plus,`zero,`plan)>;
+ else return #<aggregate(`plus,`zero,`plan)>;
+ }
+ };
+ fail
+ case call(plus,`x,`y):
+ if (!is_dataset_expr(x) || !is_dataset_expr(y))
+ fail;
+ return #<Merge(`(makePlan(x)),`(makePlan(y)))>;
+ case call(`f,...el):
+ if (!f.is_variable())
+ fail;
+ Tree ret = data_constructors.lookup(f.toString());
+ if (ret != null)
+ match ret {
+ case `v(`n,`tp):
+ Tree p = (el.length()==1)
+ ? makePlan(el.head())
+ : makePlan(#<tuple(...el)>);
+ return #<tagged_union(`n,`p)>;
+ };
+ ret = global_type_env.lookup(f.toString());
+ if (ret != null)
+ match ret {
+ case arrow(_,_):
+ Trees args = #[];
+ for ( Tree a: el )
+ args = args.append(makePlan(a));
+ return #<apply(`f,tuple(...args))>;
+ };
+ Trees tps = #[];
+ for ( Tree a: el )
+ tps = tps.append(TypeInference.type_inference(a));
+ int i = ClassImporter.find_method_number(f.toString(),tps);
+ if (i < 0)
+ error("Method "+f+tps+" has no implementation");
+ Trees sig = ClassImporter.signature(i);
+ Trees args = #[];
+ for ( int j = 0; j < el.length(); j++ ) {
+ Tree b = sig.nth(j+1);
+ if (f.equals(#<coerce>) || b.equals(tps.nth(j)) || !b.is_variable()
+ || b.equals(#<union>) || MRContainer.type_code(b.toString()) < 0)
+ args = args.append(makePlan(el.nth(j)));
+ else args = args.append(makePlan(#<call(coerce,`(el.nth(j)),
+ `(MRContainer.type_code(b.toString())))>));
+ };
+ return #<callM(`f,`i,...args)>;
+ case let(`v,`u,`body):
+ if (true)
+ fail; // disable
+ body = makePlan(body);
+ match TypeInference.type_inference(u) {
+ case `S(_):
+ // if a variable bound to a collection is used more than once in the body,
+ // materialize the collection in memory
+ if (is_collection(S) && occurences(v,body) > 1)
+ body = #<let(`v,`(makePlan(#<call(materialize,`v)>)),`body)>;
+ };
+ return #<let(`v,`(makePlan(u)),`body)>;
+ case function(tuple(...params),`outp,`body):
+ boolean is_dataset = false;
+ for ( Tree p: params )
+ match p {
+ case dataset(`v,`tp):
+ is_dataset = true;
+ };
+ body = makePlan(body);
+ return #<function(tuple(...params),`outp,`body)>;
+ case `f(...al):
+ Trees bl = #[];
+ for ( Tree a: al )
+ bl = bl.append(makePlan(a));
+ return #<`f(...bl)>;
+ };
+ return e;
+ }
+
+ /** convert an algebraic plan to a physical plan
+ * @param plan algebraic plan
+ * @return the physical plan
+ */
+ public static Tree physical_plan ( Tree plan ) {
+ match plan {
+ case MapReduce2(`mx,`my,`r,`x,`y,`o):
+ return physical_plan(#<MapAggregateReduce2(`mx,`my,`r,null,null,`x,`y,`o)>);
+ // convert a reduce-side join to a fragment-replicate join, if either of the join
+ // inputs is small to fit in memory (dataset sizes are extracted from file statistics)
+ case MapAggregateReduce2(`mx,`my,
+ lambda(`v,cmap(lambda(`vx,cmap(lambda(`vy,`b),nth(`nx,1))),nth(`ny,0))),
+ null,null,`x,`y,false):
+ if (Config.noMapJoin)
+ fail;
+ if (!nx.equals(v) || !ny.equals(v) || occurences(v,b) > 0)
+ fail;
+ Tree X = new_var();
+ Tree Y = new_var();
+ Tree nv = new_var();
+ Tree L1 = simplify_all(#<lambda(`nv,cmap(lambda(`vx,cmap(lambda(`vy,`b),nth(`nv,1))),bag(nth(`nv,0))))>);
+ nv = new_var();
+ Tree L2 = simplify_all(#<lambda(`nv,cmap(lambda(`vx,cmap(lambda(`vy,`b),bag(nth(`nv,0)))),nth(`nv,1)))>);
+ nv = new_var();
+ Tree L3 = #<lambda(`nv,cmap(lambda(`vx,cmap(lambda(`vy,`b),nth(`nv,0))),nth(`nv,1)))>;
+ Tree cond1 = makePlan(#<call(leq,dataset_size(`Y),`(Config.mapjoin_size))>);
+ Tree cond2 = makePlan(#<call(leq,dataset_size(`X),`(Config.mapjoin_size))>);
+ Tree cond3 = makePlan(#<call(lt,dataset_size(`X),dataset_size(`Y))>);
+ return #<Let(`X,`(physical_plan(x)),Let(`Y,`(physical_plan(y)),
+ If(`cond1,
+ MapJoin(`mx,`my,`L1,`X,`Y),
+ If(`cond2,
+ MapJoin(`my,`mx,`L2,`Y,`X),
+ If(`cond3,
+ MapAggregateReduce2(`my,`mx,`L3,null,null,`Y,`X,false),
+ MapAggregateReduce2(`mx,`my,
+ lambda(`v,cmap(lambda(`vx,cmap(lambda(`vy,`b),nth(`nx,1))),
+ nth(`ny,0))),
+ null,null,`X,`Y,false))))))>;
+ case MapAggregateReduce2(`mx,`my,lambda(`v,`b),null,null,`x,`y,false):
+ if (Config.noMapJoin)
+ fail;
+ Tree X = new_var();
+ Tree Y = new_var();
+ Tree nv = new_var();
+ Tree L1 = simplify_all(subst(#<nth(`v,0)>,#<bag(nth(`nv,0))>,
+ subst(#<nth(`v,1)>,#<nth(`nv,1)>,
+ #<lambda(`nv,`b)>)));
+ nv = new_var();
+ Tree L2 = subst(#<nth(`v,0)>,#<nth(`nv,1)>,
+ subst(#<nth(`v,1)>,#<nth(`nv,0)>,
+ #<lambda(`nv,`b)>));
+ Tree cond1 = makePlan(#<call(leq,dataset_size(`Y),`(Config.mapjoin_size))>);
+ Tree cond2 = makePlan(#<call(lt,dataset_size(`X),dataset_size(`Y))>);
+ return #<Let(`X,`(physical_plan(x)),Let(`Y,`(physical_plan(y)),
+ If(`cond1,
+ MapJoin(`mx,`my,`L1,`X,`Y),
+ If(`cond2,
+ MapAggregateReduce2(`my,`mx,`L2,null,null,`Y,`X,false),
+ MapAggregateReduce2(`mx,`my,lambda(`v,`b),null,null,`X,`Y,false)))))>;
+ case MapAggregateReduce2(`mx,`my,lambda(`v,`b),`acc,`zero,`x,`y,false):
+ Tree X = new_var();
+ Tree Y = new_var();
+ Tree nv = new_var();
+ Tree nr = subst(#<nth(`v,0)>,#<nth(`nv,1)>,
+ subst(#<nth(`v,1)>,#<nth(`nv,0)>,b));
+ if (!streamed_MapReduce2_reducer(#<lambda(`nv,`nr)>))
+ fail;
+ Tree cond = makePlan(#<call(lt,dataset_size(`X),dataset_size(`Y))>);
+ return #<Let(`X,`(physical_plan(x)),Let(`Y,`(physical_plan(y)),
+ If(`cond,
+ MapAggregateReduce2(`my,`mx,lambda(`nv,`nr),`acc,`zero,`Y,`X,false),
+ MapAggregateReduce2(`mx,`my,lambda(`v,`b),`acc,`zero,`X,`Y,false))))>;
+ case CrossProduct(`mx,`my,`r,`x,`y):
+ return physical_plan(#<CrossAggregateProduct(`mx,`my,`r,null,null,`x,`y)>);
+ case CrossAggregateProduct(`mx,`my,lambda(`v,`b),`acc,`zero,`x,`y):
+ Tree X = new_var();
+ Tree Y = new_var();
+ Tree nv = new_var();
+ Tree nr = subst(#<nth(`v,0)>,#<nth(`nv,1)>,
+ subst(#<nth(`v,1)>,#<nth(`nv,0)>,b));
+ Tree cond = makePlan(#<call(lt,dataset_size(`X),dataset_size(`Y))>);
+ return #<Let(`X,`(physical_plan(x)),Let(`Y,`(physical_plan(y)),
+ If(`cond,
+ CrossAggregateProduct(`my,`mx,lambda(`nv,`nr),`acc,`zero,`Y,`X),
+ CrossAggregateProduct(`mx,`my,lambda(`v,`b),`acc,`zero,`X,`Y))))>;
+ case `f(...al):
+ Trees bl = #[];
+ for ( Tree a: al )
+ bl = bl.append(physical_plan(a));
+ return #<`f(...bl)>;
+ };
+ return plan;
+ }
+}