本文分析的是Pig Logical模块的代码(newplan package下)，具体每种逻辑执行的实现类不会做具体分析。

public abstract class Operator {
    protected SourceLocation location; // The location of the operator in the original pig script.
    
    protected String name;
    protected OperatorPlan plan; // plan that contains this operator
    protected Map<String, Object> annotations;
    protected final int hashPrime = 31;

public abstract class LogicalExpression extends Operator {

    static long nextUid = 1;
    protected LogicalSchema.LogicalFieldSchema fieldSchema;
    protected LogicalSchema.LogicalFieldSchema uidOnlyFieldSchema;

deepCopy()方法需要子类实现

 

继承结构：

LogicalRelationalOperator代表关系型操作，关系型操作有Schema。以下是主要变量，LogicalRelationalOperator为他们提供了一些get/set方法。

abstract public class LogicalRelationalOperator extends Operator {
    
    protected LogicalSchema schema;
    protected int requestedParallelism;
    protected String alias;
    protected int lineNum;
    
    /**
     * Name of the customPartitioner if one is used, this is set to null otherwise.
     */
    protected String mCustomPartitioner = null;
    
    /**
     * A HashSet to indicate whether an option (such a Join Type) was pinned
     * by the user or can be chosen at runtime by the optimizer.
     */
    protected HashSet<Integer> mPinnedOptions = new HashSet<Integer>();

关于LogicalSchema类：

内部类LogicalFieldSchema具体表示每一个field的结构，可以看到与LogicalSchema是嵌套的。LogicalSchema维护一个List<LogicalFieldSchema>

    public static class LogicalFieldSchema {
        public String alias;
        public byte type;
        public long uid;
        public LogicalSchema schema;

提供基本方法如下：

除了基本方法外，还提供一套merge schema的方法

OperatorSubPlan代表的是一个OperatorPlan的一个子集的视图，OperatorSubPlan只有一个实现，使用在Rule的match过程里。所以OperatorSubPlan的作用就是提供一个子Plan，用于匹配操作。

BaseOperatorPlan实现了OperatorPlan接口，具体实现了各个图操作方法，把Operator之间的关系（包括softLink关系）用PlanEdge表示，图操作方法都借助PlanEdge类表达和实现。

public abstract class BaseOperatorPlan implements OperatorPlan {

    protected List<Operator> ops;
    protected PlanEdge fromEdges;
    protected PlanEdge toEdges;
    protected PlanEdge softFromEdges;
    protected PlanEdge softToEdges;

    private List<Operator> roots;
    private List<Operator> leaves;

比如：

toEdges.get(op)                  返回op的前辈

toEdges.get(op)==null       的op为root

fromEdges.get(op)             返回op的后辈

fromEdges.get(op)==null  的op为leave

public class PlanEdge extends MultiMap<Operator, Operator>

public class MultiMap<K, V> implements Serializable {

    // Change this if you modify the class.
    static final long serialVersionUID = 2L;

    protected Map<K, ArrayList<V>> mMap = null;

    public MultiMap() {
        mMap = new HashMap<K, ArrayList<V>>();
    }

public Pair<Integer, Integer> disconnect(Operator from, Operator to)
public void connect(Operator from, int fromPos, Operator to, int toPos)

LogicalPlan只包含关系型操作，也就是说涉及到的Operator都是LogicalRelationalOperator。

 

explain()方法既支持LogicalPlanPrinter的visit实现，也支持DotLOPrinter的dpump实现。

LogicalPlanPrinter是PlanVisitor的子类， LogicalPlanPrinter内部有一个PrintStream，在visit()过程中边遍历，边记录。

DotLOPrinter是DotPlanDumper的子类，DotPlanDumper是PlanDumper的子类，根据graphviz的dot algorithm，输出符合DOT格式的plan。

LogicalExpressionPlan处理的是LogicalExpressionOperators。

 

explain()方法借助LogicalPlanPrinter实现

PlanWalker提供的是遍历访问一个plan的能力。

PlanWalker的子类主要实现两个方法：

public abstract void walk(PlanVisitor visitor) throws FrontendException;

public abstract PlanWalker spawnChildWalker(OperatorPlan plan);

walk()方法在子类的实现中，会以不同的顺序遍历plan，最后的结果是遍历到的节点Operator会调op.accept(visitor)接受本Visitor。

继承结构

接下来具体介绍各子类遍历能力的实现。

DependencyOrderWalker按照依赖顺序访问plan，即一个node被访问的前提是它的前辈们已经被访问过了。这个访问顺序相当于，按照拓扑顺序访问图上的节点。

@Override
public PlanWalker spawnChildWalker(OperatorPlan plan) {
return new DependencyOrderWalker(plan);
}

walk()方法通过plan.getSinks()方法得到所有的leave节点，即没有后辈的节点，然后遍历他们，获取每个节点的所有前辈，再递归前辈的前辈，从而实现把所有的节点都访问一遍，最后得到结果就是一个FIFO的List。代码里的这个Graph依赖遍历的方式很不高效，但是因为访问的图的节点少，所以可接受。

递归的过程如下

