SQL Impl in Flink

跟了下Flink Table裡sql的實現，flink sql的實現比較簡單，一句話概述就是：藉助Apache Calcite做了sql解析、邏輯樹生成的過程，得到Calcite的RelRoot類，生成flink的Table，Table裡的執行計劃會轉化成DataSet的計算，經歷物理執行計劃優化等步驟。

類比Spark SQL，Calcite代替了大部分Spark SQL Catalyst的工作(Catalyst還包括了Tree/Node的定義，這部分程式碼Flink也’借鑑’來了)。兩者最終是計算一顆邏輯執行計劃樹，翻譯成各自的DataSet(Spark 2.0引入Dataset

Calcite Usage

最新Flink程式碼裡，在flink-table工程裡，使用1.7版本的calcite-core。

大致的執行過程如下：

1. TableEnvironment.sql()為呼叫入口
2. 類似Calcite的PlannerImpl，flink實現了個FlinkPlannerImpl，執行parse(sql)，validate(sqlNode)，rel(sqlNode)操作
3. 生成Table

  override def sql(query: String): Table = {

    val
 
 planner = new FlinkPlannerImpl(getFrameworkConfig, getPlanner)
    // parse the sql query
    val parsed = planner.parse(query)
    // validate the sql query
    val validated = planner.validate(parsed)
    // transform to a relational tree
    val relational = planner.rel(validated)

    new Table(this, LogicalRelNode(relational.rel))
  }
 

LogicalRelNode是flink執行計算樹裡的葉子節點。其他節點的實現類最終都會轉化成Calcite的RelBuilder生成一個可被Calcite繼續執行計劃優化的plan，邏輯在TableEnv的translate(table)方法裡。

 protected def translate[A](table: Table)(implicit tpe: TypeInformation[A]): DataSet[A] = {

    val relNode = table.getRelNode

    // decorrelate
    val decorPlan = RelDecorrelator.decorrelateQuery(relNode)

    // optimize the logical Flink plan
    val optProgram = Programs.ofRules(FlinkRuleSets.DATASET_OPT_RULES)
    val flinkOutputProps = relNode.getTraitSet.replace(DataSetConvention.INSTANCE).simplify()

    val dataSetPlan = try {
      optProgram.run(getPlanner, decorPlan, flinkOutputProps)
    }
    catch {
      // ...
    }

    dataSetPlan match {
      case node: DataSetRel =>
        node.translateToPlan(
          this,
          Some(tpe.asInstanceOf[TypeInformation[Any]])
        ).asInstanceOf[DataSet[A]]
      case _ => ???
    }
  }

在Programs.ofRules(FlinkRuleSets.DATASET_OPT_RULES)裡，flink根據Calcite的介面定義了幾個將最終物理計劃轉化為DataSet的Rule：

    // translate to Flink DataSet nodes
    DataSetAggregateRule.INSTANCE,
    DataSetCalcRule.INSTANCE,
    DataSetJoinRule.INSTANCE,
    DataSetScanRule.INSTANCE,
    DataSetUnionRule.INSTANCE,
    DataSetSortRule.INSTANCE,
    DataSetValuesRule.INSTANCE,
    BatchTableSourceScanRule.INSTANCE

每條規則會對應生成一個物理節點，org.apache.flink.api.table.plan.nodes.dataset package下。節點內，根據Calcite生成的sql的執行步驟，會進行codegen出DataSet的執行Function程式碼，在org.apache.flink.api.table.runtime package下，目前生成三種ds操作: FlatMapRunner， FlatJoinRunner， 和MapRunner.

codegen部分與Spark SQL的結構相類似。

Calcite在Flink中的使用也比較基本，單測sql package下的case就可以走通上面的呼叫過程。

整體Flink sql上的功能和實現要比Spark SQL簡單很多。並可能存在許多借鑑之處。