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-15 12:28:52 +00:00
parent 76a71b11ed
commit a85d522e20
2 changed files with 63 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -352,30 +352,78 @@ static FORCE_INLINE bool tsdbCommitIterEnd(SCommitter *pCommitter, STbDataIter *
} }
static int32_t tsdbCommitTableDataImpl(SCommitter *pCommitter, STbDataIter *pIter, SBlockIdx *pBlockIdx) { static int32_t tsdbCommitTableDataImpl(SCommitter *pCommitter, STbDataIter *pIter, SBlockIdx *pBlockIdx) {
int32_t code = 0; int32_t code = 0;
TSDBROW *pRow; int32_t c;
int32_t iBlock; int32_t iBlock;
int32_t nBlock; int32_t nBlock;
SBlock *pBlock; SBlock *pBlock;
SBlock block; SBlock block;
SBlockIdx blockIdx; // todo
// start(todo) // start ================================
tMapDataReset(&pCommitter->oBlock);
tMapDataReset(&pCommitter->nBlock);
if (pBlockIdx) {
code = tsdbReadBlock(pCommitter->pReader, pBlockIdx, &pCommitter->oBlock, NULL);
if (code) goto _err;
}
// impl ===============================
iBlock = 0;
nBlock = pCommitter->oBlock.nItem;
// impl
pRow = tsdbTbDataIterGet(pIter);
if (iBlock < nBlock) { if (iBlock < nBlock) {
pBlock = &block; pBlock = &block;
tMapDataGetItemByIdx(&pCommitter->oBlock, iBlock, pBlock, tGetBlock); tMapDataGetItemByIdx(&pCommitter->nBlock, iBlock, pBlock, tGetBlock);
} else {
pBlock = NULL;
} }
while (true) { while (true) {
if ((pRow == NULL || pRow->pTSRow->ts > pCommitter->maxKey) && pBlock == NULL) break; TSDBROW *pRow = tsdbTbDataIterGet(pIter);
bool iterEnd = ((pRow == NULL) || (pRow->pTSRow->ts > pCommitter->maxKey));
bool blockEnd = (pBlock == NULL);
if (iterEnd && blockEnd) break;
if (!iterEnd && !blockEnd) {
c = tBlockCmprFn(&(SBlock){.maxKey.ts = pRow->pTSRow->ts}, pBlock);
if (c == 0) {
// merge until pBlock->maxKey
// if (pBlock->last), merge until pCommitter->maxKey
// else merge until pBlock->maxKey
} else if (c < 0) {
// tsdbCommitTableMemData(pIter, pBlock);
// if (pBlock->last), merge until pCommitter->maxKey
// else, commit until pBlock->minKey-1
} else {
// tsdbCommitTableDiskData(pBlock);
// if (pBlock->last), merge until pCommitter->maxKey
// else, move the block to new one
}
} else if (!iterEnd) {
// no block on disk, commit to last when there are no enough data
// commit memory data to pCommitter->maxKey
// tsdbCommitTableMemData(pIter, NULL);
} else {
// tsdbCommitTableDiskData(pBlock); // only left block
// if (last block ? ) else ?
// tMapDataPutItem(&pCommitter->nBlock, pBlock, tPutBlock);
iBlock++; // get new SBlock
}
} }
// end // end ===============================
code = tsdbWriteBlock(pCommitter->pWriter, &pCommitter->nBlock, NULL, &blockIdx);
if (code) goto _err;
code = tMapDataPutItem(&pCommitter->nBlockIdx, &blockIdx, tPutBlockIdx);
if (code) goto _err;
return code; return code;
_err: _err:
tsdbError("vgId:%d commit table data impl failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }

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

@ -669,8 +669,10 @@ _err:
return code; return code;
} }
int32_t tsdbWriteBlock(SDataFWriter *pWriter, SMapData *pMapData, uint8_t **ppBuf, SBlockIdx *pBlockIdx) { int32_t tsdbWriteBlock(SDataFWriter *pWriter, SMapData *pBlockMap, uint8_t **ppBuf, SBlockIdx *pBlockIdx) {
int32_t code = 0; int32_t code = 0;
ASSERT(pBlockMap->nItem > 0);
// TODO // TODO
return code; return code;
} }