最近一年一直在做PolarDB的并行优化器，过程中调研了各种分布式数据库系统的优化和执行框架，后续几篇文章将一一分享，首先介绍对PolarDB MySQL的并行优化框架影响最大的，也就是SQL Server PDW。

SQL Server PDW介绍

SQL Server PDW（Parallel DataWarehouse）是SQL Server的MPP版本，目前已经演进为Azure DataWarehouse部署在云上，用来存储大容量数据并处理分析型查询。总体上是一个share nothing的经典MPP架构，类似于Greenplum，它也会利用单机SQL Server作为其sharding data和meta data的存储+计算实例。

基本架构

集群中每个节点都部署单个SQL Server instance + DMS服务，节点具备2种角色：

• Control node 是集群的入口点，前端应用于control node连接并发送请求，其上有一个PDW engine，做全局性的管理控制：distributed query优化、执行调度管理，DMS管理，权限检查，对外接口。内部的SQL server上有一个shell database，保存全局信息：global metadata/global statistics/数据分布/权限信息，和GP一样没有user data。
  
• compute node 存储user data，并负责分布式子计划的执行。
  

表数据的分片方式包括hash-partitioned和replicated（复制表）。

用户query到达control node后，优化生成分布式执行计划，称为DSQL。和MemSQL不同，PDW的分布式plan可以用一系列的step来描述，每个step之间不构成pipelining，step直接顺序依次执行，中间结果数据要物化到temp table后，才能开始下一step，step内是并行。

DSQL plan包含4种operation，每种operation都用SQL来描述（和MemSQL类似）

1. SQL operation，描述step的操作，直接在compute node的instance上执行SQL，返回数据
   
2. DMS operation，step执行请求发送到DMS服务上，DMS对所属instance执行SQL获取源数据，根据分发方式发送，接收方DMS把数据存入本地instance的temp table中
   
3. Temp table operation : setup temp table，并用来接收远程数据，供本地读取
   
4. Return operation，结果返回client
   

PDW 并行优化

整体来说，PDW的查询优化分为了2个步骤，分别在2个不同的组件中完成。

1. PDW Parser，获取用户query（T-SQL）做语法解析，生成AST，传递给本地SQL Server实例。
   
2. 在Control node的SQL Server实例中，利用shell database中存储的全局metadata + 统计信息等，完成单机执行计划的优化，包括：
   

• apply logical transformation
  
• cardinality estimation，基于shell database statistics
  
• apply physical implementation，为算子枚举物理执行方式，计算相应代价并做一些基本的剪枝
  

由于SQL Server的单机优化器是基于Cascades的，优化的结果保存在一个Memo的结构中，有关Cascades的原理，可参考之前的paper解读以及Orca的实现。但有所不同的是，这里并不选出最优的串行执行计划，而是将Memo中整个search space都保留下来。

3. 将Memo传递给XML generator做序列化。

4. 传递Memo给PDW optimizer，它基于memo中各种alternative的plans，根据数据分布情况，枚举可能的分布式执行方式+数据分发方式。相当于扩展了search space的新维度，加入了分布相关的信息，生成了新的候选plan集合，并基于cost，从中选出最优结果。

这种2-pass的方式有很多好处，首先，它可以完全复用SQL Server强大的单机优化器能力，众所周知SQL Server的QO应该是众多commercial database中最为优秀的，这样实现完全解耦了并行优化和单机优化，工程难度也降低了1个数量级。而且由于保存了整个Memo，不会导致串 -> 并带来的计划次优性问题。

在分布式组件PDW optimizer中，只需要去扩展DMS cost model，扩展physical property加入distribution即可。

PDW Query Optimizer实现

从上图的流程可以看到，概念上这个过程并不复杂，但工程实现的挑战还是很多的，下面一一介绍。

对现有Cascades优化器的增强

1. 能够将Memo输出
   
2. 扩展SQL语法，使其能够接受一些PDW相关的hints
   
3. 调整单机优化流程，在enumeration过程中，也会倾向对分布执行更有利的一些优化方式（这里没有细讲，个人猜想比如更倾向于subquery的展开?）
   
4. 扩展更多的逻辑/物理算子，比如partial aggregation/final aggregation，以及每种operator分布式实现，join可以是local/directed/broadcast/repartition, aggregation可以是repartition/partial+final的形式。
   
5. 扩展出DMS enforcer，不同的分发方式形成不同物理operator
   

