enhance: enable compaction dynamicly

internal/datacoord/compaction.go

@@ -53,8 +53,6 @@
 	execCompactionPlan(signal *compactionSignal, plan *datapb.CompactionPlan)
 	// getCompaction return compaction task. If planId does not exist, return nil.
 	getCompaction(planID int64) *compactionTask
-	// updateCompaction set the compaction state to timeout or completed
-	updateCompaction(ts Timestamp) error
 	// isFull return true if the task pool is full
 	isFull() bool
 	// get compaction tasks by signal id
@@ -140,41 +138,6 @@
 	}
 }
 
-func (c *compactionPlanHandler) checkResult() {
-	// deal results
-	ts, err := c.GetCurrentTS()
-	if err != nil {
-		log.Warn("fail to check result", zap.Error(err))
-		return
-	}
-	err = c.updateCompaction(ts)
-	if err != nil {
-		log.Warn("fail to update compaction", zap.Error(err))
-		return
-	}
-}
-
-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
-	}
-	return ts, nil
-}
-
-func (c *compactionPlanHandler) schedule() {
-	// schedule queuing tasks
-	tasks := c.scheduler.Schedule()
-	if len(tasks) > 0 {
-		c.notifyTasks(tasks)
-		c.scheduler.LogStatus()
-	}
-}
-
 func (c *compactionPlanHandler) start() {
 	interval := Params.DataCoordCfg.CompactionCheckIntervalInSeconds.GetAsDuration(time.Second)
 	c.stopCh = make(chan struct{})
@@ -191,7 +154,9 @@
 				log.Info("compaction handler check result loop quit")
 				return
 			case <-checkResultTicker.C:
-				c.checkResult()
+				if compactionEnabled() {
+					c.CheckResult()
+				}
 			}
 		}
 	}()
@@ -210,7 +175,9 @@
 				return
 
 			case <-scheduleTicker.C:
-				c.schedule()
+				if compactionEnabled() {
+					c.Schedule()
+				}
 			}
 		}
 	}()
@@ -225,12 +192,49 @@
 				log.Info("Compaction handler quit clean")
 				return
 			case <-cleanTicker.C:
-				c.Clean()
+				if compactionEnabled() {
+					c.Clean()
+				}
 			}
 		}
 	}()
 }
 
