diff --git a/include/libs/qcom/query.h b/include/libs/qcom/query.h
index daf008108b..f4ccc05208 100644
--- a/include/libs/qcom/query.h
+++ b/include/libs/qcom/query.h
@@ -182,8 +182,10 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
 #define SET_META_TYPE_TABLE(t)      (t) = META_TYPE_TABLE
 #define SET_META_TYPE_BOTH_TABLE(t) (t) = META_TYPE_BOTH_TABLE
 
-#define NEED_CLIENT_RM_TBLMETA_ERROR(_code) \
-  ((_code) == TSDB_CODE_PAR_TABLE_NOT_EXIST || (_code) == TSDB_CODE_VND_TB_NOT_EXIST)
+#define NEED_CLIENT_RM_TBLMETA_ERROR(_code)                                                   \
+  ((_code) == TSDB_CODE_PAR_TABLE_NOT_EXIST || (_code) == TSDB_CODE_VND_TB_NOT_EXIST ||       \
+   (_code) == TSDB_CODE_PAR_INVALID_COLUMNS_NUM || (_code) == TSDB_CODE_PAR_INVALID_COLUMN || \
+   (_code) == TSDB_CODE_PAR_TAGS_NOT_MATCHED)
 #define NEED_CLIENT_REFRESH_VG_ERROR(_code) \
   ((_code) == TSDB_CODE_VND_HASH_MISMATCH || (_code) == TSDB_CODE_VND_INVALID_VGROUP_ID)
 #define NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code) ((_code) == TSDB_CODE_TDB_TABLE_RECREATED)
@@ -194,7 +196,7 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
 #define NEED_SCHEDULER_RETRY_ERROR(_code) \
   ((_code) == TSDB_CODE_RPC_REDIRECT || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL)
 
-#define REQUEST_MAX_TRY_TIMES 5
+#define REQUEST_MAX_TRY_TIMES 1
 
 #define qFatal(...)                                                                           \
   do {                                                                                        \
diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c
index 1dda4c3024..f493f02cd6 100644
--- a/source/client/src/clientImpl.c
+++ b/source/client/src/clientImpl.c
@@ -13,18 +13,18 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "cJSON.h"
 #include "clientInt.h"
 #include "clientLog.h"
 #include "command.h"
 #include "scheduler.h"
 #include "tdatablock.h"
+#include "tdataformat.h"
 #include "tdef.h"
 #include "tglobal.h"
 #include "tmsgtype.h"
 #include "tpagedbuf.h"
 #include "tref.h"
-#include "cJSON.h"
-#include "tdataformat.h"
 
 static int32_t       initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet);
 static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest);
@@ -189,7 +189,8 @@ int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtC
       setResSchemaInfo(&pRequest->body.resInfo, (*pQuery)->pResSchema, (*pQuery)->numOfResCols);
       setResPrecision(&pRequest->body.resInfo, (*pQuery)->precision);
     }
-
+  }
+  if (TSDB_CODE_SUCCESS == code || NEED_CLIENT_HANDLE_ERROR(code)) {
     TSWAP(pRequest->dbList, (*pQuery)->pDbList);
     TSWAP(pRequest->tableList, (*pQuery)->pTableList);
   }
