Merge pull request #14514 from taosdata/fix/tsim

enh: check available memory of dnode while creating db

include/common/tmsg.h

@@ -708,6 +708,7 @@ typedef struct {
   int32_t buffer;        // MB
   int32_t pageSize;
   int32_t pages;
+  int32_t lastRowMem;
   int32_t daysPerFile;
   int32_t daysToKeep0;
   int32_t daysToKeep1;
@@ -736,6 +737,7 @@ typedef struct {
   int32_t buffer;
   int32_t pageSize;
   int32_t pages;
+  int32_t lastRowMem;
   int32_t daysPerFile;
   int32_t daysToKeep0;
   int32_t daysToKeep1;
@@ -1023,8 +1025,10 @@ typedef struct {
   int64_t     clusterId;
   int64_t     rebootTime;
   int64_t     updateTime;
-  int32_t     numOfCores;
+  float       numOfCores;
   int32_t     numOfSupportVnodes;
+  int64_t     memTotal;
+  int64_t     memAvail;
   char        dnodeEp[TSDB_EP_LEN];
   SMnodeLoad  mload;
   SQnodeLoad  qload;
@@ -1079,6 +1083,7 @@ typedef struct {
   int32_t  buffer;
   int32_t  pageSize;
   int32_t  pages;
+  int32_t  lastRowMem;
   int32_t  daysPerFile;
   int32_t  daysToKeep0;
   int32_t  daysToKeep1;
@@ -1131,6 +1136,7 @@ typedef struct {
   int32_t  buffer;
   int32_t  pageSize;
   int32_t  pages;
+  int32_t  lastRowMem;
   int32_t  daysPerFile;
   int32_t  daysToKeep0;
   int32_t  daysToKeep1;

include/util/taoserror.h

@@ -173,11 +173,12 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_MND_DNODE_NOT_EXIST           TAOS_DEF_ERROR_CODE(0, 0x0341)
 #define TSDB_CODE_MND_TOO_MANY_DNODES           TAOS_DEF_ERROR_CODE(0, 0x0342)
 #define TSDB_CODE_MND_NO_ENOUGH_DNODES          TAOS_DEF_ERROR_CODE(0, 0x0343)
-#define TSDB_CODE_MND_INVALID_CLUSTER_CFG       TAOS_DEF_ERROR_CODE(0, 0x0344)
-#define TSDB_CODE_MND_INVALID_CLUSTER_ID        TAOS_DEF_ERROR_CODE(0, 0x0345)
-#define TSDB_CODE_MND_INVALID_DNODE_CFG         TAOS_DEF_ERROR_CODE(0, 0x0346)
-#define TSDB_CODE_MND_INVALID_DNODE_EP          TAOS_DEF_ERROR_CODE(0, 0x0347)
-#define TSDB_CODE_MND_INVALID_DNODE_ID          TAOS_DEF_ERROR_CODE(0, 0x0348)
+#define TSDB_CODE_MND_NO_ENOUGH_MEM_IN_DNODE    TAOS_DEF_ERROR_CODE(0, 0x0344)
+#define TSDB_CODE_MND_INVALID_CLUSTER_CFG       TAOS_DEF_ERROR_CODE(0, 0x0345)
+#define TSDB_CODE_MND_INVALID_CLUSTER_ID        TAOS_DEF_ERROR_CODE(0, 0x0346)
+#define TSDB_CODE_MND_INVALID_DNODE_CFG         TAOS_DEF_ERROR_CODE(0, 0x0347)
+#define TSDB_CODE_MND_INVALID_DNODE_EP          TAOS_DEF_ERROR_CODE(0, 0x0348)
+#define TSDB_CODE_MND_INVALID_DNODE_ID          TAOS_DEF_ERROR_CODE(0, 0x0349)
 
 // mnode-node
 #define TSDB_CODE_MND_MNODE_ALREADY_EXIST       TAOS_DEF_ERROR_CODE(0, 0x0350)

include/util/tdef.h

@@ -334,6 +334,9 @@ typedef enum ELogicConditionType {
 #define TSDB_MIN_DB_CACHE_LAST_ROW      0
 #define TSDB_MAX_DB_CACHE_LAST_ROW      3
 #define TSDB_DEFAULT_CACHE_LAST_ROW     0
+#define TSDB_MIN_DB_LAST_ROW_MEM        1  // MB
+#define TSDB_MAX_DB_LAST_ROW_MEM        65536
+#define TSDB_DEFAULT_LAST_ROW_MEM       1
 #define TSDB_DB_STREAM_MODE_OFF         0
 #define TSDB_DB_STREAM_MODE_ON          1
 #define TSDB_DEFAULT_DB_STREAM_MODE     0

source/common/src/tmsg.c

@@ -947,8 +947,10 @@ int32_t tSerializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) {
   if (tEncodeI64(&encoder, pReq->clusterId) < 0) return -1;
   if (tEncodeI64(&encoder, pReq->rebootTime) < 0) return -1;
   if (tEncodeI64(&encoder, pReq->updateTime) < 0) return -1;
-  if (tEncodeI32(&encoder, pReq->numOfCores) < 0) return -1;
+  if (tEncodeFloat(&encoder, pReq->numOfCores) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->numOfSupportVnodes) < 0) return -1;
+  if (tEncodeI64(&encoder, pReq->memTotal) < 0) return -1;
+  if (tEncodeI64(&encoder, pReq->memAvail) < 0) return -1;
   if (tEncodeCStr(&encoder, pReq->dnodeEp) < 0) return -1;
 
   // cluster cfg
@@ -1008,8 +1010,10 @@ int32_t tDeserializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) {
   if (tDecodeI64(&decoder, &pReq->clusterId) < 0) return -1;
   if (tDecodeI64(&decoder, &pReq->rebootTime) < 0) return -1;
   if (tDecodeI64(&decoder, &pReq->updateTime) < 0) return -1;
-  if (tDecodeI32(&decoder, &pReq->numOfCores) < 0) return -1;
+  if (tDecodeFloat(&decoder, &pReq->numOfCores) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->numOfSupportVnodes) < 0) return -1;
+  if (tDecodeI64(&decoder, &pReq->memTotal) < 0) return -1;
+  if (tDecodeI64(&decoder, &pReq->memAvail) < 0) return -1;
   if (tDecodeCStrTo(&decoder, pReq->dnodeEp) < 0) return -1;
 
   // cluster cfg
@@ -1974,6 +1978,7 @@ int32_t tSerializeSCreateDbReq(void *buf, int32_t bufLen, SCreateDbReq *pReq) {
   if (tEncodeI32(&encoder, pReq->buffer) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->pageSize) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->pages) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->lastRowMem) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->daysPerFile) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->daysToKeep0) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->daysToKeep1) < 0) return -1;
