深入 Impala 查询引擎：一条 SQL 如何变成分布式流水线

Impala 查询引擎 分布式执行 SQL 源码分析

Impala 查询快，不是因为它省略了 SQL 优化，也不是简单地“绕过 MapReduce”。真正的区别在执行形态：Daemon 常驻、元数据常驻、查询被切成并行 Fragment，数据通过 Exchange 在节点间流动，Scan、Join 和 Aggregate 尽可能组成流水线，并由 Runtime Filter 在运行期缩小扫描范围。

本文以 Impala 4.5.0 为基线，从一条 SQL 进入 impalad 开始，追到 RowBatch 穿过执行树并回到客户端。源码路径均相对于 Impala 仓库。

先建立进程视图

Impala 集群的核心角色不是传统意义上的“Master + Worker”。每个 impalad 都能接收查询，其中接收客户端请求的节点在本次查询中成为 Coordinator，其他节点执行它分配的 Fragment Instance。

flowchart LR
    C[客户端] --> Q[Coordinator impalad]
    CAT[Catalogd] -->|元数据更新| Q
    ST[Statestored] -->|成员与 Topic| Q
    Q -->|ExecQueryFInstances| E1[Executor 1]
    Q -->|ExecQueryFInstances| E2[Executor 2]
    Q -->|ExecQueryFInstances| E3[Executor 3]
    E1 -->|状态与 Profile| Q
    E2 -->|状态与 Profile| Q
    E3 -->|状态与 Profile| Q
    Q -->|FetchResults| C
客户端
  → Coordinator impalad
      ├── Frontend：解析、分析、计划
      ├── Scheduler：选择执行位置
      ├── Coordinator：启动并跟踪后端实例
      └── Result Sink：向客户端返回结果

Executor impalad × N
  ├── Scan / Join / Aggregate / Sort
  ├── DataStreamSender
  └── Runtime Profile

Catalogd
  └── 表、分区、文件、统计信息与元数据版本

Statestored
  └── 成员状态、心跳和主题更新

Coordinator 与 Executor 是每条查询里的职责，不一定对应两套不同二进制。生产环境可以通过角色配置让部分节点主要承担协调工作。

第一步：SQL 进入 Frontend

客户端通过 Beeswax 或 HiveServer2 协议提交 SQL。服务层建立 QueryContext，随后进入 Java Frontend。主要入口可以从：

  • fe/src/main/java/org/apache/impala/service/Frontend.java
  • fe/src/main/java/org/apache/impala/analysis/
  • fe/src/main/java/org/apache/impala/planner/Planner.java
  • fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java

继续阅读。

Frontend 并不是只调用一次 Parser。它需要完成：

  1. 语法解析，生成 Statement AST；
  2. 名称解析，绑定数据库、表、列和函数；
  3. 类型推导与隐式转换；
  4. 权限检查；
  5. 常量折叠与表达式改写；
  6. 子查询改写、谓词传播和等价类分析；
  7. 生成单机逻辑计划；
  8. 依据数据位置切成分布式 Fragment；
  9. 估算资源并序列化为 Thrift 请求。

分析阶段失败的 SQL 不会进入后端执行。诸如列歧义、类型不兼容、聚合语义错误和权限拒绝，都在这里尽早终止。

Planner 的输入并不只是 SQL

优化器需要 Catalog 中的表结构、分区、文件块位置、列统计和行数估计。SQL 文本相同，元数据不同，计划也会不同。

以 Join 为例，Planner 要判断：

  • 哪一侧作为 Build，哪一侧作为 Probe；
  • 小表是否广播到每个执行节点；
  • 大表是否按 Join Key 重新分区；
  • Join 顺序怎样降低中间结果；
  • 是否能生成 Runtime Filter；
  • 每个节点和 Fragment 需要多少内存。

代价估算依赖统计信息。缺少 NDV、行数或列宽时，优化器只能使用默认估计，广播本不该广播的大表，或者采用代价更高的 Join 顺序。很多“Impala 突然很慢”的根因不在执行器，而在过期统计信息让 Planner 看错了数据规模。

从 PlanNode 到 PlanFragment

单机计划是一棵 PlanNode 树，例如：

