#5044

Committed 06 May 2026 02:35AM UTC coverage: 73.169% (+0.06%) from 73.107%

Build # #5044

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat: [6659794715] cpu limit (#35153)

Coverage Stats

244 of 275 new or added lines in 23 files covered. (88.73%)

526 existing lines in 141 files now uncovered.

277745 of 379596 relevant lines covered (73.17%)

133740972.66 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.47

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

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

#define _DEFAULT_SOURCE
#include "mndDb.h"
#include "audit.h"
#include "command.h"
#include "mndArbGroup.h"
#include "mndCluster.h"
#include "mndCompact.h"
#include "mndCompactDetail.h"
#include "mndConfig.h"
#include "mndDnode.h"
#include "mndEncryptAlgr.h"
#include "mndIndex.h"
#include "mndPrivilege.h"
#include "mndRetention.h"
#include "mndRetentionDetail.h"
#include "mndRsma.h"
#include "mndScan.h"
#include "mndShow.h"
#include "mndSma.h"
#include "mndSsMigrate.h"
#include "mndStb.h"
#include "mndStream.h"
#include "mndSubscribe.h"
#include "mndTopic.h"
#include "mndTrans.h"
#include "mndUser.h"
#include "mndVgroup.h"
#include "mndView.h"
#include "systable.h"
#include "thttp.h"
#include "tjson.h"

#define DB_VER_INITIAL                   1
#define DB_VER_SUPPORT_ADVANCED_SECURITY 2
#define DB_VER_SUPPORT_MAC               3
#define DB_VER_NUMBER                    DB_VER_SUPPORT_MAC

#define DB_RESERVE_SIZE 8

static SSdbRow *mndDbActionDecode(SSdbRaw *pRaw);
static int32_t  mndDbActionInsert(SSdb *pSdb, SDbObj *pDb);
static int32_t  mndDbActionDelete(SSdb *pSdb, SDbObj *pDb);
static int32_t  mndDbActionUpdate(SSdb *pSdb, SDbObj *pOld, SDbObj *pNew);
static int32_t  mndNewDbActionValidate(SMnode *pMnode, STrans *pTrans, SSdbRaw *pRaw);

static int32_t mndProcessCreateDbReq(SRpcMsg *pReq);
static int32_t mndProcessAlterDbReq(SRpcMsg *pReq);
static int32_t mndProcessDropDbReq(SRpcMsg *pReq);
static int32_t mndProcessUseDbReq(SRpcMsg *pReq);
static int32_t mndProcessTrimDbReq(SRpcMsg *pReq);
static int32_t mndProcessSsMigrateDbReq(SRpcMsg *pReq);
static int32_t mndProcessTrimDbWalReq(SRpcMsg *pReq);
static int32_t mndProcessS3MigrateDbReq(SRpcMsg *pReq);
static int32_t mndRetrieveDbs(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity);
static void    mndCancelGetNextDb(SMnode *pMnode, void *pIter);
static int32_t mndProcessGetDbCfgReq(SRpcMsg *pReq);

#ifndef TD_ENTERPRISE
int32_t mndProcessCompactDbReq(SRpcMsg *pReq) { return TSDB_CODE_OPS_NOT_SUPPORT; }
#endif

int32_t mndInitDb(SMnode *pMnode) {
  SSdbTable table = {
      .sdbType = SDB_DB,
      .keyType = SDB_KEY_BINARY,
      .encodeFp = (SdbEncodeFp)mndDbActionEncode,
      .decodeFp = (SdbDecodeFp)mndDbActionDecode,
      .insertFp = (SdbInsertFp)mndDbActionInsert,
      .updateFp = (SdbUpdateFp)mndDbActionUpdate,
      .deleteFp = (SdbDeleteFp)mndDbActionDelete,
      .validateFp = (SdbValidateFp)mndNewDbActionValidate,
  };

  mndSetMsgHandle(pMnode, TDMT_MND_CREATE_DB, mndProcessCreateDbReq);
  mndSetMsgHandle(pMnode, TDMT_MND_ALTER_DB, mndProcessAlterDbReq);
  mndSetMsgHandle(pMnode, TDMT_MND_DROP_DB, mndProcessDropDbReq);
  mndSetMsgHandle(pMnode, TDMT_MND_USE_DB, mndProcessUseDbReq);
  mndSetMsgHandle(pMnode, TDMT_MND_COMPACT_DB, mndProcessCompactDbReq);
  mndSetMsgHandle(pMnode, TDMT_MND_SCAN_DB, mndProcessScanDbReq);
  mndSetMsgHandle(pMnode, TDMT_MND_TRIM_DB, mndProcessTrimDbReq);
  mndSetMsgHandle(pMnode, TDMT_MND_TRIM_DB_WAL, mndProcessTrimDbWalReq);
  mndSetMsgHandle(pMnode, TDMT_MND_GET_DB_CFG, mndProcessGetDbCfgReq);
  mndSetMsgHandle(pMnode, TDMT_MND_SSMIGRATE_DB, mndProcessSsMigrateDbReq);
  mndSetMsgHandle(pMnode, TDMT_MND_GET_DB_INFO, mndProcessUseDbReq);

  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_DB, mndRetrieveDbs);
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_DB, mndCancelGetNextDb);

  return sdbSetTable(pMnode->pSdb, table);
}

void mndCleanupDb(SMnode *pMnode) {}

SSdbRaw *mndDbActionEncode(SDbObj *pDb) {
  int32_t code = 0;
  int32_t lino = 0;
  terrno = TSDB_CODE_OUT_OF_MEMORY;

  int32_t  size = sizeof(SDbObj) + pDb->cfg.numOfRetensions * sizeof(SRetention) + DB_RESERVE_SIZE;
  SSdbRaw *pRaw = sdbAllocRaw(SDB_DB, DB_VER_NUMBER, size);
  if (pRaw == NULL) goto _OVER;

  int32_t dataPos = 0;
  SDB_SET_BINARY(pRaw, dataPos, pDb->name, TSDB_DB_FNAME_LEN, _OVER)
  SDB_SET_BINARY(pRaw, dataPos, pDb->acct, TSDB_USER_LEN, _OVER)
  SDB_SET_BINARY(pRaw, dataPos, pDb->createUser, TSDB_USER_LEN, _OVER)
  SDB_SET_INT64(pRaw, dataPos, pDb->createdTime, _OVER)
  SDB_SET_INT64(pRaw, dataPos, pDb->updateTime, _OVER)
  SDB_SET_INT64(pRaw, dataPos, pDb->uid, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfgVersion, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->vgVersion, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.numOfVgroups, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.numOfStables, _OVER)
  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.cacheLastSize, _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)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.daysToKeep2, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.minRows, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.maxRows, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.walFsyncPeriod, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pDb->cfg.walLevel, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pDb->cfg.precision, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pDb->cfg.compression, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pDb->cfg.replications, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pDb->cfg.strict, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pDb->cfg.cacheLast, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pDb->cfg.hashMethod, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.numOfRetensions, _OVER)
  for (int32_t i = 0; i < pDb->cfg.numOfRetensions; ++i) {
    SRetention *pRetension = taosArrayGet(pDb->cfg.pRetensions, i);
    SDB_SET_INT64(pRaw, dataPos, pRetension->freq, _OVER)
    SDB_SET_INT64(pRaw, dataPos, pRetension->keep, _OVER)
    SDB_SET_INT8(pRaw, dataPos, pRetension->freqUnit, _OVER)
    SDB_SET_INT8(pRaw, dataPos, pRetension->keepUnit, _OVER)
  }
  SDB_SET_INT8(pRaw, dataPos, pDb->cfg.schemaless, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.walRetentionPeriod, _OVER)
  SDB_SET_INT64(pRaw, dataPos, pDb->cfg.walRetentionSize, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.walRollPeriod, _OVER)
  SDB_SET_INT64(pRaw, dataPos, pDb->cfg.walSegmentSize, _OVER)
  SDB_SET_INT16(pRaw, dataPos, pDb->cfg.sstTrigger, _OVER)
  SDB_SET_INT16(pRaw, dataPos, pDb->cfg.hashPrefix, _OVER)
  SDB_SET_INT16(pRaw, dataPos, pDb->cfg.hashSuffix, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.tsdbPageSize, _OVER)
  SDB_SET_INT64(pRaw, dataPos, pDb->compactStartTime, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.keepTimeOffset, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.ssChunkSize, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.ssKeepLocal, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pDb->cfg.ssCompact, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pDb->cfg.withArbitrator, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pDb->cfg.encryptAlgorithm, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->tsmaVersion, _OVER);
  SDB_SET_INT8(pRaw, dataPos, pDb->cfg.compactTimeOffset, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.compactStartTime, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.compactEndTime, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.compactInterval, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pDb->cfg.isAudit, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pDb->cfg.secureDelete, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pDb->cfg.cacheLastShardBits, _OVER)

  SDB_SET_RESERVE(pRaw, dataPos, DB_RESERVE_SIZE, _OVER)
  SDB_SET_UINT8(pRaw, dataPos, pDb->cfg.flags, _OVER)
  SDB_SET_INT64(pRaw, dataPos, pDb->ownerId, _OVER)

  SDB_SET_DATALEN(pRaw, dataPos, _OVER)

  terrno = 0;

_OVER:
  if (terrno != 0) {
    mError("db:%s, failed to encode to raw:%p since %s", pDb->name, pRaw, terrstr());
    sdbFreeRaw(pRaw);
    return NULL;
  }

  mTrace("db:%s, encode to raw:%p, row:%p", pDb->name, pRaw, pDb);
  return pRaw;
}

static SSdbRow *mndDbActionDecode(SSdbRaw *pRaw) {
  int32_t code = 0;
  int32_t lino = 0;
  terrno = TSDB_CODE_OUT_OF_MEMORY;
  SSdbRow *pRow = NULL;
  SDbObj  *pDb = NULL;

  int8_t sver = 0;
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto _OVER;

  if (sver < 1 || sver > DB_VER_NUMBER) {
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
    goto _OVER;
  }

  pRow = sdbAllocRow(sizeof(SDbObj));
  if (pRow == NULL) goto _OVER;

  pDb = sdbGetRowObj(pRow);
  if (pDb == NULL) goto _OVER;

  int32_t dataPos = 0;
  SDB_GET_BINARY(pRaw, dataPos, pDb->name, TSDB_DB_FNAME_LEN, _OVER)
  SDB_GET_BINARY(pRaw, dataPos, pDb->acct, TSDB_USER_LEN, _OVER)
  SDB_GET_BINARY(pRaw, dataPos, pDb->createUser, TSDB_USER_LEN, _OVER)
  SDB_GET_INT64(pRaw, dataPos, &pDb->createdTime, _OVER)
  SDB_GET_INT64(pRaw, dataPos, &pDb->updateTime, _OVER)
  SDB_GET_INT64(pRaw, dataPos, &pDb->uid, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfgVersion, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->vgVersion, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.numOfVgroups, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.numOfStables, _OVER)
  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.cacheLastSize, _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)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.daysToKeep2, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.minRows, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.maxRows, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.walFsyncPeriod, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.walLevel, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.precision, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.compression, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.replications, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.strict, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.cacheLast, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.hashMethod, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.numOfRetensions, _OVER)
  if (pDb->cfg.numOfRetensions > 0) {
    pDb->cfg.pRetensions = taosArrayInit(pDb->cfg.numOfRetensions, sizeof(SRetention));
    if (pDb->cfg.pRetensions == NULL) goto _OVER;
    for (int32_t i = 0; i < pDb->cfg.numOfRetensions; ++i) {
      SRetention retention = {0};
      SDB_GET_INT64(pRaw, dataPos, &retention.freq, _OVER)
      SDB_GET_INT64(pRaw, dataPos, &retention.keep, _OVER)
      SDB_GET_INT8(pRaw, dataPos, &retention.freqUnit, _OVER)
      SDB_GET_INT8(pRaw, dataPos, &retention.keepUnit, _OVER)
      if (taosArrayPush(pDb->cfg.pRetensions, &retention) == NULL) {
        goto _OVER;
      }
    }
  }
  SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.schemaless, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.walRetentionPeriod, _OVER)
  SDB_GET_INT64(pRaw, dataPos, &pDb->cfg.walRetentionSize, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.walRollPeriod, _OVER)
  SDB_GET_INT64(pRaw, dataPos, &pDb->cfg.walSegmentSize, _OVER)
  SDB_GET_INT16(pRaw, dataPos, &pDb->cfg.sstTrigger, _OVER)
  SDB_GET_INT16(pRaw, dataPos, &pDb->cfg.hashPrefix, _OVER)
  SDB_GET_INT16(pRaw, dataPos, &pDb->cfg.hashSuffix, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.tsdbPageSize, _OVER)
  SDB_GET_INT64(pRaw, dataPos, &pDb->compactStartTime, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.keepTimeOffset, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.ssChunkSize, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.ssKeepLocal, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.ssCompact, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.withArbitrator, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.encryptAlgorithm, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->tsmaVersion, _OVER);
  SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.compactTimeOffset, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.compactStartTime, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.compactEndTime, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.compactInterval, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.isAudit, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.secureDelete, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.cacheLastShardBits, _OVER)

  SDB_GET_RESERVE(pRaw, dataPos, DB_RESERVE_SIZE, _OVER)
  if (dataPos + sizeof(uint8_t) <= pRaw->dataLen) {
    SDB_GET_UINT8(pRaw, dataPos, &pDb->cfg.flags, _OVER)
  }
  if (dataPos + sizeof(int64_t) <= pRaw->dataLen) {
    SDB_GET_INT64(pRaw, dataPos, &pDb->ownerId, _OVER)
  }

  taosInitRWLatch(&pDb->lock);

  if (pDb->cfg.ssChunkSize == 0) {
    pDb->cfg.ssChunkSize = TSDB_DEFAULT_SS_CHUNK_SIZE;

    mInfo("db:%s, ssChunkSize set from %d to default %d", pDb->name, pDb->cfg.ssChunkSize, TSDB_DEFAULT_SS_CHUNK_SIZE);
  }

  if (pDb->cfg.ssKeepLocal == 0) {
    pDb->cfg.ssKeepLocal = TSDB_DEFAULT_SS_KEEP_LOCAL;

    mInfo("db:%s, ssKeepLocal set from %d to default %d", pDb->name, pDb->cfg.ssKeepLocal, TSDB_DEFAULT_SS_KEEP_LOCAL);
  }

  if (pDb->cfg.tsdbPageSize != TSDB_MIN_TSDB_PAGESIZE) {
    mInfo("db:%s, tsdbPageSize set from %d to default %d", pDb->name, pDb->cfg.tsdbPageSize,
          TSDB_DEFAULT_TSDB_PAGESIZE);
  }

  if (pDb->cfg.sstTrigger != TSDB_MIN_STT_TRIGGER) {
    mInfo("db:%s, sstTrigger set from %d to default %d", pDb->name, pDb->cfg.sstTrigger, TSDB_DEFAULT_SST_TRIGGER);
  }

  if (sver < DB_VER_SUPPORT_ADVANCED_SECURITY) {
    pDb->cfg.allowDrop = TSDB_DEFAULT_DB_ALLOW_DROP;
  }

  if (sver < DB_VER_SUPPORT_MAC) {
    pDb->cfg.securityLevel = TSDB_DEFAULT_SECURITY_LEVEL;
  }

  terrno = 0;

_OVER:
  if (terrno != 0) {
    mError("db:%s, failed to decode from raw:%p since %s", pDb == NULL ? "null" : pDb->name, pRaw, terrstr());
    taosMemoryFreeClear(pRow);
    return NULL;
  }

  mTrace("db:%s, decode from raw:%p, row:%p", pDb->name, pRaw, pDb);
  return pRow;
}

static int32_t mndNewDbActionValidate(SMnode *pMnode, STrans *pTrans, SSdbRaw *pRaw) {
  SSdb    *pSdb = pMnode->pSdb;
  SSdbRow *pRow = NULL;
  SDbObj  *pNewDb = NULL;
  int      code = -1;

  pRow = mndDbActionDecode(pRaw);
  if (pRow == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    goto _OVER;
  }
  pNewDb = sdbGetRowObj(pRow);
  if (pNewDb == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    goto _OVER;
  }

  SDbObj *pOldDb = sdbAcquire(pMnode->pSdb, SDB_DB, pNewDb->name);
  if (pOldDb != NULL) {
    mError("trans:%d, db name already in use. name: %s", pTrans->id, pNewDb->name);
    sdbRelease(pMnode->pSdb, pOldDb);
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    goto _OVER;
  }

  code = 0;
_OVER:
  if (pNewDb) mndDbActionDelete(pSdb, pNewDb);
  taosMemoryFreeClear(pRow);
  return code;
}

static int32_t mndDbActionInsert(SSdb *pSdb, SDbObj *pDb) {
  mTrace("db:%s, perform insert action, row:%p", pDb->name, pDb);
  return 0;
}

static int32_t mndDbActionDelete(SSdb *pSdb, SDbObj *pDb) {
  mTrace("db:%s, perform delete action, row:%p", pDb->name, pDb);
  taosArrayDestroy(pDb->cfg.pRetensions);
  return 0;
}

static int32_t mndDbActionUpdate(SSdb *pSdb, SDbObj *pOld, SDbObj *pNew) {
  mTrace("db:%s, perform update action, old row:%p new row:%p", pOld->name, pOld, pNew);
  taosWLockLatch(&pOld->lock);
  pOld->updateTime = pNew->updateTime;
  pOld->cfgVersion = pNew->cfgVersion;
  pOld->vgVersion = pNew->vgVersion;
  pOld->cfg.numOfVgroups = pNew->cfg.numOfVgroups;
  pOld->cfg.buffer = pNew->cfg.buffer;
  pOld->cfg.pageSize = pNew->cfg.pageSize;
  pOld->cfg.pages = pNew->cfg.pages;
  pOld->cfg.cacheLastSize = pNew->cfg.cacheLastSize;
  pOld->cfg.cacheLastShardBits = pNew->cfg.cacheLastShardBits;
  pOld->cfg.daysPerFile = pNew->cfg.daysPerFile;
  pOld->cfg.daysToKeep0 = pNew->cfg.daysToKeep0;
  pOld->cfg.daysToKeep1 = pNew->cfg.daysToKeep1;
  pOld->cfg.daysToKeep2 = pNew->cfg.daysToKeep2;
  pOld->cfg.keepTimeOffset = pNew->cfg.keepTimeOffset;
  pOld->cfg.walFsyncPeriod = pNew->cfg.walFsyncPeriod;
  pOld->cfg.walLevel = pNew->cfg.walLevel;
  pOld->cfg.walRetentionPeriod = pNew->cfg.walRetentionPeriod;
  pOld->cfg.walRetentionSize = pNew->cfg.walRetentionSize;
  pOld->cfg.strict = pNew->cfg.strict;
  pOld->cfg.cacheLast = pNew->cfg.cacheLast;
  pOld->cfg.replications = pNew->cfg.replications;
  pOld->cfg.sstTrigger = pNew->cfg.sstTrigger;
  pOld->cfg.minRows = pNew->cfg.minRows;
  pOld->cfg.maxRows = pNew->cfg.maxRows;
  pOld->cfg.tsdbPageSize = pNew->cfg.tsdbPageSize;
  pOld->cfg.ssChunkSize = pNew->cfg.ssChunkSize;
  pOld->cfg.ssKeepLocal = pNew->cfg.ssKeepLocal;
  pOld->cfg.ssCompact = pNew->cfg.ssCompact;
  pOld->cfg.withArbitrator = pNew->cfg.withArbitrator;
  pOld->cfg.compactInterval = pNew->cfg.compactInterval;
  pOld->cfg.compactStartTime = pNew->cfg.compactStartTime;
  pOld->cfg.compactEndTime = pNew->cfg.compactEndTime;
  pOld->cfg.compactTimeOffset = pNew->cfg.compactTimeOffset;
  pOld->compactStartTime = pNew->compactStartTime;
  pOld->tsmaVersion = pNew->tsmaVersion;
  pOld->cfg.isAudit = pNew->cfg.isAudit;
  pOld->cfg.secureDelete = pNew->cfg.secureDelete;
  pOld->cfg.flags = pNew->cfg.flags;
  pOld->ownerId = pNew->ownerId;
  taosWUnLockLatch(&pOld->lock);
  return 0;
}

static inline int32_t mndGetGlobalVgroupVersion(SMnode *pMnode) {
  SSdb *pSdb = pMnode->pSdb;
  return sdbGetTableVer(pSdb, SDB_VGROUP);
}

SDbObj *mndAcquireDb(SMnode *pMnode, const char *db) {
  SSdb   *pSdb = pMnode->pSdb;
  SDbObj *pDb = sdbAcquire(pSdb, SDB_DB, db);
  if (pDb == NULL) {
    if (terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
      terrno = TSDB_CODE_MND_DB_NOT_EXIST;
    } else if (terrno == TSDB_CODE_SDB_OBJ_CREATING) {
      terrno = TSDB_CODE_MND_DB_IN_CREATING;
    } else if (terrno == TSDB_CODE_SDB_OBJ_DROPPING) {
      terrno = TSDB_CODE_MND_DB_IN_DROPPING;
    } else {
      terrno = TSDB_CODE_APP_ERROR;
      mFatal("db:%s, failed to acquire db since %s", db, terrstr());
    }
  }
  return pDb;
}

