PolarDB-X 面向 HTAP 的混合执行器 一文详细说明了 PolarDB-X 执行器设计的初衷，其初衷一直是致力于为 PolarDB-X 注入并行计算的能力，兼顾 TP 和 AP 场景，逐渐为其打造一款具备 TB 级数据处理能力的数据库。为了做到这一点，借鉴了各种数据库和大数据库产品，包括分析型数据库，实时数仓等，吸取了各方面的优势来打造出一个全新的并行执行引擎。这里将对整个分布式并行执行框架做了详细的介绍，希望阅读之后对我们的执行器有一个全面的认识。

整体设计

PolarDB-X 是一个 Share Nothing 的数据库，采样了计算和存储分离的架构。其中数据以分片的形式存储于每个 DN 节点，计算节点叫 CN。在计算过程中，DN 和 CN 间、DN 和 DN、CN 和 CN 间是通过千兆、万兆网络通信。每个 CN 节点上一般都会有调度组件、资源管理组件、RPC 组件等。一个复杂的查询，会被调度到多个 CN 节点上执行，考虑到数据一般都会根据均匀策略分到各个 DN 上，所以每个 CN 节点同时会访问多个 DN。

当用户提交一条复杂的 SQL 语句，往往需要访问多个 DN 节点，这个时候就会启动并行调度策略，整个执行步骤可以简单理解：

1. 用户所连接的这个 CN 将承担查询协调者（Query Coordinator）的角色

2. Query 先发送给 Query Coordinator，会首先经过优化器生成最新的 Plan，然后会拆分到多个子计划(Fragment), 每个 Fragment 可能会包含多个执行算子。如果有的 Framgnt 负责扫描 DN 的话，它里头必定包含 Scan 算子，从 DN 拉取数据；Fragment 也可以包含 Agg 或者 Join 等其他算子

3. Query Coordinator 里头的调度器(Task Scheduler)会按照定义的调度逻辑将各个 Framgnts 封装成 Task，调度到合适的 CN 上执行，这里头可能会涉及到一些资源上的计算

4. 各个 CN 收到 Task 后，会申请执行资源，构造执行的上下文，开始启动 Task，并定期会向 Query Coordinator 汇报状态

5. 各个 Task 间会通过数据传输通道(DTL)交换数据，当所有的 Task 都执行完毕后，会将数据结果返回给 Query Coordinator，由它负责将结果返回给用户。

6. 成功返回给用户后，Query Coordinator 和被调度到的各个 CN 节点的 Task 会做清理动作，释放计算资源。

整个流程大致就这样，有细心的同学会发现我们的框架在 DN 上有一层叫 Split 概念。我们的数据是按分片存储在各个 DN 上的，Split 指的是数据分片 partition 的地址。对于包含扫描算子 Scan 的 Task，会计算出需要访问哪些 partition，这些 partition 分布在哪些 DN 上，然后封装成 splits 按比例划分给这些扫描 Task。但是实际运行过程中每个扫描 Task 并不是预分配好 splits 的，而是预分配部分 splits 给扫描 Task，看哪一个 Task 扫描的更快就会从 Query Coordinator 继续获取余下 splits，这样可以尽可能避免由于各个扫描 Task 资源不均衡导致的消费长尾现象。但是如果一个表只被分成了 2 个分片，是不是意味着扫描任务至多只能是 2，这可能起不到明显的并行加速效果。所以我们也支持在分片上继续按照分段做拆分，那么这个时候的 Split 除了会记录分片的地址，也会记录在分片上分段的位移。按照分段做拆分后，即便数据的分片数量有限，执行过程我们依然可以启动更多的扫描 Task，并行去加速扫描。

执行计划

执行引擎执行的是由优化器生成的分布式执行计划。执行计划由算子组成。因为 PolarDB-X 的数据按照分片存储到各个的 DN 节点上去，执行计划执行也会尽可能的满足数据分布的 locality，能下推的计划会被放到 DN 上执行，不能下推的计划会会被切分成一个个子计划(Fragment)，会被放到各个 CN 节点上执行。所以这里我们需要关心如何将一个从优化器出来的计划，拆分成分布式计划，放到各个 CN 上执行？

为了更好的理解这个过程，我们这里以一条简单 SQL: select * from (select useid, count(*)  as b from user_data group by userid) as T where T.b > 10  为例，经过优化器生成这样的相对最优计划

