在“国产数据库硬核技术沙龙-TDSQL-A 技术揭秘”系列分享中，5 位腾讯云技术大咖分别从整体技术架构、列式存储及相关执行优化、集群数据交互总线、分布式执行框架设计及优化策略、以及向量化执行引擎等多方面对 TDSQL-A 进行了深入解读。

本期带来了系列分享中腾讯云数据库高级工程师张倩老师主题为“TDSQL-A 分布式执行框架设计及优化策略”的分享的文字版。

作为领先的分析型数据库，TDSQL-A 是腾讯首款分布式分析型数据库，采用全并行无共享架构，具有自研列式存储引擎，支持行列混合存储，适应于海量 OLAP 关联分析查询场景。它能够支持 2000 台物理服务器以上的集群规模，存储容量能达到单数据库实例百 P 级。

一、执行框架总体设计

1.1 TDSQL-A 架构

首先介绍 TDSQL-A 的总体架构，包括上层的协调节点 CN、GTM 事务管理器、中间的数据交互总线 FN、以及下方的数据节点 DN。主要介绍的是协调节点 CN 和数据节点 DN 的相关内容，包括用户的查询怎么在 CN 和 DN 上执行、最后如何返回结果给用户等问题。

TDSQL-A 采用 MPP 架构，其特性是 share-nothing，数据分散在多个 DN 上，按照不同的分布键分布，并且不同的表可以自定义不同的分布键。如果 CN 收到了一条查询，它会将这个任务分散到多个 DN 上并行执行，从而提高执行效率，最后 CN 获得 DN 并行执行的最后结果，汇总之后再返回给客户端。

1.2 原分布式执行框架

这里先说明一下我们的原分布式执行框架一个最主要的问题。下图是一个简单的 Join 查询，如果 Join 查询正好是在这个表的分布键上进行 Join，则不涉及数据的重分布，可以直接在每个 DN 节点上进行 Join，DN 的结果汇总起来就是最终的查询结果，这是最理想的情况。

但客户的查询往往比较复杂多样，Join 经常会涉及不同节点之间的数据交换，Join 的两个表的 Join 键不一定是一个表的分布键，这种情况下就会涉及到数据的重分布。在 TDSQL-A 中，数据重分布是由 Remote Subplan 算子来执行。在执行的时候，Remote Subplan 算子会并行地创建对应下层的执行进程和对应的 DN 连接，每个 DN 都会创建对应其他 DN 的各个链接，这就会导致链接数和进程数急剧膨胀，给服务器造成很大的压力。

1.3 TDSQL-A 分布式执行框架

针对原分布式架构的缺点，我们设计了一套全新的分布式执行框架。在这种执行框架下，查询执行前 CN 会对查询计划进行分片，并创建 DN 上的各个执行进程，每个 DN 的进程间不需要再建立冗余的进程及连接。这可以减少不必要的进程和连接，减轻服务器的负担，并且能够做到比较好的线性扩展性。数据交互则是通过中间的 router——FN 节点来进行数据交换，这是当前 TDSQL-A 的分布式执行框架。

二、查询计划分片策略

2.1 查询计划分片过程

之所以要对查询计划进行分片，主要是因为一个分布式的查询计划，在绝大多数情况下，必然会包含数据的重分布。在我们的执行框架中，根据数据重分布进行查询计划的划分。

首先包括数据重分布的代价在内，优化器会生成一个代价估算最优的执行计划。在这个执行计划上，我们会做计划树的划分分片——把每一个数据重分布的节点下面的子数作为一个计划的分片，再通过 FID 来对每一个计划分片进行管理。

以下图为例，假设有一个两层的 Hash Join，每一层涉及到一些对应的数据重分布，就会有一个四分片的查询的产生。

2.2 Agg 算子执行计划

在分布式数据库里面，对其他的算子，也会生成一个分布式的执行计划，比如 OLAP 场景里面经常使用的执行聚合计算的 Agg 算子。在聚合计算中，比如 group id 正好是表的分布键的情况下，可以生成单独的分片，就像下图中 FID 1 这样的分片。每个 Agg 操作都是在 DN 本地执行，最后汇总到 CN 上得到一个最终结果。但是在有些情况下，比如聚合键不是分布键的情况下，就会在最下层的节点上做部分的聚合操作，在上层的节点经过数据重分布之后再做最后的聚合操作，得到最终结果。这就是一个分布式的 Agg 算子的执行计划。

2.3 Sort 算子 &Limit 算子执行计划

Sort 算子还有 Limit 算子也是同样的逻辑。

对于 Sort 算子，我们会在 DN 本地先做一次排序，经过数据重分布后，在上层节点再进行归并，最后得到最终的排序结果。

对于 Limit 算子，我们会把它进行下推。比如说下面这个例子中，这条搜索语句是查询前 100 名的 test order，这样的话我们会把 Limit 算子进行下推，每个 DN 只返回 Limit 100 条数据给上层节点，上层节点在收到结果之后再进行合并排序，最后取 Limit 100 的结果作为最终结果返回给上层。