SDbObj *mndAcquireAuditDb(SMnode *pMnode) {
  SSdb      *pSdb = pMnode->pSdb;
  SDbObj    *pDb = NULL;
  void      *pIter = NULL;
  ESdbStatus objStatus = 0;

  while (1) {
    pIter = sdbFetchAll(pSdb, SDB_DB, pIter, (void **)&pDb, &objStatus, true);
    if (pIter == NULL) break;
    if (pDb->cfg.isAudit == 1) {
      sdbCancelFetch(pSdb, pIter);
      break;
    }

    sdbRelease(pSdb, pDb);
  }
  return pDb;
}

void mndReleaseDb(SMnode *pMnode, SDbObj *pDb) {
  SSdb *pSdb = pMnode->pSdb;
  sdbRelease(pSdb, pDb);
}

bool mndDbIsExist(SMnode *pMnode, const char *db, int64_t uid) {
  SDbObj *pDb = mndAcquireDb(pMnode, db);
  if (pDb == NULL) {
    return false;
  } else {
    bool result = (uid == 0) || (uid == pDb->uid);
    mndReleaseDb(pMnode, pDb);
    pDb = NULL;
    return result;
  }
}

int32_t mndCheckDbName(const char *dbName, SUserObj *pUser) {
  char *pos = strstr(dbName, TS_PATH_DELIMITER);
  if (pos == NULL) {
    return TSDB_CODE_MND_INVALID_DB;
  }

  int32_t acctId;
  int32_t code = taosStr2int32(dbName, &acctId);
  if (code != 0) {
    return code;
  }

  if (acctId != pUser->acctId) {
    return TSDB_CODE_MND_INVALID_DB_ACCT;
  }

  return 0;
}

int32_t mndCheckDbCfg(SMnode *pMnode, SDbCfg *pCfg) {
  int32_t code = TSDB_CODE_MND_INVALID_DB_OPTION;

  if (pCfg->numOfVgroups < TSDB_MIN_VNODES_PER_DB || pCfg->numOfVgroups > TSDB_MAX_VNODES_PER_DB) return code;
  if (pCfg->numOfStables < TSDB_DB_STREAM_MODE_OFF || pCfg->numOfStables > TSDB_DB_STREAM_MODE_ON) return code;
  if (pCfg->buffer < TSDB_MIN_BUFFER_PER_VNODE || pCfg->buffer > TSDB_MAX_BUFFER_PER_VNODE) return code;
  if (pCfg->pageSize < TSDB_MIN_PAGESIZE_PER_VNODE || pCfg->pageSize > TSDB_MAX_PAGESIZE_PER_VNODE) return code;
  if (pCfg->pages < TSDB_MIN_PAGES_PER_VNODE || pCfg->pages > TSDB_MAX_PAGES_PER_VNODE) return code;
  if (pCfg->cacheLastSize < TSDB_MIN_DB_CACHE_SIZE || pCfg->cacheLastSize > TSDB_MAX_DB_CACHE_SIZE) return code;
  if (pCfg->cacheLastShardBits < TSDB_MIN_DB_CACHE_SHARD_BITS ||
      pCfg->cacheLastShardBits > TSDB_MAX_DB_CACHE_SHARD_BITS)
    return code;
  if (pCfg->daysPerFile < TSDB_MIN_DAYS_PER_FILE || pCfg->daysPerFile > TSDB_MAX_DAYS_PER_FILE) return code;
  if (pCfg->daysToKeep0 < TSDB_MIN_KEEP || pCfg->daysToKeep0 > TSDB_MAX_KEEP) return code;
  if (pCfg->daysToKeep1 < TSDB_MIN_KEEP || pCfg->daysToKeep1 > TSDB_MAX_KEEP) return code;
  if (pCfg->daysToKeep2 < TSDB_MIN_KEEP || pCfg->daysToKeep2 > TSDB_MAX_KEEP) return code;
  if (pCfg->daysToKeep0 < pCfg->daysPerFile) return code;
  if (pCfg->daysToKeep0 > pCfg->daysToKeep1) return code;
  if (pCfg->daysToKeep1 > pCfg->daysToKeep2) return code;
  if (pCfg->keepTimeOffset < TSDB_MIN_KEEP_TIME_OFFSET || pCfg->keepTimeOffset > TSDB_MAX_KEEP_TIME_OFFSET) return code;
  if (pCfg->minRows < TSDB_MIN_MINROWS_FBLOCK || pCfg->minRows > TSDB_MAX_MINROWS_FBLOCK) return code;
  if (pCfg->maxRows < TSDB_MIN_MAXROWS_FBLOCK || pCfg->maxRows > TSDB_MAX_MAXROWS_FBLOCK) return code;
  if (pCfg->minRows > pCfg->maxRows) return code;
  if (pCfg->walFsyncPeriod < TSDB_MIN_FSYNC_PERIOD || pCfg->walFsyncPeriod > TSDB_MAX_FSYNC_PERIOD) return code;
  if (pCfg->walLevel < TSDB_MIN_WAL_LEVEL || pCfg->walLevel > TSDB_MAX_WAL_LEVEL) return code;
  if (pCfg->precision < TSDB_MIN_PRECISION && pCfg->precision > TSDB_MAX_PRECISION) return code;
  if (pCfg->compression < TSDB_MIN_COMP_LEVEL || pCfg->compression > TSDB_MAX_COMP_LEVEL) return code;
  if (pCfg->replications < TSDB_MIN_DB_REPLICA || pCfg->replications > TSDB_MAX_DB_REPLICA) return code;
#ifdef TD_ENTERPRISE
  if ((pCfg->replications == 2) ^ (pCfg->withArbitrator == TSDB_MAX_DB_WITH_ARBITRATOR)) return code;
  // if (pCfg->encryptAlgorithm < TSDB_MIN_ENCRYPT_ALGO || pCfg->encryptAlgorithm > TSDB_MAX_ENCRYPT_ALGO) return code;
#else
  if (pCfg->replications != 1 && pCfg->replications != 3) return code;
  if (pCfg->encryptAlgorithm != TSDB_DEFAULT_ENCRYPT_ALGO) return code;
#endif

  if (pCfg->strict < TSDB_DB_STRICT_OFF || pCfg->strict > TSDB_DB_STRICT_ON) return code;
  if (pCfg->schemaless < TSDB_DB_SCHEMALESS_OFF || pCfg->schemaless > TSDB_DB_SCHEMALESS_ON) return code;
  if (pCfg->cacheLast < TSDB_CACHE_MODEL_NONE || pCfg->cacheLast > TSDB_CACHE_MODEL_BOTH) return code;
  if (pCfg->hashMethod != 1) return code;
  if (pCfg->replications > mndGetDnodeSize(pMnode)) {
    code = TSDB_CODE_MND_NO_ENOUGH_DNODES;
    TAOS_RETURN(code);
  }
  if (pCfg->walRetentionPeriod < TSDB_DB_MIN_WAL_RETENTION_PERIOD) return code;
  if (pCfg->walRetentionSize < TSDB_DB_MIN_WAL_RETENTION_SIZE) return code;
  if (pCfg->walRollPeriod < TSDB_DB_MIN_WAL_ROLL_PERIOD) return code;
  if (pCfg->walSegmentSize < TSDB_DB_MIN_WAL_SEGMENT_SIZE) return code;
  if (pCfg->sstTrigger < TSDB_MIN_STT_TRIGGER || pCfg->sstTrigger > TSDB_MAX_STT_TRIGGER) return code;
  if (pCfg->hashPrefix < TSDB_MIN_HASH_PREFIX || pCfg->hashPrefix > TSDB_MAX_HASH_PREFIX) return code;
  if (pCfg->hashSuffix < TSDB_MIN_HASH_SUFFIX || pCfg->hashSuffix > TSDB_MAX_HASH_SUFFIX) return code;
  if ((pCfg->hashSuffix * pCfg->hashPrefix) < 0) return code;
  if ((pCfg->hashPrefix + pCfg->hashSuffix) >= (TSDB_TABLE_NAME_LEN - 1)) return code;
  if (pCfg->tsdbPageSize < TSDB_MIN_TSDB_PAGESIZE || pCfg->tsdbPageSize > TSDB_MAX_TSDB_PAGESIZE) return code;
  if (taosArrayGetSize(pCfg->pRetensions) != pCfg->numOfRetensions) return code;

  if (pCfg->ssChunkSize < TSDB_MIN_SS_CHUNK_SIZE || pCfg->ssChunkSize > TSDB_MAX_SS_CHUNK_SIZE) return code;
  if (pCfg->ssKeepLocal < TSDB_MIN_SS_KEEP_LOCAL || pCfg->ssKeepLocal > TSDB_MAX_SS_KEEP_LOCAL) return code;
  if (pCfg->ssCompact < TSDB_MIN_SS_COMPACT || pCfg->ssCompact > TSDB_MAX_SS_COMPACT) return code;

  if (pCfg->compactInterval != 0 &&
      (pCfg->compactInterval < TSDB_MIN_COMPACT_INTERVAL || pCfg->compactInterval > pCfg->daysToKeep2))
    return code;
  if (pCfg->compactStartTime != 0 &&
      (pCfg->compactStartTime < -pCfg->daysToKeep2 || pCfg->compactStartTime > -pCfg->daysPerFile))
    return code;
  if (pCfg->compactEndTime != 0 &&
      (pCfg->compactEndTime < -pCfg->daysToKeep2 || pCfg->compactEndTime > -pCfg->daysPerFile))
    return code;
  if (pCfg->compactStartTime != 0 && pCfg->compactEndTime != 0 && pCfg->compactStartTime >= pCfg->compactEndTime)
    return code;
  if (pCfg->compactTimeOffset < TSDB_MIN_COMPACT_TIME_OFFSET || pCfg->compactTimeOffset > TSDB_MAX_COMPACT_TIME_OFFSET)
    return code;
  if (pCfg->isAudit < 0 || pCfg->isAudit > 1) return code;
  if (pCfg->allowDrop < TSDB_MIN_DB_ALLOW_DROP || pCfg->allowDrop > TSDB_MAX_DB_ALLOW_DROP) return code;
  if (pCfg->securityLevel < TSDB_MIN_SECURITY_LEVEL || pCfg->securityLevel > TSDB_MAX_SECURITY_LEVEL) return code;

  code = 0;
  TAOS_RETURN(code);
}

static int32_t mndCheckInChangeDbCfg(SMnode *pMnode, SDbCfg *pOldCfg, SDbCfg *pNewCfg) {
  int32_t code = TSDB_CODE_MND_INVALID_DB_OPTION;
  if (pNewCfg->buffer < TSDB_MIN_BUFFER_PER_VNODE || pNewCfg->buffer > TSDB_MAX_BUFFER_PER_VNODE) return code;
  if (pNewCfg->pages < TSDB_MIN_PAGES_PER_VNODE || pNewCfg->pages > TSDB_MAX_PAGES_PER_VNODE) return code;
  if (pNewCfg->pageSize < TSDB_MIN_PAGESIZE_PER_VNODE || pNewCfg->pageSize > TSDB_MAX_PAGESIZE_PER_VNODE) return code;
  if (pNewCfg->daysPerFile < TSDB_MIN_DAYS_PER_FILE || pNewCfg->daysPerFile > TSDB_MAX_DAYS_PER_FILE) return code;
  if (pNewCfg->daysToKeep0 < TSDB_MIN_KEEP || pNewCfg->daysToKeep0 > TSDB_MAX_KEEP) return code;
  if (pNewCfg->daysToKeep1 < TSDB_MIN_KEEP || pNewCfg->daysToKeep1 > TSDB_MAX_KEEP) return code;
  if (pNewCfg->daysToKeep2 < TSDB_MIN_KEEP || pNewCfg->daysToKeep2 > TSDB_MAX_KEEP) return code;
  if (pNewCfg->daysToKeep0 < pNewCfg->daysPerFile) return code;
  if (pNewCfg->daysToKeep0 > pNewCfg->daysToKeep1) return code;
  if (pNewCfg->daysToKeep1 > pNewCfg->daysToKeep2) return code;
  if (pNewCfg->keepTimeOffset < TSDB_MIN_KEEP_TIME_OFFSET || pNewCfg->keepTimeOffset > TSDB_MAX_KEEP_TIME_OFFSET)
    return code;
  if (pNewCfg->walFsyncPeriod < TSDB_MIN_FSYNC_PERIOD || pNewCfg->walFsyncPeriod > TSDB_MAX_FSYNC_PERIOD) return code;
  if (pNewCfg->walLevel < TSDB_MIN_WAL_LEVEL || pNewCfg->walLevel > TSDB_MAX_WAL_LEVEL) return code;
  if (pNewCfg->cacheLast < TSDB_CACHE_MODEL_NONE || pNewCfg->cacheLast > TSDB_CACHE_MODEL_BOTH) return code;
  if (pNewCfg->cacheLastSize < TSDB_MIN_DB_CACHE_SIZE || pNewCfg->cacheLastSize > TSDB_MAX_DB_CACHE_SIZE) return code;
  if (pNewCfg->cacheLastShardBits < TSDB_MIN_DB_CACHE_SHARD_BITS ||
      pNewCfg->cacheLastShardBits > TSDB_MAX_DB_CACHE_SHARD_BITS)
    return code;
  if (pNewCfg->replications < TSDB_MIN_DB_REPLICA || pNewCfg->replications > TSDB_MAX_DB_REPLICA) return code;
#ifdef TD_ENTERPRISE
  if ((pNewCfg->replications == 2) ^ (pNewCfg->withArbitrator == TSDB_MAX_DB_WITH_ARBITRATOR)) return code;
  if (pNewCfg->replications == 2 && pNewCfg->withArbitrator == TSDB_MAX_DB_WITH_ARBITRATOR) {
    if (pOldCfg->replications != 1 && pOldCfg->replications != 2) {
      terrno = TSDB_CODE_OPS_NOT_SUPPORT;
      return code;
    }
  }
  if (pNewCfg->replications != 2 && pOldCfg->replications == 2) {
    terrno = TSDB_CODE_OPS_NOT_SUPPORT;
    return code;
  }
#else
  if (pNewCfg->replications != 1 && pNewCfg->replications != 3) return code;
#endif

  if (pNewCfg->walLevel == 0 && pOldCfg->replications > 1) {
    terrno = TSDB_CODE_MND_INVALID_WAL_LEVEL;
    return code;
  }
  if (pNewCfg->replications > 1 && pOldCfg->walLevel == 0) {
    terrno = TSDB_CODE_MND_INVALID_WAL_LEVEL;
    return code;
  }

  if (pNewCfg->sstTrigger != pOldCfg->sstTrigger &&
      (pNewCfg->sstTrigger < TSDB_MIN_STT_TRIGGER || pNewCfg->sstTrigger > TSDB_MAX_STT_TRIGGER))
    return code;
  if (pNewCfg->minRows < TSDB_MIN_MINROWS_FBLOCK || pNewCfg->minRows > TSDB_MAX_MINROWS_FBLOCK) return code;
  if (pNewCfg->maxRows < TSDB_MIN_MAXROWS_FBLOCK || pNewCfg->maxRows > TSDB_MAX_MAXROWS_FBLOCK) return code;
  if (pNewCfg->minRows > pNewCfg->maxRows) return code;
  if (pNewCfg->walRetentionPeriod < TSDB_DB_MIN_WAL_RETENTION_PERIOD) return code;
  if (pNewCfg->walRetentionSize < TSDB_DB_MIN_WAL_RETENTION_SIZE) return code;
  if (pNewCfg->strict < TSDB_DB_STRICT_OFF || pNewCfg->strict > TSDB_DB_STRICT_ON) return code;
  if (pNewCfg->replications > mndGetDnodeSize(pMnode)) {
    terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
    return code;
  }
  if (pNewCfg->ssChunkSize < TSDB_MIN_SS_CHUNK_SIZE || pNewCfg->ssChunkSize > TSDB_MAX_SS_CHUNK_SIZE) return code;
  if (pNewCfg->ssKeepLocal < TSDB_MIN_SS_KEEP_LOCAL || pNewCfg->ssKeepLocal > TSDB_MAX_SS_KEEP_LOCAL) return code;
  if (pNewCfg->ssCompact < TSDB_MIN_SS_COMPACT || pNewCfg->ssCompact > TSDB_MAX_SS_COMPACT) return code;

  if (pNewCfg->compactInterval != 0 &&
      (pNewCfg->compactInterval < TSDB_MIN_COMPACT_INTERVAL || pNewCfg->compactInterval > pNewCfg->daysToKeep2))
    return code;
  if (pNewCfg->compactStartTime != 0 &&
      (pNewCfg->compactStartTime < -pNewCfg->daysToKeep2 || pNewCfg->compactStartTime > -pNewCfg->daysPerFile))
    return code;
  if (pNewCfg->compactEndTime != 0 &&
      (pNewCfg->compactEndTime < -pNewCfg->daysToKeep2 || pNewCfg->compactEndTime > -pNewCfg->daysPerFile))
    return code;
  if (pNewCfg->compactStartTime != 0 && pNewCfg->compactEndTime != 0 &&
      pNewCfg->compactStartTime >= pNewCfg->compactEndTime)
    return code;
  if (pNewCfg->compactTimeOffset < TSDB_MIN_COMPACT_TIME_OFFSET ||
      pNewCfg->compactTimeOffset > TSDB_MAX_COMPACT_TIME_OFFSET)
    return code;
  if (pNewCfg->isAudit < 0 || pNewCfg->isAudit > 1) return code;
  if (pNewCfg->allowDrop < TSDB_MIN_DB_ALLOW_DROP || pNewCfg->allowDrop > TSDB_MAX_DB_ALLOW_DROP) return code;
  if (pNewCfg->securityLevel < TSDB_MIN_SECURITY_LEVEL || pNewCfg->securityLevel > TSDB_MAX_SECURITY_LEVEL) return code;

  code = 0;
  TAOS_RETURN(code);
}

static void mndSetDefaultDbCfg(SDbCfg *pCfg) {
  if (pCfg->numOfVgroups < 0) pCfg->numOfVgroups = TSDB_DEFAULT_VN_PER_DB;
  if (pCfg->numOfStables < 0) pCfg->numOfStables = TSDB_DEFAULT_DB_SINGLE_STABLE;
  if (pCfg->buffer < 0) pCfg->buffer = TSDB_DEFAULT_BUFFER_PER_VNODE;
  if (pCfg->pageSize < 0) pCfg->pageSize = TSDB_DEFAULT_PAGESIZE_PER_VNODE;
  if (pCfg->pages < 0) pCfg->pages = TSDB_DEFAULT_PAGES_PER_VNODE;
  if (pCfg->daysPerFile < 0) pCfg->daysPerFile = TSDB_DEFAULT_DURATION_PER_FILE;
  if (pCfg->daysToKeep0 < 0) pCfg->daysToKeep0 = TSDB_DEFAULT_KEEP;
  if (pCfg->daysToKeep1 < 0) pCfg->daysToKeep1 = pCfg->daysToKeep0;
  if (pCfg->daysToKeep2 < 0) pCfg->daysToKeep2 = pCfg->daysToKeep1;
  if (pCfg->keepTimeOffset < 0) pCfg->keepTimeOffset = TSDB_DEFAULT_KEEP_TIME_OFFSET;
  if (pCfg->minRows < 0) pCfg->minRows = TSDB_DEFAULT_MINROWS_FBLOCK;
  if (pCfg->maxRows < 0) pCfg->maxRows = TSDB_DEFAULT_MAXROWS_FBLOCK;
  if (pCfg->walFsyncPeriod < 0) pCfg->walFsyncPeriod = TSDB_DEFAULT_FSYNC_PERIOD;
  if (pCfg->walLevel < 0) pCfg->walLevel = TSDB_DEFAULT_WAL_LEVEL;
  if (pCfg->precision < 0) pCfg->precision = TSDB_DEFAULT_PRECISION;
  if (pCfg->compression < 0) pCfg->compression = TSDB_DEFAULT_COMP_LEVEL;
  if (pCfg->replications < 0) pCfg->replications = TSDB_DEFAULT_DB_REPLICA;
  if (pCfg->strict < 0) pCfg->strict = TSDB_DEFAULT_DB_STRICT;
  if (pCfg->cacheLast < 0) pCfg->cacheLast = TSDB_DEFAULT_CACHE_MODEL;
  if (pCfg->cacheLastSize <= 0) pCfg->cacheLastSize = TSDB_DEFAULT_CACHE_SIZE;
  if (pCfg->numOfRetensions < 0) pCfg->numOfRetensions = 0;
  if (pCfg->schemaless < 0) pCfg->schemaless = TSDB_DB_SCHEMALESS_OFF;
  if (pCfg->walRetentionPeriod < 0 && pCfg->walRetentionPeriod != -1)
    pCfg->walRetentionPeriod = TSDB_REPS_DEF_DB_WAL_RET_PERIOD;
  if (pCfg->walRetentionSize < 0 && pCfg->walRetentionSize != -1)
    pCfg->walRetentionSize = TSDB_REPS_DEF_DB_WAL_RET_SIZE;
  if (pCfg->walRollPeriod < 0) pCfg->walRollPeriod = TSDB_REPS_DEF_DB_WAL_ROLL_PERIOD;
  if (pCfg->walSegmentSize < 0) pCfg->walSegmentSize = TSDB_DEFAULT_DB_WAL_SEGMENT_SIZE;
  if (pCfg->sstTrigger <= 0) pCfg->sstTrigger = TSDB_DEFAULT_SST_TRIGGER;
  if (pCfg->tsdbPageSize <= 0) pCfg->tsdbPageSize = TSDB_DEFAULT_TSDB_PAGESIZE;
  if (pCfg->ssChunkSize <= 0) pCfg->ssChunkSize = TSDB_DEFAULT_SS_CHUNK_SIZE;
  if (pCfg->ssKeepLocal <= 0) pCfg->ssKeepLocal = TSDB_DEFAULT_SS_KEEP_LOCAL;
  if (pCfg->ssCompact < 0) pCfg->ssCompact = TSDB_DEFAULT_SS_COMPACT;
  if (pCfg->withArbitrator < 0) pCfg->withArbitrator = TSDB_DEFAULT_DB_WITH_ARBITRATOR;
  if (pCfg->encryptAlgorithm < 0) pCfg->encryptAlgorithm = TSDB_DEFAULT_ENCRYPT_ALGO;
}

int32_t mndSetCreateDbPrepareAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
  SSdbRaw *pDbRaw = mndDbActionEncode(pDb);
  if (pDbRaw == NULL) return -1;

  if (mndTransAppendPrepareLog(pTrans, pDbRaw) != 0) return -1;
  if (sdbSetRawStatus(pDbRaw, SDB_STATUS_CREATING) != 0) return -1;
  return 0;
}

static int32_t mndSetNewVgPrepareActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroups) {
  for (int32_t v = 0; v < pDb->cfg.numOfVgroups; ++v) {
    if (mndAddNewVgPrepareAction(pMnode, pTrans, (pVgroups + v)) != 0) return -1;
  }
  return 0;
}

static int32_t mndSetCreateDbRedoLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroups) {
  int32_t  code = 0;
  SSdbRaw *pDbRaw = mndDbActionEncode(pDb);
  if (pDbRaw == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    TAOS_RETURN(code);
  }
  TAOS_CHECK_RETURN(mndTransAppendRedolog(pTrans, pDbRaw));
  TAOS_CHECK_RETURN(sdbSetRawStatus(pDbRaw, SDB_STATUS_UPDATE));

  for (int32_t v = 0; v < pDb->cfg.numOfVgroups; ++v) {
    SSdbRaw *pVgRaw = mndVgroupActionEncode(pVgroups + v);
    if (pVgRaw == NULL) {
      code = TSDB_CODE_MND_RETURN_VALUE_NULL;
      if (terrno != 0) code = terrno;
      TAOS_RETURN(code);
    }
    TAOS_CHECK_RETURN(mndTransAppendRedolog(pTrans, pVgRaw));
    TAOS_CHECK_RETURN(sdbSetRawStatus(pVgRaw, SDB_STATUS_UPDATE));
  }

  if (pDb->cfg.withArbitrator) {
    for (int32_t v = 0; v < pDb->cfg.numOfVgroups; ++v) {
      SVgObj   *pVgObj = pVgroups + v;
      SArbGroup arbGroup = {0};
      TAOS_CHECK_RETURN(mndArbGroupInitFromVgObj(pVgObj, &arbGroup));
      TAOS_CHECK_RETURN(mndSetCreateArbGroupRedoLogs(pTrans, &arbGroup));
    }
  }

  TAOS_RETURN(code);
}

static int32_t mndSetCreateDbUndoLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroups) {
  int32_t  code = 0;
  SSdbRaw *pDbRaw = mndDbActionEncode(pDb);
  if (pDbRaw == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    TAOS_RETURN(code);
  }
  TAOS_CHECK_RETURN(mndTransAppendUndolog(pTrans, pDbRaw));
  TAOS_CHECK_RETURN(sdbSetRawStatus(pDbRaw, SDB_STATUS_DROPPED));

  for (int32_t v = 0; v < pDb->cfg.numOfVgroups; ++v) {
    SSdbRaw *pVgRaw = mndVgroupActionEncode(pVgroups + v);
    if (pVgRaw == NULL) {
      code = TSDB_CODE_MND_RETURN_VALUE_NULL;
      if (terrno != 0) code = terrno;
      TAOS_RETURN(code);
    }
    TAOS_CHECK_RETURN(mndTransAppendUndolog(pTrans, pVgRaw));
    TAOS_CHECK_RETURN(sdbSetRawStatus(pVgRaw, SDB_STATUS_DROPPED));
  }

  if (pDb->cfg.withArbitrator) {
    for (int32_t v = 0; v < pDb->cfg.numOfVgroups; ++v) {
      SVgObj   *pVgObj = pVgroups + v;
      SArbGroup arbGroup = {0};
      TAOS_CHECK_RETURN(mndArbGroupInitFromVgObj(pVgObj, &arbGroup));
      TAOS_CHECK_RETURN(mndSetCreateArbGroupUndoLogs(pTrans, &arbGroup));
    }
  }

  TAOS_RETURN(code);
}

static int32_t mndSetCreateDbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroups,
                                        SUserObj *pUserDuped, SArray *pAuditOwnedDbs) {
  int32_t  code = 0;
  SSdbRaw *pDbRaw = mndDbActionEncode(pDb);
  if (pDbRaw == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    TAOS_RETURN(code);
  }
  TAOS_CHECK_RETURN(mndTransAppendCommitlog(pTrans, pDbRaw));
  TAOS_CHECK_RETURN(sdbSetRawStatus(pDbRaw, SDB_STATUS_READY));

  for (int32_t v = 0; v < pDb->cfg.numOfVgroups; ++v) {
    SSdbRaw *pVgRaw = mndVgroupActionEncode(pVgroups + v);
    if (pVgRaw == NULL) {
      code = TSDB_CODE_MND_RETURN_VALUE_NULL;
      if (terrno != 0) code = terrno;
      TAOS_RETURN(code);
    }
    TAOS_CHECK_RETURN(mndTransAppendCommitlog(pTrans, pVgRaw));
    TAOS_CHECK_RETURN(sdbSetRawStatus(pVgRaw, SDB_STATUS_READY));
  }

  if (pDb->cfg.withArbitrator) {
    for (int32_t v = 0; v < pDb->cfg.numOfVgroups; ++v) {
      SVgObj   *pVgObj = pVgroups + v;
      SArbGroup arbGroup = {0};
      TAOS_CHECK_RETURN(mndArbGroupInitFromVgObj(pVgObj, &arbGroup));
      TAOS_CHECK_RETURN(mndSetCreateArbGroupCommitLogs(pTrans, &arbGroup));
    }
  }

  if (pUserDuped) {
    SSdbRaw *pUserRaw = mndUserActionEncode(pUserDuped);
    if (pUserRaw == NULL) {
      code = TSDB_CODE_MND_RETURN_VALUE_NULL;
      if (terrno != 0) code = terrno;
      TAOS_RETURN(code);
    }
    TAOS_CHECK_RETURN(mndTransAppendCommitlog(pTrans, pUserRaw));
    TAOS_CHECK_RETURN(sdbSetRawStatus(pUserRaw, SDB_STATUS_READY));
  }

  if(pAuditOwnedDbs) {
    int32_t auditOwnedDbSize = taosArrayGetSize(pAuditOwnedDbs);
    for (int32_t i = 0; i < auditOwnedDbSize; ++i) {
      SUserObj *pAuditUser = (SUserObj *)TARRAY_GET_ELEM(pAuditOwnedDbs, i);
      SSdbRaw  *pAuditUserRaw = mndUserActionEncode(pAuditUser);
      if (pAuditUserRaw == NULL) {
        TAOS_RETURN(terrno ? terrno : TSDB_CODE_MND_RETURN_VALUE_NULL);
      }
      TAOS_CHECK_RETURN(mndTransAppendCommitlog(pTrans, pAuditUserRaw));
      TAOS_CHECK_RETURN(sdbSetRawStatus(pAuditUserRaw, SDB_STATUS_READY));
    }
  }

  TAOS_RETURN(code);
}

static int32_t mndSetCreateDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroups) {
  int32_t code = 0;
  for (int32_t vg = 0; vg < pDb->cfg.numOfVgroups; ++vg) {
    SVgObj *pVgroup = pVgroups + vg;

    for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
      SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
      TAOS_CHECK_RETURN(mndAddCreateVnodeAction(pMnode, pTrans, pDb, pVgroup, pVgid));
    }
  }

  TAOS_RETURN(code);
}

static int32_t mndSetCreateDbUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroups) {
  int32_t code = 0;
  for (int32_t vg = 0; vg < pDb->cfg.numOfVgroups; ++vg) {
    SVgObj *pVgroup = pVgroups + vg;

    for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
      SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
      TAOS_CHECK_RETURN(mndAddDropVnodeAction(pMnode, pTrans, pDb, pVgroup, pVgid, false));
    }
  }

  TAOS_RETURN(code);
}

/**
 * @brief fill audit owned dbs for users with role SYSAUDIT or SYSAUDIT_LOG.
 *
 * @param pMnode
 * @param pOperUser
 * @param pDb
 * @param pAuditOwnedDbs
 * @return int32_t
 */
static int32_t mndSetAuditOwnedDbs(SMnode *pMnode, SUserObj *pOperUser, SDbObj *pDb, SArray **pAuditOwnedDbs) {
  int32_t   code = 0, lino = 0;
  SSdb     *pSdb = pMnode->pSdb;
  SArray   *auditOwnedDbs = NULL;
  SUserObj *pUser = NULL;
  SUserObj  newUserObj = {0};
  int32_t   sysAuditLen = sizeof(TSDB_ROLE_SYSAUDIT);
  int32_t   sysAuditLogLen = sizeof(TSDB_ROLE_SYSAUDIT_LOG);

  void *pIter = NULL;
  while ((pIter = sdbFetch(pSdb, SDB_USER, pIter, (void **)&pUser))) {
    if (taosHashGetSize(pUser->roles) <= 0) {
      sdbRelease(pSdb, pUser);
      pUser = NULL;
      continue;
    }
    if(strncmp(pUser->name, pOperUser->name, TSDB_USER_LEN) == 0) {
      sdbRelease(pSdb, pUser);
      pUser = NULL;
      continue;
    }
    if (!taosHashGet(pUser->roles, TSDB_ROLE_SYSAUDIT, sysAuditLen) &&
        !taosHashGet(pUser->roles, TSDB_ROLE_SYSAUDIT_LOG, sysAuditLogLen)) {
      sdbRelease(pSdb, pUser);
      pUser = NULL;
      continue;
    }
    TAOS_CHECK_EXIT(mndUserDupObj(pUser, &newUserObj));
    TAOS_CHECK_EXIT(taosHashPut(newUserObj.ownedDbs, pDb->name, strlen(pDb->name) + 1, NULL, 0));
    if (auditOwnedDbs == NULL) {
      auditOwnedDbs = taosArrayInit(1, sizeof(SUserObj));
      if (auditOwnedDbs == NULL) {
        TAOS_CHECK_EXIT(terrno);
      }
    }
    if (!taosArrayPush(auditOwnedDbs, &newUserObj)) {
      TAOS_CHECK_EXIT(terrno);
    }
    memset(&newUserObj, 0, sizeof(SUserObj));
    sdbRelease(pSdb, pUser);
    pUser = NULL;
  }
  *pAuditOwnedDbs = auditOwnedDbs;
  auditOwnedDbs = NULL;
_exit:
  if (auditOwnedDbs != NULL) taosArrayDestroyEx(auditOwnedDbs, (FDelete)mndUserFreeObj);
  sdbRelease(pSdb, pUser);
  sdbCancelFetch(pSdb, pIter);
  mndUserFreeObj(&newUserObj);
  TAOS_RETURN(code);
}

static int32_t mndCreateDb(SMnode *pMnode, SRpcMsg *pReq, SCreateDbReq *pCreate, SUserObj *pUser, SArray *dnodeList) {
  int32_t  code = 0;
  SArray  *auditOwnedDbs = NULL;
  SUserObj newUserObj = {0};
  SDbObj   dbObj = {0};
  (void)memcpy(dbObj.name, pCreate->db, TSDB_DB_FNAME_LEN);
  (void)memcpy(dbObj.acct, pUser->acct, TSDB_USER_LEN);
  dbObj.createdTime = taosGetTimestampMs();
  dbObj.updateTime = dbObj.createdTime;
  dbObj.uid = mndGenerateUid(dbObj.name, TSDB_DB_FNAME_LEN);
  dbObj.cfgVersion = 1;
  dbObj.vgVersion = 1;
  dbObj.tsmaVersion = 1;
  (void)memcpy(dbObj.createUser, pUser->user, TSDB_USER_LEN);
  dbObj.ownerId = pUser->uid;
  dbObj.cfg = (SDbCfg){
      .numOfVgroups = pCreate->numOfVgroups,
      .numOfStables = pCreate->numOfStables,
      .buffer = pCreate->buffer,
      .pageSize = pCreate->pageSize,
      .pages = pCreate->pages,
      .cacheLastSize = pCreate->cacheLastSize,
      .cacheLastShardBits = pCreate->cacheLastShardBits,
      .daysPerFile = pCreate->daysPerFile,
      .daysToKeep0 = pCreate->daysToKeep0,
      .daysToKeep1 = pCreate->daysToKeep1,
      .daysToKeep2 = pCreate->daysToKeep2,
      .keepTimeOffset = pCreate->keepTimeOffset,
      .minRows = pCreate->minRows,
      .maxRows = pCreate->maxRows,
      .walFsyncPeriod = pCreate->walFsyncPeriod,
      .walLevel = pCreate->walLevel,
      .precision = pCreate->precision,
      .compression = pCreate->compression,
      .replications = pCreate->replications,
      .strict = pCreate->strict,
      .cacheLast = pCreate->cacheLast,
      .hashMethod = 1,
      .schemaless = pCreate->schemaless,
      .walRetentionPeriod = pCreate->walRetentionPeriod,
      .walRetentionSize = pCreate->walRetentionSize,
      .walRollPeriod = pCreate->walRollPeriod,
      .walSegmentSize = pCreate->walSegmentSize,
      .sstTrigger = pCreate->sstTrigger,
      .hashPrefix = pCreate->hashPrefix,
      .hashSuffix = pCreate->hashSuffix,
      .ssChunkSize = pCreate->ssChunkSize,
      .ssKeepLocal = pCreate->ssKeepLocal,
      .ssCompact = pCreate->ssCompact,
      .tsdbPageSize = pCreate->tsdbPageSize,
      .withArbitrator = pCreate->withArbitrator,
      .encryptAlgorithm = pCreate->encryptAlgorithm,
      .compactInterval = pCreate->compactInterval,
      .compactStartTime = pCreate->compactStartTime,
      .compactEndTime = pCreate->compactEndTime,
      .compactTimeOffset = pCreate->compactTimeOffset,
      .isAudit = pCreate->isAudit,
      .secureDelete = pCreate->secureDelete,
  };
  if (strlen(pCreate->encryptAlgrName) > 0) {
    if (strncasecmp(pCreate->encryptAlgrName, "none", TSDB_ENCRYPT_ALGR_NAME_LEN) == 0) {
      if (pCreate->isAudit == 1) {
        code = TSDB_CODE_AUDIT_MUST_ENCRYPT;
        mError("db:%s, failed to create, encrypt algorithm not match for audit db, %s", pCreate->db,
               pCreate->encryptAlgrName);
        TAOS_RETURN(code);
      }
      dbObj.cfg.encryptAlgorithm = 0;
    } else {
      SEncryptAlgrObj *pEncryptAlgr = mndAcquireEncryptAlgrByAId(pMnode, pCreate->encryptAlgrName);
      if (pEncryptAlgr != NULL) {
        if (pEncryptAlgr->type != ENCRYPT_ALGR_TYPE__SYMMETRIC_CIPHERS) {
          code = TSDB_CODE_MNODE_ENCRYPT_TYPE_NOT_MATCH;
          mError("db:%s, failed to create, encrypt algorithm not match, %s, expect type %d", pCreate->db,
                 pCreate->encryptAlgrName, ENCRYPT_ALGR_TYPE__SYMMETRIC_CIPHERS);
          TAOS_RETURN(code);
        }
        dbObj.cfg.encryptAlgorithm = pEncryptAlgr->id;
      } else {
        code = TSDB_CODE_MNODE_ENCRYPT_ALGR_NOT_EXIST;
        mError("db:%s, failed to create, encrypt algorithm not exist, %s", pCreate->db, pCreate->encryptAlgrName);
        TAOS_RETURN(code);
      }
      if (tsDataKey[0] == '\0') {
        code = TSDB_CODE_DNODE_INVALID_ENCRYPTKEY;
        mError("db:%s, failed to create db since %s", pCreate->db, tstrerror(code));
        TAOS_RETURN(code);
      }
    }
  } else {
    if (pCreate->isAudit == 1) {
      code = TSDB_CODE_AUDIT_MUST_ENCRYPT;
      mError("db:%s, failed to create, encrypt algorithm not match for audit db, %s", pCreate->db,
             pCreate->encryptAlgrName);
      TAOS_RETURN(code);
    }
  }
  dbObj.cfg.numOfRetensions = pCreate->numOfRetensions;
  dbObj.cfg.pRetensions = pCreate->pRetensions;

  if (pCreate->isAudit == 1) {
    if (dbObj.cfg.daysToKeep2 < 2628000) {
      code = TSDB_CODE_AUDIT_MUST_KEEPFORCE;
      mError("db:%s, failed to create, keep not match for audit db, %d", pCreate->db, dbObj.cfg.daysToKeep2);
      TAOS_RETURN(code);
    }
    if (dbObj.cfg.walLevel != 2) {
      code = TSDB_CODE_AUDIT_MUST_WALFORCE;
      mError("db:%s, failed to create, walLevel not match for audit db, %d", pCreate->db, dbObj.cfg.walLevel);
      TAOS_RETURN(code);
    }
  }
  dbObj.cfg.allowDrop = (uint8_t)pCreate->allowDrop;

#ifdef TD_ENTERPRISE  
  // MAC: DB securityLevel
  if (pCreate->securityLevel == 0) {
    // Explicitly specified as 0: no need to check privilege, just set it to 0
    dbObj.cfg.securityLevel = TSDB_DEFAULT_SECURITY_LEVEL;
  } else if (pCreate->securityLevel > 0) {
    // Require SYSSEC (PRIV_SECURITY_POLICY_ALTER) to set an explicit non-zero security level.
    if (!mndUserHasMacLabelPriv(pMnode, pUser)) {
      code = TSDB_CODE_MND_NO_RIGHTS;
      mError("db:%s, failed to create, user %s lacks PRIV_SECURITY_POLICY_ALTER to set security_level",
             pCreate->db, pUser->user);
      TAOS_RETURN(code);
    }
    // MAC must be active to set db security_level > 0; before activation only user levels can be set.
    if (pMnode->macActive != MAC_MODE_MANDATORY) {
      code = TSDB_CODE_MAC_INSUFFICIENT_LEVEL;
      mError("db:%s, failed to create, cannot set security_level > 0 before MAC is activated",
             pCreate->db);
      TAOS_RETURN(code);
    }
    // NRU: requested level must not exceed operator's clearance.
    // (SYSSEC requires maxSecLevel=4 under MAC, so this check always passes in practice.)
    if ((uint8_t)pCreate->securityLevel > pUser->maxSecLevel) {
      code = TSDB_CODE_MAC_INSUFFICIENT_LEVEL;
      mError("db:%s, failed to create, securityLevel(%d) exceeds user %s maxSecLevel(%d)",
             pCreate->db, pCreate->securityLevel, pUser->user, pUser->maxSecLevel);
      TAOS_RETURN(code);
    }
    dbObj.cfg.securityLevel = (uint8_t)pCreate->securityLevel;
  } else {
    // Not specified: MAC active inherit creator's maxSecLevel as default, otherwise default security_level = 0
    dbObj.cfg.securityLevel =
        pMnode->macActive == MAC_MODE_MANDATORY ? pUser->maxSecLevel : TSDB_DEFAULT_SECURITY_LEVEL;
  }
#endif
  mndSetDefaultDbCfg(&dbObj.cfg);

  if ((code = mndCheckDbName(dbObj.name, pUser)) != 0) {
    mError("db:%s, failed to create, check db name failed, since %s", pCreate->db, tstrerror(code));
    TAOS_RETURN(code);
  }

  if ((code = mndCheckDbCfg(pMnode, &dbObj.cfg)) != 0) {
    mError("db:%s, failed to create, check db cfg failed, since %s", pCreate->db, tstrerror(code));
    TAOS_RETURN(code);
  }

  if (dbObj.cfg.hashPrefix > 0) {
    int32_t dbLen = strlen(dbObj.name) + 1;
    mInfo("db:%s, hashPrefix adjust from %d to %d", dbObj.name, dbObj.cfg.hashPrefix, dbObj.cfg.hashPrefix + dbLen);
    dbObj.cfg.hashPrefix += dbLen;
  } else if (dbObj.cfg.hashPrefix < 0) {
    int32_t dbLen = strlen(dbObj.name) + 1;
    mInfo("db:%s, hashPrefix adjust from %d to %d", dbObj.name, dbObj.cfg.hashPrefix, dbObj.cfg.hashPrefix - dbLen);
    dbObj.cfg.hashPrefix -= dbLen;
  }

  SVgObj *pVgroups = NULL;
  if ((code = mndAllocVgroup(pMnode, &dbObj, &pVgroups, dnodeList)) != 0) {
    mError("db:%s, failed to create, alloc vgroup failed, since %s", pCreate->db, terrstr());
    TAOS_RETURN(code);
  }

  // add database privileges for user
  SUserObj *pNewUserDuped = NULL;
#if 0
  // We record the DB's owner id instread. The owner has the permission to create objects under the DB, but does not
  // have default permissions on objects created by others .Therefore, readDbs/writeDbs are no longer applicable.
  if (!pUser->superUser) {
    TAOS_CHECK_GOTO(mndUserDupObj(pUser, &newUserObj), NULL, _OVER);
    TAOS_CHECK_GOTO(taosHashPut(newUserObj.readDbs, dbObj.name, strlen(dbObj.name) + 1, dbObj.name, TSDB_FILENAME_LEN),
                    NULL, _OVER);
    TAOS_CHECK_GOTO(taosHashPut(newUserObj.writeDbs, dbObj.name, strlen(dbObj.name) + 1, dbObj.name, TSDB_FILENAME_LEN),
                    NULL, _OVER);
    pNewUserDuped = &newUserObj;
  }
#else
  // Considering the efficiency of use db privileges in some scenarios like insert operation, owned DBs is stored.
  bool addOwned = false;
  if (dbObj.cfg.isAudit == 1) {
    if (taosHashGet(pUser->roles, TSDB_ROLE_SYSAUDIT, sizeof(TSDB_ROLE_SYSAUDIT)) ||
        taosHashGet(pUser->roles, TSDB_ROLE_SYSAUDIT_LOG, sizeof(TSDB_ROLE_SYSAUDIT_LOG))) {
      addOwned = true;
    }
    TAOS_CHECK_GOTO(mndSetAuditOwnedDbs(pMnode, pUser, &dbObj, &auditOwnedDbs), NULL, _OVER);
  } else {
    addOwned = true;
  }
  if (addOwned) {
    TAOS_CHECK_GOTO(mndUserDupObj(pUser, &newUserObj), NULL, _OVER);
    TAOS_CHECK_GOTO(taosHashPut(newUserObj.ownedDbs, dbObj.name, strlen(dbObj.name) + 1, NULL, 0), NULL, _OVER);
    pNewUserDuped = &newUserObj;
  }
#endif

  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq, "create-db");
  if (pTrans == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    goto _OVER;
  }
  // mndTransSetSerial(pTrans);
  mInfo("trans:%d, used to create db:%s", pTrans->id, pCreate->db);

  mndTransSetDbName(pTrans, dbObj.name, NULL);
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);

  void   *pIter = NULL;
  STrans *exitTrans = NULL;

  if (!pUser->superUser) {
    while (1) {
      pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&exitTrans);
      if (pIter == NULL) break;

      if (exitTrans->conflict == TRN_CONFLICT_DB &&
          strncmp(exitTrans->opername, "create-db", TSDB_TRANS_OPER_LEN) == 0) {
        code = TSDB_CODE_MND_TRANS_CONFLICT;
        sdbRelease(pMnode->pSdb, exitTrans);
        sdbCancelFetch(pMnode->pSdb, pIter);
        goto _OVER;
      }

      sdbRelease(pMnode->pSdb, exitTrans);
    }
  }

  if (dbObj.cfg.isAudit) {
    if (dbObj.cfg.numOfVgroups > 1) {
      code = TSDB_CODE_AUDIT_DB_NOT_MULTI_VGROUP;
      TAOS_CHECK_GOTO(code, NULL, _OVER);
    }
  }

  mndTransSetOper(pTrans, MND_OPER_CREATE_DB);
  TAOS_CHECK_GOTO(mndSetCreateDbPrepareAction(pMnode, pTrans, &dbObj), NULL, _OVER);
  TAOS_CHECK_GOTO(mndSetCreateDbRedoActions(pMnode, pTrans, &dbObj, pVgroups), NULL, _OVER);
  TAOS_CHECK_GOTO(mndSetNewVgPrepareActions(pMnode, pTrans, &dbObj, pVgroups), NULL, _OVER);
  TAOS_CHECK_GOTO(mndSetCreateDbUndoLogs(pMnode, pTrans, &dbObj, pVgroups), NULL, _OVER);
  TAOS_CHECK_GOTO(mndSetCreateDbCommitLogs(pMnode, pTrans, &dbObj, pVgroups, pNewUserDuped, auditOwnedDbs), NULL, _OVER);
  TAOS_CHECK_GOTO(mndSetCreateDbUndoActions(pMnode, pTrans, &dbObj, pVgroups), NULL, _OVER);

  if (dbObj.cfg.isAudit) {
    // when create audit database, pVgroups num always is 1
    TAOS_CHECK_GOTO(mndCreateAuditStb(pMnode, &dbObj, pUser, pTrans, pVgroups), NULL, _OVER);
  }
  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);

