// Use the POCombiner package in the combine plan
// as it needs to act differently than the regular
// package operator.
mr.combinePlan = new PhysicalPlan();
CombinerPackager pkgr = new CombinerPackager(pack.getPkgr(),
bags);
POPackage combinePack = pack.clone();
combinePack.setPkgr(pkgr);
mr.combinePlan.add(combinePack);
mr.combinePlan.add(cfe);
mr.combinePlan.connect(combinePack, cfe);
// No need to connect projections in cfe to cp, because
// PigCombiner directly attaches output from package to
// root of remaining plan.
POLocalRearrange mlr = getNewRearrange(rearrange);
POPartialAgg mapAgg = null;
if(doMapAgg){
mapAgg = createPartialAgg(cfe);
}
// A specialized local rearrange operator will replace
// the normal local rearrange in the map plan. This behaves
// like the regular local rearrange in the getNext()
// as far as getting its input and constructing the
// "key" out of the input. It then returns a tuple with
// two fields - the key in the first position and the
// "value" inside a bag in the second position. This output
// format resembles the format out of a Package. This output
// will feed to the map foreach which expects this format.
// If the key field isn't in the project of the combiner or map foreach,
// it is added to the end (This is required so that we can
// set up the inner plan of the new Local Rearrange leaf in the map
// and combine plan to contain just the project of the key).
patchUpMap(mr.mapPlan, getPreCombinerLR(rearrange), mfe, mapAgg, mlr);
POLocalRearrange clr = getNewRearrange(rearrange);
mr.combinePlan.add(clr);
mr.combinePlan.connect(cfe, clr);
// Change the package operator in the reduce plan to
// be the POCombiner package, as it needs to act
// differently than the regular package operator.
pack.setPkgr(pkgr.clone());
} catch (Exception e) {
int errCode = 2018;
String msg = "Internal error. Unable to introduce the combiner for optimization.";
throw new OptimizerException(msg, errCode, PigException.BUG, e);
}