AI 数据库内核：当学习型查询优化器接管成本估算，谁为错误计划买单？

先抛出一个核心问题：数据库查询优化器的底层假设是统计信息能准确反映数据分布，但这个假设在生产环境中几乎从未成立。一张每天新增千万行数据的订单表，ANALYZE TABLE 的执行频率根本跟不上数据变化的节奏，优化器拿着过时的统计信息做出的成本估算，与实际执行成本的偏差可能高达数十倍。更棘手的是多列相关性问题——例如 WHERE city='Shanghai' AND district='Pudong' 的过滤因子并不是两列选择率的简单乘积，但优化器只能按照独立假设来估算，结果选择了 city 索引作为驱动，实际过滤效果远不如预期。

那么，既然统计信息天然不可靠，不如让优化器从历史执行记录中学习——建立查询特征到最优计划的映射模型。这就是学习型查询优化器（Learned Query Optimizer）的核心逻辑：用数据驱动替代规则驱动，用模型预测替代统计估算。

一、成本模型失灵的常态化：传统优化器的统计信息困局

学习型查询优化器并非一个独立的模块，而是嵌入传统优化器管线中的增强组件。它的架构可以拆成四个层次，如下图所示：

flowchart TB
    subgraph 传统优化器管线
        A[SQL 输入] --> B[解析与语义分析]
        B --> C[逻辑改写]
        C --> D[成本估算与计划枚举]
        D --> E[执行计划输出]
    end
    subgraph 学习型增强层
        F[查询特征编码器
编码表/列/谓词/Join拓扑]
        F --> G[计划评分模型
预测各候选计划执行耗时]
        G --> H[计划选择策略
融合模型评分与成本估算]
        H --> D
        I[执行反馈收集器
采集实际耗时/行数/I/O]
        I --> J[模型增量训练
周期性更新模型权重]
        J --> G
    end
    E --> I
    style F fill:#e53e3e,color:#fff
    style G fill:#e53e3e,color:#fff
    style H fill:#dd6b20,color:#fff
    style I fill:#38a169,color:#fff
    style J fill:#38a169,color:#fff

查询特征编码器：将 SQL 查询转化为模型可消费的向量表示。编码内容涉及表与列、谓词类型与选择率估算、Join 图的拓扑结构、排序与分组操作。关键挑战在于查询的变体极多——同一语义的 SQL 可能有数十种写法，编码器必须具备一定的语义归一化能力。

计划评分模型：这是核心推理模块。输入一组候选执行计划的特征向量，输出每个计划的预测执行耗时。主流方案使用梯度提升树（如 XGBoost）或小型神经网络——因为训练数据量通常在万级到十万级，深度模型容易过拟合。

计划选择策略：不能完全信任模型评分，必须与传统成本估算融合。常见策略是加权融合——final_score = alpha * model_score + (1 - alpha) * cost_model_score，alpha 根据模型置信度动态调整。模型对未见过的查询模式置信度低，此时退回传统成本估算。

执行反馈闭环：每次查询执行后，收集实际耗时、扫描行数、I/O 次数等指标，作为训练样本。增量训练周期通常为小时级，避免训练开销影响在线服务。

二、学习型查询优化器的架构：从特征提取到计划推荐的全链路

下面进入工程实现细节。先看查询特征编码方案的一个实践示例：

import numpy as np
from typing import List, Dict

