Milvus QueryNode v2 重构设计解读：Delegator/Worker 角色分离与无 delta channel 的删除转发架构-北京尧图网络

Milvus QueryNode v2 重构设计解读Delegator/Worker 角色分离与无 delta channel 的删除转发架构【免费下载链接】milvusMilvus is a high-performance, cloud-native vector database built for scalable vector ANN search项目地址: https://gitcode.com/GitHub_Trending/mi/milvus本文基于 20230418-querynode_v2.mdMEP已合入并随 Milvus v2.3.0 发布展开。QueryNode 承载了 Milvus 中最复杂的职责——它既是搜索/查询的执行引擎又是 shard 数据消费与分布的管理者。本文围绕Delegator 与 Worker 角色分离移除 delta channel删除转发策略与 PK Oracledelete buffer 数据完整性保障四大重构主线逐一还原设计动机、接口契约与边界条件并对照当前仓库中 internal/querynodev2 的真实实现代码帮助读者理解这套架构为什么长成这样以及如何在源码中逐行验证设计落地。一、重构背景与目标在 QueryNode v1重构前版本中一个 QueryNode 进程同时承担了三种耦合在一起的职责shard 数据消费从 DML channel 中消费 insert/delete 消息segment 分布管理维护本 shard 上 segment 的加载/释放状态纯计算执行在 segment 上执行 search/query 请求。这种耦合带来的直接后果是无法将数据/分布管理逻辑与纯计算逻辑解耦导致代码可读性差、职责难以拆分、未来难以将两类角色部署到不同组件中。该 MEP 提出重构 QueryNode期望达到四个目标见原文档## Summary分离 Delegator 与 Worker两种角色移除 delta channel改变删除记录的转发方式在 distribution 中维护 growing segments让分布式元数据视图更完整提升代码可读性。这四个目标并非孤立的只有先把数据消费分布管理收拢到 Delegator才能让 Delegator 天然拥有全部 DML 数据含删除从而顺理成章地废弃 delta channel而删除转发的正确性又依赖于 PK Oracle 与 delete buffer 等新机制的设计。整篇 MEP 实质上是在回答当一个 QueryNode 变成唯一的数据消费者后如何在不丢失、不重排删除数据的前提下把删除转发到真正需要的节点二、Delegator 与 Worker角色的分离与接口定义2.1 角色职责划分Delegator即 v1 中的 ShardLeader负责处理 segment 分布并消费 dml channel 的数据。所有分布变更load/release都必须由 Delegator 转发这样它才能始终持有该 shard 最新可用的 segment 分布信息。它是管理者数据入口。Worker纯粹的算力工人只负责在本节点上加载的 segment 上提供 search/query 服务。它是执行者。一个 QueryNode 在当下可以同时扮演 Delegator 和 Worker例如本节点 shard 的 Leader 委托其他节点加载 segment但自身也可能保存部分 segment。正因为重构时把二者拆成了两个子包未来如需将它们重排到不同组件中代价极低。2.2 接口定义原文档核心代码Delegator 接口原文档## Interface Definition// ShardDelegator is the interface definition. type ShardDelegator interface { // Search Query APIs Search(ctx context.Context, req *querypb.SearchRequest) ([]*internalpb.SearchResults, error) Query(ctx context.Context, req *querypb.QueryRequest) ([]*internalpb.RetrieveResults, error) GetStatistics(ctx context.Context, req *querypb.GetStatisticsRequest) ([]*internalpb.GetStatisticsResponse, error) // Distribution dml related APIs ProcessInsert(insertRecords map[int64]*InsertData) ProcessDelete(deleteData []*DeleteData, ts uint64) LoadGrowing(ctx context.Context, infos []*querypb.SegmentLoadInfo, version int64) error LoadSegments(ctx context.Context, req *querypb.LoadSegmentsRequest) error ReleaseSegments(ctx context.Context, req *querypb.ReleaseSegmentsRequest, force bool) error SyncDistribution(ctx context.Context, entries ...SegmentEntry) }Worker 接口// Worker is the interface definition for querynode worker role. type Worker interface { LoadSegments(context.Context, *querypb.LoadSegmentsRequest) error ReleaseSegments(context.Context, *querypb.ReleaseSegmentsRequest) error Delete(ctx context.Context, req *querypb.DeleteRequest) error Search(ctx context.Context, req *querypb.SearchRequest) (*internalpb.SearchResults, error) Query(ctx context.Context, req *querypb.QueryRequest) (*internalpb.RetrieveResults, error) GetStatistics(ctx context.Context, req *querypb.GetStatisticsRequest) (*internalpb.GetStatisticsResponse, error) IsHealthy() bool Stop() }接口设计反映了两个关键差异Delegator 的Search/Query返回slice因为它需要把请求扇出fan-out给多个 Worker 并聚合Worker 的Search/Query只返回单个结果因为它只负责自己本地的 segment。Delegator 拥有ProcessInsert/ProcessDelete/LoadGrowing/SyncDistribution这些数据面方法Worker 接口则完全没有——Worker 只能被动接受LoadSegments/ReleaseSegments/Delete指令。2.3 在源码中的落地验证当前仓库中上述设计已演化为成体系的代码internal/querynodev2/delegator/delegator.go 中定义了ShardDelegator接口与shardDelegator结构体。从 delegator.go#L144-L199 可以看到它的核心字段collectionID、replicaID、vchannelName、distribution *distribution、deleteBuffer deletebuffer.DeleteBuffer[*deletebuffer.Item]、loader segments.Loader、segmentManager等与 MEP 中Delegator 维护 distribution 消费数据持有 delete buffer的定位一一对应。Delegator 相关的能力被拆散在多个文件中delegator_data.go数据消费、distribution.go分布、delta_forward.go删除转发、snapshot.go服务快照、idf_oracle.go词频/IDF 计算、pk_filter.go主键过滤。Worker 侧对应 internal/querynodev2/local_worker.go 与 internal/querynodev2/cluster 目录——cluster 包负责 Delegator 与远程/本地 Worker 之间的管理调用cluster.Manager也被注入shardDelegator见上文 struct 中的workerManager cluster.Manager。从源码结构看MEP 设想的将两者拆为独立子包未来可分别部署的扩展性已落地delegator、segments、pipeline、pkoracle都已是 querynodev2 下的平级独立包互相只通过接口引用。三、移除 delta channel让数据消费与删除转发回到同一条链路3.1 为什么 delta channel 是负担Milvus 2.0.x 引入 Delete 能力后删除记录只出现在对应的 DML channel 上。对于那些不消费该 channel 却保存了该 shard segment 副本的 QueryNode就必须额外有一个通道把删除转发过去——这就是 delta channel。MEP 明确指出 delta channel 的三大问题消息队列 topic 数量翻倍相比早期版本系统需要双倍的 MQ topic功能耦合querynode 的 search/query 功能被与删除记录转发者绑定在一起可用性风险当时转发职责由 datanode 承担——一旦某些 datanode 宕机一段时间可能导致 search/query 不可用。由于 Delegator 作为 shard 唯一数据消费者可以消费到 MQ 中的全部 DML 数据包括 delete它天然就是删除转发的最佳宿主。于是重构的核心设计问题收敛为两个如何判定某条删除应转发给哪个 segment / 哪个 querynode如何保证所有 segment 都能拿到完整的删除数据视图第一个问题由PK Oracle回答第二个问题由delete buffer 失败重消费机制回答。3.2 Primary Key OraclePK OracleMEP 需要一个组件来判定或估算给定主键可能落在哪些 segment 上命名为PKOracle。候选实现路径有三种Delegator 持有全部 PK 列数据——内存开销极大不可行Delegator 持有全部 statslogBloom filter文件——因为此前删除过滤本来就是基于 Bloom filter 实现的成为首选引入第三方组件存储 pk 值 → segment id 映射——额外组件与一致性成本高。设计文档选择了方案 2并给出 PK Oracle 的原理示意图原文档使用img src../assets/graphs/pk_oracle.png width600/引用仓库内实际文件位于 docs/design-docs/assets/graphs/pk_oracle.png当前仓库落地情况internal/querynodev2/pkoracle包完整实现了这一机制pk_oracle.go 定义了PkOracle接口其方法比设计稿更进一步Get(pk storage.PrimaryKey, filters ...CandidateFilter) ([]int64, error)pk_oracle.go#L56返回主键可能命中的 segment id 集合、BatchGet(pks []storage.PrimaryKey, filters ...CandidateFilter) map[int64][]bool批量查询返回segment 主键过滤器矩阵、Register(candidate Candidate, workerID int64)pk_oracle.go#L99segment 加载后注册其 Bloom filter 候选、Remove(filters ...CandidateFilter)与RemoveAndRefundAll()segment 释放时注销候选bloom_filter_set.go 是Candidate的核心实现用BloomFilterSet封装了 segment 的 bloom filter 集合含 bf、part 级别的多个 filtercandidate.go、key.go、external_segment_candidate.go 提供候选的 keysegment/partition/channel 维度与对外部 segment 的支持说明 PK Oracle 机制已从内部删除转发扩展到了外部数据等更多场景。换句话说谁可能持有这条主键的问题在实现中就是pko.Get(pk)遍历已注册候选的 bloom filter 集合命中者为转发目标。3.3 删除转发的 gRPC 定义Delegator 通过 gRPC 把删除发给目标 Worker接口与消息定义如下原文档Delete grpc defDelete(context.Context, *querypb.DeleteRequest) (*commonpb.Status, error)message DeleteRequest { common.MsgBase base 1; int64 collection_id 2; int64 partition_id 3; string vchannel_name 4; int64 segment_id 5; schema.IDs primary_keys 6; repeated uint64 timestamps 7; }注意该消息携带primary_keys与timestamps两个并行数组primary_keys是本次删除的主键集合timestamps是每个主键对应的删除时间戳按消息消费顺序排列Worker 侧按segment_id vchannel_name定位本地 segment 后将 (pk, ts) 逐条写入 segment 的删除记录。批量打包主键、逐条带时间戳是保证删除必须严格按时间戳序生效这一语义能跨节点还原的前提。3.4 删除转发策略的选择为何必须即时转发Delegator 必须在任何 search/query 执行之前完成删除转发而转发策略有三种可能策略描述结论策略 1即时转发eager删除一到就转发转发失败则阻塞消费流程最终采纳策略 2惰性转发lazy删除数据随 search/query 请求捎带转发并辅以周期性的 flush 任务被否决策略 3只转发已处理的 bitset只把删除已应用到哪些行的结果bitset转发过去需要持有全部 PK 数据被否决策略 3 依赖Delegator 拥有全部主键数据这与已选定的 Bloom filter 版 PKOracle 冲突直接排除。在策略 1 与 2 之间MEP 经过调研给出关键结论所有删除记录必须严格按时间戳顺序被应用否则 segment 内部的二分查找可能为删除返回错误的 bitset。也就是说一旦删除乱序哪怕只在时间上跨节点乱序基于有序数组二分定位的删除 bitset 计算就会出错。因此在 segment 内部实现改变之前策略 1即时转发并阻塞消费是唯一选择。当前仓库 delegator/delta_forward.go 与 delegator/buffered_forwarder.go 即承载了这类消费-转发逻辑与缓冲批量转发实现buffered_forwarder_test.go覆盖了失败重试、乱序等场景。3.5 数据完整性保障delete buffer 失败重消费由于 Milvus 2.x 是分布式系统存在多个可能破坏删除数据完整性的场景原文档Data Integrity Guarantee异步加载Load Asynchronizely集合还在加载中时Delegator 无法保证所有 segment 已就绪却可能已经开始转发删除加载新 segment集合加载后compaction 等操作可能触发加载新 segment若该 segment 的消费位置已越过 safe point即之前的删除都已同步进 delta log这段时间内的删除条目会缺失balance / 节点宕机 / 滚动升级与上一种情况类似segment 在节点间搬移或重建期间部分删除记录可能丢失。解决方案——带失败重消费的 delete bufferDelegator 维护一个 delete buffer 保存近期删除数据每当一个 segment 被加载Delegator 就尝试用 buffer 中所有需要的删除数据补齐该 segment近期意味着一个容量可配置的有限双缓冲double buffer需要的删除指该 segment checkpoint 之后产生的删除记录若 segment 的 checkpoint 已经超出 delete buffer 的可回溯范围作为最后手段Delegator 会从 checkpoint 重新消费删除数据re-consume补全。该机制在当前源码中落地于 internal/querynodev2/delegator/deletebuffer 包且实现比设计稿更丰富既有DeleteBuffer接口与基于双向链表列表的list_delete_buffer.go支持按时间戳范围裁剪典型的双缓冲语义也有按时间戳组织的skiplist_buffer.go变体shardDelegator通过deleteBuffer deletebuffer.DeleteBuffer[*deletebuffer.Item]字段持有它delegator.go#L167delegator_data.go/growing_flush_source.go负责在 growing 落盘/segment 新加载时执行补删除操作。配套 delete_buffer_test.go 验证了跨 checkpoint、buffer 溢出回退等边界行为。四、其他配套改动4.1 用 Pipeline 取代 FlowgraphMEP 顺带把 v1 时代的 flowgraph 替换为更简化的pipelinepipeline 中每个节点至多一个入度、一个出度如同流水线把一段周期性重复的工作切分成多个部分每个节点由一个 goroutine 负责一段以并行提升吞吐。在 querynode 上pipeline 被用来处理来自 MsgStream 的消息原文档Use pipeline instead of flowgraphFilterNode过滤消息中的无效部分Insert Node把 insert 消息中的行写入 segmentDelete Node把 delete 消息中的删除行写入 segment并更新 TSafe。当前仓库中internal/querynodev2/pipeline包即该设计的延续且 Delegator 侧对 DML 消息的过滤如 delegator/pk_filter.go、delegator/scalar_pruner.go、delegator/segment_pruner.go比设计稿又前进了一步在消费阶段就借助主键、标量与 segment 级统计信息剪掉不可能命中的消息/segment。4.2 Search/Query 的 TSafe 等待上移到 Delegator既然 shard 上唯一的数据消费者是 Delegator等待 TSafe 的职责就顺理成章地移动到了 Delegator原文档Search/Query tsafe。shardDelegator结构体中的tsCond *syncutil.ContextCond、latestTsafe *atomic.Uint64delegator.go#L171-L172就是这条等待链的载体查询到达后Delegator 判断目标 shard 的 TSafe 是否已推进到查询时间点未达则阻塞等待直到消费追上。五、配套 Manager 接口重构为了让 Delegator/Worker 两层都能以统一方式管理资源MEP 重新定义了各 Manager 接口原文档## Interfaces。5.1 CollectionManager 与 SegmentManagertype CollectionManager interface { // Get returns collection within a LRU cache, // it will pull the collection from QueryCoord if its not in the cache, // returns error if failed to pull Get(collectionID int64) (*Collection, error) } type SegmentManager interface { // Put puts the given segments in, // and increases the ref count of the corresponding collection, // dup segments will not increase the ref count Put(segmentType SegmentType, segments ...*Segment) Get(segmentID UniqueID) *Segment GetSealed(segmentID UniqueID) *Segment GetGrowing(segmentID UniqueID) *Segment // Remove removes the given segment, // and decreases the ref count of the corresponding collection, // will not decrease the ref count if the given segment not exists Remove(segmentID UniqueID, scope querypb.DataScope) }注意两点设计意图CollectionManager.Get带LRU 缓存未命中时向 QueryCoord 拉取的能力SegmentManager通过ref count管理 collection 生命周期且GetGrowing/GetSealed分型查询与Remove(segmentID, scope)的querypb.DataScope参数正好呼应 MEP 把 growing segment 纳入 distribution 管理的目标——即 growing 与 sealed 现在统一由 SegmentManager 按 scope 区分管理。当前实现在 internal/querynodev2/segments/manager.goSegmentManager与 segments/collection.go含基于 LRU 的 collection 管理中可查证。5.2 Loadertype Loader interface { // Load loads binlogs, and spawn segments, // NOTE: make sure the ref count of the corresponding collection will never go down to 0 during this Load(ctx context.Context, collectionID int64, segmentType SegmentType, version int64, infos ...*querypb.SegmentLoadInfo) ([]Segment, error) }Loader 只负责加载 binlog 并生成 segment注释特别强调调用方要保证 collection 引用计数在加载期间不为 0避免并发释放竞态。落地位于 segments/segment_loader.go加载流程会结合 segments/index_meta.go 完成索引元数据的装配。5.3 Segment 抽象MEP 把 segment 抽象成接口统一定义属性、索引、写入与查询能力type Segment interface { // Properties ID() int64 Collection() int64 Partition() int64 Channel() string Version() int64 StartPosition() *internalpb.MsgPosition Type() SegmentType // Index related AddIndex(fieldID int64, index *IndexedFieldInfo) GetIndex(fieldID int64) *IndexedFieldInfo HaveIndex(fieldID int64) bool // Insert related Insert(entityIDs []int64, timestamps []Timestamp, record *segcorepb.InsertRecord) error Delete(entityIDs []storage.PrimaryKey, timestamps []typeutil.Timestamp) error // Query related Search(searchReq *searchRequest) (*SearchResult, error) Retrieve(plan *RetrievePlan) (*segcorepb.RetrieveResults, error) } func NewSegment(collection *Collection, segmentID int64, partitionID int64, collectionID int64, channel string, segmentType SegmentType, version int64, startPosition *internalpb.MsgPosition) (*Segment, error) func DeleteSegment(segment *Segment)Delete(entityIDs, timestamps)是专门为时间戳有序删除设计的签名体现了第 3.4 节讨论的删除语义约束。当前仓库将其扩展为Segment接口族 SegmentInternalBase由 segcore 持有的分层结构见 segments/segment_interface.go 与 segments/segment.goSegmentType/LevelSealed、Growing、L0 等也由此演化出 segment_l0.go 等文件说明架构为后续 L0 删除段等新能力预留了扩展位。5.4 Collection 与 PipelineManagerCollection 抽象原文档## Collectiontype Collection struct { } func (c *Collection) ID() UniqueID func (c *Collection) Schema() *schemapb.CollectionSchema func (c *Collection) GetPartitions() []int64 func (c *Collection) HasPartition(partitionID int64) bool func (c *Collection) AddPartition(partitionIDs ...int64) func (c *Collection) RemovePartition(partitionID int64) func (c *Collection) GetLoadType() querypb.LoadType func NewCollection(collectionID int64, schema *schemapb.CollectionSchema, loadType querypb.LoadType) *Collection func DeleteCollection(collection *Collection)PipelineManager 抽象type PipelineManager struct { } func (m *PipelineManager) Num() int func (m *PipelineManager) Add(collectionID UniqueID, dmlChannels []string) error func (m *PipelineManager) Get(collectionID UniqueID, channel Channel) (*Pipeline, error) func (m *PipelineManager) Remove(channels []Channel) func (m *PipelineManager) Close()PipelineManager以(collectionID, channel)为键管理 pipeline——这正是 shard 维度的数据消费粒度与 Delegator 一一对应。可参考 pipeline 目录下的 manager 与 node 实现。六、测试计划重构的验收标准MEP 为这次大重构给出三层验收标准原文档## Test Plan对理解其风险控制思路很有价值单元测试querynode v2 中所有包覆盖率约80%E2E 测试既有的 load / release / search / query 全部用例必须通过集成测试重点覆盖两类故障场景——Worker delete 失败的用例验证删除转发失败时的处理路径Worker 离线的用例验证节点掉线后查询的可用性/一致性。对照当前仓库这两类集成场景在 delegator/delegator_test.go、delegator/delta_forward_test.go、delegator/distribution_test.go 与 segments/segment_loader_test.go 中都有大量回归覆盖测试的演进基本沿着 MEP 划定的删除失败不丢数据、节点离线不影响正确性方向深入。七、结语与源码研读建议QueryNode v2 重构是 Milvus 查询链路走向管理面与执行面分离的里程碑Released v2.3.0。这篇 MEP 留下的核心心智模型可以浓缩为一句话把 shard 的所有数据流insert/delete与分布信息收敛到 Delegator用 PK Oracle 决定删除该去哪用 delete buffer 兜底数据完整性把 segment 上的纯计算下沉给 Worker。后续在源码中研读时建议按如下顺序阅读正好与本文的结构一一对应delegator/delegator.go先看shardDelegator结构体字段vchannel、distribution、deleteBuffer、latestTsafe建立一个 Delegator 一个 shard 的大脑的整体认知delegator/distribution.go 与 delegator/delegator_data.go理解 growing/sealed 分布如何随消费实时变更pkoracle/pk_oracle.go验证 Bloom filter 版 PK Oracle 的注册/查询路径delegator/deletebuffer看 delete buffer 如何实现有限双缓冲 checkpoint 补数据local_worker.go 与 segments看 Worker 与 segment 执行侧如何在 Delegator 调度下完成搜索。值得留意的是设计稿中的概念在今天已有演进例如 PKOracle 已服务外部 segment、删除段演进出了 L0 类型、Delegator 还增加了 leader view 回调与 BM25/IDF 等按 shard 局部计算的支持。这些演进都发生在Delegator 集中管理 shard这一重构所奠定的架构底座之上——这正是阅读历史 MEP 与当下代码对照时最有价值的地方。本文依据仓库内设计文档 docs/design-docs/design_docs/20230418-querynode_v2.md 与其引用的源码路径撰写文中当前仓库实现均指向该仓库 internal/querynodev2 下的可验证代码。【免费下载链接】milvusMilvus is a high-performance, cloud-native vector database built for scalable vector ANN search项目地址: https://gitcode.com/GitHub_Trending/mi/milvus创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

