apache calcite 优化器（二）

select 10 + 30, users.name, users.age
from users join jobs on users.id= user.id
where users.age > 30 and jobs.id>10

// org.apache.calcite.plan.hep.HepRelVertex

public class HepRelVertex extends AbstractRelNode {
  //~ Instance fields --------------------------------------------------------

  
  private RelNode currentRel;
}

//org.apache.calcite.plan.hep.HepInstruction

//note: 执行指定 rule 的 Instruction
static class RuleInstance extends HepInstruction {
  
  String ruleDescription;

  
  RelOptRule rule;

  void initialize(boolean clearCache) {
    if (!clearCache) {
      return;
    }

    if (ruleDescription !=null) {
      // Look up anew each run.
      rule=null;
    }
  }

  void execute(HepPlanner planner) {
    planner.executeInstruction(this);
  }
}

HepProgramBuilder builder=new HepProgramBuilder();
builder.addRuleInstance(FilterJoinRule.FilterIntoJoinRule.FILTER_ON_JOIN); //note: 添加 rule
HepPlanner hepPlanner=new HepPlanner(builder.build());
hepPlanner.setRoot(relNode);
relNode =hepPlanner.findBestExp();

// org.apache.calcite.plan.hep.HepPlanner
public void setRoot(RelNode rel) {
  //note: 将 RelNode 转换为 DAG 表示
  root= addRelToGraph(rel);
  //note: 仅仅是在 trace 日志中输出 Graph 信息
  dumpGraph();
}

//org.apache.calcite.plan.hep.HepPlanner
//note: 根据 RelNode 构建一个 Graph
private HepRelVertex addRelToGraph(
    RelNode rel) {
  // Check if a transformation already produced a reference
  // to an existing vertex.
  //note: 检查这个 rel 是否在 graph 中转换了
  if (graph.vertexSet().contains(rel)) {
    return (HepRelVertex) rel;
  }

  // Recursively add children, replacing this rel's inputs
  // with corresponding child vertices.
  //note: 递归地增加子节点，使用子节点相关的 vertices 代替 rel 的 input
  final List<RelNode> inputs= rel.getInputs();
  final List<RelNode> newInputs=new ArrayList<>();
  for (RelNode input1 : inputs) {
    HepRelVertex childVertex =addRelToGraph(input1); //note: 递归进行转换
    newInputs.add(childVertex); //note: 每个 HepRelVertex 只记录其 Input
  }

  if (!Util.equalShallow(inputs, newInputs)) { //note: 不相等的情况下
    RelNode oldRel= rel;
    rel = rel.copy(rel.getTraitSet(), newInputs);
    onCopy(oldRel, rel);
  }
  // Compute digest first time we add to DAG,
  // otherwise can't get equivVertex for common sub-expression
  //note: 计算 relNode 的 digest
  //note: Digest 的意思是：
  //note: A short description of this relational expression's type, inputs, and
  //note: other properties. The string uniquely identifies the node; another node
  //note: is equivalent if and only if it has the same value.
  rel.recomputeDigest();

  // try to find equivalent rel only if DAG is allowed
  //note: 如果允许 DAG 的话，检查是否有一个等价的 HepRelVertex，有的话直接返回
  if (!noDag) {
    // Now, check if an equivalent vertex already exists in graph.
    String digest= rel.getDigest();
    HepRelVertex equivVertex =mapDigestToVertex.get(digest);
    if (equivVertex !=null) { //note: 已经存在
      // Use existing vertex.
      return equivVertex;
    }
  }

  // No equivalence:  create a new vertex to represent this rel.
  //note: 创建一个 vertex 代替 rel
  HepRelVertex newVertex=new HepRelVertex(rel);
  graph.addVertex(newVertex); //note: 记录 Vertex
  updateVertex(newVertex, rel);//note: 更新相关的缓存，比如 mapDigestToVertex map

  for (RelNode input : rel.getInputs()) { //note: 设置 Edge
    graph.addEdge(newVertex, (HepRelVertex) input);//note: 记录与整个 Vertex 先关的 input
  }

  nTransformations++;
  return newVertex;
}

//org.apache.calcite.plan.hep.HepPlanner
// implement RelOptPlanner
//note: 优化器的核心，匹配规则进行优化
public RelNode findBestExp() {
  assert root !=null;

  //note: 运行 HepProgram 算法(按 HepProgram 中的 instructions 进行相应的优化)
  executeProgram(mainProgram);

  // Get rid of everything except what's in the final plan.
  //note: 垃圾收集
  collectGarbage();

  return buildFinalPlan(root); //note: 返回最后的结果，还是以 RelNode 表示
}

