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

include/common/taosmsg.h

@@ -94,15 +94,15 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_NETWORK_TEST, "nettest" )
 
 // message from mnode to vnode
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_STB_IN, "create-stb-internal" )
-TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_STB_IN, "alter-stb-internal" )	
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_STB_IN, "alter-stb-internal" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_STB_IN, "drop-stb-internal" )
 // message from mnode to mnode
 // message from mnode to qnode
 // message from mnode to dnode
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_VNODE_IN, "create-vnode-internal" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_VNODE_IN, "alter-vnode-internal" )
-TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_VNODE_IN, "drop-vnode-internal" )	
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_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_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" )
@@ -289,6 +289,37 @@ typedef struct SSchema {
   char    name[TSDB_COL_NAME_LEN];
 } SSchema;
 
+typedef struct {
+  int32_t  contLen;
+  int32_t  vgId;
+  int8_t   tableType;
+  int16_t  numOfColumns;
+  int16_t  numOfTags;
+  int32_t  tid;
+  int32_t  sversion;
+  int32_t  tversion;
+  int32_t  tagDataLen;
+  int32_t  sqlDataLen;
+  uint64_t uid;
+  uint64_t superTableUid;
+  uint64_t createdTime;
+  char     tableFname[TSDB_TABLE_FNAME_LEN];
+  char     stbFname[TSDB_TABLE_FNAME_LEN];
+  char     data[];
+} SMDCreateTableMsg;
+
+typedef struct {
+  int32_t len;  // one create table message
+  char    tableName[TSDB_TABLE_FNAME_LEN];
+  int16_t numOfTags;
+  int16_t numOfColumns;
+  int16_t sqlLen;  // the length of SQL, it starts after schema , sql is a null-terminated string
+  int8_t  igExists;
+  int8_t  rspMeta;
+  int8_t  reserved[16];
+  char    schema[];
+} SCreateTableMsg;
+
 typedef struct {
   char    name[TSDB_TABLE_FNAME_LEN];
   int8_t  igExists;
@@ -326,19 +357,6 @@ typedef struct {
   uint64_t suid;
 } SDropStbInternalMsg;
 
-typedef struct {
-  SMsgHead head;
-  char     name[TSDB_TABLE_FNAME_LEN];
-  char     stbFname[TSDB_TABLE_FNAME_LEN];
-  int8_t   tableType;
-  uint64_t suid;
-  int32_t  sversion;
-  int32_t  numOfTags;
-  int32_t  numOfColumns;
-  int32_t  tagDataLen;
-  char     data[];
-} SCreateTableMsg;
-
 typedef struct {
   SMsgHead head;
   char     name[TSDB_TABLE_FNAME_LEN];

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;
+  STqExec*  executor;
-  int32_t  status;
+  int32_t   status;
-  int64_t  size;
+  int64_t   size;
-  void*    content;
+  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;
+  void*      logReader;
-  int32_t    tail;
   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 clientId;
-  int64_t  cgId;
+  int64_t cgId;
-  void*    ahandle;
+  void*   ahandle;
-  int32_t  topicNum;
+  int32_t topicNum;
-  STqList* head;
+  STqList*     head;
-  SList*   topicList;  // SList<STqTopic>
+  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/parser/parsenodes.h

@@ -0,0 +1,167 @@
+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef _TD_PARSENODES_H_
+#define _TD_PARSENODES_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "catalog.h"
+#include "common.h"
+#include "function.h"
+#include "tmsgtype.h"
+#include "tname.h"
+#include "tvariant.h"
+
+/*
+ * The first field of a node of any type is guaranteed to be the int16_t.
+ * Hence the type of any node can be gotten by casting it to SQueryNode. 
+ */
+typedef struct SQueryNode {
+  int16_t type;
+} SQueryNode;
+
+#define nodeType(nodeptr) (((const SQueryNode*)(nodeptr))->type)
+
+typedef struct SField {
+  char     name[TSDB_COL_NAME_LEN];
+  uint8_t  type;
+  int16_t  bytes;
+} SField;
+
+typedef struct SParseBasicCtx {
+  const char *db;
+  int32_t     acctId;
+  uint64_t    requestId;
+} SParseBasicCtx;
+
+typedef struct SFieldInfo {
+  int16_t     numOfOutput;   // number of column in result
+  SField     *final;
+  SArray     *internalField; // SArray<SInternalField>
+} SFieldInfo;
+
+typedef struct SCond {
+  uint64_t uid;
+  int32_t  len;  // length of tag query condition data
+  char *   cond;
+} SCond;
+
+typedef struct SJoinNode {
+  uint64_t uid;
+  int16_t  tagColId;
+  SArray*  tsJoin;
+  SArray*  tagJoin;
+} SJoinNode;
+
+typedef struct SJoinInfo {
+  bool       hasJoin;
+  SJoinNode *joinTables[TSDB_MAX_JOIN_TABLE_NUM];
+} SJoinInfo;
+
+typedef struct STagCond {
+  int16_t   relType;     // relation between tbname list and query condition, including : TK_AND or TK_OR
+  SCond     tbnameCond;  // tbname query condition, only support tbname query condition on one table
+  SJoinInfo joinInfo;    // join condition, only support two tables join currently
+  SArray   *pCond;       // for different table, the query condition must be seperated
+} STagCond;
+
+typedef struct STableMetaInfo {
+  STableMeta    *pTableMeta;      // table meta, cached in client side and acquired by name
+  SVgroupsInfo  *vgroupList;
+  SName         name;
+  char          aliasName[TSDB_TABLE_NAME_LEN];    // alias name of table specified in query sql
+  SArray       *tagColList;                        // SArray<SColumn*>, involved tag columns
+} STableMetaInfo;
+
+typedef struct SColumnIndex {
+  int16_t tableIndex;
+  int16_t columnIndex;
+  int16_t type;               // normal column/tag/ user input constant column
+} SColumnIndex;
+
+// select statement
+typedef struct SQueryStmtInfo {
+  int16_t          command;       // the command may be different for each subclause, so keep it seperately.
+  uint32_t         type;          // query/insert type
+  STimeWindow      window;        // the whole query time window
+  SInterval        interval;      // tumble time window
+  SSessionWindow   sessionWindow; // session time window
+  SStateWindow     stateWindow;   // state window query
+  SGroupbyExpr     groupbyExpr;   // groupby tags info
+  SArray *         colList;       // SArray<SColumn*>
+  SFieldInfo       fieldsInfo;
+  SArray**         exprList;      // SArray<SExprInfo*>
+  SLimit           limit;
+  SLimit           slimit;
+  STagCond         tagCond;
+  SArray *         colCond;
+  SArray *         order;
+  int16_t          numOfTables;
+  int16_t          curTableIdx;
+  STableMetaInfo **pTableMetaInfo;
+  struct STSBuf   *tsBuf;
+
+  int16_t          fillType;      // final result fill type
+  int64_t *        fillVal;       // default value for fill
+  int32_t          numOfFillVal;  // fill value size
+
+  char *           msg;           // pointer to the pCmd->payload to keep error message temporarily
+  int64_t          clauseLimit;   // limit for current sub clause
+
+  int64_t          prjOffset;     // offset value in the original sql expression, only applied at client side
+  int64_t          vgroupLimit;    // table limit in case of super table projection query + global order + limit
+
+  int32_t          udColumnId;    // current user-defined constant output field column id, monotonically decreases from TSDB_UD_COLUMN_INDEX
+  int32_t          bufLen;
+  char*            buf;
+  SArray          *pUdfInfo;
+
+  struct SQueryStmtInfo *sibling;     // sibling
+  struct SQueryStmtInfo *pDownstream;
+  SMultiFunctionsDesc    info;
+  SArray            *pUpstream;   // SArray<struct SQueryStmtInfo>
+  int32_t            havingFieldNum;
+  int32_t            exprListLevelIndex;
+} SQueryStmtInfo;
+
+typedef enum {
+  PAYLOAD_TYPE_KV = 0,
+  PAYLOAD_TYPE_RAW = 1,
+} EPayloadType;
+
+typedef struct SVgDataBlocks {
+  SVgroupInfo vg;
+  int32_t     numOfTables;  // number of tables in current submit block
+  uint32_t    size;
+  char       *pData;        // SMsgDesc + SSubmitMsg + SSubmitBlk + ...
+} SVgDataBlocks;
+
+typedef struct SInsertStmtInfo {
+  int16_t     nodeType;
+  SArray*     pDataBlocks;         // data block for each vgroup, SArray<SVgDataBlocks*>.
+  int8_t      schemaAttache;        // denote if submit block is built with table schema or not
+  uint8_t     payloadType;         // EPayloadType. 0: K-V payload for non-prepare insert, 1: rawPayload for prepare insert
+  uint32_t    insertType;          // insert data from [file|sql statement| bound statement]
+  const char* sql;                 // current sql statement position
+} SInsertStmtInfo;
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_PARSENODES_H_*/

include/libs/parser/parser.h

@@ -20,108 +20,7 @@
 extern "C" {
 #endif
 
-#include "catalog.h"
+#include "parsenodes.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.
@@ -132,18 +31,16 @@ struct SInsertStmtInfo;
 bool qIsInsertSql(const char* pStr, size_t length);
 
 typedef struct SParseContext {
-  const char* pAcctId;
+  SParseBasicCtx  ctx;
-  const char* pDbname;
+  void            *pRpc;
-  void *pRpc;
+  struct SCatalog *pCatalog;
-  const char* pClusterId;
+  const SEpSet    *pEpSet;
-  struct SCatalog* pCatalog;
+  int64_t          id;             // query id, generated by uuid generator
-  const SEpSet* pEpSet;
+  int8_t           schemaAttached; // denote if submit block is built with table schema or not
-  int64_t id;          // query id, generated by uuid generator
+  const char      *pSql;           // sql string
-  int8_t schemaAttached; // denote if submit block is built with table schema or not
+  size_t           sqlLen;         // length of the sql string
-  const char* pSql;    // sql string
+  char            *pMsg;           // extended error message if exists to help avoid the problem in sql statement.
-  size_t sqlLen;       // length of the sql string
+  int32_t          msgLen;         // max length of the msg
-  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,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);
+int32_t qParseQuerySql(const char* pStr, size_t length, SParseBasicCtx* pParseCtx, 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;
 
 /**
  * Parse the insert sql statement.
@@ -211,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
+  int32_t   numOfTables;
-  // todo data field
+  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.
+  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    *pChildern;    // the datasource subplan,from which to fetch the result
-  SArray   *pParents;     // the data destination subplan, get data from current subplan
+  SArray    *pParents;     // the data destination subplan, get data from current subplan
-  SPhyNode *pNode;        // physical plan of 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

@@ -90,16 +90,28 @@ 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);
 
+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();
+void msgInit();
+
 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);
 
-SSchema* tGetTbnameColumnSchema();
-extern void msgInit();
-
-extern int32_t qDebugFlag;
-
 #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)
 #define qWarn(...)   do { if (qDebugFlag & DEBUG_WARN)  { taosPrintLog("QRY WARN ", 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

@@ -148,13 +148,16 @@ int   taos_init();
 void* createTscObj(const char* user, const char* auth, const char *ip, uint32_t port, SAppInstInfo* pAppInfo);
 void  destroyTscObj(void*pObj);
 
-void* createRequest(STscObj* pObj, __taos_async_fn_t fp, void* param, int32_t type);
+void *createRequest(STscObj* pObj, __taos_async_fn_t fp, void* param, int32_t type);
-void destroyRequest(SRequestObj* pRequest);
+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/clientEnv.c

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

source/client/src/clientImpl.c

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

source/client/src/clientMain.c

@@ -1,16 +1,11 @@
+#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;
+  cleanupTaskQueue();
-  tscQhandle = NULL;
-  taosCleanUpScheduler(p);
 
   id = tscConnRef;
   tscConnRef = -1;
@@ -262,3 +255,9 @@ const char *taos_data_type(int type) {
     default: return "UNKNOWN";
   }
 }
+
+const char *taos_get_client_info() { return version; }
+
+int taos_affected_rows(TAOS_RES *res) { return 1; }
+
+int taos_result_precision(TAOS_RES *res) { return TSDB_TIME_PRECISION_MILLI; }

source/client/test/clientTests.cpp

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

source/common/src/tname.c

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

source/dnode/vnode/tq/CMakeLists.txt

@@ -12,6 +12,7 @@ target_link_libraries(
     PUBLIC os
     PUBLIC util
     PUBLIC common
+    PUBLIC transport
 )
 
 if(${BUILD_TEST})

source/dnode/vnode/tq/inc/tqInt.h

@@ -18,6 +18,7 @@
 
 #include "tq.h"
 #include "tlog.h"
+#include "trpc.h"
 #ifdef __cplusplus
 extern "C" {
 #endif

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

@@ -35,7 +35,22 @@ void* tqSerializeItem(STqMsgItem* pItem, void* ptr);
 const void* tqDeserializeTopic(const void* pBytes, STqTopic* pTopic);
 const void* tqDeserializeItem(const void* pBytes, STqMsgItem* pItem);
 
-STQ* tqOpen(const char* path, STqCfg* tqConfig, STqLogReader* tqLogReader, SMemAllocatorFactory* allocFac) {
+int tqInit() {
+  int8_t old = atomic_val_compare_exchange_8(&tqMgmt.inited, 0, 1);
+  if(old == 1) return 0;
+
+  tqMgmt.timer = taosTmrInit(0, 0, 0, "TQ");
+  return 0;
+}
+
+void tqCleanUp() {
+  int8_t old = atomic_val_compare_exchange_8(&tqMgmt.inited, 1, 0);
+  if(old == 0) return;
+  taosTmrStop(tqMgmt.timer);
+  taosTmrCleanUp(tqMgmt.timer);
+}
+
+STQ* tqOpen(const char* path, STqCfg* tqConfig, STqLogHandle* tqLogHandle, SMemAllocatorFactory* allocFac) {
   STQ* pTq = malloc(sizeof(STQ));
   if (pTq == NULL) {
     terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
@@ -43,9 +58,9 @@ STQ* tqOpen(const char* path, STqCfg* tqConfig, STqLogReader* tqLogReader, SMemA
   }
   pTq->path = strdup(path);
   pTq->tqConfig = tqConfig;
-  pTq->tqLogReader = tqLogReader;
+  pTq->tqLogHandle = tqLogHandle;
 #if 0
-  pTq->tqMemRef.pAlloctorFactory = allocFac;
+  pTq->tqMemRef.pAllocatorFactory = allocFac;
   pTq->tqMemRef.pAllocator = allocFac->create(allocFac);
   if (pTq->tqMemRef.pAllocator == NULL) {
     // TODO: error code of buffer pool
@@ -53,16 +68,24 @@ STQ* tqOpen(const char* path, STqCfg* tqConfig, STqLogReader* tqLogReader, SMemA
 #endif
   pTq->tqMeta = tqStoreOpen(path, (FTqSerialize)tqSerializeGroup, (FTqDeserialize)tqDeserializeGroup, free, 0);
   if (pTq->tqMeta == NULL) {
-    // TODO: free STQ
+    free(pTq);
+#if 0
+    allocFac->destroy(allocFac, pTq->tqMemRef.pAllocator);
+#endif
     return NULL;
   }
+
   return pTq;
 }
+
 void tqClose(STQ* pTq) {
   // TODO
 }
 
-static int tqProtoCheck(STqMsgHead* pMsg) { return pMsg->protoVer == 0; }
+static int tqProtoCheck(STqMsgHead* pMsg) {
+  // TODO
+  return pMsg->protoVer == 0;
+}
 
 static int tqAckOneTopic(STqTopic* pTopic, STqOneAck* pAck, STqQueryMsg** ppQuery) {
   // clean old item and move forward
@@ -126,6 +149,13 @@ int tqCreateGroup(STQ* pTq, int64_t topicId, int64_t cgId, int64_t cId, STqGroup
   *ppGroup = pGroup;
   memset(pGroup, 0, sizeof(STqGroup));
 
+  pGroup->topicList = tdListNew(sizeof(STqTopic));
+  if(pGroup->topicList == NULL) {
+    free(pGroup);
+    return -1;
+  }
+  *ppGroup = pGroup;
+
   return 0;
 }
 
@@ -154,46 +184,55 @@ int tqDropGroup(STQ* pTq, int64_t topicId, int64_t cgId, int64_t cId) {
   return 0;
 }
 
-static int tqFetch(STqGroup* pGroup, void** msg) {
+static int tqFetch(STqGroup* pGroup, STqConsumeRsp** pRsp) {
-  STqList* head = pGroup->head;
+  STqList* pHead = pGroup->head;
-  STqList* node = head;
+  STqList* pNode = pHead;
   int      totSize = 0;
+  int      numOfMsgs = 0;
   // TODO: make it a macro
-  int            sizeLimit = 4 * 1024;
+  int   sizeLimit = 4 * 1024;
-  STqMsgContent* buffer = malloc(sizeLimit);
+
-  if (buffer == NULL) {
+  void* ptr = realloc(*pRsp, sizeof(STqConsumeRsp) + sizeLimit);
-    // TODO:memory insufficient
+  if (ptr == NULL) {
+    terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
     return -1;
   }
+  *pRsp = ptr;
+  STqMsgContent* buffer = (*pRsp)->msgs;
+
   // iterate the list to get msgs of all topics
   // until all topic iterated or msgs over sizeLimit
-  while (node->next) {
+  while (pNode->next) {
-    node = node->next;
+    pNode = pNode->next;
-    STqTopic* topicHandle = &node->topic;
+    STqTopic* pTopic = &pNode->topic;
-    int       idx = topicHandle->nextConsumeOffset % TQ_BUFFER_SIZE;
+    int       idx = pTopic->nextConsumeOffset % TQ_BUFFER_SIZE;
-    if (topicHandle->buffer[idx].content != NULL && topicHandle->buffer[idx].offset == topicHandle->nextConsumeOffset) {
+    if (pTopic->buffer[idx].content != NULL && pTopic->buffer[idx].offset == pTopic->nextConsumeOffset) {
-      totSize += topicHandle->buffer[idx].size;
+      totSize += pTopic->buffer[idx].size;
       if (totSize > sizeLimit) {
-        void* ptr = realloc(buffer, totSize);
+        void* ptr = realloc(*pRsp, sizeof(STqConsumeRsp) + totSize);
         if (ptr == NULL) {
-          totSize -= topicHandle->buffer[idx].size;
+          totSize -= pTopic->buffer[idx].size;
-          // TODO:memory insufficient
+          terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
           // return msgs already copied
           break;
         }
+        *pRsp = ptr;
+        break;
       }
-      *((int64_t*)buffer) = topicHandle->topicId;
+      *((int64_t*)buffer) = pTopic->topicId;
       buffer = POINTER_SHIFT(buffer, sizeof(int64_t));
-      *((int64_t*)buffer) = topicHandle->buffer[idx].size;
+      *((int64_t*)buffer) = pTopic->buffer[idx].size;
       buffer = POINTER_SHIFT(buffer, sizeof(int64_t));
-      memcpy(buffer, topicHandle->buffer[idx].content, topicHandle->buffer[idx].size);
+      memcpy(buffer, pTopic->buffer[idx].content, pTopic->buffer[idx].size);
-      buffer = POINTER_SHIFT(buffer, topicHandle->buffer[idx].size);
+      buffer = POINTER_SHIFT(buffer, pTopic->buffer[idx].size);
+      numOfMsgs++;
       if (totSize > sizeLimit) {
         break;
       }
     }
   }
-  return totSize;
+  (*pRsp)->bodySize = totSize;
+  return numOfMsgs;
 }
 
 STqGroup* tqGetGroup(STQ* pTq, int64_t clientId) { return tqHandleGet(pTq->tqMeta, clientId); }
@@ -275,7 +314,22 @@ int tqSetCursor(STQ* pTq, STqSetCurReq* pMsg) {
   return 0;
 }
 
-int tqConsume(STQ* pTq, STqConsumeReq* pMsg) {
+// temporary
+int tqProcessCMsg(STQ* pTq, STqConsumeReq* pMsg, STqRspHandle* pRsp) {
+  int64_t   clientId = pMsg->head.clientId;
+  STqGroup* pGroup = tqGetGroup(pTq, clientId);
+  if (pGroup == NULL) {
+    terrno = TSDB_CODE_TQ_GROUP_NOT_SET;
+    return -1;
+  }
+  pGroup->rspHandle.handle = pRsp->handle;
+  pGroup->rspHandle.ahandle = pRsp->ahandle;
+
+  return 0;
+}
+
+int tqConsume(STQ* pTq, SRpcMsg* pReq, SRpcMsg** pRsp) {
+  STqConsumeReq *pMsg = pReq->pCont;
   int64_t   clientId = pMsg->head.clientId;
   STqGroup* pGroup = tqGetGroup(pTq, clientId);
   if (pGroup == NULL) {
@@ -283,20 +337,107 @@ int tqConsume(STQ* pTq, STqConsumeReq* pMsg) {
     return -1;
   }
 
-  STqConsumeRsp* pRsp = (STqConsumeRsp*)pMsg;
+  SList* topicList = pGroup->topicList;
-  int            numOfMsgs = tqFetch(pGroup, (void**)&pRsp->msgs);
+
-  if (numOfMsgs < 0) {
+  int totSize = 0;
+  int numOfMsgs = 0;
+  int sizeLimit = 4096;
+
+
+  STqConsumeRsp *pCsmRsp = (*pRsp)->pCont;
+  void* ptr = realloc((*pRsp)->pCont, sizeof(STqConsumeRsp) + sizeLimit);
+  if (ptr == NULL) {
+    terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
     return -1;
   }
-  if (numOfMsgs == 0) {
+  (*pRsp)->pCont = ptr;
+
+  SListIter iter;
+  tdListInitIter(topicList, &iter, TD_LIST_FORWARD);
+
+  STqMsgContent* buffer = NULL;
+  SArray* pArray = taosArrayInit(0, sizeof(void*));
+
+  SListNode *pn;
+  while((pn = tdListNext(&iter)) != NULL) {
+    STqTopic* pTopic = *(STqTopic**)pn->data;
+    int idx = pTopic->floatingCursor % TQ_BUFFER_SIZE;
+    STqMsgItem* pItem = &pTopic->buffer[idx];
+    if (pItem->content != NULL && pItem->offset == pTopic->floatingCursor) {
+      if(pItem->status == TQ_ITEM_READY) {
+        //if has data
+        totSize += pTopic->buffer[idx].size;
+        if (totSize > sizeLimit) {
+          void* ptr = realloc((*pRsp)->pCont, sizeof(STqConsumeRsp) + totSize);
+          if (ptr == NULL) {
+            totSize -= pTopic->buffer[idx].size;
+            terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
+            // return msgs already copied
+            break;
+          }
+          (*pRsp)->pCont = ptr;
+          break;
+        }
+        *((int64_t*)buffer) = htobe64(pTopic->topicId);
+        buffer = POINTER_SHIFT(buffer, sizeof(int64_t));
+        *((int64_t*)buffer) = htobe64(pTopic->buffer[idx].size);
+        buffer = POINTER_SHIFT(buffer, sizeof(int64_t));
+        memcpy(buffer, pTopic->buffer[idx].content, pTopic->buffer[idx].size);
+        buffer = POINTER_SHIFT(buffer, pTopic->buffer[idx].size);
+        numOfMsgs++;
+        if (totSize > sizeLimit) {
+          break;
+        }
+      } else if(pItem->status == TQ_ITEM_PROCESS) {
+        //if not have data but in process
+
+      } else if(pItem->status == TQ_ITEM_EMPTY){
+        //if not have data and not in process
+        int32_t old = atomic_val_compare_exchange_32(&pItem->status, TQ_ITEM_EMPTY, TQ_ITEM_PROCESS);
+        if(old != TQ_ITEM_EMPTY) {
+          continue;
+        }
+        pItem->offset = pTopic->floatingCursor;
+        taosArrayPush(pArray, &pItem);
+      } else {
+        ASSERT(0);
+      }
+    }
+  }
+
+  if (numOfMsgs > 0) {
+    // set code and other msg
+    rpcSendResponse(*pRsp);
+  } else {
     // most recent data has been fetched
 
     // enable timer for blocking wait
-    // once new data written during wait time
+    // once new data written when waiting, launch query and rsp
-    // launch query and response
   }
 
   // fetched a num of msgs, rpc response
+  for(int i = 0; i < pArray->size; i++) {
+    STqMsgItem* pItem = taosArrayGet(pArray, i);
+
+    //read from wal
+    void* raw = NULL;
+    /*int code = pTq->tqLogReader->logRead(, &raw, pItem->offset);*/
+    int code = pTq->tqLogHandle->logRead(pItem->pTopic->logReader, &raw, pItem->offset);
+    if(code < 0) {
+      //TODO: error
+    }
+    //get msgType
+    //if submitblk
+    pItem->executor->assign(pItem->executor->runtimeEnv, raw);
+    SSDataBlock* content = pItem->executor->exec(pItem->executor->runtimeEnv);
+    pItem->content = content;
+    //if other type, send just put into buffer
+    /*pItem->content = raw;*/
+
+    int32_t old = atomic_val_compare_exchange_32(&pItem->status, TQ_ITEM_PROCESS, TQ_ITEM_READY);
+    ASSERT(old == TQ_ITEM_PROCESS);
+  }
+  taosArrayDestroy(pArray);
 
   return 0;
 }
@@ -378,10 +519,10 @@ void* tqSerializeTopic(STqTopic* pTopic, void* ptr) {
   ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
   *(int64_t*)ptr = pTopic->topicId;
   ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
-  *(int32_t*)ptr = pTopic->head;
+  /**(int32_t*)ptr = pTopic->head;*/
-  ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
+  /*ptr = POINTER_SHIFT(ptr, sizeof(int32_t));*/
-  *(int32_t*)ptr = pTopic->tail;
+  /**(int32_t*)ptr = pTopic->tail;*/
-  ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
+  /*ptr = POINTER_SHIFT(ptr, sizeof(int32_t));*/
   for (int i = 0; i < TQ_BUFFER_SIZE; i++) {
     ptr = tqSerializeItem(&pTopic->buffer[i], ptr);
   }
@@ -435,10 +576,10 @@ const void* tqDeserializeTopic(const void* pBytes, STqTopic* topic) {
   ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
   topic->topicId = *(int64_t*)ptr;
   ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
-  topic->head = *(int32_t*)ptr;
+  /*topic->head = *(int32_t*)ptr;*/
-  ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
+  /*ptr = POINTER_SHIFT(ptr, sizeof(int32_t));*/
-  topic->tail = *(int32_t*)ptr;
+  /*topic->tail = *(int32_t*)ptr;*/
-  ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
+  /*ptr = POINTER_SHIFT(ptr, sizeof(int32_t));*/
   for (int i = 0; i < TQ_BUFFER_SIZE; i++) {
     ptr = tqDeserializeItem(ptr, &topic->buffer[i]);
   }

source/libs/catalog/src/catalog.c

@@ -605,7 +605,7 @@ int32_t catalogGetAllMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* p
       SName *name = taosArrayGet(pReq->pTableName, i);
       STableMeta *pTableMeta = NULL;
       
-      snprintf(dbName, sizeof(dbName), "%s.%s", name->acctId, name->dbname);
+      snprintf(dbName, sizeof(dbName), "%d.%s", name->acctId, name->dbname);
 
       CTG_ERR_JRET(catalogGetTableMeta(pCatalog, pRpc, pMgmtEps, dbName, name->tname, &pTableMeta));
 

source/libs/index/inc/index_tfile.h

@@ -47,7 +47,6 @@ typedef struct TFileHeader {
 typedef struct TFileCacheKey {
   uint64_t suid;
   uint8_t  colType;
-  int32_t  version;
   char*    colName;
   int32_t  nColName;
 } TFileCacheKey;
@@ -67,6 +66,7 @@ typedef struct TFileWriter {
   uint32_t    offset;
 } TFileWriter;
 
+// multi reader and single write
 typedef struct TFileReader {
   T_REF_DECLARE()
   Fst*        fst;

source/libs/index/src/index_tfile.c

@@ -38,11 +38,14 @@ static int tfileWriteHeader(TFileWriter* writer);
 static int tfileWriteFstOffset(TFileWriter* tw, int32_t offset);
 static int tfileWriteData(TFileWriter* write, TFileValue* tval);
 
-static int tfileReadLoadHeader(TFileReader* reader);
+static int  tfileReadLoadHeader(TFileReader* reader);
-static int tfileReadLoadFst(TFileReader* reader);
+static int  tfileReadLoadFst(TFileReader* reader);
-static int tfileReadLoadTableIds(TFileReader* reader, int32_t offset, SArray* result);
+static int  tfileReadLoadTableIds(TFileReader* reader, int32_t offset, SArray* result);
+static void tfileReadRef(TFileReader* reader);
+static void tfileReadUnRef(TFileReader* reader);
 
 static int  tfileGetFileList(const char* path, SArray* result);
+static int  tfileRmExpireFile(SArray* result);
 static void tfileDestroyFileName(void* elem);
 static int  tfileCompare(const void* a, const void* b);
 static int  tfileParseFileName(const char* filename, uint64_t* suid, int* colId, int* version);
@@ -58,6 +61,8 @@ TFileCache* tfileCacheCreate(const char* path) {
   SArray* files = taosArrayInit(4, sizeof(void*));
   tfileGetFileList(path, files);
   taosArraySort(files, tfileCompare);
+  tfileRmExpireFile(files);
+
   uint64_t suid;
   int32_t  colId, version;
   for (size_t i = 0; i < taosArrayGetSize(files); i++) {
@@ -66,29 +71,29 @@ TFileCache* tfileCacheCreate(const char* path) {
       indexInfo("try parse invalid file:  %s, skip it", file);
       continue;
     }
+
     WriterCtx* wc = writerCtxCreate(TFile, file, true, 1024 * 64);
     if (wc == NULL) {
       indexError("failed to open index:  %s", file);
       goto End;
     }
+
     TFileReader* reader = tfileReaderCreate(wc);
     if (0 != tfileReadLoadHeader(reader)) {
       tfileReaderDestroy(reader);
       indexError("failed to load index header, index file: %s", file);
       goto End;
     }
+
     if (0 != tfileReadLoadFst(reader)) {
       tfileReaderDestroy(reader);
       indexError("failed to load index fst, index file: %s", file);
+      goto End;
     }
+    tfileReadRef(reader);
     // loader fst and validate it
-
     TFileHeader*  header = &reader->header;
-    TFileCacheKey key = {.suid = header->suid,
+    TFileCacheKey key = {.suid = header->suid, .colName = header->colName, .nColName = strlen(header->colName), .colType = header->colType};
-                         .version = header->version,
-                         .colName = header->colName,
-                         .nColName = strlen(header->colName),
-                         .colType = header->colType};
 
     char buf[128] = {0};
     tfileSerialCacheKey(&key, buf);
@@ -110,7 +115,7 @@ void tfileCacheDestroy(TFileCache* tcache) {
     TFileReader* p = *reader;
     indexInfo("drop table cache suid: %" PRIu64 ", colName: %s, colType: %d", p->header.suid, p->header.colName, p->header.colType);
 
-    tfileReaderDestroy(p);
+    tfileReadUnRef(p);
     reader = taosHashIterate(tcache->tableCache, reader);
   }
   taosHashCleanup(tcache->tableCache);
@@ -120,12 +125,24 @@ void tfileCacheDestroy(TFileCache* tcache) {
 TFileReader* tfileCacheGet(TFileCache* tcache, TFileCacheKey* key) {
   char buf[128] = {0};
   tfileSerialCacheKey(key, buf);
+
   TFileReader* reader = taosHashGet(tcache->tableCache, buf, strlen(buf));
+  tfileReadRef(reader);
+
   return reader;
 }
 void tfileCachePut(TFileCache* tcache, TFileCacheKey* key, TFileReader* reader) {
   char buf[128] = {0};
   tfileSerialCacheKey(key, buf);
+  // remove last version index reader
+  TFileReader** p = taosHashGet(tcache->tableCache, buf, strlen(buf));
+  if (*p != NULL) {
+    TFileReader* oldReader = *p;
+    taosHashRemove(tcache->tableCache, buf, strlen(buf));
+    tfileReadUnRef(oldReader);
+  }
+
+  tfileReadRef(reader);
   taosHashPut(tcache->tableCache, buf, strlen(buf), &reader, sizeof(void*));
   return;
 }
@@ -147,25 +164,29 @@ void tfileReaderDestroy(TFileReader* reader) {
 }
 
 int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SArray* result) {
-  SIndexTerm* term = query->term;
+  SIndexTerm*     term = query->term;
+  EIndexQueryType qtype = query->qType;
 
+  int ret = -1;
   // refactor to callback later
-  if (query->qType == QUERY_TERM) {
+  if (qtype == QUERY_TERM) {
     uint64_t offset;
     FstSlice key = fstSliceCreate(term->colVal, term->nColVal);
     if (fstGet(reader->fst, &key, &offset)) {
-      return tfileReadLoadTableIds(reader, offset, result);
+      indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, found table info in tindex", term->suid, term->colName, term->colVal);
+      ret = tfileReadLoadTableIds(reader, offset, result);
     } else {
-      indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, not found in tindex", term->suid, term->colName, term->colVal);
+      indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, not found table info in tindex", term->suid, term->colName, term->colVal);
     }
-    return 0;
+    fstSliceDestroy(&key);
-  } else if (query->qType == QUERY_PREFIX) {
+  } else if (qtype == QUERY_PREFIX) {
     // handle later
     //
   } else {
     // handle later
   }
-  return 0;
+  tfileReadUnRef(reader);
+  return ret;
 }
 
 TFileWriter* tfileWriterCreate(WriterCtx* ctx, TFileHeader* header) {
@@ -209,7 +230,6 @@ int tfileWriterPut(TFileWriter* tw, void* data) {
     int32_t tbsz = taosArrayGetSize(v->tableId);
     fstOffset += TF_TABLE_TATOAL_SIZE(tbsz);
   }
-  // check result or not
   tfileWriteFstOffset(tw, fstOffset);
 
   for (size_t i = 0; i < sz; i++) {
@@ -237,6 +257,7 @@ int tfileWriterPut(TFileWriter* tw, void* data) {
     // write reversed data in buf to tindex
     tw->ctx->write(tw->ctx, buf, offset);
   }
+  tfree(buf);
 
   // write fst
   for (size_t i = 0; i < sz; i++) {
@@ -244,11 +265,8 @@ int tfileWriterPut(TFileWriter* tw, void* data) {
     TFileValue* v = taosArrayGetP((SArray*)data, i);
     if (tfileWriteData(tw, v) == 0) {
       //
-      //
     }
   }
-
-  tfree(buf);
   return 0;
 }
 void tfileWriterDestroy(TFileWriter* tw) {
@@ -270,17 +288,19 @@ void IndexTFileDestroy(IndexTFile* tfile) {
 }
 
 int indexTFileSearch(void* tfile, SIndexTermQuery* query, SArray* result) {
-  if (tfile == NULL) { return -1; }
+  int ret = -1;
+  if (tfile == NULL) { return ret; }
   IndexTFile* pTfile = (IndexTFile*)tfile;
 
   SIndexTerm*   term = query->term;
-  TFileCacheKey key = {.suid = term->suid, .colType = term->colType, .version = 0, .colName = term->colName, .nColName = term->nColName};
+  TFileCacheKey key = {.suid = term->suid, .colType = term->colType, .colName = term->colName, .nColName = term->nColName};
+  TFileReader*  reader = tfileCacheGet(pTfile->cache, &key);
 
-  TFileReader* reader = tfileCacheGet(pTfile->cache, &key);
   return tfileReaderSearch(reader, query, result);
 }
 int indexTFilePut(void* tfile, SIndexTerm* term, uint64_t uid) {
-  TFileWriterOpt wOpt = {.suid = term->suid, .colType = term->colType, .colName = term->colName, .nColName = term->nColName, .version = 1};
+  // TFileWriterOpt wOpt = {.suid = term->suid, .colType = term->colType, .colName = term->colName, .nColName = term->nColName, .version =
+  // 1};
 
   return 0;
 }
@@ -309,8 +329,7 @@ static int tfileWriteFstOffset(TFileWriter* tw, int32_t offset) {
   return 0;
 }
 static int tfileWriteHeader(TFileWriter* writer) {
-  char  buf[TFILE_HEADER_NO_FST] = {0};
+  char buf[TFILE_HEADER_NO_FST] = {0};
-  char* p = buf;
 
   TFileHeader* header = &writer->header;
   memcpy(buf, (char*)header, sizeof(buf));
@@ -338,8 +357,7 @@ static int tfileWriteData(TFileWriter* write, TFileValue* tval) {
 }
 static int tfileReadLoadHeader(TFileReader* reader) {
   // TODO simple tfile header later
-  char  buf[TFILE_HEADER_SIZE] = {0};
+  char buf[TFILE_HEADER_SIZE] = {0};
-  char* p = buf;
 
   int64_t nread = reader->ctx->read(reader->ctx, buf, sizeof(buf));
   assert(nread == sizeof(buf));
@@ -368,20 +386,32 @@ static int tfileReadLoadFst(TFileReader* reader) {
 static int tfileReadLoadTableIds(TFileReader* reader, int32_t offset, SArray* result) {
   int32_t    nid;
   WriterCtx* ctx = reader->ctx;
-  int32_t    nread = ctx->readFrom(ctx, (char*)&nid, sizeof(nid), offset);
+
+  int32_t nread = ctx->readFrom(ctx, (char*)&nid, sizeof(nid), offset);
   assert(sizeof(nid) == nread);
 
-  char* buf = calloc(1, sizeof(uint64_t) * nid);
+  int32_t total = sizeof(uint64_t) * nid;
+  char*   buf = calloc(1, total);
   if (buf == NULL) { return -1; }
 
-  nread = ctx->read(ctx, buf, sizeof(uint64_t) * nid);
+  nread = ctx->read(ctx, buf, total);
-  uint64_t* ids = (uint64_t*)buf;
+  assert(total == nread);
+
   for (int32_t i = 0; i < nid; i++) {
-    taosArrayPush(result, ids + i);
+    taosArrayPush(result, (uint64_t*)buf + i);
   }
   free(buf);
   return 0;
 }
+static void tfileReadRef(TFileReader* reader) {
+  int ref = T_REF_INC(reader);
+  UNUSED(ref);
+}
+
+static void tfileReadUnRef(TFileReader* reader) {
+  int ref = T_REF_DEC(reader);
+  if (ref == 0) { tfileReaderDestroy(reader); }
+}
 
 static int tfileGetFileList(const char* path, SArray* result) {
   DIR* dir = opendir(path);
@@ -397,6 +427,10 @@ static int tfileGetFileList(const char* path, SArray* result) {
   closedir(dir);
   return 0;
 }
+static int tfileRmExpireFile(SArray* result) {
+  // TODO(yihao): remove expire tindex after restart
+  return 0;
+}
 static void tfileDestroyFileName(void* elem) {
   char* p = *(char**)elem;
   free(p);
@@ -423,7 +457,5 @@ static void tfileSerialCacheKey(TFileCacheKey* key, char* buf) {
   SERIALIZE_VAR_TO_BUF(buf, '_', char);
   SERIALIZE_MEM_TO_BUF(buf, key, colType);
   SERIALIZE_VAR_TO_BUF(buf, '_', char);
-  SERIALIZE_MEM_TO_BUF(buf, key, version);
-  SERIALIZE_VAR_TO_BUF(buf, '_', char);
   SERIALIZE_STR_MEM_TO_BUF(buf, key, colName, key->nColName);
 }

source/libs/parser/inc/astToMsg.h

@@ -9,5 +9,6 @@ SCreateAcctMsg* buildAcctManipulationMsg(SSqlInfo* pInfo, int32_t* outputLen, in
 SDropUserMsg* buildDropUserMsg(SSqlInfo* pInfo, int32_t* outputLen, int64_t id, char* msgBuf, int32_t msgLen);
 SShowMsg* buildShowMsg(SShowInfo* pShowInfo, int64_t id, char* msgBuf, int32_t msgLen);
 SCreateDbMsg* buildCreateDbMsg(SCreateDbInfo* pCreateDbInfo, char* msgBuf, int32_t msgLen);
+SCreateStbMsg* buildCreateTableMsg(SCreateTableSql* pCreateTableSql, int32_t* len, SParseBasicCtx* pParseCtx, SMsgBuf* pMsgBuf);
 
 #endif  // TDENGINE_ASTTOMSG_H

source/libs/parser/inc/dataBlockMgt.h

@@ -78,7 +78,7 @@ typedef struct {
 typedef struct STableDataBlocks {
   int8_t      tsSource;     // where does the UNIX timestamp come from, server or client
   bool        ordered;      // if current rows are ordered or not
-  int64_t     vgId;         // virtual group id
+  int32_t     vgId;         // virtual group id
   int64_t     prevTS;       // previous timestamp, recorded to decide if the records array is ts ascending
   int32_t     numOfTables;  // number of tables in current submit block
   int32_t     rowSize;      // row size for current table

source/libs/parser/inc/parserInt.h

@@ -38,6 +38,14 @@ typedef struct SMsgBuf {
   char   *buf;
 } SMsgBuf;
 
+// 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,
+};
+
 void clearTableMetaInfo(STableMetaInfo* pTableMetaInfo);
 
 void clearAllTableMetaInfo(SQueryStmtInfo* pQueryInfo, bool removeMeta, uint64_t id);
@@ -60,7 +68,7 @@ int32_t qParserValidateSqlNode(struct SCatalog* pCatalog, SSqlInfo* pSqlInfo, SQ
  * @param type
  * @return
  */
-int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, int64_t id, void** output, int32_t* outputLen, int32_t* type, char* msgBuf, int32_t msgBufLen);
+int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, void** output, int32_t* outputLen, int32_t* type, char* msgBuf, int32_t msgBufLen);
 
 /**
  * Evaluate the numeric and timestamp arithmetic expression in the WHERE clause.

source/libs/parser/inc/parserUtil.h

@@ -67,6 +67,8 @@ int32_t    getExprFunctionId(SExprInfo *pExprInfo);
 STableMeta* tableMetaDup(const STableMeta* pTableMeta);
 
 bool isDclSqlStatement(SSqlInfo* pSqlInfo);
+bool isDdlSqlStatement(SSqlInfo* pSqlInfo);
+bool isDqlSqlStatement(SSqlInfo* pSqlInfo);
 
 #ifdef __cplusplus
 }

source/libs/parser/src/astGenerator.c

@@ -13,9 +13,10 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "taos.h"
-#include "os.h"
 #include "astGenerator.h"
+#include <parserInt.h>
+#include "os.h"
+#include "taos.h"
 #include "tmsgtype.h"
 
 SArray *tListItemAppend(SArray *pList, SVariant *pVar, uint8_t sortOrder) {

source/libs/parser/src/astToMsg.c

@@ -1,3 +1,4 @@
+#include <astGenerator.h>
 #include "parserInt.h"
 #include "parserUtil.h"
 
@@ -219,3 +220,126 @@ SCreateDbMsg* buildCreateDbMsg(SCreateDbInfo* pCreateDbInfo, char* msgBuf, int32
 
   return pCreateMsg;
 }
+
+int32_t createSName(SName* pName, SToken* pTableName, SParseBasicCtx* pParseCtx, SMsgBuf* pMsgBuf) {
+  const char* msg1 = "name too long";
+  const char* msg2 = "acctId too long";
+
+  int32_t  code = TSDB_CODE_SUCCESS;
+  char* p  = strnchr(pTableName->z, TS_PATH_DELIMITER[0], pTableName->n, false);
+
+  if (p != NULL) { // db has been specified in sql string so we ignore current db path
+    code = tNameSetAcctId(pName, pParseCtx->acctId);
+    if (code != 0) {
+      return buildInvalidOperationMsg(pMsgBuf, msg2);
+    }
+
+    char name[TSDB_TABLE_FNAME_LEN] = {0};
+    strncpy(name, pTableName->z, pTableName->n);
+
+    code = tNameFromString(pName, name, T_NAME_DB|T_NAME_TABLE);
+    if (code != 0) {
+      return buildInvalidOperationMsg(pMsgBuf, msg1);
+    }
+  } else {  // get current DB name first, and then set it into path
+    if (pTableName->n >= TSDB_TABLE_NAME_LEN) {
+      return buildInvalidOperationMsg(pMsgBuf, msg1);
+    }
+
+    tNameSetDbName(pName, pParseCtx->acctId, pParseCtx->db, strlen(pParseCtx->db));
+
+    char name[TSDB_TABLE_FNAME_LEN] = {0};
+    strncpy(name, pTableName->z, pTableName->n);
+
+    code = tNameFromString(pName, name, T_NAME_TABLE);
+    if (code != 0) {
+      code = buildInvalidOperationMsg(pMsgBuf, msg1);
+    }
+  }
+
+  return code;
+}
+
+SCreateStbMsg* buildCreateTableMsg(SCreateTableSql* pCreateTableSql, int32_t* len, SParseBasicCtx* pParseCtx, SMsgBuf* pMsgBuf) {
+  SSchema* pSchema;
+
+  int32_t numOfCols = (int32_t) taosArrayGetSize(pCreateTableSql->colInfo.pColumns);
+  int32_t numOfTags = (int32_t) taosArrayGetSize(pCreateTableSql->colInfo.pTagColumns);
+
+  SCreateStbMsg* pCreateTableMsg = (SCreateStbMsg*)calloc(1, sizeof(SCreateStbMsg) + (numOfCols + numOfTags) * sizeof(SSchema));
+
+  char* pMsg = NULL;
+  int8_t type = pCreateTableSql->type;
+  if (type == TSQL_CREATE_TABLE) {  // create by using super table, tags value
+#if 0
+    SArray* list = pInfo->pCreateTableInfo->childTableInfo;
+
+    int32_t numOfTables = (int32_t)taosArrayGetSize(list);
+    pCreateTableMsg->numOfTables = htonl(numOfTables);
+
+    pMsg = (char*)pCreateMsg;
+    for (int32_t i = 0; i < numOfTables; ++i) {
+      SCreateTableMsg* pCreate = (SCreateTableMsg*)pMsg;
+
+      pCreate->numOfColumns = htons(pCmd->numOfCols);
+      pCreate->numOfTags = htons(pCmd->count);
+      pMsg += sizeof(SCreateTableMsg);
+
+      SCreatedTableInfo* p = taosArrayGet(list, i);
+      strcpy(pCreate->tableName, p->fullname);
+      pCreate->igExists = (p->igExist) ? 1 : 0;
+
+      // use dbinfo from table id without modifying current db info
+      pMsg = serializeTagData(&p->tagdata, pMsg);
+
+      int32_t len = (int32_t)(pMsg - (char*)pCreate);
+      pCreate->len = htonl(len);
+    }
+#endif
+  } else { // create (super) table
+    SName n = {0};
+    int32_t code = createSName(&n, &pCreateTableSql->name, pParseCtx, pMsgBuf);
+    if (code != 0) {
+      return NULL;
+    }
+
+    code = tNameExtractFullName(&n, pCreateTableMsg->name);
+    if (code != 0) {
+      buildInvalidOperationMsg(pMsgBuf, "invalid table name or database not specified");
+      return NULL;
+    }
+
+    pCreateTableMsg->igExists = pCreateTableSql->existCheck ? 1 : 0;
+
+    pCreateTableMsg->numOfColumns = htonl(numOfCols);
+    pCreateTableMsg->numOfTags = htonl(numOfTags);
+
+    pSchema = (SSchema*) pCreateTableMsg->pSchema;
+    for (int i = 0; i < numOfCols; ++i) {
+      TAOS_FIELD* pField = taosArrayGet(pCreateTableSql->colInfo.pColumns, i);
+
+      pSchema->type  = pField->type;
+      pSchema->bytes = htonl(pField->bytes);
+      strcpy(pSchema->name, pField->name);
+
+      pSchema++;
+    }
+
+    for(int32_t i = 0; i < numOfTags; ++i) {
+      TAOS_FIELD* pField = taosArrayGet(pCreateTableSql->colInfo.pTagColumns, i);
+
+      pSchema->type  = pField->type;
+      pSchema->bytes = htonl(pField->bytes);
+      strcpy(pSchema->name, pField->name);
+
+      pSchema++;
+    }
+
+    pMsg = (char*)pSchema;
+  }
+
+  int32_t msgLen = (int32_t)(pMsg - (char*)pCreateTableMsg);
+  *len = msgLen;
+
+  return pCreateTableMsg;
+}

source/libs/parser/src/astValidate.c

@@ -4171,7 +4171,144 @@ static int32_t doCheckDbOptions(SCreateDbMsg* pCreate, SMsgBuf* pMsgBuf) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, int64_t id, void** output, int32_t* outputLen, int32_t* type, char* msgBuf, int32_t msgBufLen) {
+/* is contained in pFieldList or not */
+static bool has(SArray* pFieldList, int32_t startIndex, const char* name) {
+  size_t numOfCols = taosArrayGetSize(pFieldList);
+  for (int32_t j = startIndex; j < numOfCols; ++j) {
+    TAOS_FIELD* field = taosArrayGet(pFieldList, j);
+    if (strncasecmp(name, field->name, sizeof(field->name) - 1) == 0) return true;
+  }
+
+  return false;
+}
+
+static int32_t validateTableColumns(SArray* pFieldList, int32_t maxRowLength, int32_t maxColumns, SMsgBuf* pMsgBuf) {
+  const char* msg2 = "row length exceeds max length";
+  const char* msg3 = "duplicated column names";
+  const char* msg4 = "invalid data type";
+  const char* msg5 = "invalid binary/nchar column length";
+  const char* msg6 = "invalid column name";
+  const char* msg7 = "too many columns";
+  const char* msg8 = "illegal number of columns";
+
+  size_t numOfCols = taosArrayGetSize(pFieldList);
+  if (numOfCols > maxColumns) {
+    return buildInvalidOperationMsg(pMsgBuf, msg7);
+  }
+
+  int32_t rowLen = 0;
+  for (int32_t i = 0; i < numOfCols; ++i) {
+    TAOS_FIELD* pField = taosArrayGet(pFieldList, i);
+    if (!isValidDataType(pField->type)) {
+      return buildInvalidOperationMsg(pMsgBuf, msg4);
+    }
+
+    if (pField->bytes == 0) {
+      return buildInvalidOperationMsg(pMsgBuf, msg5);
+    }
+
+    if ((pField->type == TSDB_DATA_TYPE_BINARY && (pField->bytes <= 0 || pField->bytes > TSDB_MAX_BINARY_LEN)) ||
+        (pField->type == TSDB_DATA_TYPE_NCHAR && (pField->bytes <= 0 || pField->bytes > TSDB_MAX_NCHAR_LEN))) {
+      return buildInvalidOperationMsg(pMsgBuf, msg5);
+    }
+
+    SToken nameToken = {.z = pField->name, .n = strlen(pField->name), .type = TK_ID};
+    if (parserValidateNameToken(&nameToken) != TSDB_CODE_SUCCESS) {
+      return buildInvalidOperationMsg(pMsgBuf, msg6);
+    }
+
+    // field name must be unique
+    if (has(pFieldList, i + 1, pField->name) == true) {
+      return buildInvalidOperationMsg(pMsgBuf, msg3);
+    }
+
+    rowLen += pField->bytes;
+  }
+
+  // max row length must be less than TSDB_MAX_BYTES_PER_ROW
+  if (rowLen > maxRowLength) {
+    return buildInvalidOperationMsg(pMsgBuf, msg2);
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t validateTableColumnInfo(SArray* pFieldList, SMsgBuf* pMsgBuf) {
+  assert(pFieldList != NULL);
+
+  const char* msg1 = "first column must be timestamp";
+  const char* msg2 = "row length exceeds max length";
+  const char* msg3 = "duplicated column names";
+  const char* msg4 = "invalid data type";
+  const char* msg5 = "invalid binary/nchar column length";
+  const char* msg6 = "invalid column name";
+  const char* msg7 = "too many columns";
+  const char* msg8 = "illegal number of columns";
+
+  // first column must be timestamp
+  TAOS_FIELD* pField = taosArrayGet(pFieldList, 0);
+  if (pField->type != TSDB_DATA_TYPE_TIMESTAMP) {
+    return buildInvalidOperationMsg(pMsgBuf, msg1);
+  }
+
+  // number of fields no less than 2
+  size_t numOfCols = taosArrayGetSize(pFieldList);
+  if (numOfCols <= 1) {
+    return buildInvalidOperationMsg(pMsgBuf, msg8);
+  }
+
+  return validateTableColumns(pFieldList, TSDB_MAX_BYTES_PER_ROW, TSDB_MAX_COLUMNS, pMsgBuf);
+}
+
+static int32_t validateTagParams(SArray* pTagsList, SArray* pFieldList, SMsgBuf* pMsgBuf) {
+  assert(pTagsList != NULL);
+
+  const char* msg1 = "invalid number of tag columns";
+  const char* msg3 = "duplicated column names";
+
+  // number of fields at least 1
+  size_t numOfTags = taosArrayGetSize(pTagsList);
+  if (numOfTags < 1) {
+    return buildInvalidOperationMsg(pMsgBuf, msg1);
+  }
+
+  // field name must be unique
+  for (int32_t i = 0; i < numOfTags; ++i) {
+    TAOS_FIELD* p = taosArrayGet(pTagsList, i);
+    if (has(pFieldList, 0, p->name) == true) {
+      return buildInvalidOperationMsg(pMsgBuf, msg3);
+    }
+  }
+
+  return validateTableColumns(pFieldList, TSDB_MAX_TAGS_LEN, TSDB_MAX_TAGS, pMsgBuf);
+}
+
+int32_t doCheckForCreateTable(SSqlInfo* pInfo, SMsgBuf* pMsgBuf) {
+  const char* msg1 = "invalid table name";
+
+  SCreateTableSql* pCreateTable = pInfo->pCreateTableInfo;
+
+  SArray* pFieldList = pCreateTable->colInfo.pColumns;
+  SArray* pTagList = pCreateTable->colInfo.pTagColumns;
+  assert(pFieldList != NULL);
+
+  // if sql specifies db, use it, otherwise use default db
+  SToken* pzTableName = &(pCreateTable->name);
+
+  bool dbIncluded = false;
+  if (parserValidateNameToken(pzTableName) != TSDB_CODE_SUCCESS) {
+    return buildInvalidOperationMsg(pMsgBuf, msg1);
+  }
+
+  if (validateTableColumnInfo(pFieldList, pMsgBuf) != TSDB_CODE_SUCCESS ||
+      (pTagList != NULL && validateTagParams(pTagList, pFieldList, pMsgBuf) != TSDB_CODE_SUCCESS)) {
+    return TSDB_CODE_TSC_INVALID_OPERATION;
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, void** output, int32_t* outputLen, int32_t* type, char* msgBuf, int32_t msgBufLen) {
   int32_t code = 0;
 
   SMsgBuf m = {.buf = msgBuf, .len = msgBufLen};
@@ -4224,7 +4361,7 @@ int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, int64_t id, void** output, in
         }
       }
 
-      *output = buildUserManipulationMsg(pInfo, outputLen, id, msgBuf, msgBufLen);
+      *output = buildUserManipulationMsg(pInfo, outputLen, pCtx->requestId, msgBuf, msgBufLen);
       break;
     }
 
@@ -4260,13 +4397,13 @@ int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, int64_t id, void** output, in
         }
       }
 
-      *output = buildAcctManipulationMsg(pInfo, outputLen, id, msgBuf, msgBufLen);
+      *output = buildAcctManipulationMsg(pInfo, outputLen, pCtx->requestId, msgBuf, msgBufLen);
       break;
     }
 
     case TSDB_SQL_DROP_ACCT:
     case TSDB_SQL_DROP_USER: {
-      *output = buildDropUserMsg(pInfo, outputLen, id, msgBuf, msgBufLen);
+      *output = buildDropUserMsg(pInfo, outputLen, pCtx->requestId, msgBuf, msgBufLen);
       break;
     }
     
@@ -4275,6 +4412,28 @@ int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, int64_t id, void** output, in
       break;
     }
 
+    case TSDB_SQL_USE_DB: {
+      const char* msg = "invalid db name";
+
+      SToken* pToken = taosArrayGet(pInfo->pMiscInfo->a, 0);
+      if (parserValidateNameToken(pToken) != TSDB_CODE_SUCCESS) {
+        return buildInvalidOperationMsg(pMsgBuf, msg);
+      }
+
+      SName n = {0};
+      int32_t ret = tNameSetDbName(&n, pCtx->acctId, pToken->z, pToken->n);
+      if (ret != TSDB_CODE_SUCCESS) {
+        return buildInvalidOperationMsg(pMsgBuf, msg);
+      }
+
+      SUseDbMsg *pUseDbMsg = (SUseDbMsg *) calloc(1, sizeof(SUseDbMsg));
+      tNameExtractFullName(&n, pUseDbMsg->db);
+
+      *output = pUseDbMsg;
+      *outputLen = sizeof(SUseDbMsg);
+      break;
+    }
+
     case TSDB_SQL_ALTER_DB:
     case TSDB_SQL_CREATE_DB: {
       const char* msg1 = "invalid db name";
@@ -4304,6 +4463,26 @@ int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, int64_t id, void** output, in
       break;
     }
 
+    case TSDB_SQL_CREATE_TABLE: {
+      SCreateTableSql* pCreateTable = pInfo->pCreateTableInfo;
+
+      if (pCreateTable->type == TSQL_CREATE_TABLE || pCreateTable->type == TSQL_CREATE_STABLE) {
+        if ((code = doCheckForCreateTable(pInfo, pMsgBuf)) != TSDB_CODE_SUCCESS) {
+          return code;
+        }
+        *output = buildCreateTableMsg(pCreateTable, outputLen, pCtx, pMsgBuf);
+      } else if (pCreateTable->type == TSQL_CREATE_CTABLE) {
+        //        if ((code = doCheckForCreateFromStable(pSql, pInfo)) != TSDB_CODE_SUCCESS) {
+        //          return code;
+        //        }
+
+      } else if (pCreateTable->type == TSQL_CREATE_STREAM) {
+        //        if ((code = doCheckForStream(pSql, pInfo)) != TSDB_CODE_SUCCESS) {
+        //          return code;
+      }
+
+      break;
+    }
     default:
       break;
   }

source/libs/parser/src/dataBlockMgt.c

@@ -123,7 +123,6 @@ static int32_t createDataBlock(size_t defaultSize, int32_t rowSize, int32_t star
     dataBuf->nAllocSize = dataBuf->headerSize * 2;
   }
 
-  //dataBuf->pData = calloc(1, dataBuf->nAllocSize);
   dataBuf->pData = malloc(dataBuf->nAllocSize);
   if (dataBuf->pData == NULL) {
     tfree(dataBuf);
@@ -145,8 +144,6 @@ static int32_t createDataBlock(size_t defaultSize, int32_t rowSize, int32_t star
   dataBuf->tsSource = -1;
   dataBuf->vgId     = dataBuf->pTableMeta->vgId;
 
-  // tNameAssign(&dataBuf->tableName, name);
-
   assert(defaultSize > 0 && pTableMeta != NULL && dataBuf->pTableMeta != NULL);
 
   *dataBlocks = dataBuf;

source/libs/parser/src/insertParser.c

@@ -50,15 +50,16 @@ enum {
 };
 
 typedef struct SInsertParseContext {
-  SParseContext* pComCxt;
+  SParseContext* pComCxt;       // input
-  const char* pSql;
+  const char* pSql;             // input
-  SMsgBuf msg;
+  SMsgBuf msg;                  // input
-  STableMeta* pTableMeta;
+  STableMeta* pTableMeta;       // each table
-  SParsedDataColInfo tags;
+  SParsedDataColInfo tags;      // each table
-  SKVRowBuilder tagsBuilder;
+  SKVRowBuilder tagsBuilder;    // each table
-  SHashObj* pTableBlockHashObj;
+  SHashObj* pVgroupsHashObj;    // global
-  SArray* pTableDataBlocks;
+  SHashObj* pTableBlockHashObj; // global
-  SArray* pVgDataBlocks;
+  SArray* pTableDataBlocks;     // global
+  SArray* pVgDataBlocks;        // global
   int32_t totalNum;
   SInsertStmtInfo* pOutput;
 } SInsertParseContext;
@@ -157,14 +158,14 @@ static int32_t buildName(SInsertParseContext* pCxt, SToken* pStname, char* fullD
 
   char* p = strnchr(pStname->z, TS_PATH_DELIMITER[0], pStname->n, false);
   if (NULL != p) { // db.table
-    strcpy(fullDbName, pCxt->pComCxt->pAcctId);
+    int32_t n = sprintf(fullDbName, "%d.", pCxt->pComCxt->ctx.acctId);
-    fullDbName[strlen(pCxt->pComCxt->pAcctId)] = TS_PATH_DELIMITER[0];
+    strncpy(fullDbName + n, pStname->z, p - pStname->z);
-    strncpy(fullDbName, pStname->z, p - pStname->z);
     strncpy(tableName, p + 1, pStname->n - (p - pStname->z) - 1);
   } else {
-    snprintf(fullDbName, TSDB_FULL_DB_NAME_LEN, "%s.%s", pCxt->pComCxt->pAcctId, pCxt->pComCxt->pDbname);
+    snprintf(fullDbName, TSDB_FULL_DB_NAME_LEN, "%d.%s", pCxt->pComCxt->ctx.acctId, pCxt->pComCxt->ctx.db);
     strncpy(tableName, pStname->z, pStname->n);
   }
+
   return TSDB_CODE_SUCCESS;
 }
 
@@ -173,10 +174,12 @@ static int32_t getTableMeta(SInsertParseContext* pCxt, SToken* pTname) {
   char tableName[TSDB_TABLE_NAME_LEN] = {0};
   CHECK_CODE(buildName(pCxt, pTname, fullDbName, tableName));
   CHECK_CODE(catalogGetTableMeta(pCxt->pComCxt->pCatalog, pCxt->pComCxt->pRpc, pCxt->pComCxt->pEpSet, fullDbName, tableName, &pCxt->pTableMeta));
+  SVgroupInfo vg;
+  CHECK_CODE(catalogGetTableHashVgroup(pCxt->pComCxt->pCatalog, pCxt->pComCxt->pRpc, pCxt->pComCxt->pEpSet, fullDbName, tableName, &vg));
+  CHECK_CODE(taosHashPut(pCxt->pVgroupsHashObj, (const char*)&vg.vgId, sizeof(vg.vgId), (char*)&vg, sizeof(vg)));
   return TSDB_CODE_SUCCESS;
 }
 
-// todo speedup by using hash list
 static int32_t findCol(SToken* pColname, int32_t start, int32_t end, SSchema* pSchema) {
   while (start < end) {
     if (strlen(pSchema[start].name) == pColname->n && strncmp(pColname->z, pSchema[start].name, pColname->n) == 0) {
@@ -187,16 +190,16 @@ static int32_t findCol(SToken* pColname, int32_t start, int32_t end, SSchema* pS
   return -1;
 }
 
-static void fillMsgHeader(SVgDataBlocks* dst) {
+static void buildMsgHeader(SVgDataBlocks* blocks) {
-    SMsgDesc* desc = (SMsgDesc*)dst->pData;
+    SMsgDesc* desc = (SMsgDesc*)blocks->pData;
     desc->numOfVnodes = htonl(1);
     SSubmitMsg* submit = (SSubmitMsg*)(desc + 1);
-    submit->header.vgId    = htonl(dst->vgId);
+    submit->header.vgId    = htonl(blocks->vg.vgId);
-    submit->header.contLen = htonl(dst->size - sizeof(SMsgDesc));
+    submit->header.contLen = htonl(blocks->size - sizeof(SMsgDesc));
     submit->length         = submit->header.contLen;
-    submit->numOfBlocks    = htonl(dst->numOfTables);
+    submit->numOfBlocks    = htonl(blocks->numOfTables);
     SSubmitBlk* blk = (SSubmitBlk*)(submit + 1);
-    int32_t numOfBlocks = dst->numOfTables;
+    int32_t numOfBlocks = blocks->numOfTables;
     while (numOfBlocks--) {
       int32_t dataLen = blk->dataLen;
       blk->uid = htobe64(blk->uid);
@@ -222,11 +225,11 @@ static int32_t buildOutput(SInsertParseContext* pCxt) {
     if (NULL == dst) {
       return TSDB_CODE_TSC_OUT_OF_MEMORY;
     }
-    dst->vgId = src->vgId;
+    taosHashGetClone(pCxt->pVgroupsHashObj, (const char*)&src->vgId, sizeof(src->vgId), &dst->vg);
     dst->numOfTables = src->numOfTables;
     dst->size = src->size;
     SWAP(dst->pData, src->pData, char*);
-    fillMsgHeader(dst);
+    buildMsgHeader(dst);
     taosArrayPush(pCxt->pOutput->pDataBlocks, &dst);
   }
   return TSDB_CODE_SUCCESS;
@@ -546,7 +549,7 @@ static FORCE_INLINE int32_t parseOneValue(SInsertParseContext* pCxt, SToken* pTo
     }
     case TSDB_DATA_TYPE_BINARY: {
       // too long values will return invalid sql, not be truncated automatically
-      if (pToken->n + VARSTR_HEADER_SIZE > pSchema->bytes) {  // todo refactor
+      if (pToken->n + VARSTR_HEADER_SIZE > pSchema->bytes) {
         return buildSyntaxErrMsg(&pCxt->msg, "string data overflow", pToken->z);
       }
       return func(pToken->z, pToken->n, param);
@@ -644,9 +647,7 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pTagsSchema,
     CHECK_CODE(parseOneValue(pCxt, &sToken, pSchema, precision, tmpTokenBuf, KvRowAppend, &param));
   }
 
-  destroyBoundColumnInfo(&pCxt->tags);
   SKVRow row = tdGetKVRowFromBuilder(&pCxt->tagsBuilder);
-  tdDestroyKVRowBuilder(&pCxt->tagsBuilder);
   if (NULL == row) {
     return buildInvalidOperationMsg(&pCxt->msg, "tag value expected");
   }
@@ -799,13 +800,30 @@ static int32_t parseValuesClause(SInsertParseContext* pCxt, STableDataBlocks* da
   return TSDB_CODE_SUCCESS;
 }
 
+static void destroyInsertParseContextForTable(SInsertParseContext* pCxt) {
+  tfree(pCxt->pTableMeta);
+  destroyBoundColumnInfo(&pCxt->tags);
+  tdDestroyKVRowBuilder(&pCxt->tagsBuilder);
+}
+
+static void destroyInsertParseContext(SInsertParseContext* pCxt) {
+  destroyInsertParseContextForTable(pCxt);
+  taosHashCleanup(pCxt->pVgroupsHashObj);
+  taosHashCleanup(pCxt->pTableBlockHashObj);
+  destroyBlockArrayList(pCxt->pTableDataBlocks);
+  destroyBlockArrayList(pCxt->pVgDataBlocks);
+}
+
 //   tb_name
 //       [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)]
 //       [(field1_name, ...)]
 //       VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path
 //   [...];
 static int32_t parseInsertBody(SInsertParseContext* pCxt) {
+  // for each table
   while (1) {
+    destroyInsertParseContextForTable(pCxt);
+
     SToken sToken;
     // pSql -> tb_name ...
     NEXT_TOKEN(pCxt->pSql, sToken);
@@ -867,15 +885,6 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
   return buildOutput(pCxt);
 }
 
-static void destroyInsertParseContext(SInsertParseContext* pCxt) {
-  tfree(pCxt->pTableMeta);
-  destroyBoundColumnInfo(&pCxt->tags);
-  tdDestroyKVRowBuilder(&pCxt->tagsBuilder);
-  taosHashCleanup(pCxt->pTableBlockHashObj);
-  destroyBlockArrayList(pCxt->pTableDataBlocks);
-  destroyBlockArrayList(pCxt->pVgDataBlocks);
-}
-
 // INSERT INTO
 //   tb_name
 //       [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)]
@@ -888,12 +897,13 @@ int32_t parseInsertSql(SParseContext* pContext, SInsertStmtInfo** pInfo) {
     .pSql = pContext->pSql,
     .msg = {.buf = pContext->pMsg, .len = pContext->msgLen},
     .pTableMeta = NULL,
+    .pVgroupsHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, false),
     .pTableBlockHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false),
     .totalNum = 0,
     .pOutput = calloc(1, sizeof(SInsertStmtInfo))
   };
 
-  if (NULL == context.pTableBlockHashObj || NULL == context.pOutput) {
+  if (NULL == context.pVgroupsHashObj || NULL == context.pTableBlockHashObj || NULL == context.pOutput) {
     terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
     return TSDB_CODE_FAILED;
   }

source/libs/parser/src/parser.c

@@ -31,7 +31,7 @@ bool qIsInsertSql(const char* pStr, size_t length) {
   } while (1);
 }
 
-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(const char* pStr, size_t length, SParseBasicCtx* pParseCtx, int32_t *type, void** pOutput, int32_t* outputLen, char* msg, int32_t msgLen) {
   SSqlInfo info = doGenerateAST(pStr);
   if (!info.valid) {
     strncpy(msg, info.msg, msgLen);
@@ -39,8 +39,8 @@ int32_t qParseQuerySql(const char* pStr, size_t length, int64_t id, int32_t *typ
     return terrno;
   }
 
-  if (isDclSqlStatement(&info)) {
+  if (!isDqlSqlStatement(&info)) {
-    int32_t code = qParserValidateDclSqlNode(&info, id, pOutput, outputLen, type, msg, msgLen);
+    int32_t code = qParserValidateDclSqlNode(&info, pParseCtx, pOutput, outputLen, type, msg, msgLen);
     if (code == TSDB_CODE_SUCCESS) {
       // do nothing
     }
@@ -53,7 +53,7 @@ int32_t qParseQuerySql(const char* pStr, size_t length, int64_t id, int32_t *typ
 
     struct SCatalog* pCatalog = NULL;
     int32_t code = catalogGetHandle(NULL, &pCatalog);
-    code = qParserValidateSqlNode(pCatalog, &info, pQueryInfo, id, msg, msgLen);
+    code = qParserValidateSqlNode(pCatalog, &info, pQueryInfo, pParseCtx->requestId, msg, msgLen);
     if (code == TSDB_CODE_SUCCESS) {
       *pOutput = pQueryInfo;
     }

source/libs/parser/src/parserUtil.c

@@ -1167,7 +1167,7 @@ void cleanupTagCond(STagCond* pTagCond) {
  * @param tableIndex  denote the table index for join query, where more than one table exists
  * @return
  */
-STableMetaInfo* getMetaInfo(SQueryStmtInfo* pQueryInfo, int32_t tableIndex) {
+STableMetaInfo* getMetaInfo(const SQueryStmtInfo* pQueryInfo, int32_t tableIndex) {
   assert(pQueryInfo != NULL);
   if (pQueryInfo->pTableMetaInfo == NULL) {
     assert(pQueryInfo->numOfTables == 0);
@@ -1613,7 +1613,18 @@ uint32_t convertRelationalOperator(SToken *pToken) {
 }
 
 bool isDclSqlStatement(SSqlInfo* pSqlInfo) {
-  return (pSqlInfo->type != TSDB_SQL_SELECT);
+  int32_t type = pSqlInfo->type;
+  return (type == TSDB_SQL_CREATE_USER || type == TSDB_SQL_CREATE_ACCT || type == TSDB_SQL_DROP_USER ||
+          type == TSDB_SQL_DROP_ACCT || type == TSDB_SQL_SHOW);
+}
+
+bool isDdlSqlStatement(SSqlInfo* pSqlInfo) {
+  int32_t type = pSqlInfo->type;
+  return (type == TSDB_SQL_CREATE_TABLE || type == TSDB_SQL_CREATE_DB || type == TSDB_SQL_DROP_DB);
+}
+
+bool isDqlSqlStatement(SSqlInfo* pSqlInfo) {
+  return pSqlInfo->type == TSDB_SQL_SELECT;
 }
 
 #if 0

source/libs/parser/src/queryInfoUtil.c

@@ -354,7 +354,7 @@ bool tscHasColumnFilter(SQueryStmtInfo* pQueryInfo) {
   return false;
 }
 
-int32_t getExprFunctionLevel(SQueryStmtInfo* pQueryInfo) {
+int32_t getExprFunctionLevel(const SQueryStmtInfo* pQueryInfo) {
   int32_t n = 10;
 
   int32_t level = 0;

source/libs/parser/src/sql.c

@@ -31,6 +31,7 @@
 #include <assert.h>
 #include <stdbool.h>
 #include "astGenerator.h"
+#include "parserInt.h"
 #include "tmsgtype.h"
 #include "ttoken.h"
 #include "ttokendef.h"

source/libs/parser/test/insertParserTest.cpp

@@ -43,8 +43,8 @@ protected:
 
   void bind(const char* sql) {
     reset();
-    cxt_.pAcctId = acctId_.c_str();
+    cxt_.ctx.acctId = atoi(acctId_.c_str());
-    cxt_.pDbname = db_.c_str();
+    cxt_.ctx.db = (char*) db_.c_str();
     strcpy(sqlBuf_, sql);
     cxt_.sqlLen = strlen(sql);
     sqlBuf_[cxt_.sqlLen] = '\0';
@@ -69,7 +69,7 @@ protected:
     cout << "schemaAttache:" << (int32_t)res_->schemaAttache << ", payloadType:" << (int32_t)res_->payloadType << ", insertType:" << res_->insertType << ", numOfVgs:" << num << endl;    
     for (size_t i = 0; i < num; ++i) {
       SVgDataBlocks* vg = (SVgDataBlocks*)taosArrayGetP(res_->pDataBlocks, i);
-      cout << "vgId:" << vg->vgId << ", numOfTables:" << vg->numOfTables << ", dataSize:" << vg->size << endl;
+      cout << "vgId:" << vg->vg.vgId << ", numOfTables:" << vg->numOfTables << ", dataSize:" << vg->size << endl;
       SMsgDesc* desc = (SMsgDesc*)(vg->pData);
       cout << "numOfVnodes:" << ntohl(desc->numOfVnodes) << endl;
       SSubmitMsg* submit = (SSubmitMsg*)(desc + 1);

source/libs/parser/test/mockCatalog.cpp

@@ -42,11 +42,15 @@ void generateTestST1(MockCatalogService* mcs) {
 }
 
 int32_t __catalogGetHandle(const char *clusterId, struct SCatalog** catalogHandle) {
-  return mockCatalogService->catalogGetHandle(clusterId, catalogHandle);
+  return 0;
 }
 
 int32_t __catalogGetTableMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pDBName, const char* pTableName, STableMeta** pTableMeta) {
-  return mockCatalogService->catalogGetTableMeta(pCatalog, pRpc, pMgmtEps, pDBName, pTableName, pTableMeta);
+  return mockCatalogService->catalogGetTableMeta(pDBName, pTableName, pTableMeta);
+}
+
+int32_t __catalogGetTableHashVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pDBName, const char* pTableName, SVgroupInfo* vgInfo) {
+  return mockCatalogService->catalogGetTableHashVgroup(pDBName, pTableName, vgInfo);
 }
 
 void initMetaDataEnv() {
@@ -55,6 +59,7 @@ void initMetaDataEnv() {
   static Stub stub;
   stub.set(catalogGetHandle, __catalogGetHandle);
   stub.set(catalogGetTableMeta, __catalogGetTableMeta);
+  stub.set(catalogGetTableHashVgroup, __catalogGetTableHashVgroup);
   {
     AddrAny any("libcatalog.so");
     std::map<std::string,void*> result;
@@ -71,6 +76,14 @@ void initMetaDataEnv() {
       stub.set(f.second, __catalogGetTableMeta);
     }
   }
+  {
+    AddrAny any("libcatalog.so");
+    std::map<std::string,void*> result;
+    any.get_global_func_addr_dynsym("^catalogGetTableHashVgroup$", result);
+    for (const auto& f : result) {
+      stub.set(f.second, __catalogGetTableHashVgroup);
+    }
+  }
 }
 
 void generateMetaData() {

source/libs/parser/test/mockCatalogService.cpp

@@ -90,11 +90,11 @@ public:
   MockCatalogServiceImpl() : id_(1) {
   }
 
-  int32_t catalogGetHandle(const char *clusterId, struct SCatalog** catalogHandle) const {
+  int32_t catalogGetHandle() const {
     return 0;
   }
 
-  int32_t catalogGetTableMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pDBName, const char* pTableName, STableMeta** pTableMeta) const {
+  int32_t catalogGetTableMeta(const char* pDBName, const char* pTableName, STableMeta** pTableMeta) const {
     std::unique_ptr<STableMeta> table;
     int32_t code = copyTableSchemaMeta(pDBName, pTableName, &table);
     if (TSDB_CODE_SUCCESS != code) {
@@ -104,6 +104,11 @@ public:
     return TSDB_CODE_SUCCESS;
   }
 
+  int32_t catalogGetTableHashVgroup(const char* pDBName, const char* pTableName, SVgroupInfo* vgInfo) const {
+    // todo
+    return 0;
+  }
+
   TableBuilder& createTableBuilder(const std::string& db, const std::string& tbname, int8_t tableType, int32_t numOfColumns, int32_t numOfTags) {
     builder_ = TableBuilder::createTableBuilder(tableType, numOfColumns, numOfTags);
     meta_[db][tbname] = builder_->table();
@@ -270,10 +275,10 @@ std::shared_ptr<MockTableMeta> MockCatalogService::getTableMeta(const std::strin
   return impl_->getTableMeta(db, tbname);
 }
 
-int32_t MockCatalogService::catalogGetHandle(const char *clusterId, struct SCatalog** catalogHandle) const {
+int32_t MockCatalogService::catalogGetTableMeta(const char* pDBName, const char* pTableName, STableMeta** pTableMeta) const {
-  return impl_->catalogGetHandle(clusterId, catalogHandle);
+  return impl_->catalogGetTableMeta(pDBName, pTableName, pTableMeta);
 }
 
-int32_t MockCatalogService::catalogGetTableMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pDBName, const char* pTableName, STableMeta** pTableMeta) const {
+int32_t MockCatalogService::catalogGetTableHashVgroup(const char* pDBName, const char* pTableName, SVgroupInfo* vgInfo) const {
-  return impl_->catalogGetTableMeta(pCatalog, pRpc, pMgmtEps, pDBName, pTableName, pTableMeta);
+  return impl_->catalogGetTableHashVgroup(pDBName, pTableName, vgInfo);
-}
+}

source/libs/parser/test/mockCatalogService.h

@@ -57,9 +57,8 @@ public:
   void showTables() const;
   std::shared_ptr<MockTableMeta> getTableMeta(const std::string& db, const std::string& tbname) const;
 
-  // mock interface
+  int32_t catalogGetTableMeta(const char* pDBName, const char* pTableName, STableMeta** pTableMeta) const;
-  int32_t catalogGetHandle(const char *clusterId, struct SCatalog** catalogHandle) const;
+  int32_t catalogGetTableHashVgroup(const char* pDBName, const char* pTableName, SVgroupInfo* vgInfo) const;
-  int32_t catalogGetTableMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pDBName, const char* pTableName, STableMeta** pTableMeta) const;
 
 private:
   std::unique_ptr<MockCatalogServiceImpl> impl_;

source/libs/parser/test/parserTests.cpp

@@ -717,7 +717,9 @@ TEST(testCase, show_user_Test) {
   void* output = NULL;
   int32_t type = 0;
   int32_t len = 0;
-  int32_t code = qParserValidateDclSqlNode(&info1, 1, &output, &len, &type, msg, buf.len);
+
+  SParseBasicCtx ct= {.db= "abc", .acctId = 1, .requestId = 1};
+  int32_t code = qParserValidateDclSqlNode(&info1, &ct, &output, &len, &type, msg, buf.len);
   ASSERT_EQ(code, 0);
 
   // convert the show command to be the select query
@@ -734,13 +736,14 @@ TEST(testCase, create_user_Test) {
 
   SSqlInfo info1 = doGenerateAST(sql);
   ASSERT_EQ(info1.valid, true);
-
   ASSERT_EQ(isDclSqlStatement(&info1), true);
 
   void* output = NULL;
   int32_t type = 0;
   int32_t len = 0;
-  int32_t code = qParserValidateDclSqlNode(&info1, 1, &output, &len, &type, msg, buf.len);
+
+  SParseBasicCtx ct= {.db= "abc", .acctId = 1, .requestId = 1};
+  int32_t code = qParserValidateDclSqlNode(&info1, &ct, &output, &len, &type, msg, buf.len);
   ASSERT_EQ(code, 0);
 
   destroySqlInfo(&info1);

source/libs/parser/test/plannerTest.cpp

@@ -93,7 +93,7 @@ void generateLogicplan(const char* sql) {
   ASSERT_EQ(ret, 0);
 
   struct SQueryPlanNode* n = nullptr;
-  code = createQueryPlan(pQueryInfo, &n);
+  code = createQueryPlan((const SQueryNode*)pQueryInfo, &n);
 
   char* str = NULL;
   queryPlanToString(n, &str);
@@ -156,7 +156,7 @@ TEST(testCase, planner_test) {
   ASSERT_EQ(pQueryInfo->fieldsInfo.numOfOutput, 2);
 
   struct SQueryPlanNode* n = nullptr;
-  code = createQueryPlan(pQueryInfo, &n);
+  code = createQueryPlan((const SQueryNode*)pQueryInfo, &n);
 
   char* str = NULL;
   queryPlanToString(n, &str);

source/libs/planner/inc/plannerInt.h

@@ -40,6 +40,7 @@ extern "C" {
 #define QNODE_SESSIONWINDOW 12
 #define QNODE_STATEWINDOW   13
 #define QNODE_FILL          14
+#define QNODE_INSERT        15
 
 typedef struct SQueryDistPlanNodeInfo {
   bool      stableQuery;   // super table query or not
@@ -82,7 +83,7 @@ int32_t optimizeQueryPlan(struct SQueryPlanNode* pQueryNode);
  * @param pQueryNode
  * @return
  */
-int32_t createQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryPlanNode** pQueryNode);
+int32_t createQueryPlan(const SQueryNode* pNode, struct SQueryPlanNode** pQueryPlan);
 
 /**
  * Convert the query plan to string, in order to display it in the shell.
@@ -101,7 +102,7 @@ int32_t queryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql);
 
 int32_t createDag(SQueryPlanNode* pQueryNode, struct SCatalog* pCatalog, SQueryDag** pDag);
 int32_t setSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SEpAddr* ep);
-int32_t subPlanToString(const SSubplan *pPhyNode, char** str);
+int32_t subPlanToString(const SSubplan *pPhyNode, char** str, int32_t* len);
 int32_t stringToSubplan(const char* str, SSubplan** subplan);
 
 /**
@@ -120,6 +121,9 @@ void* destroyQueryPhyPlan(struct SPhyNode* pQueryPhyNode);
 const char* opTypeToOpName(int32_t type);
 int32_t opNameToOpType(const char* name);
 
+const char* dsinkTypeToDsinkName(int32_t type);
+int32_t dsinkNameToDsinkType(const char* name);
+
 #ifdef __cplusplus
 }
 #endif

source/libs/planner/src/logicPlan.c

@@ -29,7 +29,7 @@ typedef struct SJoinCond {
   SColumn *colCond[2];
 } SJoinCond;
 
-static SArray* createQueryPlanImpl(SQueryStmtInfo* pQueryInfo);
+static SArray* createQueryPlanImpl(const SQueryStmtInfo* pQueryInfo);
 static void doDestroyQueryNode(SQueryPlanNode* pQueryNode);
 
 int32_t printExprInfo(char* buf, const SQueryPlanNode* pQueryNode, int32_t len);
@@ -37,16 +37,40 @@ int32_t optimizeQueryPlan(struct SQueryPlanNode* pQueryNode) {
   return 0;
 }
 
-int32_t createQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryPlanNode** pQueryNode) {
+int32_t createInsertPlan(const SInsertStmtInfo* pInsert, SQueryPlanNode** pQueryPlan) {
-  SArray* upstream = createQueryPlanImpl((struct SQueryStmtInfo*) pQueryInfo);
+  *pQueryPlan = calloc(1, sizeof(SQueryPlanNode));
+  SArray* blocks = taosArrayInit(taosArrayGetSize(pInsert->pDataBlocks), POINTER_BYTES);
+  if (NULL == *pQueryPlan || NULL == blocks) {
+    return TSDB_CODE_TSC_OUT_OF_MEMORY;
+  }
+  (*pQueryPlan)->info.type = QNODE_INSERT;
+  taosArrayAddAll(blocks, pInsert->pDataBlocks);
+  (*pQueryPlan)->pExtInfo = blocks;
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t createSelectPlan(const SQueryStmtInfo* pSelect, SQueryPlanNode** pQueryPlan) {
+  SArray* upstream = createQueryPlanImpl(pSelect);
   assert(taosArrayGetSize(upstream) == 1);
-
+  *pQueryPlan = taosArrayGetP(upstream, 0);
-  *pQueryNode = taosArrayGetP(upstream, 0);
-
   taosArrayDestroy(upstream);
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t createQueryPlan(const SQueryNode* pNode, SQueryPlanNode** pQueryPlan) {
+  switch (nodeType(pNode)) {
+    case TSDB_SQL_SELECT: {
+      return createSelectPlan((const SQueryStmtInfo*)pNode, pQueryPlan);
+    }
+    case TSDB_SQL_INSERT:
+      return createInsertPlan((const SInsertStmtInfo*)pNode, pQueryPlan);
+    default:
+      return TSDB_CODE_FAILED;
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
 int32_t queryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql) {
   return 0;
 }
@@ -62,7 +86,7 @@ void destroyQueryPlan(SQueryPlanNode* pQueryNode) {
 //======================================================================================================================
 
 static SQueryPlanNode* createQueryNode(int32_t type, const char* name, SQueryPlanNode** prev, int32_t numOfPrev,
-                                   SExprInfo** pExpr, int32_t numOfOutput, void* pExtInfo) {
+                                   SExprInfo** pExpr, int32_t numOfOutput, const void* pExtInfo) {
   SQueryPlanNode* pNode = calloc(1, sizeof(SQueryPlanNode));
 
   pNode->info.type = type;
@@ -123,7 +147,7 @@ static SQueryPlanNode* createQueryNode(int32_t type, const char* name, SQueryPla
     }
 
     case QNODE_FILL: { // todo !!
-      pNode->pExtInfo = pExtInfo;
+      pNode->pExtInfo = (void*)pExtInfo;
       break;
     }
 
@@ -145,7 +169,7 @@ static SQueryPlanNode* createQueryNode(int32_t type, const char* name, SQueryPla
   return pNode;
 }
 
-static SQueryPlanNode* doAddTableColumnNode(SQueryStmtInfo* pQueryInfo, STableMetaInfo* pTableMetaInfo, SQueryTableInfo* info,
+static SQueryPlanNode* doAddTableColumnNode(const SQueryStmtInfo* pQueryInfo, STableMetaInfo* pTableMetaInfo, SQueryTableInfo* info,
                                         SArray* pExprs, SArray* tableCols) {
   if (pQueryInfo->info.onlyTagQuery) {
     int32_t num = (int32_t) taosArrayGetSize(pExprs);
@@ -186,7 +210,7 @@ static SQueryPlanNode* doAddTableColumnNode(SQueryStmtInfo* pQueryInfo, STableMe
   return pNode;
 }
 
-static SQueryPlanNode* doCreateQueryPlanForSingleTableImpl(SQueryStmtInfo* pQueryInfo, SQueryPlanNode* pNode, SQueryTableInfo* info) {
+static SQueryPlanNode* doCreateQueryPlanForSingleTableImpl(const SQueryStmtInfo* pQueryInfo, SQueryPlanNode* pNode, SQueryTableInfo* info) {
   // group by column not by tag
   size_t numOfGroupCols = taosArrayGetSize(pQueryInfo->groupbyExpr.columnInfo);
 
@@ -239,7 +263,7 @@ static SQueryPlanNode* doCreateQueryPlanForSingleTableImpl(SQueryStmtInfo* pQuer
     memcpy(pInfo->val, pQueryInfo->fillVal, pNode->numOfExpr);
 
     SArray* p = pQueryInfo->exprList[0];  // top expression in select clause
-    pNode = createQueryNode(QNODE_FILL, "Fill", &pNode, 1, p, taosArrayGetSize(p), pInfo);
+    pNode = createQueryNode(QNODE_FILL, "Fill", &pNode, 1, p->pData, taosArrayGetSize(p), pInfo);
   }
 
   if (pQueryInfo->order != NULL) {
@@ -254,7 +278,7 @@ static SQueryPlanNode* doCreateQueryPlanForSingleTableImpl(SQueryStmtInfo* pQuer
   return pNode;
 }
 
-static SQueryPlanNode* doCreateQueryPlanForSingleTable(SQueryStmtInfo* pQueryInfo, STableMetaInfo* pTableMetaInfo, SArray* pExprs,
+static SQueryPlanNode* doCreateQueryPlanForSingleTable(const SQueryStmtInfo* pQueryInfo, STableMetaInfo* pTableMetaInfo, SArray* pExprs,
                                                 SArray* tableCols) {
   char name[TSDB_TABLE_FNAME_LEN] = {0};
   tstrncpy(name, pTableMetaInfo->name.tname, TSDB_TABLE_FNAME_LEN);
@@ -286,7 +310,7 @@ static bool isAllAggExpr(SArray* pList) {
   return true;
 }
 
-SArray* createQueryPlanImpl(SQueryStmtInfo* pQueryInfo) {
+SArray* createQueryPlanImpl(const SQueryStmtInfo* pQueryInfo) {
   SArray* upstream = NULL;
 
   if (pQueryInfo->pUpstream != NULL && taosArrayGetSize(pQueryInfo->pUpstream) > 0) {  // subquery in the from clause

source/libs/planner/src/physicalPlan.c

@@ -14,6 +14,7 @@
  */
 
 #include "plannerInt.h"
+#include "exception.h"
 #include "parser.h"
 
 #define STORE_CURRENT_SUBPLAN(cxt) SSubplan* _ = cxt->pCurrentSubplan
@@ -33,19 +34,67 @@ static const char* gOpName[] = {
 #undef INCLUDE_AS_NAME
 };
 
+static void* vailidPointer(void* p) {
+  if (NULL == p) {
+    THROW(TSDB_CODE_TSC_OUT_OF_MEMORY);
+  }
+  return p;
+}
+
 const char* opTypeToOpName(int32_t type) {
   return gOpName[type];
 }
 
 int32_t opNameToOpType(const char* name) {
   for (int32_t i = 1; i < sizeof(gOpName) / sizeof(gOpName[0]); ++i) {
-    if (strcmp(name, gOpName[i])) {
+    if (0 == strcmp(name, gOpName[i])) {
       return i;
     }
   }
   return OP_Unknown;
 }
 
+const char* dsinkTypeToDsinkName(int32_t type) {
+  switch (type) {
+    case DSINK_Dispatch:
+      return "Dispatch";
+    case DSINK_Insert:
+      return "Insert";
+    default:
+      break;
+  }
+  return "Unknown";
+}
+
+int32_t dsinkNameToDsinkType(const char* name) {
+  if (0 == strcmp(name, "Dispatch")) {
+    return DSINK_Dispatch;
+  } else if (0 == strcmp(name, "Insert")) {
+    return DSINK_Insert;
+  }
+  return DSINK_Unknown;
+}
+
+static SDataSink* initDataSink(int32_t type, int32_t size) {
+  SDataSink* sink = (SDataSink*)vailidPointer(calloc(1, size));
+  sink->info.type = type;
+  sink->info.name = dsinkTypeToDsinkName(type);
+  return sink;
+}
+
+static SDataSink* createDataDispatcher(SPlanContext* pCxt, SQueryPlanNode* pPlanNode) {
+  SDataDispatcher* dispatcher = (SDataDispatcher*)initDataSink(DSINK_Dispatch, sizeof(SDataDispatcher));
+  return (SDataSink*)dispatcher;
+}
+
+static SDataSink* createDataInserter(SPlanContext* pCxt, SVgDataBlocks* pBlocks) {
+  SDataInserter* inserter = (SDataInserter*)initDataSink(DSINK_Insert, sizeof(SDataInserter));
+  inserter->numOfTables = pBlocks->numOfTables;
+  inserter->size = pBlocks->size;
+  SWAP(inserter->pData, pBlocks->pData, char*);
+  return (SDataSink*)inserter;
+}
+
 static bool toDataBlockSchema(SQueryPlanNode* pPlanNode, SDataBlockSchema* dataBlockSchema) {
   dataBlockSchema->numOfCols = pPlanNode->numOfCols;
   dataBlockSchema->pSchema = malloc(sizeof(SSlotSchema) * pPlanNode->numOfCols);
@@ -72,10 +121,7 @@ static bool cloneExprArray(SArray** dst, SArray* src) {
 }
 
 static SPhyNode* initPhyNode(SQueryPlanNode* pPlanNode, int32_t type, int32_t size) {
-  SPhyNode* node = (SPhyNode*)calloc(1, size);
+  SPhyNode* node = (SPhyNode*)vailidPointer(calloc(1, size));
-  if (NULL == node) {
-    return NULL;
-  }
   node->info.type = type;
   node->info.name = opTypeToOpName(type);
   if (!cloneExprArray(&node->pTargets, pPlanNode->pExpr) || !toDataBlockSchema(pPlanNode, &(node->targetSchema))) {
@@ -138,7 +184,7 @@ static SPhyNode* createMultiTableScanNode(SQueryPlanNode* pPlanNode, SQueryTable
 }
 
 static SSubplan* initSubplan(SPlanContext* pCxt, int32_t type) {
-  SSubplan* subplan = calloc(1, sizeof(SSubplan));
+  SSubplan* subplan = vailidPointer(calloc(1, sizeof(SSubplan)));
   subplan->id = pCxt->nextId;
   ++(pCxt->nextId.subplanId);
   subplan->type = type;
@@ -146,15 +192,15 @@ static SSubplan* initSubplan(SPlanContext* pCxt, int32_t type) {
   if (NULL != pCxt->pCurrentSubplan) {
     subplan->level = pCxt->pCurrentSubplan->level + 1;
     if (NULL == pCxt->pCurrentSubplan->pChildern) {
-      pCxt->pCurrentSubplan->pChildern = taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES);
+      pCxt->pCurrentSubplan->pChildern = vailidPointer(taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES));
     }
     taosArrayPush(pCxt->pCurrentSubplan->pChildern, &subplan);
-    subplan->pParents = taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES);
+    subplan->pParents = vailidPointer(taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES));
     taosArrayPush(subplan->pParents, &pCxt->pCurrentSubplan);
   }
   SArray* currentLevel;
   if (subplan->level >= taosArrayGetSize(pCxt->pDag->pSubplans)) {
-    currentLevel = taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES);
+    currentLevel = vailidPointer(taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES));
     taosArrayPush(pCxt->pDag->pSubplans, &currentLevel);
   } else {
     currentLevel = taosArrayGetP(pCxt->pDag->pSubplans, subplan->level);
@@ -164,7 +210,17 @@ static SSubplan* initSubplan(SPlanContext* pCxt, int32_t type) {
   return subplan;
 }
 
-static void vgroupToEpSet(const SVgroupMsg* vg, SEpSet* epSet) {
+static void vgroupInfoToEpSet(const SVgroupInfo* vg, SEpSet* epSet) {
+  epSet->inUse = 0; // todo
+  epSet->numOfEps = vg->numOfEps;
+  for (int8_t i = 0; i < vg->numOfEps; ++i) {
+    epSet->port[i] = vg->epAddr[i].port;
+    strcpy(epSet->fqdn[i], vg->epAddr[i].fqdn);
+  }
+  return;
+}
+
+static void vgroupMsgToEpSet(const SVgroupMsg* vg, SEpSet* epSet) {
   epSet->inUse = 0; // todo
   epSet->numOfEps = vg->numOfEps;
   for (int8_t i = 0; i < vg->numOfEps; ++i) {
@@ -179,8 +235,9 @@ static uint64_t splitSubplanByTable(SPlanContext* pCxt, SQueryPlanNode* pPlanNod
   for (int32_t i = 0; i < pTable->pMeta->vgroupList->numOfVgroups; ++i) {
     STORE_CURRENT_SUBPLAN(pCxt);
     SSubplan* subplan = initSubplan(pCxt, QUERY_TYPE_SCAN);
-    vgroupToEpSet(&(pTable->pMeta->vgroupList->vgroups[i]), &subplan->execEpSet);
+    vgroupMsgToEpSet(&(pTable->pMeta->vgroupList->vgroups[i]), &subplan->execEpSet);
     subplan->pNode = createMultiTableScanNode(pPlanNode, pTable);
+    subplan->pDataSink = createDataDispatcher(pCxt, pPlanNode);
     RECOVERY_CURRENT_SUBPLAN(pCxt);
   }
   return pCxt->nextId.templateId++;
@@ -214,6 +271,9 @@ static SPhyNode* createPhyNode(SPlanContext* pCxt, SQueryPlanNode* pPlanNode) {
     case QNODE_TABLESCAN:
       node = createTableScanNode(pCxt, pPlanNode);
       break;
+    case QNODE_INSERT:
+      // Insert is not an operator in a physical plan.
+      break;
     default:
       assert(false);
   }
@@ -229,26 +289,46 @@ static SPhyNode* createPhyNode(SPlanContext* pCxt, SQueryPlanNode* pPlanNode) {
   return node;
 }
 
+static void splitInsertSubplan(SPlanContext* pCxt, SQueryPlanNode* pPlanNode) {
+  SArray* vgs = (SArray*)pPlanNode->pExtInfo;
+  size_t numOfVg = taosArrayGetSize(vgs);
+  for (int32_t i = 0; i < numOfVg; ++i) {
+    SSubplan* subplan = initSubplan(pCxt, QUERY_TYPE_MODIFY);
+    SVgDataBlocks* blocks = (SVgDataBlocks*)taosArrayGetP(vgs, i);
+    vgroupInfoToEpSet(&blocks->vg, &subplan->execEpSet);
+    subplan->pNode = NULL;
+    subplan->pDataSink = createDataInserter(pCxt, blocks);
+  }
+}
+
 static void createSubplanByLevel(SPlanContext* pCxt, SQueryPlanNode* pRoot) {
-  SSubplan* subplan = initSubplan(pCxt, QUERY_TYPE_MERGE);
+  if (QNODE_INSERT == pRoot->info.type) {
-  ++(pCxt->nextId.templateId);
+    splitInsertSubplan(pCxt, pRoot);
-  subplan->pNode = createPhyNode(pCxt, pRoot);
+  } else {
+    SSubplan* subplan = initSubplan(pCxt, QUERY_TYPE_MERGE);
+    ++(pCxt->nextId.templateId);
+    subplan->pNode = createPhyNode(pCxt, pRoot);
+    subplan->pDataSink = createDataDispatcher(pCxt, pRoot);
+  }
   // todo deal subquery
 }
 
 int32_t createDag(SQueryPlanNode* pQueryNode, struct SCatalog* pCatalog, SQueryDag** pDag) {
-  SPlanContext context = {
+  TRY(TSDB_MAX_TAG_CONDITIONS) {
-    .pCatalog = pCatalog,
+    SPlanContext context = {
-    .pDag = calloc(1, sizeof(SQueryDag)),
+      .pCatalog = pCatalog,
-    .pCurrentSubplan = NULL,
+      .pDag = vailidPointer(calloc(1, sizeof(SQueryDag))),
-    .nextId = {0} // todo queryid
+      .pCurrentSubplan = NULL,
-  };
+      .nextId = {0} // todo queryid
-  if (NULL == context.pDag) {
+    };
-    return TSDB_CODE_TSC_OUT_OF_MEMORY;
+    *pDag = context.pDag;
-  }
+    context.pDag->pSubplans = vailidPointer(taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES));
-  context.pDag->pSubplans = taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES);
+    createSubplanByLevel(&context, pQueryNode);
-  createSubplanByLevel(&context, pQueryNode);
+  } CATCH(code) {
-  *pDag = context.pDag;
+    CLEANUP_EXECUTE();
+    terrno = code;
+    return TSDB_CODE_FAILED;
+  } END_TRY
   return TSDB_CODE_SUCCESS;
 }
 

source/libs/planner/src/physicalPlanJson.c

@@ -651,7 +651,7 @@ static bool specificPhyNodeFromJson(const cJSON* json, void* obj) {
 static const char* jkPnodeName = "Name";
 static const char* jkPnodeTargets = "Targets";
 static const char* jkPnodeConditions = "Conditions";
-static const char* jkPnodeSchema = "Schema";
+static const char* jkPnodeSchema = "InputSchema";
 static const char* jkPnodeChildren = "Children";
 // The 'pParent' field do not need to be serialized.
 static bool phyNodeToJson(const void* obj, cJSON* jNode) {
@@ -695,6 +695,70 @@ static bool phyNodeFromJson(const cJSON* json, void* obj) {
   return res;
 }
 
+static const char* jkInserterNumOfTables = "NumOfTables";
+static const char* jkInserterDataSize = "DataSize";
+
+static bool inserterToJson(const void* obj, cJSON* json) {
+  const SDataInserter* inserter = (const SDataInserter*)obj;
+  bool res = cJSON_AddNumberToObject(json, jkInserterNumOfTables, inserter->numOfTables);
+  if (res) {
+    res = cJSON_AddNumberToObject(json, jkInserterDataSize, inserter->size);
+  }
+  // todo pData
+  return res;
+}
+
+static bool inserterFromJson(const cJSON* json, void* obj) {
+  SDataInserter* inserter = (SDataInserter*)obj;
+  inserter->numOfTables = getNumber(json, jkInserterNumOfTables);
+  inserter->size = getNumber(json, jkInserterDataSize);
+  // todo pData
+}
+
+static bool specificDataSinkToJson(const void* obj, cJSON* json) {
+  const SDataSink* dsink = (const SDataSink*)obj;
+  switch (dsink->info.type) {
+    case DSINK_Dispatch:
+      return true;
+    case DSINK_Insert:
+      return inserterToJson(obj, json);
+    default:
+      break;
+  }
+  return false;
+}
+
+static bool specificDataSinkFromJson(const cJSON* json, void* obj) {
+  SDataSink* dsink = (SDataSink*)obj;
+  switch (dsink->info.type) {
+    case DSINK_Dispatch:
+      return true;
+    case DSINK_Insert:
+      return inserterFromJson(json, obj);
+    default:
+      break;
+  }
+  return false;
+}
+
+static const char* jkDataSinkName = "Name";
+
+static bool dataSinkToJson(const void* obj, cJSON* json) {
+  const SDataSink* dsink = (const SDataSink*)obj;
+  bool res = cJSON_AddStringToObject(json, jkDataSinkName, dsink->info.name);
+  if (res) {
+    res = addObject(json, dsink->info.name, specificDataSinkToJson, dsink);
+  }
+  return res;
+}
+
+static bool dataSinkFromJson(const cJSON* json, void* obj) {
+  SDataSink* dsink = (SDataSink*)obj;
+  dsink->info.name = getString(json, jkDataSinkName);
+  dsink->info.type = dsinkNameToDsinkType(dsink->info.name);
+  return fromObject(json, dsink->info.name, specificDataSinkFromJson, dsink, true);
+}
+
 static const char* jkIdQueryId = "QueryId";
 static const char* jkIdTemplateId = "TemplateId";
 static const char* jkIdSubplanId = "SubplanId";
@@ -721,6 +785,7 @@ static bool subplanIdFromJson(const cJSON* json, void* obj) {
 
 static const char* jkSubplanId = "Id";
 static const char* jkSubplanNode = "Node";
+static const char* jkSubplanDataSink = "DataSink";
 
 static cJSON* subplanToJson(const SSubplan* subplan) {
   cJSON* jSubplan = cJSON_CreateObject();
@@ -734,6 +799,9 @@ static cJSON* subplanToJson(const SSubplan* subplan) {
   if (res) {
     res = addObject(jSubplan, jkSubplanNode, phyNodeToJson, subplan->pNode);
   }
+  if (res) {
+    res = addObject(jSubplan, jkSubplanDataSink, dataSinkToJson, subplan->pDataSink);
+  }
 
   if (!res) {
     cJSON_Delete(jSubplan);
@@ -751,6 +819,9 @@ static SSubplan* subplanFromJson(const cJSON* json) {
   if (res) {
     res = fromObjectWithAlloc(json, jkSubplanNode, phyNodeFromJson, (void**)&subplan->pNode, sizeof(SPhyNode), false);
   }
+  if (res) {
+    res = fromObjectWithAlloc(json, jkSubplanDataSink, dataSinkFromJson, (void**)&subplan->pDataSink, sizeof(SDataSink), false);
+  }
   
   if (!res) {
     qDestroySubplan(subplan);
@@ -759,13 +830,22 @@ static SSubplan* subplanFromJson(const cJSON* json) {
   return subplan;
 }
 
-int32_t subPlanToString(const SSubplan* subplan, char** str) {
+int32_t subPlanToString(const SSubplan* subplan, char** str, int32_t* len) {
+  if (QUERY_TYPE_MODIFY == subplan->type) {
+    SDataInserter* insert = (SDataInserter*)(subplan->pDataSink);
+    *len = insert->size;
+    *str = insert->pData;
+    insert->pData == NULL;
+    return TSDB_CODE_SUCCESS;
+  }
+
   cJSON* json = subplanToJson(subplan);
   if (NULL == json) {
     terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
     return TSDB_CODE_FAILED;
   }
   *str = cJSON_Print(json);
+  *len = strlen(*str);
   return TSDB_CODE_SUCCESS;
 }
 

source/libs/planner/src/planner.c

@@ -24,9 +24,9 @@ void qDestroyQueryDag(struct SQueryDag* pDag) {
   // todo
 }
 
-int32_t qCreateQueryDag(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* pQnode, struct SQueryDag** pDag) {
+int32_t qCreateQueryDag(const struct SQueryNode* pNode, struct SEpSet* pEpSet, struct SQueryDag** pDag) {
   SQueryPlanNode* logicPlan;
-  int32_t code = createQueryPlan(pQueryInfo, &logicPlan);
+  int32_t code = createQueryPlan(pNode, &logicPlan);
   if (TSDB_CODE_SUCCESS != code) {
     destroyQueryPlan(logicPlan);
     return code;
@@ -50,8 +50,8 @@ int32_t qSetSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SEpAddr
   return setSubplanExecutionNode(subplan, templateId, ep);
 }
 
-int32_t qSubPlanToString(const SSubplan *subplan, char** str) {
+int32_t qSubPlanToString(const SSubplan *subplan, char** str, int32_t* len) {
-  return subPlanToString(subplan, str);
+  return subPlanToString(subplan, str, len);
 }
 
 int32_t qStringToSubplan(const char* str, SSubplan** subplan) {

source/libs/planner/test/phyPlanTests.cpp

@@ -62,8 +62,9 @@ protected:
       size_t num = taosArrayGetSize(subplans);
       for (size_t j = 0; j < num; ++j) {
         std::cout << "no " << j << ":" << std::endl;
+        int32_t len = 0;
         char* str = nullptr;
-        ASSERT_EQ (TSDB_CODE_SUCCESS, qSubPlanToString((const SSubplan*)taosArrayGetP(subplans, j), &str));
+        ASSERT_EQ (TSDB_CODE_SUCCESS, qSubPlanToString((const SSubplan*)taosArrayGetP(subplans, j), &str, &len));
         std::cout << str << std::endl;
         free(str);
       }

source/libs/qcom/CMakeLists.txt

@@ -10,3 +10,5 @@ target_link_libraries(
    qcom
    PRIVATE os util transport
 )
+
+ADD_SUBDIRECTORY(test)

source/libs/qcom/src/queryUtil.c

@@ -1,5 +1,8 @@
 #include "os.h"
 #include "taosmsg.h"
+#include "query.h"
+#include "tglobal.h"
+#include "tsched.h"
 
 #define VALIDNUMOFCOLS(x)  ((x) >= TSDB_MIN_COLUMNS && (x) <= TSDB_MAX_COLUMNS)
 #define VALIDNUMOFTAGS(x)  ((x) >= 0 && (x) <= TSDB_MAX_TAGS)
@@ -75,4 +78,46 @@ bool tIsValidSchema(struct SSchema* pSchema, int32_t numOfCols, int32_t numOfTag
   }
 
   return true;
-}
+}
+
+static void* pTaskQueue = NULL;
+
+int32_t initTaskQueue() {
+  double factor = 4.0;
+  int32_t numOfThreads = MAX((int)(tsNumOfCores * tsNumOfThreadsPerCore / factor), 2);
+
+  int32_t queueSize = tsMaxConnections * 2;
+  pTaskQueue = taosInitScheduler(queueSize, numOfThreads, "tsc");
+  if (NULL == pTaskQueue) {
+    qError("failed to init task queue");
+    return -1;
+  }
+
+  qDebug("task queue is initialized, numOfThreads: %d", numOfThreads);
+}
+
+int32_t cleanupTaskQueue() {
+  taosCleanUpScheduler(pTaskQueue);
+}
+
+static void execHelper(struct SSchedMsg* pSchedMsg) {
+  assert(pSchedMsg != NULL && pSchedMsg->ahandle != NULL);
+
+  __async_exec_fn_t* execFn = (__async_exec_fn_t*) pSchedMsg->ahandle;
+  int32_t code = execFn(pSchedMsg->thandle);
+  if (code != 0 && pSchedMsg->msg != NULL) {
+    *(int32_t*) pSchedMsg->msg = code;
+  }
+}
+
+int32_t taosAsyncExec(__async_exec_fn_t execFn, void* execParam, int32_t* code) {
+  assert(execFn != NULL);
+
+  SSchedMsg schedMsg = {0};
+  schedMsg.fp      = execHelper;
+  schedMsg.ahandle = execFn;
+  schedMsg.thandle = execParam;
+  schedMsg.msg     = code;
+
+  taosScheduleTask(pTaskQueue, &schedMsg);
+}

source/libs/qcom/test/CMakeLists.txt

@@ -0,0 +1,19 @@
+
+MESSAGE(STATUS "build qcom unit test")
+
+# GoogleTest requires at least C++11
+SET(CMAKE_CXX_STANDARD 11)
+AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
+
+ADD_EXECUTABLE(queryUtilTest ${SOURCE_LIST})
+
+TARGET_INCLUDE_DIRECTORIES(
+        queryUtilTest
+        PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/qcom/"
+        PRIVATE "${CMAKE_SOURCE_DIR}/source/libs/qcom/inc"
+)
+
+TARGET_LINK_LIBRARIES(
+        queryUtilTest
+        PUBLIC os util gtest qcom common
+)

source/libs/qcom/test/queryTest.cpp

@@ -0,0 +1,83 @@
+/*
+ * 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/>.
+ */
+
+#include <gtest/gtest.h>
+#include <iostream>
+#include "taosmsg.h"
+#include "query.h"
+#pragma GCC diagnostic ignored "-Wwrite-strings"
+
+#pragma GCC diagnostic ignored "-Wunused-function"
+#pragma GCC diagnostic ignored "-Wunused-variable"
+#pragma GCC diagnostic ignored "-Wsign-compare"
+
+namespace {
+typedef struct SParam {
+  int32_t v;
+} SParam;
+int32_t testPrint(void* p) {
+  SParam* param = (SParam*)p;
+  printf("hello world, %d\n", param->v);
+  tfree(p);
+  return 0;
+}
+
+int32_t testPrintError(void* p) {
+  SParam* param = (SParam*) p;
+  tfree(p);
+
+  return -1;
+}
+}  // namespace
+
+class QueryTestEnv : public testing::Environment {
+ public:
+  virtual void SetUp() { initTaskQueue(); }
+
+  virtual void TearDown() { cleanupTaskQueue(); }
+
+  QueryTestEnv() {}
+  virtual ~QueryTestEnv() {}
+};
+
+int main(int argc, char** argv) {
+  testing::AddGlobalTestEnvironment(new QueryTestEnv());
+  testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}
+
+TEST(testCase, async_task_test) {
+  SParam* p = (SParam*)calloc(1, sizeof(SParam));
+  taosAsyncExec(testPrint, p, NULL);
+  usleep(5000);
+}
+
+TEST(testCase, many_async_task_test) {
+  for(int32_t i = 0; i < 50; ++i) {
+    SParam* p = (SParam*) calloc(1, sizeof(SParam));
+    p->v = i;
+    taosAsyncExec(testPrint, p, NULL);
+  }
+
+  usleep(10000);
+}
+
+TEST(testCase, error_in_async_test) {
+  int32_t code = 0;
+  SParam* p = (SParam*) calloc(1, sizeof(SParam));
+  taosAsyncExec(testPrintError, p, &code);
+  usleep(1000);
+  printf("Error code:%d after asynchronously exec function\n", code);
+}

source/libs/scheduler/src/scheduler.c

@@ -521,8 +521,8 @@ _return:
 
 int32_t schLaunchTask(SQueryJob *job, SQueryTask *task) {
   SSubplan *plan = task->plan;
-  
+  int32_t len = 0;
-  SCH_ERR_RET(qSubPlanToString(plan, &task->msg));
+  SCH_ERR_RET(qSubPlanToString(plan, &task->msg, &len));
   if (plan->execEpSet.numOfEps <= 0) {
     SCH_ERR_RET(schSetTaskExecEpSet(job, &plan->execEpSet));
   }

source/libs/wal/src/walRead.c

@@ -170,6 +170,7 @@ int32_t walReadWithHandle(SWalReadHandle *pRead, int64_t ver) {
   return 0;
 }
 
+#if 0
 int32_t walRead(SWal *pWal, SWalHead **ppHead, int64_t ver) {
   int code;
   code = walSeekVer(pWal, ver);
@@ -207,6 +208,7 @@ int32_t walRead(SWal *pWal, SWalHead **ppHead, int64_t ver) {
   return 0;
 }
 
-/*int32_t walReadWithFp(SWal *pWal, FWalWrite writeFp, int64_t verStart, int32_t readNum) {*/
+int32_t walReadWithFp(SWal *pWal, FWalWrite writeFp, int64_t verStart, int32_t readNum) {
-/*return 0;*/
+return 0;
-/*}*/
+}
+#endif