/*
 * 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.File;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Properties;
import java.util.Set;

import org.apache.pig.ExecType;
import org.apache.pig.FilterFunc;
import org.apache.pig.PigServer;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.PigContext;
import org.apache.pig.newplan.Operator;
import org.apache.pig.newplan.OperatorPlan;
import org.apache.pig.newplan.logical.optimizer.LogicalPlanOptimizer;
import org.apache.pig.newplan.logical.relational.LOCogroup;
import org.apache.pig.newplan.logical.relational.LOCross;
import org.apache.pig.newplan.logical.relational.LODistinct;
import org.apache.pig.newplan.logical.relational.LOFilter;
import org.apache.pig.newplan.logical.relational.LOForEach;
import org.apache.pig.newplan.logical.relational.LOJoin;
import org.apache.pig.newplan.logical.relational.LOLimit;
import org.apache.pig.newplan.logical.relational.LOLoad;
import org.apache.pig.newplan.logical.relational.LOSort;
import org.apache.pig.newplan.logical.relational.LOSplit;
import org.apache.pig.newplan.logical.relational.LOSplitOutput;
import org.apache.pig.newplan.logical.relational.LOStore;
import org.apache.pig.newplan.logical.relational.LOStream;
import org.apache.pig.newplan.logical.relational.LOUnion;
import org.apache.pig.newplan.logical.relational.LogicalPlan;
import org.apache.pig.newplan.logical.rules.LoadTypeCastInserter;
import org.apache.pig.newplan.logical.rules.PushUpFilter;
import org.apache.pig.newplan.optimizer.PlanOptimizer;
import org.apache.pig.newplan.optimizer.PlanTransformListener;
import org.apache.pig.newplan.optimizer.Rule;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;

/**
 * Test the logical optimizer.
 */
public class TestNewPlanPushUpFilter {
    PigContext pc = new PigContext(ExecType.LOCAL, new Properties());

    @BeforeClass
    public static void oneTimeSetup() throws Exception {
        new File("dummy").delete();
    }

    @Before
    public void tearDown() {
    }

    /**
     * A simple filter UDF for testing
     */
    static public class MyFilterFunc extends FilterFunc {

        @Override
        public Boolean exec(Tuple input) {
            return false;
        }
    }

    @Test
    // Empty plan, nothing to update
    public void testErrorEmptyInput() throws Exception {
        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( "" );

        Assert.assertTrue( !newLogicalPlan.getOperators().hasNext() );
    }

