• 接上篇 paimon sink 源码 之 paimon table 创建
• 对于 insert into table select * from xx, 经过 flink ParserImpl parse 之后会转化成 SinkModifyOperation
• SinkModifyOperation 经过 Panner 的 translate 会把 SinkModifyOperation 转化成 RelNode
  • 在转化 RelNode 之前会先执行 org.apache.flink.table.catalog.Catalog#getFactory 方法获取
    fink 的 DynamicTableFactory
  • DynamicTableFactory 用来创建 DynamicTableSource / DynamicTableSink， 对于 insert 就是会创建 DynamicTableSink
  • 然后会把 DynamicTableSink 转化成 Calcite 的 RelNode
• RelNode 经过 Panner 的 optimize
• 然后将 optimizedRelNodes 转化成 ExecNodeGraph
• 再把 ExecNodeGraph 转化成 Transformation
• Transformation 转化过程定义了 sink 的 pipeline
• 大概逻辑如下图

image.png

• 接下来看 Paimon 的 DynamicTableSink

Paimon 对于 flink DynamicTableSink 的实现

DynamicTableSink 的创建是通过 flink DynamicTableFactory 创建的
DynamicTableFactory 来源于 org.apache.flink.table.catalog.Catalog#getFactory

• Paimon 对于 org.apache.flink.table.catalog.Catalog 的实现为 FlinkCatalog 和 FlinkGenericCatalog

Paimon 的 FlinkTableFactory

以 FlinkCatalog 为例 getFactory 方法直接创建了 org.apache.paimon.flink.FlinkTableFactory

  @Override
  public Optional<Factory> getFactory() {
      return Optional.of(new FlinkTableFactory());
  }

• FlinkTableFactory 类图

  
  org.apache.paimon.flink.FlinkTableFactory

• FlinkTableFactory 实现了 createDynamicTableSource，createDynamicTableSink 用来创建 DynamicTableSource 和 DynamicTableSink

• 在上面分析对于 DynamicTable 的创建是在 Operation 转 RelNode 过程创建的

• 接下来看 Paimon 的 DynamicTableSink 实现

Paimon 的 FlinkTableSink

org.apache.paimon.flink.sink.FlinkTableSink

在 ExecNodeGraph 转 Transformation 过程会调用 FlinkTableSink 的 getSinkRuntimeProvider 方法，而这里面就定义整个 sink 算子的 pipeline 逻辑。

FINAL

• 梳理了一条 sql 到最终对 DataStream 操作的大概转换逻辑
• 在转化的过程中会有 Paimon 的 DynamicTableFactory 创建 和 Paimon DynamicTableSink 的创建
• 而 Paimon DynamicTableSink 定义了 Paimon sink 对 DataStream 操作的 pipeline
• paimon sink 源码之 dataStream 的拓扑梳理