You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ga...@apache.org on 2008/08/04 21:28:39 UTC
svn commit: r682474 [2/3] - in /incubator/pig/branches/types:
src/org/apache/pig/ src/org/apache/pig/backend/executionengine/util/
src/org/apache/pig/backend/hadoop/
src/org/apache/pig/backend/hadoop/executionengine/
src/org/apache/pig/backend/hadoop/e...
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POIsNull.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POIsNull.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POIsNull.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POIsNull.java Mon Aug 4 12:28:37 2008
@@ -20,15 +20,16 @@
import java.util.Map;
import org.apache.pig.backend.executionengine.ExecException;
-import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
-import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.NodeIdGenerator;
+import org.apache.pig.impl.plan.VisitorException;
public class POIsNull extends UnaryComparisonOperator {
@@ -49,7 +50,7 @@
@Override
public void visit(PhyPlanVisitor v) throws VisitorException {
- //v.visitIsNull(this);
+ v.visitIsNull(this);
}
@Override
@@ -187,4 +188,12 @@
}
return res;
}
+
+ @Override
+ public POIsNull clone() throws CloneNotSupportedException {
+ POIsNull clone = new POIsNull(new OperatorKey(mKey.scope,
+ NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)));
+ clone.cloneHelper(this);
+ return clone;
+ }
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POMapLookUp.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POMapLookUp.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POMapLookUp.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POMapLookUp.java Mon Aug 4 12:28:37 2008
@@ -20,14 +20,15 @@
import java.util.Map;
import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.plan.OperatorKey;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
+import org.apache.pig.impl.plan.NodeIdGenerator;
import org.apache.pig.impl.plan.VisitorException;
public class POMapLookUp extends ExpressionOperator {
@@ -149,6 +150,14 @@
public Result getNext(Tuple t) throws ExecException {
return getNext();
}
+
+ @Override
+ public POMapLookUp clone() throws CloneNotSupportedException {
+ POMapLookUp clone = new POMapLookUp(new OperatorKey(mKey.scope,
+ NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)), -1, key);
+ clone.cloneHelper(this);
+ return clone;
+ }
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/PONegative.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/PONegative.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/PONegative.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/PONegative.java Mon Aug 4 12:28:37 2008
@@ -18,11 +18,12 @@
package org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators;
import org.apache.pig.backend.executionengine.ExecException;
-import org.apache.pig.data.DataType;
-import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
+import org.apache.pig.data.DataType;
+import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.NodeIdGenerator;
import org.apache.pig.impl.plan.VisitorException;
public class PONegative extends UnaryExpressionOperator {
@@ -88,4 +89,12 @@
}
return res;
}
+
+ @Override
+ public PONegative clone() throws CloneNotSupportedException {
+ PONegative clone = new PONegative(new OperatorKey(mKey.scope,
+ NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)));
+ clone.cloneHelper(this);
+ return clone;
+ }
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/PONot.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/PONot.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/PONot.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/PONot.java Mon Aug 4 12:28:37 2008
@@ -18,11 +18,12 @@
package org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators;
import org.apache.pig.backend.executionengine.ExecException;
-import org.apache.pig.data.DataType;
-import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
+import org.apache.pig.data.DataType;
+import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.NodeIdGenerator;
import org.apache.pig.impl.plan.VisitorException;
/**
@@ -74,4 +75,12 @@
if (((Boolean)res.result).booleanValue()) return falseRes;
else return trueRes;
}
+
+ @Override
+ public PONot clone() throws CloneNotSupportedException {
+ PONot clone = new PONot(new OperatorKey(mKey.scope,
+ NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)));
+ clone.cloneHelper(this);
+ return clone;
+ }
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POOr.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POOr.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POOr.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POOr.java Mon Aug 4 12:28:37 2008
@@ -18,11 +18,12 @@
package org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators;
import org.apache.pig.backend.executionengine.ExecException;
-import org.apache.pig.data.DataType;
-import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
+import org.apache.pig.data.DataType;
+import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.NodeIdGenerator;
import org.apache.pig.impl.plan.VisitorException;
/**
@@ -69,4 +70,12 @@
// return, error, null, true, or false.
return rhs.getNext(dummyBool);
}
+
+ @Override
+ public POOr clone() throws CloneNotSupportedException {
+ POOr clone = new POOr(new OperatorKey(mKey.scope,
+ NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)));
+ clone.cloneHelper(this);
+ return clone;
+ }
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POProject.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POProject.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POProject.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POProject.java Mon Aug 4 12:28:37 2008
@@ -32,6 +32,8 @@
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
+import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.NodeIdGenerator;
import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.data.Tuple;
@@ -316,6 +318,22 @@
public void setStar(boolean star) {
this.star = star;
}
+
+ @Override
+ public POProject clone() throws CloneNotSupportedException {
+ ArrayList<Integer> cols = new ArrayList<Integer>(columns.size());
+ // Can resuse the same Integer objects, as they are immutable
+ for (Integer i : columns) {
+ cols.add(i);
+ }
+ POProject clone = new POProject(new OperatorKey(mKey.scope,
+ NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)),
+ requestedParallelism, cols);
+ clone.cloneHelper(this);
+ clone.star = star;
+ clone.overloaded = overloaded;
+ return clone;
+ }
private Result processInputBag() throws ExecException {
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/PORegexp.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/PORegexp.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/PORegexp.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/PORegexp.java Mon Aug 4 12:28:37 2008
@@ -20,11 +20,12 @@
import java.util.regex.PatternSyntaxException;
import org.apache.pig.backend.executionengine.ExecException;
-import org.apache.pig.data.DataType;
-import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
+import org.apache.pig.data.DataType;
+import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.NodeIdGenerator;
import org.apache.pig.impl.plan.VisitorException;
public class PORegexp extends BinaryComparisonOperator {
@@ -71,4 +72,12 @@
}
return left;
}
+
+ @Override
+ public PORegexp clone() throws CloneNotSupportedException {
+ PORegexp clone = new PORegexp(new OperatorKey(mKey.scope,
+ NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)));
+ clone.cloneHelper(this);
+ return clone;
+ }
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POUserComparisonFunc.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POUserComparisonFunc.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POUserComparisonFunc.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POUserComparisonFunc.java Mon Aug 4 12:28:37 2008
@@ -10,16 +10,17 @@
import org.apache.pig.ComparisonFunc;
import org.apache.pig.FuncSpec;
import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.NodeIdGenerator;
import org.apache.pig.impl.plan.VisitorException;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
public class POUserComparisonFunc extends ExpressionOperator {
@@ -150,4 +151,18 @@
return funcSpec;
}
+ @Override
+ public POUserComparisonFunc clone() throws CloneNotSupportedException {
+ FuncSpec cloneFs = null;
+ if (funcSpec != null) {
+ cloneFs = funcSpec.clone();
+ }
+ POUserComparisonFunc clone =
+ new POUserComparisonFunc(new OperatorKey(mKey.scope,
+ NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)),
+ requestedParallelism, null, cloneFs);
+ clone.cloneHelper(this);
+ return clone;
+ }
+
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POUserFunc.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POUserFunc.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POUserFunc.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POUserFunc.java Mon Aug 4 12:28:37 2008
@@ -42,6 +42,8 @@
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
+import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.NodeIdGenerator;
import org.apache.pig.impl.plan.VisitorException;
public class POUserFunc extends ExpressionOperator {
@@ -59,18 +61,26 @@
public static final byte INTERMEDIATE = 1;
public static final byte FINAL = 2;
- public POUserFunc(OperatorKey k, int rp, List inp) {
+ public POUserFunc(OperatorKey k, int rp, List<PhysicalOperator> inp) {
super(k, rp);
inputs = inp;
}
- public POUserFunc(OperatorKey k, int rp, List inp, FuncSpec funcSpec) {
+ public POUserFunc(
+ OperatorKey k,
+ int rp,
+ List<PhysicalOperator> inp,
+ FuncSpec funcSpec) {
this(k, rp, inp, funcSpec, null);
}
- public POUserFunc(OperatorKey k, int rp, List inp, FuncSpec funcSpec, EvalFunc func) {
-// super(k, rp, inp);
+ public POUserFunc(
+ OperatorKey k,
+ int rp,
+ List<PhysicalOperator> inp,
+ FuncSpec funcSpec,
+ EvalFunc func) {
super(k, rp);
super.setInputs(inp);
this.funcSpec = funcSpec;
@@ -170,7 +180,8 @@
return result;
} catch (IOException e1) {
- log.error(e1);
+ log.error("Caught error from UDF " + funcSpec.getClassName() +
+ "[" + e1.getMessage() + "]");
}
@@ -235,7 +246,7 @@
return getNext();
}
- public void setAlgebraicFunction(Byte Function) {
+ public void setAlgebraicFunction(byte Function) {
// This will only be used by the optimizer for putting correct functions
// in the mapper,
// combiner and reduce. This helps in maintaining the physical plan as
@@ -265,10 +276,11 @@
if (func instanceof Algebraic) {
return ((Algebraic) func).getInitial();
} else {
- log
- .error("Attempt to run a non-algebraic function as an algebraic function");
+ String msg = new String("Attempt to run a non-algebraic function"
+ + " as an algebraic function");
+ log.error(msg);
+ throw new RuntimeException(msg);
}
- return null;
}
public String getIntermed() {
@@ -276,10 +288,11 @@
if (func instanceof Algebraic) {
return ((Algebraic) func).getIntermed();
} else {
- log
- .error("Attempt to run a non-algebraic function as an algebraic function");
+ String msg = new String("Attempt to run a non-algebraic function"
+ + " as an algebraic function");
+ log.error(msg);
+ throw new RuntimeException(msg);
}
- return null;
}
public String getFinal() {
@@ -287,10 +300,11 @@
if (func instanceof Algebraic) {
return ((Algebraic) func).getFinal();
} else {
- log
- .error("Attempt to run a non-algebraic function as an algebraic function");
+ String msg = new String("Attempt to run a non-algebraic function"
+ + " as an algebraic function");
+ log.error(msg);
+ throw new RuntimeException(msg);
}
- return null;
}
public Type getReturnType() {
@@ -335,6 +349,20 @@
public FuncSpec getFuncSpec() {
return funcSpec;
}
+
+ public boolean combinable() {
+ return (func instanceof Algebraic);
+ }
+
+ @Override
+ public POUserFunc clone() throws CloneNotSupportedException {
+ // Inputs will be patched up later by PhysicalPlan.clone()
+ POUserFunc clone = new POUserFunc(new OperatorKey(mKey.scope,
+ NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)),
+ requestedParallelism, null, funcSpec.clone());
+ clone.setResultType(resultType);
+ return clone;
+ }
private void readObject(ObjectInputStream is) throws IOException, ClassNotFoundException{
is.defaultReadObject();
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/Subtract.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/Subtract.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/Subtract.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/Subtract.java Mon Aug 4 12:28:37 2008
@@ -18,11 +18,12 @@
package org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators;
import org.apache.pig.backend.executionengine.ExecException;
-import org.apache.pig.data.DataType;
-import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
+import org.apache.pig.data.DataType;
+import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.NodeIdGenerator;
import org.apache.pig.impl.plan.VisitorException;
public class Subtract extends BinaryExpressionOperator {
@@ -142,4 +143,12 @@
return res;
}
+ @Override
+ public Subtract clone() throws CloneNotSupportedException {
+ Subtract clone = new Subtract(new OperatorKey(mKey.scope,
+ NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)));
+ clone.cloneHelper(this);
+ return clone;
+ }
+
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/UnaryExpressionOperator.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/UnaryExpressionOperator.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/UnaryExpressionOperator.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/UnaryExpressionOperator.java Mon Aug 4 12:28:37 2008
@@ -65,4 +65,12 @@
public ExpressionOperator getExpr() {
return expr;
}
+
+ protected void cloneHelper(UnaryExpressionOperator op) {
+ // Don't clone this, as it is just a reference to something already in
+ // the plan.
+ expr = op.expr;
+ resultType = op.getResultType();
+ }
+
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/PhyPlanVisitor.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/PhyPlanVisitor.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/PhyPlanVisitor.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/PhyPlanVisitor.java Mon Aug 4 12:28:37 2008
@@ -20,42 +20,8 @@
import java.util.List;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.Add;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.ConstantExpression;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.Divide;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.EqualToExpr;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.GTOrEqualToExpr;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.GreaterThanExpr;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.LTOrEqualToExpr;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.LessThanExpr;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.Mod;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.Multiply;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.NotEqualToExpr;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POAnd;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POBinCond;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POCast;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POMapLookUp;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.PONegative;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.PONot;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POOr;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POProject;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.PORegexp;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserComparisonFunc;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserFunc;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.Subtract;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POForEach;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PODistinct;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POFilter;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POGlobalRearrange;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLimit;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLoad;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLocalRearrange;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPackage;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PORead;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POSort;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POSplit;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStore;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POUnion;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.*;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.*;
import org.apache.pig.impl.plan.PlanVisitor;
import org.apache.pig.impl.plan.PlanWalker;
import org.apache.pig.impl.plan.VisitorException;
@@ -99,7 +65,7 @@
visit();
}
}
-
+
public void visitGlobalRearrange(POGlobalRearrange gr) throws VisitorException{
//do nothing
}
@@ -176,6 +142,9 @@
public void visitRegexp(PORegexp re) throws VisitorException{
//do nothing
}
+
+ public void visitIsNull(POIsNull isNull) throws VisitorException {
+ }
public void visitAdd(Add add) throws VisitorException{
//do nothing
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/PhysicalPlan.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/PhysicalPlan.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/PhysicalPlan.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/PhysicalPlan.java Mon Aug 4 12:28:37 2008
@@ -20,7 +20,11 @@
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import org.apache.pig.data.Tuple;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
@@ -34,7 +38,7 @@
* This extends the Operator Plan.
*
*/
-public class PhysicalPlan extends OperatorPlan<PhysicalOperator> {
+public class PhysicalPlan extends OperatorPlan<PhysicalOperator> implements Cloneable {
/**
*
@@ -117,6 +121,77 @@
// TODO Auto-generated method stub
return super.equals(obj);
}
+
+ @Override
+ public PhysicalPlan clone() throws CloneNotSupportedException {
+ PhysicalPlan clone = new PhysicalPlan();
+
+ // Get all the nodes in this plan, and clone them. As we make
+ // clones, create a map between clone and original. Then walk the
+ // connections in this plan and create equivalent connections in the
+ // clone.
+ Map<PhysicalOperator, PhysicalOperator> matches =
+ new HashMap<PhysicalOperator, PhysicalOperator>(mOps.size());
+ for (PhysicalOperator op : mOps.keySet()) {
+ PhysicalOperator c = op.clone();
+ clone.add(c);
+ matches.put(op, c);
+ }
+
+ // Build the edges
+ for (PhysicalOperator op : mFromEdges.keySet()) {
+ PhysicalOperator cloneFrom = matches.get(op);
+ if (cloneFrom == null) {
+ String msg = new String("Unable to find clone for op "
+ + op.name());
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+ Collection<PhysicalOperator> toOps = mFromEdges.get(op);
+ for (PhysicalOperator toOp : toOps) {
+ PhysicalOperator cloneTo = matches.get(toOp);
+ if (cloneTo == null) {
+ String msg = new String("Unable to find clone for op "
+ + toOp.name());
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+ try {
+ clone.connect(cloneFrom, cloneTo);
+ } catch (PlanException pe) {
+ throw new RuntimeException(pe);
+ }
+ }
+ }
+
+ // Fix up all the inputs in the operators themselves.
+ for (PhysicalOperator op : mOps.keySet()) {
+ List<PhysicalOperator> inputs = op.getInputs();
+ if (inputs == null || inputs.size() == 0) continue;
+ List<PhysicalOperator> newInputs =
+ new ArrayList<PhysicalOperator>(inputs.size());
+ PhysicalOperator cloneOp = matches.get(op);
+ if (cloneOp == null) {
+ String msg = new String("Unable to find clone for op "
+ + cloneOp.name());
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+ for (PhysicalOperator iOp : inputs) {
+ PhysicalOperator cloneIOp = matches.get(iOp);
+ if (cloneIOp == null) {
+ String msg = new String("Unable to find clone for op "
+ + cloneIOp.name());
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+ newInputs.add(cloneIOp);
+ }
+ cloneOp.setInputs(newInputs);
+ }
+
+ return clone;
+ }
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PODistinct.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PODistinct.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PODistinct.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PODistinct.java Mon Aug 4 12:28:37 2008
@@ -24,18 +24,19 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.plan.OperatorKey;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
import org.apache.pig.impl.plan.VisitorException;
/**
+ * Find the distinct set of tuples in a bag.
* This is a blocking operator. All the input is put in the hashset implemented
* in DistinctDataBag which also provides the other DataBag interfaces.
*
@@ -50,27 +51,22 @@
public PODistinct(OperatorKey k, int rp, List<PhysicalOperator> inp) {
super(k, rp, inp);
- // TODO Auto-generated constructor stub
}
public PODistinct(OperatorKey k, int rp) {
super(k, rp);
- // TODO Auto-generated constructor stub
}
public PODistinct(OperatorKey k, List<PhysicalOperator> inp) {
super(k, inp);
- // TODO Auto-generated constructor stub
}
public PODistinct(OperatorKey k) {
super(k);
- // TODO Auto-generated constructor stub
}
@Override
public boolean isBlocking() {
- // TODO Auto-generated method stub
return true;
}
@@ -111,25 +107,21 @@
@Override
public String name() {
- // TODO Auto-generated method stub
return "PODistinct" + "[" + DataType.findTypeName(resultType) + "]" +" - " + mKey.toString();
}
@Override
public boolean supportsMultipleInputs() {
- // TODO Auto-generated method stub
return false;
}
@Override
public boolean supportsMultipleOutputs() {
- // TODO Auto-generated method stub
return false;
}
@Override
public void visit(PhyPlanVisitor v) throws VisitorException {
- // TODO Auto-generated method stub
v.visitDistinct(this);
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java Mon Aug 4 12:28:37 2008
@@ -1,5 +1,6 @@
package org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators;
+import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
@@ -19,6 +20,7 @@
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.NodeIdGenerator;
import org.apache.pig.impl.plan.VisitorException;
public class POForEach extends PhysicalOperator {
@@ -308,7 +310,7 @@
for(int i = 0; i < data.length; ++i) {
Object in = data[i];
- if(in instanceof Tuple) {
+ if(isToBeFlattened.get(i) && in instanceof Tuple) {
Tuple t = (Tuple)in;
for(int j = 0; j < t.size(); ++j) {
out.append(t.get(j));
@@ -345,9 +347,42 @@
getLeaves();
}
+ public void addInputPlan(PhysicalPlan plan, boolean flatten) {
+ inputPlans.add(plan);
+ planLeaves.add(plan.getLeaves().get(0));
+ isToBeFlattened.add(flatten);
+ }
+
public void setToBeFlattened(List<Boolean> flattens) {
isToBeFlattened = flattens;
}
+ public List<Boolean> getToBeFlattened() {
+ return isToBeFlattened;
+ }
+
+ /**
+ * Make a deep copy of this operator.
+ * @throws CloneNotSupportedException
+ */
+ @Override
+ public POForEach clone() throws CloneNotSupportedException {
+ List<PhysicalPlan> plans = new
+ ArrayList<PhysicalPlan>(inputPlans.size());
+ for (PhysicalPlan plan : inputPlans) {
+ plans.add(plan.clone());
+ }
+ List<Boolean> flattens = new
+ ArrayList<Boolean>(isToBeFlattened.size());
+ for (Boolean b : isToBeFlattened) {
+ // Boolean is immutable, so using same reference is ok
+ flattens.add(b);
+ }
+ return new POForEach(new OperatorKey(mKey.scope,
+ NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)),
+ requestedParallelism, plans, flattens);
+ }
+
+
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLocalRearrange.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLocalRearrange.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLocalRearrange.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLocalRearrange.java Mon Aug 4 12:28:37 2008
@@ -34,6 +34,7 @@
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.NodeIdGenerator;
import org.apache.pig.impl.plan.VisitorException;
/**
@@ -49,6 +50,8 @@
*/
private static final long serialVersionUID = 1L;
+ private static TupleFactory mTupleFactory = TupleFactory.getInstance();
+
private Log log = LogFactory.getLog(getClass());
List<PhysicalPlan> plans;
@@ -60,6 +63,15 @@
byte keyType;
+ private boolean mIsDistinct = false;
+
+ // A place holder IndexedTuple used in distinct case where we really don't
+ // have any value to pass through. But hadoop gets cranky if we pass a
+ // null, so we'll just create one instance of this empty indexed tuple and
+ // pass it for every row. We only get around to actually creating it if
+ // mIsDistinct is set to true.
+ private IndexedTuple mFakeIndexedTuple = null;
+
public POLocalRearrange(OperatorKey k) {
this(k, -1, null);
}
@@ -85,7 +97,9 @@
@Override
public String name() {
- return "Local Rearrange" + "[" + DataType.findTypeName(resultType) + "]" + "{" + DataType.findTypeName(keyType) + "}" +" - " + mKey.toString();
+ return "Local Rearrange" + "[" + DataType.findTypeName(resultType) +
+ "]" + "{" + DataType.findTypeName(keyType) + "}" + "(" +
+ mIsDistinct + ") - " + mKey.toString();
}
@Override
@@ -105,6 +119,17 @@
public void setIndex(int index) {
this.index = index;
}
+
+ public boolean isDistinct() {
+ return mIsDistinct;
+ }
+
+ public void setDistinct(boolean isDistinct) {
+ mIsDistinct = isDistinct;
+ if (mIsDistinct) {
+ mFakeIndexedTuple = new IndexedTuple(mTupleFactory.newTuple(), 0);
+ }
+ }
/**
* Overridden since the attachment of the new input should cause the old
@@ -184,7 +209,7 @@
//Construct key
Object key;
if(resLst.size()>1){
- Tuple t = TupleFactory.getInstance().newTuple(resLst.size());
+ Tuple t = mTupleFactory.newTuple(resLst.size());
int i=-1;
for(Result res : resLst)
t.set(++i, res.result);
@@ -194,16 +219,25 @@
key = resLst.get(0).result;
}
- //Create the indexed tuple out of the value
- //that is remaining in the input tuple
- IndexedTuple it = new IndexedTuple(value, index);
-
- //Put the key and the indexed tuple
- //in a tuple and return
- Tuple outPut = TupleFactory.getInstance().newTuple(2);
- outPut.set(0,key);
- outPut.set(1,it);
- return outPut;
+ Tuple outPut = mTupleFactory.newTuple(2);
+ if (mIsDistinct) {
+
+ //Put the key and the indexed tuple
+ //in a tuple and return
+ outPut.set(0,key);
+ outPut.set(1, mFakeIndexedTuple);
+ return outPut;
+ } else {
+ //Create the indexed tuple out of the value
+ //that is remaining in the input tuple
+ IndexedTuple it = new IndexedTuple(value, index);
+
+ //Put the key and the indexed tuple
+ //in a tuple and return
+ outPut.set(0,key);
+ outPut.set(1,it);
+ return outPut;
+ }
}
public byte getKeyType() {
@@ -225,4 +259,30 @@
leafOps.add((ExpressionOperator)plan.getLeaves().get(0));
}
}
+
+ /**
+ * Make a deep copy of this operator.
+ * @throws CloneNotSupportedException
+ */
+ @Override
+ public POLocalRearrange clone() throws CloneNotSupportedException {
+ List<PhysicalPlan> clonePlans = new
+ ArrayList<PhysicalPlan>(plans.size());
+ for (PhysicalPlan plan : plans) {
+ clonePlans.add(plan.clone());
+ }
+ POLocalRearrange clone = new POLocalRearrange(new OperatorKey(
+ mKey.scope,
+ NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)),
+ requestedParallelism);
+ clone.setPlans(clonePlans);
+ clone.keyType = keyType;
+ clone.index = index;
+ // Needs to be called as setDistinct so that the fake index tuple gets
+ // created.
+ clone.setDistinct(mIsDistinct);
+ return clone;
+ }
+
+
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPackage.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPackage.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPackage.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPackage.java Mon Aug 4 12:28:37 2008
@@ -30,6 +30,7 @@
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.NodeIdGenerator;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
@@ -66,7 +67,8 @@
byte keyType;
//The number of inputs to this
- //co-group
+ //co-group. 0 indicates a distinct, which means there will only be a
+ //key, no value.
int numInputs;
//Denotes if inner is specified
@@ -75,6 +77,9 @@
private final Log log = LogFactory.getLog(getClass());
+ protected static BagFactory mBagFactory = BagFactory.getInstance();
+ protected static TupleFactory mTupleFactory = TupleFactory.getInstance();
+
public POPackage(OperatorKey k) {
this(k, -1, null);
}
@@ -154,16 +159,13 @@
*/
@Override
public Result getNext(Tuple t) throws ExecException {
- //Should be removed once we start integration/perf testing
- if(indTupIter==null){
- throw new ExecException("Incorrect usage of the Package operator. " +
- "No input has been attached.");
- }
-
//Create numInputs bags
- DataBag[] dbs = new DataBag[numInputs];
- for (int i = 0; i < numInputs; i++) {
- dbs[i] = BagFactory.getInstance().newDefaultBag();
+ DataBag[] dbs = null;
+ if (numInputs > 0) {
+ dbs = new DataBag[numInputs];
+ for (int i = 0; i < numInputs; i++) {
+ dbs[i] = mBagFactory.newDefaultBag();
+ }
}
//For each indexed tup in the inp, sort them
@@ -171,7 +173,7 @@
//on the index
while (indTupIter.hasNext()) {
IndexedTuple it = indTupIter.next();
- dbs[it.index].add(it.toTuple());
+ if (numInputs > 0) dbs[it.index].add(it.toTuple());
if(reporter!=null) reporter.progress();
}
@@ -179,19 +181,21 @@
//the key and all the above constructed bags
//and return it.
Tuple res;
- res = TupleFactory.getInstance().newTuple(numInputs+1);
+ res = mTupleFactory.newTuple(numInputs+1);
res.set(0,key);
- int i=-1;
- for (DataBag bag : dbs) {
- if(inner[++i]){
- if(bag.size()==0){
- detachInput();
- Result r = new Result();
- r.returnStatus = POStatus.STATUS_NULL;
- return r;
+ if (numInputs > 0) {
+ int i=-1;
+ for (DataBag bag : dbs) {
+ if(inner[++i]){
+ if(bag.size()==0){
+ detachInput();
+ Result r = new Result();
+ r.returnStatus = POStatus.STATUS_NULL;
+ return r;
+ }
}
+ res.set(i+1,bag);
}
- res.set(i+1,bag);
}
detachInput();
Result r = new Result();
@@ -207,4 +211,24 @@
public void setKeyType(byte keyType) {
this.keyType = keyType;
}
+
+ /**
+ * Make a deep copy of this operator.
+ * @throws CloneNotSupportedException
+ */
+ @Override
+ public POPackage clone() throws CloneNotSupportedException {
+ POPackage clone = new POPackage(new OperatorKey(mKey.scope,
+ NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)));
+ clone.resultType = resultType;
+ clone.keyType = keyType;
+ clone.numInputs = numInputs;
+ clone.inner = new boolean[inner.length];
+ for (int i = 0; i < inner.length; i++) {
+ clone.inner[i] = inner[i];
+ }
+ return clone;
+ }
+
+
}
Added: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPostCombinerPackage.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPostCombinerPackage.java?rev=682474&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPostCombinerPackage.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPostCombinerPackage.java Mon Aug 4 12:28:37 2008
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators;
+
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
+import org.apache.pig.data.BagFactory;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.IndexedTuple;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.NodeIdGenerator;
+import org.apache.pig.impl.plan.VisitorException;
+/**
+ * The package operator that packages the globally rearranged tuples into
+ * output format after the combiner stage. It differs from POPackage in that
+ * instead of using the index in the IndexedTuple to find the bag to put a
+ * tuple in, it instead uses the index to find the key. All other inputs are
+ * put in a bag corresponding to their offset in the tuple.
+ */
+public class POPostCombinerPackage extends POPackage {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ private final Log log = LogFactory.getLog(getClass());
+
+ private static BagFactory mBagFactory = BagFactory.getInstance();
+ private static TupleFactory mTupleFactory = TupleFactory.getInstance();
+
+ private boolean[] mBags; // For each field, indicates whether or not it
+ // needs to be put in a bag.
+
+ /**
+ * A new POPostCombinePackage will be constructed as a near clone of the
+ * provided POPackage.
+ * @param pkg POPackage to clone.
+ * @param numFields Number of fields in each input tuple
+ */
+ public POPostCombinerPackage(POPackage pkg, boolean[] bags) {
+ super(new OperatorKey(pkg.getOperatorKey().scope,
+ NodeIdGenerator.getGenerator().getNextNodeId(pkg.getOperatorKey().scope)),
+ pkg.getRequestedParallelism(), pkg.getInputs());
+ resultType = pkg.getResultType();
+ keyType = pkg.keyType;
+ numInputs = 1;
+ inner = new boolean[1];
+ for (int i = 0; i < pkg.inner.length; i++) {
+ inner[i] = true;
+ }
+ mBags = bags;
+ }
+
+ @Override
+ public String name() {
+ return "PostCombinerPackage" + "[" + DataType.findTypeName(resultType) + "]" + "{" + DataType.findTypeName(keyType) + "}" +" - " + mKey.toString();
+ }
+
+ /**
+ * From the inputs, constructs the output tuple
+ * for this co-group in the required format which
+ * is (key, {bag of tuples from input 1}, {bag of tuples from input 2}, ...)
+ */
+ @Override
+ public Result getNext(Tuple t) throws ExecException {
+ int keyField = -1;
+ //Create numInputs bags
+ Object[] fields = new Object[mBags.length];
+ for (int i = 0; i < mBags.length; i++) {
+ if (mBags[i]) fields[i] = mBagFactory.newDefaultBag();
+ }
+
+ // For each indexed tup in the inp, split them up and place their
+ // fields into the proper bags. If the given field isn't a bag, just
+ // return set the value as is.
+ while (indTupIter.hasNext()) {
+ IndexedTuple it = indTupIter.next();
+ Tuple tup = it.toTuple();
+ for (int i = 0; i < tup.size(); i++) {
+ if (mBags[i]) ((DataBag)fields[i]).add((Tuple)tup.get(i));
+ else fields[i] = tup.get(i);
+ }
+ }
+
+ //Construct the output tuple by appending
+ //the key and all the above constructed bags
+ //and return it.
+ Tuple res;
+ res = mTupleFactory.newTuple(mBags.length);
+ for (int i = 0; i < mBags.length; i++) res.set(i, fields[i]);
+ Result r = new Result();
+ r.result = res;
+ r.returnStatus = POStatus.STATUS_OK;
+ return r;
+
+ }
+
+}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSort.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSort.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSort.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSort.java Mon Aug 4 12:28:37 2008
@@ -26,12 +26,6 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.pig.backend.executionengine.ExecException;
-import org.apache.pig.data.BagFactory;
-import org.apache.pig.data.DataBag;
-import org.apache.pig.data.DataType;
-import org.apache.pig.data.Tuple;
-import org.apache.pig.data.TupleFactory;
-import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
@@ -40,6 +34,13 @@
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.ExpressionOperator;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserComparisonFunc;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserFunc;
+import org.apache.pig.data.BagFactory;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.NodeIdGenerator;
import org.apache.pig.impl.plan.VisitorException;
/**
@@ -74,8 +75,13 @@
private DataBag sortedBag;
transient Iterator<Tuple> it;
- public POSort(OperatorKey k, int rp, List inp, List<PhysicalPlan> sortPlans,
- List<Boolean> mAscCols, POUserComparisonFunc mSortFunc) {
+ public POSort(
+ OperatorKey k,
+ int rp,
+ List inp,
+ List<PhysicalPlan> sortPlans,
+ List<Boolean> mAscCols,
+ POUserComparisonFunc mSortFunc) {
super(k, rp, inp);
//this.mSortCols = mSortCols;
this.sortPlans = sortPlans;
@@ -329,4 +335,27 @@
return (limit!=-1);
}
+ @Override
+ public POSort clone() throws CloneNotSupportedException {
+ List<PhysicalPlan> clonePlans = new
+ ArrayList<PhysicalPlan>(sortPlans.size());
+ for (PhysicalPlan plan : sortPlans) {
+ clonePlans.add(plan.clone());
+ }
+ List<Boolean> cloneAsc = new ArrayList<Boolean>(mAscCols.size());
+ for (Boolean b : mAscCols) {
+ cloneAsc.add(b);
+ }
+ POUserComparisonFunc cloneFunc = null;
+ if (mSortFunc != null) {
+ cloneFunc = mSortFunc.clone();
+ }
+ // Don't set inputs as PhysicalPlan.clone will take care of that
+ return new POSort(new OperatorKey(mKey.scope,
+ NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)),
+ requestedParallelism, null, clonePlans, cloneAsc, cloneFunc);
+ }
+
+
+
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java Mon Aug 4 12:28:37 2008
@@ -39,6 +39,7 @@
import org.apache.pig.backend.executionengine.ExecJob.JOB_STATUS;
import org.apache.pig.backend.executionengine.ExecScopedLogicalOperator;
import org.apache.pig.backend.executionengine.ExecPhysicalPlan;
+import org.apache.pig.backend.executionengine.util.ExecTools;
import org.apache.pig.backend.hadoop.executionengine.HJob;
import org.apache.pig.builtin.BinStorage;
import org.apache.pig.impl.io.FileLocalizer;
@@ -147,9 +148,9 @@
LocalLauncher launcher = new LocalLauncher();
boolean success = launcher.launchPig(plan, jobName, pigContext);
if(success)
- return new HJob(ExecJob.JOB_STATUS.COMPLETED, pigContext, spec);
+ return new LocalJob(ExecJob.JOB_STATUS.COMPLETED, pigContext, spec);
else
- return new HJob(ExecJob.JOB_STATUS.FAILED, pigContext, null);
+ return new LocalJob(ExecJob.JOB_STATUS.FAILED, pigContext, null);
} catch (Exception e) {
// There are a lot of exceptions thrown by the launcher. If this
// is an ExecException, just let it through. Else wrap it.
@@ -169,6 +170,8 @@
printer.visit();
stream.println();
+ ExecTools.checkLeafIsStore(plan, pigContext);
+
LocalLauncher launcher = new LocalLauncher();
launcher.explain(plan, pigContext, stream);
} catch (Exception ve) {
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/local/executionengine/LocalJob.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/local/executionengine/LocalJob.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/local/executionengine/LocalJob.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/local/executionengine/LocalJob.java Mon Aug 4 12:28:37 2008
@@ -19,23 +19,37 @@
package org.apache.pig.backend.local.executionengine;
import java.io.OutputStream;
+import java.io.InputStream;
import java.util.Iterator;
-import java.util.Properties;
import java.util.Map;
+import java.util.Properties;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.executionengine.ExecJob;
import org.apache.pig.data.Tuple;
-import org.apache.pig.data.DataBag;
+import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.io.FileSpec;
+import org.apache.pig.LoadFunc;
+import org.apache.pig.impl.io.FileLocalizer;
+import org.apache.pig.impl.io.BufferedPositionedInputStream;
+
public class LocalJob implements ExecJob {
- protected DataBag results;
+ private final Log log = LogFactory.getLog(getClass());
+
protected JOB_STATUS status;
+ protected PigContext pigContext;
+ protected FileSpec outFileSpec;
- public LocalJob(DataBag results, JOB_STATUS status) {
- this.results = results;
+ public LocalJob(JOB_STATUS status,
+ PigContext pigContext,
+ FileSpec outFileSpec) {
this.status = status;
+ this.pigContext = pigContext;
+ this.outFileSpec = outFileSpec;
}
public JOB_STATUS getStatus() {
@@ -47,7 +61,60 @@
}
public Iterator<Tuple> getResults() throws ExecException {
- return this.results.iterator();
+ final LoadFunc p;
+
+ try{
+ p = (LoadFunc)PigContext.instantiateFuncFromSpec(outFileSpec.getFuncSpec());
+
+ InputStream is = FileLocalizer.open(outFileSpec.getFileName(), pigContext);
+
+ p.bindTo(outFileSpec.getFileName(), new BufferedPositionedInputStream(is), 0, Long.MAX_VALUE);
+
+ }catch (Exception e){
+ throw new ExecException("Unable to get results for " + outFileSpec, e);
+ }
+
+ return new Iterator<Tuple>() {
+ Tuple t;
+ boolean atEnd;
+
+ public boolean hasNext() {
+ if (atEnd)
+ return false;
+ try {
+ if (t == null)
+ t = p.getNext();
+ if (t == null)
+ atEnd = true;
+ } catch (Exception e) {
+ log.error(e);
+ t = null;
+ atEnd = true;
+ }
+ return !atEnd;
+ }
+
+ public Tuple next() {
+ Tuple next = t;
+ if (next != null) {
+ t = null;
+ return next;
+ }
+ try {
+ next = p.getNext();
+ } catch (Exception e) {
+ log.error(e);
+ }
+ if (next == null)
+ atEnd = true;
+ return next;
+ }
+
+ public void remove() {
+ throw new RuntimeException("Removal not supported");
+ }
+
+ };
}
public Properties getContiguration() {
Modified: incubator/pig/branches/types/src/org/apache/pig/builtin/COUNT.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/COUNT.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/COUNT.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/COUNT.java Mon Aug 4 12:28:37 2008
@@ -29,6 +29,7 @@
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.util.WrappedIOException;
/**
* Generates the count of the values of the first field of a tuple. This class is Algebraic in
@@ -42,9 +43,7 @@
try {
return count(input);
} catch (ExecException ee) {
- IOException oughtToBeEE = new IOException();
- oughtToBeEE.initCause(ee);
- throw oughtToBeEE;
+ throw WrappedIOException.wrap("Caught exception in COUNT", ee);
}
}
@@ -67,9 +66,8 @@
try {
return mTupleFactory.newTuple(count(input));
} catch (ExecException ee) {
- IOException oughtToBeEE = new IOException();
- oughtToBeEE.initCause(ee);
- throw oughtToBeEE;
+ throw WrappedIOException.wrap(
+ "Caught exception in COUNT.Initial", ee);
}
}
}
@@ -81,9 +79,8 @@
try {
return mTupleFactory.newTuple(count(input));
} catch (ExecException ee) {
- IOException oughtToBeEE = new IOException();
- oughtToBeEE.initCause(ee);
- throw oughtToBeEE;
+ throw WrappedIOException.wrap(
+ "Caught exception in COUNT.Intermed", ee);
}
}
}
@@ -94,9 +91,8 @@
try {
return sum(input);
} catch (Exception ee) {
- IOException oughtToBeEE = new IOException();
- oughtToBeEE.initCause(ee);
- throw oughtToBeEE;
+ throw WrappedIOException.wrap(
+ "Caught exception in COUNT.Final", ee);
}
}
}
Modified: incubator/pig/branches/types/src/org/apache/pig/builtin/FloatSum.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/FloatSum.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/FloatSum.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/FloatSum.java Mon Aug 4 12:28:37 2008
@@ -28,6 +28,7 @@
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.util.WrappedIOException;
/**
@@ -40,9 +41,7 @@
try {
return sum(input);
} catch (ExecException ee) {
- IOException oughtToBeEE = new IOException();
- oughtToBeEE.initCause(ee);
- throw oughtToBeEE;
+ throw WrappedIOException.wrap("Caught exception in FloatSum", ee);
}
}
@@ -66,9 +65,7 @@
try {
return tfact.newTuple(sum(input));
} catch (ExecException ee) {
- IOException oughtToBeEE = new IOException();
- oughtToBeEE.initCause(ee);
- throw oughtToBeEE;
+ throw WrappedIOException.wrap("Caught exception in FloatSum.Initial", ee);
}
}
}
@@ -76,11 +73,39 @@
@Override
public Double exec(Tuple input) throws IOException {
try {
- return sum(input);
+ // Can't just call sum, because the intermediate results are
+ // now Doubles insteads of Floats.
+ DataBag values = (DataBag)input.get(0);
+
+ // if we were handed an empty bag, return NULL
+ // this is in compliance with SQL standard
+ if(values.size() == 0) {
+ return null;
+ }
+
+ long sum = 0;
+ boolean sawNonNull = false;
+ for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
+ Tuple t = (Tuple) it.next();
+ try {
+ Double d = (Double)(t.get(0));
+ if (d == null) continue;
+ sawNonNull = true;
+ sum += d;
+ }catch(RuntimeException exp) {
+ throw WrappedIOException.wrap(
+ "Caught exception in FloatSum.Final", exp);
+ }
+ }
+
+
+ if(sawNonNull) {
+ return new Double(sum);
+ } else {
+ return null;
+ }
} catch (ExecException ee) {
- IOException oughtToBeEE = new IOException();
- oughtToBeEE.initCause(ee);
- throw oughtToBeEE;
+ throw WrappedIOException.wrap("Caught exception in FloatSum.Final", ee);
}
}
}
Modified: incubator/pig/branches/types/src/org/apache/pig/builtin/IntSum.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/IntSum.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/IntSum.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/IntSum.java Mon Aug 4 12:28:37 2008
@@ -28,7 +28,7 @@
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.logicalLayer.schema.Schema;
-
+import org.apache.pig.impl.util.WrappedIOException;
/**
* Generates the sum of the Integer in the first field of a tuple.
@@ -40,9 +40,7 @@
try {
return sum(input);
} catch (ExecException ee) {
- IOException oughtToBeEE = new IOException();
- oughtToBeEE.initCause(ee);
- throw oughtToBeEE;
+ throw WrappedIOException.wrap("Caught exception in IntSum", ee);
}
}
@@ -66,9 +64,7 @@
try {
return tfact.newTuple(sum(input));
} catch (ExecException ee) {
- IOException oughtToBeEE = new IOException();
- oughtToBeEE.initCause(ee);
- throw oughtToBeEE;
+ throw WrappedIOException.wrap("Caught exception in IntSum.Initial", ee);
}
}
}
@@ -76,11 +72,39 @@
@Override
public Long exec(Tuple input) throws IOException {
try {
- return sum(input);
+ // Can't just call sum, because the intermediate results are
+ // now Longs insteads of Integers.
+ DataBag values = (DataBag)input.get(0);
+
+ // if we were handed an empty bag, return NULL
+ // this is in compliance with SQL standard
+ if(values.size() == 0) {
+ return null;
+ }
+
+ long sum = 0;
+ boolean sawNonNull = false;
+ for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
+ Tuple t = (Tuple) it.next();
+ try {
+ Long l = (Long)(t.get(0));
+ if (l == null) continue;
+ sawNonNull = true;
+ sum += l;
+ }catch(RuntimeException exp) {
+ throw WrappedIOException.wrap(
+ "Caught exception in IntSum.Final", exp);
+ }
+ }
+
+
+ if(sawNonNull) {
+ return new Long(sum);
+ } else {
+ return null;
+ }
} catch (ExecException ee) {
- IOException oughtToBeEE = new IOException();
- oughtToBeEE.initCause(ee);
- throw oughtToBeEE;
+ throw WrappedIOException.wrap("Caught exception in IntSum.Final", ee);
}
}
}
Modified: incubator/pig/branches/types/src/org/apache/pig/builtin/SUM.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/SUM.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/SUM.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/SUM.java Mon Aug 4 12:28:37 2008
@@ -32,6 +32,7 @@
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.util.WrappedIOException;
/**
@@ -44,9 +45,7 @@
try {
return sum(input);
} catch (ExecException ee) {
- IOException oughtToBeEE = new IOException();
- oughtToBeEE.initCause(ee);
- throw oughtToBeEE;
+ throw WrappedIOException.wrap("Caught exception in SUM", ee);
}
}
@@ -70,9 +69,7 @@
try {
return tfact.newTuple(sum(input));
} catch (ExecException ee) {
- IOException oughtToBeEE = new IOException();
- oughtToBeEE.initCause(ee);
- throw oughtToBeEE;
+ throw WrappedIOException.wrap("Caught exception in SUM.Initial", ee);
}
}
}
@@ -82,9 +79,7 @@
try {
return sum(input);
} catch (ExecException ee) {
- IOException oughtToBeEE = new IOException();
- oughtToBeEE.initCause(ee);
- throw oughtToBeEE;
+ throw WrappedIOException.wrap("Caught exception in SUM.Final", ee);
}
}
}
Modified: incubator/pig/branches/types/src/org/apache/pig/impl/logicalLayer/schema/Schema.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/logicalLayer/schema/Schema.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/logicalLayer/schema/Schema.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/logicalLayer/schema/Schema.java Mon Aug 4 12:28:37 2008
@@ -35,11 +35,11 @@
import org.apache.pig.impl.logicalLayer.FrontendException;
-public class Schema implements Serializable {
+public class Schema implements Serializable, Cloneable {
private static final long serialVersionUID = 2L;
- public static class FieldSchema implements Serializable{
+ public static class FieldSchema implements Serializable, Cloneable {
/**
*
*/
Modified: incubator/pig/branches/types/src/org/apache/pig/impl/plan/OperatorPlan.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/plan/OperatorPlan.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/plan/OperatorPlan.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/plan/OperatorPlan.java Mon Aug 4 12:28:37 2008
@@ -35,7 +35,18 @@
//import org.apache.commons.collections.map.MultiValueMap;
/**
- * A generic graphing class for use by LogicalPlan, PhysicalPlan, etc.
+ * A generic graphing class for use by LogicalPlan, PhysicalPlan, etc. One
+ * important aspect of this package is that it guarantees that once a graph is
+ * constructed, manipulations on that graph will maintain the ordering of
+ * inputs and outputs for a given node. That is, if a node has two inputs, 0
+ * and 1, it is guaranteed that everytime it asks for its inputs, it will
+ * receive them in the same order. This allows operators that need to
+ * distinguish their inputs (such as binary operators that need to know left
+ * from right) to work without needing to store their inputs themselves. This
+ * is an extra burden on the graph package and not in line with the way graphs
+ * are generally understood mathematically. But it greatly reducing the need
+ * for graph manipulators (such as the validators and optimizers) to
+ * understand the internals of various nodes.
*/
public abstract class OperatorPlan<E extends Operator> implements Iterable, Serializable {
protected Map<E, OperatorKey> mOps;
@@ -156,7 +167,7 @@
if (mToEdges.get(to) != null &&
!to.supportsMultipleInputs()) {
PlanException pe = new PlanException("Attempt to give operator of type " +
- from.getClass().getName() + " multiple inputs. This operator does "
+ to.getClass().getName() + " multiple inputs. This operator does "
+ "not support multiple inputs.");
log.error(pe.getMessage());
throw pe;
@@ -200,6 +211,47 @@
}
/**
+ * Trim everything below a given operator. The specified operator will
+ * NOT be removed.
+ * @param op Operator to trim everything after.
+ */
+ public void trimBelow(E op) {
+ trimBelow(getSuccessors(op));
+ }
+
+ private void trimBelow(List<E> ops) {
+ if (ops != null) {
+ // Make a copy because we'll be messing with the underlying list.
+ List<E> copy = new ArrayList<E>(ops);
+ for (E op : copy) {
+ trimBelow(getSuccessors(op));
+ remove(op);
+ }
+ }
+ }
+
+ /**
+ * Trim everything above a given operator. The specified operator will
+ * NOT be removed.
+ * @param op Operator to trim everything before.
+ */
+ public void trimAbove(E op) {
+ trimAbove(getPredecessors(op));
+ }
+
+ private void trimAbove(List<E> ops) {
+ if (ops != null) {
+ // Make a copy because we'll be messing with the underlying list.
+ List<E> copy = new ArrayList<E>(ops);
+ for (E op : copy) {
+ trimAbove(getPredecessors(op));
+ remove(op);
+ }
+ }
+ }
+
+
+ /**
* Find all of the nodes that have edges to the indicated node from
* themselves.
* @param op Node to look to
@@ -395,6 +447,51 @@
}
/**
+ * Replace an existing node in the graph with a new node. The new node
+ * will be connected to all the nodes the old node was. The old node will
+ * be removed.
+ * @param oldNode Node to be replaced
+ * @param newNode Node to add in place of oldNode
+ * @throws PlanException
+ */
+ public void replace(E oldNode, E newNode) throws PlanException {
+ checkInPlan(oldNode);
+ add(newNode);
+ mToEdges = generateNewMap(oldNode, newNode, mToEdges);
+ mFromEdges = generateNewMap(oldNode, newNode, mFromEdges);
+ remove(oldNode);
+ }
+
+ private MultiMap<E, E> generateNewMap(
+ E oldNode,
+ E newNode,
+ MultiMap<E, E> mm) {
+ // First, replace the key
+ Collection<E> targets = mm.get(oldNode);
+ if (targets != null) {
+ mm.removeKey(oldNode);
+ mm.put(newNode, targets);
+ }
+
+ // We can't just do a remove and add in the map because of our
+ // guarantee of not changing orders. So we need to walk the lists and
+ // put the new node in the same slot as the old.
+
+ // Walk all the other keys and replace any references to the oldNode
+ // in their targets.
+ MultiMap<E, E> newMap = new MultiMap<E, E>(mm.size());
+ for (E key : mm.keySet()) {
+ Collection<E> c = mm.get(key);
+ ArrayList<E> al = new ArrayList<E>(c);
+ for (int i = 0; i < al.size(); i++) {
+ if (al.get(i) == oldNode) al.set(i, newNode);
+ }
+ newMap.put(key, al);
+ }
+ return newMap;
+ }
+
+ /**
* Remove a node in a way that connects the node's predecessor (if any)
* with the node's successor (if any). This function does not handle the
* case where the node has multiple predecessors or successors.
Modified: incubator/pig/branches/types/src/org/apache/pig/impl/util/MultiMap.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/util/MultiMap.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/util/MultiMap.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/util/MultiMap.java Mon Aug 4 12:28:37 2008
@@ -38,7 +38,18 @@
*/
public class MultiMap<K, V> implements Serializable {
- private HashMap<K, ArrayList<V>> mMap = new HashMap<K, ArrayList<V>>();
+ private HashMap<K, ArrayList<V>> mMap = null;
+
+ public MultiMap() {
+ mMap = new HashMap<K, ArrayList<V>>();
+ }
+
+ /**
+ * @param size Initial size of the map
+ */
+ public MultiMap(int size) {
+ mMap = new HashMap<K, ArrayList<V>>(size);
+ }
/**
* Add an element to the map.
@@ -149,7 +160,27 @@
return values;
}
+ /**
+ * Get the number of entries in the map.
+ * @return number of entries.
+ */
+ public int size() {
+ return mMap.size();
+ }
+
+ public boolean isEmpty() {
+ return mMap.isEmpty();
+ }
+ public void clear() {
+ mMap.clear();
+ }
+ public boolean containsKey(K key) {
+ return mMap.containsKey(key);
+ }
+ public boolean containsValue(V val) {
+ return mMap.containsValue(val);
+ }
}
Added: incubator/pig/branches/types/src/org/apache/pig/impl/util/Pair.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/util/Pair.java?rev=682474&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/util/Pair.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/util/Pair.java Mon Aug 4 12:28:37 2008
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.pig.impl.util;
+
+import java.io.Serializable;
+
+/**
+ * Copy of C++ STL pair container.
+ */
+public class Pair<T, U> implements Serializable {
+ public T first;
+ public U second;
+
+ /**
+ * @param f First element in pair.
+ * @param s Second element in pair.
+ */
+ public Pair(T f, U s) {
+ first = f;
+ second = s;
+ }
+}
Modified: incubator/pig/branches/types/test/org/apache/pig/test/TestEvalPipeline.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/test/org/apache/pig/test/TestEvalPipeline.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/test/org/apache/pig/test/TestEvalPipeline.java (original)
+++ incubator/pig/branches/types/test/org/apache/pig/test/TestEvalPipeline.java Mon Aug 4 12:28:37 2008
@@ -386,10 +386,10 @@
while(iter.hasNext()){
Tuple t = iter.next();
assertEquals((Integer)numIdentity, (Integer)t.get(0));
- assertEquals((Long)5L, (Long)t.get(3));
- assertEquals(LOOP_COUNT*2.0, (Double)t.get(4), 0.01);
- assertEquals(8.0, (Double)t.get(6), 0.01);
- assertEquals(7, t.size());
+ assertEquals((Long)5L, (Long)t.get(2));
+ assertEquals(LOOP_COUNT*2.0, (Double)t.get(3), 0.01);
+ assertEquals(8.0, (Double)t.get(5), 0.01);
+ assertEquals(6, t.size());
++numIdentity;
}
assertEquals(LOOP_COUNT, numIdentity);
Modified: incubator/pig/branches/types/test/org/apache/pig/test/TestLogToPhyCompiler.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/test/org/apache/pig/test/TestLogToPhyCompiler.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/test/org/apache/pig/test/TestLogToPhyCompiler.java (original)
+++ incubator/pig/branches/types/test/org/apache/pig/test/TestLogToPhyCompiler.java Mon Aug 4 12:28:37 2008
@@ -185,6 +185,7 @@
}
public void testCogroup() throws VisitorException, IOException {
+ System.out.println("testCogroup");
String query = "cogroup (load 'a') by ($0 + $1, $0 - $1), (load 'b') by ($0 + $1, $0 - $1);";
LogicalPlan plan = buildPlan(query);
PhysicalPlan pp = buildPhysicalPlan(plan);
@@ -211,7 +212,7 @@
System.out.println(compiledPlan);
System.out.println("-------------");
//System.out.println(compiledPlan.compareTo(goldenPlan)==0);
- assertEquals(true, compiledPlan.compareTo(goldenPlan) == 0);
+ assertEquals(compiledPlan, goldenPlan);
}
public void testArithmetic() throws VisitorException, IOException, ExecException {
Modified: incubator/pig/branches/types/test/org/apache/pig/test/TestOperatorPlan.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/test/org/apache/pig/test/TestOperatorPlan.java?rev=682474&r1=682473&r2=682474&view=diff
==============================================================================
--- incubator/pig/branches/types/test/org/apache/pig/test/TestOperatorPlan.java (original)
+++ incubator/pig/branches/types/test/org/apache/pig/test/TestOperatorPlan.java Mon Aug 4 12:28:37 2008
@@ -834,6 +834,50 @@
assertFalse(transformer.mTransformed);
}
+ @Test
+ public void testReplace() throws Exception {
+ // Build a plan
+ TPlan plan = new TPlan();
+ TOperator[] ops = new TOperator[6];
+ ops[0] = new MultiOperator("1");
+ plan.add(ops[0]);
+ ops[1] = new MultiOperator("2");
+ plan.add(ops[1]);
+ ops[2] = new MultiOperator("3");
+ plan.add(ops[2]);
+ ops[3] = new MultiOperator("4");
+ plan.add(ops[3]);
+ ops[4] = new MultiOperator("5");
+ plan.add(ops[4]);
+ plan.connect(ops[0], ops[2]);
+ plan.connect(ops[1], ops[2]);
+ plan.connect(ops[2], ops[3]);
+ plan.connect(ops[2], ops[4]);
+ ops[5] = new MultiOperator("6");
+ plan.replace(ops[2], ops[5]);
+
+ assertEquals("Nodes: 1 2 4 5 6 FromEdges: 1->6 2->6 6->4 6->5 ToEdges: 4->6 5->6 6->1 6->2 ", plan.display());
+ }
+
+ @Test
+ public void testReplaceNoConnections() throws Exception {
+ // Build a plan
+ TPlan plan = new TPlan();
+ TOperator[] ops = new TOperator[4];
+ ops[0] = new MultiOperator("1");
+ plan.add(ops[0]);
+ ops[1] = new MultiOperator("2");
+ plan.add(ops[1]);
+ ops[2] = new MultiOperator("3");
+ plan.add(ops[2]);
+ plan.connect(ops[0], ops[2]);
+ ops[3] = new MultiOperator("4");
+ plan.replace(ops[1], ops[3]);
+
+ assertEquals("Nodes: 1 3 4 FromEdges: 1->3 ToEdges: 3->1 ", plan.display());
+ }
+
+
}