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

TD-13990 implement interval plan

This commit is contained in:
Xiaoyu Wang 2022-03-11 05:11:25 -05:00
parent 0595d9f1a6
commit e5286898c0
3 changed files with 30 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/libs/nodes/nodes.h
View File

@ -101,6 +101,7 @@ typedef enum ENodeType {
QUERY_NODE_LOGIC_PLAN_PROJECT,
QUERY_NODE_LOGIC_PLAN_VNODE_MODIF,
QUERY_NODE_LOGIC_PLAN_EXCHANGE,
QUERY_NODE_LOGIC_PLAN_WINDOW,
QUERY_NODE_LOGIC_SUBPLAN,
QUERY_NODE_LOGIC_PLAN,
@ -115,6 +116,7 @@ typedef enum ENodeType {
QUERY_NODE_PHYSICAL_PLAN_AGG,
QUERY_NODE_PHYSICAL_PLAN_EXCHANGE,
QUERY_NODE_PHYSICAL_PLAN_SORT,
QUERY_NODE_PHYSICAL_PLAN_INTERVAL,
QUERY_NODE_PHYSICAL_PLAN_DISPATCH,
QUERY_NODE_PHYSICAL_PLAN_INSERT,
QUERY_NODE_PHYSICAL_SUBPLAN,

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

@ -80,6 +80,11 @@ typedef struct SExchangeLogicNode {
int32_t srcGroupId;
} SExchangeLogicNode;
typedef struct SWindowLogicNode {
SLogicNode node;
SNode* pWindow;
} SWindowLogicNode;
typedef enum ESubplanType {
SUBPLAN_TYPE_MERGE = 1,
SUBPLAN_TYPE_PARTIAL,
@ -191,6 +196,14 @@ typedef struct SExchangePhysiNode {
SNodeList* pSrcEndPoints; // element is SDownstreamSource, scheduler fill by calling qSetSuplanExecutionNode
} SExchangePhysiNode;
typedef struct SIntervalPhysiNode {
SPhysiNode node;
int64_t interval;
int64_t sliding;
int64_t offset;
SFillNode* pFill;
} SIntervalPhysiNode;
typedef struct SDataSinkNode {
ENodeType type;
SDataBlockDescNode* pInputDataBlockDesc;

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

@ -257,6 +257,18 @@ static SNodeList* createColumnByRewriteExps(SLogicPlanContext* pCxt, SNodeList*
return cxt.pList;
}
static SLogicNode* createWindowLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect) {
if (NULL == pSelect->pWindow) {
return NULL;
}
SWindowLogicNode* pAgg = nodesMakeNode(QUERY_NODE_LOGIC_PLAN_WINDOW);
CHECK_ALLOC(pAgg, NULL);
pAgg->node.id = pCxt->planNodeId++;
}
static SLogicNode* createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect) {
SNodeList* pAggFuncs = NULL;
CHECK_CODE(nodesCollectFuncs(pSelect, fmIsAggFunc, &pAggFuncs), NULL);
@ -345,6 +357,9 @@ static SLogicNode* createSelectLogicNode(SLogicPlanContext* pCxt, SSelectStmt* p
pRoot->pConditions = nodesCloneNode(pSelect->pWhere);
CHECK_ALLOC(pRoot->pConditions, pRoot);
}
if (TSDB_CODE_SUCCESS == pCxt->errCode) {
pRoot = pushLogicNode(pCxt, pRoot, createWindowLogicNode(pCxt, pSelect));
}
if (TSDB_CODE_SUCCESS == pCxt->errCode) {
pRoot = pushLogicNode(pCxt, pRoot, createAggLogicNode(pCxt, pSelect));
}