flowchart TB
    subgraph Proxy["Proxy"]
        P[Proxy Service]
    end
    
    subgraph QueryNode["QueryNode"]
        Server[QueryNode Server<br/>gRPC服务入口]
        
        subgraph DelegatorMgr["ShardDelegator管理"]
            D1[ShardDelegator-1<br/>vChannel_1]
            D2[ShardDelegator-2<br/>vChannel_2]
            DN[ShardDelegator-N<br/>vChannel_N]
        end
        
        subgraph Scheduler["任务调度器"]
            Sched[Scheduler<br/>任务队列与调度]
            TaskQueue[Task Queue]
        end
        
        subgraph Pipeline["Pipeline管理器"]
            PM[Pipeline Manager]
            P1[Pipeline-1<br/>FilterNode→InsertNode→DeleteNode]
            P2[Pipeline-2<br/>FilterNode→InsertNode→DeleteNode]
        end
        
        subgraph SegmentMgr["Segment管理"]
            SM[Segment Manager<br/>Segment生命周期管理]
            Loader[Segment Loader<br/>从Object Storage加载]
            Cache[Segment Cache<br/>LRU缓存策略]
        end
        
        subgraph Worker["Worker"]
            LW[Local Worker<br/>本地执行]
            CM[Cluster Manager<br/>Worker管理]
        end
        
        Server -->|路由到Channel| D1
        Server -->|路由到Channel| D2
        Server -->|路由到Channel| DN
        
        D1 -->|提交任务| Sched
        D2 -->|提交任务| Sched
        DN -->|提交任务| Sched
        
        Sched --> TaskQueue
        TaskQueue -->|分发| LW
        TaskQueue -->|分发| CM
        
        D1 -->|消费DML| P1
        D2 -->|消费DML| P2
        
        P1 -->|Insert/Delete| D1
        P2 -->|Insert/Delete| D2
        
        D1 -->|LoadSegment| Loader
        D2 -->|LoadSegment| Loader
        
        Loader -->|管理| SM
        Loader -->|缓存| Cache
        
        LW -->|查询| SM
        CM -->|查询| SM
    end
    
    subgraph External["外部组件"]
        OS[Object Storage<br/>MinIO/S3]
        MQ[MessageQueue<br/>Pulsar/Kafka]
        QC[QueryCoord<br/>协调器]
        RemoteQN[Remote QueryNode<br/>其他QueryNode]
    end
    
    P -.->|Search/Query| Server
    Loader -.->|Download Binlog/Index| OS
    P1 -.->|Subscribe| MQ
    P2 -.->|Subscribe| MQ
    Server -.->|HeartBeat/Report| QC
    QC -.->|LoadSegment/WatchDmChannel| Server
    CM -.->|Remote Call| RemoteQN
    
    style QueryNode fill:#e1f5ff,stroke:#333,stroke-width:2px
    style DelegatorMgr fill:#fff4e6,stroke:#333,stroke-width:1px
    style Scheduler fill:#e6ffe6,stroke:#333,stroke-width:1px
    style Pipeline fill:#ffe6e6,stroke:#333,stroke-width:1px
    style SegmentMgr fill:#f0e6ff,stroke:#333,stroke-width:1px

flowchart LR
    Growing[Growing Segment<br/>内存，可写] -->|Flush| Sealed[Sealed Segment<br/>Object Storage，只读]
    Sealed -->|Build Index| Indexed[Indexed Segment<br/>带索引]
    
    style Growing fill:#ffe6e6
    style Sealed fill:#fff4e6
    style Indexed fill:#e6ffe6

// internal/querynodev2/services.go
service QueryNode {
    // 数据加载
    rpc WatchDmChannels(WatchDmChannelsRequest) returns (Status);
    rpc UnsubDmChannel(UnsubDmChannelRequest) returns (Status);
    rpc LoadSegments(LoadSegmentsRequest) returns (Status);
    rpc ReleaseSegments(ReleaseSegmentsRequest) returns (Status);
    
    // 查询服务
    rpc Search(SearchRequest) returns (SearchResults);
    rpc Query(QueryRequest) returns (RetrieveResults);
    rpc SearchSegments(SearchRequest) returns (SearchResults);  // Worker调用
    rpc QuerySegments(QueryRequest) returns (RetrieveResults);   // Worker调用
    
    // 统计与管理
    rpc GetStatistics(GetStatisticsRequest) returns (GetStatisticsResponse);
    rpc GetSegmentInfo(GetSegmentInfoRequest) returns (GetSegmentInfoResponse);
}

