Merge pull request #6179 from taosdata/feature/TD-1920

Feature/td 1920

src/common/inc/tglobal.h

@@ -142,12 +142,15 @@ extern int32_t tsMonitorInterval;
 extern int8_t tsEnableStream;
 
 // internal
+extern int8_t  tsCompactMnodeWal;
 extern int8_t  tsPrintAuth;
 extern int8_t  tscEmbedded;
 extern char    configDir[];
 extern char    tsVnodeDir[];
 extern char    tsDnodeDir[];
 extern char    tsMnodeDir[];
+extern char    tsMnodeBakDir[];
+extern char    tsMnodeTmpDir[];
 extern char    tsDataDir[];
 extern char    tsLogDir[];
 extern char    tsScriptDir[];

src/common/src/tglobal.c

@@ -176,12 +176,15 @@ int32_t tsMonitorInterval = 30;  // seconds
 int8_t  tsEnableStream = 1;
 
 // internal
+int8_t tsCompactMnodeWal = 0;
 int8_t tsPrintAuth = 0;
 int8_t tscEmbedded = 0;
 char   configDir[TSDB_FILENAME_LEN] = {0};
 char   tsVnodeDir[TSDB_FILENAME_LEN] = {0};
 char   tsDnodeDir[TSDB_FILENAME_LEN] = {0};
 char   tsMnodeDir[TSDB_FILENAME_LEN] = {0};
+char   tsMnodeTmpDir[TSDB_FILENAME_LEN] = {0};
+char   tsMnodeBakDir[TSDB_FILENAME_LEN] = {0};
 char   tsDataDir[TSDB_FILENAME_LEN] = {0};
 char   tsScriptDir[TSDB_FILENAME_LEN] = {0};
 char   tsTempDir[TSDB_FILENAME_LEN] = "/tmp/";

src/dnode/src/dnodeMain.c

@@ -40,6 +40,7 @@
 #include "dnodeShell.h"
 #include "dnodeTelemetry.h"
 #include "module.h"
+#include "mnode.h"
 
 #if !defined(_MODULE) || !defined(_TD_LINUX)
 int32_t moduleStart() { return 0; }
@@ -216,6 +217,17 @@ static int32_t dnodeInitStorage() {
   sprintf(tsDnodeDir, "%s/dnode", tsDataDir);
   // sprintf(tsVnodeBakDir, "%s/vnode_bak", tsDataDir);
 
+  if (tsCompactMnodeWal == 1) {
+    sprintf(tsMnodeTmpDir, "%s/mnode_tmp", tsDataDir);
+    tfsRmdir(tsMnodeTmpDir);
+    if (dnodeCreateDir(tsMnodeTmpDir) < 0) {
+      dError("failed to create dir: %s, reason: %s", tsMnodeTmpDir, strerror(errno));
+      return -1;
+    }
+
+    sprintf(tsMnodeBakDir, "%s/mnode_bak", tsDataDir);
+    //tfsRmdir(tsMnodeBakDir);   
+  }
   //TODO(dengyihao): no need to init here 
   if (dnodeCreateDir(tsMnodeDir) < 0) {
    dError("failed to create dir: %s, reason: %s", tsMnodeDir, strerror(errno));

src/dnode/src/dnodeSystem.c

@@ -42,6 +42,8 @@ int32_t main(int32_t argc, char *argv[]) {
       }
     } else if (strcmp(argv[i], "-C") == 0) {
       dump_config = 1;
+    } else if (strcmp(argv[i], "--compact-mnode-wal") == 0) {
+      tsCompactMnodeWal = 1;
     } else if (strcmp(argv[i], "-V") == 0) {
 #ifdef _ACCT
       char *versionStr = "enterprise";

src/inc/mnode.h

@@ -73,6 +73,9 @@ int32_t mnodeProcessPeerReq(SMnodeMsg *pMsg);
 void    mnodeProcessPeerRsp(SRpcMsg *pMsg);
 int32_t mnodeRetriveAuth(char *user, char *spi, char *encrypt, char *secret, char *ckey);
 
+int32_t mnodeCompactWal();
+int32_t mnodeCompactComponents();
+
 #ifdef __cplusplus
 }
 #endif

src/mnode/inc/mnodeAcct.h

@@ -35,6 +35,8 @@ void    mnodeDropDbFromAcct(SAcctObj *pAcct, SDbObj *pDb);
 void    mnodeAddUserToAcct(SAcctObj *pAcct, SUserObj *pUser);
 void    mnodeDropUserFromAcct(SAcctObj *pAcct, SUserObj *pUser);
 
+int32_t mnodeCompactAccts();
+
 #ifdef __cplusplus
 }
 #endif