@@ -2015,6 +2020,7 @@ int32_t tDeserializeSCreateDbReq(void *buf, int32_t bufLen, SCreateDbReq *pReq)
   if (tDecodeI32(&decoder, &pReq->buffer) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->pageSize) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->pages) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->lastRowMem) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->daysPerFile) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->daysToKeep0) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->daysToKeep1) < 0) return -1;
@@ -2069,6 +2075,7 @@ int32_t tSerializeSAlterDbReq(void *buf, int32_t bufLen, SAlterDbReq *pReq) {
   if (tEncodeI32(&encoder, pReq->buffer) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->pageSize) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->pages) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->lastRowMem) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->daysPerFile) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->daysToKeep0) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->daysToKeep1) < 0) return -1;
@@ -2094,6 +2101,7 @@ int32_t tDeserializeSAlterDbReq(void *buf, int32_t bufLen, SAlterDbReq *pReq) {
   if (tDecodeI32(&decoder, &pReq->buffer) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->pageSize) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->pages) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->lastRowMem) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->daysPerFile) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->daysToKeep0) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->daysToKeep1) < 0) return -1;
@@ -3586,6 +3594,7 @@ int32_t tSerializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *pR
   if (tEncodeI32(&encoder, pReq->buffer) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->pageSize) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->pages) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->lastRowMem) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->daysPerFile) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->daysToKeep0) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->daysToKeep1) < 0) return -1;
@@ -3643,6 +3652,7 @@ int32_t tDeserializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *
   if (tDecodeI32(&decoder, &pReq->buffer) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->pageSize) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->pages) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->lastRowMem) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->daysPerFile) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->daysToKeep0) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->daysToKeep1) < 0) return -1;
@@ -3665,7 +3675,6 @@ int32_t tDeserializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *
     SReplica *pReplica = &pReq->replicas[i];
     if (tDecodeSReplica(&decoder, pReplica) < 0) return -1;
   }
-
   if (tDecodeI32(&decoder, &pReq->numOfRetensions) < 0) return -1;
   pReq->pRetensions = taosArrayInit(pReq->numOfRetensions, sizeof(SRetention));
   if (pReq->pRetensions == NULL) {
@@ -3768,6 +3777,7 @@ int32_t tSerializeSAlterVnodeReq(void *buf, int32_t bufLen, SAlterVnodeReq *pReq
   if (tEncodeI32(&encoder, pReq->buffer) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->pageSize) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->pages) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->lastRowMem) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->daysPerFile) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->daysToKeep0) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->daysToKeep1) < 0) return -1;
@@ -3782,7 +3792,6 @@ int32_t tSerializeSAlterVnodeReq(void *buf, int32_t bufLen, SAlterVnodeReq *pReq
     SReplica *pReplica = &pReq->replicas[i];
     if (tEncodeSReplica(&encoder, pReplica) < 0) return -1;
   }
-
   tEndEncode(&encoder);
 
   int32_t tlen = encoder.pos;
@@ -3799,6 +3808,7 @@ int32_t tDeserializeSAlterVnodeReq(void *buf, int32_t bufLen, SAlterVnodeReq *pR
   if (tDecodeI32(&decoder, &pReq->buffer) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->pageSize) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->pages) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->lastRowMem) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->daysPerFile) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->daysToKeep0) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->daysToKeep1) < 0) return -1;

