other: merge 3.0

cmake/taostools_CMakeLists.txt.in

@@ -2,7 +2,7 @@
 # taos-tools
 ExternalProject_Add(taos-tools
         GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
-        GIT_TAG 7321fbb
+        GIT_TAG 719fc88
         SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
         BINARY_DIR ""
         #BUILD_IN_SOURCE TRUE

include/common/tglobal.h

@@ -119,6 +119,7 @@ extern SDiskCfg tsDiskCfg[];
 // udf
 extern bool tsStartUdfd;
 extern char tsUdfdResFuncs[];
+extern char tsUdfdLdLibPath[];
 
 // schemaless
 extern char tsSmlChildTableName[];

include/libs/catalog/catalog.h

@@ -309,6 +309,9 @@ int32_t catalogGetUdfInfo(SCatalog* pCtg, SRequestConnInfo* pConn, const char* f
 int32_t catalogChkAuth(SCatalog* pCtg, SRequestConnInfo* pConn, const char* user, const char* dbFName, AUTH_TYPE type,
                        bool* pass);
 
+int32_t catalogChkAuthFromCache(SCatalog* pCtg, SRequestConnInfo* pConn, const char* user, const char* dbFName, AUTH_TYPE type,
+                                       bool* pass, bool* exists);
+
 int32_t catalogUpdateUserAuthInfo(SCatalog* pCtg, SGetUserAuthRsp* pAuth);
 
 int32_t catalogUpdateVgEpSet(SCatalog* pCtg, const char* dbFName, int32_t vgId, SEpSet* epSet);

include/libs/sync/sync.h

@@ -200,10 +200,10 @@ typedef struct SSyncInfo {
 
 int32_t     syncInit();
 void        syncCleanUp();
+bool        syncIsInit();
 int64_t     syncOpen(SSyncInfo* pSyncInfo);
 void        syncStart(int64_t rid);
 void        syncStop(int64_t rid);
-int32_t     syncSetStandby(int64_t rid);
 ESyncState  syncGetMyRole(int64_t rid);
 bool        syncIsReady(int64_t rid);
 const char* syncGetMyRoleStr(int64_t rid);
@@ -216,7 +216,6 @@ void        syncGetEpSet(int64_t rid, SEpSet* pEpSet);
 void        syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet);
 int32_t     syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak);
 // int32_t     syncProposeBatch(int64_t rid, SRpcMsg** pMsgPArr, bool* pIsWeakArr, int32_t arrSize);
-bool        syncEnvIsStart();
 const char* syncStr(ESyncState state);
 bool        syncIsRestoreFinish(int64_t rid);
 int32_t     syncGetSnapshotByIndex(int64_t rid, SyncIndex index, SSnapshot* pSnapshot);
@@ -234,6 +233,9 @@ int32_t syncEndSnapshot(int64_t rid);
 
 int32_t syncStepDown(int64_t rid, SyncTerm newTerm);
 
+SSyncNode* syncNodeAcquire(int64_t rid);
+void       syncNodeRelease(SSyncNode* pNode);
+
 #ifdef __cplusplus
 }
 #endif

include/libs/sync/syncTools.h

@@ -28,10 +28,6 @@ typedef struct SRaftId {
   SyncGroupId vgId;
 } SRaftId;
 
-// ------------------ control -------------------
-SSyncNode* syncNodeAcquire(int64_t rid);
-void       syncNodeRelease(SSyncNode* pNode);
-
 int32_t syncGetRespRpc(int64_t rid, uint64_t index, SRpcMsg* msg);
 int32_t syncGetAndDelRespRpc(int64_t rid, uint64_t index, SRpcHandleInfo* pInfo);
 void    syncSetMsgCb(int64_t rid, const SMsgCb* msgcb);

include/util/tref.h