//org.apache.calcite.plan.hep.HepPlanner
private void executeProgram(HepProgram program) {
  HepProgram savedProgram =currentProgram; //note: 保留当前的 Program
  currentProgram= program;
  currentProgram.initialize(program ==mainProgram);//note: 如果是在同一个 Program 的话，保留上次 cache
  for (HepInstruction instruction : currentProgram.instructions) {
    instruction.execute(this); //note: 按 Rule 进行优化(会调用 executeInstruction 方法)
    int delta=nTransformations - nTransformationsLastGC;
    if (delta > graphSizeLastGC) {
      // The number of transformations performed since the last
      // garbage collection is greater than the number of vertices in
      // the graph at that time.  That means there should be a
      // reasonable amount of garbage to collect now.  We do it this
      // way to amortize garbage collection cost over multiple
      // instructions, while keeping the highwater memory usage
      // proportional to the graph size.
      //note: 进行转换的次数已经大于 DAG Graph 中的顶点数，这就意味着已经产生大量垃圾需要进行清理
      collectGarbage();
    }
  }
  currentProgram = savedProgram;
}

//org.apache.calcite.plan.hep.HepPlanner
//note: 执行相应的 RuleInstance
void executeInstruction(
    HepInstruction.RuleInstance instruction) {
  if (skippingGroup()) {
    return;
  }
  if (instruction.rule==null) {//note: 如果 rule 为 null，那么就按照 description 查找具体的 rule
    assert instruction.ruleDescription !=null;
    instruction.rule =
        getRuleByDescription(instruction.ruleDescription);
    LOGGER.trace("Looking up rule with description {}, found {}",
        instruction.ruleDescription, instruction.rule);
  }
  //note: 执行相应的 rule
  if (instruction.rule !=null) {
    applyRules(
        Collections.singleton(instruction.rule),
        true);
  }
}

//org.apache.calcite.plan.hep.HepPlanner
//note: 执行 rule（forceConversions 默认 true）
private void applyRules(
    Collection<RelOptRule> rules,
    boolean forceConversions) {
  if (currentProgram.group !=null) {
    assert currentProgram.group.collecting;
    currentProgram.group.ruleSet.addAll(rules);
    return;
  }

  LOGGER.trace("Applying rule set {}", rules);

  //note: 当遍历规则是 ARBITRARY 或 DEPTH_FIRST 时，设置为 false，此时不会从 root 节点开始，否则每次 restart 都从 root 节点开始
  boolean fullRestartAfterTransformation=
      currentProgram.matchOrder != HepMatchOrder.ARBITRARY
      && currentProgram.matchOrder != HepMatchOrder.DEPTH_FIRST;

  int nMatches=0;

  boolean fixedPoint;
  //note: 两种情况会跳出循环，一种是达到 matchLimit 限制，一种是遍历一遍不会再有新的 transform 产生
  do {
    //note: 按照遍历规则获取迭代器
    Iterator<HepRelVertex> iter= getGraphIterator(root);
    fixedPoint =true;
    while (iter.hasNext()) {
      HepRelVertex vertex =iter.next();//note: 遍历每个 HepRelVertex
      for (RelOptRule rule : rules) {//note: 遍历每个 rules
        //note: 进行规制匹配，也是真正进行相关操作的地方
        HepRelVertex newVertex=
            applyRule(rule, vertex, forceConversions);
        if (newVertex==null || newVertex== vertex) {
          continue;
        }
        ++nMatches;
        //note: 超过 MatchLimit 的限制
        if (nMatches >= currentProgram.matchLimit) {
          return;
        }
        if (fullRestartAfterTransformation) {
          //note: 发生 transformation 后，从 root 节点再次开始
          iter= getGraphIterator(root);
        } else {
          // To the extent possible, pick up where we left
          // off; have to create a new iterator because old
          // one was invalidated by transformation.
          //note: 尽可能从上次进行后的节点开始
          iter= getGraphIterator(newVertex);
          if (currentProgram.matchOrder== HepMatchOrder.DEPTH_FIRST) {
            //note: 这样做的原因就是为了防止有些 HepRelVertex 遗漏了 rule 的匹配（每次从 root 开始是最简单的算法），因为可能出现下推
            nMatches=
                depthFirstApply(iter, rules, forceConversions, nMatches);
            if (nMatches >= currentProgram.matchLimit) {
              return;
            }
          }
          // Remember to go around again since we're
          // skipping some stuff.
          //note: 再来一遍，因为前面有跳过一些节点
          fixedPoint=false;
        }
        break;
      }
    }
  } while (!fixedPoint);
}

class RelSet {
   // 记录属于这个 RelSet 的所有 RelNode
  final List<RelNode> rels=new ArrayList<>();
  
  final List<RelNode> parents=new ArrayList<>();
  //note: 具体相同物理属性的子集合（本质上 RelSubset 并不记录 RelNode，也是通过 RelSet 按物理属性过滤得到其 RelNode 子集合，见下面的 RelSubset 部分）
  final List<RelSubset> subsets=new ArrayList<>();

  
  final List<AbstractConverter> abstractConverters=new ArrayList<>();

  
  RelSet equivalentSet;
  RelNode rel;

  
  final Set<CorrelationId> variablesPropagated;

  
  final Set<CorrelationId> variablesUsed;
  final int id;

  
  boolean inMetadataQuery;
}

public class RelSubset extends AbstractRelNode {
  
