Calcite框架分享-stage1

本质上是 unparse 的过程。在解析完成后, Calcite 会生成 SqlNode, 如上述例子所展示, 本质上就是一颗树.

从而只需要遍历这个树, 基于每个节点自身的规则, 以及对应的方言需要, 生成对应的片段即可。

以 SqlSelectOperator 的 sql 生成(删除了一些代码)为例

@Override public void unparse(
    SqlWriter writer,
    SqlCall call,
    int leftPrec,
    int rightPrec) {
  SqlSelect select = (SqlSelect) call;
  final SqlWriter.Frame selectFrame =
      writer.startList(SqlWriter.FrameTypeEnum.SELECT);
  writer.sep("SELECT");

  final SqlNodeList selectClause = select.selectList;
  writer.list(SqlWriter.FrameTypeEnum.SELECT_LIST, SqlWriter.COMMA,
      selectClause);

  if (select.from != null) {
    // Calcite SQL requires FROM but MySQL does not.
    writer.sep("FROM");

    // for FROM clause, use precedence just below join operator to make
    // sure that an un-joined nested select will be properly
    // parenthesized
    final SqlWriter.Frame fromFrame =
        writer.startList(SqlWriter.FrameTypeEnum.FROM_LIST);
    select.from.unparse(
        writer,
        SqlJoin.OPERATOR.getLeftPrec() - 1,
        SqlJoin.OPERATOR.getRightPrec() - 1);
    writer.endList(fromFrame);
  }

  SqlNode where = select.where;
  if (where != null) {
    writer.sep("WHERE");

    if (!writer.isAlwaysUseParentheses()) {
      SqlNode node = where;

      // decide whether to split on ORs or ANDs
      SqlBinaryOperator whereSep = SqlStdOperatorTable.AND;
      if ((node instanceof SqlCall)
          && node.getKind() == SqlKind.OR) {
        whereSep = SqlStdOperatorTable.OR;
      }

      // unroll whereClause
      final List<SqlNode> list = new ArrayList<>(0);
      while (node.getKind() == whereSep.kind) {
        assert node instanceof SqlCall;
        final SqlCall call1 = (SqlCall) node;
        list.add(0, call1.operand(1));
        node = call1.operand(0);
      }
      list.add(0, node);

      // unparse in a WHERE_LIST frame
      writer.list(SqlWriter.FrameTypeEnum.WHERE_LIST, whereSep,
          new SqlNodeList(list, where.getParserPosition()));
    } else {
      where.unparse(writer, 0, 0);
    }
  }
  writer.endList(selectFrame);
}

和 unparse 的过程非常类似。同样看一段 SqlSelect -> RelNode 的过程

/**
 * Implementation of {@link #convertSelect(SqlSelect, boolean)};
 * derived class may override.
 */
protected void convertSelectImpl(
    final Blackboard bb,
    SqlSelect select) {
  convertFrom(
      bb,
      select.getFrom());
  convertWhere(
      bb,
      select.getWhere());

  final List<SqlNode> orderExprList = new ArrayList<>();
  final List<RelFieldCollation> collationList = new ArrayList<>();
  gatherOrderExprs(
      bb,
      select,
      select.getOrderList(),
      orderExprList,
      collationList);
  final RelCollation collation =
      cluster.traitSet().canonize(RelCollations.of(collationList));

  convertOrder(
      select, bb, collation, orderExprList, select.getOffset(),
      select.getFetch());

  bb.setRoot(bb.root(), true);
}

protected void convertFrom(
                           Blackboard bb,
                           @Nullable SqlNode from,
                           @Nullable List<String> fieldNames) {
    switch (from.getKind()) {
    case TABLE_REF:
        call = (SqlCall) from;
        convertIdentifier(bb, call.operand(0), null, call.operand(1));
        return;
    }
}