_OVER:
  taosMemoryFree(pVgroups);
  mndUserFreeObj(&newUserObj);
  taosArrayDestroyEx(auditOwnedDbs, (FDelete)mndUserFreeObj);
  mndTransDrop(pTrans);
  TAOS_RETURN(code);
}

static int32_t mndCheckDbEncryptKey(SMnode *pMnode, SCreateDbReq *pReq) {
  int32_t    code = 0;
  SSdb      *pSdb = pMnode->pSdb;
  SDnodeObj *pDnode = NULL;
  void      *pIter = NULL;

#if defined(TD_ENTERPRISE) || defined(TD_ASTRA_TODO)
  if (pReq->encryptAlgorithm == TSDB_ENCRYPT_ALGO_NONE) goto _exit;
  TAOS_CHECK_GOTO(grantCheck(TSDB_GRANT_DB_ENCRYPTION), NULL, _exit);
  if (tsEncryptionKeyStat != ENCRYPT_KEY_STAT_LOADED) {
    code = TSDB_CODE_MND_INVALID_ENCRYPT_KEY;
    mError("db:%s, failed to check encryption key:%" PRIi8 " in mnode leader since it's not loaded", pReq->db,
           tsEncryptionKeyStat);
    goto _exit;
  }

  int64_t curMs = taosGetTimestampMs();
  while ((pIter = sdbFetch(pSdb, SDB_DNODE, pIter, (void **)&pDnode))) {
    bool online = false;
    if ((pDnode->encryptionKeyStat != tsEncryptionKeyStat || pDnode->encryptionKeyChksum != tsEncryptionKeyChksum) &&
        (online = mndIsDnodeOnline(pDnode, curMs))) {
      code = TSDB_CODE_MND_INVALID_ENCRYPT_KEY;
      mError("db:%s, failed to check encryption key:%" PRIi8
             "-%u in dnode:%d since it's inconsitent with mnode leader:%" PRIi8 "-%u",
             pReq->db, pDnode->encryptionKeyStat, pDnode->encryptionKeyChksum, pDnode->id, tsEncryptionKeyStat,
             tsEncryptionKeyChksum);
      sdbCancelFetch(pSdb, pIter);
      sdbRelease(pSdb, pDnode);
      break;
    }
    sdbRelease(pSdb, pDnode);
  }
#else
  if (pReq->encryptAlgorithm != TSDB_ENCRYPT_ALGO_NONE) {
    code = TSDB_CODE_MND_INVALID_DB_OPTION;
    goto _exit;
  }
#endif
_exit:
  TAOS_RETURN(code);
}

#ifndef TD_ENTERPRISE
int32_t mndCheckDbDnodeList(SMnode *pMnode, char *db, char *dnodeListStr, SArray *dnodeList) {
  if (dnodeListStr[0] != 0) {
    terrno = TSDB_CODE_OPS_NOT_SUPPORT;
    return terrno;
  } else {
    return 0;
  }
}
#endif

static int32_t mndProcessCreateDbReq(SRpcMsg *pReq) {
  SMnode      *pMnode = pReq->info.node;
  int32_t      code = -1;
  int32_t      lino = 0;
  SDbObj      *pDb = NULL;
  SUserObj    *pUser = NULL;
  SCreateDbReq createReq = {0};
  SArray      *dnodeList = NULL;
  int64_t      tss = taosGetTimestampMs();

  dnodeList = taosArrayInit(mndGetDnodeSize(pMnode), sizeof(int32_t));
  TSDB_CHECK_NULL(dnodeList, code, lino, _OVER, TSDB_CODE_OUT_OF_MEMORY);

  TAOS_CHECK_GOTO(tDeserializeSCreateDbReq(pReq->pCont, pReq->contLen, &createReq), NULL, _OVER);
#ifdef WINDOWS
  if (taosArrayGetSize(createReq.pRetensions) > 0) {
    terrno = TSDB_CODE_MND_INVALID_PLATFORM;
    goto _OVER;
  }
#endif
  mInfo("db:%s, start to create, vgroups:%d, isAudit:%d, daysToKeep2:%d", createReq.db, createReq.numOfVgroups, createReq.isAudit, createReq.daysToKeep2);

  pDb = mndAcquireDb(pMnode, createReq.db);
  if (pDb != NULL) {
    if (createReq.ignoreExist) {
      mInfo("db:%s, already exist, ignore exist is set", createReq.db);
      code = 0;
      goto _OVER;
    } else {
      code = TSDB_CODE_MND_DB_ALREADY_EXIST;
      goto _OVER;
    }
  } else {
    if (terrno == TSDB_CODE_MND_DB_IN_CREATING) {
      code = terrno;
      goto _OVER;
    } else if (terrno == TSDB_CODE_MND_DB_IN_DROPPING) {
      goto _OVER;
    } else if (terrno == TSDB_CODE_MND_DB_NOT_EXIST) {
      // continue
    } else {  // TSDB_CODE_APP_ERROR
      goto _OVER;
    }
  }

  TAOS_CHECK_GOTO(mndCheckDbPrivilege(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_CREATE_DB, NULL), &lino, _OVER);

  TAOS_CHECK_GOTO(grantCheck(TSDB_GRANT_DB), &lino, _OVER);

  int32_t nVnodes = createReq.numOfVgroups * createReq.replications;
  TAOS_CHECK_GOTO(grantCheckEx(TSDB_GRANT_VNODE, &nVnodes), &lino, _OVER);

  if (createReq.replications == 2) {
    TAOS_CHECK_GOTO(grantCheck(TSDB_GRANT_DUAL_REPLICA_HA), &lino, _OVER);
  }

  TAOS_CHECK_GOTO(mndCheckDbEncryptKey(pMnode, &createReq), &lino, _OVER);

  TAOS_CHECK_GOTO(mndCheckDbDnodeList(pMnode, createReq.db, createReq.dnodeListStr, dnodeList), &lino, _OVER);

  TAOS_CHECK_GOTO(mndAcquireUser(pMnode, RPC_MSG_USER(pReq), &pUser), &lino, _OVER);

#ifdef TD_ENTERPRISE
  // MAC security checks on DB securityLevel (only when MAC is mandatory and level is explicitly set)
  if (pMnode->macActive == MAC_MODE_MANDATORY && createReq.securityLevel >= 0) {
    // NRU: user cannot create DB with securityLevel above their own clearance
    if (!pUser->superUser && pUser->maxSecLevel < createReq.securityLevel) {
      mError("user:%s, MAC NRU denied: cannot create DB with secLevel(%d) > maxSecLevel(%d)",
             RPC_MSG_USER(pReq), createReq.securityLevel, pUser->maxSecLevel);
      TAOS_CHECK_GOTO(TSDB_CODE_MAC_INSUFFICIENT_LEVEL, &lino, _OVER);
    }
    // NWD: user cannot create DB with securityLevel below their mandatory floor.
    // Trusted principals (PRIV_SECURITY_POLICY_ALTER / SYSSEC) are exempt as trusted subjects.
    if (!mndUserHasMacLabelPriv(pMnode, pUser) && createReq.securityLevel < pUser->minSecLevel) {
      mError("user:%s, MAC NWD denied: cannot create DB with secLevel(%d) < minSecLevel(%d)",
             RPC_MSG_USER(pReq), createReq.securityLevel, pUser->minSecLevel);
      TAOS_CHECK_GOTO(TSDB_CODE_MAC_NO_WRITE_DOWN, &lino, _OVER);
    }
  }
#endif

  if (sdbGetSize(pMnode->pSdb, SDB_MOUNT) > 0) {
    TAOS_CHECK_GOTO(TSDB_CODE_MND_MOUNT_NOT_EMPTY, &lino, _OVER);
  }

  if (createReq.isAudit == 1) {
    SDbObj *pAuditDb = mndAcquireAuditDb(pMnode);
    if (pAuditDb != NULL) {
      mndReleaseDb(pMnode, pAuditDb);
      mError("db:%s, audit db already exist, %s", createReq.db, pAuditDb->name);
      TAOS_CHECK_GOTO(TSDB_CODE_AUDIT_DB_ALREADY_EXIST, &lino, _OVER);
    }
    char *realDbName = strchr(createReq.db, '.');
    if (realDbName && strcmp(realDbName + 1, "log") == 0) {
      mError("db:%s, failed to create, db name not allowed for audit db, %s", createReq.db, realDbName + 1);
      TAOS_CHECK_GOTO(TSDB_CODE_MND_INVALID_DB, &lino, _OVER);
    }
  }

  TAOS_CHECK_GOTO(mndCreateDb(pMnode, pReq, &createReq, pUser, dnodeList), &lino, _OVER);
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;

  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {
    SName name = {0};
    if (tNameFromString(&name, createReq.db, T_NAME_ACCT | T_NAME_DB) < 0)
      mError("db:%s, failed to parse db name", createReq.db);
    int64_t tse = taosGetTimestampMs();
    double  duration = (double)(tse - tss);
    duration = duration / 1000;
    auditRecord(pReq, pMnode->clusterId, "createDB", name.dbname, "", createReq.sql, createReq.sqlLen, duration, 0);
  }

_OVER:
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
    mError("db:%s, failed to create at line:%d since %s", createReq.db, lino, tstrerror(code));
  }

  mndReleaseDb(pMnode, pDb);
  mndReleaseUser(pMnode, pUser);
  tFreeSCreateDbReq(&createReq);
  taosArrayDestroy(dnodeList);

  TAOS_RETURN(code);
}

static int32_t mndSetDbCfgFromAlterDbReq(SDbObj *pDb, SAlterDbReq *pAlter) {
  int32_t code = TSDB_CODE_MND_DB_OPTION_UNCHANGED;

  if (pAlter->buffer > 0 && pAlter->buffer != pDb->cfg.buffer) {
    pDb->cfg.buffer = pAlter->buffer;
    code = 0;
  }

  if (pAlter->pages > 0 && pAlter->pages != pDb->cfg.pages) {
    pDb->cfg.pages = pAlter->pages;
    code = 0;
  }

  if (pAlter->pageSize > 0 && pAlter->pageSize != pDb->cfg.pageSize) {
    pDb->cfg.pageSize = pAlter->pageSize;
    code = 0;
  }

  if (pAlter->daysPerFile > 0 && pAlter->daysPerFile != pDb->cfg.daysPerFile) {
    pDb->cfg.daysPerFile = pAlter->daysPerFile;
    code = 0;
  }

  if (pAlter->daysToKeep0 > 0 && pAlter->daysToKeep0 != pDb->cfg.daysToKeep0) {
    pDb->cfg.daysToKeep0 = pAlter->daysToKeep0;
    code = 0;
  }

  if (pAlter->daysToKeep1 > 0 && pAlter->daysToKeep1 != pDb->cfg.daysToKeep1) {
    pDb->cfg.daysToKeep1 = pAlter->daysToKeep1;
    code = 0;
  }

  if (pAlter->daysToKeep2 > 0 && pAlter->daysToKeep2 != pDb->cfg.daysToKeep2) {
    pDb->cfg.daysToKeep2 = pAlter->daysToKeep2;
    code = 0;
  }

  if (pAlter->keepTimeOffset >= 0 && pAlter->keepTimeOffset != pDb->cfg.keepTimeOffset) {
    pDb->cfg.keepTimeOffset = pAlter->keepTimeOffset;
    code = 0;
  }

  if (pAlter->walFsyncPeriod >= 0 && pAlter->walFsyncPeriod != pDb->cfg.walFsyncPeriod) {
    pDb->cfg.walFsyncPeriod = pAlter->walFsyncPeriod;
    code = 0;
  }

  if (pAlter->walLevel >= 0 && pAlter->walLevel != pDb->cfg.walLevel) {
    pDb->cfg.walLevel = pAlter->walLevel;
    code = 0;
  }

  if (pAlter->strict >= 0 && pAlter->strict != pDb->cfg.strict) {
#if 1
    code = TSDB_CODE_OPS_NOT_SUPPORT;
#else
    pDb->cfg.strict = pAlter->strict;
    code = 0;
#endif
  }

  if (pAlter->cacheLast >= 0 && pAlter->cacheLast != pDb->cfg.cacheLast) {
    pDb->cfg.cacheLast = pAlter->cacheLast;
    code = 0;
  }

  if (pAlter->cacheLastSize > 0 && pAlter->cacheLastSize != pDb->cfg.cacheLastSize) {
    pDb->cfg.cacheLastSize = pAlter->cacheLastSize;
    code = 0;
  }

  if (pAlter->cacheLastShardBits > 0 && pAlter->cacheLastShardBits != pDb->cfg.cacheLastShardBits) {
    pDb->cfg.cacheLastShardBits = pAlter->cacheLastShardBits;
    code = 0;
  }

  if (pAlter->replications > 0 && pAlter->replications != pDb->cfg.replications) {
    pDb->cfg.replications = pAlter->replications;
    pDb->vgVersion++;
    code = 0;
  }

  if (pAlter->sstTrigger > 0 && pAlter->sstTrigger != pDb->cfg.sstTrigger) {
    pDb->cfg.sstTrigger = pAlter->sstTrigger;
    pDb->vgVersion++;
    code = 0;
  }

  if (pAlter->minRows > 0 && pAlter->minRows != pDb->cfg.minRows) {
    pDb->cfg.minRows = pAlter->minRows;
    pDb->vgVersion++;
    code = 0;
  }

  if (pAlter->walRetentionPeriod > TSDB_DB_MIN_WAL_RETENTION_PERIOD &&
      pAlter->walRetentionPeriod != pDb->cfg.walRetentionPeriod) {
    pDb->cfg.walRetentionPeriod = pAlter->walRetentionPeriod;
    pDb->vgVersion++;
    code = 0;
  }

  if (pAlter->walRetentionSize > TSDB_DB_MIN_WAL_RETENTION_SIZE &&
      pAlter->walRetentionSize != pDb->cfg.walRetentionSize) {
    pDb->cfg.walRetentionSize = pAlter->walRetentionSize;
    pDb->vgVersion++;
    code = 0;
  }

  if (pAlter->ssKeepLocal >= TSDB_MIN_SS_KEEP_LOCAL && pAlter->ssKeepLocal <= TSDB_MAX_SS_KEEP_LOCAL && pAlter->ssKeepLocal != pDb->cfg.ssKeepLocal) {
    pDb->cfg.ssKeepLocal = pAlter->ssKeepLocal;
    pDb->vgVersion++;
    code = 0;
  }

  if (pAlter->ssCompact >= TSDB_MIN_SS_COMPACT && pAlter->ssCompact <= TSDB_MAX_SS_COMPACT && pAlter->ssCompact != pDb->cfg.ssCompact) {
    pDb->cfg.ssCompact = pAlter->ssCompact;
    pDb->vgVersion++;
    code = 0;
  }

  if (pAlter->withArbitrator >= TSDB_MIN_DB_WITH_ARBITRATOR && pAlter->withArbitrator != pDb->cfg.withArbitrator) {
    pDb->cfg.withArbitrator = pAlter->withArbitrator;
    pDb->vgVersion++;
    code = 0;
  }

  if (pAlter->compactInterval >= TSDB_DEFAULT_COMPACT_INTERVAL && pAlter->compactInterval != pDb->cfg.compactInterval) {
    pDb->cfg.compactInterval = pAlter->compactInterval;
    pDb->vgVersion++;
    code = 0;
  }

  bool compactTimeRangeChanged = false;
  if (pAlter->compactStartTime != pDb->cfg.compactStartTime &&
      (pAlter->compactStartTime == TSDB_DEFAULT_COMPACT_START_TIME ||
       pAlter->compactStartTime <= -pDb->cfg.daysPerFile)) {
    pDb->cfg.compactStartTime = pAlter->compactStartTime;
    compactTimeRangeChanged = true;
    code = 0;
  }

  if (pAlter->compactEndTime != pDb->cfg.compactEndTime &&
      (pAlter->compactEndTime == TSDB_DEFAULT_COMPACT_END_TIME || pAlter->compactEndTime <= -pDb->cfg.daysPerFile)) {
    pDb->cfg.compactEndTime = pAlter->compactEndTime;
    compactTimeRangeChanged = true;
    code = 0;
  }
  if(compactTimeRangeChanged) {
    pDb->vgVersion++;
  }

  if (pAlter->compactTimeOffset >= TSDB_MIN_COMPACT_TIME_OFFSET &&
      pAlter->compactTimeOffset != pDb->cfg.compactTimeOffset) {
    pDb->cfg.compactTimeOffset = pAlter->compactTimeOffset;
    pDb->vgVersion++;
    code = 0;
  }

  if (pAlter->isAudit > -1 && pAlter->isAudit != pDb->cfg.isAudit) {
    code = TSDB_CODE_OPS_NOT_SUPPORT;
    return code;
  }

  if (pAlter->allowDrop > -1 && pAlter->allowDrop != pDb->cfg.allowDrop) {
    pDb->cfg.allowDrop = pAlter->allowDrop;
    pDb->vgVersion++;
    code = 0;
  }

  if (pAlter->secureDelete > -1 && pAlter->secureDelete != pDb->cfg.secureDelete) {
    pDb->cfg.secureDelete = pAlter->secureDelete;
    pDb->vgVersion++;
    code = 0;
  }

  if (pAlter->securityLevel > -1 && ((uint8_t)pAlter->securityLevel != pDb->cfg.securityLevel)) {
    pDb->cfg.securityLevel = (uint8_t)pAlter->securityLevel;
    pDb->vgVersion++;
    code = 0;
  }

  TAOS_RETURN(code);
}

static int32_t mndSetAlterDbPrepareLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pOld, SDbObj *pNew) {
  int32_t  code = 0;
  SSdbRaw *pRedoRaw = mndDbActionEncode(pOld);
  if (pRedoRaw == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    TAOS_RETURN(code);
  }
  if ((code = mndTransAppendPrepareLog(pTrans, pRedoRaw)) != 0) {
    sdbFreeRaw(pRedoRaw);
    TAOS_RETURN(code);
  }

  if ((code = sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY)) != 0) {
    sdbFreeRaw(pRedoRaw);
    TAOS_RETURN(code);
  }
  return 0;
}

int32_t mndSetAlterDbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pOld, SDbObj *pNew) {
  int32_t  code = 0;
  SSdbRaw *pCommitRaw = mndDbActionEncode(pNew);
  if (pCommitRaw == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    TAOS_RETURN(code);
  }
  if ((code = mndTransAppendCommitlog(pTrans, pCommitRaw)) != 0) {
    sdbFreeRaw(pCommitRaw);
    TAOS_RETURN(code);
  }

  if ((code = sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY)) != 0) {
    sdbFreeRaw(pCommitRaw);
    TAOS_RETURN(code);
  }
  TAOS_RETURN(code);
}

static int32_t mndSetAlterDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pOldDb, SDbObj *pNewDb) {
  int32_t code = 0, lino = 0;
  SSdb   *pSdb = pMnode->pSdb;
  void   *pIter = NULL;
  SVgObj *pVgroup = NULL;
  SArray *pArray = mndBuildDnodesArray(pMnode, 0, NULL);

  TSDB_CHECK_NULL(pArray, code, lino, _err, TSDB_CODE_OUT_OF_MEMORY);

  while (1) {
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
    if (pIter == NULL) break;

    if (mndVgroupInDb(pVgroup, pNewDb->uid)) {
      SVgObj newVgroup = {0};
      TAOS_CHECK_GOTO(mndBuildAlterVgroupAction(pMnode, pTrans, pOldDb, pNewDb, pVgroup, pArray, &newVgroup), &lino,
                      _err);

      if (pNewDb->cfg.withArbitrator != pOldDb->cfg.withArbitrator) {
        if (pNewDb->cfg.withArbitrator) {
          SArbGroup arbGroup = {0};
          TAOS_CHECK_GOTO(mndArbGroupInitFromVgObj(&newVgroup, &arbGroup), &lino, _err);
          TAOS_CHECK_GOTO(mndSetCreateArbGroupCommitLogs(pTrans, &arbGroup), &lino, _err);
        } else {
          SArbGroup arbGroup = {0};
          TAOS_CHECK_GOTO(mndArbGroupInitFromVgObj(pVgroup, &arbGroup), &lino, _err);
          TAOS_CHECK_GOTO(mndSetDropArbGroupCommitLogs(pTrans, &arbGroup), &lino, _err);
        }
      }
    }

    sdbRelease(pSdb, pVgroup);
  }

  taosArrayDestroy(pArray);
  TAOS_RETURN(code);

_err:
  mError("db:%s, %s failed at %d since %s", pNewDb->name, __func__, lino, tstrerror(code));

  sdbCancelFetch(pSdb, pIter);
  sdbRelease(pSdb, pVgroup);
  taosArrayDestroy(pArray);
  TAOS_RETURN(code);
}

static int32_t mndAlterDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pOld, SDbObj *pNew) {
  int32_t code = -1;
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq, "alter-db");
  if (pTrans == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    TAOS_RETURN(code);
  }
  mInfo("trans:%d, used to alter db:%s", pTrans->id, pOld->name);
  mInfo("trans:%d, used to alter db, ableToBeKilled:%d, killMode:%d", pTrans->id, pTrans->ableToBeKilled, pTrans->killMode);

  mndTransSetDbName(pTrans, pOld->name, NULL);
  mndTransSetGroupParallel(pTrans);
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);
  TAOS_CHECK_GOTO(mndTransCheckConflictWithCompact(pMnode, pTrans), NULL, _OVER);
  TAOS_CHECK_GOTO(mndTransCheckConflictWithRetention(pMnode, pTrans), NULL, _OVER);

  TAOS_CHECK_GOTO(mndSetAlterDbPrepareLogs(pMnode, pTrans, pOld, pNew), NULL, _OVER);
  TAOS_CHECK_GOTO(mndSetAlterDbCommitLogs(pMnode, pTrans, pOld, pNew), NULL, _OVER);
  TAOS_CHECK_GOTO(mndSetAlterDbRedoActions(pMnode, pTrans, pOld, pNew), NULL, _OVER);

  mInfo("trans:%d, used to alter db, ableToBeKilled:%d, killMode:%d", pTrans->id, pTrans->ableToBeKilled, pTrans->killMode);
  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
  code = 0;

_OVER:
  mndTransDrop(pTrans);
  TAOS_RETURN(code);
}

static int32_t mndProcessAlterDbReq(SRpcMsg *pReq) {
  SMnode     *pMnode = pReq->info.node;
  int32_t     code = -1;
  SDbObj     *pDb = NULL;
  SAlterDbReq alterReq = {0};
  SDbObj      dbObj = {0};
  int64_t     tss = taosGetTimestampMs();

  TAOS_CHECK_GOTO(tDeserializeSAlterDbReq(pReq->pCont, pReq->contLen, &alterReq), NULL, _OVER);

  mInfo("db:%s, start to alter, encryt_algr:%s", alterReq.db, alterReq.encryptAlgrName);

  pDb = mndAcquireDb(pMnode, alterReq.db);
  if (pDb == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    goto _OVER;
  }

#ifdef TD_ENTERPRISE
  // MAC: only superUser or user with PRIV_SECURITY_POLICY_ALTER can ALTER DATABASE ... SECURITY_LEVEL
  // Check this BEFORE general mndCheckDbPrivilege, since holder may not have ALTER grant on the DB.
  if (alterReq.securityLevel > -1) {
    SUserObj *pUser = NULL;
    TAOS_CHECK_GOTO(mndAcquireUser(pMnode, RPC_MSG_USER(pReq), &pUser), NULL, _OVER);
    if (!mndUserHasMacLabelPriv(pMnode, pUser)) {
      mndReleaseUser(pMnode, pUser);
      code = TSDB_CODE_MND_NO_RIGHTS;
      mError("db:%s, failed to alter security_level, user %s lacks PRIV_SECURITY_POLICY_ALTER", alterReq.db,
             RPC_MSG_USER(pReq));
      goto _OVER;
    }
    // MAC must be active to set db security_level > 0; before activation only user levels can be set.
    if (alterReq.securityLevel > 0 && pMnode->macActive != MAC_MODE_MANDATORY) {
      mndReleaseUser(pMnode, pUser);
      code = TSDB_CODE_MAC_INSUFFICIENT_LEVEL;
      mError("db:%s, failed to alter, cannot set security_level > 0 before MAC is activated", alterReq.db);
      goto _OVER;
    }
    // NRU: when MAC is mandatory, new level must not exceed operator's clearance.
    // (SYSSEC requires maxSecLevel=4 under MAC, so this check always passes in practice.)
    if (pMnode->macActive == MAC_MODE_MANDATORY && (uint8_t)alterReq.securityLevel > pUser->maxSecLevel) {
      mndReleaseUser(pMnode, pUser);
      code = TSDB_CODE_MAC_INSUFFICIENT_LEVEL;
      mError("db:%s, failed to alter security_level(%d): exceeds user %s maxSecLevel(%d)", alterReq.db,
             alterReq.securityLevel, RPC_MSG_USER(pReq), pUser->maxSecLevel);
      goto _OVER;
    }
    mndReleaseUser(pMnode, pUser);
  } else {
    TAOS_CHECK_GOTO(mndCheckDbPrivilege(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_ALTER_DB, pDb), NULL,
                    _OVER);
  }
#endif
  if (alterReq.replications == 2) {
    TAOS_CHECK_GOTO(grantCheck(TSDB_GRANT_DUAL_REPLICA_HA), NULL, _OVER);
  }

  int32_t numOfTopics = 0;
  TAOS_CHECK_GOTO(mndGetNumOfTopics(pMnode, pDb->name, &numOfTopics), NULL, _OVER);

  if (numOfTopics != 0 && alterReq.walRetentionPeriod == 0) {
    code = TSDB_CODE_MND_DB_RETENTION_PERIOD_ZERO;
    mError("db:%s, not allowed to set WAL_RETENTION_PERIOD 0 when there are topics defined. numOfTopics:%d", pDb->name,
           numOfTopics);
    goto _OVER;
  }

  if (pDb->cfg.isMount) {
    code = TSDB_CODE_MND_MOUNT_OBJ_NOT_SUPPORT;
    goto _OVER;
  }

  (void)memcpy(&dbObj, pDb, sizeof(SDbObj));
  if (dbObj.cfg.pRetensions != NULL) {
    dbObj.cfg.pRetensions = taosArrayDup(pDb->cfg.pRetensions, NULL);
    if (dbObj.cfg.pRetensions == NULL) goto _OVER;
  }

  if (pDb->cfg.isAudit == 1) {
    if (alterReq.daysToKeep2 > -1 && alterReq.daysToKeep2 < 2628000) {
      code = TSDB_CODE_AUDIT_MUST_KEEPFORCE;
      mError("db:%s, failed to alter, keep not match for audit db, %d", alterReq.db, alterReq.daysToKeep2);
      goto _OVER;
    }
    if (alterReq.walLevel > -1 && alterReq.walLevel != 2) {
      code = TSDB_CODE_AUDIT_MUST_WALFORCE;
      mError("db:%s, failed to alter, walLevel not match for audit db, %d", alterReq.db, alterReq.walLevel);
      goto _OVER;
    }
    if (alterReq.isAudit == 0) {
      code = TSDB_CODE_AUDIT_DB_NOT_ALLOW_CHANGE;
      mError("db:%s, failed to alter, is not allowed to change audit db, %d", alterReq.db, alterReq.isAudit);
      goto _OVER;
    }
    if (alterReq.securityLevel > -1) {
      code = TSDB_CODE_AUDIT_DB_NOT_ALLOW_CHANGE;
      mError("db:%s, failed to alter, security_level of audit db is immutable (fixed at 4)", alterReq.db);
      goto _OVER;
    }
  }

  if (strlen(alterReq.encryptAlgrName) > 0) {
    mInfo("db:%s, check encryt algr:%s", alterReq.db, alterReq.encryptAlgrName);

    if (strncasecmp(alterReq.encryptAlgrName, "none", TSDB_ENCRYPT_ALGR_NAME_LEN) == 0) {
      if (pDb->cfg.encryptAlgorithm != 0) {
        code = TSDB_CODE_MND_ENCRYPT_NOT_ALLOW_CHANGE;
        mError("db:%s, failed to alter, not allowed to change, %s", alterReq.db, alterReq.encryptAlgrName);
        goto _OVER;
      }
    } else {
      SEncryptAlgrObj *pEncryptAlgr = mndAcquireEncryptAlgrByAId(pMnode, alterReq.encryptAlgrName);
      if (pEncryptAlgr != NULL) {
        if (pEncryptAlgr->type != ENCRYPT_ALGR_TYPE__SYMMETRIC_CIPHERS) {
          code = TSDB_CODE_MNODE_ENCRYPT_TYPE_NOT_MATCH;
          mError("db:%s, failed to alter, encrypt algorithm not match, %s, expect type %d", alterReq.db,
                 alterReq.encryptAlgrName, ENCRYPT_ALGR_TYPE__SYMMETRIC_CIPHERS);
          goto _OVER;
        }
        if (pEncryptAlgr->id != pDb->cfg.encryptAlgorithm) {
          code = TSDB_CODE_MND_ENCRYPT_NOT_ALLOW_CHANGE;
          mError("db:%s, failed to alter, not allowed to change, %s", alterReq.db, alterReq.encryptAlgrName);
          goto _OVER;
        }
      } else {
        code = TSDB_CODE_MNODE_ENCRYPT_ALGR_NOT_EXIST;
        mError("db:%s, failed to alter, encrypt algorithm not exist, %s", alterReq.db, alterReq.encryptAlgrName);
        goto _OVER;
      }
    }
  }

  // SYSSEC check for securityLevel already done above
  // MAC: When raising DB security_level, all STBs in the DB must have level >= new level.
  if (alterReq.securityLevel > -1 && (uint8_t)alterReq.securityLevel > pDb->cfg.securityLevel) {
    SSdb *pSdb2 = pMnode->pSdb;
    void *pStbIter = NULL;
    while (1) {
      SStbObj   *pStb = NULL;
      ESdbStatus stbStatus;
      pStbIter = sdbFetchAll(pSdb2, SDB_STB, pStbIter, (void **)&pStb, &stbStatus, true);
      if (pStbIter == NULL) break;
      if (pStb->dbUid == pDb->uid && pStb->securityLevel < (uint8_t)alterReq.securityLevel) {
        sdbCancelFetch(pSdb2, pStbIter);
        sdbRelease(pSdb2, pStb);
        code = TSDB_CODE_MAC_OBJ_LEVEL_BELOW_DB;
        mError("db:%s, failed to raise security_level to %d: stb %s has level %d", alterReq.db, alterReq.securityLevel,
               pStb->name, pStb->securityLevel);
        goto _OVER;
      }
      sdbRelease(pSdb2, pStb);
    }
  }

  code = mndSetDbCfgFromAlterDbReq(&dbObj, &alterReq);
  if (code != 0) {
    if (code == TSDB_CODE_MND_DB_OPTION_UNCHANGED) code = 0;
    goto _OVER;
  }

  TAOS_CHECK_GOTO(mndCheckInChangeDbCfg(pMnode, &pDb->cfg, &dbObj.cfg), NULL, _OVER);

  dbObj.cfgVersion++;
  dbObj.updateTime = taosGetTimestampMs();
  code = mndAlterDb(pMnode, pReq, pDb, &dbObj);

  if (dbObj.cfg.replications != pDb->cfg.replications) {
    // return quickly, operation executed asynchronously
    mInfo("db:%s, alter db replica from %d to %d", pDb->name, pDb->cfg.replications, dbObj.cfg.replications);
  } else {
    if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
  }

  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {
    SName name = {0};
    if (tNameFromString(&name, alterReq.db, T_NAME_ACCT | T_NAME_DB) < 0)
      mError("db:%s, failed to parse db name", alterReq.db);

    int64_t tse = taosGetTimestampMs();
    double  duration = (double)(tse - tss);
    duration = duration / 1000;
    auditRecord(pReq, pMnode->clusterId, "alterDB", name.dbname, "", alterReq.sql, alterReq.sqlLen, duration, 0);
  }

_OVER:
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
    if (terrno != 0) code = terrno;
    mError("db:%s, failed to alter since %s", alterReq.db, tstrerror(code));
  }

  mndReleaseDb(pMnode, pDb);
  taosArrayDestroy(dbObj.cfg.pRetensions);
  tFreeSAlterDbReq(&alterReq);

  TAOS_RETURN(code);
}

static void mndDumpDbCfgInfo(SDbCfgRsp *cfgRsp, SDbObj *pDb, char *algorithmsId) {
  tstrncpy(cfgRsp->db, pDb->name, sizeof(cfgRsp->db));
  cfgRsp->dbId = pDb->uid;
  cfgRsp->ownerId = pDb->ownerId;
  cfgRsp->cfgVersion = pDb->cfgVersion;
  cfgRsp->numOfVgroups = pDb->cfg.numOfVgroups;
  cfgRsp->numOfStables = pDb->cfg.numOfStables;
  cfgRsp->buffer = pDb->cfg.buffer;
  cfgRsp->cacheSize = pDb->cfg.cacheLastSize;
  cfgRsp->cacheShardBits = pDb->cfg.cacheLastShardBits;
  cfgRsp->pageSize = pDb->cfg.pageSize;
  cfgRsp->pages = pDb->cfg.pages;
  cfgRsp->daysPerFile = pDb->cfg.daysPerFile;
  cfgRsp->daysToKeep0 = pDb->cfg.daysToKeep0;
  cfgRsp->daysToKeep1 = pDb->cfg.daysToKeep1;
  cfgRsp->daysToKeep2 = pDb->cfg.daysToKeep2;
  cfgRsp->keepTimeOffset = pDb->cfg.keepTimeOffset;
  cfgRsp->minRows = pDb->cfg.minRows;
  cfgRsp->maxRows = pDb->cfg.maxRows;
  cfgRsp->walFsyncPeriod = pDb->cfg.walFsyncPeriod;
  cfgRsp->hashPrefix = pDb->cfg.hashPrefix;
  cfgRsp->hashSuffix = pDb->cfg.hashSuffix;
  cfgRsp->hashMethod = pDb->cfg.hashMethod;
  cfgRsp->walLevel = pDb->cfg.walLevel;
  cfgRsp->precision = pDb->cfg.precision;
  cfgRsp->compression = pDb->cfg.compression;
  cfgRsp->replications = pDb->cfg.replications;
  cfgRsp->strict = pDb->cfg.strict;
  cfgRsp->cacheLast = pDb->cfg.cacheLast;
  cfgRsp->tsdbPageSize = pDb->cfg.tsdbPageSize;
  cfgRsp->walRetentionPeriod = pDb->cfg.walRetentionPeriod;
  cfgRsp->walRollPeriod = pDb->cfg.walRollPeriod;
  cfgRsp->walRetentionSize = pDb->cfg.walRetentionSize;
  cfgRsp->walSegmentSize = pDb->cfg.walSegmentSize;
  cfgRsp->numOfRetensions = pDb->cfg.numOfRetensions;
  cfgRsp->pRetensions = taosArrayDup(pDb->cfg.pRetensions, NULL);
  cfgRsp->schemaless = pDb->cfg.schemaless;
  cfgRsp->sstTrigger = pDb->cfg.sstTrigger;
  cfgRsp->ssChunkSize = pDb->cfg.ssChunkSize;
  cfgRsp->ssKeepLocal = pDb->cfg.ssKeepLocal;
  cfgRsp->ssCompact = pDb->cfg.ssCompact;
  cfgRsp->withArbitrator = pDb->cfg.withArbitrator;
  // cfgRsp->encryptAlgr = pDb->cfg.encryptAlgorithm;
  cfgRsp->compactInterval = pDb->cfg.compactInterval;
  cfgRsp->compactStartTime = pDb->cfg.compactStartTime;
  cfgRsp->compactEndTime = pDb->cfg.compactEndTime;
  cfgRsp->compactTimeOffset = pDb->cfg.compactTimeOffset;
  cfgRsp->flags = pDb->cfg.flags;
  tstrncpy(cfgRsp->algorithmsId, algorithmsId, sizeof(cfgRsp->algorithmsId));
  cfgRsp->isAudit = pDb->cfg.isAudit;
  cfgRsp->secureDelete = pDb->cfg.secureDelete;
}

static int32_t mndProcessGetDbCfgReq(SRpcMsg *pReq) {
  SMnode   *pMnode = pReq->info.node;
  int32_t   code = -1;
  SDbObj   *pDb = NULL;
  SDbCfgReq cfgReq = {0};
  SDbCfgRsp cfgRsp = {0};

  TAOS_CHECK_GOTO(tDeserializeSDbCfgReq(pReq->pCont, pReq->contLen, &cfgReq), NULL, _OVER);

  const char *pRealDb = strstr(cfgReq.db, ".");
  if (pRealDb) {
    ++pRealDb;
  } else {
    pRealDb = &cfgReq.db[0];
  }
  if (!IS_SYS_DBNAME(pRealDb)) {
    pDb = mndAcquireDb(pMnode, cfgReq.db);
    if (pDb == NULL) {
      code = TSDB_CODE_MND_RETURN_VALUE_NULL;
      if (terrno != 0) code = terrno;
      goto _OVER;
    }

    char algrId[TSDB_ENCRYPT_ALGR_NAME_LEN] = {0};
    mndGetEncryptOsslAlgrNameById(pMnode, pDb->cfg.encryptAlgorithm, algrId);
    mndDumpDbCfgInfo(&cfgRsp, pDb, algrId);
  }

  int32_t contLen = tSerializeSDbCfgRsp(NULL, 0, &cfgRsp);
  void   *pRsp = rpcMallocCont(contLen);
  if (pRsp == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    goto _OVER;
  }

  int32_t ret = 0;
  if ((ret = tSerializeSDbCfgRsp(pRsp, contLen, &cfgRsp)) < 0) {
    code = ret;
    goto _OVER;
  }

  pReq->info.rsp = pRsp;
  pReq->info.rspLen = contLen;

  code = 0;

_OVER:

  tFreeSDbCfgRsp(&cfgRsp);

  if (code != 0) {
    mError("db:%s, failed to get cfg since %s", cfgReq.db, terrstr());
  }

  mndReleaseDb(pMnode, pDb);

  TAOS_RETURN(code);
}

int32_t mndSetDropDbPrepareLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
  int32_t  code = 0;
  SSdbRaw *pRedoRaw = mndDbActionEncode(pDb);
  if (pRedoRaw == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    return -1;
  }
  TAOS_CHECK_RETURN(mndTransAppendPrepareLog(pTrans, pRedoRaw));
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING));

  SSdb *pSdb = pMnode->pSdb;
  void *pIter = NULL;

  while (1) {
    SArbGroup *pArbGroup = NULL;
    pIter = sdbFetch(pSdb, SDB_ARBGROUP, pIter, (void **)&pArbGroup);
    if (pIter == NULL) break;

    if (pArbGroup->dbUid == pDb->uid) {
      if ((code = mndSetDropArbGroupPrepareLogs(pTrans, pArbGroup)) != 0) {
        sdbCancelFetch(pSdb, pIter);
        sdbRelease(pSdb, pArbGroup);
        TAOS_RETURN(code);
      }
    }

    sdbRelease(pSdb, pArbGroup);
  }

  TAOS_RETURN(code);
}

