/*
 * 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.backend.hadoop.executionengine.physicalLayer.relationalOperators;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.Job;
import org.apache.pig.FuncSpec;
import org.apache.pig.IndexableLoadFunc;
import org.apache.pig.LoadFunc;
import org.apache.pig.PigException;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
import org.apache.pig.data.DataType;
import org.apache.pig.data.SchemaTuple;
import org.apache.pig.data.SchemaTupleBackend;
import org.apache.pig.data.SchemaTupleClassGenerator.GenContext;
import org.apache.pig.data.SchemaTupleFactory;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleMaker;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.builtin.DefaultIndexableLoader;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.impl.plan.NodeIdGenerator;
import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.impl.plan.PlanException;
import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.impl.util.MultiMap;
import org.apache.pig.newplan.logical.relational.LOJoin;
import org.apache.pig.newplan.logical.relational.LOJoin.JOINTYPE;

/** This operator implements merge join algorithm to do map side joins.
 *  Currently, only two-way joins are supported. One input of join is identified as left
 *  and other is identified as right. Left input tuples are the input records in map.
 *  Right tuples are read from HDFS by opening right stream.
 *
 *    This join doesn't support outer join.
 *    Data is assumed to be sorted in ascending order. It will fail if data is sorted in descending order.
 */

public class POMergeJoin extends PhysicalOperator {
    private static final Log log = LogFactory.getLog(POMergeJoin.class);

    private static final long serialVersionUID = 1L;

    private static final String keyOrderReminder = "Remember that you should " +
        "not change the order of keys before a merge join in a FOREACH or " +
        "manipulate join keys in a UDF in a way that would change the sort " +
        "order. UDFs in a FOREACH are allowed as long as they do not change" +
        "the join key values in a way that would change the sort order.\n";

    // flag to indicate when getNext() is called first.
    private boolean firstTime = true;

    //The Local Rearrange operators modeling the join key
    private POLocalRearrange[] LRs;

    protected transient LoadFunc rightLoader;
    private OperatorKey opKey;

    private transient Object prevLeftKey;

    private transient Result prevLeftInp;

    private transient Object prevRightKey = null;

    private transient Result prevRightInp;

    //boolean denoting whether we are generating joined tuples in this getNext() call or do we need to read in more data.
    private transient boolean doingJoin;

    protected FuncSpec rightLoaderFuncSpec;

    private String rightInputFileName;

    private String indexFile;

    // Buffer to hold accumulated left tuples.
    private transient TuplesToSchemaTupleList leftTuples;

    private MultiMap<PhysicalOperator, PhysicalPlan> inpPlans;

    private PhysicalOperator rightPipelineLeaf;

    private PhysicalOperator rightPipelineRoot;

    private boolean noInnerPlanOnRightSide;

    private transient Object curJoinKey;

    private transient Tuple curJoiningRightTup;

    private int counter; // # of tuples on left side with same key.

    private transient int leftTupSize;

    private transient int rightTupSize;

    private static int ARRAY_LIST_SIZE = 1024;

    private LOJoin.JOINTYPE joinType;

    private String signature;

    private byte endOfRecordMark = POStatus.STATUS_NULL;

    // Only for Spark
    // If current operator reaches at its end, flag endOfInput is set as true.
    // The old flag parentPlan.endOfAllInput doesn't work in spark mode, because it is shared
    // between operators in the same plan, so it could be set by preceding operators even
    // current operator does not reach at its end. (see PIG-4876)
    private transient boolean endOfInput = false;
    public boolean isEndOfInput() {
        return endOfInput;
    }
    public void setEndOfInput (boolean isEndOfInput) {
        endOfInput = isEndOfInput;
    }

    // Only for spark.
    // it means that current operator reaches at its end and the last left input was
    // added into 'leftTuples', ready for join.
    private transient boolean leftInputConsumedInSpark = false;

    // This serves as the default TupleFactory