private void convertIdentifier(Blackboard bb, SqlIdentifier id,
    @Nullable SqlNodeList extendedColumns, @Nullable SqlNodeList tableHints) {
  final String datasetName =
      datasetStack.isEmpty() ? null : datasetStack.peek();
  final boolean[] usedDataset = {false};
  RelOptTable table =
      SqlValidatorUtil.getRelOptTable(fromNamespace, catalogReader,
          datasetName, usedDataset);
  assert table != null : "getRelOptTable returned null for " + fromNamespace;
  final RelNode tableRel;
  // Review Danny 2020-01-13: hacky to construct a new table scan
  // in order to apply the hint strategies.
  final List<RelHint> hints = hintStrategies.apply(
      SqlUtil.getRelHint(hintStrategies, tableHints),
      LogicalTableScan.create(cluster, table, ImmutableList.of()));
  tableRel = toRel(table, hints);
  bb.setRoot(tableRel, true);
  if (usedDataset[0]) {
    bb.setDataset(datasetName);
  }
}

  /**
   * Converts a WHERE clause.
   *
   * @param bb    Blackboard
   * @param where WHERE clause, may be null
   */
  private void convertWhere(
      final Blackboard bb,
      final @Nullable SqlNode where) {
    if (where == null) {
      return;
    }
    SqlNode newWhere = pushDownNotForIn(bb.scope(), where);
    replaceSubQueries(bb, newWhere, RelOptUtil.Logic.UNKNOWN_AS_FALSE);
    final RexNode convertedWhere = bb.convertExpression(newWhere);
    final RexNode convertedWhere2 =
        RexUtil.removeNullabilityCast(typeFactory, convertedWhere);

    // only allocate filter if the condition is not TRUE
    if (convertedWhere2.isAlwaysTrue()) {
      return;
    }

    final RelFactories.FilterFactory filterFactory =
        RelFactories.DEFAULT_FILTER_FACTORY;
    final RelNode filter =
        filterFactory.createFilter(bb.root(), convertedWhere2, ImmutableSet.of());
    final RelNode r = filter;

    bb.setRoot(r, false);
  }

这一部分为 calcite 中最为核心的逻辑，共可以分成以下几部分探讨:

1. 如何将一个关系表达式生成为另一个等价的关系表达式
2. 如何尽可能完备的生成所有的等价表达式
   1. 一个关系表达式为一颗树，其中任一子树，均可能生成等价的表达式
   2. 生成后，可能存在新的子树，也可以生成新的等价表达式
3. 如何选择最优的表达式

这三个问题在另一个部分中详述。

具体代码和 unparse, sql -> rel 等，结构上如出一辙。可参考: RelToSqlConverter 中的 visit(Project e), visit(Aggregate e) 等等。

同样看一个例子，有些直观的效果上的理解

1. sql

   select idc, domain FROM metricsdb.domain_idc_quality
   

2. sqlNode

   (SqlSelect
    (SqlSelectList (SqlIdentifier "idc")
                   (SqlIdentifier "domain"))
    (SqlFrom (SqlIdentifier
              (SqlNames "metricsdb" "domain_idc_quality"))))
   

3. relNode

   (LogicalProject (exprs $3 $4)
                   (LogicalTableScan (table "metricsdb.domain_idc_quality")))
   

这部分的逻辑是: 先粗粒度匹配, 匹配上了之后执行具体的 onMatch() .

在 onMatch 中，进行各种校验，生成的逻辑，生成等价表达式之后，调用 transformTo 将等价表达式给 planner, 进行后续的逻辑。

常见的粗粒度匹配为: 支持什么 Operand (亦即 RelNode), 仅从类型和树的结构进行描述。比如

1. AggregateJoinTransposeRule, 需要这颗子树(亦即当前的 RelNode)的顶点是 Aggregate , 这个 Aggregate 有一个 Input, 且 Input 是 Join

   /**
    * Extended instance that can push down aggregate functions.
    */
   Config EXTENDED = EMPTY.as(Config.class)
           .withOperandFor(LogicalAggregate.class, LogicalJoin.class, true);
   /**
    * Defines an operand tree for the given classes, and also sets
    * {@link #isAllowFunctions()}.
    */
   default Config withOperandFor(Class<? extends Aggregate> aggregateClass,
                                 Class<? extends Join> joinClass, boolean allowFunctions) {
       return withAllowFunctions(allowFunctions)
               .withOperandSupplier(b0 ->
                       b0.operand(aggregateClass)
                               .predicate(agg -> isAggregateSupported(agg, allowFunctions))
                               .oneInput(b1 ->
                                       b1.operand(joinClass).anyInputs()))
               .as(Config.class);
   }
   