int32_t mndSetDropDbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
  int32_t  code = 0;
  SSdbRaw *pCommitRaw = mndDbActionEncode(pDb);
  if (pCommitRaw == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    return -1;
  }
  TAOS_CHECK_RETURN(mndTransAppendCommitlog(pTrans, pCommitRaw));
  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));

  SSdb *pSdb = pMnode->pSdb;
  void *pIter = NULL;

  while (1) {
    SArbGroup *pArbGroup = NULL;
    pIter = sdbFetch(pSdb, SDB_ARBGROUP, pIter, (void **)&pArbGroup);
    if (pIter == NULL) break;

    if (pArbGroup->dbUid == pDb->uid) {
      if ((code = mndSetDropArbGroupCommitLogs(pTrans, pArbGroup)) != 0) {
        sdbCancelFetch(pSdb, pIter);
        sdbRelease(pSdb, pArbGroup);
        TAOS_RETURN(code);
      }
    }

    sdbRelease(pSdb, pArbGroup);
  }

  while (1) {
    SVgObj *pVgroup = NULL;
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
    if (pIter == NULL) break;

    if (pVgroup->dbUid == pDb->uid) {
      SSdbRaw *pVgRaw = mndVgroupActionEncode(pVgroup);
      if (pVgRaw == NULL) {
        sdbCancelFetch(pSdb, pIter);
        sdbRelease(pSdb, pVgroup);
        code = TSDB_CODE_MND_RETURN_VALUE_NULL;
        if (terrno != 0) code = terrno;
        TAOS_RETURN(code);
      }
      if ((code = mndTransAppendCommitlog(pTrans, pVgRaw)) != 0) {
        sdbCancelFetch(pSdb, pIter);
        sdbRelease(pSdb, pVgroup);
        TAOS_RETURN(code);
      }
      if ((code = sdbSetRawStatus(pVgRaw, SDB_STATUS_DROPPED)) != 0) {
        sdbCancelFetch(pSdb, pIter);
        sdbRelease(pSdb, pVgroup);
        TAOS_RETURN(code);
      }
    }

    sdbRelease(pSdb, pVgroup);
  }

  while (1) {
    SStbObj *pStb = NULL;
    ESdbStatus status;
    pIter = sdbFetchAll(pSdb, SDB_STB, pIter, (void **)&pStb, &status, true);
    if (pIter == NULL) break;

    if (pStb->dbUid == pDb->uid) {
      SSdbRaw *pStbRaw = mndStbActionEncode(pStb);
      if (pStbRaw == NULL) {
        sdbCancelFetch(pSdb, pIter);
        sdbRelease(pSdb, pStbRaw);
        code = TSDB_CODE_MND_RETURN_VALUE_NULL;
        if (terrno != 0) code = terrno;
        return -1;
      }
      if ((code = mndTransAppendCommitlog(pTrans, pStbRaw)) != 0) {
        sdbCancelFetch(pSdb, pIter);
        sdbRelease(pSdb, pStbRaw);
        return -1;
      }
      if ((code = sdbSetRawStatus(pStbRaw, SDB_STATUS_DROPPED)) != 0) {
        sdbCancelFetch(pSdb, pIter);
        sdbRelease(pSdb, pStbRaw);
        return -1;
      }
    }

    sdbRelease(pSdb, pStb);
  }

  TAOS_RETURN(code);
}

static int32_t mndBuildDropVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup) {
  int32_t code = 0;
  for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
    SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
    TAOS_CHECK_RETURN(mndAddDropVnodeAction(pMnode, pTrans, pDb, pVgroup, pVgid, true));
  }

  TAOS_RETURN(code);
}

int32_t mndSetDropDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
  int32_t code = 0;
  SSdb   *pSdb = pMnode->pSdb;
  void   *pIter = NULL;

  while (1) {
    SVgObj *pVgroup = NULL;
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
    if (pIter == NULL) break;

    if (pVgroup->dbUid == pDb->uid) {
      if ((code = mndBuildDropVgroupAction(pMnode, pTrans, pDb, pVgroup)) != 0) {
        sdbCancelFetch(pSdb, pIter);
        sdbRelease(pSdb, pVgroup);
        TAOS_RETURN(code);
      }
    }

    sdbRelease(pSdb, pVgroup);
  }

  TAOS_RETURN(code);
}

static int32_t mndBuildDropDbRsp(SDbObj *pDb, int32_t *pRspLen, void **ppRsp, bool useRpcMalloc) {
  int32_t    code = 0;
  SDropDbRsp dropRsp = {0};
  if (pDb != NULL) {
    (void)memcpy(dropRsp.db, pDb->name, TSDB_DB_FNAME_LEN);
    dropRsp.uid = pDb->uid;
  }

  int32_t rspLen = tSerializeSDropDbRsp(NULL, 0, &dropRsp);
  void   *pRsp = NULL;
  if (useRpcMalloc) {
    pRsp = rpcMallocCont(rspLen);
  } else {
    pRsp = taosMemoryMalloc(rspLen);
  }

  if (pRsp == NULL) {
    code = terrno;
    TAOS_RETURN(code);
  }

  int32_t ret = 0;
  if ((ret = tSerializeSDropDbRsp(pRsp, rspLen, &dropRsp)) < 0) return ret;
  *pRspLen = rspLen;
  *ppRsp = pRsp;
  TAOS_RETURN(code);
}

static int32_t mndDropDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb) {
  int32_t code = -1;

  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq, "drop-db");
  if (pTrans == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    goto _OVER;
  }

  mInfo("trans:%d start to drop db:%s", pTrans->id, pDb->name);

  mndTransSetDbName(pTrans, pDb->name, NULL);
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);

  if (mndTopicExistsForDb(pMnode, pDb)) {
    code = TSDB_CODE_MND_TOPIC_MUST_BE_DELETED;
    goto _OVER;
  }

  TAOS_CHECK_GOTO(mndSetDropDbPrepareLogs(pMnode, pTrans, pDb), NULL, _OVER);
  TAOS_CHECK_GOTO(mndSetDropDbCommitLogs(pMnode, pTrans, pDb), NULL, _OVER);
  TAOS_CHECK_GOTO(mndDropStreamByDb(pMnode, pTrans, pDb), NULL, _OVER);
#ifdef TD_ENTERPRISE
  TAOS_CHECK_GOTO(mndDropViewByDb(pMnode, pTrans, pDb), NULL, _OVER);
#endif
  TAOS_CHECK_GOTO(mndDropTSMAsByDb(pMnode, pTrans, pDb), NULL, _OVER);
  TAOS_CHECK_GOTO(mndDropRsmasByDb(pMnode, pTrans, pDb), NULL, _OVER);
  TAOS_CHECK_GOTO(mndDropIdxsByDb(pMnode, pTrans, pDb), NULL, _OVER);
  //TAOS_CHECK_GOTO(mndStreamSetStopStreamTasksActions(pMnode, pTrans, pDb->uid), NULL, _OVER);
  TAOS_CHECK_GOTO(mndSetDropDbRedoActions(pMnode, pTrans, pDb), NULL, _OVER);
  TAOS_CHECK_GOTO(mndPrincipalRemoveDb(pMnode, pTrans, pDb, NULL, NULL), NULL, _OVER);

  int32_t rspLen = 0;
  void   *pRsp = NULL;
  TAOS_CHECK_GOTO(mndBuildDropDbRsp(pDb, &rspLen, &pRsp, false), NULL, _OVER);
  mndTransSetRpcRsp(pTrans, pRsp, rspLen);

  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
  code = 0;

_OVER:
  mndTransDrop(pTrans);
  TAOS_RETURN(code);
}

static int32_t mndProcessDropDbReq(SRpcMsg *pReq) {
  SMnode    *pMnode = pReq->info.node;
  int32_t    code = -1;
  SDbObj    *pDb = NULL;
  SDropDbReq dropReq = {0};
  int64_t    tss = taosGetTimestampMs();

  TAOS_CHECK_GOTO(tDeserializeSDropDbReq(pReq->pCont, pReq->contLen, &dropReq), NULL, _OVER);

  mInfo("db:%s, start to drop", dropReq.db);

  pDb = mndAcquireDb(pMnode, dropReq.db);
  if (pDb == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    int32_t ret =
        mndCheckDbPrivilegeByName(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_DROP_DB, dropReq.db, true);
    if (ret != 0) {
      code = ret;
      goto _OVER;
    }
    if (dropReq.ignoreNotExists) {
      code = mndBuildDropDbRsp(pDb, &pReq->info.rspLen, &pReq->info.rsp, true);
    }
    goto _OVER;
  }

  if ((code = mndCheckDbPrivilege(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_DROP_DB, pDb))) {
    goto _OVER;
  }

  if(pDb->cfg.isMount) {
    code = TSDB_CODE_MND_MOUNT_OBJ_NOT_SUPPORT;
    goto _OVER;
  }

  if(!pDb->cfg.allowDrop) {
    code = TSDB_CODE_MND_DB_DROP_NOT_ALLOWED;
    goto _OVER;
  }

  SSdb *pSdb = pMnode->pSdb;
  void *pIter = NULL;

  while (1) {
    SVgObj *pVgroup = NULL;
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
    if (pIter == NULL) break;

    if (pVgroup->dbUid == pDb->uid) {
      bool isFound = false;
      for (int32_t i = 0; i < pVgroup->replica; i++) {
        if (pVgroup->vnodeGid[i].syncState == TAOS_SYNC_STATE_OFFLINE) {
          isFound = true;
          break;
        }
      }
      if (!isFound) {
        sdbRelease(pSdb, pVgroup);
        continue;
      }
      code = TSDB_CODE_MND_VGROUP_OFFLINE;
      sdbCancelFetch(pSdb, pIter);
      sdbRelease(pSdb, pVgroup);
      goto _OVER;
    }

    sdbRelease(pSdb, pVgroup);
  }

  bool dbStream = false;
  bool vtableStream = false;
  mstCheckDbInUse(pMnode, dropReq.db, &dbStream, &vtableStream, true);
  if (dbStream) {
    code = TSDB_CODE_MND_STREAM_DB_IN_USE;
    mError("db:%s used by streams, drop db not allowed", dropReq.db);
    goto _OVER;
  }

  if (vtableStream && !dropReq.force) {
    code = TSDB_CODE_MND_STREAM_VTABLE_EXITS;
    mError("db:%s, vtable stream exists, drop db not allowed", dropReq.db);
    goto _OVER;
  }

  code = mndDropDb(pMnode, pReq, pDb);
  if (code == TSDB_CODE_SUCCESS) {
    code = TSDB_CODE_ACTION_IN_PROGRESS;
  }

  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {
    SName name = {0};
    if (tNameFromString(&name, dropReq.db, T_NAME_ACCT | T_NAME_DB) < 0)
      mError("db:%s, failed to parse db name", dropReq.db);

    int64_t tse = taosGetTimestampMs();
    double  duration = (double)(tse - tss);
    duration = duration / 1000;
    auditRecord(pReq, pMnode->clusterId, "dropDB", name.dbname, "", dropReq.sql, dropReq.sqlLen, duration, 0);
  }

_OVER:
  if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_ACTION_IN_PROGRESS) {
    mError("db:%s, failed to drop since %s", dropReq.db, terrstr());
  }

  mndReleaseDb(pMnode, pDb);
  tFreeSDropDbReq(&dropReq);
  TAOS_RETURN(code);
}

static int32_t mndGetDBTableNum(SDbObj *pDb, SMnode *pMnode) {
  int32_t numOfTables = 0;
  int32_t vindex = 0;
  SSdb   *pSdb = pMnode->pSdb;

  void *pIter = NULL;
  while (vindex < pDb->cfg.numOfVgroups) {
    SVgObj *pVgroup = NULL;
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
    if (pIter == NULL) break;

    if (mndVgroupInDb(pVgroup, pDb->uid)) {
      numOfTables += pVgroup->numOfTables / TSDB_TABLE_NUM_UNIT;
      vindex++;
    }

    sdbRelease(pSdb, pVgroup);
  }

  sdbCancelFetch(pSdb, pIter);
  return numOfTables;
}

void mndBuildDBVgroupInfo(SDbObj *pDb, SMnode *pMnode, SArray *pVgList) {
  int32_t vindex = 0;
  SSdb   *pSdb = pMnode->pSdb;

  void *pIter = NULL;
  while (1) {
    SVgObj *pVgroup = NULL;
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
    if (pIter == NULL) break;

    if ((NULL == pDb || pVgroup->dbUid == pDb->uid) && !pVgroup->isTsma) {
      SVgroupInfo vgInfo = {0};
      vgInfo.vgId = pVgroup->vgId;
      vgInfo.hashBegin = pVgroup->hashBegin;
      vgInfo.hashEnd = pVgroup->hashEnd;
      vgInfo.numOfTable = pVgroup->numOfTables / TSDB_TABLE_NUM_UNIT;
      vgInfo.epSet.numOfEps = pVgroup->replica;
      for (int32_t gid = 0; gid < pVgroup->replica; ++gid) {
        SVnodeGid *pVgid = &pVgroup->vnodeGid[gid];
        SEp       *pEp = &vgInfo.epSet.eps[gid];
        SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
        if (pDnode != NULL) {
          (void)memcpy(pEp->fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
          pEp->port = pDnode->port;
        }
        mndReleaseDnode(pMnode, pDnode);
        if (pVgid->syncState == TAOS_SYNC_STATE_LEADER || pVgid->syncState == TAOS_SYNC_STATE_ASSIGNED_LEADER) {
          vgInfo.epSet.inUse = gid;
        }
      }
      vindex++;
      if (taosArrayPush(pVgList, &vgInfo) == NULL) {
        mError("db:%s, failed to push vgInfo to array, vgId:%d, but continue next", pDb->name, vgInfo.vgId);
      }
    }

    sdbRelease(pSdb, pVgroup);

    if (pDb && (vindex >= pDb->cfg.numOfVgroups)) {
      sdbCancelFetch(pSdb, pIter);
      break;
    }
  }
}

int32_t mndExtractDbInfo(SMnode *pMnode, SDbObj *pDb, SUseDbRsp *pRsp, const SUseDbReq *pReq) {
  int32_t code = 0;
  pRsp->pVgroupInfos = taosArrayInit(pDb->cfg.numOfVgroups, sizeof(SVgroupInfo));
  if (pRsp->pVgroupInfos == NULL) {
    code = terrno;
    TAOS_RETURN(code);
  }

  int32_t numOfTable = mndGetDBTableNum(pDb, pMnode);

  if (pReq == NULL || pReq->vgVersion < pDb->vgVersion || pReq->dbId != pDb->uid || numOfTable != pReq->numOfTable ||
      pReq->stateTs < pDb->stateTs) {
    mndBuildDBVgroupInfo(pDb, pMnode, pRsp->pVgroupInfos);
  }

  (void)memcpy(pRsp->db, pDb->name, TSDB_DB_FNAME_LEN);
  pRsp->uid = pDb->uid;
  pRsp->vgVersion = pDb->vgVersion;
  pRsp->stateTs = pDb->stateTs;
  pRsp->vgNum = taosArrayGetSize(pRsp->pVgroupInfos);
  pRsp->hashMethod = pDb->cfg.hashMethod;
  pRsp->hashPrefix = pDb->cfg.hashPrefix;
  pRsp->hashSuffix = pDb->cfg.hashSuffix;
  pRsp->flags = pDb->cfg.flags;
  TAOS_RETURN(code);
}

static int32_t mndProcessUseDbReq(SRpcMsg *pReq) {
  SMnode   *pMnode = pReq->info.node;
  int32_t   code = -1;
  SDbObj   *pDb = NULL;
  SUseDbReq usedbReq = {0};
  SUseDbRsp usedbRsp = {0};

  TAOS_CHECK_GOTO(tDeserializeSUseDbReq(pReq->pCont, pReq->contLen, &usedbReq), NULL, _OVER);

  char *p = strchr(usedbReq.db, '.');
  if (p && ((0 == strcmp(p + 1, TSDB_INFORMATION_SCHEMA_DB) || (0 == strcmp(p + 1, TSDB_PERFORMANCE_SCHEMA_DB))))) {
    (void)memcpy(usedbRsp.db, usedbReq.db, TSDB_DB_FNAME_LEN);
    int32_t vgVersion = mndGetGlobalVgroupVersion(pMnode);
    if (usedbReq.vgVersion < vgVersion) {
      usedbRsp.pVgroupInfos = taosArrayInit(10, sizeof(SVgroupInfo));
      if (usedbRsp.pVgroupInfos == NULL) goto _OVER;

      mndBuildDBVgroupInfo(NULL, pMnode, usedbRsp.pVgroupInfos);
      usedbRsp.vgVersion = vgVersion++;
    } else {
      usedbRsp.vgVersion = usedbReq.vgVersion;
    }
    usedbRsp.vgNum = taosArrayGetSize(usedbRsp.pVgroupInfos);
    code = 0;
  } else {
    pDb = mndAcquireDb(pMnode, usedbReq.db);
    if (pDb == NULL) {
      (void)memcpy(usedbRsp.db, usedbReq.db, TSDB_DB_FNAME_LEN);
      usedbRsp.uid = usedbReq.dbId;
      usedbRsp.vgVersion = usedbReq.vgVersion;
      usedbRsp.errCode = terrno;

      code = TSDB_CODE_MND_RETURN_VALUE_NULL;
      if (terrno != 0) code = terrno;
      goto _OVER;
    } else {
      TAOS_CHECK_GOTO(mndCheckDbPrivilege(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_USE_DB, pDb), NULL, _OVER);

      TAOS_CHECK_GOTO(mndExtractDbInfo(pMnode, pDb, &usedbRsp, &usedbReq), NULL, _OVER);

      mDebug("db:%s, process usedb req vgVersion:%d stateTs:%" PRId64 ", rsp vgVersion:%d stateTs:%" PRId64,
             usedbReq.db, usedbReq.vgVersion, usedbReq.stateTs, usedbRsp.vgVersion, usedbRsp.stateTs);
      code = 0;
    }
  }

  int32_t contLen = tSerializeSUseDbRsp(NULL, 0, &usedbRsp);
  void   *pRsp = rpcMallocCont(contLen);
  if (pRsp == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    code = -1;
    goto _OVER;
  }

  int32_t ret = 0;
  if ((ret = tSerializeSUseDbRsp(pRsp, contLen, &usedbRsp)) < 0) {
    code = ret;
    goto _OVER;
  }

  pReq->info.rsp = pRsp;
  pReq->info.rspLen = contLen;

_OVER:
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
    mError("db:%s, failed to process use db req since %s", usedbReq.db, tstrerror(code));
  }

  mndReleaseDb(pMnode, pDb);
  tFreeSUsedbRsp(&usedbRsp);

  TAOS_RETURN(code);
}