三、异步执行流程控制

3.1 异步执行具体流程

在生成查询计划的分片后，CN 会下发每个分片对应的执行计划片段，分别发送给各个 DN，然后每个分片在每个执行节点上会创建一个进程，执行对应的执行计划。不同层级的进程异步启动执行，通过 FN 进行数据交互。

下图中可以看到，这里有两个查询，分别是简单的 Join 查询，以及数据重分布的 Join 查询。如果是传统的数据库执行流程，就会先启动下层节点，再启动上层节点。但在我们设计的这种执行框架下，FID 1 和 FID 2 是同步启动的，它们之间通过 FN 来进行数据交互。

如果在有两个数据节点的情况下，Join 查询怎么启动执行进程呢？因为有两个分片，还有两个数据节点，所以在执行的过程中，有四个进程在同时执行。

最下面的这两个分片，都属于 FID 2，但分别在 DN 1 和 DN 2 上执行，执行对应的计划分片。对其中一个表进行扫描，再通过 FN 节点进行数据交换。上面的这两个分片都属于 FID 1，分别在 DN 1 和 DN 2 上执行，它们分别获取自己所需要的数据，同时执行自己的执行计划分片。最终，两个 FID 1 的执行进程会把最终结果发送给 CN。这四个进程是同步执行的，在数据交换的时候通过 FN 来进行。

3.2 自适应流程控制

TDSQL-A 执行框架最大的难点就在于进程间如何进行协调和控制。针对这个问题，我们设计了一个具有自适应特点的异步执行的流程控制机制。它主要有以下三个方面的特点：

●灵活控制执行进度。根据实际执行情况，DN 动态地控制各个进程之间的执行进度。

●根据前端设置按需执行，优化资源利用，快速响应异常。比如前端发送 Cancel 请求时，能够及时响应处理。如果任何执行进程发生异常，也能够快速响应处理。

●保证分布式事务一致性。涉及修改操作的分片会开启事务，并且同步执行这个事务的提交或者回滚等操作。

下面我将分别从这三个方面来介绍一下这个异步执行流程控制机制。

在各个进程同步执行的情况下，如果有的进程出现执行阻塞的情况，该怎样互相协调呢？

以下图为例，假设上层节点中的 FID 1 的这两个执行进程执行比较慢，而下层 FID 2 的这两个进程执行进度比较快的时候，下层 FID 2 的两个进程会源源不断地向上层发送它们的执行结果。如果不加控制的话，不仅会浪费下层 FID 2 的执行资源，而且会造成网络的阻塞。

针对这种情况，我们设计了进程间可以互相协调执行进度的控制机制，主要通过数据流控制来实现。如果上层节点的执行进度慢于预期的时候，下层节点会进行等待，等到上层节点能够继续执行时，下层节点才会继续做自己计划分片的执行，把数据发送给上层节点。这样可以在执行节点上达到资源分配和使用较优的效果，空出来的网络资源和 CPU/IO 资源就可以让渡给其他查询来执行。

我们的控制机制中除了数据流之外，还有控制流。由 CN 来监听并统一处理控制流消息。DN 节点的执行进程，又叫 Dprocess，在执行的过程中会随时响应控制消息。以下图为例，如果用户执行一个比较长的进程或者误执行了一个 Query，在执行几分钟后，不想再执行了，就会给 CN 发送一个 Cancel 信号取消查询，这时 CN 会把这个信号通过链接发送给每个执行进程，DN 上的执行进程收到信号后就会终止执行，及时把资源让渡出来给其他的查询使用。这是 Cancel 消息的处理过程。

除了 Cancel 消息外，我们还处理 Error 信息。在执行进程同步执行的过程中，每个执行进程之间通过 FN 来进行数据交换。如果其中一个进程发生 Error，比如在处理的过程中资源不足，或者在处理过程中遇到数据错误或其他错误等，这时它会报 Error 信号，通过链接将这个信号上报给 CN。CN 在收到执行进程 Error 消息后，会进行消息处理，然后下发给其他的执行进程，让它们终止执行。也就是说，如果任何一个并行执行的进程发生了错误，我们也能够及时取消、结束这个查询。

3.3 执行流程示例

下图是一个总体执行流程的示例。左侧是一个带有数据重分布的 Join 查询，它的整体执行流程可以用右边的这个图来表示。四个执行进程之间会有数据交换，是通过 FN 来交换数据流，最终结果也是通过 FN 数据流返还给 CN，CN 上还有一个后台线程，通过控制流控制各个执行进程之间的执行，这就是整体的执行构架。

除了查询语句外，我们还会遇到 DML 语句。DML 语句即 Insert、Update、Delete 语句，它们需要进行分布式执行事务的提交或者回滚操作。在执行过程中，我们主要是把修改操作集中在一个分片内，然后在执行修改操作的这个分片内进行事务的开启、提交和回滚等操作。这个事务的命令同样也是通过控制线程来进行发送，其他线程也同样是通过 Cancel 或者上报 Error 来处理控制消息。

