这篇简短的paper从非常high level的角度描述了下Oracle 10g对于parallel query所做的重新设计和其中的一些优化，由于Oracle RAC特殊的share-disk架构，使其在并行计算上与普通的MPP数据库有一些不同，例如对于worker的调度和分配方式以及对于资源/数据的动态调整。
PolarDB和Oracle一样都基于底层的共享存储，这让它们共享了一些基本的设计原则和架构优势，但在查询的优化和执行层面的细节上则存在不小的差别，后续我会写一篇文章专门介绍PolarDB的并行查询技术，现在先让我们看下Oracle的做法。（文中有一些我没有搞懂的点，请大神们批评指正）
基本介绍
新的parallel execution(PX) engine希望具有更好的灵活性，扩展性，对资源的高效利用和易管理性。
- 灵活性
由于底层share-disk的架构，使得用户不需要去关心数据的分区与节点间的映射关系，因此简化了应用的开发，此外由于在任意节点都可以访问所有数据，使得Oracle可以更加灵活和动态的方式去将数据片段分配到不同的worker上运行。PX engine其实是和底层的硬件特性以及数据的物理分区解耦的，因为每个节点都可以看到整体的数据视图。
- 扩展性
通过生成高效的parallel plan，并且在多个worker之间实现负载的均衡分布来达到好的扩展性，前者需要好的优化能力而后者则通过执行中的动态调整来实现。
- 资源高效利用
通过一些execution constructs来约束worker的分配，以及对每个worker的工作负载进行约束，从而对一些关键的执行资源（memory/lock/cpu...)实现更合理的利用。
- 易管理性
为了提升易管理性，引入了Parallel Single Cusor(PSC)模型，也就是对于一个SQL statement来说，所有的执行节点都共享一个相同的全局parallel plan，这样方便收集统计信息，监控执行状态。（这个点没有get到，为什么必须全局计划一样才好管理？？感觉和Oracle的历史实现方案相关。。。）
基本概念
PX engine支持节点内/节点间并行，节点内并行则通过share memory进行交互，节点间要通过网络交互。每个query的执行需要一个Parallel Execution Coordinator(PEC)和一组Parallel Execution Servers(PES)组成，也就是一个leader和一组workers。
PX可以支持各种类型的操作，包括：
- 各类关系运算(scan/join/aggregation...)
- DML (insert/update/delete...)
- DDL (create table/index/materialized view...)
- data reorganization
- data load/unload
一个Parallel Execution Plan(PEP)包含4个要素
- Dataflow operators(DFO)，分布式计划的一个片段，是一组算子的集合，等同于PolarDB的slice
- Table Queues(TQs)，执行数据分发操作，等同于PolarDB的exchange
- Granule Iterators (GRA)，控制对object(table/index)的动态分区，等同于PolarDB中的parallel scan
- Parallelizer row source，Oracle并行计划的形态就是一棵由DFO组成的树。Parallelizer腐恶调度DFO子树如何执行，每个DFO都由一组PES来并行执行。注意和普通的MPP并行数据库不同，Oracle的并行执行调度严格遵循2-DFOs的模式，也就是任意时间只有两个DFO在运行，一个producer一个consumer，当producer执行完成时，其PES空闲出来去执行下一个DFO，原来的consumer开始变为producer，而新的DFO成为了consumer。这种方式意味着2件事：
- PX的CPU占用最大为2 * DOP，每个DFO占用DOP个CPU
- 在producer和consumer之间存在temp table做中间结果的缓存，这样才能尽快切走producer把执行资源空出来，也就是DFO之间无法做到pipelining
Parallel Single Cursor模型
由PEC完成并行plan的优化，PEC/PES间共享相同的plan。
Single PEP的生成
这里对于优化流程讲的非常模糊让人十分郁闷，总的来说优化流程分为逻辑 + 物理两个步骤，
逻辑优化包含2个pass：
- 在给定的DOP下，确定join order和table的access method来使代价最低，这里计算cost时会考虑集群中node的个数，各个table/index的partition个数，以及默认的distribution方法（这个指什么？个人理解时每个算子有个默认的并行执行方式，例如group by按照group列做分布？hash join按照join key做分布？也就是按照这种分布来计算代价，然后决定access method和join order ? ）
- 真正为每个算子决定最优的distribution方法（这时才会考虑数据的statistics吗，例如数据量，可能的NDV？）
物理优化则将逻辑计划转换为实际的物理算子并构建为DFO，同时考虑一些基于数据物理属性的优化（clustering/ordering/data-fragmentation)。
吐槽下paper中的讲解。。。如果真的是个人理解的那样，为啥要搞成这种方案？不再考虑join order的同时考虑数据的各种分布方式？为啥物理属性的影响要到计划形态已确定时才考虑？
例如下图展示了一个简单的2表parallel hash join的计划
Parallelizer在PEC上负责整个计划的调度执行，Serial的部分负责对小表的串行scan通过hash redistribution分发到其他PES上（也在PEC上完成）。DFO 1负责大表的并行scan（由GRA负责多个PES扫表时对数据逻辑分片的分配，保证各个PES之间负载均衡)，然后将数据分发到DFO 2的各个worker中进行并行的hash join。
Single Global PEP的意义
一般的并行计算框架，在跨机传递执行计划时，会采用2种方案：
- 传递plan segment的某种编码形式（序列化...），例如Greenplum，PolarDB
- 传递描述plan segment的SQL语句，例如SQL Server PDW, MemSQL
Oracle认为两者都存在一定的缺陷
- 随着plan复杂度的上升，以及PX能够执行的并行功能越来与广泛，维护编码的方案将会越来越复杂且容易出错，这确实是一个问题，只要增加新的并行算子，就需要实现对应的编码/解码。
- 为每个DFO生成SQL可能无法准确描述DFO实际的物理执行计划，可能需要非常详尽的hint机制，而这也具有很高的开发+维护成本。
因此它采用了PSC的方案，即PEC在本地完成全局并行计划的优化，在相同node内的PES，共享同一个physical plan，而其他node则通过把原始SQL传递过去，完成相同的优化得到相同的parallel plan，并在那个node内与其上的PES共享。。。这样所有PES + PEC看到的都是全局一致的plan。这样有3个好处
- 易于监控各个DFO PES的执行，因为都面对相同的plan，各个PES在执行中的统计信息可以先记录在本地plan的row source当中，最后汇总到PEC上。
- 加入新的并行算子支持不需要考虑编/解码，只需要定义算子的默认分布方式（优化时用？）
- 使用SQL传递节省了内存，去掉中间表示。
但这里应该是有两个前提因素的：
- 物理执行计划和执行结构的解耦，使得多个PES能共享一个plan描述，这点MySQL就做不到
- 在各个节点上优化，能够生成完全相同的parallel plan，这点没想好是怎么做的，如何保证统计信息完全相同呢？还是优化中不考虑统计信息？
PSC的执行
通过GRA可以实现对data object(table/index)的并行读取比较均衡，具体方式是每个PES向PEC上的GRA请求一块object fragment information进行读取（多个page)，当其读取完成后会向PEC请求后续的granule，这样就可以实现根据各个PES消耗granule的速度，在多个PES间做到负载均衡。
最前面已经提到由于共享存储的特点，PX engine并不受底层data的分布限制，这样可以对数据形成某种virtual partitioning，而不受限与数据实际的user partition情况，但同时也可以利用实际的user partition做一些优化，例如paper中Full Partition-wise Join：每个PES只负载scan + join 一对partition的数据，完全避免了在PES之间的数据交互。和share-nothing架构不同，每个PES处理的partition和PES所在的计算node无关，不需要绑定在data“所在”的节点。
Cluster-aware PX
这里是指在优化中会考虑cluster的拓扑信息，例如上节提到的full partition-wise join，其形态如下：
这样确实没有任何data redistribution，但很明显并行PES的个数将受到partition个数的限制。
如上图所示，如果引入了hash redistribution，而且数据分发的成本，可以用更高的PES并行度来弥补，那么就可以提升查询的响应速度。但这里数据分发的量明显大了很多，还有更好的方法：
上图是前面2种方法的混合，利用user partition的schema，来避免跨节点的data redistribution，可以看到通过限制数据分发的目标PES，来实现只在部分partition之间做重分布，相当于把partition按照Node做了分组，每个node负责一部分partitions。这样看就变成了一个share nothing的结构，每个node绑定一些partition，在node内部完成并行join。
这种优化可以在存在partition-wise join且partition数量不足的情况下使用。当然具体采用哪种方案需要通过cost决定。
上面说的到对PES进行分组的约束是通过一个执行器的construct : PES mapper来实现了，它会对每条tuple进行检查，并约束其可能分发的PES分组。
Resource-aware PX
Memory
这里举了parallel insert partition table的例子，在并行insert时每个PES会为每个partition预留一个buffer来实现异步批量写入，那么为了避免占用内存过多，通过PES mapper限制每个PES所写入的partition数量（默认是每个PES都向所有partition写），来实现减少内存占用。
Locking
类似Memory，每个data block能够并发修改的process数量是有上限的，通过PES mapper来把PES分组，每个data block只能被某个分组的PES修改，从而限制了PES的数量。
Adaptive DOP and PES allocation
在每个query开始时，利用集群中的负载信息 + cpu数量来限制DOP，在分配PES到各个node时，要考虑node之间的负载均衡情况。
Granule allocation
在用GRA分配granule给各个PES时，应该考虑data和node的affinity或者data locality(这是指data是否在buffer pool中？还要到GCS中看block的owner吗？）
其他功能
Flexible data distribution and pipelining
对SQL做并行的一个核心要素就是对data object(静态table/index)或者data flow(动态的中间结果）实现分片。PX优化中会使用物理算子所具有的distribution requirement来完成这种分片。这里举例了window function的partition by，group by应该也是同样道理。同时优化中会考虑不同算子在物理属性上的兼容性，例如如果group by key和order by key是兼容的（group by a, b order by a)，则将执行group by的hash distribution变为range distribution，来避免再单独一个DFO做并行order by。
Cost-based Parallelization of Subqueries
Oracle对于大多数的相关子查询都可以实现去相关化，而对于无法unnesting的，则基于cost决定是否执行并行：
- subquery已经在PES中执行（在子计划片段中），自身就不能再并行了
- subquery在PEC上，但如果对每行输入，并行执行的cost太高，则也不并行
Recursive and Iterative Computation
由于producer-consumer之间存在temp table缓存中间结果，可以利用这个特点实现一些高效计算，例如在producer 生成数据时，通过redistribution把数据分布出去，而consumer在消费这些数据时，也可以不考虑temp table的分布特性（share-disk）,重新做动态的virtual partitioning，实现更优的并行方案。
Load-Balanced Table Queues
在数据分发过程中，可以通过一些动态策略来防止data skew，使数据均匀分发，例如对于range distribution，可以通过预采样的方式发现分布不均匀，从而动态调整range边界。
总结
这是一篇很老的paper了，相信其中的很多技术和现在Oracle的策略有所不同，但其充分share-disk架构的灵活性，实现灵活的执行调度 + 负载均衡策略，还是有一些借鉴意义的。
还是要吐槽下Oracle的论文风格，一如既往的粗糙。。。
