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 2013/11/25 09:23:05 UTC
svn commit: r1545171 - in /pig/branches/tez:
src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/
src/org/apache/pig/backend/hadoop/executionengine/tez/
src/org/apache/pig/impl/plan/ src/org/apache/pig/tools/pigstats/tez/
test/e2e/pig/tests/
Author: daijy
Date: Mon Nov 25 08:23:04 2013
New Revision: 1545171
URL: http://svn.apache.org/r1545171
Log:
PIG-3534: Make order by work
Modified:
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/PigProcessor.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezCompiler.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezDagBuilder.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezJobControlCompiler.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezOperPlan.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezOperator.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezPlanContainer.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezPlanContainerNode.java
pig/branches/tez/src/org/apache/pig/impl/plan/OperatorPlan.java
pig/branches/tez/src/org/apache/pig/tools/pigstats/tez/TezStats.java
pig/branches/tez/test/e2e/pig/tests/tez.conf
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java?rev=1545171&r1=1545170&r2=1545171&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java Mon Nov 25 08:23:04 2013
@@ -740,8 +740,6 @@ public class JobControlCompiler{
}
if(mro.isGlobalSort() || mro.isLimitAfterSort()){
- // Only set the quantiles file and sort partitioner if we're a
- // global sort, not for limit after sort.
if (mro.isGlobalSort()) {
String symlink = addSingleFileToDistributedCache(
pigContext, conf, mro.getQuantFile(), "pigsample");
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/PigProcessor.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/PigProcessor.java?rev=1545171&r1=1545170&r2=1545171&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/PigProcessor.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/PigProcessor.java Mon Nov 25 08:23:04 2013
@@ -30,11 +30,13 @@ import org.apache.hadoop.conf.Configurat
import org.apache.pig.PigException;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.HDataType;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
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.PhysicalPlan;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLocalRearrange;
+import org.apache.pig.backend.hadoop.executionengine.shims.HadoopShims;
import org.apache.pig.data.SchemaTupleBackend;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.PigContext;
@@ -87,6 +89,7 @@ public class PigProcessor implements Log
String execPlanString = conf.get(PLAN);
execPlan = (PhysicalPlan) ObjectSerializer.deserialize(execPlanString);
SchemaTupleBackend.initialize(conf, pc);
+ PigMapReduce.sJobContext = HadoopShims.createJobContext(conf, new org.apache.hadoop.mapreduce.JobID());
}
@Override
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezCompiler.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezCompiler.java?rev=1545171&r1=1545170&r2=1545171&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezCompiler.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezCompiler.java Mon Nov 25 08:23:04 2013
@@ -18,6 +18,7 @@
package org.apache.pig.backend.hadoop.executionengine.tez;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
@@ -27,12 +28,15 @@ import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.pig.FuncSpec;
import org.apache.pig.PigException;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.plans.ScalarPhyFinder;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.plans.UDFFinder;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.ConstantExpression;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POProject;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserFunc;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POCollectedGroup;
@@ -50,6 +54,7 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POMergeJoin;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PONative;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPackage;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPackageLite;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPackage.PackageType;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PORank;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POSkewedJoin;
@@ -61,6 +66,8 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.util.PlanHelper;
import org.apache.pig.data.DataType;
import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.builtin.FindQuantiles;
+import org.apache.pig.impl.builtin.RandomSampleLoader;
import org.apache.pig.impl.io.FileLocalizer;
import org.apache.pig.impl.io.FileSpec;
import org.apache.pig.impl.plan.DepthFirstWalker;
@@ -70,6 +77,8 @@ import org.apache.pig.impl.plan.Operator
import org.apache.pig.impl.plan.OperatorPlan;
import org.apache.pig.impl.plan.PlanException;
import org.apache.pig.impl.plan.VisitorException;
+import org.apache.pig.impl.util.Pair;
+import org.apache.pig.impl.util.Utils;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@@ -151,7 +160,7 @@ public class TezCompiler extends PhyPlan
public TezOperPlan getTezPlan() {
return tezPlan;
}
-
+
// Segment a single DAG into a DAG graph
public TezPlanContainer getPlanContainer() throws PlanException {
TezPlanContainer tezPlanContainer = new TezPlanContainer(pigContext);
@@ -317,7 +326,7 @@ public class TezCompiler extends PhyPlan
ld.setLFile(fSpec);
ret.plan.add(ld);
tezPlan.add(ret);
- tezPlan.connect(old, ret);
+ connect(tezPlan, old, ret);
return ret;
}
@@ -336,16 +345,20 @@ public class TezCompiler extends PhyPlan
tezOp.plan.addAsLeaf(op);
curTezOp = tezOp;
}
+
+ private void connect(TezOperPlan plan, TezOperator from, TezOperator to) throws PlanException {
+ plan.connect(from, to);
+ // Add edge descriptors to old and new operators
+ to.inEdges.put(from.getOperatorKey(), new TezEdgeDescriptor());
+ from.outEdges.put(to.getOperatorKey(), new TezEdgeDescriptor());
+ }
private void blocking() throws IOException, PlanException {
TezOperator newTezOp = getTezOp();
tezPlan.add(newTezOp);
for (TezOperator tezOp : compiledInputs) {
tezOp.setClosed(true);
- tezPlan.connect(tezOp, newTezOp);
- // Add edge descriptors to old and new operators
- newTezOp.inEdges.put(tezOp.getOperatorKey(), new TezEdgeDescriptor());
- tezOp.outEdges.put(newTezOp.getOperatorKey(), new TezEdgeDescriptor());
+ connect(tezPlan, tezOp, newTezOp);
}
curTezOp = newTezOp;
}
@@ -694,11 +707,593 @@ public class TezCompiler extends PhyPlan
throw new TezCompilerException(msg, errCode, PigException.BUG);
}
+ /**
+ * Returns a temporary DFS Path
+ * @return
+ * @throws IOException
+ */
+ private FileSpec getTempFileSpec() throws IOException {
+ return new FileSpec(FileLocalizer.getTemporaryPath(pigContext).toString(),
+ new FuncSpec(Utils.getTmpFileCompressorName(pigContext)));
+ }
+
+ private POStore getStore(){
+ POStore st = new POStore(new OperatorKey(scope,nig.getNextNodeId(scope)));
+ // mark store as tmp store. These could be removed by the
+ // optimizer, because it wasn't the user requesting it.
+ st.setIsTmpStore(true);
+ return st;
+ }
+
+ /**
+ * Force an end to the current vertex with a store into a temporary
+ * file.
+ * @param fSpec Temp file to force a store into.
+ * @return Tez operator that now is finished with a store.
+ * @throws PlanException
+ */
+ private TezOperator endSingleInputPlanWithStr(FileSpec fSpec) throws PlanException{
+ if(compiledInputs.length>1) {
+ int errCode = 2023;
+ String msg = "Received a multi input plan when expecting only a single input one.";
+ throw new PlanException(msg, errCode, PigException.BUG);
+ }
+ TezOperator oper = compiledInputs[0];
+ if (!oper.isClosed()) {
+ POStore str = getStore();
+ str.setSFile(fSpec);
+ oper.plan.addAsLeaf(str);
+ } else {
+ int errCode = 2022;
+ String msg = "Both map and reduce phases have been done. This is unexpected while compiling.";
+ throw new PlanException(msg, errCode, PigException.BUG);
+ }
+ return oper;
+ }
+
+ private Pair<TezOperator[],Integer> getQuantileJobs(
+ POSort inpSort,
+ TezOperator prevJob,
+ FileSpec lFile,
+ FileSpec quantFile,
+ int rp) throws PlanException, VisitorException {
+
+ POSort sort = new POSort(inpSort.getOperatorKey(), inpSort
+ .getRequestedParallelism(), null, inpSort.getSortPlans(),
+ inpSort.getMAscCols(), inpSort.getMSortFunc());
+ 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.
+ List<Boolean> ascCols = inpSort.getMAscCols();
+ String[] ascs = new String[ascCols.size()];
+ for (int i = 0; i < ascCols.size(); i++) ascs[i] = ascCols.get(i).toString();
+ // check if user defined comparator is used in the sort, if so
+ // prepend the name of the comparator as the first fields in the
+ // constructor args array to the FindQuantiles udf
+ String[] ctorArgs = ascs;
+ if(sort.isUDFComparatorUsed) {
+ String userComparatorFuncSpec = sort.getMSortFunc().getFuncSpec().toString();
+ ctorArgs = new String[ascs.length + 1];
+ ctorArgs[0] = USER_COMPARATOR_MARKER + userComparatorFuncSpec;
+ for(int j = 0; j < ascs.length; j++) {
+ ctorArgs[j+1] = ascs[j];
+ }
+ }
+
+ return getSamplingJobs(sort, prevJob, null, lFile, quantFile, rp, null, FindQuantiles.class.getName(), ctorArgs, RandomSampleLoader.class.getName());
+ }
+
+ /**
+ * Create a sampling job to collect statistics by sampling an input file. The sequence of operations is as
+ * following:
+ * <li>Transform input sample tuples into another tuple.</li>
+ * <li>Add an extra field "all" into the tuple </li>
+ * <li>Package all tuples into one bag </li>
+ * <li>Add constant field for number of reducers. </li>
+ * <li>Sorting the bag </li>
+ * <li>Invoke UDF with the number of reducers and the sorted bag.</li>
+ * <li>Data generated by UDF is stored into a file.</li>
+ *
+ * @param sort the POSort operator used to sort the bag
+ * @param prevJob previous job of current sampling job
+ * @param transformPlans PhysicalPlans to transform input samples
+ * @param lFile path of input file
+ * @param sampleFile path of output file
+ * @param rp configured parallemism
+ * @param sortKeyPlans PhysicalPlans to be set into POSort operator to get sorting keys
+ * @param udfClassName the class name of UDF
+ * @param udfArgs the arguments of UDF
+ * @param sampleLdrClassName class name for the sample loader
+ * @return pair<tezoperator[],integer>
+ * @throws PlanException
+ * @throws VisitorException
+ */
+ @SuppressWarnings("deprecation")
+ private Pair<TezOperator[],Integer> getSamplingJobs(POSort sort, TezOperator prevJob, List<PhysicalPlan> transformPlans,
+ FileSpec lFile, FileSpec sampleFile, int rp, List<PhysicalPlan> sortKeyPlans,
+ String udfClassName, String[] udfArgs, String sampleLdrClassName ) throws PlanException, VisitorException {
+
+ String[] rslargs = new String[2];
+ // SampleLoader expects string version of FuncSpec
+ // as its first constructor argument.
+
+ rslargs[0] = (new FuncSpec(Utils.getTmpFileCompressorName(pigContext))).toString();
+
+ rslargs[1] = "100"; // The value is calculated based on the file size for skewed join
+ FileSpec quantLdFilName = new FileSpec(lFile.getFileName(),
+ new FuncSpec(sampleLdrClassName, rslargs));
+
+ TezOperator[] opers = new TezOperator[2];
+
+ TezOperator oper1 = startNew(quantLdFilName, prevJob);
+ opers[0] = oper1;
+
+ // TODO: Review sort udf
+// if(sort.isUDFComparatorUsed) {
+// mro.UDFs.add(sort.getMSortFunc().getFuncSpec().toString());
+// curMROp.isUDFComparatorUsed = true;
+// }
+
+ List<Boolean> flat1 = new ArrayList<Boolean>();
+ List<PhysicalPlan> eps1 = new ArrayList<PhysicalPlan>();
+
+ // if transform plans are not specified, project the columns of sorting keys
+ if (transformPlans == null) {
+ Pair<POProject, Byte>[] sortProjs = null;
+ try{
+ sortProjs = getSortCols(sort.getSortPlans());
+ }catch(Exception e) {
+ throw new RuntimeException(e);
+ }
+ // Set up the projections of the key columns
+ if (sortProjs == null) {
+ PhysicalPlan ep = new PhysicalPlan();
+ POProject prj = new POProject(new OperatorKey(scope,
+ nig.getNextNodeId(scope)));
+ prj.setStar(true);
+ prj.setOverloaded(false);
+ prj.setResultType(DataType.TUPLE);
+ ep.add(prj);
+ eps1.add(ep);
+ flat1.add(false);
+ } else {
+ for (Pair<POProject, Byte> sortProj : sortProjs) {
+ // Check for proj being null, null is used by getSortCols for a non POProject
+ // operator. Since Order by does not allow expression operators,
+ //it should never be set to null
+ if(sortProj == null){
+ int errCode = 2174;
+ String msg = "Internal exception. Could not create a sampler job";
+ throw new TezCompilerException(msg, errCode, PigException.BUG);
+ }
+ PhysicalPlan ep = new PhysicalPlan();
+ POProject prj;
+ try {
+ prj = sortProj.first.clone();
+ } catch (CloneNotSupportedException e) {
+ //should not get here
+ throw new AssertionError(
+ "Error cloning project caught exception" + e
+ );
+ }
+ ep.add(prj);
+ eps1.add(ep);
+ flat1.add(false);
+ }
+ }
+ }else{
+ for(int i=0; i<transformPlans.size(); i++) {
+ eps1.add(transformPlans.get(i));
+ flat1.add(true);
+ }
+ }
+
+ // This foreach will pick the sort key columns from the RandomSampleLoader output
+ POForEach nfe1 = new POForEach(new OperatorKey(scope,nig.getNextNodeId(scope)),-1,eps1,flat1);
+ oper1.plan.addAsLeaf(nfe1);
+
+ // Now set up a POLocalRearrange which has "all" as the key and the output of the
+ // foreach will be the "value" out of POLocalRearrange
+ PhysicalPlan ep1 = new PhysicalPlan();
+ ConstantExpression ce = new ConstantExpression(new OperatorKey(scope,nig.getNextNodeId(scope)));
+ ce.setValue("all");
+ ce.setResultType(DataType.CHARARRAY);
+ ep1.add(ce);
+
+ List<PhysicalPlan> eps = new ArrayList<PhysicalPlan>();
+ eps.add(ep1);
+
+ POLocalRearrange lr = new POLocalRearrange(new OperatorKey(scope,nig.getNextNodeId(scope)));
+ try {
+ lr.setIndex(0);
+ } catch (ExecException e) {
+ int errCode = 2058;
+ String msg = "Unable to set index on newly created POLocalRearrange.";
+ throw new PlanException(msg, errCode, PigException.BUG, e);
+ }
+ lr.setKeyType(DataType.CHARARRAY);
+ lr.setPlans(eps);
+ lr.setResultType(DataType.TUPLE);
+ lr.addOriginalLocation(sort.getAlias(), sort.getOriginalLocations());
+ oper1.plan.add(lr);
+ oper1.plan.connect(nfe1, lr);
+
+ oper1.setClosed(true);
+
+ TezOperator oper2 = getTezOp();
+ opers[1] = oper2;
+ tezPlan.add(oper2);
+ connect(tezPlan, oper1, oper2);
+
+ POPackage pkg = new POPackage(new OperatorKey(scope,nig.getNextNodeId(scope)));
+ pkg.setKeyType(DataType.CHARARRAY);
+ pkg.setNumInps(1);
+ boolean[] inner = {false};
+ pkg.setInner(inner);
+ oper2.plan.add(pkg);
+
+ // Lets start building the plan which will have the sort
+ // for the foreach
+ PhysicalPlan fe2Plan = new PhysicalPlan();
+ // Top level project which just projects the tuple which is coming
+ // from the foreach after the package
+ POProject topPrj = new POProject(new OperatorKey(scope,nig.getNextNodeId(scope)));
+ topPrj.setColumn(1);
+ topPrj.setResultType(DataType.BAG);
+ topPrj.setOverloaded(true);
+ fe2Plan.add(topPrj);
+
+ // the projections which will form sort plans
+ List<PhysicalPlan> nesSortPlanLst = new ArrayList<PhysicalPlan>();
+ if (sortKeyPlans != null) {
+ for(int i=0; i<sortKeyPlans.size(); i++) {
+ nesSortPlanLst.add(sortKeyPlans.get(i));
+ }
+ }else{
+ Pair<POProject, Byte>[] sortProjs = null;
+ try{
+ sortProjs = getSortCols(sort.getSortPlans());
+ }catch(Exception e) {
+ throw new RuntimeException(e);
+ }
+ // Set up the projections of the key columns
+ if (sortProjs == null) {
+ PhysicalPlan ep = new PhysicalPlan();
+ POProject prj = new POProject(new OperatorKey(scope,
+ nig.getNextNodeId(scope)));
+ prj.setStar(true);
+ prj.setOverloaded(false);
+ prj.setResultType(DataType.TUPLE);
+ ep.add(prj);
+ nesSortPlanLst.add(ep);
+ } else {
+ for (int i=0; i<sortProjs.length; i++) {
+ POProject prj =
+ new POProject(new OperatorKey(scope,nig.getNextNodeId(scope)));
+
+ prj.setResultType(sortProjs[i].second);
+ if(sortProjs[i].first != null && sortProjs[i].first.isProjectToEnd()){
+ if(i != sortProjs.length -1){
+ //project to end has to be the last sort column
+ throw new AssertionError("Project-range to end (x..)" +
+ " is supported in order-by only as last sort column");
+ }
+ prj.setProjectToEnd(i);
+ break;
+ }
+ else{
+ prj.setColumn(i);
+ }
+ prj.setOverloaded(false);
+
+ PhysicalPlan ep = new PhysicalPlan();
+ ep.add(prj);
+ nesSortPlanLst.add(ep);
+ }
+ }
+ }
+
+ sort.setSortPlans(nesSortPlanLst);
+ sort.setResultType(DataType.BAG);
+ fe2Plan.add(sort);
+ fe2Plan.connect(topPrj, sort);
+
+ // The plan which will have a constant representing the
+ // degree of parallelism for the final order by map-reduce job
+ // this will either come from a "order by parallel x" in the script
+ // or will be the default number of reducers for the cluster if
+ // "parallel x" is not used in the script
+ PhysicalPlan rpep = new PhysicalPlan();
+ ConstantExpression rpce = new ConstantExpression(new OperatorKey(scope,nig.getNextNodeId(scope)));
+ rpce.setRequestedParallelism(rp);
+
+ // We temporarily set it to rp and will adjust it at runtime, because the final degree of parallelism
+ // is unknown until we are ready to submit it. See PIG-2779.
+ rpce.setValue(rp);
+
+ rpce.setResultType(DataType.INTEGER);
+ rpep.add(rpce);
+
+ List<PhysicalPlan> genEps = new ArrayList<PhysicalPlan>();
+ genEps.add(rpep);
+ genEps.add(fe2Plan);
+
+ List<Boolean> flattened2 = new ArrayList<Boolean>();
+ flattened2.add(false);
+ flattened2.add(false);
+
+ POForEach nfe2 = new POForEach(new OperatorKey(scope,nig.getNextNodeId(scope)),-1, genEps, flattened2);
+ oper2.plan.add(nfe2);
+ oper2.plan.connect(pkg, nfe2);
+
+ // Let's connect the output from the foreach containing
+ // number of quantiles and the sorted bag of samples to
+ // another foreach with the FindQuantiles udf. The input
+ // to the FindQuantiles udf is a project(*) which takes the
+ // foreach input and gives it to the udf
+ PhysicalPlan ep4 = new PhysicalPlan();
+ POProject prjStar4 = new POProject(new OperatorKey(scope,nig.getNextNodeId(scope)));
+ prjStar4.setResultType(DataType.TUPLE);
+ prjStar4.setStar(true);
+ ep4.add(prjStar4);
+
+ List<PhysicalOperator> ufInps = new ArrayList<PhysicalOperator>();
+ ufInps.add(prjStar4);
+
+ POUserFunc uf = new POUserFunc(new OperatorKey(scope,nig.getNextNodeId(scope)), -1, ufInps,
+ new FuncSpec(udfClassName, udfArgs));
+ ep4.add(uf);
+ ep4.connect(prjStar4, uf);
+
+ List<PhysicalPlan> ep4s = new ArrayList<PhysicalPlan>();
+ ep4s.add(ep4);
+ List<Boolean> flattened3 = new ArrayList<Boolean>();
+ flattened3.add(false);
+ POForEach nfe3 = new POForEach(new OperatorKey(scope,nig.getNextNodeId(scope)), -1, ep4s, flattened3);
+
+ oper2.plan.add(nfe3);
+ oper2.plan.connect(nfe2, nfe3);
+
+ POStore str = getStore();
+ str.setSFile(sampleFile);
+
+ oper2.plan.add(str);
+ oper2.plan.connect(nfe3, str);
+
+ oper2.setClosed(true);
+ oper2.requestedParallelism = 1;
+ // oper2.markSampler();
+ return new Pair<TezOperator[], Integer>(opers, rp);
+ }
+
+ private static class FindKeyTypeVisitor extends PhyPlanVisitor {
+
+ byte keyType = DataType.UNKNOWN;
+
+ FindKeyTypeVisitor(PhysicalPlan plan) {
+ super(plan,
+ new DepthFirstWalker<PhysicalOperator, PhysicalPlan>(plan));
+ }
+
+ @Override
+ public void visitProject(POProject p) throws VisitorException {
+ keyType = p.getResultType();
+ }
+ }
+
+ private Pair<POProject,Byte> [] getSortCols(List<PhysicalPlan> plans) throws PlanException, ExecException {
+ if(plans!=null){
+ @SuppressWarnings("unchecked")
+ Pair<POProject,Byte>[] ret = new Pair[plans.size()];
+ int i=-1;
+ for (PhysicalPlan plan : plans) {
+ PhysicalOperator op = plan.getLeaves().get(0);
+ POProject proj;
+ if (op instanceof POProject) {
+ if (((POProject)op).isStar()) return null;
+ proj = (POProject)op;
+ } else {
+ proj = null;
+ }
+ byte type = op.getResultType();
+ ret[++i] = new Pair<POProject, Byte>(proj, type);
+ }
+ return ret;
+ }
+ int errCode = 2026;
+ String msg = "No expression plan found in POSort.";
+ throw new PlanException(msg, errCode, PigException.BUG);
+ }
+
+ private TezOperator[] getSortJobs(
+ POSort sort,
+ TezOperator quantJob,
+ FileSpec lFile,
+ FileSpec quantFile,
+ int rp,
+ Pair<POProject, Byte>[] fields) throws PlanException{
+ TezOperator[] opers = new TezOperator[2];
+ TezOperator oper1 = startNew(lFile, quantJob);
+ tezPlan.add(oper1);
+ opers[0] = oper1;
+ oper1.setQuantFile(quantFile.getFileName());
+ oper1.setGlobalSort(true);
+ oper1.requestedParallelism = rp;
+
+ long limit = sort.getLimit();
+ oper1.limit = limit;
+
+ List<PhysicalPlan> eps1 = new ArrayList<PhysicalPlan>();
+
+ byte keyType = DataType.UNKNOWN;
+
+ boolean[] sortOrder;
+
+ List<Boolean> sortOrderList = sort.getMAscCols();
+ if(sortOrderList != null) {
+ sortOrder = new boolean[sortOrderList.size()];
+ for(int i = 0; i < sortOrderList.size(); ++i) {
+ sortOrder[i] = sortOrderList.get(i);
+ }
+ oper1.setSortOrder(sortOrder);
+ }
+
+ if (fields == null) {
+ // This is project *
+ PhysicalPlan ep = new PhysicalPlan();
+ POProject prj = new POProject(new OperatorKey(scope,nig.getNextNodeId(scope)));
+ prj.setStar(true);
+ prj.setOverloaded(false);
+ prj.setResultType(DataType.TUPLE);
+ ep.add(prj);
+ eps1.add(ep);
+ } else {
+ // Attach the sort plans to the local rearrange to get the
+ // projection.
+ eps1.addAll(sort.getSortPlans());
+
+ // Visit the first sort plan to figure out our key type. We only
+ // have to visit the first because if we have more than one plan,
+ // then the key type will be tuple.
+ try {
+ FindKeyTypeVisitor fktv =
+ new FindKeyTypeVisitor(sort.getSortPlans().get(0));
+ fktv.visit();
+ keyType = fktv.keyType;
+ } catch (VisitorException ve) {
+ int errCode = 2035;
+ String msg = "Internal error. Could not compute key type of sort operator.";
+ throw new PlanException(msg, errCode, PigException.BUG, ve);
+ }
+ }
+
+ POLocalRearrange lr = new POLocalRearrange(new OperatorKey(scope,nig.getNextNodeId(scope)));
+ try {
+ lr.setIndex(0);
+ } catch (ExecException e) {
+ int errCode = 2058;
+ String msg = "Unable to set index on newly created POLocalRearrange.";
+ throw new PlanException(msg, errCode, PigException.BUG, e);
+ }
+ lr.setKeyType((fields == null || fields.length>1) ? DataType.TUPLE :
+ keyType);
+ lr.setPlans(eps1);
+ lr.setResultType(DataType.TUPLE);
+ lr.addOriginalLocation(sort.getAlias(), sort.getOriginalLocations());
+ oper1.plan.addAsLeaf(lr);
+
+ oper1.setClosed(true);
+
+ TezOperator oper2 = getTezOp();
+ opers[1] = oper2;
+ tezPlan.add(oper2);
+ connect(tezPlan, oper1, oper2);
+
+ if (limit!=-1) {
+ POPackageLite pkg_c = new POPackageLite(new OperatorKey(scope,nig.getNextNodeId(scope)));
+ pkg_c.setKeyType((fields.length>1) ? DataType.TUPLE : keyType);
+ pkg_c.setNumInps(1);
+ oper2.inEdges.put(oper1.getOperatorKey(), new TezEdgeDescriptor());
+ PhysicalPlan combinePlan = oper2.inEdges.get(oper1.getOperatorKey()).combinePlan;
+
+ combinePlan.add(pkg_c);
+
+ List<PhysicalPlan> eps_c1 = new ArrayList<PhysicalPlan>();
+ List<Boolean> flat_c1 = new ArrayList<Boolean>();
+ PhysicalPlan ep_c1 = new PhysicalPlan();
+ POProject prj_c1 = new POProject(new OperatorKey(scope,nig.getNextNodeId(scope)));
+ prj_c1.setColumn(1);
+ prj_c1.setOverloaded(false);
+ prj_c1.setResultType(DataType.BAG);
+ ep_c1.add(prj_c1);
+ eps_c1.add(ep_c1);
+ flat_c1.add(true);
+ POForEach fe_c1 = new POForEach(new OperatorKey(scope,nig.getNextNodeId(scope)),
+ -1, eps_c1, flat_c1);
+ fe_c1.setResultType(DataType.TUPLE);
+
+ combinePlan.addAsLeaf(fe_c1);
+
+ POLimit pLimit = new POLimit(new OperatorKey(scope,nig.getNextNodeId(scope)));
+ pLimit.setLimit(limit);
+ combinePlan.addAsLeaf(pLimit);
+
+ List<PhysicalPlan> eps_c2 = new ArrayList<PhysicalPlan>();
+ eps_c2.addAll(sort.getSortPlans());
+
+ POLocalRearrange lr_c2 = new POLocalRearrange(new OperatorKey(scope,nig.getNextNodeId(scope)));
+ try {
+ lr_c2.setIndex(0);
+ } catch (ExecException e) {
+ int errCode = 2058;
+ String msg = "Unable to set index on newly created POLocalRearrange.";
+ throw new PlanException(msg, errCode, PigException.BUG, e);
+ }
+ lr_c2.setKeyType((fields.length>1) ? DataType.TUPLE : keyType);
+ lr_c2.setPlans(eps_c2);
+ lr_c2.setResultType(DataType.TUPLE);
+ combinePlan.addAsLeaf(lr_c2);
+ }
+
+ POPackageLite pkg = new POPackageLite(new OperatorKey(scope,nig.getNextNodeId(scope)));
+ pkg.setKeyType((fields == null || fields.length>1) ? DataType.TUPLE :
+ keyType);
+ pkg.setNumInps(1);
+ oper2.plan.add(pkg);
+
+ PhysicalPlan ep = new PhysicalPlan();
+ POProject prj = new POProject(new OperatorKey(scope,nig.getNextNodeId(scope)));
+ prj.setColumn(1);
+ prj.setOverloaded(false);
+ prj.setResultType(DataType.BAG);
+ ep.add(prj);
+ List<PhysicalPlan> eps2 = new ArrayList<PhysicalPlan>();
+ eps2.add(ep);
+ List<Boolean> flattened = new ArrayList<Boolean>();
+ flattened.add(true);
+ POForEach nfe1 = new POForEach(new OperatorKey(scope,nig.getNextNodeId(scope)),-1,eps2,flattened);
+ oper2.plan.add(nfe1);
+ oper2.plan.connect(pkg, nfe1);
+ if (limit!=-1)
+ {
+ POLimit pLimit2 = new POLimit(new OperatorKey(scope,nig.getNextNodeId(scope)));
+ pLimit2.setLimit(limit);
+ oper2.plan.addAsLeaf(pLimit2);
+ }
+
+ return opers;
+ }
+
@Override
public void visitSort(POSort op) throws VisitorException {
- int errCode = 2034;
- String msg = "Cannot compile " + op.getClass().getSimpleName();
- throw new TezCompilerException(msg, errCode, PigException.BUG);
+ try{
+ FileSpec fSpec = getTempFileSpec();
+ TezOperator oper = endSingleInputPlanWithStr(fSpec);
+ oper.segmentBelow = true;
+ FileSpec quantFile = getTempFileSpec();
+ int rp = op.getRequestedParallelism();
+ Pair<POProject, Byte>[] fields = getSortCols(op.getSortPlans());
+ Pair<TezOperator[], Integer> quantJobParallelismPair =
+ getQuantileJobs(op, oper, fSpec, quantFile, rp);
+ TezOperator[] opers = getSortJobs(op, quantJobParallelismPair.first[1], fSpec, quantFile,
+ quantJobParallelismPair.second, fields);
+
+ quantJobParallelismPair.first[1].segmentBelow = true;
+
+ curTezOp = opers[1];
+
+ // TODO: Review sort udf
+// if(op.isUDFComparatorUsed){
+// curTezOp.UDFs.add(op.getMSortFunc().getFuncSpec().toString());
+// curTezOp.isUDFComparatorUsed = true;
+// }
+ phyToTezOpMap.put(op, curTezOp);
+ }catch(Exception e){
+ int errCode = 2034;
+ String msg = "Error compiling operator " + op.getClass().getSimpleName();
+ throw new TezCompilerException(msg, errCode, PigException.BUG, e);
+ }
}
@Override
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezDagBuilder.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezDagBuilder.java?rev=1545171&r1=1545170&r2=1545171&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezDagBuilder.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezDagBuilder.java Mon Nov 25 08:23:04 2013
@@ -28,14 +28,20 @@ import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.RawComparator;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.api.records.LocalResourceType;
+import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.pig.LoadFunc;
import org.apache.pig.PigException;
import org.apache.pig.StoreFuncInterface;
@@ -65,6 +71,7 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigIntWritableComparator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigLongWritableComparator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigTupleWritableComparator;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.partitioners.WeightedRangePartitioner;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigCombiner;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigInputFormat;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigOutputFormat;
@@ -284,6 +291,7 @@ public class TezDagBuilder extends TezOp
if (leaves.size() == 1 && leaves.get(0) instanceof POLocalRearrange) {
byte keyType = ((POLocalRearrange)leaves.get(0)).getKeyType();
setIntermediateOutputKeyValue(keyType, conf);
+ conf.set("pig.reduce.key.type", Byte.toString(keyType));
}
// Configure the classes for incoming shuffles to this TezOp
@@ -296,11 +304,32 @@ public class TezDagBuilder extends TezOp
conf.set("pig.reduce.key.type", Byte.toString(keyType));
setIntermediateInputKeyValue(keyType, conf);
conf.setClass("pig.input.handler.class", ShuffledInputHandler.class, InputHandler.class);
+ conf.set("pig.reduce.key.type", Byte.toString(keyType));
} else {
conf.setClass("pig.input.handler.class", FileInputHandler.class, InputHandler.class);
}
conf.setClass("mapreduce.outputformat.class", PigOutputFormat.class, OutputFormat.class);
+
+ if(tezOp.isGlobalSort() || tezOp.isLimitAfterSort()){
+ if (tezOp.isGlobalSort()) {
+ FileSystem fs = FileSystem.get(conf);
+ Path quantFilePath = new Path(tezOp.getQuantFile() + "/part-r-00000");
+ FileStatus fstat = fs.getFileStatus(quantFilePath);
+ LocalResource quantFileResource = LocalResource.newInstance(
+ ConverterUtils.getYarnUrlFromPath(fstat.getPath()),
+ LocalResourceType.FILE,
+ LocalResourceVisibility.APPLICATION,
+ fstat.getLen(),
+ fstat.getModificationTime());
+ localResources.put(quantFilePath.getName(), quantFileResource);
+ conf.set("pig.quantilesFile", fstat.getPath().toString());
+ conf.set("pig.sortOrder",
+ ObjectSerializer.serialize(tezOp.getSortOrder()));
+ conf.setClass("mapreduce.job.partitioner.class", WeightedRangePartitioner.class,
+ Partitioner.class);
+ }
+ }
// Serialize the execution plan
conf.set(PigProcessor.PLAN, ObjectSerializer.serialize(tezOp.plan));
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezJobControlCompiler.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezJobControlCompiler.java?rev=1545171&r1=1545170&r2=1545171&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezJobControlCompiler.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezJobControlCompiler.java Mon Nov 25 08:23:04 2013
@@ -61,7 +61,7 @@ public class TezJobControlCompiler {
Configuration conf, TezPlanContainer planContainer)
throws JobCreationException {
int timeToSleep;
- String defaultPigJobControlSleep = pigContext.getExecType().isLocal() ? "100" : "5000";
+ String defaultPigJobControlSleep = pigContext.getExecType().isLocal() ? "100" : "1000";
String pigJobControlSleep = tezConf.get("pig.jobcontrol.sleep", defaultPigJobControlSleep);
if (!pigJobControlSleep.equals(defaultPigJobControlSleep)) {
log.info("overriding default JobControl sleep (" +
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezOperPlan.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezOperPlan.java?rev=1545171&r1=1545170&r2=1545171&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezOperPlan.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezOperPlan.java Mon Nov 25 08:23:04 2013
@@ -50,7 +50,7 @@ public class TezOperPlan extends Operato
try {
printer.visit();
} catch (VisitorException e) {
- // TODO Auto-generated catch block
+ throw new RuntimeException(e);
}
return baos.toString();
}
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezOperator.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezOperator.java?rev=1545171&r1=1545170&r2=1545171&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezOperator.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezOperator.java Mon Nov 25 08:23:04 2013
@@ -65,7 +65,26 @@ public class TezOperator extends Operato
// Indicates that the plan creation is complete
boolean closed = false;
+ // Indicate whether we need to split the DAG below the operator
+ // The result is two or more DAG connected DAG inside the same plan container
boolean segmentBelow = false;
+
+ // Indicates if this is a limit after a sort
+ boolean limitAfterSort = false;
+
+ //Indicates if this job is an order by job
+ boolean globalSort = false;
+
+ //The quantiles file name if globalSort is true
+ String quantFile;
+
+ //The sort order of the columns;
+ //asc is true and desc is false
+ boolean[] sortOrder;
+
+ // Last POLimit value in this map reduce operator, needed by LimitAdjuster
+ // to add additional map reduce operator with 1 reducer after this
+ long limit = -1;
// Types of blocking operators. For now, we only support the following ones.
private static enum OPER_FEATURE {
@@ -187,5 +206,37 @@ public class TezOperator extends Operato
public boolean needSegmentBelow() {
return segmentBelow;
}
+
+ public String getQuantFile() {
+ return quantFile;
+ }
+
+ public void setQuantFile(String quantFile) {
+ this.quantFile = quantFile;
+ }
+
+ public void setSortOrder(boolean[] sortOrder) {
+ if(null == sortOrder) return;
+ this.sortOrder = new boolean[sortOrder.length];
+ for(int i = 0; i < sortOrder.length; ++i) {
+ this.sortOrder[i] = sortOrder[i];
+ }
+ }
+
+ public boolean[] getSortOrder() {
+ return sortOrder;
+ }
+
+ public void setGlobalSort(boolean globalSort) {
+ this.globalSort = globalSort;
+ }
+
+ public boolean isGlobalSort() {
+ return globalSort;
+ }
+
+ public boolean isLimitAfterSort() {
+ return limitAfterSort;
+ }
}
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezPlanContainer.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezPlanContainer.java?rev=1545171&r1=1545170&r2=1545171&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezPlanContainer.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezPlanContainer.java Mon Nov 25 08:23:04 2013
@@ -17,9 +17,12 @@
*/
package org.apache.pig.backend.hadoop.executionengine.tez;
+import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
+import java.io.PrintStream;
import java.net.URL;
+import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -31,6 +34,7 @@ import org.apache.pig.impl.plan.NodeIdGe
import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.impl.plan.OperatorPlan;
import org.apache.pig.impl.plan.PlanException;
+import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.impl.util.JarManager;
public class TezPlanContainer extends OperatorPlan<TezPlanContainerNode> {
@@ -107,20 +111,50 @@ public class TezPlanContainer extends Op
public void split(TezPlanContainerNode planNode) throws PlanException {
TezOperPlan tezOperPlan = planNode.getNode();
+ TezOperator operToSegment = null;
+ List<TezOperator> succs = new ArrayList<TezOperator>();
for (TezOperator tezOper : tezOperPlan) {
if (tezOper.needSegmentBelow() && tezOperPlan.getSuccessors(tezOper)!=null) {
- for (TezOperator succ : tezOperPlan.getSuccessors(tezOper)) {
- tezOperPlan.disconnect(tezOper, succ);
- TezOperPlan newOperPlan = new TezOperPlan();
- tezOperPlan.moveTree(succ, newOperPlan);
- TezPlanContainerNode newPlanNode = new TezPlanContainerNode(tezOper.getOperatorKey(), newOperPlan);
- add(newPlanNode);
- connect(planNode, newPlanNode);
- split(planNode);
- split(newPlanNode);
+ operToSegment = tezOper;
+ succs.addAll(tezOperPlan.getSuccessors(tezOper));
+ break;
+ }
+ }
+ if (operToSegment != null) {
+ for (TezOperator succ : succs) {
+ tezOperPlan.disconnect(operToSegment, succ);
+ TezOperPlan newOperPlan = new TezOperPlan();
+ List<TezPlanContainerNode> containerSuccs = new ArrayList<TezPlanContainerNode>();
+ if (getSuccessors(planNode)!=null) {
+ containerSuccs.addAll(getSuccessors(planNode));
+ }
+ tezOperPlan.moveTree(succ, newOperPlan);
+ String scope = operToSegment.getOperatorKey().getScope();
+ TezPlanContainerNode newPlanNode = new TezPlanContainerNode(new OperatorKey(scope, NodeIdGenerator.getGenerator().getNextNodeId(scope)), newOperPlan);
+ add(newPlanNode);
+ for (TezPlanContainerNode containerNodeSucc : containerSuccs) {
+ disconnect(planNode, containerNodeSucc);
+ connect(newPlanNode, containerNodeSucc);
}
+ connect(planNode, newPlanNode);
+ split(newPlanNode);
}
+ split(planNode);
+ }
+ }
+
+ @Override
+ public String toString() {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ PrintStream ps = new PrintStream(baos);
+ TezPlanContainerPrinter printer = new TezPlanContainerPrinter(ps, this);
+ printer.setVerbose(true);
+ try {
+ printer.visit();
+ } catch (VisitorException e) {
+ throw new RuntimeException(e);
}
+ return baos.toString();
}
}
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezPlanContainerNode.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezPlanContainerNode.java?rev=1545171&r1=1545170&r2=1545171&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezPlanContainerNode.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezPlanContainerNode.java Mon Nov 25 08:23:04 2013
@@ -65,5 +65,10 @@ public class TezPlanContainerNode extend
public int hashCode() {
return getNode().hashCode();
}
+
+ @Override
+ public String toString() {
+ return getNode().toString();
+ }
}
Modified: pig/branches/tez/src/org/apache/pig/impl/plan/OperatorPlan.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/impl/plan/OperatorPlan.java?rev=1545171&r1=1545170&r2=1545171&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/impl/plan/OperatorPlan.java (original)
+++ pig/branches/tez/src/org/apache/pig/impl/plan/OperatorPlan.java Mon Nov 25 08:23:04 2013
@@ -290,7 +290,15 @@ public abstract class OperatorPlan<E ext
* @throws PlanException
*/
public void moveTree(E root, OperatorPlan<E> newPlan) throws PlanException {
- List<E> succs = getSuccessors(root);
+ newPlan.add(root);
+ if (getSuccessors(root) == null) {
+ remove(root);
+ return;
+ }
+
+ List<E> succs = new ArrayList<E>();
+ succs.addAll(getSuccessors(root));
+
for (E succ : succs) {
moveTree(succ, newPlan);
}
Modified: pig/branches/tez/src/org/apache/pig/tools/pigstats/tez/TezStats.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/tools/pigstats/tez/TezStats.java?rev=1545171&r1=1545170&r2=1545171&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/tools/pigstats/tez/TezStats.java (original)
+++ pig/branches/tez/src/org/apache/pig/tools/pigstats/tez/TezStats.java Mon Nov 25 08:23:04 2013
@@ -163,9 +163,11 @@ public class TezStats extends PigStats {
DAG dag = tezJob.getDag();
for (String name : tezOpVertexMap.keySet()) {
Vertex v = dag.getVertex(name);
- byte[] bb = v.getProcessorDescriptor().getUserPayload();
- Configuration conf = TezUtils.createConfFromUserPayload(bb);
- addVertexStats(name, conf, succeeded);
+ if (v != null) {
+ byte[] bb = v.getProcessorDescriptor().getUserPayload();
+ Configuration conf = TezUtils.createConfFromUserPayload(bb);
+ addVertexStats(name, conf, succeeded);
+ }
}
if (!succeeded) {
errorMessage = tezJob.getMessage();
Modified: pig/branches/tez/test/e2e/pig/tests/tez.conf
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/e2e/pig/tests/tez.conf?rev=1545171&r1=1545170&r2=1545171&view=diff
==============================================================================
--- pig/branches/tez/test/e2e/pig/tests/tez.conf (original)
+++ pig/branches/tez/test/e2e/pig/tests/tez.conf Mon Nov 25 08:23:04 2013
@@ -95,12 +95,38 @@ store c into ':OUTPATH:';\,
},
{
'num' => 2,
- 'pig' => q\a = load ':INPATH:/singlefile/studenttab20m' as (name:chararray, age:int, gpa:double);
+ 'pig' => q\a = load ':INPATH:/singlefile/studenttab10k' as (name:chararray, age:int, gpa:double);
b = foreach a generate age;
c = distinct b;
d = filter c by age > 30;
store d into ':OUTPATH:';\,
},
+ {
+ # Order by simple
+ 'num' => 3,
+ 'pig' => q\a = load ':INPATH:/singlefile/studenttab10k' as (name:chararray, age:int, gpa:double);
+b = order a by age parallel 2;
+store b into ':OUTPATH:';\,
+ 'sortArgs' => ['-t', ' ', '-k', '2,2'],
+ },
+ {
+ # Order by simple no schema
+ 'num' => 4,
+ 'pig' => q\a = load ':INPATH:/singlefile/studenttab10k';
+b = order a by $1;
+store b into ':OUTPATH:';\,
+ 'sortArgs' => ['-t', ' ', '-k', '2,2'],
+ },
+ {
+ # Order by after group
+ 'num' => 5,
+ 'pig' => q\a = load ':INPATH:/singlefile/studenttab10k' as (name:chararray, age:int, gpa:double);
+b = group a by name;
+c = foreach b generate group, AVG(a.gpa) as avg_gpa;
+d = order c by avg_gpa parallel 2;
+store d into ':OUTPATH:';\,
+ 'sortArgs' => ['-t', ' ', '-k', '2,2'],
+ },
]
}
]