pig-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sz...@apache.org
Subject svn commit: r1796639 [11/12] - in /pig/trunk: ./ bin/ ivy/ src/META-INF/services/ src/docs/src/documentation/content/xdocs/ src/org/apache/pig/ src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/ src/org/apache/pig/backend/hadoop/executio...
Date Mon, 29 May 2017 15:00:41 GMT
Added: pig/trunk/test/org/apache/pig/spark/TestIndexedKey.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/spark/TestIndexedKey.java?rev=1796639&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/spark/TestIndexedKey.java (added)
+++ pig/trunk/test/org/apache/pig/spark/TestIndexedKey.java Mon May 29 15:00:39 2017
@@ -0,0 +1,164 @@
+/*
+ * 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.pig.spark;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+import org.apache.pig.backend.hadoop.executionengine.spark.converter.IndexedKey;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import static org.junit.Assert.assertEquals;
+
+@RunWith(JUnit4.class)
+public class TestIndexedKey {
+
+    /**Case1:Compare IndexedKeys with same index value
+     * key1    key2    equal?  hashCode1        hashCode2
+     * foo     null      N     hashCode(foo)    index
+     * null    foo       N     index            hashCode(foo)
+     * foo     foo       Y     hashCode(foo)    hashCode(foo)
+     * null    null      Y     index            index
+     * (1,1)   (1,1)     Y     hashCode((1,1))  hashCode((1,1))
+     * (1,)    (1,)      Y     hashCode((1,))   hashCode((1,))
+     * (1,1)   (1,2)     N     hashCode((1,1))  hashCode((1,2))
+     */
+    @Test
+    public void testIndexedKeyWithSameIndexValue() throws Exception {
+        IndexedKey a0 = new IndexedKey(new Byte("0"), "foo");
+        IndexedKey a1 = new IndexedKey(new Byte("0"), null);
+        assertEquals(a0.equals(a1), false);
+        assertEquals(a0.hashCode()==a1.hashCode(),false);
+
+        IndexedKey a2 = new IndexedKey(new Byte("0"), null);
+        IndexedKey a3 = new IndexedKey(new Byte("0"), "foo");
+        assertEquals(a2.equals(a3),false);
+        assertEquals(a2.hashCode()==a3.hashCode(),false);
+
+        IndexedKey a4 = new IndexedKey(new Byte("0"), "foo");
+        IndexedKey a5 = new IndexedKey(new Byte("0"), "foo");
+        assertEquals(a4.equals(a5),true);
+        assertEquals(a4.hashCode()==a5.hashCode(),true);
+
+        IndexedKey a6 = new IndexedKey(new Byte("0"), null);
+        IndexedKey a7 = new IndexedKey(new Byte("0"), null);
+        assertEquals(a6.equals(a7),true);
+        assertEquals(a6.hashCode()==a7.hashCode(),true);
+
+        Tuple t1 = TupleFactory.getInstance().newTuple(2);
+        t1.set(0,"1");
+        t1.set(1,"1");
+        Tuple t2 = TupleFactory.getInstance().newTuple(2);
+        t2.set(0,"1");
+        t2.set(1,"1");
+        IndexedKey a8 = new IndexedKey(new Byte("0"), t1);
+        IndexedKey a9 = new IndexedKey(new Byte("0"), t2);
+        assertEquals(a8.equals(a9),true);
+        assertEquals(a8.hashCode()==a9.hashCode(),true);
+
+        Tuple t3 = TupleFactory.getInstance().newTuple(2);
+        t3.set(0,"1");
+        t3.set(1,null);
+        Tuple t4 = TupleFactory.getInstance().newTuple(2);
+        t4.set(0,"1");
+        t4.set(1,null);
+        IndexedKey a10 = new IndexedKey(new Byte("0"), t3);
+        IndexedKey a11 = new IndexedKey(new Byte("0"), t4);
+        assertEquals(a10.equals(a11),true);
+        assertEquals(a10.hashCode()==a11.hashCode(),true);
+
+        Tuple t5 = TupleFactory.getInstance().newTuple(2);
+        t5.set(0,"1");
+        t5.set(1,"1");
+        Tuple t6 = TupleFactory.getInstance().newTuple(2);
+        t6.set(0,"1");
+        t6.set(1,"2");
+        IndexedKey a12 = new IndexedKey(new Byte("0"), t5);
+        IndexedKey a13 = new IndexedKey(new Byte("0"), t6);
+        assertEquals(a12.equals(a13),false);
+        assertEquals(a12.hashCode()==a13.hashCode(),false);
+    }
+
+    /*
+     * Case2:Compare IndexedKeys with different index value
+     * key1    key2    equal?  hashCode1        hashCode2
+     * foo     null     N      hashCode(foo)    index2
+     * null    foo      N      index1           hashCode(foo)
+     * foo     foo      Y      hashCode(foo)    hashCode(foo)
+     * null    null     N      index1           index2
+     * (1,1)   (1,1)    Y      hashCode((1,1))  hashCode((1,1))
+     * (1,)    (1,)     N      hashCode((1,))   hashCode((1,))
+     * (1,1)   (1,2)    N      hashCode((1,1))  hashCode((1,2))
+     */
+    @Test
+    public void testIndexedKeyWithDifferentIndexValue() throws Exception {
+        IndexedKey a0 = new IndexedKey(new Byte("0"), "foo");
+        IndexedKey a1 = new IndexedKey(new Byte("1"), null);
+        assertEquals(a0.equals(a1), false);
+        assertEquals(a0.hashCode() == a1.hashCode(), false);
+
+        IndexedKey a2 = new IndexedKey(new Byte("0"), null);
+        IndexedKey a3 = new IndexedKey(new Byte("1"), "foo");
+        assertEquals(a2.equals(a3), false);
+        assertEquals(a2.hashCode() == a3.hashCode(), false);
+
+        IndexedKey a4 = new IndexedKey(new Byte("0"), "foo");
+        IndexedKey a5 = new IndexedKey(new Byte("1"), "foo");
+        assertEquals(a4.equals(a5), true);
+        assertEquals(a4.hashCode() == a5.hashCode(), true);
+
+        IndexedKey a6 = new IndexedKey(new Byte("0"), null);
+        IndexedKey a7 = new IndexedKey(new Byte("1"), null);
+        assertEquals(a6.equals(a7), false);
+        assertEquals(a6.hashCode() == a7.hashCode(), false);
+
+        Tuple t1 = TupleFactory.getInstance().newTuple(2);
+        t1.set(0, "1");
+        t1.set(1, "1");
+        Tuple t2 = TupleFactory.getInstance().newTuple(2);
+        t2.set(0, "1");
+        t2.set(1, "1");
+        IndexedKey a8 = new IndexedKey(new Byte("0"), t1);
+        IndexedKey a9 = new IndexedKey(new Byte("1"), t2);
+        assertEquals(a8.equals(a9), true);
+        assertEquals(a8.hashCode() == a9.hashCode(), true);
+
+        Tuple t3 = TupleFactory.getInstance().newTuple(2);
+        t3.set(0, "1");
+        t3.set(1, null);
+        Tuple t4 = TupleFactory.getInstance().newTuple(2);
+        t4.set(0, "1");
+        t4.set(1, null);
+        IndexedKey a10 = new IndexedKey(new Byte("0"), t3);
+        IndexedKey a11 = new IndexedKey(new Byte("1"), t4);
+        assertEquals(a10.equals(a11), false);
+        assertEquals(a10.hashCode() == a11.hashCode(), true); //hashcode of a10 and a11 are equal but they are not equal
+
+        Tuple t5 = TupleFactory.getInstance().newTuple(2);
+        t5.set(0, "1");
+        t5.set(1, "1");
+        Tuple t6 = TupleFactory.getInstance().newTuple(2);
+        t6.set(0, "1");
+        t6.set(1, "2");
+        IndexedKey a12 = new IndexedKey(new Byte("0"), t5);
+        IndexedKey a13 = new IndexedKey(new Byte("1"), t6);
+        assertEquals(a12.equals(a13), false);
+        assertEquals(a12.hashCode() == a13.hashCode(), false);
+    }
+}

Added: pig/trunk/test/org/apache/pig/spark/TestSecondarySortSpark.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/spark/TestSecondarySortSpark.java?rev=1796639&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/spark/TestSecondarySortSpark.java (added)
+++ pig/trunk/test/org/apache/pig/spark/TestSecondarySortSpark.java Mon May 29 15:00:39 2017
@@ -0,0 +1,53 @@
+/*
+ * 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.pig.spark;
+
+import org.apache.pig.backend.hadoop.executionengine.optimizer.SecondaryKeyOptimizer;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
+import org.apache.pig.backend.hadoop.executionengine.spark.optimizer.SecondaryKeyOptimizerSpark;
+import org.apache.pig.backend.hadoop.executionengine.spark.plan.SparkCompiler;
+import org.apache.pig.backend.hadoop.executionengine.spark.plan.SparkOperPlan;
+import org.apache.pig.test.MiniGenericCluster;
+import org.apache.pig.test.TestSecondarySort;
+import org.apache.pig.test.Util;
+
+/**
+ * TestSecondarySortSpark.
+ */
+public class TestSecondarySortSpark extends TestSecondarySort {
+
+    public TestSecondarySortSpark() {
+        super();
+    }
+
+    @Override
+    public MiniGenericCluster getCluster() {
+        return MiniGenericCluster.buildCluster(MiniGenericCluster.EXECTYPE_SPARK);
+    }
+
+    @Override
+    public SecondaryKeyOptimizer visitSecondaryKeyOptimizer(String query) throws Exception {
+        PhysicalPlan pp = Util.buildPp(pigServer, query);
+        SparkCompiler comp = new SparkCompiler(pp, pc);
+        comp.compile();
+        SparkOperPlan sparkPlan = comp.getSparkPlan();
+        SecondaryKeyOptimizerSpark optimizer = new SecondaryKeyOptimizerSpark(sparkPlan);
+        optimizer.visit();
+        return optimizer;
+    }
+}

