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

Merge pull request #27564 from taosdata/fix/3_liaohj 

fix(query): fix memory leak.
This commit is contained in:
Haojun Liao 2024-08-30 15:33:02 +08:00 committed by GitHub
parent 5126cd1fe0 82ad1958f4
commit 912d987184
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 37 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -254,17 +254,17 @@ static int32_t getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock,
SSortOperatorInfo* pInfo, SSDataBlock** pResBlock) {
QRY_OPTR_CHECK(pResBlock);
blockDataCleanup(pDataBlock);
int32_t lino = 0;
int32_t code = 0;
SSDataBlock* p = NULL;
int32_t code = tsortGetSortedDataBlock(pHandle, &p);
code = tsortGetSortedDataBlock(pHandle, &p);
if (p == NULL || (code != 0)) {
return code;
}
code = blockDataEnsureCapacity(p, capacity);
if (code) {
return code;
}
QUERY_CHECK_CODE(code, lino, _error);
STupleHandle* pTupleHandle;
while (1) {
@ -273,51 +273,40 @@ static int32_t getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock,
} else {
code = tsortNextTuple(pHandle, &pTupleHandle);
}
if (pTupleHandle == NULL || code != 0) {
lino = __LINE__;
break;
}
code = appendOneRowToDataBlock(p, pTupleHandle);
if (code) {
return code;
}
QUERY_CHECK_CODE(code, lino, _error);
if (p->info.rows >= capacity) {
break;
}
}
if (TSDB_CODE_SUCCESS != code) {
return code;
}
QUERY_CHECK_CODE(code, lino, _error);
if (p->info.rows > 0) {
code = blockDataEnsureCapacity(pDataBlock, capacity);
if (code) {
return code;
}
QUERY_CHECK_CODE(code, lino, _error);
// todo extract function to handle this
int32_t numOfCols = taosArrayGetSize(pColMatchInfo);
for (int32_t i = 0; i < numOfCols; ++i) {
SColMatchItem* pmInfo = taosArrayGet(pColMatchInfo, i);
if (pmInfo == NULL) {
return terrno;
}
QUERY_CHECK_NULL(pmInfo, code, lino, _error, terrno);
SColumnInfoData* pSrc = taosArrayGet(p->pDataBlock, pmInfo->srcSlotId);
if (pSrc == NULL) {
return terrno;
}
QUERY_CHECK_NULL(pSrc, code, lino, _error, terrno);
SColumnInfoData* pDst = taosArrayGet(pDataBlock->pDataBlock, pmInfo->dstSlotId);
if (pDst == NULL) {
return terrno;
}
QUERY_CHECK_NULL(pDst, code, lino, _error, terrno);
code = colDataAssign(pDst, pSrc, p->info.rows, &pDataBlock->info);
if (code) {
return code;
}
QUERY_CHECK_CODE(code, lino, _error);
}
pDataBlock->info.dataLoad = 1;
@ -329,6 +318,12 @@ static int32_t getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock,
blockDataDestroy(p);
*pResBlock = (pDataBlock->info.rows > 0) ? pDataBlock : NULL;
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) {
@ -384,13 +379,13 @@ int32_t doOpenSortOperator(SOperatorInfo* pOperator) {
code = tsortOpen(pInfo->pSortHandle);
if (code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, code);
pTaskInfo->code = code;
} else {
pOperator->cost.openCost = (taosGetTimestampUs() - pInfo->startTs) / 1000.0;
pOperator->status = OP_RES_TO_RETURN;
OPTR_SET_OPENED(pOperator);
}
pOperator->cost.openCost = (taosGetTimestampUs() - pInfo->startTs) / 1000.0;
pOperator->status = OP_RES_TO_RETURN;
OPTR_SET_OPENED(pOperator);
return code;
}
@ -735,6 +730,7 @@ int32_t doGroupSort(SOperatorInfo* pOperator, SSDataBlock** pResBlock) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
T_LONG_JMP(pOperator->pTaskInfo->env, code);
}
code = getGroupSortedBlockData(pInfo->pCurrSortHandle, pInfo->binfo.pRes, pOperator->resultInfo.capacity,
pInfo->matchInfo.pList, pInfo, &pBlock);
if (pBlock != NULL && (code == 0)) {

31
source/libs/executor/src/tsort.c
View File

@ -286,23 +286,19 @@ int32_t tsortCreateSortHandle(SArray* pSortInfo, int32_t type, int32_t pageSize,
SSDataBlock* pBlock, const char* idstr, uint64_t pqMaxRows, uint32_t pqMaxTupleLength,
uint32_t pqSortBufSize, SSortHandle** pHandle) {
int32_t code = 0;
*pHandle = NULL;
int32_t lino = 0;
QRY_OPTR_CHECK(pHandle);
SSortHandle* pSortHandle = taosMemoryCalloc(1, sizeof(SSortHandle));
if (pSortHandle == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
QUERY_CHECK_NULL(pSortHandle, code, lino, _err, terrno);
pSortHandle->type = type;
pSortHandle->pageSize = pageSize;
pSortHandle->numOfPages = numOfPages;
pSortHandle->pSortInfo = taosArrayDup(pSortInfo, NULL);
if (pSortHandle->pSortInfo == NULL) {
return terrno;
}
QUERY_CHECK_NULL(pSortHandle->pSortInfo, code, lino, _err, terrno);
pSortHandle->loops = 0;
pSortHandle->pqMaxTupleLength = pqMaxTupleLength;
if (pqMaxRows != 0) {
pSortHandle->pqSortBufSize = pqSortBufSize;
@ -312,18 +308,13 @@ int32_t tsortCreateSortHandle(SArray* pSortInfo, int32_t type, int32_t pageSize,
pSortHandle->forceUsePQSort = false;
if (pBlock != NULL) {
code = createOneDataBlock(pBlock, false, &pSortHandle->pDataBlock);
if (code) {
goto _err;
}
QUERY_CHECK_CODE(code, lino, _err);
}
pSortHandle->mergeLimit = -1;
pSortHandle->pOrderedSource = taosArrayInit(4, POINTER_BYTES);
if (pSortHandle->pOrderedSource == NULL) {
code = terrno;
goto _err;
}
QUERY_CHECK_NULL(pSortHandle->pOrderedSource, code, lino, _err, terrno);
pSortHandle->cmpParam.orderInfo = pSortInfo;
pSortHandle->cmpParam.cmpGroupId = false;
@ -346,17 +337,17 @@ int32_t tsortCreateSortHandle(SArray* pSortInfo, int32_t type, int32_t pageSize,
if (idstr != NULL) {
pSortHandle->idStr = taosStrdup(idstr);
if (pSortHandle->idStr == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
QUERY_CHECK_NULL(pSortHandle->idStr, code, lino, _err, terrno);
}
*pHandle = pSortHandle;
return code;
_err:
tsortDestroySortHandle(pSortHandle);
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
if (pSortHandle) {
tsortDestroySortHandle(pSortHandle);
}
return code;
}