流批一体向量化计算引擎 Flex 在蚂蚁的探索和实践

编者按：Flex是蚂蚁数据部自研的一款流批一体的向量化引擎，Flex是Fink和Velox的全称，也是Flexible的前缀，被赋予了灵活可插拔的寓意。本文将重点从向量化技术背景、Flex架构方案和未来规划三个方面展开论述。

作者介绍：刘勇，蚂蚁数据部分布式计算引擎技术专家，Calcite Committer、Flink Contributor。毕业8年一直对分布式计算领域保持着十年如一日的热情与专注。

一、背景

1. 什么是向量化计算？

1）并行数据处理SIMD指令

以下面循环代码为例，计算在CPU内完成需经三步：

加载（Load），从内存加载2个源操作数（a[i]和b[i]）到2个寄存器。
计算（Compute），执行加法指令，作用于2个寄存器里的源操作数副本，结果产生到目标寄存器。
存储（Store），将目标寄存器的数据存入（拷贝）到目标内存位置（c[i]）。

void addArrays(const int* a, const int* b, int* c, int num) {
  for (int i = 0; i < num; ++i) {
    c[i] = a[i] + b[i];
  }
}

该流程即对应传统的计算架构：单指令单数据（SISD）顺序架构，任意时间点只有一条指令作用于一条数据流。如果有更宽的寄存器（超机器字长，比如256位16字节），一次性从源内存同时加载更多的数据到寄存器，一条指令作用于寄存器x和y，在x和y的每个分量（比如32位4字节）上并行进行加，并将结果存入寄存器z的各对应分量，最后一次性将寄存器z里的内容存入目标内存，那么就能实现单指令并行处理数据的效果，这就是单指令多数据（SIMD）。

2）向量化执行框架具有的特性

执行引擎常规按行处理的方式，存在以下问题：

CPU Cache命中率差。一行的多列（字段）数据的内存紧挨在一起，哪怕只对其中的一个字段做操作，其他字段所占的内存也需要加载进来，这会抢占稀缺的Cache资源。Cache命中会导致被请求的数据从内存加载进Cache，等待内存操作完成会导致CPU执行指令暂停（Memory Stall），这会增加延时，还可能浪费内存带宽。
变长字段影响计算效率。假设一行包括int、string、int三列，其中int类型是固定长度，而string是变长的（一般表示为int len + bytes content），变长列的存在会导致无法通过行号算offset做快速定位。
虚函数开销。对一行的多列进行处理通常会封装在一个循环里，会抽象出一个类似handle的接口（C++虚函数）用于处理某类型数据，各字段类型会override该handle接口。虚函数的调用多一步查表，且无法被内联，循环内高频调用虚函数的性能影响不可忽。

因此，要让向量化计算发挥威力，只使用SIMD指令还不够，还需要对框架层面进行改造，数据按列组织。

数据按列组织将提高数据局部性。参与计算的列的多行数据会内存紧凑的保存在一起，CPU可以通过预取指令将接下来要处理的数据加载进Cache，从而减少Memory Stall。不参与计算的列的数据不会与被处理的列竞争Cache，这种内存交互的隔离能提高Cache亲和性。
同一列数据在循环里被施加相同的计算。批量迭代将减少函数调用次数，通过模版能减少虚函数调用，降低运行时开销。针对固定长度类型的列很容易被并行处理（通过行号offset到数据），这样的执行框架也有利于让编译器做自动向量化代码生成，显著减少分支，减轻预测失败的惩罚。结合模板，编译器会为每个实参生成特定实例化代码，避免运行时查找虚函数表，并且由于编译器知道了具体的类型信息，可以对模板函数进行内联展开。

2. 向量化计算在分布式计算领域的现状

随着技术的发展，向量化技术在硬件、指令集、配套工具、类库等得到多方位协同发展。

在指令集层面，当前大多数机器都支持SIMD指令集，比如x86平台的SSE、AVX指令集，ARM平台的NEON指令集。
在编译器层面，现代编译器如GCC、LLVM可以自动将代码中可向量化的部分转成SIMD指令。比如开源有一个基于arrow的表达式计算框架gandiva就是基于LLVM生成SIMD指令。
在类库层面，有支持跨平台的库XSimd，可同时运行在x86和arm架构。

上面讲述了向量化的技术原理和可供使用的方案，我们再看下大数据领域在向量化方向做了哪些探索。首先看下Photon，Photon是Databricks闭源的一个全新的向量化引擎，完全用C++编写。Databricks在过去几年里，一直通过各种技术手段对内部的Spark版本Databricks Runtimes进行优化。

