Pig源码分析: 逻辑执行计划优化

Whole View

本文分析的是逻辑执行计划优化的代码结构，具体每种Rule的实现不做分析。

看本文之前最好参考之前那篇逻辑执行计划模型的文章。

Architecture

几个关键类/接口的关系：

每个关键类/接口的实现和继承结构在下面各节展开。

Optimizer

PlanOptimizer是抽象类，主要和Rule、PlanTransformListener、OperatorPlan打交道。

public abstract class PlanOptimizer {
 
    protected List<Set<Rule>> ruleSets;
    protected OperatorPlan plan;
    protected List<PlanTransformListener> listeners;
    protected int maxIter;

它接受一个OperatorPlan，即Operators的DAG模型，在optimize()方法里，遍历ruleSet，得到几批Rules，即Set<Rule>。对于每批Rules，调用每个rule.match(plan)来处理传入的OperatorPlan，返回一个匹配成功的List<OperatorPlan> matches，对这些match的plans进行进一步处理。首先获得rule的transformer，然后进行transformer的check()和transform()操作。如果需要Listener操作的，还会遍历listeners，让每个PlanTransformListener监听到transformer进行的transform操作，transformer的reportChanges()方法可以返回他transform操作修改的部分。

代码如下：

    public void optimize() throws FrontendException {

        for (Set<Rule> rs : ruleSets) {
            boolean sawMatch = false;
            int numIterations = 0;
            do {
                sawMatch = false;
                for (Rule rule : rs) {
                    List<OperatorPlan> matches = rule.match(plan);
                    if (matches != null) {
                        Transformer transformer = rule.getNewTransformer();
                        for (OperatorPlan m : matches) {
                            try {
                                if (transformer.check(m)) {
                                    sawMatch = true;
                                    transformer.transform(m);
                                    if (!rule.isSkipListener()) {
                                        for(PlanTransformListener l: listeners) {
                                            l.transformed(plan, transformer.reportChanges());
                                        }
                                    }
                                }
                            } catch (Exception e) {
                                StringBuffer message = new StringBuffer("Error processing rule " + rule.name);
                                if (!rule.isMandatory()) {
                                    message.append(". Try -t " + rule.name);
                                }
                                throw new FrontendException(message.toString(), 2000, e);
                            }
                        }
                    }
                }
            } while(sawMatch && ++numIterations < maxIter);
        }
    }

实现类：

LogicalPlanOptimizer

LogicalPlanOptimizer类是PlanOptimizer的子类

默认加载两个Listener：

Listener的这两个实现在PlanTransformerListener一节具体展开讲述。

初始化的时候会buildRuleSets()，把需要添加的Rule都生成出来，然后校对该Rule是否被强制加入，或被turn off，从而选择性地放入优化规则。以下列举了所有候选的优化规则，Rule是顺序执行的：

Transformer

Transformer是抽象类，有三个方法需要子类实现：

check()方法，利用pattern来匹配plan里符合的operator集合，返回match的operator集

transform()方法，具体实施对tree的转换操作

reportChanges()方法，报告tree的哪部分被transform操作过了（只包括被修改了的或增加了的，不包括删除的node），目的是为了让Listener得知，从而可以修改schema或annotation等等。

继承结构如下：

PlanTransformerListener

PlanTransformListener监听一个plan被修改后会触发。

举例：

当一个Rule把一次join里的Filter步骤提前到join操作之间做，那么过滤部分的input schema很可能需要改变，此时一个schema listener就会被触发并执行。

PlanTransformListener是一个接口，需要实现一个方法：

public void transformed(OperatorPlan fp, OperatorPlan tp) throws FrontendException;

下面具体介绍两个实现类

ProjectionPatcher

作用是在映射操作中修补引用信息

有两个内部静态类

SchemaPatcher

使用于逻辑执行计划优化过程，当plan被transform了之后修补schema信息。

Rule

public abstract class Rule {
    protected String name = null;
    protected OperatorPlan pattern;
    transient protected OperatorPlan currentPlan;
private transient Set<Operator> matchedNodes = 
new HashSet<Operator>();
    private boolean mandatory;
    private boolean skipListener = false;

Rule已经把 match(OperatorPlan plan)方法的逻辑实现好了。

子类需要实现的是buildPattern()方法，来制定各自的”模式”，即pattern变量。

子类还需要实现getNewTransformer()方法来实例化一个transformer，transformer的check()和transform()方法会进一步处理rule匹配的operators。

Rulematch()的用途是确保plan的所有子plan都满足该rule的pattern。

实现逻辑比较繁杂。

Rule继承结构

具体每个Rule不分析了。

全文完 :)