query-optimization

查询优化

此技能可帮助 AI Agent 诊断并修复运行缓慢的数据库查询。Agent 使用 EXPLAIN/EXPLAIN ANALYZE 解读查询执行计划，识别缺失的索引和低效的扫描模式，重写查询以消除性能瓶颈，检测并解决 ORM 中的 N+1 查询问题，并推荐监控工具以持续跟踪查询性能。重点是通过优化前后的证据，带来切实且可衡量的改进。

工作流

1. 找出慢查询： 从慢查询日志、应用性能监控（APM）工具或用户报告中收集有问题的查询。记录当前执行时间、涉及的表大小以及查询运行频率。应优先处理高频慢查询，而非偶尔出现的慢查询。

2. 分析执行计划： 对查询运行 EXPLAIN ANALYZE（PostgreSQL）或 EXPLAIN FORMAT=JSON（MySQL），以获取实际执行计划。检查大型表上的顺序扫描、预估行数较高的嵌套循环连接、未建立索引的列上的排序操作，以及预估行数与实际行数之间的较大差距。

3. 找出优化机会： 根据执行计划确定具体的修复方案：为 WHERE、JOIN 和 ORDER BY 子句中的列添加索引；将子查询重写为 JOIN；用指定列替换 SELECT *；在适当情况下添加 LIMIT 子句；使用覆盖索引避免查表；删除冗余或重复条件。

4. 应用优化： 创建必要的索引、重写查询或调整 ORM 的用法。对于 N+1 问题，将延迟加载改为预加载（例如，Django 中的 select_related/prefetch_related、Prisma 中的 include、SQLAlchemy 中的 joinedload）。每次只应用一项更改，以便分别衡量每项改进。

5. 测量并验证： 对优化后的查询重新运行 EXPLAIN ANALYZE，并将执行时间、扫描行数和计划结构与原查询进行比较。验证查询返回的结果完全相同。检查新增索引是否会导致写入性能降低到不可接受的程度。

6. 设置持续监控： 配置慢查询日志，并设置合适的阈值（例如，通过 PostgreSQL 的 log_min_duration_statement 将阈值设为 100ms）。接入 pg_stat_statements、Datadog 或 Grafana 等监控工具，以跟踪查询性能趋势并及早发现性能回退。

支持的技术

• PostgreSQL： EXPLAIN ANALYZE、pg_stat_statements、pg_stat_user_indexes、auto_explain
• MySQL： EXPLAIN FORMAT=JSON、Performance Schema、慢查询日志、pt-query-digest
• ORM： SQLAlchemy、Django ORM、Prisma、ActiveRecord、Sequelize、TypeORM
• 监控： pganalyze、Datadog APM、New Relic、Grafana + Prometheus

用法

提供运行缓慢的 SQL 查询（或描述 ORM 操作），并注明数据库类型和大致的表大小。如果可能，请附上当前的 EXPLAIN 输出。Agent 将分析执行计划，推荐具体的优化方案，并提供重写后的查询和索引创建语句。Agent 还可以检查 ORM 代码中的 N+1 模式，并建议预加载修复方案。

示例

示例 1：优化运行缓慢的 JOIN 查询

问题： 在包含 500K 条订单记录的表上，连接订单、用户和产品的报表查询需要 4.2 秒。

原始查询和 EXPLAIN：

EXPLAIN ANALYZE
SELECT *
FROM orders o
JOIN users u ON o.user_id = u.id
JOIN order_items oi ON oi.order_id = o.id
JOIN products p ON oi.product_id = p.id
WHERE o.status = 'shipped'
  AND o.ordered_at >= '2025-01-01';

Nested Loop  (cost=0.00..98452.30 rows=12340 width=892) (actual time=0.08..4201.33 rows=11842 loops=1)
  -> Seq Scan on orders o  (cost=0.00..15420.00 rows=24500 width=64) (actual time=0.04..1823.12 rows=24312 loops=1)
       Filter: ((status = 'shipped') AND (ordered_at >= '2025-01-01'))
       Rows Removed by Filter: 475688
  -> Index Scan using order_items_order_id_idx on order_items oi  (...)
Planning Time: 0.45 ms
Execution Time: 4201.88 ms

诊断： 对 orders（500K 行）进行顺序扫描，并按 status 和 ordered_at 进行筛选。这些筛选列没有对应的复合索引。此外，查询选择了所有列，而实际只需要其中一部分。

修复 — 添加复合索引并重写查询：

-- Create composite index matching the WHERE clause
CREATE INDEX idx_orders_status_ordered_at ON orders(status, ordered_at);

-- Rewrite query with specific columns
EXPLAIN ANALYZE
SELECT o.id AS order_id, u.full_name, u.email,
       p.name AS product_name, oi.quantity, oi.unit_price,
       o.ordered_at
FROM orders o
JOIN users u ON o.user_id = u.id
JOIN order_items oi ON oi.order_id = o.id
JOIN products p ON oi.product_id = p.id
WHERE o.status = 'shipped'
  AND o.ordered_at >= '2025-01-01';

优化后的 EXPLAIN：