    /**
     * These TupleFactories are used for more efficient Tuple generation. This should
     * decrease the amount of memory needed for a given map task to successfully perform
     * a merge join.
     */
    private transient TupleMaker mergedTupleMaker;
    private transient TupleMaker leftTupleMaker;

    private Schema leftInputSchema;
    private Schema mergedInputSchema;

    /**
     * @param k
     * @param rp
     * @param inp
     * @param inpPlans there can only be 2 inputs each being a List<PhysicalPlan>
     * Ex. join A by ($0,$1), B by ($1,$2);
     */
    public POMergeJoin(OperatorKey k, int rp, List<PhysicalOperator> inp, MultiMap<PhysicalOperator, PhysicalPlan> inpPlans,
            List<List<Byte>> keyTypes, LOJoin.JOINTYPE joinType, Schema leftInputSchema, Schema rightInputSchema, Schema mergedInputSchema) throws PlanException{

        super(k, rp, inp);
        this.opKey = k;
        this.doingJoin = false;
        this.inpPlans = inpPlans;
        LRs = new POLocalRearrange[2];
        this.createJoinPlans(inpPlans,keyTypes);
        this.indexFile = null;
        this.joinType = joinType;
        this.leftInputSchema = leftInputSchema;
        this.mergedInputSchema = mergedInputSchema;
    }

    public POMergeJoin(POMergeJoin copy) {
        super(copy);
        this.firstTime = copy.firstTime;
        this.LRs = copy.LRs;
        this.rightLoaderFuncSpec = copy.rightLoaderFuncSpec;
        this.rightInputFileName = copy.rightInputFileName;
        this.indexFile = copy.indexFile;
        this.inpPlans = copy.inpPlans;
        this.rightPipelineLeaf = copy.rightPipelineLeaf;
        this.rightPipelineRoot = copy.rightPipelineRoot;
        this.noInnerPlanOnRightSide = copy.noInnerPlanOnRightSide;
        this.counter = copy.counter;
        this.joinType = copy.joinType;
        this.signature = copy.signature;
        this.endOfRecordMark = copy.endOfRecordMark;
        this.leftInputSchema = copy.leftInputSchema;
        this.mergedInputSchema = copy.mergedInputSchema;
    }

    /**
     * Configures the Local Rearrange operators to get keys out of tuple.
     * @throws ExecException
     */
    private void createJoinPlans(MultiMap<PhysicalOperator, PhysicalPlan> inpPlans, List<List<Byte>> keyTypes) throws PlanException{

        int i=-1;
        for (PhysicalOperator inpPhyOp : inpPlans.keySet()) {
            ++i;
            POLocalRearrange lr = new POLocalRearrange(genKey());
            try {
                lr.setIndex(i);
            } catch (ExecException e) {
                throw new PlanException(e.getMessage(),e.getErrorCode(),e.getErrorSource(),e);
            }
            lr.setResultType(DataType.TUPLE);
            lr.setKeyType(keyTypes.get(i).size() > 1 ? DataType.TUPLE : keyTypes.get(i).get(0));
            lr.setPlans(inpPlans.get(inpPhyOp));
            LRs[i]= lr;
        }
    }

    /**
     * This is a helper method that sets up all of the TupleFactory members.
     */
    private void prepareTupleFactories() {
        if (leftInputSchema != null) {
            leftTupleMaker = SchemaTupleBackend.newSchemaTupleFactory(leftInputSchema, false, GenContext.MERGE_JOIN);
        }
        if (leftTupleMaker == null) {
            log.debug("No SchemaTupleFactory available for combined left merge join schema: " + leftInputSchema);
            leftTupleMaker = mTupleFactory;
        } else {
            log.debug("Using SchemaTupleFactory for left merge join schema: " + leftInputSchema);
        }

        if (mergedInputSchema != null) {
            mergedTupleMaker = SchemaTupleBackend.newSchemaTupleFactory(mergedInputSchema, false, GenContext.MERGE_JOIN);
        }
        if (mergedTupleMaker == null) {
            log.debug("No SchemaTupleFactory available for combined left/right merge join schema: " + mergedInputSchema);
            mergedTupleMaker = mTupleFactory;
        } else {
            log.debug("Using SchemaTupleFactory for left/right merge join schema: " + mergedInputSchema);
        }

    }

