admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2025-12-06 17:18:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: Release data memory after sync task completes (#42627)

Browse Source 
Release data memory after sync task completes to prevent datanode oom
during import.

issue: https://github.com/milvus-io/milvus/issues/42608

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
This commit is contained in:
yihao.dai 2025-06-10 16:28:34 +08:00 committed by GitHub
parent c9680a5b56
commit ed55b14484
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 18 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
internal/flushcommon/syncmgr/serializer.go
View File

@ -143,3 +143,9 @@ func (p *SyncPack) WithDataSource(source string) *SyncPack {
p.dataSource = source p.dataSource = source
return p return p
} }
func (p *SyncPack) ReleaseData() {
p.insertData = nil
p.deltaData = nil
p.bm25Stats = nil
}

2
internal/flushcommon/syncmgr/task.go
View File

@ -168,6 +168,8 @@ func (t *SyncTask) Run(ctx context.Context) (err error) {
} }
} }
t.pack.ReleaseData()
actions := []metacache.SegmentAction{metacache.FinishSyncing(t.batchRows)} actions := []metacache.SegmentAction{metacache.FinishSyncing(t.batchRows)}
if t.pack.isFlush { if t.pack.isFlush {
actions = append(actions, metacache.UpdateState(commonpb.SegmentState_Flushed)) actions = append(actions, metacache.UpdateState(commonpb.SegmentState_Flushed))

10
internal/flushcommon/syncmgr/task_test.go
View File

@ -210,6 +210,12 @@ func (s *SyncTaskSuite) runTestRunNormal(storageVersion int64) {
action(seg) action(seg)
}).Return() }).Return()
isDataReleased := func(task *SyncTask) bool {
return task.pack.insertData == nil &&
task.pack.deltaData == nil &&
task.pack.bm25Stats == nil
}
s.Run("without_data", func() { s.Run("without_data", func() {
task := s.getSuiteSyncTask(new(SyncPack).WithCheckpoint( task := s.getSuiteSyncTask(new(SyncPack).WithCheckpoint(
&msgpb.MsgPosition{ &msgpb.MsgPosition{
@ -225,6 +231,7 @@ func (s *SyncTaskSuite) runTestRunNormal(storageVersion int64) {
err := task.Run(ctx) err := task.Run(ctx)
s.NoError(err) s.NoError(err)
s.True(isDataReleased(task)) // data should be released after task finished
}) })
s.Run("with_insert_delete_cp", func() { s.Run("with_insert_delete_cp", func() {
@ -244,6 +251,7 @@ func (s *SyncTaskSuite) runTestRunNormal(storageVersion int64) {
err := task.Run(ctx) err := task.Run(ctx)
s.NoError(err) s.NoError(err)
s.True(isDataReleased(task)) // data should be released after task finished
}) })
s.Run("with_flush", func() { s.Run("with_flush", func() {
@ -263,6 +271,7 @@ func (s *SyncTaskSuite) runTestRunNormal(storageVersion int64) {
} }
err := task.Run(ctx) err := task.Run(ctx)
s.NoError(err) s.NoError(err)
s.True(isDataReleased(task)) // data should be released after task finished
}) })
s.Run("with_drop", func() { s.Run("with_drop", func() {
@ -282,6 +291,7 @@ func (s *SyncTaskSuite) runTestRunNormal(storageVersion int64) {
} }
err := task.Run(ctx) err := task.Run(ctx)
s.NoError(err) s.NoError(err)
s.True(isDataReleased(task)) // data should be released after task finished
}) })
} }