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

more work

This commit is contained in:
Hongze Cheng 2022-06-20 06:41:17 +00:00
parent 4e9e0c9b28
commit 607f698fc6
6 changed files with 123 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
source/dnode/vnode/src/inc/tsdb.h
View File

@ -55,10 +55,10 @@ typedef struct SBlockData SBlockData;
typedef struct SReadH SReadH; typedef struct SReadH SReadH;
typedef struct SDelFile SDelFile; typedef struct SDelFile SDelFile;
typedef struct STsdbCacheFile STsdbCacheFile; typedef struct STsdbCacheFile STsdbCacheFile;
typedef struct STsdbIndexFile STsdbIndexFile; typedef struct SHeadFile SHeadFile;
typedef struct STsdbDataFile STsdbDataFile; typedef struct SDataFile SDataFile;
typedef struct STsdbLastFile STsdbLastFile; typedef struct SLastFile SLastFile;
typedef struct STsdbSmaFile STsdbSmaFile; typedef struct SSmaFile SSmaFile;
typedef struct SDFileSet SDFileSet; typedef struct SDFileSet SDFileSet;
typedef struct SDataFWriter SDataFWriter; typedef struct SDataFWriter SDataFWriter;
typedef struct SDataFReader SDataFReader; typedef struct SDataFReader SDataFReader;
@ -109,10 +109,10 @@ void tColDataReset(SColData *pColData);
void tColDataClear(SColData *pColData); void tColDataClear(SColData *pColData);
int32_t tColDataCmprFn(const void *p1, const void *p2); int32_t tColDataCmprFn(const void *p1, const void *p2);
// SBlockData // SBlockData
#define tsdbBlockDataCreate() ((SBlockData){0}) #define tBlockDataInit() ((SBlockData){0})
void tsdbBlockDataClear(SBlockData *pBlockData); void tBlockDataReset(SBlockData *pBlockData);
int32_t tsdbBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema); void tBlockDataClear(SBlockData *pBlockData);
void tsdbBlockDataDestroy(SBlockData *pBlockData); int32_t tBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema);
// SDelIdx // SDelIdx
int32_t tPutDelIdx(uint8_t *p, void *ph); int32_t tPutDelIdx(uint8_t *p, void *ph);
int32_t tGetDelIdx(uint8_t *p, void *ph); int32_t tGetDelIdx(uint8_t *p, void *ph);
@ -419,6 +419,34 @@ struct SBlockDataHdr {
int64_t uid; int64_t uid;
}; };
struct SHeadFile {
int64_t size;
int64_t offset;
int32_t nRef;
};
struct SDataFile {
int64_t size;
int32_t nRef;
};
struct SLastFile {
int64_t size;
int32_t nRef;
};
struct SSmaFile {
int64_t size;
int32_t nRef;
};
struct SDFileSet {
SHeadFile *pHeadFile;
SDataFile *pDataFile;
SLastFile *pLastFile;
SSmaFile *pSmaFile;
};
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

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