source/dnode/mgmt/mgmt_dnode/src/dmHandle.c

@@ -67,6 +67,8 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
   req.updateTime = pMgmt->pData->updateTime;
   req.numOfCores = tsNumOfCores;
   req.numOfSupportVnodes = tsNumOfSupportVnodes;
+  req.memTotal = tsTotalMemoryKB * 1024;
+  req.memAvail = req.memTotal - tsRpcQueueMemoryAllowed - 16 * 1024 * 1024;
   tstrncpy(req.dnodeEp, tsLocalEp, TSDB_EP_LEN);
 
   req.clusterCfg.statusInterval = tsStatusInterval;

source/dnode/mgmt/node_util/src/dmEps.c

@@ -280,7 +280,7 @@ static void dmPrintEps(SDnodeData *pData) {
   dDebug("print dnode list, num:%d", numOfEps);
   for (int32_t i = 0; i < numOfEps; i++) {
     SDnodeEp *pEp = taosArrayGet(pData->dnodeEps, i);
-    dDebug("dnode:%d, fqdn:%s port:%u is_mnode:%d", pEp->id, pEp->ep.fqdn, pEp->ep.port, pEp->isMnode);
+    dDebug("dnode:%d, fqdn:%s port:%u isMnode:%d", pEp->id, pEp->ep.fqdn, pEp->ep.port, pEp->isMnode);
   }
 }
 

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

@@ -30,12 +30,14 @@ void Testbase::InitLog(const char* path) {
   tsdbDebugFlag = 0;
   tsLogEmbedded = 1;
   tsAsyncLog = 0;
-  tsRpcQueueMemoryAllowed = 1024 * 1024 * 64;
   
   taosRemoveDir(path);
   taosMkDir(path);
   tstrncpy(tsLogDir, path, PATH_MAX);
-  if (taosInitLog("taosdlog", 1) != 0) {
+
+  taosGetSystemInfo();
+  tsRpcQueueMemoryAllowed = tsTotalMemoryKB * 0.1;
+if (taosInitLog("taosdlog", 1) != 0) {
     printf("failed to init log file\n");
   }
 }

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

@@ -148,7 +148,10 @@ typedef struct {
   int32_t    accessTimes;
   int32_t    numOfVnodes;
   int32_t    numOfSupportVnodes;
-  int32_t    numOfCores;
+  float      numOfCores;
+  int64_t    memTotal;
+  int64_t    memAvail;
+  int64_t    memUsed;
   EDndReason offlineReason;
   uint16_t   port;
   char       fqdn[TSDB_FQDN_LEN];
@@ -243,6 +246,7 @@ typedef struct {
   int32_t buffer;
   int32_t pageSize;
   int32_t pages;
+  int32_t lastRowMem;
   int32_t daysPerFile;
   int32_t daysToKeep0;
   int32_t daysToKeep1;
@@ -255,8 +259,8 @@ typedef struct {
   int8_t  compression;
   int8_t  replications;
   int8_t  strict;
-  int8_t  cacheLastRow;
   int8_t  hashMethod;  // default is 1
+  int8_t  cacheLastRow;
   int32_t numOfRetensions;
   SArray* pRetensions;
   int8_t  schemaless;

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

@@ -30,6 +30,8 @@ SSdbRaw *mndVgroupActionEncode(SVgObj *pVgroup);
 SEpSet   mndGetVgroupEpset(SMnode *pMnode, const SVgObj *pVgroup);
 int32_t  mndGetVnodesNum(SMnode *pMnode, int32_t dnodeId);
 void     mndSortVnodeGid(SVgObj *pVgroup);
+int64_t  mndGetVnodesMemory(SMnode *pMnode, int32_t dnodeId);
+int64_t  mndGetVgroupMemory(SMnode *pMnode, SDbObj *pDb, SVgObj *pVgroup);
 
 SArray *mndBuildDnodesArray(SMnode *, int32_t exceptDnodeId);
 int32_t mndAllocSmaVgroup(SMnode *, SDbObj *pDb, SVgObj *pVgroup);

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

@@ -93,6 +93,7 @@ static SSdbRaw *mndDbActionEncode(SDbObj *pDb) {
   SDB_SET_INT32(pRaw, dataPos, pDb->cfg.buffer, _OVER)
   SDB_SET_INT32(pRaw, dataPos, pDb->cfg.pageSize, _OVER)
   SDB_SET_INT32(pRaw, dataPos, pDb->cfg.pages, _OVER)
+  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.lastRowMem, _OVER)
   SDB_SET_INT32(pRaw, dataPos, pDb->cfg.daysPerFile, _OVER)
   SDB_SET_INT32(pRaw, dataPos, pDb->cfg.daysToKeep0, _OVER)
   SDB_SET_INT32(pRaw, dataPos, pDb->cfg.daysToKeep1, _OVER)
@@ -165,6 +166,7 @@ static SSdbRow *mndDbActionDecode(SSdbRaw *pRaw) {
   SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.buffer, _OVER)
   SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.pageSize, _OVER)
   SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.pages, _OVER)
+  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.lastRowMem, _OVER)
   SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.daysPerFile, _OVER)
   SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.daysToKeep0, _OVER)
   SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.daysToKeep1, _OVER)
