当前位置:网站首页>Presto中broadcast join和partition join执行计划的处理过程
Presto中broadcast join和partition join执行计划的处理过程
2022-08-04 01:51:00 【王飞活】
一. 前言
在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);
}
边栏推荐
- Continuing to invest in product research and development, Dingdong Maicai wins in supply chain investment
- nodejs+npm的安装与配置
- halcon自定义函数基本操作
- 【正则表达式】笔记
- Continuing to invest in product research and development, Dingdong Maicai wins in supply chain investment
- Use nodejs switch version (no need to uninstall and reinstall)
- Web APIs BOM- 操作浏览器:swiper 插件
- Example 039: Inserting elements into an ordered list
- esp32发布机器人电池电压到ros2(micro-ros+CoCube)
- Array_Sliding window | leecode brushing notes
猜你喜欢
随机推荐
Snake game bug analysis and function expansion
Please refer to dump files (if any exist) [date].dump, [date]-jvmRun[N].dump and [date].dumpstream.
The browser
实例037:排序
22/8/3(板子)树状dp板子+中国剩余定理+求组合数3,4+容斥原理
企业虚拟偶像产生了实质性的价值效益
如何通过API接口从淘宝(或天猫店)复制宝贝到拼多多接口代码对接教程
flask框架初学-06-对数据库的增删改查
参加Oracle OCP和MySQL OCP考试的学员怎样在VUE预约考试
Flask框架初学-05-命令管理Manager及数据库的使用
- heavy OpenCV 】 【 mapping
Example 040: Reverse List
持续投入商品研发,叮咚买菜赢在了供应链投入上
SAP SD module foreground operation
ThreadLocal
关联接口测试
螺旋矩阵_数组 | leecode刷题笔记
【QT小记】QT中信号和槽的基本使用
HBuilderX的下载安装和创建/运行项目
贪吃蛇游戏Bug解析及功能扩展