| /** |
| * 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.newplan.logical.rules; |
| |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.Map.Entry; |
| |
| import org.apache.pig.impl.logicalLayer.FrontendException; |
| import org.apache.pig.impl.util.Pair; |
| import org.apache.pig.newplan.Operator; |
| import org.apache.pig.newplan.OperatorPlan; |
| import org.apache.pig.newplan.OperatorSubPlan; |
| import org.apache.pig.newplan.logical.expression.LogicalExpression; |
| import org.apache.pig.newplan.logical.expression.LogicalExpressionPlan; |
| import org.apache.pig.newplan.logical.expression.ProjectExpression; |
| import org.apache.pig.newplan.logical.expression.UserFuncExpression; |
| import org.apache.pig.newplan.logical.relational.LOCogroup; |
| import org.apache.pig.newplan.logical.relational.LOCross; |
| import org.apache.pig.newplan.logical.relational.LODistinct; |
| import org.apache.pig.newplan.logical.relational.LOFilter; |
| import org.apache.pig.newplan.logical.relational.LOForEach; |
| import org.apache.pig.newplan.logical.relational.LOJoin; |
| import org.apache.pig.newplan.logical.relational.LOLimit; |
| import org.apache.pig.newplan.logical.relational.LOLoad; |
| import org.apache.pig.newplan.logical.relational.LONative; |
| import org.apache.pig.newplan.logical.relational.LOSort; |
| import org.apache.pig.newplan.logical.relational.LOSplit; |
| import org.apache.pig.newplan.logical.relational.LOSplitOutput; |
| import org.apache.pig.newplan.logical.relational.LOStore; |
| import org.apache.pig.newplan.logical.relational.LOStream; |
| import org.apache.pig.newplan.logical.relational.LOUnion; |
| import org.apache.pig.newplan.logical.relational.LogicalPlan; |
| import org.apache.pig.newplan.logical.relational.LogicalRelationalOperator; |
| import org.apache.pig.newplan.logical.relational.LogicalSchema; |
| import org.apache.pig.newplan.optimizer.Rule; |
| import org.apache.pig.newplan.optimizer.Transformer; |
| |
| public class PushUpFilter extends Rule { |
| |
| public PushUpFilter(String n) { |
| super(n, false); |
| } |
| |
| @Override |
| public Transformer getNewTransformer() { |
| return new PushUpFilterTransformer(); |
| } |
| |
| public class PushUpFilterTransformer extends Transformer { |
| private OperatorSubPlan subPlan; |
| |
| @Override |
| public boolean check(OperatorPlan matched) throws FrontendException { |
| // check if it is inner join |
| Operator current = matched.getSources().get(0); |
| |
| Operator pred = findNonFilterPredecessor( current ); |
| if( pred == null ) |
| return false; |
| |
| // sort and union are always okay. |
| if( pred instanceof LOSort || pred instanceof LOUnion ) { |
| return true; |
| } |
| |
| // if the predecessor is one of LOLoad/LOStore/LOStream/LOLimit/LONative |
| // if predecessor is LOForEach, it is optimized by rule FilterAboveForeach |
| // return false |
| if( pred instanceof LOLoad || pred instanceof LOStore || pred instanceof LOStream || |
| pred instanceof LOFilter || pred instanceof LOSplit || pred instanceof LOSplitOutput || |
| pred instanceof LOLimit || pred instanceof LONative || pred instanceof LOForEach) { |
| return false; |
| } |
| |
| LOFilter filter = (LOFilter)current; |
| List<Operator> preds = currentPlan.getPredecessors( pred ); |
| LogicalExpressionPlan filterPlan = filter.getFilterPlan(); |
| |
| if (OptimizerUtils.planHasNonDeterministicUdf(filterPlan)) { |
| return false; |
| } |
| |
| //if there is no nondeterministic udf, filter can be pushed above |
| // Distinct |
| if(pred instanceof LODistinct){ |
| return true; |
| } |
| |
| // collect all uids used in the filter plan |
| Set<Long> uids = collectUidFromExpPlan(filterPlan); |
| |
| if( pred instanceof LOCogroup ) { |
| LOCogroup cogrp = (LOCogroup)pred; |
| if( preds.size() == 1 ) { |
| if( hasAll( (LogicalRelationalOperator)preds.get( 0 ), uids ) ) { |
| // Order by is ok if all UIDs can be found from previous operator. |
| return true; |
| } |
| } else if ( 1 == cogrp.getExpressionPlans().get( 0 ).size() && !containUDF( filterPlan ) ) { |
| // Optimization is possible if there is only a single key. |
| // For regular cogroup, we cannot use UIDs to determine if filter can be pushed up. |
| // But if there is no UDF, it's okay, as only UDF can take bag field as input. |
| return true; |
| } |
| } |
| |
| // if the predecessor is a multi-input operator then detailed |
| // checks are required |
| if( pred instanceof LOCross || pred instanceof LOJoin ) { |
| boolean[] innerFlags = null; |
| boolean isFullOuter = true; |
| boolean isInner = true; |
| if( pred instanceof LOJoin ) { |
| innerFlags = ((LOJoin)pred).getInnerFlags(); |
| // If all innerFlag is false, means a full outer join, |
| for (boolean inner : innerFlags) { |
| if (inner) { |
| isFullOuter = false; |
| } else { |
| isInner = false; |
| } |
| } |
| if (isFullOuter) |
| return false; |
| } |
| |
| |
| for(int j=0; j<preds.size(); j++) { |
| if (hasAll((LogicalRelationalOperator)preds.get(j), uids)) { |
| // For LOJoin, innerFlag==true indicate that branch is the outer join side |
| // which has the exact opposite semantics |
| if (pred instanceof LOCross || pred instanceof LOJoin && (isInner || innerFlags[j])) |
| return true; |
| } |
| } |
| } |
| |
| return false; |
| } |
| |
| private boolean containUDF(LogicalExpressionPlan filterPlan) { |
| Iterator<Operator> it = filterPlan.getOperators(); |
| while( it.hasNext() ) { |
| if( it.next() instanceof UserFuncExpression ) |
| return true; |
| } |
| return false; |
| } |
| |
| Set<Long> collectUidFromExpPlan(LogicalExpressionPlan filterPlan) throws FrontendException { |
| Set<Long> uids = new HashSet<Long>(); |
| Iterator<Operator> iter = filterPlan.getOperators(); |
| while(iter.hasNext()) { |
| Operator op = iter.next(); |
| if (op instanceof ProjectExpression) { |
| long uid = ((ProjectExpression)op).getFieldSchema().uid; |
| uids.add(uid); |
| } |
| } |
| return uids; |
| } |
| |
| /** |
| * Starting from current operator (which is a filter), search its successors until |
| * locating a non-filter operator. Null is returned if none is found. |
| */ |
| private Operator findNonFilterPredecessor(Operator current) { |
| Operator op = current; |
| do { |
| List<Operator> predecessors = currentPlan.getPredecessors( op ); |
| |
| // if there are no predecessors return false |
| if( predecessors == null || predecessors.size() == 0 ) { |
| return null; |
| } |
| |
| Operator pred = predecessors.get( 0 ); |
| if( pred instanceof LOFilter ) { |
| op = pred; |
| continue; |
| } else { |
| return pred; |
| } |
| } while( true ); |
| |
| } |
| |
| @Override |
| public void transform(OperatorPlan matched) throws FrontendException { |
| subPlan = new OperatorSubPlan(currentPlan); |
| |
| LOFilter filter = (LOFilter)matched.getSources().get(0); |
| |
| // This is the one that we will insert filter btwn it and it's input. |
| Operator predecessor = this.findNonFilterPredecessor( filter ); |
| subPlan.add( predecessor) ; |
| |
| // Disconnect the filter in the plan without removing it from the plan. |
| Operator predec = currentPlan.getPredecessors( filter ).get( 0 ); |
| Operator succed; |
| |
| if (currentPlan.getSuccessors(filter)!=null) |
| succed = currentPlan.getSuccessors(filter).get(0); |
| else |
| succed = null; |
| |
| Pair<Integer, Integer> p1 = currentPlan.disconnect(predec, filter); |
| if (succed!=null) { |
| subPlan.add(succed); |
| Pair<Integer, Integer> p2 = currentPlan.disconnect(filter, succed); |
| currentPlan.connect(predec, p1.first, succed, p2.second); |
| } |
| |
| if( predecessor instanceof LOSort || predecessor instanceof LODistinct || |
| ( predecessor instanceof LOCogroup && currentPlan.getPredecessors( predecessor ).size() == 1 ) ) { |
| // For sort, put the filter in front of it. |
| Operator prev = currentPlan.getPredecessors( predecessor ).get( 0 ); |
| |
| insertFilter( prev, predecessor, filter ); |
| return; |
| } |
| |
| // Find the predecessor of join that contains all required uids. |
| LogicalExpressionPlan filterPlan = filter.getFilterPlan(); |
| List<Operator> preds = currentPlan.getPredecessors( predecessor ); |
| Map<Integer, Operator> inputs = findInputsToAddFilter( filterPlan, predecessor, preds ); |
| |
| LOFilter newFilter = null; |
| for( Entry<Integer, Operator> entry : inputs.entrySet() ) { |
| int inputIndex = entry.getKey(); |
| Operator pred = entry.getValue(); |
| |
| // Find projection field offset |
| int columnOffset = 0; |
| if( predecessor instanceof LOJoin || predecessor instanceof LOCross ) { |
| for( int i = 0; i < inputIndex; i++ ) { |
| columnOffset += ( (LogicalRelationalOperator)preds.get( i ) ).getSchema().size(); |
| } |
| } |
| |
| // Reuse the filter for the first match. For others, need to make a copy of the filter |
| // and add it between input and predecessor. |
| newFilter = newFilter == null ? filter : new LOFilter( (LogicalPlan)currentPlan ); |
| |
| currentPlan.add( newFilter ); |
| subPlan.add( newFilter ); |
| subPlan.add( pred ); |
| LogicalExpressionPlan fPlan = filterPlan.deepCopy(); |
| List<Operator> sinks = fPlan.getSinks(); |
| List<ProjectExpression> projExprs = new ArrayList<ProjectExpression>(); |
| for( Operator sink : sinks ) { |
| if( sink instanceof ProjectExpression ) |
| projExprs.add( (ProjectExpression)sink ); |
| } |
| |
| if( predecessor instanceof LOCogroup ) { |
| for( ProjectExpression projExpr : projExprs ) { |
| // Need to merge filter condition and cogroup by expression; |
| LogicalExpressionPlan plan = ((LOCogroup) predecessor).getExpressionPlans().get( inputIndex ).iterator().next(); |
| LogicalExpressionPlan copy = plan.deepCopy(); |
| LogicalExpression root = (LogicalExpression)copy.getSinks().get( 0 ); |
| List<Operator> predecessors = fPlan.getPredecessors( projExpr ); |
| if( predecessors == null || predecessors.size() == 0 ) { |
| fPlan.remove( projExpr ); |
| fPlan.add( root ); |
| } else { |
| fPlan.add( root ); |
| Operator pred1 = predecessors.get( 0 ); |
| Pair<Integer, Integer> pair = fPlan.disconnect( pred1, projExpr ); |
| fPlan.connect( pred1, pair.first, root, pair.second ); |
| fPlan.remove( projExpr ); |
| } |
| } |
| } |
| |
| // Now, reset the projection expressions in the new filter plan. |
| sinks = fPlan.getSinks(); |
| for( Operator sink : sinks ) { |
| if( sink instanceof ProjectExpression ) { |
| ProjectExpression projE = (ProjectExpression)sink; |
| projE.setAttachedRelationalOp( newFilter ); |
| projE.setInputNum( 0 ); |
| projE.setColNum( projE.getColNum() - columnOffset ); |
| } |
| } |
| newFilter.setFilterPlan( fPlan ); |
| |
| insertFilter( pred, predecessor, newFilter ); |
| } |
| } |
| |
| // check if a relational operator contains all of the specified uids |
| private boolean hasAll(LogicalRelationalOperator op, Set<Long> uids) throws FrontendException { |
| LogicalSchema schema = op.getSchema(); |
| if (schema==null) |
| return false; |
| for(long uid: uids) { |
| if (schema.findField(uid) == -1) { |
| return false; |
| } |
| } |
| |
| return true; |
| } |
| |
| @Override |
| public OperatorPlan reportChanges() { |
| return currentPlan; |
| } |
| |
| // Insert the filter in between the given two operators. |
| private void insertFilter(Operator prev, Operator predecessor, LOFilter filter) |
| throws FrontendException { |
| Pair<Integer, Integer> p3 = currentPlan.disconnect( prev, predecessor ); |
| currentPlan.connect( prev, p3.first, filter, 0 ); |
| currentPlan.connect( filter, 0, predecessor, p3.second ); |
| } |
| |
| // Identify those among preds that will need to have a filter between it and the predecessor. |
| private Map<Integer, Operator> findInputsToAddFilter(LogicalExpressionPlan filterPlan, Operator predecessor, |
| List<Operator> preds) throws FrontendException { |
| Map<Integer, Operator> inputs = new HashMap<Integer, Operator>(); |
| |
| if( predecessor instanceof LOUnion || predecessor instanceof LOCogroup ) { |
| for( int i = 0; i < preds.size(); i++ ) { |
| inputs.put( i, preds.get( i ) ); |
| } |
| return inputs; |
| } |
| |
| // collect all uids used in the filter plan |
| Set<Long> uids = collectUidFromExpPlan(filterPlan); |
| boolean[] innerFlags = null; |
| boolean isInner = true; |
| if (predecessor instanceof LOJoin) { |
| innerFlags = ((LOJoin)predecessor).getInnerFlags(); |
| for (boolean inner : innerFlags) { |
| if (!inner) { |
| isInner = false; |
| break; |
| } |
| } |
| } |
| |
| // Find the predecessor of join that contains all required uids. |
| for(int j=0; j<preds.size(); j++) { |
| // Filter can push to LOJoin outer branch, but no inner branch |
| if( hasAll((LogicalRelationalOperator)preds.get(j), uids) && |
| (predecessor instanceof LOCross || predecessor instanceof LOJoin && (isInner || innerFlags[j]))) { |
| Operator input = preds.get(j); |
| subPlan.add(input); |
| inputs.put( j, input ); |
| } |
| } |
| return inputs; |
| } |
| } |
| |
| @Override |
| protected OperatorPlan buildPattern() { |
| LogicalPlan plan = new LogicalPlan(); |
| LogicalRelationalOperator op1 = new LOFilter(plan); |
| plan.add( op1 ); |
| |
| return plan; |
| } |
| |
| } |
| |