Added: pig/trunk/test/org/apache/pig/spark/TestSparkCompiler.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/spark/TestSparkCompiler.java?rev=1796639&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/spark/TestSparkCompiler.java (added)
+++ pig/trunk/test/org/apache/pig/spark/TestSparkCompiler.java Mon May 29 15:00:39 2017
@@ -0,0 +1,182 @@
+/*
+ * 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.pig.spark;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.PrintStream;
+import java.util.Properties;
+import java.util.Random;
+
+import javax.xml.parsers.ParserConfigurationException;
+import javax.xml.transform.TransformerException;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.pig.PigServer;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
+import org.apache.pig.backend.hadoop.executionengine.spark.SparkLauncher;
+import org.apache.pig.backend.hadoop.executionengine.spark.SparkLocalExecType;
+import org.apache.pig.backend.hadoop.executionengine.spark.plan.DotSparkPrinter;
+import org.apache.pig.backend.hadoop.executionengine.spark.plan.SparkOperPlan;
+import org.apache.pig.backend.hadoop.executionengine.spark.plan.SparkPrinter;
+import org.apache.pig.backend.hadoop.executionengine.spark.plan.XMLSparkPrinter;
+import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.io.FileLocalizer;
+import org.apache.pig.impl.plan.NodeIdGenerator;
+import org.apache.pig.impl.plan.VisitorException;
+import org.apache.pig.test.Util;
+import org.apache.pig.test.utils.TestHelper;
+
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test cases to test the SparkCompiler. VERY IMPORTANT NOTE: The tests here
+ * compare results with a "golden" set of outputs. In each test case here, the
+ * operators generated have a random operator key which uses Java's Random
+ * class. So if there is a code change which changes the number of operators
+ * created in a plan, then  the "golden" file for that test case
+ * need to be changed.
+ */
+
+public class TestSparkCompiler {
+    private static PigContext pc;
+    private static PigServer pigServer;
+    private static final int MAX_SIZE = 100000;
+
+    private enum PlanPrinter {
+        TEXT,
+        DOT,
+        XML;
+
+        public void doPrint(PrintStream ps, SparkOperPlan plan) throws VisitorException, ParserConfigurationException, TransformerException {
+            switch (this) {
+                case DOT:
+                    throw new RuntimeException("Testing in DOT format not supported yet");
+                    //(new DotSparkPrinter(plan, ps)).dump();
+                    //break;
+                case XML:
+                    XMLSparkPrinter printer = new XMLSparkPrinter(ps, plan);
+                    printer.visit();
+                    printer.closePlan();
+                    break;
+                case TEXT:
+                default:
+                    (new SparkPrinter(ps, plan)).visit();
+                    break;
+            }
+        }
+    }
+
+    // If for some reason, the golden files need to be regenerated, set this to
+    // true - THIS WILL OVERWRITE THE GOLDEN FILES - So use this with caution
+    // and only for the test cases you need and are sure of.
+    private boolean generate = false;
+
+    @BeforeClass
+    public static void setUpBeforeClass() throws Exception {
+        resetFileLocalizer();
+        pc = new PigContext(new SparkLocalExecType(), new Properties());
+        FileUtils.deleteDirectory(new File("/tmp/pigoutput"));
+    }
+
+    @AfterClass
+    public static void tearDownAfterClass() throws Exception {
+        resetFileLocalizer();
+    }
+
+    @Before
+    public void setUp() throws ExecException {
+        resetScope();
+        pigServer = new PigServer(pc);
+    }
+
+    private void resetScope() {
+        NodeIdGenerator.reset();
+        PigServer.resetScope();
+    }
+
+    private static void resetFileLocalizer() {
+        FileLocalizer.deleteTempFiles();
+        FileLocalizer.setInitialized(false);
+        // Set random seed to generate deterministic temporary paths
+        FileLocalizer.setR(new Random(1331L));
+    }
+
+    @Test
+    public void testStoreLoad() throws Exception {
+        String query =
+                "a = load 'file:///tmp/input' as (x:int, y:int);" +
+                "store a into 'file:///tmp/pigoutput';" +
+                "b = load 'file:///tmp/pigoutput' as (x:int, y:int);" +
+                "store b into 'file:///tmp/pigoutput1';";
+
+        run(query, "test/org/apache/pig/test/data/GoldenFiles/spark/SPARKC-LoadStore-1-text.gld", PlanPrinter.TEXT);
+        run(query, "test/org/apache/pig/test/data/GoldenFiles/spark/SPARKC-LoadStore-1-xml.gld", PlanPrinter.XML);
+        //TODO: enable this when DOT file comparison is supported
+        //run(query, "test/org/apache/pig/test/data/GoldenFiles/spark/SPARKC-LoadStore-1-dot.gld", PlanPrinter.DOT);
+    }
+
+    private void run(String query, String expectedFile, PlanPrinter planPrinter) throws Exception {
+        PhysicalPlan pp = Util.buildPp(pigServer, query);
+        SparkLauncher launcher = new SparkLauncher();
+        pc.inExplain = true;
+        SparkOperPlan sparkOperPlan = launcher.compile(pp, pc);
+
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        PrintStream ps = new PrintStream(baos);
+        planPrinter.doPrint(ps, sparkOperPlan);
+        String compiledPlan = baos.toString();
+        System.out.println();
+        System.out.println("<<<" + compiledPlan + ">>>");
+
+        if (generate) {
+            FileOutputStream fos = new FileOutputStream(expectedFile);
+            fos.write(baos.toByteArray());
+            fos.close();
+            return;
+        }
+        FileInputStream fis = new FileInputStream(expectedFile);
+        byte[] b = new byte[MAX_SIZE];
+        int len = fis.read(b);
+        fis.close();
+        String goldenPlan = new String(b, 0, len);
+        if (goldenPlan.charAt(len-1) == '\n') {
+            goldenPlan = goldenPlan.substring(0, len-1);
+        }
+
+        System.out.println("-------------");
+        System.out.println("Golden");
+        System.out.println("<<<" + goldenPlan + ">>>");
+        System.out.println("-------------");
+
+        String goldenPlanClean = Util.standardizeNewline(goldenPlan).trim();
+        String compiledPlanClean = Util.standardizeNewline(compiledPlan).trim();
+        assertEquals(TestHelper.sortUDFs(Util.removeSignature(goldenPlanClean)),
+                TestHelper.sortUDFs(Util.removeSignature(compiledPlanClean)));
+    }
+
+}
+

Modified: pig/trunk/test/org/apache/pig/test/MiniGenericCluster.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/MiniGenericCluster.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/MiniGenericCluster.java (original)
+++ pig/trunk/test/org/apache/pig/test/MiniGenericCluster.java Mon May 29 15:00:39 2017
@@ -48,6 +48,7 @@ abstract public class MiniGenericCluster
 
     public static String EXECTYPE_MR = "mr";
     public static String EXECTYPE_TEZ = "tez";
