flink sql源码分析一之执行流程梳理

山行AI

发布于 2020-12-14 10:04:40

1.8K0

发布于 2020-12-14 10:04:40

文章被收录于专栏：山行AI山行AI

前言

我们在梳理flink sql 执行流程时以sql解析、sql校验、sql转化及sql优化的顺序来展开，本篇主要是对过程的梳理，不会涉及过多的代码部分，后面会针对各环节进行逐一分析。

Parser

Validate

这里以SqlQuery操作的convert过程为例：

转换过程

上面是TableEnvironmentImpl中的translate方法入口，我们来具体分析下planner.translate方法在PlannerBase转换过程如下：

我们来看一下具体的optimize过程：

将FlinkPhysicalRel DAG转换成ExecNode DAG

@VisibleForTesting
private[flink] def translateToExecNodePlan(
                                            optimizedRelNodes: Seq[RelNode]): util.List[ExecNode[_, _]] = {
  require(optimizedRelNodes.forall(_.isInstanceOf[FlinkPhysicalRel]))
  // Rewrite same rel object to different rel objects
  // in order to get the correct dag (dag reuse is based on object not digest)
  val shuttle = new SameRelObjectShuttle()
  val relsWithoutSameObj = optimizedRelNodes.map(_.accept(shuttle))
  // reuse subplan
  val reusedPlan = SubplanReuser.reuseDuplicatedSubplan(relsWithoutSameObj, config)
  // convert FlinkPhysicalRel DAG to ExecNode DAG
  reusedPlan.map(_.asInstanceOf[ExecNode[_, _]])
}

在translateToExecNodePlan方法中将FlinkPhysicalRel DAG转换成ExecNode DAG并尝试复用重复的子计划。

ExecNode转换成Transformation

下面代码是将ExecNode列表转成Transformation列表的入口：

  override protected def translateToPlan(
      execNodes: util.List[ExecNode[_, _]]): util.List[Transformation[_]] = {
    val planner = createDummyPlanner()
    planner.overrideEnvParallelism()

    execNodes.map {
          // 将execNode转成Transformation
      case node: StreamExecNode[_] => node.translateToPlan(planner)
      case _ =>
        throw new TableException("Cannot generate DataStream due to an invalid logical plan. " +
          "This is a bug and should not happen. Please file an issue.")
    }
  }