AI 数据库优化：当机器学习遇见查询计划，内核级调优的实践路径

2026/6/24 4:11:34

AI 数据库优化：当机器学习遇见查询计划，内核级调优的实践路径

一、传统优化器的天花板：为什么统计信息总在撒谎

数据库查询优化器依赖统计信息做代价估算，但统计信息天然滞后。一次大批量导入后，表的行数估算可能偏差 10 倍以上，优化器据此选择的执行计划与实际代价南辕北辙。手动ANALYZE TABLE只是临时补救，无法根治信息过时的问题。

更深层的问题在于：传统优化器的代价模型是简化的数学公式，无法捕捉数据分布的复杂特征。比如两个列的联合分布高度倾斜，单列统计信息完全无法反映这种相关性，优化器就会选择次优的连接顺序。AI 优化的切入点正在于此——用学习到的数据分布特征替代手工统计信息，用模型预测替代公式估算。

二、AI 查询优化的技术架构：从学习型代价模型到智能索引推荐

AI 数据库优化的核心思路是将机器学习嵌入查询优化的关键决策点。整体架构如下：

graph LR A[SQL 输入] --> B[解析与逻辑计划] B --> C[传统优化器] B --> D[学习型代价模型] D --> |"预测代价"| C C --> E[物理执行计划] E --> F[执行引擎] F --> |"运行时统计"| G[反馈回路] G --> D D --> |"索引推荐"| H[索引管理] H --> |"统计信息增强"| C style D fill:#f96,stroke:#333 style G fill:#9cf,stroke:#333

学习型代价模型。核心思路是用历史查询的执行统计（实际行数、IO 次数、执行时间）训练模型，预测新查询的代价。特征包括：查询算子类型、表统计信息、谓词选择性估计等。模型输出预测的 CPU 代价和 IO 代价，替代传统公式。

反馈回路。每次查询执行后，实际运行时统计回写到训练数据集。模型定期增量训练，逐步修正预测偏差。这是 AI 优化器区别于静态优化器的关键——它能从错误中学习。

智能索引推荐。基于工作负载特征，模型推荐可能提升性能的索引组合。不是穷举所有可能，而是基于历史查询模式筛选候选索引，再用代价模型评估收益。

三、生产级 AI 优化模块实现

3.1 学习型代价预测服务

import logging from dataclasses import dataclass from typing import Optional import numpy as np from sklearn.ensemble import GradientBoostingRegressor logger = logging.getLogger("ai_cost_model") @dataclass class QueryFeatures: """查询特征：用于代价预测的输入""" scan_rows: int # 预估扫描行数 filter_selectivity: float # 过滤选择性 (0~1) join_type: int # 连接类型编码 (0=NL, 1=Hash, 2=Merge) index_used: int # 是否使用索引 (0/1) agg_columns: int # 聚合列数 sort_columns: int # 排序列数 table_size_mb: float # 表大小(MB) @dataclass class CostPrediction: """代价预测结果""" predicted_time_ms: float # 预测执行时间(ms) predicted_io_count: int # 预测IO次数 confidence: float # 预测置信度 (0~1) class AICostPredictor: """ 学习型代价预测器 基于历史查询执行统计训练，预测新查询的执行代价 """ def __init__(self, model_path: Optional[str] = None): # 使用梯度提升树，对特征交互有较好的捕捉能力 self._time_model = GradientBoostingRegressor( n_estimators=200, max_depth=6, learning_rate=0.1, subsample=0.8 ) self._io_model = GradientBoostingRegressor( n_estimators=150, max_depth=5, learning_rate=0.1 ) self._is_trained = False self._min_samples = 100 # 最少训练样本数 if model_path: self._load_model(model_path) def predict(self, features: QueryFeatures) -> CostPrediction: """ 预测查询代价 未训练时返回默认值，避免线上服务中断 """ if not self._is_trained: # 未训练时返回保守估计，不阻塞查询执行 logger.warning("模型未训练，返回默认代价估计") return CostPrediction( predicted_time_ms=100.0, predicted_io_count=1000, confidence=0.0 ) x = np.array([[ features.scan_rows, features.filter_selectivity, features.join_type, features.index_used, features.agg_columns, features.sort_columns, features.table_size_mb ]]) try: pred_time = max(0.1, self._time_model.predict(x)[0]) pred_io = max(1, int(self._io_model.predict(x)[0])) # 置信度基于训练数据量，样本越多越可信 confidence = min(1.0, self._time_model.train_score_[-1]) return CostPrediction( predicted_time_ms=pred_time, predicted_io_count=pred_io, confidence=confidence ) except Exception as e: logger.error(f"代价预测异常: {e}") return CostPrediction( predicted_time_ms=100.0, predicted_io_count=1000, confidence=0.0 ) def train(self, features_list: list[QueryFeatures], actual_times: list[float], actual_ios: list[int]) -> dict: """ 增量训练模型 返回训练指标，便于监控模型质量 """ if len(features_list) < self._min_samples: logger.info( f"训练样本不足: {len(features_list)}/{self._min_samples}" ) return {"status": "insufficient_samples"} X = np.array([[ f.scan_rows, f.filter_selectivity, f.join_type, f.index_used, f.agg_columns, f.sort_columns, f.table_size_mb ] for f in features_list]) y_time = np.array(actual_times) y_io = np.array(actual_ios, dtype=float) try: self._time_model.fit(X, y_time) self._io_model.fit(X, y_io) self._is_trained = True # 评估训练质量 time_r2 = self._time_model.score(X, y_time) io_r2 = self._io_model.score(X, y_io) logger.info( f"模型训练完成: time_r2={time_r2:.4f}, io_r2={io_r2:.4f}" ) return { "status": "success", "time_r2": time_r2, "io_r2": io_r2, "sample_count": len(features_list) } except Exception as e: logger.error(f"模型训练失败: {e}") return {"status": "error", "message": str(e)}

