Merge pull request #13718 from taosdata/feature/stream

fix(stream): fix shuffle vg id not initialized

source/dnode/mnode/impl/src/mndScheduler.c

@@ -151,20 +151,23 @@ int32_t mndAddDispatcherToInnerTask(SMnode* pMnode, STrans* pTrans, SStreamObj*
     ASSERT(pDb);
 
     if (mndExtractDbInfo(pMnode, pDb, &pTask->shuffleDispatcher.dbInfo, NULL) < 0) {
-      sdbRelease(pMnode->pSdb, pDb);
-
-      SArray* pVgs = pTask->shuffleDispatcher.dbInfo.pVgroupInfos;
-      int32_t sz = taosArrayGetSize(pVgs);
-      SArray* sinkLv = taosArrayGetP(pStream->tasks, 0);
-      int32_t sinkLvSize = taosArrayGetSize(sinkLv);
-      for (int32_t i = 0; i < sz; i++) {
-        SVgroupInfo* pVgInfo = taosArrayGet(pVgs, i);
-        for (int32_t j = 0; j < sinkLvSize; j++) {
-          SStreamTask* pLastLevelTask = taosArrayGetP(sinkLv, j);
-          if (pLastLevelTask->nodeId == pVgInfo->vgId) {
-            pVgInfo->taskId = pLastLevelTask->taskId;
-            break;
-          }
+      ASSERT(0);
+      return -1;
+    }
+    sdbRelease(pMnode->pSdb, pDb);
+
+    SArray* pVgs = pTask->shuffleDispatcher.dbInfo.pVgroupInfos;
+    int32_t sz = taosArrayGetSize(pVgs);
+    SArray* sinkLv = taosArrayGetP(pStream->tasks, 0);
+    int32_t sinkLvSize = taosArrayGetSize(sinkLv);
+    for (int32_t i = 0; i < sz; i++) {
+      SVgroupInfo* pVgInfo = taosArrayGet(pVgs, i);
+      for (int32_t j = 0; j < sinkLvSize; j++) {
+        SStreamTask* pLastLevelTask = taosArrayGetP(sinkLv, j);
+        if (pLastLevelTask->nodeId == pVgInfo->vgId) {
+          pVgInfo->taskId = pLastLevelTask->taskId;
+          ASSERT(pVgInfo->taskId != 0);
+          break;
         }
       }
     }
@@ -379,7 +382,10 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
       pFinalTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK;
 
       // dispatch
-      mndAddDispatcherToInnerTask(pMnode, pTrans, pStream, pFinalTask);
+      if (mndAddDispatcherToInnerTask(pMnode, pTrans, pStream, pFinalTask) < 0) {
+        qDestroyQueryPlan(pPlan);
+        return -1;
+      }
 
       // exec
       pFinalTask->execType = TASK_EXEC__PIPE;

source/libs/stream/src/streamDispatch.c

@@ -100,7 +100,6 @@ 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));
@@ -142,11 +141,14 @@ int32_t streamBuildDispatchMsg(SStreamTask* pTask, SStreamDataBlock* data, SRpcM
         break;
       }
     }
-    ASSERT(vgId != 0);
   }
 
+  ASSERT(vgId != 0);
   req.taskId = downstreamTaskId;
 
+  qInfo("dispatch from task %d (child id %d) to down stream task %d in vnode %d", pTask->taskId, pTask->childId,
+        downstreamTaskId, vgId);
+
   // serialize
   int32_t tlen;
   tEncodeSize(tEncodeStreamDispatchReq, &req, tlen, code);