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

enh: support time line query from union/union all subquery

This commit is contained in:
dapan1121 2023-05-26 08:44:23 +08:00
parent 4c88fd08de
commit e70c433a59
5 changed files with 38 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -327,6 +327,7 @@ typedef struct SSetOperator {
SNode* pLimit;
char stmtName[TSDB_TABLE_NAME_LEN];
uint8_t precision;
ETimeLineMode timeLineResMode;
} SSetOperator;
typedef enum ESqlClause {

1
source/libs/executor/src/sortoperator.c
View File

@ -153,6 +153,7 @@ SSDataBlock* getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock, i
colDataAssign(pDst, pSrc, p->info.rows, &pDataBlock->info);
}
pDataBlock->info.dataLoad = 1;
pDataBlock->info.rows = p->info.rows;
}

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

@ -83,9 +83,10 @@ bool nodesListMatch(const SNodeList* pList, const SNodeList* pSubList) {
}
SNode* node = NULL;
bool match = false;
bool match = true;
FOREACH(node, pList) {
if (!nodesListMatchExists(pSubList, node)) {
match = false;
break;
}
}
@ -93,7 +94,6 @@ bool nodesListMatch(const SNodeList* pList, const SNodeList* pSubList) {
}
static bool columnNodeMatch(const SColumnNode* pSub, const SColumnNode* p) {
MATCH_STRING_FIELD(dbName);
if (0 == strcmp(p->colName, pSub->node.aliasName)) {
return true;
}

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

@ -752,6 +752,8 @@ static SNodeList* getProjectList(const SNode* pNode) {
static bool isTimeLineQuery(SNode* pStmt) {
if (QUERY_NODE_SELECT_STMT == nodeType(pStmt)) {
return (TIME_LINE_MULTI == ((SSelectStmt*)pStmt)->timeLineResMode) || (TIME_LINE_GLOBAL == ((SSelectStmt*)pStmt)->timeLineResMode);
} else if (QUERY_NODE_SET_OPERATOR == nodeType(pStmt)) {
return TIME_LINE_GLOBAL == ((SSetOperator*)pStmt)->timeLineResMode;
} else {
return false;
}
@ -760,6 +762,8 @@ static bool isTimeLineQuery(SNode* pStmt) {
static bool isGlobalTimeLineQuery(SNode* pStmt) {
if (QUERY_NODE_SELECT_STMT == nodeType(pStmt)) {
return TIME_LINE_GLOBAL == ((SSelectStmt*)pStmt)->timeLineResMode;
} else if (QUERY_NODE_SET_OPERATOR == nodeType(pStmt)) {
return TIME_LINE_GLOBAL == ((SSetOperator*)pStmt)->timeLineResMode;
} else {
return false;
}
@ -767,13 +771,16 @@ static bool isGlobalTimeLineQuery(SNode* pStmt) {
static bool isTimeLineAlignedQuery(SNode* pStmt) {
SSelectStmt *pSelect = (SSelectStmt *)pStmt;
SSelectStmt *pSub = (SSelectStmt *)((STempTableNode*)pSelect->pFromTable)->pSubquery;
if (isGlobalTimeLineQuery((SNode*)pSub)) {
if (isGlobalTimeLineQuery(((STempTableNode*)pSelect->pFromTable)->pSubquery)) {
return true;
}
if (!isTimeLineQuery((SNode*)pSub)) {
if (!isTimeLineQuery(((STempTableNode*)pSelect->pFromTable)->pSubquery)) {
return false;
}
if (QUERY_NODE_SELECT_STMT != nodeType(((STempTableNode*)pSelect->pFromTable)->pSubquery)) {
return false;
}
SSelectStmt *pSub = (SSelectStmt *)((STempTableNode*)pSelect->pFromTable)->pSubquery;
if (nodesListMatch(pSelect->pPartitionByList, pSub->pPartitionByList)) {
return true;
}
@ -1582,7 +1589,7 @@ static int32_t translateTimelineFunc(STranslateContext* pCxt, SFunctionNode* pFu
}
SSelectStmt* pSelect = (SSelectStmt*)pCxt->pCurrStmt;
if (NULL != pSelect->pFromTable && QUERY_NODE_TEMP_TABLE == nodeType(pSelect->pFromTable) &&
!isGlobalTimeLineQuery(((STempTableNode*)pSelect->pFromTable)->pSubquery)) {
!isGlobalTimeLineQuery(((STempTableNode*)pSelect->pFromTable)->pSubquery) && !isTimeLineAlignedQuery(pCxt->pCurrStmt)) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_NOT_ALLOWED_FUNC,
"%s function requires valid time series input", pFunc->functionName);
}
@ -3828,8 +3835,13 @@ static int32_t translateSetOperProject(STranslateContext* pCxt, SSetOperator* pS
pLeftExpr = pLeftFuncExpr;
}
snprintf(pRightExpr->aliasName, sizeof(pRightExpr->aliasName), "%s", pLeftExpr->aliasName);
if (TSDB_CODE_SUCCESS != nodesListMakeStrictAppend(&pSetOperator->pProjectionList,
createSetOperProject(pSetOperator->stmtName, pLeft))) {
SNode* pProj = createSetOperProject(pSetOperator->stmtName, pLeft);
if (QUERY_NODE_COLUMN == nodeType(pLeft) && QUERY_NODE_COLUMN == nodeType(pRight)
&& ((SColumnNode*)pLeft)->colId == PRIMARYKEY_TIMESTAMP_COL_ID
&& ((SColumnNode*)pRight)->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
((SColumnNode*)pProj)->colId = PRIMARYKEY_TIMESTAMP_COL_ID;
}
if (TSDB_CODE_SUCCESS != nodesListMakeStrictAppend(&pSetOperator->pProjectionList, pProj)) {
return TSDB_CODE_OUT_OF_MEMORY;
}
}
@ -3841,6 +3853,10 @@ static uint8_t calcSetOperatorPrecision(SSetOperator* pSetOperator) {
}
static int32_t translateSetOperOrderBy(STranslateContext* pCxt, SSetOperator* pSetOperator) {
if (NULL == pSetOperator->pOrderByList || pSetOperator->pOrderByList->length <= 0) {
return TSDB_CODE_SUCCESS;
}
bool other;
int32_t code = translateOrderByPosition(pCxt, pSetOperator->pProjectionList, pSetOperator->pOrderByList, &other);
if (TSDB_CODE_SUCCESS == code) {
@ -3853,6 +3869,14 @@ static int32_t translateSetOperOrderBy(STranslateContext* pCxt, SSetOperator* pS
if (TSDB_CODE_SUCCESS == code) {
code = replaceOrderByAlias(pCxt, pSetOperator->pProjectionList, pSetOperator->pOrderByList);
}
if (TSDB_CODE_SUCCESS == code) {
SNode* pOrder = ((SOrderByExprNode*)nodesListGetNode(pSetOperator->pOrderByList, 0))->pExpr;
if (isPrimaryKeyImpl(pOrder)) {
pSetOperator->timeLineResMode = TIME_LINE_GLOBAL;
} else {
pSetOperator->timeLineResMode = TIME_LINE_NONE;
}
}
return code;
}

4
source/libs/parser/test/parSelectTest.cpp
View File

@ -334,6 +334,10 @@ TEST_F(ParserSelectTest, subquery) {
run("SELECT SUM(a) FROM (SELECT MAX(c1) a, _wstart FROM st1s1 PARTITION BY TBNAME INTERVAL(1m) ORDER BY _WSTART) "
"INTERVAL(1n)");
run("SELECT diff(a) FROM (SELECT _wstart, tag1, tag2, MAX(c1) a FROM st1 PARTITION BY tag1 INTERVAL(1m)) PARTITION BY tag1");
run("SELECT diff(a) FROM (SELECT _wstart, tag1, tag2, MAX(c1) a FROM st1 PARTITION BY tag1 INTERVAL(1m)) PARTITION BY tag2", TSDB_CODE_PAR_NOT_ALLOWED_FUNC);
run("SELECT _C0 FROM (SELECT _ROWTS, ts FROM st1s1)");
run("SELECT ts FROM (SELECT t1.ts FROM st1s1 t1)");