You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by pr...@apache.org on 2009/11/24 20:54:34 UTC
svn commit: r883836 [17/23] - in /hadoop/pig/branches/load-store-redesign:
./ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/
contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/
contrib/zebra/ contrib/zebr...
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MultiQueryOptimizer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MultiQueryOptimizer.java?rev=883836&r1=883835&r2=883836&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MultiQueryOptimizer.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MultiQueryOptimizer.java Tue Nov 24 19:54:19 2009
@@ -24,12 +24,14 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.pig.PigException;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.plans.MROpPlanVisitor;
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.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;
@@ -43,9 +45,8 @@
import org.apache.pig.impl.plan.PlanException;
import org.apache.pig.impl.plan.ReverseDependencyOrderWalker;
import org.apache.pig.impl.plan.VisitorException;
-import org.apache.pig.impl.util.Pair;
import org.apache.pig.impl.plan.optimizer.OptimizerException;
-import org.apache.pig.PigException;
+import org.apache.pig.impl.util.Pair;
/**
@@ -204,10 +205,121 @@
}
}
+ // 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;
+ }
+
log.info("Merged " + numMerges + " out of total "
- + numSplittees + " splittees.");
+ + (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
+ // (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 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) {
+ PhysicalOperator root = pl.getRoots().get(0);
+ PhysicalOperator leaf = pl.getLeaves().get(0);
+ if (root instanceof POLoad && leaf instanceof POStore) {
+ if (pl.size() == 3) {
+ PhysicalOperator mid = pl.getSuccessors(root).get(0);
+ if (mid instanceof POForEach) {
+ rtn = true;
+ }
+ } else {
+ rtn = true;
+ }
+ }
+ }
+ }
+ return rtn;
+ }
+
+ private int mergeDiamondMROper(MapReduceOper mr, List<MapReduceOper> succs)
+ throws VisitorException {
+
+ // Only consider the cases where all inputs of the splittees are
+ // from the splitter
+ for (MapReduceOper succ : succs) {
+ List<MapReduceOper> preds = getPlan().getPredecessors(succ);
+ if (preds.size() != 1) {
+ return 0;
+ }
+ }
+
+ // first, remove the store operator from the splitter
+ PhysicalPlan pl = mr.mapPlan;
+ PhysicalOperator leaf = mr.mapPlan.getLeaves().get(0);
+ pl.remove(leaf);
+
+ // 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 =
+ new ArrayList<PhysicalOperator>(roots);
+ for (PhysicalOperator op : rootsCopy) {
+ PhysicalOperator opSucc = succ.mapPlan.getSuccessors(op).get(0);
+ PhysicalPlan clone = null;
+ try {
+ clone = pl.clone();
+ } catch (CloneNotSupportedException e) {
+ int errCode = 2127;
+ String msg = "Internal Error: Cloning of plan failed for optimization.";
+ throw new OptimizerException(msg, errCode, PigException.BUG, e);
+ }
+ succ.mapPlan.remove(op);
+ while (!clone.isEmpty()) {
+ PhysicalOperator oper = clone.getLeaves().get(0);
+ clone.remove(oper);
+ succ.mapPlan.add(oper);
+ try {
+ succ.mapPlan.connect(oper, opSucc);
+ opSucc = oper;
+ } catch (PlanException e) {
+ int errCode = 2131;
+ String msg = "Internal Error. Unable to connect split plan for optimization.";
+ throw new OptimizerException(msg, errCode, PigException.BUG, e);
+ }
+ }
+ }
+ }
+
+ // finally, remove the splitter from the MR plan
+ List<MapReduceOper> mrPreds = getPlan().getPredecessors(mr);
+ if (mrPreds != null) {
+ for (MapReduceOper pred : mrPreds) {
+ for (MapReduceOper succ : succs) {
+ try {
+ getPlan().connect(pred, succ);
+ } catch (PlanException e) {
+ int errCode = 2131;
+ String msg = "Internal Error. Unable to connect split plan for optimization.";
+ throw new OptimizerException(msg, errCode, PigException.BUG, e);
+ }
+ }
+ }
+ }
+
+ getPlan().remove(mr);
+
+ return 1;
+ }
+
private void mergeOneMapPart(MapReduceOper mapper, MapReduceOper splitter)
throws VisitorException {
PhysicalPlan splitterPl = isMapOnly(splitter) ?
@@ -509,58 +621,7 @@
// in inner plans of any POSplit operators
return curIndex;
}
-
- private int setBaseIndexOnDemux(int initial, PODemux demuxOp)
- throws VisitorException {
- int index = initial;
- demuxOp.setBaseIndex(index++);
-
- List<PhysicalPlan> pls = demuxOp.getPlans();
- for (PhysicalPlan pl : pls) {
- PhysicalOperator leaf = pl.getLeaves().get(0);
- if (leaf instanceof POLocalRearrange) {
- POLocalRearrange lr = (POLocalRearrange)leaf;
- try {
- // if the baseindex is set on the demux, then
- // POLocalRearranges in its inner plan should really
- // be sending an index out by adding the base index
- // This is because we would be replicating the demux
- // as many times as there are inner plans in the demux
- // hence the index coming out of POLocalRearranges
- // needs to be adjusted accordingly
- lr.setMultiQueryIndex(initial + lr.getIndex());
- } 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);
- }
- }
- PhysicalOperator root = pl.getRoots().get(0);
- if (root instanceof PODemux) {
- index = setBaseIndexOnDemux(index, (PODemux)root);
- } else {
- index++;
- }
- }
- return index;
- }
-
- private int setBaseIndexOnPackage(int initial, POMultiQueryPackage pkgOp) {
- int index = initial;
- pkgOp.setBaseIndex(index++);
- List<POPackage> pkgs = pkgOp.getPackages();
- for (POPackage pkg : pkgs) {
- if (pkg instanceof POMultiQueryPackage) {
- POMultiQueryPackage mpkg = (POMultiQueryPackage)pkg;
- index = setBaseIndexOnPackage(index, mpkg);
- } else {
- index++;
- }
- }
- return index;
- }
-
private void mergeOneReducePlanWithIndex(PhysicalPlan from,
PhysicalPlan to, int initial, int current, byte mapKeyType) throws VisitorException {
POPackage pk = (POPackage)from.getRoots().get(0);
@@ -572,41 +633,55 @@
// with the new indexed key
addShiftedKeyInfoIndex(initial, pk);
}
+
+ int total = current - initial;
+ POMultiQueryPackage pkg = (POMultiQueryPackage)to.getRoots().get(0);
+ int pkCount = 0;
if (pk instanceof POMultiQueryPackage) {
- POMultiQueryPackage mpkg = (POMultiQueryPackage)pk;
- setBaseIndexOnPackage(initial, mpkg);
- // we should update the keyinfo map of the
- // POPackage objects in the POMultiQueryPackage to
- // have the shifted index - The index now will be
- // starting from "initial" going up to "current"
- // ORed with the multi query bit mask
- int retIndex = addShiftedKeyInfoIndex(initial, current, mpkg);
- if(retIndex != current) {
- int errCode = 2146;
- String msg = "Internal Error. Inconsistency in key index found during optimization.";
- throw new OptimizerException(msg, errCode, PigException.BUG);
+ List<POPackage> pkgs = ((POMultiQueryPackage)pk).getPackages();
+ for (POPackage p : pkgs) {
+ pkg.addPackage(p);
+ pkCount++;
}
- }
-
- PhysicalOperator root = from.getRoots().get(0);
- if (root instanceof PODemux) {
- PODemux demux = (PODemux)root;
- setBaseIndexOnDemux(initial, demux);
- }
-
- POMultiQueryPackage pkg = (POMultiQueryPackage)to.getRoots().get(0);
- for (int i=initial; i<current; i++) {
+ addShiftedKeyInfoIndex(initial, current, (POMultiQueryPackage)pk);
+ } else {
pkg.addPackage(pk);
+ 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);
+ }
+
boolean[] keyPos = pk.getKeyPositionsInTuple();
PODemux demux = (PODemux)to.getLeaves().get(0);
- for (int i=initial; i<current; i++) {
+ int plCount = 0;
+ PhysicalOperator root = from.getRoots().get(0);
+ if (root instanceof PODemux) {
+ // flattening the inner plans of the demux operator.
+ // This is based on the fact that if a plan has a demux
+ // operator, then it's the only operator in the plan.
+ List<PhysicalPlan> pls = ((PODemux)root).getPlans();
+ for (PhysicalPlan pl : pls) {
+ demux.addPlan(pl, keyPos);
+ plCount++;
+ }
+ demux.addIsKeyWrappedList(((PODemux)root).getIsKeyWrappedList());
+ } else {
demux.addPlan(from, mapKeyType, keyPos);
+ 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 (demux.isSameMapKeyType()) {
pkg.setKeyType(pk.getKeyType());
} else {
@@ -660,11 +735,7 @@
*/
private int addShiftedKeyInfoIndex(int initialIndex, int onePastEndIndex,
POMultiQueryPackage mpkg) throws OptimizerException {
- // recursively iterate over the packages in the
- // POMultiQueryPackage adding a shifted keyInfoIndex entry
- // in the packages in order going from initialIndex upto
- // onePastEndIndex (exclusive) flattening out any nested
- // packages in nested POMultiqueryPackages as we traverse
+
List<POPackage> pkgs = mpkg.getPackages();
// if we have lesser pkgs than (onePastEndIndex - initialIndex)
// its because one or more of the pkgs is a POMultiQueryPackage which
@@ -682,12 +753,8 @@
int curIndex = initialIndex;
while (i < end) {
POPackage pkg = pkgs.get(i);
- if(pkg instanceof POMultiQueryPackage) {
- curIndex = addShiftedKeyInfoIndex(curIndex, onePastEndIndex, (POMultiQueryPackage)pkg);
- } else {
- addShiftedKeyInfoIndex(curIndex, pkg);
- curIndex++;
- }
+ addShiftedKeyInfoIndex(curIndex, pkg);
+ curIndex++;
i++;
}
return curIndex; // could be used in a caller who recursively called this function
@@ -698,37 +765,11 @@
PhysicalPlan to, int initial, int current, byte mapKeyType) throws VisitorException {
POPackage cpk = (POPackage)from.getRoots().get(0);
from.remove(cpk);
-
- if (cpk instanceof POMultiQueryPackage) {
- POMultiQueryPackage mpkg = (POMultiQueryPackage)cpk;
- setBaseIndexOnPackage(initial, mpkg);
- }
PODemux demux = (PODemux)to.getLeaves().get(0);
boolean isSameKeyType = demux.isSameMapKeyType();
- PhysicalOperator leaf = from.getLeaves().get(0);
- if (leaf instanceof POLocalRearrange) {
- POLocalRearrange clr = (POLocalRearrange)leaf;
- try {
- clr.setMultiQueryIndex(initial);
- } 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
- // multiple splittees have different map key types
- if (!isSameKeyType) {
- clr.setKeyType(DataType.TUPLE);
- }
- } else if (leaf instanceof PODemux) {
- PODemux locDemux = (PODemux)leaf;
- setBaseIndexOnDemux(initial, locDemux);
- }
-
POMultiQueryPackage pkg = (POMultiQueryPackage)to.getRoots().get(0);
// if current > initial + 1, it means we had
@@ -736,17 +777,35 @@
// 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 replicate
- // the Package as many times (in this case, the package
- // would have to be a POMultiQueryPackage since we had
- // a POSplit in the map). That Package would have a baseindex
- // correctly set (in the beginning of this method) and would
- // be able to handle the outputs from the different
+ // coming out of those POLocalRearranges, we add
+ // 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
+ // list and would be able to handle the outputs from the different
// POLocalRearranges.
- for (int i=initial; i<current; i++) {
+ 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 (!isSameKeyType) {
+ p.setKeyType(DataType.TUPLE);
+ }
+ pkCount++;
+ }
+ } else {
pkg.addPackage(cpk);
+ 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);
+ }
+
// all packages should have the same key type
if (!isSameKeyType) {
cpk.setKeyType(DataType.TUPLE);
@@ -756,11 +815,52 @@
boolean[] keyPos = cpk.getKeyPositionsInTuple();
- // See comment above for why we replicated the Package
- // in the from plan - for the same reason, we replicate
- // the Demux operators now.
- for (int i=initial; i<current; i++) {
+ // 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.
+ int plCount = 0;
+ PhysicalOperator leaf = from.getLeaves().get(0);
+ if (leaf instanceof PODemux) {
+ List<PhysicalPlan> pls = ((PODemux)leaf).getPlans();
+ for (PhysicalPlan pl : pls) {
+ demux.addPlan(pl, mapKeyType, keyPos);
+ POLocalRearrange lr = (POLocalRearrange)pl.getLeaves().get(0);
+ try {
+ 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
+ // multiple splittees have different map key types
+ if (!isSameKeyType) {
+ lr.setKeyType(DataType.TUPLE);
+ }
+ }
+ } else {
demux.addPlan(from, mapKeyType, keyPos);
+ POLocalRearrange lr = (POLocalRearrange)from.getLeaves().get(0);
+ try {
+ 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
+ // 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);
}
}
@@ -821,73 +921,7 @@
// > index + 1
int incIndex = mergeOneMapPlanWithIndex(
mrOp.mapPlan, splitOp, index, sameKeyType);
-
- // In the combine and reduce plans the Demux and POMultiQueryPackage
- // operators' baseIndex is set whenever the incIndex above is > index + 1
- // What does this 'baseIndex' mean - here is an attempt to explain it:
- // Consider a map - reduce plan layout as shown below (the comments
- // apply even if a combine plan was present) - An explanation of the "index"
- // and "baseIndex" fields follows:
- // The numbers in parenthesis () are "index" values - Note that in multiquery
- // optimizations these indices are actually ORed with a bitmask (0x80) on the
- // map side in the LocalRearrange. The POMultiQueryPackage and PODemux operators
- // then remove this bitmask to restore the original index values. In the commentary
- // below, indices will be referred to without this bitmask - the bitmask is only
- // to identify the multiquery optimization during comparsion - for details see the comments
- // in POLocalRearrange.setIndex().
- // The numbers in brackets [] are "baseIndex" values. These baseIndex values are
- // used by POMultiQueryPackage and PODemux to calculate the an arrayList index which
- // they use to pick the right package or inner plan respectively. All this is needed
- // since on the map side the indices are assigned after flattening all POLocalRearranges
- // including those nested in Splits into one flat space (as can be noticed by the
- // numbering of the indices under the split in the example below). The optimizer then
- // duplicates the POMultiQueryPackage and Demux inner plan corresponding to the split
- // in the reduce plan (the entities with * in the figure below). Now if a key with index '1'
- // is emitted, it goes to the first POMultiQueryPackage in the reduce plan below, which
- // then picks the second package in its arraylist of packages which is a
- // POMultiQueryPackage (the first with a * below). This POMultiQueryPackage then picks
- // the first package in its list (it arrives at this arraylist index of 0 by subtracting
- // the baseIndex (1) from the index coming in (1)). So the record emitted by LocalRearrange(1)
- // reaches the right package. Likewise, if LocalRearrange(2) emits a key,value they would have
- // an index 2. The first POMultiQueryPackage (with baseIndex 0 below) would pick the 3rd package
- // (arraylist index 2 arrived at by doing index - baseIndex which is 2 - 0). This is the
- // duplicated POMultiQueryPackage with baseIndex 1. This would inturn pick the second package
- // in its arraylist (arraylist index 1 arrived at by doing index - baseIndex which is 2 - 1)
- // The idea is that through duplication we make it easier to determine which POPackage to pick.
- // The exact same logic is used by PODemux to pick the right inner plan from its arraylist of
- // inner plans.
-
- // The arrows in the figure below show the correspondence between the different operators
- // and plans .i.e the end of an arrow points to the operator or plan which consumes
- // data coming from the root of the arrow
-
- // If you look at the layout below column-wise, each "column" is a MROper
- // which is being merged into the parent MROper - the Split represents a
- // MROper which inside of it has 2 MROpers merged in.
- // A star (*) next to an operator means it is the same object as the
- // other operator with a star(*) - Essentially the same object reference
- // is copied twice.
- // LocalRearrange(0) Split LocalRearrange(3)
- // | / \ |
- // | LocalRearrange(1) LocalRearrange(2) |
- // | | MAP PLAN | |
- // ----|-------------|---------------------------|-------------------|--------------------
- // | | REDUCE PLAN | |
- // | | POMultiQueryPackage[0] | |
- // V V | contains V V
- // [ POPackage, POMultiQueryPackage[1]*,POMultiQueryPackage[1]*, POPackage]
- // | / \ / \ |
- // | POPackage POPackage POPackage POPackage |
- // | \ | |
- // | \ Demux[0] | |
- // V V | contains V V
- // [ planOfRedOperators,planWithDemux*, planWithDemux*, planOfRedOperators]
- // / | | \
- // / Demux[1] Demux[1] \
- // V | | V
- // [planOfRedOps,planOfRedOps][planOfRedOps,planOfRedOps]
- //
-
+
// merge the combiner plan
if (comPl != null) {
if (!mrOp.combinePlan.isEmpty()) {
@@ -899,6 +933,7 @@
throw new OptimizerException(msg, errCode, PigException.BUG);
}
}
+
// merge the reducer plan
mergeOneReducePlanWithIndex(
mrOp.reducePlan, redPl, index, incIndex, mrOp.mapKeyType);
@@ -1091,5 +1126,5 @@
private POMultiQueryPackage getMultiQueryPackage(){
return new POMultiQueryPackage(new OperatorKey(scope, nig.getNextNodeId(scope)));
- }
+ }
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PhyPlanSetter.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PhyPlanSetter.java?rev=883836&r1=883835&r2=883836&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PhyPlanSetter.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PhyPlanSetter.java Tue Nov 24 19:54:19 2009
@@ -273,16 +273,6 @@
}
@Override
- public void visitCogroup(POCogroup cogroup) {
- cogroup.setParentPlan(parent);
- }
-
- @Override
- public void visitSplit(org.apache.pig.backend.local.executionengine.physicalLayer.relationalOperators.POSplit split) {
- split.setParentPlan(parent);
- }
-
- @Override
public void visitLocalRearrangeForIllustrate(
POLocalRearrangeForIllustrate lrfi) throws VisitorException {
super.visitLocalRearrangeForIllustrate(lrfi);
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapBase.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapBase.java?rev=883836&r1=883835&r2=883836&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapBase.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapBase.java Tue Nov 24 19:54:19 2009
@@ -47,6 +47,7 @@
import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.impl.util.ObjectSerializer;
import org.apache.pig.impl.util.SpillableMemoryManager;
+import org.apache.pig.impl.util.UDFContext;
public abstract class PigMapBase extends Mapper<Text, Tuple, PigNullableWritable, Writable> {
private static final Tuple DUMMYTUPLE = null;
@@ -162,6 +163,11 @@
// till here
pigReporter = new ProgressableReporter();
+ // Get the UDF specific context
+ UDFContext udfc = UDFContext.getUDFContext();
+ udfc.addJobConf(job);
+ udfc.deserialize();
+
if(!(mp.isEmpty())) {
PigSplit split = (PigSplit)context.getInputSplit();
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java?rev=883836&r1=883835&r2=883836&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java Tue Nov 24 19:54:19 2009
@@ -53,6 +53,8 @@
import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.impl.util.ObjectSerializer;
import org.apache.pig.impl.util.SpillableMemoryManager;
+import org.apache.pig.impl.util.UDFContext;
+import org.apache.pig.impl.util.WrappedIOException;
/**
* This class is the static Mapper & Reducer classes that
@@ -304,6 +306,12 @@
roots = rp.getRoots().toArray(new PhysicalOperator[1]);
leaf = rp.getLeaves().get(0);
}
+
+ // Get the UDF specific context
+ UDFContext udfc = UDFContext.getUDFContext();
+ udfc.addJobConf(jConf);
+ udfc.deserialize();
+
} catch (IOException ioe) {
String msg = "Problem while configuring reduce plan.";
throw new RuntimeException(msg, ioe);
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigOutputFormat.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigOutputFormat.java?rev=883836&r1=883835&r2=883836&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigOutputFormat.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigOutputFormat.java Tue Nov 24 19:54:19 2009
@@ -19,6 +19,8 @@
import java.io.IOException;
import java.util.List;
+import java.util.ArrayList;
+import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.WritableComparable;
@@ -28,9 +30,13 @@
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.pig.StoreConfig;
import org.apache.pig.StoreFunc;
import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStore;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStore;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.util.PlanHelper;
import org.apache.pig.backend.hadoop.executionengine.util.MapRedUtil;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.util.ObjectSerializer;
@@ -108,7 +114,6 @@
}
}
-
/**
* Wrapper class which will delegate calls to the actual RecordWriter - this
* should only get called in the single store case.
Added: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigSecondaryKeyComparator.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigSecondaryKeyComparator.java?rev=883836&view=auto
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigSecondaryKeyComparator.java (added)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigSecondaryKeyComparator.java Tue Nov 24 19:54:19 2009
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.pig.backend.hadoop.executionengine.mapReduceLayer;
+
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.io.WritableComparator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.io.NullableTuple;
+import org.apache.pig.impl.io.PigNullableWritable;
+import org.apache.pig.impl.util.ObjectSerializer;
+
+public class PigSecondaryKeyComparator extends WritableComparator implements Configurable {
+ private final Log mLog = LogFactory.getLog(getClass());
+ private boolean[] mAsc;
+ private boolean[] mSecondaryAsc;
+ private boolean mWholeTuple;
+ private boolean mSecondaryWholeTuple;
+ private final DataInputBuffer buffer;
+ private final NullableTuple key1;
+ private final NullableTuple key2;
+ @Override
+ public void setConf(Configuration conf) {
+ if (!(conf instanceof JobConf)) {
+ mLog.warn("Expected jobconf in setConf, got " +
+ conf.getClass().getName());
+ return;
+ }
+ JobConf jconf = (JobConf)conf;
+ try {
+ mAsc = (boolean[])ObjectSerializer.deserialize(jconf.get(
+ "pig.sortOrder"));
+ mSecondaryAsc = (boolean[])ObjectSerializer.deserialize(jconf.get(
+ "pig.secondarySortOrder"));
+ } catch (IOException ioe) {
+ mLog.error("Unable to deserialize sort order object" +
+ ioe.getMessage());
+ throw new RuntimeException(ioe);
+ }
+ if (mAsc == null) {
+ mAsc = new boolean[1];
+ mAsc[0] = true;
+ }
+ if (mSecondaryAsc == null) {
+ mSecondaryAsc = new boolean[1];
+ mSecondaryAsc[0] = true;
+ }
+ mWholeTuple = (mAsc.length == 1);
+ mSecondaryWholeTuple = (mSecondaryAsc.length == 1);
+ }
+ public Configuration getConf() {
+ return null;
+ }
+ @SuppressWarnings("unchecked")
+ public PigSecondaryKeyComparator() {
+ super(TupleFactory.getInstance().tupleClass());
+ buffer = new DataInputBuffer();
+ key1 = new NullableTuple();
+ key2 = new NullableTuple();
+ }
+ public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2){
+ try {
+ buffer.reset(b1, s1, l1); // parse key1
+ key1.readFields(buffer);
+
+ buffer.reset(b2, s2, l2); // parse key2
+ key2.readFields(buffer);
+
+ if ((key1.getIndex() & NullableTuple.mqFlag) != 0) { // this is a multi-query index
+ if ((key1.getIndex() & NullableTuple.idxSpace) < (key2.getIndex() & NullableTuple.idxSpace)) return -1;
+ else if ((key1.getIndex() & NullableTuple.idxSpace) > (key2.getIndex() & NullableTuple.idxSpace)) return 1;
+ // If equal, we fall through
+ }
+
+ Tuple t1 = (Tuple)key1.getValueAsPigType(); // t1 and t2 is guaranteed to be not null
+ Tuple t2 = (Tuple)key2.getValueAsPigType();
+
+ Object k1 = t1.get(0);
+ Object k2 = t2.get(0);
+
+ int r = compare(k1, k2, key1.getIndex(), key2.getIndex(), mAsc, mWholeTuple);
+
+ if (r!=0)
+ return r;
+
+ // main key is equal, compare the secondary key
+ k1 = t1.get(1);
+ k2 = t2.get(1);
+
+ r = compare(k1, k2, key1.getIndex(), key2.getIndex(), mSecondaryAsc, mSecondaryWholeTuple);
+
+ return r;
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ public int compare(Object k1, Object k2, int index1, int index2, boolean[] ascs, boolean wholeTuple) {
+ try {
+ if (k1!=null && k2!=null) {
+ if (k1 instanceof Tuple && k2 instanceof Tuple) {
+ Tuple t1 = (Tuple)k1;
+ Tuple t2 = (Tuple)k2;
+ int sz1 = t1.size();
+ int sz2 = t2.size();
+ if (sz2 < sz1) {
+ return 1;
+ } else if (sz2 > sz1) {
+ return -1;
+ } else {
+ for (int i = 0; i < sz1; i++) {
+ int c = DataType.compare(t1.get(i), t2.get(i));
+ if (c != 0) {
+ if (!wholeTuple && !ascs[i]) c *= -1;
+ else if (wholeTuple && !ascs[0]) c *= -1;
+ return c;
+ }
+ }
+ // If any of the field inside tuple is null, then we do not merge keys
+ // See PIG-927
+ for (int i=0;i<t1.size();i++)
+ if (t1.get(i)==null)
+ return (index1& PigNullableWritable.idxSpace) - (index2& PigNullableWritable.idxSpace);
+ return 0;
+ }
+ }
+ int c = DataType.compare(k1, k2);
+ if (!ascs[0]) c *= -1;
+ return c;
+ }
+ else if (k1==null && k2==null) {
+ // If they're both null, compare the index
+ return (index1& PigNullableWritable.idxSpace) - (index2& PigNullableWritable.idxSpace);
+ }
+ else if (k1==null) return -1;
+ else return 1;
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
Added: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/SecondaryKeyOptimizer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/SecondaryKeyOptimizer.java?rev=883836&view=auto
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/SecondaryKeyOptimizer.java (added)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/SecondaryKeyOptimizer.java Tue Nov 24 19:54:19 2009
@@ -0,0 +1,634 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.pig.backend.hadoop.executionengine.mapReduceLayer;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.plans.MROpPlanVisitor;
+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.expressionOperators.POProject;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.PORelationToExprProject;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserFunc;
+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.PODistinct;
+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.POJoinPackage;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLimit;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLocalRearrange;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPackage;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POSort;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POSortedDistinct;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POUnion;
+import org.apache.pig.data.DataType;
+import org.apache.pig.impl.io.PigNullableWritable;
+import org.apache.pig.impl.plan.DependencyOrderWalker;
+import org.apache.pig.impl.plan.DepthFirstWalker;
+import org.apache.pig.impl.plan.NodeIdGenerator;
+import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.PlanException;
+import org.apache.pig.impl.plan.VisitorException;
+
+class POToChange {
+ POToChange(PhysicalOperator oper, PhysicalPlan plan, POForEach forEach) {
+ this.oper = oper;
+ this.plan = plan;
+ this.forEach = forEach;
+ }
+
+ PhysicalOperator oper;
+
+ PhysicalPlan plan;
+
+ POForEach forEach;
+}
+
+public class SecondaryKeyOptimizer extends MROpPlanVisitor {
+ private Log log = LogFactory.getLog(getClass());
+
+ private int numMRUseSecondaryKey = 0;
+
+ private int numDistinctChanged = 0;
+
+ private int numSortRemoved = 0;
+
+ /**
+ * @param plan
+ * The MROperPlan to visit to discover keyType
+ */
+ public SecondaryKeyOptimizer(MROperPlan plan) {
+ super(plan, new DepthFirstWalker<MapReduceOper, MROperPlan>(plan));
+ }
+
+ // Build sort key structure from POLocalRearrange
+ SortKeyInfo getSortKeyInfo(POLocalRearrange rearrange) throws ExecException {
+ SortKeyInfo result = new SortKeyInfo();
+ List<PhysicalPlan> plans = rearrange.getPlans();
+ nextPlan: for (int i = 0; i < plans.size(); i++) {
+ PhysicalPlan plan = plans.get(i);
+ ColumnChainInfo columnChainInfo = new ColumnChainInfo();
+ if (plan.getRoots() == null) {
+ log.debug("POLocalRearrange plan is null");
+ return null;
+ } else if (plan.getRoots().size() != 1) {
+ // POLocalRearrange plan contains more than 1 root.
+ // Probably there is an Expression operator in the local
+ // rearrangement plan, skip this plan
+ continue nextPlan;
+ } else {
+ List<Integer> columns = new ArrayList<Integer>();
+ columns
+ .add(rearrange.getIndex()
+ & PigNullableWritable.idxSpace);
+
+ // The first item inside columnChainInfo is set to type Tuple.
+ // This value is not actually in use, but it intends to match
+ // the type of POProject in reduce side
+ columnChainInfo.insert(false, columns, DataType.TUPLE);
+
+ PhysicalOperator node = plan.getRoots().get(0);
+ while (node != null) {
+ if (node instanceof POProject) {
+ POProject project = (POProject) node;
+
+ columnChainInfo.insert(project.isStar(), project
+ .getColumns(), project.getResultType());
+
+ if (plan.getSuccessors(node) == null)
+ node = null;
+ else if (plan.getSuccessors(node).size() != 1) {
+ log.debug(node + " have more than 1 successor");
+ node = null;
+ } else
+ node = plan.getSuccessors(node).get(0);
+ } else
+ // constant, UDF, we will pass
+ continue nextPlan;
+ }
+ }
+ // Let's assume all main key is sorted ascendant, we can further
+ // optimize it to match one of the nested sort/distinct key, because we do not
+ // really care about how cogroup key are sorted; But it may not be the case
+ // if sometime we switch all the comparator to byte comparator, so just
+ // leave it as it is for now
+ result.insertColumnChainInfo(i, columnChainInfo, true);
+ }
+ return result;
+ }
+
+ @Override
+ public void visitMROp(MapReduceOper mr) throws VisitorException {
+ log.trace("Entering SecondaryKeyOptimizer.visitMROp, skip optimizing");
+ List<SortKeyInfo> sortKeyInfos = new ArrayList<SortKeyInfo>();
+ SortKeyInfo secondarySortKeyInfo = null;
+ List<POToChange> sortsToRemove = null;
+ List<POToChange> distinctsToChange = null;
+
+ // Only optimize for Cogroup case
+ if (mr.isGlobalSort())
+ return;
+
+ List<PhysicalOperator> mapLeaves = mr.mapPlan.getLeaves();
+ if (mapLeaves == null || mapLeaves.size() != 1) {
+ log
+ .debug("Expected map to have single leaf! Skip secondary key optimizing");
+ return;
+ }
+ PhysicalOperator mapLeaf = mapLeaves.get(0);
+
+ // Figure out the main key of the map-reduce job from POLocalRearrange
+ try {
+ if (mapLeaf instanceof POLocalRearrange) {
+ SortKeyInfo sortKeyInfo = getSortKeyInfo((POLocalRearrange) mapLeaf);
+ if (sortKeyInfo == null) {
+ log
+ .debug("Cannot get sortKeyInfo from POLocalRearrange, skip secondary key optimizing");
+ return;
+ }
+ sortKeyInfos.add(sortKeyInfo);
+ } else if (mapLeaf instanceof POUnion) {
+ List<PhysicalOperator> preds = mr.mapPlan
+ .getPredecessors(mapLeaf);
+ for (PhysicalOperator pred : preds) {
+ if (pred instanceof POLocalRearrange) {
+ SortKeyInfo sortKeyInfo = getSortKeyInfo((POLocalRearrange) pred);
+ if (sortKeyInfo == null) {
+ log.debug("Cannot get sortKeyInfo from POLocalRearrange, skip secondary key optimizing");
+ return;
+ }
+ sortKeyInfos.add(sortKeyInfo);
+ }
+ }
+ } else {
+ log.debug("Cannot find POLocalRearrange or POUnion in map leaf, skip secondary key optimizing");
+ return;
+ }
+ } catch (ExecException e) {
+ log
+ .debug("Cannot get sortKeyInfo from POLocalRearrange, skip secondary key optimizing");
+ return;
+ }
+
+ if (mr.reducePlan.isEmpty()) {
+ log.debug("Reduce plan is empty, skip secondary key optimizing");
+ return;
+ }
+
+ List<PhysicalOperator> reduceRoots = mr.reducePlan.getRoots();
+ if (reduceRoots.size() != 1) {
+ log
+ .debug("Expected reduce to have single root, skip secondary key optimizing");
+ return;
+ }
+
+ PhysicalOperator root = reduceRoots.get(0);
+ if (!(root instanceof POPackage)) {
+ log
+ .debug("Expected reduce root to be a POPackage, skip secondary key optimizing");
+ return;
+ }
+
+ // visit the POForEach of the reduce plan. We can have Limit and Filter
+ // in the middle
+ PhysicalOperator currentNode = root;
+ POForEach foreach = null;
+ while (currentNode != null) {
+ if (currentNode instanceof POPackage && !(currentNode instanceof POJoinPackage)
+ || currentNode instanceof POFilter
+ || currentNode instanceof POLimit) {
+ List<PhysicalOperator> succs = mr.reducePlan
+ .getSuccessors(currentNode);
+ if (succs == null) // We didn't find POForEach
+ return;
+ if (succs.size() != 1) {
+ log.debug("See multiple output for " + currentNode
+ + " in reduce plan, skip secondary key optimizing");
+ return;
+ }
+ currentNode = succs.get(0);
+ } else if (currentNode instanceof POForEach) {
+ foreach = (POForEach) currentNode;
+ break;
+ } else { // Skip optimization
+ return;
+ }
+ }
+
+ // We do not find a foreach (we shall not come here, a trick to fool findbugs)
+ if (foreach==null)
+ return;
+
+ sortsToRemove = new ArrayList<POToChange>();
+ distinctsToChange = new ArrayList<POToChange>();
+
+ for (PhysicalPlan innerPlan : foreach.getInputPlans()) {
+ // visit inner plans to figure out the sort order for distinct /
+ // sort
+ SecondaryKeyDiscoverVisitor innerPlanVisitor = new SecondaryKeyDiscoverVisitor(
+ innerPlan, sortKeyInfos, secondarySortKeyInfo);
+ innerPlanVisitor.visit();
+ secondarySortKeyInfo = innerPlanVisitor.getSecondarySortKeyInfo();
+ if (innerPlanVisitor.getSortsToRemove() != null) {
+ for (POSort sort : innerPlanVisitor.getSortsToRemove()) {
+ sortsToRemove.add(new POToChange(sort, innerPlan, foreach));
+ }
+ }
+ if (innerPlanVisitor.getDistinctsToChange() != null) {
+ for (PODistinct distinct : innerPlanVisitor
+ .getDistinctsToChange()) {
+ distinctsToChange.add(new POToChange(distinct, innerPlan,
+ foreach));
+ }
+ }
+ }
+
+ try {
+ // Change PODistinct to use POSortedDistinct, which assume the input
+ // data is sorted
+ for (POToChange distinctToChange : distinctsToChange) {
+ numDistinctChanged++;
+ PODistinct oldDistinct = (PODistinct) distinctToChange.oper;
+ String scope = oldDistinct.getOperatorKey().scope;
+ POSortedDistinct newDistinct = new POSortedDistinct(
+ new OperatorKey(scope, NodeIdGenerator.getGenerator()
+ .getNextNodeId(scope)), oldDistinct
+ .getRequestedParallelism(), oldDistinct
+ .getInputs());
+ newDistinct.setInputs(oldDistinct.getInputs());
+ newDistinct.setResultType(oldDistinct.getResultType());
+ distinctToChange.plan.replace(oldDistinct, newDistinct);
+ distinctToChange.forEach.getLeaves();
+ }
+ // Removed POSort, if the successor require a databag, we need to
+ // add a PORelationToExprProject
+ // to convert tuples into databag
+ for (POToChange sortToRemove : sortsToRemove) {
+ numSortRemoved++;
+ POSort oldSort = (POSort) sortToRemove.oper;
+ String scope = oldSort.getOperatorKey().scope;
+ List<PhysicalOperator> preds = sortToRemove.plan
+ .getPredecessors(sortToRemove.oper);
+ POProject project = null;
+ if (preds == null
+ || preds.get(0).getResultType() != DataType.BAG
+ && oldSort.getResultType() == DataType.BAG) {
+ project = new PORelationToExprProject(new OperatorKey(
+ scope, NodeIdGenerator.getGenerator()
+ .getNextNodeId(scope)), oldSort
+ .getRequestedParallelism());
+ project.setInputs(oldSort.getInputs());
+ project.setResultType(DataType.BAG);
+ project.setStar(true);
+ }
+ if (project == null)
+ sortToRemove.plan.removeAndReconnect(sortToRemove.oper);
+ else
+ sortToRemove.plan.replace(oldSort, project);
+ sortToRemove.forEach.getLeaves();
+ }
+ } catch (PlanException e) {
+ int errorCode = 2202;
+ throw new VisitorException(
+ "Error change distinct/sort to use secondary key optimizer",
+ errorCode, e);
+ }
+ if (secondarySortKeyInfo != null) {
+ // Adjust POLocalRearrange, POPackage, MapReduceOper to use the
+ // secondary key
+ numMRUseSecondaryKey++;
+ mr.setUseSecondaryKey(true);
+ mr.setSecondarySortOrder(secondarySortKeyInfo.getAscs());
+ int indexOfRearrangeToChange = -1;
+ for (ColumnChainInfo columnChainInfo : secondarySortKeyInfo
+ .getColumnChains()) {
+ ColumnInfo currentColumn = columnChainInfo.getColumnInfos()
+ .get(0);
+ int index = currentColumn.columns.get(0);
+ if (indexOfRearrangeToChange == -1)
+ indexOfRearrangeToChange = index;
+ else if (indexOfRearrangeToChange != index) {
+ int errorCode = 2203;
+ throw new VisitorException("Sort on columns from different inputs.", errorCode);
+ }
+ }
+ if (mapLeaf instanceof POLocalRearrange) {
+ ((POLocalRearrange) mapLeaf).setUseSecondaryKey(true);
+ setSecondaryPlan(mr.mapPlan, (POLocalRearrange) mapLeaf,
+ secondarySortKeyInfo);
+ } else if (mapLeaf instanceof POUnion) {
+ List<PhysicalOperator> preds = mr.mapPlan
+ .getPredecessors(mapLeaf);
+ boolean found = false;
+ for (PhysicalOperator pred : preds) {
+ POLocalRearrange rearrange = (POLocalRearrange) pred;
+ rearrange.setUseSecondaryKey(true);
+ if (rearrange.getIndex() == indexOfRearrangeToChange) {
+ // Try to find the POLocalRearrange for the secondary key
+ found = true;
+ setSecondaryPlan(mr.mapPlan, rearrange, secondarySortKeyInfo);
+ }
+ }
+ if (!found)
+ {
+ int errorCode = 2209;
+ new VisitorException("Cannot find POLocalRearrange to set secondary plan", errorCode);
+ }
+ }
+ POPackage pack = (POPackage) root;
+ pack.setUseSecondaryKey(true);
+ }
+ }
+
+ void setSecondaryPlan(PhysicalPlan plan, POLocalRearrange rearrange,
+ SortKeyInfo secondarySortKeyInfo) throws VisitorException {
+ // Put plan to project secondary key to the POLocalRearrange
+ try {
+ String scope = rearrange.getOperatorKey().scope;
+ List<PhysicalPlan> secondaryPlanList = new ArrayList<PhysicalPlan>();
+ for (ColumnChainInfo columnChainInfo : secondarySortKeyInfo
+ .getColumnChains()) {
+ PhysicalPlan secondaryPlan = new PhysicalPlan();
+ for (int i = 1; i < columnChainInfo.size(); i++) {
+ // The first item in columnChainInfo indicate the index of
+ // input, we have addressed
+ // already before we come here
+ ColumnInfo columnInfo = columnChainInfo.getColumnInfo(i);
+ POProject project = new POProject(
+ new OperatorKey(scope, NodeIdGenerator
+ .getGenerator().getNextNodeId(scope)),
+ rearrange.getRequestedParallelism());
+ if (columnInfo.star)
+ project.setStar(true);
+ else
+ project
+ .setColumns((ArrayList<Integer>) columnInfo.columns);
+ project.setResultType(columnInfo.resultType);
+ secondaryPlan.addAsLeaf(project);
+ }
+ if (secondaryPlan.isEmpty()) { // If secondary key sort on the
+ // input as a whole
+ POProject project = new POProject(
+ new OperatorKey(scope, NodeIdGenerator
+ .getGenerator().getNextNodeId(scope)),
+ rearrange.getRequestedParallelism());
+ project.setStar(true);
+ secondaryPlan.addAsLeaf(project);
+ }
+ secondaryPlanList.add(secondaryPlan);
+ }
+ rearrange.setSecondaryPlans(secondaryPlanList);
+ } catch (PlanException e) {
+ int errorCode = 2204;
+ throw new VisitorException("Error setting secondary key plan",
+ errorCode, e);
+ }
+ }
+
+ public int getNumMRUseSecondaryKey() {
+ return numMRUseSecondaryKey;
+ }
+
+ public int getNumSortRemoved() {
+ return numSortRemoved;
+ }
+
+ public int getDistinctChanged() {
+ return numDistinctChanged;
+ }
+
+ // Find eligible sort and distinct physical operators from the reduce plan.
+ // SecondaryKeyChecker will check for sort/distinct keys (for distinct, it
+ // is
+ // always the entire bag), if it is the same with the main key, then we can
+ // just
+ // remove it. If it is not, then we can have 1 secondary sort key, put the
+ // first
+ // such sort/distinct key as the secondary sort key. For subsequent
+ // sort/distinct,
+ // we cannot do any secondary key optimization because we only have 1
+ // secondary
+ // sort key.
+ private static class SecondaryKeyDiscoverVisitor extends PhyPlanVisitor {
+ List<POSort> sortsToRemove = new ArrayList<POSort>();
+
+ List<PODistinct> distinctsToChange = new ArrayList<PODistinct>();
+
+ List<SortKeyInfo> sortKeyInfos;
+
+ SortKeyInfo secondarySortKeyInfo;
+
+ ColumnChainInfo columnChainInfo = new ColumnChainInfo();
+
+ boolean sawInvalidPhysicalOper = false;
+
+ // PhysicalPlan here is foreach inner plan
+ SecondaryKeyDiscoverVisitor(PhysicalPlan plan,
+ List<SortKeyInfo> sortKeyInfos, SortKeyInfo secondarySortKeyInfo) {
+ super(plan,
+ new DependencyOrderWalker<PhysicalOperator, PhysicalPlan>(
+ plan));
+ this.sortKeyInfos = sortKeyInfos;
+ this.secondarySortKeyInfo = secondarySortKeyInfo;
+ }
+
+ // We see PODistinct, check which key it is using
+ @Override
+ public void visitDistinct(PODistinct distinct) throws VisitorException {
+ if (sawInvalidPhysicalOper)
+ return;
+ SortKeyInfo keyInfos = new SortKeyInfo();
+ try {
+ keyInfos.insertColumnChainInfo(0,
+ (ColumnChainInfo) columnChainInfo.clone(), true);
+ } catch (CloneNotSupportedException e) { // We implement Clonable,
+ // impossible to get here
+ }
+
+ // if it is part of main key
+ for (SortKeyInfo sortKeyInfo : sortKeyInfos) {
+ if (sortKeyInfo.moreSpecificThan(keyInfos)) {
+ distinctsToChange.add(distinct);
+ return;
+ }
+ }
+
+ // if it is part of secondary key
+ if (secondarySortKeyInfo != null
+ && secondarySortKeyInfo.moreSpecificThan(keyInfos)) {
+ distinctsToChange.add(distinct);
+ return;
+ }
+
+ // Now set the secondary key
+ if (secondarySortKeyInfo == null) {
+ distinctsToChange.add(distinct);
+ secondarySortKeyInfo = keyInfos;
+ }
+ }
+
+ @Override
+ public void visitLimit(POLimit limit) throws VisitorException {
+ }
+
+ // Accumulate column info
+ @Override
+ public void visitProject(POProject project) throws VisitorException {
+ columnChainInfo.insertInReduce(project.isStar(), project
+ .getColumns(), project.getResultType());
+ }
+
+ @Override
+ public void visitFilter(POFilter filter) throws VisitorException {
+ }
+
+ @Override
+ public void visitUserFunc(POUserFunc userFunc) throws VisitorException {
+ sawInvalidPhysicalOper = true;
+ }
+
+ @Override
+ public void visitUnion(POUnion union) throws VisitorException {
+ sawInvalidPhysicalOper = true;
+ }
+
+ // Accumulate column info from nested project
+ @Override
+ public void visitPOForEach(POForEach fe) throws VisitorException {
+ if (fe.getInputPlans().size() > 1) {
+ // I may be wrong, but for now let's assume foreach plan before
+ // sort/distinct only have one foreach plan
+ throw new VisitorException(
+ "POForEach has more than 1 input plans");
+ }
+ boolean r = false;
+ try {
+ r = collectColumnChain(fe.getInputPlans().get(0),
+ columnChainInfo);
+ } catch (PlanException e) {
+ int errorCode = 2205;
+ throw new VisitorException("Error visiting POForEach inner plan",
+ errorCode, e);
+ }
+ // See something other than POProject in POForEach, set the flag to stop further processing
+ if (r)
+ sawInvalidPhysicalOper = true;
+ }
+
+ // We see POSort, check which key it is using
+ @Override
+ public void visitSort(POSort sort) throws VisitorException {
+ if (sawInvalidPhysicalOper)
+ return;
+ SortKeyInfo keyInfo = new SortKeyInfo();
+ for (int i = 0; i < sort.getSortPlans().size(); i++) {
+ PhysicalPlan sortPlan = sort.getSortPlans().get(i);
+ ColumnChainInfo sortChainInfo = null;
+ try {
+ sortChainInfo = (ColumnChainInfo) columnChainInfo.clone();
+ } catch (CloneNotSupportedException e) { // We implement
+ // Clonable, impossible
+ // to get here
+ }
+ boolean r = false;
+ try {
+ r = collectColumnChain(sortPlan, sortChainInfo);
+ } catch (PlanException e) {
+ int errorCode = 2206;
+ throw new VisitorException("Error visiting POSort inner plan",
+ errorCode, e);
+ }
+ if (r) // if we saw physical operator other than project in sort plan
+ {
+ sawInvalidPhysicalOper = true;
+ return;
+ }
+ keyInfo.insertColumnChainInfo(i, sortChainInfo, sort
+ .getMAscCols().get(i));
+ }
+ // if it is part of main key
+ for (SortKeyInfo sortKeyInfo : sortKeyInfos) {
+ if (sortKeyInfo.moreSpecificThan(keyInfo)) {
+ sortsToRemove.add(sort);
+ return;
+ }
+ }
+ // if it is part of secondary key
+ if (secondarySortKeyInfo != null
+ && secondarySortKeyInfo.moreSpecificThan(keyInfo)) {
+ sortsToRemove.add(sort);
+ return;
+ }
+
+ // Now set the secondary key
+ if (secondarySortKeyInfo == null) {
+ sortsToRemove.add(sort);
+ secondarySortKeyInfo = keyInfo;
+ }
+ }
+
+ public List<POSort> getSortsToRemove() {
+ return sortsToRemove;
+ }
+
+ public List<PODistinct> getDistinctsToChange() {
+ return distinctsToChange;
+ }
+
+ public SortKeyInfo getSecondarySortKeyInfo() {
+ return secondarySortKeyInfo;
+ }
+ }
+
+ // Return true if we saw physical operators other than project in the plan
+ static private boolean collectColumnChain(PhysicalPlan plan,
+ ColumnChainInfo columnChainInfo) throws PlanException {
+ if (plan.getRoots().size() != 1) {
+ int errorCode = 2207;
+ throw new PlanException(
+ "POForEach inner plan has more than 1 root", errorCode);
+ }
+
+ PhysicalOperator currentNode = plan.getRoots().get(0);
+
+ while (currentNode != null) {
+ if (currentNode instanceof POProject) {
+ POProject project = (POProject) currentNode;
+ columnChainInfo.insertInReduce(project.isStar(), project
+ .getColumns(), project.getResultType());
+ } else {
+ return true;
+ }
+ List<PhysicalOperator> succs = plan.getSuccessors(currentNode);
+ if (succs == null)
+ break;
+ if (succs.size() != 1) {
+ int errorCode = 2208;
+ throw new PlanException(
+ "Exception visiting foreach inner plan", errorCode);
+ }
+ currentNode = succs.get(0);
+ }
+ return false;
+ }
+}
Added: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/SortKeyInfo.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/SortKeyInfo.java?rev=883836&view=auto
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/SortKeyInfo.java (added)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/SortKeyInfo.java Tue Nov 24 19:54:19 2009
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.pig.backend.hadoop.executionengine.mapReduceLayer;
+
+import java.util.ArrayList;
+import java.util.List;
+
+// Represent a set of sort keys. All the sort keys are equivalent.
+// Eg:
+// C = cogroup A by ($0, $1), B by ($0, $1)
+// This query have two sort keys, (A.$0, A.$1) and (B.$0, B.$1)
+public class SortKeyInfo {
+ private List<ColumnChainInfo> columnChains = new ArrayList<ColumnChainInfo>();
+ private List<Boolean> ascs = new ArrayList<Boolean>();
+
+ public boolean moreSpecificThan(Object o2)
+ {
+ if (!(o2 instanceof SortKeyInfo))
+ return false;
+ SortKeyInfo s2 = (SortKeyInfo)o2;
+
+ if (columnChains.size()<s2.columnChains.size())
+ return false;
+ for (int i=0;i<columnChains.size();i++)
+ {
+ if (i<s2.columnChains.size() && (!(columnChains.get(i).equals(s2.columnChains.get(i))) ||
+ !ascs.get(i).equals(s2.ascs.get(i))))
+ return false;
+ }
+ return true;
+ }
+
+ public void insertColumnChainInfo(int index, ColumnChainInfo columnChainInfo, boolean asc)
+ {
+ if (columnChains.size()<=index)
+ {
+ columnChains.add(columnChainInfo);
+ ascs.add(asc);
+ return;
+ }
+
+ ColumnChainInfo chain = columnChains.get(index);
+ chain.insertColumnChainInfo(columnChainInfo);
+ ascs.set(index, asc);
+ }
+ public String toString() {
+ return columnChains.toString();
+ }
+
+ public List<ColumnChainInfo> getColumnChains() {
+ return columnChains;
+ }
+
+ public boolean[] getAscs() {
+ boolean[] result = new boolean[ascs.size()];
+ for (int i=0;i<ascs.size();i++)
+ result[i] = ascs.get(i);
+ return result;
+ }
+}
Added: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/partitioners/SecondaryKeyPartitioner.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/partitioners/SecondaryKeyPartitioner.java?rev=883836&view=auto
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/partitioners/SecondaryKeyPartitioner.java (added)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/partitioners/SecondaryKeyPartitioner.java Tue Nov 24 19:54:19 2009
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.partitioners;
+
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.HDataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.io.PigNullableWritable;
+
+public class SecondaryKeyPartitioner extends HashPartitioner<PigNullableWritable
+, Writable> implements Configurable {
+ Byte kt;
+ @Override
+ public int getPartition(PigNullableWritable key, Writable value,
+ int numPartitions) {
+ try {
+ Tuple t = (Tuple)key.getValueAsPigType();
+ PigNullableWritable realKey = HDataType.getWritableComparableTypes(t.get(0), kt);
+ return super.getPartition(realKey, value, numPartitions);
+ } catch (ExecException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Configuration getConf() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setConf(Configuration conf) {
+ String kts = conf.get("pig.reduce.key.type");
+ kt = Byte.valueOf(kts);
+ }
+}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/plans/MRPrinter.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/plans/MRPrinter.java?rev=883836&r1=883835&r2=883836&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/plans/MRPrinter.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/plans/MRPrinter.java Tue Nov 24 19:54:19 2009
@@ -81,6 +81,8 @@
if (mr.getQuantFile() != null) {
mStream.println("Quantile file: " + mr.getQuantFile());
}
+ if (mr.getUseSecondaryKey())
+ mStream.println("Secondary sort: " + mr.getUseSecondaryKey());
mStream.println("----------------");
mStream.println("");
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/plans/POPackageAnnotator.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/plans/POPackageAnnotator.java?rev=883836&r1=883835&r2=883836&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/plans/POPackageAnnotator.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/plans/POPackageAnnotator.java Tue Nov 24 19:54:19 2009
@@ -212,7 +212,7 @@
if(keyInfo == null)
keyInfo = new HashMap<Integer, Pair<Boolean, Map<Integer, Integer>>>();
- if(keyInfo.get(lrearrange.getIndex()) != null) {
+ if(keyInfo.get(Integer.valueOf(lrearrange.getIndex())) != null) {
// something is wrong - we should not be getting key info
// for the same index from two different Local Rearranges
int errCode = 2087;
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/LogToPhyTranslationVisitor.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/LogToPhyTranslationVisitor.java?rev=883836&r1=883835&r2=883836&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/LogToPhyTranslationVisitor.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/LogToPhyTranslationVisitor.java Tue Nov 24 19:54:19 2009
@@ -62,6 +62,7 @@
import org.apache.pig.impl.plan.VisitorException;
+import org.apache.pig.impl.util.CompilerUtils;
import org.apache.pig.impl.util.LinkedMultiMap;
import org.apache.pig.impl.util.MultiMap;
@@ -834,7 +835,7 @@
POSkewedJoin skj;
try {
skj = new POSkewedJoin(new OperatorKey(scope,nodeGen.getNextNodeId(scope)),loj.getRequestedParallelism(),
- inp);
+ inp, loj.getInnerFlags());
skj.setJoinPlans(joinPlans);
}
catch (Exception e) {
@@ -843,6 +844,30 @@
throw new LogicalToPhysicalTranslatorException(msg, errCode, PigException.BUG, e);
}
skj.setResultType(DataType.TUPLE);
+
+ boolean[] innerFlags = loj.getInnerFlags();
+ for (int i=0; i < inputs.size(); i++) {
+ LogicalOperator op = inputs.get(i);
+ if (!innerFlags[i]) {
+ try {
+ Schema s = op.getSchema();
+ // if the schema cannot be determined
+ if (s == null) {
+ throw new FrontendException();
+ }
+ skj.addSchema(s);
+ } catch (FrontendException e) {
+ int errCode = 2015;
+ String msg = "Couldn't set the schema for outer join" ;
+ throw new LogicalToPhysicalTranslatorException(msg, errCode, PigException.BUG, e);
+ }
+ } else {
+ // This will never be retrieved. It just guarantees that the index will be valid when
+ // MRCompiler is trying to read the schema
+ skj.addSchema(null);
+ }
+ }
+
currentPlan.add(skj);
for (LogicalOperator op : inputs) {
@@ -855,15 +880,49 @@
}
}
logToPhyMap.put(loj, skj);
- }
-
+ }
else if(loj.getJoinType() == LOJoin.JOINTYPE.REPLICATED) {
int fragment = 0;
POFRJoin pfrj;
try {
+ boolean []innerFlags = loj.getInnerFlags();
+ boolean isLeftOuter = false;
+ // We dont check for bounds issue as we assume that a join
+ // involves atleast two inputs
+ isLeftOuter = !innerFlags[1];
+
+ Tuple nullTuple = null;
+ if( isLeftOuter ) {
+ try {
+ // We know that in a Left outer join its only a two way
+ // join, so we assume index of 1 for the right input
+ Schema inputSchema = inputs.get(1).getSchema();
+
+ // We check if we have a schema before the join
+ if(inputSchema == null) {
+ int errCode = 1109;
+ String msg = "Input (" + inputs.get(1).getAlias() + ") " +
+ "on which outer join is desired should have a valid schema";
+ throw new LogicalToPhysicalTranslatorException(msg, errCode, PigException.INPUT);
+ }
+
+ // Using the schema we decide the number of columns/fields
+ // in the nullTuple
+ nullTuple = TupleFactory.getInstance().newTuple(inputSchema.size());
+ for(int j = 0; j < inputSchema.size(); j++) {
+ nullTuple.set(j, null);
+ }
+
+ } catch( FrontendException e ) {
+ int errCode = 2104;
+ String msg = "Error while determining the schema of input";
+ throw new LogicalToPhysicalTranslatorException(msg, errCode, PigException.BUG, e);
+ }
+ }
+
pfrj = new POFRJoin(new OperatorKey(scope,nodeGen.getNextNodeId(scope)),loj.getRequestedParallelism(),
- inp, ppLists, keyTypes, null, fragment);
+ inp, ppLists, keyTypes, null, fragment, isLeftOuter, nullTuple);
} catch (ExecException e1) {
int errCode = 2058;
String msg = "Unable to set index on newly create POLocalRearrange.";
@@ -1045,85 +1104,21 @@
Schema inputSchema = null;
try {
inputSchema = joinInput.getSchema();
-
-
+
+
if(inputSchema == null) {
- int errCode = 1105;
+ int errCode = 1109;
String msg = "Input (" + joinInput.getAlias() + ") " +
"on which outer join is desired should have a valid schema";
throw new LogicalToPhysicalTranslatorException(msg, errCode, PigException.INPUT);
}
} catch (FrontendException e) {
- int errCode = 2014;
+ int errCode = 2104;
String msg = "Error while determining the schema of input";
throw new LogicalToPhysicalTranslatorException(msg, errCode, PigException.BUG, e);
}
- // we currently have POProject[bag] as the only operator in the plan
- // If the bag is an empty bag, we should replace
- // it with a bag with one tuple with null fields so that when we flatten
- // we do not drop records (flatten will drop records if the bag is left
- // as an empty bag) and actually project nulls for the fields in
- // the empty bag
-
- // So we need to get to the following state:
- // POProject[Bag]
- // \
- // POUserFunc["IsEmpty()"] Const[Bag](bag with null fields)
- // \ | POProject[Bag]
- // \ | /
- // POBinCond
-
- POProject relationProject = (POProject) fePlan.getRoots().get(0);
- try {
-
- // condition of the bincond
- POProject relationProjectForIsEmpty = relationProject.clone();
- fePlan.add(relationProjectForIsEmpty);
- String scope = relationProject.getOperatorKey().scope;
- FuncSpec isEmptySpec = new FuncSpec(IsEmpty.class.getName());
- Object f = PigContext.instantiateFuncFromSpec(isEmptySpec);
- POUserFunc isEmpty = new POUserFunc(new OperatorKey(scope, NodeIdGenerator.getGenerator().
- getNextNodeId(scope)), -1, null, isEmptySpec, (EvalFunc) f);
- isEmpty.setResultType(DataType.BOOLEAN);
- fePlan.add(isEmpty);
- fePlan.connect(relationProjectForIsEmpty, isEmpty);
-
- // lhs of bincond (const bag with null fields)
- ConstantExpression ce = new ConstantExpression(new OperatorKey(scope,
- NodeIdGenerator.getGenerator().getNextNodeId(scope)));
- // the following should give a tuple with the
- // required number of nulls
- Tuple t = TupleFactory.getInstance().newTuple(inputSchema.size());
- for(int i = 0; i < inputSchema.size(); i++) {
- t.set(i, null);
- }
- List<Tuple> bagContents = new ArrayList<Tuple>(1);
- bagContents.add(t);
- DataBag bg = new NonSpillableDataBag(bagContents);
- ce.setValue(bg);
- ce.setResultType(DataType.BAG);
- //this operator doesn't have any predecessors
- fePlan.add(ce);
-
- //rhs of bincond is the original project
- // let's set up the bincond now
- POBinCond bincond = new POBinCond(new OperatorKey(scope,
- NodeIdGenerator.getGenerator().getNextNodeId(scope)));
- bincond.setCond(isEmpty);
- bincond.setLhs(ce);
- bincond.setRhs(relationProject);
- bincond.setResultType(DataType.BAG);
- fePlan.add(bincond);
-
- fePlan.connect(isEmpty, bincond);
- fePlan.connect(ce, bincond);
- fePlan.connect(relationProject, bincond);
-
- } catch (Exception e) {
- throw new PlanException("Error setting up outerjoin", e);
- }
-
+ CompilerUtils.addEmptyBagOuterJoin(fePlan, inputSchema);
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/POStatus.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/POStatus.java?rev=883836&r1=883835&r2=883836&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/POStatus.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/POStatus.java Tue Nov 24 19:54:19 2009
@@ -30,5 +30,9 @@
// between ExecutableManager and POStream
public static final byte STATUS_EOS = 4; // end of Streaming output (i.e. output from streaming binary)
+ // successfully processing of a batch, used by accumulative UDFs
+ // this is used for accumulative UDFs
+ public static final byte STATUS_BATCH_OK = 5;
+
public static Object result;
}