2. FilterProjectTransposeRule, 需要顶点是 Filter 且 condition 中不含有 CorrelationVariable ¹, 只有一个 Input 是 Project

   Config DEFAULT = EMPTY.as(Config.class)
       .withOperandFor(Filter.class,
           f -> !RexUtil.containsCorrelation(f.getCondition()),
           Project.class, p -> true)
       .withCopyFilter(true)
       .withCopyProject(true);
   
   /** Defines an operand tree for the given 2 classes. */
   default Config withOperandFor(Class<? extends Filter> filterClass,
       Predicate<Filter> filterPredicate,
       Class<? extends Project> projectClass,
       Predicate<Project> projectPredicate) {
     return withOperandSupplier(b0 ->
         b0.operand(filterClass).predicate(filterPredicate).oneInput(b1 ->
             b1.operand(projectClass).predicate(projectPredicate).anyInputs()))
         .as(Config.class);
   }
   

具体的等价表达式的生成逻辑就各有不同了。举两个例子

先看一个例子:

• 树

  LogicalAggregate(group=[{0, 1, 2, 3, 4}], NASA2usage=[SUM($5)], NASA2videousage=[SUM($6)], NASA2audiousage=[SUM($7)], alltotalusage=[SUM($8)], allaudioUsage=[SUM($9)], allvideoUsage=[SUM($10)])
    LogicalProject(date=[$0], vid=[$1], projectName=[$14], companyid=[$15], companyname=[$16], totalusage=[$5], videousage=[$6], audio_usage=[$7], audio_video=[$10], audio=[$11], video=[$12])
      LogicalJoin(condition=[=($1, $13)], joinType=[left])
        LogicalJoin(condition=[AND(=($1, $9), =($0, $8))], joinType=[left])
          LogicalAggregate(group=[{0, 1, 2, 3, 4}], totalusage=[SUM($5)], videousage=[SUM($6)], audio_usage=[SUM($7)])
            LogicalProject(date=[$0], vid=[$5], domain=[$8], os=[$11], ver=[$12], audio_video=[$20], video=[$26], audio=[$23])
              LogicalFilter(condition=[AND(=($1, '(product_line,product_type,vid,domain,os,ver)'), OR(=($12, '3.0.0.11'), =($12, '3,0,0,16'), =($12, '3.0.0.17'), =($12, '3.0.0.19'), =($12, '3.0.0.18'), =($12, '3.0.20.20'), =($12, '3.0.20.21')), >=($0, '2020-08-01'))])
                LogicalTableScan(table=[[metricsdb, agora_vid_usage_cube_di]])
          LogicalAggregate(group=[{0, 1}], audio_video=[SUM($2)], audio=[SUM($3)], video=[SUM($4)])
            LogicalProject(date=[$0], vid=[$5], audio_video=[$20], audio=[$23], video=[$26])
              LogicalFilter(condition=[AND(>=($0, '2020-08-01'), =($1, '(product_line,product_type,vid,domain,os,ver)'))])
                LogicalTableScan(table=[[metricsdb, agora_vid_usage_cube_di]])
        LogicalProject(vid=[$0], projectName=[$1], companyid=[$2], companyname=[$3])
          LogicalTableScan(table=[[metricsdb, new_vendor_dimension]])
  

