xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #26181 from taosdata/fix/TD-30625-3.0 

fix/TD-30625
This commit is contained in:
Hongze Cheng 2024-06-20 17:06:44 +08:00 committed by GitHub
parent 70851f6bed 39c5e9844c
commit ea7709a303
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 132 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

244
source/dnode/mnode/impl/src/mndCompact.c
View File

@ -13,20 +13,19 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "mndCompact.h" #include "mndCompact.h"
#include "mndTrans.h"
#include "mndShow.h"
#include "mndDb.h"
#include "mndCompactDetail.h"
#include "mndVgroup.h"
#include "tmsgcb.h"
#include "mndDnode.h"
#include "tmisce.h"
#include "audit.h" #include "audit.h"
#include "mndCompactDetail.h"
#include "mndDb.h"
#include "mndDnode.h"
#include "mndPrivilege.h" #include "mndPrivilege.h"
#include "mndShow.h"
#include "mndTrans.h" #include "mndTrans.h"
#include "mndVgroup.h"
#include "tmisce.h"
#include "tmsgcb.h"
#define MND_COMPACT_VER_NUMBER 1 #define MND_COMPACT_VER_NUMBER 1
#define MND_COMPACT_ID_LEN 11 #define MND_COMPACT_ID_LEN 11
static int32_t mndProcessCompactTimer(SRpcMsg *pReq); static int32_t mndProcessCompactTimer(SRpcMsg *pReq);
@ -50,12 +49,9 @@ int32_t mndInitCompact(SMnode *pMnode) {
return sdbSetTable(pMnode->pSdb, table); return sdbSetTable(pMnode->pSdb, table);
} }
void mndCleanupCompact(SMnode *pMnode) { void mndCleanupCompact(SMnode *pMnode) { mDebug("mnd compact cleanup"); }
mDebug("mnd compact cleanup");
}
void tFreeCompactObj(SCompactObj *pCompact) { void tFreeCompactObj(SCompactObj *pCompact) {}
}
int32_t tSerializeSCompactObj(void *buf, int32_t bufLen, const SCompactObj *pObj) { int32_t tSerializeSCompactObj(void *buf, int32_t bufLen, const SCompactObj *pObj) {
SEncoder encoder = {0}; SEncoder encoder = {0};
@ -75,7 +71,7 @@ int32_t tSerializeSCompactObj(void *buf, int32_t bufLen, const SCompactObj *pObj
} }
int32_t tDeserializeSCompactObj(void *buf, int32_t bufLen, SCompactObj *pObj) { int32_t tDeserializeSCompactObj(void *buf, int32_t bufLen, SCompactObj *pObj) {
int8_t ex = 0; int8_t ex = 0;
SDecoder decoder = {0}; SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen); tDecoderInit(&decoder, buf, bufLen);
@ -94,7 +90,7 @@ int32_t tDeserializeSCompactObj(void *buf, int32_t bufLen, SCompactObj *pObj) {
SSdbRaw *mndCompactActionEncode(SCompactObj *pCompact) { SSdbRaw *mndCompactActionEncode(SCompactObj *pCompact) {
terrno = TSDB_CODE_SUCCESS; terrno = TSDB_CODE_SUCCESS;
void *buf = NULL; void *buf = NULL;
SSdbRaw *pRaw = NULL; SSdbRaw *pRaw = NULL;
int32_t tlen = tSerializeSCompactObj(NULL, 0, pCompact); int32_t tlen = tSerializeSCompactObj(NULL, 0, pCompact);
@ -102,8 +98,8 @@ SSdbRaw *mndCompactActionEncode(SCompactObj *pCompact) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER; goto OVER;
} }
int32_t size = sizeof(int32_t) + tlen; int32_t size = sizeof(int32_t) + tlen;
pRaw = sdbAllocRaw(SDB_COMPACT, MND_COMPACT_VER_NUMBER, size); pRaw = sdbAllocRaw(SDB_COMPACT, MND_COMPACT_VER_NUMBER, size);
if (pRaw == NULL) { if (pRaw == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -127,7 +123,6 @@ SSdbRaw *mndCompactActionEncode(SCompactObj *pCompact) {
SDB_SET_BINARY(pRaw, dataPos, buf, tlen, OVER); SDB_SET_BINARY(pRaw, dataPos, buf, tlen, OVER);
SDB_SET_DATALEN(pRaw, dataPos, OVER); SDB_SET_DATALEN(pRaw, dataPos, OVER);
OVER: OVER:
taosMemoryFreeClear(buf); taosMemoryFreeClear(buf);
if (terrno != TSDB_CODE_SUCCESS) { if (terrno != TSDB_CODE_SUCCESS) {
@ -141,9 +136,9 @@ OVER:
} }
SSdbRow *mndCompactActionDecode(SSdbRaw *pRaw) { SSdbRow *mndCompactActionDecode(SSdbRaw *pRaw) {
SSdbRow *pRow = NULL; SSdbRow *pRow = NULL;
SCompactObj *pCompact = NULL; SCompactObj *pCompact = NULL;
void *buf = NULL; void *buf = NULL;
terrno = TSDB_CODE_SUCCESS; terrno = TSDB_CODE_SUCCESS;
int8_t sver = 0; int8_t sver = 0;
@ -184,7 +179,7 @@ SSdbRow *mndCompactActionDecode(SSdbRaw *pRaw) {
goto OVER; goto OVER;
} }
//taosInitRWLatch(&pView->lock); // taosInitRWLatch(&pView->lock);
OVER: OVER:
taosMemoryFreeClear(buf); taosMemoryFreeClear(buf);
@ -210,15 +205,15 @@ int32_t mndCompactActionDelete(SSdb *pSdb, SCompactObj *pCompact) {
} }
int32_t mndCompactActionUpdate(SSdb *pSdb, SCompactObj *pOldCompact, SCompactObj *pNewCompact) { int32_t mndCompactActionUpdate(SSdb *pSdb, SCompactObj *pOldCompact, SCompactObj *pNewCompact) {
mTrace("compact:%" PRId32 ", perform update action, old row:%p new row:%p", mTrace("compact:%" PRId32 ", perform update action, old row:%p new row:%p", pOldCompact->compactId, pOldCompact,
pOldCompact->compactId, pOldCompact, pNewCompact); pNewCompact);
return 0; return 0;
} }
SCompactObj *mndAcquireCompact(SMnode *pMnode, int64_t compactId) { SCompactObj *mndAcquireCompact(SMnode *pMnode, int64_t compactId) {
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
SCompactObj *pCompact = sdbAcquire(pSdb, SDB_COMPACT, &compactId); SCompactObj *pCompact = sdbAcquire(pSdb, SDB_COMPACT, &compactId);
if (pCompact == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) { if (pCompact == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
terrno = TSDB_CODE_SUCCESS; terrno = TSDB_CODE_SUCCESS;
} }
@ -230,8 +225,8 @@ void mndReleaseCompact(SMnode *pMnode, SCompactObj *pCompact) {
sdbRelease(pSdb, pCompact); sdbRelease(pSdb, pCompact);
} }
//compact db // compact db
int32_t mndAddCompactToTran(SMnode *pMnode, STrans *pTrans, SCompactObj* pCompact, SDbObj *pDb, SCompactDbRsp *rsp){ int32_t mndAddCompactToTran(SMnode *pMnode, STrans *pTrans, SCompactObj *pCompact, SDbObj *pDb, SCompactDbRsp *rsp) {
pCompact->compactId = tGenIdPI32(); pCompact->compactId = tGenIdPI32();
strcpy(pCompact->dbname, pDb->name); strcpy(pCompact->dbname, pDb->name);
@ -251,18 +246,19 @@ int32_t mndAddCompactToTran(SMnode *pMnode, STrans *pTrans, SCompactObj* pCompac
return 0; return 0;
} }
//retrieve compact // retrieve compact
int32_t mndRetrieveCompact(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows){ int32_t mndRetrieveCompact(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
SMnode *pMnode = pReq->info.node; SMnode *pMnode = pReq->info.node;
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0; int32_t numOfRows = 0;
SCompactObj *pCompact = NULL; SCompactObj *pCompact = NULL;
char *sep = NULL; char *sep = NULL;
SDbObj *pDb = NULL; SDbObj *pDb = NULL;
if (strlen(pShow->db) > 0) { if (strlen(pShow->db) > 0) {
sep = strchr(pShow->db, '.'); sep = strchr(pShow->db, '.');
if (sep && ((0 == strcmp(sep + 1, TSDB_INFORMATION_SCHEMA_DB) || (0 == strcmp(sep + 1, TSDB_PERFORMANCE_SCHEMA_DB))))) { if (sep &&
((0 == strcmp(sep + 1, TSDB_INFORMATION_SCHEMA_DB) || (0 == strcmp(sep + 1, TSDB_PERFORMANCE_SCHEMA_DB))))) {
sep++; sep++;
} else { } else {
pDb = mndAcquireDb(pMnode, pShow->db); pDb = mndAcquireDb(pMnode, pShow->db);
@ -306,9 +302,9 @@ int32_t mndRetrieveCompact(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock,
return numOfRows; return numOfRows;
} }
//kill compact // kill compact
static void *mndBuildKillCompactReq(SMnode *pMnode, SVgObj *pVgroup, int32_t *pContLen, static void *mndBuildKillCompactReq(SMnode *pMnode, SVgObj *pVgroup, int32_t *pContLen, int32_t compactId,
int32_t compactId, int32_t dnodeid) { int32_t dnodeid) {
SVKillCompactReq req = {0}; SVKillCompactReq req = {0};
req.compactId = compactId; req.compactId = compactId;
req.vgId = pVgroup->vgId; req.vgId = pVgroup->vgId;
@ -337,8 +333,8 @@ static void *mndBuildKillCompactReq(SMnode *pMnode, SVgObj *pVgroup, int32_t *pC
return pReq; return pReq;
} }
static int32_t mndAddKillCompactAction(SMnode *pMnode, STrans *pTrans, SVgObj *pVgroup, static int32_t mndAddKillCompactAction(SMnode *pMnode, STrans *pTrans, SVgObj *pVgroup, int32_t compactId,
int32_t compactId, int32_t dnodeid) { int32_t dnodeid) {
STransAction action = {0}; STransAction action = {0};
SDnodeObj *pDnode = mndAcquireDnode(pMnode, dnodeid); SDnodeObj *pDnode = mndAcquireDnode(pMnode, dnodeid);
@ -365,7 +361,7 @@ static int32_t mndAddKillCompactAction(SMnode *pMnode, STrans *pTrans, SVgObj *p
static int32_t mndKillCompact(SMnode *pMnode, SRpcMsg *pReq, SCompactObj *pCompact) { static int32_t mndKillCompact(SMnode *pMnode, SRpcMsg *pReq, SCompactObj *pCompact) {
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB, pReq, "kill-compact"); STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB, pReq, "kill-compact");
if (pTrans == NULL) { if (pTrans == NULL) {
mError("compact:%" PRId32 ", failed to drop since %s" , pCompact->compactId, terrstr()); mError("compact:%" PRId32 ", failed to drop since %s", pCompact->compactId, terrstr());
return -1; return -1;
} }
mInfo("trans:%d, used to kill compact:%" PRId32, pTrans->id, pCompact->compactId); mInfo("trans:%d, used to kill compact:%" PRId32, pTrans->id, pCompact->compactId);
@ -388,13 +384,13 @@ static int32_t mndKillCompact(SMnode *pMnode, SRpcMsg *pReq, SCompactObj *pCompa
if (pDetail->compactId == pCompact->compactId) { if (pDetail->compactId == pCompact->compactId) {
SVgObj *pVgroup = mndAcquireVgroup(pMnode, pDetail->vgId); SVgObj *pVgroup = mndAcquireVgroup(pMnode, pDetail->vgId);
if(pVgroup == NULL){ if (pVgroup == NULL) {
mError("trans:%d, failed to append redo action since %s", pTrans->id, terrstr()); mError("trans:%d, failed to append redo action since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
} }
if(mndAddKillCompactAction(pMnode, pTrans, pVgroup, pCompact->compactId, pDetail->dnodeId) != 0){ if (mndAddKillCompactAction(pMnode, pTrans, pVgroup, pCompact->compactId, pDetail->dnodeId) != 0) {
mError("trans:%d, failed to append redo action since %s", pTrans->id, terrstr()); mError("trans:%d, failed to append redo action since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
@ -426,7 +422,7 @@ static int32_t mndKillCompact(SMnode *pMnode, SRpcMsg *pReq, SCompactObj *pCompa
return 0; return 0;
} }
int32_t mndProcessKillCompactReq(SRpcMsg *pReq){ int32_t mndProcessKillCompactReq(SRpcMsg *pReq) {
SKillCompactReq killCompactReq = {0}; SKillCompactReq killCompactReq = {0};
if (tDeserializeSKillCompactReq(pReq->pCont, pReq->contLen, &killCompactReq) != 0) { if (tDeserializeSKillCompactReq(pReq->pCont, pReq->contLen, &killCompactReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG; terrno = TSDB_CODE_INVALID_MSG;
@ -435,10 +431,10 @@ int32_t mndProcessKillCompactReq(SRpcMsg *pReq){
mInfo("start to kill compact:%" PRId32, killCompactReq.compactId); mInfo("start to kill compact:%" PRId32, killCompactReq.compactId);
SMnode *pMnode = pReq->info.node; SMnode *pMnode = pReq->info.node;
int32_t code = -1; int32_t code = -1;
SCompactObj *pCompact = mndAcquireCompact(pMnode, killCompactReq.compactId); SCompactObj *pCompact = mndAcquireCompact(pMnode, killCompactReq.compactId);
if(pCompact == NULL){ if (pCompact == NULL) {
terrno = TSDB_CODE_MND_INVALID_COMPACT_ID; terrno = TSDB_CODE_MND_INVALID_COMPACT_ID;
tFreeSKillCompactReq(&killCompactReq); tFreeSKillCompactReq(&killCompactReq);
return -1; return -1;
@ -470,9 +466,10 @@ _OVER:
return code; return code;
} }
//update progress // update progress
static int32_t mndUpdateCompactProgress(SMnode *pMnode, SRpcMsg *pReq, int32_t compactId, SQueryCompactProgressRsp* rsp) { static int32_t mndUpdateCompactProgress(SMnode *pMnode, SRpcMsg *pReq, int32_t compactId,
void* pIter = NULL; SQueryCompactProgressRsp *rsp) {
void *pIter = NULL;
while (1) { while (1) {
SCompactDetailObj *pDetail = NULL; SCompactDetailObj *pDetail = NULL;
pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT_DETAIL, pIter, (void **)&pDetail); pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT_DETAIL, pIter, (void **)&pDetail);
@ -493,36 +490,36 @@ static int32_t mndUpdateCompactProgress(SMnode *pMnode, SRpcMsg *pReq, int32_t c
return TSDB_CODE_MND_COMPACT_DETAIL_NOT_EXIST; return TSDB_CODE_MND_COMPACT_DETAIL_NOT_EXIST;
} }
int32_t mndProcessQueryCompactRsp(SRpcMsg *pReq){ int32_t mndProcessQueryCompactRsp(SRpcMsg *pReq) {
SQueryCompactProgressRsp req = {0}; SQueryCompactProgressRsp req = {0};
int32_t code = 0; int32_t code = 0;
code = tDeserializeSQueryCompactProgressRsp(pReq->pCont, pReq->contLen, &req); code = tDeserializeSQueryCompactProgressRsp(pReq->pCont, pReq->contLen, &req);
if (code != 0) { if (code != 0) {
terrno = TSDB_CODE_INVALID_MSG; terrno = TSDB_CODE_INVALID_MSG;
mError("failed to deserialize vnode-query-compact-progress-rsp, ret:%d, pCont:%p, len:%d", mError("failed to deserialize vnode-query-compact-progress-rsp, ret:%d, pCont:%p, len:%d", code, pReq->pCont,
code, pReq->pCont, pReq->contLen); pReq->contLen);
return -1; return -1;
} }
mDebug("compact:%d, receive query response, vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d", mDebug("compact:%d, receive query response, vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d", req.compactId,
req.compactId, req.vgId, req.dnodeId, req.numberFileset, req.finished); req.vgId, req.dnodeId, req.numberFileset, req.finished);
SMnode *pMnode = pReq->info.node; SMnode *pMnode = pReq->info.node;
code = mndUpdateCompactProgress(pMnode, pReq, req.compactId, &req); code = mndUpdateCompactProgress(pMnode, pReq, req.compactId, &req);
if(code != 0){ if (code != 0) {
terrno = code; terrno = code;
mError("compact:%d, failed to update progress, vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d", mError("compact:%d, failed to update progress, vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d", req.compactId,
req.compactId, req.vgId, req.dnodeId, req.numberFileset, req.finished); req.vgId, req.dnodeId, req.numberFileset, req.finished);
return -1; return -1;
} }
return 0; return 0;
} }
//timer // timer
void mndCompactSendProgressReq(SMnode *pMnode, SCompactObj *pCompact){ void mndCompactSendProgressReq(SMnode *pMnode, SCompactObj *pCompact) {
void *pIter = NULL; void *pIter = NULL;
while (1) { while (1) {
SCompactDetailObj *pDetail = NULL; SCompactDetailObj *pDetail = NULL;
@ -532,8 +529,8 @@ void mndCompactSendProgressReq(SMnode *pMnode, SCompactObj *pCompact){
if (pDetail->compactId == pCompact->compactId) { if (pDetail->compactId == pCompact->compactId) {
SEpSet epSet = {0}; SEpSet epSet = {0};
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pDetail->dnodeId); SDnodeObj *pDnode = mndAcquireDnode(pMnode, pDetail->dnodeId);
if(pDnode == NULL) break; if (pDnode == NULL) break;
addEpIntoEpSet(&epSet, pDnode->fqdn, pDnode->port); addEpIntoEpSet(&epSet, pDnode->fqdn, pDnode->port);
mndReleaseDnode(pMnode, pDnode); mndReleaseDnode(pMnode, pDnode);
@ -555,31 +552,29 @@ void mndCompactSendProgressReq(SMnode *pMnode, SCompactObj *pCompact){
sdbCancelFetch(pMnode->pSdb, pDetail); sdbCancelFetch(pMnode->pSdb, pDetail);
sdbRelease(pMnode->pSdb, pDetail); sdbRelease(pMnode->pSdb, pDetail);
continue; continue;
} }
pHead->contLen = htonl(contLen); pHead->contLen = htonl(contLen);
pHead->vgId = htonl(pDetail->vgId); pHead->vgId = htonl(pDetail->vgId);
tSerializeSQueryCompactProgressReq((char *)pHead + sizeof(SMsgHead), contLen - sizeof(SMsgHead), &req); tSerializeSQueryCompactProgressReq((char *)pHead + sizeof(SMsgHead), contLen - sizeof(SMsgHead), &req);
SRpcMsg rpcMsg = {.msgType = TDMT_VND_QUERY_COMPACT_PROGRESS, SRpcMsg rpcMsg = {.msgType = TDMT_VND_QUERY_COMPACT_PROGRESS, .contLen = contLen};
.contLen = contLen};
//rpcMsg.pCont = rpcMallocCont(contLen);
//if (rpcMsg.pCont == NULL) {
// return;
//}
//memcpy(rpcMsg.pCont, pHead, contLen); // rpcMsg.pCont = rpcMallocCont(contLen);
// if (rpcMsg.pCont == NULL) {
// return;
// }
// memcpy(rpcMsg.pCont, pHead, contLen);
rpcMsg.pCont = pHead; rpcMsg.pCont = pHead;
char detail[1024] = {0}; char detail[1024] = {0};
int32_t len = snprintf(detail, sizeof(detail), "msgType:%s numOfEps:%d inUse:%d", TMSG_INFO(TDMT_VND_QUERY_COMPACT_PROGRESS), int32_t len = snprintf(detail, sizeof(detail), "msgType:%s numOfEps:%d inUse:%d",
epSet.numOfEps, epSet.inUse); TMSG_INFO(TDMT_VND_QUERY_COMPACT_PROGRESS), epSet.numOfEps, epSet.inUse);
for (int32_t i = 0; i < epSet.numOfEps; ++i) { for (int32_t i = 0; i < epSet.numOfEps; ++i) {
len += snprintf(detail + len, sizeof(detail) - len, " ep:%d-%s:%u", i, epSet.eps[i].fqdn, len += snprintf(detail + len, sizeof(detail) - len, " ep:%d-%s:%u", i, epSet.eps[i].fqdn, epSet.eps[i].port);
epSet.eps[i].port);
} }
mDebug("compact:%d, send update progress msg to %s", pDetail->compactId, detail); mDebug("compact:%d, send update progress msg to %s", pDetail->compactId, detail);
@ -592,40 +587,51 @@ void mndCompactSendProgressReq(SMnode *pMnode, SCompactObj *pCompact){
} }
static int32_t mndSaveCompactProgress(SMnode *pMnode, int32_t compactId) { static int32_t mndSaveCompactProgress(SMnode *pMnode, int32_t compactId) {
bool needSave = false; bool needSave = false;
void* pIter = NULL; void *pIter = NULL;
while (1) { while (1) {
SCompactDetailObj *pDetail = NULL; SCompactDetailObj *pDetail = NULL;
pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT_DETAIL, pIter, (void **)&pDetail); pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT_DETAIL, pIter, (void **)&pDetail);
if (pIter == NULL) break; if (pIter == NULL) break;
if (pDetail->compactId == compactId) { if (pDetail->compactId == compactId) {
mDebug("compact:%d, check save progress, vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d, " mDebug(
"newNumberFileset:%d, newFinished:%d", "compact:%d, check save progress, vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d, "
pDetail->compactId, pDetail->vgId, pDetail->dnodeId, pDetail->numberFileset, pDetail->finished, "newNumberFileset:%d, newFinished:%d",
pDetail->newNumberFileset, pDetail->newFinished); pDetail->compactId, pDetail->vgId, pDetail->dnodeId, pDetail->numberFileset, pDetail->finished,
pDetail->newNumberFileset, pDetail->newFinished);
//these 2 number will jump back after dnode restart, so < is not used here // these 2 number will jump back after dnode restart, so < is not used here
if(pDetail->numberFileset != pDetail->newNumberFileset || pDetail->finished != pDetail->newFinished) if (pDetail->numberFileset != pDetail->newNumberFileset || pDetail->finished != pDetail->newFinished)
needSave = true; needSave = true;
} }
sdbRelease(pMnode->pSdb, pDetail); sdbRelease(pMnode->pSdb, pDetail);
} }
if(!needSave) { SCompactObj *pCompact = mndAcquireCompact(pMnode, compactId);
mDebug("compact:%" PRId32 ", no need to save" , compactId); if (pCompact == NULL) return 0;
SDbObj *pDb = mndAcquireDb(pMnode, pCompact->dbname);
if (pDb == NULL) {
needSave = true;
mWarn("compact:%" PRId32 ", no db exist, set needSave:%s", compactId, pCompact->dbname);
} else {
mndReleaseDb(pMnode, pDb);
pDb = NULL;
}
if (!needSave) {
mDebug("compact:%" PRId32 ", no need to save", compactId);
return 0; return 0;
} }
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB, NULL, "update-compact-progress"); STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB, NULL, "update-compact-progress");
if (pTrans == NULL) { if (pTrans == NULL) {
mError("trans:%" PRId32 ", failed to create since %s" , pTrans->id, terrstr()); mError("trans:%" PRId32 ", failed to create since %s", pTrans->id, terrstr());
return -1; return -1;
} }
mInfo("compact:%d, trans:%d, used to update compact progress.", compactId, pTrans->id); mInfo("compact:%d, trans:%d, used to update compact progress.", compactId, pTrans->id);
SCompactObj *pCompact = mndAcquireCompact(pMnode, compactId);
mndTransSetDbName(pTrans, pCompact->dbname, NULL); mndTransSetDbName(pTrans, pCompact->dbname, NULL);
@ -636,14 +642,15 @@ static int32_t mndSaveCompactProgress(SMnode *pMnode, int32_t compactId) {
if (pIter == NULL) break; if (pIter == NULL) break;
if (pDetail->compactId == compactId) { if (pDetail->compactId == compactId) {
mInfo("compact:%d, trans:%d, check compact progress, vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d, " mInfo(
"newNumberFileset:%d, newFinished:%d", "compact:%d, trans:%d, check compact progress, vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d, "
pDetail->compactId, pTrans->id, pDetail->vgId, pDetail->dnodeId, pDetail->numberFileset, pDetail->finished, "newNumberFileset:%d, newFinished:%d",
pDetail->newNumberFileset, pDetail->newFinished); pDetail->compactId, pTrans->id, pDetail->vgId, pDetail->dnodeId, pDetail->numberFileset, pDetail->finished,
pDetail->newNumberFileset, pDetail->newFinished);
pDetail->numberFileset = pDetail->newNumberFileset; pDetail->numberFileset = pDetail->newNumberFileset;
pDetail->finished = pDetail->newFinished; pDetail->finished = pDetail->newFinished;
SSdbRaw *pCommitRaw = mndCompactDetailActionEncode(pDetail); SSdbRaw *pCommitRaw = mndCompactDetailActionEncode(pDetail);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) { if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
mError("compact:%d, trans:%d, failed to append commit log since %s", pDetail->compactId, pTrans->id, terrstr()); mError("compact:%d, trans:%d, failed to append commit log since %s", pDetail->compactId, pTrans->id, terrstr());
@ -657,22 +664,22 @@ static int32_t mndSaveCompactProgress(SMnode *pMnode, int32_t compactId) {
} }
bool allFinished = true; bool allFinished = true;
pIter = NULL;
while (1) { while (1) {
SCompactDetailObj *pDetail = NULL; SCompactDetailObj *pDetail = NULL;
pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT_DETAIL, pIter, (void **)&pDetail); pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT_DETAIL, pIter, (void **)&pDetail);
if (pIter == NULL) break; if (pIter == NULL) break;
if(pDetail->compactId == compactId){ if (pDetail->compactId == compactId) {
mInfo("compact:%d, trans:%d, check compact finished, vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d", mInfo("compact:%d, trans:%d, check compact finished, vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d",
pDetail->compactId, pTrans->id, pDetail->vgId, pDetail->dnodeId, pDetail->numberFileset, pDetail->finished); pDetail->compactId, pTrans->id, pDetail->vgId, pDetail->dnodeId, pDetail->numberFileset, pDetail->finished);
if(pDetail->numberFileset == -1 && pDetail->finished == -1){ if (pDetail->numberFileset == -1 && pDetail->finished == -1) {
allFinished = false; allFinished = false;
sdbRelease(pMnode->pSdb, pDetail); sdbRelease(pMnode->pSdb, pDetail);
break; break;
} }
if (pDetail->numberFileset != -1 && pDetail->finished != -1 && if (pDetail->numberFileset != -1 && pDetail->finished != -1 && pDetail->numberFileset != pDetail->finished) {
pDetail->numberFileset != pDetail->finished) {
allFinished = false; allFinished = false;
sdbRelease(pMnode->pSdb, pDetail); sdbRelease(pMnode->pSdb, pDetail);
break; break;
@ -682,21 +689,33 @@ static int32_t mndSaveCompactProgress(SMnode *pMnode, int32_t compactId) {
sdbRelease(pMnode->pSdb, pDetail); sdbRelease(pMnode->pSdb, pDetail);
} }
if(allFinished){ pDb = mndAcquireDb(pMnode, pCompact->dbname);
if (pDb == NULL) {
allFinished = true;
mWarn("compact:%" PRId32 ", no db exist, set all finished:%s", compactId, pCompact->dbname);
} else {
mndReleaseDb(pMnode, pDb);
pDb = NULL;
}
if (allFinished) {
mInfo("compact:%d, all finished", pCompact->compactId); mInfo("compact:%d, all finished", pCompact->compactId);
pIter = NULL;
while (1) { while (1) {
SCompactDetailObj *pDetail = NULL; SCompactDetailObj *pDetail = NULL;
pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT_DETAIL, pIter, (void **)&pDetail); pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT_DETAIL, pIter, (void **)&pDetail);
if (pIter == NULL) break; if (pIter == NULL) break;
if (pDetail->compactId == pCompact->compactId) { if (pDetail->compactId == pCompact->compactId) {
SSdbRaw *pCommitRaw = mndCompactDetailActionEncode(pDetail); SSdbRaw *pCommitRaw = mndCompactDetailActionEncode(pDetail);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) { if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
mError("compact:%d, trans:%d, failed to append commit log since %s", pDetail->compactId, pTrans->id, terrstr()); mError("compact:%d, trans:%d, failed to append commit log since %s", pDetail->compactId, pTrans->id,
terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
} }
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED); (void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
mInfo("compact:%d, add drop compactdetail action", pDetail->compactDetailId);
} }
sdbRelease(pMnode->pSdb, pDetail); sdbRelease(pMnode->pSdb, pDetail);
@ -709,6 +728,7 @@ static int32_t mndSaveCompactProgress(SMnode *pMnode, int32_t compactId) {
return -1; return -1;
} }
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED); (void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
mInfo("compact:%d, add drop compact action", pCompact->compactId);
} }
if (mndTransPrepare(pMnode, pTrans) != 0) { if (mndTransPrepare(pMnode, pTrans) != 0) {
@ -739,8 +759,8 @@ void mndCompactPullup(SMnode *pMnode) {
for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) { for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
mInfo("begin to pull up"); mInfo("begin to pull up");
int32_t *pCompactId = taosArrayGet(pArray, i); int32_t *pCompactId = taosArrayGet(pArray, i);
SCompactObj *pCompact = mndAcquireCompact(pMnode, *pCompactId); SCompactObj *pCompact = mndAcquireCompact(pMnode, *pCompactId);
if (pCompact != NULL) { if (pCompact != NULL) {
mInfo("compact:%d, begin to pull up", pCompact->compactId); mInfo("compact:%d, begin to pull up", pCompact->compactId);
mndCompactSendProgressReq(pMnode, pCompact); mndCompactSendProgressReq(pMnode, pCompact);