sequenceDiagram
    autonumber
    participant Proxy as Proxy
    participant QN as QueryNode.Server
    participant Delegator as ShardDelegator
    participant Dist as Distribution
    participant Scheduler as Scheduler
    participant Task as SearchTask
    participant Worker as LocalWorker
    participant SegMgr as SegmentManager
    participant Seg as Segment
    participant Knowhere as Knowhere索引
    
    Note over Proxy,Knowhere: 阶段1：请求路由
    Proxy->>QN: Search(vectors, topK, channel)
    QN->>QN: 获取Collection信息
    QN->>Delegator: searchChannel(req, vChannel)
    
    Note over Proxy,Knowhere: 阶段2：等待TSafe
    Delegator->>Delegator: waitTSafe(guaranteeTS)
    Delegator->>Dist: PinReadableSegments(partitions)
    Dist-->>Delegator: sealed[], growing[], version
    
    Note over Proxy,Knowhere: 阶段3：任务组织
    Delegator->>Delegator: organizeSubTask(sealed, growing)
    Delegator->>Delegator: 按Worker分组Segment
    
    Note over Proxy,Knowhere: 阶段4：任务执行（并发）
    par 并发执行多个Worker任务
        Delegator->>Worker: SearchSegments(segmentIDs_1)
        Worker->>Scheduler: 提交SearchTask
        Scheduler->>Task: Execute()
        
        Task->>SegMgr: GetSegments(segmentIDs)
        SegMgr-->>Task: segments[]
        
        loop 对每个Segment
            Task->>Seg: Search(vector, topK, filter)
            Seg->>Knowhere: VectorIndex.Search()
            Knowhere-->>Seg: results
            Seg->>Seg: ApplyScalarFilter()
            Seg-->>Task: SearchResult
        end
        
        Task->>Task: MergeResults(allResults)
        Task-->>Worker: SearchResults_1
        Worker-->>Delegator: SearchResults_1
    and
        Delegator->>Worker: SearchSegments(segmentIDs_2)
        Note over Worker: 同上流程
        Worker-->>Delegator: SearchResults_2
    end
    
    Note over Proxy,Knowhere: 阶段5：结果合并
    Delegator->>Delegator: MergeAllResults(results[])
    Delegator-->>QN: SearchResults
    
    Note over Proxy,Knowhere: 阶段6：Reduce
    QN->>QN: ReduceSearchResults(results)
    QN-->>Proxy: Final SearchResults
    
    Delegator->>Dist: Unpin(version)

// internal/querynodev2/services.go:827
func (node *QueryNode) Search(ctx context.Context, req *querypb.SearchRequest) (*internalpb.SearchResults, error) {
    // 1. 健康检查
    if err := node.lifetime.Add(merr.IsHealthy); err != nil {
        return &internalpb.SearchResults{Status: merr.Status(err)}, nil
    }
    defer node.lifetime.Done()
    
    // 2. 获取Collection
    collection := node.manager.Collection.Get(req.GetReq().GetCollectionID())
    if collection == nil {
        return &internalpb.SearchResults{
            Status: merr.Status(merr.WrapErrCollectionNotLoaded(req.GetReq().GetCollectionID())),
        }, nil
    }
    
    // 3. 路由到Channel
    ch := req.GetDmlChannels()[0]  // 每次Search只针对一个Channel
    ret, err := node.searchChannel(ctx, req, ch)
    if err != nil {
        return &internalpb.SearchResults{Status: merr.Status(err)}, nil
    }
    
    return ret, nil
}

// internal/querynodev2/handlers.go:382
func (node *QueryNode) searchChannel(ctx context.Context, req *querypb.SearchRequest, channel string) (*internalpb.SearchResults, error) {
    // 1. 获取ShardDelegator
    sd, ok := node.delegators.Get(channel)
    if !ok {
        err := merr.WrapErrChannelNotFound(channel)
        log.Warn("Query failed, failed to get shard delegator for search", zap.Error(err))
        return nil, err
    }
    
    // 2. 调用Delegator执行Search
    results, err := sd.Search(ctx, req)
    if err != nil {
        log.Warn("failed to search on delegator", zap.Error(err))
        return nil, err
    }
    
    // 3. Reduce结果
    resp, err := segments.ReduceSearchOnQueryNode(ctx, results, req, node.manager)
    return resp, err
}