src/mnode/inc/mnodeCluster.h

@@ -25,6 +25,8 @@ void    mnodeCleanupCluster();
 void    mnodeUpdateClusterId();
 const char* mnodeGetClusterId();
 
+int32_t mnodeCompactCluster();
+
 #ifdef __cplusplus
 }
 #endif

src/mnode/inc/mnodeDb.h

@@ -41,6 +41,8 @@ void    mnodeDecDbRef(SDbObj *pDb);
 bool    mnodeCheckIsMonitorDB(char *db, char *monitordb);
 void    mnodeDropAllDbs(SAcctObj *pAcct);
 
+int32_t mnodeCompactDbs();
+
 // util func
 void mnodeAddSuperTableIntoDb(SDbObj *pDb);
 void mnodeRemoveSuperTableFromDb(SDbObj *pDb);

src/mnode/inc/mnodeDnode.h

@@ -77,6 +77,7 @@ void *  mnodeGetDnodeByEp(char *ep);
 void    mnodeUpdateDnode(SDnodeObj *pDnode);
 int32_t mnodeDropDnode(SDnodeObj *pDnode, void *pMsg);
 
+int32_t mnodeCompactDnodes();
 extern int32_t tsAccessSquence;
 
 #ifdef __cplusplus

src/mnode/inc/mnodeMnode.h

@@ -50,6 +50,7 @@ char*   mnodeGetMnodeMasterEp();
 void    mnodeGetMnodeInfos(void *mnodes);
 void    mnodeUpdateMnodeEpSet(SMInfos *pMnodes);
 
+int32_t mnodeCompactMnodes();
 #ifdef __cplusplus
 }
 #endif

src/mnode/inc/mnodeSdb.h

@@ -92,6 +92,7 @@ void    sdbUpdateMnodeRoles();
 int32_t sdbGetReplicaNum();
 
 int32_t sdbInsertRow(SSdbRow *pRow);
+int32_t sdbInsertCompactRow(SSdbRow *pRow);
 int32_t sdbDeleteRow(SSdbRow *pRow);
 int32_t sdbUpdateRow(SSdbRow *pRow);
 int32_t sdbInsertRowToQueue(SSdbRow *pRow);
@@ -106,6 +107,7 @@ int32_t  sdbGetId(void *pTable);
 uint64_t sdbGetVersion();
 bool     sdbCheckRowDeleted(void *pTable, void *pRow);
 
+int32_t mnodeCompactWal();
 #ifdef __cplusplus
 }
 #endif

src/mnode/inc/mnodeTable.h

@@ -36,6 +36,7 @@ void    mnodeCancelGetNextSuperTable(void *pIter);
 void    mnodeDropAllChildTables(SDbObj *pDropDb);
 void    mnodeDropAllSuperTables(SDbObj *pDropDb);
 void    mnodeDropAllChildTablesInVgroups(SVgObj *pVgroup);
+int32_t mnodeCompactTables();
 
 #ifdef __cplusplus
 }

src/mnode/inc/mnodeUser.h

@@ -33,6 +33,8 @@ char *    mnodeGetUserFromMsg(void *pMnodeMsg);
 int32_t   mnodeCreateUser(SAcctObj *pAcct, char *name, char *pass, void *pMsg);
 void      mnodeDropAllUsers(SAcctObj *pAcct);
 
+int32_t   mnodeCompactUsers();
+
 #ifdef __cplusplus
 }
 #endif

src/mnode/inc/mnodeVgroup.h

@@ -32,6 +32,7 @@ void    mnodeDropAllDbVgroups(SDbObj *pDropDb);
 void    mnodeSendDropAllDbVgroupsMsg(SDbObj *pDropDb);
 void    mnodeDropAllDnodeVgroups(SDnodeObj *pDropDnode);
 //void  mnodeUpdateAllDbVgroups(SDbObj *pAlterDb);
+int32_t mnodeCompactVgroups();
 
 void *  mnodeGetNextVgroup(void *pIter, SVgObj **pVgroup);
 void    mnodeCancelGetNextVgroup(void *pIter);

