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

fix(stream): fix the error when no agg tasks exist.

This commit is contained in:
Haojun Liao 2023-08-13 17:31:09 +08:00
parent 0d38f389ab
commit a81cc9aebf
1 changed files with 25 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
source/libs/stream/src/streamExec.c
View File

@ -509,7 +509,11 @@ int32_t streamProcessTranstateBlock(SStreamTask* pTask, SStreamDataBlock* pBlock
}
}
// dispatch the tran-state block to downstream task immediately
int32_t type = pTask->outputInfo.type;
// transfer the ownership of executor state
if (type == TASK_OUTPUT__FIXED_DISPATCH || type == TASK_OUTPUT__SHUFFLE_DISPATCH) {
if (level == TASK_LEVEL__SOURCE) {
qDebug("s-task:%s add transfer-state block into outputQ", id);
} else {
@ -517,10 +521,7 @@ int32_t streamProcessTranstateBlock(SStreamTask* pTask, SStreamDataBlock* pBlock
ASSERT(pTask->streamTaskId.taskId != 0 && pTask->info.fillHistory == 1);
}
// dispatch the tran-state block to downstream task immediately
int32_t type = pTask->outputInfo.type;
if ((level == TASK_LEVEL__AGG || level == TASK_LEVEL__SOURCE) &&
(type == TASK_OUTPUT__FIXED_DISPATCH || type == TASK_OUTPUT__SHUFFLE_DISPATCH)) {
if (level == TASK_LEVEL__AGG || level == TASK_LEVEL__SOURCE) {
pBlock->srcVgId = pTask->pMeta->vgId;
code = taosWriteQitem(pTask->outputInfo.queue->queue, pBlock);
if (code == 0) {
@ -529,6 +530,15 @@ int32_t streamProcessTranstateBlock(SStreamTask* pTask, SStreamDataBlock* pBlock
streamFreeQitem((SStreamQueueItem*)pBlock);
}
}
} else { // non-dispatch task, do task state transfer directly
qDebug("s-task:%s non-dispatch task, start to transfer state directly", id);
ASSERT(pTask->info.fillHistory == 1);
code = streamTransferStateToStreamTask(pTask);
if (code != TSDB_CODE_SUCCESS) {
atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);
}
}
return code;
}