int32_t mndValidateDbInfo(SMnode *pMnode, SDbCacheInfo *pDbs, int32_t numOfDbs, void **ppRsp, int32_t *pRspLen) {
  int32_t       code = 0;
  SDbHbBatchRsp batchRsp = {0};
  batchRsp.pArray = taosArrayInit(numOfDbs, sizeof(SDbHbRsp));
  if (batchRsp.pArray == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    TAOS_RETURN(code);
  }

  for (int32_t i = 0; i < numOfDbs; ++i) {
    SDbCacheInfo *pDbCacheInfo = &pDbs[i];
    pDbCacheInfo->dbId = be64toh(pDbCacheInfo->dbId);
    pDbCacheInfo->vgVersion = htonl(pDbCacheInfo->vgVersion);
    pDbCacheInfo->cfgVersion = htonl(pDbCacheInfo->cfgVersion);
    pDbCacheInfo->numOfTable = htonl(pDbCacheInfo->numOfTable);
    pDbCacheInfo->stateTs = be64toh(pDbCacheInfo->stateTs);
    pDbCacheInfo->tsmaVersion = htonl(pDbCacheInfo->tsmaVersion);

    SDbHbRsp rsp = {0};
    (void)memcpy(rsp.db, pDbCacheInfo->dbFName, TSDB_DB_FNAME_LEN);
    rsp.dbId = pDbCacheInfo->dbId;

    if ((0 == strcasecmp(pDbCacheInfo->dbFName, TSDB_INFORMATION_SCHEMA_DB) ||
         (0 == strcasecmp(pDbCacheInfo->dbFName, TSDB_PERFORMANCE_SCHEMA_DB)))) {
      int32_t vgVersion = mndGetGlobalVgroupVersion(pMnode);
      if (pDbCacheInfo->vgVersion >= vgVersion) {
        continue;
      }

      rsp.useDbRsp = taosMemoryCalloc(1, sizeof(SUseDbRsp));
      (void)memcpy(rsp.useDbRsp->db, pDbCacheInfo->dbFName, TSDB_DB_FNAME_LEN);
      rsp.useDbRsp->pVgroupInfos = taosArrayInit(10, sizeof(SVgroupInfo));

      mndBuildDBVgroupInfo(NULL, pMnode, rsp.useDbRsp->pVgroupInfos);
      rsp.useDbRsp->vgVersion = vgVersion++;

      rsp.useDbRsp->vgNum = taosArrayGetSize(rsp.useDbRsp->pVgroupInfos);

      if (taosArrayPush(batchRsp.pArray, &rsp) == NULL) {
        if (terrno != 0) code = terrno;
        return code;
      }

      continue;
    }

    SDbObj *pDb = mndAcquireDb(pMnode, pDbCacheInfo->dbFName);
    if (pDb == NULL) {
      mTrace("db:%s, no exist", pDbCacheInfo->dbFName);
      rsp.useDbRsp = taosMemoryCalloc(1, sizeof(SUseDbRsp));
      (void)memcpy(rsp.useDbRsp->db, pDbCacheInfo->dbFName, TSDB_DB_FNAME_LEN);
      rsp.useDbRsp->uid = pDbCacheInfo->dbId;
      rsp.useDbRsp->vgVersion = -1;
      if (taosArrayPush(batchRsp.pArray, &rsp) == NULL) {
        if (terrno != 0) code = terrno;
        return code;
      }
      continue;
    }

    int32_t numOfTable = mndGetDBTableNum(pDb, pMnode);

    if (pDbCacheInfo->vgVersion >= pDb->vgVersion && pDbCacheInfo->cfgVersion >= pDb->cfgVersion &&
        numOfTable == pDbCacheInfo->numOfTable && pDbCacheInfo->stateTs == pDb->stateTs &&
        pDbCacheInfo->tsmaVersion >= pDb->tsmaVersion) {
      mTrace("db:%s, valid dbinfo, vgVersion:%d cfgVersion:%d stateTs:%" PRId64
             " numOfTables:%d, not changed vgVersion:%d cfgVersion:%d stateTs:%" PRId64 " numOfTables:%d",
             pDbCacheInfo->dbFName, pDbCacheInfo->vgVersion, pDbCacheInfo->cfgVersion, pDbCacheInfo->stateTs,
             pDbCacheInfo->numOfTable, pDb->vgVersion, pDb->cfgVersion, pDb->stateTs, numOfTable);
      mndReleaseDb(pMnode, pDb);
      continue;
    } else {
      mTrace("db:%s, valid dbinfo, vgVersion:%d cfgVersion:%d stateTs:%" PRId64
             " numOfTables:%d, changed to vgVersion:%d cfgVersion:%d stateTs:%" PRId64 " numOfTables:%d",
             pDbCacheInfo->dbFName, pDbCacheInfo->vgVersion, pDbCacheInfo->cfgVersion, pDbCacheInfo->stateTs,
             pDbCacheInfo->numOfTable, pDb->vgVersion, pDb->cfgVersion, pDb->stateTs, numOfTable);
    }

    if (pDbCacheInfo->cfgVersion < pDb->cfgVersion) {
      rsp.cfgRsp = taosMemoryCalloc(1, sizeof(SDbCfgRsp));
      char algrId[TSDB_ENCRYPT_ALGR_NAME_LEN] = {0};
      mndGetEncryptOsslAlgrNameById(pMnode, pDb->cfg.encryptAlgorithm, algrId);
      mndDumpDbCfgInfo(rsp.cfgRsp, pDb, algrId);
    }

    if (pDbCacheInfo->tsmaVersion != pDb->tsmaVersion) {
      rsp.pTsmaRsp = taosMemoryCalloc(1, sizeof(STableTSMAInfoRsp));
      if (rsp.pTsmaRsp) rsp.pTsmaRsp->pTsmas = taosArrayInit(4, POINTER_BYTES);
      if (rsp.pTsmaRsp && rsp.pTsmaRsp->pTsmas) {
        bool    exist = false;
        int32_t code = mndGetDbTsmas(pMnode, 0, pDb->uid, rsp.pTsmaRsp, &exist);
        if (TSDB_CODE_SUCCESS != code) {
          mndReleaseDb(pMnode, pDb);
          if (code != TSDB_CODE_NEED_RETRY) {
            mError("db:%s, failed to get db tsmas", pDb->name);
          } else {
            mWarn("db:%s, need retry to get db tsmas", pDb->name);
          }
          taosArrayDestroyP(rsp.pTsmaRsp->pTsmas, tFreeAndClearTableTSMAInfo);
          taosMemoryFreeClear(rsp.pTsmaRsp);
          continue;
        }
        rsp.dbTsmaVersion = pDb->tsmaVersion;
        mDebug("update tsma version to %d, got tsma num: %ld", pDb->tsmaVersion, rsp.pTsmaRsp->pTsmas->size);
      }
    }

    if (pDbCacheInfo->vgVersion < pDb->vgVersion || numOfTable != pDbCacheInfo->numOfTable ||
        pDbCacheInfo->stateTs != pDb->stateTs) {
      rsp.useDbRsp = taosMemoryCalloc(1, sizeof(SUseDbRsp));
      rsp.useDbRsp->pVgroupInfos = taosArrayInit(pDb->cfg.numOfVgroups, sizeof(SVgroupInfo));
      if (rsp.useDbRsp->pVgroupInfos == NULL) {
        mndReleaseDb(pMnode, pDb);
        mError("db:%s, failed to malloc usedb response", pDb->name);
        taosArrayDestroyP(rsp.pTsmaRsp->pTsmas, tFreeAndClearTableTSMAInfo);
        taosMemoryFreeClear(rsp.pTsmaRsp);
        continue;
      }

      mndBuildDBVgroupInfo(pDb, pMnode, rsp.useDbRsp->pVgroupInfos);
      (void)memcpy(rsp.useDbRsp->db, pDb->name, TSDB_DB_FNAME_LEN);
      rsp.useDbRsp->uid = pDb->uid;
      rsp.useDbRsp->vgVersion = pDb->vgVersion;
      rsp.useDbRsp->stateTs = pDb->stateTs;
      rsp.useDbRsp->vgNum = (int32_t)taosArrayGetSize(rsp.useDbRsp->pVgroupInfos);
      rsp.useDbRsp->hashMethod = pDb->cfg.hashMethod;
      rsp.useDbRsp->hashPrefix = pDb->cfg.hashPrefix;
      rsp.useDbRsp->hashSuffix = pDb->cfg.hashSuffix;
      rsp.useDbRsp->flags = pDb->cfg.flags;
    }

    if (taosArrayPush(batchRsp.pArray, &rsp) == NULL) {
      mndReleaseDb(pMnode, pDb);
      if (terrno != 0) code = terrno;
      return code;
    }
    mndReleaseDb(pMnode, pDb);
  }

  int32_t rspLen = tSerializeSDbHbBatchRsp(NULL, 0, &batchRsp);
  void   *pRsp = taosMemoryMalloc(rspLen);
  if (pRsp == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    tFreeSDbHbBatchRsp(&batchRsp);
    return -1;
  }
  int32_t ret = 0;
  if ((ret = tSerializeSDbHbBatchRsp(pRsp, rspLen, &batchRsp)) < 0) return ret;

  *ppRsp = pRsp;
  *pRspLen = rspLen;

  tFreeSDbHbBatchRsp(&batchRsp);
  TAOS_RETURN(code);
}

static int32_t mndSetTrimDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, int64_t startTs, STimeWindow tw,
                                       SArray *vgroupIds, ETsdbOpType type, ETriggerType triggerType,
                                       STrimDbRsp *pRsp) {
  int32_t code = 0, lino = 0;
  SSdb   *pSdb = pMnode->pSdb;
  void   *pIter = NULL;

  SRetentionObj obj = {.optrType = type, .triggerType = triggerType};  // reuse SCompactObj struct
  TAOS_CHECK_EXIT(mndAddRetentionToTrans(pMnode, pTrans, &obj, pDb, pRsp));

  int32_t j = 0;
  int32_t numOfVgroups = taosArrayGetSize(vgroupIds);
  if (numOfVgroups > 0) {
    for (int32_t i = 0; i < numOfVgroups; i++) {
      int64_t vgId = *(int64_t *)taosArrayGet(vgroupIds, i);
      SVgObj *pVgroup = mndAcquireVgroup(pMnode, vgId);

      if (pVgroup == NULL) {
        mError("db:%s, vgroup:%" PRId64 " not exist", pDb->name, vgId);
        TAOS_CHECK_EXIT(TSDB_CODE_MND_VGROUP_NOT_EXIST);
      } else if (pVgroup->dbUid != pDb->uid) {
        mError("db:%s, vgroup:%" PRId64 " not belong to db:%s", pDb->name, vgId, pDb->name);
        sdbRelease(pSdb, pVgroup);
        TAOS_CHECK_EXIT(TSDB_CODE_MND_VGROUP_NOT_EXIST);
      }
      sdbRelease(pSdb, pVgroup);
    }

    for (int32_t i = 0; i < numOfVgroups; i++) {
      int64_t vgId = *(int64_t *)taosArrayGet(vgroupIds, i);
      SVgObj *pVgroup = mndAcquireVgroup(pMnode, vgId);

      if ((code = mndBuildTrimVgroupAction(pMnode, pTrans, pDb, pVgroup, startTs, tw, type, triggerType)) != 0) {
        sdbRelease(pSdb, pVgroup);
        TAOS_RETURN(code);
      }

      for (int32_t i = 0; i < pVgroup->replica; i++) {
        SVnodeGid *gid = &pVgroup->vnodeGid[i];
        if ((code = mndAddRetentionDetailToTrans(pMnode, pTrans, &obj, pVgroup, gid, j)) != 0) {
          sdbRelease(pSdb, pVgroup);
          TAOS_RETURN(code);
        }
        j++;
      }
      sdbRelease(pSdb, pVgroup);
    }
  } else {
    while (1) {
      SVgObj *pVgroup = NULL;
      pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
      if (pIter == NULL) break;

      if (pVgroup->dbUid == pDb->uid) {
        if ((code = mndBuildTrimVgroupAction(pMnode, pTrans, pDb, pVgroup, startTs, tw, type, triggerType)) != 0) {
          sdbCancelFetch(pSdb, pIter);
          sdbRelease(pSdb, pVgroup);
          TAOS_RETURN(code);
        }

        for (int32_t i = 0; i < pVgroup->replica; i++) {
          SVnodeGid *gid = &pVgroup->vnodeGid[i];
          if ((code = mndAddRetentionDetailToTrans(pMnode, pTrans, &obj, pVgroup, gid, j)) != 0) {
            sdbCancelFetch(pSdb, pIter);
            sdbRelease(pSdb, pVgroup);
            TAOS_RETURN(code);
          }
          j++;
        }
      }

      sdbRelease(pSdb, pVgroup);
    }
  }
_exit:
  TAOS_RETURN(code);
}

static int32_t mndBuildTrimDbRsp(STrimDbRsp *rsp, int32_t *pRspLen, void **ppRsp, bool useRpcMalloc) {
  int32_t code = 0;
  int32_t rspLen = tSerializeSCompactDbRsp(NULL, 0, (SCompactDbRsp *)rsp);
  void   *pRsp = NULL;
  if (useRpcMalloc) {
    pRsp = rpcMallocCont(rspLen);
  } else {
    pRsp = taosMemoryMalloc(rspLen);
  }

  if (pRsp == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    TAOS_RETURN(code);
  }

  (void)tSerializeSCompactDbRsp(pRsp, rspLen, (SCompactDbRsp *)rsp);
  *pRspLen = rspLen;
  *ppRsp = pRsp;
  TAOS_RETURN(code);
}

int32_t mndTrimDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, STimeWindow tw, SArray *vgroupIds,
                         ETsdbOpType type, ETriggerType triggerType) {
  SSdb      *pSdb = pMnode->pSdb;
  SVgObj    *pVgroup = NULL;
  void      *pIter = NULL;
  int32_t    code = 0, lino = 0;
  STrimDbRsp rsp = {0};
  bool       isExist = false;

  while (1) {
    SRetentionObj *pObj = NULL;
    pIter = sdbFetch(pMnode->pSdb, SDB_RETENTION, pIter, (void **)&pObj);
    if (pIter == NULL) break;

    if (strcmp(pObj->dbname, pDb->name) == 0) {
      isExist = true;
    }
    sdbRelease(pMnode->pSdb, pObj);
  }
  if (isExist) {
    mInfo("trim db:%s already exist", pDb->name);
    TAOS_RETURN(TSDB_CODE_MND_TRIM_ALREADY_EXIST);
  }

  int64_t startTs = taosGetTimestampMs();
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq, "trim-db");
  if (pTrans == NULL) goto _exit;
  mInfo("trans:%d, used to trim db:%s", pTrans->id, pDb->name);
  mndTransSetDbName(pTrans, pDb->name, NULL);
  TAOS_CHECK_EXIT(mndTrancCheckConflict(pMnode, pTrans));
  TAOS_CHECK_EXIT(mndSetTrimDbRedoActions(pMnode, pTrans, pDb, startTs, tw, vgroupIds, type, triggerType, &rsp));
  if (pReq) {
    int32_t rspLen = 0;
    void   *pRsp = NULL;
    rsp.bAccepted = true;
    TAOS_CHECK_EXIT(mndBuildTrimDbRsp(&rsp, &rspLen, &pRsp, false));
    mndTransSetRpcRsp(pTrans, pRsp, rspLen);
  }
  TAOS_CHECK_EXIT(mndTransPrepare(pMnode, pTrans));
_exit:
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
    mError("db:%s, failed at line %d to trim db since %s", pDb->name, lino, tstrerror(code));
  }
  mndTransDrop(pTrans);
  TAOS_RETURN(code);
}

#ifdef TD_ENTERPRISE
static int32_t mndCheckRsmaInDb(SMnode *pMnode, SDbObj *pDb) {
  int32_t   code = 0;
  void     *pIter = NULL;
  SRsmaObj *pRsma = NULL;
  while ((pIter = sdbFetch(pMnode->pSdb, SDB_RSMA, pIter, (void **)&pRsma))) {
    if (pRsma->dbUid == pDb->uid) {
      sdbRelease(pMnode->pSdb, pRsma);
      sdbCancelFetch(pMnode->pSdb, pIter);
      TAOS_RETURN(code);
    }
    sdbRelease(pMnode->pSdb, pRsma);
  }
  TAOS_RETURN(TSDB_CODE_RSMA_NOT_EXIST);
}
#endif

static int32_t mndProcessTrimDbReq(SRpcMsg *pReq) {
  SMnode    *pMnode = pReq->info.node;
  int32_t    code = 0, lino = 0;
  SDbObj    *pDb = NULL;
  STrimDbReq trimReq = {0};
  int64_t    tss = taosGetTimestampMs();

  TAOS_CHECK_EXIT(tDeserializeSTrimDbReq(pReq->pCont, pReq->contLen, &trimReq));

  mInfo("db:%s, start to trim, optr:%u, tw:%" PRId64 ",%" PRId64, trimReq.db, trimReq.optrType, trimReq.tw.skey,
        trimReq.tw.ekey);

  pDb = mndAcquireDb(pMnode, trimReq.db);
  if (pDb == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    TAOS_CHECK_EXIT(code);
  }

  TAOS_CHECK_EXIT(mndCheckDbPrivilege(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq),
                                      trimReq.optrType == TSDB_OPTR_ROLLUP ? MND_OPER_ROLLUP_DB : MND_OPER_TRIM_DB,
                                      pDb));

  if (pDb->cfg.isMount) {
    TAOS_CHECK_EXIT(TSDB_CODE_MND_MOUNT_OBJ_NOT_SUPPORT);
  }

#ifdef TD_ENTERPRISE
  if (trimReq.optrType == TSDB_OPTR_ROLLUP) {
    TAOS_CHECK_EXIT(mndCheckRsmaInDb(pMnode, pDb));
  }
#endif

  TAOS_CHECK_EXIT(mndTrimDb(pMnode, pReq, pDb, trimReq.tw, trimReq.vgroupIds, trimReq.optrType, trimReq.triggerType));
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;

  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {
    SName name = {0};
    if (tNameFromString(&name, trimReq.db, T_NAME_ACCT | T_NAME_DB) < 0) {
      mWarn("db:%s, failed at line %d to parse db name", trimReq.db, __LINE__);
    }

    char optrType[16] = {0};
    (void)snprintf(optrType, sizeof(optrType), "%u", trimReq.optrType);
    int64_t tse = taosGetTimestampMs();
    double  duration = (double)(tse - tss);
    duration = duration / 1000;
    auditRecord(pReq, pMnode->clusterId, "trimDB", name.dbname, optrType, trimReq.sql, trimReq.sqlLen, duration, 0);
  }

_exit:
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
    mError("db:%s, failed at line %d to process trim db req since %s, optr:%u", trimReq.db, lino, tstrerror(code),
           trimReq.optrType);
  }

  mndReleaseDb(pMnode, pDb);
  tFreeSTrimDbReq(&trimReq);
  TAOS_RETURN(code);
}

static int32_t mndBuildSsMigrateDbRsp(SSsMigrateDbRsp *pSsMigrateRsp, int32_t *pRspLen, void **ppRsp, bool useRpcMalloc) {
  int32_t code = 0;
  int32_t rspLen = tSerializeSSsMigrateDbRsp(NULL, 0, pSsMigrateRsp);
  void   *pRsp = NULL;
  if (useRpcMalloc) {
    pRsp = rpcMallocCont(rspLen);
  } else {
    pRsp = taosMemoryMalloc(rspLen);
  }

  if (pRsp == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    TAOS_RETURN(code);
  }

  (void)tSerializeSSsMigrateDbRsp(pRsp, rspLen, pSsMigrateRsp);
  *pRspLen = rspLen;
  *ppRsp = pRsp;
  TAOS_RETURN(code);
}

static int32_t mndTrimDbWal(SMnode *pMnode, SDbObj *pDb) {
  SSdb   *pSdb = pMnode->pSdb;
  SVgObj *pVgroup = NULL;
  void   *pIter = NULL;
  int32_t code = 0;

  // Iterate through all vgroups in this database
  while (1) {
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
    if (pIter == NULL) break;

    // Check if this vgroup belongs to the database
    if (pVgroup->dbUid != pDb->uid) {
      sdbRelease(pSdb, pVgroup);
      continue;
    }

    // Prepare message to send to vnode
    int32_t   contLen = sizeof(SMsgHead);
    SMsgHead *pHead = rpcMallocCont(contLen);
    if (pHead == NULL) {
      sdbCancelFetch(pSdb, pVgroup);
      sdbRelease(pSdb, pVgroup);
      code = TSDB_CODE_OUT_OF_MEMORY;
      break;
    }
    pHead->contLen = htonl(contLen);
    pHead->vgId = htonl(pVgroup->vgId);

    // Send TRIM WAL request to vnode
    SRpcMsg rpcMsg = {.msgType = TDMT_VND_TRIM_WAL, .pCont = pHead, .contLen = contLen};
    SEpSet  epSet = mndGetVgroupEpset(pMnode, pVgroup);
    code = tmsgSendReq(&epSet, &rpcMsg);
    if (code != 0) {
      mError("vgId:%d, failed to send vnode-trim-wal request since 0x%x", pVgroup->vgId, code);
    } else {
      mInfo("vgId:%d, send vnode-trim-wal request to vnode", pVgroup->vgId);
    }
    sdbRelease(pSdb, pVgroup);
  }

  return code;
}

static int32_t mndProcessTrimDbWalReq(SRpcMsg *pReq) {
  SMnode    *pMnode = pReq->info.node;
  int32_t    code = -1;
  SDbObj    *pDb = NULL;
  STrimDbReq trimReq = {0};

  TAOS_CHECK_GOTO(tDeserializeSTrimDbReq(pReq->pCont, pReq->contLen, &trimReq), NULL, _OVER);

  mInfo("db:%s, start to trim WAL", trimReq.db);

  pDb = mndAcquireDb(pMnode, trimReq.db);
  if (pDb == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    goto _OVER;
  }

  TAOS_CHECK_GOTO(mndCheckDbPrivilege(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_TRIM_DB, pDb), NULL, _OVER);

  code = mndTrimDbWal(pMnode, pDb);

_OVER:
  if (code != 0) {
    mError("db:%s, failed to process trim db wal req since %s", trimReq.db, terrstr());
  }

  mndReleaseDb(pMnode, pDb);
  TAOS_RETURN(code);
}

int32_t mndSsMigrateDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb) {
  int32_t       code = 0;
  SSsMigrateDbRsp ssMigrateRsp = {0};

  bool  isExist = false;
  void *pIter = NULL;
  while (1) {
    SSsMigrateObj *pSsMigrate = NULL;
    pIter = sdbFetch(pMnode->pSdb, SDB_SSMIGRATE, pIter, (void **)&pSsMigrate);
    if (pIter == NULL) break;

    if (strcmp(pSsMigrate->dbname, pDb->name) == 0) {
      isExist = true;
    }
    sdbRelease(pMnode->pSdb, pSsMigrate);
  }
  if (isExist) {
    mInfo("ssmigrate db:%s already exist", pDb->name);

    if (pReq) {
      int32_t rspLen = 0;
      void   *pRsp = NULL;
      ssMigrateRsp.ssMigrateId = 0;
      ssMigrateRsp.bAccepted = false;
      TAOS_CHECK_RETURN(mndBuildSsMigrateDbRsp(&ssMigrateRsp, &rspLen, &pRsp, true));

      pReq->info.rsp = pRsp;
      pReq->info.rspLen = rspLen;
    }

    return TSDB_CODE_MND_SSMIGRATE_ALREADY_EXIST;
  }

  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq, "ssmigrate-db");
  if (pTrans == NULL) {
    mError("failed to create ssmigrate-db trans since %s", terrstr());
    goto _OVER;
  }

  mInfo("trans:%d, used to ssmigrate db:%s", pTrans->id, pDb->name);
  mndTransSetDbName(pTrans, pDb->name, NULL);
  TAOS_CHECK_GOTO(mndTrancCheckConflict(pMnode, pTrans), NULL, _OVER);

  SSsMigrateObj ssMigrate = { .startTime = taosGetTimestampMs() };
  TAOS_CHECK_GOTO(mndAddSsMigrateToTran(pMnode, pTrans, &ssMigrate, pDb), NULL, _OVER);

  if (pReq) {
    int32_t rspLen = 0;
    void   *pRsp = NULL;
    ssMigrateRsp.ssMigrateId = ssMigrate.id;
    ssMigrateRsp.bAccepted = true;
    TAOS_CHECK_GOTO(mndBuildSsMigrateDbRsp(&ssMigrateRsp, &rspLen, &pRsp, false), NULL, _OVER);
    mndTransSetRpcRsp(pTrans, pRsp, rspLen);
  }

  if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
  code = 0;