左下图展示了不同DBR版本相对于2.1版本的性能提升情况，可以看到未开启Photon的版本性能提升在1-2倍，自从上了Photon后，性能提升可以达到3-8倍。自从Databricks论文公布其方案和效果后，大数据领域进行了各种尝试和探索。主流分布式计算引擎都在往向量化方向发力且效果显然，Flink作为流批一体实时计算引擎在湖仓一体中起到了关键核心作用，但缺少核心的向量化能力，因此我们蚂蚁在该领域发起了挑战。

由于目前开源社区已经有比较成熟的 Native Engine，如ClickHouse、Velox，具备了优秀的向量化、批处理、列计算等能力，并经过业界广泛验证和实践。因此，我们不打算重头造轮子，而是站在巨人的肩膀上。采用了类似Gluten的实现方式，基于Velox上构建。同时，为了防止闭门造车，在立项之出选择和Gluten社区合作，讨论构建方案以及加强紧密型合作。

自此，该项目在蚂蚁正式落地，内部代号Flex，Flex是Fink和Velox的全称，也是Flexible的前缀，我们希望它能够做到灵活可插拔，像胶水一样扮演其工作。

3. 流批一体向量化引擎在蚂蚁的探索

因此，今年我们在Flink 1.18中首次引入了流批一体向量化计算，通过使用C++开发，利用 SIMD、向量化、列计算等技术，在完全兼容 Flink SQL 语法的基础之上，可以提供比原生的执行引擎更高的吞吐，并可以降低用户 Flink 作业的执行成本。同时，SQL语义对齐，对用户来说，仅仅需要增加配置即可开启，降低存量用户使用成本。该技术具有较强的创新性和技术挑战性，属于自主研发，弥补了业界空白。

二、总体方案

Velox算子面向传统批处理场景，算子本身无法处理回撤消息，即天然和流式计算场景相违背，因此无法直接基于gluten的substrain plan方案。该方案实现复杂工作量大，但是也没法支持状态算子。同时，为了复用Velox一些现有的能力，不必重复造轮子，因此我们另辟蹊径采用了新的方案，新方案可以做到真正的流批一体；同时，向量化核心效果在于表达式计算。因此非常适合Calc算子，而且Calc算子不区分流和批，也就是说流任务和批任务都可以享受到向量化带来的技术红利，也可以使用到Velox的表达式计算能力。为此，我们从功能性、正确性、易用性和稳定性四个方面进行全方位建设。

1. 功能性建设

1）Native算子层优化

✨NativeCalc算子优化

在用户向量化作业验证过程中，发现对于任务中含有很多RexInputRef引用字段，这些字段数据内容长且没有加工逻辑，如果把使用到的Scheme字段全部做攒批和数据转换，转换开销将近11%。我们可以将引用字段单独拆出来，拆成两个Calc算子，在NativeCalc算子层仅仅将表达式中使用的字段进行数据转换，其他字段仅仅forward到原有费Calc codegen计算进行取值，最后再拼接JoinedRowData。为了高效的拼接JoinedRowData，自然需要插入Calc算子对字段重排序，因此需要增加规则支持projection reorder。通过该优化数据转换层开销降低到6.68%（3.96+2.72）
同时对于表达式里面含有udf，复用此方案把udf也拆到引用字段的Calc里面，就不需要整个Calc算子fallback。

✨支持NativeSource/NativeSink

对于蚂蚁内部有类似kafka的消息队列sls，sls source/sink当前支持Arrow模式，为了避免冗余的数据行转列开销，插件层直接读写Arrow格式数据，从而避免额外的行转列开销。

2）Plan层优化

为了尽可能多的将DAG中各种算子中包含projection和condition的逻辑中包含simd函数的计算逻辑委托给velox进行计算，我们在plan层增加了多种基于规则的Rule对算子进行拆分合并和交互，全方位的发挥性能极致。目前Calc、Join、Correlate算子都已经支持Native执行
由于流式计算场景需要攒批，发挥列计算的优势，但是攒批和转换数据层存在一定开销，因此对于calc算子中整个表达式tree都没有使用到simd函数，将不翻译成native算子
下面的sql RexCall以blink_json_value为例进行阐述优化规则

✨projection reorder

以下面sql为例，该规则将RexInputRef引用字段排在RexCall前面，通过一个非Native的Cal记录下引用关系。主要用于和算子层优化结合。

SELECT a, blink_json_value(a, c), b FROM MyTable