// internal/querynodev2/delegator/delegator.go:373
func (sd *shardDelegator) Search(ctx context.Context, req *querypb.SearchRequest) ([]*internalpb.SearchResults, error) {
    // 1. 等待TSafe
    waitTr := timerecord.NewTimeRecorder("wait tSafe")
    tSafe, err := sd.waitTSafe(ctx, req.Req.GetGuaranteeTimestamp())
    if err != nil {
        return nil, err
    }
    if req.GetReq().GetMvccTimestamp() == 0 {
        req.Req.MvccTimestamp = tSafe
    }
    
    // 2. Pin Segment Distribution
    sealed, growing, sealedRowCount, version, err := sd.distribution.PinReadableSegments(
        partialResultRequiredDataRatio,
        req.GetReq().GetPartitionIDs()...,
    )
    if err != nil {
        return nil, err
    }
    defer sd.distribution.Unpin(version)
    
    // 3. Segment裁剪（可选）
    if paramtable.Get().QueryNodeCfg.EnableSegmentPrune.GetAsBool() {
        PruneSegments(ctx, sd.partitionStats, nil, req.GetReq(), sd.collection.Schema(), sealed, pruneInfo)
    }
    
    // 4. 组织SubTask
    tasks, err := organizeSubTask(ctx, req, sealed, growing, sd, false, sd.modifySearchRequest)
    if err != nil {
        return nil, err
    }
    
    // 5. 执行SubTask
    results, err := executeSubTasks(ctx, tasks, NewRowCountBasedEvaluator(sealedRowCount), 
        func(ctx context.Context, req *querypb.SearchRequest, worker cluster.Worker) (*internalpb.SearchResults, error) {
            resp, err := worker.SearchSegments(ctx, req)
            // 处理Worker不可用
            status, ok := status.FromError(err)
            if ok && status.Code() == codes.Unavailable {
                sd.markSegmentOffline(req.GetSegmentIDs()...)
            }
            return resp, err
        }, "Search", log)
    
    return results, err
}

// internal/querynodev2/services.go:739
func (node *QueryNode) SearchSegments(ctx context.Context, req *querypb.SearchRequest) (*internalpb.SearchResults, error) {
    // 1. 健康检查
    if err := node.lifetime.Add(merr.IsHealthy); err != nil {
        resp.Status = merr.Status(err)
        return resp, nil
    }
    defer node.lifetime.Done()
    
    // 2. 获取Collection
    collection := node.manager.Collection.Get(req.Req.GetCollectionID())
    
    // 3. 创建SearchTask
    var task scheduler.Task
    task = tasks.NewSearchTask(ctx, collection, node.manager, req)
    
    // 4. 提交到Scheduler
    if err := node.scheduler.Add(task); err != nil {
        return nil, err
    }
    
    // 5. 等待执行完成
    if err := task.Wait(); err != nil {
        return nil, err
    }
    
    // 6. 返回结果
    return task.Result(), nil
}

// internal/querynodev2/tasks/search_task.go:133
func (t *SearchTask) Execute() error {
    // 1. 创建SearchRequest（C++ Segcore）
    searchReq, err := segcore.NewSearchRequest(t.collection.GetCCollection(), req, t.placeholderGroup)
    defer searchReq.Delete()
    
    // 2. 根据Scope查询不同类型的Segment
    var results []*segments.SearchResult
    if req.GetScope() == querypb.DataScope_Historical {
        // 查询Sealed Segment
        results, searchedSegments, err = segments.SearchHistorical(
            t.ctx,
            t.segmentManager,
            searchReq,
            req.GetReq().GetCollectionID(),
            req.GetReq().GetPartitionIDs(),
            req.GetSegmentIDs(),
        )
    } else if req.GetScope() == querypb.DataScope_Streaming {
        // 查询Growing Segment
        results, searchedSegments, err = segments.SearchStreaming(...)
    }
    defer t.segmentManager.Segment.Unpin(searchedSegments)
    defer segments.DeleteSearchResults(results)
    
    // 3. Reduce结果
    ret, err := segments.ReduceSearchResultsAndFillData(
        ctx,
        searchReq.Plan(),
        results,
        int64(len(results)),
        t.originNqs,
        t.originTopks,
    )
    
    t.result = ret
    return nil
}

