Tuple t = TupleFactory.getInstance().newTuple();
t.append(r.nextLong());
bag.add(t);
}
POCast op = new POCast(new OperatorKey("", r.nextLong()), -1);
LoadFunc load = new TestLoader();
op.setLoadFSpec(new FuncSpec(load.getClass().getName()));
POProject prj = new POProject(new OperatorKey("", r.nextLong()), -1, 0);
PhysicalPlan plan = new PhysicalPlan();
plan.add(prj);
plan.add(op);
plan.connect(prj, op);
prj.setResultType(DataType.LONG);
// Plan to test when result type is ByteArray and casting is requested
// for example casting of values coming out of map lookup.
POCast opWithInputTypeAsBA = new POCast(new OperatorKey("", r.nextLong()), -1);
PhysicalPlan planToTestBACasts = constructPlan(opWithInputTypeAsBA);
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
plan.attachInput(t);
Integer i = ((Long) t.get(0)).intValue();
Result res = op.getNext(i);
if(res.returnStatus == POStatus.STATUS_OK) {
//System.out.println(res.result + " : " + i);
assertEquals(i, res.result);
}
planToTestBACasts.attachInput(t);
res = opWithInputTypeAsBA.getNext(i);
if(res.returnStatus == POStatus.STATUS_OK)
assertEquals(i, res.result);
}
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
plan.attachInput(t);
Float f = ((Long)t.get(0)).floatValue();
Result res = op.getNext(f);
if(res.returnStatus == POStatus.STATUS_OK) {
// System.out.println(res.result + " : " + f);
assertEquals(f, res.result);
}
planToTestBACasts.attachInput(t);
res = opWithInputTypeAsBA.getNext(f);
if(res.returnStatus == POStatus.STATUS_OK)
assertEquals(f, res.result);
}
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
plan.attachInput(t);
Long l = ((Long)t.get(0)).longValue();
Result res = op.getNext(l);
if(res.returnStatus == POStatus.STATUS_OK) {
//System.out.println(res.result + " : " + l);
assertEquals(l, res.result);
}
planToTestBACasts.attachInput(t);
res = opWithInputTypeAsBA.getNext(l);
if(res.returnStatus == POStatus.STATUS_OK)
assertEquals(l, res.result);
}
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
plan.attachInput(t);
Double d = ((Long)t.get(0)).doubleValue();
Result res = op.getNext(d);
if(res.returnStatus == POStatus.STATUS_OK) {
//System.out.println(res.result + " : " + d);
assertEquals(d, res.result);
}
planToTestBACasts.attachInput(t);
res = opWithInputTypeAsBA.getNext(d);
if(res.returnStatus == POStatus.STATUS_OK)
assertEquals(d, res.result);
}
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
plan.attachInput(t);
String str = ((Long)t.get(0)).toString();
Result res = op.getNext(str);
if(res.returnStatus == POStatus.STATUS_OK) {
//System.out.println(res.result + " : " + str);
assertEquals(str, res.result);
}
planToTestBACasts.attachInput(t);
res = opWithInputTypeAsBA.getNext(str);
if(res.returnStatus == POStatus.STATUS_OK)
assertEquals(str, res.result);
}
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
plan.attachInput(t);
DataByteArray dba = new DataByteArray(((Long)t.get(0)).toString().getBytes());
Result res = op.getNext(dba);
if(res.returnStatus == POStatus.STATUS_OK) {
//System.out.println(res.result + " : " + dba);
assertEquals(dba, res.result);
}
planToTestBACasts.attachInput(t);
res = opWithInputTypeAsBA.getNext(dba);
if(res.returnStatus == POStatus.STATUS_OK)
assertEquals(dba, res.result);
}
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
plan.attachInput(t);
Map map = null;
Result res = op.getNext(map);
assertEquals(POStatus.STATUS_ERR, res.returnStatus);
}
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
plan.attachInput(t);
Result res = op.getNext(t);
assertEquals(POStatus.STATUS_ERR, res.returnStatus);
}
for(Iterator<Tuple> it = bag.iterator(); it.hasNext(); ) {
Tuple t = it.next();
plan.attachInput(t);
DataBag b = null;
Result res = op.getNext(b);
assertEquals(POStatus.STATUS_ERR, res.returnStatus);
}
{
planToTestBACasts.attachInput(dummyTuple);
try{
opWithInputTypeAsBA.getNext(dummyMap);
}catch (Exception e) {
assertEquals(ExecException.class, e.getClass());
}
}
{
planToTestBACasts.attachInput(dummyTuple);
try{
opWithInputTypeAsBA.getNext(dummyTuple);
}catch (Exception e) {
assertEquals(ExecException.class, e.getClass());
}
}
{
planToTestBACasts.attachInput(dummyTuple);
try{
opWithInputTypeAsBA.getNext(dummyBag);
}catch (Exception e) {
assertEquals(ExecException.class, e.getClass());
}
}
}