@@ -293,7 +294,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
 
   SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf};
   int32_t      code = schedulerExecJob(pTransporter, pNodeList, pDag, &pRequest->body.queryJob, pRequest->sqlstr,
-                                  pRequest->metric.start, &res);
+                                       pRequest->metric.start, &res);
   if (code != TSDB_CODE_SUCCESS) {
     if (pRequest->body.queryJob != 0) {
       schedulerFreeJob(pRequest->body.queryJob);
@@ -483,7 +484,8 @@ SRequestObj* execQuery(STscObj* pTscObj, const char* sql, int sqlLen) {
   int32_t      retryNum = 0;
   int32_t      code = 0;
 
-  while (retryNum++ < REQUEST_MAX_TRY_TIMES) {
+  do {
+    destroyRequest(pRequest);
     pRequest = launchQuery(pTscObj, sql, sqlLen);
     if (pRequest == NULL || TSDB_CODE_SUCCESS == pRequest->code || !NEED_CLIENT_HANDLE_ERROR(pRequest->code)) {
       break;
@@ -494,9 +496,7 @@ SRequestObj* execQuery(STscObj* pTscObj, const char* sql, int sqlLen) {
       pRequest->code = code;
       break;
     }
-
-    destroyRequest(pRequest);
-  }
+  } while (retryNum++ < REQUEST_MAX_TRY_TIMES);
 
   return pRequest;
 }
@@ -805,21 +805,20 @@ static int32_t doPrepareResPtr(SReqResultInfo* pResInfo) {
   return TSDB_CODE_SUCCESS;
 }
 
-static char* parseTagDatatoJson(void *p){
-  char* string = NULL;
-  cJSON *json = cJSON_CreateObject();
-  if (json == NULL)
-  {
+static char* parseTagDatatoJson(void* p) {
+  char*  string = NULL;
+  cJSON* json = cJSON_CreateObject();
+  if (json == NULL) {
     goto end;
   }
 
   int16_t nCols = kvRowNCols(p);
-  char tagJsonKey[256] = {0};
+  char    tagJsonKey[256] = {0};
   for (int j = 0; j < nCols; ++j) {
-    SColIdx * pColIdx = kvRowColIdxAt(p, j);
-    char* val = (char*)(kvRowColVal(p, pColIdx));
-    if (j == 0){
-      if(*val == TSDB_DATA_TYPE_NULL){
+    SColIdx* pColIdx = kvRowColIdxAt(p, j);
+    char*    val = (char*)(kvRowColVal(p, pColIdx));
+    if (j == 0) {
+      if (*val == TSDB_DATA_TYPE_NULL) {
         string = taosMemoryCalloc(1, 8);
         sprintf(varDataVal(string), "%s", TSDB_DATA_NULL_STR_L);
         varDataSetLen(string, strlen(varDataVal(string)));
@@ -834,19 +833,18 @@ static char* parseTagDatatoJson(void *p){
     // json value
     val += varDataTLen(val);
     char* realData = POINTER_SHIFT(val, CHAR_BYTES);
-    char type = *val;
-    if(type == TSDB_DATA_TYPE_NULL) {
+    char  type = *val;
+    if (type == TSDB_DATA_TYPE_NULL) {
       cJSON* value = cJSON_CreateNull();
-      if (value == NULL)
-      {
+      if (value == NULL) {
         goto end;
       }
       cJSON_AddItemToObject(json, tagJsonKey, value);
-    }else if(type == TSDB_DATA_TYPE_NCHAR) {
+    } else if (type == TSDB_DATA_TYPE_NCHAR) {
       cJSON* value = NULL;
-      if (varDataLen(realData) > 0){
-        char *tagJsonValue = taosMemoryCalloc(varDataLen(realData), 1);
-        int32_t length = taosUcs4ToMbs((TdUcs4 *)varDataVal(realData), varDataLen(realData), tagJsonValue);
+      if (varDataLen(realData) > 0) {
+        char*   tagJsonValue = taosMemoryCalloc(varDataLen(realData), 1);
+        int32_t length = taosUcs4ToMbs((TdUcs4*)varDataVal(realData), varDataLen(realData), tagJsonValue);
         if (length < 0) {
           tscError("charset:%s to %s. val:%s convert json value failed.", DEFAULT_UNICODE_ENCODEC, tsCharset, val);
           taosMemoryFree(tagJsonValue);
@@ -854,45 +852,41 @@ static char* parseTagDatatoJson(void *p){
         }
         value = cJSON_CreateString(tagJsonValue);
         taosMemoryFree(tagJsonValue);
-        if (value == NULL)
-        {
+        if (value == NULL) {
           goto end;
         }
-      }else if(varDataLen(realData) == 0){
+      } else if (varDataLen(realData) == 0) {
         value = cJSON_CreateString("");
-      }else{
+      } else {
         ASSERT(0);
       }
 
       cJSON_AddItemToObject(json, tagJsonKey, value);
-    }else if(type == TSDB_DATA_TYPE_DOUBLE){
+    } else if (type == TSDB_DATA_TYPE_DOUBLE) {
       double jsonVd = *(double*)(realData);
       cJSON* value = cJSON_CreateNumber(jsonVd);
-      if (value == NULL)
-      {
+      if (value == NULL) {
         goto end;
       }
       cJSON_AddItemToObject(json, tagJsonKey, value);
-//    }else if(type == TSDB_DATA_TYPE_BIGINT){
-//      int64_t jsonVd = *(int64_t*)(realData);
-//      cJSON* value = cJSON_CreateNumber((double)jsonVd);
-//      if (value == NULL)
-//      {
-//        goto end;
-//      }
-//      cJSON_AddItemToObject(json, tagJsonKey, value);
-    }else if (type == TSDB_DATA_TYPE_BOOL) {
-      char jsonVd = *(char*)(realData);
+      //    }else if(type == TSDB_DATA_TYPE_BIGINT){
+      //      int64_t jsonVd = *(int64_t*)(realData);
+      //      cJSON* value = cJSON_CreateNumber((double)jsonVd);
+      //      if (value == NULL)
+      //      {
+      //        goto end;
+      //      }
+      //      cJSON_AddItemToObject(json, tagJsonKey, value);
+    } else if (type == TSDB_DATA_TYPE_BOOL) {
+      char   jsonVd = *(char*)(realData);
       cJSON* value = cJSON_CreateBool(jsonVd);
-      if (value == NULL)
-      {
+      if (value == NULL) {
         goto end;
       }
       cJSON_AddItemToObject(json, tagJsonKey, value);
-    }else{
+    } else {
       ASSERT(0);
     }
-
   }
   string = cJSON_PrintUnformatted(json);
 end:
@@ -930,7 +924,7 @@ static int32_t doConvertUCS4(SReqResultInfo* pResultInfo, int32_t numOfRows, int
 
       pResultInfo->pCol[i].pData = pResultInfo->convertBuf[i];
       pResultInfo->row[i] = pResultInfo->pCol[i].pData;
-    }else if (type == TSDB_DATA_TYPE_JSON && colLength[i] > 0) {
+    } else if (type == TSDB_DATA_TYPE_JSON && colLength[i] > 0) {
       char* p = taosMemoryRealloc(pResultInfo->convertBuf[i], colLength[i]);
       if (p == NULL) {
         return TSDB_CODE_OUT_OF_MEMORY;
@@ -943,7 +937,6 @@ static int32_t doConvertUCS4(SReqResultInfo* pResultInfo, int32_t numOfRows, int
         if (pCol->offset[j] != -1) {
           char* pStart = pCol->offset[j] + pCol->pData;
 
-
           int32_t jsonInnerType = *pStart;
           char*   jsonInnerData = pStart + CHAR_BYTES;
           char    dst[TSDB_MAX_JSON_TAG_LEN] = {0};
@@ -951,7 +944,7 @@ static int32_t doConvertUCS4(SReqResultInfo* pResultInfo, int32_t numOfRows, int
             sprintf(varDataVal(dst), "%s", TSDB_DATA_NULL_STR_L);
             varDataSetLen(dst, strlen(varDataVal(dst)));
           } else if (jsonInnerType == TSDB_DATA_TYPE_JSON) {
-            char *jsonString = parseTagDatatoJson(jsonInnerData);
+            char* jsonString = parseTagDatatoJson(jsonInnerData);
             STR_TO_VARSTR(dst, jsonString);
             taosMemoryFree(jsonString);
           } else if (jsonInnerType == TSDB_DATA_TYPE_NCHAR) {  // value -> "value"
diff --git a/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c b/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
index cf76d3a167..c4314a57b1 100644
--- a/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
+++ b/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
@@ -126,7 +126,9 @@ int32_t mmPutRpcMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   return mmPutRpcMsgToWorker(&pMgmt->readWorker, pMsg);
 }
 
-int32_t mmPutRpcMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { return mmPutRpcMsgToWorker(&pMgmt->syncWorker, pMsg); }
+int32_t mmPutRpcMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
+  return mmPutRpcMsgToWorker(&pMgmt->syncWorker, pMsg);
+}
 
 int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
   SSingleWorkerCfg qCfg = {
diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
index 9b60260cc1..3a5e8a671c 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
@@ -70,11 +70,10 @@ static void vmProcessQueryQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
   if (code != 0) {
     if (terrno != 0) code = terrno;
     vmSendRsp(pMsg, code);
-
-    dTrace("msg:%p, is freed, code:0x%x", pMsg, code);
-    rpcFreeCont(pMsg->pCont);
-    taosFreeQitem(pMsg);
   }
+  dTrace("msg:%p, is freed, code:0x%x", pMsg, code);
+  rpcFreeCont(pMsg->pCont);
+  taosFreeQitem(pMsg);
 }
 
 static void vmProcessFetchQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
@@ -85,16 +84,15 @@ static void vmProcessFetchQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
   if (code != 0) {
     if (terrno != 0) code = terrno;
     vmSendRsp(pMsg, code);
-
-    dTrace("msg:%p, is freed, code:0x%x", pMsg, code);
-    rpcFreeCont(pMsg->pCont);
-    taosFreeQitem(pMsg);
   }
+  dTrace("msg:%p, is freed, code:0x%x", pMsg, code);
+  rpcFreeCont(pMsg->pCont);
+  taosFreeQitem(pMsg);
 }
 
 static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
-  SArray *pArray = taosArrayInit(numOfMsgs, sizeof(SRpcMsg *));
+  SArray *   pArray = taosArrayInit(numOfMsgs, sizeof(SRpcMsg *));
   if (pArray == NULL) {
     dError("failed to process %d msgs in write-queue since %s", numOfMsgs, terrstr());
     return;
@@ -216,16 +214,15 @@ static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
     if (code != 0) {
       if (terrno != 0) code = terrno;
       vmSendRsp(pMsg, code);
-
-      dTrace("msg:%p, is freed, code:0x%x", pMsg, code);
-      rpcFreeCont(pMsg->pCont);
-      taosFreeQitem(pMsg);
     }
+    dTrace("msg:%p, is freed, code:0x%x", pMsg, code);
+    rpcFreeCont(pMsg->pCont);
+    taosFreeQitem(pMsg);
   }
 }
 
 static int32_t vmPutNodeMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtype) {
-  SRpcMsg  *pRpc = pMsg;
+  SRpcMsg * pRpc = pMsg;
   SMsgHead *pHead = pRpc->pCont;
   int32_t   code = 0;
 
@@ -304,7 +301,7 @@ int32_t vmPutNodeMsgToMonitorQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
 }
 
 static int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc, EQueueType qtype) {
-  SMsgHead  *pHead = pRpc->pCont;
+  SMsgHead * pHead = pRpc->pCont;
   SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
   if (pVnode == NULL) return -1;
 
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index 9361c0e6d2..4a3b47c0d1 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -129,7 +129,7 @@ int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd) {
   return 0;
 }
 
-int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) {
+int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver, SRpcHandleInfo handleInfo) {
   if (msgType != TDMT_VND_SUBMIT) return 0;
   void*       pIter = NULL;
   STqExec*    pExec = NULL;
@@ -239,10 +239,9 @@ int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_
 
     void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
     tEncodeSMqDataBlkRsp(&abuf, &rsp);
-    pMsg->pCont = buf;
-    pMsg->contLen = tlen;
-    pMsg->code = 0;
-    tmsgSendRsp(pMsg);
+
+    SRpcMsg resp = {.info = handleInfo, .pCont = buf, .contLen = tlen, .code = 0};
+    tmsgSendRsp(&resp);
 
     atomic_store_ptr(&pExec->pushHandle.handle, NULL);
     taosWUnLockLatch(&pExec->pushHandle.lock);
@@ -407,9 +406,9 @@ int32_t tqDeserializeConsumer(STQ* pTq, const STqSerializedHead* pHead, STqConsu
       pTopic->buffer.output[j].status = 0;
       STqReadHandle* pReadHandle = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
       SReadHandle    handle = {
-             .reader = pReadHandle,
-             .meta = pTq->pVnode->pMeta,
-             .pMsgCb = &pTq->pVnode->msgCb,
+          .reader = pReadHandle,
+          .meta = pTq->pVnode->pMeta,
+          .pMsgCb = &pTq->pVnode->msgCb,
       };
       pTopic->buffer.output[j].pReadHandle = pReadHandle;
       pTopic->buffer.output[j].task = qCreateStreamExecTaskInfo(pTopic->qmsg, &handle);
@@ -663,10 +662,9 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
 
   void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
   tEncodeSMqDataBlkRsp(&abuf, &rsp);
-  pMsg->pCont = buf;
-  pMsg->contLen = tlen;
-  pMsg->code = 0;
-  tmsgSendRsp(pMsg);
+
+  SRpcMsg resp = {.info = pMsg->info, .pCont = buf, .contLen = tlen, .code = 0};
+  tmsgSendRsp(&resp);
 
   tqDebug("vg %d offset %ld from consumer %ld (epoch %d) send rsp, block num: %d, reqOffset: %ld, rspOffset: %ld",
           TD_VID(pTq->pVnode), fetchOffset, consumerId, pReq->epoch, rsp.blockNum, rsp.reqOffset, rsp.rspOffset);
@@ -845,12 +843,10 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
       /*rsp.pBlockData = pRes;*/
 
       /*taosArrayDestroyEx(rsp.pBlockData, (void (*)(void*))tDeleteSSDataBlock);*/
-      pMsg->pCont = buf;
-      pMsg->contLen = msgLen;
-      pMsg->code = 0;
+      SRpcMsg resp = {.info = pMsg->info, pCont = buf, .contLen = msgLen, .code = 0};
       tqDebug("vg %d offset %ld msgType %d from consumer %ld (epoch %d) actual rsp", TD_VID(pTq->pVnode), fetchOffset,
              pHead->msgType, consumerId, pReq->epoch);
-      tmsgSendRsp(pMsg);
+      tmsgSendRsp(&resp);
       taosMemoryFree(pHead);
       return 0;
     } else {
@@ -878,10 +874,9 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
 
   void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
   tEncodeSMqPollRspV2(&abuf, &rspV2);
-  pMsg->pCont = buf;
-  pMsg->contLen = tlen;
-  pMsg->code = 0;
-  tmsgSendRsp(pMsg);
+
+  SRpcMsg resp = {.info = pMsg->info, .pCont = buf, .contLen = tlen, .code = 0};
+  tmsgSendRsp(&resp);
   tqDebug("vg %d offset %ld from consumer %ld (epoch %d) not rsp", TD_VID(pTq->pVnode), fetchOffset, consumerId,
          pReq->epoch);
   /*}*/
@@ -990,10 +985,10 @@ int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask, int32_t parallel) {
     for (int32_t i = 0; i < parallel; i++) {
       STqReadHandle* pStreamReader = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
       SReadHandle    handle = {
-             .reader = pStreamReader,
-             .meta = pTq->pVnode->pMeta,
-             .pMsgCb = &pTq->pVnode->msgCb,
-             .vnode = pTq->pVnode,
+          .reader = pStreamReader,
+          .meta = pTq->pVnode->pMeta,
+          .pMsgCb = &pTq->pVnode->msgCb,
+          .vnode = pTq->pVnode,
       };
       pTask->exec.runners[i].inputHandle = pStreamReader;
       pTask->exec.runners[i].executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle);
diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c
index dfd78d9dca..466c300d97 100644
--- a/source/dnode/vnode/src/vnd/vnodeSvr.c
+++ b/source/dnode/vnode/src/vnd/vnodeSvr.c
@@ -162,7 +162,7 @@ int vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
 
 int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
   vTrace("message in fetch queue is processing");
-  char   *msgstr = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
+  char *  msgstr = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
   int32_t msgLen = pMsg->contLen - sizeof(SMsgHead);
   switch (pMsg->msgType) {
     case TDMT_VND_FETCH:
@@ -184,8 +184,12 @@ int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
     case TDMT_VND_TASK_PIPE_EXEC:
     case TDMT_VND_TASK_MERGE_EXEC:
       return tqProcessTaskExec(pVnode->pTq, msgstr, msgLen, 0);
-    case TDMT_VND_STREAM_TRIGGER:
-      return tqProcessStreamTrigger(pVnode->pTq, pMsg->pCont, pMsg->contLen, 0);
+    case TDMT_VND_STREAM_TRIGGER: {
+      // refactor, avoid double free
+      int code = tqProcessStreamTrigger(pVnode->pTq, pMsg->pCont, pMsg->contLen, 0);
+      pMsg->pCont = NULL;
+      return code;
+    }
     case TDMT_VND_QUERY_HEARTBEAT:
       return qWorkerProcessHbMsg(pVnode, pVnode->pQuery, pMsg);
     default:
@@ -328,12 +332,12 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq,
   SDecoder           decoder = {0};
   int                rcode = 0;
   SVCreateTbBatchReq req = {0};
-  SVCreateTbReq     *pCreateReq;
+  SVCreateTbReq *    pCreateReq;
   SVCreateTbBatchRsp rsp = {0};
   SVCreateTbRsp      cRsp = {0};
   char               tbName[TSDB_TABLE_FNAME_LEN];
-  STbUidStore       *pStore = NULL;
-  SArray            *tbUids = NULL;
+  STbUidStore *      pStore = NULL;
+  SArray *           tbUids = NULL;
 
   pRsp->msgType = TDMT_VND_CREATE_TABLE_RSP;
   pRsp->code = TSDB_CODE_SUCCESS;
@@ -517,7 +521,7 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, in
   SDecoder         decoder = {0};
   SEncoder         encoder = {0};
   int              ret;
-  SArray          *tbUids = NULL;
+  SArray *         tbUids = NULL;
 
   pRsp->msgType = TDMT_VND_DROP_TABLE_RSP;
   pRsp->pCont = NULL;
@@ -572,9 +576,9 @@ _exit:
 
 static int vnodeDebugPrintSingleSubmitMsg(SMeta *pMeta, SSubmitBlk *pBlock, SSubmitMsgIter *msgIter, const char *tags) {
   SSubmitBlkIter blkIter = {0};
-  STSchema      *pSchema = NULL;
+  STSchema *     pSchema = NULL;
   tb_uid_t       suid = 0;
-  STSRow        *row = NULL;
+  STSRow *       row = NULL;
 
   tInitSubmitBlkIter(msgIter, pBlock, &blkIter);
   if (blkIter.row == NULL) return 0;
@@ -605,8 +609,8 @@ static int vnodeDebugPrintSingleSubmitMsg(SMeta *pMeta, SSubmitBlk *pBlock, SSub
 static int vnodeDebugPrintSubmitMsg(SVnode *pVnode, SSubmitReq *pMsg, const char *tags) {
   ASSERT(pMsg != NULL);
   SSubmitMsgIter msgIter = {0};
-  SMeta         *pMeta = pVnode->pMeta;
-  SSubmitBlk    *pBlock = NULL;
+  SMeta *        pMeta = pVnode->pMeta;
+  SSubmitBlk *   pBlock = NULL;
 
   if (tInitSubmitMsgIter(pMsg, &msgIter) < 0) return -1;
   while (true) {
@@ -620,10 +624,10 @@ static int vnodeDebugPrintSubmitMsg(SVnode *pVnode, SSubmitReq *pMsg, const char
 }
 
 static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
-  SSubmitReq    *pSubmitReq = (SSubmitReq *)pReq;
+  SSubmitReq *   pSubmitReq = (SSubmitReq *)pReq;
   SSubmitRsp     submitRsp = {0};
   SSubmitMsgIter msgIter = {0};
-  SSubmitBlk    *pBlock;
+  SSubmitBlk *   pBlock;
   SSubmitRsp     rsp = {0};
   SVCreateTbReq  createTbReq = {0};
   SDecoder       decoder = {0};
diff --git a/source/libs/parser/src/parInsert.c b/source/libs/parser/src/parInsert.c
index d1b4a745b9..b452950624 100644
--- a/source/libs/parser/src/parInsert.c
+++ b/source/libs/parser/src/parInsert.c
@@ -41,6 +41,13 @@
     sToken = tStrGetToken(pSql, &index, false);  \
   } while (0)
 
+#define NEXT_VALID_TOKEN(pSql, sToken)        \
+  do {                                        \
+    sToken.n = tGetToken(pSql, &sToken.type); \
+    sToken.z = pSql;                          \
+    pSql += sToken.n;                         \
+  } while (TK_NK_SPACE == sToken.type)
+
 typedef struct SInsertParseContext {
   SParseContext*     pComCxt;             // input
   char*              pSql;                // input
@@ -482,9 +489,11 @@ static int32_t parseValueToken(char** end, SToken* pToken, SSchema* pSchema, int
           return buildSyntaxErrMsg(pMsgBuf, "invalid bool data", pToken->z);
         }
       } else if (pToken->type == TK_NK_INTEGER) {
-        return func(pMsgBuf, ((taosStr2Int64(pToken->z, NULL, 10) == 0) ? &FALSE_VALUE : &TRUE_VALUE), pSchema->bytes, param);
+        return func(pMsgBuf, ((taosStr2Int64(pToken->z, NULL, 10) == 0) ? &FALSE_VALUE : &TRUE_VALUE), pSchema->bytes,
+                    param);
       } else if (pToken->type == TK_NK_FLOAT) {
-        return func(pMsgBuf, ((taosStr2Double(pToken->z, NULL) == 0) ? &FALSE_VALUE : &TRUE_VALUE), pSchema->bytes, param);
+        return func(pMsgBuf, ((taosStr2Double(pToken->z, NULL) == 0) ? &FALSE_VALUE : &TRUE_VALUE), pSchema->bytes,
+                    param);
       } else {
         return buildSyntaxErrMsg(pMsgBuf, "invalid bool data", pToken->z);
       }
@@ -685,7 +694,7 @@ static int32_t parseBoundColumns(SInsertParseContext* pCxt, SParsedDataColInfo*
       isOrdered = false;
     }
     if (index < 0) {
-      return buildSyntaxErrMsg(&pCxt->msg, "invalid column/tag name", sToken.z);
+      return generateSyntaxErrMsg(&pCxt->msg, TSDB_CODE_PAR_INVALID_COLUMN, sToken.z);
     }
     if (pColList->cols[index].valStat == VAL_STAT_HAS) {
       return buildSyntaxErrMsg(&pCxt->msg, "duplicated column name", sToken.z);
@@ -895,8 +904,10 @@ static int32_t parseUsingClause(SInsertParseContext* pCxt, SName* name, char* tb
     return buildSyntaxErrMsg(&pCxt->msg, "( is expected", sToken.z);
   }
   CHECK_CODE(parseTagsClause(pCxt, pCxt->pTableMeta->schema, getTableInfo(pCxt->pTableMeta).precision, name->tname));
-  NEXT_TOKEN(pCxt->pSql, sToken);
-  if (TK_NK_RP != sToken.type) {
+  NEXT_VALID_TOKEN(pCxt->pSql, sToken);
+  if (TK_NK_COMMA == sToken.type) {
+    return generateSyntaxErrMsg(&pCxt->msg, TSDB_CODE_PAR_TAGS_NOT_MATCHED);
+  } else if (TK_NK_RP != sToken.type) {
     return buildSyntaxErrMsg(&pCxt->msg, ") is expected", sToken.z);
   }
 
@@ -996,8 +1007,10 @@ static int32_t parseValues(SInsertParseContext* pCxt, STableDataBlocks* pDataBlo
       pDataBlock->size += extendedRowSize;  // len;
     }
 
-    NEXT_TOKEN(pCxt->pSql, sToken);
-    if (TK_NK_RP != sToken.type) {
+    NEXT_VALID_TOKEN(pCxt->pSql, sToken);
+    if (TK_NK_COMMA == sToken.type) {
+      return generateSyntaxErrMsg(&pCxt->msg, TSDB_CODE_PAR_INVALID_COLUMNS_NUM);
+    } else if (TK_NK_RP != sToken.type) {
       return buildSyntaxErrMsg(&pCxt->msg, ") expected", sToken.z);
     }
 
@@ -1057,10 +1070,10 @@ static void destroyInsertParseContext(SInsertParseContext* pCxt) {
 //       VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path
 //   [...];
 static int32_t parseInsertBody(SInsertParseContext* pCxt) {
-  int32_t tbNum = 0;
-  char tbFName[TSDB_TABLE_FNAME_LEN];
-  bool autoCreateTbl = false;
-  STableMeta *pMeta = NULL;
+  int32_t     tbNum = 0;
+  char        tbFName[TSDB_TABLE_FNAME_LEN];
+  bool        autoCreateTbl = false;
+  STableMeta* pMeta = NULL;
 
   // for each table
   while (1) {
@@ -1121,7 +1134,7 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
                                     &dataBuf, NULL, &pCxt->createTblReq));
     pMeta = pCxt->pTableMeta;
     pCxt->pTableMeta = NULL;
-    
+
     if (TK_NK_LP == sToken.type) {
       // pSql -> field1_name, ...)
       CHECK_CODE(parseBoundColumns(pCxt, &dataBuf->boundColumnInfo, getTableColumnSchema(pMeta)));
@@ -1160,7 +1173,8 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
       return TSDB_CODE_TSC_OUT_OF_MEMORY;
     }
     memcpy(tags, &pCxt->tags, sizeof(pCxt->tags));
-    (*pCxt->pStmtCb->setInfoFn)(pCxt->pStmtCb->pStmt, pMeta, tags, tbFName, autoCreateTbl, pCxt->pVgroupsHashObj, pCxt->pTableBlockHashObj);
+    (*pCxt->pStmtCb->setInfoFn)(pCxt->pStmtCb->pStmt, pMeta, tags, tbFName, autoCreateTbl, pCxt->pVgroupsHashObj,
+                                pCxt->pTableBlockHashObj);
 
     memset(&pCxt->tags, 0, sizeof(pCxt->tags));
     pCxt->pVgroupsHashObj = NULL;
@@ -1231,14 +1245,14 @@ int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery) {
       return TSDB_CODE_OUT_OF_MEMORY;
     }
   }
-  
+
   context.pOutput->payloadType = PAYLOAD_TYPE_KV;
 
   int32_t code = skipInsertInto(&context);
   if (TSDB_CODE_SUCCESS == code) {
     code = parseInsertBody(&context);
   }
-  if (TSDB_CODE_SUCCESS == code) {
+  if (TSDB_CODE_SUCCESS == code || NEED_CLIENT_HANDLE_ERROR(code)) {
     SName* pTable = taosHashIterate(context.pTableNameHashObj, NULL);
     while (NULL != pTable) {
       taosArrayPush((*pQuery)->pTableList, pTable);
@@ -1579,9 +1593,9 @@ typedef struct SmlExecTableHandle {
 } SmlExecTableHandle;
 
 typedef struct SmlExecHandle {
-  SHashObj*           pBlockHash;
-  SmlExecTableHandle  tableExecHandle;
-  SQuery              *pQuery;
+  SHashObj*          pBlockHash;
+  SmlExecTableHandle tableExecHandle;
+  SQuery*            pQuery;
 } SSmlExecHandle;
 
 static void smlDestroyTableHandle(void* pHandle) {
@@ -1673,9 +1687,9 @@ static int32_t smlBuildTagRow(SArray* cols, SKVRowBuilder* tagsBuilder, SParsedD
     SSchema* pTagSchema = &pSchema[tags->boundColumns[i] - 1];  // colId starts with 1
     param.schema = pTagSchema;
     SSmlKv* kv = taosArrayGetP(cols, i);
-    if(IS_VAR_DATA_TYPE(kv->type)){
+    if (IS_VAR_DATA_TYPE(kv->type)) {
       KvRowAppend(msg, kv->value, kv->length, &param);
-    }else{
+    } else {
       KvRowAppend(msg, &(kv->value), kv->length, &param);
     }
   }
@@ -1688,13 +1702,13 @@ static int32_t smlBuildTagRow(SArray* cols, SKVRowBuilder* tagsBuilder, SParsedD
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t smlBindData(void *handle, SArray *tags, SArray *colsSchema, SArray *cols, bool format,
-                    STableMeta *pTableMeta, char *tableName, char *msgBuf, int16_t msgBufLen) {
+int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols, bool format, STableMeta* pTableMeta,
+                    char* tableName, char* msgBuf, int16_t msgBufLen) {
   SMsgBuf pBuf = {.buf = msgBuf, .len = msgBufLen};
 
   SSmlExecHandle* smlHandle = (SSmlExecHandle*)handle;
-  smlDestroyTableHandle(&smlHandle->tableExecHandle);   // free for each table
-  SSchema*        pTagsSchema = getTableTagSchema(pTableMeta);
+  smlDestroyTableHandle(&smlHandle->tableExecHandle);  // free for each table
+  SSchema* pTagsSchema = getTableTagSchema(pTableMeta);
   setBoundColumnInfo(&smlHandle->tableExecHandle.tags, pTagsSchema, getNumOfTags(pTableMeta));
   int ret = smlBoundColumnData(tags, &smlHandle->tableExecHandle.tags, pTagsSchema);
   if (ret != TSDB_CODE_SUCCESS) {
@@ -1702,7 +1716,8 @@ int32_t smlBindData(void *handle, SArray *tags, SArray *colsSchema, SArray *cols
     return ret;
   }
   SKVRow row = NULL;
-  ret = smlBuildTagRow(tags, &smlHandle->tableExecHandle.tagsBuilder, &smlHandle->tableExecHandle.tags, pTagsSchema, &row, &pBuf);
+  ret = smlBuildTagRow(tags, &smlHandle->tableExecHandle.tagsBuilder, &smlHandle->tableExecHandle.tags, pTagsSchema,
+                       &row, &pBuf);
   if (ret != TSDB_CODE_SUCCESS) {
     return ret;
   }
@@ -1733,7 +1748,7 @@ int32_t smlBindData(void *handle, SArray *tags, SArray *colsSchema, SArray *cols
   initRowBuilder(&pDataBlock->rowBuilder, pDataBlock->pTableMeta->sversion, &pDataBlock->boundColumnInfo);
 
   int32_t rowNum = taosArrayGetSize(cols);
-  if(rowNum <= 0) {
+  if (rowNum <= 0) {
     return buildInvalidOperationMsg(&pBuf, "cols size <= 0");
   }
   ret = allocateMemForSize(pDataBlock, extendedRowSize * rowNum);
@@ -1744,9 +1759,9 @@ int32_t smlBindData(void *handle, SArray *tags, SArray *colsSchema, SArray *cols
   for (int32_t r = 0; r < rowNum; ++r) {
     STSRow* row = (STSRow*)(pDataBlock->pData + pDataBlock->size);  // skip the SSubmitBlk header
     tdSRowResetBuf(pBuilder, row);
-    void *rowData = taosArrayGetP(cols, r);
+    void*  rowData = taosArrayGetP(cols, r);
     size_t rowDataSize = 0;
-    if(format){
+    if (format) {
       rowDataSize = taosArrayGetSize(rowData);
     }
 
@@ -1781,9 +1796,9 @@ int32_t smlBindData(void *handle, SArray *tags, SArray *colsSchema, SArray *cols
           kv->i = convertTimePrecision(kv->i, TSDB_TIME_PRECISION_NANO, pTableMeta->tableInfo.precision);
         }
 
-        if(IS_VAR_DATA_TYPE(kv->type)){
+        if (IS_VAR_DATA_TYPE(kv->type)) {
           MemRowAppend(&pBuf, kv->value, colLen, &param);
-        }else{
+        } else {
           MemRowAppend(&pBuf, &(kv->value), colLen, &param);
         }
       }
diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c
index dbb29699fc..f6d53dd15a 100644
--- a/source/libs/parser/src/parTranslater.c
+++ b/source/libs/parser/src/parTranslater.c
@@ -120,6 +120,20 @@ static int32_t getTableMeta(STranslateContext* pCxt, const char* pDbName, const
   return getTableMetaImpl(pCxt, toName(pCxt->pParseCxt->acctId, pDbName, pTableName, &name), pMeta);
 }
 
+static int32_t refreshGetTableMeta(STranslateContext* pCxt, const char* pDbName, const char* pTableName,
+                                   STableMeta** pMeta) {
+  SParseContext* pParCxt = pCxt->pParseCxt;
+  SName          name;
+  toName(pCxt->pParseCxt->acctId, pDbName, pTableName, &name);
+  int32_t code =
+      catalogRefreshGetTableMeta(pParCxt->pCatalog, pParCxt->pTransporter, &pParCxt->mgmtEpSet, &name, pMeta, false);
+  if (TSDB_CODE_SUCCESS != code) {
+    parserError("catalogRefreshGetTableMeta error, code:%s, dbName:%s, tbName:%s", tstrerror(code), pDbName,
+                pTableName);
+  }
+  return code;
+}
+
 static int32_t getTableDistVgInfo(STranslateContext* pCxt, const SName* pName, SArray** pVgInfo) {
   SParseContext* pParCxt = pCxt->pParseCxt;
   int32_t        code = collectUseDatabase(pName, pCxt->pDbs);
@@ -3201,7 +3215,7 @@ static int32_t translateExplain(STranslateContext* pCxt, SExplainStmt* pStmt) {
 }
 
 static int32_t translateDescribe(STranslateContext* pCxt, SDescribeStmt* pStmt) {
-  return getTableMeta(pCxt, pStmt->dbName, pStmt->tableName, &pStmt->pMeta);
+  return refreshGetTableMeta(pCxt, pStmt->dbName, pStmt->tableName, &pStmt->pMeta);
 }
 
 static int32_t translateKillConnection(STranslateContext* pCxt, SKillStmt* pStmt) {
diff --git a/source/libs/qworker/src/qworkerMsg.c b/source/libs/qworker/src/qworkerMsg.c
index 60270d3e06..562e550bdc 100644
--- a/source/libs/qworker/src/qworkerMsg.c
+++ b/source/libs/qworker/src/qworkerMsg.c
@@ -47,7 +47,7 @@ int32_t qwBuildAndSendQueryRsp(SRpcHandleInfo *pConn, int32_t code) {
   SQueryTableRsp rsp = {.code = code};
 
   int32_t contLen = tSerializeSQueryTableRsp(NULL, 0, &rsp);
-  void   *msg = rpcMallocCont(contLen);
+  void *  msg = rpcMallocCont(contLen);
   tSerializeSQueryTableRsp(msg, contLen, &rsp);
 
   SRpcMsg rpcRsp = {
@@ -85,7 +85,7 @@ int32_t qwBuildAndSendExplainRsp(SRpcHandleInfo *pConn, SExplainExecInfo *execIn
   SExplainRsp rsp = {.numOfPlans = num, .subplanInfo = execInfo};
 
   int32_t contLen = tSerializeSExplainRsp(NULL, 0, &rsp);
-  void   *pRsp = rpcMallocCont(contLen);
+  void *  pRsp = rpcMallocCont(contLen);
   tSerializeSExplainRsp(pRsp, contLen, &rsp);
 
   SRpcMsg rpcRsp = {
@@ -104,7 +104,7 @@ int32_t qwBuildAndSendExplainRsp(SRpcHandleInfo *pConn, SExplainExecInfo *execIn
 
 int32_t qwBuildAndSendHbRsp(SRpcHandleInfo *pConn, SSchedulerHbRsp *pStatus, int32_t code) {
   int32_t contLen = tSerializeSSchedulerHbRsp(NULL, 0, pStatus);
-  void   *pRsp = rpcMallocCont(contLen);
+  void *  pRsp = rpcMallocCont(contLen);
   tSerializeSSchedulerHbRsp(pRsp, contLen, pStatus);
 
   SRpcMsg rpcRsp = {
@@ -212,7 +212,7 @@ int32_t qwBuildAndSendShowRsp(SRpcMsg *pMsg, int32_t code) {
   showRsp.tableMeta.numOfColumns = cols;
 
   int32_t bufLen = tSerializeSShowRsp(NULL, 0, &showRsp);
-  void   *pBuf = rpcMallocCont(bufLen);
+  void *  pBuf = rpcMallocCont(bufLen);
   tSerializeSShowRsp(pBuf, bufLen, &showRsp);
 
   SRpcMsg rpcMsg = {
@@ -341,7 +341,7 @@ int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
 
   int32_t       code = 0;
   SSubQueryMsg *msg = pMsg->pCont;
-  SQWorker *mgmt = (SQWorker *)qWorkerMgmt;
+  SQWorker *    mgmt = (SQWorker *)qWorkerMgmt;
 
   if (NULL == msg || pMsg->contLen <= sizeof(*msg)) {
     QW_ELOG("invalid query msg, msg:%p, msgLen:%d", msg, pMsg->contLen);
@@ -361,7 +361,7 @@ int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   int64_t  rId = msg->refId;
 
   SQWMsg qwMsg = {.node = node, .msg = msg->msg + msg->sqlLen, .msgLen = msg->phyLen, .connInfo = pMsg->info};
-  char  *sql = strndup(msg->msg, msg->sqlLen);
+  char * sql = strndup(msg->msg, msg->sqlLen);
   QW_SCH_TASK_DLOG("processQuery start, node:%p, handle:%p, sql:%s", node, pMsg->info.handle, sql);
   taosMemoryFreeClear(sql);
 
@@ -378,8 +378,8 @@ int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   bool               queryDone = false;
   SQueryContinueReq *msg = (SQueryContinueReq *)pMsg->pCont;
   bool               needStop = false;
-  SQWTaskCtx        *handles = NULL;
-  SQWorker      *mgmt = (SQWorker *)qWorkerMgmt;
+  SQWTaskCtx *       handles = NULL;
+  SQWorker *         mgmt = (SQWorker *)qWorkerMgmt;
 
   if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
     QW_ELOG("invalid cquery msg, msg:%p, msgLen:%d", msg, pMsg->contLen);
@@ -407,7 +407,7 @@ int32_t qWorkerProcessReadyMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
     return TSDB_CODE_QRY_INVALID_INPUT;
   }
 
-  SQWorker *mgmt = (SQWorker *)qWorkerMgmt;
+  SQWorker *    mgmt = (SQWorker *)qWorkerMgmt;
   SResReadyReq *msg = pMsg->pCont;
   if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
     QW_ELOG("invalid task ready msg, msg:%p, msgLen:%d", msg, pMsg->contLen);
@@ -467,7 +467,7 @@ int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   }
 
   SResFetchReq *msg = pMsg->pCont;
-  SQWorker *mgmt = (SQWorker *)qWorkerMgmt;
+  SQWorker *    mgmt = (SQWorker *)qWorkerMgmt;
 
   if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
     QW_ELOG("invalid fetch msg, msg:%p, msgLen:%d", msg, pMsg->contLen);
@@ -505,7 +505,7 @@ int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
     return TSDB_CODE_QRY_INVALID_INPUT;
   }
 
-  SQWorker   *mgmt = (SQWorker *)qWorkerMgmt;
+  SQWorker *      mgmt = (SQWorker *)qWorkerMgmt;
   int32_t         code = 0;
   STaskCancelReq *msg = pMsg->pCont;
   if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
@@ -542,7 +542,7 @@ int32_t qWorkerProcessDropMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
 
   int32_t       code = 0;
   STaskDropReq *msg = pMsg->pCont;
-  SQWorker *mgmt = (SQWorker *)qWorkerMgmt;
+  SQWorker *    mgmt = (SQWorker *)qWorkerMgmt;
 
   if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
     QW_ELOG("invalid task drop msg, msg:%p, msgLen:%d", msg, pMsg->contLen);
@@ -581,7 +581,7 @@ int32_t qWorkerProcessHbMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
 
   int32_t         code = 0;
   SSchedulerHbReq req = {0};
-  SQWorker   *mgmt = (SQWorker *)qWorkerMgmt;
+  SQWorker *      mgmt = (SQWorker *)qWorkerMgmt;
 
   if (NULL == pMsg->pCont) {
     QW_ELOG("invalid hb msg, msg:%p, msgLen:%d", pMsg->pCont, pMsg->contLen);
diff --git a/source/libs/transport/src/trans.c b/source/libs/transport/src/trans.c
index 5c01034f35..5627dbfbf5 100644
--- a/source/libs/transport/src/trans.c
+++ b/source/libs/transport/src/trans.c
@@ -94,7 +94,9 @@ void rpcFreeCont(void* cont) {
   if (cont == NULL) {
     return;
   }
+
   taosMemoryFree((char*)cont - TRANS_MSG_OVERHEAD);
+  tTrace("free mem: %p", (char*)cont - TRANS_MSG_OVERHEAD);
 }
 void* rpcReallocCont(void* ptr, int contLen) {
   if (ptr == NULL) {
diff --git a/source/libs/transport/src/transComm.c b/source/libs/transport/src/transComm.c
index 98e9e67ede..7014cc481f 100644
--- a/source/libs/transport/src/transComm.c
+++ b/source/libs/transport/src/transComm.c
@@ -133,6 +133,7 @@ int transAllocBuffer(SConnBuffer* connBuf, uv_buf_t* uvBuf) {
   } else {
     p->cap = p->total;
     p->buf = taosMemoryRealloc(p->buf, p->cap);
+    tTrace("internal malloc mem: %p, size: %d", p->buf, p->cap);
 
     uvBuf->base = p->buf + p->len;
     uvBuf->len = p->cap - p->len;
diff --git a/source/libs/transport/src/transSrv.c b/source/libs/transport/src/transSrv.c
index 84adeb4bf6..da83a6f37f 100644
--- a/source/libs/transport/src/transSrv.c
+++ b/source/libs/transport/src/transSrv.c
@@ -469,6 +469,8 @@ static void uvStartSendResp(SSrvMsg* smsg) {
 
   if (pConn->broken == true) {
     // persist by
+    transFreeMsg(smsg->msg.pCont);
+    taosMemoryFree(smsg);
     transUnrefSrvHandle(pConn);
     return;
   }