#5011

Committed 03 Apr 2026 03:59PM UTC coverage: 72.3% (+0.008%) from 72.292%

Build # #5011

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

merge: from main to 3.0 branch #35067

Coverage Stats

4053 of 5985 new or added lines in 68 files covered. (67.72%)

732 existing lines in 143 files now uncovered.

257430 of 356056 relevant lines covered (72.3%)

131834103.52 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.97

/source/libs/planner/src/planner.c

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "planner.h"
#include "planInt.h"
#include "tutil.h"

static int32_t debugPrintNode(SNode* pNode) {
  char*   pStr = NULL;
  int32_t code = nodesNodeToString(pNode, false, &pStr, NULL);
  if (TSDB_CODE_SUCCESS == code) {
    (void)printf("%s\n", pStr);
    taosMemoryFree(pStr);
  }
  return code;
}

static int32_t dumpQueryPlan(SQueryPlan* pPlan) {
  int32_t code = 0;
  if (!(qDebugFlag & DEBUG_DEBUG)) {
    return code;
  }

  char* pStr = NULL;
  code = nodesNodeToString((SNode*)pPlan, false, &pStr, NULL);
  if (TSDB_CODE_SUCCESS == code) {
    planDebugL("QID:0x%" PRIx64 ", Query Plan, JsonPlan: %s", pPlan->queryId, pStr);
    taosMemoryFree(pStr);
  }
  return code;
}

static void printPlanNode(SLogicNode *pNode, int32_t level) {
  // print pnode and it's child for each level
  const char *nodename = nodesNodeName(nodeType((pNode)));
  for (int32_t i = 0; i < level; i++) {
    printf("    ");
  }
  printf("%s\n", nodename);
  SNode *tmp = NULL;
  FOREACH(tmp, pNode->pChildren) {
    printPlanNode((SLogicNode *)tmp, level + 1);
  }
  return;
}

static void dumpLogicPlan(SLogicSubplan* pLogicSubplan, int32_t level) {
  for (int32_t i = 0; i < level; i++) {
    printf("    ");
  }
  printf("Sub Plan:\n");
  if (pLogicSubplan->pNode) {
     printPlanNode(pLogicSubplan->pNode, level);
  }
  SNode *pNode = NULL;
  FOREACH(pNode, pLogicSubplan->pChildren) {
    dumpLogicPlan((SLogicSubplan *)pNode, level + 2);
  }
  return;
}

static void initSubQueryPlanContext(SPlanContext* pDst, SPlanContext* pSrc, SNode* pRoot) {
  memcpy(pDst, pSrc, sizeof(*pSrc));

  pDst->groupId++;
  pDst->streamCxt.hasExtWindow = false;
  pDst->hasScan = false;
  
  pDst->pAstRoot = pRoot;
}

static bool isPartitionedExternalWindowSubquery(SNode* pRoot, int32_t subQIdx) {
  if (pRoot == NULL || subQIdx < 0) {
    return false;
  }

  if (QUERY_NODE_SELECT_STMT != nodeType(pRoot)) {
    return false;
  }

  SSelectStmt* pSelect = (SSelectStmt*)pRoot;
  if (pSelect->pWindow == NULL || QUERY_NODE_EXTERNAL_WINDOW != nodeType(pSelect->pWindow)) {
    return false;
  }

  SExternalWindowNode* pExternal = (SExternalWindowNode*)pSelect->pWindow;
  if (pSelect->pPartitionByList == NULL || pExternal->pSubquery == NULL ||
      QUERY_NODE_REMOTE_TABLE != nodeType(pExternal->pSubquery)) {
    return false;
  }

  return ((SRemoteTableNode*)pExternal->pSubquery)->subQIdx == subQIdx;
}