src/mnode/src/mnodeAcct.c

@@ -238,6 +238,32 @@ static int32_t mnodeCreateRootAcct() {
   return sdbInsertRow(&row);
 }
 
+int32_t mnodeCompactAccts() {
+  void *pIter = NULL;
+  SAcctObj *pAcct = NULL;
+
+  mInfo("start to compact accts table...");
+
+  while (1) {
+    pIter = mnodeGetNextAcct(pIter, &pAcct);
+    if (pAcct == NULL) break;
+
+    SSdbRow row = {
+      .type   = SDB_OPER_GLOBAL,
+      .pTable = tsAcctSdb,
+      .pObj   = pAcct,
+    };
+
+    mInfo("compact accts %s", pAcct->user);
+    
+    sdbInsertCompactRow(&row);
+  }
+
+  mInfo("end to compact accts table...");
+
+  return 0;
+}
+
 #ifndef _ACCT
 
 int32_t acctInit() { return TSDB_CODE_SUCCESS; }

src/mnode/src/mnodeCluster.c

@@ -237,3 +237,27 @@ static int32_t mnodeRetrieveClusters(SShowObj *pShow, char *data, int32_t rows,
   pShow->numOfReads += numOfRows;
   return numOfRows;
 }
+
+int32_t mnodeCompactCluster() {
+  SClusterObj *pCluster = NULL;
+  void *pIter;
+  
+  mInfo("start to compact cluster table...");
+
+  pIter = mnodeGetNextCluster(NULL, &pCluster);
+  while (pCluster) {
+    SSdbRow row = {
+      .type   = SDB_OPER_GLOBAL,
+      .pTable = tsClusterSdb,
+      .pObj   = pCluster,      
+    };
+
+    sdbInsertCompactRow(&row);
+
+    pIter = mnodeGetNextCluster(pIter, &pCluster);
+  }
+
+  mInfo("end to compact cluster table...");
+
+  return 0;
+}

src/mnode/src/mnodeDb.c

@@ -1271,3 +1271,30 @@ void  mnodeDropAllDbs(SAcctObj *pAcct)  {
 
   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; 
+}

src/mnode/src/mnodeDnode.c

@@ -1270,3 +1270,30 @@ char* dnodeRoles[] = {
   "vnode",
   "any"
 };
+
+int32_t mnodeCompactDnodes() {
+  SDnodeObj *pDnode = NULL;
+  void *     pIter = NULL;
+
+  mInfo("start to compact dnodes table...");
+
+  while (1) {
+    pIter = mnodeGetNextDnode(pIter, &pDnode);
+    if (pDnode == NULL) break;
+
+    SSdbRow row = {
+      .type    = SDB_OPER_GLOBAL,
+      .pTable  = tsDnodeSdb,
+      .pObj    = pDnode,
+      .rowSize = sizeof(SDnodeObj),
+    };
+
+    mInfo("compact dnode %d", pDnode->dnodeId);
+    
+    sdbInsertCompactRow(&row);
+  }
+
+  mInfo("end to compact dnodes table...");
+
+  return 0;
+}

src/mnode/src/mnodeMain.c

@@ -57,6 +57,18 @@ static SStep tsMnodeSteps[] = {
   {"show",    mnodeInitShow,    mnodeCleanUpShow}
 };
 
+static SStep tsMnodeCompactSteps[] = {
+  {"cluster", mnodeCompactCluster, NULL},
+  {"dnodes",  mnodeCompactDnodes,  NULL},
+  {"mnodes",  mnodeCompactMnodes,  NULL},
+  {"accts",   mnodeCompactAccts,  NULL},
+  {"users",   mnodeCompactUsers,  NULL},
+  {"dbs",     mnodeCompactDbs,     NULL},
+  {"vgroups", mnodeCompactVgroups, NULL},
+  {"tables",  mnodeCompactTables,  NULL}, 
+
+};
+
 static void mnodeInitTimer();
 static void mnodeCleanupTimer();
 static bool mnodeNeedStart() ;
@@ -71,6 +83,11 @@ static int32_t mnodeInitComponents() {
   return dnodeStepInit(tsMnodeSteps, stepSize);
 }
 
