From dbbff4ea2c043f3447b5937520d0a5388c0660d2 Mon Sep 17 00:00:00 2001 From: slzhou Date: Fri, 1 Jul 2022 20:00:01 +0800 Subject: [PATCH 1/8] feat: push cond to join child through generating plan --- include/libs/nodes/plannodes.h | 2 ++ source/libs/executor/src/joinoperator.c | 25 ++++++++++++++++++++----- 2 files changed, 22 insertions(+), 5 deletions(-) diff --git a/include/libs/nodes/plannodes.h b/include/libs/nodes/plannodes.h index 012ba114ce..05553694f5 100644 --- a/include/libs/nodes/plannodes.h +++ b/include/libs/nodes/plannodes.h @@ -82,6 +82,7 @@ typedef struct SScanLogicNode { typedef struct SJoinLogicNode { SLogicNode node; EJoinType joinType; + SNode* pMergeCondition; SNode* pOnConditions; bool isSingleTableJoin; } SJoinLogicNode; @@ -327,6 +328,7 @@ typedef struct SInterpFuncPhysiNode { typedef struct SJoinPhysiNode { SPhysiNode node; EJoinType joinType; + SNode* pMergeCondition; SNode* pOnConditions; SNodeList* pTargets; } SJoinPhysiNode; diff --git a/source/libs/executor/src/joinoperator.c b/source/libs/executor/src/joinoperator.c index 6fbda77808..33954d5517 100644 --- a/source/libs/executor/src/joinoperator.c +++ b/source/libs/executor/src/joinoperator.c @@ -90,13 +90,9 @@ void destroyMergeJoinOperator(void* param, int32_t numOfOutput) { SJoinOperatorInfo* pJoinOperator = (SJoinOperatorInfo*)param; } -SSDataBlock* doMergeJoin(struct SOperatorInfo* pOperator) { +static void doMergeJoinImpl(struct SOperatorInfo* pOperator, SSDataBlock* pRes) { SJoinOperatorInfo* pJoinInfo = pOperator->info; - SSDataBlock* pRes = pJoinInfo->pRes; - blockDataCleanup(pRes); - blockDataEnsureCapacity(pRes, 4096); - int32_t nrows = 0; while (1) { @@ -181,7 +177,26 @@ SSDataBlock* doMergeJoin(struct SOperatorInfo* pOperator) { break; } } +} +SSDataBlock* doMergeJoin(struct SOperatorInfo* pOperator) { + SJoinOperatorInfo* pJoinInfo = pOperator->info; + + SSDataBlock* pRes = pJoinInfo->pRes; + blockDataCleanup(pRes); + blockDataEnsureCapacity(pRes, 4096); + while (true) { + int32_t numOfRowsBefore = pRes->info.rows; + doMergeJoinImpl(pOperator, pRes); + int32_t numOfNewRows = pRes->info.rows - numOfRowsBefore; + if (numOfNewRows == 0) { + break; + } + doFilter(pJoinInfo->pOnCondition, pRes); + if (pRes->info.rows >= pOperator->resultInfo.threshold) { + break; + } + } return (pRes->info.rows > 0) ? pRes : NULL; } From a380e0cd7cc40ca3cd2ca060e1492e6f62bb5877 Mon Sep 17 00:00:00 2001 From: slzhou Date: Mon, 4 Jul 2022 19:49:16 +0800 Subject: [PATCH 2/8] feat: merge condition, on condition and other conditions --- include/libs/nodes/querynodes.h | 1 + source/libs/executor/inc/executorimpl.h | 3 +- source/libs/executor/src/joinoperator.c | 22 ++++++++++---- source/libs/nodes/src/nodesCloneFuncs.c | 1 + source/libs/nodes/src/nodesCodeFuncs.c | 11 +++++++ source/libs/nodes/src/nodesTraverseFuncs.c | 3 ++ source/libs/nodes/src/nodesUtilFuncs.c | 34 ++++++++++++++++++++++ source/libs/planner/src/planOptimizer.c | 12 +++++--- source/libs/planner/src/planPhysiCreater.c | 21 ++++++++++--- 9 files changed, 93 insertions(+), 15 deletions(-) diff --git a/include/libs/nodes/querynodes.h b/include/libs/nodes/querynodes.h index 9d84f2c32f..4a798c2dfa 100644 --- a/include/libs/nodes/querynodes.h +++ b/include/libs/nodes/querynodes.h @@ -375,6 +375,7 @@ void nodesRewriteSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeRewrit typedef enum ECollectColType { COLLECT_COL_TYPE_COL = 1, COLLECT_COL_TYPE_TAG, COLLECT_COL_TYPE_ALL } ECollectColType; int32_t nodesCollectColumns(SSelectStmt* pSelect, ESqlClause clause, const char* pTableAlias, ECollectColType type, SNodeList** pCols); +int32_t nodesCollectColumnsFromNode(SNode* node, const char* pTableAlias, ECollectColType type, SNodeList** pCols); typedef bool (*FFuncClassifier)(int32_t funcId); int32_t nodesCollectFuncs(SSelectStmt* pSelect, ESqlClause clause, FFuncClassifier classifier, SNodeList** pFuncs); diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 9872f26b03..1318b6e4c1 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -678,7 +678,8 @@ typedef struct SJoinOperatorInfo { SSDataBlock *pRight; int32_t rightPos; SColumnInfo rightCol; - SNode *pOnCondition; + SNode *pOnConditions; + SNode *pOtherConditions; } SJoinOperatorInfo; #define OPTR_IS_OPENED(_optr) (((_optr)->status & OP_OPENED) == OP_OPENED) diff --git a/source/libs/executor/src/joinoperator.c b/source/libs/executor/src/joinoperator.c index 33954d5517..71f04815dc 100644 --- a/source/libs/executor/src/joinoperator.c +++ b/source/libs/executor/src/joinoperator.c @@ -53,13 +53,20 @@ SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t pOperator->info = pInfo; pOperator->pTaskInfo = pTaskInfo; - SNode* pOnCondition = pJoinNode->pOnConditions; - if (nodeType(pOnCondition) == QUERY_NODE_OPERATOR) { - SOperatorNode* pNode = (SOperatorNode*)pOnCondition; + SNode* pMergeCondition = pJoinNode->pMergeCondition; + if (nodeType(pMergeCondition) == QUERY_NODE_OPERATOR) { + SOperatorNode* pNode = (SOperatorNode*)pMergeCondition; setJoinColumnInfo(&pInfo->leftCol, (SColumnNode*)pNode->pLeft); setJoinColumnInfo(&pInfo->rightCol, (SColumnNode*)pNode->pRight); - } else if (nodeType(pOnCondition) == QUERY_NODE_LOGIC_CONDITION) { - extractTimeCondition(pInfo, (SLogicConditionNode*)pOnCondition); + } else { + ASSERT(false); + } + + //TODO: merge these two conditions + ASSERT(pJoinNode->pOnConditions); + pInfo->pOnConditions = nodesCloneNode(pJoinNode->pOnConditions); + if (pJoinNode->node.pConditions != NULL) { + pInfo->pOtherConditions = pJoinNode->node.pConditions; } pOperator->fpSet = @@ -88,6 +95,8 @@ void setJoinColumnInfo(SColumnInfo* pColumn, const SColumnNode* pColumnNode) { void destroyMergeJoinOperator(void* param, int32_t numOfOutput) { SJoinOperatorInfo* pJoinOperator = (SJoinOperatorInfo*)param; + nodesDestroyNode(pJoinOperator->pOnConditions); + nodesDestroyNode(pJoinOperator->pOtherConditions); } static void doMergeJoinImpl(struct SOperatorInfo* pOperator, SSDataBlock* pRes) { @@ -192,7 +201,8 @@ SSDataBlock* doMergeJoin(struct SOperatorInfo* pOperator) { if (numOfNewRows == 0) { break; } - doFilter(pJoinInfo->pOnCondition, pRes); + doFilter(pJoinInfo->pOnConditions, pRes); + doFilter(pJoinInfo->pOtherConditions, pRes); if (pRes->info.rows >= pOperator->resultInfo.threshold) { break; } diff --git a/source/libs/nodes/src/nodesCloneFuncs.c b/source/libs/nodes/src/nodesCloneFuncs.c index 68654b21a6..99216f5901 100644 --- a/source/libs/nodes/src/nodesCloneFuncs.c +++ b/source/libs/nodes/src/nodesCloneFuncs.c @@ -359,6 +359,7 @@ static int32_t logicScanCopy(const SScanLogicNode* pSrc, SScanLogicNode* pDst) { static int32_t logicJoinCopy(const SJoinLogicNode* pSrc, SJoinLogicNode* pDst) { COPY_BASE_OBJECT_FIELD(node, logicNodeCopy); COPY_SCALAR_FIELD(joinType); + CLONE_NODE_FIELD(pMergeCondition); CLONE_NODE_FIELD(pOnConditions); COPY_SCALAR_FIELD(isSingleTableJoin); return TSDB_CODE_SUCCESS; diff --git a/source/libs/nodes/src/nodesCodeFuncs.c b/source/libs/nodes/src/nodesCodeFuncs.c index df5a401eb1..f1d19af4ac 100644 --- a/source/libs/nodes/src/nodesCodeFuncs.c +++ b/source/libs/nodes/src/nodesCodeFuncs.c @@ -1253,6 +1253,7 @@ static int32_t jsonToLogicPlan(const SJson* pJson, void* pObj) { static const char* jkJoinLogicPlanJoinType = "JoinType"; static const char* jkJoinLogicPlanOnConditions = "OnConditions"; +static const char* jkJoinLogicPlanMergeCondition = "MergeConditions"; static int32_t logicJoinNodeToJson(const void* pObj, SJson* pJson) { const SJoinLogicNode* pNode = (const SJoinLogicNode*)pObj; @@ -1261,6 +1262,9 @@ static int32_t logicJoinNodeToJson(const void* pObj, SJson* pJson) { if (TSDB_CODE_SUCCESS == code) { code = tjsonAddIntegerToObject(pJson, jkJoinLogicPlanJoinType, pNode->joinType); } + if (TSDB_CODE_SUCCESS == code) { + code = tjsonAddObject(pJson, jkJoinLogicPlanMergeCondition, nodeToJson, pNode->pMergeCondition); + } if (TSDB_CODE_SUCCESS == code) { code = tjsonAddObject(pJson, jkJoinLogicPlanOnConditions, nodeToJson, pNode->pOnConditions); } @@ -1616,6 +1620,7 @@ static int32_t jsonToPhysiProjectNode(const SJson* pJson, void* pObj) { } static const char* jkJoinPhysiPlanJoinType = "JoinType"; +static const char* jkJoinPhysiPlanMergeCondition = "MergeCondition"; static const char* jkJoinPhysiPlanOnConditions = "OnConditions"; static const char* jkJoinPhysiPlanTargets = "Targets"; @@ -1626,6 +1631,9 @@ static int32_t physiJoinNodeToJson(const void* pObj, SJson* pJson) { if (TSDB_CODE_SUCCESS == code) { code = tjsonAddIntegerToObject(pJson, jkJoinPhysiPlanJoinType, pNode->joinType); } + if (TSDB_CODE_SUCCESS == code) { + code = tjsonAddObject(pJson, jkJoinPhysiPlanMergeCondition, nodeToJson, pNode->pMergeCondition); + } if (TSDB_CODE_SUCCESS == code) { code = tjsonAddObject(pJson, jkJoinPhysiPlanOnConditions, nodeToJson, pNode->pOnConditions); } @@ -1647,6 +1655,9 @@ static int32_t jsonToPhysiJoinNode(const SJson* pJson, void* pObj) { if (TSDB_CODE_SUCCESS == code) { code = jsonToNodeObject(pJson, jkJoinPhysiPlanOnConditions, &pNode->pOnConditions); } + if (TSDB_CODE_SUCCESS == code) { + code = jsonToNodeObject(pJson, jkJoinPhysiPlanMergeCondition, &pNode->pMergeCondition); + } if (TSDB_CODE_SUCCESS == code) { code = jsonToNodeList(pJson, jkJoinPhysiPlanTargets, &pNode->pTargets); } diff --git a/source/libs/nodes/src/nodesTraverseFuncs.c b/source/libs/nodes/src/nodesTraverseFuncs.c index 3747dde9ed..b12e3b14c7 100644 --- a/source/libs/nodes/src/nodesTraverseFuncs.c +++ b/source/libs/nodes/src/nodesTraverseFuncs.c @@ -470,6 +470,9 @@ static EDealRes dispatchPhysiPlan(SNode* pNode, ETraversalOrder order, FNodeWalk case QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN: { SJoinPhysiNode* pJoin = (SJoinPhysiNode*)pNode; res = walkPhysiNode((SPhysiNode*)pNode, order, walker, pContext); + if (DEAL_RES_ERROR != res && DEAL_RES_END != res) { + res = walkPhysiPlan(pJoin->pMergeCondition, order, walker, pContext); + } if (DEAL_RES_ERROR != res && DEAL_RES_END != res) { res = walkPhysiPlan(pJoin->pOnConditions, order, walker, pContext); } diff --git a/source/libs/nodes/src/nodesUtilFuncs.c b/source/libs/nodes/src/nodesUtilFuncs.c index 1972010e25..10081f07e3 100644 --- a/source/libs/nodes/src/nodesUtilFuncs.c +++ b/source/libs/nodes/src/nodesUtilFuncs.c @@ -717,6 +717,7 @@ void nodesDestroyNode(SNode* pNode) { case QUERY_NODE_LOGIC_PLAN_JOIN: { SJoinLogicNode* pLogicNode = (SJoinLogicNode*)pNode; destroyLogicNode((SLogicNode*)pLogicNode); + nodesDestroyNode(pLogicNode->pMergeCondition); nodesDestroyNode(pLogicNode->pOnConditions); break; } @@ -827,6 +828,7 @@ void nodesDestroyNode(SNode* pNode) { case QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN: { SJoinPhysiNode* pPhyNode = (SJoinPhysiNode*)pNode; destroyPhysiNode((SPhysiNode*)pPhyNode); + nodesDestroyNode(pPhyNode->pMergeCondition); nodesDestroyNode(pPhyNode->pOnConditions); nodesDestroyList(pPhyNode->pTargets); break; @@ -1492,6 +1494,38 @@ int32_t nodesCollectColumns(SSelectStmt* pSelect, ESqlClause clause, const char* return TSDB_CODE_SUCCESS; } +int32_t nodesCollectColumnsFromNode(SNode* node, const char* pTableAlias, ECollectColType type, SNodeList** pCols) { + if (NULL == pCols) { + return TSDB_CODE_FAILED; + } + SCollectColumnsCxt cxt = { + .errCode = TSDB_CODE_SUCCESS, + .pTableAlias = pTableAlias, + .collectType = type, + .pCols = (NULL == *pCols ? nodesMakeList() : *pCols), + .pColHash = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK)}; + if (NULL == cxt.pCols || NULL == cxt.pColHash) { + return TSDB_CODE_OUT_OF_MEMORY; + } + *pCols = NULL; + + nodesWalkExpr(node, collectColumns, &cxt); + + taosHashCleanup(cxt.pColHash); + if (TSDB_CODE_SUCCESS != cxt.errCode) { + nodesDestroyList(cxt.pCols); + return cxt.errCode; + } + if (LIST_LENGTH(cxt.pCols) > 0) { + *pCols = cxt.pCols; + } else { + nodesDestroyList(cxt.pCols); + } + + return TSDB_CODE_SUCCESS; + +} + typedef struct SCollectFuncsCxt { int32_t errCode; FFuncClassifier classifier; diff --git a/source/libs/planner/src/planOptimizer.c b/source/libs/planner/src/planOptimizer.c index 87a5096d26..90e35fec20 100644 --- a/source/libs/planner/src/planOptimizer.c +++ b/source/libs/planner/src/planOptimizer.c @@ -540,11 +540,15 @@ static bool pushDownCondOptContainPriKeyEqualCond(SJoinLogicNode* pJoin, SNode* } return hasPrimaryKeyEqualCond; } else { - return pushDownCondOptIsPriKeyEqualCond(pJoin, pCond); + bool isPriKeyEqualCond = pushDownCondOptIsPriKeyEqualCond(pJoin, pCond); + if (isPriKeyEqualCond) { + pJoin->pMergeCondition = nodesCloneNode(pCond); + } + return isPriKeyEqualCond; } } -static int32_t pushDownCondOptCheckJoinOnCond(SOptimizeContext* pCxt, SJoinLogicNode* pJoin) { +static int32_t pushDownCondOptExtractJoinMergeCond(SOptimizeContext* pCxt, SJoinLogicNode* pJoin) { if (NULL == pJoin->pOnConditions) { return generateUsageErrMsg(pCxt->pPlanCxt->pMsg, pCxt->pPlanCxt->msgLen, TSDB_CODE_PLAN_NOT_SUPPORT_CROSS_JOIN); } @@ -560,7 +564,7 @@ static int32_t pushDownCondOptDealJoin(SOptimizeContext* pCxt, SJoinLogicNode* p } if (NULL == pJoin->node.pConditions) { - return pushDownCondOptCheckJoinOnCond(pCxt, pJoin); + return pushDownCondOptExtractJoinMergeCond(pCxt, pJoin); } SNode* pOnCond = NULL; @@ -582,7 +586,7 @@ static int32_t pushDownCondOptDealJoin(SOptimizeContext* pCxt, SJoinLogicNode* p if (TSDB_CODE_SUCCESS == code) { OPTIMIZE_FLAG_SET_MASK(pJoin->node.optimizedFlag, OPTIMIZE_FLAG_PUSH_DOWN_CONDE); pCxt->optimized = true; - code = pushDownCondOptCheckJoinOnCond(pCxt, pJoin); + code = pushDownCondOptExtractJoinMergeCond(pCxt, pJoin); } else { nodesDestroyNode(pOnCond); nodesDestroyNode(pLeftChildCond); diff --git a/source/libs/planner/src/planPhysiCreater.c b/source/libs/planner/src/planPhysiCreater.c index aac9c25f77..8a0139dfb5 100644 --- a/source/libs/planner/src/planPhysiCreater.c +++ b/source/libs/planner/src/planPhysiCreater.c @@ -609,10 +609,8 @@ static int32_t createJoinPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren int32_t code = TSDB_CODE_SUCCESS; pJoin->joinType = pJoinLogicNode->joinType; - if (NULL != pJoinLogicNode->pOnConditions) { - code = setNodeSlotId(pCxt, pLeftDesc->dataBlockId, pRightDesc->dataBlockId, pJoinLogicNode->pOnConditions, - &pJoin->pOnConditions); - } + setNodeSlotId(pCxt, pLeftDesc->dataBlockId, pRightDesc->dataBlockId, pJoinLogicNode->pMergeCondition, + &pJoin->pMergeCondition); if (TSDB_CODE_SUCCESS == code) { code = setListSlotId(pCxt, pLeftDesc->dataBlockId, pRightDesc->dataBlockId, pJoinLogicNode->node.pTargets, &pJoin->pTargets); @@ -620,6 +618,21 @@ static int32_t createJoinPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren if (TSDB_CODE_SUCCESS == code) { code = addDataBlockSlots(pCxt, pJoin->pTargets, pJoin->node.pOutputDataBlockDesc); } + + SNodeList* condCols = nodesMakeList(); + if (TSDB_CODE_SUCCESS == code && NULL != pJoinLogicNode->pOnConditions) { + code = nodesCollectColumnsFromNode(pJoinLogicNode->pOnConditions, NULL, COLLECT_COL_TYPE_ALL, &condCols); + } + if (TSDB_CODE_SUCCESS == code) { + code = addDataBlockSlots(pCxt, condCols, pJoin->node.pOutputDataBlockDesc); + nodesDestroyList(condCols); + } + + if (TSDB_CODE_SUCCESS == code && NULL != pJoinLogicNode->pOnConditions) { + code = setNodeSlotId(pCxt, ((SPhysiNode*)pJoin)->pOutputDataBlockDesc->dataBlockId, -1, pJoinLogicNode->pOnConditions, + &pJoin->pOnConditions); + } + if (TSDB_CODE_SUCCESS == code) { code = setConditionsSlotId(pCxt, (const SLogicNode*)pJoinLogicNode, (SPhysiNode*)pJoin); } From a7c9ed913384ffe077436c397bc674e278590c4e Mon Sep 17 00:00:00 2001 From: slzhou Date: Mon, 4 Jul 2022 20:01:07 +0800 Subject: [PATCH 3/8] feat: push condition to child join operator --- source/libs/planner/src/planOptimizer.c | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/source/libs/planner/src/planOptimizer.c b/source/libs/planner/src/planOptimizer.c index 90e35fec20..43aebc4639 100644 --- a/source/libs/planner/src/planOptimizer.c +++ b/source/libs/planner/src/planOptimizer.c @@ -480,12 +480,18 @@ static int32_t pushDownCondOptPushCondToProject(SOptimizeContext* pCxt, SProject return pushDownCondOptAppendCond(&pProject->node.pConditions, pCond); } +static int32_t pushDownCondOptPushCondToJoin(SOptimizeContext* pCxt, SJoinLogicNode * pJoin, SNode** pCond) { + return pushDownCondOptAppendCond(&pJoin->node.pConditions, pCond); +} + static int32_t pushDownCondOptPushCondToChild(SOptimizeContext* pCxt, SLogicNode* pChild, SNode** pCond) { switch (nodeType(pChild)) { case QUERY_NODE_LOGIC_PLAN_SCAN: return pushDownCondOptPushCondToScan(pCxt, (SScanLogicNode*)pChild, pCond); case QUERY_NODE_LOGIC_PLAN_PROJECT: return pushDownCondOptPushCondToProject(pCxt, (SProjectLogicNode*)pChild, pCond); + case QUERY_NODE_LOGIC_PLAN_JOIN: + return pushDownCondOptPushCondToJoin(pCxt, (SJoinLogicNode*)pChild, pCond); default: break; } @@ -724,7 +730,8 @@ static int32_t pushDownCondOptDealAgg(SOptimizeContext* pCxt, SAggLogicNode* pAg //TODO: remove it after full implementation of pushing down to child if (1 != LIST_LENGTH(pAgg->node.pChildren) || QUERY_NODE_LOGIC_PLAN_SCAN != nodeType(nodesListGetNode(pAgg->node.pChildren, 0)) && - QUERY_NODE_LOGIC_PLAN_PROJECT != nodeType(nodesListGetNode(pAgg->node.pChildren, 0))) { + QUERY_NODE_LOGIC_PLAN_PROJECT != nodeType(nodesListGetNode(pAgg->node.pChildren, 0)) && + QUERY_NODE_LOGIC_PLAN_JOIN != nodeType(nodesListGetNode(pAgg->node.pChildren, 0))) { return TSDB_CODE_SUCCESS; } From 73e0e09dd8d1a3386279d2eb8c26a85b23deecd8 Mon Sep 17 00:00:00 2001 From: slzhou Date: Tue, 5 Jul 2022 13:42:49 +0800 Subject: [PATCH 4/8] feat: split join on condition into merge condition and on condition --- source/libs/executor/inc/executorimpl.h | 3 +- source/libs/executor/src/joinoperator.c | 26 ++++++--- source/libs/planner/src/planOptimizer.c | 75 ++++++++++++++++++++++--- 3 files changed, 85 insertions(+), 19 deletions(-) diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 8dd7057f0c..f0f0361031 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -680,8 +680,7 @@ typedef struct SJoinOperatorInfo { SSDataBlock *pRight; int32_t rightPos; SColumnInfo rightCol; - SNode *pOnConditions; - SNode *pOtherConditions; + SNode *pCondAfterMerge; } SJoinOperatorInfo; #define OPTR_IS_OPENED(_optr) (((_optr)->status & OP_OPENED) == OP_OPENED) diff --git a/source/libs/executor/src/joinoperator.c b/source/libs/executor/src/joinoperator.c index 71f04815dc..e9995ed77a 100644 --- a/source/libs/executor/src/joinoperator.c +++ b/source/libs/executor/src/joinoperator.c @@ -62,11 +62,19 @@ SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t ASSERT(false); } - //TODO: merge these two conditions - ASSERT(pJoinNode->pOnConditions); - pInfo->pOnConditions = nodesCloneNode(pJoinNode->pOnConditions); - if (pJoinNode->node.pConditions != NULL) { - pInfo->pOtherConditions = pJoinNode->node.pConditions; + if (pJoinNode->pOnConditions != NULL && pJoinNode->node.pConditions != NULL) { + pInfo->pCondAfterMerge = nodesMakeNode(QUERY_NODE_LOGIC_CONDITION); + SLogicConditionNode* pLogicCond = (SLogicConditionNode*)(pInfo->pCondAfterMerge); + pLogicCond->pParameterList = nodesMakeList(); + nodesListMakeAppend(&pLogicCond->pParameterList, nodesCloneNode(pJoinNode->pOnConditions)); + nodesListMakeAppend(&pLogicCond->pParameterList, nodesCloneNode(pJoinNode->node.pConditions)); + pLogicCond->condType = LOGIC_COND_TYPE_AND; + } else if (pJoinNode->pOnConditions != NULL) { + pInfo->pCondAfterMerge = nodesCloneNode(pJoinNode->pOnConditions); + } else if (pJoinNode->node.pConditions != NULL) { + pInfo->pCondAfterMerge = nodesCloneNode(pJoinNode->node.pConditions); + } else { + pInfo->pCondAfterMerge = NULL; } pOperator->fpSet = @@ -95,8 +103,7 @@ void setJoinColumnInfo(SColumnInfo* pColumn, const SColumnNode* pColumnNode) { void destroyMergeJoinOperator(void* param, int32_t numOfOutput) { SJoinOperatorInfo* pJoinOperator = (SJoinOperatorInfo*)param; - nodesDestroyNode(pJoinOperator->pOnConditions); - nodesDestroyNode(pJoinOperator->pOtherConditions); + nodesDestroyNode(pJoinOperator->pCondAfterMerge); } static void doMergeJoinImpl(struct SOperatorInfo* pOperator, SSDataBlock* pRes) { @@ -201,8 +208,9 @@ SSDataBlock* doMergeJoin(struct SOperatorInfo* pOperator) { if (numOfNewRows == 0) { break; } - doFilter(pJoinInfo->pOnConditions, pRes); - doFilter(pJoinInfo->pOtherConditions, pRes); + if (pJoinInfo->pCondAfterMerge != NULL) { + doFilter(pJoinInfo->pCondAfterMerge, pRes); + } if (pRes->info.rows >= pOperator->resultInfo.threshold) { break; } diff --git a/source/libs/planner/src/planOptimizer.c b/source/libs/planner/src/planOptimizer.c index 7e0c5d6f5a..495ef9e698 100644 --- a/source/libs/planner/src/planOptimizer.c +++ b/source/libs/planner/src/planOptimizer.c @@ -546,15 +546,11 @@ static bool pushDownCondOptContainPriKeyEqualCond(SJoinLogicNode* pJoin, SNode* } return hasPrimaryKeyEqualCond; } else { - bool isPriKeyEqualCond = pushDownCondOptIsPriKeyEqualCond(pJoin, pCond); - if (isPriKeyEqualCond) { - pJoin->pMergeCondition = nodesCloneNode(pCond); - } - return isPriKeyEqualCond; + return pushDownCondOptIsPriKeyEqualCond(pJoin, pCond); } } -static int32_t pushDownCondOptExtractJoinMergeCond(SOptimizeContext* pCxt, SJoinLogicNode* pJoin) { +static int32_t pushDownCondOptCheckJoinOnCond(SOptimizeContext* pCxt, SJoinLogicNode* pJoin) { if (NULL == pJoin->pOnConditions) { return generateUsageErrMsg(pCxt->pPlanCxt->pMsg, pCxt->pPlanCxt->msgLen, TSDB_CODE_PLAN_NOT_SUPPORT_CROSS_JOIN); } @@ -564,13 +560,61 @@ static int32_t pushDownCondOptExtractJoinMergeCond(SOptimizeContext* pCxt, SJoin return TSDB_CODE_SUCCESS; } +static int32_t pushDownCondOptPartJoinOnCondLogicCond(SJoinLogicNode* pJoin, SNode** ppMergeCond, SNode** ppOnCond) { + SLogicConditionNode* pLogicCond = (SLogicConditionNode*)(pJoin->pOnConditions); + + int32_t code = TSDB_CODE_SUCCESS; + SNodeList* pOnConds = NULL; + SNode* pCond = NULL; + FOREACH(pCond, pLogicCond->pParameterList) { + if (pushDownCondOptIsPriKeyEqualCond(pJoin, pCond)) { + *ppMergeCond = nodesCloneNode(pCond); + } else { + code = nodesListMakeAppend(&pOnConds, nodesCloneNode(pCond)); + } + } + + SNode* pTempOnCond = NULL; + if (TSDB_CODE_SUCCESS == code) { + code = nodesMergeConds(&pTempOnCond, &pOnConds); + } + + if (TSDB_CODE_SUCCESS == code && NULL != *ppMergeCond) { + *ppOnCond = pTempOnCond; + nodesDestroyNode(pJoin->pOnConditions); + pJoin->pOnConditions = NULL; + return TSDB_CODE_SUCCESS; + } else { + nodesDestroyList(pOnConds); + nodesDestroyNode(pTempOnCond); + return TSDB_CODE_PLAN_INTERNAL_ERROR; + } +} + +static int32_t pushDownCondOptPartJoinOnCond(SJoinLogicNode* pJoin, SNode** ppMergeCond, SNode** ppOnCond) { + if (QUERY_NODE_LOGIC_CONDITION == nodeType(pJoin->pOnConditions) && + LOGIC_COND_TYPE_AND == ((SLogicConditionNode*)(pJoin->pOnConditions))->condType) { + return pushDownCondOptPartJoinOnCondLogicCond(pJoin, ppMergeCond, ppOnCond); + } + + if (pushDownCondOptIsPriKeyEqualCond(pJoin, pJoin->pOnConditions)) { + *ppMergeCond = nodesCloneNode(pJoin->pOnConditions); + *ppOnCond = NULL; + nodesDestroyNode(pJoin->pOnConditions); + pJoin->pOnConditions = NULL; + return TSDB_CODE_SUCCESS; + } else { + return TSDB_CODE_PLAN_INTERNAL_ERROR; + } +} + static int32_t pushDownCondOptDealJoin(SOptimizeContext* pCxt, SJoinLogicNode* pJoin) { if (OPTIMIZE_FLAG_TEST_MASK(pJoin->node.optimizedFlag, OPTIMIZE_FLAG_PUSH_DOWN_CONDE)) { return TSDB_CODE_SUCCESS; } if (NULL == pJoin->node.pConditions) { - return pushDownCondOptExtractJoinMergeCond(pCxt, pJoin); + return pushDownCondOptCheckJoinOnCond(pCxt, pJoin); } SNode* pOnCond = NULL; @@ -589,11 +633,26 @@ static int32_t pushDownCondOptDealJoin(SOptimizeContext* pCxt, SJoinLogicNode* p pushDownCondOptPushCondToChild(pCxt, (SLogicNode*)nodesListGetNode(pJoin->node.pChildren, 1), &pRightChildCond); } + if (TSDB_CODE_SUCCESS == code) { + code = pushDownCondOptCheckJoinOnCond(pCxt, pJoin); + } + + SNode* pJoinMergeCond = NULL; + SNode* pJoinOnCond = NULL; + if (TSDB_CODE_SUCCESS == code) { + code = pushDownCondOptPartJoinOnCond(pJoin, &pJoinMergeCond, &pJoinOnCond); + } + if (TSDB_CODE_SUCCESS == code && NULL != pJoinMergeCond) { + pJoin->pMergeCondition = pJoinMergeCond; + pJoin->pOnConditions = pJoinOnCond; + } + if (TSDB_CODE_SUCCESS == code) { OPTIMIZE_FLAG_SET_MASK(pJoin->node.optimizedFlag, OPTIMIZE_FLAG_PUSH_DOWN_CONDE); pCxt->optimized = true; - code = pushDownCondOptExtractJoinMergeCond(pCxt, pJoin); } else { + nodesDestroyNode(pJoinMergeCond); + nodesDestroyNode(pJoinOnCond); nodesDestroyNode(pOnCond); nodesDestroyNode(pLeftChildCond); nodesDestroyNode(pRightChildCond); From 7088166e984ef4b28f83a99d57d5a53405c0a468 Mon Sep 17 00:00:00 2001 From: slzhou Date: Tue, 5 Jul 2022 14:02:33 +0800 Subject: [PATCH 5/8] fix: fix memory leak when uv_loop_close --- source/libs/function/src/tudf.c | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/source/libs/function/src/tudf.c b/source/libs/function/src/tudf.c index da9474ede0..1bc759e833 100644 --- a/source/libs/function/src/tudf.c +++ b/source/libs/function/src/tudf.c @@ -1565,6 +1565,10 @@ void constructUdfService(void *argsThread) { //TODO return value of uv_run uv_run(&udfc->uvLoop, UV_RUN_DEFAULT); uv_loop_close(&udfc->uvLoop); + + uv_walk(&udfc->uvLoop, udfUdfdCloseWalkCb, NULL); + uv_run(&udfc->uvLoop, UV_RUN_DEFAULT); + uv_loop_close(&udfc->uvLoop); } int32_t udfcOpen() { From 4a494e234ec2e6d3ee07fac184dc420a9fba57b3 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Tue, 5 Jul 2022 14:40:48 +0800 Subject: [PATCH 6/8] test: cases with 100 dnodes --- source/dnode/mnode/impl/src/mndDnode.c | 5 ++--- source/dnode/mnode/sdb/src/sdb.c | 2 +- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/source/dnode/mnode/impl/src/mndDnode.c b/source/dnode/mnode/impl/src/mndDnode.c index 1073ebc316..6ead922d95 100644 --- a/source/dnode/mnode/impl/src/mndDnode.c +++ b/source/dnode/mnode/impl/src/mndDnode.c @@ -274,15 +274,14 @@ static void mndGetDnodeData(SMnode *pMnode, SArray *pDnodeEps) { SDnodeEp dnodeEp = {0}; dnodeEp.id = pDnode->id; - dnodeEp.isMnode = 0; dnodeEp.ep.port = pDnode->port; memcpy(dnodeEp.ep.fqdn, pDnode->fqdn, TSDB_FQDN_LEN); + sdbRelease(pSdb, pDnode); + dnodeEp.isMnode = 0; if (mndIsMnode(pMnode, pDnode->id)) { dnodeEp.isMnode = 1; } - - sdbRelease(pSdb, pDnode); taosArrayPush(pDnodeEps, &dnodeEp); } } diff --git a/source/dnode/mnode/sdb/src/sdb.c b/source/dnode/mnode/sdb/src/sdb.c index fbf66da632..3db0087334 100644 --- a/source/dnode/mnode/sdb/src/sdb.c +++ b/source/dnode/mnode/sdb/src/sdb.c @@ -131,7 +131,7 @@ int32_t sdbSetTable(SSdb *pSdb, SSdbTable table) { hashType = TSDB_DATA_TYPE_BINARY; } - SHashObj *hash = taosHashInit(64, taosGetDefaultHashFunction(hashType), true, HASH_NO_LOCK); + SHashObj *hash = taosHashInit(64, taosGetDefaultHashFunction(hashType), true, HASH_ENTRY_LOCK); if (hash == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; From b719e3ce24b997854865913dd74f99043d79f8d7 Mon Sep 17 00:00:00 2001 From: slzhou Date: Tue, 5 Jul 2022 14:56:40 +0800 Subject: [PATCH 7/8] fix: extract merge cond when no where condition --- source/libs/planner/src/planOptimizer.c | 34 +++++++++++++++---------- 1 file changed, 20 insertions(+), 14 deletions(-) diff --git a/source/libs/planner/src/planOptimizer.c b/source/libs/planner/src/planOptimizer.c index 495ef9e698..f85bc789aa 100644 --- a/source/libs/planner/src/planOptimizer.c +++ b/source/libs/planner/src/planOptimizer.c @@ -608,13 +608,31 @@ static int32_t pushDownCondOptPartJoinOnCond(SJoinLogicNode* pJoin, SNode** ppMe } } +static int32_t pushDownCondOptJoinExtractMergeCond(SOptimizeContext* pCxt, SJoinLogicNode* pJoin) { + int32_t code = pushDownCondOptCheckJoinOnCond(pCxt, pJoin); + SNode* pJoinMergeCond = NULL; + SNode* pJoinOnCond = NULL; + if (TSDB_CODE_SUCCESS == code) { + code = pushDownCondOptPartJoinOnCond(pJoin, &pJoinMergeCond, &pJoinOnCond); + } + if (TSDB_CODE_SUCCESS == code) { + pJoin->pMergeCondition = pJoinMergeCond; + pJoin->pOnConditions = pJoinOnCond; + } else { + nodesDestroyNode(pJoinMergeCond); + nodesDestroyNode(pJoinOnCond); + } + return code; +} + static int32_t pushDownCondOptDealJoin(SOptimizeContext* pCxt, SJoinLogicNode* pJoin) { if (OPTIMIZE_FLAG_TEST_MASK(pJoin->node.optimizedFlag, OPTIMIZE_FLAG_PUSH_DOWN_CONDE)) { return TSDB_CODE_SUCCESS; } if (NULL == pJoin->node.pConditions) { - return pushDownCondOptCheckJoinOnCond(pCxt, pJoin); + int32_t code = pushDownCondOptJoinExtractMergeCond(pCxt, pJoin); + return code; } SNode* pOnCond = NULL; @@ -634,25 +652,13 @@ static int32_t pushDownCondOptDealJoin(SOptimizeContext* pCxt, SJoinLogicNode* p } if (TSDB_CODE_SUCCESS == code) { - code = pushDownCondOptCheckJoinOnCond(pCxt, pJoin); - } - - SNode* pJoinMergeCond = NULL; - SNode* pJoinOnCond = NULL; - if (TSDB_CODE_SUCCESS == code) { - code = pushDownCondOptPartJoinOnCond(pJoin, &pJoinMergeCond, &pJoinOnCond); - } - if (TSDB_CODE_SUCCESS == code && NULL != pJoinMergeCond) { - pJoin->pMergeCondition = pJoinMergeCond; - pJoin->pOnConditions = pJoinOnCond; + code = pushDownCondOptJoinExtractMergeCond(pCxt, pJoin); } if (TSDB_CODE_SUCCESS == code) { OPTIMIZE_FLAG_SET_MASK(pJoin->node.optimizedFlag, OPTIMIZE_FLAG_PUSH_DOWN_CONDE); pCxt->optimized = true; } else { - nodesDestroyNode(pJoinMergeCond); - nodesDestroyNode(pJoinOnCond); nodesDestroyNode(pOnCond); nodesDestroyNode(pLeftChildCond); nodesDestroyNode(pRightChildCond); From 64d24560739800207e876534428c01df5be74c57 Mon Sep 17 00:00:00 2001 From: slzhou Date: Tue, 5 Jul 2022 15:16:52 +0800 Subject: [PATCH 8/8] fix: when there are no where condtions --- source/libs/planner/src/planOptimizer.c | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/source/libs/planner/src/planOptimizer.c b/source/libs/planner/src/planOptimizer.c index f85bc789aa..a4f30228d5 100644 --- a/source/libs/planner/src/planOptimizer.c +++ b/source/libs/planner/src/planOptimizer.c @@ -632,6 +632,10 @@ static int32_t pushDownCondOptDealJoin(SOptimizeContext* pCxt, SJoinLogicNode* p if (NULL == pJoin->node.pConditions) { int32_t code = pushDownCondOptJoinExtractMergeCond(pCxt, pJoin); + if (TSDB_CODE_SUCCESS == code) { + OPTIMIZE_FLAG_SET_MASK(pJoin->node.optimizedFlag, OPTIMIZE_FLAG_PUSH_DOWN_CONDE); + pCxt->optimized = true; + } return code; }