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

order by plan implement

This commit is contained in:
Xiaoyu Wang 2022-03-26 02:58:44 -04:00
parent f6bcae1913
commit 10f073e79c
11 changed files with 215 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -120,6 +120,7 @@ typedef enum ENodeType {
QUERY_NODE_LOGIC_PLAN_VNODE_MODIF,
QUERY_NODE_LOGIC_PLAN_EXCHANGE,
QUERY_NODE_LOGIC_PLAN_WINDOW,
QUERY_NODE_LOGIC_PLAN_SORT,
QUERY_NODE_LOGIC_SUBPLAN,
QUERY_NODE_LOGIC_PLAN,

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

@ -99,6 +99,11 @@ typedef struct SWindowLogicNode {
int64_t sessionGap;
} SWindowLogicNode;
typedef struct SSortLogicNode {
SLogicNode node;
SNodeList* pSortKeys;
} SSortLogicNode;
typedef enum ESubplanType {
SUBPLAN_TYPE_MERGE = 1,
SUBPLAN_TYPE_PARTIAL,
@ -198,7 +203,7 @@ typedef struct SJoinPhysiNode {
typedef struct SAggPhysiNode {
SPhysiNode node;
SNodeList* pExprs; // these are expression list of group_by_clause and parameter expression of aggregate function
SNodeList* pGroupKeys; // SColumnRefNode list
SNodeList* pGroupKeys;
SNodeList* pAggFuncs;
} SAggPhysiNode;
@ -236,6 +241,12 @@ typedef struct SSessionWinodwPhysiNode {
int64_t gap;
} SSessionWinodwPhysiNode;
typedef struct SSortPhysiNode {
SPhysiNode node;
SNodeList* pExprs; // these are expression list of order_by_clause and parameter expression of aggregate function
SNodeList* pSortKeys; // element is SOrderByExprNode, and SOrderByExprNode::pExpr is SColumnNode
} SSortPhysiNode;
typedef struct SDataSinkNode {
ENodeType type;
SDataBlockDescNode* pInputDataBlockDesc;

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

@ -19,6 +19,11 @@
#include "taos.h"
#include "taoserror.h"
#define COPY_ALL_SCALAR_FIELDS \
do { \
memcpy((pDst), (pSrc), sizeof(*pSrc)); \
} while (0)
#define COPY_SCALAR_FIELD(fldname) \
do { \
(pDst)->fldname = (pSrc)->fldname; \
@ -195,6 +200,12 @@ static SNode* groupingSetNodeCopy(const SGroupingSetNode* pSrc, SGroupingSetNode
return (SNode*)pDst;
}
static SNode* orderByExprNodeCopy(const SOrderByExprNode* pSrc, SOrderByExprNode* pDst) {
COPY_ALL_SCALAR_FIELDS;
CLONE_NODE_FIELD(pExpr);
return (SNode*)pDst;
}
static SNode* fillNodeCopy(const SFillNode* pSrc, SFillNode* pDst) {
COPY_SCALAR_FIELD(mode);
CLONE_NODE_FIELD(pValues);
@ -280,6 +291,12 @@ static SNode* logicWindowCopy(const SWindowLogicNode* pSrc, SWindowLogicNode* pD
return (SNode*)pDst;
}
static SNode* logicSortCopy(const SSortLogicNode* pSrc, SSortLogicNode* pDst) {
COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
CLONE_NODE_LIST_FIELD(pSortKeys);
return (SNode*)pDst;
}
static SNode* logicSubplanCopy(const SLogicSubplan* pSrc, SLogicSubplan* pDst) {
CLONE_NODE_FIELD(pNode);
COPY_SCALAR_FIELD(subplanType);
@ -339,6 +356,7 @@ SNodeptr nodesCloneNode(const SNodeptr pNode) {
case QUERY_NODE_GROUPING_SET:
return groupingSetNodeCopy((const SGroupingSetNode*)pNode, (SGroupingSetNode*)pDst);
case QUERY_NODE_ORDER_BY_EXPR:
return orderByExprNodeCopy((const SOrderByExprNode*)pNode, (SOrderByExprNode*)pDst);
case QUERY_NODE_LIMIT:
break;
case QUERY_NODE_FILL:
@ -361,6 +379,8 @@ SNodeptr nodesCloneNode(const SNodeptr pNode) {
return logicExchangeCopy((const SExchangeLogicNode*)pNode, (SExchangeLogicNode*)pDst);
case QUERY_NODE_LOGIC_PLAN_WINDOW:
return logicWindowCopy((const SWindowLogicNode*)pNode, (SWindowLogicNode*)pDst);
case QUERY_NODE_LOGIC_PLAN_SORT:
return logicSortCopy((const SSortLogicNode*)pNode, (SSortLogicNode*)pDst);
case QUERY_NODE_LOGIC_SUBPLAN:
return logicSubplanCopy((const SLogicSubplan*)pNode, (SLogicSubplan*)pDst);
default:

71
source/libs/nodes/src/nodesCodeFuncs.c
View File

@ -920,6 +920,37 @@ static int32_t jsonToPhysiExchangeNode(const SJson* pJson, void* pObj) {
return code;
}
static const char* jkSortPhysiPlanExprs = "Exprs";
static const char* jkSortPhysiPlanSortKeys = "SortKeys";
static int32_t physiSortNodeToJson(const void* pObj, SJson* pJson) {
const SSortPhysiNode* pNode = (const SSortPhysiNode*)pObj;
int32_t code = physicPlanNodeToJson(pObj, pJson);
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkSortPhysiPlanExprs, pNode->pExprs);
}
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkSortPhysiPlanSortKeys, pNode->pSortKeys);
}
return code;
}
static int32_t jsonToPhysiSortNode(const SJson* pJson, void* pObj) {
SSortPhysiNode* pNode = (SSortPhysiNode*)pObj;
int32_t code = jsonToPhysicPlanNode(pJson, pObj);
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkSortPhysiPlanExprs, &pNode->pExprs);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkSortPhysiPlanSortKeys, &pNode->pSortKeys);
}
return code;
}
static const char* jkWindowPhysiPlanExprs = "Exprs";
static const char* jkWindowPhysiPlanFuncs = "Funcs";
@ -1807,6 +1838,38 @@ static int32_t groupingSetNodeToJson(const void* pObj, SJson* pJson) {
return code;
}
static const char* jkOrderByExprExpr = "Expr";
static const char* jkOrderByExprOrder = "Order";
static const char* jkOrderByExprNullOrder = "NullOrder";
static int32_t orderByExprNodeToJson(const void* pObj, SJson* pJson) {
const SOrderByExprNode* pNode = (const SOrderByExprNode*)pObj;
int32_t code = tjsonAddObject(pJson, jkOrderByExprExpr, nodeToJson, pNode->pExpr);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkOrderByExprOrder, pNode->order);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkOrderByExprNullOrder, pNode->nullOrder);
}
return code;
}
static int32_t jsonToOrderByExprNode(const SJson* pJson, void* pObj) {
SOrderByExprNode* pNode = (SOrderByExprNode*)pObj;
int32_t code = jsonToNodeObject(pJson, jkOrderByExprExpr, &pNode->pExpr);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetNumberValue(pJson, jkOrderByExprOrder, pNode->order);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetNumberValue(pJson, jkOrderByExprNullOrder, pNode->nullOrder);
}
return code;
}
static const char* jkIntervalWindowInterval = "Interval";
static const char* jkIntervalWindowOffset = "Offset";
static const char* jkIntervalWindowSliding = "Sliding";
@ -2155,6 +2218,7 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
case QUERY_NODE_GROUPING_SET:
return groupingSetNodeToJson(pObj, pJson);
case QUERY_NODE_ORDER_BY_EXPR:
return orderByExprNodeToJson(pObj, pJson);
case QUERY_NODE_LIMIT:
case QUERY_NODE_STATE_WINDOW:
case QUERY_NODE_SESSION_WINDOW:
@ -2218,7 +2282,7 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
case QUERY_NODE_PHYSICAL_PLAN_EXCHANGE:
return physiExchangeNodeToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_SORT:
break;
return physiSortNodeToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
return physiIntervalNodeToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
@ -2258,7 +2322,8 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
// break;
// case QUERY_NODE_GROUPING_SET:
// return jsonToGroupingSetNode(pJson, pObj);
// case QUERY_NODE_ORDER_BY_EXPR:
case QUERY_NODE_ORDER_BY_EXPR:
return jsonToOrderByExprNode(pJson, pObj);
// case QUERY_NODE_LIMIT:
// case QUERY_NODE_STATE_WINDOW:
// case QUERY_NODE_SESSION_WINDOW:
@ -2307,6 +2372,8 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
return jsonToPhysiAggNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_EXCHANGE:
return jsonToPhysiExchangeNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_SORT:
return jsonToPhysiSortNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
return jsonToPhysiIntervalNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:

