4. SQL Parser 实战：用 Apache Calcite 解析 SQL，提取输入表和输出表

数据血缘追踪的第一步是什么？

我个人觉得，不是画图，不是建表，而是先把 SQL 给「拆开」。你想想看，一条几百行的 ETL 脚本，里面嵌套了七八层子查询，还有 CTE、JOIN、UNION……如果靠肉眼去数输入表和输出表，那基本等于给自己挖坑。

所以这一章，我们来点硬核的。用 Apache Calcite 这个老牌 SQL 解析框架，把 SQL 拆成抽象语法树（AST），然后从中提取出我们最关心的东西——输入表（Source Table）和输出表（Target Table）。

4.1 为什么选 Calcite？

市面上能解析 SQL 的工具不少。Druid、JSqlParser、Antlr 自己写文法……我都试过。但最终在项目里落地时，我选了 Calcite。原因很简单：

  • 生态成熟：Calcite 是 Apache 顶级项目，Flink、Hive、Drill 都在用它做 SQL 解析和优化。
  • 语法支持全：不光标准 SQL，像 Hive 的 LATERAL VIEW、Spark 的 INSERT OVERWRITE，它都能处理。
  • 自带血缘分析接口：Calcite 的 SqlNode 遍历机制，天然适合做表级和字段级血缘。
我的经验：早期我用 JSqlParser 做过一版血缘工具，结果遇到 Hive 的 TRANSFORM 语法直接崩了。换成 Calcite 之后，类似的问题少了很多。不是说 Calcite 没坑，但社区响应快，踩了坑基本能找到解决方案。

4.2 核心流程：SQL → AST → 表信息

整个解析过程，说白了就三步：

  1. 解析：把 SQL 字符串变成一棵 AST（抽象语法树）。
  2. 遍历：用 Visitor 模式走遍这棵树，找到所有表引用。
  3. 分类：根据 SQL 类型（SELECT、INSERT、CREATE AS 等），区分输入表和输出表。

下面这张图，是我在项目中一直贴在墙上的流程。每次写解析逻辑前，我都会看一眼它。

Calcite SQL 解析核心流程 原始 SQL 字符串 SqlParser.parse() SqlNode (AST) SqlVisitor 遍历 输出表 (Target) 输入表 (Source) 关键：同一个 SqlNode，根据 SQL 类型决定哪些表是输入、哪些是输出

4.3 实战：写一个表血缘提取器

光说不练假把式。我们直接上手写代码。

4.3.1 引入依赖

Maven 项目里加这个：

<dependency>
    <groupId>org.apache.calcite</groupId>
    <artifactId>calcite-core</artifactId>
    <version>1.35.0</version>
</dependency>
版本坑：我曾经在 1.30 版本上遇到过一个 bug——解析 CREATE TABLE ... AS SELECT 时，输出表被识别成了输入表。升级到 1.35 后问题解决。所以建议直接用最新稳定版。

4.3.2 核心代码：表名提取器

我们需要写一个 SqlVisitor，遍历 AST 时收集所有表名。这里的关键是区分 SqlIdentifier 和 SqlSelect 的层级关系。

import org.apache.calcite.sql.*;
import org.apache.calcite.sql.parser.SqlParseException;
import org.apache.calcite.sql.parser.SqlParser;
import org.apache.calcite.sql.util.SqlBasicVisitor;

import java.util.*;

public class TableLineageExtractor {

    // 存储输入表和输出表
    private Set<String> sourceTables = new HashSet<>();
    private Set<String> targetTables = new HashSet<>();

    // 解析并提取
    public void extract(String sql) throws SqlParseException {
        SqlParser parser = SqlParser.create(sql);
        SqlNode sqlNode = parser.parseStmt();

        // 先判断 SQL 类型，确定输出表
        if (sqlNode instanceof SqlInsert) {
            SqlInsert insert = (SqlInsert) sqlNode;
            targetTables.add(extractTableName(insert.getTargetTable()));
        } else if (sqlNode instanceof SqlCreateTable) {
            SqlCreateTable create = (SqlCreateTable) sqlNode;
            targetTables.add(extractTableName(create.getName()));
        }

        // 遍历 AST，收集所有输入表
        sqlNode.accept(new SqlBasicVisitor<Void>() {
            @Override
            public Void visit(SqlIdentifier identifier) {
                // 跳过输出表，避免重复
                String tableName = identifier.toString();
                if (!targetTables.contains(tableName)) {
                    sourceTables.add(tableName);
                }
                return null;
            }
        });
    }