+func (c *compactionPlanHandler) CheckResult() {
+	// deal results
+	ts, err := c.getCurrentTs()
+	if err != nil {
+		log.Warn("fail to check result", zap.Error(err))
+		return
+	}
+	err = c.updateCompaction(ts)
+	if err != nil {
+		log.Warn("fail to update compaction", zap.Error(err))
+		return
+	}
+}
+
+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
+	}
+	return ts, nil
+}
+
+func (c *compactionPlanHandler) Schedule() {
+	// schedule queuing tasks
+	tasks := c.scheduler.Schedule()
+	if len(tasks) > 0 {
+		c.notifyTasks(tasks)
+		c.scheduler.LogStatus()
+	}
+}
+
 func (c *compactionPlanHandler) Clean() {
 	current := tsoutil.GetCurrentTime()
 	c.mu.Lock()
@@ -255,6 +259,10 @@
 }
 
 func (c *compactionPlanHandler) removeTasksByChannel(channel string) {
+	if !compactionEnabled() {
+		return
+	}
+
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	for id, task := range c.plans {
@@ -405,7 +413,9 @@
 
 // execCompactionPlan start to execute plan and return immediately
 func (c *compactionPlanHandler) execCompactionPlan(signal *compactionSignal, plan *datapb.CompactionPlan) {
-	c.enqueuePlan(signal, plan)
+	if compactionEnabled() {
+		c.enqueuePlan(signal, plan)
+	}
 }
 
 func (c *compactionPlanHandler) setSegmentsCompacting(plan *datapb.CompactionPlan, compacting bool) {
@@ -506,12 +516,17 @@
 
 // getCompaction return compaction task. If planId does not exist, return nil.
 func (c *compactionPlanHandler) getCompaction(planID int64) *compactionTask {
+	if !compactionEnabled() {
+		return nil
+	}
+
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 
 	return c.plans[planID]
 }
 
+// updateCompaction is the inner func to check compaction task states from datanode
 func (c *compactionPlanHandler) updateCompaction(ts Timestamp) error {
 	// Get executing executingTasks before GetCompactionState from DataNode to prevent false failure,
 	//  for DC might add new task while GetCompactionState.
@@ -641,6 +656,9 @@
 
 // isFull return true if the task pool is full
 func (c *compactionPlanHandler) isFull() bool {
+	if !compactionEnabled() {
+		return true
+	}
 	return c.scheduler.GetTaskCount() >= Params.DataCoordCfg.CompactionMaxParallelTasks.GetAsInt()
 }
 
@@ -658,6 +676,10 @@
 
 // get compaction tasks by signal id; if signalID == 0 return all tasks
 func (c *compactionPlanHandler) getCompactionTasksBySignalID(signalID int64) []*compactionTask {
+	if !compactionEnabled() {
+		return nil
+	}
+
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 

internal/datacoord/compaction_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/milvus-io/milvus/internal/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/util/merr"
 	"github.com/milvus-io/milvus/pkg/util/metautil"
+	"github.com/milvus-io/milvus/pkg/util/paramtable"
 	"github.com/milvus-io/milvus/pkg/util/tsoutil"
 	"github.com/milvus-io/milvus/pkg/util/typeutil"
 )
@@ -74,6 +75,13 @@ func (s *CompactionPlanHandlerSuite) TestRemoveTasksByChannel() {
 	handler.mu.Lock()
 	s.Equal(0, len(handler.plans))
 	handler.mu.Unlock()
+
+	s.Run("disable compaction", func() {
+		paramtable.Get().Save(paramtable.Get().DataCoordCfg.EnableCompaction.Key, "false")
+		defer paramtable.Get().Reset(paramtable.Get().DataCoordCfg.EnableCompaction.Key)
+
+		handler.removeTasksByChannel(ch)
+	})
 }
 
 func (s *CompactionPlanHandlerSuite) TestCheckResult() {
@@ -88,13 +96,13 @@ func (s *CompactionPlanHandlerSuite) TestCheckResult() {
 	{
 		s.mockAlloc.EXPECT().allocTimestamp(mock.Anything).Return(0, errors.New("mock")).Once()
 		handler := newCompactionPlanHandler(nil, s.mockSessMgr, nil, nil, s.mockAlloc)
-		handler.checkResult()
+		handler.CheckResult()
 	}
 
 	{
 		s.mockAlloc.EXPECT().allocTimestamp(mock.Anything).Return(19530, nil).Once()
 		handler := newCompactionPlanHandler(nil, s.mockSessMgr, nil, nil, s.mockAlloc)
-		handler.checkResult()
+		handler.CheckResult()
 	}
 }
 
@@ -440,6 +448,16 @@ func (s *CompactionPlanHandlerSuite) TestExecCompactionPlan() {
 	plan := &datapb.CompactionPlan{
 		PlanID: int64(1),
 	}
+
+	s.Run("disable compaction", func() {
+		paramtable.Get().Save(paramtable.Get().DataCoordCfg.EnableCompaction.Key, "false")
+		defer paramtable.Get().Reset(paramtable.Get().DataCoordCfg.EnableCompaction.Key)
+
+		s.NotPanics(func() {
+			handler.execCompactionPlan(&compactionSignal{id: 333}, &datapb.CompactionPlan{PlanID: 333, Channel: "ch-2"})
+		})
+	})
+
 	plan.Channel = "ch-1"
 
 	handler.execCompactionPlan(sig, plan)
@@ -658,6 +676,17 @@ func (s *CompactionPlanHandlerSuite) TestGetCompactionTask() {
 
 	task = handler.getCompaction(19530)
 	s.Nil(task)
+
+	s.Run("disable compaction", func() {
+		paramtable.Get().Save(paramtable.Get().DataCoordCfg.EnableCompaction.Key, "false")
+		defer paramtable.Get().Reset(paramtable.Get().DataCoordCfg.EnableCompaction.Key)
+
+		got := handler.getCompactionTasksBySignalID(1)
+		s.Nil(got)
+
+		task := handler.getCompaction(1)
+		s.Nil(task)
+	})
 }
 
 func (s *CompactionPlanHandlerSuite) TestUpdateCompaction() {