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

fix issue

This commit is contained in:
54liuyao 2024-08-07 16:59:19 +08:00
parent 3b71401e96
commit 8ab78009a9
8 changed files with 17 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
source/libs/executor/src/aggregateoperator.c
View File

@ -340,6 +340,7 @@ int32_t doAggregateImpl(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx) {
static int32_t createDataBlockForEmptyInput(SOperatorInfo* pOperator, SSDataBlock** ppBlock) { static int32_t createDataBlockForEmptyInput(SOperatorInfo* pOperator, SSDataBlock** ppBlock) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0; int32_t lino = 0;
SSDataBlock* pBlock = NULL;
if (!tsCountAlwaysReturnValue) { if (!tsCountAlwaysReturnValue) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -363,7 +364,6 @@ static int32_t createDataBlockForEmptyInput(SOperatorInfo* pOperator, SSDataBloc
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
SSDataBlock* pBlock = NULL;
code = createDataBlock(&pBlock); code = createDataBlock(&pBlock);
if (code) { if (code) {
return code; return code;
@ -411,6 +411,7 @@ static int32_t createDataBlockForEmptyInput(SOperatorInfo* pOperator, SSDataBloc
_end: _end:
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
blockDataDestroy(pBlock);
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code)); qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
} }
return code; return code;

3
source/libs/executor/src/countwindowoperator.c
View File

@ -294,10 +294,11 @@ int32_t createCountwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* phy
SSDataBlock* pResBlock = createDataBlockFromDescNode(pCountWindowNode->window.node.pOutputDataBlockDesc); SSDataBlock* pResBlock = createDataBlockFromDescNode(pCountWindowNode->window.node.pOutputDataBlockDesc);
QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno); QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno);
initBasicInfo(&pInfo->binfo, pResBlock);
code = blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity); code = blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity);
QUERY_CHECK_CODE(code, lino, _error); QUERY_CHECK_CODE(code, lino, _error);
initBasicInfo(&pInfo->binfo, pResBlock);
initResultRowInfo(&pInfo->binfo.resultRowInfo); initResultRowInfo(&pInfo->binfo.resultRowInfo);
pInfo->binfo.inputTsOrder = physiNode->inputTsOrder; pInfo->binfo.inputTsOrder = physiNode->inputTsOrder;
pInfo->binfo.outputTsOrder = physiNode->outputTsOrder; pInfo->binfo.outputTsOrder = physiNode->outputTsOrder;

2
source/libs/executor/src/eventwindowoperator.c
View File

@ -110,11 +110,11 @@ int32_t createEventwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* phy
SSDataBlock* pResBlock = createDataBlockFromDescNode(pEventWindowNode->window.node.pOutputDataBlockDesc); SSDataBlock* pResBlock = createDataBlockFromDescNode(pEventWindowNode->window.node.pOutputDataBlockDesc);
QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno); QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno);
initBasicInfo(&pInfo->binfo, pResBlock);
code = blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity); code = blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity);
QUERY_CHECK_CODE(code, lino, _error); QUERY_CHECK_CODE(code, lino, _error);
initBasicInfo(&pInfo->binfo, pResBlock);
initResultRowInfo(&pInfo->binfo.resultRowInfo); initResultRowInfo(&pInfo->binfo.resultRowInfo);
pInfo->binfo.inputTsOrder = physiNode->inputTsOrder; pInfo->binfo.inputTsOrder = physiNode->inputTsOrder;
pInfo->binfo.outputTsOrder = physiNode->outputTsOrder; pInfo->binfo.outputTsOrder = physiNode->outputTsOrder;

4
source/libs/executor/src/exchangeoperator.c
View File

@ -688,6 +688,7 @@ void updateLoadRemoteInfo(SLoadRemoteDataInfo* pInfo, int64_t numOfRows, int32_t
int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, SArray* pColList, char** pNextStart) { int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, SArray* pColList, char** pNextStart) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0; int32_t lino = 0;
SSDataBlock* pBlock = NULL;
if (pColList == NULL) { // data from other sources if (pColList == NULL) { // data from other sources
blockDataCleanup(pRes); blockDataCleanup(pRes);
code = blockDecode(pRes, pData, (const char**) pNextStart); code = blockDecode(pRes, pData, (const char**) pNextStart);
@ -710,7 +711,7 @@ int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, SArray* pCo
pStart += sizeof(SSysTableSchema); pStart += sizeof(SSysTableSchema);
} }
SSDataBlock* pBlock = NULL; pBlock = NULL;
code = createDataBlock(&pBlock); code = createDataBlock(&pBlock);
QUERY_CHECK_CODE(code, lino, _end); QUERY_CHECK_CODE(code, lino, _end);
@ -739,6 +740,7 @@ int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, SArray* pCo
_end: _end:
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
blockDataDestroy(pBlock);
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code)); qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
} }
return code; return code;

