Merge branch 'feature/queryredirect' of https://github.com/taosdata/TDengine into redir

include/common/tmsgdef.h

@@ -208,6 +208,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_SCH_CANCEL_TASK, "cancel-task", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_SCH_DROP_TASK, "drop-task", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_SCH_EXPLAIN, "explain", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_SCH_LINK_BROKEN, "link-broken", NULL, NULL)
 
   TD_NEW_MSG_SEG(TDMT_STREAM_MSG)
   TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_DEPLOY, "stream-task-deploy", SStreamTaskDeployReq, SStreamTaskDeployRsp)
@@ -217,9 +218,6 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_RECOVER, "stream-task-recover", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_STREAM_RETRIEVE, "stream-retrieve", NULL, NULL)
 
-  TD_NEW_MSG_SEG(TDMT_SCH_MSG)
-  TD_DEF_MSG_TYPE(TDMT_SCH_LINK_BROKEN, "link-broken", NULL, NULL)
-
   TD_NEW_MSG_SEG(TDMT_MON_MSG)
   TD_DEF_MSG_TYPE(TDMT_MON_MM_INFO, "monitor-minfo", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MON_VM_INFO, "monitor-vinfo", NULL, NULL)

include/libs/planner/planner.h

@@ -48,7 +48,7 @@ int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNo
 // @pSource one execution location of this group of datasource subplans
 int32_t qSetSubplanExecutionNode(SSubplan* pSubplan, int32_t groupId, SDownstreamSourceNode* pSource);
 
-int32_t qClearSubplanExecutionNode(SSubplan* pSubplan, int32_t groupId);
+int32_t qClearSubplanExecutionNode(SSubplan* pSubplan);
 
 // Convert to subplan to string for the scheduler to send to the executor
 int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen);

source/libs/planner/src/planner.c

@@ -85,11 +85,24 @@ int32_t qSetSubplanExecutionNode(SSubplan* subplan, int32_t groupId, SDownstream
   return setSubplanExecutionNode(subplan->pNode, groupId, pSource);
 }
 
-int32_t qClearSubplanExecutionNode(SSubplan* pSubplan, int32_t groupId) {
-  // todo
-  return TSDB_CODE_FAILED;
+static void clearSubplanExecutionNode(SPhysiNode* pNode) {
+  if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == nodeType(pNode)) {
+    SExchangePhysiNode* pExchange = (SExchangePhysiNode*)pNode;
+    NODES_DESTORY_LIST(pExchange->pSrcEndPoints);
+  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE == nodeType(pNode)) {
+    SMergePhysiNode* pMerge = (SMergePhysiNode*)pNode;
+    pMerge->numOfChannels = LIST_LENGTH(pMerge->node.pChildren);
+    SNode* pChild = NULL;
+    FOREACH(pChild, pMerge->node.pChildren) { NODES_DESTORY_LIST(((SExchangePhysiNode*)pChild)->pSrcEndPoints); }
+  }
+
+  SNode* pChild = NULL;
+  FOREACH(pChild, pNode->pChildren) { clearSubplanExecutionNode((SPhysiNode*)pChild); }
 }
 
+void qClearSubplanExecutionNode(SSubplan* pSubplan) { clearSubplanExecutionNode(pSubplan->pNode); }
+
+
 int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen) {
   if (SUBPLAN_TYPE_MODIFY == pSubplan->subplanType && NULL == pSubplan->pNode) {
     SDataInserterNode* insert = (SDataInserterNode*)pSubplan->pDataSink;

source/libs/transport/src/transSvr.c

@@ -1042,7 +1042,7 @@ void transReleaseSrvHandle(void* handle) {
   m->type = Release;
 
   tTrace("%s conn %p start to release", transLabel(pThrd->pTransInst), exh->handle);
-  transSendAsync(pThrd->asyncPool, &m->q);
+  transAsyncSend(pThrd->asyncPool, &m->q);
   transReleaseExHandle(refId);
   return;
 _return1:
@@ -1071,7 +1071,7 @@ void transSendResponse(const STransMsg* msg) {
 
   STraceId* trace = (STraceId*)&msg->info.traceId;
   tGTrace("conn %p start to send resp (1/2)", exh->handle);
-  transSendAsync(pThrd->asyncPool, &m->q);
+  transAsyncSend(pThrd->asyncPool, &m->q);
   transReleaseExHandle(refId);
   return;
 _return1:
@@ -1100,7 +1100,7 @@ void transRegisterMsg(const STransMsg* msg) {
   m->type = Register;
 
   tTrace("%s conn %p start to register brokenlink callback", transLabel(pThrd->pTransInst), exh->handle);
-  transSendAsync(pThrd->asyncPool, &m->q);
+  transAsyncSend(pThrd->asyncPool, &m->q);
   transReleaseExHandle(refId);
   return;