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

TD-13675 rollup, alter db, alter table syntax definition

This commit is contained in:
Xiaoyu Wang 2022-03-17 06:44:45 -04:00
parent 7eb4153ad0
commit d49adcd4f5
8 changed files with 433 additions and 225 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
include/common/tmsg.h
View File

@ -111,15 +111,16 @@ typedef enum _mgmt_table {
TSDB_MGMT_TABLE_MAX,
} EShowType;
#define TSDB_ALTER_TABLE_ADD_TAG 1
#define TSDB_ALTER_TABLE_DROP_TAG 2
#define TSDB_ALTER_TABLE_UPDATE_TAG_NAME 3
#define TSDB_ALTER_TABLE_UPDATE_TAG_VAL 4
#define TSDB_ALTER_TABLE_ADD_TAG 1
#define TSDB_ALTER_TABLE_DROP_TAG 2
#define TSDB_ALTER_TABLE_UPDATE_TAG_NAME 3
#define TSDB_ALTER_TABLE_UPDATE_TAG_VAL 4
#define TSDB_ALTER_TABLE_ADD_COLUMN 5
#define TSDB_ALTER_TABLE_DROP_COLUMN 6
#define TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES 7
#define TSDB_ALTER_TABLE_UPDATE_TAG_BYTES 8
#define TSDB_ALTER_TABLE_UPDATE_OPTIONS 9
#define TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME 10
#define TSDB_FILL_NONE 0
#define TSDB_FILL_NULL 1

7
include/libs/nodes/cmdnodes.h
View File

