Merge branch 'develop' into feature/query

src/client/src/tscSecondaryMerge.c

@@ -353,7 +353,7 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
   pRes->numOfGroups = 0;
 
   STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
-  STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);;
+  STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);
   
   TSKEY stime = MIN(pQueryInfo->window.skey, pQueryInfo->window.ekey);
   int64_t revisedSTime =

src/client/src/tscServer.c

@@ -1474,7 +1474,7 @@ int tscBuildTableMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
     pMsg += len;
   }
 
-  pCmd->payloadLen = pMsg - (char*)pInfoMsg;;
+  pCmd->payloadLen = pMsg - (char*)pInfoMsg;
   pCmd->msgType = TSDB_MSG_TYPE_CM_TABLE_META;
 
   tfree(tmpData);

src/client/src/tscStream.c

@@ -499,7 +499,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
   if (pSql->sqlstr == NULL) {
     tscError("%p failed to malloc sql string buffer", pSql);
     tscFreeSqlObj(pSql);
-    return NULL;;
+    return NULL;
   }
   strtolower(pSql->sqlstr, sqlstr);
 

src/dnode/src/dnodeMWrite.c

@@ -118,6 +118,8 @@ void dnodeDispatchToMnodeWriteQueue(SRpcMsg *pMsg) {
 
   SMnodeMsg *pWrite = (SMnodeMsg *)taosAllocateQitem(sizeof(SMnodeMsg));
   mnodeCreateMsg(pWrite, pMsg);
+
+  dTrace("app:%p:%p, msg:%s is put into mwrite queue", pWrite->rpcMsg.ahandle, pWrite, taosMsg[pWrite->rpcMsg.msgType]);
   taosWriteQitem(tsMWriteQueue, TAOS_QTYPE_RPC, pWrite);
 }
 
@@ -128,6 +130,7 @@ static void dnodeFreeMnodeWriteMsg(SMnodeMsg *pWrite) {
 
 void dnodeSendRpcMnodeWriteRsp(void *pRaw, int32_t code) {
   SMnodeMsg *pWrite = pRaw;
+  if (pWrite == NULL) return;
   if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) return;
   if (code == TSDB_CODE_MND_ACTION_NEED_REPROCESSED) {
     dnodeReprocessMnodeWriteMsg(pWrite);
@@ -146,19 +149,21 @@ void dnodeSendRpcMnodeWriteRsp(void *pRaw, int32_t code) {
 }
 
 static void *dnodeProcessMnodeWriteQueue(void *param) {
-  SMnodeMsg *pWriteMsg;
+  SMnodeMsg *pWrite;
   int32_t    type;
   void *     unUsed;
   
   while (1) {
-    if (taosReadQitemFromQset(tsMWriteQset, &type, (void **)&pWriteMsg, &unUsed) == 0) {
+    if (taosReadQitemFromQset(tsMWriteQset, &type, (void **)&pWrite, &unUsed) == 0) {
       dTrace("dnodeProcessMnodeWriteQueue: got no message from qset, exiting...");
       break;
     }
 
-    dTrace("%p, msg:%s will be processed in mwrite queue", pWriteMsg->rpcMsg.ahandle, taosMsg[pWriteMsg->rpcMsg.msgType]);    
-    int32_t code = mnodeProcessWrite(pWriteMsg);    
-    dnodeSendRpcMnodeWriteRsp(pWriteMsg, code);
+    dTrace("app:%p:%p, msg:%s will be processed in mwrite queue", pWrite->rpcMsg.ahandle, pWrite,
+           taosMsg[pWrite->rpcMsg.msgType]);
+
+    int32_t code = mnodeProcessWrite(pWrite);
+    dnodeSendRpcMnodeWriteRsp(pWrite, code);
   }
 
   return NULL;
@@ -168,9 +173,15 @@ void dnodeReprocessMnodeWriteMsg(void *pMsg) {
   SMnodeMsg *pWrite = pMsg;
 
   if (!mnodeIsRunning() || tsMWriteQueue == NULL) {
+    dTrace("app:%p:%p, msg:%s is redirected for mnode not running, retry times:%d", pWrite->rpcMsg.ahandle, pWrite,
+           taosMsg[pWrite->rpcMsg.msgType], pWrite->retry);
+
     dnodeSendRedirectMsg(pMsg, true);
     dnodeFreeMnodeWriteMsg(pWrite);
   } else {
+    dTrace("app:%p:%p, msg:%s is reput into mwrite queue, retry times:%d", pWrite->rpcMsg.ahandle, pWrite,
+           taosMsg[pWrite->rpcMsg.msgType], pWrite->retry);
+
     taosWriteQitem(tsMWriteQueue, TAOS_QTYPE_RPC, pWrite);
   }
 }

src/inc/taoserror.h

@@ -113,8 +113,12 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_QUERY_ID,         0, 0x030C, "mnode inva
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_STREAM_ID,        0, 0x030D, "mnode invalid stream id")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_CONN_ID,          0, 0x030E, "mnode invalid connection")
 
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_SDB_OBJ_ALREADY_THERE,    0, 0x0320, "mnode object already there")
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_SDB_ERROR,                0, 0x0321, "mnode sdb error")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_SDB_OBJ_ALREADY_THERE,    0, 0x0320, "[sdb] object already there")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_SDB_ERROR,                0, 0x0321, "[sdb] app error")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_SDB_INVALID_TABLE_TYPE,   0, 0x0322, "[sdb] invalid table type")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_SDB_OBJ_NOT_THERE,        0, 0x0323, "[sdb] object not there")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_SDB_INVAID_META_ROW,      0, 0x0324, "[sdb] invalid meta row")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_SDB_INVAID_KEY_TYPE,      0, 0x0325, "[sdb] invalid key type")
 
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_DNODE_ALREADY_EXIST,      0, 0x0330, "mnode dnode already exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_DNODE_NOT_EXIST,          0, 0x0331, "mnode dnode not exist")

src/mnode/inc/mnodeDnode.h

@@ -47,7 +47,7 @@ void    mnodeDecDnodeRef(SDnodeObj *pDnode);
 void *  mnodeGetDnode(int32_t dnodeId);
 void *  mnodeGetDnodeByEp(char *ep);
 void    mnodeUpdateDnode(SDnodeObj *pDnode);
-int32_t mnodeDropDnode(SDnodeObj *pDnode);
+int32_t mnodeDropDnode(SDnodeObj *pDnode, void *pMsg);
 
 extern int32_t tsAccessSquence;
 

src/mnode/inc/mnodeInt.h

@@ -36,10 +36,10 @@ extern int32_t sdbDebugFlag;
 #define mLWarn(...)  { monitorSaveLog(1, __VA_ARGS__); mWarn(__VA_ARGS__)  }
 #define mLPrint(...) { monitorSaveLog(0, __VA_ARGS__); mPrint(__VA_ARGS__) }
 
-#define sdbError(...) { if (sdbDebugFlag & DEBUG_ERROR) { taosPrintLog("ERROR MND-SDB ", 255, __VA_ARGS__); }}
-#define sdbWarn(...)  { if (sdbDebugFlag & DEBUG_WARN)  { taosPrintLog("WARN MND-SDB ", sdbDebugFlag, __VA_ARGS__); }}
-#define sdbTrace(...) { if (sdbDebugFlag & DEBUG_TRACE) { taosPrintLog("MND-SDB ", sdbDebugFlag, __VA_ARGS__);}}
-#define sdbPrint(...) { taosPrintLog("MND-SDB ", 255, __VA_ARGS__); }
+#define sdbError(...) { if (sdbDebugFlag & DEBUG_ERROR) { taosPrintLog("ERROR SDB ", 255, __VA_ARGS__); }}
+#define sdbWarn(...)  { if (sdbDebugFlag & DEBUG_WARN)  { taosPrintLog("WARN SDB ", sdbDebugFlag, __VA_ARGS__); }}
+#define sdbTrace(...) { if (sdbDebugFlag & DEBUG_TRACE) { taosPrintLog("SDB ", sdbDebugFlag, __VA_ARGS__);}}
+#define sdbPrint(...) { taosPrintLog("SDB ", 255, __VA_ARGS__); }
 
 #define sdbLError(...) { monitorSaveLog(2, __VA_ARGS__); sdbError(__VA_ARGS__) }
 #define sdbLWarn(...)  { monitorSaveLog(1, __VA_ARGS__); sdbWarn(__VA_ARGS__)  }

src/mnode/inc/mnodeSdb.h

