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 2022-08-26 15:48:44 +08:00
parent c3c3489db6
commit 90b32809c5
1 changed files with 380 additions and 373 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

753
source/dnode/vnode/src/tsdb/tsdbCommit.c
View File

@ -1024,6 +1024,7 @@ _err:
return code; return code;
} }
static int32_t tsdbCommitFileDataImpl(SCommitter *pCommitter);
static int32_t tsdbCommitFileData(SCommitter *pCommitter) { static int32_t tsdbCommitFileData(SCommitter *pCommitter) {
int32_t code = 0; int32_t code = 0;
STsdb *pTsdb = pCommitter->pTsdb; STsdb *pTsdb = pCommitter->pTsdb;
@ -1033,6 +1034,11 @@ static int32_t tsdbCommitFileData(SCommitter *pCommitter) {
code = tsdbCommitFileDataStart(pCommitter); code = tsdbCommitFileDataStart(pCommitter);
if (code) goto _err; if (code) goto _err;
#if 1
// impl
code = tsdbCommitFileDataImpl(pCommitter);
if (code) goto _err;
#else
// commit file data impl // commit file data impl
for (int32_t iTbData = 0; iTbData < taosArrayGetSize(pCommitter->aTbDataP); iTbData++) { for (int32_t iTbData = 0; iTbData < taosArrayGetSize(pCommitter->aTbDataP); iTbData++) {
STbData *pTbData = (STbData *)taosArrayGetP(pCommitter->aTbDataP, iTbData); STbData *pTbData = (STbData *)taosArrayGetP(pCommitter->aTbDataP, iTbData);
@ -1059,6 +1065,7 @@ static int32_t tsdbCommitFileData(SCommitter *pCommitter) {
code = tsdbCommitLastBlock(pCommitter); code = tsdbCommitLastBlock(pCommitter);
if (code) goto _err; if (code) goto _err;
} }
#endif
// commit file data end // commit file data end
code = tsdbCommitFileDataEnd(pCommitter); code = tsdbCommitFileDataEnd(pCommitter);
@ -1309,10 +1316,10 @@ static int32_t tsdbEndCommit(SCommitter *pCommitter, int32_t eno) {
tsdbFSDestroy(&pCommitter->fs); tsdbFSDestroy(&pCommitter->fs);
taosArrayDestroy(pCommitter->aTbDataP); taosArrayDestroy(pCommitter->aTbDataP);
if (pCommitter->toMerge) { // if (pCommitter->toMerge) {
code = tsdbMerge(pTsdb); // code = tsdbMerge(pTsdb);
if (code) goto _err; // if (code) goto _err;
} // }
tsdbInfo("vgId:%d, tsdb end commit", TD_VID(pTsdb->pVnode)); tsdbInfo("vgId:%d, tsdb end commit", TD_VID(pTsdb->pVnode));
return code; return code;
@ -1485,372 +1492,372 @@ _err:
return code; return code;
} }
// ================================================================================ // // ================================================================================
typedef struct { // typedef struct {
SRowInfo rowInfo; // SRowInfo rowInfo;
SDataFReader *pReader; // SDataFReader *pReader;
int32_t iLast; // int32_t iLast;
SArray *aBlockL; // SArray<SBlockL> // SArray *aBlockL; // SArray<SBlockL>
int32_t iBlockL; // int32_t iBlockL;
SBlockData bData; // SBlockData bData;
int32_t iRow; // int32_t iRow;
} SLDataIter; // } SLDataIter;
typedef struct { // typedef struct {
SRBTreeNode *pNode; // SRBTreeNode *pNode;
SRBTree rbt; // SRBTree rbt;
} SDataMerger; // } SDataMerger;
static void tDataMergerInit(SDataMerger *pMerger, SArray *aNodeP) { // static void tDataMergerInit(SDataMerger *pMerger, SArray *aNodeP) {
pMerger->pNode = NULL; // pMerger->pNode = NULL;
tRBTreeCreate(&pMerger->rbt, tRowInfoCmprFn); // tRBTreeCreate(&pMerger->rbt, tRowInfoCmprFn);
for (int32_t iNode = 0; iNode < taosArrayGetSize(aNodeP); iNode++) { // for (int32_t iNode = 0; iNode < taosArrayGetSize(aNodeP); iNode++) {
SRBTreeNode *pNode = (SRBTreeNode *)taosArrayGetP(aNodeP, iNode); // SRBTreeNode *pNode = (SRBTreeNode *)taosArrayGetP(aNodeP, iNode);
pNode = tRBTreePut(&pMerger->rbt, pNode); // pNode = tRBTreePut(&pMerger->rbt, pNode);
ASSERT(pNode); // ASSERT(pNode);
} // }
} // }
static int32_t tDataMergeNext(SDataMerger *pMerger, SRowInfo **ppInfo) { // static int32_t tDataMergeNext(SDataMerger *pMerger, SRowInfo **ppInfo) {
int32_t code = 0; // int32_t code = 0;
if (pMerger->pNode) { // if (pMerger->pNode) {
// next current iter // // next current iter
SLDataIter *pIter = (SLDataIter *)pMerger->pNode->payload; // SLDataIter *pIter = (SLDataIter *)pMerger->pNode->payload;
pIter->iRow++; // pIter->iRow++;
if (pIter->iRow < pIter->bData.nRow) { // if (pIter->iRow < pIter->bData.nRow) {
pIter->rowInfo.uid = pIter->bData.uid ? pIter->bData.uid : pIter->bData.aUid[pIter->iRow]; // pIter->rowInfo.uid = pIter->bData.uid ? pIter->bData.uid : pIter->bData.aUid[pIter->iRow];
pIter->rowInfo.row = tsdbRowFromBlockData(&pIter->bData, pIter->iRow); // pIter->rowInfo.row = tsdbRowFromBlockData(&pIter->bData, pIter->iRow);
} else { // } else {
pIter->iBlockL++; // pIter->iBlockL++;
if (pIter->iBlockL < taosArrayGetSize(pIter->aBlockL)) { // if (pIter->iBlockL < taosArrayGetSize(pIter->aBlockL)) {
SBlockL *pBlockL = (SBlockL *)taosArrayGet(pIter->aBlockL, pIter->iBlockL); // SBlockL *pBlockL = (SBlockL *)taosArrayGet(pIter->aBlockL, pIter->iBlockL);
code = tsdbReadLastBlockEx(pIter->pReader, pIter->iLast, pBlockL, &pIter->bData); // code = tsdbReadLastBlockEx(pIter->pReader, pIter->iLast, pBlockL, &pIter->bData);
if (code) goto _exit; // if (code) goto _exit;
pIter->iRow = 0; // pIter->iRow = 0;
pIter->rowInfo.suid = pIter->bData.suid; // pIter->rowInfo.suid = pIter->bData.suid;
pIter->rowInfo.uid = pIter->bData.uid ? pIter->bData.uid : pIter->bData.aUid[0]; // pIter->rowInfo.uid = pIter->bData.uid ? pIter->bData.uid : pIter->bData.aUid[0];
pIter->rowInfo.row = tsdbRowFromBlockData(&pIter->bData, 0); // pIter->rowInfo.row = tsdbRowFromBlockData(&pIter->bData, 0);
} else { // } else {
pMerger->pNode = NULL; // pMerger->pNode = NULL;
} // }
} // }
SRBTreeNode *pMinNode = tRBTreeMin(&pMerger->rbt); // SRBTreeNode *pMinNode = tRBTreeMin(&pMerger->rbt);
if (pMerger->pNode && pMinNode) { // if (pMerger->pNode && pMinNode) {
int32_t c = tRowInfoCmprFn(pMerger->pNode->payload, pMinNode->payload); // int32_t c = tRowInfoCmprFn(pMerger->pNode->payload, pMinNode->payload);
if (c > 0) { // if (c > 0) {
pMerger->pNode = tRBTreePut(&pMerger->rbt, pMerger->pNode); // pMerger->pNode = tRBTreePut(&pMerger->rbt, pMerger->pNode);
ASSERT(pMerger->pNode); // ASSERT(pMerger->pNode);
pMerger->pNode = NULL; // pMerger->pNode = NULL;
} else { // } else {
ASSERT(c); // ASSERT(c);
} // }
} // }
} // }
if (pMerger->pNode == NULL) { // if (pMerger->pNode == NULL) {
pMerger->pNode = tRBTreeMin(&pMerger->rbt); // pMerger->pNode = tRBTreeMin(&pMerger->rbt);
if (pMerger->pNode) { // if (pMerger->pNode) {
tRBTreeDrop(&pMerger->rbt, pMerger->pNode); // tRBTreeDrop(&pMerger->rbt, pMerger->pNode);
} // }
} // }
if (pMerger->pNode) { // if (pMerger->pNode) {
*ppInfo = &((SLDataIter *)pMerger->pNode->payload)[0].rowInfo; // *ppInfo = &((SLDataIter *)pMerger->pNode->payload)[0].rowInfo;
} else { // } else {
*ppInfo = NULL; // *ppInfo = NULL;
} // }
_exit: // _exit:
return code; // return code;
} // }
typedef struct { // typedef struct {
STsdb *pTsdb; // STsdb *pTsdb;
int8_t maxLast; // int8_t maxLast;
int32_t minRow; // int32_t minRow;
int32_t maxRow; // int32_t maxRow;
int8_t cmprAlg; // int8_t cmprAlg;
int64_t commitID; // int64_t commitID;
STsdbFS fs; // STsdbFS fs;
struct { // struct {
SDataFReader *pReader; // SDataFReader *pReader;
SArray *aBlockIdx; // SArray *aBlockIdx;
SLDataIter *aLDataiter[TSDB_MAX_LAST_FILE]; // SLDataIter *aLDataiter[TSDB_MAX_LAST_FILE];
SDataMerger merger; // SDataMerger merger;
} dReader; // } dReader;
struct { // struct {
SDataFWriter *pWriter; // SDataFWriter *pWriter;
SArray *aBlockIdx; // SArray *aBlockIdx;
SArray *aBlockL; // SArray *aBlockL;
SBlockData bData; // SBlockData bData;
SBlockData bDatal; // SBlockData bDatal;
} dWriter; // } dWriter;
} STsdbMerger; // } STsdbMerger;
static int32_t tsdbMergeFileDataStart(STsdbMerger *pMerger, SDFileSet *pSet) { // static int32_t tsdbMergeFileDataStart(STsdbMerger *pMerger, SDFileSet *pSet) {
int32_t code = 0; // int32_t code = 0;
STsdb *pTsdb = pMerger->pTsdb; // STsdb *pTsdb = pMerger->pTsdb;
// reader // // reader
code = tsdbDataFReaderOpen(&pMerger->dReader.pReader, pTsdb, pSet); // code = tsdbDataFReaderOpen(&pMerger->dReader.pReader, pTsdb, pSet);
if (code) goto _err; // if (code) goto _err;
code = tsdbReadBlockIdx(pMerger->dReader.pReader, pMerger->dReader.aBlockIdx); // code = tsdbReadBlockIdx(pMerger->dReader.pReader, pMerger->dReader.aBlockIdx);
if (code) goto _err; // if (code) goto _err;
pMerger->dReader.merger.pNode = NULL; // pMerger->dReader.merger.pNode = NULL;
tRBTreeCreate(&pMerger->dReader.merger.rbt, tRowInfoCmprFn); // tRBTreeCreate(&pMerger->dReader.merger.rbt, tRowInfoCmprFn);
for (int8_t iLast = 0; iLast < pSet->nLastF; iLast++) { // for (int8_t iLast = 0; iLast < pSet->nLastF; iLast++) {
SRBTreeNode *pNode = (SRBTreeNode *)taosMemoryCalloc(1, sizeof(*pNode) + sizeof(SLDataIter)); // SRBTreeNode *pNode = (SRBTreeNode *)taosMemoryCalloc(1, sizeof(*pNode) + sizeof(SLDataIter));
if (pNode == NULL) { // if (pNode == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; // code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; // goto _err;
} // }
SLDataIter *pIter = (SLDataIter *)pNode->payload; // SLDataIter *pIter = (SLDataIter *)pNode->payload;
pIter->pReader = pMerger->dReader.pReader; // pIter->pReader = pMerger->dReader.pReader;
pIter->iLast = iLast; // pIter->iLast = iLast;
pIter->aBlockL = taosArrayInit(0, sizeof(SBlockL)); // pIter->aBlockL = taosArrayInit(0, sizeof(SBlockL));
if (pIter->aBlockL == NULL) { // if (pIter->aBlockL == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; // code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; // goto _err;
} // }
code = tBlockDataCreate(&pIter->bData); // code = tBlockDataCreate(&pIter->bData);
if (code) goto _err; // if (code) goto _err;
code = tsdbReadBlockL(pMerger->dReader.pReader, iLast, pIter->aBlockL); // code = tsdbReadBlockL(pMerger->dReader.pReader, iLast, pIter->aBlockL);
if (code) goto _err; // if (code) goto _err;
if (taosArrayGetSize(pIter->aBlockL) == 0) continue; // if (taosArrayGetSize(pIter->aBlockL) == 0) continue;
pIter->iBlockL = 0; // pIter->iBlockL = 0;
SBlockL *pBlockL = (SBlockL *)taosArrayGet(pIter->aBlockL, 0); // SBlockL *pBlockL = (SBlockL *)taosArrayGet(pIter->aBlockL, 0);
code = tsdbReadLastBlockEx(pMerger->dReader.pReader, iLast, pBlockL, &pIter->bData); // code = tsdbReadLastBlockEx(pMerger->dReader.pReader, iLast, pBlockL, &pIter->bData);
if (code) goto _err; // if (code) goto _err;
pIter->iRow = 0; // pIter->iRow = 0;
pIter->rowInfo.suid = pIter->bData.suid; // pIter->rowInfo.suid = pIter->bData.suid;
pIter->rowInfo.uid = pIter->bData.uid ? pIter->bData.uid : pIter->bData.aUid[0]; // pIter->rowInfo.uid = pIter->bData.uid ? pIter->bData.uid : pIter->bData.aUid[0];
pIter->rowInfo.row = tsdbRowFromBlockData(&pIter->bData, 0); // pIter->rowInfo.row = tsdbRowFromBlockData(&pIter->bData, 0);
pNode = tRBTreePut(&pMerger->dReader.merger.rbt, pNode); // pNode = tRBTreePut(&pMerger->dReader.merger.rbt, pNode);
ASSERT(pNode); // ASSERT(pNode);
pMerger->dReader.aLDataiter[iLast] = pIter; // pMerger->dReader.aLDataiter[iLast] = pIter;
} // }
// writer // // writer
SHeadFile fHead = {.commitID = pMerger->commitID}; // SHeadFile fHead = {.commitID = pMerger->commitID};
SDataFile fData = *pSet->pDataF; // SDataFile fData = *pSet->pDataF;
SSmaFile fSma = *pSet->pSmaF; // SSmaFile fSma = *pSet->pSmaF;
SLastFile fLast = {.commitID = pMerger->commitID}; // SLastFile fLast = {.commitID = pMerger->commitID};
SDFileSet wSet = {.diskId = pSet->diskId, // SDFileSet wSet = {.diskId = pSet->diskId,
.fid = pSet->fid, // .fid = pSet->fid,
.nLastF = 1, // .nLastF = 1,
.pHeadF = &fHead, // .pHeadF = &fHead,
.pDataF = &fData, // .pDataF = &fData,
.pSmaF = &fSma, // .pSmaF = &fSma,
.aLastF[0] = &fLast}; // .aLastF[0] = &fLast};
code = tsdbDataFWriterOpen(&pMerger->dWriter.pWriter, pTsdb, &wSet); // code = tsdbDataFWriterOpen(&pMerger->dWriter.pWriter, pTsdb, &wSet);
if (code) goto _err; // if (code) goto _err;
return code; // return code;
_err: // _err:
tsdbError("vgId:%d tsdb merge file data start failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code)); // tsdbError("vgId:%d tsdb merge file data start failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code; // return code;
} // }
static int32_t tsdbMergeFileDataEnd(STsdbMerger *pMerger) { // static int32_t tsdbMergeFileDataEnd(STsdbMerger *pMerger) {
int32_t code = 0; // int32_t code = 0;
STsdb *pTsdb = pMerger->pTsdb; // STsdb *pTsdb = pMerger->pTsdb;
// write aBlockIdx // // write aBlockIdx
code = tsdbWriteBlockIdx(pMerger->dWriter.pWriter, pMerger->dWriter.aBlockIdx); // code = tsdbWriteBlockIdx(pMerger->dWriter.pWriter, pMerger->dWriter.aBlockIdx);
if (code) goto _err; // if (code) goto _err;
// write aBlockL // // write aBlockL
code = tsdbWriteBlockL(pMerger->dWriter.pWriter, pMerger->dWriter.aBlockL); // code = tsdbWriteBlockL(pMerger->dWriter.pWriter, pMerger->dWriter.aBlockL);
if (code) goto _err; // if (code) goto _err;
// update file header // // update file header
code = tsdbUpdateDFileSetHeader(pMerger->dWriter.pWriter); // code = tsdbUpdateDFileSetHeader(pMerger->dWriter.pWriter);
if (code) goto _err; // if (code) goto _err;
// upsert SDFileSet // // upsert SDFileSet
code = tsdbFSUpsertFSet(&pMerger->fs, &pMerger->dWriter.pWriter->wSet); // code = tsdbFSUpsertFSet(&pMerger->fs, &pMerger->dWriter.pWriter->wSet);
if (code) goto _err; // if (code) goto _err;
// close and sync // // close and sync
code = tsdbDataFWriterClose(&pMerger->dWriter.pWriter, 1); // code = tsdbDataFWriterClose(&pMerger->dWriter.pWriter, 1);
if (code) goto _err; // if (code) goto _err;
if (pMerger->dReader.pReader) { // if (pMerger->dReader.pReader) {
code = tsdbDataFReaderClose(&pMerger->dReader.pReader); // code = tsdbDataFReaderClose(&pMerger->dReader.pReader);
if (code) goto _err; // if (code) goto _err;
} // }
return code; // return code;
_err: // _err:
tsdbError("vgId:%d tsdb merge file data end failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code)); // tsdbError("vgId:%d tsdb merge file data end failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code; // return code;
} // }
static int32_t tsdbMergeFileData(STsdbMerger *pMerger, SDFileSet *pSet) { // static int32_t tsdbMergeFileData(STsdbMerger *pMerger, SDFileSet *pSet) {
int32_t code = 0; // int32_t code = 0;
STsdb *pTsdb = pMerger->pTsdb; // STsdb *pTsdb = pMerger->pTsdb;
// start // // start
code = tsdbMergeFileDataStart(pMerger, pSet); // code = tsdbMergeFileDataStart(pMerger, pSet);
if (code) goto _err; // if (code) goto _err;
// impl // // impl
SRowInfo *pInfo; // SRowInfo *pInfo;
TABLEID id = {0}; // TABLEID id = {0};
while (true) { // while (true) {
code = tDataMergeNext(&pMerger->dReader.merger, &pInfo); // code = tDataMergeNext(&pMerger->dReader.merger, &pInfo);
if (code) goto _err; // if (code) goto _err;
if (pInfo == NULL) { // if (pInfo == NULL) {
if (pMerger->dWriter.bData.nRow > 0) { // if (pMerger->dWriter.bData.nRow > 0) {
// TODO // // TODO
} // }
if (pMerger->dWriter.bDatal.nRow > 0) { // if (pMerger->dWriter.bDatal.nRow > 0) {
// TODO // // TODO
} // }
break; // break;
} // }
if (id.suid != pInfo->suid || id.uid != pInfo->uid) { // if (id.suid != pInfo->suid || id.uid != pInfo->uid) {
while (true) { // while (true) {
// move commit the head data // // move commit the head data
} // }
// prepare to commit next // // prepare to commit next
} // }
code = tBlockDataAppendRow(&pMerger->dWriter.bData, &pInfo->row, NULL, pInfo->uid); // code = tBlockDataAppendRow(&pMerger->dWriter.bData, &pInfo->row, NULL, pInfo->uid);
if (code) goto _err; // if (code) goto _err;
if (pMerger->dWriter.bData.nRow >= pMerger->maxRow * 4 / 5) { // if (pMerger->dWriter.bData.nRow >= pMerger->maxRow * 4 / 5) {
// code = tsdbCommitDataBlock(); // // code = tsdbCommitDataBlock();
if (code) goto _err; // if (code) goto _err;
} // }
} // }
// end // // end
code = tsdbMergeFileDataEnd(pMerger); // code = tsdbMergeFileDataEnd(pMerger);
if (code) goto _err; // if (code) goto _err;
return code; // return code;
_err: // _err:
tsdbError("vgId:%d tsdb merge file data failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code)); // tsdbError("vgId:%d tsdb merge file data failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code; // return code;
} // }
static int32_t tsdbStartMerge(STsdbMerger *pMerger, STsdb *pTsdb) { // static int32_t tsdbStartMerge(STsdbMerger *pMerger, STsdb *pTsdb) {
int32_t code = 0; // int32_t code = 0;
pMerger->pTsdb = pTsdb; // pMerger->pTsdb = pTsdb;
pMerger->maxLast = TSDB_DEFAULT_LAST_FILE; // pMerger->maxLast = TSDB_DEFAULT_LAST_FILE;
pMerger->commitID = ++pTsdb->pVnode->state.commitID; // pMerger->commitID = ++pTsdb->pVnode->state.commitID;
code = tsdbFSCopy(pTsdb, &pMerger->fs); // code = tsdbFSCopy(pTsdb, &pMerger->fs);
if (code) goto _exit; // if (code) goto _exit;
// reader // // reader
pMerger->dReader.aBlockIdx = taosArrayInit(0, sizeof(SBlockIdx)); // pMerger->dReader.aBlockIdx = taosArrayInit(0, sizeof(SBlockIdx));
if (pMerger->dReader.aBlockIdx == NULL) { // if (pMerger->dReader.aBlockIdx == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; // code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit; // goto _exit;
} // }
// for (int8_t iLast = 0; iLast < TSDB_MAX_LAST_FILE; iLast++) { // // for (int8_t iLast = 0; iLast < TSDB_MAX_LAST_FILE; iLast++) {
// pMerger->dReader.aBlockL[iLast] = taosArrayInit(0, sizeof(SBlockL)); // // pMerger->dReader.aBlockL[iLast] = taosArrayInit(0, sizeof(SBlockL));
// if (pMerger->dReader.aBlockL[iLast] == NULL) { // // if (pMerger->dReader.aBlockL[iLast] == NULL) {
// code = TSDB_CODE_OUT_OF_MEMORY; // // code = TSDB_CODE_OUT_OF_MEMORY;
// goto _exit; // // goto _exit;
// } // // }
// } // // }
// writer // // writer
pMerger->dWriter.aBlockIdx = taosArrayInit(0, sizeof(SBlockIdx)); // pMerger->dWriter.aBlockIdx = taosArrayInit(0, sizeof(SBlockIdx));
if (pMerger->dWriter.aBlockIdx == NULL) { // if (pMerger->dWriter.aBlockIdx == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; // code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit; // goto _exit;
} // }
pMerger->dWriter.aBlockL = taosArrayInit(0, sizeof(SBlockL)); // pMerger->dWriter.aBlockL = taosArrayInit(0, sizeof(SBlockL));
if (pMerger->dWriter.aBlockL == NULL) { // if (pMerger->dWriter.aBlockL == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; // code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit; // goto _exit;
} // }
_exit: // _exit:
return code; // return code;
} // }
static int32_t tsdbEndMerge(STsdbMerger *pMerger) { // static int32_t tsdbEndMerge(STsdbMerger *pMerger) {
int32_t code = 0; // int32_t code = 0;
STsdb *pTsdb = pMerger->pTsdb; // STsdb *pTsdb = pMerger->pTsdb;
code = tsdbFSCommit1(pTsdb, &pMerger->fs); // code = tsdbFSCommit1(pTsdb, &pMerger->fs);
if (code) goto _err; // if (code) goto _err;
taosThreadRwlockWrlock(&pTsdb->rwLock); // taosThreadRwlockWrlock(&pTsdb->rwLock);
code = tsdbFSCommit2(pTsdb, &pMerger->fs); // code = tsdbFSCommit2(pTsdb, &pMerger->fs);
if (code) { // if (code) {
taosThreadRwlockUnlock(&pTsdb->rwLock); // taosThreadRwlockUnlock(&pTsdb->rwLock);
goto _err; // goto _err;
} // }
taosThreadRwlockUnlock(&pTsdb->rwLock); // taosThreadRwlockUnlock(&pTsdb->rwLock);
// writer // // writer
taosArrayDestroy(pMerger->dWriter.aBlockL); // taosArrayDestroy(pMerger->dWriter.aBlockL);
taosArrayDestroy(pMerger->dWriter.aBlockIdx); // taosArrayDestroy(pMerger->dWriter.aBlockIdx);
// reader // // reader
// for (int8_t iLast = 0; iLast < TSDB_MAX_LAST_FILE; iLast++) { // // for (int8_t iLast = 0; iLast < TSDB_MAX_LAST_FILE; iLast++) {
// taosArrayDestroy(pMerger->dReader.aBlockL[iLast]); // // taosArrayDestroy(pMerger->dReader.aBlockL[iLast]);
// } // // }
taosArrayDestroy(pMerger->dReader.aBlockIdx); // taosArrayDestroy(pMerger->dReader.aBlockIdx);
tsdbFSDestroy(&pMerger->fs); // tsdbFSDestroy(&pMerger->fs);
return code; // return code;
_err: // _err:
tsdbError("vgId:%d, tsdb end merge failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code)); // tsdbError("vgId:%d, tsdb end merge failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code; // return code;
} // }
int32_t tsdbMerge(STsdb *pTsdb) { // int32_t tsdbMerge(STsdb *pTsdb) {
int32_t code = 0; // int32_t code = 0;
STsdbMerger merger = {0}; // STsdbMerger merger = {0};
code = tsdbStartMerge(&merger, pTsdb); // code = tsdbStartMerge(&merger, pTsdb);
if (code) goto _err; // if (code) goto _err;
for (int32_t iSet = 0; iSet < taosArrayGetSize(merger.fs.aDFileSet); iSet++) { // for (int32_t iSet = 0; iSet < taosArrayGetSize(merger.fs.aDFileSet); iSet++) {
SDFileSet *pSet = (SDFileSet *)taosArrayGet(merger.fs.aDFileSet, iSet); // SDFileSet *pSet = (SDFileSet *)taosArrayGet(merger.fs.aDFileSet, iSet);
if (pSet->nLastF < merger.maxLast) continue; // if (pSet->nLastF < merger.maxLast) continue;
code = tsdbMergeFileData(&merger, pSet); // code = tsdbMergeFileData(&merger, pSet);
if (code) goto _err; // if (code) goto _err;
} // }
code = tsdbEndMerge(&merger); // code = tsdbEndMerge(&merger);
if (code) goto _err; // if (code) goto _err;
return code; // return code;
_err: // _err:
tsdbError("vgId:%d tsdb merge failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code)); // tsdbError("vgId:%d tsdb merge failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code; // return code;
} // }