You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ch...@apache.org on 2013/11/28 09:56:34 UTC
svn commit: r1546314 [3/6] - in /pig/branches/tez:
src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/
src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/plans/
src/org/apache/pig/backend/hadoop/executionengine/physicalLayer...
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRUtil.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRUtil.java?rev=1546314&r1=1546313&r2=1546314&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRUtil.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRUtil.java Thu Nov 28 08:56:33 2013
@@ -36,7 +36,7 @@ public class MRUtil {
// simpleConnectMapToReduce is a utility to end a map phase and start a reduce phase in
// a mapreduce operator:
// 1. mro only contains map plan
- // 2. need to add POLocalRearrange to end map plan, and add
+ // 2. need to add POLocalRearrange to end map plan, and add
// POPackage to start a reduce plan
// 3. POLocalRearrange/POPackage are trivial
static public void simpleConnectMapToReduce(MapReduceOper mro, String scope, NodeIdGenerator nig) throws PlanException
@@ -46,10 +46,10 @@ public class MRUtil {
prjStar.setResultType(DataType.TUPLE);
prjStar.setStar(true);
ep.add(prjStar);
-
+
List<PhysicalPlan> eps = new ArrayList<PhysicalPlan>();
eps.add(ep);
-
+
POLocalRearrange lr = new POLocalRearrange(new OperatorKey(scope,nig.getNextNodeId(scope)));
try {
lr.setIndex(0);
@@ -61,19 +61,19 @@ public class MRUtil {
lr.setKeyType(DataType.TUPLE);
lr.setPlans(eps);
lr.setResultType(DataType.TUPLE);
-
+
mro.mapPlan.addAsLeaf(lr);
-
+
POPackage pkg = new POPackage(new OperatorKey(scope,nig.getNextNodeId(scope)));
- pkg.setKeyType(DataType.TUPLE);
+ pkg.getPkgr().setKeyType(DataType.TUPLE);
pkg.setNumInps(1);
boolean[] inner = {false};
- pkg.setInner(inner);
+ pkg.getPkgr().setInner(inner);
mro.reducePlan.add(pkg);
-
+
mro.reducePlan.addAsLeaf(getPlainForEachOP(scope, nig));
}
-
+
// Get a simple POForEach: ForEach X generate flatten($1)
static public POForEach getPlainForEachOP(String scope, NodeIdGenerator nig)
{
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java?rev=1546314&r1=1546313&r2=1546314&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java Thu Nov 28 08:56:33 2013
@@ -38,8 +38,8 @@ import org.apache.hadoop.mapred.JobClien
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.JobID;
import org.apache.hadoop.mapred.RunningJob;
-import org.apache.hadoop.mapred.jobcontrol.Job;
import org.apache.hadoop.mapred.TaskReport;
+import org.apache.hadoop.mapred.jobcontrol.Job;
import org.apache.pig.ExecType;
import org.apache.pig.PigConfiguration;
import org.apache.pig.PigException;
@@ -59,7 +59,7 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.plans.POPackageAnnotator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.plans.XMLMRPrinter;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POJoinPackage;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.JoinPackager;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStore;
import org.apache.pig.backend.hadoop.executionengine.shims.HadoopShims;
import org.apache.pig.impl.PigContext;
@@ -616,7 +616,7 @@ public class MapReduceLauncher extends L
String lastInputChunkSize =
pc.getProperties().getProperty(
- "last.input.chunksize", POJoinPackage.DEFAULT_CHUNK_SIZE);
+ "last.input.chunksize", JoinPackager.DEFAULT_CHUNK_SIZE);
String prop = pc.getProperties().getProperty(PigConfiguration.PROP_NO_COMBINER);
if (!pc.inIllustrator && !("true".equals(prop))) {
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MultiQueryOptimizer.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MultiQueryOptimizer.java?rev=1546314&r1=1546313&r2=1546314&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MultiQueryOptimizer.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MultiQueryOptimizer.java Thu Nov 28 08:56:33 2013
@@ -18,10 +18,10 @@
package org.apache.pig.backend.hadoop.executionengine.mapReduceLayer;
import java.util.ArrayList;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.Iterator;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -31,14 +31,15 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.plans.MROperPlan;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.MultiQueryPackager;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PODemux;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POForEach;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLoad;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLocalRearrange;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POMultiQueryPackage;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPackage;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POSplit;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStore;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.Packager;
import org.apache.pig.data.DataType;
import org.apache.pig.impl.io.PigNullableWritable;
import org.apache.pig.impl.plan.NodeIdGenerator;
@@ -50,36 +51,36 @@ import org.apache.pig.impl.plan.optimize
import org.apache.pig.impl.util.Pair;
-/**
- * An optimizer that merges all or part splittee MapReduceOpers into
- * splitter MapReduceOper.
+/**
+ * An optimizer that merges all or part splittee MapReduceOpers into
+ * splitter MapReduceOper.
* <p>
- * The merge can produce a MROperPlan that has
+ * The merge can produce a MROperPlan that has
* fewer MapReduceOpers than MapReduceOpers in the original MROperPlan.
* <p>
- * The MRCompler generates multiple MapReduceOpers whenever it encounters
- * a split operator and connects the single splitter MapReduceOper to
- * one or more splittee MapReduceOpers using store/load operators:
+ * The MRCompler generates multiple MapReduceOpers whenever it encounters
+ * a split operator and connects the single splitter MapReduceOper to
+ * one or more splittee MapReduceOpers using store/load operators:
* <p>
* ---- POStore (in splitter) -... ----
* | | ... |
* | | ... |
* POLoad POLoad ... POLoad (in splittees)
* | | |
- * <p>
- * This optimizer merges those MapReduceOpers by replacing POLoad/POStore
- * combination with POSplit operator.
+ * <p>
+ * This optimizer merges those MapReduceOpers by replacing POLoad/POStore
+ * combination with POSplit operator.
*/
class MultiQueryOptimizer extends MROpPlanVisitor {
-
+
private Log log = LogFactory.getLog(getClass());
-
+
private NodeIdGenerator nig;
-
+
private String scope;
-
+
private boolean inIllustrator = false;
-
+
MultiQueryOptimizer(MROperPlan plan, boolean inIllustrator) {
super(plan, new ReverseDependencyOrderWalker<MapReduceOper, MROperPlan>(plan));
nig = NodeIdGenerator.getGenerator();
@@ -92,22 +93,22 @@ class MultiQueryOptimizer extends MROpPl
@Override
public void visit() throws VisitorException {
super.visit();
-
+
log.info("MR plan size after optimization: " + mPlan.size());
}
-
+
@Override
public void visitMROp(MapReduceOper mr) throws VisitorException {
-
+
if (!mr.isSplitter()) {
return;
- }
-
+ }
+
// first classify all the splittees
List<MapReduceOper> mappers = new ArrayList<MapReduceOper>();
List<MapReduceOper> multiLoadMROpers = new ArrayList<MapReduceOper>();
List<MapReduceOper> mapReducers = new ArrayList<MapReduceOper>();
-
+
List<MapReduceOper> successors = getPlan().getSuccessors(mr);
for (MapReduceOper successor : successors) {
if (successor.getUseSecondaryKey()) {
@@ -122,90 +123,76 @@ class MultiQueryOptimizer extends MROpPl
}
if (isMapOnly(successor)) {
if (isSingleLoadMapperPlan(successor.mapPlan)
- && isSinglePredecessor(successor)) {
- mappers.add(successor);
- } else {
+ && isSinglePredecessor(successor)) {
+ mappers.add(successor);
+ } else {
multiLoadMROpers.add(successor);
}
} else {
if (isSingleLoadMapperPlan(successor.mapPlan)
- && isSinglePredecessor(successor)) {
- mapReducers.add(successor);
- } else {
- multiLoadMROpers.add(successor);
+ && isSinglePredecessor(successor)) {
+ mapReducers.add(successor);
+ } else {
+ multiLoadMROpers.add(successor);
}
- }
+ }
}
-
+
int numSplittees = successors.size();
-
+
// case 1: exactly one splittee and it's map-only
- if (mappers.size() == 1 && numSplittees == 1) {
+ if (mappers.size() == 1 && numSplittees == 1) {
mergeOnlyMapperSplittee(mappers.get(0), mr);
-
log.info("Merged the only map-only splittee.");
-
- return;
+ return;
}
-
+
// case 2: exactly one splittee and it has reducer
- if (isMapOnly(mr) && mapReducers.size() == 1 && numSplittees == 1) {
+ if (isMapOnly(mr) && mapReducers.size() == 1 && numSplittees == 1) {
mergeOnlyMapReduceSplittee(mapReducers.get(0), mr);
-
log.info("Merged the only map-reduce splittee.");
-
return;
- }
-
+ }
+
int numMerges = 0;
-
- PhysicalPlan splitterPl = isMapOnly(mr) ? mr.mapPlan : mr.reducePlan;
- POStore storeOp = (POStore)splitterPl.getLeaves().get(0);
-
+
+ PhysicalPlan splitterPl = isMapOnly(mr) ? mr.mapPlan : mr.reducePlan;
+ POStore storeOp = (POStore)splitterPl.getLeaves().get(0);
POSplit splitOp = null;
-
+
// case 3: multiple splittees and at least one of them is map-only
- if (mappers.size() > 0) {
- splitOp = getSplit();
- int n = mergeAllMapOnlySplittees(mappers, mr, splitOp);
-
+ if (mappers.size() > 0) {
+ splitOp = getSplit();
+ int n = mergeAllMapOnlySplittees(mappers, mr, splitOp);
log.info("Merged " + n + " map-only splittees.");
-
- numMerges += n;
- }
-
+ numMerges += n;
+ }
+
if (mapReducers.size() > 0) {
-
boolean isMapOnly = isMapOnly(mr);
int merged = 0;
-
- // case 4: multiple splittees and at least one of them has reducer
- // and the splitter is map-only
+
+ // case 4: multiple splittees and at least one of them has reducer
+ // and the splitter is map-only
if (isMapOnly) {
-
PhysicalOperator leaf = splitterPl.getLeaves().get(0);
-
splitOp = (leaf instanceof POStore) ? getSplit() : (POSplit)leaf;
-
- merged = mergeMapReduceSplittees(mapReducers, mr, splitOp);
+ merged = mergeMapReduceSplittees(mapReducers, mr, splitOp);
}
-
+
// case 5: multiple splittees and at least one of them has reducer
// and splitter has reducer
else {
-
- merged = mergeMapReduceSplittees(mapReducers, mr);
-
+ merged = mergeMapReduceSplittees(mapReducers, mr);
}
-
+
log.info("Merged " + merged + " map-reduce splittees.");
-
- numMerges += merged;
+ numMerges += merged;
}
-
- // Finally, add original store to the split operator
+
+ // Finally, add original store to the split operator
// if there is splittee that hasn't been merged into the splitter
- if (splitOp != null
+ if (splitOp != null
&& (numMerges < numSplittees)) {
PhysicalPlan storePlan = new PhysicalPlan();
@@ -216,36 +203,36 @@ class MultiQueryOptimizer extends MROpPl
int errCode = 2129;
String msg = "Internal Error. Unable to add store to the split plan for optimization.";
throw new OptimizerException(msg, errCode, PigException.BUG, e);
- }
+ }
}
// case 6: special diamond case with trivial MR operator at the head
if (numMerges == 0 && isDiamondMROper(mr)) {
int merged = mergeDiamondMROper(mr, getPlan().getSuccessors(mr));
log.info("Merged " + merged + " diamond splitter.");
- numMerges += merged;
+ numMerges += merged;
}
-
- log.info("Merged " + numMerges + " out of total "
+
+ log.info("Merged " + numMerges + " out of total "
+ (numSplittees +1) + " MR operators.");
- }
-
+ }
+
private boolean isDiamondMROper(MapReduceOper mr) {
-
+
// We'll remove this mr as part of diamond query optimization
// only if this mr is a trivial one, that is, it's plan
- // has either two operators (load followed by store) or three operators
+ // has either two operators (load followed by store) or three operators
// (the operator between the load and store must be a foreach,
// introduced by casting operation).
- //
+ //
// We won't optimize in other cases where there're more operators
- // in the plan. Otherwise those operators world run multiple times
+ // in the plan. Otherwise those operators world run multiple times
// in the successor MR operators which may not give better
// performance.
boolean rtn = false;
if (isMapOnly(mr)) {
PhysicalPlan pl = mr.mapPlan;
- if (pl.size() == 2 || pl.size() == 3) {
+ if (pl.size() == 2 || pl.size() == 3) {
PhysicalOperator root = pl.getRoots().get(0);
PhysicalOperator leaf = pl.getLeaves().get(0);
if (root instanceof POLoad && leaf instanceof POStore) {
@@ -253,7 +240,7 @@ class MultiQueryOptimizer extends MROpPl
PhysicalOperator mid = pl.getSuccessors(root).get(0);
if (mid instanceof POForEach) {
rtn = true;
- }
+ }
} else {
rtn = true;
}
@@ -262,11 +249,11 @@ class MultiQueryOptimizer extends MROpPl
}
return rtn;
}
-
- private int mergeDiamondMROper(MapReduceOper mr, List<MapReduceOper> succs)
+
+ private int mergeDiamondMROper(MapReduceOper mr, List<MapReduceOper> succs)
throws VisitorException {
-
- // Only consider the cases where all inputs of the splittees are
+
+ // Only consider the cases where all inputs of the splittees are
// from the splitter
for (MapReduceOper succ : succs) {
List<MapReduceOper> preds = getPlan().getPredecessors(succ);
@@ -274,20 +261,20 @@ class MultiQueryOptimizer extends MROpPl
return 0;
}
}
-
+
// first, remove the store operator from the splitter
PhysicalPlan pl = mr.mapPlan;
PhysicalOperator leaf = mr.mapPlan.getLeaves().get(0);
pl.remove(leaf);
-
+
POStore store = (POStore)leaf;
String ofile = store.getSFile().getFileName();
-
+
// then connect the remaining map plan to the successor of
// each root (load) operator of the splittee
for (MapReduceOper succ : succs) {
List<PhysicalOperator> roots = succ.mapPlan.getRoots();
- ArrayList<PhysicalOperator> rootsCopy =
+ ArrayList<PhysicalOperator> rootsCopy =
new ArrayList<PhysicalOperator>(roots);
for (PhysicalOperator op : rootsCopy) {
POLoad load = (POLoad)op;
@@ -309,7 +296,7 @@ class MultiQueryOptimizer extends MROpPl
throw new OptimizerException(msg, errCode, PigException.BUG, e);
}
succ.mapPlan.remove(op);
-
+
if (inIllustrator) {
// need to remove the LOAD since data from load on temporary files can't be handled by illustrator
for (Iterator<PhysicalOperator> it = pl.iterator(); it.hasNext(); )
@@ -319,7 +306,7 @@ class MultiQueryOptimizer extends MROpPl
succ.phyToMRMap.removeKey(po);
}
}
-
+
while (!clone.isEmpty()) {
PhysicalOperator oper = clone.getLeaves().get(0);
clone.remove(oper);
@@ -331,16 +318,16 @@ class MultiQueryOptimizer extends MROpPl
int errCode = 2131;
String msg = "Internal Error. Unable to connect split plan for optimization.";
throw new OptimizerException(msg, errCode, PigException.BUG, e);
- }
+ }
}
}
-
+
// PIG-2069: LoadFunc jar does not ship to backend in MultiQuery case
if (!mr.UDFs.isEmpty()) {
succ.UDFs.addAll(mr.UDFs);
}
}
-
+
// finally, remove the splitter from the MR plan
List<MapReduceOper> mrPreds = getPlan().getPredecessors(mr);
if (mrPreds != null) {
@@ -356,36 +343,36 @@ class MultiQueryOptimizer extends MROpPl
}
}
}
-
+
getPlan().remove(mr);
-
+
return 1;
}
-
+
private void mergeOneMapPart(MapReduceOper mapper, MapReduceOper splitter)
throws VisitorException {
- PhysicalPlan splitterPl = isMapOnly(splitter) ?
+ PhysicalPlan splitterPl = isMapOnly(splitter) ?
splitter.mapPlan : splitter.reducePlan;
POStore storeOp = (POStore)splitterPl.getLeaves().get(0);
- List<PhysicalOperator> storePreds = splitterPl.getPredecessors(storeOp);
-
+ List<PhysicalOperator> storePreds = splitterPl.getPredecessors(storeOp);
+
PhysicalPlan pl = mapper.mapPlan;
- PhysicalOperator load = pl.getRoots().get(0);
+ PhysicalOperator load = pl.getRoots().get(0);
pl.remove(load);
-
+
// make a copy before removing the store operator
List<PhysicalOperator> predsCopy = new ArrayList<PhysicalOperator>(storePreds);
- splitterPl.remove(storeOp);
-
+ splitterPl.remove(storeOp);
+
try {
splitterPl.merge(pl);
} catch (PlanException e) {
int errCode = 2130;
String msg = "Internal Error. Unable to merge split plans for optimization.";
throw new OptimizerException(msg, errCode, PigException.BUG, e);
- }
-
- // connect two plans
+ }
+
+ // connect two plans
List<PhysicalOperator> roots = pl.getRoots();
for (PhysicalOperator pred : predsCopy) {
for (PhysicalOperator root : roots) {
@@ -401,13 +388,13 @@ class MultiQueryOptimizer extends MROpPl
}
}
- private void mergeOnlyMapperSplittee(MapReduceOper mapper,
+ private void mergeOnlyMapperSplittee(MapReduceOper mapper,
MapReduceOper splitter) throws VisitorException {
- mergeOneMapPart(mapper, splitter);
+ mergeOneMapPart(mapper, splitter);
removeAndReconnect(mapper, splitter);
}
-
- private void mergeOnlyMapReduceSplittee(MapReduceOper mapReducer,
+
+ private void mergeOnlyMapReduceSplittee(MapReduceOper mapReducer,
MapReduceOper splitter) throws VisitorException {
mergeOneMapPart(mapReducer, splitter);
@@ -415,108 +402,108 @@ class MultiQueryOptimizer extends MROpPl
splitter.reducePlan = mapReducer.reducePlan;
splitter.setReduceDone(true);
- removeAndReconnect(mapReducer, splitter);
+ removeAndReconnect(mapReducer, splitter);
}
-
- private int mergeAllMapOnlySplittees(List<MapReduceOper> mappers,
+
+ private int mergeAllMapOnlySplittees(List<MapReduceOper> mappers,
MapReduceOper splitter, POSplit splitOp) throws VisitorException {
-
- PhysicalPlan splitterPl = isMapOnly(splitter) ?
- splitter.mapPlan : splitter.reducePlan;
+
+ PhysicalPlan splitterPl = isMapOnly(splitter) ?
+ splitter.mapPlan : splitter.reducePlan;
PhysicalOperator storeOp = splitterPl.getLeaves().get(0);
- List<PhysicalOperator> storePreds = splitterPl.getPredecessors(storeOp);
+ List<PhysicalOperator> storePreds = splitterPl.getPredecessors(storeOp);
- // merge splitee's map plans into nested plan of
+ // merge splitee's map plans into nested plan of
// the split operator
- for (MapReduceOper mapper : mappers) {
+ for (MapReduceOper mapper : mappers) {
PhysicalPlan pl = mapper.mapPlan;
PhysicalOperator load = pl.getRoots().get(0);
- pl.remove(load);
+ pl.remove(load);
splitOp.addPlan(pl);
}
-
+
// replace store operator in the splitter with split operator
- splitOp.setInputs(storePreds);
+ splitOp.setInputs(storePreds);
try {
splitterPl.replace(storeOp, splitOp);;
} catch (PlanException e) {
int errCode = 2132;
String msg = "Internal Error. Unable to replace store with split operator for optimization.";
throw new OptimizerException(msg, errCode, PigException.BUG, e);
- }
-
+ }
+
// remove all the map-only splittees from the MROperPlan
for (MapReduceOper mapper : mappers) {
- removeAndReconnect(mapper, splitter);
+ removeAndReconnect(mapper, splitter);
}
-
+
return mappers.size();
}
-
+
private boolean isSplitteeMergeable(MapReduceOper splittee) {
-
- // cannot be global sort or limit after sort, they are
+
+ // cannot be global sort or limit after sort, they are
// using a different partitioner
if (splittee.isGlobalSort() || splittee.isLimitAfterSort()) {
log.info("Cannot merge this splittee: " +
- "it is global sort or limit after sort");
+ "it is global sort or limit after sort");
return false;
}
-
+
// check the plan leaf: only merge local rearrange or split
PhysicalOperator leaf = splittee.mapPlan.getLeaves().get(0);
- if (!(leaf instanceof POLocalRearrange) &&
+ if (!(leaf instanceof POLocalRearrange) &&
! (leaf instanceof POSplit)) {
log.info("Cannot merge this splittee: " +
- "its map plan doesn't end with LR or Split operator: "
+ "its map plan doesn't end with LR or Split operator: "
+ leaf.getClass().getName());
return false;
}
-
+
// cannot have distinct combiner, it uses a different combiner
if (splittee.needsDistinctCombiner()) {
log.info("Cannot merge this splittee: " +
- "it has distinct combiner.");
- return false;
+ "it has distinct combiner.");
+ return false;
}
-
+
return true;
}
-
+
private List<MapReduceOper> getMergeList(MapReduceOper splitter,
List<MapReduceOper> mapReducers) {
List<MapReduceOper> mergeNoCmbList = new ArrayList<MapReduceOper>();
List<MapReduceOper> mergeCmbList = new ArrayList<MapReduceOper>();
List<MapReduceOper> mergeDistList = new ArrayList<MapReduceOper>();
-
+
for (MapReduceOper mrOp : mapReducers) {
if (isSplitteeMergeable(mrOp)) {
if (mrOp.combinePlan.isEmpty()) {
mergeNoCmbList.add(mrOp);
} else {
mergeCmbList.add(mrOp);
- }
+ }
} else if (splitter.reducePlan.isEmpty()
- || splitter.needsDistinctCombiner()) {
- if (mrOp.needsDistinctCombiner()) {
+ || splitter.needsDistinctCombiner()) {
+ if (mrOp.needsDistinctCombiner()) {
mergeDistList.add(mrOp);
}
}
- }
-
+ }
+
int max = Math.max(mergeNoCmbList.size(), mergeCmbList.size());
max = Math.max(max, mergeDistList.size());
-
+
if (max == mergeDistList.size()) return mergeDistList;
else if (max == mergeNoCmbList.size()) return mergeNoCmbList;
- else return mergeCmbList;
+ else return mergeCmbList;
}
-
- private int mergeMapReduceSplittees(List<MapReduceOper> mapReducers,
+
+ private int mergeMapReduceSplittees(List<MapReduceOper> mapReducers,
MapReduceOper splitter, POSplit splitOp) throws VisitorException {
-
+
List<MapReduceOper> mergeList = getMergeList(splitter, mapReducers);
-
+
if (mergeList.size() <= 1) {
// chose one to merge, prefer the one with a combiner
@@ -529,43 +516,43 @@ class MultiQueryOptimizer extends MROpPl
}
mergeList.clear();
mergeList.add(mapReducer);
- }
-
+ }
+
if (mergeList.size() == 1) {
mergeSingleMapReduceSplittee(mergeList.get(0), splitter, splitOp);
- } else {
+ } else {
mergeAllMapReduceSplittees(mergeList, splitter, splitOp);
}
-
+
return mergeList.size();
}
- private int mergeMapReduceSplittees(List<MapReduceOper> mapReducers,
+ private int mergeMapReduceSplittees(List<MapReduceOper> mapReducers,
MapReduceOper splitter) throws VisitorException {
- // In this case the splitter has non-empty reducer so we can't merge
- // MR splittees into the splitter. What we'll do is to merge multiple
+ // In this case the splitter has non-empty reducer so we can't merge
+ // MR splittees into the splitter. What we'll do is to merge multiple
// splittees (if exists) into a new MR operator and connect it to the splitter.
-
+
List<MapReduceOper> mergeList = getMergeList(splitter, mapReducers);
-
+
if (mergeList.size() <= 1) {
// nothing to merge, just return
return 0;
- }
-
+ }
+
MapReduceOper mrOper = getMROper();
MapReduceOper splittee = mergeList.get(0);
PhysicalPlan pl = splittee.mapPlan;
POLoad load = (POLoad)pl.getRoots().get(0);
-
+
mrOper.mapPlan.add(load);
-
+
// add a dummy store operator, it'll be replaced by the split operator later.
try {
mrOper.mapPlan.addAsLeaf(getStore());
- } catch (PlanException e) {
+ } catch (PlanException e) {
int errCode = 2137;
String msg = "Internal Error. Unable to add store to the plan as leaf for optimization.";
throw new OptimizerException(msg, errCode, PigException.BUG, e);
@@ -583,10 +570,10 @@ class MultiQueryOptimizer extends MROpPl
// merger the splittees into the new MR operator
mergeAllMapReduceSplittees(mergeList, mrOper, getSplit());
-
+
return (mergeList.size() - 1);
}
-
+
private boolean hasSameMapKeyType(List<MapReduceOper> splittees) {
boolean sameKeyType = true;
for (MapReduceOper outer : splittees) {
@@ -597,29 +584,29 @@ class MultiQueryOptimizer extends MROpPl
}
}
if (!sameKeyType) break;
- }
-
+ }
+
return sameKeyType;
}
-
+
private int setIndexOnLRInSplit(int initial, POSplit splitOp, boolean sameKeyType)
throws VisitorException {
int index = initial;
-
+
List<PhysicalPlan> pls = splitOp.getPlans();
for (PhysicalPlan pl : pls) {
PhysicalOperator leaf = pl.getLeaves().get(0);
if (leaf instanceof POLocalRearrange) {
POLocalRearrange lr = (POLocalRearrange)leaf;
try {
- lr.setMultiQueryIndex(index++);
- } catch (ExecException e) {
+ lr.setMultiQueryIndex(index++);
+ } catch (ExecException e) {
int errCode = 2136;
String msg = "Internal Error. Unable to set multi-query index for optimization.";
- throw new OptimizerException(msg, errCode, PigException.BUG, e);
+ throw new OptimizerException(msg, errCode, PigException.BUG, e);
}
-
- // change the map key type to tuple when
+
+ // change the map key type to tuple when
// multiple splittees have different map key types
if (!sameKeyType) {
lr.setKeyType(DataType.TUPLE);
@@ -632,26 +619,26 @@ class MultiQueryOptimizer extends MROpPl
return index;
}
-
- private int mergeOneMapPlanWithIndex(PhysicalPlan pl, POSplit splitOp,
- int index, boolean sameKeyType) throws VisitorException {
+
+ private int mergeOneMapPlanWithIndex(PhysicalPlan pl, POSplit splitOp,
+ int index, boolean sameKeyType) throws VisitorException {
PhysicalOperator load = pl.getRoots().get(0);
pl.remove(load);
-
+
int curIndex = index;
-
+
PhysicalOperator leaf = pl.getLeaves().get(0);
if (leaf instanceof POLocalRearrange) {
POLocalRearrange lr = (POLocalRearrange)leaf;
try {
- lr.setMultiQueryIndex(curIndex++);
- } catch (ExecException e) {
+ lr.setMultiQueryIndex(curIndex++);
+ } catch (ExecException e) {
int errCode = 2136;
String msg = "Internal Error. Unable to set multi-query index for optimization.";
throw new OptimizerException(msg, errCode, PigException.BUG, e);
}
-
- // change the map key type to tuple when
+
+ // change the map key type to tuple when
// multiple splittees have different map key types
if (!sameKeyType) {
lr.setKeyType(DataType.TUPLE);
@@ -673,50 +660,55 @@ class MultiQueryOptimizer extends MROpPl
POSplit spl = (POSplit)leaf;
curIndex = setIndexOnLRInSplit(index, spl, sameKeyType);
}
-
+
splitOp.addPlan(pl);
-
+
// return the updated index after setting index
// on all POLocalRearranges including ones
// in inner plans of any POSplit operators
return curIndex;
}
-
- private void mergeOneReducePlanWithIndex(PhysicalPlan from,
- PhysicalPlan to, int initial, int current, byte mapKeyType) throws VisitorException {
+
+ private void mergeOneReducePlanWithIndex(PhysicalPlan from,
+ PhysicalPlan to, int initial, int current, byte mapKeyType) throws VisitorException {
POPackage pk = (POPackage)from.getRoots().get(0);
from.remove(pk);
-
- if(!(pk instanceof POMultiQueryPackage)){
+ Packager fromPkgr = pk.getPkgr();
+
+ if (!(fromPkgr instanceof MultiQueryPackager)) {
// XXX the index of the original keyInfo map is always 0,
// we need to shift the index so that the lookups works
// with the new indexed key
- addShiftedKeyInfoIndex(initial, pk);
+ addShiftedKeyInfoIndex(initial, fromPkgr);
}
-
+
int total = current - initial;
-
- POMultiQueryPackage pkg = (POMultiQueryPackage)to.getRoots().get(0);
+
+ MultiQueryPackager toPkgr = (MultiQueryPackager) ((POPackage) to
+ .getRoots().get(0)).getPkgr();
int pkCount = 0;
- if (pk instanceof POMultiQueryPackage) {
- List<POPackage> pkgs = ((POMultiQueryPackage)pk).getPackages();
- for (POPackage p : pkgs) {
- pkg.addPackage(p);
+ if (fromPkgr instanceof MultiQueryPackager) {
+ List<Packager> pkgs = ((MultiQueryPackager) fromPkgr)
+ .getPackagers();
+ for (Packager p : pkgs) {
+ ((MultiQueryPackager) fromPkgr).addPackager(p);
pkCount++;
}
- pkg.addIsKeyWrappedList(((POMultiQueryPackage)pk).getIsKeyWrappedList());
- addShiftedKeyInfoIndex(initial, current, (POMultiQueryPackage)pk);
+ toPkgr.addIsKeyWrappedList(((MultiQueryPackager) fromPkgr)
+ .getIsKeyWrappedList());
+ addShiftedKeyInfoIndex(initial, current,
+ (MultiQueryPackager) fromPkgr);
} else {
- pkg.addPackage(pk, mapKeyType);
+ toPkgr.addPackager(fromPkgr, mapKeyType);
pkCount = 1;
}
-
+
if (pkCount != total) {
int errCode = 2146;
String msg = "Internal Error. Inconsistency in key index found during optimization.";
throw new OptimizerException(msg, errCode, PigException.BUG);
}
-
+
PODemux demux = (PODemux)to.getLeaves().get(0);
int plCount = 0;
PhysicalOperator root = from.getRoots().get(0);
@@ -733,21 +725,22 @@ class MultiQueryOptimizer extends MROpPl
demux.addPlan(from);
plCount = 1;
}
-
+
if (plCount != total) {
int errCode = 2146;
String msg = "Internal Error. Inconsistency in key index found during optimization.";
throw new OptimizerException(msg, errCode, PigException.BUG);
}
- if (pkg.isSameMapKeyType()) {
- pkg.setKeyType(pk.getKeyType());
+ if (toPkgr.isSameMapKeyType()) {
+ toPkgr.setKeyType(pk.getPkgr().getKeyType());
} else {
- pkg.setKeyType(DataType.TUPLE);
- }
+ toPkgr.setKeyType(DataType.TUPLE);
+ }
}
-
- private void addShiftedKeyInfoIndex(int index, POPackage pkg) throws OptimizerException {
+
+ private void addShiftedKeyInfoIndex(int index, Packager pkg)
+ throws OptimizerException {
/**
* we only do multi query optimization for single input MROpers
* Hence originally the keyInfo would have had only index 0. As
@@ -761,9 +754,10 @@ class MultiQueryOptimizer extends MROpPl
* addition should be the same as the "value" in the existing Entry. After
* addition, we should remove the older entry
*/
- Map<Integer, Pair<Boolean, Map<Integer, Integer>>> keyInfo = pkg.getKeyInfo();
+ Map<Integer, Pair<Boolean, Map<Integer, Integer>>> keyInfo = pkg
+ .getKeyInfo();
byte newIndex = (byte)(index | PigNullableWritable.mqFlag);
-
+
Set<Integer> existingIndices = keyInfo.keySet();
if(existingIndices.size() != 1) {
// we always maintain one entry in the keyinfo
@@ -775,7 +769,7 @@ class MultiQueryOptimizer extends MROpPl
}
int existingIndex = existingIndices.iterator().next();
keyInfo.put(Integer.valueOf(newIndex), keyInfo.get(existingIndex));
-
+
// clean up the old entry so we only keep
// the valid entry around - if we did something wrong while
// setting this up, we will fail at runtime which is better
@@ -784,17 +778,17 @@ class MultiQueryOptimizer extends MROpPl
keyInfo.remove(existingIndex);
}
}
-
+
/**
* @param initialIndex
* @param onePastEndIndex
* @param mpkg
- * @throws OptimizerException
+ * @throws OptimizerException
*/
private int addShiftedKeyInfoIndex(int initialIndex, int onePastEndIndex,
- POMultiQueryPackage mpkg) throws OptimizerException {
-
- List<POPackage> pkgs = mpkg.getPackages();
+ MultiQueryPackager mpkgr) throws OptimizerException {
+
+ List<Packager> pkgs = mpkgr.getPackagers();
// if we have lesser pkgs than (onePastEndIndex - initialIndex)
// its because one or more of the pkgs is a POMultiQueryPackage which
// internally has packages.
@@ -810,56 +804,58 @@ class MultiQueryOptimizer extends MROpPl
int i = 0;
int curIndex = initialIndex;
while (i < end) {
- POPackage pkg = pkgs.get(i);
+ Packager pkg = pkgs.get(i);
addShiftedKeyInfoIndex(curIndex, pkg);
curIndex++;
i++;
}
return curIndex; // could be used in a caller who recursively called this function
-
+
}
private void mergeOneCombinePlanWithIndex(PhysicalPlan from,
PhysicalPlan to, int initial, int current, byte mapKeyType) throws VisitorException {
POPackage cpk = (POPackage)from.getRoots().get(0);
from.remove(cpk);
-
+ Packager cpkgr = cpk.getPkgr();
+
PODemux demux = (PODemux)to.getLeaves().get(0);
-
- POMultiQueryPackage pkg = (POMultiQueryPackage)to.getRoots().get(0);
-
- boolean isSameKeyType = pkg.isSameMapKeyType();
-
+
+ MultiQueryPackager toPkgr = (MultiQueryPackager) ((POPackage) to
+ .getRoots().get(0)).getPkgr();
+
+ boolean isSameKeyType = toPkgr.isSameMapKeyType();
+
// if current > initial + 1, it means we had
// a split in the map of the MROper we are trying to
// merge. In that case we would have changed the indices
// of the POLocalRearranges in the split to be in the
// range initial to current. To handle key, value pairs
// coming out of those POLocalRearranges, we add
- // the Packages in the 'from' POMultiQueryPackage (in this case,
+ // the Packages in the 'from' POMultiQueryPackage (in this case,
// it has to be a POMultiQueryPackage since we had
- // a POSplit in the map) to the 'to' POMultiQueryPackage.
- // These Packages would have correct positions in the package
+ // a POSplit in the map) to the 'to' POMultiQueryPackage.
+ // These Packages would have correct positions in the package
// list and would be able to handle the outputs from the different
// POLocalRearranges.
int total = current - initial;
int pkCount = 0;
- if (cpk instanceof POMultiQueryPackage) {
- List<POPackage> pkgs = ((POMultiQueryPackage)cpk).getPackages();
- for (POPackage p : pkgs) {
- pkg.addPackage(p);
+ if (cpkgr instanceof MultiQueryPackager) {
+ List<Packager> pkgrs = ((MultiQueryPackager) cpkgr).getPackagers();
+ for (Packager p : pkgrs) {
+ toPkgr.addPackager(p);
if (!isSameKeyType) {
p.setKeyType(DataType.TUPLE);
}
pkCount++;
}
} else {
- pkg.addPackage(cpk);
+ toPkgr.addPackager(cpkgr);
pkCount = 1;
}
- pkg.setSameMapKeyType(isSameKeyType);
-
+ toPkgr.setSameMapKeyType(isSameKeyType);
+
if (pkCount != total) {
int errCode = 2146;
String msg = "Internal Error. Inconsistency in key index found during optimization.";
@@ -868,11 +864,11 @@ class MultiQueryOptimizer extends MROpPl
// all packages should have the same key type
if (!isSameKeyType) {
- cpk.setKeyType(DataType.TUPLE);
- }
-
- pkg.setKeyType(cpk.getKeyType());
-
+ cpk.getPkgr().setKeyType(DataType.TUPLE);
+ }
+
+ toPkgr.setKeyType(cpk.getPkgr().getKeyType());
+
// See comment above for why we flatten the Packages
// in the from plan - for the same reason, we flatten
// the inner plans of Demux operator now.
@@ -884,14 +880,14 @@ class MultiQueryOptimizer extends MROpPl
demux.addPlan(pl);
POLocalRearrange lr = (POLocalRearrange)pl.getLeaves().get(0);
try {
- lr.setMultiQueryIndex(initial + plCount++);
- } catch (ExecException e) {
+ lr.setMultiQueryIndex(initial + plCount++);
+ } catch (ExecException e) {
int errCode = 2136;
String msg = "Internal Error. Unable to set multi-query index for optimization.";
throw new OptimizerException(msg, errCode, PigException.BUG, e);
}
-
- // change the map key type to tuple when
+
+ // change the map key type to tuple when
// multiple splittees have different map key types
if (!isSameKeyType) {
lr.setKeyType(DataType.TUPLE);
@@ -901,27 +897,27 @@ class MultiQueryOptimizer extends MROpPl
demux.addPlan(from);
POLocalRearrange lr = (POLocalRearrange)from.getLeaves().get(0);
try {
- lr.setMultiQueryIndex(initial + plCount++);
- } catch (ExecException e) {
+ lr.setMultiQueryIndex(initial + plCount++);
+ } catch (ExecException e) {
int errCode = 2136;
String msg = "Internal Error. Unable to set multi-query index for optimization.";
throw new OptimizerException(msg, errCode, PigException.BUG, e);
}
-
- // change the map key type to tuple when
+
+ // change the map key type to tuple when
// multiple splittees have different map key types
if (!isSameKeyType) {
lr.setKeyType(DataType.TUPLE);
}
}
-
+
if (plCount != total) {
int errCode = 2146;
String msg = "Internal Error. Inconsistency in key index found during optimization.";
throw new OptimizerException(msg, errCode, PigException.BUG);
}
}
-
+
private boolean needCombiner(List<MapReduceOper> mapReducers) {
boolean needCombiner = false;
for (MapReduceOper mrOp : mapReducers) {
@@ -932,17 +928,17 @@ class MultiQueryOptimizer extends MROpPl
}
return needCombiner;
}
-
- private PhysicalPlan createDemuxPlan(boolean sameKeyType, boolean isCombiner)
+
+ private PhysicalPlan createDemuxPlan(boolean sameKeyType, boolean isCombiner)
throws VisitorException {
PODemux demux = getDemux(isCombiner);
- POMultiQueryPackage pkg= getMultiQueryPackage(sameKeyType, isCombiner);
-
+ POPackage pkg = getMultiQueryPackage(sameKeyType, isCombiner);
+
PhysicalPlan pl = new PhysicalPlan();
pl.add(pkg);
try {
pl.addAsLeaf(demux);
- } catch (PlanException e) {
+ } catch (PlanException e) {
int errCode = 2137;
String msg = "Internal Error. Unable to add demux to the plan as leaf for optimization.";
throw new OptimizerException(msg, errCode, PigException.BUG, e);
@@ -950,26 +946,26 @@ class MultiQueryOptimizer extends MROpPl
return pl;
}
- private void mergeAllMapReduceSplittees(List<MapReduceOper> mergeList,
+ private void mergeAllMapReduceSplittees(List<MapReduceOper> mergeList,
MapReduceOper splitter, POSplit splitOp) throws VisitorException {
-
+
boolean sameKeyType = hasSameMapKeyType(mergeList);
-
+
log.debug("Splittees have the same key type: " + sameKeyType);
-
+
// create a new reduce plan that will be the container
// for the multiple reducer plans of the MROpers in the mergeList
PhysicalPlan redPl = createDemuxPlan(sameKeyType, false);
-
+
// create a new combine plan that will be the container
- // for the multiple combiner plans of the MROpers in the mergeList
- PhysicalPlan comPl = needCombiner(mergeList) ?
+ // for the multiple combiner plans of the MROpers in the mergeList
+ PhysicalPlan comPl = needCombiner(mergeList) ?
createDemuxPlan(sameKeyType, true) : null;
log.debug("Splittees have combiner: " + (comPl != null));
-
- int index = 0;
-
+
+ int index = 0;
+
for (MapReduceOper mrOp : mergeList) {
// merge the map plan - this will recursively
@@ -979,87 +975,87 @@ class MultiQueryOptimizer extends MROpPl
// > index + 1
int incIndex = mergeOneMapPlanWithIndex(
mrOp.mapPlan, splitOp, index, sameKeyType);
-
+
// merge the combiner plan
if (comPl != null) {
- if (!mrOp.combinePlan.isEmpty()) {
+ if (!mrOp.combinePlan.isEmpty()) {
mergeOneCombinePlanWithIndex(
mrOp.combinePlan, comPl, index, incIndex, mrOp.mapKeyType);
- } else {
+ } else {
int errCode = 2141;
String msg = "Internal Error. Cannot merge non-combiner with combiners for optimization.";
throw new OptimizerException(msg, errCode, PigException.BUG);
}
}
-
+
// merge the reducer plan
mergeOneReducePlanWithIndex(
mrOp.reducePlan, redPl, index, incIndex, mrOp.mapKeyType);
-
+
index = incIndex;
-
- log.info("Merged MR job " + mrOp.getOperatorKey().getId()
+
+ log.info("Merged MR job " + mrOp.getOperatorKey().getId()
+ " into MR job " + splitter.getOperatorKey().getId());
}
PhysicalPlan splitterPl = splitter.mapPlan;
PhysicalOperator leaf = splitterPl.getLeaves().get(0);
PhysicalOperator storeOp = splitterPl.getLeaves().get(0);
- List<PhysicalOperator> storePreds = splitterPl.getPredecessors(storeOp);
-
+ List<PhysicalOperator> storePreds = splitterPl.getPredecessors(storeOp);
+
// replace store operator in the splitter with split operator
- if (leaf instanceof POStore) {
+ if (leaf instanceof POStore) {
splitOp.setInputs(storePreds);
try {
splitterPl.replace(storeOp, splitOp);;
- } catch (PlanException e) {
+ } catch (PlanException e) {
int errCode = 2132;
String msg = "Internal Error. Unable to replace store with split operator for optimization.";
throw new OptimizerException(msg, errCode, PigException.BUG, e);
}
- }
-
+ }
+
splitter.setMapDone(true);
splitter.reducePlan = redPl;
splitter.setReduceDone(true);
-
+
if (comPl != null) {
- splitter.combinePlan = comPl;
+ splitter.combinePlan = comPl;
}
-
+
for (MapReduceOper mrOp : mergeList) {
removeAndReconnect(mrOp, splitter);
}
-
+
splitter.mapKeyType = sameKeyType ?
- mergeList.get(0).mapKeyType : DataType.TUPLE;
-
- log.info("Requested parallelism of splitter: "
- + splitter.getRequestedParallelism());
+ mergeList.get(0).mapKeyType : DataType.TUPLE;
+
+ log.info("Requested parallelism of splitter: "
+ + splitter.getRequestedParallelism());
}
-
- private void mergeSingleMapReduceSplittee(MapReduceOper mapReduce,
+
+ private void mergeSingleMapReduceSplittee(MapReduceOper mapReduce,
MapReduceOper splitter, POSplit splitOp) throws VisitorException {
-
+
PhysicalPlan splitterPl = splitter.mapPlan;
PhysicalOperator leaf = splitterPl.getLeaves().get(0);
PhysicalOperator storeOp = splitterPl.getLeaves().get(0);
- List<PhysicalOperator> storePreds = splitterPl.getPredecessors(storeOp);
-
+ List<PhysicalOperator> storePreds = splitterPl.getPredecessors(storeOp);
+
PhysicalPlan pl = mapReduce.mapPlan;
PhysicalOperator load = pl.getRoots().get(0);
pl.remove(load);
-
+
splitOp.addPlan(pl);
-
+
splitter.setMapDone(true);
splitter.reducePlan = mapReduce.reducePlan;
splitter.setReduceDone(true);
splitter.combinePlan = mapReduce.combinePlan;
splitter.customPartitioner = mapReduce.customPartitioner;
-
+
// replace store operator in the splitter with split operator
- if (leaf instanceof POStore) {
+ if (leaf instanceof POStore) {
splitOp.setInputs(storePreds);
try {
splitterPl.replace(storeOp, splitOp);;
@@ -1067,25 +1063,25 @@ class MultiQueryOptimizer extends MROpPl
int errCode = 2132;
String msg = "Internal Error. Unable to replace store with split operator for optimization.";
throw new OptimizerException(msg, errCode, PigException.BUG, e);
- }
+ }
}
-
- removeAndReconnect(mapReduce, splitter);
+
+ removeAndReconnect(mapReduce, splitter);
}
-
+
/**
- * Removes the specified MR operator from the plan after the merge.
+ * Removes the specified MR operator from the plan after the merge.
* Connects its predecessors and successors to the merged MR operator
- *
+ *
* @param mr the MR operator to remove
- * @param newMR the MR operator to be connected to the predecessors and
+ * @param newMR the MR operator to be connected to the predecessors and
* the successors of the removed operator
* @throws VisitorException if connect operation fails
*/
private void removeAndReconnect(MapReduceOper mr, MapReduceOper newMR) throws VisitorException {
List<MapReduceOper> mapperSuccs = getPlan().getSuccessors(mr);
List<MapReduceOper> mapperPreds = getPlan().getPredecessors(mr);
-
+
// make a copy before removing operator
ArrayList<MapReduceOper> succsCopy = null;
ArrayList<MapReduceOper> predsCopy = null;
@@ -1095,12 +1091,12 @@ class MultiQueryOptimizer extends MROpPl
if (mapperPreds != null) {
predsCopy = new ArrayList<MapReduceOper>(mapperPreds);
}
- getPlan().remove(mr);
-
+ getPlan().remove(mr);
+
// reconnect the mapper's successors
if (succsCopy != null) {
for (MapReduceOper succ : succsCopy) {
- try {
+ try {
getPlan().connect(newMR, succ);
} catch (PlanException e) {
int errCode = 2133;
@@ -1109,14 +1105,14 @@ class MultiQueryOptimizer extends MROpPl
}
}
}
-
+
// reconnect the mapper's predecessors
if (predsCopy != null) {
for (MapReduceOper pred : predsCopy) {
if (newMR.getOperatorKey().equals(pred.getOperatorKey())) {
continue;
}
- try {
+ try {
getPlan().connect(pred, newMR);
} catch (PlanException e) {
int errCode = 2134;
@@ -1124,11 +1120,11 @@ class MultiQueryOptimizer extends MROpPl
throw new OptimizerException(msg, errCode, PigException.BUG, e);
}
}
- }
-
+ }
+
mergeMROperProperties(mr, newMR);
}
-
+
private void mergeMROperProperties(MapReduceOper from, MapReduceOper to) {
if (from.isEndOfAllInputSetInMap()) {
@@ -1138,7 +1134,7 @@ class MultiQueryOptimizer extends MROpPl
if (from.isEndOfAllInputSetInReduce()) {
to.setEndOfAllInputInReduce(true);
}
-
+
if (from.getRequestedParallelism() > to.getRequestedParallelism()) {
to.requestedParallelism = from.requestedParallelism;
}
@@ -1159,38 +1155,41 @@ class MultiQueryOptimizer extends MROpPl
private boolean isMapOnly(MapReduceOper mr) {
return mr.reducePlan.isEmpty();
}
-
+
private boolean isSingleLoadMapperPlan(PhysicalPlan pl) {
return (pl.getRoots().size() == 1);
}
-
+
private boolean isSinglePredecessor(MapReduceOper mr) {
return (getPlan().getPredecessors(mr).size() == 1);
}
-
+
private POSplit getSplit(){
return new POSplit(new OperatorKey(scope, nig.getNextNodeId(scope)));
- }
-
+ }
+
private MapReduceOper getMROper(){
return new MapReduceOper(new OperatorKey(scope, nig.getNextNodeId(scope)));
- }
-
+ }
+
private POStore getStore(){
return new POStore(new OperatorKey(scope, nig.getNextNodeId(scope)));
- }
-
+ }
+
private PODemux getDemux(boolean inCombiner){
PODemux demux = new PODemux(new OperatorKey(scope, nig.getNextNodeId(scope)));
demux.setInCombiner(inCombiner);
return demux;
- }
-
- private POMultiQueryPackage getMultiQueryPackage(boolean sameMapKeyType, boolean inCombiner){
- POMultiQueryPackage pkg =
- new POMultiQueryPackage(new OperatorKey(scope, nig.getNextNodeId(scope)));
- pkg.setInCombiner(inCombiner);
- pkg.setSameMapKeyType(sameMapKeyType);
+ }
+
+ private POPackage getMultiQueryPackage(boolean sameMapKeyType,
+ boolean inCombiner) {
+ POPackage pkg = new POPackage(new OperatorKey(scope,
+ nig.getNextNodeId(scope)));
+ MultiQueryPackager pkgr = new MultiQueryPackager();
+ pkgr.setInCombiner(inCombiner);
+ pkgr.setSameMapKeyType(sameMapKeyType);
+ pkg.setPkgr(pkgr);
return pkg;
- }
+ }
}
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PhyPlanSetter.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PhyPlanSetter.java?rev=1546314&r1=1546313&r2=1546314&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PhyPlanSetter.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PhyPlanSetter.java Thu Nov 28 08:56:33 2013
@@ -20,10 +20,56 @@ package org.apache.pig.backend.hadoop.ex
import java.util.List;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.*;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.*;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.Add;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.ConstantExpression;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.Divide;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.EqualToExpr;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.GTOrEqualToExpr;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.GreaterThanExpr;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.LTOrEqualToExpr;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.LessThanExpr;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.Mod;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.Multiply;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.NotEqualToExpr;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POAnd;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POBinCond;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POCast;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POIsNull;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POMapLookUp;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.PONegative;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.PONot;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POOr;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POProject;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.PORegexp;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserComparisonFunc;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserFunc;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.Subtract;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POCollectedGroup;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PODemux;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PODistinct;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POFRJoin;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POFilter;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POForEach;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POGlobalRearrange;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLimit;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLoad;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLocalRearrange;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POMergeCogroup;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POMergeJoin;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PONative;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POOptimizedForEach;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPackage;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPartialAgg;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPreCombinerLocalRearrange;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PORank;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POSkewedJoin;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POSort;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POSplit;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStore;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStream;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POUnion;
import org.apache.pig.impl.plan.DependencyOrderWalker;
import org.apache.pig.impl.plan.VisitorException;
@@ -84,11 +130,6 @@ public class PhyPlanSetter extends PhyPl
}
@Override
- public void visitCombinerPackage(POCombinerPackage pkg) throws VisitorException{
- pkg.setParentPlan(parent);
- }
-
- @Override
public void visitPOForEach(POForEach nfe) throws VisitorException {
super.visitPOForEach(nfe);
nfe.setParentPlan(parent);
@@ -251,11 +292,6 @@ public class PhyPlanSetter extends PhyPl
}
@Override
- public void visitJoinPackage(POJoinPackage joinPackage) throws VisitorException{
- joinPackage.setParentPlan(parent);
- }
-
- @Override
public void visitCast(POCast cast) {
cast.setParentPlan(parent);
}
@@ -278,7 +314,7 @@ public class PhyPlanSetter extends PhyPl
@Override
public void visitSkewedJoin(POSkewedJoin join) throws VisitorException {
join.setParentPlan(parent);
- }
+ }
@Override
public void visitStream(POStream stream) throws VisitorException {
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java?rev=1546314&r1=1546313&r2=1546314&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java Thu Nov 28 08:56:33 2013
@@ -35,7 +35,7 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POJoinPackage;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.JoinPackager;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPackage;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.PigContext;
@@ -47,32 +47,32 @@ import org.apache.pig.tools.pigstats.Pig
public class PigCombiner {
public static JobContext sJobContext = null;
-
- public static class Combine
+
+ public static class Combine
extends Reducer<PigNullableWritable, NullableTuple, PigNullableWritable, Writable> {
-
+
private final Log log = LogFactory.getLog(getClass());
private byte keyType;
-
+
//The reduce plan
private PhysicalPlan cp;
-
+
//The POPackage operator which is the
//root of every Map Reduce plan is
//obtained through the job conf. The portion
//remaining after its removal is the reduce
//plan
private POPackage pack;
-
+
ProgressableReporter pigReporter;
-
+
PhysicalOperator[] roots;
PhysicalOperator leaf;
-
+
PigContext pigContext = null;
private volatile boolean initialized = false;
-
+
/**
* Configures the Reduce plan, the POPackage operator
* and the reporter thread
@@ -88,7 +88,7 @@ public class PigCombiner {
pigContext = (PigContext)ObjectSerializer.deserialize(jConf.get("pig.pigContext"));
if (pigContext.getLog4jProperties()!=null)
PropertyConfigurator.configure(pigContext.getLog4jProperties());
-
+
cp = (PhysicalPlan) ObjectSerializer.deserialize(jConf
.get("pig.combinePlan"));
pack = (POPackage)ObjectSerializer.deserialize(jConf.get("pig.combine.package"));
@@ -100,10 +100,10 @@ public class PigCombiner {
cp.explain(baos);
log.debug(baos.toString());
}
-
+
keyType = ((byte[])ObjectSerializer.deserialize(jConf.get("pig.map.keytype")))[0];
// till here
-
+
pigReporter = new ProgressableReporter();
if(!(cp.isEmpty())) {
roots = cp.getRoots().toArray(new PhysicalOperator[1]);
@@ -115,7 +115,7 @@ public class PigCombiner {
}
log.info("Aliases being processed per job phase (AliasName[line,offset]): " + jConf.get("pig.alias.location"));
}
-
+
/**
* The reduce function which packages the key and List <Tuple>
* into key, Bag<Tuple> after converting Hadoop type key into Pig type.
@@ -123,7 +123,7 @@ public class PigCombiner {
* empty or after passing through the reduce plan.
*/
@Override
- protected void reduce(PigNullableWritable key, Iterable<NullableTuple> tupIter, Context context)
+ protected void reduce(PigNullableWritable key, Iterable<NullableTuple> tupIter, Context context)
throws IOException, InterruptedException {
if(!initialized) {
initialized = true;
@@ -139,13 +139,13 @@ public class PigCombiner {
PhysicalOperator.setPigLogger(pigHadoopLogger);
}
-
+
// In the case we optimize, we combine
// POPackage and POForeach - so we could get many
// tuples out of the getnext() call of POJoinPackage
- // In this case, we process till we see EOP from
+ // In this case, we process till we see EOP from
// POJoinPacakage.getNext()
- if (pack instanceof POJoinPackage)
+ if (pack.getPkgr() instanceof JoinPackager)
{
pack.attachInput(key, tupIter.iterator());
while (true)
@@ -160,9 +160,9 @@ public class PigCombiner {
pack.attachInput(key, tupIter.iterator());
processOnePackageOutput(context);
}
-
+
}
-
+
// return: false-more output
// true- end of processing
public boolean processOnePackageOutput(Context oc) throws IOException, InterruptedException {
@@ -170,18 +170,18 @@ public class PigCombiner {
Result res = pack.getNextTuple();
if(res.returnStatus==POStatus.STATUS_OK){
Tuple packRes = (Tuple)res.result;
-
+
if(cp.isEmpty()){
oc.write(null, packRes);
return false;
}
-
+
for (int i = 0; i < roots.length; i++) {
roots[i].attachInput(packRes);
}
while(true){
Result redRes = leaf.getNextTuple();
-
+
if(redRes.returnStatus==POStatus.STATUS_OK){
Tuple tuple = (Tuple)redRes.result;
Byte index = (Byte)tuple.get(0);
@@ -200,15 +200,15 @@ public class PigCombiner {
continue;
}
-
+
if(redRes.returnStatus==POStatus.STATUS_EOP) {
break;
}
-
+
if(redRes.returnStatus==POStatus.STATUS_NULL) {
continue;
}
-
+
if(redRes.returnStatus==POStatus.STATUS_ERR){
int errCode = 2090;
String msg = "Received Error while " +
@@ -220,35 +220,35 @@ public class PigCombiner {
}
}
}
-
+
if(res.returnStatus==POStatus.STATUS_NULL) {
return false;
}
-
+
if(res.returnStatus==POStatus.STATUS_ERR){
int errCode = 2091;
String msg = "Packaging error while processing group.";
throw new ExecException(msg, errCode, PigException.BUG);
}
-
+
if(res.returnStatus==POStatus.STATUS_EOP) {
return true;
}
-
- return false;
-
+
+ return false;
+
} catch (ExecException e) {
throw e;
}
}
-
+
/**
* Will be called once all the intermediate keys and values are
* processed.
* cleanup references to the PhysicalPlan
*/
- @Override
+ @Override
protected void cleanup(Context context) throws IOException, InterruptedException {
super.cleanup(context);
leaf = null;
@@ -273,5 +273,5 @@ public class PigCombiner {
this.keyType = keyType;
}
}
-
+
}