Calc(select=[a, f0 AS EXPR$1, b])
+- NativeCalc(select=[a, b, blink_json_value(a, c) AS f0])
   +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c, d)]]], fields=[a, b, c, d]

✨projections包含simd函数

该sql仅仅projections含有simd函数，condition没有使用，因此filter逻辑还是使用非native的算子。

SELECT a, blink_json_value(b, d) FROM MyTable
WHERE concat(a, '1') is not null

经过优化后的执行计划如下所示

NativeCalc(select=[a, blink_json_value(b, d) AS EXPR$1])
+- Calc(select=[a, b, d], where=[CONCAT(a, '1') IS NOT NULL])
   +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c, d)]]], fields=[a, b, c, d])

✨condition包含simd函数

该sql projections 没有使用simd函数，condition含有，因此condition部分使用native算子执行。

SELECT a, b, concat(a, '1') FROM MyTable
WHERE blink_json_value(a, c) is not null

经过优化后的执行计划如下所示

Calc(select=[a, b, concat(a, '1') AS EXPR$1], where=[f0])
+- NativeCalc(select=[a, b, blink_json_value(a, c) IS NOT NULL AS f0])
   +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c, d)]]], fields=[a, b, c, d])

✨projections和condition都包含simd函数

该sql projections和condition都含有simd函数，因此拆成两个Native算子执行。

SELECT blink_json_value(a, b), concat(c, '1') FROM MyTable
WHERE blink_json_value(a, c) is not null

经过优化后的执行计划如下所示

NativeCalc(select=[blink_json_value(a, b) AS EXPR$0, CONCAT(c, '1') AS EXPR$1])
+- Calc(select=[a, b, c], where=[f0])
   +- NativeCalc(select=[a, b, c, blink_json_value(a, c) IS NOT NULL AS f0])
      +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c, d)]]], fields=[a, b, c, d])

✨Inner Join的condition中包含simd函数

下面的双流join例子，不仅仅On条件中包含simd函数，后面的Where逻辑也有。

SELECT a, concat(d, '1') FROM(
            SELECT a, d FROM(
            SELECT a, d
                    FROM leftTable JOIN rightTable ON
                    a = d and blink_json_value(a, a) = concat(a, d))
    WHERE blink_json_value(a, d) = concat(a, d))

calcite 解析后的AST树如下


LogicalProject(a=[$0], EXPR$1=[CONCAT($1, _UTF-16LE'1')])
+- LogicalProject(a=[$0], d=[$1])
   +- LogicalFilter(condition=[=(blink_json_value($0, $1), CONCAT($0, $1))])
      +- LogicalProject(a=[$0], d=[$3])
         +- LogicalJoin(condition=[AND(=($0, $3), =(blink_json_value($0, $0), CONCAT($0, $3)))], joinType=[inner])
            :- LogicalTableScan(table=[[default_catalog, default_database, leftTable]])
            +- LogicalTableScan(table=[[default_catalog, default_database, rightTable]])

通过logical plan优化，可以看到两个simd函数，已经抽取到单独的Calc算子，跟在Join后面

