Merge pull request #1412 from taosdata/refact/slguan

Refact/slguan

src/CMakeLists.txt

@@ -12,7 +12,6 @@ ADD_SUBDIRECTORY(client)
 ADD_SUBDIRECTORY(query)
 ADD_SUBDIRECTORY(kit)
 ADD_SUBDIRECTORY(plugins)
-ADD_SUBDIRECTORY(sdb)
 ADD_SUBDIRECTORY(mnode)
 ADD_SUBDIRECTORY(vnode)
 ADD_SUBDIRECTORY(dnode)

src/dnode/CMakeLists.txt

@@ -13,7 +13,7 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
   AUX_SOURCE_DIRECTORY(src SRC)
 
   ADD_EXECUTABLE(taosd ${SRC})
-  TARGET_LINK_LIBRARIES(taosd mnode sdb taos_static monitor http tsdb)
+  TARGET_LINK_LIBRARIES(taosd mnode taos_static monitor http tsdb)
 
   #IF (TD_CLUSTER)
   #  TARGET_LINK_LIBRARIES(taosd dcluster)

src/inc/mnode.h

@@ -25,7 +25,6 @@ extern "C" {
 #include "taosdef.h"
 #include "taosmsg.h"
 #include "taoserror.h"
-#include "sdb.h"
 #include "tglobalcfg.h"
 #include "thash.h"
 #include "tidpool.h"
@@ -54,8 +53,8 @@ typedef struct {
   int8_t   numOfMnodes;
   int32_t  numOfDnodes;
   char     mnodeName[TSDB_DNODE_NAME_LEN + 1];
-  char     reserved[7];
-  char     updateEnd[1];
+  int8_t   reserved[15];
+  int8_t   updateEnd[1];
   int      syncFd;
   void    *hbTimer;
   void    *pSync;
@@ -79,8 +78,8 @@ typedef struct {
   float      lbScore;          // calc in balance function
   int32_t    customScore;      // config by user
   char       dnodeName[TSDB_DNODE_NAME_LEN + 1];
-  char       reserved[7];
-  char       updateEnd[1];
+  int8_t     reserved[15];
+  int8_t     updateEnd[1];
   SVnodeLoad vload[TSDB_MAX_VNODES];
   int32_t    status;
   uint32_t   lastReboot;       // time stamp for last reboot
@@ -121,7 +120,7 @@ typedef struct SSuperTableObj {
   int32_t  sversion;
   int32_t  numOfColumns;
   int32_t  numOfTags;
-  int8_t   reserved[5];
+  int8_t   reserved[15];
   int8_t   updateEnd[1];
   int32_t  numOfTables;
   int16_t  nextColId;
@@ -153,26 +152,29 @@ typedef struct {
   int32_t  sversion;
   int32_t  numOfColumns;
   int32_t  sqlLen;
-  int8_t   reserved[3];
+  int8_t   reserved[7];
   int8_t   updateEnd[1];
   char*    sql;  //null-terminated string
   int16_t  nextColId;
   SSchema* schema;
 } SNormalTableObj;
 
+struct _db_obj;
+
 typedef struct _vg_obj {
   uint32_t        vgId;
   char            dbName[TSDB_DB_NAME_LEN + 1];
   int64_t         createdTime;
   SVnodeGid       vnodeGid[TSDB_VNODES_SUPPORT];
   int32_t         numOfVnodes;
-  int32_t         numOfTables;
   int32_t         lbIp;
   int32_t         lbTime;
   int8_t          lbStatus;
   int8_t          reserved[14];
   int8_t          updateEnd[1];
   struct _vg_obj *prev, *next;
+  struct _db_obj *pDb;
+  int32_t         numOfTables;
   void *          idPool;
   STableInfo **   tableList;
 } SVgObj;
@@ -182,8 +184,8 @@ typedef struct _db_obj {
   int8_t  dirty;
   int64_t createdTime;
   SDbCfg  cfg;
-  char    reserved[15];
-  char    updateEnd[1];
+  int8_t  reserved[15];
+  int8_t  updateEnd[1];
   struct _db_obj *prev, *next;
   int32_t numOfVgroups;
   int32_t numOfTables;
@@ -201,7 +203,7 @@ typedef struct _user_obj {
   int64_t           createdTime;
   int8_t            superAuth;
   int8_t            writeAuth;
-  int8_t            reserved[16];
+  int8_t            reserved[13];
   int8_t            updateEnd[1];
   struct _user_obj *prev, *next;
   struct _acctObj * pAcct;

src/inc/sdb.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 TDENGINE_SDB_H
-#define TDENGINE_SDB_H
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-#include "taosmsg.h"
-#include "taosdef.h"
-
-extern uint16_t    tsMgmtMgmtPort;
-extern uint16_t    tsMgmtSyncPort;
-extern int      tsMgmtPeerHBTimer;  // seconds
-extern char *   sdbStatusStr[];
-extern char *   sdbRoleStr[];
-extern int      sdbMaster;
-extern SRpcIpSet *pSdbIpList;
-extern SRpcIpSet *pSdbPublicIpList;
-
-extern void (*sdbWorkAsMasterCallback)();  // this function pointer will be set by taosd
-
-enum _keytype {
-  SDB_KEYTYPE_STRING, SDB_KEYTYPE_UINT32, SDB_KEYTYPE_AUTO, SDB_KEYTYPE_RECYCLE, SDB_KEYTYPE_MAX
-};
-
-#define SDB_ROLE_UNAPPROVED 0
-#define SDB_ROLE_UNDECIDED  1
-#define SDB_ROLE_MASTER     2
-#define SDB_ROLE_SLAVE      3
-
-#define SDB_STATUS_OFFLINE  0
-#define SDB_STATUS_UNSYNCED 1
-#define SDB_STATUS_SYNCING  2
-#define SDB_STATUS_SERVING  3
-#define SDB_STATUS_DELETED  4
-
-enum _sdbaction {
-  SDB_TYPE_INSERT,
-  SDB_TYPE_DELETE,
-  SDB_TYPE_UPDATE,
-  SDB_TYPE_DECODE,
-  SDB_TYPE_ENCODE,
-  SDB_TYPE_BEFORE_BATCH_UPDATE,
-  SDB_TYPE_BATCH_UPDATE,
-  SDB_TYPE_AFTER_BATCH_UPDATE,
-  SDB_TYPE_RESET,
-  SDB_TYPE_DESTROY,
-  SDB_MAX_ACTION_TYPES
-};
-
-#define SDB_MAX_PEERS 4
-typedef struct {
-  uint32_t ip;
-  uint32_t publicIp;
-  char     ipstr[20];
-  char     zone[12];
-  char     role;
-  int64_t  createdTime;
-  uint64_t dbVersion;
-  int64_t  lostTime;
-  char     status;
-  char     numOfMnodes;
-  int      numOfDnodes;
-  char     updateEnd[1];
-
-  // internal
-  int   syncFd;
-  void *hbTimer;
-  void *pSync;
-} SSdbPeer;
-
-extern SSdbPeer *sdbPeer[];
-#define sdbInited (sdbPeer[0])
-#define sdbStatus (sdbPeer[0]->status)
-
-void *sdbOpenTable(int maxRows, int32_t maxRowSize, char *name, uint8_t keyType, char *directory,
-                   void *(*appTool)(char, void *, char *, int, int *));
-
-void *sdbGetRow(void *handle, void *key);
-
-int64_t sdbInsertRow(void *handle, void *row, int rowSize);
-
-int sdbDeleteRow(void *handle, void *key);
-
-int sdbUpdateRow(void *handle, void *row, int updateSize, char isUpdated);
-
-void *sdbFetchRow(void *handle, void *pNode, void **ppRow);
-
-int sdbBatchUpdateRow(void *handle, void *row, int rowSize);
-
-int64_t sdbGetId(void *handle);
-
-int64_t sdbGetNumOfRows(void *handle);
-
-void sdbSaveSnapShot(void *handle);
-
-void sdbCloseTable(void *handle);
-
-int sdbRemovePeerByIp(uint32_t ip);
-
-int sdbInitPeers(char *directory);
-
-void sdbCleanUpPeers();
-
-int64_t sdbGetVersion();
-
-int32_t sdbGetRunStatus();
-
-#define TSDB_MAX_NORMAL_TABLES 10000
-#define TSDB_MAX_SUPER_TABLES  1000
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif  // TDENGINE_SDB_H

src/inc/taosdef.h

@@ -309,6 +309,16 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
 #define TSDB_SESSIONS_PER_VNODE (300)
 #define TSDB_SESSIONS_PER_DNODE (TSDB_SESSIONS_PER_VNODE * TSDB_MAX_VNODES)
 
+#define TSDB_MAX_MNODES        5
+#define TSDB_MAX_DNODES        10
+#define TSDB_MAX_ACCOUNTS      10
+#define TSDB_MAX_USERS         20
+#define TSDB_MAX_DBS           100
+#define TSDB_MAX_VGROUPS       1000
+#define TSDB_MAX_SUPER_TABLES  100
+#define TSDB_MAX_NORMAL_TABLES 1000
+#define TSDB_MAX_CHILD_TABLES  100000
+
 enum {
   TSDB_PRECISION_MILLI,
   TSDB_PRECISION_MICRO,

src/kit/CMakeLists.txt

@@ -2,5 +2,5 @@ CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
 PROJECT(TDengine)
 
 ADD_SUBDIRECTORY(shell)
-ADD_SUBDIRECTORY(taosdemo)
-ADD_SUBDIRECTORY(taosdump)
+#ADD_SUBDIRECTORY(taosdemo)
+#ADD_SUBDIRECTORY(taosdump)

src/mnode/CMakeLists.txt

@@ -13,7 +13,7 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
   AUX_SOURCE_DIRECTORY(src SRC)
 
   ADD_LIBRARY(mnode ${SRC})
-  TARGET_LINK_LIBRARIES(mnode trpc tutil sdb pthread)
+  TARGET_LINK_LIBRARIES(mnode trpc tutil pthread)
 
   IF (TD_CLUSTER)
     TARGET_LINK_LIBRARIES(mnode)

src/mnode/inc/mgmtChildTable.h

@@ -39,6 +39,7 @@ int32_t mgmtModifyChildTableTagValueByName(SChildTableObj *pTable, char *tagName
 int32_t mgmtGetChildTableMeta(SDbObj *pDb, SChildTableObj *pTable, STableMetaMsg *pMeta, bool usePublicIp);
 
 void mgmtDropAllChildTables(SDbObj *pDropDb);
+void mgmtDropAllChildTablesInStable(SSuperTableObj *pStable);
 
 #ifdef __cplusplus
 }

src/mnode/inc/mgmtDb.h

@@ -22,11 +22,11 @@ extern "C" {
 
 #include "mnode.h"
 
-int32_t mgmtAddVgroupIntoDb(SDbObj *pDb, SVgObj *pVgroup);
-int32_t mgmtAddVgroupIntoDbTail(SDbObj *pDb, SVgObj *pVgroup);
-int32_t mgmtRemoveVgroupFromDb(SDbObj *pDb, SVgObj *pVgroup);
-int32_t mgmtMoveVgroupToTail(SDbObj *pDb, SVgObj *pVgroup);
-int32_t mgmtMoveVgroupToHead(SDbObj *pDb, SVgObj *pVgroup);
+void mgmtAddVgroupIntoDb(SVgObj *pVgroup);
+void mgmtAddVgroupIntoDbTail(SVgObj *pVgroup);
+void mgmtRemoveVgroupFromDb(SVgObj *pVgroup);
+void mgmtMoveVgroupToTail(SVgObj *pVgroup);
+void mgmtMoveVgroupToHead(SVgObj *pVgroup);
 
 int32_t mgmtInitDbs();
 void    mgmtCleanUpDbs();

src/mnode/inc/mgmtMnode.h

@@ -20,9 +20,15 @@
 extern "C" {
 #endif
 
-bool mgmtCheckRedirect(void *handle);
+int32_t mgmtInitMnodes();
+void mgmtCleanupMnodes();
+
+bool mgmtInServerStatus();   
+bool mgmtIsMaster();
 
-void mgmtGetMnodeIpList(SRpcIpSet *ipSet);
+bool mgmtCheckRedirect(void *handle);
+void mgmtGetMnodePrivateIpList(SRpcIpSet *ipSet);
+void mgmtGetMnodePublicIpList(SRpcIpSet *ipSet);
 
 int32_t mgmtAddMnode(uint32_t privateIp, uint32_t publicIp);
 int32_t mgmtRemoveMnode(uint32_t privateIp);

src/mnode/inc/mgmtProfile.h

@@ -28,6 +28,8 @@ bool mgmtCheckQhandle(uint64_t qhandle);
 void mgmtSaveQhandle(void *qhandle);
 void mgmtFreeQhandle(void *qhandle);
 
+void mgmtFreeQueuedMsg(SQueuedMsg *pMsg);
+
 #ifdef __cplusplus
 }
 #endif

src/sdb/src/sdbstr.c → src/mnode/inc/mgmtSdb.h

@@ -12,47 +12,61 @@
  * 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 "sdbint.h"
-
-int32_t (*mpeerInitMnodesFp)(char *directory) = NULL;
-void    (*mpeerCleanUpMnodesFp)() = NULL;
-int32_t (*mpeerForwardRequestFp)(SSdbTable *pTable, char type, void *cont, int32_t contLen) = NULL;
-
-char *sdbStatusStr[] = {
-  "offline",
-  "unsynced",
-  "syncing",
-  "serving",
-  "null"
-};
-
-char *sdbRoleStr[] = {
-  "unauthed",
-  "undecided",
-  "master",
-  "slave",
-  "null"
-};
-
-int32_t sdbForwardDbReqToPeer(SSdbTable *pTable, char type, char *data, int32_t dataLen) {
-  if (mpeerForwardRequestFp) {
-    return mpeerForwardRequestFp(pTable, type, data, dataLen);
-  } else {
-    return 0;
-  }
-}
 
-int32_t sdbInitPeers(char *directory) {
-  if (mpeerInitMnodesFp) {
-    return (*mpeerInitMnodesFp)(directory);
-  } else {
-    return 0;
-  }
-}
+#ifndef TDENGINE_MNODE_SDB_H
+#define TDENGINE_MNODE_SDB_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef enum {
+  SDB_KEY_TYPE_STRING, 
+  SDB_KEY_TYPE_AUTO
+} ESdbKeyType;
+
+typedef enum {
+  SDB_OPER_TYPE_GLOBAL,
+  SDB_OPER_TYPE_LOCAL
+} ESdbOperType;
+
+typedef struct {
+  ESdbOperType type;
+  void *  table;
+  void *  pObj;
+  int64_t version;
+  int32_t maxRowSize;
+  int32_t rowSize;
+  void *  rowData;
+} SSdbOperDesc;
 
-void sdbCleanUpPeers() {
-  if (mpeerCleanUpMnodesFp) {
-    (*mpeerCleanUpMnodesFp)();
-  }
+typedef struct {
+  char   *tableName;
+  int32_t hashSessions;
+  int32_t maxRowSize;
+  ESdbKeyType keyType;
+  int32_t (*insertFp)(SSdbOperDesc *pOper);
+  int32_t (*deleteFp)(SSdbOperDesc *pOper);
+  int32_t (*updateFp)(SSdbOperDesc *pOper);
+  int32_t (*encodeFp)(SSdbOperDesc *pOper);
+  int32_t (*decodeFp)(SSdbOperDesc *pDesc);  
+  int32_t (*destroyFp)(SSdbOperDesc *pDesc);
+} SSdbTableDesc;
+
+void *sdbOpenTable(SSdbTableDesc *desc);
+void sdbCloseTable(void *handle);
+
+int32_t sdbInsertRow(SSdbOperDesc *pOper);
+int32_t sdbDeleteRow(SSdbOperDesc *pOper);
+int32_t sdbUpdateRow(SSdbOperDesc *pOper);
+
+void *sdbGetRow(void *handle, void *key);
+void *sdbFetchRow(void *handle, void *pNode, void **ppRow);
+int64_t sdbGetNumOfRows(void *handle);
+uint64_t sdbGetVersion();
+
+#ifdef __cplusplus
 }
+#endif
+
+#endif
\ No newline at end of file

src/mnode/inc/mgmtVgroup.h

@@ -28,13 +28,12 @@ int32_t mgmtInitVgroups();
 void    mgmtCleanUpVgroups();
 SVgObj *mgmtGetVgroup(int32_t vgId);
 SVgObj *mgmtGetVgroupByVnode(uint32_t dnode, int32_t vnode);
+void    mgmtDropAllVgroups(SDbObj *pDropDb);
 
 void    mgmtCreateVgroup(SQueuedMsg *pMsg);
 void    mgmtDropVgroup(SVgObj *pVgroup, void *ahandle);
 void    mgmtUpdateVgroup(SVgObj *pVgroup);
-void    mgmtUpdateVgroupIp(SDnodeObj *pDnode);
-
-void    mgmtSetVgroupIdPool();
+void    mgmtAlterVgroup(SVgObj *pVgroup, void *ahandle);
 SVgObj *mgmtGetAvailableVgroup(SDbObj *pDb);
 
 void    mgmtAddTableIntoVgroup(SVgObj *pVgroup, STableInfo *pTable);

src/mnode/src/mgmtAcct.c

@@ -85,7 +85,9 @@ int32_t mgmtRemoveUserFromAcct(SAcctObj *pAcct, SUserObj *pUser) {
     pUser->prev->next = pUser->next;
   }
 
-  if (pUser->next) pUser->next->prev = pUser->prev;
+  if (pUser->next) {
+    pUser->next->prev = pUser->prev;
+  }
   
   if (pUser->prev == NULL) {
     pAcct->pUser = pUser->next;

src/mnode/src/mgmtChildTable.c

@@ -25,79 +25,47 @@
 #include "mgmtAcct.h"
 #include "mgmtChildTable.h"
 #include "mgmtDb.h"
+#include "mgmtDClient.h"
 #include "mgmtGrant.h"
+#include "mgmtMnode.h"
 #include "mgmtProfile.h"
+#include "mgmtSdb.h"
 #include "mgmtShell.h"
-#include "mgmtDClient.h"
 #include "mgmtSuperTable.h"
 #include "mgmtTable.h"
 #include "mgmtVgroup.h"
 
 void   *tsChildTableSdb;
 int32_t tsChildTableUpdateSize;
-void *(*mgmtChildTableActionFp[SDB_MAX_ACTION_TYPES])(void *row, char *str, int32_t size, int32_t *ssize);
-
-void *mgmtChildTableActionInsert(void *row, char *str, int32_t size, int32_t *ssize);
-void *mgmtChildTableActionDelete(void *row, char *str, int32_t size, int32_t *ssize);
-void *mgmtChildTableActionUpdate(void *row, char *str, int32_t size, int32_t *ssize);
-void *mgmtChildTableActionEncode(void *row, char *str, int32_t size, int32_t *ssize);
-void *mgmtChildTableActionDecode(void *row, char *str, int32_t size, int32_t *ssize);
-void *mgmtChildTableActionReset(void *row, char *str, int32_t size, int32_t *ssize);
-void *mgmtChildTableActionDestroy(void *row, char *str, int32_t size, int32_t *ssize);
 
 static void mgmtDestroyChildTable(SChildTableObj *pTable) {
-  free(pTable);
-}
-
-static void mgmtChildTableActionInit() {
-  mgmtChildTableActionFp[SDB_TYPE_INSERT]  = mgmtChildTableActionInsert;
-  mgmtChildTableActionFp[SDB_TYPE_DELETE]  = mgmtChildTableActionDelete;
-  mgmtChildTableActionFp[SDB_TYPE_UPDATE]  = mgmtChildTableActionUpdate;
-  mgmtChildTableActionFp[SDB_TYPE_ENCODE]  = mgmtChildTableActionEncode;
-  mgmtChildTableActionFp[SDB_TYPE_DECODE]  = mgmtChildTableActionDecode;
-  mgmtChildTableActionFp[SDB_TYPE_RESET]   = mgmtChildTableActionReset;
-  mgmtChildTableActionFp[SDB_TYPE_DESTROY] = mgmtChildTableActionDestroy;
+  tfree(pTable);
 }
 
-void *mgmtChildTableActionReset(void *row, char *str, int32_t size, int32_t *ssize) {
-  SChildTableObj *pTable = (SChildTableObj *) row;
-  memcpy(pTable, str, tsChildTableUpdateSize);
-  return NULL;
-}
-
-void *mgmtChildTableActionDestroy(void *row, char *str, int32_t size, int32_t *ssize) {
-  SChildTableObj *pTable = (SChildTableObj *)row;
-  mgmtDestroyChildTable(pTable);
-  return NULL;
+static int32_t mgmtChildTableActionDestroy(SSdbOperDesc *pOper) {
+  mgmtDestroyChildTable(pOper->pObj);
+  return TSDB_CODE_SUCCESS;
 }
 
-void *mgmtChildTableActionInsert(void *row, char *str, int32_t size, int32_t *ssize) {
-  SChildTableObj *pTable = (SChildTableObj *) row;
+static int32_t mgmtChildTableActionInsert(SSdbOperDesc *pOper) {
+  SChildTableObj *pTable = pOper->pObj;
 
   SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
   if (pVgroup == NULL) {
     mError("ctable:%s, not in vgroup:%d", pTable->tableId, pTable->vgId);
-    return NULL;
+    return TSDB_CODE_INVALID_VGROUP_ID;
   }
 
   SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
   if (pDb == NULL) {
     mError("ctable:%s, vgroup:%d not in db:%s", pTable->tableId, pVgroup->vgId, pVgroup->dbName);
-    return NULL;
+    return TSDB_CODE_INVALID_DB;
   }
 
   SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
   if (pAcct == NULL) {
     mError("ctable:%s, account:%s not exists", pTable->tableId, pDb->cfg.acct);
-    return NULL;
-  }
-
-  if (!sdbMaster) {
-    int32_t sid = taosAllocateId(pVgroup->idPool);
-    if (sid != pTable->sid) {
-      mError("ctable:%s, sid:%d is not matched from the master:%d", pTable->tableId, sid, pTable->sid);
-      return NULL;
-    }
+    return TSDB_CODE_INVALID_ACCT;
   }
 
   pTable->superTable = mgmtGetSuperTable(pTable->superTableId);
@@ -107,83 +75,65 @@ void *mgmtChildTableActionInsert(void *row, char *str, int32_t size, int32_t *ss
   mgmtAddTableIntoDb(pDb);
   mgmtAddTableIntoVgroup(pVgroup, (STableInfo *) pTable);
 
-  if (pVgroup->numOfTables >= pDb->cfg.maxSessions - 1 && pDb->numOfVgroups > 1) {
-    mgmtMoveVgroupToTail(pDb, pVgroup);
-  }
-
-  return NULL;
+  return TSDB_CODE_SUCCESS;
 }
 
-void *mgmtChildTableActionDelete(void *row, char *str, int32_t size, int32_t *ssize) {
-  SChildTableObj *pTable = (SChildTableObj *) row;
+static int32_t mgmtChildTableActionDelete(SSdbOperDesc *pOper) {
+  SChildTableObj *pTable = pOper->pObj;
   if (pTable->vgId == 0) {
-    return NULL;
+    return TSDB_CODE_INVALID_VGROUP_ID;
   }
 
   SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
   if (pVgroup == NULL) {
-    return NULL;
+    return TSDB_CODE_INVALID_VGROUP_ID;
   }
 
   SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
   if (pDb == NULL) {
     mError("ctable:%s, vgroup:%d not in DB:%s", pTable->tableId, pVgroup->vgId, pVgroup->dbName);
-    return NULL;
+    return TSDB_CODE_INVALID_DB;
   }
 
   SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
   if (pAcct == NULL) {
     mError("ctable:%s, account:%s not exists", pTable->tableId, pDb->cfg.acct);
-    return NULL;
+    return TSDB_CODE_INVALID_ACCT;
   }
 
   mgmtRestoreTimeSeries(pAcct, pTable->superTable->numOfColumns - 1);
   mgmtRemoveTableFromDb(pDb);
   mgmtRemoveTableFromVgroup(pVgroup, (STableInfo *) pTable);
-
   mgmtRemoveTableFromSuperTable(pTable->superTable);
 
-  if (pVgroup->numOfTables > 0) {
-    mgmtMoveVgroupToHead(pDb, pVgroup);
-  }
-
-  return NULL;
+  return TSDB_CODE_SUCCESS;
 }
 
-void *mgmtChildTableActionUpdate(void *row, char *str, int32_t size, int32_t *ssize) {
-  return mgmtChildTableActionReset(row, str, size, NULL);
+static int32_t mgmtChildTableActionUpdate(SSdbOperDesc *pOper) {
+  return TSDB_CODE_SUCCESS;
 }
 
-void *mgmtChildTableActionEncode(void *row, char *str, int32_t size, int32_t *ssize) {
-  SChildTableObj *pTable = (SChildTableObj *) row;
-  assert(row != NULL && str != NULL);
+static int32_t mgmtChildTableActionEncode(SSdbOperDesc *pOper) {
+  SChildTableObj *pTable = pOper->pObj;
+  assert(pTable != NULL && pOper->rowData != NULL);
 
-  memcpy(str, pTable, tsChildTableUpdateSize);
-  *ssize = tsChildTableUpdateSize;
+  memcpy(pOper->rowData, pTable, tsChildTableUpdateSize);
+  pOper->rowSize = tsChildTableUpdateSize;
 
-  return NULL;
+  return TSDB_CODE_SUCCESS;
 }
 
-void *mgmtChildTableActionDecode(void *row, char *str, int32_t size, int32_t *ssize) {
-  assert(str != NULL);
+static int32_t mgmtChildTableActionDecode(SSdbOperDesc *pOper) {
+  assert(pOper->rowData != NULL);
 
-  SChildTableObj *pTable = (SChildTableObj *)calloc(sizeof(SChildTableObj), 1);
-  if (pTable == NULL) return NULL;
-
-  if (size < tsChildTableUpdateSize) {
-    mgmtDestroyChildTable(pTable);
-    return NULL;
+  pOper->pObj = calloc(1, sizeof(SChildTableObj));
+  if (pOper->pObj == NULL) {
+    return TSDB_CODE_SERV_OUT_OF_MEMORY;
   }
-  memcpy(pTable, str, tsChildTableUpdateSize);
 
-  return (void *)pTable;
-}
+  memcpy(pOper->pObj, pOper->rowData, tsChildTableUpdateSize);
 
-void *mgmtChildTableAction(char action, void *row, char *str, int32_t size, int32_t *ssize) {
-  if (mgmtChildTableActionFp[(uint8_t)action] != NULL) {
-    return (*(mgmtChildTableActionFp[(uint8_t)action]))(row, str, size, ssize);
-  }
-  return NULL;
+  return TSDB_CODE_SUCCESS;
 }
 
 int32_t mgmtInitChildTables() {
@@ -191,12 +141,23 @@ int32_t mgmtInitChildTables() {
   void *pLastNode = NULL;
   SChildTableObj *pTable = NULL;
 
-  mgmtChildTableActionInit();
   SChildTableObj tObj;
-  tsChildTableUpdateSize = tObj.updateEnd - (int8_t *)&tObj;
+  tsChildTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
+
+  SSdbTableDesc tableDesc = {
+    .tableName    = "ctables",
+    .hashSessions = tsMaxTables,
+    .maxRowSize   = tsChildTableUpdateSize,
+    .keyType      = SDB_KEY_TYPE_STRING,
+    .insertFp     = mgmtChildTableActionInsert,
+    .deleteFp     = mgmtChildTableActionDelete,
+    .updateFp     = mgmtChildTableActionUpdate,
+    .encodeFp     = mgmtChildTableActionEncode,
+    .decodeFp     = mgmtChildTableActionDecode,
+    .destroyFp    = mgmtChildTableActionDestroy,
+  };
 
-  tsChildTableSdb = sdbOpenTable(tsMaxTables, tsChildTableUpdateSize,
-                                 "ctables", SDB_KEYTYPE_STRING, tsMnodeDir, mgmtChildTableAction);
+  tsChildTableSdb = sdbOpenTable(&tableDesc);
   if (tsChildTableSdb == NULL) {
     mError("failed to init child table data");
     return -1;
@@ -212,7 +173,11 @@ int32_t mgmtInitChildTables() {
     SDbObj *pDb = mgmtGetDbByTableId(pTable->tableId);
     if (pDb == NULL) {
       mError("ctable:%s, failed to get db, discard it", pTable->tableId);
-      sdbDeleteRow(tsChildTableSdb, pTable);
+      SSdbOperDesc desc = {0};
+      desc.type = SDB_OPER_TYPE_LOCAL;
+      desc.pObj = pTable;
+      desc.table = tsChildTableSdb;
+      sdbDeleteRow(&desc);
       pNode = pLastNode;
       continue;
     }
@@ -221,7 +186,11 @@ int32_t mgmtInitChildTables() {
     if (pVgroup == NULL) {
       mError("ctable:%s, failed to get vgroup:%d sid:%d, discard it", pTable->tableId, pTable->vgId, pTable->sid);
       pTable->vgId = 0;
-      sdbDeleteRow(tsChildTableSdb, pTable);
+      SSdbOperDesc desc = {0};
+      desc.type = SDB_OPER_TYPE_LOCAL;
+      desc.pObj = pTable;
+      desc.table = tsChildTableSdb;
+      sdbDeleteRow(&desc);
       pNode = pLastNode;
       continue;
     }
@@ -230,7 +199,11 @@ int32_t mgmtInitChildTables() {
       mError("ctable:%s, db:%s not match with vgroup:%d db:%s sid:%d, discard it",
              pTable->tableId, pDb->name, pTable->vgId, pVgroup->dbName, pTable->sid);
       pTable->vgId = 0;
-      sdbDeleteRow(tsChildTableSdb, pTable);
+      SSdbOperDesc desc = {0};
+      desc.type = SDB_OPER_TYPE_LOCAL;
+      desc.pObj = pTable;
+      desc.table = tsChildTableSdb;
+      sdbDeleteRow(&desc);
       pNode = pLastNode;
       continue;
     }
@@ -238,28 +211,27 @@ int32_t mgmtInitChildTables() {
     if (pVgroup->tableList == NULL) {
       mError("ctable:%s, vgroup:%d tableList is null", pTable->tableId, pTable->vgId);
       pTable->vgId = 0;
-      sdbDeleteRow(tsChildTableSdb, pTable);
+      SSdbOperDesc desc = {0};
+      desc.type = SDB_OPER_TYPE_LOCAL;
+      desc.pObj = pTable;
+      desc.table = tsChildTableSdb;
+      sdbDeleteRow(&desc);
       pNode = pLastNode;
       continue;
     }
 
-    pVgroup->tableList[pTable->sid] = (STableInfo*)pTable;
-    taosIdPoolMarkStatus(pVgroup->idPool, pTable->sid, 1);
-
     SSuperTableObj *pSuperTable = mgmtGetSuperTable(pTable->superTableId);
     if (pSuperTable == NULL) {
       mError("ctable:%s, stable:%s not exist", pTable->tableId, pTable->superTableId);
       pTable->vgId = 0;
-      sdbDeleteRow(tsChildTableSdb, pTable);
+      SSdbOperDesc desc = {0};
+      desc.type = SDB_OPER_TYPE_LOCAL;
+      desc.pObj = pTable;
+      desc.table = tsChildTableSdb;
+      sdbDeleteRow(&desc);
       pNode = pLastNode;
       continue;
     }
-
-    pTable->superTable = pSuperTable;
-    mgmtAddTableIntoSuperTable(pSuperTable);
-
-    SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
-    mgmtAddTimeSeries(pAcct, pTable->superTable->numOfColumns - 1);
   }
 
   mTrace("child table is initialized");
@@ -271,8 +243,13 @@ void mgmtCleanUpChildTables() {
 }
 
 void *mgmtBuildCreateChildTableMsg(SCMCreateTableMsg *pMsg, SChildTableObj *pTable) {
-  char    *pTagData  = pMsg->schema + TSDB_TABLE_ID_LEN + 1;
-  int32_t tagDataLen = htonl(pMsg->contLen) - sizeof(SCMCreateTableMsg) - TSDB_TABLE_ID_LEN - 1;
+  char *  pTagData = NULL;
+  int32_t tagDataLen = 0;
+  if (pMsg != NULL) {
+    pTagData = pMsg->schema + TSDB_TABLE_ID_LEN + 1;
+    tagDataLen = htonl(pMsg->contLen) - sizeof(SCMCreateTableMsg) - TSDB_TABLE_ID_LEN - 1;
+  }
+
   int32_t totalCols  = pTable->superTable->numOfColumns + pTable->superTable->numOfTags;
   int32_t contLen    = sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema) + tagDataLen;
 
@@ -305,18 +282,14 @@ void *mgmtBuildCreateChildTableMsg(SCMCreateTableMsg *pMsg, SChildTableObj *pTab
     pSchema++;
   }
 
+  if (pMsg != NULL) {
     memcpy(pCreate->data + totalCols * sizeof(SSchema), pTagData, tagDataLen);
+  }
+
   return pCreate;
 }
 
 void* mgmtCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t tid) {
-  int32_t numOfTables = sdbGetNumOfRows(tsChildTableSdb);
-  if (numOfTables >= tsMaxTables) {
-    mError("ctable:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, tsMaxTables);
-    terrno = TSDB_CODE_TOO_MANY_TABLES;
-    return NULL;
-  }
-
   char *pTagData = (char *) pCreate->schema;  // it is a tag key
   SSuperTableObj *pSuperTable = mgmtGetSuperTable(pTagData);
   if (pSuperTable == NULL) {
@@ -337,12 +310,18 @@ void* mgmtCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t
   pTable->type        = TSDB_CHILD_TABLE;
   pTable->createdTime = taosGetTimestampMs();
   pTable->uid         = (((uint64_t) pTable->vgId) << 40) + ((((uint64_t) pTable->sid) & ((1ul << 24) - 1ul)) << 16) +
-                        ((uint64_t) sdbGetVersion() & ((1ul << 16) - 1ul));
+                        (sdbGetVersion() & ((1ul << 16) - 1ul));
   pTable->sid         = tid;
   pTable->vgId        = pVgroup->vgId;
   pTable->superTable  = pSuperTable;
 
-  if (sdbInsertRow(tsChildTableSdb, pTable, 0) < 0) {
+  SSdbOperDesc desc = {0};
+  desc.type = SDB_OPER_TYPE_GLOBAL;
+  desc.pObj = pTable;
+  desc.table = tsChildTableSdb;
+  sdbInsertRow(&desc);
+
+  if (sdbInsertRow(&desc) < 0) {
     free(pTable);
     mError("ctable:%s, update sdb error", pCreate->tableId);
     terrno = TSDB_CODE_SDB_ERROR;
@@ -485,12 +464,45 @@ void mgmtDropAllChildTables(SDbObj *pDropDb) {
     }
 
     if (strncmp(pDropDb->name, pTable->tableId, dbNameLen) == 0) {
-      sdbDeleteRow(tsChildTableSdb, pTable);
+      SSdbOperDesc oper = {
+        .type = SDB_OPER_TYPE_LOCAL,
+        .table = tsChildTableSdb,
+        .pObj = pTable,
+      };
+      sdbDeleteRow(&oper);
+      pNode = pLastNode;
+      numOfTables++;
+      continue;
+    }
+  }
+
+  mTrace("db:%s, all child tables:%d is dropped from sdb", pDropDb->name, numOfTables);
+}
+
+void mgmtDropAllChildTablesInStable(SSuperTableObj *pStable) {
+  void *pNode = NULL;
+  void *pLastNode = NULL;
+  int32_t numOfTables = 0;
+  SChildTableObj *pTable = NULL;
+
+  while (1) {
+    pNode = sdbFetchRow(tsChildTableSdb, pNode, (void **)&pTable);
+    if (pTable == NULL) {
+      break;
+    }
+
+    if (pTable->superTable == pStable) {
+      SSdbOperDesc oper = {
+        .type = SDB_OPER_TYPE_LOCAL,
+        .table = tsChildTableSdb,
+        .pObj = pTable,
+      };
+      sdbDeleteRow(&oper);
       pNode = pLastNode;
-      numOfTables ++;
+      numOfTables++;
       continue;
     }
   }
 
-  mTrace("db:%s, all child tables:%d is dropped", pDropDb->name, numOfTables);
+  mTrace("stable:%s, all child tables:%d is dropped from sdb", pStable->tableId, numOfTables);
 }
\ No newline at end of file

src/mnode/src/mgmtDClient.c

@@ -42,7 +42,7 @@ int32_t mgmtInitDClient() {
   rpcInit.label        = "MND-DC";
   rpcInit.numOfThreads = 1;
   rpcInit.cfp          = mgmtProcessRspFromDnode;
-  rpcInit.sessions     = tsMaxDnodes * 5;
+  rpcInit.sessions     = 100;
   rpcInit.connType     = TAOS_CONN_CLIENT;
   rpcInit.idleTime     = tsShellActivityTimer * 1000;
   rpcInit.user         = "mgmtDClient";

src/mnode/src/mgmtDServer.c

@@ -45,7 +45,7 @@ int32_t mgmtInitDServer() {
   rpcInit.label        = "MND-DS";
   rpcInit.numOfThreads = 1;
   rpcInit.cfp          = mgmtProcessMsgFromDnode;
-  rpcInit.sessions     = tsMaxDnodes * 5;
+  rpcInit.sessions     = 100;
   rpcInit.connType     = TAOS_CONN_SERVER;
   rpcInit.idleTime     = tsShellActivityTimer * 1000;
   rpcInit.afp          = mgmtDServerRetrieveAuth;

src/mnode/src/mgmtDnode.c

@@ -547,9 +547,6 @@ void mgmtProcessDnodeStatusMsg(SRpcMsg *rpcMsg) {
     return ;
   }
   
-  uint32_t lastPrivateIp = pDnode->privateIp;
-  uint32_t lastPublicIp  = pDnode->publicIp;
-
   pDnode->privateIp        = htonl(pStatus->privateIp);
   pDnode->publicIp         = htonl(pStatus->publicIp);
   pDnode->lastReboot       = htonl(pStatus->lastReboot);
@@ -566,11 +563,6 @@ void mgmtProcessDnodeStatusMsg(SRpcMsg *rpcMsg) {
     mgmtSetDnodeMaxVnodes(pDnode);
   }
  
-  if (lastPrivateIp != pDnode->privateIp || lastPublicIp != pDnode->publicIp) {
-    mgmtUpdateVgroupIp(pDnode);
-    //mgmtUpdateMnodeIp();
-  }
-
   int32_t openVnodes = htons(pStatus->openVnodes);
   for (int32_t j = 0; j < openVnodes; ++j) {
     pDnode->vload[j].vgId          = htonl(pStatus->load[j].vgId);
@@ -599,7 +591,7 @@ void mgmtProcessDnodeStatusMsg(SRpcMsg *rpcMsg) {
     return;
   }
 
-  mgmtGetMnodeIpList(&pRsp->ipList);
+  mgmtGetMnodePrivateIpList(&pRsp->ipList);
 
   pRsp->dnodeState.dnodeId = htonl(pDnode->dnodeId);
   pRsp->dnodeState.moduleStatus = htonl(pDnode->moduleStatus);

src/mnode/src/mgmtMain.c

@@ -25,6 +25,8 @@
 #include "mgmtDClient.h"
 #include "mgmtDnode.h"
 #include "mgmtDServer.h"
+#include "mgmtMnode.h"
+#include "mgmtSdb.h"
 #include "mgmtVgroup.h"
 #include "mgmtUser.h"
 #include "mgmtTable.h"
@@ -65,7 +67,7 @@ int32_t mgmtStartSystem() {
     return 0;
   }
 
-  tsMgmtTmr = taosTmrInit((tsMaxDnodes + tsMaxShellConns) * 3, 200, 3600000, "MND");
+  tsMgmtTmr = taosTmrInit((tsMaxShellConns) * 3, 200, 3600000, "MND");
   if (tsMgmtTmr == NULL) {
     mError("failed to init timer");
     return -1;
@@ -109,8 +111,8 @@ int32_t mgmtStartSystem() {
     return -1;
   }
 
-  if (sdbInitPeers(tsMnodeDir) < 0) {
-    mError("failed to init peers");
+  if (mgmtInitMnodes() < 0) {
+    mError("failed to init mnodes");
     return -1;
   }
 
@@ -125,7 +127,7 @@ int32_t mgmtStartSystem() {
 
 
 void mgmtStopSystem() {
-  if (sdbMaster) {
+  if (mgmtIsMaster()) {
     mTrace("it is a master mgmt node, it could not be stopped");
     return;
   }
@@ -136,7 +138,7 @@ void mgmtStopSystem() {
 
 void mgmtCleanUpSystem() {
   mPrint("starting to clean up mgmt");
-  sdbCleanUpPeers();
+  mgmtCleanupMnodes();
   mgmtCleanupBalance();
   mgmtCleanUpShell();
   mgmtCleanupDClient();

src/mnode/src/mgmtMnode.c

@@ -17,48 +17,93 @@
 #include "os.h"
 #include "trpc.h"
 #include "mgmtMnode.h"
+#include "mgmtSdb.h"
 #include "mgmtUser.h"
 
-int32_t (*mgmtAddMnodeFp)(uint32_t privateIp, uint32_t publicIp) = NULL;
-int32_t (*mgmtRemoveMnodeFp)(uint32_t privateIp) = NULL;
-int32_t (*mgmtGetMnodesNumFp)() = NULL;
-void *  (*mgmtGetNextMnodeFp)(SShowObj *pShow, SMnodeObj **pMnode) = NULL;
+int32_t (*mpeerAddMnodeFp)(uint32_t privateIp, uint32_t publicIp) = NULL;
+int32_t (*mpeerRemoveMnodeFp)(uint32_t privateIp) = NULL;
+int32_t (*mpeerGetMnodesNumFp)() = NULL;
+void *  (*mpeerGetNextMnodeFp)(SShowObj *pShow, SMnodeObj **pMnode) = NULL;
+int32_t (*mpeerInitMnodesFp)() = NULL;
+void    (*mpeerCleanUpMnodesFp)() = NULL;
 
 static SMnodeObj tsMnodeObj = {0};
+static bool tsMnodeIsMaster = false;
+static bool tsMnodeIsServing = false;
 static int32_t mgmtGetMnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
 static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn);
 
+static char *mgmtMnodeStatusStr[] = {
+  "offline",
+  "unsynced",
+  "syncing",
+  "serving",
+  "null"
+};
+
+static char *mgmtMnodeRoleStr[] = {
+  "unauthed",
+  "undecided",
+  "master",
+  "slave",
+  "null"
+};
+
+int32_t mgmtInitMnodes() {
+  if (mpeerInitMnodesFp) {
+    return (*mpeerInitMnodesFp)();
+  } else {
+    tsMnodeIsServing = true;
+    tsMnodeIsMaster = true;
+    return 0;
+  }
+}
+
+void mgmtCleanupMnodes() {
+  if (mpeerCleanUpMnodesFp) {
+    (*mpeerCleanUpMnodesFp)();
+  }
+}
+
+bool mgmtInServerStatus() {
+  return tsMnodeIsServing; 
+}
+   
+bool mgmtIsMaster() { 
+  return tsMnodeIsMaster; 
+}
+
 bool mgmtCheckRedirect(void *handle) {
   return false;
 }
 
 int32_t mgmtAddMnode(uint32_t privateIp, uint32_t publicIp) {
-  if (mgmtAddMnodeFp) {
-    return (*mgmtAddMnodeFp)(privateIp, publicIp);
+  if (mpeerAddMnodeFp) {
+    return (*mpeerAddMnodeFp)(privateIp, publicIp);
   } else {
     return 0;
   }
 }
 
 int32_t mgmtRemoveMnode(uint32_t privateIp) {
-  if (mgmtRemoveMnodeFp) {
-    return (*mgmtRemoveMnodeFp)(privateIp);
+  if (mpeerRemoveMnodeFp) {
+    return (*mpeerRemoveMnodeFp)(privateIp);
   } else {
     return 0;
   }
 }
 
 static int32_t mgmtGetMnodesNum() {
-  if (mgmtGetMnodesNumFp) {
-    return (*mgmtGetMnodesNumFp)();
+  if (mpeerGetMnodesNumFp) {
+    return (*mpeerGetMnodesNumFp)();
   } else {
     return 1;
   }
 }
 
 static void *mgmtGetNextMnode(SShowObj *pShow, SMnodeObj **pMnode) {
-  if (mgmtGetNextMnodeFp) {
-    return (*mgmtGetNextMnodeFp)(pShow, pMnode);
+  if (mpeerGetNextMnodeFp) {
+    return (*mpeerGetNextMnodeFp)(pShow, pMnode);
   } else {
     if (*pMnode == NULL) {
       *pMnode = &tsMnodeObj;
@@ -148,11 +193,11 @@ static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, voi
     cols++;
 
     pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
-    strcpy(pWrite, sdbStatusStr[(uint8_t)pMnode->status]);
+    strcpy(pWrite, mgmtMnodeStatusStr[pMnode->status]);
     cols++;
 
     pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
-    strcpy(pWrite, sdbRoleStr[(uint8_t)pMnode->role]);
+    strcpy(pWrite, mgmtMnodeRoleStr[pMnode->role]);
     cols++;
 
     tinet_ntoa(ipstr, pMnode->publicIp);
@@ -167,7 +212,14 @@ static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, voi
   return numOfRows;
 }
 
-void mgmtGetMnodeIpList(SRpcIpSet *ipSet) {
+void mgmtGetMnodePrivateIpList(SRpcIpSet *ipSet) {
+  ipSet->inUse = 0;
+  ipSet->port = htons(tsMnodeDnodePort);
+  ipSet->numOfIps = 1;
+  ipSet->ip[0] = htonl(inet_addr(tsMasterIp));
+}
+
+void mgmtGetMnodePublicIpList(SRpcIpSet *ipSet) {
   ipSet->inUse = 0;
   ipSet->port = htons(tsMnodeDnodePort);
   ipSet->numOfIps = 1;

src/mnode/src/mgmtNormalTable.c

@@ -26,191 +26,132 @@
 #include "mgmtDb.h"
 #include "mgmtDClient.h"
 #include "mgmtGrant.h"
+#include "mgmtMnode.h"
 #include "mgmtNormalTable.h"
+#include "mgmtSdb.h"
 #include "mgmtSuperTable.h"
 #include "mgmtTable.h"
 #include "mgmtVgroup.h"
 
 void   *tsNormalTableSdb;
 int32_t tsNormalTableUpdateSize;
-void *(*mgmtNormalTableActionFp[SDB_MAX_ACTION_TYPES])(void *row, char *str, int32_t size, int32_t *ssize);
-
-void *mgmtNormalTableActionInsert(void *row, char *str, int32_t size, int32_t *ssize);
-void *mgmtNormalTableActionDelete(void *row, char *str, int32_t size, int32_t *ssize);
-void *mgmtNormalTableActionUpdate(void *row, char *str, int32_t size, int32_t *ssize);
-void *mgmtNormalTableActionEncode(void *row, char *str, int32_t size, int32_t *ssize);
-void *mgmtNormalTableActionDecode(void *row, char *str, int32_t size, int32_t *ssize);
-void *mgmtNormalTableActionReset(void *row, char *str, int32_t size, int32_t *ssize);
-void *mgmtNormalTableActionDestroy(void *row, char *str, int32_t size, int32_t *ssize);
 
 static void mgmtDestroyNormalTable(SNormalTableObj *pTable) {
-  free(pTable->schema);
-  free(pTable->sql);
-  free(pTable);
-}
-
-static void mgmtNormalTableActionInit() {
-  mgmtNormalTableActionFp[SDB_TYPE_INSERT] = mgmtNormalTableActionInsert;
-  mgmtNormalTableActionFp[SDB_TYPE_DELETE] = mgmtNormalTableActionDelete;
-  mgmtNormalTableActionFp[SDB_TYPE_UPDATE] = mgmtNormalTableActionUpdate;
-  mgmtNormalTableActionFp[SDB_TYPE_ENCODE] = mgmtNormalTableActionEncode;
-  mgmtNormalTableActionFp[SDB_TYPE_DECODE] = mgmtNormalTableActionDecode;
-  mgmtNormalTableActionFp[SDB_TYPE_RESET] = mgmtNormalTableActionReset;
-  mgmtNormalTableActionFp[SDB_TYPE_DESTROY] = mgmtNormalTableActionDestroy;
-}
-
-void *mgmtNormalTableActionReset(void *row, char *str, int32_t size, int32_t *ssize) {
-  SNormalTableObj *pTable = (SNormalTableObj *) row;
-  memcpy(pTable, str, tsNormalTableUpdateSize);
-
-  int32_t schemaSize = sizeof(SSchema) * (pTable->numOfColumns) + pTable->sqlLen;
-  pTable->schema = realloc(pTable->schema, schemaSize);
-  pTable->sql = (char*)pTable->schema + sizeof(SSchema) * (pTable->numOfColumns);
-  memcpy(pTable->schema, str + tsNormalTableUpdateSize, schemaSize);
-
-  return NULL;
+  tfree(pTable->schema);
+  tfree(pTable->sql);
+  tfree(pTable);
 }
 
-void *mgmtNormalTableActionDestroy(void *row, char *str, int32_t size, int32_t *ssize) {
-  SNormalTableObj *pTable = (SNormalTableObj *)row;
-  mgmtDestroyNormalTable(pTable);
-  return NULL;
+static int32_t mgmtNormalTableActionDestroy(SSdbOperDesc *pOper) {
+  mgmtDestroyNormalTable(pOper->pObj);
+  return TSDB_CODE_SUCCESS;
 }
 
-void *mgmtNormalTableActionInsert(void *row, char *str, int32_t size, int32_t *ssize) {
-  SNormalTableObj *pTable = (SNormalTableObj *) row;
+static int32_t mgmtNormalTableActionInsert(SSdbOperDesc *pOper) {
+  SNormalTableObj *pTable = pOper->pObj;
 
   SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
   if (pVgroup == NULL) {
-    mError("id:%s not in vgroup:%d", pTable->tableId, pTable->vgId);
-    return NULL;
+    mError("ntable:%s not in vgroup:%d", pTable->tableId, pTable->vgId);
+    return TSDB_CODE_INVALID_VGROUP_ID;
   }
 
   SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
   if (pDb == NULL) {
-    mError("vgroup:%d not in DB:%s", pVgroup->vgId, pVgroup->dbName);
-    return NULL;
+    mError("ntable:%s, vgroup:%d not in DB:%s", pTable->tableId, pVgroup->vgId, pVgroup->dbName);
+    return TSDB_CODE_INVALID_DB;
   }
 
   SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
   if (pAcct == NULL) {
-    mError("account not exists");
-    return NULL;
-  }
-
-  if (!sdbMaster) {
-    int32_t sid = taosAllocateId(pVgroup->idPool);
-    if (sid != pTable->sid) {
-      mError("sid:%d is not matched from the master:%d", sid, pTable->sid);
-      return NULL;
-    }
+    mError("ntable:%s, account:%s not exists", pTable->tableId, pDb->cfg.acct);
+    return TSDB_CODE_INVALID_ACCT;
   }
 
   mgmtAddTimeSeries(pAcct, pTable->numOfColumns - 1);
   mgmtAddTableIntoDb(pDb);
   mgmtAddTableIntoVgroup(pVgroup, (STableInfo *) pTable);
 
-  if (pVgroup->numOfTables >= pDb->cfg.maxSessions - 1 && pDb->numOfVgroups > 1) {
-    mgmtMoveVgroupToTail(pDb, pVgroup);
-  }
-
-  return NULL;
+  return TSDB_CODE_SUCCESS;
 }
 
-void *mgmtNormalTableActionDelete(void *row, char *str, int32_t size, int32_t *ssize) {
-  SNormalTableObj *pTable = (SNormalTableObj *) row;
+static int32_t mgmtNormalTableActionDelete(SSdbOperDesc *pOper) {
+  SNormalTableObj *pTable = pOper->pObj;
   if (pTable->vgId == 0) {
-    return NULL;
+    return TSDB_CODE_INVALID_VGROUP_ID;
   }
 
   SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
   if (pVgroup == NULL) {
-    return NULL;
+    return TSDB_CODE_INVALID_VGROUP_ID;
   }
 
   SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
   if (pDb == NULL) {
-    mError("vgroup:%d not in DB:%s", pVgroup->vgId, pVgroup->dbName);
-    return NULL;
+    mError("ntable:%s, vgroup:%d not in DB:%s", pTable->tableId, pVgroup->vgId, pVgroup->dbName);
+    return TSDB_CODE_INVALID_DB;
   }
 
   SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
   if (pAcct == NULL) {
     mError("account not exists");
-    return NULL;
+    return TSDB_CODE_INVALID_ACCT;
   }
 
   mgmtRestoreTimeSeries(pAcct, pTable->numOfColumns - 1);
   mgmtRemoveTableFromDb(pDb);
   mgmtRemoveTableFromVgroup(pVgroup, (STableInfo *) pTable);
 
-  if (pVgroup->numOfTables > 0) {
-    mgmtMoveVgroupToHead(pDb, pVgroup);
-  }
-
-  return NULL;
+  return TSDB_CODE_SUCCESS;
 }
 
-void *mgmtNormalTableActionUpdate(void *row, char *str, int32_t size, int32_t *ssize) {
-  return mgmtNormalTableActionReset(row, str, size, NULL);
+static int32_t mgmtNormalTableActionUpdate(SSdbOperDesc *pOper) {
+  return TSDB_CODE_SUCCESS;
 }
 
-void *mgmtNormalTableActionEncode(void *row, char *str, int32_t size, int32_t *ssize) {
-  SNormalTableObj *pTable = (SNormalTableObj *) row;
-  assert(row != NULL && str != NULL);
+static int32_t mgmtNormalTableActionEncode(SSdbOperDesc *pOper) {
+  SNormalTableObj *pTable = pOper->pObj;
+  assert(pOper->pObj != NULL && pOper->rowData != NULL);
 
   int32_t schemaSize = pTable->numOfColumns * sizeof(SSchema);
-  if (size < tsNormalTableUpdateSize + schemaSize + 1) {
-    *ssize = -1;
-    return NULL;
+  if (pOper->maxRowSize < tsNormalTableUpdateSize + schemaSize) {
+    return TSDB_CODE_INVALID_MSG_LEN;
   }
 
-  memcpy(str, pTable, tsNormalTableUpdateSize);
-  memcpy(str + tsNormalTableUpdateSize, pTable->schema, schemaSize);
-  memcpy(str + tsNormalTableUpdateSize + schemaSize, pTable->sql, pTable->sqlLen);
-  *ssize = tsNormalTableUpdateSize + schemaSize + pTable->sqlLen;
+  memcpy(pOper->rowData, pTable, tsNormalTableUpdateSize);
+  memcpy(pOper->rowData + tsNormalTableUpdateSize, pTable->schema, schemaSize);
+  memcpy(pOper->rowData + tsNormalTableUpdateSize + schemaSize, pTable->sql, pTable->sqlLen);
 
-  return NULL;
+  pOper->rowSize = tsNormalTableUpdateSize + schemaSize + pTable->sqlLen;
+  return TSDB_CODE_SUCCESS;
 }
 
-void *mgmtNormalTableActionDecode(void *row, char *str, int32_t size, int32_t *ssize) {
-  assert(str != NULL);
+static int32_t mgmtNormalTableActionDecode(SSdbOperDesc *pOper) {
+  assert(pOper->rowData != NULL);
 
-  SNormalTableObj *pTable = (SNormalTableObj *)malloc(sizeof(SNormalTableObj));
-  if (pTable == NULL) {
-    return NULL;
-  }
-  memset(pTable, 0, sizeof(SNormalTableObj));
+  SNormalTableObj *pTable = (SNormalTableObj *)calloc(1, sizeof(SNormalTableObj));
+  if (pTable == NULL) TSDB_CODE_SERV_OUT_OF_MEMORY;
 
-  if (size < tsNormalTableUpdateSize) {
-    mgmtDestroyNormalTable(pTable);
-    return NULL;
-  }
-  memcpy(pTable, str, tsNormalTableUpdateSize);
+  memcpy(pTable, pOper->rowData, tsNormalTableUpdateSize);
 
   int32_t schemaSize = pTable->numOfColumns * sizeof(SSchema);
   pTable->schema = (SSchema *)malloc(schemaSize);
   if (pTable->schema == NULL) {
     mgmtDestroyNormalTable(pTable);
-    return NULL;
+    return -1;
   }
 
-  memcpy(pTable->schema, str + tsNormalTableUpdateSize, schemaSize);
+  memcpy(pTable->schema, pOper->rowData + tsNormalTableUpdateSize, schemaSize);
 
   pTable->sql = (char *)malloc(pTable->sqlLen);
   if (pTable->sql == NULL) {
     mgmtDestroyNormalTable(pTable);
-    return NULL;
+    return -1;
   }
-  memcpy(pTable->sql, str + tsNormalTableUpdateSize + schemaSize, pTable->sqlLen);
-  return (void *)pTable;
-}
+  memcpy(pTable->sql, pOper->rowData + tsNormalTableUpdateSize + schemaSize, pTable->sqlLen);
   
-void *mgmtNormalTableAction(char action, void *row, char *str, int32_t size, int32_t *ssize) {
-  if (mgmtNormalTableActionFp[(uint8_t)action] != NULL) {
-    return (*(mgmtNormalTableActionFp[(uint8_t)action]))(row, str, size, ssize);
-  }
-  return NULL;
+  pOper->pObj = pTable;
+  return TSDB_CODE_SUCCESS;
 }
 
 int32_t mgmtInitNormalTables() {
@@ -218,12 +159,23 @@ int32_t mgmtInitNormalTables() {
   void *pLastNode = NULL;
   SNormalTableObj *pTable = NULL;
 
-  mgmtNormalTableActionInit();
   SNormalTableObj tObj;
-  tsNormalTableUpdateSize = tObj.updateEnd - (int8_t *)&tObj;
+  tsNormalTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
+
+  SSdbTableDesc tableDesc = {
+    .tableName    = "ntables",
+    .hashSessions = TSDB_MAX_NORMAL_TABLES,
+    .maxRowSize   = sizeof(SNormalTableObj) + sizeof(SSchema) * TSDB_MAX_COLUMNS,
+    .keyType      = SDB_KEY_TYPE_STRING,
+    .insertFp     = mgmtNormalTableActionInsert,
+    .deleteFp     = mgmtNormalTableActionDelete,
+    .updateFp     = mgmtNormalTableActionUpdate,
+    .encodeFp     = mgmtNormalTableActionEncode,
+    .decodeFp     = mgmtNormalTableActionDecode,
+    .destroyFp    = mgmtNormalTableActionDestroy,
+  };
 
-  tsNormalTableSdb = sdbOpenTable(tsMaxTables, sizeof(SNormalTableObj) + sizeof(SSchema) * TSDB_MAX_COLUMNS,
-                                 "ntables", SDB_KEYTYPE_STRING, tsMnodeDir, mgmtNormalTableAction);
+  tsNormalTableSdb = sdbOpenTable(&tableDesc);
   if (tsNormalTableSdb == NULL) {
     mError("failed to init ntables data");
     return -1;
@@ -237,7 +189,11 @@ int32_t mgmtInitNormalTables() {
     SDbObj *pDb = mgmtGetDbByTableId(pTable->tableId);
     if (pDb == NULL) {
       mError("ntable:%s, failed to get db, discard it", pTable->tableId);
-      sdbDeleteRow(tsNormalTableSdb, pTable);
+       SSdbOperDesc desc = {0};
+      desc.type = SDB_OPER_TYPE_LOCAL;
+      desc.pObj = pTable;
+      desc.table = tsNormalTableSdb;
+      sdbDeleteRow(&desc);
       pNode = pLastNode;
       continue;
     }
@@ -246,7 +202,11 @@ int32_t mgmtInitNormalTables() {
     if (pVgroup == NULL) {
       mError("ntable:%s, failed to get vgroup:%d sid:%d, discard it", pTable->tableId, pTable->vgId, pTable->sid);
       pTable->vgId = 0;
-      sdbDeleteRow(tsNormalTableSdb, pTable);
+      SSdbOperDesc desc = {0};
+      desc.type = SDB_OPER_TYPE_LOCAL;
+      desc.pObj = pTable;
+      desc.table = tsNormalTableSdb;
+      sdbDeleteRow(&desc);
       pNode = pLastNode;
       continue;
     }
@@ -255,7 +215,12 @@ int32_t mgmtInitNormalTables() {
       mError("ntable:%s, db:%s not match with vgroup:%d db:%s sid:%d, discard it",
                pTable->tableId, pDb->name, pTable->vgId, pVgroup->dbName, pTable->sid);
       pTable->vgId = 0;
-      sdbDeleteRow(tsNormalTableSdb, pTable);
+      
+      SSdbOperDesc desc = {0};
+      desc.type = SDB_OPER_TYPE_LOCAL;
+      desc.pObj = pTable;
+      desc.table = tsNormalTableSdb;
+      sdbDeleteRow(&desc);
       pNode = pLastNode;
       continue;
     }
@@ -263,19 +228,14 @@ int32_t mgmtInitNormalTables() {
     if (pVgroup->tableList == NULL) {
       mError("ntable:%s, vgroup:%d tableList is null", pTable->tableId, pTable->vgId);
       pTable->vgId = 0;
-      sdbDeleteRow(tsNormalTableSdb, pTable);
+       SSdbOperDesc desc = {0};
+      desc.type = SDB_OPER_TYPE_LOCAL;
+      desc.pObj = pTable;
+      desc.table = tsNormalTableSdb;
+      sdbDeleteRow(&desc);
       pNode = pLastNode;
       continue;
     }
-
-    mgmtAddTableIntoVgroup(pVgroup, (STableInfo *)pTable);
-    //pVgroup->tableList[pTable->sid] = (STableInfo*)pTable;
-    taosIdPoolMarkStatus(pVgroup->idPool, pTable->sid, 1);
-
-    pTable->sql = (char *)pTable->schema + sizeof(SSchema) * pTable->numOfColumns;
-
-    SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
-    mgmtAddTimeSeries(pAcct, pTable->numOfColumns - 1);
   }
 
   mTrace("ntables is initialized");
@@ -323,13 +283,6 @@ void *mgmtBuildCreateNormalTableMsg(SNormalTableObj *pTable) {
 }
 
 void *mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t sid) {
-  int32_t numOfTables = sdbGetNumOfRows(tsNormalTableSdb);
-  if (numOfTables >= TSDB_MAX_NORMAL_TABLES) {
-    mError("table:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, TSDB_MAX_NORMAL_TABLES);
-    terrno = TSDB_CODE_TOO_MANY_TABLES;
-    return NULL;
-  }
-
   SNormalTableObj *pTable = (SNormalTableObj *) calloc(sizeof(SNormalTableObj), 1);
   if (pTable == NULL) {
     mError("table:%s, failed to alloc memory", pCreate->tableId);
@@ -341,7 +294,7 @@ void *mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t
   pTable->type         = TSDB_NORMAL_TABLE;
   pTable->vgId         = pVgroup->vgId;
   pTable->createdTime  = taosGetTimestampMs();
-  pTable->uid          = (((uint64_t) pTable->createdTime) << 16) + ((uint64_t) sdbGetVersion() & ((1ul << 16) - 1ul));
+  pTable->uid          = (((uint64_t) pTable->createdTime) << 16) + (sdbGetVersion() & ((1ul << 16) - 1ul));
   pTable->sid          = sid;
   pTable->sversion     = 0;
   pTable->numOfColumns = htons(pCreate->numOfColumns);
@@ -377,7 +330,11 @@ void *mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t
     mTrace("table:%s, stream sql len:%d sql:%s", pTable->tableId, pTable->sqlLen, pTable->sql);
   }
 
-  if (sdbInsertRow(tsNormalTableSdb, pTable, 0) < 0) {
+  SSdbOperDesc desc = {0};
+  desc.type = SDB_OPER_TYPE_GLOBAL;
+  desc.pObj = pTable;
+  desc.table = tsNormalTableSdb;
+  if (sdbInsertRow(&desc) < 0) {
     mError("table:%s, update sdb error", pTable->tableId);
     free(pTable);
     terrno = TSDB_CODE_SDB_ERROR;
@@ -474,7 +431,14 @@ int32_t mgmtAddNormalTableColumn(SNormalTableObj *pTable, SSchema schema[], int3
   pTable->sversion++;
   pAcct->acctInfo.numOfTimeSeries += ncols;
 
-  sdbUpdateRow(tsNormalTableSdb, pTable, 0, 1);
+  SSdbOperDesc desc = {0};
+  desc.type = SDB_OPER_TYPE_GLOBAL;
+  desc.pObj = pTable;
+  desc.table = tsNormalTableSdb;
+  desc.rowData = pTable;
+  desc.rowSize = tsNormalTableUpdateSize;
+  sdbUpdateRow(&desc);
+
   return TSDB_CODE_SUCCESS;
 }
 
@@ -503,7 +467,14 @@ int32_t mgmtDropNormalTableColumnByName(SNormalTableObj *pTable, char *colName)
   pTable->sversion++;
 
   pAcct->acctInfo.numOfTimeSeries--;
-  sdbUpdateRow(tsNormalTableSdb, pTable, 0, 1);
+
+  SSdbOperDesc desc = {0};
+  desc.type = SDB_OPER_TYPE_GLOBAL;
+  desc.pObj = pTable;
+  desc.table = tsNormalTableSdb;
+  desc.rowData = pTable;
+  desc.rowSize = tsNormalTableUpdateSize;
+  sdbUpdateRow(&desc);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -564,12 +535,17 @@ void mgmtDropAllNormalTables(SDbObj *pDropDb) {
     if (pTable == NULL) break;
 
     if (strncmp(pDropDb->name, pTable->tableId, dbNameLen) == 0) {
-      sdbDeleteRow(tsNormalTableSdb, pTable);
+      SSdbOperDesc oper = {
+        .type = SDB_OPER_TYPE_LOCAL,
+        .table = tsNormalTableSdb,
+        .pObj = pTable,
+      };
+      sdbDeleteRow(&oper);
       pNode = pLastNode;
-      numOfTables ++;
+      numOfTables++;
       continue;
     }
   }
 
-  mTrace("db:%s, all normal tables:%d is dropped", pDropDb->name, numOfTables);
+  mTrace("db:%s, all normal tables:%d is dropped from sdb", pDropDb->name, numOfTables);
 }

src/mnode/src/mgmtProfile.c

@@ -762,3 +762,13 @@ int32_t mgmtInitProfile() {
 
 void mgmtCleanUpProfile() {
 }
+
+void mgmtFreeQueuedMsg(SQueuedMsg *pMsg) {
+  if (pMsg != NULL) {
+    if (pMsg->pCont != NULL) {
+      rpcFreeCont(pMsg->pCont);
+      pMsg->pCont = NULL;
+    }
+    free(pMsg);
+  }
+}
\ No newline at end of file

src/mnode/src/mgmtShell.c

@@ -32,6 +32,7 @@
 #include "mgmtMnode.h"
 #include "mgmtNormalTable.h"
 #include "mgmtProfile.h"
+#include "mgmtSdb.h"
 #include "mgmtShell.h"
 #include "mgmtSuperTable.h"
 #include "mgmtTable.h"
@@ -63,7 +64,7 @@ int32_t mgmtInitShell() {
   mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_HEARTBEAT, mgmtProcessHeartBeatMsg);
   mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_CONNECT, mgmtProcessConnectMsg);
 
-  tsMgmtTranQhandle = taosInitScheduler(tsMaxDnodes + tsMaxShellConns, 1, "mnodeT");
+  tsMgmtTranQhandle = taosInitScheduler(tsMaxShellConns, 1, "mnodeT");
 
   int32_t numOfThreads = tsNumOfCores * tsNumOfThreadsPerCore / 4.0;
   if (numOfThreads < 1) {
@@ -131,7 +132,7 @@ void mgmtAddToShellQueue(SQueuedMsg *queuedMsg) {
 }
 
 static void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg) {
-  if (sdbGetRunStatus() != SDB_STATUS_SERVING) {
+  if (!mgmtInServerStatus()) {
     mgmtProcessMsgWhileNotReady(rpcMsg);
     rpcFreeCont(rpcMsg->pCont);
     return;
@@ -309,20 +310,10 @@ static void mgmtProcessHeartBeatMsg(SQueuedMsg *pMsg) {
     return;
   }
 
-  pHBRsp->ipList.inUse = 0;
-  pHBRsp->ipList.port = htons(tsMnodeShellPort);
-  pHBRsp->ipList.numOfIps = 0;
-  if (pSdbPublicIpList != NULL && pSdbIpList != NULL) {
-    pHBRsp->ipList.numOfIps = htons(pSdbPublicIpList->numOfIps);
   if (connInfo.serverIp == tsPublicIpInt) {
-      for (int i = 0; i < pSdbPublicIpList->numOfIps; ++i) {
-        pHBRsp->ipList.ip[i] = htonl(pSdbPublicIpList->ip[i]);
-      }
+    mgmtGetMnodePublicIpList(&pHBRsp->ipList);
   } else {
-      for (int i = 0; i < pSdbIpList->numOfIps; ++i) {
-        pHBRsp->ipList.ip[i] = htonl(pSdbIpList->ip[i]);
-      }
-    }
+    mgmtGetMnodePrivateIpList(&pHBRsp->ipList);
   }
 
   /*
@@ -411,20 +402,11 @@ static void mgmtProcessConnectMsg(SQueuedMsg *pMsg) {
   strcpy(pConnectRsp->serverVersion, version);
   pConnectRsp->writeAuth = pUser->writeAuth;
   pConnectRsp->superAuth = pUser->superAuth;
-  pConnectRsp->ipList.inUse = 0;
-  pConnectRsp->ipList.port = htons(tsMnodeShellPort);
-  pConnectRsp->ipList.numOfIps = 0;
-  if (pSdbPublicIpList != NULL && pSdbIpList != NULL) {
-    pConnectRsp->ipList.numOfIps = htons(pSdbPublicIpList->numOfIps);
+
   if (connInfo.serverIp == tsPublicIpInt) {
-      for (int i = 0; i < pSdbPublicIpList->numOfIps; ++i) {
-        pConnectRsp->ipList.ip[i] = htonl(pSdbPublicIpList->ip[i]);
-      }
+    mgmtGetMnodePublicIpList(&pConnectRsp->ipList);
   } else {
-      for (int i = 0; i < pSdbIpList->numOfIps; ++i) {
-        pConnectRsp->ipList.ip[i] = htonl(pSdbIpList->ip[i]);
-      }
-    }
+    mgmtGetMnodePrivateIpList(&pConnectRsp->ipList);
   }
 
 connect_over:

src/mnode/src/mgmtSuperTable.c

@@ -23,6 +23,7 @@
 #include "mgmtGrant.h"
 #include "mgmtShell.h"
 #include "mgmtSuperTable.h"
+#include "mgmtSdb.h"
 #include "mgmtTable.h"
 #include "mgmtUser.h"
 #include "mgmtVgroup.h"
@@ -34,157 +35,103 @@
 static void   *tsSuperTableSdb;
 static int32_t tsSuperTableUpdateSize;
 
-static void *(*mgmtSuperTableActionFp[SDB_MAX_ACTION_TYPES])(void *row, char *str, int32_t size, int32_t *ssize);
-static void *mgmtSuperTableActionInsert(void *row, char *str, int32_t size, int32_t *ssize);
-static void *mgmtSuperTableActionDelete(void *row, char *str, int32_t size, int32_t *ssize);
-static void *mgmtSuperTableActionUpdate(void *row, char *str, int32_t size, int32_t *ssize);
-static void *mgmtSuperTableActionEncode(void *row, char *str, int32_t size, int32_t *ssize);
-static void *mgmtSuperTableActionDecode(void *row, char *str, int32_t size, int32_t *ssize);
-static void *mgmtSuperTableActionReset(void *row, char *str, int32_t size, int32_t *ssize);
-static void *mgmtSuperTableActionDestroy(void *row, char *str, int32_t size, int32_t *ssize);
 static int32_t mgmtRetrieveShowSuperTables(SShowObj *pShow, char *data, int32_t rows, void *pConn);
 static int32_t mgmtGetShowSuperTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
 
-static void mgmtDestroySuperTable(SSuperTableObj *pTable) {
-  free(pTable->schema);
-  free(pTable);
+static void mgmtDestroySuperTable(SSuperTableObj *pStable) {
+  tfree(pStable->schema);
+  tfree(pStable);
 }
 
-static void mgmtSuperTableActionInit() {
-  SSuperTableObj tObj;
-  tsSuperTableUpdateSize = tObj.updateEnd - (int8_t *)&tObj;
-
-  mgmtSuperTableActionFp[SDB_TYPE_INSERT]  = mgmtSuperTableActionInsert;
-  mgmtSuperTableActionFp[SDB_TYPE_DELETE]  = mgmtSuperTableActionDelete;
-  mgmtSuperTableActionFp[SDB_TYPE_UPDATE]  = mgmtSuperTableActionUpdate;
-  mgmtSuperTableActionFp[SDB_TYPE_ENCODE]  = mgmtSuperTableActionEncode;
-  mgmtSuperTableActionFp[SDB_TYPE_DECODE]  = mgmtSuperTableActionDecode;
-  mgmtSuperTableActionFp[SDB_TYPE_RESET]   = mgmtSuperTableActionReset;
-  mgmtSuperTableActionFp[SDB_TYPE_DESTROY] = mgmtSuperTableActionDestroy;
-}
-
-void *mgmtSuperTableActionReset(void *row, char *str, int32_t size, int32_t *ssize) {
-  SSuperTableObj *pTable = (SSuperTableObj *) row;
-  memcpy(pTable, str, tsSuperTableUpdateSize);
-
-  int32_t schemaSize = sizeof(SSchema) * (pTable->numOfColumns + pTable->numOfTags);
-  pTable->schema = realloc(pTable->schema, schemaSize);
-  memcpy(pTable->schema, str + tsSuperTableUpdateSize, schemaSize);
-
-  return NULL;
-}
-
-void *mgmtSuperTableActionDestroy(void *row, char *str, int32_t size, int32_t *ssize) {
-  SSuperTableObj *pTable = (SSuperTableObj *) row;
-  mgmtDestroySuperTable(pTable);
-  return NULL;
+static int32_t mgmtSuperTableActionDestroy(SSdbOperDesc *pOper) {
+  mgmtDestroySuperTable(pOper->pObj);
+  return TSDB_CODE_SUCCESS;
 }
 
-void *mgmtSuperTableActionInsert(void *row, char *str, int32_t size, int32_t *ssize) {
-  STableInfo *pTable = (STableInfo *) row;
-  SDbObj     *pDb    = mgmtGetDbByTableId(pTable->tableId);
-  if (pDb) {
+static int32_t mgmtSuperTableActionInsert(SSdbOperDesc *pOper) {
+  STableInfo *pStable = pOper->pObj;
+  SDbObj *pDb = mgmtGetDbByTableId(pStable->tableId);
+  if (pDb != NULL) {
     mgmtAddSuperTableIntoDb(pDb);
   }
-  return NULL;
+  return TSDB_CODE_SUCCESS;
 }
 
-void *mgmtSuperTableActionDelete(void *row, char *str, int32_t size, int32_t *ssize) {
-  STableInfo *pTable = (STableInfo *) row;
-  SDbObj     *pDb    = mgmtGetDbByTableId(pTable->tableId);
-  if (pDb) {
+static int32_t mgmtSuperTableActionDelete(SSdbOperDesc *pOper) {
+  STableInfo *pStable = pOper->pObj;
+  SDbObj *pDb = mgmtGetDbByTableId(pStable->tableId);
+  if (pDb != NULL) {
     mgmtRemoveSuperTableFromDb(pDb);
+    mgmtDropAllChildTablesInStable((SSuperTableObj *)pStable);
   }
-  return NULL;
+  return TSDB_CODE_SUCCESS;
 }
 
-void *mgmtSuperTableActionUpdate(void *row, char *str, int32_t size, int32_t *ssize) {
-  return mgmtSuperTableActionReset(row, str, size, NULL);
+static int32_t mgmtSuperTableActionUpdate(SSdbOperDesc *pOper) {
+  return TSDB_CODE_SUCCESS;
 }
 
-void *mgmtSuperTableActionEncode(void *row, char *str, int32_t size, int32_t *ssize) {
-  SSuperTableObj *pTable = (SSuperTableObj *) row;
-  assert(row != NULL && str != NULL);
+static int32_t mgmtSuperTableActionEncode(SSdbOperDesc *pOper) {
+  SSuperTableObj *pStable = pOper->pObj;
+  assert(pOper->pObj != NULL && pOper->rowData != NULL);
 
-  int32_t schemaSize = sizeof(SSchema) * (pTable->numOfColumns + pTable->numOfTags);
+  int32_t schemaSize = sizeof(SSchema) * (pStable->numOfColumns + pStable->numOfTags);
 
-  if (size < tsSuperTableUpdateSize + schemaSize + 1) {
-    *ssize = -1;
-    return NULL;
+  if (pOper->maxRowSize < tsSuperTableUpdateSize + schemaSize) {
+    return TSDB_CODE_INVALID_MSG_LEN;
   }
 
-  memcpy(str, pTable, tsSuperTableUpdateSize);
-  memcpy(str + tsSuperTableUpdateSize, pTable->schema, schemaSize);
-  *ssize = tsSuperTableUpdateSize + schemaSize;
+  memcpy(pOper->rowData, pStable, tsSuperTableUpdateSize);
+  memcpy(pOper->rowData + tsSuperTableUpdateSize, pStable->schema, schemaSize);
+  pOper->rowSize = tsSuperTableUpdateSize + schemaSize;
 
-  return NULL;
+  return TSDB_CODE_SUCCESS;
 }
 
-void *mgmtSuperTableActionDecode(void *row, char *str, int32_t size, int32_t *ssize) {
-  assert(str != NULL);
+static int32_t mgmtSuperTableActionDecode(SSdbOperDesc *pOper) {
+  assert(pOper->rowData != NULL);
 
-  SSuperTableObj *pTable = (SSuperTableObj *) malloc(sizeof(SSuperTableObj));
-  if (pTable == NULL) {
-    return NULL;
-  }
-  memset(pTable, 0, sizeof(SSuperTableObj));
+  SSuperTableObj *pStable = (SSuperTableObj *) calloc(1, sizeof(SSuperTableObj));
+  if (pStable == NULL) return TSDB_CODE_SERV_OUT_OF_MEMORY;
 
-  if (size < tsSuperTableUpdateSize) {
-    mgmtDestroySuperTable(pTable);
-    return NULL;
-  }
-  memcpy(pTable, str, tsSuperTableUpdateSize);
+  memcpy(pStable, pOper->rowData, tsSuperTableUpdateSize);
 
-  int32_t schemaSize = sizeof(SSchema) * (pTable->numOfColumns + pTable->numOfTags);
-  pTable->schema = malloc(schemaSize);
-  if (pTable->schema == NULL) {
-    mgmtDestroySuperTable(pTable);
-    return NULL;
+  int32_t schemaSize = sizeof(SSchema) * (pStable->numOfColumns + pStable->numOfTags);
+  pStable->schema = malloc(schemaSize);
+  if (pStable->schema == NULL) {
+    mgmtDestroySuperTable(pStable);
+    return -1;
   }
 
-  memcpy(pTable->schema, str + tsSuperTableUpdateSize, schemaSize);
-  return (void *) pTable;
-}
+  memcpy(pStable->schema, pOper->rowData + tsSuperTableUpdateSize, schemaSize);
+  pOper->pObj = pStable;
 
-void *mgmtSuperTableAction(char action, void *row, char *str, int32_t size, int32_t *ssize) {
-  if (mgmtSuperTableActionFp[(uint8_t) action] != NULL) {
-    return (*(mgmtSuperTableActionFp[(uint8_t) action]))(row, str, size, ssize);
-  }
-  return NULL;
+  return TSDB_CODE_SUCCESS;
 }
 
 int32_t mgmtInitSuperTables() {
-  void *pNode     = NULL;
-  void *pLastNode = NULL;
-  SSuperTableObj *pTable = NULL;
-
-  mgmtSuperTableActionInit();
-
-  tsSuperTableSdb = sdbOpenTable(tsMaxTables, tsSuperTableUpdateSize + sizeof(SSchema) * TSDB_MAX_COLUMNS,
-                          "stables", SDB_KEYTYPE_STRING, tsMnodeDir, mgmtSuperTableAction);
+  SSuperTableObj tObj;
+  tsSuperTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
+
+  SSdbTableDesc tableDesc = {
+    .tableName    = "stables",
+    .hashSessions = TSDB_MAX_SUPER_TABLES,
+    .maxRowSize   = tsSuperTableUpdateSize + sizeof(SSchema) * TSDB_MAX_COLUMNS,
+    .keyType      = SDB_KEY_TYPE_STRING,
+    .insertFp     = mgmtSuperTableActionInsert,
+    .deleteFp     = mgmtSuperTableActionDelete,
+    .updateFp     = mgmtSuperTableActionUpdate,
+    .encodeFp     = mgmtSuperTableActionEncode,
+    .decodeFp     = mgmtSuperTableActionDecode,
+    .destroyFp    = mgmtSuperTableActionDestroy,
+  };
+
+  tsSuperTableSdb = sdbOpenTable(&tableDesc);
   if (tsSuperTableSdb == NULL) {
     mError("failed to init stables data");
     return -1;
   }
 
-  pNode = NULL;
-  while (1) {
-    pNode = sdbFetchRow(tsSuperTableSdb, pNode, (void **) &pTable);
-    if (pTable == NULL) {
-      break;
-    }
-
-    SDbObj *pDb = mgmtGetDbByTableId(pTable->tableId);
-    if (pDb == NULL) {
-      mError("super table:%s, failed to get db, discard it", pTable->tableId);
-      sdbDeleteRow(tsSuperTableSdb, pTable);
-      pNode = pLastNode;
-      continue;
-    }
-
-    mgmtAddSuperTableIntoDb(pDb);
-  }
-
   mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_METRIC, mgmtGetShowSuperTableMeta);
   mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_METRIC, mgmtRetrieveShowSuperTables);
 
@@ -197,13 +144,7 @@ void mgmtCleanUpSuperTables() {
 }
 
 int32_t mgmtCreateSuperTable(SCMCreateTableMsg *pCreate) {
-  int32_t numOfTables = sdbGetNumOfRows(tsSuperTableSdb);
-  if (numOfTables >= TSDB_MAX_SUPER_TABLES) {
-    mError("stable:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, TSDB_MAX_SUPER_TABLES);
-    return TSDB_CODE_TOO_MANY_TABLES;
-  }
-
-  SSuperTableObj *pStable = (SSuperTableObj *)calloc(sizeof(SSuperTableObj), 1);
+  SSuperTableObj *pStable = (SSuperTableObj *)calloc(1, sizeof(SSuperTableObj));
   if (pStable == NULL) {
     return TSDB_CODE_SERV_OUT_OF_MEMORY;
   }
@@ -213,7 +154,7 @@ int32_t mgmtCreateSuperTable(SCMCreateTableMsg *pCreate) {
   pStable->createdTime  = taosGetTimestampMs();
   pStable->vgId         = 0;
   pStable->sid          = 0;
-  pStable->uid          = (((uint64_t) pStable->createdTime) << 16) + ((uint64_t) sdbGetVersion() & ((1ul << 16) - 1ul));
+  pStable->uid          = (((uint64_t) pStable->createdTime) << 16) + (sdbGetVersion() & ((1ul << 16) - 1ul));
   pStable->sversion     = 0;
   pStable->numOfColumns = htons(pCreate->numOfColumns);
   pStable->numOfTags    = htons(pCreate->numOfTags);
@@ -235,13 +176,21 @@ int32_t mgmtCreateSuperTable(SCMCreateTableMsg *pCreate) {
     tschema[col].bytes = htons(tschema[col].bytes);
   }
 
-  if (sdbInsertRow(tsSuperTableSdb, pStable, 0) < 0) {
-    mError("stable:%s, update sdb error", pStable->tableId);
-    return TSDB_CODE_SDB_ERROR;
-  }
+  SSdbOperDesc oper = {
+    .type = SDB_OPER_TYPE_GLOBAL,
+    .table = tsSuperTableSdb,
+    .pObj = pStable,
+    .rowSize = sizeof(SSuperTableObj) + schemaSize
+  };
 
-  mPrint("stable:%s, is created, tags:%d cols:%d", pStable->tableId, pStable->numOfTags, pStable->numOfColumns);
+  int32_t code = sdbInsertRow(&oper);
+  if (code != TSDB_CODE_SUCCESS) {
+    mgmtDestroySuperTable(pStable);
+    return TSDB_CODE_SDB_ERROR;
+  } else {
+    mLPrint("stable:%s, is created, tags:%d cols:%d", pStable->tableId, pStable->numOfTags, pStable->numOfColumns);
     return TSDB_CODE_SUCCESS;
+  }
 }
 
 int32_t mgmtDropSuperTable(SQueuedMsg *newMsg, SDbObj *pDb, SSuperTableObj *pStable) {
@@ -249,10 +198,14 @@ int32_t mgmtDropSuperTable(SQueuedMsg *newMsg, SDbObj *pDb, SSuperTableObj *pSta
     mError("stable:%s, numOfTables:%d not 0", pStable->tableId, pStable->numOfTables);
     return TSDB_CODE_OTHERS;
   } else {
-    //TODO: drop child tables
-    mError("stable:%s, is dropped from sdb", pStable->tableId);
-    mgmtRemoveSuperTableFromDb(pDb);
-    return TSDB_CODE_OTHERS;
+    SSdbOperDesc oper = {
+      .type = SDB_OPER_TYPE_GLOBAL,
+      .table = tsSuperTableSdb,
+      .pObj = pStable
+    };
+    int32_t code = sdbDeleteRow(&oper);
+    mLPrint("stable:%s, is dropped from sdb, result:%s", pStable->tableId, tstrerror(code));
+    return code;
   }
 }
 
@@ -261,10 +214,9 @@ void* mgmtGetSuperTable(char *tableId) {
 }
 
 void *mgmtGetSuperTableVgroup(SSuperTableObj *pStable) {
-  //TODO get vgroup of dnodes
   SCMSuperTableInfoRsp *rsp = rpcMallocCont(sizeof(SCMSuperTableInfoRsp) + sizeof(uint32_t) * mgmtGetDnodesNum());
-  rsp->numOfDnodes = 1;
-  rsp->dnodeIps[0] = 0;
+  rsp->numOfDnodes = htonl(1);
+  rsp->dnodeIps[0] = htonl(inet_addr(tsPrivateIp));
   return rsp;
 }
 
@@ -321,7 +273,7 @@ int32_t mgmtAddSuperTableTag(SSuperTableObj *pStable, SSchema schema[], int32_t
   pStable->sversion++;
 
   pAcct->acctInfo.numOfTimeSeries += (ntags * pStable->numOfTables);
-  sdbUpdateRow(tsSuperTableSdb, pStable, 0, 1);
+  // sdbUpdateRow(tsSuperTableSdb, pStable, tsSuperTableUpdateSize, SDB_OPER_GLOBAL);
 
   mTrace("Succeed to add tag column %s to table %s", schema[0].name, pStable->tableId);
   return TSDB_CODE_SUCCESS;
@@ -354,7 +306,7 @@ int32_t mgmtDropSuperTableTag(SSuperTableObj *pStable, char *tagName) {
   int32_t schemaSize = sizeof(SSchema) * (pStable->numOfTags + pStable->numOfColumns);
   pStable->schema = realloc(pStable->schema, schemaSize);
 
-  sdbUpdateRow(tsSuperTableSdb, pStable, 0, 1);
+  // sdbUpdateRow(tsSuperTableSdb, pStable, tsSuperTableUpdateSize, SDB_OPER_GLOBAL);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -384,9 +336,10 @@ int32_t mgmtModifySuperTableTagNameByName(SSuperTableObj *pStable, char *oldTagN
   if (msg == NULL) return TSDB_CODE_APP_ERROR;
   memset(msg, 0, size);
 
-  mgmtSuperTableActionEncode(pStable, msg, size, &rowSize);
+  // mgmtSuperTableActionEncode(pStable, msg, size, &rowSize);
 
-  int32_t ret = sdbUpdateRow(tsSuperTableSdb, msg, rowSize, 1);
+  int32_t ret = 0;
+  // int32_t ret = sdbUpdateRow(tsSuperTableSdb, msg, tsSuperTableUpdateSize, SDB_OPER_GLOBAL);
   tfree(msg);
 
   if (ret < 0) {
@@ -448,7 +401,7 @@ int32_t mgmtAddSuperTableColumn(SSuperTableObj *pStable, SSchema schema[], int32
   pStable->sversion++;
 
   pAcct->acctInfo.numOfTimeSeries += (ncols * pStable->numOfTables);
-  sdbUpdateRow(tsSuperTableSdb, pStable, 0, 1);
+  // sdbUpdateRow(tsSuperTableSdb, pStable, tsSuperTableUpdateSize, SDB_OPER_GLOBAL);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -481,7 +434,7 @@ int32_t mgmtDropSuperTableColumnByName(SSuperTableObj *pStable, char *colName) {
   pStable->schema = realloc(pStable->schema, schemaSize);
 
   pAcct->acctInfo.numOfTimeSeries -= (pStable->numOfTables);
-  sdbUpdateRow(tsSuperTableSdb, pStable, 0, 1);
+  // sdbUpdateRow(tsSuperTableSdb, pStable, tsSuperTableUpdateSize, SDB_OPER_GLOBAL);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -620,14 +573,19 @@ void mgmtDropAllSuperTables(SDbObj *pDropDb) {
     }
 
     if (strncmp(pDropDb->name, pTable->tableId, dbNameLen) == 0) {
-      sdbDeleteRow(tsSuperTableSdb, pTable);
+      SSdbOperDesc oper = {
+        .type = SDB_OPER_TYPE_LOCAL,
+        .table = tsSuperTableSdb,
+        .pObj = pTable,
+      };
+      sdbDeleteRow(&oper);
       pNode = pLastNode;
       numOfTables ++;
       continue;
     }
   }
 
-  mTrace("db:%s, all super tables:%d is dropped", pDropDb->name, numOfTables);
+  mTrace("db:%s, all super tables:%d is dropped from sdb", pDropDb->name, numOfTables);
 }
 
 void mgmtAddTableIntoSuperTable(SSuperTableObj *pStable) {

src/mnode/src/mgmtTable.c

@@ -22,10 +22,12 @@
 #include "mgmtDClient.h"
 #include "mgmtDb.h"
 #include "mgmtDnode.h"
+#include "mgmtDServer.h"
 #include "mgmtGrant.h"
 #include "mgmtMnode.h"
 #include "mgmtNormalTable.h"
 #include "mgmtProfile.h"
+#include "mgmtSdb.h"
 #include "mgmtShell.h"
 #include "mgmtSuperTable.h"
 #include "mgmtUser.h"
@@ -60,6 +62,7 @@ static void mgmtProcessDropStableRsp(SRpcMsg *rpcMsg);
 static int32_t mgmtGetShowTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
 static int32_t mgmtRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows, void *pConn);
 static void mgmtProcessGetTableMeta(STableInfo *pTable, void *thandle);
+static void mgmtProcessTableCfgMsg(SRpcMsg *rpcMsg);
 
 int32_t mgmtInitTables() {
   int32_t code = mgmtInitSuperTables();
@@ -77,8 +80,6 @@ int32_t mgmtInitTables() {
     return code;
   }
 
-  mgmtSetVgroupIdPool();
-
   mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_CREATE_TABLE, mgmtProcessCreateTableMsg);
   mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_DROP_TABLE, mgmtProcessDropTableMsg);
   mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_ALTER_TABLE, mgmtProcessAlterTableMsg);
@@ -91,6 +92,7 @@ int32_t mgmtInitTables() {
   mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_DROP_TABLE_RSP, mgmtProcessDropTableRsp);
   mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_ALTER_TABLE_RSP, mgmtProcessAlterTableRsp);
   mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_DROP_STABLE_RSP, mgmtProcessDropStableRsp);
+  mgmtAddDServerMsgHandle(TSDB_MSG_TYPE_DM_CONFIG_TABLE, mgmtProcessTableCfgMsg);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -201,13 +203,13 @@ int32_t mgmtGetShowTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn)
 
   pShow->bytes[cols] = TSDB_TABLE_NAME_LEN;
   pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
-  strcpy(pSchema[cols].name, "table_name");
+  strcpy(pSchema[cols].name, "table name");
   pSchema[cols].bytes = htons(pShow->bytes[cols]);
   cols++;
 
   pShow->bytes[cols] = 8;
   pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
-  strcpy(pSchema[cols].name, "created_time");
+  strcpy(pSchema[cols].name, "create time");
   pSchema[cols].bytes = htons(pShow->bytes[cols]);
   cols++;
 
@@ -219,7 +221,7 @@ int32_t mgmtGetShowTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn)
 
   pShow->bytes[cols] = TSDB_TABLE_NAME_LEN;
   pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
-  strcpy(pSchema[cols].name, "stable_name");
+  strcpy(pSchema[cols].name, "stable name");
   pSchema[cols].bytes = htons(pShow->bytes[cols]);
   cols++;
 
@@ -237,9 +239,6 @@ int32_t mgmtGetShowTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn)
   return 0;
 }
 
-/*
- * remove the hole in result set
- */
 static void mgmtVacuumResult(char *data, int32_t numOfCols, int32_t rows, int32_t capacity, SShowObj *pShow) {
   if (rows < capacity) {
     for (int32_t i = 0; i < numOfCols; ++i) {
@@ -433,11 +432,13 @@ void mgmtProcessCreateTableMsg(SQueuedMsg *pMsg) {
     pTable = mgmtCreateChildTable(pCreate, pVgroup, sid);
     if (pTable == NULL) {
       mgmtSendSimpleResp(pMsg->thandle, terrno);
+      mgmtFreeQueuedMsg(newMsg);
       return;
     }
     pMDCreate = mgmtBuildCreateChildTableMsg(pCreate, (SChildTableObj *) pTable);
     if (pMDCreate == NULL) {
       mgmtSendSimpleResp(pMsg->thandle, terrno);
+      mgmtFreeQueuedMsg(newMsg);
       return;
     }
   } else {
@@ -445,11 +446,13 @@ void mgmtProcessCreateTableMsg(SQueuedMsg *pMsg) {
     pTable = mgmtCreateNormalTable(pCreate, pVgroup, sid);
     if (pTable == NULL) {
       mgmtSendSimpleResp(pMsg->thandle, terrno);
+      mgmtFreeQueuedMsg(newMsg);
       return;
     }
     pMDCreate = mgmtBuildCreateNormalTableMsg((SNormalTableObj *) pTable);
     if (pMDCreate == NULL) {
       mgmtSendSimpleResp(pMsg->thandle, terrno);
+      mgmtFreeQueuedMsg(newMsg);
       return;
     }
   }
@@ -760,9 +763,19 @@ static void mgmtProcessCreateTableRsp(SRpcMsg *rpcMsg) {
 
   if (rpcMsg->code != TSDB_CODE_SUCCESS) {
     if (pTable->type == TSDB_CHILD_TABLE) {
-      sdbDeleteRow(tsChildTableSdb, pTable);
+      SSdbOperDesc oper = {
+        .type = SDB_OPER_TYPE_GLOBAL,
+        .table = tsChildTableSdb,
+        .pObj = pTable
+      };
+      sdbDeleteRow(&oper);
     } else if (pTable->type == TSDB_NORMAL_TABLE){
-      sdbDeleteRow(tsNormalTableSdb, pTable);
+      SSdbOperDesc oper = {
+        .type = SDB_OPER_TYPE_GLOBAL,
+        .table = tsNormalTableSdb,
+        .pObj = pTable
+      };
+      sdbDeleteRow(&oper);
     } else {}
     mError("table:%s, failed to create in dnode, reason:%s", pTable->tableId, tstrerror(rpcMsg->code));
     mgmtSendSimpleResp(queueMsg->thandle, rpcMsg->code);
@@ -816,14 +829,26 @@ static void mgmtProcessDropTableRsp(SRpcMsg *rpcMsg) {
   }
 
   if (pTable->type == TSDB_CHILD_TABLE) {
-    if (sdbDeleteRow(tsChildTableSdb, pTable) < 0) {
+    SSdbOperDesc oper = {
+      .type = SDB_OPER_TYPE_GLOBAL,
+      .table = tsChildTableSdb,
+      .pObj = pTable
+    };
+    int32_t code = sdbDeleteRow(&oper);
+    if (code != TSDB_CODE_SUCCESS) {
       mError("table:%s, update ctables sdb error", pTable->tableId);
       mgmtSendSimpleResp(queueMsg->thandle, TSDB_CODE_SDB_ERROR);
       free(queueMsg);
       return;
     }
   } else if (pTable->type == TSDB_NORMAL_TABLE){
-    if (sdbDeleteRow(tsNormalTableSdb, pTable) < 0) {
+    SSdbOperDesc oper = {
+      .type = SDB_OPER_TYPE_GLOBAL,
+      .table = tsNormalTableSdb,
+      .pObj = pTable
+    };
+    int32_t code = sdbDeleteRow(&oper);
+    if (code != TSDB_CODE_SUCCESS) {
       mError("table:%s, update ntables sdb error", pTable->tableId);
       mgmtSendSimpleResp(queueMsg->thandle, TSDB_CODE_SDB_ERROR);
       free(queueMsg);
@@ -844,32 +869,48 @@ static void mgmtProcessDropStableRsp(SRpcMsg *rpcMsg) {
  mTrace("drop stable rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
 }
 
-//
-//
-//static void mgmtProcessTableCfgMsg(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle) {
-//  SDMConfigTableMsg *pCfg = (SDMConfigTableMsg *) pCont;
-//  pCfg->dnode = htonl(pCfg->dnode);
-//  pCfg->vnode = htonl(pCfg->vnode);
-//  pCfg->sid   = htonl(pCfg->sid);
-//  mTrace("dnode:%s, vnode:%d, sid:%d, receive table config msg", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
-//
-//  if (!sdbMaster) {
-//    mError("dnode:%s, vnode:%d, sid:%d, not master, redirect it", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
-//    mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_REDIRECT, NULL, 0);
-//    return;
-//  }
-//
-//  STableInfo *pTable = mgmtGetTableByPos(pCfg->dnode, pCfg->vnode, pCfg->sid);
-//  if (pTable == NULL) {
-//    mError("dnode:%s, vnode:%d, sid:%d, table not found", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
-//    mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_INVALID_TABLE, NULL, 0);
-//    return;
-//  }
-//
-//  mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_SUCCESS, NULL, 0);
-//
-//  //TODO
-//  SRpcIpSet ipSet = mgmtGetIpSetFromIp(pCfg->dnode);
-//  mgmtSendCreateTableMsg(NULL, &ipSet, NULL);
-//}
-//
\ No newline at end of file
+static void mgmtProcessTableCfgMsg(SRpcMsg *rpcMsg) {
+  if (mgmtCheckRedirect(rpcMsg->handle)) return;
+
+  SDMConfigTableMsg *pCfg = (SDMConfigTableMsg *) rpcMsg->pCont;
+  pCfg->dnode = htonl(pCfg->dnode);
+  pCfg->vnode = htonl(pCfg->vnode);
+  pCfg->sid   = htonl(pCfg->sid);
+  mTrace("dnode:%s, vnode:%d, sid:%d, receive table config msg", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
+
+  STableInfo *pTable = mgmtGetTableByPos(pCfg->dnode, pCfg->vnode, pCfg->sid);
+  if (pTable == NULL) {
+    mError("dnode:%s, vnode:%d, sid:%d, table not found", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
+    mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_NOT_ACTIVE_TABLE);
+    return;
+  }
+
+  mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_SUCCESS);
+
+  SMDCreateTableMsg *pMDCreate = NULL;
+  if (pTable->type == TSDB_CHILD_TABLE) {
+    mTrace("table:%s, is a child table, vgroup:%d sid:%d", pTable->tableId, pCfg->vnode, pCfg->sid);
+    pMDCreate = mgmtBuildCreateChildTableMsg(NULL, (SChildTableObj *) pTable);
+    if (pMDCreate == NULL) {
+      return;
+    }
+  } else if (pTable->type == TSDB_NORMAL_TABLE) {
+    mTrace("table:%s, is a normal table, vgroup:%d sid:%d", pTable->tableId, pCfg->vnode, pCfg->sid);
+    pMDCreate = mgmtBuildCreateNormalTableMsg((SNormalTableObj *) pTable);
+    if (pMDCreate == NULL) {
+      return;
+    }
+  } else {
+    mError("table:%s, invalid msg type, vgroup:%d sid:%d", pTable->tableId, pCfg->vnode, pCfg->sid);
+  }
+
+  SRpcIpSet ipSet = mgmtGetIpSetFromIp(pCfg->dnode);
+  SRpcMsg rpcRsp = {
+      .handle  = NULL,
+      .pCont   = pMDCreate,
+      .contLen = htonl(pMDCreate->contLen),
+      .code    = 0,
+      .msgType = TSDB_MSG_TYPE_MD_CREATE_TABLE
+  };
+  mgmtSendMsgToDnode(&ipSet, &rpcRsp);
+}

src/mnode/src/mgmtUser.c

@@ -17,9 +17,11 @@
 #include "os.h"
 #include "trpc.h"
 #include "ttime.h"
+#include "tutil.h"
 #include "mgmtAcct.h"
 #include "mgmtGrant.h"
 #include "mgmtMnode.h"
+#include "mgmtSdb.h"
 #include "mgmtShell.h"
 #include "mgmtUser.h"
 
@@ -36,48 +38,84 @@ static void mgmtProcessCreateUserMsg(SQueuedMsg *pMsg);
 static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg);
 static void mgmtProcessDropUserMsg(SQueuedMsg *pMsg);
 
-static void *(*mgmtUserActionFp[SDB_MAX_ACTION_TYPES])(void *row, char *str, int32_t size, int32_t *ssize);
-static void *mgmtUserActionInsert(void *row, char *str, int32_t size, int32_t *ssize);
-static void *mgmtUserActionDelete(void *row, char *str, int32_t size, int32_t *ssize);
-static void *mgmtUserActionUpdate(void *row, char *str, int32_t size, int32_t *ssize);
-static void *mgmtUserActionEncode(void *row, char *str, int32_t size, int32_t *ssize);
-static void *mgmtUserActionDecode(void *row, char *str, int32_t size, int32_t *ssize);
-static void *mgmtUserActionReset(void *row, char *str, int32_t size, int32_t *ssize);
-static void *mgmtUserActionDestroy(void *row, char *str, int32_t size, int32_t *ssize);
-static void *mgmtUserAction(char action, void *row, char *str, int32_t size, int32_t *ssize);
-static void  mgmtUserActionInit();
+static int32_t mgmtUserActionDestroy(SSdbOperDesc *pOper) {
+  tfree(pOper->pObj);
+  return TSDB_CODE_SUCCESS;
+}
 
-int32_t mgmtInitUsers() {
-  void     *pNode     = NULL;
-  SUserObj *pUser     = NULL;
-  SAcctObj *pAcct     = NULL;
-  int32_t  numOfUsers = 0;
+static int32_t mgmtUserActionInsert(SSdbOperDesc *pOper) {
+  SUserObj *pUser = pOper->pObj;
+  SAcctObj *pAcct = mgmtGetAcct(pUser->acct);
 
-  mgmtUserActionInit();
+  if (pAcct != NULL) {
+    mgmtAddUserIntoAcct(pAcct, pUser);
+  }
+  else {
+    mError("user:%s, acct:%s info not exist in sdb", pUser->user, pUser->acct);
+    return TSDB_CODE_INVALID_ACCT;
+  }
 
-  SUserObj tObj;
-  tsUserUpdateSize = tObj.updateEnd - (int8_t *)&tObj;
+  return TSDB_CODE_SUCCESS;
+}
 
-  tsUserSdb = sdbOpenTable(tsMaxUsers, tsUserUpdateSize, "users", SDB_KEYTYPE_STRING, tsMnodeDir, mgmtUserAction);
-  if (tsUserSdb == NULL) {
-    mError("failed to init user data");
+static int32_t mgmtUserActionDelete(SSdbOperDesc *pOper) {
+  SUserObj *pUser = pOper->pObj;
+  SAcctObj *pAcct = mgmtGetAcct(pUser->acct);
+
+  mgmtRemoveUserFromAcct(pAcct, pUser);
+
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t mgmtUserActionUpdate(SSdbOperDesc *pOper) {
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t mgmtUserActionEncode(SSdbOperDesc *pOper) {
+  SUserObj *pUser = pOper->pObj;
+
+  if (pOper->maxRowSize < tsUserUpdateSize) {
     return -1;
+  } else {
+    memcpy(pOper->rowData, pUser, tsUserUpdateSize);
+    pOper->rowSize = tsUserUpdateSize;
+    return TSDB_CODE_SUCCESS;
   }
+}
 
-  while (1) {
-    pNode = sdbFetchRow(tsUserSdb, pNode, (void **)&pUser);
-    if (pUser == NULL) break;
-
-    pUser->prev = NULL;
-    pUser->next = NULL;
+static int32_t mgmtUserActionDecode(SSdbOperDesc *pOper) {
+  SUserObj *pUser = (SUserObj *) calloc(1, sizeof(SUserObj));
+  if (pUser == NULL) return TSDB_CODE_SERV_OUT_OF_MEMORY;
 
-    pAcct = mgmtGetAcct(pUser->acct);
-    mgmtAddUserIntoAcct(pAcct, pUser);
+  memcpy(pUser, pOper->rowData, tsUserUpdateSize);
+  pOper->pObj = pUser;
+  return TSDB_CODE_SUCCESS;
+}
 
-    numOfUsers++;
+int32_t mgmtInitUsers() {
+  SUserObj tObj;
+  tsUserUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
+
+  SSdbTableDesc tableDesc = {
+    .tableName    = "users",
+    .hashSessions = TSDB_MAX_USERS,
+    .maxRowSize   = tsUserUpdateSize,
+    .keyType      = SDB_KEY_TYPE_STRING,
+    .insertFp     = mgmtUserActionInsert,
+    .deleteFp     = mgmtUserActionDelete,
+    .updateFp     = mgmtUserActionUpdate,
+    .encodeFp     = mgmtUserActionEncode,
+    .decodeFp     = mgmtUserActionDecode,
+    .destroyFp    = mgmtUserActionDestroy,
+  };
+
+  tsUserSdb = sdbOpenTable(&tableDesc);
+  if (tsUserSdb == NULL) {
+    mError("failed to init user data");
+    return -1;
   }
 
-  pAcct = mgmtGetAcct("root");
+  SAcctObj *pAcct = mgmtGetAcct("root");
   mgmtCreateUser(pAcct, "root", "taosdata");
   mgmtCreateUser(pAcct, "monitor", tsInternalPass);
   mgmtCreateUser(pAcct, "_root", tsInternalPass);
@@ -101,16 +139,23 @@ SUserObj *mgmtGetUser(char *name) {
 }
 
 static int32_t mgmtUpdateUser(SUserObj *pUser) {
-  return sdbUpdateRow(tsUserSdb, pUser, 0, 1);
+  SSdbOperDesc oper = {
+    .type = SDB_OPER_TYPE_GLOBAL,
+    .table = tsUserSdb,
+    .pObj = pUser,
+    .rowSize = tsUserUpdateSize
+  };
+
+  int32_t code = sdbUpdateRow(&oper);
+  if (code != TSDB_CODE_SUCCESS) {
+    tfree(pUser);
+    code = TSDB_CODE_SDB_ERROR;
+  }
+
+  return code;
 }
 
 static int32_t mgmtCreateUser(SAcctObj *pAcct, char *name, char *pass) {
-  int32_t numOfUsers = sdbGetNumOfRows(tsUserSdb);
-  if (numOfUsers >= tsMaxUsers) {
-    mWarn("numOfUsers:%d, exceed tsMaxUsers:%d", numOfUsers, tsMaxUsers);
-    return TSDB_CODE_TOO_MANY_USERS;
-  }
-
   int32_t code = mgmtCheckUserLimit(pAcct);
   if (code != 0) {
     return code;
@@ -131,8 +176,7 @@ static int32_t mgmtCreateUser(SAcctObj *pAcct, char *name, char *pass) {
     return code;
   }
 
-  pUser = malloc(sizeof(SUserObj));
-  memset(pUser, 0, sizeof(SUserObj));
+  pUser = calloc(1, sizeof(SUserObj));
   strcpy(pUser->user, name);
   taosEncryptPass((uint8_t*) pass, strlen(pass), pUser->pass);
   strcpy(pUser->acct, pAcct->user);
@@ -143,8 +187,15 @@ static int32_t mgmtCreateUser(SAcctObj *pAcct, char *name, char *pass) {
     pUser->superAuth = 1;
   }
 
-  code = TSDB_CODE_SUCCESS;
-  if (sdbInsertRow(tsUserSdb, pUser, 0) < 0) {
+  SSdbOperDesc oper = {
+    .type = SDB_OPER_TYPE_GLOBAL,
+    .table = tsUserSdb,
+    .pObj = pUser,
+    .rowSize = sizeof(SUserObj)
+  };
+
+  code = sdbInsertRow(&oper);
+  if (code != TSDB_CODE_SUCCESS) {
     tfree(pUser);
     code = TSDB_CODE_SDB_ERROR;
   }
@@ -165,9 +216,18 @@ static int32_t mgmtDropUser(SAcctObj *pAcct, char *name) {
     return TSDB_CODE_NO_RIGHTS;
   }
 
-  sdbDeleteRow(tsUserSdb, pUser);
+  SSdbOperDesc oper = {
+    .type = SDB_OPER_TYPE_GLOBAL,
+    .table = tsUserSdb,
+    .pObj = pUser
+  };
 
-  return 0;
+  int32_t code = sdbDeleteRow(&oper);
+  if (code != TSDB_CODE_SUCCESS) {
+    code = TSDB_CODE_SDB_ERROR;
+  }
+
+  return code;
 }
 
 static int32_t mgmtGetUserMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
@@ -250,83 +310,6 @@ static int32_t mgmtRetrieveUsers(SShowObj *pShow, char *data, int32_t rows, void
   return numOfRows;
 }
 
-static void mgmtUserActionInit() {
-  mgmtUserActionFp[SDB_TYPE_INSERT]  = mgmtUserActionInsert;
-  mgmtUserActionFp[SDB_TYPE_DELETE]  = mgmtUserActionDelete;
-  mgmtUserActionFp[SDB_TYPE_UPDATE]  = mgmtUserActionUpdate;
-  mgmtUserActionFp[SDB_TYPE_ENCODE]  = mgmtUserActionEncode;
-  mgmtUserActionFp[SDB_TYPE_DECODE]  = mgmtUserActionDecode;
-  mgmtUserActionFp[SDB_TYPE_RESET]   = mgmtUserActionReset;
-  mgmtUserActionFp[SDB_TYPE_DESTROY] = mgmtUserActionDestroy;
-}
-
-static void *mgmtUserAction(char action, void *row, char *str, int32_t size, int32_t *ssize) {
-  if (mgmtUserActionFp[(uint8_t) action] != NULL) {
-    return (*(mgmtUserActionFp[(uint8_t) action]))(row, str, size, ssize);
-  }
-  return NULL;
-}
-
-static void *mgmtUserActionInsert(void *row, char *str, int32_t size, int32_t *ssize) {
-  SUserObj *pUser = (SUserObj *) row;
-  SAcctObj *pAcct = mgmtGetAcct(pUser->acct);
-
-  pUser->pAcct = pAcct;
-  mgmtAddUserIntoAcct(pAcct, pUser);
-
-  return NULL;
-}
-
-static void *mgmtUserActionDelete(void *row, char *str, int32_t size, int32_t *ssize) {
-  SUserObj *pUser = (SUserObj *) row;
-  SAcctObj *pAcct = mgmtGetAcct(pUser->acct);
-
-  mgmtRemoveUserFromAcct(pAcct, pUser);
-
-  return NULL;
-}
-
-static void *mgmtUserActionUpdate(void *row, char *str, int32_t size, int32_t *ssize) {
-  return mgmtUserActionReset(row, str, size, ssize);
-}
-
-static void *mgmtUserActionEncode(void *row, char *str, int32_t size, int32_t *ssize) {
-  SUserObj *pUser = (SUserObj *) row;
-
-  if (size < tsUserUpdateSize) {
-    *ssize = -1;
-  } else {
-    memcpy(str, pUser, tsUserUpdateSize);
-    *ssize = tsUserUpdateSize;
-  }
-
-  return NULL;
-}
-
-static void *mgmtUserActionDecode(void *row, char *str, int32_t size, int32_t *ssize) {
-  SUserObj *pUser = (SUserObj *) malloc(sizeof(SUserObj));
-  if (pUser == NULL) return NULL;
-  memset(pUser, 0, sizeof(SUserObj));
-
-  memcpy(pUser, str, tsUserUpdateSize);
-
-  return (void *)pUser;
-}
-
-static void *mgmtUserActionReset(void *row, char *str, int32_t size, int32_t *ssize) {
-  SUserObj *pUser = (SUserObj *)row;
-
-  memcpy(pUser, str, tsUserUpdateSize);
-
-  return NULL;
-}
-
-static void *mgmtUserActionDestroy(void *row, char *str, int32_t size, int32_t *ssize) {
-  tfree(row);
-
-  return NULL;
-}
-
 SUserObj *mgmtGetUserFromConn(void *pConn) {
   SRpcConnInfo connInfo;
   if (rpcGetConnInfo(pConn, &connInfo) == 0) {
@@ -393,7 +376,7 @@ static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg) {
       memset(pUser->pass, 0, sizeof(pUser->pass));
       taosEncryptPass((uint8_t*)pAlter->pass, strlen(pAlter->pass), pUser->pass);
       code = mgmtUpdateUser(pUser);
-      mLPrint("user:%s password is altered by %s, code:%d", pAlter->user, pUser->user, code);
+      mLPrint("user:%s password is altered by %s, result:%d", pUser->user, pOperUser->user, tstrerror(code));
     } else {
       code = TSDB_CODE_NO_RIGHTS;
     }
@@ -428,10 +411,6 @@ static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg) {
     }
 
     if (hasRight) {
-      //if (pAlter->privilege == 1) {  // super
-      //  pUser->superAuth = 1;
-      //  pUser->writeAuth = 1;
-      //}
       if (pAlter->privilege == 2) {  // read
         pUser->superAuth = 0;
         pUser->writeAuth = 0;
@@ -442,7 +421,7 @@ static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg) {
       }
 
       code = mgmtUpdateUser(pUser);
-      mLPrint("user:%s privilege is altered by %s, code:%d", pAlter->user, pUser->user, code);
+      mLPrint("user:%s privilege is altered by %s, result:%d", pUser->user, pOperUser->user, tstrerror(code));
     } else {
       code = TSDB_CODE_NO_RIGHTS;
     }
@@ -492,7 +471,7 @@ static void mgmtProcessDropUserMsg(SQueuedMsg *pMsg) {
   if (hasRight) {
     code = mgmtDropUser(pUser->pAcct, pDrop->user);
     if (code == TSDB_CODE_SUCCESS) {
-      mLPrint("user:%s is dropped by %s", pDrop->user, pUser->user);
+       mLPrint("user:%s is dropped by %s, result:%d", pUser->user, pOperUser->user, tstrerror(code));
     }
   } else {
     code = TSDB_CODE_NO_RIGHTS;

src/sdb/CMakeLists.txt

-CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
-PROJECT(TDengine)
-
-INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc)
-INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/inc)
-INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
-INCLUDE_DIRECTORIES(inc)
-
-IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
-    AUX_SOURCE_DIRECTORY(src SRC)
-    ADD_LIBRARY(sdb ${SRC})
-    TARGET_LINK_LIBRARIES(sdb trpc)
-    IF (TD_CLUSTER)
-        TARGET_LINK_LIBRARIES(sdb)
-    ENDIF()
-ENDIF ()

src/sdb/inc/sdbint.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 _sdbint_header_
-#define _sdbint_header_
-
-#include <errno.h>
-#include <pthread.h>
-#include <semaphore.h>
-#include <signal.h>
-#include <stdio.h>
-#include <stdlib.h>
-#include <string.h>
-#include <unistd.h>
-
-#include "hashint.h"
-#include "hashstr.h"
-#include "sdb.h"
-#include "tchecksum.h"
-#include "tlog.h"
-#include "trpc.h"
-#include "tutil.h"
-
-#define sdbError(...)                            \
-  if (sdbDebugFlag & DEBUG_ERROR) {              \
-    tprintf("ERROR MND-SDB ", 255, __VA_ARGS__); \
-  }
-#define sdbWarn(...)                                      \
-  if (sdbDebugFlag & DEBUG_WARN) {                        \
-    tprintf("WARN  MND-SDB ", sdbDebugFlag, __VA_ARGS__); \
-  }
-#define sdbTrace(...)                               \
-  if (sdbDebugFlag & DEBUG_TRACE) {                 \
-    tprintf("MND-SDB ", sdbDebugFlag, __VA_ARGS__); \
-  }
-#define sdbPrint(...) \
-  { tprintf("MND-SDB ", 255, __VA_ARGS__); }
-
-#define mpeerError(...)                            \
-  if (sdbDebugFlag & DEBUG_ERROR) {              \
-    tprintf("ERROR MND-MPEER ", 255, __VA_ARGS__); \
-  }
-#define mpeerWarn(...)                                      \
-  if (sdbDebugFlag & DEBUG_WARN) {                        \
-    tprintf("WARN  MND-MPEER ", sdbDebugFlag, __VA_ARGS__); \
-  }
-#define mpeerTrace(...)                               \
-  if (sdbDebugFlag & DEBUG_TRACE) {                 \
-    tprintf("MND-MPEER ", sdbDebugFlag, __VA_ARGS__); \
-  }
-#define mpeerPrint(...) \
-  { tprintf("MND-MPEER ", 255, __VA_ARGS__); }
-
-#define sdbLError(...) taosLogError(__VA_ARGS__) sdbError(__VA_ARGS__)
-#define sdbLWarn(...) taosLogWarn(__VA_ARGS__) sdbWarn(__VA_ARGS__)
-#define sdbLPrint(...) taosLogPrint(__VA_ARGS__) sdbPrint(__VA_ARGS__)
-
-#define SDB_MAX_PEERS 4
-#define SDB_DELIMITER 0xFFF00F00
-#define SDB_ENDCOMMIT 0xAFFFAAAF
-
-typedef struct {
-  uint64_t swVersion;
-  int16_t  sdbFileVersion;
-  char     reserved[6];
-  TSCKSUM  checkSum;
-} SSdbHeader;
-
-typedef struct {
-  char type;
-  // short  rowSize;
-  char *row;
-} SSdbUpdate;
-
-typedef struct _SSdbTable {
-  SSdbHeader header;
-  int        maxRows;
-  int        dbId;
-  int32_t    maxRowSize;
-  char       name[TSDB_DB_NAME_LEN];
-  char       fn[128];
-  int        keyType;
-  uint32_t   autoIndex;
-  int64_t    numOfRows;
-  int64_t    id;
-  int64_t    size;
-  void *     iHandle;
-  int        fd;
-  void *(*appTool)(char, void *, char *, int, int *);
-  pthread_mutex_t mutex;
-  SSdbUpdate *    update;
-  int             numOfUpdates;
-  int             updatePos;
-} SSdbTable;
-
-typedef struct {
-  int64_t id;
-  int64_t offset;
-  int     rowSize;
-  void *  row;
-} SRowMeta;
-
-typedef struct {
-  int32_t delimiter;
-  int32_t rowSize;
-  int64_t id;
-  char    data[];
-} SRowHead;
-
-typedef struct {
-  uint8_t  dbId;
-  char     type;
-  uint64_t version;
-  short    dataLen;
-  char     data[];
-} SForwardMsg;
-
-extern SSdbTable *tableList[];
-extern int        sdbMaxPeers;
-extern int        sdbNumOfTables;
-extern int64_t    sdbVersion;
-
-int sdbForwardDbReqToPeer(SSdbTable *pTable, char type, char *data, int dataLen);
-int mpeerRetrieveRows(int fd, SSdbTable *pTable, uint64_t version);
-void sdbResetTable(SSdbTable *pTable);
-extern const int16_t sdbFileVersion;
-
-
-#endif

src/util/inc/tglobalcfg.h

@@ -103,13 +103,7 @@ extern int   tsReplications;
 
 extern int  tsNumOfMPeers;
 extern int  tsMaxShellConns;
-extern int  tsMaxAccounts;
-extern int  tsMaxUsers;
-extern int  tsMaxDbs;
 extern int  tsMaxTables;
-extern int  tsMaxDnodes;
-extern int  tsMaxVGroups;
-extern char tsMgmtZone[];
 
 extern char tsLocalIp[];
 extern char tsDefaultDB[];

src/util/inc/tidpool.h

@@ -34,11 +34,7 @@ void taosIdPoolCleanUp(void *handle);
 
 int taosIdPoolNumOfUsed(void *handle);
 
-void taosIdPoolReinit(void *handle);
-
-void taosIdPoolMarkStatus(void *handle, int id, int status);
-
-void taosIdPoolSetFreeList(void *handle);
+void taosIdPoolMarkStatus(void *handle, int id);
 
 #ifdef __cplusplus
 }

src/util/inc/tlog.h

@@ -239,6 +239,25 @@ extern uint32_t cdebugFlag;
 #define monitorLWarn(...) taosLogWarn(__VA_ARGS__) monitorWarn(__VA_ARGS__)
 #define monitorLPrint(...) taosLogPrint(__VA_ARGS__) monitorPrint(__VA_ARGS__)
 
+#define sdbError(...)                            \
+  if (sdbDebugFlag & DEBUG_ERROR) {              \
+    tprintf("ERROR MND-SDB ", 255, __VA_ARGS__); \
+  }
+#define sdbWarn(...)                                      \
+  if (sdbDebugFlag & DEBUG_WARN) {                        \
+    tprintf("WARN  MND-SDB ", sdbDebugFlag, __VA_ARGS__); \
+  }
+#define sdbTrace(...)                               \
+  if (sdbDebugFlag & DEBUG_TRACE) {                 \
+    tprintf("MND-SDB ", sdbDebugFlag, __VA_ARGS__); \
+  }
+#define sdbPrint(...) \
+  { tprintf("MND-SDB ", 255, __VA_ARGS__); }
+
+#define sdbLError(...) taosLogError(__VA_ARGS__) sdbError(__VA_ARGS__)
+#define sdbLWarn(...) taosLogWarn(__VA_ARGS__) sdbWarn(__VA_ARGS__)
+#define sdbLPrint(...) taosLogPrint(__VA_ARGS__) sdbPrint(__VA_ARGS__)
+
 #ifdef __cplusplus
 }
 #endif

src/util/src/tglobalcfg.c

@@ -112,13 +112,7 @@ int   tsReplications = TSDB_REPLICA_MIN_NUM;
 
 int  tsNumOfMPeers = 3;
 int  tsMaxShellConns = 2000;
-int  tsMaxAccounts = 100;
-int  tsMaxUsers = 1000;
-int  tsMaxDbs = 1000;
-int  tsMaxTables = 650000;
-int  tsMaxDnodes = 1000;
-int  tsMaxVGroups = 1000;
-char tsMgmtZone[16] = "rzone";
+int  tsMaxTables = 100000;
 
 char tsLocalIp[TSDB_IPv4ADDR_LEN] = {0};
 char tsDefaultDB[TSDB_DB_NAME_LEN] = {0};
@@ -612,27 +606,9 @@ static void doInitGlobalConfig() {
                      1, 8640000, 0, TSDB_CFG_UTYPE_SECOND);
 
   // mgmt configs
-  tsInitConfigOption(cfg++, "mgmtZone", tsMgmtZone, TSDB_CFG_VTYPE_STRING,
-                     TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW | TSDB_CFG_CTYPE_B_CLUSTER,
-                     0, 0, 16, TSDB_CFG_UTYPE_NONE);
-  tsInitConfigOption(cfg++, "maxAccounts", &tsMaxAccounts, TSDB_CFG_VTYPE_INT,
-                     TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW | TSDB_CFG_CTYPE_B_CLUSTER,
-                     1, 1000, 0, TSDB_CFG_UTYPE_NONE);
-  tsInitConfigOption(cfg++, "maxUsers", &tsMaxUsers, TSDB_CFG_VTYPE_INT,
-                     TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW,
-                     1, 1000, 0, TSDB_CFG_UTYPE_NONE);
-  tsInitConfigOption(cfg++, "maxDbs", &tsMaxDbs, TSDB_CFG_VTYPE_INT,
-                     TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW,
-                     1, 10000, 0, TSDB_CFG_UTYPE_NONE);
   tsInitConfigOption(cfg++, "maxTables", &tsMaxTables, TSDB_CFG_VTYPE_INT,
                      TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW,
                      1, 100000000, 0, TSDB_CFG_UTYPE_NONE);
-  tsInitConfigOption(cfg++, "maxDnodes", &tsMaxDnodes, TSDB_CFG_VTYPE_INT,
-                     TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW | TSDB_CFG_CTYPE_B_CLUSTER,
-                     1, 1000, 0, TSDB_CFG_UTYPE_NONE);
-  tsInitConfigOption(cfg++, "maxVGroups", &tsMaxVGroups, TSDB_CFG_VTYPE_INT,
-                     TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW,
-                     1, 1000000, 0, TSDB_CFG_UTYPE_NONE);
   
   tsInitConfigOption(cfg++, "minSlidingTime", &tsMinSlidingTime, TSDB_CFG_VTYPE_INT,
                      TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW,

src/util/src/tidpool.c

@@ -15,88 +15,81 @@
 
 #include "os.h"
 #include "tlog.h"
+#include <stdbool.h>
 
 typedef struct {
   int             maxId;
   int             numOfFree;
   int             freeSlot;
-  int *           freeList;
+  bool *          freeList;
   pthread_mutex_t mutex;
 } id_pool_t;
 
 void *taosInitIdPool(int maxId) {
-  id_pool_t *pIdPool;
-  int *      idList, i;
-
-  if (maxId < 3) maxId = 3;
-
-  pIdPool = (id_pool_t *)malloc(sizeof(id_pool_t));
+  id_pool_t *pIdPool = calloc(1, sizeof(id_pool_t));
   if (pIdPool == NULL) return NULL;
 
-  idList = (int *)malloc(sizeof(int) * (size_t)maxId);
-  if (idList == NULL) {
+  pIdPool->freeList = calloc(maxId, sizeof(bool));
+  if (pIdPool->freeList == NULL) {
     free(pIdPool);
     return NULL;
   }
 
-  memset(pIdPool, 0, sizeof(id_pool_t));
   pIdPool->maxId = maxId;
-  pIdPool->numOfFree = maxId - 1;
+  pIdPool->numOfFree = maxId;
   pIdPool->freeSlot = 0;
-  pIdPool->freeList = idList;
 
   pthread_mutex_init(&pIdPool->mutex, NULL);
 
-  for (i = 1; i < maxId; ++i) idList[i - 1] = i;
-
   pTrace("pool:%p is setup, maxId:%d", pIdPool, pIdPool->maxId);
 
-  return (void *)pIdPool;
+  return pIdPool;
 }
 
 int taosAllocateId(void *handle) {
-  id_pool_t *pIdPool;
-  int        id = -1;
-  if (handle == NULL) return id;
-
-  pIdPool = (id_pool_t *)handle;
-
-  if (pIdPool->maxId < 3) pError("pool:%p is messed up, maxId:%d", pIdPool, pIdPool->maxId);
+  id_pool_t *pIdPool = handle;
+  if (handle == NULL) {
+    return -1;
+  }
 
-  if (pthread_mutex_lock(&pIdPool->mutex) != 0) perror("lock pIdPool Mutex");
+  int slot = -1;
+  pthread_mutex_lock(&pIdPool->mutex);
 
   if (pIdPool->numOfFree > 0) {
-    id = pIdPool->freeList[pIdPool->freeSlot];
-    pIdPool->freeSlot = (pIdPool->freeSlot + 1) % pIdPool->maxId;
+    for (int i = 0; i < pIdPool->maxId; ++i) {
+      slot = (i + pIdPool->freeSlot) % pIdPool->maxId;
+      if (!pIdPool->freeList[slot]) {
+        pIdPool->freeList[slot] = true;
+        pIdPool->freeSlot = slot + 1;
         pIdPool->numOfFree--;
+        break;
+      }
+    }
   }
 
-  if (pthread_mutex_unlock(&pIdPool->mutex) != 0) perror("unlock pIdPool Mutex");
-
-  return id;
+  pthread_mutex_unlock(&pIdPool->mutex);
+  return slot + 1;
 }
 
 void taosFreeId(void *handle, int id) {
-  id_pool_t *pIdPool;
-  int        slot;
+  id_pool_t *pIdPool = handle;
+  if (handle == NULL) return;
 
-  pIdPool = (id_pool_t *)handle;
-  if (pIdPool->freeList == NULL || pIdPool->maxId == 0) return;
-  if (id <= 0 || id >= pIdPool->maxId) return;
-  if (pthread_mutex_lock(&pIdPool->mutex) != 0) perror("lock pIdPool Mutex");
+  pthread_mutex_lock(&pIdPool->mutex);
 
-  slot = (pIdPool->freeSlot + pIdPool->numOfFree) % pIdPool->maxId;
-  pIdPool->freeList[slot] = id;
+  int slot = (id - 1) % pIdPool->maxId;
+  if (pIdPool->freeList[slot]) {
+    pIdPool->freeList[slot] = false;
     pIdPool->numOfFree++;
+  }
 
-  if (pthread_mutex_unlock(&pIdPool->mutex) != 0) perror("unlock pIdPool Mutex");
+  pthread_mutex_unlock(&pIdPool->mutex);
 }
 
 void taosIdPoolCleanUp(void *handle) {
-  id_pool_t *pIdPool;
+  id_pool_t *pIdPool = handle;
 
-  if (handle == NULL) return;
-  pIdPool = (id_pool_t *)handle;
+  if (pIdPool == NULL) return;
 
   pTrace("pool:%p is cleaned", pIdPool);
 
@@ -110,42 +103,21 @@ void taosIdPoolCleanUp(void *handle) {
 }
 
 int taosIdPoolNumOfUsed(void *handle) {
-  id_pool_t *pIdPool = (id_pool_t *)handle;
-
-  return pIdPool->maxId - pIdPool->numOfFree - 1;
-}
-
-void taosIdPoolReinit(void *handle) {
-  id_pool_t *pIdPool;
-
-  pIdPool = (id_pool_t *)handle;
-  pIdPool->numOfFree = 0;
-  pIdPool->freeSlot = 0;
-
-  for (int i = 0; i < pIdPool->maxId; ++i) pIdPool->freeList[i] = 0;
-}
-
-void taosIdPoolMarkStatus(void *handle, int id, int status) {
-  id_pool_t *pIdPool = (id_pool_t *)handle;
-
-  pIdPool->freeList[id] = status;
+  id_pool_t *pIdPool = handle;
+  return pIdPool->maxId - pIdPool->numOfFree;
 }
 
-void taosIdPoolSetFreeList(void *handle) {
-  id_pool_t *pIdPool;
-  int        pos = 0;
-
-  pIdPool = (id_pool_t *)handle;
-  pIdPool->numOfFree = 0;
-  pIdPool->freeSlot = 0;
+void taosIdPoolMarkStatus(void *handle, int id) {
+  id_pool_t *pIdPool = handle;
+  pthread_mutex_lock(&pIdPool->mutex);
 
-  for (int i = 1; i < pIdPool->maxId; ++i) {
-    if (pIdPool->freeList[i] == 0) {
-      pIdPool->freeList[pos] = i;
-      pIdPool->numOfFree++;
-      pos++;
-    }
+  int slot = (id - 1) % pIdPool->maxId;
+  if (!pIdPool->freeList[slot]) {
+    pIdPool->freeList[slot] = true;
+    pIdPool->numOfFree--;
   }
+
+  pthread_mutex_unlock(&pIdPool->mutex);
 }
 
 int taosUpdateIdPool(id_pool_t *handle, int maxId) {
@@ -154,18 +126,14 @@ int taosUpdateIdPool(id_pool_t *handle, int maxId) {
     return -1;
   }
 
-  int *idList, i;
-  idList = (int *)malloc(sizeof(int) * (size_t)maxId);
+  int *idList = calloc(maxId, sizeof(bool));
   if (idList == NULL) {
     return -1;
   }
-  for (i = 1; i < maxId; ++i) {
-    idList[i - 1] = i;
-  }
 
-  if (pthread_mutex_lock(&pIdPool->mutex) != 0) perror("lock pIdPool Mutex");
+  pthread_mutex_lock(&pIdPool->mutex);
 
-  memcpy(idList, pIdPool->freeList, sizeof(int) * (size_t)pIdPool->maxId);
+  memcpy(idList, pIdPool->freeList, sizeof(bool) * pIdPool->maxId);
   pIdPool->numOfFree += (maxId - pIdPool->maxId);
   pIdPool->maxId = maxId;
 
@@ -173,7 +141,7 @@ int taosUpdateIdPool(id_pool_t *handle, int maxId) {
   pIdPool->freeList = idList;
   free(oldIdList);
 
-  if (pthread_mutex_unlock(&pIdPool->mutex) != 0) perror("unlock pIdPool Mutex");
+  pthread_mutex_unlock(&pIdPool->mutex);
 
   return 0;
 }