fix: Skip updating checkpoint after dropcollection (#29220)

Signed-off-by: yangxuan <xuan.yang@zilliz.com>

internal/datanode/flow_graph_dd_node.go

@@ -115,7 +115,7 @@ func (ddn *ddNode) Operate(in []Msg) []Msg {
 	}
 
 	if load := ddn.dropMode.Load(); load != nil && load.(bool) {
-		log.Info("ddNode in dropMode",
+		log.RatedInfo(1.0, "ddNode in dropMode",
 			zap.String("vChannelName", ddn.vChannelName),
 			zap.Int64("collectionID", ddn.collectionID))
 		return []Msg{}

internal/datanode/flow_graph_time_tick_node.go

@@ -42,6 +42,7 @@ type ttNode struct {
 	writeBufferManager writebuffer.BufferManager
 	lastUpdateTime     *atomic.Time
 	cpUpdater          *channelCheckpointUpdater
+	dropMode           *atomic.Bool
 }
 
 // Name returns node name, implementing flowgraph.Node
@@ -67,6 +68,17 @@ func (ttn *ttNode) Close() {
 // Operate handles input messages, implementing flowgraph.Node
 func (ttn *ttNode) Operate(in []Msg) []Msg {
 	fgMsg := in[0].(*flowGraphMsg)
+	if fgMsg.dropCollection {
+		ttn.dropMode.Store(true)
+	}
+
+	// skip updating checkpoint for drop collection
+	// even if its the close msg
+	if ttn.dropMode.Load() {
+		log.RatedInfo(1.0, "ttnode in dropMode", zap.String("channel", ttn.vChannelName))
+		return []Msg{}
+	}
+
 	curTs, _ := tsoutil.ParseTS(fgMsg.timeRange.timestampMax)
 	if fgMsg.IsCloseMsg() {
 		if len(fgMsg.endPositions) > 0 {
@@ -129,6 +141,7 @@ func newTTNode(config *nodeConfig, wbManager writebuffer.BufferManager, cpUpdate
 		writeBufferManager: wbManager,
 		lastUpdateTime:     atomic.NewTime(time.Time{}), // set to Zero to update channel checkpoint immediately after fg started
 		cpUpdater:          cpUpdater,
+		dropMode:           atomic.NewBool(false),
 	}
 
 	return tt, nil