FlinkLogicalCalc(select=[a, CONCAT(d, '1') AS EXPR$1], where=[f0])
+- FlinkLogicalCalc(select=[a, d, AND(=(blink_json_value(a, a), CONCAT(a, d)), =(blink_json_value(a, d), CONCAT(a, d))) AS f0])
   +- FlinkLogicalJoin(condition=[=($0, $1)], joinType=[inner])
      :- FlinkLogicalCalc(select=[a])
      :  +- FlinkLogicalLegacyTableSourceScan(table=[[default_catalog, default_database, leftTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
      +- FlinkLogicalCalc(select=[d])
         +- FlinkLogicalLegacyTableSourceScan(table=[[default_catalog, default_database, rightTable, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])

如果不开启native能力，翻译后的ExecPlan如下，其中Calc推到了下面的Join算子

Calc(select=[a, CONCAT(d, '1') AS EXPR$1])
+- Join(joinType=[InnerJoin], where=[((a = d) AND (blink_json_value(a, a) = CONCAT(a, d)) AND (blink_json_value(a, d) = CONCAT(a, d)))], select=[a, d], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
   :- Exchange(distribution=[hash[a]])
   :  +- Calc(select=[a])
   :     +- TableSourceScan(table=[[default_catalog, default_database, leftTable]], fields=[a, b, c])
   +- Exchange(distribution=[hash[d]])
      +- Calc(select=[d])
         +- TableSourceScan(table=[[default_catalog, default_database, rightTable]], fields=[d, e, f])

开启native能力后，simd表达式是委托给了native计算

Calc(select=[a, CONCAT(d, '1') AS EXPR$1], where=[f0])
+- NativeCalc(select=[a, d, ((blink_json_value(a, a) = CONCAT(a, d)) AND (blink_json_value(a, d) = CONCAT(a, d))) AS f0])
   +- Join(joinType=[InnerJoin], where=[(a = d)], select=[a, d], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
      :- Exchange(distribution=[hash[a]])
      :  +- Calc(select=[a])
      :     +- TableSourceScan(table=[[default_catalog, default_database, leftTable, nativeOperator=[false]]], fields=[a, b, c])
      +- Exchange(distribution=[hash[d]])
         +- Calc(select=[d])
            +- TableSourceScan(table=[[default_catalog, default_database, rightTable, nativeOperator=[false]]], fields=[d, e, f])

✨Correlate节点对应物理算子condition中包含simd函数

同理，对于Flink内置udtf函数物理算子实现，由于包含condition，对于该场景也是可以使用native进行加速

Calc(select=[a, b, c, f0, f1], where=[((CAST(f1 AS BIGINT) = a) AND (c = f0))])
+- Correlate(invocation=[func($cor0.c)], correlate=[table(func($cor0.c))], select=[a,b,c,f0,f1], joinType=[INNER], condition=[AND(=(blink_json_value($0, _UTF-16LE'$.id'), 2), =(+($1, 1), *($1, $1)))])
   +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])

经过优化后的执行计划如下所示

Calc(select=[a, b, c, f0, f1], where=[f00])
+- NativeCalc(select=[a, b, c, f0, f1, ((blink_json_value(f0, '$.id') = 2) AND (CAST(f1 AS BIGINT) = a) AND (c = f0)) AS f00])
   +- Correlate(invocation=[func($cor0.c)], correlate=[table(func($cor0.c))], select=[a,b,c,f0,f1], joinType=[INNER], condition=[=(+($1, 1), *($1, $1))])
      +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])

3）Native层

在native层，我们支持了18个simd函数，其中字符串函数15个，数学函数3个，也补齐了大量velox不支持的Flink内置函数。

4）细粒度Fallback机制

支持细粒度的fallback机制，全部做到可配置
- 仅含有SIMD函数的表达式才翻译成NativeCalc
- 支持细粒度的函数签名级黑名单机制
- 对于SQL timestamp/decimal类型fallback机制，对这种容易出现正确性问题的类型先fallback。

5）其他

支持配置化的函数映射机制，函数覆盖优先级机制

- 支持配置化的函数映射机制对于flink velox spark/presto函数语义一直但是函数名不一样，可以不用改动c++代码，修改配置即可，函数覆盖优先级机制持配置化的函数映射机制。新引入的函数或者修复velox函数bug无需加入velox，导致整个编译时间很久。在Flex内部加入即可，编译时间从2个小时到2分钟。

2. 正确性建设

这个是重中之重。两套引擎函数行为语义无法保证是对齐的，如何通过自动化手段发现二者行为上的差异？因此我们支持了函数级和作业级两套自动化比对框架。

如何复用Flink原有函数单元测试代码，而不需要改动和重写测试逻辑。然后当前Flink内置函数有两套机制oldstack/newstack，测试框架也不同。为此我们改造原有引擎2套测试框架逻辑，通过反射方式将这些单元测试自动注入向量化配置方式进行自动化比对。每天定时跑脚本将语义不一致的函数输出出来。通过这个工具，我们发现了Flink引擎本身函数正确性问题4个，都已经提给社区，Flink和Velox函数语义不对齐问题15个。

也支持了作业级端到端比对框架，本质就是上线前双跑比对。对重要作业mock两个作业消费固定的数据集输出到不同表，每天定时跑并输出比对结果报告到钉钉群。

3. 易用性建设

为了全方位提升易用性，我们在引擎层，尽最大程度将简单交给用户，复杂留给自己

如何提前发现用户作业中使用的函数Velox是否支持，可以翻译成Native算子？我们开发了一套自动化编译工具捞取线上作业自动化执行，将各种函数不支持问题问题提前解决掉，从而减少用户干扰。
如何提前知道开发的SIMD函数效果怎么样，是否有性能回退？因此我们基于JMH框架实现了一套端到端的性能测试框架，因为数据转换层有一定开销，因此需要对比整体性能更合理，目前支持GenericRow和ColumnRow。下图是函数使用SIMD实现端到端的性能效果数据。