    /**
     * This provides a List to store Tuples in. The implementation of that list depends on whether
     * or not there is a TupleFactory available.
     * @return the list object to store Tuples in
     */
    private TuplesToSchemaTupleList newLeftTupleArray() {
        return new TuplesToSchemaTupleList(ARRAY_LIST_SIZE, leftTupleMaker);
    }

    /**
     * This is a class that extends ArrayList, making it easy to provide on the fly conversion
     * from Tuple to SchemaTuple. This is necessary because we are not getting SchemaTuples
     * from the source, though in the future that is what we would like to do.
     */
    public static class TuplesToSchemaTupleList extends ArrayList<Tuple> {
        private SchemaTupleFactory tf;

        public TuplesToSchemaTupleList(int ct, TupleMaker<?> tf) {
            super(ct);
            if (tf instanceof SchemaTupleFactory) {
                this.tf = (SchemaTupleFactory)tf;
            }
        }

        public static SchemaTuple<?> convert(Tuple t, SchemaTupleFactory tf) {
            if (t instanceof SchemaTuple<?>) {
                return (SchemaTuple<?>)t;
            }
            SchemaTuple<?> st = tf.newTuple();
            try {
                return st.set(t);
            } catch (ExecException e) {
                throw new RuntimeException("Unable to set SchemaTuple with schema ["
                        + st.getSchemaString() + "] with given Tuple in merge join.");
            }
        }

        @Override
        public boolean add(Tuple t) {
            if (tf != null) {
                t = convert(t, tf);
            }
            return super.add(t);
        }

        @Override
        public Tuple get(int i) {
            return super.get(i);
        }

        @Override
        public int size() {
            return super.size();
        }

    }

