当前位置：首页 > news >正文

学习型索引与B+树的自适应混合方案

news 2026/6/8 12:21:34

学习型索引与B+树的自适应混合方案

一、B+树的固定结构困境：无法适配数据分布

B+树作为数据库索引的标配数据结构，其核心假设是数据均匀分布——每个节点存储固定数量的键值，树的高度取决于数据总量和节点大小。然而，实际数据分布往往高度倾斜：时间序列数据按时间递增，用户ID按注册顺序递增，金额字段集中在特定区间。B+树对倾斜分布的适应性差——热点区间的节点频繁分裂，冷区间的节点大量空闲。

学习型索引（Learned Index）通过机器学习模型学习数据的累积分布函数（CDF），将查找键映射到数据在排序数组中的近似位置。理论上，学习型索引可以用更少的空间实现更高的查找效率。但纯学习型索引在最坏情况下的性能不可控——模型预测偏差可能导致多次二分查找，退化为比B+树更慢。

本文将探讨学习型索引与B+树的自适应混合方案，在保证最坏情况性能的前提下，利用学习型模型优化热点区间的查找效率。

二、混合索引架构

2.1 整体设计

graph TB A[查找请求] --> B{数据区间判断} B -->|热点区间| C[学习型模型预测] B -->|冷区间| D[B+树查找] C --> E[局部二分搜索] E --> F[结果验证] D --> F F --> G{命中?} G -->|是| H[返回结果] G -->|否| I[回退B+树] I --> H

2.2 学习型模型实现

class LearnedIndexModel: """学习型索引模型：预测键在排序数组中的位置""" def __init__(self): # 使用两层模型：顶层粗粒度，底层细粒度 self.top_model = None # 线性回归 self.bottom_models = [] # 分段线性模型 def train(self, keys: np.ndarray, positions: np.ndarray): # 顶层：全局线性回归 self.top_model = self._fit_linear(keys, positions) # 将键空间划分为多个段 predictions = self.top_model.predict(keys) errors = np.abs(predictions - positions) max_error = np.max(errors) # 按误差分桶，每个桶训练一个底层模型 n_segments = max(1, len(keys) // 10000) segment_size = len(keys) // n_segments for i in range(n_segments): start = i * segment_size end = min((i + 1) * segment_size, len(keys)) segment_keys = keys[start:end] segment_pos = positions[start:end] model = self._fit_linear(segment_keys, segment_pos) self.bottom_models.append({ 'model': model, 'min_key': segment_keys[0], 'max_key': segment_keys[-1], 'max_error': self._compute_max_error( model, segment_keys, segment_pos) }) def predict(self, key: float) -> tuple: """预测键的位置，返回(位置, 误差范围)""" # 顶层预测确定段 top_pred = self.top_model.predict(np.array([[key]]))[0] # 找到对应的底层模型 segment = self._find_segment(key) if segment is None: return top_pred, float('inf') # 底层精确预测 pred = segment['model'].predict(np.array([[key]]))[0] error_bound = segment['max_error'] return int(pred), error_bound

2.3 混合索引查找

class HybridIndex: """学习型索引与B+树的混合索引""" def __init__(self, btree, learned_model): self.btree = btree self.learned_model = learned_model self.stats = {'learned_hits': 0, 'btree_fallbacks': 0} def lookup(self, key) -> Record: # 尝试学习型索引 pred_pos, error_bound = self.learned_model.predict(key) if error_bound == float('inf'): # 模型无法预测，回退B+树 self.stats['btree_fallbacks'] += 1 return self.btree.lookup(key) # 在预测位置附近进行局部搜索 lo = max(0, pred_pos - error_bound) hi = pred_pos + error_bound result = self._local_search(key, lo, hi) if result is not None: self.stats['learned_hits'] += 1 return result # 局部搜索未命中，回退B+树 self.stats['btree_fallbacks'] += 1 return self.btree.lookup(key)