// internal/querynodev2/segments/segment.go
func (s *LocalSegment) Search(ctx context.Context, searchReq *SearchRequest) (*SearchResult, error) {
    // 1. 调用C++ Segcore执行检索
    searchResult, err := s.ptrLock.RLockIf(
        state.IsNotReleased,
        func() (*SearchResult, error) {
            // 调用C++ API
            return s.search(ctx, searchReq)
        },
    )
    
    return searchResult, err
}

func (s *LocalSegment) search(ctx context.Context, searchReq *SearchRequest) (*SearchResult, error) {
    // CGO调用C++ Segcore
    cRes := C.Search(
        s.ptr,
        searchReq.plan.cSearchPlan,
        searchReq.cPlaceholderGroup,
        C.uint64_t(searchReq.mvccTimestamp),
    )
    
    // 转换结果
    result := &SearchResult{
        cSearchResult: cRes,
    }
    
    return result, nil
}

sequenceDiagram
    autonumber
    participant QC as QueryCoord
    participant QN as QueryNode.Server
    participant Delegator as ShardDelegator
    participant Loader as SegmentLoader
    participant SM as SegmentManager
    participant OS as Object Storage
    participant Seg as LocalSegment
    
    Note over QC,Seg: 阶段1：接收加载请求
    QC->>QN: LoadSegments(segmentLoadInfos)
    QN->>QN: 获取Collection元信息
    
    alt needTransfer = true（需要Delegator处理）
        QN->>Delegator: LoadSegments(req)
        Delegator->>Delegator: 确定目标Worker
        alt 目标是本地
            Delegator->>QN: LoadSegments(needTransfer=false)
        else 目标是远程
            Delegator->>RemoteQN: LoadSegments via Worker
            RemoteQN-->>Delegator: Success
        end
    end
    
    Note over QC,Seg: 阶段2：准备加载
    QN->>Loader: Load(segmentLoadInfos)
    Loader->>Loader: prepare()：过滤已加载的Segment
    Loader->>Loader: requestResource()：检查内存限制
    
    Note over QC,Seg: 阶段3：并发加载Segment
    par 并发加载多个Segment
        loop 对每个Segment
            Loader->>OS: DownloadBinlog(insertLog)
            OS-->>Loader: Binlog数据
            
            Loader->>OS: DownloadBinlog(indexFiles)
            OS-->>Loader: 索引文件
            
            Loader->>OS: DownloadBinlog(statsLog)
            OS-->>Loader: 统计信息
            
            Loader->>OS: DownloadBinlog(deltaLog)
            OS-->>Loader: Delete数据
        end
        
        Loader->>Seg: 创建LocalSegment
        Loader->>Seg: LoadMultiFieldData()：加载字段数据
        Loader->>Seg: LoadIndex()：加载向量索引
        Loader->>Seg: LoadStatslog()：加载统计信息
        Loader->>Seg: LoadDeltaLog()：加载Delete数据
        
        Seg-->>Loader: 加载完成
    end
    
    Note over QC,Seg: 阶段4：注册Segment
    Loader->>SM: Put(segment)
    Loader->>Delegator: LoadSegments完成，更新Distribution
    Delegator->>Delegator: SyncDistribution()：同步Segment分布
    
    Note over QC,Seg: 阶段5：应用Delete
    Delegator->>Delegator: forwardDelete(segment)
    Delegator->>Delegator: applyBFInParallel()：使用BloomFilter过滤Delete
    Delegator->>Seg: Delete(pks, tss)
    
    Loader-->>QN: 加载成功
    QN-->>QC: Success
    
    Note over QC,Seg: 阶段6：心跳上报
    QN->>QC: HeartBeat：上报已加载Segment