  RelOptCost bestCost;

  
  final RelSet set;

  
  RelNode best;

  
  boolean boosted;

  //~ Constructors -----------------------------------------------------------
  RelSubset(
      RelOptCluster cluster,
      RelSet set,
      RelTraitSet traits) {
    super(cluster, traits); // 继承自 AbstractRelNode，会记录其相应的 traits 信息
    this.set=set;
    this.boosted=false;
    assert traits.allSimple();
    computeBestCost(cluster.getPlanner()); //note: 计算 best
    recomputeDigest(); //note: 计算 digest
  }
}

//org.apache.calcite.plan.volcano.RuleQueue

public void recompute(RelSubset subset, boolean force) {
  Double previousImportance =subsetImportances.get(subset);
  if (previousImportance==null) { //note: subset 还没有注册的情况下
    if (!force) { //note: 如果不是强制，可以直接先返回
      // Subset has not been registered yet. Don't worry about it.
      return;
    }

    previousImportance = Double.NEGATIVE_INFINITY;
  }

  //note: 计算器 importance 值
  double importance= computeImportance(subset);
  if (previousImportance== importance) {
    return;
  }

  //note: 缓存中更新其 importance
  updateImportance(subset, importance);
}


// 计算一个节点的 importance
double computeImportance(RelSubset subset) {
  double importance;
  if (subset== planner.root) {
    // The root always has importance=1
    //note: root RelSubset 的 importance 为1
    importance=1.0;
  } else {
    final RelMetadataQuery mq = subset.getCluster().getMetadataQuery();

    // The importance of a subset is the max of its importance to its
    // parents
    //note: 计算其相对于 parent 的最大 importance，多个 parent 的情况下，选择一个最大值
    importance=0.0;
    for (RelSubset parent : subset.getParentSubsets(planner)) {
      //note: 计算这个 RelSubset 相对于 parent 的 importance
      final double childImportance=
          computeImportanceOfChild(mq, subset, parent);
      //note: 选择最大的 importance
      importance= Math.max(importance, childImportance);
    }
  }
  LOGGER.trace("Importance of [{}] is {}", subset, importance);
  return importance;
}

//note：根据 cost 计算 child 相对于 parent 的 importance（这是个相对值）
private double computeImportanceOfChild(RelMetadataQuery mq, RelSubset child,
    RelSubset parent) {
  //note: 获取 parent 的 importance
  final double parentImportance= getImportance(parent);
  //note: 获取对应的 cost 信息
  final double childCost= toDouble(planner.getCost(child, mq));
  final double parentCost= toDouble(planner.getCost(parent, mq));
  double alpha=childCost / parentCost;
  if (alpha >=1.0) {
    // child is always less important than parent
    alpha=0.99;
  }
  //note: 根据 cost 比列计算其 importance
  final double importance=parentImportance * alpha;
  LOGGER.trace("Importance of [{}] to its parent [{}] is {} (parent importance={}, child cost={},"
      + " parent cost={})", child, parent, importance, parentImportance, childCost, parentCost);
  return importance;
}

//org.apache.calcite.plan.volcano.VolcanoPlanner
//note: Computes the cost of a RelNode.
public RelOptCost getCost(RelNode rel, RelMetadataQuery mq) {
  assert rel !=null : "pre-condition: rel !=null";
  if (rel instanceof RelSubset) { //note: 如果是 RelSubset，证明是已经计算 cost 的 subset
    return ((RelSubset) rel).bestCost;
  }
  if (rel.getTraitSet().getTrait(ConventionTraitDef.INSTANCE)
      == Convention.NONE) {
    return costFactory.makeInfiniteCost(); //note: 这种情况下也会返回 infinite Cost
  }
  //note: 计算其 cost
  RelOptCost cost= mq.getNonCumulativeCost(rel);
  if (!zeroCost.isLt(cost)) { //note: cost 比0还小的情况
    // cost must be positive, so nudge it
    cost= costFactory.makeTinyCost();
  }
  //note: RelNode 的 cost 会把其 input 全部加上
  for (RelNode input : rel.getInputs()) {
    cost = cost.plus(getCost(input, mq));
  }
  return cost;
}

//org.apache.calcite.plan.volcano.VolcanoRuleMatch

double computeImportance() {
  assert rels[0] !=null; //note: rels[0] 这个 Rule Match 对应的 RelSubset
  RelSubset subset=volcanoPlanner.getSubset(rels[0]);
  double importance=0;
  if (subset !=null) {
    //note: 获取 RelSubset 的 importance
    importance= volcanoPlanner.ruleQueue.getImportance(subset);
  }
  //note: Returns a guess as to which subset the result of this rule will belong to.
  final RelSubset targetSubset= guessSubset();
  if ((targetSubset !=null) && (targetSubset != subset)) {
    // If this rule will generate a member of an equivalence class
    // which is more important, use that importance.
    //note: 获取 targetSubset 的 importance
    final double targetImportance=
        volcanoPlanner.ruleQueue.getImportance(targetSubset);
    if (targetImportance > importance) {
      importance = targetImportance;

      // If the equivalence class is cheaper than the target, bump up
      // the importance of the rule. A converter is an easy way to
      // make the plan cheaper, so we'd hate to miss this opportunity.
      //
      // REVIEW: jhyde, 2007/12/21: This rule seems to make sense, but
      // is disabled until it has been proven.
      //
      // CHECKSTYLE: IGNORE 3
      if ((subset !=null)
          && subset.bestCost.isLt(targetSubset.bestCost)
          && false) { //note: 肯定不会进入
        importance *=
            targetSubset.bestCost.divideBy(subset.bestCost);
        importance =Math.min(importance, 0.99);
      }
    }
  }

  return importance;
}

