/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.pig.test;
import java.util.Iterator;
import java.util.Random;
import java.io.File;
import java.io.FileOutputStream;
import java.io.PrintStream;
import org.apache.pig.backend.executionengine.ExecException;
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.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.ConstantExpression;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.ExpressionOperator;
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.expressionOperators.EqualToExpr;
import org.apache.pig.impl.plan.PlanException;
import org.apache.pig.test.utils.GenPhyOp;
import org.joda.time.DateTime;
import org.junit.Before;
import junit.framework.TestCase;
public class TestPOBinCond extends TestCase {
Random r = new Random();
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
@Override
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);
// System.out.println(t + " " + op.getNext(i).result.toString());
}
}
*/
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().getNext(dummy).result;
int actual = result.intValue();
assertEquals(expected, actual);
}
}
public void testPOBinCondWithInteger() throws ExecException, PlanException {
bag= getBag(DataType.INTEGER);
TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.INTEGER, new Integer(1) );
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
testHelper.getPlan().attachInput(t);
Integer value = (Integer) t.get(0);
int expected = (value.intValue() == 1)? 1:0 ;
Integer result=(Integer)testHelper.getOperator().getNext(value).result;
int actual = result.intValue();
assertEquals( expected, actual );
}
}
public void testPOBinCondWithLong() throws ExecException, PlanException {
bag= getBag(DataType.LONG);
TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.LONG, new Long(1L) );
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
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().getNext(dummy).result;
int actual = result.intValue();
assertEquals( expected, actual );
}
}
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().getNext(dummy).result;
int actual = result.intValue();
assertEquals( expected, actual );
}
}
public void testPOBinCondWithDouble() throws ExecException, PlanException {
bag= getBag(DataType.DOUBLE);
TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.DOUBLE, new Double(1.0) );
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
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().getNext(dummy).result;
int actual = result.intValue();
assertEquals( expected, actual );
}
}
public void testPOBinCondWithDateTime() throws ExecException, PlanException {
bag= getBag(DataType.DATETIME);
TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.DATETIME, new DateTime(1L) );
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
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().getNext(dummy).result;
int actual = result.intValue();
assertEquals( expected, actual );
}
}
public void testPOBinCondIntWithNull() throws ExecException, PlanException {
bag= getBagWithNulls(DataType.INTEGER);
TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.INTEGER, new Integer(1) );
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
testHelper.getPlan().attachInput(t);
Integer value = null;
Integer result;
if (t.get(0) != null) {
value = (Integer) t.get(0);
result = (Integer) testHelper.getOperator().getNext(value).result;
} else {
result = (Integer) testHelper.getOperator().getNext(
(Integer) null).result;
}
int actual;
if (value != null) {
int expected = (value.intValue() == 1) ? 1 : 0;
actual = result.intValue();
assertEquals(expected, actual);
} else {
assertEquals(null, result);
}
}
}
public void testPOBinCondLongWithNull() throws ExecException, PlanException {
bag= getBagWithNulls(DataType.LONG);
TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.LONG, new Long(1L) );
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
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().getNext(dummy).result;
int expected;
int actual;
if ( value!=null ) {
expected=(value.intValue() == 1)? 1:0 ;
actual = result.intValue();
assertEquals( expected, actual );
} else {
assertEquals( null, result );
}
}
}
public void testPOBinCondDoubleWithNull() throws ExecException, PlanException {
bag= getBagWithNulls(DataType.DOUBLE);
TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.DOUBLE, new Double(1.0) );
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
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().getNext(dummy).result;
int expected;
int actual;
if ( value!=null ) {
expected=(value.intValue() == 1)? 1:0 ;
actual = result.intValue();
assertEquals( expected, actual );
} else {
assertEquals( null, result );
}
}
}
public void testPOBinCondDateTimeWithNull() throws ExecException, PlanException {
bag= getBagWithNulls(DataType.DATETIME);
TestPoBinCondHelper testHelper= new TestPoBinCondHelper(DataType.DATETIME, new DateTime(1L) );
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
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().getNext(dummy).result;
int expected;
int actual;
if ( value!=null ) {
expected=value.equals(new DateTime(1L))? 1:0 ;
actual = result.intValue();
assertEquals( expected, actual );
} else {
assertEquals( null, 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;
}
}