Merge branch '3.0' into feature/vnode

include/common/taosmsg.h

@@ -51,7 +51,7 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_QUERY, "mq-query" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_CONNECT, "mq-connect" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_DISCONNECT, "mq-disconnect" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_SET_CUR, "mq-set-cur" )
-TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_RSP_READY, "rsp-ready" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_RES_READY, "res-ready" )
 // message from client to mnode
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CONNECT, "connect" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_ACCT, "create-acct" )	
@@ -78,8 +78,7 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_STB, "create-stb" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_STB, "alter-stb" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_STB, "drop-stb" )	
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_VGROUP_LIST, "vgroup-list" )
-TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_QUERY, "kill-query" )	
-TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_STREAM, "kill-stream" )	
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_QUERY, "kill-query" )		
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_CONN, "kill-conn" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_HEARTBEAT, "heartbeat" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_SHOW, "show" )
@@ -95,15 +94,15 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_NETWORK_TEST, "nettest" )
 
 // message from mnode to vnode
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_STB_IN, "create-stb-internal" )
-TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_STB_IN, "alter-stb-internal" )	
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_STB_IN, "alter-stb-internal" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_STB_IN, "drop-stb-internal" )
 // message from mnode to mnode
 // message from mnode to qnode
 // message from mnode to dnode
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_VNODE_IN, "create-vnode-internal" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_VNODE_IN, "alter-vnode-internal" )
-TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_VNODE_IN, "drop-vnode-internal" )	
-TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_AUTH_VNODE_IN, "auth-vnode-internal" )	
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_VNODE_IN, "drop-vnode-internal" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_AUTH_VNODE_IN, "auth-vnode-internal" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_SYNC_VNODE_IN, "sync-vnode-internal" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_COMPACT_VNODE_IN, "compact-vnode-internal" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_MNODE_IN, "create-mnode-internal" )
@@ -290,6 +289,37 @@ typedef struct SSchema {
   char    name[TSDB_COL_NAME_LEN];
 } SSchema;
 
+typedef struct {
+  int32_t  contLen;
+  int32_t  vgId;
+  int8_t   tableType;
+  int16_t  numOfColumns;
+  int16_t  numOfTags;
+  int32_t  tid;
+  int32_t  sversion;
+  int32_t  tversion;
+  int32_t  tagDataLen;
+  int32_t  sqlDataLen;
+  uint64_t uid;
+  uint64_t superTableUid;
+  uint64_t createdTime;
+  char     tableFname[TSDB_TABLE_FNAME_LEN];
+  char     stbFname[TSDB_TABLE_FNAME_LEN];
+  char     data[];
+} SMDCreateTableMsg;
+
+typedef struct {
+  int32_t len;  // one create table message
+  char    tableName[TSDB_TABLE_FNAME_LEN];
+  int16_t numOfTags;
+  int16_t numOfColumns;
+  int16_t sqlLen;  // the length of SQL, it starts after schema , sql is a null-terminated string
+  int8_t  igExists;
+  int8_t  rspMeta;
+  int8_t  reserved[16];
+  char    schema[];
+} SCreateTableMsg;
+
 typedef struct {
   char    name[TSDB_TABLE_FNAME_LEN];
   int8_t  igExists;
@@ -327,19 +357,6 @@ typedef struct {
   uint64_t suid;
 } SDropStbInternalMsg;
 
-typedef struct {
-  SMsgHead head;
-  char     name[TSDB_TABLE_FNAME_LEN];
-  char     stbFname[TSDB_TABLE_FNAME_LEN];
-  int8_t   tableType;
-  uint64_t suid;
-  int32_t  sversion;
-  int32_t  numOfTags;
-  int32_t  numOfColumns;
-  int32_t  tagDataLen;
-  char     data[];
-} SCreateTableMsg;
-
 typedef struct {
   SMsgHead head;
   char     name[TSDB_TABLE_FNAME_LEN];
@@ -946,18 +963,6 @@ typedef struct {
   char    subSqlInfo[TSDB_SHOW_SUBQUERY_LEN];  // include subqueries' index, Obj IDs and states(C-complete/I-imcomplete)
 } SQueryDesc;
 
-typedef struct {
-  char    sql[TSDB_SHOW_SQL_LEN];
-  char    dstTable[TSDB_TABLE_NAME_LEN];
-  int32_t streamId;
-  int64_t num;  // number of computing/cycles
-  int64_t useconds;
-  int64_t ctime;
-  int64_t stime;
-  int64_t slidingTime;
-  int64_t interval;
-} SStreamDesc;
-
 typedef struct {
   int32_t connId;
   int32_t pid;
@@ -1094,6 +1099,7 @@ typedef struct {
 } SUpdateTagValRsp;
 
 typedef struct SSchedulerQueryMsg {
+  uint64_t  schedulerId;
   uint64_t  queryId;
   uint64_t  taskId;
   uint32_t  contentLen;
@@ -1101,15 +1107,39 @@ typedef struct SSchedulerQueryMsg {
 } SSchedulerQueryMsg;
 
 typedef struct SSchedulerReadyMsg {
+  uint64_t  schedulerId;
   uint64_t  queryId;
   uint64_t  taskId;
 } SSchedulerReadyMsg;
 
 typedef struct SSchedulerFetchMsg {
+  uint64_t  schedulerId;
   uint64_t  queryId;
   uint64_t  taskId;
 } SSchedulerFetchMsg;
 
+typedef struct SSchedulerStatusMsg {
+  uint64_t  schedulerId;
+} SSchedulerStatusMsg;
+
+typedef struct STaskStatus {
+  uint64_t  queryId;
+  uint64_t  taskId;
+  int8_t    status;
+} STaskStatus;
+
+typedef struct SSchedulerStatusRsp {
+  uint32_t    num;
+  STaskStatus status[];
+} SSchedulerStatusRsp;
+
+
+typedef struct SSchedulerCancelMsg {
+  uint64_t  schedulerId;
+  uint64_t  queryId;
+  uint64_t  taskId;
+} SSchedulerCancelMsg;
+
 
 #pragma pack(pop)
 

include/common/tmsgtype.h

@@ -102,14 +102,6 @@ enum {
   TSDB_DEFINE_SQL_TYPE( TSDB_SQL_MAX, "max" )
 };
 
-// create table operation type
-enum TSQL_CREATE_TABLE_TYPE {
-  TSQL_CREATE_TABLE  = 0x1,
-  TSQL_CREATE_STABLE = 0x2,
-  TSQL_CREATE_CTABLE = 0x3,
-  TSQL_CREATE_STREAM = 0x4,
-};
-
 #ifdef __cplusplus
 }
 #endif

include/common/tname.h

@@ -16,8 +16,6 @@
 #ifndef TDENGINE_TNAME_H
 #define TDENGINE_TNAME_H
 
-//#include "taosmsg.h"
-
 #define TSDB_DB_NAME_T     1
 #define TSDB_TABLE_NAME_T  2
 

include/dnode/qnode/qnode.h

@@ -19,7 +19,7 @@
 #ifdef __cplusplus
 extern "C" {
 #endif
-
+#include "trpc.h"
 
 typedef struct {
   uint64_t numOfStartTask;
@@ -32,48 +32,6 @@ typedef struct {
   uint64_t numOfErrors;
 } SQnodeStat;
 
-/* start Task msg */
-typedef struct {
-  uint32_t  schedulerIp;
-  uint16_t  schedulerPort;
-  int64_t   taskId;
-  int64_t   queryId;
-  uint32_t  srcIp;
-  uint16_t  srcPort;
-} SQnodeStartTaskMsg;
-
-/* stop Task msg */
-typedef struct {
-  int64_t   taskId;
-} SQnodeStopTaskMsg;
-
-/* start/stop Task msg response */
-typedef struct {
-  int64_t   taskId;
-  int32_t code;
-} SQnodeTaskRespMsg;
-
-/* Task status msg */
-typedef struct {
-  int64_t   taskId;
-  int32_t   status;
-  int64_t   queryId;
-} SQnodeTaskStatusMsg;
-
-/* Qnode/Scheduler heartbeat msg */
-typedef struct {
-  int32_t status;
-  int32_t load;
-  
-} SQnodeHeartbeatMsg;
-
-/* Qnode sent/received msg */
-typedef struct {
-  int8_t   msgType;
-  int32_t  msgLen;
-  char     msg[];
-} SQnodeMsg;
-
 
 /**
  * Start one Qnode in Dnode.

include/dnode/vnode/tq/tq.h

@@ -22,6 +22,8 @@
 #include "taoserror.h"
 #include "taosmsg.h"
 #include "tlist.h"
+#include "trpc.h"
+#include "ttimer.h"
 #include "tutil.h"
 
 #ifdef __cplusplus
@@ -54,6 +56,7 @@ typedef struct STqSetCurReq {
 
 typedef struct STqConsumeReq {
   STqMsgHead head;
+  int64_t    blockingTime;  // milisec
   STqAcks    acks;
 } STqConsumeReq;
 
@@ -101,33 +104,44 @@ typedef struct STqTopicVhandle {
 typedef struct STqExec {
   void* runtimeEnv;
   SSDataBlock* (*exec)(void* runtimeEnv);
-  void* (*assign)(void* runtimeEnv, SSubmitBlk* inputData);
+  void* (*assign)(void* runtimeEnv, void* inputData);
   void (*clear)(void* runtimeEnv);
   char* (*serialize)(struct STqExec*);
   struct STqExec* (*deserialize)(char*);
 } STqExec;
 
+typedef struct STqRspHandle {
+  void* handle;
+  void* ahandle;
+} STqRspHandle;
+
+typedef enum { TQ_ITEM_READY, TQ_ITEM_PROCESS, TQ_ITEM_EMPTY } STqItemStatus;
+
+typedef struct STqTopic STqTopic;
+
 typedef struct STqBufferItem {
   int64_t offset;
   // executors are identical but not concurrent
   // so there must be a copy in each item
-  STqExec* executor;
-  int32_t  status;
-  int64_t  size;
-  void*    content;
+  STqExec*  executor;
+  int32_t   status;
+  int64_t   size;
+  void*     content;
+  STqTopic* pTopic;
 } STqMsgItem;
 
-typedef struct STqTopic {
+struct STqTopic {
   // char* topic; //c style, end with '\0'
   // int64_t cgId;
   // void* ahandle;
+  // int32_t    head;
+  // int32_t    tail;
   int64_t    nextConsumeOffset;
   int64_t    floatingCursor;
   int64_t    topicId;
-  int32_t    head;
-  int32_t    tail;
+  void*      logReader;
   STqMsgItem buffer[TQ_BUFFER_SIZE];
-} STqTopic;
+};
 
 typedef struct STqListHandle {
   STqTopic              topic;
@@ -135,13 +149,13 @@ typedef struct STqListHandle {
 } STqList;
 
 typedef struct STqGroup {
-  int64_t  clientId;
-  int64_t  cgId;
-  void*    ahandle;
-  int32_t  topicNum;
-  STqList* head;
-  SList*   topicList;  // SList<STqTopic>
-  void*    returnMsg;  // SVReadMsg
+  int64_t clientId;
+  int64_t cgId;
+  void*   ahandle;
+  int32_t topicNum;
+  STqList*     head;
+  SList*       topicList;  // SList<STqTopic>
+  STqRspHandle rspHandle;
 } STqGroup;
 
 typedef struct STqQueryMsg {
@@ -149,20 +163,23 @@ typedef struct STqQueryMsg {
   struct STqQueryMsg* next;
 } STqQueryMsg;
 
-typedef struct STqLogReader {
+typedef struct STqLogHandle {
   void* logHandle;
-  int32_t (*logRead)(void* logHandle, void** data, int64_t ver);
+  void* (*openLogReader)(void* logHandle);
+  void (*closeLogReader)(void* logReader);
+  int32_t (*logRead)(void* logReader, void** data, int64_t ver);
+
   int64_t (*logGetFirstVer)(void* logHandle);
   int64_t (*logGetSnapshotVer)(void* logHandle);
   int64_t (*logGetLastVer)(void* logHandle);
-} STqLogReader;
+} STqLogHandle;
 
 typedef struct STqCfg {
   // TODO
 } STqCfg;
 
 typedef struct STqMemRef {
-  SMemAllocatorFactory* pAlloctorFactory;
+  SMemAllocatorFactory* pAllocatorFactory;
   SMemAllocator*        pAllocator;
 } STqMemRef;
 
@@ -252,19 +269,30 @@ typedef struct STQ {
   // the handle of meta kvstore
   char*         path;
   STqCfg*       tqConfig;
-  STqLogReader* tqLogReader;
+  STqLogHandle* tqLogHandle;
   STqMemRef     tqMemRef;
   STqMetaStore* tqMeta;
 } STQ;
 
+typedef struct STqMgmt {
+  int8_t inited;
+  tmr_h  timer;
+} STqMgmt;
+
+static STqMgmt tqMgmt;
+
+// init once
+int  tqInit();
+void tqCleanUp();
+
 // open in each vnode
-STQ* tqOpen(const char* path, STqCfg* tqConfig, STqLogReader* tqLogReader, SMemAllocatorFactory* allocFac);
+STQ* tqOpen(const char* path, STqCfg* tqConfig, STqLogHandle* tqLogHandle, SMemAllocatorFactory* allocFac);
 void tqClose(STQ*);
 
 // void* will be replace by a msg type
 int tqPushMsg(STQ*, void* msg, int64_t version);
 int tqCommit(STQ*);
-int tqConsume(STQ*, STqConsumeReq*);
+int tqConsume(STQ*, SRpcMsg* pReq, SRpcMsg** pRsp);
 
 int tqSetCursor(STQ*, STqSetCurReq* pMsg);
 int tqBufferSetOffset(STqTopic*, int64_t offset);

include/dnode/vnode/vnode.h

@@ -122,6 +122,16 @@ int vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs);
  */
 int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
 
+/**
+ * @brief Process a consume message.
+ *
+ * @param pVnode The vnode object.
+ * @param pMsg The request message
+ * @param pRsp The response message
+ * @return int 0 for success, -1 for failure
+ */
+int vnodeProcessCMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
+
 /**
  * @brief Process the sync request
  *

include/libs/index/index.h

@@ -44,41 +44,46 @@ typedef enum { QUERY_TERM = 0, QUERY_PREFIX = 1, QUERY_SUFFIX = 2, QUERY_REGEX =
  * @param: oper
  *
  */
-SIndexMultiTermQuery *indexMultiTermQueryCreate(EIndexOperatorType oper);
-void                  indexMultiTermQueryDestroy(SIndexMultiTermQuery *pQuery);
-int                   indexMultiTermQueryAdd(SIndexMultiTermQuery *pQuery, SIndexTerm *term, EIndexQueryType type);
+SIndexMultiTermQuery* indexMultiTermQueryCreate(EIndexOperatorType oper);
+void                  indexMultiTermQueryDestroy(SIndexMultiTermQuery* pQuery);
+int                   indexMultiTermQueryAdd(SIndexMultiTermQuery* pQuery, SIndexTerm* term, EIndexQueryType type);
 /*
  * @param:
  * @param:
  */
-int  indexOpen(SIndexOpts *opt, const char *path, SIndex **index);
-void indexClose(SIndex *index);
-int  indexPut(SIndex *index, SIndexMultiTerm *terms, uint64_t uid);
-int  indexDelete(SIndex *index, SIndexMultiTermQuery *query);
-int  indexSearch(SIndex *index, SIndexMultiTermQuery *query, SArray *result);
-int  indexRebuild(SIndex *index, SIndexOpts *opt);
+int  indexOpen(SIndexOpts* opt, const char* path, SIndex** index);
+void indexClose(SIndex* index);
+int  indexPut(SIndex* index, SIndexMultiTerm* terms, uint64_t uid);
+int  indexDelete(SIndex* index, SIndexMultiTermQuery* query);
+int  indexSearch(SIndex* index, SIndexMultiTermQuery* query, SArray* result);
+int  indexRebuild(SIndex* index, SIndexOpts* opt);
 /*
  * @param
  * @param
  */
-SIndexMultiTerm *indexMultiTermCreate();
-int              indexMultiTermAdd(SIndexMultiTerm *terms, SIndexTerm *term);
-void             indexMultiTermDestroy(SIndexMultiTerm *terms);
+SIndexMultiTerm* indexMultiTermCreate();
+int              indexMultiTermAdd(SIndexMultiTerm* terms, SIndexTerm* term);
+void             indexMultiTermDestroy(SIndexMultiTerm* terms);
 /*
  * @param:
  * @param:
  */
-SIndexOpts *indexOptsCreate();
-void        indexOptsDestroy(SIndexOpts *opts);
+SIndexOpts* indexOptsCreate();
+void        indexOptsDestroy(SIndexOpts* opts);
 
 /*
  * @param:
  * @param:
  */
 
-SIndexTerm *indexTermCreate(int64_t suid, SIndexOperOnColumn operType, uint8_t colType, const char *colName,
-    int32_t nColName, const char *colVal, int32_t nColVal);
-void        indexTermDestroy(SIndexTerm *p);
+SIndexTerm* indexTermCreate(int64_t            suid,
+                            SIndexOperOnColumn operType,
+                            uint8_t            colType,
+                            const char*        colName,
+                            int32_t            nColName,
+                            const char*        colVal,
+                            int32_t            nColVal);
+void        indexTermDestroy(SIndexTerm* p);
 
 #ifdef __cplusplus
 }

include/libs/parser/parsenodes.h

+/*
+ * 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/>.
+ */
+
+#ifndef _TD_PARSENODES_H_
+#define _TD_PARSENODES_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "catalog.h"
+#include "common.h"
+#include "function.h"
+#include "tmsgtype.h"
+#include "tname.h"
+#include "tvariant.h"
+
+/*
+ * The first field of a node of any type is guaranteed to be the int16_t.
+ * Hence the type of any node can be gotten by casting it to SQueryNode. 
+ */
+typedef struct SQueryNode {
+  int16_t type;
+} SQueryNode;
+
+#define nodeType(nodeptr) (((const SQueryNode*)(nodeptr))->type)
+
+typedef struct SField {
+  char     name[TSDB_COL_NAME_LEN];
+  uint8_t  type;
+  int16_t  bytes;
+} SField;
+
+typedef struct SParseBasicCtx {
+  const char *db;
+  int32_t     acctId;
+  uint64_t    requestId;
+} SParseBasicCtx;
+
+typedef struct SFieldInfo {
+  int16_t     numOfOutput;   // number of column in result
+  SField     *final;
+  SArray     *internalField; // SArray<SInternalField>
+} SFieldInfo;
+
+typedef struct SCond {
+  uint64_t uid;
+  int32_t  len;  // length of tag query condition data
+  char *   cond;
+} SCond;
+
+typedef struct SJoinNode {
+  uint64_t uid;
+  int16_t  tagColId;
+  SArray*  tsJoin;
+  SArray*  tagJoin;
+} SJoinNode;
+
+typedef struct SJoinInfo {
+  bool       hasJoin;
+  SJoinNode *joinTables[TSDB_MAX_JOIN_TABLE_NUM];
+} SJoinInfo;
+
+typedef struct STagCond {
+  int16_t   relType;     // relation between tbname list and query condition, including : TK_AND or TK_OR
+  SCond     tbnameCond;  // tbname query condition, only support tbname query condition on one table
+  SJoinInfo joinInfo;    // join condition, only support two tables join currently
+  SArray   *pCond;       // for different table, the query condition must be seperated
+} STagCond;
+
+typedef struct STableMetaInfo {
+  STableMeta    *pTableMeta;      // table meta, cached in client side and acquired by name
+  SVgroupsInfo  *vgroupList;
+  SName         name;
+  char          aliasName[TSDB_TABLE_NAME_LEN];    // alias name of table specified in query sql
+  SArray       *tagColList;                        // SArray<SColumn*>, involved tag columns
+} STableMetaInfo;
+
+typedef struct SColumnIndex {
+  int16_t tableIndex;
+  int16_t columnIndex;
+  int16_t type;               // normal column/tag/ user input constant column
+} SColumnIndex;
+
+// select statement
+typedef struct SQueryStmtInfo {
+  int16_t          command;       // the command may be different for each subclause, so keep it seperately.
+  uint32_t         type;          // query/insert type
+  STimeWindow      window;        // the whole query time window
+  SInterval        interval;      // tumble time window
+  SSessionWindow   sessionWindow; // session time window
+  SStateWindow     stateWindow;   // state window query
+  SGroupbyExpr     groupbyExpr;   // groupby tags info
+  SArray *         colList;       // SArray<SColumn*>
+  SFieldInfo       fieldsInfo;
+  SArray**         exprList;      // SArray<SExprInfo*>
+  SLimit           limit;
+  SLimit           slimit;
+  STagCond         tagCond;
+  SArray *         colCond;
+  SArray *         order;
+  int16_t          numOfTables;
+  int16_t          curTableIdx;
+  STableMetaInfo **pTableMetaInfo;
+  struct STSBuf   *tsBuf;
+
+  int16_t          fillType;      // final result fill type
+  int64_t *        fillVal;       // default value for fill
+  int32_t          numOfFillVal;  // fill value size
+
+  char *           msg;           // pointer to the pCmd->payload to keep error message temporarily
+  int64_t          clauseLimit;   // limit for current sub clause
+
+  int64_t          prjOffset;     // offset value in the original sql expression, only applied at client side
+  int64_t          vgroupLimit;    // table limit in case of super table projection query + global order + limit
+
+  int32_t          udColumnId;    // current user-defined constant output field column id, monotonically decreases from TSDB_UD_COLUMN_INDEX
+  int32_t          bufLen;
+  char*            buf;
+  SArray          *pUdfInfo;
+
+  struct SQueryStmtInfo *sibling;     // sibling
+  struct SQueryStmtInfo *pDownstream;
+  SMultiFunctionsDesc    info;
+  SArray            *pUpstream;   // SArray<struct SQueryStmtInfo>
+  int32_t            havingFieldNum;
+  int32_t            exprListLevelIndex;
+} SQueryStmtInfo;
+
+typedef enum {
+  PAYLOAD_TYPE_KV = 0,
+  PAYLOAD_TYPE_RAW = 1,
+} EPayloadType;
+
+typedef struct SVgDataBlocks {
+  SVgroupInfo vg;
+  int32_t     numOfTables;  // number of tables in current submit block
+  uint32_t    size;
+  char       *pData;        // SMsgDesc + SSubmitMsg + SSubmitBlk + ...
+} SVgDataBlocks;
+
+typedef struct SInsertStmtInfo {
+  int16_t     nodeType;
+  SArray*     pDataBlocks;         // data block for each vgroup, SArray<SVgDataBlocks*>.
+  int8_t      schemaAttache;        // denote if submit block is built with table schema or not
+  uint8_t     payloadType;         // EPayloadType. 0: K-V payload for non-prepare insert, 1: rawPayload for prepare insert
+  uint32_t    insertType;          // insert data from [file|sql statement| bound statement]
+  const char* sql;                 // current sql statement position
+} SInsertStmtInfo;
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_PARSENODES_H_*/

include/libs/parser/parser.h

@@ -20,108 +20,7 @@
 extern "C" {
 #endif
 
-#include "catalog.h"
-#include "common.h"
-#include "tname.h"
-#include "tvariant.h"
-#include "function.h"
-
-typedef struct SField {
-  char     name[TSDB_COL_NAME_LEN];
-  uint8_t  type;
-  int16_t  bytes;
-} SField;
-
-typedef struct SFieldInfo {
-  int16_t     numOfOutput;   // number of column in result
-  SField     *final;
-  SArray     *internalField; // SArray<SInternalField>
-} SFieldInfo;
-
-typedef struct SCond {
-  uint64_t uid;
-  int32_t  len;  // length of tag query condition data
-  char *   cond;
-} SCond;
-
-typedef struct SJoinNode {
-  uint64_t uid;
-  int16_t  tagColId;
-  SArray*  tsJoin;
-  SArray*  tagJoin;
-} SJoinNode;
-
-typedef struct SJoinInfo {
-  bool       hasJoin;
-  SJoinNode *joinTables[TSDB_MAX_JOIN_TABLE_NUM];
-} SJoinInfo;
-
-typedef struct STagCond {
-  int16_t   relType;     // relation between tbname list and query condition, including : TK_AND or TK_OR
-  SCond     tbnameCond;  // tbname query condition, only support tbname query condition on one table
-  SJoinInfo joinInfo;    // join condition, only support two tables join currently
-  SArray   *pCond;       // for different table, the query condition must be seperated
-} STagCond;
-
-typedef struct STableMetaInfo {
-  STableMeta    *pTableMeta;      // table meta, cached in client side and acquired by name
-  SVgroupsInfo  *vgroupList;
-  SName         name;
-  char          aliasName[TSDB_TABLE_NAME_LEN];    // alias name of table specified in query sql
-  SArray       *tagColList;                        // SArray<SColumn*>, involved tag columns
-} STableMetaInfo;
-
-typedef struct SQueryStmtInfo {
-  int16_t          command;       // the command may be different for each subclause, so keep it seperately.
-  uint32_t         type;          // query/insert type
-  STimeWindow      window;        // the whole query time window
-  SInterval        interval;      // tumble time window
-  SSessionWindow   sessionWindow; // session time window
-  SStateWindow     stateWindow;   // state window query
-  SGroupbyExpr     groupbyExpr;   // groupby tags info
-  SArray *         colList;       // SArray<SColumn*>
-  SFieldInfo       fieldsInfo;
-  SArray**         exprList;      // SArray<SExprInfo*>
-  SLimit           limit;
-  SLimit           slimit;
-  STagCond         tagCond;
-  SArray *         colCond;
-  SArray *         order;
-  int16_t          numOfTables;
-  int16_t          curTableIdx;
-  STableMetaInfo **pTableMetaInfo;
-  struct STSBuf   *tsBuf;
-
-  int16_t          fillType;      // final result fill type
-  int64_t *        fillVal;       // default value for fill
-  int32_t          numOfFillVal;  // fill value size
-
-  char *           msg;           // pointer to the pCmd->payload to keep error message temporarily
-  int64_t          clauseLimit;   // limit for current sub clause
-
-  int64_t          prjOffset;     // offset value in the original sql expression, only applied at client side
-  int64_t          vgroupLimit;    // table limit in case of super table projection query + global order + limit
-
-  int32_t          udColumnId;    // current user-defined constant output field column id, monotonically decreases from TSDB_UD_COLUMN_INDEX
-  int32_t          bufLen;
-  char*            buf;
-  SArray          *pUdfInfo;
-
-  struct SQueryStmtInfo *sibling;     // sibling
-  struct SQueryStmtInfo *pDownstream;
-  SMultiFunctionsDesc    info;
-  SArray            *pUpstream;   // SArray<struct SQueryStmtInfo>
-  int32_t            havingFieldNum;
-  int32_t            exprListLevelIndex;
-} SQueryStmtInfo;
-
-typedef struct SColumnIndex {
-  int16_t tableIndex;
-  int16_t columnIndex;
-  int16_t type;               // normal column/tag/ user input constant column
-} SColumnIndex;
-
-struct SInsertStmtInfo;
+#include "parsenodes.h"
 
 /**
  * True will be returned if the input sql string is insert, false otherwise.
@@ -132,17 +31,16 @@ struct SInsertStmtInfo;
 bool qIsInsertSql(const char* pStr, size_t length);
 
 typedef struct SParseContext {
-  const char* pAcctId;
-  const char* pDbname;
-  void *pRpc;
-  const char* pClusterId;
-  const SEpSet* pEpSet;
-  int64_t id;          // query id, generated by uuid generator
-  int8_t schemaAttached; // denote if submit block is built with table schema or not
-  const char* pSql;    // sql string
-  size_t sqlLen;       // length of the sql string
-  char* pMsg;           // extended error message if exists to help avoid the problem in sql statement.
-  int32_t msgLen;      // max length of the msg
+  SParseBasicCtx  ctx;
+  void            *pRpc;
+  struct SCatalog *pCatalog;
+  const SEpSet    *pEpSet;
+  int64_t          id;             // query id, generated by uuid generator
+  int8_t           schemaAttached; // denote if submit block is built with table schema or not
+  const char      *pSql;           // sql string
+  size_t           sqlLen;         // length of the sql string
+  char            *pMsg;           // extended error message if exists to help avoid the problem in sql statement.
+  int32_t          msgLen;         // max length of the msg
 } SParseContext;
 
 /**
@@ -153,27 +51,7 @@ typedef struct SParseContext {
  * @param msg      extended error message if exists.
  * @return         error code
  */
-int32_t qParseQuerySql(const char* pStr, size_t length, int64_t id, int32_t* type, void** pOutput, int32_t* outputLen, char* msg, int32_t msgLen);
-
-typedef enum {
-  PAYLOAD_TYPE_KV = 0,
-  PAYLOAD_TYPE_RAW = 1,
-} EPayloadType;
-
-typedef struct SVgDataBlocks {
-  int64_t     vgId;         // virtual group id
-  int32_t     numOfTables;  // number of tables in current submit block
-  uint32_t    size;
-  char       *pData;        // SMsgDesc + SSubmitMsg + SSubmitBlk + ...
-} SVgDataBlocks;
-
-typedef struct SInsertStmtInfo {
-  SArray*     pDataBlocks;         // data block for each vgroup, SArray<SVgDataBlocks*>.
-  int8_t      schemaAttache;        // denote if submit block is built with table schema or not
-  uint8_t     payloadType;         // EPayloadType. 0: K-V payload for non-prepare insert, 1: rawPayload for prepare insert
-  uint32_t    insertType;          // insert data from [file|sql statement| bound statement]
-  const char* sql;                 // current sql statement position
-} SInsertStmtInfo;
+int32_t qParseQuerySql(const char* pStr, size_t length, SParseBasicCtx* pParseCtx, int32_t* type, void** pOutput, int32_t* outputLen, char* msg, int32_t msgLen);
 
 /**
  * Parse the insert sql statement.
@@ -210,9 +88,9 @@ typedef struct SSourceParam {
 SExprInfo* createExprInfo(STableMetaInfo* pTableMetaInfo, const char* funcName, SSourceParam* pSource, SSchema* pResSchema, int16_t interSize);
 int32_t copyExprInfoList(SArray* dst, const SArray* src, uint64_t uid, bool deepcopy);
 int32_t copyAllExprInfo(SArray* dst, const SArray* src, bool deepcopy);
-int32_t getExprFunctionLevel(SQueryStmtInfo* pQueryInfo);
+int32_t getExprFunctionLevel(const SQueryStmtInfo* pQueryInfo);
 
-STableMetaInfo* getMetaInfo(SQueryStmtInfo* pQueryInfo, int32_t tableIndex);
+STableMetaInfo* getMetaInfo(const SQueryStmtInfo* pQueryInfo, int32_t tableIndex);
 SSchema *getOneColumnSchema(const STableMeta* pTableMeta, int32_t colIndex);
 SSchema createSchema(uint8_t type, int16_t bytes, int16_t colId, const char* name);
 

include/libs/planner/planner.h

@@ -25,6 +25,7 @@ extern "C" {
 #define QUERY_TYPE_MERGE       1
 #define QUERY_TYPE_PARTIAL     2
 #define QUERY_TYPE_SCAN        3
+#define QUERY_TYPE_MODIFY      4
 
 enum OPERATOR_TYPE_E {
   OP_Unknown,
@@ -58,18 +59,17 @@ typedef struct SQueryNodeBasicInfo {
 
 typedef struct SDataSink {
   SQueryNodeBasicInfo info;
-  SDataBlockSchema schema;
 } SDataSink;
 
 typedef struct SDataDispatcher {
   SDataSink sink;
-  // todo
 } SDataDispatcher;
 
 typedef struct SDataInserter {
   SDataSink sink;
-  uint64_t    uid;  // unique id of the table
-  // todo data field
+  int32_t   numOfTables;
+  uint32_t  size;
+  char     *pData;
 } SDataInserter;
 
 typedef struct SPhyNode {
@@ -119,12 +119,13 @@ typedef struct SSubplanId {
 
 typedef struct SSubplan {
   SSubplanId id;          // unique id of the subplan
-  int32_t   type;         // QUERY_TYPE_MERGE|QUERY_TYPE_PARTIAL|QUERY_TYPE_SCAN
-  int32_t   level;        // the execution level of current subplan, starting from 0.
-  SEpSet    execEpSet;    // for the scan sub plan, the optional execution node
-  SArray   *pChildern;    // the datasource subplan,from which to fetch the result
-  SArray   *pParents;     // the data destination subplan, get data from current subplan
-  SPhyNode *pNode;        // physical plan of current subplan
+  int32_t    type;         // QUERY_TYPE_MERGE|QUERY_TYPE_PARTIAL|QUERY_TYPE_SCAN|QUERY_TYPE_MODIFY
+  int32_t    level;        // the execution level of current subplan, starting from 0.
+  SEpSet     execEpSet;    // for the scan/modify subplan, the optional execution node
+  SArray    *pChildern;    // the datasource subplan,from which to fetch the result
+  SArray    *pParents;     // the data destination subplan, get data from current subplan
+  SPhyNode  *pNode;        // physical plan of current subplan
+  SDataSink *pDataSink;    // data of the subplan flow into the datasink
 } SSubplan;
 
 typedef struct SQueryDag {
@@ -133,10 +134,12 @@ typedef struct SQueryDag {
   SArray  *pSubplans; // Element is SArray*, and nested element is SSubplan. The execution level of subplan, starting from 0.
 } SQueryDag;
 
+struct SQueryNode;
+
 /**
  * Create the physical plan for the query, according to the AST.
  */
-int32_t qCreateQueryDag(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* pQnode, struct SQueryDag** pDag);
+int32_t qCreateQueryDag(const struct SQueryNode* pQueryInfo, struct SEpSet* pQnode, struct SQueryDag** pDag);
 
 // Set datasource of this subplan, multiple calls may be made to a subplan.
 // @subplan subplan to be schedule
@@ -144,12 +147,12 @@ int32_t qCreateQueryDag(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet*
 // @ep one execution location of this group of datasource subplans 
 int32_t qSetSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SEpAddr* ep);
 
-int32_t qExplainQuery(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* pQnode, char** str);
+int32_t qExplainQuery(const struct SQueryNode* pQueryInfo, struct SEpSet* pQnode, char** str);
 
 /**
  * Convert to subplan to string for the scheduler to send to the executor
  */
-int32_t qSubPlanToString(const SSubplan* subplan, char** str);
+int32_t qSubPlanToString(const SSubplan* subplan, char** str, int32_t* len);
 
 int32_t qStringToSubplan(const char* str, SSubplan** subplan);
 

include/libs/qcom/query.h

@@ -24,6 +24,15 @@ extern "C" {
 #include "thash.h"
 #include "tlog.h"
 
+enum {
+  JOB_TASK_STATUS_NOT_START = 1,
+  JOB_TASK_STATUS_EXECUTING,
+  JOB_TASK_STATUS_SUCCEED,
+  JOB_TASK_STATUS_FAILED,
+  JOB_TASK_STATUS_CANCELLING,
+  JOB_TASK_STATUS_CANCELLED
+};
+
 typedef struct STableComInfo {
   uint8_t numOfTags;      // the number of tags in schema
   uint8_t precision;      // the number of precision
@@ -81,15 +90,27 @@ typedef struct STableMetaOutput {
   STableMeta *tbMeta;
 } STableMetaOutput;
 
+typedef int32_t __async_exec_fn_t(void* param);
+
 bool tIsValidSchema(struct SSchema* pSchema, int32_t numOfCols, int32_t numOfTags);
 
-extern int32_t (*queryBuildMsg[TSDB_MSG_TYPE_MAX])(void* input, char **msg, int32_t msgSize, int32_t *msgLen);
-extern int32_t (*queryProcessMsgRsp[TSDB_MSG_TYPE_MAX])(void* output, char *msg, int32_t msgSize);
+int32_t initTaskQueue();
+int32_t cleanupTaskQueue();
+
+/**
+ *
+ * @param execFn      The asynchronously execution function
+ * @param execParam   The parameters of the execFn
+ * @param code        The response code during execution the execFn
+ * @return
+ */
+int32_t taosAsyncExec(__async_exec_fn_t execFn, void* execParam, int32_t* code);
 
 SSchema* tGetTbnameColumnSchema();
-extern void msgInit();
+void msgInit();
 
-extern int32_t qDebugFlag;
+extern int32_t (*queryBuildMsg[TSDB_MSG_TYPE_MAX])(void* input, char **msg, int32_t msgSize, int32_t *msgLen);
+extern int32_t (*queryProcessMsgRsp[TSDB_MSG_TYPE_MAX])(void* output, char *msg, int32_t msgSize);
 
 #define qFatal(...)  do { if (qDebugFlag & DEBUG_FATAL) { taosPrintLog("QRY FATAL ", qDebugFlag, __VA_ARGS__); }} while(0)
 #define qError(...)  do { if (qDebugFlag & DEBUG_ERROR) { taosPrintLog("QRY ERROR ", qDebugFlag, __VA_ARGS__); }} while(0)

include/libs/qworker/qworker.h

+/*
+ * 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/>.
+ */
+
+#ifndef _TD_QWORKER_H_
+#define _TD_QWORKER_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "trpc.h"
+
+typedef struct SQWorkerCfg {
+  uint32_t maxSchedulerNum;
+  uint32_t maxResCacheNum;
+  uint32_t maxSchTaskNum;
+} SQWorkerCfg;
+
+typedef struct {
+  uint64_t numOfStartTask;
+  uint64_t numOfStopTask;
+  uint64_t numOfRecvedFetch;
+  uint64_t numOfSentHb;
+  uint64_t numOfSentFetch;
+  uint64_t numOfTaskInQueue;
+  uint64_t numOfFetchInQueue;
+  uint64_t numOfErrors;
+} SQWorkerStat;
+
+
+int32_t qWorkerInit(SQWorkerCfg *cfg, void **qWorkerMgmt);
+
+int32_t qWorkerProcessQueryMsg(void *qWorkerMgmt, SSchedulerQueryMsg *msg, SRpcMsg *rsp);
+
+int32_t qWorkerProcessReadyMsg(void *qWorkerMgmt, SSchedulerReadyMsg *msg, SRpcMsg *rsp);
+
+int32_t qWorkerProcessStatusMsg(void *qWorkerMgmt, SSchedulerStatusMsg *msg, SRpcMsg *rsp);
+
+int32_t qWorkerProcessFetchMsg(void *qWorkerMgmt, SSchedulerFetchMsg *msg, SRpcMsg *rsp);
+
+int32_t qWorkerProcessCancelMsg(void *qWorkerMgmt, SSchedulerCancelMsg *msg, SRpcMsg *rsp);
+
+void qWorkerDestroy(void **qWorkerMgmt);
+
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_QWORKER_H_*/

include/libs/scheduler/scheduler.h

@@ -25,6 +25,7 @@ extern "C" {
 
 typedef struct SSchedulerCfg {
   int32_t clusterType;
+  int32_t maxJobNum;
 } SSchedulerCfg;
 
 typedef struct SQueryProfileSummary {

include/libs/wal/wal.h

@@ -174,8 +174,11 @@ SWalReadHandle *walOpenReadHandle(SWal *);
 void            walCloseReadHandle(SWalReadHandle *);
 int32_t         walReadWithHandle(SWalReadHandle *pRead, int64_t ver);
 
+// deprecated
+#if 0
 int32_t walRead(SWal *, SWalHead **, int64_t ver);
-// int32_t walReadWithFp(SWal *, FWalWrite writeFp, int64_t verStart, int32_t readNum);
+int32_t walReadWithFp(SWal *, FWalWrite writeFp, int64_t verStart, int32_t readNum);
+#endif
 
 // lifecycle check
 int64_t walGetFirstVer(SWal *);

include/util/thash.h

@@ -193,11 +193,10 @@ void  taosHashCancelIterate(SHashObj *pHashObj, void *p);
 
 /**
  * Get the corresponding key information for a given data in hash table
- * @param pHashObj
  * @param data
  * @return
  */
-int32_t taosHashGetKey(SHashObj *pHashObj, void *data, void** key, size_t* keyLen);
+int32_t taosHashGetKey(void *data, void** key, size_t* keyLen);
 
 #ifdef __cplusplus
 }

include/util/ttimer.h

@@ -16,6 +16,8 @@
 #ifndef _TD_UTIL_TIMER_H
 #define _TD_UTIL_TIMER_H
 
+#include "os.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif

source/client/inc/clientInt.h

@@ -148,13 +148,16 @@ int   taos_init();
 void* createTscObj(const char* user, const char* auth, const char *ip, uint32_t port, SAppInstInfo* pAppInfo);
 void  destroyTscObj(void*pObj);
 
-void* createRequest(STscObj* pObj, __taos_async_fn_t fp, void* param, int32_t type);
-void destroyRequest(SRequestObj* pRequest);
+void *createRequest(STscObj* pObj, __taos_async_fn_t fp, void* param, int32_t type);
+void  destroyRequest(SRequestObj* pRequest);
+
+char *getConnectionDB(STscObj* pObj);
+void  setConnectionDB(STscObj* pTscObj, const char* db);
 
 void taos_init_imp(void);
 int taos_options_imp(TSDB_OPTION option, const char *str);
 
-void* openTransporter(const char *user, const char *auth);
+void* openTransporter(const char *user, const char *auth, int32_t numOfThreads);
 
 void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet);
 void initMsgHandleFp();

source/client/src/tscEnv.c → source/client/src/clientEnv.c

@@ -13,17 +13,17 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "clientInt.h"
-#include "clientLog.h"
 #include "os.h"
 #include "taosmsg.h"
+#include "query.h"
+#include "clientInt.h"
+#include "clientLog.h"
 #include "tcache.h"
 #include "tconfig.h"
 #include "tglobal.h"
 #include "tnote.h"
 #include "tref.h"
 #include "trpc.h"
-#include "tsched.h"
 #include "ttime.h"
 #include "ttimezone.h"
 
@@ -33,10 +33,8 @@
 SAppInfo   appInfo;
 int32_t    tscReqRef  = -1;
 int32_t    tscConnRef = -1;
-void      *tscQhandle = NULL;
 
 static pthread_once_t tscinit = PTHREAD_ONCE_INIT;
-int32_t tsNumOfThreads = 1;
 volatile int32_t tscInitRes = 0;
 
 static void registerRequest(SRequestObj* pRequest) {
@@ -98,12 +96,12 @@ void closeTransporter(STscObj* pTscObj)  {
 }
 
 // TODO refactor
-void* openTransporter(const char *user, const char *auth) {
+void* openTransporter(const char *user, const char *auth, int32_t numOfThread) {
   SRpcInit rpcInit;
   memset(&rpcInit, 0, sizeof(rpcInit));
   rpcInit.localPort = 0;
   rpcInit.label = "TSC";
-  rpcInit.numOfThreads = tsNumOfThreads;
+  rpcInit.numOfThreads = numOfThread;
   rpcInit.cfp = processMsgFromServer;
   rpcInit.sessions = tsMaxConnections;
   rpcInit.connType = TAOS_CONN_CLIENT;
@@ -229,18 +227,8 @@ void taos_init_imp(void) {
 
   taosSetCoreDump(true);
 
-  double factor = 4.0;
-  int32_t numOfThreads = MAX((int)(tsNumOfCores * tsNumOfThreadsPerCore / factor), 2);
-
-  int32_t queueSize = tsMaxConnections * 2;
-  tscQhandle = taosInitScheduler(queueSize, numOfThreads, "tsc");
-  if (NULL == tscQhandle) {
-    tscError("failed to init task queue");
-    tscInitRes = -1;
-    return;
-  }
+  initTaskQueue();
 
-  tscDebug("client task queue is initialized, numOfThreads: %d", numOfThreads);
   tscConnRef = taosOpenRef(200, destroyTscObj);
   tscReqRef  = taosOpenRef(40960, doDestroyRequest);
 

source/client/src/clientImpl.c

@@ -102,9 +102,8 @@ TAOS *taos_connect_internal(const char *ip, const char *user, const char *pass,
   SAppInstInfo** pInst = taosHashGet(appInfo.pInstMap, key, strlen(key));
   if (pInst == NULL) {
     SAppInstInfo* p = calloc(1, sizeof(struct SAppInstInfo));
-
     p->mgmtEp       = epSet;
-    p->pTransporter = openTransporter(user, secretEncrypt);
+    p->pTransporter = openTransporter(user, secretEncrypt, tsNumOfCores);
     taosHashPut(appInfo.pInstMap, key, strlen(key), &p, POINTER_BYTES);
 
     pInst = &p;
@@ -152,8 +151,12 @@ TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen) {
     int32_t type = 0;
     void*   output = NULL;
     int32_t outputLen = 0;
-    code = qParseQuerySql(pRequest->sqlstr, sqlLen, pRequest->requestId, &type, &output, &outputLen, pRequest->msgBuf, ERROR_MSG_BUF_DEFAULT_SIZE);
-    if (type == TSDB_SQL_CREATE_USER || type == TSDB_SQL_SHOW || type == TSDB_SQL_DROP_USER || type == TSDB_SQL_DROP_ACCT || type == TSDB_SQL_CREATE_DB || type == TSDB_SQL_CREATE_ACCT) {
+
+    SParseBasicCtx c = {.requestId = pRequest->requestId, .acctId = pTscObj->acctId, .db = getConnectionDB(pTscObj)};
+    code = qParseQuerySql(pRequest->sqlstr, sqlLen, &c, &type, &output, &outputLen, pRequest->msgBuf, ERROR_MSG_BUF_DEFAULT_SIZE);
+    if (type == TSDB_SQL_CREATE_USER || type == TSDB_SQL_SHOW || type == TSDB_SQL_DROP_USER ||
+        type == TSDB_SQL_DROP_ACCT || type == TSDB_SQL_CREATE_DB || type == TSDB_SQL_CREATE_ACCT ||
+        type == TSDB_SQL_CREATE_TABLE || type == TSDB_SQL_USE_DB) {
       pRequest->type = type;
       pRequest->body.requestMsg = (SReqMsgInfo){.pMsg = output, .len = outputLen};
 
@@ -164,12 +167,12 @@ TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen) {
       sendMsgToServer(pTscObj->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &body, &transporterId);
 
       tsem_wait(&pRequest->body.rspSem);
-
-
       destroyRequestMsgBody(&body);
     } else {
       assert(0);
     }
+
+    tfree(c.db);
   }
 
   if (code != TSDB_CODE_SUCCESS) {
@@ -437,8 +440,19 @@ void setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t
   }
 }
 
-const char *taos_get_client_info() { return version; }
+char* getConnectionDB(STscObj* pObj) {
+  char *p = NULL;
+  pthread_mutex_lock(&pObj->mutex);
+  p = strndup(pObj->db, tListLen(pObj->db));
+  pthread_mutex_unlock(&pObj->mutex);
 
-int taos_affected_rows(TAOS_RES *res) { return 1; }
+  return p;
+}
+
+void setConnectionDB(STscObj* pTscObj, const char* db) {
+  assert(db != NULL && pTscObj != NULL);
+  pthread_mutex_lock(&pTscObj->mutex);
+  tstrncpy(pTscObj->db, db, tListLen(pTscObj->db));
+  pthread_mutex_unlock(&pTscObj->mutex);
+}
 
-int taos_result_precision(TAOS_RES *res) { return TSDB_TIME_PRECISION_MILLI; }

source/client/src/clientMain.c

+#include "os.h"
 #include "clientInt.h"
 #include "clientLog.h"
-#include "os.h"
+#include "query.h"
 #include "taosmsg.h"
-#include "tcache.h"
-#include "tconfig.h"
 #include "tglobal.h"
-#include "tnote.h"
 #include "tref.h"
 #include "trpc.h"
-#include "tsched.h"
-#include "ttime.h"
-#include "ttimezone.h"
 
 #define TSC_VAR_NOT_RELEASE 1
 #define TSC_VAR_RELEASED    0
@@ -44,9 +39,7 @@ void taos_cleanup(void) {
   tscReqRef = -1;
   taosCloseRef(id);
 
-  void* p = tscQhandle;
-  tscQhandle = NULL;
-  taosCleanUpScheduler(p);
+  cleanupTaskQueue();
 
   id = tscConnRef;
   tscConnRef = -1;
@@ -262,3 +255,9 @@ const char *taos_data_type(int type) {
     default: return "UNKNOWN";
   }
 }
+
+const char *taos_get_client_info() { return version; }
+
+int taos_affected_rows(TAOS_RES *res) { return 1; }
+
+int taos_result_precision(TAOS_RES *res) { return TSDB_TIME_PRECISION_MILLI; }

source/client/test/clientTests.cpp

@@ -16,15 +16,15 @@
 #include <gtest/gtest.h>
 #include <taoserror.h>
 #include <iostream>
-#include "tglobal.h"
 #pragma GCC diagnostic ignored "-Wwrite-strings"
 
 #pragma GCC diagnostic ignored "-Wunused-function"
 #pragma GCC diagnostic ignored "-Wunused-variable"
 #pragma GCC diagnostic ignored "-Wsign-compare"
 
-#include "../inc/clientInt.h"
 #include "taos.h"
+#include "tglobal.h"
+#include "../inc/clientInt.h"
 
 namespace {
 }  // namespace
@@ -148,3 +148,36 @@ TEST(testCase, create_db_Test) {
 
   taos_close(pConn);
 }
+
+TEST(testCase, use_db_test) {
+  TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
+  assert(pConn != NULL);
+
+  TAOS_RES* pRes = taos_query(pConn, "use abc1");
+
+  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
+  ASSERT_TRUE(pFields == NULL);
+
+  int32_t numOfFields = taos_num_fields(pRes);
+  ASSERT_EQ(numOfFields, 0);
+
+  taos_close(pConn);
+}
+
+TEST(testCase, create_stable_Test) {
+  TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
+  assert(pConn != NULL);
+
+  TAOS_RES* pRes = taos_query(pConn, "use abc1");
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "create stable st1(ts timestamp, k int) tags(a int)");
+
+  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
+  ASSERT_TRUE(pFields == NULL);
+
+  int32_t numOfFields = taos_num_fields(pRes);
+  ASSERT_EQ(numOfFields, 0);
+
+  taos_close(pConn);
+}

source/common/src/tname.c

@@ -6,21 +6,6 @@
 
 #define VALID_NAME_TYPE(x)  ((x) == TSDB_DB_NAME_T || (x) == TSDB_TABLE_NAME_T)
 
-char* extractDBName(const char* tableId, char* name) {
-  size_t offset1 = strcspn(tableId, &TS_PATH_DELIMITER[0]);
-  size_t len = strcspn(&tableId[offset1 + 1], &TS_PATH_DELIMITER[0]);
-  
-  return strncpy(name, &tableId[offset1 + 1], len);
-}
-
-// todo remove it
-size_t tableIdPrefix(const char* name, char* prefix, int32_t len) {
-  tstrncpy(prefix, name, len);
-  strcat(prefix, TS_PATH_DELIMITER);
-
-  return strlen(prefix);
-}
-
 bool tscValidateTableNameLength(size_t len) {
   return len < TSDB_TABLE_NAME_LEN;
 }
@@ -125,7 +110,7 @@ int32_t tNameExtractFullName(const SName* name, char* dst) {
     return -1;
   }
 
-  int32_t len = snprintf(dst, TSDB_FULL_DB_NAME_LEN, "%s.%s", name->acctId, name->dbname);
+  int32_t len = snprintf(dst, TSDB_FULL_DB_NAME_LEN, "%d.%s", name->acctId, name->dbname);
 
   size_t tnameLen = strlen(name->tname);
   if (tnameLen > 0) {
@@ -141,7 +126,9 @@ int32_t tNameExtractFullName(const SName* name, char* dst) {
 
 int32_t tNameLen(const SName* name) {
   assert(name != NULL);
-  int32_t len  = (int32_t) strlen(name->acctId);
+
+  char tmp[12] = {0};
+  int32_t len = sprintf(tmp, "%d", name->acctId);
   int32_t len1 = (int32_t) strlen(name->dbname);
   int32_t len2 = (int32_t) strlen(name->tname);
 
@@ -161,10 +148,6 @@ bool tNameIsValid(const SName* name) {
     return false;
   }
 
-  if (strlen(name->acctId) <= 0) {
-    return false;
-  }
-
   if (name->type == TSDB_DB_NAME_T) {
     return strlen(name->dbname) > 0;
   } else {
@@ -237,13 +220,6 @@ int32_t tNameFromString(SName* dst, const char* str, uint32_t type) {
       return -1;
     }
 
-    int32_t len = (int32_t)(p - str);
-
-    // too long account id or too long db name
-//    if ((len >= tListLen(dst->acctId)) || (len <= 0)) {
-//      return -1;
-//    }
-//    memcpy (dst->acctId, str, len);
     dst->acctId = strtoll(str, NULL, 10);
   }
 
@@ -272,9 +248,8 @@ int32_t tNameFromString(SName* dst, const char* str, uint32_t type) {
     dst->type = TSDB_TABLE_NAME_T;
     char* start = (char*) ((p == NULL)? str: (p+1));
 
-    int32_t len = (int32_t) strlen(start);
-
     // too long account id or too long db name
+    int32_t len = (int32_t) strlen(start);
     if ((len >= tListLen(dst->tname)) || (len <= 0)) {
       return -1;
     }

source/dnode/mgmt/impl/src/dndTransport.c

@@ -73,7 +73,6 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
   pMgmt->msgFp[TSDB_MSG_TYPE_DROP_STB] = dndProcessMnodeWriteMsg;
   pMgmt->msgFp[TSDB_MSG_TYPE_VGROUP_LIST] = dndProcessMnodeReadMsg;
   pMgmt->msgFp[TSDB_MSG_TYPE_KILL_QUERY] = dndProcessMnodeWriteMsg;
-  pMgmt->msgFp[TSDB_MSG_TYPE_KILL_STREAM] = dndProcessMnodeWriteMsg;
   pMgmt->msgFp[TSDB_MSG_TYPE_KILL_CONN] = dndProcessMnodeWriteMsg;
   pMgmt->msgFp[TSDB_MSG_TYPE_HEARTBEAT] = dndProcessMnodeReadMsg;
   pMgmt->msgFp[TSDB_MSG_TYPE_SHOW] = dndProcessMnodeReadMsg;

source/dnode/mgmt/impl/test/CMakeLists.txt

@@ -8,8 +8,8 @@ add_subdirectory(db)
 add_subdirectory(dnode)
 # add_subdirectory(func)
 # add_subdirectory(mnode)
-# add_subdirectory(profile)
-# add_subdirectory(show)
+add_subdirectory(profile)
+add_subdirectory(show)
 add_subdirectory(stb)
 # add_subdirectory(sync)
 # add_subdirectory(telem)
@@ -17,4 +17,4 @@ add_subdirectory(stb)
 add_subdirectory(user)
 add_subdirectory(vgroup)
 
-# add_subdirectory(common)
+add_subdirectory(sut)

source/dnode/mgmt/impl/test/acct/CMakeLists.txt

-add_executable(dnode_test_acct "")
-
-target_sources(dnode_test_acct
-    PRIVATE
-    "acct.cpp"
-    "../sut/deploy.cpp"
-)
-
+aux_source_directory(. ACCT_SRC)
+add_executable(dnode_test_acct ${ACCT_SRC})
 target_link_libraries(
     dnode_test_acct
-    PUBLIC dnode
-    PUBLIC util
-    PUBLIC os
-    PUBLIC gtest_main
-)
-
-target_include_directories(dnode_test_acct
-    PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/server/dnode/mgmt"
-    "${CMAKE_CURRENT_SOURCE_DIR}/../../inc"
-    "${CMAKE_CURRENT_SOURCE_DIR}/../sut"
+    PUBLIC sut
 )
 
 add_test(

source/dnode/mgmt/impl/test/acct/acct.cpp

@@ -9,103 +9,59 @@
  *
  */
 
-#include "deploy.h"
+#include "base.h"
 
 class DndTestAcct : public ::testing::Test {
  protected:
-   static SServer* CreateServer(const char* path, const char* fqdn, uint16_t port, const char* firstEp) {
-    SServer* pServer = createServer(path, fqdn, port, firstEp);
-    ASSERT(pServer);
-    return pServer;
-  }
-
-  static void SetUpTestSuite() {
-    initLog("/tmp/tdlog");
-
-    const char* fqdn = "localhost";
-    const char* firstEp = "localhost:9012";
-    pServer = CreateServer("/tmp/dnode_test_user", fqdn, 9012, firstEp);
-    pClient = createClient("root", "taosdata", fqdn, 9012);
-    taosMsleep(300);
-  }
-
-  static void TearDownTestSuite() {
-    stopServer(pServer);
-    dropClient(pClient);
-    pServer = NULL;
-    pClient = NULL;
-  }
-
-  static SServer* pServer;
-  static SClient* pClient;
-  static int32_t  connId;
+  static void SetUpTestSuite() { test.Init("/tmp/dnode_test_acct", 9012); }
+  static void TearDownTestSuite() { test.Cleanup(); }
+
+  static Testbase test;
+
+ public:
+  void SetUp() override {}
+  void TearDown() override {}
 };
 
-SServer* DndTestAcct::pServer;
-SClient* DndTestAcct::pClient;
-int32_t  DndTestAcct::connId;
+Testbase DndTestAcct::test;
 
 TEST_F(DndTestAcct, 01_CreateAcct) {
-  ASSERT_NE(pClient, nullptr);
-
-  SCreateAcctMsg* pReq = (SCreateAcctMsg*)rpcMallocCont(sizeof(SCreateAcctMsg));
+  int32_t contLen = sizeof(SCreateAcctMsg);
 
-  SRpcMsg rpcMsg = {0};
-  rpcMsg.pCont = pReq;
-  rpcMsg.contLen = sizeof(SCreateAcctMsg);
-  rpcMsg.msgType = TSDB_MSG_TYPE_CREATE_ACCT;
+  SCreateAcctMsg* pReq = (SCreateAcctMsg*)rpcMallocCont(contLen);
 
-  sendMsg(pClient, &rpcMsg);
-  SRpcMsg* pMsg = pClient->pRsp;
+  SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_CREATE_ACCT, pReq, contLen);
   ASSERT_NE(pMsg, nullptr);
   ASSERT_EQ(pMsg->code, TSDB_CODE_MND_MSG_NOT_PROCESSED);
 }
 
 TEST_F(DndTestAcct, 02_AlterAcct) {
-  ASSERT_NE(pClient, nullptr);
+  int32_t contLen = sizeof(SCreateAcctMsg);
 
-  SAlterAcctMsg* pReq = (SAlterAcctMsg*)rpcMallocCont(sizeof(SAlterAcctMsg));
+  SAlterAcctMsg* pReq = (SAlterAcctMsg*)rpcMallocCont(contLen);
 
-  SRpcMsg rpcMsg = {0};
-  rpcMsg.pCont = pReq;
-  rpcMsg.contLen = sizeof(SAlterAcctMsg);
-  rpcMsg.msgType = TSDB_MSG_TYPE_ALTER_ACCT;
-
-  sendMsg(pClient, &rpcMsg);
-  SRpcMsg* pMsg = pClient->pRsp;
+  SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_ALTER_ACCT, pReq, contLen);
   ASSERT_NE(pMsg, nullptr);
   ASSERT_EQ(pMsg->code, TSDB_CODE_MND_MSG_NOT_PROCESSED);
 }
 
 TEST_F(DndTestAcct, 03_DropAcct) {
-  ASSERT_NE(pClient, nullptr);
-
-  SDropAcctMsg* pReq = (SDropAcctMsg*)rpcMallocCont(sizeof(SDropAcctMsg));
+  int32_t contLen = sizeof(SDropAcctMsg);
 
-  SRpcMsg rpcMsg = {0};
-  rpcMsg.pCont = pReq;
-  rpcMsg.contLen = sizeof(SDropAcctMsg);
-  rpcMsg.msgType = TSDB_MSG_TYPE_DROP_ACCT;
+  SDropAcctMsg* pReq = (SDropAcctMsg*)rpcMallocCont(contLen);
 
-  sendMsg(pClient, &rpcMsg);
-  SRpcMsg* pMsg = pClient->pRsp;
+  SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_DROP_ACCT, pReq, contLen);
   ASSERT_NE(pMsg, nullptr);
   ASSERT_EQ(pMsg->code, TSDB_CODE_MND_MSG_NOT_PROCESSED);
 }
 
 TEST_F(DndTestAcct, 04_ShowAcct) {
-  ASSERT_NE(pClient, nullptr);
+  int32_t contLen = sizeof(SShowMsg);
 
-  SShowMsg* pReq = (SShowMsg*)rpcMallocCont(sizeof(SShowMsg));
+  SShowMsg* pReq = (SShowMsg*)rpcMallocCont(contLen);
   pReq->type = TSDB_MGMT_TABLE_ACCT;
 
-  SRpcMsg rpcMsg = {0};
-  rpcMsg.pCont = pReq;
-  rpcMsg.contLen = sizeof(SShowMsg);
-  rpcMsg.msgType = TSDB_MSG_TYPE_SHOW;
-
-  sendMsg(pClient, &rpcMsg);
-  SRpcMsg* pMsg = pClient->pRsp;
+  SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_SHOW, pReq, contLen);
   ASSERT_NE(pMsg, nullptr);
   ASSERT_EQ(pMsg->code, TSDB_CODE_MND_INVALID_MSG_TYPE);
 }
\ No newline at end of file

source/dnode/mgmt/impl/test/cluster/CMakeLists.txt

-add_executable(dnode_test_cluster "")
-
-target_sources(dnode_test_cluster
-    PRIVATE
-    "cluster.cpp"
-    "../sut/deploy.cpp"
-)
-
+aux_source_directory(. CLUSTER_SRC)
+add_executable(dnode_test_cluster ${CLUSTER_SRC})
 target_link_libraries(
     dnode_test_cluster
-    PUBLIC dnode
-    PUBLIC util
-    PUBLIC os
-    PUBLIC gtest_main
-)
-
-target_include_directories(dnode_test_cluster
-    PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/server/dnode/mgmt"
-    "${CMAKE_CURRENT_SOURCE_DIR}/../../inc"
-    "${CMAKE_CURRENT_SOURCE_DIR}/../sut"
+    PUBLIC sut
 )
 
 add_test(

source/dnode/mgmt/impl/test/cluster/cluster.cpp

@@ -9,162 +9,33 @@
  *
  */
 
-#include "deploy.h"
+#include "base.h"
 
 class DndTestCluster : public ::testing::Test {
  protected:
-  static SServer* CreateServer(const char* path, const char* fqdn, uint16_t port, const char* firstEp) {
-    SServer* pServer = createServer(path, fqdn, port, firstEp);
-    ASSERT(pServer);
-    return pServer;
-  }
+  static void SetUpTestSuite() { test.Init("/tmp/dnode_test_cluster", 9030); }
+  static void TearDownTestSuite() { test.Cleanup(); }
 
-  static void SetUpTestSuite() {
-    initLog("/tmp/tdlog");
-
-    const char* fqdn = "localhost";
-    const char* firstEp = "localhost:9030";
-    pServer = CreateServer("/tmp/dnode_test_cluster", fqdn, 9030, firstEp);
-    pClient = createClient("root", "taosdata", fqdn, 9030);
-    taosMsleep(1100);
-  }
-
-  static void TearDownTestSuite() {
-    stopServer(pServer);
-    dropClient(pClient);
-    pServer = NULL;
-    pClient = NULL;
-  }
-
-  static SServer* pServer;
-  static SClient* pClient;
-  static int32_t  connId;
+  static Testbase test;
 
  public:
   void SetUp() override {}
   void TearDown() override {}
-
-  void SendTheCheckShowMetaMsg(int8_t showType, const char* showName, int32_t columns, const char* db) {
-    SShowMsg* pShow = (SShowMsg*)rpcMallocCont(sizeof(SShowMsg));
-    pShow->type = showType;
-    if (db != NULL) {
-      strcpy(pShow->db, db);
-    }
-    SRpcMsg showRpcMsg = {0};
-    showRpcMsg.pCont = pShow;
-    showRpcMsg.contLen = sizeof(SShowMsg);
-    showRpcMsg.msgType = TSDB_MSG_TYPE_SHOW;
-
-    sendMsg(pClient, &showRpcMsg);
-    ASSERT_NE(pClient->pRsp, nullptr);
-    ASSERT_EQ(pClient->pRsp->code, 0);
-    ASSERT_NE(pClient->pRsp->pCont, nullptr);
-
-    SShowRsp* pShowRsp = (SShowRsp*)pClient->pRsp->pCont;
-    ASSERT_NE(pShowRsp, nullptr);
-    pShowRsp->showId = htonl(pShowRsp->showId);
-    pMeta = &pShowRsp->tableMeta;
-    pMeta->numOfTags = htonl(pMeta->numOfTags);
-    pMeta->numOfColumns = htonl(pMeta->numOfColumns);
-    pMeta->sversion = htonl(pMeta->sversion);
-    pMeta->tversion = htonl(pMeta->tversion);
-    pMeta->tuid = htobe64(pMeta->tuid);
-    pMeta->suid = htobe64(pMeta->suid);
-
-    showId = pShowRsp->showId;
-
-    EXPECT_NE(pShowRsp->showId, 0);
-    EXPECT_STREQ(pMeta->tbFname, showName);
-    EXPECT_EQ(pMeta->numOfTags, 0);
-    EXPECT_EQ(pMeta->numOfColumns, columns);
-    EXPECT_EQ(pMeta->precision, 0);
-    EXPECT_EQ(pMeta->tableType, 0);
-    EXPECT_EQ(pMeta->update, 0);
-    EXPECT_EQ(pMeta->sversion, 0);
-    EXPECT_EQ(pMeta->tversion, 0);
-    EXPECT_EQ(pMeta->tuid, 0);
-    EXPECT_EQ(pMeta->suid, 0);
-  }
-
-  void CheckSchema(int32_t index, int8_t type, int32_t bytes, const char* name) {
-    SSchema* pSchema = &pMeta->pSchema[index];
-    pSchema->bytes = htonl(pSchema->bytes);
-    EXPECT_EQ(pSchema->colId, 0);
-    EXPECT_EQ(pSchema->type, type);
-    EXPECT_EQ(pSchema->bytes, bytes);
-    EXPECT_STREQ(pSchema->name, name);
-  }
-
-  void SendThenCheckShowRetrieveMsg(int32_t rows) {
-    SRetrieveTableMsg* pRetrieve = (SRetrieveTableMsg*)rpcMallocCont(sizeof(SRetrieveTableMsg));
-    pRetrieve->showId = htonl(showId);
-    pRetrieve->free = 0;
-
-    SRpcMsg retrieveRpcMsg = {0};
-    retrieveRpcMsg.pCont = pRetrieve;
-    retrieveRpcMsg.contLen = sizeof(SRetrieveTableMsg);
-    retrieveRpcMsg.msgType = TSDB_MSG_TYPE_SHOW_RETRIEVE;
-
-    sendMsg(pClient, &retrieveRpcMsg);
-
-    ASSERT_NE(pClient->pRsp, nullptr);
-    ASSERT_EQ(pClient->pRsp->code, 0);
-    ASSERT_NE(pClient->pRsp->pCont, nullptr);
-
-    pRetrieveRsp = (SRetrieveTableRsp*)pClient->pRsp->pCont;
-    ASSERT_NE(pRetrieveRsp, nullptr);
-    pRetrieveRsp->numOfRows = htonl(pRetrieveRsp->numOfRows);
-    pRetrieveRsp->useconds = htobe64(pRetrieveRsp->useconds);
-    pRetrieveRsp->compLen = htonl(pRetrieveRsp->compLen);
-
-    EXPECT_EQ(pRetrieveRsp->numOfRows, rows);
-    EXPECT_EQ(pRetrieveRsp->useconds, 0);
-    // EXPECT_EQ(pRetrieveRsp->completed, completed);
-    EXPECT_EQ(pRetrieveRsp->precision, TSDB_TIME_PRECISION_MILLI);
-    EXPECT_EQ(pRetrieveRsp->compressed, 0);
-    EXPECT_EQ(pRetrieveRsp->compLen, 0);
-
-    pData = pRetrieveRsp->data;
-    pos = 0;
-  }
-
-  void CheckInt32() {
-    int32_t data = *((int32_t*)(pData + pos));
-    pos += sizeof(int32_t);
-    EXPECT_GT(data, 0);
-  }
-
-  void CheckTimestamp() {
-    int64_t data = *((int64_t*)(pData + pos));
-    pos += sizeof(int64_t);
-    EXPECT_GT(data, 0);
-  }
-
-  void CheckBinary(int32_t len) {
-    pos += sizeof(VarDataLenT);
-    char* data = (char*)(pData + pos);
-    pos += len;
-  }
-
-  int32_t            showId;
-  STableMetaMsg*     pMeta;
-  SRetrieveTableRsp* pRetrieveRsp;
-  char*              pData;
-  int32_t            pos;
 };
 
-SServer* DndTestCluster::pServer;
-SClient* DndTestCluster::pClient;
-int32_t  DndTestCluster::connId;
+Testbase DndTestCluster::test;
 
 TEST_F(DndTestCluster, 01_ShowCluster) {
-  SendTheCheckShowMetaMsg(TSDB_MGMT_TABLE_CLUSTER, "show cluster", 3, NULL);
-  CheckSchema(0, TSDB_DATA_TYPE_INT, 4, "id");
-  CheckSchema(1, TSDB_DATA_TYPE_BINARY, TSDB_CLUSTER_ID_LEN + VARSTR_HEADER_SIZE, "name");
-  CheckSchema(2, TSDB_DATA_TYPE_TIMESTAMP, 8, "create_time");
+  test.SendShowMetaMsg(TSDB_MGMT_TABLE_CLUSTER, "");
+  CHECK_META( "show cluster", 3);
+  CHECK_SCHEMA(0, TSDB_DATA_TYPE_INT, 4, "id");
+  CHECK_SCHEMA(1, TSDB_DATA_TYPE_BINARY, TSDB_CLUSTER_ID_LEN + VARSTR_HEADER_SIZE, "name");
+  CHECK_SCHEMA(2, TSDB_DATA_TYPE_TIMESTAMP, 8, "create_time");
+
+  test.SendShowRetrieveMsg();
+  EXPECT_EQ(test.GetShowRows(), 1);
 
-  SendThenCheckShowRetrieveMsg(1);
-  CheckInt32();
-  CheckBinary(TSDB_CLUSTER_ID_LEN);
+  IgnoreInt32();
+  IgnoreBinary(TSDB_CLUSTER_ID_LEN);
   CheckTimestamp();
 }
\ No newline at end of file

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

-add_executable(dnode_test_db "")
-
-target_sources(dnode_test_db
-    PRIVATE
-    "db.cpp"
-    "../sut/deploy.cpp"
-)
-
+aux_source_directory(. DB_SRC)
+add_executable(dnode_test_db ${DB_SRC})
 target_link_libraries(
     dnode_test_db
-    PUBLIC dnode
-    PUBLIC util
-    PUBLIC os
-    PUBLIC gtest_main
-)
-
-target_include_directories(dnode_test_db
-    PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/server/dnode/mgmt"
-    "${CMAKE_CURRENT_SOURCE_DIR}/../../inc"
-    "${CMAKE_CURRENT_SOURCE_DIR}/../sut"
+    PUBLIC sut
 )
 
 add_test(

source/dnode/mgmt/impl/test/db/db.cpp

@@ -9,199 +9,52 @@
  *
  */
 
-#include "deploy.h"
+#include "base.h"
 
 class DndTestDb : public ::testing::Test {
  protected:
-  static SServer* CreateServer(const char* path, const char* fqdn, uint16_t port, const char* firstEp) {
-    SServer* pServer = createServer(path, fqdn, port, firstEp);
-    ASSERT(pServer);
-    return pServer;
-  }
-
-  static void SetUpTestSuite() {
-    initLog("/tmp/tdlog");
-
-    const char* fqdn = "localhost";
-    const char* firstEp = "localhost:9040";
-    pServer = CreateServer("/tmp/dnode_test_db", fqdn, 9040, firstEp);
-    pClient = createClient("root", "taosdata", fqdn, 9040);
-    taosMsleep(1100);
-  }
-
-  static void TearDownTestSuite() {
-    stopServer(pServer);
-    dropClient(pClient);
-    pServer = NULL;
-    pClient = NULL;
-  }
+  static void SetUpTestSuite() { test.Init("/tmp/dnode_test_db", 9040); }
+  static void TearDownTestSuite() { test.Cleanup(); }
 
-  static SServer* pServer;
-  static SClient* pClient;
-  static int32_t  connId;
+  static Testbase test;
 
  public:
   void SetUp() override {}
   void TearDown() override {}
-
-  void SendTheCheckShowMetaMsg(int8_t showType, const char* showName, int32_t columns, const char* db) {
-    SShowMsg* pShow = (SShowMsg*)rpcMallocCont(sizeof(SShowMsg));
-    pShow->type = showType;
-    if (db != NULL) {
-      strcpy(pShow->db, db);
-    }
-    SRpcMsg showRpcMsg = {0};
-    showRpcMsg.pCont = pShow;
-    showRpcMsg.contLen = sizeof(SShowMsg);
-    showRpcMsg.msgType = TSDB_MSG_TYPE_SHOW;
-
-    sendMsg(pClient, &showRpcMsg);
-    ASSERT_NE(pClient->pRsp, nullptr);
-    ASSERT_EQ(pClient->pRsp->code, 0);
-    ASSERT_NE(pClient->pRsp->pCont, nullptr);
-
-    SShowRsp* pShowRsp = (SShowRsp*)pClient->pRsp->pCont;
-    ASSERT_NE(pShowRsp, nullptr);
-    pShowRsp->showId = htonl(pShowRsp->showId);
-    pMeta = &pShowRsp->tableMeta;
-    pMeta->numOfTags = htonl(pMeta->numOfTags);
-    pMeta->numOfColumns = htonl(pMeta->numOfColumns);
-    pMeta->sversion = htonl(pMeta->sversion);
-    pMeta->tversion = htonl(pMeta->tversion);
-    pMeta->tuid = htobe64(pMeta->tuid);
-    pMeta->suid = htobe64(pMeta->suid);
-
-    showId = pShowRsp->showId;
-
-    EXPECT_NE(pShowRsp->showId, 0);
-    EXPECT_STREQ(pMeta->tbFname, showName);
-    EXPECT_EQ(pMeta->numOfTags, 0);
-    EXPECT_EQ(pMeta->numOfColumns, columns);
-    EXPECT_EQ(pMeta->precision, 0);
-    EXPECT_EQ(pMeta->tableType, 0);
-    EXPECT_EQ(pMeta->update, 0);
-    EXPECT_EQ(pMeta->sversion, 0);
-    EXPECT_EQ(pMeta->tversion, 0);
-    EXPECT_EQ(pMeta->tuid, 0);
-    EXPECT_EQ(pMeta->suid, 0);
-  }
-
-  void CheckSchema(int32_t index, int8_t type, int32_t bytes, const char* name) {
-    SSchema* pSchema = &pMeta->pSchema[index];
-    pSchema->bytes = htonl(pSchema->bytes);
-    EXPECT_EQ(pSchema->colId, 0);
-    EXPECT_EQ(pSchema->type, type);
-    EXPECT_EQ(pSchema->bytes, bytes);
-    EXPECT_STREQ(pSchema->name, name);
-  }
-
-  void SendThenCheckShowRetrieveMsg(int32_t rows) {
-    SRetrieveTableMsg* pRetrieve = (SRetrieveTableMsg*)rpcMallocCont(sizeof(SRetrieveTableMsg));
-    pRetrieve->showId = htonl(showId);
-    pRetrieve->free = 0;
-
-    SRpcMsg retrieveRpcMsg = {0};
-    retrieveRpcMsg.pCont = pRetrieve;
-    retrieveRpcMsg.contLen = sizeof(SRetrieveTableMsg);
-    retrieveRpcMsg.msgType = TSDB_MSG_TYPE_SHOW_RETRIEVE;
-
-    sendMsg(pClient, &retrieveRpcMsg);
-
-    ASSERT_NE(pClient->pRsp, nullptr);
-    ASSERT_EQ(pClient->pRsp->code, 0);
-    ASSERT_NE(pClient->pRsp->pCont, nullptr);
-
-    pRetrieveRsp = (SRetrieveTableRsp*)pClient->pRsp->pCont;
-    ASSERT_NE(pRetrieveRsp, nullptr);
-    pRetrieveRsp->numOfRows = htonl(pRetrieveRsp->numOfRows);
-    pRetrieveRsp->useconds = htobe64(pRetrieveRsp->useconds);
-    pRetrieveRsp->compLen = htonl(pRetrieveRsp->compLen);
-
-    EXPECT_EQ(pRetrieveRsp->numOfRows, rows);
-    EXPECT_EQ(pRetrieveRsp->useconds, 0);
-    // EXPECT_EQ(pRetrieveRsp->completed, completed);
-    EXPECT_EQ(pRetrieveRsp->precision, TSDB_TIME_PRECISION_MILLI);
-    EXPECT_EQ(pRetrieveRsp->compressed, 0);
-    EXPECT_EQ(pRetrieveRsp->compLen, 0);
-
-    pData = pRetrieveRsp->data;
-    pos = 0;
-  }
-
-  void CheckInt8(int8_t val) {
-    int8_t data = *((int8_t*)(pData + pos));
-    pos += sizeof(int8_t);
-    EXPECT_EQ(data, val);
-  }
-
-  void CheckInt16(int16_t val) {
-    int16_t data = *((int16_t*)(pData + pos));
-    pos += sizeof(int16_t);
-    EXPECT_EQ(data, val);
-  }
-
-  void CheckInt32(int32_t val) {
-    int32_t data = *((int32_t*)(pData + pos));
-    pos += sizeof(int32_t);
-    EXPECT_EQ(data, val);
-  }
-
-  void CheckInt64(int64_t val) {
-    int64_t data = *((int64_t*)(pData + pos));
-    pos += sizeof(int64_t);
-    EXPECT_EQ(data, val);
-  }
-
-  void CheckTimestamp() {
-    int64_t data = *((int64_t*)(pData + pos));
-    pos += sizeof(int64_t);
-    EXPECT_GT(data, 0);
-  }
-
-  void CheckBinary(const char* val, int32_t len) {
-    pos += sizeof(VarDataLenT);
-    char* data = (char*)(pData + pos);
-    pos += len;
-    EXPECT_STREQ(data, val);
-  }
-
-  int32_t            showId;
-  STableMetaMsg*     pMeta;
-  SRetrieveTableRsp* pRetrieveRsp;
-  char*              pData;
-  int32_t            pos;
 };
 
-SServer* DndTestDb::pServer;
-SClient* DndTestDb::pClient;
-int32_t  DndTestDb::connId;
+Testbase DndTestDb::test;
 
 TEST_F(DndTestDb, 01_ShowDb) {
-  SendTheCheckShowMetaMsg(TSDB_MGMT_TABLE_DB, "show databases", 17, NULL);
-  CheckSchema(0, TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN - 1 + VARSTR_HEADER_SIZE, "name");
-  CheckSchema(1, TSDB_DATA_TYPE_TIMESTAMP, 8, "create_time");
-  CheckSchema(2, TSDB_DATA_TYPE_SMALLINT, 2, "vgroups");
-  CheckSchema(3, TSDB_DATA_TYPE_SMALLINT, 2, "replica");
-  CheckSchema(4, TSDB_DATA_TYPE_SMALLINT, 2, "quorum");
-  CheckSchema(5, TSDB_DATA_TYPE_SMALLINT, 2, "days");
-  CheckSchema(6, TSDB_DATA_TYPE_BINARY, 24 + VARSTR_HEADER_SIZE, "keep0,keep1,keep2");
-  CheckSchema(7, TSDB_DATA_TYPE_INT, 4, "cache");
-  CheckSchema(8, TSDB_DATA_TYPE_INT, 4, "blocks");
-  CheckSchema(9, TSDB_DATA_TYPE_INT, 4, "minrows");
-  CheckSchema(10, TSDB_DATA_TYPE_INT, 4, "maxrows");
-  CheckSchema(11, TSDB_DATA_TYPE_TINYINT, 1, "wallevel");
-  CheckSchema(12, TSDB_DATA_TYPE_INT, 4, "fsync");
-  CheckSchema(13, TSDB_DATA_TYPE_TINYINT, 1, "comp");
-  CheckSchema(14, TSDB_DATA_TYPE_TINYINT, 1, "cachelast");
-  CheckSchema(15, TSDB_DATA_TYPE_BINARY, 3 + VARSTR_HEADER_SIZE, "precision");
-  CheckSchema(16, TSDB_DATA_TYPE_TINYINT, 1, "update");
-
-  SendThenCheckShowRetrieveMsg(0);
+  test.SendShowMetaMsg(TSDB_MGMT_TABLE_DB, "");
+  CHECK_META("show databases", 17);
+  CHECK_SCHEMA(0, TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN - 1 + VARSTR_HEADER_SIZE, "name");
+  CHECK_SCHEMA(1, TSDB_DATA_TYPE_TIMESTAMP, 8, "create_time");
+  CHECK_SCHEMA(2, TSDB_DATA_TYPE_SMALLINT, 2, "vgroups");
+  CHECK_SCHEMA(3, TSDB_DATA_TYPE_SMALLINT, 2, "replica");
+  CHECK_SCHEMA(4, TSDB_DATA_TYPE_SMALLINT, 2, "quorum");
+  CHECK_SCHEMA(5, TSDB_DATA_TYPE_SMALLINT, 2, "days");
+  CHECK_SCHEMA(6, TSDB_DATA_TYPE_BINARY, 24 + VARSTR_HEADER_SIZE, "keep0,keep1,keep2");
+  CHECK_SCHEMA(7, TSDB_DATA_TYPE_INT, 4, "cache");
+  CHECK_SCHEMA(8, TSDB_DATA_TYPE_INT, 4, "blocks");
+  CHECK_SCHEMA(9, TSDB_DATA_TYPE_INT, 4, "minrows");
+  CHECK_SCHEMA(10, TSDB_DATA_TYPE_INT, 4, "maxrows");
+  CHECK_SCHEMA(11, TSDB_DATA_TYPE_TINYINT, 1, "wallevel");
+  CHECK_SCHEMA(12, TSDB_DATA_TYPE_INT, 4, "fsync");
+  CHECK_SCHEMA(13, TSDB_DATA_TYPE_TINYINT, 1, "comp");
+  CHECK_SCHEMA(14, TSDB_DATA_TYPE_TINYINT, 1, "cachelast");
+  CHECK_SCHEMA(15, TSDB_DATA_TYPE_BINARY, 3 + VARSTR_HEADER_SIZE, "precision");
+  CHECK_SCHEMA(16, TSDB_DATA_TYPE_TINYINT, 1, "update");
+
+  test.SendShowRetrieveMsg();
+  EXPECT_EQ(test.GetShowRows(), 0);
 }
 
 TEST_F(DndTestDb, 02_Create_Alter_Drop_Db) {
   {
-    SCreateDbMsg* pReq = (SCreateDbMsg*)rpcMallocCont(sizeof(SCreateDbMsg));
+    int32_t contLen = sizeof(SCreateDbMsg);
+
+    SCreateDbMsg* pReq = (SCreateDbMsg*)rpcMallocCont(contLen);
     strcpy(pReq->db, "1.d1");
     pReq->numOfVgroups = htonl(2);
     pReq->cacheBlockSize = htonl(16);
@@ -223,20 +76,16 @@ TEST_F(DndTestDb, 02_Create_Alter_Drop_Db) {
     pReq->cacheLastRow = 0;
     pReq->ignoreExist = 1;
 
-    SRpcMsg rpcMsg = {0};
-    rpcMsg.pCont = pReq;
-    rpcMsg.contLen = sizeof(SCreateDbMsg);
-    rpcMsg.msgType = TSDB_MSG_TYPE_CREATE_DB;
-
-    sendMsg(pClient, &rpcMsg);
-    SRpcMsg* pMsg = pClient->pRsp;
+    SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_CREATE_DB, pReq, contLen);
     ASSERT_NE(pMsg, nullptr);
     ASSERT_EQ(pMsg->code, 0);
-    // taosMsleep(1000000);
   }
 
-  SendTheCheckShowMetaMsg(TSDB_MGMT_TABLE_DB, "show databases", 17, NULL);
-  SendThenCheckShowRetrieveMsg(1);
+  test.SendShowMetaMsg(TSDB_MGMT_TABLE_DB, "");
+  CHECK_META("show databases", 17);
+
+  test.SendShowRetrieveMsg();
+  EXPECT_EQ(test.GetShowRows(), 1);
   CheckBinary("d1", TSDB_DB_NAME_LEN - 1);
   CheckTimestamp();
   CheckInt16(2);                      // vgroups
@@ -255,12 +104,15 @@ TEST_F(DndTestDb, 02_Create_Alter_Drop_Db) {
   CheckBinary("ms", 3);               // precision
   CheckInt8(0);                       // update
 
-  SendTheCheckShowMetaMsg(TSDB_MGMT_TABLE_VGROUP, "show vgroups", 4, "1.d1");
-  CheckSchema(0, TSDB_DATA_TYPE_INT, 4, "vgId");
-  CheckSchema(1, TSDB_DATA_TYPE_INT, 4, "tables");
-  CheckSchema(2, TSDB_DATA_TYPE_SMALLINT, 2, "v1_dnode");
-  CheckSchema(3, TSDB_DATA_TYPE_BINARY, 9 + VARSTR_HEADER_SIZE, "v1_status");
-  SendThenCheckShowRetrieveMsg(2);
+  test.SendShowMetaMsg(TSDB_MGMT_TABLE_VGROUP, "1.d1");
+  CHECK_META("show vgroups", 4);
+  CHECK_SCHEMA(0, TSDB_DATA_TYPE_INT, 4, "vgId");
+  CHECK_SCHEMA(1, TSDB_DATA_TYPE_INT, 4, "tables");
+  CHECK_SCHEMA(2, TSDB_DATA_TYPE_SMALLINT, 2, "v1_dnode");
+  CHECK_SCHEMA(3, TSDB_DATA_TYPE_BINARY, 9 + VARSTR_HEADER_SIZE, "v1_status");
+
+  test.SendShowRetrieveMsg();
+  EXPECT_EQ(test.GetShowRows(), 2);
   CheckInt32(1);
   CheckInt32(2);
   CheckInt32(0);
@@ -271,7 +123,9 @@ TEST_F(DndTestDb, 02_Create_Alter_Drop_Db) {
   CheckBinary("master", 9);
 
   {
-    SAlterDbMsg* pReq = (SAlterDbMsg*)rpcMallocCont(sizeof(SAlterDbMsg));
+    int32_t contLen = sizeof(SAlterDbMsg);
+
+    SAlterDbMsg* pReq = (SAlterDbMsg*)rpcMallocCont(contLen);
     strcpy(pReq->db, "1.d1");
     pReq->totalBlocks = htonl(12);
     pReq->daysToKeep0 = htonl(300);
@@ -282,19 +136,14 @@ TEST_F(DndTestDb, 02_Create_Alter_Drop_Db) {
     pReq->quorum = 2;
     pReq->cacheLastRow = 1;
 
-    SRpcMsg rpcMsg = {0};
-    rpcMsg.pCont = pReq;
-    rpcMsg.contLen = sizeof(SAlterDbMsg);
-    rpcMsg.msgType = TSDB_MSG_TYPE_ALTER_DB;
-
-    sendMsg(pClient, &rpcMsg);
-    SRpcMsg* pMsg = pClient->pRsp;
+    SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_ALTER_DB, pReq, contLen);
     ASSERT_NE(pMsg, nullptr);
     ASSERT_EQ(pMsg->code, 0);
   }
 
-  SendTheCheckShowMetaMsg(TSDB_MGMT_TABLE_DB, "show databases", 17, NULL);
-  SendThenCheckShowRetrieveMsg(1);
+  test.SendShowMetaMsg(TSDB_MGMT_TABLE_DB, "");
+  test.SendShowRetrieveMsg();
+  EXPECT_EQ(test.GetShowRows(), 1);
   CheckBinary("d1", TSDB_DB_NAME_LEN - 1);
   CheckTimestamp();
   CheckInt16(2);                   // vgroups
@@ -314,19 +163,14 @@ TEST_F(DndTestDb, 02_Create_Alter_Drop_Db) {
   CheckInt8(0);                    // update
 
   // restart
-  stopServer(pServer);
-  pServer = NULL;
-
-  uInfo("start all server");
+  test.Restart();
 
-  const char* fqdn = "localhost";
-  const char* firstEp = "localhost:9040";
-  pServer = startServer("/tmp/dnode_test_db", fqdn, 9040, firstEp);
+  test.SendShowMetaMsg(TSDB_MGMT_TABLE_DB, "");
+  CHECK_META("show databases", 17);
 
-  uInfo("all server is running");
+  test.SendShowRetrieveMsg();
+  EXPECT_EQ(test.GetShowRows(), 1);
 
-  SendTheCheckShowMetaMsg(TSDB_MGMT_TABLE_DB, "show databases", 17, NULL);
-  SendThenCheckShowRetrieveMsg(1);
   CheckBinary("d1", TSDB_DB_NAME_LEN - 1);
   CheckTimestamp();
   CheckInt16(2);                   // vgroups
@@ -346,27 +190,28 @@ TEST_F(DndTestDb, 02_Create_Alter_Drop_Db) {
   CheckInt8(0);                    // update
 
   {
-    SDropDbMsg* pReq = (SDropDbMsg*)rpcMallocCont(sizeof(SDropDbMsg));
-    strcpy(pReq->db, "1.d1");
+    int32_t contLen = sizeof(SDropDbMsg);
 
-    SRpcMsg rpcMsg = {0};
-    rpcMsg.pCont = pReq;
-    rpcMsg.contLen = sizeof(SDropDbMsg);
-    rpcMsg.msgType = TSDB_MSG_TYPE_DROP_DB;
+    SDropDbMsg* pReq = (SDropDbMsg*)rpcMallocCont(contLen);
+    strcpy(pReq->db, "1.d1");
 
-    sendMsg(pClient, &rpcMsg);
-    SRpcMsg* pMsg = pClient->pRsp;
+    SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_DROP_DB, pReq, contLen);
     ASSERT_NE(pMsg, nullptr);
     ASSERT_EQ(pMsg->code, 0);
   }
 
-  SendTheCheckShowMetaMsg(TSDB_MGMT_TABLE_DB, "show databases", 17, NULL);
-  SendThenCheckShowRetrieveMsg(0);
+  test.SendShowMetaMsg(TSDB_MGMT_TABLE_DB, "");
+  CHECK_META("show databases", 17);
+
+  test.SendShowRetrieveMsg();
+  EXPECT_EQ(test.GetShowRows(), 0);
 }
 
 TEST_F(DndTestDb, 03_Create_Use_Restart_Use_Db) {
   {
-    SCreateDbMsg* pReq = (SCreateDbMsg*)rpcMallocCont(sizeof(SCreateDbMsg));
+    int32_t contLen = sizeof(SCreateDbMsg);
+
+    SCreateDbMsg* pReq = (SCreateDbMsg*)rpcMallocCont(contLen);
     strcpy(pReq->db, "1.d2");
     pReq->numOfVgroups = htonl(2);
     pReq->cacheBlockSize = htonl(16);
@@ -388,33 +233,26 @@ TEST_F(DndTestDb, 03_Create_Use_Restart_Use_Db) {
     pReq->cacheLastRow = 0;
     pReq->ignoreExist = 1;
 
-    SRpcMsg rpcMsg = {0};
-    rpcMsg.pCont = pReq;
-    rpcMsg.contLen = sizeof(SCreateDbMsg);
-    rpcMsg.msgType = TSDB_MSG_TYPE_CREATE_DB;
-
-    sendMsg(pClient, &rpcMsg);
-    SRpcMsg* pMsg = pClient->pRsp;
+    SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_CREATE_DB, pReq, contLen);
     ASSERT_NE(pMsg, nullptr);
     ASSERT_EQ(pMsg->code, 0);
   }
 
-  SendTheCheckShowMetaMsg(TSDB_MGMT_TABLE_DB, "show databases", 17, NULL);
-  SendThenCheckShowRetrieveMsg(1);
+  test.SendShowMetaMsg(TSDB_MGMT_TABLE_DB, "");
+  CHECK_META("show databases", 17);
+
+  test.SendShowRetrieveMsg();
+  EXPECT_EQ(test.GetShowRows(), 1);
   CheckBinary("d2", TSDB_DB_NAME_LEN - 1);
 
   {
-    SUseDbMsg* pReq = (SUseDbMsg*)rpcMallocCont(sizeof(SUseDbMsg));
+    int32_t contLen = sizeof(SUseDbMsg);
+
+    SUseDbMsg* pReq = (SUseDbMsg*)rpcMallocCont(contLen);
     strcpy(pReq->db, "1.d2");
     pReq->vgVersion = htonl(-1);
 
-    SRpcMsg rpcMsg = {0};
-    rpcMsg.pCont = pReq;
-    rpcMsg.contLen = sizeof(SUseDbMsg);
-    rpcMsg.msgType = TSDB_MSG_TYPE_USE_DB;
-
-    sendMsg(pClient, &rpcMsg);
-    SRpcMsg* pMsg = pClient->pRsp;
+    SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_USE_DB, pReq, contLen);
     ASSERT_NE(pMsg, nullptr);
     ASSERT_EQ(pMsg->code, 0);
 

source/dnode/mgmt/impl/test/sut/CMakeLists.txt

+aux_source_directory(src SUT_SRC)
+add_library(sut STATIC ${SUT_SRC})
+target_link_libraries(
+    sut
+    PUBLIC dnode
+    PUBLIC util
+    PUBLIC os
+    PUBLIC gtest_main
+)
+
+target_include_directories(
+    sut
+    PUBLIC "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+)

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

@@ -138,7 +138,7 @@ static int32_t mndProcessShowMsg(SMnodeMsg *pMnodeMsg) {
   ShowMetaFp metaFp = pMgmt->metaFps[type];
   if (metaFp == NULL) {
     terrno = TSDB_CODE_MND_INVALID_MSG_TYPE;
-    mError("failed to process show-meta msg:%s since no message handle", mndShowStr(type));
+    mError("failed to process show-meta msg:%s since %s", mndShowStr(type), terrstr());
     return -1;
   }