class QueryFeatureEncoder:
    """将 SQL 查询编码为固定维度向量，供评分模型消费"""
    def __init__(self, table_vocab: Dict[str, int], column_vocab: Dict[str, int], 
                 max_tables: int = 16, max_predicates: int = 32):
        self.table_vocab = table_vocab      # 表名到索引的映射
        self.column_vocab = column_vocab    # 列名到索引的映射
        self.max_tables = max_tables
        self.max_predicates = max_predicates

    def encode(self, query_plan: Dict) -> np.ndarray:
        """编码查询计划为特征向量
        维度设计：表参与向量 + 列访问向量 + 谓词特征 + Join拓扑特征
        """
        # 表参与特征：one-hot 编码涉及的表
        table_feat = np.zeros(len(self.table_vocab))
        for t in query_plan.get('tables', []):
            if t in self.table_vocab:
                table_feat[self.table_vocab[t]] = 1.0

        # 列访问特征：统计每列被谓词引用的次数
        col_feat = np.zeros(len(self.column_vocab))
        for pred in query_plan.get('predicates', []):
            col_name = pred.get('column')
            if col_name in self.column_vocab:
                col_feat[self.column_vocab[col_name]] += 1.0

        # 谓词特征：[选择率估算, 谓词类型编码, 是否索引覆盖]
        pred_feat = np.zeros(self.max_predicates * 3)
        for i, pred in enumerate(query_plan.get('predicates', [])):
            if i >= self.max_predicates:
                break
            base = i * 3
            pred_feat[base] = pred.get('selectivity', 0.5)
            pred_feat[base + 1] = self._encode_predicate_type(pred.get('type'))
            pred_feat[base + 2] = 1.0 if pred.get('index_covered') else 0.0

        # Join 拓扑特征：邻接矩阵的上三角展开
        join_feat = np.zeros(self.max_tables * (self.max_tables - 1) // 2)
        join_pairs = query_plan.get('join_pairs', [])
        for pair in join_pairs:
            t1_idx = self.table_vocab.get(pair[0], -1)
            t2_idx = self.table_vocab.get(pair[1], -1)
            if t1_idx >= 0 and t2_idx >= 0:
                flat_idx = self._pair_to_flat_index(t1_idx, t2_idx)
                if flat_idx < len(join_feat):
                    join_feat[flat_idx] = 1.0

        return np.concatenate([table_feat, col_feat, pred_feat, join_feat])

    def _encode_predicate_type(self, pred_type: str) -> float:
        """谓词类型编码：等值=1.0, 范围=0.5, LIKE=0.3, 其他=0.1"""
        mapping = {'eq': 1.0, 'range': 0.5, 'like': 0.3}
        return mapping.get(pred_type, 0.1)

    def _pair_to_flat_index(self, i: int, j: int) -> int:
        """将邻接矩阵坐标转换为上三角一维索引"""
        if i > j:
            i, j = j, i
        return i * self.max_tables - i * (i + 1) // 2 + j - i - 1

然后是计划评分模型与置信度校准的实现：

import xgboost as xgb
from sklearn.calibration import CalibratedClassifierCV

class PlanScoringModel:
    """基于 XGBoost 的执行计划评分模型，带置信度校准"""
    def __init__(self, model_path: str = None):
        self.model = None
        self.calibrator = None
        self.confidence_threshold = 0.7  # 低于此阈值退回传统成本估算
        if model_path:
            self.model = xgb.Booster()
            self.model.load_model(model_path)

    def predict(self, plan_features: np.ndarray) -> tuple:
        """预测执行计划耗时，返回 (预测耗时ms, 置信度)
        置信度基于模型对同类查询的预测方差估算
        """
        if self.model is None:
            return None, 0.0
        dmatrix = xgb.DMatrix(plan_features.reshape(1, -1))
        predicted_time = self.model.predict(dmatrix)[0]
        # 置信度估算：使用模型叶子权重的方差作为不确定性指标
        # 方差越大，说明该查询落在训练数据稀疏区域
        leaf_indices = self.model.predict(dmatrix, pred_leaf=True)
        confidence = self._estimate_confidence(leaf_indices)
        return float(predicted_time), confidence

    def _estimate_confidence(self, leaf_indices: np.ndarray) -> float:
        """基于叶子节点覆盖的样本数估算置信度
        样本数越多，置信度越高
        """
        # 简化实现：实际应查询每个叶子节点的训练样本覆盖数
        # 此处用启发式：叶子索引越接近训练时常见区间，置信度越高
        return 0.8  # 生产环境需替换为真实校准逻辑

    def incremental_train(self, features: np.ndarray,
                          labels: np.ndarray,
                          sample_weights: np.ndarray = None):
        """增量训练：使用新采集的执行反馈更新模型
        sample_weights: 近期样本权重更高，衰减因子 0.95
        """
        if sample_weights is None:
            # 时间衰减权重：越近期的样本权重越高
            n = len(labels)
            sample_weights = np.array([0.95 ** (n - i - 1) for i in range(n)])
        params = {
            'objective': 'reg:squarederror',
            'max_depth': 6,
            'learning_rate': 0.1,
            'subsample': 0.8,
            'colsample_bytree': 0.8,
        }
        dtrain = xgb.DMatrix(features, label=labels, weight=sample_weights)
        self.model = xgb.train(params, dtrain, num_boost_round=100)

最后是融合策略的实现——模型与传统成本估算如何协同：

class HybridPlanSelector:
    """融合模型评分与传统成本估算的计划选择器"""
    def __init__(self, scoring_model: PlanScoringModel, alpha: float = 0.6):
        self.model = scoring_model
        self.alpha = alpha  # 模型评分权重，(1-alpha) 为成本估算权重

    def select_plan(self, candidate_plans: List[Dict]) -> Dict:
        """从候选计划中选择最优计划
        每个候选计划包含：features, cost_estimate, plan_object
        """
        best_plan = None
        best_score = float('inf')
        for plan in candidate_plans:
            model_time, confidence = self.model.predict(plan['features'])
            if confidence < self.model.confidence_threshold:
                # 置信度不足，完全依赖传统成本估算
                effective_alpha = 0.0
            else:
                # 置信度越高，模型权重越大
                effective_alpha = self.alpha * confidence
            # 归一化：将模型预测耗时和成本估算统一到 [0, 1] 区间
            model_score = model_time / 1000.0 if model_time else 1.0
            cost_score = plan['cost_estimate'] / 1e6
            final_score = (effective_alpha * model_score +
                           (1 - effective_alpha) * cost_score)
            if final_score < best_score:
                best_score = final_score
                best_plan = plan
        return best_plan

三、生产级学习型优化器的实现与关键工程决策

学习型优化器最大的风险不是模型不准，而是模型不准时你根本不知道它不准。

冷启动问题：模型在训练数据不足时，预测质量极差。新上线的数据库实例没有任何历史执行记录，模型只能退回传统成本估算。但退回机制本身也需要监控——如果模型长期处于低置信度状态，说明训练数据采集链路可能有问题，而非模型能力不足。

分布漂移：业务模式变化导致查询分布漂移。电商大促期间的查询模式与日常完全不同，模型在平时积累的知识在大促期间可能产生系统性偏差。增量训练的周期如果跟不上漂移速度，模型会持续输出错误计划。

不可解释性：模型选择了全表扫描而非索引扫描，你无法像分析 EXPLAIN 那样追溯决策路径。当生产事故发生时，“模型决定的”不是一个可接受的根因。因此，学习型优化器必须提供降级开关，允许 DBA 在紧急情况下一键回退到传统优化器。

适用边界：学习型优化器适合查询模式相对稳定、执行反馈数据充足的场景（如内部 BI 平台、数据仓库）。对于查询模式高度随机、或对计划稳定性要求极高的金融核心系统，传统优化器 + 人工 Hint 仍是更可靠的选择。

四、学习型优化器的信任危机与工程边界

总结一下：学习型查询优化器通过历史执行反馈训练预测模型，试图弥补传统成本模型在统计信息时效性和多列相关性上的固有缺陷。其架构核心是查询特征编码、计划评分模型、融合选择策略和执行反馈闭环四个组件的协同。但模型的不可解释性、冷启动阶段的低置信度、以及查询分布漂移带来的预测偏差，构成了学习型优化器在生产落地的三大工程风险。实践中应采用“模型增强而非替代”的融合策略，根据模型置信度动态调整权重，并保留完整的降级回退机制。学习型优化器的价值不在于完全取代传统成本估算，而在于为那些统计信息严重失真的高频查询提供第二意见。