static int32_t createSubQueryPlans(SPlanContext* pSrc, SQueryPlan* pParent, SArray* pExecNodeList) {
  int32_t code = TSDB_CODE_SUCCESS, lino = 0;
  SPlanContext ctx;
  SNodeList* pSubQueries = NULL;
  SNode* pNode = NULL;
  SQueryPlan* pPlan = NULL;
  SNode* pRoot = NULL;

  switch (nodeType(pSrc->pAstRoot)) {
    case QUERY_NODE_EXPLAIN_STMT:
      pRoot = ((SExplainStmt*)pSrc->pAstRoot)->pQuery;
      break;
    case QUERY_NODE_INSERT_STMT:
      pRoot = ((SInsertStmt*)pSrc->pAstRoot)->pQuery;
      break;
    default:
      pRoot = pSrc->pAstRoot;
      break;
  }
  
  switch (nodeType(pRoot)) {
    case QUERY_NODE_SELECT_STMT: {
      SSelectStmt* pSelect = (SSelectStmt*)pRoot;
      pSubQueries = pSelect->pSubQueries;
      break;
    }
    case QUERY_NODE_SET_OPERATOR: {
      SSetOperator* pSet = (SSetOperator*)pRoot;
      pSubQueries = pSet->pSubQueries;
      break;
    }
    default:
      return code;
  }

  int32_t subQIdx = 0;
  FOREACH(pNode, pSubQueries) {
    planDebug("QID:0x%" PRIx64 ", createSubQueryPlans subQIdx:%d, nodeType:%d", pSrc->queryId, subQIdx, nodeType(pNode));
    initSubQueryPlanContext(&ctx, pSrc, pNode);
    ctx.forceNoMergeDataBlock = ctx.forceNoMergeDataBlock || isPartitionedExternalWindowSubquery(pRoot, subQIdx);
    TAOS_CHECK_EXIT(qCreateQueryPlan(&ctx, &pPlan, pExecNodeList));
    TAOS_CHECK_EXIT(nodesListMakeStrictAppend(&pParent->pChildren, (SNode*)pPlan));
    pParent->numOfSubplans += pPlan->numOfSubplans;
    pPlan->subSql = nodesGetSubSql(pNode);
    nodesGetSubQType(pNode, (int32_t*)&pPlan->subQType);
    pSrc->groupId = ++ctx.groupId;
    ++subQIdx;
  }

_exit:

  return code;
}

int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNodeList) {
  SLogicSubplan*   pLogicSubplan = NULL;
  SQueryLogicPlan* pLogicPlan = NULL;
  int32_t          code = TSDB_CODE_SUCCESS;
  int32_t          lino = 0;
  bool             allocatorReleased = false;

  TAOS_CHECK_EXIT(nodesAcquireAllocator(pCxt->allocatorId));
  TAOS_CHECK_EXIT(createLogicPlan(pCxt, &pLogicSubplan));
  TAOS_CHECK_EXIT(optimizeLogicPlan(pCxt, pLogicSubplan));
  TAOS_CHECK_EXIT(splitLogicPlan(pCxt, pLogicSubplan));
  TAOS_CHECK_EXIT(scaleOutLogicPlan(pCxt, pLogicSubplan, &pLogicPlan));
  TAOS_CHECK_EXIT(createPhysiPlan(pCxt, pLogicPlan, pPlan, pExecNodeList));
  TAOS_CHECK_EXIT(validateQueryPlan(pCxt, *pPlan));
  (void)nodesReleaseAllocator(pCxt->allocatorId);
  allocatorReleased = true;
  TAOS_CHECK_EXIT(createSubQueryPlans(pCxt, *pPlan, pExecNodeList));
  TAOS_CHECK_EXIT(dumpQueryPlan(*pPlan));

_exit:
  if (TSDB_CODE_SUCCESS != code) {
    if (!allocatorReleased) {
      (void)nodesReleaseAllocator(pCxt->allocatorId);
    }
    planError("QID:0x%" PRIx64 ", qCreateQueryPlan failed at line:%d, code:%s", pCxt->queryId, lino, tstrerror(code));
  }
  nodesDestroyNode((SNode*)pLogicSubplan);
  nodesDestroyNode((SNode*)pLogicPlan);
  
  terrno = code;
  return code;
}

/** Add vgId to exchange's childrenVgIds if not present. */
static int32_t addVgIdToExchange(SExchangePhysiNode* pExchange, int32_t vgId) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;

  if (NULL == pExchange->childrenVgIds) {
    pExchange->childrenVgIds = taosArrayInit(4, sizeof(int32_t));
    QUERY_CHECK_NULL(pExchange->childrenVgIds, code, lino, _end, terrno);
  }

  int32_t vgCnt = (int32_t)taosArrayGetSize(pExchange->childrenVgIds);
  for (int32_t i = 0; i < vgCnt; ++i) {
    const int32_t* pCurr = taosArrayGet(pExchange->childrenVgIds, i);
    if (pCurr && *pCurr == vgId) {
      return TSDB_CODE_SUCCESS;
    }
  }
  QUERY_CHECK_NULL(taosArrayPush(pExchange->childrenVgIds, &vgId), code, lino,
                   _end, terrno);

