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.md（MEP，已合入并随 Milvus v2.3.0 发布）展开。QueryNode 承载了 Milvus 中最复杂的职责——它既是搜索/查询的执行引擎，又是 shard 数据消费与分布的管理者。本文围绕"Delegator 与 Worker 角色分离""移除 delta channel""删除转发策略与 PK Oracle""delete 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 Oracle（PK Oracle）

MEP 需要一个组件来"判定或估算给定主键可能落在哪些 segment 上"，命名为PKOracle。候选实现路径有三种：

Delegator 持有全部 PK 列数据——内存开销极大，不可行；
Delegator 持有全部 statslog（Bloom filter）文件——因为此前删除过滤本来就是基于 Bloom filter 实现的，成为首选；
引入第三方组件存储 pk 值 → segment id 映射——额外组件与一致性成本高。

设计文档选择了方案 2，并给出 PK Oracle 的原理示意图（原文档使用<img src="../assets/graphs/pk_oracle.png" width="600"/>引用，仓库内实际文件位于 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#L99，segment 加载后注册其 Bloom filter 候选）、Remove(filters ...CandidateFilter)与RemoveAndRefundAll()（segment 释放时注销候选）；
bloom_filter_set.go 是Candidate的核心实现，用BloomFilterSet封装了 segment 的 bloom filter 集合（含 bf、part 级别的多个 filter）；
candidate.go、key.go、external_segment_candidate.go 提供候选的 key（segment/partition/channel 维度）与对外部 segment 的支持，说明 PK Oracle 机制已从"内部删除转发"扩展到了外部数据等更多场景。

换句话说，"谁可能持有这条主键"的问题，在实现中就是pko.Get(pk)遍历已注册候选的 bloom filter 集合，命中者为转发目标。

3.3 删除转发的 gRPC 定义

Delegator 通过 gRPC 把删除发给目标 Worker，接口与消息定义如下（原文档Delete grpc def）：

Delete(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 buffer：

Delegator 维护一个 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#L167），delegator_data.go/growing_flush_source.go负责在 growing 落盘/segment 新加载时执行"补删除"操作。配套 delete_buffer_test.go 验证了跨 checkpoint、buffer 溢出回退等边界行为。

四、其他配套改动

4.1 用 Pipeline 取代 Flowgraph

MEP 顺带把 v1 时代的 flowgraph 替换为更简化的pipeline：pipeline 中每个节点至多一个入度、一个出度，如同流水线把一段周期性重复的工作切分成多个部分，每个节点由一个 goroutine 负责一段，以并行提升吞吐。

在 querynode 上，pipeline 被用来处理来自 MsgStream 的消息（原文档Use pipeline instead of flowgraph）：

FilterNode：过滤消息中的无效部分；
Insert Node：把 insert 消息中的行写入 segment；
Delete 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.Uint64（delegator.go#L171-L172）就是这条等待链的载体：查询到达后，Delegator 判断目标 shard 的 TSafe 是否已推进到查询时间点，未达则阻塞等待直到消费追上。

五、配套 Manager 接口重构

为了让 Delegator/Worker 两层都能以统一方式管理资源，MEP 重新定义了各 Manager 接口（原文档## Interfaces）。

5.1 CollectionManager 与 SegmentManager

type CollectionManager interface { // Get returns collection within a LRU cache, // it will pull the collection from QueryCoord if it's 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.go（SegmentManager）与 segments/collection.go（含基于 LRU 的 collection 管理）中可查证。

5.2 Loader

type 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.go；SegmentType/Level（Sealed、Growing、L0 等）也由此演化出 segment_l0.go 等文件，说明架构为后续 L0 删除段等新能力预留了扩展位。

5.4 Collection 与 PipelineManager

Collection 抽象（原文档## Collection）：

type 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），仅供参考