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

fix(tsdb): fix memory leak.

This commit is contained in:
Haojun Liao 2023-06-21 11:38:35 +08:00
parent a92d16a508
commit f6e3ad27f0
3 changed files with 116 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -800,7 +800,7 @@ int32_t tMergeTreeOpen2(SMergeTree *pMTree, int8_t backward, STsdb* pTsdb, uint6
bool destroyLoadInfo, const char *idStr, bool strictTimeRange, SLDataIter* pLDataIter, void* pCurrentFileSet); bool destroyLoadInfo, const char *idStr, bool strictTimeRange, SLDataIter* pLDataIter, void* pCurrentFileSet);
void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter); void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter);
bool tMergeTreeNext(SMergeTree *pMTree); bool tMergeTreeNext(SMergeTree *pMTree);
bool tMergeTreeIgnoreEarlierTs(SMergeTree *pMTree); bool tMergeTreeIgnoreEarlierTs(SMergeTree *pMTree);
void tMergeTreeClose(SMergeTree *pMTree); void tMergeTreeClose(SMergeTree *pMTree);
@ -809,6 +809,7 @@ SSttBlockLoadInfo *tCreateLastBlockLoadInfo(STSchema *pSchema, int16_t *colList,
void resetLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo); void resetLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo);
void getLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo, int64_t *blocks, double *el); void getLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo, int64_t *blocks, double *el);
void *destroyLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo); void *destroyLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo);
void destroySttBlockReader(SLDataIter* pLDataIter, int32_t numOfIter);
// tsdbCache ============================================================================================== // tsdbCache ==============================================================================================
typedef struct SCacheRowsReader { typedef struct SCacheRowsReader {

101
source/dnode/vnode/src/tsdb/tsdbMergeTree.c
View File

@ -17,6 +17,8 @@
#include "tsdbFSet2.h" #include "tsdbFSet2.h"
#include "tsdbSttFileRW.h" #include "tsdbSttFileRW.h"
static void tLDataIterClose2(SLDataIter *pIter);
// SLDataIter ================================================= // SLDataIter =================================================
SSttBlockLoadInfo *tCreateLastBlockLoadInfo(STSchema *pSchema, int16_t *colList, int32_t numOfCols, SSttBlockLoadInfo *tCreateLastBlockLoadInfo(STSchema *pSchema, int16_t *colList, int32_t numOfCols,
int32_t numOfSttTrigger) { int32_t numOfSttTrigger) {
@ -93,6 +95,12 @@ void *destroyLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo) {
return NULL; return NULL;
} }
void destroySttBlockReader(SLDataIter* pLDataIter, int32_t numOfIter) {
for(int32_t i = 0; i < numOfIter; ++i) {
tLDataIterClose2(&pLDataIter[i]);
}
}
static SBlockData *loadLastBlock(SLDataIter *pIter, const char *idStr) { static SBlockData *loadLastBlock(SLDataIter *pIter, const char *idStr) {
int32_t code = 0; int32_t code = 0;
@ -357,6 +365,7 @@ int32_t tLDataIterOpen2(struct SLDataIter *pIter, SSttFileReader *pReader, int32
void tLDataIterClose2(SLDataIter *pIter) { void tLDataIterClose2(SLDataIter *pIter) {
tsdbSttFileReaderClose(&pIter->pReader); tsdbSttFileReaderClose(&pIter->pReader);
pIter->pReader = NULL;
} }
void tLDataIterNextBlock(SLDataIter *pIter, const char *idStr) { void tLDataIterNextBlock(SLDataIter *pIter, const char *idStr) {
@ -623,64 +632,70 @@ _end:
return code; return code;
} }
int32_t tMergeTreeOpen2(SMergeTree *pMTree, int8_t backward, STsdb* pTsdb, uint64_t suid, uint64_t uid, int32_t tMergeTreeOpen2(SMergeTree *pMTree, int8_t backward, STsdb *pTsdb, uint64_t suid, uint64_t uid,
STimeWindow *pTimeWindow, SVersionRange *pVerRange, SSttBlockLoadInfo *pBlockLoadInfo, STimeWindow *pTimeWindow, SVersionRange *pVerRange, SSttBlockLoadInfo *pBlockLoadInfo,
bool destroyLoadInfo, const char *idStr, bool strictTimeRange, SLDataIter* pLDataIter, void* pCurrentFileSet) { bool destroyLoadInfo, const char *idStr, bool strictTimeRange, SLDataIter *pLDataIter,
int32_t code = TSDB_CODE_SUCCESS; void *pCurrentFileSet) {
int32_t code = TSDB_CODE_SUCCESS;
pMTree->backward = backward; pMTree->backward = backward;
pMTree->pIter = NULL; pMTree->pIter = NULL;
pMTree->idStr = idStr; pMTree->idStr = idStr;
if (!pMTree->backward) { // asc if (!pMTree->backward) { // asc
tRBTreeCreate(&pMTree->rbt, tLDataIterCmprFn); tRBTreeCreate(&pMTree->rbt, tLDataIterCmprFn);
} else { // desc } else { // desc
tRBTreeCreate(&pMTree->rbt, tLDataIterDescCmprFn); tRBTreeCreate(&pMTree->rbt, tLDataIterDescCmprFn);
} }
pMTree->pLoadInfo = pBlockLoadInfo; pMTree->pLoadInfo = pBlockLoadInfo;
pMTree->destroyLoadInfo = destroyLoadInfo; pMTree->destroyLoadInfo = destroyLoadInfo;
pMTree->ignoreEarlierTs = false; pMTree->ignoreEarlierTs = false;
// todo handle other level of stt files, here only deal with the first level stt // todo handle other level of stt files, here only deal with the first level stt
int32_t size = ((STFileSet*)pCurrentFileSet)->lvlArr[0].size; int32_t size = ((STFileSet *)pCurrentFileSet)->lvlArr[0].size;
if (size == 0) { if (size == 0) {
goto _end; goto _end;
} }
SSttLvl* pSttLevel = ((STFileSet*)pCurrentFileSet)->lvlArr[0].data[0]; SSttLvl *pSttLevel = ((STFileSet *)pCurrentFileSet)->lvlArr[0].data[0];
ASSERT(pSttLevel->level == 0); ASSERT(pSttLevel->level == 0);
for (int32_t i = 0; i < pSttLevel->fobjArr[0].size; ++i) { // open all last file for (int32_t i = 0; i < pSttLevel->fobjArr[0].size; ++i) { // open all last file
memset(&pLDataIter[i], 0, sizeof(SLDataIter)); SSttFileReader* pSttFileReader = pLDataIter[i].pReader;
memset(&pLDataIter[i], 0, sizeof(SLDataIter));
if (pSttFileReader == NULL) {
SSttFileReaderConfig conf = {.tsdb = pTsdb, .szPage = pTsdb->pVnode->config.szPage}; SSttFileReaderConfig conf = {.tsdb = pTsdb, .szPage = pTsdb->pVnode->config.szPage};
conf.file[0] = *pSttLevel->fobjArr[0].data[i]->f; conf.file[0] = *pSttLevel->fobjArr[0].data[i]->f;
SSttFileReader *pReader = NULL; code = tsdbSttFileReaderOpen(pSttLevel->fobjArr[0].data[i]->fname, &conf, &pSttFileReader);
code = tsdbSttFileReaderOpen(pSttLevel->fobjArr[0].data[i]->fname, &conf, &pReader);
code = tLDataIterOpen2(&pLDataIter[i], pReader, i, pMTree->backward, suid, uid, pTimeWindow, pVerRange,
&pMTree->pLoadInfo[i], pMTree->idStr, strictTimeRange);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto _end; return code;
}
bool hasVal = tLDataIterNextRow(&pLDataIter[i], pMTree->idStr);
if (hasVal) {
tMergeTreeAddIter(pMTree, &pLDataIter[i]);
} else {
if (!pMTree->ignoreEarlierTs) {
pMTree->ignoreEarlierTs = pLDataIter[i].ignoreEarlierTs;
}
} }
} }
return code; code = tLDataIterOpen2(&pLDataIter[i], pSttFileReader, i, pMTree->backward, suid, uid, pTimeWindow, pVerRange,
&pMTree->pLoadInfo[i], pMTree->idStr, strictTimeRange);
if (code != TSDB_CODE_SUCCESS) {
goto _end;
}
_end: bool hasVal = tLDataIterNextRow(&pLDataIter[i], pMTree->idStr);
tMergeTreeClose(pMTree); if (hasVal) {
return code; tMergeTreeAddIter(pMTree, &pLDataIter[i]);
} else {
if (!pMTree->ignoreEarlierTs) {
pMTree->ignoreEarlierTs = pLDataIter[i].ignoreEarlierTs;
}
}
}
return code;
_end:
tMergeTreeClose(pMTree);
return code;
} }
void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter) { tRBTreePut(&pMTree->rbt, (SRBTreeNode *)pIter); } void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter) { tRBTreePut(&pMTree->rbt, (SRBTreeNode *)pIter); }

