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:57 UTC
[19/21] MRQL-2: Mavenize MRQL
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/BSPTranslator.gen
----------------------------------------------------------------------
diff --git a/src/BSPTranslator.gen b/src/BSPTranslator.gen
deleted file mode 100644
index eadeabd..0000000
--- a/src/BSPTranslator.gen
+++ /dev/null
@@ -1,753 +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.*;
-
-
-/** Translates physical plans to BSP plans to be executed on Hama */
-public class BSPTranslator extends TypeInference {
- final static int orM = ClassImporter.find_method_number("or",#[bool,bool]);
- final static int andM = ClassImporter.find_method_number("and",#[bool,bool]);
- final static int notM = ClassImporter.find_method_number("not",#[bool]);
- final static int eqM = ClassImporter.find_method_number("eq",#[int,int]);
- final static int plusM = ClassImporter.find_method_number("plus",#[int,int]);
- final static int neqM = ClassImporter.find_method_number("neq",#[int,int]);
- final static int geqM = ClassImporter.find_method_number("geq",#[long,long]);
- final static int gtM = ClassImporter.find_method_number("gt",#[int,int]);
- final static int unionM = ClassImporter.find_method_number("union",#[bag(any),bag(any)]);
- final static int countM = ClassImporter.find_method_number("count",#[bag(any)]);
- final static int coerceM = ClassImporter.find_method_number("coerce",#[any,int]);
-
- final static Trees planNames = plans_with_distributed_lambdas
- .append(#[ParsedSource,Generator,BinarySource,GroupByJoin,Repeat,Closure,repeat]);
-
- private static int source_num = 0;
-
- public static void reset () {
- source_num = 0;
- }
-
- private static Tree new_source ( Tree e ) {
- match e {
- case `f(...):
- if (planNames.member(#<`f>))
- return preprocess(e);
- };
- return #<BSPSource(`(source_num++),`(preprocess(e)))>;
- }
-
- /** add a new source num to every BSP operation */
- private static Tree preprocess ( Tree e ) {
- match e {
- case lambda(`v,`b):
- return #<lambda(`v,`(preprocess(b)))>;
- case Aggregate(`acc,`zero,`S):
- return #<Aggregate(`acc,`zero,
- `(new_source(S)))>;
- case cMap(`m,`S):
- return #<cMap(`m,
- `(new_source(S)))>;
- case AggregateMap(`m,`acc,`zero,`S):
- return #<AggregateMap(`m,`acc,`zero,
- `(new_source(S)))>;
- case MapReduce(`m,`r,`S,`o):
- return #<MapReduce(`m,`r,
- `(new_source(S)),`o)>;
- case MapAggregateReduce(`m,`r,`acc,`zero,`S,`o):
- return #<MapAggregateReduce(`m,`r,`acc,`zero,
- `(new_source(S)),`o)>;
- case MapCombineReduce(`m,`c,`r,`S,`o):
- return #<MapCombineReduce(`m,`c,`r,
- `(new_source(S)),`o)>;
- case MapReduce2(`mx,`my,`r,`x,`y,`o):
- return #<MapReduce2(`mx,`my,`r,
- `(new_source(x)),
- `(new_source(y)),`o)>;
- case MapReduce2(`mx,`my,`c,`r,`x,`y,`o):
- return #<MapCombineReduce2(`mx,`my,`c,`r,
- `(new_source(x)),
- `(new_source(y)),`o)>;
- case MapAggregateReduce2(`mx,`my,`r,`acc,`zero,`x,`y,`o):
- return #<MapAggregateReduce2(`mx,`my,`r,`acc,`zero,
- `(new_source(x)),
- `(new_source(y)),`o)>;
- case MapJoin(`mx,`my,`r,`x,`y):
- return #<MapJoin(`mx,`my,`r,
- `(new_source(x)),
- `(new_source(y)))>;
- case MapAggregateJoin(`mx,`my,`r,`acc,`zero,`x,`y):
- return #<MapAggregateJoin(`mx,`my,`r,`acc,`zero,
- `(new_source(x)),
- `(new_source(y)))>;
- case CrossProduct(`mx,`my,`r,`x,`y):
- return #<CrossProduct(`mx,`my,`r,
- `(new_source(x)),
- `(new_source(y)))>;
- case CrossAggregateProduct(`mx,`my,`r,`acc,`zero,`x,`y):
- return #<CrossAggregateProduct(`mx,`my,`r,`acc,`zero,
- `(new_source(x)),
- `(new_source(y)))>;
- case GroupByJoin(`kx,`ky,`gx,`gy,`mr,`c,`r,`x,`y,`o):
- return #<GroupByJoin(`kx,`ky,`gx,`gy,`mr,`c,`r,
- `(new_source(x)),
- `(new_source(y)),`o)>;
- case Repeat(`f,`ds,`max):
- return #<Repeat(`(preprocess(f)),`(preprocess(ds)),`max)>;
- case repeat(`f,`ds,`max):
- return #<repeat(`(preprocess(f)),`(preprocess(ds)),`max)>;
- case Closure(`f,`ds,`max):
- return #<Closure(`(preprocess(f)),`(preprocess(ds)),`max)>;
- case `f(...):
- if (! #[ParsedSource,Generator,BinarySource].member(#<`f>))
- fail;
- return #<BSPSource(`(source_num++),`e)>;
- case `f(...as):
- Trees bs = #[];
- for (Tree a: as)
- bs = bs.append(preprocess(a));
- return #<`f(...bs)>;
- };
- return e;
- }
-
- /** returns the source num of a BSP operation */
- private static int source_num ( Tree e ) {
- match e {
- case BSP(`i,...):
- return (int)((LongLeaf)i).value();
- case BSPSource(`i,_):
- return (int)((LongLeaf)i).value();
- };
- return -1;
- }
-
- private static Tree subst_getCache_num ( int n, int m, Tree e ) {
- match e {
- case getCache(`cache,`k):
- if (!k.equals(#<`n>))
- fail;
- return #<getCache(`cache,`m)>;
- case `f(...as):
- Trees bs = #[];
- for (Tree a: as)
- bs = bs.append(subst_getCache_num(n,m,a));
- return #<`f(...bs)>;
- };
- return e;
- }
-
- private static Tree subst_setCache_num ( int n, int m, Tree e ) {
- match e {
- case setCache(`cache,`k,...r):
- if (!k.equals(#<`n>))
- fail;
- return #<setCache(`cache,`m,...r)>;
- case `f(...as):
- Trees bs = #[];
- for (Tree a: as)
- bs = bs.append(subst_setCache_num(n,m,a));
- return #<`f(...bs)>;
- };
- return e;
- }
-
- private static Tree set_cache_num ( Tree e, int n ) {
- match e {
- case BSP(`m,...r):
- return subst_setCache_num((int)((LongLeaf)m).value(),n,#<BSP(`n,...r)>);
- case BSPSource(_,`x):
- return #<BSPSource(`n,`x)>;
- };
- return e;
- }
-
- // return the source numbers of a BSP operation
- private static Trees source_nums ( Tree e ) {
- match e {
- case BSP(`i,...):
- return #[`i];
- case Merge(`x,`y):
- return source_nums(x).append(source_nums(y));
- case BSPSource(`i,_):
- return #[`i];
- };
- return #[];
- }
-
- private static Trees source_nums ( Trees s ) {
- Trees res = #[];
- for ( Tree x: s )
- res = res.append(source_nums(x));
- return res;
- }
-
- private static Tree getCache ( Tree var, Tree e, Tree body ) {
- Trees ns = source_nums(e);
- if (ns.length() > 0)
- return subst(var,#<getCache(cache,...ns)>,body);
- return body;
- }
-
- private static Tree getCache ( Tree var, Trees s, Tree body ) {
- Trees ns = source_nums(s);
- return subst(var,#<getCache(cache,...ns)>,body);
- }
-
- /** optimize a BSP plan after BSP fusion */
- private static Tree post_simplify ( Tree e ) {
- match e {
- case cmap(lambda(`x,`S(`a)),`b):
- if (is_collection(S))
- return post_simplify(#<map(lambda(`x,`a),`b)>);
- else fail
- case `f(...as):
- Trees bs = #[];
- for (Tree a: as)
- bs = bs.append(post_simplify(a));
- return #<`f(...bs)>;
- };
- return e;
- }
-
- private static Tree processBSP ( Tree e ) {
- return post_simplify(Simplification.simplify_all(Simplification.rename(e)));
- }
-
- private static Tree mkBSP ( Tree superstep, Tree state, Tree orderp, Tree input ) {
- int rn = source_num++;
- superstep = processBSP(subst(#<o_>,#<`rn>,getCache(#<i_>,input,superstep)));
- return #<BSP(`rn,`superstep,`state,`orderp,`input)>;
- }
-
- private static Tree mkBSPL ( Tree superstep, Tree state, Tree orderp, Trees input ) {
- int rn = source_num++;
- superstep = processBSP(subst(#<o_>,#<`rn>,getCache(#<i>,input,superstep)));
- return #<BSP(`rn,`superstep,`state,`orderp,...input)>;
- }
-
- private static Tree mkBSPL ( int[] ns, Tree superstep, Tree state, Tree orderp, Trees input ) {
- superstep = processBSP(subst(#<o_>,#<`(ns[0])>,getCache(#<i_>,input,superstep)));
- Trees s = #[];
- for ( int n: ns )
- s = s.append(#<`n>);
- return #<BSP(tuple(...s),`superstep,`state,`orderp,...input)>;
- }
-
- private static Tree mkBSPL ( int n, Tree superstep, Tree state, Tree orderp, Trees input ) {
- superstep = processBSP(subst(#<o_>,#<`n>,getCache(#<i_>,input,superstep)));
- return #<BSP(`n,`superstep,`state,`orderp,...input)>;
- }
-
- private static Tree mkBSP2 ( Tree superstep, Tree state, Tree orderp, Tree left, Tree right ) {
- int rn = source_num++;
- superstep = processBSP(subst(#<o_>,#<`rn>,getCache(#<i_>,left,
- getCache(#<j_>,right,superstep))));
- return #<BSP(`rn,`superstep,`state,`orderp,`left,`right)>;
- }
-
- /** construct a BSP plan from a physical plan
- * @param e the physical plan
- * @return the BSP plan
- */
- private static Tree mr2bsp ( Tree e ) {
- match e {
- case Aggregate(`acc,`zero,`S):
- return #<Aggregate(`acc,`zero,`(mr2bsp(S)))>;
- case cMap(`m,`S):
- return mkBSP(#<lambda(tuple(cache,ms,k,peer),
- setCache(cache,o_,cmap(`m,i_),
- tuple(BAG(),tuple(),TRUE())))>,
- #<tuple()>,
- #<false>,
- mr2bsp(S));
- case AggregateMap(`m,`acc,`zero,`S):
- return mkBSP(#<lambda(tuple(cache,ms,k,peer),
- setCache(cache,o_,aggregate(`acc,`zero,cmap(`m,i_)),
- tuple(BAG(),tuple(),TRUE())))>,
- #<tuple()>,
- #<false>,
- mr2bsp(S));
- case MapReduce(`m,`r,`S,`o):
- return mkBSP(#<lambda(tuple(cache,ms,map_step,peer),
- if(map_step,
- tuple(cmap(lambda(tuple(k,c),
- bag(tuple(k,tuple(k,c)))),
- cmap(`m,i_)),
- false,
- FALSE()),
- setCache(cache,o_,`(o.equals(#<true>) // need to sort the result?
- ? #<cmap(lambda(tuple(k,s),
- cmap(lambda(x,bag(tuple(x,k))),
- apply(`r,tuple(k,s)))),
- groupBy(ms))>
- : #<cmap(`r,groupBy(ms))>),
- tuple(BAG(),false,TRUE()))))>,
- #<true>,
- o,
- mr2bsp(S));
- case MapAggregateReduce(`m,`r,`acc,`zero,`S,_):
- return mkBSP(#<lambda(tuple(cache,ms,map_step,peer),
- if(map_step,
- tuple(cmap(lambda(tuple(k,c),
- bag(tuple(k,tuple(k,c)))),
- cmap(`m,i_)),
- false,
- FALSE()),
- setCache(cache,o_,aggregate(`acc,`zero,cmap(`r,groupBy(ms))),
- tuple(BAG(),false,TRUE()))))>,
- #<true>,
- #<false>,
- mr2bsp(S));
- case MapCombineReduce(`m,`c,`r,`S,`o):
- return mkBSP(#<lambda(tuple(cache,ms,map_step,peer),
- if(map_step,
- tuple(cmap(lambda(tuple(k,s),
- cmap(lambda(x,bag(tuple(k,tuple(k,x)))),
- apply(`c,tuple(k,s)))),
- groupBy(cmap(`m,i_))),
- false,
- FALSE()),
- setCache(cache,o_,`(o.equals(#<true>) // need to sort the result?
- ? #<cmap(lambda(tuple(k,s),
- cmap(lambda(x,bag(tuple(x,k))),
- apply(`r,tuple(k,s)))),
- groupBy(ms))>
- : #<cmap(`r,groupBy(ms))>),
- tuple(BAG(),false,TRUE()))))>,
- #<true>,
- o,
- mr2bsp(S));
- case MapCombineReduce2(`mx,`my,`c,`r,`x,`y,`o):
- return mkBSP2(#<lambda(tuple(cache,ms,map_step,peer),
- if(map_step,
- tuple(callM(union,`unionM,
- cmap(lambda(tuple(kx,x),
- bag(tuple(kx,tuple(kx,tuple(1,x))))),
- cmap(`mx,i_)),
- cmap(lambda(tuple(ky,y),
- bag(tuple(ky,tuple(ky,tuple(2,y))))),
- cmap(`my,j_))),
- false,
- FALSE()),
- setCache(cache,o_,cmap(lambda(tuple(k,s),
- cmap(lambda(tuple(kk,ss),
- cmap(lambda(x,bag(tuple(kk,x))),
- apply(`c,tuple(kk,ss)))),
- groupBy(apply(`r,
- tuple(cmap(lambda(tuple(kx,x),
- if(callM(eq,`eqM,kx,1),
- bag(x),
- bag())),
- s),
- cmap(lambda(tuple(ky,y),
- if(callM(eq,`eqM,ky,2),
- bag(y),
- bag())),
- s)))))),
- groupBy(ms)),
- tuple(BAG(),false,TRUE()))))>,
- #<true>,
- o,
- mr2bsp(x),
- mr2bsp(y));
- case MapReduce2(`mx,`my,`r,`x,`y,`o):
- return mkBSP2(#<lambda(tuple(cache,ms,map_step,peer),
- if(map_step,
- tuple(callM(union,`unionM,
- cmap(lambda(tuple(kx,x),
- bag(tuple(kx,tuple(kx,tuple(1,x))))),
- cmap(`mx,i_)),
- cmap(lambda(tuple(ky,y),
- bag(tuple(ky,tuple(ky,tuple(2,y))))),
- cmap(`my,j_))),
- false,
- FALSE()),
- setCache(cache,o_,cmap(lambda(tuple(k,s),
- cmap(lambda(x,bag(`(o.equals(#<true>) ? #<tuple(x,k)> : #<x>))),
- apply(`r,
- tuple(cmap(lambda(tuple(kx,x),
- if(callM(eq,`eqM,kx,1),
- bag(x),
- bag())),
- s),
- cmap(lambda(tuple(ky,y),
- if(callM(eq,`eqM,ky,2),
- bag(y),
- bag())),
- s))))),
- groupBy(ms)),
- tuple(BAG(),false,TRUE()))))>,
- #<true>,
- o,
- mr2bsp(x),
- mr2bsp(y));
- case MapAggregateReduce2(`mx,`my,`r,`acc,`zero,`x,`y,_):
- return mkBSP2(#<lambda(tuple(cache,ms,map_step,peer),
- if(map_step,
- tuple(callM(union,`unionM,
- cmap(lambda(tuple(kx,x),
- bag(tuple(kx,tuple(kx,tuple(1,x))))),
- cmap(`mx,i_)),
- cmap(lambda(tuple(ky,y),
- bag(tuple(ky,tuple(ky,tuple(2,y))))),
- cmap(`my,j_))),
- false,
- FALSE()),
- setCache(cache,o_,aggregate(`acc,`zero,
- cmap(lambda(tuple(k,s),
- apply(`r,tuple(cmap(lambda(tuple(kx,x),
- if(callM(eq,`eqM,kx,1),
- bag(x),
- bag())),
- s),
- cmap(lambda(tuple(ky,y),
- if(callM(eq,`eqM,ky,2),
- bag(y),
- bag())),
- s)))),
- groupBy(ms))),
- tuple(BAG(),false,TRUE()))))>,
- #<true>,
- #<false>,
- mr2bsp(x),
- mr2bsp(y));
- case MapJoin(`mx,`my,`r,`x,`y):
- return mr2bsp(#<MapReduce2(`mx,`my,
- lambda(tuple(xs,ys),cmap(lambda(x,apply(`r,tuple(x,ys))),xs)),
- `x,`y,false)>);
- case MapAggregateJoin(`mx,`my,`r,`acc,`zero,`x,`y):
- return mr2bsp(#<MapAggregateReduce2(`mx,`my,
- lambda(tuple(xs,ys),cmap(lambda(x,apply(`r,tuple(x,ys))),xs)),
- `acc,`zero,`x,`y)>);
- case GroupByJoin(`kx,`ky,lambda(`vx,`gx),lambda(`vy,`gy),`mr,`c,`r,`x,`y,`o):
- int n = 8;
- int m = 8;
- System.err.println("*** Using a groupBy join on a "+n+"*"+m+" grid of partitions");
- Tree xkey = #<cmap(lambda(i,bag(tuple(call(plus,call(mod,call(hash_code,`gx),`m),call(times,`m,i)),
- tuple(1,call(plus,call(mod,call(hash_code,`gx),`m),call(times,`m,i)),`vx)))),
- range(0,`(n-1)))>;
- Tree ykey = #<cmap(lambda(j,bag(tuple(call(plus,call(times,call(mod,call(hash_code,`gy),`n),`m),j),
- tuple(2,call(plus,call(times,call(mod,call(hash_code,`gy),`n),`m),j),`vy)))),
- range(0,`(m-1)))>;
- type_inference(xkey);
- type_inference(ykey);
- xkey = PlanGeneration.makePlan(xkey);
- ykey = PlanGeneration.makePlan(ykey);
- return mkBSP2(#<lambda(tuple(cache,ms,map_step,peer),
- if(map_step,
- tuple(callM(union,`unionM,
- cmap(lambda(`vx,`xkey),i_),
- cmap(lambda(`vy,`ykey),j_)),
- false,
- FALSE()),
- setCache(cache,o_,
- mergeGroupByJoin(`kx,`ky,lambda(`vx,`gx),lambda(`vy,`gy),`mr,`c,`r,
- cmap(lambda(tuple(kx,p,x),
- if(callM(eq,`eqM,kx,1),bag(tuple(p,x)),bag())),ms),
- cmap(lambda(tuple(ky,p,y),
- if(callM(eq,`eqM,ky,2),bag(tuple(p,y)),bag())),ms),`o),
- tuple(BAG(),false,TRUE()))))>,
- #<true>,
- o,
- mr2bsp(x),
- mr2bsp(y));
- case CrossProduct(`mx,`my,`r,`x,`y):
- return mkBSP(#<lambda(tuple(cache,ms,ys,peer),
- tuple(BAG(),
- setCache(cache,o_,
- cmap(lambda(x,
- cmap(lambda(y,apply(`r,tuple(x,y))),
- cmap(`my,ys))),
- cmap(`mx,i_)),
- tuple()),
- TRUE()))>,
- #<Collect(`(mr2bsp(y)))>,
- #<false>,
- mr2bsp(x));
- case CrossAggregateProduct(`mx,`my,`r,`acc,`zero,`x,`y):
- return mkBSP(#<lambda(tuple(cache,ms,ys,peer),
- tuple(BAG(),
- setCache(cache,o_,
- aggregate(`acc,`zero,
- cmap(lambda(x,
- cmap(lambda(y,apply(`r,tuple(x,y))),
- cmap(`my,ys))),
- cmap(`mx,i_)),
- tuple())),
- TRUE()))>,
- #<Collect(`(mr2bsp(y)))>,
- #<false>,
- mr2bsp(x));
- case Repeat(lambda(`v,`b),`ds,`max):
- Tree step = bspSimplify(mr2bsp(b));
- int step_cache_num = source_num(step);
- match step {
- case BSP(`n,`s,`k0,_,...as):
- Tree ds_source = mr2bsp(ds);
- int ds_cache_num = source_num(ds_source);
- ds_source = set_cache_num(ds_source,step_cache_num);
- // the initial values of all data sources
- Trees sources = #[`ds_source];
- Tree step_input = #<0>;
- for ( Tree x: as )
- match x {
- case BSPSource(`j,`y):
- if (y.equals(v))
- step_input = j;
- else sources = sources.append(x);
- case _: sources = sources.append(x);
- };
- s = subst_getCache_num((int)((LongLeaf)step_input).value(),step_cache_num,s);
- return mkBSPL(step_cache_num,
- #<lambda(tuple(cache,ms,tuple(k,steps),peer),
- let(tuple(ts,kk,step_end),
- apply(`s,tuple(cache,ms,k,peer)),
- if(step_end, // end of repeat step
- setCache(cache,`step_cache_num,
- map(lambda(tuple(x,bb),x),
- getCache(cache,`step_cache_num)),
- tuple(bag(),
- tuple(`k0,callM(plus,`plusM,steps,1)),
- if(callM(gt,`gtM,steps,`max),
- TRUE(), // if # of steps <= limit, exit
- callM(not,`notM, // ... else check the stopping condition
- aggregate(lambda(tuple(x,y),callM(or,`orM,x,y)),
- false,
- map(lambda(tuple(x,bb),bb),
- getCache(cache,`step_cache_num))))))),
- tuple(ts,tuple(kk,steps),FALSE()))))>,
- #<tuple(`k0,1)>,
- #<false>,
- sources);
- case `x: throw new Error("Cannot compile the repeat function: "+x);
- }
- // when the repeat variable is in memory
- case repeat(lambda(`v,`b),`ds,`max):
- Tree step = bspSimplify(mr2bsp(b));
- int step_cache_num = source_num(step);
- step = subst(v,#<map(lambda(tuple(x,b),x),getCache(cache,`step_cache_num))>,step);
- match step {
- case BSP(`n,`s,`k0,_,...as):
- // the initial values of all data sources
- Trees sources = #[];
- for ( Tree x: as )
- if (!x.equals(v))
- sources = sources.append(x);
- Tree res = mkBSPL(#<lambda(tuple(cache,ms,tuple(k,steps,firstp,S),peer),
- let(ignore,if(firstp,setCache(cache,`step_cache_num,S,0),0),
- let(tuple(ts,kk,step_end),
- apply(`s,tuple(cache,ms,k,peer)),
- if(step_end, // end of repeat step
- setCache(cache,o_,map(lambda(tuple(x,b),x),getCache(cache,`step_cache_num)),
- setCache(cache,`step_cache_num,
- distribute(peer,getCache(cache,`step_cache_num)),
- tuple(bag(),
- tuple(`k0,callM(plus,`plusM,steps,1),false,bag()),
- if(callM(gt,`gtM,steps,`max),
- TRUE(), // if # of steps > limit, exit
- callM(not,`notM, // ... else check the stopping condition
- aggregate(lambda(tuple(x,y),callM(or,`orM,x,y)),
- false,
- map(lambda(tuple(x,bb),bb),
- getCache(cache,`step_cache_num)))))))),
- tuple(ts,tuple(kk,steps,false,bag()),FALSE())))))>,
- #<tuple(`k0,1,true,map(lambda(x,tuple(x,false)),`ds))>,
- #<false>,
- sources);
- return #<Collect(`res)>; // must return a memory bag
- case `x: throw new Error("Cannot compile the repeat function: "+x);
- }
- case Closure(lambda(`v,`b),`ds,`max):
- Tree step = bspSimplify(mr2bsp(b));
- int step_cache_num = source_num(step);
- match step {
- case BSP(`n,`s,`k0,_,...as):
- Tree ds_source = mr2bsp(ds);
- int ds_cache_num = source_num(ds_source);
- ds_source = set_cache_num(ds_source,step_cache_num);
- // the initial values of all data sources
- Trees sources = #[`ds_source];
- Tree step_input = #<0>;
- for ( Tree x: as )
- match x {
- case BSPSource(`j,`y):
- if (y.equals(v))
- step_input = j;
- else sources = sources.append(x);
- case _: sources = sources.append(x);
- };
- s = subst_getCache_num((int)((LongLeaf)step_input).value(),step_cache_num,s);
- return mkBSPL(step_cache_num,
- #<lambda(tuple(cache,ms,tuple(k,steps,len),peer),
- let(tuple(ts,kk,step_end),
- apply(`s,tuple(cache,ms,k,peer)),
- if(step_end, // end of repeat step
- let(newLen,callM(count,`countM,getCache(cache,`step_cache_num)),
- tuple(bag(),
- tuple(`k0,callM(plus,`plusM,steps,1),newLen),
- if(callM(gt,`gtM,steps,`max),
- TRUE(), // if # of steps > limit, exit
- // ... else check if the new size is the same as the old size
- callM(geq,`geqM,len,newLen)))),
- tuple(ts,tuple(kk,steps,len),FALSE()))))>,
- #<tuple(`k0,1,callM(coerce,`coerceM,0,4))>,
- #<false>,
- sources);
- case `x: throw new Error("Cannot compile the closure function: "+x);
- }
- case Loop(lambda(tuple(...vs),tuple(...bs)),tuple(...ss),`max):
- Tree[] steps = new Tree[vs.length()];
- Tree[] inits = new Tree[vs.length()];
- int[] cache_num = new int[vs.length()];
- Tree[] k = new Tree[vs.length()];
- Trees sources = #[];
- Trees all_cache = #[];
- for ( int i = 0; i < vs.length(); i++ ) {
- inits[i] = mr2bsp(ss.nth(i));
- sources = sources.append(inits[i]);
- cache_num[i] = source_num(inits[i]);
- all_cache = all_cache.append(#<getCache(cache,`(cache_num[i]))>);
- };
- for ( int i = 0; i < vs.length(); i++ )
- match bspSimplify(mr2bsp(bs.nth(i))) {
- case BSP(`n,`s,`k0,_,...as):
- steps[i] = subst_setCache_num((int)((LongLeaf)n).value(),cache_num[i],s);
- k[i] = k0;
- loop: for ( Tree x: as )
- match x {
- case BSPSource(`m,`w):
- if (!vs.member(w))
- fail;
- for ( int j = 0; j < vs.length(); j++ )
- if (w.equals(vs.nth(j)))
- steps[i] = subst_getCache_num((int)((LongLeaf)m).value(),cache_num[j],steps[i]);
- case BSPSource(`n1,`d1):
- for ( Tree y: sources )
- match y {
- case BSPSource(`n2,`d2):
- if (d1.equals(d2)) {
- steps[i] = subst_getCache_num((int)((LongLeaf)n1).value(),
- (int)((LongLeaf)n2).value(),steps[i]);
- continue loop;
- }
- };
- sources = sources.append(x);
- case _: sources = sources.append(x);
- };
- };
- Tree code = #<tuple(BAG(),tuple(`(k[0]),0,1),TRUE())>;
- for ( int i = 0; i < vs.length(); i++ )
- code = #<if(callM(eq,`eqM,i,`i),
- let(tuple(ts,kk,step_end),
- apply(`(steps[i]),tuple(cache,ms,k,peer)),
- if(step_end, // end of repeat step
- `((i+1 < vs.length())
- ? #<tuple(bag(),tuple(`(k[i+1]),`(i+1),steps),FALSE())>
- : #<tuple(bag(),
- tuple(`(k[0]),0,callM(plus,`plusM,steps,1)),
- if(callM(gt,`gtM,steps,`max), // if # of steps > limit, exit
- TRUE(),
- FALSE()))>),
- tuple(ts,tuple(kk,i,steps),FALSE()))),
- `code)>;
- return mkBSPL(cache_num,
- #<lambda(tuple(cache,ms,tuple(k,i,steps),peer),`code)>,
- #<tuple(`(k[0]),0,2)>,
- #<false>,
- sources);
- case `f(...as):
- Trees bs = #[];
- for ( Tree a: as )
- bs = bs.append(mr2bsp(a));
- return #<`f(...bs)>;
- };
- return e;
- }
-
- /** simplify the BSP plan by fusing consequtive BSP plans */
- private static Tree bspSimplify ( Tree e ) {
- match e {
- case BSP(`n,`s2,`k2,`o,...r,BSP(_,`s1,`k1,_,...s),...t):
- Trees ys = r.append(t);
- Trees ns = #[];
- loop: for ( Tree x: s )
- match x {
- case BSPSource(`n1,`d1):
- for ( Tree y: ys )
- match y {
- case BSPSource(`n2,`d2):
- if (d1.equals(d2)) {
- s1 = subst_getCache_num((int)((LongLeaf)n1).value(),
- (int)((LongLeaf)n2).value(),s1);
- continue loop;
- }
- };
- ns = ns.append(x);
- case _: ns = ns.append(x);
- };
- return bspSimplify(mkBSPL((int)((LongLeaf)n).value(),
- #<lambda(tuple(cache,ms,tuple(first,k),peer),
- if(first,
- let(tuple(ts,kk,b),apply(`s1,tuple(cache,ms,k,peer)),
- let(exit,
- synchronize(peer,b), // poll all peers: do you want to exit?
- // all peers must aggree to exit the inner BSP
- // and proceed to the outer BSP
- tuple(ts,tuple(callM(not,`notM,exit),
- if(exit,`k2,kk)),FALSE()))),
- let(tuple(ts,kk,bb),apply(`s2,tuple(cache,ms,k,peer)),
- tuple(ts,tuple(false,kk),bb))))>,
- #<tuple(true,`k1)>,
- o,
- #[...r,...ns,...t]));
- case `f(...as):
- Trees bs = #[];
- for ( Tree a: as )
- bs = bs.append(bspSimplify(a));
- return #<`f(...bs)>;
- };
- return e;
- }
-
- private static Tree post_simplify_plan ( Tree e ) {
- match e {
- case setCache(`cache,`a,`v,`ret):
- return post_simplify_plan(#<setNth(`cache,`a,materialize(`v),`ret)>);
- case getCache(`cache,`a,...as):
- Tree z = #<nth(`cache,`a)>;
- for ( Tree x: as )
- z = #<callM(union,`unionM,`z,nth(`cache,`x))>;
- return z;
- case `f(...as):
- Trees bs = #[];
- for (Tree a: as)
- bs = bs.append(post_simplify_plan(a));
- return #<`f(...bs)>;
- };
- return e;
- }
-
- /** construct and simplify the BSP plan from a physical plan
- * @param plan the physical plan
- * @return the BSP plan
- */
- public static Tree constructBSPplan ( Tree plan ) {
- return post_simplify_plan(bspSimplify(mr2bsp(preprocess(plan))));
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/Bag.java
----------------------------------------------------------------------
diff --git a/src/Bag.java b/src/Bag.java
deleted file mode 100644
index 30fa4c7..0000000
--- a/src/Bag.java
+++ /dev/null
@@ -1,553 +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 java.util.*;
-import java.io.*;
-import org.apache.hadoop.io.*;
-import org.apache.hadoop.fs.*;
-
-
-/**
- * A sequence of MRData.
- * There are 3 kinds of Bag implementations, which are converted at run-time, when necessary:
- * 1) vector-based (materialized): used for small bags (when size is less than Config.max_materialized_bag);
- * 2) stream-based: can be traversed only once; implemented as Java iterators;
- * 3) spilled to a local file: can be accessed multiple times
- */
-public class Bag extends MRData implements Iterable<MRData> {
- enum Modes { STREAMED, MATERIALIZED, SPILLED };
- private Modes mode;
- private ArrayList<MRData> content; // content of a materialized bag
- private BagIterator iterator; // iterator for a streamed bag
- private boolean consumed; // true, if the stream has already been used
- private String path; // local path that contains the spilled bag
- private SequenceFile.Writer writer; // the file writer for spiled bags
-
- /**
- * create an empty bag as an ArrayList
- */
- public Bag () {
- mode = Modes.MATERIALIZED;
- content = new ArrayList<MRData>();
- }
-
- /**
- * create an empty bag as an ArrayList with a given capacity
- * @param size initial capacity
- */
- public Bag ( final int size ) {
- mode = Modes.MATERIALIZED;
- content = new ArrayList<MRData>(size);
- }
-
- /**
- * in-memory Bag construction (an ArrayList) initialized with data
- * @param as a vector of MRData to insert in the Bag
- */
- public Bag ( final MRData ...as ) {
- mode = Modes.MATERIALIZED;
- content = new ArrayList<MRData>(as.length);
- for ( MRData a: as )
- content.add(a);
- }
-
- /**
- * lazy construction (stream-based) of a Bag
- * @param i the Iterator that generates the Bag elements
- */
- public Bag ( final BagIterator i ) {
- mode = Modes.STREAMED;
- iterator = i;
- consumed = false;
- }
-
- /** is the Bag stored in an ArrayList? */
- public boolean materialized () {
- return mode == Modes.MATERIALIZED;
- }
-
- /** is the Bag stream-based? */
- public boolean streamed () {
- return mode == Modes.STREAMED;
- }
-
- /** is the Bag spilled into a file? */
- public boolean spilled () {
- return mode == Modes.SPILLED;
- }
-
- /** return the Bag size (cache it in memory if necessary) */
- public int size () {
- if (materialized())
- return content.size();
- if (streamed() && consumed)
- throw new Error("*** The collection stream has already been consumed");
- int i = 0;
- for ( MRData e: this )
- i++;
- if (streamed())
- consumed = true;
- return i;
- }
-
- /** trim the ArrayList that caches the Bag */
- public void trim () {
- if (materialized())
- content.trimToSize();
- }
-
- /** get the n'th element of a Bag (cache it in memory if necessary)
- * @param n the index
- * @return the n'th element
- */
- public MRData get ( final int n ) {
- if (materialized())
- if (n < size())
- return content.get(n);
- else throw new Error("List index out of range: "+n);
- if (streamed() && consumed)
- throw new Error("*** The collection stream has already been consumed");
- int i = 0;
- for ( MRData e: this )
- if (i++ == n)
- return e;
- if (streamed())
- consumed = true;
- throw new Error("Cannot retrieve the "+n+"th element of a sequence");
- }
-
- /** replace the n'th element of a Bag with a new value
- * @param n the index
- * @param value the new value
- * @return the Bag
- */
- public Bag set ( final int n, final MRData value ) {
- if (!materialized())
- throw new Error("Cannot replace an element of a non-materialized sequence");
- content.set(n,value);
- return this;
- }
-
- /** add a new value to a Bag (cache it in memory if necessary)
- * @param x the new value
- */
- public void add ( final MRData x ) {
- materialize();
- if (!spilled() && Config.hadoop_mode
- && size() >= Config.max_materialized_bag)
- spill();
- if (spilled())
- try {
- if (writer == null) { // writer was closed earlier for reading
- FileSystem fs = FileSystem.getLocal(Plan.conf);
- writer = SequenceFile.createWriter(fs,Plan.conf,new Path(path),
- MRContainer.class,NullWritable.class,
- SequenceFile.CompressionType.NONE);
- System.err.println("*** Appending elements to a spilled Bag: "+path);
- };
- writer.append(new MRContainer(x),NullWritable.get());
- } catch (IOException e) {
- throw new Error("Cannot append an element to a spilled Bag: "+path);
- }
- else content.add(x);
- }
-
- /** add a new value to a Bag (cache it in memory if necessary)
- * @param x the new value
- * @return the Bag
- */
- public Bag add_element ( final MRData x ) {
- add(x);
- return this;
- }
-
- /** add the elements of a Bag to the end of this Bag
- * @param b the Bag whose elements are copied
- * @return the Bag
- */
- public Bag addAll ( final Bag b ) {
- for ( MRData e: b )
- add(e);
- return this;
- }
-
- /** make this Bag empty (cache it in memory if necessary) */
- public void clear () throws IOException {
- if (materialized())
- content.clear();
- else if (streamed()) {
- if (writer != null)
- writer.close();
- writer = null;
- path = null;
- mode = Modes.MATERIALIZED;
- content = new ArrayList<MRData>(100);
- };
- mode = Modes.MATERIALIZED;
- content = new ArrayList<MRData>();
- }
-
- /** cache the Bag to an ArrayList when is absolutely necessary */
- public void materialize () {
- if (materialized() || spilled())
- return;
- Iterator<MRData> iter = iterator();
- mode = Modes.MATERIALIZED;
- writer = null;
- path = null;
- content = new ArrayList<MRData>(100);
- while ( iter.hasNext() )
- add(iter.next());
- if (materialized()) // it may have been spilled
- content.trimToSize();
- iterator = null;
- }
-
- private static Random random_generator = new Random();
-
- private static String new_path ( FileSystem fs ) throws IOException {
- Path p;
- do {
- p = new Path("file://"+Config.tmpDirectory+"/mrql"+(random_generator.nextInt(1000000)));
- } while (p.getFileSystem(Plan.conf).exists(p));
- String path = p.toString();
- Plan.temporary_paths.add(path);
- return path;
- }
-
- /** spill the Bag to a local file */
- private void spill () {
- if (!spilled() && Config.hadoop_mode)
- try {
- if (Plan.conf == null)
- Plan.conf = Evaluator.new_configuration();
- final FileSystem fs = FileSystem.getLocal(Plan.conf);
- path = new_path(fs);
- System.err.println("*** Spilling a Bag to a local file: "+path);
- final Path p = new Path(path);
- writer = SequenceFile.createWriter(fs,Plan.conf,new Path(path),
- MRContainer.class,NullWritable.class,
- SequenceFile.CompressionType.NONE);
- for ( MRData e: this )
- writer.append(new MRContainer(e),NullWritable.get());
- mode = Modes.SPILLED;
- content = null;
- iterator = null;
- } catch (Exception e) {
- throw new Error("Cannot spill a Bag to a local file");
- }
- }
-
- /**
- * sort the Bag (cache it in memory if necessary).
- * If the Bag was spilled during caching, use external sorting
- */
- public void sort () {
- materialize();
- if (spilled()) // if it was spilled during materialize()
- try { // use external sorting
- if (writer != null)
- writer.close();
- FileSystem fs = FileSystem.getLocal(Plan.conf);
- SequenceFile.Sorter sorter
- = new SequenceFile.Sorter(fs,new Plan.MRContainerKeyComparator(),
- MRContainer.class,NullWritable.class,Plan.conf);
- String out_path = new_path(fs);
- System.err.println("*** Using external sorting on a spilled bag "+path+" -> "+out_path);
- sorter.setMemory(64*1024*1024);
- sorter.sort(new Path(path),new Path(out_path));
- path = out_path;
- writer = null;
- } catch (Exception ex) {
- throw new Error("Cannot sort a spilled bag");
- }
- else Collections.sort(content);
- }
-
- /** return the Bag Iterator */
- public Iterator<MRData> iterator () {
- if (spilled())
- try {
- if (writer != null)
- writer.close();
- writer = null;
- return new BagIterator () {
- final FileSystem fs = FileSystem.getLocal(Plan.conf);
- final SequenceFile.Reader reader = new SequenceFile.Reader(fs,new Path(path),Plan.conf);
- final MRContainer key = new MRContainer();
- final NullWritable value = NullWritable.get();
- MRData data;
- public boolean hasNext () {
- try {
- if (!reader.next(key,value)) {
- reader.close();
- return false;
- };
- data = key.data();
- return true;
- } catch (IOException e) {
- throw new Error("Cannot collect values from a spilled Bag");
- }
- }
- public MRData next () {
- return data;
- }
- };
- } catch (IOException e) {
- throw new Error("Cannot collect values from a spilled Bag");
- }
- else if (materialized())
- return content.iterator();
- else {
- if (consumed) // this should never happen
- throw new Error("*** The collection stream has already been consumed");
- consumed = true;
- return iterator;
- }
- }
-
- /** cache MRData in memory by caching all Bags at any place and depth in MRData */
- public void materializeAll () {
- materialize();
- for (MRData e: this)
- e.materializeAll();
- }
-
- /** concatenate the elements of a given Bag to the elements of this Bag.
- * Does not change either Bag
- * @param s the given Bag
- * @return a new Bag
- */
- public Bag union ( final Bag s ) {
- final Iterator<MRData> i1 = iterator();
- final Iterator<MRData> i2 = s.iterator();
- return new Bag(new BagIterator () {
- boolean first = true;
- public boolean hasNext () {
- if (first)
- if (i1.hasNext())
- return true;
- else {
- first = false;
- return i2.hasNext();
- }
- else return i2.hasNext();
- }
- public MRData next () {
- if (first)
- return i1.next();
- else return i2.next();
- }
- });
- }
-
- /** does this Bag contain an element?
- * Cache this Bag in memory befor tetsing if necessary
- * @param x the element to find
- */
- public boolean contains ( final MRData x ) {
- if (materialized())
- return content.contains(x);
- if (streamed() && consumed)
- throw new Error("*** The collection stream has already been consumed");
- for ( MRData e: this )
- if (x.equals(e))
- return true;
- if (streamed())
- consumed = true;
- return false;
- }
-
- /** if this Bag is a Map from keys to values (a Bag of (key,value) pairs),
- * find the value with the given key; raise an error if not found
- * @param key the search key
- * @return the value associated with the key
- */
- public MRData map_find ( final MRData key ) {
- if (streamed() && consumed)
- throw new Error("*** The collection stream has already been consumed");
- for ( MRData e: this ) {
- Tuple p = (Tuple) e;
- if (key.equals(p.first()))
- return p.second();
- };
- if (streamed())
- consumed = true;
- throw new Error("key "+key+" not found in map");
- }
-
- /** if this Bag is a Map from keys to values (a Bag of (key,value) pairs),
- * does it contain a given key?
- * @param key the search key
- */
- public boolean map_contains ( final MRData key ) {
- if (streamed() && consumed)
- throw new Error("*** The collection stream has already been consumed");
- for ( MRData e: this )
- if (key.equals(((Tuple)e).first()))
- return true;
- if (streamed())
- consumed = true;
- return false;
- }
-
- /** the output serializer for Bag.
- * Stream-based Bags are serialized lazily (without having to cache the Bag in memory)
- */
- final public void write ( DataOutput out ) throws IOException {
- if (materialized()) {
- out.writeByte(MRContainer.BAG);
- WritableUtils.writeVInt(out,size());
- for ( MRData e: this )
- e.write(out);
- } else {
- out.writeByte(MRContainer.LAZY_BAG);
- for ( MRData e: this )
- e.write(out);
- out.writeByte(MRContainer.END_OF_LAZY_BAG);
- }
- }
-
- /** the input serializer for Bag */
- final public static Bag read ( DataInput in ) throws IOException {
- int n = WritableUtils.readVInt(in);
- Bag bag = new Bag(n);
- for ( int i = 0; i < n; i++ )
- bag.add(MRContainer.read(in));
- return bag;
- }
-
- /** a lazy input serializer for a Bag (it doesn't need to cache a Bag in memory) */
- public static Bag lazy_read ( final DataInput in ) throws IOException {
- Bag bag = new Bag(100);
- MRData data = MRContainer.read(in);
- while (data != MRContainer.end_of_lazy_bag) {
- bag.add(data);
- data = MRContainer.read(in);
- };
- if (bag.materialized())
- bag.content.trimToSize();
- return bag;
- }
-
- /** the input serializer for Bag */
- public void readFields ( DataInput in ) throws IOException {
- int n = WritableUtils.readVInt(in);
- mode = Modes.MATERIALIZED;
- iterator = null;
- path = null;
- writer = null;
- if (content == null)
- content = new ArrayList<MRData>(n);
- else {
- content.clear();
- content.ensureCapacity(n);
- };
- for ( int i = 0; i < n; i++ )
- add(MRContainer.read(in));
- }
-
- private void writeObject ( ObjectOutputStream out ) throws IOException {
- materialize();
- WritableUtils.writeVInt(out,size());
- for ( MRData e: this )
- e.write(out);
- }
-
- private void readObject ( ObjectInputStream in ) throws IOException, ClassNotFoundException {
- readFields(in);
- }
-
- /** compare this Bag with a given Bag by comparing their associated elements */
- public int compareTo ( MRData x ) {
- Bag xt = (Bag)x;
- Iterator<MRData> xi = xt.iterator();
- Iterator<MRData> yi = iterator();
- while ( xi.hasNext() && yi.hasNext() ) {
- int c = xi.next().compareTo(yi.next());
- if (c < 0)
- return -1;
- else if (c > 0)
- return 1;
- };
- if (xi.hasNext())
- return -1;
- else if (yi.hasNext())
- return 1;
- else return 0;
- }
-
- /** compare this Bag with a given Bag by comparing their associated elements */
- final public static int compare ( byte[] x, int xs, int xl, byte[] y, int ys, int yl, int[] size ) {
- try {
- int xn = WritableComparator.readVInt(x,xs);
- int xx = WritableUtils.decodeVIntSize(x[xs]);
- int yn = WritableComparator.readVInt(y,ys);
- int yy = WritableUtils.decodeVIntSize(y[ys]);
- for ( int i = 0; i < xn && i < yn; i++ ) {
- int k = MRContainer.compare(x,xs+xx,xl-xx,y,ys+yy,yl-yy,size);
- if (k != 0)
- return k;
- xx += size[0];
- yy += size[0];
- };
- size[0] = xx+1;
- if (xn > yn)
- return 1;
- if (xn < yn)
- return -1;
- return 0;
- } catch (IOException e) {
- throw new Error(e);
- }
- }
-
- /** is this Bag equal to another Bag (order is important) */
- public boolean equals ( Object x ) {
- if (!(x instanceof Bag))
- return false;
- Bag xt = (Bag) x;
- Iterator<MRData> xi = xt.iterator();
- Iterator<MRData> yi = iterator();
- while ( xi.hasNext() && yi.hasNext() )
- if ( !xi.next().equals(yi.next()) )
- return false;
- return xi.hasNext() || yi.hasNext();
- }
-
- /** the hash code of this Bag is the XOR of the hash code of its elements */
- public int hashCode () {
- int h = 127;
- for ( MRData e: this )
- h ^= e.hashCode();
- return Math.abs(h);
- }
-
- /** show the first few Bag elements (controlled by -bag_print) */
- public String toString () {
- materialize();
- StringBuffer b = new StringBuffer("{ ");
- int i = 0;
- for ( MRData e: this )
- if ( i++ < Config.max_bag_size_print )
- b.append(((i>1)?", ":"")+e);
- else return b.append(", ... }").toString();
- return b.append(" }").toString();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/BagIterator.java
----------------------------------------------------------------------
diff --git a/src/BagIterator.java b/src/BagIterator.java
deleted file mode 100644
index dfbf036..0000000
--- a/src/BagIterator.java
+++ /dev/null
@@ -1,34 +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 java.util.Iterator;
-import java.io.*;
-
-
-/** an Iterator over Bags */
-abstract public class BagIterator implements Iterator<MRData>, Serializable {
- public void remove () {
- throw new Error("Bag deletions are not permitted");
- }
- public void writeObject(ObjectOutputStream out) throws IOException {}
-
- public void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {}
-
- public void readObjectNoData() throws ObjectStreamException {}
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/BinaryDataSource.java
----------------------------------------------------------------------
diff --git a/src/BinaryDataSource.java b/src/BinaryDataSource.java
deleted file mode 100644
index 32ce828..0000000
--- a/src/BinaryDataSource.java
+++ /dev/null
@@ -1,36 +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.hadoop.conf.Configuration;
-
-
-/** A DataSource used for storing intermediate results and data dumps */
-final public class BinaryDataSource extends DataSource {
- BinaryDataSource ( int source_num, String path, Configuration conf ) {
- super(source_num,path,BinaryInputFormat.class,conf);
- }
-
- BinaryDataSource ( String path, Configuration conf ) {
- super(-1,path,BinaryInputFormat.class,conf);
- }
-
- public String toString () {
- return "Binary"+separator+source_num+separator+path;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/ClassImporter.java
----------------------------------------------------------------------
diff --git a/src/ClassImporter.java b/src/ClassImporter.java
deleted file mode 100644
index 0c093a6..0000000
--- a/src/ClassImporter.java
+++ /dev/null
@@ -1,202 +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 java.lang.reflect.Method;
-import org.apache.mrql.gen.*;
-import java.util.*;
-
-/** imports external Java methods into MRQL */
-final public class ClassImporter {
- final static boolean trace_imported_methods = false;
-
- final static String[] object_methods
- = { "hashCode", "getClass", "wait", "equals", "toString", "notify", "notifyAll" };
-
- static Vector<MethodInfo> methods = new Vector<MethodInfo>();
-
- public static void load_classes () {
- if (methods == null)
- methods = new Vector<MethodInfo>();
- if (methods.size() == 0) {
- importClass("org.apache.mrql.SystemFunctions");
- //****** import your classes with user-defined functions here
- }
- }
-
- private static boolean object_method ( String s ) {
- for (int i = 0; i < object_methods.length; i++)
- if (object_methods[i].equals(s))
- return true;
- return false;
- }
-
- private static Tree getType ( Class<?> c ) {
- String cn = c.getCanonicalName();
- Class<?>[] inf = c.getInterfaces();
- if (cn.equals("org.apache.mrql.MRData"))
- return new VariableLeaf("any");
- if (cn.startsWith("org.apache.mrql.MR_"))
- return new VariableLeaf(cn.substring(19));
- if (cn.equals("org.apache.mrql.Bag"))
- return new Node("bag",new Trees(new VariableLeaf("any")));
- if (cn.equals("org.apache.mrql.Inv"))
- return new VariableLeaf("any");
- if (cn.equals("org.apache.mrql.Union"))
- return new VariableLeaf("union");
- if (cn.equals("org.apache.mrql.Lambda"))
- return new VariableLeaf("any");
- if (inf.length > 0 && inf[0].equals("org.apache.mrql.MRData"))
- return new VariableLeaf("any");
- throw new Error("Unsupported type in imported method: "+cn);
- }
-
- private static Trees signature ( Method m ) {
- Class<?> co = m.getReturnType();
- Class<?>[] cs = m.getParameterTypes();
- Trees as = new Trees(getType(co));
- for (int i = 0; i < cs.length; i++)
- as = as.append(getType(cs[i]));
- return as;
- }
-
- public static String method_name ( int method_number ) {
- return methods.get(method_number).name;
- }
-
- public static Trees signature ( int method_number ) {
- return methods.get(method_number).signature;
- }
-
- /** import all Java methods from a given Java class */
- public static void importClass ( String class_name ) {
- try {
- Method[] ms = Class.forName(class_name).getMethods();
- Vector<MethodInfo> mv = new Vector<MethodInfo>();
- for (int i = 0; i < ms.length; i++)
- if (!object_method(ms[i].getName()) && ms[i].getModifiers() == 9)
- try {
- Trees sig = signature(ms[i]);
- MethodInfo m = new MethodInfo(ms[i].getName(),sig,ms[i]);
- mv.add(m);
- methods.add(m);
- } catch ( Exception e ) {
- System.out.println("Warning: method "+ms[i].getName()+" cannot be imported");
- System.out.println(e);
- throw new Error("");
- };
- Collections.sort(methods);
- if (Translator.functions == null)
- Translator.functions = Trees.nil;
- for ( MethodInfo m: methods )
- Translator.functions = Translator.functions.append(new Node(m.name,m.signature));
- if (trace_imported_methods) {
- System.out.print("Importing methods: ");
- for (int i = 0; i < mv.size(); i++ )
- System.out.print(mv.get(i).name+mv.get(i).signature.tail()
- +":"+mv.get(i).signature.head()+" ");
- System.out.println();
- }
- } catch (ClassNotFoundException x) {
- throw new Error("Undefined class: "+class_name);
- }
- }
-
- /** import a Java method with a given name from a given Java class */
- public static void importMethod ( String class_name, String method_name ) {
- try {
- Method[] ms = Class.forName(class_name).getMethods();
- MethodInfo m = null;
- for (int i = 0; i < ms.length; i++)
- if (ms[i].getName().equals(method_name)
- && !object_method(ms[i].getName()) && ms[i].getModifiers() == 9) {
- Trees sig = signature(ms[i]);
- m = new MethodInfo(ms[i].getName(),sig,ms[i]);
- Translator.functions = Translator.functions.append(new Node(ms[i].getName(),sig));
- break;
- };
- if (m == null)
- throw new Error("No such method: "+method_name);
- methods.add(m);
- Collections.sort(methods);
- if (trace_imported_methods)
- System.out.println("Importing method: "+m.name+m.signature.tail()
- +":"+m.signature.head()+" ");
- } catch (ClassNotFoundException x) {
- throw new Error("Undefined class: "+class_name);
- }
- }
-
- public static void print_methods () {
- for (int i = 0; i < methods.size(); i++ ) {
- MethodInfo m = methods.get(i);
- System.out.print(" "+m.name+":"+m.signature.tail()+"->"+m.signature.head());
- };
- }
-
- /** return the method specification of a system method with a given name over some expressions;
- * When the method is overloaded, find the most specific (in terms of arg subtyping)
- * @param method_name the given method name
- * @param args the method expressions
- * @return the method specification
- */
- public static Tree find_method ( String method_name, Trees args ) {
- for (int i = 0; i < methods.size(); i++ ) {
- MethodInfo m = methods.get(i);
- if (m.name.equals(method_name) && TypeInference.subtype(args,m.signature.tail()))
- return m.signature.head();
- };
- return null;
- }
-
- /** return the method number of a system method with a given name over some expressions;
- * When the method is overloaded, find the most specific (in terms of arg subtyping)
- * @param method_name the given method name
- * @param args the method expressions
- * @return the method number
- */
- public static int find_method_number ( String method_name, Trees args ) {
- for (int i = 0; i < methods.size(); i++ ) {
- MethodInfo m = methods.get(i);
- if (m.name.equals(method_name) && TypeInference.subtype(args,m.signature.tail()))
- return i;
- };
- return -1;
- }
-
- /** call a system method with a given number over MRData
- * @param method_number the method number
- * @param args in input arguments
- * @return the result of invoking this method over the args
- */
- public static MRData call ( int method_number, MRData... args ) {
- if (method_number < 0 || method_number >= methods.size())
- throw new Error("Run-time error (unknown method name)");
- MethodInfo m = methods.get(method_number);
- try {
- return (MRData)m.method.invoke(null,(Object[])args);
- } catch (Exception e) {
- Tuple t = new Tuple(args.length);
- for ( int i = 0; i < args.length; i++ )
- t.set(i,args[i]);
- System.err.println("Run-time error in method call: "+m.name+t+" of type "
- +m.signature.tail()+"->"+m.signature.head());
- throw new Error(e.toString());
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/Compiler.gen
----------------------------------------------------------------------
diff --git a/src/Compiler.gen b/src/Compiler.gen
deleted file mode 100644
index 4a333b6..0000000
--- a/src/Compiler.gen
+++ /dev/null
@@ -1,579 +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 javax.tools.*;
-import javax.tools.JavaCompiler.CompilationTask;
-import java.lang.reflect.*;
-import java.util.*;
-import java.net.*;
-import java.util.jar.*;
-
-
-/** compilation of MRQL expressions to Java code and then to Java bytecode */
-final public class Compiler extends Translator {
- static JavaCompiler compiler = ToolProvider.getSystemJavaCompiler();
- static DiagnosticCollector<JavaFileObject> diagnostics = new DiagnosticCollector<JavaFileObject>();
- final static String tmp_dir = "/tmp/mrql_jar_"+System.getProperty("user.name");
- public static String jar_path;
- static Trees in_memory_repeat_vars = #[];
- static int lambda_num = 0;
-
- /** Compile the MRQL functional arguments into Java bytecode */
- final private static class JavaSourceFromString extends SimpleJavaFileObject {
- final String code;
-
- JavaSourceFromString ( String name, String code ) {
- super(URI.create("string:///org/apache/mrql/" + name.replace('.','/') + Kind.SOURCE.extension),Kind.SOURCE);
- this.code = code;
- }
-
- @Override
- public CharSequence getCharContent ( boolean ignoreEncodingErrors ) {
- return code;
- }
- }
-
- private static String new_lambda_name () {
- return "MRQL_Lambda_"+(lambda_num++);
- }
-
- private static void add2jar ( File source, int offset, JarOutputStream target ) throws IOException {
- if (source.isDirectory()) {
- String name = source.getPath();
- if (name.length() > offset) {
- JarEntry entry = new JarEntry(name.substring(offset));
- entry.setTime(source.lastModified());
- target.putNextEntry(entry);
- target.closeEntry();
- };
- for ( File nestedFile: source.listFiles() )
- add2jar(nestedFile,offset,target);
- } else {
- JarEntry entry = new JarEntry(source.getPath().substring(offset));
- entry.setTime(source.lastModified());
- target.putNextEntry(entry);
- BufferedInputStream in = new BufferedInputStream(new FileInputStream(source));
- byte[] buffer = new byte[1024];
- int count = 1;
- while (count > 0) {
- count = in.read(buffer);
- if (count > 0)
- target.write(buffer,0,count);
- };
- target.closeEntry();
- in.close();
- }
- }
-
- private static void remove ( File file ) throws IOException {
- if (file.isDirectory())
- for ( File nestedFile: file.listFiles() )
- remove(nestedFile);
- file.delete();
- }
-
- public static void clean () throws IOException {
- in_memory_repeat_vars = #[];
- remove(new File(tmp_dir));
- }
-
- final private static Tree compile ( Tree e, StringBuffer out ) throws Exception {
- match e {
- case repeat(lambda(`v,`body),`s,`n):
- in_memory_repeat_vars = in_memory_repeat_vars.cons(v);
- return #<repeat(lambda(`v,`(compile(body,out))),`(compile(s,out)),`n)>;
- case `f(...al):
- if (!plans_with_distributed_lambdas.member(#<`f>))
- fail;
- Trees nl = #[];
- for ( Tree a: al)
- match Interpreter.closure(a,Interpreter.global_env) {
- case lambda(`v,`body):
- String fname = new_lambda_name();
- StringBuffer sb = new StringBuffer(1000);
- sb.append("final class "+fname+" extends Function {\n");
- sb.append(fname+" () {}\n");
- sb.append("final public MRData eval ( final MRData "+v
- +" ) { return "+compileE(body)+"; }\n}\n");
- out.append(sb);
- nl = nl.append(#<compiled(`fname,`a)>);
- case _: nl = nl.append(compile(a,out));
- };
- return #<`f(...nl)>;
- case `f(...al):
- Trees nl = #[];
- for ( Tree a: al)
- nl = nl.append(compile(a,out));
- return #<`f(...nl)>;
- };
- return e;
- }
-
- private static StringBuffer out;
-
- /** compile the functional arguments of the MRQL operators using the Java compiler
- * @param query the expression to compile
- * @return the query with all functional arguments compiled to Java bytecode
- */
- final public static Tree compile ( Tree query ) {
- try {
- // remove the old jar
- if (jar_path != null)
- remove(new File(jar_path));
- jar_path = tmp_dir+"/mrql_args_"+(new Random().nextInt(1000000))+".jar";
- out = new StringBuffer(1000);
- out.append("package org.apache.mrql;\n");
- Tree nq = compile(query,out);
- StringBuffer sb = new StringBuffer(1000);
- for ( String f: global_functions )
- match global_functions.lookup(f) {
- case function(tuple(...params),`otp,`body):
- sb.append("final public static "+get_MR_type(otp)+" "+f);
- if (params.is_empty())
- sb.append(" ()");
- else {
- match params.head() {
- case bind(`v,`tp):
- sb.append(" ( final "+get_MR_type(tp)+" "+v);
- };
- for ( Tree var: params.tail() )
- match var {
- case bind(`v,`tp):
- sb.append(", final "+get_MR_type(tp)+" "+v);
- }
- sb.append(" ) { return ("+get_MR_type(otp)+")");
- sb.append(compileE(body));
- sb.append("; }\n");
- }
- };
- out.append("final public class UserFunctions {\n");
- out.append(sb);
- out.append("}\n");
- String code = out.toString();
- //System.out.println(code);
- JavaFileObject file = new JavaSourceFromString("UserFunctions",code);
- Iterable<? extends JavaFileObject> compilationUnits = Arrays.asList(file);
- List<String> optionList = new ArrayList<String>();
- (new File(tmp_dir)).mkdir();
- String dir = tmp_dir+"/classes_"+(new Random().nextInt(1000000));
- File fdir = new File(dir);
- fdir.mkdir();
- ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
- String classpath = jar_path;
- String separator = System.getProperty("path.separator");
- for ( URL url: ((URLClassLoader) classLoader).getURLs() )
- classpath += separator+url.getFile();
- optionList.addAll(Arrays.asList("-classpath",classpath));
- optionList.addAll(Arrays.asList("-d",dir));
- CompilationTask task = compiler.getTask(null,null,diagnostics,optionList,null,compilationUnits);
- boolean success = task.call();
- if (!success)
- for ( Diagnostic d: diagnostics.getDiagnostics() )
- System.err.println("*** Compilation error at line "+d.getLineNumber()+" position "
- +d.getColumnNumber()+": "+d.getMessage(Locale.US));
- Manifest manifest = new Manifest();
- manifest.getMainAttributes().put(Attributes.Name.MANIFEST_VERSION,"1.0");
- JarOutputStream target = new JarOutputStream(new FileOutputStream(jar_path),manifest);
- add2jar(new File(dir+"/"),dir.length()+1,target);
- target.close();
- remove(fdir);
- return nq;
- } catch (Exception e) {
- System.err.println("*** Warning: Unable to compile the query:\n"+query);
- if (Config.trace)
- e.printStackTrace(System.err);
- return query;
- }
- }
-
- /** load the Java class of the anonymous function with name lambda_name */
- final public static Function compiled ( ClassLoader cl, String lambda_name ) throws Exception {
- URL[] urls = ((URLClassLoader) cl).getURLs();
- URL[] new_urls = new URL[urls.length+1];
- for ( int i = 0; i < urls.length; i++ )
- new_urls[i+1] = urls[i];
- new_urls[0] = new URL("file://"+jar_path);
- URLClassLoader loader = new URLClassLoader(new_urls,cl);
- Class c = loader.loadClass("org.apache.mrql."+lambda_name);
- Constructor cc = c.getDeclaredConstructors()[0];
- cc.setAccessible(true);
- return (Function)cc.newInstance();
- }
-
- /** The Java type of an MRQL type */
- private static String get_MR_type ( Tree type ) {
- match type {
- case boolean: return "MR_bool";
- case byte: return "MR_byte";
- case short: return "MR_short";
- case int: return "MR_int";
- case long: return "MR_long";
- case float: return "MR_float";
- case double: return "MR_double";
- case char: return "MR_char";
- case string: return "MR_string";
- case union: return "Union";
- case bag(...): return "Bag";
- case list(...): return "Bag";
- };
- return "MRData";
- }
-
- private static Trees remove_duplicates ( Trees al ) {
- if (al.is_empty())
- return al;
- Trees el = remove_duplicates(al.tail());
- if (el.member(al.head()))
- return el;
- else return el.cons(al.head());
- }
-
- /** lambda lifting: generate Java code from an anonymous function */
- private static String compilef ( String v, Tree body ) throws Exception {
- String fname = new_lambda_name();
- Trees free_vars = remove_duplicates(free_variables(body,#[`v]));
- StringBuffer sb = new StringBuffer(1000);
- sb.append("final class "+fname+" extends Function {\n");
- for ( Tree var: free_vars )
- sb.append("MRData "+var+"; ");
- sb.append("\npublic "+fname+" (");
- if (free_vars.is_empty())
- sb.append(") {}\n");
- else {
- sb.append(" MRData "+free_vars.head());
- for ( Tree var: free_vars.tail() )
- sb.append(", MRData "+var);
- sb.append(" ) { ");
- for ( Tree var: free_vars )
- sb.append("this."+var+" = "+var+"; ");
- sb.append("}\n");
- };
- sb.append("final public MRData eval ( final MRData "+v
- +" ) { return "+compileE(body)+"; }\n}\n");
- out.append(sb);
- String s = "new "+fname+"(";
- if (!free_vars.is_empty()) {
- s += free_vars.head();
- for ( Tree var: free_vars.tail() )
- s += ","+var;
- };
- return s+")";
- }
-
- private static String compileF ( Tree fnc ) throws Exception {
- match fnc {
- case lambda(`v,`b):
- return compilef(v.toString(),b);
- case compiled(`f,`lm):
- // recompile the function
- String s = compileF(lm);
- ((Node)fnc).children().head = new VariableLeaf(s.substring(4,s.indexOf("("))); //destructive
- return s;
- case function(tuple(...params),`tp,`body):
- String ret = "new Lambda(new Function () { "
- +"final public MRData eval ( final MRData _x ) { ";
- for ( int i = 0; i < params.length(); i++ )
- match params.nth(i) {
- case bind(`v,_):
- ret += "final MRData "+v+" = ((Tuple)_x).get("+i+"); ";
- };
- return ret+" return "+compileE(body)+"; } })";
- };
- throw new Exception("Ill-formed lambda: "+fnc);
- }
-
- private static String compileEL ( Trees el ) throws Exception {
- if (el.is_empty())
- return "";
- String ret = compileE(el.head());
- for ( Tree a: el.tail() )
- ret += ","+compileE(a);
- return ret;
- }
-
- private static String compileE ( Tree e ) throws Exception {
- if (e == null)
- return "(new MR_byte(0))";
- if (e.equals(#<true>))
- return "(new MR_bool(true))";
- else if (e.equals(#<false>))
- return "(new MR_bool(false))";
- else if (e.equals(#<null>))
- return "(new MR_byte(0))";
- else if (e.is_variable())
- if (in_memory_repeat_vars.member(e))
- return "Interpreter.lookup_global_binding(\""+e.toString()+"\")";
- else return e.toString();
- else if (e.is_long())
- return "(new MR_int("+((LongLeaf)e).value()+"))";
- else if (e.is_double())
- return "(new MR_float("+((DoubleLeaf)e).value()+"))";
- else if (e.is_string())
- return "(new MR_string("+e.toString()+"))";
- match e {
- case callM(and,_,`x,`y):
- return "(new MR_bool(((MR_bool)"+compileE(x)
- +").get() && ((MR_bool)"+compileE(y)+").get()))";
- case callM(or,_,`x,`y):
- return "(new MR_bool(((MR_bool)"+compileE(x)
- +").get() || ((MR_bool)"+compileE(y)+").get()))";
- case callM(not,_,`x):
- return "(new MR_bool(!((MR_bool)"+compileE(x)+").get()))";
- case callM(`f,`n,...args):
- if (!n.is_long())
- fail;
- String ret = "SystemFunctions."+ClassImporter.method_name((int)((LongLeaf)n).value())+"(";
- Trees sig = ClassImporter.signature((int)((LongLeaf)n).value());
- for (int i = 0; i < args.length(); i++)
- ret += ((i > 0) ? ",(" : "(")+get_MR_type(sig.nth(i+1))+")("+compileE(args.nth(i))+")";
- return ret+")";
- case lambda(`v,`body):
- return "new Lambda("+compilef(v.toString(),body)+")";
- case nth(`x,`n):
- return "(((Tuple)("+compileE(x)+")).get("+((LongLeaf)n).value()+"))";
- case setNth(`x,`n,`v,`ret):
- return "(((Tuple)("+compileE(x)+")).set("+((LongLeaf)n).value()+","+compileE(v)+","+compileE(ret)+"))";
- case materialize(`u):
- return "MapReduceAlgebra.materialize("+compileE(u)+")";
- case let(`v,`u,`body):
- return "(new Function () { public MRData eval ( final MRData "+v
- +" ) { if ("+v+" instanceof Bag) ((Bag)"+v+").materialize(); return "
- +compileE(body)+"; }; }).eval("+compileE(u)+")";
- case cmap(`m,`s):
- return "MapReduceAlgebra.cmap("+compileF(m)+",(Bag)("+compileE(s)+"))";
- case filter(`p,`m,`s):
- return "MapReduceAlgebra.filter("+compileF(p)+","+compileF(m)
- +",(Bag)"+compileE(s)+")";
- case map(`m,`s):
- return "MapReduceAlgebra.map("+compileF(m)+",(Bag)"+compileE(s)+")";
- case range(`min,`max):
- return "MapReduceAlgebra.generator(((MR_long)"+compileE(min)+").get(),"
- +"((MR_long)"+compileE(max)+").get())";
- case call(`f,...args):
- return "("+compileF(f)+".eval("+compileE(#<tuple(...args)>)+"))";
- case tuple():
- return "(new Tuple())";
- case tuple(`x):
- return "(new Tuple("+compileE(x)+"))";
- case tuple(`a,...el):
- String ret = "(new Tuple("+compileE(a);
- for ( Tree x: el )
- ret += ","+compileE(x);
- return ret+"))";
- case tagged_union(`n,`u):
- return "(new Union((byte)"+((LongLeaf)n).value()+","+compileE(u)+"))";
- case union_value(`x):
- return "(((Union)"+compileE(x)+").value())";
- case union_tag(`x):
- return "(new MR_int(((Union)"+compileE(x)+").tag()))";
- // used for shortcutting sync in bsp supersteps
- case BAG():
- return "SystemFunctions.bsp_empty_bag";
- case TRUE():
- return "SystemFunctions.bsp_true_value";
- case FALSE():
- return "SystemFunctions.bsp_false_value";
- case `T():
- if (is_collection(T))
- return "(new Bag())";
- else fail
- case `T(e):
- if (is_collection(T))
- return "(new Bag("+compileE(e)+"))";
- else fail
- case `T(`a,...el):
- if (!is_collection(T))
- fail;
- String ret = "(new Bag("+compileE(a);
- for ( Tree x: el )
- ret += ",(MRData)"+compileE(x);
- return ret+"))";
- case if(`c,`x,`y):
- return "((((MR_bool)"+compileE(c)+").get())?"+compileE(x)+":"+compileE(y)+")";
- case synchronize(`peer,`b):
- return "SystemFunctions.synchronize(((MR_string)"+compileE(peer)+"),(MR_bool)"+compileE(b)+")";
- case distribute(`peer,`s):
- return "SystemFunctions.distribute(((MR_string)"+compileE(peer)+"),(Bag)"+compileE(s)+")";
- case mapReduce(`mx,`my,`s,_):
- return "MapReduceAlgebra.mapReduce("+compileF(mx)+","+compileF(my)+",(Bag)("+compileE(s)+"))";
- case mapReduce2(`mx,`my,`r,`x,`y,_):
- return "MapReduceAlgebra.mapReduce2("+compileF(mx)+","+compileF(my)+","+compileF(r)
- +",(Bag)("+compileE(x)+"),(Bag)("+compileE(y)+"))";
- case mapJoin(`kx,`ky,`r,`x,`y):
- return "MapReduceAlgebra.mapJoin("+compileF(kx)+","+compileF(ky)+","+compileF(r)
- +",(Bag)("+compileE(x)+"),(Bag)("+compileE(y)+"))";
- case join(`kx,`ky,`r,`x,`y):
- return "MapReduceAlgebra.join("+compileF(kx)+","+compileF(ky)+","+compileF(r)
- +",(Bag)("+compileE(x)+"),(Bag)("+compileE(y)+"))";
- case groupBy(`s):
- return "MapReduceAlgebra.groupBy((Bag)("+compileE(s)+"))";
- case index(`x,`n):
- return "SystemFunctions.index((Bag)("+compileE(x)+"),"+compileE(n)+")";
- case range(`x,`i,`j):
- return "SystemFunctions.range((Bag)("+compileE(x)+"),"+compileE(i)+","+compileE(j)+")";
- case map_index(`x,`key):
- return "((Bag)("+compileE(x)+")).map_find("+compileE(key)+")";
- case aggregate(`acc,`zero,`s):
- return "MapReduceAlgebra.aggregate("+compileF(acc)+","+compileE(zero)+",(Bag)("+compileE(s)+"))";
- case Aggregate(`acc,`zero,`s):
- return "MapReducePlan.aggregate("+compileF(acc)+","+compileE(zero)+","+compileM(s)+")";
- case mergeGroupByJoin(`kx,`ky,`gx,`gy,`m,`c,`r,`x,`y,`o):
- return "MapReduceAlgebra.mergeGroupByJoin("+compileF(kx)+","+compileF(ky)
- +","+compileF(gx)+","+compileF(gy)+","+compileF(m)+","+compileF(c)
- +","+compileF(r)+",(Bag)"+compileE(x)+",(Bag)"+compileE(y)+")";
- case function(tuple(...params),`tp,`body):
- return compileF(e);
- case typed(`x,_):
- return compileE(x);
- case apply(`f,tuple(...args)):
- if (!f.is_variable())
- fail;
- match global_functions.lookup(f.toString()) {
- case function(tuple(...params),`otp,`body):
- String ret = "UserFunctions."+f+"(";
- if (args.is_empty())
- return ret+")";
- for ( int i = 0; i < params.length(); i++ )
- match params.nth(i) {
- case bind(_,`tp):
- ret += ((i==0)?"":",")+"("+get_MR_type(tp)+")"+compileE(args.nth(i));
- };
- return ret+")";
- };
- case apply(`f,`arg):
- if (!f.is_variable())
- fail;
- match global_functions.lookup(f.toString()) {
- case function(tuple(...params),`otp,`body):
- String ac = compileE(arg);
- String ret = "UserFunctions."+f+"(";
- for ( int i = 0; i < params.length(); i++ )
- match params.nth(i) {
- case bind(_,`tp):
- ret += ((i==0)?"":",")+"("+get_MR_type(tp)+")((Tuple)"+ac+").get("+i+")";
- };
- return ret+")";
- };
- case apply(`f,`arg):
- if (!f.is_variable())
- return "("+compileF(f)+").eval("+compileE(arg)+")";
- else return "(((Lambda)"+compileE(f)+").lambda().eval("+compileE(arg)+"))";
- case Collect(`s):
- return "Plan.collect("+compileM(s)+")";
- case trace(`x):
- return compileE(x);
- case _:
- return compileM(e);
- };
- throw new Exception("Cannot compile: "+e);
- }
-
- final private static String compileM ( Tree e ) throws Exception {
- match e {
- case cMap(`f,`s):
- return "MapReduceAlgebra.cmap("+compileF(f)+",(Bag)"+compileM(s)+")";
- case AggregateMap(`f,`acc,`zero,`s):
- return "new Bag(MapReduceAlgebra.aggregate("+compileF(acc)+","+compileE(zero)
- +","+compileM(#<cMap(`f,`s)>)+"))";
- case MapReduce(`m,`r,`s,_):
- return "MapReduceAlgebra.mapReduce("+compileF(m)+","
- +compileF(r)+",(Bag)"+compileM(s)+")";
- case MapAggregateReduce(`m,`r,`acc,`zero,`s,_):
- return "new Bag(MapReduceAlgebra.aggregate("+compileF(acc)
- +","+compileE(zero)+","+compileM(#<MapReduce(`m,`r,`s)>)+"))";
- case MapCombineReduce(`m,`c,`r,`s,_):
- return "MapReduceAlgebra.mapReduce("+compileF(m)
- +","+compileF(r)+",(Bag)"+compileM(s)+")";
- case MapReduce2(`mx,`my,`c,`r,`x,`y,_):
- return "MapReduceAlgebra.mapReduce2("+compileF(mx)+","+compileF(my)
- +","+compileF(r)+",(Bag)"+compileM(x)+",(Bag)"+compileM(y)+")";
- case MapReduce2(`mx,`my,`r,`x,`y,_):
- return "MapReduceAlgebra.mapReduce2("+compileF(mx)+","+compileF(my)
- +","+compileF(r)+",(Bag)"+compileM(x)+",(Bag)"+compileM(y)+")";
- case MapAggregateReduce2(`mx,`my,`r,`acc,`zero,`x,`y,_):
- return "new Bag(MapReduceAlgebra.aggregate("+compileF(acc)+","+compileE(zero)
- +","+compileM(#< MapReduce2(`mx,`my,`r,`x,`y)>)+"))";
- case MapJoin(`kx,`ky,`r,`x,`y):
- return "MapReduceAlgebra.mapJoin("+compileF(kx)+","+compileF(ky)
- +","+compileF(r)+",(Bag)"+compileM(x)+",(Bag)"+compileM(y)+")";
- case MapAggregateJoin(`kx,`ky,`r,`acc,`zero,`x,`y):
- return "new Bag(MapReduceAlgebra.aggregate("+compileF(acc)
- +","+compileE(zero)+","+compileM(#<MapJoin(`kx,`ky,`r,`x,`y)>)+"))";
- case GroupByJoin(`kx,`ky,`gx,`gy,`m,`c,`r,`x,`y,`o):
- return "MapReduceAlgebra.groupByJoin("+compileF(kx)+","+compileF(ky)
- +","+compileF(gx)+","+compileF(gy)+","+compileF(m)+","+compileF(c)
- +","+compileF(r)+",(Bag)"+compileM(x)+",(Bag)"+compileM(y)+")";
- case CrossProduct(`mx,`my,`r,`x,`y):
- return "MapReduceAlgebra.crossProduct("+compileF(mx)+","+compileF(my)
- +","+compileF(r)+",(Bag)"+compileM(x)+",(Bag)"+compileM(y)+")";
- case CrossAggregateProduct(`mx,`my,`r,`acc,`zero,`x,`y):
- return "new Bag(MapReduceAlgebra.aggregate("+compileF(acc)+","+compileE(zero)
- +","+compileM(#<CrossProduct(`mx,`my,`r,`x,`y)>)+"))";
- case BSPSource(`n,BinarySource(`file,_)):
- if (Config.hadoop_mode)
- return "Plan.binarySource("+compileE(n)+",((MR_string)"
- +compileE(file)+").get())";
- else return "(Bag)MapReduceAlgebra.read_binary("+compileE(n)+",((MR_string)"
- +compileE(file)+").get())";
- case BinarySource(`file,_):
- if (Config.hadoop_mode)
- return "Plan.binarySource(((MR_string)"+compileE(file)+").get())";
- else return "(Bag)MapReduceAlgebra.read_binary(((MR_string)"+compileE(file)+").get())";
- case BSPSource(`n,ParsedSource(`parser,`file,...args)):
- if (!(n instanceof LongLeaf))
- fail;
- if (!Config.hadoop_mode)
- return "MapReduceAlgebra.parsedSource(((MR_int)"+compileE(n)+").get(),\""
- +parser+"\",((MR_string)"+compileE(file)+").get(),"
- +reify(args)+")";
- Class<? extends Parser> p = DataSource.parserDirectory.get(parser.toString());
- if (p == null)
- throw new Error("Unknown parser: "+parser);
- return "Plan.parsedSource("+p.getName()+".class,((MR_string)"+compileE(file)+").get(),"
- +reify(args)+")";
- case ParsedSource(`parser,`file,...args):
- if (!Config.hadoop_mode)
- return "MapReduceAlgebra.parsedSource(\""+parser+"\",((MR_string)"
- +compileE(file)+").get(),"+reify(args)+")";
- Class<? extends Parser> p = DataSource.parserDirectory.get(parser.toString());
- if (p == null)
- throw new Error("Unknown parser: "+parser);
- return "Plan.parsedSource("+p.getName()+".class,((MR_string)"+compileE(file)+").get(),"
- +reify(args)+")";
- case Merge(`x,`y):
- return "((Bag)"+compileM(x)+").union((Bag)"+compileM(y)+")";
- case Generator(`min,`max,`size):
- return "MapReduceAlgebra.generator(((MR_long)"+compileE(min)+").get(),"
- +"((MR_long)"+compileE(max)+").get())";
- case BSP(`n,`superstep,`state,`o,...as):
- String ds = "";
- for ( Tree a: as )
- ds += ",(Bag)("+compileM(a)+")";
- return "MapReduceAlgebra.BSP("+((LongLeaf)n).value()+","
- +compileF(superstep)+","+compileE(state)+","+o+","
- +"new Bag[]{"+ds.substring(1)+"})";
- case `v:
- if (v.is_variable())
- return v.toString();
- };
- throw new Exception("Cannot compile: "+e);
- }
-}