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

fix(query): fix some memory leaks.

This commit is contained in:
Haojun Liao 2022-11-21 22:40:20 +08:00
parent cfb1fe9cb9
commit 6d5a256774
3 changed files with 19 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
source/dnode/vnode/src/tsdb/tsdbRead.c
View File

@ -361,11 +361,17 @@ static void resetAllDataBlockScanInfo(SHashObj* pTableMap, int64_t ts) {
STableBlockScanInfo* pInfo = *(STableBlockScanInfo**)p;
pInfo->iterInit = false;
pInfo->iter.hasVal = false;
pInfo->iiter.hasVal = false;
if (pInfo->iter.iter != NULL) {
pInfo->iter.iter = tsdbTbDataIterDestroy(pInfo->iter.iter);
}
if (pInfo->iiter.iter != NULL) {
pInfo->iiter.iter = tsdbTbDataIterDestroy(pInfo->iiter.iter);
}
pInfo->delSkyline = taosArrayDestroy(pInfo->delSkyline);
pInfo->lastKey = ts;
}
@ -373,6 +379,8 @@ static void resetAllDataBlockScanInfo(SHashObj* pTableMap, int64_t ts) {
static void clearBlockScanInfo(STableBlockScanInfo* p) {
p->iterInit = false;
p->iter.hasVal = false;
p->iiter.hasVal = false;
if (p->iter.iter != NULL) {
@ -388,9 +396,9 @@ static void clearBlockScanInfo(STableBlockScanInfo* p) {
tMapDataClear(&p->mapData);
}
static void destroyAllBlockScanInfo(SHashObj* pTableMap, bool clearEntry) {
static void destroyAllBlockScanInfo(SHashObj* pTableMap) {
void* p = NULL;
while (clearEntry && ((p = taosHashIterate(pTableMap, p)) != NULL)) {
while ((p = taosHashIterate(pTableMap, p)) != NULL) {
clearBlockScanInfo(*(STableBlockScanInfo**)p);
}
@ -2226,6 +2234,7 @@ static int32_t initMemDataIterator(STableBlockScanInfo* pBlockScanInfo, STsdbRea
if (pReader->pReadSnap->pMem != NULL) {
d = tsdbGetTbDataFromMemTable(pReader->pReadSnap->pMem, pReader->suid, pBlockScanInfo->uid);
if (d != NULL) {
ASSERT(pBlockScanInfo->iter.iter == NULL);
code = tsdbTbDataIterCreate(d, &startKey, backward, &pBlockScanInfo->iter.iter);
if (code == TSDB_CODE_SUCCESS) {
pBlockScanInfo->iter.hasVal = (tsdbTbDataIterGet(pBlockScanInfo->iter.iter) != NULL);
@ -3789,9 +3798,7 @@ int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, void* pTableL
updateBlockSMAInfo(pReader->pSchema, &pReader->suppInfo);
}
STsdbReader* p = pReader->innerReader[0] != NULL ? pReader->innerReader[0] : pReader;
pReader->status.pTableMap = createDataBlockScanInfo(p, pTableList, numOfTables);
pReader->status.pTableMap = createDataBlockScanInfo(pReader, pTableList, numOfTables);
if (pReader->status.pTableMap == NULL) {
tsdbReaderClose(pReader);
*ppReader = NULL;
@ -3849,7 +3856,7 @@ void tsdbReaderClose(STsdbReader* pReader) {
}
{
if (pReader->innerReader[0] != NULL) {
if (pReader->innerReader[0] != NULL || pReader->innerReader[1] != NULL) {
STsdbReader* p = pReader->innerReader[0];
p->status.pTableMap = NULL;
@ -3887,9 +3894,12 @@ void tsdbReaderClose(STsdbReader* pReader) {
cleanupDataBlockIterator(&pReader->status.blockIter);
size_t numOfTables = taosHashGetSize(pReader->status.pTableMap);
destroyAllBlockScanInfo(pReader->status.pTableMap, (pReader->innerReader[0] == NULL) ? true : false);
if (pReader->status.pTableMap != NULL) {
destroyAllBlockScanInfo(pReader->status.pTableMap);
clearBlockScanInfoBuf(&pReader->blockInfoBuf);
}
blockDataDestroy(pReader->pResBlock);
clearBlockScanInfoBuf(&pReader->blockInfoBuf);
if (pReader->pFileReader != NULL) {
tsdbDataFReaderClose(&pReader->pFileReader);

1
source/libs/executor/src/cachescanoperator.c
View File

@ -246,6 +246,7 @@ void destroyLastrowScanOperator(void* param) {
pInfo->pLastrowReader = tsdbCacherowsReaderClose(pInfo->pLastrowReader);
}
cleanupExprSupp(&pInfo->pseudoExprSup);
taosMemoryFreeClear(param);
}

40
source/libs/executor/src/executorimpl.c
View File

@ -20,7 +20,6 @@
#include "querynodes.h"
#include "tfill.h"
#include "tname.h"
#include "tref.h"
#include "tdatablock.h"
#include "tglobal.h"
@ -134,45 +133,6 @@ static int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock,
static void initCtxOutputBuffer(SqlFunctionCtx* pCtx, int32_t size);
static void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, int32_t numOfOutput, uint64_t groupId);
#if 0
static bool chkResultRowFromKey(STaskRuntimeEnv* pRuntimeEnv, SResultRowInfo* pResultRowInfo, char* pData,
int16_t bytes, bool masterscan, uint64_t uid) {
bool existed = false;
SET_RES_WINDOW_KEY(pRuntimeEnv->keyBuf, pData, bytes, uid);
SResultRow** p1 =
(SResultRow**)taosHashGet(pRuntimeEnv->pResultRowHashTable, pRuntimeEnv->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes));
// in case of repeat scan/reverse scan, no new time window added.
if (QUERY_IS_INTERVAL_QUERY(pRuntimeEnv->pQueryAttr)) {
if (!masterscan) { // the *p1 may be NULL in case of sliding+offset exists.
return p1 != NULL;
}
if (p1 != NULL) {
if (pResultRowInfo->size == 0) {
existed = false;
} else if (pResultRowInfo->size == 1) {
// existed = (pResultRowInfo->pResult[0] == (*p1));
} else { // check if current pResultRowInfo contains the existed pResultRow
SET_RES_EXT_WINDOW_KEY(pRuntimeEnv->keyBuf, pData, bytes, uid, pResultRowInfo);
int64_t* index =
taosHashGet(pRuntimeEnv->pResultRowListSet, pRuntimeEnv->keyBuf, GET_RES_EXT_WINDOW_KEY_LEN(bytes));
if (index != NULL) {
existed = true;
} else {
existed = false;
}
}
}
return existed;
}
return p1 != NULL;
}
#endif
SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int32_t* currentPageId, int32_t interBufSize) {
SFilePage* pData = NULL;