// internal/querynodev2/services.go:468
func (node *QueryNode) LoadSegments(ctx context.Context, req *querypb.LoadSegmentsRequest) (*commonpb.Status, error) {
    // 1. 健康检查
    if err := node.lifetime.Add(merr.IsHealthy); err != nil {
        return merr.Status(err), nil
    }
    defer node.lifetime.Done()
    
    // 2. 估算Binlog大小（如果没有提供）
    for _, s := range req.GetInfos() {
        fallbackBinlogMemorySize(s.GetBinlogPaths())
        fallbackBinlogMemorySize(s.GetDeltalogs())
    }
    
    // 3. 判断是否需要通过Delegator转发
    if req.GetNeedTransfer() {
        delegator, ok := node.delegators.Get(segment.GetInsertChannel())
        if !ok {
            return merr.Status(merr.WrapErrChannelNotFound(segment.GetInsertChannel())), nil
        }
        
        // 通过Delegator处理加载
        req.NeedTransfer = false
        err := delegator.LoadSegments(ctx, req)
        if err != nil {
            return merr.Status(err), nil
        }
        return merr.Success(), nil
    }
    
    // 4. 直接加载
    err := node.manager.Collection.PutOrRef(req.GetCollectionID(), req.GetSchema(), indexMeta, loadMeta)
    defer node.manager.Collection.Unref(req.GetCollectionID(), 1)
    
    // 5. 调用Loader加载
    loaded, err := node.loader.Load(ctx,
        req.GetCollectionID(),
        segments.SegmentTypeSealed,
        req.GetVersion(),
        req.GetInfos()...,
    )
    if err != nil {
        return merr.Status(err), nil
    }
    
    // 6. 增加Collection引用计数
    node.manager.Collection.Ref(req.GetCollectionID(), uint32(len(loaded)))
    
    return merr.Success(), nil
}

// internal/querynodev2/delegator/delegator_data.go:396
func (sd *shardDelegator) LoadSegments(ctx context.Context, req *querypb.LoadSegmentsRequest) error {
    if len(req.GetInfos()) == 0 {
        return nil
    }
    
    targetNodeID := req.GetDstNodeID()
    
    // 1. Pin Delete Buffer（防止清理）
    for _, info := range req.GetInfos() {
        sd.deleteBuffer.Pin(info.GetStartPosition().GetTimestamp(), info.GetSegmentID())
    }
    defer func() {
        for _, info := range req.GetInfos() {
            sd.deleteBuffer.Unpin(info.GetStartPosition().GetTimestamp(), info.GetSegmentID())
        }
    }()
    
    // 2. 获取目标Worker
    worker, err := sd.workerManager.GetWorker(ctx, targetNodeID)
    if err != nil {
        return err
    }
    
    // 3. 调用Worker加载Segment
    req.Base.TargetID = targetNodeID
    err = worker.LoadSegments(ctx, req)
    
    // 4. 更新Distribution
    sd.distribution.AddDistributions(/* segment info */)
    
    // 5. 应用Delete Buffer
    sd.forwardDelete(ctx, req.GetInfos(), targetNodeID)
    
    return nil
}

// internal/querynodev2/segments/segment_loader.go:249
func (loader *segmentLoader) Load(ctx context.Context,
    collectionID int64,
    segmentType SegmentType,
    version int64,
    segments ...*querypb.SegmentLoadInfo,
) ([]Segment, error) {
    if len(segments) == 0 {
        return nil, nil
    }
    
    collection := loader.manager.Collection.Get(collectionID)
    if collection == nil {
        return nil, merr.WrapErrCollectionNotFound(collectionID)
    }
    
    // 1. 过滤已加载的Segment
    infos := loader.prepare(ctx, segmentType, segments...)
    defer loader.unregister(infos...)
    
    // 2. 检查资源（内存/磁盘）
    if !isLazyLoad(collection, segmentType) {
        requestResourceResult, err := loader.requestResource(ctx, infos...)
        if err != nil {
            return nil, err
        }
        defer loader.freeRequest(requestResourceResult.Resource, requestResourceResult.LogicalResource)
    }
    
    // 3. 并发加载多个Segment
    newSegments := typeutil.NewConcurrentMap[int64, Segment]()
    loaded := typeutil.NewConcurrentMap[int64, Segment]()
    
    loadedSegments, err := loader.loadSegments(ctx, collection, segmentType, version, infos, newSegments, loaded)
    if err != nil {
        return nil, err
    }
    
    // 4. 注册到SegmentManager
    for _, segment := range loadedSegments {
        loader.manager.Segment.Put(ctx, segmentType, segment)
    }
    
    return loadedSegments, nil
}

