AI 驱动的索引推荐系统：从慢查询模式到最优索引组合的自动推导

国医中兴

1112人浏览 · 2026-06-14 11:14:31

国医中兴 · 2026-06-14 11:14:31 发布

AI 驱动的索引推荐系统：从慢查询模式到最优索引组合的自动推导

一、索引选择的经验陷阱：为什么 DBA 的"手感"不可靠

数据库索引的选择是影响查询性能的核心决策。一个缺失的索引可能导致全表扫描，查询时间从毫秒级飙升到分钟级；一个冗余的索引则浪费存储空间，增加写入开销。传统索引优化依赖 DBA 的经验判断——分析慢查询日志、解读执行计划、手动创建索引。但这种方法有三个根本性缺陷：其一，单条慢查询的索引推荐容易，但多条查询的索引组合优化是 NP-hard 问题；其二，索引的创建需要评估对写入性能的影响，DBA 往往只关注查询加速而忽略写入退化；其三，数据分布变化后，原有索引可能不再最优，但索引的定期审查几乎不会执行。

二、索引推荐架构：从慢查询分析到组合优化

AI 驱动的索引推荐核心思路是：收集慢查询和工作负载特征，通过代价模型评估候选索引的收益，再用组合优化算法在索引空间中搜索全局最优解。

flowchart TD
    A[慢查询日志] --> B[SQL 解析与模式提取]
    B --> C[工作负载特征<br/>频率/过滤条件/连接关系]
    C --> D[候选索引生成<br/>基于 WHERE/JOIN/ORDER BY]
    D --> E[代价模型评估<br/>查询加速收益]
    E --> F[写入代价评估<br/>INSERT/UPDATE 开销]
    F --> G[组合优化<br/>约束：存储上限/写入预算]
    G --> H[最优索引集推荐]
    H --> I[在线 A/B 验证]
    I --> J[效果量化与反馈]

关键设计决策在于代价模型的精度和组合优化算法的效率。代价模型需要准确预测索引对查询延迟的影响，组合优化需要在指数级的索引空间中高效搜索。

三、工程实现：SQL 解析、代价评估与组合优化

3.1 SQL 解析与候选索引生成

import sqlparse
from sqlparse.sql import Where, Identifier, IdentifierList
from dataclasses import dataclass
from typing import List, Set

@dataclass
class IndexCandidate:
    table: str
    columns: tuple[str, ...]
    index_type: str  # btree / hash / gin / gist
    estimated_size_mb: float
    covering_queries: List[str]  # 可加速的查询 ID

class IndexCandidateGenerator:
    def generate(self, slow_queries: List[dict]) -> List[IndexCandidate]:
        candidates = []

        for query in slow_queries:
            sql = query['sql']
            parsed = sqlparse.parse(sql)[0]

            # 提取 WHERE 子句中的过滤列
            where_columns = self._extract_where_columns(parsed)

            # 提取 JOIN 条件中的连接列
            join_columns = self._extract_join_columns(parsed)

            # 提取 ORDER BY 列
            order_columns = self._extract_order_columns(parsed)

            # 生成候选索引
            for table, columns in where_columns.items():
                # 单列索引
                for col in columns:
                    candidates.append(IndexCandidate(
                        table=table,
                        columns=(col,),
                        index_type='btree',
                        estimated_size_mb=0,  # 后续评估
                        covering_queries=[query['id']],
                    ))

                # 组合索引（遵循最左前缀原则）
                if len(columns) > 1:
                    # 按选择性排序：高选择性列在前
                    sorted_cols = self._sort_by_selectivity(
                        table, columns)
                    candidates.append(IndexCandidate(
                        table=table,
                        columns=tuple(sorted_cols),
                        index_type='btree',
                        estimated_size_mb=0,
                        covering_queries=[query['id']],
                    ))

            # 覆盖索引：包含 SELECT 和 WHERE 的所有列
            select_columns = self._extract_select_columns(parsed)
            for table in where_columns:
                all_cols = list(set(
                    where_columns.get(table, [])
                    + select_columns.get(table, [])
                ))
                if len(all_cols) > 1:
                    candidates.append(IndexCandidate(
                        table=table,
                        columns=tuple(all_cols),
                        index_type='btree',
                        estimated_size_mb=0,
                        covering_queries=[query['id']],
                    ))

        # 去重：相同表+相同列的候选合并
        return self._deduplicate(candidates)

3.2 代价模型评估

