diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h index 31d13d6cf4..960794792b 100644 --- a/include/libs/stream/tstream.h +++ b/include/libs/stream/tstream.h @@ -141,8 +141,7 @@ void* streamDataBlockDecode(const void* buf, SStreamDataBlock* pInput); #endif typedef struct { - int8_t parallelizable; - char* qmsg; + char* qmsg; // followings are not applicable to encoder and decoder void* inputHandle; void* executor; @@ -267,7 +266,7 @@ struct SStreamTask { // void* ahandle; }; -SStreamTask* tNewSStreamTask(int64_t streamId, int32_t childId); +SStreamTask* tNewSStreamTask(int64_t streamId); int32_t tEncodeSStreamTask(SEncoder* pEncoder, const SStreamTask* pTask); int32_t tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask); void tFreeSStreamTask(SStreamTask* pTask); diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c index 55ff34f900..d07355dfa9 100644 --- a/source/dnode/mnode/impl/src/mndScheduler.c +++ b/source/dnode/mnode/impl/src/mndScheduler.c @@ -197,7 +197,7 @@ int32_t mndAddShuffledSinkToStream(SMnode* pMnode, STrans* pTrans, SStreamObj* p sdbRelease(pSdb, pVgroup); continue; } - SStreamTask* pTask = tNewSStreamTask(pStream->uid, 0); + SStreamTask* pTask = tNewSStreamTask(pStream->uid); if (pTask == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; @@ -237,7 +237,7 @@ int32_t mndAddShuffledSinkToStream(SMnode* pMnode, STrans* pTrans, SStreamObj* p int32_t mndAddFixedSinkToStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { ASSERT(pStream->fixedSinkVgId != 0); SArray* tasks = taosArrayGetP(pStream->tasks, 0); - SStreamTask* pTask = tNewSStreamTask(pStream->uid, 0); + SStreamTask* pTask = tNewSStreamTask(pStream->uid); if (pTask == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; @@ -329,7 +329,8 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { sdbRelease(pSdb, pVgroup); continue; } - SStreamTask* pTask = tNewSStreamTask(pStream->uid, 0); + SStreamTask* pTask = tNewSStreamTask(pStream->uid); + mndAddTaskToTaskSet(taskOneLevel, pTask); // source part pTask->sourceType = TASK_SOURCE__SCAN; pTask->inputType = TASK_INPUT_TYPE__SUMBIT_BLOCK; @@ -378,14 +379,12 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { // exec part pTask->execType = TASK_EXEC__PIPE; - pTask->exec.parallelizable = 1; if (mndAssignTaskToVg(pMnode, pTrans, pTask, plan, pVgroup) < 0) { sdbRelease(pSdb, pVgroup); qDestroyQueryPlan(pPlan); return -1; } sdbRelease(pSdb, pVgroup); - mndAddTaskToTaskSet(taskOneLevel, pTask); } } else { // merge plan @@ -394,7 +393,8 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { // else, assign to vnode ASSERT(plan->subplanType == SUBPLAN_TYPE_MERGE); - SStreamTask* pTask = tNewSStreamTask(pStream->uid, 0); + SStreamTask* pTask = tNewSStreamTask(pStream->uid); + mndAddTaskToTaskSet(taskOneLevel, pTask); // source part, currently only support multi source pTask->sourceType = TASK_SOURCE__PIPE; @@ -456,7 +456,6 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { // exec part pTask->execType = TASK_EXEC__MERGE; - pTask->exec.parallelizable = 0; SVgObj* pVgroup = mndSchedFetchOneVg(pMnode, pStream->dbUid); ASSERT(pVgroup); if (mndAssignTaskToVg(pMnode, pTrans, pTask, plan, pVgroup) < 0) { @@ -465,12 +464,12 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { return -1; } sdbRelease(pSdb, pVgroup); - taosArrayPush(taskOneLevel, &pTask); } taosArrayPush(pStream->tasks, &taskOneLevel); } +#if 0 if (totLevel == 2) { void* pIter = NULL; while (1) { @@ -481,7 +480,7 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { sdbRelease(pSdb, pVgroup); continue; } - SStreamTask* pTask = tNewSStreamTask(pStream->uid, 0); + SStreamTask* pTask = tNewSStreamTask(pStream->uid); // source part pTask->sourceType = TASK_SOURCE__MERGE; @@ -495,9 +494,9 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { // exec part pTask->execType = TASK_EXEC__NONE; - pTask->exec.parallelizable = 0; } } +#endif // free memory qDestroyQueryPlan(pPlan); diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index a1b8d81d58..2f4d6c11c6 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -364,6 +364,7 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) { tdGetSTSChemaFromSSChema(&pTask->tbSink.pSchemaWrapper->pSchema, pTask->tbSink.pSchemaWrapper->nCols); ASSERT(pTask->tbSink.pTSchema); } + tqInfo("deploy stream task id %d child id %d on vg %d", pTask->taskId, pTask->childId, pTq->pVnode->config.vgId); taosHashPut(pTq->pStreamTasks, &pTask->taskId, sizeof(int32_t), pTask, sizeof(SStreamTask)); diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c index 38a3509e7b..6fb5d75d86 100644 --- a/source/libs/stream/src/streamDispatch.c +++ b/source/libs/stream/src/streamDispatch.c @@ -100,6 +100,7 @@ int32_t streamBuildDispatchMsg(SStreamTask* pTask, SStreamDataBlock* data, SRpcM .upstreamNodeId = pTask->nodeId, .blockNum = blockNum, }; + qInfo("dispatch from task %d (child id %d)", pTask->taskId, pTask->childId); req.data = taosArrayInit(blockNum, sizeof(void*)); req.dataLen = taosArrayInit(blockNum, sizeof(int32_t)); diff --git a/source/libs/stream/src/streamTask.c b/source/libs/stream/src/streamTask.c index 890ae1f3b5..5d8546bcb9 100644 --- a/source/libs/stream/src/streamTask.c +++ b/source/libs/stream/src/streamTask.c @@ -16,14 +16,13 @@ #include "executor.h" #include "tstream.h" -SStreamTask* tNewSStreamTask(int64_t streamId, int32_t childId) { +SStreamTask* tNewSStreamTask(int64_t streamId) { SStreamTask* pTask = (SStreamTask*)taosMemoryCalloc(1, sizeof(SStreamTask)); if (pTask == NULL) { return NULL; } pTask->taskId = tGenIdPI32(); pTask->streamId = streamId; - pTask->childId = childId; pTask->status = TASK_STATUS__IDLE; pTask->inputStatus = TASK_INPUT_STATUS__NORMAL; pTask->outputStatus = TASK_OUTPUT_STATUS__NORMAL; @@ -48,7 +47,6 @@ int32_t tEncodeSStreamTask(SEncoder* pEncoder, const SStreamTask* pTask) { if (tEncodeSEpSet(pEncoder, &pTask->epSet) < 0) return -1; if (pTask->execType != TASK_EXEC__NONE) { - if (tEncodeI8(pEncoder, pTask->exec.parallelizable) < 0) return -1; if (tEncodeCStr(pEncoder, pTask->exec.qmsg) < 0) return -1; } @@ -96,7 +94,6 @@ int32_t tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask) { if (tDecodeSEpSet(pDecoder, &pTask->epSet) < 0) return -1; if (pTask->execType != TASK_EXEC__NONE) { - if (tDecodeI8(pDecoder, &pTask->exec.parallelizable) < 0) return -1; if (tDecodeCStrAlloc(pDecoder, &pTask->exec.qmsg) < 0) return -1; }