//1. 初始化 VolcanoPlanner 对象，并添加相应的 Rule
VolcanoPlanner planner=new VolcanoPlanner();
planner.addRelTraitDef(ConventionTraitDef.INSTANCE);
planner.addRelTraitDef(RelDistributionTraitDef.INSTANCE);
// 添加相应的 rule
planner.addRule(FilterJoinRule.FilterIntoJoinRule.FILTER_ON_JOIN);
planner.addRule(ReduceExpressionsRule.PROJECT_INSTANCE);
planner.addRule(PruneEmptyRules.PROJECT_INSTANCE);
// 添加相应的 ConverterRule
planner.addRule(EnumerableRules.ENUMERABLE_MERGE_JOIN_RULE);
planner.addRule(EnumerableRules.ENUMERABLE_SORT_RULE);
planner.addRule(EnumerableRules.ENUMERABLE_VALUES_RULE);
planner.addRule(EnumerableRules.ENUMERABLE_PROJECT_RULE);
planner.addRule(EnumerableRules.ENUMERABLE_FILTER_RULE);
//2. Changes a relational expression to an equivalent one with a different set of traits.
RelTraitSet desiredTraits=
    relNode.getCluster().traitSet().replace(EnumerableConvention.INSTANCE);
relNode = planner.changeTraits(relNode, desiredTraits);
//3. 通过 VolcanoPlanner 的 setRoot 方法注册相应的 RelNode，并进行相应的初始化操作
planner.setRoot(relNode);
//4. 通过动态规划算法找到 cost 最小的 plan
relNode=planner.findBestExp();

EnumerableSort(sort0=[$0], dir0=[ASC])
  EnumerableProject(USER_ID=[$0], USER_NAME=[$1], USER_COMPANY=[$5], USER_AGE=[$2])
    EnumerableMergeJoin(condition=[=($0, $3)], joinType=[inner])
      EnumerableFilter(condition=[>($2, 30)])
        EnumerableTableScan(table=[[USERS]])
      EnumerableFilter(condition=[>($0, 10)])
        EnumerableTableScan(table=[[JOBS]])

//org.apache.calcite.plan.volcano.VolcanoPlanner

public VolcanoPlanner() {
  this(null, null);
}


public VolcanoPlanner(RelOptCostFactory costFactory, //
    Context externalContext) {
  super(costFactory ==null ? VolcanoCost.FACTORY : costFactory, //
      externalContext);
  this.zeroCost=this.costFactory.makeZeroCost();
}

//org.apache.calcite.plan.volcano.VolcanoPlanner
//note: 添加 RelTraitDef
@Override public boolean addRelTraitDef(RelTraitDef relTraitDef) {
  return !traitDefs.contains(relTraitDef) && traitDefs.add(relTraitDef);
}

//org.apache.calcite.plan.volcano.VolcanoPlanner
//note: 添加 rule
public boolean addRule(RelOptRule rule) {
  if (locked) {
    return false;
  }
  if (ruleSet.contains(rule)) {
    // Rule already exists.
    return false;
  }
  final boolean added = ruleSet.add(rule);
  assert added;

  final String ruleName = rule.toString();
  //note: 这里的 ruleNames 允许重复的 key 值，但是这里还是要求 rule description 保持唯一的，与 rule 一一对应
  if (ruleNames.put(ruleName, rule.getClass())) {
    Set<Class> x=ruleNames.get(ruleName);
    if (x.size() > 1) {
      throw new RuntimeException("Rule description '" + ruleName
          + "' is not unique; classes: " + x);
    }
  }

  //note: 注册一个 rule 的 description（保存在 mapDescToRule 中）
  mapRuleDescription(rule);

  // Each of this rule's operands is an 'entry point' for a rule call. Register each operand against all concrete sub-classes that could match it.
  //note: 记录每个 sub-classes 与 operand 的关系（如果能 match 的话，就记录一次）。一个 RelOptRuleOperand 只会有一个 class 与之对应，这里找的是 subclass
  for (RelOptRuleOperand operand : rule.getOperands()) {
    for (Class<? extends RelNode> subClass
        : subClasses(operand.getMatchedClass())) {
      classOperands.put(subClass, operand);
    }
  }

  // If this is a converter rule, check that it operates on one of the
  // kinds of trait we are interested in, and if so, register the rule
  // with the trait.
  //note: 对于 ConverterRule 的操作，如果其 ruleTraitDef 类型包含在我们初始化的 traitDefs 中，
  //note: 就注册这个 converterRule 到 ruleTraitDef 中
  //note: 如果不包含 ruleTraitDef，这个 ConverterRule 在本次优化的过程中是用不到的
  if (rule instanceof ConverterRule) {
    ConverterRule converterRule = (ConverterRule) rule;

    final RelTrait ruleTrait = converterRule.getInTrait();
    final RelTraitDef ruleTraitDef = ruleTrait.getTraitDef();
    if (traitDefs.contains(ruleTraitDef)) { //note: 这里注册好像也没有用到
      ruleTraitDef.registerConverterRule(this, converterRule);
    }
  }

  return true;
}

