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

enh: the fill physical scheme distinguishes between interpolated and non-interpolated expressions

This commit is contained in:
Xiaoyu Wang 2022-08-02 16:18:53 +08:00
parent 8ea865fe5c
commit f8bd778979
11 changed files with 131 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -211,6 +211,8 @@ typedef struct SWindowLogicNode {
typedef struct SFillLogicNode {
SLogicNode node;
EFillMode mode;
SNodeList* pFillExprs;
SNodeList* pNotFillExprs;
SNode* pWStartTs;
SNode* pValues; // SNodeListNode
STimeWindow timeRange;
@ -435,9 +437,10 @@ typedef SIntervalPhysiNode SStreamSemiIntervalPhysiNode;
typedef struct SFillPhysiNode {
SPhysiNode node;
EFillMode mode;
SNodeList* pFillExprs;
SNodeList* pNotFillExprs;
SNode* pWStartTs; // SColumnNode
SNode* pValues; // SNodeListNode
SNodeList* pTargets;
STimeWindow timeRange;
EOrder inputTsOrder;
} SFillPhysiNode;

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

@ -53,7 +53,13 @@ typedef struct SExprNode {
bool orderAlias;
} SExprNode;
typedef enum EColumnType { COLUMN_TYPE_COLUMN = 1, COLUMN_TYPE_TAG, COLUMN_TYPE_TBNAME } EColumnType;
typedef enum EColumnType {
COLUMN_TYPE_COLUMN = 1,
COLUMN_TYPE_TAG,
COLUMN_TYPE_TBNAME,
COLUMN_TYPE_WINDOW_PC,
COLUMN_TYPE_GROUP_KEY
} EColumnType;
typedef struct SColumnNode {
SExprNode node; // QUERY_NODE_COLUMN
@ -291,6 +297,7 @@ typedef enum ESqlClause {
SQL_CLAUSE_WHERE,
SQL_CLAUSE_PARTITION_BY,
SQL_CLAUSE_WINDOW,
SQL_CLAUSE_FILL,
SQL_CLAUSE_GROUP_BY,
SQL_CLAUSE_HAVING,
SQL_CLAUSE_DISTINCT,

20
source/libs/executor/src/executorimpl.c
View File

@ -3596,7 +3596,8 @@ void doDestroyExchangeOperatorInfo(void* param) {
}
static int32_t initFillInfo(SFillOperatorInfo* pInfo, SExprInfo* pExpr, int32_t numOfCols, SNodeListNode* pValNode,
STimeWindow win, int32_t capacity, const char* id, SInterval* pInterval, int32_t fillType, int32_t order) {
STimeWindow win, int32_t capacity, const char* id, SInterval* pInterval, int32_t fillType,
int32_t order) {
SFillColInfo* pColInfo = createFillColInfo(pExpr, numOfCols, pValNode);
STimeWindow w = getAlignQueryTimeWindow(pInterval, pInterval->precision, win.skey);
@ -3627,13 +3628,13 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode*
int32_t num = 0;
SSDataBlock* pResBlock = createResDataBlock(pPhyFillNode->node.pOutputDataBlockDesc);
SExprInfo* pExprInfo = createExprInfo(pPhyFillNode->pTargets, NULL, &num);
SExprInfo* pExprInfo = createExprInfo(pPhyFillNode->pFillExprs, NULL, &num);
SInterval* pInterval =
QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL == downstream->operatorType
? &((SMergeAlignedIntervalAggOperatorInfo*)downstream->info)->intervalAggOperatorInfo->interval
: &((SIntervalAggOperatorInfo*)downstream->info)->interval;
int32_t order = (pPhyFillNode->inputTsOrder == ORDER_ASC)? TSDB_ORDER_ASC:TSDB_ORDER_DESC;
int32_t order = (pPhyFillNode->inputTsOrder == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
int32_t type = convertFillType(pPhyFillNode->mode);
SResultInfo* pResultInfo = &pOperator->resultInfo;
@ -3641,7 +3642,7 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode*
pInfo->primaryTsCol = ((SColumnNode*)pPhyFillNode->pWStartTs)->slotId;
int32_t numOfOutputCols = 0;
SArray* pColMatchColInfo = extractColMatchInfo(pPhyFillNode->pTargets, pPhyFillNode->node.pOutputDataBlockDesc,
SArray* pColMatchColInfo = extractColMatchInfo(pPhyFillNode->pFillExprs, pPhyFillNode->node.pOutputDataBlockDesc,
&numOfOutputCols, COL_MATCH_FROM_SLOT_ID);
int32_t code = initFillInfo(pInfo, pExprInfo, num, (SNodeListNode*)pPhyFillNode->pValues, pPhyFillNode->timeRange,
@ -3835,7 +3836,7 @@ static int32_t sortTableGroup(STableListInfo* pTableListInfo, int32_t groupNum)
return TDB_CODE_SUCCESS;
}
bool groupbyTbname(SNodeList* pGroupList) {
bool groupbyTbname(SNodeList* pGroupList) {
bool bytbname = false;
if (LIST_LENGTH(pGroupList) > 0) {
SNode* p = nodesListGetNode(pGroupList, 0);
@ -3877,7 +3878,7 @@ int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle,
bool assignUid = groupbyTbname(group);
int32_t groupNum = 0;
size_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);
size_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);
for (int32_t i = 0; i < numOfTables; i++) {
STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
@ -4619,7 +4620,7 @@ void releaseQueryBuf(size_t numOfTables) {
}
int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInfoList) {
SExplainExecInfo execInfo = {0};
SExplainExecInfo execInfo = {0};
SExplainExecInfo* pExplainInfo = taosArrayPush(pExecInfoList, &execInfo);
pExplainInfo->numOfRows = operatorInfo->resultInfo.totalRows;
@ -4629,7 +4630,8 @@ int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInf
pExplainInfo->verboseInfo = NULL;
if (operatorInfo->fpSet.getExplainFn) {
int32_t code = operatorInfo->fpSet.getExplainFn(operatorInfo, &pExplainInfo->verboseInfo, &pExplainInfo->verboseLen);
int32_t code =
operatorInfo->fpSet.getExplainFn(operatorInfo, &pExplainInfo->verboseInfo, &pExplainInfo->verboseLen);
if (code) {
qError("%s operator getExplainFn failed, code:%s", GET_TASKID(operatorInfo->pTaskInfo), tstrerror(code));
return code;
@ -4640,7 +4642,7 @@ int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInf
for (int32_t i = 0; i < operatorInfo->numOfDownstream; ++i) {
code = getOperatorExplainExecInfo(operatorInfo->pDownstream[i], pExecInfoList);
if (code != TSDB_CODE_SUCCESS) {
// taosMemoryFreeClear(*pRes);
// taosMemoryFreeClear(*pRes);
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
}

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

@ -450,6 +450,8 @@ static int32_t logicWindowCopy(const SWindowLogicNode* pSrc, SWindowLogicNode* p
static int32_t logicFillCopy(const SFillLogicNode* pSrc, SFillLogicNode* pDst) {
COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
COPY_SCALAR_FIELD(mode);
CLONE_NODE_LIST_FIELD(pFillExprs);
CLONE_NODE_LIST_FIELD(pNotFillExprs);
CLONE_NODE_FIELD(pWStartTs);
CLONE_NODE_FIELD(pValues);
COPY_OBJECT_FIELD(timeRange, sizeof(STimeWindow));

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

@ -2062,9 +2062,10 @@ static int32_t jsonToPhysiIntervalNode(const SJson* pJson, void* pObj) {
}
static const char* jkFillPhysiPlanMode = "Mode";
static const char* jkFillPhysiPlanFillExprs = "FillExprs";
static const char* jkFillPhysiPlanNotFillExprs = "NotFillExprs";
static const char* jkFillPhysiPlanWStartTs = "WStartTs";
static const char* jkFillPhysiPlanValues = "Values";
static const char* jkFillPhysiPlanTargets = "Targets";
static const char* jkFillPhysiPlanStartTime = "StartTime";
static const char* jkFillPhysiPlanEndTime = "EndTime";
static const char* jkFillPhysiPlanInputTsOrder = "inputTsOrder";
@ -2076,15 +2077,18 @@ static int32_t physiFillNodeToJson(const void* pObj, SJson* pJson) {
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkFillPhysiPlanMode, pNode->mode);
}
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkFillPhysiPlanFillExprs, pNode->pFillExprs);
}
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkFillPhysiPlanNotFillExprs, pNode->pNotFillExprs);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkFillPhysiPlanWStartTs, nodeToJson, pNode->pWStartTs);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkFillPhysiPlanValues, nodeToJson, pNode->pValues);
}
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkFillPhysiPlanTargets, pNode->pTargets);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkFillPhysiPlanStartTime, pNode->timeRange.skey);
}
@ -2104,7 +2108,12 @@ static int32_t jsonToPhysiFillNode(const SJson* pJson, void* pObj) {
int32_t code = jsonToPhysicPlanNode(pJson, pObj);
if (TSDB_CODE_SUCCESS == code) {
tjsonGetNumberValue(pJson, jkFillPhysiPlanMode, pNode->mode, code);
;
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkFillPhysiPlanFillExprs, &pNode->pFillExprs);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkFillPhysiPlanNotFillExprs, &pNode->pNotFillExprs);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkFillPhysiPlanWStartTs, &pNode->pWStartTs);
@ -2112,9 +2121,6 @@ static int32_t jsonToPhysiFillNode(const SJson* pJson, void* pObj) {
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkFillPhysiPlanValues, &pNode->pValues);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkFillPhysiPlanTargets, &pNode->pTargets);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBigIntValue(pJson, jkFillPhysiPlanStartTime, &pNode->timeRange.skey);
}

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

@ -346,6 +346,7 @@ void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker wa
if (NULL != pSelect->pWindow && QUERY_NODE_INTERVAL_WINDOW == nodeType(pSelect->pWindow)) {
nodesWalkExpr(((SIntervalWindowNode*)pSelect->pWindow)->pFill, walker, pContext);
}
case SQL_CLAUSE_FILL:
nodesWalkExprs(pSelect->pGroupByList, walker, pContext);
case SQL_CLAUSE_GROUP_BY:
nodesWalkExpr(pSelect->pHaving, walker, pContext);
@ -379,6 +380,7 @@ void nodesRewriteSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeRewrit
if (NULL != pSelect->pWindow && QUERY_NODE_INTERVAL_WINDOW == nodeType(pSelect->pWindow)) {
nodesRewriteExpr(&(((SIntervalWindowNode*)pSelect->pWindow)->pFill), rewriter, pContext);
}
case SQL_CLAUSE_FILL:
nodesRewriteExprs(pSelect->pGroupByList, rewriter, pContext);
case SQL_CLAUSE_GROUP_BY:
nodesRewriteExpr(&(pSelect->pHaving), rewriter, pContext);

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

@ -931,9 +931,10 @@ void nodesDestroyNode(SNode* pNode) {
case QUERY_NODE_PHYSICAL_PLAN_FILL: {
SFillPhysiNode* pPhyNode = (SFillPhysiNode*)pNode;
destroyPhysiNode((SPhysiNode*)pPhyNode);
nodesDestroyList(pPhyNode->pFillExprs);
nodesDestroyList(pPhyNode->pNotFillExprs);
nodesDestroyNode(pPhyNode->pWStartTs);
nodesDestroyNode(pPhyNode->pValues);
nodesDestroyList(pPhyNode->pTargets);
break;
}
case QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION:

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

@ -38,6 +38,27 @@ typedef struct SRewriteExprCxt {
SNodeList* pExprs;
} SRewriteExprCxt;
static void setColumnInfo(SFunctionNode* pFunc, SColumnNode* pCol) {
switch (pFunc->funcType) {
case FUNCTION_TYPE_TBNAME:
pCol->colType = COLUMN_TYPE_TBNAME;
break;
case FUNCTION_TYPE_WSTART:
case FUNCTION_TYPE_WEND:
pCol->colId = PRIMARYKEY_TIMESTAMP_COL_ID;
pCol->colType = COLUMN_TYPE_WINDOW_PC;
break;
case FUNCTION_TYPE_WDURATION:
pCol->colType = COLUMN_TYPE_WINDOW_PC;
break;
case FUNCTION_TYPE_GROUP_KEY:
pCol->colType = COLUMN_TYPE_GROUP_KEY;
break;
default:
break;
}
}
static EDealRes doRewriteExpr(SNode** pNode, void* pContext) {
switch (nodeType(*pNode)) {
case QUERY_NODE_OPERATOR:
@ -60,11 +81,7 @@ static EDealRes doRewriteExpr(SNode** pNode, void* pContext) {
strcpy(pCol->node.aliasName, pToBeRewrittenExpr->aliasName);
strcpy(pCol->colName, ((SExprNode*)pExpr)->aliasName);
if (QUERY_NODE_FUNCTION == nodeType(pExpr)) {
if (FUNCTION_TYPE_WSTART == ((SFunctionNode*)pExpr)->funcType) {
pCol->colId = PRIMARYKEY_TIMESTAMP_COL_ID;
} else if (FUNCTION_TYPE_TBNAME == ((SFunctionNode*)pExpr)->funcType) {
pCol->colType = COLUMN_TYPE_TBNAME;
}
setColumnInfo((SFunctionNode*)pExpr, pCol);
}
nodesDestroyNode(*pNode);
*pNode = (SNode*)pCol;
@ -746,6 +763,41 @@ static int32_t createWindowLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSele
return TSDB_CODE_FAILED;
}
static EDealRes needFillValueImpl(SNode* pNode, void* pContext) {
if (QUERY_NODE_COLUMN == nodeType(pNode)) {
SColumnNode* pCol = (SColumnNode*)pNode;
if (COLUMN_TYPE_WINDOW_PC != pCol->colType && COLUMN_TYPE_GROUP_KEY != pCol->colType) {
*(bool*)pContext = true;
return DEAL_RES_END;
}
}
return DEAL_RES_CONTINUE;
}
static bool needFillValue(SNode* pNode) {
bool hasFillCol = false;
nodesWalkExpr(pNode, needFillValueImpl, &hasFillCol);
return hasFillCol;
}
static int32_t partFillExprs(SNodeList* pProjectionList, SNodeList** pFillExprs, SNodeList** pNotFillExprs) {
int32_t code = TSDB_CODE_SUCCESS;
SNode* pProject = NULL;
FOREACH(pProject, pProjectionList) {
if (needFillValue(pProject)) {
code = nodesListMakeStrictAppend(pFillExprs, nodesCloneNode(pProject));
} else {
code = nodesListMakeStrictAppend(pNotFillExprs, nodesCloneNode(pProject));
}
if (TSDB_CODE_SUCCESS != code) {
NODES_DESTORY_LIST(*pFillExprs);
NODES_DESTORY_LIST(*pNotFillExprs);
break;
}
}
return code;
}
static int32_t createFillLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
if (NULL == pSelect->pWindow || QUERY_NODE_INTERVAL_WINDOW != nodeType(pSelect->pWindow) ||
NULL == ((SIntervalWindowNode*)pSelect->pWindow)->pFill) {
@ -767,10 +819,18 @@ static int32_t createFillLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
pFill->node.resultDataOrder = DATA_ORDER_LEVEL_IN_GROUP;
pFill->inputTsOrder = ORDER_ASC;
int32_t code = nodesCollectColumns(pSelect, SQL_CLAUSE_WINDOW, NULL, COLLECT_COL_TYPE_ALL, &pFill->node.pTargets);
if (TSDB_CODE_SUCCESS == code && NULL == pFill->node.pTargets) {
code = nodesListMakeStrictAppend(&pFill->node.pTargets,
nodesCloneNode(nodesListGetNode(pCxt->pCurrRoot->pTargets, 0)));
int32_t code = partFillExprs(pSelect->pProjectionList, &pFill->pFillExprs, &pFill->pNotFillExprs);
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pFill->pFillExprs, pSelect, SQL_CLAUSE_FILL);
}
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pFill->pNotFillExprs, pSelect, SQL_CLAUSE_FILL);
}
if (TSDB_CODE_SUCCESS == code) {
code = createColumnByRewriteExprs(pFill->pFillExprs, &pFill->node.pTargets);
}
if (TSDB_CODE_SUCCESS == code) {
code = createColumnByRewriteExprs(pFill->pNotFillExprs, &pFill->node.pTargets);
}
pFill->mode = pFillNode->mode;

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

@ -311,6 +311,10 @@ static EDealRes doSetSlotId(SNode* pNode, void* pContext) {
static int32_t setNodeSlotId(SPhysiPlanContext* pCxt, int16_t leftDataBlockId, int16_t rightDataBlockId, SNode* pNode,
SNode** pOutput) {
if (NULL == pNode) {
return TSDB_CODE_SUCCESS;
}
SNode* pRes = nodesCloneNode(pNode);
if (NULL == pRes) {
return TSDB_CODE_OUT_OF_MEMORY;
@ -332,6 +336,10 @@ static int32_t setNodeSlotId(SPhysiPlanContext* pCxt, int16_t leftDataBlockId, i
static int32_t setListSlotId(SPhysiPlanContext* pCxt, int16_t leftDataBlockId, int16_t rightDataBlockId,
const SNodeList* pList, SNodeList** pOutput) {
if (NULL == pList) {
return TSDB_CODE_SUCCESS;
}
SNodeList* pRes = nodesCloneList(pList);
if (NULL == pRes) {
return TSDB_CODE_OUT_OF_MEMORY;
@ -1368,9 +1376,15 @@ static int32_t createFillPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren
pFill->inputTsOrder = pFillNode->inputTsOrder;
SDataBlockDescNode* pChildTupe = (((SPhysiNode*)nodesListGetNode(pChildren, 0))->pOutputDataBlockDesc);
int32_t code = setListSlotId(pCxt, pChildTupe->dataBlockId, -1, pFillNode->node.pTargets, &pFill->pTargets);
int32_t code = setListSlotId(pCxt, pChildTupe->dataBlockId, -1, pFillNode->pFillExprs, &pFill->pFillExprs);
if (TSDB_CODE_SUCCESS == code) {
code = addDataBlockSlots(pCxt, pFill->pTargets, pFill->node.pOutputDataBlockDesc);
code = addDataBlockSlots(pCxt, pFill->pFillExprs, pFill->node.pOutputDataBlockDesc);
}
if (TSDB_CODE_SUCCESS == code) {
code = setListSlotId(pCxt, pChildTupe->dataBlockId, -1, pFillNode->pNotFillExprs, &pFill->pNotFillExprs);
}
if (TSDB_CODE_SUCCESS == code) {
code = addDataBlockSlots(pCxt, pFill->pNotFillExprs, pFill->node.pOutputDataBlockDesc);
}
if (TSDB_CODE_SUCCESS == code) {

5
source/libs/planner/test/planIntervalTest.cpp
View File

@ -45,8 +45,9 @@ TEST_F(PlanIntervalTest, fill) {
"WHERE ts > TIMESTAMP '2022-04-01 00:00:00' and ts < TIMESTAMP '2022-04-30 23:59:59' "
"INTERVAL(10s) FILL(VALUE, 10, 20)");
run("SELECT COUNT(*) FROM st1 WHERE ts > TIMESTAMP '2022-04-01 00:00:00' and ts < TIMESTAMP '2022-04-30 23:59:59' "
"PARTITION BY TBNAME interval(10s) fill(prev)");
run("SELECT _WSTART, TBNAME, COUNT(*) FROM st1 "
"WHERE ts > '2022-04-01 00:00:00' and ts < '2022-04-30 23:59:59' "
"PARTITION BY TBNAME INTERVAL(10s) FILL(PREV)");
}
TEST_F(PlanIntervalTest, selectFunc) {

1
source/util/src/terror.c
View File

@ -463,7 +463,6 @@ TAOS_DEFINE_ERROR(TSDB_CODE_FS_APP_ERROR, "tfs out of memory")
TAOS_DEFINE_ERROR(TSDB_CODE_CTG_INTERNAL_ERROR, "catalog internal error")
TAOS_DEFINE_ERROR(TSDB_CODE_CTG_INVALID_INPUT, "invalid catalog input parameters")
TAOS_DEFINE_ERROR(TSDB_CODE_CTG_NOT_READY, "catalog is not ready")
TAOS_DEFINE_ERROR(TSDB_CODE_OUT_OF_MEMORY, "catalog memory error")
TAOS_DEFINE_ERROR(TSDB_CODE_CTG_SYS_ERROR, "catalog system error")
TAOS_DEFINE_ERROR(TSDB_CODE_CTG_DB_DROPPED, "Database is dropped")
TAOS_DEFINE_ERROR(TSDB_CODE_CTG_OUT_OF_SERVICE, "catalog is out of service")