You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by xu...@apache.org on 2011/05/07 02:15:44 UTC
svn commit: r1100420 [9/19] - in /pig/branches/branch-0.9: ./ src/
src/org/apache/pig/
src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/
src/org/apache/pig/impl/logicalLayer/
src/org/apache/pig/impl/logicalLayer/optimizer/ src/org/apache...
Modified: pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/TopLevelProjectFinder.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/TopLevelProjectFinder.java?rev=1100420&r1=1100419&r2=1100420&view=diff
==============================================================================
--- pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/TopLevelProjectFinder.java (original)
+++ pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/TopLevelProjectFinder.java Sat May 7 00:15:40 2011
@@ -1,111 +0,0 @@
-/*
- * 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.impl.plan.DepthFirstWalker;
-import org.apache.pig.impl.plan.VisitorException;
-
-/**
- * A visitor to track the top-level projection operators in a plan.
- * If there is a $1.$0 then only $1 is tracked
- */
-public class TopLevelProjectFinder extends
- LOVisitor {
-
- List<LOProject> mProjectList = new ArrayList<LOProject>();
-
- public TopLevelProjectFinder(LogicalPlan plan) {
- super(plan, new DepthFirstWalker<LogicalOperator, LogicalPlan>(plan));
- }
-
-
-
- /* (non-Javadoc)
- * @see org.apache.pig.impl.logicalLayer.LOVisitor#visit(org.apache.pig.impl.logicalLayer.LODistinct)
- */
- @Override
- protected void visit(LODistinct dt) throws VisitorException {
- }
-
-
-
- /* (non-Javadoc)
- * @see org.apache.pig.impl.logicalLayer.LOVisitor#visit(org.apache.pig.impl.logicalLayer.LOFilter)
- */
- @Override
- protected void visit(LOFilter filter) throws VisitorException {
- }
-
-
-
- /* (non-Javadoc)
- * @see org.apache.pig.impl.logicalLayer.LOVisitor#visit(org.apache.pig.impl.logicalLayer.LOForEach)
- */
- @Override
- protected void visit(LOForEach forEach) throws VisitorException {
- }
-
-
-
- /* (non-Javadoc)
- * @see org.apache.pig.impl.logicalLayer.LOVisitor#visit(org.apache.pig.impl.logicalLayer.LOSort)
- */
- @Override
- protected void visit(LOSort s) throws VisitorException {
- }
-
-
-
- /* (non-Javadoc)
- * @see org.apache.pig.impl.logicalLayer.LOVisitor#visit(org.apache.pig.impl.logicalLayer.LOProject)
- */
- @Override
- protected void visit(LOProject project) throws VisitorException {
- //If the project is a root then add it to the list
- List<LogicalOperator> projectPreds = this.getPlan().getPredecessors(project);
- if(projectPreds == null) {
- //check if the project's predecessor is null then add it to the list
- mProjectList.add(project);
- }
- }
-
- public List<LOProject> getProjectList() {
- return mProjectList;
- }
-
- public Set<LOProject> getProjectSet() {
- return new HashSet<LOProject>(mProjectList);
- }
-
- public Set<LOProject> getProjectStarSet() {
- Set<LOProject> projectStarSet = new HashSet<LOProject>();
-
- for(LOProject project: getProjectSet()) {
- if(project.isStar() && (this.getPlan().getPredecessors(project) == null)) {
- projectStarSet.add(project);
- }
- }
-
- return (projectStarSet.size() == 0? null : projectStarSet);
- }
-}
Modified: pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/UDFFinder.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/UDFFinder.java?rev=1100420&r1=1100419&r2=1100420&view=diff
==============================================================================
--- pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/UDFFinder.java (original)
+++ pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/UDFFinder.java Sat May 7 00:15:40 2011
@@ -1,81 +0,0 @@
-/*
- * 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: pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/UnaryExpressionOperator.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/UnaryExpressionOperator.java?rev=1100420&r1=1100419&r2=1100420&view=diff
==============================================================================
--- pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/UnaryExpressionOperator.java (original)
+++ pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/UnaryExpressionOperator.java Sat May 7 00:15:40 2011
@@ -1,91 +0,0 @@
-/*
- * 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.List;
-
-import org.apache.pig.impl.plan.OperatorKey;
-import org.apache.pig.impl.plan.PlanVisitor;
-import org.apache.pig.impl.plan.VisitorException;
-import org.apache.pig.data.DataType;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-/**
- * This abstract class represents the logical Unary Expression Operator The
- * unary operator has an operand and an operator. The format of the expression
- * is operator operand. The operator is implicit and not recorded in the class
- */
-public abstract class UnaryExpressionOperator extends ExpressionOperator {
- private static final long serialVersionUID = 2L;
- private static Log log = LogFactory.getLog(UnaryExpressionOperator.class);
-
- /**
- * @param plan
- * Logical plan this operator is a part of.
- * @param k
- * Operator key to assign to this node.
- * @param rp
- * degree of requested parallelism with which to execute this
- * node.
- */
- public UnaryExpressionOperator(LogicalPlan plan, OperatorKey k, int rp) {
- super(plan, k, rp);
- }
-
- /**
- * @param plan
- * Logical plan this operator is a part of.
- * @param k
- * Operator key to assign to this node.
- */
- public UnaryExpressionOperator(LogicalPlan plan, OperatorKey k) {
- super(plan, k);
-
- }
-
- public ExpressionOperator getOperand() {
- List<LogicalOperator>preds = getPlan().getPredecessors(this);
- if(preds == null)
- return null;
- return (ExpressionOperator)preds.get(0);
- }
-
- @Override
- public void visit(LOVisitor v) throws VisitorException {
- v.visit(this);
- }
-
- @Override
- public boolean supportsMultipleInputs() {
- return false;
- }
-
- /**
- * @see org.apache.pig.impl.logicalLayer.ExpressionOperator#clone()
- * Do not use the clone method directly. Operators are cloned when logical plans
- * are cloned using {@link LogicalPlanCloner}
- */
- @Override
- protected Object clone() throws CloneNotSupportedException {
- UnaryExpressionOperator unExOpClone = (UnaryExpressionOperator)super.clone();
- return unExOpClone;
- }
-
-}
Modified: pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/UnionOnSchemaSetter.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/UnionOnSchemaSetter.java?rev=1100420&r1=1100419&r2=1100420&view=diff
==============================================================================
--- pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/UnionOnSchemaSetter.java (original)
+++ pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/UnionOnSchemaSetter.java Sat May 7 00:15:40 2011
@@ -1,194 +0,0 @@
-/*
- * 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.ArrayList;
-import java.util.List;
-
-import org.apache.pig.PigException;
-import org.apache.pig.data.DataType;
-import org.apache.pig.impl.PigContext;
-import org.apache.pig.impl.logicalLayer.schema.Schema;
-import org.apache.pig.impl.logicalLayer.schema.SchemaMergeException;
-import org.apache.pig.impl.logicalLayer.validators.LogicalPlanValidationExecutor;
-import org.apache.pig.impl.logicalLayer.validators.UnionOnSchemaSetException;
-import org.apache.pig.impl.plan.CompilationMessageCollector;
-import org.apache.pig.impl.plan.DependencyOrderWalker;
-import org.apache.pig.impl.plan.NodeIdGenerator;
-import org.apache.pig.impl.plan.OperatorKey;
-import org.apache.pig.impl.plan.PlanValidationException;
-
-/**
- * A visitor that modifies the logical plan (if necessary) for union-onschema
- * functionality. It runs logical plan validator so that the correct schema
- * of its inputs is available. It inserts foreach statements in its input
- * if the input operator schema does not match the schema created by
- * merging all input schemas
- *
- */
-public class UnionOnSchemaSetter extends LOVisitor {
-
- private PigContext pigContext;
-
-
- public UnionOnSchemaSetter(LogicalPlan plan, PigContext pigContext) {
- super(plan, new DependencyOrderWalker<LogicalOperator, LogicalPlan>(plan));
- this.pigContext = pigContext;
- }
-
-
- public void visit(LOUnion loUnion) throws PlanValidationException, UnionOnSchemaSetException {
- if(!loUnion.isOnSchema()) {
- //Not union-onschema, nothing to be done
- return;
- }
- // run through validator first on inputs so that the schemas have the right
- //types for columns. It will run TypeCheckingValidator as well.
- // The compilation messages will be logged when validation is
- // done from PigServer, so not doing it here
- CompilationMessageCollector collector = new CompilationMessageCollector() ;
- boolean isBeforeOptimizer = true;
-
-
- LogicalPlanValidationExecutor validator =
- new LogicalPlanValidationExecutor(mPlan, pigContext, isBeforeOptimizer);
- validator.validate(mPlan, collector);
- List<LogicalOperator> preds = mPlan.getPredecessors(loUnion);
-
- //create the merged schema
- Schema mergedSchema ;
- try {
- mergedSchema = loUnion.getSchema();
- }catch(FrontendException e) {
- String msg = "Error creating merged schemas for union-onschema operator : "
- + e.getMessage();
- throw new UnionOnSchemaSetException(msg, 1116, PigException.INPUT, e);
- }
-
-
- //create a user defined schema list for use in LOForeach
- // using merged schema
- ArrayList<Schema> mergedSchemaList = new ArrayList<Schema>();
- for(Schema.FieldSchema fs : mergedSchema.getFields()){
- // Use NULL datatype because the type will be set by the TypeChecking
- // visitors
- mergedSchemaList.add(
- new Schema(new Schema.FieldSchema(fs.alias, DataType.NULL))
- );
- }
-
- // add a foreach for inputs that don't match mergedSchema, projecting
- // null for columns that don't exist in the input
- for(LogicalOperator lop : preds)
- {
- try {
- if(! lop.getSchema().equals(mergedSchema))
- {
- //the mergedSchema is different from this operators schema
- // so add a foreach to project columns appropriately
- int mergeSchSz = mergedSchema.size();
- ArrayList<LogicalPlan> generatePlans =
- new ArrayList<LogicalPlan>(mergeSchSz);
- ArrayList<Boolean> flattenList =
- new ArrayList<Boolean>(mergeSchSz);
-
- String scope = loUnion.getOperatorKey().getScope();
- for(Schema.FieldSchema fs : mergedSchema.getFields()) {
- LogicalPlan projectPlan = new LogicalPlan();
- Schema inpSchema = lop.getSchema();
- flattenList.add(Boolean.FALSE);
-
- int inpPos = inpSchema.getPositionSubName(fs.alias);
-
- LogicalOperator columnProj = null;
- boolean isCastNeeded = false;
- if(inpPos == -1){
- //the column is not present in schema of this input,
- // so project null
- columnProj =
- new LOConst(mPlan, getNextId(scope), null);
- // cast is necessary if the type in schema is
- // not a BYTEARRAY
- if(fs.type != DataType.BYTEARRAY){
- isCastNeeded = true;
- }
- }else {
- //project the column from input
- columnProj =
- new LOProject(projectPlan,
- new OperatorKey(
- scope,
- NodeIdGenerator.getGenerator().getNextNodeId(scope)
- ),
- lop, inpPos
- );
-
- //cast is needed if types are different.
- //compatibility of types has already been checked
- //during creation of mergedSchema
- Schema.FieldSchema inpFs = inpSchema.getFieldSubNameMatch(fs.alias);
- if(inpFs.type != fs.type)
- isCastNeeded = true;
- }
- projectPlan.add(columnProj);
-
- //add a LOCast if necessary
- if(isCastNeeded){
- LOCast loCast = new LOCast(
- projectPlan,
- getNextId(scope),
- fs.type
- );
- loCast.setFieldSchema(fs);
- projectPlan.add(loCast);
- projectPlan.connect(columnProj, loCast);
- }
- generatePlans.add(projectPlan);
-
- }
- LogicalOperator foreach = new LOForEach(
- mPlan,
- getNextId(scope),
- generatePlans, flattenList,
- mergedSchemaList
- );
- mPlan.add(foreach);
- mPlan.insertBetween(lop, foreach, loUnion);
- }
- }
- catch (FrontendException e) {
- String msg = "Error adding union operator " + loUnion.getAlias()
- + ":" + e.getMessage();
- UnionOnSchemaSetException pe = new UnionOnSchemaSetException(msg);
- pe.initCause(e);
- throw pe;
- }
-
- }
-
- }
-
-
- private OperatorKey getNextId(String scope) {
- return new OperatorKey(
- scope,
- NodeIdGenerator.getGenerator().getNextNodeId(scope)
- );
- }
-
-}
Modified: pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/ImplicitSplitInserter.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/ImplicitSplitInserter.java?rev=1100420&r1=1100419&r2=1100420&view=diff
==============================================================================
--- pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/ImplicitSplitInserter.java (original)
+++ pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/ImplicitSplitInserter.java Sat May 7 00:15:40 2011
@@ -1,166 +0,0 @@
-/*
- * 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.List;
-
-import org.apache.pig.PigException;
-import org.apache.pig.data.DataType;
-import org.apache.pig.impl.logicalLayer.LOConst;
-import org.apache.pig.impl.logicalLayer.LOSplitOutput;
-import org.apache.pig.impl.logicalLayer.LogicalOperator;
-import org.apache.pig.impl.logicalLayer.LogicalPlan;
-import org.apache.pig.impl.logicalLayer.LOSplit;
-import org.apache.pig.impl.logicalLayer.LOStore;
-import org.apache.pig.impl.plan.DepthFirstWalker;
-import org.apache.pig.impl.plan.NodeIdGenerator;
-import org.apache.pig.impl.plan.OperatorKey;
-import org.apache.pig.impl.plan.optimizer.OptimizerException;
-
-public class ImplicitSplitInserter extends LogicalTransformer {
-
- public ImplicitSplitInserter(LogicalPlan plan) {
- super(plan);
- }
-
- @Override
- public boolean check(List<LogicalOperator> nodes) throws OptimizerException {
- // Look to see if this is a non-split node with two outputs. If so
- // it matches.
- 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);
- }
- try {
- LogicalOperator op = nodes.get(0);
- List<LogicalOperator> succs = mPlan.getSuccessors(op);
- if (succs == null || succs.size() < 2) return false;
- if (op instanceof LOSplit) return false;
- if (op instanceof LOStore) return false;
- return true;
- } catch (Exception e) {
- int errCode = 2048;
- String msg = "Error while performing checks to introduce split operators.";
- throw new OptimizerException(msg, errCode, PigException.BUG, e);
- }
- }
-
- @Override
- public void transform(List<LogicalOperator> nodes)
- throws OptimizerException {
- 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);
- }
- try {
- // Insert a split and its corresponding SplitOutput nodes into the plan
- // between node 0 and 1 / 2.
- String scope = nodes.get(0).getOperatorKey().scope;
- NodeIdGenerator idGen = NodeIdGenerator.getGenerator();
- LOSplit splitOp = new LOSplit(mPlan, new OperatorKey(scope,
- idGen.getNextNodeId(scope)), new ArrayList<LogicalOperator>());
- splitOp.setAlias(nodes.get(0).getAlias());
- mPlan.add(splitOp);
-
- // Find all the successors and connect appropriately with split
- // and splitoutput operators. Keep our own copy
- // of the list, as we're changing the graph by doing these calls
- // and that will change the list of predecessors.
- List<LogicalOperator> succs =
- new ArrayList<LogicalOperator>(mPlan.getSuccessors(nodes.get(0)));
- int index = -1;
- // For two successors of nodes.get(0) here is a pictorial
- // representation of the change required:
- // BEFORE:
- // Succ1 Succ2
- // \ /
- // nodes.get(0)
-
- // SHOULD BECOME:
-
- // AFTER:
- // Succ1 Succ2
- // | |
- // SplitOutput SplitOutput
- // \ /
- // Split
- // |
- // nodes.get(0)
-
- // Here is how this will be accomplished.
- // First (the same) Split Operator will be "inserted between" nodes.get(0)
- // and all its successors. The "insertBetween" API is used which makes sure
- // the ordering of operators in the graph is preserved. So we get the following:
- // Succ1 Succ2
- // | |
- // Split Split
- // \ /
- // nodes.get(0)
-
- // Then all but the first connection between nodes.get(0) and the Split
- // Operator are removed using "disconnect" - so we get the following:
- // Succ1 Succ2
- // \ /
- // Split
- // |
- // nodes.get(0)
-
- // Now a new SplitOutputOperator is "inserted between" the Split operator
- // and the successors. So we get:
- // Succ1 Succ2
- // | |
- // SplitOutput SplitOutput
- // \ /
- // Split
- // |
- // nodes.get(0)
-
-
- for (LogicalOperator succ : succs) {
- mPlan.insertBetween(nodes.get(0), splitOp, succ);
- }
-
- for(int i = 1; i < succs.size(); i++) {
- mPlan.disconnect(nodes.get(0), splitOp);
- }
-
- for (LogicalOperator succ : succs) {
- LogicalPlan condPlan = new LogicalPlan();
- LOConst cnst = new LOConst(mPlan, new OperatorKey(scope,
- idGen.getNextNodeId(scope)), Boolean.valueOf(true));
- cnst.setType(DataType.BOOLEAN);
- condPlan.add(cnst);
- LOSplitOutput splitOutput = new LOSplitOutput(mPlan,
- new OperatorKey(scope, idGen.getNextNodeId(scope)), ++index, condPlan);
- splitOp.addOutput(splitOutput);
- mPlan.add(splitOutput);
- mPlan.insertBetween(splitOp, splitOutput, succ);
- splitOutput.setAlias(splitOp.getAlias());
- }
-
- } catch (Exception e) {
- int errCode = 2047;
- String msg = "Internal error. Unable to introduce split operators.";
- throw new OptimizerException(msg, errCode, PigException.BUG, e);
- }
- }
-}
Modified: pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalOptimizer.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalOptimizer.java?rev=1100420&r1=1100419&r2=1100420&view=diff
==============================================================================
--- pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalOptimizer.java (original)
+++ pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalOptimizer.java Sat May 7 00:15:40 2011
@@ -1,245 +0,0 @@
-/*
- * 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.List;
-import java.util.Set;
-
-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.LOForEach;
-import org.apache.pig.impl.logicalLayer.LOLimit;
-import org.apache.pig.impl.logicalLayer.LOLoad;
-import org.apache.pig.impl.logicalLayer.LONative;
-import org.apache.pig.impl.logicalLayer.LOStream;
-import org.apache.pig.impl.logicalLayer.LogicalOperator;
-import org.apache.pig.impl.logicalLayer.LogicalPlan;
-import org.apache.pig.impl.logicalLayer.RelationalOperator;
-import org.apache.pig.impl.plan.NodeIdGenerator;
-import org.apache.pig.impl.plan.OperatorKey;
-import org.apache.pig.impl.plan.optimizer.OptimizerException;
-import org.apache.pig.impl.plan.optimizer.PlanOptimizer;
-import org.apache.pig.impl.plan.optimizer.Rule;
-import org.apache.pig.impl.plan.optimizer.RuleMatcher;
-import org.apache.pig.impl.plan.optimizer.RuleOperator;
-import org.apache.pig.impl.plan.optimizer.RulePlan;
-
-/**
- * An optimizer for logical plans.
- */
-public class LogicalOptimizer extends
- PlanOptimizer<LogicalOperator, LogicalPlan> {
-
- private static final String SCOPE = "RULE";
- private static NodeIdGenerator nodeIdGen = NodeIdGenerator.getGenerator();
-
- private Set<String> mRulesOff = null;
- private Rule<LogicalOperator, LogicalPlan> pruneRule;
-
- public LogicalOptimizer(LogicalPlan plan) {
- this(plan, ExecType.MAPREDUCE);
- }
-
- public LogicalOptimizer(LogicalPlan plan, ExecType mode) {
- super(plan);
- runOptimizations(plan, mode);
- }
-
- public LogicalOptimizer(LogicalPlan plan, ExecType mode, Set<String> turnOffRules) {
- super(plan);
- mRulesOff = turnOffRules;
- runOptimizations(plan, mode);
- }
-
- private void runOptimizations(LogicalPlan plan, ExecType mode) {
- RulePlan rulePlan;
-
- // List of rules for the logical optimizer
-
- boolean turnAllRulesOff = false;
- if (mRulesOff != null) {
- for (String rule : mRulesOff) {
- if ("all".equalsIgnoreCase(rule)) {
- turnAllRulesOff = true;
- break;
- }
- }
- }
-
- // This one has to be before the type cast inserter as it expects the
- // load to only have one output.
- // Find any places in the plan that have an implicit split and make
- // it explicit. Since the RuleMatcher doesn't handle trees properly,
- // we cheat and say that we match any node. Then we'll do the actual
- // test in the transformers check method.
-
- rulePlan = new RulePlan();
- RuleOperator anyLogicalOperator = new RuleOperator(LogicalOperator.class, RuleOperator.NodeType.ANY_NODE,
- new OperatorKey(SCOPE, nodeIdGen.getNextNodeId(SCOPE)));
- rulePlan.add(anyLogicalOperator);
- mRules.add(new Rule<LogicalOperator, LogicalPlan>(rulePlan,
- new ImplicitSplitInserter(plan), "ImplicitSplitInserter"));
-
-
- // this one is ordered to be before other optimizations since later
- // optimizations may move the LOFilter that is looks for just after a
- // LOLoad
- rulePlan = new RulePlan();
- RuleOperator loLoad = new RuleOperator(LOLoad.class,
- new OperatorKey(SCOPE, nodeIdGen.getNextNodeId(SCOPE)));
- rulePlan.add(loLoad);
- mRules.add(new Rule<LogicalOperator, LogicalPlan>(rulePlan,
- new PartitionFilterOptimizer(plan), "LoadPartitionFilterOptimizer"));
-
- // Add type casting to plans where the schema has been declared (by
- // user, data, or data catalog).
- rulePlan = new RulePlan();
- loLoad = new RuleOperator(LOLoad.class,
- new OperatorKey(SCOPE, nodeIdGen.getNextNodeId(SCOPE)));
- rulePlan.add(loLoad);
- mRules.add(new Rule<LogicalOperator, LogicalPlan>(rulePlan,
- new TypeCastInserter(plan, LOLoad.class.getName()), "LoadTypeCastInserter"));
-
- // Add type casting to plans where the schema has been declared by
- // user in a statement with stream operator.
-
- rulePlan = new RulePlan();
- RuleOperator loStream= new RuleOperator(LOStream.class,
- new OperatorKey(SCOPE, nodeIdGen.getNextNodeId(SCOPE)));
- rulePlan.add(loStream);
- mRules.add(new Rule<LogicalOperator, LogicalPlan>(rulePlan, new TypeCastInserter(plan,
- LOStream.class.getName()), "StreamTypeCastInserter"));
-
- if(!turnAllRulesOff) {
-
- // Push up limit wherever possible.
- rulePlan = new RulePlan();
- RuleOperator loLimit = new RuleOperator(LOLimit.class,
- new OperatorKey(SCOPE, nodeIdGen.getNextNodeId(SCOPE)));
- rulePlan.add(loLimit);
- Rule<LogicalOperator, LogicalPlan> 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(loFilter);
- rule = new Rule<LogicalOperator, LogicalPlan>(rulePlan,
- new PushUpFilter(plan), "PushUpFilter");
- checkAndAddRule(rule);
-
- // Push foreach with flatten down wherever possible
- rulePlan = new RulePlan();
- RuleOperator loForeach = new RuleOperator(LOForEach.class,
- new OperatorKey(SCOPE, nodeIdGen.getNextNodeId(SCOPE)));
- rulePlan.add(loForeach);
- rule = new Rule<LogicalOperator, LogicalPlan>(rulePlan,
- new PushDownForeachFlatten(plan), "PushDownForeachFlatten");
- checkAndAddRule(rule);
-
- // Prune column up wherever possible
- rulePlan = new RulePlan();
- RuleOperator rulePruneColumnsOperator = new RuleOperator(RelationalOperator.class, RuleOperator.NodeType.ANY_NODE,
- new OperatorKey(SCOPE, nodeIdGen.getNextNodeId(SCOPE)));
- rulePlan.add(rulePruneColumnsOperator);
- pruneRule = new Rule<LogicalOperator, LogicalPlan>(rulePlan,
- new PruneColumns(plan), "PruneColumns", Rule.WalkerAlgo.ReverseDependencyOrderWalker);
- }
-
- }
-
- private boolean ruleEnabled(Rule<LogicalOperator, LogicalPlan> rule) {
- if(mRulesOff != null && rule != null) {
- for(String ruleOff: mRulesOff) {
- String ruleName = rule.getRuleName();
- if(ruleName == null) continue;
- if(ruleName.equalsIgnoreCase(ruleOff)) return false;
- }
- }
- mRules.add(rule);
- return true;
- }
-
- private void checkAndAddRule(Rule<LogicalOperator, LogicalPlan> rule) {
- if (ruleEnabled(rule))
- mRules.add(rule);
- }
-
- @Override
- public final int optimize() throws OptimizerException {
- //the code that follows is a copy of the code in the
- //base class. see the todo note in the base class
- boolean sawMatch = false;
- boolean initialized = false;
- int numIterations = 0;
- do {
- sawMatch = false;
- for (Rule<LogicalOperator, LogicalPlan> rule : mRules) {
- RuleMatcher<LogicalOperator, LogicalPlan> matcher = new RuleMatcher<LogicalOperator, LogicalPlan>();
- if (matcher.match(rule)) {
- // It matches the pattern. Now check if the transformer
- // approves as well.
- List<List<LogicalOperator>> matches = matcher.getAllMatches();
- for (List<LogicalOperator> match:matches)
- {
- if (rule.getTransformer().check(match)) {
- try {
- // The transformer approves.
- sawMatch = true;
- if (!initialized)
- {
- ((LogicalTransformer)rule.getTransformer()).rebuildSchemas();
- ((LogicalTransformer)rule.getTransformer()).rebuildProjectionMaps();
- initialized = true;
- }
- rule.getTransformer().transform(match);
- ((LogicalTransformer)rule.getTransformer()).rebuildSchemas();
- ((LogicalTransformer)rule.getTransformer()).rebuildProjectionMaps();
- } catch (FrontendException fee) {
- int errCode = 2145;
- String msg = "Problem while rebuilding projection map or schema in logical optimizer.";
- throw new OptimizerException(msg, errCode, PigException.BUG, fee);
- }
-
- }
- rule.getTransformer().reset();
- }
- }
- }
- } while(sawMatch && ++numIterations < mMaxIterations);
- if (pruneRule!=null && ruleEnabled(pruneRule))
- {
- RuleMatcher<LogicalOperator, LogicalPlan> matcher = new RuleMatcher<LogicalOperator, LogicalPlan>();
- if (matcher.match(pruneRule)) {
- List<List<LogicalOperator>> matches = matcher.getAllMatches();
- for (List<LogicalOperator> match:matches)
- {
- if (pruneRule.getTransformer().check(match)) {
- pruneRule.getTransformer().transform(match);
- }
- }
- ((PruneColumns)pruneRule.getTransformer()).prune();
- }
- }
- return numIterations;
- }
-}
\ No newline at end of file
Modified: pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalTransformer.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalTransformer.java?rev=1100420&r1=1100419&r2=1100420&view=diff
==============================================================================
--- pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalTransformer.java (original)
+++ pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalTransformer.java Sat May 7 00:15:40 2011
@@ -1,213 +0,0 @@
-/*
- * 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.List;
-import java.util.Map;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import org.apache.pig.impl.plan.PlanWalker;
-import org.apache.pig.impl.plan.VisitorException;
-import org.apache.pig.impl.plan.optimizer.Transformer;
-import org.apache.pig.impl.logicalLayer.FrontendException;
-import org.apache.pig.impl.logicalLayer.LOJoin;
-import org.apache.pig.impl.logicalLayer.LOJoin;
-import org.apache.pig.impl.logicalLayer.LogicalOperator;
-import org.apache.pig.impl.logicalLayer.LogicalPlan;
-import org.apache.pig.impl.logicalLayer.LOCogroup;
-import org.apache.pig.impl.logicalLayer.LOFilter;
-import org.apache.pig.impl.logicalLayer.LOForEach;
-import org.apache.pig.impl.logicalLayer.LOSort;
-import org.apache.pig.impl.logicalLayer.LOSplit;
-import org.apache.pig.impl.logicalLayer.LOSplitOutput;
-import org.apache.pig.impl.logicalLayer.ProjectFixerUpper;
-import org.apache.pig.impl.logicalLayer.ProjectionMapCalculator;
-import org.apache.pig.impl.logicalLayer.ProjectionMapRemover;
-
-public abstract class LogicalTransformer extends Transformer<LogicalOperator, LogicalPlan> {
-
- private final Log log = LogFactory.getLog(getClass());
-
- protected LogicalTransformer(
- LogicalPlan plan) {
- super(plan);
- }
-
- /**
- * Rebuild schemas after a rule has transformed the tree. This will first
- * null out existing schemas and then call getSchema to rebuild them.
- * @throws VisitorException, FrontendException
- */
- protected void rebuildSchemas() throws VisitorException, FrontendException {
- SchemaRemover sr = new SchemaRemover(mPlan);
- sr.visit();
- SchemaCalculator sc = new SchemaCalculator(mPlan);
- sc.visit();
-
- }
-
- /**
- * Rebuild projection maps after a rule has transformed the tree. This will first
- * null out existing projection maps and then call getProjectionMap to rebuild them.
- * @throws VisitorException
- */
- protected void rebuildProjectionMaps() throws VisitorException {
- ProjectionMapRemover pMapRemover = new ProjectionMapRemover(mPlan);
- pMapRemover.visit();
- ProjectionMapCalculator pMapCalculator = new ProjectionMapCalculator(mPlan);
- pMapCalculator.visit();
-
- }
-
- /**
- * Insert a node in between two existing nodes. This includes inserting
- * the node into the correct place in the plan and finding any projects in
- * successors and reconnecting them to the new node as well as rebuilding
- * all of the schemas.
- * @param after Node to insert the new node after
- * @param newNode New node to insert
- * @param before Node to insert this node before
- * @param projectionMapping A map that defines how projections in after
- * relate to projections in newnode. Keys are the projection offsets in
- * after, values are the new offsets in newnode. If this field is null,
- * then it will be assumed that the mapping is 1-1.
- * @throws VisitorException, FrontendException
- */
- protected void insertBetween(
- LogicalOperator after,
- LogicalOperator newNode,
- LogicalOperator before,
- Map<Integer, Integer> projectionMapping)
- throws VisitorException, FrontendException {
- // Insert it into the plan.
- mPlan.add(newNode);
- mPlan.insertBetween(after, newNode, before);
- }
-
- /**
- * Once a node has been inserted, inner plans associated with other nodes
- * may have references to the node that has been replaced or moved. This
- * function walks those inner plans and patches up references.
- * @param after Node that has had a new node inserted after it.
- * @param newNode node that has been inserted
- * @param before Node that has had a new node inserted before it.
- * @param projectionMapping A map that defines how projections in after
- * relate to projections in newNode. Keys are the projection offsets in
- * after, values are the new offsets in newNode. If this field is null,
- * then it will be assumed that the mapping is 1-1.
- * @throws VisitorException, FrontendException
- */
- protected void fixUpContainedPlans(
- LogicalOperator after,
- LogicalOperator newNode,
- LogicalOperator before,
- Map<Integer, Integer> projectionMapping)
- throws VisitorException, FrontendException {
- // Fix up COGroup internal wiring
- if (before instanceof LOCogroup) {
- LOCogroup cg = (LOCogroup) before ;
- cg.switchGroupByPlanOp(after, newNode);
- }
- if (before instanceof LOJoin) {
- LOJoin frj = (LOJoin) before ;
- frj.switchJoinColPlanOp(after, newNode);
- }
- if (before instanceof LOJoin) {
- LOJoin frj = (LOJoin) before ;
- frj.switchJoinColPlanOp(after, newNode);
- }
-
- // Visit all the inner plans of before and change their projects to
- // connect to newNode instead of after.
- // Find right inner plan(s) to visit
- List<LogicalPlan> plans = new ArrayList<LogicalPlan>();
- if (before instanceof LOCogroup) {
- plans.addAll((((LOCogroup)before).getGroupByPlans()).values());
- } else if (before instanceof LOJoin) {
- plans.addAll((((LOJoin)before).getJoinPlans()).values());
- } else if (before instanceof LOJoin) {
- plans.addAll((((LOJoin)before).getJoinPlans()).values());
- }
- else if (before instanceof LOSort) {
- plans.addAll(((LOSort)before).getSortColPlans());
- } else if (before instanceof LOFilter) {
- plans.add(((LOFilter)before).getComparisonPlan());
- } else if (before instanceof LOSplit) {
- // In this case we have to find each of the Split outputs, and
- // add their plans.
- List<LogicalOperator> splitOutputs = mPlan.getSuccessors(before);
- for (LogicalOperator so : splitOutputs) {
- if (!(so instanceof LOSplitOutput)) {
- String msg = "Found an LOSplit with an operator other "
- + "than LOSplitOutput after it!";
- log.error(msg);
- throw new VisitorException(msg);
- }
- plans.add(((LOSplitOutput)so).getConditionPlan());
- }
- } else if (before instanceof LOForEach) {
- plans.addAll(((LOForEach)before).getForEachPlans());
- }
-
- for (LogicalPlan lp : plans) {
- ProjectFixerUpper pfu =
- new ProjectFixerUpper(lp, after, newNode, projectionMapping);
- pfu.visit();
- }
- }
-
- /**
- * Insert a node in after an existing nodes. This includes inserting
- * the node into the correct place in the plan and finding any projects in
- * successors and reconnecting them to the new node as well as rebuilding
- * all of the schemas. This function
- * assumes that the node has only one predecessor.
- * @param after Node to insert the new node after
- * @param newNode New node to insert
- * @param projectionMapping A map that defines how projections in after
- * relate to projections in newnode. Keys are the projection offsets in
- * after, values are the new offsets in newnode. If this field is null,
- * then it will be assumed that the mapping is 1-1.
- * @throws VisitorException, FrontendException
- */
- protected void insertAfter(
- LogicalOperator after,
- LogicalOperator newNode,
- Map<Integer, Integer> projectionMapping)
- throws VisitorException, FrontendException {
- List<LogicalOperator> successors = mPlan.getSuccessors(after);
- if(successors==null) {
- mPlan.addAsLeaf(newNode);
- return;
- }
- if (successors.size() != 1) {
- throw new RuntimeException("insertAfter only valid to insert " +
- "after a node with single output.");
- }
- insertBetween(after, newNode, successors.get(0), projectionMapping);
- }
-
- @Override
- public void reset() {
- // do nothing
- }
-
-}
Modified: pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/OpLimitOptimizer.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/OpLimitOptimizer.java?rev=1100420&r1=1100419&r2=1100420&view=diff
==============================================================================
--- pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/OpLimitOptimizer.java (original)
+++ pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/OpLimitOptimizer.java Sat May 7 00:15:40 2011
@@ -1,326 +0,0 @@
-/*
- * 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.List;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.pig.ExecType;
-import org.apache.pig.PigException;
-import org.apache.pig.impl.logicalLayer.LOCogroup;
-import org.apache.pig.impl.logicalLayer.LOCross;
-import org.apache.pig.impl.logicalLayer.LODistinct;
-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.LONative;
-import org.apache.pig.impl.logicalLayer.LOSort;
-import org.apache.pig.impl.logicalLayer.LOSplit;
-import org.apache.pig.impl.logicalLayer.LOSplitOutput;
-import org.apache.pig.impl.logicalLayer.LOStream;
-import org.apache.pig.impl.logicalLayer.LOUnion;
-import org.apache.pig.impl.logicalLayer.LOJoin;
-import org.apache.pig.impl.logicalLayer.LOJoin;
-import org.apache.pig.impl.logicalLayer.LogicalOperator;
-import org.apache.pig.impl.logicalLayer.LogicalPlan;
-import org.apache.pig.impl.logicalLayer.LogicalPlanCloner;
-import org.apache.pig.impl.plan.DepthFirstWalker;
-import org.apache.pig.impl.plan.NodeIdGenerator;
-import org.apache.pig.impl.plan.OperatorKey;
-import org.apache.pig.impl.plan.PlanException;
-import org.apache.pig.impl.plan.optimizer.OptimizerException;
-
-/**
- * A visitor to discover if any schema has been specified for a file being
- * loaded. If so, a projection will be injected into the plan to cast the
- * data being loaded to the appropriate types. The optimizer can then come
- * along and move those casts as far down as possible, or in some cases remove
- * them altogether. This visitor does not handle finding the schemas for the
- * file, that has already been done as part of parsing.
- *
- */
-public class OpLimitOptimizer extends LogicalTransformer {
-
- private static final Log log = LogFactory.getLog(OpLimitOptimizer.class);
- private ExecType mode = ExecType.MAPREDUCE;
-
- public OpLimitOptimizer(LogicalPlan plan) {
- super(plan);
- }
-
- public OpLimitOptimizer(LogicalPlan plan, ExecType mode) {
- super(plan);
- this.mode = mode;
- }
-
- @Override
- public boolean check(List<LogicalOperator> nodes) throws OptimizerException {
- 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);
- }
-
- try {
- LogicalOperator lo = nodes.get(0);
- if (lo == null || !(lo instanceof LOLimit)) {
- int errCode = 2005;
- String msg = "Expected " + LOLimit.class.getSimpleName()
- + ", got "
- + (lo == null ? lo : lo.getClass().getSimpleName());
- throw new OptimizerException(msg, errCode, PigException.BUG);
- }
- List<LogicalOperator> predecessors = mPlan.getPredecessors(lo);
- if (predecessors.size()!=1) {
- int errCode = 2008;
- String msg = "Limit cannot have more than one input. Found " + predecessors.size() + " inputs.";
- throw new OptimizerException(msg, errCode, PigException.BUG);
- }
- LogicalOperator predecessor = predecessors.get(0);
-
- // Limit cannot be pushed up
- if (predecessor instanceof LOCogroup || predecessor instanceof LOFilter ||
- predecessor instanceof LOLoad || predecessor instanceof LOSplit ||
- predecessor instanceof LODistinct || predecessor instanceof LOJoin ||
- predecessor instanceof LOStream || predecessor instanceof LONative)
- {
- return false;
- }
- // Limit cannot be pushed in front of ForEach if it has a flatten
- 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;
-
- if (hasFlatten) {
- return false;
- }
- }
- } catch (Exception e) {
- int errCode = 2049;
- String msg = "Error while performing checks to optimize limit operator.";
- throw new OptimizerException(msg, errCode, PigException.BUG);
- }
-
- return true;
- }
-
- @Override
- public void transform(List<LogicalOperator> nodes) throws OptimizerException {
- 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);
- }
- try {
- LogicalOperator lo = nodes.get(0);
- if (lo == null || !(lo instanceof LOLimit)) {
- int errCode = 2005;
- String msg = "Expected " + LOLimit.class.getSimpleName() + ", got " + (lo == null ? lo : lo.getClass().getSimpleName());
- throw new OptimizerException(msg, errCode, PigException.BUG);
- }
-
- LOLimit limit = (LOLimit)lo;
-
- processNode(limit);
- } catch (OptimizerException oe) {
- throw oe;
- } catch (Exception e) {
- int errCode = 2050;
- String msg = "Internal error. Unable to optimize limit operator.";
- throw new OptimizerException(msg, errCode, PigException.BUG);
- }
- }
-
- // We recursively optimize a LOLimit, until one of the following conditions occurs:
- // 1. LOLimit can not move up
- // 2. LOLimit merged into another LOSort or another LOLimit
- // If we duplicate a LOLimit, then we leave the old LOLimit unmoved,
- // and recursively optimize the new LOLimit
- public void processNode(LOLimit limit) throws OptimizerException
- {
- try {
- List<LogicalOperator> predecessors = mPlan.getPredecessors(limit);
- if (predecessors.size()!=1) {
- int errCode = 2008;
- String msg = "Limit cannot have more than one input. Found " + predecessors.size() + " inputs.";
- throw new OptimizerException(msg, errCode, PigException.BUG);
- }
- LogicalOperator predecessor = predecessors.get(0);
-
- // Limit cannot be pushed up
- if (predecessor instanceof LOCogroup || predecessor instanceof LOFilter ||
- predecessor instanceof LOLoad || predecessor instanceof LOSplit ||
- predecessor instanceof LODistinct || predecessor instanceof LOJoin ||
- predecessor instanceof LOStream || predecessor instanceof LONative)
- {
- return;
- }
- // Limit can be pushed in front of ForEach if it does not have a flatten
- 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;
-
- // We can safely move LOLimit up
- if (!hasFlatten)
- {
- // Get operator before LOFilter
- LogicalOperator prepredecessor = mPlan.getPredecessors(predecessor).get(0);
- if (prepredecessor!=null)
- {
- try {
- mPlan.removeAndReconnect(limit);
- insertBetween(prepredecessor, limit, predecessor, null);
-
- } catch (Exception e) {
- int errCode = 2009;
- String msg = "Can not move LOLimit up";
- throw new OptimizerException(msg, errCode, PigException.BUG, e);
- }
- }
- else
- {
- int errCode = 2010;
- String msg = "LOForEach should have one input";
- throw new OptimizerException(msg, errCode, PigException.BUG);
- }
- // we can move LOLimit even further, recursively optimize LOLimit
- processNode(limit);
- }
- }
- // Limit can be duplicated, and the new instance pushed in front of an operator for the following operators
- // (that is, if you have X->limit, you can transform that to limit->X->limit):
- else if (predecessor instanceof LOCross || predecessor instanceof LOUnion)
- {
- LOLimit newLimit = null;
- List<LogicalOperator> nodesToProcess = new ArrayList<LogicalOperator>();
- for (LogicalOperator prepredecessor:mPlan.getPredecessors(predecessor))
- nodesToProcess.add(prepredecessor);
- for (LogicalOperator prepredecessor:nodesToProcess)
- {
- try {
- newLimit = limit.duplicate();
- insertBetween(prepredecessor, newLimit, predecessor, null);
- } catch (Exception e) {
- int errCode = 2011;
- String msg = "Can not insert LOLimit clone";
- throw new OptimizerException(msg, errCode, PigException.BUG, e);
- }
- // we can move the new LOLimit even further, recursively optimize LOLimit
- processNode(newLimit);
- }
- }
- // Limit can be merged into LOSort, result a "limited sort"
- else if (predecessor instanceof LOSort)
- {
- if(mode == ExecType.LOCAL) {
- //We don't need this optimisation to happen in the local mode.
- //so we do nothing here.
- } else {
- LOSort sort = (LOSort)predecessor;
- if (sort.getLimit()==-1)
- sort.setLimit(limit.getLimit());
- else
- sort.setLimit(sort.getLimit()<limit.getLimit()?sort.getLimit():limit.getLimit());
- try {
- mPlan.removeAndReconnect(limit);
- } catch (Exception e) {
- int errCode = 2012;
- String msg = "Can not remove LOLimit after LOSort";
- throw new OptimizerException(msg, errCode, PigException.BUG, e);
- }
- }
- }
- // Limit is merged into another LOLimit
- else if (predecessor instanceof LOLimit)
- {
- LOLimit beforeLimit = (LOLimit)predecessor;
- beforeLimit.setLimit(beforeLimit.getLimit()<limit.getLimit()?beforeLimit.getLimit():limit.getLimit());
- try {
- mPlan.removeAndReconnect(limit);
- } catch (Exception e) {
- int errCode = 2012;
- String msg = "Can not remove LOLimit after LOLimit";
- throw new OptimizerException(msg, errCode, PigException.BUG, e);
- }
- }
- // Limit and OrderBy (LOSort) can be separated by split
- else if (predecessor instanceof LOSplitOutput) {
- if(mode == ExecType.LOCAL) {
- //We don't need this optimisation to happen in the local mode.
- //so we do nothing here.
- } else {
- List<LogicalOperator> grandparants = mPlan
- .getPredecessors(predecessor);
- // After insertion of splitters, any node in the plan can
- // have at most one predecessor
- if (grandparants != null && grandparants.size() != 0
- && grandparants.get(0) instanceof LOSplit) {
- List<LogicalOperator> greatGrandparants = mPlan
- .getPredecessors(grandparants.get(0));
- if (greatGrandparants != null
- && greatGrandparants.size() != 0
- && greatGrandparants.get(0) instanceof LOSort) {
- LOSort sort = (LOSort)greatGrandparants.get(0);
- LOSort newSort = new LOSort(
- sort.getPlan(),
- new OperatorKey(
- sort.getOperatorKey().scope,
- NodeIdGenerator
- .getGenerator()
- .getNextNodeId(
- sort.getOperatorKey().scope)),
- sort.getSortColPlans(),
- sort.getAscendingCols(),
- sort.getUserFunc());
-
- newSort.setLimit(limit.getLimit());
- try {
- mPlan.replace(limit, newSort);
- } catch (PlanException e) {
- int errCode = 2012;
- String msg = "Can not replace LOLimit with LOSort after splitter";
- throw new OptimizerException(msg, errCode, PigException.BUG, e);
- }
- }
- }
- }
- }
- else {
- int errCode = 2013;
- String msg = "Moving LOLimit in front of " + predecessor.getClass().getSimpleName() + " is not implemented";
- throw new OptimizerException(msg, errCode, PigException.BUG);
- }
- } catch (OptimizerException oe) {
- throw oe;
- }
- }
-}
-
-
Modified: pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/PartitionFilterOptimizer.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/PartitionFilterOptimizer.java?rev=1100420&r1=1100419&r2=1100420&view=diff
==============================================================================
--- pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/PartitionFilterOptimizer.java (original)
+++ pig/branches/branch-0.9/src/org/apache/pig/impl/logicalLayer/optimizer/PartitionFilterOptimizer.java Sat May 7 00:15:40 2011
@@ -1,233 +0,0 @@
-/*
- * 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.io.IOException;
-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.hadoop.mapreduce.Job;
-import org.apache.pig.Expression;
-import org.apache.pig.LoadFunc;
-import org.apache.pig.LoadMetadata;
-import org.apache.pig.PigException;
-import org.apache.pig.Expression.BinaryExpression;
-import org.apache.pig.Expression.Column;
-import org.apache.pig.impl.logicalLayer.FrontendException;
-import org.apache.pig.impl.logicalLayer.LOFilter;
-import org.apache.pig.impl.logicalLayer.LOLoad;
-import org.apache.pig.impl.logicalLayer.LogicalOperator;
-import org.apache.pig.impl.logicalLayer.LogicalPlan;
-import org.apache.pig.impl.logicalLayer.PColFilterExtractor;
-import org.apache.pig.impl.logicalLayer.schema.Schema;
-import org.apache.pig.impl.plan.optimizer.OptimizerException;
-
-/**
- * When the load statement in a pig script is loading a table from a meta data
- * system (like owl), the load can be followed by a filter which can contain
- * conditions on partition columns. This filter can also contain conditions on
- * non partition columns. This optimizer looks at the logical plan and checks if
- * there is a load followed by such a filter which has conditions on partition
- * columns. If so, it extracts the conditions on partition columns out of the
- * filter.
- */
-public class PartitionFilterOptimizer extends
- LogicalTransformer {
-
- private String[] partitionKeys;
-
- /**
- * a reference to the LoadMetada implementation
- */
- private LoadMetadata loadMetadata;
-
- /**
- * a reference to the LoadFunc implementation
- */
- private LoadFunc loadFunc;
-
- private LOLoad loLoad;
- private LOFilter loFilter;
-
- /**
- * to ensure we only do the optimization once for performance reasons
- */
- private Set<LogicalOperator> alreadyChecked = new HashSet<LogicalOperator>();
-
- /**
- * a map between column names as reported in
- * {@link LoadMetadata#getSchema(String, org.apache.hadoop.conf.Configuration)}
- * and as present in {@link LOLoad#getSchema()}. The two will be different
- * when the user has provided a schema in the load statement
- */
- private Map<String, String> colNameMap = new HashMap<String, String>();
-
- /**
- * a map between column nameas as present in {@link LOLoad#getSchema()} and
- * as reported in
- * {@link LoadMetadata#getSchema(String, org.apache.hadoop.conf.Configuration)}.
- * The two will be different when the user has provided a schema in the
- * load statement.
- */
- private Map<String, String> reverseColNameMap = new HashMap<String, String>();
-
-
- protected PartitionFilterOptimizer(LogicalPlan plan) {
- super(plan);
- }
-
- @Override
- public boolean check(List<LogicalOperator> nodes) throws OptimizerException
- {
- 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);
- }
- if(nodes.size() != 1|| !(nodes.get(0) instanceof LOLoad )) {
- return false;
- }
- if (!alreadyChecked.add(nodes.get(0))) {
- return false;
- }
- if(nodes.get(0) instanceof LOLoad) {
- loLoad = (LOLoad)nodes.get(0);
- }
- List<LogicalOperator> sucs = mPlan.getSuccessors(loLoad);
- if(sucs == null || sucs.size() != 1 || !(sucs.get(0) instanceof LOFilter)) {
- return false;
- }
- loFilter = (LOFilter)sucs.get(0);
-
- // we have to check more only if LoadFunc implements LoadMetada
- loadFunc = loLoad.getLoadFunc();
- if(!(loadFunc instanceof LoadMetadata)) {
- return false;
- }
- loadMetadata = (LoadMetadata)loadFunc;
- try {
- partitionKeys = loadMetadata.getPartitionKeys(
- loLoad.getInputFile().getFileName(), new Job(loLoad.getConfiguration()));
- if(partitionKeys == null || partitionKeys.length == 0) {
- return false;
- }
- } catch (IOException e) {
- int errCode = 2209;
- throw new OptimizerException(
- "Internal error while processing any partition filter " +
- "conditions in the filter after the load" ,
- errCode,
- PigException.BUG
- );
- }
-
- // we found a load-filter pattern where the load returns partition keys
- return true;
- }
-
- @Override
- public void transform(List<LogicalOperator> nodes)
- throws OptimizerException {
- try {
- setupColNameMaps();
- PColFilterExtractor pColFilterFinder = new PColFilterExtractor(
- loFilter.getComparisonPlan(), getMappedKeys(partitionKeys));
- pColFilterFinder.visit();
- Expression partitionFilter = pColFilterFinder.getPColCondition();
- if(partitionFilter != null) {
- // the column names in the filter may be the ones provided by
- // the user in the schema in the load statement - we may need
- // to replace them with partition column names as given by
- // LoadFunc.getSchema()
- updateMappedColNames(partitionFilter);
- loadMetadata.setPartitionFilter(partitionFilter);
- if(pColFilterFinder.isFilterRemovable()) {
- // remove this filter from the plan
- mPlan.removeAndReconnect(loFilter);
- }
- }
- } catch (Exception e) {
- int errCode = 2209;
- throw new OptimizerException(
- "Internal error while processing any partition filter " +
- "conditions in the filter after the load:" ,
- errCode,
- PigException.BUG,
- e
- );
- }
- }
-
-
-
- /**
- * @param expr
- */
- private void updateMappedColNames(Expression expr) {
- if(expr instanceof BinaryExpression) {
- updateMappedColNames(((BinaryExpression) expr).getLhs());
- updateMappedColNames(((BinaryExpression) expr).getRhs());
- } else if (expr instanceof Column) {
- Column col = (Column) expr;
- col.setName(reverseColNameMap.get(col.getName()));
- }
- }
-
- /**
- * The partition keys in the argument are as reported by
- * {@link LoadMetadata#getPartitionKeys(String, org.apache.hadoop.conf.Configuration)}.
- * The user may have renamed these by providing a schema with different names
- * in the load statement - this method will replace the former names with
- * the latter names.
- * @param partitionKeys
- * @return
- */
- private List<String> getMappedKeys(String[] partitionKeys) {
- List<String> mappedKeys = new ArrayList<String>(partitionKeys.length);
- for (int i = 0; i < partitionKeys.length; i++) {
- mappedKeys.add(colNameMap.get(partitionKeys[i]));
- }
- return mappedKeys;
- }
-
-
-
- /**
- * @throws FrontendException
- *
- */
- private void setupColNameMaps() throws FrontendException {
- Schema loadFuncSchema = loLoad.getDeterminedSchema();
- Schema loLoadSchema = loLoad.getSchema();
- for(int i = 0; i < loadFuncSchema.size(); i++) {
- colNameMap.put(loadFuncSchema.getField(i).alias,
- (i < loLoadSchema.size() ? loLoadSchema.getField(i).alias :
- loadFuncSchema.getField(i).alias));
-
- reverseColNameMap.put((i < loLoadSchema.size() ? loLoadSchema.getField(i).alias :
- loadFuncSchema.getField(i).alias),
- loadFuncSchema.getField(i).alias);
- }
- }
-
-}