从代数到执行#

关系代数只描述语义，不描述怎么执行。也就是说，$R \bowtie S$ 告诉我们结果应该是什么，但没有告诉我们：

• 是先构建哈希表再探测？
• 还是嵌套循环逐行比较？
• 还是先排序再归并？
• 如果是并行执行，数据如何分区？

这就是执行引擎要解决的问题。

对于单线程执行来说，JOIN 语义可以非常自然地映射到一个局部算法上；但一旦进入并发和分布式环境，关系代数的“同一性”就不再自动成立了：

语义上只是一个 $R \bowtie S$，执行上却可能被拆成多个 worker、多个节点、多个数据分区。

这时就需要一个机制，把“关系代数中的一个 JOIN”映射成“可并发执行的多个局部 JOIN”。这个机制就是 Exchange。

Exchange、哈希分桶和 Broadcast 的关系#

从数学角度看，Exchange 可以被理解为一个关系重分布算子。它接收一个关系 $R$，根据某种分布规则 f，把 $R$ 映射成多个面向 worker 的输入：

不同的 f，对应不同的 Exchange 模式。

先看最普通的单线程 JOIN#

1
SELECT a.*, b.*
2
FROM table_a a
3
JOIN table_b b ON a.key = b.key;

在单机单线程执行模型中，数据库执行计划通常可以看成一棵 iterator 树。每个算子都提供类似 open()、next()、close() 的接口：

• open()：初始化算子状态
• next()：返回下一行结果
• close()：释放资源

在进入具体流程前，先澄清 Hash Join 里最关键的两个术语：build（构建）和 probe（探测）。这两个术语最早出自 Kitsuregawa、Tanaka、Moto-Oka 关于 hash 数据库机的工作《Application of Hash to Data Base Machine and Its Architecture》，后来被几乎所有 hash join 实现沿用，成为描述执行阶段的通用词汇：

• build：取两边输入中较小的那个，扫描它并在内存里建一张哈希表，这一侧称为 build side / build table。
• probe：取较大的那个输入，逐行用 join key 去刚才建好的哈希表里查找匹配，这一侧称为 probe side / probe table。

之所以让小表做 build，是因为哈希表要常驻内存：小表既省内存，又能让大表在 probe 时只扫描一遍。

一个简化的 Hash Join 执行过程如下：

1
HashJoin
2
├── Scan(table_a)  -- probe side
3
└── Scan(table_b)  -- build side

执行时通常分为两个阶段。

构建阶段（Build Phase）：

1. 扫描较小的 build table，例如 table_b
2. 按 join key 构建哈希表
3. 哈希表保存在当前执行线程的内存中

探测阶段（Probe Phase）：

1. 扫描较大的 probe table，例如 table_a
2. 对每一行计算 join key
3. 到哈希表中查找匹配项
4. 输出 JOIN 结果

可以用下面的伪代码表示：

1
hash_table = {}
2

3
for row_b in scan(table_b):
4
    hash_table[row_b.key].append(row_b)
5

6
for row_a in scan(table_a):
7
    matches = hash_table.get(row_a.key)
8
    for row_b in matches:
9
        output(row_a, row_b)

这个模型的优点是简单：所有状态都在一个线程内，JOIN 算子只需要维护自己的哈希表和扫描状态，不需要考虑线程通信、数据分区、网络传输，也不需要考虑不同 worker 之间如何协作。

但它的问题也很明显：当数据量变大时，单线程会成为瓶颈。

1. CPU 只能使用一个执行单元#

即使机器有几十个 CPU core，单线程 Hash Join 仍然只能使用其中一个 core。数据量越大，CPU 计算、哈希表构建、哈希探测都只能串行完成。

1
单线程执行：
2

3
Scan B -> Build Hash Table -> Scan A -> Probe -> Output
4

5
全部工作由一个线程完成

这意味着性能提升主要依赖单核性能，而不是系统整体计算能力。

2. 内存压力集中#

Hash Join 的 build side 哈希表全部保存在一个执行线程的上下文里。如果 build table 较大，单线程执行会面临：

• 哈希表过大
• cache locality 变差
• 内存分配压力集中
• 可能触发 spill to disk