针对并行执行计划，为了更高效的执行尽量减少数据传输，可以把执行计划按照计算过程是否需要数据重分布(ReDistribution)分为不同片段（fragment）分布到相应节点执行，并且把一些操作下推来减少扫描输出的数据，上面的计划可能就变成这样的执行计划，由多个子片段构成。

不同片段之间通过 NetWork Write/Read 算子进行数据交换。更复杂的比如多表关联（join）查询，会有更多的片段和更复杂的数据交换模式。每个片段的并发度可以不同, 并发度是基于代价推导出来的。多机调度的基本单位是 Stage，Stage 记录了上下游片段的位置信息，以便上下游之间建立网络通道(DTL)。每个片段调度到计算 CN 节点后，会被封装成逻辑执行 Task，比如 fragment-1 并发度是 2 的话，那么会将 Task-1.0 和 Task-1.1 两个 Task 分别调度到两个 CN 节点。

Task 仍然是 CN 节点计算的逻辑单元，PolarDB-X 执行器不仅仅可以支持了单机并行能力(Parallel Query)，也可以做多机并行(MPP)。所以在 CN 节点还引入了二层调度逻辑, 当然二层调度的好处不仅仅于此，后面我们还会提到。这里会继续在 Task 内部根据算子间数据交换的特性，继续做切分，切分成不同 Pipeline。

不同的 Pipeline 并发度也可以不同，每个 Pipeline 会根据处理的数据规模大小会计算出不同的并发度，生成具体的执行单元 Driver，Driver 间会根据二层调度确定上下游的本地通道(Local Channel)。

至此你应该可以了解从执行逻辑计划转化为分布式物理执行的整个过程。引入了一些新的名称，这里统一做下梳理：

• Fragment: 指的逻辑执行计划按照计算过程中数据是否需要重分布，切割成的子计划;

• Stage: 是由 Fragment 封装而成的调度逻辑单位，Stage 除了封装 Fragment 外，还会记录上下游 Stage 间的调度位置信息；

• Task: Stage 并不会在 CN 上直接运行，他们是通过并发度分解成一系列可调度到 CN 上的 Task, Task 依然是逻辑执行单元；

• Pipeline: 对 CN 上的 Task 根据二层并发度做进一步切分，切分成不同的 Pipeline；

• Driver: 一个 Pipeline 包含多个 Driver，Driver 是具体的执行单元，是一系列可运行算子的集合。

一般来说针对一个复杂查询，一个 query 包含多个 Fragment，每个 Fragment 和 Stage 一一对应，每个 Stage 包含多个 Tasks，每个 Task 会切分成不同的 Pipeline，一个 Pipeline 包含了多个 Driver。只有理解上面说的 Fragment/Stage/Task/Pipeline/Driver 这些概念，你才能更清楚了解我们接下来的设计。

调度策略

并行计算在运行之初，需要解决任务调度问题。调度的直白理解，就是将切分好的 Task 调度到各个 CN 节点去执行，充分利用各个 CN 的计算资源。这里头大家很容易有这些疑问：

1. 执行过程中各个 CN 节点的计算资源是不均衡了，那么在多机调度中是如何将各个 Task 打散到不同 CN 节点去执行？

2. 和各个 DN 交互的 Task 是如何并行的拉数据的？比如某个逻辑表分成了 16 个物理表，分布在 4 个 DN 节点上，有 4 个 Driver 去并行拉数据，每个 Driver 并不是均匀拉取 4 个物理表，而是根据自身的消费能力来确定拉取的物理表数量；多个 Driver 下发扫描任务会不会同时恰好落地一个 DN 节点上，导致某个 DN 成为瓶颈？

3. 我们完全可以在一个 CN 节点，同时调度多个 Task 执行，已经可以做到单机并行，为什么还要二层调度？

一层调度(多节点间)

为了解决(1) 和 (2) 的问题，我们在 CN 节点内部引入了调度模块(Task Scheduler)，主要负责 Task 在不同 CN 节点上的调度，这一层调度我们这里称之为一层调度，在这层调度中，同属于一个 Stage 的多个 Task 一定会被调度到不同 CN 节点上去，确保一个 CN 节点只能有相同 tage 的一个 Task。调度过程中通过心跳不断维护 Task 状态机，也维护着集群各个 CN 节点 Load 信息，整个调度是基于 CN Load 做调度的。多机调度流程如下所示

