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-06-09 16:07:06 +08:00
parent 1ee9f80e1f
commit cbd5da5dee
1 changed files with 146 additions and 140 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

160
source/dnode/vnode/src/tsdb/dev/tsdbDataFileRW.c
View File

@ -491,13 +491,6 @@ struct SDataFileWriter {
int32_t tombBlkArrayIdx; int32_t tombBlkArrayIdx;
STombBlock tombBlock[1]; STombBlock tombBlock[1];
int32_t tombBlockIdx; int32_t tombBlockIdx;
#if 0
const TBlockIdxArray *blockIdxArray;
int32_t blockIdxArrayIdx;
const TDataBlkArray *dataBlkArray;
int32_t dataBlkArrayIdx;
#endif
} ctx[1]; } ctx[1];
STFile files[TSDB_FTYPE_MAX]; STFile files[TSDB_FTYPE_MAX];
@ -512,36 +505,8 @@ struct SDataFileWriter {
TTombBlkArray tombBlkArray[1]; TTombBlkArray tombBlkArray[1];
STombBlock tombBlock[1]; STombBlock tombBlock[1];
#if 0
TBlockIdxArray blockIdxArray[1];
TDataBlkArray dataBlkArray[1];
#endif
}; };
#if 0
static int32_t tsdbDataFileWriteBlockIdx(SDataFileWriter *writer) {
int32_t code = 0;
int32_t lino = 0;
writer->headFooter->blockIdxPtr->offset = writer->files[TSDB_FTYPE_HEAD].size;
writer->headFooter->blockIdxPtr->size = TARRAY2_DATA_LEN(writer->blockIdxArray);
if (writer->headFooter->blockIdxPtr->size) {
code = tsdbWriteFile(writer->fd[TSDB_FTYPE_HEAD], writer->headFooter->blockIdxPtr->offset,
(void *)TARRAY2_DATA(writer->blockIdxArray), writer->headFooter->blockIdxPtr->size);
TSDB_CHECK_CODE(code, lino, _exit);
writer->files[TSDB_FTYPE_HEAD].size += writer->headFooter->blockIdxPtr->size;
}
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(writer->config->tsdb->pVnode), lino, code);
}
return code;
}
#endif
static int32_t tsdbDataFileWriterCloseAbort(SDataFileWriter *writer) { static int32_t tsdbDataFileWriterCloseAbort(SDataFileWriter *writer) {
ASSERT(0); ASSERT(0);
return 0; return 0;
@ -553,14 +518,14 @@ static int32_t tsdbDataFileWriterDoClose(SDataFileWriter *writer) {
} }
tTombBlockDestroy(writer->tombBlock); tTombBlockDestroy(writer->tombBlock);
// tStatisBlockDestroy(writer->statisBlock);
tBlockDataDestroy(writer->blockData);
TARRAY2_DESTROY(writer->tombBlkArray, NULL); TARRAY2_DESTROY(writer->tombBlkArray, NULL);
#if 0 tBlockDataDestroy(writer->blockData);
TARRAY2_DESTROY(writer->dataBlkArray, NULL); tBrinBlockDestroy(writer->brinBlock);
TARRAY2_DESTROY(writer->blockIdxArray, NULL); TARRAY2_DESTROY(writer->brinBlkArray, NULL);
#endif
tTombBlockDestroy(writer->ctx->tombBlock); tTombBlockDestroy(writer->ctx->tombBlock);
tBlockDataDestroy(writer->ctx->blockData);
tBrinBlockDestroy(writer->ctx->brinBlock);
for (int32_t i = 0; i < ARRAY_SIZE(writer->bufArr); ++i) { for (int32_t i = 0; i < ARRAY_SIZE(writer->bufArr); ++i) {
tFree(writer->bufArr[i]); tFree(writer->bufArr[i]);
@ -894,11 +859,6 @@ static int32_t tsdbDataFileWriteDataBlk(SDataFileWriter *writer, const TDataBlkA
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
writer->files[ftype].size += blockIdx->size; writer->files[ftype].size += blockIdx->size;
#if 0
code = TARRAY2_APPEND_PTR(writer->blockIdxArray, blockIdx);
TSDB_CHECK_CODE(code, lino, _exit);
#endif
_exit: _exit:
if (code) { if (code) {
TSDB_ERROR_LOG(TD_VID(writer->config->tsdb->pVnode), lino, code); TSDB_ERROR_LOG(TD_VID(writer->config->tsdb->pVnode), lino, code);
@ -947,20 +907,12 @@ _exit:
return code; return code;
} }
static int32_t tsdbDataFileDoWriteTSData(SDataFileWriter *writer, TSDBROW *row) { static int32_t tsdbDataFileDoWriteTableOldData(SDataFileWriter *writer, const TSDBKEY *key) {
if (writer->ctx->tbHasOldData == false) return 0;
int32_t code = 0; int32_t code = 0;
int32_t lino = 0; int32_t lino = 0;
if (writer->ctx->tbHasOldData) {
TSDBKEY key[1];
if (row->type == TSDBROW_ROW_FMT) {
key->ts = row->pTSRow->ts;
key->version = row->version;
} else {
key->ts = row->pBlockData->aTSKEY[row->iRow];
key->version = row->pBlockData->aVersion[row->iRow];
}
for (;;) { for (;;) {
for (;;) { for (;;) {
// SBlockData // SBlockData
@ -968,7 +920,7 @@ static int32_t tsdbDataFileDoWriteTSData(SDataFileWriter *writer, TSDBROW *row)
if (key->ts < writer->ctx->blockData->aTSKEY[writer->ctx->blockDataIdx] // if (key->ts < writer->ctx->blockData->aTSKEY[writer->ctx->blockDataIdx] //
|| (key->ts == writer->ctx->blockData->aTSKEY[writer->ctx->blockDataIdx] && || (key->ts == writer->ctx->blockData->aTSKEY[writer->ctx->blockDataIdx] &&
key->version < writer->ctx->blockData->aVersion[writer->ctx->blockDataIdx])) { key->version < writer->ctx->blockData->aVersion[writer->ctx->blockDataIdx])) {
goto _do_write; goto _exit;
} else { } else {
TSDBROW row1 = tsdbRowFromBlockData(writer->ctx->blockData, writer->ctx->blockDataIdx); TSDBROW row1 = tsdbRowFromBlockData(writer->ctx->blockData, writer->ctx->blockDataIdx);
code = tsdbDataFileDoWriteTSRow(writer, &row1); code = tsdbDataFileDoWriteTSRow(writer, &row1);
@ -984,13 +936,13 @@ static int32_t tsdbDataFileDoWriteTSData(SDataFileWriter *writer, TSDBROW *row)
for (; writer->ctx->brinBlockIdx < BRIN_BLOCK_SIZE(writer->ctx->brinBlock); writer->ctx->brinBlockIdx++) { for (; writer->ctx->brinBlockIdx < BRIN_BLOCK_SIZE(writer->ctx->brinBlock); writer->ctx->brinBlockIdx++) {
if (TARRAY2_GET(writer->ctx->brinBlock->uid, writer->ctx->brinBlockIdx) != writer->ctx->tbid->uid) { if (TARRAY2_GET(writer->ctx->brinBlock->uid, writer->ctx->brinBlockIdx) != writer->ctx->tbid->uid) {
writer->ctx->tbHasOldData = false; writer->ctx->tbHasOldData = false;
goto _do_write; goto _exit;
} }
if (key->ts < TARRAY2_GET(writer->ctx->brinBlock->firstKey, writer->ctx->brinBlockIdx) // if (key->ts < TARRAY2_GET(writer->ctx->brinBlock->firstKey, writer->ctx->brinBlockIdx) //
|| (key->ts == TARRAY2_GET(writer->ctx->brinBlock->firstKey, writer->ctx->brinBlockIdx) && || (key->ts == TARRAY2_GET(writer->ctx->brinBlock->firstKey, writer->ctx->brinBlockIdx) &&
key->version < TARRAY2_GET(writer->ctx->brinBlock->firstKeyVer, writer->ctx->brinBlockIdx))) { key->version < TARRAY2_GET(writer->ctx->brinBlock->firstKeyVer, writer->ctx->brinBlockIdx))) {
goto _do_write; goto _exit;
} else { } else {
SBrinRecord record[1]; SBrinRecord record[1];
tBrinBlockGet(writer->ctx->brinBlock, writer->ctx->brinBlockIdx, record); tBrinBlockGet(writer->ctx->brinBlock, writer->ctx->brinBlockIdx, record);
@ -1020,7 +972,7 @@ static int32_t tsdbDataFileDoWriteTSData(SDataFileWriter *writer, TSDBROW *row)
if (writer->ctx->brinBlkArrayIdx >= TARRAY2_SIZE(writer->ctx->brinBlkArray)) { if (writer->ctx->brinBlkArrayIdx >= TARRAY2_SIZE(writer->ctx->brinBlkArray)) {
writer->ctx->brinBlkArray = NULL; writer->ctx->brinBlkArray = NULL;
writer->ctx->tbHasOldData = false; writer->ctx->tbHasOldData = false;
goto _do_write; goto _exit;
} }
for (; writer->ctx->brinBlkArrayIdx < TARRAY2_SIZE(writer->ctx->brinBlkArray); writer->ctx->brinBlkArrayIdx++) { for (; writer->ctx->brinBlkArrayIdx < TARRAY2_SIZE(writer->ctx->brinBlkArray); writer->ctx->brinBlkArrayIdx++) {
@ -1028,7 +980,7 @@ static int32_t tsdbDataFileDoWriteTSData(SDataFileWriter *writer, TSDBROW *row)
if (brinBlk->minTbid.uid != writer->ctx->tbid->uid) { if (brinBlk->minTbid.uid != writer->ctx->tbid->uid) {
writer->ctx->tbHasOldData = false; writer->ctx->tbHasOldData = false;
goto _do_write; goto _exit;
} }
code = tsdbDataFileReadBrinBlock(writer->ctx->reader, brinBlk, writer->ctx->brinBlock); code = tsdbDataFileReadBrinBlock(writer->ctx->reader, brinBlk, writer->ctx->brinBlock);
@ -1039,9 +991,32 @@ static int32_t tsdbDataFileDoWriteTSData(SDataFileWriter *writer, TSDBROW *row)
break; break;
} }
} }
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(writer->config->tsdb->pVnode), lino, code);
}
return code;
}
static int32_t tsdbDataFileDoWriteTSData(SDataFileWriter *writer, TSDBROW *row) {
int32_t code = 0;
int32_t lino = 0;
if (writer->ctx->tbHasOldData) {
TSDBKEY key[1];
if (row->type == TSDBROW_ROW_FMT) {
key->ts = row->pTSRow->ts;
key->version = row->version;
} else {
key->ts = row->pBlockData->aTSKEY[row->iRow];
key->version = row->pBlockData->aVersion[row->iRow];
}
code = tsdbDataFileDoWriteTableOldData(writer, key);
TSDB_CHECK_CODE(code, lino, _exit);
} }
_do_write:
code = tsdbDataFileDoWriteTSRow(writer, row); code = tsdbDataFileDoWriteTSRow(writer, row);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
@ -1059,11 +1034,15 @@ static int32_t tsdbDataFileWriteTableDataEnd(SDataFileWriter *writer) {
int32_t lino = 0; int32_t lino = 0;
if (writer->ctx->tbHasOldData) { if (writer->ctx->tbHasOldData) {
for (; writer->ctx->blockDataIdx < writer->ctx->blockData->nRow; writer->ctx->blockDataIdx++) { TSDBKEY key = {
TSDBROW row = tsdbRowFromBlockData(writer->ctx->blockData, writer->ctx->blockDataIdx); .ts = TSKEY_MAX,
code = tsdbDataFileDoWriteTSRow(writer, &row); .version = VERSION_MAX,
};
code = tsdbDataFileDoWriteTableOldData(writer, &key);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
}
ASSERT(writer->ctx->tbHasOldData == false);
} }
code = tsdbDataFileDoWriteBlockData(writer, writer->blockData); code = tsdbDataFileDoWriteBlockData(writer, writer->blockData);
@ -1152,10 +1131,11 @@ static int32_t tsdbDataFileWriteHeadFooter(SDataFileWriter *writer) {
int32_t code = 0; int32_t code = 0;
int32_t lino = 0; int32_t lino = 0;
code = tsdbWriteFile(writer->fd[TSDB_FTYPE_HEAD], writer->files[TSDB_FTYPE_HEAD].size, int32_t ftype = TSDB_FTYPE_HEAD;
(const uint8_t *)writer->headFooter, sizeof(SHeadFooter)); code = tsdbWriteFile(writer->fd[ftype], writer->files[ftype].size, (const uint8_t *)writer->headFooter,
sizeof(SHeadFooter));
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
writer->files[TSDB_FTYPE_HEAD].size += sizeof(SHeadFooter); writer->files[ftype].size += sizeof(SHeadFooter);
_exit: _exit:
if (code) { if (code) {
@ -1327,6 +1307,28 @@ _exit:
return code; return code;
} }
static int32_t tsdbDataFileWriteBrinBlk(SDataFileWriter *writer) {
ASSERT(TARRAY2_SIZE(writer->brinBlkArray) > 0);
int32_t code = 0;
int32_t lino = 0;
int32_t ftype = TSDB_FTYPE_HEAD;
writer->headFooter->brinBlkPtr->offset = writer->files[ftype].size;
writer->headFooter->brinBlkPtr->size = TARRAY2_DATA_LEN(writer->brinBlkArray);
code = tsdbWriteFile(writer->fd[ftype], writer->headFooter->brinBlkPtr->offset,
(uint8_t *)TARRAY2_DATA(writer->brinBlkArray), writer->headFooter->brinBlkPtr->size);
TSDB_CHECK_CODE(code, lino, _exit);
writer->files[ftype].size += writer->headFooter->brinBlkPtr->size;
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(writer->config->tsdb->pVnode), lino, code);
}
return code;
}
static int32_t tsdbDataFileWriterCloseCommit(SDataFileWriter *writer, TFileOpArray *opArr) { static int32_t tsdbDataFileWriterCloseCommit(SDataFileWriter *writer, TFileOpArray *opArr) {
int32_t code = 0; int32_t code = 0;
int32_t lino = 0; int32_t lino = 0;
@ -1346,10 +1348,8 @@ static int32_t tsdbDataFileWriterCloseCommit(SDataFileWriter *writer, TFileOpArr
code = tsdbDataFileWriteTableDataBegin(writer, tbid); code = tsdbDataFileWriteTableDataBegin(writer, tbid);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
#if 0 code = tsdbDataFileWriteBrinBlk(writer);
code = tsdbDataFileWriteBlockIdx(writer);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
#endif
code = tsdbDataFileWriteHeadFooter(writer); code = tsdbDataFileWriteHeadFooter(writer);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
@ -1502,10 +1502,6 @@ static int32_t tsdbDataFileWriterOpenDataFD(SDataFileWriter *writer) {
if (writer->ctx->reader) { if (writer->ctx->reader) {
code = tsdbDataFileReadBrinBlk(writer->ctx->reader, &writer->ctx->brinBlkArray); code = tsdbDataFileReadBrinBlk(writer->ctx->reader, &writer->ctx->brinBlkArray);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
#if 0
code = tsdbDataFileReadBlockIdx(writer->ctx->reader, &writer->ctx->blockIdxArray);
TSDB_CHECK_CODE(code, lino, _exit);
#endif
} }
_exit: _exit:
@ -1605,6 +1601,16 @@ int32_t tsdbDataFileWriteBlockData(SDataFileWriter *writer, SBlockData *bData) {
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
} }
if (writer->ctx->tbHasOldData) {
TSDBKEY key = {
.ts = bData->aTSKEY[0],
.version = bData->aVersion[0],
};
code = tsdbDataFileDoWriteTableOldData(writer, &key);
TSDB_CHECK_CODE(code, lino, _exit);
}
if (!writer->ctx->tbHasOldData // if (!writer->ctx->tbHasOldData //
&& writer->blockData->nRow == 0 // && writer->blockData->nRow == 0 //
) { ) {