diff --git a/include/libs/nodes/nodes.h b/include/libs/nodes/nodes.h index 3860266725..639a1173cd 100644 --- a/include/libs/nodes/nodes.h +++ b/include/libs/nodes/nodes.h @@ -194,6 +194,7 @@ typedef enum ENodeType { QUERY_NODE_LOGIC_PLAN_FILL, QUERY_NODE_LOGIC_PLAN_SORT, QUERY_NODE_LOGIC_PLAN_PARTITION, + QUERY_NODE_LOGIC_PLAN_INDEF_ROWS_FUNC, QUERY_NODE_LOGIC_SUBPLAN, QUERY_NODE_LOGIC_PLAN, @@ -217,6 +218,7 @@ typedef enum ENodeType { QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION_WINDOW, QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW, QUERY_NODE_PHYSICAL_PLAN_PARTITION, + QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC, QUERY_NODE_PHYSICAL_PLAN_DISPATCH, QUERY_NODE_PHYSICAL_PLAN_INSERT, QUERY_NODE_PHYSICAL_SUBPLAN, diff --git a/include/libs/nodes/plannodes.h b/include/libs/nodes/plannodes.h index 2648a468dd..56a1343dcb 100644 --- a/include/libs/nodes/plannodes.h +++ b/include/libs/nodes/plannodes.h @@ -84,6 +84,11 @@ typedef struct SProjectLogicNode { int64_t soffset; } SProjectLogicNode; +typedef struct SIndefRowsFuncLogicNode { + SLogicNode node; + SNodeList* pVectorFuncs; +} SIndefRowsFuncLogicNode; + typedef struct SVnodeModifLogicNode { SLogicNode node; int32_t msgType; @@ -236,6 +241,12 @@ typedef struct SProjectPhysiNode { int64_t soffset; } SProjectPhysiNode; +typedef struct SIndefRowsFuncPhysiNode { + SPhysiNode node; + SNodeList* pExprs; + SNodeList* pVectorFuncs; +} SIndefRowsFuncPhysiNode; + typedef struct SJoinPhysiNode { SPhysiNode node; EJoinType joinType; diff --git a/source/libs/function/src/builtins.c b/source/libs/function/src/builtins.c index c9d556c86b..831af33add 100644 --- a/source/libs/function/src/builtins.c +++ b/source/libs/function/src/builtins.c @@ -156,14 +156,7 @@ static int32_t translatePercentile(SFunctionNode* pFunc, char* pErrBuf, int32_t return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); } - //param0 - SNode* pParamNode0 = nodesListGetNode(pFunc->pParameterList, 0); - if (nodeType(pParamNode0) != QUERY_NODE_COLUMN) { - return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, - "The first parameter of PERCENTILE function can only be column"); - } - - //param1 + // param1 SValueNode* pValue = (SValueNode*)nodesListGetNode(pFunc->pParameterList, 1); if (pValue->datum.i < 0 || pValue->datum.i > 100) { @@ -178,7 +171,7 @@ static int32_t translatePercentile(SFunctionNode* pFunc, char* pErrBuf, int32_t return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); } - //set result type + // set result type pFunc->node.resType = (SDataType){.bytes = tDataTypes[TSDB_DATA_TYPE_DOUBLE].bytes, .type = TSDB_DATA_TYPE_DOUBLE}; return TSDB_CODE_SUCCESS; } @@ -197,14 +190,7 @@ static int32_t translateApercentile(SFunctionNode* pFunc, char* pErrBuf, int32_t return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); } - //param0 - SNode* pParamNode0 = nodesListGetNode(pFunc->pParameterList, 0); - if (nodeType(pParamNode0) != QUERY_NODE_COLUMN) { - return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, - "The first parameter of APERCENTILE function can only be column"); - } - - //param1 + // param1 SNode* pParamNode1 = nodesListGetNode(pFunc->pParameterList, 1); if (nodeType(pParamNode1) != QUERY_NODE_VALUE) { return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); @@ -223,7 +209,7 @@ static int32_t translateApercentile(SFunctionNode* pFunc, char* pErrBuf, int32_t return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); } - //param2 + // param2 if (3 == numOfParams) { uint8_t para3Type = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 2))->resType.type; if (!IS_VAR_DATA_TYPE(para3Type)) { @@ -263,14 +249,7 @@ static int32_t translateTop(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); } - //param0 - SNode* pParamNode0 = nodesListGetNode(pFunc->pParameterList, 0); - if (nodeType(pParamNode0) != QUERY_NODE_COLUMN) { - return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, - "The first parameter of TOP/BOTTOM function can only be column"); - } - - //param1 + // param1 SNode* pParamNode1 = nodesListGetNode(pFunc->pParameterList, 1); if (nodeType(pParamNode1) != QUERY_NODE_VALUE) { return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); @@ -287,7 +266,7 @@ static int32_t translateTop(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { pValue->notReserved = true; - //set result type + // set result type SDataType* pType = &((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType; pFunc->node.resType = (SDataType){.bytes = pType->bytes, .type = pType->type}; return TSDB_CODE_SUCCESS; @@ -317,13 +296,6 @@ static int32_t translateElapsed(SFunctionNode* pFunc, char* pErrBuf, int32_t len return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); } - // param0 - SNode* pParaNode0 = nodesListGetNode(pFunc->pParameterList, 0); - if (QUERY_NODE_COLUMN != nodeType(pParaNode0)) { - return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, - "The first parameter of ELAPSED function can only be column"); - } - uint8_t paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type; if (TSDB_DATA_TYPE_TIMESTAMP != paraType) { return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); @@ -389,13 +361,6 @@ static int32_t translateHistogram(SFunctionNode* pFunc, char* pErrBuf, int32_t l return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); } - // param0 - SNode* pParaNode0 = nodesListGetNode(pFunc->pParameterList, 0); - if (QUERY_NODE_COLUMN != nodeType(pParaNode0)) { - return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, - "The first parameter of HISTOGRAM function can only be column"); - } - uint8_t colType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type; if (!IS_NUMERIC_TYPE(colType)) { return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); @@ -428,12 +393,6 @@ static int32_t translateHLL(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); } - SNode* pPara = nodesListGetNode(pFunc->pParameterList, 0); - if (QUERY_NODE_COLUMN != nodeType(pPara)) { - return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, - "The input parameter of HYPERLOGLOG function can only be column"); - } - pFunc->node.resType = (SDataType){.bytes = tDataTypes[TSDB_DATA_TYPE_UBIGINT].bytes, .type = TSDB_DATA_TYPE_UBIGINT}; return TSDB_CODE_SUCCESS; } @@ -444,12 +403,6 @@ static int32_t translateStateCount(SFunctionNode* pFunc, char* pErrBuf, int32_t return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); } - // param0 - SNode* pParaNode0 = nodesListGetNode(pFunc->pParameterList, 0); - if (QUERY_NODE_COLUMN != nodeType(pParaNode0)) { - return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, - "The input parameter of STATECOUNT function can only be column"); - } uint8_t colType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type; if (!IS_NUMERIC_TYPE(colType)) { return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); @@ -484,12 +437,6 @@ static int32_t translateStateDuration(SFunctionNode* pFunc, char* pErrBuf, int32 return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); } - // param0 - SNode* pParaNode0 = nodesListGetNode(pFunc->pParameterList, 0); - if (QUERY_NODE_COLUMN != nodeType(pParaNode0)) { - return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, - "The input parameter of STATEDURATION function can only be column"); - } uint8_t colType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type; if (!IS_NUMERIC_TYPE(colType)) { return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); @@ -528,12 +475,6 @@ static int32_t translateCsum(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); } - SNode* pPara = nodesListGetNode(pFunc->pParameterList, 0); - if (QUERY_NODE_COLUMN != nodeType(pPara)) { - return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, - "The input parameter of CSUM function can only be column"); - } - uint8_t colType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type; uint8_t resType; if (!IS_NUMERIC_TYPE(colType)) { @@ -559,13 +500,6 @@ static int32_t translateMavg(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); } - // param0 - SNode* pParaNode0 = nodesListGetNode(pFunc->pParameterList, 0); - if (QUERY_NODE_COLUMN != nodeType(pParaNode0)) { - return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, - "The first parameter of MAVG function can only be column"); - } - uint8_t colType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type; // param1 @@ -595,13 +529,6 @@ static int32_t translateSample(SFunctionNode* pFunc, char* pErrBuf, int32_t len) return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); } - // param0 - SNode* pParamNode0 = nodesListGetNode(pFunc->pParameterList, 0); - if (QUERY_NODE_COLUMN != nodeType(pParamNode0)) { - return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, - "The first parameter of SAMPLE function can only be column"); - } - SExprNode* pCol = (SExprNode*)nodesListGetNode(pFunc->pParameterList, 0); uint8_t colType = pCol->resType.type; @@ -639,12 +566,6 @@ static int32_t translateTail(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); } - // param0 - SNode* pPara = nodesListGetNode(pFunc->pParameterList, 0); - if (QUERY_NODE_COLUMN != nodeType(pPara)) { - return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, - "The first parameter of TAIL function can only be column"); - } SExprNode* pCol = (SExprNode*)nodesListGetNode(pFunc->pParameterList, 0); uint8_t colType = pCol->resType.type; @@ -659,8 +580,8 @@ static int32_t translateTail(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { if (pValue->datum.i < ((i > 1) ? 0 : 1) || pValue->datum.i > 100) { return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, - "TAIL function second parameter should be in range [1, 100], " - "third parameter should be in range [0, 100]"); + "TAIL function second parameter should be in range [1, 100], " + "third parameter should be in range [0, 100]"); } pValue->notReserved = true; @@ -721,20 +642,12 @@ static int32_t translateDiff(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); } - //param0 - SNode* pParamNode0 = nodesListGetNode(pFunc->pParameterList, 0); - if (nodeType(pParamNode0) != QUERY_NODE_COLUMN) { - return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, - "The first parameter of DIFF function can only be column"); - } - uint8_t colType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type; - if (!IS_SIGNED_NUMERIC_TYPE(colType) && !IS_FLOAT_TYPE(colType) && - TSDB_DATA_TYPE_BOOL != colType) { + if (!IS_SIGNED_NUMERIC_TYPE(colType) && !IS_FLOAT_TYPE(colType) && TSDB_DATA_TYPE_BOOL != colType) { return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); } - //param1 + // param1 if (numOfParams == 2) { uint8_t paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 1))->resType.type; if (!IS_INTEGER_TYPE(paraType)) { @@ -852,7 +765,7 @@ static int32_t translateSubstr(SFunctionNode* pFunc, char* pErrBuf, int32_t len) if (3 == numOfParams) { SExprNode* p2 = (SExprNode*)nodesListGetNode(pFunc->pParameterList, 2); - uint8_t para2Type = p2->resType.type; + uint8_t para2Type = p2->resType.type; if (!IS_INTEGER_TYPE(para2Type)) { return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); } diff --git a/source/libs/nodes/src/nodesCloneFuncs.c b/source/libs/nodes/src/nodesCloneFuncs.c index 68d3741b48..40c9e68620 100644 --- a/source/libs/nodes/src/nodesCloneFuncs.c +++ b/source/libs/nodes/src/nodesCloneFuncs.c @@ -408,6 +408,12 @@ static SNode* logicPartitionCopy(const SPartitionLogicNode* pSrc, SPartitionLogi return (SNode*)pDst; } +static SNode* logicIndefRowsFuncCopy(const SIndefRowsFuncLogicNode* pSrc, SIndefRowsFuncLogicNode* pDst) { + COPY_BASE_OBJECT_FIELD(node, logicNodeCopy); + CLONE_NODE_LIST_FIELD(pVectorFuncs); + return (SNode*)pDst; +} + static SNode* logicSubplanCopy(const SLogicSubplan* pSrc, SLogicSubplan* pDst) { COPY_OBJECT_FIELD(id, sizeof(SSubplanId)); CLONE_NODE_FIELD(pNode); @@ -537,6 +543,8 @@ SNodeptr nodesCloneNode(const SNodeptr pNode) { return logicSortCopy((const SSortLogicNode*)pNode, (SSortLogicNode*)pDst); case QUERY_NODE_LOGIC_PLAN_PARTITION: return logicPartitionCopy((const SPartitionLogicNode*)pNode, (SPartitionLogicNode*)pDst); + case QUERY_NODE_LOGIC_PLAN_INDEF_ROWS_FUNC: + return logicIndefRowsFuncCopy((const SIndefRowsFuncLogicNode*)pNode, (SIndefRowsFuncLogicNode*)pDst); case QUERY_NODE_LOGIC_SUBPLAN: return logicSubplanCopy((const SLogicSubplan*)pNode, (SLogicSubplan*)pDst); default: diff --git a/source/libs/nodes/src/nodesCodeFuncs.c b/source/libs/nodes/src/nodesCodeFuncs.c index 78710569cb..e47913a91b 100644 --- a/source/libs/nodes/src/nodesCodeFuncs.c +++ b/source/libs/nodes/src/nodesCodeFuncs.c @@ -198,6 +198,8 @@ const char* nodesNodeName(ENodeType type) { return "LogicSort"; case QUERY_NODE_LOGIC_PLAN_PARTITION: return "LogicPartition"; + case QUERY_NODE_LOGIC_PLAN_INDEF_ROWS_FUNC: + return "LogicIndefRowsFunc"; case QUERY_NODE_LOGIC_SUBPLAN: return "LogicSubplan"; case QUERY_NODE_LOGIC_PLAN: @@ -236,6 +238,8 @@ const char* nodesNodeName(ENodeType type) { return "PhysiStateWindow"; case QUERY_NODE_PHYSICAL_PLAN_PARTITION: return "PhysiPartition"; + case QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC: + return "PhysiIndefRowsFunc"; case QUERY_NODE_PHYSICAL_PLAN_DISPATCH: return "PhysiDispatch"; case QUERY_NODE_PHYSICAL_PLAN_INSERT: @@ -727,6 +731,30 @@ static int32_t jsonToLogicPartitionNode(const SJson* pJson, void* pObj) { return code; } +static const char* jkIndefRowsFuncLogicPlanVectorFuncs = "VectorFuncs"; + +static int32_t logicIndefRowsFuncNodeToJson(const void* pObj, SJson* pJson) { + const SIndefRowsFuncLogicNode* pNode = (const SIndefRowsFuncLogicNode*)pObj; + + int32_t code = logicPlanNodeToJson(pObj, pJson); + if (TSDB_CODE_SUCCESS == code) { + code = nodeListToJson(pJson, jkIndefRowsFuncLogicPlanVectorFuncs, pNode->pVectorFuncs); + } + + return code; +} + +static int32_t jsonToLogicIndefRowsFuncNode(const SJson* pJson, void* pObj) { + SIndefRowsFuncLogicNode* pNode = (SIndefRowsFuncLogicNode*)pObj; + + int32_t code = jsonToLogicPlanNode(pJson, pObj); + if (TSDB_CODE_SUCCESS == code) { + code = jsonToNodeList(pJson, jkIndefRowsFuncLogicPlanVectorFuncs, &pNode->pVectorFuncs); + } + + return code; +} + static const char* jkSubplanIdQueryId = "QueryId"; static const char* jkSubplanIdGroupId = "GroupId"; static const char* jkSubplanIdSubplanId = "SubplanId"; @@ -1744,6 +1772,37 @@ static int32_t jsonToPhysiPartitionNode(const SJson* pJson, void* pObj) { return code; } +static const char* jkIndefRowsFuncPhysiPlanExprs = "Exprs"; +static const char* jkIndefRowsFuncPhysiPlanVectorFuncs = "VectorFuncs"; + +static int32_t physiIndefRowsFuncNodeToJson(const void* pObj, SJson* pJson) { + const SIndefRowsFuncPhysiNode* pNode = (const SIndefRowsFuncPhysiNode*)pObj; + + int32_t code = physicPlanNodeToJson(pObj, pJson); + if (TSDB_CODE_SUCCESS == code) { + code = nodeListToJson(pJson, jkIndefRowsFuncPhysiPlanExprs, pNode->pExprs); + } + if (TSDB_CODE_SUCCESS == code) { + code = nodeListToJson(pJson, jkIndefRowsFuncPhysiPlanVectorFuncs, pNode->pVectorFuncs); + } + + return code; +} + +static int32_t jsonToPhysiIndefRowsFuncNode(const SJson* pJson, void* pObj) { + SIndefRowsFuncPhysiNode* pNode = (SIndefRowsFuncPhysiNode*)pObj; + + int32_t code = jsonToPhysicPlanNode(pJson, pObj); + if (TSDB_CODE_SUCCESS == code) { + code = jsonToNodeList(pJson, jkIndefRowsFuncPhysiPlanExprs, &pNode->pExprs); + } + if (TSDB_CODE_SUCCESS == code) { + code = jsonToNodeList(pJson, jkIndefRowsFuncPhysiPlanVectorFuncs, &pNode->pVectorFuncs); + } + + return code; +} + static const char* jkDataSinkInputDataBlockDesc = "InputDataBlockDesc"; static int32_t physicDataSinkNodeToJson(const void* pObj, SJson* pJson) { @@ -3394,6 +3453,8 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) { return logicSortNodeToJson(pObj, pJson); case QUERY_NODE_LOGIC_PLAN_PARTITION: return logicPartitionNodeToJson(pObj, pJson); + case QUERY_NODE_LOGIC_PLAN_INDEF_ROWS_FUNC: + return logicIndefRowsFuncNodeToJson(pObj, pJson); case QUERY_NODE_LOGIC_SUBPLAN: return logicSubplanToJson(pObj, pJson); case QUERY_NODE_LOGIC_PLAN: @@ -3428,6 +3489,8 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) { return physiStateWindowNodeToJson(pObj, pJson); case QUERY_NODE_PHYSICAL_PLAN_PARTITION: return physiPartitionNodeToJson(pObj, pJson); + case QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC: + return physiIndefRowsFuncNodeToJson(pObj, pJson); case QUERY_NODE_PHYSICAL_PLAN_DISPATCH: return physiDispatchNodeToJson(pObj, pJson); case QUERY_NODE_PHYSICAL_PLAN_INSERT: @@ -3505,6 +3568,8 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) { return jsonToLogicSortNode(pJson, pObj); case QUERY_NODE_LOGIC_PLAN_PARTITION: return jsonToLogicPartitionNode(pJson, pObj); + case QUERY_NODE_LOGIC_PLAN_INDEF_ROWS_FUNC: + return jsonToLogicIndefRowsFuncNode(pJson, pObj); case QUERY_NODE_LOGIC_SUBPLAN: return jsonToLogicSubplan(pJson, pObj); case QUERY_NODE_LOGIC_PLAN: @@ -3539,6 +3604,8 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) { return jsonToPhysiStateWindowNode(pJson, pObj); case QUERY_NODE_PHYSICAL_PLAN_PARTITION: return jsonToPhysiPartitionNode(pJson, pObj); + case QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC: + return jsonToPhysiIndefRowsFuncNode(pJson, pObj); case QUERY_NODE_PHYSICAL_PLAN_DISPATCH: return jsonToPhysiDispatchNode(pJson, pObj); case QUERY_NODE_PHYSICAL_SUBPLAN: diff --git a/source/libs/nodes/src/nodesTraverseFuncs.c b/source/libs/nodes/src/nodesTraverseFuncs.c index ae1ff5744b..9c433aa678 100644 --- a/source/libs/nodes/src/nodesTraverseFuncs.c +++ b/source/libs/nodes/src/nodesTraverseFuncs.c @@ -350,6 +350,7 @@ void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker wa case SQL_CLAUSE_GROUP_BY: nodesWalkExpr(pSelect->pHaving, walker, pContext); case SQL_CLAUSE_HAVING: + case SQL_CLAUSE_SELECT: case SQL_CLAUSE_DISTINCT: nodesWalkExprs(pSelect->pOrderByList, walker, pContext); case SQL_CLAUSE_ORDER_BY: @@ -382,6 +383,7 @@ void nodesRewriteSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeRewrit case SQL_CLAUSE_GROUP_BY: nodesRewriteExpr(&(pSelect->pHaving), rewriter, pContext); case SQL_CLAUSE_HAVING: + case SQL_CLAUSE_SELECT: case SQL_CLAUSE_DISTINCT: nodesRewriteExprs(pSelect->pOrderByList, rewriter, pContext); case SQL_CLAUSE_ORDER_BY: diff --git a/source/libs/nodes/src/nodesUtilFuncs.c b/source/libs/nodes/src/nodesUtilFuncs.c index e28844f2e1..5232c96140 100644 --- a/source/libs/nodes/src/nodesUtilFuncs.c +++ b/source/libs/nodes/src/nodesUtilFuncs.c @@ -230,6 +230,8 @@ SNodeptr nodesMakeNode(ENodeType type) { return makeNode(type, sizeof(SSortLogicNode)); case QUERY_NODE_LOGIC_PLAN_PARTITION: return makeNode(type, sizeof(SPartitionLogicNode)); + case QUERY_NODE_LOGIC_PLAN_INDEF_ROWS_FUNC: + return makeNode(type, sizeof(SIndefRowsFuncLogicNode)); case QUERY_NODE_LOGIC_SUBPLAN: return makeNode(type, sizeof(SLogicSubplan)); case QUERY_NODE_LOGIC_PLAN: @@ -268,6 +270,8 @@ SNodeptr nodesMakeNode(ENodeType type) { return makeNode(type, sizeof(SStateWinodwPhysiNode)); case QUERY_NODE_PHYSICAL_PLAN_PARTITION: return makeNode(type, sizeof(SPartitionPhysiNode)); + case QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC: + return makeNode(type, sizeof(SIndefRowsFuncPhysiNode)); case QUERY_NODE_PHYSICAL_PLAN_DISPATCH: return makeNode(type, sizeof(SDataDispatcherNode)); case QUERY_NODE_PHYSICAL_PLAN_INSERT: diff --git a/source/libs/planner/src/planLogicCreater.c b/source/libs/planner/src/planLogicCreater.c index 467b26b7c4..273f46ac17 100644 --- a/source/libs/planner/src/planLogicCreater.c +++ b/source/libs/planner/src/planLogicCreater.c @@ -476,6 +476,37 @@ static int32_t createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, return code; } +static int32_t createIndefRowsFuncLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SLogicNode** pLogicNode) { + // top/bottom are both an aggregate function and a indefinite rows function + if (!pSelect->hasIndefiniteRowsFunc || pSelect->hasAggFuncs) { + return TSDB_CODE_SUCCESS; + } + + SIndefRowsFuncLogicNode* pIdfRowsFunc = + (SIndefRowsFuncLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_INDEF_ROWS_FUNC); + if (NULL == pIdfRowsFunc) { + return TSDB_CODE_OUT_OF_MEMORY; + } + + int32_t code = nodesCollectFuncs(pSelect, SQL_CLAUSE_SELECT, fmIsVectorFunc, &pIdfRowsFunc->pVectorFuncs); + if (TSDB_CODE_SUCCESS == code) { + code = rewriteExprForSelect(pIdfRowsFunc->pVectorFuncs, pSelect, SQL_CLAUSE_SELECT); + } + + // set the output + if (TSDB_CODE_SUCCESS == code) { + code = createColumnByRewriteExps(pCxt, pIdfRowsFunc->pVectorFuncs, &pIdfRowsFunc->node.pTargets); + } + + if (TSDB_CODE_SUCCESS == code) { + *pLogicNode = (SLogicNode*)pIdfRowsFunc; + } else { + nodesDestroyNode(pIdfRowsFunc); + } + + return code; +} + static int32_t createWindowLogicNodeFinalize(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SWindowLogicNode* pWindow, SLogicNode** pLogicNode) { int32_t code = nodesCollectFuncs(pSelect, SQL_CLAUSE_WINDOW, fmIsWindowClauseFunc, &pWindow->pFuncs); @@ -787,6 +818,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, createIndefRowsFuncLogicNode, &pRoot); + } if (TSDB_CODE_SUCCESS == code) { code = createChildLogicNode(pCxt, pSelect, createDistinctLogicNode, &pRoot); } diff --git a/source/libs/planner/src/planPhysiCreater.c b/source/libs/planner/src/planPhysiCreater.c index a45eabefb9..69fa5499c4 100644 --- a/source/libs/planner/src/planPhysiCreater.c +++ b/source/libs/planner/src/planPhysiCreater.c @@ -800,6 +800,43 @@ static int32_t createAggPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren, return code; } +static int32_t createIndefRowsFuncPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren, + SIndefRowsFuncLogicNode* pFuncLogicNode, SPhysiNode** pPhyNode) { + SIndefRowsFuncPhysiNode* pIdfRowsFunc = (SIndefRowsFuncPhysiNode*)makePhysiNode( + pCxt, getPrecision(pChildren), (SLogicNode*)pFuncLogicNode, QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC); + if (NULL == pIdfRowsFunc) { + return TSDB_CODE_OUT_OF_MEMORY; + } + + SNodeList* pPrecalcExprs = NULL; + SNodeList* pVectorFuncs = NULL; + int32_t code = rewritePrecalcExprs(pCxt, pFuncLogicNode->pVectorFuncs, &pPrecalcExprs, &pVectorFuncs); + + 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, &pIdfRowsFunc->pExprs); + if (TSDB_CODE_SUCCESS == code) { + code = pushdownDataBlockSlots(pCxt, pIdfRowsFunc->pExprs, pChildTupe); + } + } + + if (TSDB_CODE_SUCCESS == code && NULL != pVectorFuncs) { + code = setListSlotId(pCxt, pChildTupe->dataBlockId, -1, pVectorFuncs, &pIdfRowsFunc->pVectorFuncs); + if (TSDB_CODE_SUCCESS == code) { + code = addDataBlockSlots(pCxt, pIdfRowsFunc->pVectorFuncs, pIdfRowsFunc->node.pOutputDataBlockDesc); + } + } + + if (TSDB_CODE_SUCCESS == code) { + *pPhyNode = (SPhysiNode*)pIdfRowsFunc; + } else { + nodesDestroyNode(pIdfRowsFunc); + } + + return code; +} + static int32_t createProjectPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren, SProjectLogicNode* pProjectLogicNode, SPhysiNode** pPhyNode) { SProjectPhysiNode* pProject = (SProjectPhysiNode*)makePhysiNode( @@ -949,7 +986,8 @@ static int32_t createSessionWindowPhysiNode(SPhysiPlanContext* pCxt, SNodeList* SWindowLogicNode* pWindowLogicNode, SPhysiNode** pPhyNode) { SSessionWinodwPhysiNode* pSession = (SSessionWinodwPhysiNode*)makePhysiNode( pCxt, getPrecision(pChildren), (SLogicNode*)pWindowLogicNode, - (pCxt->pPlanCxt->streamQuery ? QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION_WINDOW : QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW)); + (pCxt->pPlanCxt->streamQuery ? QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION_WINDOW + : QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW)); if (NULL == pSession) { return TSDB_CODE_OUT_OF_MEMORY; } @@ -1153,6 +1191,8 @@ static int32_t doCreatePhysiNode(SPhysiPlanContext* pCxt, SLogicNode* pLogicNode return createPartitionPhysiNode(pCxt, pChildren, (SPartitionLogicNode*)pLogicNode, pPhyNode); case QUERY_NODE_LOGIC_PLAN_FILL: return createFillPhysiNode(pCxt, pChildren, (SFillLogicNode*)pLogicNode, pPhyNode); + case QUERY_NODE_LOGIC_PLAN_INDEF_ROWS_FUNC: + return createIndefRowsFuncPhysiNode(pCxt, pChildren, (SIndefRowsFuncLogicNode*)pLogicNode, pPhyNode); default: break; } diff --git a/source/libs/planner/test/planProjectTest.cpp b/source/libs/planner/test/planProjectTest.cpp new file mode 100644 index 0000000000..3ef0038ae0 --- /dev/null +++ b/source/libs/planner/test/planProjectTest.cpp @@ -0,0 +1,34 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * This program is free software: you can use, redistribute, and/or modify + * it under the terms of the GNU Affero General Public License, version 3 + * or later ("AGPL"), as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +#include "planTestUtil.h" + +using namespace std; + +class PlanProjectTest : public PlannerTestBase {}; + +TEST_F(PlanProjectTest, basic) { + useDb("root", "test"); + + run("SELECT CEIL(c1) FROM t1"); +} + +TEST_F(PlanProjectTest, indefiniteRowsFunc) { + useDb("root", "test"); + + run("SELECT MAVG(c1, 10) FROM t1"); + + run("SELECT MAVG(CEIL(c1), 20) + 2 FROM t1"); +}