PostgreSQL 技术内幕（十三）探究 MPP 数据库分布式查询分发 Dispatcher

Dispatcher（分布式查询分发器）是 MPP 数据库的核心组件，所有的查询任务都要经过其进行分发，起着沟通用户到协调者（Coordinator，即 QD）和执行调度的关键作用。

在这次的直播中，我们为大家介绍了 Dispatcher 基本原理和实现机制，并结合实际用例进行了操作演示。以下内容根据直播文字整理而成。

Slice 与 Gang 的基本概念与分类

传统 MPP 数据库采用无共享 Shared-Nothing 架构来存储数据，节点之间不共享存储和计算资源，需要使用其他节点的数据时通常利用网络重分发。

图 1：Greenplum 数据库查询示意图

（图片来自 Greenplum: A Hybrid Database for Transactional and Analytical Workloads，SIGMOD '21 ，序号和箭头系本文作者所加）

以 Greenplum 为例，如图 1 所示，当用户连接到 Coordinator（协调者节点）进行查询操作时，会通过 Dispatcher 组件将查询任务分配到不同的 Segment，各 Segment 之间通过 Interconnect 模块来传输数据。当各节点查询执行完成后，由 QD 节点对查询结果进行收集和整理，再回传给用户。

需要注意的是，在查询任务执行时，用户不会和 QE 产生任何的连接，所有消息都是通过 QD 来中转传递，这也是 MPP 数据库的重要特征。整个过程中，涉及到两个重要的概念：

Slice：为了在查询执行期间实现最大的并行度，Greenplum 会将查询计划工作划分为 Slices。Slice 是查询计划中可以独立进行处理的部分。查询计划会为 Motion 生成 Slice，Motion 的每一侧都有一个 Slice。正是由于 Motion 算子将查询计划分割为一个个 Slice，上一层 Slice 对应的进程会读取下一层各个 Slice 进程广播或重分布操作生成的数据，然后进行计算。

Gang：属于同一个 Slice 但是运行在不同的 Segment 上的进程，称之为 Gang。在 Greenplum 中，共有 Unallocated、Reader Gang、Writer Gang、Entry Reader、Singleton Reader 五种类型的 Gang。其中：

Unallocated 运行在 QD，一般只在 Gather Motion 将各个 QE 回传的结果收取并集时才会用到。

Reader Gang 和 Writer Gang 会经常用到，而且相关的查询计划会很复杂。

● 只读的查询仅包含 Reader Gang，包含写操作的查询才会使用 Writer Gang。

● 一些既读又写的查询（例如 Create table as、Update Returning 等）可能同时包含这两类 Gang。

● 这两类 Gang 都只有 1-Gang 和 N-Gang 的情况（其中 N 为 Segment 数量），优化器不会进一步划分出更细致的构造。

Entry Reader 和 Singleton Reader 不太常见，通常在处理子查询或者其它需要保证系统中只有一份值的情形下出现。

MPP 数据库分布式的特点，使其存储容量和计算能力突破了单机的上限，而成为多个 Segment 的总和。MPP 数据库的存储和计算紧耦合，对某一部分数据的计算应尽量在这一部分数据的存储节点上直接完成，这也使得所有表在创建时都有指定的分布模式，对该表的存储和计算都应按照该模式在集群的参与者节点上分布开来，从而形成“分布式”。

图 2：Greenplum 查询执行调度示意图

（来源同图 1）

许多 MPP 数据库的初学者，经常会遇到一个问题，Slice 和 Gang 是什么关系？

从二者的作用来看，Slice 是对 Plan 的自然切割，Gang 是对 Slice 的自然实现，两者之间是相互对应的。事实上，在内核代码中两者经常会交替使用。如果是优化器，会更多使用 Slice；而在执行器、Dispatcher 部分的代码就更多使用 Gang。我们可以认为，Slice 和 Gang 是完全对应的，前者侧重规划，后者侧重执行。

需要特别强调的是，Writer Gang ≠ Writer QE。事实上，QE 并不关心自己是不是 Writer Gang，它的行为只与 QD 连接时下发的 isWriter 参数相关，这个参数通过 libpq 协议握手时 Startup 消息中的 gpqeid 参数下发到 QE 上，QE 会据此设置 Gp_is_writer 全局变量，从而控制 QE 的行为。

在 Greenplum 中，任何查询都只有一个 Writer QE，而只有写查询才会有 Writer Gang。因此，我们时常听到的“任何查询都有一个 Writer Gang”的论断是不严谨的。

查询分发

查询分发通常分为以下四个阶段：

1.创建 Gang 并初始化链接；

2.分发查询计划并执行；

3.回收 Gang；

4.会话断开时，关闭链接并销毁 Gang。

在所有查询中，Plan 类查询最为复杂，它涵盖了其余三类查询的共用方法。接下来，我们以 Plan 类查询为例，展开介绍查询分发的实现流程。

💡Plan 是指 Select、Insert 这种可由 explain 指令查看执行计划的查询。这里有一个特例，CTAS（create table as）不属于 Plan，而是属于 Utility 类别。

Plan 类查询分发的入口点是cdbdisp_dispatchX函数，它会调用 AssignGangs 方法，通过两次先序遍历查询计划树，创建所有的 Writer Gang，然后再创建其余 Gang：

bool AssignWriterGangFirst(ExecSlice *slice, ...) [  if (slice->gangType == GANGTYPE_PRIMARY _WRITER) {    slice->primaryGang = AllocateGang(...);    return true;  } else {    ListCell *cell;    foreach(cell, slice-›children) {      if (AssignWriterGangFirst(cell->element, ...))        return true;    }  }  return false;}

