Flink SQL 代码生成

从”UDF不应有状态” 切入来剖析Flink SQL代码生成

核心类: org.apache.flink.table.runtime.CRowProcessRunner

打印出codegen语句:

log4j.logger.org.apache.flink.runtime.CRowProcessRunner=DEBUG, file

alter table
  b_cdg_cft_t_trade_user_fund_merge_ods_t_test_fund_Facc_time 
set
  (
    'consumerGroup' = 't_cdg_cft_b_cdg_cft_t_trade_user_fund_merge_ods_cg_rtdw_prod_003',
    'inDebugMode' = 'true',
    'consume-from-max' = 'true'
  );

-- 自定义控制台输出下游数据表，方便查看运行结果
alter table
  console
set
  (
    -- 同上，console的schema定义的是下游表的元数据
    'schema' = 'Facc_time:varchar,Flistid:varchar,Fsg_sh_type:bigint,Ftotal_fee:bigint,Funion_id:bigint,Fsg_sh_name:varchar,Ftrade_id:varchar'
  );
-- 结果数据写入console，可以在应用运维->TM日志中查看结果输出
insert into
  console
  select
    Facc_time,
    Flistid,
    Fsg_sh_type,
    Ftotal_fee,
    Funion_id,
    Fsg_sh_name,
    Ftrade_id
from
    b_cdg_cft_t_trade_user_fund_merge_ods_t_test_fund_Facc_time

2021-10-27 15:06:43,520 DEBUG [Task-46f8730428df9ecd6d7318a02bdc405e(0/1)] org.apache.flink.table.runtime.conversion.CRowToJavaTupleMapRunner  - Compiling MapFunction: DataStreamSinkConversion$14 

 Code:

public class DataStreamSinkConversion$14 extends org.apache.flink.api.common.functions.RichMapFunction {


  final org.apache.flink.types.Row out =
      new org.apache.flink.types.Row(7);

  private org.apache.flink.types.Row in1;


  public DataStreamSinkConversion$14() throws Exception {


  }




  @Override
  public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {


  }

  @Override
  public Object map(Object _in1) throws Exception {
    in1 = (org.apache.flink.types.Row) _in1;

    boolean isNull$13 = (java.lang.String) in1.getField(6) == null;
    java.lang.String result$12;
    if (isNull$13) {
      result$12 = "";
    }
    else {
      result$12 = (java.lang.String) (java.lang.String) in1.getField(6);
    }


    boolean isNull$11 = (java.lang.String) in1.getField(5) == null;
    java.lang.String result$10;
    if (isNull$11) {
      result$10 = "";
    }
    else {
      result$10 = (java.lang.String) (java.lang.String) in1.getField(5);
    }


    boolean isNull$5 = (java.lang.Long) in1.getField(2) == null;
    long result$4;
    if (isNull$5) {
      result$4 = -1L;
    }
    else {
      result$4 = (java.lang.Long) in1.getField(2);
    }


    boolean isNull$1 = (java.lang.String) in1.getField(0) == null;
    java.lang.String result$0;
    if (isNull$1) {
      result$0 = "";
    }
    else {
      result$0 = (java.lang.String) (java.lang.String) in1.getField(0);
    }


    boolean isNull$9 = (java.lang.Long) in1.getField(4) == null;
    long result$8;
    if (isNull$9) {
      result$8 = -1L;
    }
    else {
      result$8 = (java.lang.Long) in1.getField(4);
    }


    boolean isNull$3 = (java.lang.String) in1.getField(1) == null;
    java.lang.String result$2;
    if (isNull$3) {
      result$2 = "";
    }
    else {
      result$2 = (java.lang.String) (java.lang.String) in1.getField(1);
    }


    boolean isNull$7 = (java.lang.Long) in1.getField(3) == null;
    long result$6;
    if (isNull$7) {
      result$6 = -1L;
    }
    else {
      result$6 = (java.lang.Long) in1.getField(3);
    }







    if (isNull$1) {
      out.setField(0, null);
    }
    else {
      out.setField(0, result$0);
    }



    if (isNull$3) {
      out.setField(1, null);
    }
    else {
      out.setField(1, result$2);
    }



    if (isNull$5) {
      out.setField(2, null);
    }
    else {
      out.setField(2, result$4);
    }



    if (isNull$7) {
      out.setField(3, null);
    }
    else {
      out.setField(3, result$6);
    }



    if (isNull$9) {
      out.setField(4, null);
    }
    else {
      out.setField(4, result$8);
    }



    if (isNull$11) {
      out.setField(5, null);
    }
    else {
      out.setField(5, result$10);
    }



    if (isNull$13) {
      out.setField(6, null);
    }
    else {
      out.setField(6, result$12);
    }

    return out;

  }

  @Override
  public void close() throws Exception {


  }
}

2021-10-27 15:06:43,520 DEBUG [Task-f24f04d8dcfb922ebc646161f8d38c37(0/1)] org.apache.flink.table.runtime.CRowMapRunner                  - Compiling MapFunction: DataStreamSinkConversion$29 

 Code:

public class DataStreamSinkConversion$29 extends org.apache.flink.api.common.functions.RichMapFunction {


  final org.apache.flink.types.Row out =
      new org.apache.flink.types.Row(7);

  private org.apache.flink.types.Row in1;


  public DataStreamSinkConversion$29() throws Exception {


  }




  @Override
  public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {


  }