Resource Manager(RM)是 CN 节点上个一个资源管理模块，RM 会借助 Task 心跳机制实时维护集群各个 CN 节点的负载，Task Scheduler 组件会基于负载选择合适的 CN 节点下发执行任务，比如 CN-1 负载相对集群其他 CN 节点来说高很多，那么当前查询的 Task 会分发给其他 CN 节点，避免调度到 CN-1 节点去。执行器在执行 Task 任务时，Task 并不是创建好的时候就确定了其消费 DN splits 的映射关系。各个 Task 按批次动态拉取 splits 进行消费, 直白理解就是谁的消费能力越强谁就有可能消费更多的 splits。同样为了解决同一个时刻多个任务同时消费同一个 DN 上的 splits 问题，我们在调度之初会将 splits 根据地址信息按照 Zig-zag 方式，把各个 DN 上的 splits 打散到整个 splits queue 上去，消费的时候可以尽可能分摊各个 DN 压力，这样计算过程中也会充分利用各个 DN 的资源。

有了一层调度后，我们也可以将同属于一个 Stage 的多个 Task 调度到同一个 CN，这样其实也可以做到单机并行。如果这样设计的话，我们容易忽略两个问题:

• 一层调度的逻辑比较复杂，需要多次交互，一个 CN 内部需要同时维护各个 Task 的状态，代价会比较大，这在 TP 场景是无法容忍的；

• 一层调度中，并发度越高，生成 Task 就越多，这些 Task 间需要建立更多的网络传输通道；

二层调度(节点内部)

为了解决上述一层调度的不足，为此我们在参考 Hyper 的论文[^1]，引入了二层调度，既在 CN 节点内部单独做单机并行调度，简单来说我们会在 Task 内部借助 CN 的本地调度组件(Local Scheduler)，对 Task 做进一步的并行调度，让 Task 在 CN 上执行，也可以做到并行运行。下图中，Stage-1 和 Stage-2 是上下游关系，各自并发度都是 9，调度到 3 个 CN 节点执行。如果只有一层并发度的话，每个 CN 节点还会调度运行 3 个 Task，那么上下游之间总共会建立 81 个 Channel，CN 节点内部 Task 是相互独立的，这样缺点还是很明显：

1. 多个 Channel，放大了网络开销，同一份 buffer 会被发送多次，发送和接收对 CPU 和 Memory 都有代价

2. 数据发送的对象是 Task，数据本身有倾斜，会导致同节点内 Task 之间的负载不均衡（hash skew），存在长尾问题

而一层调度和二层调度相结合的话，Stage-1 和 Stage-2 的一层并发度是 3，这样每个 CN 节点只会有 1 个 Task，Task 内部并发度 3。由于 shuffle 的对象是 Task，所以 Stage-1 和 Stage-2 间只会建立 9 个 Channel，大大减少了网络开销，同时 Task 内部的 3 个 Driver 内数据是共享的，Task 内部的所有的 Driver 可以共同消费接受到的数据，并行执行，避免长尾问题。针对于 HashJoin，假设 Ta 为大表，Tb 为小表，这两个表做 HashJoin，可以让 Ta 和 Tb 同时 shuffle 到同一个节点做计算；也可以让小表 Tb 广播到 Ta 所在节点做计算，前者的网络代价是 Ta+Tb，而后者的代价是 N*Tb(N 代表广播的份数)。所以如果只有一层调度的话，N 可能比较大，执行过程中我们可能会选择两端做 shuffle 的执行计划；而一层和二层相结合的调度策略，可以让执行过程中选择 BroadcastHashJoin,这样可以避免大表做 shuffle，提高执行效率。

此外在二层调度策略中，task 内部的多线程很容易做到数据共享，有利于更高效的算法。如下图，同样是 HashJoin 过程中，build 端的 Task 内部多个线程（driver）协同计算：build 端收到 shuffle 的数据后，多线程协同建立一个共享的 hash 表。这样一个 task 只有一个 build table，probe 端收到 shuffle 数据后，也不用做 ReDistribution 了，直接读取接受到数据，进行并行的 probe。

并行执行

聊完调度，接下来应该是关心任务是如何在 CN 上运行，运行过程中遇到异常我们系统是如何处理的呢？

线程模型

说到执行，有经验的同学可能会发现我们的调度并没有解决调度死锁问题，比如对于下面这样一个执行计划，两表 Join。一般会遇到两种问题：

