svn commit: r1746453 - in /pig/branches/spark: src/org/apache/pig/backend/hadoop/executionengine/spark/plan/SparkCompiler.java test/org/apache/pig/test/TestPigRunner.java

[xu...@apache.org]

Author: xuefu
Date: Wed Jun  1 13:02:43 2016
New Revision: 1746453

URL: http://svn.apache.org/viewvc?rev=1746453&view=rev
Log:
PIG-4898: Fix unit test failure after PIG-4771's patch was checked in (Liyun via Xuefu)

Modified:
    pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/plan/SparkCompiler.java
    pig/branches/spark/test/org/apache/pig/test/TestPigRunner.java

Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/plan/SparkCompiler.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/plan/SparkCompiler.java?rev=1746453&r1=1746452&r2=1746453&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/plan/SparkCompiler.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/plan/SparkCompiler.java Wed Jun  1 13:02:43 2016
@@ -337,6 +337,7 @@ public class SparkCompiler extends PhyPl
 	public void visitDistinct(PODistinct op) throws VisitorException {
 		try {
 			addToPlan(op);
+            phyToSparkOpMap.put(op, curSparkOp);
 		} catch (Exception e) {
 			int errCode = 2034;
 			String msg = "Error compiling operator "

Modified: pig/branches/spark/test/org/apache/pig/test/TestPigRunner.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/test/org/apache/pig/test/TestPigRunner.java?rev=1746453&r1=1746452&r2=1746453&view=diff
==============================================================================
--- pig/branches/spark/test/org/apache/pig/test/TestPigRunner.java (original)
+++ pig/branches/spark/test/org/apache/pig/test/TestPigRunner.java Wed Jun  1 13:02:43 2016
@@ -38,11 +38,13 @@ import java.util.Properties;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.mapred.Counters;
+import org.apache.pig.ExecType;
 import org.apache.pig.PigConfiguration;
 import org.apache.pig.PigRunner;
 import org.apache.pig.PigRunner.ReturnCode;
 import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
 import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRConfiguration;
+import org.apache.pig.backend.hadoop.executionengine.spark.SparkExecType;
 import org.apache.pig.impl.PigContext;
 import org.apache.pig.impl.io.FileLocalizer;
 import org.apache.pig.impl.plan.OperatorPlan;
@@ -449,14 +451,20 @@ public class TestPigRunner {
         w.close();
 
         try {
-            String[] args = { "-x", execType, PIG_FILE };
-            PigStats stats = PigRunner.run(args, new TestNotificationListener(execType));
+            String[] args = null;
+            if (execType.toUpperCase().equals((new SparkExecType()).name())) {
+                args = new String[]{"-no_multiquery", "-x", execType, PIG_FILE};
+
+            } else {
+                args = new String[]{"-x", execType, PIG_FILE};
+            }
+            PigStats stats =  PigRunner.run(args, new TestNotificationListener(execType));
             assertTrue(stats.isSuccessful());
             if (Util.isMapredExecType(cluster.getExecType())) {
                 assertEquals(3, stats.getJobGraph().size());
             } if (Util.isSparkExecType(cluster.getExecType())) {
-                // One for each store and 2 for join.
-                assertEquals(4, stats.getJobGraph().size());
+                // One for each store and 3 for join.
+                assertEquals(6, stats.getJobGraph().size());
             } else {
                 assertEquals(1, stats.getJobGraph().size());
             }