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

Merge pull request #23143 from taosdata/enh/TD-26578 

enh: optimize merge strategy
This commit is contained in:
wade zhang 2023-10-08 09:59:54 +08:00 committed by GitHub
parent f6b8500dd3 d1e59b9e0e
commit 95818f4acf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 53 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
source/dnode/vnode/src/tsdb/tsdbMerge.c
View File

@ -34,7 +34,6 @@ typedef struct {
STFileSet *fset; STFileSet *fset;
bool toData; bool toData;
int32_t level; int32_t level;
SSttLvl *lvl;
TABLEID tbid[1]; TABLEID tbid[1];
} ctx[1]; } ctx[1];
@ -107,29 +106,63 @@ static int32_t tsdbMergeFileSetBeginOpenReader(SMerger *merger) {
merger->ctx->toData = true; merger->ctx->toData = true;
merger->ctx->level = 0; merger->ctx->level = 0;
// TODO: optimize merge strategy // find the highest level that can be merged to
for (int32_t i = 0;; ++i) { for (int32_t i = 0, numCarry = 0;;) {
if (i >= TARRAY2_SIZE(merger->ctx->fset->lvlArr)) { int32_t numFile = numCarry;
merger->ctx->lvl = NULL; if (i < TARRAY2_SIZE(merger->ctx->fset->lvlArr) &&
break; merger->ctx->level == TARRAY2_GET(merger->ctx->fset->lvlArr, i)->level) {
numFile += TARRAY2_SIZE(TARRAY2_GET(merger->ctx->fset->lvlArr, i)->fobjArr);
i++;
} }
merger->ctx->lvl = TARRAY2_GET(merger->ctx->fset->lvlArr, i); numCarry = numFile / merger->sttTrigger;
if (merger->ctx->lvl->level != merger->ctx->level || if (numCarry == 0) {
TARRAY2_SIZE(merger->ctx->lvl->fobjArr) + 1 < merger->sttTrigger) { break;
} else {
merger->ctx->level++;
}
}
ASSERT(merger->ctx->level > 0);
SSttLvl *lvl;
TARRAY2_FOREACH_REVERSE(merger->ctx->fset->lvlArr, lvl) {
if (TARRAY2_SIZE(lvl->fobjArr) == 0) {
continue;
}
if (lvl->level <= merger->ctx->level) {
merger->ctx->toData = false; merger->ctx->toData = false;
merger->ctx->lvl = NULL; }
break;
}
// get number of level-0 files to merge
int32_t numFile = pow(merger->sttTrigger, merger->ctx->level);
TARRAY2_FOREACH(merger->ctx->fset->lvlArr, lvl) {
if (lvl->level == 0) continue;
if (lvl->level >= merger->ctx->level) break;
numFile = numFile - TARRAY2_SIZE(lvl->fobjArr) * pow(merger->sttTrigger, lvl->level);
}
ASSERT(numFile >= 0);
// get file system operations
TARRAY2_FOREACH(merger->ctx->fset->lvlArr, lvl) {
if (lvl->level >= merger->ctx->level) {
break; break;
} }
merger->ctx->level++; int32_t numMergeFile;
if (lvl->level == 0) {
numMergeFile = numFile;
} else {
numMergeFile = TARRAY2_SIZE(lvl->fobjArr);
}
STFileObj *fobj; for (int32_t i = 0; i < numMergeFile; ++i) {
int32_t numFile = 0; STFileObj *fobj = TARRAY2_GET(lvl->fobjArr, i);
TARRAY2_FOREACH(merger->ctx->lvl->fobjArr, fobj) {
if (numFile == merger->sttTrigger) {
break;
}
STFileOp op = { STFileOp op = {
.optype = TSDB_FOP_REMOVE, .optype = TSDB_FOP_REMOVE,
@ -151,8 +184,6 @@ static int32_t tsdbMergeFileSetBeginOpenReader(SMerger *merger) {
code = TARRAY2_APPEND(merger->sttReaderArr, reader); code = TARRAY2_APPEND(merger->sttReaderArr, reader);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
numFile++;
} }
} }
@ -434,7 +465,9 @@ int32_t tsdbMerge(void *arg) {
.sttTrigger = tsdb->pVnode->config.sttTrigger, .sttTrigger = tsdb->pVnode->config.sttTrigger,
}}; }};
ASSERT(merger->sttTrigger > 1); if (merger->sttTrigger <= 1) {
return 0;
}
code = tsdbFSCreateCopySnapshot(tsdb->pFS, &merger->fsetArr); code = tsdbFSCreateCopySnapshot(tsdb->pFS, &merger->fsetArr);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);