Optimizing Queries over Partitioned Tables in MPP Systems

随着互联网数据的爆炸性增长，传统数据库系统在单表数据容量方面承受了越来越大的压力。以前公司内部的数据库，存放的主要是来自公司业务或内部管理系统的信息，中小型公司甚至一个MySQL实例就搞定了。但现在数据源不仅更丰富，数据量也在指数级增长，从业务的角度，基于hash/range的分区表变得越来越有吸引力。

为了能够对分区表有优异的处理能力，对于查询优化系统来说一个最基本的能力就是做partition pruning，将query中并不涉及的分区提前排除掉，可想而知这可以节省大量的IO/CPU/Network资源，带来成本的降低和性能的提升。

简单图示下什么叫partition pruning，例如下图的Orders表，在date列上按照month做了range分区，每个月单独作为一个partition。

SELECT avg(amount)
FROM orders
WHERE date BETWEEN '10-01-2013' AND '12-31-
2013';

如上SQL，由于单表谓词在parititon key上，在优化期间即可确定哪些可以分区可以避免访问，即静态pruning。

上例中的数据也可以改写为star schema的形态：

SELECT avg(amount)
FROM orders
WHERE date id IN
(SELECT date id
FROM date_dim
  WHERE year = 2013 AND 
month BETWEEN 10 AND 12);

这时Orders的值将需要根据date_dim表的动态输出决定，因此对Orders表的partition pruning只能在执行期完成，类似的例子还有动态绑定变量，称为动态pruning。

这篇paper主要介绍了一种基于Cascades框架规范且统一的分区消除方法，支持根据单表/join谓词，静态/动态的做partition pruning，并在Orca中做了实现。关于Cascades和Orca在之前的文章中已有过介绍，如果不熟悉的同学，可以参考如下2篇文章：

Cascades优化器框架

Orca

之所以要介绍这篇paper是由于PolarDB的并行优化框架也参考了Cascades，因此这里提到的方法可以很自然的应用于PolarDB对分区表的优化处理中，我们也在做这方面的规划。

基本算法

引入了3个新的算子来做pruning，且不区分是动态还是静态的：

• PartitionSelector
  

PartitionSelector根据谓词生成相关partition ids，主要实现了做partition选择的功能，并将筛选后的ids传递给DynamicScan。

• DynamicScan
  

物理scan算子，基于PartitionSelector传递的ids，在做table scan时跳过不需要的partition。

• Sequence
  

是一个同步的概念，用于描述PartitionSelector->DynamicScan的生产消费关系，确定谁先执行。

三者的配合有多种方式，如下图：

(a)表示做full table scan，这时是没有filter的，PartitionSelector直接生成全量partition T1 -> T100给DynamicScan。

(b/c)表示在partition key(PK)上有单表条件(PK=5/PK<30)，这时可以只选出目标分区给DynamicScan。

(d)表示了两表 R join T on R.A = T.PK，由于T表是partition table，其scan算子是DynamicScan。这里有所不同的是，对T的过滤需要基于R.A列的值，因此PartitionSelector要加在R的scan上方，用来获取R.A列的value用于过滤分区，并传递给右侧T的DynamicScan。另外可以看到这里不再有Sequence算子了，由于这里Join算子已经保障了R和T执行的先后顺序（先R后T），也就保证了PartitionSelector->DynamicScan的顺序，Sequence不再必要。

了解了这3个算子，其实基本思路就有了：

在原有的physical plan tree中，选择PartitionSelector的放置位置，应放置在可帮助消除分区的谓词(select/join)下方，并利用谓词中pkey相关的部分做分区消除，此外应尽量往下推，靠近对应的DynamicScan，在初期过滤掉更多的分区。

输入：已包含了DynamicScan（对应partition table）的physical operator tree，其中每个DynamicScan算子有其唯一编号partScanID。

输出：插入了PartitionSelector的新tree。

先描述下PartSpec的概念，用来描述partition select的行为，包含<partScanID, partKey, partPredicate>三元组，分别表示对应哪个DynamicScan，其分区列，分区选择谓词。

针对select(单表条件)的放置算法：

看起来有些乱，但其实非常简单，面对当前plan tree的selection node（过滤条件）时：

• 如果其partScanId对应的DynamicScan不在子树中，留在node上方，作为终止位置。
  
• 如果在子树中但selection node中没有和PKey相关的谓词，则只是推入op下方，进一步递归处理。
  
• 如果在子树中且selection node中有PKey相关谓词，将相关谓词合入到PartSpec的partPredicate中，然后推入op下方进一步递归处理。
  

对单表条件做PartSelector放置