_OVER:
  mndTransDrop(pTrans);
  TAOS_RETURN(code);
}


static int32_t mndProcessSsMigrateDbReq(SRpcMsg *pReq) {
  SMnode         *pMnode = pReq->info.node;
  int32_t         code = -1;
  SDbObj         *pDb = NULL;
  SSsMigrateDbReq ssMigrateReq = {0};

  if (!tsSsEnabled) {
    code = TSDB_CODE_OPS_NOT_SUPPORT;
    goto _OVER;
  }

  TAOS_CHECK_GOTO(grantCheck(TSDB_GRANT_SHARED_STORAGE), NULL, _OVER);
  TAOS_CHECK_GOTO(tDeserializeSSsMigrateDbReq(pReq->pCont, pReq->contLen, &ssMigrateReq), NULL, _OVER);

  mInfo("db:%s, start to ssmigrate", ssMigrateReq.db);

  pDb = mndAcquireDb(pMnode, ssMigrateReq.db);
  if (pDb == NULL) {
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
    if (terrno != 0) code = terrno;
    goto _OVER;
  }

  TAOS_CHECK_GOTO(mndCheckDbPrivilege(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_SSMIGRATE_DB, pDb), NULL, _OVER);

  if(pDb->cfg.isMount) {
    code = TSDB_CODE_MND_MOUNT_OBJ_NOT_SUPPORT;
    goto _OVER;
  }

  code = mndSsMigrateDb(pMnode, pReq, pDb);

_OVER:
  if (code != 0) {
    mError("db:%s, failed to process ssmigrate db req since %s", ssMigrateReq.db, tstrerror(code));
  }

  mndReleaseDb(pMnode, pDb);
  TAOS_RETURN(code);
}

const char *mndGetDbStr(const char *src) {
  char *pos = strstr(src, TS_PATH_DELIMITER);
  if (pos != NULL) ++pos;
  if (pos == NULL) return src;
  return pos;
}

const char *mndGetStableStr(const char *src) {
  char *pos = strstr(src, TS_PATH_DELIMITER);
  if (pos != NULL) ++pos;
  if (pos == NULL) return src;
  return mndGetDbStr(pos);
}

static int64_t getValOfDiffPrecision(int8_t unit, int64_t val) {
  int64_t v = 0;
  switch (unit) {
    case 's':
      v = val / 1000;
      break;
    case 'm':
      v = val / tsTickPerMin[TSDB_TIME_PRECISION_MILLI];
      break;
    case 'h':
      v = val / (tsTickPerMin[TSDB_TIME_PRECISION_MILLI] * 60);
      break;
    case 'd':
      v = val / (tsTickPerMin[TSDB_TIME_PRECISION_MILLI] * 24 * 60);
      break;
    case 'w':
      v = val / (tsTickPerMin[TSDB_TIME_PRECISION_MILLI] * 24 * 60 * 7);
      break;
    default:
      break;
  }

  return v;
}

static char *buildRetension(SArray *pRetension) {
  size_t size = taosArrayGetSize(pRetension);
  if (size == 0) {
    return NULL;
  }

  char       *p1 = taosMemoryCalloc(1, 100);
  SRetention *p = taosArrayGet(pRetension, 0);

  int32_t len = 2;

  int64_t v1 = getValOfDiffPrecision(p->freqUnit, p->freq);
  int64_t v2 = getValOfDiffPrecision(p->keepUnit, p->keep);
  len += snprintf(p1 + len, 100 - len, "%" PRId64 "%c:%" PRId64 "%c", v1, p->freqUnit, v2, p->keepUnit);

  if (size > 1) {
    len += snprintf(p1 + len, 100 - len, ",");
    p = taosArrayGet(pRetension, 1);

    v1 = getValOfDiffPrecision(p->freqUnit, p->freq);
    v2 = getValOfDiffPrecision(p->keepUnit, p->keep);
    len += snprintf(p1 + len, 100 - len, "%" PRId64 "%c:%" PRId64 "%c", v1, p->freqUnit, v2, p->keepUnit);
  }

  if (size > 2) {
    len += snprintf(p1 + len, 100 - len, ",");
    p = taosArrayGet(pRetension, 2);

    v1 = getValOfDiffPrecision(p->freqUnit, p->freq);
    v2 = getValOfDiffPrecision(p->keepUnit, p->keep);
    len += snprintf(p1 + len, 100 - len, "%" PRId64 "%c:%" PRId64 "%c", v1, p->freqUnit, v2, p->keepUnit);
  }

  varDataSetLen(p1, len);
  return p1;
}

static const char *getCacheModelStr(int8_t cacheModel) {
  switch (cacheModel) {
    case TSDB_CACHE_MODEL_NONE:
      return TSDB_CACHE_MODEL_NONE_STR;
    case TSDB_CACHE_MODEL_LAST_ROW:
      return TSDB_CACHE_MODEL_LAST_ROW_STR;
    case TSDB_CACHE_MODEL_LAST_VALUE:
      return TSDB_CACHE_MODEL_LAST_VALUE_STR;
    case TSDB_CACHE_MODEL_BOTH:
      return TSDB_CACHE_MODEL_BOTH_STR;
    default:
      break;
  }
  return "unknown";
}

static void getEncryptAlgorithmStr(SMnode *pMnode, int8_t encryptAlgorithm, char *out) {
  switch (encryptAlgorithm) {
    case TSDB_ENCRYPT_ALGO_NONE: {
      tstrncpy(out, TSDB_ENCRYPT_ALGO_NONE_STR, TSDB_ENCRYPT_ALGR_NAME_LEN);
      break;
    }
    default: {
      SEncryptAlgrObj *obj = mndAcquireEncryptAlgrById(pMnode, encryptAlgorithm);
      if (obj != NULL) {
        tstrncpy(out, obj->algorithm_id, TSDB_ENCRYPT_ALGR_NAME_LEN);
        mndReleaseEncryptAlgr(pMnode, obj);
      }
      break;
    }
  }
}

bool mndIsDbReady(SMnode *pMnode, SDbObj *pDb) {
  if (pDb->cfg.replications == 1) return true;

  SSdb *pSdb = pMnode->pSdb;
  void *pIter = NULL;
  bool  isReady = true;
  while (1) {
    SVgObj *pVgroup = NULL;
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
    if (pIter == NULL) break;

    if (pVgroup->dbUid == pDb->uid && pVgroup->replica > 1) {
      bool hasLeader = false;
      for (int32_t i = 0; i < pVgroup->replica; ++i) {
        if (pVgroup->vnodeGid[i].syncState == TAOS_SYNC_STATE_LEADER ||
            pVgroup->vnodeGid[i].syncState == TAOS_SYNC_STATE_ASSIGNED_LEADER) {
          hasLeader = true;
        }
      }
      if (!hasLeader) isReady = false;
    }
    sdbRelease(pSdb, pVgroup);
  }

  return isReady;
}

static void mndDumpDbInfoData(SMnode *pMnode, SSDataBlock *pBlock, SDbObj *pDb, SShowObj *pShow, int32_t rows,
                              int64_t numOfTables, bool sysDb, ESdbStatus objStatus, bool sysinfo,
                              SSHashObj **ppUidNames) {
  int32_t cols = 0;
  int32_t bytes = pShow->pMeta->pSchemas[cols].bytes;
  char   *buf = taosMemoryMalloc(bytes);
  if (buf == NULL) {
    mError("db:%s, failed to malloc buffer", pDb->name);
    return;
  }
  int32_t code = 0;
  int32_t lino = 0;

  const char *name = mndGetDbStr(pDb->name);
  if (name != NULL) {
    STR_WITH_MAXSIZE_TO_VARSTR(buf, name, bytes);
  } else {
    STR_WITH_MAXSIZE_TO_VARSTR(buf, "NULL", bytes);
  }

  const char *precStr = NULL;
  switch (pDb->cfg.precision) {
    case TSDB_TIME_PRECISION_MILLI:
      precStr = TSDB_TIME_PRECISION_MILLI_STR;
      break;
    case TSDB_TIME_PRECISION_MICRO:
      precStr = TSDB_TIME_PRECISION_MICRO_STR;
      break;
    case TSDB_TIME_PRECISION_NANO:
      precStr = TSDB_TIME_PRECISION_NANO_STR;
      break;
    default:
      precStr = "none";
      break;
  }
  char precVstr[10] = {0};
  STR_WITH_MAXSIZE_TO_VARSTR(precVstr, precStr, 10);

  char *statusStr = "ready";
  if (objStatus == SDB_STATUS_CREATING) {
    statusStr = "creating";
  } else if (objStatus == SDB_STATUS_DROPPING) {
    statusStr = "dropping";
  } else {
    if (!sysDb && !mndIsDbReady(pMnode, pDb)) {
      statusStr = "unsynced";
    }
  }
  char statusVstr[24] = {0};
  STR_WITH_MAXSIZE_TO_VARSTR(statusVstr, statusStr, 24);

  if (sysDb || !sysinfo) {
    for (int32_t i = 0; i < pShow->numOfColumns; ++i) {
      SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, i);
      if (i == 0) {
        TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, buf, false), &lino, _OVER);
      } else if (i == 1) {
        TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->createdTime, false), &lino, _OVER);
      } else if (i == 3) {
        TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&numOfTables, false), &lino, _OVER);
      } else if (i == 14) {
        TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, precVstr, false), &lino, _OVER);
      } else if (i == 15) {
        TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, statusVstr, false), &lino, _OVER);
      } else if (i == 28) {
        TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.keepTimeOffset, false), &lino, _OVER);
      } else {
        colDataSetNULL(pColInfo, rows);
      }
    }
  } else {
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, buf, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->createdTime, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.numOfVgroups, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&numOfTables, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.replications, false), &lino, _OVER);

    const char *strictStr = pDb->cfg.strict ? "on" : "off";
    char        strictVstr[24] = {0};
    STR_WITH_MAXSIZE_TO_VARSTR(strictVstr, strictStr, 24);
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)strictVstr, false), &lino, _OVER);

    char    durationStr[128] = {0};
    char    durationVstr[128] = {0};
    int32_t len = formatDurationOrKeep(&durationVstr[VARSTR_HEADER_SIZE], sizeof(durationVstr) - VARSTR_HEADER_SIZE,
                                       pDb->cfg.daysPerFile);

    varDataSetLen(durationVstr, len);
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)durationVstr, false), &lino, _OVER);

    char keepVstr[128] = {0};
    char keep0Str[32] = {0};
    char keep1Str[32] = {0};
    char keep2Str[32] = {0};

    int32_t lenKeep0 = formatDurationOrKeep(keep0Str, sizeof(keep0Str), pDb->cfg.daysToKeep0);
    int32_t lenKeep1 = formatDurationOrKeep(keep1Str, sizeof(keep1Str), pDb->cfg.daysToKeep1);
    int32_t lenKeep2 = formatDurationOrKeep(keep2Str, sizeof(keep2Str), pDb->cfg.daysToKeep2);

    if (pDb->cfg.daysToKeep0 > pDb->cfg.daysToKeep1 || pDb->cfg.daysToKeep0 > pDb->cfg.daysToKeep2) {
      len = snprintf(&keepVstr[VARSTR_HEADER_SIZE], sizeof(keepVstr) - VARSTR_HEADER_SIZE, "%s,%s,%s", keep1Str,
                     keep2Str, keep0Str);
    } else {
      len = snprintf(&keepVstr[VARSTR_HEADER_SIZE], sizeof(keepVstr) - VARSTR_HEADER_SIZE, "%s,%s,%s", keep0Str,
                     keep1Str, keep2Str);
    }
    varDataSetLen(keepVstr, len);
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)keepVstr, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.buffer, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.pageSize, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.pages, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.minRows, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.maxRows, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.compression, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)precVstr, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)statusVstr, false), &lino, _OVER);

    char *rentensionVstr = buildRetension(pDb->cfg.pRetensions);
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    if (rentensionVstr == NULL) {
      colDataSetNULL(pColInfo, rows);
    } else {
      TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)rentensionVstr, false), &lino, _OVER);
      taosMemoryFree(rentensionVstr);
    }

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.numOfStables, false), &lino, _OVER);

    const char *cacheModelStr = getCacheModelStr(pDb->cfg.cacheLast);
    char        cacheModelVstr[24] = {0};
    STR_WITH_MAXSIZE_TO_VARSTR(cacheModelVstr, cacheModelStr, 24);
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)cacheModelVstr, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.cacheLastSize, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.cacheLastShardBits, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.walLevel, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.walFsyncPeriod, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.walRetentionPeriod, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.walRetentionSize, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.sstTrigger, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    int16_t hashPrefix = pDb->cfg.hashPrefix;
    if (hashPrefix > 0) {
      hashPrefix = pDb->cfg.hashPrefix - strlen(pDb->name) - 1;
    } else if (hashPrefix < 0) {
      hashPrefix = pDb->cfg.hashPrefix + strlen(pDb->name) + 1;
    }
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&hashPrefix, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.hashSuffix, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.tsdbPageSize, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.keepTimeOffset, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.ssChunkSize, false), &lino, _OVER);

    char keeplocalVstr[128] = {0};
    len = snprintf(&keeplocalVstr[VARSTR_HEADER_SIZE], sizeof(keeplocalVstr) - VARSTR_HEADER_SIZE, "%dm",
                   pDb->cfg.ssKeepLocal);
    varDataSetLen(keeplocalVstr, len);
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)keeplocalVstr, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.ssCompact, false), &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.withArbitrator, false), &lino, _OVER);

    char encryptAlgorithmStr[TSDB_ENCRYPT_ALGR_NAME_LEN] = {0};
    getEncryptAlgorithmStr(pMnode, pDb->cfg.encryptAlgorithm, encryptAlgorithmStr);
    char        encryptAlgorithmVStr[24] = {0};
    STR_WITH_MAXSIZE_TO_VARSTR(encryptAlgorithmVStr, encryptAlgorithmStr, 24);
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)encryptAlgorithmVStr, false), &lino, _OVER);

    TAOS_UNUSED(formatDurationOrKeep(durationStr, sizeof(durationStr), pDb->cfg.compactInterval));
    STR_WITH_MAXSIZE_TO_VARSTR(durationVstr, durationStr, sizeof(durationVstr));
    if ((pColInfo = taosArrayGet(pBlock->pDataBlock, cols++))) {
      TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)durationVstr, false), &lino, _OVER);
    }

    len = formatDurationOrKeep(durationStr, sizeof(durationStr), pDb->cfg.compactStartTime);
    TAOS_UNUSED(formatDurationOrKeep(durationVstr, sizeof(durationVstr), pDb->cfg.compactEndTime));
    TAOS_UNUSED(snprintf(durationStr + len, sizeof(durationStr) - len, ",%s", durationVstr));
    STR_WITH_MAXSIZE_TO_VARSTR(durationVstr, durationStr, sizeof(durationVstr));
    if ((pColInfo = taosArrayGet(pBlock->pDataBlock, cols++))) {
      TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)durationVstr, false), &lino, _OVER);
    }

    TAOS_UNUSED(snprintf(durationStr, sizeof(durationStr), "%dh", pDb->cfg.compactTimeOffset));
    STR_WITH_MAXSIZE_TO_VARSTR(durationVstr, durationStr, sizeof(durationVstr));
    if ((pColInfo = taosArrayGet(pBlock->pDataBlock, cols++))) {
      TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)durationVstr, false), &lino, _OVER);
    }

    if ((pColInfo = taosArrayGet(pBlock->pDataBlock, cols++))) {
      TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.isAudit, false), &lino, _OVER);
    }

    if (!(*ppUidNames)) {
      TAOS_CHECK_GOTO(mndBuildUidNamesHash(pMnode, ppUidNames), &lino, _OVER);
    }
    const char *ownerName = tSimpleHashGet(*ppUidNames, (const char *)&pDb->ownerId, sizeof(pDb->ownerId));
    TAOS_UNUSED(snprintf(durationStr, sizeof(durationStr), "%s", ownerName ? ownerName : "[unknown]"));
    STR_WITH_MAXSIZE_TO_VARSTR(durationVstr, durationStr, sizeof(durationVstr));
    if ((pColInfo = taosArrayGet(pBlock->pDataBlock, cols++))) {
      TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)durationVstr, false), &lino, _OVER);
    }

    if ((pColInfo = taosArrayGet(pBlock->pDataBlock, cols++))) {
      uint8_t allowDrop = pDb->cfg.allowDrop;
      TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&allowDrop, false), &lino, _OVER);
    }

    if ((pColInfo = taosArrayGet(pBlock->pDataBlock, cols++))) {
      uint8_t securityLevel = pDb->cfg.securityLevel;
      TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&securityLevel, false), &lino, _OVER);
    }
  }
_OVER:
  if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
  taosMemoryFree(buf);
}

static void setInformationSchemaDbCfg(SMnode *pMnode, SDbObj *pDbObj) {
  tstrncpy(pDbObj->name, TSDB_INFORMATION_SCHEMA_DB, tListLen(pDbObj->name));
  pDbObj->createdTime = mndGetClusterCreateTime(pMnode);
  pDbObj->cfg.numOfVgroups = 0;
  pDbObj->cfg.strict = 1;
  pDbObj->cfg.replications = 1;
  pDbObj->cfg.precision = TSDB_TIME_PRECISION_MILLI;
}

static void setPerfSchemaDbCfg(SMnode *pMnode, SDbObj *pDbObj) {
  tstrncpy(pDbObj->name, TSDB_PERFORMANCE_SCHEMA_DB, tListLen(pDbObj->name));
  pDbObj->createdTime = mndGetClusterCreateTime(pMnode);
  pDbObj->cfg.numOfVgroups = 0;
  pDbObj->cfg.strict = 1;
  pDbObj->cfg.replications = 1;
  pDbObj->cfg.precision = TSDB_TIME_PRECISION_MILLI;
}

static bool mndGetTablesOfDbFp(SMnode *pMnode, void *pObj, void *p1, void *p2, void *p3) {
  SVgObj  *pVgroup = pObj;
  int32_t *numOfTables = p1;
  int64_t  uid = *(int64_t *)p2;
  if (pVgroup->dbUid == uid) {
    *numOfTables += pVgroup->numOfTables;
  }
  return true;
}

static int32_t mndRetrieveDbs(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) {
  SMnode    *pMnode = pReq->info.node;
  SSdb      *pSdb = pMnode->pSdb;
  int32_t    numOfRows = 0;
  SDbObj    *pDb = NULL;
  SUserObj  *pOperUser = NULL;
  ESdbStatus objStatus = 0;
  SSHashObj *pUidNames = NULL;

  (void)mndAcquireUser(pMnode, RPC_MSG_USER(pReq), &pOperUser);
  if (pOperUser == NULL) return 0;
  bool sysinfo = pOperUser->sysInfo;
  char objFName[TSDB_OBJ_FNAME_LEN + 1] = {0};
  (void)snprintf(objFName, sizeof(objFName), "%d.*", pOperUser->acctId);
  bool showAnyDb = (0 == mndCheckSysObjPrivilege(pMnode, pOperUser, RPC_MSG_TOKEN(pReq), PRIV_CM_SHOW, PRIV_OBJ_DB, 0,
                                                 objFName, NULL));

  // Append the information_schema database into the result.
  if (!pShow->sysDbRsp) {
    SDbObj infoschemaDb = {0};
    setInformationSchemaDbCfg(pMnode, &infoschemaDb);
    size_t numOfTables = 0;
    getVisibleInfosTablesNum(sysinfo, &numOfTables);
    mndDumpDbInfoData(pMnode, pBlock, &infoschemaDb, pShow, numOfRows, numOfTables, true, 0, 1, &pUidNames);

    numOfRows += 1;

    SDbObj perfschemaDb = {0};
    setPerfSchemaDbCfg(pMnode, &perfschemaDb);
    numOfTables = 0;
    getPerfDbMeta(NULL, &numOfTables);
    mndDumpDbInfoData(pMnode, pBlock, &perfschemaDb, pShow, numOfRows, numOfTables, true, 0, 1, &pUidNames);

    numOfRows += 1;
    pShow->sysDbRsp = true;
  }

  while (numOfRows < rowsCapacity) {
    pShow->pIter = sdbFetchAll(pSdb, SDB_DB, pShow->pIter, (void **)&pDb, &objStatus, true);
    if (pShow->pIter == NULL) break;
    if (showAnyDb || mndCheckDbPrivilege(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_SHOW_DATABASES, pDb) == 0) {
      int32_t numOfTables = 0;
      sdbTraverse(pSdb, SDB_VGROUP, mndGetTablesOfDbFp, &numOfTables, &pDb->uid, NULL);
      mndDumpDbInfoData(pMnode, pBlock, pDb, pShow, numOfRows, numOfTables, false, objStatus, sysinfo, &pUidNames);
      numOfRows++;
    }

    sdbRelease(pSdb, pDb);
  }

  pShow->numOfRows += numOfRows;
  mndReleaseUser(pMnode, pOperUser);
  tSimpleHashCleanup(pUidNames);
  return numOfRows;
}

static void mndCancelGetNextDb(SMnode *pMnode, void *pIter) {
  SSdb *pSdb = pMnode->pSdb;
  sdbCancelFetchByType(pSdb, pIter, SDB_DB);
}

taosdata / TDengine / #5044

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous