more work
This commit is contained in:
parent
80367eb841
commit
78774c333f
|
@ -851,6 +851,112 @@ _err:
|
|||
return code;
|
||||
}
|
||||
|
||||
static int32_t tsdbCommitTableData(SCommitter *pCommitter, STbData *pTbData, SBlockIdx *pBlockIdx) {
|
||||
int32_t code = 0;
|
||||
STbDataIter *pIter = &(STbDataIter){0};
|
||||
TSDBROW *pRow;
|
||||
int32_t iBlock = 0;
|
||||
int32_t nBlock;
|
||||
|
||||
if (pTbData) {
|
||||
tsdbTbDataIterOpen(pTbData, &(TSDBKEY){.ts = pCommitter->minKey, .version = VERSION_MIN}, 0, pIter);
|
||||
pRow = tsdbTbDataIterGet(pIter);
|
||||
} else {
|
||||
pIter = NULL;
|
||||
pRow = NULL;
|
||||
}
|
||||
|
||||
if (pBlockIdx) {
|
||||
code = tsdbReadBlock(pCommitter->pReader, pBlockIdx, &pCommitter->oBlockMap, NULL);
|
||||
if (code) goto _err;
|
||||
|
||||
nBlock = pCommitter->oBlockMap.nItem;
|
||||
ASSERT(nBlock > 0);
|
||||
} else {
|
||||
nBlock = 0;
|
||||
}
|
||||
|
||||
if ((pRow == NULL || TSDBROW_TS(pRow) > pCommitter->maxKey) && nBlock == 0) goto _exit;
|
||||
|
||||
// start ===========
|
||||
SBlock *pBlock = NULL; // (todo)
|
||||
int32_t c;
|
||||
|
||||
// merge ===========
|
||||
while (true) {
|
||||
if (((pRow == NULL) || TSDBROW_TS(pRow) > pCommitter->maxKey) && pBlock == NULL) break;
|
||||
|
||||
if (pRow && TSDBROW_TS(pRow) <= pCommitter->maxKey && pBlock) {
|
||||
if (pBlock->last) {
|
||||
code = tsdbMergeBlockAndMem(pCommitter, pIter, pBlockIdx, pBlock,
|
||||
(TSDBKEY){.ts = pCommitter->maxKey + 1, .version = VERSION_MIN}, 0);
|
||||
if (code) goto _err;
|
||||
|
||||
pRow = tsdbTbDataIterGet(pIter);
|
||||
iBlock++;
|
||||
} else {
|
||||
c = tBlockCmprFn(&(SBlock){}, pBlock);
|
||||
if (c > 0) {
|
||||
code = tsdbCommitMoveDiskBlock(pCommitter, pBlock);
|
||||
if (code) goto _err;
|
||||
|
||||
iBlock++;
|
||||
} else if (c < 0) {
|
||||
code = tsdbCommitMemoryDataImpl(pCommitter, pIter, pBlock->minKey, 1);
|
||||
if (code) goto _err;
|
||||
|
||||
pRow = tsdbTbDataIterGet(pIter);
|
||||
} else {
|
||||
int64_t nOvlp = 0; // (todo)
|
||||
if (pBlock->nRow + nOvlp <= pCommitter->maxRow && pBlock->nSubBlock < TSDB_MAX_SUBBLOCKS) {
|
||||
// add as a subblock
|
||||
} else {
|
||||
if (iBlock == nBlock - 1) {
|
||||
code = tsdbMergeBlockAndMem(pCommitter, pIter, pBlockIdx, pBlock,
|
||||
(TSDBKEY){.ts = pCommitter->maxKey + 1, .version = VERSION_MIN}, 0);
|
||||
|
||||
if (code) goto _err;
|
||||
} else {
|
||||
// code = tsdbMergeBlockAndMem(pCommitter, pIter, pBlockIdx, pBlock, pBlock[1].minKey, 1);
|
||||
if (code) goto _err;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} else if (pBlock) {
|
||||
code = tsdbCommitMoveDiskBlock(pCommitter, pBlock);
|
||||
if (code) goto _err;
|
||||
|
||||
// move to next block (todo)
|
||||
} else {
|
||||
code = tsdbCommitMemoryDataImpl(pCommitter, pIter,
|
||||
(TSDBKEY){.ts = pCommitter->maxKey + 1, .version = VERSION_MIN}, 0);
|
||||
if (code) goto _err;
|
||||
|
||||
pRow = tsdbTbDataIterGet(pIter);
|
||||
ASSERT(pRow == NULL || TSDBROW_TS(pRow) > pCommitter->maxKey);
|
||||
}
|
||||
}
|
||||
|
||||
// end
|
||||
// code = tsdbWriteBlock();
|
||||
if (code) goto _err;
|
||||
|
||||
code = tMapDataPutItem(&pCommitter->nBlockIdxMap, NULL, tPutBlockIdx);
|
||||
if (code) goto _err;
|
||||
|
||||
_exit:
|
||||
if (pIter) {
|
||||
pRow = tsdbTbDataIterGet(pIter);
|
||||
if (pRow) pCommitter->nextKey = TMIN(pCommitter->nextKey, TSDBROW_TS(pRow));
|
||||
}
|
||||
return code;
|
||||
|
||||
_err:
|
||||
tsdbError("vgId:%d tsdb commit table data failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
|
||||
return code;
|
||||
}
|
||||
|
||||
static int32_t tsdbCommitFileDataImpl(SCommitter *pCommitter) {
|
||||
int32_t code = 0;
|
||||
int32_t c;
|
||||
|
|
Loading…
Reference in New Issue