From 361c54eba9e7fa7bbf0e8adf2a5d5fe974f9b6d2 Mon Sep 17 00:00:00 2001 From: jiangxinmeng1 Date: Tue, 21 May 2024 19:05:36 +0800 Subject: [PATCH] add observer in factory --- pkg/vm/engine/tae/db/merge/executor.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/pkg/vm/engine/tae/db/merge/executor.go b/pkg/vm/engine/tae/db/merge/executor.go index 05420d9df2b2..5251a135ba7d 100644 --- a/pkg/vm/engine/tae/db/merge/executor.go +++ b/pkg/vm/engine/tae/db/merge/executor.go @@ -191,7 +191,9 @@ func (e *MergeExecutor) ExecuteFor(entry *catalog.TableEntry, policy Policy) { } factory := func(ctx *tasks.Context, txn txnif.AsyncTxn) (tasks.Task, error) { - return jobs.NewMergeObjectsTask(ctx, txn, mobjs, e.rt, common.DefaultMaxOsizeObjMB*common.Const1MBytes) + task, err := jobs.NewMergeObjectsTask(ctx, txn, mobjs, e.rt, common.DefaultMaxOsizeObjMB*common.Const1MBytes) + task.AddObserver(e) + return task, err } task, err := e.rt.Scheduler.ScheduleMultiScopedTxnTask(nil, tasks.DataCompactionTask, scopes, factory) if err != nil { @@ -201,7 +203,6 @@ func (e *MergeExecutor) ExecuteFor(entry *catalog.TableEntry, policy Policy) { return } e.AddActiveTask(task.ID(), blkCnt, esize) - task.AddObserver(e) logMergeTask(e.tableName, task.ID(), mobjs, blkCnt, osize, esize) }