6
source/libs/executor/src/executil.c
View File

@ -278,7 +278,7 @@ SSDataBlock* createDataBlockFromDescNode(SDataBlockDescNode* pNode) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code)); qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
blockDataDestroy(pBlock); blockDataDestroy(pBlock);
pBlock = NULL; pBlock = NULL;
terrno = code; terrno = TSDB_CODE_INVALID_PARA;
break; break;
} }
SColumnInfoData idata = SColumnInfoData idata =
@ -1094,7 +1094,7 @@ SSDataBlock* createTagValBlockForFilter(SArray* pColList, int32_t numOfTables, S
code = blockDataEnsureCapacity(pResBlock, numOfTables); code = blockDataEnsureCapacity(pResBlock, numOfTables);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
terrno = code; terrno = code;
taosMemoryFree(pResBlock); blockDataDestroy(pResBlock);
return NULL; return NULL;
} }
@ -1166,7 +1166,7 @@ SSDataBlock* createTagValBlockForFilter(SArray* pColList, int32_t numOfTables, S
_end: _end:
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
taosMemoryFree(pResBlock); blockDataDestroy(pResBlock);
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code)); qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
terrno = code; terrno = code;
return NULL; return NULL;

3
source/libs/executor/src/scanoperator.c
View File

@ -4279,11 +4279,12 @@ static int32_t tagScanFilterByTagCond(SArray* aUidTags, SNode* pTagCond, SArray*
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0; int32_t lino = 0;
int32_t numOfTables = taosArrayGetSize(aUidTags); int32_t numOfTables = taosArrayGetSize(aUidTags);
SArray* pBlockList = NULL;
SSDataBlock* pResBlock = createTagValBlockForFilter(pInfo->filterCtx.cInfoList, numOfTables, aUidTags, pVnode, pAPI); SSDataBlock* pResBlock = createTagValBlockForFilter(pInfo->filterCtx.cInfoList, numOfTables, aUidTags, pVnode, pAPI);
QUERY_CHECK_NULL(pResBlock, code, lino, _end, terrno); QUERY_CHECK_NULL(pResBlock, code, lino, _end, terrno);
SArray* pBlockList = taosArrayInit(1, POINTER_BYTES); pBlockList = taosArrayInit(1, POINTER_BYTES);
QUERY_CHECK_NULL(pBlockList, code, lino, _end, terrno); QUERY_CHECK_NULL(pBlockList, code, lino, _end, terrno);
void* tmp = taosArrayPush(pBlockList, &pResBlock); void* tmp = taosArrayPush(pBlockList, &pResBlock);

3
source/libs/executor/src/streamtimewindowoperator.c
View File

@ -5191,6 +5191,8 @@ int32_t createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode*
SSDataBlock* pResBlock = createDataBlockFromDescNode(pPhyNode->pOutputDataBlockDesc); SSDataBlock* pResBlock = createDataBlockFromDescNode(pPhyNode->pOutputDataBlockDesc);
QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno); QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno);
initBasicInfo(&pInfo->binfo, pResBlock);
pInfo->interval = (SInterval){ pInfo->interval = (SInterval){
.interval = pIntervalPhyNode->interval, .interval = pIntervalPhyNode->interval,
.sliding = pIntervalPhyNode->sliding, .sliding = pIntervalPhyNode->sliding,
@ -5218,7 +5220,6 @@ int32_t createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode*
SExprSupp* pSup = &pOperator->exprSupp; SExprSupp* pSup = &pOperator->exprSupp;
pSup->hasWindowOrGroup = true; pSup->hasWindowOrGroup = true;
initBasicInfo(&pInfo->binfo, pResBlock);
code = initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window); code = initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);
QUERY_CHECK_CODE(code, lino, _error); QUERY_CHECK_CODE(code, lino, _error);

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

@ -1698,6 +1698,7 @@ static int32_t initRowIdSort(SSortHandle* pHandle) {
biTs.compFn = getKeyComparFunc(TSDB_DATA_TYPE_TIMESTAMP, biTs.order); biTs.compFn = getKeyComparFunc(TSDB_DATA_TYPE_TIMESTAMP, biTs.order);
void* p = taosArrayPush(pOrderInfoList, &biTs); void* p = taosArrayPush(pOrderInfoList, &biTs);
if (p == NULL) { if (p == NULL) {
taosArrayDestroy(pOrderInfoList);
return terrno; return terrno;
} }
@ -1710,6 +1711,7 @@ static int32_t initRowIdSort(SSortHandle* pHandle) {
void* px = taosArrayPush(pOrderInfoList, &biPk); void* px = taosArrayPush(pOrderInfoList, &biPk);
if (px == NULL) { if (px == NULL) {
taosArrayDestroy(pOrderInfoList);
return terrno; return terrno;
} }
} }