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不分析了。

全文完 :)