xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

enh: optimize group join performance

This commit is contained in:
dapan1121 2024-04-23 09:14:42 +08:00
parent 5f0a23bbda
commit 94c7f32247
4 changed files with 18 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/libs/nodes/plannodes.h
View File

@ -153,6 +153,7 @@ typedef struct SJoinLogicNode {
bool seqWinGroup;
bool grpJoin;
bool hashJoinHint;
bool batchScanHint;
// FOR HASH JOIN
int32_t timeRangeTarget; //table onCond filter

1
source/libs/nodes/src/nodesCloneFuncs.c
View File

@ -511,6 +511,7 @@ static int32_t logicJoinCopy(const SJoinLogicNode* pSrc, SJoinLogicNode* pDst) {
COPY_SCALAR_FIELD(seqWinGroup);
COPY_SCALAR_FIELD(grpJoin);
COPY_SCALAR_FIELD(hashJoinHint);
COPY_SCALAR_FIELD(batchScanHint);
CLONE_NODE_FIELD(pLeftOnCond);
CLONE_NODE_FIELD(pRightOnCond);
COPY_SCALAR_FIELD(timeRangeTarget);

1
source/libs/planner/src/planLogicCreater.c
View File

@ -582,6 +582,7 @@ static int32_t createJoinLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
pJoin->node.inputTsOrder = ORDER_ASC;
pJoin->node.groupAction = GROUP_ACTION_CLEAR;
pJoin->hashJoinHint = getHashJoinOptHint(pSelect->pHint);
pJoin->batchScanHint = getBatchScanOptionFromHint(pSelect->pHint);
pJoin->node.requireDataOrder = pJoin->hashJoinHint ? DATA_ORDER_LEVEL_NONE : DATA_ORDER_LEVEL_GLOBAL;
pJoin->node.resultDataOrder = DATA_ORDER_LEVEL_NONE;
pJoin->isLowLevelJoin = pJoinTable->isLowLevelJoin;

26
source/libs/planner/src/planOptimizer.c
View File

@ -2751,16 +2751,20 @@ static bool partTagsIsOptimizableNode(SLogicNode* pNode) {
if (!ret) return ret;
switch (nodeType(pNode)) {
case QUERY_NODE_LOGIC_PLAN_PARTITION: {
if (pNode->pParent && nodeType(pNode->pParent) == QUERY_NODE_LOGIC_PLAN_WINDOW) {
SWindowLogicNode* pWindow = (SWindowLogicNode*)pNode->pParent;
if (pWindow->winType == WINDOW_TYPE_INTERVAL) {
// if interval has slimit, we push down partition node to scan, and scan will set groupOrderScan to true
// we want to skip groups of blocks after slimit satisfied
// if interval only has limit, we do not push down partition node to scan
// we want to get grouped output from partition node and make use of limit
// if no slimit and no limit, we push down partition node and groupOrderScan is false, cause we do not need
// group ordered output
if (!pWindow->node.pSlimit && pWindow->node.pLimit) ret = false;
if (pNode->pParent) {
if (nodeType(pNode->pParent) == QUERY_NODE_LOGIC_PLAN_WINDOW) {
SWindowLogicNode* pWindow = (SWindowLogicNode*)pNode->pParent;
if (pWindow->winType == WINDOW_TYPE_INTERVAL) {
// if interval has slimit, we push down partition node to scan, and scan will set groupOrderScan to true
// we want to skip groups of blocks after slimit satisfied
// if interval only has limit, we do not push down partition node to scan
// we want to get grouped output from partition node and make use of limit
// if no slimit and no limit, we push down partition node and groupOrderScan is false, cause we do not need
// group ordered output
if (!pWindow->node.pSlimit && pWindow->node.pLimit) ret = false;
}
} else if (nodeType(pNode->pParent) == QUERY_NODE_LOGIC_PLAN_JOIN) {
ret = false;
}
}
} break;
@ -5607,7 +5611,7 @@ static int32_t grpJoinOptPartByTags(SLogicNode* pNode) {
static int32_t grpJoinOptRewriteGroupJoin(SOptimizeContext* pCxt, SLogicNode* pNode, SLogicSubplan* pLogicSubplan) {
SJoinLogicNode* pJoin = (SJoinLogicNode*)pNode;
int32_t code = (pJoin->allEqTags && !pJoin->hasSubQuery) ? grpJoinOptPartByTags(pNode) : grpJoinOptInsertPartitionNode(pNode);
int32_t code = (pJoin->allEqTags && !pJoin->hasSubQuery && !pJoin->batchScanHint) ? grpJoinOptPartByTags(pNode) : grpJoinOptInsertPartitionNode(pNode);
if (TSDB_CODE_SUCCESS == code) {
pJoin->grpJoin = true;
pCxt->optimized = true;