上图很好的说明了这个例子，左侧是初始plan tree，右侧上方是初始PartSelector的描述信息(可以从谓词中获取到)。由于select中没有date_id这个表不在子树中，编号为2的PartitionSelector保留在了Select算子上方，而编号为1的则推到了Select下方，并把谓词条件"month >= 10 and month <= 12"加入到了下推的PartSpec中。

针对Join的放置算法

面对当前以join node为根节点的子树时：

• 如果其partScanId对应的DynamicScan不在子树中，留在join node上方，作为终止位置。
  
• 在子树中，但对应的分区表是Join的外表，由于外->内的驱动顺序，PartitionSelector没法用内表的数据来驱动外表的pruning，只能推到外表侧，看是否可以进一步递归处理（比如利用外表单表谓词）
  
• 在子树中，且对应的分区表在内表侧，但join条件本身和partKey无关，则这个join条件对分区消除无帮助，可以推入内表侧，看是否可以进一步递归处理
  
• 在子树中，对应的分区表在内表侧，且join条件和partKey相关，则可以推入外表侧，其将join条件中partkey相关的部分融入partPredicate。
  

对join放置PartSelector

join的处理流程要更复杂些，主要受限于PartitionSelector->DynamicScan的这种先后依赖顺序，因此如果想根据join condition做动态pruning(如上图)，必须要求分区表在被驱动侧（如NL join的内表，HashJoin的probe表）。上图示例中，由于date_dim是驱动侧（build侧），1,2两个PartitionSelector都推下来。

一个完整的示例包含了针对单表条件的PartitionSelector和针对Join条件的PartitionSelector，可以看到他们都放到了正确的最低位置。尤其注意，2号PartitionSelector所对应的DynamicScan表甚至不是同层的join table，这也充分显示了这种方法的灵活性，可以在更大的子树范围内做pruning。

Greenplum中实现

这套算法实现在了Greenplum的大数据查询优化器Orca中。注意从本质上，distribution和partition是两个正交的概念，在Greenplum这种share-nothing的MPP系统中，每个segment上都可以有对应的多个partitions。

1. 基于Orca已有Physical Property概念，将partition扩展为新一维的physical property来实现，信息用PartSpec来描述，这样就从<order, distribution>扩展到了<order, distribution, partition>三元组。
   
2. DynamicScan则是针对分区表的一种physical operator。
   
3. PartitionSelector实现为enforcer，放置在合适的group中，某些group expr(scan..)上方，来满足PartSpec的属性要求，具体放置算法如上节所述。
   

上图给出了R HJ S时的4种可能的PartitionSelector enforcer放置方式（灰色是group expression，表示物理算法实现，黑色表示enforcer，用来施加特定物理属性，例如Replicate表示要广播数据）。其中R是probe侧，S是build侧。

Plan 1 表示用PartitionSelector基于R.PK做裁剪，但由于R是被驱动侧，PartitionSelector没法用join condition对R做pruning，因此没有partPredicate。

Plan 2/3 也是类似的处理，只是后续的join方式不同

Plan 4 则使用了不同的hash join顺序，S变为驱动侧(build)，R变为被驱动侧(probe)，因此这时可以将PartitionSelector推入到S上方，并基于partPredicate: R.PK = S.a对R做过滤！

对于Plan 4可能大家有疑问，为什么不是先做PartitionSelector再做Replicate呢？不是应该尽量下推吗？这是由于为了避免多一次的网络交互，在Greenplum的实现中这种PartitionSelector->DynamicScan的信息传递是不跨segment的，因此是在Replicate后，基于分发后的数据，和R在各个segment内部做局部的partition pruning。

如上图所示，上方是无效plan因为PartitionSelector和DynamicScan在不同process中（PG的多进程模型），而下图则有效。为了保证这一点，在考虑每个group中的多种可能enforcer组合时(Motion/PartitionSelector)，需要约束Motion不能在PartitionSelector上方。

总结

虽然本篇写了不少，但基本思路还是非常简单的，partition pruning是一种性价比非常高的优化策略，一般实现不会太复杂，但却可以大幅度提升查询质量。因此各类系统提出了不同的pruning方案，例如Snowflake，由于其IO/事务的基本单元都是micro-partition，每个micro-partition是immutable的，因此可以维护精确统计信息，利用这种精确统计信息可以更好的实现静/动态的pruning。

对于PolarDB来说，除了优化阶段基于sargable filter的静态剪枝外，也实现了基于hash join的bloom filter做动态pruning，本篇提到的方法实际上是一种补充，不仅适用于hash join，也可用于nest loop join。