milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: Set legacy level to l0 segment after qc restart (#35197) 

issue: #35087
after qc restarts, and target is not ready yet, if dist_handler try to
update segment dist, it will set legacy level to l0 segment, which may
cause l0 segment be moved to other node, cause search/query failed.

Signed-off-by: Wei Liu <wei.liu@zilliz.com>
pull/34491/head
wei liu 2024-08-02 10:18:13 +08:00 committed by GitHub
parent 4dbd023402
commit 27b6d58981
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 30 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
internal/proto/query_coord.proto
View File

@ -626,6 +626,7 @@ message SegmentVersionInfo {
int64 version = 5;
uint64 last_delta_timestamp = 6;
map<int64, FieldIndexInfo> index_info = 7;
data.SegmentLevel level = 8;
}
message ChannelVersionInfo {

4
internal/querycoordv2/balance/channel_level_score_balancer.go
View File

@ -147,7 +147,7 @@ func (b *ChannelLevelScoreBalancer) genStoppingSegmentPlan(replica *meta.Replica
for _, nodeID := range offlineNodes {
dist := b.dist.SegmentDistManager.GetByFilter(meta.WithCollectionID(replica.GetCollectionID()), meta.WithNodeID(nodeID), meta.WithChannel(channelName))
segments := lo.Filter(dist, func(segment *meta.Segment, _ int) bool {
return b.targetMgr.CanSegmentBeMoved(segment)
return b.targetMgr.CanSegmentBeMoved(segment.GetCollectionID(), segment.GetID())
})
plans := b.AssignSegment(replica.GetCollectionID(), segments, onlineNodes, false)
for i := range plans {
@ -168,7 +168,7 @@ func (b *ChannelLevelScoreBalancer) genSegmentPlan(replica *meta.Replica, channe
for _, node := range onlineNodes {
dist := b.dist.SegmentDistManager.GetByFilter(meta.WithCollectionID(replica.GetCollectionID()), meta.WithNodeID(node), meta.WithChannel(channelName))
segments := lo.Filter(dist, func(segment *meta.Segment, _ int) bool {
return b.targetMgr.CanSegmentBeMoved(segment)
return b.targetMgr.CanSegmentBeMoved(segment.GetCollectionID(), segment.GetID())
})
segmentDist[node] = segments
totalScore += nodeScore[node].getPriority()

2
internal/querycoordv2/balance/multi_target_balance.go
View File

@ -510,7 +510,7 @@ func (b *MultiTargetBalancer) genSegmentPlan(replica *meta.Replica, rwNodes []in
for _, node := range rwNodes {
dist := b.dist.SegmentDistManager.GetByFilter(meta.WithCollectionID(replica.GetCollectionID()), meta.WithNodeID(node))
segments := lo.Filter(dist, func(segment *meta.Segment, _ int) bool {
return b.targetMgr.CanSegmentBeMoved(segment)
return b.targetMgr.CanSegmentBeMoved(segment.GetCollectionID(), segment.GetID())
})
nodeSegments[node] = segments
globalNodeSegments[node] = b.dist.SegmentDistManager.GetByFilter(meta.WithNodeID(node))

4
internal/querycoordv2/balance/rowcount_based_balancer.go
View File

@ -219,7 +219,7 @@ func (b *RowCountBasedBalancer) genStoppingSegmentPlan(replica *meta.Replica, rw
for _, nodeID := range roNodes {
dist := b.dist.SegmentDistManager.GetByFilter(meta.WithCollectionID(replica.GetCollectionID()), meta.WithNodeID(nodeID))
segments := lo.Filter(dist, func(segment *meta.Segment, _ int) bool {
return b.targetMgr.CanSegmentBeMoved(segment)
return b.targetMgr.CanSegmentBeMoved(segment.GetCollectionID(), segment.GetID())
})
plans := b.AssignSegment(replica.GetCollectionID(), segments, rwNodes, false)
for i := range plans {
@ -240,7 +240,7 @@ func (b *RowCountBasedBalancer) genSegmentPlan(replica *meta.Replica, rwNodes []
for _, node := range rwNodes {
dist := b.dist.SegmentDistManager.GetByFilter(meta.WithCollectionID(replica.GetCollectionID()), meta.WithNodeID(node))
segments := lo.Filter(dist, func(segment *meta.Segment, _ int) bool {
return b.targetMgr.CanSegmentBeMoved(segment)
return b.targetMgr.CanSegmentBeMoved(segment.GetCollectionID(), segment.GetID())
})
rowCount := 0
for _, s := range segments {

4
internal/querycoordv2/balance/score_based_balancer.go
View File

@ -264,7 +264,7 @@ func (b *ScoreBasedBalancer) genStoppingSegmentPlan(replica *meta.Replica, onlin
for _, nodeID := range offlineNodes {
dist := b.dist.SegmentDistManager.GetByFilter(meta.WithCollectionID(replica.GetCollectionID()), meta.WithNodeID(nodeID))
segments := lo.Filter(dist, func(segment *meta.Segment, _ int) bool {
return b.targetMgr.CanSegmentBeMoved(segment)
return b.targetMgr.CanSegmentBeMoved(segment.GetCollectionID(), segment.GetID())
})
plans := b.AssignSegment(replica.GetCollectionID(), segments, onlineNodes, false)
for i := range plans {
@ -285,7 +285,7 @@ func (b *ScoreBasedBalancer) genSegmentPlan(replica *meta.Replica, onlineNodes [
for _, node := range onlineNodes {
dist := b.dist.SegmentDistManager.GetByFilter(meta.WithCollectionID(replica.GetCollectionID()), meta.WithNodeID(node))
segments := lo.Filter(dist, func(segment *meta.Segment, _ int) bool {
return b.targetMgr.CanSegmentBeMoved(segment)
return b.targetMgr.CanSegmentBeMoved(segment.GetCollectionID(), segment.GetID())
})
segmentDist[node] = segments
totalScore += nodeScore[node].getPriority()

41
internal/querycoordv2/dist/dist_handler.go vendored
View File

@ -137,36 +137,23 @@ func (dh *distHandler) handleDistResp(resp *querypb.GetDataDistributionResponse,
func (dh *distHandler) updateSegmentsDistribution(resp *querypb.GetDataDistributionResponse) {
updates := make([]*meta.Segment, 0, len(resp.GetSegments()))
for _, s := range resp.GetSegments() {
// for collection which is already loaded
segmentInfo := dh.target.GetSealedSegment(s.GetCollection(), s.GetID(), meta.CurrentTarget)
segmentInfo := dh.target.GetSealedSegment(s.GetCollection(), s.GetID(), meta.CurrentTargetFirst)
if segmentInfo == nil {
// for collection which is loading
segmentInfo = dh.target.GetSealedSegment(s.GetCollection(), s.GetID(), meta.NextTarget)
}
var segment *meta.Segment
if segmentInfo == nil {
segment = &meta.Segment{
SegmentInfo: &datapb.SegmentInfo{
ID: s.GetID(),
CollectionID: s.GetCollection(),
PartitionID: s.GetPartition(),
InsertChannel: s.GetChannel(),
},
Node: resp.GetNodeID(),
Version: s.GetVersion(),
LastDeltaTimestamp: s.GetLastDeltaTimestamp(),
IndexInfo: s.GetIndexInfo(),
}
} else {
segment = &meta.Segment{
SegmentInfo: proto.Clone(segmentInfo).(*datapb.SegmentInfo),
Node: resp.GetNodeID(),
Version: s.GetVersion(),
LastDeltaTimestamp: s.GetLastDeltaTimestamp(),
IndexInfo: s.GetIndexInfo(),
segmentInfo = &datapb.SegmentInfo{
ID: s.GetID(),
CollectionID: s.GetCollection(),
PartitionID: s.GetPartition(),
InsertChannel: s.GetChannel(),
Level: s.GetLevel(),
}
}
updates = append(updates, segment)
updates = append(updates, &meta.Segment{
SegmentInfo: proto.Clone(segmentInfo).(*datapb.SegmentInfo),
Node: resp.GetNodeID(),
Version: s.GetVersion(),
LastDeltaTimestamp: s.GetLastDeltaTimestamp(),
IndexInfo: s.GetIndexInfo(),
})
}
dh.dist.SegmentDistManager.Update(resp.GetNodeID(), updates...)

16
internal/querycoordv2/meta/target_manager.go
View File

@ -692,18 +692,16 @@ func (mgr *TargetManager) Recover(catalog metastore.QueryCoordCatalog) error {
}
// if segment isn't l0 segment, and exist in current/next target, then it can be moved
func (mgr *TargetManager) CanSegmentBeMoved(segment *Segment) bool {
if segment.GetLevel() == datapb.SegmentLevel_L0 {
return false
}
current := mgr.current.getCollectionTarget(segment.CollectionID)
if current != nil && current.segments[segment.GetID()] != nil {
func (mgr *TargetManager) CanSegmentBeMoved(collectionID, segmentID int64) bool {
mgr.rwMutex.Lock()
defer mgr.rwMutex.Unlock()
current := mgr.current.getCollectionTarget(collectionID)
if current != nil && current.segments[segmentID] != nil && current.segments[segmentID].GetLevel() != datapb.SegmentLevel_L0 {
return true
}
next := mgr.next.getCollectionTarget(segment.CollectionID)
if next != nil && next.segments[segment.GetID()] != nil {
next := mgr.next.getCollectionTarget(collectionID)
if next != nil && next.segments[segmentID] != nil && next.segments[segmentID].GetLevel() != datapb.SegmentLevel_L0 {
return true
}

1
internal/querynodev2/services.go
View File

@ -1208,6 +1208,7 @@ func (node *QueryNode) GetDataDistribution(ctx context.Context, req *querypb.Get
Partition: s.Partition(),
Channel: s.Shard().VirtualName(),
Version: s.Version(),
Level: s.Level(),
LastDeltaTimestamp: s.LastDeltaTimestamp(),
IndexInfo: lo.SliceToMap(s.Indexes(), func(info *segments.IndexedFieldInfo) (int64, *querypb.FieldIndexInfo) {
return info.IndexInfo.FieldID, info.IndexInfo