update meta sync in query service (#5937)

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

internal/proxynode/task.go

@@ -1235,8 +1235,9 @@ func (st *SearchTask) PreExecute(ctx context.Context) error {
 		return errors.New(showResp.Status.Reason)
 	}
 	log.Debug("query service show collections",
+		zap.Any("collID", collID),
 		zap.Any("collections", showResp.CollectionIDs),
-		zap.Any("collID", collID))
+	)
 	collectionLoaded := false
 	for _, collectionID := range showResp.CollectionIDs {
 		if collectionID == collID {

internal/queryservice/impl.go

@@ -160,11 +160,11 @@ func (qs *QueryService) LoadCollection(ctx context.Context, req *querypb.LoadCol
 	}
 	qs.scheduler.Enqueue([]task{loadCollectionTask})
 
-	//err := loadCollectionTask.WaitToFinish()
-	//if err != nil {
-	//	status.Reason = err.Error()
-	//	return status, err
-	//}
+	err := loadCollectionTask.WaitToFinish()
+	if err != nil {
+		status.Reason = err.Error()
+		return status, err
+	}
 	//qs.meta.setLoadCollection(collectionID, true)
 
 	log.Debug("LoadCollectionRequest completed", zap.String("role", Params.RoleName), zap.Int64("msgID", req.Base.MsgID), zap.Int64("collectionID", collectionID))
@@ -275,12 +275,12 @@ func (qs *QueryService) LoadPartitions(ctx context.Context, req *querypb.LoadPar
 		}
 		qs.scheduler.Enqueue([]task{loadPartitionTask})
 
-		//err := loadPartitionTask.WaitToFinish()
-		//if err != nil {
-		//	status.ErrorCode = commonpb.ErrorCode_UnexpectedError
-		//	status.Reason = err.Error()
-		//	return status, err
-		//}
+		err := loadPartitionTask.WaitToFinish()
+		if err != nil {
+			status.ErrorCode = commonpb.ErrorCode_UnexpectedError
+			status.Reason = err.Error()
+			return status, err
+		}
 	}
 
 	status.ErrorCode = commonpb.ErrorCode_Success

internal/queryservice/meta.go

@@ -188,6 +188,9 @@ func (m *meta) addCollection(collectionID UniqueID, schema *schemapb.CollectionS
 		if err != nil {
 			log.Error("save collectionInfo error", zap.Any("error", err.Error()), zap.Int64("collectionID", collectionID))
 		}
+		log.Debug("add collection",
+			zap.Any("collectionID", collectionID),
+		)
 		return nil
 	}
 

internal/queryservice/task_scheduler.go

@@ -518,6 +518,9 @@ func (scheduler *TaskScheduler) scheduleLoop() {
 					log.Error("scheduleLoop: process task error", zap.Any("error", err.Error()))
 					continue
 				}
+				if t.Type() == commonpb.MsgType_LoadCollection || t.Type() == commonpb.MsgType_LoadPartitions {
+					t.Notify(err)
+				}
 			}
 			log.Debug("scheduleLoop: num of child task", zap.Int("num child task", len(t.GetChildTask())))
 			for _, childTask := range t.GetChildTask() {