    @SuppressWarnings("unchecked")
    @Override
    public Result getNextTuple() throws ExecException {

        Object curLeftKey;
        Result curLeftInp;

        if(firstTime){
            prepareTupleFactories();
            leftTuples = newLeftTupleArray();

            // Do initial setup.
            curLeftInp = processInput();
            if(curLeftInp.returnStatus != POStatus.STATUS_OK)
                return curLeftInp;       // Return because we want to fetch next left tuple.

            curLeftKey = extractKeysFromTuple(curLeftInp, 0);
            if(null == curLeftKey) // We drop the tuples which have null keys.
                return new Result(endOfRecordMark, null);

            try {
                seekInRightStream(curLeftKey);
            } catch (IOException e) {
                throwProcessingException(true, e);
            } catch (ClassCastException e) {
                throwProcessingException(true, e);
            }
            leftTuples.add((Tuple)curLeftInp.result);
            firstTime = false;
            prevLeftKey = curLeftKey;
            return new Result(endOfRecordMark, null);
        }

        if(doingJoin){
            // We matched on keys. Time to do the join.

            if(counter > 0){    // We have left tuples to join with current right tuple.
                Tuple joiningLeftTup = leftTuples.get(--counter);
                leftTupSize = joiningLeftTup.size();
                Tuple joinedTup = mergedTupleMaker.newTuple(leftTupSize + rightTupSize);

                for(int i=0; i<leftTupSize; i++) {
                    joinedTup.set(i, joiningLeftTup.get(i));
                }

                for(int i=0; i < rightTupSize; i++) {
                    joinedTup.set(i+leftTupSize, curJoiningRightTup.get(i));
                }

                return new Result(POStatus.STATUS_OK, joinedTup);
            }
            // Join with current right input has ended. But bag of left tuples
            // may still join with next right tuple.

            doingJoin = false;

            while(true){
                Result rightInp = getNextRightInp();
                if(rightInp.returnStatus != POStatus.STATUS_OK){
                    prevRightInp = null;
                    return rightInp;
                }
                else{
                    Object rightKey = extractKeysFromTuple(rightInp, 1);
                    if(null == rightKey) // If we see tuple having null keys in stream, we drop them
                        continue;       // and fetch next tuple.

                    int cmpval = ((Comparable)rightKey).compareTo(curJoinKey);
                    if (cmpval == 0){
                        // Matched the very next right tuple.
                        curJoiningRightTup = (Tuple)rightInp.result;
                        rightTupSize = curJoiningRightTup.size();
                        counter = leftTuples.size();
                        doingJoin = true;
                        return this.getNextTuple();

                    }
                    else if(cmpval > 0){    // We got ahead on right side. Store currently read right tuple.
                        if(!(this.parentPlan.endOfAllInput|| leftInputConsumedInSpark)){
                            prevRightKey = rightKey;
                            prevRightInp = rightInp;
                            // There cant be any more join on this key.
                            leftTuples = newLeftTupleArray();

                            leftTuples.add((Tuple)prevLeftInp.result);
                            return new Result(endOfRecordMark, null);
                        }

                        else{           // This is end of all input and this is last join output.
                            // Right loader in this case wouldn't get a chance to close input stream. So, we close it ourself.
                            try {
                                ((IndexableLoadFunc)rightLoader).close();
                            } catch (IOException e) {
                                // Non-fatal error. We can continue.
                                log.error("Received exception while trying to close right side file: " + e.getMessage());
                            }
                            return new Result(POStatus.STATUS_EOP, null);
                        }
                    }
                    else{   // At this point right side can't be behind.
                        int errCode = 1102;
                        String errMsg = "Data is not sorted on right side. \n" +
                            keyOrderReminder +
                            "Last two tuples encountered were: \n"+
                        curJoiningRightTup+ "\n" + (Tuple)rightInp.result ;
                        throw new ExecException(errMsg,errCode);
                    }
                }
            }
        }

        curLeftInp = processInput();
        switch(curLeftInp.returnStatus){

        case POStatus.STATUS_OK:
            curLeftKey = extractKeysFromTuple(curLeftInp, 0);
            if(null == curLeftKey) // We drop the tuples which have null keys.
                return new Result(endOfRecordMark, null);

            int cmpVal = ((Comparable)curLeftKey).compareTo(prevLeftKey);
            if(cmpVal == 0){
                // Keep on accumulating.
                leftTuples.add((Tuple)curLeftInp.result);
                return new Result(endOfRecordMark, null);
            }
            else if(cmpVal > 0){ // Filled with left bag. Move on.
                curJoinKey = prevLeftKey;
                break;
            }
            else{   // Current key < Prev Key
                int errCode = 1102;
                String errMsg = "Data is not sorted on left side. \n" +
                            keyOrderReminder +
                            "Last two tuples encountered were: \n" +
                prevLeftKey+ "\n" + curLeftKey ;
                throw new ExecException(errMsg,errCode);
            }

        case POStatus.STATUS_EOP:
            if(this.parentPlan.endOfAllInput || isEndOfInput()){
                // We hit the end on left input.
                // Tuples in bag may still possibly join with right side.
                curJoinKey = prevLeftKey;
                curLeftKey = null;
                if (isEndOfInput()) {
                    leftInputConsumedInSpark = true;
                }
                break;
            }
            else    // Fetch next left input.
                return curLeftInp;

        default:    // If encountered with ERR / NULL on left side, we send it down.
            return curLeftInp;
        }

        if((null != prevRightKey)
                && !(this.parentPlan.endOfAllInput || leftInputConsumedInSpark)
                && ((Comparable)prevRightKey).compareTo(curLeftKey) >= 0){

            // This will happen when we accumulated inputs on left side and moved on, but are still behind the right side
            // In that case, throw away the tuples accumulated till now and add the one we read in this function call.
            leftTuples = newLeftTupleArray();
            leftTuples.add((Tuple)curLeftInp.result);
            prevLeftInp = curLeftInp;
            prevLeftKey = curLeftKey;
            return new Result(endOfRecordMark, null);
        }

        // Accumulated tuples with same key on left side.
        // But since we are reading ahead we still haven't checked the read ahead right tuple.
        // Accumulated left tuples may potentially join with that. So, lets check that first.

        if((null != prevRightKey) && prevRightKey.equals(prevLeftKey)){

            curJoiningRightTup = (Tuple)prevRightInp.result;
            counter = leftTuples.size();
            rightTupSize = curJoiningRightTup.size();
            doingJoin = true;
            prevLeftInp = curLeftInp;
            prevLeftKey = curLeftKey;
            return this.getNextTuple();
        }

        // We will get here only when curLeftKey > prevRightKey
        boolean slidingToNextRecord = false;
        while(true){
            // Start moving on right stream to find the tuple whose key is same as with current left bag key.
            Result rightInp;
            if (slidingToNextRecord) {
                rightInp = getNextRightInp();
                slidingToNextRecord = false;
            } else
                rightInp = getNextRightInp(prevLeftKey);

            if(rightInp.returnStatus != POStatus.STATUS_OK)
                return rightInp;

            Object extractedRightKey = extractKeysFromTuple(rightInp, 1);

            if(null == extractedRightKey) // If we see tuple having null keys in stream, we drop them
                continue;       // and fetch next tuple.

            Comparable rightKey = (Comparable)extractedRightKey;

            if( prevRightKey != null && rightKey.compareTo(prevRightKey) < 0){
                // Sanity check.
                int errCode = 1102;
                String errMsg = "Data is not sorted on right side. \n" +
                            keyOrderReminder +
                            "Last two tuples encountered were: \n"+
                prevRightKey+ "\n" + rightKey ;
                throw new ExecException(errMsg,errCode);
            }

            if (prevLeftKey != null && rightKey.compareTo(prevLeftKey) < 0) {    // still behind the left side, do nothing, fetch next right tuple.
                slidingToNextRecord = true;
                continue;
            }

            else if (prevLeftKey != null && rightKey.compareTo(prevLeftKey) == 0){  // Found matching tuple. Time to do join.

                curJoiningRightTup = (Tuple)rightInp.result;
                counter = leftTuples.size();
                rightTupSize = curJoiningRightTup.size();
                doingJoin = true;
                prevLeftInp = curLeftInp;
                prevLeftKey = curLeftKey;
                return this.getNextTuple();
            }

            else{    // We got ahead on right side. Store currently read right tuple.
                prevRightKey = rightKey;
                prevRightInp = rightInp;
                // Since we didn't find any matching right tuple we throw away the buffered left tuples and add the one read in this function call.
                leftTuples = newLeftTupleArray();
                leftTuples.add((Tuple)curLeftInp.result);
                prevLeftInp = curLeftInp;
                prevLeftKey = curLeftKey;
                if(this.parentPlan.endOfAllInput || leftInputConsumedInSpark){  // This is end of all input and this is last time we will read right input.
                    // Right loader in this case wouldn't get a chance to close input stream. So, we close it ourself.
                    try {
                        ((IndexableLoadFunc)rightLoader).close();
                    } catch (IOException e) {
                     // Non-fatal error. We can continue.
                        log.error("Received exception while trying to close right side file: " + e.getMessage());
                    }
                }
                return new Result(endOfRecordMark, null);
            }
        }
    }

