/*
 * 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 "os.h"
#include "taoserror.h"
#include "tutil.h"
#include "tgrant.h"
#include "tglobal.h"
#include "tname.h"
#include "tbn.h"
#include "tdataformat.h"
#include "tp.h"
#include "mnode.h"
#include "mnodeDef.h"
#include "mnodeInt.h"
#include "mnodeAcct.h"
#include "mnodeDb.h"
#include "mnodeDnode.h"
#include "mnodeMnode.h"
#include "mnodeProfile.h"
#include "mnodeWrite.h"
#include "mnodeSdb.h"
#include "mnodeShow.h"
#include "mnodeTable.h"
#include "mnodeUser.h"
#include "mnodeVgroup.h"

#define VG_LIST_SIZE 8
int64_t tsDbRid = -1;
void *  tsDbSdb = NULL;
static int32_t tsDbUpdateSize;

static int32_t mnodeCreateDb(SAcctObj *pAcct, SCreateDbMsg *pCreate, SMnodeMsg *pMsg);
static int32_t mnodeDropDb(SMnodeMsg *newMsg);
static int32_t mnodeSetDbDropping(SDbObj *pDb);
static int32_t mnodeGetDbMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
static int32_t mnodeRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static int32_t mnodeProcessCreateDbMsg(SMnodeMsg *pMsg);
static int32_t mnodeProcessDropDbMsg(SMnodeMsg *pMsg);
static int32_t mnodeProcessSyncDbMsg(SMnodeMsg *pMsg);
int32_t mnodeProcessAlterDbMsg(SMnodeMsg *pMsg);

#ifndef _TOPIC
int32_t tpInit() { return 0; }
void    tpCleanUp() {}
void    tpUpdateTs(int32_t vgId, int64_t *seq, void *pMsg) {}
#endif

static void mnodeDestroyDb(SDbObj *pDb) {
  pthread_mutex_destroy(&pDb->mutex);
  tfree(pDb->vgList);
  tfree(pDb);
}

static int32_t mnodeDbActionDestroy(SSdbRow *pRow) {
  mnodeDestroyDb(pRow->pObj);
  return TSDB_CODE_SUCCESS;
}

int64_t mnodeGetDbNum() {
  return sdbGetNumOfRows(tsDbSdb);
}

int32_t mnodeGetDbMaxReplica() {
  int32_t maxReplica = 0;
  SDbObj *pDb = NULL;
  void   *pIter = NULL;

  while (1) {
    pIter = mnodeGetNextDb(pIter, &pDb);
    if (pDb == NULL) break;

    if (pDb->cfg.replications > maxReplica)
      maxReplica = pDb->cfg.replications;

    mnodeDecDbRef(pDb);
  }

  return maxReplica;
}

static int32_t mnodeDbActionInsert(SSdbRow *pRow) {
  SDbObj *pDb = pRow->pObj;
  SAcctObj *pAcct = mnodeGetAcct(pDb->acct);

  pthread_mutex_init(&pDb->mutex, NULL);
  pthread_mutex_lock(&pDb->mutex);
  pDb->vgListSize = VG_LIST_SIZE;
  pDb->vgList = calloc(pDb->vgListSize, sizeof(SVgObj *));
  pDb->numOfVgroups = 0;
  pthread_mutex_unlock(&pDb->mutex);

  pDb->numOfTables = 0;
  pDb->numOfSuperTables = 0;

  if (pAcct != NULL) {
    mnodeAddDbToAcct(pAcct, pDb);
    mnodeDecAcctRef(pAcct);
  }
  else {
    mError("db:%s, acct:%s info not exist in sdb", pDb->name, pDb->acct);
    return TSDB_CODE_MND_INVALID_ACCT;
  }

  return TSDB_CODE_SUCCESS;
}

static int32_t mnodeDbActionDelete(SSdbRow *pRow) {
  SDbObj *pDb = pRow->pObj;
  SAcctObj *pAcct = mnodeGetAcct(pDb->acct);

  mnodeDropAllChildTables(pDb);
  mnodeDropAllSuperTables(pDb);
  mnodeDropAllDbVgroups(pDb);

  if (pAcct) {
    mnodeDropDbFromAcct(pAcct, pDb);
    mnodeDecAcctRef(pAcct);
  }

  return TSDB_CODE_SUCCESS;
}

static int32_t mnodeDbActionUpdate(SSdbRow *pRow) {
  SDbObj *pNew = pRow->pObj;
  SDbObj *pDb = mnodeGetDb(pNew->name);
  if (pDb != NULL && pNew != pDb) {
    memcpy(pDb, pNew, pRow->rowSize);
    free(pNew->vgList);
    free(pNew);
  }
  //mnodeUpdateAllDbVgroups(pDb);
  mnodeDecDbRef(pDb);
  return TSDB_CODE_SUCCESS;
}

static int32_t mnodeDbActionEncode(SSdbRow *pRow) {
  SDbObj *pDb = pRow->pObj;
  memcpy(pRow->rowData, pDb, tsDbUpdateSize);
  pRow->rowSize = tsDbUpdateSize;
  return TSDB_CODE_SUCCESS;
}

static int32_t mnodeDbActionDecode(SSdbRow *pRow) {
  SDbObj *pDb = (SDbObj *) calloc(1, sizeof(SDbObj));
  if (pDb == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY;
  
  memcpy(pDb, pRow->rowData, tsDbUpdateSize);
  pRow->pObj = pDb;
  return TSDB_CODE_SUCCESS;
}

static int32_t mnodeDbActionRestored() {
  return 0;
}

int32_t mnodeInitDbs() {
  SDbObj tObj;
  tsDbUpdateSize = (int32_t)((int8_t *)tObj.updateEnd - (int8_t *)&tObj);

  SSdbTableDesc desc = {
    .id           = SDB_TABLE_DB,
    .name         = "dbs",
    .hashSessions = TSDB_DEFAULT_DBS_HASH_SIZE,
    .maxRowSize   = tsDbUpdateSize,
    .refCountPos  = (int32_t)((int8_t *)(&tObj.refCount) - (int8_t *)&tObj),
    .keyType      = SDB_KEY_STRING,
    .fpInsert     = mnodeDbActionInsert,
    .fpDelete     = mnodeDbActionDelete,
    .fpUpdate     = mnodeDbActionUpdate,
    .fpEncode     = mnodeDbActionEncode,
    .fpDecode     = mnodeDbActionDecode,
    .fpDestroy    = mnodeDbActionDestroy,
    .fpRestored   = mnodeDbActionRestored
  };

  tsDbRid = sdbOpenTable(&desc);
  tsDbSdb = sdbGetTableByRid(tsDbRid);
  if (tsDbSdb == NULL) {
    mError("failed to init db data");
    return -1;
  }

  mnodeAddWriteMsgHandle(TSDB_MSG_TYPE_CM_CREATE_DB, mnodeProcessCreateDbMsg);
  mnodeAddWriteMsgHandle(TSDB_MSG_TYPE_CM_ALTER_DB, mnodeProcessAlterDbMsg);
  mnodeAddWriteMsgHandle(TSDB_MSG_TYPE_CM_DROP_DB, mnodeProcessDropDbMsg);
  mnodeAddWriteMsgHandle(TSDB_MSG_TYPE_CM_SYNC_DB, mnodeProcessSyncDbMsg);
  mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_DB, mnodeGetDbMeta);
  mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_DB, mnodeRetrieveDbs);
  mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_DB, mnodeCancelGetNextDb);
  
  mDebug("table:dbs table is created");
  return tpInit();
}

void *mnodeGetNextDb(void *pIter, SDbObj **pDb) {
  return sdbFetchRow(tsDbSdb, pIter, (void **)pDb);
}

void mnodeCancelGetNextDb(void *pIter) {
  sdbFreeIter(tsDbSdb, pIter);
}

SDbObj *mnodeGetDb(char *db) {
  return (SDbObj *)sdbGetRow(tsDbSdb, db);
}

void mnodeIncDbRef(SDbObj *pDb) {
  sdbIncRef(tsDbSdb, pDb); 
}

void mnodeDecDbRef(SDbObj *pDb) { 
  sdbDecRef(tsDbSdb, pDb); 
}

SDbObj *mnodeGetDbByTableName(char *tableName) {
  SName name = {0};
  tNameFromString(&name, tableName, T_NAME_ACCT|T_NAME_DB|T_NAME_TABLE);

  // validate the tableName?
  char db[TSDB_TABLE_FNAME_LEN] = {0};
  tNameGetFullDbName(&name, db);

  return mnodeGetDb(db);
}

static int32_t mnodeCheckDbCfg(SDbCfg *pCfg) {
  if (pCfg->cacheBlockSize < TSDB_MIN_CACHE_BLOCK_SIZE || pCfg->cacheBlockSize > TSDB_MAX_CACHE_BLOCK_SIZE) {
    mError("invalid db option cacheBlockSize:%d valid range: [%d, %d]", pCfg->cacheBlockSize, TSDB_MIN_CACHE_BLOCK_SIZE,
           TSDB_MAX_CACHE_BLOCK_SIZE);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->totalBlocks < TSDB_MIN_TOTAL_BLOCKS || pCfg->totalBlocks > TSDB_MAX_TOTAL_BLOCKS) {
    mError("invalid db option totalBlocks:%d valid range: [%d, %d]", pCfg->totalBlocks, TSDB_MIN_TOTAL_BLOCKS,
           TSDB_MAX_TOTAL_BLOCKS);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->maxTables < TSDB_MIN_TABLES || pCfg->maxTables > TSDB_MAX_TABLES) {
    mError("invalid db option maxTables:%d valid range: [%d, %d]", pCfg->maxTables, TSDB_MIN_TABLES, TSDB_MAX_TABLES);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->daysPerFile < TSDB_MIN_DAYS_PER_FILE || pCfg->daysPerFile > TSDB_MAX_DAYS_PER_FILE) {
    mError("invalid db option daysPerFile:%d valid range: [%d, %d]", pCfg->daysPerFile, TSDB_MIN_DAYS_PER_FILE,
           TSDB_MAX_DAYS_PER_FILE);
    return TSDB_CODE_MND_INVALID_DB_OPTION_DAYS;
  }

  if (pCfg->daysToKeep0 < TSDB_MIN_KEEP || pCfg->daysToKeep0 > TSDB_MAX_KEEP) {
    mError("invalid db option daysToKeep:%d valid range: [%d, %d]", pCfg->daysToKeep0, TSDB_MIN_KEEP, TSDB_MAX_KEEP);
    return TSDB_CODE_MND_INVALID_DB_OPTION_KEEP;
  }

  if (pCfg->daysToKeep0 < pCfg->daysPerFile) {
    mError("invalid db option daysToKeep:%d should larger than daysPerFile:%d", pCfg->daysToKeep0, pCfg->daysPerFile);
    return TSDB_CODE_MND_INVALID_DB_OPTION_KEEP;
  }

  if (pCfg->daysToKeep1 < pCfg->daysToKeep0 || pCfg->daysToKeep1 > TSDB_MAX_KEEP) {
    mError("invalid db option daysToKeep1:%d valid range: [%d, %d]", pCfg->daysToKeep1, pCfg->daysToKeep0, TSDB_MAX_KEEP);
    return TSDB_CODE_MND_INVALID_DB_OPTION_KEEP;
  }

  if (pCfg->daysToKeep2 < pCfg->daysToKeep1 || pCfg->daysToKeep2 > TSDB_MAX_KEEP) {
    mError("invalid db option daysToKeep2:%d valid range: [%d, %d]", pCfg->daysToKeep2, pCfg->daysToKeep1, TSDB_MAX_KEEP);
    return TSDB_CODE_MND_INVALID_DB_OPTION_KEEP;
  }


  if (pCfg->maxRowsPerFileBlock < TSDB_MIN_MAX_ROW_FBLOCK || pCfg->maxRowsPerFileBlock > TSDB_MAX_MAX_ROW_FBLOCK) {
    mError("invalid db option maxRowsPerFileBlock:%d valid range: [%d, %d]", pCfg->maxRowsPerFileBlock,
           TSDB_MIN_MAX_ROW_FBLOCK, TSDB_MAX_MAX_ROW_FBLOCK);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->minRowsPerFileBlock < TSDB_MIN_MIN_ROW_FBLOCK || pCfg->minRowsPerFileBlock > TSDB_MAX_MIN_ROW_FBLOCK) {
    mError("invalid db option minRowsPerFileBlock:%d valid range: [%d, %d]", pCfg->minRowsPerFileBlock,
           TSDB_MIN_MIN_ROW_FBLOCK, TSDB_MAX_MIN_ROW_FBLOCK);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->minRowsPerFileBlock > pCfg->maxRowsPerFileBlock) {
    mError("invalid db option minRowsPerFileBlock:%d should smaller than maxRowsPerFileBlock:%d",
           pCfg->minRowsPerFileBlock, pCfg->maxRowsPerFileBlock);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->commitTime < TSDB_MIN_COMMIT_TIME || pCfg->commitTime > TSDB_MAX_COMMIT_TIME) {
    mError("invalid db option commitTime:%d valid range: [%d, %d]", pCfg->commitTime, TSDB_MIN_COMMIT_TIME,
           TSDB_MAX_COMMIT_TIME);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->precision < TSDB_MIN_PRECISION && pCfg->precision > TSDB_MAX_PRECISION) {
    mError("invalid db option timePrecision:%d valid value: [%d, %d]", pCfg->precision, TSDB_MIN_PRECISION,
           TSDB_MAX_PRECISION);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->compression < TSDB_MIN_COMP_LEVEL || pCfg->compression > TSDB_MAX_COMP_LEVEL) {
    mError("invalid db option compression:%d valid range: [%d, %d]", pCfg->compression, TSDB_MIN_COMP_LEVEL,
           TSDB_MAX_COMP_LEVEL);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->walLevel < TSDB_MIN_WAL_LEVEL || pCfg->walLevel > TSDB_MAX_WAL_LEVEL) {
    mError("invalid db option walLevel:%d, valid range: [%d, %d]", pCfg->walLevel, TSDB_MIN_WAL_LEVEL, TSDB_MAX_WAL_LEVEL);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->fsyncPeriod < TSDB_MIN_FSYNC_PERIOD || pCfg->fsyncPeriod > TSDB_MAX_FSYNC_PERIOD) {
    mError("invalid db option fsyncPeriod:%d, valid range: [%d, %d]", pCfg->fsyncPeriod, TSDB_MIN_FSYNC_PERIOD, TSDB_MAX_FSYNC_PERIOD);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->replications < TSDB_MIN_DB_REPLICA_OPTION || pCfg->replications > TSDB_MAX_DB_REPLICA_OPTION) {
    mError("invalid db option replications:%d valid range: [%d, %d]", pCfg->replications, TSDB_MIN_DB_REPLICA_OPTION,
           TSDB_MAX_DB_REPLICA_OPTION);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->replications > mnodeGetDnodesNum()) {
    mError("no enough dnode to config replica: %d, #dnodes: %d", pCfg->replications, mnodeGetDnodesNum());
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->quorum > pCfg->replications) {
    mError("invalid db option quorum:%d larger than replica:%d", pCfg->quorum, pCfg->replications);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->quorum < TSDB_MIN_DB_QUORUM_OPTION || pCfg->quorum > TSDB_MAX_DB_QUORUM_OPTION) {
    mError("invalid db option quorum:%d valid range: [%d, %d]", pCfg->quorum, TSDB_MIN_DB_QUORUM_OPTION,
           TSDB_MAX_DB_QUORUM_OPTION);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->update < TSDB_MIN_DB_UPDATE || pCfg->update > TSDB_MAX_DB_UPDATE) {
    mError("invalid db option update:%d valid range: [%d, %d]", pCfg->update, TSDB_MIN_DB_UPDATE, TSDB_MAX_DB_UPDATE);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->cacheLastRow < TSDB_MIN_DB_CACHE_LAST_ROW || pCfg->cacheLastRow > TSDB_MAX_DB_CACHE_LAST_ROW) {
    mError("invalid db option cacheLastRow:%d valid range: [%d, %d]", pCfg->cacheLastRow, TSDB_MIN_DB_CACHE_LAST_ROW, TSDB_MAX_DB_CACHE_LAST_ROW);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->dbType < 0 || pCfg->dbType > 1) {
    mError("invalid db option dbType:%d valid range: [%d, %d]", pCfg->dbType, 0, 1);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (pCfg->partitions < TSDB_MIN_DB_PARTITON_OPTION || pCfg->partitions > TSDB_MAX_DB_PARTITON_OPTION) {
    mError("invalid db option partitions:%d valid range: [%d, %d]", pCfg->partitions, TSDB_MIN_DB_PARTITON_OPTION,
           TSDB_MAX_DB_PARTITON_OPTION);
    return TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  return TSDB_CODE_SUCCESS;
}

static void mnodeSetDefaultDbCfg(SDbCfg *pCfg) {
  if (pCfg->cacheBlockSize < 0) pCfg->cacheBlockSize = tsCacheBlockSize;
  if (pCfg->totalBlocks < 0) pCfg->totalBlocks = tsBlocksPerVnode;
  if (pCfg->maxTables < 0) pCfg->maxTables = tsMaxTablePerVnode;
  if (pCfg->daysPerFile < 0) pCfg->daysPerFile = tsDaysPerFile;
  if (pCfg->daysToKeep2 < 0) pCfg->daysToKeep2 = tsDaysToKeep;
  if (pCfg->daysToKeep1 < 0) pCfg->daysToKeep1 = pCfg->daysToKeep2;
  if (pCfg->daysToKeep0 < 0) pCfg->daysToKeep0 = pCfg->daysToKeep1;
  if (pCfg->minRowsPerFileBlock < 0) pCfg->minRowsPerFileBlock = tsMinRowsInFileBlock;
  if (pCfg->maxRowsPerFileBlock < 0) pCfg->maxRowsPerFileBlock = tsMaxRowsInFileBlock;
  if (pCfg->fsyncPeriod <0) pCfg->fsyncPeriod = tsFsyncPeriod;
  if (pCfg->commitTime < 0) pCfg->commitTime = tsCommitTime;
  if (pCfg->precision < 0) pCfg->precision = tsTimePrecision;
  if (pCfg->compression < 0) pCfg->compression = tsCompression;
  if (pCfg->walLevel < 0) pCfg->walLevel = tsWAL;
  if (pCfg->replications < 0) pCfg->replications = tsReplications;
  if (pCfg->quorum < 0) pCfg->quorum = MIN(tsQuorum, pCfg->replications);
  if (pCfg->update < 0) pCfg->update = tsUpdate;
  if (pCfg->cacheLastRow < 0) pCfg->cacheLastRow = tsCacheLastRow;
  if (pCfg->dbType < 0) pCfg->dbType = 0;
  if (pCfg->partitions < 0) pCfg->partitions = tsPartitons;
}

static int32_t mnodeCreateDbCb(SMnodeMsg *pMsg, int32_t code) {
  SDbObj *pDb = pMsg->pDb;
  if (code == TSDB_CODE_SUCCESS) {
    mLInfo("db:%s, is created by %s", pDb->name, mnodeGetUserFromMsg(pMsg));
  } else {
    mError("db:%s, failed to create by %s, reason:%s", pDb->name, mnodeGetUserFromMsg(pMsg), tstrerror(code));
  }

  return code;
}

static int32_t mnodeCreateDb(SAcctObj *pAcct, SCreateDbMsg *pCreate, SMnodeMsg *pMsg) {
  int32_t code = acctCheck(pAcct, ACCT_GRANT_DB);
  if (code != 0) return code;

  SDbObj *pDb = mnodeGetDb(pCreate->db);
  if (pDb != NULL) {
    mnodeDecDbRef(pDb); 
    if (pCreate->ignoreExist) {
      mDebug("db:%s, already exist, ignore exist is set", pCreate->db);
      return TSDB_CODE_SUCCESS;
    } else {
      mError("db:%s, already exist, ignore exist not set", pCreate->db);
      return TSDB_CODE_MND_DB_ALREADY_EXIST;
    }
  }

  code = grantCheck(TSDB_GRANT_DB);
  if (code != 0) return code;

  pDb = calloc(1, sizeof(SDbObj));
  tstrncpy(pDb->name, pCreate->db, sizeof(pDb->name));
  tstrncpy(pDb->acct, pAcct->user, sizeof(pDb->acct)); 
  pDb->createdTime = taosGetTimestampMs(); 
  pDb->cfg = (SDbCfg) {
    .cacheBlockSize      = pCreate->cacheBlockSize,
    .totalBlocks         = pCreate->totalBlocks,
    .maxTables           = pCreate->maxTables,
    .daysPerFile         = pCreate->daysPerFile,
    .daysToKeep0         = pCreate->daysToKeep0,
    .daysToKeep1         = pCreate->daysToKeep1,
    .daysToKeep2         = pCreate->daysToKeep2,
    .minRowsPerFileBlock = pCreate->minRowsPerFileBlock,
    .maxRowsPerFileBlock = pCreate->maxRowsPerFileBlock,
    .fsyncPeriod         = pCreate->fsyncPeriod,
    .commitTime          = pCreate->commitTime,
    .precision           = pCreate->precision,
    .compression         = pCreate->compression,
    .walLevel            = pCreate->walLevel,
    .replications        = pCreate->replications,
    .quorum              = pCreate->quorum,
    .update              = pCreate->update,
    .cacheLastRow        = pCreate->cacheLastRow,
    .dbType              = pCreate->dbType,
    .partitions          = pCreate->partitions
  };

  mnodeSetDefaultDbCfg(&pDb->cfg);

  code = mnodeCheckDbCfg(&pDb->cfg);
  if (code != TSDB_CODE_SUCCESS) {
    tfree(pDb);
    return code;
  }

  pMsg->pDb = pDb;
  mnodeIncDbRef(pDb);

  SSdbRow row = {
    .type     = SDB_OPER_GLOBAL,
    .pTable   = tsDbSdb,
    .pObj     = pDb,
    .rowSize  = sizeof(SDbObj),
    .pMsg     = pMsg,
    .fpRsp    = mnodeCreateDbCb
  };

  code = sdbInsertRow(&row);
  if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) {
    mError("db:%s, failed to create, reason:%s", pDb->name, tstrerror(code));
    pMsg->pDb = NULL;
    mnodeDestroyDb(pDb);
  }

  return code;
}

bool mnodeCheckIsMonitorDB(char *db, char *monitordb) {
  char dbName[TSDB_DB_NAME_LEN] = {0};
  extractDBName(db, dbName);

  size_t len = strlen(dbName);
  return (strncasecmp(dbName, monitordb, len) == 0 && len == strlen(monitordb));
}

#if 0
void mnodePrintVgroups(SDbObj *pDb, char *row) {
  mInfo("db:%s, vgroup link from head, row:%s", pDb->name, row);  
  SVgObj *pVgroup = pDb->pHead;
  while (pVgroup != NULL) {
    mInfo("vgId:%d", pVgroup->vgId);
    pVgroup = pVgroup->next;
  }

  mInfo("db:%s, vgroup link from tail", pDb->name, pDb->numOfVgroups);
  pVgroup = pDb->pTail;
  while (pVgroup != NULL) {
    mInfo("vgId:%d", pVgroup->vgId);
    pVgroup = pVgroup->prev;
  }
}
#endif

void mnodeAddVgroupIntoDb(SVgObj *pVgroup) {
  SDbObj *pDb = pVgroup->pDb;

  pthread_mutex_lock(&pDb->mutex);
  int32_t vgPos = pDb->numOfVgroups++;
  if (vgPos >= pDb->vgListSize) {
    pDb->vgList = realloc(pDb->vgList, pDb->vgListSize * 2 * sizeof(SVgObj *));
    memset(pDb->vgList + pDb->vgListSize, 0, pDb->vgListSize * sizeof(SVgObj *));
    pDb->vgListSize *= 2;
  }

  pDb->vgList[vgPos] = pVgroup;
  pthread_mutex_unlock(&pDb->mutex);
}

void mnodeRemoveVgroupFromDb(SVgObj *pVgroup) {
  SDbObj *pDb = pVgroup->pDb;

  pthread_mutex_lock(&pDb->mutex);
  for (int32_t v1 = 0; v1 < pDb->numOfVgroups; ++v1) {
    if (pDb->vgList[v1] == pVgroup) {
      for (int32_t v2 = v1; v2 < pDb->numOfVgroups - 1; ++v2) {
        pDb->vgList[v2] = pDb->vgList[v2 + 1];
      }
      pDb->numOfVgroups--;
      pDb->vgList[pDb->numOfVgroups] = NULL;
      break;
    }
  }

  pthread_mutex_unlock(&pDb->mutex);
}

void mnodeCleanupDbs() {
  tpCleanUp();
  sdbCloseTable(tsDbRid);
  tsDbSdb = NULL;
}

static int32_t mnodeGetDbMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
  int32_t cols = 0;

  SSchema *pSchema = pMeta->schema;
  SUserObj *pUser = mnodeGetUserFromConn(pConn);
  if (pUser == NULL) return 0;

  pShow->bytes[cols] = (TSDB_DB_NAME_LEN - 1) + VARSTR_HEADER_SIZE;
  pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
  strcpy(pSchema[cols].name, "name");
  pSchema[cols].bytes = htons(pShow->bytes[cols]);
  cols++;

  pShow->bytes[cols] = 8;
  pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
  strcpy(pSchema[cols].name, "created_time");
  pSchema[cols].bytes = htons(pShow->bytes[cols]);
  cols++;

  pShow->bytes[cols] = 4;
  pSchema[cols].type = TSDB_DATA_TYPE_INT;
  strcpy(pSchema[cols].name, "ntables");
  pSchema[cols].bytes = htons(pShow->bytes[cols]);
  cols++;

#ifndef __CLOUD_VERSION__
  if (strcmp(pUser->user, TSDB_DEFAULT_USER) == 0) {
#endif
    pShow->bytes[cols] = 4;
    pSchema[cols].type = TSDB_DATA_TYPE_INT;
    strcpy(pSchema[cols].name, "vgroups");
    pSchema[cols].bytes = htons(pShow->bytes[cols]);
    cols++;
#ifndef __CLOUD_VERSION__
  }
#endif

#ifndef __CLOUD_VERSION__
  if (strcmp(pUser->user, TSDB_DEFAULT_USER) == 0) {
#endif
    pShow->bytes[cols] = 2;
    pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
    strcpy(pSchema[cols].name, "replica");
    pSchema[cols].bytes = htons(pShow->bytes[cols]);
    cols++;

    pShow->bytes[cols] = 2;
    pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
    strcpy(pSchema[cols].name, "quorum");
    pSchema[cols].bytes = htons(pShow->bytes[cols]);
    cols++;

    pShow->bytes[cols] = 2;
    pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
    strcpy(pSchema[cols].name, "days");
    pSchema[cols].bytes = htons(pShow->bytes[cols]);
    cols++;
#ifndef __CLOUD_VERSION__
  }
#endif

  pShow->bytes[cols] = 24 + VARSTR_HEADER_SIZE;
  pSchema[cols].type = TSDB_DATA_TYPE_BINARY;

#ifdef _STORAGE  
  strcpy(pSchema[cols].name, "keep0,keep1,keep2");
#else
  strcpy(pSchema[cols].name, "keep");
#endif
  pSchema[cols].bytes = htons(pShow->bytes[cols]);
  cols++;

#ifndef __CLOUD_VERSION__
  if (strcmp(pUser->user, TSDB_DEFAULT_USER) == 0) {
#endif

    pShow->bytes[cols] = 4;
    pSchema[cols].type = TSDB_DATA_TYPE_INT;
    strcpy(pSchema[cols].name, "cache(MB)");
    pSchema[cols].bytes = htons(pShow->bytes[cols]);
    cols++;
    
    pShow->bytes[cols] = 4;
    pSchema[cols].type = TSDB_DATA_TYPE_INT;
    strcpy(pSchema[cols].name, "blocks");
    pSchema[cols].bytes = htons(pShow->bytes[cols]);
    cols++;
    
    pShow->bytes[cols] = 4;
    pSchema[cols].type = TSDB_DATA_TYPE_INT;
    strcpy(pSchema[cols].name, "minrows");
    pSchema[cols].bytes = htons(pShow->bytes[cols]);
    cols++;

    pShow->bytes[cols] = 4;
    pSchema[cols].type = TSDB_DATA_TYPE_INT;
    strcpy(pSchema[cols].name, "maxrows");
    pSchema[cols].bytes = htons(pShow->bytes[cols]);
    cols++;

    pShow->bytes[cols] = 1;
    pSchema[cols].type = TSDB_DATA_TYPE_TINYINT;
    strcpy(pSchema[cols].name, "wallevel");
    pSchema[cols].bytes = htons(pShow->bytes[cols]);
    cols++;

    pShow->bytes[cols] = 4;
    pSchema[cols].type = TSDB_DATA_TYPE_INT;
    strcpy(pSchema[cols].name, "fsync");
    pSchema[cols].bytes = htons(pShow->bytes[cols]);
    cols++;

    pShow->bytes[cols] = 1;
    pSchema[cols].type = TSDB_DATA_TYPE_TINYINT;
    strcpy(pSchema[cols].name, "comp");
    pSchema[cols].bytes = htons(pShow->bytes[cols]);
    cols++;

    pShow->bytes[cols] = 1;
    pSchema[cols].type = TSDB_DATA_TYPE_TINYINT;
    strcpy(pSchema[cols].name, "cachelast");
    pSchema[cols].bytes = htons(pShow->bytes[cols]);
    cols++;
#ifndef __CLOUD_VERSION__
  }
#endif

  pShow->bytes[cols] = 3 + VARSTR_HEADER_SIZE;
  pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
  strcpy(pSchema[cols].name, "precision");
  pSchema[cols].bytes = htons(pShow->bytes[cols]);
  cols++;

  pShow->bytes[cols] = 1;
  pSchema[cols].type = TSDB_DATA_TYPE_TINYINT;
  strcpy(pSchema[cols].name, "update");
  pSchema[cols].bytes = htons(pShow->bytes[cols]);
  cols++;

  pShow->bytes[cols] = 10 + VARSTR_HEADER_SIZE;
  pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
  strcpy(pSchema[cols].name, "status");
  pSchema[cols].bytes = htons(pShow->bytes[cols]);
  cols++;

  pMeta->numOfColumns = htons(cols);
  pShow->numOfColumns = cols;

  pShow->offset[0] = 0;
  for (int32_t i = 1; i < cols; ++i) {
    pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
  }

  pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
  pShow->numOfRows = pUser->pAcct->acctInfo.numOfDbs;

  mnodeDecUserRef(pUser);
  return 0;
}

char *mnodeGetDbStr(char *src) {
  char *pos = strstr(src, TS_PATH_DELIMITER);
  if (pos != NULL) ++pos;

  return pos;
}

static int32_t mnodeRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
  int32_t numOfRows = 0;
  SDbObj *pDb = NULL;
  char *  pWrite;
  int32_t cols = 0;
  SUserObj *pUser = mnodeGetUserFromConn(pConn);
  if (pUser == NULL) return 0;

  while (numOfRows < rows) {
    pShow->pIter = mnodeGetNextDb(pShow->pIter, &pDb);

    if (pDb == NULL) break;
    if (pDb->pAcct != pUser->pAcct || pDb->status != TSDB_DB_STATUS_READY) {
      mnodeDecDbRef(pDb);
      continue;
    }

    cols = 0;

    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;  
    char* name = mnodeGetDbStr(pDb->name);
    if (name != NULL) {
      STR_WITH_MAXSIZE_TO_VARSTR(pWrite, name, pShow->bytes[cols]);
    } else {
      STR_TO_VARSTR(pWrite, "NULL");
    }
    cols++;

    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
    *(int64_t *)pWrite = pDb->createdTime;
    cols++;

    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
    *(int32_t *)pWrite = pDb->numOfTables;
    cols++;

#ifndef __CLOUD_VERSION__
    if (strcmp(pUser->user, TSDB_DEFAULT_USER) == 0) {
#endif
      pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
      *(int32_t *)pWrite = pDb->numOfVgroups;
      cols++;
#ifndef __CLOUD_VERSION__
    }
#endif

#ifndef __CLOUD_VERSION__
    if (strcmp(pUser->user, TSDB_DEFAULT_USER) == 0) {
#endif
      pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
      *(int16_t *)pWrite = pDb->cfg.replications;
      cols++;

      pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
      *(int16_t *)pWrite = pDb->cfg.quorum;
      cols++;

      pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
      *(int16_t *)pWrite = pDb->cfg.daysPerFile;
      cols++;
#ifndef __CLOUD_VERSION__
    }
#endif

    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
    
    char tmp[128] = {0};
#ifdef _STORAGE      
    if (pDb->cfg.daysToKeep0 > pDb->cfg.daysToKeep1 || pDb->cfg.daysToKeep0 > pDb->cfg.daysToKeep2) { 
      sprintf(tmp, "%d,%d,%d", pDb->cfg.daysToKeep1, pDb->cfg.daysToKeep2, pDb->cfg.daysToKeep0);
    } else {
      sprintf(tmp, "%d,%d,%d", pDb->cfg.daysToKeep0, pDb->cfg.daysToKeep1, pDb->cfg.daysToKeep2);
    }
#else
    sprintf(tmp, "%d", pDb->cfg.daysToKeep0);
#endif
    STR_WITH_SIZE_TO_VARSTR(pWrite, tmp, strlen(tmp));
    cols++;

#ifndef __CLOUD_VERSION__
    if (strcmp(pUser->user, TSDB_DEFAULT_USER) == 0) {
#endif
      pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
      *(int32_t *)pWrite = pDb->cfg.cacheBlockSize;
      cols++;

      pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
      *(int32_t *)pWrite = pDb->cfg.totalBlocks;
      cols++;

      pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
      *(int32_t *)pWrite = pDb->cfg.minRowsPerFileBlock;
      cols++;

      pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
      *(int32_t *)pWrite = pDb->cfg.maxRowsPerFileBlock;
      cols++;
      
      pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
      *(int8_t *)pWrite = pDb->cfg.walLevel;
      cols++;

      pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
      *(int32_t *)pWrite = pDb->cfg.fsyncPeriod;
      cols++;

      pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
      *(int8_t *)pWrite = pDb->cfg.compression;
      cols++;

      pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
      *(int8_t *)pWrite = pDb->cfg.cacheLastRow;
      cols++;
#ifndef __CLOUD_VERSION__
    }
#endif

    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
    char *prec = (pDb->cfg.precision == TSDB_TIME_PRECISION_MILLI) ? TSDB_TIME_PRECISION_MILLI_STR
                                                                   : TSDB_TIME_PRECISION_MICRO_STR;
    STR_WITH_SIZE_TO_VARSTR(pWrite, prec, 2);
    cols++;

    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
    *(int8_t *)pWrite = pDb->cfg.update;
    cols++;

    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
    if (pDb->status == TSDB_DB_STATUS_READY) {
      const char *src = "ready";
      STR_WITH_SIZE_TO_VARSTR(pWrite, src, strlen(src));
    } else {
      const char *src = "dropping";
      STR_WITH_SIZE_TO_VARSTR(pWrite, src, strlen(src));
    }
    cols++;

    numOfRows++;
    mnodeDecDbRef(pDb);
  }

  pShow->numOfReads += numOfRows;
  mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);

  mnodeDecUserRef(pUser);
  return numOfRows;
}

void mnodeAddSuperTableIntoDb(SDbObj *pDb) {
  atomic_add_fetch_32(&pDb->numOfSuperTables, 1);
}

void mnodeRemoveSuperTableFromDb(SDbObj *pDb) {
  atomic_add_fetch_32(&pDb->numOfSuperTables, -1);
}

void mnodeAddTableIntoDb(SDbObj *pDb) {
  atomic_add_fetch_32(&pDb->numOfTables, 1);
}

void mnodeRemoveTableFromDb(SDbObj *pDb) {
  atomic_add_fetch_32(&pDb->numOfTables, -1);
}

static int32_t mnodeSetDbDropping(SDbObj *pDb) {
  if (pDb->status) return TSDB_CODE_SUCCESS;

  pDb->status = true;
  SSdbRow row = {
    .type   = SDB_OPER_GLOBAL,
    .pTable = tsDbSdb,
    .pObj   = pDb
  };

  int32_t code = sdbUpdateRow(&row);
  if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) {
    mError("db:%s, failed to set dropping state, reason:%s", pDb->name, tstrerror(code));
  }

  return code;
}

static int32_t mnodeProcessCreateDbMsg(SMnodeMsg *pMsg) {
  SCreateDbMsg *pCreate    = pMsg->rpcMsg.pCont;  
  pCreate->maxTables       = htonl(pCreate->maxTables);
  pCreate->cacheBlockSize  = htonl(pCreate->cacheBlockSize);
  pCreate->totalBlocks     = htonl(pCreate->totalBlocks);
  pCreate->daysPerFile     = htonl(pCreate->daysPerFile);
  pCreate->daysToKeep0     = htonl(pCreate->daysToKeep0);
  pCreate->daysToKeep1     = htonl(pCreate->daysToKeep1);
  pCreate->daysToKeep2     = htonl(pCreate->daysToKeep2);
  pCreate->commitTime      = htonl(pCreate->commitTime);
  pCreate->fsyncPeriod     = htonl(pCreate->fsyncPeriod);
  pCreate->partitions      = htons(pCreate->partitions);
  pCreate->minRowsPerFileBlock = htonl(pCreate->minRowsPerFileBlock);
  pCreate->maxRowsPerFileBlock = htonl(pCreate->maxRowsPerFileBlock);
  
  int32_t code;
  if (grantCheck(TSDB_GRANT_TIME) != TSDB_CODE_SUCCESS) {
    code = TSDB_CODE_GRANT_EXPIRED;
  } else if (!pMsg->pUser->writeAuth) {
    code = TSDB_CODE_MND_NO_RIGHTS;
  } else {
    code = mnodeCreateDb(pMsg->pUser->pAcct, pCreate, pMsg);
  }

  return code;
}

static SDbCfg mnodeGetAlterDbOption(SDbObj *pDb, SAlterDbMsg *pAlter) {
  SDbCfg  newCfg = pDb->cfg;
  int32_t maxTables      = htonl(pAlter->maxTables);
  int32_t cacheBlockSize = htonl(pAlter->cacheBlockSize);
  int32_t totalBlocks    = htonl(pAlter->totalBlocks);
  int32_t daysPerFile    = htonl(pAlter->daysPerFile);
  int32_t daysToKeep0    = htonl(pAlter->daysToKeep0);
  int32_t daysToKeep1    = htonl(pAlter->daysToKeep1);
  int32_t daysToKeep2    = htonl(pAlter->daysToKeep2);
  int32_t minRows        = htonl(pAlter->minRowsPerFileBlock);
  int32_t maxRows        = htonl(pAlter->maxRowsPerFileBlock);
  int32_t commitTime     = htonl(pAlter->commitTime);
  int32_t fsyncPeriod    = htonl(pAlter->fsyncPeriod);
  int8_t  compression    = pAlter->compression;
  int8_t  walLevel       = pAlter->walLevel;
  int8_t  replications   = pAlter->replications;
  int8_t  quorum         = pAlter->quorum;
  int8_t  precision      = pAlter->precision;
  int8_t  update         = pAlter->update;
  int8_t  cacheLastRow   = pAlter->cacheLastRow;
  int8_t  dbType         = pAlter->dbType;
  int16_t partitions     = htons(pAlter->partitions);
  
  terrno = TSDB_CODE_SUCCESS;

  //UPGRATE FROM LOW VERSION, reorder it
  if (pDb->cfg.daysToKeep0 > pDb->cfg.daysToKeep1 || pDb->cfg.daysToKeep0 > pDb->cfg.daysToKeep2) {
    int32_t t = pDb->cfg.daysToKeep0;
    newCfg.daysToKeep0 = pDb->cfg.daysToKeep1;
    newCfg.daysToKeep1 = pDb->cfg.daysToKeep2;
    newCfg.daysToKeep2 = t;
  }

  if (cacheBlockSize > 0 && cacheBlockSize != pDb->cfg.cacheBlockSize) {
    mError("db:%s, can't alter cache option", pDb->name);
    terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (totalBlocks > 0 && totalBlocks != pDb->cfg.totalBlocks) {
    mInfo("db:%s, blocks:%d change to %d", pDb->name, pDb->cfg.totalBlocks, totalBlocks);
    newCfg.totalBlocks = totalBlocks;
  }

  if (maxTables > 0) {
    mInfo("db:%s, maxTables:%d change to %d", pDb->name, pDb->cfg.maxTables, maxTables);
    newCfg.maxTables = maxTables;
    if (newCfg.maxTables < pDb->cfg.maxTables) {
      mError("db:%s, tables:%d should larger than origin:%d", pDb->name, newCfg.maxTables, pDb->cfg.maxTables);
      terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
    }
  }

  if (daysPerFile > 0 && daysPerFile != pDb->cfg.daysPerFile) {
    mError("db:%s, can't alter days option", pDb->name);
    terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (daysToKeep0 > 0 && (daysToKeep0 != pDb->cfg.daysToKeep0 || newCfg.daysToKeep0 != pDb->cfg.daysToKeep0)) {
    mDebug("db:%s, daysToKeep:%d change to %d", pDb->name, pDb->cfg.daysToKeep0, daysToKeep0);
    newCfg.daysToKeep0 = daysToKeep0;
  }

  if (daysToKeep1 > 0 && (daysToKeep1 != pDb->cfg.daysToKeep1 || newCfg.daysToKeep1 != pDb->cfg.daysToKeep1)) {
    mDebug("db:%s, daysToKeep1:%d change to %d", pDb->name, pDb->cfg.daysToKeep1, daysToKeep1);
    newCfg.daysToKeep1 = daysToKeep1;
  }

  if (daysToKeep2 > 0 && (daysToKeep2 != pDb->cfg.daysToKeep2 || newCfg.daysToKeep2 != pDb->cfg.daysToKeep2)) {
    mDebug("db:%s, daysToKeep2:%d change to %d", pDb->name, pDb->cfg.daysToKeep2, daysToKeep2);
    newCfg.daysToKeep2 = daysToKeep2;
  }

  if (minRows > 0 && minRows != pDb->cfg.minRowsPerFileBlock) {
    mError("db:%s, can't alter minRows option", pDb->name);
    terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (maxRows > 0 && maxRows != pDb->cfg.maxRowsPerFileBlock) {
    mError("db:%s, can't alter maxRows option", pDb->name);
    terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (commitTime > 0 && commitTime != pDb->cfg.commitTime) {
    mError("db:%s, can't alter commitTime option", pDb->name);
    terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (precision > 0 && precision != pDb->cfg.precision) {
    mError("db:%s, can't alter precision option", pDb->name);
    terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
  }

  if (compression >= 0 && compression != pDb->cfg.compression) {
    mDebug("db:%s, compression:%d change to %d", pDb->name, pDb->cfg.compression, compression);
    newCfg.compression = compression;
  }

  if (walLevel > 0 && walLevel != pDb->cfg.walLevel) {
    mDebug("db:%s, walLevel:%d change to %d", pDb->name, pDb->cfg.walLevel, walLevel);
    newCfg.walLevel = walLevel;
  }

  if (fsyncPeriod >= 0 && fsyncPeriod != pDb->cfg.fsyncPeriod) {
    mDebug("db:%s, fsyncPeriod:%d change to %d", pDb->name, pDb->cfg.fsyncPeriod, fsyncPeriod);
    newCfg.fsyncPeriod = fsyncPeriod;
  }

  if (replications > 0 && replications != pDb->cfg.replications) {
    mDebug("db:%s, replications:%d change to %d", pDb->name, pDb->cfg.replications, replications);
    newCfg.replications = replications;

    if (pDb->cfg.walLevel < TSDB_MIN_WAL_LEVEL) {
      mError("db:%s, walLevel:%d must be greater than 0", pDb->name, pDb->cfg.walLevel);
      terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
    }

    if (replications > mnodeGetDnodesNum()) {
      mError("db:%s, no enough dnode to change replica:%d", pDb->name, replications);
      terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
    }

    if (pDb->cfg.replications - replications >= 2) {
      mError("db:%s, replica number can't change from %d to %d", pDb->name, pDb->cfg.replications, replications);
      terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
    }
  }

  if (quorum >= 0 && quorum != pDb->cfg.quorum) {
    mDebug("db:%s, quorum:%d change to %d", pDb->name, pDb->cfg.quorum, quorum);
    newCfg.quorum = quorum;
  }

  if (update >= 0 && update != pDb->cfg.update) {
#if 0
    mDebug("db:%s, update:%d change to %d", pDb->name, pDb->cfg.update, update);
    newCfg.update = update;
#else
    mError("db:%s, can't alter update option", pDb->name);
    terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
#endif
  }

  if (cacheLastRow >= 0 && cacheLastRow != pDb->cfg.cacheLastRow) {
    mDebug("db:%s, cacheLastRow:%d change to %d", pDb->name, pDb->cfg.cacheLastRow, cacheLastRow);
    newCfg.cacheLastRow = cacheLastRow;
  }

  if (dbType >= 0 && dbType != pDb->cfg.dbType) {
    mDebug("db:%s, dbType:%d change to %d", pDb->name, pDb->cfg.dbType, dbType);
    newCfg.dbType = dbType;
  }

  if (partitions >= 0 && partitions != pDb->cfg.partitions) {
    mDebug("db:%s, partitions:%d change to %d", pDb->name, pDb->cfg.partitions, partitions);
    newCfg.partitions = partitions;
  }

// community version can only change daysToKeep
// but enterprise version can change all daysToKeep options
#ifndef _STORAGE
  newCfg.daysToKeep1 = newCfg.daysToKeep0;
  newCfg.daysToKeep2 = newCfg.daysToKeep0;
#endif

  return newCfg;
}

static int32_t mnodeAlterDbCb(SMnodeMsg *pMsg, int32_t code) {
  if (code != TSDB_CODE_SUCCESS) return code;
  SDbObj *pDb = pMsg->pDb;

  void *pIter = NULL;
  SVgObj *pVgroup = NULL;
    while (1) {
    pIter = mnodeGetNextVgroup(pIter, &pVgroup);
    if (pVgroup == NULL) break;
    if (pVgroup->pDb == pDb) {
      mnodeSendAlterVgroupMsg(pVgroup);
    }
    mnodeDecVgroupRef(pVgroup);
  }

  mDebug("db:%s, all vgroups is altered", pDb->name);
  mLInfo("db:%s, is alterd by %s", pDb->name, mnodeGetUserFromMsg(pMsg));

  bnNotify();

  return TSDB_CODE_SUCCESS;
}

static int32_t mnodeAlterDb(SDbObj *pDb, SAlterDbMsg *pAlter, void *pMsg) {
  mDebug("db:%s, type:%d do alter operation", pDb->name, pDb->cfg.dbType);

  SDbCfg newCfg = mnodeGetAlterDbOption(pDb, pAlter);
  if (terrno != TSDB_CODE_SUCCESS) {
    return terrno;
  }

  int32_t code = mnodeCheckDbCfg(&newCfg);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

  if (memcmp(&newCfg, &pDb->cfg, sizeof(SDbCfg)) != 0) {
    pDb->cfg = newCfg;
    pDb->dbCfgVersion++;
    SSdbRow row = {
      .type    = SDB_OPER_GLOBAL,
      .pTable  = tsDbSdb,
      .pObj    = pDb,
      .pMsg    = pMsg,
      .fpRsp   = mnodeAlterDbCb
    };

    code = sdbUpdateRow(&row);
    if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) {
      mError("db:%s, failed to alter, reason:%s", pDb->name, tstrerror(code));
    }
  }

  return code;
}

int32_t mnodeProcessAlterDbMsg(SMnodeMsg *pMsg) {
  SAlterDbMsg *pAlter = pMsg->rpcMsg.pCont;
  mDebug("db:%s, alter db msg is received from thandle:%p, dbType:%d", pAlter->db, pMsg->rpcMsg.handle, pAlter->dbType);

  if (pMsg->pDb == NULL) pMsg->pDb = mnodeGetDb(pAlter->db);
  if (pMsg->pDb == NULL) {
    mError("db:%s, failed to alter, invalid db", pAlter->db);
    return TSDB_CODE_MND_INVALID_DB;
  }
  
  if (pMsg->pDb->status != TSDB_DB_STATUS_READY) {
    mError("db:%s, status:%d, in dropping", pAlter->db, pMsg->pDb->status);
    return TSDB_CODE_MND_DB_IN_DROPPING;
  }

  return mnodeAlterDb(pMsg->pDb, pAlter, pMsg);
}

static int32_t mnodeDropDbCb(SMnodeMsg *pMsg, int32_t code) {
  SDbObj *pDb = pMsg->pDb;
  if (code != TSDB_CODE_SUCCESS) {
    mError("db:%s, failed to drop from sdb, reason:%s", pDb->name, tstrerror(code));
  } else {
    mLInfo("db:%s, is dropped by %s", pDb->name, mnodeGetUserFromMsg(pMsg));
  }

  return code;
}

static int32_t mnodeDropDb(SMnodeMsg *pMsg) {
  if (pMsg == NULL) return TSDB_CODE_MND_APP_ERROR;
  
  SDbObj *pDb = pMsg->pDb;
  mInfo("db:%s, drop db from sdb", pDb->name);

  SSdbRow row = {
    .type    = SDB_OPER_GLOBAL,
    .pTable  = tsDbSdb,
    .pObj    = pDb,
    .pMsg    = pMsg,
    .fpRsp   = mnodeDropDbCb
  };

  int32_t code = sdbDeleteRow(&row);
  if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) {
    mError("db:%s, failed to drop, reason:%s", pDb->name, tstrerror(code));
  }

  return code;
}

static int32_t mnodeProcessDropDbMsg(SMnodeMsg *pMsg) {
  SDropDbMsg *pDrop = pMsg->rpcMsg.pCont;
  mDebug("db:%s, drop db msg is received from thandle:%p", pDrop->db, pMsg->rpcMsg.handle);

  if (pMsg->pDb == NULL) pMsg->pDb = mnodeGetDb(pDrop->db);
  if (pMsg->pDb == NULL) {
    if (pDrop->ignoreNotExists) {
      mDebug("db:%s, db is not exist, treat as success", pDrop->db);
      return TSDB_CODE_SUCCESS;
    } else {
      mError("db:%s, failed to drop, invalid db", pDrop->db);
      return TSDB_CODE_MND_INVALID_DB;
    }
  }

#if 0
  if (mnodeCheckIsMonitorDB(pMsg->pDb->name, tsMonitorDbName)) {
    mError("db:%s, can't drop monitor database", pDrop->db);
    return TSDB_CODE_MND_MONITOR_DB_FORBIDDEN;
  }
#endif   

  int32_t code = mnodeSetDbDropping(pMsg->pDb);
  if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) {
    mError("db:%s, failed to drop, reason:%s", pDrop->db, tstrerror(code));
    return code;
  }

  mnodeSendDropAllDbVgroupsMsg(pMsg->pDb);

  mDebug("db:%s, all vgroups is dropped", pMsg->pDb->name);
  return mnodeDropDb(pMsg);
}

static int32_t mnodeSyncDb(SDbObj *pDb, SMnodeMsg *pMsg) {
  void *pIter = NULL;
  SVgObj *pVgroup = NULL;
    while (1) {
    pIter = mnodeGetNextVgroup(pIter, &pVgroup);
    if (pVgroup == NULL) break;
    if (pVgroup->pDb == pDb) {
      mnodeSendSyncVgroupMsg(pVgroup);
    }
    mnodeDecVgroupRef(pVgroup);
  }

  mLInfo("db:%s, is synced by %s", pDb->name, mnodeGetUserFromMsg(pMsg));

  return TSDB_CODE_SUCCESS;
}

static int32_t mnodeProcessSyncDbMsg(SMnodeMsg *pMsg) {
  SSyncDbMsg *pSyncDb = pMsg->rpcMsg.pCont;
  mDebug("db:%s, syncdb is received from thandle:%p, ignore:%d", pSyncDb->db, pMsg->rpcMsg.handle, pSyncDb->ignoreNotExists);

  if (pMsg->pDb == NULL) pMsg->pDb = mnodeGetDb(pSyncDb->db);
  if (pMsg->pDb == NULL) {
        if (pSyncDb->ignoreNotExists) {
          mDebug("db:%s, db is not exist, treat as success", pSyncDb->db);
          return TSDB_CODE_SUCCESS;
        } else {
          mError("db:%s, failed to sync, invalid db", pSyncDb->db);
          return TSDB_CODE_MND_INVALID_DB;
        }
  }

  if (pMsg->pDb->status != TSDB_DB_STATUS_READY) {
    mError("db:%s, status:%d, in dropping", pSyncDb->db, pMsg->pDb->status);
    return TSDB_CODE_MND_DB_IN_DROPPING;
  }

  return mnodeSyncDb(pMsg->pDb, pMsg);
}

void  mnodeDropAllDbs(SAcctObj *pAcct)  {
  int32_t numOfDbs = 0;
  SDbObj *pDb = NULL;
  void *  pIter = NULL;

  mInfo("acct:%s, all dbs will be dropped from sdb", pAcct->user);

  while (1) {
    pIter = mnodeGetNextDb(pIter, &pDb);
    if (pDb == NULL) break;

    if (pDb->pAcct == pAcct) {
      mInfo("db:%s, drop db from sdb for acct:%s is dropped", pDb->name, pAcct->user);
      SSdbRow row = {
        .type   = SDB_OPER_LOCAL,
        .pTable = tsDbSdb,
        .pObj   = pDb
      };
      
      sdbDeleteRow(&row);
      numOfDbs++;
    }
    mnodeDecDbRef(pDb);
  }

  mInfo("acct:%s, all dbs:%d is dropped from sdb", pAcct->user, numOfDbs);
}

int32_t mnodeCompactDbs() {
  void *pIter = NULL;
  SDbObj *pDb = NULL;

  mInfo("start to compact dbs table...");

  while (1) {
    pIter = mnodeGetNextDb(pIter, &pDb);
    if (pDb == NULL) break;

    SSdbRow row = {
      .type     = SDB_OPER_GLOBAL,
      .pTable   = tsDbSdb,
      .pObj     = pDb,
      .rowSize  = sizeof(SDbObj),
    };

    mInfo("compact dbs %s", pDb->name);
    
    sdbInsertCompactRow(&row);
  }

  mInfo("end to compact dbs table...");

  return 0; 
}