From 1d8af5d4cf18a68b340de42877f60410d0f70e94 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Sat, 7 May 2022 15:11:49 +0800 Subject: [PATCH 1/3] fix(query): fix memory leak. --- include/libs/executor/executor.h | 9 --------- source/libs/executor/src/executorimpl.c | 7 ++++--- source/libs/executor/src/scanoperator.c | 7 +++++++ 3 files changed, 11 insertions(+), 12 deletions(-) diff --git a/include/libs/executor/executor.h b/include/libs/executor/executor.h index ad57cbf4e4..e0fce2aa6a 100644 --- a/include/libs/executor/executor.h +++ b/include/libs/executor/executor.h @@ -165,15 +165,6 @@ int32_t qGetQualifiedTableIdList(void* pTableList, const char* tagCond, int32_t */ int32_t qUpdateQueriedTableIdList(qTaskInfo_t tinfo, int64_t uid, int32_t type); -/** - * release the query handle and decrease the reference count in cache - * @param pMgmt - * @param pQInfo - * @param freeHandle - * @return - */ -void** qReleaseTask(void* pMgmt, void* pQInfo, bool freeHandle); - void qProcessFetchRsp(void* parent, struct SRpcMsg* pMsg, struct SEpSet* pEpSet); int32_t qGetExplainExecInfo(qTaskInfo_t tinfo, int32_t *resNum, SExplainExecInfo **pRes); diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 46c30c43c7..f9b06f360f 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -4184,6 +4184,7 @@ static void destroyOperatorInfo(SOperatorInfo* pOperator) { pOperator->numOfDownstream = 0; } + taosMemoryFree(pOperator->pExpr); taosMemoryFreeClear(pOperator->info); taosMemoryFreeClear(pOperator); } @@ -4195,8 +4196,6 @@ int32_t doInitAggInfoSup(SAggSupporter* pAggSup, SqlFunctionCtx* pCtx, int32_t n pAggSup->resultRowSize = getResultRowSize(pCtx, numOfOutput); pAggSup->keyBuf = taosMemoryCalloc(1, keyBufSize + POINTER_BYTES + sizeof(int64_t)); pAggSup->pResultRowHashTable = taosHashInit(10, hashFn, true, HASH_NO_LOCK); - // pAggSup->pResultRowListSet = taosHashInit(100, hashFn, false, HASH_NO_LOCK); - // pAggSup->pResultRowArrayList = taosArrayInit(10, sizeof(SResultRowCell)); if (pAggSup->keyBuf == NULL /*|| pAggSup->pResultRowArrayList == NULL || pAggSup->pResultRowListSet == NULL*/ || pAggSup->pResultRowHashTable == NULL) { @@ -4358,6 +4357,7 @@ void destroySFillOperatorInfo(void* param, int32_t numOfOutput) { static void destroyProjectOperatorInfo(void* param, int32_t numOfOutput) { SProjectOperatorInfo* pInfo = (SProjectOperatorInfo*)param; doDestroyBasicInfo(&pInfo->binfo, numOfOutput); + cleanupAggSup(&pInfo->aggSup); } void destroyExchangeOperatorInfo(void* param, int32_t numOfOutput) { @@ -4687,7 +4687,7 @@ static SExecTaskInfo* createExecTaskInfo(uint64_t queryId, uint64_t taskId, EOPT char* p = taosMemoryCalloc(1, 128); snprintf(p, 128, "TID:0x%" PRIx64 " QID:0x%" PRIx64, taskId, queryId); - pTaskInfo->id.str = strdup(p); + pTaskInfo->id.str = p; return pTaskInfo; } @@ -5287,6 +5287,7 @@ void doDestroyTask(SExecTaskInfo* pTaskInfo) { qDebug("%s execTask is freed", GET_TASKID(pTaskInfo)); doDestroyTableQueryInfo(&pTaskInfo->tableqinfoGroupInfo); + destroyOperatorInfo(pTaskInfo->pRoot); // taosArrayDestroy(pTaskInfo->summary.queryProfEvents); // taosHashCleanup(pTaskInfo->summary.operatorProfResults); diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index b28a65d1d2..132c279218 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -670,6 +670,8 @@ static void destroySysScanOperator(void* param, int32_t numOfOutput) { if (strncasecmp(name, TSDB_INS_TABLE_USER_TABLES, TSDB_TABLE_FNAME_LEN) == 0) { metaCloseTbCursor(pInfo->pCur); } + + taosArrayDestroy(pInfo->scanCols); } EDealRes getDBNameFromConditionWalker(SNode* pNode, void* pContext) { @@ -1248,6 +1250,11 @@ static SSDataBlock* doTagScan(SOperatorInfo* pOperator) { SExprInfo* pExprInfo = &pOperator->pExpr[0]; SSDataBlock* pRes = pInfo->pRes; + if (taosArrayGetSize(pInfo->pTableGroups->pGroupList) == 0) { + setTaskStatus(pTaskInfo, TASK_COMPLETED); + return NULL; + } + SArray* pa = taosArrayGetP(pInfo->pTableGroups->pGroupList, 0); char str[512] = {0}; From f508da9e139e2e404e59fc53af15f231a05b64c3 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Sat, 7 May 2022 16:22:52 +0800 Subject: [PATCH 2/3] fix(query): eliminate memory leak --- include/common/tcommon.h | 2 -- source/libs/executor/src/executorimpl.c | 36 +++++++++++++++++++------ source/libs/executor/src/scanoperator.c | 2 +- 3 files changed, 29 insertions(+), 11 deletions(-) diff --git a/include/common/tcommon.h b/include/common/tcommon.h index 16653ebfee..84ea208508 100644 --- a/include/common/tcommon.h +++ b/include/common/tcommon.h @@ -115,8 +115,6 @@ void* tDecodeDataBlocks(const void* buf, SArray** blocks); void colDataDestroy(SColumnInfoData* pColData); static FORCE_INLINE void blockDestroyInner(SSDataBlock* pBlock) { - // WARNING: do not use info.numOfCols, - // sometimes info.numOfCols != array size int32_t numOfOutput = taosArrayGetSize(pBlock->pDataBlock); for (int32_t i = 0; i < numOfOutput; ++i) { SColumnInfoData* pColInfoData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, i); diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index f9b06f360f..321b7b68a5 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -1225,6 +1225,8 @@ static void* destroySqlFunctionCtx(SqlFunctionCtx* pCtx, int32_t numOfOutput) { taosVariantDestroy(&pCtx[i].tag); taosMemoryFreeClear(pCtx[i].subsidiaries.pCtx); + taosMemoryFree(pCtx[i].input.pData); + taosMemoryFree(pCtx[i].input.pColumnDataAgg); } 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 compLen, int32_t numOfOutput, int64_t startTs, uint64_t* total, SArray* pColList) { - blockDataEnsureCapacity(pRes, numOfRows); - if (pColList == NULL) { // data from other sources + blockDataEnsureCapacity(pRes, numOfRows); + int32_t dataLen = *(int32_t*)pData; pData += sizeof(int32_t); @@ -2898,20 +2900,23 @@ int32_t setSDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadI 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) { SColumnInfoData idata = {0}; idata.info.type = pSchema[i].type; idata.info.bytes = pSchema[i].bytes; idata.info.colId = pSchema[i].colId; - taosArrayPush(block.pDataBlock, &idata); + taosArrayPush(pBlock->pDataBlock, &idata); 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; 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]); ASSERT(colLen[i] >= 0); - SColumnInfoData* pColInfoData = taosArrayGet(block.pDataBlock, i); + SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i); if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) { pColInfoData->varmeta.length = colLen[i]; pColInfoData->varmeta.allocLen = colLen[i]; @@ -2943,7 +2948,10 @@ int32_t setSDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadI } // data from mnode - relocateColumnData(pRes, pColList, block.pDataBlock); + relocateColumnData(pRes, pColList, pBlock->pDataBlock); + taosArrayDestroy(pBlock->pDataBlock); + taosMemoryFree(pBlock); +// blockDataDestroy(pBlock); } pRes->info.rows = numOfRows; @@ -4184,6 +4192,17 @@ static void destroyOperatorInfo(SOperatorInfo* pOperator) { 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); taosMemoryFreeClear(pOperator->info); taosMemoryFreeClear(pOperator); @@ -4358,6 +4377,7 @@ static void destroyProjectOperatorInfo(void* param, int32_t numOfOutput) { SProjectOperatorInfo* pInfo = (SProjectOperatorInfo*)param; doDestroyBasicInfo(&pInfo->binfo, numOfOutput); cleanupAggSup(&pInfo->aggSup); + taosArrayDestroy(pInfo->pPseudoColInfo); } void destroyExchangeOperatorInfo(void* param, int32_t numOfOutput) { diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 132c279218..2b94c5fdce 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -1156,7 +1156,7 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSDataBlock* pRe pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; - pOperator->numOfExprs = pResBlock->info.numOfCols; + pOperator->numOfExprs = pResBlock->info.numOfCols; pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doSysTableScan, NULL, NULL, destroySysScanOperator, NULL, NULL, NULL); pOperator->pTaskInfo = pTaskInfo; From f867e8ea9165c20f0f09c56765e7f236401db6ad Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Sat, 7 May 2022 17:23:05 +0800 Subject: [PATCH 3/3] fix(query): eliminate memory leak. --- source/libs/executor/inc/executorimpl.h | 1 + source/libs/executor/src/executorMain.c | 15 --------------- source/libs/executor/src/executorimpl.c | 2 +- 3 files changed, 2 insertions(+), 16 deletions(-) diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 9eed76a0fe..3ed65f4a05 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -225,6 +225,7 @@ typedef struct SExecTaskInfo { char* sql; // query sql string jmp_buf env; // jump to this position when error happens. EOPTR_EXEC_MODEL execModel; // operator execution model [batch model|stream model] + struct SSubplan *plan; struct SOperatorInfo* pRoot; } SExecTaskInfo; diff --git a/source/libs/executor/src/executorMain.c b/source/libs/executor/src/executorMain.c index 7705744694..ba77950912 100644 --- a/source/libs/executor/src/executorMain.c +++ b/source/libs/executor/src/executorMain.c @@ -36,21 +36,6 @@ typedef struct STaskMgmt { bool closed; } STaskMgmt; -static void taskMgmtKillTaskFn(void* handle, void* param1) { - void** fp = (void**)handle; - qKillTask(*fp); -} - -static void freeqinfoFn(void *qhandle) { - void** handle = qhandle; - if (handle == NULL || *handle == NULL) { - return; - } - - qKillTask(*handle); - qDestroyTask(*handle); -} - int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId, SSubplan* pSubplan, qTaskInfo_t* pTaskInfo, DataSinkHandle* handle, EOPTR_EXEC_MODEL model) { assert(readHandle != NULL && pSubplan != NULL); diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 34339d3a7f..c0ea54ce4a 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -5209,6 +5209,7 @@ int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SRead goto _complete; } + (*pTaskInfo)->plan = pPlan; return code; _complete: @@ -5311,7 +5312,6 @@ void doDestroyTask(SExecTaskInfo* pTaskInfo) { // taosArrayDestroy(pTaskInfo->summary.queryProfEvents); // taosHashCleanup(pTaskInfo->summary.operatorProfResults); - destroyOperatorInfo(pTaskInfo->pRoot); taosMemoryFreeClear(pTaskInfo->sql); taosMemoryFreeClear(pTaskInfo->id.str); taosMemoryFreeClear(pTaskInfo);