fix(stream): set child id

include/libs/stream/tstream.h

@@ -141,7 +141,6 @@ void*   streamDataBlockDecode(const void* buf, SStreamDataBlock* pInput);
 #endif
 
 typedef struct {
-  int8_t parallelizable;
   char* qmsg;
   // followings are not applicable to encoder and decoder
   void* inputHandle;
@@ -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);

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);

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));
 

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));

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;
   }