* 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.

*/

    private PigServer pigServer;

    File tmpFile1;

    File tmpFile2;

    File tmpFile3;

    File tmpFile4;

    File tmpFile5;

    File tmpFile6;

    File tmpFile7;

    File tmpFile8;

    File tmpFile9;

    File tmpFile10;

    File tmpFile11;

    File tmpFile12;

    File logFile;

    private static final String simpleEchoStreamingCommand;

    static {

        String quote = "'";

        if (Util.WINDOWS) {

            quote = "\"";

        }

        simpleEchoStreamingCommand = "perl -ne " + quote + "print $_" + quote;

    }

    static public class MyFilterFunc extends FilterFunc {

        @Override

        public Boolean exec(Tuple input) {

            return true;

        }

    }

    @Before

    public void setUp() throws Exception{

        Logger logger = Logger.getLogger(ColumnPruneVisitor.class);

        logger.removeAllAppenders();

        logger.setLevel(Level.INFO);

        SimpleLayout layout = new SimpleLayout();

        logFile = File.createTempFile("log", "");

        FileAppender appender = new FileAppender(layout, logFile.toString(), false, false, 0);

        logger.addAppender(appender);

        pigServer = new PigServer("local");

        tmpFile1 = File.createTempFile("prune", "txt");

        PrintStream ps = new PrintStream(new FileOutputStream(tmpFile1));

        ps.println("1\t2\t3");

        ps.println("2\t5\t2");

        ps.close();

        tmpFile2 = File.createTempFile("prune", "txt");

        ps = new PrintStream(new FileOutputStream(tmpFile2));

        ps.println("1\t1");

        ps.println("2\t2");

        ps.close();

        tmpFile3 = File.createTempFile("prune", "txt");

        ps = new PrintStream(new FileOutputStream(tmpFile3));

        ps.println("1\t[key1#1,key2#2]");

        ps.println("2\t[key1#2,key2#4]");

        ps.close();

        tmpFile4 = File.createTempFile("prune", "txt");

        ps = new PrintStream(new FileOutputStream(tmpFile4));

        ps.println("1\t2\t3");

        ps.println("1\t2\t3");

        ps.close();

        tmpFile5 = File.createTempFile("prune", "txt");

        ps = new PrintStream(new FileOutputStream(tmpFile5));

        ps.println("1\t2\t3\t4");

        ps.println("2\t3\t4\t5");

        ps.close();

        tmpFile6 = File.createTempFile("prune", "txt");

        ps = new PrintStream(new FileOutputStream(tmpFile6));

        ps.println("\t2\t3");

        ps.println("2\t3\t4");

        ps.close();

        tmpFile7 = File.createTempFile("prune", "txt");

        ps = new PrintStream(new FileOutputStream(tmpFile7));

        ps.println("1\t1\t1");

        ps.println("2\t2\t2");

        ps.close();

        tmpFile8 = File.createTempFile("prune", "txt");

        ps = new PrintStream(new FileOutputStream(tmpFile8));

        ps.println("1\t2\t3\t4");

        ps.println("2\t5\t2\t3");

        ps.close();

        tmpFile9 = File.createTempFile("prune", "txt");

        ps = new PrintStream(new FileOutputStream(tmpFile9));

        ps.println("1\t[key1#1,key2#2]\t[key3#8,key4#9]");

        ps.println("2\t[key1#2,key2#4]\t[key3#8,key4#9]");

        ps.close();

        tmpFile10 = File.createTempFile("prune", "txt");

        ps = new PrintStream(new FileOutputStream(tmpFile10));

        ps.println("1\t[1#1,2#1]\t2");

        ps.close();

        tmpFile11 = File.createTempFile("prune", "txt");

        ps = new PrintStream(new FileOutputStream(tmpFile11));

        ps.println("1\t2\t3");

        ps.println("1\t3\t2");

        ps.println("2\t5\t2");

        ps.close();

        tmpFile12 = File.createTempFile("prune", "txt");

        ps = new PrintStream(new FileOutputStream(tmpFile12));

        ps.println("[key1#1,key2#2,cond#1]");

        ps.println("[key1#2,key2#3,cond#1]");

        ps.close();

    }

    @After

    public void tearDown() throws Exception{

        tmpFile1.delete();

        tmpFile2.delete();

        tmpFile3.delete();

        tmpFile4.delete();

        tmpFile5.delete();

        tmpFile6.delete();

        tmpFile7.delete();

        tmpFile8.delete();

        tmpFile9.delete();

        tmpFile10.delete();

        tmpFile11.delete();

        tmpFile12.delete();

        logFile.delete();

    }

    public boolean checkLogFileMessage(String[] messages) {

        BufferedReader reader = null;

        try {

            reader = new BufferedReader(new FileReader(logFile));

            List<String> logMessages=new ArrayList<String>();

            String line;

            while ((line=reader.readLine())!=null)

            {

                logMessages.add(line);

            }

            if (logMessages.size() > 0) {

                logMessages = TestHelper.sortSubFields(logMessages);

            }

            // Check if all messages appear in the log

            for (int i=0;i<messages.length;i++)

            {

                boolean found = false;

                for (int j=0;j<logMessages.size();j++)

                if (logMessages.get(j).contains(messages[i])) {

                    found = true;

                    break;

                }

                if (!found)

                    return false;

            }

            // Check no other log besides messages

            for (int i=0;i<logMessages.size();i++) {

                boolean found = false;

                for (int j=0;j<messages.length;j++) {

                    if (logMessages.get(i).contains(messages[j])) {

                        found = true;

                        break;

                    }

                }

                if (!found) {

                    if (logMessages.get(i).contains("Columns pruned for")||

                            logMessages.get(i).contains("Map key required for")) {

                        return false;

                    }

                }

            }

            return true;

        } catch (IOException e) {

            return false;

        }

    }

    public boolean emptyLogFileMessage()

    {

        if (!logFile.exists())

            return true;

        BufferedReader reader = null;

        String line;

        try {

            reader = new BufferedReader(new FileReader(logFile));

            while ((line=reader.readLine())!=null)

            {

                if (line!=null && !line.equals(""))

                    return false;

            }

            return true;

        }

        catch (IOException e) {

            return false;

        }

    }

    @Test

    public void testLoadForEach1() throws Exception{

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");

        pigServer.registerQuery("B = foreach A generate a1, a2;");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(2, t.size());

        assertEquals(2, t.get(0));

        assertEquals(3, t.get(1));

        assertTrue(iter.hasNext());

        t = iter.next();

        assertEquals(2, t.size());

        assertEquals(5, t.get(0));

        assertEquals(2, t.get(1));

        assertTrue(checkLogFileMessage(new String[]{"Columns pruned for A: $0"}));

    }

    @Test

    public void testLoadForEach2() throws Exception{

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");

        pigServer.registerQuery("B = foreach A generate a0, a2;");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(2, t.size());

        assertEquals(1, t.get(0));

        assertEquals(3, t.get(1));

        assertTrue(iter.hasNext());

        t = iter.next();

        assertEquals(2, t.size());

        assertEquals(2, t.get(0));

        assertEquals(2, t.get(1));

        assertTrue(checkLogFileMessage(new String[]{"Columns pruned for A: $1"}));

    }

    @Test

    public void testLoadForEach3() throws Exception{

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");

        pigServer.registerQuery("B = foreach A generate a0, a1;");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(2, t.size());

        assertEquals(1, t.get(0));

        assertEquals(2, t.get(1));

        assertTrue(iter.hasNext());

        t = iter.next();

        assertEquals(2, t.size());

        assertEquals(2, t.get(0));

        assertEquals(5, t.get(1));

        assertTrue(checkLogFileMessage(new String[]{"Columns pruned for A: $2"}));

    }

    @Test

    public void testJoin1() throws Exception{

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");

        pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' as (b0:int, b1:int);");

        pigServer.registerQuery("C = join A by a1, B by b1;");

        pigServer.registerQuery("D = foreach C generate a1, a2, b0, b1;");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(4, t.size());

        assertEquals(2, t.get(0));

        assertEquals(3, t.get(1));

        assertEquals(2, t.get(2));

        assertEquals(2, t.get(3));

        assertFalse(iter.hasNext());

        assertTrue(checkLogFileMessage(new String[]{"Columns pruned for A: $0"}));

    }

    @Test

    public void testJoin2() throws Exception{

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");

        pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' as (b0:int, b1:int);");

        pigServer.registerQuery("C = join A by a1, B by b1;");

        pigServer.registerQuery("D = foreach C generate a1, a2, b1;");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(3, t.size());

        assertEquals(2, t.get(0));

        assertEquals(3, t.get(1));

        assertEquals(2, t.get(2));

        assertFalse(iter.hasNext());

        assertTrue(checkLogFileMessage(new String[]{"Columns pruned for A: $0",

                "Columns pruned for B: $0"}));

    }

    @Test

    public void testForEachFilter() throws Exception{

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");

        pigServer.registerQuery("B = filter A by a2==3;");

        pigServer.registerQuery("C = foreach B generate a0, a1;");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(2, t.size());

        assertEquals(1, t.get(0));

        assertEquals(2, t.get(1));

        assertFalse(iter.hasNext());

        assertTrue(emptyLogFileMessage());

    }

    @Test

    public void testForEach1() throws Exception{

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");

        pigServer.registerQuery("B = foreach A generate a0, a1+a2;");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(2, t.size());

        assertEquals(1, t.get(0));

        assertEquals(5, t.get(1));

        assertTrue(iter.hasNext());

        t = iter.next();

        assertEquals(2, t.size());

        assertEquals(2, t.get(0));

        assertEquals(7, t.get(1));

        assertFalse(iter.hasNext());

        assertTrue(emptyLogFileMessage());

    }

    @Test

    public void testForEach2() throws Exception{

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");

        pigServer.registerQuery("B = foreach A generate a0 as b0, *;");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(4, t.size());

        assertEquals(1, t.get(0));

        assertEquals(1, t.get(1));

        assertEquals(2, t.get(2));

        assertEquals(3, t.get(3));

        assertTrue(iter.hasNext());

        t = iter.next();

        assertEquals(4, t.size());

        assertEquals(2, t.get(0));

        assertEquals(2, t.get(1));

        assertEquals(5, t.get(2));

        assertEquals(2, t.get(3));

        assertFalse(iter.hasNext());

        assertTrue(emptyLogFileMessage());

    }

    @Test

    public void testSplit1() throws Exception{

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0, a1, a2);");

        pigServer.registerQuery("split A into B if $0<=1, C if $0>1;");

        pigServer.registerQuery("D = foreach B generate $1;");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(1, t.size());

        assertEquals("2", t.get(0).toString());

        assertFalse(iter.hasNext());

        assertTrue(checkLogFileMessage(new String[]{"Columns pruned for A: $2"}));

    }

    @Test

    public void testSplit2() throws Exception{

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0, a1, a2);");

        pigServer.registerQuery("split A into B if $0<=1, C if $0>1;");

        pigServer.registerQuery("D = foreach B generate $1;");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(1, t.size());

        assertEquals("2", t.get(0).toString());

        assertFalse(iter.hasNext());

        assertTrue(checkLogFileMessage(new String[]{"Columns pruned for A: $2"}));

    }

    @Test

    public void testForeachNoSchema1() throws Exception {

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "';");

        pigServer.registerQuery("B = foreach A generate $1, $2;");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(2, t.size());

        assertEquals("2", t.get(0).toString());

        assertEquals("3", t.get(1).toString());

        assertTrue(iter.hasNext());

        t = iter.next();

        assertEquals(2, t.size());

        assertEquals("5", t.get(0).toString());

        assertEquals("2", t.get(1).toString());

        assertTrue(emptyLogFileMessage());

    }

    @Test

    public void testForeachNoSchema2() throws Exception {

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "';");

        pigServer.registerQuery("B = foreach A generate $1, 'aoeuaoeu';");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(2, t.size());

        assertEquals("2", t.get(0).toString());

        assertEquals("aoeuaoeu", t.get(1).toString());

        assertTrue(iter.hasNext());

        t = iter.next();

        assertEquals(2, t.size());

        assertEquals("5", t.get(0).toString());

        assertEquals("aoeuaoeu", t.get(1).toString());

        assertTrue(emptyLogFileMessage());

    }

    @Test

    public void testCoGroup1() throws Exception {

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1:int, a2);");

        pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' AS (b0, b1:int);");

        pigServer.registerQuery("C = cogroup A by $1, B by $1;");

        pigServer.registerQuery("D = foreach C generate AVG($1.$1);");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(1, t.size());

        assertNull(t.get(0));

        assertTrue(iter.hasNext());

        t = iter.next();

        assertEquals(1, t.size());

        assertEquals("2.0", t.get(0).toString());

        assertTrue(iter.hasNext());

        t = iter.next();

        assertEquals(1, t.size());

        assertEquals("5.0", t.get(0).toString());

        assertFalse(iter.hasNext());

        assertTrue(checkLogFileMessage(new String[]{"Columns pruned for B: $0"}));

    }

    @Test

    public void testCoGroup2() throws Exception {

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1:int, a2);");

        pigServer.registerQuery("B = group A all;");

        pigServer.registerQuery("C = foreach B generate $1;");

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

        String[] expected = new String[] {

                "({(1,2,3),(2,5,2)})"

        };

        assertTrue(iter.hasNext());

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

        assertTrue(emptyLogFileMessage());

    }

    @Test

    public void testCoGroup3() throws Exception {

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1:int, a2);");

        pigServer.registerQuery("B = group A by $1;");

        pigServer.registerQuery("C = foreach B generate $1, '1';");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(2, t.size());

        assertEquals("{(1,2,3)}", t.get(0).toString());

        assertEquals("1", t.get(1).toString());

        assertTrue(iter.hasNext());

        t = iter.next();

        assertEquals(2, t.size());

        assertEquals("{(2,5,2)}", t.get(0).toString());

        assertEquals("1", t.get(1).toString());

        assertFalse(iter.hasNext());

        assertTrue(emptyLogFileMessage());

    }

    @Test

    public void testCoGroup4() throws Exception {

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1:int, a2);");

        pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' AS (b0, b1:int);");

        pigServer.registerQuery("C = cogroup A by ($1), B by ($1);");

        pigServer.registerQuery("D = foreach C generate $1.$1, $2.$1;");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(2, t.size());

        assertEquals("{}", t.get(0).toString());

        assertEquals("{(1)}", t.get(1).toString());

        assertTrue(iter.hasNext());

        t = iter.next();

        assertEquals(2, t.size());

        assertEquals("{(2)}", t.get(0).toString());

        assertEquals("{(2)}", t.get(1).toString());

        assertTrue(iter.hasNext());

        t = iter.next();

        assertEquals(2, t.size());

        assertEquals("{(5)}", t.get(0).toString());

        assertEquals("{}", t.get(1).toString());

        assertFalse(iter.hasNext());

        assertTrue(emptyLogFileMessage());

    }

    @Test

    public void testCoGroup5() throws Exception {

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");

        pigServer.registerQuery("B = group A by (a0, a1);");

        pigServer.registerQuery("C = foreach B generate flatten(group);");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(2, t.size());

        assertEquals("1", t.get(0).toString());

        assertEquals("2", t.get(1).toString());

        assertTrue(iter.hasNext());

        t = iter.next();

        assertEquals(2, t.size());

        assertEquals("2", t.get(0).toString());

        assertEquals("5", t.get(1).toString());

        assertFalse(iter.hasNext());

        assertTrue(checkLogFileMessage(new String[]{"Columns pruned for A: $2"}));

    }

    @Test

    public void testDistinct1() throws Exception {

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile4.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");

        pigServer.registerQuery("B = distinct A;");

        pigServer.registerQuery("C = foreach B generate $0;");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(1, t.size());

        assertEquals("1", t.get(0).toString());

        assertFalse(iter.hasNext());

        assertTrue(emptyLogFileMessage());

    }

    @Test

    public void testStream1() throws Exception {

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");

        pigServer.registerQuery("B = stream A through `" + simpleEchoStreamingCommand + "`;");

        pigServer.registerQuery("C = foreach B generate $0;");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(1, t.size());

        assertEquals("1", t.get(0).toString());

        assertTrue(iter.hasNext());

        t = iter.next();

        assertEquals(1, t.size());

        assertEquals("2", t.get(0).toString());

        assertFalse(iter.hasNext());

        assertTrue(emptyLogFileMessage());

    }

    @Test

    public void testBinCond1() throws Exception {

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile5.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2, a3);");

        pigServer.registerQuery("B = foreach A generate ($1 == '2'? $2 : $3);");

        pigServer.registerQuery("C = foreach B generate $0;");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(1, t.size());

        assertEquals("3", t.get(0).toString());

        assertTrue(iter.hasNext());

        t = iter.next();

        assertEquals(1, t.size());

        assertEquals("5", t.get(0).toString());

        assertFalse(iter.hasNext());

        assertTrue(checkLogFileMessage(new String[]{"Columns pruned for A: $0"}));

    }

    @Test

    public void testCoGroup6() throws Exception {

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");

        pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' AS (b0, b1);");

        pigServer.registerQuery("C = cogroup A by ($1), B by ($1);");

        pigServer.registerQuery("D = foreach C generate A, flatten(B.($0, $1));");

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

        assertTrue(iter.hasNext());

        Tuple t = iter.next();

        assertEquals(3, t.size());

        assertEquals("{}", t.get(0).toString());

        assertEquals("1", t.get(1).toString());

        assertEquals("1", t.get(2).toString());

        assertTrue(iter.hasNext());

        t = iter.next();

        assertEquals(3, t.size());

        assertEquals("{(1,2,3)}", t.get(0).toString());

        assertEquals("2", t.get(1).toString());

        assertEquals("2", t.get(2).toString());

        assertFalse(iter.hasNext());

        assertTrue(emptyLogFileMessage());

    }

    @Test

    public void testCoGroup7() throws Exception {

        pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");

        pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' AS (b0, b1);");

        pigServer.registerQuery("C = cogroup A by ($1), B by ($1);");

        pigServer.registerQuery("D = foreach C {B = order B by $0;generate FLATTEN(A), B.($1);};");

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