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 2023-05-30 18:06:28 +08:00
parent e0889b404d
commit 03f7c6533e
7 changed files with 245 additions and 219 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/dnode/vnode/src/tsdb/dev/inc/tsdbCommit.h
View File

@ -13,7 +13,9 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "tsdbDataFileRW.h"
#include "tsdbFS.h"
#include "tsdbIter.h"
#include "tsdbSttFileRW.h"
#ifndef _TSDB_COMMIT_H_

2
source/dnode/vnode/src/tsdb/dev/inc/tsdbFile.h
View File

@ -71,7 +71,7 @@ struct STFile {
struct STFileObj {
TdThreadMutex mutex;
STFile f;
STFile f[1];
int32_t state;
int32_t ref;
char fname[TSDB_FILENAME_LEN];

10
source/dnode/vnode/src/tsdb/dev/inc/tsdbSttFileRW.h
View File

@ -60,11 +60,11 @@ struct SSttFileReaderConfig {
typedef struct SSttFileWriter SSttFileWriter;
typedef struct SSttFileWriterConfig SSttFileWriterConfig;
int32_t tsdbSttFWriterOpen(const SSttFileWriterConfig *config, SSttFileWriter **writer);
int32_t tsdbSttFWriterClose(SSttFileWriter **writer, int8_t abort, STFileOp *op);
int32_t tsdbSttFWriteTSData(SSttFileWriter *writer, SRowInfo *row);
int32_t tsdbSttFWriteTSDataBlock(SSttFileWriter *writer, SBlockData *pBlockData);
int32_t tsdbSttFWriteDLData(SSttFileWriter *writer, TABLEID *tbid, SDelData *pDelData);
int32_t tsdbSttFileWriterOpen(const SSttFileWriterConfig *config, SSttFileWriter **writer);
int32_t tsdbSttFileWriterClose(SSttFileWriter **writer, int8_t abort, STFileOp *op);
int32_t tsdbSttFileWriteTSData(SSttFileWriter *writer, SRowInfo *row);
int32_t tsdbSttFileWriteTSDataBlock(SSttFileWriter *writer, SBlockData *pBlockData);
int32_t tsdbSttFileWriteDLData(SSttFileWriter *writer, TABLEID *tbid, SDelData *pDelData);
struct SSttFileWriterConfig {
STsdb *tsdb;

366
source/dnode/vnode/src/tsdb/dev/tsdbCommit.c
View File

@ -17,63 +17,67 @@
// extern dependencies
typedef struct {
STsdb *pTsdb;
// config
STsdb *tsdb;
int32_t minutes;
int8_t precision;
int32_t minRow;
int32_t maxRow;
int8_t cmprAlg;
int8_t sttTrigger;
int64_t compactVersion;
SArray *aTbDataP; // SArray<STbData *>
TFileOpArray opArray;
int64_t eid; // edit id
struct {
int64_t now;
TSKEY nextKey;
int32_t fid;
int32_t expLevel;
TSKEY minKey;
TSKEY maxKey;
STFileSet *fset;
TABLEID tbid[1];
} ctx[1];
// context
TSKEY nextKey;
int32_t fid;
int32_t expLevel;
TSKEY minKey;
TSKEY maxKey;
STFileSet *fset;
int64_t eid; // edit id
TFileOpArray fopArray[1];
TTsdbIterArray iterArray[1];
SIterMerger *iterMerger;
// writer
SSttFileWriter *pWriter;
} SCommitter;
SDataFileWriter *dataWriter;
SSttFileWriter *sttWriter;
} SCommitter2;
static int32_t open_writer_with_new_stt(SCommitter *pCommitter) {
static int32_t tsdbCommitOpenNewSttWriter(SCommitter2 *pCommitter) {
int32_t code = 0;
int32_t lino = 0;
STsdb *pTsdb = pCommitter->pTsdb;
STsdb *pTsdb = pCommitter->tsdb;
SVnode *pVnode = pTsdb->pVnode;
int32_t vid = TD_VID(pVnode);
SSttFileWriterConfig config;
SDiskID did;
SSttFileWriterConfig config[1];
SDiskID did[1];
if (tfsAllocDisk(pVnode->pTfs, pCommitter->expLevel, &did) < 0) {
if (tfsAllocDisk(pVnode->pTfs, pCommitter->ctx->expLevel, did) < 0) {
code = TSDB_CODE_FS_NO_VALID_DISK;
TSDB_CHECK_CODE(code, lino, _exit);
}
config.tsdb = pTsdb;
config.maxRow = pCommitter->maxRow;
config.szPage = pVnode->config.tsdbPageSize;
config.cmprAlg = pCommitter->cmprAlg;
config.skmTb = NULL;
config.skmRow = NULL;
config.aBuf = NULL;
config.file.type = TSDB_FTYPE_STT;
config.file.did = did;
config.file.fid = pCommitter->fid;
config.file.cid = pCommitter->eid;
config.file.size = 0;
config.file.stt->level = 0;
config.file.stt->nseg = 0;
config->tsdb = pTsdb;
config->maxRow = pCommitter->maxRow;
config->szPage = pVnode->config.tsdbPageSize;
config->cmprAlg = pCommitter->cmprAlg;
config->skmTb = NULL;
config->skmRow = NULL;
config->aBuf = NULL;
config->file.type = TSDB_FTYPE_STT;
config->file.did = did[0];
config->file.fid = pCommitter->ctx->fid;
config->file.cid = pCommitter->eid;
config->file.size = 0;
config->file.stt->level = 0;
config->file.stt->nseg = 0;
code = tsdbSttFWriterOpen(&config, &pCommitter->pWriter);
code = tsdbSttFileWriterOpen(config, &pCommitter->sttWriter);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
@ -84,10 +88,10 @@ _exit:
}
return code;
}
static int32_t open_writer_with_exist_stt(SCommitter *pCommitter, const STFile *pFile) {
static int32_t tsdbCommitOpenExistSttWriter(SCommitter2 *pCommitter, const STFile *pFile) {
int32_t code = 0;
int32_t lino = 0;
STsdb *pTsdb = pCommitter->pTsdb;
STsdb *pTsdb = pCommitter->tsdb;
SVnode *pVnode = pTsdb->pVnode;
int32_t vid = TD_VID(pVnode);
@ -103,7 +107,7 @@ static int32_t open_writer_with_exist_stt(SCommitter *pCommitter, const STFile *
.file = *pFile //
};
code = tsdbSttFWriterOpen(&config, &pCommitter->pWriter);
code = tsdbSttFileWriterOpen(&config, &pCommitter->sttWriter);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
@ -114,105 +118,115 @@ _exit:
}
return code;
}
static int32_t open_committer_writer(SCommitter *pCommitter) {
if (!pCommitter->fset) {
return open_writer_with_new_stt(pCommitter);
static int32_t tsdbCommitOpenWriter(SCommitter2 *committer) {
if (!committer->ctx->fset) {
return tsdbCommitOpenNewSttWriter(committer);
}
const SSttLvl *lvl0 = tsdbTFileSetGetLvl(pCommitter->fset, 0);
const SSttLvl *lvl0 = tsdbTFileSetGetLvl(committer->ctx->fset, 0);
if (lvl0 == NULL) {
return open_writer_with_new_stt(pCommitter);
return tsdbCommitOpenNewSttWriter(committer);
}
ASSERT(TARRAY2_SIZE(&lvl0->farr) > 0);
STFileObj *fobj = TARRAY2_LAST(&lvl0->farr);
if (fobj->f.stt->nseg >= pCommitter->sttTrigger) {
return open_writer_with_new_stt(pCommitter);
if (fobj->f->stt->nseg >= committer->sttTrigger) {
return tsdbCommitOpenNewSttWriter(committer);
} else {
return open_writer_with_exist_stt(pCommitter, &fobj->f);
return tsdbCommitOpenExistSttWriter(committer, fobj->f);
}
}
static int32_t tsdbCommitWriteTSData(SCommitter *pCommitter, SRowInfo *pRowInfo) {
int32_t code = 0;
int32_t lino = 0;
int32_t vid = TD_VID(pCommitter->pTsdb->pVnode);
if (pCommitter->pWriter == NULL) {
code = open_committer_writer(pCommitter);
TSDB_CHECK_CODE(code, lino, _exit);
}
code = tsdbSttFWriteTSData(pCommitter->pWriter, pRowInfo);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
tsdbError("vgId:%d failed at line %d since %s", vid, lino, tstrerror(code));
}
return 0;
static int32_t tsdbCommitTSRow(SCommitter2 *committer, SRowInfo *row) {
return tsdbSttFileWriteTSData(committer->sttWriter, row);
}
static int32_t tsdbCommitWriteDelData(SCommitter *pCommitter, int64_t suid, int64_t uid, int64_t version, int64_t sKey,
static int32_t tsdbCommitWriteDelData(SCommitter2 *pCommitter, int64_t suid, int64_t uid, int64_t version, int64_t sKey,
int64_t eKey) {
int32_t code = 0;
// TODO
return code;
}
static int32_t commit_timeseries_data(SCommitter *pCommitter) {
int32_t code = 0;
int32_t lino = 0;
int64_t nRow = 0;
STsdb *pTsdb = pCommitter->pTsdb;
int32_t vid = TD_VID(pTsdb->pVnode);
SMemTable *pMem = pTsdb->imem;
static int32_t tsdbCommitTSData(SCommitter2 *committer) {
int32_t code = 0;
int32_t lino = 0;
int64_t nRow = 0;
int32_t vid = TD_VID(committer->tsdb->pVnode);
SRowInfo *row;
if (pMem->nRow == 0) goto _exit;
if (committer->tsdb->imem->nRow == 0) goto _exit;
TSDBKEY from = {.ts = pCommitter->minKey, .version = VERSION_MIN};
for (int32_t iTbData = 0; iTbData < taosArrayGetSize(pCommitter->aTbDataP); iTbData++) {
STbDataIter iter;
STbData *pTbData = (STbData *)taosArrayGetP(pCommitter->aTbDataP, iTbData);
SRowInfo rowInfo = {.suid = pTbData->suid, .uid = pTbData->uid};
// open iter and iter merger
STsdbIter *iter;
STsdbIterConfig config[1] = {{
.type = TSDB_ITER_TYPE_MEMT,
.memt = committer->tsdb->imem,
.from = {{
.ts = committer->ctx->minKey,
.version = VERSION_MIN,
}},
}};
tsdbTbDataIterOpen(pTbData, &from, 0, &iter);
code = tsdbIterOpen(config, &iter);
TSDB_CHECK_CODE(code, lino, _exit);
for (TSDBROW *pRow; (pRow = tsdbTbDataIterGet(&iter)) != NULL; tsdbTbDataIterNext(&iter)) {
TSDBKEY rowKey = TSDBROW_KEY(pRow);
code = TARRAY2_APPEND(committer->iterArray, iter);
TSDB_CHECK_CODE(code, lino, _exit);
if (rowKey.ts > pCommitter->maxKey) {
pCommitter->nextKey = TMIN(pCommitter->nextKey, rowKey.ts);
break;
code = tsdbIterMergerInit(committer->iterArray, &committer->iterMerger);
TSDB_CHECK_CODE(code, lino, _exit);
// loop iter
while ((row = tsdbIterMergerGet(committer->iterMerger)) != NULL) {
if (row->uid != committer->ctx->tbid->uid) {
committer->ctx->tbid->suid = row->suid;
committer->ctx->tbid->uid = row->uid;
// Ignore deleted table
SMetaInfo info[1];
if (metaGetInfo(committer->tsdb->pVnode->pMeta, row->uid, info, NULL) != 0) {
code = tsdbIterMergerSkipTableData(committer->iterMerger, committer->ctx->tbid);
TSDB_CHECK_CODE(code, lino, _exit);
continue;
}
}
rowInfo.row = *pRow;
code = tsdbCommitWriteTSData(pCommitter, &rowInfo);
TSKEY ts = TSDBROW_TS(&row->row);
if (ts > committer->ctx->maxKey) {
committer->ctx->nextKey = TMIN(committer->ctx->nextKey, ts);
code = tsdbIterMergerSkipTableData(committer->iterMerger, committer->ctx->tbid);
TSDB_CHECK_CODE(code, lino, _exit);
} else {
code = tsdbCommitTSRow(committer, row);
TSDB_CHECK_CODE(code, lino, _exit);
nRow++;
code = tsdbIterMergerNext(committer->iterMerger);
TSDB_CHECK_CODE(code, lino, _exit);
}
}
_exit:
if (code) {
tsdbError("vgId:%d %s failed at line %d since %s", vid, __func__, lino, tstrerror(code));
TSDB_ERROR_LOG(vid, lino, code);
} else {
tsdbDebug("vgId:%d %s done, fid:%d nRow:%" PRId64, vid, __func__, pCommitter->fid, nRow);
tsdbDebug("vgId:%d %s done, fid:%d nRow:%" PRId64, vid, __func__, committer->ctx->fid, nRow);
}
return code;
}
static int32_t commit_delete_data(SCommitter *pCommitter) {
static int32_t tsdbCommitDelData(SCommitter2 *pCommitter) {
int32_t code = 0;
int32_t lino;
return 0;
#if 0
ASSERTS(0, "TODO: Not implemented yet");
int64_t nDel = 0;
SMemTable *pMem = pCommitter->pTsdb->imem;
SMemTable *pMem = pCommitter->tsdb->imem;
if (pMem->nDel == 0) { // no del data
goto _exit;
@ -222,9 +236,9 @@ static int32_t commit_delete_data(SCommitter *pCommitter) {
STbData *pTbData = (STbData *)taosArrayGetP(pCommitter->aTbDataP, iTbData);
for (SDelData *pDelData = pTbData->pHead; pDelData; pDelData = pDelData->pNext) {
if (pDelData->eKey < pCommitter->minKey) continue;
if (pDelData->sKey > pCommitter->maxKey) {
pCommitter->nextKey = TMIN(pCommitter->nextKey, pDelData->sKey);
if (pDelData->eKey < pCommitter->ctx->minKey) continue;
if (pDelData->sKey > pCommitter->ctx->maxKey) {
pCommitter->ctx->nextKey = TMIN(pCommitter->ctx->nextKey, pDelData->sKey);
continue;
}
@ -236,44 +250,56 @@ static int32_t commit_delete_data(SCommitter *pCommitter) {
_exit:
if (code) {
tsdbError("vgId:%d failed at line %d since %s", TD_VID(pCommitter->pTsdb->pVnode), lino, tstrerror(code));
tsdbError("vgId:%d failed at line %d since %s", TD_VID(pCommitter->tsdb->pVnode), lino, tstrerror(code));
} else {
tsdbDebug("vgId:%d %s done, fid:%d nDel:%" PRId64, TD_VID(pCommitter->pTsdb->pVnode), __func__, pCommitter->fid,
tsdbDebug("vgId:%d %s done, fid:%d nDel:%" PRId64, TD_VID(pCommitter->tsdb->pVnode), __func__, pCommitter->ctx->fid,
pMem->nDel);
}
return code;
#endif
}
static int32_t commit_fset_start(SCommitter *pCommitter) {
STsdb *pTsdb = pCommitter->pTsdb;
int32_t vid = TD_VID(pTsdb->pVnode);
static int32_t tsdbCommitFileSetBegin(SCommitter2 *committer) {
int32_t code = 0;
int32_t lino = 0;
STsdb *tsdb = committer->tsdb;
int32_t vid = TD_VID(tsdb->pVnode);
pCommitter->fid = tsdbKeyFid(pCommitter->nextKey, pCommitter->minutes, pCommitter->precision);
tsdbFidKeyRange(pCommitter->fid, pCommitter->minutes, pCommitter->precision, &pCommitter->minKey,
&pCommitter->maxKey);
pCommitter->expLevel = tsdbFidLevel(pCommitter->fid, &pTsdb->keepCfg, taosGetTimestampSec());
pCommitter->nextKey = TSKEY_MAX;
committer->ctx->fid = tsdbKeyFid(committer->ctx->nextKey, committer->minutes, committer->precision);
tsdbFidKeyRange(committer->ctx->fid, committer->minutes, committer->precision, &committer->ctx->minKey,
&committer->ctx->maxKey);
committer->ctx->expLevel = tsdbFidLevel(committer->ctx->fid, &tsdb->keepCfg, committer->ctx->now);
committer->ctx->nextKey = TSKEY_MAX;
tsdbFSGetFSet(pTsdb->pFS, pCommitter->fid, &pCommitter->fset);
// TODO: use a thread safe function to get fset
tsdbFSGetFSet(tsdb->pFS, committer->ctx->fid, &committer->ctx->fset);
tsdbDebug("vgId:%d %s done, fid:%d minKey:%" PRId64 " maxKey:%" PRId64 " expLevel:%d", vid, __func__, pCommitter->fid,
pCommitter->minKey, pCommitter->maxKey, pCommitter->expLevel);
code = tsdbCommitOpenWriter(committer);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
TSDB_ERROR_LOG(vid, lino, code);
} else {
tsdbDebug("vgId:%d %s done, fid:%d minKey:%" PRId64 " maxKey:%" PRId64 " expLevel:%d", vid, __func__,
committer->ctx->fid, committer->ctx->minKey, committer->ctx->maxKey, committer->ctx->expLevel);
}
return 0;
}
static int32_t commit_fset_end(SCommitter *pCommitter) {
static int32_t tsdbCommitFileSetEnd(SCommitter2 *pCommitter) {
int32_t code = 0;
int32_t lino = 0;
int32_t vid = TD_VID(pCommitter->pTsdb->pVnode);
int32_t vid = TD_VID(pCommitter->tsdb->pVnode);
if (pCommitter->pWriter == NULL) return 0;
if (pCommitter->sttWriter == NULL) return 0;
STFileOp op;
code = tsdbSttFWriterClose(&pCommitter->pWriter, 0, &op);
code = tsdbSttFileWriterClose(&pCommitter->sttWriter, 0, &op);
TSDB_CHECK_CODE(code, lino, _exit);
if (op.optype != TSDB_FOP_NONE) {
code = TARRAY2_APPEND(&pCommitter->opArray, op);
code = TARRAY2_APPEND(pCommitter->fopArray, op);
TSDB_CHECK_CODE(code, lino, _exit);
}
@ -281,90 +307,86 @@ _exit:
if (code) {
tsdbError("vgId:%d failed at line %d since %s", vid, lino, tstrerror(code));
} else {
tsdbDebug("vgId:%d %s done, fid:%d", vid, __func__, pCommitter->fid);
tsdbDebug("vgId:%d %s done, fid:%d", vid, __func__, pCommitter->ctx->fid);
}
return code;
}
static int32_t commit_fset(SCommitter *pCommitter) {
static int32_t tsdbCommitFileSet(SCommitter2 *committer) {
int32_t code = 0;
int32_t lino = 0;
int32_t vid = TD_VID(pCommitter->pTsdb->pVnode);
int32_t vid = TD_VID(committer->tsdb->pVnode);
// fset commit start
code = commit_fset_start(pCommitter);
code = tsdbCommitFileSetBegin(committer);
TSDB_CHECK_CODE(code, lino, _exit);
// commit fset
code = commit_timeseries_data(pCommitter);
code = tsdbCommitTSData(committer);
TSDB_CHECK_CODE(code, lino, _exit);
code = commit_delete_data(pCommitter);
code = tsdbCommitDelData(committer);
TSDB_CHECK_CODE(code, lino, _exit);
// fset commit end
code = commit_fset_end(pCommitter);
code = tsdbCommitFileSetEnd(committer);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
tsdbError("vgId:%d %s failed at line %d since %s", vid, __func__, lino, tstrerror(code));
TSDB_ERROR_LOG(vid, lino, code);
} else {
tsdbDebug("vgId:%d %s done", vid, __func__);
tsdbDebug("vgId:%d %s done, fid:%d", vid, __func__, committer->ctx->fid);
}
return code;
}
static int32_t open_committer(STsdb *pTsdb, SCommitInfo *pInfo, SCommitter *pCommitter) {
static int32_t tsdbOpenCommitter(STsdb *tsdb, SCommitInfo *info, SCommitter2 *committer) {
int32_t code = 0;
int32_t lino;
int32_t lino = 0;
int32_t vid = TD_VID(tsdb->pVnode);
// set config
memset(pCommitter, 0, sizeof(SCommitter));
pCommitter->pTsdb = pTsdb;
pCommitter->minutes = pTsdb->keepCfg.days;
pCommitter->precision = pTsdb->keepCfg.precision;
pCommitter->minRow = pInfo->info.config.tsdbCfg.minRows;
pCommitter->maxRow = pInfo->info.config.tsdbCfg.maxRows;
pCommitter->cmprAlg = pInfo->info.config.tsdbCfg.compression;
pCommitter->sttTrigger = pInfo->info.config.sttTrigger;
memset(committer, 0, sizeof(committer[0]));
pCommitter->aTbDataP = tsdbMemTableGetTbDataArray(pTsdb->imem);
if (pCommitter->aTbDataP == NULL) {
taosArrayDestroy(pCommitter->aTbDataP);
TSDB_CHECK_CODE(code = TSDB_CODE_OUT_OF_MEMORY, lino, _exit);
}
TARRAY2_INIT(&pCommitter->opArray);
tsdbFSAllocEid(pTsdb->pFS, &pCommitter->eid);
committer->tsdb = tsdb;
committer->minutes = tsdb->keepCfg.days;
committer->precision = tsdb->keepCfg.precision;
committer->minRow = info->info.config.tsdbCfg.minRows;
committer->maxRow = info->info.config.tsdbCfg.maxRows;
committer->cmprAlg = info->info.config.tsdbCfg.compression;
committer->sttTrigger = info->info.config.sttTrigger;
committer->compactVersion = INT64_MAX; // TODO: use a function
// start loop
pCommitter->nextKey = pTsdb->imem->minKey; // TODO
TARRAY2_INIT(committer->fopArray);
tsdbFSAllocEid(tsdb->pFS, &committer->eid);
committer->ctx->now = taosGetTimestampSec();
committer->ctx->nextKey = tsdb->imem->minKey; // TODO
_exit:
if (code) {
tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
TSDB_ERROR_LOG(vid, lino, code);
} else {
tsdbDebug("vgId:%d %s done", TD_VID(pTsdb->pVnode), __func__);
tsdbDebug("vgId:%d %s done", TD_VID(tsdb->pVnode), __func__);
}
return code;
}
static int32_t close_committer(SCommitter *pCommiter, int32_t eno) {
static int32_t tsdbCloseCommitter(SCommitter2 *pCommiter, int32_t eno) {
int32_t code = 0;
int32_t lino = 0;
int32_t vid = TD_VID(pCommiter->pTsdb->pVnode);
int32_t vid = TD_VID(pCommiter->tsdb->pVnode);
if (eno == 0) {
code = tsdbFSEditBegin(pCommiter->pTsdb->pFS, &pCommiter->opArray, TSDB_FEDIT_COMMIT);
code = tsdbFSEditBegin(pCommiter->tsdb->pFS, pCommiter->fopArray, TSDB_FEDIT_COMMIT);
TSDB_CHECK_CODE(code, lino, _exit);
} else {
// TODO
ASSERT(0);
}
ASSERT(pCommiter->pWriter == NULL);
taosArrayDestroy(pCommiter->aTbDataP);
TARRAY2_CLEAR_FREE(&pCommiter->opArray, NULL);
ASSERT(pCommiter->sttWriter == NULL);
TARRAY2_FREE(pCommiter->fopArray);
_exit:
if (code) {
@ -376,51 +398,53 @@ _exit:
return code;
}
int32_t tsdbPreCommit(STsdb *pTsdb) {
taosThreadRwlockWrlock(&pTsdb->rwLock);
ASSERT(pTsdb->imem == NULL);
pTsdb->imem = pTsdb->mem;
pTsdb->mem = NULL;
taosThreadRwlockUnlock(&pTsdb->rwLock);
int32_t tsdbPreCommit(STsdb *tsdb) {
taosThreadRwlockWrlock(&tsdb->rwLock);
ASSERT(tsdb->imem == NULL);
tsdb->imem = tsdb->mem;
tsdb->mem = NULL;
taosThreadRwlockUnlock(&tsdb->rwLock);
return 0;
}
int32_t tsdbCommitBegin(STsdb *pTsdb, SCommitInfo *pInfo) {
if (!pTsdb) return 0;
int32_t tsdbCommitBegin(STsdb *tsdb, SCommitInfo *info) {
if (!tsdb) return 0;
int32_t code = 0;
int32_t lino = 0;
SMemTable *pMem = pTsdb->imem;
int32_t vid = TD_VID(tsdb->pVnode);
SMemTable *memt = tsdb->imem;
int64_t nRow = memt->nRow;
int64_t nDel = memt->nDel;
if (pMem->nRow == 0 && pMem->nDel == 0) {
taosThreadRwlockWrlock(&pTsdb->rwLock);
pTsdb->imem = NULL;
taosThreadRwlockUnlock(&pTsdb->rwLock);
tsdbUnrefMemTable(pMem, NULL, true);
if (!nRow && !nDel) {
taosThreadRwlockWrlock(&tsdb->rwLock);
tsdb->imem = NULL;
taosThreadRwlockUnlock(&tsdb->rwLock);
tsdbUnrefMemTable(memt, NULL, true);
} else {
SCommitter committer;
SCommitter2 committer[1];
code = open_committer(pTsdb, pInfo, &committer);
code = tsdbOpenCommitter(tsdb, info, committer);
TSDB_CHECK_CODE(code, lino, _exit);
while (committer.nextKey != TSKEY_MAX) {
code = commit_fset(&committer);
while (committer->ctx->nextKey != TSKEY_MAX) {
code = tsdbCommitFileSet(committer);
if (code) {
lino = __LINE__;
break;
}
}
code = close_committer(&committer, code);
code = tsdbCloseCommitter(committer, code);
TSDB_CHECK_CODE(code, lino, _exit);
}
_exit:
if (code) {
tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
TSDB_ERROR_LOG(vid, lino, code);
} else {
tsdbInfo("vgId:%d %s done, nRow:%" PRId64 " nDel:%" PRId64, TD_VID(pTsdb->pVnode), __func__, pMem->nRow,
pMem->nDel);
tsdbInfo("vgId:%d %s done, nRow:%" PRId64 " nDel:%" PRId64, vid, __func__, nRow, nDel);
}
return code;
}

58
source/dnode/vnode/src/tsdb/dev/tsdbFSet.c
View File

@ -37,7 +37,7 @@ static int32_t tsdbSttLvlInitEx(STsdb *pTsdb, const SSttLvl *lvl1, SSttLvl **lvl
const STFileObj *fobj1;
TARRAY2_FOREACH(&lvl1->farr, fobj1) {
STFileObj *fobj;
code = tsdbTFileObjInit(pTsdb, &fobj1->f, &fobj);
code = tsdbTFileObjInit(pTsdb, fobj1->f, &fobj);
if (code) {
tsdbSttLvlClear(lvl);
return code;
@ -66,25 +66,25 @@ static int32_t tsdbSttLvlApplyEdit(STsdb *pTsdb, const SSttLvl *lvl1, SSttLvl *l
STFileObj *fobj2 = i2 < TARRAY2_SIZE(&lvl2->farr) ? TARRAY2_GET(&lvl2->farr, i2) : NULL;
if (fobj1 && fobj2) {
if (fobj1->f.cid < fobj2->f.cid) {
if (fobj1->f->cid < fobj2->f->cid) {
// create a file obj
code = tsdbTFileObjInit(pTsdb, &fobj1->f, &fobj2);
code = tsdbTFileObjInit(pTsdb, fobj1->f, &fobj2);
if (code) return code;
code = TARRAY2_APPEND(&lvl2->farr, fobj2);
if (code) return code;
i1++;
i2++;
} else if (fobj1->f.cid > fobj2->f.cid) {
} else if (fobj1->f->cid > fobj2->f->cid) {
// remove a file obj
TARRAY2_REMOVE(&lvl2->farr, i2, tsdbSttLvlRemoveFObj);
} else {
if (tsdbIsSameTFile(&fobj1->f, &fobj2->f)) {
if (tsdbIsTFileChanged(&fobj1->f, &fobj2->f)) {
fobj2->f = fobj1->f;
if (tsdbIsSameTFile(fobj1->f, fobj2->f)) {
if (tsdbIsTFileChanged(fobj1->f, fobj2->f)) {
fobj2->f[0] = fobj1->f[0];
}
} else {
TARRAY2_REMOVE(&lvl2->farr, i2, tsdbSttLvlRemoveFObj);
code = tsdbTFileObjInit(pTsdb, &fobj1->f, &fobj2);
code = tsdbTFileObjInit(pTsdb, fobj1->f, &fobj2);
if (code) return code;
code = TARRAY2_SORT_INSERT(&lvl2->farr, fobj2, tsdbTFileObjCmpr);
if (code) return code;
@ -94,7 +94,7 @@ static int32_t tsdbSttLvlApplyEdit(STsdb *pTsdb, const SSttLvl *lvl1, SSttLvl *l
}
} else if (fobj1) {
// create a file obj
code = tsdbTFileObjInit(pTsdb, &fobj1->f, &fobj2);
code = tsdbTFileObjInit(pTsdb, fobj1->f, &fobj2);
if (code) return code;
code = TARRAY2_APPEND(&lvl2->farr, fobj2);
if (code) return code;
@ -127,7 +127,7 @@ static int32_t tsdbSttLvlToJson(const SSttLvl *lvl, cJSON *json) {
if (item == NULL) return TSDB_CODE_OUT_OF_MEMORY;
cJSON_AddItemToArray(ajson, item);
int32_t code = tsdbTFileToJson(&fobj->f, item);
int32_t code = tsdbTFileToJson(fobj->f, item);
if (code) return code;
}
@ -186,7 +186,7 @@ int32_t tsdbTFileSetToJson(const STFileSet *fset, cJSON *json) {
for (int32_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
if (fset->farr[ftype] == NULL) continue;
code = tsdbTFileToJson(&fset->farr[ftype]->f, json);
code = tsdbTFileToJson(fset->farr[ftype]->f, json);
if (code) return code;
}
@ -266,10 +266,10 @@ int32_t tsdbTFileSetEdit(STsdb *pTsdb, STFileSet *fset, const STFileOp *op) {
code = tsdbTFileObjInit(pTsdb, &op->nf, &fobj);
if (code) return code;
if (fobj->f.type == TSDB_FTYPE_STT) {
SSttLvl *lvl = tsdbTFileSetGetLvl(fset, fobj->f.stt->level);
if (fobj->f->type == TSDB_FTYPE_STT) {
SSttLvl *lvl = tsdbTFileSetGetLvl(fset, fobj->f->stt->level);
if (!lvl) {
code = tsdbSttLvlInit(fobj->f.stt->level, &lvl);
code = tsdbSttLvlInit(fobj->f->stt->level, &lvl);
if (code) return code;
code = TARRAY2_SORT_INSERT(&fset->lvlArr, lvl, tsdbSttLvlCmprFn);
@ -279,8 +279,8 @@ int32_t tsdbTFileSetEdit(STsdb *pTsdb, STFileSet *fset, const STFileOp *op) {
code = TARRAY2_SORT_INSERT(&lvl->farr, fobj, tsdbTFileObjCmpr);
if (code) return code;
} else {
ASSERT(fset->farr[fobj->f.type] == NULL);
fset->farr[fobj->f.type] = fobj;
ASSERT(fset->farr[fobj->f->type] == NULL);
fset->farr[fobj->f->type] = fobj;
}
} else if (op->optype == TSDB_FOP_REMOVE) {
// delete a file
@ -288,7 +288,7 @@ int32_t tsdbTFileSetEdit(STsdb *pTsdb, STFileSet *fset, const STFileOp *op) {
SSttLvl *lvl = tsdbTFileSetGetLvl(fset, op->of.stt->level);
ASSERT(lvl);
STFileObj tfobj = {.f = {.cid = op->of.cid}};
STFileObj tfobj = {.f[0] = {.cid = op->of.cid}};
STFileObj *tfobjp = &tfobj;
int32_t idx = TARRAY2_SEARCH_IDX(&lvl->farr, &tfobjp, tsdbTFileObjCmpr, TD_EQ);
ASSERT(idx >= 0);
@ -299,7 +299,7 @@ int32_t tsdbTFileSetEdit(STsdb *pTsdb, STFileSet *fset, const STFileOp *op) {
// TARRAY2_REMOVE(&fset->lvlArr, lvl - fset->lvlArr.data, tsdbSttLvlClear);
}
} else {
ASSERT(tsdbIsSameTFile(&op->of, &fset->farr[op->of.type]->f));
ASSERT(tsdbIsSameTFile(&op->of, fset->farr[op->of.type]->f));
tsdbTFileObjUnref(fset->farr[op->of.type]);
fset->farr[op->of.type] = NULL;
}
@ -308,14 +308,14 @@ int32_t tsdbTFileSetEdit(STsdb *pTsdb, STFileSet *fset, const STFileOp *op) {
SSttLvl *lvl = tsdbTFileSetGetLvl(fset, op->of.stt->level);
ASSERT(lvl);
STFileObj tfobj = {.f = {.cid = op->of.cid}}, *tfobjp = &tfobj;
STFileObj tfobj = {.f[0] = {.cid = op->of.cid}}, *tfobjp = &tfobj;
tfobjp = TARRAY2_SEARCH_EX(&lvl->farr, &tfobjp, tsdbTFileObjCmpr, TD_EQ);
ASSERT(tfobjp);
tfobjp->f = op->nf;
tfobjp->f[0] = op->nf;
} else {
fset->farr[op->nf.type]->f = op->nf;
fset->farr[op->nf.type]->f[0] = op->nf;
}
}
@ -334,18 +334,18 @@ int32_t tsdbTFileSetApplyEdit(STsdb *pTsdb, const STFileSet *fset1, STFileSet *f
STFileObj *fobj2 = fset2->farr[ftype];
if (fobj1 && fobj2) {
if (tsdbIsSameTFile(&fobj1->f, &fobj2->f)) {
if (tsdbIsTFileChanged(&fobj1->f, &fobj2->f)) {
fobj2->f = fobj1->f;
if (tsdbIsSameTFile(fobj1->f, fobj2->f)) {
if (tsdbIsTFileChanged(fobj1->f, fobj2->f)) {
fobj2->f[0] = fobj1->f[0];
}
} else {
tsdbTFileObjRemove(fobj2);
code = tsdbTFileObjInit(pTsdb, &fobj1->f, &fset2->farr[ftype]);
code = tsdbTFileObjInit(pTsdb, fobj1->f, &fset2->farr[ftype]);
if (code) return code;
}
} else if (fobj1) {
// create a new file
code = tsdbTFileObjInit(pTsdb, &fobj1->f, &fset2->farr[ftype]);
code = tsdbTFileObjInit(pTsdb, fobj1->f, &fset2->farr[ftype]);
if (code) return code;
} else {
// remove the file
@ -412,7 +412,7 @@ int32_t tsdbTFileSetInitEx(STsdb *pTsdb, const STFileSet *fset1, STFileSet **fse
for (int32_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
if (fset1->farr[ftype] == NULL) continue;
code = tsdbTFileObjInit(pTsdb, &fset1->farr[ftype]->f, &fset[0]->farr[ftype]);
code = tsdbTFileObjInit(pTsdb, fset1->farr[ftype]->f, &fset[0]->farr[ftype]);
if (code) {
tsdbTFileSetClear(fset);
return code;
@ -479,12 +479,12 @@ int64_t tsdbTFileSetMaxCid(const STFileSet *fset) {
int64_t maxCid = 0;
for (tsdb_ftype_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
if (fset->farr[ftype] == NULL) continue;
maxCid = TMAX(maxCid, fset->farr[ftype]->f.cid);
maxCid = TMAX(maxCid, fset->farr[ftype]->f->cid);
}
const SSttLvl *lvl;
const STFileObj *fobj;
TARRAY2_FOREACH(&fset->lvlArr, lvl) {
TARRAY2_FOREACH(&lvl->farr, fobj) { maxCid = TMAX(maxCid, fobj->f.cid); }
TARRAY2_FOREACH(&lvl->farr, fobj) { maxCid = TMAX(maxCid, fobj->f->cid); }
}
return maxCid;
}

6
source/dnode/vnode/src/tsdb/dev/tsdbFile.c
View File

@ -205,7 +205,7 @@ int32_t tsdbTFileObjInit(STsdb *pTsdb, const STFile *f, STFileObj **fobj) {
if (!fobj[0]) return TSDB_CODE_OUT_OF_MEMORY;
taosThreadMutexInit(&fobj[0]->mutex, NULL);
fobj[0]->f = *f;
fobj[0]->f[0] = f[0];
fobj[0]->state = TSDB_FSTATE_LIVE;
fobj[0]->ref = 1;
tsdbTFileName(pTsdb, f, fobj[0]->fname);
@ -295,9 +295,9 @@ bool tsdbIsTFileChanged(const STFile *f1, const STFile *f2) {
}
int32_t tsdbTFileObjCmpr(const STFileObj **fobj1, const STFileObj **fobj2) {
if (fobj1[0]->f.cid < fobj2[0]->f.cid) {
if (fobj1[0]->f->cid < fobj2[0]->f->cid) {
return -1;
} else if (fobj1[0]->f.cid > fobj2[0]->f.cid) {
} else if (fobj1[0]->f->cid > fobj2[0]->f->cid) {
return 1;
} else {
return 0;

20
source/dnode/vnode/src/tsdb/dev/tsdbMerge.c
View File

@ -91,7 +91,7 @@ static int32_t tsdbMergeToDataWriteTSDataBlock(SMerger *merger) {
// code = tsdbDataFWriteTSDataBlock(merger->dataWriter, &merger->ctx->bData);
// TSDB_CHECK_CODE(code, lino, _exit);
} else {
code = tsdbSttFWriteTSDataBlock(merger->sttWriter, &merger->ctx->bData);
code = tsdbSttFileWriteTSDataBlock(merger->sttWriter, &merger->ctx->bData);
TSDB_CHECK_CODE(code, lino, _exit);
}
@ -157,7 +157,7 @@ static int32_t tsdbMergeToUpperLevel(SMerger *merger) {
if (!merger->ctx->row) break;
code = tsdbSttFWriteTSData(merger->sttWriter, merger->ctx->row);
code = tsdbSttFileWriteTSData(merger->sttWriter, merger->ctx->row);
TSDB_CHECK_CODE(code, lino, _exit);
}
@ -187,7 +187,7 @@ static int32_t tsdbMergeFileSetBegin(SMerger *merger) {
}
fobj = TARRAY2_GET(&lvl->farr, 0);
if (fobj->f.stt->nseg < merger->tsdb->pVnode->config.sttTrigger) {
if (fobj->f->stt->nseg < merger->tsdb->pVnode->config.sttTrigger) {
merger->ctx->toData = false;
break;
} else {
@ -208,9 +208,9 @@ static int32_t tsdbMergeFileSetBegin(SMerger *merger) {
// add the operation
STFileOp op = {
.fid = fobj->f.fid,
.fid = fobj->f->fid,
.optype = TSDB_FOP_REMOVE,
.of = fobj->f,
.of = fobj->f[0],
};
code = TARRAY2_APPEND(&merger->fopArr, op);
TSDB_CHECK_CODE(code, lino, _exit);
@ -227,9 +227,9 @@ static int32_t tsdbMergeFileSetBegin(SMerger *merger) {
.skmTb = &merger->skmTb,
.skmRow = &merger->skmRow,
.aBuf = merger->aBuf,
.file = fobj->f,
.file = fobj->f[0],
};
code = tsdbSttFWriterOpen(&config, &merger->sttWriter);
code = tsdbSttFileWriterOpen(&config, &merger->sttWriter);
TSDB_CHECK_CODE(code, lino, _exit);
} else {
SSttFileWriterConfig config = {
@ -250,7 +250,7 @@ static int32_t tsdbMergeFileSetBegin(SMerger *merger) {
.stt = {{.level = merger->ctx->level, .nseg = 0}},
},
};
code = tsdbSttFWriterOpen(&config, &merger->sttWriter);
code = tsdbSttFileWriterOpen(&config, &merger->sttWriter);
TSDB_CHECK_CODE(code, lino, _exit);
}
@ -276,7 +276,7 @@ static int32_t tsdbMergeFileSetEnd(SMerger *merger) {
int32_t vid = TD_VID(merger->tsdb->pVnode);
STFileOp op;
code = tsdbSttFWriterClose(&merger->sttWriter, 0, &op);
code = tsdbSttFileWriterClose(&merger->sttWriter, 0, &op);
TSDB_CHECK_CODE(code, lino, _exit);
if (op.optype != TSDB_FOP_NONE) {
@ -356,7 +356,7 @@ int32_t tsdbMerge(STsdb *tsdb) {
fobj = TARRAY2_GET(&lvl0->farr, 0);
if (fobj->f.stt->nseg >= sttTrigger) {
if (fobj->f->stt->nseg >= sttTrigger) {
code = tsdbMergeFileSet(merger, fset);
TSDB_CHECK_CODE(code, lino, _exit);
}