Skip to content

TDengine
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
TDengine
TDengine
提交 6f26d177 编写于 作者: H Haojun Liao
浏览文件
操作

Merge remote-tracking branch 'origin/3.0' into feature/3.0_liaohj

上级 21b94f4c af32cfca
展开全部 隐藏空白更改
内联 并排
Showing with 1824 addition and 710 deletion
include/common/tmsg.h
浏览文件 @ 6f26d177
......@@ -930,6 +930,21 @@ typedef struct {
char data[];
} SRetrieveMetaTableRsp;
typedef struct SExplainExecInfo {
uint64_t startupCost;
uint64_t totalCost;
uint64_t numOfRows;
void *verboseInfo;
} SExplainExecInfo;
typedef struct {
int32_t numOfPlans;
SExplainExecInfo *subplanInfo;
} SExplainRsp;
int32_t tSerializeSExplainRsp(void* buf, int32_t bufLen, SExplainRsp* pRsp);
int32_t tDeserializeSExplainRsp(void* buf, int32_t bufLen, SExplainRsp* pRsp);
typedef struct {
char fqdn[TSDB_FQDN_LEN]; // end point, hostname:port
int32_t port;
......@@ -1067,6 +1082,7 @@ typedef struct SSubQueryMsg {
uint64_t taskId;
int64_t refId;
int8_t taskType;
int8_t explain;
uint32_t sqlLen; // the query sql,
uint32_t phyLen;
char msg[];
......
include/common/tmsgdef.h
浏览文件 @ 6f26d177
......@@ -188,6 +188,7 @@ enum {
TD_DEF_MSG_TYPE(TDMT_VND_SHOW_TABLES_FETCH, "vnode-show-tables-fetch", SVShowTablesFetchReq, SVShowTablesFetchRsp)
TD_DEF_MSG_TYPE(TDMT_VND_QUERY_CONTINUE, "vnode-query-continue", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_QUERY_HEARTBEAT, "vnode-query-heartbeat", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_EXPLAIN, "vnode-explain", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_SUBSCRIBE, "vnode-subscribe", SMVSubscribeReq, SMVSubscribeRsp)
TD_DEF_MSG_TYPE(TDMT_VND_CONSUME, "vnode-consume", SMqCVConsumeReq, SMqCVConsumeRsp)
......
include/common/ttypes.h
浏览文件 @ 6f26d177
......@@ -146,6 +146,7 @@ typedef struct {
#define IS_FLOAT_TYPE(_t) ((_t) == TSDB_DATA_TYPE_FLOAT || (_t) == TSDB_DATA_TYPE_DOUBLE)
#define IS_NUMERIC_TYPE(_t) ((IS_SIGNED_NUMERIC_TYPE(_t)) || (IS_UNSIGNED_NUMERIC_TYPE(_t)) || (IS_FLOAT_TYPE(_t)))
#define IS_MATHABLE_TYPE(_t) (IS_NUMERIC_TYPE(_t) || (_t) == (TSDB_DATA_TYPE_BOOL) || (_t) == (TSDB_DATA_TYPE_TIMESTAMP))
#define IS_VALID_TINYINT(_t) ((_t) > INT8_MIN && (_t) <= INT8_MAX)
#define IS_VALID_SMALLINT(_t) ((_t) > INT16_MIN && (_t) <= INT16_MAX)
......
include/common/tvariant.h
浏览文件 @ 6f26d177
......@@ -59,6 +59,7 @@ int32_t taosVariantDumpEx(SVariant *pVariant, char *payload, int16_t type, bool
#endif
int32_t taosVariantTypeSetType(SVariant *pVariant, char type);
char * taosVariantGet(SVariant *pVar, int32_t type);
#ifdef __cplusplus
}
......
include/libs/command/command.h
浏览文件 @ 6f26d177
......@@ -17,8 +17,14 @@
#include "tmsg.h"
#include "plannodes.h"
typedef struct SExplainCtx SExplainCtx;
int32_t qExecCommand(SNode* pStmt, SRetrieveTableRsp** pRsp);
int32_t qExecStaticExplain(SQueryPlan *pDag, SRetrieveTableRsp **pRsp);
int32_t qExecExplainBegin(SQueryPlan *pDag, SExplainCtx **pCtx, int32_t startTs);
int32_t qExecExplainEnd(SExplainCtx *pCtx, SRetrieveTableRsp **pRsp);
int32_t qExplainUpdateExecInfo(SExplainCtx *pCtx, SExplainRsp *pRspMsg, int32_t groupId, SRetrieveTableRsp **pRsp);
void qExplainFreeCtx(SExplainCtx *pCtx);
include/libs/executor/executor.h
浏览文件 @ 6f26d177
......@@ -174,6 +174,8 @@ void** qReleaseTask(void* pMgmt, void* pQInfo, bool freeHandle);
void qProcessFetchRsp(void* parent, struct SRpcMsg* pMsg, struct SEpSet* pEpSet);
int32_t qGetExplainExecInfo(qTaskInfo_t tinfo, int32_t *resNum, SExplainExecInfo **pRes);
#ifdef __cplusplus
}
#endif
......
include/libs/nodes/nodes.h
浏览文件 @ 6f26d177
......@@ -163,6 +163,8 @@ typedef enum ENodeType {
QUERY_NODE_PHYSICAL_PLAN_SORT,
QUERY_NODE_PHYSICAL_PLAN_INTERVAL,
QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW,
QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW,
QUERY_NODE_PHYSICAL_PLAN_PARTITION,
QUERY_NODE_PHYSICAL_PLAN_DISPATCH,
QUERY_NODE_PHYSICAL_PLAN_INSERT,
QUERY_NODE_PHYSICAL_SUBPLAN,
......@@ -237,6 +239,8 @@ int32_t nodesListToString(const SNodeList* pList, bool format, char** pStr, int3
int32_t nodesStringToList(const char* pStr, SNodeList** pList);
int32_t nodesNodeToSQL(SNode *pNode, char *buf, int32_t bufSize, int32_t *len);
char *nodesGetNameFromColumnNode(SNode *pNode);
int32_t nodesGetOutputNumFromSlotList(SNodeList* pSlots);
#ifdef __cplusplus
}
......
include/libs/nodes/plannodes.h
浏览文件 @ 6f26d177
......@@ -104,6 +104,7 @@ typedef struct SWindowLogicNode {
SFillNode* pFill;
int64_t sessionGap;
SNode* pTspk;
SNode* pStateExpr;
} SWindowLogicNode;
typedef struct SSortLogicNode {
......@@ -194,11 +195,20 @@ typedef struct SSystemTableScanPhysiNode {
int32_t accountId;
} SSystemTableScanPhysiNode;
typedef enum EScanRequired {
SCAN_REQUIRED_DATA_NO_NEEDED = 1,
SCAN_REQUIRED_DATA_STATIS_NEEDED,
SCAN_REQUIRED_DATA_ALL_NEEDED,
SCAN_REQUIRED_DATA_DISCARD,
} EScanRequired;
typedef struct STableScanPhysiNode {
SScanPhysiNode scan;
uint8_t scanFlag; // denotes reversed scan of data or not
STimeWindow scanRange;
double ratio;
EScanRequired scanRequired;
SNodeList* pScanReferFuncs;
} STableScanPhysiNode;
typedef STableScanPhysiNode STableSeqScanPhysiNode;
......@@ -257,17 +267,33 @@ typedef struct SIntervalPhysiNode {
SFillNode* pFill;
} SIntervalPhysiNode;
typedef struct SMultiTableIntervalPhysiNode {
SIntervalPhysiNode interval;
SNodeList* pPartitionKeys;
} SMultiTableIntervalPhysiNode;
typedef struct SSessionWinodwPhysiNode {
SWinodwPhysiNode window;
int64_t gap;
} SSessionWinodwPhysiNode;
typedef struct SStateWinodwPhysiNode {
SWinodwPhysiNode window;
SNode* pStateKey;
} SStateWinodwPhysiNode;
typedef struct SSortPhysiNode {
SPhysiNode node;
SNodeList* pExprs; // these are expression list of order_by_clause and parameter expression of aggregate function
SNodeList* pSortKeys; // element is SOrderByExprNode, and SOrderByExprNode::pExpr is SColumnNode
} SSortPhysiNode;
typedef struct SPartitionPhysiNode {
SPhysiNode node;
SNodeList* pExprs; // these are expression list of partition_by_clause
SNodeList* pPartitionKeys;
} SPartitionPhysiNode;
typedef struct SDataSinkNode {
ENodeType type;
SDataBlockDescNode* pInputDataBlockDesc;
......@@ -308,6 +334,7 @@ typedef enum EExplainMode {
typedef struct SExplainInfo {
EExplainMode mode;
bool verbose;
double ratio;
} SExplainInfo;
typedef struct SQueryPlan {
......
include/libs/nodes/querynodes.h
浏览文件 @ 6f26d177
......@@ -22,6 +22,7 @@ extern "C" {
#include "nodes.h"
#include "tmsg.h"
#include "tvariant.h"
#define TABLE_TOTAL_COL_NUM(pMeta) ((pMeta)->tableInfo.numOfColumns + (pMeta)->tableInfo.numOfTags)
#define TABLE_META_SIZE(pMeta) (NULL == (pMeta) ? 0 : (sizeof(STableMeta) + TABLE_TOTAL_COL_NUM((pMeta)) * sizeof(SSchema)))
......@@ -188,7 +189,7 @@ typedef struct SLimitNode {
typedef struct SStateWindowNode {
ENodeType type; // QUERY_NODE_STATE_WINDOW
SNode* pCol;
SNode* pExpr;
} SStateWindowNode;
typedef struct SSessionWindowNode {
......@@ -316,6 +317,8 @@ bool nodesIsTimelineQuery(const SNode* pQuery);
void* nodesGetValueFromNode(SValueNode *pNode);
char* nodesGetStrValueFromNode(SValueNode *pNode);
char *getFillModeString(EFillMode mode);
void valueNodeToVariant(const SValueNode* pNode, SVariant* pVal);
#ifdef __cplusplus
}
......
include/libs/scalar/filter.h
浏览文件 @ 6f26d177
......@@ -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);
......
include/libs/scheduler/scheduler.h
浏览文件 @ 6f26d177
......@@ -71,7 +71,7 @@ int32_t schedulerInit(SSchedulerCfg *cfg);
* @param nodeList Qnode/Vnode address list, element is SQueryNodeAddr
* @return
*/
int32_t schedulerExecJob(void *transport, SArray *nodeList, SQueryPlan* pDag, int64_t* pJob, const char* sql, SQueryResult *pRes);
int32_t schedulerExecJob(void *transport, SArray *nodeList, SQueryPlan *pDag, int64_t *pJob, const char *sql, int64_t startTs, SQueryResult *pRes);
/**
* Process the query job, generated according to the query physical plan.
......
include/util/taoserror.h
浏览文件 @ 6f26d177
......@@ -483,6 +483,9 @@ int32_t* taosGetErrno();
#define TSDB_CODE_PAR_INVALID_IDENTIFIER_NAME TAOS_DEF_ERROR_CODE(0, 0x2617)
#define TSDB_CODE_PAR_CORRESPONDING_STABLE_ERR TAOS_DEF_ERROR_CODE(0, 0x2618)
//planner
#define TSDB_CODE_PLAN_INTERNAL_ERROR TAOS_DEF_ERROR_CODE(0, 0x2700)
#ifdef __cplusplus
}
#endif
......
include/util/tlockfree.h
浏览文件 @ 6f26d177
......@@ -76,6 +76,7 @@ void taosWLockLatch(SRWLatch *pLatch);
void taosWUnLockLatch(SRWLatch *pLatch);
void taosRLockLatch(SRWLatch *pLatch);
void taosRUnLockLatch(SRWLatch *pLatch);
int32_t taosWTryLockLatch(SRWLatch *pLatch);
// copy on read
#define taosCorBeginRead(x) \
......
source/client/src/clientImpl.c
浏览文件 @ 6f26d177
......@@ -254,7 +254,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
void* pTransporter = pRequest->pTscObj->pAppInfo->pTransporter;
SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf};
int32_t code = schedulerExecJob(pTransporter, pNodeList, pDag, &pRequest->body.queryJob, pRequest->sqlstr, &res);
int32_t code = schedulerExecJob(pTransporter, pNodeList, pDag, &pRequest->body.queryJob, pRequest->sqlstr, pRequest->metric.start, &res);
if (code != TSDB_CODE_SUCCESS) {
if (pRequest->body.queryJob != 0) {
schedulerFreeJob(pRequest->body.queryJob);
......
source/common/src/tmsg.c
浏览文件 @ 6f26d177
......@@ -2769,6 +2769,48 @@ int32_t tDecodeSMqCMCommitOffsetReq(SCoder *decoder, SMqCMCommitOffsetReq *pReq)
return 0;
}
int32_t tSerializeSExplainRsp(void* buf, int32_t bufLen, SExplainRsp* pRsp) {
SCoder encoder = {0};
tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeI32(&encoder, pRsp->numOfPlans) < 0) return -1;
for (int32_t i = 0; i < pRsp->numOfPlans; ++i) {
SExplainExecInfo *info = &pRsp->subplanInfo[i];
if (tEncodeU64(&encoder, info->startupCost) < 0) return -1;
if (tEncodeU64(&encoder, info->totalCost) < 0) return -1;
if (tEncodeU64(&encoder, info->numOfRows) < 0) return -1;
}
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tCoderClear(&encoder);
return tlen;
}
int32_t tDeserializeSExplainRsp(void* buf, int32_t bufLen, SExplainRsp* pRsp) {
SCoder decoder = {0};
tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER);
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeI32(&decoder, &pRsp->numOfPlans) < 0) return -1;
if (pRsp->numOfPlans > 0) {
pRsp->subplanInfo = taosMemoryMalloc(pRsp->numOfPlans * sizeof(SExplainExecInfo));
if (pRsp->subplanInfo == NULL) return -1;
}
for (int32_t i = 0; i < pRsp->numOfPlans; ++i) {
if (tDecodeU64(&decoder, &pRsp->subplanInfo[i].startupCost) < 0) return -1;
if (tDecodeU64(&decoder, &pRsp->subplanInfo[i].totalCost) < 0) return -1;
if (tDecodeU64(&decoder, &pRsp->subplanInfo[i].numOfRows) < 0) return -1;
}
tEndDecode(&decoder);
tCoderClear(&decoder);
return 0;
}
int32_t tSerializeSSchedulerHbReq(void *buf, int32_t bufLen, SSchedulerHbReq *pReq) {
int32_t headLen = sizeof(SMsgHead);
if (buf != NULL) {
......
source/common/src/tvariant.c
浏览文件 @ 6f26d177
......@@ -1014,4 +1014,28 @@ int32_t taosVariantTypeSetType(SVariant *pVariant, char type) {
}
return 0;
}
\ No newline at end of file
}
char * taosVariantGet(SVariant *pVar, int32_t type) {
switch (type) {
case TSDB_DATA_TYPE_BOOL:
case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_INT:
case TSDB_DATA_TYPE_TIMESTAMP:
return (char *)&pVar->i;
case TSDB_DATA_TYPE_DOUBLE:
case TSDB_DATA_TYPE_FLOAT:
return (char *)&pVar->d;
case TSDB_DATA_TYPE_BINARY:
return (char *)pVar->pz;
case TSDB_DATA_TYPE_NCHAR:
return (char *)pVar->ucs4;
default:
return NULL;
}
return NULL;
}
source/libs/command/inc/commandInt.h
浏览文件 @ 6f26d177
......@@ -26,38 +26,54 @@ extern "C" {
#define EXPLAIN_MAX_GROUP_NUM 100
//newline area
#define EXPLAIN_TAG_SCAN_FORMAT "Tag Scan on %s columns=%d width=%d"
#define EXPLAIN_TBL_SCAN_FORMAT "Table Scan on %s columns=%d width=%d"
#define EXPLAIN_SYSTBL_SCAN_FORMAT "System Table Scan on %s columns=%d width=%d"
#define EXPLAIN_PROJECTION_FORMAT "Projection columns=%d width=%d"
#define EXPLAIN_JOIN_FORMAT "%s between %d tables width=%d"
#define EXPLAIN_AGG_FORMAT "Aggragate functions=%d"
#define EXPLAIN_EXCHANGE_FORMAT "Data Exchange %d:1 width=%d"
#define EXPLAIN_SORT_FORMAT "Sort on %d Column(s) width=%d"
#define EXPLAIN_INTERVAL_FORMAT "Interval on Column %s functions=%d interval=%" PRId64 "%c offset=%" PRId64 "%c sliding=%" PRId64 "%c width=%d"
#define EXPLAIN_SESSION_FORMAT "Session gap=%" PRId64 " functions=%d width=%d"
#define EXPLAIN_TAG_SCAN_FORMAT "Tag Scan on %s"
#define EXPLAIN_TBL_SCAN_FORMAT "Table Scan on %s"
#define EXPLAIN_SYSTBL_SCAN_FORMAT "System Table Scan on %s"
#define EXPLAIN_PROJECTION_FORMAT "Projection"
#define EXPLAIN_JOIN_FORMAT "%s"
#define EXPLAIN_AGG_FORMAT "Aggragate"
#define EXPLAIN_EXCHANGE_FORMAT "Data Exchange %d:1"
#define EXPLAIN_SORT_FORMAT "Sort"
#define EXPLAIN_INTERVAL_FORMAT "Interval on Column %s"
#define EXPLAIN_SESSION_FORMAT "Session"
#define EXPLAIN_ORDER_FORMAT "Order: %s"
#define EXPLAIN_FILTER_FORMAT "Filter: "
#define EXPLAIN_FILL_FORMAT "Fill: %s"
#define EXPLAIN_ON_CONDITIONS_FORMAT "Join Cond: "
#define EXPLAIN_TIMERANGE_FORMAT "Time Range: [%" PRId64 ", %" PRId64 "]"
#define EXPLAIN_OUTPUT_FORMAT "Output: "
#define EXPLAIN_TIME_WINDOWS_FORMAT "Time Window: interval=%" PRId64 "%c offset=%" PRId64 "%c sliding=%" PRId64 "%c"
#define EXPLAIN_WINDOW_FORMAT "Window: gap=%" PRId64
//append area
#define EXPLAIN_GROUPS_FORMAT " groups=%d"
#define EXPLAIN_WIDTH_FORMAT " width=%d"
#define EXPLAIN_LOOPS_FORMAT " loops=%d"
#define EXPLAIN_REVERSE_FORMAT " reverse=%d"
#define EXPLAIN_LEFT_PARENTHESIS_FORMAT " ("
#define EXPLAIN_RIGHT_PARENTHESIS_FORMAT ")"
#define EXPLAIN_BLANK_FORMAT " "
#define EXPLAIN_COST_FORMAT "cost=%.2f..%.2f"
#define EXPLAIN_ROWS_FORMAT "rows=%" PRIu64
#define EXPLAIN_COLUMNS_FORMAT "columns=%d"
#define EXPLAIN_WIDTH_FORMAT "width=%d"
#define EXPLAIN_GROUPS_FORMAT "groups=%d"
#define EXPLAIN_WIDTH_FORMAT "width=%d"
#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;
int32_t physiPlanExecNum;
int32_t physiPlanNum;
int32_t physiPlanExecIdx;
SRWLatch lock;
SSubplan *plan;
void *execInfo; //TODO
SArray *nodeExecInfo; //Array<SExplainRsp>
} SExplainGroup;
typedef struct SExplainResNode {
SNodeList* pChildren;
SPhysiNode* pNode;
void* pExecInfo;
SNodeList* pChildren;
SPhysiNode* pNode;
SArray* pExecInfo; // Array<SExplainExecInfo>
} SExplainResNode;
typedef struct SQueryExplainRowInfo {
......@@ -67,11 +83,21 @@ typedef struct SQueryExplainRowInfo {
} SQueryExplainRowInfo;
typedef struct SExplainCtx {
int32_t totalSize;
bool verbose;
char *tbuf;
SArray *rows;
SHashObj *groupHash;
EExplainMode mode;
double ratio;
bool verbose;
SRWLatch lock;
int32_t rootGroupId;
int32_t dataSize;
bool execDone;
int64_t reqStartTs;
int64_t jobStartTs;
int64_t jobDoneTs;
char *tbuf;
SArray *rows;
int32_t groupDoneNum;
SHashObj *groupHash; // Hash<SExplainGroup>
} SExplainCtx;
#define EXPLAIN_ORDER_STRING(_order) ((TSDB_ORDER_ASC == _order) ? "Ascending" : "Descending")
......
source/libs/command/src/explain.c
浏览文件 @ 6f26d177
此差异已折叠。
source/libs/executor/inc/executorimpl.h
浏览文件 @ 6f26d177
......@@ -36,6 +36,7 @@ extern "C" {
#include "thash.h"
#include "tlockfree.h"
#include "tpagedbuf.h"
#include "tmsg.h"
struct SColumnFilterElem;
......@@ -165,7 +166,7 @@ typedef struct STaskCostInfo {
typedef struct SOperatorCostInfo {
uint64_t openCost;
uint64_t execCost;
uint64_t totalCost;
} SOperatorCostInfo;
typedef struct SOrder {
......@@ -238,6 +239,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
......@@ -306,26 +308,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 {
......@@ -727,6 +730,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 *operatorInfo, SExplainExecInfo **pRes, int32_t *capacity, int32_t *resNum);
#ifdef __cplusplus
}
......
source/libs/executor/src/executorMain.c
浏览文件 @ 6f26d177
......@@ -229,3 +229,12 @@ void qDestroyTask(qTaskInfo_t qTaskHandle) {
queryCostStatis(pTaskInfo); // print the query cost summary
doDestroyTask(pTaskInfo);
}
int32_t qGetExplainExecInfo(qTaskInfo_t tinfo, int32_t *resNum, SExplainExecInfo **pRes) {
SExecTaskInfo *pTaskInfo = (SExecTaskInfo *)tinfo;
int32_t capacity = 0;
return getOperatorExplainExecInfo(pTaskInfo->pRoot, pRes, &capacity, resNum);
}
source/libs/executor/src/executorimpl.c
浏览文件 @ 6f26d177
......@@ -1304,6 +1304,12 @@ static void projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSData
pResult->info.rows = dest.numOfRows;
taosArrayDestroy(pBlockList);
}
} else if (pExpr[k].pExpr->nodeType == QUERY_NODE_VALUE) {
SColumnInfoData* pColInfoData = taosArrayGet(pResult->pDataBlock, k);
for (int32_t i = 0; i < pSrcBlock->info.rows; ++i) {
colDataAppend(pColInfoData, i, taosVariantGet(&pExpr[k].base.pParam[0].param, pExpr[k].base.pParam[0].type), TSDB_DATA_TYPE_NULL == pExpr[k].base.pParam[0].param.nType);
}
pResult->info.rows = pSrcBlock->info.rows;
} else {
ASSERT(0);
}
......@@ -1935,7 +1941,7 @@ static SqlFunctionCtx* createSqlFunctionCtx_rv(SExprInfo* pExprInfo, int32_t num
}
}
pCtx->resDataInfo.interBufSize = env.calcMemSize;
} else if (pExpr->pExpr->nodeType == QUERY_NODE_COLUMN || pExpr->pExpr->nodeType == QUERY_NODE_OPERATOR) {
} else if (pExpr->pExpr->nodeType == QUERY_NODE_COLUMN || pExpr->pExpr->nodeType == QUERY_NODE_OPERATOR || pExpr->pExpr->nodeType == QUERY_NODE_VALUE) {
pCtx->resDataInfo.interBufSize = pFunct->resSchema.bytes; // for simple column, the intermediate buffer needs to hold one element.
}
......@@ -6994,6 +7000,17 @@ SExprInfo* createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, int32_t*
// pExp->base.pParam[0].type = FUNC_PARAM_TYPE_COLUMN;
// pExp->base.pParam[0].pCol = createColumn(pTargetNode->dataBlockId, pTargetNode->slotId, pType);
} else if (nodeType(pTargetNode->pExpr) == QUERY_NODE_VALUE) {
pExp->pExpr->nodeType = QUERY_NODE_VALUE;
SValueNode* pValNode = (SValueNode*)pTargetNode->pExpr;
pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
pExp->base.numOfParams = 1;
SDataType* pType = &pValNode->node.resType;
pExp->base.resSchema = createResSchema(pType->type, pType->bytes, pTargetNode->slotId, pType->scale, pType->precision, pValNode->node.aliasName);
pExp->base.pParam[0].type = FUNC_PARAM_TYPE_VALUE;
valueNodeToVariant(pValNode, &pExp->base.pParam[0].param);
} else {
ASSERT(0);
}
......@@ -7560,3 +7577,42 @@ void releaseQueryBuf(size_t numOfTables) {
// restore value is not enough buffer available
atomic_add_fetch_64(&tsQueryBufferSizeBytes, t);
}
int32_t getOperatorExplainExecInfo(SOperatorInfo *operatorInfo, SExplainExecInfo **pRes, int32_t *capacity, int32_t *resNum) {
if (*resNum >= *capacity) {
*capacity += 10;
*pRes = taosMemoryRealloc(*pRes, (*capacity) * sizeof(SExplainExecInfo));
if (NULL == *pRes) {
qError("malloc %d failed", (*capacity) * (int32_t)sizeof(SExplainExecInfo));
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
}
(*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 < operatorInfo->numOfDownstream; ++i) {
code = getOperatorExplainExecInfo(operatorInfo->pDownstream[i], pRes, capacity, resNum);
if (code) {
taosMemoryFreeClear(*pRes);
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
}
return TSDB_CODE_SUCCESS;
}
source/libs/nodes/inc/nodesUtil.h
浏览文件 @ 6f26d177
......@@ -31,6 +31,7 @@ extern "C" {
#define NODES_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)
#define NODES_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
#ifdef __cplusplus
}
#endif
......
source/libs/nodes/src/nodesCloneFuncs.c
浏览文件 @ 6f26d177
......@@ -283,6 +283,12 @@ static SNode* logicSortCopy(const SSortLogicNode* pSrc, SSortLogicNode* pDst) {
return (SNode*)pDst;
}
static SNode* logicPartitionCopy(const SPartitionLogicNode* pSrc, SPartitionLogicNode* pDst) {
COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
CLONE_NODE_LIST_FIELD(pPartitionKeys);
return (SNode*)pDst;
}
static SNode* logicSubplanCopy(const SLogicSubplan* pSrc, SLogicSubplan* pDst) {
CLONE_NODE_FIELD(pNode);
COPY_SCALAR_FIELD(subplanType);
......@@ -367,6 +373,8 @@ SNodeptr nodesCloneNode(const SNodeptr pNode) {
return logicWindowCopy((const SWindowLogicNode*)pNode, (SWindowLogicNode*)pDst);
case QUERY_NODE_LOGIC_PLAN_SORT:
return logicSortCopy((const SSortLogicNode*)pNode, (SSortLogicNode*)pDst);
case QUERY_NODE_LOGIC_PLAN_PARTITION:
return logicPartitionCopy((const SPartitionLogicNode*)pNode, (SPartitionLogicNode*)pDst);
case QUERY_NODE_LOGIC_SUBPLAN:
return logicSubplanCopy((const SLogicSubplan*)pNode, (SLogicSubplan*)pDst);
default:
......
source/libs/nodes/src/nodesCodeFuncs.c
浏览文件 @ 6f26d177
......@@ -198,6 +198,10 @@ const char* nodesNodeName(ENodeType type) {
return "PhysiInterval";
case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
return "PhysiSessionWindow";
case QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW:
return "PhysiStateWindow";
case QUERY_NODE_PHYSICAL_PLAN_PARTITION:
return "PhysiPartition";
case QUERY_NODE_PHYSICAL_PLAN_DISPATCH:
return "PhysiDispatch";
case QUERY_NODE_PHYSICAL_PLAN_INSERT:
......@@ -1147,6 +1151,61 @@ static int32_t jsonToPhysiSessionWindowNode(const SJson* pJson, void* pObj) {
return code;
}
static const char* jkStateWindowPhysiPlanStateKey = "StateKey";
static int32_t physiStateWindowNodeToJson(const void* pObj, SJson* pJson) {
const SStateWinodwPhysiNode* pNode = (const SStateWinodwPhysiNode*)pObj;
int32_t code = physiWindowNodeToJson(pObj, pJson);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkStateWindowPhysiPlanStateKey, nodeToJson, pNode->pStateKey);
}
return code;
}
static int32_t jsonToPhysiStateWindowNode(const SJson* pJson, void* pObj) {
SStateWinodwPhysiNode* pNode = (SStateWinodwPhysiNode*)pObj;
int32_t code = jsonToPhysiWindowNode(pJson, pObj);
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkStateWindowPhysiPlanStateKey, &pNode->pStateKey);
}
return code;
}
static const char* jkPartitionPhysiPlanExprs = "Exprs";
static const char* jkPartitionPhysiPlanPartitionKeys = "PartitionKeys";
static int32_t physiPartitionNodeToJson(const void* pObj, SJson* pJson) {
const SPartitionPhysiNode* pNode = (const SPartitionPhysiNode*)pObj;
int32_t code = physicPlanNodeToJson(pObj, pJson);
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkPartitionPhysiPlanExprs, pNode->pExprs);
}
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkPartitionPhysiPlanPartitionKeys, pNode->pPartitionKeys);
}
return code;
}
static int32_t jsonToPhysiPartitionNode(const SJson* pJson, void* pObj) {
SPartitionPhysiNode* pNode = (SPartitionPhysiNode*)pObj;
int32_t code = jsonToPhysicPlanNode(pJson, pObj);
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkPartitionPhysiPlanExprs, &pNode->pExprs);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkPartitionPhysiPlanPartitionKeys, &pNode->pPartitionKeys);
}
return code;
}
static const char* jkDataSinkInputDataBlockDesc = "InputDataBlockDesc";
static int32_t physicDataSinkNodeToJson(const void* pObj, SJson* pJson) {
......@@ -2420,6 +2479,10 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
return physiIntervalNodeToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
return physiSessionWindowNodeToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW:
return physiStateWindowNodeToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_PARTITION:
return physiPartitionNodeToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_DISPATCH:
return physiDispatchNodeToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_INSERT:
......@@ -2512,6 +2575,10 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
return jsonToPhysiIntervalNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
return jsonToPhysiSessionWindowNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW:
return jsonToPhysiStateWindowNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_PARTITION:
return jsonToPhysiPartitionNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_DISPATCH:
return jsonToPhysiDispatchNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_SUBPLAN:
......
source/libs/nodes/src/nodesTraverseFuncs.c
浏览文件 @ 6f26d177
......@@ -78,7 +78,7 @@ static EDealRes walkNode(SNode* pNode, ETraversalOrder order, FNodeWalker walker
res = walkNode(((SOrderByExprNode*)pNode)->pExpr, order, walker, pContext);
break;
case QUERY_NODE_STATE_WINDOW:
res = walkNode(((SStateWindowNode*)pNode)->pCol, order, walker, pContext);
res = walkNode(((SStateWindowNode*)pNode)->pExpr, order, walker, pContext);
break;
case QUERY_NODE_SESSION_WINDOW: {
SSessionWindowNode* pSession = (SSessionWindowNode*)pNode;
......@@ -212,7 +212,7 @@ static EDealRes rewriteNode(SNode** pRawNode, ETraversalOrder order, FNodeRewrit
res = rewriteNode(&(((SOrderByExprNode*)pNode)->pExpr), order, rewriter, pContext);
break;
case QUERY_NODE_STATE_WINDOW:
res = rewriteNode(&(((SStateWindowNode*)pNode)->pCol), order, rewriter, pContext);
res = rewriteNode(&(((SStateWindowNode*)pNode)->pExpr), order, rewriter, pContext);
break;
case QUERY_NODE_SESSION_WINDOW:
res = rewriteNode(&(((SSessionWindowNode*)pNode)->pCol), order, rewriter, pContext);
......@@ -301,10 +301,9 @@ void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker wa
case SQL_CLAUSE_GROUP_BY:
nodesWalkExpr(pSelect->pHaving, walker, pContext);
case SQL_CLAUSE_HAVING:
nodesWalkExprs(pSelect->pOrderByList, walker, pContext);
case SQL_CLAUSE_ORDER_BY:
nodesWalkExprs(pSelect->pProjectionList, walker, pContext);
case SQL_CLAUSE_SELECT:
nodesWalkExprs(pSelect->pOrderByList, walker, pContext);
default:
break;
}
......
source/libs/nodes/src/nodesUtilFuncs.c
浏览文件 @ 6f26d177
......@@ -169,6 +169,8 @@ SNodeptr nodesMakeNode(ENodeType type) {
return makeNode(type, sizeof(SWindowLogicNode));
case QUERY_NODE_LOGIC_PLAN_SORT:
return makeNode(type, sizeof(SSortLogicNode));
case QUERY_NODE_LOGIC_PLAN_PARTITION:
return makeNode(type, sizeof(SPartitionLogicNode));
case QUERY_NODE_LOGIC_SUBPLAN:
return makeNode(type, sizeof(SLogicSubplan));
case QUERY_NODE_LOGIC_PLAN:
......@@ -197,6 +199,10 @@ SNodeptr nodesMakeNode(ENodeType type) {
return makeNode(type, sizeof(SIntervalPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
return makeNode(type, sizeof(SSessionWinodwPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW:
return makeNode(type, sizeof(SStateWinodwPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_PARTITION:
return makeNode(type, sizeof(SPartitionPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_DISPATCH:
return makeNode(type, sizeof(SDataDispatcherNode));
case QUERY_NODE_PHYSICAL_PLAN_INSERT:
......@@ -302,7 +308,7 @@ void nodesDestroyNode(SNodeptr pNode) {
case QUERY_NODE_LIMIT: // no pointer field
break;
case QUERY_NODE_STATE_WINDOW:
nodesDestroyNode(((SStateWindowNode*)pNode)->pCol);
nodesDestroyNode(((SStateWindowNode*)pNode)->pExpr);
break;
case QUERY_NODE_SESSION_WINDOW: {
SSessionWindowNode* pSession = (SSessionWindowNode*)pNode;
......@@ -1041,3 +1047,96 @@ int32_t nodesCollectFuncs(SSelectStmt* pSelect, FFuncClassifier classifier, SNod
return TSDB_CODE_SUCCESS;
}
char *getFillModeString(EFillMode mode) {
switch (mode) {
case FILL_MODE_NONE:
return "none";
case FILL_MODE_VALUE:
return "value";
case FILL_MODE_PREV:
return "prev";
case FILL_MODE_NULL:
return "null";
case FILL_MODE_LINEAR:
return "linear";
case FILL_MODE_NEXT:
return "next";
default:
return "unknown";
}
}
char *nodesGetNameFromColumnNode(SNode *pNode) {
if (NULL == pNode || QUERY_NODE_COLUMN != pNode->type) {
return "NULL";
}
return ((SColumnNode *)pNode)->colName;
}
int32_t nodesGetOutputNumFromSlotList(SNodeList* pSlots) {
if (NULL == pSlots || pSlots->length <= 0) {
return 0;
}
SNode* pNode = NULL;
int32_t num = 0;
FOREACH(pNode, pSlots) {
if (QUERY_NODE_SLOT_DESC != pNode->type) {
continue;
}
SSlotDescNode *descNode = (SSlotDescNode *)pNode;
if (descNode->output) {
++num;
}
}
return num;
}
void valueNodeToVariant(const SValueNode* pNode, SVariant* pVal) {
pVal->nType = pNode->node.resType.type;
pVal->nLen = pNode->node.resType.bytes;
switch (pNode->node.resType.type) {
case TSDB_DATA_TYPE_NULL:
break;
case TSDB_DATA_TYPE_BOOL:
pVal->i = pNode->datum.b;
break;
case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_INT:
case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_TIMESTAMP:
pVal->i = pNode->datum.i;
break;
case TSDB_DATA_TYPE_UTINYINT:
case TSDB_DATA_TYPE_USMALLINT:
case TSDB_DATA_TYPE_UINT:
case TSDB_DATA_TYPE_UBIGINT:
pVal->u = pNode->datum.u;
break;
case TSDB_DATA_TYPE_FLOAT:
case TSDB_DATA_TYPE_DOUBLE:
pVal->d = pNode->datum.d;
break;
case TSDB_DATA_TYPE_NCHAR:
case TSDB_DATA_TYPE_VARCHAR:
case TSDB_DATA_TYPE_VARBINARY:
pVal->pz = pNode->datum.p;
break;
case TSDB_DATA_TYPE_JSON:
case TSDB_DATA_TYPE_DECIMAL:
case TSDB_DATA_TYPE_BLOB:
// todo
default:
break;
}
}
source/libs/parser/inc/parAst.h
浏览文件 @ 6f26d177
......@@ -103,7 +103,7 @@ SNode* createJoinTableNode(SAstCreateContext* pCxt, EJoinType type, SNode* pLeft
SNode* createLimitNode(SAstCreateContext* pCxt, const SToken* pLimit, const SToken* pOffset);
SNode* createOrderByExprNode(SAstCreateContext* pCxt, SNode* pExpr, EOrder order, ENullOrder nullOrder);
SNode* createSessionWindowNode(SAstCreateContext* pCxt, SNode* pCol, SNode* pGap);
SNode* createStateWindowNode(SAstCreateContext* pCxt, SNode* pCol);
SNode* createStateWindowNode(SAstCreateContext* pCxt, SNode* pExpr);
SNode* createIntervalWindowNode(SAstCreateContext* pCxt, SNode* pInterval, SNode* pOffset, SNode* pSliding, SNode* pFill);
SNode* createFillNode(SAstCreateContext* pCxt, EFillMode mode, SNode* pValues);
SNode* createGroupingSetNode(SAstCreateContext* pCxt, SNode* pNode);
......
source/libs/parser/inc/sql.y
浏览文件 @ 6f26d177
......@@ -716,7 +716,7 @@ partition_by_clause_opt(A) ::= PARTITION BY expression_list(B).
twindow_clause_opt(A) ::= . { A = NULL; }
twindow_clause_opt(A) ::=
SESSION NK_LP column_reference(B) NK_COMMA duration_literal(C) NK_RP. { A = createSessionWindowNode(pCxt, releaseRawExprNode(pCxt, B), releaseRawExprNode(pCxt, C)); }
twindow_clause_opt(A) ::= STATE_WINDOW NK_LP column_reference(B) NK_RP. { A = createStateWindowNode(pCxt, releaseRawExprNode(pCxt, B)); }
twindow_clause_opt(A) ::= STATE_WINDOW NK_LP expression(B) NK_RP. { A = createStateWindowNode(pCxt, releaseRawExprNode(pCxt, B)); }
twindow_clause_opt(A) ::=
INTERVAL NK_LP duration_literal(B) NK_RP sliding_opt(C) fill_opt(D). { A = createIntervalWindowNode(pCxt, releaseRawExprNode(pCxt, B), NULL, C, D); }
twindow_clause_opt(A) ::=
......
source/libs/parser/src/parAstCreater.c
浏览文件 @ 6f26d177
......@@ -730,10 +730,10 @@ SNode* createSessionWindowNode(SAstCreateContext* pCxt, SNode* pCol, SNode* pGap
return (SNode*)session;
}
SNode* createStateWindowNode(SAstCreateContext* pCxt, SNode* pCol) {
SNode* createStateWindowNode(SAstCreateContext* pCxt, SNode* pExpr) {
SStateWindowNode* state = (SStateWindowNode*)nodesMakeNode(QUERY_NODE_STATE_WINDOW);
CHECK_OUT_OF_MEM(state);
state->pCol = pCol;
state->pExpr = pExpr;
return (SNode*)state;
}
......
source/libs/parser/src/parTranslater.c
浏览文件 @ 6f26d177
......@@ -438,8 +438,12 @@ static EDealRes translateValue(STranslateContext* pCxt, SValueNode* pVal) {
static EDealRes translateOperator(STranslateContext* pCxt, SOperatorNode* pOp) {
if (nodesIsUnaryOp(pOp)) {
if (OP_TYPE_MINUS == pOp->opType && !IS_NUMERIC_TYPE(((SExprNode*)(pOp->pLeft))->resType.type)) {
return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)(pOp->pLeft))->aliasName);
if (OP_TYPE_MINUS == pOp->opType) {
if (!IS_MATHABLE_TYPE(((SExprNode*)(pOp->pLeft))->resType.type)) {
return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)(pOp->pLeft))->aliasName);
}
pOp->node.resType.type = TSDB_DATA_TYPE_DOUBLE;
pOp->node.resType.bytes = tDataTypes[TSDB_DATA_TYPE_DOUBLE].bytes;
}
return DEAL_RES_CONTINUE;
}
......@@ -2338,46 +2342,6 @@ static void addCreateTbReqIntoVgroup(int32_t acctId, SHashObj* pVgroupHashmap,
}
}
static void valueNodeToVariant(const SValueNode* pNode, SVariant* pVal) {
pVal->nType = pNode->node.resType.type;
pVal->nLen = pNode->node.resType.bytes;
switch (pNode->node.resType.type) {
case TSDB_DATA_TYPE_NULL:
break;
case TSDB_DATA_TYPE_BOOL:
pVal->i = pNode->datum.b;
break;
case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_INT:
case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_TIMESTAMP:
pVal->i = pNode->datum.i;
break;
case TSDB_DATA_TYPE_UTINYINT:
case TSDB_DATA_TYPE_USMALLINT:
case TSDB_DATA_TYPE_UINT:
case TSDB_DATA_TYPE_UBIGINT:
pVal->u = pNode->datum.u;
break;
case TSDB_DATA_TYPE_FLOAT:
case TSDB_DATA_TYPE_DOUBLE:
pVal->d = pNode->datum.d;
break;
case TSDB_DATA_TYPE_NCHAR:
case TSDB_DATA_TYPE_VARCHAR:
case TSDB_DATA_TYPE_VARBINARY:
pVal->pz = pNode->datum.p;
break;
case TSDB_DATA_TYPE_JSON:
case TSDB_DATA_TYPE_DECIMAL:
case TSDB_DATA_TYPE_BLOB:
// todo
default:
break;
}
}
static int32_t addValToKVRow(STranslateContext* pCxt, SValueNode* pVal, const SSchema* pSchema, SKVRowBuilder* pBuilder) {
if (DEAL_RES_ERROR == translateValue(pCxt, pVal)) {
return pCxt->errCode;
......
source/libs/parser/src/sql.c
浏览文件 @ 6f26d177
此差异已折叠。
source/libs/planner/src/planLogicCreater.c
浏览文件 @ 6f26d177
......@@ -477,6 +477,18 @@ static int32_t createWindowLogicNodeFinalize(SLogicPlanContext* pCxt, SSelectStm
return code;
}
static int32_t createWindowLogicNodeByState(SLogicPlanContext* pCxt, SStateWindowNode* pState, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
SWindowLogicNode* pWindow = nodesMakeNode(QUERY_NODE_LOGIC_PLAN_WINDOW);
if (NULL == pWindow) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pWindow->winType = WINDOW_TYPE_STATE;
pWindow->pStateExpr = nodesCloneNode(pState->pExpr);
return createWindowLogicNodeFinalize(pCxt, pSelect, pWindow, pLogicNode);
}
static int32_t createWindowLogicNodeBySession(SLogicPlanContext* pCxt, SSessionWindowNode* pSession, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
SWindowLogicNode* pWindow = nodesMakeNode(QUERY_NODE_LOGIC_PLAN_WINDOW);
if (NULL == pWindow) {
......@@ -525,6 +537,8 @@ static int32_t createWindowLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSele
}
switch (nodeType(pSelect->pWindow)) {
case QUERY_NODE_STATE_WINDOW:
return createWindowLogicNodeByState(pCxt, (SStateWindowNode*)pSelect->pWindow, pSelect, pLogicNode);
case QUERY_NODE_SESSION_WINDOW:
return createWindowLogicNodeBySession(pCxt, (SSessionWindowNode*)pSelect->pWindow, pSelect, pLogicNode);
case QUERY_NODE_INTERVAL_WINDOW:
......@@ -642,6 +656,29 @@ static int32_t createPartitionLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pS
return TSDB_CODE_OUT_OF_MEMORY;
}
SNodeList* pCols = NULL;
int32_t code = nodesCollectColumns(pSelect, SQL_CLAUSE_PARTITION_BY, NULL, &pCols);
if (TSDB_CODE_SUCCESS == code && NULL != pCols) {
pPartition->node.pTargets = nodesCloneList(pCols);
if (NULL == pPartition->node.pTargets) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
if (TSDB_CODE_SUCCESS == code) {
pPartition->pPartitionKeys = nodesCloneList(pSelect->pPartitionByList);
if (NULL == pPartition->pPartitionKeys) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
if (TSDB_CODE_SUCCESS == code) {
*pLogicNode = (SLogicNode*)pPartition;
} else {
nodesDestroyNode(pPartition);
}
return code;
return TSDB_CODE_SUCCESS;
}
......@@ -650,7 +687,35 @@ static int32_t createDistinctLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSe
return TSDB_CODE_SUCCESS;
}
return TSDB_CODE_SUCCESS;
SAggLogicNode* pAgg = (SAggLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_AGG);
if (NULL == pAgg) {
return TSDB_CODE_OUT_OF_MEMORY;
}
int32_t code = TSDB_CODE_SUCCESS;
// set grouyp keys, agg funcs and having conditions
pAgg->pGroupKeys = nodesCloneList(pSelect->pProjectionList);
if (NULL == pAgg->pGroupKeys) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
// rewrite the expression in subsequent clauses
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExpr(pAgg->pGroupKeys, pSelect, SQL_CLAUSE_SELECT);
}
// set the output
if (TSDB_CODE_SUCCESS == code) {
code = createColumnByRewriteExps(pCxt, pAgg->pGroupKeys, &pAgg->node.pTargets);
}
if (TSDB_CODE_SUCCESS == code) {
*pLogicNode = (SLogicNode*)pAgg;
} else {
nodesDestroyNode(pAgg);
}
return code;
}
static int32_t createSelectLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
......
source/libs/planner/src/planPhysiCreater.c
浏览文件 @ 6f26d177
......@@ -176,6 +176,10 @@ static int16_t getUnsetSlotId(const SArray* pSlotIdsInfo) {
}
static int32_t addDataBlockSlotsImpl(SPhysiPlanContext* pCxt, SNodeList* pList, SDataBlockDescNode* pDataBlockDesc, const char* pStmtName, bool output) {
if (NULL == pList) {
return TSDB_CODE_SUCCESS;
}
int32_t code = TSDB_CODE_SUCCESS;
SHashObj* pHash = taosArrayGetP(pCxt->pLocationHelper, pDataBlockDesc->dataBlockId);
int16_t nextSlotId = taosHashGetSize(pHash), slotId = 0;
......@@ -219,6 +223,23 @@ static int32_t addDataBlockSlots(SPhysiPlanContext* pCxt, SNodeList* pList, SDat
return addDataBlockSlotsImpl(pCxt, pList, pDataBlockDesc, NULL, false);
}
static int32_t addDataBlockSlot(SPhysiPlanContext* pCxt, SNode** pNode, SDataBlockDescNode* pDataBlockDesc) {
if (NULL == pNode || NULL == *pNode) {
return TSDB_CODE_SUCCESS;
}
SNodeList* pList = NULL;
int32_t code = nodesListMakeAppend(&pList, *pNode);
if (TSDB_CODE_SUCCESS == code) {
code = addDataBlockSlots(pCxt, pList, pDataBlockDesc);
}
if (TSDB_CODE_SUCCESS == code) {
*pNode = nodesListGetNode(pList, 0);
}
nodesClearList(pList);
return code;
}
static int32_t addDataBlockSlotsForProject(SPhysiPlanContext* pCxt, const char* pStmtName, SNodeList* pList, SDataBlockDescNode* pDataBlockDesc) {
return addDataBlockSlotsImpl(pCxt, pList, pDataBlockDesc, pStmtName, true);
}
......@@ -244,6 +265,7 @@ static EDealRes doSetSlotId(SNode* pNode, void* pContext) {
}
// pIndex is definitely not NULL, otherwise it is a bug
if (NULL == pIndex) {
pCxt->errCode = TSDB_CODE_PLAN_INTERNAL_ERROR;
return DEAL_RES_ERROR;
}
((SColumnNode*)pNode)->dataBlockId = pIndex->dataBlockId;
......@@ -614,6 +636,25 @@ static int32_t rewritePrecalcExprs(SPhysiPlanContext* pCxt, SNodeList* pList, SN
return cxt.errCode;
}
static int32_t rewritePrecalcExpr(SPhysiPlanContext* pCxt, SNode* pNode, SNodeList** pPrecalcExprs, SNode** pRewritten) {
if (NULL == pNode) {
return TSDB_CODE_SUCCESS;
}
SNodeList* pList = NULL;
int32_t code = nodesListMakeAppend(&pList, pNode);
SNodeList* pRewrittenList = NULL;
if (TSDB_CODE_SUCCESS == code) {
code = rewritePrecalcExprs(pCxt, pList, pPrecalcExprs, &pRewrittenList);
}
if (TSDB_CODE_SUCCESS == code) {
*pRewritten = nodesListGetNode(pRewrittenList, 0);
}
nodesClearList(pList);
nodesClearList(pRewrittenList);
return code;
}
static int32_t createAggPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren, SAggLogicNode* pAggLogicNode, SPhysiNode** pPhyNode) {
SAggPhysiNode* pAgg = (SAggPhysiNode*)makePhysiNode(pCxt, (SLogicNode*)pAggLogicNode, QUERY_NODE_PHYSICAL_PLAN_AGG);
if (NULL == pAgg) {
......@@ -818,6 +859,40 @@ static int32_t createSessionWindowPhysiNode(SPhysiPlanContext* pCxt, SNodeList*
return createWindowPhysiNodeFinalize(pCxt, pChildren, &pSession->window, pWindowLogicNode, pPhyNode);
}
static int32_t createStateWindowPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren, SWindowLogicNode* pWindowLogicNode, SPhysiNode** pPhyNode) {
SStateWinodwPhysiNode* pState = (SStateWinodwPhysiNode*)makePhysiNode(pCxt, (SLogicNode*)pWindowLogicNode, QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW);
if (NULL == pState) {
return TSDB_CODE_OUT_OF_MEMORY;
}
SNodeList* pPrecalcExprs = NULL;
SNode* pStateKey = NULL;
int32_t code = rewritePrecalcExpr(pCxt, pWindowLogicNode->pStateExpr, &pPrecalcExprs, &pStateKey);
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, &pState->window.pExprs);
if (TSDB_CODE_SUCCESS == code) {
code = addDataBlockSlots(pCxt, pState->window.pExprs, pChildTupe);
}
}
if (TSDB_CODE_SUCCESS == code) {
code = setNodeSlotId(pCxt, pChildTupe->dataBlockId, -1, pStateKey, &pState->pStateKey);
if (TSDB_CODE_SUCCESS == code) {
code = addDataBlockSlot(pCxt, &pState->pStateKey, pState->window.node.pOutputDataBlockDesc);
}
}
if (TSDB_CODE_SUCCESS != code) {
nodesDestroyNode(pState);
return code;
}
return createWindowPhysiNodeFinalize(pCxt, pChildren, &pState->window, pWindowLogicNode, pPhyNode);
}
static int32_t createWindowPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren, SWindowLogicNode* pWindowLogicNode, SPhysiNode** pPhyNode) {
switch (pWindowLogicNode->winType) {
case WINDOW_TYPE_INTERVAL:
......@@ -825,7 +900,7 @@ static int32_t createWindowPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildr
case WINDOW_TYPE_SESSION:
return createSessionWindowPhysiNode(pCxt, pChildren, pWindowLogicNode, pPhyNode);
case WINDOW_TYPE_STATE:
break;
return createStateWindowPhysiNode(pCxt, pChildren, pWindowLogicNode, pPhyNode);
default:
break;
}
......@@ -867,6 +942,41 @@ static int32_t createSortPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren
return code;
}
static int32_t createPartitionPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren, SPartitionLogicNode* pPartLogicNode, SPhysiNode** pPhyNode) {
SPartitionPhysiNode* pPart = (SPartitionPhysiNode*)makePhysiNode(pCxt, (SLogicNode*)pPartLogicNode, QUERY_NODE_PHYSICAL_PLAN_PARTITION);
if (NULL == pPart) {
return TSDB_CODE_OUT_OF_MEMORY;
}
SNodeList* pPrecalcExprs = NULL;
SNodeList* pPartitionKeys = NULL;
int32_t code = rewritePrecalcExprs(pCxt, pPartLogicNode->pPartitionKeys, &pPrecalcExprs, &pPartitionKeys);
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, &pPart->pExprs);
if (TSDB_CODE_SUCCESS == code) {
code = addDataBlockSlots(pCxt, pPart->pExprs, pChildTupe);
}
}
if (TSDB_CODE_SUCCESS == code) {
code = setListSlotId(pCxt, pChildTupe->dataBlockId, -1, pPartitionKeys, &pPart->pPartitionKeys);
if (TSDB_CODE_SUCCESS == code) {
code = addDataBlockSlots(pCxt, pPart->pPartitionKeys, pPart->node.pOutputDataBlockDesc);
}
}
if (TSDB_CODE_SUCCESS == code) {
*pPhyNode = (SPhysiNode*)pPart;
} else {
nodesDestroyNode(pPart);
}
return code;
}
static int32_t doCreatePhysiNode(SPhysiPlanContext* pCxt, SLogicNode* pLogicNode, SSubplan* pSubplan, SNodeList* pChildren, SPhysiNode** pPhyNode) {
switch (nodeType(pLogicNode)) {
case QUERY_NODE_LOGIC_PLAN_SCAN:
......@@ -883,6 +993,8 @@ static int32_t doCreatePhysiNode(SPhysiPlanContext* pCxt, SLogicNode* pLogicNode
return createWindowPhysiNode(pCxt, pChildren, (SWindowLogicNode*)pLogicNode, pPhyNode);
case QUERY_NODE_LOGIC_PLAN_SORT:
return createSortPhysiNode(pCxt, pChildren, (SSortLogicNode*)pLogicNode, pPhyNode);
case QUERY_NODE_LOGIC_PLAN_PARTITION:
return createPartitionPhysiNode(pCxt, pChildren, (SPartitionLogicNode*)pLogicNode, pPhyNode);
default:
break;
}
......
source/libs/planner/test/plannerTest.cpp
浏览文件 @ 6f26d177
......@@ -217,6 +217,32 @@ TEST_F(PlannerTest, sessionWindow) {
ASSERT_TRUE(run());
}
TEST_F(PlannerTest, stateWindow) {
setDatabase("root", "test");
bind("SELECT count(*) FROM t1 state_window(c1)");
ASSERT_TRUE(run());
bind("SELECT count(*) FROM t1 state_window(c1 + 10)");
ASSERT_TRUE(run());
}
TEST_F(PlannerTest, partitionBy) {
setDatabase("root", "test");
bind("SELECT * FROM t1 partition by c1");
ASSERT_TRUE(run());
bind("SELECT count(*) FROM t1 partition by c1");
ASSERT_TRUE(run());
bind("SELECT count(*) FROM t1 partition by c1 group by c2");
ASSERT_TRUE(run());
bind("SELECT count(*) FROM st1 partition by tag1, tag2 interval(10s)");
ASSERT_TRUE(run());
}
TEST_F(PlannerTest, orderBy) {
setDatabase("root", "test");
......@@ -230,6 +256,19 @@ TEST_F(PlannerTest, orderBy) {
ASSERT_TRUE(run());
}
TEST_F(PlannerTest, distinct) {
setDatabase("root", "test");
bind("SELECT distinct c1 FROM t1");
ASSERT_TRUE(run());
bind("SELECT distinct c1, c2 + 10 FROM t1");
ASSERT_TRUE(run());
bind("SELECT distinct c1 + 10 a FROM t1 order by a");
ASSERT_TRUE(run());
}
TEST_F(PlannerTest, limit) {
setDatabase("root", "test");
......
source/libs/qworker/inc/qworkerInt.h
浏览文件 @ 6f26d177
......@@ -107,15 +107,17 @@ typedef struct SQWTaskCtx {
SRWLatch lock;
int8_t phase;
int8_t taskType;
int8_t explain;
bool emptyRes;
bool queryFetched;
bool queryEnd;
bool queryContinue;
bool queryInQueue;
int32_t rspCode;
SQWConnInfo connInfo;
SQWConnInfo ctrlConnInfo;
SQWConnInfo dataConnInfo;
int8_t events[QW_EVENT_MAX];
qTaskInfo_t taskHandle;
......
source/libs/qworker/inc/qworkerMsg.h
浏览文件 @ 6f26d177
......@@ -23,7 +23,7 @@ extern "C" {
#include "qworkerInt.h"
#include "dataSinkMgt.h"
int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, int8_t taskType);
int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, int8_t taskType, int8_t explain);
int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg);
int32_t qwProcessReady(QW_FPARAMS_DEF, SQWMsg *qwMsg);
int32_t qwProcessFetch(QW_FPARAMS_DEF, SQWMsg *qwMsg);
......@@ -38,6 +38,7 @@ void qwBuildFetchRsp(void *msg, SOutputData *input, int32_t len, bool qComple
int32_t qwBuildAndSendCQueryMsg(QW_FPARAMS_DEF, SQWConnInfo *pConn);
int32_t qwBuildAndSendReadyRsp(SQWConnInfo *pConn, int32_t code);
int32_t qwBuildAndSendQueryRsp(SQWConnInfo *pConn, int32_t code);
int32_t qwBuildAndSendExplainRsp(SQWConnInfo *pConn, SExplainExecInfo *execInfo, int32_t num);
void qwFreeFetchRsp(void *msg);
int32_t qwMallocFetchRsp(int32_t length, SRetrieveTableRsp **rsp);
int32_t qwGetSchTasksStatus(SQWorkerMgmt *mgmt, uint64_t sId, SSchedulerStatusRsp **rsp);
......
source/libs/qworker/src/qworker.c
浏览文件 @ 6f26d177
......@@ -432,8 +432,10 @@ int32_t qwKillTaskHandle(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {
void qwFreeTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {
tmsgReleaseHandle(ctx->connInfo.handle, TAOS_CONN_SERVER);
ctx->connInfo.handle = NULL;
tmsgReleaseHandle(ctx->ctrlConnInfo.handle, TAOS_CONN_SERVER);
ctx->ctrlConnInfo.handle = NULL;
// NO need to release dataConnInfo
qwFreeTaskHandle(QW_FPARAMS(), &ctx->taskHandle);
......@@ -537,6 +539,29 @@ int32_t qwDropTask(QW_FPARAMS_DEF) {
return TSDB_CODE_SUCCESS;
}
int32_t qwHandleTaskComplete(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {
qTaskInfo_t *taskHandle = &ctx->taskHandle;
if (TASK_TYPE_TEMP == ctx->taskType) {
if (ctx->explain) {
SExplainExecInfo *execInfo = NULL;
int32_t resNum = 0;
QW_ERR_RET(qGetExplainExecInfo(ctx->taskHandle, &resNum, &execInfo));
SQWConnInfo connInfo = {0};
connInfo.handle = ctx->ctrlConnInfo.handle;
QW_ERR_RET(qwBuildAndSendExplainRsp(&connInfo, execInfo, resNum));
}
qwFreeTaskHandle(QW_FPARAMS(), taskHandle);
}
return TSDB_CODE_SUCCESS;
}
int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryEnd) {
int32_t code = 0;
bool qcontinue = true;
......@@ -562,10 +587,8 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryEnd) {
QW_TASK_DLOG("qExecTask end with empty res, useconds:%"PRIu64, useconds);
dsEndPut(sinkHandle, useconds);
if (TASK_TYPE_TEMP == ctx->taskType) {
qwFreeTaskHandle(QW_FPARAMS(), taskHandle);
}
QW_ERR_RET(qwHandleTaskComplete(QW_FPARAMS(), ctx));
if (queryEnd) {
*queryEnd = true;
......@@ -658,19 +681,6 @@ int32_t qwGetResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, void
bool queryEnd = false;
int32_t code = 0;
if (ctx->emptyRes) {
QW_TASK_DLOG_E("query end with empty result");
qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_SUCCEED);
QW_ERR_RET(qwMallocFetchRsp(len, &rsp));
*rspMsg = rsp;
*dataLen = 0;
pOutput->queryEnd = true;
return TSDB_CODE_SUCCESS;
}
dsGetDataLength(ctx->sinkHandle, &len, &queryEnd);
if (len < 0) {
......@@ -760,12 +770,12 @@ int32_t qwHandlePrePhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *inpu
}
if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_DROP)) {
dropConnection = &ctx->connInfo;
dropConnection = &ctx->ctrlConnInfo;
QW_ERR_JRET(qwDropTask(QW_FPARAMS()));
dropConnection = NULL;
qwBuildAndSendDropRsp(&ctx->connInfo, code);
QW_TASK_DLOG("drop rsp send, handle:%p, code:%x - %s", ctx->connInfo.handle, code, tstrerror(code));
qwBuildAndSendDropRsp(&ctx->ctrlConnInfo, code);
QW_TASK_DLOG("drop rsp send, handle:%p, code:%x - %s", ctx->ctrlConnInfo.handle, code, tstrerror(code));
QW_ERR_JRET(TSDB_CODE_QRY_TASK_DROPPED);
break;
......@@ -798,12 +808,12 @@ int32_t qwHandlePrePhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *inpu
}
if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_DROP)) {
dropConnection = &ctx->connInfo;
dropConnection = &ctx->ctrlConnInfo;
QW_ERR_JRET(qwDropTask(QW_FPARAMS()));
dropConnection = NULL;
qwBuildAndSendDropRsp(&ctx->connInfo, code);
QW_TASK_DLOG("drop rsp send, handle:%p, code:%x - %s", ctx->connInfo.handle, code, tstrerror(code));
qwBuildAndSendDropRsp(&ctx->ctrlConnInfo, code);
QW_TASK_DLOG("drop rsp send, handle:%p, code:%x - %s", ctx->ctrlConnInfo.handle, code, tstrerror(code));
QW_ERR_JRET(TSDB_CODE_QRY_TASK_DROPPED);
}
......@@ -863,17 +873,13 @@ int32_t qwHandlePostPhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *inp
}
if (QW_PHASE_POST_QUERY == phase) {
if (NULL == ctx->taskHandle && NULL == ctx->sinkHandle) {
ctx->emptyRes = true;
}
#if 0
if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_READY)) {
readyConnection = &ctx->connInfo;
QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_READY);
}
#else
connInfo.handle = ctx->connInfo.handle;
connInfo.handle = ctx->ctrlConnInfo.handle;
readyConnection = &connInfo;
QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_READY);
......@@ -886,8 +892,8 @@ int32_t qwHandlePostPhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *inp
QW_ERR_JRET(TSDB_CODE_QRY_APP_ERROR);
}
qwBuildAndSendDropRsp(&ctx->connInfo, code);
QW_TASK_DLOG("drop rsp send, handle:%p, code:%x - %s", ctx->connInfo.handle, code, tstrerror(code));
qwBuildAndSendDropRsp(&ctx->ctrlConnInfo, code);
QW_TASK_DLOG("drop rsp send, handle:%p, code:%x - %s", ctx->ctrlConnInfo.handle, code, tstrerror(code));
QW_ERR_JRET(qwDropTask(QW_FPARAMS()));
QW_ERR_JRET(TSDB_CODE_QRY_TASK_DROPPED);
......@@ -931,7 +937,7 @@ _return:
QW_RET(code);
}
int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, int8_t taskType) {
int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, int8_t taskType, int8_t explain) {
int32_t code = 0;
bool queryRsped = false;
struct SSubplan *plan = NULL;
......@@ -947,9 +953,10 @@ int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, int8_t taskType) {
QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx));
atomic_store_8(&ctx->taskType, taskType);
atomic_store_8(&ctx->explain, explain);
atomic_store_ptr(&ctx->connInfo.handle, qwMsg->connInfo.handle);
atomic_store_ptr(&ctx->connInfo.ahandle, qwMsg->connInfo.ahandle);
atomic_store_ptr(&ctx->ctrlConnInfo.handle, qwMsg->connInfo.handle);
atomic_store_ptr(&ctx->ctrlConnInfo.ahandle, qwMsg->connInfo.ahandle);
QW_TASK_DLOGL("subplan json string, len:%d, %s", qwMsg->msgLen, qwMsg->msg);
......@@ -1011,8 +1018,8 @@ int32_t qwProcessReady(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
}
if (ctx->phase == QW_PHASE_PRE_QUERY) {
ctx->connInfo.handle == qwMsg->connInfo.handle;
ctx->connInfo.ahandle = qwMsg->connInfo.ahandle;
ctx->ctrlConnInfo.handle == qwMsg->connInfo.handle;
ctx->ctrlConnInfo.ahandle = qwMsg->connInfo.ahandle;
QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_READY);
needRsp = false;
QW_TASK_DLOG_E("ready msg will not rsp now");
......@@ -1089,10 +1096,13 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
if (rsp) {
bool qComplete = (DS_BUF_EMPTY == sOutput.bufStatus && sOutput.queryEnd);
qwBuildFetchRsp(rsp, &sOutput, dataLen, qComplete);
atomic_store_8((int8_t*)&ctx->queryEnd, qComplete);
if (qComplete) {
atomic_store_8((int8_t*)&ctx->queryEnd, true);
}
qwMsg->connInfo = ctx->connInfo;
qwMsg->connInfo = ctx->dataConnInfo;
QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_FETCH);
qwBuildAndSendFetchRsp(&qwMsg->connInfo, rsp, dataLen, code);
......@@ -1113,7 +1123,7 @@ _return:
qwFreeFetchRsp(rsp);
rsp = NULL;
qwMsg->connInfo = ctx->connInfo;
qwMsg->connInfo = ctx->dataConnInfo;
qwBuildAndSendFetchRsp(&qwMsg->connInfo, rsp, 0, code);
QW_TASK_DLOG("fetch rsp send, handle:%p, code:%x - %s, dataLen:%d", qwMsg->connInfo.handle, code, tstrerror(code), 0);
}
......@@ -1151,14 +1161,17 @@ int32_t qwProcessFetch(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
QW_ERR_JRET(qwGetResFromSink(QW_FPARAMS(), ctx, &dataLen, &rsp, &sOutput));
if (NULL == rsp) {
atomic_store_ptr(&ctx->connInfo.handle, qwMsg->connInfo.handle);
atomic_store_ptr(&ctx->connInfo.ahandle, qwMsg->connInfo.ahandle);
atomic_store_ptr(&ctx->dataConnInfo.handle, qwMsg->connInfo.handle);
atomic_store_ptr(&ctx->dataConnInfo.ahandle, qwMsg->connInfo.ahandle);
QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_FETCH);
} else {
bool qComplete = (DS_BUF_EMPTY == sOutput.bufStatus && sOutput.queryEnd);
qwBuildFetchRsp(rsp, &sOutput, dataLen, qComplete);
atomic_store_8((int8_t*)&ctx->queryEnd, qComplete);
if (qComplete) {
atomic_store_8((int8_t*)&ctx->queryEnd, true);
}
}
if ((!sOutput.queryEnd) && (DS_BUF_LOW == sOutput.bufStatus || DS_BUF_EMPTY == sOutput.bufStatus)) {
......@@ -1236,8 +1249,8 @@ int32_t qwProcessDrop(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
}
if (!rsped) {
ctx->connInfo.handle = qwMsg->connInfo.handle;
ctx->connInfo.ahandle = qwMsg->connInfo.ahandle;
ctx->ctrlConnInfo.handle = qwMsg->connInfo.handle;
ctx->ctrlConnInfo.ahandle = qwMsg->connInfo.ahandle;
QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_DROP);
}
......
source/libs/qworker/src/qworkerMsg.c
浏览文件 @ 6f26d177
......@@ -85,6 +85,27 @@ int32_t qwBuildAndSendReadyRsp(SQWConnInfo *pConn, int32_t code) {
return TSDB_CODE_SUCCESS;
}
int32_t qwBuildAndSendExplainRsp(SQWConnInfo *pConn, SExplainExecInfo *execInfo, int32_t num) {
SExplainRsp rsp = {.numOfPlans = num, .subplanInfo = execInfo};
int32_t contLen = tSerializeSExplainRsp(NULL, 0, &rsp);
void *pRsp = rpcMallocCont(contLen);
tSerializeSExplainRsp(pRsp, contLen, &rsp);
SRpcMsg rpcRsp = {
.msgType = TDMT_VND_EXPLAIN_RSP,
.handle = pConn->handle,
.ahandle = pConn->ahandle,
.pCont = pRsp,
.contLen = contLen,
.code = 0,
};
tmsgSendRsp(&rpcRsp);
return TSDB_CODE_SUCCESS;
}
int32_t qwBuildAndSendHbRsp(SQWConnInfo *pConn, SSchedulerHbRsp *pStatus, int32_t code) {
int32_t contLen = tSerializeSSchedulerHbRsp(NULL, 0, pStatus);
void *pRsp = rpcMallocCont(contLen);
......@@ -327,7 +348,7 @@ int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
QW_SCH_TASK_DLOG("processQuery start, node:%p, handle:%p, sql:%s", node, pMsg->handle, sql);
taosMemoryFreeClear(sql);
QW_ERR_RET(qwProcessQuery(QW_FPARAMS(), &qwMsg, msg->taskType));
QW_ERR_RET(qwProcessQuery(QW_FPARAMS(), &qwMsg, msg->taskType, msg->explain));
QW_SCH_TASK_DLOG("processQuery end, node:%p", node);
......
source/libs/scalar/inc/sclvector.h
浏览文件 @ 6f26d177
......@@ -75,6 +75,8 @@ static FORCE_INLINE _getDoubleValue_fn_t getVectorDoubleValueFn(int32_t srcType)
p = getVectorDoubleValue_FLOAT;
} else if (srcType == TSDB_DATA_TYPE_DOUBLE) {
p = getVectorDoubleValue_DOUBLE;
} else if (srcType == TSDB_DATA_TYPE_TIMESTAMP) {
p = getVectorDoubleValue_BIGINT;
} else {
assert(0);
}
......
source/libs/scalar/src/filter.c
浏览文件 @ 6f26d177
......@@ -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);
......
source/libs/scalar/src/scalar.c
浏览文件 @ 6f26d177
......@@ -10,7 +10,8 @@
int32_t scalarGetOperatorParamNum(EOperatorType type) {
if (OP_TYPE_IS_NULL == type || OP_TYPE_IS_NOT_NULL == type || OP_TYPE_IS_TRUE == type || OP_TYPE_IS_NOT_TRUE == type
|| OP_TYPE_IS_FALSE == type || OP_TYPE_IS_NOT_FALSE == type || OP_TYPE_IS_UNKNOWN == type || OP_TYPE_IS_NOT_UNKNOWN == type) {
|| OP_TYPE_IS_FALSE == type || OP_TYPE_IS_NOT_FALSE == type || OP_TYPE_IS_UNKNOWN == type || OP_TYPE_IS_NOT_UNKNOWN == type
|| OP_TYPE_MINUS == type) {
return 1;
}
......
source/libs/scalar/src/sclvector.c
浏览文件 @ 6f26d177
......@@ -773,6 +773,32 @@ void vectorMathRemainder(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam
doReleaseVec(pRightCol, rightConvert);
}
void vectorMathMinus(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pOut, int32_t _ord) {
SColumnInfoData *pOutputCol = pOut->columnData;
pOut->numOfRows = pLeft->numOfRows;
int32_t i = ((_ord) == TSDB_ORDER_ASC)? 0 : (pLeft->numOfRows - 1);
int32_t step = ((_ord) == TSDB_ORDER_ASC)? 1 : -1;
int32_t leftConvert = 0;
SColumnInfoData *pLeftCol = doVectorConvert(pLeft, &leftConvert);
_getDoubleValue_fn_t getVectorDoubleValueFnLeft = getVectorDoubleValueFn(pLeftCol->info.type);
double *output = (double *)pOutputCol->pData;
for (; i < pLeft->numOfRows && i >= 0; i += step, output += 1) {
*output = - getVectorDoubleValueFnLeft(pLeftCol->pData, i);
}
pOutputCol->hasNull = pLeftCol->hasNull;
if (pOutputCol->hasNull) {
memcpy(pOutputCol->nullbitmap, pLeftCol->nullbitmap, BitmapLen(pLeft->numOfRows));
}
doReleaseVec(pLeftCol, leftConvert);
}
void vectorConcat(SScalarParam* pLeft, SScalarParam* pRight, void *out, int32_t _ord) {
#if 0
int32_t len = pLeft->bytes + pRight->bytes;
......@@ -1102,6 +1128,8 @@ _bin_scalar_fn_t getBinScalarOperatorFn(int32_t binFunctionId) {
return vectorMathDivide;
case OP_TYPE_MOD:
return vectorMathRemainder;
case OP_TYPE_MINUS:
return vectorMathMinus;
case OP_TYPE_GREATER_THAN:
return vectorGreater;
case OP_TYPE_GREATER_EQUAL:
......@@ -1140,4 +1168,4 @@ _bin_scalar_fn_t getBinScalarOperatorFn(int32_t binFunctionId) {
assert(0);
return NULL;
}
}
\ No newline at end of file
}
source/libs/scalar/test/filter/filterTests.cpp
浏览文件 @ 6f26d177
......@@ -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);
}
......
source/libs/scheduler/inc/schedulerInt.h
浏览文件 @ 6f26d177
......@@ -26,6 +26,7 @@ extern "C" {
#include "scheduler.h"
#include "thash.h"
#include "trpc.h"
#include "command.h"
#define SCHEDULE_DEFAULT_MAX_JOB_NUM 1000
#define SCHEDULE_DEFAULT_MAX_TASK_NUM 1000
......@@ -165,6 +166,7 @@ typedef struct SSchJob {
SHashObj *succTasks; // succeed tasks, key:taskid, value:SQueryTask*
SHashObj *failTasks; // failed tasks, key:taskid, value:SQueryTask*
SExplainCtx *explainCtx;
int8_t status;
SQueryNodeAddr resNode;
tsem_t rspSem;
......@@ -211,6 +213,7 @@ extern SSchedulerMgmt schMgmt;
#define SCH_JOB_NEED_FETCH(_job) SCH_IS_QUERY_JOB(_job)
#define SCH_IS_WAIT_ALL_JOB(_job) (!SCH_IS_QUERY_JOB(_job))
#define SCH_IS_NEED_DROP_JOB(_job) (SCH_IS_QUERY_JOB(_job))
#define SCH_IS_EXPLAIN_JOB(_job) (EXPLAIN_MODE_ANALYZE == (_job)->attr.explainMode)
#define SCH_IS_LEVEL_UNFINISHED(_level) ((_level)->taskLaunchedNum < (_level)->taskNum)
#define SCH_GET_CUR_EP(_addr) (&(_addr)->epSet.eps[(_addr)->epSet.inUse])
......@@ -251,6 +254,8 @@ int32_t schFetchFromRemote(SSchJob *pJob);
int32_t schProcessOnTaskFailure(SSchJob *pJob, SSchTask *pTask, int32_t errCode);
int32_t schBuildAndSendHbMsg(SQueryNodeEpId *nodeEpId);
int32_t schCloneSMsgSendInfo(void *src, void **dst);
int32_t schValidateAndBuildJob(SQueryPlan *pDag, SSchJob *pJob);
void schFreeJobImpl(void *job);
#ifdef __cplusplus
......
source/libs/scheduler/src/scheduler.c
浏览文件 @ 6f26d177
......@@ -67,6 +67,81 @@ int32_t schInitTask(SSchJob *pJob, SSchTask *pTask, SSubplan *pPlan, SSchLevel *
return TSDB_CODE_SUCCESS;
}
int32_t schInitJob(SSchJob **pSchJob, SQueryPlan *pDag, void *transport, SArray *pNodeList, const char *sql,
int64_t startTs, bool syncSchedule) {
int32_t code = 0;
SSchJob *pJob = taosMemoryCalloc(1, sizeof(SSchJob));
if (NULL == pJob) {
qError("QID:%" PRIx64 " calloc %d failed", pDag->queryId, (int32_t)sizeof(SSchJob));
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
pJob->attr.explainMode = pDag->explainInfo.mode;
pJob->attr.syncSchedule = syncSchedule;
pJob->transport = transport;
pJob->sql = sql;
if (pNodeList != NULL) {
pJob->nodeList = taosArrayDup(pNodeList);
}
SCH_ERR_JRET(schValidateAndBuildJob(pDag, pJob));
if (SCH_IS_EXPLAIN_JOB(pJob)) {
SCH_ERR_JRET(qExecExplainBegin(pDag, &pJob->explainCtx, startTs));
}
pJob->execTasks =
taosHashInit(pDag->numOfSubplans, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK);
if (NULL == pJob->execTasks) {
SCH_JOB_ELOG("taosHashInit %d execTasks failed", pDag->numOfSubplans);
SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
pJob->succTasks =
taosHashInit(pDag->numOfSubplans, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK);
if (NULL == pJob->succTasks) {
SCH_JOB_ELOG("taosHashInit %d succTasks failed", pDag->numOfSubplans);
SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
pJob->failTasks =
taosHashInit(pDag->numOfSubplans, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK);
if (NULL == pJob->failTasks) {
SCH_JOB_ELOG("taosHashInit %d failTasks failed", pDag->numOfSubplans);
SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
tsem_init(&pJob->rspSem, 0, 0);
int64_t refId = taosAddRef(schMgmt.jobRef, pJob);
if (refId < 0) {
SCH_JOB_ELOG("taosAddRef job failed, error:%s", tstrerror(terrno));
SCH_ERR_JRET(terrno);
}
if (NULL == schAcquireJob(refId)) {
SCH_JOB_ELOG("schAcquireJob job failed, refId:%" PRIx64, refId);
SCH_RET(TSDB_CODE_SCH_STATUS_ERROR);
}
pJob->refId = refId;
SCH_JOB_DLOG("job refId:%" PRIx64, pJob->refId);
pJob->status = JOB_TASK_STATUS_NOT_START;
*pSchJob = pJob;
return TSDB_CODE_SUCCESS;
_return:
schFreeJobImpl(pJob);
SCH_RET(code);
}
void schFreeRpcCtx(SRpcCtx *pCtx) {
if (NULL == pCtx) {
return;
......@@ -124,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) {
......@@ -968,6 +1044,19 @@ _return:
SCH_RET(schProcessOnTaskFailure(pJob, pJob->fetchTask, code));
}
int32_t schProcessOnExplainDone(SSchJob *pJob, SSchTask *pTask, SRetrieveTableRsp *pRsp) {
SCH_TASK_DLOG("got explain rsp, rows:%d, complete:%d", htonl(pRsp->numOfRows), pRsp->completed);
atomic_store_32(&pJob->resNumOfRows, htonl(pRsp->numOfRows));
atomic_store_ptr(&pJob->resData, pRsp);
SCH_SET_TASK_STATUS(pTask, JOB_TASK_STATUS_SUCCEED);
schProcessOnDataFetched(pJob);
return TSDB_CODE_SUCCESS;
}
// Note: no more task error processing, handled in function internal
int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, char *msg, int32_t msgSize,
int32_t rspCode) {
......@@ -986,7 +1075,7 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
case TDMT_VND_CREATE_TABLE_RSP: {
SVCreateTbBatchRsp batchRsp = {0};
if (msg) {
tDeserializeSVCreateTbBatchRsp(msg, msgSize, &batchRsp);
SCH_ERR_JRET(tDeserializeSVCreateTbBatchRsp(msg, msgSize, &batchRsp));
if (batchRsp.rspList) {
int32_t num = taosArrayGetSize(batchRsp.rspList);
for (int32_t i = 0; i < num; ++i) {
......@@ -1024,7 +1113,7 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
case TDMT_VND_QUERY_RSP: {
SQueryTableRsp rsp = {0};
if (msg) {
tDeserializeSQueryTableRsp(msg, msgSize, &rsp);
SCH_ERR_JRET(tDeserializeSQueryTableRsp(msg, msgSize, &rsp));
SCH_ERR_JRET(rsp.code);
}
......@@ -1050,6 +1139,36 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
break;
}
case TDMT_VND_EXPLAIN_RSP: {
SCH_ERR_JRET(rspCode);
if (NULL == msg) {
SCH_ERR_JRET(TSDB_CODE_QRY_INVALID_INPUT);
}
if (!SCH_IS_EXPLAIN_JOB(pJob)) {
SCH_TASK_ELOG("invalid msg received for none explain query, msg type:%s", TMSG_INFO(msgType));
SCH_ERR_JRET(TSDB_CODE_QRY_INVALID_INPUT);
}
if (pJob->resData) {
SCH_TASK_ELOG("explain result is already generated, res:%p", pJob->resData);
SCH_ERR_JRET(TSDB_CODE_SCH_STATUS_ERROR);
}
SExplainRsp rsp = {0};
if (tDeserializeSExplainRsp(msg, msgSize, &rsp)) {
taosMemoryFree(rsp.subplanInfo);
SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
SRetrieveTableRsp *pRsp = NULL;
SCH_ERR_JRET(qExplainUpdateExecInfo(pJob->explainCtx, &rsp, pTask->plan->id.groupId, &pRsp));
if (pRsp) {
SCH_ERR_JRET(schProcessOnExplainDone(pJob, pTask, pRsp));
}
break;
}
case TDMT_VND_FETCH_RSP: {
SRetrieveTableRsp *rsp = (SRetrieveTableRsp *)msg;
......@@ -1058,6 +1177,24 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
SCH_ERR_JRET(TSDB_CODE_QRY_INVALID_INPUT);
}
if (SCH_IS_EXPLAIN_JOB(pJob)) {
if (rsp->completed) {
SRetrieveTableRsp *pRsp = NULL;
SCH_ERR_JRET(qExecExplainEnd(pJob->explainCtx, &pRsp));
if (pRsp) {
SCH_ERR_JRET(schProcessOnExplainDone(pJob, pTask, pRsp));
}
return TSDB_CODE_SUCCESS;
}
atomic_val_compare_exchange_32(&pJob->remoteFetch, 1, 0);
SCH_ERR_JRET(schFetchFromRemote(pJob));
return TSDB_CODE_SUCCESS;
}
if (pJob->resData) {
SCH_TASK_ELOG("got fetch rsp while res already exists, res:%p", pJob->resData);
taosMemoryFreeClear(rsp);
......@@ -1158,6 +1295,10 @@ int32_t schHandleReadyCallback(void *param, const SDataBuf *pMsg, int32_t code)
return schHandleCallback(param, pMsg, TDMT_VND_RES_READY_RSP, code);
}
int32_t schHandleExplainCallback(void *param, const SDataBuf *pMsg, int32_t code) {
return schHandleCallback(param, pMsg, TDMT_VND_EXPLAIN_RSP, code);
}
int32_t schHandleDropCallback(void *param, const SDataBuf *pMsg, int32_t code) {
SSchTaskCallbackParam *pParam = (SSchTaskCallbackParam *)param;
qDebug("QID:%" PRIx64 ",TID:%" PRIx64 " drop task rsp received, code:%x", pParam->queryId, pParam->taskId, code);
......@@ -1246,6 +1387,9 @@ int32_t schGetCallbackFp(int32_t msgType, __async_send_cb_fn_t *fp) {
case TDMT_VND_RES_READY:
*fp = schHandleReadyCallback;
break;
case TDMT_VND_EXPLAIN:
*fp = schHandleExplainCallback;
break;
case TDMT_VND_FETCH:
*fp = schHandleFetchCallback;
break;
......@@ -1266,6 +1410,43 @@ int32_t schGetCallbackFp(int32_t msgType, __async_send_cb_fn_t *fp) {
return TSDB_CODE_SUCCESS;
}
int32_t schGenerateTaskCallBackAHandle(SSchJob *pJob, SSchTask *pTask, int32_t msgType, SMsgSendInfo **pMsgSendInfo) {
int32_t code = 0;
SMsgSendInfo *msgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
if (NULL == msgSendInfo) {
SCH_TASK_ELOG("calloc %d failed", (int32_t)sizeof(SMsgSendInfo));
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
SSchTaskCallbackParam *param = taosMemoryCalloc(1, sizeof(SSchTaskCallbackParam));
if (NULL == param) {
SCH_TASK_ELOG("calloc %d failed", (int32_t)sizeof(SSchTaskCallbackParam));
SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
__async_send_cb_fn_t fp = NULL;
SCH_ERR_JRET(schGetCallbackFp(msgType, &fp));
param->queryId = pJob->queryId;
param->refId = pJob->refId;
param->taskId = SCH_TASK_ID(pTask);
param->transport = pJob->transport;
msgSendInfo->param = param;
msgSendInfo->fp = fp;
*pMsgSendInfo = msgSendInfo;
return TSDB_CODE_SUCCESS;
_return:
taosMemoryFree(param);
taosMemoryFree(msgSendInfo);
SCH_RET(code);
}
void schFreeRpcCtxVal(const void *arg) {
if (NULL == arg) {
return;
......@@ -1352,8 +1533,8 @@ _return:
int32_t schMakeQueryRpcCtx(SSchJob *pJob, SSchTask *pTask, SRpcCtx *pCtx) {
int32_t code = 0;
SSchTaskCallbackParam *param = NULL;
SMsgSendInfo *pMsgSendInfo = NULL;
SMsgSendInfo *pReadyMsgSendInfo = NULL;
SMsgSendInfo *pExplainMsgSendInfo = NULL;
pCtx->args = taosHashInit(1, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);
if (NULL == pCtx->args) {
......@@ -1361,31 +1542,18 @@ int32_t schMakeQueryRpcCtx(SSchJob *pJob, SSchTask *pTask, SRpcCtx *pCtx) {
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
if (NULL == pMsgSendInfo) {
SCH_TASK_ELOG("calloc %d failed", (int32_t)sizeof(SMsgSendInfo));
SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
SCH_ERR_JRET(schGenerateTaskCallBackAHandle(pJob, pTask, TDMT_VND_RES_READY, &pReadyMsgSendInfo));
SCH_ERR_JRET(schGenerateTaskCallBackAHandle(pJob, pTask, TDMT_VND_EXPLAIN, &pExplainMsgSendInfo));
param = taosMemoryCalloc(1, sizeof(SSchTaskCallbackParam));
if (NULL == param) {
SCH_TASK_ELOG("calloc %d failed", (int32_t)sizeof(SSchTaskCallbackParam));
int32_t msgType = TDMT_VND_RES_READY_RSP;
SRpcCtxVal ctxVal = {.val = pReadyMsgSendInfo, .clone = schCloneSMsgSendInfo, .freeFunc = schFreeRpcCtxVal};
if (taosHashPut(pCtx->args, &msgType, sizeof(msgType), &ctxVal, sizeof(ctxVal))) {
SCH_TASK_ELOG("taosHashPut msg %d to rpcCtx failed", msgType);
SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
int32_t msgType = TDMT_VND_RES_READY_RSP;
__async_send_cb_fn_t fp = NULL;
SCH_ERR_JRET(schGetCallbackFp(TDMT_VND_RES_READY, &fp));
param->queryId = pJob->queryId;
param->refId = pJob->refId;
param->taskId = SCH_TASK_ID(pTask);
param->transport = pJob->transport;
pMsgSendInfo->param = param;
pMsgSendInfo->fp = fp;
SRpcCtxVal ctxVal = {.val = pMsgSendInfo, .clone = schCloneSMsgSendInfo, .freeFunc = schFreeRpcCtxVal};
msgType = TDMT_VND_EXPLAIN_RSP;
ctxVal.val = pExplainMsgSendInfo;
if (taosHashPut(pCtx->args, &msgType, sizeof(msgType), &ctxVal, sizeof(ctxVal))) {
SCH_TASK_ELOG("taosHashPut msg %d to rpcCtx failed", msgType);
SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
......@@ -1398,8 +1566,16 @@ int32_t schMakeQueryRpcCtx(SSchJob *pJob, SSchTask *pTask, SRpcCtx *pCtx) {
_return:
taosHashCleanup(pCtx->args);
taosMemoryFreeClear(param);
taosMemoryFreeClear(pMsgSendInfo);
if (pReadyMsgSendInfo) {
taosMemoryFreeClear(pReadyMsgSendInfo->param);
taosMemoryFreeClear(pReadyMsgSendInfo);
}
if (pExplainMsgSendInfo) {
taosMemoryFreeClear(pExplainMsgSendInfo->param);
taosMemoryFreeClear(pExplainMsgSendInfo);
}
SCH_RET(code);
}
......@@ -1582,32 +1758,13 @@ int32_t schAsyncSendMsg(SSchJob *pJob, SSchTask *pTask, void *transport, SEpSet
SSchTrans *trans = (SSchTrans *)transport;
SMsgSendInfo *pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
if (NULL == pMsgSendInfo) {
SCH_TASK_ELOG("calloc %d failed", (int32_t)sizeof(SMsgSendInfo));
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
SSchTaskCallbackParam *param = taosMemoryCalloc(1, sizeof(SSchTaskCallbackParam));
if (NULL == param) {
SCH_TASK_ELOG("calloc %d failed", (int32_t)sizeof(SSchTaskCallbackParam));
SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
__async_send_cb_fn_t fp = NULL;
SCH_ERR_JRET(schGetCallbackFp(msgType, &fp));
param->queryId = pJob->queryId;
param->refId = pJob->refId;
param->taskId = SCH_TASK_ID(pTask);
param->transport = trans->transInst;
SMsgSendInfo *pMsgSendInfo = NULL;
SCH_ERR_JRET(schGenerateTaskCallBackAHandle(pJob, pTask, msgType, &pMsgSendInfo));
pMsgSendInfo->param = param;
pMsgSendInfo->msgInfo.pData = msg;
pMsgSendInfo->msgInfo.len = msgSize;
pMsgSendInfo->msgInfo.handle = trans->transHandle;
pMsgSendInfo->msgType = msgType;
pMsgSendInfo->fp = fp;
qDebug("start to send %s msg to node[%d,%s,%d], refId:%" PRIx64 "instance:%p, handle:%p", TMSG_INFO(msgType),
ntohl(((SMsgHead *)msg)->vgId), epSet->eps[epSet->inUse].fqdn, epSet->eps[epSet->inUse].port, pJob->refId,
......@@ -1624,8 +1781,11 @@ int32_t schAsyncSendMsg(SSchJob *pJob, SSchTask *pTask, void *transport, SEpSet
_return:
taosMemoryFreeClear(param);
taosMemoryFreeClear(pMsgSendInfo);
if (pMsgSendInfo) {
taosMemoryFreeClear(pMsgSendInfo->param);
taosMemoryFreeClear(pMsgSendInfo);
}
SCH_RET(code);
}
......@@ -1767,6 +1927,7 @@ int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr,
pMsg->taskId = htobe64(pTask->taskId);
pMsg->refId = htobe64(pJob->refId);
pMsg->taskType = TASK_TYPE_TEMP;
pMsg->explain = SCH_IS_EXPLAIN_JOB(pJob);
pMsg->phyLen = htonl(pTask->msgLen);
pMsg->sqlLen = htonl(len);
......@@ -2083,6 +2244,8 @@ void schFreeJobImpl(void *job) {
taosArrayDestroy(pJob->levels);
taosArrayDestroy(pJob->nodeList);
qExplainFreeCtx(pJob->explainCtx);
taosMemoryFreeClear(pJob->resData);
taosMemoryFreeClear(pJob);
......@@ -2090,70 +2253,17 @@ void schFreeJobImpl(void *job) {
}
static int32_t schExecJobImpl(void *transport, SArray *pNodeList, SQueryPlan *pDag, int64_t *job, const char *sql,
bool syncSchedule) {
int64_t startTs, bool syncSchedule) {
qDebug("QID:0x%" PRIx64 " job started", pDag->queryId);
if (pNodeList == NULL || (pNodeList && taosArrayGetSize(pNodeList) <= 0)) {
if (pNodeList == NULL || taosArrayGetSize(pNodeList) <= 0) {
qDebug("QID:0x%" PRIx64 " input exec nodeList is empty", pDag->queryId);
}
int32_t code = 0;
SSchJob *pJob = taosMemoryCalloc(1, sizeof(SSchJob));
if (NULL == pJob) {
qError("QID:%" PRIx64 " calloc %d failed", pDag->queryId, (int32_t)sizeof(SSchJob));
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
SSchJob *pJob = NULL;
SCH_ERR_JRET(schInitJob(&pJob, pDag, transport, pNodeList, sql, startTs, syncSchedule));
pJob->attr.explainMode = pDag->explainInfo.mode;
pJob->attr.syncSchedule = syncSchedule;
pJob->transport = transport;
pJob->sql = sql;
if (pNodeList != NULL) {
pJob->nodeList = taosArrayDup(pNodeList);
}
SCH_ERR_JRET(schValidateAndBuildJob(pDag, pJob));
pJob->execTasks =
taosHashInit(pDag->numOfSubplans, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK);
if (NULL == pJob->execTasks) {
SCH_JOB_ELOG("taosHashInit %d execTasks failed", pDag->numOfSubplans);
SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
pJob->succTasks =
taosHashInit(pDag->numOfSubplans, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK);
if (NULL == pJob->succTasks) {
SCH_JOB_ELOG("taosHashInit %d succTasks failed", pDag->numOfSubplans);
SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
pJob->failTasks =
taosHashInit(pDag->numOfSubplans, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK);
if (NULL == pJob->failTasks) {
SCH_JOB_ELOG("taosHashInit %d failTasks failed", pDag->numOfSubplans);
SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
tsem_init(&pJob->rspSem, 0, 0);
int64_t refId = taosAddRef(schMgmt.jobRef, pJob);
if (refId < 0) {
SCH_JOB_ELOG("taosAddRef job failed, error:%s", tstrerror(terrno));
SCH_ERR_JRET(terrno);
}
if (NULL == schAcquireJob(refId)) {
SCH_JOB_ELOG("schAcquireJob job failed, refId:%" PRIx64, refId);
SCH_RET(TSDB_CODE_SCH_STATUS_ERROR);
}
pJob->refId = refId;
SCH_JOB_DLOG("job refId:%" PRIx64, pJob->refId);
pJob->status = JOB_TASK_STATUS_NOT_START;
SCH_ERR_JRET(schLaunchJob(pJob));
*job = pJob->refId;
......@@ -2266,7 +2376,7 @@ int32_t schedulerInit(SSchedulerCfg *cfg) {
}
int32_t schedulerExecJob(void *transport, SArray *nodeList, SQueryPlan *pDag, int64_t *pJob, const char *sql,
SQueryResult *pRes) {
int64_t startTs, SQueryResult *pRes) {
if (NULL == transport || NULL == pDag || NULL == pDag->pSubplans || NULL == pJob || NULL == pRes) {
SCH_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
}
......@@ -2274,7 +2384,7 @@ int32_t schedulerExecJob(void *transport, SArray *nodeList, SQueryPlan *pDag, in
if (EXPLAIN_MODE_STATIC == pDag->explainInfo.mode) {
SCH_ERR_RET(schExecStaticExplain(transport, nodeList, pDag, pJob, sql, true));
} else {
SCH_ERR_RET(schExecJobImpl(transport, nodeList, pDag, pJob, sql, true));
SCH_ERR_RET(schExecJobImpl(transport, nodeList, pDag, pJob, sql, startTs, true));
}
SSchJob *job = schAcquireJob(*pJob);
......@@ -2292,7 +2402,11 @@ int32_t schedulerAsyncExecJob(void *transport, SArray *pNodeList, SQueryPlan *pD
SCH_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
}
SCH_ERR_RET(schExecJobImpl(transport, pNodeList, pDag, pJob, sql, false));
if (EXPLAIN_MODE_STATIC == pDag->explainInfo.mode) {
SCH_ERR_RET(schExecStaticExplain(transport, pNodeList, pDag, pJob, sql, false));
} else {
SCH_ERR_RET(schExecJobImpl(transport, pNodeList, pDag, pJob, sql, 0, false));
}
return TSDB_CODE_SUCCESS;
}
......
source/libs/scheduler/test/schedulerTests.cpp
浏览文件 @ 6f26d177
......@@ -951,7 +951,7 @@ TEST(insertTest, normalCase) {
taosThreadCreate(&(thread1), &thattr, schtSendRsp, &insertJobRefId);
SQueryResult res = {0};
code = schedulerExecJob(mockPointer, qnodeList, &dag, &insertJobRefId, "insert into tb values(now,1)", &res);
code = schedulerExecJob(mockPointer, qnodeList, &dag, &insertJobRefId, "insert into tb values(now,1)", 0, &res);
ASSERT_EQ(code, 0);
ASSERT_EQ(res.numOfRows, 20);
......
source/util/src/terror.c
浏览文件 @ 6f26d177
......@@ -432,6 +432,9 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SCH_STATUS_ERROR, "scheduler status erro
TAOS_DEFINE_ERROR(TSDB_CODE_SCH_INTERNAL_ERROR, "scheduler internal error")
TAOS_DEFINE_ERROR(TSDB_CODE_QW_MSG_ERROR, "Invalid msg order")
//planner
TAOS_DEFINE_ERROR(TSDB_CODE_PLAN_INTERNAL_ERROR, "planner internal error")
#ifdef TAOS_ERROR_C
};
#endif
......
source/util/src/tlockfree.c
浏览文件 @ 6f26d177
......@@ -53,6 +53,21 @@ void taosWLockLatch(SRWLatch *pLatch) {
}
}
int32_t taosWTryLockLatch(SRWLatch *pLatch) {
SRWLatch oLatch, nLatch;
oLatch = atomic_load_32(pLatch);
if (oLatch) {
return -1;
}
nLatch = oLatch | TD_RWLATCH_WRITE_FLAG;
if (atomic_val_compare_exchange_32(pLatch, oLatch, nLatch) == oLatch) {
return 0;
}
return -1;
}
void taosWUnLockLatch(SRWLatch *pLatch) { atomic_store_32(pLatch, 0); }
void taosRLockLatch(SRWLatch *pLatch) {
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或