You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mrql.apache.org by fe...@apache.org on 2014/06/09 07:40:19 UTC
git commit: MRQL-36: Fix evaluation errors during Junit testing
Repository: incubator-mrql
Updated Branches:
refs/heads/master 38627c386 -> aa2661c67
MRQL-36: Fix evaluation errors during Junit testing
Project: http://git-wip-us.apache.org/repos/asf/incubator-mrql/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-mrql/commit/aa2661c6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-mrql/tree/aa2661c6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-mrql/diff/aa2661c6
Branch: refs/heads/master
Commit: aa2661c6736fd7b0e7ba2a8bc32e1a4ddb3d2c5b
Parents: 38627c3
Author: fegaras <fe...@cse.uta.edu>
Authored: Mon Jun 9 08:39:37 2014 +0300
Committer: fegaras <fe...@cse.uta.edu>
Committed: Mon Jun 9 08:39:37 2014 +0300
----------------------------------------------------------------------
.../main/java/org/apache/mrql/BSPTranslator.gen | 10 +--
.../java/org/apache/mrql/MapReduceAlgebra.java | 79 +++++++++++++++++++-
.../java/org/apache/mrql/PlanGeneration.gen | 12 ++-
.../test/java/org/apache/mrql/QueryTest.java | 7 +-
.../java/org/apache/mrql/MapReduceEvaluator.gen | 4 +-
.../java/org/apache/mrql/SparkEvaluator.gen | 20 ++---
tests/queries/joins_1.mrql | 2 +-
7 files changed, 113 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/aa2661c6/core/src/main/java/org/apache/mrql/BSPTranslator.gen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/BSPTranslator.gen b/core/src/main/java/org/apache/mrql/BSPTranslator.gen
index 417c04c..5b7d3be 100644
--- a/core/src/main/java/org/apache/mrql/BSPTranslator.gen
+++ b/core/src/main/java/org/apache/mrql/BSPTranslator.gen
@@ -438,9 +438,9 @@ public class BSPTranslator extends TypeInference {
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),`acc,`zero,`r,`x,`y,`o):
- int n = (int)Math.floor(Math.sqrt(4.0*Config.nodes));
+ int n = (int)Math.floor(Math.sqrt(Config.nodes));
int m = n;
- System.err.println("Using a groupBy join on a "+n+"*"+m+" grid of partitions");
+ // 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)))>;
@@ -527,14 +527,14 @@ public class BSPTranslator extends TypeInference {
tuple(bag(),
tuple(`k0,callM(plus,`plusM,steps,1)),
if(callM(gt,`gtM,steps,`max),
- TRUE(), // if # of steps <= limit, exit
+ 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)>,
+ #<tuple(`k0,2)>,
#<false>,
sources);
case `x: throw new Error("Cannot compile the repeat function: "+x);
@@ -571,7 +571,7 @@ public class BSPTranslator extends TypeInference {
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))>,
+ #<tuple(`k0,2,true,map(lambda(x,tuple(x,false)),`ds))>,
#<false>,
sources);
return #<Collect(`res)>; // must return a memory bag
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/aa2661c6/core/src/main/java/org/apache/mrql/MapReduceAlgebra.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/MapReduceAlgebra.java b/core/src/main/java/org/apache/mrql/MapReduceAlgebra.java
index 217068b..d8935e1 100644
--- a/core/src/main/java/org/apache/mrql/MapReduceAlgebra.java
+++ b/core/src/main/java/org/apache/mrql/MapReduceAlgebra.java
@@ -399,7 +399,8 @@ final public class MapReduceAlgebra {
hashTable.clear();
}
- /** An equi-join combined with a group-by implemented using hashing
+ /** An equi-join combined with a group-by implemented using a sort-merge join
+ combined with hash-based groupby/aggregation
* @param kx left key function from a to k
* @param ky right key function from b to k
* @param gx group-by key function from a to k1
@@ -420,6 +421,82 @@ final public class MapReduceAlgebra {
Tuple pair = new Tuple(2);
pair.set(1,new Tuple(2));
final Map<MRData,MRData> hashTable = new HashMap<MRData,MRData>(1000);
+ Bag xs = map(new Function() {
+ public MRData eval ( final MRData e ) {
+ Tuple t = (Tuple)e;
+ return new Tuple(new Tuple(t.first(),kx.eval(t.second())),t.second());
+ } }, X);
+ xs.sort();
+ X = null;
+ Bag ys = map(new Function() {
+ public MRData eval ( final MRData e ) {
+ Tuple t = (Tuple)e;
+ return new Tuple(new Tuple(t.first(),ky.eval(t.second())),t.second());
+ } }, Y);
+ ys.sort();
+ Y = null;
+ Bag res = new Bag();
+ MRData partition = null;
+ int i = 0;
+ int j = 0;
+ while ( i < xs.size() && j < ys.size() ) {
+ Tuple x = (Tuple)xs.get(i);
+ Tuple y = (Tuple)ys.get(j);
+ int cmp = x.first().compareTo(y.first());
+ if ( cmp > 0 ) j++;
+ else if ( cmp < 0 ) i++;
+ else { // when cmp == 0
+ Tuple key = (Tuple)x.first();
+ if (partition == null)
+ partition = key.first();
+ else if (!partition.equals(key.first())) {
+ partition = key.first();
+ flush_table(hashTable,r,res);
+ };
+ int k = i;
+ int l = j;
+ for ( ; k < xs.size() && key.compareTo(((Tuple)xs.get(k)).first()) == 0; k++ ) {
+ MRData xx = ((Tuple)xs.get(k)).second();
+ for ( l = j; l < ys.size() && key.compareTo(((Tuple)ys.get(l)).first()) == 0; l++ ) {
+ MRData yy = ((Tuple)ys.get(l)).second();
+ Tuple gkey = new Tuple(gx.eval(xx),gy.eval(yy));
+ MRData old = hashTable.get(gkey);
+ if (old == null)
+ old = zero;
+ pair.set(0,old);
+ ((Tuple)pair.get(1)).set(0,xx).set(1,yy);
+ hashTable.put(gkey,acc.eval(pair));
+ }
+ };
+ i = k;
+ j = l;
+ }
+ };
+ flush_table(hashTable,r,res);
+ return res;
+ }
+
+ /** An equi-join combined with a group-by implemented using hashing
+ * @param kx left key function from a to k
+ * @param ky right key function from b to k
+ * @param gx group-by key function from a to k1
+ * @param gy group-by key function from b to k2
+ * @param acc accumulator from (c,(a,b)) to c
+ * @param zero of type c
+ * @param r reducer from ((k1,k2),c) to d
+ * @param X left input of type {a}
+ * @param Y right input of type {b}
+ * @return a value of type {d}
+ */
+ final public static Bag mergeGroupByJoin2 ( final Function kx, final Function ky,
+ final Function gx, final Function gy,
+ final Function acc, MRData zero,
+ final Function r,
+ Bag X, Bag Y ) {
+ Bag tbag = new Bag(2);
+ Tuple pair = new Tuple(2);
+ pair.set(1,new Tuple(2));
+ final Map<MRData,MRData> hashTable = new HashMap<MRData,MRData>(1000);
Bag xs = groupBy(map(new Function() {
public MRData eval ( final MRData e ) {
Tuple t = (Tuple)e;
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/aa2661c6/core/src/main/java/org/apache/mrql/PlanGeneration.gen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/PlanGeneration.gen b/core/src/main/java/org/apache/mrql/PlanGeneration.gen
index 4fbe493..29e56ba 100644
--- a/core/src/main/java/org/apache/mrql/PlanGeneration.gen
+++ b/core/src/main/java/org/apache/mrql/PlanGeneration.gen
@@ -700,7 +700,17 @@ final public class PlanGeneration extends AlgebraicOptimization {
lambda(`v,cmap(lambda(`vx,cmap(lambda(`vy,`b),nth(`nx,1))),
nth(`ny,0))),
null,null,`X,`Y,false))))))>;
- case MapAggregateReduce2(`mx,`my,lambda(`v,`b),null,null,`x,`y,false):
+ 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();
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/aa2661c6/core/src/test/java/org/apache/mrql/QueryTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/mrql/QueryTest.java b/core/src/test/java/org/apache/mrql/QueryTest.java
index e389cec..b0257dd 100644
--- a/core/src/test/java/org/apache/mrql/QueryTest.java
+++ b/core/src/test/java/org/apache/mrql/QueryTest.java
@@ -28,6 +28,8 @@ import junit.framework.TestCase;
import org.apache.commons.io.FileUtils;
import org.junit.BeforeClass;
+import org.apache.log4j.*;
+import java.util.Enumeration;
public abstract class QueryTest extends TestCase {
private static String TEST_QUERY_DIR = "../tests/queries";
@@ -50,6 +52,9 @@ public abstract class QueryTest extends TestCase {
if(evaluator==null)
evaluator = createEvaluator();
Translator.global_reset();
+ for ( Enumeration en = LogManager.getCurrentLoggers(); en.hasMoreElements(); )
+ ((Logger)en.nextElement()).setLevel(Level.WARN);
+ LogManager.getRootLogger().setLevel(Level.WARN);
}
public void tearDown() throws IOException {
@@ -216,7 +221,7 @@ public abstract class QueryTest extends TestCase {
Config.hadoop_mode = hm;
if (!equal_value(v1,v2)) {
System.err .println("*** "+file1+" (query "+i+"):\nFound: "+v1+"\nExpected: "+v2);
- return 0; // should be i
+ return i;
};
line1 = reader1.readLine();
line2 = reader2.readLine();
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/aa2661c6/mapreduce/src/main/java/org/apache/mrql/MapReduceEvaluator.gen
----------------------------------------------------------------------
diff --git a/mapreduce/src/main/java/org/apache/mrql/MapReduceEvaluator.gen b/mapreduce/src/main/java/org/apache/mrql/MapReduceEvaluator.gen
index f16d18b..ac48b95 100644
--- a/mapreduce/src/main/java/org/apache/mrql/MapReduceEvaluator.gen
+++ b/mapreduce/src/main/java/org/apache/mrql/MapReduceEvaluator.gen
@@ -212,9 +212,9 @@ final public class MapReduceEvaluator extends Evaluator {
case GroupByJoin(`kx,`ky,`gx,`gy,`mp,`c,`r,`x,`y,`o):
DataSet X = eval(x,env,"-");
DataSet Y = eval(y,env,"-");
- int n = (int)Math.floor(Math.sqrt(4.0*Config.nodes));
+ int n = (int)Math.floor(Math.sqrt(Config.nodes));
int m = n;
- System.err.println("Using a groupBy join on a "+n+"*"+m+" grid of partitions");
+ // System.err.println("Using a groupBy join on a "+n+"*"+m+" grid of partitions");
return GroupByJoinPlan.groupByJoin(closure(kx,env),closure(ky,env),
closure(gx,env),closure(gy,env),
closure(mp,env),closure(c,env),closure(r,env),
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/aa2661c6/spark/src/main/java/org/apache/mrql/SparkEvaluator.gen
----------------------------------------------------------------------
diff --git a/spark/src/main/java/org/apache/mrql/SparkEvaluator.gen b/spark/src/main/java/org/apache/mrql/SparkEvaluator.gen
index 90cf59c..1bf814b 100644
--- a/spark/src/main/java/org/apache/mrql/SparkEvaluator.gen
+++ b/spark/src/main/java/org/apache/mrql/SparkEvaluator.gen
@@ -271,13 +271,13 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
int limit = ((MR_int)evalE(num,env)).get();
MR_rdd[] s = new MR_rdd[vs.length()];
for ( int i = 0; i < vs.length(); i++ )
- s[i] = new MR_rdd(eval(ss.nth(i),env));
+ s[i] = new MR_rdd(eval(ss.nth(i),env).cache());
for ( int n = 0; n < limit; n++ ) {
Environment nenv = env;
for ( int i = 0; i < vs.length(); i ++ )
nenv = new Environment(vs.nth(i).toString(),s[i],nenv);
for ( int i = 0; i < vs.length(); i ++ )
- s[i] = new MR_rdd(eval(bs.nth(i),nenv));
+ s[i] = new MR_rdd(eval(bs.nth(i),nenv).cache());
};
Tuple t = new Tuple(vs.length());
for ( int i = 0; i < vs.length(); i++ )
@@ -346,9 +346,9 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
final String counter ) {
JavaRDD<MRData> rd = eval(e,env);
long count = 0;
+ rd = rd.cache();
if (!counter.equals("-")) {
final Accumulator<Integer> c = spark_context.intAccumulator(0);
- rd = rd.cache();
rd.foreach(new VoidFunction<MRData>() {
public void call ( MRData value ) {
if (((MR_bool)((Tuple)value).second()).get())
@@ -556,9 +556,9 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
}
});
case GroupByJoin(`kx,`ky,`gx,`gy,`acc,`zero,`r,`x,`y,`o):
- final int n = (int)Math.floor(Math.sqrt(4.0*Config.nodes));
+ final int n = (int)Math.floor(Math.sqrt(Config.nodes));
final int m = n;
- System.err.println("Using a groupBy join on a "+n+"*"+m+" grid of partitions");
+ // System.err.println("Using a groupBy join on a "+n+"*"+m+" grid of partitions");
final org.apache.mrql.Function fkx = evalF(kx,env);
final org.apache.mrql.Function fky = evalF(ky,env);
final org.apache.mrql.Function fgx = evalF(gx,env);
@@ -712,8 +712,8 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
public Integer call ( Integer x, Integer y ) { return x+y; }
});
i++;
- cont = true_results > 0 && i <= max_num;
- System.out.println("Repeat #"+i+": "+true_results+" true results");
+ cont = true_results > 0 && i < max_num;
+ System.err.println("Repeat #"+i+": "+true_results+" true results");
} while (cont);
return res;
case Closure(lambda(`v,`b),`s,`m):
@@ -728,8 +728,8 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
old = n;
n = res.count();
i++;
- System.out.println("Repeat #"+i+": "+(old-n)+" new records");
- } while (old < n && i <= max_num);
+ System.err.println("Repeat #"+i+": "+(old-n)+" new records");
+ } while (old < n && i < max_num);
return res;
case Generator(`min,`max,`size):
DataSet ds = Plan.generator(((MR_long)evalE(min,env)).get(),
@@ -746,7 +746,7 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
case let(`v,`u,`body):
return eval(body,new Environment(v.toString(),evalE(u,env),env));
case Let(`v,`u,`body):
- return eval(body,new Environment(v.toString(),new MR_rdd(eval(u,env)),env));
+ return eval(body,new Environment(v.toString(),new MR_rdd(eval(u,env).cache()),env));
case If(`c,`x,`y):
if (((MR_bool)evalE(c,env)).get())
return eval(x,env);
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/aa2661c6/tests/queries/joins_1.mrql
----------------------------------------------------------------------
diff --git a/tests/queries/joins_1.mrql b/tests/queries/joins_1.mrql
index ecd3fe6..2070454 100644
--- a/tests/queries/joins_1.mrql
+++ b/tests/queries/joins_1.mrql
@@ -49,7 +49,7 @@ select (k,e) from e in E group by k: e.dno;
(select (k,e) from e in E group by k: e.dno)[10];
-(select e from e in E order by e.dno)[1:2];
+(select e from e in E order by e.dno, e.name)[1:2];
select (k,e) from e in E group by k: e.dno order by inv(k);