//org.apache.calcite.plan.volcano.VolcanoPlanner
public void setRoot(RelNode rel) {
  // We're registered all the rules, and therefore RelNode classes,
  // we're interested in, and have not yet started calling metadata providers.
  // So now is a good time to tell the metadata layer what to expect.
  registerMetadataRels();

  //note: 注册相应的 RelNode，会做一系列的初始化操作, RelNode 会有对应的 RelSubset
  this.root=registerImpl(rel, null);
  if (this.originalRoot==null) {
    this.originalRoot= rel;
  }

  // Making a node the root changes its importance.
  //note: 重新计算 root subset 的 importance
  this.ruleQueue.recompute(this.root);
  //Ensures that the subset that is the root relational expression contains converters to all other subsets in its equivalence set.
  ensureRootConverters();
}

//org.apache.calcite.plan.volcano.VolcanoPlanner

private RelSubset registerImpl(
    RelNode rel,
    RelSet set) {
  if (rel instanceof RelSubset) { //note: 如果是 RelSubset 类型，已经注册过了
    return registerSubset(set, (RelSubset) rel); //note: 做相应的 merge
  }

  assert !isRegistered(rel) : "already been registered: " + rel;
  if (rel.getCluster().getPlanner() !=this) { //note: cluster 中 planner 与这里不同
    throw new AssertionError("Relational expression " + rel
        + " belongs to a different planner than is currently being used.");
  }

  // Now is a good time to ensure that the relational expression
  // implements the interface required by its calling convention.
  //note: 确保 relational expression 可以实施其 calling convention 所需的接口
  //note: 获取 RelNode 的 RelTraitSet
  final RelTraitSet traits= rel.getTraitSet();
  //note: 获取其 ConventionTraitDef
  final Convention convention= traits.getTrait(ConventionTraitDef.INSTANCE);
  assert convention !=null;
  if (!convention.getInterface().isInstance(rel)
      && !(rel instanceof Converter)) {
    throw new AssertionError("Relational expression " + rel
        + " has calling-convention " + convention
        + " but does not implement the required interface '"
        + convention.getInterface() + "' of that convention");
  }
  if (traits.size() != traitDefs.size()) {
    throw new AssertionError("Relational expression " + rel
        + " does not have the correct number of traits: " + traits.size()
        + " !=" + traitDefs.size());
  }

  // Ensure that its sub-expressions are registered.
  //note: 其实现在 AbstractRelNode 对应的方法中，实际上调用的还是 ensureRegistered 方法进行注册
  //note: 将 RelNode 的所有 inputs 注册到 planner 中
  //note: 这里会递归调用 registerImpl 注册 relNode 与 RelSet，直到其 inputs 全部注册
  //note: 返回的是一个 RelSubset 类型
  rel=rel.onRegister(this);

  // Record its provenance. (Rule call may be null.)
  //note: 记录 RelNode 的来源
  if (ruleCallStack.isEmpty()) { //note: 不知道来源时
    provenanceMap.put(rel, Provenance.EMPTY);
  } else { //note: 来自 rule 触发的情况
    final VolcanoRuleCall ruleCall= ruleCallStack.peek();
    provenanceMap.put(
        rel,
        new RuleProvenance(
            ruleCall.rule,
            ImmutableList.copyOf(ruleCall.rels),
            ruleCall.id));
  }

  // If it is equivalent to an existing expression, return the set that
  // the equivalent expression belongs to.
  //note: 根据 RelNode 的 digest（摘要，全局唯一）判断其是否已经有对应的 RelSubset，有的话直接放回
  String key= rel.getDigest();
  RelNode equivExp =mapDigestToRel.get(key);
  if (equivExp==null) { //note: 还没注册的情况
    // do nothing
  } else if (equivExp==rel) {//note: 已经有其缓存信息
    return getSubset(rel);
  } else {
    assert RelOptUtil.equal(
        "left", equivExp.getRowType(),
        "right", rel.getRowType(),
        Litmus.THROW);
    RelSet equivSet =getSet(equivExp); //note: 有 RelSubset 但对应的 RelNode 不同时，这里对其 RelSet 做下 merge
    if (equivSet !=null) {
      LOGGER.trace(
          "Register: rel#{} is equivalent to {}", rel.getId(), equivExp.getDescription());
      return registerSubset(set, getSubset(equivExp));
    }
  }

  //note： Converters are in the same set as their children.
  if (rel instanceof Converter) {
    final RelNode input = ((Converter) rel).getInput();
    final RelSet childSet = getSet(input);
    if ((set !=null)
        && (set != childSet)
        && (set.equivalentSet==null)) {
      LOGGER.trace(
          "Register #{} {} (and merge sets, because it is a conversion)",
          rel.getId(), rel.getDigest());
      merge(set, childSet);
      registerCount++;

      // During the mergers, the child set may have changed, and since
      // we're not registered yet, we won't have been informed. So
      // check whether we are now equivalent to an existing
      // expression.
      if (fixUpInputs(rel)) {
        rel.recomputeDigest();
        key = rel.getDigest();
        RelNode equivRel =mapDigestToRel.get(key);
        if ((equivRel !=rel) && (equivRel !=null)) {
          assert RelOptUtil.equal(
              "rel rowtype",
              rel.getRowType(),
              "equivRel rowtype",
              equivRel.getRowType(),
              Litmus.THROW);

          // make sure this bad rel didn't get into the
          // set in any way (fixupInputs will do this but it
          // doesn't know if it should so it does it anyway)
          set.obliterateRelNode(rel);

          // There is already an equivalent expression. Use that
          // one, and forget about this one.
          return getSubset(equivRel);
        }
      }
    } else {
      set= childSet;
    }
  }

  // Place the expression in the appropriate equivalence set.
  //note: 把 expression 放到合适的 等价集 中
  //note: 如果 RelSet 不存在，这里会初始化一个 RelSet
  if (set==null) {
    set=new RelSet(
        nextSetId++,
        Util.minus(
            RelOptUtil.getVariablesSet(rel),
            rel.getVariablesSet()),
        RelOptUtil.getVariablesUsed(rel));
    this.allSets.add(set);
  }

  // Chain to find 'live' equivalent set, just in case several sets are
  // merging at the same time.
  //note: 递归查询，一直找到最开始的 语义相等的集合，防止不同集合同时被 merge
  while (set.equivalentSet !=null) {
    set=set.equivalentSet;
  }

  // Allow each rel to register its own rules.
  registerClass(rel);

  registerCount++;
  //note: 初始时是 0
  final int subsetBeforeCount=set.subsets.size();
  //note: 向等价集中添加相应的 RelNode，并更新其 best 信息
  RelSubset subset=addRelToSet(rel, set);

  //note: 缓存相关信息，返回的 key 之前对应的 value
  final RelNode xx= mapDigestToRel.put(key, rel);
  assert xx ==null || xx== rel : rel.getDigest();

  LOGGER.trace("Register {} in {}", rel.getDescription(), subset.getDescription());

  // This relational expression may have been registered while we
  // recursively registered its children. If this is the case, we're done.
  if (xx !=null) {
    return subset;
  }

  // Create back-links from its children, which makes children more
  // important.
  //note: 如果是 root，初始化其 importance 为 1.0
  if (rel==this.root) {
    ruleQueue.subsetImportances.put(
        subset,
        1.0); // todo: remove
  }
  //note: 将 Rel 的 input 对应的 RelSubset 的 parents 设置为当前的 Rel
  //note: 也就是说，一个 RelNode 的 input 为其对应 RelSubset 的 children 节点
  for (RelNode input : rel.getInputs()) {
    RelSubset childSubset = (RelSubset) input;
    childSubset.set.parents.add(rel);

    // Child subset is more important now a new parent uses it.
    //note: 重新计算 RelSubset 的 importance
    ruleQueue.recompute(childSubset);
  }
  if (rel==this.root) {// TODO: 2019-03-11 这里为什么要删除呢？
    ruleQueue.subsetImportances.remove(subset);
  }

  // Remember abstract converters until they're satisfied
  //note: 如果是 AbstractConverter 示例，添加到 abstractConverters 集合中
  if (rel instanceof AbstractConverter) {
    set.abstractConverters.add((AbstractConverter) rel);
  }

  // If this set has any unsatisfied converters, try to satisfy them.
  //note: check set.abstractConverters
  checkForSatisfiedConverters(set, rel);

  // Make sure this rel's subset importance is updated
  //note: 强制更新（重新计算） subset 的 importance
  ruleQueue.recompute(subset, true);

  //note: 触发所有匹配的 rule，这里是添加到对应的 RuleQueue 中
  // Queue up all rules triggered by this relexp's creation.
  fireRules(rel, true);

  // It's a new subset.
  //note: 如果是一个 new subset，再做一次触发
  if (set.subsets.size() > subsetBeforeCount) {
    fireRules(subset, true);
  }

  return subset;
}

