You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by da...@apache.org on 2014/05/30 21:07:29 UTC
svn commit: r1598702 [6/23] - in /pig/trunk: ./ ivy/
shims/src/hadoop23/org/apache/pig/backend/hadoop23/
shims/test/hadoop20/org/apache/pig/test/
shims/test/hadoop23/org/apache/pig/test/ src/META-INF/services/
src/org/apache/pig/ src/org/apache/pig/bac...
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/CombinerPackager.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/CombinerPackager.java?rev=1598702&r1=1598701&r2=1598702&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/CombinerPackager.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/CombinerPackager.java Fri May 30 19:07:23 2014
@@ -39,8 +39,11 @@ import org.apache.pig.impl.util.Pair;
* offset in the tuple.
*/
public class CombinerPackager extends Packager {
+
+ private static final long serialVersionUID = 1L;
+
private boolean[] mBags; // For each field, indicates whether or not it
- // needs to be put in a bag.
+ // needs to be put in a bag.
private Map<Integer, Integer> keyLookup;
@@ -57,7 +60,7 @@ public class CombinerPackager extends Pa
super();
keyType = pkg.keyType;
numInputs = 1;
- inner = new boolean[1];
+ inner = new boolean[pkg.inner.length];
for (int i = 0; i < pkg.inner.length; i++) {
inner[i] = true;
}
@@ -105,7 +108,7 @@ public class CombinerPackager extends Pa
return new Result(POStatus.STATUS_EOP, null);
}
- //Create numInputs bags
+ // Create numInputs bags
Object[] fields = new Object[mBags.length];
for (int i = 0; i < mBags.length; i++) {
if (mBags[i]) fields[i] = createDataBag(numBags);
@@ -116,7 +119,7 @@ public class CombinerPackager extends Pa
// set the value as is.
for (Tuple tup : bags[0]) {
int tupIndex = 0; // an index for accessing elements from
- // the value (tup) that we have currently
+ // the value (tup) that we have currently
for(int i = 0; i < mBags.length; i++) {
Integer keyIndex = keyLookup.get(i);
if(keyIndex == null && mBags[i]) {
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/JoinPackager.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/JoinPackager.java?rev=1598702&r1=1598701&r2=1598702&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/JoinPackager.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/JoinPackager.java Fri May 30 19:07:23 2014
@@ -25,7 +25,6 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
import org.apache.pig.data.DataBag;
-import org.apache.pig.data.InternalCachedBag;
import org.apache.pig.data.NonSpillableDataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.plan.NodeIdGenerator;
@@ -34,6 +33,7 @@ import org.apache.pig.pen.Illustrator;
public class JoinPackager extends Packager {
+ private static final long serialVersionUID = 1L;
private POOptimizedForEach forEach;
private boolean newKey = true;
private Tuple res = null;
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/LitePackager.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/LitePackager.java?rev=1598702&r1=1598701&r2=1598702&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/LitePackager.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/LitePackager.java Fri May 30 19:07:23 2014
@@ -24,7 +24,6 @@ package org.apache.pig.backend.hadoop.ex
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
-import java.util.Map.Entry;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
@@ -45,6 +44,7 @@ import org.apache.pig.pen.util.LineageTr
*/
public class LitePackager extends Packager {
+ private static final long serialVersionUID = 1L;
private PigNullableWritable keyWritable;
@Override
@@ -79,8 +79,7 @@ public class LitePackager extends Packag
}
/**
- * @param distinct
- * the distinct to set
+ * @param distinct the distinct to set
*/
@Override
public void setDistinct(boolean distinct) {
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POCounter.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POCounter.java?rev=1598702&r1=1598701&r2=1598702&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POCounter.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POCounter.java Fri May 30 19:07:23 2014
@@ -32,7 +32,6 @@ import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.pen.util.ExampleTuple;
-import org.apache.pig.pen.util.LineageTracer;
/**
* This operator is part of the RANK operator implementation.
@@ -80,7 +79,7 @@ public class POCounter extends PhysicalO
/**
* Task ID to label each tuple analyzed by the corresponding task
**/
- private String taskID = "-1";
+ private Integer taskID = -1;
/**
* Unique identifier that links POCounter and PORank,
@@ -104,6 +103,15 @@ public class POCounter extends PhysicalO
super(k, rp, inputs);
}
+ public POCounter(POCounter copy) {
+ super(copy);
+ this.counterPlans = copy.counterPlans;
+ this.mAscCols = copy.mAscCols;
+ this.isDenseRank = copy.isDenseRank;
+ this.isRowNumber = copy.isRowNumber;
+ this.operationID = copy.operationID;
+ }
+
@SuppressWarnings({ "rawtypes", "unchecked" })
public POCounter(OperatorKey operatorKey, int requestedParallelism,
List inp, List<PhysicalPlan> counterPlans,
@@ -156,22 +164,21 @@ public class POCounter extends PhysicalO
Tuple in = (Tuple) input.result;
Tuple out = mTupleFactory.newTuple(in.getAll().size() + 2);
Long sizeBag = 0L;
- int positionBag, i = 2;
+ int positionBag, i = 1;
// Tuples are added by two stamps before the tuple content:
- // 1.- At position 0: Current taskId
- out.set(0, getTaskId());
+ // 1.- At position 0: counter value
+ // 2.- At position last: Current taskId
- // 2.- At position 1: counter value
//On this case, each tuple is analyzed independently of the tuples grouped
if(isRowNumber() || isDenseRank()) {
//Only when is Dense Rank (attached to a reduce phase) it is incremented on this way
//Otherwise, the increment is done at mapper automatically
if(isDenseRank())
- PigMapReduceCounter.PigReduceCounter.incrementCounter(POCounter.ONE);
+ incrementReduceCounter(POCounter.ONE);
- out.set(1, getLocalCounter());
+ out.set(0, getLocalCounter());
//and the local incrementer is sequentially increased.
incrementLocalCounter();
@@ -186,9 +193,9 @@ public class POCounter extends PhysicalO
//This value (the size of the tuples on the bag) is used to increment
//the current global counter and
- PigMapReduceCounter.PigReduceCounter.incrementCounter(sizeBag);
+ incrementReduceCounter(sizeBag);
- out.set(1, getLocalCounter());
+ out.set(0, getLocalCounter());
//the value for the next tuple on the current task
addToLocalCounter(sizeBag);
@@ -199,11 +206,18 @@ public class POCounter extends PhysicalO
out.set(i++, o);
}
+ // At position last: Current taskId
+ out.set(i++, getTaskId());
+
input.result = illustratorMarkup(in, out, 0);
return input;
}
+ protected void incrementReduceCounter(Long increment) {
+ PigMapReduceCounter.PigReduceCounter.incrementCounter(increment);
+ }
+
@Override
public boolean supportsMultipleInputs() {
return false;
@@ -248,7 +262,7 @@ public class POCounter extends PhysicalO
/**
* Sequential counter used at ROW NUMBER and RANK BY DENSE mode
**/
- public Long incrementLocalCounter() {
+ protected Long incrementLocalCounter() {
return localCount++;
}
@@ -260,18 +274,18 @@ public class POCounter extends PhysicalO
return this.localCount;
}
- public void addToLocalCounter(Long sizeBag) {
+ protected void addToLocalCounter(Long sizeBag) {
this.localCount += sizeBag;
}
/**
* Task ID: identifier of the task (map or reducer)
**/
- public void setTaskId(String taskID) {
+ public void setTaskId(int taskID) {
this.taskID = taskID;
}
- public String getTaskId() {
+ public int getTaskId() {
return this.taskID;
}
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POFRJoin.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POFRJoin.java?rev=1598702&r1=1598701&r2=1598702&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POFRJoin.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POFRJoin.java Fri May 30 19:07:23 2014
@@ -67,48 +67,44 @@ import org.apache.pig.impl.plan.VisitorE
*/
// We intentionally skip type checking in backend for performance reasons
-@SuppressWarnings("unchecked")
public class POFRJoin extends PhysicalOperator {
private static final Log log = LogFactory.getLog(POFRJoin.class);
- /**
- *
- */
private static final long serialVersionUID = 1L;
// The number in the input list which denotes the fragmented input
- private int fragment;
+ protected int fragment;
// There can be n inputs each being a List<PhysicalPlan>
// Ex. join A by ($0+$1,$0-$1), B by ($0*$1,$0/$1);
- private List<List<PhysicalPlan>> phyPlanLists;
+ protected List<List<PhysicalPlan>> phyPlanLists;
// The key type for each Local Rearrange operator
- private List<List<Byte>> keyTypes;
+ protected List<List<Byte>> keyTypes;
// The Local Rearrange operators modeling the join key
- private POLocalRearrange[] LRs;
+ protected POLocalRearrange[] LRs;
// The set of files that represent the replicated inputs
- private FileSpec[] replFiles;
+ protected FileSpec[] replFiles;
// Used to configure the foreach operator
- private ConstantExpression[] constExps;
+ protected ConstantExpression[] constExps;
// Used to produce the cross product of various bags
- private POForEach fe;
+ protected POForEach fe;
// The array of Hashtables one per replicated input. replicates[fragment] =
// null
// fragment is the input which is fragmented and not replicated.
- private TupleToMapKey replicates[];
+ protected TupleToMapKey replicates[];
// varaible which denotes whether we are returning tuples from the foreach
// operator
- private boolean processingPlan;
+ protected boolean processingPlan;
// A dummy tuple
- private Tuple dumTup = TupleFactory.getInstance().newTuple(1);
+ protected Tuple dumTup = TupleFactory.getInstance().newTuple(1);
// An instance of tuple factory
- private transient TupleFactory mTupleFactory;
- private boolean setUp;
+ protected transient TupleFactory mTupleFactory;
+ protected boolean setUp;
// A Boolean variable which denotes if this is a LeftOuter Join or an Inner
// Join
- private boolean isLeftOuterJoin;
+ protected boolean isLeftOuterJoin;
- // This list contains nullTuples according to schema of various inputs
- private DataBag nullBag;
- private Schema[] inputSchemas;
- private Schema[] keySchemas;
+ // This list contains nullTuples according to schema of various inputs
+ protected DataBag nullBag;
+ protected Schema[] inputSchemas;
+ protected Schema[] keySchemas;
public POFRJoin(OperatorKey k, int rp, List<PhysicalOperator> inp,
List<List<PhysicalPlan>> ppLists, List<List<Byte>> keyTypes,
@@ -152,8 +148,22 @@ public class POFRJoin extends PhysicalOp
}
}
- public List<List<PhysicalPlan>> getJoinPlans() {
- return phyPlanLists;
+ public POFRJoin(POFRJoin copy) throws ExecException {
+ super(copy);
+ this.phyPlanLists = copy.phyPlanLists;
+ this.fragment = copy.fragment;
+ this.keyTypes = copy.keyTypes;
+ this.replFiles = copy.replFiles;
+ this.replicates = copy.replicates;
+ this.LRs = copy.LRs;
+ this.fe = copy.fe;
+ this.constExps = copy.constExps;
+ this.processingPlan = copy.processingPlan;
+ this.mTupleFactory = copy.mTupleFactory;
+ this.nullBag = copy.nullBag;
+ this.isLeftOuterJoin = copy.isLeftOuterJoin;
+ this.inputSchemas = copy.inputSchemas;
+ this.keySchemas = copy.keySchemas;
}
private OperatorKey genKey(OperatorKey old) {
@@ -259,9 +269,9 @@ public class POFRJoin extends PhysicalOp
// Process the current input
inp = processInput();
if (inp.returnStatus == POStatus.STATUS_EOP
- || inp.returnStatus == POStatus.STATUS_ERR)
+ || inp.returnStatus == POStatus.STATUS_ERR) {
return inp;
- if (inp.returnStatus == POStatus.STATUS_NULL) {
+ } else if (inp.returnStatus == POStatus.STATUS_NULL) {
continue;
}
@@ -301,8 +311,9 @@ public class POFRJoin extends PhysicalOp
// If this is not LeftOuter Join and there was no match we
// skip the processing of this left tuple and move ahead
- if (!isLeftOuterJoin && noMatch)
+ if (!isLeftOuterJoin && noMatch) {
continue;
+ }
fe.attachInput(dumTup);
processingPlan = true;
@@ -318,7 +329,7 @@ public class POFRJoin extends PhysicalOp
}
}
- private static class TupleToMapKey {
+ protected static class TupleToMapKey {
private HashMap<Tuple, TuplesToSchemaTupleList> tuples;
private SchemaTupleFactory tf;
@@ -345,10 +356,10 @@ public class POFRJoin extends PhysicalOp
/**
* Builds the HashMaps by reading each replicated input from the DFS using a
* Load operator
- *
+ *
* @throws ExecException
*/
- private void setUpHashMap() throws ExecException {
+ protected void setUpHashMap() throws ExecException {
SchemaTupleFactory[] inputSchemaTupleFactories = new SchemaTupleFactory[inputSchemas.length];
SchemaTupleFactory[] keySchemaTupleFactories = new SchemaTupleFactory[inputSchemas.length];
for (int i = 0; i < inputSchemas.length; i++) {
@@ -416,7 +427,7 @@ public class POFRJoin extends PhysicalOp
log.debug("Hash Table built. Time taken: " + (time2 - time1));
}
- private boolean isKeyNull(Object key) throws ExecException {
+ protected boolean isKeyNull(Object key) throws ExecException {
if (key == null) return true;
if (key instanceof Tuple) {
Tuple t = (Tuple)key;
@@ -437,7 +448,7 @@ public class POFRJoin extends PhysicalOp
/*
* Extracts the value tuple from the LR operator's output tuple
*/
- private Tuple getValueTuple(POLocalRearrange lr, Tuple tuple)
+ protected Tuple getValueTuple(POLocalRearrange lr, Tuple tuple)
throws ExecException {
Tuple val = (Tuple) tuple.get(2);
Tuple retTup = null;
@@ -491,6 +502,14 @@ public class POFRJoin extends PhysicalOp
return retTup;
}
+ public List<List<PhysicalPlan>> getJoinPlans() {
+ return phyPlanLists;
+ }
+
+ public POLocalRearrange[] getLRs() {
+ return LRs;
+ }
+
public int getFragment() {
return fragment;
}
@@ -506,7 +525,7 @@ public class POFRJoin extends PhysicalOp
public void setReplFiles(FileSpec[] replFiles) {
this.replFiles = replFiles;
}
-
+
@Override
public Tuple illustratorMarkup(Object in, Object out, int eqClassIndex) {
// no op: all handled by the preceding POForEach
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLimit.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLimit.java?rev=1598702&r1=1598701&r2=1598702&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLimit.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLimit.java Fri May 30 19:07:23 2014
@@ -34,13 +34,13 @@ import org.apache.pig.pen.util.ExampleTu
public class POLimit extends PhysicalOperator {
/**
- *
+ *
*/
private static final long serialVersionUID = 1L;
// Counts for outputs processed
private long soFar = 0;
-
+
// Number of limited outputs
long mLimit;
@@ -62,11 +62,11 @@ public class POLimit extends PhysicalOpe
public POLimit(OperatorKey k, int rp, List<PhysicalOperator> inputs) {
super(k, rp, inputs);
}
-
+
public void setLimit(long limit) {
mLimit = limit;
}
-
+
public long getLimit() {
return mLimit;
}
@@ -80,8 +80,8 @@ public class POLimit extends PhysicalOpe
}
/**
- * Counts the number of tuples processed into static variable soFar, if the number of tuples processed reach the
- * limit, return EOP; Otherwise, return the tuple
+ * Counts the number of tuples processed into static variable soFar, if the number of tuples processed reach the
+ * limit, return EOP; Otherwise, return the tuple
*/
@Override
public Result getNextTuple() throws ExecException {
@@ -117,12 +117,12 @@ public class POLimit extends PhysicalOpe
inp = processInput();
if (inp.returnStatus == POStatus.STATUS_EOP || inp.returnStatus == POStatus.STATUS_ERR)
break;
-
+
illustratorMarkup(inp.result, null, 0);
// illustrator ignore LIMIT before the post processing
if ((illustrator == null || illustrator.getOriginalLimit() != -1) && soFar>=mLimit)
inp.returnStatus = POStatus.STATUS_EOP;
-
+
soFar++;
break;
}
@@ -161,11 +161,13 @@ public class POLimit extends PhysicalOpe
NodeIdGenerator.getGenerator().getNextNodeId(this.mKey.scope)),
this.requestedParallelism, this.inputs);
newLimit.mLimit = this.mLimit;
- newLimit.expressionPlan = this.expressionPlan.clone();
+ if (this.expressionPlan != null) {
+ newLimit.expressionPlan = this.expressionPlan.clone();
+ }
newLimit.addOriginalLocation(alias, getOriginalLocations());
return newLimit;
}
-
+
@Override
public Tuple illustratorMarkup(Object in, Object out, int eqClassIndex) {
if(illustrator != null) {
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLocalRearrange.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLocalRearrange.java?rev=1598702&r1=1598701&r2=1598702&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLocalRearrange.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLocalRearrange.java Fri May 30 19:07:23 2014
@@ -82,6 +82,8 @@ public class POLocalRearrange extends Ph
protected boolean isCross = false;
+ protected Result inp;
+
// map to store mapping of projected columns to
// the position in the "Key" where these will be projected to.
// We use this information to strip off these columns
@@ -152,6 +154,32 @@ public class POLocalRearrange extends Ph
mSecondaryProjectedColsMap = new HashMap<Integer, Integer>();
}
+ public POLocalRearrange(POLocalRearrange copy) {
+ super(copy);
+ this.plans = copy.plans;
+ this.secondaryPlans = copy.secondaryPlans;
+ this.leafOps = copy.leafOps;
+ this.secondaryLeafOps = copy.secondaryLeafOps;
+ this.index = copy.index;
+ this.keyType = copy.keyType;
+ this.mainKeyType = copy.mainKeyType;
+ this.secondaryKeyType = copy.secondaryKeyType;
+ this.mIsDistinct = copy.mIsDistinct;
+ this.isCross = copy.isCross;
+ this.mProjectedColsMap = copy.mProjectedColsMap;
+ this.mSecondaryProjectedColsMap = copy.mSecondaryProjectedColsMap;
+ this.mFakeTuple = copy.mFakeTuple;
+ this.mProjectStar = copy.mProjectStar;
+ this.mSecondaryProjectStar = copy.mSecondaryProjectStar;
+ this.isKeyTuple = copy.isKeyTuple;
+ this.isKeyCompound = copy.isKeyCompound;
+ this.isSecondaryKeyTuple = copy.isSecondaryKeyTuple;
+ this.mProjectedColsMapSize = copy.mProjectedColsMapSize;
+ this.mSecondaryProjectedColsMapSize = copy.mSecondaryProjectedColsMapSize;
+ this.useSecondaryKey = copy.useSecondaryKey;
+ this.stripKeyFromValue = copy.stripKeyFromValue;
+ }
+
@Override
public void visit(PhyPlanVisitor v) throws VisitorException {
v.visitLocalRearrange(this);
@@ -257,7 +285,7 @@ public class POLocalRearrange extends Ph
@Override
public Result getNextTuple() throws ExecException {
- Result inp = null;
+ inp = null;
Result res = ERR_RESULT;
while (true) {
inp = processInput();
@@ -501,11 +529,16 @@ public class POLocalRearrange extends Ph
return keyType;
}
+ public byte getMainKeyType() {
+ return mainKeyType;
+ }
+
public void setKeyType(byte keyType) {
if (useSecondaryKey) {
this.mainKeyType = keyType;
} else {
this.keyType = keyType;
+ this.mainKeyType = keyType;
}
}
@@ -676,32 +709,50 @@ public class POLocalRearrange extends Ph
*/
@Override
public POLocalRearrange clone() throws CloneNotSupportedException {
- List<PhysicalPlan> clonePlans = new
- ArrayList<PhysicalPlan>(plans.size());
- for (PhysicalPlan plan : plans) {
- clonePlans.add(plan.clone());
- }
POLocalRearrange clone = new POLocalRearrange(new OperatorKey(
mKey.scope,
NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)),
requestedParallelism);
+ deepCopyTo(clone);
+ return clone;
+ }
+
+ protected void deepCopyTo(POLocalRearrange clone)
+ throws CloneNotSupportedException {
+
+ clone.setParentPlan(parentPlan);
+ clone.index = index;
+ if (useSecondaryKey) {
+ clone.keyType = mainKeyType;
+ } else {
+ clone.keyType = keyType;
+ }
+ clone.setUseSecondaryKey(useSecondaryKey);
+ // Needs to be called as setDistinct so that the fake index tuple gets
+ // created.
+ clone.setDistinct(mIsDistinct);
+ clone.setCross(isCross);
+ clone.addOriginalLocation(alias, getOriginalLocations());
+ clone.setStripKeyFromValue(stripKeyFromValue);
+
try {
- clone.setPlans(clonePlans);
+ clone.setPlans(clonePlans(plans));
+ if (secondaryPlans != null) {
+ clone.setSecondaryPlans(clonePlans(secondaryPlans));
+ }
} catch (PlanException pe) {
CloneNotSupportedException cnse = new CloneNotSupportedException("Problem with setting plans of " + this.getClass().getSimpleName());
cnse.initCause(pe);
throw cnse;
}
- clone.keyType = keyType;
- clone.mainKeyType = mainKeyType;
- clone.secondaryKeyType = secondaryKeyType;
- clone.useSecondaryKey = useSecondaryKey;
- clone.index = index;
- // Needs to be called as setDistinct so that the fake index tuple gets
- // created.
- clone.setDistinct(mIsDistinct);
- clone.addOriginalLocation(alias, getOriginalLocations());
- return clone;
+ }
+
+ private List<PhysicalPlan> clonePlans(List<PhysicalPlan> origPlans) throws CloneNotSupportedException {
+ List<PhysicalPlan> clonePlans = new ArrayList<PhysicalPlan>(origPlans.size());
+ for (PhysicalPlan plan : origPlans) {
+ clonePlans.add(plan.clone());
+ }
+ return clonePlans;
}
public boolean isCross() {
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POMergeJoin.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POMergeJoin.java?rev=1598702&r1=1598701&r2=1598702&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POMergeJoin.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POMergeJoin.java Fri May 30 19:07:23 2014
@@ -123,6 +123,8 @@ public class POMergeJoin extends Physica
private String signature;
+ private byte endOfRecordMark;
+
// This serves as the default TupleFactory
private transient TupleFactory mTupleFactory;
@@ -154,14 +156,23 @@ public class POMergeJoin extends Physica
LRs = new POLocalRearrange[2];
this.createJoinPlans(inpPlans,keyTypes);
this.indexFile = null;
- this.joinType = joinType;
+ this.joinType = joinType;
this.leftInputSchema = leftInputSchema;
this.mergedInputSchema = mergedInputSchema;
+ this.endOfRecordMark = POStatus.STATUS_EOP;
+ }
+
+ // Set to POStatus.STATUS_EOP (default) for MR and POStatus.STATUS_NULL for Tez.
+ // This is because:
+ // For MR, we send EOP at the end of every record
+ // For Tez, we only use a global EOP, so send NULL for end of record
+ public void setEndOfRecordMark(byte endOfRecordMark) {
+ this.endOfRecordMark = endOfRecordMark;
}
/**
* Configures the Local Rearrange operators to get keys out of tuple.
- * @throws ExecException
+ * @throws ExecException
*/
private void createJoinPlans(MultiMap<PhysicalOperator, PhysicalPlan> inpPlans, List<List<Byte>> keyTypes) throws PlanException{
@@ -223,11 +234,11 @@ public class POMergeJoin extends Physica
* from Tuple to SchemaTuple. This is necessary because we are not getting SchemaTuples
* from the source, though in the future that is what we would like to do.
*/
- protected static class TuplesToSchemaTupleList {
+ public static class TuplesToSchemaTupleList {
private List<Tuple> tuples;
private SchemaTupleFactory tf;
- protected TuplesToSchemaTupleList(int ct, TupleMaker<?> tf) {
+ public TuplesToSchemaTupleList(int ct, TupleMaker<?> tf) {
tuples = new ArrayList<Tuple>(ct);
if (tf instanceof SchemaTupleFactory) {
this.tf = (SchemaTupleFactory)tf;
@@ -285,8 +296,8 @@ public class POMergeJoin extends Physica
curLeftKey = extractKeysFromTuple(curLeftInp, 0);
if(null == curLeftKey) // We drop the tuples which have null keys.
- return new Result(POStatus.STATUS_EOP, null);
-
+ return new Result(endOfRecordMark, null);
+
try {
seekInRightStream(curLeftKey);
} catch (IOException e) {
@@ -297,7 +308,7 @@ public class POMergeJoin extends Physica
leftTuples.add((Tuple)curLeftInp.result);
firstTime = false;
prevLeftKey = curLeftKey;
- return new Result(POStatus.STATUS_EOP, null);
+ return new Result(endOfRecordMark, null);
}
if(doingJoin){
@@ -352,6 +363,7 @@ public class POMergeJoin extends Physica
leftTuples = newLeftTupleArray();
leftTuples.add((Tuple)prevLeftInp.result);
+ return new Result(endOfRecordMark, null);
}
else{ // This is end of all input and this is last join output.
@@ -362,8 +374,8 @@ public class POMergeJoin extends Physica
// Non-fatal error. We can continue.
log.error("Received exception while trying to close right side file: " + e.getMessage());
}
+ return new Result(POStatus.STATUS_EOP, null);
}
- return new Result(POStatus.STATUS_EOP, null);
}
else{ // At this point right side can't be behind.
int errCode = 1102;
@@ -381,17 +393,17 @@ public class POMergeJoin extends Physica
case POStatus.STATUS_OK:
curLeftKey = extractKeysFromTuple(curLeftInp, 0);
if(null == curLeftKey) // We drop the tuples which have null keys.
- return new Result(POStatus.STATUS_EOP, null);
-
+ return new Result(endOfRecordMark, null);
+
int cmpVal = ((Comparable)curLeftKey).compareTo(prevLeftKey);
if(cmpVal == 0){
// Keep on accumulating.
leftTuples.add((Tuple)curLeftInp.result);
- return new Result(POStatus.STATUS_EOP, null);
+ return new Result(endOfRecordMark, null);
}
else if(cmpVal > 0){ // Filled with left bag. Move on.
curJoinKey = prevLeftKey;
- break;
+ break;
}
else{ // Current key < Prev Key
int errCode = 1102;
@@ -423,7 +435,7 @@ public class POMergeJoin extends Physica
leftTuples.add((Tuple)curLeftInp.result);
prevLeftInp = curLeftInp;
prevLeftKey = curLeftKey;
- return new Result(POStatus.STATUS_EOP, null);
+ return new Result(endOfRecordMark, null);
}
// Accumulated tuples with same key on left side.
@@ -504,14 +516,14 @@ public class POMergeJoin extends Physica
log.error("Received exception while trying to close right side file: " + e.getMessage());
}
}
- return new Result(POStatus.STATUS_EOP, null);
+ return new Result(endOfRecordMark, null);
}
}
}
-
+
private void seekInRightStream(Object firstLeftKey) throws IOException{
rightLoader = (LoadFunc)PigContext.instantiateFuncFromSpec(rightLoaderFuncSpec);
-
+
// check if hadoop distributed cache is used
if (indexFile != null && rightLoader instanceof DefaultIndexableLoader) {
DefaultIndexableLoader loader = (DefaultIndexableLoader)rightLoader;
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPackage.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPackage.java?rev=1598702&r1=1598701&r2=1598702&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPackage.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPackage.java Fri May 30 19:07:23 2014
@@ -19,10 +19,8 @@ package org.apache.pig.backend.hadoop.ex
import java.io.IOException;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
-import java.util.Map;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
@@ -42,7 +40,6 @@ import org.apache.pig.impl.io.PigNullabl
import org.apache.pig.impl.plan.NodeIdGenerator;
import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.impl.plan.VisitorException;
-import org.apache.pig.impl.util.Pair;
import org.apache.pig.pen.Illustrator;
/**
@@ -59,9 +56,7 @@ import org.apache.pig.pen.Illustrator;
* bags based on the index.
*/
public class POPackage extends PhysicalOperator {
- /**
- *
- */
+
private static final long serialVersionUID = 1L;
//The iterator of indexed Tuples
@@ -86,7 +81,7 @@ public class POPackage extends PhysicalO
protected Packager pkgr;
- private PigNullableWritable keyWritable;
+ protected PigNullableWritable keyWritable;
public POPackage(OperatorKey k) {
this(k, -1, null);
@@ -111,6 +106,13 @@ public class POPackage extends PhysicalO
this.pkgr = pkgr;
}
+ public POPackage(POPackage copy) {
+ super(copy);
+ this.numInputs = copy.numInputs;
+ this.pkgr = copy.pkgr;
+ this.pkgr.keyInfo = copy.pkgr.keyInfo;
+ }
+
@Override
public void setIllustrator(Illustrator illustrator) {
super.setIllustrator(illustrator);
@@ -211,6 +213,8 @@ public class POPackage extends PhysicalO
// all bags have reference to the sample tuples buffer
// which contains tuples from one batch
POPackageTupleBuffer buffer = new POPackageTupleBuffer();
+ buffer.setKey(key);
+ buffer.setIterator(tupIter);
for (int i = 0; i < numInputs; i++) {
dbs[i] = new AccumulativeBag(buffer, i);
}
@@ -240,8 +244,7 @@ public class POPackage extends PhysicalO
NullableTuple ntup = tupIter.next();
int index = ntup.getIndex();
if (index == numInputs - 1) {
- dbs[index] = new PeekedBag(pkgr, ntup, tupIter,
- keyWritable);
+ dbs[index] = new PeekedBag(pkgr, ntup, tupIter, keyWritable);
break;
}
Tuple copy = pkgr.getValueTuple(keyWritable, ntup, index);
@@ -296,7 +299,7 @@ public class POPackage extends PhysicalO
return clone;
}
- class POPackageTupleBuffer implements AccumulativeTupleBuffer {
+ public class POPackageTupleBuffer implements AccumulativeTupleBuffer {
private List<Tuple>[] bags;
private Iterator<NullableTuple> iter;
private int batchSize;
@@ -316,10 +319,16 @@ public class POPackage extends PhysicalO
for(int i=0; i<numInputs; i++) {
this.bags[i] = new ArrayList<Tuple>();
}
- this.iter = tupIter;
+ }
+
+ public void setKey(Object key) {
this.currKey = key;
}
+ public void setIterator(Iterator<NullableTuple> iter) {
+ this.iter = iter;
+ }
+
@Override
public boolean hasNextBatch() {
return iter.hasNext();
@@ -382,9 +391,6 @@ public class POPackage extends PhysicalO
// A ReadOnceBag that we've already "peeked" at
private static class PeekedBag extends ReadOnceBag {
- /**
- *
- */
private static final long serialVersionUID = 1L;
NullableTuple head;
int index;
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPartitionRearrange.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPartitionRearrange.java?rev=1598702&r1=1598701&r2=1598702&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPartitionRearrange.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPartitionRearrange.java Fri May 30 19:07:23 2014
@@ -19,7 +19,6 @@ package org.apache.pig.backend.hadoop.ex
import java.io.IOException;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -40,6 +39,7 @@ import org.apache.pig.impl.plan.Operator
import org.apache.pig.impl.util.Pair;
import org.apache.pig.impl.util.Utils;
+import com.google.common.collect.Maps;
/**
* The partition rearrange operator is a part of the skewed join
@@ -49,17 +49,14 @@ import org.apache.pig.impl.util.Utils;
*/
public class POPartitionRearrange extends POLocalRearrange {
- /**
- *
- */
private static final long serialVersionUID = 1L;
+ private static final BagFactory mBagFactory = BagFactory.getInstance();
private Integer totalReducers = -1;
// ReducerMap will store the tuple, max reducer index & min reducer index
- private static Map<Object, Pair<Integer, Integer> > reducerMap = new HashMap<Object, Pair<Integer, Integer> >();
- private boolean loaded;
+ private Map<Object, Pair<Integer, Integer> > reducerMap = Maps.newHashMap();
+ private boolean inited;
- protected static final BagFactory mBagFactory = BagFactory.getInstance();
private PigContext pigContext;
public POPartitionRearrange(OperatorKey k) {
@@ -81,7 +78,7 @@ public class POPartitionRearrange extend
}
/* Loads the key distribution file obtained from the sampler */
- private void loadPartitionFile() throws RuntimeException {
+ private void init() throws RuntimeException {
String keyDistFile = PigMapReduce.sJobConfInternal.get().get("pig.keyDistFile", "");
if (keyDistFile.isEmpty()) {
throw new RuntimeException(
@@ -102,7 +99,7 @@ public class POPartitionRearrange extend
// check if the partition file is empty
totalReducers = (redCnt[0] == null) ? -1 : redCnt[0];
- loaded = true;
+ inited = true;
} catch (Exception e) {
throw new RuntimeException(e);
}
@@ -128,12 +125,12 @@ public class POPartitionRearrange extend
Result res = null;
// Load the skewed join key partitioning file
- if (!loaded) {
- try {
- loadPartitionFile();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
+ if (!inited) {
+ try {
+ init();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
}
while (true) {
@@ -163,40 +160,40 @@ public class POPartitionRearrange extend
return inp;
}
- // Returns bag of tuples
+ // Returns bag of tuples
protected DataBag constructPROutput(List<Result> resLst, Tuple value) throws ExecException{
- Tuple t = super.constructLROutput(resLst, null, value);
+ Tuple t = super.constructLROutput(resLst, null, value);
//Construct key
Object key = t.get(1);
- // Construct an output bag and feed in the tuples
- DataBag opBag = mBagFactory.newDefaultBag();
+ // Construct an output bag and feed in the tuples
+ DataBag opBag = mBagFactory.newDefaultBag();
- //Put the index, key, and value
- //in a tuple and return
- Pair <Integer, Integer> indexes = reducerMap.get(key); // first -> min, second ->max
-
- // For non skewed keys, we set the partition index to be -1
- if (indexes == null) {
- indexes = new Pair <Integer, Integer>(-1,0);
- }
-
- for (Integer reducerIdx=indexes.first, cnt=0; cnt <= indexes.second; reducerIdx++, cnt++) {
- if (reducerIdx >= totalReducers) {
- reducerIdx = 0;
- }
- Tuple opTuple = mTupleFactory.newTuple(4);
- opTuple.set(0, t.get(0));
- // set the partition index
- opTuple.set(1, reducerIdx.intValue());
- opTuple.set(2, key);
- opTuple.set(3, t.get(2));
+ //Put the index, key, and value
+ //in a tuple and return
+ Pair <Integer, Integer> indexes = reducerMap.get(key); // first -> min, second ->max
+
+ // For non skewed keys, we set the partition index to be -1
+ if (indexes == null) {
+ indexes = new Pair <Integer, Integer>(-1,0);
+ }
- opBag.add(opTuple);
- }
+ for (Integer reducerIdx=indexes.first, cnt=0; cnt <= indexes.second; reducerIdx++, cnt++) {
+ if (reducerIdx >= totalReducers) {
+ reducerIdx = 0;
+ }
+ Tuple opTuple = mTupleFactory.newTuple(4);
+ opTuple.set(0, t.get(0));
+ // set the partition index
+ opTuple.set(1, reducerIdx.intValue());
+ opTuple.set(2, key);
+ opTuple.set(3, t.get(2));
+
+ opBag.add(opTuple);
+ }
- return opBag;
+ return opBag;
}
/**
@@ -219,8 +216,7 @@ public class POPartitionRearrange extend
*/
@Override
public POPartitionRearrange clone() throws CloneNotSupportedException {
- POPartitionRearrange clone = (POPartitionRearrange) super.clone();
- //clone.reducerMap = (HashMap)reducerMap.clone();
+ POPartitionRearrange clone = (POPartitionRearrange) super.clone();
return clone;
}
Added: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPoissonSample.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPoissonSample.java?rev=1598702&view=auto
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPoissonSample.java (added)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPoissonSample.java Fri May 30 19:07:23 2014
@@ -0,0 +1,234 @@
+/*
+ * 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.physicalLayer.relationalOperators;
+
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
+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.PhyPlanVisitor;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.VisitorException;
+
+public class POPoissonSample extends PhysicalOperator {
+
+ private static final long serialVersionUID = 1L;
+
+ // marker string to mark the last sample row, which has total number or rows
+ // seen by this map instance. this string will be in the 2nd last column of
+ // the last sample row it is used by GetMemNumRows.
+ public static final String NUMROWS_TUPLE_MARKER =
+ "\u4956\u3838_pig_inTeRnal-spEcial_roW_num_tuple3kt579CFLehkblah";
+
+ private static final TupleFactory tf = TupleFactory.getInstance();
+ private static Result eop = new Result(POStatus.STATUS_EOP, null);
+
+ // num of rows sampled so far
+ private int numRowsSampled = 0;
+
+ // average size of tuple in memory, for tuples sampled
+ private long avgTupleMemSz = 0;
+
+ // current row number
+ private long rowNum = 0;
+
+ // number of tuples to skip after each sample
+ private long skipInterval = -1;
+
+ // bytes in input to skip after every sample.
+ // divide this by avgTupleMemSize to get skipInterval
+ private long memToSkipPerSample = 0;
+
+ // has the special row with row number information been returned
+ private boolean numRowSplTupleReturned = false;
+
+ // 17 is not a magic number. It can be obtained by using a poisson
+ // cumulative distribution function with the mean set to 10 (empirically,
+ // minimum number of samples) and the confidence set to 95%
+ public static final int DEFAULT_SAMPLE_RATE = 17;
+
+ private int sampleRate = 0;
+
+ private float heapPerc = 0f;
+
+ // new Sample result
+ private Result newSample = null;
+
+ public POPoissonSample(OperatorKey k, int rp, int sr, float hp) {
+ super(k, rp, null);
+ numRowsSampled = 0;
+ avgTupleMemSz = 0;
+ rowNum = 0;
+ skipInterval = -1;
+ memToSkipPerSample = 0;
+ numRowSplTupleReturned = false;
+ newSample = null;
+ sampleRate = sr;
+ heapPerc = hp;
+ }
+
+ @Override
+ public Tuple illustratorMarkup(Object in, Object out, int eqClassIndex) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public void visit(PhyPlanVisitor v) throws VisitorException {
+ v.visitPoissonSample(this);
+ }
+
+ @Override
+ public Result getNextTuple() throws ExecException {
+ if (numRowSplTupleReturned) {
+ // row num special row has been returned after all inputs
+ // were read, nothing more to read
+ return eop;
+ }
+
+ Result res = null;
+ if (skipInterval == -1) {
+ // select first tuple as sample and calculate
+ // number of tuples to be skipped
+ while (true) {
+ res = processInput();
+ if (res.returnStatus == POStatus.STATUS_NULL) {
+ continue;
+ } else if (res.returnStatus == POStatus.STATUS_EOP) {
+ if (this.parentPlan.endOfAllInput) {
+ return eop;
+ } else {
+ continue;
+ }
+ } else if (res.returnStatus == POStatus.STATUS_ERR) {
+ return res;
+ }
+
+ if (res.result == null) {
+ continue;
+ }
+ long availRedMem = (long) (Runtime.getRuntime().maxMemory() * heapPerc);
+ memToSkipPerSample = availRedMem/sampleRate;
+ updateSkipInterval((Tuple)res.result);
+
+ rowNum++;
+ newSample = res;
+ break;
+ }
+ }
+
+ // skip tuples
+ for (long numSkipped = 0; numSkipped < skipInterval; numSkipped++) {
+ res = processInput();
+ if (res.returnStatus == POStatus.STATUS_NULL) {
+ continue;
+ } else if (res.returnStatus == POStatus.STATUS_EOP
+ || res.returnStatus == POStatus.STATUS_ERR) {
+ if (this.parentPlan.endOfAllInput) {
+ return createNumRowTuple((Tuple)newSample.result);
+ } else {
+ continue;
+ }
+ }
+ rowNum++;
+ }
+
+ // skipped enough, get new sample
+ while (true) {
+ res = processInput();
+ if (res.returnStatus == POStatus.STATUS_NULL) {
+ continue;
+ } else if (res.returnStatus == POStatus.STATUS_EOP
+ || res.returnStatus == POStatus.STATUS_ERR) {
+ if (this.parentPlan.endOfAllInput) {
+ return createNumRowTuple((Tuple)newSample.result);
+ } else {
+ continue;
+ }
+ }
+
+ if (res.result == null) {
+ continue;
+ }
+ updateSkipInterval((Tuple)res.result);
+ Result currentSample = newSample;
+
+ rowNum++;
+ newSample = res;
+ return currentSample;
+ }
+ }
+
+ @Override
+ public boolean supportsMultipleInputs() {
+ return false;
+ }
+
+ @Override
+ public boolean supportsMultipleOutputs() {
+ return false;
+ }
+
+ @Override
+ public String name() {
+ return getAliasString() + "PoissonSample - " + mKey.toString();
+ }
+
+ /**
+ * Update the average tuple size base on newly sampled tuple t
+ * and recalculate skipInterval
+ * @param t - tuple
+ */
+ private void updateSkipInterval(Tuple t) {
+ avgTupleMemSz =
+ ((avgTupleMemSz*numRowsSampled) + t.getMemorySize())/(numRowsSampled + 1);
+ skipInterval = memToSkipPerSample/avgTupleMemSz;
+
+ // skipping fewer number of rows the first few times, to reduce the
+ // probability of first tuples size (if much smaller than rest)
+ // resulting in very few samples being sampled. Sampling a little extra
+ // is OK
+ if(numRowsSampled < 5) {
+ skipInterval = skipInterval/(10-numRowsSampled);
+ }
+ ++numRowsSampled;
+ }
+
+ /**
+ * @param sample - sample tuple
+ * @return - Tuple appended with special marker string column, num-rows column
+ * @throws ExecException
+ */
+ private Result createNumRowTuple(Tuple sample) throws ExecException {
+ int sz = (sample == null) ? 0 : sample.size();
+ Tuple t = tf.newTuple(sz + 2);
+
+ if (sample != null) {
+ for (int i=0; i<sample.size(); i++){
+ t.set(i, sample.get(i));
+ }
+ }
+
+ t.set(sz, NUMROWS_TUPLE_MARKER);
+ t.set(sz + 1, rowNum);
+ numRowSplTupleReturned = true;
+ return new Result(POStatus.STATUS_OK, t);
+ }
+}
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PORank.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PORank.java?rev=1598702&r1=1598701&r2=1598702&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PORank.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PORank.java Fri May 30 19:07:23 2014
@@ -85,6 +85,13 @@ public class PORank extends PhysicalOper
super(k, rp, inp);
}
+ public PORank(PORank copy) {
+ super(copy);
+ this.rankPlans = copy.rankPlans;
+ this.mAscCols = copy.mAscCols;
+ this.ExprOutputTypes = copy.ExprOutputTypes;
+ }
+
@SuppressWarnings({ "rawtypes", "unchecked" })
public PORank(OperatorKey operatorKey, int requestedParallelism, List inp,
List<PhysicalPlan> rankPlans, List<Boolean> ascendingCol) {
@@ -142,20 +149,32 @@ public class PORank extends PhysicalOper
* Here is read the task identifier in order to get the corresponding cumulative sum,
* and the local counter at the tuple. These values are summed and prepended to the tuple.
* @param input processed by POCounter
- * @return input as Result. The input.result tuple owns the prepend rank value
+ * @return input as Result. The input.result tuple owns the prepend rank value
**/
public Result addRank(Result input) throws ExecException {
- int i = 1;
Tuple in = (Tuple) input.result;
- Tuple out = mTupleFactory.newTuple(in.getAll().size() - 1);
- Long taskId = Long.valueOf(in.get(0).toString());
- Long localCounter = (Long) in.get(1);
+ Long localCounter = (Long) in.get(0);
+ Integer taskId = (Integer) in.getAll().remove(in.getAll().size() - 1);
- String nameCounter = JobControlCompiler.PIG_MAP_COUNTER + getOperationID() + JobControlCompiler.PIG_MAP_SEPARATOR + String.valueOf(taskId);
+ Long rank = getRankCounterOffset(taskId);
+
+ in.set(0, rank + localCounter);
+
+ if(localCountIllustrator > 2)
+ localCountIllustrator = 0;
+
+ input.result = illustratorMarkup(in, in, localCountIllustrator);
+
+ localCountIllustrator++;
+
+ return input;
+ }
+ protected Long getRankCounterOffset(Integer taskId) {
+ String nameCounter = JobControlCompiler.PIG_MAP_COUNTER + getOperationID() + JobControlCompiler.PIG_MAP_SEPARATOR + String.valueOf(taskId);
Long rank = PigMapReduce.sJobConfInternal.get().getLong( nameCounter , -1L );
-
+
if(illustrator != null) {
rank = 0L;
}
@@ -164,23 +183,7 @@ public class PORank extends PhysicalOper
log.error("Error on reading counter "+ nameCounter);
throw new RuntimeException("Unable to read counter "+ nameCounter);
}
-
- out.set(0, rank + localCounter);
-
- //Add the content of the tuple
- List<Object> sub = in.getAll().subList(2, in.getAll().size());
-
- for (Object o : sub)
- out.set(i++, o);
-
- if(localCountIllustrator > 2)
- localCountIllustrator = 0;
-
- input.result = illustratorMarkup(in, out, localCountIllustrator);
-
- localCountIllustrator++;
-
- return input;
+ return rank;
}
@Override
Added: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POReservoirSample.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POReservoirSample.java?rev=1598702&view=auto
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POReservoirSample.java (added)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POReservoirSample.java Fri May 30 19:07:23 2014
@@ -0,0 +1,162 @@
+/*
+ * 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.physicalLayer.relationalOperators;
+
+import java.util.List;
+import java.util.Random;
+
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
+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.PhyPlanVisitor;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.VisitorException;
+
+public class POReservoirSample extends PhysicalOperator {
+
+ private static final long serialVersionUID = 1L;
+
+ // number of samples to be sampled
+ protected int numSamples;
+
+ private transient int nextSampleIdx= 0;
+
+ private int rowProcessed = 0;
+
+ private boolean sampleCollectionDone = false;
+
+ //array to store the result
+ private transient Result[] samples = null;
+
+ public POReservoirSample(OperatorKey k) {
+ this(k, -1, null);
+ }
+
+ public POReservoirSample(OperatorKey k, int rp) {
+ this(k, rp, null);
+ }
+
+ public POReservoirSample(OperatorKey k, List<PhysicalOperator> inp) {
+ this(k, -1, inp);
+ }
+
+ public POReservoirSample(OperatorKey k, int rp, List<PhysicalOperator> inp) {
+ super(k, rp, inp);
+ }
+
+ public POReservoirSample(OperatorKey k, int rp, List<PhysicalOperator> inp, int numSamples) {
+ super(k, rp, inp);
+ this.numSamples = numSamples;
+ }
+
+ @Override
+ public Tuple illustratorMarkup(Object in, Object out, int eqClassIndex) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public void visit(PhyPlanVisitor v) throws VisitorException {
+ v.visitReservoirSample(this);
+ }
+
+ @Override
+ public Result getNextTuple() throws ExecException {
+ if (sampleCollectionDone){
+ return getSample();
+ }
+ //else collect samples
+ if (samples == null) {
+ samples = new Result[numSamples];
+ }
+
+ // populate the samples array with first numSamples tuples
+ Result res = null;
+ while (rowProcessed < numSamples) {
+ res = processInput();
+ if (res.returnStatus == POStatus.STATUS_OK) {
+ samples[rowProcessed] = res;
+ rowProcessed++;
+ } else if (res.returnStatus == POStatus.STATUS_NULL) {
+ continue;
+ } else {
+ break;
+ }
+ }
+
+ int rowNum = rowProcessed;
+ Random randGen = new Random();
+
+ while (true) {
+ // pick this as sample
+ res = processInput();
+ if (res.returnStatus == POStatus.STATUS_NULL) {
+ continue;
+ } else if (res.returnStatus != POStatus.STATUS_OK) {
+ break;
+ }
+
+ // collect samples until input is exhausted
+ int rand = randGen.nextInt(rowNum);
+ if (rand < numSamples) {
+ samples[rand] = res;
+ }
+ rowNum++;
+ }
+
+ if (this.parentPlan.endOfAllInput && res.returnStatus == POStatus.STATUS_EOP) {
+ sampleCollectionDone = true;
+ }
+
+ return getSample();
+ }
+
+ private Result getSample() throws ExecException {
+ if(nextSampleIdx < samples.length){
+ if (illustrator != null) {
+ illustratorMarkup(samples[nextSampleIdx].result, samples[nextSampleIdx].result, 0);
+ }
+ Result res = samples[nextSampleIdx++];
+ if (res == null) { // Input data has lesser rows than numSamples
+ return new Result(POStatus.STATUS_NULL, null);
+ }
+ return res;
+ }
+ else{
+ Result res = new Result(POStatus.STATUS_EOP, null);
+ return res;
+ }
+ }
+
+ @Override
+ public boolean supportsMultipleInputs() {
+ return false;
+ }
+
+ @Override
+ public boolean supportsMultipleOutputs() {
+ return false;
+ }
+
+ @Override
+ public String name() {
+ return getAliasString() + "ReservoirSample - " + mKey.toString();
+ }
+}
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSkewedJoin.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSkewedJoin.java?rev=1598702&r1=1598701&r2=1598702&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSkewedJoin.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSkewedJoin.java Fri May 30 19:07:23 2014
@@ -34,23 +34,23 @@ import org.apache.pig.impl.util.MultiMap
/**
* The PhysicalOperator that represents a skewed join. It must have two inputs.
* This operator does not do any actually work, it is only a place holder. When it is
- * translated into MR plan, a POSkewedJoin is translated into a sampling job and a join
+ * translated into MR plan, a POSkewedJoin is translated into a sampling job and a join
* job.
- *
+ *
*
*/
public class POSkewedJoin extends PhysicalOperator {
-
- private static final long serialVersionUID = 1L;
- private boolean[] mInnerFlags;
-
- // The schema is used only by the MRCompiler to support outer join
- transient private List<Schema> inputSchema = new ArrayList<Schema>();
-
- // physical plans to retrive join keys
- // the key of this <code>MultiMap</code> is the PhysicalOperator that corresponds to an input
- // the value is a list of <code>PhysicalPlan</code> to retrieve each join key for this input
+
+ private static final long serialVersionUID = 1L;
+ private boolean[] mInnerFlags;
+
+ // The schema is used only by the MRCompiler to support outer join
+ transient private List<Schema> inputSchema = new ArrayList<Schema>();
+
+ // physical plans to retrive join keys
+ // the key of this <code>MultiMap</code> is the PhysicalOperator that corresponds to an input
+ // the value is a list of <code>PhysicalPlan</code> to retrieve each join key for this input
private MultiMap<PhysicalOperator, PhysicalPlan> mJoinPlans;
public POSkewedJoin(OperatorKey k) {
@@ -68,30 +68,30 @@ public class POSkewedJoin extends Physic
public POSkewedJoin(OperatorKey k, int rp, List<PhysicalOperator> inp, boolean []flags) {
super(k,rp,inp);
if (flags != null) {
- // copy the inner flags
- mInnerFlags = new boolean[flags.length];
- for (int i = 0; i < flags.length; i++) {
- mInnerFlags[i] = flags[i];
- }
+ // copy the inner flags
+ mInnerFlags = new boolean[flags.length];
+ for (int i = 0; i < flags.length; i++) {
+ mInnerFlags[i] = flags[i];
+ }
}
}
-
+
public boolean[] getInnerFlags() {
- return mInnerFlags;
+ return mInnerFlags;
}
-
+
public MultiMap<PhysicalOperator, PhysicalPlan> getJoinPlans() {
- return mJoinPlans;
+ return mJoinPlans;
}
-
+
public void setJoinPlans(MultiMap<PhysicalOperator, PhysicalPlan> joinPlans) {
mJoinPlans = joinPlans;
- }
-
- @Override
- public void visit(PhyPlanVisitor v) throws VisitorException {
- v.visitSkewedJoin(this);
- }
+ }
+
+ @Override
+ public void visit(PhyPlanVisitor v) throws VisitorException {
+ v.visitSkewedJoin(this);
+ }
@Override
public String name() {
@@ -100,26 +100,26 @@ public class POSkewedJoin extends Physic
+ mKey.toString();
}
- @Override
- public boolean supportsMultipleInputs() {
- return true;
- }
-
- @Override
- public boolean supportsMultipleOutputs() {
- return false;
- }
-
- public void addSchema(Schema s) {
- inputSchema.add(s);
- }
-
- public Schema getSchema(int i) {
- return inputSchema.get(i);
- }
-
- @Override
- public Tuple illustratorMarkup(Object in, Object out, int eqClassIndex) {
- return null;
- }
+ @Override
+ public boolean supportsMultipleInputs() {
+ return true;
+ }
+
+ @Override
+ public boolean supportsMultipleOutputs() {
+ return false;
+ }
+
+ public void addSchema(Schema s) {
+ inputSchema.add(s);
+ }
+
+ public Schema getSchema(int i) {
+ return inputSchema.get(i);
+ }
+
+ @Override
+ public Tuple illustratorMarkup(Object in, Object out, int eqClassIndex) {
+ return null;
+ }
}
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStore.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStore.java?rev=1598702&r1=1598701&r2=1598702&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStore.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStore.java Fri May 30 19:07:23 2014
@@ -50,7 +50,7 @@ import org.apache.pig.tools.pigstats.Pig
public class POStore extends PhysicalOperator {
private static final long serialVersionUID = 1L;
- private static Result empty = new Result(POStatus.STATUS_NULL, null);
+ protected static Result empty = new Result(POStatus.STATUS_NULL, null);
transient private StoreFuncInterface storer;
transient private POStoreImpl impl;
transient private String counterName = null;
@@ -93,6 +93,19 @@ public class POStore extends PhysicalOpe
super(k, rp, inp);
}
+ public POStore(POStore copy) {
+ super(copy);
+ this.sFile = copy.sFile;
+ this.schema = copy.schema;
+ this.isTmpStore = copy.isTmpStore;
+ this.isMultiStore = copy.isMultiStore;
+ this.disableCounter = copy.disableCounter;
+ this.index = copy.index;
+ this.lFile = copy.lFile;
+ this.sortInfo = copy.sortInfo;
+ this.signature = copy.signature;
+ }
+
/**
* Set up the storer
* @throws IOException
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStream.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStream.java?rev=1598702&r1=1598701&r2=1598702&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStream.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStream.java Fri May 30 19:07:23 2014
@@ -175,8 +175,8 @@ public class POStream extends PhysicalOp
// if we are here, we haven't consumed all input to be sent
// to the streaming binary - check if we are being called
// from close() on the map or reduce
+ Result r = getNextHelper((Tuple)null);
if(isFetchable || this.parentPlan.endOfAllInput) {
- Result r = getNextHelper((Tuple)null);
if(r.returnStatus == POStatus.STATUS_EOP) {
// we have now seen *ALL* possible input
// check if we ever had any real input
@@ -224,7 +224,6 @@ public class POStream extends PhysicalOp
// we are not being called from close() - so
// we must be called from either map() or reduce()
// get the next Result from helper
- Result r = getNextHelper((Tuple)null);
if(r.returnStatus == POStatus.STATUS_EOS) {
// If we received EOS, it means all output
// from the streaming binary has been sent to us
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/Packager.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/Packager.java?rev=1598702&r1=1598701&r2=1598702&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/Packager.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/Packager.java Fri May 30 19:07:23 2014
@@ -26,9 +26,6 @@ import org.apache.pig.pen.util.LineageTr
public class Packager implements Illustrable, Serializable, Cloneable {
- /**
- *
- */
private static final long serialVersionUID = 1L;
protected boolean[] readOnce;
@@ -84,8 +81,7 @@ public class Packager implements Illustr
protected POPackage parent = null;
protected static final BagFactory mBagFactory = BagFactory.getInstance();
- protected static final TupleFactory mTupleFactory = TupleFactory
- .getInstance();
+ protected static final TupleFactory mTupleFactory = TupleFactory.getInstance();
public Object getKey(PigNullableWritable key) throws ExecException {
Object keyObject = key.getValueAsPigType();
@@ -210,7 +206,7 @@ public class Packager implements Illustr
int finalValueSize = keyLookupSize + val.size();
copy = mTupleFactory.newTuple(finalValueSize);
int valIndex = 0; // an index for accessing elements from
- // the value (val) that we have currently
+ // the value (val) that we have currently
for (int i = 0; i < finalValueSize; i++) {
Integer keyIndex = keyLookup.get(i);
if (keyIndex == null) {
@@ -285,8 +281,7 @@ public class Packager implements Illustr
}
/**
- * @param keyInfo
- * the keyInfo to set
+ * @param keyInfo the keyInfo to set
*/
public void setKeyInfo(
Map<Integer, Pair<Boolean, Map<Integer, Integer>>> keyInfo) {
@@ -294,16 +289,14 @@ public class Packager implements Illustr
}
/**
- * @param keyTuple
- * the keyTuple to set
+ * @param keyTuple the keyTuple to set
*/
public void setKeyTuple(boolean keyTuple) {
this.isKeyTuple = keyTuple;
}
/**
- * @param keyCompound
- * the keyCompound to set
+ * @param keyCompound the keyCompound to set
*/
public void setKeyCompound(boolean keyCompound) {
this.isKeyCompound = keyCompound;
@@ -333,8 +326,7 @@ public class Packager implements Illustr
}
/**
- * @param distinct
- * the distinct to set
+ * @param distinct the distinct to set
*/
public void setDistinct(boolean distinct) {
this.distinct = distinct;
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/util/PlanHelper.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/util/PlanHelper.java?rev=1598702&r1=1598701&r2=1598702&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/util/PlanHelper.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/util/PlanHelper.java Fri May 30 19:07:23 2014
@@ -19,6 +19,7 @@ package org.apache.pig.backend.hadoop.ex
import java.net.URI;
import java.util.LinkedList;
+import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -68,7 +69,9 @@ import org.apache.pig.backend.hadoop.exe
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.POPartitionRearrange;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPoissonSample;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPreCombinerLocalRearrange;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POReservoirSample;
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;
@@ -76,6 +79,7 @@ import org.apache.pig.backend.hadoop.exe
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.OperatorKey;
import org.apache.pig.impl.plan.VisitorException;
import com.google.common.collect.Lists;
@@ -121,14 +125,33 @@ public class PlanHelper {
* @return a LinkedList of operators contained within the plan which implement the supplied class; empty if no such ops exist.
* @throws VisitorException
*/
- public static <C extends PhysicalOperator> LinkedList<C> getPhysicalOperators(PhysicalPlan plan,
+ public static <C> LinkedList<C> getPhysicalOperators(PhysicalPlan plan,
Class<C> opClass) throws VisitorException {
OpFinder<C> finder = new OpFinder<C>(plan, opClass);
finder.visit();
return finder.getFoundOps();
}
- private static class OpFinder<C extends PhysicalOperator> extends PhyPlanVisitor {
+ /**
+ * Finds POLocalRearrange from POSplit sub-plan
+ * @param plan physical plan
+ * @param rearrangeKey operator key of the POLocalRearrange
+ * @return POLocalRearrange with the specified operator key which is in a sub-plan of POSplit
+ * @throws VisitorException
+ */
+ public static PhysicalPlan getLocalRearrangePlanFromSplit(PhysicalPlan plan, OperatorKey rearrangeKey) throws VisitorException {
+ List<POSplit> splits = PlanHelper.getPhysicalOperators(plan, POSplit.class);
+ for (POSplit split : splits) {
+ for (PhysicalPlan subPlan : split.getPlans()) {
+ if (subPlan.getOperator(rearrangeKey) != null) {
+ return subPlan;
+ }
+ }
+ }
+ return plan;
+ }
+
+ private static class OpFinder<C> extends PhyPlanVisitor {
final Class<C> opClass;
private LinkedList<C> foundOps = Lists.newLinkedList();
@@ -147,8 +170,9 @@ public class PlanHelper {
return !foundOps.isEmpty();
}
+ @Override
@SuppressWarnings("unchecked")
- private void visit(PhysicalOperator op) {
+ public void visit(PhysicalOperator op) {
if (opClass.isAssignableFrom(op.getClass())) {
foundOps.add((C) op);
}
@@ -425,6 +449,7 @@ public class PlanHelper {
visit(stream);
}
+ @Override
public void visitSkewedJoin(POSkewedJoin sk) throws VisitorException {
super.visitSkewedJoin(sk);
visit(sk);
@@ -446,16 +471,28 @@ public class PlanHelper {
@Override
public void visitPreCombinerLocalRearrange(
- POPreCombinerLocalRearrange preCombinerLocalRearrange) {
+ POPreCombinerLocalRearrange preCombinerLocalRearrange) throws VisitorException {
super.visitPreCombinerLocalRearrange(preCombinerLocalRearrange);
visit(preCombinerLocalRearrange);
}
@Override
- public void visitPartialAgg(POPartialAgg poPartialAgg) {
+ public void visitPartialAgg(POPartialAgg poPartialAgg) throws VisitorException {
super.visitPartialAgg(poPartialAgg);
visit(poPartialAgg);
}
+
+ @Override
+ public void visitReservoirSample(POReservoirSample reservoirSample) throws VisitorException {
+ super.visitReservoirSample(reservoirSample);
+ visit(reservoirSample);
+ }
+
+ @Override
+ public void visitPoissonSample(POPoissonSample poissonSample) throws VisitorException {
+ super.visitPoissonSample(poissonSample);
+ visit(poissonSample);
+ }
}
}
Added: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/AccumulatorOptimizer.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/AccumulatorOptimizer.java?rev=1598702&view=auto
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/AccumulatorOptimizer.java (added)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/AccumulatorOptimizer.java Fri May 30 19:07:23 2014
@@ -0,0 +1,38 @@
+/*
+ * 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.tez;
+
+import org.apache.pig.backend.hadoop.executionengine.util.AccumulatorOptimizerUtil;
+import org.apache.pig.impl.plan.DepthFirstWalker;
+import org.apache.pig.impl.plan.VisitorException;
+
+/**
+ * A visitor to optimize plans that determines if a vertex plan can run in
+ * accumulative mode.
+ */
+public class AccumulatorOptimizer extends TezOpPlanVisitor {
+
+ public AccumulatorOptimizer(TezOperPlan plan) {
+ super(plan, new DepthFirstWalker<TezOperator, TezOperPlan>(plan));
+ }
+
+ public void visitTezOp(TezOperator tezOp) throws VisitorException {
+ AccumulatorOptimizerUtil.addAccumulator(tezOp.plan);
+ }
+}
Added: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/CombinerOptimizer.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/CombinerOptimizer.java?rev=1598702&view=auto
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/CombinerOptimizer.java (added)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/CombinerOptimizer.java Fri May 30 19:07:23 2014
@@ -0,0 +1,96 @@
+/**
+ * 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.tez;
+
+import java.util.List;
+
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPackage;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.util.PlanHelper;
+import org.apache.pig.backend.hadoop.executionengine.util.CombinerOptimizerUtil;
+import org.apache.pig.impl.plan.CompilationMessageCollector;
+import org.apache.pig.impl.plan.DepthFirstWalker;
+import org.apache.pig.impl.plan.VisitorException;
+
+/**
+ * Optimize tez plans to use the combiner where possible.
+ */
+public class CombinerOptimizer extends TezOpPlanVisitor {
+ private CompilationMessageCollector messageCollector = null;
+ private boolean doMapAgg;
+
+ public CombinerOptimizer(TezOperPlan plan, boolean doMapAgg) {
+ this(plan, doMapAgg, new CompilationMessageCollector());
+ }
+
+ public CombinerOptimizer(TezOperPlan plan, boolean doMapAgg,
+ CompilationMessageCollector messageCollector) {
+ super(plan, new DepthFirstWalker<TezOperator, TezOperPlan>(plan));
+ this.messageCollector = messageCollector;
+ this.doMapAgg = doMapAgg;
+ }
+
+ public CompilationMessageCollector getMessageCollector() {
+ return messageCollector;
+ }
+
+ @Override
+ public void visitTezOp(TezOperator to) throws VisitorException {
+ List<POPackage> packages = PlanHelper.getPhysicalOperators(to.plan, POPackage.class);
+ if (packages.isEmpty()) {
+ return;
+ }
+
+ List<TezOperator> predecessors = mPlan.getPredecessors(to);
+ if (predecessors == null) {
+ return;
+ }
+ if (to.isCogroup()) {
+ return;
+ }
+
+ for (TezOperator from : predecessors) {
+ List<POLocalRearrangeTez> rearranges = PlanHelper.getPhysicalOperators(from.plan, POLocalRearrangeTez.class);
+ if (rearranges.isEmpty()) {
+ continue;
+ }
+
+ POLocalRearrangeTez connectingLR = null;
+ PhysicalPlan rearrangePlan = from.plan;
+ for (POLocalRearrangeTez lr : rearranges) {
+ if (lr.getOutputKey().equals(to.getOperatorKey().toString())) {
+ connectingLR = lr;
+ break;
+ }
+ }
+
+ if (from.plan.getOperator(connectingLR.getOperatorKey()) == null) {
+ // The POLocalRearrange is sub-plan of a POSplit
+ rearrangePlan = PlanHelper.getLocalRearrangePlanFromSplit(from.plan, connectingLR.getOperatorKey());
+ }
+
+ // Detected the POLocalRearrange -> POPackage pattern. Let's add
+ // combiner if possible.
+ PhysicalPlan combinePlan = to.inEdges.get(from.getOperatorKey()).combinePlan;
+ CombinerOptimizerUtil.addCombiner(rearrangePlan, to.plan, combinePlan, messageCollector, doMapAgg);
+
+ }
+ }
+
+}
+