1. 如果先调度 f3 和 f2 的话，这个时候假设集群没有调度资源，则 f1 不能迟迟调度起来。而 HashJoin 的逻辑就是需要先构建 buildTable，这里 f1 刚好是 build table 部分。最终会导致执行死锁: f1 在等待 f3 和 f2 的计算资源释放，而 f2 和 f3 又在等待 f1 构建完 buildTable。

2. 如果 f1 先调度起来了，假设这个时候 f2 和 f3 没有调度资源，这个时候 f1 从 DN 拉出来的数据，其实是无法发送给 f3 的，因为 f3 还没有被调度起来

解决问题 1，业界有很多方式，比较常见是在调度之初构建调度依赖关系(Scheduler Depedency): f1->f3-f2。而解决问题 2，往往是将 f1 把 DN 拉出来的数据先放到内存中，实在放不下就落盘处理。可见处理上述两个问题，执行框架不仅仅需要在多机调度上做复杂的调度依赖关系，同时还需要考虑对落盘的支持。而其实我们在调度的时候，并没有去考虑调度依赖这个事情，我们是一次性把 f1/f2/f3 全部调度起来了，这个是为何呢？这就要说下我们执行中的逻辑线程模型概念。在大多数计算引擎中，一个查询首先会通过资源调度节点，在各个 CN 上申请执行线程和内存，申请成功后，这些执行资源会被调度组件占用，用来分配当前查询的 Task，不可以再被其他查询所利用，这种是真实的执行资源，和调度资源相互绑定，当 CN 上可利用的执行资源不够的时候，才会出现调度死锁问题。而在 PolarDB-X 中，我们并没有在调度的时候申请真实的线程资源，调度只需要考虑各个 CN 的负载，不需要考虑各个 CN 到底还剩多少可利用的真实资源。我们的线程模型也并没有和调度资源绑死，每个 Driver 其实不独占一个真实的线程，换句话说，真实的线程也并没有和调度资源一一对应。虽然说 Driver 是执行的基本单元，但是在调度上来看，它又是逻辑的线程模型而已。那是不是意味着只要有调度任务，都可以被成功调度到 CN 上去，答案是肯定的。一次性调度所有的执行单元到 CN 上去执行，对内存和 CPU 也是一种开销。比如 f2 被执行起来后，但是 f1 并没有执行完毕，那么 f2 也会不断执行，其数据其实也会被缓存起来，但是也不能无限缓存数据呀？为了解决这个问题，接下来就需要借助我们的时间片执行了。

时间片执行

我们在每个 CN 节点内部会有一组执行线程池来运行这些 Driver，每个 Driver 会排队进入线程池参与计算，如果 Driver 被阻塞就会退出到 Blocking 队列中，等待被唤醒。比如 f2 driver 启动后，从 DN 拉了数据放到有限空间 buffer 里头去，这个时候假设 f1 driver 都没有结束，那么 f2 driver 对应的 buffer 就会满，满了后就会阻塞住，一旦阻塞我们的执行框架就会让 f2 driver 从执行器退出来，加入到 Blocking 队列中，简单的说就是将计算资源滕让出来，等待被唤醒。直到 f1 driver 都执行完毕后， f2 driver 会被唤醒，执行框架就会将他移动到 Pending 队列中，等待被调度到执行线程池中继续运行。这里头还是会浪费点内存，但相对于 CPU 资源来说，内存资源还是比较充裕的。

时间片执行的核心就是需要判断 Driver 何时会被 Block 的，总结起来被阻塞的原因一般分为三种情况:

• 根据算子依赖模型来确定，比如图中 f1 driver 未执行完毕，那么 f2 driver 其实也会被阻塞(这个是一个可配置的选项)；

• 计算资源不足(主要指内存)，对应的 driver 会被挂起，等待资源释放；

• 等待 DN 响应，物理 SQL 下发给 DN 后，Driver 会被挂起，等待物理 SQL 执行完毕；

除此之外我们在借鉴 Linux 时间片调度机制，会在软件层面上统计 Driver 的运行时长，超过阈值(500ms),也会被强制退出执行线程，加入到 Pending 队列，等待下一轮的执行调度。这种软件层面上的时间片调度模型，可以解决复杂查询长时间占用计算资源问题。其实实现起来也挺简单的，就是每计算完一个批数据后，我们会对 driver 的运行时长进行统计，超过阈值，就退出线程池。下面贴出了 Driver 处理逻辑的部分伪代码，Driver 在执行采用的是经典的 Producer-Consumer 模型，每消费一个 Chunk 我们就会累计时间，当超过既定阈值，就会退出来。

