基于大模型的 SQL 查询自动优化建议：从执行计划到改写方案

一、SQL 调优的经验依赖：从人工分析到智能建议

数据库性能问题中，慢查询是最常见的根因。传统的 SQL 调优流程依赖 DBA 的经验——阅读执行计划、识别全表扫描、判断索引缺失、手动改写 SQL。这个过程耗时且高度依赖个人经验，不同 DBA 可能给出截然不同的优化方案。

生产环境中，SQL 调优面临三个核心痛点：第一，执行计划解读门槛高——EXPLAIN 输出包含数十个字段，非专业人员难以快速定位问题；第二，优化方案的多样性——同一慢查询可能有多种优化路径（加索引、改写 SQL、调整参数），选择最优方案需要综合考虑查询频率、数据分布和写入代价；第三，改写 SQL 的正确性——手动改写 SQL 容易引入语义错误，特别是涉及子查询和窗口函数时。

这个问题的本质是：SQL 调优需要从"经验驱动"升级为"数据+推理驱动"——大模型可以理解执行计划的语义，结合数据库统计信息，生成结构化的优化建议。

二、AI SQL 优化的底层机制

flowchart TB
    SQL[慢查询 SQL] --> EXPLAIN[执行计划采集]
    EXPLAIN --> PLAN[执行计划解析]

    SQL --> LLM[大模型推理]
    PLAN --> LLM
    STATS[表统计信息] --> LLM
    INDEX[现有索引] --> LLM
    HIST[历史优化案例] --> LLM

    subgraph 推理输出["优化建议输出"]
        LLM --> R1[问题诊断<br/>全表扫描/索引失效/排序溢出]
        LLM --> R2[索引建议<br/>新建/修改/删除索引]
        LLM --> R3[SQL改写<br/>子查询→JOIN/窗口函数优化]
        LLM --> R4[参数调整<br/>缓冲池/排序区/并行度]
    end

    R2 --> VERIFY[验证层<br/>预估代价对比]
    R3 --> VERIFY
    VERIFY --> |代价降低| APPLY[应用优化]

关键机制解析：

1. 执行计划解析：将 EXPLAIN 的表格输出解析为结构化数据——每个算子的类型（Seq Scan、Index Scan、Hash Join）、行数估计、耗时占比。大模型需要理解这些字段的语义含义。

2. 多源信息融合：优化建议不仅基于执行计划，还需要表统计信息（行数、基数、数据分布）、现有索引列表和历史优化案例。这些信息作为上下文注入 Prompt。

3. 代价验证：生成的优化建议需要通过 EXPLAIN 预估代价进行验证，确保优化后代价确实降低，避免"优化"反而变慢的情况。

三、AI SQL 优化的工程实现

3.1 执行计划解析器

import re
from dataclasses import dataclass

@dataclass
class PlanNode:
    node_type: str       # Seq Scan, Index Scan, Hash Join...
    relation: str        # 表名
    cost: float          # 启动代价
    total_cost: float    # 总代价
    rows: int            # 估计行数
    width: int           # 平均行宽
    filter: str = ""     # 过滤条件
    index_name: str = "" # 使用的索引
    children: list = None

class ExplainParser:
    """PostgreSQL执行计划解析器"""

    def parse(self, explain_output: str) -> PlanNode:
        lines = explain_output.strip().split('\n')
        return self._parse_node(lines, 0)

    def _parse_node(self, lines, start_idx):
        """递归解析执行计划树"""
        line = lines[start_idx]
        # 解析节点类型和属性
        node = self._parse_line(line)

        # 递归解析子节点
        child_idx = start_idx + 1
        while child_idx < len(lines):
            indent = len(lines[child_idx]) - len(lines[child_idx].lstrip())
            if indent <= len(line) - len(line.lstrip()):
                break
            child = self._parse_node(lines, child_idx)
            if node.children is None:
                node.children = []
            node.children.append(child)
            child_idx += 1

        return node

    def _parse_line(self, line: str) -> PlanNode:
        """解析单行执行计划"""
        # 示例: -> Seq Scan on orders  (cost=0.00..15432.00 rows=100000 width=100)
        match = re.search(
            r'(\w+(?:\s\w+)*)\s(?:on\s(\w+)\s)?'
            r'\(cost=([\d.]+)\.\.([\d.]+)\s'
            r'rows=(\d+)\swidth=(\d+)\)',
            line
        )
        if match:
            return PlanNode(
                node_type=match.group(1),
                relation=match.group(2) or "",
                cost=float(match.group(3)),
                total_cost=float(match.group(4)),
                rows=int(match.group(5)),
                width=int(match.group(6)),
            )
        return PlanNode("Unknown", "", 0, 0, 0, 0)

3.2 AI 优化建议生成

class SQLOptimizer:
    """AI驱动的SQL优化建议生成器"""

    def __init__(self, llm_client, db_client):
        self.llm = llm_client
        self.db = db_client

    def optimize(self, sql: str) -> dict:
        """生成SQL优化建议"""
        # 采集上下文信息
        plan = self._get_explain(sql)
        tables = self._extract_tables(sql)
        stats = self._get_table_stats(tables)
        indexes = self._get_existing_indexes(tables)

        # 构建Prompt
        prompt = self._build_prompt(sql, plan, stats, indexes)

        # 调用LLM
        response = self.llm.chat(prompt)

        # 解析结果
        suggestions = self._parse_response(response)

        # 验证建议
        for s in suggestions:
            if s["type"] == "sql_rewrite":
                s["verified"] = self._verify_rewrite(
                    sql, s["rewritten_sql"]
                )

        return suggestions

    def _build_prompt(self, sql, plan, stats, indexes):
        return f"""
        分析以下慢查询并给出优化建议。

        ## 原始SQL
        ```sql
        {sql}
        ```

        ## 执行计划
        {self._format_plan(plan)}

        ## 表统计信息
        {self._format_stats(stats)}

        ## 现有索引
        {self._format_indexes(indexes)}

        请按以下格式输出优化建议：
        1. 问题诊断：执行计划中的瓶颈节点
        2. 索引建议：需要新建或修改的索引（含DDL）
        3. SQL改写：优化后的SQL（保持语义等价）
        4. 预估效果：优化后的代价降低比例

        输出JSON格式。
        """

四、AI SQL 优化的边界分析

语义等价性保证

LLM 改写的 SQL 可能改变语义——特别是 NULL 处理、去重逻辑和排序稳定性。改写后的 SQL 必须通过结果集对比验证。

索引建议的写入代价

新建索引会降低写入性能。优化建议需要评估查询频率与写入频率的比值，高写入低查询的表不宜加索引。

适用边界：AI SQL 优化适合慢查询数量多、DBA 资源不足的场景。对于关键业务 SQL，仍需人工审核优化建议。

五、总结

AI SQL 优化将执行计划解读和改写建议从经验驱动升级为推理驱动。落地路线建议：

1. 起步阶段：实现执行计划解析器，将 EXPLAIN 输出结构化。
2. 优化阶段：构建多源上下文 Prompt，让 LLM 基于完整信息生成优化建议。
3. 强化阶段：实现建议验证层，通过代价预估和结果集对比确保优化有效且语义等价。
4. 精细化阶段：积累优化案例库，相似查询可复用历史优化方案。