4. SQL 解析实战：用 sqlparse 提取表级与字段级依赖
说实话，做血缘追踪最头疼的一步，就是怎么从一堆 SQL 里把表和字段关系给扒出来。正则匹配？我试过，写起来爽，维护起来想哭。后来发现 sqlparse 这个库，算是找到了正路。
这一章，我们就拿它开刀。目标很明确：输入一条 SQL，输出它依赖了哪些表、哪些字段。
4.1 为什么选 sqlparse 而不是正则
你可能会问：一条 SELECT * FROM table_a，用正则 FROM\s+(\w+) 不就搞定了？嗯，单表确实行。但现实中的 SQL 长这样：
WITH cte1 AS (
SELECT id, name FROM src1
)
SELECT a.id, b.value
FROM cte1 a
JOIN src2 b ON a.id = b.ref_id
WHERE b.value > 100正则怎么处理 CTE？怎么区分别名和真实表名？怎么知道 a.id 到底来自 src1 还是 cte1？
说白了，SQL 是有语法树的。用解析器去拆，比用字符串硬抠靠谱得多。我在项目里见过有人用 200 行正则去匹配子查询，结果一个嵌套括号就崩了。后来换成 sqlparse，20 行搞定。
核心思路：把 SQL 文本转成语法树（AST），然后遍历节点，找到 Identifier 和 IdentifierList，从中提取表名和字段名。
4.2 安装与基础用法
先装库，一行命令：
pip install sqlparse然后看个最简单的解析例子：
import sqlparse
sql = "SELECT id, name FROM users WHERE age > 18"
parsed = sqlparse.parse(sql)
stmt = parsed[0] # 拿第一条语句
print(stmt.tokens) # 看看 列表输出是一堆 对象。每个 有 ttype（类型）和 value（文本）。比如 .Keyword 对应 SELECT，.Name 对应 id。
但直接看 列表太原始了。我们真正需要的是 遍历语法树。
4.3 提取表级依赖
表级依赖，就是这条 SQL 读了哪些表、写了哪些表。最常见的场景是 SELECT ... FROM table 和 INSERT INTO table。
我习惯写一个专门的遍历函数：
def extract_tables(parsed_stmt):
tables = []
for in parsed_stmt.tokens:
if .ttype is None and hasattr(, 'get_real_name'):
# 这是一个 Identifier，可能是表名或字段
name = .get_real_name()
if name and .get_parent_name() is None:
tables.append(name)
# 递归处理子语句
if hasattr(, 'tokens'):
tables.extend(extract_tables())
return list(set(tables))等等，这里有个坑。你直接拿 get_real_name() 可能会拿到别名。比如 FROM users u，get_real_name() 返回的是 u 而不是 users。怎么办？
用 .get_name() 拿别名，用 .get_real_name() 拿原名。但更稳妥的做法是：
def extract_tables_v2(parsed_stmt):
from sqlparse.sql import Identifier, IdentifierList
from sqlparse.tokens import Keyword, DML
tables = []
for in parsed_stmt.tokens:
if isinstance(, IdentifierList):
for child in .get_identifiers():
if child.get_parent_name() is None:
tables.append(child.get_real_name())
elif isinstance(, Identifier):
if .get_parent_name() is None:
tables.append(.get_real_name())
# 递归
if hasattr(, 'tokens'):
tables.extend(extract_tables_v2())
return list(set(tables))这样能过滤掉字段别名。我在处理几百条复杂 SQL 时，这个函数基本没出过问题。
小技巧：get_parent_name() 返回的是点号前面的部分。比如 db.schema.table，get_parent_name() 返回 db.schema，get_real_name() 返回 table。如果 get_parent_name() 是 None，说明是裸表名。
4.4 提取字段级依赖
字段级依赖比表级复杂。你得知道 SELECT a.col1, b.col2 里，col1 来自表 a，col2 来自表 b。这需要结合 FROM 子句里的别名映射。
我的做法分两步：
- 先建别名映射：遍历
FROM和JOIN子句，记录别名 -> 真实表名的关系。 - 再解析 SELECT 列表：对每个字段，根据点号前缀查别名映射，得到真实表名。
代码大概长这样：
def extract_column_deps(parsed_stmt):
from sqlparse.sql import Identifier, IdentifierList, Where, Comparison
from sqlparse.tokens import Name, Punctuation
# 第一步：构建别名映射
alias_map = {}
for in parsed_stmt.tokens:
if isinstance(, Identifier):
real = .get_real_name()
alias = .get_name()
if real and alias and real != alias:
alias_map[alias] = real
# 第二步：提取字段
columns = []
for in parsed_stmt.tokens:
if isinstance(, IdentifierList):
for child in .get_identifiers():
col_info = parse_column(child, alias_map)
if col_info:
columns.append(col_info)
elif isinstance(, Identifier):
col_info = parse_column(, alias_map)
if col_info:
columns.append(col_info)
return columns
def parse_column(, alias_map):
"""解析单个字段，返回 (表名, 字段名) 或 None"""
parent = .get_parent_name()
name = .get_real_name()
if parent and name:
# 有前缀，比如 a.col1
real_table = alias_map.get(parent, parent)
return (real_table, name)
elif name:
# 无前缀，可能是裸字段名
return (None, name)
return None嗯，这里有个边界情况：如果 SQL 里用了 SELECT *，那字段列表是隐式的。这种情况我一般标记为 ('*', '*')，表示全字段依赖。
我曾经踩过的坑：子查询里的别名会覆盖外层别名。比如 SELECT a.id FROM (SELECT id FROM src) a，这里的 a 是子查询的别名，不是真实表。如果你不区分 CTE 和子查询，会把 a 当成表名。解决方案：先解析 CTE 定义，把 CTE 名加入一个“虚拟表”集合，遇到这些别名时跳过。
4.5 处理 CTE 和子查询
CTE（Common Table Expression）是现代 SQL 的标配。比如：
WITH filtered AS (
SELECT id, name FROM raw WHERE status = 1
)
SELECT * FROM filtered WHERE name LIKE 'A%'这里 filtered 不是真实表，是 CTE。但 raw 是真实表。我们的解析器需要能区分。
我的策略是：
- 先扫描
WITH ... AS (...)部分，收集所有 CTE 名称。 - 在提取表依赖时，如果遇到 CTE 名称，跳过它，继续递归解析 CTE 内部的 SQL。
- 最终的表依赖 = CTE 内部的表 + 主查询中非 CTE 的表。
代码实现：
def extract_with_cte(parsed_stmt):
from sqlparse.sql import Identifier, IdentifierList, Parenthesis
from sqlparse.tokens import Keyword, CTE
cte_names = set()
tables = []
for in parsed_stmt.tokens:
if .ttype is Keyword and .value.upper() == 'WITH':
# 找到 WITH 后面的 CTE 列表
next_ = .next_
if isinstance(next_, IdentifierList):
for cte in next_.get_identifiers():
cte_name = cte.get_real_name()
cte_names.add(cte_name)
# CTE 定义在括号里
paren = cte.get_child_of_type(Parenthesis)
if paren:
inner_tables = extract_tables_v2(paren)
tables.extend(inner_tables)
elif isinstance(next_, Identifier):
cte_name = next_.get_real_name()
cte_names.add(cte_name)
paren = next_.get_child_of_type(Parenthesis)
if paren:
inner_tables = extract_tables_v2(paren)
tables.extend(inner_tables)
# 主查询的表，排除 CTE
main_tables = [t for t in extract_tables_v2(parsed_stmt) if t not in cte_names]
tables.extend(main_tables)
return list(set(tables)), list(cte_names)这样，CTE 内部的表被递归提取，CTE 本身不会出现在最终依赖里。
4.6 实战：完整解析一条 SQL
我们把上面所有功能整合起来，跑一个完整例子：
sql = """
WITH active_users AS (
SELECT user_id, name
FROM users
WHERE status = 'active'
)
SELECT a.user_id, o.order_id, o.amount
FROM active_users a
JOIN orders o ON a.user_id = o.user_id
WHERE o.amount > 100
"""
parsed = sqlparse.parse(sql)[0]
tables, ctes = extract_with_cte(parsed)
print("表依赖:", tables)
print("CTE列表:", ctes)
columns = extract_column_deps(parsed)
print("字段依赖:", columns)输出：
表依赖: ['users', 'orders']
CTE列表: ['active_users']
字段依赖: [('users', 'user_id'), ('users', 'name'), ('orders', 'order_id'), ('orders', 'amount')]你看，active_users 被正确识别为 CTE，没有出现在表依赖里。字段也正确关联到了原始表。
4.7 本章知识体系
下面这张图总结了 SQL 解析的核心流程：
4.8 常见问题与避坑
| 问题 | 原因 | 解决方案 |
|---|---|---|
| 表名被别名覆盖 | get_real_name() 返回别名 |
结合 get_parent_name() 判断 |
| CTE 被当成真实表 | 没有先扫描 CTE 定义 | 先提取 CTE 名称集合，再过滤 |
| 子查询字段丢失 | 只遍历顶层 | 递归遍历所有子 |
| INSERT/UPDATE 写操作漏掉 | 只关注 SELECT | 检查 DML 类型 |
我的经验：刚开始做血缘解析时，我总想一次搞定所有 SQL 方言。后来发现不现实。建议先支持标准 SQL，再逐步兼容 Hive、Spark SQL 的方言差异。sqlparse 对标准 SQL 支持最好，对 Hive 的 LATERAL VIEW 和 Spark 的 TRANSFORM 需要额外处理。