spork-master/test/org/apache/pig/tez/TestTezCompiler.java
/*
 * 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.tez;

import static org.junit.Assert.assertEquals;

import java.io.ByteArrayOutputStream;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.PrintStream;
import java.util.Properties;

import org.apache.pig.PigConfiguration;
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.tez.TezLauncher;
import org.apache.pig.backend.hadoop.executionengine.tez.TezLocalExecType;
import org.apache.pig.backend.hadoop.executionengine.tez.plan.TezPlanContainer;
import org.apache.pig.backend.hadoop.executionengine.tez.plan.TezPlanContainerPrinter;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.plan.NodeIdGenerator;
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;

/**
 * Test cases to test the TezCompiler. 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 TestTezCompiler {
    private static PigContext pc;
    private static PigServer pigServer;
    private static final int MAX_SIZE = 100000;

    // 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 {
        pc = new PigContext(new TezLocalExecType(), new Properties());
    }

    @AfterClass
    public static void tearDownAfterClass() throws Exception {
    }

    @Before
    public void setUp() throws ExecException {
        resetScope();
        pc.getProperties().remove(PigConfiguration.PIG_OPT_MULTIQUERY);
        pc.getProperties().remove(PigConfiguration.PIG_TEZ_OPT_UNION);
        pc.getProperties().remove(PigConfiguration.PIG_EXEC_NO_SECONDARY_KEY);
        pigServer = new PigServer(pc);
    }

    private void resetScope() {
        NodeIdGenerator.reset();
        PigServer.resetScope();
        TezPlanContainer.resetScope();
    }

    @Test
    public void testFilter() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:int);" +
                "b = filter a by x > 0;" +
                "c = foreach b generate y;" +
                "store c into 'file:///tmp/output';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Filter-1.gld");
    }

    @Test
    public void testGroupBy() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:int);" +
                "b = group a by x;" +
                "c = foreach b generate group, COUNT(a.x);" +
                "store c into 'file:///tmp/output';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Group-1.gld");
    }

    @Test
    public void testJoin() throws Exception {
        String query =
                "a = load 'file:///tmp/input1' as (x:int, y:int);" +
                "b = load 'file:///tmp/input2' as (x:int, z:int);" +
                "c = join a by x, b by x;" +
                "d = foreach c generate a::x as x, y, z;" +
                "store d into 'file:///tmp/output';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Join-1.gld");
    }

    @Test
    public void testSkewedJoin() throws Exception {
        String query =
                "a = load 'file:///tmp/input1' as (x:int, y:int);" +
                "b = load 'file:///tmp/input2' as (x:int, z:int);" +
                "c = join a by x, b by x using 'skewed';" +
                "d = foreach c generate a::x as x, y, z;" +
                "store d into 'file:///tmp/output';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-SkewJoin-1.gld");
    }

    @Test
    public void testLimit() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:int);" +
                "b = limit a 10;" +
                "c = foreach b generate y;" +
                "store c into 'file:///tmp/output';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Limit-1.gld");
    }

    @Test
    public void testLimitOrderby() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:int);" +
                "b = order a by x, y;" +
                "c = limit b 10;" +
                "store c into 'file:///tmp/output';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Limit-2.gld");
    }

    @Test
    public void testLimitScalarOrderby() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:int);" +
                "b = order a by x, y;" +
                "g = group a all;" +
                "h = foreach g generate COUNT(a) as sum;" +
                "c = limit b h.sum/2;" +
                "store c into 'file:///tmp/output';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Limit-3.gld");
    }

    @Test
    public void testDistinct() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:int);" +
                "b = distinct a;" +
                "c = foreach b generate y;" +
                "store c into 'file:///tmp/output';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Distinct-1.gld");
    }

    @Test
    public void testDistinctAlgebraicUdfCombiner() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:int);" +
                "b = group a by x;" +
                "c = foreach b { d = distinct a; generate COUNT(d); };" +
                "store c into 'file:///tmp/output';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Distinct-2.gld");
    }



    @Test
    public void testReplicatedJoinInMapper() throws Exception {
        String query =
                "a = load 'file:///tmp/input1' as (x:int, y:int);" +
                "b = load 'file:///tmp/input2' as (x:int, z:int);" +
                "c = load 'file:///tmp/input3' as (x:int, z:int);" +
                "d = join a by x, b by x, c by x using 'replicated';" +
                "store d into 'file:///tmp/output/d';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-FRJoin-1.gld");
    }

    @Test
    public void testReplicatedJoinInReducer() throws Exception {
        String query =
                "a = load 'file:///tmp/input1' as (x:int, y:int);" +
                "b = group a by x;" +
                "b1 = foreach b generate group, COUNT(a.y);" +
                "c = load 'file:///tmp/input2' as (x:int, z:int);" +
                "d = join b1 by group, c by x using 'replicated';" +
                "store d into 'file:///tmp/output/e';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-FRJoin-2.gld");
    }

    @Test
    public void testStream() throws Exception {
        String query =
                "a = load 'file:///tmp/input' using PigStorage(',') as (x:int, y:int);" +
                "b = stream a through `stream.pl -n 5`;" +
                "STORE b INTO 'file:///tmp/output';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Stream-1.gld");
    }

    @Test
    public void testSecondaryKeySort() throws Exception {
        String query =
                "a = load 'file:///tmp/input' using PigStorage(',') as (x:int, y:int, z:int);" +
                "b = group a by $0;" +
                "c = foreach b { d = limit a 10; e = order d by $1; f = order e by $0; generate group, f;};"+
                "store c INTO 'file:///tmp/output';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-SecKeySort-1.gld");

        // With optimization turned off
        setProperty(PigConfiguration.PIG_EXEC_NO_SECONDARY_KEY, "true");
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-SecKeySort-2.gld");
    }

    @Test
    public void testOrderBy() throws Exception {
        String query =
                "a = load 'file:///tmp/input' using PigStorage(',') as (x:int, y:int);" +
                "b = order a by x;" +
                "STORE b INTO 'file:///tmp/output';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Order-1.gld");
    }

    // PIG-3759, PIG-3781
    // Combiner should not be added in case of co-group
    @Test
    public void testCogroupWithAlgebraiceUDF() throws Exception {
        String query =
                "a = load 'file:///tmp/input1' as (x:int, y:int);" +
                "b = load 'file:///tmp/input2' as (x:int, z:int);" +
                "c = cogroup a by x, b by x;" +
                "d = foreach c generate group, COUNT(a.y), COUNT(b.z);" +
                "store d into 'file:///tmp/output/d';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Cogroup-1.gld");
    }

    @Test
    public void testMulitQueryWithSplitSingleVertex() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:int);" +
                "split a into b if x <= 5, c if x <= 10, d if x >10;" +
                "store b into 'file:///tmp/output/b';" +
                "store c into 'file:///tmp/output/c';" +
                "store d into 'file:///tmp/output/d';";

        setProperty(PigConfiguration.PIG_OPT_MULTIQUERY, "" + true);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-MQ-1.gld");
        setProperty(PigConfiguration.PIG_OPT_MULTIQUERY, "" + false);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-MQ-1-OPTOFF.gld");
    }

    @Test
    public void testMulitQueryWithSplitMultiVertex() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:int);" +
                "split a into b if x <= 5, c if x <= 10, d if x >10;" +
                "split b into e if x < 3, f if x >= 3;" +
                // No Combiner on the edge to b1/b2 vertex as both b1 and b2 are stored
                "b1 = group b by x;" +
                "b2 = foreach b1 generate group, SUM(b.x);" +
                // Case of two outputs within a split going to same edge as input
                "c1 = join c by x, b by x;" +
                "c2 = group c by x;" +
                // Combiner on the edge to c3 vertex
                "c3 = foreach c2 generate group, SUM(c.x);" +
                "d1 = filter d by x == 5;" +
                "e1 = order e by x;" +
                // TODO: Physical plan has extra split for f1 - 1-2: Split - scope-80
                // POSplit has only 1 sub plan. Optimized and removed in MR plan.
                // Needs to be removed in Tez plan as well.
                "f1 = limit f 1;" +
                "f2 = union d1, f1;" +
                "store b1 into 'file:///tmp/output/b1';" +
                "store b2 into 'file:///tmp/output/b2';" +
                "store c1 into 'file:///tmp/output/c1';" +
                "store c3 into 'file:///tmp/output/c1';" +
                "store d1 into 'file:///tmp/output/d1';" +
                "store e1 into 'file:///tmp/output/e1';" +
                "store f1 into 'file:///tmp/output/f1';" +
                "store f2 into 'file:///tmp/output/f2';";

        setProperty(PigConfiguration.PIG_OPT_MULTIQUERY, "" + true);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-MQ-2.gld");
        setProperty(PigConfiguration.PIG_OPT_MULTIQUERY, "" + false);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-MQ-2-OPTOFF.gld");
    }

    @Test
    public void testMultiQueryWithGroupBy() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:int);" +
                "b = group a by x;" +
                "b = foreach b generate group, COUNT(a.x);" +
                "c = group a by (x,y);" +
                "c = foreach c generate group, COUNT(a.y);" +
                "store b into 'file:///tmp/output/b';" +
                "store c into 'file:///tmp/output/c';";

        setProperty(PigConfiguration.PIG_OPT_MULTIQUERY, "" + true);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-MQ-3.gld");
        setProperty(PigConfiguration.PIG_OPT_MULTIQUERY, "" + false);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-MQ-3-OPTOFF.gld");
    }

    @Test
    public void testMultiQueryWithJoin() throws Exception {
        String query =
                "a = load 'file:///tmp/input1' as (x:int, y:int);" +
                "b = load 'file:///tmp/input2' as (x:int, z:int);" +
                "c = join a by x, b by x;" +
                "d = foreach c generate $0, $1, $3;" +
                "e = foreach c generate $0, $1, $2, $3;" +
                "store c into 'file:///tmp/output/c';" +
                "store d into 'file:///tmp/output/d';" +
                "store e into 'file:///tmp/output/e';";

        setProperty(PigConfiguration.PIG_OPT_MULTIQUERY, "" + true);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-MQ-4.gld");
        setProperty(PigConfiguration.PIG_OPT_MULTIQUERY, "" + false);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-MQ-4-OPTOFF.gld");
    }

    @Test
    public void testMultiQueryWithNestedSplit() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:int);" +
                "b = group a by x;" + //b: {group: int,a: {(x: int,y: int)}}
                "store b into 'file:///tmp/output/b';" +
                "c = foreach b generate a.x, a.y;" + //c: {{(x: int)},{(y: int)}}
                "store c into 'file:///tmp/output/c';" +
                "d = foreach b GENERATE FLATTEN(a);" + //d: {a::x: int,a::y: int}
                "store d into 'file:///tmp/output/d';" +
                "e = foreach d GENERATE a::x, a::y;" +
                "store e into 'file:///tmp/output/e';";

        setProperty(PigConfiguration.PIG_OPT_MULTIQUERY, "" + true);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-MQ-5.gld");
        setProperty(PigConfiguration.PIG_OPT_MULTIQUERY, "" + false);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-MQ-5-OPTOFF.gld");
    }

    @Test
    public void testUnionStore() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:chararray);" +
                "b = load 'file:///tmp/input' as (y:chararray, x:int);" +
                "c = union onschema a, b;" +
                "store c into 'file:///tmp/output';";

        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + true);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-1.gld");
        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + false);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-1-OPTOFF.gld");
    }

    @Test
    public void testUnionGroupBy() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:int);" +
                "b = load 'file:///tmp/input' as (y:int, x:int);" +
                "c = union onschema a, b;" +
                "d = group c by x;" +
                "e = foreach d generate group, SUM(c.y);" +
                "store e into 'file:///tmp/output';";

        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + true);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-2.gld");
        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + false);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-2-OPTOFF.gld");
    }

    @Test
    public void testUnionJoin() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:chararray);" +
                "b = load 'file:///tmp/input' as (y:chararray, x:int);" +
                "c = union onschema a, b;" +
                "d = load 'file:///tmp/input1' as (x:int, z:chararray);" +
                "e = join c by x, d by x;" +
                "store e into 'file:///tmp/output';";

        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + true);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-3.gld");
        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + false);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-3-OPTOFF.gld");
    }


    @Test
    public void testUnionReplicateJoin() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:chararray);" +
                "b = load 'file:///tmp/input' as (y:chararray, x:int);" +
                "c = union onschema a, b;" +
                "d = load 'file:///tmp/input1' as (x:int, z:chararray);" +
                "e = join c by x, d by x using 'replicated';" +
                "store e into 'file:///tmp/output';";

        //TODO: PIG-3856 Not optimized
        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + true);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-4.gld");
        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + false);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-4-OPTOFF.gld");

        query =
                "a = load 'file:///tmp/input' as (x:int, y:chararray);" +
                "b = load 'file:///tmp/input' as (y:chararray, x:int);" +
                "c = union onschema a, b;" +
                "d = load 'file:///tmp/input1' as (x:int, z:chararray);" +
                "e = join d by x, c by x using 'replicated';" +
                "store e into 'file:///tmp/output';";

        // Optimized
        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + true);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-5.gld");
        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + false);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-5-OPTOFF.gld");
    }

    @Test
    public void testUnionSkewedJoin() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:chararray);" +
                "b = load 'file:///tmp/input' as (y:chararray, x:int);" +
                "c = union onschema a, b;" +
                "d = load 'file:///tmp/input1' as (x:int, z:chararray);" +
                "e = join c by x, d by x using 'skewed';" +
                "store e into 'file:///tmp/output';";

        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + true);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-6.gld");
        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + false);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-6-OPTOFF.gld");
    }

    @Test
    public void testUnionOrderby() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:chararray);" +
                "b = load 'file:///tmp/input' as (y:chararray, x:int);" +
                "c = union onschema a, b;" +
                "d = order c by x;" +
                "store d into 'file:///tmp/output';";

        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + true);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-7.gld");
        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + false);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-7-OPTOFF.gld");
    }

    //TODO: PIG-3854 Limit is too convoluted and can be simplified.
    @Test
    public void testUnionLimit() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:chararray);" +
                "b = load 'file:///tmp/input' as (y:chararray, x:int);" +
                "c = union onschema a, b;" +
                "d = limit c 1;" +
                "store d into 'file:///tmp/output';";

        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + true);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-8.gld");
        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + false);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-8-OPTOFF.gld");
    }

    @Test
    public void testUnionSplit() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:chararray);" +
                "b = load 'file:///tmp/input' as (y:chararray, x:int);" +
                "split a into a1 if x > 100, a2 otherwise;" +
                "c = union onschema a1, b;" +
                "split c into d if x > 500, e otherwise;" +
                "store a2 into 'file:///tmp/output/a2';" +
                "store d into 'file:///tmp/output/d';" +
                "store e into 'file:///tmp/output/e';";

        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + true);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-9.gld");
        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + false);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-9-OPTOFF.gld");
    }

    @Test
    public void testUnionUnion() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:chararray);" +
                "b = load 'file:///tmp/input' as (y:chararray, x:int);" +
                "c = union onschema a, b;" +
                "d = load 'file:///tmp/input1' as (x:int, y:chararray);" +
                "e = union onschema c, d;" +
                "f = group e by x;" +
                "store f into 'file:///tmp/output';";

        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + true);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-10.gld");
        resetScope();
        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + false);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-10-OPTOFF.gld");
    }

    //TODO: union followed by union followed by store does not work.
    //@Test
    public void testUnionUnionStore() throws Exception {
        String query =
                "a = load 'file:///tmp/input' as (x:int, y:chararray);" +
                "b = load 'file:///tmp/input' as (y:chararray, x:int);" +
                "c = union onschema a, b;" +
                "d = load 'file:///tmp/input1' as (x:int, y:chararray);" +
                "e = union onschema c, d;" +
                "store e into 'file:///tmp/output';";

        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + true);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-11.gld");
        resetScope();
        setProperty(PigConfiguration.PIG_TEZ_OPT_UNION, "" + false);
        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Union-11-OPTOFF.gld");
    }

    @Test
    public void testRank() throws Exception {
        String query =
                "a = load 'file:///tmp/input1' as (x:int, y:int);" +
                "b = rank a;" +
                "store b into 'file:///tmp/output/d';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Rank-1.gld");
    }

    @Test
    public void testRankBy() throws Exception {
        //TODO: Physical plan (affects both MR and Tez) has extra job before order by. Does not look right.
        String query =
                "a = load 'file:///tmp/input1' as (x:int, y:int);" +
                "b = rank a by x;" +
                "store b into 'file:///tmp/output/d';";

        run(query, "test/org/apache/pig/test/data/GoldenFiles/tez/TEZC-Rank-2.gld");
    }

    private void setProperty(String property, String value) {
        pigServer.getPigContext().getProperties().setProperty(property, value);
    }

    private void run(String query, String expectedFile) throws Exception {
        PhysicalPlan pp = Util.buildPp(pigServer, query);
        TezLauncher launcher = new TezLauncher();
        pc.inExplain = true;
        TezPlanContainer tezPlanContainer = launcher.compile(pp, pc);

        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        PrintStream ps = new PrintStream(baos);
        TezPlanContainerPrinter printer = new TezPlanContainerPrinter(ps, tezPlanContainer);
        printer.visit();
        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)));
    }
}

