【分布式 OLAP 查询引擎】Spark SQL 与 Catalyst：逻辑/物理计划与 AQE

第 12 篇 把 Trino 476 从 SqlQueryExecution 到 Iceberg Split 的 Coordinator 路径讲完了。数据平台里与 Trino 并列的另一条 JVM 查询栈是 Spark SQL：批 ETL、湖仓 MERGE INTO、Structured Streaming 批微批模式都跑在同一套 Catalyst 优化器与 Tungsten 运行时之上。

Spark 读者转 Trino 时最常见的混淆有三处：

• Spark 的 logical plan 与 physical plan 都叫 SparkPlan，和 Trino 的 PlanNode 如何对照？
• Whole-stage codegen 把算子 fuse 成 JVM 字节码，Trino 为何仍以 Interpreter 式 Operator 为主？
• AQE 能在运行时改 shuffle partition 数、补 skew join；Trino 476 的等价能力边界在哪？

本文回答上述问题，并给出 Spark 读 Iceberg V2 与 Trino Iceberg connector 的下推对照表（能力来自官方文档，不做跨引擎 latency 排名）。

环境说明：本机 WSL2、无 JVM、Docker Compose 不可用。Catalyst/AQE 描述来自 Spark 3.5 SQL Guide 与 apache/spark 源码包 org.apache.spark.sql.catalyst；Iceberg 行为引用 Iceberg Spark 文档。本文 不粘贴未执行的 explain 输出；PLAN.md 实验台账中的 Spark vs Trino 计划对照需读者在本地 local[*] 或集群上自行完成。

版本锚定：Spark 3.5+；Trino 476+ 作对照；Iceberg V2 table；对比口径与 lakehouse 第 18 篇 一致。

一、Catalyst 在 Spark SQL 中的位置

Spark SQL 入口 SparkSession.sql(text) 触发（来源：Spark 3.5 Documentation, SQL Programming Guide）：

flowchart LR
  P["Parser<br/>Catalyst SqlParser"]
  A["Analyzer<br/>Unresolved → LogicalPlan"]
  O["Optimizer<br/>Rules on LogicalPlan"]
  PP["Planner<br/>Logical → SparkPlan physical"]
  PREP["prepareForExecution<br/>AQE hooks · subquery"]
  EX["execute()<br/>RDD / SparkPlan exec"]
  P --> A --> O --> PP --> PREP --> EX

Trino 侧对应关系（第 2–3 篇）：Analyzer ≈ Trino Analyzer；Logical optimize ≈ PlanOptimizers；Physical + Fragment ≈ Trino distributed plan + PlanFragmenter；execute ≈ Trino Worker Task（无 Driver 侧 codegen 主路径）。

二、LogicalPlan 与 SparkPlan

2.1 LogicalPlan（未解析 → 已解析 → 已优化）

Catalyst 用 TreeNode 表示计划（源码：org.apache.spark.sql.catalyst.plans.logical.LogicalPlan）。常见节点：

• Relation / SubqueryAlias：表或视图
• Project、Filter
• Join、Aggregate、Sort

逻辑计划优化在 RuleExecutor 上批量应用 Rule（如 PushDownPredicate、ColumnPruning），与 Calcite 风格规则链类似（第 3 篇）。

2.2 SparkPlan（物理计划）

物理计划同样是 SparkPlan 子类（org.apache.spark.sql.execution.SparkPlan），但节点带 执行语义：

df.explain("formatted")（Spark 3.x）按 CTEs / Main 分段展示；explain("cost") 在启用 CBO 时展示代价（Spark 3.5 文档 Performance Tuning — Adaptive Query Execution 前置条件含 spark.sql.cbo.enabled）。

2.3 Whole-stage codegen 边界

Whole-stage codegen（WSCG）把 同一 shuffle 边界内 的无分支算子链编译为单个 JVM 函数，减少虚调用与中间行对象分配（来源：Spark 文档 Whole-Stage Codegen；源码 org.apache.spark.sql.execution.WholeStageCodegenExec）。

Pipeline 断点与 Trino 类似：

• ShuffleExchange / BroadcastExchange
• Sort、HashAggregate（mode 切换）
• 支持 codegen 的算子组合受限（复杂 UDF、某些 outer join 形态可能 fallback 到 WholeStageCodegenExec 外的 Volcano 式迭代）

Trino 476 以 预编译 OperatorFactory + 列式 Page 为主，不做查询级 JVM bytecode 生成；同等优化落在 vectorized expression 与 plan 规则（第 7 篇）。这是交互式引擎与批引擎的常见分工差异，不宜直接等同于「谁更快」。

三、Shuffle、Stage 与 Trino Exchange 对照

Spark 一次 action（collect、count 等）触发 DAGScheduler 切 Stage，边界是 宽依赖（shuffle）（来源：Spark Programming Guide, Job Scheduling）。

Spark shuffle 写 local disk（或 CE 插件的外部 shuffle service）；Trino shuffle 默认 内存 OutputBuffer + 网络，可选 exchange manager spill。批处理栈容忍 disk shuffle；交互式栈优先低延迟内存路径（第 16 篇 对照 spill 哲学）。

四、Adaptive Query Execution（AQE）

Spark 3.2+ 默认开启 AQE（spark.sql.adaptive.enabled=true，3.5 文档 Performance Tuning）。AQE 在 shuffle map stage 完成后 根据 运行时统计 重新优化下游 plan。

4.1 核心机制

