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

more code

This commit is contained in:
Hongze Cheng 2025-02-07 18:52:53 +08:00
parent 054ca189bb
commit cfb8cdb991
16 changed files with 156 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/libs/tfs/tfs.h
View File

@ -94,6 +94,8 @@ int32_t tfsGetLevel(STfs *pTfs);
*/
int32_t tfsAllocDisk(STfs *pTfs, int32_t expLevel, SDiskID *pDiskId);
int32_t tfsAllocDiskAtLevel(STfs *pTfs, int32_t level, SDiskID *pDiskId);
/**
* @brief Get the primary path.
*

15
source/dnode/vnode/src/tsdb/tsdbCommit2.c
View File

@ -37,8 +37,6 @@ typedef struct {
struct {
SFileSetCommitInfo *info;
int32_t expLevel;
SDiskID did;
TSKEY minKey;
TSKEY maxKey;
TABLEID tbid[1];
@ -75,7 +73,6 @@ static int32_t tsdbCommitOpenWriter(SCommitter2 *committer) {
.cmprAlg = committer->cmprAlg,
.fid = committer->ctx->info->fid,
.cid = committer->cid,
.did = committer->ctx->did,
.level = 0,
};
@ -324,16 +321,9 @@ static int32_t tsdbCommitFileSetBegin(SCommitter2 *committer) {
// check if can commit
tsdbFSCheckCommit(tsdb, committer->ctx->info->fid);
committer->ctx->expLevel = tsdbFidLevel(committer->ctx->info->fid, &tsdb->keepCfg, committer->now);
tsdbFidKeyRange(committer->ctx->info->fid, committer->minutes, committer->precision, &committer->ctx->minKey,
&committer->ctx->maxKey);
TAOS_CHECK_GOTO(tfsAllocDisk(committer->tsdb->pVnode->pTfs, committer->ctx->expLevel, &committer->ctx->did), &lino,
_exit);
if (tfsMkdirRecurAt(committer->tsdb->pVnode->pTfs, committer->tsdb->path, committer->ctx->did) != 0) {
tsdbError("vgId:%d failed to create directory %s", TD_VID(committer->tsdb->pVnode), committer->tsdb->path);
}
committer->ctx->tbid->suid = 0;
committer->ctx->tbid->uid = 0;
@ -345,9 +335,8 @@ _exit:
if (code) {
tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));
} else {
tsdbDebug("vgId:%d %s done, fid:%d minKey:%" PRId64 " maxKey:%" PRId64 " expLevel:%d", TD_VID(tsdb->pVnode),
__func__, committer->ctx->info->fid, committer->ctx->minKey, committer->ctx->maxKey,
committer->ctx->expLevel);
tsdbDebug("vgId:%d %s done, fid:%d minKey:%" PRId64 " maxKey:%" PRId64, TD_VID(tsdb->pVnode), __func__,
committer->ctx->info->fid, committer->ctx->minKey, committer->ctx->maxKey);
}
return code;
}

21
source/dnode/vnode/src/tsdb/tsdbDataFileRW.c
View File

@ -13,8 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "meta.h"
#include "tsdbDataFileRW.h"
#include "meta.h"
// SDataFileReader =============================================
struct SDataFileReader {
@ -733,10 +733,13 @@ _exit:
return code;
}
extern int32_t tsdbAllocateDisk(STsdb *tsdb, int32_t fid, const char *label, SDiskID *diskId);
static int32_t tsdbDataFileWriterDoOpen(SDataFileWriter *writer) {
int32_t code = 0;
int32_t lino = 0;
int32_t ftype;
SDiskID diskId = {0};
if (!writer->config->skmTb) writer->config->skmTb = writer->skmTb;
if (!writer->config->skmRow) writer->config->skmRow = writer->skmRow;
@ -750,9 +753,11 @@ static int32_t tsdbDataFileWriterDoOpen(SDataFileWriter *writer) {
// .head
ftype = TSDB_FTYPE_HEAD;
code = tsdbAllocateDisk(writer->config->tsdb, writer->config->fid, tsdbFTypeLabel(ftype), &diskId);
TSDB_CHECK_CODE(code, lino, _exit);
writer->files[ftype] = (STFile){
.type = ftype,
.did = writer->config->did,
.did = diskId,
.fid = writer->config->fid,
.cid = writer->config->cid,
.size = 0,
@ -765,9 +770,11 @@ static int32_t tsdbDataFileWriterDoOpen(SDataFileWriter *writer) {
if (writer->config->files[ftype].exist) {
writer->files[ftype] = writer->config->files[ftype].file;
} else {
code = tsdbAllocateDisk(writer->config->tsdb, writer->config->fid, tsdbFTypeLabel(ftype), &diskId);
TSDB_CHECK_CODE(code, lino, _exit);
writer->files[ftype] = (STFile){
.type = ftype,
.did = writer->config->did,
.did = diskId,
.fid = writer->config->fid,
.cid = writer->config->cid,
.size = 0,
@ -782,9 +789,11 @@ static int32_t tsdbDataFileWriterDoOpen(SDataFileWriter *writer) {
if (writer->config->files[ftype].exist) {
writer->files[ftype] = writer->config->files[ftype].file;
} else {
code = tsdbAllocateDisk(writer->config->tsdb, writer->config->fid, tsdbFTypeLabel(ftype), &diskId);
TSDB_CHECK_CODE(code, lino, _exit);
writer->files[ftype] = (STFile){
.type = ftype,
.did = writer->config->did,
.did = diskId,
.fid = writer->config->fid,
.cid = writer->config->cid,
.size = 0,
@ -795,9 +804,11 @@ static int32_t tsdbDataFileWriterDoOpen(SDataFileWriter *writer) {
// .tomb
ftype = TSDB_FTYPE_TOMB;
code = tsdbAllocateDisk(writer->config->tsdb, writer->config->fid, tsdbFTypeLabel(ftype), &diskId);
TSDB_CHECK_CODE(code, lino, _exit);
writer->files[ftype] = (STFile){
.type = ftype,
.did = writer->config->did,
.did = diskId,
.fid = writer->config->fid,
.cid = writer->config->cid,
.size = 0,

1
source/dnode/vnode/src/tsdb/tsdbDataFileRW.h
View File

@ -75,7 +75,6 @@ typedef struct SDataFileWriterConfig {
int32_t szPage;
int32_t fid;
int64_t cid;
SDiskID did;
int64_t compactVersion;
int32_t lcn;
struct {

3
source/dnode/vnode/src/tsdb/tsdbFSetRW.c
View File

@ -155,7 +155,6 @@ int32_t tsdbFSetWriterOpen(SFSetWriterConfig *config, SFSetWriter **writer) {
.szPage = config->szPage,
.fid = config->fid,
.cid = config->cid,
.did = config->did,
.compactVersion = config->compactVersion,
.skmTb = writer[0]->skmTb,
.skmRow = writer[0]->skmRow,
@ -178,14 +177,12 @@ int32_t tsdbFSetWriterOpen(SFSetWriterConfig *config, SFSetWriter **writer) {
.szPage = config->szPage,
.cmprAlg = config->cmprAlg,
.compactVersion = config->compactVersion,
.did = config->did,
.fid = config->fid,
.cid = config->cid,
.level = config->level,
.skmTb = writer[0]->skmTb,
.skmRow = writer[0]->skmRow,
.buffers = writer[0]->buffers,
};
code = tsdbSttFileWriterOpen(&sttWriterConfig, &writer[0]->sttWriter);
TSDB_CHECK_CODE(code, lino, _exit);

1
source/dnode/vnode/src/tsdb/tsdbFSetRW.h
View File

@ -35,7 +35,6 @@ typedef struct {
int8_t cmprAlg;
int32_t fid;
int64_t cid;
SDiskID did;
int32_t level;
int32_t lcn;
struct {

2
source/dnode/vnode/src/tsdb/tsdbFile2.c
View File

@ -455,3 +455,5 @@ int32_t tsdbTFileObjCmpr(const STFileObj **fobj1, const STFileObj **fobj2) {
return 0;
}
}
const char *tsdbFTypeLabel(tsdb_ftype_t ftype) { return g_tfile_info[ftype].suffix; }

2
source/dnode/vnode/src/tsdb/tsdbFile2.h
View File

@ -82,6 +82,8 @@ struct STFileObj {
char fname[TSDB_FILENAME_LEN];
};
const char *tsdbFTypeLabel(tsdb_ftype_t ftype);
#ifdef __cplusplus
}
#endif

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

@ -265,14 +265,6 @@ static int32_t tsdbMergeFileSetBeginOpenWriter(SMerger *merger) {
int32_t lino = 0;
int32_t vid = TD_VID(merger->tsdb->pVnode);
SDiskID did;
int32_t level = tsdbFidLevel(merger->ctx->fset->fid, &merger->tsdb->keepCfg, merger->ctx->now);
TAOS_CHECK_GOTO(tfsAllocDisk(merger->tsdb->pVnode->pTfs, level, &did), &lino, _exit);
code = tfsMkdirRecurAt(merger->tsdb->pVnode->pTfs, merger->tsdb->path, did);
TSDB_CHECK_CODE(code, lino, _exit);
SFSetWriterConfig config = {
.tsdb = merger->tsdb,
.toSttOnly = true,
@ -283,7 +275,6 @@ static int32_t tsdbMergeFileSetBeginOpenWriter(SMerger *merger) {
.cmprAlg = merger->cmprAlg,
.fid = merger->ctx->fset->fid,
.cid = merger->cid,
.did = did,
.level = merger->ctx->level,
};

91
source/dnode/vnode/src/tsdb/tsdbRetention.c
View File

@ -240,59 +240,68 @@ _exit:
return code;
}
extern int32_t tsdbAllocateDiskAtLevel(STsdb *tsdb, int32_t fid, int32_t level, const char *label, SDiskID *diskId);
static int32_t tsdbRemoveOrMoveFileObject(SRTNer *rtner, STFileObj *fobj) {
int32_t code = 0;
int32_t lino = 0;
if (fobj == NULL) {
return code;
}
int32_t expLevel = tsdbFidLevel(fobj->f->fid, &rtner->tsdb->keepCfg, rtner->now);
if (expLevel < 0) {
// remove the file
code = tsdbDoRemoveFileObject(rtner, fobj);
TSDB_CHECK_CODE(code, lino, _exit);
} else if (expLevel > fobj->f->did.level) {
// Try to move the file to a new level
SDiskID diskId = {0};
code = tsdbAllocateDiskAtLevel(rtner->tsdb, fobj->f->fid, expLevel, tsdbFTypeLabel(fobj->f->type), &diskId);
if (code) {
tsdbTrace("vgId:%d, cannot allocate disk for file %s, level:%d, reason:%s, skip!", TD_VID(rtner->tsdb->pVnode),
fobj->fname, expLevel, tstrerror(code));
code = 0;
} else {
tsdbInfo("vgId:%d start to migrate file %s from level %d to %d, size:%" PRId64, TD_VID(rtner->tsdb->pVnode),
fobj->fname, fobj->f->did.level, diskId.level, fobj->f->size);
code = tsdbDoMigrateFileObj(rtner, fobj, &diskId);
TSDB_CHECK_CODE(code, lino, _exit);
tsdbInfo("vgId:%d end to migrate file %s from level %d to %d, size:%" PRId64, TD_VID(rtner->tsdb->pVnode),
fobj->fname, fobj->f->did.level, diskId.level, fobj->f->size);
}
}
_exit:
if (code) {
tsdbError("vgId:%d, %s failed at %s:%d since %s", TD_VID(rtner->tsdb->pVnode), __func__, __FILE__, lino,
tstrerror(code));
}
return code;
}
static int32_t tsdbDoRetention(SRTNer *rtner) {
int32_t code = 0;
int32_t lino = 0;
STFileObj *fobj = NULL;
STFileSet *fset = rtner->fset;
int32_t expLevel = tsdbFidLevel(fset->fid, &rtner->tsdb->keepCfg, rtner->now);
if (expLevel < 0) { // remove the fileset
for (int32_t ftype = 0; (ftype < TSDB_FTYPE_MAX) && (fobj = fset->farr[ftype], 1); ++ftype) {
if (fobj == NULL) continue;
TAOS_CHECK_GOTO(tsdbDoRemoveFileObject(rtner, fobj), &lino, _exit);
}
SSttLvl *lvl;
TARRAY2_FOREACH(fset->lvlArr, lvl) {
TARRAY2_FOREACH(lvl->fobjArr, fobj) { TAOS_CHECK_GOTO(tsdbDoRemoveFileObject(rtner, fobj), &lino, _exit); }
}
} else if (expLevel == 0) { // only migrate to upper level
return 0;
} else { // migrate
SDiskID did;
TAOS_CHECK_GOTO(tfsAllocDisk(rtner->tsdb->pVnode->pTfs, expLevel, &did), &lino, _exit);
code = tfsMkdirRecurAt(rtner->tsdb->pVnode->pTfs, rtner->tsdb->path, did);
// handle data file sets
for (int32_t ftype = 0; ftype < TSDB_FTYPE_MAX; ++ftype) {
code = tsdbRemoveOrMoveFileObject(rtner, fset->farr[ftype]);
TSDB_CHECK_CODE(code, lino, _exit);
// data
for (int32_t ftype = 0; ftype < TSDB_FTYPE_MAX && (fobj = fset->farr[ftype], 1); ++ftype) {
if (fobj == NULL) continue;
if (fobj->f->did.level == did.level) {
continue;
}
if (fobj->f->did.level > did.level) {
continue;
}
tsdbInfo("file:%s size: %" PRId64 " do migrate from %d to %d", fobj->fname, fobj->f->size, fobj->f->did.level,
did.level);
TAOS_CHECK_GOTO(tsdbDoMigrateFileObj(rtner, fobj, &did), &lino, _exit);
}
// stt
// handle stt file
SSttLvl *lvl;
TARRAY2_FOREACH(fset->lvlArr, lvl) {
TARRAY2_FOREACH(lvl->fobjArr, fobj) {
if (fobj->f->did.level == did.level) {
continue;
}
TAOS_CHECK_GOTO(tsdbDoMigrateFileObj(rtner, fobj, &did), &lino, _exit);
}
code = tsdbRemoveOrMoveFileObject(rtner, fobj);
TSDB_CHECK_CODE(code, lino, _exit);
}
}

1
source/dnode/vnode/src/tsdb/tsdbSnapshot.c
View File

@ -790,7 +790,6 @@ static int32_t tsdbSnapWriteFileSetOpenWriter(STsdbSnapWriter* writer) {
.cmprAlg = writer->cmprAlg,
.fid = writer->ctx->fid,
.cid = writer->commitID,
.did = writer->ctx->did,
.level = 0,
};
// merge stt files to either data or a new stt file

10
source/dnode/vnode/src/tsdb/tsdbSttFileRW.c
View File

@ -778,9 +778,12 @@ static int32_t tsdbSttFileDoWriteFooter(SSttFileWriter *writer) {
return tsdbFileWriteSttFooter(writer->fd, writer->footer, &writer->file->size, encryptAlgorithm, encryptKey);
}
extern int32_t tsdbAllocateDisk(STsdb *tsdb, int32_t fid, const char *label, SDiskID *diskId);
static int32_t tsdbSttFWriterDoOpen(SSttFileWriter *writer) {
int32_t code = 0;
int32_t lino = 0;
STsdb *tsdb = writer->config->tsdb;
// set
if (!writer->config->skmTb) writer->config->skmTb = writer->skmTb;
@ -790,9 +793,14 @@ static int32_t tsdbSttFWriterDoOpen(SSttFileWriter *writer) {
writer->buffers = writer->local;
}
// alloc disk id
SDiskID diskId = {0};
code = tsdbAllocateDisk(tsdb, writer->config->fid, "stt", &diskId);
TSDB_CHECK_CODE(code, lino, _exit);
writer->file[0] = (STFile){
.type = TSDB_FTYPE_STT,
.did = writer->config->did,
.did = diskId,
.fid = writer->config->fid,
.cid = writer->config->cid,
.size = 0,

1
source/dnode/vnode/src/tsdb/tsdbSttFileRW.h
View File

@ -82,7 +82,6 @@ struct SSttFileWriterConfig {
int32_t szPage;
int8_t cmprAlg;
int64_t compactVersion;
SDiskID did;
int32_t fid;
int64_t cid;
int32_t level;

48
source/dnode/vnode/src/tsdb/tsdbUtil.c
View File

@ -622,7 +622,7 @@ void tsdbRowGetColVal(TSDBROW *pRow, STSchema *pTSchema, int32_t iCol, SColVal *
SColData *pColData = tBlockDataGetColData(pRow->pBlockData, pTColumn->colId);
if (pColData) {
if (tColDataGetValue(pColData, pRow->iRow, pColVal) != 0){
if (tColDataGetValue(pColData, pRow->iRow, pColVal) != 0) {
tsdbError("failed to tColDataGetValue");
}
} else {
@ -647,7 +647,7 @@ void tColRowGetPrimaryKey(SBlockData *pBlock, int32_t irow, SRowKey *key) {
SColData *pColData = &pBlock->aColData[i];
if (pColData->cflag & COL_IS_KEY) {
SColVal cv;
if (tColDataGetValue(pColData, irow, &cv) != 0){
if (tColDataGetValue(pColData, irow, &cv) != 0) {
break;
}
key->pks[key->numOfPKs] = cv.value;
@ -723,7 +723,8 @@ SColVal *tsdbRowIterNext(STSDBRowIter *pIter) {
}
if (pIter->iColData <= pIter->pRow->pBlockData->nColData) {
if (tColDataGetValue(&pIter->pRow->pBlockData->aColData[pIter->iColData - 1], pIter->pRow->iRow, &pIter->cv) != 0){
if (tColDataGetValue(&pIter->pRow->pBlockData->aColData[pIter->iColData - 1], pIter->pRow->iRow, &pIter->cv) !=
0) {
return NULL;
}
++pIter->iColData;
@ -1802,3 +1803,44 @@ uint32_t tsdbCvtTimestampAlg(uint32_t alg) {
return 0;
}
int32_t tsdbAllocateDisk(STsdb *tsdb, int32_t fid, const char *label, SDiskID *diskId) {
int32_t code = 0;
int32_t lino = 0;
SDiskID did = {0};
STfs *tfs = tsdb->pVnode->pTfs;
int32_t expectedLevel = tsdbFidLevel(fid, &tsdb->keepCfg, taosGetTimestampSec());
code = tfsAllocDisk(tfs, expectedLevel, &did);
if (code) {
tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));
return code;
}
if (tfsMkdirRecurAt(tfs, tsdb->path, did) != 0) {
tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));
}
if (diskId) {
*diskId = did;
}
return code;
}
int32_t tsdbAllocateDiskAtLevel(STsdb *tsdb, int32_t fid, int32_t level, const char *label, SDiskID *diskId) {
int32_t code = 0;
SDiskID did = {
.level = level,
};
STfs *tfs = tsdb->pVnode->pTfs;
code = tfsAllocDiskAtLevel(tfs, level, &did);
if (code) {
return code;
}
if (diskId) {
*diskId = did;
}
return 0;
}

8
source/libs/tfs/inc/tfsInt.h
View File

@ -61,22 +61,22 @@ typedef struct {
STfsDisk *pDisk;
} SDiskIter;
typedef struct STfsDir {
struct STfsDir {
SDiskIter iter;
SDiskID did;
char dirName[TSDB_FILENAME_LEN];
STfsFile tfile;
TdDirPtr pDir;
STfs *pTfs;
} STfsDir;
};
typedef struct STfs {
struct STfs {
TdThreadSpinlock lock;
SDiskSize size;
int32_t nlevel;
STfsTier tiers[TFS_MAX_TIERS];
SHashObj *hash; // name to did map
} STfs;
};
int32_t tfsCheckAndFormatCfg(STfs *pTfs, SDiskCfg *pCfg);
int32_t tfsNewDisk(int32_t level, int32_t id, int8_t disable, const char *dir, STfsDisk **ppDisk);

20
source/libs/tfs/src/tfs.c
View File

@ -152,6 +152,26 @@ int32_t tfsGetDisksAtLevel(STfs *pTfs, int32_t level) {
int32_t tfsGetLevel(STfs *pTfs) { return pTfs->nlevel; }
int32_t tfsAllocDiskAtLevel(STfs *pTfs, int32_t level, SDiskID *pDiskId) {
pDiskId->level = level;
pDiskId->id = -1;
if (pDiskId->level >= pTfs->nlevel) {
pDiskId->level = pTfs->nlevel - 1;
}
if (pDiskId->level < 0) {
pDiskId->level = 0;
}
pDiskId->id = tfsAllocDiskOnTier(&pTfs->tiers[pDiskId->level]);
if (pDiskId->id < 0) {
TAOS_RETURN(TSDB_CODE_FS_NO_VALID_DISK);
}
TAOS_RETURN(0);
}
int32_t tfsAllocDisk(STfs *pTfs, int32_t expLevel, SDiskID *pDiskId) {
pDiskId->level = expLevel;
pDiskId->id = -1;