| /** |
| * 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 { |
| |
| /** return the monoid associated with the aggregation e */ |
| private static Tree get_monoid ( Tree e ) { |
| match e { |
| case call(`f,`u): |
| Tree etp = #<none>; |
| match TypeInference.type_inference2(u) { |
| case `S(`tp): etp = tp; |
| case _: return #<none>; |
| }; |
| for ( Tree monoid: monoids ) // system & user-defined aggregations |
| match monoid { |
| case `aggr(`mtp,`plus,`zero,`unit): |
| if (!aggr.equals(f.toString())) |
| continue; |
| if (TypeInference.unify(etp,mtp) == null) |
| continue; |
| return monoid; |
| } |
| }; |
| return #<none>; |
| } |
| |
| /** 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): |
| match get_monoid(e) { |
| case `nm(`mtp,`plus,`zero,`unit): |
| 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; |
| } |
| |
| private static boolean contains_without ( Tree e, Tree with, Tree without ) { |
| if (e.equals(with)) |
| return false; |
| if (e.equals(without)) |
| return true; |
| match e { |
| case `f(...r): |
| for ( Tree a: r ) |
| if (contains_without(a,with,without)) |
| return true; |
| }; |
| return false; |
| } |
| |
| private static Trees extract_combiners ( Tree e, Tree with, Tree var ) { |
| Trees res = #[ ]; |
| match e { |
| case call(`c,`u): |
| if (contains_without(e,with,var) || occurences(with,e) == 0) |
| fail; |
| if (get_monoid(e).equals(#<none>)) |
| fail; |
| return #[`e]; |
| case `f(...r): |
| for ( Tree a: r ) |
| res = res.append(extract_combiners(a,with,var)); |
| }; |
| return res; |
| } |
| |
| private static Tree get_nested_query ( Tree var, Tree e ) { |
| match e { |
| case MapReduce(lambda(`v,`b),`r,`s,...): |
| if (!repeat_variables.member(s)) |
| fail; |
| if (!free_variables(b,#[`var,`v]).is_empty()) |
| fail; |
| return e; |
| case mapReduce(lambda(`v,`b),`r,`s,...): |
| if (!repeat_variables.member(s)) |
| fail; |
| if (!free_variables(b,#[`var,`v]).is_empty()) |
| fail; |
| return e; |
| case `f(...r): |
| for ( Tree a: r ) |
| match get_nested_query(var,a) { |
| case none: ; |
| case `c: |
| return c; |
| } |
| }; |
| return #<none>; |
| } |
| |
| /** 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 into a groupByJoin (generalized matrix multiplication) |
| case mapReduce(lambda(`vm,bag(`bm)),lambda(`vr,bag(`br)),`s,`o): |
| if (!bm.equals(vm) || !is_dataset_expr(s)) |
| fail; |
| match s { |
| case mapReduce2(lambda(`mvx,bag(tuple(`jx,`mx))), |
| lambda(`mvy,bag(tuple(`jy,`my))), |
| lambda(`v,cmap(lambda(`x,cmap(lambda(`y,bag(tuple(tuple(`gx,`gy),`mxy))), |
| 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,#[`y]).is_empty() && free_variables(gy,#[`x]).is_empty()) { |
| gxx = gy; |
| gyy = gx; |
| } else if (!free_variables(gx,#[`x]).is_empty() || !free_variables(gy,#[`y]).is_empty()) |
| fail; |
| Trees combiners = extract_combiners(br,#<nth(`vr,1)>,vr); |
| if (combiners.is_empty()) |
| fail; |
| Tree nvc = new_var(); |
| Tree nvr = new_var(); |
| Tree nm = subst(x,#<nth(nth(`nvc,1),0)>,subst(y,#<nth(nth(`nvc,1),1)>,mxy)); |
| int i = 0; |
| Trees accs = #[ ]; |
| Trees zeros = #[ ]; |
| for ( Tree combiner: combiners ) |
| match combiner { |
| case call(_,cmap(lambda(`vc,bag(`u)),_)): |
| match get_monoid(combiner) { |
| case `aggr(`mtp,`plus,`zero,`unit): |
| br = subst(combiner,#<nth(nth(`nvr,1),`i)>,br); |
| zeros = zeros.append(zero); |
| accs = accs.append(#<apply(`plus,tuple(nth(nth(`nvc,0),`i), |
| apply(`unit,apply(lambda(`vc,`u),`nm))))>); |
| i++; |
| } |
| case _: throw new Error("Unrecognized aggregation: "+combiner); |
| }; |
| type_env.insert(nvc.toString(),TypeInference.type_inference(#<tuple(tuple(...zeros),tuple(`x,`y))>)); |
| type_env.insert(nvr.toString(), |
| TypeInference.type_inference(#<tuple(nth(`vr,0),tuple(...zeros))>)); |
| Tree reducer = makePlan(subst(#<nth(`vr,0)>,#<nth(`nvr,0)>,br)); |
| if (!free_variables(reducer,#[`nvr]).is_empty()) |
| fail; |
| Tree accumulator = makePlan(simplify_all(#<tuple(...accs)>)); |
| Tree zero = makePlan(#<tuple(...zeros)>); |
| return #<GroupByJoin(lambda(`mvx,`(makePlan(jx))), |
| lambda(`mvy,`(makePlan(jy))), |
| lambda(`x,`(makePlan(gxx))), |
| lambda(`y,`(makePlan(gyy))), |
| lambda(`nvc,`accumulator), |
| `zero, |
| lambda(`nvr,`reducer), |
| `(makePlan(X)), |
| `(makePlan(Y)),`o)>; |
| }; |
| fail |
| case mapReduce(lambda(`vm,`bm),`nr,`s,`o): |
| if (!is_dataset_expr(s) || Config.bsp_mode) |
| fail; |
| Tree pm = makePlan(bm); |
| match get_nested_query(vm,pm) { |
| case MapReduce(`km,`kr,`ks,`ko): |
| Tree nv = new_var(); |
| pm = subst(#<MapReduce(`km,`kr,`ks,`ko)>, |
| #<mapReduce(`km,`kr,`nv,`ko)>, |
| pm); // must be done in memory |
| TypeInference.global_type_env.insert(nv.toString(),TypeInference.type_inference(ks)); |
| return #<let(`nv,DataSetCollect(`ks), |
| MapReduce(lambda(`vm,`pm), |
| `(makePlan(nr)), |
| `(makePlan(s)), |
| `o))>; |
| case mapReduce(`km,`kr,`ks,`ko): |
| Tree nv = new_var(); |
| pm = subst(ks,nv,pm); |
| TypeInference.global_type_env.insert(nv.toString(),TypeInference.type_inference(ks)); |
| return #<let(`nv,DataSetCollect(`ks), |
| MapReduce(lambda(`vm,`pm), |
| `(makePlan(nr)), |
| `(makePlan(s)), |
| `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; |
| TypeInference.type_inference(e); |
| 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 makePlan(s) { |
| // 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 outerMerge(`m,`x,`y): |
| if (is_dataset_expr(x) && is_dataset_expr(y)) |
| return #<OuterMerge(`(makePlan(m)), |
| `(makePlan(x)), |
| `(makePlan(y)))>; |
| else fail |
| case rightOuterMerge(`m,`x,`y): |
| if (is_dataset_expr(x) && is_dataset_expr(y)) |
| return #<RightOuterMerge(`(makePlan(m)), |
| `(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 call(stream,binary,`file,`tp): |
| return #<BinaryStream(`file,`tp)>; |
| case call(stream,gen,`f,`len,`ulen): |
| return #<SequenceStream(`(makePlan(f)),`(makePlan(len)), |
| `(makePlan(ulen)))>; |
| case call(stream,`parser,`host,`port,...args): |
| if (!port.is_long()) |
| fail; |
| Trees el = #[]; |
| for ( Tree a: args ) |
| el = el.append(makePlan(a)); |
| return #<SocketStream(`parser,`(makePlan(host)),`port,...el)>; |
| case call(stream,`parser,`file,...args): |
| Trees el = #[]; |
| for ( Tree a: args ) |
| el = el.append(makePlan(a)); |
| return #<ParsedStream(`parser,`(makePlan(file)),...el)>; |
| case stream(lambda(`v,`b),`u): |
| return #<Stream(lambda(`v,`(makePlan(b))),`(makePlan(u)))>; |
| 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),tuple(...s),`n): |
| int i = 0; |
| for ( Tree v: vs ) |
| if (!is_dataset_expr(s.nth(i))) |
| fail; |
| repeat_variables = repeat_variables.append(vs); |
| return #<Loop(lambda(tuple(...vs),`(makePlan(b))),`(makePlan(#<tuple(...s)>)),`(makePlan(n)))>; |
| case incr(`z,lambda(tuple(`s,`v),`m),`a): |
| repeat_variables = repeat_variables.append(v); |
| return #<incr(`(makePlan(z)),lambda(tuple(`s,`v),`(makePlan(m))),`(makePlan(a)))>; |
| 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); |
| if (tp.equals(#<XML>) || tp.equals(global_datatype_env.lookup(#<XML>.toString()))) |
| return makePlan(#<call(XMLchildren,`(new StringLeaf(a.toString())),`x)>); |
| match tp { |
| case `S(`xtp): |
| if (is_collection(S)) |
| if (xtp.equals(#<XML>) || xtp.equals(global_datatype_env.lookup(#<XML>.toString()))) |
| return makePlan(#<call(XMLchildren,`(new StringLeaf(a.toString())),`x)>); |
| }; |
| match TypeInference.expand(tp) { |
| case record(...bl): |
| int i = 0; |
| for ( Tree b: bl ) |
| match b { |
| case bind(`c,_): |
| if (a.equals(c)) |
| return makePlan(#<nth(`x,`i)>); |
| else fail |
| case _: i++; |
| }; |
| case union(...tl): |
| for ( Tree t: tl ) |
| match TypeInference.expand(t) { |
| case _(record(...bl)): |
| int i = 0; |
| for ( Tree b: bl ) |
| match b { |
| case bind(`c,_): |
| if (a.equals(c)) |
| return makePlan(#<nth(union_value(`x),`i)>); |
| else fail |
| case _: i++; |
| }; |
| case _(bag(tuple(string,`tv))): |
| return #<map_index(`(makePlan(#<union_value(`x)>)), |
| `(new StringLeaf(a.toString())))>; |
| case `tt: error("wrong projection: "+e+" ("+tt+")"); |
| }; |
| case `S(`ttp): |
| if (!is_collection(S)) |
| fail; |
| match TypeInference.expand(ttp) { |
| case tuple(string,`tv): |
| return #<map_index(`(makePlan(x)), |
| `(new StringLeaf(a.toString())))>; |
| case record(...bl): |
| Tree nv = new_var(); |
| int i = 0; |
| for ( Tree b: bl ) |
| match b { |
| case bind(`c,_): |
| if (a.equals(c)) |
| return makePlan(#<cmap(lambda(`nv,nth(`nv,`i)),`x)>); |
| else fail |
| case _: i++; |
| }; |
| case union(...tl): |
| Tree nv = new_var(); |
| int j = 0; |
| for ( Tree t: tl ) { |
| match t { |
| case `c(record(...bl)): |
| int i = 0; |
| for ( Tree b: bl ) |
| match b { |
| case bind(`w,_): |
| if (a.equals(w)) |
| return makePlan(#<cmap(lambda(`nv,if(call(eq,union_tag(`nv),`j), |
| bag(nth(union_value(`nv),`i)), |
| bag())),`x)>); |
| else fail |
| case _: i++; |
| }; |
| case `c(bag(tuple(string,`tv))): |
| return makePlan(#<cmap(lambda(`nv,if(call(eq,union_tag(`nv),`j), |
| bag(map_index(union_value(`nv), |
| `(new StringLeaf(a.toString())))), |
| bag())),`x)>); |
| case `tt: error("wrong projection: "+e+" ("+tt+")"); |
| }; |
| j++; |
| } |
| }; |
| case `t: error("wrong projection: "+e+" ("+t+")"); |
| }; |
| case typed(`x,`tp): |
| if (tp.is_variable() && !tp.equals(#<string>) && MRContainer.type_code(tp.toString()) >= 0) |
| return makePlan(#<call(coerce,`x, |
| `(MRContainer.type_code(tp.toString())))>); |
| else fail |
| case index(`x,`n): |
| match TypeInference.type_inference2(x) { |
| case bag(tuple(`kt,`vt)): |
| if (TypeInference.type_inference2(n).equals(kt)) |
| return #<map_index(`(makePlan(x)), |
| `(makePlan(n)))>; |
| case Bag(tuple(`kt,`vt)): |
| if (TypeInference.type_inference2(n).equals(kt)) |
| return #<map_index(`(makePlan(x)), |
| `(makePlan(n)))>; |
| case union(...tl): |
| for ( Tree t: tl ) |
| match TypeInference.expand(t) { |
| case _(bag(tuple(`kt,`vt))): |
| if (TypeInference.type_inference2(n).equals(kt)) |
| return #<map_index(`(makePlan(#<union_value(`x)>)), |
| `(makePlan(n)))>; |
| case _(list(`tp)): |
| return #<index(`(makePlan(#<union_value(`x)>)), |
| `(makePlan(n)))>; |
| case `tt: error("wrong indexing: "+e+" ("+tt+")"); |
| }; |
| }; |
| return #<index(`(makePlan(x)), |
| `(makePlan(n)))>; |
| case call(count,mapReduce(`m,lambda(`vr,`br),`X,`o)): |
| Tree nv = new_var(); |
| type_env.insert(nv.toString(),TypeInference.type_inference(vr)); |
| Tree nr = simplify(#<lambda(`nv,bag(call(count,`(subst(vr,nv,br)))))>); |
| Tree plan = #<call(sum,mapReduce(`m,`nr,`X,false))>; |
| return makePlan(plan); |
| case call(`f,`u): |
| for ( Tree monoid: monoids ) |
| match monoid { |
| case `aggr(`mtp,`plus,`zero,`unit): |
| if (aggr.equals(f.toString())) { |
| Tree plan = makePlan(u); |
| Tree nx = new_var(); |
| Tree np = new_var(); |
| Tree na = new_var(); |
| Tree tp = TypeInference.type_inference(e); |
| type_env.insert(np.toString(),#<tuple(`tp,`tp)>); |
| match TypeInference.type_inference(u) { |
| case `T(`t): |
| if (TypeInference.unify(mtp,t) == null) |
| continue; |
| type_env.insert(na.toString(),#<tuple(`tp,`t)>); |
| type_env.insert(nx.toString(),t); |
| }; |
| plus = makePlan(simplify_all(#<lambda(`np,apply(`plus,`np))>)); |
| Tree acc = makePlan(simplify_all(#<lambda(`na,apply(`plus,tuple(nth(`na,0), |
| apply(`unit,nth(`na,1)))))>)); |
| zero = makePlan((f.equals(#<avg>)) ? zero : #<typed(`zero,`tp)>); |
| match plan { |
| case MapCombineReduce(`m,`c,`r,`s,_): |
| plan = #<MapAggregateReduce(`m,`r,`acc,`zero,`s,false)>; |
| case MapReduce(`m,`r,`s,_): |
| plan = #<MapAggregateReduce(`m,`r,`acc,`zero,`s,false)>; |
| case MapReduce2(`mx,`my,`r,`x,`y,_): |
| plan = #<MapAggregateReduce2(`mx,`my,`r,`acc,`zero,`x,`y,false)>; |
| case MapJoin(`kx,`ky,`r,`x,`y): |
| plan = #<MapAggregateJoin(`kx,`ky,`r,`acc,`zero,`x,`y)>; |
| case CrossProduct(`mx,`my,`r,`x,`y): |
| plan = #<CrossAggregateProduct(`mx,`my,`r,`acc,`zero,`x,`y)>; |
| case cMap(`m,`s): |
| plan = #<AggregateMap(`m,`acc,`zero,`s)>; |
| case _: |
| if (is_dataset_expr(u)) |
| plan = #<AggregateMap(lambda(`nx,bag(`nx)),`acc,`zero,`plan)>; |
| else return #<aggregate(`acc,`zero,`plan)>; |
| }; |
| if (is_dataset_expr(u)) |
| return #<Aggregate(`plus,`zero,`plan)>; |
| else return #<aggregate(`plus,`zero,`plan)>; |
| } |
| }; |
| fail |
| case call(plus,`x,`y): |
| if (!is_dataset_expr(x) || !is_dataset_expr(y)) |
| fail; |
| return #<Merge(`(makePlan(x)),`(makePlan(y)))>; |
| case trace(`msg,`tp,`x): |
| return #<trace(`msg,`tp,`(makePlan(x)))>; |
| 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 provenance(`x,...s): |
| Tree px = makePlan(x); |
| if (is_dataset_expr(x)) |
| return #<Provenance(`px,...s)>; |
| else return #<provenance(`px,...s)>; |
| 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)>); |
| case MapAggregateReduce2(`mx,`my,lambda(`v,`b),`acc,`zero,`x,`y,false): |
| // dissable mapjoins in Flink and Spark mode, because to select between a mapjoin |
| // and a coGroup at run-time, it needs to know the input sizes |
| if (!Config.flink_mode && !Config.spark_mode) |
| fail; |
| return #<MapAggregateReduce2(`mx,`my,lambda(`v,`b),`acc,`zero, |
| `(physical_plan(x)), |
| `(physical_plan(y)),false)>; |
| // convert a reduce-side join to a fragment-replicate join, if either of the join |
| // inputs is small to fit in memory (dataset sizes are extracted from file statistics) |
| case MapAggregateReduce2(`mx,`my, |
| lambda(`v,cmap(lambda(`vx,cmap(lambda(`vy,`b),nth(`nx,1))),nth(`ny,0))), |
| null,null,`x,`y,false): |
| if (Config.noMapJoin) |
| fail; |
| if (!nx.equals(v) || !ny.equals(v) || occurences(v,b) > 0) |
| fail; |
| Tree X = new_var(); |
| Tree Y = new_var(); |
| Tree nv = new_var(); |
| Tree L1 = simplify_all(#<lambda(`nv,cmap(lambda(`vx,cmap(lambda(`vy,`b),nth(`nv,1))),bag(nth(`nv,0))))>); |
| nv = new_var(); |
| Tree L2 = simplify_all(#<lambda(`nv,cmap(lambda(`vx,cmap(lambda(`vy,`b),bag(nth(`nv,0)))),nth(`nv,1)))>); |
| nv = new_var(); |
| Tree L3 = #<lambda(`nv,cmap(lambda(`vx,cmap(lambda(`vy,`b),nth(`nv,0))),nth(`nv,1)))>; |
| Tree cond1 = makePlan(#<call(leq,dataset_size(`Y),`(Config.mapjoin_size))>); |
| Tree cond2 = makePlan(#<call(leq,dataset_size(`X),`(Config.mapjoin_size))>); |
| Tree cond3 = makePlan(#<call(lt,dataset_size(`X),dataset_size(`Y))>); |
| return #<Let(`X,`(physical_plan(x)),Let(`Y,`(physical_plan(y)), |
| If(`cond1, |
| MapJoin(`mx,`my,`L1,`X,`Y), |
| If(`cond2, |
| MapJoin(`my,`mx,`L2,`Y,`X), |
| If(`cond3, |
| MapAggregateReduce2(`my,`mx,`L3,null,null,`Y,`X,false), |
| MapAggregateReduce2(`mx,`my, |
| lambda(`v,cmap(lambda(`vx,cmap(lambda(`vy,`b),nth(`nx,1))), |
| nth(`ny,0))), |
| null,null,`X,`Y,false))))))>; |
| case MapAggregateReduce2(`mx,`my, |
| lambda(`v,cmap(lambda(`vy,cmap(lambda(`vx,`b),nth(`ny,0))),nth(`nx,1))), |
| null,null,`x,`y,false): |
| if (Config.noMapJoin) |
| fail; |
| if (!nx.equals(v) || !ny.equals(v) || occurences(v,b) > 0) |
| fail; |
| return physical_plan(#<MapAggregateReduce2(`mx,`my, |
| lambda(`v,cmap(lambda(`vx,cmap(lambda(`vy,`b),nth(`nx,1))),nth(`ny,0))), |
| null,null,`x,`y,false)>); |
| case xxxMapAggregateReduce2(`mx,`my,lambda(`v,`b),null,null,`x,`y,false): |
| if (Config.noMapJoin) |
| fail; |
| Tree X = new_var(); |
| Tree Y = new_var(); |
| Tree nv = new_var(); |
| Tree L1 = simplify_all(subst(#<nth(`v,0)>,#<bag(nth(`nv,0))>, |
| subst(#<nth(`v,1)>,#<nth(`nv,1)>, |
| #<lambda(`nv,`b)>))); |
| nv = new_var(); |
| Tree L2 = subst(#<nth(`v,0)>,#<nth(`nv,1)>, |
| subst(#<nth(`v,1)>,#<nth(`nv,0)>, |
| #<lambda(`nv,`b)>)); |
| Tree cond1 = makePlan(#<call(leq,dataset_size(`Y),`(Config.mapjoin_size))>); |
| Tree cond2 = makePlan(#<call(lt,dataset_size(`X),dataset_size(`Y))>); |
| return #<Let(`X,`(physical_plan(x)),Let(`Y,`(physical_plan(y)), |
| If(`cond1, |
| MapJoin(`mx,`my,`L1,`X,`Y), |
| If(`cond2, |
| MapAggregateReduce2(`my,`mx,`L2,null,null,`Y,`X,false), |
| MapAggregateReduce2(`mx,`my,lambda(`v,`b),null,null,`X,`Y,false)))))>; |
| case MapAggregateReduce2(`mx,`my,lambda(`v,`b),`acc,`zero,`x,`y,false): |
| Tree X = new_var(); |
| Tree Y = new_var(); |
| Tree nv = new_var(); |
| Tree nr = subst(#<nth(`v,0)>,#<nth(`nv,1)>, |
| subst(#<nth(`v,1)>,#<nth(`nv,0)>,b)); |
| if (!streamed_MapReduce2_reducer(#<lambda(`nv,`nr)>)) |
| fail; |
| Tree cond = makePlan(#<call(lt,dataset_size(`X),dataset_size(`Y))>); |
| return #<Let(`X,`(physical_plan(x)),Let(`Y,`(physical_plan(y)), |
| If(`cond, |
| MapAggregateReduce2(`my,`mx,lambda(`nv,`nr),`acc,`zero,`Y,`X,false), |
| MapAggregateReduce2(`mx,`my,lambda(`v,`b),`acc,`zero,`X,`Y,false))))>; |
| case CrossProduct(`mx,`my,`r,`x,`y): |
| return physical_plan(#<CrossAggregateProduct(`mx,`my,`r,null,null,`x,`y)>); |
| case CrossAggregateProduct(`mx,`my,lambda(`v,`b),`acc,`zero,`x,`y): |
| if (Config.flink_mode || Config.spark_mode) |
| fail; |
| 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; |
| } |
| } |