Milvus QueryNode v2 重构设计解读：Delegator/Worker 角色分离与无 delta channel 的删除转发架构

相关推荐

使用 claude-task-master 的 Hamster Brief 端到端任务执行工作流

LocalAI 模型别名（Model Aliases）完全指南：零客户端改动的模型重定向与灰度切换

两级式SST技术：高效电力转换与工业应用解析

更多资讯

美业门店效率革命：数据驱动与流程优化实战

LFFD+DSFD双检测器人脸系统：毕设级PyTorch全流程实现

Pydantic validate_call 验证装饰器实战指南：基于类型注解的函数入参与返回值校验

分子模拟异构算力适配开发教程（5）：gpu_utils 源码解析——DeviceContext/DeviceStream/DeviceBuffer 如何把 CUDA 藏进三个类

ESLint 自定义处理器（Custom Processors）完全指南：从零为 Markdown/HTML 文件编写 preprocess 与 postprocess

Temu 怎么开店？2026 年 Temu 入驻全流程详解

今日更新

Linux 内核 ARM 虚拟内存布局全解析：从 4GB 地址空间划分到源码级验证

AIGC降重工具测评与教育应用指南

数据容灾核心指标与实战方案解析

本周精选

超人会飞不算本事：系统稳定依赖清晰规则与边界设计

超人VS蜘蛛侠：拆解超级IP的影响力与传播方法论

基于CNN的调制信号识别：MATLAB实现时频图分类实战

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？