    private void seekInRightStream(Object firstLeftKey) throws IOException {
        rightLoader = getRightLoader();

        // Pass signature of the loader to rightLoader
        // make a copy of the conf to use in calls to rightLoader.
        rightLoader.setUDFContextSignature(signature);
        Job job = new Job(new Configuration(PigMapReduce.sJobConfInternal.get()));
        rightLoader.setLocation(rightInputFileName, job);
        ((IndexableLoadFunc)rightLoader).initialize(job.getConfiguration());
        ((IndexableLoadFunc)rightLoader).seekNear(
                firstLeftKey instanceof Tuple ? (Tuple)firstLeftKey : mTupleFactory.newTuple(firstLeftKey));
    }

    /**
     * Instantiate right loader
     *
     * @return
     * @throws IOException
     * @throws ExecException
     */
    protected LoadFunc getRightLoader() throws ExecException, IOException {
        LoadFunc loader = (LoadFunc) PigContext.instantiateFuncFromSpec(rightLoaderFuncSpec);
        // check if hadoop distributed cache is used
        if (indexFile != null && loader instanceof DefaultIndexableLoader) {
            DefaultIndexableLoader defLoader = (DefaultIndexableLoader) loader;
            defLoader.setIndexFile(indexFile);
        }
        return loader;
    }

