/*
 * 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 static org.apache.pig.newplan.logical.relational.LOTestHelper.newLOLoad;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.util.Collection;
import java.util.List;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;
import org.apache.pig.ExecType;
import org.apache.pig.FuncSpec;
import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
import org.apache.pig.builtin.PigStorage;
import org.apache.pig.data.DataType;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.io.FileSpec;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.util.MultiMap;
import org.apache.pig.impl.util.Pair;
import org.apache.pig.newplan.BaseOperatorPlan;
import org.apache.pig.newplan.DependencyOrderWalker;
import org.apache.pig.newplan.DepthFirstWalker;
import org.apache.pig.newplan.Operator;
import org.apache.pig.newplan.OperatorPlan;
import org.apache.pig.newplan.PlanEdge;
import org.apache.pig.newplan.PlanVisitor;
import org.apache.pig.newplan.PlanWalker;
import org.apache.pig.newplan.ReverseDependencyOrderWalker;
import org.apache.pig.newplan.logical.expression.AndExpression;
import org.apache.pig.newplan.logical.expression.ConstantExpression;
import org.apache.pig.newplan.logical.expression.EqualExpression;
import org.apache.pig.newplan.logical.expression.LogicalExpressionPlan;
import org.apache.pig.newplan.logical.expression.LogicalExpressionVisitor;
import org.apache.pig.newplan.logical.expression.ProjectExpression;
import org.apache.pig.newplan.logical.relational.LOFilter;
import org.apache.pig.newplan.logical.relational.LOJoin;
import org.apache.pig.newplan.logical.relational.LOJoin.JOINTYPE;
import org.apache.pig.newplan.logical.relational.LOLoad;
import org.apache.pig.newplan.logical.relational.LogicalPlan;
import org.apache.pig.newplan.logical.relational.LogicalRelationalNodesVisitor;
import org.apache.pig.newplan.logical.relational.LogicalSchema;
import org.junit.Before;
import org.junit.Test;

public class TestNewPlanOperatorPlan {

    public static class FooLoad extends PigStorage {
        public FooLoad(String[] params) {
        }

    }

    private static class SillyPlan extends BaseOperatorPlan {

        SillyPlan() {
            super();
        }
    }

    static public class DummyLoad extends PigStorage {
        public DummyLoad(String a, String b) {
        }

        public DummyLoad(String a) {
        }

    }

    private static class SillyOperator extends Operator {
        private String name;

        SillyOperator(String n, SillyPlan p) {
            super(n, p);
            name = n;
        }

        @Override
        public void accept(PlanVisitor v) {
            if (v instanceof SillyVisitor) {
                ((SillyVisitor)v).visitSillyOperator(this);
            }
        }

        @Override
        public boolean isEqual(Operator operator) {
            return (name.compareTo(operator.getName()) == 0);
        }
    }

    private static class SillyVisitor extends PlanVisitor {

        StringBuffer buf;

        protected SillyVisitor(OperatorPlan plan, PlanWalker walker) {
            super(plan, walker);
            buf = new StringBuffer();
        }

        public void visitSillyOperator(SillyOperator so) {
            buf.append(so.getName());
        }

        public String getVisitPattern() {
            return buf.toString();
        }

    }

    Configuration conf = null;

    @Before
    public void setUp() throws Exception {
        PigContext pc = new PigContext(ExecType.LOCAL, new Properties());
        pc.connect();
        conf = new Configuration(
                ConfigurationUtil.toConfiguration(pc.getFs().getConfiguration())
                );
    }

    // Tests for PlanEdge

    @Test
    public void testPlanEdgeInsert() {
        SillyPlan plan = new SillyPlan();
        SillyOperator fred = new SillyOperator("fred", plan);
        SillyOperator joe = new SillyOperator("joe", plan);
        PlanEdge edges = new PlanEdge();

        // Test initial entry
        edges.put(fred, joe, 0);
        Collection<Operator> c = edges.get(fred);
        assertEquals(1, c.size());
        Operator[] a = new Operator[1];
        Operator[] b = c.toArray(a);
        assertEquals(joe, b[0]);

        // Test entry with no position
        SillyOperator bob = new SillyOperator("bob", plan);
        edges.put(fred, bob);
        c = edges.get(fred);
        assertEquals(2, c.size());
        a = new Operator[2];
        b = c.toArray(a);
        assertEquals(joe, b[0]);
        assertEquals(bob, b[1]);

        // Test entry with position
        SillyOperator jill = new SillyOperator("jill", plan);
        edges.put(fred, jill, 1);
        c = edges.get(fred);
        assertEquals(3, c.size());
        a = new Operator[3];
        b = c.toArray(a);
        assertEquals(joe, b[0]);
        assertEquals(jill, b[1]);
        assertEquals(bob, b[2]);
    }

    // Test that entry with invalid position cannot be made.
    @Test
    public void testPlanEdgeInsertFirstIndexBad() {
        SillyPlan plan = new SillyPlan();
        SillyOperator fred = new SillyOperator("fred", plan);
        SillyOperator joe = new SillyOperator("joe", plan);
        PlanEdge edges = new PlanEdge();
        boolean caught = false;
        try {
            edges.put(fred, joe, 1);
        } catch (IndexOutOfBoundsException e) {
            caught = true;
        }
        assertTrue(caught);

        caught = false;
        edges.put(fred, joe);
        SillyOperator bob = new SillyOperator("bob", plan);
        try {
            edges.put(fred, bob, 2);
        } catch (IndexOutOfBoundsException e) {
            caught = true;
        }
        assertTrue(caught);
    }

    // Test OperatorPlan
    @Test
    public void testOperatorPlan() throws FrontendException {
        SillyPlan plan = new SillyPlan();
        SillyOperator fred = new SillyOperator("fred", plan);
        SillyOperator joe = new SillyOperator("joe", plan);
        SillyOperator bob = new SillyOperator("bob", plan);
        SillyOperator jim = new SillyOperator("jim", plan);
        SillyOperator sam = new SillyOperator("sam", plan);

        // Test that roots and leaves are empty when there are no operators in
        // plan.
        List<Operator> list = plan.getSources();
        assertEquals(0, list.size());
        list = plan.getSinks();
        assertEquals(0, list.size());

        plan.add(fred);
        plan.add(joe);
        plan.add(bob);
        plan.add(jim);
        plan.add(sam);

        // Test that when not connected all nodes are roots and leaves.
        list = plan.getSources();
        assertEquals(5, list.size());
        list = plan.getSinks();
        assertEquals(5, list.size());

        // Connect them up
        plan.connect(fred, bob);
        plan.connect(joe, bob);
        plan.connect(bob, jim);
        plan.connect(bob, sam);

        // Check that the roots and leaves came out right
        list = plan.getSources();
        assertEquals(2, list.size());
        for (Operator op : list) {
            assertTrue(fred.isEqual(op) || joe.isEqual(op));
        }
        list = plan.getSinks();
        assertEquals(2, list.size());
        for (Operator op : list) {
            assertTrue(jim.isEqual(op) || sam.isEqual(op));
        }

        // Check each of their successors and predecessors
        list = plan.getSuccessors(fred);
        assertEquals(1, list.size());
        assertEquals(bob, list.get(0));

        list = plan.getSuccessors(joe);
        assertEquals(1, list.size());
        assertEquals(bob, list.get(0));

        list = plan.getPredecessors(jim);
        assertEquals(1, list.size());
        assertEquals(bob, list.get(0));

        list = plan.getPredecessors(sam);
        assertEquals(1, list.size());
        assertEquals(bob, list.get(0));

        list = plan.getPredecessors(bob);
        assertEquals(2, list.size());
        assertEquals(fred, list.get(0));
        assertEquals(joe, list.get(1));

        list = plan.getSuccessors(bob);
        assertEquals(2, list.size());
        assertEquals(jim, list.get(0));
        assertEquals(sam, list.get(1));

        // Now try swapping two, and check that all comes out as planned
        Pair<Integer, Integer> p1 = plan.disconnect(bob, jim);
        Pair<Integer, Integer> p2 = plan.disconnect(fred, bob);

        plan.connect(bob, p1.first, fred, p1.second);
        plan.connect(jim, p2.first, bob, p2.second);

        // Check that the roots and leaves came out right
        list = plan.getSources();
        assertEquals(2, list.size());
        for (Operator op : list) {
            assertTrue(jim.isEqual(op) || joe.isEqual(op));
        }
        list = plan.getSinks();
        assertEquals(2, list.size());
        for (Operator op : list) {
            assertTrue(fred.isEqual(op) || sam.isEqual(op));
        }

        // Check each of their successors and predecessors
        list = plan.getSuccessors(jim);
        assertEquals(1, list.size());
        assertEquals(bob, list.get(0));

        list = plan.getSuccessors(joe);
        assertEquals(1, list.size());
        assertEquals(bob, list.get(0));

        list = plan.getPredecessors(fred);
        assertEquals(1, list.size());
        assertEquals(bob, list.get(0));

        list = plan.getPredecessors(sam);
        assertEquals(1, list.size());
        assertEquals(bob, list.get(0));

        list = plan.getPredecessors(bob);
        assertEquals(2, list.size());
        assertEquals(jim, list.get(0));
        assertEquals(joe, list.get(1));

        list = plan.getSuccessors(bob);
        assertEquals(2, list.size());
        assertEquals(fred, list.get(0));
        assertEquals(sam, list.get(1));

    }

    @Test
    public void testDisconnectAndRemove() throws FrontendException {
        SillyPlan plan = new SillyPlan();
        SillyOperator fred = new SillyOperator("fred", plan);
        SillyOperator joe = new SillyOperator("joe", plan);
        SillyOperator bob = new SillyOperator("bob", plan);

        plan.add(fred);
        plan.add(joe);
        plan.add(bob);

        plan.connect(fred, joe);

        plan.remove(bob);
        plan.disconnect(fred, joe);

        List<Operator> list = plan.getSources();
        assertEquals(2, list.size());
        list = plan.getSinks();
        assertEquals(2, list.size());

        plan.remove(fred);
        plan.remove(joe);

        assertEquals(0, plan.size());

        list = plan.getSources();
        assertEquals(0, list.size());
        list = plan.getSinks();
        assertEquals(0, list.size());
    }

    // Test bad remove
    @Test
    public void testRemoveNegative() {
        SillyPlan plan = new SillyPlan();
        SillyOperator fred = new SillyOperator("fred", plan);
        SillyOperator joe = new SillyOperator("joe", plan);

        plan.add(fred);
        plan.add(joe);

        plan.connect(fred, joe);

        boolean caught = false;
        try {
            plan.remove(fred);
        } catch (FrontendException e) {
            caught = true;
        }
        assertTrue(caught);

        caught = false;
        try {
            plan.remove(joe);
        } catch (FrontendException e) {
            caught = true;
        }
        assertTrue(caught);

    }

    @Test
    public void testDisconnectNegative() {
        SillyPlan plan = new SillyPlan();
        SillyOperator fred = new SillyOperator("fred", plan);
        SillyOperator joe = new SillyOperator("joe", plan);

        plan.add(fred);
        plan.add(joe);

        boolean caught = false;
        try {
            plan.disconnect(fred, joe);
        } catch (FrontendException e) {
            caught = true;
        }
        assertTrue(caught);

    }

    // Tests for DependencyOrderWalker

    @Test
    public void testDependencyOrderWalkerLinear() throws FrontendException {
        SillyPlan plan = new SillyPlan();
        SillyOperator fred = new SillyOperator("fred", plan);
        SillyOperator joe = new SillyOperator("joe", plan);
        SillyOperator bob = new SillyOperator("bob", plan);

        plan.add(fred);
        plan.add(joe);
        plan.add(bob);

        plan.connect(fred, joe);
        plan.connect(joe, bob);

        SillyVisitor v =
                new SillyVisitor(plan, new DependencyOrderWalker(plan));

        v.visit();

        String s = v.getVisitPattern();

        assertEquals("fredjoebob", s);
    }

    @Test
    public void testDependencyOrderWalkerTree() throws FrontendException {
        SillyPlan plan = new SillyPlan();
        SillyOperator fred = new SillyOperator("fred", plan);
        SillyOperator joe = new SillyOperator("joe", plan);
        SillyOperator bob = new SillyOperator("bob", plan);
        SillyOperator jill = new SillyOperator("jill", plan);
        SillyOperator jane = new SillyOperator("jane", plan);

        plan.add(fred);
        plan.add(joe);
        plan.add(bob);
        plan.add(jill);
        plan.add(jane);

        plan.connect(fred, bob);
        plan.connect(joe, bob);
        plan.connect(bob, jill);
        plan.connect(jane, jill);

        SillyVisitor v =
                new SillyVisitor(plan, new DependencyOrderWalker(plan));

        v.visit();

        String s = v.getVisitPattern();

        if (!s.equals("fredjoebobjanejill") &&
                !s.equals("joefredbobjanejill") &&
                !s.equals("janefredjoebobjill") &&
                !s.equals("janejoefredbobjill")) {
            System.out.println("Invalid order " + s);
            fail();
        }
    }

    @Test
    public void testDependencyOrderWalkerGraph() throws FrontendException {
        SillyPlan plan = new SillyPlan();
        SillyOperator fred = new SillyOperator("fred", plan);
        SillyOperator joe = new SillyOperator("joe", plan);
        SillyOperator bob = new SillyOperator("bob", plan);
        SillyOperator jill = new SillyOperator("jill", plan);
        SillyOperator jane = new SillyOperator("jane", plan);

        plan.add(fred);
        plan.add(joe);
        plan.add(bob);
        plan.add(jill);
        plan.add(jane);

        plan.connect(fred, bob);
        plan.connect(joe, bob);
        plan.connect(bob, jill);
        plan.connect(bob, jane);

        SillyVisitor v =
                new SillyVisitor(plan, new DependencyOrderWalker(plan));

        v.visit();

        String s = v.getVisitPattern();

        if (!s.equals("fredjoebobjanejill") &&
                !s.equals("joefredbobjanejill") &&
                !s.equals("fredjoebobjilljane") &&
                !s.equals("joefredbobjilljane")) {
            System.out.println("Invalid order " + s);
            fail();
        }
    }

    // Tests for DepthFirstWalker

    @Test
    public void testDepthFirstWalkerLinear() throws FrontendException {
        SillyPlan plan = new SillyPlan();
        SillyOperator fred = new SillyOperator("fred", plan);
        SillyOperator joe = new SillyOperator("joe", plan);
        SillyOperator bob = new SillyOperator("bob", plan);

        plan.add(fred);
        plan.add(joe);
        plan.add(bob);

        plan.connect(fred, joe);
        plan.connect(joe, bob);

        SillyVisitor v =
                new SillyVisitor(plan, new DepthFirstWalker(plan));

        v.visit();

        String s = v.getVisitPattern();

        assertEquals("fredjoebob", s);
    }

    @Test
    public void testDepthFirstWalkerTree() throws FrontendException {
        SillyPlan plan = new SillyPlan();
        SillyOperator fred = new SillyOperator("fred", plan);
        SillyOperator joe = new SillyOperator("joe", plan);
        SillyOperator bob = new SillyOperator("bob", plan);
        SillyOperator jill = new SillyOperator("jill", plan);
        SillyOperator jane = new SillyOperator("jane", plan);

        plan.add(fred);
        plan.add(joe);
        plan.add(bob);
        plan.add(jill);
        plan.add(jane);

        plan.connect(fred, bob);
        plan.connect(fred, joe);
        plan.connect(joe, jill);
        plan.connect(joe, jane);

        SillyVisitor v =
                new SillyVisitor(plan, new DepthFirstWalker(plan));

        v.visit();

        String s = v.getVisitPattern();

        assertEquals("fredbobjoejilljane", s);
    }

    @Test
    public void testDepthFirstWalkerGraph() throws FrontendException {
        SillyPlan plan = new SillyPlan();
        SillyOperator fred = new SillyOperator("fred", plan);
        SillyOperator joe = new SillyOperator("joe", plan);
        SillyOperator bob = new SillyOperator("bob", plan);
        SillyOperator jill = new SillyOperator("jill", plan);
        SillyOperator jane = new SillyOperator("jane", plan);

        plan.add(fred);
        plan.add(joe);
        plan.add(bob);
        plan.add(jill);
        plan.add(jane);

        plan.connect(fred, bob);
        plan.connect(joe, bob);
        plan.connect(bob, jill);
        plan.connect(bob, jane);

        SillyVisitor v =
                new SillyVisitor(plan, new DepthFirstWalker(plan));

        v.visit();

        String s = v.getVisitPattern();

        if (!s.equals("fredbobjilljanejoe") &&
                !s.equals("joebobjilljanefred")) {
            System.out.println("Invalid order " + s);
            fail();
        }
    }

    // Tests for ReverseDependencyOrderWalker

    @Test
    public void testReverseDependencyOrderWalkerLinear() throws FrontendException {
        SillyPlan plan = new SillyPlan();
        SillyOperator fred = new SillyOperator("fred", plan);
        SillyOperator joe = new SillyOperator("joe", plan);
        SillyOperator bob = new SillyOperator("bob", plan);

        plan.add(fred);
        plan.add(joe);
        plan.add(bob);

        plan.connect(fred, joe);
        plan.connect(joe, bob);

        SillyVisitor v =
                new SillyVisitor(plan, new ReverseDependencyOrderWalker(plan));

        v.visit();

        String s = v.getVisitPattern();

        assertEquals("bobjoefred", s);
    }

    @Test
    public void testReverseDependencyOrderWalkerTree() throws FrontendException {
        SillyPlan plan = new SillyPlan();
        SillyOperator fred = new SillyOperator("fred", plan);
        SillyOperator joe = new SillyOperator("joe", plan);
        SillyOperator bob = new SillyOperator("bob", plan);
        SillyOperator jill = new SillyOperator("jill", plan);
        SillyOperator jane = new SillyOperator("jane", plan);

        plan.add(fred);
        plan.add(joe);
        plan.add(bob);
        plan.add(jill);
        plan.add(jane);

        plan.connect(fred, bob);
        plan.connect(joe, bob);
        plan.connect(bob, jill);
        plan.connect(jane, jill);

        SillyVisitor v =
                new SillyVisitor(plan, new ReverseDependencyOrderWalker(plan));

        v.visit();

        String s = v.getVisitPattern();

        if (!s.equals("jilljanebobjoefred") &&
                !s.equals("jilljanebobfredjoe") &&
                !s.equals("jillbobjoefredjane") &&
                !s.equals("jillbobjoejanefred") &&
                !s.equals("jillbobfredjoejane") &&
                !s.equals("jillbobfredjanejoe") &&
                !s.equals("jillbobjanejoefred") &&
                !s.equals("jillbobjanefredjoe")) {
            System.out.println("Invalid order " + s);
            fail();
        }
    }

    @Test
    public void testReverseDependencyOrderWalkerGraph() throws FrontendException {
        SillyPlan plan = new SillyPlan();
        SillyOperator fred = new SillyOperator("fred", plan);
        SillyOperator joe = new SillyOperator("joe", plan);
        SillyOperator bob = new SillyOperator("bob", plan);
        SillyOperator jill = new SillyOperator("jill", plan);
        SillyOperator jane = new SillyOperator("jane", plan);

        plan.add(fred);
        plan.add(joe);
        plan.add(bob);
        plan.add(jill);
        plan.add(jane);

        plan.connect(fred, bob);
        plan.connect(joe, bob);
        plan.connect(bob, jill);
        plan.connect(bob, jane);

        SillyVisitor v =
                new SillyVisitor(plan, new ReverseDependencyOrderWalker(plan));

        v.visit();

        String s = v.getVisitPattern();

        if (!s.equals("jilljanebobjoefred") &&
                !s.equals("jilljanebobfredjoe") &&
                !s.equals("janejillbobjoefred") &&
                !s.equals("janejillbobfredjoe")) {
            System.out.println("Invalid order " + s);
            fail();
        }
    }

    private static class TestLogicalVisitor extends LogicalRelationalNodesVisitor {

        StringBuffer bf = new StringBuffer();

        protected TestLogicalVisitor(OperatorPlan plan) throws FrontendException {
            super(plan, new DepthFirstWalker(plan));
        }

        @Override
        public void visit(LOLoad load) {
            bf.append("load ");
        }

        String getVisitPlan() {
            return bf.toString();
        }

    }

    @Test
    public void testLogicalPlanVisitor() throws FrontendException {
        LogicalPlan lp = new LogicalPlan();
        LOLoad load = newLOLoad(null, null, lp, conf);
        /*
         * lp.add((LogicalRelationalOperator)null, load,
         * (LogicalRelationalOperator)null);
         */
        lp.add(load);

        TestLogicalVisitor v = new TestLogicalVisitor(lp);
        v.visit();

        assertEquals("load ", v.getVisitPlan());
    }

    @Test
    public void testBinaryOperatorOrder() throws FrontendException {
        LogicalExpressionPlan ep = new LogicalExpressionPlan();
        ConstantExpression c = new ConstantExpression(ep, new Integer(5));
        ProjectExpression p = new ProjectExpression(ep, 0, 0, null);
        EqualExpression e = new EqualExpression(ep, p, c);
        assertEquals(p, e.getLhs());
        assertEquals(c, e.getRhs());
    }

    private static class TestExpressionVisitor extends LogicalExpressionVisitor {

        StringBuffer bf = new StringBuffer();

        protected TestExpressionVisitor(OperatorPlan plan) throws FrontendException {
            super(plan, new DepthFirstWalker(plan));
        }

        @Override
        public void visit(AndExpression andExpr) {
            bf.append("and ");
        }

        @Override
        public void visit(EqualExpression equal) {
            bf.append("equal ");
        }

        @Override
        public void visit(ProjectExpression project) {
            bf.append("project ");
        }

        @Override
        public void visit(ConstantExpression constant) {
            bf.append("constant ");
        }

        String getVisitPlan() {
            return bf.toString();
        }
    }

    @Test
    public void testExpressionPlanVisitor() throws FrontendException {
        LogicalExpressionPlan ep = new LogicalExpressionPlan();
        ConstantExpression c = new ConstantExpression(ep, new Integer(5));
        ProjectExpression p = new ProjectExpression(ep, 0, 0, null);
        EqualExpression e = new EqualExpression(ep, p, c);
        ConstantExpression c2 = new ConstantExpression(ep, new Boolean("true"));
        new AndExpression(ep, e, c2);

        TestExpressionVisitor v = new TestExpressionVisitor(ep);
        v.visit();
        assertEquals("and equal project constant constant ", v.getVisitPlan());
    }

    @Test
    public void testExpressionEquality() throws FrontendException {
        LogicalExpressionPlan ep1 = new LogicalExpressionPlan();
        ConstantExpression c1 = new ConstantExpression(ep1, new Integer(5));
        ProjectExpression p1 = new ProjectExpression(ep1, 0, 0, null);
        EqualExpression e1 = new EqualExpression(ep1, p1, c1);
        ConstantExpression ca1 = new ConstantExpression(ep1, new Boolean("true"));
        AndExpression a1 = new AndExpression(ep1, e1, ca1);

        LogicalExpressionPlan ep2 = new LogicalExpressionPlan();
        ConstantExpression c2 = new ConstantExpression(ep2, new Integer(5));
        ProjectExpression p2 = new ProjectExpression(ep2, 0, 0, null);
        EqualExpression e2 = new EqualExpression(ep2, p2, c2);
        ConstantExpression ca2 = new ConstantExpression(ep2, new Boolean("true"));
        AndExpression a2 = new AndExpression(ep2, e2, ca2);

        assertTrue(ep1.isEqual(ep2));
        assertTrue(c1.isEqual(c2));
        assertTrue(p1.isEqual(p2));
        assertTrue(e1.isEqual(e2));
        assertTrue(ca1.isEqual(ca2));
        assertTrue(a1.isEqual(a2));

        LogicalExpressionPlan ep3 = new LogicalExpressionPlan();
        ConstantExpression c3 = new ConstantExpression(ep3, new Integer(3));
        ProjectExpression p3 = new ProjectExpression(ep3, 0, 1, null);
        EqualExpression e3 = new EqualExpression(ep3, p3, c3);
        ConstantExpression ca3 = new ConstantExpression(ep3, "true");
        AndExpression a3 = new AndExpression(ep3, e3, ca3);

        assertFalse(ep1.isEqual(ep3));
        assertFalse(c1.isEqual(c3));
        assertFalse(p1.isEqual(p3));
        assertFalse(e1.isEqual(e3));
        assertFalse(ca1.isEqual(ca3));
        assertFalse(a1.isEqual(a3));

        LogicalExpressionPlan ep4 = new LogicalExpressionPlan();
        ProjectExpression p4 = new ProjectExpression(ep4, 1, 0, null);

        assertFalse(ep1.isEqual(ep4));
        assertFalse(p1.isEqual(p4));
    }

    @Test
    public void testRelationalEquality() throws FrontendException {
        // Build a plan that is the logical plan for
        // A = load 'bla' as (x);
        // B = load 'morebla' as (y);
        // C = join A on x, B on y;
        // D = filter C by y > 0;

        // A = load
        LogicalPlan lp = new LogicalPlan();
        {
            LogicalSchema aschema = new LogicalSchema();
            aschema.addField(new LogicalSchema.LogicalFieldSchema(
                    "x", null, DataType.INTEGER));
            LOLoad A = newLOLoad(new FileSpec("/abc",
                    new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "y" })), aschema,
                    lp, conf);
            lp.add(A);

            // B = load
            LogicalSchema bschema = new LogicalSchema();
            bschema.addField(new LogicalSchema.LogicalFieldSchema(
                    "y", null, DataType.INTEGER));
            LOLoad B = newLOLoad(new FileSpec("/def",
                    new FuncSpec("PigStorage", "\t")), bschema, lp, conf);
            lp.add(B);

            // C = join
            LogicalSchema cschema = new LogicalSchema();
            cschema.addField(new LogicalSchema.LogicalFieldSchema(
                    "x", null, DataType.INTEGER));
            cschema.addField(new LogicalSchema.LogicalFieldSchema(
                    "y", null, DataType.INTEGER));
            LogicalExpressionPlan aprojplan = new LogicalExpressionPlan();
            new ProjectExpression(aprojplan, 0, 0, null);
            LogicalExpressionPlan bprojplan = new LogicalExpressionPlan();
            new ProjectExpression(bprojplan, 1, 0, null);
            MultiMap<Integer, LogicalExpressionPlan> mm =
                    new MultiMap<Integer, LogicalExpressionPlan>();
            mm.put(0, aprojplan);
            mm.put(1, bprojplan);
            LOJoin C = new LOJoin(lp, mm, JOINTYPE.HASH, new boolean[] { true, true });
            C.neverUseForRealSetSchema(cschema);
            lp.add(C);
            lp.connect(A, C);
            lp.connect(B, C);

            // D = filter
            LogicalExpressionPlan filterPlan = new LogicalExpressionPlan();
            ProjectExpression fy = new ProjectExpression(filterPlan, 0, 1, null);
            ConstantExpression fc = new ConstantExpression(filterPlan, new Integer(0));
            new EqualExpression(filterPlan, fy, fc);
            LOFilter D = new LOFilter(lp, filterPlan);
            D.neverUseForRealSetSchema(cschema);
            lp.add(D);
            lp.connect(C, D);
        }

        // Build a second similar plan to test equality
        // A = load
        LogicalPlan lp1 = new LogicalPlan();
        {
            LogicalSchema aschema = new LogicalSchema();
            aschema.addField(new LogicalSchema.LogicalFieldSchema(
                    "x", null, DataType.INTEGER));
            LOLoad A = newLOLoad(new FileSpec("/abc",
                    new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "y" })), aschema,
                    lp1, conf);
            lp1.add(A);

            // B = load
            LogicalSchema bschema = new LogicalSchema();
            bschema.addField(new LogicalSchema.LogicalFieldSchema(
                    "y", null, DataType.INTEGER));
            LOLoad B = newLOLoad(new FileSpec("/def",
                    new FuncSpec("PigStorage", "\t")), bschema, lp1, conf);
            lp1.add(B);

            // C = join
            LogicalSchema cschema = new LogicalSchema();
            cschema.addField(new LogicalSchema.LogicalFieldSchema(
                    "x", null, DataType.INTEGER));
            cschema.addField(new LogicalSchema.LogicalFieldSchema(
                    "y", null, DataType.INTEGER));
            LogicalExpressionPlan aprojplan = new LogicalExpressionPlan();
            new ProjectExpression(aprojplan, 0, 0, null);
            LogicalExpressionPlan bprojplan = new LogicalExpressionPlan();
            new ProjectExpression(bprojplan, 1, 0, null);
            MultiMap<Integer, LogicalExpressionPlan> mm =
                    new MultiMap<Integer, LogicalExpressionPlan>();
            mm.put(0, aprojplan);
            mm.put(1, bprojplan);
            LOJoin C = new LOJoin(lp1, mm, JOINTYPE.HASH, new boolean[] { true, true });
            C.neverUseForRealSetSchema(cschema);
            lp1.add(C);
            lp1.connect(A, C);
            lp1.connect(B, C);

            // D = filter
            LogicalExpressionPlan filterPlan = new LogicalExpressionPlan();
            ProjectExpression fy = new ProjectExpression(filterPlan, 0, 1, null);
            ConstantExpression fc = new ConstantExpression(filterPlan, new Integer(0));
            new EqualExpression(filterPlan, fy, fc);
            LOFilter D = new LOFilter(lp1, filterPlan);
            D.neverUseForRealSetSchema(cschema);
            lp1.add(D);
            lp1.connect(C, D);

        }

        assertTrue(lp.isEqual(lp1));
    }

    @Test
    public void testLoadEqualityDifferentFuncSpecCtorArgs() throws FrontendException {
        LogicalPlan lp = new LogicalPlan();

        LogicalSchema aschema1 = new LogicalSchema();
        aschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad load1 = newLOLoad(new FileSpec("/abc",
                new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "y" })), aschema1, lp,
                conf);
        lp.add(load1);

        LOLoad load2 = newLOLoad(new FileSpec("/abc",
                new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "z" })), aschema1, lp,
                conf);
        lp.add(load2);

        assertFalse(load1.isEqual(load2));
    }

    @Test
    public void testLoadEqualityDifferentNumFuncSpecCstorArgs() throws FrontendException {
        LogicalPlan lp = new LogicalPlan();

        LogicalSchema aschema1 = new LogicalSchema();
        aschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad load1 = newLOLoad(new FileSpec("/abc",
                new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "y" })), aschema1, lp,
                conf);
        lp.add(load1);

        LOLoad load3 = newLOLoad(new FileSpec("/abc",
                new FuncSpec(DummyLoad.class.getName(), "x")), aschema1, lp, conf);
        lp.add(load3);

        assertFalse(load1.isEqual(load3));
    }

    @Test
    public void testLoadEqualityDifferentFunctionNames() throws FrontendException {
        LogicalPlan lp = new LogicalPlan();

        LogicalSchema aschema1 = new LogicalSchema();
        aschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad load1 = newLOLoad(new FileSpec("/abc",
                new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "y" })), aschema1, lp,
                conf);
        lp.add(load1);

        // Different function names in FuncSpec
        LOLoad load4 = newLOLoad(new FileSpec("/abc",
                new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "z" })), aschema1, lp,
                conf);
        lp.add(load4);

        assertFalse(load1.isEqual(load4));
    }

    @Test
    public void testLoadEqualityDifferentFileName() throws FrontendException {
        LogicalPlan lp = new LogicalPlan();
        LogicalSchema aschema1 = new LogicalSchema();
        aschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad load1 = newLOLoad(new FileSpec("/abc",
                new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "y" })), aschema1, lp,
                conf);
        lp.add(load1);

        // Different file name
        LOLoad load5 = newLOLoad(new FileSpec("/def",
                new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "z" })), aschema1, lp,
                conf);
        lp.add(load5);

        assertFalse(load1.isEqual(load5));
    }

    @Test
    public void testRelationalEqualityDifferentSchema() throws FrontendException {
        LogicalPlan lp = new LogicalPlan();
        LogicalSchema aschema1 = new LogicalSchema();
        aschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad load1 = newLOLoad(new FileSpec("/abc",
                new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "y" })), aschema1, lp,
                conf);
        lp.add(load1);

        // Different schema
        LogicalSchema aschema2 = new LogicalSchema();
        aschema2.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.CHARARRAY));

        LOLoad load6 = newLOLoad(new FileSpec("/abc",
                new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "z" })), aschema2, lp,
                conf);
        lp.add(load6);

        assertFalse(load1.isEqual(load6));
    }

    @Test
    public void testRelationalEqualityNullSchemas() throws FrontendException {
        LogicalPlan lp = new LogicalPlan();
        // Test that two loads with no schema are still equal
        LOLoad load7 = newLOLoad(new FileSpec("/abc",
                new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "y" })), null, lp, conf);
        lp.add(load7);

        LOLoad load8 = newLOLoad(new FileSpec("/abc",
                new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "y" })), null, lp, conf);
        lp.add(load8);

        assertTrue(load7.isEqual(load8));
    }

    @Test
    public void testRelationalEqualityOneNullOneNotNullSchema() throws FrontendException {
        LogicalPlan lp = new LogicalPlan();
        LogicalSchema aschema1 = new LogicalSchema();
        aschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad load1 = newLOLoad(new FileSpec("/abc",
                new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "y" })), aschema1, lp,
                conf);
        lp.add(load1);

        // Test that one with schema and one without breaks equality
        LOLoad load9 = newLOLoad(new FileSpec("/abc",
                new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "z" })), null, lp, conf);
        lp.add(load9);

        assertFalse(load1.isEqual(load9));
    }

    @Test
    public void testFilterDifferentPredicates() throws FrontendException {
        LogicalPlan lp = new LogicalPlan();

        LogicalExpressionPlan fp1 = new LogicalExpressionPlan();
        ProjectExpression fy1 = new ProjectExpression(fp1, 0, 1, null);
        ConstantExpression fc1 = new ConstantExpression(fp1,
                new Integer(0));
        new EqualExpression(fp1, fy1, fc1);
        LOFilter D1 = new LOFilter(lp, fp1);
        LogicalSchema cschema = new LogicalSchema();
        cschema.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        cschema.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        D1.neverUseForRealSetSchema(cschema);
        lp.add(D1);

        LogicalExpressionPlan fp2 = new LogicalExpressionPlan();
        ProjectExpression fy2 = new ProjectExpression(fp2, 0, 1, null);
        ConstantExpression fc2 = new ConstantExpression(fp2,
                new Integer(1));
        new EqualExpression(fp2, fy2, fc2);
        LOFilter D2 = new LOFilter(lp, fp2);
        D2.neverUseForRealSetSchema(cschema);
        lp.add(D2);

        assertFalse(D1.isEqual(D2));
    }

    // No tests for LOStore because it tries to actually instantiate the store
    // func, and I don't want to mess with that here.

    @Test
    public void testJoinDifferentJoinTypes() throws FrontendException {
        LogicalPlan lp = new LogicalPlan();
        LogicalSchema jaschema1 = new LogicalSchema();
        jaschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad A1 = newLOLoad(new FileSpec("/abc",
                new FuncSpec("org.apache.pig.test.TestNewPlanOperatorPlan$FooLoad", new String[] {
                                "x", "y" })), jaschema1, lp, conf);
        lp.add(A1);

        // B = load
        LogicalSchema jbschema1 = new LogicalSchema();
        jbschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LOLoad B1 = newLOLoad(new FileSpec("/def",
                new FuncSpec("PigStorage", "\t")), jbschema1, lp, conf);
        lp.add(B1);

        // C = join
        LogicalSchema jcschema1 = new LogicalSchema();
        jcschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        jcschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LogicalExpressionPlan aprojplan1 = new LogicalExpressionPlan();
        new ProjectExpression(aprojplan1, 0, 0, null);
        LogicalExpressionPlan bprojplan1 = new LogicalExpressionPlan();
        new ProjectExpression(bprojplan1, 1, 0, null);
        MultiMap<Integer, LogicalExpressionPlan> mm1 =
                new MultiMap<Integer, LogicalExpressionPlan>();
        mm1.put(0, aprojplan1);
        mm1.put(1, bprojplan1);
        LOJoin C1 = new LOJoin(lp, mm1, JOINTYPE.HASH, new boolean[] { true, true });
        C1.neverUseForRealSetSchema(jcschema1);
        lp.add(C1);
        lp.connect(A1, C1);
        lp.connect(B1, C1);

        // A = load
        LogicalSchema jaschema2 = new LogicalSchema();
        jaschema2.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad A2 = newLOLoad(new FileSpec("/abc",
                new FuncSpec("org.apache.pig.test.TestNewPlanOperatorPlan$FooLoad", new String[] {
                                "x", "y" })), jaschema2, lp, conf);
        lp.add(A2);

        // B = load
        LogicalSchema jbschema2 = new LogicalSchema();
        jbschema2.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LOLoad B2 = newLOLoad(new FileSpec("/def",
                new FuncSpec("PigStorage", "\t")), jbschema2, lp, conf);
        lp.add(B2);

        // C = join
        LogicalSchema jcschema2 = new LogicalSchema();
        jcschema2.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        jcschema2.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LogicalExpressionPlan aprojplan2 = new LogicalExpressionPlan();
        new ProjectExpression(aprojplan2, 0, 0, null);
        LogicalExpressionPlan bprojplan2 = new LogicalExpressionPlan();
        new ProjectExpression(bprojplan2, 1, 0, null);
        MultiMap<Integer, LogicalExpressionPlan> mm2 =
                new MultiMap<Integer, LogicalExpressionPlan>();
        mm2.put(0, aprojplan2);
        mm2.put(1, bprojplan2);
        LOJoin C2 = new LOJoin(lp, mm2, JOINTYPE.SKEWED, new boolean[] { true, true });
        C2.neverUseForRealSetSchema(jcschema2);
        lp.add(C2);
        lp.connect(A2, C2);
        lp.connect(B2, C2);

        assertFalse(C1.isEqual(C2));
    }

    @Test
    public void testJoinDifferentInner() throws FrontendException {
        LogicalPlan lp = new LogicalPlan();
        LogicalSchema jaschema1 = new LogicalSchema();
        jaschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad A1 = newLOLoad(new FileSpec("/abc",
                new FuncSpec("org.apache.pig.test.TestNewPlanOperatorPlan$FooLoad", new String[] {
                                "x", "y" })), jaschema1, lp, conf);
        lp.add(A1);

        // B = load
        LogicalSchema jbschema1 = new LogicalSchema();
        jbschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LOLoad B1 = newLOLoad(new FileSpec("/def",
                new FuncSpec("PigStorage", "\t")), jbschema1, lp, conf);
        lp.add(B1);

        // C = join
        LogicalSchema jcschema1 = new LogicalSchema();
        jcschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        jcschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LogicalExpressionPlan aprojplan1 = new LogicalExpressionPlan();
        new ProjectExpression(aprojplan1, 0, 0, null);
        LogicalExpressionPlan bprojplan1 = new LogicalExpressionPlan();
        new ProjectExpression(bprojplan1, 1, 0, null);
        MultiMap<Integer, LogicalExpressionPlan> mm1 =
                new MultiMap<Integer, LogicalExpressionPlan>();
        mm1.put(0, aprojplan1);
        mm1.put(1, bprojplan1);
        LOJoin C1 = new LOJoin(lp, mm1, JOINTYPE.HASH, new boolean[] { true, true });
        C1.neverUseForRealSetSchema(jcschema1);
        lp.add(C1);
        lp.connect(A1, C1);
        lp.connect(B1, C1);

        // Test different inner status
        // A = load
        LogicalSchema jaschema3 = new LogicalSchema();
        jaschema3.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad A3 = newLOLoad(new FileSpec("/abc",
                new FuncSpec("org.apache.pig.test.TestNewPlanOperatorPlan$FooLoad", new String[] {
                                "x", "y" })), jaschema3, lp, conf);
        lp.add(A3);

        // B = load
        LogicalSchema jbschema3 = new LogicalSchema();
        jbschema3.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LOLoad B3 = newLOLoad(new FileSpec("/def",
                new FuncSpec("PigStorage", "\t")), jbschema3, lp, conf);
        lp.add(B3);

        // C = join
        LogicalSchema jcschema3 = new LogicalSchema();
        jcschema3.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        jcschema3.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LogicalExpressionPlan aprojplan3 = new LogicalExpressionPlan();
        new ProjectExpression(aprojplan3, 0, 0, null);
        LogicalExpressionPlan bprojplan3 = new LogicalExpressionPlan();
        new ProjectExpression(bprojplan3, 1, 0, null);
        MultiMap<Integer, LogicalExpressionPlan> mm3 =
                new MultiMap<Integer, LogicalExpressionPlan>();
        mm3.put(0, aprojplan3);
        mm3.put(1, bprojplan3);
        LOJoin C3 = new LOJoin(lp, mm3, JOINTYPE.HASH, new boolean[] { true, false });
        C3.neverUseForRealSetSchema(jcschema3);
        lp.add(C3);
        lp.connect(A3, C3);
        lp.connect(B3, C3);

        assertFalse(C1.isEqual(C3));
    }

    @Test
    public void testJoinDifferentNumInputs() throws FrontendException {
        LogicalPlan lp = new LogicalPlan();
        LogicalSchema jaschema1 = new LogicalSchema();
        jaschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad A1 = newLOLoad(new FileSpec("/abc",
                new FuncSpec("org.apache.pig.test.TestNewPlanOperatorPlan$FooLoad", new String[] {
                                "x", "y" })), jaschema1, lp, conf);
        lp.add(A1);

        // B = load
        LogicalSchema jbschema1 = new LogicalSchema();
        jbschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LOLoad B1 = newLOLoad(new FileSpec("/def",
                new FuncSpec("PigStorage", "\t")), jbschema1, lp, conf);
        lp.add(B1);

        // C = join
        LogicalSchema jcschema1 = new LogicalSchema();
        jcschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        jcschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LogicalExpressionPlan aprojplan1 = new LogicalExpressionPlan();
        new ProjectExpression(aprojplan1, 0, 0, null);
        LogicalExpressionPlan bprojplan1 = new LogicalExpressionPlan();
        new ProjectExpression(bprojplan1, 1, 0, null);
        MultiMap<Integer, LogicalExpressionPlan> mm1 =
                new MultiMap<Integer, LogicalExpressionPlan>();
        mm1.put(0, aprojplan1);
        mm1.put(1, bprojplan1);
        LOJoin C1 = new LOJoin(lp, mm1, JOINTYPE.HASH, new boolean[] { true, true });
        C1.neverUseForRealSetSchema(jcschema1);
        lp.add(C1);
        lp.connect(A1, C1);
        lp.connect(B1, C1);

        // A = load
        LogicalSchema jaschema5 = new LogicalSchema();
        jaschema5.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad A5 = newLOLoad(new FileSpec("/abc",
                new FuncSpec("org.apache.pig.test.TestNewPlanOperatorPlan$FooLoad", new String[] {
                                "x", "y" })), jaschema5, lp, conf);
        lp.add(A5);

        // B = load
        LogicalSchema jbschema5 = new LogicalSchema();
        jbschema5.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LOLoad B5 = newLOLoad(new FileSpec("/def",
                new FuncSpec("PigStorage", "\t")), jbschema5, lp, conf);
        lp.add(B5);

        // Beta = load
        LogicalSchema jbetaschema5 = new LogicalSchema();
        jbetaschema5.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LOLoad Beta5 = newLOLoad(new FileSpec("/ghi",
                new FuncSpec("PigStorage", "\t")), jbetaschema5, lp, conf);
        lp.add(Beta5);

        // C = join
        LogicalSchema jcschema5 = new LogicalSchema();
        jcschema5.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        jcschema5.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LogicalExpressionPlan aprojplan5 = new LogicalExpressionPlan();
        new ProjectExpression(aprojplan5, 0, 0, null);
        LogicalExpressionPlan bprojplan5 = new LogicalExpressionPlan();
        new ProjectExpression(bprojplan5, 1, 0, null);
        LogicalExpressionPlan betaprojplan5 = new LogicalExpressionPlan();
        new ProjectExpression(betaprojplan5, 1, 0, null);
        MultiMap<Integer, LogicalExpressionPlan> mm5 =
                new MultiMap<Integer, LogicalExpressionPlan>();
        mm5.put(0, aprojplan5);
        mm5.put(1, bprojplan5);
        mm5.put(2, betaprojplan5);
        LOJoin C5 = new LOJoin(lp, mm5, JOINTYPE.HASH, new boolean[] { true, true });
        C5.neverUseForRealSetSchema(jcschema5);
        lp.add(C5);
        lp.connect(A5, C5);
        lp.connect(B5, C5);
        lp.connect(Beta5, C5);

        assertFalse(C1.isEqual(C5));
    }

    @Test
    public void testJoinDifferentJoinKeys() throws FrontendException {
        LogicalPlan lp = new LogicalPlan();

        // Test different join keys
        LogicalSchema jaschema6 = new LogicalSchema();
        jaschema6.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad A6 = newLOLoad(new FileSpec("/abc",
                new FuncSpec("org.apache.pig.test.TestNewPlanOperatorPlan$FooLoad", new String[] {
                                "x", "y" })), jaschema6, lp, conf);
        lp.add(A6);

        // B = load
        LogicalSchema jbschema6 = new LogicalSchema();
        jbschema6.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        jbschema6.addField(new LogicalSchema.LogicalFieldSchema(
                "z", null, DataType.LONG));
        LOLoad B6 = newLOLoad(new FileSpec("/def",
                new FuncSpec("PigStorage", "\t")), jbschema6, lp, conf);
        lp.add(B6);

        // C = join
        LogicalSchema jcschema6 = new LogicalSchema();
        jcschema6.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        jcschema6.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LogicalExpressionPlan aprojplan6 = new LogicalExpressionPlan();
        new ProjectExpression(aprojplan6, 0, 0, null);
        LogicalExpressionPlan bprojplan6 = new LogicalExpressionPlan();
        new ProjectExpression(bprojplan6, 1, 0, null);
        LogicalExpressionPlan b2projplan6 = new LogicalExpressionPlan();
        new ProjectExpression(b2projplan6, 1, 1, null);
        MultiMap<Integer, LogicalExpressionPlan> mm6 =
                new MultiMap<Integer, LogicalExpressionPlan>();
        mm6.put(0, aprojplan6);
        mm6.put(1, bprojplan6);
        mm6.put(1, b2projplan6);
        LOJoin C6 = new LOJoin(lp, mm6, JOINTYPE.HASH, new boolean[] { true, true });
        C6.neverUseForRealSetSchema(jcschema6);
        lp.add(C6);
        lp.connect(A6, C6);
        lp.connect(B6, C6);

        LogicalSchema jaschema7 = new LogicalSchema();
        jaschema7.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad A7 = newLOLoad(new FileSpec("/abc",
                new FuncSpec("org.apache.pig.test.TestNewPlanOperatorPlan$FooLoad", new String[] {
                                "x", "y" })), jaschema7, lp, conf);
        lp.add(A7);

        // B = load
        LogicalSchema jbschema7 = new LogicalSchema();
        jbschema7.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        jbschema7.addField(new LogicalSchema.LogicalFieldSchema(
                "z", null, DataType.LONG));
        LOLoad B7 = newLOLoad(new FileSpec("/def",
                new FuncSpec("PigStorage", "\t")), jbschema7, lp, conf);
        lp.add(B7);

        // C = join
        LogicalSchema jcschema7 = new LogicalSchema();
        jcschema7.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        jcschema7.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LogicalExpressionPlan aprojplan7 = new LogicalExpressionPlan();
        new ProjectExpression(aprojplan7, 0, 0, null);
        LogicalExpressionPlan bprojplan7 = new LogicalExpressionPlan();
        new ProjectExpression(bprojplan7, 1, 1, null);
        LogicalExpressionPlan b2projplan7 = new LogicalExpressionPlan();
        new ProjectExpression(b2projplan7, 1, 0, null);
        MultiMap<Integer, LogicalExpressionPlan> mm7 =
                new MultiMap<Integer, LogicalExpressionPlan>();
        mm7.put(0, aprojplan7);
        mm7.put(1, bprojplan7);
        mm7.put(1, b2projplan7);
        LOJoin C7 = new LOJoin(lp, mm7, JOINTYPE.HASH, new boolean[] { true, true });
        C7.neverUseForRealSetSchema(jcschema7);
        lp.add(C7);
        lp.connect(A7, C7);
        lp.connect(B7, C7);

        assertFalse(C6.isEqual(C7));
    }

    @Test
    public void testJoinDifferentNumJoinKeys() throws FrontendException {
        LogicalPlan lp = new LogicalPlan();

        // Test different join keys
        LogicalSchema jaschema6 = new LogicalSchema();
        jaschema6.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad A6 = newLOLoad(new FileSpec("/abc",
                new FuncSpec("org.apache.pig.test.TestNewPlanOperatorPlan$FooLoad", new String[] {
                                "x", "y" })), jaschema6, lp, conf);
        lp.add(A6);

        // B = load
        LogicalSchema jbschema6 = new LogicalSchema();
        jbschema6.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        jbschema6.addField(new LogicalSchema.LogicalFieldSchema(
                "z", null, DataType.LONG));
        LOLoad B6 = newLOLoad(new FileSpec("/def",
                new FuncSpec("PigStorage", "\t")), jbschema6, lp, conf);
        lp.add(B6);

        // C = join
        LogicalSchema jcschema6 = new LogicalSchema();
        jcschema6.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        jcschema6.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LogicalExpressionPlan aprojplan6 = new LogicalExpressionPlan();
        new ProjectExpression(aprojplan6, 0, 0, null);
        LogicalExpressionPlan bprojplan6 = new LogicalExpressionPlan();
        new ProjectExpression(bprojplan6, 1, 0, null);
        LogicalExpressionPlan b2projplan6 = new LogicalExpressionPlan();
        new ProjectExpression(b2projplan6, 1, 1, null);
        MultiMap<Integer, LogicalExpressionPlan> mm6 =
                new MultiMap<Integer, LogicalExpressionPlan>();
        mm6.put(0, aprojplan6);
        mm6.put(1, bprojplan6);
        mm6.put(1, b2projplan6);
        LOJoin C6 = new LOJoin(lp, mm6, JOINTYPE.HASH, new boolean[] { true, true });
        C6.neverUseForRealSetSchema(jcschema6);
        lp.add(C6);
        lp.connect(A6, C6);
        lp.connect(B6, C6);

        // Test different different number of join keys
        LogicalSchema jaschema8 = new LogicalSchema();
        jaschema8.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad A8 = newLOLoad(new FileSpec("/abc",
                new FuncSpec("org.apache.pig.test.TestNewPlanOperatorPlan$FooLoad", new String[] {
                                "x", "y" })), jaschema8, lp, conf);
        lp.add(A8);

        // B = load
        LogicalSchema jbschema8 = new LogicalSchema();
        jbschema8.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        jbschema8.addField(new LogicalSchema.LogicalFieldSchema(
                "z", null, DataType.LONG));
        LOLoad B8 = newLOLoad(new FileSpec("/def",
                new FuncSpec("PigStorage", "\t")), jbschema8, lp, conf);
        lp.add(B8);

        // C = join
        LogicalSchema jcschema8 = new LogicalSchema();
        jcschema8.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        jcschema8.addField(new LogicalSchema.LogicalFieldSchema(
                "y", null, DataType.INTEGER));
        LogicalExpressionPlan aprojplan8 = new LogicalExpressionPlan();
        new ProjectExpression(aprojplan8, 0, 0, null);
        LogicalExpressionPlan bprojplan8 = new LogicalExpressionPlan();
        new ProjectExpression(bprojplan8, 1, 0, null);
        MultiMap<Integer, LogicalExpressionPlan> mm8 =
                new MultiMap<Integer, LogicalExpressionPlan>();
        mm8.put(0, aprojplan8);
        mm8.put(1, bprojplan8);
        LOJoin C8 = new LOJoin(lp, mm8, JOINTYPE.HASH, new boolean[] { true, true });
        C8.neverUseForRealSetSchema(jcschema8);
        lp.add(C8);
        lp.connect(A8, C8);
        lp.connect(B8, C8);

        assertFalse(C6.isEqual(C8));
    }

    @Test
    public void testRelationalSameOpDifferentPreds() throws FrontendException {
        LogicalPlan lp1 = new LogicalPlan();
        LogicalSchema aschema1 = new LogicalSchema();
        aschema1.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        LOLoad A1 = newLOLoad(new FileSpec("/abc",
                new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "y" })), aschema1, lp1,
                conf);
        lp1.add(A1);

        LogicalExpressionPlan fp1 = new LogicalExpressionPlan();
        ProjectExpression fy1 = new ProjectExpression(fp1, 0, 0, null);
        ConstantExpression fc1 = new ConstantExpression(fp1,
                new Integer(0));
        new EqualExpression(fp1, fy1, fc1);
        LOFilter D1 = new LOFilter(lp1, fp1);
        LogicalSchema cschema = new LogicalSchema();
        cschema.addField(new LogicalSchema.LogicalFieldSchema(
                "x", null, DataType.INTEGER));
        D1.neverUseForRealSetSchema(cschema);
        lp1.add(D1);
        lp1.connect(A1, D1);

        LogicalPlan lp2 = new LogicalPlan();
        LOLoad A2 = newLOLoad(new FileSpec("/abc",
                new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "z" })), null, lp2,
                conf);
        lp2.add(A2);

        LogicalExpressionPlan fp2 = new LogicalExpressionPlan();
        ProjectExpression fy2 = new ProjectExpression(fp2, 0, 0, null);
        ConstantExpression fc2 = new ConstantExpression(fp2,
                new Integer(0));
        new EqualExpression(fp2, fy2, fc2);
        LOFilter D2 = new LOFilter(lp2, fp2);
        D2.neverUseForRealSetSchema(cschema);
        lp2.add(D2);
        lp2.connect(A2, D2);

        assertTrue(D1.isEqual(D2));
    }

    @Test
    public void testReplace1() throws FrontendException {
        // has multiple inputs
        SillyPlan plan = new SillyPlan();
        Operator load1 = new SillyOperator("load1", plan);
        Operator load2 = new SillyOperator("load2", plan);
        Operator filter1 = new SillyOperator("filter1", plan);
        Operator join1 = new SillyOperator("join1", plan);
        Operator filter2 = new SillyOperator("filter2", plan);
        plan.add(load1);
        plan.add(load2);
        plan.add(filter1);
        plan.add(filter2);
        plan.add(join1);
        plan.connect(load1, join1);
        plan.connect(load2, filter1);
        plan.connect(filter1, join1);
        plan.connect(join1, filter2);

        Operator join2 = new SillyOperator("join2", plan);
        plan.replace(join1, join2);

        List<Operator> preds = plan.getPredecessors(join2);
        assertEquals(2, preds.size());
        assertTrue(preds.contains(load1));
        assertTrue(preds.contains(filter1));

        List<Operator> succs = plan.getSuccessors(join2);
        assertEquals(1, succs.size());
        assertTrue(succs.contains(filter2));
    }

    @Test
    public void testReplace2() throws FrontendException {
        // has multiple outputs
        SillyPlan plan = new SillyPlan();
        Operator load1 = new SillyOperator("load1", plan);
        Operator split1 = new SillyOperator("split1", plan);
        Operator filter1 = new SillyOperator("filter1", plan);
        Operator filter2 = new SillyOperator("filter2", plan);
        plan.add(load1);
        plan.add(split1);
        plan.add(filter1);
        plan.add(filter2);
        plan.connect(load1, split1);
        plan.connect(split1, filter1);
        plan.connect(split1, filter2);

        Operator split2 = new SillyOperator("split2", plan);
        plan.replace(split1, split2);

        List<Operator> preds = plan.getPredecessors(split2);
        assertEquals(1, preds.size());
        assertTrue(preds.contains(load1));

        List<Operator> succs = plan.getSuccessors(split2);
        assertEquals(2, succs.size());
        assertTrue(succs.contains(filter1));
        assertTrue(succs.contains(filter2));
    }

    @Test
    public void testReplace3() throws FrontendException {
        // single input/output
        SillyPlan plan = new SillyPlan();
        Operator load1 = new SillyOperator("load1", plan);
        Operator filter1 = new SillyOperator("filter1", plan);
        Operator filter2 = new SillyOperator("filter2", plan);
        plan.add(load1);
        plan.add(filter1);
        plan.add(filter2);
        plan.connect(load1, filter1);
        plan.connect(filter1, filter2);

        Operator filter3 = new SillyOperator("filter3", plan);
        plan.replace(filter1, filter3);

        List<Operator> preds = plan.getPredecessors(filter3);
        assertEquals(1, preds.size());
        assertTrue(preds.contains(load1));

        List<Operator> succs = plan.getSuccessors(filter3);
        assertEquals(1, succs.size());
        assertTrue(succs.contains(filter2));
    }

    @Test
    public void testReplace4() throws FrontendException {
        // output is null
        SillyPlan plan = new SillyPlan();
        Operator load1 = new SillyOperator("load1", plan);
        Operator filter1 = new SillyOperator("filter1", plan);
        Operator filter2 = new SillyOperator("filter2", plan);
        plan.add(load1);
        plan.add(filter1);
        plan.add(filter2);
        plan.connect(load1, filter1);
        plan.connect(filter1, filter2);

        Operator filter3 = new SillyOperator("filter3", plan);
        plan.replace(filter2, filter3);

        List<Operator> preds = plan.getPredecessors(filter3);
        assertEquals(1, preds.size());
        assertTrue(preds.contains(filter1));

        List<Operator> succs = plan.getSuccessors(filter3);
        assertNull(succs);
    }

    @Test
    public void testReplace5() throws FrontendException {
        // input is null
        SillyPlan plan = new SillyPlan();
        Operator load1 = new SillyOperator("load1", plan);
        Operator filter1 = new SillyOperator("filter1", plan);
        Operator filter2 = new SillyOperator("filter2", plan);
        plan.add(load1);
        plan.add(filter1);
        plan.add(filter2);
        plan.connect(load1, filter1);
        plan.connect(filter1, filter2);

        Operator load2 = new SillyOperator("load2", plan);
        plan.replace(load1, load2);

        List<Operator> preds = plan.getPredecessors(load2);
        assertNull(preds);

        List<Operator> succs = plan.getSuccessors(load2);
        assertEquals(1, succs.size());
        assertTrue(succs.contains(filter1));
    }

    @Test
    public void testReplace6() throws FrontendException {
        // has multiple inputs/outputs
        SillyPlan plan = new SillyPlan();
        Operator load1 = new SillyOperator("load1", plan);
        Operator load2 = new SillyOperator("load2", plan);
        Operator filter1 = new SillyOperator("filter1", plan);
        // fake operator to take multiple inputs/outputs
        Operator fake1 = new SillyOperator("fake1", plan);
        Operator filter2 = new SillyOperator("filter2", plan);
        Operator filter3 = new SillyOperator("filter3", plan);
        plan.add(load1);
        plan.add(load2);
        plan.add(filter1);
        plan.add(filter2);
        plan.add(filter3);
        plan.add(fake1);
        plan.connect(load1, fake1);
        plan.connect(load2, filter1);
        plan.connect(filter1, fake1);
        plan.connect(fake1, filter2);
        plan.connect(fake1, filter3);

        Operator fake2 = new SillyOperator("fake2", plan);
        plan.replace(fake1, fake2);

        List<Operator> preds = plan.getPredecessors(fake2);
        assertEquals(2, preds.size());
        assertTrue(preds.contains(load1));
        assertTrue(preds.contains(filter1));

        List<Operator> succs = plan.getSuccessors(fake2);
        assertEquals(2, succs.size());
        assertTrue(succs.contains(filter2));
        assertTrue(succs.contains(filter3));
    }

    @Test
    public void testRemove1() throws FrontendException {
        // single input/output
        SillyPlan plan = new SillyPlan();
        Operator load1 = new SillyOperator("load1", plan);
        Operator load2 = new SillyOperator("load2", plan);
        Operator filter1 = new SillyOperator("filter1", plan);
        Operator join1 = new SillyOperator("join1", plan);
        Operator filter2 = new SillyOperator("filter2", plan);
        plan.add(load1);
        plan.add(load2);
        plan.add(filter1);
        plan.add(filter2);
        plan.add(join1);
        plan.connect(load1, join1);
        plan.connect(load2, filter1);
        plan.connect(filter1, join1);
        plan.connect(join1, filter2);

        plan.removeAndReconnect(filter1);

        List<Operator> preds = plan.getPredecessors(join1);
        assertEquals(2, preds.size());
        assertTrue(preds.contains(load2));
    }

    @Test
    public void testRemove2() throws FrontendException {
        // input is null
        SillyPlan plan = new SillyPlan();
        Operator load1 = new SillyOperator("load1", plan);
        Operator load2 = new SillyOperator("load2", plan);
        Operator filter1 = new SillyOperator("filter1", plan);
        Operator join1 = new SillyOperator("join1", plan);
        Operator filter2 = new SillyOperator("filter2", plan);
        plan.add(load1);
        plan.add(load2);
        plan.add(filter1);
        plan.add(filter2);
        plan.add(join1);
        plan.connect(load1, join1);
        plan.connect(load2, filter1);
        plan.connect(filter1, join1);
        plan.connect(join1, filter2);

        plan.removeAndReconnect(load1);

        List<Operator> preds = plan.getPredecessors(join1);
        assertEquals(1, preds.size());
        assertTrue(preds.contains(filter1));

        plan.removeAndReconnect(filter1);
        preds = plan.getPredecessors(join1);
        assertEquals(1, preds.size());
        assertTrue(preds.contains(load2));
    }

    @Test
    public void testRemove3() throws FrontendException {
        // output is null
        SillyPlan plan = new SillyPlan();
        Operator load1 = new SillyOperator("load1", plan);
        Operator filter1 = new SillyOperator("filter1", plan);
        Operator filter2 = new SillyOperator("filter2", plan);
        plan.add(load1);
        plan.add(filter1);
        plan.add(filter2);
        plan.connect(load1, filter1);
        plan.connect(filter1, filter2);

        plan.removeAndReconnect(filter2);

        List<Operator> succs = plan.getSuccessors(filter2);
        assertNull(succs);
    }

    @Test
    public void testRemove4() throws FrontendException {
        // has multiple inputs
        SillyPlan plan = new SillyPlan();
        Operator load1 = new SillyOperator("load1", plan);
        Operator load2 = new SillyOperator("load2", plan);
        Operator filter1 = new SillyOperator("filter1", plan);
        Operator join1 = new SillyOperator("join1", plan);
        Operator fake1 = new SillyOperator("fake1", plan);
        plan.add(load1);
        plan.add(load2);
        plan.add(filter1);
        plan.add(join1);
        plan.connect(load1, join1);
        plan.connect(load2, filter1);
        plan.connect(filter1, join1);
        plan.connect(join1, fake1);

        plan.removeAndReconnect(join1);

        List<Operator> preds = plan.getPredecessors(fake1);
        assertEquals(2, preds.size());
        assertTrue(preds.contains(load1));
        assertTrue(preds.contains(filter1));
    }

    @Test
    public void testRemove5() throws FrontendException {
        // has multiple outputs
        SillyPlan plan = new SillyPlan();
        Operator load1 = new SillyOperator("load1", plan);
        Operator split1 = new SillyOperator("split1", plan);
        Operator split2 = new SillyOperator("split2", plan);
        Operator filter1 = new SillyOperator("filter1", plan);
        Operator filter2 = new SillyOperator("filter2", plan);
        plan.add(load1);
        plan.add(split1);
        plan.add(filter1);
        plan.add(filter2);
        plan.connect(load1, split1);
        plan.connect(split1, split2);
        plan.connect(split2, filter1);
        plan.connect(split2, filter2);

        plan.removeAndReconnect(split2);

        List<Operator> succs = plan.getSuccessors(split1);
        assertEquals(2, succs.size());
        assertTrue(succs.contains(filter1));
        assertTrue(succs.contains(filter2));
    }

    @Test(expected = FrontendException.class)
    public void testRemove6() throws FrontendException {
        // has multiple inputs/outputs
        SillyPlan plan = new SillyPlan();
        Operator load1 = new SillyOperator("load1", plan);
        Operator load2 = new SillyOperator("load2", plan);
        Operator fake1 = new SillyOperator("fake1", plan);
        Operator filter1 = new SillyOperator("filter1", plan);
        Operator filter2 = new SillyOperator("filter2", plan);
        plan.add(load1);
        plan.add(load2);
        plan.add(fake1);
        plan.add(filter1);
        plan.add(filter2);
        plan.connect(load1, fake1);
        plan.connect(load2, fake1);
        plan.connect(fake1, filter1);
        plan.connect(fake1, filter2);

        try {
            plan.removeAndReconnect(fake1);
        } catch (FrontendException e) {
            assertEquals(2256, e.getErrorCode());
            throw e;
        }
    }

    @Test
    public void testInsertBetween1() throws FrontendException {
        // single input
        SillyPlan plan = new SillyPlan();
        Operator load1 = new SillyOperator("load1", plan);
        Operator filter1 = new SillyOperator("filter1", plan);
        plan.add(load1);
        plan.add(filter1);
        plan.connect(load1, filter1);

        Operator filter2 = new SillyOperator("filter2", plan);
        plan.insertBetween(load1, filter2, filter1);

        List<Operator> succs = plan.getSuccessors(filter2);
        assertEquals(1, succs.size());
        assertTrue(succs.contains(filter1));

        List<Operator> preds = plan.getPredecessors(filter2);
        assertEquals(1, preds.size());
        assertTrue(preds.contains(load1));
    }
}