Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
提交 fb2b42f9 编写于 作者: H Hongze Cheng
浏览文件
操作

refact more code

上级 545e4d7f
隐藏空白更改
内联 并排
Showing with 276 addition and 122 deletion
source/dnode/vnode/src/tsdb/tsdbSnapshot.c
浏览文件 @ fb2b42f9
...@@ -21,6 +21,7 @@ extern int32_t tsdbReadDataBlockEx(SDataFReader* pReader, SDataBlk* pDataBlk, SB ...@@ -21,6 +21,7 @@ extern int32_t tsdbReadDataBlockEx(SDataFReader* pReader, SDataBlk* pDataBlk, SB
#define TSDB_MEM_TABLE_DATA_ITER 0 #define TSDB_MEM_TABLE_DATA_ITER 0
#define TSDB_DATA_FILE_DATA_ITER 1 #define TSDB_DATA_FILE_DATA_ITER 1
#define TSDB_STT_FILE_DATA_ITER 2 #define TSDB_STT_FILE_DATA_ITER 2
#define TSDB_TOMB_FILE_DATA_ITER 3
typedef struct STsdbDataIter2 STsdbDataIter2; typedef struct STsdbDataIter2 STsdbDataIter2;
struct STsdbDataIter2 { struct STsdbDataIter2 {
...@@ -55,6 +56,14 @@ struct STsdbDataIter2 { ...@@ -55,6 +56,14 @@ struct STsdbDataIter2 {
int32_t iSttBlk; int32_t iSttBlk;
int32_t iRow; int32_t iRow;
} sIter; } sIter;
// TSDB_TOMB_FILE_DATA_ITER
struct {
SDelFReader* pReader;
SArray* aDelIdx;
SArray* aDelData;
int32_t iDelIdx;
int32_t iDelData;
} tIter;
}; };
}; };
...@@ -152,6 +161,49 @@ _exit: ...@@ -152,6 +161,49 @@ _exit:
return code; return code;
} }
static int32_t tsdbOpenTombFileDataIter(SDelFReader* pReader, STsdbDataIter2** ppIter) {
int32_t code = 0;
int32_t lino = 0;
STsdbDataIter2* pIter = (STsdbDataIter2*)taosMemoryCalloc(1, sizeof(*pIter));
if (pIter == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
pIter->type = TSDB_TOMB_FILE_DATA_ITER;
pIter->tIter.pReader = pReader;
if ((pIter->tIter.aDelIdx = taosArrayInit(0, sizeof(SDelIdx))) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
if ((pIter->tIter.aDelData = taosArrayInit(0, sizeof(SDelData))) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
code = tsdbReadDelIdx(pReader, pIter->tIter.aDelIdx);
TSDB_CHECK_CODE(code, lino, _exit);
if (taosArrayGetSize(pIter->tIter.aDelIdx) == 0) goto _clear;
pIter->tIter.iDelIdx = 0;
pIter->tIter.iDelData = 0;
_exit:
if (code) {
if (pIter) {
_clear:
taosArrayDestroy(pIter->tIter.aDelIdx);
taosArrayDestroy(pIter->tIter.aDelData);
taosMemoryFree(pIter);
pIter = NULL;
}
}
*ppIter = pIter;
return code;
}
/* close */ /* close */
static void tsdbCloseDataFileDataIter(STsdbDataIter2* pIter) { static void tsdbCloseDataFileDataIter(STsdbDataIter2* pIter) {
tBlockDataDestroy(&pIter->dIter.bData, 1); tBlockDataDestroy(&pIter->dIter.bData, 1);
...@@ -921,19 +973,19 @@ struct STsdbSnapWriter { ...@@ -921,19 +973,19 @@ struct STsdbSnapWriter {
uint8_t* aBuf[5]; uint8_t* aBuf[5];
STsdbFS fs; STsdbFS fs;
TABLEID tbid;
// time-series data // time-series data
SBlockData inData; SBlockData inData;
int32_t fid; int32_t fid;
TABLEID tbid;
SSkmInfo skmTable; SSkmInfo skmTable;
/* reader */ /* reader */
SDataFReader* pDataFReader; SDataFReader* pDataFReader;
STsdbDataIter2* iterList; STsdbDataIter2* iterList;
STsdbDataIter2* pDIter; STsdbDataIter2* pDIter;
STsdbDataIter2* pIter; STsdbDataIter2* pSIter;
SRBTree rbt; // SRBTree<STsdbDataIter2> SRBTree rbt; // SRBTree<STsdbDataIter2>
/* writer */ /* writer */
...@@ -945,12 +997,14 @@ struct STsdbSnapWriter { ...@@ -945,12 +997,14 @@ struct STsdbSnapWriter {
SBlockData sData; SBlockData sData;
// tombstone data // tombstone data
SDelFReader* pDelFReader; /* reader */
SDelFReader* pDelFReader;
STsdbDataIter2* pTIter;
/* writer */
SDelFWriter* pDelFWriter; SDelFWriter* pDelFWriter;
int32_t iDelIdx; SArray* aDelIdx;
SArray* aDelIdxR;
SArray* aDelData; SArray* aDelData;
SArray* aDelIdxW;
}; };
// SNAP_DATA_TSDB // SNAP_DATA_TSDB
...@@ -1207,6 +1261,7 @@ static int32_t tsdbSnapWriteTableDataEnd(STsdbSnapWriter* pWriter) { ...@@ -1207,6 +1261,7 @@ static int32_t tsdbSnapWriteTableDataEnd(STsdbSnapWriter* pWriter) {
if (pWriter->bData.nRow > 0) { if (pWriter->bData.nRow > 0) {
if (pWriter->bData.nRow < pWriter->minRow) { if (pWriter->bData.nRow < pWriter->minRow) {
ASSERT(TABLE_SAME_SCHEMA(pWriter->sData.suid, pWriter->sData.uid, pWriter->tbid.suid, pWriter->tbid.uid));
for (int32_t iRow = 0; iRow < pWriter->bData.nRow; iRow++) { for (int32_t iRow = 0; iRow < pWriter->bData.nRow; iRow++) {
code = code =
tBlockDataAppendRow(&pWriter->sData, &tsdbRowFromBlockData(&pWriter->bData, iRow), NULL, pWriter->tbid.uid); tBlockDataAppendRow(&pWriter->sData, &tsdbRowFromBlockData(&pWriter->bData, iRow), NULL, pWriter->tbid.uid);
...@@ -1262,7 +1317,7 @@ static int32_t tsdbSnapWriteFileDataStart(STsdbSnapWriter* pWriter, int32_t fid) ...@@ -1262,7 +1317,7 @@ static int32_t tsdbSnapWriteFileDataStart(STsdbSnapWriter* pWriter, int32_t fid)
pWriter->pDataFReader = NULL; pWriter->pDataFReader = NULL;
pWriter->iterList = NULL; pWriter->iterList = NULL;
pWriter->pDIter = NULL; pWriter->pDIter = NULL;
pWriter->pIter = NULL; pWriter->pSIter = NULL;
tRBTreeCreate(&pWriter->rbt, tsdbDataIterCmprFn); tRBTreeCreate(&pWriter->rbt, tsdbDataIterCmprFn);
if (pSet) { if (pSet) {
code = tsdbDataFReaderOpen(&pWriter->pDataFReader, pTsdb, pSet); code = tsdbDataFReaderOpen(&pWriter->pDataFReader, pTsdb, pSet);
...@@ -1276,23 +1331,23 @@ static int32_t tsdbSnapWriteFileDataStart(STsdbSnapWriter* pWriter, int32_t fid) ...@@ -1276,23 +1331,23 @@ static int32_t tsdbSnapWriteFileDataStart(STsdbSnapWriter* pWriter, int32_t fid)
} }
for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) { for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) {
code = tsdbOpenSttFileDataIter(pWriter->pDataFReader, iStt, &pWriter->pIter); code = tsdbOpenSttFileDataIter(pWriter->pDataFReader, iStt, &pWriter->pSIter);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
if (pWriter->pIter) { if (pWriter->pSIter) {
code = tsdbSttFileDataIterNext(pWriter->pIter); code = tsdbSttFileDataIterNext(pWriter->pSIter);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
// add to tree // add to tree
tRBTreePut(&pWriter->rbt, &pWriter->pIter->rbtn); tRBTreePut(&pWriter->rbt, &pWriter->pSIter->rbtn);
// add to list // add to list
pWriter->pIter->next = pWriter->iterList; pWriter->pSIter->next = pWriter->iterList;
pWriter->iterList = pWriter->pIter; pWriter->iterList = pWriter->pSIter;
} }
} }
pWriter->pIter = NULL; pWriter->pSIter = NULL;
} }
// open writer // open writer
...@@ -1348,7 +1403,7 @@ static int32_t tsdbSnapWriteTableData(STsdbSnapWriter* pWriter, SRowInfo* pRowIn ...@@ -1348,7 +1403,7 @@ static int32_t tsdbSnapWriteTableData(STsdbSnapWriter* pWriter, SRowInfo* pRowIn
// switch to new table if need // switch to new table if need
if (pRowInfo == NULL || pRowInfo->uid != pWriter->tbid.uid) { if (pRowInfo == NULL || pRowInfo->uid != pWriter->tbid.uid) {
if (pWriter->tbid.uid != 0) { if (pWriter->tbid.uid) {
code = tsdbSnapWriteTableDataEnd(pWriter); code = tsdbSnapWriteTableDataEnd(pWriter);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
} }
...@@ -1357,11 +1412,10 @@ static int32_t tsdbSnapWriteTableData(STsdbSnapWriter* pWriter, SRowInfo* pRowIn ...@@ -1357,11 +1412,10 @@ static int32_t tsdbSnapWriteTableData(STsdbSnapWriter* pWriter, SRowInfo* pRowIn
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
} }
// end with a NULL row if (pRowInfo == NULL) goto _exit;
if (pRowInfo) {
code = tsdbSnapWriteTableRow(pWriter, &pRowInfo->row); code = tsdbSnapWriteTableRow(pWriter, &pRowInfo->row);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
}
_exit: _exit:
if (code) { if (code) {
...@@ -1374,19 +1428,19 @@ static int32_t tsdbSnapWriteNextRow(STsdbSnapWriter* pWriter, SRowInfo** ppRowIn ...@@ -1374,19 +1428,19 @@ static int32_t tsdbSnapWriteNextRow(STsdbSnapWriter* pWriter, SRowInfo** ppRowIn
int32_t code = 0; int32_t code = 0;
int32_t lino = 0; int32_t lino = 0;
if (pWriter->pIter) { if (pWriter->pSIter) {
code = tsdbDataIterNext2(pWriter->pIter); code = tsdbDataIterNext2(pWriter->pSIter);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
if (pWriter->pIter->rowInfo.suid == 0 && pWriter->pIter->rowInfo.uid == 0) { if (pWriter->pSIter->rowInfo.suid == 0 && pWriter->pSIter->rowInfo.uid == 0) {
pWriter->pIter = NULL; pWriter->pSIter = NULL;
} else { } else {
SRBTreeNode* pNode = tRBTreeMin(&pWriter->rbt); SRBTreeNode* pNode = tRBTreeMin(&pWriter->rbt);
if (pNode) { if (pNode) {
int32_t c = tsdbDataIterCmprFn(&pWriter->pIter->rbtn, pNode); int32_t c = tsdbDataIterCmprFn(&pWriter->pSIter->rbtn, pNode);
if (c > 0) { if (c > 0) {
tRBTreePut(&pWriter->rbt, &pWriter->pIter->rbtn); tRBTreePut(&pWriter->rbt, &pWriter->pSIter->rbtn);
pWriter->pIter = NULL; pWriter->pSIter = NULL;
} else if (c == 0) { } else if (c == 0) {
ASSERT(0); ASSERT(0);
} }
...@@ -1394,17 +1448,17 @@ static int32_t tsdbSnapWriteNextRow(STsdbSnapWriter* pWriter, SRowInfo** ppRowIn ...@@ -1394,17 +1448,17 @@ static int32_t tsdbSnapWriteNextRow(STsdbSnapWriter* pWriter, SRowInfo** ppRowIn
} }
} }
if (pWriter->pIter == NULL) { if (pWriter->pSIter == NULL) {
SRBTreeNode* pNode = tRBTreeMin(&pWriter->rbt); SRBTreeNode* pNode = tRBTreeMin(&pWriter->rbt);
if (pNode) { if (pNode) {
tRBTreeDrop(&pWriter->rbt, pNode); tRBTreeDrop(&pWriter->rbt, pNode);
pWriter->pIter = TSDB_RBTN_TO_DATA_ITER(pNode); pWriter->pSIter = TSDB_RBTN_TO_DATA_ITER(pNode);
} }
} }
if (ppRowInfo) { if (ppRowInfo) {
if (pWriter->pIter) { if (pWriter->pSIter) {
*ppRowInfo = &pWriter->pIter->rowInfo; *ppRowInfo = &pWriter->pSIter->rowInfo;
} else { } else {
*ppRowInfo = NULL; *ppRowInfo = NULL;
} }
...@@ -1421,8 +1475,8 @@ static int32_t tsdbSnapWriteGetRow(STsdbSnapWriter* pWriter, SRowInfo** ppRowInf ...@@ -1421,8 +1475,8 @@ static int32_t tsdbSnapWriteGetRow(STsdbSnapWriter* pWriter, SRowInfo** ppRowInf
int32_t code = 0; int32_t code = 0;
int32_t lino = 0; int32_t lino = 0;
if (pWriter->pIter) { if (pWriter->pSIter) {
*ppRowInfo = &pWriter->pIter->rowInfo; *ppRowInfo = &pWriter->pSIter->rowInfo;
goto _exit; goto _exit;
} }
...@@ -1558,143 +1612,238 @@ _exit: ...@@ -1558,143 +1612,238 @@ _exit:
} }
// SNAP_DATA_DEL // SNAP_DATA_DEL
static int32_t tsdbSnapMoveWriteDelData(STsdbSnapWriter* pWriter, TABLEID* pId) { static int32_t tsdbSnapWriteDelTableDataStart(STsdbSnapWriter* pWriter, TABLEID* pId) {
int32_t code = 0; int32_t code = 0;
int32_t lino = 0;
while (true) { if (pId) {
if (pWriter->iDelIdx >= taosArrayGetSize(pWriter->aDelIdxR)) break; pWriter->tbid = *pId;
} else {
pWriter->tbid = (TABLEID){.suid = INT64_MAX, .uid = INT64_MAX};
}
SDelIdx* pDelIdx = (SDelIdx*)taosArrayGet(pWriter->aDelIdxR, pWriter->iDelIdx); if (pWriter->pTIter) {
ASSERT(pWriter->pTIter->tIter.iDelData >= taosArrayGetSize(pWriter->pTIter->tIter.aDelData));
if (tTABLEIDCmprFn(pDelIdx, pId) >= 0) break; for (;;) {
if (pWriter->pTIter->tIter.iDelIdx >= taosArrayGetSize(pWriter->pTIter->tIter.aDelIdx)) {
break;
}
code = tsdbReadDelData(pWriter->pDelFReader, pDelIdx, pWriter->aDelData); SDelIdx* pDelIdx = taosArrayGet(pWriter->pTIter->tIter.aDelIdx, pWriter->pTIter->tIter.iDelIdx);
if (code) goto _exit;
SDelIdx delIdx = *pDelIdx; int32_t c = tTABLEIDCmprFn(pDelIdx, &pWriter->tbid);
code = tsdbWriteDelData(pWriter->pDelFWriter, pWriter->aDelData, &delIdx); if (c < 0) {
if (code) goto _exit; code = tsdbReadDelData(pWriter->pDelFReader, pDelIdx, pWriter->pTIter->tIter.aDelData);
TSDB_CHECK_CODE(code, lino, _exit);
if (taosArrayPush(pWriter->aDelIdxW, &delIdx) == NULL) { SDelIdx* pDelIdxNew = taosArrayReserve(pWriter->pTIter->tIter.aDelIdx, 1);
code = TSDB_CODE_OUT_OF_MEMORY; if (pDelIdxNew == NULL) {
goto _exit; code = TSDB_CODE_OUT_OF_MEMORY;
} TSDB_CHECK_CODE(code, lino, _exit);
}
pDelIdxNew->suid = pDelIdx->suid;
pDelIdxNew->uid = pDelIdx->uid;
pWriter->iDelIdx++; code = tsdbWriteDelData(pWriter->pDelFWriter, pWriter->pTIter->tIter.aDelData, pDelIdxNew);
TSDB_CHECK_CODE(code, lino, _exit);
pWriter->pTIter->tIter.iDelIdx++;
} else if (c == 0) {
code = tsdbReadDelData(pWriter->pDelFReader, pDelIdx, pWriter->pTIter->tIter.aDelData);
TSDB_CHECK_CODE(code, lino, _exit);
if (taosArrayAddBatch(pWriter->aDelData, pWriter->pTIter->tIter.aDelData->pData,
taosArrayGetSize(pWriter->pTIter->tIter.aDelData)) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
pWriter->pTIter->tIter.iDelData = taosArrayGetSize(pWriter->pTIter->tIter.aDelData);
pWriter->pTIter->tIter.iDelIdx++;
break;
} else {
pWriter->pTIter->tIter.iDelData = taosArrayGetSize(pWriter->pTIter->tIter.aDelData);
break;
}
}
} }
taosArrayClear(pWriter->aDelData);
_exit: _exit:
if (code) {
tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, lino, tstrerror(code));
} else {
tsdbTrace("vgId:%d %s done, suid:%" PRId64 " uid:%" PRId64, TD_VID(pWriter->pTsdb->pVnode), __func__, pId->suid,
pId->uid);
}
return code; return code;
} }
static int32_t tsdbSnapWriteDelData(STsdbSnapWriter* pWriter, SSnapDataHdr* pHdr) { static int32_t tsdbSnapWriteDelTableDataEnd(STsdbSnapWriter* pWriter) {
int32_t code = 0; int32_t code = 0;
STsdb* pTsdb = pWriter->pTsdb; int32_t lino = 0;
// Open del file if not opened yet if (taosArrayGetSize(pWriter->aDelData) > 0) {
if (pWriter->pDelFWriter == NULL) { SDelIdx* pDelIdx = taosArrayReserve(pWriter->aDelIdx, 1);
SDelFile* pDelFile = pWriter->fs.pDelFile; if (pDelIdx == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
// reader pDelIdx->suid = pWriter->tbid.suid;
if (pDelFile) { pDelIdx->uid = pWriter->tbid.uid;
code = tsdbDelFReaderOpen(&pWriter->pDelFReader, pDelFile, pTsdb);
if (code) goto _err;
code = tsdbReadDelIdx(pWriter->pDelFReader, pWriter->aDelIdxR); code = tsdbWriteDelData(pWriter->pDelFWriter, pWriter->aDelData, pDelIdx);
if (code) goto _err; TSDB_CHECK_CODE(code, lino, _exit);
} else { }
taosArrayClear(pWriter->aDelIdxR);
_exit:
if (code) {
tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, lino, tstrerror(code));
} else {
tsdbTrace("vgId:%d %s done", TD_VID(pWriter->pTsdb->pVnode), __func__);
}
return code;
}
static int32_t tsdbSnapWriteDelTableData(STsdbSnapWriter* pWriter, TABLEID* pId, uint8_t* pData, int64_t size) {
int32_t code = 0;
int32_t lino = 0;
if (pId == NULL || pId->uid != pWriter->tbid.uid) {
if (pWriter->tbid.uid) {
code = tsdbSnapWriteDelTableDataEnd(pWriter);
TSDB_CHECK_CODE(code, lino, _exit);
} }
pWriter->iDelIdx = 0;
// writer code = tsdbSnapWriteDelTableDataStart(pWriter, pId);
SDelFile delFile = {.commitID = pWriter->commitID}; TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbDelFWriterOpen(&pWriter->pDelFWriter, &delFile, pTsdb);
if (code) goto _err;
taosArrayClear(pWriter->aDelIdxW);
} }
TABLEID id = *(TABLEID*)pHdr->data; if (pId == NULL) goto _exit;
// Move write data < id int64_t n = 0;
code = tsdbSnapMoveWriteDelData(pWriter, &id); while (n < size) {
if (code) goto _err; SDelData delData;
n += tGetDelData(pData + n, &delData);
// Merge incoming data with current if (taosArrayPush(pWriter->aDelData, &delData) < 0) {
if (pWriter->iDelIdx < taosArrayGetSize(pWriter->aDelIdxR) && code = TSDB_CODE_OUT_OF_MEMORY;
tTABLEIDCmprFn(taosArrayGet(pWriter->aDelIdxR, pWriter->iDelIdx), &id) == 0) { TSDB_CHECK_CODE(code, lino, _exit);
SDelIdx* pDelIdx = (SDelIdx*)taosArrayGet(pWriter->aDelIdxR, pWriter->iDelIdx); }
}
code = tsdbReadDelData(pWriter->pDelFReader, pDelIdx, pWriter->aDelData); ASSERT(n == size);
if (code) goto _err;
pWriter->iDelIdx++; _exit:
} else { if (code) {
taosArrayClear(pWriter->aDelData); tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, lino, tstrerror(code));
} }
return code;
}
int64_t n = sizeof(TABLEID); static int32_t tsdbSnapWriteDelDataStart(STsdbSnapWriter* pWriter) {
while (n < pHdr->size) { int32_t code = 0;
SDelData delData; int32_t lino = 0;
n += tGetDelData(pHdr->data + n, &delData); STsdb* pTsdb = pWriter->pTsdb;
SDelFile* pDelFile = pWriter->fs.pDelFile;
if (taosArrayPush(pWriter->aDelData, &delData) == NULL) { pWriter->tbid = (TABLEID){0};
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; // reader
} if (pDelFile) {
code = tsdbDelFReaderOpen(&pWriter->pDelFReader, pDelFile, pTsdb);
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbOpenTombFileDataIter(pWriter->pDelFReader, &pWriter->pTIter);
TSDB_CHECK_CODE(code, lino, _exit);
} }
SDelIdx delIdx = {.suid = id.suid, .uid = id.uid}; // writer
code = tsdbWriteDelData(pWriter->pDelFWriter, pWriter->aDelData, &delIdx); code = tsdbDelFWriterOpen(&pWriter->pDelFWriter, &(SDelFile){.commitID = pWriter->commitID}, pTsdb);
if (code) goto _err; TSDB_CHECK_CODE(code, lino, _exit);
if (taosArrayPush(pWriter->aDelIdxW, &delIdx) == NULL) { if ((pWriter->aDelIdx = taosArrayInit(0, sizeof(SDelIdx))) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; TSDB_CHECK_CODE(code, lino, _exit);
}
if ((pWriter->aDelData = taosArrayInit(0, sizeof(SDelData))) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
} }
return code; _exit:
if (code) {
_err: tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
tsdbError("vgId:%d, vnode snapshot tsdb write del for %s failed since %s", TD_VID(pTsdb->pVnode), pTsdb->path, } else {
tstrerror(code)); tsdbDebug("vgId:%d %s done", TD_VID(pTsdb->pVnode), __func__);
}
return code; return code;
} }
static int32_t tsdbSnapWriteDelEnd(STsdbSnapWriter* pWriter) { static int32_t tsdbSnapWriteDelDataEnd(STsdbSnapWriter* pWriter) {
int32_t code = 0; int32_t code = 0;
STsdb* pTsdb = pWriter->pTsdb; int32_t lino = 0;
if (pWriter->pDelFWriter == NULL) return code; STsdb* pTsdb = pWriter->pTsdb;
TABLEID id = {.suid = INT64_MAX, .uid = INT64_MAX}; // end remaining table (TODO)
code = tsdbSnapMoveWriteDelData(pWriter, &id); code = tsdbSnapWriteDelTableData(pWriter, NULL, NULL, 0);
if (code) goto _err; TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbWriteDelIdx(pWriter->pDelFWriter, pWriter->aDelIdxW); // update file-level info
if (code) goto _err; code = tsdbWriteDelIdx(pWriter->pDelFWriter, pWriter->aDelIdx);
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbUpdateDelFileHdr(pWriter->pDelFWriter); code = tsdbUpdateDelFileHdr(pWriter->pDelFWriter);
if (code) goto _err; TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbFSUpsertDelFile(&pWriter->fs, &pWriter->pDelFWriter->fDel); code = tsdbFSUpsertDelFile(&pWriter->fs, &pWriter->pDelFWriter->fDel);
if (code) goto _err; TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbDelFWriterClose(&pWriter->pDelFWriter, 1); code = tsdbDelFWriterClose(&pWriter->pDelFWriter, 1);
if (code) goto _err; TSDB_CHECK_CODE(code, lino, _exit);
if (pWriter->pDelFReader) { if (pWriter->pDelFReader) {
code = tsdbDelFReaderClose(&pWriter->pDelFReader); code = tsdbDelFReaderClose(&pWriter->pDelFReader);
if (code) goto _err; TSDB_CHECK_CODE(code, lino, _exit);
} }
tsdbInfo("vgId:%d, vnode snapshot tsdb write del for %s end", TD_VID(pTsdb->pVnode), pTsdb->path); _exit:
if (code) {
tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
} else {
tsdbInfo("vgId:%d %s done", TD_VID(pTsdb->pVnode), __func__);
}
return code; return code;
}
_err: static int32_t tsdbSnapWriteDelData(STsdbSnapWriter* pWriter, SSnapDataHdr* pHdr) {
tsdbError("vgId:%d, vnode snapshot tsdb write del end for %s failed since %s", TD_VID(pTsdb->pVnode), pTsdb->path, int32_t code = 0;
tstrerror(code)); int32_t lino = 0;
STsdb* pTsdb = pWriter->pTsdb;
// start to write del data if need
if (pWriter->pDelFWriter == NULL) {
code = tsdbSnapWriteDelDataStart(pWriter);
TSDB_CHECK_CODE(code, lino, _exit);
}
code = tsdbSnapWriteDelTableData(pWriter, (TABLEID*)pHdr->data, pHdr->data + sizeof(TABLEID),
pHdr->size - sizeof(TABLEID));
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
tsdbError("vgId:%d %s failed since %s", TD_VID(pTsdb->pVnode), __func__, tstrerror(code));
} else {
tsdbTrace("vgId:%d %s done", TD_VID(pTsdb->pVnode), __func__);
}
return code; return code;
} }
...@@ -1738,7 +1887,7 @@ int32_t tsdbSnapWriterOpen(STsdb* pTsdb, int64_t sver, int64_t ever, STsdbSnapWr ...@@ -1738,7 +1887,7 @@ int32_t tsdbSnapWriterOpen(STsdb* pTsdb, int64_t sver, int64_t ever, STsdbSnapWr
_exit: _exit:
if (code) { if (code) {
tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code)); tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
if (pWriter) { if (pWriter) {
tBlockDataDestroy(&pWriter->sData, 1); tBlockDataDestroy(&pWriter->sData, 1);
tBlockDataDestroy(&pWriter->bData, 1); tBlockDataDestroy(&pWriter->bData, 1);
...@@ -1755,20 +1904,26 @@ _exit: ...@@ -1755,20 +1904,26 @@ _exit:
int32_t tsdbSnapWriterPrepareClose(STsdbSnapWriter* pWriter) { int32_t tsdbSnapWriterPrepareClose(STsdbSnapWriter* pWriter) {
int32_t code = 0; int32_t code = 0;
int32_t lino = 0;
if (pWriter->pDataFWriter) { if (pWriter->pDataFWriter) {
code = tsdbSnapWriteFileDataEnd(pWriter); code = tsdbSnapWriteFileDataEnd(pWriter);
if (code) goto _exit; TSDB_CHECK_CODE(code, lino, _exit);
} }
code = tsdbSnapWriteDelEnd(pWriter); if (pWriter->pDelFWriter) {
if (code) goto _exit; code = tsdbSnapWriteDelDataEnd(pWriter);
TSDB_CHECK_CODE(code, lino, _exit);
}
code = tsdbFSPrepareCommit(pWriter->pTsdb, &pWriter->fs); code = tsdbFSPrepareCommit(pWriter->pTsdb, &pWriter->fs);
if (code) goto _exit; TSDB_CHECK_CODE(code, lino, _exit);
_exit: _exit:
if (code) { if (code) {
tsdbError("vgId:%d, %s failed since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, tstrerror(code)); tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, lino, tstrerror(code));
} else {
tsdbDebug("vgId:%d %s done", TD_VID(pWriter->pTsdb->pVnode), __func__);
} }
return code; return code;
} }
...@@ -1795,9 +1950,8 @@ int32_t tsdbSnapWriterClose(STsdbSnapWriter** ppWriter, int8_t rollback) { ...@@ -1795,9 +1950,8 @@ int32_t tsdbSnapWriterClose(STsdbSnapWriter** ppWriter, int8_t rollback) {
} }
// SNAP_DATA_DEL // SNAP_DATA_DEL
taosArrayDestroy(pWriter->aDelIdxW);
taosArrayDestroy(pWriter->aDelData); taosArrayDestroy(pWriter->aDelData);
taosArrayDestroy(pWriter->aDelIdxR); taosArrayDestroy(pWriter->aDelIdx);
// SNAP_DATA_TSDB // SNAP_DATA_TSDB
tBlockDataDestroy(&pWriter->sData, 1); tBlockDataDestroy(&pWriter->sData, 1);
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或