TOP-N
  └── AGGREGATE (merge)
      └── EXCHANGE
          └── AGGREGATE (partial)
              └── HASH JOIN
                  ├── HDFS SCAN fact
                  └── HDFS SCAN dim

分布式规划器在需要改变数据分布的地方插入 Exchange，并把树切成多个 PlanFragment。

Fragment 不是一台机器上的完整查询，而是一段可以并行实例化的执行子图。一个 Fragment 可以在多台 Executor 上各运行多个 Instance。ExchangeNode 是接收端，子 Fragment 末端的 DataStreamSender 是发送端。

常见数据分区方式包括：

  • UNPARTITIONED：单实例执行，常用于最终汇总；
  • HASH_PARTITIONED：按表达式哈希，把相同 Key 送到同一接收端；
  • RANDOM：均匀分散数据；
  • BROADCAST：把 Build Side 复制到多个接收端；
  • DIRECTED：按指定目标路由。

源码中 DistributedPlanner 的 create*Fragment() 方法和 ExchangeNode 展示了这些边界如何形成。

Scheduler：把逻辑实例落到具体主机

Fragment 只说明并行关系，还没有决定在哪运行。调度阶段会结合：

  • HDFS 文件块或数据源位置；
  • 当前可用 Executor；
  • Scan Range 数量；
  • 节点负载与资源池；
  • Fragment 的并行度配置；
  • Coordinator 与数据节点距离。

Scan 尽量靠近数据，但本地性不是唯一目标。如果一个节点拥有大量数据块，把所有 Scan 都压到它上面会失去并行收益。调度器要在本地读取和负载均衡之间取舍。

生成的 Query Schedule 包含每个 Backend 要运行的 Fragment Instance、Scan Range、数据流目标和 Runtime Filter 路由。

Coordinator 怎样启动整个查询

C++ 后端中的关键入口包括：

  • be/src/runtime/coordinator.cc
  • be/src/runtime/coordinator-backend-state.cc
  • be/src/runtime/query-exec-mgr.cc
  • be/src/runtime/query-state.cc
  • be/src/runtime/fragment-instance-state.cc

Coordinator::Exec() 建立查询级状态和后端状态，随后 StartBackendExec() 为每个 Backend 组织 ExecQueryFInstances RPC。远端 QueryExecMgr::StartQuery() 创建或取得 QueryState，再为本机每个 Fragment Instance 建立 FragmentInstanceState。

sequenceDiagram
    participant C as Coordinator
    participant B as BackendState
    participant Q as QueryExecMgr
    participant S as QueryState
    participant F as FragmentInstanceState
    participant E as ExecNode Tree
    C->>B: StartBackendExec
    B->>Q: ExecQueryFInstances RPC
    Q->>S: 创建或取得查询状态
    S->>F: 创建 Fragment Instance
    F->>E: Prepare / Open
    E-->>F: RowBatch 流
    F-->>C: 状态与 Runtime Profile
Coordinator::Exec
  → StartBackendExec
  → BackendState::Exec
  → ExecQueryFInstances RPC
  → QueryExecMgr::StartQuery
  → QueryState
  → FragmentState
  → FragmentInstanceState
  → ExecNode / DataSink

启动不是完全串行的。Coordinator 会并行向后端发送 RPC，并在确认关键启动阶段后进入运行状态。任何 Backend 返回错误，都可能触发查询取消并向其他实例传播 Cancel。

ExecNode 生命周期：Prepare、Open、GetNext、Close

执行树上的节点共享统一接口，定义在 be/src/exec/exec-node.h：

Prepare()  建立描述符、表达式、计数器和内存资源
Open()     打开子节点，完成阻塞型准备
GetNext()  生产下一批 RowBatch
Reset()    在子计划等场景中重新执行
Close()    释放资源，必须能处理部分初始化和失败路径

GetNext() 体现了 Pull 模型：父节点向子节点要数据。ScanNode 从 Scanner 取得 RowBatch，Filter 丢弃不匹配行，Aggregate 更新哈希表，Exchange 从网络接收批次。

但不能把整个引擎理解成纯同步迭代器。Scan Range 读取、数据流 RPC、Runtime Filter 和后端状态上报都大量异步执行；RowBatch 只是执行算子之间的主要数据载体。

