Merge branch '3.0' into hotfix/resetquerycache

include/common/tmsg.h

@@ -510,7 +510,8 @@ typedef struct {
   int8_t   superUser;
   int8_t   connType;
   SEpSet   epSet;
-  char     sVersion[128];
+  char     sVer[TSDB_VERSION_LEN];
+  char     sDetailVer[128];
 } SConnectRsp;
 
 int32_t tSerializeSConnectRsp(void* buf, int32_t bufLen, SConnectRsp* pRsp);
@@ -836,6 +837,20 @@ typedef struct {
 int32_t tSerializeSDnodeListReq(void* buf, int32_t bufLen, SDnodeListReq* pReq);
 int32_t tDeserializeSDnodeListReq(void* buf, int32_t bufLen, SDnodeListReq* pReq);
 
+typedef struct {
+  int32_t useless;  // useless
+} SServerVerReq;
+
+int32_t tSerializeSServerVerReq(void* buf, int32_t bufLen, SServerVerReq* pReq);
+int32_t tDeserializeSServerVerReq(void* buf, int32_t bufLen, SServerVerReq* pReq);
+
+typedef struct {
+  char ver[TSDB_VERSION_LEN];
+} SServerVerRsp;
+
+int32_t tSerializeSServerVerRsp(void* buf, int32_t bufLen, SServerVerRsp* pRsp);
+int32_t tDeserializeSServerVerRsp(void* buf, int32_t bufLen, SServerVerRsp* pRsp);
+
 
 typedef struct SQueryNodeAddr {
   int32_t nodeId;  // vgId or qnodeId
@@ -1229,6 +1244,21 @@ typedef struct {
 int32_t tSerializeSShowVariablesReq(void* buf, int32_t bufLen, SShowVariablesReq* pReq);
 int32_t tDeserializeSShowVariablesReq(void* buf, int32_t bufLen, SShowVariablesReq* pReq);
 
+typedef struct {
+  char  name[TSDB_CONFIG_OPTION_LEN + 1];
+  char  value[TSDB_CONFIG_VALUE_LEN + 1];
+} SVariablesInfo;
+
+typedef struct {
+  SArray *variables;  //SArray<SVariablesInfo>
+} SShowVariablesRsp;
+
+int32_t tSerializeSShowVariablesRsp(void* buf, int32_t bufLen, SShowVariablesRsp* pReq);
+int32_t tDeserializeSShowVariablesRsp(void* buf, int32_t bufLen, SShowVariablesRsp* pReq);
+
+void tFreeSShowVariablesRsp(SShowVariablesRsp* pRsp);
+
+
 /*
  * sql: show tables like '%a_%'
  * payload is the query condition, e.g., '%a_%'

include/common/tmsgdef.h

@@ -163,6 +163,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_MND_REDISTRIBUTE_VGROUP, "redistribute-vgroup", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_SPLIT_VGROUP, "split-vgroup", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_SHOW_VARIABLES, "show-variables", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_SERVER_VERSION, "server-version", NULL, NULL)
 
   TD_NEW_MSG_SEG(TDMT_VND_MSG)
   TD_DEF_MSG_TYPE(TDMT_VND_SUBMIT, "submit", SSubmitReq, SSubmitRsp)

include/libs/catalog/catalog.h

@@ -71,6 +71,7 @@ typedef struct SCatalogReq {
   SArray* pTableCfg;      // element is SNAME
   bool    qNodeRequired;  // valid qnode
   bool    dNodeRequired;  // valid dnode
+  bool    svrVerRequired;
   bool    forceUpdate;
 } SCatalogReq;
 
@@ -80,18 +81,19 @@ typedef struct SMetaRes {
 } SMetaRes;
 
 typedef struct SMetaData {
-  SArray* pDbVgroup;    // pRes = SArray<SVgroupInfo>*
-  SArray* pDbCfg;       // pRes = SDbCfgInfo*
-  SArray* pDbInfo;      // pRes = SDbInfo*
-  SArray* pTableMeta;   // pRes = STableMeta*
-  SArray* pTableHash;   // pRes = SVgroupInfo*
-  SArray* pTableIndex;  // pRes = SArray<STableIndexInfo>*
-  SArray* pUdfList;     // pRes = SFuncInfo*
-  SArray* pIndex;       // pRes = SIndexInfo*
-  SArray* pUser;        // pRes = bool*
-  SArray* pQnodeList;   // pRes = SArray<SQueryNodeLoad>*
-  SArray* pTableCfg;    // pRes = STableCfg*
-  SArray* pDnodeList;   // pRes = SArray<SEpSet>*
+  SArray*   pDbVgroup;    // pRes = SArray<SVgroupInfo>*
+  SArray*   pDbCfg;       // pRes = SDbCfgInfo*
+  SArray*   pDbInfo;      // pRes = SDbInfo*
+  SArray*   pTableMeta;   // pRes = STableMeta*
+  SArray*   pTableHash;   // pRes = SVgroupInfo*
+  SArray*   pTableIndex;  // pRes = SArray<STableIndexInfo>*
+  SArray*   pUdfList;     // pRes = SFuncInfo*
+  SArray*   pIndex;       // pRes = SIndexInfo*
+  SArray*   pUser;        // pRes = bool*
+  SArray*   pQnodeList;   // pRes = SArray<SQueryNodeLoad>*
+  SArray*   pTableCfg;    // pRes = STableCfg*
+  SArray*   pDnodeList;   // pRes = SArray<SEpSet>*
+  SMetaRes* pSvrVer;      // pRes = char*
 } SMetaData;
 
 typedef struct SCatalogCfg {
@@ -260,7 +262,7 @@ int32_t catalogGetTableHashVgroup(SCatalog* pCatalog, SRequestConnInfo* pConn, c
  */
 int32_t catalogGetAllMeta(SCatalog* pCatalog, SRequestConnInfo* pConn, const SCatalogReq* pReq, SMetaData* pRsp);
 
-int32_t catalogAsyncGetAllMeta(SCatalog* pCtg, SRequestConnInfo* pConn, uint64_t reqId, const SCatalogReq* pReq, catalogCallback fp, void* param, int64_t* jobId);
+int32_t catalogAsyncGetAllMeta(SCatalog* pCtg, SRequestConnInfo* pConn, const SCatalogReq* pReq, catalogCallback fp, void* param, int64_t* jobId);
 
 int32_t catalogGetQnodeList(SCatalog* pCatalog, SRequestConnInfo* pConn, SArray* pQnodeList);
 
@@ -290,6 +292,8 @@ int32_t catalogUpdateUserAuthInfo(SCatalog* pCtg, SGetUserAuthRsp* pAuth);
 
 int32_t catalogUpdateVgEpSet(SCatalog* pCtg, const char* dbFName, int32_t vgId, SEpSet *epSet);
 
+int32_t catalogGetServerVersion(SCatalog* pCtg, SRequestConnInfo *pConn, char** pVersion);
+
 int32_t ctgdLaunchAsyncCall(SCatalog* pCtg, SRequestConnInfo* pConn, uint64_t reqId, bool forceUpdate);
 
 int32_t catalogClearCache(void);

include/libs/parser/parser.h

@@ -51,6 +51,8 @@ typedef struct SParseContext {
   bool             isSuperUser;
   bool             async;
   int8_t           schemalessType;
+  const char*      svrVer;
+  bool             nodeOffline;
 } SParseContext;
 
 int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery);

source/client/inc/clientInt.h

@@ -54,6 +54,11 @@ enum {
   RES_TYPE__TMQ_META,
 };
 
+#define SHOW_VARIABLES_RESULT_COLS      2
+#define SHOW_VARIABLES_RESULT_FIELD1_LEN (TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE)
+#define SHOW_VARIABLES_RESULT_FIELD2_LEN (TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE)
+
+
 #define TD_RES_QUERY(res)    (*(int8_t*)res == RES_TYPE__QUERY)
 #define TD_RES_TMQ(res)      (*(int8_t*)res == RES_TYPE__TMQ)
 #define TD_RES_TMQ_META(res) (*(int8_t*)res == RES_TYPE__TMQ_META)
@@ -104,6 +109,8 @@ typedef struct SHeartBeatInfo {
 struct SAppInstInfo {
   int64_t            numOfConns;
   SCorEpSet          mgmtEp;
+  int32_t            totalDnodes;
+  int32_t            onlineDnodes;
   TdThreadMutex      qnodeMutex;
   SArray*            pQnodeList;
   SAppClusterSummary summary;
@@ -127,7 +134,8 @@ typedef struct STscObj {
   char          user[TSDB_USER_LEN];
   char          pass[TSDB_PASSWORD_LEN];
   char          db[TSDB_DB_FNAME_LEN];
-  char          ver[128];
+  char          sVer[TSDB_VERSION_LEN];
+  char          sDetailVer[128];
   int8_t        connType;
   int32_t       acctId;
   uint32_t      connId;

source/client/src/clientHb.c

@@ -161,6 +161,9 @@ static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
             
         updateEpSet_s(&pTscObj->pAppInfo->mgmtEp, &pRsp->query->epSet);
       }
+      
+      pTscObj->pAppInfo->totalDnodes = pRsp->query->totalDnodes;
+      pTscObj->pAppInfo->onlineDnodes = pRsp->query->onlineDnodes;
       pTscObj->connId = pRsp->query->connId;
 
       if (pRsp->query->killRid) {

source/client/src/clientImpl.c

@@ -178,7 +178,9 @@ int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtC
                        .pStmtCb = pStmtCb,
                        .pUser = pTscObj->user,
                        .schemalessType = pTscObj->schemalessType,
-                       .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER))};
+                       .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
+                       .svrVer = pTscObj->sVer,
+                       .nodeOffline = (pTscObj->pAppInfo->onlineDnodes < pTscObj->pAppInfo->totalDnodes)};
 
   cxt.mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
   int32_t code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &cxt.pCatalog);

source/client/src/clientMain.c

@@ -623,7 +623,7 @@ const char *taos_get_server_info(TAOS *taos) {
 
   releaseTscObj(*(int64_t *)taos);
 
-  return pTscObj->ver;
+  return pTscObj->sDetailVer;
 }
 
 typedef struct SqlParseWrapper {
@@ -766,7 +766,7 @@ void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) {
                            .requestObjRefId = pCxt->requestRid,
                            .mgmtEps = pCxt->mgmtEpSet};
 