AQE 依赖 map stage 输出的统计（row count、size per partition）。这与 Trino 规划期 CBO（第 4–5 篇）形成鲜明对照：Trino 476 不在运行时 自动拆分 skew join partition；倾斜排查靠 EXPLAIN ANALYZE（第 11 篇）。

4.2 何时 AQE 帮不上忙

• Scan 阶段已全表读：与 Trino 一样，需 partition/file pruning（Iceberg manifest）。
• UDF 阻断谓词下推：优化器看不到真实过滤率。
• Broadcast 仍 OOM：build 表 runtime 估计仍偏小。

五、Spark 读 Iceberg：V2 reader 与下推

Spark 3.x 通过 Iceberg Spark runtime（Maven org.apache.iceberg:iceberg-spark-runtime-3.5_*）注册 spark_catalog 或 iceberg catalog（Iceberg 文档 Spark Getting Started）。

5.1 计划层

• DataSource V2：SparkScanBuilder → SparkBatchQueryScan（Iceberg 源码 org.apache.iceberg.spark.source.SparkScanBuilder）。
• 谓词下推：SupportsPushDownFilters、SupportsPushDownRequiredColumns（V2 connector API）。
• 与 Trino 相同，partition / file pruning 发生在 打开 data file 之前，依赖 manifest stats（lakehouse 第 8 篇）。

5.2 文件层

选中文件后，Spark 使用 Parquet reader + Iceberg 的 column stats / delete file 处理（MOR 表需 merge-on-read 语义，Trino 同样支持 delete file 边界，细节见 Iceberg 文档 Spark Reads / Trino Iceberg）。

5.3 Trino vs Spark Iceberg 能力对照（文档级，非 benchmark）

深度 planner 哪一步注入 constraint 见本系列 第 15 篇；读湖漏斗全文见 lakehouse 第 18 篇。

六、从 Spark 视角读一条 SQL 的路径

sequenceDiagram
  participant D as Spark Driver
  participant E as Executors
  D->>D: parse · analyze · optimize
  D->>D: physical plan · WSCG insert
  D->>E: launch Stage 0 tasks (scan)
  E->>D: shuffle map statistics
  D->>D: AQE re-optimize (optional)
  D->>E: launch Stage 1 tasks (join)
  E->>D: action result / metrics

与 第 12 篇 Trino 时序 对比：

• Spark Driver 承担更多 物理执行协调（尤其 local[*]）；Trino Coordinator 专职调度，Worker 更「哑」。
• Spark AQE 在 Driver 上 二次优化；Trino 规划一次为主，除非 adaptive plan（476 部分特性，非 AQE 同级）。
• 两者 Iceberg planning 都绕不开 planFiles() 类 metadata 扫描；慢 planning 的根因往往相通（issue #11708 讨论）。

七、轻量实验指引（需 JVM 环境）

本机未运行；步骤供读者在具备 JDK 17 + Spark 3.5 的环境复现 PLAN.md 实验台账第 13 项。

# 示例：local[*] 启动 spark-shell，配置 Iceberg REST（与 reproduce 栈 REST URI 对齐）
# spark.sql.catalog.rest=org.apache.iceberg.spark.SparkCatalog
# spark.sql.catalog.rest.uri=http://localhost:8181
# spark.sql.catalog.rest.type=rest

# 同一张 Iceberg 表：
# spark.sql("SELECT count(*) FROM rest.db.orders WHERE o_orderdate >= DATE '1996-06-01'").explain("formatted")
# 与 Trino：EXPLAIN SELECT ... 对照 Scan 节点与 filter 下推

对照看什么（不比较 wall time）：

1. Scan 节点是否仍含 Filter 算子（谓词未完全下推）。
2. Join 两侧是否出现 Exchange / ShuffleExchangeExec。
3. 是否 BroadcastHashJoin vs SortMergeJoin。
4. AQE 开启前后 explain 是否变化（仅 Spark）。

DuckDB 嵌入式对照见 第 14 篇。

八、选型边界（本篇范围）

完整决策树见 第 18 篇。

九、小结

Spark SQL 的 Catalyst 把 SQL 走 parse → analyze → optimize → physical → execute；Whole-stage codegen 在 shuffle 边界内 fuse 算子；AQE 用 map stage 统计在运行时 coalesce partition、缓解 skew、切换 join 策略。Trino 476 在 MPP 调度模型（第 10 篇）与 Iceberg manifest → Split（第 12 篇）上与 Spark 同构，差异主要在 codegen vs Operator Page 流 与 运行时自适应 强弱。

下一篇进入嵌入式对照：DuckDB 与 DataFusion（第 14 篇）。

参考资料

1. Apache Spark 3.5 Documentation, SQL Programming Guide（Catalyst 流水线）。
2. Apache Spark 3.5 Documentation, Performance Tuning — Adaptive Query Execution（AQE 规则与配置）。
3. Apache Spark 3.5 Documentation, Whole-Stage Codegen。
4. Apache Spark 3.5 Documentation, Job Scheduling（Stage / Task）。
5. apache/spark branch 3.5, sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/（LogicalPlan）。
6. apache/spark branch 3.5, sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala（物理计划）。
7. Apache Iceberg Documentation, Spark Configuration / Spark Reads（V2 reader、catalog）。
8. Trino 476 Documentation, Iceberg connector（对照下推）。
9. 本系列 第 12 篇（Trino 全路径）。
10. lakehouse 第 18 篇（读湖漏斗）。

返回 系列目录 ｜ 上一篇：Trino 查询路径全拆解 ｜ 下一篇：DuckDB 与 DataFusion