Merge remote-tracking branch 'origin/3.0' into feature/shm

include/client/taos.h

@@ -85,11 +85,7 @@ typedef struct taosField {
   int32_t bytes;
 } TAOS_FIELD;
 
-#ifdef _TD_GO_DLL_
-#define DLL_EXPORT __declspec(dllexport)
-#else
 #define DLL_EXPORT
-#endif
 
 typedef void (*__taos_async_fn_t)(void *param, TAOS_RES *, int code);
 

include/common/tcommon.h

@@ -54,25 +54,28 @@ typedef struct SColumnDataAgg {
 } SColumnDataAgg;
 
 typedef struct SDataBlockInfo {
-  STimeWindow    window;
-  int32_t        rows;
-  int32_t        rowSize;
-  int16_t        numOfCols;
-  int16_t        hasVarCol;
-  union {int64_t uid; int64_t blockId;};
+  STimeWindow window;
+  int32_t     rows;
+  int32_t     rowSize;
+  int16_t     numOfCols;
+  int16_t     hasVarCol;
+  union {
+    int64_t uid;
+    int64_t blockId;
+  };
 } SDataBlockInfo;
 
-//typedef struct SConstantItem {
-//  SColumnInfo info;
-//  int32_t     startRow;  // run-length-encoding to save the space for multiple rows
-//  int32_t     endRow;
-//  SVariant    value;
-//} SConstantItem;
+// typedef struct SConstantItem {
+//   SColumnInfo info;
+//   int32_t     startRow;  // run-length-encoding to save the space for multiple rows
+//   int32_t     endRow;
+//   SVariant    value;
+// } SConstantItem;
 
 // info.numOfCols = taosArrayGetSize(pDataBlock) + taosArrayGetSize(pConstantList);
 typedef struct SSDataBlock {
-  SColumnDataAgg *pBlockAgg;
-  SArray         *pDataBlock;    // SArray<SColumnInfoData>
+  SColumnDataAgg* pBlockAgg;
+  SArray*         pDataBlock;  // SArray<SColumnInfoData>
   SDataBlockInfo  info;
 } SSDataBlock;
 
@@ -98,23 +101,37 @@ void*   blockDataDestroy(SSDataBlock* pBlock);
 int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock);
 void*   tDecodeDataBlock(const void* buf, SSDataBlock* pBlock);
 
-static FORCE_INLINE void tDeleteSSDataBlock(SSDataBlock* pBlock) {
-  if (pBlock == NULL) {
-    return;
+static FORCE_INLINE void blockDestroyInner(SSDataBlock* pBlock) {
+  // WARNING: do not use info.numOfCols,
+  // sometimes info.numOfCols != array size
+  int32_t numOfOutput = taosArrayGetSize(pBlock->pDataBlock);
+  for (int32_t i = 0; i < numOfOutput; ++i) {
+    SColumnInfoData* pColInfoData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, i);
+    if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
+      tfree(pColInfoData->varmeta.offset);
+    } else {
+      tfree(pColInfoData->nullbitmap);
+    }
+
+    tfree(pColInfoData->pData);
   }
-  blockDataDestroy(pBlock);
+
+  taosArrayDestroy(pBlock->pDataBlock);
+  tfree(pBlock->pBlockAgg);
 }
 