Plan Enumeration

这是本篇paper的核心，也就是如何枚举分布式算子，基本原理如下

引入针对特定列(join列/group列)的distribution propery：(哪列，如何分布)

• 自底向上，对Memo中的group进行预处理：
  

1. 修正一些分布式执行算子的cardinality estimation，例如partial/final aggregation，在单机节点上虽然枚举了出来但没有分布式的statistics，这个card是不准确的，需要校正。
   
2. 从分布式的角度，对每个group，合并等价的group expression。
   
3. 自顶向下，生成每个group的interesting property（扩展了System-R的interesting order，添加了distribution属性）。
   

• 自底向上，对每个group枚举分布计划
  

1. 针对input group中所有可能expression，枚举本group内所有可能的分布式执行方式，生成新的group expression。
   
2. 基于cost做剪枝，对每个physical equivalent class（相同具有输出property），保留一个lowest cost plan，此外保留一个总体的best子计划。
   
3. 基于之前生成的interesting property，枚举可能的move enforcer，然后再做和步骤2类似的cost-based pruning。
   

• 提取最优plan
  

从root group中获取最优的plan tree

• 生成DSQL plan
  

Cost model

从paper中的描述来看，PDW的cost model只考虑数据传输+DMS做temp table物化的代价，而不考虑执行SQL操作的代价，这有几个原因

1. 这部分代价确实占据了执行中的绝大多数时间
   
2. 由于考虑的更少，使得cost model处理的范围更小，易管理，开发、调试、测试都会简化
   

其基本假设是

1. DSQL step顺序执行没有流水线，这有cost的累计就对应了执行时间的累加
   
2. 不考虑并发query的影响
   
3. 认为各node在硬件上同质的
   
4. 数据在节点间均匀分布，也就是不考虑data skew
   

但很遗憾paper没有给出设计细节。

DMS operation

DMS支持多种数据分发方式，这和PDW的并行执行能力相关，例如

Shuffle Move(N : N) ，最经典的redistribution，基于shuffle key的hash value

Partition Move(N : 1)，相当于汇总

Broadcast Move (N : ALL)，数据从部分compute node分发到所有compute node

Replicated broadcast(1 : ALL)，从1个compute node分发到所有compute node

...

总的来说，DMS被分为source + target 两个component，source发送和targe接收是同时进行的:

• source从本地instance获取数据 + network(网络分发) , 本地read和network分发是异步的
  
• target从Network接收数据放入本地buffer + bulkcpy(批量写入本地instance temp table), 本地收 和 bulkcpy也是异步的，代价公式是
  

由于以上操作都是异步的，而cost衡量的就是query执行时间，其cost formula是

Csource = max(Creader, Cnetwork).

Ctarget = max(Cwriter, CSQLBulkCpy).

Cdms = max(Csource, Ctarget).

在这篇著名的paper 中，我们都看到这样一个结论：由于Cardinality Estimation通常具有很大的误差，导致cost model的精确性变得不那么重要。

这里PDW的设计者们也做出了类似的判断，认为精细的cost model会使得plan对于数据分布/统计信息/执行环境等的微小变化都更为敏感，反而不够robust，此外维护和调试成本也更高，因此这里每个Cxx的计算都非常简单：Cxx = B * λ，B是操作处理的字节数，λ是不同操作对应的常量代价因子，可以看到cost model保持了尽可能的简单。

一个栗子

paper中给出了一个简单的例子，SQL是

SELECT *
FROM CUSTOMER C, ORDERS O
WHERE C.C_CUSTKEY = O.O_CUSTKEY
AND O.O_TOTALPRICE > 1000;

上图非常清晰的描述了4个主要步骤都发生了什么：

(a)用户输入到PDW engine

(b)Parser生成AST

(c)在单机SQL Server中完成单机优化，形成Final (Serial) Memo，传递给PDW optimizer，枚举Shuffle/Replicate等算子，生成Augmented (Parallel) Memo。

(d)从c的Parallel Memo中选出最优分布式计划

(e)转换为DSQL的plan描述，其中包括DMS operation + SQL operation...

总结

SQL Server PDW的并行优化方案还是非常优雅的，从本质上，它利用Cascades的top-down strategy完成单机串行的优化，再利用System-R的bottom-up完成并行的分布式优化，将两者结合了起来，是非常聪明的解决方案。不需要太多调整原有的Cascades组件，又简洁高效的生成了分布式计划。