Presto 之Hash Join的Partition

一. 前言

  在Presto中，当两表Join为Hash Join并且join_distribution_type为PARTITIONED的时候，Presto会将Build表分区（Partition）后再进行Join操作。在Presto中的Join操作中，对表的分区有两级，第一级是将Hash和Probe表的Hash列不同的Hash值的行分到不同的worker之间并行Join计算，第二级是节点内部再进行二次分区，将Hash列的Hash值再Hash计算，然后分配到不同的HashBuilderOperator中并行Join计算。本文主要简述Presto中这两级Partition的实现。

二. 节点间Partition

  为了使得可以多个节点之间并行Join，但是又不可能将Hash包和Probe表的数据都装到同一个节点上，就得要求Hash表和Probe表Join Key列的相同的Hash值的数据在调度的时候要被调度到相同的一个节点上才能进行碰撞操作。Presto此功能的实现是通过依赖PartitionedOutputOperator算子的实现的。        

   在PartitionedOutputOperator中，对于每一个TableScan中过来的Page，都会根据Hash列计算Hash值，然后与worker的数据取余求得改行数据应该被调度到哪个worker上进行Join计算。主要代码如下所示：

  求取任意一行的数据应该所在的分区：bucketCount就是worker的数量：

  对TableScan进行Partition：

 PageBuilders是个数组，不同worker拉取不同下标的Page，比如worker0固定是拉取pageBuilders[0]的page，worker1固定拉取pageBuilders[1]的page，从而实现将数据Partition到多个节点中，但是相同Key值又在同一个节点中的Partition处理。

二. 节点内部Partition

   除了将Hash表和Probe表Partition到多个节点计算外，Presto还做了节点内部的二次Partition并行计算优化。二次Partition的实现原理是对Build表的每一行数据的Hash Key的Hash值再次Hash，根据二次Hash值内部重新将一个节点的所有输入数据Partition到多个HashBuilderOperator中，从而加快Hash表的建立和一个节点内部的并行计算。

 节点内部二次Partition是在PartitioningExchanger算子中实现的，其实现的关键代码如下所示：

 节点内部build表被Partition成几个分区是由task.concurrency参数决定的，默认会将Build表Hash到16个分区中进行Join操作。

 Probe表在节点内部不再进行重新Partition。