From 2c6dec89d2215a04b2ce605cc701258bc7d9fcdc Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Wed, 6 Apr 2022 11:32:21 +0800 Subject: [PATCH] feature/qnode --- include/common/tmsg.h | 1 + include/libs/scalar/filter.h | 2 +- source/libs/command/inc/commandInt.h | 1 + source/libs/command/src/explain.c | 57 +++++++++++++++++-- source/libs/executor/inc/executorimpl.h | 44 +++++++------- source/libs/executor/src/executorimpl.c | 24 ++++++-- source/libs/scalar/src/filter.c | 3 +- .../libs/scalar/test/filter/filterTests.cpp | 22 +++---- source/libs/scheduler/src/scheduler.c | 3 + 9 files changed, 114 insertions(+), 43 deletions(-) diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 27b9a11625..9e04b5a618 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -919,6 +919,7 @@ typedef struct SExplainExecInfo { uint64_t startupCost; uint64_t totalCost; uint64_t numOfRows; + void *verboseInfo; } SExplainExecInfo; typedef struct { diff --git a/include/libs/scalar/filter.h b/include/libs/scalar/filter.h index a93180800e..c81cb49b64 100644 --- a/include/libs/scalar/filter.h +++ b/include/libs/scalar/filter.h @@ -40,7 +40,7 @@ extern int32_t filterInitFromNode(SNode *pNode, SFilterInfo **pinfo, uint32_t op extern bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, int8_t** p, SColumnDataAgg *statis, int16_t numOfCols); extern int32_t filterSetDataFromSlotId(SFilterInfo *info, void *param); extern int32_t filterSetDataFromColId(SFilterInfo *info, void *param); -extern int32_t filterGetTimeRange(SFilterInfo *info, STimeWindow *win); +extern int32_t filterGetTimeRange(SNode *pNode, STimeWindow *win, bool *isStrict); extern int32_t filterConverNcharColumns(SFilterInfo* pFilterInfo, int32_t rows, bool *gotNchar); extern int32_t filterFreeNcharColumns(SFilterInfo* pFilterInfo); extern void filterFreeInfo(SFilterInfo *info); diff --git a/source/libs/command/inc/commandInt.h b/source/libs/command/inc/commandInt.h index 7e9ea12ea4..23e4f2b24f 100644 --- a/source/libs/command/inc/commandInt.h +++ b/source/libs/command/inc/commandInt.h @@ -58,6 +58,7 @@ extern "C" { #define EXPLAIN_LOOPS_FORMAT "loops=%d" #define EXPLAIN_REVERSE_FORMAT "reverse=%d" #define EXPLAIN_FUNCTIONS_FORMAT "functions=%d" +#define EXPLAIN_EXECINFO_FORMAT "cost=%" PRIu64 "..%" PRIu64 " rows=%" PRIu64 typedef struct SExplainGroup { int32_t nodeNum; diff --git a/source/libs/command/src/explain.c b/source/libs/command/src/explain.c index c8502388ba..2a36113749 100644 --- a/source/libs/command/src/explain.c +++ b/source/libs/command/src/explain.c @@ -228,7 +228,10 @@ int32_t qExplainGenerateResNode(SPhysiNode *pNode, SExplainGroup *group, SExplai int32_t code = 0; resNode->pNode = pNode; - QRY_ERR_JRET(qExplainGenerateResNodeExecInfo(&resNode->pExecInfo, group)); + + if (group->nodeExecInfo) { + QRY_ERR_JRET(qExplainGenerateResNodeExecInfo(&resNode->pExecInfo, group)); + } QRY_ERR_JRET(qExplainGenerateResChildren(pNode, group, &resNode->pChildren)); @@ -247,14 +250,52 @@ _return: int32_t qExplainBufAppendExecInfo(SArray *pExecInfo, char *tbuf, int32_t *len) { int32_t tlen = *len; + int32_t nodeNum = taosArrayGetSize(pExecInfo); + SExplainExecInfo maxExecInfo = {0}; - EXPLAIN_ROW_APPEND("(exec info here)"); + for (int32_t i = 0; i < nodeNum; ++i) { + SExplainExecInfo *execInfo = taosArrayGet(pExecInfo, i); + if (execInfo->startupCost > maxExecInfo.startupCost) { + maxExecInfo.startupCost = execInfo->startupCost; + } + if (execInfo->totalCost > maxExecInfo.totalCost) { + maxExecInfo.totalCost = execInfo->totalCost; + } + if (execInfo->numOfRows > maxExecInfo.numOfRows) { + maxExecInfo.numOfRows = execInfo->numOfRows; + } + } + EXPLAIN_ROW_APPEND(EXPLAIN_EXECINFO_FORMAT, maxExecInfo.startupCost, maxExecInfo.totalCost, maxExecInfo.numOfRows); + *len = tlen; return TSDB_CODE_SUCCESS; } +int32_t qExplainBufAppendVerboseExecInfo(SArray *pExecInfo, char *tbuf, int32_t *len) { + int32_t tlen = 0; + bool gotVerbose = false; + int32_t nodeNum = taosArrayGetSize(pExecInfo); + SExplainExecInfo maxExecInfo = {0}; + + for (int32_t i = 0; i < nodeNum; ++i) { + SExplainExecInfo *execInfo = taosArrayGet(pExecInfo, i); + if (execInfo->verboseInfo) { + gotVerbose = true; + } + } + + if (gotVerbose) { + EXPLAIN_ROW_APPEND("exec verbose info"); + } + + *len = tlen; + + return TSDB_CODE_SUCCESS; +} + + int32_t qExplainResAppendRow(SExplainCtx *ctx, char *tbuf, int32_t len, int32_t level) { SQueryExplainRowInfo row = {0}; row.buf = taosMemoryMalloc(len); @@ -322,6 +363,14 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i EXPLAIN_ROW_NEW(level + 1, EXPLAIN_ORDER_FORMAT, EXPLAIN_ORDER_STRING(pTagScanNode->order)); EXPLAIN_ROW_END(); QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1)); + + if (pResNode->pExecInfo) { + QRY_ERR_RET(qExplainBufAppendVerboseExecInfo(pResNode->pExecInfo, tbuf, &tlen)); + if (tlen) { + EXPLAIN_ROW_END(); + QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1)); + } + } } break; } @@ -532,8 +581,8 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i EXPLAIN_ROW_APPEND(EXPLAIN_LEFT_PARENTHESIS_FORMAT); if (pResNode->pExecInfo) { QRY_ERR_RET(qExplainBufAppendExecInfo(pResNode->pExecInfo, tbuf, &tlen)); + EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT); } - EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT); EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pExchNode->node.pOutputDataBlockDesc->totalRowSize); EXPLAIN_ROW_APPEND(EXPLAIN_RIGHT_PARENTHESIS_FORMAT); EXPLAIN_ROW_END(); @@ -710,7 +759,7 @@ int32_t qExplainAppendGroupResRows(void *pCtx, int32_t groupId, int32_t level) { QRY_ERR_RET(qExplainGenerateResNode(group->plan->pNode, group, &node)); - if (group->physiPlanNum != group->physiPlanExecNum) { + if ((EXPLAIN_MODE_ANALYZE == ctx->mode) && (group->physiPlanNum != group->physiPlanExecNum)) { qError("physiPlanNum %d mismatch with physiExecNum %d in group %d", group->physiPlanNum, group->physiPlanExecNum, groupId); QRY_ERR_JRET(TSDB_CODE_QRY_APP_ERROR); } diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index f73ef1af38..2f667582e5 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -237,6 +237,7 @@ typedef bool (*__optr_decode_fn_t)(struct SOperatorInfo* pOperator, char *result typedef int32_t (*__optr_open_fn_t)(struct SOperatorInfo* pOptr); typedef SSDataBlock* (*__optr_fn_t)(struct SOperatorInfo* pOptr, bool* newgroup); typedef void (*__optr_close_fn_t)(void* param, int32_t num); +typedef int32_t (*__optr_get_explain_fn_t)(struct SOperatorInfo* pOptr, void **pOptrExplain); typedef struct STaskIdInfo { uint64_t queryId; // this is also a request id @@ -305,26 +306,27 @@ enum { }; typedef struct SOperatorInfo { - uint8_t operatorType; - bool blockingOptr; // block operator or not - uint8_t status; // denote if current operator is completed - int32_t numOfOutput; // number of columns of the current operator results - char* name; // name, used to show the query execution plan - void* info; // extension attribution - SExprInfo* pExpr; - STaskRuntimeEnv* pRuntimeEnv; // todo remove it - SExecTaskInfo* pTaskInfo; - SOperatorCostInfo cost; - SResultInfo resultInfo; - struct SOperatorInfo** pDownstream; // downstram pointer list - int32_t numOfDownstream; // number of downstream. The value is always ONE expect for join operator - __optr_open_fn_t _openFn; // DO NOT invoke this function directly - __optr_fn_t getNextFn; - __optr_fn_t getStreamResFn; // execute the aggregate in the stream model. - __optr_fn_t cleanupFn; // call this function to release the allocated resources ASAP - __optr_close_fn_t closeFn; - __optr_encode_fn_t encodeResultRow; - __optr_decode_fn_t decodeResultRow; + uint8_t operatorType; + bool blockingOptr; // block operator or not + uint8_t status; // denote if current operator is completed + int32_t numOfOutput; // number of columns of the current operator results + char* name; // name, used to show the query execution plan + void* info; // extension attribution + SExprInfo* pExpr; + STaskRuntimeEnv* pRuntimeEnv; // todo remove it + SExecTaskInfo* pTaskInfo; + SOperatorCostInfo cost; + SResultInfo resultInfo; + struct SOperatorInfo** pDownstream; // downstram pointer list + int32_t numOfDownstream; // number of downstream. The value is always ONE expect for join operator + __optr_open_fn_t _openFn; // DO NOT invoke this function directly + __optr_fn_t getNextFn; + __optr_fn_t getStreamResFn; // execute the aggregate in the stream model. + __optr_fn_t cleanupFn; // call this function to release the allocated resources ASAP + __optr_close_fn_t closeFn; + __optr_encode_fn_t encodeResultRow; + __optr_decode_fn_t decodeResultRow; + __optr_get_explain_fn_t getExplainFn; } SOperatorInfo; typedef struct { @@ -715,7 +717,7 @@ int32_t getMaximumIdleDurationSec(); void doInvokeUdf(struct SUdfInfo* pUdfInfo, SqlFunctionCtx* pCtx, int32_t idx, int32_t type); void setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status); int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId, EOPTR_EXEC_MODEL model); -int32_t getOperatorExplainExecInfo(SOperatorInfo *operator, SExplainExecInfo **pRes, int32_t *capacity, int32_t *resNum); +int32_t getOperatorExplainExecInfo(SOperatorInfo *operatorInfo, SExplainExecInfo **pRes, int32_t *capacity, int32_t *resNum); #ifdef __cplusplus } diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 498426d1a4..729d9efd86 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -5585,6 +5585,10 @@ SOperatorInfo* createTableScanOperatorInfo(void* pTsdbReadHandle, int32_t order, pOperator->getNextFn = doTableScan; pOperator->pTaskInfo = pTaskInfo; + //pOperator->cost.openCost = 1314; + //pOperator->cost.totalCost = 2324; + //pOperator->resultInfo.totalRows = 3334; + return pOperator; } @@ -9517,7 +9521,7 @@ void releaseQueryBuf(size_t numOfTables) { atomic_add_fetch_64(&tsQueryBufferSizeBytes, t); } -int32_t getOperatorExplainExecInfo(SOperatorInfo *operator, SExplainExecInfo **pRes, int32_t *capacity, int32_t *resNum) { +int32_t getOperatorExplainExecInfo(SOperatorInfo *operatorInfo, SExplainExecInfo **pRes, int32_t *capacity, int32_t *resNum) { if (*resNum >= *capacity) { *capacity += 10; @@ -9528,15 +9532,23 @@ int32_t getOperatorExplainExecInfo(SOperatorInfo *operator, SExplainExecInfo **p } } - (*pRes)[*resNum].numOfRows = operator->resultInfo.totalRows; - (*pRes)[*resNum].startupCost = operator->cost.openCost; - (*pRes)[*resNum].totalCost = operator->cost.totalCost; + (*pRes)[*resNum].numOfRows = operatorInfo->resultInfo.totalRows; + (*pRes)[*resNum].startupCost = operatorInfo->cost.openCost; + (*pRes)[*resNum].totalCost = operatorInfo->cost.totalCost; + + if (operatorInfo->getExplainFn) { + int32_t code = (*operatorInfo->getExplainFn)(operatorInfo, &(*pRes)->verboseInfo); + if (code) { + qError("operator getExplainFn failed, error:%s", tstrerror(code)); + return code; + } + } ++(*resNum); int32_t code = 0; - for (int32_t i = 0; i < operator->numOfDownstream; ++i) { - code = getOperatorExplainExecInfo(operator->pDownstream[i], pRes, capacity, resNum); + for (int32_t i = 0; i < operatorInfo->numOfDownstream; ++i) { + code = getOperatorExplainExecInfo(operatorInfo->pDownstream[i], pRes, capacity, resNum); if (code) { taosMemoryFreeClear(*pRes); return TSDB_CODE_QRY_OUT_OF_MEMORY; diff --git a/source/libs/scalar/src/filter.c b/source/libs/scalar/src/filter.c index b0632bbc34..0502774d1b 100644 --- a/source/libs/scalar/src/filter.c +++ b/source/libs/scalar/src/filter.c @@ -3314,7 +3314,8 @@ bool filterRangeExecute(SFilterInfo *info, SColumnDataAgg *pDataStatis, int32_t -int32_t filterGetTimeRange(SFilterInfo *info, STimeWindow *win) { +int32_t filterGetTimeRange(SNode *pNode, STimeWindow *win, bool *isStrict) { + SFilterInfo *info = NULL; SFilterRange ra = {0}; SFilterRangeCtx *prev = filterInitRangeCtx(TSDB_DATA_TYPE_TIMESTAMP, FLT_OPTION_TIMESTAMP); SFilterRangeCtx *tmpc = filterInitRangeCtx(TSDB_DATA_TYPE_TIMESTAMP, FLT_OPTION_TIMESTAMP); diff --git a/source/libs/scalar/test/filter/filterTests.cpp b/source/libs/scalar/test/filter/filterTests.cpp index 54f82eae2d..696a697a91 100644 --- a/source/libs/scalar/test/filter/filterTests.cpp +++ b/source/libs/scalar/test/filter/filterTests.cpp @@ -234,15 +234,16 @@ TEST(timerangeTest, greater) { flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tsmall); flttMakeOpNode(&opNode1, OP_TYPE_GREATER_THAN, TSDB_DATA_TYPE_BOOL, pcol, pval); - SFilterInfo *filter = NULL; - int32_t code = filterInitFromNode(opNode1, &filter, FLT_OPTION_NO_REWRITE|FLT_OPTION_TIMESTAMP); - ASSERT_EQ(code, 0); + //SFilterInfo *filter = NULL; + //int32_t code = filterInitFromNode(opNode1, &filter, FLT_OPTION_NO_REWRITE|FLT_OPTION_TIMESTAMP); + //ASSERT_EQ(code, 0); STimeWindow win = {0}; - code = filterGetTimeRange(filter, &win); + bool isStrict = false; + int32_t code = filterGetTimeRange(opNode1, &win, &isStrict); ASSERT_EQ(code, 0); ASSERT_EQ(win.skey, tsmall); ASSERT_EQ(win.ekey, INT64_MAX); - filterFreeInfo(filter); + //filterFreeInfo(filter); nodesDestroyNode(opNode1); } @@ -263,15 +264,16 @@ TEST(timerangeTest, greater_and_lower) { flttMakeLogicNode(&logicNode, LOGIC_COND_TYPE_AND, list, 2); - SFilterInfo *filter = NULL; - int32_t code = filterInitFromNode(logicNode, &filter, FLT_OPTION_NO_REWRITE|FLT_OPTION_TIMESTAMP); - ASSERT_EQ(code, 0); + //SFilterInfo *filter = NULL; + //int32_t code = filterInitFromNode(logicNode, &filter, FLT_OPTION_NO_REWRITE|FLT_OPTION_TIMESTAMP); + //ASSERT_EQ(code, 0); STimeWindow win = {0}; - code = filterGetTimeRange(filter, &win); + bool isStrict = false; + int32_t code = filterGetTimeRange(logicNode, &win, &isStrict); ASSERT_EQ(code, 0); ASSERT_EQ(win.skey, tsmall); ASSERT_EQ(win.ekey, tbig); - filterFreeInfo(filter); + //filterFreeInfo(filter); nodesDestroyNode(logicNode); } diff --git a/source/libs/scheduler/src/scheduler.c b/source/libs/scheduler/src/scheduler.c index 55eaae1b54..1675f91330 100644 --- a/source/libs/scheduler/src/scheduler.c +++ b/source/libs/scheduler/src/scheduler.c @@ -199,6 +199,7 @@ int32_t schValidateTaskReceivedMsgType(SSchJob *pJob, SSchTask *pTask, int32_t m int32_t reqMsgType = msgType - 1; switch (msgType) { case TDMT_SCH_LINK_BROKEN: + case TDMT_VND_EXPLAIN_RSP: return TSDB_CODE_SUCCESS; case TDMT_VND_QUERY_RSP: // query_rsp may be processed later than ready_rsp if (lastMsgType != reqMsgType && -1 != lastMsgType && TDMT_VND_FETCH != lastMsgType) { @@ -1187,6 +1188,8 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch return TSDB_CODE_SUCCESS; } + atomic_val_compare_exchange_32(&pJob->remoteFetch, 1, 0); + SCH_ERR_JRET(schFetchFromRemote(pJob)); return TSDB_CODE_SUCCESS; -- GitLab