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

fix(query): fix memory leak.

This commit is contained in:
Haojun Liao 2024-08-30 10:19:16 +08:00
parent 251bdf51c1
commit 5a5bfa5b8c
1 changed files with 21 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
source/libs/executor/src/sortoperator.c
View File

@ -263,17 +263,17 @@ static int32_t getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock,
SSortOperatorInfo* pInfo, SSDataBlock** pResBlock) { SSortOperatorInfo* pInfo, SSDataBlock** pResBlock) {
QRY_OPTR_CHECK(pResBlock); QRY_OPTR_CHECK(pResBlock);
blockDataCleanup(pDataBlock); blockDataCleanup(pDataBlock);
int32_t lino = 0;
int32_t code = 0;
SSDataBlock* p = NULL; SSDataBlock* p = NULL;
int32_t code = tsortGetSortedDataBlock(pHandle, &p); code = tsortGetSortedDataBlock(pHandle, &p);
if (p == NULL || (code != 0)) { if (p == NULL || (code != 0)) {
return code; return code;
} }
code = blockDataEnsureCapacity(p, capacity); code = blockDataEnsureCapacity(p, capacity);
if (code) { QUERY_CHECK_CODE(code, lino, _error);
return code;
}
STupleHandle* pTupleHandle; STupleHandle* pTupleHandle;
while (1) { while (1) {
@ -282,51 +282,40 @@ static int32_t getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock,
} else { } else {
code = tsortNextTuple(pHandle, &pTupleHandle); code = tsortNextTuple(pHandle, &pTupleHandle);
} }
if (pTupleHandle == NULL || code != 0) { if (pTupleHandle == NULL || code != 0) {
lino = __LINE__;
break; break;
} }
code = appendOneRowToDataBlock(p, pTupleHandle); code = appendOneRowToDataBlock(p, pTupleHandle);
if (code) { QUERY_CHECK_CODE(code, lino, _error);
return code;
}
if (p->info.rows >= capacity) { if (p->info.rows >= capacity) {
break; break;
} }
} }
if (TSDB_CODE_SUCCESS != code) {
return code; QUERY_CHECK_CODE(code, lino, _error);
}
if (p->info.rows > 0) { if (p->info.rows > 0) {
code = blockDataEnsureCapacity(pDataBlock, capacity); code = blockDataEnsureCapacity(pDataBlock, capacity);
if (code) { QUERY_CHECK_CODE(code, lino, _error);
return code;
}
// todo extract function to handle this // todo extract function to handle this
int32_t numOfCols = taosArrayGetSize(pColMatchInfo); int32_t numOfCols = taosArrayGetSize(pColMatchInfo);
for (int32_t i = 0; i < numOfCols; ++i) { for (int32_t i = 0; i < numOfCols; ++i) {
SColMatchItem* pmInfo = taosArrayGet(pColMatchInfo, i); SColMatchItem* pmInfo = taosArrayGet(pColMatchInfo, i);
if (pmInfo == NULL) { QUERY_CHECK_NULL(pmInfo, code, lino, _error, terrno);
return terrno;
}
SColumnInfoData* pSrc = taosArrayGet(p->pDataBlock, pmInfo->srcSlotId); SColumnInfoData* pSrc = taosArrayGet(p->pDataBlock, pmInfo->srcSlotId);
if (pSrc == NULL) { QUERY_CHECK_NULL(pSrc, code, lino, _error, terrno);
return terrno;
}
SColumnInfoData* pDst = taosArrayGet(pDataBlock->pDataBlock, pmInfo->dstSlotId); SColumnInfoData* pDst = taosArrayGet(pDataBlock->pDataBlock, pmInfo->dstSlotId);
if (pDst == NULL) { QUERY_CHECK_NULL(pDst, code, lino, _error, terrno);
return terrno;
}
code = colDataAssign(pDst, pSrc, p->info.rows, &pDataBlock->info); code = colDataAssign(pDst, pSrc, p->info.rows, &pDataBlock->info);
if (code) { QUERY_CHECK_CODE(code, lino, _error);
return code;
}
} }
pDataBlock->info.dataLoad = 1; pDataBlock->info.dataLoad = 1;
@ -338,6 +327,12 @@ static int32_t getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock,
blockDataDestroy(p); blockDataDestroy(p);
*pResBlock = (pDataBlock->info.rows > 0) ? pDataBlock : NULL; *pResBlock = (pDataBlock->info.rows > 0) ? pDataBlock : NULL;
return code; return code;
_error:
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
blockDataDestroy(p);
return code;
} }
int32_t loadNextDataBlock(void* param, SSDataBlock** ppBlock) { int32_t loadNextDataBlock(void* param, SSDataBlock** ppBlock) {
@ -746,6 +741,7 @@ int32_t doGroupSort(SOperatorInfo* pOperator, SSDataBlock** pResBlock) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code)); qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
T_LONG_JMP(pOperator->pTaskInfo->env, code); T_LONG_JMP(pOperator->pTaskInfo->env, code);
} }
code = getGroupSortedBlockData(pInfo->pCurrSortHandle, pInfo->binfo.pRes, pOperator->resultInfo.capacity, code = getGroupSortedBlockData(pInfo->pCurrSortHandle, pInfo->binfo.pRes, pOperator->resultInfo.capacity,
pInfo->matchInfo.pList, pInfo, &pBlock); pInfo->matchInfo.pList, pInfo, &pBlock);
if (pBlock != NULL && (code == 0)) { if (pBlock != NULL && (code == 0)) {