Scan：真正决定查询下限的地方

没有任何 Join 优化能挽救一次不必要的全表读取。Impala 的 Scan 性能来自多层裁剪：

Catalog 分区裁剪
  → 文件与 Scan Range 裁剪
  → Parquet Row Group 统计裁剪
  → Page Index / Dictionary 过滤
  → Runtime Filter
  → 解码后的谓词判断

Scan Range

文件被拆为多个 Scan Range，分配给 Scanner 线程。I/O Manager 负责异步读取、缓存与缓冲区管理，使磁盘或远端存储 IO 能与解码计算重叠。

Parquet 列裁剪

只读取查询需要的列。列式存储把同列值放在一起，使压缩和批量解码更高效。嵌套类型还需要根据 Definition/Repetition Level 重建逻辑结构。

Predicate Pushdown

谓词不一定等数据完全解码后才执行。Min/Max、字典和 Page Index 可以提前证明某个数据块不可能命中，从而跳过读取或解码。

be/src/exec/hdfs-scan-node*、hdfs-scanner.h 和 parquet/ 目录是理解扫描路径的主要入口。

Runtime Filter：用 Join 的结果反过来过滤 Scan

静态优化发生在执行前，Runtime Filter 则在查询运行时生成。Hash Join Build Side 完成后，可以从 Join Key 构造 Bloom Filter、Min/Max Filter 或 In-List Filter，再把它发布给 Probe Side 的 Scan。

Build Side: dim.region_id = {3, 7, 9}
             ↓ 生成 Runtime Filter
Probe Scan: fact.region_id
             ↓ 提前排除绝大多数不可能匹配的数据
Hash Join: 只处理剩余行

过滤器可能是本地的，也可能跨节点聚合并广播。过滤器到得太晚时，Scan 已经读完，收益为零；等得太久又会延迟首批数据。因此执行器允许 Scan 等待有限时间，并在过滤器到达后应用到后续 Range。

Runtime Filter 的收益取决于 Build Side 基数、过滤选择性、发布时间和数据源是否支持下推。不是 Filter 数量越多越好，大 Filter 的构建、合并和网络传输同样有成本。

Hash Join：广播还是重新分区

假设事实表很大，维表较小：

Broadcast Join

维表在 Build Fragment 中生成哈希表或 Build Rows，并发送到每个 Probe 节点。

优点是事实表不用跨网络重新分区；缺点是 Build 数据要复制 N 份，内存成本随 Executor 数量增加。

Partitioned Join

两侧都按 Join Key 哈希分区，相同 Key 被路由到同一个接收实例。

优点是 Build 内存分散；缺点是两侧可能发生大量网络 Shuffle。

优化器的选择可以粗略写成：

Broadcast 成本 ≈ BuildSize × ExecutorCount
Partition 成本 ≈ BuildSize + ProbeSize 的网络重分布

实际模型还要考虑并行度、已有分区、基数和内存上限。统计信息错误时，这个选择很容易出错。

Exchange：分布式流水线的关节

发送端实现位于 be/src/runtime/krpc-data-stream-sender.cc。KrpcDataStreamSender 根据分区策略把 RowBatch 分到 Channel，序列化后通过 KRPC 发送。接收端由 DataStreamMgr 与 ExchangeNode 管理。

flowchart LR
    P[上游 RowBatch] --> H{分区策略}
    H -->|Hash Key A| C1[Channel 1]
    H -->|Hash Key B| C2[Channel 2]
    H -->|Hash Key C| C3[Channel 3]
    C1 -->|KRPC| X1[ExchangeNode 1]
    C2 -->|KRPC| X2[ExchangeNode 2]
    C3 -->|KRPC| X3[ExchangeNode 3]
    X1 --> N1[下游 Pipeline]
    X2 --> N2[下游 Pipeline]
    X3 --> N3[下游 Pipeline]

为了避免无限堆积，Channel 需要处理背压：前一批 RPC 未完成时，发送端不能无界地产生缓冲区。发送、序列化和网络延迟会体现在 Profile 的 DataStreamSender 指标中。

Exchange 也是性能分界线：它切断单机 Pipeline，引入序列化、网络和接收端排队。看执行计划时，不能只数 Scan 和 Join；每个 Exchange 都意味着一次数据分布改变。

