- 接上篇 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 之前会先执行 org.apache.flink.table.catalog.Catalog#getFactory 方法获取
- 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 的拓扑梳理
