You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ju...@apache.org on 2012/05/12 16:58:37 UTC
svn commit: r1337558 - in /pig/trunk: ./
src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/
src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/
src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperat...
Author: julien
Date: Sat May 12 14:58:36 2012
New Revision: 1337558
URL: http://svn.apache.org/viewvc?rev=1337558&view=rev
Log:
PIG-2659: add source location of the aliases in the physical plan
Added:
pig/trunk/test/org/apache/pig/newplan/
pig/trunk/test/org/apache/pig/newplan/logical/
pig/trunk/test/org/apache/pig/newplan/logical/relational/
pig/trunk/test/org/apache/pig/newplan/logical/relational/TestLocationInPhysicalPlan.java
Modified:
pig/trunk/CHANGES.txt
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/CombinerOptimizer.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigGenericMapBase.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigGenericMapReduce.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/PhysicalOperator.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLimit.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLocalRearrange.java
pig/trunk/src/org/apache/pig/newplan/logical/expression/ExpToPhyTranslationVisitor.java
pig/trunk/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java
pig/trunk/src/org/apache/pig/tools/pigstats/JobStats.java
pig/trunk/src/org/apache/pig/tools/pigstats/ScriptState.java
pig/trunk/src/org/apache/pig/tools/pigstats/SimplePigStats.java
pig/trunk/test/org/apache/pig/test/data/GoldenFiles/MRC15.gld
pig/trunk/test/org/apache/pig/test/data/GoldenFiles/MRC18.gld
Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1337558&r1=1337557&r2=1337558&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Sat May 12 14:58:36 2012
@@ -24,6 +24,8 @@ INCOMPATIBLE CHANGES
IMPROVEMENTS
+PIG-2659: add source location of the aliases in the physical plan (julien)
+
PIG-2547: Easier UDFs: Convenient EvalFunc super-classes (billgraham, dvryaboy)
PIG-2639: Utils.getSchemaFromString should automatically give name to all types, but fails on boolean (jcoveney)
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/CombinerOptimizer.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/CombinerOptimizer.java?rev=1337558&r1=1337557&r2=1337558&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/CombinerOptimizer.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/CombinerOptimizer.java Sat May 12 14:58:36 2012
@@ -338,7 +338,7 @@ public class CombinerOptimizer extends M
String scope = combineFE.getOperatorKey().scope;
POPartialAgg poAgg = new POPartialAgg(new OperatorKey(scope,
NodeIdGenerator.getGenerator().getNextNodeId(scope)));
- poAgg.setAlias(combineFE.getAlias());
+ poAgg.addOriginalLocation(combineFE.getAlias(), combineFE.getOriginalLocations());
poAgg.setResultType(combineFE.getResultType());
//first plan in combine foreach is the group key
@@ -484,7 +484,7 @@ public class CombinerOptimizer extends M
private POForEach createForEachWithGrpProj(POForEach foreach, byte keyType) {
String scope = foreach.getOperatorKey().scope;
POForEach newFE = new POForEach(createOperatorKey(scope), new ArrayList<PhysicalPlan>());
- newFE.setAlias(foreach.getAlias());
+ newFE.addOriginalLocation(foreach.getAlias(), foreach.getOriginalLocations());
newFE.setResultType(foreach.getResultType());
//create plan that projects the group column
PhysicalPlan grpProjPlan = new PhysicalPlan();
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java?rev=1337558&r1=1337557&r2=1337558&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java Sat May 12 14:58:36 2012
@@ -2130,7 +2130,7 @@ public class MRCompiler extends PhyPlanV
keyType);
lr.setPlans(eps1);
lr.setResultType(DataType.TUPLE);
- lr.setAlias(sort.getAlias());
+ lr.addOriginalLocation(sort.getAlias(), sort.getOriginalLocations());
mro.mapPlan.addAsLeaf(lr);
mro.setMapDone(true);
@@ -2220,7 +2220,7 @@ public class MRCompiler extends PhyPlanV
POSort sort = new POSort(inpSort.getOperatorKey(), inpSort
.getRequestedParallelism(), null, inpSort.getSortPlans(),
inpSort.getMAscCols(), inpSort.getMSortFunc());
- sort.setAlias(inpSort.getAlias());
+ sort.addOriginalLocation(inpSort.getAlias(), inpSort.getOriginalLocations());
// Turn the asc/desc array into an array of strings so that we can pass it
// to the FindQuantiles function.
@@ -2425,7 +2425,7 @@ public class MRCompiler extends PhyPlanV
lr.setKeyType(DataType.CHARARRAY);
lr.setPlans(eps);
lr.setResultType(DataType.TUPLE);
- lr.setAlias(sort.getAlias());
+ lr.addOriginalLocation(sort.getAlias(), sort.getOriginalLocations());
mro.mapPlan.add(lr);
mro.mapPlan.connect(nfe1, lr);
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java?rev=1337558&r1=1337557&r2=1337558&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java Sat May 12 14:58:36 2012
@@ -285,6 +285,8 @@ public class MapReduceLauncher extends L
if (mro != null) {
String alias = ScriptState.get().getAlias(mro);
log.info("Processing aliases " + alias);
+ String aliasLocation = ScriptState.get().getAliasLocation(mro);
+ log.info("detailed locations: " + aliasLocation);
}
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java?rev=1337558&r1=1337557&r2=1337558&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java Sat May 12 14:58:36 2012
@@ -115,6 +115,7 @@ public class PigCombiner {
String msg = "Problem while configuring combiner's reduce plan.";
throw new RuntimeException(msg, ioe);
}
+ log.info("Aliases being processed per job phase (AliasName[line,offset]): " + jConf.get("pig.alias.location"));
}
/**
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigGenericMapBase.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigGenericMapBase.java?rev=1337558&r1=1337557&r2=1337558&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigGenericMapBase.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigGenericMapBase.java Sat May 12 14:58:36 2012
@@ -208,6 +208,7 @@ public abstract class PigGenericMapBase
PigStatusReporter.setContext(context);
+ log.info("Aliases being processed per job phase (AliasName[line,offset]): " + job.get("pig.alias.location"));
}
/**
@@ -246,6 +247,7 @@ public abstract class PigGenericMapBase
pigHadoopLogger.setReporter(PigStatusReporter.getInstance());
PhysicalOperator.setPigLogger(pigHadoopLogger);
+
}
if (mp.isEmpty()) {
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigGenericMapReduce.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigGenericMapReduce.java?rev=1337558&r1=1337557&r2=1337558&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigGenericMapReduce.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigGenericMapReduce.java Sat May 12 14:58:36 2012
@@ -348,6 +348,7 @@ public class PigGenericMapReduce {
String msg = "Problem while configuring reduce plan.";
throw new RuntimeException(msg, ioe);
}
+ log.info("Aliases being processed per job phase (AliasName[line,offset]): " + jConf.get("pig.alias.location"));
}
/**
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/PhysicalOperator.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/PhysicalOperator.java?rev=1337558&r1=1337557&r2=1337558&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/PhysicalOperator.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/PhysicalOperator.java Sat May 12 14:58:36 2012
@@ -17,6 +17,9 @@
*/
package org.apache.pig.backend.hadoop.executionengine.physicalLayer;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -34,6 +37,7 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.impl.plan.Operator;
import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.impl.plan.VisitorException;
+import org.apache.pig.parser.SourceLocation;
import org.apache.pig.pen.util.LineageTracer;
import org.apache.pig.pen.Illustrator;
import org.apache.pig.pen.Illustrable;
@@ -137,6 +141,8 @@ public abstract class PhysicalOperator e
private boolean accum;
private transient boolean accumStart;
+ private List<OriginalLocation> originalLocations = new ArrayList<OriginalLocation>();
+
public PhysicalOperator(OperatorKey k) {
this(k, -1, null);
}
@@ -185,8 +191,18 @@ public abstract class PhysicalOperator e
return (alias == null) ? "" : (alias + ": ");
}
- public void setAlias(String alias) {
+ public void addOriginalLocation(String alias, SourceLocation sourceLocation) {
+ this.alias = alias;
+ this.originalLocations.add(new OriginalLocation(alias, sourceLocation.line(), sourceLocation.offset()));
+ }
+
+ public void addOriginalLocation(String alias, List<OriginalLocation> originalLocations) {
this.alias = alias;
+ this.originalLocations.addAll(originalLocations);
+ }
+
+ public List<OriginalLocation> getOriginalLocations() {
+ return Collections.unmodifiableList(originalLocations);
}
public void setAccumulative() {
@@ -441,6 +457,7 @@ public abstract class PhysicalOperator e
protected void cloneHelper(PhysicalOperator op) {
resultType = op.resultType;
+ originalLocations.addAll(op.originalLocations);
}
/**
@@ -462,4 +479,33 @@ public abstract class PhysicalOperator e
return pigLogger;
}
+ public static class OriginalLocation implements Serializable {
+ private String alias;
+ private int line;
+ private int offset;
+
+ public OriginalLocation(String alias, int line, int offset) {
+ super();
+ this.alias = alias;
+ this.line = line;
+ this.offset = offset;
+}
+
+ public String getAlias() {
+ return alias;
+ }
+
+ public int getLine() {
+ return line;
+ }
+
+ public int getOffset() {
+ return offset;
+ }
+
+ @Override
+ public String toString() {
+ return alias+"["+line+","+offset+"]";
+ }
+ }
}
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java?rev=1337558&r1=1337557&r2=1337558&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java Sat May 12 14:58:36 2012
@@ -593,7 +593,7 @@ public class POForEach extends PhysicalO
requestedParallelism, plans, flattens);
clone.setOpsToBeReset(ops);
clone.setResultType(getResultType());
- clone.setAlias(alias);
+ clone.addOriginalLocation(alias, getOriginalLocations());
return clone;
}
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=1337558&r1=1337557&r2=1337558&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 Sat May 12 14:58:36 2012
@@ -162,7 +162,7 @@ public class POLimit extends PhysicalOpe
this.requestedParallelism, this.inputs);
newLimit.mLimit = this.mLimit;
newLimit.expressionPlan = this.expressionPlan.clone();
- newLimit.setAlias(alias);
+ newLimit.addOriginalLocation(alias, getOriginalLocations());
return newLimit;
}
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=1337558&r1=1337557&r2=1337558&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 Sat May 12 14:58:36 2012
@@ -687,7 +687,7 @@ public class POLocalRearrange extends Ph
// Needs to be called as setDistinct so that the fake index tuple gets
// created.
clone.setDistinct(mIsDistinct);
- clone.setAlias(alias);
+ clone.addOriginalLocation(alias, getOriginalLocations());
return clone;
}
Modified: pig/trunk/src/org/apache/pig/newplan/logical/expression/ExpToPhyTranslationVisitor.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/expression/ExpToPhyTranslationVisitor.java?rev=1337558&r1=1337557&r2=1337558&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/expression/ExpToPhyTranslationVisitor.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/expression/ExpToPhyTranslationVisitor.java Sat May 12 14:58:36 2012
@@ -287,7 +287,7 @@ public class ExpToPhyTranslationVisitor
nodeGen.getNextNodeId(DEFAULT_SCOPE)));
((POMapLookUp)physOp).setLookUpKey(op.getLookupKey() );
physOp.setResultType(op.getType());
- physOp.setAlias(op.getFieldSchema().alias);
+ physOp.addOriginalLocation(op.getFieldSchema().alias, op.getLocation());
currentPlan.add(physOp);
logToPhyMap.put(op, physOp);
Modified: pig/trunk/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java?rev=1337558&r1=1337557&r2=1337558&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java Sat May 12 14:58:36 2012
@@ -82,6 +82,7 @@ import org.apache.pig.newplan.logical.Ut
import org.apache.pig.newplan.logical.expression.ExpToPhyTranslationVisitor;
import org.apache.pig.newplan.logical.expression.LogicalExpressionPlan;
import org.apache.pig.newplan.logical.expression.ProjectExpression;
+import org.apache.pig.parser.SourceLocation;
public class LogToPhyTranslationVisitor extends LogicalRelationalNodesVisitor {
@@ -122,7 +123,7 @@ public class LogToPhyTranslationVisitor
// splittable due to LoadStore Refactor
POLoad load = new POLoad(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), loLoad.getLoadFunc());
- load.setAlias(loLoad.getAlias());
+ load.addOriginalLocation(loLoad.getAlias(), loLoad.getLocation());
load.setLFile(loLoad.getFileSpec());
load.setPc(pc);
load.setResultType(DataType.BAG);
@@ -155,7 +156,7 @@ public class LogToPhyTranslationVisitor
PONative poNative = new PONative(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)));
- poNative.setAlias(loNative.getAlias());
+ poNative.addOriginalLocation(loNative.getAlias(), loNative.getLocation());
poNative.setNativeMRjar(loNative.getNativeMRJar());
poNative.setParams(loNative.getParams());
poNative.setResultType(DataType.BAG);
@@ -190,7 +191,7 @@ public class LogToPhyTranslationVisitor
// System.err.println("Entering Filter");
POFilter poFilter = new POFilter(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), filter.getRequestedParallelisam());
- poFilter.setAlias(filter.getAlias());
+ poFilter.addOriginalLocation(filter.getAlias(), filter.getLocation());
poFilter.setResultType(DataType.BAG);
currentPlan.add(poFilter);
logToPhyMap.put(filter, poFilter);
@@ -267,7 +268,7 @@ public class LogToPhyTranslationVisitor
.getNextNodeId(scope)), sort.getRequestedParallelisam(), null,
sortPlans, sort.getAscendingCols(), comparator);
}
- poSort.setAlias(sort.getAlias());
+ poSort.addOriginalLocation(sort.getAlias(), sort.getLocation());
poSort.setLimit(sort.getLimit());
// sort.setRequestedParallelism(s.getType());
logToPhyMap.put(sort, poSort);
@@ -300,7 +301,7 @@ public class LogToPhyTranslationVisitor
List<Operator> inputs = cross.getPlan().getPredecessors(cross);
if (cross.isNested()) {
POCross physOp = new POCross(new OperatorKey(scope,nodeGen.getNextNodeId(scope)), cross.getRequestedParallelisam());
- physOp.setAlias(physOp.getAlias());
+ physOp.addOriginalLocation(physOp.getAlias(), physOp.getOriginalLocations());
currentPlan.add(physOp);
physOp.setResultType(DataType.BAG);
logToPhyMap.put(cross, physOp);
@@ -318,10 +319,10 @@ public class LogToPhyTranslationVisitor
POGlobalRearrange poGlobal = new POGlobalRearrange(new OperatorKey(
scope, nodeGen.getNextNodeId(scope)), cross
.getRequestedParallelisam());
- poGlobal.setAlias(cross.getAlias());
+ poGlobal.addOriginalLocation(cross.getAlias(), cross.getLocation());
POPackage poPackage = new POPackage(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), cross.getRequestedParallelisam());
- poGlobal.setAlias(cross.getAlias());
+ poGlobal.addOriginalLocation(cross.getAlias(), cross.getLocation());
currentPlan.add(poGlobal);
currentPlan.add(poPackage);
@@ -349,7 +350,7 @@ public class LogToPhyTranslationVisitor
ce1.setResultType(DataType.TUPLE);*/
POUserFunc gfc = new POUserFunc(new OperatorKey(scope, nodeGen.getNextNodeId(scope)),cross.getRequestedParallelisam(), Arrays.asList((PhysicalOperator)ce1,(PhysicalOperator)ce2), new FuncSpec(GFCross.class.getName()));
- gfc.setAlias(cross.getAlias());
+ gfc.addOriginalLocation(cross.getAlias(), cross.getLocation());
gfc.setResultType(DataType.BAG);
fep1.addAsLeaf(gfc);
gfc.setInputs(Arrays.asList((PhysicalOperator)ce1,(PhysicalOperator)ce2));
@@ -359,7 +360,7 @@ public class LogToPhyTranslationVisitor
PhysicalPlan fep2 = new PhysicalPlan();
POProject feproj = new POProject(new OperatorKey(scope, nodeGen.getNextNodeId(scope)), cross.getRequestedParallelisam());
- feproj.setAlias(cross.getAlias());
+ feproj.addOriginalLocation(cross.getAlias(), cross.getLocation());
feproj.setResultType(DataType.TUPLE);
feproj.setStar(true);
feproj.setOverloaded(false);
@@ -367,19 +368,19 @@ public class LogToPhyTranslationVisitor
List<PhysicalPlan> fePlans = Arrays.asList(fep1, fep2);
POForEach fe = new POForEach(new OperatorKey(scope, nodeGen.getNextNodeId(scope)), cross.getRequestedParallelisam(), fePlans, flattenLst );
- fe.setAlias(cross.getAlias());
+ fe.addOriginalLocation(cross.getAlias(), cross.getLocation());
currentPlan.add(fe);
currentPlan.connect(logToPhyMap.get(op), fe);
POLocalRearrange physOp = new POLocalRearrange(new OperatorKey(
scope, nodeGen.getNextNodeId(scope)), cross
.getRequestedParallelisam());
- physOp.setAlias(cross.getAlias());
+ physOp.addOriginalLocation(cross.getAlias(), cross.getLocation());
List<PhysicalPlan> lrPlans = new ArrayList<PhysicalPlan>();
for(int i=0;i<inputs.size();i++){
PhysicalPlan lrp1 = new PhysicalPlan();
POProject lrproj1 = new POProject(new OperatorKey(scope, nodeGen.getNextNodeId(scope)), cross.getRequestedParallelisam(), i);
- lrproj1.setAlias(cross.getAlias());
+ lrproj1.addOriginalLocation(cross.getAlias(), cross.getLocation());
lrproj1.setOverloaded(false);
lrproj1.setResultType(DataType.INTEGER);
lrp1.add(lrproj1);
@@ -420,7 +421,7 @@ public class LogToPhyTranslationVisitor
for(int i=1;i<=count;i++){
PhysicalPlan fep1 = new PhysicalPlan();
POProject feproj1 = new POProject(new OperatorKey(scope, nodeGen.getNextNodeId(scope)), cross.getRequestedParallelisam(), i);
- feproj1.setAlias(cross.getAlias());
+ feproj1.addOriginalLocation(cross.getAlias(), cross.getLocation());
feproj1.setResultType(DataType.BAG);
feproj1.setOverloaded(false);
fep1.add(feproj1);
@@ -429,7 +430,7 @@ public class LogToPhyTranslationVisitor
}
POForEach fe = new POForEach(new OperatorKey(scope, nodeGen.getNextNodeId(scope)), cross.getRequestedParallelisam(), fePlans, flattenLst );
- fe.setAlias(cross.getAlias());
+ fe.addOriginalLocation(cross.getAlias(), cross.getLocation());
currentPlan.add(fe);
try{
currentPlan.connect(poPackage, fe);
@@ -448,7 +449,7 @@ public class LogToPhyTranslationVisitor
POStream poStream = new POStream(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), stream.getExecutableManager(),
stream.getStreamingCommand(), this.pc.getProperties());
- poStream.setAlias(stream.getAlias());
+ poStream.addOriginalLocation(stream.getAlias(), stream.getLocation());
currentPlan.add(poStream);
logToPhyMap.put(stream, poStream);
@@ -596,7 +597,7 @@ public class LogToPhyTranslationVisitor
}
POForEach poFE = new POForEach(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), foreach.getRequestedParallelisam(), innerPlans, flattenList);
- poFE.setAlias(foreach.getAlias());
+ poFE.addOriginalLocation(foreach.getAlias(), foreach.getLocation());
poFE.setResultType(DataType.BAG);
logToPhyMap.put(foreach, poFE);
currentPlan.add(poFE);
@@ -670,7 +671,7 @@ public class LogToPhyTranslationVisitor
// System.err.println("Entering Store");
POStore store = new POStore(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)));
- store.setAlias(loStore.getAlias());
+ store.addOriginalLocation(loStore.getAlias(), loStore.getLocation());
store.setSFile(loStore.getOutputSpec());
store.setInputSpec(loStore.getInputSpec());
store.setSignature(loStore.getSignature());
@@ -743,7 +744,7 @@ public class LogToPhyTranslationVisitor
List<LogicalExpressionPlan> exprPlans = cg.getExpressionPlans().get(0);
POCollectedGroup physOp = new POCollectedGroup(new OperatorKey(
DEFAULT_SCOPE, nodeGen.getNextNodeId(DEFAULT_SCOPE)));
- physOp.setAlias(cg.getAlias());
+ physOp.addOriginalLocation(cg.getAlias(), cg.getLocation());
List<PhysicalPlan> pExprPlans = translateExpressionPlans(cg, exprPlans);
try {
@@ -794,7 +795,7 @@ public class LogToPhyTranslationVisitor
List<LogicalExpressionPlan> plans = innerPlans.get(i);
POLocalRearrange poInnerLR = new POLocalRearrange(new OperatorKey(DEFAULT_SCOPE, nodeGen.getNextNodeId(DEFAULT_SCOPE)));
- poInnerLR.setAlias(relationalOp.getAlias());
+ poInnerLR.addOriginalLocation(relationalOp.getAlias(), relationalOp.getLocation());
// LR will contain list of physical plans, because there could be
// multiple keys and each key can be an expression.
List<PhysicalPlan> exprPlans = translateExpressionPlans(relationalOp, plans);
@@ -832,7 +833,7 @@ public class LogToPhyTranslationVisitor
validateMapSideMerge(inputs, cg.getPlan());
POMergeCogroup poCogrp = compileToMergeCogrp(cg, cg.getExpressionPlans());
poCogrp.setResultType(DataType.TUPLE);
- poCogrp.setAlias(cg.getAlias());
+ poCogrp.addOriginalLocation(cg.getAlias(), cg.getLocation());
currentPlan.add(poCogrp);
for (Operator op : inputs) {
try {
@@ -895,6 +896,7 @@ public class LogToPhyTranslationVisitor
boolean[] innerFlags = loj.getInnerFlags();
String alias = loj.getAlias();
+ SourceLocation location = loj.getLocation();
int parallel = loj.getRequestedParallelisam();
for (int i=0;i<inputs.size();i++) {
@@ -920,7 +922,7 @@ public class LogToPhyTranslationVisitor
try {
skj = new POSkewedJoin(new OperatorKey(scope,nodeGen.getNextNodeId(scope)),
parallel,inp, innerFlags);
- skj.setAlias(alias);
+ skj.addOriginalLocation(alias, location);
skj.setJoinPlans(joinPlans);
}
catch (Exception e) {
@@ -1006,7 +1008,7 @@ public class LogToPhyTranslationVisitor
pfrj = new POFRJoin(new OperatorKey(scope,nodeGen.getNextNodeId(scope)),parallel,
inp, ppLists, keyTypes, null, fragment, isLeftOuter, nullTuple);
- pfrj.setAlias(alias);
+ pfrj.addOriginalLocation(alias, location);
} catch (ExecException e1) {
int errCode = 2058;
String msg = "Unable to set index on newly create POLocalRearrange.";
@@ -1054,7 +1056,7 @@ public class LogToPhyTranslationVisitor
smj.setResultType(DataType.TUPLE);
currentPlan.add(smj);
- smj.setAlias(alias);
+ smj.addOriginalLocation(alias, location);
for (Operator op : inputs) {
try {
currentPlan.connect(logToPhyMap.get(op), smj);
@@ -1068,7 +1070,7 @@ public class LogToPhyTranslationVisitor
if(!usePOMergeJoin){
// Now create and configure foreach which will flatten the output
// of cogroup.
- POForEach fe = compileFE4Flattening(innerFlags, scope, parallel, alias, inputs);
+ POForEach fe = compileFE4Flattening(innerFlags, scope, parallel, alias, location, inputs);
currentPlan.add(fe);
try {
currentPlan.connect(smj, fe);
@@ -1082,7 +1084,7 @@ public class LogToPhyTranslationVisitor
}
else if (loj.getJoinType() == LOJoin.JOINTYPE.HASH){
POPackage poPackage = compileToLR_GR_PackTrio(loj, loj.getCustomPartitioner(), innerFlags, loj.getExpressionPlans());
- POForEach fe = compileFE4Flattening(innerFlags, scope, parallel, alias, inputs);
+ POForEach fe = compileFE4Flattening(innerFlags, scope, parallel, alias, location, inputs);
currentPlan.add(fe);
try {
currentPlan.connect(poPackage, fe);
@@ -1101,11 +1103,11 @@ public class LogToPhyTranslationVisitor
POGlobalRearrange poGlobal = new POGlobalRearrange(new OperatorKey(
DEFAULT_SCOPE, nodeGen.getNextNodeId(DEFAULT_SCOPE)), relationalOp.getRequestedParallelisam());
- poGlobal.setAlias(relationalOp.getAlias());
+ poGlobal.addOriginalLocation(relationalOp.getAlias(), relationalOp.getLocation());
poGlobal.setCustomPartitioner(customPartitioner);
POPackage poPackage = new POPackage(new OperatorKey(DEFAULT_SCOPE, nodeGen
.getNextNodeId(DEFAULT_SCOPE)), relationalOp.getRequestedParallelisam());
- poPackage.setAlias(relationalOp.getAlias());
+ poPackage.addOriginalLocation(relationalOp.getAlias(), relationalOp.getLocation());
currentPlan.add(poGlobal);
currentPlan.add(poPackage);
@@ -1125,7 +1127,7 @@ public class LogToPhyTranslationVisitor
List<LogicalExpressionPlan> plans = innerPlans.get(i);
POLocalRearrange physOp = new POLocalRearrange(new OperatorKey(
DEFAULT_SCOPE, nodeGen.getNextNodeId(DEFAULT_SCOPE)), relationalOp.getRequestedParallelisam());
- physOp.setAlias(relationalOp.getAlias());
+ physOp.addOriginalLocation(relationalOp.getAlias(), relationalOp.getLocation());
List<PhysicalPlan> exprPlans = translateExpressionPlans(relationalOp, plans);
try {
physOp.setPlans(exprPlans);
@@ -1170,7 +1172,7 @@ public class LogToPhyTranslationVisitor
}
private POForEach compileFE4Flattening(boolean[] innerFlags,String scope,
- int parallel, String alias, List<Operator> inputs)
+ int parallel, String alias, SourceLocation location, List<Operator> inputs)
throws FrontendException {
List<PhysicalPlan> fePlans = new ArrayList<PhysicalPlan>();
@@ -1181,7 +1183,7 @@ public class LogToPhyTranslationVisitor
PhysicalPlan fep1 = new PhysicalPlan();
POProject feproj1 = new POProject(new OperatorKey(scope, nodeGen.getNextNodeId(scope)),
parallel, i+1); //i+1 since the first column is the "group" field
- feproj1.setAlias(alias);
+ feproj1.addOriginalLocation(alias, location);
feproj1.setResultType(DataType.BAG);
feproj1.setOverloaded(false);
fep1.add(feproj1);
@@ -1202,7 +1204,7 @@ public class LogToPhyTranslationVisitor
fe = new POForEach(new OperatorKey(scope, nodeGen.getNextNodeId(scope)),
parallel, fePlans, flattenLst );
- fe.setAlias(alias);
+ fe.addOriginalLocation(alias, location);
}catch (PlanException e1) {
int errCode = 2015;
@@ -1216,7 +1218,7 @@ public class LogToPhyTranslationVisitor
public void visit(LOUnion loUnion) throws FrontendException {
String scope = DEFAULT_SCOPE;
POUnion physOp = new POUnion(new OperatorKey(scope,nodeGen.getNextNodeId(scope)), loUnion.getRequestedParallelisam());
- physOp.setAlias(loUnion.getAlias());
+ physOp.addOriginalLocation(loUnion.getAlias(), loUnion.getLocation());
currentPlan.add(physOp);
physOp.setResultType(DataType.BAG);
logToPhyMap.put(loUnion, physOp);
@@ -1238,7 +1240,7 @@ public class LogToPhyTranslationVisitor
public void visit(LODistinct loDistinct) throws FrontendException {
String scope = DEFAULT_SCOPE;
PODistinct physOp = new PODistinct(new OperatorKey(scope,nodeGen.getNextNodeId(scope)), loDistinct.getRequestedParallelisam());
- physOp.setAlias(loDistinct.getAlias());
+ physOp.addOriginalLocation(loDistinct.getAlias(), loDistinct.getLocation());
currentPlan.add(physOp);
physOp.setResultType(DataType.BAG);
logToPhyMap.put(loDistinct, physOp);
@@ -1260,7 +1262,7 @@ public class LogToPhyTranslationVisitor
POLimit poLimit = new POLimit(new OperatorKey(scope, nodeGen.getNextNodeId(scope)),
loLimit.getRequestedParallelisam());
poLimit.setLimit(loLimit.getLimit());
- poLimit.setAlias(loLimit.getAlias());
+ poLimit.addOriginalLocation(loLimit.getAlias(), loLimit.getLocation());
poLimit.setResultType(DataType.BAG);
currentPlan.add(poLimit);
logToPhyMap.put(loLimit, poLimit);
@@ -1297,7 +1299,7 @@ public class LogToPhyTranslationVisitor
String scope = DEFAULT_SCOPE;
POSplit physOp = new POSplit(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), loSplit.getRequestedParallelisam());
- physOp.setAlias(loSplit.getAlias());
+ physOp.addOriginalLocation(loSplit.getAlias(), loSplit.getLocation());
FileSpec splStrFile;
try {
splStrFile = new FileSpec(FileLocalizer.getTemporaryPath(pc).toString(),new FuncSpec(Utils.getTmpFileCompressorName(pc)));
@@ -1350,7 +1352,7 @@ public class LogToPhyTranslationVisitor
// System.err.println("Entering Filter");
POFilter poFilter = new POFilter(new OperatorKey(scope, nodeGen
.getNextNodeId(scope)), loSplitOutput.getRequestedParallelisam());
- poFilter.setAlias(loSplitOutput.getAlias());
+ poFilter.addOriginalLocation(loSplitOutput.getAlias(), loSplitOutput.getLocation());
poFilter.setResultType(DataType.BAG);
currentPlan.add(poFilter);
logToPhyMap.put(loSplitOutput, poFilter);
Modified: pig/trunk/src/org/apache/pig/tools/pigstats/JobStats.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/tools/pigstats/JobStats.java?rev=1337558&r1=1337557&r2=1337558&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/tools/pigstats/JobStats.java (original)
+++ pig/trunk/src/org/apache/pig/tools/pigstats/JobStats.java Sat May 12 14:58:36 2012
@@ -63,6 +63,7 @@ import org.apache.pig.tools.pigstats.Sim
public final class JobStats extends Operator {
public static final String ALIAS = "JobStatistics:alias";
+ public static final String ALIAS_LOCATION = "JobStatistics:alias_location";
public static final String FEATURE = "JobStatistics:feature";
public static final String SUCCESS_HEADER = "JobId\tMaps\tReduces\t" +
@@ -199,6 +200,10 @@ public final class JobStats extends Oper
return (String)getAnnotation(ALIAS);
}
+ public String getAliasLocation() {
+ return (String)getAnnotation(ALIAS_LOCATION);
+ }
+
public String getFeature() {
return (String)getAnnotation(FEATURE);
}
@@ -266,13 +271,13 @@ public final class JobStats extends Oper
if (conf == null) return;
this.conf = conf;
try {
- mapStores = (List<POStore>) ObjectSerializer.deserialize(conf
+ this.mapStores = (List<POStore>) ObjectSerializer.deserialize(conf
.get(JobControlCompiler.PIG_MAP_STORES));
- reduceStores = (List<POStore>) ObjectSerializer.deserialize(conf
+ this.reduceStores = (List<POStore>) ObjectSerializer.deserialize(conf
.get(JobControlCompiler.PIG_REDUCE_STORES));
- loads = (ArrayList<FileSpec>) ObjectSerializer.deserialize(conf
+ this.loads = (ArrayList<FileSpec>) ObjectSerializer.deserialize(conf
.get("pig.inputs"));
- disableCounter = conf.getBoolean("pig.disable.counter", false);
+ this.disableCounter = conf.getBoolean("pig.disable.counter", false);
} catch (IOException e) {
LOG.warn("Failed to deserialize the store list", e);
}
@@ -440,6 +445,7 @@ public final class JobStats extends Oper
void setAlias(MapReduceOper mro) {
annotate(ALIAS, ScriptState.get().getAlias(mro));
+ annotate(ALIAS_LOCATION, ScriptState.get().getAliasLocation(mro));
annotate(FEATURE, ScriptState.get().getPigFeature(mro));
}
Modified: pig/trunk/src/org/apache/pig/tools/pigstats/ScriptState.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/tools/pigstats/ScriptState.java?rev=1337558&r1=1337557&r2=1337558&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/tools/pigstats/ScriptState.java (original)
+++ pig/trunk/src/org/apache/pig/tools/pigstats/ScriptState.java Sat May 12 14:58:36 2012
@@ -46,6 +46,7 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.NativeMapReduceOper;
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.PhysicalOperator.OriginalLocation;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POCollectedGroup;
@@ -115,7 +116,8 @@ public class ScriptState {
JOB_PARENTS ("pig.parent.jobid"),
JOB_FEATURE ("pig.job.feature"),
SCRIPT_FEATURES ("pig.script.features"),
- JOB_ALIAS ("pig.alias");
+ JOB_ALIAS ("pig.alias"),
+ JOB_ALIAS_LOCATION ("pig.alias.location");
private String displayStr;
@@ -186,7 +188,8 @@ public class ScriptState {
private PigContext pigContext;
private Map<MapReduceOper, String> featureMap = null;
- private Map<MapReduceOper, String> aliasMap = null;
+ private Map<MapReduceOper, String> aliasMap = new HashMap<MapReduceOper, String>();
+ private Map<MapReduceOper, String> aliasLocationMap = new HashMap<MapReduceOper, String>();
private List<PigProgressNotificationListener> listeners
= new ArrayList<PigProgressNotificationListener>();
@@ -237,6 +240,7 @@ public class ScriptState {
}
}
+
public void emitJobsSubmittedNotification(int numJobsSubmitted) {
for (PigProgressNotificationListener listener: listeners) {
listener.jobsSubmittedNotification(id, numJobsSubmitted);
@@ -371,6 +375,7 @@ public class ScriptState {
public String getPigVersion() {
if (pigVersion == null) {
String findContainingJar = JarManager.findContainingJar(ScriptState.class);
+ if (findContainingJar != null) {
try {
JarFile jar = new JarFile(findContainingJar);
final Manifest manifest = jar.getManifest();
@@ -380,6 +385,9 @@ public class ScriptState {
} catch (Exception e) {
LOG.warn("unable to read pigs manifest file");
}
+ } else {
+ LOG.warn("unable to read pigs manifest file. Not running from the Pig jar");
+ }
}
return (pigVersion == null) ? "" : pigVersion;
}
@@ -427,6 +435,7 @@ public class ScriptState {
String.valueOf(scriptFeatures));
}
conf.set(PIG_PROPERTY.JOB_ALIAS.toString(), getAlias(mro));
+ conf.set(PIG_PROPERTY.JOB_ALIAS_LOCATION.toString(), getAliasLocation(mro));
}
private void setJobParents(MapReduceOper mro, Configuration conf) {
@@ -462,27 +471,45 @@ public class ScriptState {
}
public String getAlias(MapReduceOper mro) {
- if (aliasMap == null) {
- aliasMap = new HashMap<MapReduceOper, String>();
+ if (!aliasMap.containsKey(mro)) {
+ setAlias(mro);
}
- String retStr = aliasMap.get(mro);
- if (retStr == null) {
+ return aliasMap.get(mro);
+ }
+
+ private void setAlias(MapReduceOper mro) {
ArrayList<String> alias = new ArrayList<String>();
+ String aliasLocationStr = "";
try {
- new AliasVisitor(mro.mapPlan, alias).visit();
- new AliasVisitor(mro.reducePlan, alias).visit();
+ ArrayList<String> aliasLocation = new ArrayList<String>();
+ new AliasVisitor(mro.mapPlan, alias, aliasLocation).visit();
+ aliasLocationStr += "M: "+LoadFunc.join(aliasLocation, ",");
+ if (mro.combinePlan != null) {
+ aliasLocation = new ArrayList<String>();
+ new AliasVisitor(mro.combinePlan, alias, aliasLocation).visit();
+ aliasLocationStr += " C: "+LoadFunc.join(aliasLocation, ",");
+ }
+ aliasLocation = new ArrayList<String>();
+ new AliasVisitor(mro.reducePlan, alias, aliasLocation).visit();
+ aliasLocationStr += " R: "+LoadFunc.join(aliasLocation, ",");
if (!alias.isEmpty()) {
Collections.sort(alias);
}
} catch (VisitorException e) {
LOG.warn("unable to get alias", e);
}
- retStr = LoadFunc.join(alias, ",");
- aliasMap.put(mro, retStr);
+ aliasMap.put(mro, LoadFunc.join(alias, ","));
+ aliasLocationMap.put(mro, aliasLocationStr);
}
- return retStr;
+
+ public String getAliasLocation(MapReduceOper mro) {
+ if (!aliasLocationMap.containsKey(mro)) {
+ setAlias(mro);
+ }
+ return aliasLocationMap.get(mro);
}
+
public String getPigFeature(MapReduceOper mro) {
if (featureMap == null) {
featureMap = new HashMap<MapReduceOper, String>();
@@ -732,10 +759,13 @@ public class ScriptState {
private List<String> alias;
- public AliasVisitor(PhysicalPlan plan, List<String> alias) {
+ private final List<String> aliasLocation;
+
+ public AliasVisitor(PhysicalPlan plan, List<String> alias, List<String> aliasLocation) {
super(plan, new DepthFirstWalker<PhysicalOperator, PhysicalPlan>(
plan));
this.alias = alias;
+ this.aliasLocation = aliasLocation;
aliasSet = new HashSet<String>();
if (!alias.isEmpty()) {
for (String s : alias) aliasSet.add(s);
@@ -745,73 +775,87 @@ public class ScriptState {
@Override
public void visitLoad(POLoad load) throws VisitorException {
setAlias(load);
+ super.visitLoad(load);
}
@Override
public void visitFRJoin(POFRJoin join) throws VisitorException {
setAlias(join);
+ super.visitFRJoin(join);
}
@Override
public void visitMergeJoin(POMergeJoin join) throws VisitorException {
setAlias(join);
+ super.visitMergeJoin(join);
}
@Override
public void visitMergeCoGroup(POMergeCogroup mergeCoGrp)
throws VisitorException {
setAlias(mergeCoGrp);
+ super.visitMergeCoGroup(mergeCoGrp);
}
@Override
public void visitCollectedGroup(POCollectedGroup mg)
throws VisitorException {
setAlias(mg);
+ super.visitCollectedGroup(mg);
}
@Override
public void visitDistinct(PODistinct distinct) throws VisitorException {
setAlias(distinct);
+ super.visitDistinct(distinct);
}
@Override
public void visitStream(POStream stream) throws VisitorException {
setAlias(stream);
+ super.visitStream(stream);
}
@Override
public void visitFilter(POFilter fl) throws VisitorException {
setAlias(fl);
+ super.visitFilter(fl);
}
@Override
public void visitLocalRearrange(POLocalRearrange lr) throws VisitorException {
setAlias(lr);
+ super.visitLocalRearrange(lr);
}
@Override
public void visitPOForEach(POForEach nfe) throws VisitorException {
setAlias(nfe);
+ super.visitPOForEach(nfe);
}
@Override
public void visitUnion(POUnion un) throws VisitorException {
setAlias(un);
+ super.visitUnion(un);
}
@Override
public void visitSort(POSort sort) throws VisitorException {
setAlias(sort);
+ super.visitSort(sort);
}
@Override
public void visitLimit(POLimit lim) throws VisitorException {
setAlias(lim);
+ super.visitLimit(lim);
}
@Override
public void visitSkewedJoin(POSkewedJoin sk) throws VisitorException {
setAlias(sk);
+ super.visitSkewedJoin(sk);
}
private void setAlias(PhysicalOperator op) {
@@ -822,7 +866,11 @@ public class ScriptState {
aliasSet.add(s);
}
}
+ List<OriginalLocation> originalLocations = op.getOriginalLocations();
+ for (OriginalLocation originalLocation : originalLocations) {
+ aliasLocation.add(originalLocation.toString());
}
}
+ }
}
Modified: pig/trunk/src/org/apache/pig/tools/pigstats/SimplePigStats.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/tools/pigstats/SimplePigStats.java?rev=1337558&r1=1337557&r2=1337558&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/tools/pigstats/SimplePigStats.java (original)
+++ pig/trunk/src/org/apache/pig/tools/pigstats/SimplePigStats.java Sat May 12 14:58:36 2012
@@ -24,6 +24,7 @@ import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
@@ -43,6 +44,8 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.NativeMapReduceOper;
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.relationalOperators.POStore;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.util.PlanHelper;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.plan.DependencyOrderWalker;
Added: pig/trunk/test/org/apache/pig/newplan/logical/relational/TestLocationInPhysicalPlan.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/newplan/logical/relational/TestLocationInPhysicalPlan.java?rev=1337558&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/newplan/logical/relational/TestLocationInPhysicalPlan.java (added)
+++ pig/trunk/test/org/apache/pig/newplan/logical/relational/TestLocationInPhysicalPlan.java Sat May 12 14:58:36 2012
@@ -0,0 +1,48 @@
+package org.apache.pig.newplan.logical.relational;
+
+import java.io.File;
+import java.util.List;
+
+import junit.framework.Assert;
+
+import org.apache.pig.ExecType;
+import org.apache.pig.PigServer;
+import org.apache.pig.backend.executionengine.ExecJob;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator.OriginalLocation;
+import org.apache.pig.test.Util;
+import org.apache.pig.tools.pigstats.JobStats;
+import org.junit.Test;
+
+public class TestLocationInPhysicalPlan {
+
+ @Test
+ public void test() throws Exception {
+ File input = File.createTempFile("test", "input");
+ input.deleteOnExit();
+ File output = File.createTempFile("test", "output");
+ output.delete();
+ Util.createLocalInputFile(input.getAbsolutePath(), new String[] {
+ "1,2,3",
+ "1,1,3",
+ "1,1,1",
+ "3,1,1",
+ "1,2,1",
+ });
+ PigServer pigServer = new PigServer(ExecType.LOCAL);
+ pigServer.setBatchOn();
+ pigServer.registerQuery(
+ "A = LOAD '" + input.getAbsolutePath() + "' using PigStorage();\n"
+ + "B = GROUP A BY $0;\n"
+ + "A = FOREACH B GENERATE COUNT(A);\n"
+ + "STORE A INTO '" + output.getAbsolutePath() + "';");
+ ExecJob job = pigServer.executeBatch().get(0);
+ List<OriginalLocation> originalLocations = job.getPOStore().getOriginalLocations();
+ Assert.assertEquals(1, originalLocations.size());
+ OriginalLocation originalLocation = originalLocations.get(0);
+ Assert.assertEquals(4, originalLocation.getLine());
+ Assert.assertEquals(0, originalLocation.getOffset());
+ Assert.assertEquals("A", originalLocation.getAlias());
+ JobStats jStats = (JobStats)job.getStatistics().getJobGraph().getSinks().get(0);
+ Assert.assertEquals("M: A[1,4],A[3,4],B[2,4] C: A[3,4],B[2,4] R: A[3,4]", jStats.getAliasLocation());
+ }
+}
Modified: pig/trunk/test/org/apache/pig/test/data/GoldenFiles/MRC15.gld
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/data/GoldenFiles/MRC15.gld?rev=1337558&r1=1337557&r2=1337558&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/data/GoldenFiles/MRC15.gld (original)
+++ pig/trunk/test/org/apache/pig/test/data/GoldenFiles/MRC15.gld Sat May 12 14:58:36 2012
@@ -88,4 +88,4 @@ MapReduce(1,GFCross) - -29:
| | |
| | Constant(true) - --6860857982727545176
| |
- | |---Load(DummyFil:DummyLdr) - -737965411848544085
+ | |---Load(DummyFil:DummyLdr) - -737965411848544085
\ No newline at end of file
Modified: pig/trunk/test/org/apache/pig/test/data/GoldenFiles/MRC18.gld
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/data/GoldenFiles/MRC18.gld?rev=1337558&r1=1337557&r2=1337558&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/data/GoldenFiles/MRC18.gld (original)
+++ pig/trunk/test/org/apache/pig/test/data/GoldenFiles/MRC18.gld Sat May 12 14:58:36 2012
@@ -18,4 +18,4 @@ Reduce Plan Empty
| | |
| | Project[tuple][*] - scope-111
| |
- | |---b: Load(/tmp/input2:org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MergeJoinIndexer('org.apache.pig.builtin.PigStorage','kmonaaafhdhcaabdgkgbhggbcohfhegjgmcoebhchcgbhjemgjhdhehiibncbnjjmhgbjnadaaabejaaaehdgjhkgfhihaaaaaaaabhhaeaaaaaaabhdhcaaeogphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccohagmgbgohdcofagihjhdgjgdgbgmfagmgbgoaaaaaaaaaaaaaaabacaaacfkaaangfgogeepggebgmgmejgohahfheemaaafgphagngbhaheaacdemgphcghcpgbhagbgdgigfcphagjghcpgjgnhagmcphfhegjgmcpenhfgmhegjengbhadlhihcaacfgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcfagmgbgoimlohdjmoaemmdgiacaaaiemaaakgneghcgpgnefgeghgfhdhbaahoaaademaaafgnelgfhjhdheaaapemgkgbhggbcphfhegjgmcpengbhadlemaaahgnemgfgbhggfhdheaabaemgkgbhggbcphfhegjgmcpemgjhdhedlemaaaegnephahdhbaahoaaafemaaaggnfcgpgphehdhbaahoaaagemaaaognfdgpggheeghcgpgnefgeghgfhdhbaahoaaademaaamgnfdgpgghefegpefgeghgfhdhbaahoaaademaaaignfegpefgeghgfhdhbaahoaaad
hihahdhcaacbgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohfhegjgmcoenhfgmhegjengbhaaaaaaaaaaaaaaaacacaaabemaaaegnengbhahbaahoaaafhihahdhcaabbgkgbhggbcohfhegjgmcoeigbhdgiengbhaafahnkmbmdbgganbadaaacegaaakgmgpgbgeeggbgdhegphcejaaajhegihcgfhdgigpgmgehihadpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaaahihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaabhdhcaacegphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcelgfhjaaaaaaaaaaaaaaabacaaacekaaacgjgeemaaafhdgdgphagfheaabcemgkgbhggbcpgmgbgoghcpfdhehcgjgoghdlhihaaaaaaaaaaaaaaagiheaaafhdgdgphagfhdhcaafjgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccogfhihahcgfhdhdgjgpgoephagfhcgbhegphchdcofaepfahcgpgkgfgdheaaaaaaaaaaaaaaabacaaagfkaaaogjhdfahcgpgkgfgdhefegpefgogefkaaakgphggfhcgmgpgbgegfgefkaabfhahcgpgdgfhdhdgjgoghecgbghepggfehfhagmgfhdfkaabehcgfhdhfgmhefdgjgoghgmgffehfhagmgfecgbghejaaaihdhegbhcheedgpgmemaaahgdgpgmhfgngohdheaabfemgkgbhggbcphfhegjgmcpebhchcgbhjemgjhdhed
lhihcaagcgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccogfhihahcgfhdhdgjgpgoephagfhcgbhegphchdcoefhihahcgfhdhdgjgpgoephagfhcgbhegphcaaaaaaaaaaaaaaabacaaaahihcaaemgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccofagihjhdgjgdgbgmephagfhcgbhegphcaaaaaaaaaaaaaaabacaaalfkaaafgbgdgdhfgnfkaaangjgohahfheebhehegbgdgigfgeejaabehcgfhbhfgfhdhegfgefagbhcgbgmgmgfgmgjhdgnecaaakhcgfhdhfgmhefehjhagfemaaafgbgmgjgbhdhbaahoaaaoemaaafgjgohahfheheaablemgphcghcpgbhagbgdgigfcphagjghcpgegbhegbcpfehfhagmgfdlemaaaggjgohahfhehdhbaahoaaagemaaangmgjgogfgbghgffehcgbgdgfhcheaachemgphcghcpgbhagbgdgigfcphagjghcphagfgocphfhegjgmcpemgjgogfgbghgffehcgbgdgfhcdlemaaahgphfhehahfhehdhbaahoaaagemaaakhagbhcgfgohefagmgbgoheaafaemgphcghcpgbhagbgdgigfcphagjghcpgcgbgdglgfgogecpgigbgegpgphacpgfhigfgdhfhegjgpgogfgoghgjgogfcphagihjhdgjgdgbgmemgbhjgfhccphagmgbgohdcpfagihjhdgjgdgbgmfa
gmgbgodlemaaadhcgfhdheaaeeemgphcghcpgbhagbgdgigfcphagjghcpgcgbgdglgfgogecpgigbgegpgphacpgfhigfgdhfhegjgpgogfgoghgjgogfcphagihjhdgjgdgbgmemgbhjgfhccpfcgfhdhfgmhedlhihcaacbgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcaaaaaaaaaaaaaaabacaaabemaaaegnelgfhjheaacgemgphcghcpgbhagbgdgigfcphagjghcpgjgnhagmcphagmgbgocpephagfhcgbhegphcelgfhjdlhihahbaahoaaapaaaappppppppdchahahahahahahdhcaaecgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccofcgfhdhfgmheaaaaaaaaaaaaaaabacaaacecaaamhcgfhehfhcgofdhegbhehfhdemaaaghcgfhdhfgmheheaabcemgkgbhggbcpgmgbgoghcpepgcgkgfgdhedlhihaachaaaaaaaaaaaaaaaaahdhbaahoaaaaaaaaaaabhhaeaaaaaaakhdhcaabbgkgbhggbcogmgbgoghcoejgohegfghgfhcbcockakephibihdiacaaabejaaafhggbgmhfgfhihcaabagkgbhggbcogmgbgoghcoeohfgngcgfhcigkmjfbnaljeoailacaaaahihaaaaaaaaahihihdhbaahoaaaaaaaaaaabhhaeaaaaaaakhbaahoaablhihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaabhbaahoaablhbaahoaaaphihdhbaahoaaaaaaa
aaaaahhaeaaaaaaakhihdhbaahoaaaihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaaahihdhbaahoaaaihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaaahihdhbaahoaaaihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaaahiaahahi','','b','scope','true')) - scope-102
\ No newline at end of file
+ | |---b: Load(/tmp/input2:org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MergeJoinIndexer('org.apache.pig.builtin.PigStorage','kmonaaafhdhcaabdgkgbhggbcohfhegjgmcoebhchcgbhjemgjhdhehiibncbnjjmhgbjnadaaabejaaaehdgjhkgfhihaaaaaaaabhhaeaaaaaaabhdhcaaeogphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccohagmgbgohdcofagihjhdgjgdgbgmfagmgbgoaaaaaaaaaaaaaaabacaaacfkaaangfgogeepggebgmgmejgohahfheemaaafgphagngbhaheaacdemgphcghcpgbhagbgdgigfcphagjghcpgjgnhagmcphfhegjgmcpenhfgmhegjengbhadlhihcaacfgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcfagmgbgoimlohdjmoaemmdgiacaaaiemaaakgneghcgpgnefgeghgfhdhbaahoaaademaaafgnelgfhjhdheaaapemgkgbhggbcphfhegjgmcpengbhadlemaaahgnemgfgbhggfhdheaabaemgkgbhggbcphfhegjgmcpemgjhdhedlemaaaegnephahdhbaahoaaafemaaaggnfcgpgphehdhbaahoaaagemaaaognfdgpggheeghcgpgnefgeghgfhdhbaahoaaademaaamgnfdgpgghefegpefgeghgfhdhbaahoaaademaaaignfegpefgeghgfhdhbaahoaaad
hihahdhcaacbgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohfhegjgmcoenhfgmhegjengbhaaaaaaaaaaaaaaaacacaaabemaaaegnengbhahbaahoaaafhihahdhcaabbgkgbhggbcohfhegjgmcoeigbhdgiengbhaafahnkmbmdbgganbadaaacegaaakgmgpgbgeeggbgdhegphcejaaajhegihcgfhdgigpgmgehihadpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaaahihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaabhdhcaacegphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcelgfhjaaaaaaaaaaaaaaabacaaacekaaacgjgeemaaafhdgdgphagfheaabcemgkgbhggbcpgmgbgoghcpfdhehcgjgoghdlhihaaaaaaaaaaaaaaagiheaaafhdgdgphagfhdhcaafjgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccogfhihahcgfhdhdgjgpgoephagfhcgbhegphchdcofaepfahcgpgkgfgdheaaaaaaaaaaaaaaabacaaagfkaaaogjhdfahcgpgkgfgdhefegpefgogefkaaakgphggfhcgmgpgbgegfgefkaabfhahcgpgdgfhdhdgjgoghecgbghepggfehfhagmgfhdfkaabehcgfhdhfgmhefdgjgoghgmgffehfhagmgfecgbghejaaaihdhegbhcheedgpgmemaaahgdgpgmhfgngohdheaabfemgkgbhggbcphfhegjgmcpebhchcgbhjemgjhdhed
lhihcaagcgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccogfhihahcgfhdhdgjgpgoephagfhcgbhegphchdcoefhihahcgfhdhdgjgpgoephagfhcgbhegphcaaaaaaaaaaaaaaabacaaaahihcaaemgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccofagihjhdgjgdgbgmephagfhcgbhegphcaaaaaaaaaaaaaaabacaaamfkaaafgbgdgdhfgnfkaaangjgohahfheebhehegbgdgigfgeejaabehcgfhbhfgfhdhegfgefagbhcgbgmgmgfgmgjhdgnecaaakhcgfhdhfgmhefehjhagfemaaafgbgmgjgbhdhbaahoaaaoemaaafgjgohahfheheaablemgphcghcpgbhagbgdgigfcphagjghcpgegbhegbcpfehfhagmgfdlemaaaggjgohahfhehdhbaahoaaagemaaangmgjgogfgbghgffehcgbgdgfhcheaachemgphcghcpgbhagbgdgigfcphagjghcphagfgocphfhegjgmcpemgjgogfgbghgffehcgbgdgfhcdlemaabbgphcgjghgjgogbgmemgpgdgbhegjgpgohdhbaahoaaagemaaahgphfhehahfhehdhbaahoaaagemaaakhagbhcgfgohefagmgbgoheaafaemgphcghcpgbhagbgdgigfcphagjghcpgcgbgdglgfgogecpgigbgegpgphacpgfhigfgdhfhegjgpgogfgoghgjgogfcphagihjhd
gjgdgbgmemgbhjgfhccphagmgbgohdcpfagihjhdgjgdgbgmfagmgbgodlemaaadhcgfhdheaaeeemgphcghcpgbhagbgdgigfcphagjghcpgcgbgdglgfgogecpgigbgegpgphacpgfhigfgdhfhegjgpgogfgoghgjgogfcphagihjhdgjgdgbgmemgbhjgfhccpfcgfhdhfgmhedlhihcaacbgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcaaaaaaaaaaaaaaabacaaabemaaaegnelgfhjheaacgemgphcghcpgbhagbgdgigfcphagjghcpgjgnhagmcphagmgbgocpephagfhcgbhegphcelgfhjdlhihahbaahoaaapaaaappppppppdchahahahahdhbaahoaaaaaaaaaaaahhaeaaaaaaakhihahahdhcaaecgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccofcgfhdhfgmheaaaaaaaaaaaaaaabacaaacecaaamhcgfhehfhcgofdhegbhehfhdemaaaghcgfhdhfgmheheaabcemgkgbhggbcpgmgbgoghcpepgcgkgfgdhedlhihaachaaaaaaaaaaaaaaaaahdhbaahoaaaaaaaaaaabhhaeaaaaaaakhdhcaabbgkgbhggbcogmgbgoghcoejgohegfghgfhcbcockakephibihdiacaaabejaaafhggbgmhfgfhihcaabagkgbhggbcogmgbgoghcoeohfgngcgfhcigkmjfbnaljeoailacaaaahihaaaaaaaaahihihdhbaahoaaaaaaaaaaabhhaeaaaaaaakhbaahoaablhih
dhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaabhbaahoaablhbaahoaaaphihdhbaahoaaaaaaaaaaaahhaeaaaaaaakhihdhbaahoaaaihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaaahihdhbaahoaaaihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaaahihdhbaahoaaaihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaaahiaahahi','','b','scope','true')) - scope-102
\ No newline at end of file