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 2013/08/27 14:27:53 UTC
[15/21] MRQL-2: Mavenize MRQL
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/Normalization.gen
----------------------------------------------------------------------
diff --git a/src/Normalization.gen b/src/Normalization.gen
deleted file mode 100644
index e98ec06..0000000
--- a/src/Normalization.gen
+++ /dev/null
@@ -1,406 +0,0 @@
-/**
- * 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 */
- 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 */
- 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/f5e66d20/src/ParsedDataSource.java
----------------------------------------------------------------------
diff --git a/src/ParsedDataSource.java b/src/ParsedDataSource.java
deleted file mode 100644
index ce5220b..0000000
--- a/src/ParsedDataSource.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/**
- * 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,ParsedInputFormat.class,conf);
- this.parser = parser;
- this.args = args;
- }
-
- ParsedDataSource ( String path,
- Class<? extends Parser> parser,
- Trees args,
- Configuration conf ) {
- super(-1,path,ParsedInputFormat.class,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/f5e66d20/src/Parser.java
----------------------------------------------------------------------
diff --git a/src/Parser.java b/src/Parser.java
deleted file mode 100644
index c1728f1..0000000
--- a/src/Parser.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/**
- * 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/f5e66d20/src/Plan.java
----------------------------------------------------------------------
diff --git a/src/Plan.java b/src/Plan.java
deleted file mode 100644
index bb2e424..0000000
--- a/src/Plan.java
+++ /dev/null
@@ -1,390 +0,0 @@
-/**
- * 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 and BSP 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
- : "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 MRQLFileInputFormat.collect(x,strip);
- }
-
- /** The collect physical operator */
- public final static Bag collect ( final DataSet x ) throws Exception {
- return MRQLFileInputFormat.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));
- }
-
- /** dump both the MRQL data and their MRQL type into files */
- public final static void dump ( String file, Tree type, MRData data ) throws Exception {
- Path path = new Path(file);
- FileSystem fs = path.getFileSystem(conf);
- PrintStream ftp = new PrintStream(fs.create(path.suffix(".type")));
- ftp.print("2@"+type.toString()+"\n");
- ftp.close();
- SequenceFile.Writer writer
- = new SequenceFile.Writer(fs,conf,path,
- MRContainer.class,MRContainer.class);
- if (data instanceof MR_dataset)
- data = Plan.collect(((MR_dataset)data).dataset());
- if (data instanceof Bag) {
- Bag s = (Bag)data;
- long i = 0;
- for ( MRData e: s ) {
- counter_key.set(i++);
- value_container.set(e);
- writer.append(counter_container,value_container);
- }
- } else {
- counter_key.set(0);
- value_container.set(data);
- writer.append(counter_container,value_container);
- };
- writer.close();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/PlanGeneration.gen
----------------------------------------------------------------------
diff --git a/src/PlanGeneration.gen b/src/PlanGeneration.gen
deleted file mode 100644
index fd80e44..0000000
--- a/src/PlanGeneration.gen
+++ /dev/null
@@ -1,718 +0,0 @@
-/**
- * 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;
- match TypeInference.type_inference2(x) {
- case `T(_):
- if (is_collection(T))
- return #<Merge(`(makePlan(x)),
- `(makePlan(y)))>;
- };
- fail
- 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 (!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):
- 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;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/Printer.gen
----------------------------------------------------------------------
diff --git a/src/Printer.gen b/src/Printer.gen
deleted file mode 100644
index 79f4460..0000000
--- a/src/Printer.gen
+++ /dev/null
@@ -1,466 +0,0 @@
-/**
- * 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.Iterator;
-
-
-/** printers for types, expressions, plans, etc */
-public class Printer {
-
- public static String print_type ( Tree tp ) {
- match tp {
- case tuple(...tl):
- if (tl.is_empty())
- return "()";
- String s = "( "+print_type(tl.head());
- for ( Tree t: tl.tail() )
- s += ", "+print_type(t);
- return s+" )";
- case record(...tl):
- if (tl.is_empty())
- return "< >";
- String s = "< ";
- match tl.head() {
- case bind(`a,`t):
- s += a+": "+print_type(t);
- };
- for ( Tree t: tl.tail() )
- match t {
- case bind(`a,`at):
- s += ", "+a+": "+print_type(at);
- };
- return s+" >";
- case arrow(`itp,`otp):
- return print_type(itp)+" -> "+print_type(otp);
- case persistent(`t):
- return "!"+print_type(t);
- case `f():
- return f+"()";
- case persistent(`t):
- return "!"+print_type(t);
- case Bag(`etp):
- return "!bag("+print_type(etp)+")";
- case List(`etp):
- return "!list("+print_type(etp)+")";
- case `f(...tl):
- String s = f+"( "+print_type(tl.head());
- for ( Tree t: tl.tail() )
- s += ", "+print_type(t);
- return s+" )";
- };
- return tp.toString();
- }
-
- private static String print_query_list ( Trees el ) {
- if (el.length() == 0)
- return "";
- String s = " ";
- s += print_query(el.head());
- for ( Tree a: el.tail() )
- s += ", "+print_query(a);
- return s+" ";
- }
-
- public static String print_query ( Tree e ) {
- match e {
- case select(`opt_dist,`u,from(...bl),where(`c),groupby(...gs),orderby(...os)):
- String s = "select "+(opt_dist.equals(#<none>) ? "" : "distinct ");
- s += print_query(u)+" from ";
- match bl.head() {
- case bind(`p,`d):
- s += print_query(p)+" in "+print_query(d);
- };
- for ( Tree b: bl.tail() )
- match b {
- case bind(`p,`d):
- s += ", "+print_query(p)+" in "+print_query(d);
- };
- if (!c.equals(#<true>))
- s += " where "+print_query(c);
- match #<groupby(...gs)> {
- case groupby(`h,...gl):
- s += " group by ";
- match gl.head() {
- case bind(`gp,`gd):
- s += print_query(gp)+": "+print_query(gd);
- };
- for ( Tree g: gl.tail() )
- match g {
- case bind(`gp,`gd):
- s += ", "+print_query(gp)+": "+print_query(gd);
- };
- if (!h.equals(#<true>))
- s += " having "+print_query(h);
- };
- match #<orderby(...os)> {
- case orderby(`l,...ol):
- s += " order by "+print_query(ol.length() == 1 ? ol.head() : #<tuple(...ol)>);
- if (!l.equals(#<none>))
- s += " limit "+print_query(l);
- };
- return s;
- case tuple(...el):
- return "("+print_query_list(el)+")";
- case record(...el):
- String s = "< ";
- match el.head() {
- case bind(`v,`b):
- s += v+": "+print_query(b);
- };
- for ( Tree a: el.tail() )
- match a {
- case bind(`v,`b):
- s += ", "+v+": "+print_query(b);
- };
- return s+" >";
- case project(`a,`v):
- return (a.is_variable()) ? print_query(a)+"."+v : "("+print_query(a)+")."+v;
- case index(`a,`i):
- return (a.is_variable()) ? print_query(a)+"["+print_query(i)+"]"
- : "("+print_query(a)+")["+print_query(i)+"]";
- case nth(`x,`n):
- return (x.is_variable()) ? print_query(x)+"#"+print_query(n)
- : "("+print_query(x)+")#"+print_query(n);
- case call(`f,...el):
- return f+"("+print_query_list(el)+")";
- case list(...el):
- return "["+print_query_list(el)+"]";
- case bag(...el):
- return "{"+print_query_list(el)+"}";
- case `f(...el):
- return f+"("+print_query_list(el)+")";
- };
- return e.toString();
- }
-
- private static String print_XML ( final Union x ) {
- if (x.tag() == 1)
- return ((MR_string)x.value()).get();
- Tuple t = (Tuple)x.value();
- String s = "<"+((MR_string)t.get(0)).get();
- for ( MRData a: (Bag)t.get(1) ) {
- Tuple attr = (Tuple)a;
- s += " "+((MR_string)attr.first()).get()+"=\""
- +((MR_string)attr.second()).get()+"\"";
- };
- Bag c = (Bag)t.get(2);
- if (c.size() == 0)
- return s+"/>";
- s += ">";
- for ( MRData e: c )
- s += print_XML((Union)e);
- return s+"</"+((MR_string)t.get(0)).get()+">";
- }
-
- private static String print_JSON ( final Union x ) {
- switch (x.tag()) {
- case 0:
- String s = "{ ";
- for ( MRData e: (Bag)x.value() ) {
- Tuple t = (Tuple)e;
- s += t.get(0)+": "+print_JSON((Union)t.get(1))+", ";
- };
- return s.substring(0,s.length()-2)+" }";
- case 1:
- String q = "[ ";
- for ( MRData e: (Bag)x.value() )
- q += print_JSON((Union)e)+", ";
- return q.substring(0,q.length()-2)+" ]";
- };
- return ""+x.value();
- }
-
- /** An MRData printer based on type information */
- final static String print ( final MRData x, final Tree type ) {
- try {
- if (x instanceof Inv)
- return print(((Inv)x).value(),type);
- if (type.equals(#<XML>))
- return print_XML((Union)x);
- if (type.equals(#<JSON>))
- return print_JSON((Union)x);
- match TypeInference.expand(type) {
- case persistent(`tp):
- return print(x,tp);
- case Bag(`tp):
- if (x instanceof MR_dataset) {
- DataSet ds = ((MR_dataset)x).dataset();
- return print(Plan.collect(ds),#<bag(`tp)>);
- } else return print(x,#<bag(`tp)>);
- case List(`tp):
- if (x instanceof MR_dataset) {
- DataSet ds = ((MR_dataset)x).dataset();
- return print(Plan.collect(ds),#<list(`tp)>);
- } else return print(x,#<list(`tp)>);
- case bag(`tp):
- Bag b = (Bag)x;
- Iterator<MRData> bi = b.iterator();
- if (!bi.hasNext())
- return "{}";
- String s = "{ "+print(bi.next(),tp);
- for ( long i = 1; bi.hasNext() && (Config.max_bag_size_print < 0
- || i < Config.max_bag_size_print); i++ )
- s += ", "+print(bi.next(),tp);
- if (bi.hasNext())
- return s+", ... }";
- else return s+" }";
- case list(`tp):
- Bag b = (Bag)x;
- Iterator<MRData> bi = b.iterator();
- if (!bi.hasNext())
- return "[]";
- String s = "[ "+print(bi.next(),tp);
- for ( long i = 1; bi.hasNext() && (Config.max_bag_size_print < 0
- || i < Config.max_bag_size_print); i++ )
- s += ", "+print(bi.next(),tp);
- if (bi.hasNext())
- return s+", ... ]";
- else return s+" ]";
- case tuple(...el):
- Tuple t = (Tuple)x;
- if (t.size() == 0)
- return "()";
- String s = "("+print(t.get((short)0),el.nth(0));
- for ( short i = 1; i < t.size(); i++ )
- s += ","+print(t.get(i),el.nth(i));
- return s+")";
- case record(...el):
- Tuple t = (Tuple)x;
- if (t.size() == 0)
- return "<>";
- String s = "< ";
- match el.nth(0) {
- case bind(`a,`tp):
- s += a+": "+print(t.get((short)0),tp);
- };
- for ( short i = 1; i < t.size(); i++ )
- match el.nth(i) {
- case bind(`a,`tp):
- s += ", "+a+": "+print(t.get(i),tp);
- };
- return s+" >";
- case union(...el):
- Union u = (Union)x;
- match el.nth(u.tag()) {
- case `c(tuple(...ts)):
- return c+print(u.value(),#<tuple(...ts)>);
- case `c(`tp):
- return c+"("+print(u.value(),tp)+")";
- }
- };
- return x.toString();
- } catch (Exception ex) {
- throw new Error(ex);
- }
- }
-
- private final static String tab ( int n ) {
- String s = "";
- for ( int i = 0; i < n; i++ )
- s += " ";
- return s;
- }
-
- /** print a physical plan
- * @param e the plan
- * @param n tab (# of spaces to put in the beginning of the line)
- * @param pv is this a variable bound to physical plan?
- * @return the printout
- */
- public final static String print_plan ( Tree e, int n, boolean pv ) {
- match e {
- case cMap(`f,`s):
- return "cMap:\n"+tab(n+3)+"input: "+print_plan(s,n+10,true);
- case AggregateMap(`f,`a,`z,`s):
- return "AggregateMap:\n"+tab(n+3)+"input: "+print_plan(s,n+10,true);
- case MapReduce(`m,`r,`s,_):
- return "MapReduce:\n"+tab(n+3)+"input: "+print_plan(s,n+10,true);
- case MapCombineReduce(`m,`c,`r,`s,_):
- return "MapCombineReduce:\n"+tab(n+3)+"input: "+print_plan(s,n+10,true);
- case MapAggregateReduce(`m,`r,`a,`z,`s,_):
- return "MapAggregateReduce:\n"+tab(n+3)+"input: "+print_plan(s,n+10,true);
- case MapReduce2(`mx,`my,`r,`x,`y,_):
- return "MapReduce2:\n"+tab(n+3)+"left: "+print_plan(x,n+9,true)+"\n"
- +tab(n+3)+"right: "+print_plan(y,n+10,true);
- case MapCombineReduce2(`mx,`my,`c,`r,`x,`y,_):
- return "MapCombineReduce2:\n"+tab(n+3)+"left: "+print_plan(x,n+9,true)+"\n"
- +tab(n+3)+"right: "+print_plan(y,n+10,true);
- case MapAggregateReduce2(`mx,`my,`r,`a,null,`x,`y,...):
- return "MapReduce2:\n"+tab(n+3)+"left: "+print_plan(x,n+9,true)+"\n"
- +tab(n+3)+"right: "+print_plan(y,n+10,true);
- case MapAggregateReduce2(`mx,`my,`r,`a,`z,`x,`y,...):
- return "MapAggregateReduce2:\n"+tab(n+3)+"left: "+print_plan(x,n+9,true)+"\n"
- +tab(n+3)+"right: "+print_plan(y,n+10,true);
- case MapJoin(`kx,`ky,`r,`x,`y):
- return "MapJoin:\n"+tab(n+3)+"left: "+print_plan(x,n+9,true)+"\n"
- +tab(n+3)+"right: "+print_plan(y,n+10,true);
- case MapAggregateJoin(`kx,`ky,`r,`a,null,`x,`y):
- return "MapJoin:\n"+tab(n+3)+"left: "+print_plan(x,n+9,true)+"\n"
- +tab(n+3)+"right: "+print_plan(y,n+10,true);
- case MapAggregateJoin(`kx,`ky,`r,`a,`z,`x,`y):
- return "MapAggregateJoin:\n"+tab(n+3)+"left: "+print_plan(x,n+9,true)+"\n"
- +tab(n+3)+"right: "+print_plan(y,n+10,true);
- case GroupByJoin(`kx,`ky,`gx,`gy,`m,`c,`r,`x,`y,_):
- return "GroupByJoin:\n"+tab(n+3)+"left: "+print_plan(x,n+9,true)+"\n"
- +tab(n+3)+"right: "+print_plan(y,n+10,true);
- case CrossProduct(`mx,`my,`r,`x,`y):
- return "CrossProduct:\n"+tab(n+3)+"left: "+print_plan(x,n+9,true)+"\n"
- +tab(n+3)+"right: "+print_plan(y,n+10,true);
- case CrossAggregateProduct(`mx,`my,`r,`a,null,`x,`y):
- return "CrossProduct:\n"+tab(n+3)+"left: "+print_plan(x,n+9,true)+"\n"
- +tab(n+3)+"right: "+print_plan(y,n+10,true);
- case CrossAggregateProduct(`mx,`my,`r,`a,`z,`x,`y):
- return "CrossAggregateProduct:\n"+tab(n+3)+"left: "+print_plan(x,n+9,true)+"\n"
- +tab(n+3)+"right: "+print_plan(y,n+10,true);
- case Aggregate(`a,`z,`s):
- return "Aggregate:\n"+tab(n+3)+"input: "+print_plan(s,n+10,true);
- case BinarySource(`k,`file,_):
- return "Source (binary): "+file;
- case BinarySource(`file,_):
- return "Source (binary): "+file;
- case ParsedSource(`m,`parser,`file,...args):
- if (m instanceof LongLeaf)
- return "Source ("+parser+"): "+file;
- else fail
- case ParsedSource(`parser,`file,...args):
- return "Source ("+parser+"): "+file;
- case Generator(...):
- return "Generator";
- case Merge(`x,`y):
- return "Merge:\n"+tab(n+3)+"left: "+print_plan(x,n+9,true)+"\n"
- +tab(n+3)+"right: "+print_plan(y,n+10,true);
- case BSP(_,_,_,_,...ds):
- String ret = "BSP:\n";
- for ( Tree d: ds )
- ret += tab(n+3)+"input: "+print_plan(d,n+10,true);
- return ret;
- case Loop(lambda(tuple(...vs),tuple(...bs)),tuple(...ss),...):
- String ret = "Loop ("+vs+"):\n"+tab(n+3)+"init: "
- +vs.head()+" = "+print_plan(ss.head(),n+vs.head().toString().length()+12,true)+"\n";
- ss = ss.tail();
- for ( Trees s = vs.tail(); !s.is_empty() && !ss.is_empty(); s = s.tail(), ss = ss.tail() )
- ret += tab(n+9)+s.head()+" = "+print_plan(ss.head(),n+s.head().toString().length()+12,true)+"\n";
- ret += tab(n+3)+"step: "
- +vs.head()+" = "+print_plan(bs.head(),n+vs.head().toString().length()+12,true)+"\n";
- bs = bs.tail();
- for ( Trees s = vs.tail(); !s.is_empty() && !bs.is_empty(); s = s.tail(), bs = bs.tail() )
- ret += tab(n+9)+s.head()+" = "+print_plan(bs.head(),n+s.head().toString().length()+12,true)+"\n";
- return ret;
- case `f(lambda(`v,`b),`s,...):
- if (! #[Repeat,repeat,Closure,closure].member(#<`f>))
- fail;
- return f+" ("+v+"):\n"+tab(n+3)+"init: "+print_plan(s,n+9,true)+"\n"
- +tab(n+3)+"step: "+print_plan(b,n+9,true);
- case Let(`v,`u,`body):
- return "let "+v+" = "+print_plan(u,n+10+v.toString().length(),pv)+"\n"
- +tab(n)+print_plan(body,n,pv);
- case If(_,`x1,If(_,`x2,If(_,`x3,`x4))):
- return "Choice 1: "+print_plan(x1,n+10,pv)+"\n"
- +tab(n)+"Choice 2: "+print_plan(x2,n+10,pv)+"\n"
- +tab(n)+"Choice 3: "+print_plan(x3,n+10,pv)+"\n"
- +tab(n)+"Choice 4: "+print_plan(x4,n+10,pv);
- case If(_,`x1,If(_,`x2,`x3)):
- return "Choice 1: "+print_plan(x1,n+10,pv)+"\n"
- +tab(n)+"Choice 2: "+print_plan(x2,n+10,pv)+"\n"
- +tab(n)+"Choice 3: "+print_plan(x3,n+10,pv);
- case If(`c,`x,`y):
- return "Choice 1: "+print_plan(x,n+10,pv)+"\n"
- +tab(n)+"Choice 2: "+print_plan(y,n+10,pv);
- case `f(...as):
- String s = "";
- for (Tree a: as) {
- String ps = print_plan(a,n,pv);
- if (!ps.equals("") && !a.is_variable())
- s += ps+(ps.endsWith("\n")?"":"\n");
- };
- return s;
- };
- if (pv && e.is_variable())
- return e.toString();
- return "";
- }
-
- /** given an MRData value, construct an expression that builds this data
- * @param x the MRData
- * @param type the type of x
- * @return an expression that constructs x
- */
- public final static Tree reify ( final MRData x, Tree type ) {
- if (x instanceof MR_variable)
- return new VariableLeaf("t_"+((MR_variable)x).var_num);
- type = TypeInference.expand(type);
- match type {
- case `T(`tp):
- if (!Translator.is_collection(T))
- fail;
- Bag b = (Bag)x;
- Trees as = #[];
- for ( MRData e: b)
- as = as.append(reify(e,tp));
- return #<`T(...as)>;
- case tuple(...el):
- Tuple t = (Tuple)x;
- Trees as = #[];
- for ( short i = 0; i < t.size(); i++ )
- as = as.append(reify(t.get(i),el.nth(i)));
- return #<tuple(...as)>;
- case record(...el):
- Tuple t = (Tuple)x;
- Trees as = #[];
- for ( short i = 0; i < t.size(); i++ )
- match el.nth(i) {
- case bind(`a,`tp):
- as = as.append(#<bind(`a,`(reify(t.get(i),tp)))>);
- };
- return #<record(...as)>;
- case union(...el):
- Union u = (Union)x;
- match el.nth(u.tag()) {
- case `c(tuple(...ts)):
- return #<call(`c,`(reify(u.value(),#<tuple(...ts)>)))>;
- case `c(`tp):
- return #<call(`c,`(reify(u.value(),tp)))>;
- };
- case string:
- String[] s = ((MR_string)x).get().split("\\x7B\\x7B");
- if (s.length == 1)
- return new StringLeaf(s[0]);
- Trees as = s[0].length() == 0 ? #[] : #[].append(new StringLeaf(s[0]));
- for ( int i = 1; i < s.length; i++ ) {
- String[] d = s[i].split("\\x7D\\x7D",2);
- if (d.length < 2)
- throw new Error("");
- as = as.append(new VariableLeaf("t_"+Integer.parseInt(d[0])));
- if (d[1].length() > 0)
- as = as.append(new StringLeaf(d[1]));
- };
- Tree res = as.reverse().head();
- for ( Tree a: as.reverse().tail() )
- res = #<call(plus,`a,`res)>;
- return res;
- case short: return #<typed(`(((MR_short)x).get()),`type)>;
- case int: return #<typed(`(((MR_int)x).get()),`type)>;
- case long: return #<typed(`((int)((MR_long)x).get()),`type)>;
- case float: return #<typed(`(((MR_float)x).get()),`type)>;
- case double: return #<typed(`((float)(((MR_double)x).get())),`type)>;
- };
- throw new Error("wrong type: "+type);
- }
-}