fix: Clean the compaction plan info to avoid the object leak (#29365)

issue: #29296

Signed-off-by: SimFG <bang.fu@zilliz.com>

Makefile

@@ -199,6 +199,12 @@ build-3rdparty:
 	@echo "Build 3rdparty ..."
 	@(env bash $(PWD)/scripts/3rdparty_build.sh)
 
+generated-proto-without-cpp: download-milvus-proto
+	@echo "Generate proto ..."
+	@mkdir -p ${GOPATH}/bin
+	@which protoc-gen-go 1>/dev/null || (echo "Installing protoc-gen-go" && cd /tmp && go install github.com/golang/protobuf/protoc-gen-go@v1.3.2)
+	@(env bash $(PWD)/scripts/generate_proto.sh)
+
 generated-proto: download-milvus-proto build-3rdparty
 	@echo "Generate proto ..."
 	@mkdir -p ${GOPATH}/bin

internal/datacoord/compaction.go

@@ -134,14 +134,24 @@ func newCompactionPlanHandler(sessions SessionManager, cm *ChannelManager, meta
 
 func (c *compactionPlanHandler) checkResult() {
 	// deal results
+	ts, err := c.GetCurrentTS()
+	if err != nil {
+		log.Warn("fail to check result", zap.Error(err))
+		return
+	}
+	_ = c.updateCompaction(ts)
+}
+
+func (c *compactionPlanHandler) GetCurrentTS() (Timestamp, error) {
 	interval := Params.DataCoordCfg.CompactionRPCTimeout.GetAsDuration(time.Second)
 	ctx, cancel := context.WithTimeout(context.Background(), interval)
 	defer cancel()
 	ts, err := c.allocator.allocTimestamp(ctx)
 	if err != nil {
 		log.Warn("unable to alloc timestamp", zap.Error(err))
+		return 0, err
 	}
-	_ = c.updateCompaction(ts)
+	return ts, nil
 }
 
 func (c *compactionPlanHandler) schedule() {
@@ -156,7 +166,7 @@ func (c *compactionPlanHandler) schedule() {
 func (c *compactionPlanHandler) start() {
 	interval := Params.DataCoordCfg.CompactionCheckIntervalInSeconds.GetAsDuration(time.Second)
 	c.stopCh = make(chan struct{})
-	c.stopWg.Add(2)
+	c.stopWg.Add(3)
 
 	go func() {
 		defer c.stopWg.Done()
@@ -192,6 +202,37 @@ func (c *compactionPlanHandler) start() {
 			}
 		}
 	}()
+
+	go func() {
+		defer c.stopWg.Done()
+		cleanTicker := time.NewTicker(30 * time.Minute)
+		defer cleanTicker.Stop()
+		for {
+			select {
+			case <-c.stopCh:
+				log.Info("Compaction handler quit clean")
+				return
+			case <-cleanTicker.C:
+				c.Clean()
+			}
+		}
+	}()
+}
+
+func (c *compactionPlanHandler) Clean() {
+	current := tsoutil.GetCurrentTime()
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	for id, task := range c.plans {
+		if task.state == executing || task.state == pipelining {
+			continue
+		}
+		// after timeout + 1h, the plan will be cleaned
+		if c.isTimeout(current, task.plan.GetStartTime(), task.plan.GetTimeoutInSeconds()+60*60) {
+			delete(c.plans, id)
+		}
+	}
 }
 
 func (c *compactionPlanHandler) stop() {
@@ -355,9 +396,8 @@ func (c *compactionPlanHandler) completeCompaction(result *datapb.CompactionPlan
 	default:
 		return errors.New("unknown compaction type")
 	}
-	c.plans[planID] = c.plans[planID].shadowClone(setState(completed), setResult(result))
-	// TODO: when to clean task list
 	UpdateCompactionSegmentSizeMetrics(result.GetSegments())
+	c.plans[planID] = c.plans[planID].shadowClone(setState(completed), setResult(result), cleanLogPath())
 	return nil
 }
 
@@ -574,6 +614,26 @@ func setResult(result *datapb.CompactionPlanResult) compactionTaskOpt {
 	}
 }
 
+// cleanLogPath clean the log info in the compactionTask object for avoiding the memory leak
+func cleanLogPath() compactionTaskOpt {
+	return func(task *compactionTask) {
+		if task.plan.GetSegmentBinlogs() != nil {
+			for _, binlogs := range task.plan.GetSegmentBinlogs() {
+				binlogs.FieldBinlogs = nil
+				binlogs.Field2StatslogPaths = nil
+				binlogs.Deltalogs = nil
+			}
+		}
+		if task.result.GetSegments() != nil {
+			for _, segment := range task.result.GetSegments() {
+				segment.InsertLogs = nil
+				segment.Deltalogs = nil
+				segment.Field2StatslogPaths = nil
+			}
+		}
+	}
+}
+
 // 0.5*min(8, NumCPU/2)
 func calculateParallel() int {
 	// TODO after node memory management enabled, use this config as hard limit

internal/datacoord/compaction_test.go

@@ -81,8 +81,6 @@ func (s *CompactionPlanHandlerSuite) TestRemoveTasksByChannel() {
 }
 
 func (s *CompactionPlanHandlerSuite) TestCheckResult() {
-	s.mockAlloc.EXPECT().allocTimestamp(mock.Anything).Return(19530, nil)
-
 	session := &SessionManagerImpl{
 		sessions: struct {
 			sync.RWMutex
@@ -102,8 +100,48 @@ func (s *CompactionPlanHandlerSuite) TestCheckResult() {
 			},
 		},
 	}
-	handler := newCompactionPlanHandler(session, nil, nil, s.mockAlloc)
+	{
-	handler.checkResult()
+		s.mockAlloc.EXPECT().allocTimestamp(mock.Anything).Return(0, errors.New("mock")).Once()
+		handler := newCompactionPlanHandler(session, nil, nil, s.mockAlloc)
+		handler.checkResult()
+	}
+
+	{
+		s.mockAlloc.EXPECT().allocTimestamp(mock.Anything).Return(19530, nil).Once()
+		handler := newCompactionPlanHandler(session, nil, nil, s.mockAlloc)
+		handler.checkResult()
+	}
+}
+
+func (s *CompactionPlanHandlerSuite) TestClean() {
+	startTime := tsoutil.ComposeTSByTime(time.Now(), 0)
+	cleanTime := tsoutil.ComposeTSByTime(time.Now().Add(-2*time.Hour), 0)
+	c := &compactionPlanHandler{
+		allocator: s.mockAlloc,
+		plans: map[int64]*compactionTask{
+			1: {
+				state: executing,
+			},
+			2: {
+				state: pipelining,
+			},
+			3: {
+				state: completed,
+				plan: &datapb.CompactionPlan{
+					StartTime: startTime,
+				},
+			},
+			4: {
+				state: completed,
+				plan: &datapb.CompactionPlan{
+					StartTime: cleanTime,
+				},
+			},
+		},
+	}
+
+	c.Clean()
+	s.Len(c.plans, 3)
 }
 
 func (s *CompactionPlanHandlerSuite) TestHandleL0CompactionResults() {
@@ -679,6 +717,9 @@ func TestCompactionPlanHandler_completeCompaction(t *testing.T) {
 
 		err := c.completeCompaction(&compactionResult)
 		assert.NoError(t, err)
+		assert.Nil(t, compactionResult.GetSegments()[0].GetInsertLogs())
+		assert.Nil(t, compactionResult.GetSegments()[0].GetField2StatslogPaths())
+		assert.Nil(t, compactionResult.GetSegments()[0].GetDeltalogs())
 	})
 
 	t.Run("test empty result merge compaction task", func(t *testing.T) {