Calcite CBO ③ - VolcanoPlanner 之

TopDownRuleDriver 是 cascades 论文的标准实现，我们以下面的 case 来跟踪代码：

LogicalAggregate(group=[{0}], groups=[[{0}, {}]], C=[COUNT()])
  LogicalProject(DEPTNO=[$2])
    LogicalValues(tuples=[[{ 100, 'Fred', 10, null, null, 40, 25, true, false, 1996-08-03 }, { 110, 'Eric', 20, 'M', 'San Francisco', 3, 80, null, false, 2001-01-01 }, { 110, 'John', 40, 'M', 'Vancouver', 2, null, false, true, 2002-05-03 }, { 120, 'Wilma', 20, 'F', null, 1, 5, null, true, 2005-09-07 }, { 130, 'Alice', 40, 'F', 'Vancouver', 2, null, false, true, 2007-01-01 }]])

TopDownRuleDriver 主要有以下成员：

• TopDownRuleQueue ruleQueue：用来管理 TopDownRuleDriver 用到的 RuleMatchs
• Stack<Task> tasks：用来管理 Tasks

一、概述

总结来说，TopDownRuleDriver#drive 的执行逻辑如下：

@Override public void drive() {
  TaskDescriptor description = new TaskDescriptor();

  // Starting from the root's OptimizeGroup task.
  tasks.push(
      new OptimizeGroup(
          requireNonNull(planner.root, "planner.root"),
          planner.infCost));

  // Iterates until the root is fully optimized.
  while (!tasks.isEmpty()) {
    Task task = tasks.pop();
    description.log(task);
    task.perform();
  }
}

整个优化过程由上面的循环驱动：不断从栈顶取出 Task 执行，Task 执行中又会产生新的 Task，重复这个过程直到栈为空（或者某个 Task 执行抛异常）。

二、各类 Task 串联

可以看到，一切优化都是从 OptimizeGroup(root) 的 task 开始的。

2.1、Round1

2.2、Round2

以 VolcanoPlanner 的 RelSubset root 成员为起点，自下而上，不停的探索 inputs，根据不同情况，执行创建 Task、Task 入栈、Task 出栈、Task执行等操作，直到 Stack<Task> tasks 为空或过程中抛异常为止。以 Stack 模拟了递归的实现。

具体来说，基于 RelSubset root 创建 OptimizeGroup 类型的 Task 并入栈，从 tasks 出栈，得到 Task 并执行 perform()。每个 Task 都有绑定的 RelNode，我们下述统称为 boundRel

• 如果 Task 是 OptimizeGroup 类型：对于该 Task 的 boundRel 的 set.rels

  • 先对于其中的 logical 类型的 RelNode 创建 OptimizeMExpr 类型 Task 并入栈
  • 再对其中的 physical 类型的 RelNode 的 inputs 创建 OptimizeInput1/OptimizeInputs 类型 Task 并入栈

• 如果 Task 是 OptimizeMExpr 类型

  • 创建对于 boundRel 的ApplyRules 类型的 Task 并入栈
  • 对于 boundRel.getInputs() 中的每个 input，创建一个 ExploreInput 类型的 Task 并入栈

• 如果 Task 是 OptimizeInput1 类型：为只有一个输入的 physical node 优化其 input 的 Task

  • 创建 CheckInput 类型 Task 并入栈，主体是 physical node
  • 创建 OptimizeGroup 类型 Task 并入栈，主体是 physical node 的 input0

• 如果 Task 是 OptimizeInputs类型：用于优化 physical node 的 inputs。此 Task 计算的适当上限（RelOptCost upperBound）并调用 OptimizeGroup Task。 当 input 的 upperBound 小于 input 的下限时，Group 剪枝主要发生在这里

• 如果 Task 是 CheckInput 类型：

  • 如果 input 发生变更（可能由其他规则的action导致），对于 input 创建 OptimizeGroup Task 重新 explore
  • 如果 input 的 inputs 都已经 explore 过，函数返回
  • 如果是 OptimizeInputs Task 创建的 CheckInput，会计算当前的 lower cost 并向上更新 OptimizeInputs 对应的 lowerBoundSum ，即 OptimizeInputs 对应的多个 inputs 的 cost 之和

• 如果 Task 是 ApplyRules 类型：从 TopDownRuleDriver 的 TopDownRuleQueue ruleQueue 成员 pop 出适用于 boundRel 的 ruleMatchs 并为每一个 ruleMatch 创建一个 ApplyRule （主体是 boundRel 对应的 RelSubset 及 ruleMatch）类型的 Task 并入栈

• 如果 Task 是 ExploreInput 类型：对于 boundRel.set.rels 中每个 logical node，创建一个 OptimizeMExpr 类型的 Task 用于进一步 explore inputs

