Merge branch 'feature/qnode' of https://github.com/taosdata/TDengine into feature/qnode

include/common/tmsg.h

@@ -651,7 +651,6 @@ typedef struct SQueryNodeAddr {
   SEpSet  epSet;
 } SQueryNodeAddr;
 
-
 typedef struct {
   SArray* pArray;  // Array of SUseDbRsp
 } SUseDbBatchRsp;
@@ -724,7 +723,7 @@ typedef struct {
 
 int32_t tSerializeSRetrieveFuncRsp(void* buf, int32_t bufLen, SRetrieveFuncRsp* pRsp);
 int32_t tDeserializeSRetrieveFuncRsp(void* buf, int32_t bufLen, SRetrieveFuncRsp* pRsp);
-void tFreeSFuncInfo(SFuncInfo *pInfo);
+void    tFreeSFuncInfo(SFuncInfo* pInfo);
 void    tFreeSRetrieveFuncRsp(SRetrieveFuncRsp* pRsp);
 
 typedef struct {

include/libs/transport/trpc.h

@@ -38,11 +38,12 @@ typedef struct SRpcConnInfo {
 
 typedef struct SRpcMsg {
   tmsg_t  msgType;
-  void   *pCont;
+  void *  pCont;
   int     contLen;
   int32_t code;
-  void   *handle;         // rpc handle returned to app
-  void   *ahandle;        // app handle set by client
+  void *  handle;         // rpc handle returned to app
+  void *  ahandle;        // app handle set by client
+  int64_t refId;          // refid, used by server
   int     noResp;         // has response or not(default 0, 0: resp, 1: no resp);
   int     persistHandle;  // persist handle or not
 
@@ -54,8 +55,8 @@ typedef struct {
   uint16_t clientPort;
   SRpcMsg  rpcMsg;
   int32_t  rspLen;
-  void    *pRsp;
-  void    *pNode;
+  void *   pRsp;
+  void *   pNode;
 } SNodeMsg;
 
 typedef void (*RpcCfp)(void *parent, SRpcMsg *, SEpSet *);
@@ -64,7 +65,7 @@ typedef int (*RpcRfp)(void *parent, SRpcMsg *, SEpSet *);
 
 typedef struct SRpcInit {
   uint16_t localPort;     // local port
-  char    *label;         // for debug purpose
+  char *   label;         // for debug purpose
   int      numOfThreads;  // number of threads to handle connections
   int      sessions;      // number of sessions allowed
   int8_t   connType;      // TAOS_CONN_UDP, TAOS_CONN_TCPC, TAOS_CONN_TCPS
@@ -97,23 +98,23 @@ typedef struct {
 
 typedef struct {
   int32_t msgType;
-  void   *val;
+  void *  val;
   int32_t (*clone)(void *src, void **dst);
   void (*freeFunc)(const void *arg);
 } SRpcBrokenlinkVal;
 
 typedef struct {
-  SHashObj         *args;
+  SHashObj *        args;
   SRpcBrokenlinkVal brokenVal;
 } SRpcCtx;
 
 int32_t rpcInit();
 void    rpcCleanup();
-void   *rpcOpen(const SRpcInit *pRpc);
+void *  rpcOpen(const SRpcInit *pRpc);
 void    rpcClose(void *);
-void   *rpcMallocCont(int contLen);
+void *  rpcMallocCont(int contLen);
 void    rpcFreeCont(void *pCont);
-void   *rpcReallocCont(void *ptr, int contLen);
+void *  rpcReallocCont(void *ptr, int contLen);
 
 // Because taosd supports multi-process mode
 // These functions should not be used on the server side

include/libs/wal/wal.h

@@ -92,7 +92,7 @@ typedef struct SWalReadHead {
   int8_t  headVer;
   int8_t  reserved;
   int16_t msgType;
-  int32_t len;
+  int32_t bodyLen;
   int64_t ingestTs;  // not implemented
   int64_t version;
 

include/util/tprocess.h

@@ -53,8 +53,8 @@ int32_t   taosProcRun(SProcObj *pProc);
 void      taosProcStop(SProcObj *pProc);
 
 int32_t taosProcPutToChildQ(SProcObj *pProc, const void *pHead, int16_t headLen, const void *pBody, int32_t bodyLen,
-                            void *handle, EProcFuncType ftype);
-void    taosProcRemoveHandle(SProcObj *pProc, void *handle);
+                            void *handle, int64_t handleRef, EProcFuncType ftype);
+int64_t taosProcRemoveHandle(SProcObj *pProc, void *handle);
 void    taosProcCloseHandles(SProcObj *pProc, void (*HandleFp)(void *handle));
 void    taosProcPutToParentQ(SProcObj *pProc, const void *pHead, int16_t headLen, const void *pBody, int32_t bodyLen,
                              EProcFuncType ftype);

source/common/src/tglobal.c

@@ -237,18 +237,20 @@ static int32_t taosLoadCfg(SConfig *pCfg, const char *inputCfgDir, const char *e
   char cfgFile[PATH_MAX + 100] = {0};
 
   taosExpandDir(inputCfgDir, cfgDir, PATH_MAX);
+  if (taosIsDir(cfgDir)) {
     snprintf(cfgFile, sizeof(cfgFile), "%s" TD_DIRSEP "taos.cfg", cfgDir);
+  } else {
+    tstrncpy(cfgFile, cfgDir, sizeof(cfgDir));
+  }
 
   if (cfgLoad(pCfg, CFG_STYPE_APOLLO_URL, apolloUrl) != 0) {
     uError("failed to load from apollo url:%s since %s", apolloUrl, terrstr());
     return -1;
   }
 
-  if (cfgLoad(pCfg, CFG_STYPE_CFG_FILE, cfgDir) != 0) {
   if (cfgLoad(pCfg, CFG_STYPE_CFG_FILE, cfgFile) != 0) {
-      uInfo("cfg file:%s not read since %s", cfgFile, terrstr());
-      return 0;
-    }
+    uError("failed to load from cfg file:%s since %s", cfgFile, terrstr());
+    return -1;
   }
 
   if (cfgLoad(pCfg, CFG_STYPE_ENV_FILE, envFile) != 0) {

source/dnode/mgmt/exe/dmMain.c

@@ -190,7 +190,7 @@ int main(int argc, char const *argv[]) {
   }
 
   if (dmInitLog() != 0) {
-    printf("failed to start since init log error\n");
+    dError("failed to start since init log error");
     return -1;
   }
 

source/dnode/mgmt/implement/src/dmTransport.c

@@ -59,6 +59,10 @@ static inline int32_t dmBuildMsg(SNodeMsg *pMsg, SRpcMsg *pRpc) {
   pMsg->clientIp = connInfo.clientIp;
   pMsg->clientPort = connInfo.clientPort;
   memcpy(&pMsg->rpcMsg, pRpc, sizeof(SRpcMsg));
+  if ((pRpc->msgType & 1u)) {
+    assert(pRpc->refId != 0);
+  }
+  // assert(pRpc->handle != NULL && pRpc->refId != 0 && pMsg->rpcMsg.refId != 0);
   return 0;
 }
 
@@ -84,7 +88,7 @@ static void dmProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSe
     dTrace("msg:%p, is created and put into child queue, type:%s handle:%p user:%s", pMsg, TMSG_INFO(msgType),
            pRpc->handle, pMsg->user);
     code = taosProcPutToChildQ(pWrapper->procObj, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen, pRpc->handle,
-                               PROC_FUNC_REQ);
+                               pRpc->refId, PROC_FUNC_REQ);
   } else {
     dTrace("msg:%p, should not processed in child process, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user);
     ASSERT(1);
@@ -107,7 +111,7 @@ _OVER:
         }
       }
 
-      SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = code};
+      SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = code, .refId = pRpc->refId};
       tmsgSendRsp(&rsp);
     }
     dTrace("msg:%p, is freed", pMsg);
@@ -134,7 +138,8 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
   if (pDnode->status != DND_STAT_RUNNING) {
     dError("msg:%s ignored since dnode not running, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle);
     if (isReq) {
-      SRpcMsg rspMsg = {.handle = pMsg->handle, .code = TSDB_CODE_APP_NOT_READY, .ahandle = pMsg->ahandle};
+      SRpcMsg rspMsg = {
+          .handle = pMsg->handle, .code = TSDB_CODE_APP_NOT_READY, .ahandle = pMsg->ahandle, .refId = pMsg->refId};
       rpcSendResponse(&rspMsg);
     }
     rpcFreeCont(pMsg->pCont);
@@ -143,7 +148,8 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
 
   if (isReq && pMsg->pCont == NULL) {
     dError("req:%s not processed since its empty, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle);
-    SRpcMsg rspMsg = {.handle = pMsg->handle, .code = TSDB_CODE_INVALID_MSG_LEN, .ahandle = pMsg->ahandle};
+    SRpcMsg rspMsg = {
+        .handle = pMsg->handle, .code = TSDB_CODE_INVALID_MSG_LEN, .ahandle = pMsg->ahandle, .refId = pMsg->refId};
     rpcSendResponse(&rspMsg);
     return;
   }
@@ -151,7 +157,8 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
   if (pWrapper == NULL) {
     dError("msg:%s not processed since no handle, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle);
     if (isReq) {
-      SRpcMsg rspMsg = {.handle = pMsg->handle, .code = TSDB_CODE_MSG_NOT_PROCESSED, .ahandle = pMsg->ahandle};
+      SRpcMsg rspMsg = {
+          .handle = pMsg->handle, .code = TSDB_CODE_MSG_NOT_PROCESSED, .ahandle = pMsg->ahandle, .refId = pMsg->refId};
       rpcSendResponse(&rspMsg);
     }
     rpcFreeCont(pMsg->pCont);
@@ -170,6 +177,9 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
   }
 
   dTrace("msg:%s will be processed by %s, app:%p", TMSG_INFO(msgType), pWrapper->name, pMsg->ahandle);
+  if (isReq) {
+    assert(pMsg->refId != 0);
+  }
   dmProcessRpcMsg(pWrapper, pMsg, pEpSet);
 }
 
@@ -317,7 +327,7 @@ static void dmConsumeChildQueue(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int16_t
   if (code != 0) {
     dError("msg:%p, failed to process since code:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
     if (pRpc->msgType & 1U) {
-      SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = terrno};
+      SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = terrno, .refId = pRpc->refId};
       dmSendRsp(pWrapper, &rsp);
     }
 
@@ -346,7 +356,7 @@ static void dmConsumeParentQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, int16_t
       dmSendRpcReq(pWrapper->pDnode, (SEpSet *)((char *)pMsg + sizeof(SRpcMsg)), pMsg);
       break;
     case PROC_FUNC_RSP:
-      taosProcRemoveHandle(pWrapper->procObj, pMsg->handle);
+      pMsg->refId = taosProcRemoveHandle(pWrapper->procObj, pMsg->handle);
       dmSendRpcRsp(pWrapper->pDnode, pMsg);
       break;
     default:

source/dnode/mgmt/implement/src/dmWorker.c

@@ -105,7 +105,7 @@ void dmStopMonitorThread(SDnode *pDnode) {
 }
 
 static void dmProcessMgmtQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
-  SDnode  *pDnode = pInfo->ahandle;
+  SDnode * pDnode = pInfo->ahandle;
   SRpcMsg *pRpc = &pMsg->rpcMsg;
   int32_t  code = -1;
   dTrace("msg:%p, will be processed in dnode-mgmt queue", pMsg);
@@ -150,7 +150,7 @@ static void dmProcessMgmtQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
 
   if (pRpc->msgType & 1u) {
     if (code != 0) code = terrno;
-    SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = code};
+    SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = code, .refId = pRpc->refId};
     rpcSendResponse(&rsp);
   }
 

source/dnode/mgmt/interface/src/dmInt.c

@@ -177,7 +177,7 @@ void dmProcessServerStatusReq(SDnode *pDnode, SRpcMsg *pReq) {
   SServerStatusRsp statusRsp = {0};
   dmGetServerStatus(pDnode, &statusRsp);
 
-  SRpcMsg rspMsg = {.handle = pReq->handle, .ahandle = pReq->ahandle};
+  SRpcMsg rspMsg = {.handle = pReq->handle, .ahandle = pReq->ahandle, .refId = pReq->refId};
   int32_t rspLen = tSerializeSServerStatusRsp(NULL, 0, &statusRsp);
   if (rspLen < 0) {
     rspMsg.code = TSDB_CODE_OUT_OF_MEMORY;

source/dnode/mgmt/mgmt_bnode/src/bmHandle.c

@@ -53,7 +53,7 @@ int32_t bmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
     return -1;
   }
 
-  if (createReq.dnodeId != pDnode->data.dnodeId) {
+  if (pDnode->data.dnodeId != 0 && createReq.dnodeId != pDnode->data.dnodeId) {
     terrno = TSDB_CODE_INVALID_OPTION;
     dError("failed to create bnode since %s, input:%d cur:%d", terrstr(), createReq.dnodeId, pDnode->data.dnodeId);
     return -1;

source/dnode/mgmt/mgmt_bnode/src/bmWorker.c

@@ -17,7 +17,8 @@
 #include "bmInt.h"
 
 static void bmSendErrorRsp(SNodeMsg *pMsg, int32_t code) {
-  SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle, .code = code};
+  SRpcMsg rpcRsp = {
+      .handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle, .code = code, .refId = pMsg->rpcMsg.refId};
   tmsgSendRsp(&rpcRsp);
 
   dTrace("msg:%p, is freed", pMsg);
@@ -38,6 +39,7 @@ static void bmSendErrorRsps(STaosQall *qall, int32_t numOfMsgs, int32_t code) {
 static inline void bmSendRsp(SNodeMsg *pMsg, int32_t code) {
   SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
                  .ahandle = pMsg->rpcMsg.ahandle,
+                 .refId = pMsg->rpcMsg.refId,
                  .code = code,
                  .pCont = pMsg->pRsp,
                  .contLen = pMsg->rspLen};
@@ -101,7 +103,7 @@ static void bmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 }
 
 int32_t bmProcessWriteMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SBnodeMgmt   *pMgmt = pWrapper->pMgmt;
+  SBnodeMgmt *  pMgmt = pWrapper->pMgmt;
   SMultiWorker *pWorker = &pMgmt->writeWorker;
 
   dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
@@ -110,7 +112,7 @@ int32_t bmProcessWriteMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 }
 
 int32_t bmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SBnodeMgmt    *pMgmt = pWrapper->pMgmt;
+  SBnodeMgmt *   pMgmt = pWrapper->pMgmt;
   SSingleWorker *pWorker = &pMgmt->monitorWorker;
 
   dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);

source/dnode/mgmt/mgmt_mnode/src/mmHandle.c

@@ -134,9 +134,9 @@ int32_t mmProcessAlterReq(SMnodeMgmt *pMgmt, SNodeMsg *pMsg) {
     return -1;
   }
 
-  if (alterReq.dnodeId != pDnode->data.dnodeId) {
+  if (pDnode->data.dnodeId != 0 && alterReq.dnodeId != pDnode->data.dnodeId) {
     terrno = TSDB_CODE_INVALID_OPTION;
-    dError("failed to alter mnode since %s, dnodeId:%d input:%d", terrstr(), pDnode->data.dnodeId, alterReq.dnodeId);
+    dError("failed to alter mnode since %s, input:%d cur:%d", terrstr(), alterReq.dnodeId, pDnode->data.dnodeId);
     return -1;
   } else {
     return mmAlter(pMgmt, &alterReq);

source/dnode/mgmt/mgmt_mnode/src/mmWorker.c

@@ -19,6 +19,7 @@
 static inline void mmSendRsp(SNodeMsg *pMsg, int32_t code) {
   SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
                  .ahandle = pMsg->rpcMsg.ahandle,
+                 .refId = pMsg->rpcMsg.refId,
                  .code = code,
                  .pCont = pMsg->pRsp,
                  .contLen = pMsg->rspLen};

source/dnode/mgmt/mgmt_qnode/src/qmWorker.c

@@ -19,6 +19,7 @@
 static inline void qmSendRsp(SNodeMsg *pMsg, int32_t code) {
   SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
                  .ahandle = pMsg->rpcMsg.ahandle,
+                 .refId = pMsg->rpcMsg.refId,
                  .code = code,
                  .pCont = pMsg->pRsp,
                  .contLen = pMsg->rspLen};

source/dnode/mgmt/mgmt_snode/src/smWorker.c

@@ -19,6 +19,7 @@
 static inline void smSendRsp(SNodeMsg *pMsg, int32_t code) {
   SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
                  .ahandle = pMsg->rpcMsg.ahandle,
+                 .refId = pMsg->rpcMsg.refId,
                  .code = code,
                  .pCont = pMsg->pRsp,
                  .contLen = pMsg->rspLen};
@@ -149,7 +150,7 @@ static FORCE_INLINE int32_t smGetSWTypeFromMsg(SRpcMsg *pMsg) {
 }
 
 int32_t smProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SSnodeMgmt   *pMgmt = pWrapper->pMgmt;
+  SSnodeMgmt *  pMgmt = pWrapper->pMgmt;
   SMultiWorker *pWorker = taosArrayGetP(pMgmt->uniqueWorkers, 0);
   if (pWorker == NULL) {
     terrno = TSDB_CODE_INVALID_MSG;
@@ -162,7 +163,7 @@ int32_t smProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 }
 
 int32_t smProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SSnodeMgmt    *pMgmt = pWrapper->pMgmt;
+  SSnodeMgmt *   pMgmt = pWrapper->pMgmt;
   SSingleWorker *pWorker = &pMgmt->monitorWorker;
 
   dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
@@ -171,7 +172,7 @@ int32_t smProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 }
 
 int32_t smProcessUniqueMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SSnodeMgmt   *pMgmt = pWrapper->pMgmt;
+  SSnodeMgmt *  pMgmt = pWrapper->pMgmt;
   int32_t       index = smGetSWIdFromMsg(&pMsg->rpcMsg);
   SMultiWorker *pWorker = taosArrayGetP(pMgmt->uniqueWorkers, index);
   if (pWorker == NULL) {
@@ -185,7 +186,7 @@ int32_t smProcessUniqueMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 }
 
 int32_t smProcessSharedMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SSnodeMgmt    *pMgmt = pWrapper->pMgmt;
+  SSnodeMgmt *   pMgmt = pWrapper->pMgmt;
   SSingleWorker *pWorker = &pMgmt->sharedWorker;
 
   dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);

source/dnode/mgmt/mgmt_vnode/src/vmWorker.c

@@ -23,6 +23,7 @@
 static inline void vmSendRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) {
   SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
                  .ahandle = pMsg->rpcMsg.ahandle,
+                 .refId = pMsg->rpcMsg.refId,
                  .code = code,
                  .pCont = pMsg->pRsp,
                  .contLen = pMsg->rspLen};
@@ -126,6 +127,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
     rsp.code = 0;
     rsp.handle = pRpc->handle;
     rsp.ahandle = pRpc->ahandle;
+    rsp.refId = pRpc->refId;
 
     int32_t code = vnodeProcessWriteReq(pVnode->pImpl, pRpc, version++, &rsp);
     tmsgSendRsp(&rsp);
@@ -134,13 +136,14 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
   // sync integration response
   for (int i = 0; i < taosArrayGetSize(pArray); i++) {
     SNodeMsg *pMsg;
-    SRpcMsg  *pRpc;
+    SRpcMsg * pRpc;
 
     pMsg = *(SNodeMsg **)taosArrayGet(pArray, i);
     pRpc = &pMsg->rpcMsg;
 
     rsp.ahandle = pRpc->ahandle;
     rsp.handle = pRpc->handle;
+    rsp.refId = pRpc->refId;
     rsp.pCont = NULL;
     rsp.contLen = 0;
 
@@ -172,11 +175,9 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 
 static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
-  SNodeMsg  *pMsg = NULL;
+  SNodeMsg * pMsg = NULL;
   SRpcMsg    rsp;
 
-  // static int64_t version = 0;
-
   for (int32_t i = 0; i < numOfMsgs; ++i) {
 #if 1
     // sync integration
@@ -208,6 +209,7 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
     if (pMsg->rpcMsg.handle != NULL && pMsg->rpcMsg.ahandle != NULL) {
       rsp.ahandle = pMsg->rpcMsg.ahandle;
       rsp.handle = pMsg->rpcMsg.handle;
+      rsp.refId = pMsg->rpcMsg.refId;
       tmsgSendRsp(&rsp);
     }
 #endif
@@ -216,7 +218,7 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 
 static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
-  SNodeMsg  *pMsg = NULL;
+  SNodeMsg * pMsg = NULL;
 
   for (int32_t i = 0; i < numOfMsgs; ++i) {
     taosGetQitem(qall, (void **)&pMsg);
@@ -229,7 +231,7 @@ static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOf
 
 static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
-  SNodeMsg  *pMsg = NULL;
+  SNodeMsg * pMsg = NULL;
 
   for (int32_t i = 0; i < numOfMsgs; ++i) {
     taosGetQitem(qall, (void **)&pMsg);
@@ -246,7 +248,7 @@ static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 }
 
 static int32_t vmPutNodeMsgToQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg, EQueueType qtype) {
-  SRpcMsg  *pRpc = &pMsg->rpcMsg;
+  SRpcMsg * pRpc = &pMsg->rpcMsg;
   SMsgHead *pHead = pRpc->pCont;
   pHead->contLen = ntohl(pHead->contLen);
   pHead->vgId = ntohl(pHead->vgId);
@@ -315,7 +317,7 @@ int32_t vmProcessMergeMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 }
 
 int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SVnodesMgmt   *pMgmt = pWrapper->pMgmt;
+  SVnodesMgmt *  pMgmt = pWrapper->pMgmt;
   SSingleWorker *pWorker = &pMgmt->mgmtWorker;
   dTrace("msg:%p, will be written to vnode-mgmt queue, worker:%s", pMsg, pWorker->name);
   taosWriteQitem(pWorker->queue, pMsg);
@@ -323,7 +325,7 @@ int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 }
 
 int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SVnodesMgmt   *pMgmt = pWrapper->pMgmt;
+  SVnodesMgmt *  pMgmt = pWrapper->pMgmt;
   SSingleWorker *pWorker = &pMgmt->monitorWorker;
 
   dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
@@ -333,7 +335,7 @@ int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 
 static int32_t vmPutRpcMsgToQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, EQueueType qtype) {
   SVnodesMgmt *pMgmt = pWrapper->pMgmt;
-  SMsgHead    *pHead = pRpc->pCont;
+  SMsgHead *   pHead = pRpc->pCont;
 
   SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
   if (pVnode == NULL) return -1;
@@ -346,6 +348,7 @@ static int32_t vmPutRpcMsgToQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, EQueueT
   } else {
     dTrace("msg:%p, is created, type:%s", pMsg, TMSG_INFO(pRpc->msgType));
     pMsg->rpcMsg = *pRpc;
+    // if (pMsg->rpcMsg.handle != NULL) assert(pMsg->rpcMsg.refId != 0);
     switch (qtype) {
       case QUERY_QUEUE:
         dTrace("msg:%p, will be put into vnode-query queue", pMsg);

source/dnode/mnode/impl/inc/mndDef.h

@@ -135,6 +135,7 @@ typedef struct {
   int32_t    failedTimes;
   void*      rpcHandle;
   void*      rpcAHandle;
+  int64_t    rpcRefId;
   void*      rpcRsp;
   int32_t    rpcRspLen;
   SArray*    redoLogs;

source/dnode/mnode/impl/src/mndOffset.c

@@ -172,7 +172,7 @@ static int32_t mndProcessCommitOffsetReq(SNodeMsg *pMsg) {
     bool          create = false;
     SMqOffsetObj *pOffsetObj = mndAcquireOffset(pMnode, key);
     if (pOffsetObj == NULL) {
-      pOffsetObj = taosMemoryMalloc(sizeof(SMqOffset));
+      pOffsetObj = taosMemoryMalloc(sizeof(SMqOffsetObj));
       memcpy(pOffsetObj->key, key, TSDB_PARTITION_KEY_LEN);
       create = true;
     }

source/dnode/mnode/impl/src/mndTrans.c

@@ -193,9 +193,9 @@ TRANS_ENCODE_OVER:
 static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
   terrno = TSDB_CODE_OUT_OF_MEMORY;
 
-  SSdbRow     *pRow = NULL;
-  STrans      *pTrans = NULL;
-  char        *pData = NULL;
+  SSdbRow *    pRow = NULL;
+  STrans *     pTrans = NULL;
+  char *       pData = NULL;
   int32_t      dataLen = 0;
   int8_t       sver = 0;
   int32_t      redoLogNum = 0;
@@ -456,7 +456,7 @@ static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *pOld, STrans *pNew) {
 }
 
 static STrans *mndAcquireTrans(SMnode *pMnode, int32_t transId) {
-  SSdb   *pSdb = pMnode->pSdb;
+  SSdb *  pSdb = pMnode->pSdb;
   STrans *pTrans = sdbAcquire(pSdb, SDB_TRANS, &transId);
   if (pTrans == NULL) {
     terrno = TSDB_CODE_MND_TRANS_NOT_EXIST;
@@ -484,6 +484,7 @@ STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnType type, const S
   pTrans->createdTime = taosGetTimestampMs();
   pTrans->rpcHandle = pReq->handle;
   pTrans->rpcAHandle = pReq->ahandle;
+  pTrans->rpcRefId = pReq->refId;
   pTrans->redoLogs = taosArrayInit(MND_TRANS_ARRAY_SIZE, sizeof(void *));
   pTrans->undoLogs = taosArrayInit(MND_TRANS_ARRAY_SIZE, sizeof(void *));
   pTrans->commitLogs = taosArrayInit(MND_TRANS_ARRAY_SIZE, sizeof(void *));
@@ -625,7 +626,7 @@ static int32_t mndCheckTransCanBeStartedInParallel(SMnode *pMnode, STrans *pNewT
   if (mndIsBasicTrans(pNewTrans)) return 0;
 
   STrans *pTrans = NULL;
-  void   *pIter = NULL;
+  void *  pIter = NULL;
   int32_t code = 0;
 
   while (1) {
@@ -703,6 +704,7 @@ int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans) {
 
   pNew->rpcHandle = pTrans->rpcHandle;
   pNew->rpcAHandle = pTrans->rpcAHandle;
+  pNew->rpcRefId = pTrans->rpcRefId;
   pNew->rpcRsp = pTrans->rpcRsp;
   pNew->rpcRspLen = pTrans->rpcRspLen;
   pTrans->rpcRsp = NULL;
@@ -767,6 +769,7 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
     SRpcMsg rspMsg = {.handle = pTrans->rpcHandle,
                       .code = pTrans->code,
                       .ahandle = pTrans->rpcAHandle,
+                      .refId = pTrans->rpcRefId,
                       .pCont = rpcCont,
                       .contLen = pTrans->rpcRspLen};
     tmsgSendRsp(&rspMsg);
@@ -827,7 +830,7 @@ HANDLE_ACTION_RSP_OVER:
 }
 
 static int32_t mndTransExecuteLogs(SMnode *pMnode, SArray *pArray) {
-  SSdb   *pSdb = pMnode->pSdb;
+  SSdb *  pSdb = pMnode->pSdb;
   int32_t arraySize = taosArrayGetSize(pArray);
 
   if (arraySize == 0) return 0;
@@ -1202,11 +1205,11 @@ static int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) {
 }
 
 static int32_t mndProcessKillTransReq(SNodeMsg *pReq) {
-  SMnode       *pMnode = pReq->pNode;
+  SMnode *      pMnode = pReq->pNode;
   SKillTransReq killReq = {0};
   int32_t       code = -1;
-  SUserObj     *pUser = NULL;
-  STrans       *pTrans = NULL;
+  SUserObj *    pUser = NULL;
+  STrans *      pTrans = NULL;
 
   if (tDeserializeSKillTransReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &killReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
@@ -1246,7 +1249,7 @@ KILL_OVER:
 
 void mndTransPullup(SMnode *pMnode) {
   STrans *pTrans = NULL;
-  void   *pIter = NULL;
+  void *  pIter = NULL;
 
   while (1) {
     pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
@@ -1261,11 +1264,11 @@ void mndTransPullup(SMnode *pMnode) {
 
 static int32_t mndRetrieveTrans(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
   SMnode *pMnode = pReq->pNode;
-  SSdb   *pSdb = pMnode->pSdb;
+  SSdb *  pSdb = pMnode->pSdb;
   int32_t numOfRows = 0;
   STrans *pTrans = NULL;
   int32_t cols = 0;
-  char   *pWrite;
+  char *  pWrite;
 
   while (numOfRows < rows) {
     pShow->pIter = sdbFetch(pSdb, SDB_TRANS, pShow->pIter, (void **)&pTrans);

source/dnode/vnode/src/vnd/vnodeQuery.c

@@ -22,21 +22,21 @@ int vnodeQueryOpen(SVnode *pVnode) {
 void vnodeQueryClose(SVnode *pVnode) { qWorkerDestroy((void **)&pVnode->pQuery); }
 
 int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
-  STbCfg         *pTbCfg = NULL;
-  STbCfg         *pStbCfg = NULL;
+  STbCfg *        pTbCfg = NULL;
+  STbCfg *        pStbCfg = NULL;
   tb_uid_t        uid;
   int32_t         nCols;
   int32_t         nTagCols;
   SSchemaWrapper *pSW = NULL;
-  STableMetaRsp  *pTbMetaMsg = NULL;
+  STableMetaRsp * pTbMetaMsg = NULL;
   STableMetaRsp   metaRsp = {0};
-  SSchema        *pTagSchema;
+  SSchema *       pTagSchema;
   SRpcMsg         rpcMsg;
   int             msgLen = 0;
   int32_t         code = 0;
   char            tableFName[TSDB_TABLE_FNAME_LEN];
   int32_t         rspLen = 0;
-  void           *pRsp = NULL;
+  void *          pRsp = NULL;
 
   STableInfoReq infoReq = {0};
   if (tDeserializeSTableInfoReq(pMsg->pCont, pMsg->contLen, &infoReq) != 0) {
@@ -142,6 +142,7 @@ _exit:
 
   rpcMsg.handle = pMsg->handle;
   rpcMsg.ahandle = pMsg->ahandle;
+  rpcMsg.refId = pMsg->refId;
   rpcMsg.pCont = pRsp;
   rpcMsg.contLen = rspLen;
   rpcMsg.code = code;

source/dnode/vnode/src/vnd/vnodeSync.c

@@ -113,7 +113,7 @@ void vnodeSyncCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cb
         pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), beginIndex);
     syncRpcMsgLog2(logBuf, (SRpcMsg *)pMsg);
 
-    SVnode       *pVnode = (SVnode *)(pFsm->data);
+    SVnode *      pVnode = (SVnode *)(pFsm->data);
     SyncApplyMsg *pSyncApplyMsg = syncApplyMsgBuild2(pMsg, pVnode->config.vgId, &cbMeta);
     SRpcMsg       applyMsg;
     syncApplyMsg2RpcMsg(pSyncApplyMsg, &applyMsg);
@@ -133,6 +133,7 @@ void vnodeSyncCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cb
     if (ret == 1 && cbMeta.state == TAOS_SYNC_STATE_LEADER) {
       applyMsg.handle = saveRpcMsg.handle;
       applyMsg.ahandle = saveRpcMsg.ahandle;
+      applyMsg.refId = saveRpcMsg.refId;
     } else {
       applyMsg.handle = NULL;
       applyMsg.ahandle = NULL;

source/libs/qworker/inc/qworkerInt.h

@@ -60,7 +60,6 @@ enum {
   QW_WRITE,
 };
 
-
 enum {
   QW_NOT_EXIST_RET_ERR = 1,
   QW_NOT_EXIST_ADD,
@@ -73,14 +72,15 @@ typedef struct SQWDebug {
 } SQWDebug;
 
 typedef struct SQWConnInfo {
-  void *handle;
-  void *ahandle;
+  void *  handle;
+  void *  ahandle;
+  int64_t refId;
 } SQWConnInfo;
 
 typedef struct SQWMsg {
-  void        *node;
+  void *      node;
   int32_t     code;
-  char        *msg;
+  char *      msg;
   int32_t     msgLen;
   SQWConnInfo connInfo;
 } SQWMsg;
@@ -100,7 +100,6 @@ typedef struct SQWPhaseOutput {
 #endif
 } SQWPhaseOutput;
 
-
 typedef struct SQWTaskStatus {
   int64_t refId;  // job's refId
   int32_t code;
@@ -124,8 +123,8 @@ typedef struct SQWTaskCtx {
 
   int8_t events[QW_EVENT_MAX];
 
-  qTaskInfo_t     taskHandle;
-  DataSinkHandle  sinkHandle;
+  void *taskHandle;
+  void *sinkHandle;
 } SQWTaskCtx;
 
 typedef struct SQWSchStatus {
@@ -134,7 +133,7 @@ typedef struct SQWSchStatus {
   SQWConnInfo    hbConnInfo;
   SQueryNodeEpId hbEpId;
   SRWLatch       tasksLock;
-  SHashObj      *tasksHash;   // key:queryId+taskId, value: SQWTaskStatus
+  SHashObj *     tasksHash;  // key:queryId+taskId, value: SQWTaskStatus
 } SQWSchStatus;
 
 // Qnode/Vnode level task management
@@ -142,7 +141,7 @@ typedef struct SQWorkerMgmt {
   SQWorkerCfg cfg;
   int8_t      nodeType;
   int32_t     nodeId;
-  void       *timer;
+  void *      timer;
   tmr_h       hbTimer;
   SRWLatch    schLock;
   // SRWLatch ctxLock;
@@ -171,41 +170,86 @@ typedef struct SQWorkerMgmt {
 
 #define QW_TASK_NOT_EXIST(code)     (TSDB_CODE_QRY_SCH_NOT_EXIST == (code) || TSDB_CODE_QRY_TASK_NOT_EXIST == (code))
 #define QW_TASK_ALREADY_EXIST(code) (TSDB_CODE_QRY_TASK_ALREADY_EXIST == (code))
-#define QW_TASK_READY(status) (status == JOB_TASK_STATUS_SUCCEED || status == JOB_TASK_STATUS_FAILED || status == JOB_TASK_STATUS_CANCELLED || status == JOB_TASK_STATUS_PARTIAL_SUCCEED)
-#define QW_SET_QTID(id, qId, tId) do { *(uint64_t *)(id) = (qId); *(uint64_t *)((char *)(id) + sizeof(qId)) = (tId); } while (0)
-#define QW_GET_QTID(id, qId, tId) do { (qId) = *(uint64_t *)(id); (tId) = *(uint64_t *)((char *)(id) + sizeof(qId)); } while (0)
-
-#define QW_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; return _code; } } while (0)
-#define QW_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)
-#define QW_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
+#define QW_TASK_READY(status)                                                                                      \
+  (status == JOB_TASK_STATUS_SUCCEED || status == JOB_TASK_STATUS_FAILED || status == JOB_TASK_STATUS_CANCELLED || \
+   status == JOB_TASK_STATUS_PARTIAL_SUCCEED)
+#define QW_SET_QTID(id, qId, tId)                      \
+  do {                                                 \
+    *(uint64_t *)(id) = (qId);                         \
+    *(uint64_t *)((char *)(id) + sizeof(qId)) = (tId); \
+  } while (0)
+#define QW_GET_QTID(id, qId, tId)                      \
+  do {                                                 \
+    (qId) = *(uint64_t *)(id);                         \
+    (tId) = *(uint64_t *)((char *)(id) + sizeof(qId)); \
+  } while (0)
+
+#define QW_ERR_RET(c)                 \
+  do {                                \
+    int32_t _code = c;                \
+    if (_code != TSDB_CODE_SUCCESS) { \
+      terrno = _code;                 \
+      return _code;                   \
+    }                                 \
+  } while (0)
+#define QW_RET(c)                     \
+  do {                                \
+    int32_t _code = c;                \
+    if (_code != TSDB_CODE_SUCCESS) { \
+      terrno = _code;                 \
+    }                                 \
+    return _code;                     \
+  } while (0)
+#define QW_ERR_JRET(c)               \
+  do {                               \
+    code = c;                        \
+    if (code != TSDB_CODE_SUCCESS) { \
+      terrno = code;                 \
+      goto _return;                  \
+    }                                \
+  } while (0)
 
 #define QW_ELOG(param, ...) qError("QW:%p " param, mgmt, __VA_ARGS__)
 #define QW_DLOG(param, ...) qDebug("QW:%p " param, mgmt, __VA_ARGS__)
 
-#define QW_DUMP(param, ...) do { if (gQWDebug.dumpEnable) { qDebug("QW:%p " param, mgmt, __VA_ARGS__); } } while (0)
-
-
-#define QW_SCH_ELOG(param, ...) qError("QW:%p SID:%"PRIx64" " param, mgmt, sId, __VA_ARGS__)
-#define QW_SCH_DLOG(param, ...) qDebug("QW:%p SID:%"PRIx64" " param, mgmt, sId, __VA_ARGS__)
-
-#define QW_TASK_ELOG(param, ...) qError("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
-#define QW_TASK_WLOG(param, ...) qWarn("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
-#define QW_TASK_DLOG(param, ...) qDebug("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
-#define QW_TASK_DLOGL(param, ...) qDebugL("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
-
-#define QW_TASK_ELOG_E(param) qError("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId)
-#define QW_TASK_WLOG_E(param) qWarn("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId)
-#define QW_TASK_DLOG_E(param) qDebug("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId)
-
-#define QW_SCH_TASK_ELOG(param, ...) qError("QW:%p SID:0x%"PRIx64",QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)
-#define QW_SCH_TASK_WLOG(param, ...) qWarn("QW:%p SID:0x%"PRIx64",QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)
-#define QW_SCH_TASK_DLOG(param, ...) qDebug("QW:%p SID:0x%"PRIx64",QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)
-
-#define QW_LOCK_DEBUG(...) do { if (gQWDebug.lockEnable) { qDebug(__VA_ARGS__); } } while (0)
+#define QW_DUMP(param, ...)                      \
+  do {                                           \
+    if (gQWDebug.dumpEnable) {                   \
+      qDebug("QW:%p " param, mgmt, __VA_ARGS__); \
+    }                                            \
+  } while (0)
+
+#define QW_SCH_ELOG(param, ...) qError("QW:%p SID:%" PRIx64 " " param, mgmt, sId, __VA_ARGS__)
+#define QW_SCH_DLOG(param, ...) qDebug("QW:%p SID:%" PRIx64 " " param, mgmt, sId, __VA_ARGS__)
+
+#define QW_TASK_ELOG(param, ...) qError("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId, __VA_ARGS__)
+#define QW_TASK_WLOG(param, ...) qWarn("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId, __VA_ARGS__)
+#define QW_TASK_DLOG(param, ...) qDebug("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId, __VA_ARGS__)
+#define QW_TASK_DLOGL(param, ...) \
+  qDebugL("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId, __VA_ARGS__)
+
+#define QW_TASK_ELOG_E(param) qError("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId)
+#define QW_TASK_WLOG_E(param) qWarn("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId)
+#define QW_TASK_DLOG_E(param) qDebug("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId)
+
+#define QW_SCH_TASK_ELOG(param, ...) \
+  qError("QW:%p SID:0x%" PRIx64 ",QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, sId, qId, tId, __VA_ARGS__)
+#define QW_SCH_TASK_WLOG(param, ...) \
+  qWarn("QW:%p SID:0x%" PRIx64 ",QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, sId, qId, tId, __VA_ARGS__)
+#define QW_SCH_TASK_DLOG(param, ...) \
+  qDebug("QW:%p SID:0x%" PRIx64 ",QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, sId, qId, tId, __VA_ARGS__)
+
+#define QW_LOCK_DEBUG(...)     \
+  do {                         \
+    if (gQWDebug.lockEnable) { \
+      qDebug(__VA_ARGS__);     \
+    }                          \
+  } while (0)
 
 #define TD_RWLATCH_WRITE_FLAG_COPY 0x40000000
 
-#define QW_LOCK(type, _lock) do {   \
+#define QW_LOCK(type, _lock)                                                                       \
+  do {                                                                                             \
     if (QW_READ == (type)) {                                                                       \
       assert(atomic_load_32((_lock)) >= 0);                                                        \
       QW_LOCK_DEBUG("QW RLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
@@ -219,9 +263,10 @@ typedef struct SQWorkerMgmt {
       QW_LOCK_DEBUG("QW WLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
       assert(atomic_load_32((_lock)) == TD_RWLATCH_WRITE_FLAG_COPY);                               \
     }                                                                                              \
-} while (0)
+  } while (0)
 
-#define QW_UNLOCK(type, _lock) do {                       \
+#define QW_UNLOCK(type, _lock)                                                                      \
+  do {                                                                                              \
     if (QW_READ == (type)) {                                                                        \
       assert(atomic_load_32((_lock)) > 0);                                                          \
       QW_LOCK_DEBUG("QW RULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
@@ -235,7 +280,7 @@ typedef struct SQWorkerMgmt {
       QW_LOCK_DEBUG("QW WULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
       assert(atomic_load_32((_lock)) >= 0);                                                         \
     }                                                                                               \
-} while (0)
+  } while (0)
 
 #ifdef __cplusplus
 }

source/libs/qworker/src/qworkerMsg.c

-#include "qworker.h"
-#include "tcommon.h"
+#include "qworkerMsg.h"
+#include "dataSinkMgt.h"
 #include "executor.h"
 #include "planner.h"
 #include "query.h"
+#include "qworker.h"
 #include "qworkerInt.h"
-#include "qworkerMsg.h"
+#include "tcommon.h"
 #include "tmsg.h"
 #include "tname.h"
-#include "dataSinkMgt.h"
-
 
 int32_t qwMallocFetchRsp(int32_t length, SRetrieveTableRsp **rsp) {
   int32_t msgSize = sizeof(SRetrieveTableRsp) + length;
@@ -26,8 +25,6 @@ int32_t qwMallocFetchRsp(int32_t length, SRetrieveTableRsp **rsp) {
   return TSDB_CODE_SUCCESS;
 }
 
-
-
 void qwBuildFetchRsp(void *msg, SOutputData *input, int32_t len, bool qComplete) {
   SRetrieveTableRsp *rsp = (SRetrieveTableRsp *)msg;
 
@@ -39,7 +36,6 @@ void qwBuildFetchRsp(void *msg, SOutputData *input, int32_t len, bool qComplete)
   rsp->numOfRows = htonl(input->numOfRows);
 }
 
-
 void qwFreeFetchRsp(void *msg) {
   if (msg) {
     rpcFreeCont(msg);
@@ -50,13 +46,14 @@ int32_t qwBuildAndSendQueryRsp(SQWConnInfo *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 = {
       .msgType = TDMT_VND_QUERY_RSP,
       .handle = pConn->handle,
       .ahandle = pConn->ahandle,
+      .refId = pConn->refId,
       .pCont = msg,
       .contLen = contLen,
       .code = code,
@@ -74,6 +71,7 @@ int32_t qwBuildAndSendReadyRsp(SQWConnInfo *pConn, int32_t code) {
   SRpcMsg rpcRsp = {
       .msgType = TDMT_VND_RES_READY_RSP,
       .handle = pConn->handle,
+      .refId = pConn->refId,
       .ahandle = NULL,
       .pCont = pRsp,
       .contLen = sizeof(*pRsp),
@@ -89,13 +87,14 @@ int32_t qwBuildAndSendExplainRsp(SQWConnInfo *pConn, SExplainExecInfo *execInfo,
   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 = {
       .msgType = TDMT_VND_EXPLAIN_RSP,
       .handle = pConn->handle,
       .ahandle = pConn->ahandle,
+      .refId = pConn->refId,
       .pCont = pRsp,
       .contLen = contLen,
       .code = 0,
@@ -108,13 +107,14 @@ int32_t qwBuildAndSendExplainRsp(SQWConnInfo *pConn, SExplainExecInfo *execInfo,
 
 int32_t qwBuildAndSendHbRsp(SQWConnInfo *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 = {
       .msgType = TDMT_VND_QUERY_HEARTBEAT_RSP,
       .handle = pConn->handle,
       .ahandle = pConn->ahandle,
+      .refId = pConn->refId,
       .pCont = pRsp,
       .contLen = contLen,
       .code = code,
@@ -136,6 +136,7 @@ int32_t qwBuildAndSendFetchRsp(SQWConnInfo *pConn, SRetrieveTableRsp *pRsp, int3
       .msgType = TDMT_VND_FETCH_RSP,
       .handle = pConn->handle,
       .ahandle = pConn->ahandle,
+      .refId = pConn->refId,
       .pCont = pRsp,
       .contLen = sizeof(*pRsp) + dataLength,
       .code = code,
@@ -154,6 +155,7 @@ int32_t qwBuildAndSendCancelRsp(SQWConnInfo *pConn, int32_t code) {
       .msgType = TDMT_VND_CANCEL_TASK_RSP,
       .handle = pConn->handle,
       .ahandle = pConn->ahandle,
+      .refId = pConn->refId,
       .pCont = pRsp,
       .contLen = sizeof(*pRsp),
       .code = code,
@@ -171,6 +173,7 @@ int32_t qwBuildAndSendDropRsp(SQWConnInfo *pConn, int32_t code) {
       .msgType = TDMT_VND_DROP_TASK_RSP,
       .handle = pConn->handle,
       .ahandle = pConn->ahandle,
+      .refId = pConn->refId,
       .pCont = pRsp,
       .contLen = sizeof(*pRsp),
       .code = code,
@@ -220,12 +223,13 @@ 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 = {
       .handle = pMsg->handle,
       .ahandle = pMsg->ahandle,
+      .refId = pMsg->refId,
       .pCont = pBuf,
       .contLen = bufLen,
       .code = code,
@@ -235,7 +239,7 @@ int32_t qwBuildAndSendShowRsp(SRpcMsg *pMsg, int32_t code) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t qwBuildAndSendShowFetchRsp(SRpcMsg *pMsg, SVShowTablesFetchReq* pFetchReq) {
+int32_t qwBuildAndSendShowFetchRsp(SRpcMsg *pMsg, SVShowTablesFetchReq *pFetchReq) {
   SVShowTablesFetchRsp *pRsp = (SVShowTablesFetchRsp *)rpcMallocCont(sizeof(SVShowTablesFetchRsp));
   int32_t               handle = htonl(pFetchReq->id);
 
@@ -243,6 +247,7 @@ int32_t qwBuildAndSendShowFetchRsp(SRpcMsg *pMsg, SVShowTablesFetchReq* pFetchRe
   SRpcMsg rpcMsg = {
       .handle = pMsg->handle,
       .ahandle = pMsg->ahandle,
+      .refId = pMsg->refId,
       .pCont = pRsp,
       .contLen = sizeof(*pRsp),
       .code = 0,
@@ -253,7 +258,7 @@ int32_t qwBuildAndSendShowFetchRsp(SRpcMsg *pMsg, SVShowTablesFetchReq* pFetchRe
 }
 
 int32_t qwBuildAndSendCQueryMsg(QW_FPARAMS_DEF, SQWConnInfo *pConn) {
-  SQueryContinueReq * req = (SQueryContinueReq *)rpcMallocCont(sizeof(SQueryContinueReq));
+  SQueryContinueReq *req = (SQueryContinueReq *)rpcMallocCont(sizeof(SQueryContinueReq));
   if (NULL == req) {
     QW_SCH_TASK_ELOG("rpcMallocCont %d failed", (int32_t)sizeof(SQueryContinueReq));
     QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
@@ -268,6 +273,7 @@ int32_t qwBuildAndSendCQueryMsg(QW_FPARAMS_DEF, SQWConnInfo *pConn) {
       .handle = pConn->handle,
       .ahandle = pConn->ahandle,
       .msgType = TDMT_VND_QUERY_CONTINUE,
+      .refId = pConn->refId,
       .pCont = req,
       .contLen = sizeof(SQueryContinueReq),
       .code = 0,
@@ -285,9 +291,8 @@ int32_t qwBuildAndSendCQueryMsg(QW_FPARAMS_DEF, SQWConnInfo *pConn) {
   return TSDB_CODE_SUCCESS;
 }
 
-
 int32_t qwRegisterQueryBrokenLinkArg(QW_FPARAMS_DEF, SQWConnInfo *pConn) {
-  STaskDropReq * req = (STaskDropReq *)rpcMallocCont(sizeof(STaskDropReq));
+  STaskDropReq *req = (STaskDropReq *)rpcMallocCont(sizeof(STaskDropReq));
   if (NULL == req) {
     QW_SCH_TASK_ELOG("rpcMallocCont %d failed", (int32_t)sizeof(STaskDropReq));
     QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
@@ -302,6 +307,7 @@ int32_t qwRegisterQueryBrokenLinkArg(QW_FPARAMS_DEF, SQWConnInfo *pConn) {
   SRpcMsg pMsg = {
       .handle = pConn->handle,
       .ahandle = pConn->ahandle,
+      .refId = pConn->refId,
       .msgType = TDMT_VND_DROP_TASK,
       .pCont = req,
       .contLen = sizeof(STaskDropReq),
@@ -337,6 +343,7 @@ int32_t qwRegisterHbBrokenLinkArg(SQWorkerMgmt *mgmt, uint64_t sId, SQWConnInfo
   SRpcMsg pMsg = {
       .handle = pConn->handle,
       .ahandle = pConn->ahandle,
+      .refId = pConn->refId,
       .msgType = TDMT_VND_QUERY_HEARTBEAT,
       .pCont = msg,
       .contLen = msgSize,
@@ -348,8 +355,6 @@ int32_t qwRegisterHbBrokenLinkArg(SQWorkerMgmt *mgmt, uint64_t sId, SQWConnInfo
   return TSDB_CODE_SUCCESS;
 }
 
-
-
 int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   if (NULL == node || NULL == qWorkerMgmt || NULL == pMsg) {
     QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
@@ -379,8 +384,9 @@ int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   SQWMsg qwMsg = {.node = node, .msg = msg->msg + msg->sqlLen, .msgLen = msg->phyLen};
   qwMsg.connInfo.handle = pMsg->handle;
   qwMsg.connInfo.ahandle = pMsg->ahandle;
+  qwMsg.connInfo.refId = pMsg->refId;
 
-  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->handle, sql);
   taosMemoryFreeClear(sql);
 
@@ -397,8 +403,8 @@ int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   bool               queryDone = false;
   SQueryContinueReq *msg = (SQueryContinueReq *)pMsg->pCont;
   bool               needStop = false;
-  SQWTaskCtx *handles = NULL;
-  SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt;
+  SQWTaskCtx *       handles = NULL;
+  SQWorkerMgmt *     mgmt = (SQWorkerMgmt *)qWorkerMgmt;
 
   if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
     QW_ELOG("invalid cquery msg, msg:%p, msgLen:%d", msg, pMsg->contLen);
@@ -413,6 +419,7 @@ int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0};
   qwMsg.connInfo.handle = pMsg->handle;
   qwMsg.connInfo.ahandle = pMsg->ahandle;
+  qwMsg.connInfo.refId = pMsg->refId;
 
   QW_SCH_TASK_DLOG("processCQuery start, node:%p, handle:%p", node, pMsg->handle);
 
@@ -423,7 +430,7 @@ int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t qWorkerProcessReadyMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg){
+int32_t qWorkerProcessReadyMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   if (NULL == node || NULL == qWorkerMgmt || NULL == pMsg) {
     return TSDB_CODE_QRY_INVALID_INPUT;
   }
@@ -447,6 +454,7 @@ int32_t qWorkerProcessReadyMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg){
   SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0};
   qwMsg.connInfo.handle = pMsg->handle;
   qwMsg.connInfo.ahandle = pMsg->ahandle;
+  qwMsg.connInfo.refId = pMsg->refId;
 
   QW_SCH_TASK_DLOG("processReady start, node:%p, handle:%p", node, pMsg->handle);
 
@@ -475,11 +483,11 @@ int32_t qWorkerProcessStatusMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
 
   SSchedulerStatusRsp *sStatus = NULL;
 
-  //QW_ERR_JRET(qwGetSchTasksStatus(qWorkerMgmt, msg->sId, &sStatus));
+  // QW_ERR_JRET(qwGetSchTasksStatus(qWorkerMgmt, msg->sId, &sStatus));
 
 _return:
 
-  //QW_ERR_RET(qwBuildAndSendStatusRsp(pMsg, sStatus));
+  // QW_ERR_RET(qwBuildAndSendStatusRsp(pMsg, sStatus));
 
   return TSDB_CODE_SUCCESS;
 }
@@ -509,6 +517,7 @@ int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0};
   qwMsg.connInfo.handle = pMsg->handle;
   qwMsg.connInfo.ahandle = pMsg->ahandle;
+  qwMsg.connInfo.refId = pMsg->refId;
 
   QW_SCH_TASK_DLOG("processFetch start, node:%p, handle:%p", node, pMsg->handle);
 
@@ -529,7 +538,7 @@ int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
     return TSDB_CODE_QRY_INVALID_INPUT;
   }
 
-  SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt;
+  SQWorkerMgmt *  mgmt = (SQWorkerMgmt *)qWorkerMgmt;
   int32_t         code = 0;
   STaskCancelReq *msg = pMsg->pCont;
   if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
@@ -550,8 +559,9 @@ int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0};
   qwMsg.connInfo.handle = pMsg->handle;
   qwMsg.connInfo.ahandle = pMsg->ahandle;
+  qwMsg.connInfo.refId = pMsg->refId;
 
-  //QW_ERR_JRET(qwCancelTask(qWorkerMgmt, msg->sId, msg->queryId, msg->taskId));
+  // QW_ERR_JRET(qwCancelTask(qWorkerMgmt, msg->sId, msg->queryId, msg->taskId));
 
 _return:
 
@@ -588,6 +598,7 @@ int32_t qWorkerProcessDropMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0, .code = pMsg->code};
   qwMsg.connInfo.handle = pMsg->handle;
   qwMsg.connInfo.ahandle = pMsg->ahandle;
+  qwMsg.connInfo.refId = pMsg->refId;
 
   if (TSDB_CODE_RPC_NETWORK_UNAVAIL == pMsg->code) {
     QW_SCH_TASK_DLOG("receive drop task due to network broken, error:%s", tstrerror(pMsg->code));
@@ -609,7 +620,7 @@ int32_t qWorkerProcessHbMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
 
   int32_t         code = 0;
   SSchedulerHbReq req = {0};
-  SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt;
+  SQWorkerMgmt *  mgmt = (SQWorkerMgmt *)qWorkerMgmt;
 
   if (NULL == pMsg->pCont) {
     QW_ELOG("invalid hb msg, msg:%p, msgLen:%d", pMsg->pCont, pMsg->contLen);
@@ -626,6 +637,7 @@ int32_t qWorkerProcessHbMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   SQWMsg   qwMsg = {.node = node, .msg = NULL, .msgLen = 0, .code = pMsg->code};
   qwMsg.connInfo.handle = pMsg->handle;
   qwMsg.connInfo.ahandle = pMsg->ahandle;
+  qwMsg.connInfo.refId = pMsg->refId;
 
   if (TSDB_CODE_RPC_NETWORK_UNAVAIL == pMsg->code) {
     QW_SCH_DLOG("receive Hb msg due to network broken, error:%s", tstrerror(pMsg->code));

source/libs/scheduler/src/scheduler.c

@@ -369,7 +369,7 @@ int32_t schBuildTaskRalation(SSchJob *pJob, SHashObj *planToTask) {
       }
 
       for (int32_t n = 0; n < childNum; ++n) {
-        SSubplan  *child = (SSubplan *)nodesListGetNode(pPlan->pChildren, n);
+        SSubplan * child = (SSubplan *)nodesListGetNode(pPlan->pChildren, n);
         SSchTask **childTask = taosHashGet(planToTask, &child, POINTER_BYTES);
         if (NULL == childTask || NULL == *childTask) {
           SCH_TASK_ELOG("subplan children relationship error, level:%d, taskIdx:%d, childIdx:%d", i, m, n);
@@ -401,7 +401,7 @@ int32_t schBuildTaskRalation(SSchJob *pJob, SHashObj *planToTask) {
       }
 
       for (int32_t n = 0; n < parentNum; ++n) {
-        SSubplan  *parent = (SSubplan *)nodesListGetNode(pPlan->pParents, n);
+        SSubplan * parent = (SSubplan *)nodesListGetNode(pPlan->pParents, n);
         SSchTask **parentTask = taosHashGet(planToTask, &parent, POINTER_BYTES);
         if (NULL == parentTask || NULL == *parentTask) {
           SCH_TASK_ELOG("subplan parent relationship error, level:%d, taskIdx:%d, childIdx:%d", i, m, n);
@@ -491,7 +491,7 @@ int32_t schValidateAndBuildJob(SQueryPlan *pDag, SSchJob *pJob) {
   SSchLevel      level = {0};
   SNodeListNode *plans = NULL;
   int32_t        taskNum = 0;
-  SSchLevel     *pLevel = NULL;
+  SSchLevel *    pLevel = NULL;
 
   level.status = JOB_TASK_STATUS_NOT_START;
 
@@ -1267,7 +1267,7 @@ int32_t schUpdateTaskExecNodeHandle(SSchTask *pTask, void *handle, int32_t rspCo
 int32_t schHandleCallback(void *param, const SDataBuf *pMsg, int32_t msgType, int32_t rspCode) {
   int32_t                code = 0;
   SSchTaskCallbackParam *pParam = (SSchTaskCallbackParam *)param;
-  SSchTask              *pTask = NULL;
+  SSchTask *             pTask = NULL;
 
   SSchJob *pJob = schAcquireJob(pParam->refId);
   if (NULL == pJob) {
@@ -1617,8 +1617,8 @@ _return:
 int32_t schMakeHbRpcCtx(SSchJob *pJob, SSchTask *pTask, SRpcCtx *pCtx) {
   int32_t              code = 0;
   SSchHbCallbackParam *param = NULL;
-  SMsgSendInfo        *pMsgSendInfo = NULL;
-  SQueryNodeAddr      *addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx);
+  SMsgSendInfo *       pMsgSendInfo = NULL;
+  SQueryNodeAddr *     addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx);
   SQueryNodeEpId       epId = {0};
 
   epId.nodeId = addr->nodeId;
@@ -1759,10 +1759,10 @@ int32_t schCloneHbRpcCtx(SRpcCtx *pSrc, SRpcCtx *pDst) {
   }
 
   SRpcCtxVal dst = {0};
-  void      *pIter = taosHashIterate(pSrc->args, NULL);
+  void *     pIter = taosHashIterate(pSrc->args, NULL);
   while (pIter) {
     SRpcCtxVal *pVal = (SRpcCtxVal *)pIter;
-    int32_t    *msgType = taosHashGetKey(pIter, NULL);
+    int32_t *   msgType = taosHashGetKey(pIter, NULL);
 
     dst = *pVal;
     dst.val = NULL;
@@ -1916,7 +1916,7 @@ _return:
 
 int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr, int32_t msgType) {
   uint32_t msgSize = 0;
-  void    *msg = NULL;
+  void *   msg = NULL;
   int32_t  code = 0;
   bool     isCandidateAddr = false;
   bool     persistHandle = false;
@@ -2673,7 +2673,7 @@ int32_t schedulerGetTasksStatus(int64_t job, SArray *pSub) {
     SSchLevel *pLevel = taosArrayGet(pJob->levels, i);
 
     for (int32_t m = 0; m < pLevel->taskNum; ++m) {
-      SSchTask     *pTask = taosArrayGet(pLevel->subTasks, m);
+      SSchTask *    pTask = taosArrayGet(pLevel->subTasks, m);
       SQuerySubDesc subDesc = {.tid = pTask->taskId, .status = pTask->status};
 
       taosArrayPush(pSub, &subDesc);

source/libs/sync/src/syncEnv.c

@@ -29,7 +29,8 @@ static void      syncEnvTick(void *param, void *tmrId);
 int32_t syncEnvStart() {
   int32_t ret = 0;
   taosSeedRand(taosGetTimestampSec());
-  gSyncEnv = doSyncEnvStart(gSyncEnv);
+  //gSyncEnv = doSyncEnvStart(gSyncEnv);
+  gSyncEnv = doSyncEnvStart();
   assert(gSyncEnv != NULL);
   sTrace("syncEnvStart ok!");
   return ret;
@@ -91,8 +92,12 @@ static SSyncEnv *doSyncEnvStart() {
 }
 
 static int32_t doSyncEnvStop(SSyncEnv *pSyncEnv) {
+  assert(pSyncEnv == gSyncEnv);
+  if (pSyncEnv != NULL) {
     taosTmrCleanUp(pSyncEnv->pTimerManager);
     taosMemoryFree(pSyncEnv);
+  }
+  gSyncEnv = NULL;
   return 0;
 }
 

source/libs/sync/src/syncRaftLog.c

@@ -71,7 +71,7 @@ SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
     SWalReadHandle* pWalHandle = walOpenReadHandle(pWal);
     assert(walReadWithHandle(pWalHandle, index) == 0);
 
-    SSyncRaftEntry* pEntry = syncEntryBuild(pWalHandle->pHead->head.len);
+    SSyncRaftEntry* pEntry = syncEntryBuild(pWalHandle->pHead->head.bodyLen);
     assert(pEntry != NULL);
 
     pEntry->msgType = TDMT_VND_SYNC_CLIENT_REQUEST;
@@ -80,8 +80,8 @@ SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
     pEntry->isWeak = pWalHandle->pHead->head.syncMeta.isWeek;
     pEntry->term = pWalHandle->pHead->head.syncMeta.term;
     pEntry->index = index;
-    assert(pEntry->dataLen == pWalHandle->pHead->head.len);
-    memcpy(pEntry->data, pWalHandle->pHead->head.body, pWalHandle->pHead->head.len);
+    assert(pEntry->dataLen == pWalHandle->pHead->head.bodyLen);
+    memcpy(pEntry->data, pWalHandle->pHead->head.body, pWalHandle->pHead->head.bodyLen);
 
     // need to hold, do not new every time!!
     walCloseReadHandle(pWalHandle);

source/libs/transport/inc/transComm.h

@@ -327,6 +327,10 @@ void transQueueClear(STransQueue* queue);
  */
 void transQueueDestroy(STransQueue* queue);
 
+/*
+ * init global func
+ */
+void transThreadOnce();
 #ifdef __cplusplus
 }
 #endif

source/libs/transport/src/transComm.c

@@ -16,6 +16,8 @@
 
 #include "transComm.h"
 
+// static TdThreadOnce transModuleInit = PTHREAD_ONCE_INIT;
+
 int transAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey) {
   T_MD5_CTX context;
   int       ret = -1;
@@ -361,5 +363,10 @@ void transQueueDestroy(STransQueue* queue) {
   transQueueClear(queue);
   taosArrayDestroy(queue->q);
 }
-
+// int32_t transGetExHandle() {
+//  static
+//}
+// void transThreadOnce() {
+// taosThreadOnce(&transModuleInit, );
+//}
 #endif

source/libs/transport/src/transSrv.c

@@ -17,6 +17,11 @@
 
 #include "transComm.h"
 
+static TdThreadOnce transModuleInit = PTHREAD_ONCE_INIT;
+
+static char* notify = "a";
+static int   transSrvInst = 0;
+
 typedef struct {
   int       notifyCount;  //
   int       init;         // init or not
@@ -45,6 +50,7 @@ typedef struct SSrvConn {
   struct sockaddr_in addr;
   struct sockaddr_in locaddr;
 
+  int64_t refId;
   int     spi;
   char    info[64];
   char    user[TSDB_UNI_LEN];  // user ID for the link
@@ -89,7 +95,12 @@ typedef struct SServerObj {
   uv_async_t* pAcceptAsync;  // just to quit from from accept thread
 } SServerObj;
 
-static const char* notify = "a";
+// handle
+typedef struct SExHandle {
+  void*         handle;
+  int64_t       refId;
+  SWorkThrdObj* pThrd;
+} SExHandle;
 
 static void uvAllocConnBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);
 static void uvAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);
@@ -130,6 +141,17 @@ static void uvHandleRegister(SSrvMsg* msg, SWorkThrdObj* thrd);
 static void (*transAsyncHandle[])(SSrvMsg* msg, SWorkThrdObj* thrd) = {uvHandleResp, uvHandleQuit, uvHandleRelease,
                                                                        uvHandleRegister};
 
+static int32_t exHandlesMgt;
+
+void       uvInitExHandleMgt();
+void       uvOpenExHandleMgt(int size);
+void       uvCloseExHandleMgt();
+int64_t    uvAddExHandle(void* p);
+int32_t    uvRemoveExHandle(int64_t refId);
+int32_t    uvReleaseExHandle(int64_t refId);
+void       uvDestoryExHandle(void* handle);
+SExHandle* uvAcquireExHandle(int64_t refId);
+
 static void uvDestroyConn(uv_handle_t* handle);
 
 // server and worker thread
@@ -168,6 +190,40 @@ static bool addHandleToAcceptloop(void* arg);
     uv_loop_close(loop);           \
   } while (0);
 
+#define ASYNC_ERR_JRET(thrd)                            \
+  do {                                                  \
+    if (thrd->quit) {                                   \
+      tTrace("worker thread already quit, ignore msg"); \
+      goto _return1;                                    \
+    }                                                   \
+  } while (0)
+
+#define ASYNC_CHECK_HANDLE(exh1, refId)                                                                               \
+  do {                                                                                                                \
+    if (refId > 0) {                                                                                                  \
+      tTrace("server handle step1");                                                                                  \
+      SExHandle* exh2 = uvAcquireExHandle(refId);                                                                     \
+      if (exh2 == NULL || refId != exh2->refId) {                                                                     \
+        tTrace("server handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, \
+               exh1->refId, refId);                                                                                   \
+        goto _return1;                                                                                                \
+      }                                                                                                               \
+    } else if (refId == 0) {                                                                                          \
+      tTrace("server handle step2");                                                                                  \
+      SExHandle* exh2 = uvAcquireExHandle(refId);                                                                     \
+      if (exh2 == NULL || refId != exh2->refId) {                                                                     \
+        tTrace("server handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, \
+               refId, exh2 ? exh2->refId : 0);                                                                        \
+        goto _return1;                                                                                                \
+      } else {                                                                                                        \
+        refId = exh1->refId;                                                                                          \
+      }                                                                                                               \
+    } else if (refId == -1) {                                                                                         \
+      tTrace("server handle step3");                                                                                  \
+      goto _return2;                                                                                                  \
+    }                                                                                                                 \
+  } while (0)
+
 void uvAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
   SSrvConn*    conn = handle->data;
   SConnBuffer* pBuf = &conn->readBuf;
@@ -233,7 +289,15 @@ static void uvHandleReq(SSrvConn* pConn) {
   // 1. server application should not send resp on handle
   // 2. once send out data, cli conn released to conn pool immediately
   // 3. not mixed with persist
-  transMsg.handle = pConn;
+
+  transMsg.handle = (void*)uvAcquireExHandle(pConn->refId);
+  tTrace("server handle %p conn: %p translated to app, refId: %" PRIu64 "", transMsg.handle, pConn, pConn->refId);
+  transMsg.refId = pConn->refId;
+  assert(transMsg.handle != NULL);
+  if (pHead->noResp == 1) {
+    transMsg.refId = -1;
+  }
+  uvReleaseExHandle(pConn->refId);
 
   STrans* pTransInst = pConn->pTransInst;
   (*pTransInst->cfp)(pTransInst->parent, &transMsg, NULL);
@@ -404,6 +468,7 @@ static void destroySmsg(SSrvMsg* smsg) {
   taosMemoryFree(smsg);
 }
 static void destroyAllConn(SWorkThrdObj* pThrd) {
+  tTrace("thread %p destroy all conn ", pThrd);
   while (!QUEUE_IS_EMPTY(&pThrd->conn)) {
     queue* h = QUEUE_HEAD(&pThrd->conn);
     QUEUE_REMOVE(h);
@@ -436,7 +501,26 @@ void uvWorkerAsyncCb(uv_async_t* handle) {
       tError("unexcept occurred, continue");
       continue;
     }
+    // release handle to rpc init
+    if (msg->type == Quit) {
       (*transAsyncHandle[msg->type])(msg, pThrd);
+      continue;
+    } else {
+      STransMsg transMsg = msg->msg;
+
+      SExHandle* exh1 = transMsg.handle;
+      int64_t    refId = transMsg.refId;
+      SExHandle* exh2 = uvAcquireExHandle(refId);
+      if (exh2 == NULL || exh1 != exh2) {
+        tTrace("server handle %p except msg, ignore it", exh1);
+        uvReleaseExHandle(refId);
+        destroySmsg(msg);
+        continue;
+      }
+      msg->pConn = exh1->handle;
+      uvReleaseExHandle(refId);
+      (*transAsyncHandle[msg->type])(msg, pThrd);
+    }
   }
 }
 static void uvWalkCb(uv_handle_t* handle, void* arg) {
@@ -658,8 +742,15 @@ static SSrvConn* createConn(void* hThrd) {
   pConn->broken = false;
   pConn->status = ConnNormal;
 
+  SExHandle* exh = taosMemoryMalloc(sizeof(SExHandle));
+  exh->handle = pConn;
+  exh->pThrd = pThrd;
+  exh->refId = uvAddExHandle(exh);
+  uvAcquireExHandle(exh->refId);
+
+  pConn->refId = exh->refId;
   transRefSrvHandle(pConn);
-  tTrace("server conn %p created", pConn);
+  tTrace("server handle %p, conn %p created, refId: %" PRId64 "", exh, pConn, pConn->refId);
   return pConn;
 }
 
@@ -667,6 +758,7 @@ static void destroyConn(SSrvConn* conn, bool clear) {
   if (conn == NULL) {
     return;
   }
+
   transDestroyBuffer(&conn->readBuf);
   if (clear) {
     tTrace("server conn %p to be destroyed", conn);
@@ -681,6 +773,9 @@ static void uvDestroyConn(uv_handle_t* handle) {
   }
   SWorkThrdObj* thrd = conn->hostThrd;
 
+  uvReleaseExHandle(conn->refId);
+  uvRemoveExHandle(conn->refId);
+
   tDebug("server conn %p destroy", conn);
   // uv_timer_stop(&conn->pTimer);
   transQueueDestroy(&conn->srvMsgs);
@@ -705,6 +800,10 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
   srv->port = port;
   uv_loop_init(srv->loop);
 
+  taosThreadOnce(&transModuleInit, uvInitExHandleMgt);
+  transSrvInst++;
+  // uvOpenExHandleMgt(10000);
+
   for (int i = 0; i < srv->numOfThreads; i++) {
     SWorkThrdObj* thrd = (SWorkThrdObj*)taosMemoryCalloc(1, sizeof(SWorkThrdObj));
     thrd->quit = false;
@@ -749,6 +848,44 @@ End:
   transCloseServer(srv);
   return NULL;
 }
+
+void uvInitExHandleMgt() {
+  // init exhandle mgt
+  uvOpenExHandleMgt(10000);
+}
+void uvOpenExHandleMgt(int size) {
+  // added into once later
+  exHandlesMgt = taosOpenRef(size, uvDestoryExHandle);
+}
+void uvCloseExHandleMgt() {
+  // close ref
+  taosCloseRef(exHandlesMgt);
+}
+int64_t uvAddExHandle(void* p) {
+  // acquire extern handle
+  return taosAddRef(exHandlesMgt, p);
+}
+int32_t uvRemoveExHandle(int64_t refId) {
+  // acquire extern handle
+  return taosRemoveRef(exHandlesMgt, refId);
+}
+
+SExHandle* uvAcquireExHandle(int64_t refId) {
+  // acquire extern handle
+  return (SExHandle*)taosAcquireRef(exHandlesMgt, refId);
+}
+
+int32_t uvReleaseExHandle(int64_t refId) {
+  // release extern handle
+  return taosReleaseRef(exHandlesMgt, refId);
+}
+void uvDestoryExHandle(void* handle) {
+  if (handle == NULL) {
+    return;
+  }
+  taosMemoryFree(handle);
+}
+
 void uvHandleQuit(SSrvMsg* msg, SWorkThrdObj* thrd) {
   thrd->quit = true;
   if (QUEUE_IS_EMPTY(&thrd->conn)) {
@@ -759,7 +896,6 @@ void uvHandleQuit(SSrvMsg* msg, SWorkThrdObj* thrd) {
   taosMemoryFree(msg);
 }
 void uvHandleRelease(SSrvMsg* msg, SWorkThrdObj* thrd) {
-  // release handle to rpc init
   SSrvConn* conn = msg->pConn;
   if (conn->status == ConnAcquire) {
     if (!transQueuePush(&conn->srvMsgs, msg)) {
@@ -840,6 +976,12 @@ void transCloseServer(void* arg) {
   taosMemoryFree(srv->pipe);
 
   taosMemoryFree(srv);
+
+  transSrvInst--;
+  if (transSrvInst == 0) {
+    transModuleInit = PTHREAD_ONCE_INIT;
+    uvCloseExHandleMgt();
+  }
 }
 
 void transRefSrvHandle(void* handle) {
@@ -862,57 +1004,98 @@ void transUnrefSrvHandle(void* handle) {
 }
 
 void transReleaseSrvHandle(void* handle) {
-  if (handle == NULL) {
-    return;
-  }
-  SSrvConn*     pConn = handle;
-  SWorkThrdObj* pThrd = pConn->hostThrd;
+  SExHandle* exh = handle;
+  int64_t    refId = exh->refId;
+
+  ASYNC_CHECK_HANDLE(exh, refId);
+
+  SWorkThrdObj* pThrd = exh->pThrd;
+  ASYNC_ERR_JRET(pThrd);
 
-  STransMsg tmsg = {.code = 0, .handle = handle, .ahandle = NULL};
+  STransMsg tmsg = {.code = 0, .handle = exh, .ahandle = NULL, .refId = refId};
 
   SSrvMsg* srvMsg = taosMemoryCalloc(1, sizeof(SSrvMsg));
   srvMsg->msg = tmsg;
   srvMsg->type = Release;
-  srvMsg->pConn = pConn;
 
-  tTrace("server conn %p start to release", pConn);
+  tTrace("server conn %p start to release", exh->handle);
   transSendAsync(pThrd->asyncPool, &srvMsg->q);
-}
-void transSendResponse(const STransMsg* pMsg) {
-  if (pMsg->handle == NULL) {
+  uvReleaseExHandle(refId);
   return;
-  }
-  SSrvConn*     pConn = pMsg->handle;
-  SWorkThrdObj* pThrd = pConn->hostThrd;
-  if (pThrd->quit) {
+_return1:
+  tTrace("server handle %p failed to send to release handle", exh);
+  uvReleaseExHandle(refId);
   return;
-  }
+_return2:
+  tTrace("server handle %p failed to send to release handle", exh);
+  return;
+}
+void transSendResponse(const STransMsg* msg) {
+  SExHandle* exh = msg->handle;
+  int64_t    refId = msg->refId;
+  ASYNC_CHECK_HANDLE(exh, refId);
+  assert(refId != 0);
+
+  STransMsg tmsg = *msg;
+  tmsg.refId = refId;
+
+  SWorkThrdObj* pThrd = exh->pThrd;
+  ASYNC_ERR_JRET(pThrd);
 
   SSrvMsg* srvMsg = taosMemoryCalloc(1, sizeof(SSrvMsg));
-  srvMsg->pConn = pConn;
-  srvMsg->msg = *pMsg;
+  srvMsg->msg = tmsg;
   srvMsg->type = Normal;
-  tTrace("server conn %p start to send resp (1/2)", pConn);
+  tTrace("server conn %p start to send resp (1/2)", exh->handle);
   transSendAsync(pThrd->asyncPool, &srvMsg->q);
+  uvReleaseExHandle(refId);
+  return;
+_return1:
+  tTrace("server handle %p failed to send resp", exh);
+  rpcFreeCont(msg->pCont);
+  uvReleaseExHandle(refId);
+  return;
+_return2:
+  tTrace("server handle %p failed to send resp", exh);
+  rpcFreeCont(msg->pCont);
+  return;
 }
 void transRegisterMsg(const STransMsg* msg) {
-  if (msg->handle == NULL) {
-    return;
-  }
-  SSrvConn*     pConn = msg->handle;
-  SWorkThrdObj* pThrd = pConn->hostThrd;
+  SExHandle* exh = msg->handle;
+  int64_t    refId = msg->refId;
+  ASYNC_CHECK_HANDLE(exh, refId);
+
+  STransMsg tmsg = *msg;
+  tmsg.refId = refId;
+
+  SWorkThrdObj* pThrd = exh->pThrd;
+  ASYNC_ERR_JRET(pThrd);
 
   SSrvMsg* srvMsg = taosMemoryCalloc(1, sizeof(SSrvMsg));
-  srvMsg->pConn = pConn;
-  srvMsg->msg = *msg;
+  srvMsg->msg = tmsg;
   srvMsg->type = Register;
-  tTrace("server conn %p start to register brokenlink callback", pConn);
+  tTrace("server conn %p start to register brokenlink callback", exh->handle);
   transSendAsync(pThrd->asyncPool, &srvMsg->q);
+  uvReleaseExHandle(refId);
+  return;
+_return1:
+  tTrace("server handle %p failed to send to register brokenlink", exh);
+  rpcFreeCont(msg->pCont);
+  uvReleaseExHandle(refId);
+  return;
+_return2:
+  tTrace("server handle %p failed to send to register brokenlink", exh);
+  rpcFreeCont(msg->pCont);
 }
+
 int transGetConnInfo(void* thandle, STransHandleInfo* pInfo) {
-  SSrvConn*          pConn = thandle;
-  struct sockaddr_in addr = pConn->addr;
+  if (thandle == NULL) {
+    tTrace("invalid handle %p, failed to Get Conn info", thandle);
+    return -1;
+  }
+  SExHandle* ex = thandle;
+  SSrvConn*  pConn = ex->handle;
 
+  struct sockaddr_in addr = pConn->addr;
   pInfo->clientIp = (uint32_t)(addr.sin_addr.s_addr);
   pInfo->clientPort = ntohs(addr.sin_port);
   tstrncpy(pInfo->user, pConn->user, sizeof(pInfo->user));

source/libs/wal/inc/walInt.h

@@ -16,10 +16,10 @@
 #ifndef _TD_WAL_INT_H_
 #define _TD_WAL_INT_H_
 
-#include "tcompare.h"
 #include "taoserror.h"
 #include "tchecksum.h"
 #include "tcoding.h"
+#include "tcompare.h"
 #include "wal.h"
 
 #ifdef __cplusplus
@@ -41,10 +41,10 @@ typedef struct WalIdxEntry {
 } SWalIdxEntry;
 
 static inline int tSerializeWalIdxEntry(void** buf, SWalIdxEntry* pIdxEntry) {
-  int tlen;
+  int tlen = 0;
   tlen += taosEncodeFixedI64(buf, pIdxEntry->ver);
   tlen += taosEncodeFixedI64(buf, pIdxEntry->offset);
-  return 0;
+  return tlen;
 }
 
 static inline void* tDeserializeWalIdxEntry(void* buf, SWalIdxEntry* pIdxEntry) {
@@ -103,7 +103,7 @@ static inline int walValidHeadCksum(SWalHead* pHead) {
 }
 
 static inline int walValidBodyCksum(SWalHead* pHead) {
-  return taosCheckChecksum((uint8_t*)pHead->head.body, pHead->head.len, pHead->cksumBody);
+  return taosCheckChecksum((uint8_t*)pHead->head.body, pHead->head.bodyLen, pHead->cksumBody);
 }
 
 static inline int walValidCksum(SWalHead* pHead, void* body, int64_t bodyLen) {

source/libs/wal/src/walRead.c

@@ -144,12 +144,12 @@ int32_t walReadWithHandle_s(SWalReadHandle *pRead, int64_t ver, SWalReadHead **p
     taosThreadMutexUnlock(&pRead->mutex);
     return -1;
   }
-  *ppHead = taosMemoryMalloc(sizeof(SWalReadHead) + pRead->pHead->head.len);
+  *ppHead = taosMemoryMalloc(sizeof(SWalReadHead) + pRead->pHead->head.bodyLen);
   if (*ppHead == NULL) {
     taosThreadMutexUnlock(&pRead->mutex);
     return -1;
   }
-  memcpy(*ppHead, &pRead->pHead->head, sizeof(SWalReadHead) + pRead->pHead->head.len);
+  memcpy(*ppHead, &pRead->pHead->head, sizeof(SWalReadHead) + pRead->pHead->head.bodyLen);
   taosThreadMutexUnlock(&pRead->mutex);
   return 0;
 }
@@ -178,16 +178,18 @@ int32_t walReadWithHandle(SWalReadHandle *pRead, int64_t ver) {
     terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
     return -1;
   }
-  if (pRead->capacity < pRead->pHead->head.len) {
-    void *ptr = taosMemoryRealloc(pRead->pHead, sizeof(SWalHead) + pRead->pHead->head.len);
+  if (pRead->capacity < pRead->pHead->head.bodyLen) {
+    void *ptr = taosMemoryRealloc(pRead->pHead, sizeof(SWalHead) + pRead->pHead->head.bodyLen);
     if (ptr == NULL) {
       terrno = TSDB_CODE_WAL_OUT_OF_MEMORY;
       return -1;
     }
     pRead->pHead = ptr;
-    pRead->capacity = pRead->pHead->head.len;
+    pRead->capacity = pRead->pHead->head.bodyLen;
   }
-  if (pRead->pHead->head.len != taosReadFile(pRead->pReadLogTFile, pRead->pHead->head.body, pRead->pHead->head.len)) {
+
+  if (pRead->pHead->head.bodyLen !=
+      taosReadFile(pRead->pReadLogTFile, pRead->pHead->head.body, pRead->pHead->head.bodyLen)) {
     return -1;
   }
 

source/libs/wal/src/walWrite.c

@@ -295,7 +295,7 @@ int64_t walWriteWithSyncInfo(SWal *pWal, int64_t index, tmsg_t msgType, SSyncLog
   pWal->writeHead.head.version = index;
 
   int64_t offset = walGetCurFileOffset(pWal);
-  pWal->writeHead.head.len = bodyLen;
+  pWal->writeHead.head.bodyLen = bodyLen;
   pWal->writeHead.head.msgType = msgType;
 
   // sync info

source/libs/wal/test/walMetaTest.cpp

@@ -320,7 +320,7 @@ TEST_F(WalKeepEnv, readHandleRead) {
     char newStr[100];
     sprintf(newStr, "%s-%d", ranStr, ver);
     int len = strlen(newStr);
-    ASSERT_EQ(pRead->pHead->head.len, len);
+    ASSERT_EQ(pRead->pHead->head.bodyLen, len);
     for (int j = 0; j < len; j++) {
       EXPECT_EQ(newStr[j], pRead->pHead->head.body[j]);
     }
@@ -372,7 +372,7 @@ TEST_F(WalRetentionEnv, repairMeta1) {
     char newStr[100];
     sprintf(newStr, "%s-%d", ranStr, ver);
     int len = strlen(newStr);
-    ASSERT_EQ(pRead->pHead->head.len, len);
+    ASSERT_EQ(pRead->pHead->head.bodyLen, len);
     for (int j = 0; j < len; j++) {
       EXPECT_EQ(newStr[j], pRead->pHead->head.body[j]);
     }
@@ -402,7 +402,7 @@ TEST_F(WalRetentionEnv, repairMeta1) {
     char newStr[100];
     sprintf(newStr, "%s-%d", ranStr, ver);
     int len = strlen(newStr);
-    ASSERT_EQ(pRead->pHead->head.len, len);
+    ASSERT_EQ(pRead->pHead->head.bodyLen, len);
     for (int j = 0; j < len; j++) {
       EXPECT_EQ(newStr[j], pRead->pHead->head.body[j]);
     }

source/util/src/tconfig.c

@@ -139,12 +139,6 @@ static int32_t cfgCheckAndSetDir(SConfigItem *pItem, const char *inputDir) {
     return -1;
   }
 
-  if (taosRealPath(fullDir, NULL, PATH_MAX) != 0) {
-    terrno = TAOS_SYSTEM_ERROR(errno);
-    uError("failed to get realpath of dir:%s since %s", inputDir, terrstr());
-    return -1;
-  }
-
   taosMemoryFreeClear(pItem->str);
   pItem->str = strdup(fullDir);
   if (pItem->str == NULL) {
@@ -172,9 +166,8 @@ static int32_t cfgSetBool(SConfigItem *pItem, const char *value, ECfgSrcType sty
 static int32_t cfgSetInt32(SConfigItem *pItem, const char *value, ECfgSrcType stype) {
   int32_t ival = (int32_t)atoi(value);
   if (ival < pItem->imin || ival > pItem->imax) {
-    uError("cfg:%s, type:%s src:%s value:%d out of range[%" PRId64 ", %" PRId64 "], use last src:%s value:%d",
-           pItem->name, cfgDtypeStr(pItem->dtype), cfgStypeStr(stype), ival, pItem->imin, pItem->imax,
-           cfgStypeStr(pItem->stype), pItem->i32);
+    uError("cfg:%s, type:%s src:%s value:%d out of range[%" PRId64 ", %" PRId64 "]", pItem->name,
+           cfgDtypeStr(pItem->dtype), cfgStypeStr(stype), ival, pItem->imin, pItem->imax);
     terrno = TSDB_CODE_OUT_OF_RANGE;
     return -1;
   }
@@ -187,10 +180,8 @@ static int32_t cfgSetInt32(SConfigItem *pItem, const char *value, ECfgSrcType st
 static int32_t cfgSetInt64(SConfigItem *pItem, const char *value, ECfgSrcType stype) {
   int64_t ival = (int64_t)atoi(value);
   if (ival < pItem->imin || ival > pItem->imax) {
-    uError("cfg:%s, type:%s src:%s value:%" PRId64 " out of range[%" PRId64 ", %" PRId64
-           "], use last src:%s value:%" PRId64,
-           pItem->name, cfgDtypeStr(pItem->dtype), cfgStypeStr(stype), ival, pItem->imin, pItem->imax,
-           cfgStypeStr(pItem->stype), pItem->i64);
+    uError("cfg:%s, type:%s src:%s value:%" PRId64 " out of range[%" PRId64 ", %" PRId64 "]", pItem->name,
+           cfgDtypeStr(pItem->dtype), cfgStypeStr(stype), ival, pItem->imin, pItem->imax);
     terrno = TSDB_CODE_OUT_OF_RANGE;
     return -1;
   }
@@ -203,9 +194,8 @@ static int32_t cfgSetInt64(SConfigItem *pItem, const char *value, ECfgSrcType st
 static int32_t cfgSetFloat(SConfigItem *pItem, const char *value, ECfgSrcType stype) {
   float fval = (float)atof(value);
   if (fval < pItem->fmin || fval > pItem->fmax) {
-    uError("cfg:%s, type:%s src:%s value:%f out of range[%f, %f], use last src:%s value:%f", pItem->name,
-           cfgDtypeStr(pItem->dtype), cfgStypeStr(stype), fval, pItem->fmin, pItem->fmax, cfgStypeStr(pItem->stype),
-           pItem->fval);
+    uError("cfg:%s, type:%s src:%s value:%f out of range[%f, %f]", pItem->name, cfgDtypeStr(pItem->dtype),
+           cfgStypeStr(stype), fval, pItem->fmin, pItem->fmax);
     terrno = TSDB_CODE_OUT_OF_RANGE;
     return -1;
   }
@@ -219,8 +209,8 @@ static int32_t cfgSetString(SConfigItem *pItem, const char *value, ECfgSrcType s
   char *tmp = strdup(value);
   if (tmp == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
-    uError("cfg:%s, type:%s src:%s value:%s failed to dup since %s, use last src:%s value:%s", pItem->name,
-           cfgDtypeStr(pItem->dtype), cfgStypeStr(stype), value, terrstr(), cfgStypeStr(pItem->stype), pItem->str);
+    uError("cfg:%s, type:%s src:%s value:%s failed to dup since %s", pItem->name, cfgDtypeStr(pItem->dtype),
+           cfgStypeStr(stype), value, terrstr());
     return -1;
   }
 
@@ -232,9 +222,8 @@ static int32_t cfgSetString(SConfigItem *pItem, const char *value, ECfgSrcType s
 
 static int32_t cfgSetDir(SConfigItem *pItem, const char *value, ECfgSrcType stype) {
   if (cfgCheckAndSetDir(pItem, value) != 0) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    uError("cfg:%s, type:%s src:%s value:%s failed to dup since %s, use last src:%s value:%s", pItem->name,
-           cfgDtypeStr(pItem->dtype), cfgStypeStr(stype), value, terrstr(), cfgStypeStr(pItem->stype), pItem->str);
+    uError("cfg:%s, type:%s src:%s value:%s failed to dup since %s", pItem->name, cfgDtypeStr(pItem->dtype),
+           cfgStypeStr(stype), value, terrstr());
     return -1;
   }
 
@@ -245,8 +234,8 @@ static int32_t cfgSetDir(SConfigItem *pItem, const char *value, ECfgSrcType styp
 static int32_t cfgSetLocale(SConfigItem *pItem, const char *value, ECfgSrcType stype) {
   if (cfgCheckAndSetLocale(pItem, value) != 0) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
-    uError("cfg:%s, type:%s src:%s value:%s failed to dup since %s, use last src:%s value:%s", pItem->name,
-           cfgDtypeStr(pItem->dtype), cfgStypeStr(stype), value, terrstr(), cfgStypeStr(pItem->stype), pItem->str);
+    uError("cfg:%s, type:%s src:%s value:%s failed to dup since %s", pItem->name, cfgDtypeStr(pItem->dtype),
+           cfgStypeStr(stype), value, terrstr());
     return -1;
   }
 
@@ -257,8 +246,8 @@ static int32_t cfgSetLocale(SConfigItem *pItem, const char *value, ECfgSrcType s
 static int32_t cfgSetCharset(SConfigItem *pItem, const char *value, ECfgSrcType stype) {
   if (cfgCheckAndSetCharset(pItem, value) != 0) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
-    uError("cfg:%s, type:%s src:%s value:%s failed to dup since %s, use last src:%s value:%s", pItem->name,
-           cfgDtypeStr(pItem->dtype), cfgStypeStr(stype), value, terrstr(), cfgStypeStr(pItem->stype), pItem->str);
+    uError("cfg:%s, type:%s src:%s value:%s failed to dup since %s", pItem->name, cfgDtypeStr(pItem->dtype),
+           cfgStypeStr(stype), value, terrstr());
     return -1;
   }
 
@@ -269,8 +258,8 @@ static int32_t cfgSetCharset(SConfigItem *pItem, const char *value, ECfgSrcType
 static int32_t cfgSetTimezone(SConfigItem *pItem, const char *value, ECfgSrcType stype) {
   if (cfgCheckAndSetTimezone(pItem, value) != 0) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
-    uError("cfg:%s, type:%s src:%s value:%s failed to dup since %s, use last src:%s value:%s", pItem->name,
-           cfgDtypeStr(pItem->dtype), cfgStypeStr(stype), value, terrstr(), cfgStypeStr(pItem->stype), pItem->str);
+    uError("cfg:%s, type:%s src:%s value:%s failed to dup since %s", pItem->name, cfgDtypeStr(pItem->dtype),
+           cfgStypeStr(stype), value, terrstr());
     return -1;
   }
 
@@ -606,16 +595,20 @@ int32_t cfgLoadFromCfgFile(SConfig *pConfig, const char *filepath) {
   char   *line = NULL, *name, *value, *value2, *value3;
   int32_t olen, vlen, vlen2, vlen3;
   ssize_t _bytes = 0;
-
-  if (taosIsDir(filepath)) {
-    return -1;
-  }
+  int32_t code = 0;
 
   TdFilePtr pFile = taosOpenFile(filepath, TD_FILE_READ | TD_FILE_STREAM);
   if (pFile == NULL) {
+    // success when the file does not exist
+    if (errno == ENOENT) {
       terrno = TAOS_SYSTEM_ERROR(errno);
+      uInfo("failed to load from cfg file %s since %s, use default parameters", filepath, terrstr());
+      return 0;
+    } else {
+      uError("failed to load from cfg file %s since %s", filepath, terrstr());
       return -1;
     }
+  }
 
   while (!taosEOFFile(pFile)) {
     name = value = value2 = value3 = NULL;
@@ -643,17 +636,24 @@ int32_t cfgLoadFromCfgFile(SConfig *pConfig, const char *filepath) {
       if (vlen3 != 0) value3[vlen3] = 0;
     }
 
-    cfgSetItem(pConfig, name, value, CFG_STYPE_CFG_FILE);
+    code = cfgSetItem(pConfig, name, value, CFG_STYPE_CFG_FILE);
+    if (code != 0 && terrno != TSDB_CODE_CFG_NOT_FOUND) break;
     if (value2 != NULL && value3 != NULL && value2[0] != 0 && value3[0] != 0 && strcasecmp(name, "dataDir") == 0) {
-      cfgSetTfsItem(pConfig, name, value, value2, value3, CFG_STYPE_CFG_FILE);
+      code = cfgSetTfsItem(pConfig, name, value, value2, value3, CFG_STYPE_CFG_FILE);
+      if (code != 0 && terrno != TSDB_CODE_CFG_NOT_FOUND) break;
     }
   }
 
   taosCloseFile(&pFile);
   if (line != NULL) taosMemoryFreeClear(line);
 
+  if (code == 0 || (code != 0 && terrno == TSDB_CODE_CFG_NOT_FOUND)) {
     uInfo("load from cfg file %s success", filepath);
     return 0;
+  } else {
+    uError("failed to load from cfg file %s since %s", filepath, terrstr());
+    return -1;
+  }
 }
 
 int32_t cfgLoadFromApollUrl(SConfig *pConfig, const char *url) {

source/util/src/tlog.c

@@ -137,7 +137,14 @@ static void taosStopLog() {
   }
 }
 
-static void taosLogBuffDestroy() {
+void taosCloseLog() {
+  if (tsLogObj.logHandle != NULL) {
+    taosStopLog();
+    if (tsLogObj.logHandle != NULL && taosCheckPthreadValid(tsLogObj.logHandle->asyncThread)) {
+      taosThreadJoin(tsLogObj.logHandle->asyncThread, NULL);
+    }
+    tsLogInited = 0;
+
     taosThreadMutexDestroy(&tsLogObj.logHandle->buffMutex);
     taosCloseFile(&tsLogObj.logHandle->pFile);
     taosMemoryFreeClear(tsLogObj.logHandle->buffer);
@@ -146,15 +153,7 @@ static void taosLogBuffDestroy() {
     taosMemoryFreeClear(tsLogObj.logHandle);
     memset(&tsLogObj.logHandle, 0, sizeof(tsLogObj.logHandle));
     tsLogObj.logHandle = NULL;
-}
-
-void taosCloseLog() {
-  taosStopLog();
-  if (taosCheckPthreadValid(tsLogObj.logHandle->asyncThread)) {
-    taosThreadJoin(tsLogObj.logHandle->asyncThread, NULL);
   }
-  tsLogInited = 0;
-  taosLogBuffDestroy(tsLogObj.logHandle);
 }
 
 static bool taosLockLogFile(TdFilePtr pFile) {

source/util/src/tprocess.c

@@ -154,7 +154,8 @@ static void taosProcCleanupQueue(SProcQueue *pQueue) {
 }
 
 static int32_t taosProcQueuePush(SProcObj *pProc, SProcQueue *pQueue, const char *pHead, int16_t rawHeadLen,
-                                 const char *pBody, int32_t rawBodyLen, int64_t handle, EProcFuncType ftype) {
+                                 const char *pBody, int32_t rawBodyLen, int64_t handle, int64_t handleRef,
+                                 EProcFuncType ftype) {
   if (rawHeadLen == 0 || pHead == NULL) {
     terrno = TSDB_CODE_INVALID_PARA;
     return -1;
@@ -172,7 +173,7 @@ static int32_t taosProcQueuePush(SProcObj *pProc, SProcQueue *pQueue, const char
   }
 
   if (handle != 0 && ftype == PROC_FUNC_REQ) {
-    if (taosHashPut(pProc->hash, &handle, sizeof(int64_t), &handle, sizeof(int64_t)) != 0) {
+    if (taosHashPut(pProc->hash, &handle, sizeof(int64_t), &handleRef, sizeof(int64_t)) != 0) {
       taosThreadMutexUnlock(&pQueue->mutex);
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       return -1;
@@ -286,13 +287,13 @@ static int32_t taosProcQueuePop(SProcQueue *pQueue, void **ppHead, int16_t *pHea
       pQueue->head = headLen + bodyLen;
     } else if (remain < 8 + headLen) {
       memcpy(pHead, pQueue->pBuffer + pQueue->head + 8, remain - 8);
-      memcpy((char*)pHead + remain - 8, pQueue->pBuffer, headLen - (remain - 8));
+      memcpy((char *)pHead + remain - 8, pQueue->pBuffer, headLen - (remain - 8));
       memcpy(pBody, pQueue->pBuffer + headLen - (remain - 8), bodyLen);
       pQueue->head = headLen - (remain - 8) + bodyLen;
     } else if (remain < 8 + headLen + bodyLen) {
       memcpy(pHead, pQueue->pBuffer + pQueue->head + 8, headLen);
       memcpy(pBody, pQueue->pBuffer + pQueue->head + 8 + headLen, remain - 8 - headLen);
-      memcpy((char*)pBody + remain - 8 - headLen, pQueue->pBuffer, bodyLen - (remain - 8 - headLen));
+      memcpy((char *)pBody + remain - 8 - headLen, pQueue->pBuffer, bodyLen - (remain - 8 - headLen));
       pQueue->head = bodyLen - (remain - 8 - headLen);
     } else {
       memcpy(pHead, pQueue->pBuffer + pQueue->head + 8, headLen);
@@ -454,19 +455,25 @@ void taosProcCleanup(SProcObj *pProc) {
 }
 
 int32_t taosProcPutToChildQ(SProcObj *pProc, const void *pHead, int16_t headLen, const void *pBody, int32_t bodyLen,
-                            void *handle, EProcFuncType ftype) {
+                            void *handle, int64_t handleRef, EProcFuncType ftype) {
   if (ftype != PROC_FUNC_REQ) {
     terrno = TSDB_CODE_INVALID_PARA;
     return -1;
   }
-  return taosProcQueuePush(pProc, pProc->pChildQueue, pHead, headLen, pBody, bodyLen, (int64_t)handle, ftype);
+  return taosProcQueuePush(pProc, pProc->pChildQueue, pHead, headLen, pBody, bodyLen, (int64_t)handle, handleRef,
+                           ftype);
 }
 
-void taosProcRemoveHandle(SProcObj *pProc, void *handle) {
+int64_t taosProcRemoveHandle(SProcObj *pProc, void *handle) {
   int64_t h = (int64_t)handle;
   taosThreadMutexLock(&pProc->pChildQueue->mutex);
+
+  int64_t *handleRef = taosHashGet(pProc->hash, &h, sizeof(int64_t));
   taosHashRemove(pProc->hash, &h, sizeof(int64_t));
   taosThreadMutexUnlock(&pProc->pChildQueue->mutex);
+
+  if (handleRef == NULL) return 0;
+  return *handleRef;
 }
 
 void taosProcCloseHandles(SProcObj *pProc, void (*HandleFp)(void *handle)) {
@@ -484,7 +491,7 @@ void taosProcCloseHandles(SProcObj *pProc, void (*HandleFp)(void *handle)) {
 void taosProcPutToParentQ(SProcObj *pProc, const void *pHead, int16_t headLen, const void *pBody, int32_t bodyLen,
                           EProcFuncType ftype) {
   int32_t retry = 0;
-  while (taosProcQueuePush(pProc, pProc->pParentQueue, pHead, headLen, pBody, bodyLen, 0, ftype) != 0) {
+  while (taosProcQueuePush(pProc, pProc->pParentQueue, pHead, headLen, pBody, bodyLen, 0, 0, ftype) != 0) {
     uWarn("proc:%s, failed to put to queue:%p since %s, retry:%d", pProc->name, pProc->pParentQueue, terrstr(), retry);
     retry++;
     taosMsleep(retry);

source/util/test/procTest.cpp

@@ -120,20 +120,20 @@ TEST_F(UtilTesProc, 01_Push_Pop_Child) {
   SProcObj *cproc = taosProcInit(&cfg);
   ASSERT_NE(cproc, nullptr);
 
-  ASSERT_NE(taosProcPutToChildQ(cproc, &head, 0, body, 0, 0, PROC_FUNC_RSP), 0);
-  ASSERT_NE(taosProcPutToChildQ(cproc, &head, 0, body, 0, 0, PROC_FUNC_REGIST), 0);
-  ASSERT_NE(taosProcPutToChildQ(cproc, &head, 0, body, 0, 0, PROC_FUNC_RELEASE), 0);
-  ASSERT_NE(taosProcPutToChildQ(cproc, NULL, 12, body, 0, 0, PROC_FUNC_REQ), 0);
-  ASSERT_NE(taosProcPutToChildQ(cproc, &head, 0, body, 0, 0, PROC_FUNC_REQ), 0);
-  ASSERT_NE(taosProcPutToChildQ(cproc, &head, shm.size, body, 0, 0, PROC_FUNC_REQ), 0);
-  ASSERT_NE(taosProcPutToChildQ(cproc, &head, sizeof(STestMsg), body, shm.size, 0, PROC_FUNC_REQ), 0);
+  ASSERT_NE(taosProcPutToChildQ(cproc, &head, 0, body, 0, 0, 0, PROC_FUNC_RSP), 0);
+  ASSERT_NE(taosProcPutToChildQ(cproc, &head, 0, body, 0, 0, 0, PROC_FUNC_REGIST), 0);
+  ASSERT_NE(taosProcPutToChildQ(cproc, &head, 0, body, 0, 0, 0, PROC_FUNC_RELEASE), 0);
+  ASSERT_NE(taosProcPutToChildQ(cproc, NULL, 12, body, 0, 0, 0, PROC_FUNC_REQ), 0);
+  ASSERT_NE(taosProcPutToChildQ(cproc, &head, 0, body, 0, 0, 0, PROC_FUNC_REQ), 0);
+  ASSERT_NE(taosProcPutToChildQ(cproc, &head, shm.size, body, 0, 0, 0, PROC_FUNC_REQ), 0);
+  ASSERT_NE(taosProcPutToChildQ(cproc, &head, sizeof(STestMsg), body, shm.size, 0, 0, PROC_FUNC_REQ), 0);
 
   for (int32_t j = 0; j < 1000; j++) {
     int32_t i = 0;
     for (i = 0; i < 20; ++i) {
-      ASSERT_EQ(taosProcPutToChildQ(cproc, &head, sizeof(STestMsg), body, i, 0, PROC_FUNC_REQ), 0);
+      ASSERT_EQ(taosProcPutToChildQ(cproc, &head, sizeof(STestMsg), body, i, 0, 0, PROC_FUNC_REQ), 0);
     }
-    ASSERT_NE(taosProcPutToChildQ(cproc, &head, sizeof(STestMsg), body, i, 0, PROC_FUNC_REQ), 0);
+    ASSERT_NE(taosProcPutToChildQ(cproc, &head, sizeof(STestMsg), body, i, 0, 0, PROC_FUNC_REQ), 0);
 
     cfg.isChild = true;
     cfg.name = "1235_p";
@@ -236,7 +236,7 @@ TEST_F(UtilTesProc, 03_Handle) {
     int32_t i = 0;
     for (i = 0; i < 20; ++i) {
       head.handle = (void *)((int64_t)i);
-      ASSERT_EQ(taosProcPutToChildQ(cproc, &head, sizeof(STestMsg), body, i, (void *)((int64_t)i), PROC_FUNC_REQ), 0);
+      ASSERT_EQ(taosProcPutToChildQ(cproc, &head, sizeof(STestMsg), body, i, (void *)((int64_t)i), i, PROC_FUNC_REQ), 0);
     }
 
     cfg.isChild = true;
@@ -246,9 +246,14 @@ TEST_F(UtilTesProc, 03_Handle) {
     taosProcRun(pproc);
     taosProcCleanup(pproc);
 
-    taosProcRemoveHandle(cproc, (void *)((int64_t)3));
-    taosProcRemoveHandle(cproc, (void *)((int64_t)5));
-    taosProcRemoveHandle(cproc, (void *)((int64_t)6));
+    int64_t ref = 0;
+    
+    ref = taosProcRemoveHandle(cproc, (void *)((int64_t)3));
+    EXPECT_EQ(ref, 3);
+    ref = taosProcRemoveHandle(cproc, (void *)((int64_t)5));
+    EXPECT_EQ(ref, 5);
+    ref = taosProcRemoveHandle(cproc, (void *)((int64_t)6));
+    EXPECT_EQ(ref, 6);
     taosProcCloseHandles(cproc, processHandle);
   }
 

tests/script/tsim/tmq/basic1.sim

@@ -175,7 +175,7 @@ $totalMsgOfCtb = $rowsPerCtb
 $expectmsgcnt  = $totalMsgOfCtb
 sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
 
-print == start consumer to pull msgs from stb
+print == start consumer to pull msgs from ctb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
 system tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
 
@@ -246,7 +246,7 @@ $totalMsgOfNtb = $rowsPerCtb
 $expectmsgcnt  = $totalMsgOfNtb
 sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
 
-print == start consumer to pull msgs from stb
+print == start consumer to pull msgs from ntb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
 system tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
 

tests/script/tsim/tmq/basic1Of2Cons.sim

+#### test scenario, please refer to https://jira.taosdata.com:18090/pages/viewpage.action?pageId=135120406
+#basic1Of2Cons.sim: vgroups=1, one topic for 2 consumers, firstly insert data, then start consume. Include six topics
+#basic2Of2Cons.sim: vgroups=1, multi topics for 2 consumers, firstly insert data, then start consume. Include six topics
+#basic3Of2Cons.sim: vgroups=4, one topic for 2 consumers, firstly insert data, then start consume. Include six topics
+#basic4Of2Cons.sim: vgroups=4, multi topics for 2 consumers, firstly insert data, then start consume. Include six topics
+
+# notes1: Scalar function: ABS/ACOS/ASIN/ATAN/CEIL/COS/FLOOR/LOG/POW/ROUND/SIN/SQRT/TAN
+# The above use cases are combined with where filter conditions, such as: where ts > "2017-08-12 18:25:58.128Z" and sin(a) > 0.5;
+#
+# notes2: not support aggregate functions(such as sum/count/min/max) and time-windows(interval).
+#
+
+run tsim/tmq/prepareBasicEnv-1vgrp.sim
+
+#---- global parameters start ----#
+$dbName    = db
+$vgroups    = 1
+$stbPrefix  = stb
+$ctbPrefix  = ctb
+$ntbPrefix  = ntb
+$stbNum     = 1
+$ctbNum     = 10
+$ntbNum     = 10
+$rowsPerCtb = 10
+$tstart     = 1640966400000  # 2022-01-01 00:00:00.000
+#---- global parameters end ----#
+
+$pullDelay    = 5
+$ifcheckdata  = 1
+$showMsg      = 1
+$showRow      = 0
+
+sql connect
+sql use $dbName
+
+print == create topics from super table
+sql create topic topic_stb_column as select ts, c3 from stb
+sql create topic topic_stb_all as select ts, c1, c2, c3 from stb
+sql create topic topic_stb_function as select ts, abs(c1), sin(c2) from stb
+
+print == create topics from child table
+sql create topic topic_ctb_column as select ts, c3 from ctb0
+sql create topic topic_ctb_all as select * from ctb0
+sql create topic topic_ctb_function as select ts, abs(c1), sin(c2) from ctb0
+
+print == create topics from normal table
+sql create topic topic_ntb_column as select ts, c3 from ntb0
+sql create topic topic_ntb_all as select * from ntb0
+sql create topic topic_ntb_function as select ts, abs(c1), sin(c2) from ntb0
+
+#sql show topics
+#if $rows != 9 then 
+#  return -1
+#endi
+
+$keyList = ' . group.id:cgrp1
+$keyList = $keyList . '
+
+$cdb_index = 0
+#=============================== start consume =============================#
+
+print ================ test consume from stb
+$loop_cnt = 0
+loop_consume_diff_topic_from_stb:
+
+#######################################################################################
+# clear consume info and consume result 
+#run tsim/tmq/clearConsume.sim
+# because drop table function no stable, so by create new db for consume info and result. Modify it later
+$cdb_index = $cdb_index + 1
+$cdbName = cdb . $cdb_index
+sql create database $cdbName vgroups 1
+sleep 500
+sql use $cdbName
+
+print == create consume info table and consume result table for stb
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
+
+sql show tables
+if $rows != 2 then 
+  return -1
+endi
+#######################################################################################
+
+if $loop_cnt == 0 then
+  print == scenario 1: topic_stb_column
+  $topicList = ' . topic_stb_column
+  $topicList = $topicList . '
+elif $loop_cnt == 1 then
+  print == scenario 2: topic_stb_all
+  $topicList = ' . topic_stb_all
+  $topicList = $topicList . '
+elif $loop_cnt == 2 then
+  print == scenario 3: topic_stb_function
+  $topicList = ' . topic_stb_function
+  $topicList = $topicList . '
+else
+  goto loop_consume_diff_topic_from_stb_end
+endi
+
+$consumerId    = 0
+$totalMsgOfStb = $ctbNum * $rowsPerCtb
+$expectmsgcnt  = $totalMsgOfStb
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+
+$consumerId    = 1
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+
+print == start consumer to pull msgs from stb
+print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
+system tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
+
+print == check consume result
+wait_consumer_end_from_stb:
+sql select * from consumeresult
+print ==> rows: $rows 
+print ==> rows[0]: $data[0][0] $data[0][1] $data[0][2] $data[0][3] $data[0][4] $data[0][5] $data[0][6]
+print ==> rows[1]: $data[1][0] $data[1][1] $data[1][2] $data[1][3] $data[1][4] $data[1][5] $data[1][6]
+if $rows != 2 then
+  sleep 1000
+  goto wait_consumer_end_from_stb
+endi
+if $data[0][1] == 0 then
+  if $data[1][1] != 1 then
+    return -1
+  endi
+endi
+if $data[0][1] == 1 then
+  if $data[1][1] != 0 then
+    return -1
+  endi
+endi
+
+# either $data[0][2] == $totalMsgOfStb and $data[1][2] == 0
+# or     $data[0][2] == 0              and $data[1][2] == $totalMsgOfStb
+if $data[0][2] == $totalMsgOfStb then
+  if $data[1][2] == 0 then
+    goto check_ok_0
+  endi
+elif $data[0][2] == 0 then
+  if $data[1][2] == $totalMsgOfStb then
+    goto check_ok_0
+  endi
+endi
+return -1 
+check_ok_0:
+
+if $data[0][3] == $totalMsgOfStb then
+  if $data[1][3] == 0 then
+    goto check_ok_1
+  endi
+elif $data[0][3] == 0 then
+  if $data[1][3] == $totalMsgOfStb then
+    goto check_ok_1
+  endi
+endi
+return -1 
+check_ok_1:
+
+$loop_cnt = $loop_cnt + 1
+goto loop_consume_diff_topic_from_stb
+loop_consume_diff_topic_from_stb_end:
+
+print ================ test consume from ctb
+$loop_cnt = 0
+loop_consume_diff_topic_from_ctb:
+
+#######################################################################################
+# clear consume info and consume result 
+#run tsim/tmq/clearConsume.sim
+# because drop table function no stable, so by create new db for consume info and result. Modify it later
+$cdb_index = $cdb_index + 1
+$cdbName = cdb . $cdb_index
+sql create database $cdbName vgroups 1
+sleep 500
+sql use $cdbName
+
+print == create consume info table and consume result table for ctb
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
+
+sql show tables
+if $rows != 2 then 
+  return -1
+endi
+#######################################################################################
+
+if $loop_cnt == 0 then
+  print == scenario 1: topic_ctb_column
+  $topicList = ' . topic_ctb_column
+  $topicList = $topicList . '
+elif $loop_cnt == 1 then
+  print == scenario 2: topic_ctb_all
+  $topicList = ' . topic_ctb_all
+  $topicList = $topicList . '
+elif $loop_cnt == 2 then
+  print == scenario 3: topic_ctb_function
+  $topicList = ' . topic_ctb_function
+  $topicList = $topicList . '
+else
+  goto loop_consume_diff_topic_from_ctb_end
+endi
+
+$consumerId    = 0
+$totalMsgOfCtb = $rowsPerCtb
+$expectmsgcnt  = $totalMsgOfCtb
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+$consumerId    = 1
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+
+print == start consumer to pull msgs from ctb
+print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
+system tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
+
+print == check consume result
+wait_consumer_end_from_ctb:
+sql select * from consumeresult
+print ==> rows: $rows 
+print ==> rows[0]: $data[0][0] $data[0][1] $data[0][2] $data[0][3] $data[0][4] $data[0][5] $data[0][6]
+print ==> rows[1]: $data[1][0] $data[1][1] $data[1][2] $data[1][3] $data[1][4] $data[1][5] $data[1][6]
+if $rows != 2 then
+  sleep 1000
+  goto wait_consumer_end_from_ctb
+endi
+if $data[0][1] == 0 then
+  if $data[1][1] != 1 then
+    return -1
+  endi
+endi
+if $data[0][1] == 1 then
+  if $data[1][1] != 0 then
+    return -1
+  endi
+endi
+
+# either $data[0][2] == $totalMsgOfCtb and $data[1][2] == 0
+# or     $data[0][2] == 0              and $data[1][2] == $totalMsgOfCtb
+if $data[0][2] == $totalMsgOfCtb then
+  if $data[1][2] == 0 then
+    goto check_ok_2
+  endi
+elif $data[0][2] == 0 then
+  if $data[1][2] == $totalMsgOfCtb then
+    goto check_ok_2
+  endi
+endi
+return -1 
+check_ok_2:
+
+if $data[0][3] == $totalMsgOfCtb then
+  if $data[1][3] == 0 then
+    goto check_ok_3
+  endi
+elif $data[0][3] == 0 then
+  if $data[1][3] == $totalMsgOfCtb then
+    goto check_ok_3
+  endi
+endi
+return -1 
+check_ok_3:
+
+$loop_cnt = $loop_cnt + 1
+goto loop_consume_diff_topic_from_ctb
+loop_consume_diff_topic_from_ctb_end:
+
+print ================ test consume from ntb
+$loop_cnt = 0
+loop_consume_diff_topic_from_ntb:
+
+#######################################################################################
+# clear consume info and consume result 
+#run tsim/tmq/clearConsume.sim
+# because drop table function no stable, so by create new db for consume info and result. Modify it later
+$cdb_index = $cdb_index + 1
+$cdbName = cdb . $cdb_index
+sql create database $cdbName vgroups 1
+sleep 500
+sql use $cdbName
+
+print == create consume info table and consume result table for ntb
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
+
+sql show tables
+if $rows != 2 then 
+  return -1
+endi
+#######################################################################################
+
+if $loop_cnt == 0 then
+  print == scenario 1: topic_ntb_column
+  $topicList = ' . topic_ntb_column
+  $topicList = $topicList . '
+elif $loop_cnt == 1 then
+  print == scenario 2: topic_ntb_all
+  $topicList = ' . topic_ntb_all
+  $topicList = $topicList . '
+elif $loop_cnt == 2 then
+  print == scenario 3: topic_ntb_function
+  $topicList = ' . topic_ntb_function
+  $topicList = $topicList . '
+else
+  goto loop_consume_diff_topic_from_ntb_end
+endi
+
+$consumerId    = 0
+$totalMsgOfNtb = $rowsPerCtb
+$expectmsgcnt  = $totalMsgOfNtb
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+$consumerId    = 1
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+
+print == start consumer to pull msgs from ntb
+print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
+system tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
+
+print == check consume result from ntb
+wait_consumer_end_from_ntb:
+sql select * from consumeresult
+print ==> rows: $rows 
+print ==> rows[0]: $data[0][0] $data[0][1] $data[0][2] $data[0][3] $data[0][4] $data[0][5] $data[0][6]
+print ==> rows[1]: $data[1][0] $data[1][1] $data[1][2] $data[1][3] $data[1][4] $data[1][5] $data[1][6]
+if $rows != 2 then
+  sleep 1000
+  goto wait_consumer_end_from_ntb
+endi
+if $data[0][1] == 0 then
+  if $data[1][1] != 1 then
+    return -1
+  endi
+endi
+if $data[1][1] == 0 then
+  if $data[0][1] != 1 then
+    return -1
+  endi
+endi
+
+# either $data[0][2] == $totalMsgOfNtb and $data[1][2] == 0
+# or     $data[0][2] == 0              and $data[1][2] == $totalMsgOfNtb
+if $data[0][2] == $totalMsgOfNtb then
+  if $data[1][2] == 0 then
+    goto check_ok_4
+  endi
+elif $data[0][2] == 0 then
+  if $data[1][2] == $totalMsgOfNtb then
+    goto check_ok_4
+  endi
+endi
+return -1 
+check_ok_4:
+
+if $data[0][3] == $totalMsgOfNtb then
+  if $data[1][3] == 0 then
+    goto check_ok_5
+  endi
+elif $data[0][3] == 0 then
+  if $data[1][3] == $totalMsgOfNtb then
+    goto check_ok_5
+  endi
+endi
+return -1 
+check_ok_5:
+
+$loop_cnt = $loop_cnt + 1
+goto loop_consume_diff_topic_from_ntb
+loop_consume_diff_topic_from_ntb_end:
+
+#------ not need stop consumer, because it exit after pull msg overthan expect msg
+#system tsim/tmq/consume.sh -s stop -x SIGINT
+
+system sh/exec.sh -n dnode1 -s stop -x SIGINT

tests/script/tsim/tmq/basic2.sim

@@ -133,7 +133,7 @@ $totalMsgOfCtb = $rowsPerCtb * $topicNum
 $expectmsgcnt  = $totalMsgOfCtb
 sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
 
-print == start consumer to pull msgs from stb
+print == start consumer to pull msgs from ctb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
 system tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
 
@@ -190,7 +190,7 @@ $totalMsgOfNtb = $rowsPerCtb * $topicNum
 $expectmsgcnt  = $totalMsgOfNtb
 sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
 
-print == start consumer to pull msgs from stb
+print == start consumer to pull msgs from ntb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
 system tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
 

tests/script/tsim/tmq/basic3.sim

@@ -175,7 +175,7 @@ $totalMsgOfCtb = $rowsPerCtb
 $expectmsgcnt  = $totalMsgOfCtb
 sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
 
-print == start consumer to pull msgs from stb
+print == start consumer to pull msgs from ctb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
 system tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
 
@@ -246,7 +246,7 @@ $totalMsgOfNtb = $rowsPerCtb
 $expectmsgcnt  = $totalMsgOfNtb
 sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
 
-print == start consumer to pull msgs from stb
+print == start consumer to pull msgs from ntb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
 system tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
 

tests/script/tsim/tmq/basic4.sim

@@ -130,7 +130,7 @@ $totalMsgOfCtb = $rowsPerCtb * $topicNum
 $expectmsgcnt  = $totalMsgOfCtb
 sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
 
-print == start consumer to pull msgs from stb
+print == start consumer to pull msgs from ctb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
 system tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
 
@@ -187,7 +187,7 @@ $totalMsgOfNtb = $rowsPerCtb * $topicNum
 $expectmsgcnt  = $totalMsgOfNtb
 sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
 
-print == start consumer to pull msgs from stb
+print == start consumer to pull msgs from ntb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
 system tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
 

tests/test/c/tmqSim.c

@@ -371,6 +371,8 @@ void *consumeThreadFunc(void *param) {
   
   loop_consume(pInfo);
 
+  tmq_commit(pInfo->tmq, NULL, 0);
+
   err = tmq_unsubscribe(pInfo->tmq);
   if (err) {
     printf("tmq_unsubscribe() fail, reason: %s\n", tmq_err2str(err));