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

fix: keep processing on tqNextBlock failure

This commit is contained in:
Benguang Zhao 2023-02-17 09:16:44 +08:00
parent 2a08e2a549
commit dea3c0285b
1 changed files with 2 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1630,13 +1630,10 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
SFetchRet ret = {0};
if (tqNextBlock(pInfo->tqReader, &ret) < 0) {
qError("failed to get next log block since %s", terrstr());
return NULL;
}
if (ret.fetchType == FETCH_TYPE__DATA) {
blockDataCleanup(pInfo->pRes);
if (setBlockIntoRes(pInfo, &ret.data, true) < 0) {
return NULL;
}
setBlockIntoRes(pInfo, &ret.data, true);
if (pInfo->pRes->info.rows > 0) {
pOperator->status = OP_EXEC_RECV;
qDebug("queue scan log return %d rows", pInfo->pRes->info.rows);
@ -1644,7 +1641,7 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
}
} else if (ret.fetchType == FETCH_TYPE__META) {
qError("unexpected ret.fetchType:%d", ret.fetchType);
return NULL;
continue;
// pTaskInfo->streamInfo.lastStatus = ret.offset;
// pTaskInfo->streamInfo.metaBlk = ret.meta;
// return NULL;