enhance: Avoid redundant meta operations of import (milvus-io#33518) (m…

…ilvus-io#33519)

issue: milvus-io#33513

pr: milvus-io#33518

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>

internal/datacoord/import_checker.go

@@ -341,6 +341,9 @@ func (c *importChecker) checkCollection(collectionID int64, jobs []ImportJob) {
 		return
 	}
 	if !has {
+		jobs = lo.Filter(jobs, func(job ImportJob, _ int) bool {
+			return job.GetState() != internalpb.ImportJobState_Failed
+		})
 		for _, job := range jobs {
 			err = c.imeta.UpdateJob(job.GetJobID(), UpdateJobState(internalpb.ImportJobState_Failed),
 				UpdateJobReason(fmt.Sprintf("collection %d dropped", collectionID)))
@@ -388,6 +391,8 @@ func (c *importChecker) checkGC(job ImportJob) {
 		err := c.imeta.RemoveJob(job.GetJobID())
 		if err != nil {
 			log.Warn("remove import job failed", zap.Int64("jobID", job.GetJobID()), zap.Error(err))
+			return
 		}
+		log.Info("import job removed", zap.Int64("jobID", job.GetJobID()))
 	}
 }

internal/datacoord/import_scheduler.go

@@ -350,9 +350,11 @@ func (s *importScheduler) processFailed(task ImportTask) {
 				return
 			}
 		}
-		err := s.imeta.UpdateTask(task.GetTaskID(), UpdateSegmentIDs(nil))
-		if err != nil {
-			log.Warn("update import task segments failed", WrapTaskLog(task, zap.Error(err))...)
+		if len(segments) > 0 {
+			err := s.imeta.UpdateTask(task.GetTaskID(), UpdateSegmentIDs(nil))
+			if err != nil {
+				log.Warn("update import task segments failed", WrapTaskLog(task, zap.Error(err))...)
+			}
 		}
 	}
 	err := DropImportTask(task, s.cluster, s.imeta)