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

fix(query): eliminate memory leak

This commit is contained in:
Haojun Liao 2022-05-07 16:22:52 +08:00
parent 1d8af5d4cf
commit f508da9e13
3 changed files with 29 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/common/tcommon.h
View File

@ -115,8 +115,6 @@ void* tDecodeDataBlocks(const void* buf, SArray** blocks);
void colDataDestroy(SColumnInfoData* pColData); void colDataDestroy(SColumnInfoData* pColData);
static FORCE_INLINE void blockDestroyInner(SSDataBlock* pBlock) { static FORCE_INLINE void blockDestroyInner(SSDataBlock* pBlock) {
// WARNING: do not use info.numOfCols,
// sometimes info.numOfCols != array size
int32_t numOfOutput = taosArrayGetSize(pBlock->pDataBlock); int32_t numOfOutput = taosArrayGetSize(pBlock->pDataBlock);
for (int32_t i = 0; i < numOfOutput; ++i) { for (int32_t i = 0; i < numOfOutput; ++i) {
SColumnInfoData* pColInfoData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, i); SColumnInfoData* pColInfoData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, i);

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

@ -1225,6 +1225,8 @@ static void* destroySqlFunctionCtx(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
taosVariantDestroy(&pCtx[i].tag); taosVariantDestroy(&pCtx[i].tag);
taosMemoryFreeClear(pCtx[i].subsidiaries.pCtx); taosMemoryFreeClear(pCtx[i].subsidiaries.pCtx);
taosMemoryFree(pCtx[i].input.pData);
taosMemoryFree(pCtx[i].input.pColumnDataAgg);
} }
taosMemoryFreeClear(pCtx); taosMemoryFreeClear(pCtx);
@ -2840,9 +2842,9 @@ void relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray
int32_t setSDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadInfo, int32_t numOfRows, char* pData, int32_t setSDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadInfo, int32_t numOfRows, char* pData,
int32_t compLen, int32_t numOfOutput, int64_t startTs, uint64_t* total, int32_t compLen, int32_t numOfOutput, int64_t startTs, uint64_t* total,
SArray* pColList) { SArray* pColList) {
if (pColList == NULL) { // data from other sources
blockDataEnsureCapacity(pRes, numOfRows); blockDataEnsureCapacity(pRes, numOfRows);
if (pColList == NULL) { // data from other sources
int32_t dataLen = *(int32_t*)pData; int32_t dataLen = *(int32_t*)pData;
pData += sizeof(int32_t); pData += sizeof(int32_t);
@ -2898,20 +2900,23 @@ int32_t setSDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadI
pStart += sizeof(SSysTableSchema); pStart += sizeof(SSysTableSchema);
} }
SSDataBlock block = {.pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData)), .info.numOfCols = numOfCols}; SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
pBlock->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
pBlock->info.numOfCols = numOfCols;
for (int32_t i = 0; i < numOfCols; ++i) { for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData idata = {0}; SColumnInfoData idata = {0};
idata.info.type = pSchema[i].type; idata.info.type = pSchema[i].type;
idata.info.bytes = pSchema[i].bytes; idata.info.bytes = pSchema[i].bytes;
idata.info.colId = pSchema[i].colId; idata.info.colId = pSchema[i].colId;
taosArrayPush(block.pDataBlock, &idata); taosArrayPush(pBlock->pDataBlock, &idata);
if (IS_VAR_DATA_TYPE(idata.info.type)) { if (IS_VAR_DATA_TYPE(idata.info.type)) {
block.info.hasVarCol = true; pBlock->info.hasVarCol = true;
} }
} }
blockDataEnsureCapacity(&block, numOfRows); blockDataEnsureCapacity(pBlock, numOfRows);
int32_t dataLen = *(int32_t*)pStart; int32_t dataLen = *(int32_t*)pStart;
uint64_t groupId = *(uint64_t*)(pStart + sizeof(int32_t)); uint64_t groupId = *(uint64_t*)(pStart + sizeof(int32_t));
@ -2924,7 +2929,7 @@ int32_t setSDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadI
colLen[i] = htonl(colLen[i]); colLen[i] = htonl(colLen[i]);
ASSERT(colLen[i] >= 0); ASSERT(colLen[i] >= 0);
SColumnInfoData* pColInfoData = taosArrayGet(block.pDataBlock, i); SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) { if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
pColInfoData->varmeta.length = colLen[i]; pColInfoData->varmeta.length = colLen[i];
pColInfoData->varmeta.allocLen = colLen[i]; pColInfoData->varmeta.allocLen = colLen[i];
@ -2943,7 +2948,10 @@ int32_t setSDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadI
} }
// data from mnode // data from mnode
relocateColumnData(pRes, pColList, block.pDataBlock); relocateColumnData(pRes, pColList, pBlock->pDataBlock);
taosArrayDestroy(pBlock->pDataBlock);
taosMemoryFree(pBlock);
// blockDataDestroy(pBlock);
} }
pRes->info.rows = numOfRows; pRes->info.rows = numOfRows;
@ -4184,6 +4192,17 @@ static void destroyOperatorInfo(SOperatorInfo* pOperator) {
pOperator->numOfDownstream = 0; pOperator->numOfDownstream = 0;
} }
if (pOperator->pExpr != NULL) {
for (int32_t i = 0; i < pOperator->numOfExprs; ++i) {
SExprInfo* pExprInfo = &pOperator->pExpr[i];
if (pExprInfo->pExpr->nodeType == QUERY_NODE_COLUMN) {
taosMemoryFree(pExprInfo->base.pParam[0].pCol);
}
taosMemoryFree(pExprInfo->base.pParam);
taosMemoryFree(pExprInfo->pExpr);
}
}
taosMemoryFree(pOperator->pExpr); taosMemoryFree(pOperator->pExpr);
taosMemoryFreeClear(pOperator->info); taosMemoryFreeClear(pOperator->info);
taosMemoryFreeClear(pOperator); taosMemoryFreeClear(pOperator);
@ -4358,6 +4377,7 @@ static void destroyProjectOperatorInfo(void* param, int32_t numOfOutput) {
SProjectOperatorInfo* pInfo = (SProjectOperatorInfo*)param; SProjectOperatorInfo* pInfo = (SProjectOperatorInfo*)param;
doDestroyBasicInfo(&pInfo->binfo, numOfOutput); doDestroyBasicInfo(&pInfo->binfo, numOfOutput);
cleanupAggSup(&pInfo->aggSup); cleanupAggSup(&pInfo->aggSup);
taosArrayDestroy(pInfo->pPseudoColInfo);
} }
void destroyExchangeOperatorInfo(void* param, int32_t numOfOutput) { void destroyExchangeOperatorInfo(void* param, int32_t numOfOutput) {