    private Result getNextRightInp(Object leftKey) throws ExecException{

        /*
         * Only call seekNear if the merge join is 'merge-sparse'.  DefaultIndexableLoader does not
         * support more than a single call to seekNear per split - so don't call seekNear.
         */
    	if (joinType == LOJoin.JOINTYPE.MERGESPARSE) {
    		try {
    			((IndexableLoadFunc)rightLoader).seekNear(leftKey instanceof Tuple ? (Tuple)leftKey : mTupleFactory.newTuple(leftKey));
    			prevRightKey = null;
    		} catch (IOException e) {
    			throwProcessingException(true, e);
    		}
    	}
    	return this.getNextRightInp();
    }

    private Result getNextRightInp() throws ExecException{

        try {
            if(noInnerPlanOnRightSide){
                Tuple t = rightLoader.getNext();
                if(t == null) { // no more data on right side
                    return new Result(POStatus.STATUS_EOP, null);
                } else {
                    return new Result(POStatus.STATUS_OK, t);
                }
            } else {
                Result res = rightPipelineLeaf.getNextTuple();
                rightPipelineLeaf.detachInput();
                switch(res.returnStatus){
                case POStatus.STATUS_OK:
                    return res;

                case POStatus.STATUS_EOP:
                    Tuple t = rightLoader.getNext();
                    if(t == null) { // no more data on right side
                        return new Result(POStatus.STATUS_EOP, null);
                    } else {
                        // run the tuple through the pipeline
                        rightPipelineRoot.attachInput(t);
                        return this.getNextRightInp();

                    }
                    default: // We don't deal with ERR/NULL. just pass them down
                        throwProcessingException(false, null);

                }
            }
        } catch (IOException e) {
            throwProcessingException(true, e);
        }
        // we should never get here!
        return new Result(POStatus.STATUS_ERR, null);
    }

    public void throwProcessingException (boolean withCauseException, Exception e) throws ExecException {
        int errCode = 2176;
        String errMsg = "Error processing right input during merge join";
        if(withCauseException) {
            throw new ExecException(errMsg, errCode, PigException.BUG, e);
        } else {
            throw new ExecException(errMsg, errCode, PigException.BUG);
        }
    }

    private Object extractKeysFromTuple(Result inp, int lrIdx) throws ExecException{

        //Separate Key & Value of input using corresponding LR operator
        POLocalRearrange lr = LRs[lrIdx];
        lr.attachInput((Tuple)inp.result);
        Result lrOut = lr.getNextTuple();
        lr.detachInput();
        if(lrOut.returnStatus!=POStatus.STATUS_OK){
            int errCode = 2167;
            String errMsg = "LocalRearrange used to extract keys from tuple isn't configured correctly";
            throw new ExecException(errMsg,errCode,PigException.BUG);
        }

        return ((Tuple) lrOut.result).get(1);
    }

