/*
 * 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.io.PrintStream;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Random;

import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.pig.EvalFunc;
import org.apache.pig.PigConfiguration;
import org.apache.pig.PigException;
import org.apache.pig.PigServer;
import org.apache.pig.backend.executionengine.ExecJob;
import org.apache.pig.builtin.BinStorage;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.DataType;
import org.apache.pig.data.DefaultBagFactory;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.PigImplConstants;
import org.apache.pig.impl.io.FileLocalizer;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.impl.util.LogUtils;
import org.apache.pig.impl.util.ObjectSerializer;
import org.apache.pig.test.utils.Identity;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;

@RunWith(JUnit4.class)
public class TestEvalPipeline2 {

    static MiniGenericCluster cluster = MiniGenericCluster.buildCluster();
    private PigServer pigServer;

    TupleFactory mTf = TupleFactory.getInstance();
    BagFactory mBf = BagFactory.getInstance();

    @Before
    public void setUp() throws Exception{
        pigServer = new PigServer(cluster.getExecType(), cluster.getProperties());
    }

    @AfterClass
    public static void oneTimeTearDown() throws Exception {
        cluster.shutDown();
    }

    @Test
    public void testUdfInputOrder() throws IOException {
        String[] input = {
                "(123)",
                "((123)",
                "(123123123123)",
                "(asdf)"
        };

        Util.createInputFile(cluster, "table_udfInp", input);
        pigServer.registerQuery("a = load 'table_udfInp' as (i:int);");
        pigServer.registerQuery("b = foreach a {dec = 'hello'; str1 = " +  Identity.class.getName() +
                    "(dec,'abc','def');" +
                    "generate dec,str1; };");
        Iterator<Tuple> it = pigServer.openIterator("b");

        Tuple tup=null;

        //tuple 1
        tup = it.next();
        Tuple out = (Tuple)tup.get(1);

        Assert.assertEquals( out.get(0).toString(), "hello");
        Assert.assertEquals(out.get(1).toString(), "abc");
        Assert.assertEquals(out.get(2).toString(), "def");

        Util.deleteFile(cluster, "table_udfInp");
    }


    @Test
    public void testUDFwithStarInput() throws Exception {
        int LOOP_COUNT = 10;
        File tmpFile = Util.createTempFileDelOnExit("test", "txt");
        PrintStream ps = new PrintStream(new FileOutputStream(tmpFile));
        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);
            }
        }
        ps.close();

        pigServer.registerQuery("A = LOAD '"
                + Util.generateURI(tmpFile.toString(), pigServer
                        .getPigContext()) + "';");
        pigServer.registerQuery("B = group A by $0;");
        String query = "C = foreach B {"
        + "generate " + Identity.class.getName() +"(*);"
        + "};";

        pigServer.registerQuery(query);
        Iterator<Tuple> iter = pigServer.openIterator("C");
        if(!iter.hasNext()) Assert.fail("No output found");
        List<Tuple> actualResList = new ArrayList<Tuple>();
        while(iter.hasNext()){
            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()));
            Assert.assertEquals(numIdentity, group);
            Assert.assertTrue(t.get(1) instanceof DataBag);
            DataBag bag = (DataBag)t.get(1);
            Assert.assertEquals(10, bag.size());
            Assert.assertEquals(2, t.size());
            ++numIdentity;
        }
        Assert.assertEquals(LOOP_COUNT, numIdentity);

    }
    @Test
    public void testBinStorageByteArrayCastsSimple() throws IOException {
        // Test for PIG-544 fix
        // Tries to read data in BinStorage bytearrays as other pig types,
        // should return null if the conversion fails.
        // This test case does not use a practical example , it just tests
        // if the conversion happens when minimum conditions for conversion
        // such as expected number of bytes are met.
        String[] input = {
                    "asdf\t12\t1.1\t231\t234\t3024123\t3.2492",
                    "sa\t1231\t123.4\t12345678\t1234.567\t5081123453\t9.181817",
                    "asdff\t1232123\t1.45345\t123456789\t123456789.9\t1234567\t1.234567"
                    };

        Util.createInputFile(cluster, "table_bs_ac", input);

        // test with BinStorage
        pigServer.registerQuery("a = load 'table_bs_ac';");
        String output = "/pig/out/TestEvalPipeline2_BinStorageByteArrayCasts";
        pigServer.deleteFile(output);
        pigServer.store("a", output, BinStorage.class.getName());

        pigServer.registerQuery("b = load '" + output + "' using BinStorage('Utf8StorageConverter') "
                + "as (name: int, age: int, gpa: float, lage: long, dgpa: double, bi:biginteger, bd:bigdecimal);");

        Iterator<Tuple> it = pigServer.openIterator("b");

        Tuple tup=null;

        // I have separately verified only few of the successful conversions,
        // assuming the rest are correct.
        // It is primarily testing if null is being returned when conversions
        // are expected to fail

        //tuple 1
        tup = it.next();

        Assert.assertTrue((Integer)tup.get(0) == null);
        Assert.assertTrue((Integer)tup.get(1) == 12);
        Assert.assertTrue((Float)tup.get(2) == 1.1F);
        Assert.assertTrue((Long)tup.get(3) == 231L);
        Assert.assertTrue((Double)tup.get(4) == 234.0);
        Assert.assertEquals((BigInteger)tup.get(5), new BigInteger("3024123"));
        Assert.assertEquals((BigDecimal)tup.get(6), new BigDecimal("3.2492"));

        //tuple 2
        tup = it.next();
        Assert.assertTrue(tup.get(0) == null);
        Assert.assertTrue((Integer)tup.get(1) == 1231);
        Assert.assertTrue((Float)tup.get(2) == 123.4F);
        Assert.assertTrue((Long)tup.get(3) == 12345678L);
        Assert.assertTrue((Double)tup.get(4) == 1234.567);
        Assert.assertEquals((BigInteger)tup.get(5), new BigInteger("5081123453"));
        Assert.assertEquals((BigDecimal)tup.get(6), new BigDecimal("9.181817"));

        //tuple 3
        tup = it.next();
        Assert.assertTrue(tup.get(0) == null);
        Assert.assertTrue((Integer)tup.get(1) == 1232123);
        Assert.assertTrue((Float)tup.get(2) == 1.45345F);
        Assert.assertTrue((Long)tup.get(3) == 123456789L);
        Assert.assertTrue((Double)tup.get(4) == 1.234567899E8);
        Assert.assertEquals((BigInteger)tup.get(5), new BigInteger("1234567"));
        Assert.assertEquals((BigDecimal)tup.get(6), new BigDecimal("1.234567"));

        Util.deleteFile(cluster, "table");
    }
    @Test
    public void testBinStorageByteArrayCastsComplexBag() throws IOException {
        // Test for PIG-544 fix

        // Tries to read data in BinStorage bytearrays as other pig bags,
        // should return null if the conversion fails.

        String[] input = {
                "{(asdf)}",
                "{(2344)}",
                "{(2344}",
                "{(323423423423434)}",
                "{(323423423423434L)}",
                "{(asdff)}"
        };

        Util.createInputFile(cluster, "table_bs_ac_clx", input);

        // test with BinStorage
        pigServer.registerQuery("a = load 'table_bs_ac_clx' as (f1);");
        pigServer.registerQuery("b = foreach a generate (bag{tuple(int)})f1;");

        Iterator<Tuple> it = pigServer.openIterator("b");

        Tuple tup=null;

        //tuple 1
        tup = it.next();
        Assert.assertTrue(tup.get(0) != null);

        //tuple 2
        tup = it.next();
        Assert.assertTrue(tup.get(0) != null);

        //tuple 3 - malformed
        tup = it.next();
        Assert.assertTrue(tup.get(0) == null);

        //tuple 4 - integer exceeds size limit
        tup = it.next();
        Assert.assertTrue(tup.get(0) instanceof DataBag);
        DataBag db = (DataBag)tup.get(0);
        Assert.assertTrue(db.iterator().hasNext());
        Tuple innerTuple = db.iterator().next();
        Assert.assertTrue(innerTuple.get(0)==null);

        //tuple 5
        tup = it.next();
        Assert.assertTrue(tup.get(0) != null);

        //tuple 6
        tup = it.next();
        Assert.assertTrue(tup.get(0) != null);

        Util.deleteFile(cluster, "table_bs_ac_clx");
    }
    @Test
    public void testBinStorageByteArrayCastsComplexTuple() throws IOException {
        // Test for PIG-544 fix

        // Tries to read data in BinStorage bytearrays as other pig bags,
        // should return null if the conversion fails.

        String[] input = {
                "(123)",
                "((123)",
                "(123123123123)",
                "(asdf)"
        };

        Util.createInputFile(cluster, "table_bs_ac_clxt", input);

        // test with BinStorage
        pigServer.registerQuery("a = load 'table_bs_ac_clxt' as (t:tuple(t:tuple(i:int)));");
        Iterator<Tuple> it = pigServer.openIterator("a");

        Tuple tup=null;

        //tuple 1
        tup = it.next();
        Assert.assertTrue(tup.get(0) == null);

        //tuple 2 -malformed tuple
        tup = it.next();
        Assert.assertTrue(tup.get(0) == null);

        //tuple 3 - integer exceeds size limit
        tup = it.next();
        Assert.assertTrue(tup.get(0) == null);

        //tuple 4
        tup = it.next();
        Assert.assertTrue(tup.get(0) == null);

        Util.deleteFile(cluster, "table_bs_ac_clxt");
    }

    @Test
    public void testPigStorageWithCtrlChars() throws Exception {
        String[] inputData = { "hello\u0001world", "good\u0001morning", "nice\u0001day" };
        Util.createInputFile(cluster, "testPigStorageWithCtrlCharsInput.txt", inputData);
        String script = "a = load 'testPigStorageWithCtrlCharsInput.txt' using PigStorage('\u0001');" +
        		"b = foreach a generate $0, CONCAT($0, '\u0005'), $1; " +
        		"store b into 'testPigStorageWithCtrlCharsOutput.txt' using PigStorage('\u0001');" +
        		"c = load 'testPigStorageWithCtrlCharsOutput.txt' using PigStorage('\u0001') as (f1:chararray, f2:chararray, f3:chararray);";
        Util.registerMultiLineQuery(pigServer, script);
        Iterator<Tuple> it  = pigServer.openIterator("c");
        HashMap<String, Tuple> expectedResults = new HashMap<String, Tuple>();
        expectedResults.put("hello", (Tuple) Util.getPigConstant("('hello','hello\u0005','world')"));
        expectedResults.put("good", (Tuple) Util.getPigConstant("('good','good\u0005','morning')"));
        expectedResults.put("nice", (Tuple) Util.getPigConstant("('nice','nice\u0005','day')"));
        HashMap<String, Boolean> seen = new HashMap<String, Boolean>();
        int numRows = 0;
        while(it.hasNext()) {
            Tuple t = it.next();
            String firstCol = (String) t.get(0);
            Assert.assertFalse(seen.containsKey(firstCol));
            seen.put(firstCol, true);
            Assert.assertEquals(expectedResults.get(firstCol), t);
            numRows++;
        }
        Assert.assertEquals(3, numRows);
        Util.deleteFile(cluster, "testPigStorageWithCtrlCharsInput.txt");
    }

    @Test
    // Test case added for PIG-850
    public void testLimitedSortWithDump() throws Exception{
        int LOOP_COUNT = 40;
        File tmpFile = Util.createTempFileDelOnExit("test", "txt");
        PrintStream ps = new PrintStream(new FileOutputStream(tmpFile));
        Random r = new Random(1);
        int rand;
        for(int i = 0; i < LOOP_COUNT; i++) {
            rand = r.nextInt(100);
            ps.println(rand);
        }
        ps.close();

        pigServer.registerQuery("A = LOAD '"
                + Util.generateURI(tmpFile.toString(), pigServer
                        .getPigContext()) + "' AS (num:int);");
        pigServer.registerQuery("B = order A by num parallel 2;");
        pigServer.registerQuery("C = limit B 10;");
        Iterator<Tuple> result = pigServer.openIterator("C");

        int numIdentity = 0;
        while (result.hasNext())
        {
            result.next();
            ++numIdentity;
        }
        Assert.assertEquals(10, numIdentity);
    }

    @Test
    public void testLimitAfterSort() throws Exception{
        int LOOP_COUNT = 40;
        File tmpFile = Util.createTempFileDelOnExit("test", "txt");
        PrintStream ps = new PrintStream(new FileOutputStream(tmpFile));
        Random r = new Random(1);
        int rand;
        for(int i = 0; i < LOOP_COUNT; i++) {
            rand = r.nextInt(100);
            ps.println(rand);
        }
        ps.close();

        pigServer.registerQuery("A = LOAD '"
                + Util.generateURI(tmpFile.toString(), pigServer
                        .getPigContext()) + "' AS (num:int);");
        pigServer.registerQuery("B = order A by num parallel 2;");
        pigServer.registerQuery("C = limit B 10;");
        Iterator<Tuple> iter = pigServer.openIterator("C");
        if(!iter.hasNext()) Assert.fail("No output found");
        int numIdentity = 0;
        int oldNum = Integer.MIN_VALUE;
        int newNum;
        while(iter.hasNext()){
            Tuple t = iter.next();
            newNum = (Integer)t.get(0);
            Assert.assertTrue(newNum>=oldNum);
            oldNum = newNum;
            ++numIdentity;
        }
        Assert.assertEquals(10, numIdentity);
    }

    @Test
    public void testLimitAfterSortDesc() throws Exception{
        int LOOP_COUNT = 40;
        File tmpFile = Util.createTempFileDelOnExit("test", "txt");
        PrintStream ps = new PrintStream(new FileOutputStream(tmpFile));
        Random r = new Random(1);
        int rand;
        for(int i = 0; i < LOOP_COUNT; i++) {
            rand = r.nextInt(100);
            ps.println(rand);
        }
        ps.close();

        pigServer.registerQuery("A = LOAD '"
                + Util.generateURI(tmpFile.toString(), pigServer
                        .getPigContext()) + "' AS (num:int);");
        pigServer.registerQuery("B = order A by num desc parallel 2;");
        pigServer.registerQuery("C = limit B 10;");
        Iterator<Tuple> iter = pigServer.openIterator("C");
        if(!iter.hasNext()) Assert.fail("No output found");
        int numIdentity = 0;
        int oldNum = Integer.MAX_VALUE;
        int newNum;
        while(iter.hasNext()){
            Tuple t = iter.next();
            newNum = (Integer)t.get(0);
            Assert.assertTrue(newNum<=oldNum);
            oldNum = newNum;
            ++numIdentity;
        }
        Assert.assertEquals(10, numIdentity);
    }

    @Test
    // See PIG-894
    public void testEmptySort() throws Exception{
        File tmpFile = Util.createTempFileDelOnExit("test", "txt");
        pigServer.registerQuery("A = LOAD '"
                + Util.generateURI(tmpFile.toString(), pigServer
                        .getPigContext()) + "';");
        pigServer.registerQuery("B = order A by $0;");
        Iterator<Tuple> iter = pigServer.openIterator("B");

        Assert.assertTrue(iter.hasNext()==false);
    }

    // See PIG-761
    @Test
    public void testLimitPOPackageAnnotator() throws Exception{
        File tmpFile1 = Util.createTempFileDelOnExit("test1", "txt");
        PrintStream ps1 = new PrintStream(new FileOutputStream(tmpFile1));
        ps1.println("1\t2\t3");
        ps1.println("2\t5\t2");
        ps1.close();

        File tmpFile2 = Util.createTempFileDelOnExit("test2", "txt");
        PrintStream ps2 = new PrintStream(new FileOutputStream(tmpFile2));
        ps2.println("1\t1");
        ps2.println("2\t2");
        ps2.close();

        pigServer.registerQuery("A = LOAD '" + Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
        pigServer.registerQuery("B = LOAD '" + Util.generateURI(tmpFile2.toString(), pigServer.getPigContext()) + "' AS (b0, b1);");
        pigServer.registerQuery("C = LIMIT B 100;");
        pigServer.registerQuery("D = COGROUP C BY b0, A BY a0 PARALLEL 2;");
        Iterator<Tuple> iter = pigServer.openIterator("D");

        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(iter.hasNext());
            t = iter.next();

            Assert.assertTrue(t.toString().equals("(1,{(1,1)},{(1,2,3)})"));

            Assert.assertFalse(iter.hasNext());
        }
    }

    // See PIG-1195
    @Test
    public void testNestedDescSort() throws Exception{
        Util.createInputFile(cluster, "table_testNestedDescSort", new String[]{"3","4"});
        pigServer.registerQuery("A = LOAD 'table_testNestedDescSort' as (a0:int);");
        pigServer.registerQuery("B = group A ALL;");
        pigServer.registerQuery("C = foreach B { D = order A by a0 desc;generate D;};");
        Iterator<Tuple> iter = pigServer.openIterator("C");

        Assert.assertTrue(iter.hasNext());
        Tuple t = iter.next();

        Assert.assertTrue(t.toString().equals("({(4),(3)})"));
        Assert.assertFalse(iter.hasNext());

        Util.deleteFile(cluster, "table_testNestedDescSort");
    }

    // See PIG-972
    @Test
    public void testDescribeNestedAlias() throws Exception{
        String[] input = {
                "1\t3",
                "2\t4",
                "3\t5"
        };

        Util.createInputFile(cluster, "table_testDescribeNestedAlias", input);
        pigServer.registerQuery("A = LOAD 'table_testDescribeNestedAlias' as (a0, a1);");
        pigServer.registerQuery("P = GROUP A by a1;");
        // Test RelationalOperator
        pigServer.registerQuery("B = FOREACH P { D = ORDER A by $0; generate group, D.$0; };");

        // Test ExpressionOperator - negative test case
        pigServer.registerQuery("C = FOREACH A { D = a0/a1; E=a1/a0; generate E as newcol; };");
        Schema schema = pigServer.dumpSchemaNested("B", "D");
        Assert.assertTrue(schema.toString().equalsIgnoreCase("{a0: bytearray,a1: bytearray}"));
        try {
            schema = pigServer.dumpSchemaNested("C", "E");
        } catch (FrontendException e) {
            Assert.assertTrue(e.getErrorCode() == 1113);
        }
    }

    // See PIG-1484
    @Test
    public void testBinStorageCommaSeperatedPath() throws Exception{
        String[] input = {
                "1\t3",
                "2\t4",
                "3\t5"
        };

        Util.createInputFile(cluster, "table_simple1", input);
        pigServer.setBatchOn();
        pigServer.registerQuery("A = LOAD 'table_simple1' as (a0, a1);");
        pigServer.registerQuery("store A into 'table_simple1.bin' using BinStorage();");
        pigServer.registerQuery("store A into 'table_simple2.bin' using BinStorage();");

        pigServer.executeBatch();

        pigServer.registerQuery("A = LOAD 'table_simple1.bin,table_simple2.bin' using BinStorage();");
        Iterator<Tuple> iter = pigServer.openIterator("A");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().equals("(1,3)"));

        t = iter.next();
        Assert.assertTrue(t.toString().equals("(2,4)"));

        t = iter.next();
        Assert.assertTrue(t.toString().equals("(3,5)"));

        t = iter.next();
        Assert.assertTrue(t.toString().equals("(1,3)"));

        t = iter.next();
        Assert.assertTrue(t.toString().equals("(2,4)"));

        t = iter.next();
        Assert.assertTrue(t.toString().equals("(3,5)"));

        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-1543
    @Test
    public void testEmptyBagIterator() throws Exception{
        String[] input1 = {
                "1",
                "1",
                "1"
        };

        String[] input2 = {
                "2",
                "2"
        };

        Util.createInputFile(cluster, "input1", input1);
        Util.createInputFile(cluster, "input2", input2);
        pigServer.registerQuery("A = load 'input1' as (a1:int);");
        pigServer.registerQuery("B = load 'input2' as (b1:int);");
        pigServer.registerQuery("C = COGROUP A by a1, B by b1;");
        pigServer.registerQuery("C1 = foreach C { Alim = limit A 1; Blim = limit B 1; generate Alim, Blim; };");
        pigServer.registerQuery("D1 = FOREACH C1 generate Alim,Blim, (IsEmpty(Alim)? 0:1), (IsEmpty(Blim)? 0:1), COUNT(Alim), COUNT(Blim);");

        Iterator<Tuple> iter = pigServer.openIterator("D1");

        List<Tuple> expectedResults = Util.getTuplesFromConstantTupleStrings(
                new String[] {
                        "({(1)},{},1,0,1L,0L)",
                        "({},{(2)},0,1,0L,1L)" });
        Util.checkQueryOutputsAfterSort(iter, expectedResults);
    }

    // See PIG-1669
    @Test
    public void testPushUpFilterScalar() throws Exception{
        String[] input1 = {
                "jason\t14\t4.7",
                "jack\t18\t4.6"
        };

        String[] input2 = {
                "jason\t14",
                "jack\t18"
        };

        Util.createInputFile(cluster, "table_PushUpFilterScalar1", input1);
        Util.createInputFile(cluster, "table_PushUpFilterScalar2", input2);
        pigServer.registerQuery("A = load 'table_PushUpFilterScalar1' as (name, age, gpa);");
        pigServer.registerQuery("B = load 'table_PushUpFilterScalar2' as (name, age);");
        pigServer.registerQuery("C = filter A by age < 20;");
        pigServer.registerQuery("D = filter B by age < 20;");
        pigServer.registerQuery("simple_scalar = limit D 1;");
        pigServer.registerQuery("E = join C by name, D by name;");
        pigServer.registerQuery("F = filter E by C::age==(int)simple_scalar.age;");

        Iterator<Tuple> iter = pigServer.openIterator("F");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().equals("(jason,14,4.7,jason,14)"));

        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-1683
    @Test
    public void testDuplicateReferenceInnerPlan() throws Exception{
        String[] input1 = {
                "1\t1\t1",
        };

        String[] input2 = {
                "1\t1",
                "2\t2"
        };

        Util.createInputFile(cluster, "table_testDuplicateReferenceInnerPlan1", input1);
        Util.createInputFile(cluster, "table_testDuplicateReferenceInnerPlan2", input2);
        pigServer.registerQuery("a = load 'table_testDuplicateReferenceInnerPlan1' as (a0, a1, a2);");
        pigServer.registerQuery("b = load 'table_testDuplicateReferenceInnerPlan2' as (b0, b1);");
        pigServer.registerQuery("c = join a by a0, b by b0;");
        pigServer.registerQuery("d = foreach c {d0 = a::a1;d1 = a::a2;generate ((d0 is not null)? d0 : d1);};");

        Iterator<Tuple> iter = pigServer.openIterator("d");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().equals("(1)"));

        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-1719
    @Test
    public void testBinCondSchema() throws IOException {
        String[] inputData = new String[] {"hello world\t2"};
        Util.createInputFile(cluster, "table_testSchemaSerialization.txt", inputData);
        pigServer.registerQuery("a = load 'table_testSchemaSerialization.txt' as (a0:chararray, a1:int);");
        pigServer.registerQuery("b = foreach a generate FLATTEN((a1<=1?{('null')}:TOKENIZE(a0)));");
        pigServer.registerQuery("c = foreach b generate UPPER($0);");

        Iterator<Tuple> it = pigServer.openIterator("c");
        Tuple t = it.next();
        Assert.assertTrue(t.get(0).equals("HELLO"));
        t = it.next();
        Assert.assertTrue(t.get(0).equals("WORLD"));
    }

    // See PIG-1721
    @Test
    public void testDuplicateInnerAlias() throws Exception{
        String[] input1 = {
                "1\t[key1#5]", "1\t[key2#5]", "2\t[key1#3]"
        };

        Util.createInputFile(cluster, "table_testDuplicateInnerAlias", input1);
        pigServer.registerQuery("a = load 'table_testDuplicateInnerAlias' as (a0:int, a1:map[]);");
        pigServer.registerQuery("b = filter a by a0==1;");
        pigServer.registerQuery("c = foreach b { b0 = a1#'key1'; generate ((b0 is null or b0 == '')?1:0);};");

        Iterator<Tuple> iter = pigServer.openIterator("c");

        Tuple t = iter.next();
        Assert.assertTrue((Integer)t.get(0)==0);

        t = iter.next();
        Assert.assertTrue((Integer)t.get(0)==1);

        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-3379
    @Test
    public void testNestedOperatorReuse() throws Exception{
        String[] input1 = {
        		"60000\tdv1\txuaHeartBeat",
        		"70000\tdv2\txuaHeartBeat",
        		"80000\tdv1\txuaPowerOff",
        		"90000\tdv1\txuaPowerOn",
        		"110000\tdv2\txuaHeartBeat",
        		"120000\tdv2\txuaPowerOff",
        		"140000\tdv2\txuaPowerOn",
        		"150000\tdv1\txuaHeartBeat",
        		"160000\tdv2\txuaHeartBeat",
        		"250000\tdv1\txuaHeartBeat",
        		"310000\tdv2\txuaPowerOff",
        		"360000\tdv1\txuaPowerOn",
        		"420000\tdv3\txuaHeartBeat",
        		"450000\tdv3\txuaHeartBeat",
        		"540000\tdv4\txuaPowerOn",
        		"550000\tdv3\txuaHeartBeat",
        		"560000\tdv5\txuaHeartBeat" };
        Util.createInputFile( cluster, "table_testNestedOperatorReuse", input1 );
        String query = "Events = LOAD 'table_testNestedOperatorReuse' AS (eventTime:long, deviceId:chararray, eventName:chararray);" +
        		"Events = FOREACH Events GENERATE eventTime, deviceId, eventName;" +
        		"EventsPerMinute = GROUP Events BY (eventTime / 60000);" +
        		"EventsPerMinute = FOREACH EventsPerMinute {" +
        		"  DistinctDevices = DISTINCT Events.deviceId;" +
        		"  nbDevices = SIZE(DistinctDevices);" +
        		"  DistinctDevices = FILTER Events BY eventName == 'xuaHeartBeat';" +
        		"  nbDevicesWatching = SIZE(DistinctDevices);" +
        		"  GENERATE $0*60000 as timeStamp, nbDevices as nbDevices, nbDevicesWatching as nbDevicesWatching;" +
        		"}" +
        		"EventsPerMinute = FILTER EventsPerMinute BY timeStamp >= 0  AND timeStamp < 300000;";

        pigServer.registerQuery(query);
        Iterator<Tuple> iter = pigServer.openIterator("EventsPerMinute");

        List<Tuple> expectedResults = Util.getTuplesFromConstantTupleStrings(
            new String[]{"(60000L,2L,3L)", "(120000L,2L,2L)", "(240000L,1L,1L)"});

        Util.checkQueryOutputs(iter, expectedResults, Util.isSparkExecType(cluster.getExecType()));
    }

    // See PIG-1729
    @Test
    public void testDereferenceInnerPlan() throws Exception{
        String[] input1 = {
                "1\t2\t3"
        };

        String[] input2 = {
                "1\t1"
        };

        Util.createInputFile(cluster, "table_testDereferenceInnerPlan1", input1);
        Util.createInputFile(cluster, "table_testDereferenceInnerPlan2", input2);
        pigServer.registerQuery("a = load 'table_testDereferenceInnerPlan1' as (a0:int, a1:int, a2:int);");
        pigServer.registerQuery("b = load 'table_testDereferenceInnerPlan2' as (b0:int, b1:int);");
        pigServer.registerQuery("c = cogroup a by a0, b by b0;");
        pigServer.registerQuery("d = foreach c generate ((COUNT(a)==0L)?null : a.a0) as d0;");
        pigServer.registerQuery("e = foreach d generate flatten(d0);");
        pigServer.registerQuery("f = group e all;");

        Iterator<Tuple> iter = pigServer.openIterator("f");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().equals("(all,{(1)})"));

        Assert.assertFalse(iter.hasNext());
    }

    @Test
    // See PIG-1725
    public void testLOGenerateSchema() throws Exception{
        String[] input1 = {
                "1\t2\t{(1)}",
        };

        Util.createInputFile(cluster, "table_testLOGenerateSchema", input1);
        pigServer.registerQuery("a = load 'table_testLOGenerateSchema' as (a0:int, a1, a2:bag{});");
        pigServer.registerQuery("b = foreach a generate a0 as b0, a1 as b1, flatten(a2) as b2:int;");
        pigServer.registerQuery("c = filter b by b0==1;");
        pigServer.registerQuery("d = foreach c generate b0+1, b2;");

        Iterator<Tuple> iter = pigServer.openIterator("d");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().equals("(2,1)"));

        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-1737
    @Test
    public void testMergeSchemaErrorMessage() throws IOException {
        try {
            pigServer.registerQuery("a = load '1.txt' as (a0, a1, a2);");
            pigServer.registerQuery("b = group a by (a0, a1);");
            pigServer.registerQuery("c = foreach b generate flatten(group) as c0;");
            pigServer.openIterator("c");
        } catch (Exception e) {
            PigException pe = LogUtils.getPigException(e);
            Util.checkStrContainsSubStr(pe.getMessage(), "Incompatible schema");
            return;
        }
        Assert.fail();
    }

    // See PIG-1732
    @Test
    public void testForEachDupColumn() throws Exception{
        String[] input1 = {
                "1\t2",
        };

        String[] input2 = {
                "1\t1\t3",
                "2\t4\t2"
        };

        Util.createInputFile(cluster, "table_testForEachDupColumn1", input1);
        Util.createInputFile(cluster, "table_testForEachDupColumn2", input2);
        pigServer.registerQuery("a = load 'table_testForEachDupColumn1' as (a0, a1:int);");
        pigServer.registerQuery("b = load 'table_testForEachDupColumn2' as (b0, b1:int, b2);");
        pigServer.registerQuery("c = foreach a generate a0, a1, a1 as a2;");
        pigServer.registerQuery("d = union b, c;");
        pigServer.registerQuery("e = foreach d generate $1;");

        Iterator<Tuple> iter = pigServer.openIterator("e");

        Map<Object, Object> expected = new HashMap<Object, Object>(3);
        expected.put(1, null);
        expected.put(2, null);
        expected.put(4, null);

        Tuple t = iter.next();
        Assert.assertTrue(t.size()==1);
        Assert.assertTrue(expected.containsKey(t.get(0)));

        t = iter.next();
        Assert.assertTrue(t.size()==1);
        Assert.assertTrue(expected.containsKey(t.get(0)));

        t = iter.next();
        Assert.assertTrue(t.size()==1);
        Assert.assertTrue(expected.containsKey(t.get(0)));

        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-1745
    @Test
    public void testBinStorageByteCast() throws Exception{
        String[] input1 = {
                "1\t2\t3",
        };

        Util.createInputFile(cluster, "table_testBinStorageByteCast", input1);
        pigServer.registerQuery("a = load 'table_testBinStorageByteCast' as (a0, a1, a2);");
        pigServer.store("a", "table_testBinStorageByteCast.temp", BinStorage.class.getName());

        pigServer.registerQuery("a = load 'table_testBinStorageByteCast.temp' using BinStorage() as (a0, a1, a2);");
        pigServer.registerQuery("b = foreach a generate (long)a0;");

        try {
            pigServer.openIterator("b");
        } catch (Exception e) {
            PigException pe = LogUtils.getPigException(e);
            //This changes in hadoop 23, we get error code 2997
            //Assert.assertTrue(pe.getErrorCode()==1118);
            return;
        }

        Assert.fail();
    }

    // See PIG-1761
    @Test
    public void testBagDereferenceInMiddle1() throws Exception{
        String[] input1 = {
                "foo@apache#44",
        };

        Util.createInputFile(cluster, "table_testBagDereferenceInMiddle1", input1);
        pigServer.registerQuery("a = load 'table_testBagDereferenceInMiddle1' as (a0:chararray);");
        pigServer.registerQuery("b = foreach a generate UPPER(REGEX_EXTRACT_ALL(a0, '.*@(.*)#.*').$0);");

        Iterator<Tuple> iter = pigServer.openIterator("b");
        Tuple t = iter.next();
        Assert.assertTrue(t.size()==1);
        Assert.assertTrue(t.get(0).equals("APACHE"));
    }

    // See PIG-1843
    @Test
    public void testBagDereferenceInMiddle2() throws Exception{
        String[] input1 = {
                "foo apache",
        };

        Util.createInputFile(cluster, "table_testBagDereferenceInMiddle2", input1);
        pigServer.registerQuery("a = load 'table_testBagDereferenceInMiddle2' as (a0:chararray);");
        pigServer.registerQuery("b = foreach a generate " + MapGenerate.class.getName() + " (STRSPLIT(a0).$0);");

        Iterator<Tuple> iter = pigServer.openIterator("b");
        Tuple t = iter.next();
        Assert.assertTrue(t.size()==1);
        Assert.assertTrue(t.toString().equals("([key#1])"));
    }

    // See PIG-1766
    @Test
    public void testForEachSameOriginColumn1() throws Exception {
        String[] input1 = {
                "1\t2",
                "1\t3",
                "2\t4",
                "2\t5",
        };

        String[] input2 = {
                "1\tone",
                "2\ttwo",
        };

        Util.createInputFile(cluster, "table_testForEachSameOriginColumn1_1", input1);
        Util.createInputFile(cluster, "table_testForEachSameOriginColumn1_2", input2);
        pigServer.registerQuery("A = load 'table_testForEachSameOriginColumn1_1' AS (a0:int, a1:int);");
        pigServer.registerQuery("B = load 'table_testForEachSameOriginColumn1_2' AS (b0:int, b1:chararray);");
        pigServer.registerQuery("C = join A by a0, B by b0;");
        pigServer.registerQuery("D = foreach B generate b0 as d0, b1 as d1;");
        pigServer.registerQuery("E = join C by a1, D by d0;");
        pigServer.registerQuery("F = foreach E generate b1, d1;");

        Iterator<Tuple> iter = pigServer.openIterator("F");
        Tuple t = iter.next();
        Assert.assertTrue(t.size()==2);
        Assert.assertTrue(t.get(0).equals("one"));
        Assert.assertTrue(t.get(1).equals("two"));
    }

    // See PIG-1771
    @Test
    public void testLoadWithDifferentSchema() throws Exception{
        String[] input1 = {
                "hello\thello\t(hello)\t[key#value]",
        };

        Util.createInputFile(cluster, "table_testLoadWithDifferentSchema1", input1);
        pigServer.registerQuery("a = load 'table_testLoadWithDifferentSchema1' as (a0:chararray, a1:chararray, a2, a3:map[]);");
        pigServer.store("a", "table_testLoadWithDifferentSchema1.bin", "org.apache.pig.builtin.BinStorage");

        pigServer.registerQuery("b = load 'table_testLoadWithDifferentSchema1.bin' USING BinStorage('Utf8StorageConverter') AS (b0:chararray, b1:chararray, b2:tuple(), b3:map[]);");
        Iterator<Tuple> iter = pigServer.openIterator("b");

        Tuple t = iter.next();
        Assert.assertTrue(t.size()==4);
        Assert.assertTrue(t.toString().equals("(hello,hello,(hello),[key#value])"));
    }

    static public class MapGenerate extends EvalFunc<Map<String, Integer>> {
        @Override
        public Map<String, Integer> exec(Tuple input) throws IOException {
            Map<String, Integer> m = new HashMap<String, Integer>();
            m.put("key", new Integer(input.size()));
            return m;
        }

        @Override
        public Schema outputSchema(Schema input) {
            return new Schema(new Schema.FieldSchema(getSchemaName("parselong", input), DataType.MAP));
        }
    }

    // See PIG-1277
    @Test
    public void testWrappingUnknownKey1() throws Exception{
        String[] input1 = {
                "1",
        };

        Util.createInputFile(cluster, "table_testWrappingUnknownKey1", input1);

        pigServer.registerQuery("a = load 'table_testWrappingUnknownKey1' as (a0);");
        pigServer.registerQuery("b = foreach a generate a0, "+ MapGenerate.class.getName() + "(*) as m:map[];");
        pigServer.registerQuery("c = foreach b generate a0, m#'key' as key;");
        pigServer.registerQuery("d = group c by key;");

        Iterator<Tuple> iter = pigServer.openIterator("d");

        Tuple t = iter.next();
        Assert.assertTrue(t.size()==2);
        Assert.assertTrue(t.toString().equals("(1,{(1,1)})"));
        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-999
    @Test
    public void testWrappingUnknownKey2() throws Exception{
        String[] input1 = {
                "1",
        };

        Util.createInputFile(cluster, "table_testWrappingUnknownKey2", input1);

        pigServer.registerQuery("a = load 'table_testWrappingUnknownKey2' as (a0);");
        pigServer.registerQuery("b = foreach a generate a0, "+ MapGenerate.class.getName() + "(*) as m:map[];");
        pigServer.registerQuery("c = foreach b generate a0, m#'key' as key;");
        pigServer.registerQuery("d = order c by key;");

        Iterator<Tuple> iter = pigServer.openIterator("d");

        Tuple t = iter.next();
        Assert.assertTrue(t.size()==2);
        Assert.assertTrue(t.toString().equals("(1,1)"));
        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-1065
    @Test
    public void testWrappingUnknownKey3() throws Exception{
        String[] input1 = {
                "1\t2",
                "2\t3"
        };

        String[] input2 = {
                "1",
        };

        Util.createInputFile(cluster, "table_testWrappingUnknownKey3_1", input1);
        Util.createInputFile(cluster, "table_testWrappingUnknownKey3_2", input2);

        pigServer.registerQuery("a = load 'table_testWrappingUnknownKey3_1' as (a0:chararray, a1:chararray);");
        pigServer.registerQuery("b = load 'table_testWrappingUnknownKey3_2' as (b0:chararray);");
        pigServer.registerQuery("c = union a, b;");
        pigServer.registerQuery("d = order c by $0;");

        Collection<String> results = new HashSet<String>();
        results.add("(1,2)");
        results.add("(1)");
        results.add("(2,3)");

        Iterator<Tuple> iter = pigServer.openIterator("d");

        Tuple t = iter.next();
        Assert.assertTrue(results.contains(t.toString()));
        t = iter.next();
        Assert.assertTrue(results.contains(t.toString()));
        t = iter.next();
        Assert.assertTrue(results.contains(t.toString()));
        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-1787
    @Test
    public void testOrderByLimitJoin() throws Exception{
        String[] input1 = {
                "1\t1",
                "1\t2"
        };

        Util.createInputFile(cluster, "table_testOrderByLimitJoin", input1);

        pigServer.registerQuery("a = load 'table_testOrderByLimitJoin' as (a0, a1);");
        pigServer.registerQuery("b = group a by a0;");
        pigServer.registerQuery("c = foreach b generate group as c0, COUNT(a) as c1;");
        pigServer.registerQuery("d = order c by c1 parallel 2;");
        pigServer.registerQuery("e = limit d 10;");
        pigServer.registerQuery("f = join e by c0, a by a0;");

        Iterator<Tuple> iter = pigServer.openIterator("f");

        String[] expected = new String[] {"(1,2,1,1)", "(1,2,1,2)"};

        Util.checkQueryOutputsAfterSortRecursive(iter, expected, org.apache.pig.newplan.logical.Util.translateSchema(pigServer.dumpSchema("f")));

    }

    // See PIG-1785
    @Test
    public void testForEachSameOriginColumn2() throws Exception{
        String[] input1 = {
                "{(1,2),(2,3)}",
        };

        Util.createInputFile(cluster, "table_testForEachSameOriginColumn2", input1);

        pigServer.registerQuery("a = load 'table_testForEachSameOriginColumn2' as (a0:bag{t:tuple(i0:int, i1:int)});");
        pigServer.registerQuery("b = foreach a generate flatten(a0) as (b0, b1), flatten(a0) as (b2, b3);");
        pigServer.registerQuery("c = filter b by b0>b2;");

        Iterator<Tuple> iter = pigServer.openIterator("c");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().contains("(2,3,1,2)"));
        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-1785
    @Test
    public void testForEachSameOriginColumn3() throws Exception{
        String[] input1 = {
                "1\t1\t2",
                "1\t2\t3",
        };

        Util.createInputFile(cluster, "table_testForEachSameOriginColumn3", input1);

        pigServer.registerQuery("a = load 'table_testForEachSameOriginColumn3' as (a0:int, a1:int, a2:int);");
        pigServer.registerQuery("b = group a by a0;");
        pigServer.registerQuery("c = foreach b generate flatten(a.(a1,a2)) as (b0, b1), flatten(a.(a1,a2)) as (b2, b3);");
        pigServer.registerQuery("d = filter c by b0>b2;");

        Iterator<Tuple> iter = pigServer.openIterator("d");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().contains("(2,3,1,2)"));
        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-1785
    @Test
    public void testAddingTwoBag() {
        try {
            pigServer.registerQuery("a = load '1.txt' as (name:chararray, age:int, gpa:double);");
            pigServer.registerQuery("b = group a by name;");
            pigServer.registerQuery("c = foreach b generate group, SUM(a.age*a.gpa);");
            pigServer.openIterator("c");
        } catch (Exception e) {
            PigException pe = LogUtils.getPigException(e);
            Assert.assertTrue(pe.getErrorCode()==1039);
            Assert.assertTrue(pe.getMessage().contains("incompatible types"));
            return;
        }
        Assert.fail();
    }

    public static class BagGenerateNoSchema extends EvalFunc<DataBag> {
        @Override
        public DataBag exec(Tuple input) throws IOException {
            DataBag bg = DefaultBagFactory.getInstance().newDefaultBag();
            bg.add(input);
            return bg;
        }
    }

    // See PIG-1813
    @Test
    public void testUDFNoSchemaPropagate1() throws Exception{
        String[] input1 = {
                "[key#1,key2#2]",
                "[key#2,key2#3]",
        };

        Util.createInputFile(cluster, "table_testUDFNoSchemaPropagate1", input1);

        pigServer.registerQuery("a = load 'table_testUDFNoSchemaPropagate1' as (a0:map[]);");
        pigServer.registerQuery("b = foreach a generate " + BagGenerateNoSchema.class.getName() + "(*) as b0;");
        pigServer.registerQuery("c = foreach b generate flatten(IdentityColumn(b0));");
        pigServer.registerQuery("d = foreach c generate $0#'key';");

        Iterator<Tuple> iter = pigServer.openIterator("d");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().contains("(1)"));
        t = iter.next();
        Assert.assertTrue(t.toString().contains("(2)"));
        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-1813
    @Test
    public void testUDFNoSchemaPropagate2() throws Exception{
        String[] input1 = {
                "[key#1,key2#2]",
                "[key#2,key2#3]",
        };

        Util.createInputFile(cluster, "table_testUDFNoSchemaPropagate2", input1);

        pigServer.registerQuery("a = load 'table_testUDFNoSchemaPropagate2' as (a0:map[]);");
        pigServer.registerQuery("b = foreach a generate flatten(" + BagGenerateNoSchema.class.getName() + "(*)) as b0;");
        pigServer.registerQuery("c = foreach b generate IdentityColumn(b0);");
        pigServer.registerQuery("d = foreach c generate $0#'key';");

        Iterator<Tuple> iter = pigServer.openIterator("d");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().contains("(1)"));
        t = iter.next();
        Assert.assertTrue(t.toString().contains("(2)"));
        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-1812
    @Test
    public void testLocalRearrangeInReducer() throws Exception{
        String[] input1 = {
                "1\t1",
                "1\t1",
                "1\t2",
        };

        String[] input2 = {
                "1\t1",
        };

        Util.createInputFile(cluster, "table_testLocalRearrangeInReducer1", input1);
        Util.createInputFile(cluster, "table_testLocalRearrangeInReducer2", input2);

        pigServer.registerQuery("a = load 'table_testLocalRearrangeInReducer1' as (a0, a1);");
        pigServer.registerQuery("b = distinct a;");
        pigServer.registerQuery("c = load 'table_testLocalRearrangeInReducer2' as (c0, c1);");
        pigServer.registerQuery("d = cogroup b by a0, c by c0;");
        pigServer.registerQuery("e = foreach d { e1 = order c by c1; generate e1;};");

        Iterator<Tuple> iter = pigServer.openIterator("e");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().contains("({(1,1)})"));
        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-1850
    @Test
    public void testProjectNullSchema() throws Exception{
        String[] input = {
                "0\t1",
        };

        Util.createInputFile(cluster, "table_testProjectNullSchema", input);

        pigServer.registerQuery("a = load 'table_testProjectNullSchema';");
        pigServer.registerQuery("b = foreach a generate ASIN($0), $1;");
        pigServer.registerQuery("c = order b by $0;");

        Iterator<Tuple> iter = pigServer.openIterator("c");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().contains("(0.0,1)"));
        Assert.assertFalse(iter.hasNext());
    }


    // See PIG-1188
    @Test
    public void testSchemaDataNotMatch() throws Exception{
        String[] input = {
                "0\t1\t2",
                "3\t4",
                "5"
        };

        Util.createInputFile(cluster, "table_testSchemaDataNotMatch", input);

        pigServer.registerQuery("a = load 'table_testSchemaDataNotMatch' as (a0, a1);");

        Iterator<Tuple> iter = pigServer.openIterator("a");

        Tuple t = iter.next();
        Assert.assertTrue(t.size()==2);
        Assert.assertTrue(t.get(0).toString().equals("0"));
        Assert.assertTrue(t.get(1).toString().equals("1"));

        t = iter.next();
        Assert.assertTrue(t.size()==2);
        Assert.assertTrue(t.get(0).toString().equals("3"));
        Assert.assertTrue(t.get(1).toString().equals("4"));

        t = iter.next();
        Assert.assertTrue(t.size()==2);
        Assert.assertTrue(t.get(0).toString().equals("5"));
        Assert.assertTrue(t.get(1)==null);

        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-1912
    @Test
    public void testDuplicateLoadFuncSignature() throws Exception{
        String[] input = {
                "0\t1\ta",
        };

        Util.createInputFile(cluster, "table_testDuplicateLoadFuncSignature", input);
        pigServer.setBatchOn();
        pigServer.registerQuery("a = load 'table_testDuplicateLoadFuncSignature' as (a0, a1, a2);");
        pigServer.registerQuery("b = foreach a generate a0, a1;");
        pigServer.registerQuery("a = load 'table_testDuplicateLoadFuncSignature' as (a0, a1, a2);");
        pigServer.registerQuery("c = foreach a generate a0, a2;");
        pigServer.registerQuery("store b into 'testDuplicateLoadFuncSignatureOutput1';");
        pigServer.registerQuery("store c into 'testDuplicateLoadFuncSignatureOutput2';");

        pigServer.executeBatch();

        pigServer.registerQuery("a = load 'testDuplicateLoadFuncSignatureOutput1';");
        Iterator<Tuple> iter = pigServer.openIterator("a");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().equals("(0,1)"));
        Assert.assertFalse(iter.hasNext());

        pigServer.registerQuery("a = load 'testDuplicateLoadFuncSignatureOutput2';");
        iter = pigServer.openIterator("a");

        t = iter.next();
        Assert.assertTrue(t.toString().equals("(0,a)"));
        Assert.assertFalse(iter.hasNext());

    }

    // See PIG-1927
    @Test
    public void testDereferencePartialAlias() throws Exception{

        pigServer.registerQuery("a = load '1.txt' as (a0:int, a1);");
        pigServer.registerQuery("b = group a by a0;");
        pigServer.registerQuery("c = foreach b generate flatten(a);");
        pigServer.registerQuery("d = cogroup c by (a0);");
        pigServer.registerQuery("e = foreach d generate c.a0 as e0;");
        pigServer.registerQuery("f = foreach e generate e0;");

        // Shall not throw exception
        pigServer.explain("f", System.out);
    }

    // See PIG-1866
    @Test
    public void testProjBagInTuple() throws Exception{
        String[] input = {
                "(1,{(one),(two)})",
        };

        Util.createInputFile(cluster, "table_testProjBagInTuple", input);

        pigServer.registerQuery("a = load 'table_testProjBagInTuple' as (t : tuple(i: int, b1: bag { b_tuple : tuple ( b_str: chararray) }));");
        pigServer.registerQuery("b = foreach a generate t.b1;");

        Iterator<Tuple> iter = pigServer.openIterator("b");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().equals("({(one),(two)})"));
        Assert.assertFalse(iter.hasNext());
    }

    //See PIG-1974
    @Test
    public void testCastMap() throws Exception{
        String[] input = {
                "([key#1])",
                "([key#2])",
        };

        Util.createInputFile(cluster, "table_testCastMap", input);

        pigServer.registerQuery("a = load 'table_testCastMap' as (m:map[]);");
        pigServer.registerQuery("b = foreach a generate (map[int])m;");
        pigServer.registerQuery("c = foreach b generate m#'key' + 1;");

        Iterator<Tuple> iter = pigServer.openIterator("c");

        Tuple t = iter.next();
        Assert.assertEquals(t.get(0), 2);
        t = iter.next();
        Assert.assertEquals(t.get(0), 3);
        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-1979
    @Test
    public void testDereferenceUidBug() throws Exception{
        String[] input1 = {
                "0\t0\t{(1,2)}\t1",
        };
        String[] input2 = {
                "0\t0",
        };

        Util.createInputFile(cluster, "table_testDereferenceUidBug1", input1);
        Util.createInputFile(cluster, "table_testDereferenceUidBug2", input2);
        pigServer.registerQuery("a = load 'table_testDereferenceUidBug1' as (a0:int, a1:int, a2:{t:(i0:int, i1:int)}, a3:int);");
        pigServer.registerQuery("b = foreach a generate a0, a1, a0+a1 as sum, a2.i0 as a2, a3;");
        pigServer.registerQuery("c = filter b by sum==0;");
        pigServer.registerQuery("d = load 'table_testDereferenceUidBug2' as (d0:int, d1:int);");
        pigServer.registerQuery("e = join c by a0, d by d0;");
        pigServer.registerQuery("f = foreach e generate c::a2;");

        Iterator<Tuple> iter = pigServer.openIterator("f");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().equals("({(1)})"));
        Assert.assertFalse(iter.hasNext());
    }

    static public class UDFWithNonStandardType extends EvalFunc<Tuple>{
        @Override
        public Tuple exec(Tuple input) throws IOException {
            Tuple t = TupleFactory.getInstance().newTuple();
            t.append(new ArrayList<Integer>());
            return t;
        }
    }

    // See PIG-1826
    @Test
    public void testNonStandardData() throws Exception{
        String[] input1 = {
                "0",
        };

        Util.createInputFile(cluster, "table_testNonStandardData", input1);
        pigServer.registerQuery("a = load 'table_testNonStandardData' as (a0);");
        pigServer.registerQuery("b = foreach a generate " + UDFWithNonStandardType.class.getName() + "(a0);");

        try {
            pigServer.openIterator("b");
            Assert.fail();
        } catch (Exception e) {
            // Tez does not construct exceptions from stacktrace as it will have multiple ones.
            // So e.getCause().getCause() will be null
            Throwable cause = e.getCause().getCause() ==  null ? e.getCause() : e.getCause().getCause();
            String message = cause.getMessage();
            Assert.assertTrue(message.contains(ArrayList.class.getName()));
        }
    }

    // See PIG-1826
    @Test
    public void testNonStandardDataWithoutFetch() throws Exception{
        Properties props = pigServer.getPigContext().getProperties();
        props.setProperty(PigConfiguration.PIG_OPT_FETCH, "false");
        String[] input1 = {
                "0",
        };
        try {
            Util.createInputFile(cluster, "table_testNonStandardDataWithoutFetch", input1);
            pigServer.registerQuery("a = load 'table_testNonStandardDataWithoutFetch' as (a0);");
            pigServer.registerQuery("b = foreach a generate " + UDFWithNonStandardType.class.getName() + "(a0);");

            try {
                pigServer.openIterator("b");
                Assert.fail();
            } catch (Exception e) {
                Assert.assertTrue(e.getMessage().contains(ArrayList.class.getName()));
            }
        }
        finally {
            props.setProperty(PigConfiguration.PIG_OPT_FETCH, "true");
        }
    }

    // See PIG-2078
    @Test
    public void testProjectNullBag() throws Exception{
        String[] input1 = {
                "{(1)}\t2",
                "\t3"
        };

        HashSet<String> optimizerRules = new HashSet<String>();
        optimizerRules.add("MergeForEach");
        pigServer.getPigContext().getProperties().setProperty(
                PigImplConstants.PIG_OPTIMIZER_RULES_KEY,
                ObjectSerializer.serialize(optimizerRules));

        Util.createInputFile(cluster, "table_testProjectNullBag", input1);
        pigServer.registerQuery("a = load 'table_testProjectNullBag' as (a0:bag{}, a1:int);");
        pigServer.registerQuery("b = foreach a generate a0;");

        Iterator<Tuple> iter = pigServer.openIterator("b");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().equals("({(1)})"));

        t = iter.next();
        Assert.assertTrue(t.toString().equals("()"));

        Assert.assertFalse(iter.hasNext());

        pigServer.getPigContext().getProperties().remove(PigImplConstants.PIG_OPTIMIZER_RULES_KEY);
    }

    // See PIG-2159
    @Test
    public void testUnionOnSchemaUidGeneration() throws Exception{
        String[] input1 = {
        		"100,101,102,103,104,105",
        		"110,111,112,113,114,115"
        };

        String[] input2 = {
        		"200,201,202,203,204,205",
        		"210,211,212,213,214,215"
        };

        String[] input0 = {
        		"200,201,202,203,204,205",
        		"210,211,212,213,214,215"
        };

        Util.createInputFile(cluster, "table_testUnionOnSchemaUidGeneration1", input1);
        Util.createInputFile(cluster, "table_testUnionOnSchemaUidGeneration2", input2);
        Util.createInputFile(cluster, "table_testUnionOnSchemaUidGeneration0", input0);

        pigServer.registerQuery("A = load 'table_testUnionOnSchemaUidGeneration1' using PigStorage(',')  as (f1:int,f2:int,f3:int,f4:long,f5:double);");
        pigServer.registerQuery("B = load 'table_testUnionOnSchemaUidGeneration2' using PigStorage(',')  as (f1:int,f2:int,f3:int,f4:long,f5:double);");
        pigServer.registerQuery("C = load 'table_testUnionOnSchemaUidGeneration0' using PigStorage(',')  as (f1:int,f2:int,f3:int);");
        pigServer.registerQuery("U = UNION ONSCHEMA A,B;");
        pigServer.registerQuery("J = join C by (f1,f2,f3) LEFT OUTER, U by (f1,f2,f3);");
        pigServer.registerQuery("Porj = foreach J generate C::f1 as f1 ,C::f2 as f2,C::f3 as f3,U::f4 as f4,U::f5 as f5;");
        pigServer.registerQuery("G = GROUP Porj by (f1,f2,f3,f5);");
        pigServer.registerQuery("Final = foreach G generate SUM(Porj.f4) as total;");

        Iterator<Tuple> iter = pigServer.openIterator("Final");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().equals("(203)"));

        t = iter.next();
        Assert.assertTrue(t.toString().equals("(213)"));

        Assert.assertFalse(iter.hasNext());

    }

    // See PIG-2185
    @Test
    public void testProjectEmptyBag() throws Exception{
        String[] input = {
                "{(12)}",
                "{(23)}",
                ""
        };

        Util.createInputFile(cluster, "table_testProjectEmptyBag", input);

        pigServer.registerQuery("A = load 'table_testProjectEmptyBag' as (bg:bag{});");
        pigServer.registerQuery("B = FOREACH A { x = FILTER bg BY $0 == '12'; GENERATE x; };");

        Iterator<Tuple> iter = pigServer.openIterator("B");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().equals("({(12)})"));

        t = iter.next();
        Assert.assertTrue(t.toString().equals("({})"));

        Assert.assertTrue(iter.hasNext());

        t = iter.next();
        Assert.assertTrue(t.toString().equals("({})"));

        Assert.assertFalse(iter.hasNext());
    }

    // See PIG-2231
    @Test
    public void testLimitFlatten() throws Exception{
        String[] input = {
                "1\tA",
                "1\tB",
                "2\tC",
                "3\tD",
                "3\tE",
                "3\tF"
        };

        Util.createInputFile(cluster, "table_testLimitFlatten", input);

        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);");

        Iterator<Tuple> iter = pigServer.openIterator("flattened");

        String[] expected = new String[] {"(1,A)", "(1,B)", "(2,C)"};

        Util.checkQueryOutputsAfterSortRecursive(iter, expected,
            org.apache.pig.newplan.logical.Util.translateSchema(pigServer.dumpSchema("flattened")));
    }

    // See PIG-2237
    @Test
    public void testLimitAutoReducer() throws Exception{
        String[] input = {
                "1\tA",
                "4\tB",
                "2\tC",
                "3\tD",
                "6\tE",
                "5\tF"
        };

        Util.createInputFile(cluster, "table_testLimitAutoReducer", input);

        pigServer.getPigContext().getProperties().setProperty("pig.exec.reducers.bytes.per.reducer", "16");
        pigServer.registerQuery("A = load 'table_testLimitAutoReducer';");
        pigServer.registerQuery("B = order A by $0;");
        pigServer.registerQuery("C = limit B 2;");

        Iterator<Tuple> iter = pigServer.openIterator("C");

        Tuple t = iter.next();
        Assert.assertTrue(t.toString().equals("(1,A)"));

        t = iter.next();
        Assert.assertTrue(t.toString().equals("(2,C)"));

        Assert.assertFalse(iter.hasNext());
    }

    @SuppressWarnings("unchecked")
    @Test
    public void testCrossAfterGroupAll() throws Exception{
        String[] input = {
                "1\tA",
                "2\tB",
                "3\tC",
                "4\tD",
        };

        Util.createInputFile(cluster, "table_testCrossAfterGroupAll", input);

        try {
            pigServer.getPigContext().getProperties().setProperty("pig.exec.reducers.bytes.per.reducer", "40");
            pigServer.registerQuery("A = load 'table_testCrossAfterGroupAll' as (a0:int, a1:chararray);");
            pigServer.registerQuery("B = group A all;");
            pigServer.registerQuery("C = foreach B generate COUNT(A);");
            pigServer.registerQuery("D = cross A, C;");
            Path output = FileLocalizer.getTemporaryPath(pigServer.getPigContext());
            ExecJob job = pigServer.store("D", output.toString());
            FileSystem fs = output.getFileSystem(cluster.getConfiguration());
            FileStatus[] partFiles = fs.listStatus(output, new PathFilter() {
                @Override
                public boolean accept(Path path) {
                    if (path.getName().startsWith("part")) {
                        return true;
                    }
                    return false;
                }
            });

            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>();
            while (iter.hasNext()) {
                results.add(iter.next());
            }
            Collections.sort(results);
            Assert.assertEquals(4, results.size());
            Assert.assertEquals("(1,A,4)", results.get(0).toString());
            Assert.assertEquals("(2,B,4)", results.get(1).toString());
            Assert.assertEquals("(3,C,4)", results.get(2).toString());
            Assert.assertEquals("(4,D,4)", results.get(3).toString());
        } finally {
            pigServer.getPigContext().getProperties().remove("pig.exec.reducers.bytes.per.reducer");
        }
    }

    // see PIG-4392
    @Test
    public void testRankWithEmptyReduce() throws Exception {
        Util.createInputFile(cluster, "table_testRankWithEmptyReduce", new String[]{"1\t2\t3", "4\t5\t6", "7\t8\t9"});
        pigServer.setDefaultParallel(4);

        pigServer.registerQuery("d = load 'table_testRankWithEmptyReduce' as (a:int, b:int, c:int);");
        pigServer.registerQuery("e = rank d by a parallel 4;");

        Iterator<Tuple> iter = pigServer.openIterator("e");

        Collection<String> results = new HashSet<String>();
        results.add("(1,1,2,3)");
        results.add("(2,4,5,6)");
        results.add("(3,7,8,9)");

        Assert.assertTrue(results.contains(iter.next().toString()));
        Assert.assertTrue(results.contains(iter.next().toString()));
        Assert.assertTrue(results.contains(iter.next().toString()));
        Assert.assertFalse(iter.hasNext());
    }
}