| /** |
| * 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.*; |
| |
| /** |
| * Optimize MRQL algebra expressions using normalization rules and heuristic optimizations |
| */ |
| public class AlgebraicOptimization extends Simplification { |
| |
| /** Is this a group-by operation?*/ |
| private static boolean is_groupBy ( Tree e ) { |
| match e { |
| case groupBy(...): return true; |
| case orderBy(...): return true; |
| }; |
| return false; |
| } |
| |
| /** |
| * algebraic optimization (algebra to algebra) |
| * @param e the algebraic form to be optimized |
| * @return the optimized form |
| */ |
| public static Tree translate ( Tree e ) { |
| match e { |
| // convert self-joins to single-source mapreduce; Hadoop doesn't like self-joins anyway |
| case mapReduce2(lambda(`vx,`bx),lambda(`vy,`by),lambda(`s,`f),`X,`Y,`o): |
| if (!alpha_equivalent(X,Y) || !Config.selfJoinOpt) |
| fail; |
| Tree ny = subst(vy,vx,by); |
| Tree tx = null; |
| Tree ty = null; |
| match TypeInference.type_inference(bx) { |
| case _(tuple(_,`t)): tx = t; |
| }; |
| match TypeInference.type_inference(by) { |
| case _(tuple(_,`t)): ty = t; |
| }; |
| Tree b = subst(s,#<tuple(cmap(lambda(tuple(n,v1,v2), |
| if(call(eq,n,1),bag(v1),bag())),s), |
| cmap(lambda(tuple(n,v1,v2), |
| if(call(eq,n,2),bag(v2),bag())),s))>, |
| f); |
| Tree res = #<mapReduce(lambda(`vx,call(plus, |
| cmap(lambda(x,bag(tuple(nth(x,0),tuple(1,nth(x,1),typed(null,`ty))))),`bx), |
| cmap(lambda(y,bag(tuple(nth(y,0),tuple(2,typed(null,`tx),nth(y,1))))),`ny))), |
| lambda(tuple(k,s),`b), |
| `X,`o)>; |
| res = simplify_all(rename(res)); |
| TypeInference.type_inference(res); |
| return translate(res); |
| case mapReduce2(`mx,`my,`r,cmap(lambda(`v,`b),`X),`Y,`o): |
| match X { |
| case groupBy(...): ; |
| case _: |
| Tree nmx = simplify(#<cmap(`mx,`b)>); |
| return translate(#<mapReduce2(lambda(`v,`nmx),`my,`r,`X,`Y,`o)>); |
| }; |
| fail |
| case mapReduce2(`mx,`my,`r,`X,cmap(lambda(`v,`b),`Y),`o): |
| match Y { |
| case groupBy(...): ; |
| case _: |
| Tree nmy = simplify(#<cmap(`my,`b)>); |
| return translate(#<mapReduce2(`mx,lambda(`v,`nmy),`r,`X,`Y,`o)>); |
| }; |
| fail |
| case crossProduct(`mx,`my,`r,cmap(lambda(`vx,`bx),`X),cmap(lambda(`vy,`by),`Y)): |
| return translate(#<crossProduct(lambda(`vx,cmap(`mx,`bx)), |
| lambda(`vy,cmap(`my,`by)),`r,`X,`Y)>); |
| case crossProduct(`mx,`my,`r,cmap(lambda(`v,`b),`X),`Y): |
| return translate(#<crossProduct(lambda(`v,cmap(`mx,`b)),`my,`r,`X,`Y)>); |
| case crossProduct(`mx,`my,`r,`X,cmap(lambda(`v,`b),`Y)): |
| return translate(#<crossProduct(`mx,lambda(`v,cmap(`my,`b)),`r,`X,`Y)>); |
| case cmap(`m,crossProduct(`mx,`my,lambda(`v,`b),`X,`Y)): |
| return translate(#<crossProduct(`mx,`my,lambda(`v,cmap(`m,`b)),`X,`Y)>); |
| case cmap(`r,`groupBy1(cmap(`m,`groupBy2(`s)))): |
| if (! #[groupBy,orderBy].member(#<`groupBy1>) |
| || ! #[groupBy,orderBy].member(#<`groupBy2>)) |
| fail; |
| return #<mapReduce(`(identity()), |
| `(translate(r)), |
| `(translate(#<cmap(`m,`groupBy2(`s))>)), |
| `((#<`groupBy1>.equals(#<orderBy>)) ? #<true> : #<false>))>; |
| case cmap(`r,`groupBy(cmap(`m,`s))): |
| if (! #[groupBy,orderBy].member(#<`groupBy>)) |
| fail; |
| return #<mapReduce(`(translate(m)), |
| `(translate(r)), |
| `(translate(s)), |
| `((#<`groupBy>.equals(#<orderBy>)) ? #<true> : #<false>))>; |
| case `groupBy(cmap(`m,groupBy(`s))): |
| if (! #[groupBy,orderBy].member(#<`groupBy>)) |
| fail; |
| return #<mapReduce(`(identity()), |
| `(identity()), |
| `(translate(#<cmap(`m,groupBy(`s))>)), |
| `((#<`groupBy>.equals(#<orderBy>)) ? #<true> : #<false>))>; |
| case `groupBy(cmap(`m,`s)): |
| if (! #[groupBy,orderBy].member(#<`groupBy>)) |
| fail; |
| return #<mapReduce(`(translate(m)), |
| `(identity()), |
| `(translate(s)), |
| `((#<`groupBy>.equals(#<orderBy>)) ? #<true> : #<false>))>; |
| case cmap(`r,`groupBy(`s)): |
| if (! #[groupBy,orderBy].member(#<`groupBy>)) |
| fail; |
| return #<mapReduce(`(identity()), |
| `(translate(r)), |
| `(translate(s)), |
| `((#<`groupBy>.equals(#<orderBy>)) ? #<true> : #<false>))>; |
| case `groupBy(`s): |
| if (! #[groupBy,orderBy].member(#<`groupBy>)) |
| fail; |
| return #<mapReduce(`(identity()), |
| `(identity()), |
| `(translate(s)), |
| `((#<`groupBy>.equals(#<orderBy>)) ? #<true> : #<false>))>; |
| case cmap(`m,`s): |
| return #<cmap(`(translate(m)), |
| `(translate(s)))>; |
| // convert self-joins to single-source mapreduce; Hadoop doesn't like self-joins anyway |
| case join(lambda(`vx,`bx),lambda(`vy,`by),lambda(`s,`f),`x,`y): |
| if (!x.equals(y) || !Config.selfJoinOpt) |
| fail; |
| Tree ny = subst(vy,vx,by); |
| Tree b = subst(s, |
| #<tuple(cmap(lambda(tuple(n,v), |
| if(call(eq,n,1),bag(v),bag())),s), |
| cmap(lambda(tuple(n,v), |
| if(call(eq,n,2),bag(v),bag())),s))>, |
| f); |
| Tree res = #<mapReduce(lambda(`vx,bag(tuple(`bx,tuple(1,`vx)), |
| tuple(`ny,tuple(2,`vx)))), |
| lambda(tuple(k,s),`b),`x,false)>; |
| res = simplify_all(rename(res)); |
| TypeInference.type_inference(res); |
| return translate(res); |
| case join(lambda(`vx,`bx),lambda(`vy,`by),`f,`x,`y): |
| return translate(#<mapReduce2(lambda(`vx,bag(tuple(`bx,`vx))), |
| lambda(`vy,bag(tuple(`by,`vy))), |
| `f,`x,`y,false)>); |
| case nth(`x,`n): |
| match TypeInference.type_inference2(x) { |
| case `S(tuple(...bl)): |
| if (!is_collection(S)) |
| fail; |
| Tree nv = new_var(); |
| type_env.insert(nv.toString(),bl.nth((int)n.longValue())); |
| return translate(#<cmap(lambda(`nv,`S(nth(`nv,`n))),`x)>); |
| }; |
| fail |
| case project(`x,`a): |
| match TypeInference.type_inference2(x) { |
| case `S(record(...bl)): |
| if (!is_collection(S)) |
| fail; |
| for ( Tree b: bl ) |
| match b { |
| case bind(`c,_): |
| if (!a.equals(c)) |
| fail; |
| Tree nv = new_var(); |
| type_env.insert(nv.toString(),#<record(...bl)>); |
| return translate(#<cmap(lambda(`nv,`S(project(`nv,`a))),`x)>); |
| }; |
| }; |
| fail |
| case provenance(`x,...s): |
| return #<provenance(`(translate(x)),...s)>; |
| case `f(...al): |
| Trees bl = #[]; |
| for ( Tree a: al ) |
| bl = bl.append(translate(a)); |
| return #<`f(...bl)>; |
| }; |
| return e; |
| } |
| |
| /** apply algebraic optimizations multiple times until no change */ |
| public static Tree translate_all ( Tree e ) { |
| Tree ne = translate(e); |
| if (e.equals(ne)) |
| return e; |
| else return translate_all(ne); |
| } |
| |
| /** |
| * does a form contain a bulk plan that doesn't refer to a given variable? |
| * @param e the form |
| * @param v the given variable |
| * @return true if e contains a bulk plan that doesn't refer to v |
| */ |
| private static boolean contains_plan ( Tree e, Tree v ) { |
| match e { |
| case lambda(`x,`u): return false; |
| case let(...): return false; |
| case Let(...): return false; |
| case `f(...as): |
| if (plan_names.member(#<`f>) && !free_variables(e,#[]).member(v)) |
| return true; |
| for (Tree a: as) |
| if (contains_plan(a,v)) |
| return true; |
| return false; |
| }; |
| return false; |
| } |
| |
| /** |
| * extract the common factors (common sub-expressions) from a form |
| * that do not refer to a given variable |
| * @param e the form |
| * @param v the given variable |
| * @return the list of common factors |
| */ |
| private static Trees common_factors ( Tree e, Tree v ) { |
| match e { |
| case lambda(`x,`u): return #[]; |
| case let(...): return #[]; |
| case Let(...): return #[]; |
| case `f(...as): |
| if (!contains_plan(e,v)) |
| fail; |
| if (plan_names.member(#<`f>) && !free_variables(e,#[]).member(v)) |
| return #[`e]; |
| Trees bs = #[]; |
| for ( Tree a: as ) |
| bs = bs.append(common_factors(a,v)); |
| return bs; |
| }; |
| return #[]; |
| } |
| |
| /** |
| * if a term is used multiple times in a query, factor it out using let-expressions |
| * @param e the expression to be factored-out |
| * @return the factored-out expression |
| */ |
| public static Tree common_factoring ( Tree e ) { |
| match e { |
| case `f(...as): |
| if (!plan_names.member(#<`f>)) |
| fail; |
| Trees bs = #[]; |
| Trees binds = #[]; |
| for ( Tree a: as ) |
| match a { |
| case lambda(`v,`u): |
| if (!contains_plan(u,v)) |
| fail; |
| Trees gs = common_factors(u,v); |
| Tree nb = u; |
| for ( Tree g: gs) { |
| Tree nv = new_var(); |
| nb = subst(g,nv,nb); |
| binds = binds.append(#<bind(`nv,`g)>); |
| }; |
| bs = bs.append(#<lambda(`v,`(common_factoring(nb)))>); |
| case _: bs = bs.append(common_factoring(a)); |
| }; |
| Tree res = #<`f(...bs)>; |
| for ( Tree bind: binds ) |
| match bind { |
| case bind(`v,`x): |
| res = #<let(`v,`x,`res)>; |
| }; |
| return res; |
| case `f(...as): |
| Trees bs = #[]; |
| for ( Tree a: as ) |
| bs = bs.append(common_factoring(a)); |
| return #<`f(...bs)>; |
| }; |
| return e; |
| } |
| } |