// internal/querynodev2/segments/segment_loader.go:1002
func (loader *segmentLoader) LoadSegment(ctx context.Context,
    seg Segment,
    loadInfo *querypb.SegmentLoadInfo,
) (err error) {
    segment, ok := seg.(*LocalSegment)
    if !ok {
        return merr.WrapErrParameterInvalid("LocalSegment", fmt.Sprintf("%T", seg))
    }
    
    collection := loader.manager.Collection.Get(segment.Collection())
    pkField := GetPkField(collection.Schema())
    
    if segment.Type() == SegmentTypeSealed {
        // 加载Sealed Segment
        if err := loader.loadSealedSegment(ctx, loadInfo, segment); err != nil {
            return err
        }
    } else {
        // 加载Growing Segment
        if err := segment.LoadMultiFieldData(ctx); err != nil {
            return err
        }
    }
    
    // 加载统计信息
    if segment.segmentType == SegmentTypeGrowing {
        if err := loader.loadStatslog(ctx, collection, segment, loadInfo.GetStatsLogs()); err != nil {
            return err
        }
    }
    
    // 加载Delete数据
    if err := loader.loadDeltaLogs(ctx, segment, loadInfo.GetDeltalogs()); err != nil {
        return err
    }
    
    return nil
}

// internal/querynodev2/segments/segment_loader.go
func (loader *segmentLoader) loadSealedSegment(ctx context.Context, loadInfo *querypb.SegmentLoadInfo, segment *LocalSegment) error {
    // 1. 加载字段数据
    if err := segment.LoadMultiFieldData(ctx); err != nil {
        return err
    }
    
    // 2. 加载索引
    if err := loader.LoadIndex(ctx, segment, loadInfo, version); err != nil {
        return err
    }
    
    return nil
}

func (loader *segmentLoader) LoadIndex(ctx context.Context, segment *LocalSegment, loadInfo *querypb.SegmentLoadInfo, version int64) error {
    for _, indexInfo := range loadInfo.GetIndexInfos() {
        fieldID := indexInfo.GetFieldID()
        
        // 下载索引文件
        indexData, err := loader.downloadIndexFiles(ctx, indexInfo)
        if err != nil {
            return err
        }
        
        // 加载到Segment
        if err := segment.LoadIndex(ctx, fieldID, indexData, indexInfo); err != nil {
            return err
        }
    }
    
    return nil
}

sequenceDiagram
    autonumber
    participant Proxy as Proxy
    participant QN as QueryNode.Server
    participant Delegator1 as ShardDelegator-1
    participant Delegator2 as ShardDelegator-2
    participant Worker as LocalWorker
    participant SegMgr as SegmentManager
    
    Note over Proxy,SegMgr: 阶段1：请求分发
    Proxy->>QN: Query(channels[ch1, ch2])
    
    par 并发查询多个Channel
        QN->>Delegator1: queryChannel(ch1)
        Delegator1->>Delegator1: waitTSafe()
        Delegator1->>Delegator1: PinReadableSegments()
        Delegator1->>Delegator1: organizeSubTask()
        Delegator1->>Worker: QuerySegments(segmentIDs_1)
        Worker->>SegMgr: Query on segments
        SegMgr-->>Worker: RetrieveResults_1
        Worker-->>Delegator1: RetrieveResults_1
        Delegator1-->>QN: RetrieveResults_1
    and
        QN->>Delegator2: queryChannel(ch2)
        Note over Delegator2,Worker: 同上流程
        Delegator2-->>QN: RetrieveResults_2
    end
    
    Note over Proxy,SegMgr: 阶段2：合并结果
    QN->>QN: CreateInternalReducer()
    QN->>QN: Reduce(results[])
    QN->>QN: 去重、排序、限制数量
    QN-->>Proxy: Final RetrieveResults

