TiDB 自动统计信息优先级队列内存化改造从周期重建到增量维护的设计与实现【免费下载链接】tidbTiDB is built for agentic workloads that grow unpredictably, with ACID guarantees and native support for transactions, analytics, and vector search. No data silos. No noisy neighbors. No infrastructure ceiling.项目地址: https://gitcode.com/GitHub_Trending/ti/tidb导读本文基于 TiDB 仓库中的设计文档 2024-09-06-maintain-priority-queue-in-memory.md深入剖析自动统计信息auto analyze优先级队列的一次关键架构升级将每 3 秒扫描全量 schema/stats 缓存重建队列的做法改造为在内存中维护一个线程安全的堆并通过 DML/DDL 增量事件驱动更新。读完本文你将理解优先级打分公式的由来、DML 增量更新与 DDL 事件订阅的完整方案、mysql.ddl_notifier系统表与DDLNotifier订阅框架的设计细节以及这些设计在当前仓库源码pkg/statistics/handle/autoanalyze/priorityqueue、pkg/ddl/notifier中的落地形态。背景为什么不能继续每次重建优先级队列TiDB 使用优先级队列来自动更新统计信息。旧方案的问题是每次都以全量扫描的方式重建队列这会带来两个难以忍受的代价Information Schema v2 之后不再全量缓存表信息。TiDB 需要从 TiKV 拉取表信息当表数量很大时速度很慢。因此任何需要加载全部表信息的队列操作都必须被避免。扫描过于频繁导致 CPU 开销高。每 3 秒重建一次队列意味着每 3 秒就要扫描一遍全部 stats cache 与 schema cache表一多就会显著消耗 CPU对应 issuehttps://github.com/pingcap/tidb/issues/49972。设计文档给出的结论非常直接与其反复重建不如把优先级队列常驻内存用增量事件来维护它。这一背景的延续脉络可以参考更早的设计文档 2023-11-29-priority-queue-for-auto-analyze.md后者确立了加权排序的优先级打分思想而本文档要解决的是它的工程落地代价——队列维护成本。当前实现优先级打分公式新设计并不改变打分语义而是沿用既有加权打分体系。队列中的每个分析任务AnalysisJob由四项指标加权求和得到优先级分数。四项核心指标指标含义分数计算数据来源Percentage of Change距上次分析的变更比例log10(1 变更比例)未分析过的表变更比例视为 100%Stats CacheTable Size表大小 行数 × 参与分析的列数小表应优先于大表对log10(1 表大小)取惩罚项即1 - log10(1 表大小)Stats Cache Table InfoAnalysis Interval距上次分析执行的时间间隔间隔越大优先级越高log10(1 √分析间隔)对较大值进一步压缩增长速率Stats CacheSpecial Event特殊事件例如表新增了索引但尚未分析HasNewIndexWithoutStats: 2Table Info全表扫描最终加权公式四项指标按 60% / 10% / 30% 加权再加上特殊事件加分priority_score (0.6 * math.Log10(1 ChangeRatio) 0.1 * (1 - math.Log10(1 TableSize)) 0.3 * math.Log10(1 math.Sqrt(AnalysisInterval)) special_event[event])源码中的对应实现该公式在 calculator.go 中有精确落地。权重与特殊事件以常量形式定义const ( EventNone 0.0 EventNewIndex 2.0 ) const ( changeRatioWeight 0.6 sizeWeight 0.1 analysisInterval 0.3 )PriorityCalculator.CalculateWeight的实现与设计文档公式一致其中有一点值得注意它对ChangePercentage先乘以 100 放大数量级再套log10目的是让结果更有区分度同时实现中还保留了TODO: make these configurable注释说明权重未来有望做成可配置项。特殊事件逻辑GetSpecialEvent判定job.HasNewlyAddedIndex()时返回EventNewIndex2.0。旧实现中该队列每 3 秒重建一次count与modify_count从 stats cache 获取——这正是开销的根源也是本次改造的切入点。详细设计让队列随事件增量演进设计目标避免每次都扫描 schema cache。让 DDL 变更及时反映到优先级队列。保证 DDL 事件能被可靠、有序地接收。两类驱动信息从指标表可以看出队列所需信息分为两类DML 引发的表数据变化由于默认就会在 stats cache 中加载所有表的modify_count和count因此当表数据变化时只需重新计算分数即可。DDL 引发的表结构变化需要一种可靠途径获取已发生的 DDL 事件并据此重算分数。增量更新DML 事件当前 stats 模块在 stats cache 中维护所有表最基本的元信息modify_count和count优先级队列可以沿着增量更新的思路来维护。具体流程如下启动一个后台 worker每隔一个 stats lease默认 3 秒调用statsHandle.Update(do.InfoSchema())。在 stats cache 结构体中维护一个原子时间戳maxTblStatsVer见 statscacheinner.go。每次调用statsHandle.Update时以当前maxTblStatsVer为起点增量查询新更新SELECT version, table_id, modify_count, count from mysql.stats_meta where version %? order by version拿到所有发生更新的表后更新对应统计缓存同时推进maxTblStatsVer。这样就不必扫描整个mysql.stats_meta内存中始终保留最新元信息。优先级队列复用同一套机制每当发现 stats meta 表中有更新就知道对应表的modify_count/count可能变了需要重算分数。双时间戳的进度追踪设计文档强调了一个容易踩坑的问题如果依赖 stats cache 的更新结果同步计算最新分数可能会阻塞 stats cache 的正常更新。因此优先级队列自身维护另一个时间戳nextStatsVersion只扫描nextStatsVersion与maxTblStatsVer之间发生变化的表决定是否需要重建这些表的分析任务。图中t2即maxTblStatsVert1即nextStatsVersion。必须保证t1 t2否则优先级队列无法从 stats cache 拿到最新的modify_count和count。在当前的 queue.go 实现中这一机制演化为lastDMLUpdateFetchTimestamp字段并设置了多个刷新周期常量const ( lastAnalysisDurationRefreshInterval time.Minute * 10 dmlChangesFetchInterval time.Minute * 2 mustRetryJobRequeueInterval time.Minute * 5 )即DML 变更每 2 分钟拉取一次分析间隔等时间类指标每 10 分钟重算一次失败任务重入队间隔 5 分钟。队列注释中说明拉取 DML 变更时会先加锁再更新取数时间戳宁可让部分变更被处理两次幂等可接受也不允许在取数过程中漏掉变更。同时代码中特别标注对 100 万张表全量扫描 stats 并处理 DML 变更耗时不到 100ms——这直接验证了内存化 增量更新路线的性能收益。增量更新DDL 事件DML 可以靠 stats meta 增量查询但 DDL 事件当时只有 DDL 模块到 stats 模块之间的内存 channel一旦节点重启或切换事件就会丢失因此需要一个可靠且有序的存储与拉取机制。核心设计决策DDL 模块在完成 DDL 的同一事务内把事件写入一张带预置主键的物理表stats 模块按主键排序读取这些事件——因为 DDL 事件按顺序执行其 ID 也自然有序stats 模块用一个大事务批量处理事件处理完即删除该方案无需在故障切换时维护特定 checkpoint且天然保证exactly-once 投递每个事件只被读取和处理一次。新系统表 mysql.ddl_notifier承载 DDL 事件的系统表定义如下CREATE TABLE mysql.ddl_notifier ( ddl_job_id BIGINT, multi_schema_change_id BIGINT COMMENT -1 if the schema change does not belong to a multi-schema change DDL. 0 or positive numbers representing the sub-job index of a multi-schema change DDL, schema_change JSON COMMENT SchemaChange at rest, processed_by_flag BIGINT UNSIGNED DEFAULT 0 COMMENT flag to mark which subscriber has processed the event, PRIMARY KEY(ddl_job_id, multi_schema_change_id) )各字段含义ddl_job_idDDL job 的 ID与multi_schema_change_id共同构成主键multi_schema_change_id-1表示该 schema change 不属于 multi-schema change DDL0或正数表示 multi-schema change DDL 中的子任务索引schema_change序列化后的 SchemaChange 事件JSONprocessed_by_flag无符号大整数按位标记哪些订阅者已处理过该事件。在仓库落地版本中表结构有所演进见 store.go 的OpenTableStore注释multi_schema_change_id更名为sub_job_id语义扩展为不属于 multi-schema change DDL 或 merged DDL 时为 -1否则为子任务索引。Store接口的四个方法Insert/UpdateProcessed/DeleteAndCommit/List分别对应写入、标记已处理、删除已消费事件和按序读取事件。订阅者的 Go 类型视图订阅方看到的是统一的SchemaChange结构通过Type字段区分事件类型再用对应的 getter 获取具体信息// SchemaChange stands for a schema change event. DDL will // generate one SchemaChange or multiple SchemaChange (only // for multi-schema change DDL). The caller should check the // Type field of SchemaChange and call the corresponding getter // function to retrieve the needed information. type SchemaChange struct { Type model.ActionType // unexported fields ... } // GetAddPartitioningInfo retrieves information from a // ActionAddTablePartition type SchemaChange. func (c *SchemaChange) GetAddPartitioningInfo() (...)在仓库中对应 events.go 的SchemaChangeEvent。它已支持非常丰富的事件类型CreateTable、TruncateTable、DropTable、AddColumn、ModifyColumn、AddPartition、TruncatePartition、DropPartition、ExchangePartition、ReorganizePartition、AddPartitioning、RemovePartitioning、AddIndex、FlashbackCluster、DropSchema 等每种类型都有配套的NewXxxEvent构造函数与GetXxxInfo取值函数例如NewAddIndexEventGetAddIndexInfo返回表信息、新增索引列表及是否已分析标记。发布者的 Go 类型视图从发布者目前只有 DDL 模块视角内部细节被隐藏起来。SchemaChange持有一个仅用于 JSON 序列化的内部结构体type SchemaChange struct { Type model.ActionType inner *schemaChange4Persist } // schemaChange4Persist is used by SchemaChange when needed to // (un)marshal data, because the Golang JSON library needs every // fields to be exported but we want to hide the details for // subscribers so SchemaChange has unexported fields. type schemaChange4Persist struct { Type model.ActionType TableInfo *model.TableInfo json:omitempty PartInfo *model.PartitionInfo json:omitempty ... } // MarshalJSON implements json.Marshaler. func (c *SchemaChange) MarshalJSON() ([]byte, error) { return json.Marshal(c.inner) } // UnmarshalJSON implements json.Unmarshaler. func (c *SchemaChange) UnmarshalJSON(data []byte) error { p : schemaChange4Persist{} err : json.Unmarshal(data, p) if err ! nil {...} c.Type p.Type c.inner p return nil }仓库实现里这个隐藏内部细节的结构体叫jsonSchemaChangeEvent见 events.go包含table_info、old_table_info、added_partition_info、dropped_partition_info、columns、indexes、analyzed、old_table_id_for_partition、type等 JSON 字段。核心原则是只有 SchemaChange 结构体自己知道实现细节对订阅者不暴露任何内部信息。订阅 API 行为SchemaChangeHandler是订阅者注册的回调其契约相当严格// SchemaChangeHandler function is used by subscribers to // handle the SchemaChange generated by the DDL module. It // will be called at least once for every SchemaChange. The // sctx has already started a pessimistic transaction and // handler should execute exactly once SQL modification // logic with it. After the function is returned, subscribing // framework will commit the whole transaction with internal // flag modification to provide exactly-once delivery. The // handler will be called periodically, with no guarantee about // the latency between the execution time and SchemaChange // happening time. // // The handler function must be registered by RegisterHandler // before the DDLNotifier is started. If the handler cant // immediately serve the handling after regsitering, it // can return nil to tell the DDLNotifier to act like the // change has been handled, or return ErrNotReadyRetryLater // to hold the change and re-handle later. type SchemaChangeHandler func(ctx context.Context, sctx sessionctx.Context, change SchemaChange) error // ErrNotReadyRetryLater should be used by a registered handler // that is not ready to process the events. var ErrNotReadyRetryLater errors.New(...) // RegisterHandler must be called with an exclusive and fixed ID // for each handler to register the handler. Illegal ID will // panic. RegisterHandler should not be called after the global // DDLNotifier is started. func RegisterHandler(id int, handler SchemaChangeHandler) {...}关键约束handler 在悲观事务内执行且只应执行一次 SQL 修改逻辑返回后由订阅框架连同内部标记修改一起提交实现exactly-oncehandler 被周期性调用延迟无保证必须在DDLNotifier启动前通过RegisterHandler注册若 handler 暂时无法处理可返回nil视作已处理或返回ErrNotReadyRetryLater保留事件稍后重试。仓库中HandlerID是持久化的整数 ID每个 ID 在BIGINT列中占一个 bit最多只能有 64 个。目前定义了三个见 subscribe.goconst ( // TestHandlerID is used for testing only. TestHandlerID HandlerID 0 // StatsMetaHandlerID is used to update statistics system table. StatsMetaHandlerID HandlerID 1 // PriorityQueueHandlerID is used to update the priority queue. PriorityQueueHandlerID HandlerID 2 )其中PriorityQueueHandlerID 2正是本设计的主角——优先级队列作为 DDL 事件的订阅者。RegisterHandler会对非法 ID小于 0 或大于等于 64直接 panic。订阅的实现DDLNotifier为避免每个订阅者各自拉取事件设计引入一个后台 worker 统一处理mysql.ddl_notifier构造一个DDLNotifier让其他模块向它注册事件 handler。type DDLNotifier struct { sctx sessionctx.Context handlers map[int]SchemaChangeHandler pollInterval time.Duration } func NewDDLNotifier(sctx sessionctx.Context, pollInterval time.Duration) *DDLNotifier { return DDLNotifier{ sctx: sctx, handlers: make(map[int]SchemaChangeHandler), pollInterval: pollInterval, } } func (n *DDLNotifier) Start(ctx context.Context) error { ticker : time.NewTicker(n.pollInterval) defer ticker.Stop() for { select { case -ctx.Done(): return ctx.Err() case -ticker.C: if err : n.processEvents(ctx); err ! nil { log.Printf(Error processing events: %v, err) } } } }处理流程processEvents按事件逐个执行所有已注册 handler并用processed_by_flag位图记录进度func (n *DDLNotifier) processEventForHandler(ctx context.Context, event DDLEvent, handlerID int, handler SchemaChangeHandler) (err error) { if n.hasProcessed(event, handlerID) { return nil } if _, _, err ExecRows(n.sctx, BEGIN PESSIMISTIC); err ! nil { return err } defer func() { err finishTransaction(n.sctx, err) }() if err : handler(ctx, n.sctx, event.Event); err ! nil { return err } if err : n.markProcessed(n.sctx, event.JobID, handlerID); err ! nil { return err } event.ProcessedBy | 1 uint(handlerID) return nil }辅助判定函数func (n *DDLNotifier) hasProcessed(event DDLEvent, handlerID int) bool { return (event.ProcessedBy (1 handlerID)) ! 0 } func (n *DDLNotifier) allHandlersProcessed(event DDLEvent) bool { return event.ProcessedBy (1len(n.handlers))-1 }设计要点每个事件分别以独立事务处理每个 handler减少各 handler 之间的相互影响只有当所有 handler 都处理完某个事件后才删除该事件事件按 job ID 排序拉取保证顺序消费。仓库实现进一步细化了这些逻辑见 subscribe.goprocessEvents每轮从 session pool 取两个 session一个用于List读事件一个用于processEventForHandler处理批量大小由ProcessEventsBatchSize 1024控制采用键集分页WHERE (ddl_job_id, sub_job_id) (?, ?) ORDER BY ddl_job_id, sub_job_id LIMIT ?按序流式读取为保证同一 handler 收到的事件严格有序若某 handler 在前序事件上报错后续事件直接跳过该 handlerskipHandlers机制processEventForHandler中BeginPessimistic开启悲观事务handler 执行 SQL 修改逻辑store.UpdateProcessed通过乐观的条件更新WHERE ... AND processed_by_flag old更新位图并随事务一起提交实现 exactly-oncehandler 耗时超过 5 秒会记录慢日志UpdateProcessed如果影响行数为 0会报错提示可能已被其他 owner 更新见 store.go从实现层面印证了设计文档中关于竞态防护的考虑。运行位置与未来扩展由于目前只有 stats 模块订阅且需要内存操作DDLNotifier只在stats owner 节点上启动。仓库实现与此完全一致subscribe.go 中DDLNotifier实现了owner.Listener接口var _ owner.Listener (*DDLNotifier)(nil)通过OnBecomeOwner/OnRetireOwner回调随 stats owner 的选举与退位而启停并明确注释了三点理由优先级队列在内存中处理事件、stats handler 与 DDLNotifier 同节点可保证数据完整性、避免跨节点分布式处理引发竞态或不一致。未来可能性若将来其他模块也要订阅需要避免只在 stats owner 节点运行其他模块的内存操作可能不在该节点上。一个可能方案是在每个 TiDB 节点上各启动一个 DDLNotifier、各节点承载不同订阅者。但这会引入订阅者管理、跨节点删除时机、以及相同订阅逻辑在每个节点运行算一个还是多个订阅者等粒度问题。设计文档明确将其列为 future possibility而非本次目标。时间因素的处理分数中有个指标是分析间隔距上次分析的时间它天然随时间变化因此需要周期性重算。设计决定优先级队列的后台 worker每 5 分钟为相关表重算一次分析间隔与分数。在仓库落地中这个周期被细化为lastAnalysisDurationRefreshInterval time.Minute * 1010 分钟并且重算所需的数据来源清晰可见——interval.go 中通过查询mysql.analyze_jobs表来计算指标平均分析时长取最近 5 次成功分析state finished AND fail_reason IS NULL的AVG(TIMESTAMPDIFF(SECOND, start_time, end_time))分区表则跨分区取样上次失败时长取最近一次失败分析state failed距当前时间戳的秒数。故障切换Failover故障切换只有两类场景需要处理节点重启owner 切换。两种情况都需要重建优先级队列重新扫描并重构整个队列DML 事件扫描整个 stats cache取最新的modify_count与countDDL 事件由于所有操作都在事务中完成等队列就绪后让 notifier 重新开始处理即可。设计文档给出了重建队列的实测成本从 TiKV 收集全部表信息Table CountTime (s)690676.365261488974759.0119024717350516.08587708427133824.15543368933269729.865601857可以看到从 TiKV 收集全部表信息相当慢100 万张表需要超过一分钟。原因是 Information Schema v2 之后SchemaTableInfosAPI 会直接访问 TiKV 并解析表信息。不过目前这个代价可以容忍因为优先级队列重建的频率很低仅在故障切换时发生与原来每 3 秒重建一次相比已是数量级的改善。新的优先级队列内存堆的 API 设计由于要不断增删改堆需要一个线程安全的堆来存放分析任务。设计要求的操作复杂度如下OperationTime complexityAddO(log n)UpdateO(log n)DeleteO(log n)PopO(1)GetO(1)存储的信息type Indicators struct { // ChangePercentage is the percentage of the changed rows. // Usually, the more the changed rows, the higher the priority. // It is calculated by modifiedCount / last time analysis count. ChangePercentage float64 // TableSize is the table size in rows * len(columns). TableSize float64 // LastAnalysisDuration is the duration from the last analysis to now. LastAnalysisDuration time.Duration } type AnalysisJob interface { // SetWeight sets the weight of the job. SetWeight(weight float64) // GetWeight gets the weight of the job. GetWeight() float64 // HasNewlyAddedIndex checks whether the job has a newly added index. HasNewlyAddedIndex() bool // GetIndicators gets the indicators of the job. GetIndicators() Indicators }泛型堆接口// K is the key type of the object. It has to be comparable. // T is the type of the object. type Heap[K comparable, T any] interface { // Add/Update elements Add(obj T) error BulkAdd(list []T) error AddIfNotPresent(obj T) error Update(obj T) error // Same as Add // Remove element Delete(obj T) error DeleteByKey(key K) error // View/Pop top element Peek() (T, error) Pop() (T, error) // Query operations List() []T ListKeys() []K Get(obj T) (T, bool, error) GetByKey(key K) (T, bool, error) // State operations Close() IsClosed() bool } // Constructor func NewHeapK comparable, T any *Heap[K, T] // Helper function type definitions type KeyFunc[K comparable, T any] func(T) (K, error) type LessFunc[T any] func(T, T) bool该接口用 Go 泛型实现K必须可比较用于去重与查询T是任意任务类型通过KeyFunc与LessFunc解耦取键与比较优先级逻辑。仓库落地版本见 heap.go 与 queue.go。pqHeap接口保留了设计中的核心操作getByKey、addOrUpdate、update、delete、list、pop、peek、isEmpty、len并对分析任务做了具体化非分区表、静态分区表、动态分区表各有独立的 analysis job 实现non_partitioned_table_analysis_job.go、static_partitioned_table_analysis_job.go、dynamic_partitioned_table_analysis_job.go并支持TableStatsVer等字段记录统计版本为 DML 增量更新提供依据。测试设计该特性需要同时关注正确性与性能正确性测试验证优先级计算的准确性以及 DML/DDL 变更处理的正确性性能测试确保优先级队列的运行不会对系统性能产生负面影响。功能测试优先级队列操作测试 Add、Update、Delete、Pop、Get 等基本操作验证线程安全DML 变更处理验证modify_count和count变化时正确重算分数测试来自 stats cache 的增量更新DDL 变更处理测试 DDL 事件订阅验证不同类型 DDL 变更如 add index、exchange partition被正确处理测试 DDL 事件的 exactly-once 投递基于时间的更新验证分析间隔与分数的周期性重算故障切换场景测试节点重启、owner 切换后的队列重建。场景测试混合 DML/DDL 变更模拟真实场景验证优先级计算与队列更新的正确性大规模测试100 万 表规模下的性能与正确性验证并发操作并发 DML/DDL 操作下的队列一致性与优先级更新。兼容性测试本次改动只影响 stats 模块不涉及兼容性问题。基准测试内存占用监控内存堆的内存消耗测试不同表数量、更新频率、DDL/DML 频率下的内存使用模式性能测量处理大量 DDL 事件的性能测量故障切换时重建队列的性能。仓库中对应测试非常完备heap_test.go、queue_test.go、calculator_test.go、queue_ddl_handler_test.go、events_test.go、store_test.go 等覆盖了上述各测试要点。尚未解决的问题Unresolved Questions设计文档保留了一个开放问题优先级队列如何查找发生更新的表 ID当前设计会查询mysql.stats_meta表且 version 列有索引理论上不会太慢但也可以考虑直接从 stats cache 取数——cache 内存中维护了所有表的count、modify_count和 version可遍历 cache 做版本比较筛选。但后者可能需要取大量 segment 锁、做 n 次版本比较。两种方案孰优孰劣尚不确定需要更多基准测试来定论。总结把优先级队列常驻内存并增量维护是 TiDB 自动统计信息链路的一次重要架构演进DML 侧复用 stats meta 的增量更新机制DDL 侧新建mysql.ddl_notifier系统表与DDLNotifier订阅框架以同事务写入 按主键有序读取 位图标记 事务删除的组合拳实现了可靠有序且 exactly-once 的事件投递队列本身则收敛为线程安全的内存堆。这套设计已在仓库中完整落地相关实现可继续在 pkg/statistics/handle/autoanalyze/priorityqueue、pkg/ddl/notifier 与 pkg/statistics/handle/cache/statscacheinner.go 中深入研读。【免费下载链接】tidbTiDB is built for agentic workloads that grow unpredictably, with ACID guarantees and native support for transactions, analytics, and vector search. No data silos. No noisy neighbors. No infrastructure ceiling.项目地址: https://gitcode.com/GitHub_Trending/ti/tidb创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考