Greenplum: 基于PostgreSQL的分布式数据库内核揭秘(下篇)
http://www.postgres.cn/v2/news/viewone/1/454
原作者：姚延栋 创作时间：2019-05-08 17:25:25+08 采编：wangliyun
发布时间：2019-05-09 08:25:28
欢迎大家踊跃投稿，投稿信箱：[email protected]
评论：0 浏览：1620
作者介绍
姚延栋，山东大学本科，中科院软件所研究生。PostgreSQL中文社区委员，致力于Greenplum/PostgreSQL开源数据库产品、社区和生态的发展。
上篇（链接地址：https://mp.weixin.qq.com/s/DI4U8UoddOHBRiJPzwfr-Q）介绍了集群概述、分布式数据存储和分布式查询优化。本篇继续介绍分布式查询执行、分布式事务、数据洗牌和集群管理等方面。
一、分布式执行器
现在有了分布式数据存储机制，也生成了分布式查询计划，下一步是如何在集群里执行分布式计划，最终返回结果给用户。
Greenplum 执行器相关概念
先看一个 SQL 例子及其计划：
test=# CREATE TABLE students (id int, name text) DISTRIBUTED BY (id);
test=# CREATE TABLE classes(id int, classname text, student_id int) DISTRIBUTED BY (id);
test=# INSERT INTO students VALUES (1, 'steven'), (2, 'changchang'), (3, 'guoguo');
test=# INSERT INTO classes VALUES (1, 'math', 1), (2, 'math', 2), (3, 'physics', 3);
test=# explain SELECT s.name student_name, c.classname
test-# FROM students s, classes c
test-# WHERE s.id=c.student_id;
QUERY PLAN
-----------------------------------------------------------------------------------------------
Gather Motion 2:1 (slice2; segments: 2) (cost=2.07..4.21 rows=4 width=14)
-> Hash Join (cost=2.07..4.21 rows=2 width=14)
Hash Cond: c.student_id = s.id
-> Redistribute Motion 2:2 (slice1; segments: 2) (cost=0.00..2.09 rows=2 width=10)
Hash Key: c.student_id
-> Seq Scan on classes c (cost=0.00..2.03 rows=2 width=10)
-> Hash (cost=2.03..2.03 rows=2 width=12)
-> Seq Scan on students s (cost=0.00..2.03 rows=2 width=12)
Optimizer status: legacy query optimizer
这个图展示了上面例子中的 SQL 在2个segment的Greenplum集群中执行时的示意图。
QD（Query Dispatcher、查询调度器）：Master 节点上负责处理用户查询请求的进程称为 QD（PostgreSQL 中称之为 Backend 进程）。 QD 收到用户发来的 SQL 请求后，进行解析、重写和优化，将优化后的并行计划分发给每个 segment 上执行，并将最终结果返回给用户。此外还负责整个 SQL 语句涉及到的所有的QE进程间的通讯控制和协调，譬如某个 QE 执行时出现错误时，QD 负责收集错误详细信息，并取消所有其他 QEs；如果 LIMIT n 语句已经满足，则中止所有 QE 的执行等。QD 的入口是 exec_simple_query()。
QE（Query Executor、查询执行器）：Segment 上负责执行 QD 分发来的查询任务的进程称为 QE。Segment 实例运行的也是一个 PostgreSQL，所以对于 QE 而言，QD 是一个 PostgreSQL 的客户端，它们之间通过 PostgreSQL 标准的libpq 协议进行通讯。对于 QD 而言，QE 是负责执行其查询请求的PostgreSQL Backend进程。通常 QE 执行整个查询的一部分（称为 Slice）。QE 的入口是 exec_mpp_query()。
Slice：为了提高查询执行并行度和效率，Greenplum 把一个完整的分布式查询计划从下到上分成多个 Slice，每个 Slice 负责计划的一部分。划分slice的边界为 Motion，每遇到 Motion 则一刀将 Motion切成发送方和接收方，得到两颗子树。每个 slice 由一个QE进程处理。上面例子中一共有三个 slice。
Gang：在不同 segments 上执行同一个 slice 的所有 QEs 进程称为 Gang。上例中有两组 Gang，第一组Gang负责在2个segments上分别对表 classes 顺序扫描，并把结果数据重分布发送给第二组Gang；第二组Gang 在2个segments 上分别对表students顺序扫描，与第一组Gang发送到本segment的 classes 数据进行哈希关联，并将最终结果发送给 Master。
并行执行流程
下图展示了查询在 Greenplum 集群中并行执行的流程。该图假设有2个segments，查询计划有两个slices，一共有 4 个 QEs，它们之间通过网络进行通讯。
QD 和 QE 都是 PostgreSQL backend 进程，其执行逻辑非常相似。对于数据操作（DML）语句（数据定义语句的执行逻辑更简单），其核心执行逻辑由 ExecutorStart, ExecutorRun, ExecutorEnd 实现。
QD：
●ExecutorStart 负责执行器的初始化和启动。Greenplum 通过 CdbDispatchPlan 把完整的查询计划发送给每个 Gang 中的每个 QE 进程。Greenplum 有两种发送计划给 QE 的方式：1）异步方式，使用 libpq 的异步 API 以非阻塞方式发送查询计划给QE；2）同步多线程方式：使用 libpq 的同步 API，使用多个线程同时发送查询计划给 QE。GUC gp_connections_per_thread 控制使用线程数量，缺省值为0，表示采用异步方式。Greenplum 从6.0开始去掉了异步方式。
●ExecutorRun 启动执行器，执行查询树中每个算子的代码，并以火山模型（volcano）风格返回结果元组给客户端。在 QD 上，ExecutorRun 调用 ExecutePlan 处理查询树，该查询树的最下端的节点是一个 Motion 算子。其对应的函数为 ExecMotion，该函数等待来自于各个 QE 的结果。QD 获得来自于 QE 的元组后，执行某些必要操作（譬如排序）然后返回给最终用户。
●ExecutorEnd 负责执行器的清理工作，包括检查结果，关闭 interconnect 连接等。
QE 上的 ExecutorStart/ExecutorRun/ExecutorEnd 函数和单节点的PostgreSQL代码逻辑类似。主要的区别在 QE 执行的是 Greenplum 分布式计划中的一个 slice，因而其查询树的根节点一定是个 Motion 节点。其对应的执行函数为 ExecMotion，该算子从查询树下部获得元组，并根据 Motion 的类型发送给不同的接收方。低一级的 Gang 的QE把 Motion 节点的结果元组发送给上一级 Gang 的QE，最顶层 Gang 的 QE 的 Motion 会把结果元组发送给 QD。Motion 的 Flow 类型确定了数据传输的方式，有两种：广播和重分布。广播方式将数据发送给上一级 Gang的每一个 QE；重分布方式将数据根据重分布键计算其对应的QE处理节点，并发送给该 QE。
QD 和 QE 之间有两种类型的网络连接：
●libpq：QD 通过 libpq 与各个QE间传输控制信息，包括发送查询计划、收集错误信息、处理取消操作等。libpq 是 PostgreSQL 的标准协议， Greenplum 对该协议进行了增强，譬如新增了 ‘M’ 消息类型 （QD 使用该消息发送查询计划给 QE）。libpq 是基于 TCP 的。
●interconnect：QD 和 QE、QE 和 QE 之间的表元组数据传输通过 interconnect 实现。Greenplum 有两种 interconnect 实现方式，一种基于 TCP，一种基于UDP。缺省方式为 UDP interconnect 连接方式。
Direct Dispatch 优化
有一类特殊的 SQL，执行时只需要单个 segment 执行即可。譬如主键查询：SELECT * FROM tbl WHERE id = 1;
为了提高资源利用率和效率，Greenplum 对这类 SQL 进行了专门的优化，称为 Direct Dispatch 优化：生成查询计划阶段，优化器根据分布键和WHERE子句的条件，判断查询计划是否为 Direct Dispatch 类型查询；在执行阶段，如果计划是 Direct Dispatch，QD 则只会把计划发送给需要执行该计划的单个 segment 执行，而不是发送给所有的 segments 执行。
二. 分布式事务
Greenplum 使用两阶段提交（2PC）协议实现分布式事务。2PC 是数据库经典算法，此处不再赘述。本节概要介绍两个 Greenplum 分布式事务的实现细节：
●分布式事务快照：实现 master和不同segment间一致性
●共享本地快照：实现 segment 内不同 QEs 间的一致性
分布式快照
在分布式环境下，SQL 在不同节点上的执行顺序可能不同。譬如下面例子中 segment1 首先执行SQL1，然后执行 SQL2，所以新插入的数据对 SQL1 不可见；而segment2上先执行 SQL2 后执行 SQL1，因而 SQL1 可以看到新插入的数据。这就造成了数据的不一致。
Greenplum 使用分布式快照和本地映射实现跨节点的数据一致性。Greenplum QD 进程承担分布式事务管理器的角色，在QD开始一个新的事务（StartTransaction）时，它会创建一个新的分布式事务id、设置时间戳及相应的状态信息；在获取快照（GetSnapshotData）时，QD 创建分布式快照并保存在当前快照中。和单节点的快照类似，分布式快照记录了 xmin/xmax/xip 等信息，结构体如下所示：
typedef struct DistributedSnapshot
{
DistributedTransactionTimeStamp distribTransactionTimeStamp;
DistributedTransactionId xminAllDistributedSnapshots;
DistributedSnapshotId distribSnapshotId;
DistributedTransactionId xmin; /* XID < xmin 则可见 */
DistributedTransactionId xmax; /* XID >= xmax 则不可见 */
int32 count; /* inProgressXidArray 数组中分布式事务的个数 */
int32 maxCount;
/* 正在执行的分布式事务数组 */
DistributedTransactionId *inProgressXidArray;
} DistributedSnapshot;