Presto中broadcast join和partition join执行计划的处理过程

一. 前言

    在presto中，partition join和broadcast join的差异主要在执行计划的方式上。如果join的一侧直接以REPLICATE exchange的方式将数据传送到另外一侧，则为broadcast join，如下图1的执行计划所示。 如果join 两侧的数据经过REPARTITION后再进行exchange到下游执行join操作，则为partition join，如下图2。

                                    图1： broadcast join执行计划样例

                       图2： partition join执行计划样例

  

  在presto中，可以通过设置join_distribution_type参数强制使用任意一种join方式。

  在presto中，生成partition join和broadcast join的执行计划在AddExchanges::visitJoin中实现的。

public PlanWithProperties visitJoin(JoinNode node, PreferredProperties preferredProperties)
{
   ....
   if (distributionType == JoinNode.DistributionType.REPLICATED) {
      // 生成broadcast  join执行计划
      planReplicatedJoin(node, left);
   }
   else {
       // 生成partition join执行计划
       return planPartitionedJoin(node, leftSymbols, rightSymbols);
   }
}

二. broadcast join

private PlanWithProperties planReplicatedJoin(JoinNode node, PlanWithProperties left)
{
    PlanWithProperties right = node.getRight().accept(this, PreferredProperties.any());

    // 如果左表已经是单节点运行的任务了（比如coordaintor的aggregate）, 则右表添加ExchangeNode.Type.GATHER的exchange 汇聚到左表执行join操作
    if (left.getProperties().isSingleNode()) {
        if (!right.getProperties().isSingleNode() ||
                (!isColocatedJoinEnabled(session) && hasMultipleSources(left.getNode(), right.getNode()))) {
            right = withDerivedProperties(
                    gatheringExchange(idAllocator.getNextId(), REMOTE, right.getNode()),
                    right.getProperties());
        }
    }
    else {
        // 右表直接以ExchangeNode.Type.REPLICATE的形式复制到左表，实现数据广播
        right = withDerivedProperties(
                replicatedExchange(idAllocator.getNextId(), REMOTE, right.getNode()),
                right.getProperties());
    }

    return buildJoin(node, left, right, JoinNode.DistributionType.REPLICATED);
}

三. partition join

   partition join的实现比broadcast join复杂一些，因为涉及到以哪一列进行hash的问题。其实现的过程如以下代码所示：

private PlanWithProperties planPartitionedJoin(JoinNode node, List<Symbol> leftSymbols, List<Symbol> rightSymbols, PlanWithProperties left)
{
    SetMultimap<Symbol, Symbol> rightToLeft = createMapping(rightSymbols, leftSymbols);
    SetMultimap<Symbol, Symbol> leftToRight = createMapping(leftSymbols, rightSymbols);

    PlanWithProperties right;

    // 左表的join列且是分区列
    if (left.getProperties().isNodePartitionedOn(leftSymbols) && !left.getProperties().isSingleNode()) {
        // 找到左表join列对应的右表列，对该列进行hash Partitioning
        Partitioning rightPartitioning = left.getProperties().translate(createTranslator(leftToRight)).getNodePartitioning().get();
        right = node.getRight().accept(this, PreferredProperties.partitioned(rightPartitioning));
        if (!right.getProperties().isCompatibleTablePartitioningWith(left.getProperties(), rightToLeft::get, metadata, session)) {
            // 如下一行代码会添加一个ExchangeNode.Type.REPARTITION的exchage，hash的列为右表的join列，从而实现对右表进行REPARTITION到下游进行join操作 
            right = withDerivedProperties(
                    partitionedExchange(idAllocator.getNextId(), REMOTE, right.getNode(), new PartitioningScheme(rightPartitioning, right.getNode().getOutputSymbols())),
                    right.getProperties());
        }
    }
    else {
        right = node.getRight().accept(this, PreferredProperties.partitioned(ImmutableSet.copyOf(rightSymbols)));

        // 如下的代码用意和上边相同，只是操作的左表，尝试将左表进行REPARTITION到下游进行join操作 
        if (right.getProperties().isNodePartitionedOn(rightSymbols) && !right.getProperties().isSingleNode()) {
            Partitioning leftPartitioning = right.getProperties().translate(createTranslator(rightToLeft)).getNodePartitioning().get();
            left = withDerivedProperties(
                    partitionedExchange(idAllocator.getNextId(), REMOTE, left.getNode(), new PartitioningScheme(leftPartitioning, left.getNode().getOutputSymbols())),
                    left.getProperties());
        }
        else {
            // 如下是处理无分区列的情况，左右表都按照自己Join列进行Hash partition，然后exchange到下游执行join 操作
            left = withDerivedProperties(
                    partitionedExchange(idAllocator.getNextId(), REMOTE, left.getNode(), leftSymbols, Optional.empty()),
                    left.getProperties());
            right = withDerivedProperties(
                    partitionedExchange(idAllocator.getNextId(), REMOTE, right.getNode(), rightSymbols, Optional.empty()),
                    right.getProperties());
        }
    }

    ....

    return buildJoin(node, left, right, JoinNode.DistributionType.PARTITIONED);
}