4
source/libs/nodes/src/nodesTraverseFuncs.c
View File

@ -294,10 +294,10 @@ void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker wa
case SQL_CLAUSE_GROUP_BY:
nodesWalkNode(pSelect->pHaving, walker, pContext);
case SQL_CLAUSE_HAVING:
nodesWalkList(pSelect->pProjectionList, walker, pContext);
case SQL_CLAUSE_SELECT:
nodesWalkList(pSelect->pOrderByList, walker, pContext);
case SQL_CLAUSE_ORDER_BY:
nodesWalkList(pSelect->pProjectionList, walker, pContext);
case SQL_CLAUSE_SELECT:
default:
break;
}

6
source/libs/nodes/src/nodesUtilFuncs.c
View File

@ -159,6 +159,8 @@ SNodeptr nodesMakeNode(ENodeType type) {
return makeNode(type, sizeof(SExchangeLogicNode));
case QUERY_NODE_LOGIC_PLAN_WINDOW:
return makeNode(type, sizeof(SWindowLogicNode));
case QUERY_NODE_LOGIC_PLAN_SORT:
return makeNode(type, sizeof(SSortLogicNode));
case QUERY_NODE_LOGIC_SUBPLAN:
return makeNode(type, sizeof(SLogicSubplan));
case QUERY_NODE_LOGIC_PLAN:
@ -182,7 +184,7 @@ SNodeptr nodesMakeNode(ENodeType type) {
case QUERY_NODE_PHYSICAL_PLAN_EXCHANGE:
return makeNode(type, sizeof(SExchangePhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_SORT:
return makeNode(type, sizeof(SNode));
return makeNode(type, sizeof(SSortPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
return makeNode(type, sizeof(SIntervalPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
@ -555,7 +557,7 @@ static EDealRes collectColumns(SNode* pNode, void* pContext) {
if (QUERY_NODE_COLUMN == nodeType(pNode)) {
SColumnNode* pCol = (SColumnNode*)pNode;
int32_t colId = pCol->colId;
if (0 == strcmp(pCxt->pTableAlias, pCol->tableAlias)) {
if (NULL == pCxt->pTableAlias || 0 == strcmp(pCxt->pTableAlias, pCol->tableAlias)) {
return doCollect(pCxt, colId, pNode);
}
}

26
source/libs/planner/inc/planInt.h
View File

@ -22,32 +22,6 @@ extern "C" {
#include "planner.h"
#define CHECK_ALLOC(p, res) \
do { \
if (NULL == (p)) { \
pCxt->errCode = TSDB_CODE_OUT_OF_MEMORY; \
return (res); \
} \
} while (0)
#define CHECK_CODE(exec, res) \
do { \
int32_t code = (exec); \
if (TSDB_CODE_SUCCESS != code) { \
pCxt->errCode = code; \
return (res); \
} \
} while (0)
#define CHECK_CODE_EXT(exec) \
do { \
int32_t code = (exec); \
if (TSDB_CODE_SUCCESS != code) { \
pCxt->errCode = code; \
return code; \
} \
} while (0)
#define planFatal(param, ...) qFatal("PLAN: " param, __VA_ARGS__)
#define planError(param, ...) qError("PLAN: " param, __VA_ARGS__)
#define planWarn(param, ...) qWarn("PLAN: " param, __VA_ARGS__)

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

@ -45,7 +45,9 @@ static EDealRes doRewriteExpr(SNode** pNode, void* pContext) {
}
if (nodesEqualNode(pExpr, *pNode)) {
SColumnNode* pCol = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN);
CHECK_ALLOC(pCol, DEAL_RES_ERROR);
if (NULL == pCol) {
return DEAL_RES_ERROR;
}
SExprNode* pToBeRewrittenExpr = (SExprNode*)(*pNode);
pCol->node.resType = pToBeRewrittenExpr->resType;
strcpy(pCol->node.aliasName, pToBeRewrittenExpr->aliasName);
@ -311,20 +313,22 @@ static EDealRes doCreateColumn(SNode* pNode, void* pContext) {
switch (nodeType(pNode)) {
case QUERY_NODE_COLUMN: {
SNode* pCol = nodesCloneNode(pNode);
CHECK_ALLOC(pCol, DEAL_RES_ERROR);
CHECK_CODE(nodesListAppend(pCxt->pList, pCol), DEAL_RES_ERROR);
return DEAL_RES_IGNORE_CHILD;
if (NULL == pCol) {
return DEAL_RES_ERROR;
}
return (TSDB_CODE_SUCCESS == nodesListAppend(pCxt->pList, pCol) ? DEAL_RES_IGNORE_CHILD : DEAL_RES_ERROR);
}
case QUERY_NODE_OPERATOR:
case QUERY_NODE_LOGIC_CONDITION:
case QUERY_NODE_FUNCTION: {
SExprNode* pExpr = (SExprNode*)pNode;
SColumnNode* pCol = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN);
CHECK_ALLOC(pCol, DEAL_RES_ERROR);
if (NULL == pCol) {
return DEAL_RES_ERROR;
}
pCol->node.resType = pExpr->resType;
strcpy(pCol->colName, pExpr->aliasName);
CHECK_CODE(nodesListAppend(pCxt->pList, (SNode*)pCol), DEAL_RES_ERROR);
return DEAL_RES_IGNORE_CHILD;
return (TSDB_CODE_SUCCESS == nodesListAppend(pCxt->pList, pCol) ? DEAL_RES_IGNORE_CHILD : DEAL_RES_ERROR);
}
default:
break;
@ -485,6 +489,41 @@ static int32_t createWindowLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSele
return TSDB_CODE_FAILED;
}
static int32_t createSortLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
if (NULL == pSelect->pOrderByList) {
return TSDB_CODE_SUCCESS;
}
SSortLogicNode* pSort = nodesMakeNode(QUERY_NODE_LOGIC_PLAN_SORT);
if (NULL == pSort) {
return TSDB_CODE_OUT_OF_MEMORY;
}
SNodeList* pCols = NULL;
int32_t code = nodesCollectColumns(pSelect, SQL_CLAUSE_ORDER_BY, NULL, &pCols);
if (TSDB_CODE_SUCCESS == code && NULL != pCols) {
pSort->node.pTargets = nodesCloneList(pCols);
if (NULL == pSort->node.pTargets) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
if (TSDB_CODE_SUCCESS == code) {
pSort->pSortKeys = nodesCloneList(pSelect->pOrderByList);
if (NULL == pSort->pSortKeys) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
if (TSDB_CODE_SUCCESS == code) {
*pLogicNode = (SLogicNode*)pSort;
} else {
nodesDestroyNode(pSort);
}
return code;
}
static int32_t createColumnByProjections(SLogicPlanContext* pCxt, const char* pStmtName, SNodeList* pExprs, SNodeList** pCols) {
SNodeList* pList = nodesMakeList();
if (NULL == pList) {
@ -539,6 +578,9 @@ static int32_t createSelectLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSele
if (TSDB_CODE_SUCCESS == code) {
code = createChildLogicNode(pCxt, pSelect, createAggLogicNode, &pRoot);
}
if (TSDB_CODE_SUCCESS == code) {
code = createChildLogicNode(pCxt, pSelect, createSortLogicNode, &pRoot);
}
if (TSDB_CODE_SUCCESS == code) {
code = createChildLogicNode(pCxt, pSelect, createProjectLogicNode, &pRoot);
}

42
source/libs/planner/src/planPhysiCreater.c
View File

@ -31,6 +31,10 @@ typedef struct SPhysiPlanContext {
} SPhysiPlanContext;
static int32_t getSlotKey(SNode* pNode, const char* pStmtName, char* pKey) {
if (QUERY_NODE_ORDER_BY_EXPR == nodeType(pNode)) {
return getSlotKey(((SOrderByExprNode*)pNode)->pExpr, pStmtName, pKey);
}
if (QUERY_NODE_COLUMN == nodeType(pNode)) {
SColumnNode* pCol = (SColumnNode*)pNode;
if (NULL != pStmtName) {
@ -41,6 +45,7 @@ static int32_t getSlotKey(SNode* pNode, const char* pStmtName, char* pKey) {
}
return sprintf(pKey, "%s.%s", pCol->tableAlias, pCol->colName);
}
if (NULL != pStmtName) {
return sprintf(pKey, "%s.%s", pStmtName, ((SExprNode*)pNode)->aliasName);
}
@ -815,6 +820,41 @@ static int32_t createWindowPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildr
return TSDB_CODE_FAILED;
}
static int32_t createSortPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren, SSortLogicNode* pSortLogicNode, SPhysiNode** pPhyNode) {
SSortPhysiNode* pSort = (SSortPhysiNode*)makePhysiNode(pCxt, (SLogicNode*)pSortLogicNode, QUERY_NODE_PHYSICAL_PLAN_SORT);
if (NULL == pSort) {
return TSDB_CODE_OUT_OF_MEMORY;
}
SNodeList* pPrecalcExprs = NULL;
SNodeList* pSortKeys = NULL;
int32_t code = rewritePrecalcExprs(pCxt, pSortLogicNode->pSortKeys, &pPrecalcExprs, &pSortKeys);
SDataBlockDescNode* pChildTupe = (((SPhysiNode*)nodesListGetNode(pChildren, 0))->pOutputDataBlockDesc);
// push down expression to pOutputDataBlockDesc of child node
if (TSDB_CODE_SUCCESS == code && NULL != pPrecalcExprs) {
code = setListSlotId(pCxt, pChildTupe->dataBlockId, -1, pPrecalcExprs, &pSort->pExprs);
if (TSDB_CODE_SUCCESS == code) {
code = addDataBlockSlots(pCxt, pSort->pExprs, pChildTupe);
}
}
if (TSDB_CODE_SUCCESS == code) {
code = setListSlotId(pCxt, pChildTupe->dataBlockId, -1, pSortKeys, &pSort->pSortKeys);
if (TSDB_CODE_SUCCESS == code) {
code = addDataBlockSlots(pCxt, pSort->pSortKeys, pSort->node.pOutputDataBlockDesc);
}
}
if (TSDB_CODE_SUCCESS == code) {
*pPhyNode = (SPhysiNode*)pSort;
} else {
nodesDestroyNode(pSort);
}
return code;
}
static int32_t doCreatePhysiNode(SPhysiPlanContext* pCxt, SLogicNode* pLogicNode, SSubplan* pSubplan, SNodeList* pChildren, SPhysiNode** pPhyNode) {
switch (nodeType(pLogicNode)) {
case QUERY_NODE_LOGIC_PLAN_SCAN:
@ -829,6 +869,8 @@ static int32_t doCreatePhysiNode(SPhysiPlanContext* pCxt, SLogicNode* pLogicNode
return createExchangePhysiNode(pCxt, (SExchangeLogicNode*)pLogicNode, pPhyNode);
case QUERY_NODE_LOGIC_PLAN_WINDOW:
return createWindowPhysiNode(pCxt, pChildren, (SWindowLogicNode*)pLogicNode, pPhyNode);
case QUERY_NODE_LOGIC_PLAN_SORT:
return createSortPhysiNode(pCxt, pChildren, (SSortLogicNode*)pLogicNode, pPhyNode);
default:
break;
}

4
source/libs/planner/src/planSpliter.c
View File

@ -65,7 +65,9 @@ static int32_t stsMatch(SSplitContext* pCxt, SLogicSubplan* pSubplan) {
SLogicNode* pSplitNode = stsMatchByNode(pSubplan->pNode);
if (NULL != pSplitNode) {
SStsInfo* pInfo = calloc(1, sizeof(SStsInfo));
CHECK_ALLOC(pInfo, TSDB_CODE_OUT_OF_MEMORY);
if (NULL == pInfo) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pInfo->pScan = (SScanLogicNode*)pSplitNode;
pInfo->pSubplan = pSubplan;
pCxt->pInfo = pInfo;

13
source/libs/planner/test/plannerTest.cpp
View File

@ -201,6 +201,19 @@ TEST_F(PlannerTest, sessionWindow) {
ASSERT_TRUE(run());
}
TEST_F(PlannerTest, orderBy) {
setDatabase("root", "test");
bind("SELECT * FROM t1 order by c1");
ASSERT_TRUE(run());
bind("SELECT c1 FROM t1 order by c2");
ASSERT_TRUE(run());
bind("SELECT * FROM t1 order by c1 + 10, c2");
ASSERT_TRUE(run());
}
TEST_F(PlannerTest, showTables) {
setDatabase("root", "test");