_end:
  if (TSDB_CODE_SUCCESS != code) {
    planError("%s, failed to add vgId to exchange at line %d, code:%d",
              __func__, lino, code);
  }
  return code;
}

static int32_t setSubplanExecutionNode(SPhysiNode* pNode, int32_t groupId,
                                       SDownstreamSourceNode* pSource) {
  int32_t code = TSDB_CODE_SUCCESS;

  if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == nodeType(pNode)) {
    SExchangePhysiNode* pExchange = (SExchangePhysiNode*)pNode;
    if (groupId >= pExchange->srcStartGroupId && groupId <= pExchange->srcEndGroupId) {
      code = addVgIdToExchange(pExchange, pSource->addr.nodeId);
      if (TSDB_CODE_SUCCESS != code) {
        planError("%s, failed to add vgId to exchange at line %d, code:%d",
                  __func__, __LINE__, code);
        return code;
      }

      SNode* pNew = NULL;
      code = nodesCloneNode((SNode*)pSource, &pNew);
      if (TSDB_CODE_SUCCESS == code) {
        return nodesListMakeStrictAppend(&pExchange->pSrcEndPoints, pNew);
      }
    }
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE == nodeType(pNode)) {
    SMergePhysiNode* pMerge = (SMergePhysiNode*)pNode;
    if (pMerge->srcGroupId <= groupId && pMerge->srcEndGroupId >= groupId) {
      SExchangePhysiNode* pMergeExchange =
          (SExchangePhysiNode*)nodesListGetNode(pMerge->node.pChildren, pMerge->numOfChannels - 1);
      if (1 == pMerge->numOfChannels) {
        pMerge->numOfChannels = LIST_LENGTH(pMerge->node.pChildren);
      } else {
        --(pMerge->numOfChannels);
      }
      code = addVgIdToExchange(pMergeExchange, pSource->addr.nodeId);
      if (TSDB_CODE_SUCCESS != code) {
        planError("%s, failed to add vgId to merge exchange at line %d, code:%d",
                  __func__, __LINE__, code);
        return code;
      }

      SNode* pNew = NULL;
      code = nodesCloneNode((SNode*)pSource, &pNew);
      if (TSDB_CODE_SUCCESS == code) {
        return nodesListMakeStrictAppend(&pMergeExchange->pSrcEndPoints, pNew);
      }
    }
  }

  SNode* pChild = NULL;
  if (TSDB_CODE_SUCCESS == code) {
  FOREACH(pChild, pNode->pChildren) {
      if (TSDB_CODE_SUCCESS != (code = setSubplanExecutionNode((SPhysiNode*)pChild, groupId, pSource))) {
        return code;
      }
    }
  }
  return code;
}

int32_t qContinuePlanPostQuery(void* pPostPlan) {
  // TODO
  return TSDB_CODE_SUCCESS;
}

int32_t qSetSubplanExecutionNode(SSubplan* subplan, int32_t groupId, SDownstreamSourceNode* pSource) {
  planDebug("QID:0x%" PRIx64 ", set subplan execution node, groupId:%d", subplan->id.queryId, groupId);
  return setSubplanExecutionNode(subplan->pNode, groupId, pSource);
}

static void clearSubplanExecutionNode(SPhysiNode* pNode) {
  if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == nodeType(pNode)) {
    SExchangePhysiNode* pExchange = (SExchangePhysiNode*)pNode;
    NODES_DESTORY_LIST(pExchange->pSrcEndPoints);
    taosArrayDestroy(pExchange->childrenVgIds);
    pExchange->childrenVgIds = NULL;
  } 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);
      taosArrayDestroy(((SExchangePhysiNode*)pChild)->childrenVgIds);
      ((SExchangePhysiNode*)pChild)->childrenVgIds = NULL;
    }
  }

  SNode* pChild = NULL;
  FOREACH(pChild, pNode->pChildren) { clearSubplanExecutionNode((SPhysiNode*)pChild); }
}

