diff --git a/source/dnode/mgmt/mgmt_mnode/inc/mmInt.h b/source/dnode/mgmt/mgmt_mnode/inc/mmInt.h
index 9a0cfdfc9302160a21eccbe98445aad259250335..22691300bcc86a77d0ff83ed1880728ccb92bf2c 100644
--- a/source/dnode/mgmt/mgmt_mnode/inc/mmInt.h
+++ b/source/dnode/mgmt/mgmt_mnode/inc/mmInt.h
@@ -56,6 +56,7 @@ int32_t mmProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg);
 int32_t mmProcessAlterReq(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t mmProcessGetMonitorInfoReq(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t mmProcessGetLoadsReq(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t mndPreprocessQueryMsg(SMnode * pMnode, SRpcMsg * pMsg);
 
 // mmWorker.c
 int32_t mmStartWorker(SMnodeMgmt *pMgmt);
diff --git a/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c b/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
index 460f2242f2752f10959c637c2b814a3a20f2e9cd..42fa7b718e333b9e146e66fbc58c9e25f5235ea5 100644
--- a/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
+++ b/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
@@ -89,6 +89,8 @@ int32_t mmPutNodeMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
 }
 
 int32_t mmPutNodeMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
+  mndPreprocessQueryMsg(pMgmt->pMnode, pMsg);
+
   return mmPutNodeMsgToWorker(&pMgmt->queryWorker, pMsg);
 }
 
diff --git a/source/dnode/mgmt/mgmt_qnode/inc/qmInt.h b/source/dnode/mgmt/mgmt_qnode/inc/qmInt.h
index 54e9da24a4e1f581cbb7aa009e29efb9ceac221e..acc101386bd112862ea243bea89c2e3e0c867fbc 100644
--- a/source/dnode/mgmt/mgmt_qnode/inc/qmInt.h
+++ b/source/dnode/mgmt/mgmt_qnode/inc/qmInt.h
@@ -51,6 +51,8 @@ int32_t qmPutNodeMsgToQueryQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t qmPutNodeMsgToFetchQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t qmPutNodeMsgToMonitorQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg);
 
+int32_t qndPreprocessQueryMsg(SQnode *pQnode, SRpcMsg * pMsg);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c b/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c
index e36efa83db24b0d6b706518ab3c789ece80c21d5..6814643b59d9a17735a69cb3d345c97ca1b5c7d0 100644
--- a/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c
+++ b/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c
@@ -57,6 +57,8 @@ static int32_t qmPutNodeMsgToWorker(SSingleWorker *pWorker, SRpcMsg *pMsg) {
 }
 
 int32_t qmPutNodeMsgToQueryQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg) {
+  qndPreprocessQueryMsg(pMgmt->pQnode, pMsg);
+
   return qmPutNodeMsgToWorker(&pMgmt->queryWorker, pMsg);
 }
 
diff --git a/source/dnode/mnode/impl/src/mndQuery.c b/source/dnode/mnode/impl/src/mndQuery.c
index 97594f2b913334ac17e2bd5e6c8fc95e19a03e9e..12b39e5b78a10c22fee3d953fd102f62b01330a1 100644
--- a/source/dnode/mnode/impl/src/mndQuery.c
+++ b/source/dnode/mnode/impl/src/mndQuery.c
@@ -18,6 +18,14 @@
 #include "mndMnode.h"
 #include "qworker.h"
 
