Merge pull request #20222 from taosdata/feat/compact

more compact to enterprise

source/dnode/mnode/impl/CMakeLists.txt

@@ -2,8 +2,9 @@ aux_source_directory(src MNODE_SRC)
 IF (TD_PRIVILEGE)
   ADD_DEFINITIONS(-D_PRIVILEGE)
 ENDIF ()
-IF (TD_PRIVILEGE)
+IF (TD_ENTERPRISE)
   LIST(APPEND MNODE_SRC ${TD_ENTERPRISE_DIR}/src/plugins/privilege/src/privilege.c)
+  LIST(APPEND MNODE_SRC ${TD_ENTERPRISE_DIR}/src/plugins/mnode/src/mndDb.c)
 ENDIF ()
 
 add_library(mnode STATIC ${MNODE_SRC})

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

@@ -33,6 +33,8 @@ bool    mndIsDbReady(SMnode *pMnode, SDbObj *pDb);
 SSdbRaw    *mndDbActionEncode(SDbObj *pDb);
 const char *mndGetDbStr(const char *src);
 
+int32_t mndProcessCompactDbReq(SRpcMsg *pReq);
+
 #ifdef __cplusplus
 }
 #endif

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

@@ -41,12 +41,15 @@ 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  mndProcessCompactDbReq(SRpcMsg *pReq);
 static int32_t  mndProcessTrimDbReq(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,
@@ -1395,98 +1398,6 @@ int32_t mndValidateDbInfo(SMnode *pMnode, SDbVgVersion *pDbs, int32_t numOfDbs,
   return 0;
 }
 
-static int32_t mndSetCompactDbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, int64_t compactTs) {
-  SDbObj dbObj = {0};
-  memcpy(&dbObj, pDb, sizeof(SDbObj));
-  dbObj.compactStartTime = compactTs;
-
-  SSdbRaw *pCommitRaw = mndDbActionEncode(&dbObj);
-  if (pCommitRaw == NULL) return -1;
-  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
-    sdbFreeRaw(pCommitRaw);
-    return -1;
-  }
-
-  (void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
-  return 0;
-}
-
-static int32_t mndSetCompactDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, int64_t compactTs) {
-  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 (mndVgroupInDb(pVgroup, pDb->uid)) {
-      if (mndBuildCompactVgroupAction(pMnode, pTrans, pDb, pVgroup, compactTs) != 0) {
-        sdbCancelFetch(pSdb, pIter);
-        sdbRelease(pSdb, pVgroup);
-        return -1;
-      }
-    }
-
-    sdbRelease(pSdb, pVgroup);
-  }
-
-  return 0;
-}
-
-static int32_t mndCompactDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb) {
-  int64_t compactTs = taosGetTimestampMs();
-  int32_t code = -1;
-  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq, "compact-db");
-  if (pTrans == NULL) goto _OVER;
-
-  mInfo("trans:%d, used to compact db:%s", pTrans->id, pDb->name);
-  mndTransSetDbName(pTrans, pDb->name, NULL);
-  if (mndTrancCheckConflict(pMnode, pTrans) != 0) goto _OVER;
-  if (mndSetCompactDbCommitLogs(pMnode, pTrans, pDb, compactTs) != 0) goto _OVER;
-  if (mndSetCompactDbRedoActions(pMnode, pTrans, pDb, compactTs) != 0) goto _OVER;
-  if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
-  code = 0;
-
-_OVER:
-  mndTransDrop(pTrans);
-  return code;
-}
-
-static int32_t mndProcessCompactDbReq(SRpcMsg *pReq) {
-  SMnode       *pMnode = pReq->info.node;
-  int32_t       code = -1;
-  SDbObj       *pDb = NULL;
-  SCompactDbReq compactReq = {0};
-
-  if (tDeserializeSCompactDbReq(pReq->pCont, pReq->contLen, &compactReq) != 0) {
-    terrno = TSDB_CODE_INVALID_MSG;
-    goto _OVER;
-  }
-
-  mInfo("db:%s, start to compact", compactReq.db);
-
-  pDb = mndAcquireDb(pMnode, compactReq.db);
-  if (pDb == NULL) {
-    goto _OVER;
-  }
-
-  if (mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_COMPACT_DB, pDb) != 0) {
-    goto _OVER;
-  }
-
-  code = mndCompactDb(pMnode, pReq, pDb);
-  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
-
-_OVER:
-  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
-    mError("db:%s, failed to process compact db req since %s", compactReq.db, terrstr());
-  }
-
-  mndReleaseDb(pMnode, pDb);
-  return code;
-}
-
 static int32_t mndTrimDb(SMnode *pMnode, SDbObj *pDb) {
   SSdb       *pSdb = pMnode->pSdb;
   SVgObj     *pVgroup = NULL;

source/dnode/vnode/CMakeLists.txt

@@ -14,7 +14,6 @@ target_sources(
     "src/vnd/vnodeSvr.c"
     "src/vnd/vnodeSync.c"
     "src/vnd/vnodeSnapshot.c"
-    "src/vnd/vnodeCompact.c"
     "src/vnd/vnodeRetention.c"
 
     # meta
@@ -53,7 +52,6 @@ target_sources(
     "src/tsdb/tsdbCacheRead.c"
     "src/tsdb/tsdbRetention.c"
     "src/tsdb/tsdbDiskData.c"
-    "src/tsdb/tsdbCompact.c"
     "src/tsdb/tsdbMergeTree.c"
     "src/tsdb/tsdbDataIter.c"
 
@@ -69,10 +67,20 @@ target_sources(
     "src/tq/tqSnapshot.c"
     "src/tq/tqOffsetSnapshot.c"
 )
+
+IF (TD_VNODE_PLUGINS)
+  target_sources(
+    vnode
+    PRIVATE
+    ${TD_ENTERPRISE_DIR}/src/plugins/vnode/src/tsdbCompact.c
+    ${TD_ENTERPRISE_DIR}/src/plugins/vnode/src/vnodeCompact.c
+  )
+ENDIF ()
+
 target_include_directories(
     vnode
     PUBLIC "inc"
-    PRIVATE "src/inc"
+    PUBLIC "src/inc"
     PUBLIC "${TD_SOURCE_DIR}/include/libs/scalar"
 )
 target_link_libraries(

source/dnode/vnode/src/inc/vnd.h

@@ -106,10 +106,6 @@ int32_t vnodeSyncCommit(SVnode* pVnode);
 int32_t vnodeAsyncCommit(SVnode* pVnode);
 bool    vnodeShouldRollback(SVnode* pVnode);
 
-// vnodeCompact.c
-int32_t vnodeAsyncCompact(SVnode* pVnode);
-int32_t vnodeSyncCompact(SVnode* pVnode);
-
 // vnodeSync.c
 int32_t vnodeSyncOpen(SVnode* pVnode, char* path);
 int32_t vnodeSyncStart(SVnode* pVnode);

source/dnode/vnode/src/vnd/vnodeCompact.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/>.
- */
-
-#include "vnd.h"
-
-extern int32_t tsdbCommitCompact(STsdb *pTsdb);
-
-static int32_t vnodeCompactTask(void *param) {
-  int32_t code = 0;
-  int32_t lino = 0;
-
-  SCompactInfo *pInfo = (SCompactInfo *)param;
-  SVnode       *pVnode = pInfo->pVnode;
-
-  // do compact
-  code = tsdbCompact(pInfo->pVnode->pTsdb, pInfo);
-  TSDB_CHECK_CODE(code, lino, _exit);
-
-  // end compact
-  char dir[TSDB_FILENAME_LEN] = {0};
-  if (pVnode->pTfs) {
-    snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path);
-  } else {
-    snprintf(dir, TSDB_FILENAME_LEN, "%s", pVnode->path);
-  }
-
-  vnodeCommitInfo(dir);
-
-  tsdbCommitCompact(pVnode->pTsdb);
-
-_exit:
-  tsem_post(&pInfo->pVnode->canCommit);
-  taosMemoryFree(pInfo);
-  return code;
-}
-static int32_t vnodePrepareCompact(SVnode *pVnode, SCompactInfo *pInfo) {
-  int32_t code = 0;
-  int32_t lino = 0;
-
-  tsem_wait(&pVnode->canCommit);
-
-  pInfo->pVnode = pVnode;
-  pInfo->flag = 0;
-  pInfo->commitID = ++pVnode->state.commitID;
-
-  char       dir[TSDB_FILENAME_LEN] = {0};
-  SVnodeInfo info = {0};
-
-  if (pVnode->pTfs) {
-    snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path);
-  } else {
-    snprintf(dir, TSDB_FILENAME_LEN, "%s", pVnode->path);
-  }
-
-  if (vnodeLoadInfo(dir, &info) < 0) {
-    code = terrno;
-    goto _exit;
-  }
-
-  info.state.commitID = pInfo->commitID;
-
-  if (vnodeSaveInfo(dir, &info) < 0) {
-    code = terrno;
-    goto _exit;
-  }
-
-_exit:
-  if (code) {
-    vError("vgId:%d %s failed at line %d since %s, commit ID:%" PRId64, TD_VID(pVnode), __func__, lino, tstrerror(code),
-           pVnode->state.commitID);
-  } else {
-    vDebug("vgId:%d %s done, commit ID:%" PRId64, TD_VID(pVnode), __func__, pVnode->state.commitID);
-  }
-  return code;
-}
-int32_t vnodeAsyncCompact(SVnode *pVnode) {
-  int32_t code = 0;
-  int32_t lino = 0;
-
-  SCompactInfo *pInfo = taosMemoryCalloc(1, sizeof(*pInfo));
-  if (pInfo == NULL) {
-    code = TSDB_CODE_OUT_OF_MEMORY;
-    TSDB_CHECK_CODE(code, lino, _exit);
-  }
-
-  vnodeAsyncCommit(pVnode);
-
-  code = vnodePrepareCompact(pVnode, pInfo);
-  TSDB_CHECK_CODE(code, lino, _exit);
-
-  vnodeScheduleTask(vnodeCompactTask, pInfo);
-
-_exit:
-  if (code) {
-    vError("vgId:%d %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
-    if (pInfo) taosMemoryFree(pInfo);
-  } else {
-    vInfo("vgId:%d %s done", TD_VID(pVnode), __func__);
-  }
-  return code;
-}
-
-int32_t vnodeSyncCompact(SVnode *pVnode) {
-  vnodeAsyncCompact(pVnode);
-  tsem_wait(&pVnode->canCommit);
-  tsem_post(&pVnode->canCommit);
-  return 0;
-}
\ No newline at end of file

source/dnode/vnode/src/vnd/vnodeSvr.c

@@ -1641,17 +1641,13 @@ static int32_t vnodeProcessDropIndexReq(SVnode *pVnode, int64_t version, void *p
   return TSDB_CODE_SUCCESS;
 }
 
+extern int32_t vnodeProcessCompactVnodeReqImpl(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
+
 static int32_t vnodeProcessCompactVnodeReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
-  SCompactVnodeReq req = {0};
-  if (tDeserializeSCompactVnodeReq(pReq, len, &req) != 0) {
-    terrno = TSDB_CODE_INVALID_MSG;
-    return TSDB_CODE_INVALID_MSG;
-  }
-  vInfo("vgId:%d, compact msg will be processed, db:%s dbUid:%" PRId64 " compactStartTime:%" PRId64, TD_VID(pVnode),
-        req.db, req.dbUid, req.compactStartTime);
+  return vnodeProcessCompactVnodeReqImpl(pVnode, version, pReq, len, pRsp);
+}
 
-  vnodeAsyncCompact(pVnode);
-  vnodeBegin(pVnode);
 
-  return 0;
-}
+#ifndef TD_ENTERPRISE
+int32_t vnodeProcessCompactVnodeReqImpl(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) { return 0; }
+#endif