/*
 * 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.utils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;

import org.apache.pig.FuncSpec;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.test.PORead;
import org.apache.pig.builtin.PigStorage;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.io.FileLocalizer;
import org.apache.pig.impl.io.FileSpec;
import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MapReduceOper;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.*;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.*;
import org.apache.pig.impl.plan.PlanException;
import org.apache.pig.impl.util.Utils;

public class GenPhyOp{
    static Random r = new Random();
    
    public static final byte GTE = 1;
    public static final byte GT = 2;
    public static final byte LTE = 3;
    public static final byte LT = 4;

    public static class PlansAndFlattens {
        public List<PhysicalPlan> plans;
        public List<Boolean> flattens;

        public PlansAndFlattens(List<PhysicalPlan> p, List<Boolean> f) {
            plans = p;
            flattens = f;
        }
    };
    
    public static PigContext pc;
    
    public static ConstantExpression exprConst() {
        ConstantExpression ret = new ConstantExpression(new OperatorKey("", r
                .nextLong()));
        return ret;
    }

    public static GreaterThanExpr compGreaterThanExpr() {
        GreaterThanExpr ret = new GreaterThanExpr(new OperatorKey("", r
                .nextLong()));
        return ret;
    }

    public static GreaterThanExpr compGreaterThanExpr(
            ExpressionOperator lhs,
            ExpressionOperator rhs,
            byte type) {
        GreaterThanExpr ret = new GreaterThanExpr(new OperatorKey("", r
                .nextLong()));
        ret.setLhs(lhs);
        ret.setRhs(rhs);
        ret.setOperandType(type);
        return ret;
    }

    public static POAnd compAndExpr(
            ExpressionOperator lhs,
            ExpressionOperator rhs) {
        POAnd ret = new POAnd(new OperatorKey("", r
                .nextLong()));
        ret.setLhs(lhs);
        ret.setRhs(rhs);
        ret.setOperandType(DataType.BOOLEAN);
        return ret;
    }

    public static POProject exprProject() {
        POProject ret = new POProject(new OperatorKey("", r.nextLong()));
        return ret;
    }

    public static POProject exprProject(int col) {
        POProject ret = new POProject(new OperatorKey("", r.nextLong()), 1, col);
        return ret;
    }

    public static GTOrEqualToExpr compGTOrEqualToExpr() {
        GTOrEqualToExpr ret = new GTOrEqualToExpr(new OperatorKey("", r
                .nextLong()));
        return ret;
    }

    public static EqualToExpr compEqualToExpr() {
        EqualToExpr ret = new EqualToExpr(new OperatorKey("", r.nextLong()));
        return ret;
    }

    public static EqualToExpr compEqualToExpr(
            ExpressionOperator lhs,
            ExpressionOperator rhs,
            byte type) {
        EqualToExpr ret = new EqualToExpr(new OperatorKey("", r
                .nextLong()));
        ret.setLhs(lhs);
        ret.setRhs(rhs);
        ret.setOperandType(type);
        return ret;
    }

    public static NotEqualToExpr compNotEqualToExpr() {
        NotEqualToExpr ret = new NotEqualToExpr(new OperatorKey("", r
                .nextLong()));
        return ret;
    }

    public static POIsNull compIsNullExpr() {
        POIsNull ret = new POIsNull(new OperatorKey("", r
                .nextLong()));
        return ret;
    }

    public static LessThanExpr compLessThanExpr() {
        LessThanExpr ret = new LessThanExpr(new OperatorKey("", r.nextLong()));
        return ret;
    }

    public static LTOrEqualToExpr compLTOrEqualToExpr() {
        LTOrEqualToExpr ret = new LTOrEqualToExpr(new OperatorKey("", r
                .nextLong()));
        return ret;
    }

    public static POLocalRearrange topLocalRearrangeOp() {
        POLocalRearrange ret = new POLocalRearrange(new OperatorKey("", r
                .nextLong()));
        List<PhysicalPlan> plans = new LinkedList<PhysicalPlan>();
        try {
            ret.setPlans(plans);
        } catch(PlanException pe) {
        }
        return ret;
    }
    
    public static POGlobalRearrange topGlobalRearrangeOp(){
        POGlobalRearrange ret = new POGlobalRearrange(new OperatorKey("", r
                .nextLong()));
        return ret;
    }
    
    public static POPackage topPackageOp(){
        POPackage ret = new POPackage(new OperatorKey("", r.nextLong()));
        return ret;
    }
    
    public static POForEach topForEachOp() {
        POForEach ret = new POForEach(new OperatorKey("", r
                .nextLong()));
        return ret;
    }

    public static POUnion topUnionOp() {
        POUnion ret = new POUnion(new OperatorKey("", r.nextLong()));
        return ret;
    }
    
    public static POPartialAgg topPOPartialAgg(){
        POPartialAgg partAgg =  new POPartialAgg(getOK());
        return partAgg;
    }
    
    /**
     * creates the PlansAndFlattens struct for 
     * generate grpCol, *.
     * 
     * @param grpCol - The column to be grouped on
     * @param sample - The sample tuple that is used to infer
     *                  result types and #projects for *
     * @return - The PlansAndFlattens struct which has the exprplan
     *              for generate grpCol, * set.
     * @throws ExecException
     */
    public static PlansAndFlattens topGenerateOpWithExPlan(
            int grpCol,
            Tuple sample) throws ExecException {
        POProject prj1 = new POProject(new OperatorKey("", r.nextLong()), -1, grpCol);
        prj1.setResultType(sample.getType(grpCol));
        prj1.setOverloaded(false);
        
        

        List<Boolean> toBeFlattened = new LinkedList<Boolean>();
        toBeFlattened.add(false);
        

        PhysicalPlan plan1 = new PhysicalPlan();
        plan1.add(prj1);
        
        List<PhysicalPlan> inputs = new LinkedList<PhysicalPlan>();
        inputs.add(plan1);
        
        POProject rest[] = new POProject[sample.size()];
        int i=-1;
        for (POProject project : rest) {
            project = new POProject(new OperatorKey("", r.nextLong()), -1, ++i);
            project.setResultType(sample.getType(i));
            project.setOverloaded(false);
            
            PhysicalPlan pl = new PhysicalPlan();
            pl.add(project);
            
            toBeFlattened.add(false);
            inputs.add(pl);
        }

        

        return new PlansAndFlattens(inputs, toBeFlattened);
    }
    
    /**
     * creates the PlansAndFlattens struct for 
     * generate grpCol, *.
     * 
     * @param grpCol - The column to be grouped on
     * @param sample - The sample tuple that is used to infer
     *                  result types and #projects for *
     * @return - The PlansAndFlattens struct which has the exprplan
     *              for generate grpCol, * set.
     * @throws ExecException
     * @throws PlanException 
     */
    public static PlansAndFlattens topGenerateOpWithExPlanLR(
            int grpCol,
            Tuple sample) throws ExecException, PlanException {
        POProject prj1 = new POProject(new OperatorKey("", r.nextLong()), -1, grpCol);
        prj1.setResultType(sample.getType(grpCol));
        prj1.setOverloaded(false);
        
        POCastDummy cst = new POCastDummy(new OperatorKey("",r.nextLong()));
        cst.setResultType(sample.getType(grpCol));

        List<Boolean> toBeFlattened = new LinkedList<Boolean>();
        toBeFlattened.add(false);
        

        PhysicalPlan plan1 = new PhysicalPlan();
        plan1.add(prj1);
        plan1.add(cst);
        plan1.connect(prj1, cst);
        
        List<PhysicalPlan> inputs = new LinkedList<PhysicalPlan>();
        inputs.add(plan1);
        
        POProject rest[] = new POProject[sample.size()];
        POCastDummy csts[] = new POCastDummy[sample.size()];
        int i=-1;
        for (POProject project : rest) {
            project = new POProject(new OperatorKey("", r.nextLong()), -1, ++i);
            project.setResultType(sample.getType(i));
            project.setOverloaded(false);
            
            csts[i] = new POCastDummy(new OperatorKey("",r.nextLong()));
            csts[i].setResultType(sample.getType(i));
            
            PhysicalPlan pl = new PhysicalPlan();
            pl.add(project);
            pl.add(csts[i]);
            pl.connect(project, csts[i]);
            
            toBeFlattened.add(false);
            inputs.add(pl);
        }

        

        return new PlansAndFlattens(inputs, toBeFlattened);
    }
    
    /**
     * creates the PlansAndFlattens struct for 
     * 'generate field'.
     * 
     * @param field - The column to be generated
     * @param sample - The sample tuple that is used to infer
     *                  result type
     * @return - The PlansAndFlattens struct which has the exprplan
     *              for 'generate field' set.
     * @throws ExecException
     */
    public static PlansAndFlattens topGenerateOpWithExPlanForFe(
            int field,
            Tuple sample) throws ExecException {
        POProject prj1 = new POProject(new OperatorKey("", r.nextLong()), -1, field);
        prj1.setResultType(sample.getType(field));
        prj1.setOverloaded(false);
        
        

        List<Boolean> toBeFlattened = new LinkedList<Boolean>();
        toBeFlattened.add(false);
        

        PhysicalPlan plan1 = new PhysicalPlan();
        plan1.add(prj1);
        
        List<PhysicalPlan> inputs = new LinkedList<PhysicalPlan>();
        inputs.add(plan1);
        
        return new PlansAndFlattens(inputs, toBeFlattened);
    }
    
    /**
     * creates the PlansAndFlattens struct for 
     * 'generate flatten(field)'.
     * 
     * @param field - The column to be generated
     * @param sample - The sample tuple that is used to infer
     *                  result type
     * @return - The PlansAndFlattens struct which has the exprplan
     *              for 'generate field' set.
     * @throws ExecException
     */
    public static PlansAndFlattens topGenerateOpWithExPlanForFeFlat(int field) throws ExecException {
        POProject prj1 = new POProject(new OperatorKey("", r.nextLong()), -1, field);
        prj1.setResultType(DataType.BAG);
        prj1.setOverloaded(false);

        List<Boolean> toBeFlattened = new LinkedList<Boolean>();
        toBeFlattened.add(true);

        PhysicalPlan plan1 = new PhysicalPlan();
        plan1.add(prj1);
        
        List<PhysicalPlan> inputs = new LinkedList<PhysicalPlan>();
        inputs.add(plan1);
        
        return new PlansAndFlattens(inputs, toBeFlattened);
    }
    
    /**
     * creates the PlansAndFlattens struct for 
     * 'generate field[0] field[1] ...'.
     *
     * @param fields - The columns to be generated
     * @param sample - The sample tuple that is used to infer
     *                  result type
     * @return - The PlansAndFlattens struct which has the exprplan
     *              for 'generate field[0] field[1]' set.
     * @throws ExecException
     * @throws PlanException 
     */
    public static PlansAndFlattens topGenerateOpWithExPlanForFe(
            int[] fields,
            Tuple sample) throws ExecException, PlanException {
        POProject[] prj = new POProject[fields.length];

        for(int i=0;i<prj.length;i++){
            prj[i] = new POProject(new OperatorKey("", r.nextLong()), -1, fields[i]);
            prj[i].setResultType(sample.getType(fields[i]));
            prj[i].setOverloaded(false);
        }
        
        
        POCastDummy[] cst = new POCastDummy[fields.length];

        List<Boolean> toBeFlattened = new LinkedList<Boolean>();
        for (POProject project : prj)
            toBeFlattened.add(false);
        
        

        List<PhysicalPlan> inputs = new LinkedList<PhysicalPlan>();
        PhysicalPlan[] plans = new PhysicalPlan[fields.length];
        for (int i=0;i<plans.length;i++) {
            plans[i] = new PhysicalPlan();
            plans[i].add(prj[i]);
            cst[i] = new POCastDummy(new OperatorKey("",r.nextLong()));
            cst[i].setResultType(sample.getType(fields[i]));
            plans[i].add(cst[i]);
            plans[i].connect(prj[i], cst[i]);
            inputs.add(plans[i]);
        }
        
        return new PlansAndFlattens(inputs, toBeFlattened);
    }
    
    /**
     * creates the PlansAndFlattens struct for 
     * 'generate field[0] field[1] ...'.
     * with the flatten list as specified
     * @param fields - The columns to be generated
     * @param toBeFlattened - The columns to be flattened
     * @param sample - The sample tuple that is used to infer
     *                  result type
     * @return - The PlansAndFlattens struct which has the exprplan
     *              for 'generate field[0] field[1]' set.
     * @throws ExecException
     * @throws PlanException 
     */
    public static PlansAndFlattens topGenerateOpWithExPlanForFe(
            int[] fields,
            Tuple sample,
            List<Boolean> toBeFlattened) throws ExecException, PlanException {
        POProject[] prj = new POProject[fields.length];

        for(int i=0;i<prj.length;i++){
            prj[i] = new POProject(new OperatorKey("", r.nextLong()), -1, fields[i]);
            prj[i].setResultType(sample.getType(fields[i]));
            prj[i].setOverloaded(false);
        }
        
        
        POCastDummy[] cst = new POCastDummy[fields.length];

        List<PhysicalPlan> inputs = new LinkedList<PhysicalPlan>();
        PhysicalPlan[] plans = new PhysicalPlan[fields.length];
        for (int i=0;i<plans.length;i++) {
            plans[i] = new PhysicalPlan();
            plans[i].add(prj[i]);
            cst[i] = new POCastDummy(new OperatorKey("",r.nextLong()));
            cst[i].setResultType(sample.getType(fields[i]));
            plans[i].add(cst[i]);
            plans[i].connect(prj[i], cst[i]);
            inputs.add(plans[i]);
        }
        
        return new PlansAndFlattens(inputs, toBeFlattened);
    }
    
    /**
     * creates the POLocalRearrange operator with the given index for
     * group by grpCol
     * @param index - The input index of this POLocalRearrange operator
     * @param grpCol - The column to be grouped on
     * @param sample - Sample tuple needed for topGenerateOpWithExPlan
     * @return - The POLocalRearrange operator
     * @throws ExecException
     */
    public static POLocalRearrange topLocalRearrangeOPWithPlan(int index, int grpCol, Tuple sample) throws ExecException, PlanException{
        POLocalRearrange lr = topLocalRearrangeOPWithPlanPlain(index, grpCol, sample);
        List<PhysicalPlan> plans = lr.getPlans(); 
        PhysicalPlan ep = plans.get(0);
        POCastDummy cst = new POCastDummy(new OperatorKey("", r.nextLong()));
        cst.setResultType(sample.getType(grpCol));
        ep.addAsLeaf(cst);
        lr.setPlans(plans);
        return lr;
    }
    
    /**
     * creates the POLocalRearrange operator with the given index for
     * group by grpCol
     * @param index - The input index of this POLocalRearrange operator
     * @param grpCol - The column to be grouped on
     * @param sample - Sample tuple needed for topGenerateOpWithExPlan
     * @return - The POLocalRearrange operator
     * @throws ExecException
     */
    public static POLocalRearrange topLocalRearrangeOPWithPlanPlain(int index, int grpCol, Tuple sample) throws ExecException, PlanException{
        POProject prj1 = new POProject(new OperatorKey("", r.nextLong()), -1, grpCol);
        prj1.setResultType(sample.getType(grpCol));
        prj1.setOverloaded(false);

        PhysicalPlan plan1 = new PhysicalPlan();
        plan1.add(prj1);
        
        List<PhysicalPlan> plans = new ArrayList<PhysicalPlan>();
        plans.add(plan1);
        POLocalRearrange ret = topLocalRearrangeOp();
        ret.setPlans(plans);
        ret.setIndex(index);
        ret.setResultType(DataType.TUPLE);
        ret.setKeyType(sample.getType(grpCol));
        return ret;
    }
    
    /**
     * creates the POForEach operator for
     * foreach A generate field
     * @param field - The column to be generated
     * @param sample - Sample tuple needed for topGenerateOpWithExPlanForFe
     * @return - The POForEach operator
     * @throws ExecException
     */
    public static POForEach topForEachOPWithPlan(int field, Tuple sample) throws ExecException, PlanException{
        PlansAndFlattens pf = topGenerateOpWithExPlanForFe(field, sample);
        
        POForEach ret = topForEachOp();
        ret.setInputPlans(pf.plans);
        ret.setToBeFlattened(pf.flattens);
        ret.setResultType(DataType.TUPLE);
        return ret;
    }

    /**
     * creates the POForEach operator for
     * foreach A generate field[0] field[1]
     * @param fields - The columns to be generated
     * @param sample - Sample tuple needed for topGenerateOpWithExPlanForFe
     * @return - The POForEach operator
     * @throws ExecException
     */
    public static POForEach topForEachOPWithPlan(int[] fields, Tuple sample) throws ExecException, PlanException{
        PlansAndFlattens pf = topGenerateOpWithExPlanForFe(fields, sample);
        
        POForEach ret = topForEachOp();
        ret.setInputPlans(pf.plans);
        ret.setToBeFlattened(pf.flattens);
        ret.setResultType(DataType.TUPLE);
        return ret;
    }
    
    /**
     * creates the POForEach operator for
     * foreach A generate field[0] field[1]
     * @param fields - The columns to be generated
     * @param sample - Sample tuple needed for topGenerateOpWithExPlanForFe
     * @return - The POForEach operator
     * @throws ExecException
     */
    public static POForEach topForEachOPWithPlan(
            int[] fields,
            Tuple sample,
            List<Boolean> toBeFlattened) throws ExecException, PlanException{
        PlansAndFlattens pf =
            topGenerateOpWithExPlanForFe(fields, sample, toBeFlattened);
        
        POForEach ret = topForEachOp();
        ret.setInputPlans(pf.plans);
        ret.setToBeFlattened(toBeFlattened);
        ret.setResultType(DataType.TUPLE);
        return ret;
    }
    
    /**
     * creates the POForEach operator for
     * foreach A generate flatten(field)
     * @param fields - The columns to be generated
     * @param sample - Sample tuple needed for topGenerateOpWithExPlanForFe
     * @return - The POForEach operator
     * @throws ExecException
     */
    public static POForEach topForEachOPWithPlan(int field) throws ExecException, PlanException{
        PlansAndFlattens pf = topGenerateOpWithExPlanForFeFlat(field);
        
        POForEach ret = topForEachOp();
        ret.setInputPlans(pf.plans);
        ret.setToBeFlattened(pf.flattens);
        ret.setResultType(DataType.TUPLE);
        return ret;
    }

    
    public static POLoad topLoadOp() {
        POLoad ret = new POLoad(new OperatorKey("", r.nextLong()));
        ret.setPc(pc);
        return ret;
    }

    public static POFilter topFilterOp() {
        POFilter ret = new POFilter(new OperatorKey("", r.nextLong()));
        ConstantExpression ex = GenPhyOp.exprConst();
        ex.setValue(new Boolean(true));
        PhysicalPlan pp = new PhysicalPlan();
        pp.add(ex);
        ret.setPlan(pp);
        return ret;
    }
    
    public static POFilter connectedFilterOp(PhysicalOperator input) {
        List<PhysicalOperator> ops = new ArrayList<PhysicalOperator>(1);
        ops.add(input);
        POFilter ret = new POFilter(new OperatorKey("", r.nextLong()), ops);
        return ret;
    }
    
    public static POLimit topLimitOp() {
        POLimit ret = new POLimit(new OperatorKey("", r.nextLong()));
        return ret;
    }

    public static POFilter topFilterOpWithExPlan(int lhsVal, int rhsVal)
            throws ExecException, PlanException {
        POFilter ret = new POFilter(new OperatorKey("", r.nextLong()));

        ConstantExpression ce1 = GenPhyOp.exprConst();
        ce1.setValue(lhsVal);

        ConstantExpression ce2 = GenPhyOp.exprConst();
        ce2.setValue(rhsVal);

        GreaterThanExpr gr = GenPhyOp.compGreaterThanExpr();
        gr.setLhs(ce1);
        gr.setRhs(ce2);
        gr.setOperandType(DataType.INTEGER);

        PhysicalPlan ep = new PhysicalPlan();
        ep.add(ce1);
        ep.add(ce2);
        ep.add(gr);

        ep.connect(ce1, gr);
        ep.connect(ce2, gr);

        ret.setPlan(ep);

        return ret;
    }

    public static POFilter topFilterOpWithProj(int col, int rhsVal)
            throws ExecException, PlanException {
        POFilter ret = new POFilter(new OperatorKey("", r.nextLong()));

        POProject proj = exprProject();
        proj.setResultType(DataType.INTEGER);
        proj.setColumn(col);
        proj.setOverloaded(false);

        ConstantExpression ce2 = GenPhyOp.exprConst();
        ce2.setValue(rhsVal);

        GreaterThanExpr gr = GenPhyOp.compGreaterThanExpr();
        gr.setLhs(proj);
        gr.setRhs(ce2);
        gr.setOperandType(DataType.INTEGER);

        PhysicalPlan ep = new PhysicalPlan();
        ep.add(proj);
        ep.add(ce2);
        ep.add(gr);

        ep.connect(proj, gr);
        ep.connect(ce2, gr);

        ret.setPlan(ep);

        return ret;
    }
    
    public static POFilter topFilterOpWithProj(int col, int rhsVal,
            byte CompType) throws ExecException, PlanException {
        POFilter ret = new POFilter(new OperatorKey("", r.nextLong()));

        POProject proj = exprProject();
        proj.setResultType(DataType.INTEGER);
        proj.setColumn(col);
        proj.setOverloaded(false);

        ConstantExpression ce2 = GenPhyOp.exprConst();
        ce2.setValue(rhsVal);

        BinaryComparisonOperator cop = null;
        switch (CompType) {
        case GenPhyOp.GTE:
            cop = GenPhyOp.compGTOrEqualToExpr();
            break;
        case GenPhyOp.GT:
            cop = GenPhyOp.compGreaterThanExpr();
            break;
        case GenPhyOp.LTE:
            cop = GenPhyOp.compLTOrEqualToExpr();
            break;
        case GenPhyOp.LT:
            cop = GenPhyOp.compLessThanExpr();
            break;
        }

        cop.setLhs(proj);
        cop.setRhs(ce2);
        cop.setOperandType(DataType.INTEGER);

        PhysicalPlan ep = new PhysicalPlan();
        ep.add(proj);
        ep.add(ce2);
        ep.add(cop);

        ep.connect(proj, cop);
        ep.connect(ce2, cop);

        ret.setPlan(ep);

        return ret;
    }
    
    public static POFilter topFilterOpWithProjWithCast(int col, int rhsVal, byte CompType)
            throws ExecException, PlanException {
        POFilter ret = new POFilter(new OperatorKey("", r.nextLong()));

        POProject proj = exprProject();
        proj.setResultType(DataType.INTEGER);
        proj.setColumn(col);
        proj.setOverloaded(false);

        ConstantExpression ce2 = GenPhyOp.exprConst();
        ce2.setValue(rhsVal);
        
        BinaryComparisonOperator cop = null;
        switch(CompType){
        case GenPhyOp.GTE:
            cop = GenPhyOp.compGTOrEqualToExpr();
            break;
        case GenPhyOp.GT:
            cop = GenPhyOp.compGreaterThanExpr();
            break;
        case GenPhyOp.LTE:
            cop = GenPhyOp.compLTOrEqualToExpr();
            break;
        case GenPhyOp.LT:
            cop = GenPhyOp.compLessThanExpr();
            break;
        }
        
        POCastDummy cst = new POCastDummy(new OperatorKey("",r.nextLong()));
        
        cop.setLhs(cst);
        cop.setRhs(ce2);
        cop.setOperandType(DataType.INTEGER);

        PhysicalPlan ep = new PhysicalPlan();
        ep.add(cst);
        ep.add(proj);
        ep.add(ce2);
        ep.add(cop);

        ep.connect(proj, cst);
        ep.connect(cst, cop);
        ep.connect(ce2, cop);

        ret.setPlan(ep);

        return ret;
    }
    
    public static PORead topReadOp(DataBag bag) {
        PORead ret = new PORead(new OperatorKey("", r.nextLong()), bag);
        return ret;
    }
    
    public static POStore dummyPigStorageOp() {
        POStore ret = new POStore(new OperatorKey("", r.nextLong()));
        ret.setSFile(new FileSpec("DummyFil", new FuncSpec(PigStorage.class.getName() + "()")));
        return ret;
    }

    public static POStore topStoreOp() {
        POStore ret = new POStore(new OperatorKey("", r.nextLong()));
        return ret;
    }

    public static void setR(Random r) {
        GenPhyOp.r = r;
    }
    
    public static MapReduceOper MROp(){
        MapReduceOper ret = new MapReduceOper(new OperatorKey("",r.nextLong()));
        return ret;
    }
    
    private static FileSpec getTempFileSpec() throws IOException {
        return new FileSpec(FileLocalizer.getTemporaryPath(pc).toString(),
                new FuncSpec(Utils.getTmpFileCompressorName(pc))
        );
    }
    
    public static POSplit topSplitOp() throws IOException{
        POSplit ret = new POSplit(new OperatorKey("",r.nextLong()));
        ret.setSplitStore(getTempFileSpec());
        return ret;
    }
    
    public static PhysicalPlan grpChain() throws ExecException, PlanException{
        PhysicalPlan grpChain = new PhysicalPlan();
        POLocalRearrange lr = GenPhyOp.topLocalRearrangeOp();
        POGlobalRearrange gr = GenPhyOp.topGlobalRearrangeOp();
        POPackage pk = GenPhyOp.topPackageOp();
        
        grpChain.add(lr);
        grpChain.add(gr);
        grpChain.add(pk);
        
        grpChain.connect(lr, gr);
        grpChain.connect(gr, pk);
        
        return grpChain;
    }
    
    public static PhysicalPlan loadedGrpChain() throws ExecException, PlanException{
        PhysicalPlan ret = new PhysicalPlan();
        POLoad ld = GenPhyOp.topLoadOp();
        POLocalRearrange lr = GenPhyOp.topLocalRearrangeOp();
        POGlobalRearrange gr = GenPhyOp.topGlobalRearrangeOp();
        POPackage pk = GenPhyOp.topPackageOp();
        
        ret.add(ld);
        ret.add(lr);
        ret.add(gr);
        ret.add(pk);
        
        ret.connect(ld, lr);
        ret.connect(lr, gr);
        ret.connect(gr, pk);
        
        return ret;
    }
    
    public static PhysicalPlan loadedFilter() throws ExecException, PlanException{
        PhysicalPlan ret = new PhysicalPlan();
        POLoad ld = GenPhyOp.topLoadOp();
        POFilter fl = GenPhyOp.topFilterOp();
        ret.add(ld);
        ret.add(fl);
        
        ret.connect(ld, fl);
        return ret;
    }
    
    public static POForEach topForEachOPWithUDF(List<String> clsName) throws PlanException{
        List<PhysicalPlan> ep4s = new ArrayList<PhysicalPlan>();
        List<Boolean> flattened3 = new ArrayList<Boolean>();
        for (String string : clsName) {
            PhysicalPlan ep4 = new PhysicalPlan();
            POProject prjStar4 = new POProject(new OperatorKey("", r.nextLong()));
            prjStar4.setResultType(DataType.TUPLE);
            prjStar4.setStar(true);
            ep4.add(prjStar4);
            
            List ufInps = new ArrayList();
            ufInps.add(prjStar4);
            POUserFunc uf = new POUserFunc(new OperatorKey("", r.nextLong()), -1, ufInps, new FuncSpec(string));
            ep4.add(uf);
            ep4.connect(prjStar4, uf);
            ep4s.add(ep4);
            flattened3.add(false);
        }
        
        POForEach fe3 = new POForEach(new OperatorKey("", r.nextLong()), 1,
            ep4s, flattened3);
        fe3.setResultType(DataType.TUPLE);
        return fe3;
    }
    
    public static PhysicalPlan arithPlan() throws PlanException{
        PhysicalPlan ep = new PhysicalPlan();
        ConstantExpression ce[] = new ConstantExpression[7];
        for(int i=0;i<ce.length;i++){
            ce[i] = GenPhyOp.exprConst();
            ce[i].setValue(i);
            ep.add(ce[i]);
        }
        
        Add ad = new Add(getOK());
        ep.add(ad);
        ep.connect(ce[0], ad);
        ep.connect(ce[1], ad);
        
        Divide div = new Divide(getOK());
        ep.add(div);
        ep.connect(ce[2], div);
        ep.connect(ce[3], div);
        
        Subtract sub = new Subtract(getOK());
        ep.add(sub);
        ep.connect(ad, sub);
        ep.connect(div, sub);
        
        Mod mod = new Mod(getOK());
        ep.add(mod);
        ep.connect(ce[4], mod);
        ep.connect(ce[5], mod);
        
        Multiply mul1 = new Multiply(getOK());
        ep.add(mul1);
        ep.connect(mod, mul1);
        ep.connect(ce[6], mul1);
        
        Multiply mul2 = new Multiply(getOK());
        ep.add(mul2);
        ep.connect(sub, mul2);
        ep.connect(mul1, mul2);
        
        return ep;
    }
    
    public static OperatorKey getOK(){
        return new OperatorKey("",r.nextLong());
    }
    
    public static void setPc(PigContext pc) {
        GenPhyOp.pc = pc;
    }
}