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

fix(stream): recover the overwritten code

This commit is contained in:
54liuyao 2022-07-25 11:19:52 +08:00
parent 65d90a66a7
commit 86c4d34339
1 changed files with 7 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
source/libs/executor/src/executorimpl.c
View File

@ -3328,6 +3328,7 @@ static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
if (pLimitInfo->remainGroupOffset > 0) { if (pLimitInfo->remainGroupOffset > 0) {
if (pLimitInfo->currentGroupId == 0 || pLimitInfo->currentGroupId == pBlock->info.groupId) { // it is the first group if (pLimitInfo->currentGroupId == 0 || pLimitInfo->currentGroupId == pBlock->info.groupId) { // it is the first group
pLimitInfo->currentGroupId = pBlock->info.groupId; pLimitInfo->currentGroupId = pBlock->info.groupId;
ASSERT(pTaskInfo->execModel != OPTR_EXEC_MODEL_STREAM);
continue; continue;
} else if (pLimitInfo->currentGroupId != pBlock->info.groupId) { } else if (pLimitInfo->currentGroupId != pBlock->info.groupId) {
// now it is the data from a new group // now it is the data from a new group
@ -3336,6 +3337,7 @@ static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
// ignore data block in current group // ignore data block in current group
if (pLimitInfo->remainGroupOffset > 0) { if (pLimitInfo->remainGroupOffset > 0) {
ASSERT(pTaskInfo->execModel != OPTR_EXEC_MODEL_STREAM);
continue; continue;
} }
} }
@ -3380,10 +3382,12 @@ static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
if (pLimitInfo->remainOffset >= pInfo->pRes->info.rows) { if (pLimitInfo->remainOffset >= pInfo->pRes->info.rows) {
pLimitInfo->remainOffset -= pInfo->pRes->info.rows; pLimitInfo->remainOffset -= pInfo->pRes->info.rows;
blockDataCleanup(pInfo->pRes); blockDataCleanup(pInfo->pRes);
ASSERT(pTaskInfo->execModel != OPTR_EXEC_MODEL_STREAM);
continue; continue;
} else if (pLimitInfo->remainOffset < pInfo->pRes->info.rows && pLimitInfo->remainOffset > 0) { } else if (pLimitInfo->remainOffset < pInfo->pRes->info.rows && pLimitInfo->remainOffset > 0) {
blockDataTrimFirstNRows(pInfo->pRes, pLimitInfo->remainOffset); blockDataTrimFirstNRows(pInfo->pRes, pLimitInfo->remainOffset);
pLimitInfo->remainOffset = 0; pLimitInfo->remainOffset = 0;
ASSERT(pTaskInfo->execModel != OPTR_EXEC_MODEL_STREAM);
} }
// check for the limitation in each group // check for the limitation in each group
@ -3391,6 +3395,7 @@ static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
pLimitInfo->numOfOutputRows + pInfo->pRes->info.rows >= pLimitInfo->limit.limit) { pLimitInfo->numOfOutputRows + pInfo->pRes->info.rows >= pLimitInfo->limit.limit) {
int32_t keepRows = (int32_t)(pLimitInfo->limit.limit - pLimitInfo->numOfOutputRows); int32_t keepRows = (int32_t)(pLimitInfo->limit.limit - pLimitInfo->numOfOutputRows);
blockDataKeepFirstNRows(pInfo->pRes, keepRows); blockDataKeepFirstNRows(pInfo->pRes, keepRows);
ASSERT(pTaskInfo->execModel != OPTR_EXEC_MODEL_STREAM);
if (pLimitInfo->slimit.limit > 0 && pLimitInfo->slimit.limit <= pLimitInfo->numOfOutputGroups) { if (pLimitInfo->slimit.limit > 0 && pLimitInfo->slimit.limit <= pLimitInfo->numOfOutputGroups) {
pOperator->status = OP_EXEC_DONE; pOperator->status = OP_EXEC_DONE;
} }
@ -3412,7 +3417,8 @@ static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
// continue merge data, ignore the group id // continue merge data, ignore the group id
blockDataMerge(pFinalRes, pInfo->pRes); blockDataMerge(pFinalRes, pInfo->pRes);
if (pFinalRes->info.rows + pInfo->pRes->info.rows <= pOperator->resultInfo.threshold) { if (pFinalRes->info.rows + pInfo->pRes->info.rows <= pOperator->resultInfo.threshold &&
pTaskInfo->execModel != OPTR_EXEC_MODEL_STREAM) {
continue; continue;
} }
} }