+    public static String EXECTYPE_SPARK = "spark";
 
     /**
      * Returns the single instance of class MiniGenericCluster that represents
@@ -75,6 +76,8 @@ abstract public class MiniGenericCluster
                 INSTANCE = new MiniCluster();
             } else if (execType.equalsIgnoreCase(EXECTYPE_TEZ)) {
                 INSTANCE = new TezMiniCluster();
+            } else if (execType.equalsIgnoreCase(EXECTYPE_SPARK)) {
+                INSTANCE = new SparkMiniCluster();
             } else {
                 throw new RuntimeException("Unknown test.exec.type: " + execType);
             }
@@ -157,7 +160,9 @@ abstract public class MiniGenericCluster
             return MiniCluster.getLauncher();
         } else if (execType.equalsIgnoreCase(EXECTYPE_TEZ)) {
             return TezMiniCluster.getLauncher();
-        } else {
+        } else if(execType.equalsIgnoreCase(EXECTYPE_SPARK)){
+            return SparkMiniCluster.getLauncher();
+        } else{
             throw new RuntimeException("Unknown test.exec.type: " + execType);
         }
     }

Added: pig/trunk/test/org/apache/pig/test/SparkMiniCluster.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/SparkMiniCluster.java?rev=1796639&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/test/SparkMiniCluster.java (added)
+++ pig/trunk/test/org/apache/pig/test/SparkMiniCluster.java Mon May 29 15:00:39 2017
@@ -0,0 +1,52 @@
+/*
+ * 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.pig.test;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.Map;
+
+import org.apache.commons.lang.ArrayUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.mapreduce.v2.MiniMRYarnCluster;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.pig.ExecType;
+import org.apache.pig.backend.hadoop.executionengine.Launcher;
+import org.apache.pig.backend.hadoop.executionengine.spark.SparkExecType;
+import org.apache.pig.backend.hadoop.executionengine.spark.SparkLauncher;
+
+public class SparkMiniCluster extends YarnMiniCluster {
+
+    private static final Log LOG = LogFactory
+            .getLog(SparkMiniCluster.class);
+    private ExecType spark = new SparkExecType();
+
+    @Override
+    public ExecType getExecType() {
+        return spark;
+    }
+
+    static public Launcher getLauncher() {
+        return new SparkLauncher();
+    }
+}

Modified: pig/trunk/test/org/apache/pig/test/TestAssert.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestAssert.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestAssert.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestAssert.java Mon May 29 15:00:39 2017
@@ -116,7 +116,8 @@ public class TestAssert {
       try {
           pigServer.openIterator("A");
       } catch (FrontendException fe) {
-          if (pigServer.getPigContext().getExecType().toString().startsWith("TEZ")) {
+          if (pigServer.getPigContext().getExecType().toString().startsWith("TEZ")
+                  || pigServer.getPigContext().getExecType().toString().startsWith("SPARK")) {
               Assert.assertTrue(fe.getCause().getMessage().contains(
                       "Assertion violated: i should be greater than 1"));
           } else {
@@ -147,7 +148,8 @@ public class TestAssert {
       try {
           pigServer.openIterator("A");
       } catch (FrontendException fe) {
-          if (pigServer.getPigContext().getExecType().toString().startsWith("TEZ")) {
+          if (pigServer.getPigContext().getExecType().toString().startsWith("TEZ")
+                  || pigServer.getPigContext().getExecType().toString().startsWith("SPARK")) {
               Assert.assertTrue(fe.getCause().getMessage().contains(
                       "Assertion violated: i should be greater than 1"));
           } else {

Modified: pig/trunk/test/org/apache/pig/test/TestCase.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestCase.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestCase.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestCase.java Mon May 29 15:00:39 2017
@@ -29,6 +29,7 @@ import org.apache.pig.PigServer;
 import org.apache.pig.builtin.mock.Storage.Data;
 import org.apache.pig.data.Tuple;
 import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
 import org.junit.Test;
 
 public class TestCase {
@@ -267,10 +268,15 @@ public class TestCase {
         pigServer.registerQuery("STORE C INTO 'bar' USING mock.Storage();");
 
         List<Tuple> out = data.get("bar");
-        assertEquals(3, out.size());
-        assertEquals(tuple(1, "3n+1", bag(tuple("a","x"), tuple("a","y"))), out.get(0));
-        assertEquals(tuple(2, "3n+2", bag(tuple("b","x"), tuple("b","y"))), out.get(1));
-        assertEquals(tuple(3, "3n",   bag(tuple("c","x"), tuple("c","y"))), out.get(2));
+
+        String[] expected = new String[] {
+                "(1,3n+1,{(a,x),(a,y)})",
+                "(2,3n+2,{(b,x),(b,y)})",
+                "(3,3n,{(c,x),(c,y)})"
+        };
+        Schema s = pigServer.dumpSchema("C");
+        Util.checkQueryOutputs(out.iterator(), expected, org.apache.pig.newplan.logical.Util.translateSchema(s),
+                Util.isSparkExecType(Util.getLocalTestMode()));
     }
 
     /**

Modified: pig/trunk/test/org/apache/pig/test/TestCollectedGroup.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestCollectedGroup.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestCollectedGroup.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestCollectedGroup.java Mon May 29 15:00:39 2017
@@ -25,7 +25,6 @@ import java.util.Iterator;
 import java.util.List;
 
 import org.apache.pig.CollectableLoadFunc;
-import org.apache.pig.ExecType;
 import org.apache.pig.PigServer;
 import org.apache.pig.backend.executionengine.ExecException;
 import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRCompilerException;
@@ -296,13 +295,13 @@ public class TestCollectedGroup {
     @Test
     public void testMapsideGroupWithMergeJoin() throws IOException{
         pigServer = new PigServer(cluster.getExecType(), cluster.getProperties());
-        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' using "+DummyCollectableLoader.class.getName() +"() as (id, name, grade);");
-        pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "' using "+DummyCollectableLoader.class.getName() +"() as (id, name, grade);");
+        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' using " + DummyCollectableLoader.class.getName() + "() as (id, name, grade);");
+        pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "' using " + DummyCollectableLoader.class.getName() + "() as (id, name, grade);");
         try {
             DataBag dbfrj = BagFactory.getInstance().newDefaultBag();
             DataBag dbshj = BagFactory.getInstance().newDefaultBag();
             {
-            	pigServer.registerQuery("C = join A by id, B by id using 'merge';");
+                pigServer.registerQuery("C = join A by id, B by id using 'merge';");
                 pigServer.registerQuery("D = group C by A::id using 'collected';");
                 pigServer.registerQuery("E = foreach D generate group, COUNT(C);");
                 Iterator<Tuple> iter = pigServer.openIterator("E");
@@ -312,7 +311,7 @@ public class TestCollectedGroup {
                 }
             }
             {
-            	pigServer.registerQuery("F = join A by id, B by id;");
+                pigServer.registerQuery("F = join A by id, B by id;");
                 pigServer.registerQuery("G = group F by A::id;");
                 pigServer.registerQuery("H = foreach G generate group, COUNT(F);");
                 Iterator<Tuple> iter = pigServer.openIterator("H");
@@ -321,7 +320,7 @@ public class TestCollectedGroup {
                     dbshj.add(iter.next());
                 }
             }
-            Assert.assertTrue(dbfrj.size()>0 && dbshj.size()>0);
+            Assert.assertTrue(dbfrj.size() > 0 && dbshj.size() > 0);
             Assert.assertEquals(true, TestHelper.compareBags(dbfrj, dbshj));
 
         } catch (Exception e) {
@@ -337,4 +336,4 @@ public class TestCollectedGroup {
         }
 
     }
-}
+}
\ No newline at end of file

Modified: pig/trunk/test/org/apache/pig/test/TestCombiner.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestCombiner.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestCombiner.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestCombiner.java Mon May 29 15:00:39 2017
@@ -116,13 +116,17 @@ public class TestCombiner {
         inputLines.add("a,c,1");
         String inputFileName = loadWithTestLoadFunc("A", pig, inputLines);
 
-        pig.registerQuery("B = group A by ($0, $1);");
-        pig.registerQuery("C = foreach B generate flatten(group), COUNT($1);");
-        Iterator<Tuple> resultIterator = pig.openIterator("C");
-        Tuple tuple = resultIterator.next();
-        assertEquals("(a,b,2)", tuple.toString());
-        tuple = resultIterator.next();
-        assertEquals("(a,c,1)", tuple.toString());
+        pig.registerQuery("B = foreach A generate $0 as (c0:chararray), $1 as (c1:chararray), $2 as (c2:int);");
+        pig.registerQuery("C = group B by ($0, $1);");
+        pig.registerQuery("D = foreach C generate flatten(group), COUNT($1) as int;");
+        // Since the input has no schema, using Util.getTuplesFromConstantTupleStrings fails assert.
+        List<Tuple> resultTuples = Util.getTuplesFromConstantTupleStrings(
+            new String[]{
+                "('a','b',2)",
+                "('a','c',1)",
+            });
+        Iterator<Tuple> resultIterator = pig.openIterator("D");
+        Util.checkQueryOutputsAfterSort(resultIterator, resultTuples);
 
         return inputFileName;
     }
@@ -185,7 +189,7 @@ public class TestCombiner {
         ByteArrayOutputStream baos = new ByteArrayOutputStream();
         PrintStream ps = new PrintStream(baos);
         pigServer.explain("c", ps);
-        assertTrue(baos.toString().matches("(?si).*combine plan.*"));
+        checkCombinerUsed(pigServer, "c", true);
 
         Iterator<Tuple> it = pigServer.openIterator("c");
         Tuple t = it.next();
@@ -235,7 +239,7 @@ public class TestCombiner {
         ByteArrayOutputStream baos = new ByteArrayOutputStream();
         PrintStream ps = new PrintStream(baos);
         pigServer.explain("c", ps);
-        assertTrue(baos.toString().matches("(?si).*combine plan.*"));
+        checkCombinerUsed(pigServer, "c", true);
 
         HashMap<String, Object[]> results = new HashMap<String, Object[]>();
         results.put("pig1", new Object[] { "pig1", 3L, 57L, 5.2, 75L, 9.4, 3L, 3L, 57L });
@@ -256,6 +260,56 @@ public class TestCombiner {
     }
 
     @Test
+    public void testGroupAndUnion() throws Exception {
+        // test use of combiner when group elements are accessed in the foreach
+        String input1[] = {
+                "ABC\t1\ta\t1",
+                "ABC\t1\tb\t2",
+                "ABC\t1\ta\t3",
+                "ABC\t2\tb\t4",
+        };
+
+        Util.createInputFile(cluster, "testGroupElements1.txt", input1);
+        PigServer pigServer = new PigServer(cluster.getExecType(), properties);
+        pigServer.debugOn();
+        pigServer.registerQuery("a1 = load 'testGroupElements1.txt' " +
+                "as (str:chararray, num1:int, alph : chararray, num2 : int);");
+        pigServer.registerQuery("b1 = group a1 by str;");
+
+        // check if combiner is present or not for various forms of foreach
+        pigServer.registerQuery("c1 = foreach b1  generate flatten(group), COUNT(a1.alph), SUM(a1.num2); ");
+
+        String input2[] = {
+                "DEF\t2\ta\t3",
+                "DEF\t2\td\t5",
+        };
+
+        Util.createInputFile(cluster, "testGroupElements2.txt", input2);
+        pigServer.registerQuery("a2 = load 'testGroupElements2.txt' " +
+                "as (str:chararray, num1:int, alph : chararray, num2 : int);");
+        pigServer.registerQuery("b2 = group a2 by str;");
+
+        // check if combiner is present or not for various forms of foreach
+        pigServer.registerQuery("c2 = foreach b2  generate flatten(group), COUNT(a2.alph), SUM(a2.num2); ");
+
+        pigServer.registerQuery("c = union c1, c2;");
+
+        List<Tuple> expectedRes =
+                Util.getTuplesFromConstantTupleStrings(
+                        new String[]{
+                                "('ABC',4L,10L)",
+                                "('DEF',2L,8L)",
+                        });
+
+        Iterator<Tuple> it = pigServer.openIterator("c");
+        Util.checkQueryOutputsAfterSort(it, expectedRes);
+
+        Util.deleteFile(cluster, "testGroupElements1.txt");
+        Util.deleteFile(cluster, "testGroupElements2.txt");
+        pigServer.shutdown();
+    }
+
+    @Test
     public void testGroupElements() throws Exception {
         // test use of combiner when group elements are accessed in the foreach
         String input[] = {
@@ -352,13 +406,19 @@ public class TestCombiner {
         pigServer.shutdown();
     }
 
-    private void checkCombinerUsed(PigServer pigServer, String string, boolean combineExpected)
+    private void checkCombinerUsed(PigServer pigServer, String alias, boolean combineExpected)
             throws IOException {
         // make sure there is a combine plan in the explain output
         ByteArrayOutputStream baos = new ByteArrayOutputStream();
         PrintStream ps = new PrintStream(baos);
-        pigServer.explain("c", ps);
-        boolean combinerFound = baos.toString().matches("(?si).*combine plan.*");
+        pigServer.explain(alias, ps);
+        boolean combinerFound;
+        if (pigServer.getPigContext().getExecType().name().equalsIgnoreCase("spark")) {
+            combinerFound = baos.toString().contains("Reduce By");
+        } else {
+            combinerFound = baos.toString().matches("(?si).*combine plan.*");
+        }
+
         System.out.println(baos.toString());
         assertEquals("is combiner present as expected", combineExpected, combinerFound);
     }

Modified: pig/trunk/test/org/apache/pig/test/TestCubeOperator.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestCubeOperator.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestCubeOperator.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestCubeOperator.java Mon May 29 15:00:39 2017
@@ -566,8 +566,9 @@ public class TestCubeOperator {
     public void testIllustrate() throws Exception {
 	// test for illustrate
         Assume.assumeTrue("illustrate does not work in tez (PIG-3993)", !Util.getLocalTestMode().toString().startsWith("TEZ"));
-	String query = "a = load 'input' USING mock.Storage() as (a1:chararray,b1:chararray,c1:long); "
-	        + "b = cube a by cube(a1,b1);";
+        Assume.assumeTrue("illustrate does not work in spark (PIG-4621)", !Util.getLocalTestMode().toString().startsWith("SPARK"));
+        String query = "a = load 'input' USING mock.Storage() as (a1:chararray,b1:chararray,c1:long); "
+            + "b = cube a by cube(a1,b1);";
 
         Util.registerMultiLineQuery(pigServer, query);
         Map<Operator, DataBag> examples = pigServer.getExamples("b");

Modified: pig/trunk/test/org/apache/pig/test/TestEmptyInputDir.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestEmptyInputDir.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestEmptyInputDir.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestEmptyInputDir.java Mon May 29 15:00:39 2017
@@ -33,6 +33,7 @@ import org.apache.pig.PigRunner;
 import org.apache.pig.tools.pigstats.PigStats;
 import org.apache.pig.tools.pigstats.mapreduce.MRJobStats;
 import org.junit.AfterClass;
+import org.junit.Assume;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
@@ -85,6 +86,8 @@ public class TestEmptyInputDir {
                 assertEquals(0, js.getNumberMaps());
             }
 
+            //Spark doesn't create an empty result file part-*, only a _SUCCESS file if input dir was empty
+            Assume.assumeTrue("Skip this test for Spark. See PIG-5140", !Util.isSparkExecType(cluster.getExecType()));
             assertEmptyOutputFile();
         } finally {
             new File(PIG_FILE).delete();

Modified: pig/trunk/test/org/apache/pig/test/TestEvalPipeline.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestEvalPipeline.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestEvalPipeline.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestEvalPipeline.java Mon May 29 15:00:39 2017
@@ -24,6 +24,7 @@ import java.io.PrintStream;
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -422,9 +423,15 @@ public class TestEvalPipeline {
         pigServer.registerQuery(query);
         Iterator<Tuple> iter = pigServer.openIterator("C");
         if(!iter.hasNext()) Assert.fail("No output found");
-        int numIdentity = 0;
+        List<Tuple> actualResList = new ArrayList<Tuple>();
         while(iter.hasNext()){
-            Tuple t = iter.next();
+            actualResList.add(iter.next());
+        }
+
+        Util.sortQueryOutputsIfNeed(actualResList, Util.isSparkExecType(cluster.getExecType()));
+
+        int numIdentity = 0;
+        for (Tuple t : actualResList) {
             Assert.assertEquals((Integer)numIdentity, (Integer)t.get(0));
             Assert.assertEquals((Long)5L, (Long)t.get(2));
             Assert.assertEquals(LOOP_COUNT*2.0, (Double)t.get(3), 0.01);
@@ -467,9 +474,15 @@ public class TestEvalPipeline {
         pigServer.registerQuery(query);
         Iterator<Tuple> iter = pigServer.openIterator("C");
         if(!iter.hasNext()) Assert.fail("No output found");
-        int numIdentity = 0;
+        List<Tuple> actualResList = new ArrayList<Tuple>();
         while(iter.hasNext()){
-            Tuple t = iter.next();
+            actualResList.add(iter.next());
+        }
+
+        Util.sortQueryOutputsIfNeed(actualResList, Util.isSparkExecType(cluster.getExecType()));
+
+        int numIdentity = 0;
+        for (Tuple t : actualResList) {
             Assert.assertEquals((Integer)numIdentity, (Integer)t.get(0));
             Assert.assertEquals((Long)5L, (Long)t.get(2));
             Assert.assertEquals(LOOP_COUNT*2.0, (Double)t.get(3), 0.01);
@@ -842,9 +855,15 @@ public class TestEvalPipeline {
         pigServer.registerQuery(query);
         Iterator<Tuple> iter = pigServer.openIterator("C");
         if(!iter.hasNext()) Assert.fail("No output found");
-        int numIdentity = 0;
+        List<Tuple> actualResList = new ArrayList<Tuple>();
         while(iter.hasNext()){
-            Tuple t = iter.next();
+            actualResList.add(iter.next());
+        }
+
+        Util.sortQueryOutputsIfNeed(actualResList, Util.isSparkExecType(cluster.getExecType()));
+
+        int numIdentity = 0;
+        for (Tuple t : actualResList) {
             Assert.assertEquals((Integer)((numIdentity + 1) * 10), (Integer)t.get(0));
             Assert.assertEquals((Long)10L, (Long)t.get(1));
             Assert.assertEquals((Long)5L, (Long)t.get(2));
@@ -873,6 +892,10 @@ public class TestEvalPipeline {
         pigServer.registerQuery("A = LOAD '"
                 + Util.generateURI(tmpFile.toString(), pigContext) + "';");
         pigServer.registerQuery("B = distinct A;");
+        if (Util.isSparkExecType(cluster.getExecType())) {
+            pigServer.registerQuery("B = order B by *;");
+        }
+
         String query = "C = foreach B {"
         + "C1 = $1 - $0;"
         + "C2 = $1%2;"
@@ -883,7 +906,6 @@ public class TestEvalPipeline {
         pigServer.registerQuery(query);
         Iterator<Tuple> iter = pigServer.openIterator("C");
         if(!iter.hasNext()) Assert.fail("No output found");
-
         int numRows = 0;
         for(int i = 0; i < LOOP_COUNT; i++) {
             for(int j = 0; j < LOOP_COUNT; j+=2){
@@ -920,6 +942,10 @@ public class TestEvalPipeline {
         pigServer.registerQuery("A = LOAD '"
                 + Util.generateURI(tmpFile.toString(), pigContext) + "';");
         pigServer.registerQuery("B = distinct A;");
+        if (Util.isSparkExecType(cluster.getExecType())) {
+            pigServer.registerQuery("B = order B by *;");
+        }
+
         String query = "C = foreach B {"
         + "C1 = $0 + $1;"
         + "C2 = C1 + $0;"
@@ -929,7 +955,6 @@ public class TestEvalPipeline {
         pigServer.registerQuery(query);
         Iterator<Tuple> iter = pigServer.openIterator("C");
         if(!iter.hasNext()) Assert.fail("No output found");
-
         int numRows = 0;
         for(int i = 0; i < LOOP_COUNT; i++) {
             for(int j = 0; j < LOOP_COUNT; j+=2){

Modified: pig/trunk/test/org/apache/pig/test/TestEvalPipeline2.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestEvalPipeline2.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestEvalPipeline2.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestEvalPipeline2.java Mon May 29 15:00:39 2017
@@ -138,9 +138,15 @@ public class TestEvalPipeline2 {
         pigServer.registerQuery(query);
         Iterator<Tuple> iter = pigServer.openIterator("C");
         if(!iter.hasNext()) Assert.fail("No output found");
-        int numIdentity = 0;
+        List<Tuple> actualResList = new ArrayList<Tuple>();
         while(iter.hasNext()){
-            Tuple tuple = iter.next();
+            actualResList.add(iter.next());
+        }
+
+        Util.sortQueryOutputsIfNeed(actualResList,Util.isSparkExecType(cluster.getExecType()));
+
+        int numIdentity = 0;
+        for (Tuple tuple : actualResList) {
             Tuple t = (Tuple)tuple.get(0);
             Assert.assertEquals(DataByteArray.class, t.get(0).getClass());
             int group = Integer.parseInt(new String(((DataByteArray)t.get(0)).get()));
@@ -476,17 +482,24 @@ public class TestEvalPipeline2 {
         pigServer.registerQuery("D = COGROUP C BY b0, A BY a0 PARALLEL 2;");
         Iterator<Tuple> iter = pigServer.openIterator("D");
 
-        Assert.assertTrue(iter.hasNext());
-        Tuple t = iter.next();
+        if (Util.isSparkExecType(cluster.getExecType())) {
+            String[] expectedResults =
+                new String[] {"(2,{(2,2)},{(2,5,2)})", "(1,{(1,1)},{(1,2,3)})" };
+            Util.checkQueryOutputs(iter, expectedResults,
+                org.apache.pig.newplan.logical.Util.translateSchema(pigServer.dumpSchema("D")), Util.isSparkExecType(cluster.getExecType()));
+        } else {
+            Assert.assertTrue(iter.hasNext());
+            Tuple t = iter.next();
 
-        Assert.assertTrue(t.toString().equals("(2,{(2,2)},{(2,5,2)})"));
+            Assert.assertTrue(t.toString().equals("(2,{(2,2)},{(2,5,2)})"));
 
-        Assert.assertTrue(iter.hasNext());
-        t = iter.next();
+            Assert.assertTrue(iter.hasNext());
+            t = iter.next();
 
-        Assert.assertTrue(t.toString().equals("(1,{(1,1)},{(1,2,3)})"));
+            Assert.assertTrue(t.toString().equals("(1,{(1,1)},{(1,2,3)})"));
 
-        Assert.assertFalse(iter.hasNext());
+            Assert.assertFalse(iter.hasNext());
+        }
     }
 
     // See PIG-1195
@@ -739,16 +752,10 @@ public class TestEvalPipeline2 {
         pigServer.registerQuery(query);
         Iterator<Tuple> iter = pigServer.openIterator("EventsPerMinute");
 
-        Tuple t = iter.next();
-        Assert.assertTrue( (Long)t.get(0) == 60000 && (Long)t.get(1) == 2 && (Long)t.get(2) == 3 );
-
-        t = iter.next();
-        Assert.assertTrue( (Long)t.get(0) == 120000 && (Long)t.get(1) == 2 && (Long)t.get(2) == 2 );
-
-        t = iter.next();
-        Assert.assertTrue( (Long)t.get(0) == 240000 && (Long)t.get(1) == 1 && (Long)t.get(2) == 1 );
+        List<Tuple> expectedResults = Util.getTuplesFromConstantTupleStrings(
+            new String[]{"(60000L,2L,3L)", "(120000L,2L,2L)", "(240000L,1L,1L)"});
 
-        Assert.assertFalse(iter.hasNext());
+        Util.checkQueryOutputs(iter, expectedResults, Util.isSparkExecType(cluster.getExecType()));
     }
 
     // See PIG-1729
@@ -1580,6 +1587,9 @@ public class TestEvalPipeline2 {
 
         pigServer.registerQuery("data = load 'table_testLimitFlatten' as (k,v);");
         pigServer.registerQuery("grouped = GROUP data BY k;");
+        if (Util.isSparkExecType(cluster.getExecType())) {
+            pigServer.registerQuery("grouped = ORDER grouped BY group;");
+        }
         pigServer.registerQuery("selected = LIMIT grouped 2;");
         pigServer.registerQuery("flattened = FOREACH selected GENERATE FLATTEN (data);");
 
@@ -1587,7 +1597,9 @@ public class TestEvalPipeline2 {
 
         String[] expected = new String[] {"(1,A)", "(1,B)", "(2,C)"};
 
-        Util.checkQueryOutputsAfterSortRecursive(iter, expected, org.apache.pig.newplan.logical.Util.translateSchema(pigServer.dumpSchema("flattened")));
+        Util.checkQueryOutputs(iter, expected,
+            org.apache.pig.newplan.logical.Util.translateSchema(pigServer.dumpSchema("flattened")), 
+            Util.isSparkExecType(cluster.getExecType()));
     }
 
     // See PIG-2237
@@ -1650,8 +1662,14 @@ public class TestEvalPipeline2 {
                     return false;
                 }
             });
-            // auto-parallelism is 2 in MR, 20 in Tez, so check >=2
-            Assert.assertTrue(partFiles.length >= 2);
+
+            if (Util.isSparkExecType(cluster.getExecType())) {
+                // TODO: Fix this when we implement auto-parallelism in Spark
+                Assert.assertTrue(partFiles.length == 1);
+            } else {
+                // auto-parallelism is 2 in MR, 20 in Tez, so check >=2
+                Assert.assertTrue(partFiles.length >= 2);
+            }
             // Check the output
             Iterator<Tuple> iter = job.getResults();
             List<Tuple> results = new ArrayList<Tuple>();

Modified: pig/trunk/test/org/apache/pig/test/TestEvalPipelineLocal.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestEvalPipelineLocal.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestEvalPipelineLocal.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestEvalPipelineLocal.java Mon May 29 15:00:39 2017
@@ -37,6 +37,7 @@ import java.util.Properties;
 import java.util.Random;
 import java.util.Set;
 import java.util.StringTokenizer;
+import java.util.Collections;
 
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.log4j.Appender;
@@ -884,7 +885,7 @@ public class TestEvalPipelineLocal {
         int LOOP_COUNT = 10;
         File tmpFile = File.createTempFile("test", "txt");
         PrintStream ps = new PrintStream(new FileOutputStream(tmpFile));
-        for(int i = 0; i < LOOP_COUNT; i++) {
+        for(int i=0; i<LOOP_COUNT; i++) {
             for(int j=0;j<LOOP_COUNT;j+=2){
                 ps.println(i+"\t"+j);
                 ps.println(i+"\t"+j);
@@ -908,18 +909,29 @@ public class TestEvalPipelineLocal {
         Iterator<Tuple> iter = pigServer.openIterator("C");
         if(!iter.hasNext()) Assert.fail("No output found");
         int numIdentity = 0;
+        // When running with spark, output can be in a different order than that 
+        // when running in mr mode.
+        List<Tuple> resList = new ArrayList<Tuple>();
         while(iter.hasNext()){
-            Tuple t = iter.next();
-            Assert.assertEquals((Integer)((numIdentity + 1) * 10), (Integer)t.get(0));
+            resList.add(iter.next());
+        }
+
+        numIdentity = resList.size();
+        Util.sortQueryOutputsIfNeed(resList, Util.isSparkExecType(Util.getLocalTestMode()));
+        Assert.assertEquals(LOOP_COUNT, numIdentity);
+        // Since delta differences in some cases are allowed, utility function 
+        // to compare tuple-lists cannot be used here.
+        // This loop generates sorted expected data
+        for (int i=0; i<numIdentity; i++) {
+            Tuple t = resList.get(i);
+            Assert.assertEquals((Integer)((i + 1) * 10), (Integer)t.get(0));
             Assert.assertEquals((Long)10L, (Long)t.get(1));
             Assert.assertEquals((Long)5L, (Long)t.get(2));
             Assert.assertEquals(LOOP_COUNT*2.0, (Double)t.get(3), 0.01);
             Assert.assertEquals(8.0, (Double)t.get(5), 0.01);
             Assert.assertEquals(5L, ((DataBag)t.get(6)).size());
             Assert.assertEquals(7, t.size());
-            ++numIdentity;
         }
-        Assert.assertEquals(LOOP_COUNT, numIdentity);
     }
 
     @Test
@@ -927,12 +939,25 @@ public class TestEvalPipelineLocal {
         int LOOP_COUNT = 10;
         File tmpFile = File.createTempFile("test", "txt");
         PrintStream ps = new PrintStream(new FileOutputStream(tmpFile));
+
+        List<Tuple> expectedList = new ArrayList<Tuple>();
         for(int i = 0; i < LOOP_COUNT; i++) {
             for(int j=0;j<LOOP_COUNT;j+=2){
                 ps.println(i+"\t"+j);
                 ps.println(i+"\t"+j);
+                //  Generating expected data
+                Tuple t = mTf.newTuple();
+                t.append(new Double(j - i));
+                t.append((Integer)(j%2));
+                if(j == 0) {
+                    t.append(0.0);
+                } else {
+                    t.append((Double)((double)i/j));
+                }
+                expectedList.add(t);
             }
         }
+        Util.sortQueryOutputsIfNeed(expectedList, Util.isSparkExecType(Util.getLocalTestMode()));
         ps.close();
 
         pigServer.registerQuery("A = LOAD '"
@@ -949,25 +974,30 @@ public class TestEvalPipelineLocal {
         pigServer.registerQuery(query);
         Iterator<Tuple> iter = pigServer.openIterator("C");
         if(!iter.hasNext()) Assert.fail("No output found");
+        // When ruuning with spark, output can be in a different order than when
+        // running in mr mode.
+        List<Tuple> resList = new ArrayList<Tuple>();
+        while(iter.hasNext()){
+            resList.add(iter.next());
+        }
 
-        int numRows = 0;
+        Util.sortQueryOutputsIfNeed(resList, Util.isSparkExecType(Util.getLocalTestMode()));
+        Assert.assertEquals((LOOP_COUNT * LOOP_COUNT)/2, resList.size());
+
+        // Since delta difference in some cases is allowed, utility function 
+        // to compare tuple-lists cannot be used here.
         for(int i = 0; i < LOOP_COUNT; i++) {
             for(int j = 0; j < LOOP_COUNT; j+=2){
-                Tuple t = null;
-                if(iter.hasNext()) t = iter.next();
-                Assert.assertEquals(3, t.size());
-                Assert.assertEquals(new Double(j - i), (Double)t.get(0), 0.01);
-                Assert.assertEquals((Integer)(j%2), (Integer)t.get(1));
-                if(j == 0) {
-                    Assert.assertEquals(0.0, (Double)t.get(2), 0.01);
-                } else {
-                    Assert.assertEquals((Double)((double)i/j), (Double)t.get(2), 0.01);
-                }
-                ++numRows;
+                int k = i*LOOP_COUNT/2 + j/2;
+                Tuple res_t = resList.get(k);
+                Tuple expec_t = expectedList.get(k);
+
+                Assert.assertEquals(expec_t.size(), res_t.size());
+                Assert.assertEquals((Double)expec_t.get(0), (Double)res_t.get(0), 0.01);
+                Assert.assertEquals((Integer)expec_t.get(1), (Integer)res_t.get(1));
+                Assert.assertEquals((Double)expec_t.get(2), (Double)res_t.get(2), 0.01);
             }
         }
-
-        Assert.assertEquals((LOOP_COUNT * LOOP_COUNT)/2, numRows);
     }
 
     @Test
@@ -975,10 +1005,16 @@ public class TestEvalPipelineLocal {
         int LOOP_COUNT = 10;
         File tmpFile = File.createTempFile("test", "txt");
         PrintStream ps = new PrintStream(new FileOutputStream(tmpFile));
+        List<Tuple> expectedList = new ArrayList<Tuple>();
         for(int i = 0; i < LOOP_COUNT; i++) {
             for(int j=0;j<LOOP_COUNT;j+=2){
                 ps.println(i+"\t"+j);
                 ps.println(i+"\t"+j);
+                // Generating expected data.
+                Tuple t = mTf.newTuple();
+                t.append(new Double(i+j));
+                t.append(new Double(i + j + i));
+                expectedList.add(t);
             }
         }
         ps.close();
@@ -996,20 +1032,9 @@ public class TestEvalPipelineLocal {
         pigServer.registerQuery(query);
         Iterator<Tuple> iter = pigServer.openIterator("C");
         if(!iter.hasNext()) Assert.fail("No output found");
-
-        int numRows = 0;
-        for(int i = 0; i < LOOP_COUNT; i++) {
-            for(int j = 0; j < LOOP_COUNT; j+=2){
-                Tuple t = null;
-                if(iter.hasNext()) t = iter.next();
-                Assert.assertEquals(2, t.size());
-                Assert.assertEquals(new Double(i + j), (Double)t.get(0), 0.01);
-                Assert.assertEquals(new Double(i + j + i), (Double)t.get(1));
-                ++numRows;
-            }
-        }
-
-        Assert.assertEquals((LOOP_COUNT * LOOP_COUNT)/2, numRows);
+        // When ruuning with spark, output can be in a different order than that
+        // when running in mr mode.
+        Util.checkQueryOutputs(iter, expectedList, Util.isSparkExecType(Util.getLocalTestMode()));
     }
 
     @Test
@@ -1128,9 +1153,15 @@ public class TestEvalPipelineLocal {
         pigServer.registerQuery("b = foreach a generate TOTUPLE(x, y) as t, z;");
         pigServer.registerQuery("c = group b by t;");
         Iterator<Tuple> iter = pigServer.openIterator("c");
-        Assert.assertTrue(iter.next().toString().equals("((1,2),{((1,2),3)})"));
-        Assert.assertTrue(iter.next().toString().equals("((4,5),{((4,5),6)})"));
-        Assert.assertFalse(iter.hasNext());
+        // When ruuning with spark, output can be in a different order than that 
+        // when running in mr mode.
+        List<Tuple> expectedRes =
+                Util.getTuplesFromConstantTupleStrings(
+                        new String[] {
+                                "((1,2),{((1,2),3)})",
+                                "((4,5),{((4,5),6)})"
+                        });
+        Util.checkQueryOutputs(iter, expectedRes, Util.isSparkExecType(Util.getLocalTestMode()));
     }
     
     @Test
@@ -1297,7 +1328,7 @@ public class TestEvalPipelineLocal {
 
         logger.removeAppender(appender);
 
-        Assert.assertTrue(bos.toString().contains("New For Each(false,false)[tuple]"));
+        Assert.assertTrue(bos.toString().contains("New For Each"));
     }
 
     @Test

Modified: pig/trunk/test/org/apache/pig/test/TestFinish.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestFinish.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestFinish.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestFinish.java Mon May 29 15:00:39 2017
@@ -45,7 +45,7 @@ public class TestFinish {
     BagFactory mBf = BagFactory.getInstance();
     File f1;
 
-    static MiniGenericCluster cluster = MiniGenericCluster.buildCluster();
+    private static MiniGenericCluster cluster = MiniGenericCluster.buildCluster();
 
     static public class MyEvalFunction extends EvalFunc<Tuple> {
         String execType;

Modified: pig/trunk/test/org/apache/pig/test/TestFlatten.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestFlatten.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestFlatten.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestFlatten.java Mon May 29 15:00:39 2017
@@ -95,6 +95,6 @@ public class TestFlatten {
         List<Tuple> expectedResults = Util.getTuplesFromConstantTupleStrings(
                 new String[] {
                 "('a', 'b', '1', '2')", "('a', 'b', '3', '4')", "('c', 'd', '1', '2')", "('c', 'd', '3', '4')" });
-        Util.checkQueryOutputs(actualResults.iterator(), expectedResults);
+        Util.checkQueryOutputsAfterSort(actualResults.iterator(), expectedResults);
     }
 }

Modified: pig/trunk/test/org/apache/pig/test/TestForEachNestedPlanLocal.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestForEachNestedPlanLocal.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestForEachNestedPlanLocal.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestForEachNestedPlanLocal.java Mon May 29 15:00:39 2017
@@ -101,14 +101,8 @@ public class TestForEachNestedPlanLocal
         pig.registerQuery("D = foreach C {"
                 + "crossed = cross user, session;"
                 + "generate crossed;" + "}");
-        Iterator<Tuple> expectedItr = expectedResults.iterator();
         Iterator<Tuple> actualItr = pig.openIterator("D");
-        while (expectedItr.hasNext() && actualItr.hasNext()) {
-            Tuple expectedTuple = expectedItr.next();
-            Tuple actualTuple = actualItr.next();
-            assertEquals(expectedTuple, actualTuple);
-        }
-        assertEquals(expectedItr.hasNext(), actualItr.hasNext());
+        Util.checkQueryOutputsAfterSort(actualItr, expectedResults);
     }
 
     @Test
@@ -156,14 +150,8 @@ public class TestForEachNestedPlanLocal
                 + "crossed = cross user, distinct_session;"
                 + "filtered = filter crossed by user::region == distinct_session::region;"
                 + "generate filtered;" + "}");
-        Iterator<Tuple> expectedItr = expectedResults.iterator();
         Iterator<Tuple> actualItr = pig.openIterator("D");
-        while (expectedItr.hasNext() && actualItr.hasNext()) {
-            Tuple expectedTuple = expectedItr.next();
-            Tuple actualTuple = actualItr.next();
-            assertEquals(expectedTuple, actualTuple);
-        }
-        assertEquals(expectedItr.hasNext(), actualItr.hasNext());
+        Util.checkQueryOutputsAfterSort(actualItr, expectedResults);
     }
 
     @Test
@@ -187,14 +175,8 @@ public class TestForEachNestedPlanLocal
         pig.registerQuery("D = foreach C {"
                 + "crossed = cross user, session, profile;"
                 + "generate crossed;" + "}");
-        Iterator<Tuple> expectedItr = expectedResults.iterator();
         Iterator<Tuple> actualItr = pig.openIterator("D");
-        while (expectedItr.hasNext() && actualItr.hasNext()) {
-            Tuple expectedTuple = expectedItr.next();
-            Tuple actualTuple = actualItr.next();
-            assertEquals(expectedTuple, actualTuple);
-        }
-        assertEquals(expectedItr.hasNext(), actualItr.hasNext());
+        Util.checkQueryOutputsAfterSort(actualItr, expectedResults);
     }
 
     /*

Modified: pig/trunk/test/org/apache/pig/test/TestGrunt.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestGrunt.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestGrunt.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestGrunt.java Mon May 29 15:00:39 2017
@@ -70,7 +70,6 @@ import org.junit.BeforeClass;
 import org.junit.Test;
 
 public class TestGrunt {
-
     static MiniGenericCluster cluster = MiniGenericCluster.buildCluster();
     private String basedir = "test/org/apache/pig/test/data";
 
@@ -915,6 +914,15 @@ public class TestGrunt {
 
     @Test
     public void testKeepGoigFailed() throws Throwable {
+        // in mr mode, the output file 'baz' will be automatically deleted if the mr job fails
+        // when "cat baz;" is executed, it throws "Encountered IOException. Directory baz does not exist"
+        // in GruntParser#processCat() and variable "caught" is true
+        // in spark mode, the output file 'baz' will not be automatically deleted even the job fails(see SPARK-7953)
+        // when "cat baz;" is executed, it does not throw exception and the variable "caught" is false
+        // TODO: Enable this for Spark when SPARK-7953 is resolved
+        Assume.assumeTrue(
+            "Skip this test for Spark until SPARK-7953 is resolved!",
+            !Util.isSparkExecType(cluster.getExecType()));
         PigServer server = new PigServer(cluster.getExecType(), cluster.getProperties());
         PigContext context = server.getPigContext();
         Util.copyFromLocalToCluster(cluster, "test/org/apache/pig/test/data/passwd", "passwd");
@@ -936,7 +944,6 @@ public class TestGrunt {
         InputStreamReader reader = new InputStreamReader(cmd);
 
         Grunt grunt = new Grunt(new BufferedReader(reader), context);
-
         boolean caught = false;
         try {
             grunt.exec();
@@ -1004,7 +1011,13 @@ public class TestGrunt {
             grunt.exec();
         } catch (PigException e) {
             caught = true;
-            assertTrue(e.getErrorCode() == 6017);
+            if (!Util.isSparkExecType(cluster.getExecType())) {
+                assertTrue(e.getErrorCode() == 6017);
+            } else {
+                //In spark mode, We wrap ExecException to RunTimeException and is thrown out in JobGraphBuilder#sparkOperToRDD,
+                //So unwrap the exception here
+                assertTrue(((ExecException) e.getCause()).getErrorCode() == 6017);
+            }
         }
 
         if (Util.isMapredExecType(cluster.getExecType())) {
@@ -1621,7 +1634,7 @@ public class TestGrunt {
         boolean found = false;
         for (String line : lines) {
             if (line.matches(".*Added jar .*" + jarName + ".*")) {
-                // MR mode
+                // MR and Spark mode
                 found = true;
             } else if (line.matches(".*Local resource.*" + jarName + ".*")) {
                 // Tez mode

Modified: pig/trunk/test/org/apache/pig/test/TestHBaseStorage.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestHBaseStorage.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestHBaseStorage.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestHBaseStorage.java Mon May 29 15:00:39 2017
@@ -18,6 +18,7 @@ package org.apache.pig.test;
 
 import java.io.File;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -45,6 +46,7 @@ import org.apache.pig.data.DataBag;
 import org.apache.pig.data.DataByteArray;
 import org.apache.pig.data.DataType;
 import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Assert;
@@ -1443,23 +1445,20 @@ public class TestHBaseStorage {
 
         Iterator<Tuple> it = pig.openIterator("c");
         int index = 0;
-        while (it.hasNext()) {
-            Tuple t = it.next();
-            String rowKey = (String) t.get(0);
-            int col_a = (Integer) t.get(1);
-            Assert.assertNotNull(t.get(2));
-            double col_b = (Double) t.get(2);
-            String col_c = (String) t.get(3);
-
-            Assert.assertEquals("00".substring((index + "").length()) + index,
-                    rowKey);
-            Assert.assertEquals(index, col_a);
-            Assert.assertEquals(index + 0.0, col_b, 1e-6);
-            Assert.assertEquals("Text_" + index, col_c);
+        List<Tuple> expected = new ArrayList<Tuple>();
+        while (index<TEST_ROW_COUNT) {
+            Tuple t = TupleFactory.getInstance().newTuple();
+            t.append("00".substring((index + "").length()) + index);
+            t.append(index);
+            t.append(index + 0.0);
+            t.append("Text_" + index);
+            t.append(index);
+            t.append(new DataByteArray("Text_" + index));
             index++;
+            expected.add(t);
         }
-        Assert.assertEquals(index, TEST_ROW_COUNT);
-    }
+        Util.checkQueryOutputsAfterSort(it, expected);
+}
 
     @Test
     // See PIG-4151

Modified: pig/trunk/test/org/apache/pig/test/TestLimitVariable.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestLimitVariable.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestLimitVariable.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestLimitVariable.java Mon May 29 15:00:39 2017
@@ -80,15 +80,17 @@ public class TestLimitVariable {
         Iterator<Tuple> it = pigServer.openIterator("g");
 
         List<Tuple> expectedRes = Util.getTuplesFromConstantTupleStrings(new String[] {
-                "(5.0,36)"});
-        Util.checkQueryOutputs(it, expectedRes);
+                "(5.0,36L)"});
+        Util.checkQueryOutputsAfterSort(it, expectedRes);
         pigServer.getPigContext().getProperties().remove(PigConfiguration.PIG_EXEC_MAP_PARTAGG);
     }
 
     @Test
     public void testLimitVariable2() throws IOException {
+        //add field type here to use  Util.checkQueryOutputsAfterSort comparing the expected and actual
+        //results
         String query =
-            "a = load '" + inputFile.getName() + "' as (id, num);" +
+            "a = load '" + inputFile.getName() + "' as (id:int, num:int);" +
             "b = filter a by id == 2;" + // only 1 tuple returned (2,3)
             "c = order a by id ASC;" +
             "d = limit c b.num;" + // test bytearray to long implicit cast
@@ -99,18 +101,20 @@ public class TestLimitVariable {
         Iterator<Tuple> itD = pigServer.openIterator("d");
         List<Tuple> expectedResD = Util.getTuplesFromConstantTupleStrings(new String[] {
                 "(1,11)", "(2,3)", "(3,10)" });
-        Util.checkQueryOutputs(itD, expectedResD);
+        Util.checkQueryOutputsAfterSort(itD, expectedResD);
 
         Iterator<Tuple> itE = pigServer.openIterator("e");
         List<Tuple> expectedResE = Util.getTuplesFromConstantTupleStrings(new String[] {
                 "(1,11)", "(2,3)", "(3,10)", "(4,11)", "(5,10)", "(6,15)" });
-        Util.checkQueryOutputs(itE, expectedResE);
+        Util.checkQueryOutputsAfterSort(itE, expectedResE);
     }
 
     @Test
     public void testLimitVariable3() throws IOException {
+        //add field type here to use  Util.checkQueryOutputsAfterSort comparing the expected and actual
+        //results
         String query =
-            "a = load '" + inputFile.getName() + "' ;" +
+            "a = load '" + inputFile.getName() + "' as (id:int, num:int);" +
             "b = group a all;" +
             "c = foreach b generate COUNT(a) as sum;" +
             "d = order a by $0 ASC;" +
@@ -121,7 +125,7 @@ public class TestLimitVariable {
         Iterator<Tuple> itE = pigServer.openIterator("e");
         List<Tuple> expectedResE = Util.getTuplesFromConstantTupleStrings(new String[] {
                 "(1,11)", "(2,3)", "(3,10)", "(4,11)", "(5,10)", "(6,15)" });
-        Util.checkQueryOutputs(itE, expectedResE);
+        Util.checkQueryOutputsAfterSort(itE, expectedResE);
     }
 
     @Test

Modified: pig/trunk/test/org/apache/pig/test/TestLineageFindRelVisitor.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestLineageFindRelVisitor.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestLineageFindRelVisitor.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestLineageFindRelVisitor.java Mon May 29 15:00:39 2017
@@ -259,9 +259,10 @@ public class TestLineageFindRelVisitor {
         pig.registerQuery("E = FOREACH D GENERATE (chararray) tupleD.a1;\n");
         Iterator<Tuple> iter  = pig.openIterator("E");
 
-        Assert.assertEquals("123", iter.next().get(0));
-        Assert.assertEquals("456", iter.next().get(0));
-        Assert.assertEquals("789", iter.next().get(0));
+        Util.checkQueryOutputs(iter,
+                new String[]{"(123)", "(456)", "(789)"},
+                org.apache.pig.newplan.logical.Util.translateSchema(pig.dumpSchema("E")), Util.isSparkExecType(Util
+                .getLocalTestMode()));
     }
 
     @Test

Modified: pig/trunk/test/org/apache/pig/test/TestMapSideCogroup.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestMapSideCogroup.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestMapSideCogroup.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestMapSideCogroup.java Mon May 29 15:00:39 2017
@@ -311,12 +311,11 @@ public class TestMapSideCogroup {
         pigServer.registerQuery("A = LOAD '" + INPUT_FILE1 + "," + INPUT_FILE4 + "' using "+ DummyCollectableLoader.class.getName() +"() as (c1:chararray,c2:int);");
         pigServer.registerQuery("B = LOAD '" + INPUT_FILE5 + "' using "+ DummyIndexableLoader.class.getName()   +"() as (c1:chararray,c2:int);");
 
-        DataBag dbMergeCogrp = BagFactory.getInstance().newDefaultBag();
+        List<Tuple> dbMergeCogrp = new ArrayList<Tuple>();
 
         pigServer.registerQuery("C = cogroup A by c1, B by c1 using 'merge';");
         Iterator<Tuple> iter = pigServer.openIterator("C");
 
-
         while(iter.hasNext()) {
             Tuple t = iter.next();
             dbMergeCogrp.add(t);
@@ -335,12 +334,29 @@ public class TestMapSideCogroup {
                 "(3,{(3,3),(3,2),(3,1)},{(3,1),(3,2),(3,3)})"
         };
 
-        assertEquals(9, dbMergeCogrp.size());
+        List<Tuple> expected = Util.getTuplesFromConstantTupleStrings(results);
+
+        //We need sort dbMergeCogrp because the result is different in sequence between spark and other mode when
+        //multiple files are loaded(LOAD INPUT_FILE1,INPUT_FILE4...)
+        for (Tuple t : dbMergeCogrp) {
+            Util.convertBagToSortedBag(t);
+        }
+        for (Tuple t : expected) {
+            Util.convertBagToSortedBag(t);
+        }
+
+        Collections.sort(dbMergeCogrp);
+        Collections.sort(expected);
+        assertEquals(dbMergeCogrp.size(), expected.size());
+
+        //Since TestMapSideCogroup.DummyIndexableLoader.getNext() does not
+        //apply schema for each input tuple,Util#checkQueryOutputsAfterSortRecursive fails to assert.
+        // The schema for C is (int,{(chararray,int),(chararray,int),(chararray,int)},{(chararray,int),(chararray,int),(chararray,int)}).
+        //But the schema for result "dbMergeCogrp" is (int,{(chararray,int),(chararray,int),(chararray,int)},{(chararray,chararray),(chararray,chararray),(chararray,chararray)})
         Iterator<Tuple> itr = dbMergeCogrp.iterator();
-        for(int i=0; i<9; i++){
-            assertEquals(itr.next().toString(), results[i]);   
+        for (int i = 0; i < dbMergeCogrp.size(); i++) {
+            assertEquals(itr.next().toString(), expected.get(i).toString());
         }
-        assertFalse(itr.hasNext());
     }
 
     @Test

Modified: pig/trunk/test/org/apache/pig/test/TestMergeJoinOuter.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestMergeJoinOuter.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestMergeJoinOuter.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestMergeJoinOuter.java Mon May 29 15:00:39 2017
@@ -168,7 +168,7 @@ public class TestMergeJoinOuter {
     
     @Test
     public void testLeftOuter() throws IOException {
-        
+
         pigServer.registerQuery("A = LOAD '"+INPUT_FILE1+"' using "+ DummyCollectableLoader.class.getName() +"() as (c1:chararray, c2:chararray);");
         pigServer.registerQuery("B = LOAD '"+INPUT_FILE2+"' using "+ DummyIndexableLoader.class.getName() +"() as (c1:chararray, c2:chararray);");
 
@@ -197,7 +197,7 @@ public class TestMergeJoinOuter {
     
     @Test
     public void testRightOuter() throws IOException{
-        
+
         pigServer.registerQuery("A = LOAD '"+INPUT_FILE1+"' using "+ DummyCollectableLoader.class.getName() +"() as (c1:chararray, c2:chararray);");
         pigServer.registerQuery("B = LOAD '"+INPUT_FILE2+"' using "+ DummyIndexableLoader.class.getName() +"() as (c1:chararray, c2:chararray);");
         pigServer.registerQuery("C = join A by c1 right, B by c1 using 'merge';");
@@ -224,7 +224,7 @@ public class TestMergeJoinOuter {
     
     @Test
     public void testFullOuter() throws IOException{
-        
+
         pigServer.registerQuery("A = LOAD '"+INPUT_FILE1+"' using "+ DummyCollectableLoader.class.getName() +"() as (c1:chararray, c2:chararray);");
         pigServer.registerQuery("B = LOAD '"+INPUT_FILE2+"' using "+ DummyIndexableLoader.class.getName() +"() as (c1:chararray, c2:chararray);");
         pigServer.registerQuery("C = join A by c1 full, B by c1 using 'merge';");

Modified: pig/trunk/test/org/apache/pig/test/TestMultiQuery.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestMultiQuery.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestMultiQuery.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestMultiQuery.java Mon May 29 15:00:39 2017
@@ -31,6 +31,9 @@ import org.apache.pig.backend.executione
 import org.apache.pig.builtin.mock.Storage;
 import org.apache.pig.data.Tuple;
 import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.tools.pigstats.InputStats;
+
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Before;
@@ -106,33 +109,25 @@ public class TestMultiQuery {
         myPig.registerQuery("E = load 'output1' as (a:int, b:int);");
         Iterator<Tuple> iter = myPig.openIterator("E");
 
-        List<Tuple> expectedResults = Util.getTuplesFromConstantTupleStrings(
-                new String[] {
+        String[] expectedResults = new String[]{
                         "(1,2)",
                         "(2,3)"
-                });
+        };
+        Schema s = myPig.dumpSchema("E");
+        Util.checkQueryOutputs(iter, expectedResults, org.apache.pig.newplan.logical.Util.translateSchema(s), Util
+                .isSparkExecType(Util.getLocalTestMode()));
 
-        int counter = 0;
-        while (iter.hasNext()) {
-            assertEquals(expectedResults.get(counter++).toString(), iter.next().toString());
-        }
-        assertEquals(expectedResults.size(), counter);
 
         myPig.registerQuery("E = load 'output2' as (a:int, b:int);");
         iter = myPig.openIterator("E");
 
-        expectedResults = Util.getTuplesFromConstantTupleStrings(
-                new String[] {
+        expectedResults = new String[]{
                         "(2,3)",
                         "(3,4)"
-                });
-
-        counter = 0;
-        while (iter.hasNext()) {
-            assertEquals(expectedResults.get(counter++).toString(), iter.next().toString());
-        }
-
-        assertEquals(expectedResults.size(), counter);
+        };
+        s = myPig.dumpSchema("E");
+        Util.checkQueryOutputs(iter, expectedResults, org.apache.pig.newplan.logical.Util.translateSchema(s), Util
+                .isSparkExecType(Util.getLocalTestMode()));
     }
 
     @Test
@@ -165,20 +160,15 @@ public class TestMultiQuery {
 
         Iterator<Tuple> iter = myPig.openIterator("F");
 
-        List<Tuple> expectedResults = Util.getTuplesFromConstantTupleStrings(
-                new String[] {
+        String[] expectedResults = new String[]{
                         "(1,2)",
                         "(2,3)",
                         "(3,5)",
                         "(5,6)"
-                });
-
-        int counter = 0;
-        while (iter.hasNext()) {
-            assertEquals(expectedResults.get(counter++).toString(), iter.next().toString());
-        }
-
-        assertEquals(expectedResults.size(), counter);
+        };
+        Schema s = myPig.dumpSchema("F");
+        Util.checkQueryOutputs(iter, expectedResults, org.apache.pig.newplan.logical.Util.translateSchema(s), Util
+                .isSparkExecType(Util.getLocalTestMode()));
     }
 
     @Test
@@ -299,19 +289,14 @@ public class TestMultiQuery {
 
         Iterator<Tuple> iter = myPig.openIterator("E");
 
-        List<Tuple> expectedResults = Util.getTuplesFromConstantTupleStrings(
-                new String[] {
-                        "(1L,'apple',3,1L,'apple',1L,{(1L)})",
-                        "(2L,'orange',4,2L,'orange',2L,{(2L)})",
-                        "(3L,'persimmon',5,3L,'persimmon',3L,{(3L)})"
-                });
-
-        int counter = 0;
-        while (iter.hasNext()) {
-            assertEquals(expectedResults.get(counter++).toString(), iter.next().toString());
-        }
-
-        assertEquals(expectedResults.size(), counter);
+        String[] expectedResults = new String[]{
+                "(1L,apple,3,1L,apple,1L,{(1L)})",
+                "(2L,orange,4,2L,orange,2L,{(2L)})",
+                "(3L,persimmon,5,3L,persimmon,3L,{(3L)})"
+        };
+        Schema s = myPig.dumpSchema("E");
+        Util.checkQueryOutputs(iter, expectedResults, org.apache.pig.newplan.logical.Util.translateSchema(s), Util
+                .isSparkExecType(Util.getLocalTestMode()));
     }
 
     @Test
@@ -345,19 +330,14 @@ public class TestMultiQuery {
 
         Iterator<Tuple> iter = myPig.openIterator("joined_session_info");
 
-        List<Tuple> expectedResults = Util.getTuplesFromConstantTupleStrings(
-                new String[] {
-                        "('apple',{},{('apple','jar',1L)})",
-                        "('orange',{},{('orange','box',1L)})",
-                        "('strawberry',{(30,'strawberry','quit','bot')},{})"
-                });
-
-        int counter = 0;
-        while (iter.hasNext()) {
-            assertEquals(expectedResults.get(counter++).toString(), iter.next().toString());
-        }
-
-        assertEquals(expectedResults.size(), counter);
+        String[] expectedResults = new String[]{
+                "(apple,{},{(apple,jar,1L)})",
+                "(orange,{},{(orange,box,1L)})",
+                "(strawberry,{(30,strawberry,quit,bot)},{})"};
+
+        Schema s = myPig.dumpSchema("joined_session_info");
+        Util.checkQueryOutputs(iter, expectedResults, org.apache.pig.newplan.logical.Util.translateSchema(s), Util
+                .isSparkExecType(Util.getLocalTestMode()));
     }
 
     @Test
@@ -861,12 +841,12 @@ public class TestMultiQuery {
         List<Tuple> actualResults = data.get("output1");
         List<Tuple> expectedResults = Util.getTuplesFromConstantTupleStrings(
                 new String[] {"(1)", "(2)"});
-        Util.checkQueryOutputs(actualResults.iterator(), expectedResults);
+        Util.checkQueryOutputsAfterSort(actualResults.iterator(), expectedResults);
 
         actualResults = data.get("output2");
         expectedResults = Util.getTuplesFromConstantTupleStrings(
                 new String[] {"(1, 'world')", "(2, 'world')"});
-        Util.checkQueryOutputs(actualResults.iterator(), expectedResults);
+        Util.checkQueryOutputsAfterSort(actualResults.iterator(), expectedResults);
     }
 
     @Test
@@ -908,6 +888,27 @@ public class TestMultiQuery {
     }
 
     @Test
+    public void testMultiQueryJiraPig4899() throws Exception {
+        myPig.setBatchOn();
+
+        myPig.registerQuery("a = load 'passwd' "
+                + "using PigStorage(':') as (uname:chararray, passwd:chararray, uid:int, gid:int);");
+        myPig.registerQuery("b1 = foreach a generate uname;");
+        myPig.registerQuery("b2 = foreach a generate uid;");
+        myPig.registerQuery("store b1 into 'output1';");
+        myPig.registerQuery("store b2 into 'output2';");
+
+        List<ExecJob> jobs = myPig.executeBatch();
+        for (ExecJob job : jobs) {
+            assertTrue(job.getStatus() == ExecJob.JOB_STATUS.COMPLETED);
+            List<InputStats> stats = job.getStatistics().getInputStats();
+            assertEquals(1,stats.size());
+            InputStats stat = stats.get(0);
+            assertEquals("Number of records in passwd file is 14",14,stat.getNumberRecords());
+        }
+    }
+
+    @Test
     public void testMultiQueryJiraPig4883() throws Exception {
         Storage.Data data = Storage.resetData(myPig);
         data.set("inputLocation",
@@ -934,20 +935,24 @@ public class TestMultiQuery {
 
         List<Tuple> actualResults = data.get("output1");
         String[] expectedResults = new String[]{"(12, 1)"};
-        Util.checkQueryOutputsAfterSortRecursive(actualResults.iterator(), expectedResults, org.apache.pig.newplan.logical.Util.translateSchema(myPig.dumpSchema("B1")));
+        Util.checkQueryOutputs(actualResults.iterator(), expectedResults, org.apache.pig.newplan
+                .logical.Util.translateSchema(myPig.dumpSchema("B1")), Util.isSparkExecType(Util.getLocalTestMode()));
 
 
         actualResults = data.get("output2");
         expectedResults = new String[]{"(c,1)"};
-        Util.checkQueryOutputsAfterSortRecursive(actualResults.iterator(), expectedResults, org.apache.pig.newplan.logical.Util.translateSchema(myPig.dumpSchema("B2")));
+        Util.checkQueryOutputs(actualResults.iterator(), expectedResults, org.apache.pig.newplan.logical.Util
+                .translateSchema(myPig.dumpSchema("B2")), Util.isSparkExecType(Util.getLocalTestMode()));
 
         actualResults = data.get("output3");
         expectedResults = new String[]{"(-12, 1)"};
-        Util.checkQueryOutputsAfterSortRecursive(actualResults.iterator(), expectedResults, org.apache.pig.newplan.logical.Util.translateSchema(myPig.dumpSchema("C1")));
+        Util.checkQueryOutputs(actualResults.iterator(), expectedResults, org.apache.pig.newplan.logical.Util
+                .translateSchema(myPig.dumpSchema("C1")), Util.isSparkExecType(Util.getLocalTestMode()));
 
         actualResults = data.get("output4");
         expectedResults = new String[]{"(d,1)"};
-        Util.checkQueryOutputsAfterSortRecursive(actualResults.iterator(), expectedResults, org.apache.pig.newplan.logical.Util.translateSchema(myPig.dumpSchema("C2")));
+        Util.checkQueryOutputs(actualResults.iterator(), expectedResults, org.apache.pig.newplan.logical.Util
+                .translateSchema(myPig.dumpSchema("C2")), Util.isSparkExecType(Util.getLocalTestMode()));
     }
 
     // --------------------------------------------------------------------------

Modified: pig/trunk/test/org/apache/pig/test/TestMultiQueryLocal.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestMultiQueryLocal.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestMultiQueryLocal.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestMultiQueryLocal.java Mon May 29 15:00:39 2017
@@ -487,6 +487,7 @@ public class TestMultiQueryLocal {
     public void testMultiQueryWithIllustrate() throws Exception {
 
         Assume.assumeTrue("illustrate does not work in tez (PIG-3993)", !Util.getLocalTestMode().toString().startsWith("TEZ"));
+        Assume.assumeTrue("illustrate does not work in spark (PIG-4621)", !Util.getLocalTestMode().toString().startsWith("SPARK"));
         System.out.println("===== test multi-query with illustrate =====");
 
         try {

Modified: pig/trunk/test/org/apache/pig/test/TestNativeMapReduce.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestNativeMapReduce.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestNativeMapReduce.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestNativeMapReduce.java Mon May 29 15:00:39 2017
@@ -29,6 +29,7 @@ import java.util.List;
 
 import org.apache.hadoop.mapred.FileAlreadyExistsException;
 import org.apache.pig.PigServer;
+import org.apache.pig.backend.executionengine.ExecException;
 import org.apache.pig.backend.executionengine.ExecJob;
 import org.apache.pig.backend.executionengine.ExecJob.JOB_STATUS;
 import org.apache.pig.backend.hadoop.executionengine.JobCreationException;
@@ -210,6 +211,13 @@ public class TestNativeMapReduce  {
         } catch (JobCreationException e) {
             // Running in Tez mode throw exception
             assertTrue(e.getCause() instanceof FileAlreadyExistsException);
+        } catch (ExecException e) {
+            // Running in spark mode throw exception
+            if (e.getCause() instanceof RuntimeException) {
+                RuntimeException re = (RuntimeException) e.getCause();
+                JobCreationException jce = (JobCreationException) re.getCause();
+                assertTrue(jce.getCause() instanceof FileAlreadyExistsException);
+            }
         }
         finally{
             // We have to manually delete intermediate mapreduce files

Modified: pig/trunk/test/org/apache/pig/test/TestNullConstant.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestNullConstant.java?rev=1796639&r1=1796638&r2=1796639&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestNullConstant.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestNullConstant.java Mon May 29 15:00:39 2017
@@ -109,16 +109,7 @@ public class TestNullConstant {
         pigServer.registerQuery("d = foreach c generate flatten((SIZE(a) == 0 ? null: a)), flatten((SIZE(b) == 0 ? null : b));");
         Iterator<Tuple> it = pigServer.openIterator("d");
         Object[][] results = new Object[][]{{10, "will_join", 10, "will_join"}, {11, "will_not_join", null}, {null, 12, "will_not_join"}};
-        int i = 0;
-        while(it.hasNext()) {
-
-            Tuple t = it.next();
-            Object[] result = results[i++];
-            assertEquals(result.length, t.size());
-            for (int j = 0; j < result.length; j++) {
-                assertEquals(result[j], t.get(j));
-            }
-        }
+        Util.checkQueryOutputsAfterSort(it,results);
     }
 
     @Test



Mime
View raw message