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

enh: control stt max level

This commit is contained in:
Hongze Cheng 2023-10-09 14:00:11 +08:00
parent 880e8c1941
commit 2c9a0dbcda
1 changed files with 127 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -15,6 +15,8 @@
#include "tsdbMerge.h" #include "tsdbMerge.h"
#define TSDB_MAX_LEVEL 6 // means max level is 7
typedef struct { typedef struct {
STsdb *tsdb; STsdb *tsdb;
TFileSetArray *fsetArr; TFileSetArray *fsetArr;
@ -100,90 +102,142 @@ _exit:
} }
static int32_t tsdbMergeFileSetBeginOpenReader(SMerger *merger) { static int32_t tsdbMergeFileSetBeginOpenReader(SMerger *merger) {
int32_t code = 0; int32_t code = 0;
int32_t lino = 0; int32_t lino = 0;
merger->ctx->toData = true;
merger->ctx->level = 0;
// find the highest level that can be merged to
for (int32_t i = 0, numCarry = 0;;) {
int32_t numFile = numCarry;
if (i < TARRAY2_SIZE(merger->ctx->fset->lvlArr) &&
merger->ctx->level == TARRAY2_GET(merger->ctx->fset->lvlArr, i)->level) {
numFile += TARRAY2_SIZE(TARRAY2_GET(merger->ctx->fset->lvlArr, i)->fobjArr);
i++;
}
numCarry = numFile / merger->sttTrigger;
if (numCarry == 0) {
break;
} else {
merger->ctx->level++;
}
}
ASSERT(merger->ctx->level > 0);
SSttLvl *lvl; SSttLvl *lvl;
bool hasLevelLargerThanMax = false;
TARRAY2_FOREACH_REVERSE(merger->ctx->fset->lvlArr, lvl) { TARRAY2_FOREACH_REVERSE(merger->ctx->fset->lvlArr, lvl) {
if (TARRAY2_SIZE(lvl->fobjArr) == 0) { if (lvl->level <= TSDB_MAX_LEVEL) {
continue; break;
} } else if (TARRAY2_SIZE(lvl->fobjArr) > 0) {
hasLevelLargerThanMax = true;
if (lvl->level <= merger->ctx->level) {
merger->ctx->toData = false;
}
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;
} }
}
int32_t numMergeFile; if (hasLevelLargerThanMax) {
if (lvl->level == 0) { // merge all stt files
numMergeFile = numFile; merger->ctx->toData = true;
} else { merger->ctx->level = TSDB_MAX_LEVEL;
numMergeFile = TARRAY2_SIZE(lvl->fobjArr);
TARRAY2_FOREACH(merger->ctx->fset->lvlArr, lvl) {
int32_t numMergeFile = TARRAY2_SIZE(lvl->fobjArr);
for (int32_t i = 0; i < numMergeFile; ++i) {
STFileObj *fobj = TARRAY2_GET(lvl->fobjArr, i);
STFileOp op = {
.optype = TSDB_FOP_REMOVE,
.fid = merger->ctx->fset->fid,
.of = fobj->f[0],
};
code = TARRAY2_APPEND(merger->fopArr, op);
TSDB_CHECK_CODE(code, lino, _exit);
SSttFileReader *reader;
SSttFileReaderConfig config = {
.tsdb = merger->tsdb,
.szPage = merger->szPage,
.file[0] = fobj->f[0],
};
code = tsdbSttFileReaderOpen(fobj->fname, &config, &reader);
TSDB_CHECK_CODE(code, lino, _exit);
code = TARRAY2_APPEND(merger->sttReaderArr, reader);
TSDB_CHECK_CODE(code, lino, _exit);
}
}
} else {
// do regular merge
merger->ctx->toData = true;
merger->ctx->level = 0;
// find the highest level that can be merged to
for (int32_t i = 0, numCarry = 0;;) {
int32_t numFile = numCarry;
if (i < TARRAY2_SIZE(merger->ctx->fset->lvlArr) &&
merger->ctx->level == TARRAY2_GET(merger->ctx->fset->lvlArr, i)->level) {
numFile += TARRAY2_SIZE(TARRAY2_GET(merger->ctx->fset->lvlArr, i)->fobjArr);
i++;
}
numCarry = numFile / merger->sttTrigger;
if (numCarry == 0) {
break;
} else {
merger->ctx->level++;
}
} }
for (int32_t i = 0; i < numMergeFile; ++i) { ASSERT(merger->ctx->level > 0);
STFileObj *fobj = TARRAY2_GET(lvl->fobjArr, i);
STFileOp op = { if (merger->ctx->level <= TSDB_MAX_LEVEL) {
.optype = TSDB_FOP_REMOVE, TARRAY2_FOREACH_REVERSE(merger->ctx->fset->lvlArr, lvl) {
.fid = merger->ctx->fset->fid, if (TARRAY2_SIZE(lvl->fobjArr) == 0) {
.of = fobj->f[0], continue;
}; }
code = TARRAY2_APPEND(merger->fopArr, op);
TSDB_CHECK_CODE(code, lino, _exit);
SSttFileReader *reader; if (lvl->level >= merger->ctx->level) {
SSttFileReaderConfig config = { merger->ctx->toData = false;
.tsdb = merger->tsdb, }
.szPage = merger->szPage, break;
.file[0] = fobj->f[0], }
}; }
code = tsdbSttFileReaderOpen(fobj->fname, &config, &reader); // get number of level-0 files to merge
TSDB_CHECK_CODE(code, lino, _exit); 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;
code = TARRAY2_APPEND(merger->sttReaderArr, reader); numFile = numFile - TARRAY2_SIZE(lvl->fobjArr) * pow(merger->sttTrigger, lvl->level);
TSDB_CHECK_CODE(code, lino, _exit); }
ASSERT(numFile >= 0);
// get file system operations
TARRAY2_FOREACH(merger->ctx->fset->lvlArr, lvl) {
if (lvl->level >= merger->ctx->level) {
break;
}
int32_t numMergeFile;
if (lvl->level == 0) {
numMergeFile = numFile;
} else {
numMergeFile = TARRAY2_SIZE(lvl->fobjArr);
}
for (int32_t i = 0; i < numMergeFile; ++i) {
STFileObj *fobj = TARRAY2_GET(lvl->fobjArr, i);
STFileOp op = {
.optype = TSDB_FOP_REMOVE,
.fid = merger->ctx->fset->fid,
.of = fobj->f[0],
};
code = TARRAY2_APPEND(merger->fopArr, op);
TSDB_CHECK_CODE(code, lino, _exit);
SSttFileReader *reader;
SSttFileReaderConfig config = {
.tsdb = merger->tsdb,
.szPage = merger->szPage,
.file[0] = fobj->f[0],
};
code = tsdbSttFileReaderOpen(fobj->fname, &config, &reader);
TSDB_CHECK_CODE(code, lino, _exit);
code = TARRAY2_APPEND(merger->sttReaderArr, reader);
TSDB_CHECK_CODE(code, lino, _exit);
}
}
if (merger->ctx->level > TSDB_MAX_LEVEL) {
merger->ctx->level = TSDB_MAX_LEVEL;
} }
} }