    @Test
    //Test to ensure that the right exception is thrown when the input list is empty
    public void testErrorNonFilterInput() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);store A into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get(0);
        Assert.assertTrue( op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue( op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue( op instanceof LOStore );
        Assert.assertTrue( newLogicalPlan.getSuccessors(op) == null );
    }

    @Test
    public void testFilterLoad() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = filter A by $1 < 18;" +
            "store B into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get(0);
        Assert.assertTrue( op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue( op instanceof LOStore );
    }

    @Test
    public void testFilterStreaming() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = stream A through `" + "ps -u" + "`;" +
            "C = filter B by $1 < 18;" +
            "D = STORE C into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue(  op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStream );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
    }

    @Test
    public void testFilterSort() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = order A by $1, $2;" +
            "C = filter B by $1 < 18;" +
            "D = STORE C into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue(  op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOSort );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    public void testFilterConstantConditionSort() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = order A by $1, $2;" +
            "C = filter B by 1 == 1;" +
            "D = STORE C into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue(  op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOSort );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    public void testFilterUDFSort() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = order A by $1, $2;" +
            "C = filter B by " + MyFilterFunc.class.getName() + "($1) ;" +
            "D = STORE C into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue(  op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOSort );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    public void testFilterDistinct() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = distinct A;" +
            "C = filter B by $1 < 18;" +
            "D = STORE C into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue(  op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LODistinct );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    public void testFilterConstantConditionDistinct() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = distinct A;" +
            "C = filter B by 1 == 1;" +
            "D = STORE C into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue(  op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LODistinct );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    public void testFilterUDFDistinct() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = distinct A;" +
            "C = filter B by " + MyFilterFunc.class.getName() + "($1) ;" +
            "D = STORE C into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue(  op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LODistinct );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    public void testFilterFilter() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = filter A by $0 != 'name';" +
            "C = filter B by $1 < 18;" +
            "D = STORE C into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue(  op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        Assert.assertTrue( ((LOFilter)op).getAlias().equals( "B" ) );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        Assert.assertTrue( ((LOFilter)op).getAlias().equals( "C" ) );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    public void testFilterSplitOutput() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "split A into B if $1 < 18, C if $1 >= 18;" +
            "C = filter B by $1 < 10;" +
            "D = STORE C into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue(  op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOSplit );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOSplitOutput );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    public void testFilterLimit() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = limit A 10;" +
            "C = filter B by $1 < 18;" +
            "D = STORE C into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue(  op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOLimit );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    public void testFilterUnion() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = union A, B;" +
            "D = filter C by $1 < 18;" +
            "E = STORE D into'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator loadA = null;
        Operator loadB = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )  {
            loadA = loads.get( 0 );
            loadB = loads.get( 1 );
        } else {
            loadA = loads.get( 1 );
            loadB = loads.get( 0 );
        }

        Operator foreachA = newLogicalPlan.getSuccessors(loadA).get( 0 );
        Assert.assertTrue(  foreachA instanceof LOForEach );
        Operator filterA = newLogicalPlan.getSuccessors(foreachA).get( 0 );
        Assert.assertTrue(  filterA instanceof LOFilter );
        Operator foreachB = newLogicalPlan.getSuccessors(loadB).get( 0 );
        Assert.assertTrue(  foreachB instanceof LOForEach );
        Operator filterB = newLogicalPlan.getSuccessors(foreachB).get( 0 );
        Assert.assertTrue(  filterB instanceof LOFilter );

        Operator unionA = newLogicalPlan.getSuccessors( filterA ).get( 0 );
        Assert.assertTrue(  unionA instanceof LOUnion );
        Operator unionB = newLogicalPlan.getSuccessors( filterB ).get( 0 );
        Assert.assertTrue(  unionB instanceof LOUnion );
        Assert.assertTrue(  unionB == unionA );

        Operator store = newLogicalPlan.getSuccessors(unionA).get( 0 );
        Assert.assertTrue(  store instanceof LOStore );
    }

    @Test
    public void testFilterConstantConditionUnion() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = union A, B;" +
            "D = filter C by 1 == 1;" +
            "E = STORE D into'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator loadA = null;
        Operator loadB = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )  {
            loadA = loads.get( 0 );
            loadB = loads.get( 1 );
        } else {
            loadA = loads.get( 1 );
            loadB = loads.get( 0 );
        }

        Operator foreachA = newLogicalPlan.getSuccessors(loadA).get( 0 );
        Assert.assertTrue(  foreachA instanceof LOForEach );
        Operator filterA = newLogicalPlan.getSuccessors(foreachA).get( 0 );
        Assert.assertTrue(  filterA instanceof LOFilter );
        Operator foreachB = newLogicalPlan.getSuccessors(loadB).get( 0 );
        Assert.assertTrue(  foreachB instanceof LOForEach );
        Operator filterB = newLogicalPlan.getSuccessors(foreachB).get( 0 );
        Assert.assertTrue(  filterB instanceof LOFilter );

        Operator unionA = newLogicalPlan.getSuccessors( filterA ).get( 0 );
        Assert.assertTrue(  unionA instanceof LOUnion );
        Operator unionB = newLogicalPlan.getSuccessors( filterB ).get( 0 );
        Assert.assertTrue(  unionB instanceof LOUnion );
        Assert.assertTrue(  unionB == unionA );

        Operator store = newLogicalPlan.getSuccessors(unionA).get( 0 );
        Assert.assertTrue(  store instanceof LOStore );
    }

    @Test
    public void testFilterUDFUnion() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = union A, B;" +
            "D = filter C by " + MyFilterFunc.class.getName() + "() ;" +
            "E = STORE D into'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator loadA = null;
        Operator loadB = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )  {
            loadA = loads.get( 0 );
            loadB = loads.get( 1 );
        } else {
            loadA = loads.get( 1 );
            loadB = loads.get( 0 );
        }

        Operator foreachA = newLogicalPlan.getSuccessors(loadA).get( 0 );
        Assert.assertTrue(  foreachA instanceof LOForEach );
        Operator filterA = newLogicalPlan.getSuccessors(foreachA).get( 0 );
        Assert.assertTrue(  filterA instanceof LOFilter );
        Operator foreachB = newLogicalPlan.getSuccessors(loadB).get( 0 );
        Assert.assertTrue(  foreachB instanceof LOForEach );
        Operator filterB = newLogicalPlan.getSuccessors(foreachB).get( 0 );
        Assert.assertTrue(  filterB instanceof LOFilter );

        Operator unionA = newLogicalPlan.getSuccessors( filterA ).get( 0 );
        Assert.assertTrue(  unionA instanceof LOUnion );
        Operator unionB = newLogicalPlan.getSuccessors( filterB ).get( 0 );
        Assert.assertTrue(  unionB instanceof LOUnion );
        Assert.assertTrue(  unionB == unionA );

        Operator store = newLogicalPlan.getSuccessors(unionA).get( 0 );
        Assert.assertTrue(  store instanceof LOStore );
    }

    @Test
    public void testFilterCross() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = cross A, B;" +
            "D = filter C by $5 < 18;" +
            "E = limit D 10;" +
            "F = STORE E into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator op = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "B" ) )
            op = loads.get( 0 );
        else
            op = loads.get( 1 );

        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOCross );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOLimit );
    }

    @Test
    public void testFilterCross1() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = cross A, B;" +
            "D = filter C by $1 < 18;" +
            "E = STORE D into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator op = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
            op = loads.get( 0 );
        else
            op = loads.get( 1 );

        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue( op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOCross );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }


    @Test
    public void testFilterCross2() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = cross A, B;" +
            "D = filter C by $1 < 18 and $5 < 18;" +
            "E = STORE D into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator loadA = null;
        Operator loadB = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) ) {
            loadA = loads.get( 0 );
            loadB = loads.get( 1 );
        } else {
            loadA = loads.get( 1 );
            loadB = loads.get( 0 );
        }

        Operator foreachA = newLogicalPlan.getSuccessors(loadA).get( 0 );
        Assert.assertTrue(  foreachA instanceof LOForEach );
        Operator op = newLogicalPlan.getSuccessors(foreachA).get( 0 );
        Assert.assertTrue(  op instanceof LOCross );
        Operator foreachB = newLogicalPlan.getSuccessors(loadB).get( 0 );
        Assert.assertTrue(  foreachB instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(foreachB).get( 0 );
        Assert.assertTrue(  op instanceof LOCross );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
    }

    @Test
    public void testFilterConstantConditionCross() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = cross A, B;" +
            "D = filter C by 1 == 1;" +
            "E = STORE D into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator loadA = null;
        Operator loadB = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) ) {
            loadA = loads.get( 0 );
            loadB = loads.get( 1 );
        } else {
            loadA = loads.get( 1 );
            loadB = loads.get( 0 );
        }
        Operator op = newLogicalPlan.getSuccessors(loadA).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(loadB).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOCross );
    }

    @Test
    public void testFilterUDFCross() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = cross A, B;" +
            "D = filter C by " + MyFilterFunc.class.getName() + "($0) ;" +
            "E = STORE D into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator op = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
            op = loads.get( 0 );
        else
            op = loads.get( 1 );

        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOCross );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    public void testFilterCogroup() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = cogroup A by $0, B by $0;" +
            "D = filter C by $0 < 'name';" +
            "E = STORE D into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator loadA = null;
        Operator loadB = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )  {
            loadA = loads.get( 0 );
            loadB = loads.get( 1 );
        } else {
            loadA = loads.get( 1 );
            loadB = loads.get( 0 );
        }

        Operator foreachA = newLogicalPlan.getSuccessors(loadA).get( 0 );
        Assert.assertTrue(  foreachA instanceof LOForEach );
        Operator filterA = newLogicalPlan.getSuccessors(foreachA).get( 0 );
        Assert.assertTrue(  filterA instanceof LOFilter );

        Operator foreachB = newLogicalPlan.getSuccessors(loadB).get( 0 );
        Assert.assertTrue(  foreachB instanceof LOForEach );
        Operator filterB = newLogicalPlan.getSuccessors(foreachB).get( 0 );
        Assert.assertTrue(  filterB instanceof LOFilter );

        Operator cogrpA = newLogicalPlan.getSuccessors( filterA ).get( 0 );
        Assert.assertTrue(  cogrpA instanceof LOCogroup );
        Operator cogrpB = newLogicalPlan.getSuccessors( filterB ).get( 0 );
        Assert.assertTrue(  cogrpB instanceof LOCogroup );
        Assert.assertTrue(  cogrpB == cogrpA );

        Operator store = newLogicalPlan.getSuccessors(cogrpA).get( 0 );
        Assert.assertTrue(  store instanceof LOStore );
    }

    @Test
    public void testFilterConstantConditionCogroup() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = cogroup A by $0, B by $0;" +
            "D = filter C by 1 == 1;" +
            "E = STORE D into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator loadA = null;
        Operator loadB = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )  {
            loadA = loads.get( 0 );
            loadB = loads.get( 1 );
        } else {
            loadA = loads.get( 1 );
            loadB = loads.get( 0 );
        }

        Operator foreachA = newLogicalPlan.getSuccessors(loadA).get( 0 );
        Assert.assertTrue(  foreachA instanceof LOForEach );
        Operator filterA = newLogicalPlan.getSuccessors(foreachA).get( 0 );
        Assert.assertTrue(  filterA instanceof LOFilter );

        Operator foreachB = newLogicalPlan.getSuccessors(loadB).get( 0 );
        Assert.assertTrue(  foreachB instanceof LOForEach );
        Operator filterB = newLogicalPlan.getSuccessors(foreachB).get( 0 );
        Assert.assertTrue(  filterB instanceof LOFilter );

        Operator cogrpA = newLogicalPlan.getSuccessors( filterA ).get( 0 );
        Assert.assertTrue(  cogrpA instanceof LOCogroup );
        Operator cogrpB = newLogicalPlan.getSuccessors( filterB ).get( 0 );
        Assert.assertTrue(  cogrpB instanceof LOCogroup );
        Assert.assertTrue(  cogrpB == cogrpA );

        Operator store = newLogicalPlan.getSuccessors(cogrpA).get( 0 );
        Assert.assertTrue(  store instanceof LOStore );
    }


    @Test
    public void testFilterUDFCogroup() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = cogroup A by $0, B by $0;" +
            "D = filter C by " + MyFilterFunc.class.getName() + "($1) ;" +
            "E = STORE D into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator loadA = null;
        Operator loadB = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )  {
            loadA = loads.get( 0 );
            loadB = loads.get( 1 );
        } else {
            loadA = loads.get( 1 );
            loadB = loads.get( 0 );
        }

        Operator foreachA = newLogicalPlan.getSuccessors(loadA).get( 0 );
        Assert.assertTrue(  foreachA instanceof LOForEach );
        Operator cogroupA = newLogicalPlan.getSuccessors(foreachA).get( 0 );
        Assert.assertTrue(  cogroupA instanceof LOCogroup );

        Operator foreachB = newLogicalPlan.getSuccessors(loadB).get( 0 );
        Assert.assertTrue(  foreachB instanceof LOForEach );
        Operator cogroupB = newLogicalPlan.getSuccessors(foreachB).get( 0 );
        Assert.assertTrue(  cogroupB instanceof LOCogroup );

        Operator filter = newLogicalPlan.getSuccessors( cogroupA ).get( 0 );
        Assert.assertTrue(  filter instanceof LOFilter );
        filter = newLogicalPlan.getSuccessors( cogroupB ).get( 0 );
        Assert.assertTrue(  cogroupB instanceof LOCogroup );
        Assert.assertTrue(  cogroupB == cogroupA );

        Operator store = newLogicalPlan.getSuccessors(filter).get( 0 );
        Assert.assertTrue(  store instanceof LOStore );
    }

    @Test
    public void testFilterCogroupOuter() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = cogroup A by $0, B by $0 outer;" +
            "D = filter C by $0 < 'name';" +
            "E = STORE D into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator loadA = null;
        Operator loadB = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )  {
            loadA = loads.get( 0 );
            loadB = loads.get( 1 );
        } else {
            loadA = loads.get( 1 );
            loadB = loads.get( 0 );
        }

        Operator foreachA = newLogicalPlan.getSuccessors(loadA).get( 0 );
        Assert.assertTrue(  foreachA instanceof LOForEach );
        Operator filterA = newLogicalPlan.getSuccessors(foreachA).get( 0 );
        Assert.assertTrue(  filterA instanceof LOFilter );

        Operator foreachB = newLogicalPlan.getSuccessors(loadB).get( 0 );
        Assert.assertTrue(  foreachB instanceof LOForEach );
        Operator filterB = newLogicalPlan.getSuccessors(foreachB).get( 0 );
        Assert.assertTrue(  filterB instanceof LOFilter );

        Operator cogrpA = newLogicalPlan.getSuccessors( filterA ).get( 0 );
        Assert.assertTrue(  cogrpA instanceof LOCogroup );
        Operator cogrpB = newLogicalPlan.getSuccessors( filterB ).get( 0 );
        Assert.assertTrue(  cogrpB instanceof LOCogroup );
        Assert.assertTrue(  cogrpB == cogrpA );

        Operator store = newLogicalPlan.getSuccessors(cogrpA).get( 0 );
        Assert.assertTrue(  store instanceof LOStore );
    }

    @Test
    public void testFilterConstantConditionCogroupOuter() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = cogroup A by $0, B by $0 outer;" +
            "D = filter C by 1 == 1;" +
            "E = STORE D into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator loadA = null;
        Operator loadB = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )  {
            loadA = loads.get( 0 );
            loadB = loads.get( 1 );
        } else {
            loadA = loads.get( 1 );
            loadB = loads.get( 0 );
        }

        Operator foreachA = newLogicalPlan.getSuccessors(loadA).get( 0 );
        Assert.assertTrue(  foreachA instanceof LOForEach );
        Operator filterA = newLogicalPlan.getSuccessors(foreachA).get( 0 );
        Assert.assertTrue(  filterA instanceof LOFilter );

        Operator foreachB = newLogicalPlan.getSuccessors(loadB).get( 0 );
        Assert.assertTrue(  foreachB instanceof LOForEach );
        Operator filterB = newLogicalPlan.getSuccessors(foreachB).get( 0 );
        Assert.assertTrue(  filterB instanceof LOFilter );

        Operator cogrpA = newLogicalPlan.getSuccessors( filterA ).get( 0 );
        Assert.assertTrue(  cogrpA instanceof LOCogroup );
        Operator cogrpB = newLogicalPlan.getSuccessors( filterB ).get( 0 );
        Assert.assertTrue(  cogrpB instanceof LOCogroup );
        Assert.assertTrue(  cogrpB == cogrpA );

        Operator store = newLogicalPlan.getSuccessors(cogrpA).get( 0 );
        Assert.assertTrue(  store instanceof LOStore );
    }

    @Test
    public void testFilterUDFCogroupOuter() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = cogroup A by $0, B by $0 outer;" +
            "D = filter C by " + MyFilterFunc.class.getName() + "() ;" +
            "E = STORE D into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator loadA = null;
        Operator loadB = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )  {
            loadA = loads.get( 0 );
            loadB = loads.get( 1 );
        } else {
            loadA = loads.get( 1 );
            loadB = loads.get( 0 );
        }

        Operator foreachA = newLogicalPlan.getSuccessors(loadA).get( 0 );
        Assert.assertTrue(  foreachA instanceof LOForEach );
        Operator cogroupA = newLogicalPlan.getSuccessors(foreachA).get( 0 );
        Assert.assertTrue(  cogroupA instanceof LOCogroup );

        Operator foreachB = newLogicalPlan.getSuccessors(loadB).get( 0 );
        Assert.assertTrue(  foreachB instanceof LOForEach );
        Operator cogroupB = newLogicalPlan.getSuccessors(foreachB).get( 0 );
        Assert.assertTrue(  cogroupB instanceof LOCogroup );

        Operator filter = newLogicalPlan.getSuccessors( cogroupA ).get( 0 );
        Assert.assertTrue(  filter instanceof LOFilter );
        filter = newLogicalPlan.getSuccessors( cogroupB ).get( 0 );
        Assert.assertTrue(  filter instanceof LOFilter );
        Assert.assertTrue(  cogroupB == cogroupA );

        Operator store = newLogicalPlan.getSuccessors(filter).get( 0 );
        Assert.assertTrue(  store instanceof LOStore );
    }

    @Test
    public void testFilterGroupBy() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = group A by $0;" +
            "C = filter B by $0 < 'name';" +
            "D = STORE C into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue(  op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOCogroup );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    public void testFilterConstantConditionGroupBy() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = group A by $0;" +
            "C = filter B by 1 == 1;" +
            "D = STORE C into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue(  op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOCogroup );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    public void testFilterUDFGroupBy() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = group A by $0;" +
            "C = filter B by " + MyFilterFunc.class.getName() + "($1) ;" +
            "D = STORE C into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue(  op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue( op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOCogroup );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    public void testFilterGroupByOuter() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = group A by $0 outer;" +
            "C = filter B by $0 < 'name';" +
            "D = STORE C into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue(  op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOCogroup );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    public void testFilterConstantConditionGroupByOuter() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = group A by $0 outer;" +
            "C = filter B by 1 == 1;" +
            "D = STORE C into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue(  op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOCogroup );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    public void testFilterUDFGroupByOuter() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = group A by $0 outer;" +
            "C = filter B by " + MyFilterFunc.class.getName() + "($1) ;" +
            "D = STORE C into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        Operator op = newLogicalPlan.getSources().get( 0 );
        Assert.assertTrue(  op instanceof LOLoad );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOCogroup );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    public void testFilterFRJoin() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = join A by $0, B by $0 using 'replicated';" +
            "D = filter C by $0 < 'name';" +
            "E = limit D 10;" +
            "F = STORE E into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator op = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
            op = loads.get( 0 );
        else
            op = loads.get( 1 );

        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOJoin );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOLimit );
    }

    @Test
    public void testFilterFRJoin1() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = join A by $0, B by $0 using 'replicated';" +
            "D = filter C by $4 < 'name';" +
            "E = limit D 10;" +
            "F = STORE E into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator op = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "B" ) )
            op = loads.get( 0 );
        else
            op = loads.get( 1 );

        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOJoin );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOLimit );
    }

    @Test
    // Constant filter condition, the filter will be pushed up to the first branch of join.
    public void testFilterConstantConditionFRJoin() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = join A by $0, B by $0 using 'replicated';" +
            "D = filter C by 1 == 1;" +
            "E = STORE D into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator op = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
            op = loads.get( 0 );
        else
            op = loads.get( 1 );

        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOJoin );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    // UDF takes on argument, so it's constant. As a result, filter will pushed up to the first branch of the join.
    public void testFilterUDFFRJoin() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = join A by $0, B by $0 using 'replicated';" +
            "D = filter C by " + MyFilterFunc.class.getName() + "();" +
            "E = STORE D into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator op = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
            op = loads.get( 0 );
        else
            op = loads.get( 1 );

        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOJoin );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    @Test
    // UDF takes all input, so filter connot be pushed up.
    public void testFilterUDFFRJoin1() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = join A by $0, B by $0 using 'replicated';" +
            "D = filter C by TupleSize(*) > 5;" +
            "E = STORE D into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator op = newLogicalPlan.getSuccessors( loads.get( 0 ) ).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOJoin );

        op = newLogicalPlan.getSuccessors( loads.get( 1 ) ).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOJoin );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
    }

    @Test
    public void testFilterInnerJoin() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = join A by $0, B by $0;" +
            "D = filter C by $0 < 'name';" +
            "E = limit D 10;" +
            "F = STORE E into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator op = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
            op = loads.get( 0 );
        else
            op = loads.get( 1 );

        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOJoin );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOLimit );
    }

    @Test
    public void testFilterInnerJoin1() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = join A by $0, B by $0;" +
            "D = filter C by $4 < 'name';" +
            "E = limit D 10;" +
            "F = STORE E into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator op = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "B" ) )
            op = loads.get( 0 );
        else
            op = loads.get( 1 );

        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOJoin );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOLimit );
    }

    @Test
    public void testFilterInnerJoin2() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = join A by $0, B by $0;" +
            "D = filter C by $0 < 'jonh' OR $1 > 50;" +
            "E = limit D 10;" +
            "F = STORE E into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator op = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
            op = loads.get( 0 );
        else
            op = loads.get( 1 );

        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOJoin );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOLimit );
    }

    @Test
    public void testFilterInnerJoinNegative() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = join A by $0, B by $0;" +
            "D = filter C by $4 < 'name' AND $0 == 'joe';" +
            "E = limit D 10;" +
            "F = STORE E into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );

        Operator op = newLogicalPlan.getSuccessors( loads.get( 0 ) ).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOJoin );
        op = newLogicalPlan.getSuccessors( loads.get( 1 ) ).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOJoin );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
    }

    @Test
    public void testFilterUDFInnerJoin() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name, age, preference);" +
            "C = join A by $0, B by $0;" +
            "D = filter C by " + MyFilterFunc.class.getName() + "() ;" +
            "E = STORE D into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan( query );

        List<Operator> loads = newLogicalPlan.getSources();
        Assert.assertTrue( loads.size() == 2 );
        Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
        Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
        Operator op = null;
        if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
            op = loads.get( 0 );
        else
            op = loads.get( 1 );

        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOForEach );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOJoin );
        op = newLogicalPlan.getSuccessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOStore );
    }

    // See PIG-1289
    @Test
    public void testOutJoin() throws Exception {
        String query = "A = load 'myfile' as (name, age, gpa);" +
            "B = load 'anotherfile' as (name);" +
            "C = join A by name LEFT OUTER, B by name;" +
            "D = filter C by B::name is null;" +
            "store D into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan(query);

        Operator op = newLogicalPlan.getSinks().get(0);
        Assert.assertTrue( op instanceof LOStore );
        op = newLogicalPlan.getPredecessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
    }

    // See PIG-1507
    @Test
    public void testFullOutJoin() throws Exception {
        String query = "A = load 'myfile' as (d1:int);" +
            "B = load 'anotherfile' as (d2:int);" +
            "c = join A by d1 full outer, B by d2;" +
            "d = filter c by d2 is null;" +
            "store d into 'dummy';";

        LogicalPlan newLogicalPlan = migrateAndOptimizePlan(query);

        Operator op = newLogicalPlan.getSinks().get(0);
        Assert.assertTrue( op instanceof LOStore );
        op = newLogicalPlan.getPredecessors(op).get( 0 );
        Assert.assertTrue(  op instanceof LOFilter );
    }

    /**
     * Cascading joins. First one has outer, but filter should be able to push up above the second one, but below
     * the first one.
     */
    @Test
    public void testFullOutJoin1() throws Exception {
        String query = "A = load 'myfile' as (d1:int);" +
            "B = load 'anotherfile' as (d2:int);" +
            "C = join A by d1 full outer, B by d2;" +
            "D = load 'xxx' as (d3:int);" +
            "E = join C by d1, D by d3;" +
            "F = filter E by d1 > 5;" +
            "G = store F into 'dummy';";
        LogicalPlan newLogicalPlan = migrateAndOptimizePlan(query);

        List<Operator> ops = newLogicalPlan.getSinks();
        Assert.assertTrue( ops.size() == 1 );
        Operator op = ops.get( 0 );
        Assert.assertTrue( op instanceof LOStore );
        Operator join = newLogicalPlan.getPredecessors(op).get( 0 );
        Assert.assertTrue( join instanceof LOJoin );
        ops = newLogicalPlan.getPredecessors(join);
        Assert.assertEquals( 2, ops.size() );
        Assert.assertTrue( ops.get( 0 ) instanceof LOFilter || ops.get( 1 ) instanceof LOFilter );
    }

    private LogicalPlan migrateAndOptimizePlan(String query) throws Exception {
        PigServer pigServer = new PigServer( pc );
        LogicalPlan newLogicalPlan = Util.buildLp(pigServer, query);
        PlanOptimizer optimizer = new MyPlanOptimizer( newLogicalPlan, 3 );
        optimizer.optimize();
        return newLogicalPlan;
    }

    public class MyPlanOptimizer extends LogicalPlanOptimizer {
        protected MyPlanOptimizer(OperatorPlan p,  int iterations) {
            super(p, iterations, new HashSet<String>());
        }

        @Override
        public void addPlanTransformListener(PlanTransformListener listener) {
            super.addPlanTransformListener(listener);
        }

       @Override
    protected List<Set<Rule>> buildRuleSets() {
            List<Set<Rule>> ls = new ArrayList<Set<Rule>>();

            Set<Rule> s = new HashSet<Rule>();
            // add split filter rule
            Rule r = new LoadTypeCastInserter( "TypeCastInserter" );
            s.add(r);
            ls.add(s);

            s = new HashSet<Rule>();
            r = new PushUpFilter( "PushUpFilter" );
            s.add(r);
            ls.add(s);

            return ls;
        }
    }
}