else if (loj.getJoinType() == LOJoin.JOINTYPE.HASH){
POGlobalRearrange poGlobal = new POGlobalRearrange(new OperatorKey(
scope, nodeGen.getNextNodeId(scope)), loj
.getRequestedParallelisam());
poGlobal.setAlias(loj.getAlias());
POPackage poPackage = new POPackage(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), loj.getRequestedParallelisam());
poPackage.setAlias(loj.getAlias());
currentPlan.add(poGlobal);
currentPlan.add(poPackage);
int count = 0;
Byte type = null;
try {
currentPlan.connect(poGlobal, poPackage);
for (int i=0; i<inputs.size(); i++) {
Operator op = inputs.get(i);
List<LogicalExpressionPlan> plans =
(List<LogicalExpressionPlan>) loj.getJoinPlan(i);
POLocalRearrange physOp = new POLocalRearrange(new OperatorKey(
scope, nodeGen.getNextNodeId(scope)), loj
.getRequestedParallelisam());
List<PhysicalPlan> exprPlans = translateExpressionPlans(loj, plans);
// currentPlans.push(currentPlan);
// for (LogicalExpressionPlan lp : plans) {
// currentPlan = new PhysicalPlan();
// PlanWalker childWalker = currentWalker
// .spawnChildWalker(lp);
// pushWalker(childWalker);
// //currentWalker.walk(this);
// currentWalker.walk(
// new ExpToPhyTranslationVisitor(currentWalker.getPlan(),
// childWalker) );
// exprPlans.add(currentPlan);
// popWalker();
//
// }
// currentPlan = currentPlans.pop();
try {
physOp.setPlans(exprPlans);
} catch (PlanException pe) {
int errCode = 2071;
String msg = "Problem with setting up local rearrange's plans.";
throw new LogicalToPhysicalTranslatorException(msg, errCode, PigException.BUG, pe);
}
try {
physOp.setIndex(count++);
} catch (ExecException e1) {
int errCode = 2058;
String msg = "Unable to set index on newly create POLocalRearrange.";
throw new VisitorException(msg, errCode, PigException.BUG, e1);
}
if (plans.size() > 1) {
type = DataType.TUPLE;
physOp.setKeyType(type);
} else {
type = exprPlans.get(0).getLeaves().get(0).getResultType();
physOp.setKeyType(type);
}
physOp.setResultType(DataType.TUPLE);
currentPlan.add(physOp);
try {
currentPlan.connect(logToPhyMap.get(op), physOp);
currentPlan.connect(physOp, poGlobal);
} catch (PlanException e) {
int errCode = 2015;
String msg = "Invalid physical operators in the physical plan" ;
throw new LogicalToPhysicalTranslatorException(msg, errCode, PigException.BUG, e);
}
}
} catch (PlanException e1) {
int errCode = 2015;
String msg = "Invalid physical operators in the physical plan" ;
throw new LogicalToPhysicalTranslatorException(msg, errCode, PigException.BUG, e1);
}
poPackage.setKeyType(type);
poPackage.setResultType(DataType.TUPLE);
poPackage.setNumInps(count);
boolean[] innerFlags = loj.getInnerFlags();
poPackage.setInner(innerFlags);
List<PhysicalPlan> fePlans = new ArrayList<PhysicalPlan>();
List<Boolean> flattenLst = new ArrayList<Boolean>();
try{