From cea25e7f9d9da27d55da219b4b2cf06a94a14e4b Mon Sep 17 00:00:00 2001
From: dapan1121 <wpan@taosdata.com>
Date: Mon, 28 Oct 2024 18:44:41 +0800
Subject: [PATCH] fix: scheduler ut issues

---
 source/libs/executor/test/queryPlanTests.cpp  |  2 ++
 source/libs/scheduler/src/schJob.c            |  4 ++--
 source/libs/scheduler/test/schedulerTests.cpp | 19 ++++++++++++++++++-
 3 files changed, 22 insertions(+), 3 deletions(-)

diff --git a/source/libs/executor/test/queryPlanTests.cpp b/source/libs/executor/test/queryPlanTests.cpp
index 1e0129ce5b..2b0c0e1118 100755
--- a/source/libs/executor/test/queryPlanTests.cpp
+++ b/source/libs/executor/test/queryPlanTests.cpp
@@ -371,6 +371,8 @@ int32_t qptGetColumnRandLen(int32_t colType) {
       assert(0);
       break;
   }
+
+  return 0;
 }
 
 
diff --git a/source/libs/scheduler/src/schJob.c b/source/libs/scheduler/src/schJob.c
index c52f779b75..ab9bc6a98f 100644
--- a/source/libs/scheduler/src/schJob.c
+++ b/source/libs/scheduler/src/schJob.c
@@ -334,7 +334,7 @@ int32_t schValidateAndBuildJob(SQueryPlan *pDag, SSchJob *pJob) {
       HASH_NO_LOCK);
   if (NULL == planToTask) {
     SCH_JOB_ELOG("taosHashInit %d failed", SCHEDULE_DEFAULT_MAX_TASK_NUM);
-    SCH_ERR_RET(terrno);
+    SCH_ERR_JRET(terrno);
   }
 
   pJob->levels = taosArrayInit(levelNum, sizeof(SSchLevel));
@@ -363,7 +363,7 @@ int32_t schValidateAndBuildJob(SQueryPlan *pDag, SSchJob *pJob) {
     pLevel = taosArrayGet(pJob->levels, i);
     if (NULL == pLevel) {
       SCH_JOB_ELOG("fail to get the %dth level, levelNum: %d", i, levelNum);
-      SCH_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
+      SCH_ERR_JRET(TSDB_CODE_QRY_INVALID_INPUT);
     }
 
     pLevel->level = i;
diff --git a/source/libs/scheduler/test/schedulerTests.cpp b/source/libs/scheduler/test/schedulerTests.cpp
index 44d32b9480..6e13e37e88 100644
--- a/source/libs/scheduler/test/schedulerTests.cpp
+++ b/source/libs/scheduler/test/schedulerTests.cpp
@@ -203,6 +203,10 @@ void schtBuildQueryDag(SQueryPlan *dag) {
     return;
   }
   scanPlan->msgType = TDMT_SCH_QUERY;
+  code = nodesMakeNode(QUERY_NODE_PHYSICAL_PLAN_DISPATCH, (SNode**)&scanPlan->pDataSink);
+  if (NULL == scanPlan->pDataSink) {
+    return;
+  }
 
   mergePlan->id.queryId = qId;
   mergePlan->id.groupId = schtMergeTemplateId;
@@ -223,6 +227,10 @@ void schtBuildQueryDag(SQueryPlan *dag) {
     return;
   }
   mergePlan->msgType = TDMT_SCH_QUERY;
+  code = nodesMakeNode(QUERY_NODE_PHYSICAL_PLAN_DISPATCH, (SNode**)&mergePlan->pDataSink);
+  if (NULL == mergePlan->pDataSink) {
+    return;
+  }
 
   merge->pNodeList = NULL;
   code = nodesMakeList(&merge->pNodeList);
@@ -235,6 +243,7 @@ void schtBuildQueryDag(SQueryPlan *dag) {
     return;
   }
 
+
   (void)nodesListAppend(merge->pNodeList, (SNode *)mergePlan);
   (void)nodesListAppend(scan->pNodeList, (SNode *)scanPlan);
 
@@ -250,7 +259,7 @@ void schtBuildQueryFlowCtrlDag(SQueryPlan *dag) {
   int32_t  scanPlanNum = 20;
 
   dag->queryId = qId;
-  dag->numOfSubplans = 2;
+  dag->numOfSubplans = scanPlanNum + 1;
   dag->pSubplans = NULL;
   int32_t code = nodesMakeList(&dag->pSubplans);
   if (NULL == dag->pSubplans) {
@@ -289,6 +298,10 @@ void schtBuildQueryFlowCtrlDag(SQueryPlan *dag) {
   if (NULL == mergePlan->pChildren) {
     return;
   }
+  code = nodesMakeNode(QUERY_NODE_PHYSICAL_PLAN_DISPATCH, (SNode**)&mergePlan->pDataSink);
+  if (NULL == mergePlan->pDataSink) {
+    return;
+  }
 
   for (int32_t i = 0; i < scanPlanNum; ++i) {
     SSubplan *scanPlan = NULL;
@@ -322,6 +335,10 @@ void schtBuildQueryFlowCtrlDag(SQueryPlan *dag) {
       return;
     }
     scanPlan->msgType = TDMT_SCH_QUERY;
+    code = nodesMakeNode(QUERY_NODE_PHYSICAL_PLAN_DISPATCH, (SNode**)&scanPlan->pDataSink);
+    if (NULL == scanPlan->pDataSink) {
+      return;
+    }
 
     (void)nodesListAppend(scanPlan->pParents, (SNode *)mergePlan);
     (void)nodesListAppend(mergePlan->pChildren, (SNode *)scanPlan);