From 5524d42c2d8dd1fc353d8d07182e6bb5ba74ee9d Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Tue, 6 Aug 2024 10:19:22 +0800 Subject: [PATCH] fix: check return value and fix bugs. --- source/dnode/vnode/src/tsdb/tsdbRead2.c | 44 ++++++++++-------- source/libs/executor/src/cachescanoperator.c | 2 +- source/libs/executor/src/operator.c | 45 +++++++++++++++---- source/libs/executor/src/projectoperator.c | 1 + source/libs/executor/src/querytask.c | 7 ++- source/libs/executor/src/scanoperator.c | 21 +++++---- source/libs/executor/src/sortoperator.c | 5 +++ .../executor/src/streamtimewindowoperator.c | 4 +- source/libs/executor/src/sysscanoperator.c | 3 ++ source/libs/executor/src/timewindowoperator.c | 16 ++----- 10 files changed, 90 insertions(+), 58 deletions(-) diff --git a/source/dnode/vnode/src/tsdb/tsdbRead2.c b/source/dnode/vnode/src/tsdb/tsdbRead2.c index 51e3672c56..661ab3e92e 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead2.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead2.c @@ -5762,6 +5762,7 @@ int32_t tsdbTakeReadSnap2(STsdbReader* pReader, _query_reseek_func_t reseek, STs int32_t code = 0; STsdb* pTsdb = pReader->pTsdb; SVersionRange* pRange = &pReader->info.verRange; + *ppSnap = NULL; // lock code = taosThreadMutexLock(&pTsdb->mutex); @@ -5774,7 +5775,7 @@ int32_t tsdbTakeReadSnap2(STsdbReader* pReader, _query_reseek_func_t reseek, STs STsdbReadSnap* pSnap = (STsdbReadSnap*)taosMemoryCalloc(1, sizeof(STsdbReadSnap)); if (pSnap == NULL) { (void) taosThreadMutexUnlock(&pTsdb->mutex); - code = TSDB_CODE_OUT_OF_MEMORY; + code = terrno; goto _exit; } @@ -5784,7 +5785,7 @@ int32_t tsdbTakeReadSnap2(STsdbReader* pReader, _query_reseek_func_t reseek, STs pSnap->pNode = taosMemoryMalloc(sizeof(*pSnap->pNode)); if (pSnap->pNode == NULL) { (void) taosThreadMutexUnlock(&pTsdb->mutex); - code = TSDB_CODE_OUT_OF_MEMORY; + code = terrno; goto _exit; } @@ -5798,8 +5799,10 @@ int32_t tsdbTakeReadSnap2(STsdbReader* pReader, _query_reseek_func_t reseek, STs pSnap->pIMem = pTsdb->imem; pSnap->pINode = taosMemoryMalloc(sizeof(*pSnap->pINode)); if (pSnap->pINode == NULL) { + tsdbUnrefMemTable(pTsdb->mem, pSnap->pNode, true); // unref the previous refed mem + code = terrno; + (void) taosThreadMutexUnlock(&pTsdb->mutex); - code = TSDB_CODE_OUT_OF_MEMORY; goto _exit; } @@ -5811,28 +5814,33 @@ int32_t tsdbTakeReadSnap2(STsdbReader* pReader, _query_reseek_func_t reseek, STs // fs code = tsdbFSCreateRefSnapshotWithoutLock(pTsdb->pFS, &pSnap->pfSetArray); + if (code) { + if (pSnap->pNode) { + tsdbUnrefMemTable(pTsdb->mem, pSnap->pNode, true); // unref the previous refed mem + } + + if (pSnap->pINode) { + tsdbUnrefMemTable(pTsdb->imem, pSnap->pINode, true); + } + + (void) taosThreadMutexUnlock(&pTsdb->mutex); + goto _exit; + } // unlock (void) taosThreadMutexUnlock(&pTsdb->mutex); + *ppSnap = pSnap; - if (code == TSDB_CODE_SUCCESS) { - tsdbTrace("vgId:%d, take read snapshot", TD_VID(pTsdb->pVnode)); - } + tsdbTrace("vgId:%d, take read snapshot", TD_VID(pTsdb->pVnode)); + return code; _exit: - if (code != TSDB_CODE_SUCCESS) { - tsdbError("vgId:%d take read snapshot failed, code:%s", TD_VID(pTsdb->pVnode), tstrerror(code)); - - *ppSnap = NULL; - if (pSnap) { - if (pSnap->pNode) taosMemoryFree(pSnap->pNode); - if (pSnap->pINode) taosMemoryFree(pSnap->pINode); - taosMemoryFree(pSnap); - } - } else { - *ppSnap = pSnap; + tsdbError("vgId:%d take read snapshot failed, code:%s", TD_VID(pTsdb->pVnode), tstrerror(code)); + if (pSnap) { + if (pSnap->pNode) taosMemoryFree(pSnap->pNode); + if (pSnap->pINode) taosMemoryFree(pSnap->pINode); + taosMemoryFree(pSnap); } - return code; } diff --git a/source/libs/executor/src/cachescanoperator.c b/source/libs/executor/src/cachescanoperator.c index 73f770a599..0752f4028c 100644 --- a/source/libs/executor/src/cachescanoperator.c +++ b/source/libs/executor/src/cachescanoperator.c @@ -98,7 +98,6 @@ int32_t createCacherowsScanOperator(SLastRowScanPhysiNode* pScanNode, SReadHandl if (pInfo == NULL || pOperator == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; - tableListDestroy(pTableListInfo); goto _error; } @@ -227,6 +226,7 @@ _error: if (code != TSDB_CODE_SUCCESS) { qError("%s failed at line %d since %s", __func__, lino, tstrerror(code)); } + pInfo->pTableList = NULL; destroyCacheScanOperator(pInfo); taosMemoryFree(pOperator); return code; diff --git a/source/libs/executor/src/operator.c b/source/libs/executor/src/operator.c index c204494710..c00afefd15 100644 --- a/source/libs/executor/src/operator.c +++ b/source/libs/executor/src/operator.c @@ -333,10 +333,10 @@ int32_t createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHand pTaskInfo->cost.pRecoder = &pScanInfo->base.readRecorder; } else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN == type) { STableMergeScanPhysiNode* pTableScanNode = (STableMergeScanPhysiNode*)pPhyNode; - STableListInfo* pTableListInfo = tableListCreate(); - code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, true, pHandle, - pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo); + STableListInfo* pTableListInfo = tableListCreate(); + code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, true, pHandle, pTableListInfo, + pTagCond, pTagIndexCond, pTaskInfo); if (code) { pTaskInfo->code = code; tableListDestroy(pTableListInfo); @@ -366,7 +366,7 @@ int32_t createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHand } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN == type) { STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode; STableListInfo* pTableListInfo = tableListCreate(); - if (pTableListInfo == NULL){ + if (pTableListInfo == NULL) { pTaskInfo->code = terrno; return terrno; } @@ -384,6 +384,11 @@ int32_t createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHand //pTaskInfo->schemaInfo.qsw = extractQueriedColumnSchema(&pTableScanNode->scan); code = createStreamScanOperatorInfo(pHandle, pTableScanNode, pTagCond, pTableListInfo, pTaskInfo, &pOperator); + if (code) { + pTaskInfo->code = code; + tableListDestroy(pTableListInfo); + return code; + } } else if (QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN == type) { SSystemTableScanPhysiNode* pSysScanPhyNode = (SSystemTableScanPhysiNode*)pPhyNode; code = createSysTableScanOperatorInfo(pHandle, pSysScanPhyNode, pUser, pTaskInfo, &pOperator); @@ -404,10 +409,17 @@ int32_t createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHand if (code != TSDB_CODE_SUCCESS) { pTaskInfo->code = code; qError("failed to getTableList, code: %s", tstrerror(code)); + tableListDestroy(pTableListInfo); return code; } } - code = createTagScanOperatorInfo(pHandle, pTagScanPhyNode, pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo, &pOperator); + code = createTagScanOperatorInfo(pHandle, pTagScanPhyNode, pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo, + &pOperator); + if (code) { + pTaskInfo->code = code; + tableListDestroy(pTableListInfo); + return code; + } } else if (QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN == type) { SBlockDistScanPhysiNode* pBlockNode = (SBlockDistScanPhysiNode*)pPhyNode; STableListInfo* pTableListInfo = tableListCreate(); @@ -422,6 +434,7 @@ int32_t createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHand if (code != TSDB_CODE_SUCCESS) { pTaskInfo->code = code; taosArrayDestroy(pList); + tableListDestroy(pTableListInfo); return code; } @@ -429,13 +442,14 @@ int32_t createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHand for (int32_t i = 0; i < num; ++i) { uint64_t* id = taosArrayGet(pList, i); if (id == NULL) { - pTaskInfo->code = terrno; - return terrno; + continue; } code = tableListAddTableInfo(pTableListInfo, *id, 0); if (code) { pTaskInfo->code = code; + tableListDestroy(pTableListInfo); + taosArrayDestroy(pList); return code; } } @@ -445,11 +459,17 @@ int32_t createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHand code = tableListAddTableInfo(pTableListInfo, pBlockNode->uid, 0); if (code) { pTaskInfo->code = code; + tableListDestroy(pTableListInfo); return code; } } code = createDataBlockInfoScanOperator(pHandle, pBlockNode, pTableListInfo, pTaskInfo, &pOperator); + if (code) { + pTaskInfo->code = code; + tableListDestroy(pTableListInfo); + return code; + } } else if (QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN == type) { SLastRowScanPhysiNode* pScanNode = (SLastRowScanPhysiNode*)pPhyNode; STableListInfo* pTableListInfo = tableListCreate(); @@ -458,20 +478,27 @@ int32_t createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHand return terrno; } - code = createScanTableListInfo(&pScanNode->scan, pScanNode->pGroupTags, true, pHandle, pTableListInfo, - pTagCond, pTagIndexCond, pTaskInfo); + code = createScanTableListInfo(&pScanNode->scan, pScanNode->pGroupTags, true, pHandle, pTableListInfo, pTagCond, + pTagIndexCond, pTaskInfo); if (code != TSDB_CODE_SUCCESS) { pTaskInfo->code = code; + tableListDestroy(pTableListInfo); return code; } code = initQueriedTableSchemaInfo(pHandle, &pScanNode->scan, dbname, pTaskInfo); if (code != TSDB_CODE_SUCCESS) { pTaskInfo->code = code; + tableListDestroy(pTableListInfo); return code; } code = createCacherowsScanOperator(pScanNode, pHandle, pTableListInfo, pTaskInfo, &pOperator); + if (code) { + tableListDestroy(pTableListInfo); + pTaskInfo->code = code; + return code; + } } else if (QUERY_NODE_PHYSICAL_PLAN_PROJECT == type) { code = createProjectOperatorInfo(NULL, (SProjectPhysiNode*)pPhyNode, pTaskInfo, &pOperator); } else { diff --git a/source/libs/executor/src/projectoperator.c b/source/libs/executor/src/projectoperator.c index 7185f74254..899bd72fff 100644 --- a/source/libs/executor/src/projectoperator.c +++ b/source/libs/executor/src/projectoperator.c @@ -485,6 +485,7 @@ int32_t createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhysiNode* } SSDataBlock* pResBlock = createDataBlockFromDescNode(pPhyNode->node.pOutputDataBlockDesc); + TSDB_CHECK_NULL(pInfo->binfo.pRes, code, lino, _error, terrno); // Make sure the size of SSDataBlock will never exceed the size of 2MB. int32_t TWOMB = 2 * 1024 * 1024; diff --git a/source/libs/executor/src/querytask.c b/source/libs/executor/src/querytask.c index 6f0bc40d29..5c4d3fe009 100644 --- a/source/libs/executor/src/querytask.c +++ b/source/libs/executor/src/querytask.c @@ -136,8 +136,7 @@ void cleanupQueriedTableScanInfo(void* p) { int32_t initQueriedTableSchemaInfo(SReadHandle* pHandle, SScanPhysiNode* pScanNode, const char* dbName, SExecTaskInfo* pTaskInfo) { SMetaReader mr = {0}; if (pHandle == NULL) { - terrno = TSDB_CODE_INVALID_PARA; - return terrno; + return TSDB_CODE_INVALID_PARA; } SStorageAPI* pAPI = &pTaskInfo->storageAPI; @@ -149,7 +148,7 @@ int32_t initQueriedTableSchemaInfo(SReadHandle* pHandle, SScanPhysiNode* pScanNo GET_TASKID(pTaskInfo)); pAPI->metaReaderFn.clearReader(&mr); - return terrno; + return code; } SSchemaInfo schemaInfo = {0}; @@ -169,7 +168,7 @@ int32_t initQueriedTableSchemaInfo(SReadHandle* pHandle, SScanPhysiNode* pScanNo pAPI->metaReaderFn.clearReader(&mr); taosMemoryFree(schemaInfo.tablename); taosMemoryFree(schemaInfo.dbname); - return terrno; + return code; } schemaInfo.sw = tCloneSSchemaWrapper(&mr.me.stbEntry.schemaRow); diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 37c5bc1aeb..f89cd7af44 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -1418,6 +1418,7 @@ int32_t createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHa _error: if (pInfo != NULL) { + pInfo->base.pTableListInfo = NULL; // this attribute will be destroy outside of this function destroyTableScanOperatorInfo(pInfo); } @@ -3876,7 +3877,6 @@ int32_t createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* if (pInfo == NULL || pOperator == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; - tableListDestroy(pTableListInfo); goto _error; } @@ -3889,7 +3889,6 @@ int32_t createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* int32_t numOfCols = 0; code = extractColMatchInfo(pScanPhyNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID, &pInfo->matchInfo); if (code != TSDB_CODE_SUCCESS) { - tableListDestroy(pTableListInfo); goto _error; } @@ -3919,8 +3918,7 @@ int32_t createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* if (pTableScanNode->pSubtable != NULL) { SExprInfo* pSubTableExpr = taosMemoryCalloc(1, sizeof(SExprInfo)); if (pSubTableExpr == NULL) { - terrno = TSDB_CODE_OUT_OF_MEMORY; - tableListDestroy(pTableListInfo); + code = terrno; goto _error; } @@ -3929,7 +3927,6 @@ int32_t createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* QUERY_CHECK_CODE(code, lino, _error); if (initExprSupp(&pInfo->tbnameCalSup, pSubTableExpr, 1, &pTaskInfo->storageAPI.functionStore) != 0) { - tableListDestroy(pTableListInfo); goto _error; } } @@ -3939,20 +3936,17 @@ int32_t createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* SExprInfo* pTagExpr = createExpr(pTableScanNode->pTags, &numOfTags); if (pTagExpr == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; - tableListDestroy(pTableListInfo); goto _error; } if (initExprSupp(&pInfo->tagCalSup, pTagExpr, numOfTags, &pTaskInfo->storageAPI.functionStore) != 0) { - terrno = TSDB_CODE_OUT_OF_MEMORY; - tableListDestroy(pTableListInfo); + code = TSDB_CODE_OUT_OF_MEMORY; goto _error; } } pInfo->pBlockLists = taosArrayInit(4, sizeof(SPackedData)); if (pInfo->pBlockLists == NULL) { - terrno = TSDB_CODE_OUT_OF_MEMORY; - tableListDestroy(pTableListInfo); + code = terrno; goto _error; } @@ -4636,6 +4630,10 @@ int32_t createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* p return code; _error: + if (pInfo) { + pInfo->pTableListInfo = NULL; + } + taosMemoryFree(pInfo); taosMemoryFree(pOperator); return code; @@ -5839,7 +5837,8 @@ int32_t createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SR return code; _error: - pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY; + pTaskInfo->code = code; + pInfo->base.pTableListInfo = NULL; taosMemoryFree(pInfo); taosMemoryFree(pOperator); return code; diff --git a/source/libs/executor/src/sortoperator.c b/source/libs/executor/src/sortoperator.c index 858f26ad18..e397a46513 100644 --- a/source/libs/executor/src/sortoperator.c +++ b/source/libs/executor/src/sortoperator.c @@ -101,7 +101,10 @@ int32_t createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortN } pInfo->binfo.pRes = createDataBlockFromDescNode(pDescNode); + TSDB_CHECK_NULL(pInfo->binfo.pRes, code, lino, _error, terrno); + pInfo->pSortInfo = createSortInfo(pSortNode->pSortKeys); + TSDB_CHECK_NULL(pInfo->pSortInfo, code, lino, _error, terrno); if (pSortNode->calcGroupId) { int32_t keyLen; @@ -788,6 +791,8 @@ int32_t createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSortPhysiNo QUERY_CHECK_NULL(pOperator->exprSupp.pCtx, code, lino, _error, terrno); pInfo->binfo.pRes = createDataBlockFromDescNode(pDescNode); + TSDB_CHECK_NULL(pInfo->binfo.pRes, code, lino, _error, terrno); + code = blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity); TSDB_CHECK_CODE(code, lino, _error); diff --git a/source/libs/executor/src/streamtimewindowoperator.c b/source/libs/executor/src/streamtimewindowoperator.c index e95ffeb658..563b78144f 100644 --- a/source/libs/executor/src/streamtimewindowoperator.c +++ b/source/libs/executor/src/streamtimewindowoperator.c @@ -3738,9 +3738,7 @@ int32_t createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode SSDataBlock* pResBlock = createDataBlockFromDescNode(pPhyNode->pOutputDataBlockDesc); QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno); code = initBasicInfoEx(&pInfo->binfo, pExpSup, pExprInfo, numOfCols, pResBlock, &pTaskInfo->storageAPI.functionStore); - if (code != TSDB_CODE_SUCCESS) { - goto _error; - } + QUERY_CHECK_CODE(code, lino, _error); pInfo->twAggSup = (STimeWindowAggSupp){ .waterMark = pSessionNode->window.watermark, diff --git a/source/libs/executor/src/sysscanoperator.c b/source/libs/executor/src/sysscanoperator.c index 14fda685c0..5c57f1889b 100644 --- a/source/libs/executor/src/sysscanoperator.c +++ b/source/libs/executor/src/sysscanoperator.c @@ -2748,6 +2748,9 @@ int32_t createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDistScanP return code; _error: + if (pInfo) { + pInfo->pTableListInfo = NULL; + } taosMemoryFreeClear(pInfo); taosMemoryFreeClear(pOperator); return code; diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index 7a5ab58331..a8423301cb 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -1706,9 +1706,7 @@ int32_t createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionWinodwPh code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str, pTaskInfo->streamInfo.pState, &pTaskInfo->storageAPI.functionStore); - if (code != TSDB_CODE_SUCCESS) { - goto _error; - } + QUERY_CHECK_CODE(code, lino, _error); pInfo->twAggSup.waterMark = pSessionNode->window.watermark; pInfo->twAggSup.calTrigger = pSessionNode->window.triggerType; @@ -1732,15 +1730,11 @@ int32_t createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionWinodwPh QUERY_CHECK_CODE(code, lino, _error); code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar, &pTaskInfo->storageAPI.functionStore); - if (code != TSDB_CODE_SUCCESS) { - goto _error; - } + QUERY_CHECK_CODE(code, lino, _error); } code = filterInitFromNode((SNode*)pSessionNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0); - if (code != TSDB_CODE_SUCCESS) { - goto _error; - } + QUERY_CHECK_CODE(code, lino, _error); setOperatorInfo(pOperator, "SessionWindowAggOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION, true, OP_NOT_OPENED, pInfo, pTaskInfo); @@ -1748,9 +1742,7 @@ int32_t createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionWinodwPh optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL); pOperator->pTaskInfo = pTaskInfo; code = appendDownstream(pOperator, &downstream, 1); - if (code != TSDB_CODE_SUCCESS) { - goto _error; - } + QUERY_CHECK_CODE(code, lino, _error); *pOptrInfo = pOperator; return code;