test/org/apache/pig/test/TestPOBinCond.java - pig - Git at Google

 /*
  * 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.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;

 import java.util.Iterator;
 import java.util.Random;

 import org.apache.pig.backend.executionengine.ExecException;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.ConstantExpression;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.EqualToExpr;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POBinCond;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POProject;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
 import org.apache.pig.data.BagFactory;
 import org.apache.pig.data.DataBag;
 import org.apache.pig.data.DataType;
 import org.apache.pig.data.DefaultBagFactory;
 import org.apache.pig.data.Tuple;
 import org.apache.pig.data.TupleFactory;
 import org.apache.pig.impl.plan.OperatorKey;
 import org.apache.pig.impl.plan.PlanException;
 import org.apache.pig.test.utils.GenPhyOp;
 import org.joda.time.DateTime;
 import org.junit.Before;
 import org.junit.Test;

 public class TestPOBinCond {
     Random r = new Random(42L);
     DataBag bag = BagFactory.getInstance().newDefaultBag();
     DataBag bagDefault = BagFactory.getInstance().newDefaultBag();
     DataBag bagWithNull = BagFactory.getInstance().newDefaultBag();
     DataBag bagWithBoolean = BagFactory.getInstance().newDefaultBag();
     DataBag bagWithBooleanAndNull = BagFactory.getInstance().newDefaultBag();

     final int MAX = 10;

     /***
      *  POBinCondition tests
      *
      *  (r1, 1, 0 )
      *  (r2, 1, 0 )
      *  (r3, 1, 0 )
      *  ...
      *  (rn, 1, 0 )
      *
      *   where r is a random number ( r1 .. rn )
      *
      *   The POBinCondition to test is:  Integer(result)= ( r == 1 )? Ingeger(1), Ingeger(0);
      *   but the condition can be of any datatype: Interger, Float, Double...
      *
      * @throws ExecException
      */

     @Before
     public void setUp() {
         //default bag as shown above
         for(int i = 0; i < 10; i ++) {
             Tuple t = TupleFactory.getInstance().newTuple();
             t.append(r.nextInt(2));
             t.append(1);
             t.append(0);
             bagDefault.add(t);
         }

         //same as default bag but contains nulls
         for(int i = 0; i < 10; i ++) {
             Tuple t = TupleFactory.getInstance().newTuple();
             if (r.nextInt(4)%3 == 0){
                 t.append(null);
             }else{
                 t.append(r.nextInt(2));
             }
             t.append(1);
             t.append(0);
             bagWithNull.add(t);

         }

         //r is a boolean
         for(int i = 0; i < 10; i ++) {
             Tuple t = TupleFactory.getInstance().newTuple();
             if (r.nextInt(2)%2 == 0 ){
                     t.append(true);
             } else  {
                     t.append(false);
             }
             t.append(1);
             t.append(0);
             bagWithBoolean.add(t);
         }

         //r is a boolean with nulls
         for(int i = 0; i < 10; i ++) {

             Tuple t = TupleFactory.getInstance().newTuple();
             if (r.nextInt(3)%2 == 0){

                  t.append(null);

             }else{

                    if (r.nextInt(2)%2 == 0 ){
                     t.append(true);
                 } else {
                     t.append(false);
                 }

             }
             t.append(1);
             t.append(0);
             bagWithBooleanAndNull.add(t);

         }


     }

     /* ORIGINAL TEST
     public void testPOBinCond() throws ExecException, PlanException {
         ConstantExpression rt = (ConstantExpression) GenPhyOp.exprConst();
         rt.setValue(1);
         rt.setResultType(DataType.INTEGER);

         POProject prj1 = GenPhyOp.exprProject();
         prj1.setColumn(0);
         prj1.setResultType(DataType.INTEGER);

         EqualToExpr equal = (EqualToExpr) GenPhyOp.compEqualToExpr();
         equal.setLhs(prj1);
         equal.setRhs(rt);
         equal.setOperandType(DataType.INTEGER);

         POProject prjLhs = GenPhyOp.exprProject();
         prjLhs.setResultType(DataType.INTEGER);
         prjLhs.setColumn(1);

         POProject prjRhs = GenPhyOp.exprProject();
         prjRhs.setResultType(DataType.INTEGER);
         prjRhs.setColumn(2);

         POBinCond op = new POBinCond(new OperatorKey("", r.nextLong()), -1, equal, prjLhs, prjRhs);
         op.setResultType(DataType.INTEGER);

         PhysicalPlan plan = new PhysicalPlan();
         plan.add(op);
         plan.add(prjLhs);
         plan.add(prjRhs);
         plan.add(equal);
         plan.connect(equal, op);
         plan.connect(prjLhs, op);
         plan.connect(prjRhs, op);

         plan.add(prj1);
         plan.add(rt);
         plan.connect(prj1, equal);
         plan.connect(rt, equal);

         for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
             Tuple t = it.next();
             plan.attachInput(t);
             Integer i = (Integer) t.get(0);
             assertEquals(1, i | (Integer)op.getNext(i).result);
         }


     }
   */

     @Test
     public void testPOBinCondWithBoolean() throws ExecException, PlanException {
         bag = getBag(DataType.BOOLEAN);
         TestPoBinCondHelper testHelper = new TestPoBinCondHelper(
                 DataType.BOOLEAN, Boolean.TRUE);

         for (Iterator<Tuple> it = bag.iterator(); it.hasNext();) {
             Tuple t = it.next();
             testHelper.getPlan().attachInput(t);
             Boolean value = (Boolean) t.get(0);
             int expected = (value.booleanValue() == true) ? 1 : 0;
             Integer dummy = new Integer(0);
             Integer result = (Integer) testHelper.getOperator().getNextInteger().result;
             int actual = result.intValue();
             assertEquals(expected, actual);
         }
     }

     @Test
     public void testPOBinCondWithInteger() throws  ExecException, PlanException {

         bag= getBag(DataType.INTEGER);
         TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.INTEGER, new Integer(1) );

         for (Tuple t : bag) {
             testHelper.getPlan().attachInput(t);
             Integer value = (Integer) t.get(0);
             int expected = (value.intValue() == 1)? 1:0 ;
             Integer result=(Integer)testHelper.getOperator().getNextInteger().result;
             int actual = result.intValue();
             assertEquals( expected, actual );
         }

     }

     @Test
     public void testPOBinCondWithLong() throws  ExecException, PlanException {
         bag= getBag(DataType.LONG);
         TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.LONG, new Long(1L) );

         for (Tuple t : bag) {
             testHelper.getPlan().attachInput(t);
             Long value = (Long) t.get(0);
             int expected = (value.longValue() == 1L )? 1:0 ;
             Integer dummy = new Integer(0);
             Integer result=(Integer)testHelper.getOperator().getNextInteger().result;
             int actual = result.intValue();
             assertEquals( expected, actual );
         }
     }

     @Test
     public void testPOBinCondWithFloat() throws  ExecException, PlanException {
        bag= getBag(DataType.FLOAT);
        TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.FLOAT, new Float(1.0f) );

        for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
            Tuple t = it.next();
            testHelper.getPlan().attachInput(t);
            Float value = (Float) t.get(0);
            int expected = (value.floatValue() == 1.0f )? 1:0 ;
            Integer dummy = new Integer(0);
            Integer result=(Integer)testHelper.getOperator().getNextInteger().result;
            int actual = result.intValue();
            assertEquals( expected, actual );
         }

     }

     @Test
     public void testPOBinCondWithDouble() throws  ExecException, PlanException {
        bag= getBag(DataType.DOUBLE);
        TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.DOUBLE, new Double(1.0) );

        for (Tuple t : bag) {
            testHelper.getPlan().attachInput(t);
            Double value = (Double) t.get(0);
            int expected = (value.doubleValue() == 1.0 )? 1:0 ;
            Integer dummy = new Integer(0);
            Integer result=(Integer)testHelper.getOperator().getNextInteger().result;
            int actual = result.intValue();
            assertEquals( expected, actual );
         }

     }

     @Test
     public void testPOBinCondWithDateTime() throws  ExecException, PlanException {
         bag= getBag(DataType.DATETIME);
         TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.DATETIME, new DateTime(1L) );

         for (Tuple t : bag) {
            testHelper.getPlan().attachInput(t);
            DateTime value = (DateTime) t.get(0);
            int expected = value.equals(new DateTime(1L))? 1:0 ;
            Integer dummy = new Integer(0);
            Integer result=(Integer)testHelper.getOperator().getNextInteger().result;
            int actual = result.intValue();
            assertEquals( expected, actual );
         }
     }

     @Test
     public void testPOBinCondIntWithNull() throws  ExecException, PlanException {
        bag= getBagWithNulls(DataType.INTEGER);
        TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.INTEGER, new Integer(1) );

        for (Tuple t : bag) {
             testHelper.getPlan().attachInput(t);
             Integer value = null;
             Integer result;

             if (t.get(0) != null) {
                 value = (Integer) t.get(0);
                 result = (Integer) testHelper.getOperator().getNextInteger().result;
             } else {
                 result = (Integer) testHelper.getOperator().getNextInteger().result;
             }
             int actual;
             if (value != null) {
                 int expected = (value.intValue() == 1) ? 1 : 0;
                 actual = result.intValue();
                 assertEquals(expected, actual);
             } else {
                 assertNull(result);
             }
        }
    }

     @Test
     public void testPOBinCondLongWithNull() throws  ExecException, PlanException {
        bag= getBagWithNulls(DataType.LONG);
        TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.LONG, new Long(1L) );

        for (Tuple t : bag) {
            testHelper.getPlan().attachInput(t);

            Long value=null;
            if ( t.get(0)!=null){
                value = (Long) t.get(0);
            }
            Integer dummy = new Integer(0);
            Integer result=(Integer)testHelper.getOperator().getNextInteger().result;
            int expected;
            int actual;
            if ( value!=null ) {
                expected=(value.intValue() == 1)? 1:0 ;
                actual  = result.intValue();
                assertEquals( expected, actual );
            } else {
                assertNull(result);
            }
        }
     }

     @Test
     public void testPOBinCondDoubleWithNull() throws  ExecException, PlanException {
        bag= getBagWithNulls(DataType.DOUBLE);
        TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.DOUBLE, new Double(1.0) );

        for (Tuple t : bag) {
            testHelper.getPlan().attachInput(t);

            Double value=null;
            if ( t.get(0)!=null){
                value = (Double) t.get(0);
            }
            Integer dummy = new Integer(0);
            Integer result=(Integer)testHelper.getOperator().getNextInteger().result;

            int expected;
            int actual;
            if ( value!=null ) {
                expected=(value.intValue() == 1)? 1:0 ;
                actual  = result.intValue();
                assertEquals( expected, actual );
            } else {
                assertNull(result);
            }
        }
     }

     @Test
     public void testPOBinCondDateTimeWithNull() throws  ExecException, PlanException {
         bag= getBagWithNulls(DataType.DATETIME);
         TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.DATETIME, new DateTime(1L) );

         for (Tuple t : bag) {
            testHelper.getPlan().attachInput(t);

            DateTime value=null;
            if ( t.get(0)!=null){
                value = (DateTime) t.get(0);
            }
            Integer dummy = new Integer(0);
            Integer result=(Integer)testHelper.getOperator().getNextInteger().result;
            int expected;
            int actual;
            if ( value!=null ) {
                expected=value.equals(new DateTime(1L))? 1:0 ;
                actual  = result.intValue();
                assertEquals( expected, actual );
            } else {
                assertNull(result);
            }
        }
     }

     protected class TestPoBinCondHelper {
         PhysicalPlan plan= null;
         POBinCond op= null;

         public <U> TestPoBinCondHelper(   byte type,  U value  )  throws  ExecException, PlanException {

             ConstantExpression rt = (ConstantExpression) GenPhyOp.exprConst();
             rt.setValue(value);
             rt.setResultType(type);

             POProject prj1 = GenPhyOp.exprProject();
             prj1.setColumn(0);
             prj1.setResultType(type);


             EqualToExpr equal = (EqualToExpr) GenPhyOp.compEqualToExpr();
             equal.setLhs(prj1);
             equal.setRhs(rt);
             equal.setOperandType(type);

             POProject prjLhs = GenPhyOp.exprProject();
             prjLhs.setResultType(DataType.INTEGER);
             prjLhs.setColumn(1);

             POProject prjRhs =prjRhs = GenPhyOp.exprProject();
             prjRhs.setResultType(DataType.INTEGER);
             prjRhs.setColumn(2);

             op = new POBinCond(new OperatorKey("", r.nextLong()), -1, equal, prjLhs, prjRhs);
             op.setResultType(DataType.INTEGER);

             plan= new PhysicalPlan();
             plan.add(op);
             plan.add(prjLhs);
             plan.add(prjRhs);
             plan.add(equal);
             plan.connect(equal, op);
             plan.connect(prjLhs, op);
             plan.connect(prjRhs, op);

             plan.add(prj1);
             plan.add(rt);
             plan.connect(prj1, equal);
             plan.connect(rt, equal);

            // File tmpFile = File.createTempFile("test", ".txt" );
            //PrintStream ps = new PrintStream(new FileOutputStream(tmpFile));
            //plan.explain(ps);
            //ps.close();

         }

         public PhysicalPlan getPlan(){
             return plan;
         }

         public POBinCond getOperator(){
             return op;
         }
     }

     private DataBag getBag(byte type) {
         DataBag bag = DefaultBagFactory.getInstance().newDefaultBag();
         for(int i = 0; i < 10; i ++) {
             Tuple t = TupleFactory.getInstance().newTuple();
             switch(type) {
                 case DataType.BOOLEAN:
                     t.append(r.nextBoolean());
                     break;
                 case DataType.INTEGER:
                     t.append(r.nextInt(2));
                     break;
                 case DataType.LONG:
                     t.append(r.nextLong() % 2L);
                     break;
                 case DataType.FLOAT:
                     t.append((i % 2 == 0 ? 1.0f : 0.0f));
                     break;
                 case DataType.DOUBLE:
                     t.append((i % 2 == 0 ? 1.0 : 0.0));
                     break;
                 case DataType.DATETIME:
                     t.append(new DateTime(r.nextLong() % 2L));
                     break;
             }
             t.append(1);
             t.append(0);
             bag.add(t);
         }
         return bag;
     }

     private DataBag getBagWithNulls(byte type) {
         DataBag bag = DefaultBagFactory.getInstance().newDefaultBag();
         for(int i = 0; i < 10; i ++) {
             Tuple t = TupleFactory.getInstance().newTuple();
             if (r.nextInt(4)%3 == 0){
                 t.append(null);
             }else{
                 switch(type) {
                     case DataType.BOOLEAN:
                         t.append(r.nextBoolean());
                         break;
                     case DataType.INTEGER:
                         t.append(r.nextInt(2));
                         break;
                     case DataType.LONG:
                         t.append(r.nextLong() % 2L);
                         break;
                     case DataType.FLOAT:
                         t.append( (i % 2 == 0 ? 1.0f : 0.0f));
                         break;
                     case DataType.DOUBLE:
                         t.append( (i % 2 == 0 ? 1.0 : 0.0));
                         break;
                     case DataType.DATETIME:
                         t.append(new DateTime(r.nextLong() % 2L));
                         break;
                 }
             }
             t.append(1);
             t.append(0);
             bag.add(t);
         }
         return bag;
     }
 }
	/*
	* 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.junit.Assert.assertEquals;
	import static org.junit.Assert.assertNull;

	import java.util.Iterator;
	import java.util.Random;

	import org.apache.pig.backend.executionengine.ExecException;
	import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.ConstantExpression;
	import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.EqualToExpr;
	import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POBinCond;
	import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POProject;
	import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
	import org.apache.pig.data.BagFactory;
	import org.apache.pig.data.DataBag;
	import org.apache.pig.data.DataType;
	import org.apache.pig.data.DefaultBagFactory;
	import org.apache.pig.data.Tuple;
	import org.apache.pig.data.TupleFactory;
	import org.apache.pig.impl.plan.OperatorKey;
	import org.apache.pig.impl.plan.PlanException;
	import org.apache.pig.test.utils.GenPhyOp;
	import org.joda.time.DateTime;
	import org.junit.Before;
	import org.junit.Test;

	public class TestPOBinCond {
	Random r = new Random(42L);
	DataBag bag = BagFactory.getInstance().newDefaultBag();
	DataBag bagDefault = BagFactory.getInstance().newDefaultBag();
	DataBag bagWithNull = BagFactory.getInstance().newDefaultBag();
	DataBag bagWithBoolean = BagFactory.getInstance().newDefaultBag();
	DataBag bagWithBooleanAndNull = BagFactory.getInstance().newDefaultBag();

	final int MAX = 10;

	/***
	* POBinCondition tests
	*
	* (r1, 1, 0 )
	* (r2, 1, 0 )
	* (r3, 1, 0 )
	* ...
	* (rn, 1, 0 )
	*
	* where r is a random number ( r1 .. rn )
	*
	* The POBinCondition to test is: Integer(result)= ( r == 1 )? Ingeger(1), Ingeger(0);
	* but the condition can be of any datatype: Interger, Float, Double...
	*
	* @throws ExecException
	*/

	@Before
	public void setUp() {
	//default bag as shown above
	for(int i = 0; i < 10; i ++) {
	Tuple t = TupleFactory.getInstance().newTuple();
	t.append(r.nextInt(2));
	t.append(1);
	t.append(0);
	bagDefault.add(t);
	}

	//same as default bag but contains nulls
	for(int i = 0; i < 10; i ++) {
	Tuple t = TupleFactory.getInstance().newTuple();
	if (r.nextInt(4)%3 == 0){
	t.append(null);
	}else{
	t.append(r.nextInt(2));
	}
	t.append(1);
	t.append(0);
	bagWithNull.add(t);

	}

	//r is a boolean
	for(int i = 0; i < 10; i ++) {
	Tuple t = TupleFactory.getInstance().newTuple();
	if (r.nextInt(2)%2 == 0 ){
	t.append(true);
	} else {
	t.append(false);
	}
	t.append(1);
	t.append(0);
	bagWithBoolean.add(t);
	}

	//r is a boolean with nulls
	for(int i = 0; i < 10; i ++) {

	Tuple t = TupleFactory.getInstance().newTuple();
	if (r.nextInt(3)%2 == 0){

	t.append(null);

	}else{

	if (r.nextInt(2)%2 == 0 ){
	t.append(true);
	} else {
	t.append(false);
	}

	}
	t.append(1);
	t.append(0);
	bagWithBooleanAndNull.add(t);

	}


	}

	/* ORIGINAL TEST
	public void testPOBinCond() throws ExecException, PlanException {
	ConstantExpression rt = (ConstantExpression) GenPhyOp.exprConst();
	rt.setValue(1);
	rt.setResultType(DataType.INTEGER);

	POProject prj1 = GenPhyOp.exprProject();
	prj1.setColumn(0);
	prj1.setResultType(DataType.INTEGER);

	EqualToExpr equal = (EqualToExpr) GenPhyOp.compEqualToExpr();
	equal.setLhs(prj1);
	equal.setRhs(rt);
	equal.setOperandType(DataType.INTEGER);

	POProject prjLhs = GenPhyOp.exprProject();
	prjLhs.setResultType(DataType.INTEGER);
	prjLhs.setColumn(1);

	POProject prjRhs = GenPhyOp.exprProject();
	prjRhs.setResultType(DataType.INTEGER);
	prjRhs.setColumn(2);

	POBinCond op = new POBinCond(new OperatorKey("", r.nextLong()), -1, equal, prjLhs, prjRhs);
	op.setResultType(DataType.INTEGER);

	PhysicalPlan plan = new PhysicalPlan();
	plan.add(op);
	plan.add(prjLhs);
	plan.add(prjRhs);
	plan.add(equal);
	plan.connect(equal, op);
	plan.connect(prjLhs, op);
	plan.connect(prjRhs, op);

	plan.add(prj1);
	plan.add(rt);
	plan.connect(prj1, equal);
	plan.connect(rt, equal);

	for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
	Tuple t = it.next();
	plan.attachInput(t);
	Integer i = (Integer) t.get(0);
	assertEquals(1, i \| (Integer)op.getNext(i).result);
	}


	}
	*/

	@Test
	public void testPOBinCondWithBoolean() throws ExecException, PlanException {
	bag = getBag(DataType.BOOLEAN);
	TestPoBinCondHelper testHelper = new TestPoBinCondHelper(
	DataType.BOOLEAN, Boolean.TRUE);

	for (Iterator<Tuple> it = bag.iterator(); it.hasNext();) {
	Tuple t = it.next();
	testHelper.getPlan().attachInput(t);
	Boolean value = (Boolean) t.get(0);
	int expected = (value.booleanValue() == true) ? 1 : 0;
	Integer dummy = new Integer(0);
	Integer result = (Integer) testHelper.getOperator().getNextInteger().result;
	int actual = result.intValue();
	assertEquals(expected, actual);
	}
	}

	@Test
	public void testPOBinCondWithInteger() throws ExecException, PlanException {

	bag= getBag(DataType.INTEGER);
	TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.INTEGER, new Integer(1) );

	for (Tuple t : bag) {
	testHelper.getPlan().attachInput(t);
	Integer value = (Integer) t.get(0);
	int expected = (value.intValue() == 1)? 1:0 ;
	Integer result=(Integer)testHelper.getOperator().getNextInteger().result;
	int actual = result.intValue();
	assertEquals( expected, actual );
	}

	}

	@Test
	public void testPOBinCondWithLong() throws ExecException, PlanException {
	bag= getBag(DataType.LONG);
	TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.LONG, new Long(1L) );

	for (Tuple t : bag) {
	testHelper.getPlan().attachInput(t);
	Long value = (Long) t.get(0);
	int expected = (value.longValue() == 1L )? 1:0 ;
	Integer dummy = new Integer(0);
	Integer result=(Integer)testHelper.getOperator().getNextInteger().result;
	int actual = result.intValue();
	assertEquals( expected, actual );
	}
	}

	@Test
	public void testPOBinCondWithFloat() throws ExecException, PlanException {
	bag= getBag(DataType.FLOAT);
	TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.FLOAT, new Float(1.0f) );

	for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
	Tuple t = it.next();
	testHelper.getPlan().attachInput(t);
	Float value = (Float) t.get(0);
	int expected = (value.floatValue() == 1.0f )? 1:0 ;
	Integer dummy = new Integer(0);
	Integer result=(Integer)testHelper.getOperator().getNextInteger().result;
	int actual = result.intValue();
	assertEquals( expected, actual );
	}

	}

	@Test
	public void testPOBinCondWithDouble() throws ExecException, PlanException {
	bag= getBag(DataType.DOUBLE);
	TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.DOUBLE, new Double(1.0) );

	for (Tuple t : bag) {
	testHelper.getPlan().attachInput(t);
	Double value = (Double) t.get(0);
	int expected = (value.doubleValue() == 1.0 )? 1:0 ;
	Integer dummy = new Integer(0);
	Integer result=(Integer)testHelper.getOperator().getNextInteger().result;
	int actual = result.intValue();
	assertEquals( expected, actual );
	}

	}

	@Test
	public void testPOBinCondWithDateTime() throws ExecException, PlanException {
	bag= getBag(DataType.DATETIME);
	TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.DATETIME, new DateTime(1L) );

	for (Tuple t : bag) {
	testHelper.getPlan().attachInput(t);
	DateTime value = (DateTime) t.get(0);
	int expected = value.equals(new DateTime(1L))? 1:0 ;
	Integer dummy = new Integer(0);
	Integer result=(Integer)testHelper.getOperator().getNextInteger().result;
	int actual = result.intValue();
	assertEquals( expected, actual );
	}
	}

	@Test
	public void testPOBinCondIntWithNull() throws ExecException, PlanException {
	bag= getBagWithNulls(DataType.INTEGER);
	TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.INTEGER, new Integer(1) );

	for (Tuple t : bag) {
	testHelper.getPlan().attachInput(t);
	Integer value = null;
	Integer result;

	if (t.get(0) != null) {
	value = (Integer) t.get(0);
	result = (Integer) testHelper.getOperator().getNextInteger().result;
	} else {
	result = (Integer) testHelper.getOperator().getNextInteger().result;
	}
	int actual;
	if (value != null) {
	int expected = (value.intValue() == 1) ? 1 : 0;
	actual = result.intValue();
	assertEquals(expected, actual);
	} else {
	assertNull(result);
	}
	}
	}

	@Test
	public void testPOBinCondLongWithNull() throws ExecException, PlanException {
	bag= getBagWithNulls(DataType.LONG);
	TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.LONG, new Long(1L) );

	for (Tuple t : bag) {
	testHelper.getPlan().attachInput(t);

	Long value=null;
	if ( t.get(0)!=null){
	value = (Long) t.get(0);
	}
	Integer dummy = new Integer(0);
	Integer result=(Integer)testHelper.getOperator().getNextInteger().result;
	int expected;
	int actual;
	if ( value!=null ) {
	expected=(value.intValue() == 1)? 1:0 ;
	actual = result.intValue();
	assertEquals( expected, actual );
	} else {
	assertNull(result);
	}
	}
	}

	@Test
	public void testPOBinCondDoubleWithNull() throws ExecException, PlanException {
	bag= getBagWithNulls(DataType.DOUBLE);
	TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.DOUBLE, new Double(1.0) );

	for (Tuple t : bag) {
	testHelper.getPlan().attachInput(t);

	Double value=null;
	if ( t.get(0)!=null){
	value = (Double) t.get(0);
	}
	Integer dummy = new Integer(0);
	Integer result=(Integer)testHelper.getOperator().getNextInteger().result;

	int expected;
	int actual;
	if ( value!=null ) {
	expected=(value.intValue() == 1)? 1:0 ;
	actual = result.intValue();
	assertEquals( expected, actual );
	} else {
	assertNull(result);
	}
	}
	}

	@Test
	public void testPOBinCondDateTimeWithNull() throws ExecException, PlanException {
	bag= getBagWithNulls(DataType.DATETIME);
	TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.DATETIME, new DateTime(1L) );

	for (Tuple t : bag) {
	testHelper.getPlan().attachInput(t);

	DateTime value=null;
	if ( t.get(0)!=null){
	value = (DateTime) t.get(0);
	}
	Integer dummy = new Integer(0);
	Integer result=(Integer)testHelper.getOperator().getNextInteger().result;
	int expected;
	int actual;
	if ( value!=null ) {
	expected=value.equals(new DateTime(1L))? 1:0 ;
	actual = result.intValue();
	assertEquals( expected, actual );
	} else {
	assertNull(result);
	}
	}
	}

	protected class TestPoBinCondHelper {
	PhysicalPlan plan= null;
	POBinCond op= null;

	public <U> TestPoBinCondHelper( byte type, U value ) throws ExecException, PlanException {

	ConstantExpression rt = (ConstantExpression) GenPhyOp.exprConst();
	rt.setValue(value);
	rt.setResultType(type);

	POProject prj1 = GenPhyOp.exprProject();
	prj1.setColumn(0);
	prj1.setResultType(type);


	EqualToExpr equal = (EqualToExpr) GenPhyOp.compEqualToExpr();
	equal.setLhs(prj1);
	equal.setRhs(rt);
	equal.setOperandType(type);

	POProject prjLhs = GenPhyOp.exprProject();
	prjLhs.setResultType(DataType.INTEGER);
	prjLhs.setColumn(1);

	POProject prjRhs =prjRhs = GenPhyOp.exprProject();
	prjRhs.setResultType(DataType.INTEGER);
	prjRhs.setColumn(2);

	op = new POBinCond(new OperatorKey("", r.nextLong()), -1, equal, prjLhs, prjRhs);
	op.setResultType(DataType.INTEGER);

	plan= new PhysicalPlan();
	plan.add(op);
	plan.add(prjLhs);
	plan.add(prjRhs);
	plan.add(equal);
	plan.connect(equal, op);
	plan.connect(prjLhs, op);
	plan.connect(prjRhs, op);

	plan.add(prj1);
	plan.add(rt);
	plan.connect(prj1, equal);
	plan.connect(rt, equal);

	// File tmpFile = File.createTempFile("test", ".txt" );
	//PrintStream ps = new PrintStream(new FileOutputStream(tmpFile));
	//plan.explain(ps);
	//ps.close();

	}

	public PhysicalPlan getPlan(){
	return plan;
	}

	public POBinCond getOperator(){
	return op;
	}
	}

	private DataBag getBag(byte type) {
	DataBag bag = DefaultBagFactory.getInstance().newDefaultBag();
	for(int i = 0; i < 10; i ++) {
	Tuple t = TupleFactory.getInstance().newTuple();
	switch(type) {
	case DataType.BOOLEAN:
	t.append(r.nextBoolean());
	break;
	case DataType.INTEGER:
	t.append(r.nextInt(2));
	break;
	case DataType.LONG:
	t.append(r.nextLong() % 2L);
	break;
	case DataType.FLOAT:
	t.append((i % 2 == 0 ? 1.0f : 0.0f));
	break;
	case DataType.DOUBLE:
	t.append((i % 2 == 0 ? 1.0 : 0.0));
	break;
	case DataType.DATETIME:
	t.append(new DateTime(r.nextLong() % 2L));
	break;
	}
	t.append(1);
	t.append(0);
	bag.add(t);
	}
	return bag;
	}

	private DataBag getBagWithNulls(byte type) {
	DataBag bag = DefaultBagFactory.getInstance().newDefaultBag();
	for(int i = 0; i < 10; i ++) {
	Tuple t = TupleFactory.getInstance().newTuple();
	if (r.nextInt(4)%3 == 0){
	t.append(null);
	}else{
	switch(type) {
	case DataType.BOOLEAN:
	t.append(r.nextBoolean());
	break;
	case DataType.INTEGER:
	t.append(r.nextInt(2));
	break;
	case DataType.LONG:
	t.append(r.nextLong() % 2L);
	break;
	case DataType.FLOAT:
	t.append( (i % 2 == 0 ? 1.0f : 0.0f));
	break;
	case DataType.DOUBLE:
	t.append( (i % 2 == 0 ? 1.0 : 0.0));
	break;
	case DataType.DATETIME:
	t.append(new DateTime(r.nextLong() % 2L));
	break;
	}
	}
	t.append(1);
	t.append(0);
	bag.add(t);
	}
	return bag;
	}
	}