+int32_t mndPreprocessQueryMsg(SMnode * pMnode, SRpcMsg * pMsg) {
+  if (TDMT_VND_QUERY != pMsg->msgType) {
+    return 0;
+  }
+
+  return qWorkerPreprocessQueryMsg(pMnode->pQuery, pMsg);
+}
+
 int32_t mndProcessQueryMsg(SRpcMsg *pMsg) {
   int32_t     code = -1;
   SMnode     *pMnode = pMsg->info.node;
diff --git a/source/dnode/qnode/src/qnode.c b/source/dnode/qnode/src/qnode.c
index 438982ac6ae2ca13f2244acd978bdc58c723d6de..45b88318c474dfd3f053bb8803e92be32defcecf 100644
--- a/source/dnode/qnode/src/qnode.c
+++ b/source/dnode/qnode/src/qnode.c
@@ -63,6 +63,14 @@ int32_t qndGetLoad(SQnode *pQnode, SQnodeLoad *pLoad) {
   return 0; 
 }
 
+int32_t qndPreprocessQueryMsg(SQnode *pQnode, SRpcMsg * pMsg) {
+  if (TDMT_VND_QUERY != pMsg->msgType) {
+    return 0;
+  }
+
+  return qWorkerPreprocessQueryMsg(pQnode->pQuery, pMsg);
+}
+
 int32_t qndProcessQueryMsg(SQnode *pQnode, int64_t ts, SRpcMsg *pMsg) {
   int32_t     code = -1;
   SReadHandle handle = {.pMsgCb = &pQnode->msgCb};
diff --git a/source/libs/qworker/src/qwDbg.c b/source/libs/qworker/src/qwDbg.c
index 27fe22295d3706eb21a237f8d662e34b4dce9b36..3914541157e86aa67a55c56cd56afb76da2061cf 100644
--- a/source/libs/qworker/src/qwDbg.c
+++ b/source/libs/qworker/src/qwDbg.c
@@ -36,7 +36,8 @@ int32_t qwDbgValidateStatus(QW_FPARAMS_DEF, int8_t oriStatus, int8_t newStatus,
 
       break;
     case JOB_TASK_STATUS_NOT_START:
-      if (newStatus != JOB_TASK_STATUS_CANCELLED) {
+      if (newStatus != JOB_TASK_STATUS_DROPPING && newStatus != JOB_TASK_STATUS_EXECUTING
+        && newStatus != JOB_TASK_STATUS_FAILED) {
         QW_ERR_JRET(TSDB_CODE_QRY_APP_ERROR);
       }
 
diff --git a/source/libs/qworker/src/qwMsg.c b/source/libs/qworker/src/qwMsg.c
index 317877c2532cdc3f0c4beb9517a953320e621388..2f1ecc51723a4799071292f5d7eb8d3b3d2f65ab 100644
--- a/source/libs/qworker/src/qwMsg.c
+++ b/source/libs/qworker/src/qwMsg.c
@@ -300,12 +300,12 @@ int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, int
     QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
   }
 
-  msg->sId = be64toh(msg->sId);
-  msg->queryId = be64toh(msg->queryId);
-  msg->taskId = be64toh(msg->taskId);
-  msg->refId = be64toh(msg->refId);
-  msg->phyLen = ntohl(msg->phyLen);
-  msg->sqlLen = ntohl(msg->sqlLen);
+  msg->sId = msg->sId;
+  msg->queryId = msg->queryId;
+  msg->taskId = msg->taskId;
+  msg->refId = msg->refId;
+  msg->phyLen = msg->phyLen;
+  msg->sqlLen = msg->sqlLen;
 
   uint64_t sId = msg->sId;
   uint64_t qId = msg->queryId;
diff --git a/source/libs/qworker/src/qworker.c b/source/libs/qworker/src/qworker.c
index 9ad36a32c4716fc44091de4686fdaf10162dc8eb..8c0366fa0be1150b595fb9bec71035e7f2fe58ed 100644
--- a/source/libs/qworker/src/qworker.c
+++ b/source/libs/qworker/src/qworker.c
@@ -446,6 +446,8 @@ int32_t qwPrerocessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
 
   QW_ERR_JRET(qwAddAcquireTaskCtx(QW_FPARAMS(), &ctx));
 
+  ctx->ctrlConnInfo = qwMsg->connInfo;
+
   QW_ERR_JRET(qwAddTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_NOT_START));
 
 _return:
@@ -475,8 +477,6 @@ int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, int8_t taskType, int8_t ex
   atomic_store_8(&ctx->taskType, taskType);
   atomic_store_8(&ctx->explain, explain);
 
-  ctx->ctrlConnInfo = qwMsg->connInfo;
-
   /*QW_TASK_DLOGL("subplan json string, len:%d, %s", qwMsg->msgLen, qwMsg->msg);*/
 
   code = qStringToSubplan(qwMsg->msg, &plan);
diff --git a/source/libs/scheduler/src/schRemote.c b/source/libs/scheduler/src/schRemote.c
index a940e45b43d0baa364490d93f7f168eb7fc79b0d..88caf7438e069b77c664dff05d466eeb5991fb34 100644
--- a/source/libs/scheduler/src/schRemote.c
+++ b/source/libs/scheduler/src/schRemote.c
@@ -774,13 +774,20 @@ int32_t schEnsureHbConnection(SSchJob *pJob, SSchTask *pTask) {
   strcpy(epId.ep.fqdn, pEp->fqdn);
   epId.ep.port = pEp->port;
 
-  SSchHbTrans *hb = taosHashGet(schMgmt.hbConnections, &epId, sizeof(SQueryNodeEpId));
-  if (NULL == hb) {
-    bool exist = false;
-    SCH_ERR_RET(schRegisterHbConnection(pJob, pTask, &epId, &exist));
-    if (!exist) {
-      SCH_ERR_RET(schBuildAndSendHbMsg(&epId, NULL));
+  SSchHbTrans *hb = NULL;
+  while (true) {
+    hb = taosHashGet(schMgmt.hbConnections, &epId, sizeof(SQueryNodeEpId));
+    if (NULL == hb) {
+      bool exist = false;
+      SCH_ERR_RET(schRegisterHbConnection(pJob, pTask, &epId, &exist));
+      if (!exist) {
+        SCH_ERR_RET(schBuildAndSendHbMsg(&epId, NULL));
+      }
+
+      continue;
     }
+
+    break;
   }
 
   atomic_add_fetch_64(&hb->taskNum, 1);