@@ -20,6 +20,8 @@
 extern "C" {
 #endif
 
+struct SMnodeMsg;
+
 typedef enum {
   SDB_TABLE_DNODE   = 0,
   SDB_TABLE_MNODE   = 1,
@@ -48,8 +50,11 @@ typedef struct {
   ESdbOper type;
   void *   table;
   void *   pObj;
-  int32_t  rowSize;
   void *   rowData;
+  int32_t  rowSize;
+  int32_t  retCode; // for callback in sdb queue
+  int32_t  (*cb)(struct SMnodeMsg *pMsg, int32_t code);
+  struct SMnodeMsg *pMsg;
 } SSdbOper;
 
 typedef struct {

src/mnode/inc/mnodeUser.h

@@ -28,7 +28,8 @@ void *    mnodeGetNextUser(void *pIter, SUserObj **pUser);
 void      mnodeIncUserRef(SUserObj *pUser);
 void      mnodeDecUserRef(SUserObj *pUser);
 SUserObj *mnodeGetUserFromConn(void *pConn);
-int32_t   mnodeCreateUser(SAcctObj *pAcct, char *name, char *pass);
+char *    mnodeGetUserFromMsg(void *pMnodeMsg);
+int32_t   mnodeCreateUser(SAcctObj *pAcct, char *name, char *pass, void *pMsg);
 void      mnodeDropAllUsers(SAcctObj *pAcct);
 
 #ifdef __cplusplus

src/mnode/src/mnodeAcct.c

@@ -78,7 +78,9 @@ static int32_t mnodeAcctActionDecode(SSdbOper *pOper) {
 }
 
 static int32_t mnodeAcctActionRestored() {
-  if (dnodeIsFirstDeploy()) {
+  int32_t numOfRows = sdbGetNumOfRows(tsAcctSdb);
+  if (numOfRows <= 0 && dnodeIsFirstDeploy()) {
+    mPrint("dnode first deploy, create root acct");
     int32_t code = mnodeCreateRootAcct();
     if (code != TSDB_CODE_SUCCESS) {
       mError("failed to create root account, reason:%s", tstrerror(code));

src/mnode/src/mnodeDb.c

@@ -41,7 +41,7 @@
 static void *  tsDbSdb = NULL;
 static int32_t tsDbUpdateSize;
 
-static int32_t mnodeCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate);
+static int32_t mnodeCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate, void *pMsg);
 static int32_t mnodeDropDb(SMnodeMsg *newMsg);
 static int32_t mnodeSetDbDropping(SDbObj *pDb);
 static int32_t mnodeGetDbMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
@@ -311,7 +311,7 @@ static void mnodeSetDefaultDbCfg(SDbCfg *pCfg) {
   if (pCfg->replications < 0) pCfg->replications = tsReplications;
 }
 
-static int32_t mnodeCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate) {
+static int32_t mnodeCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate, void *pMsg) {
   int32_t code = acctCheck(pAcct, ACCT_GRANT_DB);
   if (code != 0) return code;
 
@@ -364,12 +364,15 @@ static int32_t mnodeCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate) {
     .table   = tsDbSdb,
     .pObj    = pDb,
     .rowSize = sizeof(SDbObj),
+    .pMsg    = pMsg
   };
 
   code = sdbInsertRow(&oper);
   if (code != TSDB_CODE_SUCCESS) {
     tfree(pDb);
-    code = TSDB_CODE_MND_SDB_ERROR;
+  } else {
+    mLPrint("db:%s, is created by %s", pDb->name, mnodeGetUserFromMsg(pMsg));
+    if (pMsg != NULL) code = TSDB_CODE_MND_ACTION_IN_PROGRESS;
   }
 
   return code;
@@ -771,12 +774,7 @@ static int32_t mnodeProcessCreateDbMsg(SMnodeMsg *pMsg) {
   } else if (!pMsg->pUser->writeAuth) {
     code = TSDB_CODE_MND_NO_RIGHTS;
   } else {
-    code = mnodeCreateDb(pMsg->pUser->pAcct, pCreate);
-    if (code == TSDB_CODE_SUCCESS) {
-      mLPrint("db:%s, is created by %s", pCreate->db, pMsg->pUser->user);
-    } else {
-      mError("db:%s, failed to create, reason:%s", pCreate->db, tstrerror(code));
-    }
+    code = mnodeCreateDb(pMsg->pUser->pAcct, pCreate, pMsg);
   }
 
   return code;
@@ -893,7 +891,31 @@ static SDbCfg mnodeGetAlterDbOption(SDbObj *pDb, SCMAlterDbMsg *pAlter) {
   return newCfg;
 }
 
-static int32_t mnodeAlterDb(SDbObj *pDb, SCMAlterDbMsg *pAlter) {
+static int32_t mnodeAlterDbCb(SMnodeMsg *pMsg, int32_t code) {
+  if (code != TSDB_CODE_SUCCESS) return code;
+  SDbObj *pDb = pMsg->pDb;
+
+  void *pIter = NULL;
+  while (1) {
+    SVgObj *pVgroup = NULL;
+    pIter = mnodeGetNextVgroup(pIter, &pVgroup);
+    if (pVgroup == NULL) break;
+    if (pVgroup->pDb == pDb) {
+      mnodeSendCreateVgroupMsg(pVgroup, NULL);
+    }
+    mnodeDecVgroupRef(pVgroup);
+  }
+  sdbFreeIter(pIter);
+
+  mTrace("db:%s, all vgroups is altered", pDb->name);
+  mLPrint("db:%s, is alterd by %s", pDb->name, mnodeGetUserFromMsg(pMsg));
+
+  balanceNotify();
+
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t mnodeAlterDb(SDbObj *pDb, SCMAlterDbMsg *pAlter, void *pMsg) {
   SDbCfg newCfg = mnodeGetAlterDbOption(pDb, pAlter);
   if (terrno != TSDB_CODE_SUCCESS) {
     return terrno;
@@ -904,38 +926,24 @@ static int32_t mnodeAlterDb(SDbObj *pDb, SCMAlterDbMsg *pAlter) {
     return code;
   }
 
-  int32_t oldReplica = pDb->cfg.replications;
-
   if (memcmp(&newCfg, &pDb->cfg, sizeof(SDbCfg)) != 0) {
     pDb->cfg = newCfg;
     pDb->cfgVersion++;
     SSdbOper oper = {
       .type  = SDB_OPER_GLOBAL,
       .table = tsDbSdb,
-      .pObj = pDb
+      .pObj  = pDb,
+      .pMsg  = pMsg,
+      .cb    = mnodeAlterDbCb
     };
 
-    int32_t code = sdbUpdateRow(&oper);
-    if (code != TSDB_CODE_SUCCESS) {
-      return TSDB_CODE_MND_SDB_ERROR;
-    }
-  }
-
-  void *pIter = NULL;
-  while (1) {
-    SVgObj *pVgroup = NULL;
-    pIter = mnodeGetNextVgroup(pIter, &pVgroup);
-    if (pVgroup == NULL) break;   
-    mnodeSendCreateVgroupMsg(pVgroup, NULL);
-    mnodeDecVgroupRef(pVgroup);
+    code = sdbUpdateRow(&oper);
+    if (code == TSDB_CODE_SUCCESS) {
+      if (pMsg != NULL) code = TSDB_CODE_MND_ACTION_IN_PROGRESS;
     }
-  sdbFreeIter(pIter);
-
-  if (oldReplica != pDb->cfg.replications) {
-    balanceNotify();
   }
 
-  return TSDB_CODE_SUCCESS;
+  return code;
 }
 
 static int32_t mnodeProcessAlterDbMsg(SMnodeMsg *pMsg) {
@@ -948,14 +956,7 @@ static int32_t mnodeProcessAlterDbMsg(SMnodeMsg *pMsg) {
     return TSDB_CODE_MND_INVALID_DB;
   }
 
-  int32_t code = mnodeAlterDb(pMsg->pDb, pAlter);
-  if (code != TSDB_CODE_SUCCESS) {
-    mError("db:%s, failed to alter, invalid db option", pAlter->db);
-    return code;
-  }
-
-  mTrace("db:%s, all vgroups is altered", pMsg->pDb->name);
-  return TSDB_CODE_SUCCESS;
+  return mnodeAlterDb(pMsg->pDb, pAlter, pMsg);
 }
 
 static int32_t mnodeDropDb(SMnodeMsg *pMsg) {
@@ -965,11 +966,14 @@ static int32_t mnodeDropDb(SMnodeMsg *pMsg) {
   SSdbOper oper = {
     .type  = SDB_OPER_GLOBAL,
     .table = tsDbSdb,
-    .pObj = pDb
+    .pObj  = pDb,
+    .pMsg  = pMsg
   };
+
   int32_t code = sdbDeleteRow(&oper);
-  if (code != 0) {
-    code = TSDB_CODE_MND_SDB_ERROR;
+  if (code == TSDB_CODE_SUCCESS) {
+    mLPrint("db:%s, is dropped by %s", pDb->name, mnodeGetUserFromMsg(pMsg));
+    if (pMsg != NULL) code = TSDB_CODE_MND_ACTION_IN_PROGRESS;
   }
 
   return code;

src/mnode/src/mnodeDnode.c

@@ -44,7 +44,7 @@ static int32_t tsDnodeUpdateSize = 0;
 extern void *  tsMnodeSdb;
 extern void *  tsVgroupSdb;
 
-static int32_t mnodeCreateDnode(char *ep);
+static int32_t mnodeCreateDnode(char *ep, SMnodeMsg *pMsg);
 static int32_t mnodeProcessCreateDnodeMsg(SMnodeMsg *pMsg);
 static int32_t mnodeProcessDropDnodeMsg(SMnodeMsg *pMsg);
 static int32_t mnodeProcessCfgDnodeMsg(SMnodeMsg *pMsg);
@@ -117,7 +117,8 @@ static int32_t mnodeDnodeActionDecode(SSdbOper *pOper) {
 static int32_t mnodeDnodeActionRestored() {
   int32_t numOfRows = sdbGetNumOfRows(tsDnodeSdb);
   if (numOfRows <= 0 && dnodeIsFirstDeploy()) {
-    mnodeCreateDnode(tsLocalEp);
+    mPrint("dnode first deploy, create dnode:%s", tsLocalEp);
+    mnodeCreateDnode(tsLocalEp, NULL);
     SDnodeObj *pDnode = mnodeGetDnodeByEp(tsLocalEp);
     mnodeAddMnode(pDnode->dnodeId);
     mnodeDecDnodeRef(pDnode);
@@ -391,7 +392,7 @@ static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mnodeCreateDnode(char *ep) {
+static int32_t mnodeCreateDnode(char *ep, SMnodeMsg *pMsg) {
   int32_t grantCode = grantCheck(TSDB_GRANT_DNODE);
   if (grantCode != TSDB_CODE_SUCCESS) {
     return grantCode;
@@ -415,7 +416,8 @@ static int32_t mnodeCreateDnode(char *ep) {
     .type = SDB_OPER_GLOBAL,
     .table = tsDnodeSdb,
     .pObj = pDnode,
-    .rowSize = sizeof(SDnodeObj)
+    .rowSize = sizeof(SDnodeObj),
+    .pMsg = pMsg
   };
 
   int32_t code = sdbInsertRow(&oper);
@@ -423,30 +425,32 @@ static int32_t mnodeCreateDnode(char *ep) {
     int dnodeId = pDnode->dnodeId;
     tfree(pDnode);
     mError("failed to create dnode:%d, result:%s", dnodeId, tstrerror(code));
-    return TSDB_CODE_MND_SDB_ERROR;
+  } else {
+    mPrint("dnode:%d is created, result:%s", pDnode->dnodeId, tstrerror(code));
+    if (pMsg != NULL) code = TSDB_CODE_MND_ACTION_IN_PROGRESS;
   }
 
-  mPrint("dnode:%d is created, result:%s", pDnode->dnodeId, tstrerror(code));
   return code;
 }
 
-int32_t mnodeDropDnode(SDnodeObj *pDnode) {
+int32_t mnodeDropDnode(SDnodeObj *pDnode, void *pMsg) {
   SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsDnodeSdb,
-    .pObj = pDnode
+    .pObj = pDnode,
+    .pMsg = pMsg
   };
 
   int32_t code = sdbDeleteRow(&oper);
-  if (code != TSDB_CODE_SUCCESS) {
-    code = TSDB_CODE_MND_SDB_ERROR;
+  if (code == TSDB_CODE_SUCCESS) {
+    mLPrint("dnode:%d, is dropped from cluster, result:%s", pDnode->dnodeId, tstrerror(code));
+    if (pMsg != NULL) code = TSDB_CODE_MND_ACTION_IN_PROGRESS;
   }
 
-  mLPrint("dnode:%d, is dropped from cluster, result:%s", pDnode->dnodeId, tstrerror(code));
   return code;
 }
 
-static int32_t mnodeDropDnodeByEp(char *ep) {
+static int32_t mnodeDropDnodeByEp(char *ep, SMnodeMsg *pMsg) {
   SDnodeObj *pDnode = mnodeGetDnodeByEp(ep);
   if (pDnode == NULL) {
     mError("dnode:%s, is not exist", ep);
@@ -461,7 +465,7 @@ static int32_t mnodeDropDnodeByEp(char *ep) {
 
   mPrint("dnode:%d, start to drop it", pDnode->dnodeId);
 #ifndef _SYNC
-  return mnodeDropDnode(pDnode);
+  return mnodeDropDnode(pDnode, pMsg);
 #else
   return balanceDropDnode(pDnode);
 #endif
@@ -473,17 +477,7 @@ static int32_t mnodeProcessCreateDnodeMsg(SMnodeMsg *pMsg) {
   if (strcmp(pMsg->pUser->user, "root") != 0) {
     return TSDB_CODE_MND_NO_RIGHTS;
   } else {
-    int32_t code = mnodeCreateDnode(pCreate->ep);
-
-    if (code == TSDB_CODE_SUCCESS) {
-      SDnodeObj *pDnode = mnodeGetDnodeByEp(pCreate->ep);
-      mLPrint("dnode:%d, %s is created by %s", pDnode->dnodeId, pCreate->ep, pMsg->pUser->user);
-      mnodeDecDnodeRef(pDnode);
-    } else {
-      mError("failed to create dnode:%s, reason:%s", pCreate->ep, tstrerror(code));
-    }
-
-    return code;
+    return mnodeCreateDnode(pCreate->ep, pMsg);
   }
 }
 
@@ -493,15 +487,7 @@ static int32_t mnodeProcessDropDnodeMsg(SMnodeMsg *pMsg) {
   if (strcmp(pMsg->pUser->user, "root") != 0) {
     return TSDB_CODE_MND_NO_RIGHTS;
   } else {
-    int32_t code = mnodeDropDnodeByEp(pDrop->ep);
-
-    if (code == TSDB_CODE_SUCCESS) {
-      mLPrint("dnode:%s is dropped by %s", pDrop->ep, pMsg->pUser->user);
-    } else {
-      mError("failed to drop dnode:%s, reason:%s", pDrop->ep, tstrerror(code));
-    }
-
-    return code;
+    return mnodeDropDnodeByEp(pDrop->ep, pMsg);
   }
 }
 

src/mnode/src/mnodeSdb.c

@@ -24,6 +24,7 @@
 #include "tsync.h"
 #include "tglobal.h"
 #include "dnode.h"
+#include "mnode.h"
 #include "mnodeDef.h"
 #include "mnodeInt.h"
 #include "mnodeMnode.h"
@@ -31,6 +32,7 @@
 #include "mnodeSdb.h"
 
 #define SDB_TABLE_LEN 12
+#define SDB_SYNC_HACK 16
 
 typedef enum {
   SDB_ACTION_INSERT,
@@ -83,8 +85,29 @@ typedef struct {
   void *  row;
 } SSdbRow;
 
+typedef struct {
+  pthread_t thread;
+  int32_t   workerId;
+} SSdbWriteWorker;
+
+typedef struct {
+  int32_t num;
+  SSdbWriteWorker *writeWorker;
+} SSdbWriteWorkerPool;
+
 static SSdbObject tsSdbObj = {0};
+static taos_qset  tsSdbWriteQset;
+static taos_qall  tsSdbWriteQall;
+static taos_queue tsSdbWriteQueue;
+static SSdbWriteWorkerPool tsSdbPool;
+
 static int     sdbWrite(void *param, void *data, int type);
+static int     sdbWriteToQueue(void *param, void *data, int type);
+static void *  sdbWorkerFp(void *param);
+static int32_t sdbInitWriteWorker();
+static void    sdbCleanupWriteWorker();
+static int32_t sdbAllocWriteQueue();
+static void    sdbFreeWritequeue();
 
 int32_t sdbGetId(void *handle) {
   return ((SSdbTable *)handle)->autoIndex;
@@ -302,7 +325,7 @@ void sdbUpdateSync() {
   syncInfo.ahandle = NULL;
   syncInfo.getWalInfo = sdbGetWalInfo;
   syncInfo.getFileInfo = sdbGetFileInfo;
-  syncInfo.writeToCache = sdbWrite;
+  syncInfo.writeToCache = sdbWriteToQueue;
   syncInfo.confirmForward = sdbConfirmForward; 
   syncInfo.notifyRole = sdbNotifyRole;
   tsSdbObj.cfg = syncCfg;
@@ -319,6 +342,10 @@ int32_t sdbInit() {
   pthread_mutex_init(&tsSdbObj.mutex, NULL);
   sem_init(&tsSdbObj.sem, 0, 0);
 
+  if (sdbInitWriteWorker() != 0) {
+    return -1;
+  }
+
   if (sdbInitWal() != 0) {
     return -1;
   }
@@ -340,6 +367,8 @@ void sdbCleanUp() {
 
   tsSdbObj.status = SDB_STATUS_CLOSING;
   
+  sdbCleanupWriteWorker();
+
   if (tsSdbObj.sync) {
     syncStop(tsSdbObj.sync);
     tsSdbObj.sync = NULL;
@@ -475,7 +504,7 @@ static int32_t sdbDeleteHash(SSdbTable *pTable, SSdbOper *pOper) {
   pTable->numOfRows--;
   pthread_mutex_unlock(&pTable->mutex);
 
-  sdbTrace("table:%s, delete record:%s from hash, numOfRows:%" PRId64 "version:%" PRIu64, pTable->tableName,
+  sdbTrace("table:%s, delete record:%s from hash, numOfRows:%" PRId64 " version:%" PRIu64, pTable->tableName,
            sdbGetKeyStrFromObj(pTable, pOper->pObj), pTable->numOfRows, sdbGetVersion());
 
   int8_t *updateEnd = pOper->pObj + pTable->refCountPos - 1;
@@ -494,6 +523,7 @@ static int32_t sdbUpdateHash(SSdbTable *pTable, SSdbOper *pOper) {
 }
 
 static int sdbWrite(void *param, void *data, int type) {
+  SSdbOper *pOper = param;
   SWalHead *pHead = data;
   int32_t tableId = pHead->msgType / 10;
   int32_t action = pHead->msgType % 10;
@@ -531,21 +561,22 @@ static int sdbWrite(void *param, void *data, int type) {
     pthread_mutex_unlock(&tsSdbObj.mutex);
     return code;
   }
-  walFsync(tsSdbObj.wal);
   
   code = sdbForwardToPeer(pHead);
   pthread_mutex_unlock(&tsSdbObj.mutex);
 
   // from app, oper is created
-  if (param != NULL) {
-    //sdbTrace("request from app is disposed, version:%" PRIu64 " code:%s", pHead->version, tstrerror(code));
+  if (pOper != NULL) {
+    sdbTrace("record from app is disposed, version:%" PRIu64 " result:%s", pHead->version, tstrerror(code));
     return code;
   }
   
   // from wal or forward msg, oper not created, should add into hash
   if (tsSdbObj.sync != NULL) {
-    sdbTrace("forward request is received, version:%" PRIu64 " result:%s, confirm it", pHead->version, tstrerror(code));
+    sdbTrace("record from wal forward is disposed, version:%" PRIu64 " confirm it", pHead->version);
     syncConfirmForward(tsSdbObj.sync, pHead->version, code);
+  } else {
+    sdbTrace("record from wal restore is disposed, version:%" PRIu64 , pHead->version);
   }
 
   if (action == SDB_ACTION_INSERT) {
@@ -568,7 +599,7 @@ static int sdbWrite(void *param, void *data, int type) {
 
 int32_t sdbInsertRow(SSdbOper *pOper) {
   SSdbTable *pTable = (SSdbTable *)pOper->table;
-  if (pTable == NULL) return -1;
+  if (pTable == NULL) return TSDB_CODE_MND_SDB_INVALID_TABLE_TYPE;
 
   if (sdbGetRowFromObj(pTable, pOper->pObj)) {
     sdbError("table:%s, failed to insert record:%s, already exist", pTable->tableName, sdbGetKeyStrFromObj(pTable, pOper->pObj));
@@ -587,9 +618,21 @@ int32_t sdbInsertRow(SSdbOper *pOper) {
     pthread_mutex_unlock(&pTable->mutex);
   }
 
-  if (pOper->type == SDB_OPER_GLOBAL) {
-    int32_t   size = sizeof(SWalHead) + pTable->maxRowSize;
-    SWalHead *pHead = taosAllocateQitem(size);
+  int32_t code = sdbInsertHash(pTable, pOper);
+  if (code != TSDB_CODE_SUCCESS) {
+    sdbError("table:%s, failed to insert into hash", pTable->tableName);
+    return code;
+  }
+
+  // just insert data into memory
+  if (pOper->type != SDB_OPER_GLOBAL) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  int32_t size = sizeof(SSdbOper) + sizeof(SWalHead) + pTable->maxRowSize + SDB_SYNC_HACK;
+  SSdbOper *pNewOper = taosAllocateQitem(size);
+  
+  SWalHead *pHead = (void *)pNewOper + sizeof(SSdbOper) + SDB_SYNC_HACK;
   pHead->version = 0;
   pHead->len = pOper->rowSize;
   pHead->msgType = pTable->tableId * 10 + SDB_ACTION_INSERT;
@@ -598,28 +641,43 @@ int32_t sdbInsertRow(SSdbOper *pOper) {
   (*pTable->encodeFp)(pOper);
   pHead->len = pOper->rowSize;
 
-    int32_t code = sdbWrite(pOper, pHead, pHead->msgType);
-    taosFreeQitem(pHead);
-    if (code < 0) return code;
+  memcpy(pNewOper, pOper, sizeof(SSdbOper));
+
+  if (pNewOper->pMsg != NULL) {
+    sdbTrace("app:%p:%p, insert action is add to sdb queue", pNewOper->pMsg->rpcMsg.ahandle, pNewOper->pMsg);
   }
 
-  return sdbInsertHash(pTable, pOper);
+  taosWriteQitem(tsSdbWriteQueue, TAOS_QTYPE_RPC, pNewOper);
+  return TSDB_CODE_SUCCESS;
 }
 
 int32_t sdbDeleteRow(SSdbOper *pOper) {
   SSdbTable *pTable = (SSdbTable *)pOper->table;
-  if (pTable == NULL) return -1;
+  if (pTable == NULL) return TSDB_CODE_MND_SDB_INVALID_TABLE_TYPE;
 
   SSdbRow *pMeta = sdbGetRowMetaFromObj(pTable, pOper->pObj);
   if (pMeta == NULL) {
     sdbTrace("table:%s, record is not there, delete failed", pTable->tableName);
-    return -1;
+    return TSDB_CODE_MND_SDB_OBJ_NOT_THERE;
   }
 
-  void * pMetaRow = pMeta->row;
-  assert(pMetaRow != NULL);
+  void *pMetaRow = pMeta->row;
+  if (pMetaRow == NULL) {
+    sdbError("table:%s, record meta is null", pTable->tableName);
+    return TSDB_CODE_MND_SDB_INVAID_META_ROW;
+  }
+
+  int32_t code = sdbDeleteHash(pTable, pOper);
+  if (code != TSDB_CODE_SUCCESS) {
+    sdbError("table:%s, failed to delete from hash", pTable->tableName);
+    return code;
+  }
+
+  // just delete data from memory
+  if (pOper->type != SDB_OPER_GLOBAL) {
+    return TSDB_CODE_SUCCESS;
+  }
 
-  if (pOper->type == SDB_OPER_GLOBAL) {
   void *  key = sdbGetObjKey(pTable, pOper->pObj);
   int32_t keySize = 0;
   switch (pTable->keyType) {
@@ -632,40 +690,59 @@ int32_t sdbDeleteRow(SSdbOper *pOper) {
       keySize = sizeof(uint32_t);
       break;
     default:
-        return -1;
+      return TSDB_CODE_MND_SDB_INVAID_KEY_TYPE;
   }
 
-    int32_t   size = sizeof(SWalHead) + keySize;
-    SWalHead *pHead = taosAllocateQitem(size);
+  int32_t size = sizeof(SSdbOper) + sizeof(SWalHead) + keySize + SDB_SYNC_HACK;
+  SSdbOper *pNewOper = taosAllocateQitem(size);
+
+  SWalHead *pHead = (void *)pNewOper + sizeof(SSdbOper) + SDB_SYNC_HACK;
   pHead->version = 0;
   pHead->len = keySize;
   pHead->msgType = pTable->tableId * 10 + SDB_ACTION_DELETE;
   memcpy(pHead->cont, key, keySize);
 
-    int32_t code = sdbWrite(pOper, pHead, pHead->msgType);
-    taosFreeQitem(pHead);
-    if (code < 0) return code;
+  memcpy(pNewOper, pOper, sizeof(SSdbOper));
+
+  if (pNewOper->pMsg != NULL) {
+    sdbTrace("app:%p:%p, delete action is add to sdb queue", pNewOper->pMsg->rpcMsg.ahandle, pNewOper->pMsg);
   }
 
-  return sdbDeleteHash(pTable, pOper);
+  taosWriteQitem(tsSdbWriteQueue, TAOS_QTYPE_RPC, pNewOper);
+  return TSDB_CODE_SUCCESS;
 }
 
 int32_t sdbUpdateRow(SSdbOper *pOper) {
   SSdbTable *pTable = (SSdbTable *)pOper->table;
-  if (pTable == NULL) return -1;
+  if (pTable == NULL) return TSDB_CODE_MND_SDB_INVALID_TABLE_TYPE;
 
   SSdbRow *pMeta = sdbGetRowMetaFromObj(pTable, pOper->pObj);
   if (pMeta == NULL) {
-    sdbTrace("table:%s, record is not there, delete failed", pTable->tableName);
-    return -1;
+    sdbTrace("table:%s, record is not there, update failed", pTable->tableName);
+    return TSDB_CODE_MND_SDB_OBJ_NOT_THERE;
+  }
+
+  void *pMetaRow = pMeta->row;
+  if (pMetaRow == NULL) {
+    sdbError("table:%s, record meta is null", pTable->tableName);
+    return TSDB_CODE_MND_SDB_INVAID_META_ROW;
+  }
+
+  int32_t code = sdbUpdateHash(pTable, pOper);
+  if (code != TSDB_CODE_SUCCESS) {
+    sdbError("table:%s, failed to update hash", pTable->tableName);
+    return code;
+  }
+
+  // just update data in memory
+  if (pOper->type != SDB_OPER_GLOBAL) {
+    return TSDB_CODE_SUCCESS;
   }
 
-  void * pMetaRow = pMeta->row;
-  assert(pMetaRow != NULL);
+  int32_t size = sizeof(SSdbOper) + sizeof(SWalHead) + pTable->maxRowSize + SDB_SYNC_HACK;
+  SSdbOper *pNewOper = taosAllocateQitem(size);
 
-  if (pOper->type == SDB_OPER_GLOBAL) {
-    int32_t   size = sizeof(SWalHead) + pTable->maxRowSize;
-    SWalHead *pHead = taosAllocateQitem(size);
+  SWalHead *pHead = (void *)pNewOper + sizeof(SSdbOper) + SDB_SYNC_HACK;
   pHead->version = 0;
   pHead->msgType = pTable->tableId * 10 + SDB_ACTION_UPDATE;
 
@@ -673,12 +750,14 @@ int32_t sdbUpdateRow(SSdbOper *pOper) {
   (*pTable->encodeFp)(pOper);
   pHead->len = pOper->rowSize;
 
-    int32_t code = sdbWrite(pOper, pHead, pHead->msgType);
-    taosFreeQitem(pHead);
-    if (code < 0) return code;
+  memcpy(pNewOper, pOper, sizeof(SSdbOper));
+
+  if (pNewOper->pMsg != NULL) {
+    sdbTrace("app:%p:%p, update action is add to sdb queue", pNewOper->pMsg->rpcMsg.ahandle, pNewOper->pMsg);
   }
 
-  return sdbUpdateHash(pTable, pOper);
+  taosWriteQitem(tsSdbWriteQueue, TAOS_QTYPE_RPC, pNewOper);
+  return TSDB_CODE_SUCCESS;
 }
 
 void *sdbFetchRow(void *handle, void *pNode, void **ppRow) {
@@ -775,3 +854,158 @@ void sdbCloseTable(void *handle) {
   free(pTable);
 }
 
+int32_t sdbInitWriteWorker() {
+  tsSdbPool.num = 1;
+  tsSdbPool.writeWorker = (SSdbWriteWorker *)calloc(sizeof(SSdbWriteWorker), tsSdbPool.num);
+
+  if (tsSdbPool.writeWorker == NULL) return -1;
+  for (int32_t i = 0; i < tsSdbPool.num; ++i) {
+    SSdbWriteWorker *pWorker = tsSdbPool.writeWorker + i;
+    pWorker->workerId = i;
+  }
+
+  sdbAllocWriteQueue();
+  
+  mPrint("sdb write is opened");
+  return 0;
+}
+
+void sdbCleanupWriteWorker() {
+  for (int32_t i = 0; i < tsSdbPool.num; ++i) {
+    SSdbWriteWorker *pWorker = tsSdbPool.writeWorker + i;
+    if (pWorker->thread) {
+      taosQsetThreadResume(tsSdbWriteQset);
+    }
+  }
+
+  for (int32_t i = 0; i < tsSdbPool.num; ++i) {
+    SSdbWriteWorker *pWorker = tsSdbPool.writeWorker + i;
+    if (pWorker->thread) {
+      pthread_join(pWorker->thread, NULL);
+    }
+  }
+
+  sdbFreeWritequeue();
+
+  mPrint("sdb write is closed");
+}
+
+int32_t sdbAllocWriteQueue() {
+  tsSdbWriteQueue = taosOpenQueue();
+  if (tsSdbWriteQueue == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY;
+
+  tsSdbWriteQset = taosOpenQset();
+  if (tsSdbWriteQset == NULL) {
+    taosCloseQueue(tsSdbWriteQueue);
+    return TSDB_CODE_MND_OUT_OF_MEMORY;
+  }
+  taosAddIntoQset(tsSdbWriteQset, tsSdbWriteQueue, NULL);
+
+  tsSdbWriteQall = taosAllocateQall();
+  if (tsSdbWriteQall == NULL) {
+    taosCloseQset(tsSdbWriteQset);
+    taosCloseQueue(tsSdbWriteQueue);
+    return TSDB_CODE_MND_OUT_OF_MEMORY;
+  }
+  
+  for (int32_t i = 0; i < tsSdbPool.num; ++i) {
+    SSdbWriteWorker *pWorker = tsSdbPool.writeWorker + i;
+    pWorker->workerId = i;
+
+    pthread_attr_t thAttr;
+    pthread_attr_init(&thAttr);
+    pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);
+
+    if (pthread_create(&pWorker->thread, &thAttr, sdbWorkerFp, pWorker) != 0) {
+      mError("failed to create thread to process sdb write queue, reason:%s", strerror(errno));
+      taosFreeQall(tsSdbWriteQall);
+      taosCloseQset(tsSdbWriteQset);
+      taosCloseQueue(tsSdbWriteQueue);
+      return TSDB_CODE_MND_OUT_OF_MEMORY;
+    }
+
+    pthread_attr_destroy(&thAttr);
+    mTrace("sdb write worker:%d is launched, total:%d", pWorker->workerId, tsSdbPool.num);
+  }
+
+  mTrace("sdb write queue:%p is allocated", tsSdbWriteQueue);
+  return TSDB_CODE_SUCCESS;
+}
+
+void sdbFreeWritequeue() {
+  taosCloseQset(tsSdbWriteQueue);
+  taosFreeQall(tsSdbWriteQall);
+  taosCloseQset(tsSdbWriteQset);
+  tsSdbWriteQall = NULL;
+  tsSdbWriteQset = NULL;
+  tsSdbWriteQueue = NULL;
+}
+
+int sdbWriteToQueue(void *param, void *data, int type) {
+  SWalHead *pHead = data;
+  int size = sizeof(SWalHead) + pHead->len;
+  SWalHead *pWal = (SWalHead *)taosAllocateQitem(size);
+  memcpy(pWal, pHead, size);
+
+  taosWriteQitem(tsSdbWriteQueue, type, pWal);
+  return 0;
+}
+
+static void *sdbWorkerFp(void *param) {
+  SWalHead *pHead;
+  SSdbOper *pOper;
+  int32_t   type;
+  int32_t   numOfMsgs;
+  void *    item;
+  void *    unUsed;
+
+  while (1) {
+    numOfMsgs = taosReadAllQitemsFromQset(tsSdbWriteQset, tsSdbWriteQall, &unUsed);
+    if (numOfMsgs == 0) {
+      sdbTrace("sdbWorkerFp: got no message from qset, exiting...");
+      break;
+    }
+
+    for (int32_t i = 0; i < numOfMsgs; ++i) {
+      taosGetQitem(tsSdbWriteQall, &type, &item);
+      if (type == TAOS_QTYPE_RPC) {
+        pOper = (SSdbOper *)item;
+        pHead = (void *)pOper + sizeof(SSdbOper) + SDB_SYNC_HACK;
+      } else {
+        pHead = (SWalHead *)item;
+        pOper = NULL;
+      }
+
+      if (pOper != NULL && pOper->pMsg != NULL) {
+        sdbTrace("app:%p:%p, will be processed in sdb queue", pOper->pMsg->rpcMsg.ahandle, pOper->pMsg);
+      }
+
+      int32_t code = sdbWrite(pOper, pHead, type);
+      if (pOper) pOper->retCode = code;
+    }
+
+    walFsync(tsSdbObj.wal);
+
+    // browse all items, and process them one by one
+    taosResetQitems(tsSdbWriteQall);
+    for (int32_t i = 0; i < numOfMsgs; ++i) {
+      taosGetQitem(tsSdbWriteQall, &type, &item);
+      if (type == TAOS_QTYPE_RPC) {
+        pOper = (SSdbOper *)item;
+        if (pOper != NULL && pOper->cb != NULL) {
+          pOper->retCode = (*pOper->cb)(pOper->pMsg, pOper->retCode);
+        }
+          
+        if (pOper != NULL && pOper->pMsg != NULL) {
+          sdbTrace("app:%p:%p, msg is processed, result:%s", pOper->pMsg->rpcMsg.ahandle, pOper->pMsg,
+                   tstrerror(pOper->retCode));
+        }
+
+        dnodeSendRpcMnodeWriteRsp(pOper->pMsg, pOper->retCode);
+      }
+      taosFreeQitem(item);
+    }
+  }
+
+  return NULL;
+}

src/mnode/src/mnodeUser.c

@@ -102,11 +102,13 @@ static int32_t mnodeUserActionDecode(SSdbOper *pOper) {
 }
 
 static int32_t mnodeUserActionRestored() {
-  if (dnodeIsFirstDeploy()) {
+  int32_t numOfRows = sdbGetNumOfRows(tsUserSdb);
+  if (numOfRows <= 0 && dnodeIsFirstDeploy()) {
+    mPrint("dnode first deploy, create root user");
     SAcctObj *pAcct = mnodeGetAcct("root");
-    mnodeCreateUser(pAcct, "root", "taosdata");
-    mnodeCreateUser(pAcct, "monitor", tsInternalPass);
-    mnodeCreateUser(pAcct, "_root", tsInternalPass);
+    mnodeCreateUser(pAcct, "root", "taosdata", NULL);
+    mnodeCreateUser(pAcct, "monitor", tsInternalPass, NULL);
+    mnodeCreateUser(pAcct, "_root", tsInternalPass, NULL);
     mnodeDecAcctRef(pAcct);
   }
 
@@ -170,22 +172,24 @@ void mnodeDecUserRef(SUserObj *pUser) {
   return sdbDecRef(tsUserSdb, pUser); 
 }
 
-static int32_t mnodeUpdateUser(SUserObj *pUser) {
+static int32_t mnodeUpdateUser(SUserObj *pUser, void *pMsg) {
   SSdbOper oper = {
     .type  = SDB_OPER_GLOBAL,
     .table = tsUserSdb,
-    .pObj = pUser
+    .pObj  = pUser,
+    .pMsg  = pMsg
   };
 
   int32_t code = sdbUpdateRow(&oper);
-  if (code != TSDB_CODE_SUCCESS) {
-    code = TSDB_CODE_MND_SDB_ERROR;
+  if (code == TSDB_CODE_SUCCESS) {
+    mLPrint("user:%s, is altered by %s", pUser->user, mnodeGetUserFromMsg(pMsg));
+    if (pMsg != NULL) code = TSDB_CODE_MND_ACTION_IN_PROGRESS;
   }
 
   return code;
 }
 
-int32_t mnodeCreateUser(SAcctObj *pAcct, char *name, char *pass) {
+int32_t mnodeCreateUser(SAcctObj *pAcct, char *name, char *pass, void *pMsg) {
   int32_t code = acctCheck(pAcct, ACCT_GRANT_USER);
   if (code != TSDB_CODE_SUCCESS) {
     return code;
@@ -226,28 +230,33 @@ int32_t mnodeCreateUser(SAcctObj *pAcct, char *name, char *pass) {
     .type    = SDB_OPER_GLOBAL,
     .table   = tsUserSdb,
     .pObj    = pUser,
-    .rowSize = sizeof(SUserObj)
+    .rowSize = sizeof(SUserObj),
+    .pMsg    = pMsg
   };
 
   code = sdbInsertRow(&oper);
   if (code != TSDB_CODE_SUCCESS) {
     tfree(pUser);
-    code = TSDB_CODE_MND_SDB_ERROR;
+  } else {
+    mLPrint("user:%s, is created by %s", pUser->user, mnodeGetUserFromMsg(pMsg));
+    if (pMsg != NULL) code = TSDB_CODE_MND_ACTION_IN_PROGRESS;
   }
 
   return code;
 }
 
-static int32_t mnodeDropUser(SUserObj *pUser) {
+static int32_t mnodeDropUser(SUserObj *pUser, void *pMsg) {
   SSdbOper oper = {
     .type  = SDB_OPER_GLOBAL,
     .table = tsUserSdb,
-    .pObj = pUser
+    .pObj  = pUser,
+    .pMsg  = pMsg
   };
 
   int32_t code = sdbDeleteRow(&oper);
-  if (code != TSDB_CODE_SUCCESS) {
-    code = TSDB_CODE_MND_SDB_ERROR;
+  if (code == TSDB_CODE_SUCCESS) {
+    mLPrint("user:%s, is dropped by %s", pUser->user, mnodeGetUserFromMsg(pMsg));
+    if (pMsg != NULL) code = TSDB_CODE_MND_ACTION_IN_PROGRESS;
   }
 
   return code;
@@ -357,22 +366,25 @@ SUserObj *mnodeGetUserFromConn(void *pConn) {
   }
 }
 
+char *mnodeGetUserFromMsg(void *pMsg) {
+  SMnodeMsg *pMnodeMsg = pMsg;
+  if (pMnodeMsg != NULL &&pMnodeMsg->pUser != NULL) {
+    return pMnodeMsg->pUser->user;
+  } else {
+    return "system";
+  }
+}
+
 static int32_t mnodeProcessCreateUserMsg(SMnodeMsg *pMsg) {
-  int32_t code;
   SUserObj *pOperUser = pMsg->pUser;
   
   if (pOperUser->superAuth) {
     SCMCreateUserMsg *pCreate = pMsg->rpcMsg.pCont;
-    code = mnodeCreateUser(pOperUser->pAcct, pCreate->user, pCreate->pass);
-    if (code == TSDB_CODE_SUCCESS) {
-      mLPrint("user:%s, is created by %s", pCreate->user, pOperUser->user);
-    }
+    return mnodeCreateUser(pOperUser->pAcct, pCreate->user, pCreate->pass, pMsg);
   } else {
     mError("user:%s, no rights to create user", pOperUser->user);
-    code = TSDB_CODE_MND_NO_RIGHTS;
+    return TSDB_CODE_MND_NO_RIGHTS;
   }
-
-  return code;
 }
 
 static int32_t mnodeProcessAlterUserMsg(SMnodeMsg *pMsg) {
@@ -409,8 +421,7 @@ static int32_t mnodeProcessAlterUserMsg(SMnodeMsg *pMsg) {
     if (hasRight) {
       memset(pUser->pass, 0, sizeof(pUser->pass));
       taosEncryptPass((uint8_t*)pAlter->pass, strlen(pAlter->pass), pUser->pass);
-      code = mnodeUpdateUser(pUser);
-      mLPrint("user:%s, password is altered by %s, result:%s", pUser->user, pOperUser->user, tstrerror(code));
+      code = mnodeUpdateUser(pUser, pMsg);
     } else {
       mError("user:%s, no rights to alter user", pOperUser->user);
       code = TSDB_CODE_MND_NO_RIGHTS;
@@ -450,8 +461,7 @@ static int32_t mnodeProcessAlterUserMsg(SMnodeMsg *pMsg) {
         pUser->writeAuth = 1;
       }
 
-      code = mnodeUpdateUser(pUser);
-      mLPrint("user:%s, privilege is altered by %s, result:%s", pUser->user, pOperUser->user, tstrerror(code));
+      code = mnodeUpdateUser(pUser, pMsg);
     } else {
       mError("user:%s, no rights to alter user", pOperUser->user);
       code = TSDB_CODE_MND_NO_RIGHTS;
@@ -497,10 +507,7 @@ static int32_t mnodeProcessDropUserMsg(SMnodeMsg *pMsg) {
   }
 
   if (hasRight) {
-    code = mnodeDropUser(pUser);
-    if (code == TSDB_CODE_SUCCESS) {
-      mLPrint("user:%s, is dropped by %s, result:%s", pUser->user, pOperUser->user, tstrerror(code));
-    }
+    code = mnodeDropUser(pUser, pMsg);
   } else {
     code = TSDB_CODE_MND_NO_RIGHTS;
   }

src/mnode/src/mnodeVgroup.c

@@ -149,7 +149,7 @@ static int32_t mnodeVgroupActionUpdate(SSdbOper *pOper) {
       }
     }
 
-    memcpy(pVgroup, pNew, pOper->rowSize);
+    memcpy(pVgroup, pNew, sizeof(SVgObj));
     free(pNew);
 
     for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
@@ -299,6 +299,27 @@ void *mnodeGetNextVgroup(void *pIter, SVgObj **pVgroup) {
   return sdbFetchRow(tsVgroupSdb, pIter, (void **)pVgroup); 
 }
 
+static int32_t mnodeCreateVgroupCb(SMnodeMsg *pMsg, int32_t code) {
+  if (code != TSDB_CODE_SUCCESS) {
+    pMsg->pVgroup = NULL;
+    return code;
+  }
+
+  SVgObj *pVgroup = pMsg->pVgroup;
+  SDbObj *pDb = pMsg->pDb;
+
+  mPrint("vgId:%d, is created in mnode, db:%s replica:%d", pVgroup->vgId, pDb->name, pVgroup->numOfVnodes);
+  for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
+    mPrint("vgId:%d, index:%d, dnode:%d", pVgroup->vgId, i, pVgroup->vnodeGid[i].dnodeId);
+  }
+
+  mnodeIncVgroupRef(pVgroup);
+  pMsg->expected = pVgroup->numOfVnodes;
+  mnodeSendCreateVgroupMsg(pVgroup, pMsg);
+
+  return TSDB_CODE_MND_ACTION_IN_PROGRESS;
+}
+
 int32_t mnodeCreateVgroup(SMnodeMsg *pMsg, SDbObj *pDb) {
   SVgObj *pVgroup = (SVgObj *)calloc(1, sizeof(SVgObj));
   strcpy(pVgroup->dbName, pDb->name);
@@ -314,26 +335,22 @@ int32_t mnodeCreateVgroup(SMnodeMsg *pMsg, SDbObj *pDb) {
     .type = SDB_OPER_GLOBAL,
     .table = tsVgroupSdb,
     .pObj = pVgroup,
-    .rowSize = sizeof(SVgObj)
+    .rowSize = sizeof(SVgObj),
+    .pMsg = pMsg,
+    .cb = mnodeCreateVgroupCb
   };
 
+  pMsg->pVgroup = pVgroup;
+
   int32_t code = sdbInsertRow(&oper);
   if (code != TSDB_CODE_SUCCESS) {
+    pMsg->pVgroup = NULL;
     tfree(pVgroup);
-    return TSDB_CODE_MND_SDB_ERROR;
-  }
-
-  mPrint("vgId:%d, is created in mnode, db:%s replica:%d", pVgroup->vgId, pDb->name, pVgroup->numOfVnodes);
-  for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
-    mPrint("vgId:%d, index:%d, dnode:%d", pVgroup->vgId, i, pVgroup->vnodeGid[i].dnodeId);
+  } else {
+    if (pMsg != NULL) code = TSDB_CODE_MND_ACTION_IN_PROGRESS;
   }
 
-  mnodeIncVgroupRef(pVgroup);
-  pMsg->pVgroup = pVgroup;
-  pMsg->expected = pVgroup->numOfVnodes;
-  mnodeSendCreateVgroupMsg(pVgroup, pMsg);
-
-  return TSDB_CODE_MND_ACTION_IN_PROGRESS;
+  return code;
 }
 
 void mnodeDropVgroup(SVgObj *pVgroup, void *ahandle) {
@@ -596,7 +613,6 @@ SRpcIpSet mnodeGetIpSetFromIp(char *ep) {
 }
 
 void mnodeSendCreateVnodeMsg(SVgObj *pVgroup, SRpcIpSet *ipSet, void *ahandle) {
-  mTrace("vgId:%d, send create vnode:%d msg, ahandle:%p db:%s", pVgroup->vgId, pVgroup->vgId, ahandle, pVgroup->dbName);
   SMDCreateVnodeMsg *pCreate = mnodeBuildCreateVnodeMsg(pVgroup);
   SRpcMsg rpcMsg = {
     .handle  = ahandle,
@@ -609,9 +625,12 @@ void mnodeSendCreateVnodeMsg(SVgObj *pVgroup, SRpcIpSet *ipSet, void *ahandle) {
 }
 
 void mnodeSendCreateVgroupMsg(SVgObj *pVgroup, void *ahandle) {
-  mTrace("vgId:%d, send create all vnodes msg, ahandle:%p", pVgroup->vgId, ahandle);
+  mTrace("vgId:%d, send create all vnodes msg, numOfVnodes:%d db:%s", pVgroup->vgId, pVgroup->numOfVnodes,
+         pVgroup->dbName);
   for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
     SRpcIpSet ipSet = mnodeGetIpSetFromIp(pVgroup->vnodeGid[i].pDnode->dnodeEp);
+    mTrace("vgId:%d, index:%d, send create vnode msg to dnode %s, ahandle:%p", pVgroup->vgId,
+           i, pVgroup->vnodeGid[i].pDnode->dnodeEp, ahandle);
     mnodeSendCreateVnodeMsg(pVgroup, &ipSet, ahandle);
   }
 }
@@ -729,6 +748,7 @@ static int32_t mnodeProcessVnodeCfgMsg(SMnodeMsg *pMsg) {
   }
   mnodeDecVgroupRef(pVgroup);
 
+  mTrace("vgId:%d, send create vnode msg to dnode %s for vnode cfg msg", pVgroup->vgId, pDnode->dnodeEp);
   SRpcIpSet ipSet = mnodeGetIpSetFromIp(pDnode->dnodeEp);
   mnodeSendCreateVnodeMsg(pVgroup, &ipSet, NULL);
 

src/mnode/src/mnodeWrite.c

@@ -43,7 +43,7 @@ void mnodeAddWriteMsgHandle(uint8_t msgType, int32_t (*fp)(SMnodeMsg *mnodeMsg))
 
 int32_t mnodeProcessWrite(SMnodeMsg *pMsg) {
   if (pMsg->rpcMsg.pCont == NULL) {
-    mError("%p, msg:%s  in mwrite queue, content is null", pMsg->rpcMsg.ahandle, taosMsg[pMsg->rpcMsg.msgType]);
+    mError("app:%p:%p, msg:%s content is null", pMsg->rpcMsg.ahandle, pMsg, taosMsg[pMsg->rpcMsg.msgType]);
     return TSDB_CODE_MND_INVALID_MSG_LEN;
   }
 
@@ -54,27 +54,31 @@ int32_t mnodeProcessWrite(SMnodeMsg *pMsg) {
     rpcRsp->rsp = ipSet;
     rpcRsp->len = sizeof(SRpcIpSet);
 
-    mTrace("%p, msg:%s in mwrite queue, will be redireced inUse:%d", pMsg->rpcMsg.ahandle, taosMsg[pMsg->rpcMsg.msgType], ipSet->inUse);
+    mTrace("app:%p:%p, msg:%s will be redireced inUse:%d", pMsg->rpcMsg.ahandle, pMsg, taosMsg[pMsg->rpcMsg.msgType],
+           ipSet->inUse);
     for (int32_t i = 0; i < ipSet->numOfIps; ++i) {
-      mTrace("mnode index:%d ip:%s:%d", i, ipSet->fqdn[i], htons(ipSet->port[i]));
+      mTrace("app:%p:%p, mnode index:%d ip:%s:%d", pMsg->rpcMsg.ahandle, pMsg, i, ipSet->fqdn[i],
+             htons(ipSet->port[i]));
     }
 
     return TSDB_CODE_RPC_REDIRECT;
   }
 
   if (tsMnodeProcessWriteMsgFp[pMsg->rpcMsg.msgType] == NULL) {
-    mError("%p, msg:%s in mwrite queue, not processed", pMsg->rpcMsg.ahandle, taosMsg[pMsg->rpcMsg.msgType]);
+    mError("app:%p:%p, msg:%s not processed", pMsg->rpcMsg.ahandle, pMsg, taosMsg[pMsg->rpcMsg.msgType]);
     return TSDB_CODE_MND_MSG_NOT_PROCESSED;
   }
 
   int32_t code = mnodeInitMsg(pMsg);
   if (code != TSDB_CODE_SUCCESS) {
-    mError("%p, msg:%s in mwrite queue, not processed reason:%s", pMsg->rpcMsg.ahandle, taosMsg[pMsg->rpcMsg.msgType], tstrerror(code));
+    mError("app:%p:%p, msg:%s not processed, reason:%s", pMsg->rpcMsg.ahandle, pMsg, taosMsg[pMsg->rpcMsg.msgType],
+           tstrerror(code));
     return code;
   }
 
   if (!pMsg->pUser->writeAuth) {
-    mError("%p, msg:%s  in mwrite queue, not processed, no write auth", pMsg->rpcMsg.ahandle, taosMsg[pMsg->rpcMsg.msgType]);
+    mError("app:%p:%p, msg:%s not processed, no write auth", pMsg->rpcMsg.ahandle, pMsg,
+           taosMsg[pMsg->rpcMsg.msgType]);
     return TSDB_CODE_MND_NO_RIGHTS;
   }
 

src/query/src/qExecutor.c

@@ -3356,7 +3356,7 @@ void setWindowResOutputBufInitCtx(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *
 
 int32_t setAdditionalInfo(SQInfo *pQInfo, STableId* pTableId, STableQueryInfo *pTableQueryInfo) {
   SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
-  assert(pTableQueryInfo->lastKey >= TSKEY_INITIAL_VAL);
+  //assert(pTableQueryInfo->lastKey >= TSKEY_INITIAL_VAL);
 
   setTagVal(pRuntimeEnv, pTableId, pQInfo->tsdb);
 
@@ -4252,7 +4252,7 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
       SArray* group = taosArrayGetP(pQInfo->groupInfo.pGroupList, pQInfo->groupIndex);
 
       qTrace("QInfo:%p last_row query on group:%d, total group:%d, current group:%d", pQInfo, pQInfo->groupIndex,
-             numOfGroups);
+             numOfGroups, group);
 
       STsdbQueryCond cond = {
           .twindow = pQuery->window,
@@ -4286,10 +4286,10 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
       
       setTagVal(pRuntimeEnv, (STableId*) taosArrayGet(s, 0), pQInfo->tsdb);
         
-      taosArrayDestroy(s);
       if (isFirstLastRowQuery(pQuery)) {
         assert(taosArrayGetSize(s) == 1);
       }
+      taosArrayDestroy(s);
       
       // here we simply set the first table as current table
       pQuery->current = ((SGroupItem*) taosArrayGet(group, 0))->info;
@@ -5405,6 +5405,7 @@ static int32_t createFilterInfo(void *pQInfo, SQuery *pQuery) {
 
         if ((lower == TSDB_RELATION_GREATER_EQUAL || lower == TSDB_RELATION_GREATER) &&
             (upper == TSDB_RELATION_LESS_EQUAL || upper == TSDB_RELATION_LESS)) {
+          assert(rangeFilterArray != NULL);
           if (lower == TSDB_RELATION_GREATER_EQUAL) {
             if (upper == TSDB_RELATION_LESS_EQUAL) {
               pSingleColFilter->fp = rangeFilterArray[4];
@@ -5419,11 +5420,12 @@ static int32_t createFilterInfo(void *pQInfo, SQuery *pQuery) {
             }
           }
         } else {  // set callback filter function
+          assert(filterArray != NULL);
           if (lower != TSDB_RELATION_INVALID) {
             pSingleColFilter->fp = filterArray[lower];
 
             if (upper != TSDB_RELATION_INVALID) {
-              qError("pQInfo:%p failed to get filter function, invalid filter condition", pQInfo, type);
+              qError("pQInfo:%p failed to get filter function, invalid filter condition: %d", pQInfo, type);
               return TSDB_CODE_QRY_INVALID_MSG;
             }
           } else {

src/query/src/tvariant.c

@@ -887,7 +887,7 @@ int32_t tVariantTypeSetType(tVariant *pVariant, char type) {
         free(pVariant->pz);
         pVariant->dKey = v;
       } else if (pVariant->nType >= TSDB_DATA_TYPE_BOOL && pVariant->nType <= TSDB_DATA_TYPE_BIGINT) {
-        pVariant->dKey = pVariant->i64Key;
+        pVariant->dKey = (double)(pVariant->i64Key);
       }
       
       pVariant->nType = TSDB_DATA_TYPE_DOUBLE;

src/rpc/src/rpcTcp.c

@@ -117,7 +117,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
     code = pthread_mutex_init(&(pThreadObj->mutex), NULL);
     if (code < 0) {
       tError("%s failed to init TCP process data mutex(%s)", label, strerror(errno));
-      break;;
+      break;
     }
 
     pThreadObj->pollFd = epoll_create(10);  // size does not matter
@@ -367,7 +367,7 @@ static void taosReportBrokenLink(SFdObj *pFdObj) {
     recvInfo.ip = 0;
     recvInfo.port = 0;
     recvInfo.shandle = pThreadObj->shandle;
-    recvInfo.thandle = pFdObj->thandle;;
+    recvInfo.thandle = pFdObj->thandle;
     recvInfo.chandle = NULL;
     recvInfo.connType = RPC_CONN_TCP;
     (*(pThreadObj->processData))(&recvInfo);
@@ -414,7 +414,7 @@ static int taosReadTcpData(SFdObj *pFdObj, SRecvInfo *pInfo) {
   pInfo->ip = pFdObj->ip;
   pInfo->port = pFdObj->port;
   pInfo->shandle = pThreadObj->shandle;
-  pInfo->thandle = pFdObj->thandle;;
+  pInfo->thandle = pFdObj->thandle;
   pInfo->chandle = pFdObj;
   pInfo->connType = RPC_CONN_TCP;
 

src/util/src/tmem.c

@@ -29,7 +29,7 @@ static FILE* fpAllocLog = NULL;
 // memory allocator which fails randomly
 
 extern int32_t taosGetTimestampSec();
-static int32_t startTime = INT32_MAX;;
+static int32_t startTime = INT32_MAX;
 
 static bool random_alloc_fail(size_t size, const char* file, uint32_t line) {
   if (taosGetTimestampSec() < startTime) {

src/vnode/src/vnodeMain.c

@@ -119,7 +119,7 @@ int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg) {
   tsdbCfg.minRowsPerFileBlock = pVnodeCfg->cfg.minRowsPerFileBlock;
   tsdbCfg.maxRowsPerFileBlock = pVnodeCfg->cfg.maxRowsPerFileBlock;
   tsdbCfg.precision           = pVnodeCfg->cfg.precision;
-  tsdbCfg.compression         = pVnodeCfg->cfg.compression;;
+  tsdbCfg.compression         = pVnodeCfg->cfg.compression;
 
   char tsdbDir[TSDB_FILENAME_LEN] = {0};
   sprintf(tsdbDir, "%s/vnode%d/tsdb", tsVnodeDir, pVnodeCfg->cfg.vgId);

tests/script/jenkins/basic.txt

@@ -359,3 +359,4 @@ cd ../../../debug; make
 ./test.sh -f unique/arbitrator/sync_replica3_alterTable_drop.sim
 ./test.sh -f unique/arbitrator/sync_replica3_dropDb.sim
 ./test.sh -f unique/arbitrator/sync_replica3_dropTable.sim
+

tests/script/tmp/prepare.sim

@@ -20,10 +20,10 @@ system sh/cfg.sh -n dnode2 -c mnodeEqualVnodeNum -v 4
 system sh/cfg.sh -n dnode3 -c mnodeEqualVnodeNum -v 4
 system sh/cfg.sh -n dnode4 -c mnodeEqualVnodeNum -v 4
 
-system sh/cfg.sh -n dnode1 -c numOfTotalVnodes -v 4
-system sh/cfg.sh -n dnode2 -c numOfTotalVnodes -v 4
-system sh/cfg.sh -n dnode3 -c numOfTotalVnodes -v 4
-system sh/cfg.sh -n dnode4 -c numOfTotalVnodes -v 4
+system sh/cfg.sh -n dnode1 -c numOfTotalVnodes -v 20
+system sh/cfg.sh -n dnode2 -c numOfTotalVnodes -v 20
+system sh/cfg.sh -n dnode3 -c numOfTotalVnodes -v 20
+system sh/cfg.sh -n dnode4 -c numOfTotalVnodes -v 20
 
 system sh/cfg.sh -n dnode1 -c http -v 1
 system sh/cfg.sh -n dnode2 -c http -v 1

tests/script/unique/arbitrator/dn3_mn1_vnode_corruptFile_offline.sim

@@ -155,22 +155,31 @@ if $dnode2Vtatus != master then
   goto wait_dnode3_vgroup_offline
 endi
 
-#system_content ls ../../../sim/dnode3/data/vnode/vnode2/tsdb/data/ -l | grep "^-" | wc -l | sed 's/^[ \t]*//g'
-#system_content ls ../../../sim/dnode3/data/vnode/vnode2/tsdb/data/ -l | grep "^-" | wc -l | sed 's/[ \t]*$//g'
+#$expectCnt = 3 . : 
+#print expectCnt: [ $expectCnt ]
+#system_content ls ../../../sim/dnode3/data/vnode/vnode2/tsdb/data/ -l | grep "^-" | wc -l | tr '\n' ':'
+#system_content ls ../../../sim/dnode3/data/vnode/vnode2/tsdb/data/ -l | grep "^-" | wc -l | tr '\n' ':'
 #print --2-->dnode3 data files: [ $system_content ]
 
-system_content ls ../../../sim/dnode2/data/vnode/vnode2/tsdb/data/ -l | grep "^-" | wc -l
-print ---->dnode2 data files: [ $system_content ], expect is 0
 
-system_content ls ../../../sim/dnode3/data/vnode/vnode2/tsdb/data/ -l | grep "^-" | wc -l
-print ---->dnode3 data files: [ $system_content ], expect is 3
-#if $system_content != 3 then 
-#  return -1
-#endi
+system_content ls ../../../sim/dnode2/data/vnode/vnode2/tsdb/data/ -l | grep "^-" | wc -l | tr -d '\n'
+print ---->dnode2 data files: $system_content   expect: 0
+if $system_content != 0 then
+  return -1
+endi
+
+system_content ls ../../../sim/dnode3/data/vnode/vnode2/tsdb/data/ -l | grep "^-" | wc -l | tr -d '\n'
+print ---->dnode3 data files: $system_content   expect: 3
+if $system_content != 3 then 
+  return -1
+endi
 
 system echo "haha, nothing......" > ../../../sim/dnode3/data/vnode/vnode2/tsdb/data/f1643.data
+
+print ============== step3-1: insert some news data for let version changed
 sql insert into $tb values ( now + 0a , $x ) ( now + 1a , $x ) ( now + 2a , $x )
-$totalRows = $totalRows + 3
+sql insert into $tb values ( now + 10a , $x ) ( now + 11a , $x ) ( now + 12a , $x )
+$totalRows = $totalRows + 6
 sql select count(*) from $stb
 print data00 $data00
 if $data00 != $totalRows then
@@ -242,18 +251,18 @@ if $data00 != $totalRows then
 	return -1
 endi
 
-system_content ls ../../../sim/dnode2/data/vnode/vnode2/tsdb/data/ -l |grep "^-"|wc -l
-print ----> dnode2 data files: [ $system_content ], expect is 0
-#if $system_content != 0 then 
-#  return -1
-#endi
+system_content ls ../../../sim/dnode2/data/vnode/vnode2/tsdb/data/ -l |grep "^-"|wc -l | tr -d '\n'
+print ----> dnode2 data files: $system_content  expect: 0
+if $system_content != 0 then 
+  return -1
+endi
 
-system_content ls ../../../sim/dnode3/data/vnode/vnode2/tsdb/data/ -l |grep "^-"|wc -l
-print ----> dnode3 data files: [ $system_content ], expect is 0
-#if $system_content != 0 then
-#  print there should be no data file in dnode3 after sync 
-#  return -1
-#endi
+system_content ls ../../../sim/dnode3/data/vnode/vnode2/tsdb/data/ -l |grep "^-"|wc -l | tr -d '\n'
+print ----> dnode3 data files: $system_content   expect: 0
+if $system_content != 0 then
+  print there should be no data file in dnode3 after sync 
+  return -1
+endi
 
 print ============== step5: stop dnode2, and check if dnode3 sync ok
 system sh/exec.sh -n dnode2 -s stop -x SIGINT

tests/script/unique/arbitrator/replica_changeWithArbitrator.sim

@@ -25,9 +25,9 @@ system sh/cfg.sh -n dnode3 -c numOfTotalVnodes -v 8
 system sh/cfg.sh -n dnode4 -c numOfTotalVnodes -v 8
 
 system sh/cfg.sh -n dnode1 -c mnodeEqualVnodeNum -v 4
-#system sh/cfg.sh -n dnode2 -c mnodeEqualVnodeNum -v 4
-#system sh/cfg.sh -n dnode3 -c mnodeEqualVnodeNum -v 4
-#system sh/cfg.sh -n dnode4 -c mnodeEqualVnodeNum -v 4
+system sh/cfg.sh -n dnode2 -c mnodeEqualVnodeNum -v 4
+system sh/cfg.sh -n dnode3 -c mnodeEqualVnodeNum -v 4
+system sh/cfg.sh -n dnode4 -c mnodeEqualVnodeNum -v 4
 
 system sh/cfg.sh -n dnode1 -c alternativeRole -v 0
 system sh/cfg.sh -n dnode2 -c alternativeRole -v 0

tests/script/unique/dnode/balance3.sim

@@ -177,7 +177,6 @@ endi
 
 print ========== step5
 sql create dnode $hostname6
-system sh/deploy.sh -n dnode6 -i 6
 system sh/exec.sh -n dnode6 -s start
 
 $x = 0

tests/test/c/insertPerRow.c

@@ -33,6 +33,7 @@ typedef struct {
   int       threadIndex;
   char      dbName[32];
   char      stableName[64];
+  float     createTableSpeed;
   pthread_t thread;
 } SInfo;
 
@@ -49,8 +50,8 @@ int64_t numOfThreads = 1;
 int64_t numOfTablesPerThread = 200;
 char    dbName[32] = "db";
 char    stableName[64] = "st";
-int32_t cache = 16384;
-int32_t tables = 1000;
+int32_t cache = 16;
+int32_t tables = 5000;
 
 int main(int argc, char *argv[]) {
   shellParseArgument(argc, argv);
@@ -63,9 +64,8 @@ int main(int argc, char *argv[]) {
 void createDbAndTable() {
   pPrint("start to create table");
 
+  TAOS_RES *     pSql;
   TAOS *         con;
-  struct timeval systemTime;
-  int64_t        st, et;
   char           qstr[64000];
 
   char fqdn[TSDB_FQDN_LEN];
@@ -77,21 +77,23 @@ void createDbAndTable() {
     exit(1);
   }
 
-  sprintf(qstr, "create database if not exists %s cache %d tables %d", dbName, cache, tables);
-  if (taos_query(con, qstr)) {
-    pError("failed to create database:%s, code:%d reason:%s", dbName, taos_errno(con), taos_errstr(con));
+  sprintf(qstr, "create database if not exists %s cache %d maxtables %d", dbName, cache, tables);
+  pSql = taos_query(con, qstr);
+  int32_t code = taos_errno(pSql);
+  if (code != 0) {
+    pError("failed to create database:%s, sql:%s, code:%d reason:%s", dbName, qstr, taos_errno(con), taos_errstr(con));
     exit(0);
   }
+  taos_free_result(pSql);
 
   sprintf(qstr, "use %s", dbName);
-  if (taos_query(con, qstr)) {
+  pSql = taos_query(con, qstr);
+  code = taos_errno(pSql);
+  if (code != 0) {
     pError("failed to use db, code:%d reason:%s", taos_errno(con), taos_errstr(con));
     exit(0);
   }
-
-  gettimeofday(&systemTime, NULL);
-  st = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
-  int64_t totalTables = numOfTablesPerThread * numOfThreads;
+  taos_free_result(pSql);
 
   if (strcmp(stableName, "no") != 0) {
     int len = sprintf(qstr, "create table if not exists %s(ts timestamp", stableName);
@@ -100,36 +102,14 @@ void createDbAndTable() {
     }
     sprintf(qstr + len, ") tags(t int)");
 
-    if (taos_query(con, qstr)) {
+    pSql = taos_query(con, qstr);
+    code = taos_errno(pSql);
+    if (code != 0) {
       pError("failed to create stable, code:%d reason:%s", taos_errno(con), taos_errstr(con));
       exit(0);
     }
-    
-    for (int64_t t = 0; t < totalTables; ++t) {
-      sprintf(qstr, "create table if not exists %s%ld using %s tags(%ld)", stableName, t, stableName, t);
-      if (taos_query(con, qstr)) {
-        pError("failed to create table %s%d, reason:%s", stableName, t, taos_errstr(con));
-        exit(0);
-      }
-    }
-  } else {
-    for (int64_t t = 0; t < totalTables; ++t) {
-      int len = sprintf(qstr, "create table if not exists %s%ld(ts timestamp", stableName, t);
-      for (int64_t f = 0; f < pointsPerTable; ++f) {
-        len += sprintf(qstr + len, ", f%ld double", f);
-      }
-      sprintf(qstr + len, ")");
-
-      if (taos_query(con, qstr)) {
-        pError("failed to create table %s%ld, reason:%s", stableName, t, taos_errstr(con));
-        exit(0);
-      }
-    }
+    taos_free_result(pSql);
   }
-
-  gettimeofday(&systemTime, NULL);
-  et = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
-  pPrint("%.1f seconds to create %ld tables", (et - st) / 1000.0 / 1000.0, totalTables);
 }
 
 void insertData() {
@@ -144,7 +124,7 @@ void insertData() {
   pthread_attr_t thattr;
   pthread_attr_init(&thattr);
   pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
-  SInfo *pInfo = (SInfo *)malloc(sizeof(SInfo) * numOfThreads);
+  SInfo *pInfo = (SInfo *)calloc(numOfThreads, sizeof(SInfo));
 
   // Start threads to write
   for (int i = 0; i < numOfThreads; ++i) {
@@ -173,10 +153,15 @@ void insertData() {
   double  speedOfRows = totalRows / seconds;
   double  speedOfPoints = totalPoints / seconds;
 
+  float createTableSpeed = 0;
+  for (int i = 0; i < numOfThreads; ++i) {
+    createTableSpeed += pInfo[i].createTableSpeed;
+  }
+
   pPrint(
       "%sall threads:%ld finished, use %.1lf seconds, tables:%.ld rows:%ld points:%ld, speed RowsPerSecond:%.1lf "
-      "PointsPerSecond:%.1lf%s",
-      GREEN, numOfThreads, seconds, totalTables, totalRows, totalPoints, speedOfRows, speedOfPoints, NC);
+      "PointsPerSecond:%.1lf CreateTableSpeed:%.1f t/s %s",
+      GREEN, numOfThreads, seconds, totalTables, totalRows, totalPoints, speedOfRows, speedOfPoints, createTableSpeed, NC);
 
   pPrint("threads exit");
 
@@ -191,6 +176,7 @@ void *syncTest(void *param) {
   int64_t        st, et;
   char           qstr[65000];
   int            maxBytes = 60000;
+  int            code;
 
   pPrint("thread:%d, start to run", pInfo->threadIndex);
 
@@ -210,6 +196,48 @@ void *syncTest(void *param) {
   gettimeofday(&systemTime, NULL);
   st = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
 
+  if (strcmp(stableName, "no") != 0) {
+    for (int64_t t = pInfo->tableBeginIndex; t < pInfo->tableEndIndex; ++t) {
+      sprintf(qstr, "create table if not exists %s%ld using %s tags(%ld)", stableName, t, stableName, t);
+      TAOS_RES *pSql = taos_query(con, qstr);
+      code = taos_errno(pSql);
+      if (code != 0) {
+        pError("failed to create table %s%d, reason:%s", stableName, t, taos_errstr(con));
+        exit(0);
+      }
+      taos_free_result(pSql);
+    }
+  } else {
+    for (int64_t t = pInfo->tableBeginIndex; t < pInfo->tableEndIndex; ++t) {
+      int len = sprintf(qstr, "create table if not exists %s%ld(ts timestamp", stableName, t);
+      for (int64_t f = 0; f < pointsPerTable; ++f) {
+        len += sprintf(qstr + len, ", f%ld double", f);
+      }
+      sprintf(qstr + len, ")");
+
+      TAOS_RES *pSql = taos_query(con, qstr);
+      code = taos_errno(pSql);
+      if (code != 0) {
+        pError("failed to create table %s%ld, reason:%s", stableName, t, taos_errstr(con));
+        exit(0);
+      }
+      taos_free_result(pSql);
+    }
+  }
+
+  gettimeofday(&systemTime, NULL);
+  et = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
+  float seconds = (et - st) / 1000.0 / 1000.0;
+  int64_t tables = pInfo->tableEndIndex - pInfo->tableBeginIndex;
+  pInfo->createTableSpeed = (float)tables / seconds;
+  pPrint("thread:%d, %.1f seconds to create %ld tables, speed:%.1f", pInfo->threadIndex, seconds, tables,
+         pInfo->createTableSpeed);
+
+  if (pInfo->rowsPerTable == 0) return NULL;       
+
+  gettimeofday(&systemTime, NULL);
+  st = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
+
   int64_t start = 1430000000000;
   int64_t interval = 1000;  // 1000 ms
 
@@ -227,10 +255,13 @@ void *syncTest(void *param) {
       }
       len += sprintf(sql + len, ")");
       if (len > maxBytes) {
-        if (taos_query(con, qstr)) {
+        TAOS_RES *pSql = taos_query(con, qstr);
+        int32_t code = taos_errno(pSql);
+        if (code != 0) {
           pError("thread:%d, failed to insert table:%s%ld row:%ld, reason:%s", pInfo->threadIndex, pInfo->stableName,
                  table, row, taos_errstr(con));
         }
+        taos_free_result(pSql);
 
         // "insert into"
         len = sprintf(sql, "%s", inserStr);
@@ -239,7 +270,8 @@ void *syncTest(void *param) {
   }
 
   if (len != strlen(inserStr)) {
-    taos_query(con, qstr);
+    TAOS_RES *pSql = taos_query(con, qstr);
+    taos_free_result(pSql);
   }
 
   gettimeofday(&systemTime, NULL);

tests/test/c/insertPerTable.c

@@ -49,8 +49,8 @@ int64_t numOfThreads = 1;
 int64_t numOfTablesPerThread = 1;
 char    dbName[32] = "db";
 char    stableName[64] = "st";
-int32_t cache = 16384;
-int32_t tables = 1000;
+int32_t cache = 16;
+int32_t tables = 5000;
 
 int main(int argc, char *argv[]) {
   shellParseArgument(argc, argv);
@@ -63,6 +63,7 @@ int main(int argc, char *argv[]) {
 void createDbAndTable() {
   pPrint("start to create table");
 
+  TAOS_RES *     pSql;
   TAOS *         con;
   struct timeval systemTime;
   int64_t        st, et;
@@ -79,17 +80,22 @@ void createDbAndTable() {
     exit(1);
   }
 
-  sprintf(qstr, "create database if not exists %s cache %d tables %d", dbName, cache, tables);
-  if (taos_query(con, qstr)) {
-    pError("failed to create database:%s, code:%d reason:%s", dbName, taos_errno(con), taos_errstr(con));
+  sprintf(qstr, "create database if not exists %s cache %d maxtables %d", dbName, cache, tables);
+  pSql = taos_query(con, qstr);
+  int32_t code = taos_errno(pSql);
+  if (code != 0) {
+    pError("failed to create database:%s, sql:%s, code:%d reason:%s", dbName, qstr, taos_errno(con), taos_errstr(con));
     exit(0);
   }
 
   sprintf(qstr, "use %s", dbName);
-  if (taos_query(con, qstr)) {
+  pSql = taos_query(con, qstr);
+  code = taos_errno(pSql);
+  if (code != 0) {
     pError("failed to use db, code:%d reason:%s", taos_errno(con), taos_errstr(con));
     exit(0);
   }
+  taos_stop_query(pSql);
 
   gettimeofday(&systemTime, NULL);
   st = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
@@ -102,17 +108,23 @@ void createDbAndTable() {
     }
     sprintf(qstr + len, ") tags(t int)");
 
-    if (taos_query(con, qstr)) {
+    pSql = taos_query(con, qstr);
+    code = taos_errno(pSql);
+    if (code != 0) {
       pError("failed to create stable, code:%d reason:%s", taos_errno(con), taos_errstr(con));
       exit(0);
     }
+    taos_stop_query(pSql);
 
     for (int64_t t = 0; t < totalTables; ++t) {
       sprintf(qstr, "create table if not exists %s%ld using %s tags(%ld)", stableName, t, stableName, t);
-      if (taos_query(con, qstr)) {
+      pSql = taos_query(con, qstr);
+      code = taos_errno(pSql);
+      if (code != 0) {
         pError("failed to create table %s%d, reason:%s", stableName, t, taos_errstr(con));
         exit(0);
       }
+      taos_stop_query(pSql);
     }
   } else {
     for (int64_t t = 0; t < totalTables; ++t) {
@@ -122,16 +134,20 @@ void createDbAndTable() {
       }
       sprintf(qstr + len, ")");
 
-      if (taos_query(con, qstr)) {
+      pSql = taos_query(con, qstr);
+      code = taos_errno(pSql);
+      if (code != 0) {
         pError("failed to create table %s%ld, reason:%s", stableName, t, taos_errstr(con));
         exit(0);
       }
+      taos_stop_query(pSql);
     }
   }
 
   gettimeofday(&systemTime, NULL);
   et = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
-  pPrint("%.1f seconds to create %ld tables", (et - st) / 1000.0 / 1000.0, totalTables);
+  float seconds = (et - st) / 1000.0 / 1000.0;
+  pPrint("%.1f seconds to create %ld tables, speed:%.1f", seconds, totalTables, totalTables / seconds);
 }
 
 void insertData() {
@@ -141,7 +157,12 @@ void insertData() {
   gettimeofday(&systemTime, NULL);
   st = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
 
+  if (rowsPerTable <= 0) {
+    pPrint("not insert data for rowsPerTable is :%d", rowsPerTable);
+    exit(0);
+  } else {
     pPrint("%d threads are spawned to insert data", numOfThreads);
+  }
 
   pthread_attr_t thattr;
   pthread_attr_init(&thattr);
@@ -230,10 +251,13 @@ void *syncTest(void *param) {
       }
       len += sprintf(sql + len, ")");
       if (len > maxBytes) {
-        if (taos_query(con, qstr)) {
+        TAOS_RES *pSql = taos_query(con, qstr);
+        int32_t code = taos_errno(pSql);
+        if (code != 0) {
           pError("thread:%d, failed to insert table:%s%ld row:%ld, reason:%s", pInfo->threadIndex, pInfo->stableName,
                  table, row, taos_errstr(con));
         }
+        taos_stop_query(pSql);
 
         // "insert into"
         len = sprintf(sql, "%s", inserStr);