任务状态机

高并发系统，频繁地等待或者任务切换是常见的系统瓶颈。异步处理是一种已经被证明行之有效的避免这些瓶颈，把高并发系统性能推到极致的方法。所以 PolarDB-X 执行器的整个后端，统一使用全异步的执行框架；同时 MPP 执行过程涉及到多机的协调，所以这就要求我们在系统内部维护这些异步状态。异步状态的维护特别重要，比如某个查询下的 Task 执行失败，需要立即通知到整个集群中该查询正在运行的 Task 任务，以便立即中止，以防出现 Suspend Task，造成资源不释放问题。

所以在执行器内部，我们从三个维度(Task Stage Query)去维护状态, 这三种 State 是相互依赖耦合的，比如 Query 被 Cancel，会立即通知其所有的 Stage，各个 Stage 监听到状态变化，会及时通知给其所有的 Task，只有等待 Task 都被 Cancel 后，Stage 最后的状态才变更为 Cancel，最终 Query 的状态才被标记为 Cancel。在这个过程中我们会引入对状态机异步监听机制，一旦状态发送变更就会异步回调相关处理逻辑。通过维护这些状态，我们也可以及时通过查询或者监控诊断到任务是否异常，异常发生在哪个环节，也便于我们后期排查问题。

资源隔离

如果并发请求过多的时候，资源紧张会让请求线程参与排队。但是正在运行的线程，需要耗费比较多的计算资源(CPU 和 Memory)的时候，会严重影响到其他正常正在运行的 Driver。这对我们这种面向 HTAP 场景的执行器是决定不被允许的。所以在资源隔离这一块，我们会针对不同 WorkLoad 做计算资源隔离，但这种隔离是抢占式的。

CPU

在 CPU 层面上我们是基于 CGroup 做资源隔离的，根据 WorkLoad 不同我们把 CPU 资源分为 AP Group 和 TP Group 两组，其中对 TP Group 的 CPU 资源不限制；而对 AP Group 是基于 CGroup 做硬隔离，其 CPU 使用水位的最小阈值(cpu.min.cfsquota)和最大阈值(cpu.max.cfsquota)来做限制。执行线程分为三组: TP Core Pool 、AP Core Pool、SlowQuery AP Core Pool，其中后两者会被划分到 AP Croup 一组，做严格的 CPU 限制。Driver 会根据 WorkLoad 划分到不同的 Pool 执行。看似很美的实现，这里头依然存在两个问题:

1. 基于 COST 识别的 WorkLoad 不准怎么办？

2. AP 查询比较耗资源，在同一个 Group 下的多个慢查询相互影响怎么办？

出现问题(1)主要的场景是我们把 AP 类型的查询识别成了 TP，结果会导致 AP 影响到 TP，这是不可以接受的。所以我们在执行过程中会监视 TP Driver 的执行时长，超过一定阈值后扔没有结束的查询，会主动退出时间片，然后将其它调度到 AP Core Pool 执行。而为了解决问题(2), 我们会将 AP Core Pool 中长时间运行都未结束的 Driver，进一步做优雅降级，调度到 SlowQuery AP Core Pool 执行。其中 SlowQuery AP Core Pool 会设置执行权重，尽可能降低其执行 Driver 的频率。

MEMORY

在内存层面上，会将 CN 节点堆内内存区域大致可以分为四大块:

• TP Memory：用于存放 TP 计算过程中的临时数据

• AP Memory：用于存放 TP 计算过程中的临时数据

• Other: 存放数据结构、临时对象和元数据等

• System Reserverd: 系统保留内存

TP 和 AP Memory 分别会有最大阈值和最小阈值限制，两者内存使用过程中可以相互抢占，但是基本原则是：TP Memory 可以抢占 AP Memory，直到查询结束才释放；而 AP Memory 可以抢占内存 TP，但是一旦 TP 需要内存的时候，AP Memory 需要立即释放内存，释放方式可以是自杀或者落盘。

数据传输层(DTL)

并行计算是充分利用各个 CN 资源参与计算，那么 DN 与 DN 之间必然会存在数据交互。各个 DN 上的上下游的 Task 数据需要传输，比如上游的 Task 数量 N，下游的 Task 数量是 M，那么他们之间的数据传输通道需要用到 M*N 个通道(Channel)，同样的我们将这些通道（Channel）的概念抽象成数据传输层。这个传输层的设计往往也会面临两个问题:

