You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by sm...@apache.org on 2009/06/29 19:20:25 UTC
svn commit: r789388 [1/2] - in /hadoop/pig/trunk:
src/org/apache/pig/impl/logicalLayer/
src/org/apache/pig/impl/logicalLayer/optimizer/
src/org/apache/pig/impl/plan/ src/org/apache/pig/impl/plan/optimizer/
src/org/apache/pig/pen/util/ test/org/apache/p...
Author: sms
Date: Mon Jun 29 17:20:24 2009
New Revision: 789388
URL: http://svn.apache.org/viewvc?rev=789388&view=rev
Log:
PIG-697: Proposed improvements to pig's optimizer (sms)
Added:
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/CastFinder.java
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/UDFFinder.java
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/PushUpFilter.java
hadoop/pig/trunk/test/org/apache/pig/test/TestPushUpFilter.java
Modified:
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCogroup.java
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCross.java
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOFRJoin.java
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOForEach.java
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOLoad.java
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOUnion.java
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LogicalPlan.java
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/ProjectFixerUpper.java
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalOptimizer.java
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalTransformer.java
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/OpLimitOptimizer.java
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/StreamOptimizer.java
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/TypeCastInserter.java
hadoop/pig/trunk/src/org/apache/pig/impl/plan/OperatorPlan.java
hadoop/pig/trunk/src/org/apache/pig/impl/plan/ProjectionMap.java
hadoop/pig/trunk/src/org/apache/pig/impl/plan/optimizer/PlanOptimizer.java
hadoop/pig/trunk/src/org/apache/pig/impl/plan/optimizer/Transformer.java
hadoop/pig/trunk/src/org/apache/pig/pen/util/FunctionalLogicalOptimizer.java
hadoop/pig/trunk/test/org/apache/pig/test/TestOperatorPlan.java
hadoop/pig/trunk/test/org/apache/pig/test/TestProjectionMap.java
Added: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/CastFinder.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/CastFinder.java?rev=789388&view=auto
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/CastFinder.java (added)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/CastFinder.java Mon Jun 29 17:20:24 2009
@@ -0,0 +1,81 @@
+/*
+ * 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.logicalLayer;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.ArrayList;
+
+import org.apache.pig.FuncSpec;
+import org.apache.pig.impl.plan.DepthFirstWalker;
+import org.apache.pig.impl.plan.VisitorException;
+
+/**
+ * A visitor to track the casts in a plan.
+ */
+public class CastFinder extends LOVisitor {
+
+ List<LOCast> mCastList = new ArrayList<LOCast>();
+
+ /**
+ *
+ * @param plan
+ * logical plan to query the presence of UDFs
+ */
+ public CastFinder(LogicalPlan plan) {
+ super(plan, new DepthFirstWalker<LogicalOperator, LogicalPlan>(plan));
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see
+ * org.apache.pig.impl.logicalLayer.LOVisitor#visit(org.apache.pig.impl.
+ * logicalLayer.LOUserFunc)
+ */
+ @Override
+ protected void visit(LOCast cast) throws VisitorException {
+ mCastList.add(cast);
+ }
+
+ /**
+ *
+ * @return list of casts found in the plan
+ */
+ public List<LOCast> getCastList() {
+ return mCastList;
+ }
+
+ /**
+ *
+ * @return set of casts found in the plan
+ */
+ public Set<LOCast> getCastSet() {
+ return new HashSet<LOCast>(mCastList);
+ }
+
+ /**
+ *
+ * @return true if the plan contained any Casts; false otherwise
+ */
+ public boolean foundAnyCast() {
+ return (mCastList.size() == 0 ? false : true);
+ }
+
+}
\ No newline at end of file
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCogroup.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCogroup.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCogroup.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCogroup.java Mon Jun 29 17:20:24 2009
@@ -557,7 +557,7 @@
MultiMap<LogicalOperator, LogicalPlan> groupByPlans = getGroupByPlans();
boolean groupByAdded = false;
- MultiMap<Integer, Pair<Integer, Integer>> mapFields = new MultiMap<Integer, Pair<Integer, Integer>>();
+ MultiMap<Integer, ProjectionMap.Column> mapFields = new MultiMap<Integer, ProjectionMap.Column>();
List<Pair<Integer, Integer>> removedFields = new ArrayList<Pair<Integer, Integer>>();
for(int inputNum = 0; (inputNum < predecessors.size()) && (!groupByAdded); ++inputNum) {
@@ -575,10 +575,22 @@
if(leaves.get(0) instanceof LOProject) {
//find out if this project is a chain of projects
- LOProject topProject = LogicalPlan.chainOfProjects(predecessorPlan);
- if(topProject != null) {
- inputColumn = topProject.getCol();
- mapFields.put(0, new Pair<Integer, Integer>(inputNum, inputColumn));
+ Pair<LOProject, LOCast> pair = LogicalPlan.chainOfProjects(predecessorPlan);
+ if (pair != null) {
+ LOProject topProject = pair.first;
+ if (topProject != null) {
+ inputColumn = topProject.getCol();
+ LOCast cast = pair.second;
+ if (cast != null) {
+ mapFields.put(0,
+ new ProjectionMap.Column(
+ new Pair<Integer, Integer>(inputNum, inputColumn), true, cast.getType()
+ )
+ );
+ } else {
+ mapFields.put(0, new ProjectionMap.Column(new Pair<Integer, Integer>(inputNum, inputColumn)));
+ }
+ }
}
} else {
groupByAdded = true;
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCross.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCross.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCross.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCross.java Mon Jun 29 17:20:24 2009
@@ -214,7 +214,7 @@
return mProjectionMap;
}
- MultiMap<Integer, Pair<Integer, Integer>> mapFields = new MultiMap<Integer, Pair<Integer, Integer>>();
+ MultiMap<Integer, ProjectionMap.Column> mapFields = new MultiMap<Integer, ProjectionMap.Column>();
List<Integer> addedFields = new ArrayList<Integer>();
boolean[] unknownSchema = new boolean[predecessors.size()];
boolean anyUnknownInputSchema = false;
@@ -239,7 +239,7 @@
} else {
unknownSchema[inputNum] = false;
for(int inputColumn = 0; inputColumn < inputSchema.size(); ++inputColumn) {
- mapFields.put(outputColumnNum++, new Pair<Integer, Integer>(inputNum, inputColumn));
+ mapFields.put(outputColumnNum++, new ProjectionMap.Column(new Pair<Integer, Integer>(inputNum, inputColumn)));
}
}
}
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOFRJoin.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOFRJoin.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOFRJoin.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOFRJoin.java Mon Jun 29 17:20:24 2009
@@ -302,7 +302,7 @@
return mProjectionMap;
}
- MultiMap<Integer, Pair<Integer, Integer>> mapFields = new MultiMap<Integer, Pair<Integer, Integer>>();
+ MultiMap<Integer, ProjectionMap.Column> mapFields = new MultiMap<Integer, ProjectionMap.Column>();
List<Integer> addedFields = new ArrayList<Integer>();
boolean[] unknownSchema = new boolean[predecessors.size()];
boolean anyUnknownInputSchema = false;
@@ -327,7 +327,8 @@
} else {
unknownSchema[inputNum] = false;
for(int inputColumn = 0; inputColumn < inputSchema.size(); ++inputColumn) {
- mapFields.put(outputColumnNum++, new Pair<Integer, Integer>(inputNum, inputColumn));
+ mapFields.put(outputColumnNum++,
+ new ProjectionMap.Column(new Pair<Integer, Integer>(inputNum, inputColumn)));
}
}
}
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOForEach.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOForEach.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOForEach.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOForEach.java Mon Jun 29 17:20:24 2009
@@ -492,7 +492,7 @@
List<LogicalPlan> foreachPlans = getForEachPlans();
List<Boolean> flattenList = getFlatten();
- MultiMap<Integer, Pair<Integer, Integer>> mapFields = new MultiMap<Integer, Pair<Integer, Integer>>();
+ MultiMap<Integer, ProjectionMap.Column> mapFields = new MultiMap<Integer, ProjectionMap.Column>();
List<Integer> addedFields = new ArrayList<Integer>();
int outputColumn = 0;
@@ -506,14 +506,19 @@
int inputColumn = -1;
boolean mapped = false;
+ LOCast cast = null;
if(leaves.get(0) instanceof LOProject || leaves.get(0) instanceof LOCast) {
//find out if this project is a chain of projects
- LOProject topProject = LogicalPlan.chainOfProjects(foreachPlan);
- if(topProject != null) {
- inputColumn = topProject.getCol();
+ Pair<LOProject, LOCast> pair = LogicalPlan.chainOfProjects(foreachPlan);
+ if (pair != null) {
+ LOProject topProject = pair.first;
+ cast = pair.second;
+ if (topProject != null) {
+ inputColumn = topProject.getCol();
mapped = true;
+ }
}
}
@@ -567,7 +572,17 @@
for(int j = 0; j < innerSchema.size(); ++j) {
if(mapped) {
//map each flattened column to the original column
- mapFields.put(outputColumn++, new Pair<Integer, Integer>(0, inputColumn));
+ if (cast != null) {
+ mapFields.put(outputColumn++,
+ new ProjectionMap.Column(
+ new Pair<Integer, Integer>(0, inputColumn), true, cast.getType()
+ )
+ );
+ } else {
+ mapFields.put(outputColumn++,
+ new ProjectionMap.Column(new Pair<Integer, Integer>(0, inputColumn))
+ );
+ }
} else {
addedFields.add(outputColumn++);
}
@@ -578,12 +593,32 @@
//flattening a null schema results in a bytearray
if(mapped) {
//map each flattened column to the original column
- mapFields.put(outputColumn++, new Pair<Integer, Integer>(0, inputColumn));
+ if (cast != null) {
+ mapFields.put(outputColumn++,
+ new ProjectionMap.Column(
+ new Pair<Integer, Integer>(0, inputColumn), true, cast.getType()
+ )
+ );
+ } else {
+ mapFields.put(outputColumn++,
+ new ProjectionMap.Column(new Pair<Integer, Integer>(0, inputColumn))
+ );
+ }
} else {
addedFields.add(outputColumn++);
}
} else {
- mapFields.put(outputColumn++, new Pair<Integer, Integer>(0, inputColumn));
+ if (cast != null) {
+ mapFields.put(outputColumn++,
+ new ProjectionMap.Column(
+ new Pair<Integer, Integer>(0, inputColumn), true, cast.getType()
+ )
+ );
+ } else {
+ mapFields.put(outputColumn++,
+ new ProjectionMap.Column(new Pair<Integer, Integer>(0, inputColumn))
+ );
+ }
}
}
} else {
@@ -592,18 +627,48 @@
//flattening a null schema results in a bytearray
if(mapped) {
//map each flattened column to the original column
- mapFields.put(outputColumn++, new Pair<Integer, Integer>(0, inputColumn));
+ if (cast != null) {
+ mapFields.put(outputColumn++,
+ new ProjectionMap.Column(
+ new Pair<Integer, Integer>(0, inputColumn), true, cast.getType()
+ )
+ );
+ } else {
+ mapFields.put(outputColumn++,
+ new ProjectionMap.Column(new Pair<Integer, Integer>(0, inputColumn))
+ );
+ }
} else {
addedFields.add(outputColumn++);
}
} else {
- mapFields.put(outputColumn++, new Pair<Integer, Integer>(0, inputColumn));
+ if (cast != null) {
+ mapFields.put(outputColumn++,
+ new ProjectionMap.Column(
+ new Pair<Integer, Integer>(0, inputColumn), true, cast.getType()
+ )
+ );
+ } else {
+ mapFields.put(outputColumn++,
+ new ProjectionMap.Column(new Pair<Integer, Integer>(0, inputColumn))
+ );
+ }
}
}
} else {
//not a flattened column
if(mapped) {
- mapFields.put(outputColumn++, new Pair<Integer, Integer>(0, inputColumn));
+ if (cast != null) {
+ mapFields.put(outputColumn++,
+ new ProjectionMap.Column(
+ new Pair<Integer, Integer>(0, inputColumn), true, cast.getType()
+ )
+ );
+ } else {
+ mapFields.put(outputColumn++,
+ new ProjectionMap.Column(new Pair<Integer, Integer>(0, inputColumn))
+ );
+ }
} else {
addedFields.add(outputColumn++);
}
@@ -637,9 +702,9 @@
if(mapFields != null) {
Set<Integer> mappedSet = new HashSet<Integer>();
for(Integer key: mapFields.keySet()) {
- List<Pair<Integer, Integer>> values = (ArrayList<Pair<Integer, Integer>>)mapFields.get(key);
- for(Pair<Integer, Integer> value: values) {
- mappedSet.add(value.second);
+ List<ProjectionMap.Column> values = (ArrayList<ProjectionMap.Column>) mapFields.get(key);
+ for (ProjectionMap.Column value : values) {
+ mappedSet.add(value.getInputColumn().second);
}
}
removedSet.removeAll(mappedSet);
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOLoad.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOLoad.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOLoad.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOLoad.java Mon Jun 29 17:20:24 2009
@@ -275,10 +275,10 @@
mProjectionMap = new ProjectionMap(false);
return mProjectionMap;
} else {
- MultiMap<Integer, Pair<Integer, Integer>> mapFields = new MultiMap<Integer, Pair<Integer, Integer>>();
+ MultiMap<Integer, ProjectionMap.Column> mapFields = new MultiMap<Integer, ProjectionMap.Column>();
//compute the mapping assuming its a prefix projection
for(int i = 0; i < inputSchema.size(); ++i) {
- mapFields.put(i, new Pair<Integer, Integer>(0, i));
+ mapFields.put(i, new ProjectionMap.Column(new Pair<Integer, Integer>(0, i)));
}
mProjectionMap = new ProjectionMap(mapFields, null, null);
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOUnion.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOUnion.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOUnion.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOUnion.java Mon Jun 29 17:20:24 2009
@@ -161,7 +161,7 @@
return mProjectionMap;
}
- MultiMap<Integer, Pair<Integer, Integer>> mapFields = new MultiMap<Integer, Pair<Integer, Integer>>();
+ MultiMap<Integer, ProjectionMap.Column> mapFields = new MultiMap<Integer, ProjectionMap.Column>();
for(int inputNum = 0; inputNum < predecessors.size(); ++inputNum) {
LogicalOperator predecessor = predecessors.get(inputNum);
@@ -179,8 +179,7 @@
return mProjectionMap;
} else {
for(int inputColumn = 0; inputColumn < inputSchema.size(); ++inputColumn) {
- mapFields.put(inputColumn, new Pair<Integer, Integer>(inputNum, inputColumn));
- //removedFields.add(new Pair<Integer, Integer>(inputNum, inputColumn));
+ mapFields.put(inputColumn, new ProjectionMap.Column(new Pair<Integer, Integer>(inputNum, inputColumn)));
}
}
}
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LogicalPlan.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LogicalPlan.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LogicalPlan.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LogicalPlan.java Mon Jun 29 17:20:24 2009
@@ -32,6 +32,7 @@
import org.apache.pig.impl.plan.OperatorPlan;
import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.impl.plan.PlanException;
+import org.apache.pig.impl.util.Pair;
public class LogicalPlan extends OperatorPlan<LogicalOperator> {
private static final long serialVersionUID = 2L;
@@ -156,12 +157,14 @@
*
* @param plan
* input plan
- * @return the top most projection operator if there is a chain of
- * projection operators with or without casts; null otherwise
+ * @return a pair consisting of the top most projection operator and the
+ * cast (if present) if there is a chain of projection operators
+ * with or without casts; null otherwise
*/
- public static LOProject chainOfProjects(LogicalPlan plan) {
+ public static Pair<LOProject, LOCast> chainOfProjects(LogicalPlan plan) {
LOProject topProject = null;
+ LOCast cast = null;
if (plan == null) {
return null;
@@ -181,14 +184,14 @@
while (true) {
if (node == null) {
- //a node cannot be null
+ // a node cannot be null
return null;
}
if (node instanceof LOProject) {
topProject = (LOProject) node;
} else if (node instanceof LOCast) {
- // continue
+ cast = (LOCast) node;
} else {
// not a projection or a cast return null
return null;
@@ -198,7 +201,7 @@
if (predecessors == null) {
// we have reached the root
- return topProject;
+ return new Pair<LOProject, LOCast>(topProject, cast);
}
if (predecessors.size() > 1) {
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/ProjectFixerUpper.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/ProjectFixerUpper.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/ProjectFixerUpper.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/ProjectFixerUpper.java Mon Jun 29 17:20:24 2009
@@ -144,7 +144,7 @@
return;
}
- MultiMap<Integer, Pair<Integer, Integer>> oldNodeMappedFields = oldNodeMap
+ MultiMap<Integer, ProjectionMap.Column> oldNodeMappedFields = oldNodeMap
.getMappedFields();
if (oldNodeMappedFields == null) {
// there is no mapping available bail out
@@ -155,9 +155,9 @@
}
- List<Pair<Integer, Integer>> pairs = (List<Pair<Integer, Integer>>) oldNodeMappedFields
- .get(oldNodeColumn);
- if (pairs == null) {
+ List<ProjectionMap.Column> columns = (List<ProjectionMap.Column>) oldNodeMappedFields.get(oldNodeColumn);
+
+ if (columns == null) {
// there is no mapping for oldNodeColumn
// it could be an added field; bail out
int errCode = 2148;
@@ -167,7 +167,8 @@
}
boolean foundMapping = false;
- for (Pair<Integer, Integer> pair : pairs) {
+ for (ProjectionMap.Column column : columns) {
+ Pair<Integer, Integer> pair = column.getInputColumn();
if (pair.first.equals(mPredecessorIndex)) {
ArrayList<Integer> newColumns = new ArrayList<Integer>();
newColumns.add(pair.second);
@@ -209,7 +210,7 @@
return;
}
- MultiMap<Integer, Pair<Integer, Integer>> newNodeMappedFields = newNodeMap
+ MultiMap<Integer, ProjectionMap.Column> newNodeMappedFields = newNodeMap
.getMappedFields();
if (newNodeMappedFields == null) {
// there is no mapping available bail out
@@ -223,9 +224,9 @@
boolean foundMapping = false;
for (Integer key : newNodeMappedFields.keySet()) {
- List<Pair<Integer, Integer>> pairs = (List<Pair<Integer, Integer>>) newNodeMappedFields
+ List<ProjectionMap.Column> columns = (List<ProjectionMap.Column>) newNodeMappedFields
.get(key);
- if (pairs == null) {
+ if (columns == null) {
// should not happen
int errCode = 2152;
String msg = "Error during fixing projections. Could not locate mapping for column: "
@@ -234,7 +235,8 @@
PigException.BUG);
}
- for (Pair<Integer, Integer> pair : pairs) {
+ for (ProjectionMap.Column column : columns) {
+ Pair<Integer, Integer> pair = column.getInputColumn();
if (pair.first.equals(mPredecessorIndex)
&& pair.second.equals(oldNodeColumn)) {
ArrayList<Integer> newColumns = new ArrayList<Integer>();
Added: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/UDFFinder.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/UDFFinder.java?rev=789388&view=auto
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/UDFFinder.java (added)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/UDFFinder.java Mon Jun 29 17:20:24 2009
@@ -0,0 +1,81 @@
+/*
+ * 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.logicalLayer;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.ArrayList;
+
+import org.apache.pig.FuncSpec;
+import org.apache.pig.impl.plan.DepthFirstWalker;
+import org.apache.pig.impl.plan.VisitorException;
+
+/**
+ * A visitor to track the UDFs in a plan.
+ */
+public class UDFFinder extends LOVisitor {
+
+ List<FuncSpec> mUDFList = new ArrayList<FuncSpec>();
+
+ /**
+ *
+ * @param plan
+ * logical plan to query the presence of UDFs
+ */
+ public UDFFinder(LogicalPlan plan) {
+ super(plan, new DepthFirstWalker<LogicalOperator, LogicalPlan>(plan));
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see
+ * org.apache.pig.impl.logicalLayer.LOVisitor#visit(org.apache.pig.impl.
+ * logicalLayer.LOUserFunc)
+ */
+ @Override
+ protected void visit(LOUserFunc udf) throws VisitorException {
+ mUDFList.add(udf.getFuncSpec());
+ }
+
+ /**
+ *
+ * @return list of function specifications found in the plan
+ */
+ public List<FuncSpec> getUDFList() {
+ return mUDFList;
+ }
+
+ /**
+ *
+ * @return set of function specifications found in the plan
+ */
+ public Set<FuncSpec> getUDFSet() {
+ return new HashSet<FuncSpec>(mUDFList);
+ }
+
+ /**
+ *
+ * @return true if the plan had any UDFs; false otherwise
+ */
+ public boolean foundAnyUDF() {
+ return (mUDFList.size() == 0 ? false : true);
+ }
+
+}
\ No newline at end of file
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalOptimizer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalOptimizer.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalOptimizer.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalOptimizer.java Mon Jun 29 17:20:24 2009
@@ -23,6 +23,7 @@
import org.apache.pig.ExecType;
import org.apache.pig.PigException;
import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.logicalLayer.LOFilter;
import org.apache.pig.impl.logicalLayer.LOLimit;
import org.apache.pig.impl.logicalLayer.LOLoad;
import org.apache.pig.impl.logicalLayer.LOPrinter;
@@ -107,24 +108,35 @@
mRules.add(new Rule<LogicalOperator, LogicalPlan>(rulePlan, new TypeCastInserter(plan,
LOStream.class.getName()), "StreamTypeCastInserter"));
- // Optimize when LOAD precedes STREAM and the loader class
- // is the same as the serializer for the STREAM.
- // Similarly optimize when STREAM is followed by store and the
- // deserializer class is same as the Storage class.
if(!turnAllRulesOff) {
- Rule rule = new Rule<LogicalOperator, LogicalPlan>(rulePlan, new StreamOptimizer(plan,
- LOStream.class.getName()), "StreamOptimizer");
+ /*
+ * Optimize when LOAD precedes STREAM and the loader class is the
+ * same as the serializer for the STREAM. Similarly optimize when
+ * STREAM is followed by store and the deserializer class is same as
+ * the Storage class.
+ */
+
+ Rule rule = new Rule<LogicalOperator, LogicalPlan>(rulePlan,
+ new StreamOptimizer(plan, LOStream.class.getName()),
+ "StreamOptimizer");
checkAndAddRule(rule);
- }
- // Push up limit where ever possible.
- if(!turnAllRulesOff) {
+ // Push up limit wherever possible.
rulePlan = new RulePlan();
- RuleOperator loLimit = new RuleOperator(LOLimit.class,
+ RuleOperator loLimit = new RuleOperator(LOLimit.class,
+ new OperatorKey(SCOPE, nodeIdGen.getNextNodeId(SCOPE)));
+ rulePlan.add(loLimit);
+ rule = new Rule<LogicalOperator, LogicalPlan>(rulePlan,
+ new OpLimitOptimizer(plan, mode), "LimitOptimizer");
+ checkAndAddRule(rule);
+
+ // Push filters up wherever possible
+ rulePlan = new RulePlan();
+ RuleOperator loFilter = new RuleOperator(LOFilter.class,
new OperatorKey(SCOPE, nodeIdGen.getNextNodeId(SCOPE)));
- rulePlan.add(loLimit);
- Rule rule = new Rule<LogicalOperator, LogicalPlan>(rulePlan,
- new OpLimitOptimizer(plan, mode), "LimitOptimizer");
+ rulePlan.add(loFilter);
+ rule = new Rule<LogicalOperator, LogicalPlan>(rulePlan,
+ new PushUpFilter(plan), "PushUpFilter");
checkAndAddRule(rule);
}
@@ -178,6 +190,7 @@
}
}
+ rule.getTransformer().reset();
}
}
}
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalTransformer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalTransformer.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalTransformer.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalTransformer.java Mon Jun 29 17:20:24 2009
@@ -270,5 +270,10 @@
// Now rebuild the schemas
// rebuildSchemas();
}
-
+
+ @Override
+ public void reset() {
+ // do nothing
+ }
+
}
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/OpLimitOptimizer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/OpLimitOptimizer.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/OpLimitOptimizer.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/OpLimitOptimizer.java Mon Jun 29 17:20:24 2009
@@ -77,7 +77,9 @@
LogicalOperator lo = nodes.get(0);
if (lo == null || !(lo instanceof LOLimit)) {
int errCode = 2005;
- String msg = "Expected " + LOLimit.class.getSimpleName() + ", got " + lo.getClass().getSimpleName();
+ String msg = "Expected " + LOLimit.class.getSimpleName()
+ + ", got "
+ + (lo == null ? lo : lo.getClass().getSimpleName());
throw new OptimizerException(msg, errCode, PigException.BUG);
}
} catch (Exception e) {
@@ -168,7 +170,7 @@
else
{
int errCode = 2010;
- String msg = "LOFilter should have one input";
+ String msg = "LOForEach should have one input";
throw new OptimizerException(msg, errCode, PigException.BUG);
}
// we can move LOLimit even further, recursively optimize LOLimit
Added: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/PushUpFilter.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/PushUpFilter.java?rev=789388&view=auto
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/PushUpFilter.java (added)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/PushUpFilter.java Mon Jun 29 17:20:24 2009
@@ -0,0 +1,425 @@
+/*
+ * 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.logicalLayer.optimizer;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.pig.data.DataType;
+import org.apache.pig.impl.logicalLayer.CastFinder;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.logicalLayer.LOCast;
+import org.apache.pig.impl.logicalLayer.LOCogroup;
+import org.apache.pig.impl.logicalLayer.LOCross;
+import org.apache.pig.impl.logicalLayer.LOFRJoin;
+import org.apache.pig.impl.logicalLayer.LOFilter;
+import org.apache.pig.impl.logicalLayer.LOForEach;
+import org.apache.pig.impl.logicalLayer.LOLimit;
+import org.apache.pig.impl.logicalLayer.LOLoad;
+import org.apache.pig.impl.logicalLayer.LOSplit;
+import org.apache.pig.impl.logicalLayer.LOStore;
+import org.apache.pig.impl.logicalLayer.LOProject;
+import org.apache.pig.impl.logicalLayer.LOStream;
+import org.apache.pig.impl.logicalLayer.LOSplitOutput;
+import org.apache.pig.impl.logicalLayer.LOUnion;
+import org.apache.pig.impl.logicalLayer.LogicalOperator;
+import org.apache.pig.impl.logicalLayer.LogicalPlan;
+import org.apache.pig.impl.logicalLayer.UDFFinder;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.plan.DepthFirstWalker;
+import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.ProjectionMap;
+import org.apache.pig.impl.plan.RequiredFields;
+import org.apache.pig.impl.plan.optimizer.OptimizerException;
+import org.apache.pig.FuncSpec;
+import org.apache.pig.PigException;
+import org.apache.pig.impl.streaming.StreamingCommand;
+import org.apache.pig.impl.streaming.StreamingCommand.HandleSpec;
+import org.apache.pig.impl.util.MultiMap;
+import org.apache.pig.impl.util.Pair;
+
+/**
+ * A visitor to discover if a filter can be pushed as high up the tree as
+ * possible.
+ */
+public class PushUpFilter extends LogicalTransformer {
+
+ // boolean to remember if the filter has to be swapped
+ private boolean mSwap = false;
+
+ // boolean to remember if the filter has to be pushed into one of the
+ // filter's predecessor's inputs
+ private boolean mPushBefore = false;
+
+ // the input of the predecessor where the filter has to be pushed
+ private int mPushBeforeInput = -1;
+
+ public PushUpFilter(LogicalPlan plan) {
+ super(plan, new DepthFirstWalker<LogicalOperator, LogicalPlan>(plan));
+ }
+
+ /**
+ *
+ * @return true if the filter has to swapped; false otherwise
+ */
+ public boolean getSwap() {
+ return mSwap;
+ }
+
+ /**
+ *
+ * @return true if the filter has to be pushed before its predecessor; false
+ * otherwise
+ */
+ public boolean getPushBefore() {
+ return mPushBefore;
+ }
+
+ /**
+ *
+ * @return return the input of the predecessor where the filter has to be
+ * pushed
+ */
+ public int getPushBeforeInput() {
+ return mPushBeforeInput;
+ }
+
+ @Override
+ public boolean check(List<LogicalOperator> nodes) throws OptimizerException {
+ try {
+ LOFilter filter = (LOFilter) getOperator(nodes);
+ List<LogicalOperator> predecessors = (mPlan.getPredecessors(filter) == null ? null
+ : new ArrayList<LogicalOperator>(mPlan
+ .getPredecessors(filter)));
+
+ // if there are no predecessors return false
+ if (predecessors == null) {
+ return false;
+ }
+
+ // if the filter has no predecessors or more than one predecessor
+ // return false
+ if (predecessors.size() == 0 || predecessors.size() > 1) {
+ return false;
+ }
+
+ LogicalOperator predecessor = predecessors.get(0);
+
+ // if the predecessor is one of LOLoad/LOStore/LOStream/LOLimit
+ // return false
+ if (predecessor instanceof LOLoad || predecessor instanceof LOStore
+ || predecessor instanceof LOStream
+ || predecessor instanceof LOLimit) {
+ return false;
+ }
+
+ // TODO
+ // for now filters cannot be combined
+ // remove this check when filters can be combined
+ if (predecessor instanceof LOFilter)
+ return false;
+
+ // TODO
+ // same rule as filters
+ if (predecessor instanceof LOSplitOutput) {
+ return false;
+ }
+ if (predecessor instanceof LOSplit) {
+ return false;
+ }
+
+ UDFFinder udfFinder = new UDFFinder(filter.getComparisonPlan());
+ udfFinder.visit();
+
+ // if the filter's inner plan contains any UDF then return false
+ if (udfFinder.foundAnyUDF()) {
+ return false;
+ }
+
+ CastFinder castFinder = new CastFinder(filter.getComparisonPlan());
+ castFinder.visit();
+
+ // if the filter's inner plan contains any casts then return false
+ if (castFinder.foundAnyCast()) {
+ return false;
+ }
+
+ List<RequiredFields> filterRequiredFields = filter
+ .getRequiredFields();
+ if (filterRequiredFields == null) {
+ return false;
+ }
+ RequiredFields requiredField = filterRequiredFields.get(0);
+
+ // the filter's conditions contain constant expression
+ // return false
+ if (requiredField.needNoFields()) {
+ return false;
+ }
+
+ // if the predecessor is a multi-input operator then detailed
+ // checks are required
+ if (predecessor instanceof LOCross
+ || predecessor instanceof LOUnion
+ || predecessor instanceof LOCogroup
+ || predecessor instanceof LOFRJoin) {
+
+ // check if the filter's required fields in conjunction with the
+ // predecessor's projection map. If the filter needs more than
+ // one input then the filter's expressions have to be split
+
+ List<LogicalOperator> grandParents = mPlan
+ .getPredecessors(predecessor);
+
+ // if the predecessor does not have predecessors return false
+ if (grandParents == null || grandParents.size() == 0) {
+ return false;
+ }
+
+ // check if the predecessor is a group by
+ if (grandParents.size() == 1) {
+ if (predecessor instanceof LOCogroup) {
+ mSwap = true;
+ return true;
+ } else {
+ // only a group by can have a single input
+ return false;
+ }
+ }
+
+ if (requiredField.needAllFields()) {
+ return false;
+ }
+
+ Pair<Boolean, Set<Integer>> mappingResult = isRequiredFieldMapped(requiredField, predecessor.getProjectionMap());
+ boolean mapped = mappingResult.first;
+ Set<Integer> grandParentIndexes = mappingResult.second;
+ if (!mapped) {
+ return false;
+ }
+
+ // TODO
+ // the filter's conditions requires more than one input of its
+ // predecessor
+ // when the filter's conditions are splittable return true
+ if ((grandParentIndexes == null)
+ || (grandParentIndexes.size() == 0)
+ || (grandParentIndexes.size() > 1)) {
+ return false;
+ }
+
+ if (predecessor instanceof LOCogroup) {
+ // check for outer
+ if (isAnyOuter((LOCogroup) predecessor)) {
+ return false;
+ }
+ }
+ mPushBefore = true;
+ mPushBeforeInput = grandParentIndexes.iterator().next();
+ return true;
+
+ } else if (predecessor instanceof LOForEach) {
+
+ LOForEach loForEach = (LOForEach) predecessor;
+ List<Boolean> mFlatten = loForEach.getFlatten();
+ boolean hasFlatten = false;
+ for (Boolean b : mFlatten) {
+ if (b.equals(true)) {
+ hasFlatten = true;
+ }
+ }
+
+ // TODO
+ // A better check is to examine each column in the filter's
+ // required fields. If the column is the result of a flatten
+ // then
+ // return false else return true
+
+ // for now if the foreach has a flatten then return false
+ if (hasFlatten) {
+ return false;
+ }
+
+ Pair<Boolean, Set<Integer>> mappingResult = isRequiredFieldMapped(requiredField, predecessor.getProjectionMap());
+ boolean mapped = mappingResult.first;
+ if (!mapped) {
+ return false;
+ }
+ }
+
+ mSwap = true;
+ return true;
+ } catch (OptimizerException oe) {
+ throw oe;
+ } catch (Exception e) {
+ int errCode = 2149;
+ String msg = "Internal error while trying to check if filters can be pushed up.";
+ throw new OptimizerException(msg, errCode, PigException.BUG, e);
+ }
+ }
+
+ private LogicalOperator getOperator(List<LogicalOperator> nodes)
+ throws FrontendException {
+ if ((nodes == null) || (nodes.size() <= 0)) {
+ int errCode = 2052;
+ String msg = "Internal error. Cannot retrieve operator from null or empty list.";
+ throw new OptimizerException(msg, errCode, PigException.BUG);
+ }
+
+ LogicalOperator lo = nodes.get(0);
+ if (lo == null || !(lo instanceof LOFilter)) {
+ // we should never be called with any other operator class name
+ int errCode = 1101;
+ String msg = "Expected " + LOFilter.class.getSimpleName()
+ + ", got "
+ + (lo == null ? lo : lo.getClass().getSimpleName());
+ throw new OptimizerException(msg, errCode, PigException.INPUT);
+ } else {
+ return lo;
+ }
+
+ }
+
+ @Override
+ public void transform(List<LogicalOperator> nodes)
+ throws OptimizerException {
+ try {
+ LOFilter filter = (LOFilter) getOperator(nodes);
+ LogicalOperator predecessor = mPlan.getPredecessors(filter).get(0);
+ if (mSwap) {
+ mPlan.swap(predecessor, filter);
+ } else if (mPushBefore) {
+ if (mPushBeforeInput == -1) {
+ // something is wrong!
+ int errCode = 2150;
+ String msg = "Internal error. The push before input is not set.";
+ throw new OptimizerException(msg, errCode, PigException.BUG);
+ }
+ mPlan.pushBefore(predecessor, filter, mPushBeforeInput);
+ }
+ } catch (OptimizerException oe) {
+ throw oe;
+ } catch (Exception e) {
+ int errCode = 2151;
+ String msg = "Internal error while pushing filters up.";
+ throw new OptimizerException(msg, errCode, PigException.BUG, e);
+ }
+ }
+
+ @Override
+ public void reset() {
+ mPushBefore = false;
+ mPushBeforeInput = -1;
+ mSwap = false;
+ }
+
+ /**
+ *
+ * A method to check if there any grouping column has the outer clause in a
+ * grouping operator
+ *
+ * @param cogroup
+ * the cogroup operator to be examined for presence of outer
+ * clause
+ * @return true if the cogroup contains any input that has an outer clause;
+ * false otherwise
+ */
+ private boolean isAnyOuter(LOCogroup cogroup) {
+ boolean[] innerList = cogroup.getInner();
+ for (boolean inner : innerList) {
+ if (!inner) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * A method to check if the required field contains elements that are mapped
+ * in the predecessor's inputs without a cast
+ *
+ * @param requiredField
+ * the required field of the operator
+ * @param predProjectionMap
+ * the projection map of the predecessor
+ * @return a pair of boolean and a set of integers; the first element of the
+ * pair is true if the field is mapped without a cast; false
+ * otherwise; the second element of the pair is the set of
+ * predecessor's inputs that are required for the mapping
+ */
+ private Pair<Boolean, Set<Integer>> isRequiredFieldMapped(RequiredFields requiredField,
+ ProjectionMap predProjectionMap) {
+
+ if(requiredField == null) {
+ return new Pair<Boolean, Set<Integer>>(false, null);
+ }
+
+ // if predecessor projection map is null then return false
+ if (predProjectionMap == null) {
+ return new Pair<Boolean, Set<Integer>>(false, null);
+ }
+
+ // if the predecessor does not change its output return true
+ if (!predProjectionMap.changes()) {
+ return new Pair<Boolean, Set<Integer>>(true, null);
+ }
+
+ MultiMap<Integer, ProjectionMap.Column> mappedFields = predProjectionMap
+ .getMappedFields();
+ // if there is no mapping in the predecessor then return false
+ if (mappedFields == null) {
+ return new Pair<Boolean, Set<Integer>>(false, null);
+ }
+
+ Set<Integer> predInputs = new HashSet<Integer>();
+ for (Pair<Integer, Integer> pair : requiredField.getFields()) {
+ predInputs.add(pair.second);
+ }
+
+ boolean mapped = false;
+ Set<Integer> grandParentIndexes = new HashSet<Integer>();
+
+ for (Integer input : predInputs) {
+ List<ProjectionMap.Column> inputList = (List<ProjectionMap.Column>) mappedFields
+ .get(input);
+ for (ProjectionMap.Column column : inputList) {
+ // TODO
+ // Check if the column has a cast
+ // if a cast is not used then consider it as mapped
+ // in the future this should go away and the cast
+ // type should be used to move around the projections
+ if (!column.cast()) {
+ mapped = true;
+ }
+ Pair<Integer, Integer> pair = column.getInputColumn();
+ grandParentIndexes.add(pair.first);
+ }
+ }
+
+ if (!mapped) {
+ return new Pair<Boolean, Set<Integer>>(false, null);
+ }
+
+ return new Pair<Boolean, Set<Integer>>(true, grandParentIndexes);
+ }
+}
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/StreamOptimizer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/StreamOptimizer.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/StreamOptimizer.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/StreamOptimizer.java Mon Jun 29 17:20:24 2009
@@ -75,7 +75,9 @@
LogicalOperator lo = nodes.get(0);
if (lo == null || !(lo instanceof LOStream)) {
int errCode = 2005;
- String msg = "Expected " + LOStream.class.getSimpleName() + ", got " + lo.getClass().getSimpleName();
+ String msg = "Expected " + LOStream.class.getSimpleName()
+ + ", got "
+ + (lo == null ? lo : lo.getClass().getSimpleName());
throw new OptimizerException(msg, errCode, PigException.BUG);
}
LOStream stream = (LOStream)lo;
@@ -231,6 +233,12 @@
throw new OptimizerException(msg, errCode, PigException.BUG, e);
}
}
+
+ @Override
+ public void reset() {
+ mOptimizeLoad = false;
+ mOptimizeStore = false;
+ }
}
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/TypeCastInserter.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/TypeCastInserter.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/TypeCastInserter.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/TypeCastInserter.java Mon Jun 29 17:20:24 2009
@@ -108,7 +108,9 @@
if(LOLoad.class.getName().equals(operatorClassName)) {
if (lo == null || !(lo instanceof LOLoad)) {
int errCode = 2005;
- String msg = "Expected " + LOLoad.class.getSimpleName() + ", got " + lo.getClass().getSimpleName();
+ String msg = "Expected " + LOLoad.class.getSimpleName()
+ + ", got "
+ + (lo == null ? lo : lo.getClass().getSimpleName());
throw new OptimizerException(msg, errCode, PigException.BUG);
}
@@ -116,7 +118,9 @@
} else if(LOStream.class.getName().equals(operatorClassName)){
if (lo == null || !(lo instanceof LOStream)) {
int errCode = 2005;
- String msg = "Expected " + LOStream.class.getSimpleName() + ", got " + lo.getClass().getSimpleName();
+ String msg = "Expected " + LOStream.class.getSimpleName()
+ + ", got "
+ + (lo == null ? lo : lo.getClass().getSimpleName());
throw new OptimizerException(msg, errCode, PigException.BUG);
}
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/plan/OperatorPlan.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/plan/OperatorPlan.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/plan/OperatorPlan.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/plan/OperatorPlan.java Mon Jun 29 17:20:24 2009
@@ -1128,7 +1128,7 @@
throw new PlanException(msg, errCode, PigException.INPUT);
}
- List<E> firstNodeSuccessors = (ArrayList<E>)mFromEdges.get(firstNode);
+ List<E> firstNodeSuccessors = (mFromEdges.get(firstNode) == null? null : new ArrayList<E>(mFromEdges.get(firstNode)));
if(firstNodeSuccessors == null) {
int errCode = 1088;
@@ -1217,10 +1217,8 @@
//Rewire is used within reconnectPredecessors. However, rewire is explicitly turned off in insertBetween
//The rewiring is done explicitly here to avoid exceptions that are generated due to precondition
//violations in insertBetween
- IndexHelper<E> indexHelper = new IndexHelper<E>(secondNodePredecessors);
- secondNode.rewire(firstNode, indexHelper.getIndex(firstNode), firstNodePredecessors.get(inputNum), true);
+ secondNode.rewire(firstNode, inputNum, firstNodePredecessors.get(inputNum), true);
secondNode.regenerateProjectionMap();
- //firstNode.rewire(firstNodePredecessors.get(inputNum), inputNum, secondNode, false);
firstNode.rewire(firstNodePredecessors.get(inputNum), 0, secondNode, false);
markDirty();
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/plan/ProjectionMap.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/plan/ProjectionMap.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/plan/ProjectionMap.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/plan/ProjectionMap.java Mon Jun 29 17:20:24 2009
@@ -21,6 +21,7 @@
import java.lang.StringBuilder;
import java.util.List;
+import org.apache.pig.data.DataType;
import org.apache.pig.impl.util.MultiMap;
import org.apache.pig.impl.util.Pair;
@@ -35,11 +36,12 @@
/**
* Map of field changes, with keys being the output fields of the operator
- * and values being the input fields. Fields are numbered from 0. So for a
- * foreach operator derived from 'B = foreach A generate $0, $2, $3,
- * udf($1)' would produce a mapping of 0->(0, 0), 2->(0, 1), 3->(0, 2)
+ * and values being the projection map columns. Fields are numbered from 0.
+ * So for a foreach operator derived from 'B = foreach A generate $0, $2,
+ * (int)$3, udf($1)' would produce a mapping of 0->(0, 0), 2->(0, 1),
+ * 3->[(0, 2), DataType.INTEGER]
*/
- private MultiMap<Integer, Pair<Integer, Integer>> mMappedFields;
+ private MultiMap<Integer, Column> mMappedFields;
/**
* List of fields removed from the input. This includes fields that were
@@ -73,7 +75,7 @@
* @param addedFields
* the list of columns that are added to the output
*/
- public ProjectionMap(MultiMap<Integer, Pair<Integer, Integer>> mapFields,
+ public ProjectionMap(MultiMap<Integer, Column> mapFields,
List<Pair<Integer, Integer>> removedFields,
List<Integer> addedFields) {
this(mapFields, removedFields, addedFields, true);
@@ -90,7 +92,7 @@
* @param changes
* to indicate if this projection map changes its input or not
*/
- private ProjectionMap(MultiMap<Integer, Pair<Integer, Integer>> mapFields,
+ private ProjectionMap(MultiMap<Integer, Column> mapFields,
List<Pair<Integer, Integer>> removedFields,
List<Integer> addedFields, boolean changes) {
mMappedFields = mapFields;
@@ -103,7 +105,7 @@
*
* @return the mapping of input column to output column
*/
- public MultiMap<Integer, Pair<Integer, Integer>> getMappedFields() {
+ public MultiMap<Integer, Column> getMappedFields() {
return mMappedFields;
}
@@ -112,7 +114,7 @@
* @param fields
* the mapping of input column to output column
*/
- public void setMappedFields(MultiMap<Integer, Pair<Integer, Integer>> fields) {
+ public void setMappedFields(MultiMap<Integer, Column> fields) {
mMappedFields = fields;
}
@@ -185,4 +187,95 @@
sb.append(" removed fields: " + mRemovedFields);
return sb.toString();
}
+
+ /**
+ *
+ * A wrapper for projection map columns to contain the input number, input
+ * column. Additionally, if a cast is used, record the type of the cast
+ *
+ */
+ public static class Column {
+ private Pair<Integer, Integer> mInputColumn;
+ private boolean mCast = false;
+ private byte mCastType = DataType.UNKNOWN;
+
+ /**
+ *
+ * @param inputColumn
+ * A pair of integers representing the input number and the
+ * input column number
+ */
+ public Column(Pair<Integer, Integer> inputColumn) {
+ this(inputColumn, false, DataType.UNKNOWN);
+ }
+
+ /**
+ *
+ * @param inputColumn
+ * A pair of integers representing the input number and the
+ * input column number
+ * @param cast
+ * true if the input column has a cast
+ * @param castType
+ * the data type of the cast
+ */
+ public Column(Pair<Integer, Integer> inputColumn, boolean cast,
+ byte castType) {
+ mInputColumn = inputColumn;
+ mCast = cast;
+ mCastType = castType;
+ }
+
+ /**
+ *
+ * Get the mapped column details
+ *
+ * @return A pair of integers representing the input number and the
+ * input column number
+ */
+ public Pair<Integer, Integer> getInputColumn() {
+ return mInputColumn;
+ }
+
+ /**
+ *
+ * Check if the column is cast
+ *
+ * @return true if the input column has a cast; false otherwise
+ */
+ public boolean getCast() {
+ return mCast;
+ }
+
+
+ /**
+ *
+ * Check if the column is cast
+ *
+ * @return true if the input column has a cast; false otherwise
+ */
+ public boolean cast() {
+ return getCast();
+ }
+
+ /**
+ *
+ * Get the data type of the cast
+ *
+ * @return the data type of the cast
+ */
+ public byte getCastType() {
+ return mCastType;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append(mInputColumn);
+ if (mCast) {
+ sb.append(" cast to: " + DataType.findTypeName(mCastType));
+ }
+ return sb.toString();
+ }
+ }
}
\ No newline at end of file
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/plan/optimizer/PlanOptimizer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/plan/optimizer/PlanOptimizer.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/plan/optimizer/PlanOptimizer.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/plan/optimizer/PlanOptimizer.java Mon Jun 29 17:20:24 2009
@@ -89,6 +89,7 @@
sawMatch = true;
rule.getTransformer().transform(match);
}
+ rule.getTransformer().reset();
}
}
}
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/plan/optimizer/Transformer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/plan/optimizer/Transformer.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/plan/optimizer/Transformer.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/plan/optimizer/Transformer.java Mon Jun 29 17:20:24 2009
@@ -60,6 +60,13 @@
* @throws OptimizerException
*/
public abstract void transform(List<O> nodes) throws OptimizerException;
+
+ /**
+ * Reset any internal state after the transformation.
+ *
+ * @throws OptimizerException
+ */
+ public abstract void reset() throws OptimizerException;
}
Modified: hadoop/pig/trunk/src/org/apache/pig/pen/util/FunctionalLogicalOptimizer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/pen/util/FunctionalLogicalOptimizer.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/pen/util/FunctionalLogicalOptimizer.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/pen/util/FunctionalLogicalOptimizer.java Mon Jun 29 17:20:24 2009
@@ -23,6 +23,7 @@
import java.util.List;
import java.util.Map;
+import org.apache.pig.impl.logicalLayer.LOFilter;
import org.apache.pig.impl.logicalLayer.LOLoad;
import org.apache.pig.impl.logicalLayer.LOStream;
import org.apache.pig.impl.logicalLayer.LogicalOperator;
@@ -85,6 +86,15 @@
rulePlan.add(loStream);
mRules.add(new Rule<LogicalOperator, LogicalPlan>(rulePlan, new TypeCastInserter(plan,
LOStream.class.getName()), "StreamTypeCastInserter"));
+
+ // Push up filters as far as we can
+ rulePlan = new RulePlan();
+ RuleOperator loFilter = new RuleOperator(LOFilter.class,
+ new OperatorKey(SCOPE, nodeIdGen.getNextNodeId(SCOPE)));
+ rulePlan.add(loFilter);
+ mRules.add(new Rule<LogicalOperator, LogicalPlan>(rulePlan,
+ new TypeCastInserter(plan, LOFilter.class.getName()),
+ "PushUpFilter"));
}
Modified: hadoop/pig/trunk/test/org/apache/pig/test/TestOperatorPlan.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/TestOperatorPlan.java?rev=789388&r1=789387&r2=789388&view=diff
==============================================================================
--- hadoop/pig/trunk/test/org/apache/pig/test/TestOperatorPlan.java (original)
+++ hadoop/pig/trunk/test/org/apache/pig/test/TestOperatorPlan.java Mon Jun 29 17:20:24 2009
@@ -286,6 +286,10 @@
public int getNumberOfChecks() {
return mNumChecks;
}
+
+ @Override
+ public void reset() {
+ }
}
class NeverTransform extends Transformer<TOperator, TPlan> {
@@ -308,6 +312,10 @@
public int getNumberOfChecks() {
return mNumChecks;
}
+
+ @Override
+ public void reset() {
+ }
}
@Test