sequenceDiagram
    autonumber
    participant QC as QueryCoord
    participant QN as QueryNode.Server
    participant CM as ClusterManager
    participant Delegator as ShardDelegator
    participant PM as PipelineManager
    participant Pipeline as Pipeline
    participant MQ as MessageQueue
    
    Note over QC,MQ: 阶段1：接收订阅请求
    QC->>QN: WatchDmChannels(channel, seekPosition)
    QN->>QN: 获取Collection元信息
    
    Note over QC,MQ: 阶段2：创建ShardDelegator
    QN->>Delegator: NewShardDelegator(channel)
    Delegator->>Delegator: 初始化Distribution
    Delegator->>Delegator: 初始化DeleteBuffer
    Delegator->>Delegator: 初始化PKOracle
    QN->>QN: delegators.Insert(channel, delegator)
    
    Note over QC,MQ: 阶段3：创建Pipeline
    QN->>PM: Add(collectionID, channel)
    PM->>Pipeline: NewPipeline(channel, delegator)
    Pipeline->>Pipeline: 创建FilterNode
    Pipeline->>Pipeline: 创建EmbeddingNode（可选）
    Pipeline->>Pipeline: 创建InsertNode
    Pipeline->>Pipeline: 创建DeleteNode
    Pipeline->>Pipeline: 连接Node流程
    
    Note over QC,MQ: 阶段4：加载Growing Segment
    loop 对每个Growing Segment
        QN->>Delegator: LoadGrowing(segmentInfo)
        Delegator->>Delegator: 创建Growing Segment
        Delegator->>Delegator: AddExcludedSegments()
    end
    
    Note over QC,MQ: 阶段5：订阅消息流
    QN->>Pipeline: ConsumeMsgStream(seekPosition)
    Pipeline->>MQ: Subscribe(channel, seekPosition)
    MQ-->>Pipeline: Subscription确认
    
    Note over QC,MQ: 阶段6：启动Pipeline
    QN->>Pipeline: Start()
    Pipeline->>Pipeline: 启动各个Node
    QN->>Delegator: Start()
    Delegator->>Delegator: 设置为Serviceable状态
    
    QN-->>QC: Success
    
    Note over QC,MQ: 阶段7：持续消费DML消息
    loop 持续消费
        MQ->>Pipeline: DML Messages
        Pipeline->>FilterNode: 过滤消息
        FilterNode->>InsertNode: InsertMsg
        FilterNode->>DeleteNode: DeleteMsg
        InsertNode->>Delegator: ProcessInsert()
        DeleteNode->>Delegator: ProcessDelete()
    end

flowchart LR
    MQ[MessageQueue] -->|MsgPack| Filter[FilterNode]
    
    subgraph Filter Processing
        Filter -->|验证消息| Check{消息类型}
        Check -->|Insert| ValidateInsert[验证SegmentID<br/>是否Excluded]
        Check -->|Delete| ValidateDelete[验证Delete]
        Check -->|SchemaChange| ValidateSchema[验证Schema]
    end
    
    ValidateInsert -->|Pass| Embed[EmbeddingNode<br/>可选]
    ValidateDelete -->|Pass| Embed
    ValidateSchema -->|Pass| Embed
    
    Embed -->|InsertData| Insert[InsertNode]
    Embed -->|DeleteData| Delete[DeleteNode]
    
    subgraph Insert Processing
        Insert -->|groupBySegment| Group[按SegmentID分组]
        Group -->|ProcessInsert| Delegator1[Delegator.ProcessInsert]
        Delegator1 -->|AppendRows| GrowingSeg[Growing Segment]
        Delegator1 -->|UpdatePK| PKOracle[PKOracle<br/>BloomFilter]
    end
    
    subgraph Delete Processing
        Delete -->|批量Delete| Delegator2[Delegator.ProcessDelete]
        Delegator2 -->|Put| DeleteBuffer[DeleteBuffer]
        Delegator2 -->|Forward| StreamingNode[StreamingNode<br/>可选]
    end
    
    style Filter fill:#e1f5ff
    style Insert fill:#e6ffe6
    style Delete fill:#ffe6e6

