Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: Wrong behavior of CurrentTargetFirst/NextTargetFirst in target manager #31378

Merged
merged 2 commits into from
Mar 19, 2024
+142 −98
Merged

fix: Wrong behavior of CurrentTargetFirst/NextTargetFirst in target manager #31378

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4fd5ec1
fix: Wrong behavior of CurrentTargetFirst/NextTargetFirst in target m…
weiliu1031 Mar 18, 2024
31b6fa0
fix dead lock
weiliu1031 Mar 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
195 changes: 117 additions & 78 deletions internal/querycoordv2/meta/target_manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -330,22 +330,49 @@
return NewCollectionTarget(segments, channels)
}

func (mgr *TargetManager) getCollectionTarget(scope TargetScope, collectionID int64) *CollectionTarget {
func (mgr *TargetManager) getCollectionTarget(scope TargetScope, collectionID int64) []*CollectionTarget {
switch scope {
case CurrentTarget:
return mgr.current.getCollectionTarget(collectionID)

ret := make([]*CollectionTarget, 0, 1)
current := mgr.current.getCollectionTarget(collectionID)
if current != nil {
ret = append(ret, current)
}
return ret
case NextTarget:
return mgr.next.getCollectionTarget(collectionID)
ret := make([]*CollectionTarget, 0, 1)
next := mgr.next.getCollectionTarget(collectionID)
if next != nil {
ret = append(ret, next)
}
return ret
case CurrentTargetFirst:
if current := mgr.current.getCollectionTarget(collectionID); current != nil {
return current
ret := make([]*CollectionTarget, 0, 2)
current := mgr.current.getCollectionTarget(collectionID)
if current != nil {
ret = append(ret, current)
}

next := mgr.next.getCollectionTarget(collectionID)
if next != nil {
ret = append(ret, next)
}
return mgr.next.getCollectionTarget(collectionID)

return ret
case NextTargetFirst:
if next := mgr.next.getCollectionTarget(collectionID); next != nil {
return next
ret := make([]*CollectionTarget, 0, 2)
next := mgr.next.getCollectionTarget(collectionID)
if next != nil {
ret = append(ret, next)
}

current := mgr.current.getCollectionTarget(collectionID)
if current != nil {
ret = append(ret, current)
}
return mgr.current.getCollectionTarget(collectionID)

return ret
}
return nil
}
Expand All @@ -356,18 +383,20 @@
mgr.rwMutex.RLock()
defer mgr.rwMutex.RUnlock()

collectionTarget := mgr.getCollectionTarget(scope, collectionID)
targets := mgr.getCollectionTarget(scope, collectionID)

if collectionTarget == nil {
return nil
}
for _, t := range targets {
segments := typeutil.NewUniqueSet()
for _, channel := range t.GetAllDmChannels() {
segments.Insert(channel.GetUnflushedSegmentIds()...)
}

segments := typeutil.NewUniqueSet()
for _, channel := range collectionTarget.GetAllDmChannels() {
segments.Insert(channel.GetUnflushedSegmentIds()...)
if len(segments) > 0 {
return segments
}
}

return segments
return nil
}

func (mgr *TargetManager) GetGrowingSegmentsByChannel(collectionID int64,
Expand All @@ -377,20 +406,21 @@
mgr.rwMutex.RLock()
defer mgr.rwMutex.RUnlock()

collectionTarget := mgr.getCollectionTarget(scope, collectionID)

if collectionTarget == nil {
return nil
}
targets := mgr.getCollectionTarget(scope, collectionID)
for _, t := range targets {
segments := typeutil.NewUniqueSet()
for _, channel := range t.GetAllDmChannels() {
if channel.ChannelName == channelName {
segments.Insert(channel.GetUnflushedSegmentIds()...)
}
}

segments := typeutil.NewUniqueSet()
for _, channel := range collectionTarget.GetAllDmChannels() {
if channel.ChannelName == channelName {
segments.Insert(channel.GetUnflushedSegmentIds()...)
if len(segments) > 0 {
return segments
}
}

return segments
return nil
}

func (mgr *TargetManager) GetSealedSegmentsByCollection(collectionID int64,
Expand All @@ -399,12 +429,13 @@
mgr.rwMutex.RLock()
defer mgr.rwMutex.RUnlock()

collectionTarget := mgr.getCollectionTarget(scope, collectionID)
targets := mgr.getCollectionTarget(scope, collectionID)

if collectionTarget == nil {
return nil
for _, t := range targets {
return t.GetAllSegments()
}
return collectionTarget.GetAllSegments()

return nil
}

func (mgr *TargetManager) GetSealedSegmentsByChannel(collectionID int64,
Expand All @@ -414,19 +445,21 @@
mgr.rwMutex.RLock()
defer mgr.rwMutex.RUnlock()

collectionTarget := mgr.getCollectionTarget(scope, collectionID)
if collectionTarget == nil {
return nil
}
targets := mgr.getCollectionTarget(scope, collectionID)
for _, t := range targets {
ret := make(map[int64]*datapb.SegmentInfo)
for k, v := range t.GetAllSegments() {
if v.GetInsertChannel() == channelName {
ret[k] = v
}
}

ret := make(map[int64]*datapb.SegmentInfo)
for k, v := range collectionTarget.GetAllSegments() {
if v.GetInsertChannel() == channelName {
ret[k] = v
if len(ret) > 0 {
return ret
}
}

return ret
return nil
}

func (mgr *TargetManager) GetDroppedSegmentsByChannel(collectionID int64,
Expand All @@ -436,86 +469,92 @@
mgr.rwMutex.RLock()
defer mgr.rwMutex.RUnlock()

collectionTarget := mgr.getCollectionTarget(scope, collectionID)

if collectionTarget == nil {
return nil
}

channel := collectionTarget.dmChannels[channelName]
if channel == nil {
return nil
targets := mgr.getCollectionTarget(scope, collectionID)
for _, t := range targets {
if channel, ok := t.dmChannels[channelName]; ok {
return channel.GetDroppedSegmentIds()
}
}

return channel.GetDroppedSegmentIds()
return nil

Check warning on line 479 in internal/querycoordv2/meta/target_manager.go

View check run for this annotation

Codecov / codecov/patch

internal/querycoordv2/meta/target_manager.go#L479 

Added line #L479 was not covered by tests
}

func (mgr *TargetManager) GetSealedSegmentsByPartition(collectionID int64,
partitionID int64, scope TargetScope,
partitionID int64,
scope TargetScope,
) map[int64]*datapb.SegmentInfo {
mgr.rwMutex.RLock()
defer mgr.rwMutex.RUnlock()

collectionTarget := mgr.getCollectionTarget(scope, collectionID)

if collectionTarget == nil {
return nil
}
targets := mgr.getCollectionTarget(scope, collectionID)
for _, t := range targets {
segments := make(map[int64]*datapb.SegmentInfo)
for _, s := range t.GetAllSegments() {
if s.GetPartitionID() == partitionID {
segments[s.GetID()] = s
}
}

segments := make(map[int64]*datapb.SegmentInfo)
for _, s := range collectionTarget.GetAllSegments() {
if s.GetPartitionID() == partitionID {
segments[s.GetID()] = s
if len(segments) > 0 {
return segments
}
}

return segments
return nil
}

func (mgr *TargetManager) GetDmChannelsByCollection(collectionID int64, scope TargetScope) map[string]*DmChannel {
mgr.rwMutex.RLock()
defer mgr.rwMutex.RUnlock()

collectionTarget := mgr.getCollectionTarget(scope, collectionID)
targets := mgr.getCollectionTarget(scope, collectionID)

if collectionTarget == nil {
return nil
for _, t := range targets {
return t.GetAllDmChannels()
}
return collectionTarget.GetAllDmChannels()

return nil
}

func (mgr *TargetManager) GetDmChannel(collectionID int64, channel string, scope TargetScope) *DmChannel {
mgr.rwMutex.RLock()
defer mgr.rwMutex.RUnlock()

collectionTarget := mgr.getCollectionTarget(scope, collectionID)

if collectionTarget == nil {
return nil
targets := mgr.getCollectionTarget(scope, collectionID)
for _, t := range targets {
if ch, ok := t.GetAllDmChannels()[channel]; ok {
return ch
}
}
return collectionTarget.GetAllDmChannels()[channel]
return nil
}

func (mgr *TargetManager) GetSealedSegment(collectionID int64, id int64, scope TargetScope) *datapb.SegmentInfo {
mgr.rwMutex.RLock()
defer mgr.rwMutex.RUnlock()
collectionTarget := mgr.getCollectionTarget(scope, collectionID)

if collectionTarget == nil {
return nil
targets := mgr.getCollectionTarget(scope, collectionID)
for _, t := range targets {
if s, ok := t.GetAllSegments()[id]; ok {
return s
}
}
return collectionTarget.GetAllSegments()[id]

return nil
}

func (mgr *TargetManager) GetCollectionTargetVersion(collectionID int64, scope TargetScope) int64 {
mgr.rwMutex.RLock()
defer mgr.rwMutex.RUnlock()
collectionTarget := mgr.getCollectionTarget(scope, collectionID)

if collectionTarget == nil {
return 0
targets := mgr.getCollectionTarget(scope, collectionID)
for _, t := range targets {
if t.GetTargetVersion() > 0 {
return t.GetTargetVersion()
}
}
return collectionTarget.GetTargetVersion()

return 0
}

func (mgr *TargetManager) IsCurrentTargetExist(collectionID int64) bool {
Expand Down
Loading
Loading