• 图

  Breadth-first from root:  {
      rel#57:HepRelVertex(rel#56:LogicalAggregate.(input=HepRelVertex#55,group={0, 1, 2, 3, 4},NASA2usage=SUM($5),NASA2videousage=SUM($6),NASA2audiousage=SUM($7),alltotalusage=SUM($8),allaudioUsage=SUM($9),allvideoUsage=SUM($10))) = rel#56:LogicalAggregate.(input=HepRelVertex#55,group={0, 1, 2, 3, 4},NASA2usage=SUM($5),NASA2videousage=SUM($6),NASA2audiousage=SUM($7),alltotalusage=SUM($8),allaudioUsage=SUM($9),allvideoUsage=SUM($10)), rowcount=1.585, cumulative cost=457.1676257395745
      rel#55:HepRelVertex(rel#54:LogicalProject.(input=HepRelVertex#53,inputs=0..1,exprs=[$14, $15, $16, $5, $6, $7, $10, $11, $12])) = rel#54:LogicalProject.(input=HepRelVertex#53,inputs=0..1,exprs=[$14, $15, $16, $5, $6, $7, $10, $11, $12]), rowcount=15.85, cumulative cost=454.27500028610234
      rel#53:HepRelVertex(rel#52:LogicalJoin.(left=HepRelVertex#48,right=HepRelVertex#51,condition==($1, $13),joinType=left)) = rel#52:LogicalJoin.(left=HepRelVertex#48,right=HepRelVertex#51,condition==($1, $13),joinType=left), rowcount=15.85, cumulative cost=438.4250002861023
      rel#48:HepRelVertex(rel#47:LogicalJoin.(left=HepRelVertex#40,right=HepRelVertex#46,condition=AND(=($1, $9), =($0, $8)),joinType=left)) = rel#47:LogicalJoin.(left=HepRelVertex#40,right=HepRelVertex#46,condition=AND(=($1, $9), =($0, $8)),joinType=left), rowcount=1.0, cumulative cost=222.5750002861023
      rel#51:HepRelVertex(rel#50:LogicalProject.(input=HepRelVertex#49,inputs=0..3)) = rel#50:LogicalProject.(input=HepRelVertex#49,inputs=0..3), rowcount=100.0, cumulative cost=200.0
      rel#40:HepRelVertex(rel#39:LogicalAggregate.(input=HepRelVertex#38,group={0, 1, 2, 3, 4},totalusage=SUM($5),videousage=SUM($6),audio_usage=SUM($7))) = rel#39:LogicalAggregate.(input=HepRelVertex#38,group={0, 1, 2, 3, 4},totalusage=SUM($5),videousage=SUM($6),audio_usage=SUM($7)), rowcount=1.0, cumulative cost=105.16250014305115
      rel#46:HepRelVertex(rel#45:LogicalAggregate.(input=HepRelVertex#44,group={0, 1},audio_video=SUM($2),audio=SUM($3),video=SUM($4))) = rel#45:LogicalAggregate.(input=HepRelVertex#44,group={0, 1},audio_video=SUM($2),audio=SUM($3),video=SUM($4)), rowcount=1.0, cumulative cost=116.41250014305115
      rel#49:HepRelVertex(rel#28:LogicalTableScan.(table=[metricsdb, new_vendor_dimension])) = rel#28:LogicalTableScan.(table=[metricsdb, new_vendor_dimension]), rowcount=100.0, cumulative cost=100.0
      rel#38:HepRelVertex(rel#37:LogicalProject.(input=HepRelVertex#36,inputs=0,exprs=[$5, $8, $11, $12, $20, $26, $23])) = rel#37:LogicalProject.(input=HepRelVertex#36,inputs=0,exprs=[$5, $8, $11, $12, $20, $26, $23]), rowcount=1.875, cumulative cost=103.75
      rel#44:HepRelVertex(rel#43:LogicalProject.(input=HepRelVertex#42,inputs=0,exprs=[$5, $20, $23, $26])) = rel#43:LogicalProject.(input=HepRelVertex#42,inputs=0,exprs=[$5, $20, $23, $26]), rowcount=7.5, cumulative cost=115.0
      rel#36:HepRelVertex(rel#35:LogicalFilter.(input=HepRelVertex#34,condition=AND(=($1, '(product_line,product_type,vid,domain,os,ver)'), OR(=($12, '3.0.0.11'), =($12, '3,0,0,16'), =($12, '3.0.0.17'), =($12, '3.0.0.19'), =($12, '3.0.0.18'), =($12, '3.0.20.20'), =($12, '3.0.20.21')), >=($0, '2020-08-01')))) = rel#35:LogicalFilter.(input=HepRelVertex#34,condition=AND(=($1, '(product_line,product_type,vid,domain,os,ver)'), OR(=($12, '3.0.0.11'), =($12, '3,0,0,16'), =($12, '3.0.0.17'), =($12, '3.0.0.19'), =($12, '3.0.0.18'), =($12, '3.0.20.20'), =($12, '3.0.20.21')), >=($0, '2020-08-01'))), rowcount=1.875, cumulative cost=101.875
      rel#42:HepRelVertex(rel#41:LogicalFilter.(input=HepRelVertex#34,condition=AND(>=($0, '2020-08-01'), =($1, '(product_line,product_type,vid,domain,os,ver)')))) = rel#41:LogicalFilter.(input=HepRelVertex#34,condition=AND(>=($0, '2020-08-01'), =($1, '(product_line,product_type,vid,domain,os,ver)'))), rowcount=7.5, cumulative cost=107.5
      rel#34:HepRelVertex(rel#16:LogicalTableScan.(table=[metricsdb, agora_vid_usage_cube_di])) = rel#16:LogicalTableScan.(table=[metricsdb, agora_vid_usage_cube_di]), rowcount=100.0, cumulative cost=100.0
  }
  

优化的基础就是: 将树生成图，而后基于某种遍历方式，将每个顶点(代表一颗子树)都尝试生成对应的等价表达式, 并加入图中。 ² 最终基于图中的信息，获得最终的执行计划。