    // 从 SqlNode 中提取表名字符串
    private String extractTableName(SqlNode node) {
        if (node instanceof SqlIdentifier) {
            return ((SqlIdentifier) node).toString();
        }
        return node.toString();
    }

    public Set<String> getSourceTables() { return sourceTables; }
    public Set<String> getTargetTables() { return targetTables; }
}

4.3.3 测试一下

写个 main 方法跑跑看：

public static void main(String[] args) throws Exception {
    TableLineageExtractor extractor = new TableLineageExtractor();

    String sql = "INSERT INTO dws.daily_sales " +
                 "SELECT t1.order_id, t1.amount, t2.user_name " +
                 "FROM ods.orders t1 " +
                 "JOIN dim.users t2 ON t1.user_id = t2.user_id " +
                 "WHERE t1.dt = '2024-01-01'";

    extractor.extract(sql);

    System.out.println("输出表: " + extractor.getTargetTables());
    System.out.println("输入表: " + extractor.getSourceTables());
}

输出结果：

输出表: [dws.daily_sales]
输入表: [ods.orders, dim.users]

嗯，符合预期。输入表是 ods.orders 和 dim.users，输出表是 dws.daily_sales。

4.4 处理复杂场景

上面的例子比较简单。实际生产环境里，SQL 花样多得很。我遇到过几个典型场景，这里列出来供你参考。

SQL 类型 输入表 输出表 注意事项
SELECT ... FROM A JOIN B A, B 无 纯查询，无输出
INSERT INTO T SELECT ... FROM A A T 最常见的 ETL 模式
CREATE TABLE T AS SELECT ... FROM A A T 注意 CTAS 语法
WITH cte AS (SELECT ... FROM A) SELECT ... FROM cte JOIN B A, B 无 CTE 是中间表，不是物理表
MERGE INTO T USING S ON ... WHEN MATCHED ... S T MERGE 的输出表是目标表
避坑指南：CTE（Common Table Expression）里的临时表名，千万别当成物理表加进去。我曾经犯过这个错，结果血缘图里多了一堆不存在的表，排查了半天才发现是 CTE 别名被误识别了。解决办法很简单——在 Visitor 里维护一个 CTE 名称列表，遇到这些名字直接跳过。

4.5 进阶：处理子查询和别名

子查询是另一个容易出问题的地方。比如：

SELECT a.col1, b.col2
FROM (
    SELECT id, col1 FROM ods.table_a
) a
JOIN dim.table_b b ON a.id = b.id

这里 a 是子查询的别名，不是物理表。真正的输入表是 ods.table_a 和 dim.table_b。

怎么处理？我的做法是：在 Visitor 里遇到 SqlSelect 时，先递归处理它的 FROM 子句。如果 FROM 是一个子查询，就继续往下钻，直到找到底层的 SqlIdentifier。同时，把子查询的别名记录下来，在后续遍历中忽略它。

@Override
public Void visit(SqlSelect select) {
    SqlNode from = select.getFrom();
    if (from instanceof SqlBasicCall) {
        SqlBasicCall call = (SqlBasicCall) from;
        // 处理 JOIN
        for (SqlNode operand : call.getOperands()) {
            extractFromNode(operand);
        }
    } else {
        extractFromNode(from);
    }
    return null;
}

private void extractFromNode(SqlNode node) {
    if (node instanceof SqlIdentifier) {
        sourceTables.add(node.toString());
    } else if (node instanceof SqlSelect) {
        // 递归处理子查询
        visit((SqlSelect) node);
    }
    // 忽略别名节点
}

4.6 性能与工程化建议

如果你要把这个能力集成到数据治理平台里，有几点需要注意：

  • 批量解析：不要一条一条解析。把一批 SQL 打包，用多线程并行解析。Calcite 的 SqlParser 是线程安全的，放心用。
  • 缓存 AST：如果同一个 SQL 会被多次分析（比如字段级血缘），把解析后的 SqlNode 缓存起来，避免重复解析。
  • 异常处理：生产环境的 SQL 千奇百怪，解析失败是常态。我习惯在解析失败时记录原始 SQL 和异常堆栈，然后跳过，不影响其他任务。
我的习惯：在项目里，我会把解析结果落库到一张 table_lineage 表里，字段包括 source_table、target_table、sql_hash、parse_time。这样后续做血缘追溯时，直接查表就行，不用每次都重新解析。
← 上一章 📖 返回目录 下一章 →