107
source/dnode/vnode/src/tsdb/tsdbRead2.c
View File

@ -910,20 +910,13 @@ static int32_t doLoadBlockIndex(STsdbReader* pReader, SDataFileReader* pFileRead
ASSERT(pBrinBlk->minTbid.suid <= pReader->suid && pBrinBlk->maxTbid.suid >= pReader->suid); ASSERT(pBrinBlk->minTbid.suid <= pReader->suid && pBrinBlk->maxTbid.suid >= pReader->suid);
if (pBrinBlk->minTbid.uid < pList->tableUidList[j]) { if (pBrinBlk->maxTbid.uid < pList->tableUidList[j]) {
i += 1; i += 1;
continue; continue;
} }
if (pBrinBlk->minTbid.uid > pList->tableUidList[j]) {
j += 1;
continue;
}
// todo maxTbid.uid == xxx?
if (pBrinBlk->minTbid.uid == pList->tableUidList[j]) {
// this block belongs to a table that is not queried. // this block belongs to a table that is not queried.
STableBlockScanInfo* pScanInfo = getTableBlockScanInfo(pReader->status.pTableMap, pBrinBlk->minTbid.uid, pReader->idStr); STableBlockScanInfo* pScanInfo = getTableBlockScanInfo(pReader->status.pTableMap, pList->tableUidList[j], pReader->idStr);
if (pScanInfo == NULL) { if (pScanInfo == NULL) {
// tsdbBICacheRelease(pFileReader->pTsdb->biCache, handle); // tsdbBICacheRelease(pFileReader->pTsdb->biCache, handle);
return terrno; return terrno;
@ -934,11 +927,8 @@ static int32_t doLoadBlockIndex(STsdbReader* pReader, SDataFileReader* pFileRead
} }
taosArrayPush(pIndexList, pBrinBlk); taosArrayPush(pIndexList, pBrinBlk);
i += 1;
j += 1; j += 1;
} }
}
int64_t et2 = taosGetTimestampUs(); int64_t et2 = taosGetTimestampUs();
tsdbDebug("load block index for %d/%d tables completed, elapsed time:%.2f ms, set BrinBlk:%.2f ms, size:%.2f Kb %s", tsdbDebug("load block index for %d/%d tables completed, elapsed time:%.2f ms, set BrinBlk:%.2f ms, size:%.2f Kb %s",
@ -1020,6 +1010,14 @@ static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SBlockN
SBrinRecord record = {0}; SBrinRecord record = {0};
for (int32_t j = 0; j < TARRAY2_SIZE(block.numRow); ++j) { for (int32_t j = 0; j < TARRAY2_SIZE(block.numRow); ++j) {
tBrinBlockGet(&block, j, &record); tBrinBlockGet(&block, j, &record);
if (record.suid < pReader->suid) {
continue;
}
if (record.suid > pReader->suid) {
break;
}
{ {
while (record.uid > uid) { while (record.uid > uid) {
k += 1; k += 1;
@ -1539,9 +1537,9 @@ static int32_t doLoadFileBlockData(STsdbReader* pReader, SDataBlockIter* pBlockI
SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter); SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter);
SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo; SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
SBrinRecord* pRecord = &pBlockInfo->record;
// SDataBlk* pBlock = getCurrentBlock(pBlockIter); code = tsdbDataFileReadBlockData(pReader->pFileReader, pRecord, pBlockData);
code = tsdbDataFileReadBlockData(pReader->pFileReader, &pBlockInfo->record, pBlockData);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("%p error occurs in loading file block, global index:%d, table index:%d, brange:%" PRId64 "-%" PRId64 tsdbError("%p error occurs in loading file block, global index:%d, table index:%d, brange:%" PRId64 "-%" PRId64
", rows:%d, code:%s %s", ", rows:%d, code:%s %s",
@ -1552,7 +1550,6 @@ static int32_t doLoadFileBlockData(STsdbReader* pReader, SDataBlockIter* pBlockI
double elapsedTime = (taosGetTimestampUs() - st) / 1000.0; double elapsedTime = (taosGetTimestampUs() - st) / 1000.0;
SBrinRecord* pRecord = &pBlockInfo->record;
tsdbDebug("%p load file block into buffer, global index:%d, index in table block list:%d, brange:%" PRId64 "-%" PRId64 tsdbDebug("%p load file block into buffer, global index:%d, index in table block list:%d, brange:%" PRId64 "-%" PRId64
", rows:%d, minVer:%" PRId64 ", maxVer:%" PRId64 ", elapsed time:%.2f ms, %s", ", rows:%d, minVer:%" PRId64 ", maxVer:%" PRId64 ", elapsed time:%.2f ms, %s",
pReader, pBlockIter->index, pBlockInfo->tbBlockIdx, pRecord->firstKey, pRecord->lastKey, pRecord->numRow, pReader, pBlockIter->index, pBlockInfo->tbBlockIdx, pRecord->firstKey, pRecord->lastKey, pRecord->numRow,
@ -1983,7 +1980,7 @@ static bool fileBlockShouldLoad(STsdbReader* pReader, SFileDataBlockInfo* pBlock
return loadDataBlock; return loadDataBlock;
} }
static bool isCleanFileDataBlock(STsdbReader* pReader, SFileDataBlockInfo* pBlockInfo, SDataBlk* pBlock, static bool isCleanFileDataBlock(STsdbReader* pReader, SFileDataBlockInfo* pBlockInfo,
STableBlockScanInfo* pScanInfo, TSDBKEY keyInBuf, SLastBlockReader* pLastBlockReader) { STableBlockScanInfo* pScanInfo, TSDBKEY keyInBuf, SLastBlockReader* pLastBlockReader) {
SDataBlockToLoadInfo info = {0}; SDataBlockToLoadInfo info = {0};
getBlockToLoadInfo(&info, pBlockInfo, pScanInfo, keyInBuf, pLastBlockReader, pReader); getBlockToLoadInfo(&info, pBlockInfo, pScanInfo, keyInBuf, pLastBlockReader, pReader);
@ -2870,7 +2867,6 @@ static int32_t loadNeighborIfOverlap(SFileDataBlockInfo* pBlockInfo, STableBlock
// SBlockIndex nxtBIndex = {0}; // SBlockIndex nxtBIndex = {0};
*loadNeighbor = false; *loadNeighbor = false;
SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter);
SBrinRecord rec = {0}; SBrinRecord rec = {0};
bool hasNeighbor = getNeighborBlockOfSameTable(pBlockInfo, pBlockScanInfo, &nextIndex, pReader->order, &rec); bool hasNeighbor = getNeighborBlockOfSameTable(pBlockInfo, pBlockScanInfo, &nextIndex, pReader->order, &rec);
@ -2928,13 +2924,14 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
int64_t st = taosGetTimestampUs(); int64_t st = taosGetTimestampUs();
int32_t step = asc ? 1 : -1; int32_t step = asc ? 1 : -1;
double el = 0; double el = 0;
SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter); SBrinRecord* pRecord = &pBlockInfo->record;
SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo; SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
STableBlockScanInfo* pBlockScanInfo = NULL; STableBlockScanInfo* pBlockScanInfo = NULL;
if (pBlockInfo != NULL) { if (pBlockInfo != NULL) {
if (pReader->pIgnoreTables && taosHashGet(*pReader->pIgnoreTables, &pBlockInfo->uid, sizeof(pBlockInfo->uid))) { if (pReader->pIgnoreTables && taosHashGet(*pReader->pIgnoreTables, &pBlockInfo->uid, sizeof(pBlockInfo->uid))) {
setBlockAllDumped(pDumpInfo, pBlock->maxKey.ts, pReader->order); setBlockAllDumped(pDumpInfo, pRecord->lastKey, pReader->order);
return code; return code;
} }
@ -2943,11 +2940,12 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
goto _end; goto _end;
} }
pRecord = &pBlockInfo->record;
TSDBKEY keyInBuf = getCurrentKeyInBuf(pBlockScanInfo, pReader); TSDBKEY keyInBuf = getCurrentKeyInBuf(pBlockScanInfo, pReader);
// it is a clean block, load it directly // it is a clean block, load it directly
if (isCleanFileDataBlock(pReader, pBlockInfo, pBlock, pBlockScanInfo, keyInBuf, pLastBlockReader) && if (isCleanFileDataBlock(pReader, pBlockInfo, pBlockScanInfo, keyInBuf, pLastBlockReader) &&
pBlock->nRow <= pReader->resBlockInfo.capacity) { (pRecord->numRow <= pReader->resBlockInfo.capacity)) {
if (asc || (!hasDataInLastBlock(pLastBlockReader))) { if (asc || (!hasDataInLastBlock(pLastBlockReader))) {
code = copyBlockDataToSDataBlock(pReader); code = copyBlockDataToSDataBlock(pReader);
if (code) { if (code) {
@ -2955,14 +2953,15 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
} }
// record the last key value // record the last key value
pBlockScanInfo->lastKey = asc ? pBlock->maxKey.ts : pBlock->minKey.ts; pBlockScanInfo->lastKey = asc ? pRecord->lastKey : pRecord->firstKey;
goto _end; goto _end;
} }
} }
} else { // file blocks not exist } else { // file blocks not exist
ASSERT(0);
pBlockScanInfo = *pReader->status.pTableIter; pBlockScanInfo = *pReader->status.pTableIter;
if (pReader->pIgnoreTables && taosHashGet(*pReader->pIgnoreTables, &pBlockScanInfo->uid, sizeof(pBlockScanInfo->uid))) { if (pReader->pIgnoreTables && taosHashGet(*pReader->pIgnoreTables, &pBlockScanInfo->uid, sizeof(pBlockScanInfo->uid))) {
setBlockAllDumped(pDumpInfo, pBlock->maxKey.ts, pReader->order); // setBlockAllDumped(pDumpInfo, pBlock->maxKey.ts, pReader->order);
return code; return code;
} }
} }
@ -2972,8 +2971,8 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
while (1) { while (1) {
bool hasBlockData = false; bool hasBlockData = false;
{ {
while (pBlockData->nRow > 0 && while (pBlockData->nRow > 0 && pBlockData->uid == pBlockScanInfo->uid) {
pBlockData->uid == pBlockScanInfo->uid) { // find the first qualified row in data block // find the first qualified row in data block
if (isValidFileBlockRow(pBlockData, pDumpInfo, pBlockScanInfo, pReader)) { if (isValidFileBlockRow(pBlockData, pDumpInfo, pBlockScanInfo, pReader)) {
hasBlockData = true; hasBlockData = true;
break; break;
@ -2981,8 +2980,7 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
pDumpInfo->rowIndex += step; pDumpInfo->rowIndex += step;
pBlock = getCurrentBlock(&pReader->status.blockIter); if (pDumpInfo->rowIndex >= pBlockData->nRow || pDumpInfo->rowIndex < 0) {
if (pDumpInfo->rowIndex >= pBlock->nRow || pDumpInfo->rowIndex < 0) {
pBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter); // NOTE: get the new block info pBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter); // NOTE: get the new block info
// continue check for the next file block if the last ts in the current block // continue check for the next file block if the last ts in the current block
@ -2990,7 +2988,7 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
bool loadNeighbor = false; bool loadNeighbor = false;
code = loadNeighborIfOverlap(pBlockInfo, pBlockScanInfo, pReader, &loadNeighbor); code = loadNeighborIfOverlap(pBlockInfo, pBlockScanInfo, pReader, &loadNeighbor);
if ((!loadNeighbor) || (code != 0)) { if ((!loadNeighbor) || (code != 0)) {
setBlockAllDumped(pDumpInfo, pBlock->maxKey.ts, pReader->order); setBlockAllDumped(pDumpInfo, pRecord->lastKey, pReader->order);
break; break;
} }
} }
@ -3009,8 +3007,8 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
// currently loaded file data block is consumed // currently loaded file data block is consumed
if ((pBlockData->nRow > 0) && (pDumpInfo->rowIndex >= pBlockData->nRow || pDumpInfo->rowIndex < 0)) { if ((pBlockData->nRow > 0) && (pDumpInfo->rowIndex >= pBlockData->nRow || pDumpInfo->rowIndex < 0)) {
pBlock = getCurrentBlock(&pReader->status.blockIter); // pBlock = getCurrentBlock(&pReader->status.blockIter);
setBlockAllDumped(pDumpInfo, pBlock->maxKey.ts, pReader->order); setBlockAllDumped(pDumpInfo, pRecord->lastKey, pReader->order);
break; break;
} }
@ -3290,7 +3288,6 @@ static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) {
// reset the index in last block when handing a new file // reset the index in last block when handing a new file
doCleanupTableScanInfo(pScanInfo); doCleanupTableScanInfo(pScanInfo);
// pStatus->mapDataCleaned = true;
bool hasDataInLastFile = initLastBlockReader(pLastBlockReader, pScanInfo, pReader); bool hasDataInLastFile = initLastBlockReader(pLastBlockReader, pScanInfo, pReader);
if (!hasDataInLastFile) { if (!hasDataInLastFile) {
@ -3625,21 +3622,24 @@ static int32_t buildBlockFromBufferSequentially(STsdbReader* pReader) {
// set the correct start position in case of the first/last file block, according to the query time window // set the correct start position in case of the first/last file block, according to the query time window
static void initBlockDumpInfo(STsdbReader* pReader, SDataBlockIter* pBlockIter) { static void initBlockDumpInfo(STsdbReader* pReader, SDataBlockIter* pBlockIter) {
int64_t lastKey = ASCENDING_TRAVERSE(pReader->order) ? INT64_MIN : INT64_MAX; int64_t lastKey = ASCENDING_TRAVERSE(pReader->order) ? INT64_MIN : INT64_MAX;
SDataBlk* pBlock = getCurrentBlock(pBlockIter);
SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter); SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter);
SReaderStatus* pStatus = &pReader->status;
SFileBlockDumpInfo* pDumpInfo = &pStatus->fBlockDumpInfo;
if (pBlockInfo) { if (pBlockInfo) {
STableBlockScanInfo* pScanInfo = tSimpleHashGet(pBlockIter->pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid)); STableBlockScanInfo* pScanInfo = tSimpleHashGet(pBlockIter->pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid));
if (pScanInfo) { if (pScanInfo) {
lastKey = pScanInfo->lastKey; lastKey = pScanInfo->lastKey;
} }
pDumpInfo->totalRows = pBlockInfo->record.numRow;
pDumpInfo->rowIndex = ASCENDING_TRAVERSE(pReader->order) ? 0 : pBlockInfo->record.numRow - 1;
} else {
pDumpInfo->totalRows = 0;
pDumpInfo->rowIndex = 0;
} }
SReaderStatus* pStatus = &pReader->status;
SFileBlockDumpInfo* pDumpInfo = &pStatus->fBlockDumpInfo;
pDumpInfo->totalRows = pBlock->nRow;
pDumpInfo->allDumped = false; pDumpInfo->allDumped = false;
pDumpInfo->rowIndex = ASCENDING_TRAVERSE(pReader->order) ? 0 : pBlock->nRow - 1;
pDumpInfo->lastKey = lastKey; pDumpInfo->lastKey = lastKey;
} }
@ -4041,7 +4041,7 @@ typedef enum {
CHECK_FILEBLOCK_QUIT = 0x2, CHECK_FILEBLOCK_QUIT = 0x2,
} CHECK_FILEBLOCK_STATE; } CHECK_FILEBLOCK_STATE;
static int32_t checkForNeighborFileBlock(STsdbReader* pReader, STableBlockScanInfo* pScanInfo, SDataBlk* pBlock, static int32_t checkForNeighborFileBlock(STsdbReader* pReader, STableBlockScanInfo* pScanInfo,
SFileDataBlockInfo* pFBlock, SRowMerger* pMerger, int64_t key, SFileDataBlockInfo* pFBlock, SRowMerger* pMerger, int64_t key,
CHECK_FILEBLOCK_STATE* state) { CHECK_FILEBLOCK_STATE* state) {
SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo; SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
@ -4085,13 +4085,13 @@ int32_t doMergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pSc
CHECK_FILEBLOCK_STATE st; CHECK_FILEBLOCK_STATE st;
SFileDataBlockInfo* pFileBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter); SFileDataBlockInfo* pFileBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter);
SDataBlk* pCurrentBlock = getCurrentBlock(&pReader->status.blockIter); // SDataBlk* pCurrentBlock = getCurrentBlock(&pReader->status.blockIter);
if (pFileBlockInfo == NULL) { if (pFileBlockInfo == NULL) {
st = CHECK_FILEBLOCK_QUIT; st = CHECK_FILEBLOCK_QUIT;
break; break;
} }
checkForNeighborFileBlock(pReader, pScanInfo, pCurrentBlock, pFileBlockInfo, pMerger, key, &st); checkForNeighborFileBlock(pReader, pScanInfo, pFileBlockInfo, pMerger, key, &st);
if (st == CHECK_FILEBLOCK_QUIT) { if (st == CHECK_FILEBLOCK_QUIT) {
break; break;
} }
@ -4767,14 +4767,10 @@ void tsdbReaderClose2(STsdbReader* pReader) {
pReader->pReadSnap = NULL; pReader->pReadSnap = NULL;
tsdbReleaseReader(pReader); tsdbReleaseReader(pReader);
tsdbUninitReaderLock(pReader); tsdbUninitReaderLock(pReader);
taosMemoryFreeClear(pReader->status.pLDataIter);
taosMemoryFreeClear(pReader->status.uidList.tableUidList);
SIOCostSummary* pCost = &pReader->cost; SIOCostSummary* pCost = &pReader->cost;
SFilesetIter* pFilesetIter = &pReader->status.fileIter;
SFilesetIter* pFilesetIter = &pReader->status.fileIter;
if (pFilesetIter->pLastBlockReader != NULL) { if (pFilesetIter->pLastBlockReader != NULL) {
SLastBlockReader* pLReader = pFilesetIter->pLastBlockReader; SLastBlockReader* pLReader = pFilesetIter->pLastBlockReader;
tMergeTreeClose(&pLReader->mergeTree); tMergeTreeClose(&pLReader->mergeTree);
@ -4782,9 +4778,15 @@ void tsdbReaderClose2(STsdbReader* pReader) {
getLastBlockLoadInfo(pLReader->pInfo, &pCost->lastBlockLoad, &pCost->lastBlockLoadTime); getLastBlockLoadInfo(pLReader->pInfo, &pCost->lastBlockLoad, &pCost->lastBlockLoadTime);
pLReader->pInfo = destroyLastBlockLoadInfo(pLReader->pInfo); pLReader->pInfo = destroyLastBlockLoadInfo(pLReader->pInfo);
// todo dynamic allocate the number of stt data iter
destroySttBlockReader(pReader->status.pLDataIter, pReader->pTsdb->pVnode->config.sttTrigger);
taosMemoryFree(pLReader); taosMemoryFree(pLReader);
} }
taosMemoryFreeClear(pReader->status.pLDataIter);
taosMemoryFreeClear(pReader->status.uidList.tableUidList);
tsdbDebug( tsdbDebug(
"%p :io-cost summary: head-file:%" PRIu64 ", head-file time:%.2f ms, SMA:%" PRId64 "%p :io-cost summary: head-file:%" PRIu64 ", head-file time:%.2f ms, SMA:%" PRId64
" SMA-time:%.2f ms, fileBlocks:%" PRId64 " SMA-time:%.2f ms, fileBlocks:%" PRId64
@ -5210,9 +5212,10 @@ int32_t tsdbRetrieveDatablockSMA2(STsdbReader* pReader, SSDataBlock* pDataBlock,
} }
int64_t st = taosGetTimestampUs(); int64_t st = taosGetTimestampUs();
ASSERT(0);
SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter); // SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter);
if (tDataBlkHasSma(pBlock)) { if (/*tDataBlkHasSma(pBlock)*/1) {
// code = tsdbReadBlockSma(pReader->pFileReader, pBlock, pSup->pColAgg); // code = tsdbReadBlockSma(pReader->pFileReader, pBlock, pSup->pColAgg);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbDebug("vgId:%d, failed to load block SMA for uid %" PRIu64 ", code:%s, %s", 0, pFBlock->uid, tstrerror(code), tsdbDebug("vgId:%d, failed to load block SMA for uid %" PRIu64 ", code:%s, %s", 0, pFBlock->uid, tstrerror(code),
@ -5250,7 +5253,7 @@ int32_t tsdbRetrieveDatablockSMA2(STsdbReader* pReader, SSDataBlock* pDataBlock,
} }
// do fill all null column value SMA info // do fill all null column value SMA info
if (doFillNullColSMA(pSup, pBlock->nRow, numOfCols, pTsAgg)) { if (doFillNullColSMA(pSup, pFBlock->record.numRow, numOfCols, pTsAgg)) {
*hasNullSMA = true; *hasNullSMA = true;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -5470,11 +5473,13 @@ int32_t tsdbGetFileBlocksDistInfo2(STsdbReader* pReader, STableBlockDistInfo* pT
pTableBlockInfo->numOfTables = numOfTables; pTableBlockInfo->numOfTables = numOfTables;
bool hasNext = (pBlockIter->numOfBlocks > 0); bool hasNext = (pBlockIter->numOfBlocks > 0);
ASSERT(0);
while (true) { while (true) {
if (hasNext) { if (hasNext) {
SDataBlk* pBlock = getCurrentBlock(pBlockIter); // SDataBlk* pBlock = getCurrentBlock(pBlockIter);
int32_t numOfRows = pBlock->nRow; int32_t numOfRows = 0;//pFB->nRow;
pTableBlockInfo->totalRows += numOfRows; pTableBlockInfo->totalRows += numOfRows;
if (numOfRows > pTableBlockInfo->maxRows) { if (numOfRows > pTableBlockInfo->maxRows) {
@ -5489,7 +5494,7 @@ int32_t tsdbGetFileBlocksDistInfo2(STsdbReader* pReader, STableBlockDistInfo* pT
pTableBlockInfo->numOfSmallBlocks += 1; pTableBlockInfo->numOfSmallBlocks += 1;
} }
pTableBlockInfo->totalSize += pBlock->aSubBlock[0].szBlock; pTableBlockInfo->totalSize += 0;//pBlock->aSubBlock[0].szBlock;
int32_t bucketIndex = getBucketIndex(pTableBlockInfo->defMinRows, bucketRange, numOfRows, numOfBuckets); int32_t bucketIndex = getBucketIndex(pTableBlockInfo->defMinRows, bucketRange, numOfRows, numOfBuckets);
pTableBlockInfo->blockRowsHisto[bucketIndex]++; pTableBlockInfo->blockRowsHisto[bucketIndex]++;