如果能把数据拆成多个分区，每个 worker 只处理其中一部分 key，单个 worker 的内存压力就会明显下降。

3. 无法利用数据局部性#

在分布式数据库中，数据天然分布在不同节点上。比如：

1
节点1: table_a 的一部分
2
节点2: table_a 的一部分
3
节点3: table_b 的一部分
4
节点4: table_b 的一部分

如果仍然使用单线程思路，就需要把数据集中到一个地方再 JOIN。这会导致：

• 网络 I/O 集中
• 单节点 CPU 压力集中
• 单节点内存压力集中
• 集群资源无法充分利用

4. JOIN 需要“同 key 同地”#

JOIN 的语义要求相同 join key 的两边数据必须相遇。

例如：

1
table_a: a1(key=1), a2(key=2)
2
table_b: b1(key=1), b2(key=2)

要完成 JOIN，a1 必须能和 b1 相遇，a2 必须能和 b2 相遇。单线程模型中，这件事天然成立，因为所有数据都在同一个执行上下文里。

但在并发或分布式执行中，这件事不再天然成立。数据可能分布在不同线程、不同进程、不同机器上。于是问题变成：

如何保证相同 join key 的数据被送到同一个 Join worker？

这个问题正是 Exchange 要解决的核心问题。

问题 1：谁来决定一行数据属于哪个 worker？#

对于 Hash Join，通常需要按 join key 做 hash 分区：

1
target_worker = hash(join_key) % worker_count

如果 table_a 和 table_b 都使用同一个分区函数，那么相同 join key 的数据就会被送到同一个 worker。

问题 2：谁来负责在线程或节点之间传输数据？#

Scan 算子只负责读取数据，Join 算子只负责关联数据。如果让每个 Join 算子都自己处理网络传输、线程同步、缓冲队列和分区逻辑，所有算子都会变得非常复杂。

数据库执行引擎需要一个统一的位置来处理这些事情。

问题 3：如何保持 iterator 模型不变？#

传统执行引擎中，上层算子只需要调用下层算子的 next()。如果引入并发后，每个算子都要知道下层有多少线程、多少节点、多少缓冲区，执行模型就会被破坏。

理想状态是：

• Join 算子仍然像单线程一样调用输入
• Scan 算子仍然像单线程一样产生数据
• 并发、分区、通信由一个独立算子负责

这个独立算子就是 Exchange。

从单线程管道到并发管道#

单线程执行中，数据流是这样的：

1
row1 -> row2 -> row3 -> row4

并发执行中，数据流会被拆成多个分区：

1
             -> worker 1: row1, row4, row8
2
upstream rows -> worker 2: row2, row5, row9
3
             -> worker 3: row3, row6, row7

Exchange 的职责就是完成这个拆分和转发。

Exchange 不改变 JOIN 语义#

Exchange 并不负责真正的 JOIN 匹配。它只保证数据在进入 Join 算子之前已经被组织成适合并发执行的形态。

对 Join 算子来说，它看到的仍然是两路输入：

1
HashJoin
2
├── ExchangeReceiver(table_a partition)
3
└── ExchangeReceiver(table_b partition)

每个 Join worker 都只处理自己的分区。只要 Exchange 保证相同 key 的两边数据进入同一个 worker，局部 Join 的结果合起来就是全局 Join 的结果。

1. 封装并行性#

没有 Exchange 时，如果要把一个算子并行化，可能需要修改每个算子的内部实现。例如：

• Scan 要知道自己应该把数据发给哪些线程
• Join 要知道上游有多少并发生产者
• Aggregate 要知道如何接收多个 worker 的局部结果
• Sort 要知道如何处理跨节点的有序数据

这会让每个算子都耦合并发执行细节。

Exchange 的做法是把这些复杂性集中到一个算子中：

• 启动多个执行副本，也就是论文中提到的 clone
• 管理 clone 的 startup 和 teardown
• 在 producer 和 consumer 之间建立数据通道
• 处理线程、进程或节点之间的数据传输
• 根据分区策略把数据送到正确的目标

这样，普通算子仍然可以按照 iterator 模型实现，而不需要感知自己是否运行在并行环境中。

2. 架构无关性#

不同硬件架构下，并行执行的底层机制并不一样：

如果每个算子都直接依赖这些机制，执行引擎就很难在不同架构之间复用。

Exchange 的抽象价值在于：

对上层算子来说，它只是一个 iterator；对执行引擎来说，它可以映射成线程队列、进程通信或网络 shuffle。

这也是论文标题中 “Architecture-Independence” 的含义。

1. Hash Exchange（哈希交换）#

Hash Exchange 的分布函数是哈希函数。给定关系 $R$ 和分区键 key，它定义：

因此，Hash Exchange 得到的是一组水平分区：

它的核心性质是：如果两个元组的 join key 相同，那么它们会被映射到同一个分区。

所以，对于等值 JOIN，只要 $R$ 和 $S$ 使用同一个分区函数，就有：

适用场景：

• Hash Join
• 按 key 聚合
• 需要让相同 key 数据聚集到同一 worker 的场景

示例：

1
SELECT a.*, b.*
2
FROM table_a a
3
JOIN table_b b ON a.user_id = b.user_id;

在这个查询中，table_a 和 table_b 都可以按 user_id 做 Hash Exchange。只要两边使用相同的 hash 函数和 worker 数量，相同 user_id 的数据就会进入同一个 Join worker。

1
Before Exchange:
2

3
节点1: a(user_id=1), a(user_id=2)
4
节点2: b(user_id=1), b(user_id=2)
5

6
After Hash Exchange:
7

8
worker-1: a(user_id=1), b(user_id=1)
9
worker-2: a(user_id=2), b(user_id=2)

从数学上看，Hash Exchange 的目标不是平均分配数据本身，而是保证同 key 同分区。数据均衡是性能目标，同 key 同分区才是 JOIN 正确性的要求。

2. Broadcast Exchange（广播交换）#

Broadcast Exchange 的分布函数不是切分，而是复制。给定关系 $S$，它为每个 worker 生成一份完整副本：

如果另一个关系 $R$ 已经被切分为：

那么 JOIN 可以写成：

这里和 Hash Exchange 最大的区别是：$S$ 不需要按照 join key 分区，因为每个 worker 都持有完整的 $S$。只要 $r \in R_i$ 需要匹配某个 $s \in S$，这个 $s$ 一定可以在当前 worker 本地找到。

适用场景：

• 小表 JOIN 大表
• 维度表 JOIN 事实表
• 小结果集需要被多个下游 worker 使用

例如：

1
SELECT f.*, d.description
2
FROM fact_table f
3
JOIN dim_table d ON f.dim_id = d.id;

如果 dim_table 很小，可以把它广播到所有处理 fact_table 的 worker：

1
Broadcast dim_table:
2

3
worker-1: fact partition 1 + full dim_table
4
worker-2: fact partition 2 + full dim_table
5
worker-3: fact partition 3 + full dim_table

Broadcast 的数学含义是通过复制消除跨分区匹配问题；工程上的收益是避免移动大表。它的代价是小表会被复制多份，当 worker 数量很多，或者所谓“小表”并不小的时候，广播成本会迅速上升。

3. Gather Exchange（汇总交换）#

Gather Exchange 的方向和 Hash / Broadcast 相反。Hash 和 Broadcast 是把一个输入关系分发给多个 worker，而 Gather 是把多个 worker 的局部结果收敛成一个输出关系。

如果每个 worker 产生局部结果 $Q_i$，那么 Gather 的数学形式是：

也可以写成：

适用场景：

• 查询最终结果输出
• 全局 LIMIT
• 单点汇总
• 部分全局聚合

1
worker-1 result \
2
worker-2 result  -> coordinator
3
worker-3 result /

Gather 是最容易理解的一种 Exchange，但也最容易成为瓶颈。因为它把并发流重新收敛成单个输出流。如果 $Q_i$ 很大，Gather 节点就会成为 CPU、内存和网络的集中点。

4. Range Exchange（范围交换）#

Range Exchange 的分布函数是一个范围映射。它根据排序键或范围键，把关系切分成有序、不重叠的区间。

例如定义边界：

则可以定义：

因此：

并且分区之间满足范围顺序：

适用场景：

• ORDER BY
• 全局排序
• 分布式 Top-N
• 需要保持范围有序的任务

例如：

1
worker-1: key < 100
2
worker-2: 100 <= key < 200
3
worker-3: key >= 200

Range Exchange 的优点是可以支持有序输出。只要每个分区内部有序，并且分区之间范围不重叠，最终结果就可以按分区顺序合并。缺点是需要估计或采样数据分布，如果范围划分不合理，可能导致数据倾斜。

5. Round-Robin Exchange（轮询交换）#

Round-Robin Exchange 不依赖 key，而是按照输入元组的到达顺序分配给不同 worker。可以把第 $k$ 条输入元组映射为：

也就是：

1
row1 -> worker-1
2
row2 -> worker-2
3
row3 -> worker-3
4
row4 -> worker-1

适用场景：

• 不关心 key 语义的负载均衡
• 简单并发处理
• 上游数据需要均匀打散

从数学语义看，Round-Robin 仍然是一种水平切分：

但它不保证任何 key 相关性质。也就是说，通常不能推出：

因为相同 join key 的 $r$ 和 $s$ 可能被分到不同 worker。所以 Round-Robin 更适合无 key 依赖的并发处理，而不适合直接用于等值 JOIN 的 key 对齐。

1. Shuffle Join / Repartition Join#

两边表都按 join key 做 Hash Exchange。

设：

那么：

这表示两个关系都被切分成同样的分区，然后每个 worker 只负责一个局部 JOIN。

1
table_a -- Hash Exchange on key --\
2
                                  HashJoin workers
3
table_b -- Hash Exchange on key --/

适合场景：

• 两张表都比较大
• 两边原始分布都不能保证 join key 对齐
• 无法通过广播小表解决

优点：

• 能充分并行
• 每个 worker 只处理一部分 key
• 适合大表 JOIN 大表

缺点：

• 两边都需要网络 shuffle
• 对数据倾斜敏感
• 网络和序列化成本较高

2. Broadcast Join#

小表通过 Broadcast Exchange 复制到所有 worker，大表保持原地或按原分区扫描。

设大表 $R$ 被切分为：

而小表 $S$ 被复制到每个 worker：

那么：

1
small_table -- Broadcast Exchange --\
2
                                  HashJoin workers
3
large_table -- local scan ---------/

这里不要求 $S$ 先按 join key 分桶，因为每个 worker 都持有完整的 $S$。于是每个局部 worker 都可以独立完成：

适合场景：

• 一边表很小
• 另一边表很大
• 广播小表的成本低于 shuffle 大表

优点：

• 避免大表移动
• 执行简单
• 对大事实表 JOIN 小维表非常有效

缺点：

• 小表会复制多份
• worker 数量越多，广播总成本越高
• 如果小表估算错误，可能导致内存压力

3. Colocated Join#

如果两张表本来就按照相同的分布规则存储，那么可以不再做额外的数据重分布。

设：

并且数据已经物理上放在对应节点上，那么：

但这里的关键是：这个分解不需要通过 Exchange 再次搬运数据，因为 $R_i$ 和 $S_i$ 本来就已经在同一个节点上。

1
节点1: table_a bucket 1 + table_b bucket 1 -> local join
2
节点2: table_a bucket 2 + table_b bucket 2 -> local join
3
节点3: table_a bucket 3 + table_b bucket 3 -> local join

适合场景：

• 两表分布已对齐
• 分布键和 join key 一致
• bucket / tablet 对齐

优点：

• 几乎不需要网络传输
• 局部性最好
• 通常是最理想的执行方式

缺点：

• 对建表分布设计有要求
• 优化器需要准确识别分布兼容性

4. Bucket Shuffle Join#

Bucket Shuffle Join 可以看成是对“部分分布已知”的场景做优化。

假设 $R$ 已经按 bucket 分布：

而 $S$ 只需要被重分布到与 $R$ 相同的 bucket 上：

那么：

但与完全 Shuffle Join 不同的是，$R$ 不需要再移动，只需要让 $S$ 对齐到 $R$ 的分布。

适合场景：

• 一侧大表已经按 join key 分布
• 另一侧表需要对齐到大表分布
• 完全 shuffle 两边成本过高

优点：

• 比双边 shuffle 成本低
• 比 broadcast 更适合中等规模表
• 能复用已有数据分布

缺点：

• 依赖分布信息
• 优化器选择更复杂

5. 统一理解#

从数学角度看，这些 JOIN 策略的区别只是：

• 是否对两边都做分区
• 是否只对一边做分区，另一边复制
• 是否根本不做额外分区，只复用已有分布

它们最终都在回答同一件事：如何把 $R \bowtie S$ 分解成若干个可以并发执行的局部 JOIN，再把这些局部结果组合成全局结果。

Sender：实现分布函数#

Exchange Sender 的核心职责是计算分布函数。

对于一条输入元组 $r$：

然后把 $r$ 放入目标 $i$ 对应的发送缓冲区。

不同 Exchange 类型的差异，主要体现在 f 的定义上：

Receiver：恢复局部关系#

Exchange Receiver 的职责是把多个 Sender 发来的数据合并成当前 worker 的局部输入。

数学上，worker $i$ 接收到的是：

如果是 Broadcast，则每个 worker 接收到的是完整关系：

下游算子不需要知道 $R_i$ 是本地扫描得到的，还是从网络接收得到的。它只需要像单线程 iterator 一样不断调用 next()。

这就是 Exchange 封装并行性的关键：

数学上，它定义关系如何分解；物理上，它负责把分解后的数据送到对应 worker；接口上，它仍然表现为普通 iterator。

生命周期管理#

从论文角度看，Exchange 的 startup 和 teardown 很重要。因为并发执行不是只多开几个线程，还要确保：

• 每个 clone 正确启动
• 上下游 worker 数量匹配
• 数据通道建立完成后再开始传输
• 出错时能够取消其他 clone
• 所有输入结束后能够正确关闭

这些生命周期管理能力，是 Exchange 从“数据传输算子”变成“并行执行控制算子”的关键。

1. 减少不必要的 Exchange#

Exchange 的收益来自并发，但成本来自数据移动。优化器应尽量避免不必要的数据重分布。

常见手段包括：

• 利用 colocated join
• 复用已有分布属性
• 谓词下推减少 Exchange 前的数据量
• 投影下推减少传输列数

2. 选择合适的分布策略#

不同策略适合不同场景：

3. 控制并发度#

并发度不是越高越好。

并发度提高后，可能带来：

• 更多 worker
• 更多网络连接
• 更多缓冲区
• 更多上下文切换
• 更高的内存占用

因此需要在 CPU、内存、网络带宽之间做平衡。

4. 处理数据倾斜#

Hash Exchange 对数据倾斜敏感。如果某些 join key 特别热，大量数据会被发送到同一个 worker，导致局部瓶颈。

常见优化方式包括：

• 统计信息辅助优化器选择策略
• 对热点 key 做特殊处理
• 使用 broadcast 避免某些 shuffle
• 增加局部聚合减少传输量
• 对倾斜 key 做拆分或 salting

5. 批量传输和压缩#

Exchange 通常不会逐行发送数据，而是以 block、batch 或 chunk 为单位传输。

这样可以：

• 降低网络调用次数
• 提升吞吐量
• 减少序列化开销
• 更好地利用压缩

但 batch 太大也会增加延迟和内存占用，所以需要在吞吐和延迟之间平衡。

性能监控指标#

查询计划查看#

可以通过 EXPLAIN 查看查询计划中的 Exchange 操作：

1
EXPLAIN SELECT * FROM t1 JOIN t2 ON t1.id = t2.id;

计划中如果出现类似 EXCHANGE、HASH_PARTITIONED、BROADCAST 等信息，通常说明优化器在执行计划中引入了数据重分布或广播。

一个简化示意如下：

1
Hash Join
2
├── Exchange Receiver
3
│   └── Exchange Sender
4
│       └── Scan t1
5
└── Exchange Receiver
6
    └── Exchange Sender
7
        └── Scan t2

对于不同 JOIN，Doris 优化器可能选择不同策略：

• 如果两表都大，可能选择 Shuffle Join
• 如果一边很小，可能选择 Broadcast Join
• 如果两表分布已经对齐，可能选择 Colocated Join
• 如果一侧分桶可以复用，可能选择 Bucket Shuffle Join

这与前面从单线程到并发的分析是一致的：JOIN 算子本身负责匹配数据，Exchange 负责让数据以正确的方式进入对应的并发 Join 实例。