    public void setupRightPipeline(PhysicalPlan rightPipeline) throws FrontendException{

        if(rightPipeline != null){
            if(rightPipeline.getLeaves().size() != 1 || rightPipeline.getRoots().size() != 1){
                int errCode = 2168;
                String errMsg = "Expected physical plan with exactly one root and one leaf.";
                throw new FrontendException(errMsg,errCode,PigException.BUG);
            }

            noInnerPlanOnRightSide = false;
            this.rightPipelineLeaf = rightPipeline.getLeaves().get(0);
            this.rightPipelineRoot = rightPipeline.getRoots().get(0);
            this.rightPipelineRoot.setInputs(null);
        }
        else
            noInnerPlanOnRightSide = true;
    }

    private void readObject(ObjectInputStream is) throws IOException, ClassNotFoundException, ExecException{

        is.defaultReadObject();
    }


    private OperatorKey genKey(){
        return new OperatorKey(opKey.scope,NodeIdGenerator.getGenerator().getNextNodeId(opKey.scope));
    }

    public void setRightLoaderFuncSpec(FuncSpec rightLoaderFuncSpec) {
        this.rightLoaderFuncSpec = rightLoaderFuncSpec;
    }

    public List<PhysicalPlan> getInnerPlansOf(int index) {
        return inpPlans.get(inputs.get(index));
    }

    @Override
    public void visit(PhyPlanVisitor v) throws VisitorException {
        v.visitMergeJoin(this);
    }

    @Override
    public String name() {
        String name = getAliasString() + "MergeJoin";
        if (joinType==LOJoin.JOINTYPE.MERGESPARSE)
            name+="(sparse)";
        name+="[" + DataType.findTypeName(resultType) + "]" + " - " + mKey.toString();
        return name;
    }

    @Override
    public boolean supportsMultipleInputs() {
        return true;
    }

    /* (non-Javadoc)
     * @see org.apache.pig.impl.plan.Operator#supportsMultipleOutputs()
     */
    @Override
    public boolean supportsMultipleOutputs() {
        return false;
    }

    /**
     * @param rightInputFileName the rightInputFileName to set
     */
    public void setRightInputFileName(String rightInputFileName) {
        this.rightInputFileName = rightInputFileName;
    }

    public String getSignature() {
        return signature;
    }

    public void setSignature(String signature) {
        this.signature = signature;
    }

    public void setIndexFile(String indexFile) {
        this.indexFile = indexFile;
    }

    public String getIndexFile() {
        return indexFile;
    }

    @Override
    public Tuple illustratorMarkup(Object in, Object out, int eqClassIndex) {
        return null;
    }

    public LOJoin.JOINTYPE getJoinType() {
        return joinType;
    }

    public POLocalRearrange[] getLRs() {
        return LRs;
    }

    @Override
    public POMergeJoin clone() throws CloneNotSupportedException {
        POMergeJoin clone = (POMergeJoin) super.clone();
        clone.LRs = new POLocalRearrange[this.LRs.length];
        for (int i = 0; i < this.LRs.length; i++) {
            clone.LRs[i] = this.LRs[i].clone();
        }
        clone.rightLoaderFuncSpec = this.rightLoaderFuncSpec.clone();
        clone.inpPlans = new MultiMap<PhysicalOperator, PhysicalPlan>();
        for (PhysicalOperator op : this.inpPlans.keySet()) {
            PhysicalOperator cloneOp = op.clone();
            for (PhysicalPlan phyPlan : this.inpPlans.get(op)) {
                clone.inpPlans.put(cloneOp, phyPlan.clone());
            }
        }
        clone.rightPipelineLeaf = this.rightPipelineLeaf.clone();
        clone.rightPipelineRoot = this.rightPipelineRoot.clone();
        clone.leftInputSchema = this.leftInputSchema.clone();
        clone.mergedInputSchema = this.mergedInputSchema.clone();
        return clone;
    }
}