Postgres EXPLAIN慢查询分析器

--- name: postgres-explain-slow-query-analyzer description: 读取 Postgres EXPLAIN 和 EXPLAIN ANALYZE 计划（文本或JSON），找出查询缓慢的原因，，大型关系上的顺序扫描、行估计巨大的嵌套循环、错误估计（估计值与实际值）、高缓冲区读取、溢出到磁盘的 Sort/Hash、缺少 LIMIT 下推以及按实际时间计算的高成本节点。对修复方案进行排序（索引、重写、统计信息）。离线标准库 Python 分析器。 --- # Postgres EXPLAIN 慢查询分析器 你帮助开发人员和 DBA 阅读 Postgres 查询计划。慢查询很少仅通过查看 SQL 就能修复：计划会显示 Postgres 是选择了顺序扫描、爆炸式增长的嵌套循环、错误的行估计、溢出到磁盘的 Sort，还是读取了千兆字节缓冲区的 Hash Join。你接收 `EXPLAIN` 或 `EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT|JSON)` 计划，指出高成本节点，解释它们为什么有影响，并对具体的修复方案进行排序。这项技能在于阅读计划，而不是在没有证据的情况下从头重写 SQL。 你不执行破坏性语句。优先在副本上或可以回滚的事务中使用 `EXPLAIN (ANALYZE, BUFFERS)`；警告 ANALYZE 实际上会运行查询。 ## 本技能包含的文件 - `scripts/explain_analyze.py` - 解析文本或 JSON 的 EXPLAIN / EXPLAIN ANALYZE 输出；按实际时间或成本对节点评分；报告 EX01-EX12（顺序扫描、嵌套循环爆炸、错误估计、磁盘排序/哈希、高共享读、未执行节点、并行不平衡）；对建议的修复进行排序（仅限 Python 3.8+ 标准库） - `references/plan-nodes-and-buffers.md` - 如何阅读节点类型、成本、行数、循环、缓冲区和临时文件 - `references/checks-and-codes.md` - 每个发现代码、严重性和应对措施，以及输入提示 - `templates/slow-query-review.md` - 你连同计划摘录和排序后的修复方案一起交回的审查笔记 - `examples/orders_by_email.explain.txt` - 慢查询查找的文本 EXPLAIN ANALYZE 示例（顺序扫描 + 嵌套循环） - `examples/review-report.md` - 示例上的真实分析器输出 ## 工作流程 ### 1. 收集计划 询问 Postgres 版本、SQL，以及 `EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)`（或 JSON）。如果在生产环境上 ANALYZE 不安全，则使用不带 ANALYZE 的 `EXPLAIN (BUFFERS)`，并说明时间仅为估计值。当顺序扫描看起来可疑时，捕获表大小（`pg_stat_user_tables`，`pg_relation_size`）。 ### 2. 运行分析器 ``` python3 scripts/explain_analyze.py examples/orders_by_email.explain.txt python3 scripts/explain_analyze.py plan.json --json-out python3 scripts/explain_analyze.py plan.txt --min-ms 5 python3 scripts/explain_analyze.py --self-test ``` 退出码 0 = 无高危发现，1 = 有高危发现，2 = 输入错误。如果没有 shell，则使用 `references/plan-nodes-and-buffers.md` 自底向上遍历计划树，并应用 `references/checks-and-codes.md` 中的代码。 ### 3. 读取结果 - 首先看高危项：EX01 在大型关系上带有选择性过滤器的顺序扫描，EX03 实际行数乘以循环数巨大的嵌套循环，EX05 热点节点上超过 100 倍的错误估计比例，EX07 溢出到临时文件的 Sort/Hash。 - 中等：EX02 未索引的连接键，EX04 扫描后过滤器移除大部分行，EX06 高共享读缓冲区，EX08 未执行但在纸面上看起来仍昂贵的分支。 - 低/信息：EX09 并行工作进程不均，EX10 规划时间高于执行时间，EX11 未下推 LIMIT，EX12 强制完整计算的 CTE MATERIALIZED。 ### 4. 对修复进行排序 优先选择有证据支持的更改：添加或调整与 Filter/Join Filter 匹配的索引，更新统计信息（`ANALYZE`），重写变成嵌套循环的相关子查询，仅当 Sort/Hash 溢出且集合适合内存时提高 `work_mem`，或者在 Postgres 12+ 不必要地物化 CTE 时将其改为 `NOT MATERIALIZED`。不要在未从计划中指明列的情况下凭空发明索引。 ### 5. 交付 使用 `templates/slow-query-review.md` 返回结论、按实际时间和缓冲区计算的最昂贵节点、发现列表以及简短的排序后修复列表。 ## 示例结果 在示例计划中，嵌套循环占用了 98% 的时间：外层对 `customers` 的顺序扫描由 `lower(email)` 过滤（无表达式索引），驱动了内层对 `orders_customer_id_idx` 的索引扫描，该扫描运行了 48,210 次循环。外层扫描的估计行数为 42，实际为 48,210（错误估计）。`orders` 上的共享读缓冲区很高。排序后的修复：在 `lower(email)` 上建立表达式索引，然后确认连接选择性；对 `customers` 执行 `ANALYZE`。 ## 规则 - 每项声明都基于计划中的一个节点（类型、成本/实际值、行数、循环、缓冲区）。 - 区分仅估计的 EXPLAIN 与 ANALYZE；绝不将成本视为毫秒。 - 不建议在没有测量的情况下全局禁用顺序扫描或将 random_page_cost 设置为幻想值。 - 警告 EXPLAIN ANALYZE 会运行查询（触发器、写入、锁）。 - 这是计划阅读辅助工具，不能替代 `pg_stat_statements` 监控或完整的模式审查。 FILE:examples/orders_by_email.explain.txt QUERY PLAN ----------------------------------------------------------------------------------------------------------------------------- Nested Loop (cost=0.42..184220.50 rows=12 width=72) (actual time=0.085..912.440 rows=96 loops=1) Buffers: shared hit=1204 read=88440 -> Seq Scan on public.customers c (cost=0.00..1850.00 rows=42 width=40) (actual time=0.040..88.200 rows=48210 loops=1) Filter: (lower((email)::text) = 'ada@example.com'::text) Rows Removed by Filter: 151790 Buffers: shared hit=20 read=1830 -> Index Scan using orders_customer_id_idx on public.orders o (cost=0.42..4.20 rows=8 width=40) (actual time=0.015..0.016 rows=0 loops=48210) Index Cond: (customer_id = c.id) Filter: (created_at >= (now() - '90 days'::interval)) Rows Removed by Filter: 3 Buffers: shared hit=1184 read=86610 Planning Time: 0.412 ms Execution Time: 912.610 ms FILE:examples/review-report.md # Postgres EXPLAIN 分析 来源：`examples/orders_by_email.explain.txt` 执行时间：912.61 毫秒 规划时间：0.412 毫秒 **结论：需要优化** - 3 项高危发现。 ## 按时间（或成本）排序的最重要节点 | 节点 | 关系 | 实际毫秒 (x 循环) | 成本 | 行数估计 -> 实际 | 循环 | 共享读取 | |------|----------|---------------------|------|-----------------|-------|-------------| | 嵌套循环 | - | 912.4 | 184220 | 12 -> 96 | 1 | 88440 | | 索引扫描 | public.orders | 771.4 | 4 | 8 -> 0 | 48210 | 86610 | | 顺序扫描 | public.customers | 88.2 | 1850 | 42 -> 48210 | 1 | 1830 | ## 发现 | 严重性 | 代码 | 详情 | |-----|------|--------| | 高 | EX01 | 在 public.customers 上带有选择性 Filter 的顺序扫描 ((lower((email)::text) = 'ada@example.com'::text))；移除了 151790 行 | | 高 | EX03 | 嵌套循环子节点索引扫描运行 loops=48210 (实际行/循环~0) 消耗约 771.4 毫秒 | | 高 | EX05 | 顺序扫描 public.customers 上的错误估计：估计行数=42 实际=48210 (比例约 1148 倍) | | 中 | EX02 | 嵌套循环在 public.customers 上有内部/外部顺序扫描 | | 中 | EX06 | 嵌套循环 ? 共享读取=88440 块 | | 中 | EX06 | 索引扫描 public.orders 共享读取=86610 块 | ## 排序后的修复建议 1. 添加与顺序扫描 Filter 匹配的 btree/表达式索引（检查像 lower() 这样的函数包装）。 2. 减少嵌套循环外部行数或为内部连接键建立索引；在 ANALYZE 后重新检查连接顺序。 3. 对涉及的表执行 ANALYZE；考虑对相关列使用扩展统计信息。 4. 通过更好的索引或提前过滤来降低共享读取，使嵌套循环不必重新读取堆。 FILE:references/checks-and-codes.md # 检查项和代码 | 代码 | 严重性 | 含义 | 应对措施 | |------|----------|---------|------------| | EX01 | 高 | 在带有选择性 Filter 且实际行数大或实际时间高的关系上进行顺序扫描 | 添加与过滤器匹配的表达式或 btree 索引；避免将列包装在函数中而没有匹配的表达式索引 | | EX02 | 中 | 连接条件看起来未索引（嵌套循环内部顺序扫描或连接键上带 Filter 的高循环） | 为内部连接键建立索引；确认数据类型匹配（没有破坏索引的转换） | | EX03 | 高 | 嵌套循环具有非常高的 循环数 * 实际行数（行爆炸） | 首先减少外部行数；考虑哈希/合并连接；修复导致选择嵌套循环的错误估计 | | EX04 | 中 | 扫描后 Filter 移除大部分行（进入行数 >> 输出行数） | 将谓词推入索引条件；重写以便规划器可以使用索引 | | EX05 | 高 | 错误估计：实际/估计 或 估计/实际 在重要节点上 >= 100 | 执行 ANALYZE；对相关列使用扩展统计信息；避免在谓词中使用非不可变函数 | | EX06 | 中 | 热点节点上的高共享读缓冲区 | 通过缓存/索引减少堆获取；检查嵌套循环是否重新读取相同页面 | | EX07 | 高 | Sort 或 Hash 溢出到临时文件（外部合并 / Batches > 1 且写入临时文件） | 在排序/哈希之前减少行数；谨慎提高此查询会话的 work_mem；添加支持性索引以避免排序 | | EX08 | 低 | 节点未执行（实际 loops=0）但主导估计成本 | 运行时忽略；在比较没有 ANALYZE 的计划时有用 | | EX09 | 低 | 并行查询工作进程贡献不均 | 检查 parallel_* 设置以及 gather 是否值得 | | EX10 | 信息 | 规划时间相对于执行时间较大 | 使用预编译语句；简化一次性规划；分区裁剪问题 | | EX11 | 中 | LIMIT 位于大型集合的完整 Sort 或完整聚合之上 | 重写以便排序/过滤允许索引扫描提前停止 | | EX12 | 中 | CTE 扫描带有 MATERIALIZED 强制完整计算 | 使用 WITH ... AS NOT MATERIALIZED (PG12+) 或内联 CTE | ## 输入提示 - 文本计划：粘贴完整树；不要删除缩进。 - JSON 计划：Postgres `FORMAT JSON` 数组，包含一个计划对象；脚本接受原始数组或 `{Plan: ...}`。 - 像 `QUERY PLAN` 或虚线规则这样的可选标题行会被忽略。 FILE:references/plan-nodes-and-buffers.md # 计划节点和缓冲区 ## 捕获有用的计划 ```sql EXPLAIN (ANALYZE, BUFFERS, VERBOSE, FORMAT TEXT) <query>; -- 或 EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) <query>; ``` 在生产环境上，优先使用热备副本或将写入包装在可回滚的事务中。`ANALYZE` 会执行语句。 ## 阅读文本节点行 示例：`Seq Scan on public.customers (cost=0.00..1550.00 rows=42 width=64) (actual time=0.020..85.110 rows=48210 loops=1)` - 节点类型和关系 - `cost=启动..总计`（规划器单位，不是毫秒） - `rows` = 估计行数；在 ANALYZE 下还有 `actual time`、`rows` 和 `loops` - 该节点的总时间 ~= (实际总时间) * 循环数 (对于子树贡献模式)；叶节点时间比父节点更清晰，因为父节点包含子节点 ## 常见节点类型 - **顺序扫描**：读取整个堆。适用于小表或大多数行匹配的情况；在大型表上带有选择性 Filter 时会很痛苦。 - **索引扫描 / 仅索引扫描 / 位图堆扫描**：选择性高时效果好。位图堆扫描带有许多重新检查时仍可能有影响。 - **嵌套循环**：对于每个外部行，运行内部。外部行少 + 内部有索引时效果好；外部行多时是灾难性的。 - **哈希连接 / 合并连接**：更适合较大的集合；注意哈希溢出（`Batches` > 1，写入临时文件）。 - **排序**：`Sort Method: external merge` 表示溢出到磁盘；考虑 `work_mem` 或提前减少行数。 - **聚合 / 分组聚合 / 哈希聚合**：检查组列是否已索引或预先过滤。 - **CTE 扫描**：Postgres 12+ 可能 MATERIALIZE CTE；会阻塞过滤器下推。 - **限制**：应位于廉价路径之上；如果 Limit 位于对所有内容排序的完整排序之上，则向下推条件。 ## 缓冲区（BUFFERS 选项） - `shared hit` = 在共享缓冲区中找到（廉价） - `shared read` = 从操作系统/磁盘读入缓冲区（缓存未命中时昂贵） - `temp read/written` = work_mem 溢出文件 在重复的嵌套循环内部侧具有大量 `shared read` 的节点通常需要更好的外部过滤器或索引。 ## 行数错误估计 比较 `rows`（估计值）和实际 `rows`。在昂贵节点上超过 10 倍的比率（尤其是 100 倍以上）意味着统计数据不好、`ANALYZE` 过期、列相关，或函数包装索引列（`lower(email)`）。 ## 此技能不做什么 它不替代 `pg_stat_statements`、auto_explain 或从模式单独开始的索引设计审查。它从一个计划开始并解释该计划。 FILE:scripts/explain_analyze.py #!/usr/bin/env python3 """Postgres EXPLAIN / EXPLAIN ANALYZE 计划分析器（仅限标准库，Python 3.8+）。 解析文本或 JSON 计划，对昂贵节点评分，并报告侧重于阅读计划的发现项 EX01-EX12（顺序扫描、嵌套循环爆炸、错误估计、缓冲区压力、磁盘溢出）。 用法： explain_analyze.py PLAN.txt|.json [--min-ms N] [--json-out] explain_analyze.py --self-test 退出码：0 = 无高危发现，1 = 有高危发现，2 = 输入错误。 """ import argparse, json, re, sys SEV = {"high": 0, "medium": 1, "low": 2, "info": 3} NODE_RE = re.compile( r"^(?P<indent>\s*)(?:->\s*)?(?P<type>.+?)" r"(?=\s+using\s+|\s+on\s+|\s*\(cost=)" r"(?:\s+using\s+(?P<idx>[\w.]+))?" r"(?:\s+on\s+(?P<rel>[\w.]+))?" r"(?:\s+(?P<alias>\w+))?" r"\s*\(cost=(?P<cstart>[\d.]+)\.\.(?P<ctotal>[\d.]+)\s+rows=(?P<erows>[\d]+)\s+width=(?P<width>[\d]+)\)" r"(?:\s*\(actual time=(?P<astart>[\d.]+)\.\.(?P<atotal>[\d.]+)\s+rows=(?P<arows>[\d]+)\s+loops=(?P<loops>[\d]+)\))?" ) BUF_RE = re.compile( r"Buffers:\s*(?:shared hit=(?P<shit>\d+))?" r"(?:\s*read=(?P<sread>\d+))?" r"(?:.*?temp read=(?P<tread>\d+))?" r"(?:.*?written=(?P<twrite>\d+))?", re.I, ) FILTER_RE = re.compile(r"^\s+Filter:\s*(.+)$") REMOVED_RE = re.compile(r"Rows Removed by Filter:\s*(\d+)") SORT_RE = re.compile(r"Sort Method:\s*(.+)$", re.I) HASH_BATCH_RE = re.compile(r"Batches:\s*(\d+)", re.I) PLAN_TIME_RE = re.compile(r"Planning Time:\s*([\d.]+)\s*ms", re.I) EXEC_TIME_RE = re.compile(r"Execution Time:\s*([\d.]+)\s*ms", re.I) class InputError(Exception): pass def parse_text(text): lines = text.splitlines() nodes = [] extras = {"planning_ms": None, "execution_ms": None} stack = [] i = 0 while i < len(lines): line = lines[i].rstrip() if not line or set(line) <= set("-=") or line.strip() == "QUERY PLAN": i += 1 continue m = PLAN_TIME_RE.search(line) if m: extras["planning_ms"] = float(m.group(1)); i += 1; continue m = EXEC_TIME_RE.search(line) if m: extras["execution_ms"] = float(m.group(1)); i += 1; continue m = NODE_RE.match(line) if m: raw_indent = re.match(r"^(\s*)", line).group(1) ind = len(raw_indent) node = { "type": m.group("type").strip(), "rel": m.group("rel"), "idx": m.group("idx"), "cost_start": float(m.group("cstart")), "cost_total": float(m.group("ctotal")), "est_rows": int(m.group("erows")), "width": int(m.group("width")), "act_start": float(m.group("astart")) if m.group("astart") else None, "act_total": float(m.group("atotal")) if m.group("atotal") else None, "act_rows": int(m.group("arows")) if m.group("arows") else None, "loops": int(m.group("loops")) if m.group("loops") else None, "filter": None, "removed": None, "buffers": {}, "sort_method": None, "hash_batches": None, "indent": ind, "children": [], } while stack and stack[-1][0] >= ind: stack.pop() if stack: stack[-1][1]["children"].append(node) else: nodes.append(node) stack.append((ind, node)) j = i + 1 while j < len(lines): pl = lines[j] if not pl.strip(): break if NODE_RE.match(pl) or PLAN_TIME_RE.search(pl) or EXEC_TIME_RE.search(pl): break if not pl.startswith(" ") and not pl.startswith("\t"): break fm = FILTER_RE.match(pl) if fm: node["filter"] = fm.group(1).strip() rm = REMOVED_RE.search(pl) if rm: node["removed"] = int(rm.group(1)) bm = BUF_RE.search(pl) if bm: node["buffers"] = {k: int(bm.group(k)) for k in ("shit", "sread", "tread", "twrite") if bm.group(k)} sm = SORT_RE.search(pl) if sm: node["sort_method"] = sm.group(1).strip() hb = HASH_BATCH_RE.search(pl) if hb: node["hash_batches"] = int(hb.group(1)) j += 1 i = j continue i += 1 if not nodes: raise InputError("未找到计划节点（需要包含 cost=... rows=... 的 EXPLAIN 文本）") return nodes, extras def parse_json_plan(obj): if isinstance(obj, list): if not obj: raise InputError("空的 JSON 计划数组") obj = obj[0] if not isinstance(obj, dict): raise InputError("JSON 计划必须是对象或对象数组") plan = obj.get("Plan") or obj.get("plan") if not plan: raise InputError("JSON 缺少 Plan 对象") extras = { "planning_ms": obj.get("Planning Time"), "execution_ms": obj.get("Execution Time"), } def conv(p, indent=0): node = { "type": p.get("Node Type", "Unknown"), "rel": p.get("Relation Name") or p.get("Relation"), "idx": p.get("Index Name"), "cost_start": float(p.get("Startup Cost", 0)), "cost_total": float(p.get("Total Cost", 0)), "est_rows": int(p.get("Plan Rows", 0)), "width": int(p.get("Plan Width", 0)), "act_start": p.get("Actual Startup Time"), "act_total": p.get("Actual Total Time"), "act_rows": p.get("Actual Rows"), "loops": p.get("Actual Loops"), "filter": p.get("Filter"), "removed": p.get("Rows Removed by Filter"), "buffers": {}, "sort_method": p.get("Sort Method"), "hash_batches": p.get("Hash Batches") or p.get("Batches"), "indent": indent, "children": [], } if "Shared Hit Blocks" in p or "Shared Read Blocks" in p: node["buffers"] = { k: int(p[src]) for k, src in (("shit", "Shared Hit Blocks"), ("sread", "Shared Read Blocks"), ("tread", "Temp Read Blocks"), ("twrite", "Temp Written Blocks")) if src in p } for ch in p.get("Plans") or []: node["children"].append(conv(ch, indent + 2)) return node return [conv(plan)], extras def walk(nodes): out = [] def w(n): out.append(n) for c in n["children"]: w(c) for n in nodes: w(n) return out def time_ms(n): if n["act_total"] is None or n["loops"] is None: return None return float(n["act_total"]) * float(n["loops"]) def analyze(nodes, extras, min_ms=1.0): findings = [] flat = walk(nodes) def add(sev, code, msg, node=None): findings.append((sev, code, msg, node)) for n in flat: t = n["type"] ms = time_ms(n) est, act = n["est_rows"], n["act_rows"] loops = n["loops"] or 1 rel = n["rel"] or "?" if t.startswith("Seq Scan") and n.get("filter") and (n.get("removed") or 0) > 1000: if (act or 0) > 1000 or (ms or 0) >= min_ms * 10: add("high", "EX01", "在 %s 上带有选择性 Filter (%s) 的顺序扫描；移除了 %s 行" % (rel, n["filter"], n.get("removed")), n) if t.startswith("Nested Loop"): for c in n["children"]: if c["type"].startswith("Seq Scan"): add("medium", "EX02", "嵌套循环在 %s 上有内部/外部顺序扫描" % (c.get("rel") or "?"), c) if t.startswith("Nested Loop"): for c in n["children"]: cl = c.get("loops") or 0 if cl >= 10000 and (time_ms(c) or 0) >= min_ms * 20: add("high", "EX03", "嵌套循环子节点 %s 运行 loops=%d (实际行/循环约 %s) 消耗约 %.1f 毫秒" % ( c["type"], cl, c.get("act_rows"), time_ms(c) or 0), c) if n.get("removed") and act is not None: total_in = n["removed"] + act if total_in > 1000 and n["removed"] / total_in >= 0.9: add("medium", "EX04", "%s %s 上的 Filter 移除了 %.0f%% 的行（移除了 %d，保留了 %d）" % ( t, rel, 100.0 * n["removed"] / total_in, n["removed"], act), n) if act is not None and est is not None and (ms or 0) >= min_ms: if est == 0: ratio = float("inf") if act > 0 else 1.0 else: ratio = max(act / max(est, 1), est / max(act, 1)) if ratio >= 100: add("high", "EX05", "%s %s 上的错误估计：估计行数=%d 实际=%d (比例约 %.0f 倍)" % (t, rel, est, act, ratio if ratio != float("inf") else 9999), n) sread = n.get("buffers", {}).get("sread") or 0 if sread >= 5000 and (ms or 0) >= min_ms: add("medium", "EX06", "%s %s 共享读取=%d 块" % (t, rel, sread), n) sm = (n.get("sort_method") or "").lower() if "external" in sm: add("high", "EX07", "排序溢出到磁盘（%s）在 %s 上" % (n["sort_method"], t), n) if n.get("hash_batches") and n["hash_batches"] > 1 and (n.get("buffers", {}).get("twrite") or 0) > 0: add("high", "EX07", "哈希溢出：Batches=%d 临时写入=%s" % (n["hash_batches"], n["buffers"].get("twrite")), n) if loops == 0 and n["cost_total"] >= 1000: add("low", "EX08", "节点 %s 未执行但成本=%.0f" % (t, n["cost_total"]), n) if t.startswith("Limit"): for c in n["children"]: if c["type"].startswith("Sort") and (c.get("act_rows") or c.get("est_rows") or 0) > 1000: add("medium", "EX11", "LIMIT 位于大型集合的 Sort 之上（%s 行）" % (c.get("act_rows") or c.get("est_rows")), n) if "CTE" in t and (ms or n["cost_total"] > 1000): add("medium", "EX12", "CTE 扫描 %s 可能已被物化；如果过滤器应下推，考虑使用 NOT MATERIALIZED" % (n.get("rel") or ""), n) if extras.get("planning_ms") and extras.get("execution_ms"): if extras["planning_ms"] > extras["execution_ms"] and extras["planning_ms"] > 20: add("info", "EX10", "规划时间 %.1f 毫秒 > 执行时间 %.1f 毫秒" % (extras["planning_ms"], extras["execution_ms"]), None) findings.sort(key=lambda x: (SEV.get(x[0], 9), x[1])) return findings def report(nodes, extras, findings, src): flat = walk(nodes) lines = [] lines.append("# Postgres EXPLAIN 分析") lines.append("") lines.append("来源：`%s`" % src) if extras.get("execution_ms") is not None: lines.append("执行时间：%s 毫秒" % extras["execution_ms"]) if extras.get("planning_ms") is not None: lines.append("规划时间：%s 毫秒" % extras["planning_ms"]) high = sum(1 for f in findings if f[0] == "high") lines.append("") if high: lines.append("**结论：需要优化** - %d 项高危发现。" % high) elif findings: lines.append("**结论：有备注通过** - %d 项发现，均非高危。" % len(findings)) else: lines.append("**结论：通过** - 无超过阈值的发现。") lines.append("") lines.append("## 按时间（或成本）排序的最重要节点") lines.append("") scored = [] for n in flat: ms = time_ms(n) scored.append((ms if ms is not None else -1, n["cost_total"], n)) scored.sort(key=lambda x: (x[0] if x[0] >= 0 else -1, x[1]), reverse=True) lines.append("| 节点 | 关系 | 实际毫秒 (x 循环) | 成本 | 行数估计 -> 实际 | 循环 | 共享读取 |") lines.append("|------|----------|---------------------|------|-----------------|-------|-------------|") for ms, cost, n in scored[:12]: am = ("%.1f" % ms) if ms is not None and ms >= 0 else "-" ra = "%s -> %s" % (n["est_rows"], n["act_rows"] if n["act_rows"] is not None else "-") lines.append("| %s | %s | %s | %.0f | %s | %s | %s |" % ( n["type"], n["rel"] or "-", am, cost, ra, n["loops"] if n["loops"] is not None else "-", n.get("buffers", {}).get("sread", "-"))) lines.append("") lines.append("## 发现") lines.append("") if not findings: lines.append("无。") else: lines.append("| 严重性 | 代码 | 详情 |") lines.append("|-----|------|--------|") for sev, code, msg, _ in findings: lines.append("| %s | %s | %s |" % (sev, code, msg.replace("|", "/"))) lines.append("") lines.append("## 排序后的修复建议") lines.append("") fixes = [] codes = {f[1] for f in findings} if "EX01" in codes: fixes.append("添加与顺序扫描 Filter 匹配的 btree/表达式索引（检查像 lower() 这样的函数包装）。") if "EX03" in codes or "EX02" in codes: fixes.append("减少嵌套循环外部行数或为内部连接键建立索引；在 ANALYZE 后重新检查连接顺序。") if "EX05" in codes: fixes.append("对涉及的表执行 ANALYZE；考虑对相关列使用扩展统计信息。") if "EX07" in codes: fixes.append("在 Sort/Hash 之前减少行数，或在测量后仅为本次会话提高 work_mem。") if "EX06" in codes: fixes.append("通过更好的索引或提前过滤来降低共享读取，使嵌套循环不必重新读取堆。") if "EX11" in codes: fixes.append("重写以便 LIMIT 可以在有序索引扫描上停止，而不是对整个集合进行排序。") if "EX12" in codes: fixes.append("尝试使用 WITH ... AS NOT MATERIALIZED 或内联 CTE，以便过滤器可以下推。") if not fixes: fixes.append("根据阈值没有高优先级的结构性修复；将计划置于 pg_stat_statements 监视之下。") for i, f in enumerate(fixes, 1): lines.append("%d. %s" % (i, f)) lines.append("") return "\n".join(lines) + "\n" def load_plan(path): raw = open(path, encoding="utf-8").read() if path.endswith(".json") or raw.lstrip().startswith(("[", "{")): try: obj = json.loads(raw) except json.JSONDecodeError as e: raise InputError("无效的 JSON：%s" % e) return parse_json_plan(obj) return parse_text(raw) def self_test(): tests = [] def t(name, ok): tests.append((name, bool(ok))) sample = """Nested Loop (cost=0.42..184220.50 rows=12 width=72) (actual time=0.085..912.440 rows=96 loops=1) Buffers: shared hit=1204 read=88440 -> Seq Scan on public.customers c (cost=0.00..1850.00 rows=42 width=40) (actual time=0.040..88.200 rows=48210 loops=1) Filter: (lower((email)::text) = 'ada@example.com'::text) Rows Removed by Filter: 151790 Buffers: shared hit=20 read=1830 -> Index Scan using orders_customer_id_idx on public.orders o (cost=0.42..4.20 rows=8 width=40) (actual time=0.015..0.016 rows=0 loops=48210) Index Cond: (customer_id = c.id) Buffers: shared hit=1184 read=86610 Planning Time: 0.412 ms Execution Time: 912.610 ms """ nodes, extras = parse_text(sample) t("解析根嵌套循环", nodes[0]["type"] == "Nested Loop") t("两个子节点", len(nodes[0]["children"]) == 2) t("执行时间", extras["execution_ms"] == 912.610) findings = analyze(nodes, extras) codes = {f[1] for f in findings} t("EX01 选择性顺序扫描", "EX01" in codes) t("EX03 嵌套循环爆炸", "EX03" in codes) t("EX05 错误估计", "EX05" in codes) t("EX06 共享读取", "EX06" in codes) r = report(nodes, extras, findings, "t") t("报告结论需要优化", "NEEDS WORK" in r) tiny = """Seq Scan on public.t (cost=0.00..1.05 rows=5 width=4) (actual time=0.002..0.003 rows=5 loops=1) Execution Time: 0.020 ms """ nodes2, ex2 = parse_text(tiny) f2 = analyze(nodes2, ex2) t("小表无高危", not any(x[0] == "high" for x in f2)) spill = """Sort (cost=1000..1100 rows=10000 width=8) (actual time=50..60 rows=10000 loops=1) Sort Method: external merge Disk: 4096kB """ ns, es = parse_text(spill) fs = analyze(ns, es) t("EX07 外部排序", any(x[1] == "EX07" for x in fs)) j = [{"Plan": {"Node Type": "Limit", "Startup Cost": 0, "Total Cost": 10, "Plan Rows": 10, "Plan Width": 4, "Actual Startup Time": 0.1, "Actual Total Time": 5.0, "Actual Rows": 10, "Actual Loops": 1, "Plans": [{"Node Type": "Sort", "Startup Cost": 0, "Total Cost": 9, "Plan Rows": 5000, "Plan Width": 4, "Actual Startup Time": 0.1, "Actual Total Time": 4.9, "Actual Rows": 5000, "Actual Loops": 1, "Sort Method": "quicksort"}]}, "Planning Time": 1.0, "Execution Time": 5.1}] nj, ej = parse_json_plan(j) fj = analyze(nj, ej) t("JSON 限制+排序 -> EX11", any(x[1] == "EX11" for x in fj)) try: parse_text("not a plan") ok = False except InputError: ok = True t("拒绝错误输入", ok) for name, ok in tests: print(("PASS " if ok else "FAIL ") + name) passed = sum(ok for ok in tests) print("%d/%d 通过" % (passed, len(tests))) return 0 if passed == len(tests) else 1 def main(argv=None): ap = argparse.ArgumentParser(description="分析 Postgres EXPLAIN / EXPLAIN ANALYZE 计划。") ap.add_argument("plan", nargs="?") ap.add_argument("--min-ms", type=float, default=1.0) ap.add_argument("--json-out", action="store_true") ap.add_argument("--self-test", action="store_true") a = ap.parse_args(argv) if a.self_test: return self_test() if not a.plan: ap.print_usage(sys.stderr) return 2 try: nodes, extras = load_plan(a.plan) findings = analyze(nodes, extras, min_ms=a.min_ms) except (OSError, InputError) as e: print("错误：%s" % e, file=sys.stderr) return 2 if a.json_out: print(json.dumps({ "execution_ms": extras.get("execution_ms"), "planning_ms": extras.get("planning_ms"), "findings": [{"severity": s, "code": c, "detail": m} for s, c, m, _ in findings], }, indent=2)) else: sys.stdout.write(report(nodes, extras, findings, a.plan)) return 1 if any(f[0] == "high" for f in findings) else 0 if __name__ == "__main__": sys.exit(main()) FILE:templates/slow-query-review.md # 慢查询计划审查 - 日期： - 数据库 / Postgres 版本： - 环境（生产副本 / 暂存）： - 查询名称或工单号： ## SQL（简短） ```sql ``` ## 结论 - 总体： - 主要时间消耗点： ## 重要节点 | 节点 | 实际时间 / 成本 | 行数估计 -> 实际 | 循环 | 缓冲区 | 备注 | |------|--------------------|-----------------|-------|---------|-------| | | | | | | | ## 发现 | 代码 | 严重性 | 详情 | |------|----------|--------| | | | | ## 排序后的修复建议 1. 2. 3. ## 后续事项 - [ ] 更改后捕获 `EXPLAIN (ANALYZE, BUFFERS)` - [ ] 在添加更多索引前检查 `pg_stat_user_indexes` / 未使用的索引 - [ ] 注意基础表上的 ANALYZE 是否已过期