-  code = catalogAsyncGetAllMeta(pCxt->pCatalog, &conn, pRequest->requestId, &catalogReq, retrieveMetaCallback, pWrapper,
+  code = catalogAsyncGetAllMeta(pCxt->pCatalog, &conn, &catalogReq, retrieveMetaCallback, pWrapper,
                                 &pRequest->body.queryJob);
   if (code == TSDB_CODE_SUCCESS) {
     return;
@@ -934,7 +934,7 @@ int taos_load_table_info(TAOS *taos, const char *tableNameList) {
 
   conn.mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
 
-  code = catalogAsyncGetAllMeta(pCtg, &conn, pRequest->requestId, &catalogReq, syncCatalogFn, &param, NULL);
+  code = catalogAsyncGetAllMeta(pCtg, &conn, &catalogReq, syncCatalogFn, &param, NULL);
   if (code) {
     goto _return;
   }

source/client/src/clientMsgHandler.c

@@ -82,7 +82,8 @@ int32_t processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
 
   pTscObj->connId = connectRsp.connId;
   pTscObj->acctId = connectRsp.acctId;
-  tstrncpy(pTscObj->ver, connectRsp.sVersion, tListLen(pTscObj->ver));
+  tstrncpy(pTscObj->sVer, connectRsp.sVer, tListLen(pTscObj->sVer));
+  tstrncpy(pTscObj->sDetailVer, connectRsp.sDetailVer, tListLen(pTscObj->sDetailVer));
 
   // update the appInstInfo
   pTscObj->pAppInfo->clusterId = connectRsp.clusterId;
@@ -287,6 +288,103 @@ int32_t processAlterStbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
   return code;
 }
 
+static int32_t buildShowVariablesBlock(SArray* pVars, SSDataBlock** block) {
+  SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
+  pBlock->info.hasVarCol = true;
+
+  pBlock->pDataBlock = taosArrayInit(SHOW_VARIABLES_RESULT_COLS, sizeof(SColumnInfoData));
+
+  SColumnInfoData infoData = {0};
+  infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
+  infoData.info.bytes = SHOW_VARIABLES_RESULT_FIELD1_LEN;
+
+  taosArrayPush(pBlock->pDataBlock, &infoData);
+
+  infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
+  infoData.info.bytes = SHOW_VARIABLES_RESULT_FIELD2_LEN;
+  taosArrayPush(pBlock->pDataBlock, &infoData);
+
+  int32_t numOfCfg = taosArrayGetSize(pVars);
+  blockDataEnsureCapacity(pBlock, numOfCfg);
+
+  for (int32_t i = 0, c = 0; i < numOfCfg; ++i, c = 0) {
+    SVariablesInfo *pInfo = taosArrayGet(pVars, i);
+
+    char name[TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE] = {0};
+    STR_WITH_MAXSIZE_TO_VARSTR(name, pInfo->name, TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE);
+    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, c++);
+    colDataAppend(pColInfo, i, name, false);
+    
+    char value[TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE] = {0};
+    STR_WITH_MAXSIZE_TO_VARSTR(value, pInfo->value, TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE);
+    pColInfo = taosArrayGet(pBlock->pDataBlock, c++);
+    colDataAppend(pColInfo, i, value, false);
+  }
+
+  pBlock->info.rows = numOfCfg;
+
+  *block = pBlock;
+  
+  return TSDB_CODE_SUCCESS;
+}
+
+
+static int32_t buildShowVariablesRsp(SArray* pVars, SRetrieveTableRsp** pRsp) {
+  SSDataBlock* pBlock = NULL;
+  int32_t code = buildShowVariablesBlock(pVars, &pBlock);
+  if (code) {
+    return code;
+  }
+
+  size_t rspSize = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
+  *pRsp = taosMemoryCalloc(1, rspSize);
+  if (NULL == *pRsp) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  (*pRsp)->useconds = 0;
+  (*pRsp)->completed = 1;
+  (*pRsp)->precision = 0;
+  (*pRsp)->compressed = 0;
+  (*pRsp)->compLen = 0;
+  (*pRsp)->numOfRows = htonl(pBlock->info.rows);
+  (*pRsp)->numOfCols = htonl(SHOW_VARIABLES_RESULT_COLS);
+
+  int32_t len = 0;
+  blockCompressEncode(pBlock, (*pRsp)->data, &len, SHOW_VARIABLES_RESULT_COLS, false);
+  ASSERT(len == rspSize - sizeof(SRetrieveTableRsp));
+
+  blockDataDestroy(pBlock);
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t processShowVariablesRsp(void* param, const SDataBuf* pMsg, int32_t code) {
+  SRequestObj* pRequest = param;
+  if (code != TSDB_CODE_SUCCESS) {
+    setErrno(pRequest, code);
+  } else {
+    SShowVariablesRsp rsp = {0};
+    SRetrieveTableRsp* pRes = NULL;
+    code = tDeserializeSShowVariablesRsp(pMsg->pData, pMsg->len, &rsp);
+    if (TSDB_CODE_SUCCESS == code) {
+      code = buildShowVariablesRsp(rsp.variables, &pRes);
+    }
+    if (TSDB_CODE_SUCCESS == code) {
+      code = setQueryResultFromRsp(&pRequest->body.resInfo, pRes, false, false);
+    }
+    
+    tFreeSShowVariablesRsp(&rsp);
+  }
+
+  if (pRequest->body.queryFp != NULL) {
+    pRequest->body.queryFp(pRequest->body.param, pRequest, code);
+  } else {
+    tsem_post(&pRequest->body.rspSem);
+  }
+  return code;
+}
+
+
 __async_send_cb_fn_t getMsgRspHandle(int32_t msgType) {
   switch (msgType) {
     case TDMT_MND_CONNECT:
@@ -301,6 +399,8 @@ __async_send_cb_fn_t getMsgRspHandle(int32_t msgType) {
       return processDropDbRsp;
     case TDMT_MND_ALTER_STB:
       return processAlterStbRsp;
+    case TDMT_MND_SHOW_VARIABLES:
+      return processShowVariablesRsp;
     default:
       return genericRspCallback;
   }

source/common/src/tmsg.c

@@ -2250,6 +2250,56 @@ int32_t tDeserializeSDnodeListReq(void *buf, int32_t bufLen, SDnodeListReq *pReq
   return 0;
 }
 
+int32_t tSerializeSServerVerReq(void *buf, int32_t bufLen, SServerVerReq *pReq) {
+  SEncoder encoder = {0};
+  tEncoderInit(&encoder, buf, bufLen);
+
+  if (tStartEncode(&encoder) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->useless) < 0) return -1;
+  tEndEncode(&encoder);
+
+  int32_t tlen = encoder.pos;
+  tEncoderClear(&encoder);
+  return tlen;
+}
+
+int32_t tDeserializeSServerVerReq(void *buf, int32_t bufLen, SServerVerReq *pReq) {
+  SDecoder decoder = {0};
+  tDecoderInit(&decoder, buf, bufLen);
+
+  if (tStartDecode(&decoder) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->useless) < 0) return -1;
+
+  tEndDecode(&decoder);
+  tDecoderClear(&decoder);
+  return 0;
+}
+
+int32_t tSerializeSServerVerRsp(void *buf, int32_t bufLen, SServerVerRsp *pRsp) {
+  SEncoder encoder = {0};
+  tEncoderInit(&encoder, buf, bufLen);
+
+  if (tStartEncode(&encoder) < 0) return -1;
+  if (tEncodeCStr(&encoder, pRsp->ver) < 0) return -1;
+  tEndEncode(&encoder);
+
+  int32_t tlen = encoder.pos;
+  tEncoderClear(&encoder);
+  return tlen;
+}
+
+int32_t tDeserializeSServerVerRsp(void *buf, int32_t bufLen, SServerVerRsp *pRsp) {
+  SDecoder decoder = {0};
+  tDecoderInit(&decoder, buf, bufLen);
+
+  if (tStartDecode(&decoder) < 0) return -1;
+  if (tDecodeCStrTo(&decoder, pRsp->ver) < 0) return -1;
+
+  tEndDecode(&decoder);
+  tDecoderClear(&decoder);
+  return 0;
+}
+
 
 int32_t tSerializeSQnodeListRsp(void *buf, int32_t bufLen, SQnodeListRsp *pRsp) {
   SEncoder encoder = {0};
@@ -2859,6 +2909,67 @@ int32_t tDeserializeSShowVariablesReq(void *buf, int32_t bufLen, SShowVariablesR
   return 0;
 }
 
+int32_t tEncodeSVariablesInfo(SEncoder* pEncoder, SVariablesInfo* pInfo) {
+  if (tEncodeCStr(pEncoder, pInfo->name) < 0) return -1;
+  if (tEncodeCStr(pEncoder, pInfo->value) < 0) return -1;
+  return 0;
+}
+
+int32_t tDecodeSVariablesInfo(SDecoder* pDecoder, SVariablesInfo* pInfo) {
+  if (tDecodeCStrTo(pDecoder, pInfo->name) < 0) return -1;
+  if (tDecodeCStrTo(pDecoder, pInfo->value) < 0) return -1;
+  return 0;
+}
+
+
+int32_t tSerializeSShowVariablesRsp(void* buf, int32_t bufLen, SShowVariablesRsp* pRsp) {
+  SEncoder encoder = {0};
+  tEncoderInit(&encoder, buf, bufLen);
+
+  if (tStartEncode(&encoder) < 0) return -1;
+  int32_t varNum = taosArrayGetSize(pRsp->variables);
+  if (tEncodeI32(&encoder, varNum) < 0) return -1;
+  for (int32_t i = 0; i < varNum; ++i) {
+    SVariablesInfo* pInfo = taosArrayGet(pRsp->variables, i);
+    if (tEncodeSVariablesInfo(&encoder, pInfo) < 0) return -1;
+  }
+  tEndEncode(&encoder);
+
+  int32_t tlen = encoder.pos;
+  tEncoderClear(&encoder);
+  return tlen;
+}
+
+int32_t tDeserializeSShowVariablesRsp(void* buf, int32_t bufLen, SShowVariablesRsp* pRsp) {
+  SDecoder decoder = {0};
+  tDecoderInit(&decoder, buf, bufLen);
+
+  if (tStartDecode(&decoder) < 0) return -1;
+  int32_t varNum = 0;
+  if (tDecodeI32(&decoder, &varNum) < 0) return -1;
+  if (varNum > 0) {
+    pRsp->variables = taosArrayInit(varNum, sizeof(SVariablesInfo));
+    if (NULL == pRsp->variables) return -1;
+    for (int32_t i = 0; i < varNum; ++i) {
+      SVariablesInfo info = {0};
+      if (tDecodeSVariablesInfo(&decoder, &info) < 0) return -1;
+      if (NULL == taosArrayPush(pRsp->variables, &info)) return -1;
+    }
+  }
+
+  tEndDecode(&decoder);
+  tDecoderClear(&decoder);
+  return 0;
+}
+
+void tFreeSShowVariablesRsp(SShowVariablesRsp* pRsp) {
+  if (NULL == pRsp) {
+    return;
+  }
+  
+  taosArrayDestroy(pRsp->variables);
+}
+
 int32_t tSerializeSShowReq(void *buf, int32_t bufLen, SShowReq *pReq) {
   SEncoder encoder = {0};
   tEncoderInit(&encoder, buf, bufLen);
@@ -3400,7 +3511,8 @@ int32_t tSerializeSConnectRsp(void *buf, int32_t bufLen, SConnectRsp *pRsp) {
   if (tEncodeI8(&encoder, pRsp->superUser) < 0) return -1;
   if (tEncodeI8(&encoder, pRsp->connType) < 0) return -1;
   if (tEncodeSEpSet(&encoder, &pRsp->epSet) < 0) return -1;
-  if (tEncodeCStr(&encoder, pRsp->sVersion) < 0) return -1;
+  if (tEncodeCStr(&encoder, pRsp->sVer) < 0) return -1;
+  if (tEncodeCStr(&encoder, pRsp->sDetailVer) < 0) return -1;
   tEndEncode(&encoder);
 
   int32_t tlen = encoder.pos;
@@ -3420,7 +3532,8 @@ int32_t tDeserializeSConnectRsp(void *buf, int32_t bufLen, SConnectRsp *pRsp) {
   if (tDecodeI8(&decoder, &pRsp->superUser) < 0) return -1;
   if (tDecodeI8(&decoder, &pRsp->connType) < 0) return -1;
   if (tDecodeSEpSet(&decoder, &pRsp->epSet) < 0) return -1;
-  if (tDecodeCStrTo(&decoder, pRsp->sVersion) < 0) return -1;
+  if (tDecodeCStrTo(&decoder, pRsp->sVer) < 0) return -1;
+  if (tDecodeCStrTo(&decoder, pRsp->sDetailVer) < 0) return -1;
   tEndDecode(&decoder);
 
   tDecoderClear(&decoder);

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

@@ -206,6 +206,8 @@ SArray *mmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_MND_SYSTABLE_RETRIEVE, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_GRANT, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_AUTH, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_SHOW_VARIABLES, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_SERVER_VERSION, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
 
   if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY, mmPutMsgToQueryQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_CONTINUE, mmPutMsgToQueryQueue, 1) == NULL) goto _OVER;

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

@@ -27,12 +27,10 @@ void    mndCleanupScheduler(SMnode* pMnode);
 
 int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscribeObj* pSub);
 
-int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream);
-
 int32_t mndConvertRsmaTask(char** pDst, int32_t* pDstLen, const char* ast, int64_t uid, int8_t triggerType,
                            int64_t watermark);
 
-int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream);
+int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream);
 
 #ifdef __cplusplus
 }

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

@@ -48,6 +48,7 @@ static int32_t  mndDnodeActionInsert(SSdb *pSdb, SDnodeObj *pDnode);
 static int32_t  mndDnodeActionDelete(SSdb *pSdb, SDnodeObj *pDnode);
 static int32_t  mndDnodeActionUpdate(SSdb *pSdb, SDnodeObj *pOld, SDnodeObj *pNew);
 static int32_t  mndProcessDnodeListReq(SRpcMsg *pReq);
+static int32_t  mndProcessShowVariablesReq(SRpcMsg *pReq);
 
 static int32_t mndProcessCreateDnodeReq(SRpcMsg *pReq);
 static int32_t mndProcessDropDnodeReq(SRpcMsg *pReq);
@@ -78,6 +79,7 @@ int32_t mndInitDnode(SMnode *pMnode) {
   mndSetMsgHandle(pMnode, TDMT_DND_CONFIG_DNODE_RSP, mndProcessConfigDnodeRsp);
   mndSetMsgHandle(pMnode, TDMT_MND_STATUS, mndProcessStatusReq);
   mndSetMsgHandle(pMnode, TDMT_MND_DNODE_LIST, mndProcessDnodeListReq);
+  mndSetMsgHandle(pMnode, TDMT_MND_SHOW_VARIABLES, mndProcessShowVariablesReq);
 
   mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_CONFIGS, mndRetrieveConfigs);
   mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_CONFIGS, mndCancelGetNextConfig);
@@ -554,6 +556,60 @@ _OVER:
   return code;
 }
 
+static int32_t mndProcessShowVariablesReq(SRpcMsg *pReq) {
+  SShowVariablesRsp rsp = {0};
+  int32_t       code = -1;
+
+  rsp.variables = taosArrayInit(4, sizeof(SVariablesInfo));
+  if (NULL == rsp.variables) {
+    mError("failed to alloc SVariablesInfo array while process show variables req");
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    goto _OVER;
+  }
+
+  SVariablesInfo info = {0};
+
+  strcpy(info.name, "statusInterval");
+  snprintf(info.value, TSDB_CONFIG_VALUE_LEN, "%d", tsStatusInterval);
+  taosArrayPush(rsp.variables, &info);
+
+  strcpy(info.name, "timezone");
+  snprintf(info.value, TSDB_CONFIG_VALUE_LEN, "%s", tsTimezoneStr);
+  taosArrayPush(rsp.variables, &info);
+  
+  strcpy(info.name, "locale");
+  snprintf(info.value, TSDB_CONFIG_VALUE_LEN, "%s", tsLocale);
+  taosArrayPush(rsp.variables, &info);
+
+  strcpy(info.name, "charset");
+  snprintf(info.value, TSDB_CONFIG_VALUE_LEN, "%s", tsCharset);
+  taosArrayPush(rsp.variables, &info);
+
+  int32_t rspLen = tSerializeSShowVariablesRsp(NULL, 0, &rsp);
+  void   *pRsp = rpcMallocCont(rspLen);
+  if (pRsp == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    goto _OVER;
+  }
+
+  tSerializeSShowVariablesRsp(pRsp, rspLen, &rsp);
+
+  pReq->info.rspLen = rspLen;
+  pReq->info.rsp = pRsp;
+  code = 0;
+
+_OVER:
+
+  if (code != 0) {
+    mError("failed to get show variables info since %s", terrstr());
+  }
+
+  tFreeSShowVariablesRsp(&rsp);
+
+  return code;
+}
+
+
 static int32_t mndProcessCreateDnodeReq(SRpcMsg *pReq) {
   SMnode         *pMnode = pReq->info.node;
   int32_t         code = -1;

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

@@ -70,6 +70,7 @@ static void      mndCancelGetNextQuery(SMnode *pMnode, void *pIter);
 static void      mndFreeApp(SAppObj *pApp);
 static int32_t   mndRetrieveApps(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
 static void      mndCancelGetNextApp(SMnode *pMnode, void *pIter);
+static int32_t   mndProcessSvrVerReq(SRpcMsg *pReq);
 
 int32_t mndInitProfile(SMnode *pMnode) {
   SProfileMgmt *pMgmt = &pMnode->profileMgmt;
@@ -94,6 +95,7 @@ int32_t mndInitProfile(SMnode *pMnode) {
   mndSetMsgHandle(pMnode, TDMT_MND_CONNECT, mndProcessConnectReq);
   mndSetMsgHandle(pMnode, TDMT_MND_KILL_QUERY, mndProcessKillQueryReq);
   mndSetMsgHandle(pMnode, TDMT_MND_KILL_CONN, mndProcessKillConnReq);
+  mndSetMsgHandle(pMnode, TDMT_MND_SERVER_VERSION, mndProcessSvrVerReq);
 
   mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_CONNS, mndRetrieveConns);
   mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_CONNS, mndCancelGetNextConn);
@@ -262,8 +264,9 @@ static int32_t mndProcessConnectReq(SRpcMsg *pReq) {
   connectRsp.connId = pConn->id;
   connectRsp.connType = connReq.connType;
   connectRsp.dnodeNum = mndGetDnodeSize(pMnode);
-
-  snprintf(connectRsp.sVersion, sizeof(connectRsp.sVersion), "ver:%s\nbuild:%s\ngitinfo:%s", version, buildinfo,
+  
+  strcpy(connectRsp.sVer, version);
+  snprintf(connectRsp.sDetailVer, sizeof(connectRsp.sDetailVer), "ver:%s\nbuild:%s\ngitinfo:%s", version, buildinfo,
            gitinfo);
   mndGetMnodeEpSet(pMnode, &connectRsp.epSet);
 
@@ -460,6 +463,27 @@ static int32_t mndUpdateAppInfo(SMnode *pMnode, SClientHbReq *pHbReq, SRpcConnIn
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t mndGetOnlineDnodeNum(SMnode *pMnode, int32_t *num) {
+  SSdb      *pSdb = pMnode->pSdb;
+  SDnodeObj *pDnode = NULL;
+  int64_t    curMs = taosGetTimestampMs();
+  void      *pIter = NULL;
+  
+  while (true) {
+    pIter = sdbFetch(pSdb, SDB_DNODE, pIter, (void **)&pDnode);
+    if (pIter == NULL) break;
+    
+    bool online = mndIsDnodeOnline(pDnode, curMs);
+    if (online) {
+      (*num)++;
+    }
+    
+    sdbRelease(pSdb, pDnode);
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t mndProcessQueryHeartBeat(SMnode *pMnode, SRpcMsg *pMsg, SClientHbReq *pHbReq,
                                         SClientHbBatchRsp *pBatchRsp) {
   SProfileMgmt *pMgmt = &pMnode->profileMgmt;
@@ -503,7 +527,7 @@ static int32_t mndProcessQueryHeartBeat(SMnode *pMnode, SRpcMsg *pMsg, SClientHb
 
     rspBasic->connId = pConn->id;
     rspBasic->totalDnodes = mndGetDnodeSize(pMnode);
-    rspBasic->onlineDnodes = 1;  // TODO
+    mndGetOnlineDnodeNum(pMnode, &rspBasic->onlineDnodes);
     mndGetMnodeEpSet(pMnode, &rspBasic->epSet);
 
     mndCreateQnodeList(pMnode, &rspBasic->pQnodeList, -1);
@@ -694,6 +718,28 @@ static int32_t mndProcessKillConnReq(SRpcMsg *pReq) {
   }
 }
 
+static int32_t mndProcessSvrVerReq(SRpcMsg *pReq) {
+  int32_t code = -1;
+  SServerVerRsp rsp = {0};
+  strcpy(rsp.ver, version);
+  
+  int32_t contLen = tSerializeSServerVerRsp(NULL, 0, &rsp);
+  if (contLen < 0) goto _over;
+  void *pRsp = rpcMallocCont(contLen);
+  if (pRsp == NULL) goto _over;
+  tSerializeSServerVerRsp(pRsp, contLen, &rsp);
+
+  pReq->info.rspLen = contLen;
+  pReq->info.rsp = pRsp;
+
+  code = 0;
+
+_over:
+
+  return code;
+}
+
+
 static int32_t mndRetrieveConns(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
   SMnode   *pMnode = pReq->info.node;
   SSdb     *pSdb = pMnode->pSdb;

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

@@ -97,37 +97,7 @@ END:
   return terrno;
 }
 
-int32_t mndPersistTaskDeployReq(STrans* pTrans, SStreamTask* pTask, const SEpSet* pEpSet, tmsg_t type, int32_t nodeId) {
-  SEncoder encoder;
-  tEncoderInit(&encoder, NULL, 0);
-  tEncodeSStreamTask(&encoder, pTask);
-  int32_t size = encoder.pos;
-  int32_t tlen = sizeof(SMsgHead) + size;
-  tEncoderClear(&encoder);
-  void* buf = taosMemoryCalloc(1, tlen);
-  if (buf == NULL) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return -1;
-  }
-  ((SMsgHead*)buf)->vgId = htonl(nodeId);
-  void* abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
-  tEncoderInit(&encoder, abuf, size);
-  tEncodeSStreamTask(&encoder, pTask);
-  tEncoderClear(&encoder);
-
-  STransAction action = {0};
-  memcpy(&action.epSet, pEpSet, sizeof(SEpSet));
-  action.pCont = buf;
-  action.contLen = tlen;
-  action.msgType = type;
-  if (mndTransAppendRedoAction(pTrans, &action) != 0) {
-    taosMemoryFree(buf);
-    return -1;
-  }
-  return 0;
-}
-
-int32_t mndAddSinkToTask(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream, SStreamTask* pTask) {
+int32_t mndAddSinkToTask(SMnode* pMnode, SStreamObj* pStream, SStreamTask* pTask) {
   pTask->dispatchType = TASK_DISPATCH__NONE;
   // sink
   if (pStream->smaId != 0) {
@@ -142,7 +112,7 @@ int32_t mndAddSinkToTask(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream, SS
   return 0;
 }
 
-int32_t mndAddDispatcherToInnerTask(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream, SStreamTask* pTask) {
+int32_t mndAddDispatcherToInnerTask(SMnode* pMnode, SStreamObj* pStream, SStreamTask* pTask) {
   pTask->sinkType = TASK_SINK__NONE;
   if (pStream->fixedSinkVgId == 0) {
     pTask->dispatchType = TASK_DISPATCH__SHUFFLE;
@@ -187,7 +157,7 @@ int32_t mndAddDispatcherToInnerTask(SMnode* pMnode, STrans* pTrans, SStreamObj*
   return 0;
 }
 
-int32_t mndAssignTaskToVg(SMnode* pMnode, STrans* pTrans, SStreamTask* pTask, SSubplan* plan, const SVgObj* pVgroup) {
+int32_t mndAssignTaskToVg(SMnode* pMnode, SStreamTask* pTask, SSubplan* plan, const SVgObj* pVgroup) {
   int32_t msgLen;
   pTask->nodeId = pVgroup->vgId;
   pTask->epSet = mndGetVgroupEpset(pMnode, pVgroup);
@@ -196,11 +166,11 @@ int32_t mndAssignTaskToVg(SMnode* pMnode, STrans* pTrans, SStreamTask* pTask, SS
   plan->execNode.epSet = pTask->epSet;
 
   if (qSubPlanToString(plan, &pTask->exec.qmsg, &msgLen) < 0) {
+    ASSERT(0);
     terrno = TSDB_CODE_QRY_INVALID_INPUT;
     return -1;
   }
   ASSERT(pTask->dispatchType != TASK_DISPATCH__NONE || pTask->sinkType != TASK_SINK__NONE);
-  mndPersistTaskDeployReq(pTrans, pTask, &plan->execNode.epSet, TDMT_STREAM_TASK_DEPLOY, pVgroup->vgId);
   return 0;
 }
 
@@ -212,8 +182,7 @@ SSnodeObj* mndSchedFetchOneSnode(SMnode* pMnode) {
   return pObj;
 }
 
-int32_t mndAssignTaskToSnode(SMnode* pMnode, STrans* pTrans, SStreamTask* pTask, SSubplan* plan,
-                             const SSnodeObj* pSnode) {
+int32_t mndAssignTaskToSnode(SMnode* pMnode, SStreamTask* pTask, SSubplan* plan, const SSnodeObj* pSnode) {
   int32_t msgLen;
 
   pTask->nodeId = SNODE_HANDLE;
@@ -223,10 +192,10 @@ int32_t mndAssignTaskToSnode(SMnode* pMnode, STrans* pTrans, SStreamTask* pTask,
   plan->execNode.epSet = pTask->epSet;
 
   if (qSubPlanToString(plan, &pTask->exec.qmsg, &msgLen) < 0) {
+    ASSERT(0);
     terrno = TSDB_CODE_QRY_INVALID_INPUT;
     return -1;
   }
-  mndPersistTaskDeployReq(pTrans, pTask, &plan->execNode.epSet, TDMT_STREAM_TASK_DEPLOY, SNODE_HANDLE);
   return 0;
 }
 
@@ -245,7 +214,7 @@ SVgObj* mndSchedFetchOneVg(SMnode* pMnode, int64_t dbUid) {
   return pVgroup;
 }
 
-int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
+int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, SStreamObj* pStream) {
   SSdb*   pSdb = pMnode->pSdb;
   void*   pIter = NULL;
   SArray* tasks = taosArrayGetP(pStream->tasks, 0);
@@ -262,6 +231,7 @@ int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, STrans* pTrans, SStreamOb
     }
     SStreamTask* pTask = tNewSStreamTask(pStream->uid);
     if (pTask == NULL) {
+      sdbRelease(pSdb, pVgroup);
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       return -1;
     }
@@ -290,13 +260,11 @@ int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, STrans* pTrans, SStreamOb
 
     // dispatch
     pTask->dispatchType = TASK_DISPATCH__NONE;
-
-    mndPersistTaskDeployReq(pTrans, pTask, &pTask->epSet, TDMT_STREAM_TASK_DEPLOY, pVgroup->vgId);
   }
   return 0;
 }
 
-int32_t mndAddFixedSinkTaskToStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
+int32_t mndAddFixedSinkTaskToStream(SMnode* pMnode, SStreamObj* pStream) {
   ASSERT(pStream->fixedSinkVgId != 0);
   SArray*      tasks = taosArrayGetP(pStream->tasks, 0);
   SStreamTask* pTask = tNewSStreamTask(pStream->uid);
@@ -337,13 +305,10 @@ int32_t mndAddFixedSinkTaskToStream(SMnode* pMnode, STrans* pTrans, SStreamObj*
 
   // dispatch
   pTask->dispatchType = TASK_DISPATCH__NONE;
-
-  mndPersistTaskDeployReq(pTrans, pTask, &pTask->epSet, TDMT_STREAM_TASK_DEPLOY, pStream->fixedSinkVg.vgId);
-
   return 0;
 }
 
-int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
+int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
   SSdb*       pSdb = pMnode->pSdb;
   SQueryPlan* pPlan = qStringToQueryPlan(pStream->physicalPlan);
   if (pPlan == NULL) {
@@ -368,9 +333,15 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
     // add extra sink
     hasExtraSink = true;
     if (pStream->fixedSinkVgId == 0) {
-      mndAddShuffleSinkTasksToStream(pMnode, pTrans, pStream);
+      if (mndAddShuffleSinkTasksToStream(pMnode, pStream) < 0) {
+        // TODO free
+        return -1;
+      }
     } else {
-      mndAddFixedSinkTaskToStream(pMnode, pTrans, pStream);
+      if (mndAddFixedSinkTaskToStream(pMnode, pStream) < 0) {
+        // TODO free
+        return -1;
+      }
     }
   }
 
@@ -386,6 +357,11 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
       ASSERT(plan->subplanType == SUBPLAN_TYPE_MERGE);
 
       pInnerTask = tNewSStreamTask(pStream->uid);
+      if (pInnerTask == NULL) {
+        terrno = TSDB_CODE_OUT_OF_MEMORY;
+        qDestroyQueryPlan(pPlan);
+        return -1;
+      }
       mndAddTaskToTaskSet(taskInnerLevel, pInnerTask);
 
       pInnerTask->childEpInfo = taosArrayInit(0, sizeof(void*));
@@ -397,7 +373,7 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
       pInnerTask->triggerParam = pStream->triggerParam;
 
       // dispatch
-      if (mndAddDispatcherToInnerTask(pMnode, pTrans, pStream, pInnerTask) < 0) {
+      if (mndAddDispatcherToInnerTask(pMnode, pStream, pInnerTask) < 0) {
         qDestroyQueryPlan(pPlan);
         return -1;
       }
@@ -409,14 +385,13 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
         SSnodeObj* pSnode = mndSchedFetchOneSnode(pMnode);
         if (pSnode == NULL) {
           SVgObj* pVgroup = mndSchedFetchOneVg(pMnode, pStream->sourceDbUid);
-          if (mndAssignTaskToVg(pMnode, pTrans, pInnerTask, plan, pVgroup) < 0) {
+          if (mndAssignTaskToVg(pMnode, pInnerTask, plan, pVgroup) < 0) {
             sdbRelease(pSdb, pVgroup);
             qDestroyQueryPlan(pPlan);
             return -1;
           }
         } else {
-          if (mndAssignTaskToSnode(pMnode, pTrans, pInnerTask, plan, pSnode) < 0) {
-            ASSERT(0);
+          if (mndAssignTaskToSnode(pMnode, pInnerTask, plan, pSnode) < 0) {
             sdbRelease(pSdb, pSnode);
             qDestroyQueryPlan(pPlan);
             return -1;
@@ -424,7 +399,7 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
         }
       } else {
         SVgObj* pVgroup = mndSchedFetchOneVg(pMnode, pStream->sourceDbUid);
-        if (mndAssignTaskToVg(pMnode, pTrans, pInnerTask, plan, pVgroup) < 0) {
+        if (mndAssignTaskToVg(pMnode, pInnerTask, plan, pVgroup) < 0) {
           sdbRelease(pSdb, pVgroup);
           qDestroyQueryPlan(pPlan);
           return -1;
@@ -450,6 +425,12 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
         continue;
       }
       SStreamTask* pTask = tNewSStreamTask(pStream->uid);
+      if (pInnerTask == NULL) {
+        terrno = TSDB_CODE_OUT_OF_MEMORY;
+        sdbRelease(pSdb, pVgroup);
+        qDestroyQueryPlan(pPlan);
+        return -1;
+      }
       mndAddTaskToTaskSet(taskSourceLevel, pTask);
 
       // source
@@ -466,7 +447,7 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
 
       // exec
       pTask->execType = TASK_EXEC__PIPE;
-      if (mndAssignTaskToVg(pMnode, pTrans, pTask, plan, pVgroup) < 0) {
+      if (mndAssignTaskToVg(pMnode, pTask, plan, pVgroup) < 0) {
         sdbRelease(pSdb, pVgroup);
         qDestroyQueryPlan(pPlan);
         return -1;
@@ -507,6 +488,11 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
         continue;
       }
       SStreamTask* pTask = tNewSStreamTask(pStream->uid);
+      if (pTask == NULL) {
+        sdbRelease(pSdb, pVgroup);
+        qDestroyQueryPlan(pPlan);
+        return -1;
+      }
       mndAddTaskToTaskSet(taskOneLevel, pTask);
 
       // source
@@ -517,14 +503,14 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
 
       // sink or dispatch
       if (hasExtraSink) {
-        mndAddDispatcherToInnerTask(pMnode, pTrans, pStream, pTask);
+        mndAddDispatcherToInnerTask(pMnode, pStream, pTask);
       } else {
-        mndAddSinkToTask(pMnode, pTrans, pStream, pTask);
+        mndAddSinkToTask(pMnode, pStream, pTask);
       }
 
       // exec
       pTask->execType = TASK_EXEC__PIPE;
-      if (mndAssignTaskToVg(pMnode, pTrans, pTask, plan, pVgroup) < 0) {
+      if (mndAssignTaskToVg(pMnode, pTask, plan, pVgroup) < 0) {
         sdbRelease(pSdb, pVgroup);
         qDestroyQueryPlan(pPlan);
         return -1;

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

@@ -533,7 +533,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
 #if 0
   smaObj.timezone = pCreate->timezone;
 #endif
-  smaObj.timezone = tsTimezone; // use timezone of server
+  smaObj.timezone = tsTimezone;  // use timezone of server
   smaObj.interval = pCreate->interval;
   smaObj.offset = pCreate->offset;
   smaObj.sliding = pCreate->sliding;
@@ -623,7 +623,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
   if (mndSetUpdateSmaStbCommitLogs(pMnode, pTrans, pStb) != 0) goto _OVER;
   // if (mndSetCreateSmaRedoActions(pMnode, pTrans, pDb, &smaObj) != 0) goto _OVER;
   if (mndSetCreateSmaVgroupRedoActions(pMnode, pTrans, pDb, &streamObj.fixedSinkVg, &smaObj) != 0) goto _OVER;
-  if (mndScheduleStream(pMnode, pTrans, &streamObj) != 0) goto _OVER;
+  if (mndScheduleStream(pMnode, &streamObj) != 0) goto _OVER;
   if (mndPersistStream(pMnode, pTrans, &streamObj) != 0) goto _OVER;
   if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
 

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

@@ -320,70 +320,72 @@ FAIL:
   return 0;
 }
 
-int32_t mndPersistStream(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream) {
-  SSdbRaw *pCommitRaw = mndStreamActionEncode(pStream);
-  if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
-    mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
-    mndTransDrop(pTrans);
+int32_t mndPersistTaskDeployReq(STrans *pTrans, const SStreamTask *pTask) {
+  SEncoder encoder;
+  tEncoderInit(&encoder, NULL, 0);
+  tEncodeSStreamTask(&encoder, pTask);
+  int32_t size = encoder.pos;
+  int32_t tlen = sizeof(SMsgHead) + size;
+  tEncoderClear(&encoder);
+  void *buf = taosMemoryCalloc(1, tlen);
+  if (buf == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
   }
-  sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
-  return 0;
-}
+  ((SMsgHead *)buf)->vgId = htonl(pTask->nodeId);
+  void *abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
+  tEncoderInit(&encoder, abuf, size);
+  tEncodeSStreamTask(&encoder, pTask);
+  tEncoderClear(&encoder);
 
-int32_t mndPersistDropStreamLog(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream) {
-  SSdbRaw *pCommitRaw = mndStreamActionEncode(pStream);
-  if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
-    mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
-    mndTransDrop(pTrans);
+  STransAction action = {0};
+  memcpy(&action.epSet, &pTask->epSet, sizeof(SEpSet));
+  action.pCont = buf;
+  action.contLen = tlen;
+  action.msgType = TDMT_STREAM_TASK_DEPLOY;
+  if (mndTransAppendRedoAction(pTrans, &action) != 0) {
+    taosMemoryFree(buf);
     return -1;
   }
-  sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
   return 0;
 }
 
-int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast, STrans *pTrans) {
-  SNode *pAst = NULL;
-
-  if (nodesStringToNode(ast, &pAst) < 0) {
-    return -1;
-  }
-
-  if (qExtractResultSchema(pAst, (int32_t *)&pStream->outputSchema.nCols, &pStream->outputSchema.pSchema) != 0) {
-    nodesDestroyNode(pAst);
-    return -1;
-  }
-  // free
-  nodesDestroyNode(pAst);
-
-#if 0
-  printf("|");
-  for (int i = 0; i < pStream->outputSchema.nCols; i++) {
-    printf(" %15s |", (char *)pStream->outputSchema.pSchema[i].name);
+int32_t mndPersistStreamTasks(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream) {
+  int32_t level = taosArrayGetSize(pStream->tasks);
+  for (int32_t i = 0; i < level; i++) {
+    SArray *pLevel = taosArrayGetP(pStream->tasks, i);
+    int32_t sz = taosArrayGetSize(pLevel);
+    for (int32_t j = 0; j < sz; j++) {
+      SStreamTask *pTask = taosArrayGetP(pLevel, j);
+      if (mndPersistTaskDeployReq(pTrans, pTask) < 0) {
+        return -1;
+      }
+    }
   }
-  printf("\n=======================================================\n");
-
-#endif
+  return 0;
+}
 
-  if (TSDB_CODE_SUCCESS != mndStreamGetPlanString(ast, pStream->trigger, pStream->watermark, &pStream->physicalPlan)) {
-    mError("topic:%s, failed to get plan since %s", pStream->name, terrstr());
+int32_t mndPersistStream(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream) {
+  if (mndPersistStreamTasks(pMnode, pTrans, pStream) < 0) {
     return -1;
   }
-
-  if (mndScheduleStream(pMnode, pTrans, pStream) < 0) {
-    mError("stream:%ld, schedule stream since %s", pStream->uid, terrstr());
+  SSdbRaw *pCommitRaw = mndStreamActionEncode(pStream);
+  if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
+    mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
     return -1;
   }
-  mDebug("trans:%d, used to create stream:%s", pTrans->id, pStream->name);
+  sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
+  return 0;
+}
 
+int32_t mndPersistDropStreamLog(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream) {
   SSdbRaw *pCommitRaw = mndStreamActionEncode(pStream);
   if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
     mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
     mndTransDrop(pTrans);
     return -1;
   }
-  sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
-
+  sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
   return 0;
 }
 
@@ -503,64 +505,6 @@ static int32_t mndDropStreamTasks(SMnode *pMnode, STrans *pTrans, SStreamObj *pS
   return 0;
 }
 
-static int32_t mndCreateStream(SMnode *pMnode, SRpcMsg *pReq, SCMCreateStreamReq *pCreate, SDbObj *pDb) {
-  mDebug("stream:%s to create", pCreate->name);
-  SStreamObj streamObj = {0};
-  tstrncpy(streamObj.name, pCreate->name, TSDB_STREAM_FNAME_LEN);
-  tstrncpy(streamObj.sourceDb, pDb->name, TSDB_DB_FNAME_LEN);
-  tstrncpy(streamObj.targetSTbName, pCreate->targetStbFullName, TSDB_TABLE_FNAME_LEN);
-  streamObj.createTime = taosGetTimestampMs();
-  streamObj.updateTime = streamObj.createTime;
-  streamObj.uid = mndGenerateUid(pCreate->name, strlen(pCreate->name));
-  streamObj.targetStbUid = mndGenerateUid(pCreate->targetStbFullName, TSDB_TABLE_FNAME_LEN);
-  streamObj.sourceDbUid = pDb->uid;
-  streamObj.version = 1;
-  streamObj.sql = pCreate->sql;
-  // TODO
-  streamObj.fixedSinkVgId = 0;
-  streamObj.smaId = 0;
-  streamObj.trigger = pCreate->triggerType;
-  streamObj.watermark = pCreate->watermark;
-  streamObj.triggerParam = pCreate->maxDelay;
-
-  if (streamObj.targetSTbName[0]) {
-    pDb = mndAcquireDbByStb(pMnode, streamObj.targetSTbName);
-    if (pDb == NULL) {
-      terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
-      return -1;
-    }
-    tstrncpy(streamObj.targetDb, pDb->name, TSDB_DB_FNAME_LEN);
-  }
-
-  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq);
-  if (pTrans == NULL) {
-    mError("stream:%s, failed to create since %s", pCreate->name, terrstr());
-    return -1;
-  }
-  mDebug("trans:%d, used to create stream:%s", pTrans->id, pCreate->name);
-
-  if (mndAddStreamToTrans(pMnode, &streamObj, pCreate->ast, pTrans) != 0) {
-    mError("trans:%d, failed to add stream since %s", pTrans->id, terrstr());
-    mndTransDrop(pTrans);
-    return -1;
-  }
-
-  if (streamObj.targetSTbName[0] && mndCreateStbForStream(pMnode, pTrans, &streamObj, pReq->info.conn.user) < 0) {
-    mError("trans:%d, failed to create stb for stream since %s", pTrans->id, terrstr());
-    mndTransDrop(pTrans);
-    return -1;
-  }
-
-  if (mndTransPrepare(pMnode, pTrans) != 0) {
-    mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
-    mndTransDrop(pTrans);
-    return -1;
-  }
-
-  mndTransDrop(pTrans);
-  return 0;
-}
-
 static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
   SMnode            *pMnode = pReq->info.node;
   int32_t            code = -1;
@@ -631,7 +575,7 @@ static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
   }
 
   // schedule stream task for stream obj
-  if (mndScheduleStream(pMnode, pTrans, &streamObj) < 0) {
+  if (mndScheduleStream(pMnode, &streamObj) < 0) {
     mError("stream:%s, failed to schedule since %s", createStreamReq.name, terrstr());
     mndTransDrop(pTrans);
     goto _OVER;
@@ -653,8 +597,6 @@ static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
 
   mndTransDrop(pTrans);
 
-  /*code = mndCreateStream(pMnode, pReq, &createStreamReq, pDb);*/
-  /*if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;*/
   code = TSDB_CODE_ACTION_IN_PROGRESS;
 
 _OVER:

source/dnode/mnode/impl/test/db/CMakeLists.txt

@@ -5,9 +5,7 @@ target_link_libraries(
     PUBLIC sut
 )
 
-if(NOT TD_WINDOWS)
-    add_test(
-        NAME dbTest
-        COMMAND dbTest
-    )
-endif(NOT TD_WINDOWS)
+add_test(
+    NAME dbTest
+    COMMAND dbTest
+)

source/dnode/mnode/impl/test/func/CMakeLists.txt

@@ -5,9 +5,7 @@ target_link_libraries(
     PUBLIC sut
 )
 
-# if(NOT TD_WINDOWS)
-    add_test(
-        NAME funcTest
-        COMMAND funcTest
-    )
+add_test(
+    NAME funcTest
+    COMMAND funcTest
+)
\ No newline at end of file
-# endif(NOT TD_WINDOWS)

source/dnode/mnode/impl/test/profile/CMakeLists.txt

@@ -5,9 +5,7 @@ target_link_libraries(
     PUBLIC sut
 )
 
-# if(NOT TD_WINDOWS)
-    add_test(
-        NAME profileTest
-        COMMAND profileTest
-    )
-# endif(NOT TD_WINDOWS)
+add_test(
+    NAME profileTest
+    COMMAND profileTest
+)

source/dnode/mnode/impl/test/show/CMakeLists.txt

@@ -5,9 +5,7 @@ target_link_libraries(
     PUBLIC sut
 )
 
-# if(NOT TD_WINDOWS)
-    add_test(
-        NAME showTest
-        COMMAND showTest
-    )
-# endif(NOT TD_WINDOWS)
+add_test(
+    NAME showTest
+    COMMAND showTest
+)

source/dnode/mnode/impl/test/sma/CMakeLists.txt

@@ -5,9 +5,7 @@ target_link_libraries(
     PUBLIC sut
 )
 
-if(NOT TD_WINDOWS)
-    add_test(
-        NAME smaTest
-        COMMAND smaTest
-    )
-endif(NOT TD_WINDOWS)
+add_test(
+    NAME smaTest
+    COMMAND smaTest
+)

source/dnode/mnode/impl/test/stb/CMakeLists.txt

@@ -5,9 +5,7 @@ target_link_libraries(
     PUBLIC sut
 )
 
-if(NOT TD_WINDOWS)
-    add_test(
-        NAME stbTest
-        COMMAND stbTest
-    )
+add_test(
+    NAME stbTest
+    COMMAND stbTest
+)
\ No newline at end of file
-endif(NOT TD_WINDOWS)
\ No newline at end of file

source/dnode/mnode/impl/test/user/CMakeLists.txt

@@ -5,9 +5,7 @@ target_link_libraries(
     PUBLIC sut
 )
 
-if(NOT TD_WINDOWS)
-    add_test(
-        NAME userTest
-        COMMAND userTest
-    )
-endif(NOT TD_WINDOWS)
+add_test(
+    NAME userTest
+    COMMAND userTest
+)

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

@@ -146,8 +146,7 @@ int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReadHandle* pHandle, uint64_
                             int32_t* pNumOfRows) {
   *pUid = 0;
 
-  // TODO set to real sversion
-  /*int32_t sversion = 1;*/
+  // TODO: cache multiple schema
   int32_t sversion = htonl(pHandle->pBlock->sversion);
   if (pHandle->cachedSchemaSuid == 0 || pHandle->cachedSchemaVer != sversion ||
       pHandle->cachedSchemaSuid != pHandle->msgIter.suid) {
@@ -161,7 +160,6 @@ int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReadHandle* pHandle, uint64_
       return -1;
     }
 
-    // this interface use suid instead of uid
     if (pHandle->pSchemaWrapper) tDeleteSSchemaWrapper(pHandle->pSchemaWrapper);
     pHandle->pSchemaWrapper = metaGetTableSchema(pHandle->pVnodeMeta, pHandle->msgIter.uid, sversion, true);
     if (pHandle->pSchemaWrapper == NULL) {

source/libs/catalog/inc/catalogInt.h

@@ -76,6 +76,7 @@ typedef enum {
   CTG_TASK_GET_INDEX,
   CTG_TASK_GET_UDF,
   CTG_TASK_GET_USER,
+  CTG_TASK_GET_SVR_VER,
 } CTG_TASK_TYPE;
 
 typedef enum {
@@ -224,6 +225,7 @@ typedef struct SCtgJob {
   int32_t           dbInfoNum;
   int32_t           tbIndexNum;
   int32_t           tbCfgNum;
+  int32_t           svrVerNum;
 } SCtgJob;
 
 typedef struct SCtgMsgCtx {
@@ -592,8 +594,9 @@ int32_t ctgGetTbMetaFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SNa
 int32_t ctgGetTbMetaFromVnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, SVgroupInfo *vgroupInfo, STableMetaOutput* out, SCtgTask* pTask);
 int32_t ctgGetTableCfgFromVnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, SVgroupInfo *vgroupInfo, STableCfg **out, SCtgTask* pTask);
 int32_t ctgGetTableCfgFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, STableCfg **out, SCtgTask* pTask);
+int32_t ctgGetSvrVerFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, char **out, SCtgTask* pTask);
 
-int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgJob** job, uint64_t reqId, const SCatalogReq* pReq, catalogCallback fp, void* param, int32_t* taskNum);
+int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgJob** job, const SCatalogReq* pReq, catalogCallback fp, void* param, int32_t* taskNum);
 int32_t ctgLaunchJob(SCtgJob *pJob);
 int32_t ctgMakeAsyncRes(SCtgJob *pJob);
 int32_t ctgLaunchSubTask(SCtgTask *pTask, CTG_TASK_TYPE type, ctgSubTaskCbFp fp, void* param);

source/libs/catalog/src/catalog.c

@@ -1034,7 +1034,7 @@ _return:
   CTG_API_LEAVE(code);
 }
 
-int32_t catalogAsyncGetAllMeta(SCatalog* pCtg, SRequestConnInfo *pConn, uint64_t reqId, const SCatalogReq* pReq, catalogCallback fp, void* param, int64_t* jobId) {
+int32_t catalogAsyncGetAllMeta(SCatalog* pCtg, SRequestConnInfo *pConn, const SCatalogReq* pReq, catalogCallback fp, void* param, int64_t* jobId) {
   CTG_API_ENTER();
 
   if (NULL == pCtg || NULL == pConn || NULL == pReq || NULL == fp || NULL == param) {
@@ -1043,7 +1043,7 @@ int32_t catalogAsyncGetAllMeta(SCatalog* pCtg, SRequestConnInfo *pConn, uint64_t
 
   int32_t code = 0, taskNum = 0;
   SCtgJob *pJob = NULL;
-  CTG_ERR_JRET(ctgInitJob(pCtg, pConn, &pJob, reqId, pReq, fp, param, &taskNum));
+  CTG_ERR_JRET(ctgInitJob(pCtg, pConn, &pJob, pReq, fp, param, &taskNum));
   if (taskNum <= 0) {
     SMetaData* pMetaData = taosMemoryCalloc(1, sizeof(SMetaData));
     fp(pMetaData, param, TSDB_CODE_SUCCESS);
@@ -1230,6 +1230,22 @@ _return:
   CTG_API_LEAVE(code);
 }
 
+int32_t catalogGetServerVersion(SCatalog* pCtg, SRequestConnInfo *pConn, char** pVersion) {
+  CTG_API_ENTER();
+  
+  if (NULL == pCtg || NULL == pConn || NULL == pVersion) {
+    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
+  }
+
+  int32_t code = 0;
+  CTG_ERR_JRET(ctgGetSvrVerFromMnode(pCtg, pConn, pVersion, NULL));
+  
+_return:
+
+  CTG_API_LEAVE(code);
+}
+
+
 int32_t catalogUpdateUserAuthInfo(SCatalog* pCtg, SGetUserAuthRsp* pAuth) {
   CTG_API_ENTER();
 

source/libs/catalog/src/ctgAsync.c

@@ -255,6 +255,20 @@ int32_t ctgInitGetUserTask(SCtgJob *pJob, int32_t taskIdx, void* param) {
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t ctgInitGetSvrVerTask(SCtgJob *pJob, int32_t taskIdx, void* param) {
+  SCtgTask task = {0};
+
+  task.type = CTG_TASK_GET_SVR_VER;
+  task.taskId = taskIdx;
+  task.pJob = pJob;
+
+  taosArrayPush(pJob->pTasks, &task);
+
+  qDebug("QID:0x%" PRIx64 " [%dth] task type %s initialized", pJob->queryId, taskIdx, ctgTaskTypeStr(task.type));
+
+  return TSDB_CODE_SUCCESS;
+}
+
 int32_t ctgInitGetTbIndexTask(SCtgJob *pJob, int32_t taskIdx, void* param) {
   SName *name = (SName*)param;
   SCtgTask task = {0};
@@ -413,7 +427,7 @@ int32_t ctgInitTask(SCtgJob *pJob, CTG_TASK_TYPE type, void* param, int32_t *tas
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgJob** job, uint64_t reqId, const SCatalogReq* pReq, catalogCallback fp, void* param, int32_t* taskNum) {
+int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgJob** job, const SCatalogReq* pReq, catalogCallback fp, void* param, int32_t* taskNum) {
   int32_t code = 0;
   int32_t tbMetaNum = (int32_t)taosArrayGetSize(pReq->pTableMeta);
   int32_t dbVgNum = (int32_t)taosArrayGetSize(pReq->pDbVgroup);
@@ -421,6 +435,7 @@ int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgJob** job, uint6
   int32_t udfNum = (int32_t)taosArrayGetSize(pReq->pUdf);
   int32_t qnodeNum = pReq->qNodeRequired ? 1 : 0;
   int32_t dnodeNum = pReq->dNodeRequired ? 1 : 0;
+  int32_t svrVerNum = pReq->svrVerRequired ? 1 : 0;
   int32_t dbCfgNum = (int32_t)taosArrayGetSize(pReq->pDbCfg);
   int32_t indexNum = (int32_t)taosArrayGetSize(pReq->pIndex);
   int32_t userNum = (int32_t)taosArrayGetSize(pReq->pUser);
@@ -428,21 +443,21 @@ int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgJob** job, uint6
   int32_t tbIndexNum = (int32_t)taosArrayGetSize(pReq->pTableIndex);
   int32_t tbCfgNum = (int32_t)taosArrayGetSize(pReq->pTableCfg);
 
-  *taskNum = tbMetaNum + dbVgNum + udfNum + tbHashNum + qnodeNum + dnodeNum + dbCfgNum + indexNum + userNum + dbInfoNum + tbIndexNum + tbCfgNum;
+  *taskNum = tbMetaNum + dbVgNum + udfNum + tbHashNum + qnodeNum + dnodeNum + svrVerNum + dbCfgNum + indexNum + userNum + dbInfoNum + tbIndexNum + tbCfgNum;
   if (*taskNum <= 0) {
-    ctgDebug("Empty input for job, no need to retrieve meta, reqId:0x%" PRIx64, reqId);
+    ctgDebug("Empty input for job, no need to retrieve meta, reqId:0x%" PRIx64, pConn->requestId);
     return TSDB_CODE_SUCCESS;
   }
 
   *job = taosMemoryCalloc(1, sizeof(SCtgJob));
   if (NULL == *job) {
-    ctgError("failed to calloc, size:%d, reqId:0x%" PRIx64, (int32_t)sizeof(SCtgJob), reqId);
+    ctgError("failed to calloc, size:%d, reqId:0x%" PRIx64, (int32_t)sizeof(SCtgJob), pConn->requestId);
     CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
   }
 
   SCtgJob *pJob = *job;
   
-  pJob->queryId = reqId;
+  pJob->queryId = pConn->requestId;
   pJob->userFp = fp;
   pJob->pCtg    = pCtg;
   pJob->conn    = *pConn;
@@ -460,6 +475,7 @@ int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgJob** job, uint6
   pJob->dbInfoNum = dbInfoNum;
   pJob->tbIndexNum = tbIndexNum;
   pJob->tbCfgNum = tbCfgNum;
+  pJob->svrVerNum = svrVerNum;
 
   pJob->pTasks = taosArrayInit(*taskNum, sizeof(SCtgTask));
 
@@ -530,6 +546,10 @@ int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgJob** job, uint6
     CTG_ERR_JRET(ctgInitTask(pJob, CTG_TASK_GET_DNODE, NULL, NULL));
   }
 
+  if (svrVerNum) {
+    CTG_ERR_JRET(ctgInitTask(pJob, CTG_TASK_GET_SVR_VER, NULL, NULL));
+  }
+
   pJob->refId = taosAddRef(gCtgMgmt.jobPool, pJob);
   if (pJob->refId < 0) {
     ctgError("add job to ref failed, error: %s", tstrerror(terrno));
@@ -728,6 +748,21 @@ int32_t ctgDumpUserRes(SCtgTask* pTask) {
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t ctgDumpSvrVer(SCtgTask* pTask) {
+  SCtgJob* pJob = pTask->pJob;
+  if (NULL == pJob->jobRes.pSvrVer) {
+    pJob->jobRes.pSvrVer = taosMemoryCalloc(1, sizeof(SMetaRes));
+    if (NULL == pJob->jobRes.pSvrVer) {
+      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+    }
+  }
+
+  pJob->jobRes.pSvrVer->code = pTask->code;
+  pJob->jobRes.pSvrVer->pRes = pTask->res;
+  
+  return TSDB_CODE_SUCCESS;
+}
+
 int32_t ctgInvokeSubCb(SCtgTask *pTask) {
   int32_t code = 0;
   
@@ -1156,6 +1191,20 @@ _return:
   CTG_RET(code);
 }
 
+int32_t ctgHandleGetSvrVerRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *pMsg, int32_t rspCode) {
+  int32_t code = 0;
+
+  CTG_ERR_JRET(ctgProcessRspMsg(&pTask->msgCtx.out, reqType, pMsg->pData, pMsg->len, rspCode, pTask->msgCtx.target));
+
+  TSWAP(pTask->res, pTask->msgCtx.out);
+  
+_return:
+
+  ctgHandleTaskEnd(pTask, code);
+
+  CTG_RET(code);
+}
+
 int32_t ctgAsyncRefreshTbMeta(SCtgTask *pTask) {
   SCatalog* pCtg = pTask->pJob->pCtg; 
   SRequestConnInfo* pConn = &pTask->pJob->conn;
@@ -1459,6 +1508,15 @@ int32_t ctgLaunchGetUserTask(SCtgTask *pTask) {
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t ctgLaunchGetSvrVerTask(SCtgTask *pTask) {
+  SCatalog* pCtg = pTask->pJob->pCtg; 
+  SRequestConnInfo* pConn = &pTask->pJob->conn;
+
+  CTG_ERR_RET(ctgGetSvrVerFromMnode(pCtg, pConn, NULL, pTask));
+
+  return TSDB_CODE_SUCCESS;
+}
+
 int32_t ctgRelaunchGetTbMetaTask(SCtgTask *pTask) {
   ctgResetTbMetaTask(pTask);
 
@@ -1532,6 +1590,7 @@ SCtgAsyncFps gCtgAsyncFps[] = {
   {ctgInitGetIndexTask,   ctgLaunchGetIndexTask,   ctgHandleGetIndexRsp,   ctgDumpIndexRes,   NULL,               NULL},
   {ctgInitGetUdfTask,     ctgLaunchGetUdfTask,     ctgHandleGetUdfRsp,     ctgDumpUdfRes,     NULL,               NULL},
   {ctgInitGetUserTask,    ctgLaunchGetUserTask,    ctgHandleGetUserRsp,    ctgDumpUserRes,    NULL,               NULL},
+  {ctgInitGetSvrVerTask,  ctgLaunchGetSvrVerTask,  ctgHandleGetSvrVerRsp,  ctgDumpSvrVer,     NULL,               NULL},
 };
 
 int32_t ctgMakeAsyncRes(SCtgJob *pJob) {
@@ -1633,7 +1692,7 @@ int32_t ctgLaunchJob(SCtgJob *pJob) {
   for (int32_t i = 0; i < taskNum; ++i) {
     SCtgTask *pTask = taosArrayGet(pJob->pTasks, i);
 
-    qDebug("QID:0x%" PRIx64 " ctg start to launch task %d", pJob->queryId, pTask->taskId);
+    qDebug("QID:0x%" PRIx64 " ctg launch [%dth] task", pJob->queryId, pTask->taskId);
     CTG_ERR_RET((*gCtgAsyncFps[pTask->type].launchFp)(pTask));
     pTask->status = CTG_TASK_LAUNCHED;
   }

source/libs/catalog/src/ctgDbg.c

@@ -210,7 +210,7 @@ int32_t ctgdLaunchAsyncCall(SCatalog* pCtg, SRequestConnInfo* pConn, uint64_t re
   
   int64_t jobId = 0;
 
-  CTG_ERR_JRET(catalogAsyncGetAllMeta(pCtg, pConn, reqId, &req, ctgdUserCallback, param, &jobId));
+  CTG_ERR_JRET(catalogAsyncGetAllMeta(pCtg, pConn, &req, ctgdUserCallback, param, &jobId));
 
 _return:
 

source/libs/catalog/src/ctgRemote.c

@@ -217,6 +217,21 @@ int32_t ctgProcessRspMsg(void* out, int32_t reqType, char* msg, int32_t msgSize,
       qDebug("Got stb cfg from mnode, tbFName:%s", target);
       break;
     }    
+    case TDMT_MND_SERVER_VERSION: {
+      if (TSDB_CODE_SUCCESS != rspCode) {
+        qError("error rsp for svr ver from mnode, error:%s", tstrerror(rspCode));
+        CTG_ERR_RET(rspCode);
+      }
+      
+      code = queryProcessMsgRsp[TMSG_INDEX(reqType)](out, msg, msgSize);
+      if (code) {
+        qError("Process svr ver rsp failed, error:%s", tstrerror(code));
+        CTG_ERR_RET(code);
+      }
+      
+      qDebug("Got svr ver from mnode");
+      break;
+    }
     default:
       qError("invalid req type %s", TMSG_INFO(reqType));
       return TSDB_CODE_APP_ERROR;
@@ -811,4 +826,38 @@ int32_t ctgGetTableCfgFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const S
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t ctgGetSvrVerFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, char **out, SCtgTask* pTask) {
+  char *msg = NULL;
+  int32_t msgLen = 0;
+  int32_t reqType = TDMT_MND_SERVER_VERSION;
+  void*(*mallocFp)(int32_t) = pTask ? taosMemoryMalloc : rpcMallocCont;
+
+  qDebug("try to get svr ver from mnode");
+
+  int32_t code = queryBuildMsg[TMSG_INDEX(reqType)](NULL, &msg, 0, &msgLen, mallocFp);
+  if (code) {
+    ctgError("Build get svr ver msg failed, code:%s", tstrerror(code));
+    CTG_ERR_RET(code);
+  }
+
+  if (pTask) {
+    CTG_ERR_RET(ctgUpdateMsgCtx(&pTask->msgCtx, reqType, NULL, NULL));
+    
+    CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask, reqType, msg, msgLen));
+  }
+  
+  SRpcMsg rpcMsg = {
+      .msgType = reqType,
+      .pCont   = msg,
+      .contLen = msgLen,
+  };
+
+  SRpcMsg rpcRsp = {0};
+  rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp);
+
+  CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, NULL));
+  
+  return TSDB_CODE_SUCCESS;
+}
+
 

source/libs/catalog/src/ctgUtil.c

@@ -45,6 +45,8 @@ char *ctgTaskTypeStr(CTG_TASK_TYPE type) {
       return "[get udf]";
     case CTG_TASK_GET_USER:
       return "[get user]";
+    case CTG_TASK_GET_SVR_VER:
+      return "[get svr ver]";
     default:
       return "unknown";
   }
@@ -107,8 +109,13 @@ void ctgFreeSMetaData(SMetaData* pData) {
   taosArrayDestroy(pData->pQnodeList);
   pData->pQnodeList = NULL;
 
+  taosArrayDestroy(pData->pDnodeList);
+  pData->pDnodeList = NULL;
+
   taosArrayDestroy(pData->pTableCfg);
   pData->pTableCfg = NULL;
+
+  taosMemoryFreeClear(pData->pSvrVer);
 }
 
 void ctgFreeSCtgUserAuth(SCtgUserAuth *userCache) {
@@ -371,20 +378,8 @@ void ctgResetTbMetaTask(SCtgTask* pTask) {
 
 void ctgFreeTaskRes(CTG_TASK_TYPE type, void **pRes) {
   switch (type) {
-    case CTG_TASK_GET_QNODE: {
-      taosArrayDestroy((SArray*)*pRes);
-      *pRes = NULL;
-      break;
-    }
-    case CTG_TASK_GET_DNODE: {
-      taosArrayDestroy((SArray*)*pRes);
-      *pRes = NULL;
-      break;
-    }
-    case CTG_TASK_GET_TB_META: {
-      taosMemoryFreeClear(*pRes);
-      break;
-    }
+    case CTG_TASK_GET_QNODE:
+    case CTG_TASK_GET_DNODE:
     case CTG_TASK_GET_DB_VGROUP: {
       taosArrayDestroy((SArray*)*pRes);
       *pRes = NULL;
@@ -398,14 +393,6 @@ void ctgFreeTaskRes(CTG_TASK_TYPE type, void **pRes) {
       }
       break;
     }
-    case CTG_TASK_GET_DB_INFO: {
-      taosMemoryFreeClear(*pRes);
-      break;
-    }
-    case CTG_TASK_GET_TB_HASH: {
-      taosMemoryFreeClear(*pRes);
-      break;
-    }
     case CTG_TASK_GET_TB_INDEX: {
       taosArrayDestroyEx(*pRes, tFreeSTableIndexInfo);
       *pRes = NULL;
@@ -419,15 +406,13 @@ void ctgFreeTaskRes(CTG_TASK_TYPE type, void **pRes) {
       }
       break;
     }
-    case CTG_TASK_GET_INDEX: {
-      taosMemoryFreeClear(*pRes);
-      break;
-    }
-    case CTG_TASK_GET_UDF: {
-      taosMemoryFreeClear(*pRes);
-      break;
-    }
-    case CTG_TASK_GET_USER: {
+    case CTG_TASK_GET_TB_HASH:
+    case CTG_TASK_GET_DB_INFO:
+    case CTG_TASK_GET_INDEX:
+    case CTG_TASK_GET_UDF: 
+    case CTG_TASK_GET_USER: 
+    case CTG_TASK_GET_SVR_VER:
+    case CTG_TASK_GET_TB_META: {
       taosMemoryFreeClear(*pRes);
       break;
     }
@@ -440,20 +425,12 @@ void ctgFreeTaskRes(CTG_TASK_TYPE type, void **pRes) {
 
 void ctgFreeSubTaskRes(CTG_TASK_TYPE type, void **pRes) {
   switch (type) {
-    case CTG_TASK_GET_QNODE: {
-      taosArrayDestroy((SArray*)*pRes);
-      *pRes = NULL;
-      break;
-    }
+    case CTG_TASK_GET_QNODE:
     case CTG_TASK_GET_DNODE: {
       taosArrayDestroy((SArray*)*pRes);
       *pRes = NULL;
       break;
     }
-    case CTG_TASK_GET_TB_META: {
-      taosMemoryFreeClear(*pRes);
-      break;
-    }
     case CTG_TASK_GET_DB_VGROUP: {
       if (*pRes) {
         SDBVgInfo* pInfo = (SDBVgInfo*)*pRes;
@@ -470,14 +447,6 @@ void ctgFreeSubTaskRes(CTG_TASK_TYPE type, void **pRes) {
       }
       break;
     }
-    case CTG_TASK_GET_DB_INFO: {
-      taosMemoryFreeClear(*pRes);
-      break;
-    }
-    case CTG_TASK_GET_TB_HASH: {
-      taosMemoryFreeClear(*pRes);
-      break;
-    }
     case CTG_TASK_GET_TB_INDEX: {
       taosArrayDestroyEx(*pRes, tFreeSTableIndexInfo);
       *pRes = NULL;
@@ -491,14 +460,12 @@ void ctgFreeSubTaskRes(CTG_TASK_TYPE type, void **pRes) {
       }
       break;
     }
-    case CTG_TASK_GET_INDEX: {
-      taosMemoryFreeClear(*pRes);
-      break;
-    }
-    case CTG_TASK_GET_UDF: {
-      taosMemoryFreeClear(*pRes);
-      break;
-    }
+    case CTG_TASK_GET_TB_META:    
+    case CTG_TASK_GET_DB_INFO:
+    case CTG_TASK_GET_TB_HASH:
+    case CTG_TASK_GET_INDEX: 
+    case CTG_TASK_GET_UDF: 
+    case CTG_TASK_GET_SVR_VER: 
     case CTG_TASK_GET_USER: {
       taosMemoryFreeClear(*pRes);
       break;
@@ -522,10 +489,6 @@ void ctgClearSubTaskRes(SCtgSubRes *pRes) {
 
 void ctgFreeTaskCtx(SCtgTask* pTask) {
   switch (pTask->type) {
-    case CTG_TASK_GET_QNODE: {
-      taosMemoryFreeClear(pTask->taskCtx);      
-      break;
-    }
     case CTG_TASK_GET_TB_META: {
       SCtgTbMetaCtx* taskCtx = (SCtgTbMetaCtx*)pTask->taskCtx;
       taosMemoryFreeClear(taskCtx->pName);
@@ -536,18 +499,6 @@ void ctgFreeTaskCtx(SCtgTask* pTask) {
       taosMemoryFreeClear(pTask->taskCtx);
       break;
     }
-    case CTG_TASK_GET_DB_VGROUP: {
-      taosMemoryFreeClear(pTask->taskCtx);      
-      break;
-    }
-    case CTG_TASK_GET_DB_CFG: {
-      taosMemoryFreeClear(pTask->taskCtx);
-      break;
-    }
-    case CTG_TASK_GET_DB_INFO: {
-      taosMemoryFreeClear(pTask->taskCtx);      
-      break;
-    }
     case CTG_TASK_GET_TB_HASH: {
       SCtgTbHashCtx* taskCtx = (SCtgTbHashCtx*)pTask->taskCtx;
       taosMemoryFreeClear(taskCtx->pName);
@@ -567,14 +518,12 @@ void ctgFreeTaskCtx(SCtgTask* pTask) {
       taosMemoryFreeClear(pTask->taskCtx);
       break;
     }
-    case CTG_TASK_GET_INDEX: {
-      taosMemoryFreeClear(pTask->taskCtx);
-      break;
-    }
-    case CTG_TASK_GET_UDF: {
-      taosMemoryFreeClear(pTask->taskCtx);
-      break;
-    }
+    case CTG_TASK_GET_DB_VGROUP:
+    case CTG_TASK_GET_DB_CFG:
+    case CTG_TASK_GET_DB_INFO:    
+    case CTG_TASK_GET_INDEX:
+    case CTG_TASK_GET_UDF:
+    case CTG_TASK_GET_QNODE:    
     case CTG_TASK_GET_USER: {
       taosMemoryFreeClear(pTask->taskCtx);
       break;

source/libs/executor/src/executor.c

@@ -40,15 +40,8 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
     SStreamBlockScanInfo* pInfo = pOperator->info;
     pInfo->assignBlockUid = assignUid;
 
-    // no need to check
-#if 0
-    if (pInfo->blockType == 0) {
-      pInfo->blockType = type;
-    } else if (pInfo->blockType != type) {
-      ASSERT(0);
-      return TSDB_CODE_QRY_APP_ERROR;
-    }
-#endif
+    // TODO: if a block was set but not consumed,
+    // prevent setting a different type of block
     pInfo->blockType = type;
 
     if (type == STREAM_DATA_TYPE_SUBMIT_BLOCK) {

source/libs/executor/src/executorimpl.c

@@ -1321,7 +1321,7 @@ void doFilter(const SNode* pFilterNode, SSDataBlock* pBlock) {
   // todo move to the initialization function
   int32_t code = filterInitFromNode((SNode*)pFilterNode, &filter, 0);
 
-  size_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
+  size_t             numOfCols = taosArrayGetSize(pBlock->pDataBlock);
   SFilterColumnParam param1 = {.numOfCols = numOfCols, .pDataBlock = pBlock->pDataBlock};
   code = filterSetDataFromSlotId(filter, &param1);
 
@@ -2048,7 +2048,7 @@ int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLo
                                      SArray* pColList) {
   if (pColList == NULL) {  // data from other sources
     blockDataCleanup(pRes);
-//    blockDataEnsureCapacity(pRes, numOfRows);
+    //    blockDataEnsureCapacity(pRes, numOfRows);
     blockCompressDecode(pRes, numOfOutput, numOfRows, pData);
   } else {  // extract data according to pColList
     ASSERT(numOfOutput == taosArrayGetSize(pColList));
@@ -2402,14 +2402,14 @@ SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode
 
   tsem_init(&pInfo->ready, 0, 0);
 
-  pInfo->seqLoadData  = false;
+  pInfo->seqLoadData = false;
   pInfo->pTransporter = pTransporter;
-  pInfo->pResult      = createResDataBlock(pExNode->node.pOutputDataBlockDesc);
-  pOperator->name     = "ExchangeOperator";
+  pInfo->pResult = createResDataBlock(pExNode->node.pOutputDataBlockDesc);
+  pOperator->name = "ExchangeOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_EXCHANGE;
   pOperator->blocking = false;
-  pOperator->status   = OP_NOT_OPENED;
-  pOperator->info     = pInfo;
+  pOperator->status = OP_NOT_OPENED;
+  pOperator->info = pInfo;
   pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pResult->pDataBlock);
   pOperator->pTaskInfo = pTaskInfo;
 
@@ -4365,6 +4365,62 @@ _error:
   return NULL;
 }
 
+int32_t extractTableScanNode(SPhysiNode* pNode, STableScanPhysiNode** ppNode) {
+  if (pNode->pChildren == NULL || LIST_LENGTH(pNode->pChildren) == 0) {
+    if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == pNode->type) {
+      *ppNode = (STableScanPhysiNode*)pNode;
+      return 0;
+    } else {
+      ASSERT(0);
+      terrno = TSDB_CODE_QRY_APP_ERROR;
+      return -1;
+    }
+  } else {
+    if (LIST_LENGTH(pNode->pChildren) != 1) {
+      ASSERT(0);
+      terrno = TSDB_CODE_QRY_APP_ERROR;
+      return -1;
+    }
+    SPhysiNode* pChildNode = (SPhysiNode*)nodesListGetNode(pNode->pChildren, 0);
+    return extractTableScanNode(pChildNode, ppNode);
+  }
+  return -1;
+}
+
+int32_t doRebuildReader(SOperatorInfo* pOperator, SSubplan* plan, SReadHandle* pHandle) {
+  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
+    if (pOperator->numOfDownstream == 0) {
+      qError("failed to find stream scan operator");
+      return TSDB_CODE_QRY_APP_ERROR;
+    }
+
+    if (pOperator->numOfDownstream > 1) {
+      qError("join not supported for stream block scan");
+      return TSDB_CODE_QRY_APP_ERROR;
+    }
+    return doRebuildReader(pOperator->pDownstream[0], plan, pHandle);
+  } else {
+    SStreamBlockScanInfo* pInfo = pOperator->info;
+    ASSERT(pInfo->pSnapshotReadOp->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN);
+    STableScanInfo* pTableScanInfo = pInfo->pSnapshotReadOp->info;
+
+    tsdbCleanupReadHandle(pTableScanInfo->dataReader);
+    STableScanPhysiNode* pNode = NULL;
+    if (extractTableScanNode(plan->pNode, &pNode) < 0) {
+      ASSERT(0);
+    }
+
+    STableListInfo info = {0};
+    pTableScanInfo->dataReader = doCreateDataReader(pNode, pHandle, &info, 0, 0);
+    if (pTableScanInfo->dataReader == NULL) {
+      ASSERT(0);
+      qError("failed to create data reader");
+      return TSDB_CODE_QRY_APP_ERROR;
+    }
+  }
+  return 0;
+}
+
 int32_t encodeOperator(SOperatorInfo* ops, char** result, int32_t* length) {
   int32_t code = TDB_CODE_SUCCESS;
   char*   pCurrent = NULL;

source/libs/parser/src/parInsertData.c

@@ -26,6 +26,7 @@
 typedef struct SBlockKeyTuple {
   TSKEY skey;
   void* payloadAddr;
+  int16_t index;
 } SBlockKeyTuple;
 
 typedef struct SBlockKeyInfo {
@@ -36,7 +37,6 @@ typedef struct SBlockKeyInfo {
 static int32_t rowDataCompar(const void* lhs, const void* rhs) {
   TSKEY left = *(TSKEY*)lhs;
   TSKEY right = *(TSKEY*)rhs;
-
   if (left == right) {
     return 0;
   } else {
@@ -44,6 +44,16 @@ static int32_t rowDataCompar(const void* lhs, const void* rhs) {
   }
 }
 
+static int32_t rowDataComparStable(const void* lhs, const void* rhs) {
+  TSKEY left = *(TSKEY*)lhs;
+  TSKEY right = *(TSKEY*)rhs;
+  if (left == right) {
+    return ((SBlockKeyTuple*)lhs)->index - ((SBlockKeyTuple*)rhs)->index;
+  } else {
+    return left > right ? 1 : -1;
+  }
+}
+
 void setBoundColumnInfo(SParsedDataColInfo* pColList, SSchema* pSchema, col_id_t numOfCols) {
   pColList->numOfCols = numOfCols;
   pColList->numOfBound = numOfCols;
@@ -343,6 +353,7 @@ int sortRemoveDataBlockDupRows(STableDataBlocks* dataBuf, SBlockKeyInfo* pBlkKey
   while (n < nRows) {
     pBlkKeyTuple->skey = TD_ROW_KEY((STSRow*)pBlockData);
     pBlkKeyTuple->payloadAddr = pBlockData;
+    pBlkKeyTuple->index = n;
 
     // next loop
     pBlockData += extendedRowSize;
@@ -354,7 +365,7 @@ int sortRemoveDataBlockDupRows(STableDataBlocks* dataBuf, SBlockKeyInfo* pBlkKey
     pBlkKeyTuple = pBlkKeyInfo->pKeyTuple;
 
     // todo. qsort is unstable, if timestamp is same, should get the last one
-    qsort(pBlkKeyTuple, nRows, sizeof(SBlockKeyTuple), rowDataCompar);
+    qsort(pBlkKeyTuple, nRows, sizeof(SBlockKeyTuple), rowDataComparStable);
 
     pBlkKeyTuple = pBlkKeyInfo->pKeyTuple;
     int32_t i = 0;

source/libs/parser/src/parTranslater.c

@@ -4641,7 +4641,7 @@ static int32_t extractShowCreateTableResultSchema(int32_t* numOfCols, SSchema**
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t extractShowLocalVariablesResultSchema(int32_t* numOfCols, SSchema** pSchema) {
+static int32_t extractShowVariablesResultSchema(int32_t* numOfCols, SSchema** pSchema) {
   *numOfCols = 2;
   *pSchema = taosMemoryCalloc((*numOfCols), sizeof(SSchema));
   if (NULL == (*pSchema)) {
@@ -4678,7 +4678,8 @@ int32_t extractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pS
     case QUERY_NODE_SHOW_CREATE_STABLE_STMT:
       return extractShowCreateTableResultSchema(numOfCols, pSchema);
     case QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT:
-      return extractShowLocalVariablesResultSchema(numOfCols, pSchema);
+    case QUERY_NODE_SHOW_VARIABLES_STMT:
+      return extractShowVariablesResultSchema(numOfCols, pSchema);
     default:
       break;
   }
@@ -5909,7 +5910,6 @@ static int32_t rewriteQuery(STranslateContext* pCxt, SQuery* pQuery) {
     case QUERY_NODE_SHOW_CLUSTER_STMT:
     case QUERY_NODE_SHOW_TOPICS_STMT:
     case QUERY_NODE_SHOW_TRANSACTIONS_STMT:
-    case QUERY_NODE_SHOW_VARIABLES_STMT:
     case QUERY_NODE_SHOW_APPS_STMT:
     case QUERY_NODE_SHOW_CONSUMERS_STMT:
     case QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT:
@@ -6011,6 +6011,14 @@ static int32_t setQuery(STranslateContext* pCxt, SQuery* pQuery) {
     case QUERY_NODE_ALTER_LOCAL_STMT:
       pQuery->execMode = QUERY_EXEC_MODE_LOCAL;
       break;
+    case QUERY_NODE_SHOW_VARIABLES_STMT:
+      pQuery->haveResultSet = true;
+      pQuery->execMode = QUERY_EXEC_MODE_RPC;
+      if (NULL != pCxt->pCmdMsg) {
+        TSWAP(pQuery->pCmdMsg, pCxt->pCmdMsg);
+        pQuery->msgType = pQuery->pCmdMsg->msgType;
+      }
+      break;      
     default:
       pQuery->execMode = QUERY_EXEC_MODE_RPC;
       if (NULL != pCxt->pCmdMsg) {

source/libs/parser/test/parTestUtil.h

@@ -54,12 +54,6 @@ class ParserDdlTest : public ParserTestBase {
   virtual void checkDdl(const SQuery* pQuery, ParserStage stage) {
     ASSERT_NE(pQuery, nullptr);
     ASSERT_NE(pQuery->pRoot, nullptr);
-    if (QUERY_EXEC_MODE_RPC == pQuery->execMode) {
-      ASSERT_EQ(pQuery->haveResultSet, false);
-      ASSERT_EQ(pQuery->numOfResCols, 0);
-      ASSERT_EQ(pQuery->pResSchema, nullptr);
-      ASSERT_EQ(pQuery->precision, 0);
-    }
     if (nullptr != checkDdl_) {
       checkDdl_(pQuery, stage);
     }

source/libs/planner/test/CMakeLists.txt

@@ -32,9 +32,7 @@ if(${BUILD_WINGETOPT})
     target_link_libraries(plannerTest PUBLIC wingetopt)
 endif()
 
-# if(NOT TD_WINDOWS)
-    add_test(
-        NAME plannerTest
-        COMMAND plannerTest
-    )
-# endif(NOT TD_WINDOWS)
+add_test(
+    NAME plannerTest
+    COMMAND plannerTest
+)

source/libs/qcom/src/querymsg.c

@@ -144,6 +144,23 @@ int32_t queryBuildDnodeListMsg(void *input, char **msg, int32_t msgSize, int32_t
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t queryBuildGetSerVerMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen, void*(*mallcFp)(int32_t)) {
+  if (NULL == msg || NULL == msgLen) {
+    return TSDB_CODE_TSC_INVALID_INPUT;
+  }
+
+  SServerVerReq req = {0};
+
+  int32_t bufLen = tSerializeSServerVerReq(NULL, 0, &req);
+  void   *pBuf = (*mallcFp)(bufLen);
+  tSerializeSServerVerReq(pBuf, bufLen, &req);
+
+  *msg = pBuf;
+  *msgLen = bufLen;
+
+  return TSDB_CODE_SUCCESS;
+}
+
 
 int32_t queryBuildGetDBCfgMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen, void*(*mallcFp)(int32_t)) {
   if (NULL == msg || NULL == msgLen) {
@@ -467,6 +484,26 @@ int32_t queryProcessDnodeListRsp(void *output, char *msg, int32_t msgSize) {
   return code;
 }
 
+int32_t queryProcessGetSerVerRsp(void *output, char *msg, int32_t msgSize) {
+  SServerVerRsp out = {0};
+  int32_t       code = 0;
+
+  if (NULL == output || NULL == msg || msgSize <= 0) {
+    code = TSDB_CODE_TSC_INVALID_INPUT;
+    return code;
+  }
+
+  if (tDeserializeSServerVerRsp(msg, msgSize, &out) != 0) {
+    qError("invalid svr ver rsp msg, msgSize:%d", msgSize);
+    code = TSDB_CODE_INVALID_MSG;
+    return code;
+  }
+
+  *(char**)output = strdup(out.ver);
+
+  return code;
+}
+
 
 int32_t queryProcessGetDbCfgRsp(void *output, char *msg, int32_t msgSize) {
   SDbCfgRsp out = {0};
@@ -583,6 +620,7 @@ void initQueryModuleMsgHandle() {
   queryBuildMsg[TMSG_INDEX(TDMT_MND_GET_TABLE_INDEX)]  = queryBuildGetTbIndexMsg;
   queryBuildMsg[TMSG_INDEX(TDMT_VND_TABLE_CFG)]        = queryBuildGetTbCfgMsg;
   queryBuildMsg[TMSG_INDEX(TDMT_MND_TABLE_CFG)]        = queryBuildGetTbCfgMsg;
+  queryBuildMsg[TMSG_INDEX(TDMT_MND_SERVER_VERSION)]   = queryBuildGetSerVerMsg;
 
   queryProcessMsgRsp[TMSG_INDEX(TDMT_VND_TABLE_META)]      = queryProcessTableMetaRsp;
   queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_TABLE_META)]      = queryProcessTableMetaRsp;
@@ -596,6 +634,7 @@ void initQueryModuleMsgHandle() {
   queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_GET_TABLE_INDEX)] = queryProcessGetTbIndexRsp;
   queryProcessMsgRsp[TMSG_INDEX(TDMT_VND_TABLE_CFG)]       = queryProcessGetTbCfgRsp;
   queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_TABLE_CFG)]       = queryProcessGetTbCfgRsp;
+  queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_SERVER_VERSION)]  = queryProcessGetSerVerRsp;
 }
 
 #pragma GCC diagnostic pop

source/libs/transport/inc/transComm.h

@@ -125,6 +125,7 @@ typedef struct {
   STransMsg* pRsp;    // for synchronous API
   tsem_t*    pSem;    // for synchronous API
   SCvtAddr   cvtAddr;
+  bool       setMaxRetry;
 
   int hThrdIdx;
 } STransConnCtx;

source/libs/transport/src/transCli.c

@@ -980,7 +980,7 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
     tTrace("try to send req to next node");
     pMsg->st = taosGetTimestampUs();
     pCtx->retryCount += 1;
-    if (pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL) {
+    if (pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL && pCtx->setMaxRetry == false) {
       if (pCtx->retryCount < pEpSet->numOfEps * 3) {
         pEpSet->inUse = (++pEpSet->inUse) % pEpSet->numOfEps;
         if (pThrd->quit == false) {
@@ -997,6 +997,7 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
         }
       }
     } else if (pCtx->retryCount < TRANS_RETRY_COUNT_LIMIT) {
+      pCtx->setMaxRetry = true;
       if (pResp->contLen == 0) {
         pEpSet->inUse = (++pEpSet->inUse) % pEpSet->numOfEps;
         transPrintEpSet(&pCtx->epSet);
@@ -1012,8 +1013,10 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
                pCtx->retryCount + 1, TRANS_RETRY_COUNT_LIMIT);
       }
       if (pThrd->quit == false) {
-        if (pConn->status != ConnInPool) {
-          addConnToPool(pThrd->pool, pConn);
+        if (pResp->code != TSDB_CODE_RPC_NETWORK_UNAVAIL) {
+          if (pConn->status != ConnInPool) addConnToPool(pThrd->pool, pConn);
+        } else {
+          transUnrefCliHandle(pConn);
         }
         STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg));
         arg->param1 = pMsg;

tests/pytest/util/sqlset.py

@@ -15,7 +15,7 @@ from util.sql import tdSql
 
 class TDSetSql:
     def init(self, conn, logSql):
-        tdSql.init(conn.cursor(), logSql)
+        
         self.stbname = 'stb'
 
     def set_create_normaltable_sql(self, ntbname='ntb', 

tests/script/sh/cfg.bat

@@ -16,12 +16,12 @@ rem echo NODE:       %NODE%
 set SCRIPT_DIR=%~dp0..\
 rem echo SCRIPT_DIR: %SCRIPT_DIR%
 
-set BUILD_DIR=%SCRIPT_DIR%..\..\..\debug\build\bin\
+echo %cd% | grep community > nul && set "BUILD_DIR=%SCRIPT_DIR%..\..\..\debug\build\bin\" || set "BUILD_DIR=%SCRIPT_DIR%..\..\debug\build\bin\"
 set TSIM=%BUILD_DIR%tsim
 rem echo BUILD_DIR:  %BUILD_DIR%
 rem echo TSIM:       %TSIM%
 
-set SIM_DIR=%SCRIPT_DIR%..\..\..\sim\
+echo %cd% | grep community > nul && set "SIM_DIR=%SCRIPT_DIR%..\..\..\sim\" || set "SIM_DIR=%SCRIPT_DIR%..\..\sim\"
 rem echo SIM_DIR:    %SIM_DIR%
 
 set NODE_DIR=%SIM_DIR%%NODE_NAME%\

tests/script/sh/copy_udf.bat

@@ -4,7 +4,7 @@ echo Executing copy_udf.bat
 set SCRIPT_DIR=%cd%
 echo SCRIPT_DIR: %SCRIPT_DIR%
 
-cd ..\..\..
+echo %cd% | grep community > nul && cd ..\..\.. || cd ..\..
 set TAOS_DIR=%cd%
 echo find udf library in %TAOS_DIR%
 set UDF1_DIR=%TAOS_DIR%\debug\build\lib\udf1.dll

tests/script/sh/deploy.bat

@@ -13,12 +13,12 @@ rem echo NODE:       %NODE%
 set SCRIPT_DIR=%~dp0..\
 rem echo SCRIPT_DIR: %SCRIPT_DIR%
 
-set BUILD_DIR=%SCRIPT_DIR%..\..\..\debug\build\bin\
+echo %cd% | grep community > nul && set "BUILD_DIR=%SCRIPT_DIR%..\..\..\debug\build\bin\" || set "BUILD_DIR=%SCRIPT_DIR%..\..\debug\build\bin\"
 set TSIM=%BUILD_DIR%tsim
 rem echo BUILD_DIR:  %BUILD_DIR%
 rem echo TSIM:       %TSIM%
 
-set SIM_DIR=%SCRIPT_DIR%..\..\..\sim\
+echo %cd% | grep community > nul && set "SIM_DIR=%SCRIPT_DIR%..\..\..\sim\" || set "SIM_DIR=%SCRIPT_DIR%..\..\sim\"
 rem echo SIM_DIR:    %SIM_DIR%
 
 set NODE_DIR=%SIM_DIR%%NODE_NAME%\

tests/script/sh/exec.bat

@@ -13,12 +13,12 @@ rem echo NODE:       %EXEC_OPTON%
 set SCRIPT_DIR=%~dp0..\
 rem echo SCRIPT_DIR: %SCRIPT_DIR%
 
-set BUILD_DIR=%SCRIPT_DIR%..\..\..\debug\build\bin\
+echo %cd% | grep community > nul && set "BUILD_DIR=%SCRIPT_DIR%..\..\..\debug\build\bin\" || set "BUILD_DIR=%SCRIPT_DIR%..\..\debug\build\bin\"
 set TAOSD=%BUILD_DIR%taosd
 rem echo BUILD_DIR:  %BUILD_DIR%
 rem echo TAOSD:      %TAOSD%
 
-set SIM_DIR=%SCRIPT_DIR%..\..\..\sim\
+echo %cd% | grep community > nul && set "SIM_DIR=%SCRIPT_DIR%..\..\..\sim\" || set "SIM_DIR=%SCRIPT_DIR%..\..\sim\"
 rem echo SIM_DIR:    %SIM_DIR%
 
 set NODE_DIR=%SIM_DIR%%NODE_NAME%\

tests/script/tsim/dnode/redistribute_vgroup_replica3_v1_follower.sim

@@ -112,7 +112,7 @@ if $data(5)[4] != ready then
   goto step2
 endi
 
-print =============== step31: move follower
+print =============== step3: create database
 $leaderExist = 0
 $leaderVnode = 0
 $follower1 = 0
@@ -165,7 +165,7 @@ if $rows != 1 then
   return -1
 endi
 
-print =============== step32: move follower2
+print =============== step31: 
 print redistribute vgroup 2 dnode $leaderVnode dnode $follower2 dnode 5 
 sql redistribute vgroup 2 dnode $leaderVnode dnode $follower2 dnode 5 
 sql show d1.vgroups
@@ -178,7 +178,7 @@ endi
 
 return
 
-print =============== step33: move follower1
+print =============== step32: 
 print redistribute vgroup 2 dnode $leaderVnode dnode $follower1 dnode 5 
 sql redistribute vgroup 2 dnode $leaderVnode dnode $follower1 dnode 5 
 sql show d1.vgroups
@@ -189,7 +189,7 @@ if $rows != 1 then
   return -1
 endi
 
-print =============== step34: move follower2
+print =============== step33: 
 print redistribute vgroup 2 dnode $leaderVnode dnode 5 dnode $follower2
 sql redistribute vgroup 2 dnode $leaderVnode dnode 5 dnode $follower2
 sql show d1.vgroups
@@ -200,7 +200,7 @@ if $rows != 1 then
   return -1
 endi
 
-print =============== step35: move follower1
+print =============== step34: 
 print redistribute vgroup 2 dnode $leaderVnode dnode 5 dnode $follower1
 sql redistribute vgroup 2 dnode $leaderVnode dnode 5 dnode $follower1
 sql show d1.vgroups
@@ -211,9 +211,9 @@ if $rows != 1 then
   return -1
 endi
 
-print =============== step36: move follower2
-print redistribute vgroup 2 dnode $leaderVnode dnode $follower2 dnode 5 
-sql redistribute vgroup 2 dnode $leaderVnode dnode $follower2 dnode 5 
+print =============== step34: 
+print redistribute vgroup 2 dnode $leaderVnode dnode $follower2 dnode $follower1
+sql redistribute vgroup 2 dnode $leaderVnode dnode $follower2 dnode $follower1
 sql show d1.vgroups
 print ===> $data00 $data01 $data02 $data03 $data04 $data05 $data06 $data07 $data08 $data09
 
@@ -222,7 +222,7 @@ if $rows != 1 then
   return -1
 endi
 
-print =============== step37: move follower1
+print =============== step36: 
 print redistribute vgroup 2 dnode $leaderVnode dnode $follower1 dnode 5 
 sql redistribute vgroup 2 dnode $leaderVnode dnode $follower1 dnode 5 
 sql show d1.vgroups
@@ -233,7 +233,7 @@ if $rows != 1 then
   return -1
 endi
 
-print =============== step38: move follower2
+print =============== step37: 
 print redistribute vgroup 2 dnode $leaderVnode dnode 5 dnode $follower2
 sql redistribute vgroup 2 dnode $leaderVnode dnode 5 dnode $follower2
 sql show d1.vgroups
@@ -244,7 +244,7 @@ if $rows != 1 then
   return -1
 endi
 
-print =============== step39: move follower1
+print =============== step38: 
 print redistribute vgroup 2 dnode $leaderVnode dnode 5 dnode $follower1
 sql redistribute vgroup 2 dnode $leaderVnode dnode 5 dnode $follower1
 sql show d1.vgroups
@@ -255,6 +255,17 @@ if $rows != 1 then
   return -1
 endi
 
+print =============== step39: 
+print redistribute vgroup 2 dnode $leaderVnode dnode $follower1 dnode $follower2
+sql redistribute vgroup 2 dnode $leaderVnode dnode $follower1 dnode $follower2
+sql show d1.vgroups
+print ===> $data00 $data01 $data02 $data03 $data04 $data05 $data06 $data07 $data08 $data09
+
+sql show d1.tables
+if $rows != 1 then
+  return -1
+endi
+
 system sh/exec.sh -n dnode1 -s stop -x SIGINT
 system sh/exec.sh -n dnode2 -s stop -x SIGINT
 system sh/exec.sh -n dnode3 -s stop -x SIGINT

tests/script/tsim/dnode/redistribute_vgroup_replica3_v1_leader.sim

@@ -91,7 +91,7 @@ if $data(5)[4] != ready then
   goto step2
 endi
 
-print =============== step31: move follower
+print =============== step3: create database
 $leaderExist = 0
 $leaderVnode = 0
 $follower1 = 0
@@ -144,7 +144,7 @@ if $rows != 1 then
   return -1
 endi
 
-print =============== step32: move leader
+print =============== step40: move leader
 print redistribute vgroup 2 dnode $follower1 dnode $follower2 dnode 5 
 sql redistribute vgroup 2 dnode $follower1 dnode $follower2 dnode 5 
 sql show d1.vgroups
@@ -155,9 +155,7 @@ if $rows != 1 then
   return -1
 endi
 
-return 
-
-print =============== step33: move follower1
+print =============== step41: 
 print redistribute vgroup 2 dnode $leaderVnode dnode $follower2 dnode 5 
 sql redistribute vgroup 2 dnode $leaderVnode dnode $follower2 dnode 5 
 sql show d1.vgroups
@@ -168,7 +166,7 @@ if $rows != 1 then
   return -1
 endi
 
-print =============== step34: move follower2
+print =============== step42: 
 print redistribute vgroup 2 dnode $follower1 dnode 5 dnode $leaderVnode
 sql redistribute vgroup 2 dnode $follower1 dnode 5 dnode $leaderVnode
 sql show d1.vgroups
@@ -179,7 +177,18 @@ if $rows != 1 then
   return -1
 endi
 
-print =============== step35: move leader
+print =============== step43:
+print redistribute vgroup 2 dnode $follower1 dnode $follower2 dnode $leaderVnode
+sql redistribute vgroup 2 dnode $follower1 dnode $follower2 dnode $leaderVnode
+sql show d1.vgroups
+print ===> $data00 $data01 $data02 $data03 $data04 $data05 $data06 $data07 $data08 $data09
+
+sql show d1.tables
+if $rows != 1 then
+  return -1
+endi
+
+print =============== step44:
 print redistribute vgroup 2 dnode $follower2 dnode 5 dnode $follower1
 sql redistribute vgroup 2 dnode $follower2 dnode 5 dnode $follower1
 sql show d1.vgroups
@@ -190,7 +199,7 @@ if $rows != 1 then
   return -1
 endi
 
-print =============== step36: move follower1
+print =============== step45:
 print redistribute vgroup 2 dnode $leaderVnode dnode $follower2 dnode 5 
 sql redistribute vgroup 2 dnode $leaderVnode dnode $follower2 dnode 5 
 sql show d1.vgroups
@@ -201,7 +210,7 @@ if $rows != 1 then
   return -1
 endi
 
-print =============== step37: move follower2
+print =============== step46: 
 print redistribute vgroup 2 dnode $leaderVnode dnode $follower1 dnode 5 
 sql redistribute vgroup 2 dnode $leaderVnode dnode $follower1 dnode 5 
 sql show d1.vgroups
@@ -212,7 +221,7 @@ if $rows != 1 then
   return -1
 endi
 
-print =============== step38: move leader
+print =============== step47: 
 print redistribute vgroup 2 dnode $follower1 dnode 5 dnode $follower2
 sql redistribute vgroup 2 dnode $follower1 dnode 5 dnode $follower2
 sql show d1.vgroups
@@ -223,7 +232,18 @@ if $rows != 1 then
   return -1
 endi
 
-print =============== step39: move follower1
+print =============== step48: 
+print redistribute vgroup 2 dnode $leaderVnode dnode 5 dnode $follower2
+sql redistribute vgroup 2 dnode $leaderVnode dnode 5 dnode $follower2
+sql show d1.vgroups
+print ===> $data00 $data01 $data02 $data03 $data04 $data05 $data06 $data07 $data08 $data09
+
+sql show d1.tables
+if $rows != 1 then
+  return -1
+endi
+
+print =============== step49: 
 print redistribute vgroup 2 dnode $leaderVnode dnode 5 dnode $follower2
 sql redistribute vgroup 2 dnode $leaderVnode dnode 5 dnode $follower2
 sql show d1.vgroups

tests/script/tsim/sync/3Replica1VgElect.sim

@@ -197,22 +197,22 @@ print $switch_loop_cnt
 
 if $switch_loop_cnt == 1 then
   sql show vgroups
-  $dnodeId = $data[1][3]
+  $dnodeId = $data[0][3]
   $dnodeId = dnode . $dnodeId
   goto switch_leader_to_offine_loop
 elif $switch_loop_cnt == 2 then
   sql show vgroups
-  $dnodeId = $data[2][3]
+  $dnodeId = $data[0][3]
   $dnodeId = dnode . $dnodeId
   goto switch_leader_to_offine_loop
 elif $switch_loop_cnt == 3 then
   sql show vgroups
-  $dnodeId = $data[3][3]
+  $dnodeId = $data[0][3]
   $dnodeId = dnode . $dnodeId
   goto switch_leader_to_offine_loop
 elif $switch_loop_cnt == 4 then
   sql show vgroups
-  $dnodeId = $data[4][3]  
+  $dnodeId = $data[0][3]  
   $dnodeId = dnode . $dnodeId
   goto switch_leader_to_offine_loop
 else

tests/script/tsim/sync/3Replica5VgElect.sim

@@ -290,22 +290,22 @@ print $switch_loop_cnt
 
 if $switch_loop_cnt == 1 then
   sql show vgroups
-  $dnodeId = $data[1][3]
+  $dnodeId = $data[0][3]
   $dnodeId = dnode . $dnodeId
   goto switch_leader_to_offine_loop
 elif $switch_loop_cnt == 2 then
   sql show vgroups
-  $dnodeId = $data[2][3]
+  $dnodeId = $data[0][3]
   $dnodeId = dnode . $dnodeId
   goto switch_leader_to_offine_loop
 elif $switch_loop_cnt == 3 then
   sql show vgroups
-  $dnodeId = $data[3][3]
+  $dnodeId = $data[0][3]
   $dnodeId = dnode . $dnodeId
   goto switch_leader_to_offine_loop
 elif $switch_loop_cnt == 4 then
   sql show vgroups
-  $dnodeId = $data[4][3]  
+  $dnodeId = $data[0][3]  
   $dnodeId = dnode . $dnodeId
   goto switch_leader_to_offine_loop
 else

tests/script/tsim/sync/oneReplica5VgElect.sim

@@ -227,22 +227,22 @@ print $switch_loop_cnt
 
 if $switch_loop_cnt == 1 then
   sql show vgroups
-  $dnodeId = $data[1][3]
+  $dnodeId = $data[0][3]
   $dnodeId = dnode . $dnodeId
   goto switch_leader_to_offine_loop
 elif $switch_loop_cnt == 2 then
   sql show vgroups
-  $dnodeId = $data[2][3]
+  $dnodeId = $data[0][3]
   $dnodeId = dnode . $dnodeId
   goto switch_leader_to_offine_loop
 elif $switch_loop_cnt == 3 then
   sql show vgroups
-  $dnodeId = $data[3][3]
+  $dnodeId = $data[0][3]
   $dnodeId = dnode . $dnodeId
   goto switch_leader_to_offine_loop
 elif $switch_loop_cnt == 4 then
   sql show vgroups
-  $dnodeId = $data[4][3]  
+  $dnodeId = $data[0][3]  
   $dnodeId = dnode . $dnodeId
   goto switch_leader_to_offine_loop
 else

tests/script/tsim/tmq/consume.bat

+
+@echo off
+
+set EXEC_OPTON=start
+set DB_NAME=db 
+set CDB_NAME=db
+set /a POLL_DELAY=5
+set /a VALGRIND=0
+set SIGNAL=SIGINT
+set /a SHOW_MSG=0
+set /a SHOW_ROW=0
+set /a EXP_USE_SNAPSHOT=0
+
+:param
+if "%1"=="" (
+    goto :end
+)
+if %1 == -d ( set "DB_NAME=%2" && shift && shift && goto :param )
+if %1 == -g ( set "SHOW_MSG=%2" && shift && shift && goto :param )
+if %1 == -r ( set "SHOW_ROW=%2" && shift && shift && goto :param )
+if %1 == -s ( set "EXEC_OPTON=%2" && shift && shift && goto :param )
+if %1 == -v ( set "VALGRIND=1" && shift && goto :param )
+if %1 == -y ( set "POLL_DELAY=%2" && shift && shift && goto :param )
+if %1 == -x ( set "SIGNAL=%2" && shift && shift && goto :param )
+if %1 == -w ( set "CDB_NAME=%2" && shift && shift && goto :param )
+if %1 == -e ( set "EXP_USE_SNAPSHOT=%2" && shift && shift && goto :param )
+echo unkown argument %1
+goto :eof
+:end
+
+echo EXEC_OPTON %EXEC_OPTON%
+echo DB_NAME %DB_NAME%
+echo CDB_NAME %CDB_NAME%
+echo POLL_DELAY %POLL_DELAY%
+echo VALGRIND %VALGRIND%
+echo SIGNAL %SIGNAL%
+echo SHOW_MSG %SHOW_MSG%
+echo SHOW_ROW %SHOW_ROW%
+echo EXP_USE_SNAPSHOT %EXP_USE_SNAPSHOT%
+
+echo %cd% | grep community > nul && cd  ..\..\.. || cd  ..\..
+set BUILD_DIR=%cd%\debug\build\bin
+set SIM_DIR=%cd%\sim
+set PRG_DIR=%SIM_DIR%\tsim
+set CFG_DIR=%PRG_DIR%\cfg
+set LOG_DIR=%PRG_DIR%\log
+set PROGRAM=%BUILD_DIR%\tmq_sim.exe
+
+echo ------------------------------------------------------------------------
+echo BUILD_DIR  : %BUILD_DIR%
+echo SIM_DIR    : %SIM_DIR%
+echo CFG_DIR    : %CFG_DIR%
+echo PRG_DIR    : %PRG_DIR%
+echo CFG_DIR    : %CFG_DIR%
+echo LOG_DIR    : %LOG_DIR%
+echo PROGRAM    : %PROGRAM%
+echo POLL_DELAY : %POLL_DELAY% 
+echo DB_NAME    : %DB_NAME%
+echo ------------------------------------------------------------------------
+
+if "%EXEC_OPTON%" == "start" ( 
+  echo  mintty -h never %PROGRAM% -c %CFG_DIR% -y %POLL_DELAY% -d %DB_NAME% -g %SHOW_MSG% -r %SHOW_ROW% -w %CDB_NAME% -e %EXP_USE_SNAPSHOT%
+  mintty -h never %PROGRAM% -c %CFG_DIR% -y %POLL_DELAY% -d %DB_NAME% -g %SHOW_MSG% -r %SHOW_ROW% -w %CDB_NAME% -e %EXP_USE_SNAPSHOT%
+) else (
+  if "%SIGNAL%" == "SIGKILL" ( ps | grep tmq_sim | awk '{print $2}' | xargs kill -9 ) else ( ps | grep tmq_sim | awk '{print $2}' | xargs kill -SIGINT )
+)
+goto :eof

tests/script/wtest.bat

@@ -6,12 +6,12 @@ rem echo Start TDengine Testing Case ...
 set "SCRIPT_DIR=%~dp0"
 rem echo SCRIPT_DIR: %SCRIPT_DIR%
 
-set "BUILD_DIR=%SCRIPT_DIR%..\..\..\debug\build\bin\"
+echo %cd% | grep community > nul && set "BUILD_DIR=%SCRIPT_DIR%..\..\..\debug\build\bin\" || set "BUILD_DIR=%SCRIPT_DIR%..\..\debug\build\bin\"
 set "TSIM=%BUILD_DIR%tsim"
 rem echo BUILD_DIR:  %BUILD_DIR%
 rem echo TSIM:       %TSIM%
 
-set "SIM_DIR=%SCRIPT_DIR%..\..\..\sim\"
+echo %cd% | grep community > nul && set "SIM_DIR=%SCRIPT_DIR%..\..\..\sim\" || set "SIM_DIR=%SCRIPT_DIR%..\..\sim\"
 rem echo SIM_DIR:    %SIM_DIR%
 
 set "TSIM_DIR=%SIM_DIR%tsim\"

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

@@ -44,13 +44,13 @@ def taos_command (buildPath, key, value, expectString, cfgDir, sqlString='', key
 
     tdLog.info ("taos cmd: %s" % taosCmd)
 
-    child = taosExpect.spawn(taosCmd, timeout=3)
+    child = taosExpect.spawn(taosCmd, timeout=10)
     #output = child.readline()
     #print (output.decode())
     if len(expectString) != 0:
-        i = child.expect([expectString, taosExpect.TIMEOUT, taosExpect.EOF], timeout=6)
+        i = child.expect([expectString, taosExpect.TIMEOUT, taosExpect.EOF], timeout=10)
     else:
-        i = child.expect([taosExpect.TIMEOUT, taosExpect.EOF], timeout=6)
+        i = child.expect([taosExpect.TIMEOUT, taosExpect.EOF], timeout=10)
 
     if platform.system().lower() == 'windows':
         retResult = child.before

tests/system-test/2-query/To_unixtimestamp.py

@@ -12,7 +12,37 @@ class TDTestCase:
     def init(self, conn, logSql):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor())
+        # name of normal table
+        self.ntbname = 'ntb'    
+        # name of stable
+        self.stbname = 'stb'    
+        # structure of column
+        self.column_dict = {    
+            'ts':'timestamp',
+            'c1':'int',
+            'c2':'float',
+            'c3':'binary(20)',
+            'c4':'nchar(20)'
+        }
+        # structure of tag
+        self.tag_dict = {       
+            't0':'int'
+        }
+        # number of child tables
+        self.tbnum = 2       
+        # values of tag,the number of values should equal to tbnum
+        self.tag_values = [     
+            f'10',
+            f'100'
+        ]
+        # values of rows, structure should be same as column
+        self.values_list = [
+            f'now,10,99.99,"2020-1-1 00:00:00"',
+            f'today(),100,11.111,22.222222'
 
+        ]
+        self.error_param = [1,'now()']
+    
     def run(self):  # sourcery skip: extract-duplicate-method
         tdSql.prepare()
         tdLog.printNoPrefix("==========step1:create tables==========")

tests/system-test/2-query/first.py

@@ -60,10 +60,10 @@ class TDTestCase:
             tdSql.checkRows(1)
             tdSql.checkData(0, 1, None)
         #!bug TD-16561
-        # for i in ['stb','db.stb']:
-        #     tdSql.query(f"select first(*) from {i}")
-        #     tdSql.checkRows(1)
-        #     tdSql.checkData(0, 1, None)
+        for i in ['stb','db.stb']:
+            tdSql.query(f"select first(*) from {i}")
+            tdSql.checkRows(1)
+            tdSql.checkData(0, 1, None)
         for i in column_list:
             for j in ['stb_1','db.stb_1','stb_1','db.stb_1']:
                 tdSql.query(f"select first({i}) from {j}")
@@ -125,10 +125,10 @@ class TDTestCase:
             tdSql.execute(f"create table {stbname}_{i} using {stbname} tags('beijing')")
             tdSql.execute(f"insert into {stbname}_{i}(ts) values(%d)" % (self.ts - 1-i))
         #!bug TD-16561
-        # for i in [f'{stbname}', f'{dbname}.{stbname}']:
-        #     tdSql.query(f"select first(*) from {i}")
-        #     tdSql.checkRows(1)
-        #     tdSql.checkData(0, 1, None)
+        for i in [f'{stbname}', f'{dbname}.{stbname}']:
+            tdSql.query(f"select first(*) from {i}")
+            tdSql.checkRows(1)
+            tdSql.checkData(0, 1, None)
         tdSql.query('show tables')
         vgroup_list = []
         for i in range(len(tdSql.queryResult)):
@@ -170,8 +170,8 @@ class TDTestCase:
                 elif 'nchar' in v:
                     tdSql.checkData(0, 0, f'{self.nchar_str}1')
         #!bug TD-16569
-        # tdSql.query(f"select first(*),last(*) from {stbname} where ts < 23 interval(1s)")
-        # tdSql.checkRows(0)
+        tdSql.query(f"select first(*),last(*) from {stbname} where ts < 23 interval(1s)")
+        tdSql.checkRows(0)
         tdSql.execute(f'drop database {dbname}')
         
         

tests/system-test/2-query/timezone.py

@@ -2,7 +2,7 @@
 from util.log import *
 from util.sql import *
 from util.cases import *
-
+from util.sqlset import *
 import platform
 import os
 if platform.system().lower() == 'windows':
@@ -14,10 +14,39 @@ class TDTestCase:
     def init(self, conn, logSql):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor())
+        self.setsql = TDSetSql()
+        self.arithmetic_operators = ['+','-','*','/']
+        self.arithmetic_values = [0,1,100,15.5]
+        # name of normal table
+        self.ntbname = 'ntb'    
+        # name of stable
+        self.stbname = 'stb'    
+        # structure of column
+        self.column_dict = {    
+            'ts':'timestamp',
+            'c1':'int',
+            'c2':'float',
+            'c3':'double'
+        }
+        # structure of tag
+        self.tag_dict = {       
+            't0':'int'
+        }
+        # number of child tables
+        self.tbnum = 2       
+        # values of tag,the number of values should equal to tbnum
+        self.tag_values = [     
+            f'10',
+            f'100'
+        ]
+        # values of rows, structure should be same as column
+        self.values_list = [
+            f'now,10,99.99,11.111111',
+            f'today(),100,11.111,22.222222'
 
-    def run(self):  # sourcery skip: extract-duplicate-method
-        tdSql.prepare()
-        # get system timezone
+        ]
+        self.error_param = [1,'now()']
+    def get_system_timezone(self):
         if platform.system().lower() == 'windows':
             time_zone_1 = tzlocal.get_localzone_name()
             time_zone_2 = time.strftime('(UTC, %z)')
@@ -32,122 +61,61 @@ class TDTestCase:
                 time_zone_2 = os.popen('date "+(%Z, %z)"').read().strip()
                 time_zone = time_zone_1 + " " + time_zone_2
         print("expected time zone: " + time_zone)
-
-        tdLog.printNoPrefix("==========step1:create tables==========")
-        tdSql.execute(
-            '''create table if not exists ntb
-            (ts timestamp, c1 int, c2 float,c3 double)
-            '''
-        )
-        tdSql.execute(
-            '''create table if not exists stb
-            (ts timestamp, c1 int, c2 float,c3 double) tags(t0 int)
-            '''
-        )
-        tdSql.execute(
-            '''create table if not exists stb_1 using stb tags(100)
-            '''
-        )
-
-        tdLog.printNoPrefix("==========step2:insert data==========")
-        tdSql.execute(
-            "insert into ntb values(now,10,99.99,11.111111)(today(),100,11.111,22.222222)")
-        tdSql.execute(
-            "insert into stb_1 values(now,111,99.99,11.111111)(today(),1,11.111,22.222222)")
-
-        tdLog.printNoPrefix("==========step3:query data==========")
+        return time_zone
+    
+    def tb_type_check(self,tb_type):
+        if tb_type in ['normal_table','child_table']:
+            tdSql.checkRows(len(self.values_list))
+        elif tb_type == 'stable':
+            tdSql.checkRows(len(self.values_list*self.tbnum))
+    def data_check(self,timezone,tbname,tb_type):
+        tdSql.query(f"select timezone() from {tbname}")
+        self.tb_type_check(tb_type)
+        tdSql.checkData(0,0,timezone)
+        for symbol in self.arithmetic_operators:
+            tdSql.query(f"select timezone(){symbol}null from {tbname}")
+            self.tb_type_check(tb_type)
+            tdSql.checkData(0,0,None)
+        for i in self.arithmetic_values:
+            for symbol in self.arithmetic_operators:
+                tdSql.query(f"select timezone(){symbol}{i} from {tbname}")
+                self.tb_type_check(tb_type)
+                if symbol == '+':
+                    tdSql.checkData(0,0,i)
+                elif symbol == '-':
+                    tdSql.checkData(0,0,-i)
+                elif symbol in ['*','/','%']:
+                    if i == 0 and symbol == '/':
+                        tdSql.checkData(0,0,None)
+                    else:
+                        tdSql.checkData(0,0,0)
+        for param in self.error_param:
+            tdSql.error(f'select timezone({param}) from {tbname}')
+        tdSql.query(f"select * from {tbname} where timezone()='{timezone}'")
+        self.tb_type_check(tb_type)
+    def timezone_check_ntb(self,timezone):
+        tdSql.prepare()
+        tdSql.execute(self.setsql.set_create_normaltable_sql(self.ntbname,self.column_dict))
+        for value in self.values_list:
+            tdSql.execute(
+                f'insert into {self.ntbname} values({value})')
+        self.data_check(timezone,self.ntbname,'normal_table')
+        tdSql.execute('drop database db')
+    def timezone_check_stb(self,timezone):
+        tdSql.prepare()
+        tdSql.execute(self.setsql.set_create_stable_sql(self.stbname,self.column_dict,self.tag_dict))
+        for i in range(self.tbnum):
+            tdSql.execute(f'create table if not exists {self.stbname}_{i} using {self.stbname} tags({self.tag_values[i]})')
+            for j in self.values_list:
+                tdSql.execute(f'insert into {self.stbname}_{i} values({j})')
+        self.data_check(timezone,self.stbname,'stable')
+        for i in range(self.tbnum):
+            self.data_check(timezone,f'{self.stbname}_{i}','child_table')
+    def run(self):  # sourcery skip: extract-duplicate-method
+        timezone = self.get_system_timezone()
+        self.timezone_check_ntb(timezone)
+        self.timezone_check_stb(timezone)
         
-        tdSql.query("select timezone() from ntb")
-        tdSql.checkRows(2)
-        tdSql.checkData(0, 0, time_zone)
-        tdSql.query("select timezone() from db.ntb")
-        tdSql.checkRows(2)
-        tdSql.checkData(0, 0, time_zone)
-        tdSql.query("select timezone() from stb")
-        tdSql.checkRows(2)
-        tdSql.checkData(0, 0, time_zone)
-        tdSql.query("select timezone() from db.stb")
-        tdSql.checkRows(2)
-        tdSql.checkData(0, 0, time_zone)
-        tdSql.query("select timezone() from stb_1")
-        tdSql.checkRows(2)
-        tdSql.checkData(0, 0, time_zone)
-        tdSql.query("select timezone() from db.stb_1 ")
-        tdSql.checkRows(2)
-        tdSql.checkData(0, 0, time_zone)
-
-        tdSql.error("select timezone(1) from stb")
-        tdSql.error("select timezone(1) from db.stb")
-        tdSql.error("select timezone(1) from ntb")
-        tdSql.error("select timezone(1) from db.ntb")
-        tdSql.error("select timezone(1) from stb_1")
-        tdSql.error("select timezone(1) from db.stb_1")
-        tdSql.error("select timezone(now()) from stb")
-        tdSql.error("select timezone(now()) from db.stb")
-
-        tdSql.query(f"select * from ntb where timezone()='{time_zone}'")
-        tdSql.checkRows(2)
-        tdSql.query("select timezone()+1 from ntb")
-        tdSql.checkRows(2)
-        tdSql.query("select timezone()+1 from db.ntb")
-        tdSql.checkRows(2)
-        tdSql.query("select timezone()+1 from stb")
-        tdSql.checkRows(2)
-        tdSql.query("select timezone()+1 from db.stb")
-        tdSql.checkRows(2)
-        tdSql.query("select timezone()+1 from stb_1")
-        tdSql.checkRows(2)
-        tdSql.query("select timezone()+1 from db.stb_1")
-        tdSql.checkRows(2)
-        tdSql.query("select timezone()+1.5 from ntb")
-        tdSql.checkRows(2)
-        tdSql.query("select timezone()+1.5 from db.ntb")
-        tdSql.checkRows(2)
-        tdSql.query("select timezone()-100 from ntb")
-        tdSql.checkRows(2)
-        tdSql.query("select timezone()*100 from ntb")
-        tdSql.checkRows(2)
-        tdSql.query("select timezone()/10 from ntb")
-        # tdSql.query("select timezone()/0 from ntb")
-
-
-        tdSql.query("select timezone()+null from ntb")
-        tdSql.checkRows(2)
-        tdSql.checkData(0,0,None)
-        tdSql.query("select timezone()-null from ntb")
-        tdSql.checkRows(2)
-        tdSql.checkData(0,0,None)
-        tdSql.query("select timezone()*null from ntb")
-        tdSql.checkRows(2)
-        tdSql.checkData(0,0,None)
-        tdSql.query("select timezone()/null from ntb")
-        tdSql.checkRows(2)
-        tdSql.checkData(0,0,None)
-        # tdSql.query("select timezone()")
-        tdSql.query("select timezone()+null from stb")
-        tdSql.checkRows(2)
-        tdSql.checkData(0,0,None)
-        tdSql.query("select timezone()-null from stb")
-        tdSql.checkRows(2)
-        tdSql.checkData(0,0,None)
-        tdSql.query("select timezone()*null from stb")
-        tdSql.checkRows(2)
-        tdSql.checkData(0,0,None)
-        tdSql.query("select timezone()/null from stb")
-        tdSql.checkRows(2)
-        tdSql.checkData(0,0,None)
-        tdSql.query("select timezone()+null from stb_1")
-        tdSql.checkRows(2)
-        tdSql.checkData(0,0,None)
-        tdSql.query("select timezone()-null from stb_1")
-        tdSql.checkRows(2)
-        tdSql.checkData(0,0,None)
-        tdSql.query("select timezone()*null from stb_1")
-        tdSql.checkRows(2)
-        tdSql.checkData(0,0,None)
-        tdSql.query("select timezone()/null from stb_1")
-        tdSql.checkRows(2)
-        tdSql.checkData(0,0,None)
     def stop(self):
         tdSql.close()
         tdLog.success(f"{__file__} successfully executed")

tests/system-test/simpletest.bat

 
-@REM python3 .\test.py -f 0-others\taosShell.py
-@REM python3 .\test.py -f 0-others\taosShellError.py
+python3 .\test.py -f 0-others\taosShell.py
+python3 .\test.py -f 0-others\taosShellError.py
 python3 .\test.py -f 0-others\taosShellNetChk.py
 python3 .\test.py -f 0-others\telemetry.py
 python3 .\test.py -f 0-others\taosdMonitor.py
 python3 .\test.py -f 0-others\udfTest.py
-@REM python3 .\test.py -f 0-others\udf_create.py
-@REM python3 .\test.py -f 0-others\udf_restart_taosd.py
+python3 .\test.py -f 0-others\udf_create.py
+python3 .\test.py -f 0-others\udf_restart_taosd.py
 @REM python3 .\test.py -f 0-others\cachelast.py
 
 @REM python3 .\test.py -f 0-others\user_control.py

tests/system-test/test.py

@@ -38,7 +38,9 @@ def checkRunTimeError():
     while 1:
         time.sleep(1)
         timeCount = timeCount + 1
+        print("checkRunTimeError",timeCount)
         if (timeCount>900):
+            print("stop the test.")
             os.system("TASKKILL /F /IM taosd.exe")
             os.system("TASKKILL /F /IM taos.exe")
             os.system("TASKKILL /F /IM tmq_sim.exe")