+static FORCE_INLINE void tDeleteSSDataBlock(SSDataBlock* pBlock) { blockDestroyInner(pBlock); }
+
 static FORCE_INLINE int32_t tEncodeSMqPollRsp(void** buf, const SMqPollRsp* pRsp) {
   int32_t tlen = 0;
   int32_t sz = 0;
-  tlen += taosEncodeFixedI64(buf, pRsp->consumerId);
+  // tlen += taosEncodeFixedI64(buf, pRsp->consumerId);
   tlen += taosEncodeFixedI64(buf, pRsp->reqOffset);
   tlen += taosEncodeFixedI64(buf, pRsp->rspOffset);
   tlen += taosEncodeFixedI32(buf, pRsp->skipLogNum);
   tlen += taosEncodeFixedI32(buf, pRsp->numOfTopics);
   if (pRsp->numOfTopics == 0) return tlen;
-  tlen += tEncodeSSchemaWrapper(buf, pRsp->schemas);
+  tlen += tEncodeSSchemaWrapper(buf, pRsp->schema);
   if (pRsp->pBlockData) {
     sz = taosArrayGetSize(pRsp->pBlockData);
   }
@@ -128,15 +145,15 @@ static FORCE_INLINE int32_t tEncodeSMqPollRsp(void** buf, const SMqPollRsp* pRsp
 
 static FORCE_INLINE void* tDecodeSMqPollRsp(void* buf, SMqPollRsp* pRsp) {
   int32_t sz;
-  buf = taosDecodeFixedI64(buf, &pRsp->consumerId);
+  // buf = taosDecodeFixedI64(buf, &pRsp->consumerId);
   buf = taosDecodeFixedI64(buf, &pRsp->reqOffset);
   buf = taosDecodeFixedI64(buf, &pRsp->rspOffset);
   buf = taosDecodeFixedI32(buf, &pRsp->skipLogNum);
   buf = taosDecodeFixedI32(buf, &pRsp->numOfTopics);
   if (pRsp->numOfTopics == 0) return buf;
-  pRsp->schemas = (SSchemaWrapper*)calloc(1, sizeof(SSchemaWrapper));
-  if (pRsp->schemas == NULL) return NULL;
-  buf = tDecodeSSchemaWrapper(buf, pRsp->schemas);
+  pRsp->schema = (SSchemaWrapper*)calloc(1, sizeof(SSchemaWrapper));
+  if (pRsp->schema == NULL) return NULL;
+  buf = tDecodeSSchemaWrapper(buf, pRsp->schema);
   buf = taosDecodeFixedI32(buf, &sz);
   pRsp->pBlockData = taosArrayInit(sz, sizeof(SSDataBlock));
   for (int32_t i = 0; i < sz; i++) {
@@ -148,13 +165,13 @@ static FORCE_INLINE void* tDecodeSMqPollRsp(void* buf, SMqPollRsp* pRsp) {
 }
 
 static FORCE_INLINE void tDeleteSMqConsumeRsp(SMqPollRsp* pRsp) {
-  if (pRsp->schemas) {
-    if (pRsp->schemas->nCols) {
-      tfree(pRsp->schemas->pSchema);
+  if (pRsp->schema) {
+    if (pRsp->schema->nCols) {
+      tfree(pRsp->schema->pSchema);
     }
-    free(pRsp->schemas);
+    free(pRsp->schema);
   }
-  taosArrayDestroyEx(pRsp->pBlockData, (void (*)(void*))tDeleteSSDataBlock);
+  taosArrayDestroyEx(pRsp->pBlockData, (void (*)(void*))blockDestroyInner);
   pRsp->pBlockData = NULL;
 }
 
@@ -196,7 +213,7 @@ typedef struct SGroupbyExpr {
 
 typedef struct SFunctParam {
   int32_t  type;
-  SColumn *pCol;
+  SColumn* pCol;
   SVariant param;
 } SFunctParam;
 
@@ -214,12 +231,12 @@ typedef struct SResSchame {
 typedef struct SExprBasicInfo {
   SResSchema   resSchema;
   int16_t      numOfParams;  // argument value of each function
-  SFunctParam *pParam;
+  SFunctParam* pParam;
 } SExprBasicInfo;
 
 typedef struct SExprInfo {
-  struct SExprBasicInfo  base;
-  struct tExprNode      *pExpr;
+  struct SExprBasicInfo base;
+  struct tExprNode*     pExpr;
 } SExprInfo;
 
 typedef struct SStateWindow {

include/common/tmsg.h

@@ -1282,7 +1282,7 @@ static FORCE_INLINE SMqRebSubscribe* tNewSMqRebSubscribe(const char* key) {
   if (pRebSub == NULL) {
     goto _err;
   }
-  pRebSub->key = key;
+  pRebSub->key = strdup(key);
   pRebSub->lostConsumers = taosArrayInit(0, sizeof(int64_t));
   if (pRebSub->lostConsumers == NULL) {
     goto _err;
@@ -2116,25 +2116,16 @@ typedef struct {
   int8_t  mqMsgType;
   int32_t code;
   int32_t epoch;
+  int64_t consumerId;
 } SMqRspHead;
 
-typedef struct {
-  int64_t         consumerId;
-  SSchemaWrapper* schemas;
-  int64_t         reqOffset;
-  int64_t         rspOffset;
-  int32_t         skipLogNum;
-  int32_t         numOfTopics;
-  SArray*         pBlockData;  // SArray<SSDataBlock>
-} SMqPollRsp;
-
-// one req for one vg+topic
 typedef struct {
   SMsgHead head;
 
   int64_t consumerId;
   int64_t blockingTime;
   int32_t epoch;
+  int8_t  withSchema;
   char    cgroup[TSDB_CGROUP_LEN];
 
   int64_t currentOffset;
@@ -2153,19 +2144,22 @@ typedef struct {
 } SMqSubTopicEp;
 
 typedef struct {
-  int64_t consumerId;
-  char    cgroup[TSDB_CGROUP_LEN];
-  SArray* topics;  // SArray<SMqSubTopicEp>
-} SMqCMGetSubEpRsp;
+  SMqRspHead head;
+  int64_t    reqOffset;
+  int64_t    rspOffset;
+  int32_t    skipLogNum;
+  // TODO: replace with topic name
+  int32_t numOfTopics;
+  // TODO: remove from msg
+  SSchemaWrapper* schema;
+  SArray*         pBlockData;  // SArray<SSDataBlock>
+} SMqPollRsp;
 
 typedef struct {
   SMqRspHead head;
-  union {
-    SMqPollRsp       consumeRsp;
-    SMqCMGetSubEpRsp getEpRsp;
-  };
-  void* extra;
-} SMqMsgWrapper;
+  char       cgroup[TSDB_CGROUP_LEN];
+  SArray*    topics;  // SArray<SMqSubTopicEp>
+} SMqCMGetSubEpRsp;
 
 typedef struct {
   int32_t curBlock;
@@ -2173,11 +2167,13 @@ typedef struct {
   void**  uData;
 } SMqRowIter;
 
-struct tmq_message_t_v1 {
-  SMqPollRsp rsp;
+struct tmq_message_t {
+  SMqPollRsp msg;
+  void*      vg;
   SMqRowIter iter;
 };
 
+#if 0
 struct tmq_message_t {
   SMqRspHead head;
   union {
@@ -2189,6 +2185,7 @@ struct tmq_message_t {
   int32_t curRow;
   void**  uData;
 };
+#endif
 
 static FORCE_INLINE void tDeleteSMqSubTopicEp(SMqSubTopicEp* pSubTopicEp) { taosArrayDestroy(pSubTopicEp->vgs); }
 
@@ -2241,8 +2238,7 @@ static FORCE_INLINE void* tDecodeSMqSubTopicEp(void* buf, SMqSubTopicEp* pTopicE
 
 static FORCE_INLINE int32_t tEncodeSMqCMGetSubEpRsp(void** buf, const SMqCMGetSubEpRsp* pRsp) {
   int32_t tlen = 0;
-  tlen += taosEncodeFixedI64(buf, pRsp->consumerId);
-  tlen += taosEncodeString(buf, pRsp->cgroup);
+  // tlen += taosEncodeString(buf, pRsp->cgroup);
   int32_t sz = taosArrayGetSize(pRsp->topics);
   tlen += taosEncodeFixedI32(buf, sz);
   for (int32_t i = 0; i < sz; i++) {
@@ -2253,8 +2249,7 @@ static FORCE_INLINE int32_t tEncodeSMqCMGetSubEpRsp(void** buf, const SMqCMGetSu
 }
 
 static FORCE_INLINE void* tDecodeSMqCMGetSubEpRsp(void* buf, SMqCMGetSubEpRsp* pRsp) {
-  buf = taosDecodeFixedI64(buf, &pRsp->consumerId);
-  buf = taosDecodeStringTo(buf, pRsp->cgroup);
+  // buf = taosDecodeStringTo(buf, pRsp->cgroup);
   int32_t sz;
   buf = taosDecodeFixedI32(buf, &sz);
   pRsp->topics = taosArrayInit(sz, sizeof(SMqSubTopicEp));
@@ -2275,8 +2270,8 @@ enum {
 };
 
 typedef struct {
-  void*  inputHandle;
-  void** executor;
+  void* inputHandle;
+  void* executor[4];
 } SStreamTaskParRunner;
 
 typedef struct {

include/os/osDir.h

@@ -17,6 +17,7 @@
 #define _TD_OS_DIR_H_
 
 // If the error is in a third-party library, place this header file under the third-party library header file.
+// When you want to use this feature, you should find or add the same function in the following section.
 #ifndef ALLOW_FORBID_FUNC
     #define opendir OPENDIR_FUNC_TAOS_FORBID
     #define readdir READDIR_FUNC_TAOS_FORBID

include/os/osFile.h

@@ -23,6 +23,7 @@ extern "C" {
 #include "osSocket.h"
 
 // If the error is in a third-party library, place this header file under the third-party library header file.
+// When you want to use this feature, you should find or add the same function in the following sectio
 #ifndef ALLOW_FORBID_FUNC
     #define open OPEN_FUNC_TAOS_FORBID
     #define fopen FOPEN_FUNC_TAOS_FORBID

include/os/osLocale.h

@@ -23,6 +23,7 @@ extern "C" {
 #endif
 
 // If the error is in a third-party library, place this header file under the third-party library header file.
+// When you want to use this feature, you should find or add the same function in the following section.
 #ifndef ALLOW_FORBID_FUNC
     #define setlocale SETLOCALE_FUNC_TAOS_FORBID
 #endif

include/os/osRand.h

@@ -21,6 +21,7 @@ extern "C" {
 #endif
 
 // If the error is in a third-party library, place this header file under the third-party library header file.
+// When you want to use this feature, you should find or add the same function in the following section.
 #ifndef ALLOW_FORBID_FUNC
     #define rand RAND_FUNC_TAOS_FORBID
     #define srand SRAND_FUNC_TAOS_FORBID

include/os/osSleep.h

@@ -21,6 +21,7 @@ extern "C" {
 #endif
 
 // If the error is in a third-party library, place this header file under the third-party library header file.
+// When you want to use this feature, you should find or add the same function in the following section.
 #ifndef ALLOW_FORBID_FUNC
     #define Sleep SLEEP_FUNC_TAOS_FORBID
     #define sleep SLEEP_FUNC_TAOS_FORBID

include/os/osSocket.h

@@ -17,6 +17,7 @@
 #define _TD_OS_SOCKET_H_
 
 // If the error is in a third-party library, place this header file under the third-party library header file.
+// When you want to use this feature, you should find or add the same function in the following section.
 #ifndef ALLOW_FORBID_FUNC
     #define socket SOCKET_FUNC_TAOS_FORBID
     #define bind   BIND_FUNC_TAOS_FORBID
@@ -52,9 +53,6 @@ extern "C" {
 
 #if (defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)) 
   #define htobe64 htonll
-  #if defined(_TD_GO_DLL_)
-    uint64_t htonll(uint64_t val);
-  #endif
 #endif
 
 #if defined(_TD_DARWIN_64)

include/os/osString.h

@@ -24,6 +24,7 @@ typedef wchar_t TdWchar;
 typedef int32_t TdUcs4;
 
 // If the error is in a third-party library, place this header file under the third-party library header file.
+// When you want to use this feature, you should find or add the same function in the following section.
 #ifndef ALLOW_FORBID_FUNC
     #define iconv_open ICONV_OPEN_FUNC_TAOS_FORBID
     #define iconv_close ICONV_CLOSE_FUNC_TAOS_FORBID

include/os/osSystem.h

@@ -21,6 +21,7 @@ extern "C" {
 #endif
 
 // If the error is in a third-party library, place this header file under the third-party library header file.
+// When you want to use this feature, you should find or add the same function in the following section.
 #ifndef ALLOW_FORBID_FUNC
     #define popen POPEN_FUNC_TAOS_FORBID
     #define pclose PCLOSE_FUNC_TAOS_FORBID

include/os/osThread.h

@@ -37,6 +37,7 @@ typedef pthread_condattr_t TdThreadCondAttr;
 #define taosThreadCleanupPop pthread_cleanup_pop
 
 // If the error is in a third-party library, place this header file under the third-party library header file.
+// When you want to use this feature, you should find or add the same function in the following section.
 #ifndef ALLOW_FORBID_FUNC
     #define pthread_t PTHREAD_T_TYPE_TAOS_FORBID
     #define pthread_spinlock_t PTHREAD_SPINLOCK_T_TYPE_TAOS_FORBID

include/os/osTime.h

@@ -21,6 +21,7 @@ extern "C" {
 #endif
 
 // If the error is in a third-party library, place this header file under the third-party library header file.
+// When you want to use this feature, you should find or add the same function in the following section.
 #ifndef ALLOW_FORBID_FUNC
     #define strptime STRPTIME_FUNC_TAOS_FORBID
     #define gettimeofday GETTIMEOFDAY_FUNC_TAOS_FORBID
@@ -33,11 +34,7 @@ extern "C" {
 
   #define CLOCK_REALTIME 	0
 
-  #ifdef _TD_GO_DLL_
-    #define MILLISECOND_PER_SECOND (1000LL)
-  #else
-    #define MILLISECOND_PER_SECOND (1000i64)
-  #endif
+  #define MILLISECOND_PER_SECOND (1000i64)
 #else
   #define MILLISECOND_PER_SECOND ((int64_t)1000L)
 #endif

include/os/osTimer.h

@@ -21,6 +21,7 @@ extern "C" {
 #endif
 
 // If the error is in a third-party library, place this header file under the third-party library header file.
+// When you want to use this feature, you should find or add the same function in the following section.
 #ifndef ALLOW_FORBID_FUNC
     #define timer_create TIMER_CREATE_FUNC_TAOS_FORBID
     #define timer_settime TIMER_SETTIME_FUNC_TAOS_FORBID

include/os/osTimezone.h

@@ -21,6 +21,7 @@ extern "C" {
 #endif
 
 // If the error is in a third-party library, place this header file under the third-party library header file.
+// When you want to use this feature, you should find or add the same function in the following section.
 #ifndef ALLOW_FORBID_FUNC
     #define tzset TZSET_FUNC_TAOS_FORBID
 #endif

source/client/src/clientEnv.c

@@ -48,8 +48,8 @@ static void registerRequest(SRequestObj *pRequest) {
   if (pTscObj->pAppInfo) {
     SInstanceSummary *pSummary = &pTscObj->pAppInfo->summary;
 
-    int32_t total = atomic_add_fetch_32(&pSummary->totalRequests, 1);
-    int32_t currentInst = atomic_add_fetch_32(&pSummary->currentRequests, 1);
+    int32_t total = atomic_add_fetch_64(&pSummary->totalRequests, 1);
+    int32_t currentInst = atomic_add_fetch_64(&pSummary->currentRequests, 1);
     tscDebug("0x%" PRIx64 " new Request from connObj:0x%" PRIx64
              ", current:%d, app current:%d, total:%d, reqId:0x%" PRIx64,
              pRequest->self, pRequest->pTscObj->id, num, currentInst, total, pRequest->requestId);
@@ -62,7 +62,7 @@ static void deregisterRequest(SRequestObj *pRequest) {
   STscObj *         pTscObj = pRequest->pTscObj;
   SInstanceSummary *pActivity = &pTscObj->pAppInfo->summary;
 
-  int32_t currentInst = atomic_sub_fetch_32(&pActivity->currentRequests, 1);
+  int32_t currentInst = atomic_sub_fetch_64(&pActivity->currentRequests, 1);
   int32_t num = atomic_sub_fetch_32(&pTscObj->numOfReqs, 1);
 
   int64_t duration = taosGetTimestampMs() - pRequest->metric.start;

source/client/src/tmq.c

@@ -681,7 +681,7 @@ static char* formatTimestamp(char* buf, int64_t val, int precision) {
 
 int32_t tmqGetSkipLogNum(tmq_message_t* tmq_message) {
   if (tmq_message == NULL) return 0;
-  SMqPollRsp* pRsp = &tmq_message->consumeRsp;
+  SMqPollRsp* pRsp = &tmq_message->msg;
   return pRsp->skipLogNum;
 }
 
@@ -690,15 +690,15 @@ void tmqShowMsg(tmq_message_t* tmq_message) {
 
   static bool noPrintSchema;
   char        pBuf[128];
-  SMqPollRsp* pRsp = &tmq_message->consumeRsp;
-  int32_t     colNum = pRsp->schemas->nCols;
+  SMqPollRsp* pRsp = &tmq_message->msg;
+  int32_t     colNum = pRsp->schema->nCols;
   if (!noPrintSchema) {
     printf("|");
     for (int32_t i = 0; i < colNum; i++) {
       if (i == 0)
-        printf(" %25s |", pRsp->schemas->pSchema[i].name);
+        printf(" %25s |", pRsp->schema->pSchema[i].name);
       else
-        printf(" %15s |", pRsp->schemas->pSchema[i].name);
+        printf(" %15s |", pRsp->schema->pSchema[i].name);
     }
     printf("\n");
     printf("===============================================\n");
@@ -735,7 +735,7 @@ int32_t tmqPollCb(void* param, const SDataBuf* pMsg, int32_t code) {
   SMqClientVg*    pVg = pParam->pVg;
   tmq_t*          tmq = pParam->tmq;
   if (code != 0) {
-    printf("msg discard\n");
+    printf("msg discard %x\n", code);
     goto WRITE_QUEUE_FAIL;
   }
 
@@ -778,19 +778,19 @@ int32_t tmqPollCb(void* param, const SDataBuf* pMsg, int32_t code) {
     goto WRITE_QUEUE_FAIL;
   }
   memcpy(pRsp, pMsg->pData, sizeof(SMqRspHead));
-  tDecodeSMqPollRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &pRsp->consumeRsp);
-  pRsp->curBlock = 0;
-  pRsp->curRow = 0;
+  tDecodeSMqPollRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &pRsp->msg);
+  pRsp->iter.curBlock = 0;
+  pRsp->iter.curRow = 0;
   // TODO: alloc mem
   /*pRsp->*/
   /*printf("rsp commit off:%ld rsp off:%ld has data:%d\n", pRsp->committedOffset, pRsp->rspOffset, pRsp->numOfTopics);*/
-  if (pRsp->consumeRsp.numOfTopics == 0) {
+  if (pRsp->msg.numOfTopics == 0) {
     /*printf("no data\n");*/
     taosFreeQitem(pRsp);
     goto WRITE_QUEUE_FAIL;
   }
 
-  pRsp->extra = pParam->pVg;
+  pRsp->vg = pParam->pVg;
   taosWriteQitem(tmq->mqueue, pRsp);
   atomic_add_fetch_32(&tmq->readyRequest, 1);
   tsem_post(&tmq->rspSem);
@@ -860,14 +860,14 @@ int32_t tmqAskEpCb(void* param, const SDataBuf* pMsg, int32_t code) {
     }
     tDeleteSMqCMGetSubEpRsp(&rsp);
   } else {
-    tmq_message_t* pRsp = taosAllocateQitem(sizeof(tmq_message_t));
+    SMqCMGetSubEpRsp* pRsp = taosAllocateQitem(sizeof(SMqCMGetSubEpRsp));
     if (pRsp == NULL) {
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       code = -1;
       goto END;
     }
     memcpy(pRsp, pMsg->pData, sizeof(SMqRspHead));
-    tDecodeSMqCMGetSubEpRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &pRsp->getEpRsp);
+    tDecodeSMqCMGetSubEpRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), pRsp);
 
     taosWriteQitem(tmq->mqueue, pRsp);
     tsem_post(&tmq->rspSem);
@@ -983,6 +983,7 @@ SMqPollReq* tmqBuildConsumeReqImpl(tmq_t* tmq, int64_t blockingTime, SMqClientTo
   return pReq;
 }
 
+#if 0
 tmq_message_t* tmqSyncPollImpl(tmq_t* tmq, int64_t blockingTime) {
   tmq_message_t* msg = NULL;
   for (int i = 0; i < taosArrayGetSize(tmq->clientTopics); i++) {
@@ -1050,6 +1051,7 @@ tmq_message_t* tmqSyncPollImpl(tmq_t* tmq, int64_t blockingTime) {
   }
   return NULL;
 }
+#endif
 
 int32_t tmqPollImpl(tmq_t* tmq, int64_t blockingTime) {
   /*printf("call poll\n");*/
@@ -1111,11 +1113,12 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t blockingTime) {
 }
 
 // return
-int32_t tmqHandleRes(tmq_t* tmq, tmq_message_t* rspMsg, bool* pReset) {
-  if (rspMsg->head.mqMsgType == TMQ_MSG_TYPE__EP_RSP) {
+int32_t tmqHandleRes(tmq_t* tmq, SMqRspHead* rspHead, bool* pReset) {
+  if (rspHead->mqMsgType == TMQ_MSG_TYPE__EP_RSP) {
     /*printf("ep %d %d\n", rspMsg->head.epoch, tmq->epoch);*/
-    if (rspMsg->head.epoch > atomic_load_32(&tmq->epoch)) {
-      tmqUpdateEp(tmq, rspMsg->head.epoch, &rspMsg->getEpRsp);
+    if (rspHead->epoch > atomic_load_32(&tmq->epoch)) {
+      SMqCMGetSubEpRsp* rspMsg = (SMqCMGetSubEpRsp*)rspHead;
+      tmqUpdateEp(tmq, rspHead->epoch, rspMsg);
       tmqClearUnhandleMsg(tmq);
       *pReset = true;
     } else {
@@ -1129,21 +1132,22 @@ int32_t tmqHandleRes(tmq_t* tmq, tmq_message_t* rspMsg, bool* pReset) {
 
 tmq_message_t* tmqHandleAllRsp(tmq_t* tmq, int64_t blockingTime, bool pollIfReset) {
   while (1) {
-    tmq_message_t* rspMsg = NULL;
-    taosGetQitem(tmq->qall, (void**)&rspMsg);
-    if (rspMsg == NULL) {
+    SMqRspHead* rspHead = NULL;
+    taosGetQitem(tmq->qall, (void**)&rspHead);
+    if (rspHead == NULL) {
       taosReadAllQitems(tmq->mqueue, tmq->qall);
-      taosGetQitem(tmq->qall, (void**)&rspMsg);
-      if (rspMsg == NULL) return NULL;
+      taosGetQitem(tmq->qall, (void**)&rspHead);
+      if (rspHead == NULL) return NULL;
     }
 
-    if (rspMsg->head.mqMsgType == TMQ_MSG_TYPE__POLL_RSP) {
+    if (rspHead->mqMsgType == TMQ_MSG_TYPE__POLL_RSP) {
+      tmq_message_t* rspMsg = (tmq_message_t*)rspHead;
       atomic_sub_fetch_32(&tmq->readyRequest, 1);
       /*printf("handle poll rsp %d\n", rspMsg->head.mqMsgType);*/
-      if (rspMsg->head.epoch == atomic_load_32(&tmq->epoch)) {
+      if (rspMsg->msg.head.epoch == atomic_load_32(&tmq->epoch)) {
         /*printf("epoch match\n");*/
-        SMqClientVg* pVg = rspMsg->extra;
-        pVg->currentOffset = rspMsg->consumeRsp.rspOffset;
+        SMqClientVg* pVg = rspMsg->vg;
+        pVg->currentOffset = rspMsg->msg.rspOffset;
         atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
         return rspMsg;
       } else {
@@ -1153,8 +1157,8 @@ tmq_message_t* tmqHandleAllRsp(tmq_t* tmq, int64_t blockingTime, bool pollIfRese
     } else {
       /*printf("handle ep rsp %d\n", rspMsg->head.mqMsgType);*/
       bool reset = false;
-      tmqHandleRes(tmq, rspMsg, &reset);
-      taosFreeQitem(rspMsg);
+      tmqHandleRes(tmq, rspHead, &reset);
+      taosFreeQitem(rspHead);
       if (pollIfReset && reset) {
         printf("reset and repoll\n");
         tmqPollImpl(tmq, blockingTime);
@@ -1163,6 +1167,7 @@ tmq_message_t* tmqHandleAllRsp(tmq_t* tmq, int64_t blockingTime, bool pollIfRese
   }
 }
 
+#if 0
 tmq_message_t* tmq_consumer_poll_v1(tmq_t* tmq, int64_t blocking_time) {
   tmq_message_t* rspMsg = NULL;
   int64_t        startTime = taosGetTimestampMs();
@@ -1185,6 +1190,7 @@ tmq_message_t* tmq_consumer_poll_v1(tmq_t* tmq, int64_t blocking_time) {
       return NULL;
   }
 }
+#endif
 
 tmq_message_t* tmq_consumer_poll(tmq_t* tmq, int64_t blocking_time) {
   tmq_message_t* rspMsg;
@@ -1350,7 +1356,7 @@ tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* tmq_topic_v
 
 void tmq_message_destroy(tmq_message_t* tmq_message) {
   if (tmq_message == NULL) return;
-  SMqPollRsp* pRsp = &tmq_message->consumeRsp;
+  SMqPollRsp* pRsp = &tmq_message->msg;
   tDeleteSMqConsumeRsp(pRsp);
   /*free(tmq_message);*/
   taosFreeQitem(tmq_message);
@@ -1366,24 +1372,24 @@ const char* tmq_err2str(tmq_resp_err_t err) {
 }
 
 TAOS_ROW tmq_get_row(tmq_message_t* message) {
-  SMqPollRsp* rsp = &message->consumeRsp;
+  SMqPollRsp* rsp = &message->msg;
   while (1) {
-    if (message->curBlock < taosArrayGetSize(rsp->pBlockData)) {
-      SSDataBlock* pBlock = taosArrayGet(rsp->pBlockData, message->curBlock);
-      if (message->curRow < pBlock->info.rows) {
+    if (message->iter.curBlock < taosArrayGetSize(rsp->pBlockData)) {
+      SSDataBlock* pBlock = taosArrayGet(rsp->pBlockData, message->iter.curBlock);
+      if (message->iter.curRow < pBlock->info.rows) {
         for (int i = 0; i < pBlock->info.numOfCols; i++) {
           SColumnInfoData* pData = taosArrayGet(pBlock->pDataBlock, i);
-          if (colDataIsNull_s(pData, message->curRow))
-            message->uData[i] = NULL;
+          if (colDataIsNull_s(pData, message->iter.curRow))
+            message->iter.uData[i] = NULL;
           else {
-            message->uData[i] = colDataGetData(pData, message->curRow);
+            message->iter.uData[i] = colDataGetData(pData, message->iter.curRow);
           }
         }
-        message->curRow++;
-        return message->uData;
+        message->iter.curRow++;
+        return message->iter.uData;
       } else {
-        message->curBlock++;
-        message->curRow = 0;
+        message->iter.curBlock++;
+        message->iter.curRow = 0;
         continue;
       }
     }

source/common/src/tmsg.c

@@ -2720,6 +2720,8 @@ int32_t tEncodeSStreamTask(SCoder *pEncoder, const SStreamTask *pTask) {
   if (tEncodeI32(pEncoder, pTask->taskId) < 0) return -1;
   if (tEncodeI32(pEncoder, pTask->level) < 0) return -1;
   if (tEncodeI8(pEncoder, pTask->status) < 0) return -1;
+  if (tEncodeI8(pEncoder, pTask->pipeEnd) < 0) return -1;
+  if (tEncodeI8(pEncoder, pTask->parallel) < 0) return -1;
   if (tEncodeSEpSet(pEncoder, &pTask->NextOpEp) < 0) return -1;
   if (tEncodeCStr(pEncoder, pTask->qmsg) < 0) return -1;
   tEndEncode(pEncoder);
@@ -2732,6 +2734,8 @@ int32_t tDecodeSStreamTask(SCoder *pDecoder, SStreamTask *pTask) {
   if (tDecodeI32(pDecoder, &pTask->taskId) < 0) return -1;
   if (tDecodeI32(pDecoder, &pTask->level) < 0) return -1;
   if (tDecodeI8(pDecoder, &pTask->status) < 0) return -1;
+  if (tDecodeI8(pDecoder, &pTask->pipeEnd) < 0) return -1;
+  if (tDecodeI8(pDecoder, &pTask->parallel) < 0) return -1;
   if (tDecodeSEpSet(pDecoder, &pTask->NextOpEp) < 0) return -1;
   if (tDecodeCStrAlloc(pDecoder, &pTask->qmsg) < 0) return -1;
   tEndDecode(pDecoder);

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

@@ -77,8 +77,9 @@ static void vmProcessFetchQueue(SVnodeObj *pVnode, SNodeMsg *pMsg) {
   }
 
   dTrace("msg:%p, is freed, result:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
-  rpcFreeCont(pMsg->rpcMsg.pCont);
-  taosFreeQitem(pMsg);
+  // TODO: handle invalid write
+  /*rpcFreeCont(pMsg->rpcMsg.pCont);*/
+  /*taosFreeQitem(pMsg);*/
 }
 
 static void vmProcessWriteQueue(SVnodeObj *pVnode, STaosQall *qall, int32_t numOfMsgs) {

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

@@ -166,7 +166,7 @@ static int32_t mndStreamActionDelete(SSdb *pSdb, SStreamObj *pStream) {
 
 static int32_t mndStreamActionUpdate(SSdb *pSdb, SStreamObj *pOldStream, SStreamObj *pNewStream) {
   mTrace("stream:%s, perform update action", pOldStream->name);
-  atomic_exchange_32(&pOldStream->updateTime, pNewStream->updateTime);
+  atomic_exchange_64(&pOldStream->updateTime, pNewStream->updateTime);
   atomic_exchange_32(&pOldStream->version, pNewStream->version);
 
   taosWLockLatch(&pOldStream->lock);

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

@@ -272,7 +272,6 @@ static int32_t mndProcessGetSubEpReq(SNodeMsg *pMsg) {
   /*sdbWrite(pMnode->pSdb, pConsumerRaw);*/
 
   strcpy(rsp.cgroup, pReq->cgroup);
-  rsp.consumerId = consumerId;
   if (epoch != pConsumer->epoch) {
     mInfo("send new assignment to consumer, consumer epoch %d, server epoch %d", epoch, pConsumer->epoch);
     SArray *pTopics = pConsumer->currentTopics;
@@ -322,6 +321,7 @@ static int32_t mndProcessGetSubEpReq(SNodeMsg *pMsg) {
   }
   ((SMqRspHead *)buf)->mqMsgType = TMQ_MSG_TYPE__EP_RSP;
   ((SMqRspHead *)buf)->epoch = pConsumer->epoch;
+  ((SMqRspHead *)buf)->consumerId = pConsumer->consumerId;
 
   void *abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
   tEncodeSMqCMGetSubEpRsp(&abuf, &rsp);
@@ -337,21 +337,21 @@ static int32_t mndSplitSubscribeKey(const char *key, char *topic, char *cgroup)
   while (key[i] != TMQ_SEPARATOR) {
     i++;
   }
-  memcpy(topic, key, i - 1);
-  topic[i] = 0;
-  strcpy(cgroup, &key[i + 1]);
+  memcpy(cgroup, key, i);
+  cgroup[i] = 0;
+  strcpy(topic, &key[i + 1]);
   return 0;
 }
 
 static SMqRebSubscribe *mndGetOrCreateRebSub(SHashObj *pHash, const char *key) {
-  SMqRebSubscribe *pRebSub = taosHashGet(pHash, key, strlen(key));
+  SMqRebSubscribe *pRebSub = taosHashGet(pHash, key, strlen(key) + 1);
   if (pRebSub == NULL) {
     pRebSub = tNewSMqRebSubscribe(key);
     if (pRebSub == NULL) {
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       return NULL;
     }
-    taosHashPut(pHash, key, strlen(key), pRebSub, sizeof(SMqRebSubscribe));
+    taosHashPut(pHash, key, strlen(key) + 1, pRebSub, sizeof(SMqRebSubscribe));
   }
   return pRebSub;
 }
@@ -441,6 +441,7 @@ static int32_t mndProcessDoRebalanceMsg(SNodeMsg *pMsg) {
     if (pIter == NULL) break;
     SMqRebSubscribe *pRebSub = (SMqRebSubscribe *)pIter;
     SMqSubscribeObj *pSub = mndAcquireSubscribeByKey(pMnode, pRebSub->key);
+    tfree(pRebSub->key);
 
     mInfo("mq rebalance subscription: %s", pSub->key);
 
@@ -503,7 +504,8 @@ static int32_t mndProcessDoRebalanceMsg(SNodeMsg *pMsg) {
             atomic_store_32(&pRebConsumer->status, MQ_CONSUMER_STATUS__IDLE);
           }
 
-          mInfo("mq consumer:%" PRId64 ", status change from %d to %d", pRebConsumer->consumerId, status, pRebConsumer->status);
+          mInfo("mq consumer:%" PRId64 ", status change from %d to %d", pRebConsumer->consumerId, status,
+                pRebConsumer->status);
 
           SSdbRaw *pConsumerRaw = mndConsumerActionEncode(pRebConsumer);
           sdbSetRawStatus(pConsumerRaw, SDB_STATUS_READY);
@@ -537,14 +539,14 @@ static int32_t mndProcessDoRebalanceMsg(SNodeMsg *pMsg) {
               mndSplitSubscribeKey(pSub->key, topic, cgroup);
               SMqTopicObj *pTopic = mndAcquireTopic(pMnode, topic);
 
-              mInfo("mq set conn: assign vgroup %d of topic %s to consumer %" PRId64 "", pConsumerEp->vgId, topic,
-                    pConsumerEp->consumerId);
+              mInfo("mq set conn: assign vgroup %d of topic %s to consumer %" PRId64 " cgroup: %s", pConsumerEp->vgId,
+                    topic, pConsumerEp->consumerId, cgroup);
 
               mndPersistMqSetConnReq(pMnode, pTrans, pTopic, cgroup, pConsumerEp);
               mndReleaseTopic(pMnode, pTopic);
             } else {
-              mInfo("mq rebalance: assign vgroup %d, from consumer %" PRId64 " to consumer %" PRId64 "", pConsumerEp->vgId,
-                    pConsumerEp->oldConsumerId, pConsumerEp->consumerId);
+              mInfo("mq rebalance: assign vgroup %d, from consumer %" PRId64 " to consumer %" PRId64 "",
+                    pConsumerEp->vgId, pConsumerEp->oldConsumerId, pConsumerEp->consumerId);
 
               mndPersistRebalanceMsg(pMnode, pTrans, pConsumerEp);
             }
@@ -1099,7 +1101,8 @@ static int32_t mndProcessSubscribeReq(SNodeMsg *pMsg) {
       SMqSubscribeObj *pSub = mndAcquireSubscribe(pMnode, cgroup, newTopicName);
       bool             createSub = false;
       if (pSub == NULL) {
-        mDebug("create new subscription by consumer %" PRId64 ", group: %s, topic %s", consumerId, cgroup, newTopicName);
+        mDebug("create new subscription by consumer %" PRId64 ", group: %s, topic %s", consumerId, cgroup,
+               newTopicName);
         pSub = mndCreateSubscription(pMnode, pTopic, cgroup);
         createSub = true;
 

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

@@ -176,7 +176,7 @@ static int32_t mndTopicActionDelete(SSdb *pSdb, SMqTopicObj *pTopic) {
 
 static int32_t mndTopicActionUpdate(SSdb *pSdb, SMqTopicObj *pOldTopic, SMqTopicObj *pNewTopic) {
   mTrace("topic:%s, perform update action", pOldTopic->name);
-  atomic_exchange_32(&pOldTopic->updateTime, pNewTopic->updateTime);
+  atomic_exchange_64(&pOldTopic->updateTime, pNewTopic->updateTime);
   atomic_exchange_32(&pOldTopic->version, pNewTopic->version);
 
   taosWLockLatch(&pOldTopic->lock);
@@ -241,12 +241,12 @@ static int32_t mndGetPlanString(SCMCreateTopicReq *pCreate, char **pStr) {
     return TSDB_CODE_SUCCESS;
   }
 
-  SNode* pAst = NULL;
+  SNode  *pAst = NULL;
   int32_t code = nodesStringToNode(pCreate->ast, &pAst);
 
-  SQueryPlan* pPlan = NULL;
+  SQueryPlan *pPlan = NULL;
   if (TSDB_CODE_SUCCESS == code) {
-    SPlanContext cxt = { .pAstRoot = pAst, .topicQuery = true };
+    SPlanContext cxt = {.pAstRoot = pAst, .topicQuery = true};
     code = qCreateQueryPlan(&cxt, &pPlan, NULL);
   }
 
@@ -274,7 +274,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq
   topicObj.logicalPlan = "";
   topicObj.sqlLen = strlen(pCreate->sql);
 
-  char* pPlanStr = NULL;
+  char *pPlanStr = NULL;
   if (TSDB_CODE_SUCCESS != mndGetPlanString(pCreate, &pPlanStr)) {
     mError("topic:%s, failed to get plan since %s", pCreate->name, terrstr());
     return -1;

source/dnode/snode/src/snode.c

@@ -57,7 +57,9 @@ void sndMetaDelete(SStreamMeta *pMeta) {
 }
 
 int32_t sndMetaDeployTask(SStreamMeta *pMeta, SStreamTask *pTask) {
-  pTask->runner.executor = qCreateStreamExecTaskInfo(pTask->qmsg, NULL);
+  for (int i = 0; i < pTask->parallel; i++) {
+    pTask->runner.executor[i] = qCreateStreamExecTaskInfo(pTask->qmsg, NULL);
+  }
   return taosHashPut(pMeta->pHash, &pTask->taskId, sizeof(int32_t), pTask, sizeof(void *));
 }
 
@@ -95,6 +97,7 @@ void sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg) {
     SStreamTask *pTask = malloc(sizeof(SStreamTask));
     if (pTask == NULL) {
       ASSERT(0);
+      return;
     }
     SCoder decoder;
     tCoderInit(&decoder, TD_LITTLE_ENDIAN, msg, pMsg->contLen - sizeof(SMsgHead), TD_DECODER);

source/dnode/vnode/src/tq/tq.c

@@ -245,7 +245,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
   }
 
   SMqPollRsp rsp = {
-      .consumerId = consumerId,
+      /*.consumerId = consumerId,*/
       .numOfTopics = 0,
       .pBlockData = NULL,
   };
@@ -298,7 +298,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
         }
 
         taosArrayPush(pRes, pDataBlock);
-        rsp.schemas = pTopic->buffer.output[pos].pReadHandle->pSchemaWrapper;
+        rsp.schema = pTopic->buffer.output[pos].pReadHandle->pSchemaWrapper;
         rsp.rspOffset = fetchOffset;
 
         rsp.numOfTopics = 1;
@@ -312,10 +312,11 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
         }
         ((SMqRspHead*)buf)->mqMsgType = TMQ_MSG_TYPE__POLL_RSP;
         ((SMqRspHead*)buf)->epoch = pReq->epoch;
+        ((SMqRspHead*)buf)->consumerId = consumerId;
 
         void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
         tEncodeSMqPollRsp(&abuf, &rsp);
-        taosArrayDestroyEx(rsp.pBlockData, (void (*)(void*))tDeleteSSDataBlock);
+        /*taosArrayDestroyEx(rsp.pBlockData, (void (*)(void*))tDeleteSSDataBlock);*/
         pMsg->pCont = buf;
         pMsg->contLen = tlen;
         pMsg->code = 0;

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

@@ -14,8 +14,8 @@
  */
 
 #define _DEFAULT_SOURCE
-#include "vnd.h"
 #include "sync.h"
+#include "vnd.h"
 // #include "vnodeInt.h"
 
 int32_t vnodeAlter(SVnode *pVnode, const SVnodeCfg *pCfg) { return 0; }
@@ -41,6 +41,6 @@ int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad) {
 }
 
 int vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
-  vInfo("sync message is processed");
+  /*vInfo("sync message is processed");*/
   return 0;
 }

source/libs/catalog/inc/catalogInt.h

@@ -278,7 +278,7 @@ typedef struct SCtgAction {
 #define CTG_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
 
 #define CTG_API_LEAVE(c) do { int32_t __code = c; CTG_UNLOCK(CTG_READ, &gCtgMgmt.lock); CTG_API_DEBUG("CTG API leave %s", __FUNCTION__); CTG_RET(__code); } while (0)
-#define CTG_API_ENTER() do { CTG_API_DEBUG("CTG API enter %s", __FUNCTION__); CTG_LOCK(CTG_READ, &gCtgMgmt.lock); if (atomic_load_8(&gCtgMgmt.exit)) { CTG_API_LEAVE(TSDB_CODE_CTG_OUT_OF_SERVICE); }  } while (0)
+#define CTG_API_ENTER() do { CTG_API_DEBUG("CTG API enter %s", __FUNCTION__); CTG_LOCK(CTG_READ, &gCtgMgmt.lock); if (atomic_load_8((int8_t*)&gCtgMgmt.exit)) { CTG_API_LEAVE(TSDB_CODE_CTG_OUT_OF_SERVICE); }  } while (0)
 
 
 

source/libs/catalog/src/catalog.c

@@ -2044,7 +2044,7 @@ void* ctgUpdateThreadFunc(void* param) {
   while (true) {
     tsem_wait(&gCtgMgmt.sem);
     
-    if (atomic_load_8(&gCtgMgmt.exit)) {
+    if (atomic_load_8((int8_t*)&gCtgMgmt.exit)) {
       break;
     }
 
@@ -2090,7 +2090,7 @@ int32_t catalogInit(SCatalogCfg *cfg) {
     CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
   }
 
-  atomic_store_8(&gCtgMgmt.exit, false);
+  atomic_store_8((int8_t*)&gCtgMgmt.exit, false);
 
   if (cfg) {
     memcpy(&gCtgMgmt.cfg, cfg, sizeof(*cfg));
@@ -2705,11 +2705,11 @@ int32_t catalogGetExpiredDBs(SCatalog* pCtg, SDbVgVersion **dbs, uint32_t *num)
 void catalogDestroy(void) {
   qInfo("start to destroy catalog");
   
-  if (NULL == gCtgMgmt.pCluster || atomic_load_8(&gCtgMgmt.exit)) {
+  if (NULL == gCtgMgmt.pCluster || atomic_load_8((int8_t*)&gCtgMgmt.exit)) {
     return;
   }
 
-  atomic_store_8(&gCtgMgmt.exit, true);
+  atomic_store_8((int8_t*)&gCtgMgmt.exit, true);
 
   tsem_post(&gCtgMgmt.sem);
 

source/libs/executor/src/tsimplehash.c

@@ -40,7 +40,7 @@ typedef struct SHNode {
 typedef struct SSHashObj {
   SHNode         **hashList;
   size_t           capacity;     // number of slots
-  size_t           size;         // number of elements in hash table
+  int64_t          size;         // number of elements in hash table
   _hash_fn_t       hashFp;       // hash function
   _equal_fn_t      equalFp;      // equal function
   int32_t          keyLen;
@@ -91,7 +91,7 @@ int32_t tSimpleHashGetSize(const SSHashObj *pHashObj) {
   if (pHashObj == NULL) {
     return 0;
   }
-  return (int32_t)atomic_load_64(&pHashObj->size);
+  return (int32_t)atomic_load_64((int64_t*)&pHashObj->size);
 }
 
 static SHNode *doCreateHashNode(const void *key, size_t keyLen, const void *pData, size_t dsize, uint32_t hashVal) {

source/libs/function/src/functionMgt.c

@@ -97,7 +97,7 @@ bool fmIsAggFunc(int32_t funcId) {
 
 void fmFuncMgtDestroy() {
   void* m = gFunMgtService.pFuncNameHashTable;
-  if (m != NULL && atomic_val_compare_exchange_ptr(&gFunMgtService.pFuncNameHashTable, m, 0) == m) {
+  if (m != NULL && atomic_val_compare_exchange_ptr((void**)&gFunMgtService.pFuncNameHashTable, m, 0) == m) {
     taosHashCleanup(m);
   }
 }

source/libs/parser/test/mockCatalogService.cpp

@@ -16,8 +16,8 @@
 #include <iomanip>
 #include <iostream>
 #include <map>
-#include "tdatablock.h"
 #include "mockCatalogService.h"
+#include "tdatablock.h"
 
 #include "tname.h"
 #include "ttypes.h"

source/libs/qworker/test/qworkerTests.cpp

@@ -50,7 +50,7 @@ namespace {
 int32_t qwtTestMaxExecTaskUsec = 2;
 int32_t qwtTestReqMaxDelayUsec = 2;
 
-uint64_t qwtTestQueryId = 0;
+int64_t qwtTestQueryId = 0;
 bool qwtTestEnableSleep = true;
 bool qwtTestStop = false;
 bool qwtTestDeadLoop = false;

source/libs/sync/inc/syncIO.h

@@ -35,7 +35,7 @@ extern "C" {
 typedef struct SSyncIO {
   STaosQueue *pMsgQ;
   STaosQset * pQset;
-  TdThread   consumerTid;
+  TdThread    consumerTid;
 
   void * serverRpc;
   void * clientRpc;

source/libs/sync/inc/syncInt.h

@@ -217,10 +217,12 @@ int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode);
 int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode);
 
 // utils --------------
-int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, SRpcMsg* pMsg);
-int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, SRpcMsg* pMsg);
-cJSON*  syncNode2Json(const SSyncNode* pSyncNode);
-char*   syncNode2Str(const SSyncNode* pSyncNode);
+int32_t    syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, SRpcMsg* pMsg);
+int32_t    syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, SRpcMsg* pMsg);
+cJSON*     syncNode2Json(const SSyncNode* pSyncNode);
+char*      syncNode2Str(const SSyncNode* pSyncNode);
+SSyncNode* syncNodeAcquire(int64_t rid);
+void       syncNodeRelease(SSyncNode* pNode);
 
 // raft state change --------------
 void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term);

source/libs/sync/src/syncMain.c

@@ -575,6 +575,17 @@ char* syncNode2Str(const SSyncNode* pSyncNode) {
   return serialized;
 }
 
+SSyncNode* syncNodeAcquire(int64_t rid) {
+  SSyncNode* pNode = taosAcquireRef(tsNodeRefId, rid);
+  if (pNode == NULL) {
+    sTrace("failed to acquire node from refId:%" PRId64, rid);
+  }
+
+  return pNode;
+}
+
+void syncNodeRelease(SSyncNode* pNode) { taosReleaseRef(tsNodeRefId, pNode->rid); }
+
 // raft state change --------------
 void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term) {
   if (term > pSyncNode->pRaftStore->currentTerm) {

source/libs/sync/test/CMakeLists.txt

@@ -28,6 +28,7 @@ add_executable(syncRpcMsgTest "")
 add_executable(syncPingTimerTest2 "")
 add_executable(syncPingSelfTest "")
 add_executable(syncElectTest "")
+add_executable(syncElectTest2 "")
 add_executable(syncEncodeTest "")
 add_executable(syncWriteTest "")
 add_executable(syncReplicateTest "")
@@ -154,6 +155,10 @@ target_sources(syncElectTest
     PRIVATE
     "syncElectTest.cpp"
 )
+target_sources(syncElectTest2
+    PRIVATE
+    "syncElectTest2.cpp"
+)
 target_sources(syncEncodeTest
     PRIVATE
     "syncEncodeTest.cpp"
@@ -322,7 +327,7 @@ target_include_directories(syncElectTest
     "${CMAKE_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
-target_include_directories(syncElectTest
+target_include_directories(syncElectTest2
     PUBLIC
     "${CMAKE_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
@@ -469,6 +474,10 @@ target_link_libraries(syncElectTest
     sync
     gtest_main
 )
+target_link_libraries(syncElectTest2
+    sync
+    gtest_main
+)
 target_link_libraries(syncEncodeTest
     sync
     gtest_main

source/libs/sync/test/syncElectTest.cpp

@@ -120,6 +120,7 @@ int main(int argc, char** argv) {
         "elect sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, electTimerMS:%d",
         gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm,
         gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS);
+    taosMsleep(1000);
   }
 
   return 0;

source/libs/sync/test/syncElectTest2.cpp

+#include <gtest/gtest.h>
+#include <stdio.h>
+#include "syncEnv.h"
+#include "syncIO.h"
+#include "syncInt.h"
+#include "syncRaftLog.h"
+#include "syncRaftStore.h"
+#include "syncUtil.h"
+#include "tref.h"
+
+void logTest() {
+  sTrace("--- sync log test: trace");
+  sDebug("--- sync log test: debug");
+  sInfo("--- sync log test: info");
+  sWarn("--- sync log test: warn");
+  sError("--- sync log test: error");
+  sFatal("--- sync log test: fatal");
+}
+
+uint16_t ports[] = {7010, 7110, 7210, 7310, 7410};
+int32_t  replicaNum = 3;
+int32_t  myIndex = 0;
+
+SRaftId   ids[TSDB_MAX_REPLICA];
+SSyncInfo syncInfo;
+SSyncFSM* pFsm;
+SWal*     pWal;
+
+int64_t syncNodeInit() {
+  syncInfo.vgId = 1234;
+  syncInfo.rpcClient = gSyncIO->clientRpc;
+  syncInfo.FpSendMsg = syncIOSendMsg;
+  syncInfo.queue = gSyncIO->pMsgQ;
+  syncInfo.FpEqMsg = syncIOEqMsg;
+  syncInfo.pFsm = pFsm;
+  snprintf(syncInfo.path, sizeof(syncInfo.path), "./elect2_test_%d", myIndex);
+
+  int code = walInit();
+  assert(code == 0);
+  SWalCfg walCfg;
+  memset(&walCfg, 0, sizeof(SWalCfg));
+  walCfg.vgId = syncInfo.vgId;
+  walCfg.fsyncPeriod = 1000;
+  walCfg.retentionPeriod = 1000;
+  walCfg.rollPeriod = 1000;
+  walCfg.retentionSize = 1000;
+  walCfg.segSize = 1000;
+  walCfg.level = TAOS_WAL_FSYNC;
+
+  char tmpdir[128];
+  snprintf(tmpdir, sizeof(tmpdir), "./elect2_test_wal_%d", myIndex);
+  pWal = walOpen(tmpdir, &walCfg);
+  assert(pWal != NULL);
+
+  syncInfo.pWal = pWal;
+
+  SSyncCfg* pCfg = &syncInfo.syncCfg;
+  pCfg->myIndex = myIndex;
+  pCfg->replicaNum = replicaNum;
+
+  for (int i = 0; i < replicaNum; ++i) {
+    pCfg->nodeInfo[i].nodePort = ports[i];
+    snprintf(pCfg->nodeInfo[i].nodeFqdn, sizeof(pCfg->nodeInfo[i].nodeFqdn), "%s", "127.0.0.1");
+    // taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn);
+  }
+
+  int64_t rid = syncStart(&syncInfo);
+  assert(rid > 0);
+
+  SSyncNode* pSyncNode = (SSyncNode*)syncNodeAcquire(rid);
+  assert(pSyncNode != NULL);
+
+  gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing;
+  gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
+  gSyncIO->FpOnSyncRequestVote = pSyncNode->FpOnRequestVote;
+  gSyncIO->FpOnSyncRequestVoteReply = pSyncNode->FpOnRequestVoteReply;
+  gSyncIO->FpOnSyncAppendEntries = pSyncNode->FpOnAppendEntries;
+  gSyncIO->FpOnSyncAppendEntriesReply = pSyncNode->FpOnAppendEntriesReply;
+  gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing;
+  gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
+  gSyncIO->FpOnSyncTimeout = pSyncNode->FpOnTimeout;
+  gSyncIO->pSyncNode = pSyncNode;
+
+  syncNodeRelease(pSyncNode);
+
+  return rid;
+}
+
+void initRaftId(SSyncNode* pSyncNode) {
+  for (int i = 0; i < replicaNum; ++i) {
+    ids[i] = pSyncNode->replicasId[i];
+    char* s = syncUtilRaftId2Str(&ids[i]);
+    printf("raftId[%d] : %s\n", i, s);
+    free(s);
+  }
+}
+
+int main(int argc, char** argv) {
+  // taosInitLog((char *)"syncTest.log", 100000, 10);
+  tsAsyncLog = 0;
+  sDebugFlag = 143 + 64;
+
+  myIndex = 0;
+  if (argc >= 2) {
+    myIndex = atoi(argv[1]);
+  }
+
+  int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
+  assert(ret == 0);
+
+  ret = syncInit();
+  assert(ret == 0);
+
+  int64_t rid = syncNodeInit();
+  assert(rid > 0);
+
+  SSyncNode* pSyncNode = (SSyncNode*)syncNodeAcquire(rid);
+  assert(pSyncNode != NULL);
+
+  syncNodePrint2((char*)"", pSyncNode);
+  initRaftId(pSyncNode);
+
+  //---------------------------
+  while (1) {
+    sTrace(
+        "elect sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, electTimerMS:%d",
+        pSyncNode->state, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm,
+        pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS);
+    taosMsleep(1000);
+  }
+
+  syncNodeRelease(pSyncNode);
+
+  return 0;
+}

source/os/src/osSocket.c

@@ -216,12 +216,6 @@ int32_t taosShutDownSocketServerRDWR(TdSocketServerPtr pSocketServer) {
 #endif
 }
 
-#if (defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32))
-  #if defined(_TD_GO_DLL_)
-    uint64_t htonll(uint64_t val) { return (((uint64_t)htonl(val)) << 32) + htonl(val >> 32); }
-  #endif
-#endif
-
 void taosWinSocketInit1() {
 #if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
   static char flag = 0;

source/util/src/tcache.c

@@ -456,7 +456,7 @@ void *taosCacheAcquireByKey(SCacheObj *pCacheObj, const void *key, size_t keyLen
   }
 
   if (pCacheObj->numOfElems == 0) {
-    atomic_add_fetch_32(&pCacheObj->statistics.missCount, 1);
+    atomic_add_fetch_64(&pCacheObj->statistics.missCount, 1);
     return NULL;
   }
 
@@ -475,15 +475,15 @@ void *taosCacheAcquireByKey(SCacheObj *pCacheObj, const void *key, size_t keyLen
 
   void *pData = (pNode != NULL) ? pNode->data : NULL;
   if (pData != NULL) {
-    atomic_add_fetch_32(&pCacheObj->statistics.hitCount, 1);
+    atomic_add_fetch_64(&pCacheObj->statistics.hitCount, 1);
     uDebug("cache:%s, key:%p, %p is retrieved from cache, refcnt:%d", pCacheObj->name, key, pData,
            T_REF_VAL_GET(pNode));
   } else {
-    atomic_add_fetch_32(&pCacheObj->statistics.missCount, 1);
+    atomic_add_fetch_64(&pCacheObj->statistics.missCount, 1);
     uDebug("cache:%s, key:%p, not in cache, retrieved failed", pCacheObj->name, key);
   }
 
-  atomic_add_fetch_32(&pCacheObj->statistics.totalAccess, 1);
+  atomic_add_fetch_64(&pCacheObj->statistics.totalAccess, 1);
   return pData;
 }
 

source/util/src/thash.c

@@ -55,7 +55,7 @@ typedef struct SHashEntry {
 struct SHashObj {
   SHashEntry **     hashList;
   size_t            capacity;      // number of slots
-  size_t            size;          // number of elements in hash table
+  int64_t           size;          // number of elements in hash table
   _hash_fn_t        hashFp;        // hash function
   _equal_fn_t       equalFp;       // equal function
   _hash_free_fn_t   freeFp;        // hash node free callback function
@@ -186,7 +186,7 @@ static SHashNode *doCreateHashNode(const void *key, size_t keyLen, const void *p
 static FORCE_INLINE void doUpdateHashNode(SHashObj *pHashObj, SHashEntry* pe, SHashNode* prev, SHashNode *pNode, SHashNode *pNewNode) {
   assert(pNode->keyLen == pNewNode->keyLen);
 
-  atomic_sub_fetch_32(&pNode->refCount, 1);
+  atomic_sub_fetch_16(&pNode->refCount, 1);
   if (prev != NULL) {
     prev->next = pNewNode;
   } else {
@@ -302,7 +302,7 @@ int32_t taosHashGetSize(const SHashObj *pHashObj) {
   if (pHashObj == NULL) {
     return 0;
   }
-  return (int32_t)atomic_load_64(&pHashObj->size);
+  return (int32_t)atomic_load_64((int64_t*)&pHashObj->size);
 }
 
 int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, void *data, size_t size) {
@@ -508,7 +508,7 @@ int32_t taosHashRemove(SHashObj *pHashObj, const void *key, size_t keyLen) {
         pNode->removed == 0) {
       code = 0;  // it is found
 
-      atomic_sub_fetch_32(&pNode->refCount, 1);
+      atomic_sub_fetch_16(&pNode->refCount, 1);
       pNode->removed = 1;
       if (pNode->refCount <= 0) {
         if (prevNode == NULL) {
@@ -755,7 +755,7 @@ static void *taosHashReleaseNode(SHashObj *pHashObj, void *p, int *slot) {
       pNode = pNode->next;
     }
 
-    atomic_sub_fetch_32(&pOld->refCount, 1);
+    atomic_sub_fetch_16(&pOld->refCount, 1);
     if (pOld->refCount <=0) {
       if (prevNode) {
         prevNode->next = pOld->next;

source/util/src/ttimer.c

@@ -132,7 +132,7 @@ static timer_map_t timerMap;
 static uintptr_t getNextTimerId() {
   uintptr_t id;
   do {
-    id = atomic_add_fetch_ptr(&nextTimerId, 1);
+    id = (uintptr_t)atomic_add_fetch_ptr((void **)&nextTimerId, 1);
   } while (id == 0);
   return id;
 }

tests/test/c/tmqDemo.c

@@ -65,11 +65,11 @@ typedef struct {
 static SConfInfo g_stConfInfo = {
     "tmqdb",
     "stb",
-	"./tmqResult.txt",	// output_file
-	"/data2/dnode/data/vnode/vnode2/wal",
+    "./tmqResult.txt",	// output_file
+    "/data2/dnode/data/vnodes/vnode2/wal",
     1, // threads
     1, // tables
-	1, // vgroups
+    1, // vgroups
     0, // run mode
     1, // columns
     1, // ratio
@@ -306,7 +306,7 @@ int32_t init_env() {
   }
 
   //const char* sql = "select * from tu1";
-  sprintf(sqlStr, "create topic test_stb_topic_1 as select * from %s%d", g_stConfInfo.stbName, 0);
+  sprintf(sqlStr, "create topic test_stb_topic_1 as select * from %s", g_stConfInfo.stbName);
   /*pRes = tmq_create_topic(pConn, "test_stb_topic_1", sqlStr, strlen(sqlStr));*/
   pRes = taos_query(pConn, sqlStr);
   if (taos_errno(pRes) != 0) {