在 calcite 中, planner 负责规则的协调/迭代, 树的等价生成，树的选择等。

默认实现中, 共提供了两种 planner, 分别为 HepPlanner 与 VolcanoPlanner.

常规意义上，一般认为 HepPlanner 是 Rule Based, 而 VolcanoPlanner 是 Cost Based.

• Hep 会在图中进行替换，使用新表达式替换已有，而后基于新图，继续生成，直至图不变为止。
• Volcanno 会保留所有的等价集合, 最终选择一个 cost 最低的方案。

1. Hep 的生成规则相对简单, 遍历所有节点, 与所有规则
2. 规则匹配完成后, 生成新表达式则加入图中, 并移除旧表达式, 重新执行 1
   1. 如果生成多个新表达式, 则基于 cost, 取一个局部最优
   2. 移除旧表达式的过程是指: 将所有以旧表达式为输入的节点, 其引用调整为新表达式
3. 当所有节点遍历后, 不会生成新表达式, 则结束
4. 最终由根节点遍历，即可获取最终生成的 Plan

1. 任何关系表达式，会将其所有节点进行规则匹配，将匹配完成的规则放入待执行队列.
   • 匹配逻辑为: 规则描述树的结构(每一层的节点类型), 基于节点类型，找到相关的规则。基于当前类型在规则中的位置，进行匹配。
   • 比如: 一个规则 Project <- Aggregate
     • 匹配 Aggregate 节点时, 找到该规则，找其上级节点是否为 Project
     • 匹配 Project 节点时, 找到该规则，找其下级节点是否为 Aggregate
2. 遍历匹配的规则清单，执行，完成后，如果生成新的表达式，同 1
3. 规则清单为空时，则完成生成，退出。³

生成完成后，是一张图。其中每个节点为一个集合，集合内是等价表达式。 表达式间会有依赖(比如 Aggregate 作为 Project 的 input )。

图中存在多条路径可以达到 root 节点(目标的关系表达式), 最终就会需要从这张图中找到一个开销最低的路径.(亦即下图中由顶点开始的蓝线部分)

同样看一下例子:

• sql

SELECT
          date,
          vid,
          domain,
          os,
          ver,
          sum(audio_video) AS totalusage,
          sum(video) AS videousage,
          sum(audio) AS audio_usage
        FROM
          metricsdb.agora_vid_usage_cube_di
        WHERE
          group_mark = '(product_line,product_type,vid,domain,os,ver)'
          AND ver IN (
            '3.0.0.11',
            '3,0,0,16',
            '3.0.0.17',
            '3.0.0.19',
            '3.0.0.18',
            '3.0.20.20',
            '3.0.20.21'
          )
          AND date >= '2020-08-01'
        GROUP BY
          1,
          2,
          3,
          4,
          5

• 关系表达式

LogicalAggregate(subset=[rel#27:RelSubset#3.ENUMERABLE.[]], group=[{0, 1, 2, 3, 4}], totalusage=[SUM($5)], videousage=[SUM($6)], audio_usage=[SUM($7)]): rowcount = 1.0, cumulative cost = {1.4125001430511475 rows, 0.0 cpu, 0.0 io}, id = 25
  LogicalProject(subset=[rel#24:RelSubset#2.NONE.[]], date=[$0], vid=[$2], domain=[$3], os=[$4], ver=[$5], audio_video=[$6], video=[$8], audio=[$7]): rowcount = 1.0, cumulative cost = {1.0 rows, 8.0 cpu, 0.0 io}, id = 23
    LogicalFilter(subset=[rel#22:RelSubset#1.NONE.[]], condition=[AND(=($1, '(product_line,product_type,vid,domain,os,ver)'), OR(=($5, '3.0.0.11'), =($5, '3,0,0,16'), =($5, '3.0.0.17'), =($5, '3.0.0.19'), =($5, '3.0.0.18'), =($5, '3.0.20.20'), =($5, '3.0.20.21')), >=($0, '2020-08-01'))]): rowcount = 1.0, cumulative cost = {1.0 rows, 1.0 cpu, 0.0 io}, id = 21
      LogicalTableScan(subset=[rel#20:RelSubset#0.NONE.[]], table=[[metricsdb, agora_vid_usage_cube_di]]): rowcount = 1.0, cumulative cost = {1.0 rows, 2.0 cpu, 0.0 io}, id = 16

• 初始图

• 完成图