    protected void doAllPredecessors(Operator node,
                                   Set<Operator> seen,
                                   Collection<Operator> fifo) throws FrontendException {
        if (!seen.contains(node)) {
            // We haven't seen this one before.
            Collection<Operator> preds = Utils.mergeCollection(plan.getPredecessors(node), plan.getSoftLinkPredecessors(node));
            if (preds != null && preds.size() > 0) {
                // Do all our predecessors before ourself
                for (Operator op : preds) {
                    doAllPredecessors(op, seen, fifo);
                }
            }
            // Now do ourself
            seen.add(node);
            fifo.add(node);
        }
    }

@Override
public PlanWalker spawnChildWalker(OperatorPlan plan) {
return new DepthFirstWalker(plan);
}

    private void depthFirst(Operator node,
                            Collection<Operator> successors,
                            Set<Operator> seen,
                            PlanVisitor visitor) throws FrontendException {
        if (successors == null) return;

        for (Operator suc : successors) {
            if (seen.add(suc)) {
                suc.accept(visitor);
                Collection<Operator> newSuccessors = Utils.mergeCollection(plan.getSuccessors(suc), plan.getSoftLinkSuccessors(suc));
                depthFirst(suc, newSuccessors, seen, visitor);
            }
        }
    }

public PlanWalker spawnChildWalker(OperatorPlan plan) {
return new DepthFirstWalker(plan);
}

    private void depthFirst(Operator node, Collection<Operator> predecessors, Set<Operator> seen,
            PlanVisitor visitor) throws FrontendException {
        if (predecessors == null)
            return;

        boolean thisBranchFlag = branchFlag;
        for (Operator pred : predecessors) {
            if (seen.add(pred)) {
                branchFlag = thisBranchFlag;
                pred.accept(visitor);
                Collection<Operator> newPredecessors = Utils.mergeCollection(plan.getPredecessors(pred), plan.getSoftLinkPredecessors(pred));
                depthFirst(pred, newPredecessors, seen, visitor);
            }
        }
    }

ReserveDependencyOrderWalker是逆向的依赖顺序遍历，即一个节点访问之后才能访问它依赖的节点，即N节点要想被访问，需要依赖N节点的节点先被访问。

@Override
public PlanWalker spawnChildWalker(OperatorPlan plan) {
return new ReverseDependencyOrderWalker(plan);
}

walk()方法的访问模式类似DependencyOrderWalker，区别在于先获得所有的root节点，然后进行遍历操作，遍历root节点的所有后辈，递归后辈的后辈，使root节点最后访问。

递归如下：

    protected void doAllSuccessors(Operator node,
                                   Set<Operator> seen,
                                   Collection<Operator> fifo) throws FrontendException {
        if (!seen.contains(node)) {
            // We haven't seen this one before.
            Collection<Operator> succs = Utils.mergeCollection(plan.getSuccessors(node), plan.getSoftLinkSuccessors(node));
            if (succs != null && succs.size() > 0) {
                // Do all our successors before ourself
                for (Operator op : succs) {
                    doAllSuccessors(op, seen, fifo);
                }
            }
            // Now do ourself
            seen.add(node);
            fifo.add(node);
        }
    }

ReverseDependencyOrderWalkerWOSeenChk

ReverseDependencyOrderWalkerWOSeenChk也是逆向的依赖顺序遍历，同ReserveDependencyOrderWalker一样。

 

子Walker是ReserveDependencyOrderWalker

@Override
public PlanWalker spawnChildWalker(OperatorPlan plan) {
return new ReverseDependencyOrderWalker(plan);
}

walk()方法和ReserveDependencyOrderWalker的区别在于，每次遍历的时候不记录一个seen的Set<Operator>集。

全文完 :)