+int32_t mnodeCompactComponents() {
+  int32_t stepSize = sizeof(tsMnodeCompactSteps) / sizeof(SStep);
+  return dnodeStepInit(tsMnodeCompactSteps, stepSize);
+}
+
 int32_t mnodeStartSystem() {
   if (tsMgmtIsRunning) {
     mInfo("mnode module already started...");

src/mnode/src/mnodeMnode.c

@@ -566,3 +566,30 @@ static int32_t mnodeRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, vo
 
   return numOfRows;
 }
+
+int32_t mnodeCompactMnodes() {
+  void *pIter = NULL;
+  SMnodeObj *pMnode = NULL;
+
+  mInfo("start to compact mnodes table...");
+
+  while (1) {
+    pIter = mnodeGetNextMnode(pIter, &pMnode);
+    if (pMnode == NULL) break;
+
+    SSdbRow row = {
+      .type    = SDB_OPER_GLOBAL,
+      .pTable  = tsMnodeSdb,
+      .pObj    = pMnode,
+      .rowSize = sizeof(SMnodeObj),
+    };
+
+    mInfo("compact mnode %d", pMnode->mnodeId);
+    
+    sdbInsertCompactRow(&row);
+  }
+
+  mInfo("end to compact mnodes table...");
+
+  return 0; 
+}

src/mnode/src/mnodeSdb.c

@@ -20,6 +20,7 @@
 #include "tutil.h"
 #include "tref.h"
 #include "tbn.h"
+#include "tfs.h"
 #include "tqueue.h"
 #include "twal.h"
 #include "tsync.h"
@@ -450,6 +451,12 @@ int32_t sdbInit() {
   }
 
   tsSdbMgmt.status = SDB_STATUS_SERVING;
+
+  if (tsCompactMnodeWal) {
+    mnodeCompactWal();
+    exit(EXIT_SUCCESS);
+  }
+
   return TSDB_CODE_SUCCESS;
 }
 
@@ -726,6 +733,12 @@ static int32_t sdbProcessWrite(void *wparam, void *hparam, int32_t qtype, void *
   }
 }
 