Aggregate 为什么通常分两阶段

GROUP BY 常被拆成局部聚合和全局合并：

每个 Scanner/Executor
  → Partial Aggregate
  → 按 Group Key Exchange
  → Merge/Final Aggregate

局部聚合先把大量明细压缩成较少的分组状态，减少网络数据量。若 Group Key 基数接近输入行数，压缩效果会很差，哈希表还会占据大量内存。

聚合器会监控哈希表和内存使用，必要时采用分区与 Spill。Spill 保住查询正确性，却把内存计算变成磁盘 IO，尾延迟通常显著上升。

Sort 与 Top-N 的阻塞边界

有些节点可以边读边产出，有些必须看到大量输入才能返回结果。

  • Filter、Project 通常是流式节点；
  • Hash Join 的 Build Side 是阻塞的；
  • 完整 Sort 需要生成 Run 并归并；
  • Top-N 只保留候选集合，通常比完整排序省内存；
  • Final Aggregate 是否阻塞取决于执行形态。

SortNode::Open() 读取输入并驱动 Sorter，最终归并则可以在 GetNext() 中按需产生结果。外部排序在内存不足时写临时 Run，再执行多路归并。

Codegen 为什么能加速表达式执行

通用执行器需要处理类型分派、虚函数、Nullable 判断和表达式树调用。Impala 使用 LLVM 将热点表达式和部分循环编译为查询专用机器码，例如：

  • Join Key 哈希；
  • 谓词计算；
  • Tuple Materialization；
  • 聚合更新；
  • DataStream 分区哈希。

Codegen 消除了部分解释开销，但编译本身有成本。短查询可能在代码生成完成前就能执行结束，因此 Impala 会区分 Codegen 准备和实际执行，并在 Profile 中记录编译时间。

RowBatch、Tuple 与内存所有权

Impala 4.5 的经典执行路径仍大量使用 RowBatch。一个 RowBatch 保存 TupleRow 指针，Tuple 中按 SlotDescriptor 布局字段，变长数据由关联内存持有。

GetNext() 的所有权规则很重要：父节点取得的行可能仍引用子节点或 RowBatch 的内存，不能过早释放。RowBatch::TransferResourceOwnership() 等机制让缓冲区在算子间转移，而不是复制每个字符串。

减少复制提升速度，也让生命周期错误更危险。源码中 ExecNode::GetNext() 和 Close() 的注释明确规定了资源在成功、EOS、取消和错误路径上的处理方式。

内存不是一个全局数字

查询内存由多层对象共同约束：

进程 MemTracker
  └── Query MemTracker
      ├── Fragment / Instance
      ├── Hash Join Build
      ├── Aggregator
      ├── Sorter
      ├── Scan Buffers
      └── Data Stream Buffers

Buffer Pool 管理可回收页，Reservation 机制避免多个算子都假设自己能拿到剩余内存。算子在 Prepare/Open 阶段声明最低和理想资源，Scheduler 据此做 Admission Control。

超过内存限制时，支持 Spill 的算子尝试落盘；不支持或无法继续 Spill 时查询失败。把 MEM_LIMIT 调大只能改变边界，不能修复错误 Join 顺序、数据倾斜或失控基数。

Admission Control：查询开始前的背压

如果所有查询都立即启动，节点会在内存和 CPU 已经耗尽后才失败。Admission Control 在运行前依据资源池配置、并发查询、内存估算和队列状态决定：

  • 立即准入；
  • 排队等待；
  • 超时拒绝；
  • 因资源请求不可能满足而直接拒绝。

这不是简单限流。一个大查询可能占用多个 Backend 的资源，准入判断必须考虑整个调度集合，而不是只看 Coordinator。

Catalog 与 Statestore 为什么会影响查询

Catalogd 管理元数据更新并通过 Topic 分发。Impalad 本地维护 Catalog Cache，因此规划查询时不必每次同步访问 Hive Metastore。

Statestored 负责成员信息和 Topic 更新传播。它不在 RowBatch 数据路径里，但 Executor 成员变化、Catalog 版本传播和节点故障发现都依赖这套机制。

元数据不一致通常表现为“文件明明存在却查不到”或分区版本落后。INVALIDATE METADATA 和 REFRESH 影响范围不同，不应当作无差别修复按钮。

