Flink SQL算子篇（二）聊聊选择与投影

select  name
       ,score+1
from test
where score>15

protected Transformation<RowData> translateToPlanInternal(PlannerBase planner) {
    final ExecEdge inputEdge = getInputEdges().get(0);
    // 取得执行图上的前驱 Transformation
    final Transformation<RowData> inputTransform =
            (Transformation<RowData>) inputEdge.translateToPlan(planner);
    // 得到 CodeGen 的上下文
    final CodeGeneratorContext ctx =
            new CodeGeneratorContext(planner.getTableConfig())
                    .setOperatorBaseClass(operatorBaseClass);
    // 生成 CodeGenOperatorFactory 类型的 Operator 算子
    final CodeGenOperatorFactory<RowData> substituteStreamOperator =
            CalcCodeGenerator.generateCalcOperator(
                    ctx,
                    inputTransform,
                    (RowType) getOutputType(),
                    JavaScalaConversionUtil.toScala(projection),
                    JavaScalaConversionUtil.toScala(Optional.ofNullable(this.condition)),
                    retainHeader,
                    getClass().getSimpleName());
    // 构造出 Transformation
    return new OneInputTransformation<>(
            inputTransform,
            getDescription(),
            substituteStreamOperator,
            InternalTypeInfo.of(getOutputType()),
            inputTransform.getParallelism());
}

public OneInputTransformation(
        Transformation<IN> input,
        String name,
        OneInputStreamOperator<IN, OUT> operator,
        TypeInformation<OUT> outputType,
        int parallelism
    // 调用Operator工厂类为参数的构造方法
    this(input, name, SimpleOperatorFactory.of(operator), outputType, parallelism);
}
public OneInputTransformation(
        Transformation<IN> input,
        String name,
        StreamOperatorFactory<OUT> operatorFactory,
        TypeInformation<OUT> outputType,
        int parallelism) {
    super(name, outputType, parallelism);
    this.input = input;
    this.operatorFactory = operatorFactory;
}

select a+b from test

public class StreamExecCalc$11 extends org.apache.flink.table.runtime.operators.TableStreamOperator
          implements org.apache.flink.streaming.api.operators.OneInputStreamOperator {
...
}

public void processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord element) throws Exception {
  // 接收输入数据
  org.apache.flink.table.data.RowData in1 = (org.apache.flink.table.data.RowData) element.getValue();
  // 一次性创建后续需要用到的所有字段
  int field$4;
  boolean isNull$4;
  boolean isNull$5;
  boolean result$6;
  org.apache.flink.table.data.binary.BinaryStringData field$7;
  boolean isNull$7;
  org.apache.flink.table.data.binary.BinaryStringData field$9;
  boolean isNull$10;
  int result$11;
  // 获取score的值
  isNull$4 = in1.isNullAt(1);
  field$4 = -1;
  if (!isNull$4) {
    field$4 = in1.getInt(1);
  }
  // 判断score大于15
  isNull$5 = isNull$4 || false;
  result$6 = false;
  if (!isNull$5) {
    result$6 = field$4 > ((int) 15);
  }
  // 如果score大于15，则输出name和score+1的结果记录
  if (result$6) {
  isNull$7 = in1.isNullAt(0);
  field$7 = org.apache.flink.table.data.binary.BinaryStringData.EMPTY_UTF8;
  if (!isNull$7) {
    field$7 = ((org.apache.flink.table.data.binary.BinaryStringData) in1.getString(0));
  }
  field$9 = field$7;
  if (!isNull$7) {
    field$9 = (org.apache.flink.table.data.binary.BinaryStringData) (typeSerializer$8.copy(field$9));
  }
  out.setRowKind(in1.getRowKind());
  if (isNull$7) {
    out.setNullAt(0);
  } else {
    out.setNonPrimitiveValue(0, field$9);
  }
  isNull$10 = isNull$4 || false;
  result$11 = -1;
  if (!isNull$10) { 
    result$11 = (int) (field$4 + ((int) 1));  
  }  
  if (isNull$10) {
    out.setNullAt(1);
  } else {
    out.setInt(1, result$11);
  }             
  output.collect(outElement.replace(out));
  }
}

public void processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord element) throws Exception {
   接收输入数据
  org.apache.flink.table.data.RowData in1 = (org.apache.flink.table.data.RowData) element.getValue();
  ...
  // 获取score的值
  field$4 = in1.getInt(1);
  // 判断score大于15
  result$6 = field$4 > ((int) 15);
  ...
  / 如果score大于15，则输出name和score+1的结果记录
  if (result$6) {
    ...
    // 拿到并设置name的值
    field$7 = ((org.apache.flink.table.data.binary.BinaryStringData) in1.getString(0));
    ...
    field$9 = field$7;
    ...
    out.setNonPrimitiveValue(0, field$9);
    ...
    // 计算score+1并设置到输出结果中
    result$11 = (int) (field$4 + ((int) 1));  
    ...
    out.setInt(1, result$11);
    ...
    // 输出
    output.collect(outElement.replace(out));
  }
}

+I[John, 19]
+I[Tom, 19]
+I[Tom, 20]