/*
 * 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.IOException;
import java.util.Iterator;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.pig.ExecType;
import org.apache.pig.IndexableLoadFunc;
import org.apache.pig.LoadCaster;
import org.apache.pig.LoadFunc;
import org.apache.pig.PigException;
import org.apache.pig.PigServer;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRConfiguration;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.plans.MROperPlan;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.impl.util.LogUtils;
import org.apache.pig.test.utils.TestHelper;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

public class TestMergeJoin {

    private static final String INPUT_FILE = "testMergeJoinInput.txt";
    private static final String INPUT_FILE2 = "testMergeJoinInput2.txt";
    private static final String INPUT_FILE3 = "testMergeJoinInput3.txt";
    private PigServer pigServer;
    private static MiniGenericCluster cluster = MiniGenericCluster.buildCluster();

    public TestMergeJoin() throws ExecException{

        Properties props = cluster.getProperties();
        props.setProperty(MRConfiguration.MAP_MAX_ATTEMPTS, "1");
        props.setProperty(MRConfiguration.REDUCE_MAX_ATTEMPTS, "1");
        pigServer = new PigServer(cluster.getExecType(), props);
    }
    /**
     * @throws java.lang.Exception
     */
    @Before
    public void setUp() throws Exception {
        int LOOP_SIZE = 3;
        String[] input = new String[LOOP_SIZE*LOOP_SIZE];
        int k = 0;
        for(int i = 1; i <= LOOP_SIZE; i++) {
            String si = i + "";
            for(int j=1;j<=LOOP_SIZE;j++)
                input[k++] = si + "\t" + j;
        }
        Util.createInputFile(cluster, INPUT_FILE, input);
        
        Util.createInputFile(cluster, INPUT_FILE2, new String[]{"2"});

        String[] input3 = new String[LOOP_SIZE];
        for (int i = 0; i<= LOOP_SIZE-1; i++) {
            input3[i] = "(" + (i + 1) + ")\t + {";
            for(int j=1;j<=LOOP_SIZE;j++) {
                input3[i] = input3[i] + "(" + j + ")";
                if (j!=LOOP_SIZE) {
                    input3[i] = input3[i] + ",";
                }
            }
            input3[i] = input3[i] + "}";
        }
        Util.createInputFile(cluster, INPUT_FILE3, input3);
    }

    @AfterClass
    public static void oneTimeTearDown() throws Exception {
        cluster.shutDown();
    }
    /**
     * @throws java.lang.Exception
     */
    @After
    public void tearDown() throws Exception {
        Util.deleteFile(cluster, INPUT_FILE);
        Util.deleteFile(cluster, INPUT_FILE2);
        Util.deleteFile(cluster, INPUT_FILE3);
    }

    @Test
    public void testRecursiveFileListing() throws IOException{
        Util.createInputFile(cluster, "foo/bar/test.dat", new String[]{"2"});
        pigServer.registerQuery("A = LOAD 'foo';");
        pigServer.registerQuery("B = LOAD 'foo';");
        DataBag dbMergeJoin = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        {
            pigServer.registerQuery("C = join A by $0, B by $0 using 'merge';");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbMergeJoin.add(iter.next());
            }
        }
        {
            pigServer.registerQuery("C = join A by $0, B by $0;");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }
        Assert.assertEquals(dbMergeJoin.size(), dbshj.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbMergeJoin, dbshj));
        Util.deleteFile(cluster,"foo/bar/test.dat");
    }
    
    @Test
    public void testMergeJoinSimplest() throws IOException{
        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "';");
        pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "';");
        DataBag dbMergeJoin = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        {
            pigServer.registerQuery("C = join A by $0, B by $0 using 'merge';");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbMergeJoin.add(iter.next());
            }
        }
        {
            pigServer.registerQuery("C = join A by $0, B by $0;");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }
        Assert.assertEquals(dbMergeJoin.size(), dbshj.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbMergeJoin, dbshj));
    }

    @Test
    public void testMergeJoinWithReplicatedJoin() throws IOException{
        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' as (f1:int,f2:int);");
        pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "' as (f1:int,f2:int);");
        pigServer.registerQuery("C = LOAD '" + INPUT_FILE + "' as (f1:int,f2:int);");
        DataBag dbMergeJoin = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        {
            pigServer.registerQuery("D = join A by f1, B by f1 using 'replicated';");
            pigServer.registerQuery("E = join D by A::f1, C by f1 using 'merge';");
            Iterator<Tuple> iter = pigServer.openIterator("E");

            while(iter.hasNext()) {
                dbMergeJoin.add(iter.next());
            }
        }
        {
            pigServer.registerQuery("D = join A by f1, B by f1 using 'replicated';");
            pigServer.registerQuery("E = join D by A::f1, C by f1;");
            Iterator<Tuple> iter = pigServer.openIterator("E");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }
        Assert.assertEquals(dbshj.size(), dbMergeJoin.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbMergeJoin, dbshj));
    }

    @Test
    public void testMergeJoinWithForeachFlatten() throws IOException{
        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' as (f1:int,f2:int);");
        pigServer.registerQuery("B = LOAD '" + INPUT_FILE3 + "' as (t:(f1:int), b:{(f1:int)});");
        pigServer.registerQuery("C = foreach B generate flatten(t) as f1:int, flatten(b);");
        pigServer.registerQuery("D = join C by f1, A by f1 using 'merge';");
        DataBag dbMergeJoin = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        {
            pigServer.registerQuery("D = join C by f1, A by f1 using 'merge';");
            Iterator<Tuple> iter = pigServer.openIterator("D");

            while(iter.hasNext()) {
                dbMergeJoin.add(iter.next());
            }
        }
        {
            pigServer.registerQuery("D = join C by f1, A by f1;");
            Iterator<Tuple> iter = pigServer.openIterator("D");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }
        Assert.assertEquals(dbshj.size(), dbMergeJoin.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbMergeJoin, dbshj));
    }

    @Test
    public void testMergeJoinOnMultiFields() throws IOException{
        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "';");
        pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "';");
        DataBag dbMergeJoin = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        {
            pigServer.registerQuery("C = join A by ($0,$1), B by ($0,$1) using 'merge';");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbMergeJoin.add(iter.next());
            }
        }
        {
            pigServer.registerQuery("C = join A by ($0,$1), B by ($0,$1);");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }
        Assert.assertEquals(dbMergeJoin.size(), dbshj.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbMergeJoin, dbshj));
    }

    @Test
    public void testMergeJoinWithExpr() throws IOException{
        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "';");
        pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "';");
        DataBag dbMergeJoin = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        {
            pigServer.registerQuery("C = join A by ($0+10), B by ($0+10) using 'merge';");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbMergeJoin.add(iter.next());
            }
        }
        {
            pigServer.registerQuery("C = join A by ($0+10), B by ($0+10);");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }
        Assert.assertEquals(dbMergeJoin.size(), dbshj.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbMergeJoin, dbshj));
    }

    @Test
    public void testMergeJoinOutWithSchema() throws IOException{

        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' as (x:int,y:int);");
        pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "' as (x:int,y:int);");
        DataBag dbMergeJoin = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        {
            pigServer.registerQuery("C = join A by $0, B by $0 using 'merge';");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbMergeJoin.add(iter.next());
            }
        }
        {
            pigServer.registerQuery("C = join A by $0, B by $0;");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }
        Assert.assertEquals(dbMergeJoin.size(), dbshj.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbMergeJoin, dbshj));
    }

    @Test
    public void testMergeJoinOutWithFilters() throws IOException{

        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "';");
        pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "';");
        pigServer.registerQuery("C = FILTER A by $1 > 1;"); 
        pigServer.registerQuery("D = FILTER B by $1 > 1;"); 
        DataBag dbMergeJoin = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        {
            pigServer.registerQuery("E = join C by $0, D by $0 using 'merge';");
            Iterator<Tuple> iter = pigServer.openIterator("E");

            while(iter.hasNext()) {
                dbMergeJoin.add(iter.next());
            }
        }
        {
            pigServer.registerQuery("E = join C by $0, D by $0;");
            Iterator<Tuple> iter = pigServer.openIterator("E");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }
        Assert.assertEquals(dbMergeJoin.size(), dbshj.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbMergeJoin, dbshj));
    }

    @Test
    public void testMergeJoinOutWithProjects() throws IOException{

        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' as (f1,f2);");
        pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "'as (f1,f2);");
        pigServer.registerQuery("C = foreach A generate f1;"); 
        pigServer.registerQuery("D = foreach B generate f1;"); 
        DataBag dbMergeJoin = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        {
            pigServer.registerQuery("E = join C by f1, D by f1 using 'merge';");
            Iterator<Tuple> iter = pigServer.openIterator("E");

            while(iter.hasNext()) {
                dbMergeJoin.add(iter.next());
            }
        }
        {
            pigServer.registerQuery("E = join C by f1, D by f1;");
            Iterator<Tuple> iter = pigServer.openIterator("E");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }
        Assert.assertEquals(dbMergeJoin.size(), dbshj.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbMergeJoin, dbshj));
    }

    @Test
    public void testMergeJoinOutPipeline() throws IOException{
        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' as (f1:int,f2:int);");
        pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "' as (f1:int,f2:int);");
        DataBag dbmrj = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        {
            pigServer.registerQuery("C = join A by $0, B by $0 using 'merge';");
            pigServer.registerQuery("G = LOAD '" + INPUT_FILE + "' as (f1:int,f2:int);");
            pigServer.registerQuery("H = LOAD '" + INPUT_FILE + "' as (f1:int,f2:int);");
            pigServer.registerQuery("D = join G by $0, H by $0 using 'merge';");
            pigServer.registerQuery("E = union C,D;");
            pigServer.registerQuery("F = filter E by 1 == 1;");
            Iterator<Tuple> iter = pigServer.openIterator("F");

            while(iter.hasNext()) {
                dbmrj.add(iter.next());
            }
        }
        // Note that these two queries are setup little differently. This is because if there is a Split in Plan, Merge Join fails.
        // When work in MRCompiler finishes, this query along with merge join preceded by order-by should be two test-cases for it.
        {
            pigServer.registerQuery("C = join A by $0, B by $0;");
            pigServer.registerQuery("D = join A by $0, B by $0;");
            pigServer.registerQuery("E = union C,D;");
            pigServer.registerQuery("F = filter E by 1 == 1;");
            Iterator<Tuple> iter = pigServer.openIterator("F");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }
        Assert.assertEquals(dbmrj.size(), dbshj.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbmrj, dbshj));
    }

    @Test
    public void testMergeJoinWithNulls() throws IOException{

        String[] input = new String[3*3];
        input[0] = "\t2";
        input[1] = "1\t2";
        input[2] = "1\t2";
        input[3] = "\t2";
        input[4] = "3\t2";
        input[5] = "\t";
        input[6] = "5\t";
        input[7] = "7\t";
        input[8] = "7\t1";
        Util.createInputFile(cluster, "temp_file", input);
        pigServer.registerQuery("A = LOAD 'temp_file';");
        pigServer.registerQuery("B = LOAD 'temp_file';");
        DataBag dbmrj = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        {
            pigServer.registerQuery("C = join A by $0, B by $0 using 'merge';");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbmrj.add(iter.next());
            }
        }
        {
            pigServer.registerQuery("C = join A by $0, B by $0;");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }
        Util.deleteFile(cluster, "temp_file");
        Assert.assertEquals(dbmrj.size(),dbshj.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbmrj, dbshj));
    }

    @Test
    public void testMergeJoinWithMRBoundaryLater() throws IOException{
        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' as (x:int,y:int);");
        pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "' as (x:int,y:int);");
        DataBag dbMergeJoin = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        {
            pigServer.registerQuery("C = join A by $0, B by $0 using 'merge';");
            pigServer.registerQuery("D = group C by $0;");
            pigServer.registerQuery("E = filter D by $0 > 0;");
            Iterator<Tuple> iter = pigServer.openIterator("E");

            while(iter.hasNext()) {
                dbMergeJoin.add(iter.next());
            }
        }
        {
            pigServer.registerQuery("C = join A by $0, B by $0;");
            pigServer.registerQuery("D = group C by $0 ;");
            pigServer.registerQuery("E = filter D by $0 > 0;");
            Iterator<Tuple> iter = pigServer.openIterator("E");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }
        Assert.assertEquals(dbMergeJoin.size(), dbshj.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbMergeJoin, dbshj));
    }

    @Test
    public void testMergeJoinWithUDF() throws Exception{
        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' as (x:int,y:double);");
        pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "' as (x:int,y:double);");
        pigServer.registerQuery("A = FOREACH A GENERATE x, ABS(y) AS y;");

        DataBag dbMergeJoin = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        {
            pigServer.registerQuery("C = JOIN A BY x, B BY x USING 'merge';");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbMergeJoin.add(iter.next());
            }
        }
        {
            pigServer.registerQuery("C = JOIN A BY x, B BY x;");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }
        Assert.assertEquals(dbMergeJoin.size(), dbshj.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbMergeJoin, dbshj));
    }

    @Test
    public void testMergeJoin3Way() throws IOException{
        try {
            pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' as (id, name, n);");
            pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "' as (id, name);");
            pigServer.registerQuery("C = LOAD '" + INPUT_FILE + "' as (id, name);");
            pigServer.registerQuery("D = join A by id, B by id, C by id using 'merge';");
        }catch(Exception e) {
            PigException pe = LogUtils.getPigException(e);
            Assert.assertTrue( pe.getMessage().contains( "Merge join can only be applied for 2-way joins" ) );
            return;
        }
        Assert.fail("Should throw exception, do not support 3 way join");
    }       

    @Test
    public void testMergeJoinWithOrderAndSplit() throws Exception{
        String query = "A = LOAD '" + INPUT_FILE + "' as (id, name, n);\n" +
                "B = LOAD '" + INPUT_FILE + "' as (id, name);\n" +
                "C = ORDER A by $0 parallel 5;\n" +
                "D = join A by id, C by id using 'merge';\n" +
                "store D into '/dev/null/1';";
        // verify that this passes parsing sanity checks.
        Util.buildPp(pigServer, query);
    }

    @Test
    public void testMergeJoinWithOrder() throws Exception{
        String query = "A = LOAD '" + INPUT_FILE + "' as (id, name, n);\n" +
                "B = LOAD '" + INPUT_FILE + "' as (id, name);\n" +
                "C = ORDER B by $0 parallel 5;\n" +
                "D = join A by id, C by id using 'merge';\n" +
                "store D into '/dev/null/1';";
        // verify that this passes parsing sanity checks.
        Util.buildPp(pigServer, query);
    }

    @Test
    public void testMergeJoinFailure2() throws IOException{
        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' as (id, name, n);");
        pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "' as (id, name);");
        pigServer.registerQuery("C = GROUP B by id;");
        pigServer.registerQuery("D = join A by id, C by $0 using 'merge';");
        try {
            pigServer.openIterator("D");
        }catch(Exception e) {
            PigException pe = LogUtils.getPigException(e);
            Assert.assertEquals(1103,pe.getErrorCode());
            return;
        }
        Assert.fail("Should fail to compile");
    }       

    @Test
    public void testEmptyRightFile() throws IOException{
        DataBag dbmrj = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        Util.createInputFile(cluster, "temp_file", new String[]{});
        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "';");
        pigServer.registerQuery("B = LOAD 'temp_file';");
        pigServer.registerQuery("C = join A by $0, B by $0 using 'merge';");
        pigServer.openIterator("C");
        {
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) 
                dbmrj.add(iter.next());
        }
        {
            pigServer.registerQuery("C = join A by $0, B by $0;");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }
        Assert.assertEquals(dbmrj.size(), dbshj.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbmrj, dbshj));
        Util.deleteFile(cluster, "temp_file");
    }       

    @Test
    public void testParallelism() throws Exception{
        String query = "A = LOAD '" + INPUT_FILE + "';" +
                       "B = LOAD '" + INPUT_FILE + "';" +
                       "C = join A by $0, B by $0 using 'merge' parallel 50;" + 
                       "store C into 'out';";
	PigContext pc = new PigContext(ExecType.MAPREDUCE,cluster.getProperties());
    pc.connect();
	MROperPlan mro = Util.buildMRPlan(Util.buildPp(pigServer, query),pc);
        Assert.assertEquals(1,mro.getRoots().get(0).getRequestedParallelism());
    }

    @Test
    public void testIndexer() throws IOException{
        Util.createInputFile(cluster, "temp_file1", new String[]{1+""});
        Util.createInputFile(cluster, "temp_file2", new String[]{2+""});
        Util.createInputFile(cluster, "temp_file3", new String[]{10+""});
        pigServer.registerQuery("A = LOAD 'temp_file*' as (a:int);");
        pigServer.registerQuery("B = LOAD 'temp_file*' as (a:int);");
        DataBag dbmrj = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        {
            pigServer.registerQuery("C = join A by $0, B by $0 using 'merge';");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbmrj.add(iter.next());
            }
        }
        {
            pigServer.registerQuery("C = join A by $0, B by $0;");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }
        Util.deleteFile(cluster, "temp_file1");
        Util.deleteFile(cluster, "temp_file2");
        Util.deleteFile(cluster, "temp_file3");
        Assert.assertEquals(dbmrj.size(),dbshj.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbmrj, dbshj));
    }
    
    @Test
    public void testExpression() throws IOException{
        Util.createInputFile(cluster, "temp_file1", new String[]{"8", "9"});
        Util.createInputFile(cluster, "temp_file2", new String[]{"10", "11"});
        Util.createInputFile(cluster, "temp_file3", new String[]{"20"});
        Util.createInputFile(cluster, "leftinput", new String[] { "9", "11"});
        pigServer.registerQuery("A = LOAD 'leftinput' as (a:int);");
        pigServer.registerQuery("B = LOAD 'temp_file*' as (a:int);");
        DataBag dbmrj = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        {
            pigServer.registerQuery("C = join A by $0 + 10, B by $0 + 10 using 'merge';");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbmrj.add(iter.next());
            }
        }
        {
            pigServer.registerQuery("C = join A by $0 + 10, B by $0 + 10;");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }
        Util.deleteFile(cluster, "temp_file1");
        Util.deleteFile(cluster, "temp_file2");
        Util.deleteFile(cluster, "temp_file3");
        Util.deleteFile(cluster, "leftinput");
        Assert.assertEquals(dbmrj.size(),dbshj.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbmrj, dbshj));
    }
    
    @Test
    public void testExpressionFail() throws IOException{
        pigServer.registerQuery("A = LOAD 'leftinput' as (a:int);");
        pigServer.registerQuery("B = LOAD 'temp_file*' using " +
                DummyIndexableLoader.class.getName() + "() as (a:int);");
        boolean exceptionThrown = false;
        try {
            pigServer.registerQuery("C = join A by $0 + 10, B by $0 + 10 using 'merge';");
            Iterator<Tuple> iter = pigServer.openIterator("C");

        }catch (Exception e) {
            e.printStackTrace();
            PigException pe = LogUtils.getPigException(e);
            Assert.assertEquals(1106, pe.getErrorCode());
            exceptionThrown = true;
        }
        Assert.assertEquals(true, exceptionThrown);
    }
    
    @Test
    public void testMergeJoinSch1() throws IOException{
        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' as (x:int,y:int);");
        pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "' as (x:int,y:int);");
        Schema mjSch = null, shjSch = null;
        pigServer.registerQuery("C = join A by $0, B by $0 using 'merge';");
        mjSch = pigServer.dumpSchema("C");
        pigServer.registerQuery("C = join A by $0, B by $0;");
        shjSch = pigServer.dumpSchema("C");
        Assert.assertEquals(true, shjSch.equals(mjSch));
    }

    @Test
    public void testMergeJoinSch2() throws IOException{
        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "';");
        pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "';");
        Schema mjSch = null, shjSch = null;
        pigServer.registerQuery("C = join A by ($0,$1), B by ($0,$1) using 'merge';");
        mjSch = pigServer.dumpSchema("C");
        pigServer.registerQuery("C = join A by ($0,$1), B by ($0,$1);");
        shjSch = pigServer.dumpSchema("C");
        Assert.assertTrue(shjSch == null);
    }
    
    @Test
    public void testMergeJoinWithCommaSeparatedFilePaths() throws IOException{

        pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "';");
        pigServer.registerQuery("B = LOAD 'temp_file,righinput_file' using " +
                DummyIndexableLoader.class.getName() + "();");

        pigServer.registerQuery("C = join A by $0, B by $0 using 'merge';");
            
        Iterator<Tuple> iter = pigServer.openIterator("C");
        Assert.assertFalse(iter.hasNext());
    }
    
    @Test
    public void testMergeJoinEmptyIndex() throws IOException{
        DataBag dbMergeJoin = BagFactory.getInstance().newDefaultBag(), dbshj = BagFactory.getInstance().newDefaultBag();
        pigServer.registerQuery("A = LOAD '" + INPUT_FILE2 + "';");
        pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "';");
        
        {
            pigServer.registerQuery("C = join A by $0, B by $0 using 'merge';");
            Iterator<Tuple> iter = pigServer.openIterator("C");
            while(iter.hasNext()) {
                dbMergeJoin.add(iter.next());
            }
        }
        
        {
            pigServer.registerQuery("C = join A by $0, B by $0;");
            Iterator<Tuple> iter = pigServer.openIterator("C");

            while(iter.hasNext()) {
                dbshj.add(iter.next());
            }
        }

        Assert.assertEquals(dbMergeJoin.size(), dbshj.size());
        Assert.assertEquals(true, TestHelper.compareBags(dbMergeJoin, dbshj));        
    }
    
    /**
     * A dummy loader which implements {@link IndexableLoadFunc} to test
     * that expressions are not allowed as merge join keys when the right input's
     * loader implements {@link IndexableLoadFunc}
     */
    public static class DummyIndexableLoader extends LoadFunc implements IndexableLoadFunc{

        /**
         * 
         */
        public DummyIndexableLoader() {
        }
 
        @Override
        public void close() throws IOException {
 
        }

        @Override
        public void seekNear(Tuple keys) throws IOException {
 
        }

        @Override
        public Tuple getNext() throws IOException {
            return null;
        }

        @Override
        public void initialize(Configuration conf) throws IOException {
        }

        @Override
        public InputFormat getInputFormat() throws IOException {
            return null;
        }

        @Override
        public LoadCaster getLoadCaster() throws IOException {            
            return null;
        }

        @Override
        public void prepareToRead(RecordReader reader, PigSplit split)
                throws IOException {

        }

        @Override
        public void setLocation(String location, Job job) throws IOException {

        }
        
    }
}