AssignGangs 第一次遍历代码示例

bool InventorySliceTree(ExecSlice *slice, ...) {  if (slice->gangType == GANGTYPE_UNALLOCATED) {    slice->primaryGang = NULL;  } else if (!slice->primaryGang) {    slice->primaryGang = AllocateGang(...);  }
  ListCell *cell;  foreach(cell, slice->children) {    InventorySliceTree(cell->element, ...);  }}

AssignGangs 第二次遍历代码示例

注意： AssignWriterGangFirst 有一个“提前终止”的特性，表明查询计划只能有一个 Writer Gang。

我们看到这两段代码都是调用 AllocateGang 函数完成具体一个 Gang 的初始化，上方的代码只能调用一次，下方的代码可以调用多次。

创建并初始化 Gang

这里引入了另一个概念：Segment Type， 因为 AllocateGang 函数并不直接基于 Gang Type 去工作，需要将 Gang Type 转化为 Segment Type。二者之间对应关系如下：

● Writer Gang 一定要求一个 Writer QE，即一个 isWriter 为 true 的 Gang；

● extended query（libpq 规定的扩展查询协议，即游标 CURSOR）一定要求一个 Reader QE，即 isWriter 为 false 的 Gang；

● 其他情况（包括 Reader Gang 在内），适用任何节点。

通过以上的对应关系能够提供更好的兼容性，并进一步提高 Gang 进程的复用率。

下一步会调用：

cdbgang_createGang_asynccdbcomponent_allocateIdleQE

这两个函数。从名字可以看出， Gang 的创建是一个异步的过程，会一次创建一个 Gang 中的所有链接，然后通过轮询等待创建完成。在创建过程中，首先判断 freelist 中回收上来的 QE 的相容性，如果相容则直接使用；否则异步分配新的 Gang。

异步的过程如下所示：

// cdbgang_createGang_asyncint size = list_length(segments);for (i = 0; i < size; i++) {  segdbDesc = newGangDefinition->db_descriptors[i];  cdbconn_doConnectStart(segdbDesc, ...); // -> PQconnectStartParams -> PQconnectPoll
  connStatusDone[i] = false;  pollingStatus[i] = PGRES_POLLING_WRITING;}
for (;;) {  for (i = 0; i < size; i++) {    segdbDesc = newGangDefinition->db_descriptors[i];    if (connStatusDone[i]) continue;    switch (pollingStatus[i]) {      case PGRES_POLLING_OK:        ... // -> cdbconn_doConnectComplete        connStatusDone[i] = true;        continue;      case PGRES_POLLING_READING:         ... // set poll to wait till ready to read        break;      case PGRES_POLLING_WRITING:        ... // set poll to wait till write won't get blocked        break;      case PGRES_POLLING_FAILED: ... // throw exception    }  }    nready = poll(fds, nfds, poll_timeout);  if (nready > 0) {    for (i = 0; i < size; i++)      segdbDesc = newGangDefinition->db_descriptors[i];      pollingStatus[i] = PQconnectPoll(segdbDesc->conn);  }}

分发查询计划并执行

在这一阶段，以 Gang 为单位异步分发 M 类型消息（QD 到 QE 的消息为 M 类型消息，调用 exec_mpp_query 入口方法，而用户收到 QD 的消息则为 Q 类型消息，调用 exec_simple_query 入口方法）到所有 Gang，并再次通过轮询等待分发完成。

回收 Gang

回收 Gang 时，对每个 Gang 中的所有 QE 调用 cdbcomponent_recycleIdleQE 方法，将回收的 QE 放到 freelist 中。在下次分配时，会尽量使用已经创建好的链接。这种方法保证了 Writer QE 永远在 freelist 的前部，同时因为 Writer QE 具有良好的相容性，就使得 freelist 具备“有序”的性质，在分配 Gang（allocateIdleQE）时就无需遍历整个 freelist ，提升了 Gang 的分配效率。

销毁 Gang

销毁 Gang 时，通过 libpq 对每个 idle QE 进行终止握手，并释放状态。

其它类型查询

● Plan：到所有 QE;parsed statement

● Command：到 Writer QE;raw SQL

● Utility Statement：到 Writer QE;parsed statement

● Set Command：到所有 QE;raw SQL

掌握了 Plan 类型查询分发机制之后，就能更好地理解其余的三类查询，它们是 {到所有 QE，到 Writer QE} 和对 {parsed statement， raw SQL} 两两组合形成的四种结果。这里的 parsed statement 指经过查询解析和规划后的 PlannedStmt。

SharedSnapshot

Postgres 有本地快照机制，保证本地进程的一致性，而 Greenplum 作为一个分布式系统，必须通过分布式快照，来确保跨 Segment 的全局一致性。

本地快照保证进程本地一致性，分布式快照保证跨 Segment 一致性，但在 MPP 数据库中，一个 Segment 可能有多个进程，这就需要一个新的机制来保证跨进程的 Segment 本地一致性。

SharedSnapshot 作为基于共享内存的快照同步，成为沟通本地快照和分布式快照的一个桥梁。

Postgres 中的 SnapshotData 自带了进程本地一致性，在这基础之上通过 SharedSnapshot 实现跨进程的 Segment 本地一致性，最终通过分布式快照 DistributedSnapshot ，实现跨 Segment 的系统全局一致性，从而向用户提供全局一致的视图。