  @Override
  public Object map(Object _in1) throws Exception {
    in1 = (org.apache.flink.types.Row) _in1;

    boolean isNull$28 = (java.lang.String) in1.getField(6) == null;
    java.lang.String result$27;
    if (isNull$28) {
      result$27 = "";
    }
    else {
      result$27 = (java.lang.String) (java.lang.String) in1.getField(6);
    }


    boolean isNull$26 = (java.lang.String) in1.getField(5) == null;
    java.lang.String result$25;
    if (isNull$26) {
      result$25 = "";
    }
    else {
      result$25 = (java.lang.String) (java.lang.String) in1.getField(5);
    }


    boolean isNull$20 = (java.lang.Long) in1.getField(2) == null;
    long result$19;
    if (isNull$20) {
      result$19 = -1L;
    }
    else {
      result$19 = (java.lang.Long) in1.getField(2);
    }


    boolean isNull$16 = (java.lang.String) in1.getField(0) == null;
    java.lang.String result$15;
    if (isNull$16) {
      result$15 = "";
    }
    else {
      result$15 = (java.lang.String) (java.lang.String) in1.getField(0);
    }


    boolean isNull$24 = (java.lang.Long) in1.getField(4) == null;
    long result$23;
    if (isNull$24) {
      result$23 = -1L;
    }
    else {
      result$23 = (java.lang.Long) in1.getField(4);
    }


    boolean isNull$18 = (java.lang.String) in1.getField(1) == null;
    java.lang.String result$17;
    if (isNull$18) {
      result$17 = "";
    }
    else {
      result$17 = (java.lang.String) (java.lang.String) in1.getField(1);
    }


    boolean isNull$22 = (java.lang.Long) in1.getField(3) == null;
    long result$21;
    if (isNull$22) {
      result$21 = -1L;
    }
    else {
      result$21 = (java.lang.Long) in1.getField(3);
    }







    if (isNull$16) {
      out.setField(0, null);
    }
    else {
      out.setField(0, result$15);
    }



    if (isNull$18) {
      out.setField(1, null);
    }
    else {
      out.setField(1, result$17);
    }



    if (isNull$20) {
      out.setField(2, null);
    }
    else {
      out.setField(2, result$19);
    }



    if (isNull$22) {
      out.setField(3, null);
    }
    else {
      out.setField(3, result$21);
    }



    if (isNull$24) {
      out.setField(4, null);
    }
    else {
      out.setField(4, result$23);
    }



    if (isNull$26) {
      out.setField(5, null);
    }
    else {
      out.setField(5, result$25);
    }



    if (isNull$28) {
      out.setField(6, null);
    }
    else {
      out.setField(6, result$27);
    }

    return out;

  }

  @Override
  public void close() throws Exception {


  }
}

一个样例

CREATE TABLE mysql_orders (
  ei STRING,
  sei STRING,
  ui STRING,
  etime STRING
) WITH (
'connector.type' = 'jdbc',
   'connector.driver' = 'com.mysql.cj.jdbc.Driver',
   'connector.url' = 'jdbc:mysql://localhost:3306/flink_demo',
   'connector.table' = 'orders',
   'connector.username' = 'root',
   'connector.password' = 'abc.ABC.123'
) 

insert into mysql_orders select concat(ei,'#',sei) as pk,sei,ui,etime from orders

calcite解析后:

org.apache.calcite.sql2rel,DEBUG,Plan after converting SqlNode to RelNode
LogicalProject(pk=[CONCAT($0, _UTF-16LE'#', $1)], sei=[$1], ui=[$2], etime=[$3])
  LogicalTableScan(table=[[default_catalog, default_database, Unregistered_TableSource_566698125, source: [CsvTableSource(read fields: ei, sei, ui, etime)]]])

优化:

org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram,DEBUG,optimize convert table references before rewriting sub-queries to semi-join cost 31 ms.
optimize result:
 LogicalSink(name=[`default_catalog`.`default_database`.`mysql_orders`], fields=[ei, sei, ui, etime])
+- LogicalProject(ei=[CONCAT($0, _UTF-16LE'#', $1)], sei=[CAST($1):VARCHAR(2147483647) CHARACTER SET "UTF-16LE"], ui=[CAST($2):VARCHAR(2147483647) CHARACTER SET "UTF-16LE"], etime=[CAST($3):VARCHAR(2147483647) CHARACTER SET "UTF-16LE"])
   +- LogicalTableScan(table=[[default_catalog, default_database, Unregistered_TableSource_566698125, source: [CsvTableSource(read fields: ei, sei, ui, etime)]]])

org.apache.calcite.plan.RelOptPlanner,DEBUG,Cheapest plan:
FlinkLogicalSink(name=[`default_catalog`.`default_database`.`mysql_orders`], fields=[ei, sei, ui, etime]): rowcount = 1.0E8, cumulative cost = {3.0E8 rows, 3.0E8 cpu, 4.8E9 io, 0.0 network, 0.0 memory}, id = 73
  FlinkLogicalCalc(select=[CONCAT(ei, _UTF-16LE'#', sei) AS ei, CAST(sei) AS sei, CAST(ui) AS ui, CAST(etime) AS etime]): rowcount = 1.0E8, cumulative cost = {2.0E8 rows, 2.0E8 cpu, 4.8E9 io, 0.0 network, 0.0 memory}, id = 72
    FlinkLogicalTableSourceScan(table=[[default_catalog, default_database, Unregistered_TableSource_566698125, source: [CsvTableSource(read fields: ei, sei, ui, etime)]]], fields=[ei, sei, ui, etime]): rowcount = 1.0E8, cumulative cost = {1.0E8 rows, 1.0E8 cpu, 4.8E9 io, 0.0 network, 0.0 memory}, id = 71

实际的执行计划分为三层: