优化器如何解决算子间的数据流不匹配问题一、为什么 Planner 需要 AddLocalExchange在查询优化器Optimizer对逻辑计划进行物理规划Physical Planning时一个核心问题始终存在如何确保每个算子Operator都能获得符合它真正要求的数据流Dataflow这里所说的数据流并不是数据本身而是数据在本地执行环境中的组织方式例如数据是否按照某些列进行分区、是否保持特定顺序、以及由多少并行执行实例消费等。不同算子对数据流有着不同的要求。例如Aggregation 需要相同 Group Key 的数据进入同一个数据分区Data Partition从而保证负责该分区的执行实例能够维护完整的聚合状态。Sort 希望最终获得单个有序的数据流以保证全局排序结果正确当然这也并非绝对在支持分布式排序的情况下Sort 需要 FIXED distribution 的数据流。TableWriter 则可能要求固定并行度的数据流以满足写入策略的要求。然而上游算子天然产生的数据流并不一定能够满足下游算子的执行要求。换句话说Operator 与 Operator 之间的数据流并不是自然而然就兼容的Dataflow Mismatch。因此在物理规划阶段Optimizer 必须负责协调这种数据流的不匹配而不是把这一职责交给各个算子自己完成。说明从数据流规划角度数据分区Partition描述数据应该如何被划分和路由执行实例则描述“哪些实例负责消费这些分区数据”。Local Exchange 的作用就是建立两者之间的映射关系。一个典型的例子例如下面这个查询SELECTa,b,COUNT(*)FROMtGROUPBYa,b;对于 Aggregation 来说它真正需要的是Need: HASH on (a, b)也就是说相同 (a,b) 的数据必须进入同一个执行分区使对应的 Aggregation driver执行实例 能够维护完整的聚合状态。而上游输入节点例如未携带有效分区信息 TableScan 或 Remote Exchange往往会提供类似于Have: MULTIPLE (unknown)即多个并行的数据流。这里的 unknown 指的是 Optimizer 无法证明数据已按特定列分区。因此上下游算子之间产生了数据流属性的不匹配为了满足 Aggregation 的执行要求Optimizer 必须主动插入一个 Local Exchange重新组织数据流需要注意的是Local Exchange 改变的是数据流的组织方式而不是数据本身。它负责在本地执行环境中重新组织输入数据流将数据按照目标分区方式分发给下游执行实例使下游 Operator 获得满足执行要求的数据布局。AddLocalExchanges 的职责这正是 AddLocalExchanges 优化规则存在的原因。它是 Optimizer 在物理规划阶段负责本地数据流规划Local Dataflow Planning的核心组件其职责可以概括为三点判断上下游算子之间是否存在数据流不匹配Need vs Have在必要的位置插入合适类型的 Local Exchange对数据流进行重组在插入 Local Exchange 后重新推导数据流属性确保整个执行计划始终满足各个算子的执行要求。AddLocalExchanges 在优化流程中的位置在 Presto 的物理计划优化阶段Exchange 相关处理由多个优化规则共同完成各阶段关注的问题不同。其中AddExchanges 主要负责生成 Remote Exchange解决跨 Worker 节点的数据重分布问题例如根据 Join Key 或 Grouping Key 将数据发送到正确的执行节点。AddLocalExchanges 位于 AddExchanges 之后主要负责生成 Local Exchange解决单个 Worker 内部不同 Pipeline 执行片段Driver之间的数据组织问题。即使 Remote Exchange 已经保证相同 Key 的数据位于同一个 Worker仍需要保证这些数据在 Worker 内部进一步按照执行需求进行组织从而满足 Aggregation、Join 等算子的输入属性要求。需要注意的是AddLocalExchanges 的核心目标是保证物理计划的正确执行而不是进行成本优化。它通过算子语义推导数据流属性并在属性不满足时插入必要的 Local Exchange。至于 Exchange 是否冗余、是否可以消除、并行度是否最优、是否存在更低成本的数据流方案则由后续优化规则如 Exchange 简化、冗余消除以及其他物理优化阶段进一步处理。在介绍完 AddLocalExchanges 优化规则所处的位置和其职责之后一个新的问题也随之产生Optimizer 如何知道一个算子需要什么样的数据流又如何判断上游算子实际提供了什么样的数据流为了回答这两个问题Presto 引入了一套完整的数据流属性约束机制Dataflow Property Enforcement Mechanism。下一章将围绕这一机制展开分析 AddLocalExchanges 如何通过声明需求、比较属性、重组数据流来完成整个本地数据流规划过程。二、数据流属性约束机制Dataflow Property Enforcement Mechanism2.1 数据流属性约束理解了为什么需要之后我们来看 AddLocalExchanges 的设计哲学。想要理解 AddLocalExchanges首先需要理解 Optimizer 如何在物理规划阶段分析并满足 Operator 对数据流属性的要求。本文将这一过程概括为“数据流属性约束机制”。简单来说数据流属性约束描述了 Operator 对输入数据流属性的要求以及 Optimizer 如何通过属性推导、需求匹配和 Exchange 插入等机制将逻辑上的算子需求转换为满足执行要求的物理数据流组织方式。在这一机制中Parent Operator 声明需求Need例如HASH partitioning on (a,b)SINGLE data streamORDERED data streamChild Operator 提供实际数据流属性Have例如已经按照 (a,b) 分区未知分区多个并行输入流Optimizer 负责协调两者之间的差异如果满足需求直接连接如果不满足则插入 Local Exchange 进行数据流转换。注意这里的 Parent / Child 指 Plan Tree 中的父子关系数据流实际方向是由 Child 流向 Parent。2.2 数据流属性约束的核心价值Optimizer 不要求每个算子自己产生正确分布的数据流而是允许算子只声明需要什么由优化器负责在必要时重组数据流。这意味着算子可以专注于业务逻辑聚合、连接、排序等数据流管理由优化器统一负责整个系统更模块化、更可扩展这实现了 Operator 计算逻辑与数据流管理逻辑的解耦使新的 Operator 可以专注于计算语义而无需自行处理数据重新分布问题。2.3 数据流属性约束的实施者在 AddLocalExchanges 的优化过程中planAndEnforce() 是数据流属性约束机制的核心实施入口。它负责接收 Parent Operator 的属性需求规划 Child Operator 并获取其实际属性在需求不满足时插入对应的 Local Exchange。三、数据流属性系统——如何描述算子对数据流的 Need 与 Have3.1 为什么需要统一的属性描述有了前面数据流属性约束的概念我们现在需要一个核心工具一套统一的语言来描述当前算子需要什么和上游算子实际提供什么。如果没有统一的描述语言Aggregation 说我需要 HASH 分区TableScan 说我提供 MULTIPLE 数据流两者说的不是同一套概念无法比较因此Presto 引入了一套数据流属性描述体系。其中StreamPreferredProperties 用于描述 Parent Operator 对输入数据流属性的期望要求StreamProperties 用于描述 Child Operator 实际产生的数据流属性。3.2 StreamPreferredProperties描述 Operator 的输入数据流属性需求父节点用 StreamPreferredProperties 向子节点表达需求publicclassStreamPreferredProperties{privatefinalbooleansingleStream;// 是否期望单流privatefinalbooleanfixedParallelism;// 是否期望固定并行privatefinalOptionalListVariableReferenceExpressionpartitioningColumns;// 期望分区列privatefinalbooleanorderSensitive;// 是否对排序敏感}需要注意的是父节点表达的需求包含两种类型Required Properties 和 Preferred Properties它们都使用 StreamPreferredProperties 进行定义。planAndEnforce() 的执行流程并不是直接使用 Required Properties 强制修改子节点而是首先利用 Preferred Properties 递归规划子节点使其产生实际的数据流属性StreamProperties。随后Planner 再将子节点实际提供的属性Have与父节点要求的 Required PropertiesNeed进行匹配。如果实际属性已经满足 Required则直接复用当前计划如果不满足则进入 enforce() 阶段通过插入 Local Exchange 对数据流进行重组使其最终满足算子的执行要求。3.3 StreamProperties描述 Operator 输出的数据流属性publicclassStreamProperties{// 数据流在 Pipeline 间的分布publicenumStreamDistribution{SINGLE,// 单个本地数据流MULTIPLE,// 多个数据流分区未知FIXED// 多个数据流按照明确 partitioning scheme 分布}privatefinalStreamDistributiondistribution;privatefinalOptionalListVariableReferenceExpressionpartitioningColumns;privatefinalbooleanordered;privatefinalOptionalActualPropertiesotherActualProperties;// ...}其描述的核心问题是数据分布在多少个数据流中在哪些列上分区是否有序以及每个数据流内部的数据物理特性如排序、分组等等。3.4 分布类型的本地语义类型本地语义典型场景SINGLE单个数据流VALUES、GATHER 之后MULTIPLE多个数据流分区未知TableScan、REPLICATE 之后FIXED多个数据流分区明确REPARTITION 之后3.5 分区列的包含关系这是一个容易被误解但很重要的概念// 当前流在 (a, b) 上分区 // 判断能否满足在 (a, b, c) 上分区的要求 // 答案可以相同的 (a,b,c) 值一定在同一个分区中 actual.isPartitionedOn([a, b, c]) → true // 当前流在 (a, b, c) 上分区 // 判断能否满足在 (a, b) 上分区的要求 // 答案不可以相同的 (a,b) 值可能分布在不同分区c 不同 actual.isPartitionedOn([a, b]) → false这个逻辑在本地 Pipeline 间同样适用决定了哪些操作可以在不 shuffle 的情况下完成。3.6 ActualProperties完整的物理属性描述除了分布属性Planner 还需要知道数据在单个数据流内部的组织方式该属性通过 ActualProperties 来描述publicclassActualProperties{privatefinalGlobalglobal;// 描述了全局相关的数据分布特性包括数据在多个节点之间如何分布以及数据在多个数据流切片之间如何分布privatefinalListLocalPropertyVariableReferenceExpressionlocalProperties;// 描述数据在单个数据流切片内部的特性如排序、分组等privatefinalMapVariableReferenceExpression,ConstantExpressionconstants;// 常量列描述// ...}需要注意的是在 AddLocalExchanges 所关注的 StreamProperties 中我们只对其 ActualProperties 中的 localProperties 感兴趣。持有整个 ActualProperties 对象只是因为 PropertyDerivations 要求输入类型为 ActualProperties。StreamProperties 和 ActualProperties 的关系如下所示属性类型关注维度描述内容StreamProperties数据流之间数据如何分布在多个数据流中AddLocalExchanges 专门关心ActualProperties数据流内部 全局节点之间以及数据流之间完整的物理属性描述包含分布、排序、常量等一个更准确的理解StreamPropertiesAddLocalExchanges 专门关心的数据流分布抽象ActualPropertiesPlanner 推导出的完整物理属性描述从关注范围来看StreamProperties 可以理解为 ActualProperties 中与本地数据流组织相关的一部分属性抽象是 AddLocalExchanges 进行本地数据流规划时重点关注的信息。3.7 数据流属性约束体系的整体流程Operator 根据自身语义定义属性需求 ↓ Visitor 将这些规则编码为 Planner 可执行逻辑 ↓ StreamPreferredProperties 描述 Need ↓ StreamProperties 描述 Have ↓ planAndEnforce() 执行规划并比较 Need 与 Have ↓ 匹配直接复用现有数据流 不匹配插入 Local Exchange ↓ 重新推导输出属性四、Operator 的数据流属性语义前面介绍了 Presto 如何使用 StreamProperties 描述当前数据流状态以及使用 StreamPreferredProperties 表达算子的输入需求。但是Optimizer 如何知道一个算子需要什么样的数据流又如何知道其执行完成后数据流会具有什么属性答案是每个 Operator 都定义了自身的数据流属性语义Dataflow Property Semantics用于描述输入约束和输出属性。这些语义是物理规划阶段进行属性推导和 Exchange 插入的基础。4.1 输入需求与输出属性对于一个 OperatorOptimizer 需要关注两个问题1. 对输入数据流的需求和建议Input Requirement Preferred当前算子需要以及建议上游提供什么样的数据流。例如某些算子要求数据按照特定字段分区某些算子要求单流执行某些算子要求输入满足排序条件。这些需求最终通过 StreamPreferredProperties 来定义用于指导子节点规划并执行数据流属性约束。2. 实际输出数据流的属性Output Property当前算子被规划并执行之后下游节点可以依赖的数据流特征。例如Filter 通常保持输入的数据分布Exchange 会改变数据分布Project 会根据字段映射传播已有属性某些算子会丢失部分输入属性。这些结果会通过 StreamProperties 来定义用于后续节点的属性推导。4.2 算子语义与物理规划的关系需要注意AddLocalExchanges 并不关心算子的具体执行逻辑而只关心算子对数据流组织形式的要求。例如Aggregation 在执行层面可能涉及Hash AggregationStreaming AggregationPartial / Final Aggregation。但在数据流规划阶段Planner 只需要知道grouping key 是否满足数据聚合要求Join 也不关心具体 Join Operator 如何执行而关注输入数据分布是否满足 Join 策略要求因此Operator 属性语义提供的是一种物理约束抽象而不是执行实现描述。4.3 Operator 属性语义与 Visitor在源码实现中Operator 的属性语义通过 Visitor 模式进行编码。Visitor 模式的本质其本身并不是优化算法而是将不同 PlanNode 对输入数据流属性的要求以及它们产生的输出属性编码为 Optimizer 可以理解和操作的形式。真正负责比较 Need 与 Have、插入 Exchange、重新推导属性的是 planAndEnforce() 方法。Visitor 定义的是“属性语义”而 planAndEnforce() 执行的是“满足属性约束”。说明Operator 属性语义与具体执行策略密切相关并且仍处于持续演进过程中不同版本之间可能存在差异。因此本文只讨论其在物理规划中的抽象作用即算子如何声明数据流约束以及 Planner 如何基于这些约束完成属性推导和 Exchange 调整。对于具体算子的当前实现细节建议直接参考对应版本源码。五、planAndEnforce —— 属性约束满足的核心流程前面的章节介绍了数据流不匹配问题、属性约束机制、属性描述语言、算子属性语义等。现在需要一个过程将这些属性语义转换为满足约束的物理计划。planAndEnforce() 正是完成这一过程的核心方法。它接收父节点的属性需求递归处理子节点比较需求与实际属性的差距在必要时插入 Local Exchange 重组数据流最终确保每个节点都获得满足执行要求的数据流。5.1 planAndEnforce 的职责planAndEnforce() 的核心职责概括为六步Step 1: 接收需求 ── 从父节点获取 Required 和 Preferred 属性需求Step 2: 递归规划 ── 使用 Preferred 建议对每个子节点进行递归规划传递裁剪后的需求Step 3: 收集属性 ── 收集子节点规划处理后的实际数据流属性Step 4: 比较判断 ── 检查 Have 是否满足 Need (Required)Step 5: 修正实施 ── 不满足时插入 Local Exchange重新推导属性Step 6: 返回属性 —— 将当前节点输出属性作为父节点的 Have其核心思想是不断将父节点的建议需求preferred向下传播并在数据流无法满足强制要求required时通过 Exchange 主动修正。方法签名如下privatePlanWithPropertiesplanAndEnforce(PlanNodenode,StreamPreferredPropertiesrequired,// 父节点必须满足的需求StreamPreferredPropertiespreferred);// 父节点偏好的需求优化提示PlanWithProperties 返回值包含处理后的节点及其产出的实际数据流属性。Required vs PreferredRequired 是硬约束不满足时 Planner 必须通过 Exchange 等机制进行修正。如 Aggregation 必须在 grouping keys 上分区。Preferred 是软约束满足可提升性能不满足也正确。如某些节点偏向单流处理。属性需求裁剪在递归规划过程中父节点的需求不会直接传递给所有子节点。而是会根据子节点实际输出的数据列对需求进行裁剪Join Need: HASH(a, b) / \ Left Child Right Child 输出: a, c 输出: b, d ↓ 裁剪后传递 ↓ 裁剪后传递 Need: HASH(a) Need: HASH(b)确保每个子节点只接收与自己相关的需求。5.2 Need vs HaveStreamPreferredProperties 如何判断属性满足递归规划完成后Planner 会比较Need: 父节点要求的数据流属性Have: 子节点实际产生的数据流属性属性匹配本质上是判断已有数据流属性是否足够支持当前算子的执行要求。例如更强的数据流属性可以满足较弱的数据流需求典型情况例如SINGLE ORDERED 可以满足 SINGLE HASH(a) 可以满足 HASH(a, b) --分区包含关系isPartitionedOn详见上面3.5节的说明综上所述isSatisfiedBy() 并不是简单比较 distribution 是否完全相等而是根据数据流属性的强弱关系判断是否满足。在属性匹配过程中Planner 会根据 StreamProperties 的 distribution、partitioning 和 ordering 判断 Child Stream Properties 是否满足 Parent Requirement。如果不能满足则进入 enforce 阶段。5.3 enforce通过 Local Exchange 修正属性在 enforce 阶段Planner 根据缺失的属性类型插入对应的 Local Exchange。如下所示Exchange 类型属性变化典型场景GATHER任意 → SINGLE全局排序、最终汇聚HASH REPARTITION任意 → FIXED partitioningAggregation、JoinROUND ROBIN任意 → MULTIPLE并行负载均衡例如Aggregation 需要HASH(a, b)但输入数据流为MULTIPLE则 HASH Exchange on (a, b) MULTIPLE ----------------------------- FIXED(a, b)插入之后 Aggregation 即可正确执行。5.4 Exchange 后重新推导属性由于 Exchange 改变了数据流组织形式因此插入 Exchange 后必须重新计算属性。例如插入前数据流分布为 MULTIPLE 或 FIXED插入后数据流分布变为了 SINGLE 插入前数据流分布为 SINGLE 或 MULTIPLE插入厚数据流分布变为了 FIXED(a, b)因此在 Local Exchange 插入之后我们需要重新执行属性推导derive properties以得到新的数据流属性StreamProperties。属性推导是自底向上的子节点属性会向上传播并决定父节点属性。planAndEnforce() 返回前通过 checkState() 验证所有硬约束均已满足确保计划执行时不会因数据流属性不匹配而失败。总结planAndEnforce() 通过递归规划 → 属性比较 → 插入 Exchange → 重新推导的闭环将算子属性语义转化为满足约束的物理执行计划是 AddLocalExchanges 优化规则得以实施的核心引擎。六、完整执行流程示例Local Exchange 如何完善执行计划需要注意的是Presto 的物理计划生成通常包含两个层面的数据重组Remote Exchange解决 Task/Stage 之间的数据传输和重新分区问题Local Exchange解决单个 Task 内部多个 Pipeline 之间的数据组织问题。本示例重点关注 AddLocalExchanges 如何基于已有物理计划属性调整 Local 数据流。Remote Exchange 仅作为背景展示不展开分析。以如下的分组聚合语句作为示例SELECTa,b,COUNT(*)FROMtGROUPBYa,b;经过前序物理规划后计划已经包含跨 Task 数据重分布Output | RemoteExchange(GATHER) | Aggregation | RemoteExchange(MULTIPLE) | TableScan这里RemoteExchange(MULTIPLE) 负责 Stage 间的数据重新分区。但是在每个 Task 内部仍然需要确定各 Pipeline 如何消费这些数据。AddLocalExchanges 处理 Aggregation 的实际行为如下所示初始计划片段Aggregation | RemoteExchange (MULTIPLE)planAndEnforce() 开始执行对于 Aggregation 节点处理逻辑为递归处理子节点 RemoteExchange规划并获取 RemoteExchange 的输出数据流分布类型: MULTIPLE比较 Aggregation Need vs RemoteExchange HaveNeed: FIXED on (a,b)Have: MULTIPLE, 分区未知不匹配调用 enforce()插入 LocalExchange (HASH on a,b)修正后计划Output | RemoteExchange(GATHER) | Aggregation │ LocalExchange (HASH on a,b) ← 插入的 Local Exchange │ RemoteExchange(MULTIPLE) | TableScan重新推导属性LocalExchange 后: FIXED on (a,b)Aggregation 后: FIXED on (a,b)继承了分区特性Output 检查: 可以接受多流排序敏感允许保留校验最终计划满足所有需求。七、总结AddLocalExchanges 本质上解决的是查询执行计划生成过程中的一个核心问题如何将逻辑算子所表达的计算语义转换为满足执行要求的数据流组织方式。在分布式查询引擎中不同算子对输入数据流具有不同约束例如 Aggregation 需要满足 grouping key 的分区要求Join 需要满足关联键的数据布局要求Sort 和 TableWriter 也分别依赖特定的数据流属性。这些约束并不是由算子自身完成协调而是由 Optimizer 通过属性抽象和约束满足机制统一处理。AddLocalExchanges 通过 Visitor 定义算子属性语义利用 StreamPreferredProperties 和 StreamProperties 描述需求与实际状态并由 planAndEnforce() 递归的完成属性传播、需求匹配以及必要的 Exchange 插入。从设计思想上看AddLocalExchanges 体现了现代查询优化器中的 Property Enforcement 模型Optimizer 不直接决定执行细节而是通过物理属性描述执行约束再由规划规则自动调整数据流使最终执行计划满足所有算子要求。因此Local Exchange 并不是简单的数据搬运节点而是连接算子语义与物理执行的数据流控制机制。它将逻辑计划中的计算需求逐步转换为满足并行执行要求的数据流结构是 Presto 从逻辑计划走向可执行物理计划的重要阶段。作者王冬PrestoDB Global Committer | Presto Iceberg Code OwnerGitHub: https://github.com/hantangwangdEmail: mingwbdgmail.com欢迎就相关技术问题与实践经验进行交流与探讨本文章同步发表于https://hantangwangd.github.io/zh/posts/2026-07-27-add-local-exchanges.html 
长河君 · 2026/10/26 1:37:14 · 阅读 1.2k Presto 查询引擎内核详解：AddLocalExchanges——基于数据流属性约束的本地执行规划
优化器如何解决算子间的数据流不匹配问题一、为什么 Planner 需要 AddLocalExchange在查询优化器Optimizer对逻辑计划进行物理规划Physical Planning时一个核心问题始终存在如何确保每个算子Operator都能获得符合它真正要求的数据流Dataflow这里所说的数据流并不是数据本身而是数据在本地执行环境中的组织方式例如数据是否按照某些列进行分区、是否保持特定顺序、以及由多少并行执行实例消费等。不同算子对数据流有着不同的要求。例如Aggregation 需要相同 Group Key 的数据进入同一个数据分区Data Partition从而保证负责该分区的执行实例能够维护完整的聚合状态。Sort 希望最终获得单个有序的数据流以保证全局排序结果正确当然这也并非绝对在支持分布式排序的情况下Sort 需要 FIXED distribution 的数据流。TableWriter 则可能要求固定并行度的数据流以满足写入策略的要求。然而上游算子天然产生的数据流并不一定能够满足下游算子的执行要求。换句话说Operator 与 Operator 之间的数据流并不是自然而然就兼容的Dataflow Mismatch。因此在物理规划阶段Optimizer 必须负责协调这种数据流的不匹配而不是把这一职责交给各个算子自己完成。说明从数据流规划角度数据分区Partition描述数据应该如何被划分和路由执行实例则描述“哪些实例负责消费这些分区数据”。Local Exchange 的作用就是建立两者之间的映射关系。一个典型的例子例如下面这个查询SELECTa,b,COUNT(*)FROMtGROUPBYa,b;对于 Aggregation 来说它真正需要的是Need: HASH on (a, b)也就是说相同 (a,b) 的数据必须进入同一个执行分区使对应的 Aggregation driver执行实例 能够维护完整的聚合状态。而上游输入节点例如未携带有效分区信息 TableScan 或 Remote Exchange往往会提供类似于Have: MULTIPLE (unknown)即多个并行的数据流。这里的 unknown 指的是 Optimizer 无法证明数据已按特定列分区。因此上下游算子之间产生了数据流属性的不匹配为了满足 Aggregation 的执行要求Optimizer 必须主动插入一个 Local Exchange重新组织数据流需要注意的是Local Exchange 改变的是数据流的组织方式而不是数据本身。它负责在本地执行环境中重新组织输入数据流将数据按照目标分区方式分发给下游执行实例使下游 Operator 获得满足执行要求的数据布局。AddLocalExchanges 的职责这正是 AddLocalExchanges 优化规则存在的原因。它是 Optimizer 在物理规划阶段负责本地数据流规划Local Dataflow Planning的核心组件其职责可以概括为三点判断上下游算子之间是否存在数据流不匹配Need vs Have在必要的位置插入合适类型的 Local Exchange对数据流进行重组在插入 Local Exchange 后重新推导数据流属性确保整个执行计划始终满足各个算子的执行要求。AddLocalExchanges 在优化流程中的位置在 Presto 的物理计划优化阶段Exchange 相关处理由多个优化规则共同完成各阶段关注的问题不同。其中AddExchanges 主要负责生成 Remote Exchange解决跨 Worker 节点的数据重分布问题例如根据 Join Key 或 Grouping Key 将数据发送到正确的执行节点。AddLocalExchanges 位于 AddExchanges 之后主要负责生成 Local Exchange解决单个 Worker 内部不同 Pipeline 执行片段Driver之间的数据组织问题。即使 Remote Exchange 已经保证相同 Key 的数据位于同一个 Worker仍需要保证这些数据在 Worker 内部进一步按照执行需求进行组织从而满足 Aggregation、Join 等算子的输入属性要求。需要注意的是AddLocalExchanges 的核心目标是保证物理计划的正确执行而不是进行成本优化。它通过算子语义推导数据流属性并在属性不满足时插入必要的 Local Exchange。至于 Exchange 是否冗余、是否可以消除、并行度是否最优、是否存在更低成本的数据流方案则由后续优化规则如 Exchange 简化、冗余消除以及其他物理优化阶段进一步处理。在介绍完 AddLocalExchanges 优化规则所处的位置和其职责之后一个新的问题也随之产生Optimizer 如何知道一个算子需要什么样的数据流又如何判断上游算子实际提供了什么样的数据流为了回答这两个问题Presto 引入了一套完整的数据流属性约束机制Dataflow Property Enforcement Mechanism。下一章将围绕这一机制展开分析 AddLocalExchanges 如何通过声明需求、比较属性、重组数据流来完成整个本地数据流规划过程。二、数据流属性约束机制Dataflow Property Enforcement Mechanism2.1 数据流属性约束理解了为什么需要之后我们来看 AddLocalExchanges 的设计哲学。想要理解 AddLocalExchanges首先需要理解 Optimizer 如何在物理规划阶段分析并满足 Operator 对数据流属性的要求。本文将这一过程概括为“数据流属性约束机制”。简单来说数据流属性约束描述了 Operator 对输入数据流属性的要求以及 Optimizer 如何通过属性推导、需求匹配和 Exchange 插入等机制将逻辑上的算子需求转换为满足执行要求的物理数据流组织方式。在这一机制中Parent Operator 声明需求Need例如HASH partitioning on (a,b)SINGLE data streamORDERED data streamChild Operator 提供实际数据流属性Have例如已经按照 (a,b) 分区未知分区多个并行输入流Optimizer 负责协调两者之间的差异如果满足需求直接连接如果不满足则插入 Local Exchange 进行数据流转换。注意这里的 Parent / Child 指 Plan Tree 中的父子关系数据流实际方向是由 Child 流向 Parent。2.2 数据流属性约束的核心价值Optimizer 不要求每个算子自己产生正确分布的数据流而是允许算子只声明需要什么由优化器负责在必要时重组数据流。这意味着算子可以专注于业务逻辑聚合、连接、排序等数据流管理由优化器统一负责整个系统更模块化、更可扩展这实现了 Operator 计算逻辑与数据流管理逻辑的解耦使新的 Operator 可以专注于计算语义而无需自行处理数据重新分布问题。2.3 数据流属性约束的实施者在 AddLocalExchanges 的优化过程中planAndEnforce() 是数据流属性约束机制的核心实施入口。它负责接收 Parent Operator 的属性需求规划 Child Operator 并获取其实际属性在需求不满足时插入对应的 Local Exchange。三、数据流属性系统——如何描述算子对数据流的 Need 与 Have3.1 为什么需要统一的属性描述有了前面数据流属性约束的概念我们现在需要一个核心工具一套统一的语言来描述当前算子需要什么和上游算子实际提供什么。如果没有统一的描述语言Aggregation 说我需要 HASH 分区TableScan 说我提供 MULTIPLE 数据流两者说的不是同一套概念无法比较因此Presto 引入了一套数据流属性描述体系。其中StreamPreferredProperties 用于描述 Parent Operator 对输入数据流属性的期望要求StreamProperties 用于描述 Child Operator 实际产生的数据流属性。3.2 StreamPreferredProperties描述 Operator 的输入数据流属性需求父节点用 StreamPreferredProperties 向子节点表达需求publicclassStreamPreferredProperties{privatefinalbooleansingleStream;// 是否期望单流privatefinalbooleanfixedParallelism;// 是否期望固定并行privatefinalOptionalListVariableReferenceExpressionpartitioningColumns;// 期望分区列privatefinalbooleanorderSensitive;// 是否对排序敏感}需要注意的是父节点表达的需求包含两种类型Required Properties 和 Preferred Properties它们都使用 StreamPreferredProperties 进行定义。planAndEnforce() 的执行流程并不是直接使用 Required Properties 强制修改子节点而是首先利用 Preferred Properties 递归规划子节点使其产生实际的数据流属性StreamProperties。随后Planner 再将子节点实际提供的属性Have与父节点要求的 Required PropertiesNeed进行匹配。如果实际属性已经满足 Required则直接复用当前计划如果不满足则进入 enforce() 阶段通过插入 Local Exchange 对数据流进行重组使其最终满足算子的执行要求。3.3 StreamProperties描述 Operator 输出的数据流属性publicclassStreamProperties{// 数据流在 Pipeline 间的分布publicenumStreamDistribution{SINGLE,// 单个本地数据流MULTIPLE,// 多个数据流分区未知FIXED// 多个数据流按照明确 partitioning scheme 分布}privatefinalStreamDistributiondistribution;privatefinalOptionalListVariableReferenceExpressionpartitioningColumns;privatefinalbooleanordered;privatefinalOptionalActualPropertiesotherActualProperties;// ...}其描述的核心问题是数据分布在多少个数据流中在哪些列上分区是否有序以及每个数据流内部的数据物理特性如排序、分组等等。3.4 分布类型的本地语义类型本地语义典型场景SINGLE单个数据流VALUES、GATHER 之后MULTIPLE多个数据流分区未知TableScan、REPLICATE 之后FIXED多个数据流分区明确REPARTITION 之后3.5 分区列的包含关系这是一个容易被误解但很重要的概念// 当前流在 (a, b) 上分区 // 判断能否满足在 (a, b, c) 上分区的要求 // 答案可以相同的 (a,b,c) 值一定在同一个分区中 actual.isPartitionedOn([a, b, c]) → true // 当前流在 (a, b, c) 上分区 // 判断能否满足在 (a, b) 上分区的要求 // 答案不可以相同的 (a,b) 值可能分布在不同分区c 不同 actual.isPartitionedOn([a, b]) → false这个逻辑在本地 Pipeline 间同样适用决定了哪些操作可以在不 shuffle 的情况下完成。3.6 ActualProperties完整的物理属性描述除了分布属性Planner 还需要知道数据在单个数据流内部的组织方式该属性通过 ActualProperties 来描述publicclassActualProperties{privatefinalGlobalglobal;// 描述了全局相关的数据分布特性包括数据在多个节点之间如何分布以及数据在多个数据流切片之间如何分布privatefinalListLocalPropertyVariableReferenceExpressionlocalProperties;// 描述数据在单个数据流切片内部的特性如排序、分组等privatefinalMapVariableReferenceExpression,ConstantExpressionconstants;// 常量列描述// ...}需要注意的是在 AddLocalExchanges 所关注的 StreamProperties 中我们只对其 ActualProperties 中的 localProperties 感兴趣。持有整个 ActualProperties 对象只是因为 PropertyDerivations 要求输入类型为 ActualProperties。StreamProperties 和 ActualProperties 的关系如下所示属性类型关注维度描述内容StreamProperties数据流之间数据如何分布在多个数据流中AddLocalExchanges 专门关心ActualProperties数据流内部 全局节点之间以及数据流之间完整的物理属性描述包含分布、排序、常量等一个更准确的理解StreamPropertiesAddLocalExchanges 专门关心的数据流分布抽象ActualPropertiesPlanner 推导出的完整物理属性描述从关注范围来看StreamProperties 可以理解为 ActualProperties 中与本地数据流组织相关的一部分属性抽象是 AddLocalExchanges 进行本地数据流规划时重点关注的信息。3.7 数据流属性约束体系的整体流程Operator 根据自身语义定义属性需求 ↓ Visitor 将这些规则编码为 Planner 可执行逻辑 ↓ StreamPreferredProperties 描述 Need ↓ StreamProperties 描述 Have ↓ planAndEnforce() 执行规划并比较 Need 与 Have ↓ 匹配直接复用现有数据流 不匹配插入 Local Exchange ↓ 重新推导输出属性四、Operator 的数据流属性语义前面介绍了 Presto 如何使用 StreamProperties 描述当前数据流状态以及使用 StreamPreferredProperties 表达算子的输入需求。但是Optimizer 如何知道一个算子需要什么样的数据流又如何知道其执行完成后数据流会具有什么属性答案是每个 Operator 都定义了自身的数据流属性语义Dataflow Property Semantics用于描述输入约束和输出属性。这些语义是物理规划阶段进行属性推导和 Exchange 插入的基础。4.1 输入需求与输出属性对于一个 OperatorOptimizer 需要关注两个问题1. 对输入数据流的需求和建议Input Requirement Preferred当前算子需要以及建议上游提供什么样的数据流。例如某些算子要求数据按照特定字段分区某些算子要求单流执行某些算子要求输入满足排序条件。这些需求最终通过 StreamPreferredProperties 来定义用于指导子节点规划并执行数据流属性约束。2. 实际输出数据流的属性Output Property当前算子被规划并执行之后下游节点可以依赖的数据流特征。例如Filter 通常保持输入的数据分布Exchange 会改变数据分布Project 会根据字段映射传播已有属性某些算子会丢失部分输入属性。这些结果会通过 StreamProperties 来定义用于后续节点的属性推导。4.2 算子语义与物理规划的关系需要注意AddLocalExchanges 并不关心算子的具体执行逻辑而只关心算子对数据流组织形式的要求。例如Aggregation 在执行层面可能涉及Hash AggregationStreaming AggregationPartial / Final Aggregation。但在数据流规划阶段Planner 只需要知道grouping key 是否满足数据聚合要求Join 也不关心具体 Join Operator 如何执行而关注输入数据分布是否满足 Join 策略要求因此Operator 属性语义提供的是一种物理约束抽象而不是执行实现描述。4.3 Operator 属性语义与 Visitor在源码实现中Operator 的属性语义通过 Visitor 模式进行编码。Visitor 模式的本质其本身并不是优化算法而是将不同 PlanNode 对输入数据流属性的要求以及它们产生的输出属性编码为 Optimizer 可以理解和操作的形式。真正负责比较 Need 与 Have、插入 Exchange、重新推导属性的是 planAndEnforce() 方法。Visitor 定义的是“属性语义”而 planAndEnforce() 执行的是“满足属性约束”。说明Operator 属性语义与具体执行策略密切相关并且仍处于持续演进过程中不同版本之间可能存在差异。因此本文只讨论其在物理规划中的抽象作用即算子如何声明数据流约束以及 Planner 如何基于这些约束完成属性推导和 Exchange 调整。对于具体算子的当前实现细节建议直接参考对应版本源码。五、planAndEnforce —— 属性约束满足的核心流程前面的章节介绍了数据流不匹配问题、属性约束机制、属性描述语言、算子属性语义等。现在需要一个过程将这些属性语义转换为满足约束的物理计划。planAndEnforce() 正是完成这一过程的核心方法。它接收父节点的属性需求递归处理子节点比较需求与实际属性的差距在必要时插入 Local Exchange 重组数据流最终确保每个节点都获得满足执行要求的数据流。5.1 planAndEnforce 的职责planAndEnforce() 的核心职责概括为六步Step 1: 接收需求 ── 从父节点获取 Required 和 Preferred 属性需求Step 2: 递归规划 ── 使用 Preferred 建议对每个子节点进行递归规划传递裁剪后的需求Step 3: 收集属性 ── 收集子节点规划处理后的实际数据流属性Step 4: 比较判断 ── 检查 Have 是否满足 Need (Required)Step 5: 修正实施 ── 不满足时插入 Local Exchange重新推导属性Step 6: 返回属性 —— 将当前节点输出属性作为父节点的 Have其核心思想是不断将父节点的建议需求preferred向下传播并在数据流无法满足强制要求required时通过 Exchange 主动修正。方法签名如下privatePlanWithPropertiesplanAndEnforce(PlanNodenode,StreamPreferredPropertiesrequired,// 父节点必须满足的需求StreamPreferredPropertiespreferred);// 父节点偏好的需求优化提示PlanWithProperties 返回值包含处理后的节点及其产出的实际数据流属性。Required vs PreferredRequired 是硬约束不满足时 Planner 必须通过 Exchange 等机制进行修正。如 Aggregation 必须在 grouping keys 上分区。Preferred 是软约束满足可提升性能不满足也正确。如某些节点偏向单流处理。属性需求裁剪在递归规划过程中父节点的需求不会直接传递给所有子节点。而是会根据子节点实际输出的数据列对需求进行裁剪Join Need: HASH(a, b) / \ Left Child Right Child 输出: a, c 输出: b, d ↓ 裁剪后传递 ↓ 裁剪后传递 Need: HASH(a) Need: HASH(b)确保每个子节点只接收与自己相关的需求。5.2 Need vs HaveStreamPreferredProperties 如何判断属性满足递归规划完成后Planner 会比较Need: 父节点要求的数据流属性Have: 子节点实际产生的数据流属性属性匹配本质上是判断已有数据流属性是否足够支持当前算子的执行要求。例如更强的数据流属性可以满足较弱的数据流需求典型情况例如SINGLE ORDERED 可以满足 SINGLE HASH(a) 可以满足 HASH(a, b) --分区包含关系isPartitionedOn详见上面3.5节的说明综上所述isSatisfiedBy() 并不是简单比较 distribution 是否完全相等而是根据数据流属性的强弱关系判断是否满足。在属性匹配过程中Planner 会根据 StreamProperties 的 distribution、partitioning 和 ordering 判断 Child Stream Properties 是否满足 Parent Requirement。如果不能满足则进入 enforce 阶段。5.3 enforce通过 Local Exchange 修正属性在 enforce 阶段Planner 根据缺失的属性类型插入对应的 Local Exchange。如下所示Exchange 类型属性变化典型场景GATHER任意 → SINGLE全局排序、最终汇聚HASH REPARTITION任意 → FIXED partitioningAggregation、JoinROUND ROBIN任意 → MULTIPLE并行负载均衡例如Aggregation 需要HASH(a, b)但输入数据流为MULTIPLE则 HASH Exchange on (a, b) MULTIPLE ----------------------------- FIXED(a, b)插入之后 Aggregation 即可正确执行。5.4 Exchange 后重新推导属性由于 Exchange 改变了数据流组织形式因此插入 Exchange 后必须重新计算属性。例如插入前数据流分布为 MULTIPLE 或 FIXED插入后数据流分布变为了 SINGLE 插入前数据流分布为 SINGLE 或 MULTIPLE插入厚数据流分布变为了 FIXED(a, b)因此在 Local Exchange 插入之后我们需要重新执行属性推导derive properties以得到新的数据流属性StreamProperties。属性推导是自底向上的子节点属性会向上传播并决定父节点属性。planAndEnforce() 返回前通过 checkState() 验证所有硬约束均已满足确保计划执行时不会因数据流属性不匹配而失败。总结planAndEnforce() 通过递归规划 → 属性比较 → 插入 Exchange → 重新推导的闭环将算子属性语义转化为满足约束的物理执行计划是 AddLocalExchanges 优化规则得以实施的核心引擎。六、完整执行流程示例Local Exchange 如何完善执行计划需要注意的是Presto 的物理计划生成通常包含两个层面的数据重组Remote Exchange解决 Task/Stage 之间的数据传输和重新分区问题Local Exchange解决单个 Task 内部多个 Pipeline 之间的数据组织问题。本示例重点关注 AddLocalExchanges 如何基于已有物理计划属性调整 Local 数据流。Remote Exchange 仅作为背景展示不展开分析。以如下的分组聚合语句作为示例SELECTa,b,COUNT(*)FROMtGROUPBYa,b;经过前序物理规划后计划已经包含跨 Task 数据重分布Output | RemoteExchange(GATHER) | Aggregation | RemoteExchange(MULTIPLE) | TableScan这里RemoteExchange(MULTIPLE) 负责 Stage 间的数据重新分区。但是在每个 Task 内部仍然需要确定各 Pipeline 如何消费这些数据。AddLocalExchanges 处理 Aggregation 的实际行为如下所示初始计划片段Aggregation | RemoteExchange (MULTIPLE)planAndEnforce() 开始执行对于 Aggregation 节点处理逻辑为递归处理子节点 RemoteExchange规划并获取 RemoteExchange 的输出数据流分布类型: MULTIPLE比较 Aggregation Need vs RemoteExchange HaveNeed: FIXED on (a,b)Have: MULTIPLE, 分区未知不匹配调用 enforce()插入 LocalExchange (HASH on a,b)修正后计划Output | RemoteExchange(GATHER) | Aggregation │ LocalExchange (HASH on a,b) ← 插入的 Local Exchange │ RemoteExchange(MULTIPLE) | TableScan重新推导属性LocalExchange 后: FIXED on (a,b)Aggregation 后: FIXED on (a,b)继承了分区特性Output 检查: 可以接受多流排序敏感允许保留校验最终计划满足所有需求。七、总结AddLocalExchanges 本质上解决的是查询执行计划生成过程中的一个核心问题如何将逻辑算子所表达的计算语义转换为满足执行要求的数据流组织方式。在分布式查询引擎中不同算子对输入数据流具有不同约束例如 Aggregation 需要满足 grouping key 的分区要求Join 需要满足关联键的数据布局要求Sort 和 TableWriter 也分别依赖特定的数据流属性。这些约束并不是由算子自身完成协调而是由 Optimizer 通过属性抽象和约束满足机制统一处理。AddLocalExchanges 通过 Visitor 定义算子属性语义利用 StreamPreferredProperties 和 StreamProperties 描述需求与实际状态并由 planAndEnforce() 递归的完成属性传播、需求匹配以及必要的 Exchange 插入。从设计思想上看AddLocalExchanges 体现了现代查询优化器中的 Property Enforcement 模型Optimizer 不直接决定执行细节而是通过物理属性描述执行约束再由规划规则自动调整数据流使最终执行计划满足所有算子要求。因此Local Exchange 并不是简单的数据搬运节点而是连接算子语义与物理执行的数据流控制机制。它将逻辑计划中的计算需求逐步转换为满足并行执行要求的数据流结构是 Presto 从逻辑计划走向可执行物理计划的重要阶段。作者王冬PrestoDB Global Committer | Presto Iceberg Code OwnerGitHub: https://github.com/hantangwangdEmail: mingwbdgmail.com欢迎就相关技术问题与实践经验进行交流与探讨本文章同步发表于https://hantangwangd.github.io/zh/posts/2026-07-27-add-local-exchanges.html