• 如果 Task 是 ApplyRule 类型：说白了就是将一个 ruleMatch apply 到 boundRel，生成新的 plan，新的 plan 又会进入到某个 RelSubset 中，进而又会进一步触发后续的优化任务（这部分位于 onProduce）

  • 如果产生的是 logical plan 则生成 OptimizeMExpr
  • 如果产生的是 physical plan 则生成 OptimizeInputs

• 如果 Task 是 ExploreInput 类型：和 OptimizeGroup 对等的作用，也是优化 group，差别是这里只处理 group.set.rels 中的逻辑算子。还有一个很容易被忽略的细节，这里 explore 的值：

  • 如果是通过 OptimizeGroup 生成的 OptimizeMExpr，explore=false
  • 而通过ExploreInput生成的 OptimizeMExpr，explore=true

  explore 会影响 applyRules 的时候的 rule 的筛选，为 true，只会 apply TransformationRule

上述这些 task 共同构成了 top-down 优化的递归过程。上图是各个 task 之间的调用关系，蓝色回边意味着递归进入下一层节点

三、关键要点

3.1、停止条件

• tasks 为空
• Task 执行抛异常

3.2、复用

以 OptimizeInputs、OptimizeGroup 的 perform 方法为例：

    @Override public void perform() {
      RelOptCost winner = group.getWinnerCost();
      if (winner != null) {
        return;
      }

当 RelSubset 已经被 fully optimized 后（也就是 RelSubset 中的每个 RelNode 都被优化过或剪枝了），Task 直接返回无需再执行

四、剪枝

4.1、通过 lower bound 与 upper bound 剪枝

在 TopDown 优化中，对于一个 RelSubset，会有一个 RelNode 会率先计算出 cost，此时会暂时将该 cost 设置为 RelSubset.bestCost，同时赋值给 RelSubset.upperBound，当要对同 RelSubset 的其他 RelNode 及其 inputs 进行向下搜索时。

有没有可能在搜索之前，就可以判断其代价已经过大，无法产生比当前 bestCost 更好的 cost，从而实现 purning 呢？

比如，此时 best（RelNode）如下，bestCost/upperBound 为 {84.80000000000001 rows, 1702.8593150429992 cpu, 0.0 io}

EnumerableMergeJoin(subset=[rel#11141:RelSubset#6.ENUMERABLE.[2]], condition=[=($0, $2)], joinType=[inner])
  EnumerableProject(subset=[rel#11087:RelSubset#3.ENUMERABLE.[0]], DEPTNO=[$0])
    EnumerableTableScan(subset=[rel#11091:RelSubset#2.ENUMERABLE.[0]], table=[[scott, DEPT]])
  EnumerableProject(subset=[rel#11110:RelSubset#1.ENUMERABLE.[1]], EMPNO=[$0], DEPTNO=[$7])
    EnumerableSort(subset=[rel#11113:RelSubset#0.ENUMERABLE.[7]], sort0=[$7], dir0=[ASC])
      EnumerableTableScan(subset=[rel#11100:RelSubset#0.ENUMERABLE.[0]], table=[[scott, EMP]])

接下来，要去 check 同一个 RelSubset 下的另一个 RelNode（记为 anotherRel）及其包含的各级 children 来看是否会有更加 cheap 的 cost。

EnumerableHashJoin(condition=[=($1, $2)], joinType=[inner])
  EnumerableProject(subset=[rel#11110:RelSubset#1.ENUMERABLE.[1]], EMPNO=[$0], DEPTNO=[$7])
    EnumerableSort(subset=[rel#11113:RelSubset#0.ENUMERABLE.[7]], sort0=[$7], dir0=[ASC])
      EnumerableTableScan(subset=[rel#11100:RelSubset#0.ENUMERABLE.[0]], table=[[scott, EMP]])
  EnumerableProject(subset=[rel#11145:RelSubset#3.ENUMERABLE.[]], DEPTNO=[$0])
    EnumerableTableScan(subset=[rel#11091:RelSubset#2.ENUMERABLE.[0]], table=[[scott, DEPT]])

这里会计算两个值：

• RelOptCost upperForInput

  • upperForInput = planner.upperBoundForInputs(anotherRel, upperBound)
  • 得到 {35.453197385386396 rows, 1702.8593150429992 cpu, 0.0 io}

• RelOptCost lowerBoundSum：

  • 即 anotherRel 的所有 planner.getLowerBound(input) 的 cost 之和
  • 得到 {42.0 rows, 1668.6593150429992 cpu, 0.0 io}

当 upperForInput < lowerBoundSum 时，无需再向下搜索 anotherRel 的各级 children 了，达到了一个剪枝的目的。目前在 Calcite 中，cost 的比较主要看 rowCount，不怎么看 cpu 和 io。