// internal/querynodev2/services.go:187
func (node *QueryNode) WatchDmChannels(ctx context.Context, req *querypb.WatchDmChannelsRequest) (*commonpb.Status, error) {
    // 1. 健康检查
    if err := node.lifetime.Add(merr.IsHealthy); err != nil {
        return merr.Status(err), nil
    }
    defer node.lifetime.Done()
    
    // 2. 获取Channel信息
    channel := req.GetInfos()[0]  // 每次只Watch一个Channel
    
    // 3. 注册Collection
    err := node.manager.Collection.PutOrRef(req.GetCollectionID(), req.GetSchema(), indexMeta, loadMeta)
    defer func() {
        if err != nil {
            node.manager.Collection.Unref(req.GetCollectionID(), 1)
        }
    }()
    
    // 4. 创建ShardDelegator
    delegator, err := delegator.NewShardDelegator(
        ctx,
        req.GetCollectionID(),
        req.GetReplicaID(),
        channel.GetChannelName(),
        req.GetVersion(),
        node.clusterManager,
        node.manager,
        node.loader,
        channel.GetSeekPosition().GetTimestamp(),
        node.queryHook,
        node.chunkManager,
        queryView,
    )
    if err != nil {
        return merr.Status(err), nil
    }
    node.delegators.Insert(channel.GetChannelName(), delegator)
    
    // 5. 创建Pipeline
    pipeline, err := node.pipelineManager.Add(req.GetCollectionID(), channel.GetChannelName())
    if err != nil {
        return merr.Status(err), nil
    }
    
    // 6. 加载Growing Segment
    delegator.AddExcludedSegments(growingInfo)   // 标记已有的Growing
    delegator.AddExcludedSegments(flushedInfo)   // 标记已Flush的
    delegator.AddExcludedSegments(droppedInfo)   // 标记已Drop的
    
    // 7. 加载L0 Segment（Delete Segment）
    err = loadL0Segments(ctx, delegator, req)
    
    // 8. 订阅消息流
    err = pipeline.ConsumeMsgStream(ctx, position)
    if err != nil {
        return merr.Status(err), nil
    }
    
    // 9. 启动
    pipeline.Start()
    delegator.Start()
    
    return merr.Success(), nil
}

// internal/querynodev2/pipeline/pipeline.go:48
func NewPipeLine(
    collection *Collection,
    channel string,
    manager *DataManager,
    dispatcher msgdispatcher.Client,
    delegator delegator.ShardDelegator,
) (Pipeline, error) {
    collectionID := collection.ID()
    
    // 创建Pipeline
    p := &pipeline{
        collectionID:   collectionID,
        StreamPipeline: base.NewPipelineWithStream(dispatcher, nodeCtxTtInterval, enableTtChecker, channel, replicateConfig),
    }
    
    // 创建Node
    filterNode := newFilterNode(collectionID, channel, manager, delegator, pipelineQueueLength)
    embeddingNode, err := newEmbeddingNode(collectionID, channel, manager, pipelineQueueLength)
    insertNode := newInsertNode(collectionID, channel, manager, delegator, pipelineQueueLength)
    deleteNode := newDeleteNode(collectionID, channel, manager, delegator, pipelineQueueLength)
    
    // 连接Node
    if embeddingNode != nil {
        p.Add(filterNode, embeddingNode, insertNode, deleteNode)
    } else {
        p.Add(filterNode, insertNode, deleteNode)
    }
    
    return p, nil
}

// internal/querynodev2/pipeline/filter_node.go:50
func (fNode *filterNode) Operate(in Msg) Msg {
    streamMsgPack, ok := in.(*msgstream.MsgPack)
    if !ok {
        return nil
    }
    
    collection := fNode.manager.Collection.Get(fNode.collectionID)
    
    out := &insertNodeMsg{
        insertMsgs: []*InsertMsg{},
        deleteMsgs: []*DeleteMsg{},
        timeRange: TimeRange{
            timestampMin: streamMsgPack.BeginTs,
            timestampMax: streamMsgPack.EndTs,
        },
    }
    
    // 过滤每条消息
    for _, msg := range streamMsgPack.Msgs {
        err := fNode.filtrate(collection, msg)
        if err != nil {
            log.Debug("filter invalid message", zap.Error(err))
        } else {
            out.append(msg)  // 添加到输出
        }
    }
    
    // 尝试清理ExcludedSegments
    fNode.delegator.TryCleanExcludedSegments(streamMsgPack.EndTs)
    
    return out
}

func (fNode *filterNode) filtrate(c *Collection, msg msgstream.TsMsg) error {
    switch msg.Type() {
    case commonpb.MsgType_Insert:
        insertMsg := msg.(*msgstream.InsertMsg)
        
        // 检查Segment是否Excluded
        ok := fNode.delegator.VerifyExcludedSegments(insertMsg.SegmentID, insertMsg.EndTimestamp)
        if !ok {
            return merr.WrapErrServiceInternal("skip msg due to segment excluded")
        }
        return nil
        
    case commonpb.MsgType_Delete:
        deleteMsg := msg.(*msgstream.DeleteMsg)
        return nil  // Delete消息总是接受
        
    default:
        return merr.WrapErrParameterInvalid("msgType is Insert or Delete", "not")
    }
}