//org.apache.calcite.plan.volcano.VolcanoPlanner

void fireRules(
    RelNode rel,
    boolean deferred) {
  for (RelOptRuleOperand operand : classOperands.get(rel.getClass())) {
    if (operand.matches(rel)) { //note: rule 匹配的情况
      final VolcanoRuleCall ruleCall;
      if (deferred) { //note: 这里默认都是 true，会把 RuleMatch 添加到 queue 中
        ruleCall=new DeferringRuleCall(this, operand);
      } else {
        ruleCall =new VolcanoRuleCall(this, operand);
      }
      ruleCall.match(rel);
    }
  }
}


private static class DeferringRuleCall extends VolcanoRuleCall {
  DeferringRuleCall(
      VolcanoPlanner planner,
      RelOptRuleOperand operand) {
    super(planner, operand);
  }

  
  protected void onMatch() {
    final VolcanoRuleMatch match =new VolcanoRuleMatch(
            volcanoPlanner,
            getOperand0(), //note: 其实就是 operand
            rels,
            nodeInputs);
    volcanoPlanner.ruleQueue.addMatch(match);
  }
}

//org.apache.calcite.plan.volcano.RuleQueue

void addMatch(VolcanoRuleMatch match) {
  final String matchName = match.toString();
  for (PhaseMatchList matchList : matchListMap.values()) {
    if (!matchList.names.add(matchName)) {
      // Identical match has already been added.
      continue;
    }

    String ruleClassName = match.getRule().getClass().getSimpleName();

    Set<String> phaseRuleSet=phaseRuleMapping.get(matchList.phase);
    //note: 如果 phaseRuleSet 不为 ALL_RULES，并且 phaseRuleSet 不包含这个 ruleClassName 时，就跳过(其他三个阶段都属于这个情况)
    //note: 在添加 rule match 时，phaseRuleSet 可以控制哪些 match 可以添加、哪些不能添加
    //note: 这里的话，默认只有处在 OPTIMIZE 阶段的 PhaseMatchList 可以添加相应的 rule match
    if (phaseRuleSet != ALL_RULES) {
      if (!phaseRuleSet.contains(ruleClassName)) {
        continue;
      }
    }

    LOGGER.trace("{} Rule-match queued: {}", matchList.phase.toString(), matchName);

    matchList.list.add(match);

    matchList.matchMap.put(
        planner.getSubset(match.rels[0]), match);
  }
}

