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

feat: sql function 'last_row'

This commit is contained in:
Xiaoyu Wang 2022-06-21 19:52:26 +08:00
parent 78caa7e005
commit 45a806f789
15 changed files with 103 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/libs/function/functionMgt.h
View File

@ -190,6 +190,7 @@ bool fmIsForbidWindowFunc(int32_t funcId);
bool fmIsForbidGroupByFunc(int32_t funcId);
bool fmIsIntervalInterpoFunc(int32_t funcId);
bool fmIsInterpFunc(int32_t funcId);
bool fmIsLastRowFunc(int32_t funcId);
int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc, SFunctionNode** pMergeFunc);

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

@ -59,10 +59,10 @@ extern "C" {
for (SListCell* cell = (NULL != (list) ? (list)->pHead : NULL); \
(NULL != cell ? (node = &(cell->pNode), true) : (node = NULL, false)); cell = cell->pNext)
#define DESTORY_LIST(list) \
do { \
nodesDestroyList((list)); \
(list) = NULL; \
#define NODES_DESTORY_LIST(list) \
do { \
nodesDestroyList((list)); \
(list) = NULL; \
} while (0)
#define NODES_CLEAR_LIST(list) \
@ -219,6 +219,7 @@ typedef enum ENodeType {
QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN,
QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN,
QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN,
QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN,
QUERY_NODE_PHYSICAL_PLAN_PROJECT,
QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN,
QUERY_NODE_PHYSICAL_PLAN_HASH_AGG,

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

@ -40,7 +40,8 @@ typedef enum EScanType {
SCAN_TYPE_SYSTEM_TABLE,
SCAN_TYPE_STREAM,
SCAN_TYPE_TABLE_MERGE,
SCAN_TYPE_BLOCK_INFO
SCAN_TYPE_BLOCK_INFO,
SCAN_TYPE_LAST_ROW
} EScanType;
typedef struct SScanLogicNode {
@ -260,6 +261,7 @@ typedef struct SScanPhysiNode {
typedef SScanPhysiNode STagScanPhysiNode;
typedef SScanPhysiNode SBlockDistScanPhysiNode;
typedef SScanPhysiNode SLastRowScanPhysiNode;
typedef struct SSystemTableScanPhysiNode {
SScanPhysiNode scan;

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

@ -258,6 +258,7 @@ typedef struct SSelectStmt {
bool hasUniqueFunc;
bool hasTailFunc;
bool hasInterpFunc;
bool hasLastRowFunc;
} SSelectStmt;
typedef enum ESetOperatorType { SET_OP_TYPE_UNION_ALL = 1, SET_OP_TYPE_UNION } ESetOperatorType;

2
source/libs/function/src/builtins.c
View File

@ -1799,7 +1799,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "last_row",
.type = FUNCTION_TYPE_LAST_ROW,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC,
.classification = FUNC_MGT_MULTI_RES_FUNC,
.translateFunc = translateLastRow,
.getEnvFunc = getMinmaxFuncEnv,
.initFunc = minmaxFunctionSetup,

7
source/libs/function/src/functionMgt.c
View File

@ -186,6 +186,13 @@ bool fmIsInterpFunc(int32_t funcId) {
return FUNCTION_TYPE_INTERP == funcMgtBuiltins[funcId].type;
}
bool fmIsLastRowFunc(int32_t funcId) {
if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
return false;
}
return FUNCTION_TYPE_LAST_ROW == funcMgtBuiltins[funcId].type;
}
void fmFuncMgtDestroy() {
void* m = gFunMgtService.pFuncNameHashTable;
if (m != NULL && atomic_val_compare_exchange_ptr((void**)&gFunMgtService.pFuncNameHashTable, m, 0) == m) {

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

@ -220,6 +220,8 @@ const char* nodesNodeName(ENodeType type) {
return "PhysiSystemTableScan";
case QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN:
return "PhysiBlockDistScan";
case QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN:
return "PhysiLastRowScan";
case QUERY_NODE_PHYSICAL_PLAN_PROJECT:
return "PhysiProject";
case QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN:
@ -4105,6 +4107,7 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
return logicPlanToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN:
return physiTagScanNodeToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN:
@ -4245,6 +4248,7 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
return jsonToLogicPlan(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN:
return jsonToPhysiTagScanNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN:

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

@ -273,6 +273,8 @@ SNode* nodesMakeNode(ENodeType type) {
return makeNode(type, sizeof(SSystemTableScanPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN:
return makeNode(type, sizeof(SBlockDistScanPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN:
return makeNode(type, sizeof(SLastRowScanPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_PROJECT:
return makeNode(type, sizeof(SProjectPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN:
@ -781,6 +783,7 @@ void nodesDestroyNode(SNode* pNode) {
case QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN:
destroyScanPhysiNode((SScanPhysiNode*)pNode);
break;
case QUERY_NODE_PHYSICAL_PLAN_PROJECT: {

2
source/libs/parser/src/parCalcConst.c
View File

@ -227,7 +227,7 @@ static int32_t calcConstGroupBy(SCalcConstContext* pCxt, SSelectStmt* pSelect) {
}
}
}
DESTORY_LIST(pSelect->pGroupByList);
NODES_DESTORY_LIST(pSelect->pGroupByList);
}
return code;
}

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

@ -1176,6 +1176,7 @@ static void setFuncClassification(SSelectStmt* pSelect, SFunctionNode* pFunc) {
pSelect->hasUniqueFunc = pSelect->hasUniqueFunc ? true : (FUNCTION_TYPE_UNIQUE == pFunc->funcType);
pSelect->hasTailFunc = pSelect->hasTailFunc ? true : (FUNCTION_TYPE_TAIL == pFunc->funcType);
pSelect->hasInterpFunc = pSelect->hasInterpFunc ? true : (FUNCTION_TYPE_INTERP == pFunc->funcType);
pSelect->hasLastRowFunc = pSelect->hasLastRowFunc ? true : (FUNCTION_TYPE_LAST_ROW == pFunc->funcType);
}
}

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

@ -296,6 +296,59 @@ static int32_t createScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
return code;
}
static int32_t createColumnByLastRow(SNodeList* pFuncs, SNodeList** pOutput) {
int32_t code = TSDB_CODE_SUCCESS;
SNodeList* pCols = NULL;
SNode* pFunc = NULL;
FOREACH(pFunc, pFuncs) {
SFunctionNode* pLastRow = (SFunctionNode*)pFunc;
SColumnNode* pCol = (SColumnNode*)nodesListGetNode(pLastRow->pParameterList, 0);
snprintf(pCol->colName, sizeof(pCol->colName), "%s", pLastRow->node.aliasName);
snprintf(pCol->node.aliasName, sizeof(pCol->colName), "%s", pLastRow->node.aliasName);
NODES_CLEAR_LIST(pLastRow->pParameterList);
code = nodesListMakeStrictAppend(&pCols, (SNode*)pCol);
if (TSDB_CODE_SUCCESS != code) {
break;
}
}
if (TSDB_CODE_SUCCESS == code) {
*pOutput = pCols;
} else {
nodesDestroyList(pCols);
}
return code;
}
static int32_t createLastRowScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SRealTableNode* pRealTable,
SLogicNode** pLogicNode) {
SScanLogicNode* pScan = NULL;
int32_t code = makeScanLogicNode(pCxt, pRealTable, false, (SLogicNode**)&pScan);
SNodeList* pFuncs = NULL;
if (TSDB_CODE_SUCCESS == code) {
pScan->scanType = SCAN_TYPE_LAST_ROW;
code = nodesCollectFuncs(pSelect, SQL_CLAUSE_FROM, fmIsLastRowFunc, &pFuncs);
}
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pFuncs, pSelect, SQL_CLAUSE_FROM);
}
if (TSDB_CODE_SUCCESS == code) {
code = createColumnByLastRow(pFuncs, &pScan->pScanCols);
}
if (TSDB_CODE_SUCCESS == code) {
code = createColumnByRewriteExprs(pScan->pScanCols, &pScan->node.pTargets);
}
if (TSDB_CODE_SUCCESS == code) {
*pLogicNode = (SLogicNode*)pScan;
} else {
nodesDestroyNode((SNode*)pScan);
}
nodesDestroyList(pFuncs);
return code;
}
static int32_t createSubqueryLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, STempTableNode* pTable,
SLogicNode** pLogicNode) {
return createQueryLogicNode(pCxt, pTable->pSubquery, pLogicNode);
@ -367,7 +420,11 @@ static int32_t doCreateLogicNodeByTable(SLogicPlanContext* pCxt, SSelectStmt* pS
SLogicNode** pLogicNode) {
switch (nodeType(pTable)) {
case QUERY_NODE_REAL_TABLE:
return createScanLogicNode(pCxt, pSelect, (SRealTableNode*)pTable, pLogicNode);
if (pSelect->hasLastRowFunc) {
return createLastRowScanLogicNode(pCxt, pSelect, (SRealTableNode*)pTable, pLogicNode);
} else {
return createScanLogicNode(pCxt, pSelect, (SRealTableNode*)pTable, pLogicNode);
}
case QUERY_NODE_TEMP_TABLE:
return createSubqueryLogicNode(pCxt, pSelect, (STempTableNode*)pTable, pLogicNode);
case QUERY_NODE_JOIN_TABLE:

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

@ -1126,7 +1126,7 @@ static int32_t partTagsOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSub
break;
}
}
DESTORY_LIST(((SAggLogicNode*)pNode)->pGroupKeys);
NODES_DESTORY_LIST(((SAggLogicNode*)pNode)->pGroupKeys);
}
if (TSDB_CODE_SUCCESS == code) {
code = partTagsOptRebuildTbanme(pScan->pPartTags);

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

@ -462,6 +462,8 @@ static ENodeType getScanOperatorType(EScanType scanType) {
return QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN;
case SCAN_TYPE_BLOCK_INFO:
return QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN;
case SCAN_TYPE_LAST_ROW:
return QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN;
default:
break;
}
@ -559,6 +561,7 @@ static int32_t createScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan* pSubplan,
switch (pScanLogicNode->scanType) {
case SCAN_TYPE_TAG:
case SCAN_TYPE_BLOCK_INFO:
case SCAN_TYPE_LAST_ROW:
return createSimpleScanPhysiNode(pCxt, pSubplan, pScanLogicNode, pPhyNode);
case SCAN_TYPE_TABLE:
return createTableScanPhysiNode(pCxt, pSubplan, pScanLogicNode, pPhyNode);
@ -732,7 +735,7 @@ static int32_t rewritePrecalcExprs(SPhysiPlanContext* pCxt, SNodeList* pList, SN
SRewritePrecalcExprsCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pPrecalcExprs = *pPrecalcExprs};
nodesRewriteExprs(*pRewrittenList, doRewritePrecalcExprs, &cxt);
if (0 == LIST_LENGTH(cxt.pPrecalcExprs) || TSDB_CODE_SUCCESS != cxt.errCode) {
DESTORY_LIST(*pPrecalcExprs);
NODES_DESTORY_LIST(*pPrecalcExprs);
}
return cxt.errCode;
}

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

@ -934,7 +934,7 @@ static int32_t unionSplitSubplan(SSplitContext* pCxt, SLogicSubplan* pUnionSubpl
}
if (TSDB_CODE_SUCCESS == code) {
nodesDestroyList(pSubplanChildren);
DESTORY_LIST(pSplitNode->pChildren);
NODES_DESTORY_LIST(pSplitNode->pChildren);
}
return code;
}

12
source/libs/planner/test/planBasicTest.cpp
View File

@ -83,3 +83,15 @@ TEST_F(PlanBasicTest, interpFunc) {
run("SELECT INTERP(c1) FROM t1 RANGE('2017-7-14 18:00:00', '2017-7-14 19:00:00') EVERY(5s) FILL(LINEAR)");
}
TEST_F(PlanBasicTest, lastRowFunc) {
useDb("root", "test");
run("SELECT LAST_ROW(c1) FROM t1");
run("SELECT LAST_ROW(*) FROM t1");
run("SELECT LAST_ROW(c1, c2) FROM t1");
run("SELECT LAST_ROW(c1) FROM st1");
}