Merge pull request #23242 from taosdata/fix/liaohj
refactor: do some internal refactor.
This commit is contained in:
commit
bd72d080a9
|
@ -788,10 +788,10 @@ struct SDiskDataBuilder {
|
|||
SBlkInfo bi;
|
||||
};
|
||||
|
||||
typedef struct SLDataIter {
|
||||
struct SLDataIter {
|
||||
SRBTreeNode node;
|
||||
SSttBlk *pSttBlk;
|
||||
int32_t iStt; // for debug purpose
|
||||
int64_t cid; // for debug purpose
|
||||
int8_t backward;
|
||||
int32_t iSttBlk;
|
||||
int32_t iRow;
|
||||
|
@ -802,7 +802,7 @@ typedef struct SLDataIter {
|
|||
SSttBlockLoadInfo *pBlockLoadInfo;
|
||||
bool ignoreEarlierTs;
|
||||
struct SSttFileReader *pReader;
|
||||
} SLDataIter;
|
||||
};
|
||||
|
||||
#define tMergeTreeGetRow(_t) (&((_t)->pIter->rInfo.row))
|
||||
int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFReader, uint64_t suid, uint64_t uid,
|
||||
|
|
|
@ -115,8 +115,9 @@ static SBlockData *loadLastBlock(SLDataIter *pIter, const char *idStr) {
|
|||
SSttBlockLoadInfo *pInfo = pIter->pBlockLoadInfo;
|
||||
if (pInfo->blockIndex[0] == pIter->iSttBlk) {
|
||||
if (pInfo->currentLoadBlockIndex != 0) {
|
||||
tsdbDebug("current load index is set to 0, block index:%d, file index:%d, due to uid:%" PRIu64 ", load data, %s",
|
||||
pIter->iSttBlk, pIter->iStt, pIter->uid, idStr);
|
||||
tsdbDebug("current load index is set to 0, block index:%d, fileVer:%" PRId64 ", due to uid:%" PRIu64
|
||||
", load data, %s",
|
||||
pIter->iSttBlk, pIter->cid, pIter->uid, idStr);
|
||||
pInfo->currentLoadBlockIndex = 0;
|
||||
}
|
||||
return &pInfo->blockData[0];
|
||||
|
@ -124,8 +125,9 @@ static SBlockData *loadLastBlock(SLDataIter *pIter, const char *idStr) {
|
|||
|
||||
if (pInfo->blockIndex[1] == pIter->iSttBlk) {
|
||||
if (pInfo->currentLoadBlockIndex != 1) {
|
||||
tsdbDebug("current load index is set to 1, block index:%d, file index:%d, due to uid:%" PRIu64 ", load data, %s",
|
||||
pIter->iSttBlk, pIter->iStt, pIter->uid, idStr);
|
||||
tsdbDebug("current load index is set to 1, block index:%d, fileVer:%" PRId64 ", due to uid:%" PRIu64
|
||||
", load data, %s",
|
||||
pIter->iSttBlk, pIter->cid, pIter->uid, idStr);
|
||||
pInfo->currentLoadBlockIndex = 1;
|
||||
}
|
||||
return &pInfo->blockData[1];
|
||||
|
@ -150,10 +152,12 @@ static SBlockData *loadLastBlock(SLDataIter *pIter, const char *idStr) {
|
|||
pInfo->cost.blockElapsedTime += el;
|
||||
pInfo->cost.loadBlocks += 1;
|
||||
|
||||
tsdbDebug("read last block, total load:%"PRId64", trigger by uid:%" PRIu64
|
||||
", last file index:%d, last block index:%d, entry:%d, rows:%d, %p, elapsed time:%.2f ms, %s",
|
||||
pInfo->cost.loadBlocks, pIter->uid, pIter->iStt, pIter->iSttBlk, pInfo->currentLoadBlockIndex, pBlock->nRow,
|
||||
pBlock, el, idStr);
|
||||
tsdbDebug("read last block, total load:%" PRId64 ", trigger by uid:%" PRIu64 ", stt-fileVer:%" PRId64
|
||||
", last block index:%d, entry:%d, rows:%d, uidRange:%" PRId64 "-%" PRId64 " tsRange:%" PRId64 "-%" PRId64
|
||||
" %p, elapsed time:%.2f ms, %s",
|
||||
pInfo->cost.loadBlocks, pIter->uid, pIter->cid, pIter->iSttBlk, pInfo->currentLoadBlockIndex, pBlock->nRow,
|
||||
pIter->pSttBlk->minUid, pIter->pSttBlk->maxUid, pIter->pSttBlk->minKey, pIter->pSttBlk->maxKey, pBlock, el,
|
||||
idStr);
|
||||
|
||||
pInfo->blockIndex[pInfo->currentLoadBlockIndex] = pIter->iSttBlk;
|
||||
pIter->iRow = (pIter->backward) ? pInfo->blockData[pInfo->currentLoadBlockIndex].nRow : -1;
|
||||
|
@ -424,14 +428,14 @@ static int32_t doLoadSttFilesBlk(SSttBlockLoadInfo *pBlockLoadInfo, SLDataIter *
|
|||
return code;
|
||||
}
|
||||
|
||||
int32_t tLDataIterOpen2(SLDataIter *pIter, SSttFileReader *pSttFileReader, int32_t iStt, int8_t backward,
|
||||
int32_t tLDataIterOpen2(SLDataIter *pIter, SSttFileReader *pSttFileReader, int32_t cid, int8_t backward,
|
||||
uint64_t suid, uint64_t uid, STimeWindow *pTimeWindow, SVersionRange *pRange,
|
||||
SSttBlockLoadInfo *pBlockLoadInfo, const char *idStr, bool strictTimeRange,
|
||||
_load_tomb_fn loadTombFn, void *pReader1) {
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
|
||||
pIter->uid = uid;
|
||||
pIter->iStt = iStt;
|
||||
pIter->cid = cid;
|
||||
pIter->backward = backward;
|
||||
pIter->verRange.minVer = pRange->minVer;
|
||||
pIter->verRange.maxVer = pRange->maxVer;
|
||||
|
@ -538,12 +542,15 @@ void tLDataIterNextBlock(SLDataIter *pIter, const char *idStr) {
|
|||
|
||||
pIter->pSttBlk = NULL;
|
||||
if (index != -1) {
|
||||
SSttBlk *p = taosArrayGet(pIter->pBlockLoadInfo->aSttBlk, index);
|
||||
|
||||
pIter->iSttBlk = index;
|
||||
pIter->pSttBlk = (SSttBlk *)taosArrayGet(pIter->pBlockLoadInfo->aSttBlk, pIter->iSttBlk);
|
||||
tsdbDebug("try next last file block:%d from stt fileIdx:%d, trigger by uid:%" PRIu64 ", file index:%d, %s",
|
||||
pIter->iSttBlk, oldIndex, pIter->uid, pIter->iStt, idStr);
|
||||
tsdbDebug("try next stt-file block:%d from %d, trigger by uid:%" PRIu64 ", stt-fileVer:%" PRId64
|
||||
", uidRange:%" PRId64 "-%" PRId64 " %s",
|
||||
pIter->iSttBlk, oldIndex, pIter->uid, pIter->cid, p->minUid, p->maxUid, idStr);
|
||||
} else {
|
||||
tsdbDebug("no more last block qualified, uid:%" PRIu64 ", file index:%d, %s", pIter->uid, oldIndex, idStr);
|
||||
tsdbDebug("no more last block qualified, uid:%" PRIu64 ", stt-file block:%d, %s", pIter->uid, oldIndex, idStr);
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -787,7 +794,9 @@ int32_t tMergeTreeOpen2(SMergeTree *pMTree, SMergeTreeConf *pConf) {
|
|||
}
|
||||
|
||||
memset(pIter, 0, sizeof(SLDataIter));
|
||||
code = tLDataIterOpen2(pIter, pSttFileReader, i, pMTree->backward, pConf->suid, pConf->uid, &pConf->timewindow,
|
||||
|
||||
int64_t cid = pSttLevel->fobjArr->data[i]->f->cid;
|
||||
code = tLDataIterOpen2(pIter, pSttFileReader, cid, pMTree->backward, pConf->suid, pConf->uid, &pConf->timewindow,
|
||||
&pConf->verRange, pLoadInfo, pMTree->idStr, pConf->strictTimeRange, pConf->loadTombFn,
|
||||
pConf->pReader);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
|
|
|
@ -524,7 +524,7 @@ static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SBlockN
|
|||
int64_t st = taosGetTimestampUs();
|
||||
|
||||
// clear info for the new file
|
||||
cleanupInfoFoxNextFileset(pReader->status.pTableMap);
|
||||
cleanupInfoForNextFileset(pReader->status.pTableMap);
|
||||
|
||||
int32_t k = 0;
|
||||
int32_t numOfTables = tSimpleHashGetSize(pReader->status.pTableMap);
|
||||
|
@ -1424,7 +1424,6 @@ static bool tryCopyDistinctRowFromSttBlock(TSDBROW* fRow, SLastBlockReader* pLas
|
|||
STableBlockScanInfo* pScanInfo, int64_t ts, STsdbReader* pReader,
|
||||
bool* copied) {
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
|
||||
*copied = false;
|
||||
|
||||
bool hasVal = nextRowFromLastBlocks(pLastBlockReader, pScanInfo, &pReader->info.verRange);
|
||||
|
@ -1507,12 +1506,6 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo*
|
|||
if (ps == NULL) {
|
||||
return terrno;
|
||||
}
|
||||
|
||||
int32_t code = tsdbRowMergerInit(pMerger, ps);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
tsdbError("failed to init row merger, code:%s", tstrerror(code));
|
||||
return code;
|
||||
}
|
||||
}
|
||||
|
||||
int64_t minKey = 0;
|
||||
|
@ -1655,7 +1648,7 @@ static int32_t doMergeFileBlockAndLastBlock(SLastBlockReader* pLastBlockReader,
|
|||
tsdbTrace("fRow ptr:%p, %d, uid:%" PRIu64 ", ts:%" PRId64 " %s", pRow->pBlockData, pRow->iRow, pLastBlockReader->uid,
|
||||
fRow.pBlockData->aTSKEY[fRow.iRow], pReader->idStr);
|
||||
|
||||
// only last block exists
|
||||
// only stt block exists
|
||||
if ((!mergeBlockData) || (tsLastBlock != pBlockData->aTSKEY[pDumpInfo->rowIndex])) {
|
||||
code = tryCopyDistinctRowFromSttBlock(&fRow, pLastBlockReader, pBlockScanInfo, tsLastBlock, pReader, &copied);
|
||||
if (code) {
|
||||
|
@ -1734,12 +1727,6 @@ static int32_t mergeFileBlockAndLastBlock(STsdbReader* pReader, SLastBlockReader
|
|||
if (ps == NULL) {
|
||||
return terrno;
|
||||
}
|
||||
|
||||
int32_t code = tsdbRowMergerInit(pMerger, ps);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
tsdbError("failed to init row merger, code:%s", tstrerror(code));
|
||||
return code;
|
||||
}
|
||||
}
|
||||
|
||||
if (hasDataInFileBlock(pBlockData, pDumpInfo)) {
|
||||
|
@ -1841,12 +1828,6 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
|
|||
if (ps == NULL) {
|
||||
return terrno;
|
||||
}
|
||||
|
||||
code = tsdbRowMergerInit(pMerger, ps);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
tsdbError("failed to init row merger, code:%s", tstrerror(code));
|
||||
return code;
|
||||
}
|
||||
}
|
||||
|
||||
int64_t minKey = 0;
|
||||
|
@ -2169,12 +2150,6 @@ int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBloc
|
|||
if (ps == NULL) {
|
||||
return terrno;
|
||||
}
|
||||
|
||||
code = tsdbRowMergerInit(pMerger, ps);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
tsdbError("failed to init row merger, code:%s", tstrerror(code));
|
||||
return code;
|
||||
}
|
||||
}
|
||||
|
||||
if (copied) {
|
||||
|
|
|
@ -243,7 +243,7 @@ static void doCleanupInfoForNextFileset(STableBlockScanInfo* pScanInfo) {
|
|||
taosArrayClear(pScanInfo->pFileDelData); // del data from each file set
|
||||
}
|
||||
|
||||
void cleanupInfoFoxNextFileset(SSHashObj* pTableMap) {
|
||||
void cleanupInfoForNextFileset(SSHashObj* pTableMap) {
|
||||
STableBlockScanInfo** p = NULL;
|
||||
|
||||
int32_t iter = 0;
|
||||
|
|
|
@ -238,7 +238,7 @@ SSHashObj* createDataBlockScanInfo(STsdbReader* pTsdbReader, SBlockInfoBuf* pBuf
|
|||
void clearBlockScanInfo(STableBlockScanInfo* p);
|
||||
void destroyAllBlockScanInfo(SSHashObj* pTableMap);
|
||||
void resetAllDataBlockScanInfo(SSHashObj* pTableMap, int64_t ts, int32_t step);
|
||||
void cleanupInfoFoxNextFileset(SSHashObj* pTableMap);
|
||||
void cleanupInfoForNextFileset(SSHashObj* pTableMap);
|
||||
int32_t ensureBlockScanInfoBuf(SBlockInfoBuf* pBuf, int32_t numOfTables);
|
||||
void clearBlockScanInfoBuf(SBlockInfoBuf* pBuf);
|
||||
void* getPosInBlockInfoBuf(SBlockInfoBuf* pBuf, int32_t index);
|
||||
|
|
Loading…
Reference in New Issue