Merge branch '3.0' of https://github.com/taosdata/TDengine into feature/3.0_mhli

include/common/taosmsg.h

@@ -52,6 +52,10 @@ 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_RES_READY, "res-ready" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_TASKS_STATUS, "tasks-status" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CANCEL_TASK, "cancel-task" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_TASK, "drop-task" )
+
 // 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" )	
@@ -79,7 +83,6 @@ 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_CONN, "kill-conn" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_HEARTBEAT, "heartbeat" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_SHOW, "show" )
@@ -290,13 +293,52 @@ 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 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    tableName[TSDB_TABLE_FNAME_LEN];
+  int16_t numOfColumns;
+  int16_t numOfTags;
+  int8_t  igExists;
+  int8_t  rspMeta;
+  char    schema[];
+} SCreateCTableMsg;
+
 typedef struct {
   char    name[TSDB_TABLE_FNAME_LEN];
   int8_t  igExists;
   int32_t numOfTags;
   int32_t numOfColumns;
   SSchema pSchema[];
-} SCreateStbMsg;
+} SCreateStbMsg, SCreateTableMsg;
 
 typedef struct {
   char   name[TSDB_TABLE_FNAME_LEN];
@@ -327,19 +369,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];
@@ -352,6 +381,7 @@ typedef struct {
 typedef struct {
   SMsgHead head;
   char     name[TSDB_TABLE_FNAME_LEN];
+  int8_t   ignoreNotExists;
 } SDropTableMsg;
 
 typedef struct {
@@ -946,18 +976,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;
@@ -1093,29 +1111,29 @@ typedef struct {
   /* data */
 } SUpdateTagValRsp;
 
-typedef struct SSchedulerQueryMsg {
+typedef struct SSubQueryMsg {
   uint64_t  schedulerId;
   uint64_t  queryId;
   uint64_t  taskId;
   uint32_t  contentLen;
   char      msg[];
-} SSchedulerQueryMsg;
+} SSubQueryMsg;
 
-typedef struct SSchedulerReadyMsg {
+typedef struct SResReadyMsg {
   uint64_t  schedulerId;
   uint64_t  queryId;
   uint64_t  taskId;
-} SSchedulerReadyMsg;
+} SResReadyMsg;
 
-typedef struct SSchedulerFetchMsg {
+typedef struct SResFetchMsg {
   uint64_t  schedulerId;
   uint64_t  queryId;
   uint64_t  taskId;
-} SSchedulerFetchMsg;
+} SResFetchMsg;
 
-typedef struct SSchedulerStatusMsg {
+typedef struct SSchTasksStatusMsg {
   uint64_t  schedulerId;
-} SSchedulerStatusMsg;
+} SSchTasksStatusMsg;
 
 typedef struct STaskStatus {
   uint64_t  queryId;
@@ -1129,11 +1147,17 @@ typedef struct SSchedulerStatusRsp {
 } SSchedulerStatusRsp;
 
 
-typedef struct SSchedulerCancelMsg {
+typedef struct STaskCancelMsg {
+  uint64_t  schedulerId;
+  uint64_t  queryId;
+  uint64_t  taskId;
+} STaskCancelMsg;
+
+typedef struct STaskDropMsg {
   uint64_t  schedulerId;
   uint64_t  queryId;
   uint64_t  taskId;
-} SSchedulerCancelMsg;
+} STaskDropMsg;
 
 
 #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/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,12 +104,21 @@ 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
@@ -115,19 +127,21 @@ typedef struct STqBufferItem {
   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;
@@ -141,7 +155,7 @@ typedef struct STqGroup {
   int32_t topicNum;
   STqList*     head;
   SList*       topicList;  // SList<STqTopic>
-  void*    returnMsg;  // SVReadMsg
+  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/catalog/catalog.h

@@ -115,14 +115,14 @@ int32_t catalogGetTableDistVgroup(struct SCatalog* pCatalog, void *pRpc, const S
 /**
  * Get a table's vgroup from its name's hash value.
  * @param pCatalog (input, got with catalogGetHandle)
- * @param pRpc (input, rpc object)
+ * @param pTransporter (input, rpc object)
  * @param pMgmtEps (input, mnode EPs)
  * @param pDBName (input, full db name)
  * @param pTableName (input, table name, NOT including db name)
  * @param vgInfo (output, vgroup info)
  * @return error code
  */
-int32_t catalogGetTableHashVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pDBName, const char* pTableName, SVgroupInfo* vgInfo);
+int32_t catalogGetTableHashVgroup(struct SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const char* pDBName, const char* pTableName, SVgroupInfo* vgInfo);
 
 
 /**

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;
+  int32_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;
+
+typedef struct SDclStmtInfo {
+  int16_t     nodeType;
+  int16_t     msgType;
+  char*       pMsg;
+  int32_t     msgLen;
+} SDclStmtInfo;
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_PARSENODES_H_*/

include/libs/parser/parser.h

@@ -20,129 +20,17 @@
 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;
-
-/**
- * True will be returned if the input sql string is insert, false otherwise.
- * @param pStr    sql string
- * @param length  length of the sql string
- * @return
- */
-bool qIsInsertSql(const char* pStr, size_t length);
+#include "parsenodes.h"
 
 typedef struct SParseContext {
-  const char* pAcctId;
-  const char* pDbname;
+  SParseBasicCtx  ctx;
   void            *pRpc;
-  const char* pClusterId;
-  struct SCatalog* pCatalog;
-  const SEpSet* pEpSet;
-  int64_t id;          // query id, generated by uuid generator
+  struct SCatalog *pCatalog;
+  const SEpSet    *pEpSet;
   int8_t           schemaAttached; // denote if submit block is built with table schema or not
-  const char* pSql;    // sql string
+  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.
+  char            *pMsg;           // extended error message if exists to help avoid the problem in sql statement.
   int32_t          msgLen;         // max length of the msg
 } SParseContext;
 
@@ -154,37 +42,9 @@ 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);
+int32_t qParseQuerySql(SParseContext* pContext, SQueryNode** pQuery);
 
-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;
-
-/**
- * Parse the insert sql statement.
- * @param pStr            sql string
- * @param length          length of the sql string
- * @param id              operator id, generated by uuid generator.
- * @param msg             extended error message if exists to help avoid the problem in sql statement.
- * @return                data in binary format to submit to vnode directly.
- */
- int32_t qParseInsertSql(SParseContext* pContext, struct SInsertStmtInfo** pInfo);
+bool qIsDclQuery(const SQueryNode* pQuery);
 
 /**
  * Convert a normal sql statement to only query tags information to enable that the subscribe client can be aware quickly of the true vgroup ids that
@@ -211,9 +71,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    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 sub plan, the optional execution node
+  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

@@ -25,12 +25,15 @@ extern "C" {
 #include "tlog.h"
 
 enum {
+  JOB_TASK_STATUS_NULL = 0,
   JOB_TASK_STATUS_NOT_START = 1,
   JOB_TASK_STATUS_EXECUTING,
+  JOB_TASK_STATUS_PARTIAL_SUCCEED,
   JOB_TASK_STATUS_SUCCEED,
   JOB_TASK_STATUS_FAILED,
   JOB_TASK_STATUS_CANCELLING,
-  JOB_TASK_STATUS_CANCELLED
+  JOB_TASK_STATUS_CANCELLED,
+  JOB_TASK_STATUS_DROPPING,
 };
 
 typedef struct STableComInfo {
@@ -90,15 +93,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 initQueryModuleMsgHandle();
 
-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

@@ -42,15 +42,17 @@ typedef struct {
 
 int32_t qWorkerInit(SQWorkerCfg *cfg, void **qWorkerMgmt);
 
-int32_t qWorkerProcessQueryMsg(void *qWorkerMgmt, SSchedulerQueryMsg *msg, SRpcMsg *rsp);
+int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SRpcMsg **rsp);
 
-int32_t qWorkerProcessReadyMsg(void *qWorkerMgmt, SSchedulerReadyMsg *msg, SRpcMsg *rsp);
+int32_t qWorkerProcessReadyMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SRpcMsg *rsp);
 
-int32_t qWorkerProcessStatusMsg(void *qWorkerMgmt, SSchedulerStatusMsg *msg, SRpcMsg *rsp);
+int32_t qWorkerProcessStatusMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SRpcMsg *rsp);
 
-int32_t qWorkerProcessFetchMsg(void *qWorkerMgmt, SSchedulerFetchMsg *msg, SRpcMsg *rsp);
+int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SRpcMsg *rsp);
 
-int32_t qWorkerProcessCancelMsg(void *qWorkerMgmt, SSchedulerCancelMsg *msg, SRpcMsg *rsp);
+int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SRpcMsg *rsp);
+
+int32_t qWorkerProcessDropMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SRpcMsg *rsp);
 
 void qWorkerDestroy(void **qWorkerMgmt);
 

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/taoserror.h

@@ -314,6 +314,11 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_QRY_INVALID_TIME_CONDITION    TAOS_DEF_ERROR_CODE(0, 0x070D)  //"invalid time condition")
 #define TSDB_CODE_QRY_SYS_ERROR                 TAOS_DEF_ERROR_CODE(0, 0x070E)  //"System error")
 #define TSDB_CODE_QRY_INVALID_INPUT             TAOS_DEF_ERROR_CODE(0, 0x070F)  //"invalid input")
+#define TSDB_CODE_QRY_SCH_NOT_EXIST             TAOS_DEF_ERROR_CODE(0, 0x0710)  //"Scheduler not exist")
+#define TSDB_CODE_QRY_TASK_NOT_EXIST            TAOS_DEF_ERROR_CODE(0, 0x0711)  //"Task not exist")
+#define TSDB_CODE_QRY_TASK_ALREADY_EXIST        TAOS_DEF_ERROR_CODE(0, 0x0712)  //"Task already exist")
+#define TSDB_CODE_QRY_RES_CACHE_NOT_EXIST       TAOS_DEF_ERROR_CODE(0, 0x0713)  //"Task result cache not exist")
+#define TSDB_CODE_QRY_TASK_CANCELLED            TAOS_DEF_ERROR_CODE(0, 0x0714)  //"Task cancelled")
 
 
 // grant

include/util/thash.h

@@ -33,6 +33,8 @@ typedef void (*_hash_free_fn_t)(void *);
 
 #define HASH_INDEX(v, c) ((v) & ((c)-1))
 
+#define HASH_NODE_EXIST(code) (code == -2)
+
 /**
  * murmur hash algorithm
  * @key  usually string

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

@@ -137,24 +137,26 @@ typedef struct SRequestMsgBody {
 
 extern SAppInfo   appInfo;
 extern int32_t    tscReqRef;
-extern void      *tscQhandle;
 extern int32_t    tscConnRef;
 
-extern int (*buildRequestMsgFp[TSDB_SQL_MAX])(SRequestObj *pRequest, SRequestMsgBody *pMsgBody);
-extern int (*handleRequestRspFp[TSDB_SQL_MAX])(SRequestObj *pRequest, const char* pMsg, int32_t msgLen);
+SRequestMsgBody buildRequestMsgImpl(SRequestObj *pRequest);
+extern int (*handleRequestRspFp[TSDB_MSG_TYPE_MAX])(SRequestObj *pRequest, const char* pMsg, int32_t msgLen);
 
 int   taos_init();
 
-void* createTscObj(const char* user, const char* auth, const char *ip, uint32_t port, SAppInstInfo* pAppInfo);
+void* createTscObj(const char* user, const char* auth, const char *db, SAppInstInfo* pAppInfo);
 void  destroyTscObj(void*pObj);
 
-void* createRequest(STscObj* pObj, __taos_async_fn_t fp, void* param, int32_t type);
+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,9 +13,11 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include <catalog.h>
 #include "clientInt.h"
 #include "clientLog.h"
 #include "os.h"
+#include "query.h"
 #include "taosmsg.h"
 #include "tcache.h"
 #include "tconfig.h"
@@ -23,7 +25,6 @@
 #include "tnote.h"
 #include "tref.h"
 #include "trpc.h"
-#include "tsched.h"
 #include "ttime.h"
 #include "ttimezone.h"
 
@@ -33,10 +34,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 +97,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;
@@ -131,7 +130,7 @@ void destroyTscObj(void *pObj) {
   tfree(pTscObj);
 }
 
-void* createTscObj(const char* user, const char* auth, const char *ip, uint32_t port, SAppInstInfo* pAppInfo) {
+void* createTscObj(const char* user, const char* auth, const char *db, SAppInstInfo* pAppInfo) {
   STscObj *pObj = (STscObj *)calloc(1, sizeof(STscObj));
   if (NULL == pObj) {
     terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
@@ -146,6 +145,10 @@ void* createTscObj(const char* user, const char* auth, const char *ip, uint32_t
   tstrncpy(pObj->user, user, sizeof(pObj->user));
   memcpy(pObj->pass, auth, TSDB_PASSWORD_LEN);
 
+  if (db != NULL) {
+    tstrncpy(pObj->db, db, tListLen(pObj->db));
+  }
+
   pthread_mutex_init(&pObj->mutex, NULL);
   pObj->id = taosAddRef(tscConnRef, pObj);
 
@@ -222,25 +225,19 @@ void taos_init_imp(void) {
 
   taosInitNotes();
   initMsgHandleFp();
+  initQueryModuleMsgHandle();
 
   rpcInit();
 
+  SCatalogCfg cfg = {.enableVgroupCache = true, .maxDBCacheNum = 100, .maxTblCacheNum = 100};
+  catalogInit(&cfg);
+
   tscDebug("starting to initialize TAOS driver, local ep: %s", tsLocalEp);
 
   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;
@@ -145,33 +144,66 @@ TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen) {
 
   tscDebugL("0x%"PRIx64" SQL: %s", pRequest->requestId, pRequest->sqlstr);
 
-  int32_t code = 0;
-  if (qIsInsertSql(pRequest->sqlstr, sqlLen)) {
-    // todo add
-  } else {
-    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) {
-      pRequest->type = type;
-      pRequest->body.requestMsg = (SReqMsgInfo){.pMsg = output, .len = outputLen};
+  SParseContext cxt = {
+    .ctx = {.requestId = pRequest->requestId, .acctId = pTscObj->acctId, .db = getConnectionDB(pTscObj)},
+    .pSql = pRequest->sqlstr,
+    .sqlLen = sqlLen,
+    .pMsg = pRequest->msgBuf,
+    .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE
+  };
+  SQueryNode* pQuery = NULL;
+  int32_t code = qParseQuerySql(&cxt, &pQuery);
+  if (qIsDclQuery(pQuery)) {
+    SDclStmtInfo* pDcl = (SDclStmtInfo*)pQuery;
+    pRequest->type = pDcl->msgType;
+    pRequest->body.requestMsg = (SReqMsgInfo){.pMsg = pDcl->pMsg, .len = pDcl->msgLen};
+
+    SRequestMsgBody body = buildRequestMsgImpl(pRequest);
+    SEpSet* pEpSet = &pTscObj->pAppInfo->mgmtEp.epSet;
+
+    if (pDcl->msgType == TSDB_MSG_TYPE_CREATE_TABLE) {
+      struct SCatalog* pCatalog = NULL;
+
+      char buf[12] = {0};
+      sprintf(buf, "%d", pTscObj->pAppInfo->clusterId);
+      code = catalogGetHandle(buf, &pCatalog);
+      if (code != 0) {
+        pRequest->code = code;
+        return pRequest;
+      }
 
-      SRequestMsgBody body = {0};
-      buildRequestMsgFp[type](pRequest, &body);
+      SCreateTableMsg* pMsg = body.msgInfo.pMsg;
 
-      int64_t transporterId = 0;
-      sendMsgToServer(pTscObj->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &body, &transporterId);
+      SName t = {0};
+      tNameFromString(&t, pMsg->name, T_NAME_ACCT|T_NAME_DB|T_NAME_TABLE);
 
-      tsem_wait(&pRequest->body.rspSem);
+      char db[TSDB_DB_NAME_LEN + TS_PATH_DELIMITER_LEN + TSDB_ACCT_ID_LEN] = {0};
+      tNameGetFullDbName(&t, db);
 
+      SVgroupInfo info = {0};
+      catalogGetTableHashVgroup(pCatalog, pTscObj->pTransporter, pEpSet, db, tNameGetTableName(&t), &info);
 
-      destroyRequestMsgBody(&body);
+      int64_t transporterId = 0;
+      SEpSet ep = {0};
+      ep.inUse = info.inUse;
+      ep.numOfEps = info.numOfEps;
+      for(int32_t i = 0; i < ep.numOfEps; ++i) {
+        ep.port[i] = info.epAddr[i].port;
+        tstrncpy(ep.fqdn[i], info.epAddr[i].fqdn, tListLen(ep.fqdn[i]));
+      }
+
+      sendMsgToServer(pTscObj->pTransporter, &ep, &body, &transporterId);
     } else {
-      assert(0);
+      int64_t transporterId = 0;
+      sendMsgToServer(pTscObj->pTransporter, pEpSet, &body, &transporterId);
     }
+
+    tsem_wait(&pRequest->body.rspSem);
+    destroyRequestMsgBody(&body);
   }
 
+  tfree(cxt.ctx.db);
+
   if (code != TSDB_CODE_SUCCESS) {
     pRequest->code = code;
     return pRequest;
@@ -217,13 +249,13 @@ int initEpSetFromCfg(const char *firstEp, const char *secondEp, SCorEpSet *pEpSe
 }
 
 STscObj* taosConnectImpl(const char *ip, const char *user, const char *auth, const char *db, uint16_t port, __taos_async_fn_t fp, void *param, SAppInstInfo* pAppInfo) {
-  STscObj *pTscObj = createTscObj(user, auth, ip, port, pAppInfo);
+  STscObj *pTscObj = createTscObj(user, auth, db, pAppInfo);
   if (NULL == pTscObj) {
     terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
     return pTscObj;
   }
 
-  SRequestObj *pRequest = createRequest(pTscObj, fp, param, TSDB_SQL_CONNECT);
+  SRequestObj *pRequest = createRequest(pTscObj, fp, param, TSDB_MSG_TYPE_CONNECT);
   if (pRequest == NULL) {
     destroyTscObj(pTscObj);
     terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
@@ -265,16 +297,11 @@ static int32_t buildConnectMsg(SRequestObj *pRequest, SRequestMsgBody* pMsgBody)
     return -1;
   }
 
-  // TODO refactor full_name
-  char *db;  // ugly code to move the space
-
   STscObj *pObj = pRequest->pTscObj;
-  pthread_mutex_lock(&pObj->mutex);
-  db = strstr(pObj->db, TS_PATH_DELIMITER);
 
-  db = (db == NULL) ? pObj->db : db + 1;
+  char* db = getConnectionDB(pObj);
   tstrncpy(pConnect->db, db, sizeof(pConnect->db));
-  pthread_mutex_unlock(&pObj->mutex);
+  tfree(db);
 
   pConnect->pid = htonl(appInfo.pid);
   pConnect->startTime = htobe64(appInfo.startTime);
@@ -392,10 +419,9 @@ void* doFetchRow(SRequestObj* pRequest) {
   SReqResultInfo* pResultInfo = &pRequest->body.resInfo;
 
   if (pResultInfo->pData == NULL || pResultInfo->current >= pResultInfo->numOfRows) {
-    pRequest->type = TSDB_SQL_RETRIEVE_MNODE;
+    pRequest->type = TSDB_MSG_TYPE_SHOW_RETRIEVE;
 
-    SRequestMsgBody body = {0};
-    buildRequestMsgFp[pRequest->type](pRequest, &body);
+    SRequestMsgBody body = buildRequestMsgImpl(pRequest);
 
     int64_t transporterId = 0;
     STscObj* pTscObj = pRequest->pTscObj;
@@ -437,8 +463,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,7 +16,6 @@
 #include <gtest/gtest.h>
 #include <taoserror.h>
 #include <iostream>
-#include "tglobal.h"
 #pragma GCC diagnostic ignored "-Wwrite-strings"
 
 #pragma GCC diagnostic ignored "-Wunused-function"
@@ -25,8 +24,22 @@
 
 #include "../inc/clientInt.h"
 #include "taos.h"
+#include "tglobal.h"
 
 namespace {
+void showDB(TAOS* pConn) {
+  TAOS_RES* pRes = taos_query(pConn, "show databases");
+  TAOS_ROW  pRow = NULL;
+
+  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
+  int32_t     numOfFields = taos_num_fields(pRes);
+
+  char str[512] = {0};
+  while ((pRow = taos_fetch_row(pRes)) != NULL) {
+    int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
+    printf("%s\n", str);
+  }
+}
 }  // namespace
 
 int main(int argc, char** argv) {
@@ -34,92 +47,208 @@ int main(int argc, char** argv) {
   return RUN_ALL_TESTS();
 }
 
-TEST(testCase, driverInit_Test) {
-  taos_init();
-}
+TEST(testCase, driverInit_Test) { taos_init(); }
 
-TEST(testCase, connect_Test) {
-  TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
+// TEST(testCase, connect_Test) {
+//  TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
+////  assert(pConn != NULL);
+//  taos_close(pConn);
+//}
+//
+// TEST(testCase, create_user_Test) {
+//  TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
 //  assert(pConn != NULL);
-  taos_close(pConn);
-}
-
-TEST(testCase, create_user_Test) {
-  TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
+//
+//  TAOS_RES* pRes = taos_query(pConn, "create user abc pass 'abc'");
+//  if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
+//    printf("failed to create user, reason:%s\n", taos_errstr(pRes));
+//  }
+//
+//  taos_free_result(pRes);
+//  taos_close(pConn);
+//}
+//
+// TEST(testCase, create_account_Test) {
+//  TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
 //  assert(pConn != NULL);
+//
+//  TAOS_RES* pRes = taos_query(pConn, "create account aabc pass 'abc'");
+//  if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
+//    printf("failed to create user, reason:%s\n", taos_errstr(pRes));
+//  }
+//
+//  taos_free_result(pRes);
+//  taos_close(pConn);
+//}
+//
+// TEST(testCase, drop_account_Test) {
+//  TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
+//  assert(pConn != NULL);
+//
+//  TAOS_RES* pRes = taos_query(pConn, "drop account aabc");
+//  if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
+//    printf("failed to create user, reason:%s\n", taos_errstr(pRes));
+//  }
+//
+//  taos_free_result(pRes);
+//  taos_close(pConn);
+//}
+//
+// TEST(testCase, show_user_Test) {
+//  TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
+////  assert(pConn != NULL);
+//
+//  TAOS_RES* pRes = taos_query(pConn, "show users");
+//  TAOS_ROW pRow = NULL;
+//
+//  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
+//  int32_t numOfFields = taos_num_fields(pRes);
+//
+//  char str[512] = {0};
+//  while((pRow = taos_fetch_row(pRes)) != NULL) {
+//    int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
+//    printf("%s\n", str);
+//  }
+//
+//  taos_close(pConn);
+//}
+//
+// TEST(testCase, drop_user_Test) {
+//  TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
+//  assert(pConn != NULL);
+//
+//  TAOS_RES* pRes = taos_query(pConn, "drop user abc");
+//  if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
+//    printf("failed to create user, reason:%s\n", taos_errstr(pRes));
+//  }
+//
+//  taos_free_result(pRes);
+//  taos_close(pConn);
+//}
+//
+// TEST(testCase, show_db_Test) {
+//  TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
+////  assert(pConn != NULL);
+//
+//  TAOS_RES* pRes = taos_query(pConn, "show databases");
+//  TAOS_ROW pRow = NULL;
+//
+//  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
+//  int32_t numOfFields = taos_num_fields(pRes);
+//
+//  char str[512] = {0};
+//  while((pRow = taos_fetch_row(pRes)) != NULL) {
+//    int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
+//    printf("%s\n", str);
+//  }
+//
+//  taos_close(pConn);
+//}
 
-  TAOS_RES* pRes = taos_query(pConn, "create user abc pass 'abc'");
-  if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
-    printf("failed to create user, reason:%s\n", taos_errstr(pRes));
-  }
-
-  taos_free_result(pRes);
-  taos_close(pConn);
-}
-
-TEST(testCase, create_account_Test) {
+TEST(testCase, create_db_Test) {
   TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
   assert(pConn != NULL);
 
-  TAOS_RES* pRes = taos_query(pConn, "create account aabc pass 'abc'");
-  if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
-    printf("failed to create user, reason:%s\n", taos_errstr(pRes));
+  TAOS_RES* pRes = taos_query(pConn, "create database abc1");
+  if (taos_errno(pRes) != 0) {
+    printf("error in create db, reason:%s\n", taos_errstr(pRes));
   }
 
-  taos_free_result(pRes);
-  taos_close(pConn);
-}
-
-TEST(testCase, drop_account_Test) {
-  TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
+  ASSERT_TRUE(pFields == NULL);
 
-  TAOS_RES* pRes = taos_query(pConn, "drop account aabc");
-  if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
-    printf("failed to create user, reason:%s\n", taos_errstr(pRes));
-  }
+  int32_t numOfFields = taos_num_fields(pRes);
+  ASSERT_EQ(numOfFields, 0);
 
-  taos_free_result(pRes);
   taos_close(pConn);
 }
 
-TEST(testCase, show_user_Test) {
+TEST(testCase, use_db_test) {
   TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
-//  assert(pConn != NULL);
+  assert(pConn != NULL);
 
-  TAOS_RES* pRes = taos_query(pConn, "show users");
-  TAOS_ROW pRow = NULL;
+  TAOS_RES* pRes = taos_query(pConn, "use abc1");
+  if (taos_errno(pRes) != 0) {
+    printf("error in use db, reason:%s\n", taos_errstr(pRes));
+  }
 
   TAOS_FIELD* pFields = taos_fetch_fields(pRes);
-  int32_t numOfFields = taos_num_fields(pRes);
+  ASSERT_TRUE(pFields == NULL);
 
-  char str[512] = {0};
-  while((pRow = taos_fetch_row(pRes)) != NULL) {
-    int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
-    printf("%s\n", str);
-  }
+  int32_t numOfFields = taos_num_fields(pRes);
+  ASSERT_EQ(numOfFields, 0);
 
   taos_close(pConn);
 }
 
-TEST(testCase, drop_user_Test) {
+TEST(testCase, drop_db_test) {
   TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
   assert(pConn != NULL);
 
-  TAOS_RES* pRes = taos_query(pConn, "drop user abc");
-  if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
-    printf("failed to create user, reason:%s\n", taos_errstr(pRes));
+  showDB(pConn);
+
+  TAOS_RES* pRes = taos_query(pConn, "drop database abc1");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to drop db, reason:%s\n", taos_errstr(pRes));
   }
 
   taos_free_result(pRes);
+
+  showDB(pConn);
   taos_close(pConn);
 }
 
-TEST(testCase, show_db_Test) {
-  TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
+// TEST(testCase, create_stable_Test) {
+//  TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
 //  assert(pConn != NULL);
+//
+//  TAOS_RES* pRes = taos_query(pConn, "create database abc1");
+//  if (taos_errno(pRes) != 0) {
+//    printf("error in create db, reason:%s\n", taos_errstr(pRes));
+//  }
+//  taos_free_result(pRes);
+//
+//  pRes = taos_query(pConn, "use abc1");
+//  if (taos_errno(pRes) != 0) {
+//    printf("error in use db, reason:%s\n", taos_errstr(pRes));
+//  }
+//  taos_free_result(pRes);
+//
+//  pRes = taos_query(pConn, "create stable st1(ts timestamp, k int) tags(a int)");
+//  if (taos_errno(pRes) != 0) {
+//    printf("error in create stable, reason:%s\n", taos_errstr(pRes));
+//  }
+//
+//  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
+//  ASSERT_TRUE(pFields == NULL);
+//
+//  int32_t numOfFields = taos_num_fields(pRes);
+//  ASSERT_EQ(numOfFields, 0);
+//
+//  taos_free_result(pRes);
+//  taos_close(pConn);
+//}
+
+TEST(testCase, create_table_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 table tm0(ts timestamp, k int)");
+  //  taos_free_result(pRes);
+  //
+  //  taos_close(pConn);
+}
 
-  TAOS_RES* pRes = taos_query(pConn, "show databases");
+TEST(testCase, create_ctable_Test) {}
+
+TEST(testCase, show_stable_Test) {
+  TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
+  assert(pConn != NULL);
+
+  TAOS_RES* pRes = taos_query(pConn, "show stables");
   TAOS_ROW pRow = NULL;
 
   TAOS_FIELD* pFields = taos_fetch_fields(pRes);
@@ -131,20 +260,45 @@ TEST(testCase, show_db_Test) {
     printf("%s\n", str);
   }
 
+  taos_free_result(pRes);
+
   taos_close(pConn);
 }
 
-TEST(testCase, create_db_Test) {
+TEST(testCase, drop_stable_Test) {
   TAOS* pConn = taos_connect("ubuntu", "root", "taosdata", NULL, 0);
   assert(pConn != NULL);
 
   TAOS_RES* pRes = taos_query(pConn, "create database abc1");
+  if (taos_errno(pRes) != 0) {
+    printf("error in creating db, reason:%s\n", taos_errstr(pRes));
+  }
+  taos_free_result(pRes);
 
-  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
-  ASSERT_TRUE(pFields == NULL);
+  pRes = taos_query(pConn, "use abc1");
+  if (taos_errno(pRes) != 0) {
+    printf("error in using db, reason:%s\n", taos_errstr(pRes));
+  }
+  taos_free_result(pRes);
 
-  int32_t numOfFields = taos_num_fields(pRes);
-  ASSERT_EQ(numOfFields, 0);
+  pRes = taos_query(pConn, "drop stable st1");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to drop stable, reason:%s\n", taos_errstr(pRes));
+  }
 
+  taos_free_result(pRes);
   taos_close(pConn);
 }
+
+//TEST(testCase, show_table_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, "show tables");
+//  taos_free_result(pRes);
+//
+//  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

@@ -45,6 +45,10 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
   pMgmt->msgFp[TSDB_MSG_TYPE_MQ_CONNECT] = dndProcessVnodeWriteMsg;
   pMgmt->msgFp[TSDB_MSG_TYPE_MQ_DISCONNECT] = dndProcessVnodeWriteMsg;
   pMgmt->msgFp[TSDB_MSG_TYPE_MQ_SET_CUR] = dndProcessVnodeWriteMsg;
+  pMgmt->msgFp[TSDB_MSG_TYPE_RES_READY] = dndProcessVnodeFetchMsg;
+  pMgmt->msgFp[TSDB_MSG_TYPE_TASKS_STATUS] = dndProcessVnodeFetchMsg;
+  pMgmt->msgFp[TSDB_MSG_TYPE_CANCEL_TASK] = dndProcessVnodeFetchMsg;
+  pMgmt->msgFp[TSDB_MSG_TYPE_DROP_TASK] = dndProcessVnodeFetchMsg;
 
   // msg from client to mnode
   pMgmt->msgFp[TSDB_MSG_TYPE_CONNECT] = dndProcessMnodeReadMsg;
@@ -73,7 +77,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/dnode/CMakeLists.txt

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