Nested Loop  (cost=1.12..3842.56 rows=12340 width=198) (actual time=0.06..87.42 rows=11842 loops=1)
  -> Index Scan using idx_orders_status_ordered_at on orders o  (cost=0.42..892.15 rows=24500 width=24) (actual time=0.03..12.68 rows=24312 loops=1)
       Index Cond: ((status = 'shipped') AND (ordered_at >= '2025-01-01'))
  -> Index Scan using order_items_order_id_idx on order_items oi  (...)
Planning Time: 0.52 ms
Execution Time: 88.04 ms

结果： 通过将顺序扫描替换为索引扫描，并仅选择指定列以减少传输的数据，执行时间从 4,201ms 降至 88ms，性能提升 48 倍。

示例 2：修复 Django ORM 应用中的 N+1 查询

问题： 一个列出 100 个订单及其用户名和产品详情的视图会生成 201 条 SQL 查询（1 条查询订单 + 100 条查询用户 + 100 条查询产品），耗时 1.8 秒。

修复前 — N+1 模式：

# views.py — Triggers N+1 queries
def order_list(request):
    orders = Order.objects.filter(status="shipped").order_by("-ordered_at")[:100]
    results = []
    for order in orders:
        results.append({
            "id": order.id,
            "customer": order.user.full_name,       # Lazy load: 1 query per order
            "items": [
                {"product": item.product.name, "qty": item.quantity}
                for item in order.items.all()        # Lazy load: 1 query per order
            ],
        })
    return JsonResponse(results, safe=False)

Django Debug Toolbar 输出： 201 条查询，耗时 1,823 毫秒。

修复后 — 使用 select_related 和 prefetch_related 进行预加载：

# views.py — Fixed with eager loading
def order_list(request):
    orders = (
        Order.objects
        .filter(status="shipped")
        .select_related("user")                      # JOIN for user (1:1/FK)
        .prefetch_related("items__product")           # Prefetch items + products (1:N)
        .order_by("-ordered_at")[:100]
    )
    results = []
    for order in orders:
        results.append({
            "id": order.id,
            "customer": order.user.full_name,         # No extra query
            "items": [
                {"product": item.product.name, "qty": item.quantity}
                for item in order.items.all()          # No extra query
            ],
        })
    return JsonResponse(results, safe=False)

Django Debug Toolbar 输出： 3 条查询，耗时 42 毫秒。

结果： 查询数量从 201 条降至 3 条，响应时间从 1,823 毫秒降至 42 毫秒（提升 43 倍）。select_related 使用 SQL JOIN 查询用户外键，而 prefetch_related 则通过一条 IN 查询获取所有订单项及其产品。

最佳实践

• 始终使用 EXPLAIN ANALYZE，而不只是 EXPLAIN — ANALYZE 变体会运行查询，并显示实际行数和耗时；这些数据往往与估算值有显著差异，能揭示真正的瓶颈。
• 创建与 WHERE + ORDER BY 模式匹配的复合索引 — 在 (status, ordered_at) 上创建复合索引，效果远胜于分别为每列创建索引，因为数据库可以使用单次索引范围扫描。
• **避免在生产查询中使用 SELECT *** — 选择所有列会迫使数据库读取更宽的行、增加 I/O，并阻碍使用覆盖索引。始终只指定所需的列。
• 在 ORM 层面修复 N+1 问题 — 使用 select_related（Django）、joinedload（SQLAlchemy）、include（Prisma）或 includes（ActiveRecord），将关联对象的加载批处理为一两条查询，而不是数百条。
• 持续监控查询性能 — 在 PostgreSQL 中启用 pg_stat_statements，或在 MySQL 中启用 Performance Schema，按总执行时间（而不仅是单次查询耗时）跟踪最耗时的查询。
• 测试索引对写入的影响 — 每个索引都会加快读取，但会拖慢写入（INSERT、UPDATE、DELETE）。添加索引后，对写入密集型操作进行基准测试，确保这种权衡可以接受。

边缘情况

• 统计信息过时导致计划不佳： 表数据发生显著变化时（例如大量导入数据后），查询规划器可能会使用过时的统计信息。运行 ANALYZE（PostgreSQL）或 ANALYZE TABLE（MySQL）以刷新统计信息并获得准确的执行计划。
• 频繁变更表上的索引膨胀： 频繁更新和删除的表可能产生膨胀的索引，导致性能下降。定期安排 REINDEX（PostgreSQL）或 OPTIMIZE TABLE（MySQL）以回收空间。
• 视图中隐藏的相关子查询： 看似简单的查询可能引用了包含相关子查询的视图，而该子查询会对每一行执行一次。进行 EXPLAIN 分析时，务必展开视图以查看完整的执行计划。
• 参数嗅探 / 计划缓存： 为某个参数值缓存的查询计划，对另一个参数值可能表现不佳。在 PostgreSQL 中，对于参数选择性差异很大的查询，可使用 PREPARE/EXECUTE，或设置 plan_cache_mode = force_custom_plan。
• ORM 生成了不必要的 JOIN 查询： ORM 有时会在 INNER JOIN 足够时生成 LEFT JOIN，或添加不必要的子查询。使用 QuerySet.query（Django）或 .toSQL()（Knex）检查实际 SQL；如果 ORM 生成的 SQL 不理想，则改用原始查询。