Merge branch '3.0' into feature/vnode

include/common/tmsg.h

@@ -262,6 +262,26 @@ typedef struct {
   char     data[];
 } SMDCreateTableMsg;
 
+// typedef struct {
+//  int32_t len;  // one create table message
+//  char    tableName[TSDB_TABLE_FNAME_LEN];
+//  int16_t numOfColumns;
+//  int16_t sqlLen;  // the length of SQL, it starts after schema , sql is a null-terminated string
+//  int8_t  igExists;
+//  int8_t  rspMeta;
+//  int8_t  reserved[16];
+//  char    schema[];
+//} SCreateTableMsg;
+
+typedef struct {
+  char    tableName[TSDB_TABLE_FNAME_LEN];
+  int16_t numOfColumns;
+  int16_t numOfTags;
+  int8_t  igExists;
+  int8_t  rspMeta;
+  char    schema[];
+} SCreateCTableMsg;
+
 typedef struct {
   char    name[TSDB_TABLE_FNAME_LEN];
   int8_t  igExists;
@@ -342,6 +362,18 @@ typedef struct SEpSet {
   char     fqdn[TSDB_MAX_REPLICA][TSDB_FQDN_LEN];
 } SEpSet;
 
+static FORCE_INLINE int taosEncodeSEpSet(void** buf, const SEpSet* pEp) {
+  if(buf == NULL) return sizeof(SEpSet);
+  memcpy(buf, pEp, sizeof(SEpSet));
+  //TODO: endian conversion
+  return sizeof(SEpSet);
+}
+
+static FORCE_INLINE void* taosDecodeSEpSet(void* buf, SEpSet* pEpSet) {
+  memcpy(pEpSet, buf, sizeof(SEpSet));
+  return buf;
+}
+
 typedef struct {
   int32_t acctId;
   int64_t clusterId;
@@ -685,8 +717,6 @@ typedef struct {
 typedef struct {
   int32_t dnodeId;
   int64_t clusterId;
-  int8_t  dropped;
-  char    reserved[7];
 } SDnodeCfg;
 
 typedef struct {
@@ -1101,26 +1131,94 @@ typedef struct STaskDropRsp {
 } STaskDropRsp;
 
 typedef struct {
-  int8_t igExists;
-  char*  name;
-  char*  phyPlan;
+  int8_t  igExists;
+  char*   name;
+  char*   physicalPlan;
+  char*   logicalPlan;
 } SCMCreateTopicReq;
 
 static FORCE_INLINE int tSerializeSCMCreateTopicReq(void** buf, const SCMCreateTopicReq* pReq) {
   int tlen = 0;
   tlen += taosEncodeString(buf, pReq->name);
   tlen += taosEncodeFixedI8(buf, pReq->igExists);
-  tlen += taosEncodeString(buf, pReq->phyPlan);
+  tlen += taosEncodeString(buf, pReq->physicalPlan);
+  tlen += taosEncodeString(buf, pReq->logicalPlan);
   return tlen;
 }
 
 static FORCE_INLINE void* tDeserializeSCMCreateTopicReq(void* buf, SCMCreateTopicReq* pReq) {
   buf = taosDecodeFixedI8(buf, &(pReq->igExists));
   buf = taosDecodeString(buf, &(pReq->name));
-  buf = taosDecodeString(buf, &(pReq->phyPlan));
+  buf = taosDecodeString(buf, &(pReq->physicalPlan));
+  buf = taosDecodeString(buf, &(pReq->logicalPlan));
   return buf;
 }
 
+typedef struct {
+  int64_t topicId;
+} SCMCreateTopicRsp;
+
+static FORCE_INLINE int tSerializeSCMCreateTopicRsp(void** buf, const SCMCreateTopicRsp* pRsp) {
+  int tlen = 0;
+  tlen += taosEncodeFixedI64(buf, pRsp->topicId);
+  return tlen;
+}
+
+static FORCE_INLINE void* tDeserializeSCMCreateTopicRsp(void* buf, SCMCreateTopicRsp* pRsp) {
+  buf = taosDecodeFixedI64(buf, &pRsp->topicId);
+  return buf;
+}
+
+typedef struct {
+  char* topicName;
+  char* consumerGroup;
+  int64_t consumerId;
+} SCMSubscribeReq;
+
+static FORCE_INLINE int tSerializeSCMSubscribeReq(void** buf, const SCMSubscribeReq* pReq) {
+  int tlen = 0;
+  tlen += taosEncodeString(buf, pReq->topicName);
+  tlen += taosEncodeString(buf, pReq->consumerGroup);
+  tlen += taosEncodeFixedI64(buf, pReq->consumerId);
+  return tlen;
+}
+
+static FORCE_INLINE void* tDeserializeSCMSubscribeReq(void* buf, SCMSubscribeReq* pReq) {
+  buf = taosDecodeString(buf, &pReq->topicName);
+  buf = taosDecodeString(buf, &pReq->consumerGroup);
+  buf = taosDecodeFixedI64(buf, &pReq->consumerId);
+  return buf;
+}
+
+typedef struct {
+  int32_t vgId;
+  SEpSet pEpSet;
+} SCMSubscribeRsp;
+
+static FORCE_INLINE int tSerializeSCMSubscribeRsp(void** buf, const SCMSubscribeRsp* pRsp) {
+  int tlen = 0;
+  tlen += taosEncodeFixedI32(buf, pRsp->vgId);
+  tlen += taosEncodeSEpSet(buf, &pRsp->pEpSet);
+  return tlen;
+}
+
+static FORCE_INLINE void* tDeserializeSCMSubscribeRsp(void* buf, SCMSubscribeRsp* pRsp) {
+  buf = taosDecodeFixedI32(buf, &pRsp->vgId);
+  buf = taosDecodeSEpSet(buf, &pRsp->pEpSet);
+  return buf;
+}
+
+typedef struct {
+  int64_t topicId;
+  int64_t consumerId;
+  int64_t consumerGroupId;
+  int64_t offset;
+} SMVSubscribeReq;
+
+typedef struct {
+  int64_t newOffset;
+} SMVSubscribeRsp;
+
 typedef struct {
   char    name[TSDB_TOPIC_FNAME_LEN];
   int8_t  igExists;

include/common/tmsgdef.h

@@ -116,9 +116,10 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_MND_TRANS, "mnode-trans", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_GRANT, "mnode-grant", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_AUTH, "mnode-auth", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_TOPIC, "mnode-create-topic", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_TOPIC, "mnode-create-topic", SCMCreateTopicReq, SCMCreateTopicRsp)
   TD_DEF_MSG_TYPE(TDMT_MND_ALTER_TOPIC, "mnode-alter-topic", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_DROP_TOPIC, "mnode-drop-topic", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_SUBSCRIBE, "mnode-subscribe", SCMSubscribeReq, SCMSubscribeRsp)
 
   // Requests handled by VNODE
   TD_NEW_MSG_SEG(TDMT_VND_MSG)
@@ -149,6 +150,9 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_SHOW_TABLES, "vnode-show-tables", SVShowTablesReq, SVShowTablesRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_SHOW_TABLES_FETCH, "vnode-show-tables-fetch", SVShowTablesFetchReq, SVShowTablesFetchRsp)
 
+  TD_DEF_MSG_TYPE(TDMT_VND_SUBSCRIBE, "vnode-subscribe", SMVSubscribeReq, SMVSubscribeRsp)
+
+
   // Requests handled by QNODE
   TD_NEW_MSG_SEG(TDMT_QND_MSG)
 
@@ -158,4 +162,4 @@ enum {
 #if defined(TD_MSG_NUMBER_)
   TDMT_MAX
 #endif
-};
+};
\ No newline at end of file

include/dnode/mnode/sdb/sdb.h

@@ -25,7 +25,7 @@ extern "C" {
 #define SDB_GET_INT64(pData, pRow, dataPos, val)   \
   {                                                \
     if (sdbGetRawInt64(pRaw, dataPos, val) != 0) { \
-      sdbFreeRow(pRow);                            \
+      tfree(pRow);                                 \
       return NULL;                                 \
     }                                              \
     dataPos += sizeof(int64_t);                    \
@@ -34,7 +34,7 @@ extern "C" {
 #define SDB_GET_INT32(pData, pRow, dataPos, val)   \
   {                                                \
     if (sdbGetRawInt32(pRaw, dataPos, val) != 0) { \
-      sdbFreeRow(pRow);                            \
+      tfree(pRow);                                 \
       return NULL;                                 \
     }                                              \
     dataPos += sizeof(int32_t);                    \
@@ -43,7 +43,7 @@ extern "C" {
 #define SDB_GET_INT16(pData, pRow, dataPos, val)   \
   {                                                \
     if (sdbGetRawInt16(pRaw, dataPos, val) != 0) { \
-      sdbFreeRow(pRow);                            \
+      tfree(pRow);                                 \
       return NULL;                                 \
     }                                              \
     dataPos += sizeof(int16_t);                    \
@@ -52,7 +52,7 @@ extern "C" {
 #define SDB_GET_INT8(pData, pRow, dataPos, val)   \
   {                                               \
     if (sdbGetRawInt8(pRaw, dataPos, val) != 0) { \
-      sdbFreeRow(pRow);                           \
+      tfree(pRow);                                \
       return NULL;                                \
     }                                             \
     dataPos += sizeof(int8_t);                    \
@@ -61,7 +61,7 @@ extern "C" {
 #define SDB_GET_BINARY(pRaw, pRow, dataPos, val, valLen)    \
   {                                                         \
     if (sdbGetRawBinary(pRaw, dataPos, val, valLen) != 0) { \
-      sdbFreeRow(pRow);                                     \
+      tfree(pRow);                                          \
       return NULL;                                          \
     }                                                       \
     dataPos += valLen;                                      \
@@ -71,7 +71,7 @@ extern "C" {
   {                                                         \
     char val[valLen] = {0};                                 \
     if (sdbGetRawBinary(pRaw, dataPos, val, valLen) != 0) { \
-      sdbFreeRow(pRow);                                     \
+      tfree(pRow);                                          \
       return NULL;                                          \
     }                                                       \
     dataPos += valLen;                                      \
@@ -161,12 +161,14 @@ typedef enum {
   SDB_USER = 5,
   SDB_AUTH = 6,
   SDB_ACCT = 7,
-  SDB_TOPIC = 8,
-  SDB_VGROUP = 9,
-  SDB_STB = 10,
-  SDB_DB = 11,
-  SDB_FUNC = 12,
-  SDB_MAX = 13
+  SDB_CONSUMER = 8,
+  SDB_CGROUP = 9,
+  SDB_TOPIC = 10,
+  SDB_VGROUP = 11,
+  SDB_STB = 12,
+  SDB_DB = 13,
+  SDB_FUNC = 14,
+  SDB_MAX = 15
 } ESdbType;
 
 typedef struct SSdb SSdb;
@@ -325,7 +327,7 @@ int32_t  sdbGetRawSoftVer(SSdbRaw *pRaw, int8_t *sver);
 int32_t  sdbGetRawTotalSize(SSdbRaw *pRaw);
 
 SSdbRow *sdbAllocRow(int32_t objSize);
-void     sdbFreeRow(SSdbRow *pRow);
+void     sdbFreeRow(SSdb *pSdb, SSdbRow *pRow);
 void    *sdbGetRowObj(SSdbRow *pRow);
 
 #ifdef __cplusplus

include/dnode/vnode/vnode.h

@@ -162,16 +162,6 @@ int vnodeProcessQueryReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
  */
 int vnodeProcessFetchReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
 
-/**
- * @brief Process a consume message.
- *
- * @param pVnode The vnode object.
- * @param pMsg The request message
- * @param pRsp The response message
- * @return int 0 for success, -1 for failure
- */
-int vnodeProcessConsumeReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
-
 /* ------------------------ SVnodeCfg ------------------------ */
 /**
  * @brief Initialize VNODE options.

include/libs/catalog/catalog.h

@@ -62,6 +62,18 @@ int32_t catalogGetHandle(const char *clusterId, struct SCatalog** catalogHandle)
 
 int32_t catalogGetDBVgroupVersion(struct SCatalog* pCatalog, const char* dbName, int32_t* version);
 
+/**
+ * Get a DB's all vgroup info.
+ * @param pCatalog (input, got with catalogGetHandle)
+ * @param pRpc (input, rpc object)
+ * @param pMgmtEps (input, mnode EPs)
+ * @param pDBName (input, full db name)
+ * @param forceUpdate (input, force update db vgroup info from mnode) 
+ * @param pVgroupList (output, vgroup info list, element is SVgroupInfo, NEED to simply free the array by caller)
+ * @return error code
+ */
+int32_t catalogGetDBVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pDBName, int32_t forceUpdate, SArray** pVgroupList);
+
 int32_t catalogUpdateDBVgroup(struct SCatalog* pCatalog, const char* dbName, SDBVgroupInfo* dbInfo);
 
 /**

include/libs/qcom/query.h

@@ -75,6 +75,7 @@ typedef struct STableMeta {
 } STableMeta;
 
 typedef struct SDBVgroupInfo {
+  int32_t   lock;
   int32_t   vgVersion;  
   int8_t    hashMethod;
   SHashObj *vgInfo;  //key:vgId, value:SVgroupInfo

include/util/thash.h

@@ -144,6 +144,16 @@ void *taosHashGet(SHashObj *pHashObj, const void *key, size_t keyLen);
  */
 void *taosHashGetClone(SHashObj *pHashObj, const void *key, size_t keyLen, void* destBuf);
 
+/**
+ * Clone the result to interval allocated buffer
+ * @param pHashObj
+ * @param key
+ * @param keyLen
+ * @param destBuf
+ * @return
+ */
+void* taosHashGetCloneExt(SHashObj *pHashObj, const void *key, size_t keyLen, void (*fp)(void *), void** d, size_t *sz);
+
 /**
  * remove item with the specified key
  * @param pHashObj
@@ -200,6 +210,26 @@ void  taosHashCancelIterate(SHashObj *pHashObj, void *p);
  */
 int32_t taosHashGetKey(void *data, void** key, size_t* keyLen);
 
+/**
+ * return the payload data with the specified key(reference number added)
+ *
+ * @param pHashObj
+ * @param key
+ * @param keyLen
+ * @return
+ */
+void* taosHashAcquire(SHashObj *pHashObj, const void *key, size_t keyLen);
+
+/**
+ * release the prevous acquired obj
+ *
+ * @param pHashObj
+ * @param data
+ * @return
+ */
+void taosHashRelease(SHashObj *pHashObj, void *p);
+
+
 #ifdef __cplusplus
 }
 #endif

source/client/src/clientImpl.c

@@ -210,6 +210,9 @@ TAOS_RES *tmq_create_topic(TAOS* taos, const char* name, const char* sql, int sq
   terrno = TSDB_CODE_SUCCESS;
 
   CHECK_CODE_GOTO(buildRequest(pTscObj, sql, sqlLen, &pRequest), _return);
+
+//temporary disabled until planner ready
+#if 0
   CHECK_CODE_GOTO(parseSql(pRequest, &pQuery), _return);
   //TODO: check sql valid
 
@@ -219,15 +222,24 @@ TAOS_RES *tmq_create_topic(TAOS* taos, const char* name, const char* sql, int sq
   if(dagStr == NULL) {
     //TODO
   }
+#endif
 
   SCMCreateTopicReq req = {
     .name = (char*)name,
     .igExists = 0,
-    .phyPlan = dagStr,
+    /*.physicalPlan = dagStr,*/
+    .physicalPlan = (char*)sql,
+    .logicalPlan = "",
   };
 
-  void* buf = NULL;
-  int tlen = tSerializeSCMCreateTopicReq(&buf, &req);
+  int tlen = tSerializeSCMCreateTopicReq(NULL, &req);
+  void* buf = malloc(tlen);
+  if(buf == NULL) {
+    goto _return;
+  }
+  void* abuf = buf;
+  tSerializeSCMCreateTopicReq(&abuf, &req);
+  /*printf("formatted: %s\n", dagStr);*/
 
   pRequest->body.requestMsg = (SDataBuf){ .pData = buf, .len = tlen };
 
@@ -239,8 +251,6 @@ TAOS_RES *tmq_create_topic(TAOS* taos, const char* name, const char* sql, int sq
 
   tsem_wait(&pRequest->body.rspSem);
 
-  destroySendMsgInfo(body);
-
 _return:
   qDestroyQuery(pQuery);
   qDestroyQueryDag(pDag); 

source/client/src/clientMsgHandler.c

@@ -188,14 +188,21 @@ int32_t processCreateDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
 }
 
 int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
-  SUseDbRsp* pUseDbRsp = (SUseDbRsp*) pMsg->pData;
-  SName name = {0};
-  tNameFromString(&name, pUseDbRsp->db, T_NAME_ACCT|T_NAME_DB);
+  SRequestObj* pRequest = param;
+
+  if (code != TSDB_CODE_SUCCESS) {
+    pRequest->code = code;
+    tsem_post(&pRequest->body.rspSem);
+    return code;
+  }
+
+  SUseDbRsp* pUseDbRsp = (SUseDbRsp*)pMsg->pData;
+  SName      name = {0};
+  tNameFromString(&name, pUseDbRsp->db, T_NAME_ACCT | T_NAME_DB);
 
   char db[TSDB_DB_NAME_LEN] = {0};
   tNameGetDbName(&name, db);
 
-  SRequestObj* pRequest = param;
   setConnectionDB(pRequest->pTscObj, db);
 
   tsem_post(&pRequest->body.rspSem);

source/client/test/clientTests.cpp

@@ -49,6 +49,7 @@ int main(int argc, char** argv) {
 
 TEST(testCase, driverInit_Test) { taos_init(); }
 
+#if 0
 TEST(testCase, connect_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
   assert(pConn != NULL);
@@ -398,6 +399,7 @@ TEST(testCase, drop_stable_Test) {
   taos_free_result(pRes);
   taos_close(pConn);
 }
+#endif
 
 //TEST(testCase, create_topic_Test) {
 //  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);

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

@@ -139,7 +139,7 @@ void dmnWaitSignal() {
 void dmnInitOption(SDnodeOpt *pOption) {
   pOption->sver = 30000000; //3.0.0.0
   pOption->numOfCores = tsNumOfCores;
-  pOption->numOfSupportVnodes = 1;
+  pOption->numOfSupportVnodes = 16;
   pOption->numOfCommitThreads = 1;
   pOption->statusInterval = tsStatusInterval;
   pOption->numOfThreadsPerCore = tsNumOfThreadsPerCore;

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

@@ -393,13 +393,11 @@ void dndSendStatusMsg(SDnode *pDnode) {
 
 static void dndUpdateDnodeCfg(SDnode *pDnode, SDnodeCfg *pCfg) {
   SDnodeMgmt *pMgmt = &pDnode->dmgmt;
-  if (pMgmt->dnodeId == 0 || pMgmt->dropped != pCfg->dropped) {
-    dInfo("set dnodeId:%d clusterId:% " PRId64 " dropped:%d", pCfg->dnodeId, pCfg->clusterId, pCfg->dropped);
-
+  if (pMgmt->dnodeId == 0) {
+    dInfo("set dnodeId:%d clusterId:% " PRId64, pCfg->dnodeId, pCfg->clusterId);
     taosWLockLatch(&pMgmt->latch);
     pMgmt->dnodeId = pCfg->dnodeId;
     pMgmt->clusterId = pCfg->clusterId;
-    pMgmt->dropped = pCfg->dropped;
     dndWriteDnodes(pDnode);
     taosWUnLockLatch(&pMgmt->latch);
   }
@@ -430,6 +428,11 @@ static void dndProcessStatusRsp(SDnode *pDnode, SRpcMsg *pMsg) {
 
   if (pMsg->code != TSDB_CODE_SUCCESS) {
     pMgmt->statusSent = 0;
+    if (pMsg->code == TSDB_CODE_MND_DNODE_NOT_EXIST && !pMgmt->dropped && pMgmt->dnodeId > 0) {
+      dInfo("dnode:%d, set to dropped since not exist in mnode", pMgmt->dnodeId);
+      pMgmt->dropped = 1;
+      dndWriteDnodes(pDnode);
+    }
     return;
   }
 
@@ -439,11 +442,6 @@ static void dndProcessStatusRsp(SDnode *pDnode, SRpcMsg *pMsg) {
   pCfg->clusterId = htobe64(pCfg->clusterId);
   dndUpdateDnodeCfg(pDnode, pCfg);
 
-  if (pCfg->dropped) {
-    pMgmt->statusSent = 0;
-    return;
-  }
-
   SDnodeEps *pDnodeEps = &pRsp->dnodeEps;
   pDnodeEps->num = htonl(pDnodeEps->num);
   for (int32_t i = 0; i < pDnodeEps->num; ++i) {
@@ -487,7 +485,7 @@ static void *dnodeThreadRoutine(void *param) {
     pthread_testcancel();
     taosMsleep(ms);
 
-    if (dndGetStat(pDnode) == DND_STAT_RUNNING && !pMgmt->statusSent) {
+    if (dndGetStat(pDnode) == DND_STAT_RUNNING && !pMgmt->statusSent && !pMgmt->dropped) {
       dndSendStatusMsg(pDnode);
     }
   }
@@ -522,6 +520,11 @@ int32_t dndInitDnode(SDnode *pDnode) {
     return -1;
   }
 
+  if (pMgmt->dropped) {
+    dError("dnode will not start for its already dropped");
+    return -1;
+  }
+
   if (dndInitMgmtWorker(pDnode) != 0) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;

source/dnode/mnode/impl/inc/mndConsumer.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_MND_CONSUMER_H_
+#define _TD_MND_CONSUMER_H_
+
+#include "mndInt.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+int32_t mndInitConsumer(SMnode *pMnode);
+void    mndCleanupConsumer(SMnode *pMnode);
+
+SConsumerObj *mndAcquireConsumer(SMnode *pMnode, int32_t consumerId);
+void          mndReleaseConsumer(SMnode *pMnode, SConsumerObj *pConsumer);
+
+SCGroupObj *mndAcquireCGroup(SMnode *pMnode, char *consumerGroup);
+void        mndReleaseCGroup(SMnode *pMnode, SCGroupObj *pCGroup);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_MND_CONSUMER_H_*/

source/dnode/mnode/impl/inc/mndDef.h

@@ -124,12 +124,8 @@ typedef struct {
   int64_t    rebootTime;
   int64_t    lastAccessTime;
   int32_t    accessTimes;
-  int16_t    numOfMnodes;
   int16_t    numOfVnodes;
-  int16_t    numOfQnodes;
-  int16_t    numOfSupportMnodes;
   int16_t    numOfSupportVnodes;
-  int16_t    numOfSupportQnodes;
   int16_t    numOfCores;
   EDndStatus status;
   EDndReason offlineReason;
@@ -305,8 +301,34 @@ typedef struct {
   void*    executor;
   int32_t  sqlLen;
   char*    sql;
+  char*    logicalPlan;
+  char*    physicalPlan;
 } STopicObj;
 
+typedef struct {
+  char name[TSDB_TOPIC_FNAME_LEN];
+  char db[TSDB_DB_FNAME_LEN];
+  int64_t createTime;
+  int64_t updateTime;
+  uint64_t uid;
+  //uint64_t dbUid;
+  int32_t version;
+  SRWLatch lock;
+
+} SConsumerObj;
+
+typedef struct {
+  char name[TSDB_TOPIC_FNAME_LEN];
+  char db[TSDB_DB_FNAME_LEN];
+  int64_t createTime;
+  int64_t updateTime;
+  uint64_t uid;
+  //uint64_t dbUid;
+  int32_t version;
+  SRWLatch lock;
+
+} SCGroupObj;
+
 typedef struct SMnodeMsg {
   char    user[TSDB_USER_LEN];
   char    db[TSDB_DB_FNAME_LEN];

source/dnode/mnode/impl/inc/mndInt.h

@@ -19,6 +19,7 @@
 #include "mndDef.h"
 #include "sdb.h"
 #include "tcache.h"
+#include "tep.h"
 #include "tqueue.h"
 #include "ttime.h"
 

source/dnode/mnode/impl/inc/mndVgroup.h

@@ -29,6 +29,7 @@ void     mndReleaseVgroup(SMnode *pMnode, SVgObj *pVgroup);
 SSdbRaw *mndVgroupActionEncode(SVgObj *pVgroup);
 int32_t  mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups);
 SEpSet   mndGetVgroupEpset(SMnode *pMnode, SVgObj *pVgroup);
+int32_t  mndGetVnodesNum(SMnode *pMnode, int32_t dnodeId);
 
 SCreateVnodeMsg *mndBuildCreateVnodeMsg(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup);
 SDropVnodeMsg   *mndBuildDropVnodeMsg(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup);

source/dnode/mnode/impl/src/mndConsumer.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/>.
+ */
+
+#define _DEFAULT_SOURCE
+#include "mndConsumer.h"
+#include "mndDb.h"
+#include "mndDnode.h"
+#include "mndMnode.h"
+#include "mndShow.h"
+#include "mndStb.h"
+#include "mndTopic.h"
+#include "mndTrans.h"
+#include "mndUser.h"
+#include "mndVgroup.h"
+#include "tname.h"
+
+#define MND_CONSUMER_VER_NUMBER 1
+#define MND_CONSUMER_RESERVE_SIZE 64
+
+static SSdbRaw *mndConsumerActionEncode(SConsumerObj *pConsumer);
+static SSdbRow *mndConsumerActionDecode(SSdbRaw *pRaw);
+static int32_t  mndConsumerActionInsert(SSdb *pSdb, SConsumerObj *pConsumer);
+static int32_t  mndConsumerActionDelete(SSdb *pSdb, SConsumerObj *pConsumer);
+static int32_t  mndConsumerActionUpdate(SSdb *pSdb, SConsumerObj *pConsumer, SConsumerObj *pNewConsumer);
+static int32_t  mndProcessCreateConsumerMsg(SMnodeMsg *pMsg);
+static int32_t  mndProcessDropConsumerMsg(SMnodeMsg *pMsg);
+static int32_t  mndProcessDropConsumerInRsp(SMnodeMsg *pMsg);
+static int32_t  mndProcessConsumerMetaMsg(SMnodeMsg *pMsg);
+static int32_t  mndGetConsumerMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMsg *pMeta);
+static int32_t  mndRetrieveConsumer(SMnodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows);
+static void     mndCancelGetNextConsumer(SMnode *pMnode, void *pIter);
+
+static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg);
+static int32_t mndProcessSubscribeRsp(SMnodeMsg *pMsg);
+static int32_t mndProcessSubscribeInternalReq(SMnodeMsg *pMsg);
+static int32_t mndProcessSubscribeInternalRsp(SMnodeMsg *pMsg);
+
+int32_t mndInitConsumer(SMnode *pMnode) {
+  SSdbTable table = {.sdbType = SDB_CONSUMER,
+                     .keyType = SDB_KEY_BINARY,
+                     .encodeFp = (SdbEncodeFp)mndConsumerActionEncode,
+                     .decodeFp = (SdbDecodeFp)mndConsumerActionDecode,
+                     .insertFp = (SdbInsertFp)mndConsumerActionInsert,
+                     .updateFp = (SdbUpdateFp)mndConsumerActionUpdate,
+                     .deleteFp = (SdbDeleteFp)mndConsumerActionDelete};
+
+  mndSetMsgHandle(pMnode, TDMT_MND_SUBSCRIBE, mndProcessSubscribeReq);
+  mndSetMsgHandle(pMnode, TDMT_MND_SUBSCRIBE_RSP, mndProcessSubscribeRsp);
+  mndSetMsgHandle(pMnode, TDMT_VND_SUBSCRIBE, mndProcessSubscribeInternalReq);
+  mndSetMsgHandle(pMnode, TDMT_VND_SUBSCRIBE_RSP, mndProcessSubscribeInternalRsp);
+
+  return sdbSetTable(pMnode->pSdb, table);
+}
+
+void mndCleanupConsumer(SMnode *pMnode) {}
+
+static SSdbRaw *mndConsumerActionEncode(SConsumerObj *pConsumer) {
+  int32_t  size = sizeof(SConsumerObj) + MND_CONSUMER_RESERVE_SIZE;
+  SSdbRaw *pRaw = sdbAllocRaw(SDB_CONSUMER, MND_CONSUMER_VER_NUMBER, size);
+  if (pRaw == NULL) return NULL;
+
+  int32_t dataPos = 0;
+  SDB_SET_BINARY(pRaw, dataPos, pConsumer->name, TSDB_TABLE_FNAME_LEN);
+  SDB_SET_BINARY(pRaw, dataPos, pConsumer->db, TSDB_DB_FNAME_LEN);
+  SDB_SET_INT64(pRaw, dataPos, pConsumer->createTime);
+  SDB_SET_INT64(pRaw, dataPos, pConsumer->updateTime);
+  SDB_SET_INT64(pRaw, dataPos, pConsumer->uid);
+  /*SDB_SET_INT64(pRaw, dataPos, pConsumer->dbUid);*/
+  SDB_SET_INT32(pRaw, dataPos, pConsumer->version);
+
+  SDB_SET_RESERVE(pRaw, dataPos, MND_CONSUMER_RESERVE_SIZE);
+  SDB_SET_DATALEN(pRaw, dataPos);
+
+  return pRaw;
+}
+
+static SSdbRow *mndConsumerActionDecode(SSdbRaw *pRaw) {
+  int8_t sver = 0;
+  if (sdbGetRawSoftVer(pRaw, &sver) != 0) return NULL;
+
+  if (sver != MND_CONSUMER_VER_NUMBER) {
+    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
+    mError("failed to decode consumer since %s", terrstr());
+    return NULL;
+  }
+
+  int32_t       size = sizeof(SConsumerObj) + TSDB_MAX_COLUMNS * sizeof(SSchema);
+  SSdbRow      *pRow = sdbAllocRow(size);
+  SConsumerObj *pConsumer = sdbGetRowObj(pRow);
+  if (pConsumer == NULL) return NULL;
+
+  int32_t dataPos = 0;
+  SDB_GET_BINARY(pRaw, pRow, dataPos, pConsumer->name, TSDB_TABLE_FNAME_LEN);
+  SDB_GET_BINARY(pRaw, pRow, dataPos, pConsumer->db, TSDB_DB_FNAME_LEN);
+  SDB_GET_INT64(pRaw, pRow, dataPos, &pConsumer->createTime);
+  SDB_GET_INT64(pRaw, pRow, dataPos, &pConsumer->updateTime);
+  SDB_GET_INT64(pRaw, pRow, dataPos, &pConsumer->uid);
+  /*SDB_GET_INT64(pRaw, pRow, dataPos, &pConsumer->dbUid);*/
+  SDB_GET_INT32(pRaw, pRow, dataPos, &pConsumer->version);
+
+  SDB_GET_RESERVE(pRaw, pRow, dataPos, MND_CONSUMER_RESERVE_SIZE);
+
+  return pRow;
+}
+
+static int32_t mndConsumerActionInsert(SSdb *pSdb, SConsumerObj *pConsumer) {
+  mTrace("consumer:%s, perform insert action", pConsumer->name);
+  return 0;
+}
+
+static int32_t mndConsumerActionDelete(SSdb *pSdb, SConsumerObj *pConsumer) {
+  mTrace("consumer:%s, perform delete action", pConsumer->name);
+  return 0;
+}
+
+static int32_t mndConsumerActionUpdate(SSdb *pSdb, SConsumerObj *pOldConsumer, SConsumerObj *pNewConsumer) {
+  mTrace("consumer:%s, perform update action", pOldConsumer->name);
+  atomic_exchange_32(&pOldConsumer->updateTime, pNewConsumer->updateTime);
+  atomic_exchange_32(&pOldConsumer->version, pNewConsumer->version);
+
+  taosWLockLatch(&pOldConsumer->lock);
+
+  // TODO handle update
+
+  taosWUnLockLatch(&pOldConsumer->lock);
+  return 0;
+}
+
+SConsumerObj *mndAcquireConsumer(SMnode *pMnode, int32_t consumerId) {
+  SSdb         *pSdb = pMnode->pSdb;
+  SConsumerObj *pConsumer = sdbAcquire(pSdb, SDB_CONSUMER, &consumerId);
+  if (pConsumer == NULL) {
+    /*terrno = TSDB_CODE_MND_CONSUMER_NOT_EXIST;*/
+  }
+  return pConsumer;
+}
+
+void mndReleaseConsumer(SMnode *pMnode, SConsumerObj *pConsumer) {
+  SSdb *pSdb = pMnode->pSdb;
+  sdbRelease(pSdb, pConsumer);
+}
+
+static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) {
+  SMnode          *pMnode = pMsg->pMnode;
+  char            *msgStr = pMsg->rpcMsg.pCont;
+  SCMSubscribeReq *pSubscribe;
+  tDeserializeSCMSubscribeReq(msgStr, pSubscribe);
+  // add consumerGroupId -> list<consumerId> to sdb
+  // add consumerId -> list<consumer> to sdb
+  // add consumer -> list<consumerId> to sdb
+  return 0;
+}
+
+static int32_t mndProcessSubscribeRsp(SMnodeMsg *pMsg) { return 0; }
+
+static int32_t mndProcessSubscribeInternalReq(SMnodeMsg *pMsg) { return 0; }
+
+static int32_t mndProcessSubscribeInternalRsp(SMnodeMsg *pMsg) { return 0; }
+
+static int32_t mndProcessDropConsumerInRsp(SMnodeMsg *pMsg) {
+  mndTransProcessRsp(pMsg);
+  return 0;
+}
+
+static int32_t mndProcessConsumerMetaMsg(SMnodeMsg *pMsg) {
+  SMnode        *pMnode = pMsg->pMnode;
+  STableInfoMsg *pInfo = pMsg->rpcMsg.pCont;
+
+  mDebug("consumer:%s, start to retrieve meta", pInfo->tableFname);
+
+#if 0
+  SDbObj *pDb = mndAcquireDbByConsumer(pMnode, pInfo->tableFname);
+  if (pDb == NULL) {
+    terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
+    mError("consumer:%s, failed to retrieve meta since %s", pInfo->tableFname, terrstr());
+    return -1;
+  }
+
+  SConsumerObj *pConsumer = mndAcquireConsumer(pMnode, pInfo->tableFname);
+  if (pConsumer == NULL) {
+    mndReleaseDb(pMnode, pDb);
+    terrno = TSDB_CODE_MND_INVALID_CONSUMER;
+    mError("consumer:%s, failed to get meta since %s", pInfo->tableFname, terrstr());
+    return -1;
+  }
+
+  taosRLockLatch(&pConsumer->lock);
+  int32_t totalCols = pConsumer->numOfColumns + pConsumer->numOfTags;
+  int32_t contLen = sizeof(STableMetaMsg) + totalCols * sizeof(SSchema);
+
+  STableMetaMsg *pMeta = rpcMallocCont(contLen);
+  if (pMeta == NULL) {
+    taosRUnLockLatch(&pConsumer->lock);
+    mndReleaseDb(pMnode, pDb);
+    mndReleaseConsumer(pMnode, pConsumer);
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    mError("consumer:%s, failed to get meta since %s", pInfo->tableFname, terrstr());
+    return -1;
+  }
+
+  memcpy(pMeta->consumerFname, pConsumer->name, TSDB_TABLE_FNAME_LEN);
+  pMeta->numOfTags = htonl(pConsumer->numOfTags);
+  pMeta->numOfColumns = htonl(pConsumer->numOfColumns);
+  pMeta->precision = pDb->cfg.precision;
+  pMeta->tableType = TSDB_SUPER_TABLE;
+  pMeta->update = pDb->cfg.update;
+  pMeta->sversion = htonl(pConsumer->version);
+  pMeta->tuid = htonl(pConsumer->uid);
+
+  for (int32_t i = 0; i < totalCols; ++i) {
+    SSchema *pSchema = &pMeta->pSchema[i];
+    SSchema *pSrcSchema = &pConsumer->pSchema[i];
+    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
+    pSchema->type = pSrcSchema->type;
+    pSchema->colId = htonl(pSrcSchema->colId);
+    pSchema->bytes = htonl(pSrcSchema->bytes);
+  }
+  taosRUnLockLatch(&pConsumer->lock);
+  mndReleaseDb(pMnode, pDb);
+  mndReleaseConsumer(pMnode, pConsumer);
+
+  pMsg->pCont = pMeta;
+  pMsg->contLen = contLen;
+
+  mDebug("consumer:%s, meta is retrieved, cols:%d tags:%d", pInfo->tableFname, pConsumer->numOfColumns, pConsumer->numOfTags);
+#endif
+  return 0;
+}
+
+static int32_t mndGetNumOfConsumers(SMnode *pMnode, char *dbName, int32_t *pNumOfConsumers) {
+  SSdb *pSdb = pMnode->pSdb;
+
+  SDbObj *pDb = mndAcquireDb(pMnode, dbName);
+  if (pDb == NULL) {
+    terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
+    return -1;
+  }
+
+  int32_t numOfConsumers = 0;
+  void   *pIter = NULL;
+  while (1) {
+    SConsumerObj *pConsumer = NULL;
+    pIter = sdbFetch(pSdb, SDB_CONSUMER, pIter, (void **)&pConsumer);
+    if (pIter == NULL) break;
+
+    if (strcmp(pConsumer->db, dbName) == 0) {
+      numOfConsumers++;
+    }
+
+    sdbRelease(pSdb, pConsumer);
+  }
+
+  *pNumOfConsumers = numOfConsumers;
+  return 0;
+}
+
+static int32_t mndGetConsumerMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMsg *pMeta) {
+  SMnode *pMnode = pMsg->pMnode;
+  SSdb   *pSdb = pMnode->pSdb;
+
+  if (mndGetNumOfConsumers(pMnode, pShow->db, &pShow->numOfRows) != 0) {
+    return -1;
+  }
+
+  int32_t  cols = 0;
+  SSchema *pSchema = pMeta->pSchema;
+
+  pShow->bytes[cols] = TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE;
+  pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
+  strcpy(pSchema[cols].name, "name");
+  pSchema[cols].bytes = htonl(pShow->bytes[cols]);
+  cols++;
+
+  pShow->bytes[cols] = 8;
+  pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
+  strcpy(pSchema[cols].name, "create_time");
+  pSchema[cols].bytes = htonl(pShow->bytes[cols]);
+  cols++;
+
+  pShow->bytes[cols] = 4;
+  pSchema[cols].type = TSDB_DATA_TYPE_INT;
+  strcpy(pSchema[cols].name, "columns");
+  pSchema[cols].bytes = htonl(pShow->bytes[cols]);
+  cols++;
+
+  pShow->bytes[cols] = 4;
+  pSchema[cols].type = TSDB_DATA_TYPE_INT;
+  strcpy(pSchema[cols].name, "tags");
+  pSchema[cols].bytes = htonl(pShow->bytes[cols]);
+  cols++;
+
+  pMeta->numOfColumns = htonl(cols);
+  pShow->numOfColumns = cols;
+
+  pShow->offset[0] = 0;
+  for (int32_t i = 1; i < cols; ++i) {
+    pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
+  }
+
+  pShow->numOfRows = sdbGetSize(pSdb, SDB_CONSUMER);
+  pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
+  strcpy(pMeta->tbFname, mndShowStr(pShow->type));
+
+  return 0;
+}
+
+static int32_t mndRetrieveConsumer(SMnodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows) {
+  SMnode       *pMnode = pMsg->pMnode;
+  SSdb         *pSdb = pMnode->pSdb;
+  int32_t       numOfRows = 0;
+  SConsumerObj *pConsumer = NULL;
+  int32_t       cols = 0;
+  char         *pWrite;
+  char          prefix[64] = {0};
+
+  tstrncpy(prefix, pShow->db, 64);
+  strcat(prefix, TS_PATH_DELIMITER);
+  int32_t prefixLen = (int32_t)strlen(prefix);
+
+  while (numOfRows < rows) {
+    pShow->pIter = sdbFetch(pSdb, SDB_CONSUMER, pShow->pIter, (void **)&pConsumer);
+    if (pShow->pIter == NULL) break;
+
+    if (strncmp(pConsumer->name, prefix, prefixLen) != 0) {
+      sdbRelease(pSdb, pConsumer);
+      continue;
+    }
+
+    cols = 0;
+
+    char consumerName[TSDB_TABLE_NAME_LEN] = {0};
+    tstrncpy(consumerName, pConsumer->name + prefixLen, TSDB_TABLE_NAME_LEN);
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    STR_TO_VARSTR(pWrite, consumerName);
+    cols++;
+
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    *(int64_t *)pWrite = pConsumer->createTime;
+    cols++;
+
+    /*pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;*/
+    /**(int32_t *)pWrite = pConsumer->numOfColumns;*/
+    /*cols++;*/
+
+    /*pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;*/
+    /**(int32_t *)pWrite = pConsumer->numOfTags;*/
+    /*cols++;*/
+
+    numOfRows++;
+    sdbRelease(pSdb, pConsumer);
+  }
+
+  pShow->numOfReads += numOfRows;
+  mndVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
+  return numOfRows;
+}
+
+static void mndCancelGetNextConsumer(SMnode *pMnode, void *pIter) {
+  SSdb *pSdb = pMnode->pSdb;
+  sdbCancelFetch(pSdb, pIter);
+}

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

@@ -18,8 +18,7 @@
 #include "mndMnode.h"
 #include "mndShow.h"
 #include "mndTrans.h"
-#include "tep.h"
-#include "ttime.h"
+#include "mndVgroup.h"
 
 #define TSDB_DNODE_VER_NUMBER 1
 #define TSDB_DNODE_RESERVE_SIZE 64
@@ -370,7 +369,6 @@ static int32_t mndProcessStatusMsg(SMnodeMsg *pMsg) {
   }
 
   pRsp->dnodeCfg.dnodeId = htonl(pDnode->id);
-  pRsp->dnodeCfg.dropped = 0;
   pRsp->dnodeCfg.clusterId = htobe64(pMnode->clusterId);
   mndGetDnodeData(pMnode, &pRsp->dnodeEps, numOfEps);
 
@@ -700,7 +698,7 @@ static int32_t mndRetrieveDnodes(SMnodeMsg *pMsg, SShowObj *pShow, char *data, i
     cols++;
 
     pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
-    *(int16_t *)pWrite = pDnode->numOfVnodes;
+    *(int16_t *)pWrite = mndGetVnodesNum(pMnode, pDnode->id);
     cols++;
 
     pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;

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

@@ -118,17 +118,17 @@ static SConnObj *mndCreateConn(SMnode *pMnode, SRpcConnInfo *pInfo, int32_t pid,
   SConnObj *pConn = taosCachePut(pMgmt->cache, &connId, sizeof(int32_t), &connObj, sizeof(connObj), keepTime * 1000);
   if (pConn == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
-    mError("conn:%d, data:%p failed to put into cache since %s, user:%s", connId, pConn, pInfo->user, terrstr());
+    mError("conn:%d, failed to put into cache since %s, user:%s", connId, pInfo->user, terrstr());
     return NULL;
   } else {
-    mTrace("conn:%d, data:%p created, user:%s", pConn->id, pConn, pInfo->user);
+    mTrace("conn:%d, is created, data:%p user:%s", pConn->id, pConn, pInfo->user);
     return pConn;
   }
 }
 
 static void mndFreeConn(SConnObj *pConn) {
   tfree(pConn->pQueries);
-  mTrace("conn:%d, data:%p destroyed", pConn->id, pConn);
+  mTrace("conn:%d, is destroyed, data:%p", pConn->id, pConn);
 }
 
 static SConnObj *mndAcquireConn(SMnode *pMnode, int32_t connId) {
@@ -143,13 +143,13 @@ static SConnObj *mndAcquireConn(SMnode *pMnode, int32_t connId) {
   int32_t keepTime = pMnode->cfg.shellActivityTimer * 3;
   pConn->lastAccessTimeMs = keepTime * 1000 + (uint64_t)taosGetTimestampMs();
 
-  mTrace("conn:%d, data:%p acquired from cache", pConn->id, pConn);
+  mTrace("conn:%d, acquired from cache, data:%p", pConn->id, pConn);
   return pConn;
 }
 
 static void mndReleaseConn(SMnode *pMnode, SConnObj *pConn) {
   if (pConn == NULL) return;
-  mTrace("conn:%d, data:%p released from cache", pConn->id, pConn);
+  mTrace("conn:%d, released from cache, data:%p", pConn->id, pConn);
 
   SProfileMgmt *pMgmt = &pMnode->profileMgmt;
   taosCacheRelease(pMgmt->cache, (void **)&pConn, false);

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

@@ -55,32 +55,25 @@ static SShowObj *mndCreateShowObj(SMnode *pMnode, SShowMsg *pMsg) {
   int32_t showId = atomic_add_fetch_32(&pMgmt->showId, 1);
   if (showId == 0) atomic_add_fetch_32(&pMgmt->showId, 1);
 
-  int32_t   size = sizeof(SShowObj) + pMsg->payloadLen;
-  SShowObj *pShow = calloc(1, size);
-  if (pShow != NULL) {
-    pShow->id = showId;
-    pShow->pMnode = pMnode;
-    pShow->type = pMsg->type;
-    pShow->payloadLen = pMsg->payloadLen;
-    memcpy(pShow->db, pMsg->db, TSDB_DB_FNAME_LEN);
-    memcpy(pShow->payload, pMsg->payload, pMsg->payloadLen);
-  } else {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    mError("failed to process show-meta msg:%s since %s", mndShowStr(pMsg->type), terrstr());
-    return NULL;
-  }
+  int32_t  size = sizeof(SShowObj) + pMsg->payloadLen;
+  SShowObj showObj = {0};
+  showObj.id = showId;
+  showObj.pMnode = pMnode;
+  showObj.type = pMsg->type;
+  showObj.payloadLen = pMsg->payloadLen;
+  memcpy(showObj.db, pMsg->db, TSDB_DB_FNAME_LEN);
+  memcpy(showObj.payload, pMsg->payload, pMsg->payloadLen);
 
   int32_t   keepTime = pMnode->cfg.shellActivityTimer * 6 * 1000;
-  SShowObj *pShowRet = taosCachePut(pMgmt->cache, &showId, sizeof(int32_t), pShow, size, keepTime);
-  free(pShow);
-  if (pShowRet == NULL) {
+  SShowObj *pShow = taosCachePut(pMgmt->cache, &showId, sizeof(int32_t), &showObj, size, keepTime);
+  if (pShow == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     mError("show:%d, failed to put into cache since %s", showId, terrstr());
     return NULL;
-  } else {
-    mTrace("show:%d, data:%p created", showId, pShowRet);
-    return pShowRet;
   }
+
+  mTrace("show:%d, is created, data:%p", showId, pShow);
+  return pShow;
 }
 
 static void mndFreeShowObj(SShowObj *pShow) {
@@ -94,7 +87,7 @@ static void mndFreeShowObj(SShowObj *pShow) {
     }
   }
 
-  mTrace("show:%d, data:%p destroyed", pShow->id, pShow);
+  mTrace("show:%d, is destroyed, data:%p", pShow->id, pShow);
 }
 
 static SShowObj *mndAcquireShowObj(SMnode *pMnode, int32_t showId) {
@@ -106,14 +99,14 @@ static SShowObj *mndAcquireShowObj(SMnode *pMnode, int32_t showId) {
     return NULL;
   }
 
-  mTrace("show:%d, data:%p acquired from cache", pShow->id, pShow);
+  mTrace("show:%d, acquired from cache, data:%p", pShow->id, pShow);
   return pShow;
 }
 
 static void mndReleaseShowObj(SShowObj *pShow, bool forceRemove) {
   if (pShow == NULL) return;
-  mTrace("show:%d, data:%p released from cache, force:%d", pShow->id, pShow, forceRemove);
-  
+  mTrace("show:%d, released from cache, data:%p force:%d", pShow->id, pShow, forceRemove);
+
   // A bug in tcache.c
   forceRemove = 0;
 
@@ -158,8 +151,8 @@ static int32_t mndProcessShowMsg(SMnodeMsg *pMnodeMsg) {
   }
 
   int32_t code = (*metaFp)(pMnodeMsg, pShow, &pRsp->tableMeta);
-  mDebug("show:%d, data:%p get meta finished, numOfRows:%d cols:%d type:%s result:%s", pShow->id, pShow,
-         pShow->numOfRows, pShow->numOfColumns, mndShowStr(type), tstrerror(code));
+  mDebug("show:%d, get meta finished, numOfRows:%d cols:%d type:%s result:%s", pShow->id, pShow->numOfRows,
+         pShow->numOfColumns, mndShowStr(type), tstrerror(code));
 
   if (code == TSDB_CODE_SUCCESS) {
     pMnodeMsg->contLen = sizeof(SShowRsp) + sizeof(SSchema) * pShow->numOfColumns;
@@ -195,16 +188,15 @@ static int32_t mndProcessRetrieveMsg(SMnodeMsg *pMnodeMsg) {
   if (retrieveFp == NULL) {
     mndReleaseShowObj(pShow, false);
     terrno = TSDB_CODE_MSG_NOT_PROCESSED;
-    mError("show:%d, data:%p failed to retrieve data since %s", pShow->id, pShow, terrstr());
+    mError("show:%d, failed to retrieve data since %s", pShow->id, terrstr());
     return -1;
   }
 
-  mDebug("show:%d, data:%p start retrieve data, numOfReads:%d numOfRows:%d type:%s", pShow->id, pShow,
-         pShow->numOfReads, pShow->numOfRows, mndShowStr(pShow->type));
+  mDebug("show:%d, start retrieve data, numOfReads:%d numOfRows:%d type:%s", pShow->id, pShow->numOfReads,
+         pShow->numOfRows, mndShowStr(pShow->type));
 
   if (mndCheckRetrieveFinished(pShow)) {
-    mDebug("show:%d, data:%p read finished, numOfReads:%d numOfRows:%d", pShow->id, pShow, pShow->numOfReads,
-           pShow->numOfRows);
+    mDebug("show:%d, read finished, numOfReads:%d numOfRows:%d", pShow->id, pShow->numOfReads, pShow->numOfRows);
     pShow->numOfReads = pShow->numOfRows;
   }
 
@@ -227,7 +219,7 @@ static int32_t mndProcessRetrieveMsg(SMnodeMsg *pMnodeMsg) {
   if (pRsp == NULL) {
     mndReleaseShowObj(pShow, false);
     terrno = TSDB_CODE_OUT_OF_MEMORY;
-    mError("show:%d, data:%p failed to retrieve data since %s", pShow->id, pShow, terrstr());
+    mError("show:%d, failed to retrieve data since %s", pShow->id, terrstr());
     return -1;
   }
 
@@ -236,7 +228,7 @@ static int32_t mndProcessRetrieveMsg(SMnodeMsg *pMnodeMsg) {
     rowsRead = (*retrieveFp)(pMnodeMsg, pShow, pRsp->data, rowsToRead);
   }
 
-  mDebug("show:%d, data:%p stop retrieve data, rowsRead:%d rowsToRead:%d", pShow->id, pShow, rowsRead, rowsToRead);
+  mDebug("show:%d, stop retrieve data, rowsRead:%d rowsToRead:%d", pShow->id, rowsRead, rowsToRead);
 
   pRsp->numOfRows = htonl(rowsRead);
   pRsp->precision = TSDB_TIME_PRECISION_MILLI;  // millisecond time precision
@@ -246,10 +238,10 @@ static int32_t mndProcessRetrieveMsg(SMnodeMsg *pMnodeMsg) {
 
   if (rowsRead == 0 || rowsToRead == 0 || (rowsRead == rowsToRead && pShow->numOfRows == pShow->numOfReads)) {
     pRsp->completed = 1;
-    mDebug("show:%d, data:%p retrieve completed", pShow->id, pShow);
+    mDebug("show:%d, retrieve completed", pShow->id);
     mndReleaseShowObj(pShow, true);
   } else {
-    mDebug("show:%d, data:%p retrieve not completed yet", pShow->id, pShow);
+    mDebug("show:%d, retrieve not completed yet", pShow->id);
     mndReleaseShowObj(pShow, false);
   }
 

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

@@ -33,10 +33,7 @@ static int32_t  mndTopicActionInsert(SSdb *pSdb, STopicObj *pTopic);
 static int32_t  mndTopicActionDelete(SSdb *pSdb, STopicObj *pTopic);
 static int32_t  mndTopicActionUpdate(SSdb *pSdb, STopicObj *pTopic, STopicObj *pNewTopic);
 static int32_t  mndProcessCreateTopicMsg(SMnodeMsg *pMsg);
-static int32_t  mndProcessAlterTopicMsg(SMnodeMsg *pMsg);
 static int32_t  mndProcessDropTopicMsg(SMnodeMsg *pMsg);
-static int32_t  mndProcessCreateTopicInRsp(SMnodeMsg *pMsg);
-static int32_t  mndProcessAlterTopicInRsp(SMnodeMsg *pMsg);
 static int32_t  mndProcessDropTopicInRsp(SMnodeMsg *pMsg);
 static int32_t  mndProcessTopicMetaMsg(SMnodeMsg *pMsg);
 static int32_t  mndGetTopicMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMsg *pMeta);
@@ -53,19 +50,8 @@ int32_t mndInitTopic(SMnode *pMnode) {
                      .deleteFp = (SdbDeleteFp)mndTopicActionDelete};
 
   mndSetMsgHandle(pMnode, TDMT_MND_CREATE_TOPIC, mndProcessCreateTopicMsg);
-  mndSetMsgHandle(pMnode, TDMT_MND_ALTER_TOPIC, mndProcessAlterTopicMsg);
   mndSetMsgHandle(pMnode, TDMT_MND_DROP_TOPIC, mndProcessDropTopicMsg);
-  mndSetMsgHandle(pMnode, TDMT_VND_CREATE_TOPIC_RSP, mndProcessCreateTopicInRsp);
-  mndSetMsgHandle(pMnode, TDMT_VND_ALTER_TOPIC_RSP, mndProcessAlterTopicInRsp);
   mndSetMsgHandle(pMnode, TDMT_VND_DROP_TOPIC_RSP, mndProcessDropTopicInRsp);
-  mndSetMsgHandle(pMnode, TDMT_VND_TABLE_META, mndProcessTopicMetaMsg);
-
-  /*mndAddShowMetaHandle(pMnode, TSDB_MGMT_TOPIC, mndGetTopicMeta);*/
-  /*mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TOPIC, mndRetrieveTopic);*/
-  /*mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TOPIC, mndCancelGetNextTopic);*/
-
-  mndSetMsgHandle(pMnode, TDMT_MND_CREATE_TOPIC, mndProcessCreateTopicMsg);
-  mndSetMsgHandle(pMnode, TDMT_VND_CREATE_TOPIC_RSP, mndProcessCreateTopicInRsp);
 
   return sdbSetTable(pMnode->pSdb, table);
 }
@@ -145,24 +131,9 @@ static int32_t mndTopicActionUpdate(SSdb *pSdb, STopicObj *pOldTopic, STopicObj
   atomic_exchange_32(&pOldTopic->version, pNewTopic->version);
 
   taosWLockLatch(&pOldTopic->lock);
-#if 0
-
-  pOldTopic->numOfColumns = pNewTopic->numOfColumns;
-  pOldTopic->numOfTags = pNewTopic->numOfTags;
-  int32_t totalCols = pNewTopic->numOfTags + pNewTopic->numOfColumns;
-  int32_t totalSize = totalCols * sizeof(SSchema);
+  
+  //TODO handle update
 
-  if (pOldTopic->numOfTags + pOldTopic->numOfColumns < totalCols) {
-    void *pSchema = malloc(totalSize);
-    if (pSchema != NULL) {
-      free(pOldTopic->pSchema);
-      pOldTopic->pSchema = pSchema;
-    }
-  }
-
-  memcpy(pOldTopic->pSchema, pNewTopic->pSchema, totalSize);
-
-#endif
   taosWUnLockLatch(&pOldTopic->lock);
   return 0;
 }
@@ -191,41 +162,6 @@ static SDbObj *mndAcquireDbByTopic(SMnode *pMnode, char *topicName) {
   return mndAcquireDb(pMnode, db);
 }
 
-static SCreateTopicInternalMsg *mndBuildCreateTopicMsg(SMnode *pMnode, SVgObj *pVgroup, STopicObj *pTopic) {
-  int32_t totalCols = 0;
-  int32_t contLen = sizeof(SCreateTopicInternalMsg) + pTopic->execLen + pTopic->sqlLen;
-
-  SCreateTopicInternalMsg *pCreate = calloc(1, contLen);
-  if (pCreate == NULL) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return NULL;
-  }
-
-  pCreate->head.contLen = htonl(contLen);
-  pCreate->head.vgId = htonl(pVgroup->vgId);
-  memcpy(pCreate->name, pTopic->name, TSDB_TABLE_FNAME_LEN);
-  pCreate->tuid = htobe64(pTopic->uid);
-  pCreate->sverson = htonl(pTopic->version);
-
-  pCreate->sql = malloc(pTopic->sqlLen);
-  if (pCreate->sql == NULL) {
-    free(pCreate);
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return NULL;
-  }
-  memcpy(pCreate->sql, pTopic->sql, pTopic->sqlLen);
-
-  pCreate->executor = malloc(pTopic->execLen);
-  if (pCreate->executor == NULL) {
-    free(pCreate);
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return NULL;
-  }
-  memcpy(pCreate->executor, pTopic->executor, pTopic->execLen);
-
-  return pCreate;
-}
-
 static SDropTopicInternalMsg *mndBuildDropTopicMsg(SMnode *pMnode, SVgObj *pVgroup, STopicObj *pTopic) {
   int32_t contLen = sizeof(SDropTopicInternalMsg);
 
@@ -243,109 +179,12 @@ static SDropTopicInternalMsg *mndBuildDropTopicMsg(SMnode *pMnode, SVgObj *pVgro
   return pDrop;
 }
 
-static int32_t mndCheckCreateTopicMsg(SCreateTopicMsg *pCreate) {
+static int32_t mndCheckCreateTopicMsg(SCMCreateTopicReq *pCreate) {
   // deserialize and other stuff
   return 0;
 }
 
-static int32_t mndSetCreateTopicRedoLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, STopicObj *pTopic) {
-  SSdbRaw *pRedoRaw = mndTopicActionEncode(pTopic);
-  if (pRedoRaw == NULL) return -1;
-  if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1;
-  if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_CREATING) != 0) return -1;
-
-  return 0;
-}
-
-static int32_t mndSetCreateTopicUndoLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, STopicObj *pTopic) {
-  SSdbRaw *pUndoRaw = mndTopicActionEncode(pTopic);
-  if (pUndoRaw == NULL) return -1;
-  if (mndTransAppendUndolog(pTrans, pUndoRaw) != 0) return -1;
-  if (sdbSetRawStatus(pUndoRaw, SDB_STATUS_DROPPED) != 0) return -1;
-
-  return 0;
-}
-
-static int32_t mndSetCreateTopicCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, STopicObj *pTopic) {
-  SSdbRaw *pCommitRaw = mndTopicActionEncode(pTopic);
-  if (pCommitRaw == NULL) return -1;
-  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
-  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) return -1;
-
-  return 0;
-}
-
-static int32_t mndSetCreateTopicRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, STopicObj *pTopic) {
-  SSdb   *pSdb = pMnode->pSdb;
-  SVgObj *pVgroup = NULL;
-  void   *pIter = NULL;
-
-  while (1) {
-    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
-    if (pIter == NULL) break;
-    if (pVgroup->dbUid != pDb->uid) continue;
-
-    SCreateTopicInternalMsg *pMsg = mndBuildCreateTopicMsg(pMnode, pVgroup, pTopic);
-    if (pMsg == NULL) {
-      sdbCancelFetch(pSdb, pIter);
-      sdbRelease(pSdb, pVgroup);
-      terrno = TSDB_CODE_OUT_OF_MEMORY;
-      return -1;
-    }
-
-    STransAction action = {0};
-    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
-    action.pCont = pMsg;
-    action.contLen = htonl(pMsg->head.contLen);
-    action.msgType = TDMT_VND_CREATE_TOPIC;
-    if (mndTransAppendRedoAction(pTrans, &action) != 0) {
-      free(pMsg);
-      sdbCancelFetch(pSdb, pIter);
-      sdbRelease(pSdb, pVgroup);
-      return -1;
-    }
-    sdbRelease(pSdb, pVgroup);
-  }
-
-  return 0;
-}
-
-static int32_t mndSetCreateTopicUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, STopicObj *pTopic) {
-  SSdb   *pSdb = pMnode->pSdb;
-  SVgObj *pVgroup = NULL;
-  void   *pIter = NULL;
-
-  while (1) {
-    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
-    if (pIter == NULL) break;
-    if (pVgroup->dbUid != pDb->uid) continue;
-
-    SDropTopicInternalMsg *pMsg = mndBuildDropTopicMsg(pMnode, pVgroup, pTopic);
-    if (pMsg == NULL) {
-      sdbCancelFetch(pSdb, pIter);
-      sdbRelease(pSdb, pVgroup);
-      terrno = TSDB_CODE_OUT_OF_MEMORY;
-      return -1;
-    }
-
-    STransAction action = {0};
-    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
-    action.pCont = pMsg;
-    action.contLen = sizeof(SDropTopicInternalMsg);
-    action.msgType = TDMT_VND_DROP_TOPIC;
-    if (mndTransAppendUndoAction(pTrans, &action) != 0) {
-      free(pMsg);
-      sdbCancelFetch(pSdb, pIter);
-      sdbRelease(pSdb, pVgroup);
-      return -1;
-    }
-    sdbRelease(pSdb, pVgroup);
-  }
-
-  return 0;
-}
-
-static int32_t mndCreateTopic(SMnode *pMnode, SMnodeMsg *pMsg, SCreateTopicMsg *pCreate, SDbObj *pDb) {
+static int32_t mndCreateTopic(SMnode *pMnode, SMnodeMsg *pMsg, SCMCreateTopicReq *pCreate, SDbObj *pDb) {
   STopicObj topicObj = {0};
   tstrncpy(topicObj.name, pCreate->name, TSDB_TABLE_FNAME_LEN);
   tstrncpy(topicObj.db, pDb->name, TSDB_DB_FNAME_LEN);
@@ -355,66 +194,17 @@ static int32_t mndCreateTopic(SMnode *pMnode, SMnodeMsg *pMsg, SCreateTopicMsg *
   topicObj.dbUid = pDb->uid;
   topicObj.version = 1;
 
-#if 0
-  int32_t totalCols = topicObj.numOfColumns + topicObj.numOfTags;
-  int32_t totalSize = totalCols * sizeof(SSchema);
-  topicObj.sql = malloc(totalSize);
-  if (topicObj.sql == NULL) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return -1;
-  }
-  memcpy(topicObj.sql, pCreate->sql, totalSize);
-#endif
-
-  int32_t code = 0;
-  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, &pMsg->rpcMsg);
-  if (pTrans == NULL) {
-    mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
-    return -1;
-  }
-  mDebug("trans:%d, used to create topic:%s", pTrans->id, pCreate->name);
-
-  if (mndSetCreateTopicRedoLogs(pMnode, pTrans, pDb, &topicObj) != 0) {
-    mError("trans:%d, failed to set redo log since %s", pTrans->id, terrstr());
-    goto CREATE_TOPIC_OVER;
-  }
-
-  if (mndSetCreateTopicUndoLogs(pMnode, pTrans, pDb, &topicObj) != 0) {
-    mError("trans:%d, failed to set undo log since %s", pTrans->id, terrstr());
-    goto CREATE_TOPIC_OVER;
-  }
-
-  if (mndSetCreateTopicCommitLogs(pMnode, pTrans, pDb, &topicObj) != 0) {
-    mError("trans:%d, failed to set commit log since %s", pTrans->id, terrstr());
-    goto CREATE_TOPIC_OVER;
-  }
-
-  if (mndSetCreateTopicRedoActions(pMnode, pTrans, pDb, &topicObj) != 0) {
-    mError("trans:%d, failed to set redo actions since %s", pTrans->id, terrstr());
-    goto CREATE_TOPIC_OVER;
-  }
-
-  if (mndSetCreateTopicUndoActions(pMnode, pTrans, pDb, &topicObj) != 0) {
-    mError("trans:%d, failed to set redo actions since %s", pTrans->id, terrstr());
-    goto CREATE_TOPIC_OVER;
-  }
-
-  if (mndTransPrepare(pMnode, pTrans) != 0) {
-    mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
-    mndTransDrop(pTrans);
-    return -1;
-  }
-
-  code = 0;
-
-CREATE_TOPIC_OVER:
-  mndTransDrop(pTrans);
-  return code;
+  SSdbRaw *pTopicRaw = mndTopicActionEncode(&topicObj);
+  if (pTopicRaw == NULL) return -1;
+  if (sdbSetRawStatus(pTopicRaw, SDB_STATUS_READY) != 0) return -1;
+  return sdbWrite(pMnode->pSdb, pTopicRaw);
 }
 
 static int32_t mndProcessCreateTopicMsg(SMnodeMsg *pMsg) {
-  SMnode          *pMnode = pMsg->pMnode;
-  SCreateTopicMsg *pCreate = pMsg->rpcMsg.pCont;
+  SMnode        *pMnode = pMsg->pMnode;
+  char *msgStr = pMsg->rpcMsg.pCont;
+  SCMCreateTopicReq* pCreate;
+  tDeserializeSCMCreateTopicReq(msgStr, pCreate);
 
   mDebug("topic:%s, start to create", pCreate->name);
 
@@ -436,15 +226,6 @@ static int32_t mndProcessCreateTopicMsg(SMnodeMsg *pMsg) {
     }
   }
 
-  // topic should have different name with stb
-  SStbObj *pStb = mndAcquireStb(pMnode, pCreate->name);
-  if (pStb != NULL) {
-    sdbRelease(pMnode->pSdb, pStb);
-    terrno = TSDB_CODE_MND_NAME_CONFLICT_WITH_STB;
-    mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
-    return -1;
-  }
-
   SDbObj *pDb = mndAcquireDbByTopic(pMnode, pCreate->name);
   if (pDb == NULL) {
     terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
@@ -464,144 +245,7 @@ static int32_t mndProcessCreateTopicMsg(SMnodeMsg *pMsg) {
   return TSDB_CODE_MND_ACTION_IN_PROGRESS;
 }
 
-static int32_t mndCheckAlterTopicMsg(SAlterTopicMsg *pAlter) {
-  SSchema *pSchema = &pAlter->schema;
-  pSchema->colId = htonl(pSchema->colId);
-  pSchema->bytes = htonl(pSchema->bytes);
-
-  if (pSchema->type <= 0) {
-    terrno = TSDB_CODE_MND_INVALID_TOPIC_OPTION;
-    return -1;
-  }
-  if (pSchema->colId < 0 || pSchema->colId >= (TSDB_MAX_COLUMNS + TSDB_MAX_TAGS)) {
-    terrno = TSDB_CODE_MND_INVALID_TOPIC_OPTION;
-    return -1;
-  }
-  if (pSchema->bytes <= 0) {
-    terrno = TSDB_CODE_MND_INVALID_TOPIC_OPTION;
-    return -1;
-  }
-  if (pSchema->name[0] == 0) {
-    terrno = TSDB_CODE_MND_INVALID_TOPIC_OPTION;
-    return -1;
-  }
-
-  return 0;
-}
-
-static int32_t mndUpdateTopic(SMnode *pMnode, SMnodeMsg *pMsg, STopicObj *pOldTopic, STopicObj *pNewTopic) { return 0; }
-
-static int32_t mndProcessAlterTopicMsg(SMnodeMsg *pMsg) {
-  SMnode         *pMnode = pMsg->pMnode;
-  SAlterTopicMsg *pAlter = pMsg->rpcMsg.pCont;
-
-  mDebug("topic:%s, start to alter", pAlter->name);
-
-  if (mndCheckAlterTopicMsg(pAlter) != 0) {
-    mError("topic:%s, failed to alter since %s", pAlter->name, terrstr());
-    return -1;
-  }
-
-  STopicObj *pTopic = mndAcquireTopic(pMnode, pAlter->name);
-  if (pTopic == NULL) {
-    terrno = TSDB_CODE_MND_TOPIC_NOT_EXIST;
-    mError("topic:%s, failed to alter since %s", pAlter->name, terrstr());
-    return -1;
-  }
-
-  STopicObj topicObj = {0};
-  memcpy(&topicObj, pTopic, sizeof(STopicObj));
-
-  int32_t code = mndUpdateTopic(pMnode, pMsg, pTopic, &topicObj);
-  mndReleaseTopic(pMnode, pTopic);
-
-  if (code != 0) {
-    mError("topic:%s, failed to alter since %s", pAlter->name, tstrerror(code));
-    return code;
-  }
-
-  return TSDB_CODE_MND_ACTION_IN_PROGRESS;
-}
-
-static int32_t mndProcessAlterTopicInRsp(SMnodeMsg *pMsg) {
-  mndTransProcessRsp(pMsg);
-  return 0;
-}
-
-static int32_t mndSetDropTopicRedoLogs(SMnode *pMnode, STrans *pTrans, STopicObj *pTopic) {
-  SSdbRaw *pRedoRaw = mndTopicActionEncode(pTopic);
-  if (pRedoRaw == NULL) return -1;
-  if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1;
-  if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING) != 0) return -1;
-
-  return 0;
-}
-
-static int32_t mndSetDropTopicUndoLogs(SMnode *pMnode, STrans *pTrans, STopicObj *pTopic) {
-  SSdbRaw *pUndoRaw = mndTopicActionEncode(pTopic);
-  if (pUndoRaw == NULL) return -1;
-  if (mndTransAppendUndolog(pTrans, pUndoRaw) != 0) return -1;
-  if (sdbSetRawStatus(pUndoRaw, SDB_STATUS_READY) != 0) return -1;
-
-  return 0;
-}
-
-static int32_t mndSetDropTopicCommitLogs(SMnode *pMnode, STrans *pTrans, STopicObj *pTopic) {
-  SSdbRaw *pCommitRaw = mndTopicActionEncode(pTopic);
-  if (pCommitRaw == NULL) return -1;
-  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
-  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED) != 0) return -1;
-
-  return 0;
-}
-
-static int32_t mndSetDropTopicRedoActions(SMnode *pMnode, STrans *pTrans, STopicObj *pTopic) { return 0; }
-
-static int32_t mndSetDropTopicUndoActions(SMnode *pMnode, STrans *pTrans, STopicObj *pTopic) { return 0; }
-
 static int32_t mndDropTopic(SMnode *pMnode, SMnodeMsg *pMsg, STopicObj *pTopic) {
-  int32_t code = -1;
-  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, &pMsg->rpcMsg);
-  if (pTrans == NULL) {
-    mError("topic:%s, failed to drop since %s", pTopic->name, terrstr());
-    return -1;
-  }
-  mDebug("trans:%d, used to drop topic:%s", pTrans->id, pTopic->name);
-
-  if (mndSetDropTopicRedoLogs(pMnode, pTrans, pTopic) != 0) {
-    mError("trans:%d, failed to set redo log since %s", pTrans->id, terrstr());
-    goto DROP_TOPIC_OVER;
-  }
-
-  if (mndSetDropTopicUndoLogs(pMnode, pTrans, pTopic) != 0) {
-    mError("trans:%d, failed to set undo log since %s", pTrans->id, terrstr());
-    goto DROP_TOPIC_OVER;
-  }
-
-  if (mndSetDropTopicCommitLogs(pMnode, pTrans, pTopic) != 0) {
-    mError("trans:%d, failed to set commit log since %s", pTrans->id, terrstr());
-    goto DROP_TOPIC_OVER;
-  }
-
-  if (mndSetDropTopicRedoActions(pMnode, pTrans, pTopic) != 0) {
-    mError("trans:%d, failed to set redo actions since %s", pTrans->id, terrstr());
-    goto DROP_TOPIC_OVER;
-  }
-
-  if (mndSetDropTopicUndoActions(pMnode, pTrans, pTopic) != 0) {
-    mError("trans:%d, failed to set redo actions since %s", pTrans->id, terrstr());
-    goto DROP_TOPIC_OVER;
-  }
-
-  if (mndTransPrepare(pMnode, pTrans) != 0) {
-    mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
-    goto DROP_TOPIC_OVER;
-  }
-
-  code = 0;
-
-DROP_TOPIC_OVER:
-  mndTransDrop(pTrans);
   return 0;
 }
 
@@ -705,11 +349,6 @@ static int32_t mndProcessTopicMetaMsg(SMnodeMsg *pMsg) {
   return 0;
 }
 
-static int32_t mndProcessCreateTopicInRsp(SMnodeMsg *pMsg) {
-  mndTransProcessRsp(pMsg);
-  return 0;
-}
-
 static int32_t mndGetNumOfTopics(SMnode *pMnode, char *dbName, int32_t *pNumOfTopics) {
   SSdb *pSdb = pMnode->pSdb;
 

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

@@ -294,18 +294,18 @@ TRANS_DECODE_OVER:
     return NULL;
   }
 
-  mTrace("trans:%d, decode from raw:%p", pTrans->id, pRaw);
+  mTrace("trans:%d, decode from raw:%p, data:%p", pTrans->id, pRaw, pTrans);
   return pRow;
 }
 
 static int32_t mndTransActionInsert(SSdb *pSdb, STrans *pTrans) {
   pTrans->stage = TRN_STAGE_PREPARE;
-  mTrace("trans:%d, perform insert action", pTrans->id);
+  mTrace("trans:%d, perform insert action, data:%p", pTrans->id, pTrans);
   return 0;
 }
 
 static int32_t mndTransActionDelete(SSdb *pSdb, STrans *pTrans) {
-  mTrace("trans:%d, perform delete action", pTrans->id);
+  mTrace("trans:%d, perform delete action, data:%p", pTrans->id, pTrans);
 
   mndTransDropLogs(pTrans->redoLogs);
   mndTransDropLogs(pTrans->undoLogs);
@@ -317,7 +317,7 @@ static int32_t mndTransActionDelete(SSdb *pSdb, STrans *pTrans) {
 }
 
 static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *pOldTrans, STrans *pNewTrans) {
-  mTrace("trans:%d, perform update action", pOldTrans->id);
+  mTrace("trans:%d, perform update action, data:%p", pOldTrans->id, pOldTrans);
   pOldTrans->stage = pNewTrans->stage;
   return 0;
 }
@@ -362,14 +362,14 @@ STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, SRpcMsg *pMsg) {
     return NULL;
   }
 
-  mDebug("trans:%d, is created", pTrans->id);
+  mDebug("trans:%d, is created, data:%p", pTrans->id, pTrans);
   return pTrans;
 }
 
 static void mndTransDropLogs(SArray *pArray) {
   for (int32_t i = 0; i < pArray->size; ++i) {
     SSdbRaw *pRaw = taosArrayGetP(pArray, i);
-    tfree(pRaw);
+    sdbFreeRaw(pRaw);
   }
 
   taosArrayDestroy(pArray);
@@ -391,7 +391,7 @@ void mndTransDrop(STrans *pTrans) {
   mndTransDropActions(pTrans->redoActions);
   mndTransDropActions(pTrans->undoActions);
 
-  // mDebug("trans:%d, is dropped, data:%p", pTrans->id, pTrans);
+  mDebug("trans:%d, is dropped, data:%p", pTrans->id, pTrans);
   tfree(pTrans);
 }
 
@@ -442,7 +442,7 @@ static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) {
   }
   sdbSetRawStatus(pRaw, SDB_STATUS_READY);
 
-  mTrace("trans:%d, sync to other nodes", pTrans->id);
+  mDebug("trans:%d, sync to other nodes", pTrans->id);
   int32_t code = mndSyncPropose(pMnode, pRaw);
   if (code != 0) {
     mError("trans:%d, failed to sync since %s", pTrans->id, terrstr());
@@ -450,7 +450,7 @@ static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) {
     return -1;
   }
 
-  mTrace("trans:%d, sync finished", pTrans->id);
+  mDebug("trans:%d, sync finished", pTrans->id);
 
   code = sdbWrite(pMnode->pSdb, pRaw);
   if (code != 0) {

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

@@ -86,7 +86,6 @@ SSdbRaw *mndVgroupActionEncode(SVgObj *pVgroup) {
   for (int8_t i = 0; i < pVgroup->replica; ++i) {
     SVnodeGid *pVgid = &pVgroup->vnodeGid[i];
     SDB_SET_INT32(pRaw, dataPos, pVgid->dnodeId)
-    SDB_SET_INT8(pRaw, dataPos, pVgid->role)
   }
   SDB_SET_RESERVE(pRaw, dataPos, TSDB_VGROUP_RESERVE_SIZE)
   SDB_SET_DATALEN(pRaw, dataPos);
@@ -121,7 +120,6 @@ SSdbRow *mndVgroupActionDecode(SSdbRaw *pRaw) {
   for (int8_t i = 0; i < pVgroup->replica; ++i) {
     SVnodeGid *pVgid = &pVgroup->vnodeGid[i];
     SDB_GET_INT32(pRaw, pRow, dataPos, &pVgid->dnodeId)
-    SDB_GET_INT8(pRaw, pRow, dataPos, (int8_t *)&pVgid->role)
   }
   SDB_GET_RESERVE(pRaw, pRow, dataPos, TSDB_VGROUP_RESERVE_SIZE)
 
@@ -237,44 +235,95 @@ SDropVnodeMsg *mndBuildDropVnodeMsg(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *p
   return pDrop;
 }
 
-static int32_t mndGetAvailableDnode(SMnode *pMnode, SVgObj *pVgroup) {
+static SArray *mndBuildDnodesArray(SMnode *pMnode) {
   SSdb   *pSdb = pMnode->pSdb;
-  int32_t allocedVnodes = 0;
-  void   *pIter = NULL;
+  int32_t numOfDnodes = mndGetDnodeSize(pMnode);
+  SArray *pArray = taosArrayInit(numOfDnodes, sizeof(SDnodeObj));
+  if (pArray == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return NULL;
+  }
 
-  while (allocedVnodes < pVgroup->replica) {
+  void *pIter = NULL;
+  while (1) {
     SDnodeObj *pDnode = NULL;
     pIter = sdbFetch(pSdb, SDB_DNODE, pIter, (void **)&pDnode);
     if (pIter == NULL) break;
 
-    // todo
-    if (mndIsDnodeInReadyStatus(pMnode, pDnode)) {
-      SVnodeGid *pVgid = &pVgroup->vnodeGid[allocedVnodes];
-      pVgid->dnodeId = pDnode->id;
-      if (pVgroup->replica == 1) {
-        pVgid->role = TAOS_SYNC_STATE_LEADER;
-      } else {
-        pVgid->role = TAOS_SYNC_STATE_FOLLOWER;
-      }
-      allocedVnodes++;
+    int32_t numOfVnodes = mndGetVnodesNum(pMnode, pDnode->id);
+
+    bool isMnode = mndIsMnode(pMnode, pDnode->id);
+    if (isMnode) {
+      pDnode->numOfVnodes++;
+    }
+
+    bool isReady = mndIsDnodeInReadyStatus(pMnode, pDnode);
+    if (isReady) {
+      taosArrayPush(pArray, pDnode);
     }
+
+    mDebug("dnode:%d, numOfVnodes:%d numOfSupportVnodes:%d isMnode:%d ready:%d", pDnode->id, numOfVnodes,
+           pDnode->numOfSupportVnodes, isMnode, isReady);
     sdbRelease(pSdb, pDnode);
   }
 
-  if (allocedVnodes != pVgroup->replica) {
-    terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
-    return -1;
+  return pArray;
+}
+
+static int32_t mndCompareDnodeVnodes(SDnodeObj *pDnode1, SDnodeObj *pDnode2) {
+  float d1Score = (float)pDnode1->numOfVnodes / pDnode1->numOfSupportVnodes;
+  float d2Score = (float)pDnode2->numOfVnodes / pDnode2->numOfSupportVnodes;
+  return d1Score > d2Score ? 0 : 1;
+}
+
+static int32_t mndGetAvailableDnode(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray) {
+  SSdb   *pSdb = pMnode->pSdb;
+  int32_t allocedVnodes = 0;
+  void   *pIter = NULL;
+
+  taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes);
+
+  for (int32_t v = 0; v < pVgroup->replica; ++v) {
+    SVnodeGid *pVgid = &pVgroup->vnodeGid[v];
+    SDnodeObj *pDnode = taosArrayGet(pArray, v);
+    if (pDnode == NULL || pDnode->numOfVnodes > pDnode->numOfSupportVnodes) {
+      terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
+      return -1;
+    }
+
+    pVgid->dnodeId = pDnode->id;
+    if (pVgroup->replica == 1) {
+      pVgid->role = TAOS_SYNC_STATE_LEADER;
+    } else {
+      pVgid->role = TAOS_SYNC_STATE_FOLLOWER;
+    }
+
+    mDebug("db:%s, vgId:%d, vindex:%d dnodeId:%d is alloced", pVgroup->dbName, pVgroup->vgId, v, pVgid->dnodeId);
+    pDnode->numOfVnodes++;
   }
+
   return 0;
 }
 
 int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups) {
-  SVgObj *pVgroups = calloc(pDb->cfg.numOfVgroups, sizeof(SVgObj));
+  int32_t code = -1;
+  SArray *pArray = NULL;
+  SVgObj *pVgroups = NULL;
+
+  pVgroups = calloc(pDb->cfg.numOfVgroups, sizeof(SVgObj));
   if (pVgroups == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return -1;
+    goto ALLOC_VGROUP_OVER;
   }
 
+  pArray = mndBuildDnodesArray(pMnode);
+  if (pArray == NULL) {
+    goto ALLOC_VGROUP_OVER;
+  }
+
+  mDebug("db:%s, total %d dnodes used to create %d vgroups (%d vnodes)", pDb->name, (int32_t)taosArrayGetSize(pArray),
+         pDb->cfg.numOfVgroups, pDb->cfg.numOfVgroups * pDb->cfg.replications);
+
   int32_t  allocedVgroups = 0;
   int32_t  maxVgId = sdbGetMaxId(pMnode->pSdb, SDB_VGROUP);
   uint32_t hashMin = 0;
@@ -298,17 +347,23 @@ int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups) {
     pVgroup->dbUid = pDb->uid;
     pVgroup->replica = pDb->cfg.replications;
 
-    if (mndGetAvailableDnode(pMnode, pVgroup) != 0) {
+    if (mndGetAvailableDnode(pMnode, pVgroup, pArray) != 0) {
       terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
-      free(pVgroups);
-      return -1;
+      goto ALLOC_VGROUP_OVER;
     }
 
     allocedVgroups++;
   }
 
   *ppVgroups = pVgroups;
-  return 0;
+  code = 0;
+
+  mDebug("db:%s, %d vgroups is alloced, replica:%d", pDb->name, pDb->cfg.numOfVgroups, pDb->cfg.replications);
+
+ALLOC_VGROUP_OVER:
+  if (code != 0) free(pVgroups);
+  taosArrayDestroy(pArray);
+  return code;
 }
 
 SEpSet mndGetVgroupEpset(SMnode *pMnode, SVgObj *pVgroup) {
@@ -348,6 +403,7 @@ static int32_t mndProcessDropVnodeRsp(SMnodeMsg *pMsg) {
 }
 
 static int32_t mndProcessSyncVnodeRsp(SMnodeMsg *pMsg) { return 0; }
+
 static int32_t mndProcessCompactVnodeRsp(SMnodeMsg *pMsg) { return 0; }
 
 static int32_t mndGetVgroupMaxReplica(SMnode *pMnode, char *dbName, int8_t *pReplica, int32_t *pNumOfVgroups) {
@@ -478,7 +534,7 @@ static void mndCancelGetNextVgroup(SMnode *pMnode, void *pIter) {
   sdbCancelFetch(pSdb, pIter);
 }
 
-static int32_t mndGetVnodesNum(SMnode *pMnode, int32_t dnodeId) {
+int32_t mndGetVnodesNum(SMnode *pMnode, int32_t dnodeId) {
   SSdb   *pSdb = pMnode->pSdb;
   int32_t numOfVnodes = 0;
   void   *pIter = NULL;

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

@@ -146,15 +146,15 @@ static int32_t mndInitSteps(SMnode *pMnode) {
   if (mndAllocStep(pMnode, "mnode-sdb", mndInitSdb, mndCleanupSdb) != 0) return -1;
   if (mndAllocStep(pMnode, "mnode-trans", mndInitTrans, mndCleanupTrans) != 0) return -1;
   if (mndAllocStep(pMnode, "mnode-cluster", mndInitCluster, mndCleanupCluster) != 0) return -1;
-  if (mndAllocStep(pMnode, "mnode-dnode", mndInitDnode, mndCleanupDnode) != 0) return -1;
   if (mndAllocStep(pMnode, "mnode-mnode", mndInitMnode, mndCleanupMnode) != 0) return -1;
-  if (mndAllocStep(pMnode, "mnode-acct", mndInitAcct, mndCleanupAcct) != 0) return -1;
-  if (mndAllocStep(pMnode, "mnode-auth", mndInitAuth, mndCleanupAuth) != 0) return -1;
+  if (mndAllocStep(pMnode, "mnode-dnode", mndInitDnode, mndCleanupDnode) != 0) return -1;
   if (mndAllocStep(pMnode, "mnode-user", mndInitUser, mndCleanupUser) != 0) return -1;
-  if (mndAllocStep(pMnode, "mnode-db", mndInitDb, mndCleanupDb) != 0) return -1;
+  if (mndAllocStep(pMnode, "mnode-auth", mndInitAuth, mndCleanupAuth) != 0) return -1;
+  if (mndAllocStep(pMnode, "mnode-acct", mndInitAcct, mndCleanupAcct) != 0) return -1;
+  if (mndAllocStep(pMnode, "mnode-topic", mndInitTopic, mndCleanupTopic) != 0) return -1;
   if (mndAllocStep(pMnode, "mnode-vgroup", mndInitVgroup, mndCleanupVgroup) != 0) return -1;
   if (mndAllocStep(pMnode, "mnode-stb", mndInitStb, mndCleanupStb) != 0) return -1;
-  if (mndAllocStep(pMnode, "mnode-topic", mndInitTopic, mndCleanupTopic) != 0) return -1;
+  if (mndAllocStep(pMnode, "mnode-db", mndInitDb, mndCleanupDb) != 0) return -1;
   if (mndAllocStep(pMnode, "mnode-func", mndInitFunc, mndCleanupFunc) != 0) return -1;
   if (pMnode->clusterId <= 0) {
     if (mndAllocStep(pMnode, "mnode-sdb-deploy", mndDeploySdb, NULL) != 0) return -1;

source/dnode/mnode/sdb/inc/sdbInt.h

@@ -72,6 +72,7 @@ typedef struct SSdb {
 } SSdb;
 
 int32_t sdbWriteFile(SSdb *pSdb);
+void    sdbPrintOper(SSdb *pSdb, SSdbRow *pRow, const char *oper);
 
 #ifdef __cplusplus
 }

source/dnode/mnode/sdb/src/sdb.c

@@ -80,16 +80,12 @@ void sdbCleanup(SSdb *pSdb) {
     SHashObj *hash = pSdb->hashObjs[i];
     if (hash == NULL) continue;
 
-    SdbDeleteFp deleteFp = pSdb->deleteFps[i];
-    SSdbRow   **ppRow = taosHashIterate(hash, NULL);
+    SSdbRow **ppRow = taosHashIterate(hash, NULL);
     while (ppRow != NULL) {
       SSdbRow *pRow = *ppRow;
       if (pRow == NULL) continue;
 
-      if (deleteFp != NULL) {
-        (*deleteFp)(pSdb, pRow->pObj);
-      }
-      sdbFreeRow(pRow);
+      sdbFreeRow(pSdb, pRow);
       ppRow = taosHashIterate(hash, ppRow);
     }
   }

source/dnode/mnode/sdb/src/sdbFile.c

@@ -151,7 +151,7 @@ int32_t sdbWriteFile(SSdb *pSdb) {
         if (taosWriteFile(fd, pRaw, writeLen) != writeLen) {
           code = TAOS_SYSTEM_ERROR(terrno);
           taosHashCancelIterate(hash, ppRow);
-          free(pRaw);
+          sdbFreeRaw(pRaw);
           break;
         }
 
@@ -159,7 +159,7 @@ int32_t sdbWriteFile(SSdb *pSdb) {
         if (taosWriteFile(fd, &cksum, sizeof(int32_t)) != sizeof(int32_t)) {
           code = TAOS_SYSTEM_ERROR(terrno);
           taosHashCancelIterate(hash, ppRow);
-          free(pRaw);
+          sdbFreeRaw(pRaw);
           break;
         }
       } else {
@@ -168,7 +168,7 @@ int32_t sdbWriteFile(SSdb *pSdb) {
         break;
       }
 
-      free(pRaw);
+      sdbFreeRaw(pRaw);
       ppRow = taosHashIterate(hash, ppRow);
     }
     taosWUnLockLatch(pLock);

source/dnode/mnode/sdb/src/sdbHash.c

@@ -16,6 +16,50 @@
 #define _DEFAULT_SOURCE
 #include "sdbInt.h"
 
+static const char *sdbTableName(ESdbType type) {
+  switch (type) {
+    case SDB_TRANS:
+      return "trans";
+    case SDB_CLUSTER:
+      return "cluster";
+    case SDB_MNODE:
+      return "mnode";
+    case SDB_DNODE:
+      return "dnode";
+    case SDB_USER:
+      return "user";
+    case SDB_AUTH:
+      return "auth";
+    case SDB_ACCT:
+      return "acct";
+    case SDB_TOPIC:
+      return "topic";
+    case SDB_VGROUP:
+      return "vgId";
+    case SDB_STB:
+      return "stb";
+    case SDB_DB:
+      return "db";
+    case SDB_FUNC:
+      return "func";
+    default:
+      return "undefine";
+  }
+}
+
+void sdbPrintOper(SSdb *pSdb, SSdbRow *pRow, const char *oper) {
+  EKeyType keyType = pSdb->keyTypes[pRow->type];
+
+  if (keyType == SDB_KEY_BINARY) {
+    mTrace("%s:%s, refCount:%d oper:%s", sdbTableName(pRow->type), (char *)pRow->pObj, pRow->refCount, oper);
+  } else if (keyType == SDB_KEY_INT32) {
+    mTrace("%s:%d, refCount:%d oper:%s", sdbTableName(pRow->type), *(int32_t *)pRow->pObj, pRow->refCount, oper);
+  } else if (keyType == SDB_KEY_INT64) {
+    mTrace("%s:%" PRId64 ", refCount:%d oper:%s", sdbTableName(pRow->type), *(int64_t *)pRow->pObj, pRow->refCount, oper);
+  } else {
+  }
+}
+
 static SHashObj *sdbGetHash(SSdb *pSdb, int32_t type) {
   if (type >= SDB_MAX || type <= SDB_START) {
     terrno = TSDB_CODE_SDB_INVALID_TABLE_TYPE;
@@ -55,17 +99,18 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
   SSdbRow *pOldRow = taosHashGet(hash, pRow->pObj, keySize);
   if (pOldRow != NULL) {
     taosWUnLockLatch(pLock);
-    sdbFreeRow(pRow);
+    sdbFreeRow(pSdb, pRow);
     terrno = TSDB_CODE_SDB_OBJ_ALREADY_THERE;
     return terrno;
   }
 
-  pRow->refCount = 1;
+  pRow->refCount = 0;
   pRow->status = pRaw->status;
+  sdbPrintOper(pSdb, pRow, "insertRow");
 
   if (taosHashPut(hash, pRow->pObj, keySize, &pRow, sizeof(void *)) != 0) {
     taosWUnLockLatch(pLock);
-    sdbFreeRow(pRow);
+    sdbFreeRow(pSdb, pRow);
     terrno = TSDB_CODE_SDB_OBJ_ALREADY_THERE;
     return terrno;
   }
@@ -83,7 +128,7 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
       taosWLockLatch(pLock);
       taosHashRemove(hash, pRow->pObj, keySize);
       taosWUnLockLatch(pLock);
-      sdbFreeRow(pRow);
+      sdbFreeRow(pSdb, pRow);
       terrno = code;
       return terrno;
     }
@@ -113,7 +158,7 @@ static int32_t sdbUpdateRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
     code = (*updateFp)(pSdb, pOldRow->pObj, pNewRow->pObj);
   }
 
-  sdbFreeRow(pNewRow);
+  sdbFreeRow(pSdb, pNewRow);
   return code;
 }
 
@@ -123,14 +168,10 @@ static int32_t sdbDeleteRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
   SRWLatch *pLock = &pSdb->locks[pRow->type];
   taosWLockLatch(pLock);
 
-  // remove attached object such as trans
-  SdbDeleteFp deleteFp = pSdb->deleteFps[pRow->type];
-  if (deleteFp != NULL) (*deleteFp)(pSdb, pRow->pObj);
-
   SSdbRow **ppOldRow = taosHashGet(hash, pRow->pObj, keySize);
   if (ppOldRow == NULL || *ppOldRow == NULL) {
     taosWUnLockLatch(pLock);
-    sdbFreeRow(pRow);
+    sdbFreeRow(pSdb, pRow);
     terrno = TSDB_CODE_SDB_OBJ_NOT_THERE;
     return terrno;
   }
@@ -140,8 +181,8 @@ static int32_t sdbDeleteRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
   taosHashRemove(hash, pOldRow->pObj, keySize);
   taosWUnLockLatch(pLock);
 
-  sdbRelease(pSdb, pOldRow->pObj);
-  sdbFreeRow(pRow);
+  // sdbRelease(pSdb, pOldRow->pObj);
+  sdbFreeRow(pSdb, pRow);
   return code;
 }
 
@@ -206,6 +247,7 @@ void *sdbAcquire(SSdb *pSdb, ESdbType type, void *pKey) {
     case SDB_STATUS_UPDATING:
       atomic_add_fetch_32(&pRow->refCount, 1);
       pRet = pRow->pObj;
+      sdbPrintOper(pSdb, pRow, "acquireRow");
       break;
     case SDB_STATUS_CREATING:
       terrno = TSDB_CODE_SDB_OBJ_CREATING;
@@ -232,13 +274,9 @@ void sdbRelease(SSdb *pSdb, void *pObj) {
   taosRLockLatch(pLock);
 
   int32_t ref = atomic_sub_fetch_32(&pRow->refCount, 1);
+  sdbPrintOper(pSdb, pRow, "releaseRow");
   if (ref <= 0 && pRow->status == SDB_STATUS_DROPPED) {
-    SdbDeleteFp deleteFp = pSdb->deleteFps[pRow->type];
-    if (deleteFp != NULL) {
-      (*deleteFp)(pSdb, pRow->pObj);
-    }
-
-    sdbFreeRow(pRow);
+    sdbFreeRow(pSdb, pRow);
   }
 
   taosRUnLockLatch(pLock);
@@ -255,9 +293,9 @@ void *sdbFetch(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj) {
 
   if (pIter != NULL) {
     SSdbRow *pLastRow = *(SSdbRow **)pIter;
-    int32_t  ref = atomic_sub_fetch_32(&pLastRow->refCount, 1);
+    int32_t  ref = atomic_load_32(&pLastRow->refCount);
     if (ref <= 0 && pLastRow->status == SDB_STATUS_DROPPED) {
-      sdbFreeRow(pLastRow);
+      sdbFreeRow(pSdb, pLastRow);
     }
   }
 
@@ -270,6 +308,7 @@ void *sdbFetch(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj) {
     }
 
     atomic_add_fetch_32(&pRow->refCount, 1);
+    sdbPrintOper(pSdb, pRow, "fetchRow");
     *ppObj = pRow->pObj;
     break;
   }

source/dnode/mnode/sdb/src/sdbRaw.c

@@ -27,12 +27,12 @@ SSdbRaw *sdbAllocRaw(ESdbType type, int8_t sver, int32_t dataLen) {
   pRaw->sver = sver;
   pRaw->dataLen = dataLen;
 
-  // mTrace("raw:%p, is created, len:%d", pRaw, dataLen);
+  mTrace("raw:%p, is created, len:%d", pRaw, dataLen);
   return pRaw;
 }
 
 void sdbFreeRaw(SSdbRaw *pRaw) {
-  // mTrace("raw:%p, is freed", pRaw);
+  mTrace("raw:%p, is freed", pRaw);
   free(pRaw);
 }
 

source/dnode/mnode/sdb/src/sdbRow.c

@@ -35,4 +35,13 @@ void *sdbGetRowObj(SSdbRow *pRow) {
   return pRow->pObj;
 }
 
-void sdbFreeRow(SSdbRow *pRow) { tfree(pRow); }
+void sdbFreeRow(SSdb *pSdb, SSdbRow *pRow) {
+  // remove attached object such as trans
+  SdbDeleteFp deleteFp = pSdb->deleteFps[pRow->type];
+  if (deleteFp != NULL) {
+    (*deleteFp)(pSdb, pRow->pObj);
+  }
+
+  sdbPrintOper(pSdb, pRow, "freeRow");
+  tfree(pRow);
+}

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

@@ -28,8 +28,3 @@ int vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
   vInfo("sync message is processed");
   return 0;
 }
-
-int vnodeProcessConsumeReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
-  vInfo("consume message is processed");
-  return 0;
-}

source/libs/catalog/inc/catalogInt.h

@@ -31,6 +31,11 @@ extern "C" {
 
 #define CTG_DEFAULT_INVALID_VERSION (-1)
 
+enum {
+  CTG_READ = 1,
+  CTG_WRITE,
+};
+
 typedef struct SVgroupListCache {
   int32_t vgroupVersion;
   SHashObj *cache;        // key:vgId, value:SVgroupInfo
@@ -41,6 +46,7 @@ typedef struct SDBVgroupCache {
 } SDBVgroupCache;
 
 typedef struct STableMetaCache {
+  SRWLatch  stableLock;
   SHashObj *cache;           //key:fulltablename, value:STableMeta
   SHashObj *stableCache;     //key:suid, value:STableMeta*
 } STableMetaCache;
@@ -71,6 +77,31 @@ typedef uint32_t (*tableNameHashFp)(const char *, uint32_t);
 #define CTG_ERR_LRET(c,...) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { ctgError(__VA_ARGS__); terrno = _code; return _code; } } while (0)
 #define CTG_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
 
+#define CTG_LOCK(type, _lock) do {   \
+  if (CTG_READ == (type)) {          \
+    if ((*(_lock)) < 0) assert(0);  \
+    taosRLockLatch(_lock);           \
+    ctgDebug("CTG RLOCK%p, %s:%d", (_lock), __FILE__, __LINE__); \
+  } else {                                                \
+    if ((*(_lock)) < 0) assert(0);                          \
+    taosWLockLatch(_lock);                                \
+    ctgDebug("CTG WLOCK%p, %s:%d", (_lock), __FILE__, __LINE__);  \
+  }                                                       \
+} while (0)
+
+#define CTG_UNLOCK(type, _lock) do {                       \
+  if (CTG_READ == (type)) {                                \
+    if ((*(_lock)) <= 0) assert(0);                         \
+    taosRUnLockLatch(_lock);                              \
+    ctgDebug("CTG RULOCK%p, %s:%d", (_lock), __FILE__, __LINE__); \
+  } else {                                                \
+    if ((*(_lock)) <= 0) assert(0);                         \
+    taosWUnLockLatch(_lock);                              \
+    ctgDebug("CTG WULOCK%p, %s:%d", (_lock), __FILE__, __LINE__); \
+  }                                                       \
+} while (0)
+
+
 #ifdef __cplusplus
 }
 #endif

source/libs/catalog/src/catalog.c

@@ -20,24 +20,28 @@
 
 SCatalogMgmt ctgMgmt = {0};
 
-int32_t ctgGetDBVgroupFromCache(struct SCatalog* pCatalog, const char *dbName, SDBVgroupInfo *dbInfo, int32_t *exist) {
+int32_t ctgGetDBVgroupFromCache(struct SCatalog* pCatalog, const char *dbName, SDBVgroupInfo **dbInfo, bool *inCache) {
   if (NULL == pCatalog->dbCache.cache) {
-    *exist = 0;
+    *inCache = false;
     return TSDB_CODE_SUCCESS;
   }
 
-  SDBVgroupInfo *info = taosHashGet(pCatalog->dbCache.cache, dbName, strlen(dbName));
+  SDBVgroupInfo *info = taosHashAcquire(pCatalog->dbCache.cache, dbName, strlen(dbName));
 
   if (NULL == info) {
-    *exist = 0;
+    *inCache = false;
     return TSDB_CODE_SUCCESS;
   }
 
-  if (dbInfo) {
-    *dbInfo = *info;
+  CTG_LOCK(CTG_READ, &info->lock);
+  if (NULL == info->vgInfo) {
+    CTG_UNLOCK(CTG_READ, &info->lock);
+    *inCache = false;
+    return TSDB_CODE_SUCCESS;
   }
 
-  *exist = 1;
+  *dbInfo = info;
+  *inCache = true;
   
   return TSDB_CODE_SUCCESS;
 }
@@ -80,46 +84,51 @@ int32_t ctgGetTableMetaFromCache(struct SCatalog* pCatalog, const SName* pTableN
   char tbFullName[TSDB_TABLE_FNAME_LEN];
   tNameExtractFullName(pTableName, tbFullName);
 
-  STableMeta *tbMeta = taosHashGet(pCatalog->tableCache.cache, tbFullName, strlen(tbFullName));
+  *pTableMeta = NULL;
+
+  size_t sz = 0;
+  STableMeta *tbMeta = taosHashGetCloneExt(pCatalog->tableCache.cache, tbFullName, strlen(tbFullName), NULL, (void **)pTableMeta, &sz);
 
-  if (NULL == tbMeta) {
+  if (NULL == *pTableMeta) {
     *exist = 0;
     return TSDB_CODE_SUCCESS;
   }
 
-  if (tbMeta->tableType == TSDB_CHILD_TABLE) {
-    STableMeta **stbMeta = taosHashGet(pCatalog->tableCache.stableCache, &tbMeta->suid, sizeof(tbMeta->suid));
-    if (NULL == stbMeta || NULL == *stbMeta) {
-      *exist = 0;
-      return TSDB_CODE_SUCCESS;
-    }
-
-    if ((*stbMeta)->suid != tbMeta->suid) {
-      ctgError("stable cache error, expected suid:%"PRId64 ",actual suid:%"PRId64, tbMeta->suid, (*stbMeta)->suid);
-      CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
-    }
+  *exist = 1;
 
-    int32_t metaSize = sizeof(STableMeta) + ((*stbMeta)->tableInfo.numOfTags + (*stbMeta)->tableInfo.numOfColumns) * sizeof(SSchema);
-    *pTableMeta = calloc(1, metaSize);
-    if (NULL == *pTableMeta) {
-      ctgError("calloc size[%d] failed", metaSize);
-      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
-    }
+  if (tbMeta->tableType != TSDB_CHILD_TABLE) {
+    return TSDB_CODE_SUCCESS;
+  }
+  
+  CTG_LOCK(CTG_READ, &pCatalog->tableCache.stableLock);
+  
+  STableMeta **stbMeta = taosHashGet(pCatalog->tableCache.stableCache, &tbMeta->suid, sizeof(tbMeta->suid));
+  if (NULL == stbMeta || NULL == *stbMeta) {
+    CTG_UNLOCK(CTG_READ, &pCatalog->tableCache.stableLock);
+    qError("no stable:%"PRIx64 " meta in cache", tbMeta->suid);
+    tfree(*pTableMeta);
+    *exist = 0;
+    return TSDB_CODE_SUCCESS;
+  }
 
-    memcpy(*pTableMeta, tbMeta, sizeof(SCTableMeta));
-    memcpy(&(*pTableMeta)->sversion, &(*stbMeta)->sversion, metaSize - sizeof(SCTableMeta));
-  } else {
-    int32_t metaSize = sizeof(STableMeta) + (tbMeta->tableInfo.numOfTags + tbMeta->tableInfo.numOfColumns) * sizeof(SSchema);
-    *pTableMeta = calloc(1, metaSize);
-    if (NULL == *pTableMeta) {
-      ctgError("calloc size[%d] failed", metaSize);
-      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
-    }
+  if ((*stbMeta)->suid != tbMeta->suid) {    
+    CTG_UNLOCK(CTG_READ, &pCatalog->tableCache.stableLock);
+    tfree(*pTableMeta);
+    ctgError("stable cache error, expected suid:%"PRId64 ",actual suid:%"PRId64, tbMeta->suid, (*stbMeta)->suid);
+    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
+  }
 
-    memcpy(*pTableMeta, tbMeta, metaSize);
+  int32_t metaSize = sizeof(STableMeta) + ((*stbMeta)->tableInfo.numOfTags + (*stbMeta)->tableInfo.numOfColumns) * sizeof(SSchema);
+  *pTableMeta = realloc(*pTableMeta, metaSize);
+  if (NULL == *pTableMeta) {    
+    CTG_UNLOCK(CTG_READ, &pCatalog->tableCache.stableLock);
+    ctgError("calloc size[%d] failed", metaSize);
+    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
   }
 
-  *exist = 1;
+  memcpy(&(*pTableMeta)->sversion, &(*stbMeta)->sversion, metaSize - sizeof(SCTableMeta));
+
+  CTG_UNLOCK(CTG_READ, &pCatalog->tableCache.stableLock);
   
   return TSDB_CODE_SUCCESS;
 }
@@ -223,9 +232,11 @@ int32_t ctgGetHashFunction(int8_t hashMethod, tableNameHashFp *fp) {
 int32_t ctgGetVgInfoFromDB(struct SCatalog *pCatalog, void *pRpc, const SEpSet *pMgmtEps, SDBVgroupInfo *dbInfo, SArray** vgroupList) {
   SHashObj *vgroupHash = NULL;
   SVgroupInfo *vgInfo = NULL;
+  SArray *vgList = NULL;
+  int32_t code = 0;
 
-  *vgroupList = taosArrayInit(taosHashGetSize(dbInfo->vgInfo), sizeof(SVgroupInfo));
-  if (NULL == *vgroupList) {
+  vgList = taosArrayInit(taosHashGetSize(dbInfo->vgInfo), sizeof(SVgroupInfo));
+  if (NULL == vgList) {
     ctgError("taosArrayInit failed");
     CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);    
   }
@@ -234,19 +245,34 @@ int32_t ctgGetVgInfoFromDB(struct SCatalog *pCatalog, void *pRpc, const SEpSet *
   while (pIter) {
     vgInfo = pIter;
 
-    if (NULL == taosArrayPush(*vgroupList, vgInfo)) {
+    if (NULL == taosArrayPush(vgList, vgInfo)) {
       ctgError("taosArrayPush failed");
-      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
+      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
     }
     
     pIter = taosHashIterate(dbInfo->vgInfo, pIter);
     vgInfo = NULL;
   }
 
+  *vgroupList = vgList;
+  vgList = NULL;
+
   return TSDB_CODE_SUCCESS;
+
+_return:
+
+  if (vgList) {
+    taosArrayDestroy(vgList);
+  }
+
+  CTG_RET(code);
 }
 
 int32_t ctgGetVgInfoFromHashValue(SDBVgroupInfo *dbInfo, const SName *pTableName, SVgroupInfo *pVgroup) {
+  int32_t code = 0;
+  
+  CTG_LOCK(CTG_READ, &dbInfo->lock);
+  
   int32_t vgNum = taosHashGetSize(dbInfo->vgInfo);
   char db[TSDB_DB_FNAME_LEN] = {0};
   tNameGetFullDbName(pTableName, db);
@@ -259,7 +285,7 @@ int32_t ctgGetVgInfoFromHashValue(SDBVgroupInfo *dbInfo, const SName *pTableName
   tableNameHashFp fp = NULL;
   SVgroupInfo *vgInfo = NULL;
 
-  CTG_ERR_RET(ctgGetHashFunction(dbInfo->hashMethod, &fp));
+  CTG_ERR_JRET(ctgGetHashFunction(dbInfo->hashMethod, &fp));
 
   char tbFullName[TSDB_TABLE_FNAME_LEN];
   tNameExtractFullName(pTableName, tbFullName);
@@ -279,19 +305,23 @@ int32_t ctgGetVgInfoFromHashValue(SDBVgroupInfo *dbInfo, const SName *pTableName
 
   if (NULL == vgInfo) {
     ctgError("no hash range found for hashvalue[%u]", hashValue);
-    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
+    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
   }
 
   *pVgroup = *vgInfo;
 
-  return TSDB_CODE_SUCCESS;
+_return:
+
+  CTG_UNLOCK(CTG_READ, &dbInfo->lock);
+  
+  CTG_RET(TSDB_CODE_SUCCESS);
 }
 
 int32_t ctgGetTableMetaImpl(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, bool forceUpdate, STableMeta** pTableMeta) {
   if (NULL == pCatalog || NULL == pRpc || NULL == pMgmtEps || NULL == pTableName || NULL == pTableMeta) {
     CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
   }
-
+  
   int32_t exist = 0;
 
   if (!forceUpdate) {  
@@ -316,21 +346,23 @@ int32_t ctgGetTableMetaImpl(struct SCatalog* pCatalog, void *pRpc, const SEpSet*
 
 
 int32_t ctgUpdateTableMetaCache(struct SCatalog *pCatalog, STableMetaOutput *output) {
+  int32_t code = 0;
+  
   if (output->metaNum != 1 && output->metaNum != 2) {
     ctgError("invalid table meta number[%d] got from meta rsp", output->metaNum);
-    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
+    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
   }
 
   if (NULL == output->tbMeta) {
     ctgError("no valid table meta got from meta rsp");
-    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
+    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
   }
 
   if (NULL == pCatalog->tableCache.cache) {
     pCatalog->tableCache.cache = taosHashInit(ctgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
     if (NULL == pCatalog->tableCache.cache) {
       ctgError("init hash[%d] for tablemeta cache failed", ctgMgmt.cfg.maxTblCacheNum);
-      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
+      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
     }
   }
 
@@ -338,50 +370,59 @@ int32_t ctgUpdateTableMetaCache(struct SCatalog *pCatalog, STableMetaOutput *out
     pCatalog->tableCache.stableCache = taosHashInit(ctgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), true, HASH_ENTRY_LOCK);
     if (NULL == pCatalog->tableCache.stableCache) {
       ctgError("init hash[%d] for stablemeta cache failed", ctgMgmt.cfg.maxTblCacheNum);
-      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
+      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
     }
   }
 
   if (output->metaNum == 2) {
     if (taosHashPut(pCatalog->tableCache.cache, output->ctbFname, strlen(output->ctbFname), &output->ctbMeta, sizeof(output->ctbMeta)) != 0) {
       ctgError("push ctable[%s] to table cache failed", output->ctbFname);
-      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
+      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
     }
 
     if (TSDB_SUPER_TABLE != output->tbMeta->tableType) {
       ctgError("table type[%d] error, expected:%d", output->tbMeta->tableType, TSDB_SUPER_TABLE);
-      CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
+      CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
     }    
   }
 
   int32_t tbSize = sizeof(*output->tbMeta) + sizeof(SSchema) * (output->tbMeta->tableInfo.numOfColumns + output->tbMeta->tableInfo.numOfTags);
-  if (taosHashPut(pCatalog->tableCache.cache, output->tbFname, strlen(output->tbFname), output->tbMeta, tbSize) != 0) {
-    ctgError("push table[%s] to table cache failed", output->tbFname);
-    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
-  }
 
   if (TSDB_SUPER_TABLE == output->tbMeta->tableType) {
-    if (taosHashPut(pCatalog->tableCache.stableCache, &output->tbMeta->suid, sizeof(output->tbMeta->suid), &output->tbMeta, POINTER_BYTES) != 0) {
+    CTG_LOCK(CTG_WRITE, &pCatalog->tableCache.stableLock);
+    if (taosHashPut(pCatalog->tableCache.cache, output->tbFname, strlen(output->tbFname), output->tbMeta, tbSize) != 0) {
+      CTG_UNLOCK(CTG_WRITE, &pCatalog->tableCache.stableLock);
+      ctgError("push table[%s] to table cache failed", output->tbFname);
+      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
+    }
+
+    STableMeta *tbMeta = taosHashGet(pCatalog->tableCache.cache, output->tbFname, strlen(output->tbFname));
+    if (taosHashPut(pCatalog->tableCache.stableCache, &output->tbMeta->suid, sizeof(output->tbMeta->suid), &tbMeta, POINTER_BYTES) != 0) {
+      CTG_UNLOCK(CTG_WRITE, &pCatalog->tableCache.stableLock);
       ctgError("push suid[%"PRIu64"] to stable cache failed", output->tbMeta->suid);
-      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
+      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
+    }
+    CTG_UNLOCK(CTG_WRITE, &pCatalog->tableCache.stableLock);
+  } else {
+    if (taosHashPut(pCatalog->tableCache.cache, output->tbFname, strlen(output->tbFname), output->tbMeta, tbSize) != 0) {
+      ctgError("push table[%s] to table cache failed", output->tbFname);
+      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
     }
   }
+
+_return:
+  tfree(output->tbMeta);
   
-  return TSDB_CODE_SUCCESS;
+  CTG_RET(code);
 }
 
 
-int32_t ctgGetDBVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* dbName, int32_t forceUpdate, SDBVgroupInfo* dbInfo) {
-  if (NULL == pCatalog || NULL == dbName || NULL == pRpc || NULL == pMgmtEps) {
-    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
-  }
-
-  int32_t exist = 0;
-
+int32_t ctgGetDBVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* dbName, int32_t forceUpdate, SDBVgroupInfo** dbInfo) {
+  bool inCache = false;
   if (0 == forceUpdate) {
-    CTG_ERR_RET(ctgGetDBVgroupFromCache(pCatalog, dbName, dbInfo, &exist));
+    CTG_ERR_RET(ctgGetDBVgroupFromCache(pCatalog, dbName, dbInfo, &inCache));
 
-    if (exist) {
+    if (inCache) {
       return TSDB_CODE_SUCCESS;
     }
   }
@@ -397,9 +438,7 @@ int32_t ctgGetDBVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgm
 
   CTG_ERR_RET(catalogUpdateDBVgroup(pCatalog, dbName, &DbOut.dbVgroup));
 
-  if (dbInfo) {
-    *dbInfo = DbOut.dbVgroup;
-  }
+  CTG_ERR_RET(ctgGetDBVgroupFromCache(pCatalog, dbName, dbInfo, &inCache));
 
   return TSDB_CODE_SUCCESS;
 }
@@ -479,17 +518,68 @@ int32_t catalogGetDBVgroupVersion(struct SCatalog* pCatalog, const char* dbName,
     return TSDB_CODE_SUCCESS;
   }
 
-  SDBVgroupInfo * dbInfo = taosHashGet(pCatalog->dbCache.cache, dbName, strlen(dbName));
+  SDBVgroupInfo * dbInfo = taosHashAcquire(pCatalog->dbCache.cache, dbName, strlen(dbName));
   if (NULL == dbInfo) {
     *version = CTG_DEFAULT_INVALID_VERSION;
     return TSDB_CODE_SUCCESS;
   }
 
   *version = dbInfo->vgVersion;
+  taosHashRelease(pCatalog->dbCache.cache, dbInfo);
 
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t catalogGetDBVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* dbName, int32_t forceUpdate, SArray** vgroupList) {
+  if (NULL == pCatalog || NULL == dbName || NULL == pRpc || NULL == pMgmtEps || NULL == vgroupList) {
+    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
+  }
+
+  SDBVgroupInfo* db = NULL;
+  int32_t code = 0;
+  SVgroupInfo *vgInfo = NULL;
+  SArray *vgList = NULL;
+  
+  CTG_ERR_JRET(ctgGetDBVgroup(pCatalog, pRpc, pMgmtEps, dbName, forceUpdate, &db));
+
+  vgList = taosArrayInit(taosHashGetSize(db->vgInfo), sizeof(SVgroupInfo));
+  if (NULL == vgList) {
+    ctgError("taosArrayInit failed");
+    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);    
+  }
+
+  void *pIter = taosHashIterate(db->vgInfo, NULL);
+  while (pIter) {
+    vgInfo = pIter;
+
+    if (NULL == taosArrayPush(vgList, vgInfo)) {
+      ctgError("taosArrayPush failed");
+      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
+    }
+    
+    pIter = taosHashIterate(db->vgInfo, pIter);
+    vgInfo = NULL;
+  }
+
+  *vgroupList = vgList;
+  vgList = NULL;
+
+_return:
+
+  if (db) {
+    CTG_UNLOCK(CTG_READ, &db->lock);
+    taosHashRelease(pCatalog->dbCache.cache, db);
+  }
+
+  if (vgList) {
+    taosArrayDestroy(vgList);
+    vgList = NULL;
+  }
+
+  CTG_RET(code);  
+}
+
+
 int32_t catalogUpdateDBVgroup(struct SCatalog* pCatalog, const char* dbName, SDBVgroupInfo* dbInfo) {
   if (NULL == pCatalog || NULL == dbName || NULL == dbInfo) {
     CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
@@ -497,13 +587,17 @@ int32_t catalogUpdateDBVgroup(struct SCatalog* pCatalog, const char* dbName, SDB
 
   if (dbInfo->vgVersion < 0) {
     if (pCatalog->dbCache.cache) {
-      SDBVgroupInfo *oldInfo = taosHashGet(pCatalog->dbCache.cache, dbName, strlen(dbName));
-      if (oldInfo && oldInfo->vgInfo) {
-        taosHashCleanup(oldInfo->vgInfo);
-        oldInfo->vgInfo = NULL;
+      SDBVgroupInfo *oldInfo = taosHashAcquire(pCatalog->dbCache.cache, dbName, strlen(dbName));
+      if (oldInfo) {
+        CTG_LOCK(CTG_WRITE, &oldInfo->lock);
+        if (oldInfo->vgInfo) {
+          taosHashCleanup(oldInfo->vgInfo);
+          oldInfo->vgInfo = NULL;
+        }
+        CTG_UNLOCK(CTG_WRITE, &oldInfo->lock);
+
+        taosHashRelease(pCatalog->dbCache.cache, oldInfo);
       }
-
-      taosHashRemove(pCatalog->dbCache.cache, dbName, strlen(dbName));
     }
     
     ctgWarn("remove db [%s] from cache", dbName);
@@ -517,10 +611,16 @@ int32_t catalogUpdateDBVgroup(struct SCatalog* pCatalog, const char* dbName, SDB
       CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
     }
   } else {
-    SDBVgroupInfo *oldInfo = taosHashGet(pCatalog->dbCache.cache, dbName, strlen(dbName));
-    if (oldInfo && oldInfo->vgInfo) {
-      taosHashCleanup(oldInfo->vgInfo);
-      oldInfo->vgInfo = NULL;
+    SDBVgroupInfo *oldInfo = taosHashAcquire(pCatalog->dbCache.cache, dbName, strlen(dbName));
+    if (oldInfo) {
+      CTG_LOCK(CTG_WRITE, &oldInfo->lock);
+      if (oldInfo->vgInfo) {
+        taosHashCleanup(oldInfo->vgInfo);
+        oldInfo->vgInfo = NULL;
+      }
+      CTG_UNLOCK(CTG_WRITE, &oldInfo->lock);
+    
+      taosHashRelease(pCatalog->dbCache.cache, oldInfo);
     }
   }
 
@@ -573,7 +673,10 @@ int32_t catalogGetTableDistVgroup(struct SCatalog* pCatalog, void *pRpc, const S
   STableMeta *tbMeta = NULL;
   int32_t code = 0;
   SVgroupInfo vgroupInfo = {0};
-  SDBVgroupInfo dbVgroup = {0};
+  SDBVgroupInfo* dbVgroup = NULL;
+  SArray *vgList = NULL;
+
+  *pVgroupList = NULL;
   
   CTG_ERR_JRET(catalogGetTableMeta(pCatalog, pRpc, pMgmtEps, pTableName, &tbMeta));
 
@@ -582,38 +685,48 @@ int32_t catalogGetTableDistVgroup(struct SCatalog* pCatalog, void *pRpc, const S
   CTG_ERR_JRET(ctgGetDBVgroup(pCatalog, pRpc, pMgmtEps, db, false, &dbVgroup));
 
   if (tbMeta->tableType == TSDB_SUPER_TABLE) {
-    CTG_ERR_JRET(ctgGetVgInfoFromDB(pCatalog, pRpc, pMgmtEps, &dbVgroup, pVgroupList));
+    CTG_ERR_JRET(ctgGetVgInfoFromDB(pCatalog, pRpc, pMgmtEps, dbVgroup, pVgroupList));
   } else {
     int32_t vgId = tbMeta->vgId;
-    if (NULL == taosHashGetClone(dbVgroup.vgInfo, &vgId, sizeof(vgId), &vgroupInfo)) {
+    if (NULL == taosHashGetClone(dbVgroup->vgInfo, &vgId, sizeof(vgId), &vgroupInfo)) {
       ctgError("vgId[%d] not found in vgroup list", vgId);
       CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);    
     }
 
-    *pVgroupList = taosArrayInit(1, sizeof(SVgroupInfo));
-    if (NULL == *pVgroupList) {
+    vgList = taosArrayInit(1, sizeof(SVgroupInfo));
+    if (NULL == vgList) {
       ctgError("taosArrayInit failed");
       CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);    
     }
 
-    if (NULL == taosArrayPush(*pVgroupList, &vgroupInfo)) {
+    if (NULL == taosArrayPush(vgList, &vgroupInfo)) {
       ctgError("push vgroupInfo to array failed");
       CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
     }
-  }
 
-  tfree(tbMeta);
-  return TSDB_CODE_SUCCESS;
+    *pVgroupList = vgList;
+    vgList = NULL;
+  }
 
 _return:
   tfree(tbMeta);
-  taosArrayDestroy(*pVgroupList);
+
+  if (dbVgroup) {
+    CTG_UNLOCK(CTG_READ, &dbVgroup->lock);
+    taosHashRelease(pCatalog->dbCache.cache, dbVgroup);
+  }
+
+  if (vgList) {
+    taosArrayDestroy(vgList);
+    vgList = NULL;
+  }
+  
   CTG_RET(code);
 }
 
 
 int32_t catalogGetTableHashVgroup(struct SCatalog *pCatalog, void *pTransporter, const SEpSet *pMgmtEps, const SName *pTableName, SVgroupInfo *pVgroup) {
-  SDBVgroupInfo dbInfo = {0};
+  SDBVgroupInfo* dbInfo = NULL;
   int32_t code = 0;
 
   char db[TSDB_DB_FNAME_LEN] = {0};
@@ -621,12 +734,14 @@ int32_t catalogGetTableHashVgroup(struct SCatalog *pCatalog, void *pTransporter,
 
   CTG_ERR_RET(ctgGetDBVgroup(pCatalog, pTransporter, pMgmtEps, db, false, &dbInfo));
 
-  if (dbInfo.vgVersion < 0 || NULL == dbInfo.vgInfo) {
-    ctgError("db[%s] vgroup cache invalid, vgroup version:%d, vgInfo:%p", db, dbInfo.vgVersion, dbInfo.vgInfo);
-    CTG_ERR_RET(TSDB_CODE_TSC_DB_NOT_SELECTED);
-  }
+  CTG_ERR_JRET(ctgGetVgInfoFromHashValue(dbInfo, pTableName, pVgroup));
 
-  CTG_ERR_RET(ctgGetVgInfoFromHashValue(&dbInfo, pTableName, pVgroup));
+_return:
+
+  if (dbInfo) {
+    CTG_UNLOCK(CTG_READ, &dbInfo->lock);  
+    taosHashRelease(pCatalog->dbCache.cache, dbInfo);
+  }
 
   CTG_RET(code);
 }

source/libs/qworker/inc/qworkerInt.h

@@ -106,11 +106,11 @@ typedef struct SQWorkerMgmt {
   if (QW_READ == (type)) {          \
     if ((*(_lock)) < 0) assert(0);    \
     taosRLockLatch(_lock);          \
-    qDebug("RLOCK%p, %s:%d", (_lock), __FILE__, __LINE__); \
+    qDebug("QW RLOCK%p, %s:%d", (_lock), __FILE__, __LINE__); \
   } else {                                                \
     if ((*(_lock)) < 0) assert(0);                          \
     taosWLockLatch(_lock);                                \
-    qDebug("WLOCK%p, %s:%d", (_lock), __FILE__, __LINE__);  \
+    qDebug("QW WLOCK%p, %s:%d", (_lock), __FILE__, __LINE__);  \
   }                                                       \
 } while (0)
 
@@ -118,11 +118,11 @@ typedef struct SQWorkerMgmt {
   if (QW_READ == (type)) {                                \
     if ((*(_lock)) <= 0) assert(0);                         \
     taosRUnLockLatch(_lock);                              \
-    qDebug("RULOCK%p, %s:%d", (_lock), __FILE__, __LINE__); \
+    qDebug("QW RULOCK%p, %s:%d", (_lock), __FILE__, __LINE__); \
   } else {                                                \
     if ((*(_lock)) <= 0) assert(0);                         \
     taosWUnLockLatch(_lock);                              \
-    qDebug("WULOCK%p, %s:%d", (_lock), __FILE__, __LINE__); \
+    qDebug("QW WULOCK%p, %s:%d", (_lock), __FILE__, __LINE__); \
   }                                                       \
 } while (0)
 

source/util/CMakeLists.txt

@@ -12,4 +12,6 @@ target_link_libraries(
     PUBLIC zlib
     PUBLIC lz4_static
     PUBLIC api
-)
+)
\ No newline at end of file
+
+ADD_SUBDIRECTORY(test)

source/util/src/thash.c

@@ -362,7 +362,7 @@ void* taosHashGetCloneExt(SHashObj *pHashObj, const void *key, size_t keyLen, vo
   return data;
 }
 
-void* taosHashGetClone(SHashObj *pHashObj, const void *key, size_t keyLen, void* d) {
+void* taosHashGetCloneImpl(SHashObj *pHashObj, const void *key, size_t keyLen, void* d, bool acquire) {
   if (taosHashTableEmpty(pHashObj) || keyLen == 0 || key == NULL) {
     return NULL;
   }
@@ -404,6 +404,10 @@ void* taosHashGetClone(SHashObj *pHashObj, const void *key, size_t keyLen, void*
       memcpy(d, GET_HASH_NODE_DATA(pNode), pNode->dataLen);
     }
 
+    if (acquire) {
+      pNode->count++;
+    }
+
     data = GET_HASH_NODE_DATA(pNode);
   }
 
@@ -415,6 +419,15 @@ void* taosHashGetClone(SHashObj *pHashObj, const void *key, size_t keyLen, void*
   return data;
 }
 
+void* taosHashGetClone(SHashObj *pHashObj, const void *key, size_t keyLen, void* d) {
+  return taosHashGetCloneImpl(pHashObj, key, keyLen, d, false);
+}
+
+void* taosHashAcquire(SHashObj *pHashObj, const void *key, size_t keyLen) {
+  return taosHashGetCloneImpl(pHashObj, key, keyLen, NULL, true);
+}
+
+
 int32_t taosHashRemove(SHashObj *pHashObj, const void *key, size_t keyLen/*, void *data, size_t dsize*/) {
   if (pHashObj == NULL || taosHashTableEmpty(pHashObj)) {
     return -1;
@@ -919,3 +932,9 @@ void taosHashCancelIterate(SHashObj *pHashObj, void *p) {
 
   __rd_unlock(&pHashObj->lock, pHashObj->type);
 }
+
+void taosHashRelease(SHashObj *pHashObj, void *p) {
+  taosHashCancelIterate(pHashObj, p);
+}
+
+

source/util/test/CMakeLists.txt

@@ -13,17 +13,22 @@ IF (HEADER_GTEST_INCLUDE_DIR AND (LIB_GTEST_STATIC_DIR OR LIB_GTEST_SHARED_DIR))
 
     LIST(REMOVE_ITEM SOURCE_LIST ${CMAKE_CURRENT_SOURCE_DIR}/trefTest.c)
     ADD_EXECUTABLE(utilTest ${SOURCE_LIST})
-    TARGET_LINK_LIBRARIES(utilTest tutil common os gtest pthread gcov)
+    TARGET_LINK_LIBRARIES(utilTest util common os gtest pthread gcov)
 
+    LIST(REMOVE_ITEM SOURCE_LIST ${CMAKE_CURRENT_SOURCE_DIR}/cacheTest.cpp)
+    LIST(APPEND SOURCE_LIST ${CMAKE_CURRENT_SOURCE_DIR}/hashTest.cpp)
+    ADD_EXECUTABLE(hashTest ${SOURCE_LIST})
+    TARGET_LINK_LIBRARIES(hashTest util common os gtest pthread gcov)
+    
     LIST(APPEND BIN_SRC ${CMAKE_CURRENT_SOURCE_DIR}/trefTest.c)
     ADD_EXECUTABLE(trefTest ${BIN_SRC})
-    TARGET_LINK_LIBRARIES(trefTest common tutil)
+    TARGET_LINK_LIBRARIES(trefTest common util)
 
 ENDIF()
 
 #IF (TD_LINUX)
 #    ADD_EXECUTABLE(trefTest ./trefTest.c)
-#    TARGET_LINK_LIBRARIES(trefTest tutil common)
+#    TARGET_LINK_LIBRARIES(trefTest util common)
 #ENDIF ()
 
 INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)

source/util/test/hashTest.cpp

@@ -4,10 +4,15 @@
 #include <taosdef.h>
 #include <iostream>
 
-#include "hash.h"
+#include "thash.h"
 #include "taos.h"
 
 namespace {
+
+typedef struct TESTSTRUCT {
+    char *p;
+}TESTSTRUCT;
+
 // the simple test code for basic operations
 void simpleTest() {
   SHashObj* hashTable = (SHashObj*) taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);
@@ -141,6 +146,52 @@ void invalidOperationTest() {
 
 }
 
+void acquireRleaseTest() {
+  SHashObj* hashTable = (SHashObj*) taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
+  ASSERT_EQ(taosHashGetSize(hashTable), 0);
+
+  int32_t key = 2;
+  int32_t code = 0;
+  int32_t num = 0;
+  TESTSTRUCT data = {0};
+  char *str1 = "abcdefg";
+  char *str2 = "aaaaaaa";
+  char *str3 = "123456789";
+
+  data.p = (char *)malloc(10);
+  strcpy(data.p, str1);
+
+  code = taosHashPut(hashTable, &key, sizeof(key), &data, sizeof(data));
+  ASSERT_EQ(code, 0);
+
+  TESTSTRUCT* pdata = (TESTSTRUCT*)taosHashAcquire(hashTable, &key, sizeof(key));
+  ASSERT_TRUE(pdata != nullptr);
+  ASSERT_TRUE(strcmp(pdata->p, str1) == 0);
+  
+  code = taosHashRemove(hashTable, &key, sizeof(key));
+  ASSERT_EQ(code, 0);
+  ASSERT_TRUE(strcmp(pdata->p, str1) == 0);
+
+  num = taosHashGetSize(hashTable);
+  ASSERT_EQ(num, 1);
+  
+  strcpy(pdata->p, str3);
+
+  data.p = (char *)malloc(10);
+  strcpy(data.p, str2);
+  code = taosHashPut(hashTable, &key, sizeof(key), &data, sizeof(data));
+  ASSERT_EQ(code, 0);
+  num = taosHashGetSize(hashTable);
+  ASSERT_EQ(num, 2);
+
+  printf("%s,expect:%s", pdata->p, str3);
+  ASSERT_TRUE(strcmp(pdata->p, str3) == 0);
+  
+  taosHashRelease(hashTable, pdata);
+  num = taosHashGetSize(hashTable);
+  ASSERT_EQ(num, 1);
+}
+
 }
 
 int main(int argc, char** argv) {
@@ -153,4 +204,5 @@ TEST(testCase, hashTest) {
   stringKeyTest();
   noLockPerformanceTest();
   multithreadsTest();
+  acquireRleaseTest();
 }

tests/script/general/db/basic1.sim

@@ -59,9 +59,29 @@ if $data03 != 0 then
 endi
 
 print =============== show vgroups
+sql show databases
+
+if $rows != 1 then 
+  return -1
+endi
+
+sql_error use d1
+
 sql use d4
+sql show vgroups
+
+if $rows != 2 then
+  return -1
+endi
+
+print =============== show dnodes
+sql show dnodes
+
+if $data00 != 1 then
+  return -1
+endi
 
-if $rows == 0 then 
+if $data02 != 2 then
   return -1
 endi
 

tests/script/general/table/basic1.sim

@@ -23,7 +23,7 @@ endi
 
 print $data00 $data01 $data02
 
-sql create table st2 (ts timestamp, i float) tags (j bigint)
+sql create table st2 (ts timestamp, i float) tags (j int)
 sql show stables
 if $rows != 2 then 
   return -1
@@ -39,15 +39,14 @@ if $rows != 1 then
   return -1
 endi
 
-print -->
 print $data00 $data01 $data02
 print $data10 $data11 $data12
 
-return
-
 print =============== create child table
 sql create table c1 using st tags(1)
 sql create table c2 using st tags(2)
+
+return 
 sql show tables
 if $rows != 2 then 
   return -1

tests/script/sh/exec.sh

@@ -17,7 +17,7 @@ OS_TYPE=`$UNAME_BIN`
 NODE_NAME=
 EXEC_OPTON=
 CLEAR_OPTION="false"
-while getopts "n:s:u:x:ct" arg
+while getopts "n:s:u:x:cv" arg
 do
   case $arg in
     n)
@@ -29,7 +29,7 @@ do
     c)
       CLEAR_OPTION="clear"
       ;;
-    t)
+    v)
       SHELL_OPTION="true"
       ;;
     u)

tests/script/tmp/dnodes.sim

@@ -2,20 +2,19 @@ system sh/stop_dnodes.sh
 
 
 ############## config parameter  #####################
-$node1 = 192.168.101.174 
+$node1 = 192.168.0.201 
 $node2 = 192.168.0.202
-$node2 = 192.168.0.203
-$node3 = 192.168.0.204
+$node3 = 192.168.0.203
+$node4 = 192.168.0.204
 
-$first = 1
-$num = 5
 $self = $node1
+$num = 25
 
 ###############  deploy firstEp  #####################
 
 $firstEp = $node1 . :7100
 $firstPort = 7100
-if $first == 1 then 
+if $self == $node1 then 
   system sh/deploy.sh -n dnode1 -i 1
   system sh/cfg.sh -n dnode1 -c firstEp -v $firstEp
   system sh/cfg.sh -n dnode1 -c secondEp -v $firstEp
@@ -28,7 +27,7 @@ if $first == 1 then
   $i = 0
   while $i < $num
     $port = $i * 100
-    $port = $port + 8000
+    $port = $port + 8100
     $i = $i + 1
     sql create dnode $node1 port $port
   endw
@@ -36,7 +35,7 @@ if $first == 1 then
   $i = 0
   while $i < $num
     $port = $i * 100
-    $port = $port + 8000
+    $port = $port + 8100
     $i = $i + 1
     sql create dnode $node2 port $port
   endw
@@ -44,7 +43,7 @@ if $first == 1 then
   $i = 0
   while $i < $num
     $port = $i * 100
-    $port = $port + 8000
+    $port = $port + 8100
     $i = $i + 1
     sql create dnode $node3 port $port
   endw
@@ -52,7 +51,7 @@ if $first == 1 then
   $i = 0
   while $i < $num
     $port = $i * 100
-    $port = $port + 8000
+    $port = $port + 8100
     $i = $i + 1
     sql create dnode $node4 port $port
   endw
@@ -64,7 +63,7 @@ $i = 0
 while $i < $num
   $index = $i + 80
   $port = $i * 100
-  $port = $port + 8000
+  $port = $port + 8100
   $dnodename = dnode . $index
   $i = $i + 1 
 
@@ -74,5 +73,5 @@ while $i < $num
   system sh/cfg.sh -n $dnodename -c fqdn -v $self
   system sh/cfg.sh -n $dnodename -c serverPort -v $port
 
-  #system sh/exec.sh -n $dnodename -s start
+  system sh/exec.sh -n $dnodename -s start
 endw