admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2026-02-02 01:06:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

enhance: [2.4] Provide secondary index criteria when filter leaderview (#37777) (#37802)

Browse Source 
Cherry-pick from master
pr: #37777 
Related to #37630

---------

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
This commit is contained in:
congqixia 2024-11-21 10:48:33 +08:00 committed by GitHub
parent 28adfe4629
commit 0bd26171d5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 12 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
internal/querycoordv2/observers/collection_observer.go
View File

@ -325,7 +325,9 @@ func (ob *CollectionObserver) observePartitionLoadStatus(ctx context.Context, pa
}
subChannelCount := loadedCount
for _, segment := range segmentTargets {
views := ob.dist.LeaderViewManager.GetByFilter(meta.WithSegment2LeaderView(segment.GetID(), false))
views := ob.dist.LeaderViewManager.GetByFilter(
meta.WithChannelName2LeaderView(segment.GetInsertChannel()),
meta.WithSegment2LeaderView(segment.GetID(), false))
nodes := lo.Map(views, func(view *meta.LeaderView, _ int) int64 { return view.ID })
group := utils.GroupNodesByReplica(ob.meta.ReplicaManager, partition.GetCollectionID(), nodes)
loadedCount += len(group)

4
internal/querycoordv2/task/action.go
View File

@ -123,7 +123,9 @@ func (action *SegmentAction) IsFinished(distMgr *meta.DistributionManager) bool
}
// segment found in leader view
views := distMgr.LeaderViewManager.GetByFilter(meta.WithSegment2LeaderView(action.segmentID, false))
views := distMgr.LeaderViewManager.GetByFilter(
meta.WithChannelName2LeaderView(action.Shard()),
meta.WithSegment2LeaderView(action.segmentID, false))
if len(views) == 0 {
return false
}

4
internal/querycoordv2/task/scheduler.go
View File

@ -345,7 +345,9 @@ func (scheduler *taskScheduler) preAdd(task Task) error {
taskType := GetTaskType(task)
if taskType == TaskTypeMove {
views := scheduler.distMgr.LeaderViewManager.GetByFilter(meta.WithSegment2LeaderView(task.SegmentID(), false))
views := scheduler.distMgr.LeaderViewManager.GetByFilter(
meta.WithChannelName2LeaderView(task.Shard()),
meta.WithSegment2LeaderView(task.SegmentID(), false))
if len(views) == 0 {
return merr.WrapErrServiceInternal("segment's delegator not found, stop balancing")
}

3
internal/querycoordv2/task/task_test.go
View File

@ -476,6 +476,7 @@ func (suite *TaskSuite) TestLoadSegmentTask() {
ID: targetNode,
CollectionID: suite.collection,
Segments: map[int64]*querypb.SegmentDist{},
Channel: channel.ChannelName,
}
for _, segment := range suite.loadSegments {
view.Segments[segment] = &querypb.SegmentDist{NodeID: targetNode, Version: 0}
@ -575,6 +576,7 @@ func (suite *TaskSuite) TestLoadSegmentTaskNotIndex() {
ID: targetNode,
CollectionID: suite.collection,
Segments: map[int64]*querypb.SegmentDist{},
Channel: channel.ChannelName,
}
for _, segment := range suite.loadSegments {
view.Segments[segment] = &querypb.SegmentDist{NodeID: targetNode, Version: 0}
@ -1139,6 +1141,7 @@ func (suite *TaskSuite) TestSegmentTaskStale() {
ID: targetNode,
CollectionID: suite.collection,
Segments: map[int64]*querypb.SegmentDist{},
Channel: channel.ChannelName,
}
for _, segment := range suite.loadSegments[1:] {
view.Segments[segment] = &querypb.SegmentDist{NodeID: targetNode, Version: 0}