@@ -25,7 +25,8 @@ extern "C" {
 
 // open a reference set, max is the mod used by hash, fp is the pointer to free resource function
 // return rsetId which will be used by other APIs. On error, -1 is returned, and terrno is set appropriately
-int32_t taosOpenRef(int32_t max, void (*fp)(void *));
+typedef void (*RefFp)(void *);
+int32_t taosOpenRef(int32_t max, RefFp fp);
 
 // close the reference set, refId is the return value by taosOpenRef
 // return 0 if success. On error, -1 is returned, and terrno is set appropriately

source/common/src/tglobal.c

@@ -163,7 +163,8 @@ int32_t tsTtlUnit = 86400;
 int32_t tsTtlPushInterval = 86400;
 int32_t tsGrantHBInterval = 60;
 int32_t tsUptimeInterval = 300;     // seconds
-char    tsUdfdResFuncs[1024] = "";  // udfd resident funcs that teardown when udfd exits
+char    tsUdfdResFuncs[512] = "";  // udfd resident funcs that teardown when udfd exits
+char    tsUdfdLdLibPath[512] = "";
 
 #ifndef _STORAGE
 int32_t taosSetTfsCfg(SConfig *pCfg) {
@@ -424,6 +425,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
 
   if (cfgAddBool(pCfg, "udf", tsStartUdfd, 0) != 0) return -1;
   if (cfgAddString(pCfg, "udfdResFuncs", tsUdfdResFuncs, 0) != 0) return -1;
+  if (cfgAddString(pCfg, "udfdLdLibPath", tsUdfdLdLibPath, 0) != 0) return -1;
   GRANT_CFG_ADD;
   return 0;
 }
@@ -722,7 +724,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
 
   tsStartUdfd = cfgGetItem(pCfg, "udf")->bval;
   tstrncpy(tsUdfdResFuncs, cfgGetItem(pCfg, "udfdResFuncs")->str, sizeof(tsUdfdResFuncs));
-
+  tstrncpy(tsUdfdLdLibPath, cfgGetItem(pCfg, "udfdLdLibPath")->str, sizeof(tsUdfdLdLibPath));
   if (tsQueryBufferSize >= 0) {
     tsQueryBufferSizeBytes = tsQueryBufferSize * 1048576UL;
   }

source/dnode/mgmt/CMakeLists.txt

@@ -13,4 +13,18 @@ target_include_directories(
     taosd
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/node_mgmt/inc"
 )
-target_link_libraries(taosd dnode)
+
+IF (TD_LINUX_64 AND JEMALLOC_ENABLED)
+  ADD_DEFINITIONS(-DTD_JEMALLOC_ENABLED -I${CMAKE_BINARY_DIR}/build/include -L${CMAKE_BINARY_DIR}/build/lib -Wl,-rpath,${CMAKE_BINARY_DIR}/build/lib -ljemalloc)
+  SET(LINK_JEMALLOC "-L${CMAKE_BINARY_DIR}/build/lib -ljemalloc")
+ELSE ()
+  SET(LINK_JEMALLOC "")
+ENDIF ()
+
+IF (TD_LINUX_64 AND JEMALLOC_ENABLED)
+    ADD_DEPENDENCIES(taosd jemalloc)
+    target_link_libraries(taosd dnode ${LINK_JEMALLOC})
+ELSE ()
+    target_link_libraries(taosd dnode)
+ENDIF ()
+

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

@@ -481,7 +481,7 @@ int32_t mndProcessSyncCtrlMsg(SRpcMsg *pMsg) {
 
   mInfo("vgId:%d, process sync ctrl msg", 1);
 
-  if (!syncEnvIsStart()) {
+  if (!syncIsInit()) {
     mError("failed to process sync msg:%p type:%s since syncEnv stop", pMsg, TMSG_INFO(pMsg->msgType));
     terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
     return -1;
@@ -518,7 +518,7 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
   SSyncMgmt *pMgmt = &pMnode->syncMgmt;
   int32_t    code = 0;
 
-  if (!syncEnvIsStart()) {
+  if (!syncIsInit()) {
     mError("failed to process sync msg:%p type:%s since syncEnv stop", pMsg, TMSG_INFO(pMsg->msgType));
     terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
     return -1;
@@ -581,11 +581,6 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
     code = syncNodeOnSnapshotReply(pSyncNode, pSyncMsg);
     syncSnapshotRspDestroy(pSyncMsg);
 
-  } else if (pMsg->msgType == TDMT_SYNC_SET_MNODE_STANDBY) {
-    code = syncSetStandby(pMgmt->sync);
-    SRpcMsg rsp = {.code = code, .info = pMsg->info};
-    tmsgSendRsp(&rsp);
-
   } else if (pMsg->msgType == TDMT_SYNC_LOCAL_CMD) {
     SyncLocalCmd *pSyncMsg = syncLocalCmdFromRpcMsg2(pMsg);
     code = syncNodeOnLocalCmd(pSyncNode, pSyncMsg);

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

@@ -17,11 +17,44 @@
 #include "mndSync.h"
 #include "mndTrans.h"
 
+static int32_t mndSyncEqCtrlMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) {
+  if (pMsg == NULL || pMsg->pCont == NULL) {
+    return -1;
+  }
+
+  SMsgHead *pHead = pMsg->pCont;
+  pHead->contLen = htonl(pHead->contLen);
+  pHead->vgId = htonl(pHead->vgId);
+
+  if (msgcb == NULL || msgcb->putToQueueFp == NULL) {
+    rpcFreeCont(pMsg->pCont);
+    pMsg->pCont = NULL;
+    return -1;
+  }
+
+  int32_t code = tmsgPutToQueue(msgcb, SYNC_CTRL_QUEUE, pMsg);
+  if (code != 0) {
+    rpcFreeCont(pMsg->pCont);
+    pMsg->pCont = NULL;
+  }
+  return code;
+}
+
 static int32_t mndSyncEqMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) {
+  if (pMsg == NULL || pMsg->pCont == NULL) {
+    return -1;
+  }
+
   SMsgHead *pHead = pMsg->pCont;
   pHead->contLen = htonl(pHead->contLen);
   pHead->vgId = htonl(pHead->vgId);
 
+  if (msgcb == NULL || msgcb->putToQueueFp == NULL) {
+    rpcFreeCont(pMsg->pCont);
+    pMsg->pCont = NULL;
+    return -1;
+  }
+
   int32_t code = tmsgPutToQueue(msgcb, SYNC_QUEUE, pMsg);
   if (code != 0) {
     rpcFreeCont(pMsg->pCont);
@@ -212,7 +245,7 @@ int32_t mndInitSync(SMnode *pMnode) {
       .msgcb = NULL,
       .FpSendMsg = mndSyncSendMsg,
       .FpEqMsg = mndSyncEqMsg,
-      .FpEqCtrlMsg = NULL,
+      .FpEqCtrlMsg = mndSyncEqCtrlMsg,
   };
 
   snprintf(syncInfo.path, sizeof(syncInfo.path), "%s%ssync", pMnode->path, TD_DIRSEP);

source/dnode/vnode/src/tsdb/tsdbRead.c

@@ -830,7 +830,8 @@ int32_t getEndPosInDataBlock(STsdbReader* pReader, SBlockData* pBlockData, SData
   } else if (!asc && pReader->window.skey <= pBlock->minKey.ts) {
     endPos = 0;
   } else {
-    endPos = doBinarySearchKey(pBlockData->aTSKEY, pBlock->nRow, pos, pReader->window.ekey, pReader->order);
+    int64_t key = asc ? pReader->window.ekey : pReader->window.skey;
+    endPos = doBinarySearchKey(pBlockData->aTSKEY, pBlock->nRow, pos, key, pReader->order);
   }
 
   return endPos;
@@ -861,8 +862,9 @@ static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader, STableBlockScanIn
       // pDumpInfo->rowIndex = pBlock->nRow - 1;
     } else {
       int32_t pos = asc ? pBlock->nRow - 1 : 0;
-      int32_t order = (pReader->order == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC;
-      pDumpInfo->rowIndex = doBinarySearchKey(pBlockData->aTSKEY, pBlock->nRow, pos, pReader->window.skey, order);
+      int32_t order = asc ? TSDB_ORDER_DESC : TSDB_ORDER_ASC;
+      int64_t key = asc ? pReader->window.skey : pReader->window.ekey;
+      pDumpInfo->rowIndex = doBinarySearchKey(pBlockData->aTSKEY, pBlock->nRow, pos, key, order);
     }
   }
 

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

@@ -234,7 +234,7 @@ int32_t vnodeProcessSyncCtrlMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
   int32_t         code = 0;
   const STraceId *trace = &pMsg->info.traceId;
 
-  if (!syncEnvIsStart()) {
+  if (!syncIsInit()) {
     vGError("vgId:%d, msg:%p failed to process since sync env not start", pVnode->config.vgId, pMsg);
     terrno = TSDB_CODE_APP_ERROR;
     return -1;
@@ -277,7 +277,7 @@ int32_t vnodeProcessSyncMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
   int32_t         code = 0;
   const STraceId *trace = &pMsg->info.traceId;
 
-  if (!syncEnvIsStart()) {
+  if (!syncIsInit()) {
     vGError("vgId:%d, msg:%p failed to process since sync env not start", pVnode->config.vgId, pMsg);
     terrno = TSDB_CODE_APP_ERROR;
     return -1;
@@ -370,7 +370,13 @@ int32_t vnodeProcessSyncMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
 }
 
 static int32_t vnodeSyncEqCtrlMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) {
-  if (msgcb == NULL) {
+  if (pMsg == NULL || pMsg->pCont == NULL) {
+    return -1;
+  }
+
+  if (msgcb == NULL || msgcb->putToQueueFp == NULL) {
+    rpcFreeCont(pMsg->pCont);
+    pMsg->pCont = NULL;
     return -1;
   }
 
@@ -383,7 +389,13 @@ static int32_t vnodeSyncEqCtrlMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) {
 }
 
 static int32_t vnodeSyncEqMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) {
-  if (msgcb == NULL) {
+  if (pMsg == NULL || pMsg->pCont == NULL) {
+    return -1;
+  }
+
+  if (msgcb == NULL || msgcb->putToQueueFp == NULL) {
+    rpcFreeCont(pMsg->pCont);
+    pMsg->pCont = NULL;
     return -1;
   }
 

source/libs/catalog/src/catalog.c

@@ -320,7 +320,7 @@ _return:
 }
 
 int32_t ctgChkAuth(SCatalog* pCtg, SRequestConnInfo* pConn, const char* user, const char* dbFName, AUTH_TYPE type,
-                   bool* pass) {
+                   bool* pass, bool* exists) {
   bool    inCache = false;
   int32_t code = 0;
 
@@ -329,6 +329,13 @@ int32_t ctgChkAuth(SCatalog* pCtg, SRequestConnInfo* pConn, const char* user, co
   CTG_ERR_RET(ctgChkAuthFromCache(pCtg, (char*)user, (char*)dbFName, type, &inCache, pass));
 
   if (inCache) {
+    if (exists) {
+      *exists = true;
+    }
+    
+    return TSDB_CODE_SUCCESS;
+  } else if (exists) {
+    *exists = false;
     return TSDB_CODE_SUCCESS;
   }
 
@@ -1032,7 +1039,7 @@ int32_t catalogChkTbMetaVersion(SCatalog* pCtg, SRequestConnInfo* pConn, SArray*
       switch (tbType) {
         case TSDB_CHILD_TABLE: {
           SName stb = name;
-          strcpy(stb.tname, stbName);
+          tstrncpy(stb.tname, stbName, sizeof(stb.tname));
           ctgRemoveTbMeta(pCtg, &stb);
           break;
         }
@@ -1373,13 +1380,30 @@ int32_t catalogChkAuth(SCatalog* pCtg, SRequestConnInfo* pConn, const char* user
   }
 
   int32_t code = 0;
-  CTG_ERR_JRET(ctgChkAuth(pCtg, pConn, user, dbFName, type, pass));
+  CTG_ERR_JRET(ctgChkAuth(pCtg, pConn, user, dbFName, type, pass, NULL));
 
 _return:
 
   CTG_API_LEAVE(code);
 }
 
+int32_t catalogChkAuthFromCache(SCatalog* pCtg, SRequestConnInfo* pConn, const char* user, const char* dbFName, AUTH_TYPE type,
+                                        bool* pass, bool* exists) {
+  CTG_API_ENTER();
+
+  if (NULL == pCtg || NULL == pConn || NULL == user || NULL == dbFName || NULL == pass || NULL == exists) {
+    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
+  }
+
+  int32_t code = 0;
+  CTG_ERR_JRET(ctgChkAuth(pCtg, pConn, user, dbFName, type, pass, exists));
+
+_return:
+
+  CTG_API_LEAVE(code);
+}
+
+
 int32_t catalogGetServerVersion(SCatalog* pCtg, SRequestConnInfo* pConn, char** pVersion) {
   CTG_API_ENTER();
 

source/libs/catalog/src/ctgUtil.c

@@ -924,6 +924,11 @@ int32_t ctgGetVgInfosFromHashValue(SCatalog* pCtg, SCtgTaskReq* tReq, SDBVgInfo*
 
   if (1 == vgNum) {
     void* pIter = taosHashIterate(dbInfo->vgHash, NULL);
+    if (NULL == pIter) {
+      ctgError("empty vgHash, db:%s, vgroup number:%d", dbFName, vgNum);
+      CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
+    }
+    
     for (int32_t i = 0; i < tbNum; ++i) {
       vgInfo = taosMemoryMalloc(sizeof(SVgroupInfo));
       if (NULL == vgInfo) {

source/libs/catalog/test/catalogTests.cpp

@@ -2771,10 +2771,30 @@ TEST(apiTest, catalogChkAuth_test) {
   ASSERT_EQ(code, 0);
 
   bool pass = false;
+  bool exists = false;
+  code = catalogChkAuthFromCache(pCtg, mockPointer, ctgTestUsername, ctgTestDbname, AUTH_TYPE_READ, &pass, &exists);
+  ASSERT_EQ(code, 0);
+  ASSERT_EQ(exists, false);
+  
   code = catalogChkAuth(pCtg, mockPointer, ctgTestUsername, ctgTestDbname, AUTH_TYPE_READ, &pass);
   ASSERT_EQ(code, 0);
   ASSERT_EQ(pass, true);
 
+  while (true) {
+    uint64_t n = 0;
+    ctgdGetStatNum("runtime.numOfOpDequeue", (void *)&n);
+    if (n != 1) {
+      taosMsleep(50);
+    } else {
+      break;
+    }
+  }
+
+  code = catalogChkAuthFromCache(pCtg, mockPointer, ctgTestUsername, ctgTestDbname, AUTH_TYPE_READ, &pass, &exists);
+  ASSERT_EQ(code, 0);
+  ASSERT_EQ(pass, true);
+  ASSERT_EQ(exists, true);
+
   catalogDestroy();
 }
 

source/libs/executor/src/groupoperator.c

@@ -62,7 +62,7 @@ static int32_t initGroupOptrInfo(SArray** pGroupColVals, int32_t* keyLen, char**
 
   int32_t numOfGroupCols = taosArrayGetSize(pGroupColList);
   for (int32_t i = 0; i < numOfGroupCols; ++i) {
-    SColumn* pCol = (SColumn*) taosArrayGet(pGroupColList, i);
+    SColumn* pCol = (SColumn*)taosArrayGet(pGroupColList, i);
     (*keyLen) += pCol->bytes;  // actual data + null_flag
 
     SGroupKeys key = {0};
@@ -397,7 +397,7 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
   return buildGroupResultDataBlock(pOperator);
 }
 
-SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode *pAggNode, SExecTaskInfo* pTaskInfo) {
+SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode* pAggNode, SExecTaskInfo* pTaskInfo) {
   SGroupbyOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SGroupbyOperatorInfo));
   SOperatorInfo*        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
@@ -442,8 +442,8 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode
   pOperator->info = pInfo;
   pOperator->pTaskInfo = pTaskInfo;
 
-  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, hashGroupbyAggregate, NULL, NULL,
-                                         destroyGroupOperatorInfo, NULL);
+  pOperator->fpSet =
+      createOperatorFpSet(operatorDummyOpenFn, hashGroupbyAggregate, NULL, NULL, destroyGroupOperatorInfo, NULL);
   code = appendDownstream(pOperator, &downstream, 1);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
@@ -765,7 +765,6 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition
     goto _error;
   }
 
-
   int32_t    numOfCols = 0;
   SExprInfo* pExprInfo = createExprInfo(pPartNode->pTargets, NULL, &numOfCols);
   pInfo->pGroupCols = extractPartitionColInfo(pPartNode->pPartitionKeys);
@@ -819,8 +818,8 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition
   pOperator->info = pInfo;
   pOperator->pTaskInfo = pTaskInfo;
 
-  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, hashPartition, NULL, NULL, destroyPartitionOperatorInfo,
-                                         NULL);
+  pOperator->fpSet =
+      createOperatorFpSet(operatorDummyOpenFn, hashPartition, NULL, NULL, destroyPartitionOperatorInfo, NULL);
 
   code = appendDownstream(pOperator, &downstream, 1);
   return pOperator;
@@ -965,6 +964,7 @@ static SSDataBlock* doStreamHashPartition(SOperatorInfo* pOperator) {
       case STREAM_DELETE_DATA: {
         copyDataBlock(pInfo->pDelRes, pBlock);
         pInfo->pDelRes->info.type = STREAM_DELETE_RESULT;
+        printDataBlock(pInfo->pDelRes, "stream partitionby delete");
         return pInfo->pDelRes;
       } break;
       default:
@@ -1014,6 +1014,9 @@ void initParDownStream(SOperatorInfo* downstream, SPartitionBySupporter* pParSup
   SStreamScanInfo* pScanInfo = downstream->info;
   pScanInfo->partitionSup = *pParSup;
   pScanInfo->pPartScalarSup = pExpr;
+  if (!pScanInfo->pUpdateInfo) {
+    pScanInfo->pUpdateInfo = updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, 0);
+  }
 }
 
 SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStreamPartitionPhysiNode* pPartNode,
@@ -1108,7 +1111,6 @@ _error:
   return NULL;
 }
 
-
 SArray* extractColumnInfo(SNodeList* pNodeList) {
   size_t  numOfCols = LIST_LENGTH(pNodeList);
   SArray* pList = taosArrayInit(numOfCols, sizeof(SColumn));

source/libs/executor/src/scanoperator.c

@@ -364,7 +364,8 @@ void applyLimitOffset(SLimitInfo* pLimitInfo, SSDataBlock* pBlock, SExecTaskInfo
     if (pLimitInfo->remainOffset >= pBlock->info.rows) {
       pLimitInfo->remainOffset -= pBlock->info.rows;
       pBlock->info.rows = 0;
-      qDebug("current block ignore due to offset, current:%"PRId64", %s", pLimitInfo->remainOffset, GET_TASKID(pTaskInfo));
+      qDebug("current block ignore due to offset, current:%" PRId64 ", %s", pLimitInfo->remainOffset,
+             GET_TASKID(pTaskInfo));
     } else {
       blockDataTrimFirstNRows(pBlock, pLimitInfo->remainOffset);
       pLimitInfo->remainOffset = 0;
@@ -377,7 +378,7 @@ void applyLimitOffset(SLimitInfo* pLimitInfo, SSDataBlock* pBlock, SExecTaskInfo
     int32_t keep = pBlock->info.rows - overflowRows;
 
     blockDataKeepFirstNRows(pBlock, keep);
-    qDebug("output limit %"PRId64" has reached, %s", pLimit->limit, GET_TASKID(pTaskInfo));
+    qDebug("output limit %" PRId64 " has reached, %s", pLimit->limit, GET_TASKID(pTaskInfo));
     pOperator->status = OP_EXEC_DONE;
   }
 }
@@ -770,13 +771,13 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
         return NULL;
       }
 
-      int32_t num = 0;
+      int32_t        num = 0;
       STableKeyInfo* pList = NULL;
       tableListGetGroupList(pTaskInfo->pTableInfoList, pInfo->currentGroupId, &pList, &num);
       ASSERT(pInfo->dataReader == NULL);
 
-      int32_t code = tsdbReaderOpen(pInfo->readHandle.vnode, &pInfo->cond, pList, num, (STsdbReader**)&pInfo->dataReader,
-                                    GET_TASKID(pTaskInfo));
+      int32_t code = tsdbReaderOpen(pInfo->readHandle.vnode, &pInfo->cond, pList, num,
+                                    (STsdbReader**)&pInfo->dataReader, GET_TASKID(pTaskInfo));
       if (code != TSDB_CODE_SUCCESS) {
         T_LONG_JMP(pTaskInfo->env, code);
       }
@@ -798,7 +799,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
     pInfo->limitInfo.numOfOutputRows = 0;
     pInfo->limitInfo.remainOffset = pInfo->limitInfo.limit.offset;
 
-    int32_t num = 0;
+    int32_t        num = 0;
     STableKeyInfo* pList = NULL;
     tableListGetGroupList(pTaskInfo->pTableInfoList, pInfo->currentGroupId, &pList, &num);
 
@@ -841,8 +842,8 @@ static void destroyTableScanOperatorInfo(void* param) {
   taosMemoryFreeClear(param);
 }
 
-SOperatorInfo*  createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,
-                                            SExecTaskInfo* pTaskInfo) {
+SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,
+                                           SExecTaskInfo* pTaskInfo) {
   STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));
   SOperatorInfo*  pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
@@ -906,7 +907,7 @@ SOperatorInfo*  createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
   pOperator->cost.openCost = 0;
   return pOperator;
 
-  _error:
+_error:
   if (pInfo != NULL) {
     destroyTableScanOperatorInfo(pInfo);
   }
@@ -1046,7 +1047,8 @@ static int32_t initTableblockDistQueryCond(uint64_t uid, SQueryTableDataCond* pC
   return TSDB_CODE_SUCCESS;
 }
 
-SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDistScanPhysiNode* pBlockScanNode, SExecTaskInfo* pTaskInfo) {
+SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDistScanPhysiNode* pBlockScanNode,
+                                               SExecTaskInfo* pTaskInfo) {
   SBlockDistInfo* pInfo = taosMemoryCalloc(1, sizeof(SBlockDistInfo));
   SOperatorInfo*  pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
@@ -1063,8 +1065,8 @@ SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDi
     }
 
     STableListInfo* pTableListInfo = pTaskInfo->pTableInfoList;
-    size_t num = tableListGetSize(pTableListInfo);
-    void*  pList = tableListGetInfo(pTableListInfo, 0);
+    size_t          num = tableListGetSize(pTableListInfo);
+    void*           pList = tableListGetInfo(pTableListInfo, 0);
 
     tsdbReaderOpen(readHandle->vnode, &cond, pList, num, &pInfo->pHandle, pTaskInfo->id.str);
     cleanupQueryTableDataCond(&cond);
@@ -1092,7 +1094,7 @@ SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDi
       createOperatorFpSet(operatorDummyOpenFn, doBlockInfoScan, NULL, NULL, destroyBlockDistScanOperatorInfo, NULL);
   return pOperator;
 
-  _error:
+_error:
   taosMemoryFreeClear(pInfo);
   taosMemoryFreeClear(pOperator);
   return NULL;
@@ -1157,8 +1159,8 @@ static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbU
   blockDataCleanup(pBlock);
 
   STsdbReader* pReader = NULL;
-  int32_t code = tsdbReaderOpen(pTableScanInfo->readHandle.vnode, &cond, &tblInfo, 1, (STsdbReader**)&pReader,
-                                GET_TASKID(pTaskInfo));
+  int32_t      code = tsdbReaderOpen(pTableScanInfo->readHandle.vnode, &cond, &tblInfo, 1, (STsdbReader**)&pReader,
+                                     GET_TASKID(pTaskInfo));
   if (code != TSDB_CODE_SUCCESS) {
     terrno = code;
     return NULL;
@@ -1183,7 +1185,8 @@ static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbU
 
   tsdbReaderClose(pReader);
   qDebug("retrieve prev rows:%d, skey:%" PRId64 ", ekey:%" PRId64 " uid:%" PRIu64 ", max ver:%" PRId64
-             ", suid:%" PRIu64, pBlock->info.rows, startTs, endTs, tbUid, maxVersion, cond.suid);
+         ", suid:%" PRIu64,
+         pBlock->info.rows, startTs, endTs, tbUid, maxVersion, cond.suid);
 
   return pBlock->info.rows > 0 ? pBlock : NULL;
 }
@@ -1395,7 +1398,7 @@ static int32_t generateIntervalScanRange(SStreamScanInfo* pInfo, SSDataBlock* pS
   if (rows == 0) {
     return TSDB_CODE_SUCCESS;
   }
-  int32_t code = blockDataEnsureCapacity(pDestBlock, rows * 2);
+  int32_t code = blockDataEnsureCapacity(pDestBlock, rows);
   if (code != TSDB_CODE_SUCCESS) {
     return code;
   }
@@ -1438,39 +1441,33 @@ static int32_t generateIntervalScanRange(SStreamScanInfo* pInfo, SSDataBlock* pS
 }
 
 static int32_t generateDeleteResultBlock(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock, SSDataBlock* pDestBlock) {
-  if (pSrcBlock->info.rows == 0) {
+  blockDataCleanup(pDestBlock);
+  int32_t rows = pSrcBlock->info.rows;
+  if (rows == 0) {
     return TSDB_CODE_SUCCESS;
   }
-  blockDataCleanup(pDestBlock);
-  int32_t code = blockDataEnsureCapacity(pDestBlock, pSrcBlock->info.rows);
+  int32_t code = blockDataEnsureCapacity(pDestBlock, rows);
   if (code != TSDB_CODE_SUCCESS) {
     return code;
   }
-  ASSERT(taosArrayGetSize(pSrcBlock->pDataBlock) >= 3);
-  SColumnInfoData* pStartTsCol = taosArrayGet(pSrcBlock->pDataBlock, START_TS_COLUMN_INDEX);
-  TSKEY*           startData = (TSKEY*)pStartTsCol->pData;
-  SColumnInfoData* pEndTsCol = taosArrayGet(pSrcBlock->pDataBlock, END_TS_COLUMN_INDEX);
-  TSKEY*           endData = (TSKEY*)pEndTsCol->pData;
-  SColumnInfoData* pUidCol = taosArrayGet(pSrcBlock->pDataBlock, UID_COLUMN_INDEX);
-  uint64_t*        uidCol = (uint64_t*)pUidCol->pData;
 
-  SColumnInfoData* pDestStartCol = taosArrayGet(pDestBlock->pDataBlock, START_TS_COLUMN_INDEX);
-  SColumnInfoData* pDestEndCol = taosArrayGet(pDestBlock->pDataBlock, END_TS_COLUMN_INDEX);
-  SColumnInfoData* pDestUidCol = taosArrayGet(pDestBlock->pDataBlock, UID_COLUMN_INDEX);
-  SColumnInfoData* pDestGpCol = taosArrayGet(pDestBlock->pDataBlock, GROUPID_COLUMN_INDEX);
-  SColumnInfoData* pDestCalStartTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
-  SColumnInfoData* pDestCalEndTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
-  int32_t          dummy = 0;
-  int64_t          version = pSrcBlock->info.version - 1;
+  SColumnInfoData* pSrcStartTsCol = (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, START_TS_COLUMN_INDEX);
+  SColumnInfoData* pSrcEndTsCol = (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, END_TS_COLUMN_INDEX);
+  SColumnInfoData* pSrcUidCol = taosArrayGet(pSrcBlock->pDataBlock, UID_COLUMN_INDEX);
+  uint64_t*        srcUidData = (uint64_t*)pSrcUidCol->pData;
+  SColumnInfoData* pSrcGpCol = taosArrayGet(pSrcBlock->pDataBlock, GROUPID_COLUMN_INDEX);
+  uint64_t*        srcGp = (uint64_t*)pSrcGpCol->pData;
+  ASSERT(pSrcStartTsCol->info.type == TSDB_DATA_TYPE_TIMESTAMP);
+  TSKEY*  srcStartTsCol = (TSKEY*)pSrcStartTsCol->pData;
+  TSKEY*  srcEndTsCol = (TSKEY*)pSrcEndTsCol->pData;
+  int64_t version = pSrcBlock->info.version - 1;
   for (int32_t i = 0; i < pSrcBlock->info.rows; i++) {
-    uint64_t groupId = getGroupIdByData(pInfo, uidCol[i], startData[i], version);
-    colDataAppend(pDestStartCol, i, (const char*)(startData + i), false);
-    colDataAppend(pDestEndCol, i, (const char*)(endData + i), false);
-    colDataAppendNULL(pDestUidCol, i);
-    colDataAppend(pDestGpCol, i, (const char*)&groupId, false);
-    colDataAppendNULL(pDestCalStartTsCol, i);
-    colDataAppendNULL(pDestCalEndTsCol, i);
-    pDestBlock->info.rows++;
+    uint64_t srcUid = srcUidData[i];
+    uint64_t groupId = srcGp[i];
+    if (groupId == 0) {
+      groupId = getGroupIdByData(pInfo, srcUid, srcStartTsCol[i], version);
+    }
+    appendOneRowToStreamSpecialBlock(pDestBlock, srcStartTsCol + i, srcEndTsCol + i, srcUidData + i, &groupId, NULL);
   }
   return TSDB_CODE_SUCCESS;
 }
@@ -1481,6 +1478,8 @@ static int32_t generateScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock,
     code = generateIntervalScanRange(pInfo, pSrcBlock, pDestBlock);
   } else if (isSessionWindow(pInfo) || isStateWindow(pInfo)) {
     code = generateSessionScanRange(pInfo, pSrcBlock, pDestBlock);
+  } else {
+    code = generateDeleteResultBlock(pInfo, pSrcBlock, pDestBlock);
   }
   pDestBlock->info.type = STREAM_CLEAR;
   pDestBlock->info.version = pSrcBlock->info.version;
@@ -1909,8 +1908,8 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
 #endif
 
   size_t total = taosArrayGetSize(pInfo->pBlockLists);
-  // TODO: refactor
-  FETCH_NEXT_BLOCK:
+// TODO: refactor
+FETCH_NEXT_BLOCK:
   if (pInfo->blockType == STREAM_INPUT__DATA_BLOCK) {
     if (pInfo->validBlockIndex >= total) {
       doClearBufferedBlocks(pInfo);
@@ -2037,7 +2036,7 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
 
     int32_t totBlockNum = taosArrayGetSize(pInfo->pBlockLists);
 
-    NEXT_SUBMIT_BLK:
+  NEXT_SUBMIT_BLK:
     while (1) {
       if (pInfo->tqReader->pMsg == NULL) {
         if (pInfo->validBlockIndex >= totBlockNum) {
@@ -2295,7 +2294,7 @@ SOperatorInfo* createRawScanOperatorInfo(SReadHandle* pHandle, SExecTaskInfo* pT
   pOperator->fpSet = createOperatorFpSet(NULL, doRawScan, NULL, NULL, destroyRawScanOperatorInfo, NULL);
   return pOperator;
 
-  _end:
+_end:
   taosMemoryFree(pInfo);
   taosMemoryFree(pOperator);
   pTaskInfo->code = code;
@@ -2405,7 +2404,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
     }
 
     STableKeyInfo* pList = NULL;
-    int32_t num = 0;
+    int32_t        num = 0;
     tableListGetGroupList(pTaskInfo->pTableInfoList, 0, &pList, &num);
 
     if (pHandle->initTableReader) {
@@ -2484,7 +2483,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
 
   return pOperator;
 
-  _error:
+_error:
   if (pColIds != NULL) {
     taosArrayDestroy(pColIds);
   }
@@ -4119,7 +4118,7 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScan
 
   return pOperator;
 
-  _error:
+_error:
   taosMemoryFreeClear(pInfo);
   taosMemoryFreeClear(pOperator);
   terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
@@ -4258,7 +4257,7 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi
 
   return pOperator;
 
-  _error:
+_error:
   taosMemoryFree(pInfo);
   taosMemoryFree(pOperator);
   terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -4281,8 +4280,8 @@ int32_t createMultipleDataReaders2(SQueryTableDataCond* pQueryCond, SReadHandle*
                                    STableListInfo* pTableListInfo, int32_t tableStartIdx, int32_t tableEndIdx,
                                    STsdbReader** ppReader, const char* idstr) {
   STsdbReader* pReader = NULL;
-  void*   pStart = tableListGetInfo(pTableListInfo, tableStartIdx);
-  int32_t num = tableEndIdx - tableStartIdx + 1;
+  void*        pStart = tableListGetInfo(pTableListInfo, tableStartIdx);
+  int32_t      num = tableEndIdx - tableStartIdx + 1;
 
   int32_t code = tsdbReaderOpen(pHandle->vnode, pQueryCond, pStart, num, &pReader, idstr);
   if (code != 0) {
@@ -4443,7 +4442,7 @@ static int32_t loadDataBlockFromOneTable(SOperatorInfo* pOperator, STableMergeSc
 
     bool             allColumnsHaveAgg = true;
     SColumnDataAgg** pColAgg = NULL;
-//    STsdbReader* reader = pTableScanInfo->pReader; // taosArrayGetP(pTableScanInfo->dataReaders, readerIdx);
+    //    STsdbReader* reader = pTableScanInfo->pReader; // taosArrayGetP(pTableScanInfo->dataReaders, readerIdx);
 
     if (allColumnsHaveAgg == true) {
       int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
@@ -4541,7 +4540,7 @@ static SSDataBlock* getTableDataBlockTemp(void* param) {
 
   int64_t st = taosGetTimestampUs();
 
-  void* p = tableListGetInfo(pInfo->tableListInfo, readIdx + pInfo->tableStartIndex);
+  void*        p = tableListGetInfo(pInfo->tableListInfo, readIdx + pInfo->tableStartIndex);
   SReadHandle* pHandle = &pInfo->readHandle;
   tsdbReaderOpen(pHandle->vnode, pQueryCond, p, 1, &pInfo->pReader, GET_TASKID(pTaskInfo));
 

source/libs/executor/src/timewindowoperator.c

@@ -1687,7 +1687,9 @@ void initIntervalDownStream(SOperatorInfo* downstream, uint16_t type, SAggSuppor
   SStreamScanInfo* pScanInfo = downstream->info;
   pScanInfo->windowSup.parentType = type;
   pScanInfo->windowSup.pIntervalAggSup = pSup;
-  pScanInfo->pUpdateInfo = updateInfoInitP(pInterval, pTwSup->waterMark);
+  if (!pScanInfo->pUpdateInfo) {
+    pScanInfo->pUpdateInfo = updateInfoInitP(pInterval, pTwSup->waterMark);
+  }
   pScanInfo->interval = *pInterval;
   pScanInfo->twAggSup = *pTwSup;
 }
@@ -2453,7 +2455,6 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
           } else {  // non-linear interpolation
             pSliceInfo->current =
                 taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
-
           }
         }
 

source/libs/function/CMakeLists.txt

@@ -12,6 +12,17 @@ target_include_directories(
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
 )
 
+IF (TD_LINUX_64 AND JEMALLOC_ENABLED)
+  ADD_DEFINITIONS(-DTD_JEMALLOC_ENABLED -I${CMAKE_BINARY_DIR}/build/include -L${CMAKE_BINARY_DIR}/build/lib -Wl,-rpath,${CMAKE_BINARY_DIR}/build/lib -ljemalloc)
+  SET(LINK_JEMALLOC "-L${CMAKE_BINARY_DIR}/build/lib -ljemalloc")
+ELSE ()
+  SET(LINK_JEMALLOC "")
+ENDIF ()
+
+IF (TD_LINUX_64 AND JEMALLOC_ENABLED)
+    ADD_DEPENDENCIES(function jemalloc)
+ENDIF ()
+
 target_link_libraries(
     function
     PRIVATE os 
@@ -21,7 +32,7 @@ target_link_libraries(
     PRIVATE qcom 
     PRIVATE scalar 
     PRIVATE transport 
-    PRIVATE stream
+    PRIVATE stream ${LINK_JEMALLOC}
     PUBLIC uv_a
 )
 
@@ -37,10 +48,15 @@ target_include_directories(
             "${TD_SOURCE_DIR}/include/os"
         PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
 )
+
+IF (TD_LINUX_64 AND JEMALLOC_ENABLED)
+    ADD_DEPENDENCIES(runUdf jemalloc)
+ENDIF ()
+
 target_link_libraries(
         runUdf
         PUBLIC uv_a
-        PRIVATE os util common nodes function
+        PRIVATE os util common nodes function ${LINK_JEMALLOC}
 )
 
 add_library(udf1 STATIC MODULE test/udf1.c)
@@ -54,8 +70,13 @@ target_include_directories(
             "${TD_SOURCE_DIR}/include/os"
         PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
 )
+
+IF (TD_LINUX_64 AND JEMALLOC_ENABLED)
+    ADD_DEPENDENCIES(udf1 jemalloc)
+ENDIF ()
+
 target_link_libraries(
-    udf1 PUBLIC os)
+    udf1 PUBLIC os ${LINK_JEMALLOC})
 
 add_library(udf2 STATIC MODULE test/udf2.c)
 target_include_directories(
@@ -68,8 +89,13 @@ target_include_directories(
             "${TD_SOURCE_DIR}/include/os"
         PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
 )
+
+IF (TD_LINUX_64 AND JEMALLOC_ENABLED)
+    ADD_DEPENDENCIES(udf2 jemalloc)
+ENDIF ()
+
 target_link_libraries(
-    udf2 PUBLIC os
+    udf2 PUBLIC os ${LINK_JEMALLOC}
 )
 
 #SET(EXECUTABLE_OUTPUT_PATH ${CMAKE_BINARY_DIR}/build/bin)
@@ -86,9 +112,13 @@ target_include_directories(
         PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
 )
 
+IF (TD_LINUX_64 AND JEMALLOC_ENABLED)
+    ADD_DEPENDENCIES(udfd jemalloc)
+ENDIF ()
+
 target_link_libraries(
-        udfd
-        PUBLIC uv_a
-        PRIVATE os util common nodes function
-)
+    udfd
+    PUBLIC uv_a
+    PRIVATE os util common nodes function ${LINK_JEMALLOC}
+    )
 

source/libs/function/src/tudf.c

@@ -117,10 +117,29 @@ static int32_t udfSpawnUdfd(SUdfdData *pData) {
   char dnodeIdEnvItem[32] = {0};
   char thrdPoolSizeEnvItem[32] = {0};
   snprintf(dnodeIdEnvItem, 32, "%s=%d", "DNODE_ID", pData->dnodeId);
+
   float numCpuCores = 4;
   taosGetCpuCores(&numCpuCores);
   snprintf(thrdPoolSizeEnvItem, 32, "%s=%d", "UV_THREADPOOL_SIZE", (int)numCpuCores * 2);
-  char *envUdfd[] = {dnodeIdEnvItem, thrdPoolSizeEnvItem, NULL};
+
+  char pathTaosdLdLib[512] = {0};
+  size_t taosdLdLibPathLen = sizeof(pathTaosdLdLib);
+  uv_os_getenv("LD_LIBRARY_PATH", pathTaosdLdLib, &taosdLdLibPathLen);
+
+  char udfdPathLdLib[1024] = {0};
+  size_t udfdLdLibPathLen = strlen(tsUdfdLdLibPath);
+  strncpy(udfdPathLdLib, tsUdfdLdLibPath, udfdLdLibPathLen);
+  udfdPathLdLib[udfdLdLibPathLen] = ':';
+  strncpy(udfdPathLdLib + udfdLdLibPathLen + 1, pathTaosdLdLib, sizeof(udfdPathLdLib) - udfdLdLibPathLen);
+  if (udfdLdLibPathLen + taosdLdLibPathLen < 1024) {
+    fnInfo("udfd LD_LIBRARY_PATH: %s", udfdPathLdLib);
+  } else {
+    fnError("can not set correct udfd LD_LIBRARY_PATH");
+  }
+  char ldLibPathEnvItem[1024 + 32] = {0};
+  snprintf(ldLibPathEnvItem, 1024 + 32, "%s=%s", "LD_LIBRARY_PATH", udfdPathLdLib);
+
+  char *envUdfd[] = {dnodeIdEnvItem, thrdPoolSizeEnvItem, ldLibPathEnvItem, NULL};
   options.env = envUdfd;
 
   int err = uv_spawn(&pData->loop, &pData->process, &options);

source/libs/qcom/src/queryUtil.c

@@ -456,6 +456,7 @@ int32_t cloneDbVgInfo(SDBVgInfo* pSrc, SDBVgInfo** pDst) {
     (*pDst)->vgHash = taosHashInit(taosHashGetSize(pSrc->vgHash), taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true,
                                    HASH_ENTRY_LOCK);
     if (NULL == (*pDst)->vgHash) {
+      taosMemoryFreeClear(*pDst);
       return TSDB_CODE_TSC_OUT_OF_MEMORY;
     }
 

source/libs/scalar/src/filter.c

@@ -1087,7 +1087,7 @@ int32_t filterAddUnitImpl(SFilterInfo *info, uint8_t optr, SFilterFieldId *left,
     if (tmp == NULL) {
       return TSDB_CODE_OUT_OF_MEMORY;
     }
-    info->units = tmp;
+    info->units = (SFilterUnit*)tmp;
     memset(info->units + psize, 0, sizeof(*info->units) * FILTER_DEFAULT_UNIT_SIZE);
   }
 
@@ -1633,12 +1633,12 @@ void filterDumpInfoToString(SFilterInfo *info, const char *msg, int32_t options)
 
           SValueNode *var = (SValueNode *)field->desc;
           SDataType  *dType = &var->node.resType;
-          if (dType->type == TSDB_DATA_TYPE_VALUE_ARRAY) {
-            qDebug("VAL%d => [type:TS][val:[%" PRIi64 "] - [%" PRId64 "]]", i, *(int64_t *)field->data,
-                   *(((int64_t *)field->data) + 1));
-          } else {
+          //if (dType->type == TSDB_DATA_TYPE_VALUE_ARRAY) {
+          //  qDebug("VAL%d => [type:TS][val:[%" PRIi64 "] - [%" PRId64 "]]", i, *(int64_t *)field->data,
+          //         *(((int64_t *)field->data) + 1));
+          //} else {
             qDebug("VAL%d => [type:%d][val:%" PRIx64 "]", i, dType->type, var->datum.i);  // TODO
-          }
+          //}
         } else if (field->data) {
           qDebug("VAL%d => [type:NIL][val:NIL]", i);  // TODO
         }
@@ -4059,11 +4059,13 @@ bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, SColumnInfoData **p, SC
     SArray *pList = taosArrayInit(1, POINTER_BYTES);
     taosArrayPush(pList, &pSrc);
 
-    FLT_ERR_RET(scalarCalculate(info->sclCtx.node, pList, &output));
-    *p = output.columnData;
-
+    int32_t code = scalarCalculate(info->sclCtx.node, pList, &output);
     taosArrayDestroy(pList);
 
+    FLT_ERR_RET(code);
+
+    *p = output.columnData;
+
     if (output.numOfQualified == output.numOfRows) {
       *pResultStatus = FILTER_RESULT_ALL_QUALIFIED;
     } else if (output.numOfQualified == 0) {

source/libs/scalar/src/scalar.c

@@ -896,6 +896,10 @@ int32_t sclExecCaseWhen(SCaseWhenNode *node, SScalarCtx *ctx, SScalarParam *outp
 
   SCL_ERR_JRET(sclGetNodeRes(pWhenThen->pWhen, ctx, &pWhen));
   SCL_ERR_JRET(sclGetNodeRes(pWhenThen->pThen, ctx, &pThen));
+  if (NULL == pWhen || NULL == pThen) {
+    sclError("invalid when/then in whenThen list");
+    SCL_ERR_JRET(TSDB_CODE_INVALID_PARA);
+  }
 
   if (pCase) {
     vectorCompare(pCase, pWhen, &comp, TSDB_ORDER_ASC, OP_TYPE_EQUAL);

source/libs/sync/inc/syncEnv.h

@@ -20,13 +20,7 @@
 extern "C" {
 #endif
 
-#include <stdint.h>
-#include <stdio.h>
-#include <stdlib.h>
 #include "syncInt.h"
-#include "taosdef.h"
-#include "trpc.h"
-#include "ttimer.h"
 
 #define TIMER_MAX_MS         0x7FFFFFFF
 #define ENV_TICK_TIMER_MS    1000
@@ -57,12 +51,12 @@ typedef struct SSyncEnv {
 
 } SSyncEnv;
 
-extern SSyncEnv* gSyncEnv;
+SSyncEnv* syncEnv();
 
-int32_t syncEnvStart();
-int32_t syncEnvStop();
-int32_t syncEnvStartTimer();
-int32_t syncEnvStopTimer();
+int64_t    syncNodeAdd(SSyncNode* pNode);
+void       syncNodeRemove(int64_t rid);
+SSyncNode* syncNodeAcquire(int64_t rid);
+void       syncNodeRelease(SSyncNode* pNode);
 
 #ifdef __cplusplus
 }

source/libs/sync/inc/syncInt.h

@@ -24,6 +24,8 @@ extern "C" {
 #include "syncTools.h"
 #include "tlog.h"
 #include "ttimer.h"
+#include "taosdef.h"
+#include "ttimer.h"
 
 // clang-format off
 #define sFatal(...) do { if (sDebugFlag & DEBUG_FATAL) { taosPrintLog("SYN FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }}     while(0)
@@ -255,9 +257,6 @@ void      syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* newConfig, Sync
 SyncIndex syncMinMatchIndex(SSyncNode* pSyncNode);
 char*     syncNodePeerState2Str(const SSyncNode* pSyncNode);
 
-SSyncNode* syncNodeAcquire(int64_t rid);
-void       syncNodeRelease(SSyncNode* pNode);
-
 // raft state change --------------
 void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term);
 void syncNodeUpdateTermWithoutStepDown(SSyncNode* pSyncNode, SyncTerm term);
@@ -302,9 +301,6 @@ bool syncNodeNeedSendAppendEntries(SSyncNode* ths, const SRaftId* pDestId, const
 int32_t syncGetSnapshotMeta(int64_t rid, struct SSnapshotMeta* sMeta);
 int32_t syncGetSnapshotMetaByIndex(int64_t rid, SyncIndex snapshotIndex, struct SSnapshotMeta* sMeta);
 
-void syncStartNormal(int64_t rid);
-void syncStartStandBy(int64_t rid);
-
 bool syncNodeCanChange(SSyncNode* pSyncNode);
 bool syncNodeCheckNewConfig(SSyncNode* pSyncNode, const SSyncCfg* pNewCfg);
 

source/libs/sync/src/syncEnv.c

@@ -13,118 +13,111 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#define _DEFAULT_SOURCE
 #include "syncEnv.h"
-// #include <ASSERT.h>
+#include "tref.h"
 
-SSyncEnv *gSyncEnv = NULL;
+static SSyncEnv gSyncEnv = {0};
+static int32_t  gNodeRefId = -1;
+bool            gRaftDetailLog = false;
+static void     syncEnvTick(void *param, void *tmrId);
 
-// local function -----------------
-static SSyncEnv *doSyncEnvStart();
-static int32_t   doSyncEnvStop(SSyncEnv *pSyncEnv);
-static int32_t   doSyncEnvStartTimer(SSyncEnv *pSyncEnv);
-static int32_t   doSyncEnvStopTimer(SSyncEnv *pSyncEnv);
-static void      syncEnvTick(void *param, void *tmrId);
-// --------------------------------
+SSyncEnv *syncEnv() { return &gSyncEnv; }
 
-bool syncEnvIsStart() {
-  if (gSyncEnv == NULL) {
-    return false;
-  }
+bool syncIsInit() { return atomic_load_8(&gSyncEnv.isStart); }
 
-  return atomic_load_8(&(gSyncEnv->isStart));
-}
+int32_t syncInit() {
+  if (syncIsInit()) return 0;
 
-int32_t syncEnvStart() {
-  int32_t  ret = 0;
   uint32_t seed = (uint32_t)(taosGetTimestampNs() & 0x00000000FFFFFFFF);
   taosSeedRand(seed);
-  // gSyncEnv = doSyncEnvStart(gSyncEnv);
-  gSyncEnv = doSyncEnvStart();
-  ASSERT(gSyncEnv != NULL);
-  sTrace("sync env start ok");
-  return ret;
-}
 
-int32_t syncEnvStop() {
-  int32_t ret = doSyncEnvStop(gSyncEnv);
-  return ret;
-}
+  memset(&gSyncEnv, 0, sizeof(SSyncEnv));
+  gSyncEnv.envTickTimerCounter = 0;
+  gSyncEnv.envTickTimerMS = ENV_TICK_TIMER_MS;
+  gSyncEnv.FpEnvTickTimer = syncEnvTick;
+  atomic_store_64(&gSyncEnv.envTickTimerLogicClock, 0);
+  atomic_store_64(&gSyncEnv.envTickTimerLogicClockUser, 0);
 
-int32_t syncEnvStartTimer() {
-  int32_t ret = doSyncEnvStartTimer(gSyncEnv);
-  return ret;
-}
+  // start tmr thread
+  gSyncEnv.pTimerManager = taosTmrInit(1000, 50, 10000, "SYNC-ENV");
+  atomic_store_8(&gSyncEnv.isStart, 1);
+
+  gNodeRefId = taosOpenRef(200, (RefFp)syncNodeClose);
+  if (gNodeRefId < 0) {
+    sError("failed to init node ref");
+    syncCleanUp();
+    return -1;
+  }
 
-int32_t syncEnvStopTimer() {
-  int32_t ret = doSyncEnvStopTimer(gSyncEnv);
-  return ret;
+  sDebug("sync rsetId:%d is open", gNodeRefId);
+  return 0;
 }
 
-// local function -----------------
-static void syncEnvTick(void *param, void *tmrId) {
-  SSyncEnv *pSyncEnv = (SSyncEnv *)param;
-  if (atomic_load_64(&pSyncEnv->envTickTimerLogicClockUser) <= atomic_load_64(&pSyncEnv->envTickTimerLogicClock)) {
-    ++(pSyncEnv->envTickTimerCounter);
-    sTrace("syncEnvTick do ... envTickTimerLogicClockUser:%" PRIu64 ", envTickTimerLogicClock:%" PRIu64
-           ", envTickTimerCounter:%" PRIu64
-           ", "
-           "envTickTimerMS:%d, tmrId:%p",
-           pSyncEnv->envTickTimerLogicClockUser, pSyncEnv->envTickTimerLogicClock, pSyncEnv->envTickTimerCounter,
-           pSyncEnv->envTickTimerMS, tmrId);
+void syncCleanUp() {
+  atomic_store_8(&gSyncEnv.isStart, 0);
+  taosTmrCleanUp(gSyncEnv.pTimerManager);
+  memset(&gSyncEnv, 0, sizeof(SSyncEnv));
 
-    // do something, tick ...
-    taosTmrReset(syncEnvTick, pSyncEnv->envTickTimerMS, pSyncEnv, pSyncEnv->pTimerManager, &pSyncEnv->pEnvTickTimer);
-  } else {
-    sTrace("syncEnvTick pass ... envTickTimerLogicClockUser:%" PRIu64 ", envTickTimerLogicClock:%" PRIu64
-           ", envTickTimerCounter:%" PRIu64
-           ", "
-           "envTickTimerMS:%d, tmrId:%p",
-           pSyncEnv->envTickTimerLogicClockUser, pSyncEnv->envTickTimerLogicClock, pSyncEnv->envTickTimerCounter,
-           pSyncEnv->envTickTimerMS, tmrId);
+  if (gNodeRefId != -1) {
+    sDebug("sync rsetId:%d is closed", gNodeRefId);
+    taosCloseRef(gNodeRefId);
+    gNodeRefId = -1;
   }
 }
 
-static SSyncEnv *doSyncEnvStart() {
-  SSyncEnv *pSyncEnv = (SSyncEnv *)taosMemoryMalloc(sizeof(SSyncEnv));
-  ASSERT(pSyncEnv != NULL);
-  memset(pSyncEnv, 0, sizeof(SSyncEnv));
+int64_t syncNodeAdd(SSyncNode *pNode) {
+  pNode->rid = taosAddRef(gNodeRefId, pNode);
+  if (pNode->rid < 0) return -1;
 
-  pSyncEnv->envTickTimerCounter = 0;
-  pSyncEnv->envTickTimerMS = ENV_TICK_TIMER_MS;
-  pSyncEnv->FpEnvTickTimer = syncEnvTick;
-  atomic_store_64(&pSyncEnv->envTickTimerLogicClock, 0);
-  atomic_store_64(&pSyncEnv->envTickTimerLogicClockUser, 0);
+  sDebug("vgId:%d, sync rid:%" PRId64 " is added to rsetId:%d", pNode->vgId, pNode->rid, gNodeRefId);
+  return pNode->rid;
+}
 
-  // start tmr thread
-  pSyncEnv->pTimerManager = taosTmrInit(1000, 50, 10000, "SYNC-ENV");
+void syncNodeRemove(int64_t rid) { taosRemoveRef(gNodeRefId, rid); }
+
+SSyncNode *syncNodeAcquire(int64_t rid) {
+  SSyncNode *pNode = taosAcquireRef(gNodeRefId, rid);
+  if (pNode == NULL) {
+    sTrace("failed to acquire node from refId:%" PRId64, rid);
+  }
 
-  atomic_store_8(&(pSyncEnv->isStart), 1);
-  return pSyncEnv;
+  return pNode;
 }
 
-static int32_t doSyncEnvStop(SSyncEnv *pSyncEnv) {
-  ASSERT(pSyncEnv == gSyncEnv);
-  if (pSyncEnv != NULL) {
-    atomic_store_8(&(pSyncEnv->isStart), 0);
-    taosTmrCleanUp(pSyncEnv->pTimerManager);
-    taosMemoryFree(pSyncEnv);
-  }
-  gSyncEnv = NULL;
-  return 0;
+void syncNodeRelease(SSyncNode *pNode) { taosReleaseRef(gNodeRefId, pNode->rid); }
+
+#if 0
+void syncEnvStartTimer() {
+  taosTmrReset(gSyncEnv.FpEnvTickTimer, gSyncEnv.envTickTimerMS, &gSyncEnv, gSyncEnv.pTimerManager,
+               &gSyncEnv.pEnvTickTimer);
+  atomic_store_64(&gSyncEnv.envTickTimerLogicClock, gSyncEnv.envTickTimerLogicClockUser);
 }
 
-static int32_t doSyncEnvStartTimer(SSyncEnv *pSyncEnv) {
+void syncEnvStopTimer() {
   int32_t ret = 0;
-  taosTmrReset(pSyncEnv->FpEnvTickTimer, pSyncEnv->envTickTimerMS, pSyncEnv, pSyncEnv->pTimerManager,
-               &pSyncEnv->pEnvTickTimer);
-  atomic_store_64(&pSyncEnv->envTickTimerLogicClock, pSyncEnv->envTickTimerLogicClockUser);
+  atomic_add_fetch_64(&gSyncEnv.envTickTimerLogicClockUser, 1);
+  taosTmrStop(gSyncEnv.pEnvTickTimer);
+  gSyncEnv.pEnvTickTimer = NULL;
   return ret;
 }
+#endif
 
-static int32_t doSyncEnvStopTimer(SSyncEnv *pSyncEnv) {
-  int32_t ret = 0;
-  atomic_add_fetch_64(&pSyncEnv->envTickTimerLogicClockUser, 1);
-  taosTmrStop(pSyncEnv->pEnvTickTimer);
-  pSyncEnv->pEnvTickTimer = NULL;
-  return ret;
+static void syncEnvTick(void *param, void *tmrId) {
+  SSyncEnv *pSyncEnv = param;
+  if (atomic_load_64(&gSyncEnv.envTickTimerLogicClockUser) <= atomic_load_64(&gSyncEnv.envTickTimerLogicClock)) {
+    gSyncEnv.envTickTimerCounter++;
+    sTrace("syncEnvTick do ... envTickTimerLogicClockUser:%" PRIu64 ", envTickTimerLogicClock:%" PRIu64
+           ", envTickTimerCounter:%" PRIu64 ", envTickTimerMS:%d, tmrId:%p",
+           gSyncEnv.envTickTimerLogicClockUser, gSyncEnv.envTickTimerLogicClock, gSyncEnv.envTickTimerCounter,
+           gSyncEnv.envTickTimerMS, tmrId);
+
+    // do something, tick ...
+    taosTmrReset(syncEnvTick, gSyncEnv.envTickTimerMS, pSyncEnv, gSyncEnv.pTimerManager, &gSyncEnv.pEnvTickTimer);
+  } else {
+    sTrace("syncEnvTick pass ... envTickTimerLogicClockUser:%" PRIu64 ", envTickTimerLogicClock:%" PRIu64
+           ", envTickTimerCounter:%" PRIu64 ", envTickTimerMS:%d, tmrId:%p",
+           gSyncEnv.envTickTimerLogicClockUser, gSyncEnv.envTickTimerLogicClock, gSyncEnv.envTickTimerCounter,
+           gSyncEnv.envTickTimerMS, tmrId);
+  }
 }

source/libs/sync/src/syncTimeout.c

@@ -76,7 +76,7 @@ int32_t syncNodeTimerRoutine(SSyncNode* ths) {
     SSyncLogStoreData* pData = ths->pLogStore->data;
     int32_t            code = walEndSnapshot(pData->pWal);
     if (code != 0) {
-      sError("vgId:%d, wal snapshot end error since:%s", ths->vgId, terrstr(terrno));
+      sError("vgId:%d, timer wal snapshot end error since:%s", ths->vgId, terrstr());
       return -1;
     } else {
       do {

source/libs/sync/test/syncElectTest.cpp

@@ -98,7 +98,7 @@ int main(int argc, char** argv) {
   init();
   int32_t ret = syncIOStart((char*)"127.0.0.1", gPorts[myIndex]);
   assert(ret == 0);
-  ret = syncEnvStart();
+  ret = syncInit();
   assert(ret == 0);
 
   char walPath[128];

source/libs/sync/test/syncEncodeTest.cpp

@@ -152,7 +152,7 @@ int main(int argc, char **argv) {
   int32_t ret = syncIOStart((char *)"127.0.0.1", ports[myIndex]);
   assert(ret == 0);
 
-  ret = syncEnvStart();
+  ret = syncInit();
   assert(ret == 0);
 
   taosRemoveDir("./wal_test");

source/libs/sync/test/syncEnqTest.cpp

@@ -81,7 +81,7 @@ int main(int argc, char** argv) {
   int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
   assert(ret == 0);
 
-  ret = syncEnvStart();
+  ret = syncInit();
   assert(ret == 0);
 
   SSyncNode* pSyncNode = syncInitTest();

source/libs/sync/test/syncEnvTest.cpp

@@ -22,7 +22,7 @@ int main() {
 
   logTest();
 
-  ret = syncEnvStart();
+  ret = syncInit();
   assert(ret == 0);
 
   for (int i = 0; i < 5; ++i) {
@@ -37,8 +37,6 @@ int main() {
     taosMsleep(5000);
   }
 
-  ret = syncEnvStop();
-  assert(ret == 0);
-
+  syncCleanUp();
   return 0;
 }

source/libs/sync/test/syncIOSendMsgTest.cpp

@@ -82,7 +82,7 @@ int main(int argc, char** argv) {
   int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
   assert(ret == 0);
 
-  ret = syncEnvStart();
+  ret = syncInit();
   assert(ret == 0);
 
   SSyncNode* pSyncNode = syncInitTest();

source/libs/sync/test/syncInitTest.cpp

@@ -81,7 +81,7 @@ int main(int argc, char** argv) {
   int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
   assert(ret == 0);
 
-  ret = syncEnvStart();
+  ret = syncInit();
   assert(ret == 0);
 
   SSyncNode* pSyncNode = syncInitTest();
@@ -91,7 +91,7 @@ int main(int argc, char** argv) {
   initRaftId(pSyncNode);
 
   syncNodeClose(pSyncNode);
-  syncEnvStop();
+  syncCleanUp();
   // syncIOStop();
 
   // taosCloseLog();

source/libs/sync/test/syncPingSelfTest.cpp

@@ -81,7 +81,7 @@ int main(int argc, char** argv) {
   int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
   assert(ret == 0);
 
-  ret = syncEnvStart();
+  ret = syncInit();
   assert(ret == 0);
 
   SSyncNode* pSyncNode = syncInitTest();

source/libs/sync/test/syncPingTimerTest.cpp

@@ -81,7 +81,7 @@ int main(int argc, char** argv) {
   int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
   assert(ret == 0);
 
-  ret = syncEnvStart();
+  ret = syncInit();
   assert(ret == 0);
 
   SSyncNode* pSyncNode = syncInitTest();

source/libs/sync/test/syncPingTimerTest2.cpp

@@ -81,7 +81,7 @@ int main(int argc, char** argv) {
   int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
   assert(ret == 0);
 
-  ret = syncEnvStart();
+  ret = syncInit();
   assert(ret == 0);
 
   SSyncNode* pSyncNode = syncInitTest();

source/libs/sync/test/syncSnapshotTest.cpp

@@ -179,7 +179,7 @@ int main(int argc, char **argv) {
   int32_t ret = syncIOStart((char *)"127.0.0.1", ports[myIndex]);
   assert(ret == 0);
 
-  ret = syncEnvStart();
+  ret = syncInit();
   assert(ret == 0);
 
   // taosRemoveDir(pWalDir);

source/libs/sync/test/syncVotesGrantedTest.cpp

@@ -82,7 +82,7 @@ int main(int argc, char** argv) {
   int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
   assert(ret == 0);
 
-  ret = syncEnvStart();
+  ret = syncInit();
   assert(ret == 0);
 
   SSyncNode* pSyncNode = syncInitTest();

source/libs/sync/test/syncVotesRespondTest.cpp

@@ -82,7 +82,7 @@ int main(int argc, char** argv) {
   int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
   assert(ret == 0);
 
-  ret = syncEnvStart();
+  ret = syncInit();
   assert(ret == 0);
 
   SSyncNode* pSyncNode = syncInitTest();

source/libs/sync/test/syncWriteTest.cpp

@@ -154,7 +154,7 @@ int main(int argc, char **argv) {
   int32_t ret = syncIOStart((char *)"127.0.0.1", ports[myIndex]);
   assert(ret == 0);
 
-  ret = syncEnvStart();
+  ret = syncInit();
   assert(ret == 0);
 
   taosRemoveDir("./wal_test");

source/libs/transport/inc/transComm.h

@@ -428,6 +428,7 @@ void    transDestoryExHandle(void* handle);
 int32_t transGetRefMgt();
 int32_t transGetInstMgt();
 
+void transHttpEnvDestroy();
 #ifdef __cplusplus
 }
 #endif

source/libs/transport/src/thttp.c

@@ -20,21 +20,52 @@
 #include "thttp.h"
 #include "taoserror.h"
 #include "tlog.h"
+#include "transComm.h"
 
 // clang-format on
 
 #define HTTP_RECV_BUF_SIZE 1024
 
+typedef struct SHttpModule {
+  uv_loop_t*  loop;
+  SAsyncPool* asyncPool;
+  TdThread    thread;
+} SHttpModule;
+
+typedef struct SHttpMsg {
+  queue         q;
+  char*         server;
+  int32_t       port;
+  char*         cont;
+  int32_t       len;
+  EHttpCompFlag flag;
+  int8_t        quit;
+  SHttpModule*  http;
+
+} SHttpMsg;
+
 typedef struct SHttpClient {
-  uv_connect_t conn;
-  uv_tcp_t     tcp;
-  uv_write_t   req;
-  uv_buf_t*    wbuf;
-  char*        rbuf;
-  char*        addr;
-  uint16_t     port;
+  uv_connect_t       conn;
+  uv_tcp_t           tcp;
+  uv_write_t         req;
+  uv_buf_t*          wbuf;
+  char*              rbuf;
+  char*              addr;
+  uint16_t           port;
+  struct sockaddr_in dest;
 } SHttpClient;
 
+static TdThreadOnce transHttpInit = PTHREAD_ONCE_INIT;
+static SHttpModule* thttp = NULL;
+static void         transHttpEnvInit();
+
+static void    httpHandleReq(SHttpMsg* msg);
+static void    httpHandleQuit(SHttpMsg* msg);
+static int32_t httpSendQuit();
+
+static int32_t taosSendHttpReportImpl(const char* server, uint16_t port, char* pCont, int32_t contLen,
+                                      EHttpCompFlag flag);
+
 static int32_t taosBuildHttpHeader(const char* server, int32_t contLen, char* pHead, int32_t headLen,
                                    EHttpCompFlag flag) {
   if (flag == HTTP_FLAT) {
@@ -53,6 +84,7 @@ static int32_t taosBuildHttpHeader(const char* server, int32_t contLen, char* pH
                     "Content-Length: %d\n\n",
                     server, contLen);
   } else {
+    terrno = TSDB_CODE_INVALID_CFG;
     return -1;
   }
 }
@@ -126,116 +158,217 @@ _OVER:
   return code;
 }
 
+static FORCE_INLINE int32_t taosBuildDstAddr(const char* server, uint16_t port, struct sockaddr_in* dest) {
+  uint32_t ip = taosGetIpv4FromFqdn(server);
+  if (ip == 0xffffffff) {
+    tError("http-report failed to get http server:%s since %s", server, errno == 0 ? "invalid http server" : terrstr());
+    return -1;
+  }
+  char buf[128] = {0};
+  tinet_ntoa(buf, ip);
+  uv_ip4_addr(buf, port, dest);
+  return 0;
+}
+
+static void* httpThread(void* arg) {
+  SHttpModule* http = (SHttpModule*)arg;
+  setThreadName("http-cli-send-thread");
+  uv_run(http->loop, UV_RUN_DEFAULT);
+  return NULL;
+}
+
+static void httpDestroyMsg(SHttpMsg* msg) {
+  if (msg == NULL) return;
+
+  taosMemoryFree(msg->server);
+  taosMemoryFree(msg->cont);
+  taosMemoryFree(msg);
+}
+static void httpAsyncCb(uv_async_t* handle) {
+  SAsyncItem*  item = handle->data;
+  SHttpModule* http = item->pThrd;
+
+  SHttpMsg *msg = NULL, *quitMsg = NULL;
+
+  queue wq;
+  taosThreadMutexLock(&item->mtx);
+  QUEUE_MOVE(&item->qmsg, &wq);
+  taosThreadMutexUnlock(&item->mtx);
+
+  int count = 0;
+  while (!QUEUE_IS_EMPTY(&wq)) {
+    queue* h = QUEUE_HEAD(&wq);
+    QUEUE_REMOVE(h);
+    msg = QUEUE_DATA(h, SHttpMsg, q);
+    if (msg->quit) {
+      quitMsg = msg;
+    } else {
+      httpHandleReq(msg);
+    }
+  }
+  if (quitMsg) httpHandleQuit(quitMsg);
+}
+
 static FORCE_INLINE void destroyHttpClient(SHttpClient* cli) {
+  taosMemoryFree(cli->wbuf[0].base);
+  taosMemoryFree(cli->wbuf[1].base);
   taosMemoryFree(cli->wbuf);
   taosMemoryFree(cli->rbuf);
   taosMemoryFree(cli->addr);
   taosMemoryFree(cli);
 }
+
 static FORCE_INLINE void clientCloseCb(uv_handle_t* handle) {
   SHttpClient* cli = handle->data;
   destroyHttpClient(cli);
 }
+
 static FORCE_INLINE void clientAllocBuffCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
   SHttpClient* cli = handle->data;
   buf->base = cli->rbuf;
   buf->len = HTTP_RECV_BUF_SIZE;
 }
+
 static FORCE_INLINE void clientRecvCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) {
   SHttpClient* cli = handle->data;
   if (nread < 0) {
-    uError("http-report recv error:%s", uv_err_name(nread));
+    tError("http-report recv error:%s", uv_err_name(nread));
   } else {
-    uTrace("http-report succ to recv %d bytes", (int32_t)nread);
+    tTrace("http-report succ to recv %d bytes", (int32_t)nread);
   }
   if (!uv_is_closing((uv_handle_t*)&cli->tcp)) {
     uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
-  } else {
-    destroyHttpClient(cli);
   }
 }
 static void clientSentCb(uv_write_t* req, int32_t status) {
   SHttpClient* cli = req->data;
   if (status != 0) {
-    terrno = TAOS_SYSTEM_ERROR(status);
-    uError("http-report failed to send data %s", uv_strerror(status));
+    tError("http-report failed to send data, reason: %s, dst:%s:%d", uv_strerror(status), cli->addr, cli->port);
     if (!uv_is_closing((uv_handle_t*)&cli->tcp)) {
       uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
-    } else {
-      destroyHttpClient(cli);
     }
     return;
   } else {
-    uTrace("http-report succ to send data");
+    tTrace("http-report succ to send data");
   }
   status = uv_read_start((uv_stream_t*)&cli->tcp, clientAllocBuffCb, clientRecvCb);
   if (status != 0) {
-    terrno = TAOS_SYSTEM_ERROR(status);
-    uError("http-report failed to recv data,reason:%s, dst:%s:%d", uv_strerror(status), cli->addr, cli->port);
+    tError("http-report failed to recv data,reason:%s, dst:%s:%d", uv_strerror(status), cli->addr, cli->port);
     if (!uv_is_closing((uv_handle_t*)&cli->tcp)) {
       uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
-    } else {
-      destroyHttpClient(cli);
     }
   }
 }
 static void clientConnCb(uv_connect_t* req, int32_t status) {
   SHttpClient* cli = req->data;
   if (status != 0) {
-    terrno = TAOS_SYSTEM_ERROR(status);
-    uError("http-report failed to conn to server, reason:%s, dst:%s:%d", uv_strerror(status), cli->addr, cli->port);
+    tError("http-report failed to conn to server, reason:%s, dst:%s:%d", uv_strerror(status), cli->addr, cli->port);
     if (!uv_is_closing((uv_handle_t*)&cli->tcp)) {
       uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
-    } else {
-      destroyHttpClient(cli);
     }
     return;
   }
   status = uv_write(&cli->req, (uv_stream_t*)&cli->tcp, cli->wbuf, 2, clientSentCb);
   if (0 != status) {
-    terrno = TAOS_SYSTEM_ERROR(status);
-    uError("http-report failed to send data,reason:%s, dst:%s:%d", uv_strerror(status), cli->addr, cli->port);
+    tError("http-report failed to send data,reason:%s, dst:%s:%d", uv_strerror(status), cli->addr, cli->port);
     if (!uv_is_closing((uv_handle_t*)&cli->tcp)) {
       uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
-    } else {
-      destroyHttpClient(cli);
     }
   }
 }
 
-static FORCE_INLINE int32_t taosBuildDstAddr(const char* server, uint16_t port, struct sockaddr_in* dest) {
-  uint32_t ip = taosGetIpv4FromFqdn(server);
-  if (ip == 0xffffffff) {
-    terrno = TAOS_SYSTEM_ERROR(errno);
-    uError("http-report failed to get http server:%s since %s", server, errno == 0 ? "invalid http server" : terrstr());
+int32_t httpSendQuit() {
+  SHttpMsg* msg = taosMemoryCalloc(1, sizeof(SHttpMsg));
+  msg->quit = 1;
+
+  SHttpModule* load = atomic_load_ptr(&thttp);
+  if (load == NULL) {
+    httpDestroyMsg(msg);
+    tError("http-report already released");
     return -1;
+  } else {
+    msg->http = load;
   }
-  char buf[128] = {0};
-  tinet_ntoa(buf, ip);
-  uv_ip4_addr(buf, port, dest);
+  transAsyncSend(load->asyncPool, &(msg->q));
   return 0;
 }
-int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag) {
-  struct sockaddr_in dest = {0};
-  if (taosBuildDstAddr(server, port, &dest) < 0) {
+
+static int32_t taosSendHttpReportImpl(const char* server, uint16_t port, char* pCont, int32_t contLen,
+                                      EHttpCompFlag flag) {
+  SHttpMsg* msg = taosMemoryMalloc(sizeof(SHttpMsg));
+  msg->server = strdup(server);
+  msg->port = port;
+  msg->cont = taosMemoryMalloc(contLen);
+  memcpy(msg->cont, pCont, contLen);
+  msg->len = contLen;
+  msg->flag = flag;
+  msg->quit = 0;
+
+  SHttpModule* load = atomic_load_ptr(&thttp);
+  if (load == NULL) {
+    httpDestroyMsg(msg);
+    tError("http-report already released");
     return -1;
+  } else {
+    msg->http = load;
+    transAsyncSend(load->asyncPool, &(msg->q));
+  }
+
+  return 0;
+}
+
+static void httpDestroyClientCb(uv_handle_t* handle) {
+  SHttpClient* http = handle->data;
+  destroyHttpClient(http);
+}
+static void httpWalkCb(uv_handle_t* handle, void* arg) {
+  // impl later
+  if (!uv_is_closing(handle)) {
+    uv_handle_type type = uv_handle_get_type(handle);
+    if (uv_handle_get_type(handle) == UV_TCP) {
+      uv_close(handle, httpDestroyClientCb);
+    } else {
+      uv_close(handle, NULL);
+    }
   }
-  if (flag == HTTP_GZIP) {
-    int32_t dstLen = taosCompressHttpRport(pCont, contLen);
+  return;
+}
+static void httpHandleQuit(SHttpMsg* msg) {
+  SHttpModule* http = msg->http;
+  taosMemoryFree(msg);
+
+  uv_walk(http->loop, httpWalkCb, NULL);
+}
+static void httpHandleReq(SHttpMsg* msg) {
+  SHttpModule* http = msg->http;
+
+  struct sockaddr_in dest = {0};
+  if (taosBuildDstAddr(msg->server, msg->port, &dest) < 0) {
+    goto END;
+  }
+  if (msg->flag == HTTP_GZIP) {
+    int32_t dstLen = taosCompressHttpRport(msg->cont, msg->len);
     if (dstLen > 0) {
-      contLen = dstLen;
+      msg->len = dstLen;
     } else {
-      flag = HTTP_FLAT;
+      msg->flag = HTTP_FLAT;
+    }
+    if (dstLen < 0) {
+      goto END;
     }
   }
-  terrno = 0;
 
-  char    header[2048] = {0};
-  int32_t headLen = taosBuildHttpHeader(server, contLen, header, sizeof(header), flag);
+  int32_t len = 2048;
+  char*   header = taosMemoryCalloc(1, len);
+  int32_t headLen = taosBuildHttpHeader(msg->server, msg->len, header, len, msg->flag);
+  if (headLen < 0) {
+    taosMemoryFree(header);
+    goto END;
+  }
 
   uv_buf_t* wb = taosMemoryCalloc(2, sizeof(uv_buf_t));
-  wb[0] = uv_buf_init((char*)header, headLen);  // stack var
-  wb[1] = uv_buf_init((char*)pCont, contLen);   //  heap var
+  wb[0] = uv_buf_init((char*)header, strlen(header));  //  heap var
+  wb[1] = uv_buf_init((char*)msg->cont, msg->len);     //  heap var
 
   SHttpClient* cli = taosMemoryCalloc(1, sizeof(SHttpClient));
   cli->conn.data = cli;
@@ -243,41 +376,71 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32
   cli->req.data = cli;
   cli->wbuf = wb;
   cli->rbuf = taosMemoryCalloc(1, HTTP_RECV_BUF_SIZE);
-  cli->addr = tstrdup(server);
-  cli->port = port;
+  cli->addr = msg->server;
+  cli->port = msg->port;
+  cli->dest = dest;
+
+  taosMemoryFree(msg);
+
+  uv_tcp_init(http->loop, &cli->tcp);
 
-  uv_loop_t* loop = taosMemoryMalloc(sizeof(uv_loop_t));
-  int        err = uv_loop_init(loop);
-  if (err != 0) {
-    uError("http-report failed to init uv_loop, reason: %s", uv_strerror(err));
-    taosMemoryFree(loop);
-    terrno = TAOS_SYSTEM_ERROR(err);
-    destroyHttpClient(cli);
-    return terrno;
-  }
-  uv_tcp_init(loop, &cli->tcp);
   // set up timeout to avoid stuck;
   int32_t fd = taosCreateSocketWithTimeout(5);
+  int     ret = uv_tcp_open((uv_tcp_t*)&cli->tcp, fd);
+  if (ret != 0) {
+    tError("http-report failed to open socket, reason:%s, dst:%s:%d", uv_strerror(ret), cli->addr, cli->port);
+    destroyHttpClient(cli);
+    return;
+  }
 
-  int ret = uv_tcp_open((uv_tcp_t*)&cli->tcp, fd);
+  ret = uv_tcp_connect(&cli->conn, &cli->tcp, (const struct sockaddr*)&cli->dest, clientConnCb);
   if (ret != 0) {
-    uError("http-report failed to open socket, reason:%s, dst:%s:%d", uv_strerror(ret), cli->addr, cli->port);
+    tError("http-report failed to connect to http-server, reason:%s, dst:%s:%d", uv_strerror(ret), cli->addr,
+           cli->port);
     destroyHttpClient(cli);
-    uv_stop(loop);
-    terrno = TAOS_SYSTEM_ERROR(ret);
-  } else {
-    ret = uv_tcp_connect(&cli->conn, &cli->tcp, (const struct sockaddr*)&dest, clientConnCb);
-    if (ret != 0) {
-      uError("http-report failed to connect to http-server, reason:%s, dst:%s:%d", uv_strerror(ret), cli->addr,
-             cli->port);
-      destroyHttpClient(cli);
-      uv_stop(loop);
-      terrno = TAOS_SYSTEM_ERROR(ret);
-    }
   }
+  return;
+
+END:
+  tError("http-report failed to report, reason: %s, addr: %s:%d", terrstr(), msg->server, msg->port);
+  httpDestroyMsg(msg);
+}
+
+int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag) {
+  taosThreadOnce(&transHttpInit, transHttpEnvInit);
+  return taosSendHttpReportImpl(server, port, pCont, contLen, flag);
+}
+
+static void transHttpEnvInit() {
+  SHttpModule* http = taosMemoryMalloc(sizeof(SHttpModule));
+
+  http->loop = taosMemoryMalloc(sizeof(uv_loop_t));
+  uv_loop_init(http->loop);
+
+  http->asyncPool = transAsyncPoolCreate(http->loop, 1, http, httpAsyncCb);
+
+  int err = taosThreadCreate(&http->thread, NULL, httpThread, (void*)http);
+  if (err != 0) {
+    taosMemoryFree(http->loop);
+    taosMemoryFree(http);
+    http = NULL;
+  }
+  atomic_store_ptr(&thttp, http);
+}
+
+void transHttpEnvDestroy() {
+  SHttpModule* load = atomic_load_ptr(&thttp);
+  if (load == NULL) {
+    return;
+  }
+  httpSendQuit();
+  taosThreadJoin(load->thread, NULL);
+
+  TRANS_DESTROY_ASYNC_POOL_MSG(load->asyncPool, SHttpMsg, httpDestroyMsg);
+  transAsyncPoolDestroy(load->asyncPool);
+  uv_loop_close(load->loop);
+  taosMemoryFree(load->loop);
+  taosMemoryFree(load);
 
-  uv_run(loop, UV_RUN_DEFAULT);
-  uv_loop_close(loop);
-  taosMemoryFree(loop);
-  return terrno;
+  atomic_store_ptr(&thttp, NULL);
 }

source/libs/transport/src/trans.c

@@ -172,6 +172,8 @@ int32_t rpcInit() {
 }
 void rpcCleanup(void) {
   transCleanup();
+  transHttpEnvDestroy();
+
   return;
 }
 

source/libs/transport/src/transCli.c

@@ -187,18 +187,8 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
     snprintf(key, sizeof(key), "%s:%d", ip, (int)port); \
   } while (0)
 
-#define CONN_HOST_THREAD_IDX1(idx, exh, refId, pThrd) \
-  do {                                                \
-    if (exh == NULL) {                                \
-      idx = -1;                                       \
-    } else {                                          \
-      ASYNC_CHECK_HANDLE((exh), refId);               \
-      pThrd = (SCliThrd*)(exh)->pThrd;                \
-    }                                                 \
-  } while (0)
-#define CONN_PERSIST_TIME(para)    ((para) <= 90000 ? 90000 : (para))
-#define CONN_GET_HOST_THREAD(conn) (conn ? ((SCliConn*)conn)->hostThrd : NULL)
-#define CONN_GET_INST_LABEL(conn)  (((STrans*)(((SCliThrd*)(conn)->hostThrd)->pTransInst))->label)
+#define CONN_PERSIST_TIME(para)   ((para) <= 90000 ? 90000 : (para))
+#define CONN_GET_INST_LABEL(conn) (((STrans*)(((SCliThrd*)(conn)->hostThrd)->pTransInst))->label)
 
 #define CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle)                         \
   do {                                                                    \
@@ -217,6 +207,7 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
       tDebug("msg found, %" PRIu64 "", ahandle);                          \
     }                                                                     \
   } while (0)
+
 #define CONN_GET_NEXT_SENDMSG(conn)                 \
   do {                                              \
     int i = 0;                                      \
@@ -231,21 +222,6 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
     }                                               \
   } while (0)
 
-#define CONN_HANDLE_THREAD_QUIT(thrd) \
-  do {                                \
-    if (thrd->quit) {                 \
-      return;                         \
-    }                                 \
-  } while (0)
-
-#define CONN_HANDLE_BROKEN(conn) \
-  do {                           \
-    if (conn->broken) {          \
-      cliHandleExcept(conn);     \
-      return;                    \
-    }                            \
-  } while (0)
-
 #define CONN_SET_PERSIST_BY_APP(conn) \
   do {                                \
     if (conn->status == ConnNormal) { \

source/util/src/tref.c

@@ -57,7 +57,7 @@ static void    taosIncRsetCount(SRefSet *pSet);
 static void    taosDecRsetCount(SRefSet *pSet);
 static int32_t taosDecRefCount(int32_t rsetId, int64_t rid, int32_t remove);
 
-int32_t taosOpenRef(int32_t max, void (*fp)(void *)) {
+int32_t taosOpenRef(int32_t max, RefFp fp) {
   SRefNode **nodeList;
   SRefSet   *pSet;
   int64_t   *lockedBy;

tests/develop-test/5-taos-tools/taosbenchmark/auto_create_table_json.py

@@ -25,7 +25,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor())
 

tests/develop-test/5-taos-tools/taosbenchmark/commandline.py

@@ -27,7 +27,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
 

tests/develop-test/5-taos-tools/taosbenchmark/custom_col_tag.py

@@ -24,7 +24,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
 

tests/develop-test/5-taos-tools/taosbenchmark/default_json.py

@@ -23,7 +23,7 @@ class TDTestCase:
         [TD-11510] taosBenchmark test cases
         """
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
 

tests/develop-test/5-taos-tools/taosbenchmark/demo.py

@@ -27,7 +27,7 @@ class TDTestCase:
         """
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
 

tests/develop-test/5-taos-tools/taosbenchmark/insert_alltypes_json.py

@@ -24,7 +24,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
 

tests/develop-test/5-taos-tools/taosbenchmark/invalid_commandline.py

@@ -24,7 +24,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
 

tests/develop-test/5-taos-tools/taosbenchmark/json_tag.py

@@ -24,7 +24,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
 

tests/develop-test/5-taos-tools/taosbenchmark/limit_offset_json.py

@@ -24,7 +24,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
 

tests/develop-test/5-taos-tools/taosbenchmark/query_json.py

@@ -29,7 +29,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
 

tests/develop-test/5-taos-tools/taosbenchmark/sample_csv_json.py

@@ -24,7 +24,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
 

tests/develop-test/5-taos-tools/taosbenchmark/sml_interlace.py

@@ -24,7 +24,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
 

tests/develop-test/5-taos-tools/taosbenchmark/sml_json_alltypes.py

@@ -24,7 +24,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
 

tests/develop-test/5-taos-tools/taosbenchmark/sml_telnet_alltypes.py

@@ -24,7 +24,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
 

tests/develop-test/5-taos-tools/taosbenchmark/taosadapter_json.py

@@ -25,7 +25,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
 

tests/develop-test/5-taos-tools/taosbenchmark/telnet_tcp.py

@@ -25,7 +25,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
 

tests/develop-test/5-taos-tools/taosdump/taosdumpTestInspect.py

@@ -27,7 +27,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
         self.tmpdir = "tmp"

tests/develop-test/5-taos-tools/taosdump/taosdumpTestTypeBigInt.py

@@ -27,7 +27,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
         self.tmpdir = "tmp"

tests/develop-test/5-taos-tools/taosdump/taosdumpTestTypeBinary.py

@@ -27,7 +27,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
         self.tmpdir = "tmp"

tests/develop-test/5-taos-tools/taosdump/taosdumpTestTypeBool.py

@@ -27,7 +27,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
         self.tmpdir = "tmp"

tests/develop-test/5-taos-tools/taosdump/taosdumpTestTypeDouble.py

@@ -28,7 +28,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
         self.tmpdir = "tmp"

tests/develop-test/5-taos-tools/taosdump/taosdumpTestTypeFloat.py

@@ -28,7 +28,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
         self.tmpdir = "tmp"

tests/develop-test/5-taos-tools/taosdump/taosdumpTestTypeInt.py

@@ -27,7 +27,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
         self.tmpdir = "tmp"

tests/develop-test/5-taos-tools/taosdump/taosdumpTestTypeJson.py

@@ -27,7 +27,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
         self.tmpdir = "tmp"

tests/develop-test/5-taos-tools/taosdump/taosdumpTestTypeSmallInt.py

@@ -27,7 +27,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
         self.tmpdir = "tmp"

tests/develop-test/5-taos-tools/taosdump/taosdumpTestTypeTinyInt.py

@@ -27,7 +27,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
         self.tmpdir = "tmp"

tests/develop-test/5-taos-tools/taosdump/taosdumpTestTypeUnsignedBigInt.py

@@ -27,7 +27,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
         self.tmpdir = "tmp"

tests/develop-test/5-taos-tools/taosdump/taosdumpTestTypeUnsignedInt.py

@@ -27,7 +27,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
         self.tmpdir = "tmp"

tests/develop-test/5-taos-tools/taosdump/taosdumpTestTypeUnsignedSmallInt.py

@@ -27,7 +27,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
         self.tmpdir = "tmp"

tests/develop-test/5-taos-tools/taosdump/taosdumpTestTypeUnsignedTinyInt.py

@@ -27,7 +27,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVarl=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(), logSql)
         self.tmpdir = "tmp"

tests/develop-test/test.py

@@ -72,8 +72,9 @@ if __name__ == "__main__":
     queryPolicy = 1
     createDnodeNums = 1
     restful = False
-    opts, args = getopt.gnu_getopt(sys.argv[1:], 'f:p:m:l:scghrd:k:e:N:M:Q:C:RD:', [
-        'file=', 'path=', 'master', 'logSql', 'stop', 'cluster', 'valgrind', 'help', 'restart', 'updateCfgDict', 'killv', 'execCmd','dnodeNums','mnodeNums','queryPolicy','createDnodeNums','restful','adaptercfgupdate'])
+    replicaVar = 1
+    opts, args = getopt.gnu_getopt(sys.argv[1:], 'f:p:m:l:scghrd:k:e:N:M:Q:C:RD:n:', [
+        'file=', 'path=', 'master', 'logSql', 'stop', 'cluster', 'valgrind', 'help', 'restart', 'updateCfgDict', 'killv', 'execCmd','dnodeNums','mnodeNums','queryPolicy','createDnodeNums','restful','adaptercfgupdate','replicaVar'])
     for key, value in opts:
         if key in ['-h', '--help']:
             tdLog.printNoPrefix(
@@ -95,8 +96,7 @@ if __name__ == "__main__":
             tdLog.printNoPrefix('-C create Dnode Numbers in one cluster')
             tdLog.printNoPrefix('-R restful realization form')
             tdLog.printNoPrefix('-D taosadapter update cfg dict ')
-
-
+            tdLog.printNoPrefix('-n the number of replicas')
             sys.exit(0)
 
         if key in ['-r', '--restart']:
@@ -168,6 +168,9 @@ if __name__ == "__main__":
                 print('adapter cfg update convert fail.')
                 sys.exit(0)
 
+        if key in ['-n', '--replicaVar']:
+            replicaVar = value
+
     if not execCmd == "":
         if restful:
             tAdapter.init(deployPath)
@@ -194,7 +197,7 @@ if __name__ == "__main__":
             processID = subprocess.check_output(psCmd, shell=True)
 
         for port in range(6030, 6041):
-            usePortPID = "lsof -i tcp:%d | grep LISTEn | awk '{print $2}'" % port
+            usePortPID = "lsof -i tcp:%d | grep LISTEN | awk '{print $2}'" % port
             processID = subprocess.check_output(usePortPID, shell=True)
 
             if processID:
@@ -206,11 +209,13 @@ if __name__ == "__main__":
             time.sleep(2)
 
         if restful:
-            toBeKilled = "taosadapter"
+            toBeKilled = "taosadapt"
 
-            killCmd = "ps -ef|grep -w %s| grep -v grep | awk '{print $2}' | xargs kill -TERM > /dev/null 2>&1" % toBeKilled
+            # killCmd = "ps -ef|grep -w %s| grep -v grep | awk '{print $2}' | xargs kill -TERM > /dev/null 2>&1" % toBeKilled
+            killCmd = f"pkill {toBeKilled}"
 
             psCmd = "ps -ef|grep -w %s| grep -v grep | awk '{print $2}'" % toBeKilled
+            # psCmd = f"pgrep {toBeKilled}"
             processID = subprocess.check_output(psCmd, shell=True)
 
             while(processID):
@@ -218,15 +223,15 @@ if __name__ == "__main__":
                 time.sleep(1)
                 processID = subprocess.check_output(psCmd, shell=True)
 
-            for port in range(6030, 6041):
-                usePortPID = "lsof -i tcp:%d | grep LISTEn | awk '{print $2}'" % port
-                processID = subprocess.check_output(usePortPID, shell=True)
+            port = 6041
+            usePortPID = f"lsof -i tcp:{port} | grep LISTEN | awk '{{print $2}}'"
+            processID = subprocess.check_output(usePortPID, shell=True)
 
-                if processID:
-                    killCmd = "kill -TERM %s" % processID
-                    os.system(killCmd)
-                fuserCmd = "fuser -k -n tcp %d" % port
-                os.system(fuserCmd)
+            if processID:
+                killCmd = f"kill -TERM {processID}"
+                os.system(killCmd)
+            fuserCmd = f"fuser -k -n tcp {port}"
+            os.system(fuserCmd)
 
             tdLog.info('stop taosadapter')
 
@@ -319,7 +324,7 @@ if __name__ == "__main__":
             for dnode in tdDnodes.dnodes:
                 tdDnodes.starttaosd(dnode.index)
             tdCases.logSql(logSql)
-
+                            
             if restful:
                 tAdapter.deploy(adapter_cfg_dict)
                 tAdapter.start()
@@ -339,6 +344,26 @@ if __name__ == "__main__":
                     print("check dnode ready")
             except Exception as r:
                 print(r)
+            if queryPolicy != 1:
+                queryPolicy=int(queryPolicy)
+                if restful:
+                    conn = taosrest.connect(url=f"http://{host}:6041")
+                else:
+                    conn = taos.connect(host,config=tdDnodes.getSimCfgPath())
+
+                cursor = conn.cursor()
+                cursor.execute("create qnode on dnode 1")
+                cursor.execute(f'alter local "queryPolicy" "{queryPolicy}"')
+                cursor.execute("show local variables")
+                res = cursor.fetchall()
+                for i in range(cursor.rowcount):
+                    if res[i][0] == "queryPolicy" :
+                        if int(res[i][1]) == int(queryPolicy):
+                            tdLog.success(f'alter queryPolicy to {queryPolicy} successfully')
+                        else:
+                            tdLog.debug(res)
+                            tdLog.exit(f"alter queryPolicy to  {queryPolicy} failed")
+                            
         if ucase is not None and hasattr(ucase, 'noConn') and ucase.noConn == True:
             conn = None
         else:
@@ -453,6 +478,26 @@ if __name__ == "__main__":
             except Exception as r:
                 print(r)
 
+            if queryPolicy != 1:
+                queryPolicy=int(queryPolicy)
+                if restful:
+                    conn = taosrest.connect(url=f"http://{host}:6041")
+                else:
+                    conn = taos.connect(host,config=tdDnodes.getSimCfgPath())
+
+                cursor = conn.cursor()
+                cursor.execute("create qnode on dnode 1")
+                cursor.execute(f'alter local "queryPolicy" "{queryPolicy}"')
+                cursor.execute("show local variables")
+                res = cursor.fetchall()
+                for i in range(cursor.rowcount):
+                    if res[i][0] == "queryPolicy" :
+                        if int(res[i][1]) == int(queryPolicy):
+                            tdLog.success(f'alter queryPolicy to {queryPolicy} successfully')
+                        else:
+                            tdLog.debug(res)
+                            tdLog.exit(f"alter queryPolicy to  {queryPolicy} failed")
+                            
 
         if testCluster:
             tdLog.info("Procedures for testing cluster")
@@ -470,7 +515,7 @@ if __name__ == "__main__":
             if fileName == "all":
                 tdCases.runAllLinux(conn)
             else:
-                tdCases.runOneLinux(conn, fileName)
+                tdCases.runOneLinux(conn, fileName, replicaVar)
 
         if restart:
             if fileName == "all":
@@ -487,7 +532,7 @@ if __name__ == "__main__":
                         conn = taosrest.connect(url=f"http://{host}:6041")
                     tdLog.info("Procedures for tdengine deployed in %s" % (host))
                     tdLog.info("query test after taosd restart")
-                    tdCases.runOneLinux(conn, sp[0] + "_" + "restart.py")
+                    tdCases.runOneLinux(conn, sp[0] + "_" + "restart.py", replicaVar)
                 else:
                     tdLog.info("not need to query")
 

tests/pytest/util/cases.py

@@ -63,14 +63,14 @@ class TDCases:
 
         tdLog.info("total %d Linux test case(s) executed" % (runNum))
 
-    def runOneLinux(self, conn, fileName):
+    def runOneLinux(self, conn, fileName, replicaVar):
         testModule = self.__dynamicLoadModule(fileName)
 
         runNum = 0
         for tmp in self.linuxCases:
             if tmp.name.find(fileName) != -1:
                 case = testModule.TDTestCase()
-                case.init(conn, self._logSql)
+                case.init(conn, self._logSql, replicaVar)
                 try:
                     case.run()
                 except Exception as e:

tests/script/tsim/stream/scalar.sim

+$loop_all = 0
+looptest:
+
+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/exec.sh -n dnode1 -s start
+sleep 50
+sql connect
+
+sql drop database if exists test;
+sql create database test  vgroups 1;
+sql use test;
+sql create stable st(ts timestamp,a int,b int,c int, d double) tags(ta int,tb int,tc int);
+sql create table t1 using st tags(1,1,1);
+sql create table t2 using st tags(2,2,2);
+sql create stream streams0 into streamt0 as select  ts c1, a, abs(b) c4 from t1 partition by a;
+sql create stream streams1 into streamt1 as select  ts c1, a, abs(b) c4 from t1;
+sql create stream streams2 into streamt2 as select  ts c1, a, abs(b) c4 from st partition by tbname;
+
+sql insert into t1 values(1648791213000,1,1,1,1);
+sql insert into t1 values(1648791213001,1,1,1,1);
+sql insert into t1 values(1648791213002,1,1,1,1);
+
+sql insert into t2 values(1648791213000,1,2,2,2);
+sql insert into t2 values(1648791213001,1,1,1,1);
+sql insert into t2 values(1648791213002,1,1,1,1);
+
+sql insert into t1 values(1648791213001,2,11,11,11);
+
+
+$loop_count = 0
+loop1:
+
+sleep 200
+
+sql select * from streamt0 order by a desc;
+
+$loop_count = $loop_count + 1
+
+if $loop_count == 10 then
+  return -1
+endi
+
+if $rows != 3 then
+  print ======streamt0=rows=$rows
+  goto loop1
+endi
+
+if $data01 != 2 then
+  print ======streamt0=data01=$data01
+  goto loop1
+endi
+
+if $data02 != 11 then
+  print ======streamt0=data02=$data02
+  goto loop1
+endi
+
+
+sql select * from streamt1 order by a desc;
+
+if $rows != 3 then
+  print ======streamt1=rows=$rows
+  goto loop1
+endi
+
+if $data01 != 2 then
+  print ======streamt1=data01=$data01
+  goto loop1
+endi
+
+if $data02 != 11 then
+  print ======streamt1=data02=$data02
+  goto loop1
+endi
+
+sql select * from streamt2 order by a desc;
+
+if $rows != 6 then
+  print ======streamt2=rows=$rows
+  goto loop1
+endi
+
+if $data01 != 2 then
+  print ======streamt2=data01=$data01
+  goto loop1
+endi
+
+if $data02 != 11 then
+  print ======streamt2=data02=$data02
+  goto loop1
+endi
+
+system sh/stop_dnodes.sh
\ No newline at end of file

tests/system-test/0-others/cachemodel.py

@@ -14,7 +14,7 @@ class TDTestCase:
     updatecfgDict = {'debugFlag': 143 ,"cDebugFlag":143,"uDebugFlag":143 ,"rpcDebugFlag":143 , "tmrDebugFlag":143 ,
     "jniDebugFlag":143 ,"simDebugFlag":143,"dDebugFlag":143, "dDebugFlag":143,"vDebugFlag":143,"mDebugFlag":143,"qDebugFlag":143,
     "wDebugFlag":143,"sDebugFlag":143,"tsdbDebugFlag":143,"tqDebugFlag":143 ,"fsDebugFlag":143 ,"udfDebugFlag":143}
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor(), True)
 

tests/system-test/0-others/compatibility.py

@@ -22,7 +22,7 @@ class TDTestCase:
         '''
         return
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor())
 

tests/system-test/0-others/fsync.py

@@ -26,7 +26,7 @@ TS_TYPE_COL = [ TS_COL, ]
 
 class TDTestCase:
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor())
 

tests/system-test/0-others/show.py

@@ -20,7 +20,7 @@ from util.sqlset import *
 
 
 class TDTestCase:
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor())
         self.setsql = TDSetSql()

tests/system-test/0-others/sysinfo.py

@@ -20,7 +20,7 @@ from util.common import *
 
 
 class TDTestCase:
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor())
         self.dbname = 'db'

tests/system-test/0-others/taosShell.py

@@ -108,7 +108,7 @@ class TDTestCase:
 
     print ("===================: ", updatecfgDict)
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor())
 

tests/system-test/0-others/taosShellError.py

@@ -110,7 +110,7 @@ class TDTestCase:
 
     print ("===================: ", updatecfgDict)
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor())
 

tests/system-test/0-others/taosShellNetChk.py

@@ -110,7 +110,7 @@ class TDTestCase:
 
     print ("===================: ", updatecfgDict)
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor())
 

tests/system-test/0-others/taosdMonitor.py

@@ -284,7 +284,7 @@ class TDTestCase:
 
     print ("===================: ", updatecfgDict)
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor())
 

tests/system-test/0-others/taosdShell.py

@@ -44,7 +44,7 @@ class TDTestCase:
 
     # print ("===================: ", updatecfgDict)
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor())
 

tests/system-test/0-others/taosdlog.py

@@ -10,7 +10,7 @@ from util.dnodes import *
 
 class TDTestCase:
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor())
 

tests/system-test/0-others/telemetry.py

@@ -173,7 +173,7 @@ class TDTestCase:
 
     print ("===================: ", updatecfgDict)
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor())
 

tests/system-test/0-others/udfTest.py

@@ -13,7 +13,7 @@ import subprocess
 
 class TDTestCase:
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor(), logSql)
 

tests/system-test/0-others/udf_cfg1.py

@@ -15,7 +15,7 @@ class TDTestCase:
     updatecfgDict = {'debugFlag': 143, "cDebugFlag": 143, "uDebugFlag": 143, "rpcDebugFlag": 143, "tmrDebugFlag": 143,
                      "jniDebugFlag": 143, "simDebugFlag": 143, "dDebugFlag": 143, "dDebugFlag": 143, "vDebugFlag": 143, "mDebugFlag": 143, "qDebugFlag": 143,
                      "wDebugFlag": 143, "sDebugFlag": 143, "tsdbDebugFlag": 143, "tqDebugFlag": 143, "fsDebugFlag": 143, "fnDebugFlag": 143 ,"udf":0}
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor(), logSql)
 

tests/system-test/0-others/udf_cfg2.py

@@ -15,7 +15,7 @@ class TDTestCase:
     updatecfgDict = {'debugFlag': 143, "cDebugFlag": 143, "uDebugFlag": 143, "rpcDebugFlag": 143, "tmrDebugFlag": 143,
                      "jniDebugFlag": 143, "simDebugFlag": 143, "dDebugFlag": 143, "dDebugFlag": 143, "vDebugFlag": 143, "mDebugFlag": 143, "qDebugFlag": 143,
                      "wDebugFlag": 143, "sDebugFlag": 143, "tsdbDebugFlag": 143, "tqDebugFlag": 143, "fsDebugFlag": 143, "fnDebugFlag": 143 ,"udf":1}
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor(), logSql)
 

tests/system-test/0-others/udf_cluster.py

@@ -19,7 +19,7 @@ class MyDnodes(TDDnodes):
 
 class TDTestCase:
 
-    def init(self,conn ,logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         self.TDDnodes = None
         self.depoly_cluster(3)

tests/system-test/0-others/udf_create.py

@@ -15,7 +15,7 @@ import threading
 
 class TDTestCase:
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor(), logSql)
 

tests/system-test/0-others/udf_restart_taosd.py

@@ -12,7 +12,7 @@ import subprocess
 
 class TDTestCase:
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor(), logSql)
 

tests/system-test/0-others/user_control.py

@@ -154,7 +154,7 @@ class User:
 
 class TDTestCase:
 
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor())
 

tests/system-test/1-insert/alter_database.py

@@ -11,7 +11,7 @@ from util.cases import *
 from util.dnodes import *
 
 class TDTestCase:
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor(),logSql)
         self.buffer_boundary = [3,4097,8193,12289,16384]

tests/system-test/1-insert/alter_stable.py

@@ -20,7 +20,7 @@ from util.sqlset import *
 from util import constant
 from util.common import *
 class TDTestCase:
-    def init(self, conn, logSql):
+    def init(self, conn, logSql, replicaVar=1):
         tdLog.debug("start to execute %s" % __file__)
         tdSql.init(conn.cursor())
         self.setsql = TDSetSql()