void qClearSubplanExecutionNode(SSubplan* pSubplan) {
  planDebug("QID:0x%" PRIx64 ", clear subplan execution node, groupId:%d", pSubplan->id.queryId, pSubplan->id.groupId);
  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;
    *pLen = insert->size;
    *pStr = insert->pData;
    insert->pData = NULL;
    return TSDB_CODE_SUCCESS;
  }
  return nodesNodeToString((const SNode*)pSubplan, false, pStr, pLen);
}

int32_t qStringToSubplan(const char* pStr, SSubplan** pSubplan) { return nodesStringToNode(pStr, (SNode**)pSubplan); }

int32_t qSubPlanToMsg(const SSubplan* pSubplan, char** pStr, int32_t* pLen) {
  if (NULL == pSubplan) {
    return terrno = TSDB_CODE_INVALID_PARA;
  }
  
  if (SUBPLAN_TYPE_MODIFY == pSubplan->subplanType && NULL == pSubplan->pNode) {
    SDataInserterNode* insert = (SDataInserterNode*)pSubplan->pDataSink;
    *pLen = insert->size;
    *pStr = insert->pData;
    insert->pData = NULL;
    return TSDB_CODE_SUCCESS;
  }
  return nodesNodeToMsg((const SNode*)pSubplan, pStr, pLen);
}

int32_t qMsgToSubplan(const char* pStr, int32_t len, SSubplan** pSubplan) {
  return nodesMsgToNode(pStr, len, (SNode**)pSubplan);
}

SQueryPlan* qStringToQueryPlan(const char* pStr) {
  SQueryPlan* pPlan = NULL;
  if (TSDB_CODE_SUCCESS != nodesStringToNode(pStr, (SNode**)&pPlan)) {
    return NULL;
  }
  return pPlan;
}