查询结果怎样返回

根 Fragment 通常运行在 Coordinator，PlanRootSink 或 BufferedPlanRootSink 收集最终 RowBatch。客户端调用 FetchResults 时，Coordinator::GetNext() 从结果 Sink 取数据，转换成协议结果集。

这意味着客户端消费速度也会影响查询尾部。如果客户端迟迟不 Fetch，结果缓冲区和查询状态不能立刻释放。大型导出应控制 Fetch Size，并避免让 Coordinator 成为单点序列化瓶颈。

Profile：从哪里判断查询慢在何处

Impala Runtime Profile 是执行器真实发生过什么的证据。阅读时建议按边界排查：

计划是否错了

  • Estimated Cardinality 与实际 RowsReturned 差多少；
  • Broadcast Build 是否过大；
  • Exchange 前后的数据量是否异常；
  • 是否缺少 Runtime Filter 或分区裁剪。

Scan 是否慢

  • BytesRead、ScanRangesComplete、RowsRead；
  • IO Wait 与 Scanner Threads；
  • Parquet 解码时间；
  • Runtime Filter 到达时间和过滤比例；
  • 本地、缓存、远程读取比例。

计算是否慢

  • Hash Join Build/Probe 时间；
  • 聚合哈希表基数；
  • Sort Run 和 Spill 数据量；
  • Codegen 时间；
  • 单个 Instance 是否远慢于其他实例。

是否存在倾斜

平均值会隐藏倾斜。应比较各 Fragment Instance 的最大值和分布：某个分区承载大量相同 Join Key 时，一个实例会拖住整个查询。

一条 Join SQL 的完整路径

把前面的组件串起来：

SELECT d.region, SUM(f.amount)
FROM fact_orders f
JOIN dim_store d ON f.store_id = d.store_id
WHERE f.order_date >= '2026-09-01'
GROUP BY d.region;

执行过程大致是：

  1. Analyzer 解析表、列、类型与权限；
  2. 分区裁剪只保留目标日期范围；
  3. Planner 根据统计选择维表为 Build Side；
  4. 维表较小时选择 Broadcast Join；
  5. Build Side 生成 Store ID Runtime Filter；
  6. Filter 发布到事实表 Scan，提前跳过不命中数据；
  7. 每个 Executor 执行本地 Hash Join 和 Partial Aggregate；
  8. 结果按 Region 经 Exchange 重分区；
  9. 上层 Fragment 合并聚合；
  10. 根 Fragment 把结果交给客户端。

任何一步都可能成为瓶颈。维表统计过期会改变 Join 策略，Runtime Filter 太晚会失效，Region 极度倾斜会拖慢最终聚合，客户端读取缓慢则延长结果阶段。

Impala 查询快的真实原因

可以把它归结为几件相互配合的事：

  • Daemon 和元数据常驻，避免每次查询启动完整计算框架；
  • Fragment Instance 将 Scan 和计算分散到数据附近；
  • Exchange 明确控制数据分布；
  • 列裁剪与多层谓词下推减少读取；
  • Runtime Filter 把 Join 信息推回 Scan；
  • Pipeline 让 IO、解码、计算和网络重叠；
  • LLVM Codegen 降低热点表达式开销；
  • Buffer Pool、Spill 和 Admission Control 给资源使用设边界。

没有哪一项能单独保证性能。优化器估错基数，执行器再快也会做错工作；数据倾斜严重，并行度越高也无法均匀；内存设置不合理，Pipeline 最终仍会被 Spill 打断。

结语

理解 Impala 最有效的方式不是背组件名，而是跟踪两个东西：数据在哪里改变分布，资源在哪里形成阻塞。

Exchange 标记分布变化，Scan、Join Build、Aggregate、Sort 和 Result Sink 标记主要资源边界。Runtime Profile 则把计划中的估计与执行中的事实放在一起。

一条 SQL 从文本变成结果，经历的是 Frontend 语义分析、Planner 代价选择、Scheduler 实例放置、Coordinator RPC 启动、Executor 流水线运行和结果回收。能沿这条链找到具体 Fragment、Instance 和算子，才算真正开始理解 Impala 查询引擎。