src/org/apache/pig/newplan/optimizer/Rule.java - pig - Git at Google

 /*
  * 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.optimizer;

 import java.util.ArrayList;
 import java.util.Deque;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Set;

 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.pig.impl.logicalLayer.FrontendException;
 import org.apache.pig.newplan.Operator;
 import org.apache.pig.newplan.OperatorPlan;
 import org.apache.pig.newplan.OperatorSubPlan;

 /**
  * Rules describe a pattern of operators.  They also reference a Transformer.
  * If the pattern of operators is found one or more times in the provided plan,
  * then the optimizer will use the associated Transformer to transform the
  * plan.
  * Note: the pattern matching logic implemented here has a limitation
  * that it assumes that all the leaves in the pattern are siblings. See more
  * detailed description here - https://issues.apache.org/jira/browse/PIG-1742
  * If new rules use patterns that don't work with this limitation, the pattern
  * match logic will need to be updated.
  */
 public abstract class Rule {

     protected String name = null;
     protected OperatorPlan pattern;
     transient protected OperatorPlan currentPlan;
     protected static final Log log = LogFactory.getLog(Rule.class);
     private transient Set<Operator> matchedNodes = new HashSet<Operator>();
     private boolean mandatory;
     private boolean skipListener = false;

     /**
      * Create this rule by using the default pattern that this rule provided
      * @param n Name of this rule
      * @param mandatory if it is set to false, this rule can be disabled by user
      */
     public Rule(String n, boolean mandatory) {
         name = n;
         pattern = buildPattern();
         this.mandatory = mandatory;
     }

     /**
      * @param n Name of this rule
      * @param p Pattern to look for.
      */
     public Rule(String n, OperatorPlan p) {
         name = n;
         pattern = p;
     }

     /**
      * Build the pattern that this rule will look for
      * @return  the pattern to look for by this rule
      */
     abstract protected OperatorPlan buildPattern();

     /**
      * Get the transformer for this rule.  Abstract because the rule
      * may want to choose how to instantiate the transformer.
      * This should never return a cached transformer, it should
      * always return a fresh one with no state.
      * @return Transformer to use with this rule
      */
     abstract public Transformer getNewTransformer();

     /**
      * Return the pattern to be matched for this rule
      * @return the pattern to be matched for this rule
      */
     public OperatorPlan getPattern() {
         return pattern;
     }

     protected boolean isSkipListener() {
         return this.skipListener;
     }

     protected void setSkipListener(boolean skip) {
         this.skipListener = skip;
     }

     /**
      * Search for all the sub-plans that matches the pattern
      * defined by this rule.
      * See class description above for limitations on the patterns supported.
      * @return A list of all matched sub-plans. The returned plans are
      *        partial views of the original OperatorPlan. Each is a
      *        sub-set of the original plan and represents the same
      *        topology as the pattern, but operators in the returned plan
      *        are the same objects as the original plan. Therefore,
      *        a call getPlan() from any node in the return plan would
      *        return the original plan.
      *
      * @param plan the OperatorPlan to look for matches to the pattern
      */
     public List<OperatorPlan> match(OperatorPlan plan) throws FrontendException {
         currentPlan = plan;

         List<Operator> leaves = pattern.getSinks();

         Iterator<Operator> iter = plan.getOperators();
         List<OperatorPlan> matchedList = new ArrayList<OperatorPlan>();
         matchedNodes.clear();

         while(iter.hasNext()) {
             Operator op = iter.next();

             // find a node that matches the first leaf of the pattern
             if (match(op, leaves.get(0))) {
                 List<Operator> planOps = new ArrayList<Operator>();
                 planOps.add(op);

                 // if there is more than 1 leaves in the pattern, we check
                 // if other leaves match the siblings of this node
                 if (leaves.size()>1) {
                     boolean matched = true;


                     List<Operator> preds = null;
                     preds = plan.getPredecessors(op);

                     // if this node has no predecessor, it must be a root
                     if (preds == null) {
                         preds = new ArrayList<Operator>();
                         preds.add(null);
                     }

                     for(Operator s: preds) {
                         matched = true;
                         List<Operator> siblings = null;
                         if (s != null) {
                             siblings = plan.getSuccessors(s);
                         }else{
                             // for a root, we get its siblings by getting all roots
                             siblings = plan.getSources();
                         }

                         int index = siblings.indexOf(op);
                         if (siblings.size()-index < leaves.size()) {
                             continue;
                         }


                         for(int j=1; j<leaves.size(); j++) {
                             if (!match(siblings.get(index+j), leaves.get(j))) {
                                 matched = false;
                                 break;
                             }
                         }

                         if (matched) {
                             for(int j=1; j<leaves.size(); j++) {
                                 planOps.add(siblings.get(index+j));
                             }
                         }

                     }

                     // we have move on to next operator as this one doesn't have siblings to
                     // match all the leaves
                     if (!matched) {
                         continue;
                     }
                 }


                 PatternMatchOperatorPlan match = new PatternMatchOperatorPlan(plan);
                 if (match.check(planOps)) {
                     // we find a matched pattern,
                     // add the operators into matchedNodes
                     Iterator<Operator> iter2 = match.getOperators();
                     while(iter2.hasNext()) {
                         Operator opt = iter2.next();
                         matchedNodes.add(opt);
                     }

                     // add pattern
                     matchedList.add(match);
                 }
             }
         }

         return matchedList;
     }

     public String getName() {
         return name;
     }

     public boolean isMandatory() {
         return mandatory;
     }

     /**
      * Check if two operators match each other, we want to find matches
      * that don't share nodes
      */
     private boolean match(Operator planNode, Operator patternNode) {
         return planNode.getClass().equals(patternNode.getClass()) && !matchedNodes.contains(planNode);
     }


     private class PatternMatchOperatorPlan extends OperatorSubPlan {

         public PatternMatchOperatorPlan(OperatorPlan basePlan) {
             super(basePlan);
         }

         protected boolean check(List<Operator> planOps) throws FrontendException {
             List<Operator> patternOps = pattern.getSinks();
             if (planOps.size() != patternOps.size()) {
                 return false;
             }

             for(int i=0; i<planOps.size(); i++) {
                 Deque<Operator> s = new LinkedList<Operator>();
                 if (!check(planOps.get(i), patternOps.get(i), s)) {
                     return false;
                 }
                 Iterator<Operator> iter = s.iterator();
                 while(iter.hasNext()) {
                     add(iter.next());
                 }
             }

             if (size() == pattern.size()) {
                 return true;
             }

             return false;
         }

         /**
          * Check if the plan operator and its sub-tree has a match to the pattern
          * operator and its sub-tree. This algorithm only search and return one match.
          * It doesn't recursively search for all possible matches. For example,
          * for a plan that looks like
          *                   join
          *                  /     \
          *                 load   load
          * if we are looking for join->load pattern, only one match will be returned instead
          * of two, so that the matched subsets don't share nodes.
          */
         private boolean check(Operator planOp, Operator patternOp, Deque<Operator> opers) throws FrontendException {
             if (!match(planOp, patternOp)) {
                 return false;
             }

             // check if their predecessors match
             List<Operator> preds1 = getBasePlan().getPredecessors(planOp);
             List<Operator> preds2 = pattern.getPredecessors(patternOp);
             if (preds1 == null && preds2 != null) {
                 return false;
             }

             if (preds1 != null && preds2 != null && preds1.size() < preds2.size()) {
                 return false;
             }

             // we've reached the root of the pattern, so a match is found
             if (preds2 == null || preds2.size() == 0) {
                 opers.push(planOp);
                 return true;
             }

             int index = 0;
             // look for predecessors
             while(index < preds1.size()) {
                 boolean match = true;
                 if (match(preds1.get(index), preds2.get(0))) {
                     if ( (preds1.size() - index) < preds2.size()) {
                         return false;
                     }

                     int oldSize = opers.size();
                     for(int i=0; i<preds2.size(); i++) {
                         if (!check(preds1.get(index+i), preds2.get(i), opers)) {
                             for(int j=opers.size(); j>oldSize; j--) {
                                 opers.pop();
                             }
                             match = false;
                             break;
                         }
                     }
                     if (match) {
                         opers.push(planOp);
                         return true;
                     }
                 }
                 index++;
             }

             return false;
         }
     }
 }
	/*
	* 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.optimizer;

	import java.util.ArrayList;
	import java.util.Deque;
	import java.util.HashSet;
	import java.util.Iterator;
	import java.util.LinkedList;
	import java.util.List;
	import java.util.Set;

	import org.apache.commons.logging.Log;
	import org.apache.commons.logging.LogFactory;
	import org.apache.pig.impl.logicalLayer.FrontendException;
	import org.apache.pig.newplan.Operator;
	import org.apache.pig.newplan.OperatorPlan;
	import org.apache.pig.newplan.OperatorSubPlan;

	/**
	* Rules describe a pattern of operators. They also reference a Transformer.
	* If the pattern of operators is found one or more times in the provided plan,
	* then the optimizer will use the associated Transformer to transform the
	* plan.
	* Note: the pattern matching logic implemented here has a limitation
	* that it assumes that all the leaves in the pattern are siblings. See more
	* detailed description here - https://issues.apache.org/jira/browse/PIG-1742
	* If new rules use patterns that don't work with this limitation, the pattern
	* match logic will need to be updated.
	*/
	public abstract class Rule {

	protected String name = null;
	protected OperatorPlan pattern;
	transient protected OperatorPlan currentPlan;
	protected static final Log log = LogFactory.getLog(Rule.class);
	private transient Set<Operator> matchedNodes = new HashSet<Operator>();
	private boolean mandatory;
	private boolean skipListener = false;

	/**
	* Create this rule by using the default pattern that this rule provided
	* @param n Name of this rule
	* @param mandatory if it is set to false, this rule can be disabled by user
	*/
	public Rule(String n, boolean mandatory) {
	name = n;
	pattern = buildPattern();
	this.mandatory = mandatory;
	}

	/**
	* @param n Name of this rule
	* @param p Pattern to look for.
	*/
	public Rule(String n, OperatorPlan p) {
	name = n;
	pattern = p;
	}

	/**
	* Build the pattern that this rule will look for
	* @return the pattern to look for by this rule
	*/
	abstract protected OperatorPlan buildPattern();

	/**
	* Get the transformer for this rule. Abstract because the rule
	* may want to choose how to instantiate the transformer.
	* This should never return a cached transformer, it should
	* always return a fresh one with no state.
	* @return Transformer to use with this rule
	*/
	abstract public Transformer getNewTransformer();

	/**
	* Return the pattern to be matched for this rule
	* @return the pattern to be matched for this rule
	*/
	public OperatorPlan getPattern() {
	return pattern;
	}

	protected boolean isSkipListener() {
	return this.skipListener;
	}

	protected void setSkipListener(boolean skip) {
	this.skipListener = skip;
	}

	/**
	* Search for all the sub-plans that matches the pattern
	* defined by this rule.
	* See class description above for limitations on the patterns supported.
	* @return A list of all matched sub-plans. The returned plans are
	* partial views of the original OperatorPlan. Each is a
	* sub-set of the original plan and represents the same
	* topology as the pattern, but operators in the returned plan
	* are the same objects as the original plan. Therefore,
	* a call getPlan() from any node in the return plan would
	* return the original plan.
	*
	* @param plan the OperatorPlan to look for matches to the pattern
	*/
	public List<OperatorPlan> match(OperatorPlan plan) throws FrontendException {
	currentPlan = plan;

	List<Operator> leaves = pattern.getSinks();

	Iterator<Operator> iter = plan.getOperators();
	List<OperatorPlan> matchedList = new ArrayList<OperatorPlan>();
	matchedNodes.clear();

	while(iter.hasNext()) {
	Operator op = iter.next();

	// find a node that matches the first leaf of the pattern
	if (match(op, leaves.get(0))) {
	List<Operator> planOps = new ArrayList<Operator>();
	planOps.add(op);

	// if there is more than 1 leaves in the pattern, we check
	// if other leaves match the siblings of this node
	if (leaves.size()>1) {
	boolean matched = true;


	List<Operator> preds = null;
	preds = plan.getPredecessors(op);

	// if this node has no predecessor, it must be a root
	if (preds == null) {
	preds = new ArrayList<Operator>();
	preds.add(null);
	}

	for(Operator s: preds) {
	matched = true;
	List<Operator> siblings = null;
	if (s != null) {
	siblings = plan.getSuccessors(s);
	}else{
	// for a root, we get its siblings by getting all roots
	siblings = plan.getSources();
	}

	int index = siblings.indexOf(op);
	if (siblings.size()-index < leaves.size()) {
	continue;
	}


	for(int j=1; j<leaves.size(); j++) {
	if (!match(siblings.get(index+j), leaves.get(j))) {
	matched = false;
	break;
	}
	}

	if (matched) {
	for(int j=1; j<leaves.size(); j++) {
	planOps.add(siblings.get(index+j));
	}
	}

	}

	// we have move on to next operator as this one doesn't have siblings to
	// match all the leaves
	if (!matched) {
	continue;
	}
	}


	PatternMatchOperatorPlan match = new PatternMatchOperatorPlan(plan);
	if (match.check(planOps)) {
	// we find a matched pattern,
	// add the operators into matchedNodes
	Iterator<Operator> iter2 = match.getOperators();
	while(iter2.hasNext()) {
	Operator opt = iter2.next();
	matchedNodes.add(opt);
	}

	// add pattern
	matchedList.add(match);
	}
	}
	}

	return matchedList;
	}

	public String getName() {
	return name;
	}

	public boolean isMandatory() {
	return mandatory;
	}

	/**
	* Check if two operators match each other, we want to find matches
	* that don't share nodes
	*/
	private boolean match(Operator planNode, Operator patternNode) {
	return planNode.getClass().equals(patternNode.getClass()) && !matchedNodes.contains(planNode);
	}


	private class PatternMatchOperatorPlan extends OperatorSubPlan {

	public PatternMatchOperatorPlan(OperatorPlan basePlan) {
	super(basePlan);
	}

	protected boolean check(List<Operator> planOps) throws FrontendException {
	List<Operator> patternOps = pattern.getSinks();
	if (planOps.size() != patternOps.size()) {
	return false;
	}

	for(int i=0; i<planOps.size(); i++) {
	Deque<Operator> s = new LinkedList<Operator>();
	if (!check(planOps.get(i), patternOps.get(i), s)) {
	return false;
	}
	Iterator<Operator> iter = s.iterator();
	while(iter.hasNext()) {
	add(iter.next());
	}
	}

	if (size() == pattern.size()) {
	return true;
	}

	return false;
	}

	/**
	* Check if the plan operator and its sub-tree has a match to the pattern
	* operator and its sub-tree. This algorithm only search and return one match.
	* It doesn't recursively search for all possible matches. For example,
	* for a plan that looks like
	* join
	* / \
	* load load
	* if we are looking for join->load pattern, only one match will be returned instead
	* of two, so that the matched subsets don't share nodes.
	*/
	private boolean check(Operator planOp, Operator patternOp, Deque<Operator> opers) throws FrontendException {
	if (!match(planOp, patternOp)) {
	return false;
	}

	// check if their predecessors match
	List<Operator> preds1 = getBasePlan().getPredecessors(planOp);
	List<Operator> preds2 = pattern.getPredecessors(patternOp);
	if (preds1 == null && preds2 != null) {
	return false;
	}

	if (preds1 != null && preds2 != null && preds1.size() < preds2.size()) {
	return false;
	}

	// we've reached the root of the pattern, so a match is found
	if (preds2 == null \|\| preds2.size() == 0) {
	opers.push(planOp);
	return true;
	}

	int index = 0;
	// look for predecessors
	while(index < preds1.size()) {
	boolean match = true;
	if (match(preds1.get(index), preds2.get(0))) {
	if ( (preds1.size() - index) < preds2.size()) {
	return false;
	}

	int oldSize = opers.size();
	for(int i=0; i<preds2.size(); i++) {
	if (!check(preds1.get(index+i), preds2.get(i), opers)) {
	for(int j=opers.size(); j>oldSize; j--) {
	opers.pop();
	}
	match = false;
	break;
	}
	}
	if (match) {
	opers.push(planOp);
	return true;
	}
	}
	index++;
	}

	return false;
	}
	}
	}