void qDestroyQueryPlan(SQueryPlan* pPlan) { nodesDestroyNode((SNode*)pPlan); }

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#include "planner.h"
17	#include "planInt.h"
18	#include "tutil.h"
19
20	static int32_t debugPrintNode(SNode* pNode) {	×
21	char* pStr = NULL;	×
22	int32_t code = nodesNodeToString(pNode, false, &pStr, NULL);	×
23	if (TSDB_CODE_SUCCESS == code) {	×
24	(void)printf("%s\n", pStr);	×
25	taosMemoryFree(pStr);	×
26	}
27	return code;	×
28	}
29
30	static int32_t dumpQueryPlan(SQueryPlan* pPlan) {	917,719,590✔
31	int32_t code = 0;	917,719,590✔
32	if (!(qDebugFlag & DEBUG_DEBUG)) {	917,719,590✔
33	return code;	289,094,701✔
34	}
35
36	char* pStr = NULL;	628,624,889✔
37	code = nodesNodeToString((SNode*)pPlan, false, &pStr, NULL);	628,650,234✔
38	if (TSDB_CODE_SUCCESS == code) {	628,677,966✔
39	planDebugL("QID:0x%" PRIx64 ", Query Plan, JsonPlan: %s", pPlan->queryId, pStr);	628,677,966✔
40	taosMemoryFree(pStr);	628,679,265✔
41	}
42	return code;	628,679,265✔
43	}
44
45	static void printPlanNode(SLogicNode *pNode, int32_t level) {	×
46	// print pnode and it's child for each level
47	const char *nodename = nodesNodeName(nodeType((pNode)));	×
48	for (int32_t i = 0; i < level; i++) {	×
49	printf(" ");	×
50	}
51	printf("%s\n", nodename);	×
52	SNode *tmp = NULL;	×
53	FOREACH(tmp, pNode->pChildren) {	×
54	printPlanNode((SLogicNode *)tmp, level + 1);	×
55	}
56	return;	×
57	}
58
59	static void dumpLogicPlan(SLogicSubplan* pLogicSubplan, int32_t level) {	×
60	for (int32_t i = 0; i < level; i++) {	×
61	printf(" ");	×
62	}
63	printf("Sub Plan:\n");	×
64	if (pLogicSubplan->pNode) {	×
65	printPlanNode(pLogicSubplan->pNode, level);	×
66	}
67	SNode *pNode = NULL;	×
68	FOREACH(pNode, pLogicSubplan->pChildren) {	×
69	dumpLogicPlan((SLogicSubplan *)pNode, level + 2);	×
70	}
71	return;	×
72	}
73
74	static void initSubQueryPlanContext(SPlanContext* pDst, SPlanContext* pSrc, SNode* pRoot) {	170,730,139✔
75	memcpy(pDst, pSrc, sizeof(*pSrc));	170,730,139✔
76
77	pDst->groupId++;	170,730,139✔
78	pDst->streamCxt.hasExtWindow = false;	170,730,139✔
79	pDst->hasScan = false;	170,730,139✔
80
81	pDst->pAstRoot = pRoot;	170,730,139✔
82	}	170,730,139✔
83
84	static bool isPartitionedExternalWindowSubquery(SNode* pRoot, int32_t subQIdx) {	170,699,539✔
85	if (pRoot == NULL \|\| subQIdx < 0) {	170,699,539✔
NEW 86	return false;	×
87	}
88
89	if (QUERY_NODE_SELECT_STMT != nodeType(pRoot)) {	170,725,171✔
90	return false;	6,035,370✔
91	}
92
93	SSelectStmt* pSelect = (SSelectStmt*)pRoot;	164,689,801✔
94	if (pSelect->pWindow == NULL \|\| QUERY_NODE_EXTERNAL_WINDOW != nodeType(pSelect->pWindow)) {	164,689,801✔
95	return false;	164,560,194✔
96	}
97
98	SExternalWindowNode* pExternal = (SExternalWindowNode*)pSelect->pWindow;	129,607✔
99	if (pSelect->pPartitionByList == NULL \|\| pExternal->pSubquery == NULL \|\|	129,607✔
100	QUERY_NODE_REMOTE_TABLE != nodeType(pExternal->pSubquery)) {	55,895✔
101	return false;	73,712✔
102	}
103
104	return ((SRemoteTableNode*)pExternal->pSubquery)->subQIdx == subQIdx;	55,895✔
105	}
106
107	static int32_t createSubQueryPlans(SPlanContext* pSrc, SQueryPlan* pParent, SArray* pExecNodeList) {	917,734,589✔
108	int32_t code = TSDB_CODE_SUCCESS, lino = 0;	917,734,589✔
109	SPlanContext ctx;	736,809,263✔
110	SNodeList* pSubQueries = NULL;	917,756,110✔
111	SNode* pNode = NULL;	917,756,110✔
112	SQueryPlan* pPlan = NULL;	917,756,110✔
113	SNode* pRoot = NULL;	917,770,826✔
114
115	switch (nodeType(pSrc->pAstRoot)) {	917,770,826✔
116	case QUERY_NODE_EXPLAIN_STMT:	49,067,647✔
117	pRoot = ((SExplainStmt*)pSrc->pAstRoot)->pQuery;	49,067,647✔
118	break;	49,067,647✔
119	case QUERY_NODE_INSERT_STMT:	526,042✔
120	pRoot = ((SInsertStmt*)pSrc->pAstRoot)->pQuery;	526,042✔
121	break;	526,042✔
122	default:	868,105,399✔
123	pRoot = pSrc->pAstRoot;	868,105,399✔
124	break;	868,169,281✔
125	}
126
127	switch (nodeType(pRoot)) {	917,762,970✔
128	case QUERY_NODE_SELECT_STMT: {	346,713,865✔
129	SSelectStmt* pSelect = (SSelectStmt*)pRoot;	346,713,865✔
130	pSubQueries = pSelect->pSubQueries;	346,713,865✔
131	break;	346,713,046✔
132	}
133	case QUERY_NODE_SET_OPERATOR: {	24,111,093✔
134	SSetOperator* pSet = (SSetOperator*)pRoot;	24,111,093✔
135	pSubQueries = pSet->pSubQueries;	24,111,093✔
136	break;	24,062,258✔
137	}
138	default:	546,924,650✔
139	return code;	546,924,650✔
140	}
141
142	int32_t subQIdx = 0;	370,775,304✔
143	FOREACH(pNode, pSubQueries) {	541,511,576✔
144	planDebug("QID:0x%" PRIx64 ", createSubQueryPlans subQIdx:%d, nodeType:%d", pSrc->queryId, subQIdx, nodeType(pNode));	170,764,329✔
145	initSubQueryPlanContext(&ctx, pSrc, pNode);	170,764,329✔
146	ctx.forceNoMergeDataBlock = ctx.forceNoMergeDataBlock \|\| isPartitionedExternalWindowSubquery(pRoot, subQIdx);	170,706,668✔
147	TAOS_CHECK_EXIT(qCreateQueryPlan(&ctx, &pPlan, pExecNodeList));	170,756,703✔
148	TAOS_CHECK_EXIT(nodesListMakeStrictAppend(&pParent->pChildren, (SNode*)pPlan));	170,747,776✔
149	pParent->numOfSubplans += pPlan->numOfSubplans;	170,756,173✔
150	pPlan->subSql = nodesGetSubSql(pNode);	170,756,173✔
151	nodesGetSubQType(pNode, (int32_t*)&pPlan->subQType);	170,732,992✔
152	pSrc->groupId = ++ctx.groupId;	170,737,844✔
153	++subQIdx;	170,736,272✔
154	}
155
156	_exit:	370,750,910✔
157
158	return code;	370,750,976✔
159	}
160
161	int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNodeList) {	918,021,319✔
162	SLogicSubplan* pLogicSubplan = NULL;	918,021,319✔
163	SQueryLogicPlan* pLogicPlan = NULL;	918,033,695✔
164	int32_t code = TSDB_CODE_SUCCESS;	918,038,471✔
165	int32_t lino = 0;	918,038,471✔
166	bool allocatorReleased = false;	918,038,471✔
167
168	TAOS_CHECK_EXIT(nodesAcquireAllocator(pCxt->allocatorId));	918,038,471✔
169	TAOS_CHECK_EXIT(createLogicPlan(pCxt, &pLogicSubplan));	918,081,999✔
170	TAOS_CHECK_EXIT(optimizeLogicPlan(pCxt, pLogicSubplan));	918,012,907✔
171	TAOS_CHECK_EXIT(splitLogicPlan(pCxt, pLogicSubplan));	917,718,842✔
172	TAOS_CHECK_EXIT(scaleOutLogicPlan(pCxt, pLogicSubplan, &pLogicPlan));	917,751,302✔
173	TAOS_CHECK_EXIT(createPhysiPlan(pCxt, pLogicPlan, pPlan, pExecNodeList));	917,767,323✔
174	TAOS_CHECK_EXIT(validateQueryPlan(pCxt, *pPlan));	917,773,132✔
175	(void)nodesReleaseAllocator(pCxt->allocatorId);	917,765,217✔
176	allocatorReleased = true;	917,813,940✔
177	TAOS_CHECK_EXIT(createSubQueryPlans(pCxt, *pPlan, pExecNodeList));	917,813,940✔
178	TAOS_CHECK_EXIT(dumpQueryPlan(*pPlan));	917,745,086✔
179
180	_exit:	918,022,657✔
181	if (TSDB_CODE_SUCCESS != code) {	918,027,880✔
182	if (!allocatorReleased) {	260,889✔
183	(void)nodesReleaseAllocator(pCxt->allocatorId);	258,834✔
184	}
185	planError("QID:0x%" PRIx64 ", qCreateQueryPlan failed at line:%d, code:%s", pCxt->queryId, lino, tstrerror(code));	260,889✔
186	}
187	nodesDestroyNode((SNode*)pLogicSubplan);	918,027,880✔
188	nodesDestroyNode((SNode*)pLogicPlan);	918,020,529✔
189
190	terrno = code;	918,077,581✔
191	return code;	918,080,910✔
192	}
193
194	/** Add vgId to exchange's childrenVgIds if not present. */
195	static int32_t addVgIdToExchange(SExchangePhysiNode* pExchange, int32_t vgId) {	260,944,607✔
196	int32_t code = TSDB_CODE_SUCCESS;	260,944,607✔
197	int32_t lino = 0;	260,944,607✔
198
199	if (NULL == pExchange->childrenVgIds) {	260,944,607✔
200	pExchange->childrenVgIds = taosArrayInit(4, sizeof(int32_t));	147,344,221✔
201	QUERY_CHECK_NULL(pExchange->childrenVgIds, code, lino, _end, terrno);	147,346,417✔
202	}
203
204	int32_t vgCnt = (int32_t)taosArrayGetSize(pExchange->childrenVgIds);	260,946,803✔
205	for (int32_t i = 0; i < vgCnt; ++i) {	505,995,303✔
206	const int32_t* pCurr = taosArrayGet(pExchange->childrenVgIds, i);	257,411,240✔
207	if (pCurr && *pCurr == vgId) {	257,410,079✔
208	return TSDB_CODE_SUCCESS;	12,362,433✔
209	}
210	}
211	QUERY_CHECK_NULL(taosArrayPush(pExchange->childrenVgIds, &vgId), code, lino,	497,166,342✔
212	_end, terrno);
213
214	_end:	248,582,279✔
215	if (TSDB_CODE_SUCCESS != code) {	248,582,279✔
216	planError("%s, failed to add vgId to exchange at line %d, code:%d",	×
217	__func__, lino, code);
218	}
219	return code;	248,581,119✔
220	}
221
222	static int32_t setSubplanExecutionNode(SPhysiNode* pNode, int32_t groupId,	2,147,483,647✔
223	SDownstreamSourceNode* pSource) {
224	int32_t code = TSDB_CODE_SUCCESS;	2,147,483,647✔
225
226	if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == nodeType(pNode)) {	2,147,483,647✔
227	SExchangePhysiNode* pExchange = (SExchangePhysiNode*)pNode;	2,147,483,647✔
228	if (groupId >= pExchange->srcStartGroupId && groupId <= pExchange->srcEndGroupId) {	2,147,483,647✔
229	code = addVgIdToExchange(pExchange, pSource->addr.nodeId);	215,628,115✔
230	if (TSDB_CODE_SUCCESS != code) {	215,624,700✔
231	planError("%s, failed to add vgId to exchange at line %d, code:%d",	×
232	__func__, __LINE__, code);
233	return code;	34,406,470✔
234	}
235
236	SNode* pNew = NULL;	215,624,700✔
237	code = nodesCloneNode((SNode*)pSource, &pNew);	215,626,383✔
238	if (TSDB_CODE_SUCCESS == code) {	215,626,410✔
239	return nodesListMakeStrictAppend(&pExchange->pSrcEndPoints, pNew);	215,626,410✔
240	}
241	}
242	} else if (QUERY_NODE_PHYSICAL_PLAN_MERGE == nodeType(pNode)) {	657,096,116✔
243	SMergePhysiNode* pMerge = (SMergePhysiNode*)pNode;	65,534,699✔
244	if (pMerge->srcGroupId <= groupId && pMerge->srcEndGroupId >= groupId) {	65,534,699✔
245	SExchangePhysiNode* pMergeExchange =
246	(SExchangePhysiNode*)nodesListGetNode(pMerge->node.pChildren, pMerge->numOfChannels - 1);	45,319,035✔
247	if (1 == pMerge->numOfChannels) {	45,318,612✔
248	pMerge->numOfChannels = LIST_LENGTH(pMerge->node.pChildren);	18,514,493✔
249	} else {
250	--(pMerge->numOfChannels);	26,804,523✔
251	}
252	code = addVgIdToExchange(pMergeExchange, pSource->addr.nodeId);	45,319,016✔
253	if (TSDB_CODE_SUCCESS != code) {	45,318,174✔
254	planError("%s, failed to add vgId to merge exchange at line %d, code:%d",	×
255	__func__, __LINE__, code);
256	return code;	9,357,006✔
257	}
258
259	SNode* pNew = NULL;	45,318,174✔
260	code = nodesCloneNode((SNode*)pSource, &pNew);	45,318,174✔
261	if (TSDB_CODE_SUCCESS == code) {	45,318,690✔
262	return nodesListMakeStrictAppend(&pMergeExchange->pSrcEndPoints, pNew);	45,318,690✔
263	}
264	}
265	}
266
267	SNode* pChild = NULL;	2,147,483,647✔
268	if (TSDB_CODE_SUCCESS == code) {	2,147,483,647✔
269	FOREACH(pChild, pNode->pChildren) {	2,147,483,647✔
270	if (TSDB_CODE_SUCCESS != (code = setSubplanExecutionNode((SPhysiNode*)pChild, groupId, pSource))) {	2,147,483,647✔
271	return code;	×
272	}
273	}
274	}
275	return code;	2,147,483,647✔
276	}
277
278	int32_t qContinuePlanPostQuery(void* pPostPlan) {	×
279	// TODO
280	return TSDB_CODE_SUCCESS;	×
281	}
282
283	int32_t qSetSubplanExecutionNode(SSubplan* subplan, int32_t groupId, SDownstreamSourceNode* pSource) {	261,064,239✔
284	planDebug("QID:0x%" PRIx64 ", set subplan execution node, groupId:%d", subplan->id.queryId, groupId);	261,064,239✔
285	return setSubplanExecutionNode(subplan->pNode, groupId, pSource);	261,064,239✔
286	}
287
288	static void clearSubplanExecutionNode(SPhysiNode* pNode) {	6,249,078✔
289	if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == nodeType(pNode)) {	6,249,078✔
290	SExchangePhysiNode* pExchange = (SExchangePhysiNode*)pNode;	1,478✔
291	NODES_DESTORY_LIST(pExchange->pSrcEndPoints);	1,478✔
292	taosArrayDestroy(pExchange->childrenVgIds);	1,478✔
293	pExchange->childrenVgIds = NULL;	1,478✔
294	} else if (QUERY_NODE_PHYSICAL_PLAN_MERGE == nodeType(pNode)) {	6,247,600✔
295	SMergePhysiNode* pMerge = (SMergePhysiNode*)pNode;	×
296	pMerge->numOfChannels = LIST_LENGTH(pMerge->node.pChildren);	×
297	SNode* pChild = NULL;	×
298	FOREACH(pChild, pMerge->node.pChildren) {	×
299	NODES_DESTORY_LIST(((SExchangePhysiNode*)pChild)->pSrcEndPoints);	×
300	taosArrayDestroy(((SExchangePhysiNode*)pChild)->childrenVgIds);	×
301	((SExchangePhysiNode*)pChild)->childrenVgIds = NULL;	×
302	}
303	}
304
305	SNode* pChild = NULL;	6,249,078✔
306	FOREACH(pChild, pNode->pChildren) { clearSubplanExecutionNode((SPhysiNode*)pChild); }	8,260,893✔
307	}	6,249,078✔
308
309	void qClearSubplanExecutionNode(SSubplan* pSubplan) {	4,237,263✔
310	planDebug("QID:0x%" PRIx64 ", clear subplan execution node, groupId:%d", pSubplan->id.queryId, pSubplan->id.groupId);	4,237,263✔
311	clearSubplanExecutionNode(pSubplan->pNode);	4,237,263✔
312	}	4,237,263✔
313
314	int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen) {	23,828✔
315	if (SUBPLAN_TYPE_MODIFY == pSubplan->subplanType && NULL == pSubplan->pNode) {	23,828✔
316	SDataInserterNode* insert = (SDataInserterNode*)pSubplan->pDataSink;	×
317	*pLen = insert->size;	×
318	*pStr = insert->pData;	×
319	insert->pData = NULL;	×
320	return TSDB_CODE_SUCCESS;	×
321	}
322	return nodesNodeToString((const SNode*)pSubplan, false, pStr, pLen);	23,828✔
323	}
324
325	int32_t qStringToSubplan(const char* pStr, SSubplan pSubplan) { return nodesStringToNode(pStr, (SNode)pSubplan); }	1,500,206✔
326
327	int32_t qSubPlanToMsg(const SSubplan* pSubplan, char** pStr, int32_t* pLen) {	1,095,822,625✔
328	if (NULL == pSubplan) {	1,095,822,625✔
329	return terrno = TSDB_CODE_INVALID_PARA;	1,960✔
330	}
331
332	if (SUBPLAN_TYPE_MODIFY == pSubplan->subplanType && NULL == pSubplan->pNode) {	1,095,820,665✔
333	SDataInserterNode* insert = (SDataInserterNode*)pSubplan->pDataSink;	564,489,004✔
334	*pLen = insert->size;	564,485,515✔
335	*pStr = insert->pData;	564,482,428✔
336	insert->pData = NULL;	564,496,557✔
337	return TSDB_CODE_SUCCESS;	564,476,127✔
338	}
339	return nodesNodeToMsg((const SNode*)pSubplan, pStr, pLen);	531,265,005✔
340	}
341
342	int32_t qMsgToSubplan(const char* pStr, int32_t len, SSubplan** pSubplan) {	375,682,079✔
343	return nodesMsgToNode(pStr, len, (SNode**)pSubplan);	375,682,079✔
344	}
345
346	SQueryPlan* qStringToQueryPlan(const char* pStr) {	×
347	SQueryPlan* pPlan = NULL;	×
348	if (TSDB_CODE_SUCCESS != nodesStringToNode(pStr, (SNode**)&pPlan)) {	×
349	return NULL;	×
350	}
351	return pPlan;	×
352	}
353
354	void qDestroyQueryPlan(SQueryPlan* pPlan) { nodesDestroyNode((SNode*)pPlan); }	746,536,564✔

taosdata / TDengine / #5011

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous