Merge remote-tracking branch 'origin/3.0' into feature/qnode

include/common/taosmsg.h

@@ -82,8 +82,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" )
@@ -99,15 +98,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" )
@@ -294,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];
@@ -331,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];
@@ -356,6 +381,7 @@ typedef struct {
 typedef struct {
   SMsgHead head;
   char     name[TSDB_TABLE_FNAME_LEN];
+  int8_t   ignoreNotExists;
 } SDropTableMsg;
 
 typedef struct {
@@ -950,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;

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,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/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/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;
+  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;
+
+#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

@@ -93,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/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/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 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,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;
@@ -152,24 +151,63 @@ 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_MSG_TYPE_CREATE_USER || type == TSDB_MSG_TYPE_SHOW || type == TSDB_MSG_TYPE_DROP_USER ||
+        type == TSDB_MSG_TYPE_DROP_ACCT || type == TSDB_MSG_TYPE_CREATE_DB || type == TSDB_MSG_TYPE_CREATE_ACCT ||
+        type == TSDB_MSG_TYPE_CREATE_TABLE || type == TSDB_MSG_TYPE_CREATE_STB || type == TSDB_MSG_TYPE_USE_DB ||
+        type == TSDB_MSG_TYPE_DROP_DB || type == TSDB_MSG_TYPE_DROP_STB) {
       pRequest->type = type;
       pRequest->body.requestMsg = (SReqMsgInfo){.pMsg = output, .len = outputLen};
 
-      SRequestMsgBody body = {0};
-      buildRequestMsgFp[type](pRequest, &body);
+      SRequestMsgBody body = buildRequestMsgImpl(pRequest);
+      SEpSet* pEpSet = &pTscObj->pAppInfo->mgmtEp.epSet;
 
-      int64_t transporterId = 0;
-      sendMsgToServer(pTscObj->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &body, &transporterId);
+      if (type == TSDB_MSG_TYPE_CREATE_TABLE) {
+        struct SCatalog* pCatalog = NULL;
 
-      tsem_wait(&pRequest->body.rspSem);
+        char buf[12] = {0};
+        sprintf(buf, "%d", pTscObj->pAppInfo->clusterId);
+        code = catalogGetHandle(buf, &pCatalog);
+        if (code != 0) {
+          pRequest->code = code;
+          return pRequest;
+        }
+
+        SCreateTableMsg* pMsg = body.msgInfo.pMsg;
+
+        SName t = {0};
+        tNameFromString(&t, pMsg->name, T_NAME_ACCT|T_NAME_DB|T_NAME_TABLE);
 
+        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);
+
+        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 {
+        int64_t transporterId = 0;
+        sendMsgToServer(pTscObj->pTransporter, pEpSet, &body, &transporterId);
+      }
 
+      tsem_wait(&pRequest->body.rspSem);
       destroyRequestMsgBody(&body);
     } else {
       assert(0);
     }
+
+    tfree(c.db);
   }
 
   if (code != TSDB_CODE_SUCCESS) {
@@ -217,13 +255,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 +303,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 +425,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 +469,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

@@ -77,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(

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

-add_executable(dndTestProfile "")
-
-target_sources(dndTestProfile
-    PRIVATE
-    "profile.cpp"
-    "../sut/deploy.cpp"
-)
-
+aux_source_directory(. PROFILE_SRC)
+add_executable(dnode_test_profile ${PROFILE_SRC})
 target_link_libraries(
-    dndTestProfile
-    PUBLIC dnode
-    PUBLIC util
-    PUBLIC os
-    PUBLIC gtest_main
-)
-
-target_include_directories(dndTestProfile
-    PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/server/dnode/mgmt"
-    "${CMAKE_CURRENT_SOURCE_DIR}/../../inc"
-    "${CMAKE_CURRENT_SOURCE_DIR}/../sut"
+    dnode_test_profile
+    PUBLIC sut
 )
 
 add_test(
-    NAME dndTestProfile
-    COMMAND dndTestProfile
+    NAME dnode_test_profile
+    COMMAND dnode_test_profile
 )

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;
   }