@ -131,7 +131,12 @@ typedef struct SAlterTableStmt {
ENodeType type;
char dbName[TSDB_DB_NAME_LEN];
char tableName[TSDB_TABLE_NAME_LEN];
// todo
int8_t alterType;
char colName[TSDB_COL_NAME_LEN];
char newColName[TSDB_COL_NAME_LEN];
STableOptions* pOptions;
SDataType dataType;
SValueNode* pVal;
} SAlterTableStmt;
typedef struct SCreateUserStmt {

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

@ -166,6 +166,7 @@ SNodeList* nodesMakeList();
int32_t nodesListAppend(SNodeList* pList, SNodeptr pNode);
int32_t nodesListStrictAppend(SNodeList* pList, SNodeptr pNode);
int32_t nodesListAppendList(SNodeList* pTarget, SNodeList* pSrc);
int32_t nodesListStrictAppendList(SNodeList* pTarget, SNodeList* pSrc);
SListCell* nodesListErase(SNodeList* pList, SListCell* pCell);
SNodeptr nodesListGetNode(SNodeList* pList, int32_t index);
void nodesDestroyList(SNodeList* pList);

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

@ -359,6 +359,17 @@ int32_t nodesListAppendList(SNodeList* pTarget, SNodeList* pSrc) {
return TSDB_CODE_SUCCESS;
}
int32_t nodesListStrictAppendList(SNodeList* pTarget, SNodeList* pSrc) {
if (NULL == pSrc) {
return TSDB_CODE_OUT_OF_MEMORY;
}
int32_t code = nodesListAppendList(pTarget, pSrc);
if (TSDB_CODE_SUCCESS != code) {
nodesDestroyList(pSrc);
}
return code;
}
SListCell* nodesListErase(SNodeList* pList, SListCell* pCell) {
if (NULL == pCell->pPrev) {
pList->pHead = pCell->pNext;
@ -573,7 +584,7 @@ typedef struct SCollectFuncsCxt {
static EDealRes collectFuncs(SNode* pNode, void* pContext) {
SCollectFuncsCxt* pCxt = (SCollectFuncsCxt*)pContext;
if (QUERY_NODE_FUNCTION == nodeType(pNode) && pCxt->classifier(((SFunctionNode*)pNode)->funcId)) {
pCxt->errCode = nodesListAppend(pCxt->pFuncs, pNode);
pCxt->errCode = nodesListStrictAppend(pCxt->pFuncs, nodesCloneNode(pNode));
return (TSDB_CODE_SUCCESS == pCxt->errCode ? DEAL_RES_IGNORE_CHILD : DEAL_RES_ERROR);
}
return DEAL_RES_CONTINUE;

14
source/libs/parser/inc/sql.y
View File

@ -181,18 +181,18 @@ cmd ::= ALTER STABLE alter_table_clause(A).
alter_table_clause(A) ::= full_table_name(B) alter_table_options(C). { A = createAlterTableOption(pCxt, B, C); }
alter_table_clause(A) ::=
full_table_name(B) ADD COLUMN column_name(C) type_name(D). { A = createAlterTableAddModifyCol(pCxt, B, TSDB_ALTER_TABLE_ADD_COLUMN, &C, D); }
alter_table_clause(A) ::= full_table_name(B) DROP COLUMN column_name(C). { A = createAlterTableDropCol(pCxt, B, TSDB_ALTER_TABLE_ADD_COLUMN, &C); }
alter_table_clause(A) ::= full_table_name(B) DROP COLUMN column_name(C). { A = createAlterTableDropCol(pCxt, B, TSDB_ALTER_TABLE_DROP_COLUMN, &C); }
alter_table_clause(A) ::=
full_table_name(B) MODIFY COLUMN column_name(C) type_name(D). { A = createAlterTableAddModifyCol(pCxt, B, TSDB_ALTER_TABLE_ADD_COLUMN, &C, D); }
full_table_name(B) MODIFY COLUMN column_name(C) type_name(D). { A = createAlterTableAddModifyCol(pCxt, B, TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES, &C, D); }
alter_table_clause(A) ::=
full_table_name(B) RENAME COLUMN column_name(C) column_name(D). { A = createAlterTableRenameCol(pCxt, B, TSDB_ALTER_TABLE_ADD_COLUMN, &C, &D); }
full_table_name(B) RENAME COLUMN column_name(C) column_name(D). { A = createAlterTableRenameCol(pCxt, B, TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME, &C, &D); }
alter_table_clause(A) ::=
full_table_name(B) ADD TAG column_name(C) type_name(D). { A = createAlterTableAddModifyCol(pCxt, B, TSDB_ALTER_TABLE_ADD_COLUMN, &C, D); }
alter_table_clause(A) ::= full_table_name(B) DROP TAG column_name(C). { A = createAlterTableDropCol(pCxt, B, TSDB_ALTER_TABLE_ADD_COLUMN, &C); }
full_table_name(B) ADD TAG column_name(C) type_name(D). { A = createAlterTableAddModifyCol(pCxt, B, TSDB_ALTER_TABLE_ADD_TAG, &C, D); }
alter_table_clause(A) ::= full_table_name(B) DROP TAG column_name(C). { A = createAlterTableDropCol(pCxt, B, TSDB_ALTER_TABLE_DROP_TAG, &C); }
alter_table_clause(A) ::=
full_table_name(B) MODIFY TAG column_name(C) type_name(D). { A = createAlterTableAddModifyCol(pCxt, B, TSDB_ALTER_TABLE_ADD_COLUMN, &C, D); }
full_table_name(B) MODIFY TAG column_name(C) type_name(D). { A = createAlterTableAddModifyCol(pCxt, B, TSDB_ALTER_TABLE_UPDATE_TAG_BYTES, &C, D); }
alter_table_clause(A) ::=
full_table_name(B) RENAME TAG column_name(C) column_name(D). { A = createAlterTableRenameCol(pCxt, B, TSDB_ALTER_TABLE_ADD_COLUMN, &C, &D); }
full_table_name(B) RENAME TAG column_name(C) column_name(D). { A = createAlterTableRenameCol(pCxt, B, TSDB_ALTER_TABLE_UPDATE_TAG_NAME, &C, &D); }
alter_table_clause(A) ::=
full_table_name(B) SET TAG column_name(C) NK_EQ literal(D). { A = createAlterTableSetTag(pCxt, B, &C, D); }

33
source/libs/parser/src/parAstCreater.c
View File

@ -961,23 +961,46 @@ SNode* createDropSuperTableStmt(SAstCreateContext* pCxt, bool ignoreNotExists, S
}
SNode* createAlterTableOption(SAstCreateContext* pCxt, SNode* pRealTable, SNode* pOptions) {
SAlterTableStmt* pStmt = nodesMakeNode(QUERY_NODE_ALTER_TABLE_STMT);
CHECK_OUT_OF_MEM(pStmt);
pStmt->alterType = TSDB_ALTER_TABLE_UPDATE_OPTIONS;
pStmt->pOptions = (STableOptions*)pOptions;
return (SNode*)pStmt;
}
SNode* createAlterTableAddModifyCol(SAstCreateContext* pCxt, SNode* pRealTable, int8_t alterType, const SToken* pColName, SDataType dataType) {
SAlterTableStmt* pStmt = nodesMakeNode(QUERY_NODE_ALTER_TABLE_STMT);
CHECK_OUT_OF_MEM(pStmt);
pStmt->alterType = alterType;
strncpy(pStmt->colName, pColName->z, pColName->n);
pStmt->dataType = dataType;
return (SNode*)pStmt;
}
SNode* createAlterTableDropCol(SAstCreateContext* pCxt, SNode* pRealTable, int8_t alterType, const SToken* pColName) {
SAlterTableStmt* pStmt = nodesMakeNode(QUERY_NODE_ALTER_TABLE_STMT);
CHECK_OUT_OF_MEM(pStmt);
pStmt->alterType = alterType;
strncpy(pStmt->colName, pColName->z, pColName->n);
return (SNode*)pStmt;
}
SNode* createAlterTableRenameCol(SAstCreateContext* pCxt, SNode* pRealTable, int8_t alterType, const SToken* pOldColName, const SToken* pNewColName) {
SAlterTableStmt* pStmt = nodesMakeNode(QUERY_NODE_ALTER_TABLE_STMT);
CHECK_OUT_OF_MEM(pStmt);
pStmt->alterType = alterType;
strncpy(pStmt->colName, pOldColName->z, pOldColName->n);
strncpy(pStmt->newColName, pNewColName->z, pNewColName->n);
return (SNode*)pStmt;
}
SNode* createAlterTableSetTag(SAstCreateContext* pCxt, SNode* pRealTable, const SToken* pTagName, SNode* pVal) {
SAlterTableStmt* pStmt = nodesMakeNode(QUERY_NODE_ALTER_TABLE_STMT);
CHECK_OUT_OF_MEM(pStmt);
pStmt->alterType = TSDB_ALTER_TABLE_UPDATE_TAG_VAL;
strncpy(pStmt->colName, pTagName->z, pTagName->n);
pStmt->pVal = (SValueNode*)pVal;
return (SNode*)pStmt;
}
SNode* createUseDatabaseStmt(SAstCreateContext* pCxt, const SToken* pDbName) {

58
source/libs/parser/src/parTranslater.c
View File

@ -962,8 +962,56 @@ static int32_t translateDropSuperTable(STranslateContext* pCxt, SDropSuperTableS
return doTranslateDropSuperTable(pCxt, &tableName, pStmt->ignoreNotExists);
}
static int32_t setAlterTableField(SAlterTableStmt* pStmt, SMAltertbReq* pAlterReq) {
pAlterReq->pFields = taosArrayInit(2, sizeof(TAOS_FIELD));
if (NULL == pAlterReq->pFields) {
return TSDB_CODE_OUT_OF_MEMORY;
}
switch (pStmt->alterType) {
case TSDB_ALTER_TABLE_ADD_TAG:
case TSDB_ALTER_TABLE_DROP_TAG:
case TSDB_ALTER_TABLE_ADD_COLUMN:
case TSDB_ALTER_TABLE_DROP_COLUMN:
case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES:
case TSDB_ALTER_TABLE_UPDATE_TAG_BYTES: {
TAOS_FIELD field = { .type = pStmt->dataType.type, .bytes = pStmt->dataType.bytes };
strcpy(field.name, pStmt->colName);
taosArrayPush(pAlterReq->pFields, &field);
break;
}
case TSDB_ALTER_TABLE_UPDATE_TAG_NAME:
case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME: {
TAOS_FIELD oldField = {0};
strcpy(oldField.name, pStmt->colName);
taosArrayPush(pAlterReq->pFields, &oldField);
TAOS_FIELD newField = {0};
strcpy(oldField.name, pStmt->newColName);
taosArrayPush(pAlterReq->pFields, &newField);
break;
}
default:
break;
}
return TSDB_CODE_SUCCESS;
}
static int32_t translateAlterTable(STranslateContext* pCxt, SAlterTableStmt* pStmt) {
SMAltertbReq alterReq = {0};
SName tableName = { .type = TSDB_TABLE_NAME_T, .acctId = pCxt->pParseCxt->acctId };
strcpy(tableName.dbname, pStmt->dbName);
strcpy(tableName.tname, pStmt->tableName);
tNameExtractFullName(&tableName, alterReq.name);
alterReq.alterType = pStmt->alterType;
alterReq.numOfFields = 1;
if (TSDB_ALTER_TABLE_UPDATE_OPTIONS == pStmt->alterType) {
// todo
} else {
if (TSDB_CODE_SUCCESS != setAlterTableField(pStmt, &alterReq)) {
return TSDB_CODE_OUT_OF_MEMORY;
}
}
pCxt->pCmdMsg = malloc(sizeof(SCmdMsgInfo));
if (NULL == pCxt->pCmdMsg) {
@ -1910,6 +1958,11 @@ static int32_t rewriteCreateMultiTable(STranslateContext* pCxt, SQuery* pQuery)
return rewriteToVnodeModifOpStmt(pQuery, pBufArray);
}
static int32_t rewriteAlterTable(STranslateContext* pCxt, SQuery* pQuery) {
// todo
return TSDB_CODE_SUCCESS;
}
static int32_t rewriteQuery(STranslateContext* pCxt, SQuery* pQuery) {
int32_t code = TSDB_CODE_SUCCESS;
switch (nodeType(pQuery->pRoot)) {
@ -1921,6 +1974,11 @@ static int32_t rewriteQuery(STranslateContext* pCxt, SQuery* pQuery) {
case QUERY_NODE_CREATE_MULTI_TABLE_STMT:
code = rewriteCreateMultiTable(pCxt, pQuery);
break;
case QUERY_NODE_ALTER_TABLE_STMT:
if (TSDB_ALTER_TABLE_UPDATE_TAG_VAL == ((SAlterTableStmt*)pQuery->pRoot)->alterType) {
code = rewriteAlterTable(pCxt, pQuery);
}
break;
default:
break;
}

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

@ -18,13 +18,13 @@
#include "functionMgt.h"
typedef struct SLogicPlanContext {
SPlanContext* pPlanCxt;
int32_t errCode;
int32_t planNodeId;
int32_t acctId;
} SLogicPlanContext;
static SLogicNode* createQueryLogicNode(SLogicPlanContext* pCxt, SNode* pStmt);
static SLogicNode* createLogicNodeByTable(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SNode* pTable);
static int32_t doCreateLogicNodeByTable(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SNode* pTable, SLogicNode** pLogicNode);
static int32_t createQueryLogicNode(SLogicPlanContext* pCxt, SNode* pStmt, SLogicNode** pLogicNode);
typedef struct SRewriteExprCxt {
int32_t errCode;
@ -94,123 +94,75 @@ static int32_t rewriteExpr(int32_t planNodeId, int32_t rewriteId, SNodeList* pEx
return cxt.errCode;
}
static SLogicNode* pushLogicNode(SLogicPlanContext* pCxt, SLogicNode* pRoot, SLogicNode* pNode) {
if (TSDB_CODE_SUCCESS != pCxt->errCode) {
goto error;
}
if (NULL == pRoot) {
return pNode;
}
if (NULL == pNode) {
return pRoot;
}
if (NULL == pNode->pChildren) {
pNode->pChildren = nodesMakeList();
if (NULL == pNode->pChildren) {
goto error;
static int32_t pushLogicNode(SLogicPlanContext* pCxt, SLogicNode** pOldRoot, SLogicNode* pNewRoot) {
if (NULL == pNewRoot->pChildren) {
pNewRoot->pChildren = nodesMakeList();
if (NULL == pNewRoot->pChildren) {
return TSDB_CODE_OUT_OF_MEMORY;
}
}
if (TSDB_CODE_SUCCESS != nodesListAppend(pNode->pChildren, (SNode*)pRoot)) {
goto error;
if (TSDB_CODE_SUCCESS != nodesListAppend(pNewRoot->pChildren, (SNode*)*pOldRoot)) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pRoot->pParent = pNode;
return pNode;
error:
nodesDestroyNode((SNode*)pNode);
return pRoot;
(*pOldRoot)->pParent = pNewRoot;
*pOldRoot = pNewRoot;
return TSDB_CODE_SUCCESS;
}
static SLogicNode* createScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SRealTableNode* pRealTable) {
SScanLogicNode* pScan = (SScanLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_SCAN);
CHECK_ALLOC(pScan, NULL);
pScan->node.id = pCxt->planNodeId++;
TSWAP(pScan->pMeta, pRealTable->pMeta, STableMeta*);
TSWAP(pScan->pVgroupList, pRealTable->pVgroupList, SVgroupsInfo*);
// set columns to scan
SNodeList* pCols = NULL;
CHECK_CODE(nodesCollectColumns(pSelect, SQL_CLAUSE_FROM, pRealTable->table.tableAlias, &pCols), (SLogicNode*)pScan);
if (NULL != pCols) {
pScan->pScanCols = nodesCloneList(pCols);
CHECK_ALLOC(pScan->pScanCols, (SLogicNode*)pScan);
static SColumnNode* createColumnByExpr(SExprNode* pExpr) {
SColumnNode* pCol = nodesMakeNode(QUERY_NODE_COLUMN);
if (NULL == pCol) {
return NULL;
}
// set output
if (NULL != pCols) {
pScan->node.pTargets = nodesCloneList(pCols);
CHECK_ALLOC(pScan->node.pTargets, (SLogicNode*)pScan);
}
pScan->scanType = SCAN_TYPE_TABLE;
pScan->scanFlag = MAIN_SCAN;
pScan->scanRange = TSWINDOW_INITIALIZER;
pScan->tableName.type = TSDB_TABLE_NAME_T;
pScan->tableName.acctId = pCxt->acctId;
strcpy(pScan->tableName.dbname, pRealTable->table.dbName);
strcpy(pScan->tableName.tname, pRealTable->table.tableName);
return (SLogicNode*)pScan;
pCol->node.resType = pExpr->resType;
strcpy(pCol->colName, pExpr->aliasName);
return pCol;
}
static SLogicNode* createSubqueryLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, STempTableNode* pTable) {
SLogicNode* pRoot = createQueryLogicNode(pCxt, pTable->pSubquery);
CHECK_ALLOC(pRoot, NULL);
static int32_t createColumnByProjections(SLogicPlanContext* pCxt, SNodeList* pExprs, SNodeList** pCols) {
SNodeList* pList = nodesMakeList();
if (NULL == pList) {
return TSDB_CODE_OUT_OF_MEMORY;
}
SNode* pNode;
FOREACH(pNode, pRoot->pTargets) {
strcpy(((SColumnNode*)pNode)->tableAlias, pTable->table.tableAlias);
FOREACH(pNode, pExprs) {
if (TSDB_CODE_SUCCESS != nodesListAppend(pList, createColumnByExpr((SExprNode*)pNode))) {
nodesDestroyList(pList);
return TSDB_CODE_OUT_OF_MEMORY;
}
}
return pRoot;
*pCols = pList;
return TSDB_CODE_SUCCESS;
}
static SLogicNode* createJoinLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SJoinTableNode* pJoinTable) {
SJoinLogicNode* pJoin = (SJoinLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_JOIN);
CHECK_ALLOC(pJoin, NULL);
pJoin->node.id = pCxt->planNodeId++;
pJoin->joinType = pJoinTable->joinType;
// set left and right node
pJoin->node.pChildren = nodesMakeList();
CHECK_ALLOC(pJoin->node.pChildren, (SLogicNode*)pJoin);
SLogicNode* pLeft = createLogicNodeByTable(pCxt, pSelect, pJoinTable->pLeft);
CHECK_ALLOC(pLeft, (SLogicNode*)pJoin);
CHECK_CODE(nodesListAppend(pJoin->node.pChildren, (SNode*)pLeft), (SLogicNode*)pJoin);
SLogicNode* pRight = createLogicNodeByTable(pCxt, pSelect, pJoinTable->pRight);
CHECK_ALLOC(pRight, (SLogicNode*)pJoin);
CHECK_CODE(nodesListAppend(pJoin->node.pChildren, (SNode*)pRight), (SLogicNode*)pJoin);
// set on conditions
if (NULL != pJoinTable->pOnCond) {
pJoin->pOnConditions = nodesCloneNode(pJoinTable->pOnCond);
CHECK_ALLOC(pJoin->pOnConditions, (SLogicNode*)pJoin);
static int32_t createProjectLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
SProjectLogicNode* pProject = (SProjectLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_PROJECT);
if (NULL == pProject) {
return TSDB_CODE_OUT_OF_MEMORY;
}
// set the output
pJoin->node.pTargets = nodesCloneList(pLeft->pTargets);
CHECK_ALLOC(pJoin->node.pTargets, (SLogicNode*)pJoin);
SNodeList* pTargets = nodesCloneList(pRight->pTargets);
CHECK_ALLOC(pTargets, (SLogicNode*)pJoin);
nodesListAppendList(pJoin->node.pTargets, pTargets);
int32_t code = TSDB_CODE_SUCCESS;
return (SLogicNode*)pJoin;
}
static SLogicNode* createLogicNodeByTable(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SNode* pTable) {
switch (nodeType(pTable)) {
case QUERY_NODE_REAL_TABLE:
return createScanLogicNode(pCxt, pSelect, (SRealTableNode*)pTable);
case QUERY_NODE_TEMP_TABLE:
return createSubqueryLogicNode(pCxt, pSelect, (STempTableNode*)pTable);
case QUERY_NODE_JOIN_TABLE:
return createJoinLogicNode(pCxt, pSelect, (SJoinTableNode*)pTable);
default:
break;
pProject->pProjections = nodesCloneList(pSelect->pProjectionList);
if (NULL == pProject->pProjections) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
return NULL;
if (TSDB_CODE_SUCCESS == code) {
code = createColumnByProjections(pCxt,pSelect->pProjectionList, &pProject->node.pTargets);
}
if (TSDB_CODE_SUCCESS == code) {
*pLogicNode = (SLogicNode*)pProject;
} else {
nodesDestroyNode(pProject);
}
return code;
}
typedef struct SCreateColumnCxt {
@ -245,196 +197,353 @@ static EDealRes doCreateColumn(SNode* pNode, void* pContext) {
return DEAL_RES_CONTINUE;
}
static SNodeList* createColumnByRewriteExps(SLogicPlanContext* pCxt, SNodeList* pExprs) {
SCreateColumnCxt cxt = { .errCode = TSDB_CODE_SUCCESS, .pList = nodesMakeList() };
CHECK_ALLOC(cxt.pList, NULL);
static int32_t createColumnByRewriteExps(SLogicPlanContext* pCxt, SNodeList* pExprs, SNodeList** pList) {
SCreateColumnCxt cxt = { .errCode = TSDB_CODE_SUCCESS, .pList = (NULL == *pList ? nodesMakeList() : *pList) };
if (NULL == cxt.pList) {
return TSDB_CODE_OUT_OF_MEMORY;
}
nodesWalkList(pExprs, doCreateColumn, &cxt);
if (TSDB_CODE_SUCCESS != cxt.errCode) {
nodesDestroyList(cxt.pList);
return NULL;
return cxt.errCode;
}
return cxt.pList;
if (NULL == *pList) {
*pList = cxt.pList;
}
return cxt.errCode;
}
static SLogicNode* createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect) {
static int32_t createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
SNodeList* pAggFuncs = NULL;
CHECK_CODE(nodesCollectFuncs(pSelect, fmIsAggFunc, &pAggFuncs), NULL);
int32_t code = nodesCollectFuncs(pSelect, fmIsAggFunc, &pAggFuncs);
if (TSDB_CODE_SUCCESS != code) {
return code;
}
if (NULL == pAggFuncs && NULL == pSelect->pGroupByList) {
return NULL;
return TSDB_CODE_SUCCESS;
}
SAggLogicNode* pAgg = (SAggLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_AGG);
CHECK_ALLOC(pAgg, NULL);
pAgg->node.id = pCxt->planNodeId++;
if (NULL == pAgg) {
return TSDB_CODE_OUT_OF_MEMORY;
}
// set grouyp keys, agg funcs and having conditions
if (NULL != pSelect->pGroupByList) {
pAgg->pGroupKeys = nodesCloneList(pSelect->pGroupByList);
CHECK_ALLOC(pAgg->pGroupKeys, (SLogicNode*)pAgg);
if (NULL == pAgg->pGroupKeys) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
if (NULL != pAggFuncs) {
if (TSDB_CODE_SUCCESS == code && NULL != pAggFuncs) {
pAgg->pAggFuncs = nodesCloneList(pAggFuncs);
CHECK_ALLOC(pAgg->pAggFuncs, (SLogicNode*)pAgg);
if (NULL == pAgg->pAggFuncs) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
// rewrite the expression in subsequent clauses
CHECK_CODE(rewriteExpr(pAgg->node.id, 1, pAgg->pGroupKeys, pSelect, SQL_CLAUSE_GROUP_BY), (SLogicNode*)pAgg);
CHECK_CODE(rewriteExpr(pAgg->node.id, 1 + LIST_LENGTH(pAgg->pGroupKeys), pAgg->pAggFuncs, pSelect, SQL_CLAUSE_GROUP_BY), (SLogicNode*)pAgg);
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExpr(pAgg->node.id, 1, pAgg->pGroupKeys, pSelect, SQL_CLAUSE_GROUP_BY);
}
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExpr(pAgg->node.id, 1 + LIST_LENGTH(pAgg->pGroupKeys), pAgg->pAggFuncs, pSelect, SQL_CLAUSE_GROUP_BY);
}
if (NULL != pSelect->pHaving) {
if (TSDB_CODE_SUCCESS == code && NULL != pSelect->pHaving) {
pAgg->node.pConditions = nodesCloneNode(pSelect->pHaving);
CHECK_ALLOC(pAgg->node.pConditions, (SLogicNode*)pAgg);
if (NULL == pAgg->node.pConditions) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
// set the output
pAgg->node.pTargets = nodesMakeList();
CHECK_ALLOC(pAgg->node.pTargets, (SLogicNode*)pAgg);
if (NULL != pAgg->pGroupKeys) {
SNodeList* pTargets = createColumnByRewriteExps(pCxt, pAgg->pGroupKeys);
CHECK_ALLOC(pAgg->node.pTargets, (SLogicNode*)pAgg);
nodesListAppendList(pAgg->node.pTargets, pTargets);
if (TSDB_CODE_SUCCESS == code && NULL != pAgg->pGroupKeys) {
code = createColumnByRewriteExps(pCxt, pAgg->pGroupKeys, &pAgg->node.pTargets);
}
if (NULL != pAgg->pAggFuncs) {
SNodeList* pTargets = createColumnByRewriteExps(pCxt, pAgg->pAggFuncs);
CHECK_ALLOC(pTargets, (SLogicNode*)pAgg);
nodesListAppendList(pAgg->node.pTargets, pTargets);
if (TSDB_CODE_SUCCESS == code && NULL != pAgg->pAggFuncs) {
code = createColumnByRewriteExps(pCxt, pAgg->pAggFuncs, &pAgg->node.pTargets);
}
return (SLogicNode*)pAgg;
if (TSDB_CODE_SUCCESS == code) {
*pLogicNode = (SLogicNode*)pAgg;
} else {
nodesDestroyNode(pAgg);
}
return code;
}
static SLogicNode* createWindowLogicNodeByInterval(SLogicPlanContext* pCxt, SIntervalWindowNode* pInterval, SSelectStmt* pSelect) {
static int32_t createWindowLogicNodeByInterval(SLogicPlanContext* pCxt, SIntervalWindowNode* pInterval, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
SWindowLogicNode* pWindow = nodesMakeNode(QUERY_NODE_LOGIC_PLAN_WINDOW);
CHECK_ALLOC(pWindow, NULL);
pWindow->node.id = pCxt->planNodeId++;
if (NULL == pWindow) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pWindow->winType = WINDOW_TYPE_INTERVAL;
SValueNode* pIntervalNode = (SValueNode*)((SRawExprNode*)(pInterval->pInterval))->pNode;
pWindow->winType = WINDOW_TYPE_INTERVAL;
pWindow->interval = pIntervalNode->datum.i;
pWindow->offset = (NULL != pInterval->pOffset ? ((SValueNode*)pInterval->pOffset)->datum.i : 0);
pWindow->sliding = (NULL != pInterval->pSliding ? ((SValueNode*)pInterval->pSliding)->datum.i : pWindow->interval);
int32_t code = TSDB_CODE_SUCCESS;
if (NULL != pInterval->pFill) {
pWindow->pFill = nodesCloneNode(pInterval->pFill);
CHECK_ALLOC(pWindow->pFill, (SLogicNode*)pWindow);
if (NULL == pWindow->pFill) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
SNodeList* pFuncs = NULL;
CHECK_CODE(nodesCollectFuncs(pSelect, fmIsAggFunc, &pFuncs), NULL);
if (NULL != pFuncs) {
pWindow->pFuncs = nodesCloneList(pFuncs);
CHECK_ALLOC(pWindow->pFuncs, (SLogicNode*)pWindow);
if (TSDB_CODE_SUCCESS == code) {
code = nodesCollectFuncs(pSelect, fmIsAggFunc, &pWindow->pFuncs);
}
CHECK_CODE(rewriteExpr(pWindow->node.id, 1, pWindow->pFuncs, pSelect, SQL_CLAUSE_WINDOW), (SLogicNode*)pWindow);
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExpr(pWindow->node.id, 1, pWindow->pFuncs, pSelect, SQL_CLAUSE_WINDOW);
}
pWindow->node.pTargets = createColumnByRewriteExps(pCxt, pWindow->pFuncs);
CHECK_ALLOC(pWindow->node.pTargets, (SLogicNode*)pWindow);
if (TSDB_CODE_SUCCESS == code) {
code = createColumnByRewriteExps(pCxt, pWindow->pFuncs, &pWindow->node.pTargets);
}
return (SLogicNode*)pWindow;
if (TSDB_CODE_SUCCESS == code) {
*pLogicNode = (SLogicNode*)pWindow;
} else {
nodesDestroyNode(pWindow);
}
return code;
}
static SLogicNode* createWindowLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect) {
static int32_t createWindowLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
if (NULL == pSelect->pWindow) {
return NULL;
return TSDB_CODE_SUCCESS;
}
switch (nodeType(pSelect->pWindow)) {
case QUERY_NODE_INTERVAL_WINDOW:
return createWindowLogicNodeByInterval(pCxt, (SIntervalWindowNode*)pSelect->pWindow, pSelect);
return createWindowLogicNodeByInterval(pCxt, (SIntervalWindowNode*)pSelect->pWindow, pSelect, pLogicNode);
default:
break;
}
return NULL;
return TSDB_CODE_FAILED;
}
static SNodeList* createColumnByProjections(SLogicPlanContext* pCxt, SNodeList* pExprs) {
SNodeList* pList = nodesMakeList();
CHECK_ALLOC(pList, NULL);
SNode* pNode;
FOREACH(pNode, pExprs) {
SExprNode* pExpr = (SExprNode*)pNode;
SColumnNode* pCol = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN);
if (NULL == pCol) {
goto error;
}
pCol->node.resType = pExpr->resType;
strcpy(pCol->colName, pExpr->aliasName);
if (TSDB_CODE_SUCCESS != nodesListAppend(pList, (SNode*)pCol)) {
goto error;
static int32_t createJoinLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SJoinTableNode* pJoinTable, SLogicNode** pLogicNode) {
SJoinLogicNode* pJoin = (SJoinLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_JOIN);
if (NULL == pJoin) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pJoin->joinType = pJoinTable->joinType;
int32_t code = TSDB_CODE_SUCCESS;
// set left and right node
pJoin->node.pChildren = nodesMakeList();
if (NULL == pJoin->node.pChildren) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
SLogicNode* pLeft = NULL;
if (TSDB_CODE_SUCCESS == code) {
code = doCreateLogicNodeByTable(pCxt, pSelect, pJoinTable->pLeft, &pLeft);
if (TSDB_CODE_SUCCESS == code) {
code = nodesListStrictAppend(pJoin->node.pChildren, (SNode*)pLeft);
}
}
return pList;
error:
nodesDestroyList(pList);
return NULL;
SLogicNode* pRight = NULL;
if (TSDB_CODE_SUCCESS == code) {
code = doCreateLogicNodeByTable(pCxt, pSelect, pJoinTable->pRight, &pRight);
if (TSDB_CODE_SUCCESS == code) {
code = nodesListStrictAppend(pJoin->node.pChildren, (SNode*)pRight);
}
}
// set on conditions
if (TSDB_CODE_SUCCESS == code && NULL != pJoinTable->pOnCond) {
pJoin->pOnConditions = nodesCloneNode(pJoinTable->pOnCond);
if (NULL == pJoin->pOnConditions) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
// set the output
if (TSDB_CODE_SUCCESS == code) {
pJoin->node.pTargets = nodesCloneList(pLeft->pTargets);
if (NULL == pJoin->pOnConditions) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
if (TSDB_CODE_SUCCESS == code) {
code = nodesListStrictAppendList(pJoin->node.pTargets, nodesCloneList(pRight->pTargets));
}
}
if (TSDB_CODE_SUCCESS == code) {
*pLogicNode = (SLogicNode*)pJoin;
} else {
nodesDestroyNode(pJoin);
}
return code;
}
static SLogicNode* createProjectLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect) {
SProjectLogicNode* pProject = (SProjectLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_PROJECT);
CHECK_ALLOC(pProject, NULL);
pProject->node.id = pCxt->planNodeId++;
pProject->pProjections = nodesCloneList(pSelect->pProjectionList);
pProject->node.pTargets = createColumnByProjections(pCxt,pSelect->pProjectionList);
CHECK_ALLOC(pProject->node.pTargets, (SLogicNode*)pProject);
return (SLogicNode*)pProject;
static int32_t createSubqueryLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, STempTableNode* pTable, SLogicNode** pLogicNode) {
int32_t code = createQueryLogicNode(pCxt, pTable->pSubquery, pLogicNode);
if (TSDB_CODE_SUCCESS == code) {
SNode* pNode;
FOREACH(pNode, (*pLogicNode)->pTargets) {
strcpy(((SColumnNode*)pNode)->tableAlias, pTable->table.tableAlias);
}
}
return code;
}
static SLogicNode* createSelectLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect) {
SLogicNode* pRoot = createLogicNodeByTable(pCxt, pSelect, pSelect->pFromTable);
if (TSDB_CODE_SUCCESS == pCxt->errCode && NULL != pSelect->pWhere) {
pRoot->pConditions = nodesCloneNode(pSelect->pWhere);
CHECK_ALLOC(pRoot->pConditions, pRoot);
static int32_t createScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SRealTableNode* pRealTable, SLogicNode** pLogicNode) {
SScanLogicNode* pScan = (SScanLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_SCAN);
if (NULL == pScan) {
return TSDB_CODE_OUT_OF_MEMORY;
}
if (TSDB_CODE_SUCCESS == pCxt->errCode) {
pRoot = pushLogicNode(pCxt, pRoot, createWindowLogicNode(pCxt, pSelect));
TSWAP(pScan->pMeta, pRealTable->pMeta, STableMeta*);
TSWAP(pScan->pVgroupList, pRealTable->pVgroupList, SVgroupsInfo*);
pScan->scanType = SCAN_TYPE_TABLE;
pScan->scanFlag = MAIN_SCAN;
pScan->scanRange = TSWINDOW_INITIALIZER;
pScan->tableName.type = TSDB_TABLE_NAME_T;
pScan->tableName.acctId = pCxt->pPlanCxt->acctId;
strcpy(pScan->tableName.dbname, pRealTable->table.dbName);
strcpy(pScan->tableName.tname, pRealTable->table.tableName);
// set columns to scan
SNodeList* pCols = NULL;
int32_t code = nodesCollectColumns(pSelect, SQL_CLAUSE_FROM, pRealTable->table.tableAlias, &pCols);
if (TSDB_CODE_SUCCESS == code && NULL != pCols) {
pScan->pScanCols = nodesCloneList(pCols);
if (NULL == pScan) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
if (TSDB_CODE_SUCCESS == pCxt->errCode) {
pRoot = pushLogicNode(pCxt, pRoot, createAggLogicNode(pCxt, pSelect));
// set output
if (TSDB_CODE_SUCCESS == code && NULL != pCols) {
pScan->node.pTargets = nodesCloneList(pCols);
if (NULL == pScan) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
if (TSDB_CODE_SUCCESS == pCxt->errCode) {
pRoot = pushLogicNode(pCxt, pRoot, createProjectLogicNode(pCxt, pSelect));
if (TSDB_CODE_SUCCESS == code) {
*pLogicNode = (SLogicNode*)pScan;
} else {
nodesDestroyNode(pScan);
}
return pRoot;
return code;
}
static int32_t doCreateLogicNodeByTable(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SNode* pTable, SLogicNode** pLogicNode) {
switch (nodeType(pTable)) {
case QUERY_NODE_REAL_TABLE:
return createScanLogicNode(pCxt, pSelect, (SRealTableNode*)pTable, pLogicNode);
case QUERY_NODE_TEMP_TABLE:
return createSubqueryLogicNode(pCxt, pSelect, (STempTableNode*)pTable, pLogicNode);
case QUERY_NODE_JOIN_TABLE:
return createJoinLogicNode(pCxt, pSelect, (SJoinTableNode*)pTable, pLogicNode);
default:
break;
}
return TSDB_CODE_FAILED;
}
static int32_t createLogicNodeByTable(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SNode* pTable, SLogicNode** pLogicNode) {
SLogicNode* pNode = NULL;
int32_t code = doCreateLogicNodeByTable(pCxt, pSelect, pTable, &pNode);
if (TSDB_CODE_SUCCESS == code) {
pNode->pConditions = nodesCloneNode(pSelect->pWhere);
if (NULL != pSelect->pWhere && NULL == pNode->pConditions) {
nodesDestroyNode(pNode);
return TSDB_CODE_OUT_OF_MEMORY;
}
*pLogicNode = pNode;
}
return code;
}
typedef int32_t (*FCreateLogicNode)(SLogicPlanContext*, SSelectStmt*, SLogicNode**);
static int32_t createChildLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, FCreateLogicNode func, SLogicNode** pRoot) {
SLogicNode* pNode = NULL;
int32_t code = func(pCxt, pSelect, &pNode);
if (TSDB_CODE_SUCCESS == code) {
code = pushLogicNode(pCxt, pRoot, pNode);
}
if (TSDB_CODE_SUCCESS != code) {
nodesDestroyNode(pNode);
}
return code;
}
static int32_t createSelectLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
SLogicNode* pRoot = NULL;
int32_t code = createLogicNodeByTable(pCxt, pSelect, pSelect->pFromTable, &pRoot);
if (TSDB_CODE_SUCCESS == code) {
code = createChildLogicNode(pCxt, pSelect, createWindowLogicNode, &pRoot);
}
if (TSDB_CODE_SUCCESS == code) {
code = createChildLogicNode(pCxt, pSelect, createAggLogicNode, &pRoot);
}
if (TSDB_CODE_SUCCESS == code) {
code = createChildLogicNode(pCxt, pSelect, createProjectLogicNode, &pRoot);
}
if (TSDB_CODE_SUCCESS == code) {
*pLogicNode = pRoot;
} else {
nodesDestroyNode(pRoot);
}
return code;
}
static int32_t getMsgType(ENodeType sqlType) {
return (QUERY_NODE_CREATE_TABLE_STMT == sqlType || QUERY_NODE_CREATE_MULTI_TABLE_STMT == sqlType) ? TDMT_VND_CREATE_TABLE : TDMT_VND_SUBMIT;
}
static SLogicNode* createVnodeModifLogicNode(SLogicPlanContext* pCxt, SVnodeModifOpStmt* pStmt) {
SVnodeModifLogicNode* pModif = (SVnodeModifLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_VNODE_MODIF);
CHECK_ALLOC(pModif, NULL);
static int32_t createVnodeModifLogicNode(SLogicPlanContext* pCxt, SVnodeModifOpStmt* pStmt, SLogicNode** pLogicNode) {
SVnodeModifLogicNode* pModif = nodesMakeNode(QUERY_NODE_LOGIC_PLAN_VNODE_MODIF);
if (NULL == pModif) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pModif->pDataBlocks = pStmt->pDataBlocks;
pModif->msgType = getMsgType(pStmt->sqlNodeType);
return (SLogicNode*)pModif;
*pLogicNode = (SLogicNode*)pModif;
return TSDB_CODE_SUCCESS;
}
static SLogicNode* createQueryLogicNode(SLogicPlanContext* pCxt, SNode* pStmt) {
static int32_t createQueryLogicNode(SLogicPlanContext* pCxt, SNode* pStmt, SLogicNode** pLogicNode) {
switch (nodeType(pStmt)) {
case QUERY_NODE_SELECT_STMT:
return createSelectLogicNode(pCxt, (SSelectStmt*)pStmt);
return createSelectLogicNode(pCxt, (SSelectStmt*)pStmt, pLogicNode);
case QUERY_NODE_VNODE_MODIF_STMT:
return createVnodeModifLogicNode(pCxt, (SVnodeModifOpStmt*)pStmt);
return createVnodeModifLogicNode(pCxt, (SVnodeModifOpStmt*)pStmt, pLogicNode);
default:
break;
}
return NULL; // to avoid compiler error
return TSDB_CODE_FAILED;
}
int32_t createLogicPlan(SPlanContext* pCxt, SLogicNode** pLogicNode) {
SLogicPlanContext cxt = { .errCode = TSDB_CODE_SUCCESS, .planNodeId = 1, .acctId = pCxt->acctId };
SLogicNode* pRoot = createQueryLogicNode(&cxt, pCxt->pAstRoot);
if (TSDB_CODE_SUCCESS != cxt.errCode) {
nodesDestroyNode((SNode*)pRoot);
return cxt.errCode;
SLogicPlanContext cxt = { .pPlanCxt = pCxt, .errCode = TSDB_CODE_SUCCESS, .planNodeId = 1 };
int32_t code = createQueryLogicNode(&cxt, pCxt->pAstRoot, pLogicNode);
if (TSDB_CODE_SUCCESS != code) {
return code;
}
*pLogicNode = pRoot;
return TSDB_CODE_SUCCESS;
}