fix(stream): fix error in extract data from inputQ.

This commit is contained in:
Haojun Liao 2023-07-03 15:33:08 +08:00
parent ceb06635fe
commit 9e62b9d0d2
1 changed files with 3 additions and 2 deletions

View File

@ -347,7 +347,7 @@ static int32_t extractMsgFromInputQ(SStreamTask* pTask, SStreamQueueItem** pInpu
return TSDB_CODE_SUCCESS;
}
if (pInput == NULL) {
if (*pInput == NULL) {
ASSERT((*numOfBlocks) == 0);
*pInput = qItem;
} else {
@ -358,7 +358,8 @@ static int32_t extractMsgFromInputQ(SStreamTask* pTask, SStreamQueueItem** pInpu
streamQueueProcessFail(pTask->inputQueue);
return TSDB_CODE_SUCCESS;
}
pInput = newRet;
*pInput = newRet;
}
*numOfBlocks += 1;