String msg = "Invalid physical operators in the physical plan" ;
throw new LogicalToPhysicalTranslatorException(msg, errCode, PigException.BUG, e);
}
}
} else {
POGlobalRearrange poGlobal = new POGlobalRearrange(new OperatorKey(
scope, nodeGen.getNextNodeId(scope)), cross
.getRequestedParallelism());
poGlobal.addOriginalLocation(cross.getAlias(), cross.getLocation());
POPackage poPackage = new POPackage(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), cross.getRequestedParallelism());
poGlobal.addOriginalLocation(cross.getAlias(), cross.getLocation());
poGlobal.setCross(true);
currentPlan.add(poGlobal);
currentPlan.add(poPackage);
int count = 0;
try {
currentPlan.connect(poGlobal, poPackage);
List<Boolean> flattenLst = Arrays.asList(true, true);
for (Operator op : inputs) {
PhysicalPlan fep1 = new PhysicalPlan();
ConstantExpression ce1 = new ConstantExpression(new OperatorKey(scope, nodeGen.getNextNodeId(scope)),cross.getRequestedParallelism());
ce1.setValue(inputs.size());
ce1.setResultType(DataType.INTEGER);
fep1.add(ce1);
ConstantExpression ce2 = new ConstantExpression(new OperatorKey(scope, nodeGen.getNextNodeId(scope)),cross.getRequestedParallelism());
ce2.setValue(count);
ce2.setResultType(DataType.INTEGER);
fep1.add(ce2);
/*Tuple ce1val = TupleFactory.getInstance().newTuple(2);
ce1val.set(0,inputs.size());
ce1val.set(1,count);
ce1.setValue(ce1val);
ce1.setResultType(DataType.TUPLE);*/
POUserFunc gfc = new POUserFunc(new OperatorKey(scope, nodeGen.getNextNodeId(scope)),cross.getRequestedParallelism(),
Arrays.asList((PhysicalOperator)ce1,(PhysicalOperator)ce2), new FuncSpec(GFCross.class.getName()
+ "('" + poGlobal.getOperatorKey().toString() + "')"));
gfc.addOriginalLocation(cross.getAlias(), cross.getLocation());
gfc.setResultType(DataType.BAG);
fep1.addAsLeaf(gfc);
gfc.setInputs(Arrays.asList((PhysicalOperator)ce1,(PhysicalOperator)ce2));
/*fep1.add(gfc);