fix schedule

source/dnode/vnode/src/tq/tq.c

@@ -751,12 +751,13 @@ int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg) {
 }
 
 int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg, bool exec) {
+  ASSERT(0);
   char*              msgStr = pMsg->pCont;
   char*              msgBody = POINTER_SHIFT(msgStr, sizeof(SMsgHead));
   int32_t            msgLen = pMsg->contLen - sizeof(SMsgHead);
   SStreamDispatchReq req;
   SDecoder           decoder;
-  tDecoderInit(&decoder, msgBody, msgLen);
+  tDecoderInit(&decoder, (uint8_t*)msgBody, msgLen);
   tDecodeStreamDispatchReq(&decoder, &req);
   int32_t       taskId = req.taskId;
   SStreamTask** ppTask = (SStreamTask**)taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));

source/libs/stream/src/stream.c

@@ -295,10 +295,11 @@ int32_t streamProcessRetrieveReq(SStreamTask* pTask, SStreamRetrieveReq* pReq, S
   streamTaskEnqueueRetrieve(pTask, pReq, pRsp);
 
   ASSERT(pTask->execType != TASK_EXEC__NONE);
-  streamTryExec(pTask);
+  streamSchedExec(pTask);
+  /*streamTryExec(pTask);*/
 
-  ASSERT(pTask->dispatchType != TASK_DISPATCH__NONE);
-  streamDispatch(pTask);
+  /*ASSERT(pTask->dispatchType != TASK_DISPATCH__NONE);*/
+  /*streamDispatch(pTask);*/
 
   return 0;
 }

source/libs/stream/src/streamExec.c

@@ -162,10 +162,7 @@ int32_t streamExecForAll(SStreamTask* pTask) {
         data = qItem;
         streamQueueProcessSuccess(pTask->inputQueue);
         if (pTask->execType == TASK_EXEC__NONE) {
-          ASSERT(((SStreamQueueItem*)data)->type == STREAM_INPUT__DATA_BLOCK);
-          streamTaskOutput(pTask, data);
-          data = NULL;
-          continue;
+          break;
         }
       } else {
         void* newRet;
@@ -187,7 +184,13 @@ int32_t streamExecForAll(SStreamTask* pTask) {
     }
 
     if (data == NULL) {
-      return 0;
+      break;
+    }
+
+    if (pTask->execType == TASK_EXEC__NONE) {
+      ASSERT(((SStreamQueueItem*)data)->type == STREAM_INPUT__DATA_BLOCK);
+      streamTaskOutput(pTask, data);
+      continue;
     }
 
     SArray* pRes = taosArrayInit(0, sizeof(SSDataBlock));