Merge branch '3.0' of github.com:taosdata/TDengine into test/chr/TD-14699

Jenkinsfile2

@@ -33,16 +33,17 @@ def abort_previous(){
     milestone(buildNumber)
 }
 def pre_test(){
-    sh 'hostname'
     sh '''
+        hostname
         date
-        sudo rmtaos || echo "taosd has not installed"
     '''
     sh '''
-        killall -9 taosd ||echo "no taosd running"
-        killall -9 gdb || echo "no gdb running"
-        killall -9 python3.8 || echo "no python program running"
+        cd ${WK}
+        git reset --hard
+        git fetch || git fetch
         cd ${WKC}
+        git reset --hard
+        git fetch || git fetch
     '''
     script {
         if (env.CHANGE_TARGET == 'master') {
@@ -81,10 +82,10 @@ def pre_test(){
             git pull >/dev/null
             git fetch origin +refs/pull/${CHANGE_ID}/merge
             git checkout -qf FETCH_HEAD
-            git log|head -n20
+            git log -5
             cd ${WK}
             git pull >/dev/null
-            git log|head -n20
+            git log -5
         '''
     } else if (env.CHANGE_URL =~ /\/TDinternal\//) {
         sh '''
@@ -92,10 +93,10 @@ def pre_test(){
             git pull >/dev/null
             git fetch origin +refs/pull/${CHANGE_ID}/merge
             git checkout -qf FETCH_HEAD
-            git log|head -n20
+            git log -5
             cd ${WKC}
             git pull >/dev/null
-            git log|head -n20
+            git log -5
         '''
     } else {
         sh '''
@@ -106,21 +107,10 @@ def pre_test(){
         cd ${WKC}
         git submodule update --init --recursive
     '''
-    sh '''
-        cd ${WK}
-        export TZ=Asia/Harbin
-        date
-        rm -rf debug
-        mkdir debug
-        cd debug
-        cmake .. > /dev/null
-        make -j4> /dev/null
-    '''
     sh '''
         cd ${WKPY}
         git reset --hard
         git pull
-        pip3 install .
     '''
     return 1
 }
@@ -131,12 +121,14 @@ def pre_test_win(){
         time /t
         taskkill /f /t /im python.exe
         taskkill /f /t /im bash.exe
-        cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDengine
-        rd /s /Q C:\\workspace\\%EXECUTOR_NUMBER%\\TDengine\\debug
+        rd /s /Q C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\debug
         exit 0
     '''
     bat '''
-        cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDengine
+        cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
+        git reset --hard
+        git fetch || git fetch
+        cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
         git reset --hard
         git fetch || git fetch
         git checkout -f
@@ -144,39 +136,73 @@ def pre_test_win(){
     script {
         if (env.CHANGE_TARGET == 'master') {
             bat '''
-                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDengine
+                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
+                git checkout master
+                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
                 git checkout master
             '''
         } else if(env.CHANGE_TARGET == '2.0') {
             bat '''
-                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDengine
+                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
+                git checkout 2.0
+                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
                 git checkout 2.0
             '''
         } else if(env.CHANGE_TARGET == '3.0') {
             bat '''
-                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDengine
+                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
+                git checkout 3.0
+                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
                 git checkout 3.0
             '''
         } else {
             bat '''
-                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDengine
+                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
+                git checkout develop
+                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
                 git checkout develop
             '''
         }
     }
+    script {
+        if (env.CHANGE_URL =~ /\/TDengine\//) {
+            bat '''
+                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
+                git pull
+                git log -5
+                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
+                git pull
+                git fetch origin +refs/pull/${CHANGE_ID}/merge
+                git checkout -qf FETCH_HEAD
+                git log -5
+            '''
+        } else if (env.CHANGE_URL =~ /\/TDinternal\//) {
             bat '''
-        cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDengine
-        git branch
-        git pull || git pull
-        git fetch origin +refs/pull/%CHANGE_ID%/merge
+                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
+                git pull
+                git fetch origin +refs/pull/${CHANGE_ID}/merge
                 git checkout -qf FETCH_HEAD
+                git log -5
+                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
+                git pull
+                git log -5
+            '''
+        } else {
+            sh '''
+                echo "unmatched reposiotry ${CHANGE_URL}"
+            '''
+        }
+    }
+    bat '''
+        cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
+        git submodule update --init --recursive
     '''
 }
 def pre_test_build_win() {
     bat '''
         echo "building ..."
         time /t
-        cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDengine
+        cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
         mkdir debug
         cd debug
         call "C:\\Program Files (x86)\\Microsoft Visual Studio\\2017\\Community\\VC\\Auxiliary\\Build\\vcvarsall.bat" x64
@@ -192,6 +218,7 @@ pipeline {
     agent none
     options { skipDefaultCheckout() } 
     environment{
+        WKDIR = '/var/lib/jenkins/workspace'
         WK = '/var/lib/jenkins/workspace/TDinternal'
         WKC = '/var/lib/jenkins/workspace/TDinternal/community'
         WKPY = '/var/lib/jenkins/workspace/taos-connector-python'
@@ -206,40 +233,24 @@ pipeline {
                         changeRequest()
                     }
                     steps {
-                        timeout(time: 45, unit: 'MINUTES'){
+                        timeout(time: 40, unit: 'MINUTES'){
                             pre_test()
                             script {
-                                if (env.CHANGE_URL =~ /\/TDengine\//) {
                                 sh '''
-                                        cd ${WK}/debug
-                                        ctest -VV
-                                    '''
-                                    sh '''
-                                        export LD_LIBRARY_PATH=${WK}/debug/build/lib
-                                        cd ${WKC}/tests/system-test
-                                        ./fulltest.sh
-                                    '''
-                                } else if (env.CHANGE_URL =~ /\/TDinternal\//) {
-                                    sh '''
-                                        cd ${WKC}/debug
-                                        ctest -VV
+                                    cd ${WKC}/tests/parallel_test
+                                    date
+                                    time ./container_build.sh -w ${WKDIR} -t 8 -e
+                                    rm -f /tmp/cases.task
+                                    ./collect_cases.sh -e
                                 '''
                                 sh '''
-                                        export LD_LIBRARY_PATH=${WKC}/debug/build/lib
-                                        cd ${WKC}/tests/system-test
-                                        ./fulltest.sh
-                                    '''
-                                } else {
-                                    sh '''
-                                        echo "unmatched reposiotry ${CHANGE_URL}"
+                                    cd ${WKC}/tests/parallel_test
+                                    export DEFAULT_RETRY_TIME=2
+                                    date
+                                    timeout 2100 time ./run.sh -e -m /home/m.json -t /tmp/cases.task -b ${BRANCH_NAME} -l ${WKDIR}/log -o 480
                                 '''
                             }
                         }
-                            sh '''
-                                cd ${WKC}/tests
-                                ./test-all.sh b1fq
-                            '''
-                        }
                     }
                 }
             }

contrib/CMakeLists.txt

@@ -365,7 +365,7 @@ if(${BUILD_ADDR2LINE})
     if(HAVE_LIBELF_H OR HAVE_LIBELF_LIBELF_H)
         target_link_libraries(libdwarf PUBLIC libelf)
     endif()
-    target_include_directories(libdwarf SYSTEM PUBLIC "libdwarf/src/lib/libdwarf" ${CMAKE_BINARY_DIR}/contrib)
+    target_include_directories(libdwarf SYSTEM PUBLIC "libdwarf/src/lib/libdwarf" ${CMAKE_CURRENT_BINARY_DIR})
     file(READ "addr2line/addr2line.c" ADDR2LINE_CONTENT)
     string(REPLACE "static int" "int" ADDR2LINE_CONTENT "${ADDR2LINE_CONTENT}")
     string(REPLACE "static void" "void" ADDR2LINE_CONTENT "${ADDR2LINE_CONTENT}")

example/src/tmq.c

@@ -22,7 +22,7 @@
 static int  running = 1;
 static void msg_process(TAOS_RES* msg) {
   char buf[1024];
-  memset(buf, 0, 1024);
+  /*memset(buf, 0, 1024);*/
   printf("topic: %s\n", tmq_get_topic_name(msg));
   printf("vg: %d\n", tmq_get_vgroup_id(msg));
   while (1) {
@@ -61,7 +61,7 @@ int32_t init_env() {
   taos_free_result(pRes);
 
   pRes =
-      taos_query(pConn, "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(10)) tags(t1 int)");
+      taos_query(pConn, "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int)");
   if (taos_errno(pRes) != 0) {
     printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes));
     return -1;
@@ -106,8 +106,8 @@ int32_t create_topic() {
   }
   taos_free_result(pRes);
 
-  /*pRes = taos_query(pConn, "create topic topic_ctb_column as abc1");*/
-  pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from st1");
+  pRes = taos_query(pConn, "create topic topic_ctb_column as abc1");
+  /*pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from st1");*/
   if (taos_errno(pRes) != 0) {
     printf("failed to create topic topic_ctb_column, reason:%s\n", taos_errstr(pRes));
     return -1;

include/common/tcommon.h

@@ -48,6 +48,7 @@ enum {
 typedef enum EStreamType {
   STREAM_NORMAL = 1,
   STREAM_INVERT,
+  STREAM_REPROCESS,
   STREAM_INVALID,
 } EStreamType;
 
@@ -203,7 +204,6 @@ typedef struct {
   const char*   key;
   int32_t       keyLen;
   uint8_t       type;
-  int16_t length;
   union{
     const char* value;
     int64_t     i;
@@ -211,7 +211,7 @@ typedef struct {
     double      d;
     float       f;
   };
-  int32_t valueLen;
+  int32_t       length;
 } SSmlKv;
 
 #define QUERY_ASC_FORWARD_STEP  1

include/common/tglobal.h

@@ -34,7 +34,6 @@ extern int32_t  tsVersion;
 extern int32_t  tsStatusInterval;
 
 // common
-extern int32_t tsMaxConnections;
 extern int32_t tsMaxShellConns;
 extern int32_t tsShellActivityTimer;
 extern int32_t tsCompressMsgSize;

include/common/tmsgcb.h

@@ -37,11 +37,12 @@ typedef enum {
   QUEUE_MAX,
 } EQueueType;
 
-typedef int32_t (*PutToQueueFp)(SMgmtWrapper* pWrapper, SRpcMsg* pReq);
-typedef int32_t (*GetQueueSizeFp)(SMgmtWrapper* pWrapper, int32_t vgId, EQueueType qtype);
+typedef int32_t (*PutToQueueFp)(void *pMgmt, SRpcMsg* pReq);
+typedef int32_t (*GetQueueSizeFp)(void *pMgmt, int32_t vgId, EQueueType qtype);
 typedef int32_t (*SendReqFp)(SMgmtWrapper* pWrapper, const SEpSet* epSet, SRpcMsg* pReq);
 typedef int32_t (*SendMnodeReqFp)(SMgmtWrapper* pWrapper, SRpcMsg* pReq);
 typedef void (*SendRspFp)(SMgmtWrapper* pWrapper, const SRpcMsg* pRsp);
+typedef void (*SendMnodeRecvFp)(SMgmtWrapper* pWrapper, SRpcMsg* pReq, SRpcMsg* pRsp);
 typedef void (*SendRedirectRspFp)(SMgmtWrapper* pWrapper, const SRpcMsg* pRsp, const SEpSet* pNewEpSet);
 typedef void (*RegisterBrokenLinkArgFp)(SMgmtWrapper* pWrapper, SRpcMsg* pMsg);
 typedef void (*ReleaseHandleFp)(SMgmtWrapper* pWrapper, void* handle, int8_t type);
@@ -49,23 +50,26 @@ typedef void (*ReportStartup)(SMgmtWrapper* pWrapper, const char* name, const ch
 
 typedef struct {
   SMgmtWrapper*           pWrapper;
+  void*                   pMgmt;
+  void*                   clientRpc;
   PutToQueueFp            queueFps[QUEUE_MAX];
   GetQueueSizeFp          qsizeFp;
   SendReqFp               sendReqFp;
   SendRspFp               sendRspFp;
+  SendMnodeRecvFp         sendMnodeRecvFp;
   SendRedirectRspFp       sendRedirectRspFp;
   RegisterBrokenLinkArgFp registerBrokenLinkArgFp;
   ReleaseHandleFp         releaseHandleFp;
   ReportStartup           reportStartupFp;
-  void*                   clientRpc;
 } SMsgCb;
 
 void    tmsgSetDefaultMsgCb(const SMsgCb* pMsgCb);
 int32_t tmsgPutToQueue(const SMsgCb* pMsgCb, EQueueType qtype, SRpcMsg* pReq);
 int32_t tmsgGetQueueSize(const SMsgCb* pMsgCb, int32_t vgId, EQueueType qtype);
 int32_t tmsgSendReq(const SMsgCb* pMsgCb, const SEpSet* epSet, SRpcMsg* pReq);
-void    tmsgSendRsp(const SRpcMsg* pRsp);
-void    tmsgSendRedirectRsp(const SRpcMsg* pRsp, const SEpSet* pNewEpSet);
+void    tmsgSendRsp(SRpcMsg* pRsp);
+void    tmsgSendMnodeRecv(SRpcMsg* pReq, SRpcMsg* pRsp);
+void    tmsgSendRedirectRsp(SRpcMsg* pRsp, const SEpSet* pNewEpSet);
 void    tmsgRegisterBrokenLinkArg(const SMsgCb* pMsgCb, SRpcMsg* pMsg);
 void    tmsgReleaseHandle(void* handle, int8_t type);
 void    tmsgReportStartup(const char* name, const char* desc);

include/libs/function/function.h

@@ -126,7 +126,7 @@ enum {
 
 enum {
   MAIN_SCAN     = 0x0u,
-  REVERSE_SCAN  = 0x1u,
+  REVERSE_SCAN  = 0x1u,  // todo remove it
   REPEAT_SCAN   = 0x2u,  //repeat scan belongs to the master scan
   MERGE_STAGE   = 0x20u,
 };
@@ -173,6 +173,7 @@ typedef struct SqlFunctionCtx {
   SInputColumnInfoData input;
   SResultDataInfo      resDataInfo;
   uint32_t             order;  // data block scanner order: asc|desc
+  uint8_t              scanFlag;  // record current running step, default: 0
   ////////////////////////////////////////////////////////////////
   int32_t          startRow;   // start row index
   int32_t          size;       // handled processed row number
@@ -183,7 +184,6 @@ typedef struct SqlFunctionCtx {
   bool             hasNull;      // null value exist in current block, TODO remove it
   bool             requireNull;  // require null in some function, TODO remove it
   int32_t          columnIndex;  // TODO remove it
-  uint8_t          currentStage;  // record current running step, default: 0
   bool             isAggSet;
   int64_t          startTs;       // timestamp range of current query when function is executed on a specific data block, TODO remove it
   bool             stableQuery;
@@ -222,13 +222,6 @@ enum {
 typedef struct tExprNode {
   int32_t nodeType;
   union {
-    struct {
-      int32_t           optr;   // binary operator
-      void             *info;   // support filter operation on this expression only available for leaf node
-      struct tExprNode *pLeft;  // left child pointer
-      struct tExprNode *pRight; // right child pointer
-    } _node;
-
     SSchema            *pSchema;// column node
     struct SVariant    *pVal;   // value node
 
@@ -237,12 +230,6 @@ typedef struct tExprNode {
       int32_t           functionId;
       int32_t           num;
       struct SFunctionNode    *pFunctNode;
-      // Note that the attribute of pChild is not the parameter of function, it is the columns that involved in the
-      // calculation instead.
-      // E.g., Cov(col1, col2), the column information, w.r.t. the col1 and col2, is kept in pChild nodes.
-      //  The concat function, concat(col1, col2), is a binary scalar
-      //  operator and is kept in the attribute of _node.
-      struct tExprNode **pChild;
     } _function;
 
     struct {
@@ -273,6 +260,7 @@ typedef struct SAggFunctionInfo {
 struct SScalarParam {
   SColumnInfoData *columnData;
   SHashObj        *pHashFilter;
+  void            *param;  // other parameter, such as meta handle from vnode, to extract table name/tag value
   int32_t          numOfRows;
 };
 
@@ -281,10 +269,6 @@ int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionI
 
 bool qIsValidUdf(SArray* pUdfInfo, const char* name, int32_t len, int32_t* functionId);
 
-tExprNode* exprTreeFromBinary(const void* data, size_t size);
-
-tExprNode* exprdup(tExprNode* pTree);
-
 void resetResultRowEntryResult(SqlFunctionCtx* pCtx, int32_t num);
 void cleanupResultRowEntry(struct SResultRowEntryInfo* pCell);
 int32_t getNumOfResult(SqlFunctionCtx* pCtx, int32_t num, SSDataBlock* pResBlock);

include/libs/nodes/plannodes.h

@@ -193,7 +193,6 @@ typedef struct SScanPhysiNode {
 } SScanPhysiNode;
 
 typedef SScanPhysiNode STagScanPhysiNode;
-typedef SScanPhysiNode SStreamScanPhysiNode;
 
 typedef struct SSystemTableScanPhysiNode {
   SScanPhysiNode scan;
@@ -217,6 +216,7 @@ typedef struct STableScanPhysiNode {
 } STableScanPhysiNode;
 
 typedef STableScanPhysiNode STableSeqScanPhysiNode;
+typedef STableScanPhysiNode SStreamScanPhysiNode;
 
 typedef struct SProjectPhysiNode {
   SPhysiNode node;

include/libs/parser/parser.h

@@ -77,7 +77,7 @@ int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char*
 
 void*   smlInitHandle(SQuery* pQuery);
 void    smlDestroyHandle(void* pHandle);
-int32_t smlBindData(void* handle, SArray* tags, SArray* colsFormat, SArray* colsSchema, SArray* cols, bool format,
+int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols, bool format,
                     STableMeta* pTableMeta, char* tableName, char* msgBuf, int16_t msgBufLen);
 int32_t smlBuildOutput(void* handle, SHashObj* pVgHash);
 

include/util/tdef.h

@@ -425,9 +425,12 @@ enum {
   SND_WORKER_TYPE__UNIQUE,
 };
 
-#define MNODE_HANDLE   -1
-#define QNODE_HANDLE   1
 #define DEFAULT_HANDLE 0
+#define MNODE_HANDLE   -1
+#define QNODE_HANDLE   -2
+#define SNODE_HANDLE   -3
+#define VNODE_HANDLE   -4
+#define BNODE_HANDLE   -5
 
 #define TSDB_CONFIG_OPTION_LEN 16
 #define TSDB_CONIIG_VALUE_LEN  48

include/util/tjson.h

@@ -22,17 +22,12 @@
 extern "C" {
 #endif
 
-#ifdef WINDOWS
-#define tjsonGetNumberValue(pJson, pName, val) -1
-#else
-#define tjsonGetNumberValue(pJson, pName, val) \
-  ({ \
+#define tjsonGetNumberValue(pJson, pName, val, code)  \
+  do {                                                \
     uint64_t _tmp = 0;                                \
-    int32_t _code = tjsonGetUBigIntValue(pJson, pName, &_tmp); \
+    code = tjsonGetUBigIntValue(pJson, pName, &_tmp); \
     val = _tmp;                                       \
-    _code; \
-  })
-#endif
+  } while (0)
 
 typedef void SJson;
 

source/client/inc/clientStmt.h

@@ -63,7 +63,7 @@ typedef struct SStmtBindInfo {
   int8_t       tbType;
   bool         tagsCached;
   void*        boundTags;  
-  char         tbName[TSDB_TABLE_FNAME_LEN];;
+  char         tbName[TSDB_TABLE_FNAME_LEN];
   char         tbFName[TSDB_TABLE_FNAME_LEN];
   char         stbFName[TSDB_TABLE_FNAME_LEN];
   SName        sname;
@@ -71,7 +71,6 @@ typedef struct SStmtBindInfo {
 
 typedef struct SStmtExecInfo {
   int32_t      affectedRows;
-  bool         emptyRes;
   SRequestObj* pRequest;
   SHashObj*    pVgHash;
   SHashObj*    pBlockHash;
@@ -87,7 +86,6 @@ typedef struct SStmtSQLInfo {
   char*             sqlStr;
   int32_t           sqlLen;
   SArray*           nodeList;
-  SQueryPlan*       pQueryPlan;
   SStmtQueryResInfo queryRes;
   bool              autoCreateTbl;
 } SStmtSQLInfo;

source/client/src/clientEnv.c

@@ -91,7 +91,7 @@ void *openTransporter(const char *user, const char *auth, int32_t numOfThread) {
   rpcInit.label = "TSC";
   rpcInit.numOfThreads = numOfThread;
   rpcInit.cfp = processMsgFromServer;
-  rpcInit.sessions = tsMaxConnections;
+  rpcInit.sessions = 1024;
   rpcInit.connType = TAOS_CONN_CLIENT;
   rpcInit.user = (char *)user;
   rpcInit.idleTime = tsShellActivityTimer * 1000;

source/client/src/clientMain.c

@@ -303,6 +303,7 @@ int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int num_fields)
         break;
     }
   }
+  str[len] = 0;
 
   return len;
 }
@@ -567,7 +568,7 @@ void taos_query_a(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param
     // todo directly call fp
   }
 
-  taos_query_l(taos, sql, (int32_t) strlen(sql));
+  taos_query_l(taos, sql, (int32_t)strlen(sql));
 }
 
 void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {

source/client/src/clientStmt.c

@@ -279,7 +279,6 @@ int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool freeRequest) {
   }
 
   pStmt->exec.autoCreateTbl = false;
-  pStmt->exec.emptyRes = false;
   
   if (keepTable) {
     return TSDB_CODE_SUCCESS;
@@ -298,7 +297,6 @@ int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
   taosMemoryFree(pStmt->sql.queryRes.userFields);
   taosMemoryFree(pStmt->sql.sqlStr);
   qDestroyQuery(pStmt->sql.pQuery);
-  qDestroyQueryPlan(pStmt->sql.pQueryPlan);
   taosArrayDestroy(pStmt->sql.nodeList);
 
   void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
@@ -599,6 +597,8 @@ int32_t stmtFetchColFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD** fiel
 int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
   STscStmt* pStmt = (STscStmt*)stmt;
 
+  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
+
   if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
       STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
     pStmt->bInfo.needParse = false;
@@ -617,19 +617,40 @@ int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
     STMT_ERR_RET(stmtParseSql(pStmt));
   }
 
-  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
-
   if (STMT_TYPE_QUERY == pStmt->sql.type) {
-    if (NULL == pStmt->sql.pQueryPlan) {
-      STMT_ERR_RET(getQueryPlan(pStmt->exec.pRequest, pStmt->sql.pQuery, &pStmt->sql.nodeList));
-      pStmt->sql.pQueryPlan = pStmt->exec.pRequest->body.pDag;
-      pStmt->exec.pRequest->body.pDag = NULL;
-      STMT_ERR_RET(stmtBackupQueryFields(pStmt));
-    } else {
-      STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
+    STMT_ERR_RET(qStmtBindParams(pStmt->sql.pQuery, bind, colIdx, pStmt->exec.pRequest->requestId));
+    
+    SParseContext ctx = {.requestId = pStmt->exec.pRequest->requestId,
+                         .acctId = pStmt->taos->acctId,
+                         .db = pStmt->exec.pRequest->pDb,
+                         .topicQuery = false,
+                         .pSql = pStmt->sql.sqlStr,
+                         .sqlLen = pStmt->sql.sqlLen,
+                         .pMsg = pStmt->exec.pRequest->msgBuf,
+                         .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
+                         .pTransporter = pStmt->taos->pAppInfo->pTransporter,
+                         .pStmtCb = NULL,
+                         .pUser = pStmt->taos->user};
+    ctx.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
+    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &ctx.pCatalog));
+    
+    STMT_ERR_RET(qStmtParseQuerySql(&ctx, pStmt->sql.pQuery));
+
+    if (pStmt->sql.pQuery->haveResultSet) {
+      setResSchemaInfo(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->pResSchema, pStmt->sql.pQuery->numOfResCols);
+      setResPrecision(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->precision);
     }
   
-    STMT_RET(qStmtBindParam(pStmt->sql.pQueryPlan, bind, colIdx, pStmt->exec.pRequest->requestId, &pStmt->exec.emptyRes));
+    TSWAP(pStmt->exec.pRequest->dbList, pStmt->sql.pQuery->pDbList);
+    TSWAP(pStmt->exec.pRequest->tableList, pStmt->sql.pQuery->pTableList);  
+
+    //if (STMT_TYPE_QUERY == pStmt->sql.queryRes) {
+    //  STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
+    //}
+
+    //STMT_ERR_RET(stmtBackupQueryFields(pStmt));
+
+    return TSDB_CODE_SUCCESS;
   }
  
   STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
@@ -736,11 +757,7 @@ int stmtExec(TAOS_STMT *stmt) {
   STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
 
   if (STMT_TYPE_QUERY == pStmt->sql.type) {
-    if (pStmt->exec.emptyRes) {
-      pStmt->exec.pRequest->type = TSDB_SQL_RETRIEVE_EMPTY_RESULT;
-    } else {
-      scheduleQuery(pStmt->exec.pRequest, pStmt->sql.pQueryPlan, pStmt->sql.nodeList, NULL);
-    }
+    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, TSDB_CODE_SUCCESS, true, NULL);
   } else {
     STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->exec.pVgHash, pStmt->exec.pBlockHash));
     launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, TSDB_CODE_SUCCESS, true, (autoCreateTbl ? (void**)&pRsp : NULL));
@@ -839,16 +856,7 @@ int stmtGetParamNum(TAOS_STMT* stmt, int* nums) {
   }
 
   if (STMT_TYPE_QUERY == pStmt->sql.type) {
-    if (NULL == pStmt->sql.pQueryPlan) {
-      STMT_ERR_RET(getQueryPlan(pStmt->exec.pRequest, pStmt->sql.pQuery, &pStmt->sql.nodeList));
-      pStmt->sql.pQueryPlan = pStmt->exec.pRequest->body.pDag;
-      pStmt->exec.pRequest->body.pDag = NULL;
-      STMT_ERR_RET(stmtBackupQueryFields(pStmt));
-    } else {
-      STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
-    }
-
-    *nums = taosArrayGetSize(pStmt->sql.pQueryPlan->pPlaceholderValues);
+    *nums = taosArrayGetSize(pStmt->sql.pQuery->pPlaceholderValues);
   } else {
     STMT_ERR_RET(stmtFetchColFields(stmt, nums, NULL));
   }

source/client/src/tmq.c

@@ -185,6 +185,7 @@ typedef struct {
   int32_t        async;
   tsem_t         rspSem;
   tmq_resp_err_t rspErr;
+  SArray*        offsets;
 } SMqCommitCbParam;
 
 tmq_conf_t* tmq_conf_new() {
@@ -246,10 +247,13 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
   if (strcmp(key, "msg.with.table.name") == 0) {
     if (strcmp(value, "true") == 0) {
       conf->withTbName = 1;
+      return TMQ_CONF_OK;
     } else if (strcmp(value, "false") == 0) {
       conf->withTbName = 0;
+      return TMQ_CONF_OK;
     } else if (strcmp(value, "none") == 0) {
       conf->withTbName = -1;
+      return TMQ_CONF_OK;
     } else {
       return TMQ_CONF_INVALID;
     }
@@ -395,6 +399,9 @@ int32_t tmqCommitCb(void* param, const SDataBuf* pMsg, int32_t code) {
   if (!pParam->async)
     tsem_post(&pParam->rspSem);
   else {
+    if (pParam->offsets) {
+      taosArrayDestroy(pParam->offsets);
+    }
     tsem_destroy(&pParam->rspSem);
     /*if (pParam->pArray) {*/
     /*taosArrayDestroy(pParam->pArray);*/
@@ -540,10 +547,10 @@ tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, in
   // build msg
   // send to mnode
   SMqCMCommitOffsetReq req;
-  SArray*              pArray = NULL;
+  SArray*              pOffsets = NULL;
 
   if (offsets == NULL) {
-    pArray = taosArrayInit(0, sizeof(SMqOffset));
+    pOffsets = taosArrayInit(0, sizeof(SMqOffset));
     for (int i = 0; i < taosArrayGetSize(tmq->clientTopics); i++) {
       SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i);
       for (int j = 0; j < taosArrayGetSize(pTopic->vgs); j++) {
@@ -553,11 +560,11 @@ tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, in
         strcpy(offset.cgroup, tmq->groupId);
         offset.vgId = pVg->vgId;
         offset.offset = pVg->currentOffset;
-        taosArrayPush(pArray, &offset);
+        taosArrayPush(pOffsets, &offset);
       }
     }
-    req.num = pArray->size;
-    req.offsets = pArray->pData;
+    req.num = pOffsets->size;
+    req.offsets = pOffsets->pData;
   } else {
     req.num = taosArrayGetSize(&offsets->container);
     req.offsets = (SMqOffset*)offsets->container.pData;
@@ -591,6 +598,7 @@ tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, in
   pParam->tmq = tmq;
   tsem_init(&pParam->rspSem, 0, 0);
   pParam->async = async;
+  pParam->offsets = pOffsets;
 
   pRequest->body.requestMsg = (SDataBuf){
       .pData = buf,
@@ -613,8 +621,8 @@ tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, in
     tsem_destroy(&pParam->rspSem);
     taosMemoryFree(pParam);
 
-    if (pArray) {
-      taosArrayDestroy(pArray);
+    if (pOffsets) {
+      taosArrayDestroy(pOffsets);
     }
   }
 
@@ -1015,7 +1023,7 @@ int32_t tmqAskEp(tmq_t* tmq, bool async) {
   atomic_store_32(&tmq->epSkipCnt, 0);
 #endif
   int32_t      tlen = sizeof(SMqAskEpReq);
-  SMqAskEpReq* req = taosMemoryMalloc(tlen);
+  SMqAskEpReq* req = taosMemoryCalloc(1, tlen);
   if (req == NULL) {
     tscError("failed to malloc get subscribe ep buf");
     /*atomic_store_8(&tmq->epStatus, 0);*/
@@ -1025,7 +1033,7 @@ int32_t tmqAskEp(tmq_t* tmq, bool async) {
   req->epoch = htonl(tmq->epoch);
   strcpy(req->cgroup, tmq->groupId);
 
-  SMqAskEpCbParam* pParam = taosMemoryMalloc(sizeof(SMqAskEpCbParam));
+  SMqAskEpCbParam* pParam = taosMemoryCalloc(1, sizeof(SMqAskEpCbParam));
   if (pParam == NULL) {
     tscError("failed to malloc subscribe param");
     taosMemoryFree(req);
@@ -1107,7 +1115,7 @@ SMqPollReq* tmqBuildConsumeReqImpl(tmq_t* tmq, int64_t waitTime, SMqClientTopic*
     reqOffset = tmq->resetOffsetCfg;
   }
 
-  SMqPollReq* pReq = taosMemoryMalloc(sizeof(SMqPollReq));
+  SMqPollReq* pReq = taosMemoryCalloc(1, sizeof(SMqPollReq));
   if (pReq == NULL) {
     return NULL;
   }

source/common/src/systable.c

@@ -26,7 +26,7 @@ static const SSysDbTableSchema dnodesSchema[] = {
     {.name = "id", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT},
     {.name = "endpoint", .bytes = TSDB_EP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
     {.name = "vnodes", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT},
-    {.name = "max_vnodes", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT},
+    {.name = "support_vnodes", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT},
     {.name = "status", .bytes = 10 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
     {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
     {.name = "note", .bytes = 256 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},

source/common/src/tdatablock.c

@@ -1447,6 +1447,10 @@ void blockDebugShowData(const SArray* dataBlocks) {
       for (int32_t k = 0; k < colNum; k++) {
         SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
         void*            var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
+        if (pColInfoData->hasNull) {
+          printf(" %15s |", "NULL");
+          continue;
+        }
         switch (pColInfoData->info.type) {
           case TSDB_DATA_TYPE_TIMESTAMP:
             formatTimestamp(pBuf, *(uint64_t*)var, TSDB_TIME_PRECISION_MILLI);
@@ -1464,6 +1468,9 @@ void blockDebugShowData(const SArray* dataBlocks) {
           case TSDB_DATA_TYPE_UBIGINT:
             printf(" %15lu |", *(uint64_t*)var);
             break;
+          case TSDB_DATA_TYPE_DOUBLE:
+            printf(" %15f |", *(double*)var);
+            break;
         }
       }
       printf("\n");

source/common/src/tglobal.c

@@ -33,18 +33,17 @@ int32_t  tsStatusInterval = 1;  // second
 
 // common
 int32_t tsMaxShellConns = 50000;
-int32_t tsMaxConnections = 50000;
 int32_t tsShellActivityTimer = 3;  // second
 bool    tsEnableSlaveQuery = true;
 bool    tsPrintAuth = false;
 
 // multi process
 bool    tsMultiProcess = false;
-int32_t tsMnodeShmSize = TSDB_MAX_WAL_SIZE * 2;
-int32_t tsVnodeShmSize = TSDB_MAX_WAL_SIZE * 10;
-int32_t tsQnodeShmSize = TSDB_MAX_WAL_SIZE * 4;
-int32_t tsSnodeShmSize = TSDB_MAX_WAL_SIZE * 4;
-int32_t tsBnodeShmSize = TSDB_MAX_WAL_SIZE * 4;
+int32_t tsMnodeShmSize = TSDB_MAX_WAL_SIZE * 2 + 128;
+int32_t tsVnodeShmSize = TSDB_MAX_WAL_SIZE * 10 + 128;
+int32_t tsQnodeShmSize = TSDB_MAX_WAL_SIZE * 4 + 128;
+int32_t tsSnodeShmSize = TSDB_MAX_WAL_SIZE * 4 + 128;
+int32_t tsBnodeShmSize = TSDB_MAX_WAL_SIZE * 4 + 128;
 
 // queue & threads
 int32_t tsNumOfRpcThreads = 1;
@@ -280,6 +279,7 @@ int32_t taosAddClientLogCfg(SConfig *pCfg) {
   if (cfgAddInt32(pCfg, "cDebugFlag", cDebugFlag, 0, 255, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "uDebugFlag", uDebugFlag, 0, 255, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "rpcDebugFlag", rpcDebugFlag, 0, 255, 1) != 0) return -1;
+  if (cfgAddInt32(pCfg, "qDebugFlag", qDebugFlag, 0, 255, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "tmrDebugFlag", tmrDebugFlag, 0, 255, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "jniDebugFlag", jniDebugFlag, 0, 255, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "simDebugFlag", 143, 0, 255, 1) != 0) return -1;
@@ -351,15 +351,14 @@ static int32_t taosAddSystemCfg(SConfig *pCfg) {
 }
 
 static int32_t taosAddServerCfg(SConfig *pCfg) {
-  if (cfgAddInt32(pCfg, "supportVnodes", 256, 0, 4096, 0) != 0) return -1;
   if (cfgAddDir(pCfg, "dataDir", tsDataDir, 0) != 0) return -1;
   if (cfgAddFloat(pCfg, "minimalDataDirGB", 2.0f, 0.001f, 10000000, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "maxNumOfDistinctRes", tsMaxNumOfDistinctResults, 10 * 10000, 10000 * 10000, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "maxConnections", tsMaxConnections, 1, 100000, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "supportVnodes", 256, 0, 4096, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "maxShellConns", tsMaxShellConns, 10, 50000000, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "statusInterval", tsStatusInterval, 1, 30, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "minSlidingTime", tsMinSlidingTime, 10, 1000000, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "minIntervalTime", tsMinIntervalTime, 1, 1000000, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "maxNumOfDistinctRes", tsMaxNumOfDistinctResults, 10 * 10000, 10000 * 10000, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "maxStreamCompDelay", tsMaxStreamComputDelay, 10, 1000000000, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "maxFirstStreamCompDelay", tsStreamCompStartDelay, 1000, 1000000000, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "retryStreamCompDelay", tsRetryStreamCompDelay, 10, 1000000000, 0) != 0) return -1;
@@ -371,11 +370,11 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   if (cfgAddBool(pCfg, "deadLockKillQuery", tsDeadLockKillQuery, 0) != 0) return -1;
 
   if (cfgAddBool(pCfg, "multiProcess", tsMultiProcess, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "mnodeShmSize", tsMnodeShmSize, 4096, INT32_MAX, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "vnodeShmSize", tsVnodeShmSize, 4096, INT32_MAX, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "qnodeShmSize", tsQnodeShmSize, 4096, INT32_MAX, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "snodeShmSize", tsSnodeShmSize, 4096, INT32_MAX, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "bnodeShmSize", tsBnodeShmSize, 4096, INT32_MAX, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "mnodeShmSize", tsMnodeShmSize, TSDB_MAX_WAL_SIZE + 128, INT32_MAX, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "vnodeShmSize", tsVnodeShmSize, TSDB_MAX_WAL_SIZE + 128, INT32_MAX, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "qnodeShmSize", tsQnodeShmSize, TSDB_MAX_WAL_SIZE + 128, INT32_MAX, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "snodeShmSize", tsSnodeShmSize, TSDB_MAX_WAL_SIZE + 128, INT32_MAX, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "bnodeShmSize", tsBnodeShmSize, TSDB_MAX_WAL_SIZE + 128, INT32_MAX, 0) != 0) return -1;
 
   tsNumOfRpcThreads = tsNumOfCores / 2;
   tsNumOfRpcThreads = TRANGE(tsNumOfRpcThreads, 1, 4);
@@ -458,6 +457,7 @@ static void taosSetClientLogCfg(SConfig *pCfg) {
   tsLogKeepDays = cfgGetItem(pCfg, "logKeepDays")->i32;
   cDebugFlag = cfgGetItem(pCfg, "cDebugFlag")->i32;
   uDebugFlag = cfgGetItem(pCfg, "uDebugFlag")->i32;
+  qDebugFlag = cfgGetItem(pCfg, "qDebugFlag")->i32;
   rpcDebugFlag = cfgGetItem(pCfg, "rpcDebugFlag")->i32;
   tmrDebugFlag = cfgGetItem(pCfg, "tmrDebugFlag")->i32;
   jniDebugFlag = cfgGetItem(pCfg, "jniDebugFlag")->i32;
@@ -533,12 +533,11 @@ static void taosSetSystemCfg(SConfig *pCfg) {
 
 static int32_t taosSetServerCfg(SConfig *pCfg) {
   tsDataSpace.reserved = cfgGetItem(pCfg, "minimalDataDirGB")->fval;
-  tsMaxNumOfDistinctResults = cfgGetItem(pCfg, "maxNumOfDistinctRes")->i32;
-  tsMaxConnections = cfgGetItem(pCfg, "maxConnections")->i32;
   tsMaxShellConns = cfgGetItem(pCfg, "maxShellConns")->i32;
   tsStatusInterval = cfgGetItem(pCfg, "statusInterval")->i32;
   tsMinSlidingTime = cfgGetItem(pCfg, "minSlidingTime")->i32;
   tsMinIntervalTime = cfgGetItem(pCfg, "minIntervalTime")->i32;
+  tsMaxNumOfDistinctResults = cfgGetItem(pCfg, "maxNumOfDistinctRes")->i32;
   tsMaxStreamComputDelay = cfgGetItem(pCfg, "maxStreamCompDelay")->i32;
   tsStreamCompStartDelay = cfgGetItem(pCfg, "maxFirstStreamCompDelay")->i32;
   tsRetryStreamCompDelay = cfgGetItem(pCfg, "retryStreamCompDelay")->i32;

source/common/src/tmsgcb.c

@@ -19,12 +19,16 @@
 
 static SMsgCb tsDefaultMsgCb;
 
-void tmsgSetDefaultMsgCb(const SMsgCb* pMsgCb) { tsDefaultMsgCb = *pMsgCb; }
+void tmsgSetDefaultMsgCb(const SMsgCb* pMsgCb) {
+  // if (tsDefaultMsgCb.pWrapper == NULL) {
+  tsDefaultMsgCb = *pMsgCb;
+  //}
+}
 
 int32_t tmsgPutToQueue(const SMsgCb* pMsgCb, EQueueType qtype, SRpcMsg* pReq) {
   PutToQueueFp fp = pMsgCb->queueFps[qtype];
   if (fp != NULL) {
-    return (*fp)(pMsgCb->pWrapper, pReq);
+    return (*fp)(pMsgCb->pMgmt, pReq);
   } else {
     terrno = TSDB_CODE_INVALID_PTR;
     return -1;
@@ -34,7 +38,7 @@ int32_t tmsgPutToQueue(const SMsgCb* pMsgCb, EQueueType qtype, SRpcMsg* pReq) {
 int32_t tmsgGetQueueSize(const SMsgCb* pMsgCb, int32_t vgId, EQueueType qtype) {
   GetQueueSizeFp fp = pMsgCb->qsizeFp;
   if (fp != NULL) {
-    return (*fp)(pMsgCb->pWrapper, vgId, qtype);
+    return (*fp)(pMsgCb->pMgmt, vgId, qtype);
   } else {
     terrno = TSDB_CODE_INVALID_PTR;
     return -1;
@@ -51,7 +55,7 @@ int32_t tmsgSendReq(const SMsgCb* pMsgCb, const SEpSet* epSet, SRpcMsg* pReq) {
   }
 }
 
-void tmsgSendRsp(const SRpcMsg* pRsp) {
+void tmsgSendRsp(SRpcMsg* pRsp) {
   SendRspFp fp = tsDefaultMsgCb.sendRspFp;
   if (fp != NULL) {
     return (*fp)(tsDefaultMsgCb.pWrapper, pRsp);
@@ -60,7 +64,7 @@ void tmsgSendRsp(const SRpcMsg* pRsp) {
   }
 }
 
-void tmsgSendRedirectRsp(const SRpcMsg* pRsp, const SEpSet* pNewEpSet) {
+void tmsgSendRedirectRsp(SRpcMsg* pRsp, const SEpSet* pNewEpSet) {
   SendRedirectRspFp fp = tsDefaultMsgCb.sendRedirectRspFp;
   if (fp != NULL) {
     (*fp)(tsDefaultMsgCb.pWrapper, pRsp, pNewEpSet);
@@ -69,6 +73,15 @@ void tmsgSendRedirectRsp(const SRpcMsg* pRsp, const SEpSet* pNewEpSet) {
   }
 }
 
+void tmsgSendMnodeRecv(SRpcMsg* pReq, SRpcMsg* pRsp) {
+  SendMnodeRecvFp fp = tsDefaultMsgCb.sendMnodeRecvFp;
+  if (fp != NULL) {
+    (*fp)(tsDefaultMsgCb.pWrapper, pReq, pRsp);
+  } else {
+    terrno = TSDB_CODE_INVALID_PTR;
+  }
+}
+
 void tmsgRegisterBrokenLinkArg(const SMsgCb* pMsgCb, SRpcMsg* pMsg) {
   RegisterBrokenLinkArgFp fp = pMsgCb->registerBrokenLinkArgFp;
   if (fp != NULL) {

source/common/src/tname.c

@@ -317,7 +317,11 @@ void buildChildTableName(RandTableName* rName) {
   for (int j = 0; j < size; ++j) {
     SSmlKv* tagKv = taosArrayGetP(rName->tags, j);
     taosStringBuilderAppendStringLen(&sb, tagKv->key, tagKv->keyLen);
-    taosStringBuilderAppendStringLen(&sb, tagKv->value, tagKv->valueLen);
+    if(IS_VAR_DATA_TYPE(tagKv->type)){
+      taosStringBuilderAppendStringLen(&sb, tagKv->value, tagKv->length);
+    }else{
+      taosStringBuilderAppendStringLen(&sb, (char*)(&(tagKv->value)), tagKv->length);
+    }
   }
   size_t    len = 0;
   char*     keyJoined = taosStringBuilderGetResult(&sb, &len);

source/dnode/mgmt/CMakeLists.txt

-add_subdirectory(interface)
-add_subdirectory(implement)
+add_subdirectory(node_mgmt)
+add_subdirectory(node_util)
 add_subdirectory(mgmt_bnode)
 add_subdirectory(mgmt_mnode)
 add_subdirectory(mgmt_qnode)
 add_subdirectory(mgmt_snode)
 add_subdirectory(mgmt_vnode)
+add_subdirectory(mgmt_dnode)
 add_subdirectory(test)
 
 aux_source_directory(exe EXEC_SRC)
 add_executable(taosd ${EXEC_SRC})
 target_include_directories(
     taosd
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/implement/inc"
+    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/node_mgmt/inc"
 )
 target_link_libraries(taosd dnode)

source/dnode/mgmt/exe/dmMain.c

@@ -14,7 +14,7 @@
  */
 
 #define _DEFAULT_SOURCE
-#include "dmImp.h"
+#include "dmMgmt.h"
 #include "tconfig.h"
 
 #define DM_APOLLO_URL    "The apollo string to use when configuring the server, such as: -a 'jsonFile:./tests/cfg.json', cfg.json text can be '{\"fqdn\":\"td1\"}'."
@@ -163,14 +163,14 @@ static SDnodeOpt dmGetOpt() {
 
 static int32_t dmInitLog() {
   char logName[12] = {0};
-  snprintf(logName, sizeof(logName), "%slog", dmLogName(global.ntype));
+  snprintf(logName, sizeof(logName), "%slog", dmNodeLogName(global.ntype));
   return taosCreateLog(logName, 1, configDir, global.envCmd, global.envFile, global.apolloUrl, global.pArgs, 0);
 }
 
 static void dmSetProcInfo(int32_t argc, char **argv) {
   taosSetProcPath(argc, argv);
   if (global.ntype != DNODE && global.ntype != NODE_END) {
-    const char *name = dmProcName(global.ntype);
+    const char *name = dmNodeProcName(global.ntype);
     taosSetProcName(argc, argv, name);
   }
 }

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

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http:www.gnu.org/licenses/>.
- */
-
-#define _DEFAULT_SOURCE
-#include "dmImp.h"
-
-static void dmUpdateDnodeCfg(SDnode *pDnode, SDnodeCfg *pCfg) {
-  if (pDnode->data.dnodeId == 0 || pDnode->data.clusterId == 0) {
-    dInfo("set dnodeId:%d clusterId:%" PRId64, pCfg->dnodeId, pCfg->clusterId);
-    taosWLockLatch(&pDnode->data.latch);
-    pDnode->data.dnodeId = pCfg->dnodeId;
-    pDnode->data.clusterId = pCfg->clusterId;
-    dmWriteEps(pDnode);
-    taosWUnLockLatch(&pDnode->data.latch);
-  }
-}
-
-static int32_t dmProcessStatusRsp(SDnode *pDnode, SRpcMsg *pRsp) {
-  if (pRsp->code != TSDB_CODE_SUCCESS) {
-    if (pRsp->code == TSDB_CODE_MND_DNODE_NOT_EXIST && !pDnode->data.dropped && pDnode->data.dnodeId > 0) {
-      dInfo("dnode:%d, set to dropped since not exist in mnode", pDnode->data.dnodeId);
-      pDnode->data.dropped = 1;
-      dmWriteEps(pDnode);
-    }
-  } else {
-    SStatusRsp statusRsp = {0};
-    if (pRsp->pCont != NULL && pRsp->contLen > 0 &&
-        tDeserializeSStatusRsp(pRsp->pCont, pRsp->contLen, &statusRsp) == 0) {
-      pDnode->data.dnodeVer = statusRsp.dnodeVer;
-      dmUpdateDnodeCfg(pDnode, &statusRsp.dnodeCfg);
-      dmUpdateEps(pDnode, statusRsp.pDnodeEps);
-    }
-    rpcFreeCont(pRsp->pCont);
-    tFreeSStatusRsp(&statusRsp);
-  }
-
-  return TSDB_CODE_SUCCESS;
-}
-
-void dmSendStatusReq(SDnode *pDnode) {
-  SStatusReq req = {0};
-
-  taosRLockLatch(&pDnode->data.latch);
-  req.sver = tsVersion;
-  req.dnodeVer = pDnode->data.dnodeVer;
-  req.dnodeId = pDnode->data.dnodeId;
-  req.clusterId = pDnode->data.clusterId;
-  if (req.clusterId == 0) req.dnodeId = 0;
-  req.rebootTime = pDnode->data.rebootTime;
-  req.updateTime = pDnode->data.updateTime;
-  req.numOfCores = tsNumOfCores;
-  req.numOfSupportVnodes = pDnode->data.supportVnodes;
-  tstrncpy(req.dnodeEp, pDnode->data.localEp, TSDB_EP_LEN);
-
-  req.clusterCfg.statusInterval = tsStatusInterval;
-  req.clusterCfg.checkTime = 0;
-  char timestr[32] = "1970-01-01 00:00:00.00";
-  (void)taosParseTime(timestr, &req.clusterCfg.checkTime, (int32_t)strlen(timestr), TSDB_TIME_PRECISION_MILLI, 0);
-  memcpy(req.clusterCfg.timezone, tsTimezoneStr, TD_TIMEZONE_LEN);
-  memcpy(req.clusterCfg.locale, tsLocale, TD_LOCALE_LEN);
-  memcpy(req.clusterCfg.charset, tsCharset, TD_LOCALE_LEN);
-  taosRUnLockLatch(&pDnode->data.latch);
-
-  SMonVloadInfo vinfo = {0};
-  dmGetVnodeLoads(pDnode, &vinfo);
-  req.pVloads = vinfo.pVloads;
-  pDnode->data.unsyncedVgId = 0;
-  pDnode->data.vndState = TAOS_SYNC_STATE_LEADER;
-  for (int32_t i = 0; i < taosArrayGetSize(req.pVloads); ++i) {
-    SVnodeLoad *pLoad = taosArrayGet(req.pVloads, i);
-    if (pLoad->syncState != TAOS_SYNC_STATE_LEADER && pLoad->syncState != TAOS_SYNC_STATE_FOLLOWER) {
-      pDnode->data.unsyncedVgId = pLoad->vgId;
-      pDnode->data.vndState = pLoad->syncState;
-    }
-  }
-
-  SMonMloadInfo minfo = {0};
-  dmGetMnodeLoads(pDnode, &minfo);
-  pDnode->data.isMnode = minfo.isMnode;
-  pDnode->data.mndState = minfo.load.syncState;
-
-  int32_t contLen = tSerializeSStatusReq(NULL, 0, &req);
-  void   *pHead = rpcMallocCont(contLen);
-  tSerializeSStatusReq(pHead, contLen, &req);
-  tFreeSStatusReq(&req);
-
-  SRpcMsg rpcMsg = {.pCont = pHead, .contLen = contLen, .msgType = TDMT_MND_STATUS, .ahandle = (void *)0x9527};
-  SRpcMsg rpcRsp = {0};
-
-  dTrace("send req:%s to mnode, app:%p", TMSG_INFO(rpcMsg.msgType), rpcMsg.ahandle);
-  dmSendToMnodeRecv(pDnode, &rpcMsg, &rpcRsp);
-  dmProcessStatusRsp(pDnode, &rpcRsp);
-}
-
-int32_t dmProcessAuthRsp(SDnode *pDnode, SNodeMsg *pMsg) {
-  SRpcMsg *pRsp = &pMsg->rpcMsg;
-  dError("auth rsp is received, but not supported yet");
-  return 0;
-}
-
-int32_t dmProcessGrantRsp(SDnode *pDnode, SNodeMsg *pMsg) {
-  SRpcMsg *pRsp = &pMsg->rpcMsg;
-  dError("grant rsp is received, but not supported yet");
-  return 0;
-}
-
-int32_t dmProcessConfigReq(SDnode *pDnode, SNodeMsg *pMsg) {
-  SRpcMsg       *pReq = &pMsg->rpcMsg;
-  SDCfgDnodeReq *pCfg = pReq->pCont;
-  dError("config req is received, but not supported yet");
-  return TSDB_CODE_OPS_NOT_SUPPORT;
-}
-
-int32_t dmProcessCreateNodeReq(SDnode *pDnode, EDndNodeType ntype, SNodeMsg *pMsg) {
-  SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);
-  if (pWrapper != NULL) {
-    dmReleaseWrapper(pWrapper);
-    terrno = TSDB_CODE_NODE_ALREADY_DEPLOYED;
-    dError("failed to create node since %s", terrstr());
-    return -1;
-  }
-
-  taosThreadMutexLock(&pDnode->mutex);
-  pWrapper = &pDnode->wrappers[ntype];
-
-  if (taosMkDir(pWrapper->path) != 0) {
-    terrno = TAOS_SYSTEM_ERROR(errno);
-    dError("failed to create dir:%s since %s", pWrapper->path, terrstr());
-    return -1;
-  }
-
-  int32_t code = (*pWrapper->fp.createFp)(pWrapper, pMsg);
-  if (code != 0) {
-    dError("node:%s, failed to create since %s", pWrapper->name, terrstr());
-  } else {
-    dDebug("node:%s, has been created", pWrapper->name);
-    (void)dmOpenNode(pWrapper);
-    pWrapper->required = true;
-    pWrapper->deployed = true;
-    pWrapper->procType = pDnode->ptype;
-  }
-
-  taosThreadMutexUnlock(&pDnode->mutex);
-  return code;
-}
-
-int32_t dmProcessDropNodeReq(SDnode *pDnode, EDndNodeType ntype, SNodeMsg *pMsg) {
-  SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);
-  if (pWrapper == NULL) {
-    terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
-    dError("failed to drop node since %s", terrstr());
-    return -1;
-  }
-
-  taosThreadMutexLock(&pDnode->mutex);
-
-  int32_t code = (*pWrapper->fp.dropFp)(pWrapper, pMsg);
-  if (code != 0) {
-    dError("node:%s, failed to drop since %s", pWrapper->name, terrstr());
-  } else {
-    dDebug("node:%s, has been dropped", pWrapper->name);
-    pWrapper->required = false;
-    pWrapper->deployed = false;
-  }
-
-  dmReleaseWrapper(pWrapper);
-
-  if (code == 0) {
-    dmCloseNode(pWrapper);
-    taosRemoveDir(pWrapper->path);
-  }
-  taosThreadMutexUnlock(&pDnode->mutex);
-  return code;
-}
-
-static void dmSetMgmtMsgHandle(SMgmtWrapper *pWrapper) {
-  // Requests handled by DNODE
-  dmSetMsgHandle(pWrapper, TDMT_DND_CREATE_MNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
-  dmSetMsgHandle(pWrapper, TDMT_DND_DROP_MNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
-  dmSetMsgHandle(pWrapper, TDMT_DND_CREATE_QNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
-  dmSetMsgHandle(pWrapper, TDMT_DND_DROP_QNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
-  dmSetMsgHandle(pWrapper, TDMT_DND_CREATE_SNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
-  dmSetMsgHandle(pWrapper, TDMT_DND_DROP_SNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
-  dmSetMsgHandle(pWrapper, TDMT_DND_CREATE_BNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
-  dmSetMsgHandle(pWrapper, TDMT_DND_DROP_BNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
-  dmSetMsgHandle(pWrapper, TDMT_DND_CONFIG_DNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
-
-  // Requests handled by MNODE
-  dmSetMsgHandle(pWrapper, TDMT_MND_GRANT_RSP, dmProcessMgmtMsg, DEFAULT_HANDLE);
-  dmSetMsgHandle(pWrapper, TDMT_MND_AUTH_RSP, dmProcessMgmtMsg, DEFAULT_HANDLE);
-}
-
-static int32_t dmStartMgmt(SMgmtWrapper *pWrapper) {
-  if (dmStartStatusThread(pWrapper->pDnode) != 0) {
-    return -1;
-  }
-  if (dmStartMonitorThread(pWrapper->pDnode) != 0) {
-    return -1;
-  }
-  return 0;
-}
-
-static void dmStopMgmt(SMgmtWrapper *pWrapper) {
-  dmStopMonitorThread(pWrapper->pDnode);
-  dmStopStatusThread(pWrapper->pDnode);
-}
-
-static int32_t dmInitMgmt(SMgmtWrapper *pWrapper) {
-  dInfo("dnode-mgmt start to init");
-  SDnode *pDnode = pWrapper->pDnode;
-
-  pDnode->data.dnodeHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
-  if (pDnode->data.dnodeHash == NULL) {
-    dError("failed to init dnode hash");
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return -1;
-  }
-
-  if (dmReadEps(pDnode) != 0) {
-    dError("failed to read file since %s", terrstr());
-    return -1;
-  }
-
-  if (pDnode->data.dropped) {
-    dError("dnode will not start since its already dropped");
-    return -1;
-  }
-
-  if (dmStartWorker(pDnode) != 0) {
-    return -1;
-  }
-
-  if (dmInitServer(pDnode) != 0) {
-    dError("failed to init transport since %s", terrstr());
-    return -1;
-  }
-  dmReportStartup(pDnode, "dnode-transport", "initialized");
-
-  if (udfStartUdfd(pDnode->data.dnodeId) != 0) {
-    dError("failed to start udfd");
-  }
-
-  dInfo("dnode-mgmt is initialized");
-  return 0;
-}
-
-static void dmCleanupMgmt(SMgmtWrapper *pWrapper) {
-  dInfo("dnode-mgmt start to clean up");
-  SDnode *pDnode = pWrapper->pDnode;
-
-  udfStopUdfd();
-
-  dmStopWorker(pDnode);
-
-  taosWLockLatch(&pDnode->data.latch);
-  if (pDnode->data.dnodeEps != NULL) {
-    taosArrayDestroy(pDnode->data.dnodeEps);
-    pDnode->data.dnodeEps = NULL;
-  }
-  if (pDnode->data.dnodeHash != NULL) {
-    taosHashCleanup(pDnode->data.dnodeHash);
-    pDnode->data.dnodeHash = NULL;
-  }
-  taosWUnLockLatch(&pDnode->data.latch);
-
-  dmCleanupClient(pDnode);
-  dmCleanupServer(pDnode);
-  dInfo("dnode-mgmt is cleaned up");
-}
-
-static int32_t dmRequireMgmt(SMgmtWrapper *pWrapper, bool *required) {
-  *required = true;
-  return 0;
-}
-
-void dmSetMgmtFp(SMgmtWrapper *pWrapper) {
-  SMgmtFp mgmtFp = {0};
-  mgmtFp.openFp = dmInitMgmt;
-  mgmtFp.closeFp = dmCleanupMgmt;
-  mgmtFp.startFp = dmStartMgmt;
-  mgmtFp.stopFp = dmStopMgmt;
-  mgmtFp.requiredFp = dmRequireMgmt;
-
-  dmSetMgmtMsgHandle(pWrapper);
-  pWrapper->name = "dnode";
-  pWrapper->fp = mgmtFp;
-}

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

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#define _DEFAULT_SOURCE
-#include "dmImp.h"
-
-static void dmGetMonitorBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) {
-  pInfo->protocol = 1;
-  pInfo->dnode_id = pDnode->data.dnodeId;
-  pInfo->cluster_id = pDnode->data.clusterId;
-  tstrncpy(pInfo->dnode_ep, tsLocalEp, TSDB_EP_LEN);
-}
-
-static void dmGetMonitorDnodeInfo(SDnode *pDnode, SMonDnodeInfo *pInfo) {
-  pInfo->uptime = (taosGetTimestampMs() - pDnode->data.rebootTime) / (86400000.0f);
-  pInfo->has_mnode = pDnode->wrappers[MNODE].required;
-  pInfo->has_qnode = pDnode->wrappers[QNODE].required;
-  pInfo->has_snode = pDnode->wrappers[SNODE].required;
-  pInfo->has_bnode = pDnode->wrappers[BNODE].required;
-  tstrncpy(pInfo->logdir.name, tsLogDir, sizeof(pInfo->logdir.name));
-  pInfo->logdir.size = tsLogSpace.size;
-  tstrncpy(pInfo->tempdir.name, tsTempDir, sizeof(pInfo->tempdir.name));
-  pInfo->tempdir.size = tsTempSpace.size;
-}
-
-static void dmGetMonitorInfo(SDnode *pDnode, SMonDmInfo *pInfo) {
-  dmGetMonitorBasicInfo(pDnode, &pInfo->basic);
-  dmGetMonitorSysInfo(&pInfo->sys);
-  dmGetMonitorDnodeInfo(pDnode, &pInfo->dnode);
-}
-
-void dmSendMonitorReport(SDnode *pDnode) {
-  if (!tsEnableMonitor || tsMonitorFqdn[0] == 0 || tsMonitorPort == 0) return;
-  dTrace("send monitor report to %s:%u", tsMonitorFqdn, tsMonitorPort);
-
-  SMonDmInfo dmInfo = {0};
-  SMonMmInfo mmInfo = {0};
-  SMonVmInfo vmInfo = {0};
-  SMonQmInfo qmInfo = {0};
-  SMonSmInfo smInfo = {0};
-  SMonBmInfo bmInfo = {0};
-
-  SRpcMsg req = {0};
-  SRpcMsg rsp;
-  SEpSet  epset = {.inUse = 0, .numOfEps = 1};
-  tstrncpy(epset.eps[0].fqdn, pDnode->data.localFqdn, TSDB_FQDN_LEN);
-  epset.eps[0].port = tsServerPort;
-
-  SMgmtWrapper *pWrapper = NULL;
-  dmGetMonitorInfo(pDnode, &dmInfo);
-
-  bool getFromAPI = !tsMultiProcess;
-  pWrapper = &pDnode->wrappers[MNODE];
-  if (getFromAPI) {
-    if (dmMarkWrapper(pWrapper) == 0) {
-      mmGetMonitorInfo(pWrapper, &mmInfo);
-      dmReleaseWrapper(pWrapper);
-    }
-  } else {
-    if (pWrapper->required) {
-      req.msgType = TDMT_MON_MM_INFO;
-      dmSendRecv(pDnode, &epset, &req, &rsp);
-      if (rsp.code == 0 && rsp.contLen > 0) {
-        tDeserializeSMonMmInfo(rsp.pCont, rsp.contLen, &mmInfo);
-      }
-      rpcFreeCont(rsp.pCont);
-    }
-  }
-
-  pWrapper = &pDnode->wrappers[VNODE];
-  if (getFromAPI) {
-    if (dmMarkWrapper(pWrapper) == 0) {
-      vmGetMonitorInfo(pWrapper, &vmInfo);
-      dmReleaseWrapper(pWrapper);
-    }
-  } else {
-    if (pWrapper->required) {
-      req.msgType = TDMT_MON_VM_INFO;
-      dmSendRecv(pDnode, &epset, &req, &rsp);
-      if (rsp.code == 0 && rsp.contLen > 0) {
-        tDeserializeSMonVmInfo(rsp.pCont, rsp.contLen, &vmInfo);
-      }
-      rpcFreeCont(rsp.pCont);
-    }
-  }
-
-  pWrapper = &pDnode->wrappers[QNODE];
-  if (getFromAPI) {
-    if (dmMarkWrapper(pWrapper) == 0) {
-      qmGetMonitorInfo(pWrapper, &qmInfo);
-      dmReleaseWrapper(pWrapper);
-    }
-  } else {
-    if (pWrapper->required) {
-      req.msgType = TDMT_MON_QM_INFO;
-      dmSendRecv(pDnode, &epset, &req, &rsp);
-      if (rsp.code == 0 && rsp.contLen > 0) {
-        tDeserializeSMonQmInfo(rsp.pCont, rsp.contLen, &qmInfo);
-      }
-      rpcFreeCont(rsp.pCont);
-    }
-  }
-
-  pWrapper = &pDnode->wrappers[SNODE];
-  if (getFromAPI) {
-    if (dmMarkWrapper(pWrapper) == 0) {
-      smGetMonitorInfo(pWrapper, &smInfo);
-      dmReleaseWrapper(pWrapper);
-    }
-  } else {
-    if (pWrapper->required) {
-      req.msgType = TDMT_MON_SM_INFO;
-      dmSendRecv(pDnode, &epset, &req, &rsp);
-      if (rsp.code == 0 && rsp.contLen > 0) {
-        tDeserializeSMonSmInfo(rsp.pCont, rsp.contLen, &smInfo);
-      }
-      rpcFreeCont(rsp.pCont);
-    }
-  }
-
-  pWrapper = &pDnode->wrappers[BNODE];
-  if (getFromAPI) {
-    if (dmMarkWrapper(pWrapper) == 0) {
-      bmGetMonitorInfo(pWrapper, &bmInfo);
-      dmReleaseWrapper(pWrapper);
-    }
-  } else {
-    if (pWrapper->required) {
-      req.msgType = TDMT_MON_BM_INFO;
-      dmSendRecv(pDnode, &epset, &req, &rsp);
-      if (rsp.code == 0 && rsp.contLen > 0) {
-        tDeserializeSMonBmInfo(rsp.pCont, rsp.contLen, &bmInfo);
-      }
-      rpcFreeCont(rsp.pCont);
-    }
-  }
-
-  monSetDmInfo(&dmInfo);
-  monSetMmInfo(&mmInfo);
-  monSetVmInfo(&vmInfo);
-  monSetQmInfo(&qmInfo);
-  monSetSmInfo(&smInfo);
-  monSetBmInfo(&bmInfo);
-  tFreeSMonMmInfo(&mmInfo);
-  tFreeSMonVmInfo(&vmInfo);
-  tFreeSMonQmInfo(&qmInfo);
-  tFreeSMonSmInfo(&smInfo);
-  tFreeSMonBmInfo(&bmInfo);
-  monSendReport();
-}
-
-void dmGetVnodeLoads(SDnode *pDnode, SMonVloadInfo *pInfo) {
-  SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, VNODE);
-  if (pWrapper == NULL) return;
-
-  bool getFromAPI = !tsMultiProcess;
-  if (getFromAPI) {
-    vmGetVnodeLoads(pWrapper, pInfo);
-  } else {
-    SRpcMsg req = {.msgType = TDMT_MON_VM_LOAD};
-    SRpcMsg rsp = {0};
-    SEpSet  epset = {.inUse = 0, .numOfEps = 1};
-    tstrncpy(epset.eps[0].fqdn, pDnode->data.localFqdn, TSDB_FQDN_LEN);
-    epset.eps[0].port = tsServerPort;
-
-    dmSendRecv(pDnode, &epset, &req, &rsp);
-    if (rsp.code == 0 && rsp.contLen > 0) {
-      tDeserializeSMonVloadInfo(rsp.pCont, rsp.contLen, pInfo);
-    }
-    rpcFreeCont(rsp.pCont);
-  }
-  dmReleaseWrapper(pWrapper);
-}
-
-void dmGetMnodeLoads(SDnode *pDnode, SMonMloadInfo *pInfo) {
-  SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, MNODE);
-  if (pWrapper == NULL) {
-    pInfo->isMnode = 0;
-    return;
-  }
-
-  bool getFromAPI = !tsMultiProcess;
-  if (getFromAPI) {
-    mmGetMnodeLoads(pWrapper, pInfo);
-  } else {
-    SRpcMsg req = {.msgType = TDMT_MON_MM_LOAD};
-    SRpcMsg rsp = {0};
-    SEpSet  epset = {.inUse = 0, .numOfEps = 1};
-    tstrncpy(epset.eps[0].fqdn, pDnode->data.localFqdn, TSDB_FQDN_LEN);
-    epset.eps[0].port = tsServerPort;
-
-    dmSendRecv(pDnode, &epset, &req, &rsp);
-    if (rsp.code == 0 && rsp.contLen > 0) {
-      tDeserializeSMonMloadInfo(rsp.pCont, rsp.contLen, pInfo);
-    }
-    rpcFreeCont(rsp.pCont);
-  }
-  dmReleaseWrapper(pWrapper);
-}

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

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#define _DEFAULT_SOURCE
-#include "dmImp.h"
-
-static int32_t dmInitVars(SDnode *pDnode, const SDnodeOpt *pOption) {
-  pDnode->data.dnodeId = 0;
-  pDnode->data.clusterId = 0;
-  pDnode->data.dnodeVer = 0;
-  pDnode->data.updateTime = 0;
-  pDnode->data.rebootTime = taosGetTimestampMs();
-  pDnode->data.dropped = 0;
-  pDnode->data.localEp = strdup(pOption->localEp);
-  pDnode->data.localFqdn = strdup(pOption->localFqdn);
-  pDnode->data.firstEp = strdup(pOption->firstEp);
-  pDnode->data.secondEp = strdup(pOption->secondEp);
-  pDnode->data.dataDir = strdup(pOption->dataDir);
-  pDnode->data.disks = pOption->disks;
-  pDnode->data.numOfDisks = pOption->numOfDisks;
-  pDnode->data.supportVnodes = pOption->numOfSupportVnodes;
-  pDnode->data.serverPort = pOption->serverPort;
-  pDnode->ntype = pOption->ntype;
-
-  if (pDnode->data.dataDir == NULL || pDnode->data.localEp == NULL || pDnode->data.localFqdn == NULL ||
-      pDnode->data.firstEp == NULL || pDnode->data.secondEp == NULL) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return -1;
-  }
-
-  if (!tsMultiProcess || pDnode->ntype == DNODE || pDnode->ntype == NODE_END) {
-    pDnode->data.lockfile = dmCheckRunning(pDnode->data.dataDir);
-    if (pDnode->data.lockfile == NULL) {
-      return -1;
-    }
-  }
-
-  taosInitRWLatch(&pDnode->data.latch);
-  taosThreadMutexInit(&pDnode->mutex, NULL);
-  return 0;
-}
-
-static void dmClearVars(SDnode *pDnode) {
-  for (EDndNodeType n = DNODE; n < NODE_END; ++n) {
-    SMgmtWrapper *pMgmt = &pDnode->wrappers[n];
-    taosMemoryFreeClear(pMgmt->path);
-  }
-  if (pDnode->data.lockfile != NULL) {
-    taosUnLockFile(pDnode->data.lockfile);
-    taosCloseFile(&pDnode->data.lockfile);
-    pDnode->data.lockfile = NULL;
-  }
-  taosMemoryFreeClear(pDnode->data.localEp);
-  taosMemoryFreeClear(pDnode->data.localFqdn);
-  taosMemoryFreeClear(pDnode->data.firstEp);
-  taosMemoryFreeClear(pDnode->data.secondEp);
-  taosMemoryFreeClear(pDnode->data.dataDir);
-  taosThreadMutexDestroy(&pDnode->mutex);
-  memset(&pDnode->mutex, 0, sizeof(pDnode->mutex));
-  taosMemoryFree(pDnode);
-  dDebug("dnode memory is cleared, data:%p", pDnode);
-}
-
-SDnode *dmCreate(const SDnodeOpt *pOption) {
-  dDebug("start to create dnode");
-  int32_t code = -1;
-  char    path[PATH_MAX] = {0};
-  SDnode *pDnode = NULL;
-
-  pDnode = taosMemoryCalloc(1, sizeof(SDnode));
-  if (pDnode == NULL) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    goto _OVER;
-  }
-
-  if (dmInitVars(pDnode, pOption) != 0) {
-    dError("failed to init variables since %s", terrstr());
-    goto _OVER;
-  }
-
-  dmSetStatus(pDnode, DND_STAT_INIT);
-  dmSetMgmtFp(&pDnode->wrappers[DNODE]);
-  mmSetMgmtFp(&pDnode->wrappers[MNODE]);
-  vmSetMgmtFp(&pDnode->wrappers[VNODE]);
-  qmSetMgmtFp(&pDnode->wrappers[QNODE]);
-  smSetMgmtFp(&pDnode->wrappers[SNODE]);
-  bmSetMgmtFp(&pDnode->wrappers[BNODE]);
-
-  for (EDndNodeType n = DNODE; n < NODE_END; ++n) {
-    SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
-    snprintf(path, sizeof(path), "%s%s%s", pDnode->data.dataDir, TD_DIRSEP, pWrapper->name);
-    pWrapper->path = strdup(path);
-    pWrapper->procShm.id = -1;
-    pWrapper->pDnode = pDnode;
-    pWrapper->ntype = n;
-    pWrapper->procType = DND_PROC_SINGLE;
-    taosInitRWLatch(&pWrapper->latch);
-
-    if (pWrapper->path == NULL) {
-      terrno = TSDB_CODE_OUT_OF_MEMORY;
-      goto _OVER;
-    }
-
-    if (n != DNODE && dmReadShmFile(pWrapper) != 0) {
-      dError("node:%s, failed to read shm file since %s", pWrapper->name, terrstr());
-      goto _OVER;
-    }
-  }
-
-  if (dmInitMsgHandle(pDnode) != 0) {
-    dError("failed to init msg handles since %s", terrstr());
-    goto _OVER;
-  }
-
-  dInfo("dnode is created, data:%p", pDnode);
-  code = 0;
-
-_OVER:
-  if (code != 0 && pDnode) {
-    dmClearVars(pDnode);
-    pDnode = NULL;
-    dError("failed to create dnode since %s", terrstr());
-  }
-
-  return pDnode;
-}
-
-void dmClose(SDnode *pDnode) {
-  if (pDnode == NULL) return;
-  dmClearVars(pDnode);
-  dInfo("dnode is closed, data:%p", pDnode);
-}