1. 通道分为发送端和接受端，当发送端源源不断发送数据，而接受端无法处理的话就会造成内存雪崩；

2. 数据在传输过程中丢失；

在业界实现数据传输主要有两种传输方式:Push 和 Pull。Push 就是发送端往接受端推送数据，这里头为了避免接收端处理不过来，需要引入流控逻辑，一般的做法都是在接收端预留了槽位，当槽位被数据占满时会通知发送端暂停发送数据，当有接收端数据被消费空闲槽位出现时通知发送端继续发送，这里头会涉及到发送端和接收端的多次交互，流控机制相对比较复杂。Pull 就是发送端将数据先发送到 buffer 里头去，接收端按需从发送端的的 buffer 拉数据，而当发送端发送的数据到 buffer，接收端假设长时间不来拉数据，最后发送端 buffer 满了，也会触发上游反压，为了避免频繁反压，往往发送端的 buffer 不应该设置太小。综合起来我们选择了 pull 方式来做。采样 pull 方式也会遇到两个问题

1. 每个 receiver 一般会和上游多个 sender 建立连接，那么每次都是通过广播的方式从上游所有的 sender 拉数据吗？

2. 一次从 sender 端到底请求多少的数据呢，即 averageBytesPerRequest?

我们先回答问题(2)，我们这里会记录上一次请求的数据量 lastAverageBytesPerRequest、当前建连通道个数 n 以及上一次总共返回的数据量 responseBytes，来计算出当前 averageBytesPerRequest，具体的公式下面也给出了。至于问题(1),有了当前的 averageBytesPerRequest 后，结合目前 receiver 上 buffer 剩余空间，可以估算出这一次需要向上游几个 sender 发送请求。

$$averageBytesPerRequest = lastAverageBytesPerRequest* (n-1)/n + responseBytes/ n$$

在异步通信过程中为了保证传输可靠性，我们采用了类似 tcp ack 的方式，当 receiver 端带着 token 去上游拉数据的时候，则表示当前 token 之前的数据均已经被 receiver 端消费完毕，sender 可以释放这些数据，然后将下一批数据以及 nextToken 返回给 receiver 端。

效果展示

前后说了很多干货，下面咱们来点简单实际的东西。这里以 TPCH Q13 为例来演示下执行器在不同场景下的加速效果，为了方便截图在 Q13 后面都加了 limit。该测试环环境下，CN 和 DN 规格都是 2*16C64G。

单机单线程下运行，耗时 3min31s

使用 Parallel Query 加速，既单机多线程执行，耗时 23.3s

使用 MPP 加速，既同时利用两个 CN 节点的资源计算，耗时 11.4s

总结

不管是简单查询，还是 Parallel Query 和 MPP 场景下的复杂查询，共用的都是一套执行框架。不同场景下对执行器的要求，更多的是并发度设置和调度策略的差异。相对于业界其他产品来说，PolarDB-X 执行器主要特点：

1. 在资源模式上使用的是轻量化的资源管理,不像大数据计算引擎，需要额外引入的资源管理的节点，做严格的资源预分配，主要考虑到我们的场景是针对于小集群的在线计算；

2. 在调度模型上执行器支持 DAG 调度，相对于 MPP 调度可以做到更加灵活的并发控制模型，各个 Stage 间、Pipeline 间的并发可以不一样；

3. 区别与其他产品，AP 加速引用的是外挂并行计算引擎，PolarDB-X 并行执行器是内置的，不同查询间共用一套执行模型，确保 TP 和 AP 享有一致的 SQL 兼容性;

PolarDB-X 并行计算在线上已经平稳运行了近两年，这两年来我们不仅仅在执行框架上做了很多稳定性工作，在算子层的优化我们也沉淀了不少的技术。但这些还不够，目前比较热的是自适应执行，结合 Pipeline 模式的自适应执行挑战比较大，我们近期也在研究，欢迎感兴趣的朋友来拍拍砖，一起进步！

Reference

[1] V. Leis, et al., Morsel-Driven Parallelism: A NUMA-Aware Query Evaluation Framework for the Many-Core Age, in SIGMOD, 2014.

[2] Presto: SQL on Everything.

[3] A Deep Dive into Query Execution Engine of Spark SQL.

[4] Impala: A Modern, Open-Source SQL Engine for Hadoop

[5] FusionInsight LibrA: Huawei's Enterprise Cloud Data Analytics Platform. Proc. VLDB Endow. 11(12): 1822-1834 (2018)