AI 驱动的索引推荐：从执行计划分析到智能优化建议
一、索引优化的经验陷阱：DBA 的直觉不是银弹
数据库索引优化是性能调优的核心环节，但索引的选择远非"给 WHERE 条件的列加索引"这么简单。复合索引的列顺序、覆盖索引的设计、索引的选择性评估、写入代价的权衡——每个决策都需要综合考量查询模式、数据分布和业务优先级。传统索引优化依赖 DBA 的经验判断，但经验有两个局限：其一，面对数百张表和数千条慢查询时，人工审查无法覆盖全量；其二，数据分布变化后，曾经的优化可能变为负优化。
AI 驱动的索引推荐通过分析执行计划、查询模式和统计信息，自动识别索引缺失和冗余，并给出带代价评估的推荐方案。核心思路是：将 DBA 的分析过程形式化为可自动执行的规则和模型，覆盖全量查询，持续适应数据变化。
二、AI 索引推荐架构：从慢查询到索引方案的决策链
AI 索引推荐的输入是慢查询日志和统计信息，输出是带代价评估的索引推荐。决策链分为四步：慢查询采集 → 执行计划分析 → 索引方案生成 → 代价评估与排序。
flowchart TB
A[慢查询日志] –> B[执行计划采集<br/>EXPLAIN ANALYZE]
B –> C[扫描类型识别<br/>全表扫描/索引扫描]
C –> D{瓶颈判定}
D –>|全表扫描| E[索引缺失]
D –>|低效索引扫描| F[索引不优]
D –>|排序开销大| G[缺少排序索引]
E –> H[索引方案生成]
F –> H
G –> H
H –> I[候选索引列表]
I –> J[代价评估模型]
J –> K[查询加速收益]
J –> L[写入开销代价]
J –> M[存储空间成本]
K –> N[综合评分排序]
L –> N
M –> N
N –> O[推荐索引 TOP-N]
O –> P[人工审核 → 上线]
代价评估是推荐质量的关键。一个索引可能将查询延迟从 500ms 降到 5ms，但如果该查询每天只执行 10 次，而写入操作每秒 1000 次，索引的写入开销可能远超查询收益。AI 推荐必须量化"收益 – 代价"的净值。
三、生产级代码实现：执行计划分析与索引推荐
3.1 执行计划解析器
import re
from dataclasses import dataclass, field
from typing import List, Optional
@dataclass
class ExecutionPlan:
"""MySQL 执行计划解析结果"""
id: int
select_type: str
table: str
partitions: Optional[str]
type: str # ALL, index, range, ref, eq_ref, const
possible_keys: List[str]
key: Optional[str] # 实际使用的索引
key_len: Optional[int]
ref: Optional[str]
rows: int # 预估扫描行数
filtered: float # 过滤比例
extra: str # Using where, Using filesort, etc.
@dataclass
class QueryAnalysis:
"""查询分析结果"""
sql: str
tables: List[str]
full_scan_tables: List[str] # 全表扫描的表
filesort_tables: List[str] # 需要文件排序的表
low_efficiency_indexes: List[dict] # 低效索引使用
where_columns: List[str] # WHERE 条件列
order_by_columns: List[str] # ORDER BY 列
group_by_columns: List[str] # GROUP BY 列
class ExecutionPlanAnalyzer:
"""执行计划分析器"""
def analyze(self, sql: str, explain_rows: list) -> QueryAnalysis:
"""分析执行计划，识别性能瓶颈"""
plans = [self._parse_row(row) for row in explain_rows]
full_scan = []
filesort = []
low_efficiency = []
for plan in plans:
# 识别全表扫描
# 为什么关注 type=ALL：全表扫描意味着
# 没有索引可用，是最大的性能瓶颈；
# 其他扫描类型（index, range, ref）
# 至少利用了部分索引
if plan.type == "ALL" and plan.rows > 1000:
full_scan.append({
"table": plan.table,
"estimated_rows": plan.rows,
"possible_keys": plan.possible_keys,
})
# 识别文件排序
# 为什么关注 Using filesort：文件排序意味着
# MySQL 无法利用索引的有序性，需要额外排序；
# 大数据量下排序开销可能超过查询本身
if "Using filesort" in plan.extra:
filesort.append({
"table": plan.table,
"extra": plan.extra,
})
# 识别低效索引使用
if plan.type == "index" and plan.rows > 10000:
low_efficiency.append({
"table": plan.table,
"index": plan.key,
"estimated_rows": plan.rows,
"reason": "索引扫描行数过多，可能缺少更精确的索引",
})
# 提取查询涉及的列
where_cols = self._extract_where_columns(sql)
order_cols = self._extract_order_by_columns(sql)
group_cols = self._extract_group_by_columns(sql)
return QueryAnalysis(
sql=sql,
tables=[p.table for p in plans],
full_scan_tables=full_scan,
filesort_tables=filesort,
low_efficiency_indexes=low_efficiency,
where_columns=where_cols,
order_by_columns=order_cols,
group_by_columns=group_cols,
)
def _parse_row(self, row: dict) -> ExecutionPlan:
"""解析 EXPLAIN 输出的一行"""
return ExecutionPlan(
id=row.get("id", 0),
select_type=row.get("select_type", ""),
table=row.get("table", ""),
partitions=row.get("partitions"),
type=row.get("type", "ALL"),
possible_keys=row.get("possible_keys", "").split(",")
if row.get("possible_keys") else [],
key=row.get("key"),
key_len=row.get("key_len"),
ref=row.get("ref"),
rows=row.get("rows", 0),
filtered=row.get("filtered", 100.0),
extra=row.get("Extra", ""),
)
3.2 索引方案生成器
class IndexRecommender:
"""索引推荐生成器"""
def recommend(self, analysis: QueryAnalysis,
table_stats: dict) -> List[IndexRecommendation]:
"""根据查询分析生成索引推荐"""
recommendations = []
# 为全表扫描的表生成索引推荐
for scan in analysis.full_scan_tables:
table = scan["table"]
# 获取该表在查询中涉及的列
relevant_cols = self._get_relevant_columns(
table, analysis)
if not relevant_cols:
continue
# 生成复合索引方案
# 为什么生成复合索引而非单列索引：
# 复合索引可以同时覆盖 WHERE、ORDER BY
# 和 GROUP BY，一个复合索引的收益通常
# 大于多个单列索引；且复合索引数量更少，
# 写入开销更低
index_candidates = self._generate_index_candidates(
table, relevant_cols, analysis)
for candidate in index_candidates:
# 评估索引收益
benefit = self._estimate_query_benefit(
candidate, scan, analysis)
# 评估索引代价
cost = self._estimate_write_cost(
candidate, table, table_stats)
recommendations.append(IndexRecommendation(
table=table,
columns=candidate["columns"],
index_type=candidate["type"],
benefit_score=benefit,
cost_score=cost,
net_score=benefit – cost,
reason=candidate["reason"],
))
# 为文件排序生成排序索引
for sort in analysis.filesort_tables:
table = sort["table"]
order_cols = self._get_order_columns(table, analysis)
where_cols = self._get_where_columns(table, analysis)
if order_cols:
# 排序索引：WHERE 列 + ORDER BY 列
# 为什么 WHERE 列在前：索引的最左前缀
# 规则要求 WHERE 条件列在前面才能利用
# 索引过滤，ORDER BY 列在后面才能利用
# 索引的有序性避免排序
combined = where_cols + order_cols
recommendations.append(IndexRecommendation(
table=table,
columns=combined,
index_type="btree",
benefit_score=8.0,
cost_score=2.0,
net_score=6.0,
reason=f"消除文件排序: WHERE {where_cols} "
f"ORDER BY {order_cols}",
))
# 按净收益排序
recommendations.sort(key=lambda r: r.net_score, reverse=True)
return recommendations
def _generate_index_candidates(self, table, columns,
analysis):
"""生成索引候选方案"""
candidates = []
# 方案1: 等值条件列 + 排序列
eq_cols = [c for c in columns
if self._is_equality_condition(c, analysis)]
sort_cols = analysis.order_by_columns
if eq_cols:
candidates.append({
"columns": eq_cols + sort_cols,
"type": "btree",
"reason": f"等值过滤 + 排序: {eq_cols} + {sort_cols}",
})
# 方案2: 范围条件列
range_cols = [c for c in columns
if self._is_range_condition(c, analysis)]
if range_cols:
# 范围列放在最后
# 为什么范围列放最后：范围条件（>, <, BETWEEN）
# 之后的索引列无法利用索引有序性；
# 将范围列放在复合索引最后，确保前面的
# 等值列能充分利用索引
prefix_cols = [c for c in eq_cols
if c not in range_cols]
candidates.append({
"columns": prefix_cols + range_cols,
"type": "btree",
"reason": f"等值 + 范围: {prefix_cols} + {range_cols}",
})
return candidates
def _estimate_query_benefit(self, candidate, scan,
analysis):
"""估算查询加速收益（0-10 分）"""
estimated_rows = scan.get("estimated_rows", 0)
if estimated_rows == 0:
return 0
# 全表扫描行数越多，索引收益越大
# 为什么用对数缩放：100 万行到 10 万行的收益
# 远大于 1 万行到 1000 行；对数缩放
# 避免大表评分过高
import math
row_score = min(10, math.log10(estimated_rows))
# 覆盖索引额外加分
if candidate["type"] == "covering":
row_score += 2
return min(10, row_score)
def _estimate_write_cost(self, candidate, table,
table_stats):
"""估算写入代价（0-10 分）"""
stats = table_stats.get(table, {})
write_qps = stats.get("write_qps", 0)
# 写入 QPS 越高，索引代价越大
# 为什么关注写入 QPS：每次 INSERT/UPDATE/DELETE
# 都需要更新所有相关索引，写入密集的表
# 上加索引会显著增加写入延迟
if write_qps > 1000:
return 7
elif write_qps > 100:
return 4
elif write_qps > 10:
return 2
return 1
3.3 冗余索引检测
class RedundantIndexDetector:
"""冗余索引检测器"""
def detect(self, table: str,
existing_indexes: List[dict]) -> List[dict]:
"""检测冗余索引"""
redundant = []
# 按列前缀分组
# 为什么检测冗余：冗余索引浪费存储空间
# 和写入性能；如果 INDEX(a, b) 已存在，
# INDEX(a) 就是冗余的，因为 (a, b) 的
# 最左前缀可以替代 (a)
for i, idx1 in enumerate(existing_indexes):
for idx2 in existing_indexes[i + 1:]:
cols1 = idx1["columns"]
cols2 = idx2["columns"]
# 检查前缀关系
if len(cols1) > len(cols2):
longer, shorter = cols1, cols2
longer_idx, shorter_idx = idx1, idx2
else:
longer, shorter = cols2, cols1
longer_idx, shorter_idx = idx2, idx1
# shorter 是 longer 的前缀 → shorter 冗余
if longer[:len(shorter)] == shorter:
redundant.append({
"redundant_index": shorter_idx["name"],
"covered_by": longer_idx["name"],
"reason": (f"索引 {shorter_idx['name']}({shorter}) "
f"是 {longer_idx['name']}({longer}) 的前缀"),
})
return redundant
四、AI 索引推荐的架构权衡：推荐精度、数据时效与安全边界
推荐精度的局限：MySQL 的 EXPLAIN 预估行数基于统计信息，可能不准确。统计信息过期时，推荐方案的收益评估会偏差。建议在推荐前先执行 ANALYZE TABLE 更新统计信息，但 ANALYZE TABLE 在大表上可能耗时较长。
数据分布变化的影响：索引的选择性依赖数据分布。一个在当前数据下高选择性的索引，在数据分布变化后可能变为低选择性。AI 推荐需要定期重新评估，而非一次推荐永久有效。建议设置每周自动重新分析慢查询。
安全边界：自动上线 vs 人工审核：AI 推荐的索引不应自动上线——错误的索引可能导致写入性能严重下降，甚至锁表。建议 AI 推荐只生成方案，DBA 审核后在预发环境验证，确认无副作用后再上线。Critical 表（如订单表）的索引变更必须经过压测。
虚拟索引验证：部分数据库（如 MySQL 8.0 的 sys 库）支持虚拟索引（Invisible Index），索引存在但不被优化器使用。可以先创建虚拟索引，用 FORCE INDEX 验证效果，确认收益后再设为可见。
五、总结
AI 驱动的索引推荐将 DBA 的分析过程形式化为可自动执行的流程：慢查询采集 → 执行计划分析 → 索引方案生成 → 代价评估。复合索引的设计遵循"等值列在前、范围列在后、排序列在尾"的原则。代价评估必须量化查询收益和写入开销的净值。落地时建议先在非核心表上验证推荐效果，建立信任后再扩展到核心表。AI 推荐是辅助决策工具，最终上线决策权在 DBA。