3.2 智能索引推荐

from collections import defaultdict from typing import List, Tuple class IndexAdvisor: """ 基于工作负载的智能索引推荐 分析查询模式，推荐高收益索引组合 """ def __init__(self, cost_predictor: AICostPredictor): self.predictor = cost_predictor # 记录列的查询频率和过滤频率 self._column_access = defaultdict(int) self._column_filter = defaultdict(int) self._column_combos = defaultdict(int) def analyze_workload(self, queries: List[dict]) -> List[dict]: """ 分析工作负载，提取索引候选 queries: [{"table": str, "filter_cols": list, "sort_cols": list, "join_cols": list}] """ for q in queries: table = q["table"] # 过滤列是最高优先级的索引候选 for col in q.get("filter_cols", []): key = f"{table}.{col}" self._column_filter[key] += 1 # 排序列次之 for col in q.get("sort_cols", []): key = f"{table}.{col}" self._column_access[key] += 1 # 连接列也是索引候选 for col in q.get("join_cols", []): key = f"{table}.{col}" self._column_access[key] += 2 # 连接列权重更高 # 多列组合模式 filter_cols = q.get("filter_cols", []) if len(filter_cols) >= 2: combo = tuple(sorted(filter_cols)) self._column_combos[(table, combo)] += 1 return self._generate_recommendations() def _generate_recommendations(self) -> List[dict]: """生成索引推荐列表，按预估收益排序""" recommendations = [] # 单列索引推荐 for key, freq in sorted( self._column_filter.items(), key=lambda x: x[1], reverse=True ): table, col = key.split(".") # 过滤频率超过阈值才推荐 if freq >= 5: recommendations.append({ "type": "single_column", "table": table, "columns": [col], "estimated_benefit": freq, "reason": f"过滤频率={freq}，高频查询条件列" }) # 组合索引推荐 for (table, cols), freq in sorted( self._column_combos.items(), key=lambda x: x[1], reverse=True ): if freq >= 3: recommendations.append({ "type": "composite", "table": table, "columns": list(cols), "estimated_benefit": freq * 2, "reason": f"组合过滤频率={freq}，多列联合查询" }) # 按收益降序排列 recommendations.sort( key=lambda x: x["estimated_benefit"], reverse=True ) return recommendations[:10] # 最多推荐10个索引

四、AI 优化的现实约束与妥协

冷启动问题。模型需要足够的历史数据才能有效预测。新上线的数据库没有历史查询统计，AI 优化器在冷启动阶段基本等于摆设。解决方案是在灰度阶段先用传统优化器，积累数据后再切换。

模型漂移。数据分布随业务变化而变化，训练好的模型会逐渐失效。需要持续收集运行时统计并定期重训练。但重训练本身消耗资源，频率过高影响在线服务，过低则模型退化。这是一个需要根据业务变化速度动态调整的参数。

可解释性缺失。优化器选择了一个"反直觉"的执行计划，DBA 无法理解原因。这在生产环境中是严重的信任问题。AI 优化器必须提供决策依据，至少要能解释"为什么这个计划的预测代价更低"。

安全边界。AI 推荐的执行计划不能比传统优化器差太多。必须设置回退机制：当 AI 预测的代价与实际代价偏差超过阈值时，自动回退到传统优化器。这是生产环境的基本安全网。

五、总结

AI 数据库优化的核心价值在于用数据驱动替代规则驱动，从历史执行统计中学习更准确的代价预测。学习型代价模型和智能索引推荐是两个最成熟的落地方向。但 AI 优化不是万能的，冷启动、模型漂移、可解释性和安全边界是必须正视的约束。落地路径应该是：先用传统优化器积累数据，再在只读查询上灰度验证 AI 优化，最后逐步扩大覆盖范围。每一步都需要有回退机制，确保 AI 优化不会比传统方案更差。