//org.apache.calcite.plan.volcano.VolcanoPlanner

public RelNode findBestExp() {
  //note: 确保 root relational expression 的 subset（RelSubset）在它的等价集（RelSet）中包含所有 RelSubset 的 converter
  //note: 来保证 planner 从其他的 subsets 找到的实现方案可以转换为 root，否则可能因为 convention 不同，无法实施
  ensureRootConverters();
  //note: materialized views 相关，这里可以先忽略~
  registerMaterializations();
  int cumulativeTicks=0; //note: 四个阶段通用的变量
  //note: 不同的阶段，总共四个阶段，实际上只有 OPTIMIZE 这个阶段有效，因为其他阶段不会有 RuleMatch
  for (VolcanoPlannerPhase phase : VolcanoPlannerPhase.values()) {
    //note: 在不同的阶段，初始化 RelSubSets 相应的 importance
    //note: root 节点往下子节点的 importance 都会被初始化
    setInitialImportance();

    //note: 默认是 VolcanoCost
    RelOptCost targetCost= costFactory.makeHugeCost();
    int tick=0;
    int firstFiniteTick=-1;
    int splitCount=0;
    int giveUpTick= Integer.MAX_VALUE;

    while (true) {
      ++tick;
      ++cumulativeTicks;
      //note: 第一次运行是 false，两个不是一个对象，一个是 costFactory.makeHugeCost， 一个是 costFactory.makeInfiniteCost
      //note: 如果低于目标 cost，这里再重新设置一个新目标、新的 giveUpTick
      if (root.bestCost.isLe(targetCost)) {
        //note: 本阶段第一次运行，目的是为了调用 clearImportanceBoost 方法，清除相应的 importance 信息
        if (firstFiniteTick < 0) {
          firstFiniteTick = cumulativeTicks;

          //note: 对于那些手动提高 importance 的 RelSubset 进行重新计算
          clearImportanceBoost();
        }
        if (ambitious) {
          // Choose a slightly more ambitious target cost, and
          // try again. If it took us 1000 iterations to find our
          // first finite plan, give ourselves another 100
          // iterations to reduce the cost by 10%.
          //note: 设置 target 为当前 best cost 的 0.9，调整相应的目标，再进行优化
          targetCost=root.bestCost.multiplyBy(0.9);
          ++splitCount;
          if (impatient) {
            if (firstFiniteTick < 10) {
              // It's possible pre-processing can create
              // an implementable plan -- give us some time
              // to actually optimize it.
              //note: 有可能在 pre-processing 阶段就实现一个 implementable plan，所以先设置一个值，后面再去优化
              giveUpTick=cumulativeTicks + 25;
            } else {
              giveUpTick =
                  cumulativeTicks
                      + Math.max(firstFiniteTick / 10, 25);
            }
          }
        } else {
          break;
        }
      //note: 最近没有任何进步（超过 giveUpTick 限制，还没达到目标值），直接采用当前的 best plan
      } else if (cumulativeTicks > giveUpTick) {
        // We haven't made progress recently. Take the current best.
        break;
      } else if (root.bestCost.isInfinite() && ((tick % 10)==0)) {
        injectImportanceBoost();
      }

      LOGGER.debug("PLANNER={}; TICK={}/{}; PHASE={}; COST={}",
          this, cumulativeTicks, tick, phase.toString(), root.bestCost);

      VolcanoRuleMatch match = ruleQueue.popMatch(phase);
      //note: 如果没有规则，会直接退出当前的阶段
      if (match==null) {
        break;
      }

      assert match.getRule().matches(match);
      //note: 做相应的规则匹配
      match.onMatch();

      // The root may have been merged with another
      // subset. Find the new root subset.
      root= canonize(root);
    }

    //note: 当期阶段完成，移除 ruleQueue 中记录的 rule-match list
    ruleQueue.phaseCompleted(phase);
  }
  if (LOGGER.isTraceEnabled()) {
    StringWriter sw =new StringWriter();
    final PrintWriter pw =new PrintWriter(sw);
    dump(pw);
    pw.flush();
    LOGGER.trace(sw.toString());
  }
  //note: 根据 plan 构建其 RelNode 树
  RelNode cheapest=root.buildCheapestPlan(this);
  if (LOGGER.isDebugEnabled()) {
    LOGGER.debug(
        "Cheapest plan:
{}", RelOptUtil.toString(cheapest, SqlExplainLevel.ALL_ATTRIBUTES));

    LOGGER.debug("Provenance:
{}", provenance(cheapest));
  }
  return cheapest;
}

/org.apache.calcite.plan.volcano.RuleQueue

VolcanoRuleMatch popMatch(VolcanoPlannerPhase phase) {
  dump();

  //note: 选择当前阶段对应的 PhaseMatchList
  PhaseMatchList phaseMatchList=matchListMap.get(phase);
  if (phaseMatchList==null) {
    throw new AssertionError("Used match list for phase " + phase
        + " after phase complete");
  }

  final List<VolcanoRuleMatch> matchList= phaseMatchList.list;
  VolcanoRuleMatch match;
  for (;;) {
    //note: 按照前面的逻辑只有在 OPTIMIZE 阶段，PhaseMatchList 才不为空，其他阶段都是空
    // 参考 addMatch 方法
    if (matchList.isEmpty()) {
      return null;
    }
    if (LOGGER.isTraceEnabled()) {
      matchList.sort(MATCH_COMPARATOR);
      match =matchList.remove(0);

      StringBuilder b =new StringBuilder();
      b.append("Sorted rule queue:");
      for (VolcanoRuleMatch match2 : matchList) {
        final double importance= match2.computeImportance();
        b.append("
");
        b.append(match2);
        b.append(" importance ");
        b.append(importance);
      }

      LOGGER.trace(b.toString());
    } else { //note: 直接遍历找到 importance 最大的 match（上面先做排序，是为了输出日志）
      // If we're not tracing, it's not worth the effort of sorting the
      // list to find the minimum.
      match=null;
      int bestPos=-1;
      int i=-1;
      for (VolcanoRuleMatch match2 : matchList) {
        ++i;
        if (match==null
            || MATCH_COMPARATOR.compare(match2, match) < 0) {
          bestPos = i;
          match = match2;
        }
      }
      match = matchList.remove(bestPos);
    }

    if (skipMatch(match)) {
      LOGGER.debug("Skip match: {}", match);
    } else {
      break;
    }
  }

  // A rule match's digest is composed of the operand RelNodes' digests,
  // which may have changed if sets have merged since the rule match was
  // enqueued.
  //note: 重新计算一下这个 RuleMatch 的 digest
  match.recomputeDigest();

  //note: 从 phaseMatchList 移除这个 RuleMatch
  phaseMatchList.matchMap.remove(
      planner.getSubset(match.rels[0]), match);

  LOGGER.debug("Pop match: {}", match);
  return match;
}

protected VolcanoPlannerPhaseRuleMappingInitializer
    getPhaseRuleMappingInitializer() {
  return phaseRuleMap -> {
    // Disable all phases except OPTIMIZE by adding one useless rule name.
    //note: 通过添加一个无用的 rule name 来 disable 优化器的其他三个阶段
    phaseRuleMap.get(VolcanoPlannerPhase.PRE_PROCESS_MDR).add("xxx");
    phaseRuleMap.get(VolcanoPlannerPhase.PRE_PROCESS).add("xxx");
    phaseRuleMap.get(VolcanoPlannerPhase.CLEANUP).add("xxx");
  };
}

for (VolcanoPlannerPhase phase : VolcanoPlannerPhase.values()) {
  // empty phases get converted to "all rules"
  //note: 如果阶段对应的 rule set 为空，那么就给这个阶段对应的 rule set 添加一个 【ALL_RULES】
  //也就是只有 OPTIMIZE 这个阶段对应的会添加 ALL_RULES
  if (phaseRuleMapping.get(phase).isEmpty()) {
    phaseRuleMapping.put(phase, ALL_RULES);
  }
}