Presto 查询引擎内核详解：Spill-to-Disk 执行

Presto Spill-to-Disk Execution从内存压力到算子溢写在分布式查询引擎中内存通常是决定查询能够处理多大数据规模的关键资源。Hash Aggregation、Hash Join、Sort、Window 等算子都可能随着输入数据增长而维护越来越大的中间状态。当这些状态超过 Worker 可用内存时如果查询引擎只能继续申请内存最终就只能以 OOM 或 Query Failure 结束。Spill-to-Disk 是解决这一问题的重要机制当内存压力达到一定程度时将部分可以外置的中间状态写入本地磁盘释放内存并让查询继续执行。但 Presto 的 Spill 并不是简单的“把 Operator 的内存状态序列化到磁盘之后再全部加载回来”。它实际上是一套由 Memory Management、Memory Revocation、Driver Scheduling 和 Operator-specific External-Memory Execution 共同构成的执行机制。从整体上看可以将它概括为Memory Pressure ↓ Memory Revocation ↓ Driver Execution ↓ Operator-specific Spill ↓ Memory Reclamation ↓ Execution Resumes本文从 Worker 的内存管理开始逐步分析 Presto 如何发现内存压力、如何选择需要 Spill 的 Operator、为什么必须主动唤醒 Driver以及 Operator 如何完成实际的磁盘溢写最终形成完整的执行闭环。1. 为什么需要 Spill从 Worker 内存压力说起1.1 查询执行中的内存为什么会不断增长很多查询算子都不是简单地读取一条数据、处理一条数据然后立即输出。例如对于如下的聚合查询SELECT key, count(*) FROM table GROUP BY key;Hash Aggregation 需要维护一个类似这样的结构Hash Table ┌────────────────────────────────┐ │ │ │ (key₁, aggregation state₁) │ │ (key₂, aggregation state₂) │ │ (key₃, aggregation state₃) │ │ (key₄, aggregation state₄) │ │ ... │ │ │ └────────────────────────────────┘随着输入数据不断到达不同的 key 越来越多Hash Table 也不断增长。Hash Join 也类似Build Side 通常需要首先建立 Hash Table然后再与 Probe Side 执行连接。Sort 则需要保留需要排序的数据Window Operator 也可能需要保存大量中间状态。因此一个 Query 的执行过程实际上可能持续消耗大量内存Input ↓ Operator ↓ Intermediate State ↓ Memory Usage ↑当 Query 数据规模扩大以后单个 Operator 或整个 Worker 的内存使用都可能达到很高水平。1.2 为什么不能简单地继续申请内存假设一个 Worker 总共只有 100 GB 内存。现在多个 Query 同时执行Query A → 40 GB Query B → 30 GB Query C → 20 GB ---------------- Total 90 GB此时某个 Hash Aggregation 又需要 20 GB90 GB 20 GB 110 GB如果查询引擎简单地允许继续申请那么最终可能因为无法满足内存需求而触发 OOM甚至导致 Worker 进程失败而如果仅仅通过严格的 Query 级内存上限来限制查询一旦查询的内存需求超过该上限就直接失败对于需要处理大规模数据的查询来说同样不够友好也限制了查询引擎处理超出单机内存规模的数据集的能力。对于一个成熟的查询引擎而言更合理的方式是让部分不需要持续驻留在内存中的执行状态能够被外置。例如Hash Aggregation 的部分中间聚合状态可以被写入磁盘之后再通过特定的外部内存算法重新处理这些磁盘上的中间结果最终生成查询结果。这就是 Spill-to-Disk 的基本思想。1.3 Presto Spill 的核心并不是“保存 Operator 状态”Presto 的 Spill 并非某种通用的 Operator Suspension 机制其更接近Operator / \ Memory Disk │ │ current state spilled state \ / \ / continue execution在该机制中Operator 并没有整体被“暂停并保存”。相反Operator 会将部分中间数据转化为磁盘上的 spill data同时保留继续执行所需的必要状态并在内存释放完成后继续推进后续计算。因此从执行模型上说Presto Spill 更准确地描述为Operator-specific external-memory execution即由具体 Operator 自己实现的外部内存执行算法。这也是理解 Presto Spill 后续所有机制的基础。2. Presto 的内存管理Memory 如何控制 Operator 执行理解 Spill首先必须理解 Presto 的 Memory Management。这部分的详情可以参见我写的另一篇文章《Presto 查询引擎内核详解集群资源管理机制解析》。2.1 Operator 的内存申请与 Worker MemoryPoolOperator 在执行过程中需要申请和释放内存。这些内存申请通过 Presto 的层次化 Memory Context 进行管理并最终汇聚到 Worker 的 MemoryPool。从概念上可以理解为Operator ↓ OperatorContext ↓ MemoryTrackingContext │ ┌──────────┼──────────┐ ▼ ▼ ▼ Local Aggregated Memory Memory ... Context Context └──────────┼──────────┘ ↓ Leveled Memory Accounting ↓ MemoryPool这种层次化设计使得 Operator 的局部内存需求能够最终纳入 Worker 全局的内存管理体系。当多个 Operator 和 Query 的内存使用共同导致 MemoryPool 面临压力时上层 Memory Management 机制便可以进一步进行 Memory Blocking 或 Memory Revocation。例如在算子内部当使用了一定数量的内存之后执行memoryContext.setBytes(bytes);这个操作会逐层向上传递并最终影响 Worker MemoryPool 中的内存状态。因此Operator 的一次普通内存申请实际上已经与整个 Worker 的内存管理体系建立了联系。2.2 Memory 不仅仅是 Accounting如果 Memory Context 只是记录Operator A 10 GB Operator B 20 GB Operator C 30 GB那么它只是一个 accounting system。但 Presto 的 Memory Management 更进一步Memory Management 会直接参与执行流控。当 Operator 申请内存时如果 MemoryPool 能够满足则后续执行继续进行reserve ↓ Memory available ↓ success ↓ Operator continues但是如果 MemoryPool 无法满足则会通过返回的 blocked future 来阻塞 Operator 的后续执行reserve ↓ Memory insufficient ↓ blocked Future ↓ Operator waits于是 MemoryPool 的状态就可以反向影响到 Operator 的执行了。2.3 Memory Blocked Future可以将这个 Future 理解为“当前内存不足等有人释放内存后再继续”。例如Operator │ request 10 MB ▼ MemoryPool │ insufficient ▼ blocked FutureOperatorContext 持有与之相关的内存等待状态memoryFuture和revocableMemoryFuture分别代表了在申请 user memory 和 revocable memory 时由于内存不足而导致的阻塞状态。Driver 在判断某个 Operator 是否处于阻塞状态以及 Driver 本身是否整体处于阻塞状态时都会去检查 OperatorContext 中的这种状态。这也说明 Memory Pool 已经进入了 Query Execution 的控制路径。2.4 Revocable Memory对于 Spill 来说更重要的是 Revocable Memory。但并不是所有 Operator 使用的内存都可以随意回收。有些内存一旦释放就会破坏当前 Operator 的执行状态。另外一些内存可以通过 Spill 转化为磁盘上的中间数据因此具有可回收性。这其中的后者就是 Revocable Memory。可以粗略理解为Worker Memory ├── User / Non-revocable Memory │ └── 一般不能直接要求 Operator 释放 │ └── Revocable Memory └── 执行状态可以被 externalize因此可以通过 Spill reclaim 的内存所以 Memory Revocation 的目标并不是“随便找一个 Operator让它释放内存。”而是找到持有可回收状态的 Operator让它通过 Spill 主动释放 Revocable Memory并以 external-memory execution 的模式继续执行。3. Memory Revocation内存压力如何转化为 Spill 请求3.1 Memory Pressure随着多个 Operator 持续执行Memory Reservation ↑ ↓ MemoryPool Usage ↑当内存使用达到需要 Revocation 的条件时Presto 的 MemoryRevokingScheduler 会介入。MemoryRevokingScheduler 会将 onMemoryReserved() 监听逻辑注册到其管理的每一个 MemoryPool 上。因此当 Operator 的内存 reservation 导致 MemoryPool 状态发生变化时相关状态变化会进入 MemoryRevokingScheduler 的监听路径由 MemoryRevokingScheduler 判断当前是否需要启动 Revocation。registers listener ┌──────────────────────────┐ │ │ ▼ │ ┌───────────────────┐ ┌──────────┴─────────────┐ │ MemoryPool │────►│ MemoryRevokingScheduler│ └───────────────────┘ └───────────┬────────────┘ ▲ │ │ reservation change │ evaluate pressure │ ▼ Operator reserve Trigger Revocation3.2 MemoryRevokingScheduler 做什么MemoryRevokingScheduler 的核心职责并不是实际执行 Spill。它主要解决的是现在是否需要释放内存如果需要的话应该让哪些 Operator 来释放因此可以把它理解成一个 Memory-Reclamation Coordinator。流程大致是┌──────────────────────────┐ │ MemoryRevokingScheduler │ │ │ Memory Pressure ─► Evaluate pressure │ │ ↓ │ │ Calculate required │ │ revocation │ │ ↓ │ │ Select suitable operators│ └──────────┬───────────────┘ │ request revocation ▼ Selected Operators这里的职责边界非常重要Scheduler 决定“谁应该释放内存”而 Operator 本身决定“如何释放这些内存”。3.3 requestMemoryRevoking()当 Scheduler 找到合适的 Operator 后会调用operatorContext.requestMemoryRevoking();这会在 OperatorContext 中设置 Memory Revocation Request 状态。在逻辑上可以将其理解为“这个 Operator 下一次获得执行机会时需要先执行 Memory Revocation。”于是一个看似简单的问题随之出现既然 Scheduler 已经确定了需要释放内存的 Operator为什么不直接调用 Operator 的 startMemoryRevoke()而是只设置一个 Revocation Request答案涉及 Presto 的核心执行模型Memory Revocation 并不是由 MemoryRevokingScheduler 直接驱动 Operator 执行而是需要回到 Driver 的统一调度执行路径由 Driver 在合适的执行时机处理 Revocation Request。这也就引出了下一章要讨论的问题Revocation Request 是如何从 MemoryRevokingScheduler 进入 Driver并最终驱动 Operator 执行实际的 Spill4. Driver 如何驱动 Memory Revocation4.1 一个看似矛盾的问题假设 Operator 当前因为内存不足而 blocked而现在 MemoryRevokingScheduler 又要求“请你执行 Spill释放内存。”可是此时 Operator 甚至整个 Driver 可能都已经 blocked 了如果它完全依赖正常执行流程Driver blocked ↓ 被调度到了也无法运行 ↓ 不能驱动 Operator 执行 Spill ↓ 不能释放 Memory ↓ 其他 Operators 也无法获得 Memory于是就可能进入一种无法破解的整体阻塞状态。Presto 对此的解决方式是Memory Revocation Request 会主动唤醒 Driver。4.2 Driver 的 driverBlockedFutureDriver 中维护AtomicReferenceSettableFuture? driverBlockedFuture它代表 Driver 当前是否处于 blocked 状态。在 Driver 初始化时会为其持有的每一个 active Operators 注册 Memory Revocation Listener。该 listener 监听每一个算子上的 requestMemoryRevoking 动作并在该动作发生的时候执行driverBlockedFuture.set(null)以取消当前 Driver 的阻塞状态如下图所示OperatorContext │ │ requestMemoryRevoking() ▼ Memory Revocation Listener │ ▼ driverBlockedFuture.set(null) │ ▼ Driver becomes runnable因此Memory Revocation Request 会打破 Driver 原本因为某些等待条件而形成的 blocked 状态使其重新进入可运行/可调度路径从而有机会执行 Memory Revoke。注意这里一个非常容易产生的误解Driver 被唤醒并不意味着内存已经恢复。恰恰相反Driver 被唤醒是为了让它驱动具体的 Operator 执行释放内存的动作。4.3 Driver processFor() 中的判断Driver 每次获得执行机会时会首先检查自己的 blocked 状态。如果 Driver 当前确实因为某个 Future 等待driverBlockedFuture not completed那么它不会继续执行。而当 Memory Revocation Request 触发时设置了driverBlockedFuture.set(null)Driver 就可以在下一次调度时继续执行。于是Memory Pressure ↓ requestMemoryRevoking() ↓ wake Driver ↓ Driver executes ↓ handleMemoryRevoke()4.4 handleMemoryRevoke()Driver 获得执行机会后首先会执行方法handleMemoryRevoke()。其中会遍历 active Operators对于满足如下条件的 Operatornot already revoking isMemoryRevokingRequested()调用如下方法以启动异步的溢写及内存回收行为operator.startMemoryRevoke();返回的 Future 会维护在 revokingOperators 中。于是整个状态变成requestMemoryRevoking() ↓ Driver awakened ↓ handleMemoryRevoke() ↓ startMemoryRevoke() ↓ spill in progress ↓ 溢出执行的状态维护在 revokingOperators4.5 为什么需要 revokingOperators因为 Spill 是异步执行的。如下所示startMemoryRevoke() ↓ serialize ↓ write disk ↓ flush ↓ Future completeDriver 不能在 Spill 尚未完成时继续正常操作这个 Operator。因此在 Future 完成之前将 Operator 维护在 revokingOperators 中。这样一来Driver 会把这个 Operator 视为处于 Memory Revocation 状态。然后在其checkOperatorFinishedRevoking()方法中检查并确认 Spill 成功之后将 Operator 从 revokingOperators 里移除并调用finishMemoryRevoke()最后清除 memoryRevokingRequest 标志整个过程就完成了。4.6 几种 Future 的职责必须区分这一套代码中涉及到的异步操作 Future 很多很容易混淆。可以用下面的方式理解Future 类型含义driverBlockedFuture表示Driver当前是否整个处于阻塞状态memoryFuture表示Operator是否因为普通内存User Memory不足而等待revocableMemoryFuture表示Operator是否因为可撤销内存Revocable Memory不足而等待memory revoke/spill Future表示当前内存回收Memory Revocation或溢出Spill操作是否完成它们解决的是不同层次的问题Memory Future: “我没有内存不能继续。”Revoke Future: “我要释放内存但 Spill 还没完成。”Driver Future: “当前 Driver 是否处于阻塞状态。”理解这一点之后Presto 的整个控制流就会清晰很多。5. Operator-Specific SpillPresto 如何把算子状态溢写到磁盘5.1 Spill 并不是通用的 Operator SnapshotPresto 为 Operator 定义了如下的接口方法ListenableFuture? startMemoryRevoke(); void finishMemoryRevoke();但是这两个接口本身并没有规定“所有 Operator 应该如何执行 Spill。”换句话说Presto 的 Memory Revocation API 抽象的是“释放可撤销内存”的生命周期而不是“如何持久化 Operator State”的数据模型。具体实现由 Operator 自己负责即HashAggregationOperator - Aggregation-specific SpillOrderByOperator - Sort-specific SpillHashBuilderOperator - Join-specific Spill…因此Presto Spill 的一个核心设计思想是Spilling is operator-specific.5.2 为什么不做一个通用的 Spill Framework因为不同 Operator 的中间状态完全不同。对于 Hash Aggregation Hash Table对于 Sort Rows / sorted runs对于 Hash Join Build-side hash structures / partitions这些数据不仅结构不同而且最终如何重新合并的算法也各有不同。例如Sort 可以采用External Merge Sort 算法而 Hash Aggregation 则需要Re-aggregate spilled data。这些差异并不是简单的实现细节而是不同 Operator 的核心执行算法本身因此很难由一个通用 Spill Framework 完全抽象。因此Presto 选择了另一条路线由 Operator 自己实现适合其执行逻辑的数据外置与恢复算法。5.3 Spill 后 Operator 如何继续推进这是理解 Presto Spill 的关键。在 Presto 中Spill 并不是Memory - Disk - Operator suspended而是Operator / \ Memory Disk │ │ current state spilled data │ │ └─────┬──────┘ ↓ continue executionSpill 并不意味着整个 Operator 状态被序列化写出并暂停。对于支持增量 Spill 的 OperatorstartMemoryRevoke() 触发其将当前可外置的中间数据转换为 Spill Data并释放相应的 Revocable Memory。Operator 随后可以继续处理新的输入直至产生最终输出结果。因此一个 Operator 在运行过程中可以同时维护In-memory intermediate state Disk-resident intermediate data甚至可以多次执行 SpillMemory ↓ Spill #1 - Continue ↓ Memory grows again ↓ Spill #2 - Continue ↓ ......最终算子的中间状态包含了内存中和溢写到磁盘上的状态Memory State Spill File #1 Spill File #2 Spill File #3 ...这是一种典型的 External-Memory Execution 模式。5.4 Spill Data 如何参与后续计算最终产生输出时Operator 不能只处理当前内存中的状态例如 Hash Table。它还必须处理之前 Spill 到磁盘的中间数据。因此最终过程可以抽象为Spill Data Memory State ↓ Operator-specific processing ├── External Merge ├── Re-aggregate ├── Partition Processing └── ...这正是 Operator-specific External-Memory Algorithm 的体现。5.5 HashAggregationOperator以 Hash Aggregation 为例。当其正常执行时Input ↓ HashAggregationOperator ↓ InMemoryHashAggregationBuilder ↓ Hash Table随着后续输入不断到达Hash Table 变得越来越大占用的内存也越来越大。当 MemoryRevokingScheduler 要求它释放内存时MemoryRevokingScheduler │ requestMemoryRevoking() ▼ HashAggregationOperator │ startMemoryRevoke() ▼ 启动异步 Spill │ ├──────────────► Spiller │ │ │ ▼ │ 以 group key 有序的方式 │ 写出 Aggregation State ▼ 释放可撤销内存 │ ▼ 清理 / 重置内存中的 Aggregation State此时 Hash Aggregation 的逻辑状态就变成HashAggregationBuilder │ ├── In-memory aggregation state │ └── Spilled intermediate data ├── spill file #1 ├── spill file #2 └── ...溢写完成之后 Operator 可以继续接收新的输入并且有可能随着内存的不断增加再次触发 requestMemoryRevoking()。注意发生 Spill 后内存中的聚合状态不再能够直接作为最终聚合结果输出。由于相同的 Group By Key 可能同时存在于内存中的聚合状态以及已经 Spill 到磁盘的中间聚合结果中因此最终需要对这些 intermediate aggregation states 再次进行合并聚合。为此在执行 Spill 时Presto 会将当前内存中的聚合状态转换为适合外部处理的 intermediate/partial aggregation representation并通过 buildHashSortedResult() 等逻辑构造有序的中间结果然后写入 Spill 文件。最终生成结果时将内存中的以及多个 Spill 文件中的有序中间聚合结果进行 Merge并对属于相同 Group By Key 的 aggregation states 执行最终合并。通过这种方式可以流式的产生最终聚合结果从而避免在内存中维护一个覆盖全部数据的巨大 Hash Table。最后需要强调的是HashAggregation Spill 的关键并不是单纯的把 Hash Table 写到磁盘而是通过 buildHashSortedResult() 将内存中的 Hash-based aggregation state 转换成有序的 intermediate representation使后续能够通过 External Merge 和流式 Re-aggregation 在有限内存下完成最终聚合。5.6 不同 Operator 的 Spill 算法不同前面的 HashAggregation 只是 Operator-Specific Spill 的一个具体例子。Presto 没有规定所有 Operator 必须采用统一的 Spill Algorithm而是由具体的 Operator 根据自身的数据结构、计算语义以及最终结果的生成方式选择合适的 External-Memory Processing 方案。因此不同 Operator 的 Spill 行为可能存在很大的差异。例如OrderBy 的核心问题是如何将无法完全放入内存的数据转换为多个有序数据集并最终通过 External Merge 产生有序输出Input ↓ In-memory sorting ↓ sorted run ↓ spill ↓ multiple sorted runs ↓ external merge ↓ output而 Hash Join 通常通过 Partitioning 将数据划分成多个可以独立处理的 Partition当内存不足时将部分 Partition 外置之后逐个处理这些 PartitionBuild data ↓ partition ↓ spill partitions ↓ process partitions ↓ join因此不能简单地说“Presto 可以把 Operator 的 state 溢写到磁盘。”更准确的说法是Presto 将 Operator 中可以外置的中间数据转换成磁盘上的 Spill Data并由 Operator 自己负责后续的 External-Memory Processing。6. Spill 完成后内存如何真正释放并恢复执行前面的流程描述的是“如何让 Operator 执行 Spill”现在还需要解决另外一个问题Spill 完成之后释放出来的内存如何让此前因内存不足而阻塞的 Operator 恢复执行6.1 finishMemoryRevoke()Driver 被调度执行时会在其checkOperatorFinishedRevoking()方法中检查 Operator 的 Spill future 是否完成。在确认 startMemoryRevoke() 返回的 future 完成以后调用该算子的finishMemoryRevoke()这个阶段主要完成本次 Memory Revocation 的收尾工作。其中最重要的一步是更新 Operator 的 Memory Accounting释放已经被溢写到外部存储的 Revocable Memory。于是形成如下的反馈Operator │ finishMemoryRevoke() ▼ Update Memory Accounting │ ▼ Revocable Memory Reservation ↓ │ ▼ MemoryPool.free() │ ▼ Available Memory ↑Memory Revocation 真正完成的标志不是 Spill File 已经写入磁盘而是 Operator 已经完成 Memory Accounting 的更新使对应的 Revocable Memory Reservation 被释放。6.2 内存释放如何解除被阻塞的 Operator之前内存不足时MemoryPool ↓ Available Memory insufficient ↓ Memory Future not completed ↓ Operator blocked on memory现在当 MemoryPool 的可用内存因为 Revoke 而增加后之前等待内存资源的 Operator 所关联的 Memory Future 得以重新满足完成条件从而解除其内存阻塞MemoryPool.free() ↓ free bytes ↑ ↓ Operator memory future completed ↓ Operator unblocked ↓ Driver runnable ↓ continue execution7. 总结Presto Spill 是一个 Memory-Driven Execution Feedback Loop如果只从“数据落盘”的角度看Spill 似乎只应该是Memory → Disk。但从 Worker Execution Engine 的角度看Spill 的核心并不是数据落盘而是通过 Operator-specific External-Memory Processing 将内存压力转化为可执行的资源回收动作并最终恢复系统的执行能力。MemoryRevokingScheduler 负责发现内存压力并选择需要释放 Revocable Memory 的 OperatorDriver 负责响应 Revocation Request并获得执行机会以驱动 Operator 执行 startMemoryRevoke()最终由 Operator 根据自身的 External-Memory Algorithm 完成状态外置并在释放内存后继续执行。因此Presto Spill 最准确的架构抽象并不是“Presto 在内存不足时把数据写到磁盘。”而是Presto 实现了一套由内存压力驱动的外部内存执行机制通过将算子中可外置的中间数据转换为 Spill Data 并释放相应的 Revocable Memory使查询能够在有限内存下继续执行。这也是理解 Presto Spill 最重要的一点Spill 并不是一个独立的磁盘 I/O 功能而是整个执行引擎围绕内存压力形成的一套反馈控制机制。而 Operator-specific External-Memory Execution则是这套机制能够真正落地的关键Memory Management 决定“什么时候需要释放”Driver 负责“如何驱动释放”Operator 决定“如何将自己的中间数据外置并继续执行”。三者共同构成从 Memory Pressure → Memory Revocation → Spill → Memory Release → Execution Recovery 的完整闭环。作者王冬PrestoDB Committer | Presto Iceberg Code OwnerGitHub: https://github.com/hantangwangdEmail: mingwbdgmail.com本文章同步发表于https://hantangwangd.github.io/zh/posts/2026-08-27-spill-to-disk-execution.html