@@ -230,8 +232,9 @@ static int32_t mndDbActionUpdate(SSdb *pSdb, SDbObj *pOld, SDbObj *pNew) {
   pOld->cfgVersion = pNew->cfgVersion;
   pOld->vgVersion = pNew->vgVersion;
   pOld->cfg.buffer = pNew->cfg.buffer;
-  pOld->cfg.pages = pNew->cfg.pages;
   pOld->cfg.pageSize = pNew->cfg.pageSize;
+  pOld->cfg.pages = pNew->cfg.pages;
+  pOld->cfg.lastRowMem = pNew->cfg.lastRowMem;
   pOld->cfg.daysPerFile = pNew->cfg.daysPerFile;
   pOld->cfg.daysToKeep0 = pNew->cfg.daysToKeep0;
   pOld->cfg.daysToKeep1 = pNew->cfg.daysToKeep1;
@@ -288,6 +291,7 @@ static int32_t mndCheckDbCfg(SMnode *pMnode, SDbCfg *pCfg) {
   if (pCfg->buffer < TSDB_MIN_BUFFER_PER_VNODE || pCfg->buffer > TSDB_MAX_BUFFER_PER_VNODE) return -1;
   if (pCfg->pageSize < TSDB_MIN_PAGESIZE_PER_VNODE || pCfg->pageSize > TSDB_MAX_PAGESIZE_PER_VNODE) return -1;
   if (pCfg->pages < TSDB_MIN_PAGES_PER_VNODE || pCfg->pages > TSDB_MAX_PAGES_PER_VNODE) return -1;
+  if (pCfg->lastRowMem < TSDB_MIN_DB_LAST_ROW_MEM || pCfg->lastRowMem > TSDB_MAX_DB_LAST_ROW_MEM) return -1;
   if (pCfg->daysPerFile < TSDB_MIN_DAYS_PER_FILE || pCfg->daysPerFile > TSDB_MAX_DAYS_PER_FILE) return -1;
   if (pCfg->daysToKeep0 < TSDB_MIN_KEEP || pCfg->daysToKeep0 > TSDB_MAX_KEEP) return -1;
   if (pCfg->daysToKeep1 < TSDB_MIN_KEEP || pCfg->daysToKeep1 > TSDB_MAX_KEEP) return -1;
@@ -336,6 +340,7 @@ static void mndSetDefaultDbCfg(SDbCfg *pCfg) {
   if (pCfg->replications < 0) pCfg->replications = TSDB_DEFAULT_DB_REPLICA;
   if (pCfg->strict < 0) pCfg->strict = TSDB_DEFAULT_DB_STRICT;
   if (pCfg->cacheLastRow < 0) pCfg->cacheLastRow = TSDB_DEFAULT_CACHE_LAST_ROW;
+  if (pCfg->lastRowMem <= 0) pCfg->lastRowMem = TSDB_DEFAULT_LAST_ROW_MEM;
   if (pCfg->numOfRetensions < 0) pCfg->numOfRetensions = 0;
   if (pCfg->schemaless < 0) pCfg->schemaless = TSDB_DB_SCHEMALESS_OFF;
 }
@@ -434,6 +439,7 @@ static int32_t mndCreateDb(SMnode *pMnode, SRpcMsg *pReq, SCreateDbReq *pCreate,
       .buffer = pCreate->buffer,
       .pageSize = pCreate->pageSize,
       .pages = pCreate->pages,
+      .lastRowMem = pCreate->lastRowMem,
       .daysPerFile = pCreate->daysPerFile,
       .daysToKeep0 = pCreate->daysToKeep0,
       .daysToKeep1 = pCreate->daysToKeep1,
@@ -475,7 +481,7 @@ static int32_t mndCreateDb(SMnode *pMnode, SRpcMsg *pReq, SCreateDbReq *pCreate,
   int32_t code = -1;
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq);
   if (pTrans == NULL) goto _OVER;
-
+  // mndTransSetSerial(pTrans);
   mDebug("trans:%d, used to create db:%s", pTrans->id, pCreate->db);
 
   mndTransSetDbName(pTrans, dbObj.name, NULL);
@@ -622,6 +628,11 @@ static int32_t mndSetDbCfgFromAlterDbReq(SDbObj *pDb, SAlterDbReq *pAlter) {
     terrno = 0;
   }
 
+  if (pAlter->lastRowMem > 0 && pAlter->lastRowMem != pDb->cfg.lastRowMem) {
+    pDb->cfg.lastRowMem = pAlter->lastRowMem;
+    terrno = 0;
+  }
+
   if (pAlter->replications > 0 && pAlter->replications != pDb->cfg.replications) {
 #if 1
     terrno = TSDB_CODE_OPS_NOT_SUPPORT;

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

@@ -15,8 +15,8 @@
 
 #define _DEFAULT_SOURCE
 #include "mndDnode.h"
-#include "mndPrivilege.h"
 #include "mndMnode.h"
+#include "mndPrivilege.h"
 #include "mndQnode.h"
 #include "mndShow.h"
 #include "mndSnode.h"
@@ -432,7 +432,8 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) {
     }
 
     if (!online) {
-      mInfo("dnode:%d, from offline to online", pDnode->id);
+      mInfo("dnode:%d, from offline to online, memory avail:%" PRId64 " total:%" PRId64 " cores:%.2f", pDnode->id,
+            statusReq.memAvail, statusReq.memTotal, statusReq.numOfCores);
     } else {
       mDebug("dnode:%d, send dnode epset, online:%d dnodeVer:%" PRId64 ":%" PRId64 " reboot:%d", pDnode->id, online,
              statusReq.dnodeVer, dnodeVer, reboot);
@@ -441,6 +442,8 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) {
     pDnode->rebootTime = statusReq.rebootTime;
     pDnode->numOfCores = statusReq.numOfCores;
     pDnode->numOfSupportVnodes = statusReq.numOfSupportVnodes;
+    pDnode->memAvail = statusReq.memAvail;
+    pDnode->memTotal = statusReq.memTotal;
 
     SStatusRsp statusRsp = {0};
     statusRsp.dnodeVer = dnodeVer;

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

@@ -207,6 +207,7 @@ void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVg
   createReq.buffer = pDb->cfg.buffer;
   createReq.pageSize = pDb->cfg.pageSize;
   createReq.pages = pDb->cfg.pages;
+  createReq.lastRowMem = pDb->cfg.lastRowMem;
   createReq.daysPerFile = pDb->cfg.daysPerFile;
   createReq.daysToKeep0 = pDb->cfg.daysToKeep0;
   createReq.daysToKeep1 = pDb->cfg.daysToKeep1;
@@ -274,8 +275,9 @@ void *mndBuildAlterVnodeReq(SMnode *pMnode, SDbObj *pDb, SVgObj *pVgroup, int32_
   SAlterVnodeReq alterReq = {0};
   alterReq.vgVersion = pVgroup->version;
   alterReq.buffer = pDb->cfg.buffer;
-  alterReq.pages = pDb->cfg.pages;
   alterReq.pageSize = pDb->cfg.pageSize;
+  alterReq.pages = pDb->cfg.pages;
+  alterReq.lastRowMem = pDb->cfg.lastRowMem;
   alterReq.daysPerFile = pDb->cfg.daysPerFile;
   alterReq.daysToKeep0 = pDb->cfg.daysToKeep0;
   alterReq.daysToKeep1 = pDb->cfg.daysToKeep1;
@@ -392,9 +394,10 @@ static bool mndBuildDnodesArrayFp(SMnode *pMnode, void *pObj, void *p1, void *p2
   bool    online = mndIsDnodeOnline(pDnode, curMs);
   bool    isMnode = mndIsMnode(pMnode, pDnode->id);
   pDnode->numOfVnodes = mndGetVnodesNum(pMnode, pDnode->id);
+  pDnode->memUsed = mndGetVnodesMemory(pMnode, pDnode->id);
 
-  mDebug("dnode:%d, vnodes:%d support_vnodes:%d is_mnode:%d online:%d", pDnode->id, pDnode->numOfVnodes,
-         pDnode->numOfSupportVnodes, isMnode, online);
+  mDebug("dnode:%d, vnodes:%d supportVnodes:%d isMnode:%d online:%d memory avail:%" PRId64 " used:%" PRId64, pDnode->id,
+         pDnode->numOfVnodes, pDnode->numOfSupportVnodes, isMnode, online, pDnode->memAvail, pDnode->memUsed);
 
   if (isMnode) {
     pDnode->numOfVnodes++;
@@ -426,15 +429,7 @@ static int32_t mndCompareDnodeId(int32_t *dnode1Id, int32_t *dnode2Id) { return
 static int32_t mndCompareDnodeVnodes(SDnodeObj *pDnode1, SDnodeObj *pDnode2) {
   float d1Score = (float)pDnode1->numOfVnodes / pDnode1->numOfSupportVnodes;
   float d2Score = (float)pDnode2->numOfVnodes / pDnode2->numOfSupportVnodes;
-#if 0  
-  if (d1Score == d2Score) {
-    return pDnode2->id - pDnode1->id;
-  } else {
   return d1Score >= d2Score ? 1 : 0;
-  }
-#else
-  return d1Score >= d2Score ? 1 : 0;
-#endif
 }
 
 void mndSortVnodeGid(SVgObj *pVgroup) {
@@ -447,7 +442,7 @@ void mndSortVnodeGid(SVgObj *pVgroup) {
   }
 }
 
-static int32_t mndGetAvailableDnode(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray) {
+static int32_t mndGetAvailableDnode(SMnode *pMnode, SDbObj *pDb, SVgObj *pVgroup, SArray *pArray) {
   SSdb   *pSdb = pMnode->pSdb;
   int32_t allocedVnodes = 0;
   void   *pIter = NULL;
@@ -470,6 +465,16 @@ static int32_t mndGetAvailableDnode(SMnode *pMnode, SVgObj *pVgroup, SArray *pAr
       return -1;
     }
 
+    int64_t vgMem = mndGetVgroupMemory(pMnode, pDb, pVgroup);
+    if (pDnode->memAvail - vgMem - pDnode->memUsed <= 0) {
+      mError("db:%s, vgId:%d, no enough memory:%" PRId64 " in dnode:%d, avail:%" PRId64 " used:%" PRId64,
+             pVgroup->dbName, pVgroup->vgId, vgMem, pDnode->id, pDnode->memAvail, pDnode->memUsed);
+      terrno = TSDB_CODE_MND_NO_ENOUGH_MEM_IN_DNODE;
+      return -1;
+    } else {
+      pDnode->memUsed += vgMem;
+    }
+
     pVgid->dnodeId = pDnode->id;
     if (pVgroup->replica == 1) {
       pVgid->role = TAOS_SYNC_STATE_LEADER;
@@ -477,7 +482,8 @@ static int32_t mndGetAvailableDnode(SMnode *pMnode, SVgObj *pVgroup, SArray *pAr
       pVgid->role = TAOS_SYNC_STATE_FOLLOWER;
     }
 
-    mInfo("db:%s, vgId:%d, vn:%d dnode:%d is alloced", pVgroup->dbName, pVgroup->vgId, v, pVgid->dnodeId);
+    mInfo("db:%s, vgId:%d, vn:%d is alloced, memory:%" PRId64 ", dnode:%d avail:%" PRId64 " used:%" PRId64,
+          pVgroup->dbName, pVgroup->vgId, v, vgMem, pVgid->dnodeId, pDnode->memAvail, pDnode->memUsed);
     pDnode->numOfVnodes++;
   }
 
@@ -498,7 +504,7 @@ int32_t mndAllocSmaVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj *pVgroup) {
   pVgroup->dbUid = pDb->uid;
   pVgroup->replica = 1;
 
-  if (mndGetAvailableDnode(pMnode, pVgroup, pArray) != 0) return -1;
+  if (mndGetAvailableDnode(pMnode, pDb, pVgroup, pArray) != 0) return -1;
 
   mInfo("db:%s, sma vgId:%d is alloced", pDb->name, pVgroup->vgId);
   return 0;
@@ -546,8 +552,7 @@ int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups) {
     pVgroup->dbUid = pDb->uid;
     pVgroup->replica = pDb->cfg.replications;
 
-    if (mndGetAvailableDnode(pMnode, pVgroup, pArray) != 0) {
-      terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
+    if (mndGetAvailableDnode(pMnode, pDb, pVgroup, pArray) != 0) {
       goto _OVER;
     }
 
@@ -728,6 +733,43 @@ int32_t mndGetVnodesNum(SMnode *pMnode, int32_t dnodeId) {
   return numOfVnodes;
 }
 
+int64_t mndGetVgroupMemory(SMnode *pMnode, SDbObj *pDbInput, SVgObj *pVgroup) {
+  SDbObj *pDb = pDbInput;
+  if (pDbInput == NULL) {
+    pDb = mndAcquireDb(pMnode, pVgroup->dbName);
+  }
+
+  int64_t vgroupMemroy = (int64_t)pDb->cfg.buffer * 1024 * 1024 + (int64_t)pDb->cfg.pages * pDb->cfg.pageSize * 1024;
+  if (pDb->cfg.cacheLastRow > 0) {
+    vgroupMemroy += (int64_t)pDb->cfg.lastRowMem * 1024 * 1024;
+  }
+
+  if (pDbInput == NULL) {
+    mndReleaseDb(pMnode, pDb);
+  }
+  return vgroupMemroy;
+}
+
+static bool mndGetVnodeMemroyFp(SMnode *pMnode, void *pObj, void *p1, void *p2, void *p3) {
+  SVgObj  *pVgroup = pObj;
+  int32_t  dnodeId = *(int32_t *)p1;
+  int64_t *pVnodeMemory = (int64_t *)p2;
+
+  for (int32_t v = 0; v < pVgroup->replica; ++v) {
+    if (pVgroup->vnodeGid[v].dnodeId == dnodeId) {
+      *pVnodeMemory += mndGetVgroupMemory(pMnode, NULL, pVgroup);
+    }
+  }
+
+  return true;
+}
+
+int64_t mndGetVnodesMemory(SMnode *pMnode, int32_t dnodeId) {
+  int64_t vnodeMemory = 0;
+  sdbTraverse(pMnode->pSdb, SDB_VGROUP, mndGetVnodeMemroyFp, &dnodeId, &vnodeMemory, NULL);
+  return vnodeMemory;
+}
+
 static int32_t mndRetrieveVnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
   SMnode *pMnode = pReq->info.node;
   SSdb   *pSdb = pMnode->pSdb;
@@ -807,9 +849,20 @@ int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray) {
       return -1;
     }
 
+    int64_t vgMem = mndGetVgroupMemory(pMnode, NULL, pVgroup);
+    if (pDnode->memAvail - vgMem - pDnode->memUsed <= 0) {
+      mError("db:%s, vgId:%d, no enough memory:%" PRId64 " in dnode:%d avail:%" PRId64 " used:%" PRId64,
+             pVgroup->dbName, pVgroup->vgId, vgMem, pDnode->id, pDnode->memAvail, pDnode->memUsed);
+      terrno = TSDB_CODE_MND_NO_ENOUGH_MEM_IN_DNODE;
+      return -1;
+    } else {
+      pDnode->memUsed += vgMem;
+    }
+
     pVgid->dnodeId = pDnode->id;
     pVgid->role = TAOS_SYNC_STATE_ERROR;
-    mInfo("db:%s, vgId:%d, vn:%d dnode:%d, is added", pVgroup->dbName, pVgroup->vgId, pVgroup->replica, pVgid->dnodeId);
+    mInfo("db:%s, vgId:%d, vn:%d is added, memory:%" PRId64 ", dnode:%d avail:%" PRId64 " used:%" PRId64,
+          pVgroup->dbName, pVgroup->vgId, pVgroup->replica, vgMem, pVgid->dnodeId, pDnode->memAvail, pDnode->memUsed);
 
     pVgroup->replica++;
     pDnode->numOfVnodes++;
@@ -835,7 +888,10 @@ int32_t mndRemoveVnodeFromVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray
     for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
       SVnodeGid *pVgid = &pVgroup->vnodeGid[vn];
       if (pVgid->dnodeId == pDnode->id) {
-        mInfo("db:%s, vgId:%d, vn:%d dnode:%d, is removed", pVgroup->dbName, pVgroup->vgId, vn, pVgid->dnodeId);
+        int64_t vgMem = mndGetVgroupMemory(pMnode, NULL, pVgroup);
+        pDnode->memUsed -= vgMem;
+        mInfo("db:%s, vgId:%d, vn:%d is removed, memory:%" PRId64 ", dnode:%d avail:%" PRId64 " used:%" PRId64,
+              pVgroup->dbName, pVgroup->vgId, vn, vgMem, pVgid->dnodeId, pDnode->memAvail, pDnode->memUsed);
         pDnode->numOfVnodes--;
         pVgroup->replica--;
         *pDelVgid = *pVgid;
@@ -1161,6 +1217,17 @@ static int32_t mndRedistributeVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb,
       terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
       goto _OVER;
     }
+
+    int64_t vgMem = mndGetVgroupMemory(pMnode, NULL, pVgroup);
+    if (pNew1->memAvail - vgMem - pNew1->memUsed <= 0) {
+      mError("db:%s, vgId:%d, no enough memory:%" PRId64 " in dnode:%d avail:%" PRId64 " used:%" PRId64,
+             pVgroup->dbName, pVgroup->vgId, vgMem, pNew1->id, pNew1->memAvail, pNew1->memUsed);
+      terrno = TSDB_CODE_MND_NO_ENOUGH_MEM_IN_DNODE;
+      return -1;
+    } else {
+      pNew1->memUsed += vgMem;
+    }
+
     if (mndAddIncVgroupReplicaToTrans(pMnode, pTrans, pDb, &newVg, pNew1->id) != 0) goto _OVER;
     if (mndAddDecVgroupReplicaFromTrans(pMnode, pTrans, pDb, &newVg, pOld1->id) != 0) goto _OVER;
   }
@@ -1173,6 +1240,15 @@ static int32_t mndRedistributeVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb,
       terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
       goto _OVER;
     }
+    int64_t vgMem = mndGetVgroupMemory(pMnode, NULL, pVgroup);
+    if (pNew2->memAvail - vgMem - pNew2->memUsed <= 0) {
+      mError("db:%s, vgId:%d, no enough memory:%" PRId64 " in dnode:%d avail:%" PRId64 " used:%" PRId64,
+             pVgroup->dbName, pVgroup->vgId, vgMem, pNew2->id, pNew2->memAvail, pNew2->memUsed);
+      terrno = TSDB_CODE_MND_NO_ENOUGH_MEM_IN_DNODE;
+      return -1;
+    } else {
+      pNew2->memUsed += vgMem;
+    }
     if (mndAddIncVgroupReplicaToTrans(pMnode, pTrans, pDb, &newVg, pNew2->id) != 0) goto _OVER;
     if (mndAddDecVgroupReplicaFromTrans(pMnode, pTrans, pDb, &newVg, pOld2->id) != 0) goto _OVER;
   }
@@ -1185,6 +1261,15 @@ static int32_t mndRedistributeVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb,
       terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
       goto _OVER;
     }
+    int64_t vgMem = mndGetVgroupMemory(pMnode, NULL, pVgroup);
+    if (pNew3->memAvail - vgMem - pNew3->memUsed <= 0) {
+      mError("db:%s, vgId:%d, no enough memory:%" PRId64 " in dnode:%d avail:%" PRId64 " used:%" PRId64,
+             pVgroup->dbName, pVgroup->vgId, vgMem, pNew3->id, pNew3->memAvail, pNew3->memUsed);
+      terrno = TSDB_CODE_MND_NO_ENOUGH_MEM_IN_DNODE;
+      return -1;
+    } else {
+      pNew3->memUsed += vgMem;
+    }
     if (mndAddIncVgroupReplicaToTrans(pMnode, pTrans, pDb, &newVg, pNew3->id) != 0) goto _OVER;
     if (mndAddDecVgroupReplicaFromTrans(pMnode, pTrans, pDb, &newVg, pOld3->id) != 0) goto _OVER;
   }

source/libs/parser/src/parTranslater.c

@@ -3138,6 +3138,7 @@ static void buildAlterDbReq(STranslateContext* pCxt, SAlterDatabaseStmt* pStmt,
   pReq->buffer = pStmt->pOptions->buffer;
   pReq->pageSize = -1;
   pReq->pages = pStmt->pOptions->pages;
+  pReq->lastRowMem = -1;
   pReq->daysPerFile = -1;
   pReq->daysToKeep0 = pStmt->pOptions->keep[0];
   pReq->daysToKeep1 = pStmt->pOptions->keep[1];

source/libs/tfs/src/tfs.c

@@ -291,7 +291,7 @@ int32_t tfsRmdir(STfs *pTfs, const char *rname) {
     for (int32_t id = 0; id < pTier->ndisk; id++) {
       STfsDisk *pDisk = pTier->disks[id];
       snprintf(aname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, rname);
-      uInfo("====> tfs remove dir : path:%s aname:%s rname:[%s]", pDisk->path, aname, rname);
+      uInfo("tfs remove dir : path:%s aname:%s rname:[%s]", pDisk->path, aname, rname);
       taosRemoveDir(aname);
     }
   }

source/libs/wal/src/walWrite.c

@@ -332,21 +332,25 @@ int32_t walWriteWithSyncInfo(SWal *pWal, int64_t index, tmsg_t msgType, SSyncLog
     terrno = TSDB_CODE_WAL_SIZE_LIMIT;
     return -1;
   }
+  taosThreadMutexLock(&pWal->mutex);
 
   if (index == pWal->vers.lastVer + 1) {
     if (taosArrayGetSize(pWal->fileInfoSet) == 0) {
       pWal->vers.firstVer = index;
       if (walRoll(pWal) < 0) {
+        taosThreadMutexUnlock(&pWal->mutex);
         return -1;
       }
     } else {
       int64_t passed = walGetSeq() - pWal->lastRollSeq;
       if (pWal->cfg.rollPeriod != -1 && pWal->cfg.rollPeriod != 0 && passed > pWal->cfg.rollPeriod) {
         if (walRoll(pWal) < 0) {
+          taosThreadMutexUnlock(&pWal->mutex);
           return -1;
         }
       } else if (pWal->cfg.segSize != -1 && pWal->cfg.segSize != 0 && walGetLastFileSize(pWal) > pWal->cfg.segSize) {
         if (walRoll(pWal) < 0) {
+          taosThreadMutexUnlock(&pWal->mutex);
           return -1;
         }
       }
@@ -355,6 +359,7 @@ int32_t walWriteWithSyncInfo(SWal *pWal, int64_t index, tmsg_t msgType, SSyncLog
     // reject skip log or rewrite log
     // must truncate explicitly first
     terrno = TSDB_CODE_WAL_INVALID_VER;
+    taosThreadMutexUnlock(&pWal->mutex);
     return -1;
   }
 
@@ -362,8 +367,6 @@ int32_t walWriteWithSyncInfo(SWal *pWal, int64_t index, tmsg_t msgType, SSyncLog
 
   ASSERT(pWal->writeCur >= 0);
 
-  taosThreadMutexLock(&pWal->mutex);
-
   if (pWal->pWriteIdxTFile == NULL || pWal->pWriteLogTFile == NULL) {
     walSetWrite(pWal);
     taosLSeekFile(pWal->pWriteLogTFile, 0, SEEK_END);

source/util/src/terror.c

@@ -177,6 +177,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_DNODE_ALREADY_EXIST,      "Dnode already exists"
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_DNODE_NOT_EXIST,          "Dnode does not exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_DNODES,          "Too many dnodes")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_NO_ENOUGH_DNODES,         "Out of dnodes")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_NO_ENOUGH_MEM_IN_DNODE,   "No enough memory in dnode")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_CLUSTER_CFG,      "Cluster cfg inconsistent")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_CLUSTER_ID,       "Cluster id not match")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_DNODE_CFG,        "Invalid dnode cfg")

source/util/src/ttimer.c

@@ -18,6 +18,7 @@
 #include "taoserror.h"
 #include "tlog.h"
 #include "tsched.h"
+#include "tdef.h"
 
 #define tmrFatal(...)                                                     \
   {                                                                       \
@@ -110,7 +111,7 @@ typedef struct time_wheel_t {
   tmr_obj_t**     slots;
 } time_wheel_t;
 
-static int32_t tsMaxTmrCtrl = 512;
+static int32_t tsMaxTmrCtrl = TSDB_MAX_VNODES_PER_DB + 100;
 
 static TdThreadOnce  tmrModuleInit = PTHREAD_ONCE_INIT;
 static TdThreadMutex tmrCtrlMutex;