@ -23,19 +23,21 @@ typedef struct {
int32_t minRow; int32_t minRow;
int32_t maxRow; int32_t maxRow;
// -------------- // --------------
TSKEY nextKey; // need to be reset by each table commit TSKEY nextKey; // reset by each table commit
int32_t commitFid; int32_t commitFid;
TSKEY minKey; TSKEY minKey;
TSKEY maxKey; TSKEY maxKey;
// commit file data // commit file data
SDataFReader *pReader; SDataFReader *pReader;
SMapData oBlockIdx; // SMapData<SBlockIdx>, read from reader SMapData oBlockIdxMap; // SMapData<SBlockIdx>, read from reader
SMapData oBlock; // SMapData<SBlock>, read from reader SMapData oBlockMap; // SMapData<SBlock>, read from reader
SBlockData bDataO; SBlock oBlock;
SBlockData oBlockData;
SDataFWriter *pWriter; SDataFWriter *pWriter;
SMapData nBlockIdx; // SMapData<SBlockIdx>, build by committer SMapData nBlockIdxMap; // SMapData<SBlockIdx>, build by committer
SMapData nBlock; // SMapData<SBlock> SMapData nBlockMap; // SMapData<SBlock>
SBlockData bDataN; SBlock nBlock;
SBlockData nBlockData;
/* commit del */ /* commit del */
SDelFReader *pDelFReader; SDelFReader *pDelFReader;
SMapData oDelIdxMap; // SMapData<SDelIdx>, old SMapData oDelIdxMap; // SMapData<SDelIdx>, old
@ -55,13 +57,12 @@ int32_t tsdbBegin(STsdb *pTsdb) {
int32_t code = 0; int32_t code = 0;
code = tsdbMemTableCreate(pTsdb, &pTsdb->mem); code = tsdbMemTableCreate(pTsdb, &pTsdb->mem);
if (code) { if (code) goto _err;
goto _err;
}
return code; return code;
_err: _err:
tsdbError("vgId:%d tsdb begin failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
@ -73,7 +74,8 @@ int32_t tsdbCommit(STsdb *pTsdb) {
SMemTable *pMemTable = pTsdb->mem; SMemTable *pMemTable = pTsdb->mem;
// check // check
if (pMemTable->nRow == 0 && pMemTable->nDel == 0) { // TODO if (pMemTable->nRow == 0 && pMemTable->nDel == 0) {
// TODO: lock?
pTsdb->mem = NULL; pTsdb->mem = NULL;
tsdbMemTableDestroy(pMemTable); tsdbMemTableDestroy(pMemTable);
goto _exit; goto _exit;
@ -81,31 +83,21 @@ int32_t tsdbCommit(STsdb *pTsdb) {
// start commit // start commit
code = tsdbStartCommit(pTsdb, &commith); code = tsdbStartCommit(pTsdb, &commith);
if (code) { if (code) goto _err;
goto _err;
}
// commit impl // commit impl
code = tsdbCommitData(&commith); code = tsdbCommitData(&commith);
if (code) { if (code) goto _err;
goto _err;
}
code = tsdbCommitDel(&commith); code = tsdbCommitDel(&commith);
if (code) { if (code) goto _err;
goto _err;
}
code = tsdbCommitCache(&commith); code = tsdbCommitCache(&commith);
if (code) { if (code) goto _err;
goto _err;
}
// end commit // end commit
code = tsdbEndCommit(&commith, 0); code = tsdbEndCommit(&commith, 0);
if (code) { if (code) goto _err;
goto _err;
}
_exit: _exit:
return code; return code;
@ -355,35 +347,35 @@ static int32_t tsdbCommitMemoryData(SCommitter *pCommitter, SBlockIdx *pBlockIdx
pRow = tsdbTbDataIterGet(pIter); pRow = tsdbTbDataIterGet(pIter);
if (pRow == NULL || tsdbKeyCmprFn(&(TSDBKEY){.ts = pRow->pTSRow->ts, .version = pRow->version}, &eKey) > 0) { if (pRow == NULL || tsdbKeyCmprFn(&(TSDBKEY){.ts = pRow->pTSRow->ts, .version = pRow->version}, &eKey) > 0) {
if (pCommitter->bDataN.nRow == 0) { if (pCommitter->nBlockData.nRow == 0) {
break; break;
} else { } else {
goto _write_block_data; goto _write_block_data;
} }
} }
code = tsdbBlockDataAppendRow(&pCommitter->bDataN, pRow, NULL /*TODO*/); code = tBlockDataAppendRow(&pCommitter->nBlockData, pRow, NULL /*TODO*/);
if (code) goto _err; if (code) goto _err;
if (pCommitter->bDataN.nRow < pCommitter->maxRow * 4 / 5) { if (pCommitter->nBlockData.nRow < pCommitter->maxRow * 4 / 5) {
continue; continue;
} }
_write_block_data: _write_block_data:
if (!toDataOnly && pCommitter->bDataN.nRow < pCommitter->minKey) { if (!toDataOnly && pCommitter->nBlockData.nRow < pCommitter->minKey) {
block.last = 1; block.last = 1;
} else { } else {
block.last = 0; block.last = 0;
} }
code = tsdbWriteBlockData(pCommitter->pWriter, &pCommitter->bDataN, NULL, NULL, pBlockIdx, &block); code = tsdbWriteBlockData(pCommitter->pWriter, &pCommitter->nBlockData, NULL, NULL, pBlockIdx, &block);
if (code) goto _err; if (code) goto _err;
code = tMapDataPutItem(&pCommitter->nBlock, &block, tPutBlock); code = tMapDataPutItem(&pCommitter->nBlockMap, &block, tPutBlock);
if (code) goto _err; if (code) goto _err;
tBlockReset(&block); tBlockReset(&block);
tsdbBlockDataClear(&pCommitter->bDataN); tBlockDataReset(&pCommitter->nBlockData);
} }
return code; return code;
@ -431,19 +423,19 @@ static int32_t tsdbMergeCommitImpl(SCommitter *pCommitter, SBlockIdx *pBlockIdx,
TSDBKEY key1; TSDBKEY key1;
TSDBKEY key2; TSDBKEY key2;
tsdbBlockDataClear(&pCommitter->bDataN); tBlockDataReset(&pCommitter->nBlockData);
// load last and merge until {pCommitter->maxKey, INT64_MAX} // load last and merge until {pCommitter->maxKey, INT64_MAX}
code = tsdbReadBlockData(pCommitter->pReader, pBlockIdx, pBlock, &pCommitter->bDataO, NULL, 0, NULL, NULL); code = tsdbReadBlockData(pCommitter->pReader, pBlockIdx, pBlock, &pCommitter->oBlockData, NULL, 0, NULL, NULL);
if (code) goto _err; if (code) goto _err;
iRow = 0; iRow = 0;
nRow = pCommitter->bDataO.nRow; nRow = pCommitter->oBlockData.nRow;
pRow = tsdbTbDataIterGet(pIter); pRow = tsdbTbDataIterGet(pIter);
while (true) { while (true) {
if ((pRow == NULL || pRow->pTSRow->ts > pCommitter->maxKey) && (iRow >= nRow)) { if ((pRow == NULL || pRow->pTSRow->ts > pCommitter->maxKey) && (iRow >= nRow)) {
if (pCommitter->bDataN.nRow > 0) { if (pCommitter->nBlockData.nRow > 0) {
goto _write_block_data; goto _write_block_data;
} else { } else {
break; break;
@ -453,16 +445,16 @@ static int32_t tsdbMergeCommitImpl(SCommitter *pCommitter, SBlockIdx *pBlockIdx,
// TODO // TODO
_write_block_data: _write_block_data:
block.last = pCommitter->bDataN.nRow < pCommitter->minRow ? 1 : 0; block.last = pCommitter->nBlockData.nRow < pCommitter->minRow ? 1 : 0;
code = tsdbWriteBlockData(pCommitter->pWriter, &pCommitter->bDataN, NULL, NULL, pBlockIdx, &block); code = tsdbWriteBlockData(pCommitter->pWriter, &pCommitter->nBlockData, NULL, NULL, pBlockIdx, &block);
if (code) goto _err; if (code) goto _err;
code = tMapDataPutItem(&pCommitter->nBlock, &block, tPutBlock); code = tMapDataPutItem(&pCommitter->nBlockMap, &block, tPutBlock);
if (code) goto _err; if (code) goto _err;
} }
tBlockReset(&block); tBlockReset(&block);
tsdbBlockDataClear(&pCommitter->bDataN); tBlockDataReset(&pCommitter->nBlockData);
return code; return code;
@ -475,7 +467,7 @@ static int32_t tsdbMergeCommit(SCommitter *pCommitter, SBlockIdx *pBlockIdx, STb
int32_t code = 0; int32_t code = 0;
TSDBROW *pRow; TSDBROW *pRow;
SBlock block = tBlockInit(); SBlock block = tBlockInit();
SBlockData bDataN; SBlockData nBlockData;
TSDBKEY key; TSDBKEY key;
int32_t c; int32_t c;
@ -503,7 +495,7 @@ static int32_t tsdbMergeCommit(SCommitter *pCommitter, SBlockIdx *pBlockIdx, STb
c = tBlockCmprFn(&(SBlock){.info.maxKey = key, .info.minKey = key}, pBlock); c = tBlockCmprFn(&(SBlock){.info.maxKey = key, .info.minKey = key}, pBlock);
if (c > 0) { if (c > 0) {
// move block // move block
code = tMapDataPutItem(&pCommitter->nBlock, pBlock, tPutBlock); code = tMapDataPutItem(&pCommitter->nBlockMap, pBlock, tPutBlock);
if (code) goto _err; if (code) goto _err;
} else if (c == 0) { } else if (c == 0) {
int32_t nOverlap = tsdbGetOverlapRowNumber(pIter, pBlock); int32_t nOverlap = tsdbGetOverlapRowNumber(pIter, pBlock);
@ -545,10 +537,10 @@ static int32_t tsdbCommitTableData(SCommitter *pCommitter, STbData *pTbData, SBl
if (ROW_END(pRow, pCommitter->maxKey) && pBlockIdx == NULL) goto _exit; if (ROW_END(pRow, pCommitter->maxKey) && pBlockIdx == NULL) goto _exit;
// start ================================ // start ================================
tMapDataReset(&pCommitter->oBlock); tMapDataReset(&pCommitter->oBlockMap);
tMapDataReset(&pCommitter->nBlock); tMapDataReset(&pCommitter->nBlockMap);
if (pBlockIdx) { if (pBlockIdx) {
code = tsdbReadBlock(pCommitter->pReader, pBlockIdx, &pCommitter->oBlock, NULL); code = tsdbReadBlock(pCommitter->pReader, pBlockIdx, &pCommitter->oBlockMap, NULL);
if (code) goto _err; if (code) goto _err;
} }
@ -556,12 +548,12 @@ static int32_t tsdbCommitTableData(SCommitter *pCommitter, STbData *pTbData, SBl
SBlock block; SBlock block;
SBlock *pBlock = &block; SBlock *pBlock = &block;
int32_t iBlock = 0; int32_t iBlock = 0;
int32_t nBlock = pCommitter->oBlock.nItem; int32_t nBlockMap = pCommitter->oBlockMap.nItem;
// merge // merge
pRow = tsdbTbDataIterGet(pIter); pRow = tsdbTbDataIterGet(pIter);
while (!ROW_END(pRow, pCommitter->maxKey) && iBlock < nBlock) { while (!ROW_END(pRow, pCommitter->maxKey) && iBlock < nBlockMap) {
tMapDataGetItemByIdx(&pCommitter->oBlock, iBlock, pBlock, tGetBlock); tMapDataGetItemByIdx(&pCommitter->oBlockMap, iBlock, pBlock, tGetBlock);
code = tsdbMergeCommit(pCommitter, &blockIdx, pIter, pBlock); code = tsdbMergeCommit(pCommitter, &blockIdx, pIter, pBlock);
if (code) goto _err; if (code) goto _err;
@ -579,8 +571,8 @@ static int32_t tsdbCommitTableData(SCommitter *pCommitter, STbData *pTbData, SBl
} }
// disk // disk
while (iBlock < nBlock) { while (iBlock < nBlockMap) {
tMapDataGetItemByIdx(&pCommitter->oBlock, iBlock, pBlock, tGetBlock); tMapDataGetItemByIdx(&pCommitter->oBlockMap, iBlock, pBlock, tGetBlock);
code = tsdbMergeCommit(pCommitter, &blockIdx, NULL, pBlock); code = tsdbMergeCommit(pCommitter, &blockIdx, NULL, pBlock);
if (code) goto _err; if (code) goto _err;
@ -589,10 +581,10 @@ static int32_t tsdbCommitTableData(SCommitter *pCommitter, STbData *pTbData, SBl
} }
// end =============================== // end ===============================
code = tsdbWriteBlock(pCommitter->pWriter, &pCommitter->nBlock, NULL, &blockIdx); code = tsdbWriteBlock(pCommitter->pWriter, &pCommitter->nBlockMap, NULL, &blockIdx);
if (code) goto _err; if (code) goto _err;
code = tMapDataPutItem(&pCommitter->nBlockIdx, &blockIdx, tPutBlockIdx); code = tMapDataPutItem(&pCommitter->nBlockIdxMap, &blockIdx, tPutBlockIdx);
if (code) goto _err; if (code) goto _err;
_exit: _exit:
@ -619,21 +611,25 @@ static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) {
// memory // memory
pCommitter->nextKey = TSKEY_MAX; pCommitter->nextKey = TSKEY_MAX;
tMapDataReset(&pCommitter->oBlockIdx);
tMapDataReset(&pCommitter->oBlock);
tMapDataReset(&pCommitter->nBlockIdx);
tMapDataReset(&pCommitter->nBlock);
// load old // old
tMapDataReset(&pCommitter->oBlockIdxMap);
tMapDataReset(&pCommitter->oBlockMap);
tBlockReset(&pCommitter->oBlock);
tBlockDataReset(&pCommitter->oBlockData);
if (pRSet) { if (pRSet) {
code = tsdbDataFReaderOpen(&pCommitter->pReader, pTsdb, pRSet); code = tsdbDataFReaderOpen(&pCommitter->pReader, pTsdb, pRSet);
if (code) goto _err; if (code) goto _err;
code = tsdbReadBlockIdx(pCommitter->pReader, &pCommitter->oBlockIdx, NULL); code = tsdbReadBlockIdx(pCommitter->pReader, &pCommitter->oBlockIdxMap, NULL);
if (code) goto _err; if (code) goto _err;
} }
// create new // new
tMapDataReset(&pCommitter->nBlockIdxMap);
tMapDataReset(&pCommitter->nBlockMap);
tBlockReset(&pCommitter->nBlock);
tBlockDataReset(&pCommitter->nBlockData);
code = tsdbDataFWriterOpen(&pCommitter->pWriter, pTsdb, pWSet); code = tsdbDataFWriterOpen(&pCommitter->pWriter, pTsdb, pWSet);
if (code) goto _err; if (code) goto _err;
@ -653,18 +649,16 @@ static int32_t tsdbCommitFileDataImpl(SCommitter *pCommitter) {
int32_t iTbData = 0; int32_t iTbData = 0;
int32_t nTbData = taosArrayGetSize(pMemTable->aTbData); int32_t nTbData = taosArrayGetSize(pMemTable->aTbData);
int32_t iBlockIdx = 0; int32_t iBlockIdx = 0;
int32_t nBlockIdx = pCommitter->oBlockIdx.nItem; int32_t nBlockIdx = pCommitter->oBlockIdxMap.nItem;
STbData *pTbData; STbData *pTbData;
SBlockIdx *pBlockIdx;
SBlockIdx blockIdx; SBlockIdx blockIdx;
SBlockIdx *pBlockIdx = &blockIdx;
ASSERT(nTbData > 0); ASSERT(nTbData > 0);
pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData); pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData);
if (iBlockIdx < nBlockIdx) { if (iBlockIdx < nBlockIdx) {
pBlockIdx = &blockIdx; tMapDataGetItemByIdx(&pCommitter->oBlockIdxMap, iBlockIdx, pBlockIdx, tGetBlockIdx);
code = tMapDataGetItemByIdx(&pCommitter->oBlockIdx, iBlockIdx, pBlockIdx, tGetBlockIdx);
if (code) goto _err;
} else { } else {
pBlockIdx = NULL; pBlockIdx = NULL;
} }
@ -706,9 +700,7 @@ static int32_t tsdbCommitFileDataImpl(SCommitter *pCommitter) {
iBlockIdx++; iBlockIdx++;
if (iBlockIdx < nBlockIdx) { if (iBlockIdx < nBlockIdx) {
pBlockIdx = &blockIdx; tMapDataGetItemByIdx(&pCommitter->oBlockIdxMap, iBlockIdx, pBlockIdx, tGetBlockIdx);
code = tMapDataGetItemByIdx(&pCommitter->oBlockIdx, iBlockIdx, pBlockIdx, tGetBlockIdx);
if (code) goto _err;
} else { } else {
pBlockIdx = NULL; pBlockIdx = NULL;
} }
@ -726,9 +718,7 @@ static int32_t tsdbCommitFileDataImpl(SCommitter *pCommitter) {
pTbData = NULL; pTbData = NULL;
} }
if (iBlockIdx < nBlockIdx) { if (iBlockIdx < nBlockIdx) {
pBlockIdx = &blockIdx; tMapDataGetItemByIdx(&pCommitter->oBlockIdxMap, iBlockIdx, pBlockIdx, tGetBlockIdx);
code = tMapDataGetItemByIdx(&pCommitter->oBlockIdx, iBlockIdx, pBlockIdx, tGetBlockIdx);
if (code) goto _err;
} else { } else {
pBlockIdx = NULL; pBlockIdx = NULL;
} }
@ -746,7 +736,7 @@ static int32_t tsdbCommitFileDataEnd(SCommitter *pCommitter) {
int32_t code = 0; int32_t code = 0;
// write blockIdx // write blockIdx
code = tsdbWriteBlockIdx(pCommitter->pWriter, &pCommitter->nBlockIdx, NULL); code = tsdbWriteBlockIdx(pCommitter->pWriter, &pCommitter->nBlockIdxMap, NULL);
if (code) goto _err; if (code) goto _err;
// update file header // update file header
@ -775,25 +765,20 @@ static int32_t tsdbCommitFileData(SCommitter *pCommitter) {
// commit file data start // commit file data start
code = tsdbCommitFileDataStart(pCommitter); code = tsdbCommitFileDataStart(pCommitter);
if (code) { if (code) goto _err;
goto _err;
}
// commit file data impl // commit file data impl
code = tsdbCommitFileDataImpl(pCommitter); code = tsdbCommitFileDataImpl(pCommitter);
if (code) { if (code) goto _err;
goto _err;
}
// commit file data end // commit file data end
code = tsdbCommitFileDataEnd(pCommitter); code = tsdbCommitFileDataEnd(pCommitter);
if (code) { if (code) goto _err;
goto _err;
}
return code; return code;
_err: _err:
tsdbError("vgId:%d commit file data failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
@ -809,6 +794,10 @@ static int32_t tsdbStartCommit(STsdb *pTsdb, SCommitter *pCommitter) {
// unlock(); // unlock();
pCommitter->pTsdb = pTsdb; pCommitter->pTsdb = pTsdb;
pCommitter->minutes = pTsdb->keepCfg.days;
pCommitter->precision = pTsdb->keepCfg.precision;
pCommitter->minRow = pTsdb->pVnode->config.tsdbCfg.minRows;
pCommitter->maxRow = pTsdb->pVnode->config.tsdbCfg.maxRows;
return code; return code;
} }

28
source/dnode/vnode/src/tsdb/tsdbFile.c
View File

@ -19,34 +19,6 @@ static const char *tsdbFileSuffix[] = {".tombstone", ".cache", ".index", ".data"
// .tombstone // .tombstone
struct STsdbIndexFile {
int64_t size;
int64_t offset;
int32_t nRef;
};
struct STsdbDataFile {
int64_t size;
int32_t nRef;
};
struct STsdbLastFile {
int64_t size;
int32_t nRef;
};
struct STsdbSmaFile {
int64_t size;
int32_t nRef;
};
struct SDFileSet {
STsdbIndexFile *pIndexF;
STsdbDataFile *pDataF;
STsdbLastFile *pLastF;
STsdbSmaFile *pSmaF;
};
// SDelFile =============================================== // SDelFile ===============================================
char *tsdbDelFileName(STsdb *pTsdb, SDelFile *pFile) { char *tsdbDelFileName(STsdb *pTsdb, SDelFile *pFile) {
char *pName = NULL; char *pName = NULL;

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

@ -493,7 +493,7 @@ static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, STbData *pTbData, i
SSubmitBlkIter blkIter = {0}; SSubmitBlkIter blkIter = {0};
TSDBKEY key = {.version = version}; TSDBKEY key = {.version = version};
SMemSkipListNode *pos[SL_MAX_LEVEL]; SMemSkipListNode *pos[SL_MAX_LEVEL];
TSDBROW row = {.version = version, .pTSRow = NULL}; TSDBROW row = tsdbRowFromTSRow(version, NULL);
int32_t nRow = 0; int32_t nRow = 0;
tInitSubmitBlkIter(pMsgIter, pBlock, &blkIter); tInitSubmitBlkIter(pMsgIter, pBlock, &blkIter);

16
source/dnode/vnode/src/tsdb/tsdbReaderWriter.c
View File

@ -441,15 +441,15 @@ _err:
return code; return code;
} }
int32_t tsdbReadBlockIdx(SDataFReader *pReader, SMapData *pMapData, uint8_t **ppBuf) { int32_t tsdbReadBlockIdx(SDataFReader *pReader, SMapData *mBlockIdx, uint8_t **ppBuf) {
int32_t code = 0; int32_t code = 0;
int64_t offset = -1; // TODO int64_t offset = pReader->pSet->pHeadFile->offset;
int64_t size = -1; // TODO int64_t size = pReader->pSet->pHeadFile->size;
int64_t n; int64_t n;
uint32_t delimiter; uint32_t delimiter;
// alloc // alloc
if (!ppBuf) ppBuf = &pMapData->pBuf; if (!ppBuf) ppBuf = &mBlockIdx->pBuf;
code = tsdbRealloc(ppBuf, size); code = tsdbRealloc(ppBuf, size);
if (code) goto _err; if (code) goto _err;
@ -479,7 +479,7 @@ int32_t tsdbReadBlockIdx(SDataFReader *pReader, SMapData *pMapData, uint8_t **pp
n = 0; n = 0;
n += tGetU32(*ppBuf + n, &delimiter); n += tGetU32(*ppBuf + n, &delimiter);
ASSERT(delimiter == TSDB_FILE_DLMT); ASSERT(delimiter == TSDB_FILE_DLMT);
n += tGetMapData(*ppBuf + n, pMapData); n += tGetMapData(*ppBuf + n, mBlockIdx);
ASSERT(n + sizeof(TSCKSUM) == size); ASSERT(n + sizeof(TSCKSUM) == size);
return code; return code;
@ -489,7 +489,7 @@ _err:
return code; return code;
} }
int32_t tsdbReadBlock(SDataFReader *pReader, SBlockIdx *pBlockIdx, SMapData *pMapData, uint8_t **ppBuf) { int32_t tsdbReadBlock(SDataFReader *pReader, SBlockIdx *pBlockIdx, SMapData *mBlockIdx, uint8_t **ppBuf) {
int32_t code = 0; int32_t code = 0;
int64_t offset = pBlockIdx->offset; int64_t offset = pBlockIdx->offset;
int64_t size = pBlockIdx->size; int64_t size = pBlockIdx->size;
@ -499,7 +499,7 @@ int32_t tsdbReadBlock(SDataFReader *pReader, SBlockIdx *pBlockIdx, SMapData *pMa
tb_uid_t uid; tb_uid_t uid;
// alloc // alloc
if (!ppBuf) ppBuf = &pMapData->pBuf; if (!ppBuf) ppBuf = &mBlockIdx->pBuf;
code = tsdbRealloc(ppBuf, size); code = tsdbRealloc(ppBuf, size);
if (code) goto _err; if (code) goto _err;
@ -533,7 +533,7 @@ int32_t tsdbReadBlock(SDataFReader *pReader, SBlockIdx *pBlockIdx, SMapData *pMa
ASSERT(suid == pBlockIdx->suid); ASSERT(suid == pBlockIdx->suid);
n += tGetI64(*ppBuf + n, &uid); n += tGetI64(*ppBuf + n, &uid);
ASSERT(uid == pBlockIdx->uid); ASSERT(uid == pBlockIdx->uid);
n += tGetMapData(*ppBuf + n, pMapData); n += tGetMapData(*ppBuf + n, mBlockIdx);
ASSERT(n + sizeof(TSCKSUM) == size); ASSERT(n + sizeof(TSCKSUM) == size);
return code; return code;

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

@ -147,7 +147,7 @@ int32_t tGetMapData(uint8_t *p, SMapData *pMapData) {
default: default:
ASSERT(0); ASSERT(0);
} }
n += tGetBinary(p ? p + n : p, &pMapData->pData, &pMapData->nData); n += tGetBinary(p + n, &pMapData->pData, &pMapData->nData);
return n; return n;
} }
@ -697,14 +697,7 @@ static int32_t tsdbBlockDataAppendRow1(SBlockData *pBlockData, TSDBROW *pRow) {
return code; return code;
} }
void tsdbBlockDataClear(SBlockData *pBlockData) { int32_t tBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema) {
pBlockData->nRow = 0;
for (int32_t iCol = 0; iCol < pBlockData->nCol; iCol++) {
pBlockData->aColData[iCol] = (SColData){.cid = 0, .type = 0, .bytes = 0, .flags = 0, .nData = 0};
}
}
int32_t tsdbBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema) {
int32_t code = 0; int32_t code = 0;
if (pRow->type == 0) { if (pRow->type == 0) {
@ -716,7 +709,12 @@ int32_t tsdbBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *
return code; return code;
} }
void tsdbBlockDataDestroy(SBlockData *pBlockData) { void tBlockDataReset(SBlockData *pBlockData) {
pBlockData->nRow = 0;
pBlockData->nCol = 0;
}
void tBlockDataClear(SBlockData *pBlockData) {
tsdbFree((uint8_t *)pBlockData->aVersion); tsdbFree((uint8_t *)pBlockData->aVersion);
tsdbFree((uint8_t *)pBlockData->aTSKEY); tsdbFree((uint8_t *)pBlockData->aTSKEY);
for (int32_t iCol = 0; iCol < pBlockData->nCol; iCol++) { for (int32_t iCol = 0; iCol < pBlockData->nCol; iCol++) {