Merge branch '3.0' into feature/tq

cmake/libuv_CMakeLists.txt.in

@@ -4,9 +4,9 @@ ExternalProject_Add(libuv
         GIT_REPOSITORY https://github.com/libuv/libuv.git
         GIT_TAG v1.42.0
         SOURCE_DIR "${CMAKE_CONTRIB_DIR}/libuv"
-        BINARY_DIR ""
-        CONFIGURE_COMMAND ""
-        BUILD_COMMAND ""
+        BINARY_DIR "${CMAKE_CONTRIB_DIR}/libuv"
+        CONFIGURE_COMMAND "" 
+        BUILD_COMMAND ""  
         INSTALL_COMMAND ""
         TEST_COMMAND ""
-        )
+        )
\ No newline at end of file

include/common/common.h

@@ -62,6 +62,12 @@ typedef struct SConstantItem {
   SVariant    value;
 } SConstantItem;
 
+typedef struct {
+  uint32_t  numOfTables;
+  SArray   *pGroupList;
+  SHashObj *map;  // speedup acquire the tableQueryInfo by table uid
+} STableGroupInfo;
+
 typedef struct SSDataBlock {
   SColumnDataAgg *pBlockAgg;
   SArray         *pDataBlock;    // SArray<SColumnInfoData>

include/common/tmsg.h

@@ -1020,6 +1020,14 @@ typedef struct {
   char     msg[];
 } SSubQueryMsg;
 
+typedef struct {
+  SMsgHead header;
+  uint64_t sId;
+  uint64_t queryId;
+  uint64_t taskId;
+} SSinkDataReq;
+
+
 typedef struct {
   SMsgHead header;
   uint64_t sId;

include/dnode/bnode/bnode.h

@@ -23,22 +23,18 @@ extern "C" {
 /* ------------------------ TYPES EXPOSED ------------------------ */
 typedef struct SDnode SDnode;
 typedef struct SBnode SBnode;
-typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
-typedef int32_t (*SendReqToMnodeFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
-typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
+typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *pMsg);
+typedef int32_t (*SendReqToMnodeFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
+typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
 
 typedef struct {
   int64_t numOfErrors;
 } SBnodeLoad;
 
 typedef struct {
-  int32_t sver;
-} SBnodeCfg;
-
-typedef struct {
+  int32_t           sver;
   int32_t           dnodeId;
   int64_t           clusterId;
-  SBnodeCfg         cfg;
   SDnode           *pDnode;
   SendReqToDnodeFp  sendReqToDnodeFp;
   SendReqToMnodeFp  sendReqToMnodeFp;

include/dnode/mgmt/dnode.h

@@ -22,15 +22,39 @@
 extern "C" {
 #endif
 
-/* ------------------------ TYPES EXPOSED ------------------------ */
+/* ------------------------ TYPES EXPOSED ---------------- */
 typedef struct SDnode SDnode;
 
+/* ------------------------ Environment ------------------ */
+typedef struct {
+  int32_t sver;
+  int32_t numOfCores;
+  int16_t numOfCommitThreads;
+  int8_t  enableTelem;
+  char    timezone[TSDB_TIMEZONE_LEN];
+  char    locale[TSDB_LOCALE_LEN];
+  char    charset[TSDB_LOCALE_LEN];
+  char    buildinfo[64];
+  char    gitinfo[48];
+} SDnodeEnvCfg;
+
+/**
+ * @brief Initialize the environment
+ *
+ * @param pOption Option of the environment
+ * @return int32_t 0 for success and -1 for failure
+ */
+int32_t dndInit(const SDnodeEnvCfg *pCfg);
+
+/**
+ * @brief clear the environment
+ *
+ */
+void dndCleanup();
+
+/* ------------------------ SDnode ----------------------- */
 typedef struct {
-  int32_t  sver;
-  int32_t  numOfCores;
   int32_t  numOfSupportVnodes;
-  int16_t  numOfCommitThreads;
-  int8_t   enableTelem;
   int32_t  statusInterval;
   float    numOfThreadsPerCore;
   float    ratioOfQueryCores;
@@ -41,28 +65,22 @@ typedef struct {
   char     localEp[TSDB_EP_LEN];
   char     localFqdn[TSDB_FQDN_LEN];
   char     firstEp[TSDB_EP_LEN];
-  char     timezone[TSDB_TIMEZONE_LEN];
-  char     locale[TSDB_LOCALE_LEN];
-  char     charset[TSDB_LOCALE_LEN];
-  char     buildinfo[64];
-  char     gitinfo[48];
-} SDnodeOpt;
+} SDnodeObjCfg;
 
-/* ------------------------ SDnode ------------------------ */
 /**
  * @brief Initialize and start the dnode.
  *
- * @param pOption Option of the dnode.
+ * @param pCfg Config of the dnode.
  * @return SDnode* The dnode object.
  */
-SDnode *dndInit(SDnodeOpt *pOption);
+SDnode *dndCreate(SDnodeObjCfg *pCfg);
 
 /**
  * @brief Stop and cleanup the dnode.
  *
  * @param pDnode The dnode object to close.
  */
-void dndCleanup(SDnode *pDnode);
+void dndClose(SDnode *pDnode);
 
 #ifdef __cplusplus
 }

include/dnode/qnode/qnode.h

@@ -23,9 +23,9 @@ extern "C" {
 /* ------------------------ TYPES EXPOSED ------------------------ */
 typedef struct SDnode SDnode;
 typedef struct SQnode SQnode;
-typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
-typedef int32_t (*SendReqToMnodeFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
-typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
+typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *pMsg);
+typedef int32_t (*SendReqToMnodeFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
+typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
 
 typedef struct {
   int64_t numOfStartTask;
@@ -39,13 +39,9 @@ typedef struct {
 } SQnodeLoad;
 
 typedef struct {
-  int32_t sver;
-} SQnodeCfg;
-
-typedef struct {
+  int32_t           sver;
   int32_t           dnodeId;
   int64_t           clusterId;
-  SQnodeCfg         cfg;
   SDnode           *pDnode;
   SendReqToDnodeFp  sendReqToDnodeFp;
   SendReqToMnodeFp  sendReqToMnodeFp;

include/dnode/snode/snode.h

@@ -23,22 +23,18 @@ extern "C" {
 /* ------------------------ TYPES EXPOSED ------------------------ */
 typedef struct SDnode SDnode;
 typedef struct SSnode SSnode;
-typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
-typedef int32_t (*SendReqToMnodeFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
-typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
+typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *pMsg);
+typedef int32_t (*SendReqToMnodeFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
+typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
 
 typedef struct {
   int64_t numOfErrors;
 } SSnodeLoad;
 
 typedef struct {
-  int32_t sver;
-} SSnodeCfg;
-
-typedef struct {
+  int32_t           sver;
   int32_t           dnodeId;
   int64_t           clusterId;
-  SSnodeCfg         cfg;
   SDnode           *pDnode;
   SendReqToDnodeFp  sendReqToDnodeFp;
   SendReqToMnodeFp  sendReqToMnodeFp;

source/libs/executor/inc/dataSinkMgt.h → include/libs/executor/dataSinkMgt.h

@@ -21,36 +21,38 @@ extern "C" {
 #endif
 
 #include "os.h"
-#include "executorimpl.h"
+#include "thash.h"
+#include "executor.h"
 
-#define DS_CAPACITY_ENOUGH 1
-#define DS_CAPACITY_FULL   2
-#define DS_NEED_SCHEDULE   3
-#define DS_END             4
-#define DS_IN_PROCESS      5
+#define DS_BUF_LOW   1
+#define DS_BUF_FULL  2
+#define DS_BUF_EMPTY 3
 
 struct SDataSink;
 struct SSDataBlock;
 
 typedef struct SDataSinkMgtCfg {
-  uint32_t maxDataBlockNum;
+  uint32_t maxDataBlockNum;           // todo: this should be numOfRows?
   uint32_t maxDataBlockNumPerQuery;
 } SDataSinkMgtCfg;
 
 int32_t dsDataSinkMgtInit(SDataSinkMgtCfg *cfg);
 
-typedef void* DataSinkHandle;
-
 typedef struct SInputData {
-  const SSDataBlock* pData;
+  const struct SSDataBlock* pData;
   SHashObj* pTableRetrieveTsMap;
 } SInputData;
 
-typedef struct SOutPutData {
+typedef struct SOutputData {
   int32_t numOfRows;
   int8_t  compressed;
   char*   pData;
-} SOutPutData;
+  bool    queryEnd;
+  bool    needSchedule;
+  int32_t bufStatus;
+  int64_t useconds;
+  int8_t  precision;
+} SOutputData;
 
 /**
  * Create a subplan's datasinker handle for all later operations. 
@@ -66,16 +68,16 @@ int32_t dsCreateDataSinker(const struct SDataSink *pDataSink, DataSinkHandle* pH
  * @param pRes
  * @return error code
  */
-int32_t dsPutDataBlock(DataSinkHandle handle, const SInputData* pInput, int32_t* pStatus);
+int32_t dsPutDataBlock(DataSinkHandle handle, const SInputData* pInput, bool* pContinue);
 
-void dsEndPut(DataSinkHandle handle);
+void dsEndPut(DataSinkHandle handle, int64_t useconds);
 
 /**
  * Get the length of the data returned by the next call to dsGetDataBlock.
  * @param handle
- * @return data length
+ * @param pLen data length
  */
-int32_t dsGetDataLength(DataSinkHandle handle, int32_t* pStatus);
+void dsGetDataLength(DataSinkHandle handle, int32_t* pLen, bool* pQueryEnd);
 
 /**
  * Get data, the caller needs to allocate data memory.
@@ -84,7 +86,7 @@ int32_t dsGetDataLength(DataSinkHandle handle, int32_t* pStatus);
  * @param pStatus output
  * @return error code
  */
-int32_t dsGetDataBlock(DataSinkHandle handle, SOutPutData* pOutput, int32_t* pStatus);
+int32_t dsGetDataBlock(DataSinkHandle handle, SOutputData* pOutput);
 
 /**
  * After dsGetStatus returns DS_NEED_SCHEDULE, the caller need to put this into the work queue.

include/libs/executor/executor.h

@@ -21,24 +21,30 @@ extern "C" {
 #endif
 
 typedef void* qTaskInfo_t;
+typedef void* DataSinkHandle;
+struct SSubplan;
 
-/**
- * create the qinfo object according to QueryTableMsg
- * @param tsdb
- * @param pQueryTableMsg
- * @param pTaskInfo
- * @return
- */
-int32_t qCreateTask(void* tsdb, int32_t vgId, void* pQueryTableMsg, qTaskInfo_t* pTaskInfo, uint64_t qId);
+
+ /**
+  * Create the exec task object according to task json
+  * @param tsdb
+  * @param vgId
+  * @param pTaskInfoMsg
+  * @param pTaskInfo
+  * @param qId
+  * @return
+  */
+int32_t qCreateExecTask(void* tsdb, int32_t vgId, struct SSubplan* pPlan, qTaskInfo_t* pTaskInfo);
 
 /**
- * the main query execution function, including query on both table and multiple tables,
+ * The main task execution function, including query on both table and multiple tables,
  * which are decided according to the tag or table name query conditions
  *
- * @param qinfo
+ * @param tinfo
+ * @param handle
  * @return
  */
-bool qExecTask(qTaskInfo_t qinfo, uint64_t *qId);
+int32_t qExecTask(qTaskInfo_t tinfo, DataSinkHandle* handle);
 
 /**
  * Retrieve the produced results information, if current query is not paused or completed,
@@ -60,7 +66,7 @@ int32_t qRetrieveQueryResultInfo(qTaskInfo_t qinfo, bool* buildRes, void* pRspCo
  * @param contLen payload length
  * @return
  */
-int32_t qDumpRetrieveResult(qTaskInfo_t qinfo, SRetrieveTableRsp** pRsp, int32_t* contLen, bool* continueExec);
+//int32_t qDumpRetrieveResult(qTaskInfo_t qinfo, SRetrieveTableRsp** pRsp, int32_t* contLen, bool* continueExec);
 
 /**
  * return the transporter context (RPC)
@@ -81,7 +87,7 @@ int32_t qKillTask(qTaskInfo_t qinfo);
  * @param qinfo
  * @return
  */
-int32_t qIsQueryCompleted(qTaskInfo_t qinfo);
+int32_t qIsTaskCompleted(qTaskInfo_t qinfo);
 
 /**
  * destroy query info structure
@@ -113,7 +119,7 @@ int32_t qGetQualifiedTableIdList(void* pTableList, const char* tagCond, int32_t
  * @param numOfIndex
  * @return
  */
-int32_t qCreateTableGroupByGroupExpr(SArray* pTableIdList, TSKEY skey, STableGroupInfo groupInfo, SColIndex* groupByIndex, int32_t numOfIndex);
+//int32_t qCreateTableGroupByGroupExpr(SArray* pTableIdList, TSKEY skey, STableGroupInfo groupInfo, SColIndex* groupByIndex, int32_t numOfIndex);
 
 /**
  * Update the table id list of a given query.

include/libs/parser/parsenodes.h

@@ -43,15 +43,6 @@ typedef struct SField {
   int32_t  bytes;
 } SField;
 
-typedef struct SParseBasicCtx {
-  uint64_t         requestId;
-  int32_t          acctId;
-  const char      *db;
-  void            *pTransporter;
-  SEpSet           mgmtEpSet;
-  struct SCatalog *pCatalog;
-} SParseBasicCtx;
-
 typedef struct SFieldInfo {
   int16_t     numOfOutput;   // number of column in result
   SField     *final;

include/libs/parser/parser.h

@@ -23,11 +23,17 @@ extern "C" {
 #include "parsenodes.h"
 
 typedef struct SParseContext {
-  SParseBasicCtx   ctx;
+  uint64_t         requestId;
+  int32_t          acctId;
+  const char      *db;
+  void            *pTransporter;
+  SEpSet           mgmtEpSet;
   const char      *pSql;           // sql string
   size_t           sqlLen;         // length of the sql string
   char            *pMsg;           // extended error message if exists to help identifying the problem in sql statement.
   int32_t          msgLen;         // max length of the msg
+
+  struct SCatalog *pCatalog;
 } SParseContext;
 
 /**

include/libs/planner/planner.h

@@ -91,8 +91,10 @@ typedef struct SPhyNode {
 
 typedef struct SScanPhyNode {
   SPhyNode    node;
-  uint64_t    uid;  // unique id of the table
+  uint64_t    uid;           // unique id of the table
   int8_t      tableType;
+  int32_t     order;         // scan order: TSDB_ORDER_ASC|TSDB_ORDER_DESC
+  int32_t     count;         // repeat count
 } SScanPhyNode;
 
 typedef SScanPhyNode SSystemTableScanPhyNode;
@@ -117,6 +119,25 @@ typedef struct SExchangePhyNode {
   SArray     *pSrcEndPoints;  // SEpAddr, scheduler fill by calling qSetSuplanExecutionNode
 } SExchangePhyNode;
 
+typedef enum EAggAlgo {
+	AGG_ALGO_PLAIN = 1, // simple agg across all input rows
+	AGG_ALGO_SORTED,    // grouped agg, input must be sorted
+	AGG_ALGO_HASHED     // grouped agg, use internal hashtable
+} EAggAlgo;
+
+typedef enum EAggSplit {
+	AGG_SPLIT_PRE = 1, // first level agg, maybe don't need calculate the final result
+	AGG_SPLIT_FINAL    // second level agg, must calculate the final result
+} EAggSplit;
+
+typedef struct SAggPhyNode {
+  SPhyNode    node;
+  EAggAlgo    aggAlgo;  // algorithm used by agg operator 
+  EAggSplit   aggSplit; // distributed splitting mode
+  SArray     *pExprs;   // SExprInfo list, these are expression list of group_by_clause and parameter expression of aggregate function
+  SArray     *pGroupByList; // SColIndex list, but these must be column node
+} SAggPhyNode;
+
 typedef struct SSubplanId {
   uint64_t queryId;
   uint64_t templateId;

include/libs/planner/plannerOp.h

@@ -30,7 +30,7 @@ OP_ENUM_MACRO(TagScan)
 OP_ENUM_MACRO(SystemTableScan)
 OP_ENUM_MACRO(Aggregate)
 OP_ENUM_MACRO(Project)
-OP_ENUM_MACRO(Groupby)
+// OP_ENUM_MACRO(Groupby)
 OP_ENUM_MACRO(Limit)
 OP_ENUM_MACRO(SLimit)
 OP_ENUM_MACRO(TimeWindow)

include/libs/scheduler/scheduler.h

@@ -20,8 +20,10 @@
 extern "C" {
 #endif
 
-#include "planner.h"
 #include "catalog.h"
+#include "planner.h"
+
+struct SSchJob;
 
 typedef struct SSchedulerCfg {
   uint32_t maxJobNum;
@@ -65,7 +67,7 @@ int32_t schedulerInit(SSchedulerCfg *cfg);
  * @param nodeList  Qnode/Vnode address list, element is SQueryNodeAddr
  * @return
  */
-int32_t scheduleExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, void** pJob, SQueryResult *pRes);
+int32_t scheduleExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, struct SSchJob** pJob, SQueryResult *pRes);
 
 /**
  * Process the query job, generated according to the query physical plan.
@@ -73,7 +75,7 @@ int32_t scheduleExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, void
  * @param nodeList  Qnode/Vnode address list, element is SQueryNodeAddr
  * @return
  */
-int32_t scheduleAsyncExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, void** pJob);
+int32_t scheduleAsyncExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, struct SSchJob** pJob);
 
 /**
  * Fetch query result from the remote query executor
@@ -81,7 +83,7 @@ int32_t scheduleAsyncExecJob(void *transport, SArray *nodeList, SQueryDag* pDag,
  * @param data
  * @return
  */
-int32_t scheduleFetchRows(void *pJob, void **data);
+int32_t scheduleFetchRows(struct SSchJob *pJob, void **data);
 
 
 /**
@@ -89,7 +91,7 @@ int32_t scheduleFetchRows(void *pJob, void **data);
  * @param pJob
  * @return
  */
-int32_t scheduleCancelJob(void *pJob);
+//int32_t scheduleCancelJob(void *pJob);
 
 /**
  * Free the query job

include/libs/tfs/tfs.h

@@ -49,7 +49,7 @@ typedef struct {
 } STierMeta;
 
 int  tfsInit(SDiskCfg *pDiskCfg, int ndisk);
-void tfsDestroy();
+void tfsCleanup();
 void tfsUpdateInfo(SFSMeta *pFSMeta, STierMeta *tierMetas, int8_t numLevels);
 void tfsGetMeta(SFSMeta *pMeta);
 void tfsAllocDisk(int expLevel, int *level, int *id);

include/libs/tkv/tkv.h

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#ifndef _TD_TKV_H_
-#define _TD_TKV_H_
-
-#if 0
-#include "os.h"
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-// Types exported
-typedef struct STkvDb        STkvDb;
-typedef struct STkvOpts      STkvOpts;
-typedef struct STkvCache     STkvCache;
-typedef struct STkvReadOpts  STkvReadOpts;
-typedef struct STkvWriteOpts STkvWriteOpts;
-
-// DB operations
-STkvDb *tkvOpen(const STkvOpts *options, const char *path);
-void    tkvClose(STkvDb *db);
-void    tkvPut(STkvDb *db, const STkvWriteOpts *, const char *key, size_t keylen, const char *val, size_t vallen);
-char *  tkvGet(STkvDb *db, const STkvReadOpts *, const char *key, size_t keylen, size_t *vallen);
-void    tkvCommit(STkvDb *db);
-
-// DB options
-STkvOpts *tkvOptsCreate();
-void      tkvOptsDestroy(STkvOpts *);
-void      tkvOptionsSetCache(STkvOpts *, STkvCache *);
-void      tkvOptsSetCreateIfMissing(STkvOpts *, unsigned char);
-
-// DB cache
-typedef enum { TKV_LRU_CACHE = 0, TKV_LFU_CACHE = 1 } ETkvCacheType;
-STkvCache *tkvCacheCreate(size_t capacity, ETkvCacheType type);
-void       tkvCacheDestroy(STkvCache *);
-
-// STkvReadOpts
-STkvReadOpts *tkvReadOptsCreate();
-void          tkvReadOptsDestroy(STkvReadOpts *);
-
-// STkvWriteOpts
-STkvWriteOpts *tkvWriteOptsCreate();
-void           tkvWriteOptsDestroy(STkvWriteOpts *);
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif
-#endif /*_TD_TKV_H_*/
\ No newline at end of file

include/util/taoserror.h

@@ -70,6 +70,8 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_INVALID_MSG                   TAOS_DEF_ERROR_CODE(0, 0x0108)
 #define TSDB_CODE_MSG_NOT_PROCESSED             TAOS_DEF_ERROR_CODE(0, 0x0109)
 #define TSDB_CODE_INVALID_PARA                  TAOS_DEF_ERROR_CODE(0, 0x010A)
+#define TSDB_CODE_REPEAT_INIT                   TAOS_DEF_ERROR_CODE(0, 0x010B)
+
 #define TSDB_CODE_REF_NO_MEMORY                 TAOS_DEF_ERROR_CODE(0, 0x0110)
 #define TSDB_CODE_REF_FULL                      TAOS_DEF_ERROR_CODE(0, 0x0111)
 #define TSDB_CODE_REF_ID_REMOVED                TAOS_DEF_ERROR_CODE(0, 0x0112)

include/util/tarray.h

@@ -23,10 +23,23 @@ extern "C" {
 #include "os.h"
 #include "talgo.h"
 
+#if 0
+#define TARRAY(TYPE)             \
+  struct {                       \
+    int32_t      tarray_size_;   \
+    int32_t      tarray_neles_;  \
+    struct TYPE* td_array_data_; \
+  }
+
+#define TARRAY_SIZE(ARRAY) (ARRAY)->tarray_size_
+#define TARRAY_NELES(ARRAY) (ARRAY)->tarray_neles_
+#define TARRAY_ELE_AT(ARRAY, IDX) ((ARRAY)->td_array_data_ + idx)
+#endif
+
 #define TARRAY_MIN_SIZE 8
 #define TARRAY_GET_ELEM(array, index) ((void*)((char*)((array)->pData) + (index) * (array)->elemSize))
-#define TARRAY_ELEM_IDX(array, ele)   (POINTER_DISTANCE(ele, (array)->pData) / (array)->elemSize)
-#define TARRAY_GET_START(array)       ((array)->pData)
+#define TARRAY_ELEM_IDX(array, ele) (POINTER_DISTANCE(ele, (array)->pData) / (array)->elemSize)
+#define TARRAY_GET_START(array) ((array)->pData)
 
 typedef struct SArray {
   size_t size;
@@ -57,7 +70,7 @@ int32_t taosArrayEnsureCap(SArray* pArray, size_t tsize);
  * @param nEles
  * @return
  */
-void *taosArrayAddBatch(SArray *pArray, const void *pData, int nEles);
+void* taosArrayAddBatch(SArray* pArray, const void* pData, int nEles);
 
 /**
  *
@@ -65,7 +78,7 @@ void *taosArrayAddBatch(SArray *pArray, const void *pData, int nEles);
  * @param pData           position array list
  * @param numOfElems      the number of removed position
  */
-void taosArrayRemoveBatch(SArray *pArray, const int32_t* pData, int32_t numOfElems);
+void taosArrayRemoveBatch(SArray* pArray, const int32_t* pData, int32_t numOfElems);
 
 /**
  *
@@ -73,7 +86,7 @@ void taosArrayRemoveBatch(SArray *pArray, const int32_t* pData, int32_t numOfEle
  * @param comparFn
  * @param fp
  */
-void taosArrayRemoveDuplicate(SArray *pArray, __compar_fn_t comparFn, void (*fp)(void*));
+void taosArrayRemoveDuplicate(SArray* pArray, __compar_fn_t comparFn, void (*fp)(void*));
 
 /**
  *  add all element from the source array list into the destination
@@ -242,19 +255,18 @@ int32_t taosArraySearchIdx(const SArray* pArray, const void* key, __compar_fn_t
  */
 char* taosArraySearchString(const SArray* pArray, const char* key, __compar_fn_t comparFn, int flags);
 
-
 /**
  * sort the pointer data in the array
  * @param pArray
- * @param compar 
- * @param param  
+ * @param compar
+ * @param param
  * @return
  */
 
-void taosArraySortPWithExt(SArray* pArray, __ext_compar_fn_t fn, const void *param);
+void taosArraySortPWithExt(SArray* pArray, __ext_compar_fn_t fn, const void* param);
 
 #ifdef __cplusplus
 }
 #endif
 
-#endif  /*_TD_UTIL_ARRAY_H*/
+#endif /*_TD_UTIL_ARRAY_H*/

source/client/inc/clientInt.h

@@ -133,13 +133,12 @@ typedef struct SReqResultInfo {
   const char  *pData;
   TAOS_FIELD  *fields;
   uint32_t     numOfCols;
-
   int32_t     *length;
   TAOS_ROW     row;
   char       **pCol;
-
   uint32_t     numOfRows;
   uint32_t     current;
+  bool         completed;
 } SReqResultInfo;
 
 typedef struct SShowReqInfo {
@@ -153,6 +152,7 @@ typedef struct SRequestSendRecvBody {
   tsem_t          rspSem;        // not used now
   void*           fp;
   SShowReqInfo    showInfo;      // todo this attribute will be removed after the query framework being completed.
+  struct SSchJob *pQueryJob;     // query job, created according to sql query DAG.
   SDataBuf        requestMsg;
   SReqResultInfo  resInfo;
 } SRequestSendRecvBody;
@@ -169,7 +169,6 @@ typedef struct SRequestObj {
   char            *msgBuf;
   void            *pInfo;   // sql parse info, generated by parser module
   int32_t          code;
-  uint64_t         affectedRows;
   SQueryExecMetric metric;
   SRequestSendRecvBody body;
 } SRequestObj;
@@ -201,12 +200,13 @@ int  taos_options_imp(TSDB_OPTION option, const char *str);
 void* openTransporter(const char *user, const char *auth, int32_t numOfThreads);
 
 void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet);
+
 void initMsgHandleFp();
 
 TAOS *taos_connect_internal(const char *ip, const char *user, const char *pass, const char *auth, const char *db, uint16_t port);
-TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen);
 
 void *doFetchRow(SRequestObj* pRequest);
+
 void  setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t numOfCols, int32_t numOfRows);
 
 // --- heartbeat 

source/client/src/clientImpl.c

 
+#include "../../libs/scheduler/inc/schedulerInt.h"
 #include "clientInt.h"
 #include "clientLog.h"
+#include "parser.h"
+#include "planner.h"
+#include "scheduler.h"
 #include "tdef.h"
 #include "tep.h"
 #include "tglobal.h"
@@ -8,9 +12,6 @@
 #include "tnote.h"
 #include "tpagedfile.h"
 #include "tref.h"
-#include "parser.h"
-#include "planner.h"
-#include "scheduler.h"
 
 #define CHECK_CODE_GOTO(expr, lable) \
   do {                               \
@@ -24,8 +25,9 @@
 static int32_t initEpSetFromCfg(const char *firstEp, const char *secondEp, SCorEpSet *pEpSet);
 static SMsgSendInfo* buildConnectMsg(SRequestObj *pRequest);
 static void destroySendMsgInfo(SMsgSendInfo* pMsgBody);
+static void setQueryResultByRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* pRsp);
 
-static bool stringLengthCheck(const char* str, size_t maxsize) {
+  static bool stringLengthCheck(const char* str, size_t maxsize) {
   if (str == NULL) {
     return false;
   }
@@ -57,6 +59,7 @@ static char* getClusterKey(const char* user, const char* auth, const char* ip, i
 }
 
 static STscObj* taosConnectImpl(const char *ip, const char *user, const char *auth, const char *db, uint16_t port, __taos_async_fn_t fp, void *param, SAppInstInfo* pAppInfo);
+static void  setResSchemaInfo(SReqResultInfo* pResInfo, const SDataBlockSchema* pDataBlockSchema);
 
 TAOS *taos_connect_internal(const char *ip, const char *user, const char *pass, const char *auth, const char *db, uint16_t port) {
   if (taos_init() != TSDB_CODE_SUCCESS) {
@@ -149,23 +152,26 @@ int32_t parseSql(SRequestObj* pRequest, SQueryNode** pQuery) {
   STscObj* pTscObj = pRequest->pTscObj;
 
   SParseContext cxt = {
-    .ctx = {.requestId = pRequest->requestId, .acctId = pTscObj->acctId, .db = getConnectionDB(pTscObj), .pTransporter = pTscObj->pTransporter},
+    .requestId = pRequest->requestId,
+    .acctId = pTscObj->acctId,
+    .db     = getConnectionDB(pTscObj),
+    .pTransporter = pTscObj->pTransporter,
     .pSql   = pRequest->sqlstr,
     .sqlLen = pRequest->sqlLen,
     .pMsg   = pRequest->msgBuf,
     .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE
   };
 
-  cxt.ctx.mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
-  int32_t code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &cxt.ctx.pCatalog);
+  cxt.mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
+  int32_t code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &cxt.pCatalog);
   if (code != TSDB_CODE_SUCCESS) {
-    tfree(cxt.ctx.db);
+    tfree(cxt.db);
     return code;
   }
 
   code = qParseQuerySql(&cxt, pQuery);
 
-  tfree(cxt.ctx.db);
+  tfree(cxt.db);
   return code;
 }
 
@@ -198,27 +204,59 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQueryNode* pQuery) {
 
 int32_t getPlan(SRequestObj* pRequest, SQueryNode* pQueryNode, SQueryDag** pDag) {
   pRequest->type = pQueryNode->type;
-  return qCreateQueryDag(pQueryNode, pDag, pRequest->requestId);
+
+  SReqResultInfo* pResInfo = &pRequest->body.resInfo;
+  int32_t code = qCreateQueryDag(pQueryNode, pDag, pRequest->requestId);
+  if (code != 0) {
+    return code;
+  }
+
+  if (pQueryNode->type == TSDB_SQL_SELECT) {
+    SArray* pa = taosArrayGetP((*pDag)->pSubplans, 0);
+
+    SSubplan* pPlan = taosArrayGetP(pa, 0);
+    SDataBlockSchema* pDataBlockSchema = &(pPlan->pDataSink->schema);
+    setResSchemaInfo(pResInfo, pDataBlockSchema);
+
+    pRequest->type = TDMT_VND_QUERY;
+  }
+
+  return code;
+}
+
+void setResSchemaInfo(SReqResultInfo* pResInfo, const SDataBlockSchema* pDataBlockSchema) {
+  assert(pDataBlockSchema != NULL && pDataBlockSchema->numOfCols > 0);
+
+  pResInfo->numOfCols = pDataBlockSchema->numOfCols;
+  pResInfo->fields = calloc(pDataBlockSchema->numOfCols, sizeof(pDataBlockSchema->pSchema[0]));
+
+  for (int32_t i = 0; i < pResInfo->numOfCols; ++i) {
+    SSchema* pSchema = &pDataBlockSchema->pSchema[i];
+    pResInfo->fields[i].bytes = pSchema->bytes;
+    pResInfo->fields[i].type  = pSchema->type;
+    tstrncpy(pResInfo->fields[i].name, pSchema->name, tListLen(pResInfo->fields[i].name));
+  }
 }
 
-int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag, void** pJob) {
+int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag) {
   if (TSDB_SQL_INSERT == pRequest->type || TSDB_SQL_CREATE_TABLE == pRequest->type) {
     SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf};
 
-    int32_t code = scheduleExecJob(pRequest->pTscObj->pTransporter, NULL, pDag, pJob, &res);
+    int32_t code = scheduleExecJob(pRequest->pTscObj->pTransporter, NULL, pDag, &pRequest->body.pQueryJob, &res);
     if (code != TSDB_CODE_SUCCESS) {
       // handle error and retry
     } else {
-      if (*pJob != NULL) {
-        scheduleFreeJob(*pJob);
+      if (pRequest->body.pQueryJob != NULL) {
+        scheduleFreeJob(pRequest->body.pQueryJob);
       }
     }
 
-    pRequest->affectedRows = res.numOfRows;
-    return res.code;
+    pRequest->body.resInfo.numOfRows = res.numOfRows;
+    pRequest->code = res.code;
+    return pRequest->code;
   }
 
-  return scheduleAsyncExecJob(pRequest->pTscObj->pTransporter, NULL /*todo appInfo.xxx*/, pDag, pJob);
+  return scheduleAsyncExecJob(pRequest->pTscObj->pTransporter, NULL, pDag, &pRequest->body.pQueryJob);
 }
 
 typedef struct tmq_t tmq_t;
@@ -409,7 +447,6 @@ TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen) {
   SRequestObj *pRequest = NULL;
   SQueryNode  *pQuery   = NULL;
   SQueryDag   *pDag     = NULL;
-  void        *pJob     = NULL;
 
   terrno = TSDB_CODE_SUCCESS;
   CHECK_CODE_GOTO(buildRequest(pTscObj, sql, sqlLen, &pRequest), _return);
@@ -419,9 +456,8 @@ TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen) {
     CHECK_CODE_GOTO(execDdlQuery(pRequest, pQuery), _return);
   } else {
     CHECK_CODE_GOTO(getPlan(pRequest, pQuery, &pDag), _return);
-    CHECK_CODE_GOTO(scheduleQuery(pRequest, pDag, &pJob), _return);
+    CHECK_CODE_GOTO(scheduleQuery(pRequest, pDag), _return);
     pRequest->code = terrno;
-    return pRequest;
   }
 
 _return:
@@ -430,6 +466,7 @@ _return:
   if (NULL != pRequest && TSDB_CODE_SUCCESS != terrno) {
     pRequest->code = terrno;
   }
+
   return pRequest;
 }
 
@@ -628,7 +665,21 @@ void* doFetchRow(SRequestObj* pRequest) {
   SReqResultInfo* pResultInfo = &pRequest->body.resInfo;
 
   if (pResultInfo->pData == NULL || pResultInfo->current >= pResultInfo->numOfRows) {
-    if (pRequest->type == TDMT_MND_SHOW) {
+    if (pRequest->type == TDMT_VND_QUERY) {
+      // All data has returned to App already, no need to try again
+      if (pResultInfo->completed) {
+        return NULL;
+      }
+
+      scheduleFetchRows(pRequest->body.pQueryJob, (void **)&pRequest->body.resInfo.pData);
+      setQueryResultByRsp(&pRequest->body.resInfo, (SRetrieveTableRsp*)pRequest->body.resInfo.pData);
+
+      if (pResultInfo->numOfRows == 0) {
+        return NULL;
+      }
+
+      goto _return;
+    } else if (pRequest->type == TDMT_MND_SHOW) {
       pRequest->type = TDMT_MND_SHOW_RETRIEVE;
     } else if (pRequest->type == TDMT_VND_SHOW_TABLES) {
       pRequest->type = TDMT_VND_SHOW_TABLES_FETCH;
@@ -671,6 +722,8 @@ void* doFetchRow(SRequestObj* pRequest) {
     }
   }
 
+_return:
+
   for(int32_t i = 0; i < pResultInfo->numOfCols; ++i) {
     pResultInfo->row[i] = pResultInfo->pCol[i] + pResultInfo->fields[i].bytes * pResultInfo->current;
     if (IS_VAR_DATA_TYPE(pResultInfo->fields[i].type)) {
@@ -683,12 +736,23 @@ void* doFetchRow(SRequestObj* pRequest) {
   return pResultInfo->row;
 }
 
+static void doPrepareResPtr(SReqResultInfo* pResInfo) {
+  if (pResInfo->row == NULL) {
+    pResInfo->row    = calloc(pResInfo->numOfCols, POINTER_BYTES);
+    pResInfo->pCol   = calloc(pResInfo->numOfCols, POINTER_BYTES);
+    pResInfo->length = calloc(pResInfo->numOfCols, sizeof(int32_t));
+  }
+}
+
 void setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t numOfCols, int32_t numOfRows) {
   assert(numOfCols > 0 && pFields != NULL && pResultInfo != NULL);
   if (numOfRows == 0) {
     return;
   }
 
+  // todo check for the failure of malloc
+  doPrepareResPtr(pResultInfo);
+
   int32_t offset = 0;
   for (int32_t i = 0; i < numOfCols; ++i) {
     pResultInfo->length[i] = pResultInfo->fields[i].bytes;
@@ -714,3 +778,14 @@ void setConnectionDB(STscObj* pTscObj, const char* db) {
   pthread_mutex_unlock(&pTscObj->mutex);
 }
 
+void setQueryResultByRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* pRsp) {
+  assert(pResultInfo != NULL && pRsp != NULL);
+
+  pResultInfo->pRspMsg   = (const char*) pRsp;
+  pResultInfo->pData     = (void*) pRsp->data;
+  pResultInfo->numOfRows = htonl(pRsp->numOfRows);
+  pResultInfo->current   = 0;
+  pResultInfo->completed = (pRsp->completed == 1);
+
+  setResultDataPtr(pResultInfo, pResultInfo->fields, pResultInfo->numOfCols, pResultInfo->numOfRows);
+}
\ No newline at end of file

source/client/src/clientMain.c

@@ -265,7 +265,13 @@ const char *taos_data_type(int type) {
 const char *taos_get_client_info() { return version; }
 
 int taos_affected_rows(TAOS_RES *res) {
-  return ((SRequestObj*)res)->affectedRows;
+  if (res == NULL) {
+    return 0;
+  }
+
+  SRequestObj* pRequest = (SRequestObj*) res;
+  SReqResultInfo* pResInfo = &pRequest->body.resInfo;
+  return pResInfo->numOfRows;
 }
 
 int taos_result_precision(TAOS_RES *res) { return TSDB_TIME_PRECISION_MILLI; }

source/client/src/clientMsgHandler.c

@@ -157,9 +157,6 @@ int32_t processShowRsp(void* param, const SDataBuf* pMsg, int32_t code) {
 
   pResInfo->fields    = pFields;
   pResInfo->numOfCols = pMetaMsg->numOfColumns;
-  pResInfo->row       = calloc(pResInfo->numOfCols, POINTER_BYTES);
-  pResInfo->pCol      = calloc(pResInfo->numOfCols, POINTER_BYTES);
-  pResInfo->length    = calloc(pResInfo->numOfCols, sizeof(int32_t));
 
   pRequest->body.showInfo.execId = pShow->showId;
 

source/common/src/tglobal.c

@@ -37,7 +37,7 @@ uint16_t tsServerPort = 6030;
 int32_t  tsStatusInterval = 1;  // second
 int8_t   tsEnableTelemetryReporting = 0;
 char     tsEmail[TSDB_FQDN_LEN] = {0};
-int32_t  tsNumOfSupportVnodes = 16;
+int32_t  tsNumOfSupportVnodes = 128;
 
 // common
 int32_t tsRpcTimer = 300;

source/dnode/bnode/inc/bndInt.h

@@ -30,12 +30,7 @@ extern "C" {
 #endif
 
 typedef struct SBnode {
-  int32_t           dnodeId;
-  int64_t           clusterId;
-  SBnodeCfg         cfg;
-  SendReqToDnodeFp  sendReqToDnodeFp;
-  SendReqToMnodeFp  sendReqToMnodeFp;
-  SendRedirectRspFp sendRedirectRspFp;
+  SBnodeOpt opt;
 } SBnode;
 
 #ifdef __cplusplus

source/dnode/mgmt/daemon/src/daemon.c

@@ -28,11 +28,11 @@ static struct {
   bool printAuth;
   bool printVersion;
   char configDir[PATH_MAX];
-} global = {0};
+} dmn = {0};
 
 void dmnSigintHandle(int signum, void *info, void *ctx) {
   uInfo("singal:%d is received", signum);
-  global.stop = true;
+  dmn.stop = true;
 }
 
 void dmnSetSignalHandle() {
@@ -44,7 +44,7 @@ void dmnSetSignalHandle() {
 }
 
 int dmnParseOption(int argc, char const *argv[]) {
-  tstrncpy(global.configDir, "/etc/taos", PATH_MAX);
+  tstrncpy(dmn.configDir, "/etc/taos", PATH_MAX);
 
   for (int i = 1; i < argc; ++i) {
     if (strcmp(argv[i], "-c") == 0) {
@@ -53,19 +53,19 @@ int dmnParseOption(int argc, char const *argv[]) {
           printf("config file path overflow");
           return -1;
         }
-        tstrncpy(global.configDir, argv[i], PATH_MAX);
+        tstrncpy(dmn.configDir, argv[i], PATH_MAX);
       } else {
         printf("'-c' requires a parameter, default is %s\n", configDir);
         return -1;
       }
     } else if (strcmp(argv[i], "-C") == 0) {
-      global.dumpConfig = true;
+      dmn.dumpConfig = true;
     } else if (strcmp(argv[i], "-k") == 0) {
-      global.generateGrant = true;
+      dmn.generateGrant = true;
     } else if (strcmp(argv[i], "-A") == 0) {
-      global.printAuth = true;
+      dmn.printAuth = true;
     } else if (strcmp(argv[i], "-V") == 0) {
-      global.printVersion = true;
+      dmn.printVersion = true;
     } else {
     }
   }
@@ -92,7 +92,7 @@ void dmnPrintVersion() {
 }
 
 int dmnReadConfig(const char *path) {
-  tstrncpy(configDir, global.configDir, PATH_MAX);
+  tstrncpy(configDir, dmn.configDir, PATH_MAX);
   taosInitGlobalCfg();
   taosReadGlobalLogCfg();
 
@@ -114,12 +114,12 @@ int dmnReadConfig(const char *path) {
   }
 
   if (taosReadCfgFromFile() != 0) {
-    uError("failed to read global config");
+    uError("failed to read config");
     return -1;
   }
 
   if (taosCheckAndPrintCfg() != 0) {
-    uError("failed to check global config");
+    uError("failed to check config");
     return -1;
   }
 
@@ -131,38 +131,50 @@ void dmnDumpConfig() { taosDumpGlobalCfg(); }
 
 void dmnWaitSignal() {
   dmnSetSignalHandle();
-  while (!global.stop) {
+  while (!dmn.stop) {
     taosMsleep(100);
   }
 }
 
-void dmnInitOption(SDnodeOpt *pOption) {
-  pOption->sver = 30000000; //3.0.0.0
-  pOption->numOfCores = tsNumOfCores;
-  pOption->numOfSupportVnodes = tsNumOfSupportVnodes;
-  pOption->numOfCommitThreads = tsNumOfCommitThreads;
-  pOption->statusInterval = tsStatusInterval;
-  pOption->numOfThreadsPerCore = tsNumOfThreadsPerCore;
-  pOption->ratioOfQueryCores = tsRatioOfQueryCores;
-  pOption->maxShellConns = tsMaxShellConns;
-  pOption->shellActivityTimer = tsShellActivityTimer;
-  pOption->serverPort = tsServerPort;
-  tstrncpy(pOption->dataDir, tsDataDir, TSDB_FILENAME_LEN);
-  tstrncpy(pOption->localEp, tsLocalEp, TSDB_EP_LEN);
-  tstrncpy(pOption->localFqdn, tsLocalFqdn, TSDB_FQDN_LEN);
-  tstrncpy(pOption->firstEp, tsFirst, TSDB_EP_LEN);
-  tstrncpy(pOption->timezone, tsTimezone, TSDB_TIMEZONE_LEN);
-  tstrncpy(pOption->locale, tsLocale, TSDB_LOCALE_LEN);
-  tstrncpy(pOption->charset, tsCharset, TSDB_LOCALE_LEN);
-  tstrncpy(pOption->buildinfo, buildinfo, 64);
-  tstrncpy(pOption->gitinfo, gitinfo, 48);
+void dnmInitEnvCfg(SDnodeEnvCfg *pCfg) {
+  pCfg->sver = 30000000;  // 3.0.0.0
+  pCfg->numOfCores = tsNumOfCores;
+  pCfg->numOfCommitThreads = tsNumOfCommitThreads;
+  pCfg->enableTelem = 0;
+  tstrncpy(pCfg->timezone, tsTimezone, TSDB_TIMEZONE_LEN);
+  tstrncpy(pCfg->locale, tsLocale, TSDB_LOCALE_LEN);
+  tstrncpy(pCfg->charset, tsCharset, TSDB_LOCALE_LEN);
+  tstrncpy(pCfg->buildinfo, buildinfo, 64);
+  tstrncpy(pCfg->gitinfo, gitinfo, 48);
+}
+
+void dmnInitObjCfg(SDnodeObjCfg *pCfg) {
+  pCfg->numOfSupportVnodes = tsNumOfSupportVnodes;
+  pCfg->statusInterval = tsStatusInterval;
+  pCfg->numOfThreadsPerCore = tsNumOfThreadsPerCore;
+  pCfg->ratioOfQueryCores = tsRatioOfQueryCores;
+  pCfg->maxShellConns = tsMaxShellConns;
+  pCfg->shellActivityTimer = tsShellActivityTimer;
+  pCfg->serverPort = tsServerPort;
+  tstrncpy(pCfg->dataDir, tsDataDir, TSDB_FILENAME_LEN);
+  tstrncpy(pCfg->localEp, tsLocalEp, TSDB_EP_LEN);
+  tstrncpy(pCfg->localFqdn, tsLocalFqdn, TSDB_FQDN_LEN);
+  tstrncpy(pCfg->firstEp, tsFirst, TSDB_EP_LEN);
 }
 
 int dmnRunDnode() {
-  SDnodeOpt option = {0};
-  dmnInitOption(&option);
+  SDnodeEnvCfg envCfg = {0};
+  SDnodeObjCfg objCfg = {0};
+
+  dnmInitEnvCfg(&envCfg);
+  dmnInitObjCfg(&objCfg);
+
+  if (dndInit(&envCfg) != 0) {
+    uInfo("Failed to start TDengine, please check the log at %s", tsLogDir);
+    return -1;
+  }
 
-  SDnode *pDnode = dndInit(&option);
+  SDnode *pDnode = dndCreate(&objCfg);
   if (pDnode == NULL) {
     uInfo("Failed to start TDengine, please check the log at %s", tsLogDir);
     return -1;
@@ -172,7 +184,8 @@ int dmnRunDnode() {
   dmnWaitSignal();
   uInfo("TDengine is shut down!");
 
-  dndCleanup(pDnode);
+  dndClose(pDnode);
+  dndCleanup();
   taosCloseLog();
   return 0;
 }
@@ -182,21 +195,21 @@ int main(int argc, char const *argv[]) {
     return -1;
   }
 
-  if (global.generateGrant) {
+  if (dmn.generateGrant) {
     dmnGenerateGrant();
     return 0;
   }
 
-  if (global.printVersion) {
+  if (dmn.printVersion) {
     dmnPrintVersion();
     return 0;
   }
 
-  if (dmnReadConfig(global.configDir) != 0) {
+  if (dmnReadConfig(dmn.configDir) != 0) {
     return -1;
   }
 
-  if (global.dumpConfig) {
+  if (dmn.dumpConfig) {
     dmnDumpConfig();
     return 0;
   }

source/dnode/mgmt/impl/inc/dndBnode.h

@@ -19,7 +19,7 @@
 #ifdef __cplusplus
 extern "C" {
 #endif
-#include "dndInt.h"
+#include "dndEnv.h"
 
 int32_t dndInitBnode(SDnode *pDnode);
 void    dndCleanupBnode(SDnode *pDnode);

source/dnode/mgmt/impl/inc/dndEnv.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef _TD_DND_ENV_H_
+#define _TD_DND_ENV_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "dndInt.h"
+
+typedef struct {
+  EWorkerType type;
+  const char *name;
+  int32_t     minNum;
+  int32_t     maxNum;
+  void       *queueFp;
+  SDnode     *pDnode;
+  STaosQueue *queue;
+  union {
+    SWorkerPool  pool;
+    SMWorkerPool mpool;
+  };
+} SDnodeWorker;
+
+typedef struct {
+  char *dnode;
+  char *mnode;
+  char *snode;
+  char *bnode;
+  char *vnodes;
+} SDnodeDir;
+
+typedef struct {
+  int32_t      dnodeId;
+  int32_t      dropped;
+  int64_t      clusterId;
+  int64_t      dver;
+  int64_t      rebootTime;
+  int64_t      updateTime;
+  int8_t       statusSent;
+  SEpSet       mnodeEpSet;
+  char        *file;
+  SHashObj    *dnodeHash;
+  SDnodeEps   *dnodeEps;
+  pthread_t   *threadId;
+  SRWLatch     latch;
+  SDnodeWorker mgmtWorker;
+  SDnodeWorker statusWorker;
+} SDnodeMgmt;
+
+typedef struct {
+  int32_t      refCount;
+  int8_t       deployed;
+  int8_t       dropped;
+  SMnode      *pMnode;
+  SRWLatch     latch;
+  SDnodeWorker readWorker;
+  SDnodeWorker writeWorker;
+  SDnodeWorker syncWorker;
+  int8_t       replica;
+  int8_t       selfIndex;
+  SReplica     replicas[TSDB_MAX_REPLICA];
+} SMnodeMgmt;
+
+typedef struct {
+  int32_t      refCount;
+  int8_t       deployed;
+  int8_t       dropped;
+  SQnode      *pQnode;
+  SRWLatch     latch;
+  SDnodeWorker queryWorker;
+  SDnodeWorker fetchWorker;
+} SQnodeMgmt;
+
+typedef struct {
+  int32_t      refCount;
+  int8_t       deployed;
+  int8_t       dropped;
+  SSnode      *pSnode;
+  SRWLatch     latch;
+  SDnodeWorker writeWorker;
+} SSnodeMgmt;
+
+typedef struct {
+  int32_t      refCount;
+  int8_t       deployed;
+  int8_t       dropped;
+  SBnode      *pBnode;
+  SRWLatch     latch;
+  SDnodeWorker writeWorker;
+} SBnodeMgmt;
+
+typedef struct {
+  SHashObj    *hash;
+  int32_t      openVnodes;
+  int32_t      totalVnodes;
+  SRWLatch     latch;
+  SWorkerPool  queryPool;
+  SWorkerPool  fetchPool;
+  SMWorkerPool syncPool;
+  SMWorkerPool writePool;
+} SVnodesMgmt;
+
+typedef struct {
+  void    *serverRpc;
+  void    *clientRpc;
+  DndMsgFp msgFp[TDMT_MAX];
+} STransMgmt;
+
+typedef struct SDnode {
+  EStat        stat;
+  SDnodeObjCfg cfg;
+  SDnodeEnvCfg env;
+  SDnodeDir    dir;
+  FileFd       lockFd;
+  SDnodeMgmt   dmgmt;
+  SMnodeMgmt   mmgmt;
+  SQnodeMgmt   qmgmt;
+  SSnodeMgmt   smgmt;
+  SBnodeMgmt   bmgmt;
+  SVnodesMgmt  vmgmt;
+  STransMgmt   tmgmt;
+  SStartupReq  startup;
+} SDnode;
+
+typedef struct {
+  int8_t       once;
+  SDnodeEnvCfg cfg;
+} SDnodeEnv;
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_DND_ENV_H_*/
\ No newline at end of file

source/dnode/mgmt/impl/inc/dndInt.h

@@ -55,124 +55,12 @@ extern int32_t dDebugFlag;
 
 typedef enum { DND_STAT_INIT, DND_STAT_RUNNING, DND_STAT_STOPPED } EStat;
 typedef enum { DND_WORKER_SINGLE, DND_WORKER_MULTI } EWorkerType;
+typedef enum { DND_ENV_INIT = 0, DND_ENV_READY = 1, DND_ENV_CLEANUP = 2 } EEnvStat;
 typedef void (*DndMsgFp)(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEps);
 
-typedef struct {
-  EWorkerType    type;
-  const char    *name;
-  int32_t        minNum;
-  int32_t        maxNum;
-  void          *queueFp;
-  SDnode        *pDnode;
-  STaosQueue    *queue;
-  union {
-    SWorkerPool  pool;
-    SMWorkerPool mpool;
-  };
-} SDnodeWorker;
-
-typedef struct {
-  char *dnode;
-  char *mnode;
-  char *snode;
-  char *bnode;
-  char *vnodes;
-} SDnodeDir;
-
-typedef struct {
-  int32_t      dnodeId;
-  int32_t      dropped;
-  int64_t      clusterId;
-  int64_t      dver;
-  int64_t      rebootTime;
-  int64_t      updateTime;
-  int8_t       statusSent;
-  SEpSet       mnodeEpSet;
-  char        *file;
-  SHashObj    *dnodeHash;
-  SDnodeEps   *dnodeEps;
-  pthread_t   *threadId;
-  SRWLatch     latch;
-  SDnodeWorker mgmtWorker;
-} SDnodeMgmt;
-
-typedef struct {
-  int32_t      refCount;
-  int8_t       deployed;
-  int8_t       dropped;
-  SMnode      *pMnode;
-  SRWLatch     latch;
-  SDnodeWorker readWorker;
-  SDnodeWorker writeWorker;
-  SDnodeWorker syncWorker;
-  int8_t       replica;
-  int8_t       selfIndex;
-  SReplica     replicas[TSDB_MAX_REPLICA];
-} SMnodeMgmt;
-
-typedef struct {
-  int32_t      refCount;
-  int8_t       deployed;
-  int8_t       dropped;
-  SQnode      *pQnode;
-  SRWLatch     latch;
-  SDnodeWorker queryWorker;
-  SDnodeWorker fetchWorker;
-} SQnodeMgmt;
-
-typedef struct {
-  int32_t      refCount;
-  int8_t       deployed;
-  int8_t       dropped;
-  SSnode      *pSnode;
-  SRWLatch     latch;
-  SDnodeWorker writeWorker;
-} SSnodeMgmt;
-
-typedef struct {
-  int32_t      refCount;
-  int8_t       deployed;
-  int8_t       dropped;
-  SBnode      *pBnode;
-  SRWLatch     latch;
-  SDnodeWorker writeWorker;
-} SBnodeMgmt;
-
-typedef struct {
-  SHashObj    *hash;
-  int32_t      openVnodes;
-  int32_t      totalVnodes;
-  SRWLatch     latch;
-  SWorkerPool  queryPool;
-  SWorkerPool  fetchPool;
-  SMWorkerPool syncPool;
-  SMWorkerPool writePool;
-} SVnodesMgmt;
-
-typedef struct {
-  void    *serverRpc;
-  void    *clientRpc;
-  DndMsgFp msgFp[TDMT_MAX];
-} STransMgmt;
-
-typedef struct SDnode {
-  EStat       stat;
-  SDnodeOpt   opt;
-  SDnodeDir   dir;
-  FileFd      lockFd;
-  SDnodeMgmt  dmgmt;
-  SMnodeMgmt  mmgmt;
-  SQnodeMgmt  qmgmt;
-  SSnodeMgmt  smgmt;
-  SBnodeMgmt  bmgmt;
-  SVnodesMgmt vmgmt;
-  STransMgmt  tmgmt;
-  SStartupReq startup;
-} SDnode;
-
-EStat dndGetStat(SDnode *pDnode);
-void  dndSetStat(SDnode *pDnode, EStat stat);
-char *dndStatStr(EStat stat);
+EStat       dndGetStat(SDnode *pDnode);
+void        dndSetStat(SDnode *pDnode, EStat stat);
+const char *dndStatStr(EStat stat);
 
 void dndReportStartup(SDnode *pDnode, char *pName, char *pDesc);
 void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup);

source/dnode/mgmt/impl/inc/dndMgmt.h

@@ -19,7 +19,7 @@
 #ifdef __cplusplus
 extern "C" {
 #endif
-#include "dndInt.h"
+#include "dndEnv.h"
 
 int32_t dndInitMgmt(SDnode *pDnode);
 void    dndStopMgmt(SDnode *pDnode);

source/dnode/mgmt/impl/inc/dndMnode.h

@@ -19,7 +19,7 @@
 #ifdef __cplusplus
 extern "C" {
 #endif
-#include "dndInt.h"
+#include "dndEnv.h"
 
 int32_t dndInitMnode(SDnode *pDnode);
 void    dndCleanupMnode(SDnode *pDnode);

source/dnode/mgmt/impl/inc/dndQnode.h

@@ -19,7 +19,7 @@
 #ifdef __cplusplus
 extern "C" {
 #endif
-#include "dndInt.h"
+#include "dndEnv.h"
 
 int32_t dndInitQnode(SDnode *pDnode);
 void    dndCleanupQnode(SDnode *pDnode);

source/dnode/mgmt/impl/inc/dndSnode.h

@@ -19,7 +19,7 @@
 #ifdef __cplusplus
 extern "C" {
 #endif
-#include "dndInt.h"
+#include "dndEnv.h"
 
 int32_t dndInitSnode(SDnode *pDnode);
 void    dndCleanupSnode(SDnode *pDnode);

source/dnode/mgmt/impl/inc/dndTransport.h

@@ -19,7 +19,7 @@
 #ifdef __cplusplus
 extern "C" {
 #endif
-#include "dndInt.h"
+#include "dndEnv.h"
 
 int32_t dndInitTrans(SDnode *pDnode);
 void    dndCleanupTrans(SDnode *pDnode);

source/dnode/mgmt/impl/inc/dndVnodes.h

@@ -19,7 +19,7 @@
 #ifdef __cplusplus
 extern "C" {
 #endif
-#include "dndInt.h"
+#include "dndEnv.h"
 
 int32_t dndInitVnodes(SDnode *pDnode);
 void    dndCleanupVnodes(SDnode *pDnode);
@@ -36,6 +36,8 @@ int32_t dndProcessAuthVnodeReq(SDnode *pDnode, SRpcMsg *pReq);
 int32_t dndProcessSyncVnodeReq(SDnode *pDnode, SRpcMsg *pReq);
 int32_t dndProcessCompactVnodeReq(SDnode *pDnode, SRpcMsg *pReq);
 
+int32_t dndPutReqToVQueryQ(SDnode *pDnode, SRpcMsg *pReq);
+
 #ifdef __cplusplus
 }
 #endif

source/dnode/mgmt/impl/inc/dndWorker.h

@@ -19,7 +19,7 @@
 #ifdef __cplusplus
 extern "C" {
 #endif
-#include "dndInt.h"
+#include "dndEnv.h"
 
 int32_t dndInitWorker(SDnode *pDnode, SDnodeWorker *pWorker, EWorkerType type, const char *name, int32_t minNum,
                       int32_t maxNum, void *queueFp);

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

@@ -179,7 +179,7 @@ static void dndBuildBnodeOption(SDnode *pDnode, SBnodeOpt *pOption) {
   pOption->sendRedirectRspFp = dndSendRedirectRsp;
   pOption->dnodeId = dndGetDnodeId(pDnode);
   pOption->clusterId = dndGetClusterId(pDnode);
-  pOption->cfg.sver = pDnode->opt.sver;
+  pOption->sver = pDnode->env.sver;
 }
 
 static int32_t dndOpenBnode(SDnode *pDnode) {

source/dnode/mgmt/impl/src/dnode.c → source/dnode/mgmt/impl/src/dndEnv.c

@@ -22,8 +22,10 @@
 #include "dndTransport.h"
 #include "dndVnodes.h"
 #include "sync.h"
-#include "wal.h"
 #include "tfs.h"
+#include "wal.h"
+
+static SDnodeEnv dndEnv = {0};
 
 EStat dndGetStat(SDnode *pDnode) { return pDnode->stat; }
 
@@ -32,7 +34,7 @@ void dndSetStat(SDnode *pDnode, EStat stat) {
   pDnode->stat = stat;
 }
 
-char *dndStatStr(EStat stat) {
+const char *dndStatStr(EStat stat) {
   switch (stat) {
     case DND_STAT_INIT:
       return "init";
@@ -79,25 +81,26 @@ static FileFd dndCheckRunning(char *dataDir) {
   return fd;
 }
 
-static int32_t dndInitEnv(SDnode *pDnode, SDnodeOpt *pOption) {
-  pDnode->lockFd = dndCheckRunning(pOption->dataDir);
+static int32_t dndCreateImp(SDnode *pDnode, SDnodeObjCfg *pCfg) {
+  pDnode->lockFd = dndCheckRunning(pCfg->dataDir);
   if (pDnode->lockFd < 0) {
     return -1;
   }
 
   char path[PATH_MAX + 100];
-  snprintf(path, sizeof(path), "%s%smnode", pOption->dataDir, TD_DIRSEP);
+  snprintf(path, sizeof(path), "%s%smnode", pCfg->dataDir, TD_DIRSEP);
   pDnode->dir.mnode = tstrdup(path);
-  snprintf(path, sizeof(path), "%s%svnode", pOption->dataDir, TD_DIRSEP);
+  snprintf(path, sizeof(path), "%s%svnode", pCfg->dataDir, TD_DIRSEP);
   pDnode->dir.vnodes = tstrdup(path);
-  snprintf(path, sizeof(path), "%s%sdnode", pOption->dataDir, TD_DIRSEP);
+  snprintf(path, sizeof(path), "%s%sdnode", pCfg->dataDir, TD_DIRSEP);
   pDnode->dir.dnode = tstrdup(path);
-  snprintf(path, sizeof(path), "%s%ssnode", pOption->dataDir, TD_DIRSEP);
+  snprintf(path, sizeof(path), "%s%ssnode", pCfg->dataDir, TD_DIRSEP);
   pDnode->dir.snode = tstrdup(path);
-  snprintf(path, sizeof(path), "%s%sbnode", pOption->dataDir, TD_DIRSEP);
+  snprintf(path, sizeof(path), "%s%sbnode", pCfg->dataDir, TD_DIRSEP);
   pDnode->dir.bnode = tstrdup(path);
 
-  if (pDnode->dir.mnode == NULL || pDnode->dir.vnodes == NULL || pDnode->dir.dnode == NULL) {
+  if (pDnode->dir.mnode == NULL || pDnode->dir.vnodes == NULL || pDnode->dir.dnode == NULL ||
+      pDnode->dir.snode == NULL || pDnode->dir.bnode == NULL) {
     dError("failed to malloc dir object");
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
@@ -133,11 +136,12 @@ static int32_t dndInitEnv(SDnode *pDnode, SDnodeOpt *pOption) {
     return -1;
   }
 
-  memcpy(&pDnode->opt, pOption, sizeof(SDnodeOpt));
+  memcpy(&pDnode->cfg, pCfg, sizeof(SDnodeObjCfg));
+  memcpy(&pDnode->env, &dndEnv.cfg, sizeof(SDnodeEnvCfg));
   return 0;
 }
 
-static void dndCleanupEnv(SDnode *pDnode) {
+static void dndCloseImp(SDnode *pDnode) {
   tfree(pDnode->dir.mnode);
   tfree(pDnode->dir.vnodes);
   tfree(pDnode->dir.dnode);
@@ -149,118 +153,95 @@ static void dndCleanupEnv(SDnode *pDnode) {
     taosCloseFile(pDnode->lockFd);
     pDnode->lockFd = 0;
   }
-
-  taosStopCacheRefreshWorker();
 }
 
-SDnode *dndInit(SDnodeOpt *pOption) {
-  taosIgnSIGPIPE();
-  taosBlockSIGPIPE();
-  taosResolveCRC();
+SDnode *dndCreate(SDnodeObjCfg *pCfg) {
+  dInfo("start to create dnode object");
 
   SDnode *pDnode = calloc(1, sizeof(SDnode));
   if (pDnode == NULL) {
-    dError("failed to create dnode object");
     terrno = TSDB_CODE_OUT_OF_MEMORY;
+    dError("failed to create dnode object since %s", terrstr());
     return NULL;
   }
 
-  dInfo("start to initialize TDengine");
   dndSetStat(pDnode, DND_STAT_INIT);
 
-  if (dndInitEnv(pDnode, pOption) != 0) {
-    dError("failed to init env");
-    dndCleanup(pDnode);
-    return NULL;
-  }
-
-  if (rpcInit() != 0) {
-    dError("failed to init rpc env");
-    dndCleanup(pDnode);
-    return NULL;
-  }
-
-  if (walInit() != 0) {
-    dError("failed to init wal env");
-    dndCleanup(pDnode);
+  if (dndCreateImp(pDnode, pCfg) != 0) {
+    dError("failed to init dnode dir since %s", terrstr());
+    dndClose(pDnode);
     return NULL;
   }
 
   SDiskCfg dCfg;
-  strcpy(dCfg.dir, pDnode->opt.dataDir);
+  strcpy(dCfg.dir, pDnode->cfg.dataDir);
   dCfg.level = 0;
   dCfg.primary = 1;
   if (tfsInit(&dCfg, 1) != 0) {
-    dError("failed to init tfs env");
-    dndCleanup(pDnode);
-    return NULL;
-  }
-
-  if (vnodeInit(pDnode->opt.numOfCommitThreads) != 0) {
-    dError("failed to init vnode env");
-    dndCleanup(pDnode);
+    dError("failed to init tfs since %s", terrstr());
+    dndClose(pDnode);
     return NULL;
   }
 
   if (dndInitMgmt(pDnode) != 0) {
-    dError("failed to init dnode");
-    dndCleanup(pDnode);
+    dError("failed to init mgmt since %s", terrstr());
+    dndClose(pDnode);
     return NULL;
   }
 
   if (dndInitVnodes(pDnode) != 0) {
-    dError("failed to init vnodes");
-    dndCleanup(pDnode);
+    dError("failed to init vnodes since %s", terrstr());
+    dndClose(pDnode);
     return NULL;
   }
 
   if (dndInitQnode(pDnode) != 0) {
-    dError("failed to init qnode");
-    dndCleanup(pDnode);
+    dError("failed to init qnode since %s", terrstr());
+    dndClose(pDnode);
     return NULL;
   }
 
   if (dndInitSnode(pDnode) != 0) {
-    dError("failed to init snode");
-    dndCleanup(pDnode);
+    dError("failed to init snode since %s", terrstr());
+    dndClose(pDnode);
     return NULL;
   }
 
   if (dndInitBnode(pDnode) != 0) {
-    dError("failed to init bnode");
-    dndCleanup(pDnode);
+    dError("failed to init bnode since %s", terrstr());
+    dndClose(pDnode);
     return NULL;
   }
 
   if (dndInitMnode(pDnode) != 0) {
-    dError("failed to init mnode");
-    dndCleanup(pDnode);
+    dError("failed to init mnode since %s", terrstr());
+    dndClose(pDnode);
     return NULL;
   }
 
   if (dndInitTrans(pDnode) != 0) {
-    dError("failed to init transport");
-    dndCleanup(pDnode);
+    dError("failed to init transport since %s", terrstr());
+    dndClose(pDnode);
     return NULL;
   }
 
   dndSetStat(pDnode, DND_STAT_RUNNING);
   dndSendStatusReq(pDnode);
   dndReportStartup(pDnode, "TDengine", "initialized successfully");
-  dInfo("TDengine is initialized successfully, pDnode:%p", pDnode);
+  dInfo("dnode object is created, data:%p", pDnode);
 
   return pDnode;
 }
 
-void dndCleanup(SDnode *pDnode) {
+void dndClose(SDnode *pDnode) {
   if (pDnode == NULL) return;
 
   if (dndGetStat(pDnode) == DND_STAT_STOPPED) {
-    dError("dnode is shutting down");
+    dError("dnode is shutting down, data:%p", pDnode);
     return;
   }
 
-  dInfo("start to cleanup TDengine");
+  dInfo("start to close dnode, data:%p", pDnode);
   dndSetStat(pDnode, DND_STAT_STOPPED);
   dndCleanupTrans(pDnode);
   dndStopMgmt(pDnode);
@@ -270,12 +251,66 @@ void dndCleanup(SDnode *pDnode) {
   dndCleanupQnode(pDnode);
   dndCleanupVnodes(pDnode);
   dndCleanupMgmt(pDnode);
-  vnodeClear();
-  tfsDestroy();
-  walCleanUp();
-  rpcCleanup();
+  tfsCleanup();
 
-  dndCleanupEnv(pDnode);
+  dndCloseImp(pDnode);
   free(pDnode);
-  dInfo("TDengine is cleaned up successfully");
+  dInfo("dnode object is closed, data:%p", pDnode);
+}
+
+int32_t dndInit(const SDnodeEnvCfg *pCfg) {
+  if (atomic_val_compare_exchange_8(&dndEnv.once, DND_ENV_INIT, DND_ENV_READY) != DND_ENV_INIT) {
+    terrno = TSDB_CODE_REPEAT_INIT;
+    dError("failed to init dnode env since %s", terrstr());
+    return -1;
+  }
+
+  taosIgnSIGPIPE();
+  taosBlockSIGPIPE();
+  taosResolveCRC();
+
+  if (rpcInit() != 0) {
+    dError("failed to init rpc since %s", terrstr());
+    dndCleanup();
+    return -1;
+  }
+
+  if (walInit() != 0) {
+    dError("failed to init wal since %s", terrstr());
+    dndCleanup();
+    return -1;
+  }
+
+  SVnodeOpt vnodeOpt = {
+      .sver = pCfg->sver,
+      .timezone = pCfg->timezone,
+      .locale = pCfg->locale,
+      .charset = pCfg->charset,
+      .nthreads = pCfg->numOfCommitThreads,
+      .putReqToVQueryQFp = dndPutReqToVQueryQ,
+  };
+
+  if (vnodeInit(&vnodeOpt) != 0) {
+    dError("failed to init vnode since %s", terrstr());
+    dndCleanup();
+    return -1;
+  }
+
+  memcpy(&dndEnv.cfg, pCfg, sizeof(SDnodeEnvCfg));
+  dInfo("dnode env is initialized");
+  return 0;
 }
+
+void dndCleanup() {
+  if (atomic_val_compare_exchange_8(&dndEnv.once, DND_ENV_READY, DND_ENV_CLEANUP) != DND_ENV_READY) {
+    dError("dnode env is already cleaned up");
+    return;
+  }
+
+  walCleanUp();
+  vnodeCleanup();
+  rpcCleanup();
+
+  taosStopCacheRefreshWorker();
+  dInfo("dnode env is cleaned up");
+}
\ No newline at end of file

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

@@ -86,7 +86,7 @@ void dndSendRedirectRsp(SDnode *pDnode, SRpcMsg *pReq) {
   dDebug("RPC %p, req:%s is redirected, num:%d use:%d", pReq->handle, TMSG_INFO(msgType), epSet.numOfEps, epSet.inUse);
   for (int32_t i = 0; i < epSet.numOfEps; ++i) {
     dDebug("mnode index:%d %s:%u", i, epSet.fqdn[i], epSet.port[i]);
-    if (strcmp(epSet.fqdn[i], pDnode->opt.localFqdn) == 0 && epSet.port[i] == pDnode->opt.serverPort) {
+    if (strcmp(epSet.fqdn[i], pDnode->cfg.localFqdn) == 0 && epSet.port[i] == pDnode->cfg.serverPort) {
       epSet.inUse = (i + 1) % epSet.numOfEps;
     }
 
@@ -289,8 +289,8 @@ PRASE_DNODE_OVER:
   if (root != NULL) cJSON_Delete(root);
   if (fp != NULL) fclose(fp);
 
-  if (dndIsEpChanged(pDnode, pMgmt->dnodeId, pDnode->opt.localEp)) {
-    dError("localEp %s different with %s and need reconfigured", pDnode->opt.localEp, pMgmt->file);
+  if (dndIsEpChanged(pDnode, pMgmt->dnodeId, pDnode->cfg.localEp)) {
+    dError("localEp %s different with %s and need reconfigured", pDnode->cfg.localEp, pMgmt->file);
     return -1;
   }
 
@@ -298,7 +298,7 @@ PRASE_DNODE_OVER:
     pMgmt->dnodeEps = calloc(1, sizeof(SDnodeEps) + sizeof(SDnodeEp));
     pMgmt->dnodeEps->num = 1;
     pMgmt->dnodeEps->eps[0].isMnode = 1;
-    taosGetFqdnPortFromEp(pDnode->opt.firstEp, pMgmt->dnodeEps->eps[0].fqdn, &pMgmt->dnodeEps->eps[0].port);
+    taosGetFqdnPortFromEp(pDnode->cfg.firstEp, pMgmt->dnodeEps->eps[0].fqdn, &pMgmt->dnodeEps->eps[0].port);
   }
 
   dndResetDnodes(pDnode, pMgmt->dnodeEps);
@@ -362,24 +362,24 @@ void dndSendStatusReq(SDnode *pDnode) {
 
   SDnodeMgmt *pMgmt = &pDnode->dmgmt;
   taosRLockLatch(&pMgmt->latch);
-  pStatus->sver = htonl(pDnode->opt.sver);
+  pStatus->sver = htonl(pDnode->env.sver);
   pStatus->dver = htobe64(pMgmt->dver);
   pStatus->dnodeId = htonl(pMgmt->dnodeId);
   pStatus->clusterId = htobe64(pMgmt->clusterId);
   pStatus->rebootTime = htobe64(pMgmt->rebootTime);
   pStatus->updateTime = htobe64(pMgmt->updateTime);
-  pStatus->numOfCores = htonl(pDnode->opt.numOfCores);
-  pStatus->numOfSupportVnodes = htonl(pDnode->opt.numOfSupportVnodes);
-  tstrncpy(pStatus->dnodeEp, pDnode->opt.localEp, TSDB_EP_LEN);
+  pStatus->numOfCores = htonl(pDnode->env.numOfCores);
+  pStatus->numOfSupportVnodes = htonl(pDnode->cfg.numOfSupportVnodes);
+  tstrncpy(pStatus->dnodeEp, pDnode->cfg.localEp, TSDB_EP_LEN);
 
-  pStatus->clusterCfg.statusInterval = htonl(pDnode->opt.statusInterval);
+  pStatus->clusterCfg.statusInterval = htonl(pDnode->cfg.statusInterval);
   pStatus->clusterCfg.checkTime = 0;
   char timestr[32] = "1970-01-01 00:00:00.00";
   (void)taosParseTime(timestr, &pStatus->clusterCfg.checkTime, (int32_t)strlen(timestr), TSDB_TIME_PRECISION_MILLI, 0);
   pStatus->clusterCfg.checkTime = htonl(pStatus->clusterCfg.checkTime);
-  tstrncpy(pStatus->clusterCfg.timezone, pDnode->opt.timezone, TSDB_TIMEZONE_LEN);
-  tstrncpy(pStatus->clusterCfg.locale, pDnode->opt.locale, TSDB_LOCALE_LEN);
-  tstrncpy(pStatus->clusterCfg.charset, pDnode->opt.charset, TSDB_LOCALE_LEN);
+  tstrncpy(pStatus->clusterCfg.timezone, pDnode->env.timezone, TSDB_TIMEZONE_LEN);
+  tstrncpy(pStatus->clusterCfg.locale, pDnode->env.locale, TSDB_LOCALE_LEN);
+  tstrncpy(pStatus->clusterCfg.charset, pDnode->env.charset, TSDB_LOCALE_LEN);
   taosRUnLockLatch(&pMgmt->latch);
 
   dndGetVnodeLoads(pDnode, &pStatus->vnodeLoads);
@@ -485,7 +485,7 @@ void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) {
 static void *dnodeThreadRoutine(void *param) {
   SDnode     *pDnode = param;
   SDnodeMgmt *pMgmt = &pDnode->dmgmt;
-  int32_t     ms = pDnode->opt.statusInterval * 1000;
+  int32_t     ms = pDnode->cfg.statusInterval * 1000;
 
   while (true) {
     pthread_testcancel();
@@ -536,6 +536,11 @@ int32_t dndInitMgmt(SDnode *pDnode) {
     return -1;
   }
 
+  if (dndInitWorker(pDnode, &pMgmt->statusWorker, DND_WORKER_SINGLE, "dnode-status", 1, 1, dndProcessMgmtQueue) != 0) {
+    dError("failed to start dnode mgmt worker since %s", terrstr());
+    return -1;
+  }
+
   pMgmt->threadId = taosCreateThread(dnodeThreadRoutine, pDnode);
   if (pMgmt->threadId == NULL) {
     dError("failed to init dnode thread");
@@ -550,6 +555,7 @@ int32_t dndInitMgmt(SDnode *pDnode) {
 void dndStopMgmt(SDnode *pDnode) {
   SDnodeMgmt *pMgmt = &pDnode->dmgmt;
   dndCleanupWorker(&pMgmt->mgmtWorker);
+  dndCleanupWorker(&pMgmt->statusWorker);
 
   if (pMgmt->threadId != NULL) {
     taosDestoryThread(pMgmt->threadId);
@@ -587,7 +593,12 @@ void dndProcessMgmtMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
     dndUpdateMnodeEpSet(pDnode, pEpSet);
   }
 
-  if (dndWriteMsgToWorker(&pMgmt->mgmtWorker, pMsg, sizeof(SRpcMsg)) != 0) {
+  SDnodeWorker *pWorker = &pMgmt->mgmtWorker;
+  if (pMsg->msgType == TDMT_MND_STATUS_RSP) {
+    pWorker = &pMgmt->statusWorker;
+  }
+
+  if (dndWriteMsgToWorker(pWorker, pMsg, sizeof(SRpcMsg)) != 0) {
     if (pMsg->msgType & 1u) {
       SRpcMsg rsp = {.handle = pMsg->handle, .code = TSDB_CODE_OUT_OF_MEMORY};
       rpcSendResponse(&rsp);

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

@@ -247,7 +247,7 @@ static bool dndNeedDeployMnode(SDnode *pDnode) {
     return false;
   }
 
-  if (strcmp(pDnode->opt.localEp, pDnode->opt.firstEp) != 0) {
+  if (strcmp(pDnode->cfg.localEp, pDnode->cfg.firstEp) != 0) {
     return false;
   }
 
@@ -266,15 +266,15 @@ static void dndInitMnodeOption(SDnode *pDnode, SMnodeOpt *pOption) {
   pOption->putReqToMWriteQFp = dndPutMsgToMWriteQ;
   pOption->dnodeId = dndGetDnodeId(pDnode);
   pOption->clusterId = dndGetClusterId(pDnode);
-  pOption->cfg.sver = pDnode->opt.sver;
-  pOption->cfg.enableTelem = pDnode->opt.enableTelem;
-  pOption->cfg.statusInterval = pDnode->opt.statusInterval;
-  pOption->cfg.shellActivityTimer = pDnode->opt.shellActivityTimer;
-  pOption->cfg.timezone = pDnode->opt.timezone;
-  pOption->cfg.charset = pDnode->opt.charset;
-  pOption->cfg.locale = pDnode->opt.locale;
-  pOption->cfg.gitinfo = pDnode->opt.gitinfo;
-  pOption->cfg.buildinfo = pDnode->opt.buildinfo;
+  pOption->cfg.sver = pDnode->env.sver;
+  pOption->cfg.enableTelem = pDnode->env.enableTelem;
+  pOption->cfg.statusInterval = pDnode->cfg.statusInterval;
+  pOption->cfg.shellActivityTimer = pDnode->cfg.shellActivityTimer;
+  pOption->cfg.timezone = pDnode->env.timezone;
+  pOption->cfg.charset = pDnode->env.charset;
+  pOption->cfg.locale = pDnode->env.locale;
+  pOption->cfg.gitinfo = pDnode->env.gitinfo;
+  pOption->cfg.buildinfo = pDnode->env.buildinfo;
 }
 
 static void dndBuildMnodeDeployOption(SDnode *pDnode, SMnodeOpt *pOption) {
@@ -283,8 +283,8 @@ static void dndBuildMnodeDeployOption(SDnode *pDnode, SMnodeOpt *pOption) {
   pOption->selfIndex = 0;
   SReplica *pReplica = &pOption->replicas[0];
   pReplica->id = 1;
-  pReplica->port = pDnode->opt.serverPort;
-  memcpy(pReplica->fqdn, pDnode->opt.localFqdn, TSDB_FQDN_LEN);
+  pReplica->port = pDnode->cfg.serverPort;
+  memcpy(pReplica->fqdn, pDnode->cfg.localFqdn, TSDB_FQDN_LEN);
 
   SMnodeMgmt *pMgmt = &pDnode->mmgmt;
   pMgmt->selfIndex = pOption->selfIndex;

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

@@ -185,7 +185,7 @@ static void dndBuildQnodeOption(SDnode *pDnode, SQnodeOpt *pOption) {
   pOption->sendRedirectRspFp = dndSendRedirectRsp;
   pOption->dnodeId = dndGetDnodeId(pDnode);
   pOption->clusterId = dndGetClusterId(pDnode);
-  pOption->cfg.sver = pDnode->opt.sver;
+  pOption->sver = pDnode->env.sver;
 }
 
 static int32_t dndOpenQnode(SDnode *pDnode) {

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

@@ -179,7 +179,7 @@ static void dndBuildSnodeOption(SDnode *pDnode, SSnodeOpt *pOption) {
   pOption->sendRedirectRspFp = dndSendRedirectRsp;
   pOption->dnodeId = dndGetDnodeId(pDnode);
   pOption->clusterId = dndGetClusterId(pDnode);
-  pOption->cfg.sver = pDnode->opt.sver;
+  pOption->sver = pDnode->env.sver;
 }
 
 static int32_t dndOpenSnode(SDnode *pDnode) {

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

@@ -176,7 +176,7 @@ static int32_t dndInitClient(SDnode *pDnode) {
   rpcInit.cfp = dndProcessResponse;
   rpcInit.sessions = 1024;
   rpcInit.connType = TAOS_CONN_CLIENT;
-  rpcInit.idleTime = pDnode->opt.shellActivityTimer * 1000;
+  rpcInit.idleTime = pDnode->cfg.shellActivityTimer * 1000;
   rpcInit.user = INTERNAL_USER;
   rpcInit.ckey = INTERNAL_CKEY;
   rpcInit.secret = INTERNAL_SECRET;
@@ -325,20 +325,20 @@ static int32_t dndInitServer(SDnode *pDnode) {
   STransMgmt *pMgmt = &pDnode->tmgmt;
   dndInitMsgFp(pMgmt);
 
-  int32_t numOfThreads = (int32_t)((pDnode->opt.numOfCores * pDnode->opt.numOfThreadsPerCore) / 2.0);
+  int32_t numOfThreads = (int32_t)((pDnode->env.numOfCores * pDnode->cfg.numOfThreadsPerCore) / 2.0);
   if (numOfThreads < 1) {
     numOfThreads = 1;
   }
 
   SRpcInit rpcInit;
   memset(&rpcInit, 0, sizeof(rpcInit));
-  rpcInit.localPort = pDnode->opt.serverPort;
+  rpcInit.localPort = pDnode->cfg.serverPort;
   rpcInit.label = "DND-S";
   rpcInit.numOfThreads = numOfThreads;
   rpcInit.cfp = dndProcessRequest;
-  rpcInit.sessions = pDnode->opt.maxShellConns;
+  rpcInit.sessions = pDnode->cfg.maxShellConns;
   rpcInit.connType = TAOS_CONN_SERVER;
-  rpcInit.idleTime = pDnode->opt.shellActivityTimer * 1000;
+  rpcInit.idleTime = pDnode->cfg.shellActivityTimer * 1000;
   rpcInit.afp = dndRetrieveUserAuthInfo;
   rpcInit.parent = pDnode;
 

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

@@ -381,7 +381,8 @@ static void *dnodeOpenVnodeFunc(void *param) {
              pMgmt->openVnodes, pMgmt->totalVnodes);
     dndReportStartup(pDnode, "open-vnodes", stepDesc);
 
-    SVnode *pImpl = vnodeOpen(pCfg->path, NULL, pCfg->vgId);
+    SVnodeCfg cfg = {.pDnode = pDnode, .vgId = pCfg->vgId};
+    SVnode   *pImpl = vnodeOpen(pCfg->path, &cfg);
     if (pImpl == NULL) {
       dError("vgId:%d, failed to open vnode by thread:%d", pCfg->vgId, pThread->threadIndex);
       pThread->failed++;
@@ -419,7 +420,7 @@ static int32_t dndOpenVnodes(SDnode *pDnode) {
 
   pMgmt->totalVnodes = numOfVnodes;
 
-  int32_t threadNum = pDnode->opt.numOfCores;
+  int32_t threadNum = pDnode->env.numOfCores;
   int32_t vnodesPerThread = numOfVnodes / threadNum + 1;
 
   SVnodeThread *threads = calloc(threadNum, sizeof(SVnodeThread));
@@ -581,7 +582,8 @@ int32_t dndProcessCreateVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
     return -1;
   }
 
-  SVnode *pImpl = vnodeOpen(wrapperCfg.path, NULL /*pCfg*/, pCreate->vgId);
+  vnodeCfg.pDnode = pDnode;
+  SVnode *pImpl = vnodeOpen(wrapperCfg.path, &vnodeCfg);
   if (pImpl == NULL) {
     dError("vgId:%d, failed to create vnode since %s", pCreate->vgId, terrstr());
     return -1;
@@ -800,7 +802,7 @@ static void dndProcessVnodeSyncQueue(SVnodeObj *pVnode, STaosQall *qall, int32_t
   }
 }
 
-static int32_t dndWriteRpcMsgToVnodeQueue(STaosQueue *pQueue, SRpcMsg *pRpcMsg) {
+static int32_t dndWriteRpcMsgToVnodeQueue(STaosQueue *pQueue, SRpcMsg *pRpcMsg, bool sendRsp) {
   int32_t code = 0;
 
   if (pQueue == NULL) {
@@ -817,13 +819,15 @@ static int32_t dndWriteRpcMsgToVnodeQueue(STaosQueue *pQueue, SRpcMsg *pRpcMsg)
     }
   }
 
-  if (code != TSDB_CODE_SUCCESS) {
+  if (code != TSDB_CODE_SUCCESS && sendRsp) {
     if (pRpcMsg->msgType & 1u) {
       SRpcMsg rsp = {.handle = pRpcMsg->handle, .code = code};
       rpcSendResponse(&rsp);
     }
     rpcFreeCont(pRpcMsg->pCont);
   }
+
+  return code;
 }
 
 static SVnodeObj *dndAcquireVnodeFromMsg(SDnode *pDnode, SRpcMsg *pMsg) {
@@ -846,7 +850,7 @@ static SVnodeObj *dndAcquireVnodeFromMsg(SDnode *pDnode, SRpcMsg *pMsg) {
 void dndProcessVnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
   SVnodeObj *pVnode = dndAcquireVnodeFromMsg(pDnode, pMsg);
   if (pVnode != NULL) {
-    dndWriteRpcMsgToVnodeQueue(pVnode->pWriteQ, pMsg);
+    (void)dndWriteRpcMsgToVnodeQueue(pVnode->pWriteQ, pMsg, true);
     dndReleaseVnode(pDnode, pVnode);
   }
 }
@@ -854,7 +858,7 @@ void dndProcessVnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
 void dndProcessVnodeSyncMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
   SVnodeObj *pVnode = dndAcquireVnodeFromMsg(pDnode, pMsg);
   if (pVnode != NULL) {
-    dndWriteRpcMsgToVnodeQueue(pVnode->pSyncQ, pMsg);
+    (void)dndWriteRpcMsgToVnodeQueue(pVnode->pSyncQ, pMsg, true);
     dndReleaseVnode(pDnode, pVnode);
   }
 }
@@ -862,7 +866,7 @@ void dndProcessVnodeSyncMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
 void dndProcessVnodeQueryMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
   SVnodeObj *pVnode = dndAcquireVnodeFromMsg(pDnode, pMsg);
   if (pVnode != NULL) {
-    dndWriteRpcMsgToVnodeQueue(pVnode->pQueryQ, pMsg);
+    (void)dndWriteRpcMsgToVnodeQueue(pVnode->pQueryQ, pMsg, true);
     dndReleaseVnode(pDnode, pVnode);
   }
 }
@@ -870,11 +874,23 @@ void dndProcessVnodeQueryMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
 void dndProcessVnodeFetchMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
   SVnodeObj *pVnode = dndAcquireVnodeFromMsg(pDnode, pMsg);
   if (pVnode != NULL) {
-    dndWriteRpcMsgToVnodeQueue(pVnode->pFetchQ, pMsg);
+    (void)dndWriteRpcMsgToVnodeQueue(pVnode->pFetchQ, pMsg, true);
     dndReleaseVnode(pDnode, pVnode);
   }
 }
 
+int32_t dndPutReqToVQueryQ(SDnode *pDnode, SRpcMsg *pMsg) {
+  SMsgHead *pHead = pMsg->pCont;
+  // pHead->vgId = htonl(pHead->vgId);
+
+  SVnodeObj *pVnode = dndAcquireVnode(pDnode, pHead->vgId);
+  if (pVnode == NULL) return -1;
+
+  int32_t code = dndWriteRpcMsgToVnodeQueue(pVnode->pFetchQ, pMsg, false);
+  dndReleaseVnode(pDnode, pVnode);
+  return code;
+}
+
 static int32_t dndPutMsgIntoVnodeApplyQueue(SDnode *pDnode, int32_t vgId, SRpcMsg *pMsg) {
   SVnodeObj *pVnode = dndAcquireVnode(pDnode, vgId);
   if (pVnode == NULL) return -1;
@@ -888,11 +904,11 @@ static int32_t dndInitVnodeWorkers(SDnode *pDnode) {
   SVnodesMgmt *pMgmt = &pDnode->vmgmt;
 
   int32_t maxFetchThreads = 4;
-  int32_t minFetchThreads = MIN(maxFetchThreads, pDnode->opt.numOfCores);
-  int32_t minQueryThreads = MAX((int32_t)(pDnode->opt.numOfCores * pDnode->opt.ratioOfQueryCores), 1);
+  int32_t minFetchThreads = MIN(maxFetchThreads, pDnode->env.numOfCores);
+  int32_t minQueryThreads = MAX((int32_t)(pDnode->env.numOfCores * pDnode->cfg.ratioOfQueryCores), 1);
   int32_t maxQueryThreads = minQueryThreads;
-  int32_t maxWriteThreads = MAX(pDnode->opt.numOfCores, 1);
-  int32_t maxSyncThreads = MAX(pDnode->opt.numOfCores / 2, 1);
+  int32_t maxWriteThreads = MAX(pDnode->env.numOfCores, 1);
+  int32_t maxSyncThreads = MAX(pDnode->env.numOfCores / 2, 1);
 
   SWorkerPool *pPool = &pMgmt->queryPool;
   pPool->name = "vnode-query";

source/dnode/mgmt/impl/test/sut/inc/server.h

@@ -24,7 +24,7 @@ class TestServer {
   bool DoStart();
 
  private:
-  SDnodeOpt BuildOption(const char* path, const char* fqdn, uint16_t port, const char* firstEp);
+  SDnodeObjCfg BuildOption(const char* path, const char* fqdn, uint16_t port, const char* firstEp);
 
  private:
   SDnode*    pDnode;

source/dnode/mgmt/impl/test/sut/src/server.cpp

@@ -22,30 +22,27 @@ void* serverLoop(void* param) {
   }
 }
 
-SDnodeOpt TestServer::BuildOption(const char* path, const char* fqdn, uint16_t port, const char* firstEp) {
-  SDnodeOpt option = {0};
-  option.sver = 1;
-  option.numOfCores = 1;
-  option.numOfSupportVnodes = 16;
-  option.numOfCommitThreads = 1;
-  option.statusInterval = 1;
-  option.numOfThreadsPerCore = 1;
-  option.ratioOfQueryCores = 1;
-  option.maxShellConns = 1000;
-  option.shellActivityTimer = 30;
-  option.serverPort = port;
-  strcpy(option.dataDir, path);
-  snprintf(option.localEp, TSDB_EP_LEN, "%s:%u", fqdn, port);
-  snprintf(option.localFqdn, TSDB_FQDN_LEN, "%s", fqdn);
-  snprintf(option.firstEp, TSDB_EP_LEN, "%s", firstEp);
-  return option;
+SDnodeObjCfg TestServer::BuildOption(const char* path, const char* fqdn, uint16_t port, const char* firstEp) {
+  SDnodeObjCfg cfg = {0};
+  cfg.numOfSupportVnodes = 16;
+  cfg.statusInterval = 1;
+  cfg.numOfThreadsPerCore = 1;
+  cfg.ratioOfQueryCores = 1;
+  cfg.maxShellConns = 1000;
+  cfg.shellActivityTimer = 30;
+  cfg.serverPort = port;
+  strcpy(cfg.dataDir, path);
+  snprintf(cfg.localEp, TSDB_EP_LEN, "%s:%u", fqdn, port);
+  snprintf(cfg.localFqdn, TSDB_FQDN_LEN, "%s", fqdn);
+  snprintf(cfg.firstEp, TSDB_EP_LEN, "%s", firstEp);
+  return cfg;
 }
 
 bool TestServer::DoStart() {
-  SDnodeOpt option = BuildOption(path, fqdn, port, firstEp);
+  SDnodeObjCfg cfg = BuildOption(path, fqdn, port, firstEp);
   taosMkDir(path);
 
-  pDnode = dndInit(&option);
+  pDnode = dndCreate(&cfg);
   if (pDnode != NULL) {
     return false;
   }
@@ -81,7 +78,7 @@ void TestServer::Stop() {
   }
 
   if (pDnode != NULL) {
-    dndCleanup(pDnode);
+    dndClose(pDnode);
     pDnode = NULL;
   }
 }

source/dnode/mgmt/impl/test/sut/src/sut.cpp

@@ -43,6 +43,11 @@ void Testbase::InitLog(const char* path) {
 }
 
 void Testbase::Init(const char* path, int16_t port) {
+  SDnodeEnvCfg cfg = {0};
+  cfg.numOfCommitThreads = 1;
+  cfg.numOfCores = 1;
+  dndInit(&cfg);
+
   char fqdn[] = "localhost";
   char firstEp[TSDB_EP_LEN] = {0};
   snprintf(firstEp, TSDB_EP_LEN, "%s:%u", fqdn, port);
@@ -56,6 +61,7 @@ void Testbase::Init(const char* path, int16_t port) {
 void Testbase::Cleanup() {
   server.Stop();
   client.Cleanup();
+  dndCleanup();
 }
 
 void Testbase::Restart() { server.Restart(); }

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

@@ -740,7 +740,7 @@ static int32_t mndBuildDropVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *
     if (pReq == NULL) return -1;
 
     action.pCont = pReq;
-    action.contLen = sizeof(SCreateVnodeReq);
+    action.contLen = sizeof(SDropVnodeReq);
     action.msgType = TDMT_DND_DROP_VNODE;
     action.acceptableCode = TSDB_CODE_DND_VNODE_NOT_DEPLOYED;
     if (mndTransAppendRedoAction(pTrans, &action) != 0) {

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

@@ -319,6 +319,14 @@ static int32_t mndGetAvailableDnode(SMnode *pMnode, SVgObj *pVgroup, SArray *pAr
 
   taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes);
 
+  int32_t size = taosArrayGetSize(pArray);
+  if (size < pVgroup->replica) {
+    mError("db:%s, vgId:%d, no enough online dnodes:%d to alloc %d replica", pVgroup->dbName, pVgroup->vgId, size,
+           pVgroup->replica);
+    terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
+    return -1;
+  }
+
   for (int32_t v = 0; v < pVgroup->replica; ++v) {
     SVnodeGid *pVgid = &pVgroup->vnodeGid[v];
     SDnodeObj *pDnode = taosArrayGet(pArray, v);

source/dnode/mnode/impl/test/profile/profile.cpp

@@ -13,7 +13,7 @@
 
 class MndTestProfile : public ::testing::Test {
  protected:
-  static void SetUpTestSuite() { test.Init("/tmp/mnode_test_profile", 9022); }
+  static void SetUpTestSuite() { test.Init("/tmp/mnode_test_profile", 9031); }
   static void TearDownTestSuite() { test.Cleanup(); }
 
   static Testbase test;
@@ -53,7 +53,7 @@ TEST_F(MndTestProfile, 01_ConnectMsg) {
 
   EXPECT_EQ(pRsp->epSet.inUse, 0);
   EXPECT_EQ(pRsp->epSet.numOfEps, 1);
-  EXPECT_EQ(pRsp->epSet.port[0], 9022);
+  EXPECT_EQ(pRsp->epSet.port[0], 9031);
   EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
 
   connId = pRsp->connId;
@@ -127,7 +127,7 @@ TEST_F(MndTestProfile, 04_HeartBeatMsg) {
 
   EXPECT_EQ(pRsp->epSet.inUse, 0);
   EXPECT_EQ(pRsp->epSet.numOfEps, 1);
-  EXPECT_EQ(pRsp->epSet.port[0], 9022);
+  EXPECT_EQ(pRsp->epSet.port[0], 9031);
   EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
 }
 
@@ -185,7 +185,7 @@ TEST_F(MndTestProfile, 05_KillConnMsg) {
 
     EXPECT_EQ(pRsp->epSet.inUse, 0);
     EXPECT_EQ(pRsp->epSet.numOfEps, 1);
-    EXPECT_EQ(pRsp->epSet.port[0], 9022);
+    EXPECT_EQ(pRsp->epSet.port[0], 9031);
     EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
 
     connId = pRsp->connId;
@@ -249,7 +249,7 @@ TEST_F(MndTestProfile, 07_KillQueryMsg) {
 
     EXPECT_EQ(pRsp->epSet.inUse, 0);
     EXPECT_EQ(pRsp->epSet.numOfEps, 1);
-    EXPECT_EQ(pRsp->epSet.port[0], 9022);
+    EXPECT_EQ(pRsp->epSet.port[0], 9031);
     EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
   }
 }

source/dnode/qnode/inc/qndInt.h

@@ -29,12 +29,7 @@ extern "C" {
 #endif
 
 typedef struct SQnode {
-  int32_t           dnodeId;
-  int64_t           clusterId;
-  SQnodeCfg         cfg;
-  SendReqToDnodeFp  sendReqToDnodeFp;
-  SendReqToMnodeFp  sendReqToMnodeFp;
-  SendRedirectRspFp sendRedirectRspFp;
+  SQnodeOpt opt;
 } SQnode;
 
 #ifdef __cplusplus

source/dnode/snode/inc/sndInt.h

@@ -29,12 +29,7 @@ extern "C" {
 #endif
 
 typedef struct SSnode {
-  int32_t           dnodeId;
-  int64_t           clusterId;
-  SSnodeCfg         cfg;
-  SendReqToDnodeFp  sendReqToDnodeFp;
-  SendReqToMnodeFp  sendReqToMnodeFp;
-  SendRedirectRspFp sendRedirectRspFp;
+  SSnodeOpt cfg;
 } SSnode;
 
 #ifdef __cplusplus

source/dnode/vnode/CMakeLists.txt

-add_subdirectory(meta)
-add_subdirectory(tq)
-add_subdirectory(tsdb)
-add_subdirectory(impl)
+aux_source_directory(src/meta META_SRC)
+aux_source_directory(src/tq TQ_SRC)
+aux_source_directory(src/tsdb TSDB_SRC)
+aux_source_directory(src/vnd VND_SRC)
\ No newline at end of file
+list(APPEND
+    VNODE_SRC
+    ${META_SRC}
+    ${TQ_SRC}
+    ${TSDB_SRC}
+    ${VND_SRC}
+)
+
+add_library(vnode STATIC ${VNODE_SRC})
+target_include_directories(
+    vnode
+    PUBLIC inc
+    PRIVATE src/inc
+)
+target_link_libraries(
+    vnode
+    PUBLIC os
+    PUBLIC util
+    PUBLIC common
+    PUBLIC transport
+    PUBLIC bdb
+    PUBLIC tfs
+    PUBLIC wal
+    PUBLIC qworker
+)
+
+if(${BUILD_TEST})
+    # add_subdirectory(test)
+endif(${BUILD_TEST})

source/dnode/vnode/impl/CMakeLists.txt

-aux_source_directory(src VNODE_SRC)
-add_library(vnode STATIC ${VNODE_SRC})
-target_include_directories(
-    vnode
-    PUBLIC "${CMAKE_SOURCE_DIR}/include/dnode/vnode"
-    private "${CMAKE_CURRENT_SOURCE_DIR}/inc"
-)
-target_link_libraries(
-    vnode
-    PUBLIC os
-    PUBLIC transport
-    PUBLIC meta
-    PUBLIC tq
-    PUBLIC tsdb
-    PUBLIC wal
-    PUBLIC sync
-    PUBLIC cjson
-    PUBLIC qworker
-)
-
-# test
-if(${BUILD_TEST})
-#    add_subdirectory(test)
-endif(${BUILD_TEST})

source/dnode/vnode/impl/inc/vnodeMAF.h

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#ifndef _TD_VNODE_MAF_H_
-#define _TD_VNODE_MAF_H_
-
-#include "vnode.h"
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-int  vnodeOpenMAF(SVnode *pVnode);
-void vnodeCloseMAF(SVnode *pVnode);
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif /*_TD_VNODE_MAF_H_*/
\ No newline at end of file

source/dnode/vnode/impl/inc/vnodeRead.h

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#ifndef _TD_VNODE_READ_H_
-#define _TD_VNODE_READ_H_
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-#include "vnodeInt.h"
-
-void vnodeProcessReadMsg(SVnode *pVnode, SVnodeMsg *pMsg);
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif /*_TD_VNODE_READ_H_*/

source/dnode/vnode/impl/inc/vnodeRequest.h

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#ifndef _TD_VNODE_REQUEST_H_
-#define _TD_VNODE_REQUEST_H_
-
-#include "vnode.h"
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-// SVDropTbReq
-// int   vnodeBuildDropTableReq(void **buf, const SVDropTbReq *pReq);
-// void *vnodeParseDropTableReq(void *buf, SVDropTbReq *pReq);
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif /*_TD_VNODE_REQUEST_H_*/
\ No newline at end of file

source/dnode/vnode/impl/src/vnodeFS.c

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#include "vnodeDef.h"
\ No newline at end of file

source/dnode/vnode/impl/src/vnodeRead.c

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#include "vnodeDef.h"
\ No newline at end of file

source/dnode/vnode/impl/src/vnodeRequest.c

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#include "vnodeDef.h"
-
-#if 0
-
-static int   vnodeBuildCreateTableReq(void **buf, const SVCreateTableReq *pReq);
-static void *vnodeParseCreateTableReq(void *buf, SVCreateTableReq *pReq);
-
-int vnodeBuildReq(void **buf, const SVnodeReq *pReq, tmsg_t type) {
-  int tsize = 0;
-
-  tsize += taosEncodeFixedU64(buf, pReq->ver);
-  switch (type) {
-    case TDMT_VND_CREATE_STB:
-      tsize += vnodeBuildCreateTableReq(buf, &(pReq->ctReq));
-      break;
-    case TDMT_VND_SUBMIT:
-      /* code */
-      break;
-    default:
-      break;
-  }
-  /* TODO */
-  return tsize;
-}
-
-void *vnodeParseReq(void *buf, SVnodeReq *pReq, tmsg_t type) {
-  buf = taosDecodeFixedU64(buf, &(pReq->ver));
-
-  switch (type) {
-    case TDMT_VND_CREATE_STB:
-      buf = vnodeParseCreateTableReq(buf, &(pReq->ctReq));
-      break;
-
-    default:
-      break;
-  }
-
-  // TODO
-  return buf;
-}
-
-static int vnodeBuildCreateTableReq(void **buf, const SVCreateTableReq *pReq) {
-  int tsize = 0;
-
-  tsize += taosEncodeString(buf, pReq->name);
-  tsize += taosEncodeFixedU32(buf, pReq->ttl);
-  tsize += taosEncodeFixedU32(buf, pReq->keep);
-  tsize += taosEncodeFixedU8(buf, pReq->type);
-
-  switch (pReq->type) {
-    case META_SUPER_TABLE:
-      tsize += taosEncodeFixedU64(buf, pReq->stbCfg.suid);
-      tsize += tdEncodeSchema(buf, pReq->stbCfg.pSchema);
-      tsize += tdEncodeSchema(buf, pReq->stbCfg.pTagSchema);
-      break;
-    case META_CHILD_TABLE:
-      tsize += taosEncodeFixedU64(buf, pReq->ctbCfg.suid);
-      tsize += tdEncodeKVRow(buf, pReq->ctbCfg.pTag);
-      break;
-    case META_NORMAL_TABLE:
-      tsize += tdEncodeSchema(buf, pReq->ntbCfg.pSchema);
-      break;
-    default:
-      break;
-  }
-
-  return tsize;
-}
-
-static void *vnodeParseCreateTableReq(void *buf, SVCreateTableReq *pReq) {
-  buf = taosDecodeString(buf, &(pReq->name));
-  buf = taosDecodeFixedU32(buf, &(pReq->ttl));
-  buf = taosDecodeFixedU32(buf, &(pReq->keep));
-  buf = taosDecodeFixedU8(buf, &(pReq->type));
-
-  switch (pReq->type) {
-    case META_SUPER_TABLE:
-      buf = taosDecodeFixedU64(buf, &(pReq->stbCfg.suid));
-      buf = tdDecodeSchema(buf, &(pReq->stbCfg.pSchema));
-      buf = tdDecodeSchema(buf, &(pReq->stbCfg.pTagSchema));
-      break;
-    case META_CHILD_TABLE:
-      buf = taosDecodeFixedU64(buf, &(pReq->ctbCfg.suid));
-      buf = tdDecodeKVRow(buf, &(pReq->ctbCfg.pTag));
-      break;
-    case META_NORMAL_TABLE:
-      buf = tdDecodeSchema(buf, &(pReq->ntbCfg.pSchema));
-      break;
-    default:
-      break;
-  }
-
-  return buf;
-}
-
-int vnodeBuildDropTableReq(void **buf, const SVDropTbReq *pReq) {
-  // TODO
-  return 0;
-}
-
-void *vnodeParseDropTableReq(void *buf, SVDropTbReq *pReq) {
-  // TODO
-}
-#endif
\ No newline at end of file

source/dnode/vnode/impl/src/vnodeSync.c

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
\ No newline at end of file

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

-# Vnode API test
-add_executable(vnodeApiTests "")
-target_sources(vnodeApiTests
-    PRIVATE
-    "vnodeApiTests.cpp"
-)
-target_link_libraries(vnodeApiTests vnode gtest gtest_main)
-
-add_test(
-    NAME vnode_api_tests 
-    COMMAND ${CMAKE_CURRENT_BINARY_DIR}/vnodeApiTests
-    )
\ No newline at end of file

source/dnode/vnode/impl/test/vBenchmarkTest.cpp

-// https://stackoverflow.com/questions/8565666/benchmarking-with-googletest
-// https://github.com/google/benchmark
\ No newline at end of file

source/dnode/vnode/impl/test/vnodeApiTests.cpp

-/**
- * @file vnodeApiTests.cpp
- * @author hzcheng (hzcheng@taosdata.com)
- * @brief VNODE module API tests
- * @version 0.1
- * @date 2021-12-13
- *
- * @copyright Copyright (c) 2021
- *
- */
-
-#include <gtest/gtest.h>
-#include <iostream>
-
-#include "vnode.h"
-
-static STSchema *vtCreateBasicSchema() {
-  STSchemaBuilder sb;
-  STSchema *      pSchema = NULL;
-
-  tdInitTSchemaBuilder(&sb, 0);
-
-  tdAddColToSchema(&sb, TSDB_DATA_TYPE_TIMESTAMP, 0, 0);
-  for (int i = 1; i < 10; i++) {
-    tdAddColToSchema(&sb, TSDB_DATA_TYPE_INT, i, 0);
-  }
-
-  pSchema = tdGetSchemaFromBuilder(&sb);
-
-  tdDestroyTSchemaBuilder(&sb);
-
-  return pSchema;
-}
-
-static STSchema *vtCreateBasicTagSchema() {
-  STSchemaBuilder sb;
-  STSchema *      pSchema = NULL;
-
-  tdInitTSchemaBuilder(&sb, 0);
-
-  tdAddColToSchema(&sb, TSDB_DATA_TYPE_TIMESTAMP, 0, 0);
-  for (int i = 10; i < 12; i++) {
-    tdAddColToSchema(&sb, TSDB_DATA_TYPE_BINARY, i, 20);
-  }
-
-  pSchema = tdGetSchemaFromBuilder(&sb);
-
-  tdDestroyTSchemaBuilder(&sb);
-
-  return pSchema;
-}
-
-static SKVRow vtCreateBasicTag() {
-  SKVRowBuilder rb;
-  SKVRow        pTag;
-
-  tdInitKVRowBuilder(&rb);
-
-  for (int i = 0; i < 2; i++) {
-    void *pVal = malloc(sizeof(VarDataLenT) + strlen("foo"));
-    varDataLen(pVal) = strlen("foo");
-    memcpy(varDataVal(pVal), "foo", strlen("foo"));
-
-    tdAddColToKVRow(&rb, i, TSDB_DATA_TYPE_BINARY, pVal);
-    free(pVal);
-  }
-
-  pTag = tdGetKVRowFromBuilder(&rb);
-  tdDestroyKVRowBuilder(&rb);
-
-  return pTag;
-}
-
-static void vtBuildCreateStbReq(tb_uid_t suid, char *tbname, SRpcMsg **ppMsg) {
-  SRpcMsg * pMsg;
-  STSchema *pSchema;
-  STSchema *pTagSchema;
-  int       zs;
-  void *    pBuf;
-
-  pSchema = vtCreateBasicSchema();
-  pTagSchema = vtCreateBasicTagSchema();
-
-  SVnodeReq vCreateSTbReq;
-  vnodeSetCreateStbReq(&vCreateSTbReq, tbname, UINT32_MAX, UINT32_MAX, suid, pSchema, pTagSchema);
-
-  zs = vnodeBuildReq(NULL, &vCreateSTbReq, TDMT_VND_CREATE_STB);
-  pMsg = (SRpcMsg *)malloc(sizeof(SRpcMsg) + zs);
-  pMsg->msgType = TDMT_VND_CREATE_STB;
-  pMsg->contLen = zs;
-  pMsg->pCont = POINTER_SHIFT(pMsg, sizeof(SRpcMsg));
-
-  pBuf = pMsg->pCont;
-  vnodeBuildReq(&pBuf, &vCreateSTbReq, TDMT_VND_CREATE_STB);
-  META_CLEAR_TB_CFG(&vCreateSTbReq);
-
-  tdFreeSchema(pSchema);
-  tdFreeSchema(pTagSchema);
-
-  *ppMsg = pMsg;
-}
-
-static void vtBuildCreateCtbReq(tb_uid_t suid, char *tbname, SRpcMsg **ppMsg) {
-  SRpcMsg *pMsg;
-  int      tz;
-  SKVRow   pTag = vtCreateBasicTag();
-
-  SVnodeReq vCreateCTbReq;
-  vnodeSetCreateCtbReq(&vCreateCTbReq, tbname, UINT32_MAX, UINT32_MAX, suid, pTag);
-
-  tz = vnodeBuildReq(NULL, &vCreateCTbReq, TDMT_VND_CREATE_TABLE);
-  pMsg = (SRpcMsg *)malloc(sizeof(SRpcMsg) + tz);
-  pMsg->msgType = TDMT_VND_CREATE_TABLE;
-  pMsg->contLen = tz;
-  pMsg->pCont = POINTER_SHIFT(pMsg, sizeof(*pMsg));
-  void *pBuf = pMsg->pCont;
-
-  vnodeBuildReq(&pBuf, &vCreateCTbReq, TDMT_VND_CREATE_TABLE);
-  META_CLEAR_TB_CFG(&vCreateCTbReq);
-  free(pTag);
-
-  *ppMsg = pMsg;
-}
-
-static void vtBuildCreateNtbReq(char *tbname, SRpcMsg **ppMsg) {
-  // TODO
-}
-
-static void vtBuildSubmitReq(SRpcMsg **ppMsg) {
-  SRpcMsg *   pMsg;
-  SSubmitMsg *pSubmitMsg;
-  SSubmitBlk *pSubmitBlk;
-  int         tz = 1024;  // TODO
-
-  pMsg = (SRpcMsg *)malloc(sizeof(*pMsg) + tz);
-  pMsg->msgType = TDMT_VND_SUBMIT;
-  pMsg->contLen = tz;
-  pMsg->pCont = POINTER_SHIFT(pMsg, sizeof(*pMsg));
-
-  // For submit msg header
-  pSubmitMsg = (SSubmitMsg *)(pMsg->pCont);
-  // pSubmitMsg->header.contLen = 0;
-  // pSubmitMsg->header.vgId = 0;
-  // pSubmitMsg->length = 0;
-  pSubmitMsg->numOfBlocks = 1;
-
-  // For submit blk
-  pSubmitBlk = (SSubmitBlk *)(pSubmitMsg->blocks);
-  pSubmitBlk->uid = 0;
-  pSubmitBlk->tid = 0;
-  pSubmitBlk->padding = 0;
-  pSubmitBlk->sversion = 0;
-  pSubmitBlk->dataLen = 0;
-  pSubmitBlk->numOfRows = 0;
-
-  // For row batch
-
-  *ppMsg = pMsg;
-}
-
-static void vtClearMsgBatch(SArray *pMsgArr) {
-  SRpcMsg *pMsg;
-  for (size_t i = 0; i < taosArrayGetSize(pMsgArr); i++) {
-    pMsg = *(SRpcMsg **)taosArrayGet(pMsgArr, i);
-    free(pMsg);
-  }
-
-  taosArrayClear(pMsgArr);
-}
-
-static void vtProcessAndApplyReqs(SVnode *pVnode, SArray *pMsgArr) {
-  int      rcode;
-  SRpcMsg *pReq;
-  SRpcMsg *pRsp;
-
-  rcode = vnodeProcessWMsgs(pVnode, pMsgArr);
-  GTEST_ASSERT_EQ(rcode, 0);
-
-  for (size_t i = 0; i < taosArrayGetSize(pMsgArr); i++) {
-    pReq = *(SRpcMsg **)taosArrayGet(pMsgArr, i);
-    rcode = vnodeApplyWMsg(pVnode, pReq, NULL);
-    GTEST_ASSERT_EQ(rcode, 0);
-  }
-}
-
-TEST(vnodeApiTest, vnode_simple_create_table_test) {
-  tb_uid_t suid = 1638166374163;
-  SRpcMsg *pMsg;
-  SArray * pMsgArr = NULL;
-  SVnode * pVnode;
-  int      rcode;
-  int      ntables = 1000000;
-  int      batch = 10;
-  char     tbname[128];
-
-  pMsgArr = (SArray *)taosArrayInit(batch, sizeof(pMsg));
-
-  vnodeDestroy("vnode1");
-  GTEST_ASSERT_GE(vnodeInit(2), 0);
-
-  // CREATE AND OPEN A VNODE
-  pVnode = vnodeOpen("vnode1", NULL);
-  ASSERT_NE(pVnode, nullptr);
-
-  // CREATE A SUPER TABLE
-  sprintf(tbname, "st");
-  vtBuildCreateStbReq(suid, tbname, &pMsg);
-  taosArrayPush(pMsgArr, &pMsg);
-  vtProcessAndApplyReqs(pVnode, pMsgArr);
-  vtClearMsgBatch(pMsgArr);
-
-  // CREATE A LOT OF CHILD TABLES
-  for (int i = 0; i < ntables / batch; i++) {
-    // Build request batch
-    for (int j = 0; j < batch; j++) {
-      sprintf(tbname, "ct%d", i * batch + j + 1);
-      vtBuildCreateCtbReq(suid, tbname, &pMsg);
-      taosArrayPush(pMsgArr, &pMsg);
-    }
-
-    // Process request batch
-    vtProcessAndApplyReqs(pVnode, pMsgArr);
-
-    // Clear request batch
-    vtClearMsgBatch(pMsgArr);
-  }
-
-  // CLOSE THE VNODE
-  vnodeClose(pVnode);
-  vnodeClear();
-
-  taosArrayDestroy(pMsgArr);
-}
-
-TEST(vnodeApiTest, vnode_simple_insert_test) {
-  const char *vname = "vnode2";
-  char        tbname[128];
-  tb_uid_t    suid = 1638166374163;
-  SRpcMsg *   pMsg;
-  SArray *    pMsgArr;
-  int         rcode;
-  SVnode *    pVnode;
-  int         batch = 1;
-  int         loop = 1000000;
-
-  pMsgArr = (SArray *)taosArrayInit(0, sizeof(pMsg));
-
-  vnodeDestroy(vname);
-
-  GTEST_ASSERT_GE(vnodeInit(2), 0);
-
-  // Open a vnode
-  pVnode = vnodeOpen(vname, NULL);
-  GTEST_ASSERT_NE(pVnode, nullptr);
-
-  // 1. CREATE A SUPER TABLE
-  sprintf(tbname, "st");
-  vtBuildCreateStbReq(suid, tbname, &pMsg);
-  taosArrayPush(pMsgArr, &pMsg);
-  vtProcessAndApplyReqs(pVnode, pMsgArr);
-  vtClearMsgBatch(pMsgArr);
-
-  // 2. CREATE A CHILD TABLE
-  sprintf(tbname, "t0");
-  vtBuildCreateCtbReq(suid, tbname, &pMsg);
-  taosArrayPush(pMsgArr, &pMsg);
-  vtProcessAndApplyReqs(pVnode, pMsgArr);
-  vtClearMsgBatch(pMsgArr);
-
-  // 3. WRITE A LOT OF TIME-SERIES DATA
-  for (int j = 0; j < loop; j++) {
-    for (int i = 0; i < batch; i++) {
-      vtBuildSubmitReq(&pMsg);
-      taosArrayPush(pMsgArr, &pMsg);
-    }
-    vtProcessAndApplyReqs(pVnode, pMsgArr);
-    vtClearMsgBatch(pMsgArr);
-  }
-
-  // Close the vnode
-  vnodeClose(pVnode);
-  vnodeClear();
-
-  taosArrayDestroy(pMsgArr);
-}
\ No newline at end of file

include/dnode/vnode/tsdb/tsdb.h → source/dnode/vnode/inc/tsdb.h

@@ -18,6 +18,7 @@
 
 #include "mallocator.h"
 #include "meta.h"
+#include "common.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -39,6 +40,10 @@ typedef struct STable {
   STSchema *pSchema;
 } STable;
 
+#define BLOCK_LOAD_OFFSET_SEQ_ORDER   1
+#define BLOCK_LOAD_TABLE_SEQ_ORDER    2
+#define BLOCK_LOAD_TABLE_RR_ORDER     3
+
 #define TABLE_TID(t) (t)->tid
 #define TABLE_UID(t) (t)->uid
 
@@ -58,6 +63,22 @@ typedef struct STsdbCfg {
   int8_t   compression;
 } STsdbCfg;
 
+// query condition to build multi-table data block iterator
+typedef struct STsdbQueryCond {
+  STimeWindow  twindow;
+  int32_t      order;             // desc|asc order to iterate the data block
+  int32_t      numOfCols;
+  SColumnInfo *colList;
+  bool         loadExternalRows;  // load external rows or not
+  int32_t      type;              // data block load type:
+} STsdbQueryCond;
+
+typedef struct {
+  void    *pTable;
+  TSKEY    lastKey;
+  uint64_t uid;
+} STableKeyInfo;
+
 // STsdb
 STsdb *tsdbOpen(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF, SMeta *pMeta);
 void   tsdbClose(STsdb *);
@@ -70,6 +91,119 @@ int    tsdbCommit(STsdb *pTsdb);
 int  tsdbOptionsInit(STsdbCfg *);
 void tsdbOptionsClear(STsdbCfg *);
 
+typedef void* tsdbReadHandleT;
+/**
+ * Get the data block iterator, starting from position according to the query condition
+ *
+ * @param tsdb       tsdb handle
+ * @param pCond      query condition, including time window, result set order, and basic required columns for each block
+ * @param tableInfoGroup  table object list in the form of set, grouped into different sets according to the
+ *                        group by condition
+ * @param qinfo      query info handle from query processor
+ * @return
+ */
+tsdbReadHandleT *tsdbQueryTables(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfoGroup, uint64_t qId,
+                                  void *pRef);
+
+/**
+ * Get the last row of the given query time window for all the tables in STableGroupInfo object.
+ * Note that only one data block with only row will be returned while invoking retrieve data block function for
+ * all tables in this group.
+ *
+ * @param tsdb   tsdb handle
+ * @param pCond  query condition, including time window, result set order, and basic required columns for each block
+ * @param tableInfo  table list.
+ * @return
+ */
+//tsdbReadHandleT tsdbQueryLastRow(STsdbRepo *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfo, uint64_t qId,
+//                                  SMemRef *pRef);
+
+
+tsdbReadHandleT tsdbQueryCacheLast(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, uint64_t qId, void* pMemRef);
+
+bool isTsdbCacheLastRow(tsdbReadHandleT* pTsdbReadHandle);
+
+/**
+ * get num of rows in mem table
+ *
+ * @param pHandle
+ * @return row size
+ */
+
+int64_t tsdbGetNumOfRowsInMemTable(tsdbReadHandleT* pHandle);
+
+/**
+ * move to next block if exists
+ *
+ * @param pTsdbReadHandle
+ * @return
+ */
+bool tsdbNextDataBlock(tsdbReadHandleT pTsdbReadHandle);
+
+/**
+ * Get current data block information
+ *
+ * @param pTsdbReadHandle
+ * @param pBlockInfo
+ * @return
+ */
+void tsdbRetrieveDataBlockInfo(tsdbReadHandleT *pTsdbReadHandle, SDataBlockInfo *pBlockInfo);
+
+/**
+ *
+ * Get the pre-calculated information w.r.t. current data block.
+ *
+ * In case of data block in cache, the pBlockStatis will always be NULL.
+ * If a block is not completed loaded from disk, the pBlockStatis will be NULL.
+
+ * @pBlockStatis the pre-calculated value for current data blocks. if the block is a cache block, always return 0
+ * @return
+ */
+int32_t tsdbRetrieveDataBlockStatisInfo(tsdbReadHandleT *pTsdbReadHandle, SDataStatis **pBlockStatis);
+
+/**
+ *
+ * The query condition with primary timestamp is passed to iterator during its constructor function,
+ * the returned data block must be satisfied with the time window condition in any cases,
+ * which means the SData data block is not actually the completed disk data blocks.
+ *
+ * @param pTsdbReadHandle      query handle
+ * @param pColumnIdList     required data columns id list
+ * @return
+ */
+SArray *tsdbRetrieveDataBlock(tsdbReadHandleT *pTsdbReadHandle, SArray *pColumnIdList);
+
+/**
+ * destroy the created table group list, which is generated by tag query
+ * @param pGroupList
+ */
+void tsdbDestroyTableGroup(STableGroupInfo *pGroupList);
+
+/**
+ * create the table group result including only one table, used to handle the normal table query
+ *
+ * @param tsdb        tsdbHandle
+ * @param uid         table uid
+ * @param pGroupInfo  the generated result
+ * @return
+ */
+int32_t tsdbGetOneTableGroup(STsdb *tsdb, uint64_t uid, TSKEY startKey, STableGroupInfo *pGroupInfo);
+
+/**
+ *
+ * @param tsdb
+ * @param pTableIdList
+ * @param pGroupInfo
+ * @return
+ */
+int32_t tsdbGetTableGroupFromIdList(STsdb *tsdb, SArray *pTableIdList, STableGroupInfo *pGroupInfo);
+
+/**
+ * clean up the query handle
+ * @param queryHandle
+ */
+void tsdbCleanupQueryHandle(tsdbReadHandleT queryHandle);
+
 #ifdef __cplusplus
 }
 #endif

include/dnode/vnode/vnode.h → source/dnode/vnode/inc/vnode.h

@@ -31,56 +31,50 @@ extern "C" {
 
 /* ------------------------ TYPES EXPOSED ------------------------ */
 typedef struct SVnode SVnode;
+typedef struct SDnode SDnode;
+typedef int32_t (*PutReqToVQueryQFp)(SDnode *pDnode, struct SRpcMsg *pReq);
+
 typedef struct SVnodeCfg {
   int32_t vgId;
-
-  /** vnode buffer pool options */
+  SDnode *pDnode;
   struct {
-    /** write buffer size */
     uint64_t wsize;
     uint64_t ssize;
     uint64_t lsize;
-    /** use heap allocator or arena allocator */
     bool isHeapAllocator;
   };
-
-  /** time to live of tables in this vnode */
   uint32_t ttl;
-
-  /** data to keep in this vnode */
   uint32_t keep;
-
-  /** if TS data is eventually consistency */
   bool isWeak;
-
-  /** TSDB config */
   STsdbCfg tsdbCfg;
-
-  /** META config */
   SMetaCfg metaCfg;
-
-  /** TQ config */
   STqCfg tqCfg;
-
-  /** WAL config */
   SWalCfg walCfg;
 } SVnodeCfg;
 
+typedef struct {
+  int32_t           sver;
+  char             *timezone;
+  char             *locale;
+  char             *charset;
+  uint16_t          nthreads;  // number of commit threads. 0 for no threads and a schedule queue should be given (TODO)
+  PutReqToVQueryQFp putReqToVQueryQFp;
+} SVnodeOpt;
+
 /* ------------------------ SVnode ------------------------ */
 /**
  * @brief Initialize the vnode module
  *
- * @param nthreads number of commit threads. 0 for no threads and
- *        a schedule queue should be given (TODO)
+ * @param pOption Option of the vnode mnodule
  * @return int 0 for success and -1 for failure
  */
-int vnodeInit(uint16_t nthreads);
+int vnodeInit(const SVnodeOpt *pOption);
 
 /**
- * @brief clear a vnode
+ * @brief Cleanup the vnode module
  *
  */
-void vnodeClear();
+void vnodeCleanup();
 
 /**
  * @brief Open a VNODE.
@@ -89,7 +83,7 @@ void vnodeClear();
  * @param pVnodeCfg options of the vnode
  * @return SVnode* The vnode object
  */
-SVnode *vnodeOpen(const char *path, const SVnodeCfg *pVnodeCfg, int32_t vid);
+SVnode *vnodeOpen(const char *path, const SVnodeCfg *pVnodeCfg);
 
 /**
  * @brief Close a VNODE

source/dnode/vnode/meta/CMakeLists.txt

-set(META_DB_IMPL_LIST "BDB" "SQLITE")
-set(META_DB_IMPL "BDB" CACHE STRING "Use BDB as the default META implementation")
-set_property(CACHE META_DB_IMPL PROPERTY STRINGS ${META_DB_IMPL_LIST})
-
-if(META_DB_IMPL IN_LIST META_DB_IMPL_LIST)
-    message(STATUS "META DB Impl: ${META_DB_IMPL}==============")
-else()
-    message(FATAL_ERROR "Invalid META DB IMPL: ${META_DB_IMPL}==============")
-endif()
-
-aux_source_directory(src META_SRC)
-if(${META_DB_IMPL} STREQUAL "BDB")
-    list(REMOVE_ITEM META_SRC "src/metaSQLiteImpl.c")
-elseif(${META_DB_IMPL} STREQUAL "SQLITE")
-    list(REMOVE_ITEM META_SRC "src/metaBDBImpl.c")
-endif()
-
-add_library(meta STATIC ${META_SRC})
-target_include_directories(
-    meta
-    PUBLIC "${CMAKE_SOURCE_DIR}/include/dnode/vnode/meta"
-    PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/index"
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
-)
-target_link_libraries(
-    meta
-    PUBLIC common
-    PUBLIC index 
-)
-
-if(${META_DB_IMPL} STREQUAL "BDB")
-    target_link_libraries(
-        meta
-        PUBLIC bdb
-    )
-elseif(${META_DB_IMPL} STREQUAL "SQLITE")
-    target_link_libraries(
-        meta
-        PUBLIC sqlite
-    )
-endif()
-
-
-if(${BUILD_TEST})
-    add_subdirectory(test)
-endif(${BUILD_TEST})

source/dnode/vnode/meta/src/metaSQLiteImpl.c

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#include "metaDef.h"
-#include "sqlite3.h"
-
-struct SMetaDB {
-  sqlite3 *pDB;
-};
-
-int metaOpenDB(SMeta *pMeta) {
-  char  dir[128];
-  int   rc;
-  char *err = NULL;
-
-  pMeta->pDB = (SMetaDB *)calloc(1, sizeof(SMetaDB));
-  if (pMeta->pDB == NULL) {
-    // TODO: handle error
-    return -1;
-  }
-
-  sprintf(dir, "%s/meta.db", pMeta->path);
-  rc = sqlite3_open(dir, &(pMeta->pDB->pDB));
-  if (rc != SQLITE_OK) {
-    // TODO: handle error
-    printf("failed to open meta.db\n");
-  }
-
-  // For all tables
-  rc = sqlite3_exec(pMeta->pDB->pDB,
-                    "CREATE TABLE IF NOT EXISTS tb ("
-                    "  tbname VARCHAR(256) NOT NULL UNIQUE,"
-                    "  tb_uid INTEGER NOT NULL UNIQUE "
-                    ");",
-                    NULL, NULL, &err);
-  if (rc != SQLITE_OK) {
-    // TODO: handle error
-    printf("failed to create meta table tb since %s\n", err);
-  }
-
-  // For super tables
-  rc = sqlite3_exec(pMeta->pDB->pDB,
-                    "CREATE TABLE IF NOT EXISTS stb ("
-                    "    tb_uid INTEGER NOT NULL UNIQUE,"
-                    "    tbname VARCHAR(256) NOT NULL UNIQUE,"
-                    "    tb_schema BLOB NOT NULL,"
-                    "    tag_schema BLOB NOT NULL"
-                    ");",
-                    NULL, NULL, &err);
-  if (rc != SQLITE_OK) {
-    // TODO: handle error
-    printf("failed to create meta table stb since %s\n", err);
-  }
-
-  // For normal tables
-  rc = sqlite3_exec(pMeta->pDB->pDB,
-                    "CREATE TABLE IF NOT EXISTS ntb ("
-                    "    tb_uid INTEGER NOT NULL UNIQUE,"
-                    "    tbname VARCHAR(256) NOT NULL,"
-                    "    tb_schema BLOB NOT NULL"
-                    ");",
-                    NULL, NULL, &err);
-  if (rc != SQLITE_OK) {
-    // TODO: handle error
-    printf("failed to create meta table ntb since %s\n", err);
-  }
-
-  sqlite3_exec(pMeta->pDB->pDB, "BEGIN;", NULL, NULL, &err);
-
-  tfree(err);
-
-  return 0;
-}
-
-void metaCloseDB(SMeta *pMeta) {
-  if (pMeta->pDB) {
-    sqlite3_exec(pMeta->pDB->pDB, "COMMIT;", NULL, NULL, NULL);
-    sqlite3_close(pMeta->pDB->pDB);
-    free(pMeta->pDB);
-    pMeta->pDB = NULL;
-  }
-
-  // TODO
-}
-
-int metaSaveTableToDB(SMeta *pMeta, const STbCfg *pTbCfg) {
-  char          sql[256];
-  char *        err = NULL;
-  int           rc;
-  tb_uid_t      uid;
-  sqlite3_stmt *stmt;
-  char          buf[256];
-  void *        pBuf;
-
-  switch (pTbCfg->type) {
-    case META_SUPER_TABLE:
-      uid = pTbCfg->stbCfg.suid;
-      sprintf(sql,
-              "INSERT INTO tb VALUES (\'%s\', %" PRIu64
-              ");"
-              "CREATE TABLE IF NOT EXISTS stb_%" PRIu64
-              " ("
-              "    tb_uid INTEGER NOT NULL UNIQUE,"
-              "    tbname VARCHAR(256),"
-              "    tag1 INTEGER);",
-              pTbCfg->name, uid, uid);
-      rc = sqlite3_exec(pMeta->pDB->pDB, sql, NULL, NULL, &err);
-      if (rc != SQLITE_OK) {
-        printf("failed to create normal table since %s\n", err);
-      }
-
-      sprintf(sql, "INSERT INTO stb VALUES (%" PRIu64 ", %s, ?, ?)", uid, pTbCfg->name);
-      sqlite3_prepare_v2(pMeta->pDB->pDB, sql, -1, &stmt, NULL);
-
-      pBuf = buf;
-      tdEncodeSchema(&pBuf, pTbCfg->stbCfg.pSchema);
-      sqlite3_bind_blob(stmt, 1, buf, POINTER_DISTANCE(pBuf, buf), NULL);
-      pBuf = buf;
-      tdEncodeSchema(&pBuf, pTbCfg->stbCfg.pTagSchema);
-      sqlite3_bind_blob(stmt, 2, buf, POINTER_DISTANCE(pBuf, buf), NULL);
-
-      sqlite3_step(stmt);
-
-      sqlite3_finalize(stmt);
-
-#if 0
-      sprintf(sql,
-              "INSERT INTO tb VALUES (?, ?);"
-              // "INSERT INTO stb VALUES (?, ?, ?, ?);"
-              // "CREATE TABLE IF NOT EXISTS stb_%" PRIu64
-              // " ("
-              // "    tb_uid INTEGER NOT NULL UNIQUE,"
-              // "    tbname VARCHAR(256),"
-              // "    tag1 INTEGER);"
-              ,
-              uid);
-      rc = sqlite3_prepare_v2(pMeta->pDB->pDB, sql, -1, &stmt, NULL);
-      if (rc != SQLITE_OK) {
-        return -1;
-      }
-      sqlite3_bind_text(stmt, 1, pTbCfg->name, -1, SQLITE_TRANSIENT);
-      sqlite3_bind_int64(stmt, 2, uid);
-      sqlite3_step(stmt);
-      sqlite3_finalize(stmt);
-
-
-      // sqlite3_bind_int64(stmt, 3, uid);
-      // sqlite3_bind_text(stmt, 4, pTbCfg->name, -1, SQLITE_TRANSIENT);
-      // pBuf = buf;
-      // tdEncodeSchema(&pBuf, pTbCfg->stbCfg.pSchema);
-      // sqlite3_bind_blob(stmt, 5, buf, POINTER_DISTANCE(pBuf, buf), NULL);
-      // pBuf = buf;
-      // tdEncodeSchema(&pBuf, pTbCfg->stbCfg.pTagSchema);
-      // sqlite3_bind_blob(stmt, 6, buf, POINTER_DISTANCE(pBuf, buf), NULL);
-
-      rc = sqliteVjj3_step(stmt);
-      if (rc != SQLITE_OK) {
-        printf("failed to create normal table since %s\n", sqlite3_errmsg(pMeta->pDB->pDB));
-      }
-      sqlite3_finalize(stmt);
-#endif
-      break;
-    case META_NORMAL_TABLE:
-      // uid = metaGenerateUid(pMeta);
-      // sprintf(sql,
-      //         "INSERT INTO tb VALUES (\'%s\', %" PRIu64
-      //         ");"
-      //         "INSERT INTO ntb VALUES (%" PRIu64 ", \'%s\', );",
-      //         pTbCfg->name, uid, uid, pTbCfg->name, );
-
-      // rc = sqlite3_exec(pMeta->pDB->pDB, sql, NULL, NULL, &err);
-      // if (rc != SQLITE_OK) {
-      //   printf("failed to create normal table since %s\n", err);
-      // }
-      break;
-    case META_CHILD_TABLE:
-#if 0
-      uid = metaGenerateUid(pMeta);
-      // sprintf(sql, "INSERT INTO tb VALUES (\'%s\', %" PRIu64
-      //              ");"
-      //              "INSERT INTO stb_%" PRIu64 " VALUES (%" PRIu64 ", \'%s\', );");
-      rc = sqlite3_exec(pMeta->pDB->pDB, sql, NULL, NULL, &err);
-      if (rc != SQLITE_OK) {
-        printf("failed to create child table since %s\n", err);
-      }
-#endif
-      break;
-    default:
-      break;
-  }
-
-  tfree(err);
-
-  return 0;
-}
-
-int metaRemoveTableFromDb(SMeta *pMeta, tb_uid_t uid) {
-  /* TODO */
-  return 0;
-}
\ No newline at end of file

source/dnode/vnode/meta/test/CMakeLists.txt

-# add_executable(metaTest "")
-# target_sources(metaTest
-#     PRIVATE
-#     "../src/metaMain.c"
-#     "../src/metaUid.c"
-#     "metaTests.cpp"
-# )
-# target_include_directories(metaTest 
-#     PUBLIC
-#     "${CMAKE_SOURCE_DIR}/include/server/vnode/meta"
-#     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
-# )
-# target_link_libraries(metaTest
-#     os
-#     util
-#     common
-#     gtest_main
-#     tkv
-# )
-# enable_testing()
-# add_test(
-#     NAME meta_test
-#     COMMAND metaTest
-# )

source/dnode/vnode/meta/test/metaAPITest.cpp

-#if 0
-#include <gtest/gtest.h>
-#include <string.h>
-#include <iostream>
-
-#include "meta.h"
-
-static STSchema *metaGetSimpleSchema() {
-  STSchema *      pSchema = NULL;
-  STSchemaBuilder sb = {0};
-
-  tdInitTSchemaBuilder(&sb, 0);
-  tdAddColToSchema(&sb, TSDB_DATA_TYPE_TIMESTAMP, 0, 8);
-  tdAddColToSchema(&sb, TSDB_DATA_TYPE_INT, 1, 4);
-
-  pSchema = tdGetSchemaFromBuilder(&sb);
-  tdDestroyTSchemaBuilder(&sb);
-
-  return pSchema;
-}
-
-static SKVRow metaGetSimpleTags() {
-  SKVRowBuilder kvrb = {0};
-  SKVRow        row;
-
-  tdInitKVRowBuilder(&kvrb);
-  int64_t ts = 1634287978000;
-  int32_t a = 10;
-
-  tdAddColToKVRow(&kvrb, 0, TSDB_DATA_TYPE_TIMESTAMP, (void *)(&ts));
-  tdAddColToKVRow(&kvrb, 0, TSDB_DATA_TYPE_INT, (void *)(&a));
-
-  row = tdGetKVRowFromBuilder(&kvrb);
-
-  tdDestroyKVRowBuilder(&kvrb);
-
-  return row;
-}
-
-TEST(MetaTest, DISABLED_meta_create_1m_normal_tables_test) {
-  // Open Meta
-  SMeta *meta = metaOpen(NULL, NULL);
-  std::cout << "Meta is opened!" << std::endl;
-
-  // Create 1000000 normal tables
-  META_TABLE_OPTS_DECLARE(tbOpts);
-  STSchema *pSchema = metaGetSimpleSchema();
-  char      tbname[128];
-
-  for (size_t i = 0; i < 1000000; i++) {
-    sprintf(tbname, "ntb%ld", i);
-    metaNormalTableOptsInit(&tbOpts, tbname, pSchema);
-    metaCreateTable(meta, &tbOpts);
-    metaTableOptsClear(&tbOpts);
-  }
-
-  tdFreeSchema(pSchema);
-
-  // Close Meta
-  metaClose(meta);
-  std::cout << "Meta is closed!" << std::endl;
-
-  // Destroy Meta
-  metaDestroy("meta");
-  std::cout << "Meta is destroyed!" << std::endl;
-}
-
-TEST(MetaTest, meta_create_1m_child_tables_test) {
-  // Open Meta
-  SMeta *meta = metaOpen(NULL);
-  std::cout << "Meta is opened!" << std::endl;
-
-  // Create a super tables
-  tb_uid_t uid = 477529885843758ul;
-  META_TABLE_OPTS_DECLARE(tbOpts);
-  STSchema *pSchema = metaGetSimpleSchema();
-  STSchema *pTagSchema = metaGetSimpleSchema();
-
-  metaSuperTableOptsInit(&tbOpts, "st", uid, pSchema, pTagSchema);
-  metaCreateTable(meta, &tbOpts);
-  metaTableOptsClear(&tbOpts);
-
-  tdFreeSchema(pSchema);
-  tdFreeSchema(pTagSchema);
-
-  // Create 1000000 child tables
-  char   name[128];
-  SKVRow row = metaGetSimpleTags();
-  for (size_t i = 0; i < 1000000; i++) {
-    sprintf(name, "ctb%ld", i);
-    metaChildTableOptsInit(&tbOpts, name, uid, row);
-    metaCreateTable(meta, &tbOpts);
-    metaTableOptsClear(&tbOpts);
-  }
-  kvRowFree(row);
-
-  // Close Meta
-  metaClose(meta);
-  std::cout << "Meta is closed!" << std::endl;
-
-  // Destroy Meta
-  metaDestroy("meta");
-  std::cout << "Meta is destroyed!" << std::endl;
-}
-#endif
\ No newline at end of file

source/dnode/vnode/impl/inc/vnodeDef.h → source/dnode/vnode/src/inc/vnodeDef.h

@@ -17,7 +17,7 @@
 #define _TD_VNODE_DEF_H_
 
 #include "mallocator.h"
-#include "sync.h"
+// #include "sync.h"
 #include "tcoding.h"
 #include "tlist.h"
 #include "tlockfree.h"
@@ -30,12 +30,9 @@
 #include "vnodeBufferPool.h"
 #include "vnodeCfg.h"
 #include "vnodeCommit.h"
-#include "vnodeFS.h"
 #include "vnodeMemAllocator.h"
 #include "vnodeQuery.h"
-#include "vnodeRequest.h"
 #include "vnodeStateMgr.h"
-#include "vnodeSync.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -57,6 +54,8 @@ typedef struct SVnodeMgr {
   pthread_cond_t  hasTask;
   TD_DLIST(SVnodeTask) queue;
   // For vnode Mgmt
+  SDnode*           pDnode;
+  PutReqToVQueryQFp putReqToVQueryQFp;
 } SVnodeMgr;
 
 extern SVnodeMgr vnodeMgr;
@@ -71,14 +70,25 @@ struct SVnode {
   STsdb*      pTsdb;
   STQ*        pTq;
   SWal*       pWal;
-  SVnodeSync* pSync;
-  SVnodeFS*   pFs;
   tsem_t      canCommit;
   SQHandle*   pQuery;
+  SDnode*     pDnode;
 };
 
 int vnodeScheduleTask(SVnodeTask* task);
 
+int32_t vnodePutReqToVQueryQ(SVnode *pVnode, struct SRpcMsg *pReq);
+
+// For Log
+extern int32_t vDebugFlag;
+
+#define vFatal(...) do { if (vDebugFlag & DEBUG_FATAL) { taosPrintLog("TDB FATAL ", 255, __VA_ARGS__); }}     while(0)
+#define vError(...) do { if (vDebugFlag & DEBUG_ERROR) { taosPrintLog("TDB ERROR ", 255, __VA_ARGS__); }}     while(0)
+#define vWarn(...)  do { if (vDebugFlag & DEBUG_WARN)  { taosPrintLog("TDB WARN ", 255, __VA_ARGS__); }}      while(0)
+#define vInfo(...)  do { if (vDebugFlag & DEBUG_INFO)  { taosPrintLog("TDB ", 255, __VA_ARGS__); }}           while(0)
+#define vDebug(...) do { if (vDebugFlag & DEBUG_DEBUG) { taosPrintLog("TDB ", tsdbDebugFlag, __VA_ARGS__); }} while(0)
+#define vTrace(...) do { if (vDebugFlag & DEBUG_TRACE) { taosPrintLog("TDB ", tsdbDebugFlag, __VA_ARGS__); }} while(0)
+
 #ifdef __cplusplus
 }
 #endif