class IndexCostModel:
    def __init__(self, db_stats: DatabaseStatistics):
        self.stats = db_stats

    def estimate_query_benefit(self, query: dict,
                                index: IndexCandidate) -> float:
        """估算索引对查询的加速收益（毫秒）"""
        table_stats = self.stats.get_table_stats(index.table)
        current_cost = self._estimate_scan_cost(query, table_stats)

        # 使用索引后的代价
        indexed_cost = self._estimate_index_scan_cost(
            query, index, table_stats)

        benefit_ms = current_cost - indexed_cost
        return max(0, benefit_ms)

    def estimate_write_penalty(self, index: IndexCandidate,
                                write_rate: int) -> float:
        """估算索引对写入的额外开销（毫秒/事务）"""
        # B-Tree 索引的写入开销约为数据写入的 20%-40%
        index_maintenance_cost = (
            write_rate * 0.3 * index.estimated_size_mb / 100
        )
        return index_maintenance_cost

    def _estimate_scan_cost(self, query: dict,
                             table: TableStats) -> float:
        # 全表扫描代价 ≈ 行数 × 行大小 / 磁盘带宽
        seq_scan_cost = (
            table.row_count * table.avg_row_bytes
            / self.stats.disk_bandwidth_bytes_per_ms
        )
        return seq_scan_cost

    def _estimate_index_scan_cost(self, query: dict,
                                   index: IndexCandidate,
                                   table: TableStats) -> float:
        # 索引扫描代价 ≈ 索引高度 × 页访问 + 结果行 × 回表代价
        selectivity = self._estimate_selectivity(query, index, table)
        result_rows = table.row_count * selectivity

        index_height = max(3, int(
            table.row_count / (8160 / (len(index.columns) * 8))
        ) ** 0.5)

        index_seek_cost = index_height * 0.01  # 每层约 0.01ms
        table_fetch_cost = result_rows * 0.005  # 每行回表约 0.005ms

        return index_seek_cost + table_fetch_cost

3.3 组合优化

class IndexCombinationOptimizer:
    def optimize(self, candidates: List[IndexCandidate],
                 cost_model: IndexCostModel,
                 queries: List[dict],
                 storage_budget_mb: float,
                 write_budget_ms: float) -> List[IndexCandidate]:
        """贪心算法搜索最优索引组合"""
        selected = []
        remaining_budget = storage_budget_mb
        remaining_write_budget = write_budget_ms

        # 计算每个候选的收益/成本比
        scored = []
        for candidate in candidates:
            total_benefit = sum(
                cost_model.estimate_query_benefit(q, candidate)
                for q in queries
                if q['id'] in candidate.covering_queries
            )
            write_penalty = cost_model.estimate_write_penalty(
                candidate, self._get_write_rate(candidate.table))

            score = total_benefit / max(
                candidate.estimated_size_mb, 1)

            scored.append((candidate, total_benefit,
                          write_penalty, score))

        # 按收益/成本比降序排列
        scored.sort(key=lambda x: x[3], reverse=True)

        # 贪心选择
        for candidate, benefit, penalty, score in scored:
            if (candidate.estimated_size_mb <= remaining_budget
                and penalty <= remaining_write_budget):
                selected.append(candidate)
                remaining_budget -= candidate.estimated_size_mb
                remaining_write_budget -= penalty

        return selected

四、索引推荐的精度边界与工程风险

代价模型的统计信息依赖：代价模型依赖数据库的统计信息（行数、唯一值数量、数据分布）。统计信息过期时，代价估算可能严重偏离实际。建议在推荐前先执行 ANALYZE TABLE 更新统计信息，但大规模表的 ANALYZE 本身可能耗时较长。

组合优化的局部最优：贪心算法无法保证全局最优。两个单独收益不高的索引，组合后可能产生"1+1>2"的效果（如覆盖索引消除回表）。更精确的方案是整数线性规划（ILP），但 ILP 的求解时间随候选索引数量指数增长，实际场景中通常限制候选数量在 50 以内。

索引创建的锁风险：CREATE INDEX 在线创建（CONCURRENTLY）不会阻塞读写，但创建时间可能很长（大表数小时）。创建过程中会占用额外的 CPU 和 I/O 资源，可能影响在线业务。建议在低峰期执行索引创建，并监控创建进度。

索引维护的长期成本：索引推荐系统通常只考虑当前工作负载，但工作负载会随业务变化而变化。三个月前创建的索引，可能已经不再被任何查询使用。建议定期审计索引使用率（pg_stat_user_indexes 或 sys.schema_unused_indexes），清理无用索引。

五、总结

AI 驱动索引推荐的本质是将"DBA 经验驱动的索引决策"转化为"代价模型评估 + 组合优化搜索"的系统化方案。本文方案的核心链路为：慢查询解析 → 候选索引生成 → 代价模型评估 → 组合优化 → 在线验证。落地时需重点关注三个参数：存储预算（建议不超过数据量的 30%）、写入预算（建议索引维护不超过写入延迟的 20%）、候选索引上限（建议 50 个以内）。建议从单表查询的索引推荐开始验证，逐步扩展到多表 JOIN 和覆盖索引场景，并建立索引使用率的定期审计机制。