+int32_t sdbInsertCompactRow(SSdbRow *pRow) {
+  SSdbTable *pTable = pRow->pTable;
+  if (pTable == NULL) return TSDB_CODE_MND_SDB_INVALID_TABLE_TYPE;
+  return sdbWriteRowToQueue(pRow, SDB_ACTION_INSERT);
+}
+
 int32_t sdbInsertRow(SSdbRow *pRow) {
   SSdbTable *pTable = pRow->pTable;
   if (pTable == NULL) return TSDB_CODE_MND_SDB_INVALID_TABLE_TYPE;
@@ -1138,3 +1151,46 @@ static void *sdbWorkerFp(void *pWorker) {
 int32_t sdbGetReplicaNum() {
   return tsSdbMgmt.cfg.replica;
 }
+
+int32_t mnodeCompactWal() {
+  sdbInfo("vgId:1, start compact mnode wal...");
+
+  // close old wal
+  walFsync(tsSdbMgmt.wal, true);
+  walClose(tsSdbMgmt.wal);
+
+  // reset version,then compacted wal log can start from version 1
+  tsSdbMgmt.version = 0;
+
+  // change wal to wal_tmp dir
+  SWalCfg walCfg = {.vgId = 1, .walLevel = TAOS_WAL_FSYNC, .keep = TAOS_WAL_KEEP, .fsyncPeriod = 0};
+  char    temp[TSDB_FILENAME_LEN] = {0};
+  sprintf(temp, "%s/wal", tsMnodeTmpDir);
+  tsSdbMgmt.wal = walOpen(temp, &walCfg);
+  walRenew(tsSdbMgmt.wal);
+
+  // compact memory tables info to wal tmp dir
+  if (mnodeCompactComponents() != 0) {
+    tfsRmdir(tsMnodeTmpDir);
+    return -1;
+  }
+
+  // close wal
+  walFsync(tsSdbMgmt.wal, true);
+  walClose(tsSdbMgmt.wal);
+
+  // rename old wal to wal_bak
+  if (taosRename(tsMnodeDir, tsMnodeBakDir) != 0) {
+    return -1;
+  }
+
+  // rename wal_tmp to wal
+  if (taosRename(tsMnodeTmpDir, tsMnodeDir) != 0) {
+    return -1;
+  }
+  
+  // del wal_tmp dir
+  sdbInfo("vgId:1, compact mnode wal success");
+
+  return 0;
+}

src/mnode/src/mnodeTable.c

@@ -3242,3 +3242,65 @@ static int32_t mnodeRetrieveStreamTables(SShowObj *pShow, char *data, int32_t ro
 
   return numOfRows;
 }
+
+static int32_t mnodeCompactSuperTables() {
+  void *pIter = NULL;
+  SSTableObj *pTable = NULL;
+
+  mInfo("start to compact super table...");
+
+  while (1) {
+    pIter = mnodeGetNextSuperTable(pIter, &pTable);
+    if (pTable == NULL) break;
+
+    int32_t schemaSize = (pTable->numOfColumns + pTable->numOfTags) * sizeof(SSchema);
+    SSdbRow row = {
+      .type    = SDB_OPER_GLOBAL,
+      .pTable  = tsSuperTableSdb,
+      .pObj    = pTable,
+      .rowSize = sizeof(SSTableObj) + schemaSize,
+    };
+
+    mInfo("compact super %" PRIu64, pTable->uid);
+    
+    sdbInsertCompactRow(&row);
+  }
+
+  mInfo("end to compact super table...");
+
+  return 0; 
+}
+
+static int32_t mnodeCompactChildTables() {
+  void *pIter = NULL;
+  SCTableObj *pTable = NULL;
+
+  mInfo("start to compact child table...");
+
+  while (1) {
+    pIter = mnodeGetNextChildTable(pIter, &pTable);
+    if (pTable == NULL) break;
+
+    SSdbRow row = {
+      .type   = SDB_OPER_GLOBAL,
+      .pObj   = pTable,
+      .pTable = tsChildTableSdb,
+    };
+
+    mInfo("compact child %" PRIu64 ":%d", pTable->uid, pTable->tid);
+    
+    sdbInsertCompactRow(&row);
+  }
+
+  mInfo("end to compact child table...");
+
+  return 0; 
+}
+
+int32_t mnodeCompactTables() {
+  mnodeCompactSuperTables();
+
+  mnodeCompactChildTables();
+
+  return 0;
+}

src/mnode/src/mnodeUser.c

@@ -617,3 +617,30 @@ static int32_t mnodeProcessAuthMsg(SMnodeMsg *pMsg) {
   
   return mnodeRetriveAuth(pAuthMsg->user, &pAuthRsp->spi, &pAuthRsp->encrypt, pAuthRsp->secret, pAuthRsp->ckey);
 }
+
+int32_t mnodeCompactUsers() {
+  void *pIter = NULL;
+  SUserObj *pUser = NULL;
+
+  mInfo("start to compact users table...");
+
+  while (1) {
+    pIter = mnodeGetNextUser(pIter, &pUser);
+    if (pUser == NULL) break;
+
+    SSdbRow row = {
+      .type     = SDB_OPER_GLOBAL,
+      .pTable   = tsUserSdb,
+      .pObj     = pUser,
+      .rowSize  = sizeof(SUserObj),
+    };
+
+    mInfo("compact users %s", pUser->user);
+    
+    sdbInsertCompactRow(&row);
+  }
+
+  mInfo("end to compact users table...");
+
+  return 0;
+}

src/mnode/src/mnodeVgroup.c

@@ -1302,3 +1302,30 @@ void mnodeSetVgidVer(int8_t *cver, uint64_t iver) {
   cver[1] = (int8_t)((int32_t)(iver % 100000) / 100);
   cver[2] = (int8_t)(iver % 100);
 }
+
+int32_t mnodeCompactVgroups() {
+  void *pIter = NULL;
+  SVgObj *pVgroup = NULL;
+
+  mInfo("start to compact vgroups table...");
+
+  while (1) {
+    pIter = mnodeGetNextVgroup(pIter, &pVgroup);
+    if (pVgroup == NULL) break;
+
+    SSdbRow row = {
+      .type     = SDB_OPER_GLOBAL,
+      .pTable   = tsVgroupSdb,
+      .pObj     = pVgroup,
+      .rowSize  = sizeof(SVgObj),
+    };
+
+    mInfo("compact vgroups %d", pVgroup->vgId);
+    
+    sdbInsertCompactRow(&row);
+  }
+
+  mInfo("end to compact vgroups table...");
+
+  return 0; 
+}