基于flink1.14的源码做解析

公司内有很多业务方都在使用我们Flink sql平台做TopN的计算，今天同事突然问到我，Flink sql 是怎么实现topN的 ？

蒙圈了，这块源码没看过啊 ，业务要问起来怎么办，赶快打开源码补一下

拿到这个问题先冷静分析一下范围

首先肯定属于Flink sql模块，源码里面肯定是在flink-table-planner包里面，接着topN那不就是ROW_NUMBER嘛，是个函数呀

既然如此那就从flink源码的系统函数作为线索开始找起来，来到 org.apache.calcite.sql.fun.SqlStdOperatorTable类

果然找到了，那calcite的某个rule肯定有个地方判断了它，继续查调用链

不出所料，FlinkLogicalRankRuleBase这个calcite的rule里面果然根据这个function的类型来确定rank的类型了

看下这个rule的匹配条件

 这里也好理解，overAgg的时候会判断这个rank以及对应的类型

这是只是做了一下简单的提取了rank的字段啊，提取谓语啊，提取表达式啊这一些拿信息的操作

然后直接生成新的relNode叫FlinkLogicalRank通过transformTo直接返回了这个等价节点

既然是relNode那肯定又会有calcite的rule去处理它，来找一找

批处理的就不管了，从名字就可以看出来我们要找的类了

看个不带window的吧

 返回StreamPhysicalRank

这个类是一个FlinkPhysicalRel是可以转换成execNode的

返回的这个StreamExecRank就是可以转换成具体的Flink的算子了，具体逻辑就在里面了

接下来看下row_number的具体逻辑，找到方法translateToPlanInternal

根据策略主要分为三种类型

AppendFastStrategy  （输入仅包含插入时）

RetractStrategy   （输入包含update和delete）

UpdateFastStrategy     (输入不应包含删除且输入有给定的primaryKeys且按字段排序时)

来看个retractStrategy的吧

先通过sort的字段获取一个用于排序RowData的比较器 ComparableRecordComparator

根据比较器创建 RetractableTopNFunction

这个类还有两个主要的状态数据结构

dataState这个map用来存放当key相同的所有数据会放在同一个list里面

treeMap这个可排序的map就是通过上面我们sql里面定义的sort by 来排序数据的，Long是指这个相同的key有多少个record

!!!!!!!!!!!  那就是用java的treeMap排序呗

继续往下看

 主逻辑就是这个了

每进入一条数据，会根据这条数据的类型划分

当数据是Insert , UPDATE_AFTER类型是会走 emitRecordsWithRowNumber（）方法

当数据是UPDATE_BEFORE，DELETE类型会走 retractRecordWithRowNumber （）方法

来看下具体逻辑先看INSERT的

 遍历treeMap

解读一下，当数据是insert数据的时候

INSERT数据会先放到treeMap里面去，delete则不会

按顺序遍历treeMap

当遍历过程中发现遍历的key与当前数据的key相同时，和当前数据key相同的所有数据数据（dataState中的LIST），全部撤回并且更新他们的rowNumber+1

继续遍历treeMap

之后的数据全部撤回UpdateBefore，并且向下游发送UpdateAfter使rowNumber+1，遍历直到已经到第TopN个数据循环结束

 

当数据是DELETE类型的时候，会和Insert反过来，当前key之后的数据全部撤回，然后rowNumber-1

 

整个处理流程差不多就结束了，可以看到rowNumber当N较大且排序变化频繁的时候，性能消耗还是非常大的，极端情况下游的数据会翻很多倍

这个还需要注意在其他两个策略中还有一个参数，table.exec.topn.cache-size

 

 影响下面这个本地lruCache的大小

 

 调大可以减少状态的访问，可以按需要添加