这里举一个最典型的例子。执行 Insert into 语句时，如果后面跟的是 Select From，也就是在其他的表中经过查询操作获得一个结果集，把这个结果集插入到一个表中，此时我们在其他分片上执行只读操作，只在第一个包含 Insert 的分片上执行修改操作，这个修改操作就涉及事务的提交和回滚。

3.4 中止处理流程

这里重点介绍中止处理流程，它和 Cancel 流程不一样。中止处理流程是 CN 在获取了部分的查询结果集后中止执行。典型的应用场景是把查询结果做分页展示。在很多前端的应用中，查询结果就是用分页展示的形式展现在客户端页面上的。

比如一个查询，第一页可能有 1000 条查询结果，下一页则是下 1000 条查询结果。CN 在查询执行的时候，只要执行获取到 1000 条结果，就可以返回给前端，让前端做展示或者处理。因为前端程序处理查询结果也需要时间，在这时，后端就可以继续执行获取下 1000 条查询结果，这样就能实现前端和后端并行执行，取得执行效率整体最优化。

在这种执行流程下，CN 会先获取前 1000 条结果——该数值用户可以自由设置，在获取到指定结果集之后 CN 先返回给前端，前端处理完之后，如果需要再获取，CN 就会继续返回下一批结果。

如果前端查询取消，比如用户可能看 5 页或者 6 页之后不想再看，或者是前端应用处理到第几批数据之后不再处理直接返回，在这样的情况下，查询其实不需要再继续执行，这时 CN 会下发一个 End query 信号，然后在并行执行流程上也会及时响应这个信号来结束查询。

左下角是分页场景下的执行性能对比。最左边的这个柱子显示的是，如果这个查询在正常执行情况下，在返回第一条结果的时候所需要的时间，第二个柱子是如果设置了 fetch size 是 1000 条时，它所需要执行的时间。如果没有设置 fetch size，在传统的执行方式下，这个查询的执行时间是非常长的，但如果我们先设置返回 1000 条结果，这个查询时间可以大幅缩小。同时在继续执行的时候，后续的每一批的查询结果的执行时间几乎可以忽略不计。因为前端在接受查询时，我们后端也在同时处理继续获取查询结果。

四、子查询执行优化

在 OLAP 场景下一些比较典型的包含有子查询的执行优化。OLAP 子查询基本上可以分为两类：一类是非相关的子查询，一类是相关的子查询。

4.1 非相关子查询执行

非相关的子查询，指的是子查询的结果集是一个固定的值，跟外层的查询没有关联。对非相关子查询，我们设计了“异步执行、一次执行”的机制。子查询对我们的执行框架来说，是另外的一个分片，它跟父查询可以并行执行。当父查询需要子查询的结果时，子查询已经执行完毕了，父查询可以直接获取结果继续执行。

下图中，FID 3 分片就是代表子查询的执行分片。Hash Join 在执行过程中，每个分片都是并行执行的，在 FID 2 做扫描的时候，如果它不需要子查询的结果，就可以不用等待 FID 3 的执行结果。当它需要子查询的执行结果时，因为 FID 3 和 FID 2 是并行执行，就可以直接获取到这个结果并使用。这是非相关子查询的执行。

4.2 相关子查询执行

更为复杂的是相关子查询的执行。在执行过程中，相关子查询的执行结果是跟父查询的传递条件是有关系的。

在 order 1 和 order 2 的 pid 是相等的情况下，查询会从 order 2 这个表中取出最大的 tax 值。这个 tax 的值再和外层的 order 1 的 tax 值做等值比较，最后获取等值比较成立的那个结果，作为最终的查询结果。

相关子查询的执行，一般情况是由父分片传递参数到子分片上，子分片会设置这个参数值，然后返回查询结果。比如 FID 2 先做一个 scan 的操作，它在要获取子查询的值时，会先把 order 1 的 pid 先通过 fragment 之间的连接传递给 FID 3，FID 3 在取得并设置了 order 1 的 pid 值后，执行它本身的执行计划，最后获取的结果再传递给 FID 2，然后 FID 2 获取结果后再继续进行计算，可以看到这是一个非并行的执行。之所以这样，主要是因为子查询 FID 3 的每一条执行结果其实是和 FID 2 下发的参数值是有关的。因此它们俩不能并行执行，这样的子查询执行效率就比较低。

针对这种情况，我们做了相关子查询的优化，会在计划生成阶段由优化器自动改写查询计划。在很多应用中，查询语句可能是由前端应用自动生成的，并且数量很大，如果都用人工来进行优化改写，工作量会非常大。在这种情况下，我们在优化器中实现了一套基于代数变换规则的自动改写，会把相关子查询，根据一定的规则改写成等价的 Join 查询，之后再进行其他优化，生成最后的查询计划。

经过优化后，相关子查询的性能提升非常明显。下面这个图就是在子查询改写之后，它的优化性能对比。可以看到，如果按照原来的执行方式，每个子查询每一次设置参数之后都需要执行一次，整个查询的执行时间非常长。如果改写成等价的 Join 查询之后，它的执行效率非常高。