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

[td-11818] refactor.

This commit is contained in:
Haojun Liao 2022-01-20 16:02:09 +08:00
parent 1a1cd55111 7fba094c50
commit b45da2b208
45 changed files with 2072 additions and 805 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
include/common/tmsg.h
View File

@ -1089,6 +1089,10 @@ static FORCE_INLINE void* tDeserializeSMVSubscribeReq(void* buf, SMVSubscribeReq
return buf;
}
typedef struct SMqTmrMsg {
int32_t reserved;
} SMqTmrMsg;
typedef struct {
int64_t status;
} SMVSubscribeRsp;
@ -1548,6 +1552,25 @@ static FORCE_INLINE void* tDecodeSMqSetCVgReq(void* buf, SMqSetCVgReq* pReq) {
return buf;
}
typedef struct SMqCVConsumeReq {
int64_t reqId;
int64_t offset;
int64_t clientId;
char topicName[TSDB_TOPIC_FNAME_LEN];
char cgroup[TSDB_CONSUMER_GROUP_LEN];
} SMqCVConsumeReq;
typedef struct SMqCVConsumeRsp {
int64_t reqId;
int64_t clientId;
int64_t committedOffset;
int64_t receiveOffset;
int64_t rspOffset;
int32_t skipLogNum;
int32_t bodyLen;
char topicName[TSDB_TOPIC_FNAME_LEN];
char body[];
} SMqCvConsumeRsp;
#ifdef __cplusplus
}

2
include/common/tmsgdef.h
View File

@ -140,6 +140,7 @@ enum {
TD_DEF_MSG_TYPE(TDMT_MND_ALTER_TOPIC, "mnode-alter-topic", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_DROP_TOPIC, "mnode-drop-topic", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_SUBSCRIBE, "mnode-subscribe", SCMSubscribeReq, SCMSubscribeRsp)
TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mnode-timer", SMqTmrMsg, SMqTmrMsg)
// Requests handled by VNODE
TD_NEW_MSG_SEG(TDMT_VND_MSG)
@ -174,6 +175,7 @@ enum {
TD_DEF_MSG_TYPE(TDMT_VND_SCHEDULE_DATA_SINK, "vnode-schedule-data-sink", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_SUBSCRIBE, "vnode-subscribe", SMVSubscribeReq, SMVSubscribeRsp)
TD_DEF_MSG_TYPE(TDMT_VND_CONSUME, "vnode-consume", SMqCVConsumeReq, SMqCVConsumeRsp)
// Requests handled by QNODE

2
include/libs/function/function.h
View File

@ -134,6 +134,8 @@ typedef struct SFunctionFpSet {
void (*combine)(struct SQLFunctionCtx *pCtx);
} SFunctionFpSet;
extern SFunctionFpSet fpSet[1];
// sql function runtime context
typedef struct SQLFunctionCtx {
int32_t size; // number of rows

66
include/libs/function/functionMgt.h Normal file
View File

@ -0,0 +1,66 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_FUNCTION_MGT_H_
#define _TD_FUNCTION_MGT_H_
#ifdef __cplusplus
extern "C" {
#endif
#include "nodes.h"
struct SQLFunctionCtx;
struct SResultRowEntryInfo;
struct STimeWindow;
typedef struct SFuncExecEnv {
int32_t calcMemSize;
} SFuncExecEnv;
typedef void* FuncMgtHandle;
typedef bool (*FExecGetEnv)(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
typedef bool (*FExecInit)(struct SQLFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo);
typedef void (*FExecProcess)(struct SQLFunctionCtx *pCtx);
typedef void (*FExecFinalize)(struct SQLFunctionCtx *pCtx);
typedef struct SFuncExecFuncs {
FExecGetEnv getEnv;
FExecInit init;
FExecProcess process;
FExecFinalize finalize;
} SFuncExecFuncs;
int32_t fmFuncMgtInit();
int32_t fmGetHandle(FuncMgtHandle* pHandle);
int32_t fmGetFuncId(FuncMgtHandle handle, const char* name);
int32_t fmGetFuncResultType(FuncMgtHandle handle, SFunctionNode* pFunc);
bool fmIsAggFunc(int32_t funcId);
bool fmIsStringFunc(int32_t funcId);
bool fmIsTimestampFunc(int32_t funcId);
bool fmIsTimelineFunc(int32_t funcId);
bool fmIsTimeorderFunc(int32_t funcId);
bool fmIsNonstandardSQLFunc(int32_t funcId);
int32_t fmFuncScanType(int32_t funcId);
int32_t fmGetFuncExecFuncs(FuncMgtHandle handle, int32_t funcId, SFuncExecFuncs* pFpSet);
#ifdef __cplusplus
}
#endif
#endif // _TD_FUNCTION_MGT_H_

1
include/libs/parser/parser.h
View File

@ -91,6 +91,7 @@ SSchema *getOneColumnSchema(const STableMeta* pTableMeta, int32_t colIndex);
int32_t getNewResColId();
void addIntoSourceParam(SSourceParam* pSourceParam, tExprNode* pNode, SColumn* pColumn);
SExprInfo* createBinaryExprInfo(struct tExprNode* pNode, SSchema* pResSchema);
#ifdef __cplusplus
}

271
include/nodes/nodes.h Normal file
View File

@ -0,0 +1,271 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_NODES_H_
#define _TD_NODES_H_
#ifdef __cplusplus
extern "C" {
#endif
#include "tarray.h"
#include "tdef.h"
typedef enum ENodeType {
QUERY_NODE_COLUMN = 1,
QUERY_NODE_VALUE,
QUERY_NODE_OPERATOR,
QUERY_NODE_LOGIC_CONDITION,
QUERY_NODE_IS_NULL_CONDITION,
QUERY_NODE_FUNCTION,
QUERY_NODE_REAL_TABLE,
QUERY_NODE_TEMP_TABLE,
QUERY_NODE_JOIN_TABLE,
QUERY_NODE_GROUPING_SET,
QUERY_NODE_ORDER_BY_EXPR,
QUERY_NODE_STATE_WINDOW,
QUERY_NODE_SESSION_WINDOW,
QUERY_NODE_INTERVAL_WINDOW,
QUERY_NODE_SET_OPERATOR,
QUERY_NODE_SELECT_STMT
} ENodeType;
/**
* The first field of a node of any type is guaranteed to be the ENodeType.
* Hence the type of any node can be gotten by casting it to SNode.
*/
typedef struct SNode {
ENodeType type;
} SNode;
#define nodeType(nodeptr) (((const SNode*)(nodeptr))->type)
typedef struct SDataType {
uint8_t type;
uint8_t precision;
uint8_t scale;
int32_t bytes;
} SDataType;
typedef struct SExprNode {
ENodeType nodeType;
SDataType resType;
char aliasName[TSDB_COL_NAME_LEN];
} SExprNode;
typedef enum EColumnType {
COLUMN_TYPE_COLUMN = 1,
COLUMN_TYPE_TAG
} EColumnType;
typedef struct SColumnNode {
SExprNode node; // QUERY_NODE_COLUMN
int16_t colId;
EColumnType colType; // column or tag
char dbName[TSDB_DB_NAME_LEN];
char tableName[TSDB_TABLE_NAME_LEN];
char colName[TSDB_COL_NAME_LEN];
} SColumnNode;
typedef struct SValueNode {
SExprNode type; // QUERY_NODE_VALUE
char* literal;
} SValueNode;
typedef enum EOperatorType {
// arithmetic operator
OP_TYPE_ADD = 1,
OP_TYPE_SUB,
OP_TYPE_MULTI,
OP_TYPE_DIV,
OP_TYPE_MOD,
// comparison operator
OP_TYPE_GREATER_THAN,
OP_TYPE_GREATER_EQUAL,
OP_TYPE_LOWER_THAN,
OP_TYPE_LOWER_EQUAL,
OP_TYPE_EQUAL,
OP_TYPE_NOT_EQUAL,
OP_TYPE_IN,
OP_TYPE_NOT_IN,
OP_TYPE_LIKE,
OP_TYPE_NOT_LIKE,
OP_TYPE_MATCH,
OP_TYPE_NMATCH,
// json operator
OP_TYPE_JSON_GET_VALUE,
OP_TYPE_JSON_CONTAINS
} EOperatorType;
typedef struct SOperatorNode {
SExprNode type; // QUERY_NODE_OPERATOR
EOperatorType opType;
SNode* pLeft;
SNode* pRight;
} SOperatorNode;
typedef enum ELogicConditionType {
LOGIC_COND_TYPE_AND,
LOGIC_COND_TYPE_OR,
LOGIC_COND_TYPE_NOT,
} ELogicConditionType;
typedef struct SLogicConditionNode {
ENodeType type; // QUERY_NODE_LOGIC_CONDITION
ELogicConditionType condType;
SArray* pParameterList;
} SLogicConditionNode;
typedef struct SIsNullCondNode {
ENodeType type; // QUERY_NODE_IS_NULL_CONDITION
SNode* pExpr;
bool isNot;
} SIsNullCondNode;
typedef struct SFunctionNode {
SExprNode type; // QUERY_NODE_FUNCTION
char functionName[TSDB_FUNC_NAME_LEN];
int32_t funcId;
SArray* pParameterList; // SNode
} SFunctionNode;
typedef struct STableNode {
ENodeType type;
char tableName[TSDB_TABLE_NAME_LEN];
char tableAliasName[TSDB_COL_NAME_LEN];
} STableNode;
typedef struct SRealTableNode {
STableNode type; // QUERY_NODE_REAL_TABLE
char dbName[TSDB_DB_NAME_LEN];
} SRealTableNode;
typedef struct STempTableNode {
STableNode type; // QUERY_NODE_TEMP_TABLE
SNode* pSubquery;
} STempTableNode;
typedef enum EJoinType {
JOIN_TYPE_INNER = 1
} EJoinType;
typedef struct SJoinTableNode {
STableNode type; // QUERY_NODE_JOIN_TABLE
EJoinType joinType;
SNode* pLeft;
SNode* pRight;
SNode* pOnCond;
} SJoinTableNode;
typedef enum EGroupingSetType {
GP_TYPE_NORMAL = 1
} EGroupingSetType;
typedef struct SGroupingSetNode {
ENodeType type; // QUERY_NODE_GROUPING_SET
EGroupingSetType groupingSetType;
SArray* pParameterList;
} SGroupingSetNode;
typedef enum EOrder {
ORDER_ASC = 1,
ORDER_DESC
} EOrder;
typedef enum ENullOrder {
NULL_ORDER_FIRST = 1,
NULL_ORDER_LAST
} ENullOrder;
typedef struct SOrderByExprNode {
ENodeType type; // QUERY_NODE_ORDER_BY_EXPR
SNode* pExpr;
EOrder order;
ENullOrder nullOrder;
} SOrderByExprNode;
typedef struct SLimitInfo {
uint64_t limit;
uint64_t offset;
} SLimitInfo;
typedef struct SStateWindowNode {
ENodeType type; // QUERY_NODE_STATE_WINDOW
SNode* pCol;
} SStateWindowNode;
typedef struct SSessionWindowNode {
ENodeType type; // QUERY_NODE_SESSION_WINDOW
int64_t gap; // gap between two session window(in microseconds)
SNode* pCol;
} SSessionWindowNode;
typedef struct SIntervalWindowNode {
ENodeType type; // QUERY_NODE_INTERVAL_WINDOW
int64_t interval;
int64_t sliding;
int64_t offset;
} SIntervalWindowNode;
typedef struct SSelectStmt {
ENodeType type; // QUERY_NODE_SELECT_STMT
bool isDistinct;
SArray* pProjectionList; // SNode
SNode* pFromTable;
SNode* pWhereCond;
SArray* pPartitionByList; // SNode
SNode* pWindowClause;
SArray* pGroupByList; // SGroupingSetNode
SArray* pOrderByList; // SOrderByExprNode
SLimitInfo limit;
SLimitInfo slimit;
} SSelectStmt;
typedef enum ESetOperatorType {
SET_OP_TYPE_UNION_ALL = 1
} ESetOperatorType;
typedef struct SSetOperator {
ENodeType type; // QUERY_NODE_SET_OPERATOR
ESetOperatorType opType;
SNode* pLeft;
SNode* pRight;
} SSetOperator;
typedef bool (*FQueryNodeWalker)(SNode* pNode, void* pContext);
bool nodeArrayWalker(SArray* pArray, FQueryNodeWalker walker, void* pContext);
bool nodeTreeWalker(SNode* pNode, FQueryNodeWalker walker, void* pContext);
bool stmtWalker(SNode* pNode, FQueryNodeWalker walker, void* pContext);
bool nodeEqual(const SNode* a, const SNode* b);
void cloneNode(const SNode* pNode);
int32_t nodeToString(const SNode* pNode, char** pStr, int32_t* pLen);
int32_t stringToNode(const char* pStr, SNode** pNode);
bool isTimeorderQuery(const SNode* pQuery);
bool isTimelineQuery(const SNode* pQuery);
#ifdef __cplusplus
}
#endif
#endif /*_TD_NODES_H_*/

3
source/CMakeLists.txt
View File

@ -3,4 +3,5 @@ add_subdirectory(util)
add_subdirectory(common)
add_subdirectory(libs)
add_subdirectory(client)
add_subdirectory(dnode)
add_subdirectory(dnode)
add_subdirectory(nodes)

1166
source/client/test/clientTests.cpp
View File

File diff suppressed because it is too large Load Diff

2
source/dnode/mgmt/impl/src/dndTransport.c
View File

@ -100,7 +100,7 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_VGROUP_LIST)] = dndProcessMnodeReadMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_KILL_QUERY)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_KILL_CONN)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_HEARTBEAT)] = dndProcessMnodeReadMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_HEARTBEAT)] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_SHOW)] = dndProcessMnodeReadMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_SHOW_RETRIEVE)] = dndProcessMnodeReadMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_MND_STATUS)] = dndProcessMnodeReadMsg;

1
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -354,6 +354,7 @@ typedef struct SMqSubscribeObj {
char key[TSDB_SUBSCRIBE_KEY_LEN];
int32_t epoch;
//TODO: replace with priority queue
int32_t nextConsumerIdx;
SArray* availConsumer; // SArray<int64_t> (consumerId)
SArray* assigned; // SArray<SMqConsumerEp>
SArray* unassignedConsumer; // SArray<SMqConsumerEp>

1
source/dnode/mnode/impl/inc/mndInt.h
View File

@ -80,6 +80,7 @@ typedef struct SMnode {
SReplica replicas[TSDB_MAX_REPLICA];
tmr_h timer;
tmr_h transTimer;
tmr_h mqTimer;
char *path;
SMnodeCfg cfg;
int64_t checkTime;

119
source/dnode/mnode/impl/src/mndSubscribe.c
View File

@ -13,13 +13,13 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "mndSubscribe.h"
#include "mndConsumer.h"
#include "mndDb.h"
#include "mndDnode.h"
#include "mndMnode.h"
#include "mndShow.h"
#include "mndStb.h"
#include "mndSubscribe.h"
#include "mndTopic.h"
#include "mndTrans.h"
#include "mndUser.h"
@ -40,6 +40,10 @@ static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg);
static int32_t mndProcessSubscribeRsp(SMnodeMsg *pMsg);
static int32_t mndProcessSubscribeInternalReq(SMnodeMsg *pMsg);
static int32_t mndProcessSubscribeInternalRsp(SMnodeMsg *pMsg);
static int32_t mndProcessMqTimerMsg(SMnodeMsg *pMsg);
static int mndBuildMqSetConsumerVgReq(SMnode *pMnode, STrans *pTrans, SMqConsumerObj *pConsumer,
SMqConsumerTopic *pConsumerTopic, SMqTopicObj *pTopic);
int32_t mndInitSubscribe(SMnode *pMnode) {
SSdbTable table = {.sdbType = SDB_SUBSCRIBE,
@ -54,9 +58,90 @@ int32_t mndInitSubscribe(SMnode *pMnode) {
/*mndSetMsgHandle(pMnode, TDMT_MND_SUBSCRIBE_RSP, mndProcessSubscribeRsp);*/
/*mndSetMsgHandle(pMnode, TDMT_VND_SUBSCRIBE, mndProcessSubscribeInternalReq);*/
mndSetMsgHandle(pMnode, TDMT_VND_SUBSCRIBE_RSP, mndProcessSubscribeInternalRsp);
mndSetMsgHandle(pMnode, TDMT_MND_MQ_TIMER, mndProcessMqTimerMsg);
return sdbSetTable(pMnode->pSdb, table);
}
static int32_t mndSplitSubscribeKey(char *key, char **topic, char **cgroup) {
int i = 0;
while (key[i] != ':') {
i++;
}
key[i] = 0;
*topic = strdup(key);
key[i] = ':';
*cgroup = strdup(&key[i + 1]);
return 0;
}
static int32_t mndProcessMqTimerMsg(SMnodeMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;
SSdb *pSdb = pMnode->pSdb;
SMqSubscribeObj *pSub = NULL;
void *pIter = sdbFetch(pSdb, SDB_SUBSCRIBE, NULL, (void **)&pSub);
int sz;
while (pIter != NULL) {
if ((sz = taosArrayGetSize(pSub->unassignedVg)) > 0) {
char *topic = NULL;
char *cgroup = NULL;
mndSplitSubscribeKey(pSub->key, &topic, &cgroup);
SMqTopicObj *pTopic = mndAcquireTopic(pMnode, topic);
// create trans
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, &pMsg->rpcMsg);
for (int i = 0; i < sz; i++) {
int64_t consumerId = *(int64_t *)taosArrayGet(pSub->availConsumer, pSub->nextConsumerIdx);
SMqConsumerEp *pCEp = taosArrayPop(pSub->unassignedVg);
pCEp->consumerId = consumerId;
taosArrayPush(pSub->assigned, pCEp);
pSub->nextConsumerIdx++;
// build msg
SMqSetCVgReq req = {
.vgId = pCEp->vgId,
.consumerId = consumerId,
};
strcpy(req.cGroup, cgroup);
strcpy(req.topicName, topic);
strcpy(req.sql, pTopic->sql);
strcpy(req.logicalPlan, pTopic->logicalPlan);
strcpy(req.physicalPlan, pTopic->physicalPlan);
int32_t tlen = tEncodeSMqSetCVgReq(NULL, &req);
void *reqStr = malloc(tlen);
if (reqStr == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
void *abuf = reqStr;
tEncodeSMqSetCVgReq(abuf, &req);
// persist msg
STransAction action = {0};
action.epSet = pCEp->epset;
action.pCont = reqStr;
action.contLen = tlen;
action.msgType = TDMT_VND_MQ_SET_CONN;
mndTransAppendRedoAction(pTrans, &action);
// persist raw
SSdbRaw *pRaw = mndSubActionEncode(pSub);
mndTransAppendRedolog(pTrans, pRaw);
tfree(topic);
tfree(cgroup);
}
if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
}
mndReleaseTopic(pMnode, pTopic);
mndTransDrop(pTrans);
}
pIter = sdbFetch(pSdb, SDB_SUBSCRIBE, NULL, (void **)&pSub);
}
return 0;
}
static int mndInitUnassignedVg(SMnode *pMnode, SMqTopicObj *pTopic, SArray *unassignedVg) {
SMqConsumerEp CEp;
CEp.lastConsumerHbTs = CEp.lastVgHbTs = -1;
@ -76,7 +161,7 @@ static int mndInitUnassignedVg(SMnode *pMnode, SMqTopicObj *pTopic, SArray *unas
}
static int mndBuildMqSetConsumerVgReq(SMnode *pMnode, STrans *pTrans, SMqConsumerObj *pConsumer,
SMqConsumerTopic *pConsumerTopic) {
SMqConsumerTopic *pConsumerTopic, SMqTopicObj *pTopic) {
int32_t sz = taosArrayGetSize(pConsumerTopic->pVgInfo);
for (int32_t i = 0; i < sz; i++) {
int32_t vgId = *(int32_t *)taosArrayGet(pConsumerTopic->pVgInfo, i);
@ -86,7 +171,10 @@ static int mndBuildMqSetConsumerVgReq(SMnode *pMnode, STrans *pTrans, SMqConsume
.consumerId = pConsumer->consumerId,
};
strcpy(req.cGroup, pConsumer->cgroup);
strcpy(req.topicName, pConsumerTopic->name);
strcpy(req.topicName, pTopic->name);
strcpy(req.sql, pTopic->sql);
strcpy(req.logicalPlan, pTopic->logicalPlan);
strcpy(req.physicalPlan, pTopic->physicalPlan);
int32_t tlen = tEncodeSMqSetCVgReq(NULL, &req);
void *reqStr = malloc(tlen);
if (reqStr == NULL) {
@ -94,7 +182,7 @@ static int mndBuildMqSetConsumerVgReq(SMnode *pMnode, STrans *pTrans, SMqConsume
return -1;
}
void *abuf = reqStr;
tEncodeSMqSetCVgReq(abuf, &req);
tEncodeSMqSetCVgReq(&abuf, &req);
STransAction action = {0};
action.epSet = mndGetVgroupEpset(pMnode, pVgObj);
@ -278,20 +366,20 @@ static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) {
int i = 0, j = 0;
while (i < newTopicNum || j < oldTopicNum) {
char* newTopicName = NULL;
char* oldTopicName = NULL;
char *newTopicName = NULL;
char *oldTopicName = NULL;
if (i >= newTopicNum) {
// encode unset topic msg to all vnodes related to that topic
oldTopicName = ((SMqConsumerTopic*)taosArrayGet(oldSub, j))->name;
oldTopicName = ((SMqConsumerTopic *)taosArrayGet(oldSub, j))->name;
j++;
} else if (j >= oldTopicNum) {
newTopicName = taosArrayGet(newSub, i);
i++;
} else {
newTopicName = taosArrayGet(newSub, i);
oldTopicName = ((SMqConsumerTopic*)taosArrayGet(oldSub, j))->name;
oldTopicName = ((SMqConsumerTopic *)taosArrayGet(oldSub, j))->name;
int comp = compareLenPrefixedStr(newTopicName, oldTopicName);
int comp = compareLenPrefixedStr(newTopicName, oldTopicName);
if (comp == 0) {
// do nothing
oldTopicName = newTopicName = NULL;
@ -374,19 +462,25 @@ static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) {
// set unassigned vg
mndInitUnassignedVg(pMnode, pTopic, pSub->unassignedVg);
}
taosArrayPush(pSub->availConsumer, &consumerId);
// TODO: no need
SMqConsumerTopic *pConsumerTopic = tNewConsumerTopic(consumerId, pTopic, pSub);
taosArrayPush(pConsumer->topics, pConsumerTopic);
if (taosArrayGetSize(pConsumerTopic->pVgInfo) > 0) {
int32_t vgId = *(int32_t *)taosArrayGetLast(pConsumerTopic->pVgInfo);
// send setmsg to vnode
if (mndBuildMqSetConsumerVgReq(pMnode, pTrans, pConsumer, pConsumerTopic) < 0) {
if (mndBuildMqSetConsumerVgReq(pMnode, pTrans, pConsumer, pConsumerTopic, pTopic) < 0) {
// TODO
return -1;
}
}
taosArrayDestroy(pConsumerTopic->pVgInfo);
free(pConsumerTopic);
SSdbRaw *pRaw = mndSubActionEncode(pSub);
/*sdbSetRawStatus(pRaw, SDB_STATUS_READY);*/
mndTransAppendRedolog(pTrans, pRaw);
#if 0
SMqCGroup *pGroup = taosHashGet(pTopic->cgroups, consumerGroup, cgroupLen);
if (pGroup == NULL) {
@ -448,7 +542,10 @@ static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) {
return 0;
}
static int32_t mndProcessSubscribeInternalRsp(SMnodeMsg *pMsg) { return 0; }
static int32_t mndProcessSubscribeInternalRsp(SMnodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
static int32_t mndProcessConsumerMetaMsg(SMnodeMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;

25
source/dnode/mnode/impl/src/mnode.c
View File

@ -18,6 +18,7 @@
#include "mndAuth.h"
#include "mndBnode.h"
#include "mndCluster.h"
#include "mndConsumer.h"
#include "mndDb.h"
#include "mndDnode.h"
#include "mndFunc.h"
@ -27,6 +28,7 @@
#include "mndShow.h"
#include "mndSnode.h"
#include "mndStb.h"
#include "mndSubscribe.h"
#include "mndSync.h"
#include "mndTelem.h"
#include "mndTopic.h"
@ -69,15 +71,15 @@ static void mndTransReExecute(void *param, void *tmrId) {
taosTmrReset(mndTransReExecute, 3000, pMnode, pMnode->timer, &pMnode->transTimer);
}
static void mndCalMqRebalance(void* param, void* tmrId) {
SMnode* pMnode = param;
static void mndCalMqRebalance(void *param, void *tmrId) {
SMnode *pMnode = param;
if (mndIsMaster(pMnode)) {
// iterate cgroup, cal rebalance
// sync with raft
// write sdb
SMqTmrMsg *pMsg = rpcMallocCont(sizeof(SMqTmrMsg));
SRpcMsg rpcMsg = {.msgType = TDMT_MND_MQ_TIMER, .pCont = pMsg, .contLen = sizeof(SMqTmrMsg)};
pMnode->putReqToMWriteQFp(pMnode->pDnode, &rpcMsg);
}
taosTmrReset(mndCalMqRebalance, 3000, pMnode, pMnode->timer, &pMnode->transTimer);
taosTmrReset(mndCalMqRebalance, 3000, pMnode, pMnode->timer, &pMnode->mqTimer);
}
static int32_t mndInitTimer(SMnode *pMnode) {
@ -95,6 +97,11 @@ static int32_t mndInitTimer(SMnode *pMnode) {
return -1;
}
if (taosTmrReset(mndCalMqRebalance, 3000, pMnode, pMnode->timer, &pMnode->mqTimer)) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
return 0;
}
@ -102,6 +109,8 @@ static void mndCleanupTimer(SMnode *pMnode) {
if (pMnode->timer != NULL) {
taosTmrStop(pMnode->transTimer);
pMnode->transTimer = NULL;
taosTmrStop(pMnode->mqTimer);
pMnode->mqTimer = NULL;
taosTmrCleanUp(pMnode->timer);
pMnode->timer = NULL;
}
@ -171,6 +180,8 @@ static int32_t mndInitSteps(SMnode *pMnode) {
if (mndAllocStep(pMnode, "mnode-auth", mndInitAuth, mndCleanupAuth) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-acct", mndInitAcct, mndCleanupAcct) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-topic", mndInitTopic, mndCleanupTopic) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-consumer", mndInitConsumer, mndCleanupConsumer) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-subscribe", mndInitSubscribe, mndCleanupSubscribe) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-vgroup", mndInitVgroup, mndCleanupVgroup) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-stb", mndInitStb, mndCleanupStb) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-db", mndInitDb, mndCleanupDb) != 0) return -1;
@ -377,7 +388,7 @@ SMnodeMsg *mndInitMsg(SMnode *pMnode, SRpcMsg *pRpcMsg) {
return NULL;
}
if (pRpcMsg->msgType != TDMT_MND_TRANS) {
if (pRpcMsg->msgType != TDMT_MND_TRANS && pRpcMsg->msgType != TDMT_MND_MQ_TIMER) {
SRpcConnInfo connInfo = {0};
if ((pRpcMsg->msgType & 1U) && rpcGetConnInfo(pRpcMsg->handle, &connInfo) != 0) {
taosFreeQitem(pMsg);

1
source/dnode/vnode/CMakeLists.txt
View File

@ -25,6 +25,7 @@ target_link_libraries(
PUBLIC bdb
PUBLIC tfs
PUBLIC wal
PUBLIC scheduler
PUBLIC qworker
)

60
source/dnode/vnode/inc/tq.h
View File

@ -18,14 +18,15 @@
#include "common.h"
#include "mallocator.h"
#include "meta.h"
#include "os.h"
#include "scheduler.h"
#include "taoserror.h"
#include "tmsg.h"
#include "tlist.h"
#include "tmsg.h"
#include "trpc.h"
#include "ttimer.h"
#include "tutil.h"
#include "meta.h"
#ifdef __cplusplus
extern "C" {
@ -150,15 +151,40 @@ typedef struct STqListHandle {
} STqList;
typedef struct STqGroup {
int64_t clientId;
int64_t cgId;
void* ahandle;
int32_t topicNum;
int64_t clientId;
int64_t cgId;
void* ahandle;
int32_t topicNum;
STqList* head;
SList* topicList; // SList<STqTopic>
STqRspHandle rspHandle;
} STqGroup;
typedef struct STqTaskItem {
int32_t status;
void* dst;
SSubQueryMsg* pMsg;
} STqTaskItem;
// new version
typedef struct STqBuffer {
int64_t firstOffset;
int64_t lastOffset;
STqTaskItem output[TQ_BUFFER_SIZE];
} STqBuffer;
typedef struct STqClientHandle {
int64_t clientId;
char topicName[TSDB_TOPIC_FNAME_LEN];
char cGroup[TSDB_TOPIC_FNAME_LEN];
char* sql;
char* logicalPlan;
char* physicalPlan;
int64_t committedOffset;
int64_t currentOffset;
STqBuffer buffer;
} STqClientHandle;
typedef struct STqQueryMsg {
STqMsgItem* item;
struct STqQueryMsg* next;
@ -253,7 +279,7 @@ typedef struct STqMetaStore {
// a table head
STqMetaList* unpersistHead;
// topics that are not connectted
STqMetaList* unconnectTopic;
STqMetaList* unconnectTopic;
// TODO:temporaral use, to be replaced by unified tfile
int fileFd;
@ -316,24 +342,22 @@ const void* tqDeserializeGroup(const STqSerializedHead*, STqGroup**);
static int tqQueryExecuting(int32_t status) { return status; }
typedef struct STqReadHandle {
int64_t ver;
SSubmitMsg* pMsg;
SSubmitBlk* pBlock;
int64_t ver;
SSubmitMsg* pMsg;
SSubmitBlk* pBlock;
SSubmitMsgIter msgIter;
SSubmitBlkIter blkIter;
SMeta* pMeta;
SMeta* pMeta;
} STqReadHandle;
typedef struct SSubmitBlkScanInfo {
} SSubmitBlkScanInfo;
STqReadHandle* tqInitSubmitMsgScanner(SMeta* pMeta, SSubmitMsg *pMsg);
bool tqNextDataBlock(STqReadHandle* pHandle);
int tqRetrieveDataBlockInfo(STqReadHandle* pHandle, SDataBlockInfo *pBlockInfo);
//return SArray<SColumnInfoData>
SArray *tqRetrieveDataBlock(STqReadHandle* pHandle, SArray* pColumnIdList);
//int tqLoadDataBlock(SExecTaskInfo* pTaskInfo, SSubmitBlkScanInfo* pSubmitBlkScanInfo, SSDataBlock* pBlock, uint32_t status);
STqReadHandle* tqInitSubmitMsgScanner(SMeta* pMeta, SSubmitMsg* pMsg);
bool tqNextDataBlock(STqReadHandle* pHandle);
int tqRetrieveDataBlockInfo(STqReadHandle* pHandle, SDataBlockInfo* pBlockInfo);
// return SArray<SColumnInfoData>
SArray* tqRetrieveDataBlock(STqReadHandle* pHandle, SArray* pColumnIdList);
#ifdef __cplusplus
}

2
source/dnode/vnode/src/inc/vnd.h
View File

@ -177,4 +177,4 @@ bool vmaIsFull(SVMemAllocator* pVMA);
}
#endif
#endif /*_TD_VNODE_DEF_H_*/
#endif /*_TD_VNODE_DEF_H_*/

2
source/dnode/vnode/src/vnd/vnodeQuery.c
View File

@ -57,6 +57,8 @@ int vnodeProcessFetchReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
// return qWorkerProcessShowFetchMsg(pVnode->pMeta, pVnode->pQuery, pMsg);
case TDMT_VND_TABLE_META:
return vnodeGetTableMeta(pVnode, pMsg, pRsp);
case TDMT_VND_CONSUME:
return 0;
default:
vError("unknown msg type:%d in fetch queue", pMsg->msgType);
return TSDB_CODE_VND_APP_ERROR;

35
source/dnode/vnode/src/vnd/vnodeWrite.c
View File

@ -14,6 +14,7 @@
*/
#include "vnd.h"
#include "tq.h"
int vnodeProcessNoWalWMsgs(SVnode *pVnode, SRpcMsg *pMsg) {
switch (pMsg->msgType) {
@ -108,6 +109,40 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
// TODO: handle error
}
break;
case TDMT_VND_MQ_SET_CONN: {
char* reqStr = ptr;
SMqSetCVgReq req;
tDecodeSMqSetCVgReq(reqStr, &req);
STqClientHandle* pHandle = calloc(sizeof(STqClientHandle), 1);
if (pHandle == NULL) {
// TODO: handle error
}
strcpy(pHandle->topicName, req.topicName);
strcpy(pHandle->cGroup, req.cGroup);
strcpy(pHandle->sql, req.sql);
strcpy(pHandle->logicalPlan, req.logicalPlan);
strcpy(pHandle->physicalPlan, req.physicalPlan);
SArray *pArray;
//TODO: deserialize to SQueryDag
SQueryDag *pDag;
// convert to task
if (schedulerConvertDagToTaskList(pDag, &pArray) < 0) {
// TODO: handle error
}
ASSERT(taosArrayGetSize(pArray) == 0);
STaskInfo *pInfo = taosArrayGet(pArray, 0);
SArray* pTasks;
schedulerCopyTask(pInfo, &pTasks, TQ_BUFFER_SIZE);
pHandle->buffer.firstOffset = -1;
pHandle->buffer.lastOffset = -1;
for (int i = 0; i < TQ_BUFFER_SIZE; i++) {
SSubQueryMsg* pMsg = taosArrayGet(pTasks, i);
pHandle->buffer.output[i].pMsg = pMsg;
pHandle->buffer.output[i].status = 0;
}
// write mq meta
}
break;
default:
ASSERT(0);
break;

9
source/libs/executor/inc/executorimpl.h
View File

@ -425,14 +425,15 @@ typedef struct SOptrBasicInfo {
typedef struct SOptrBasicInfo STableIntervalOperatorInfo;
typedef struct SAggOperatorInfo {
SOptrBasicInfo binfo;
uint32_t seed;
SDiskbasedResultBuf* pResultBuf; // query result buffer based on blocked-wised disk file
SOptrBasicInfo binfo;
uint32_t seed;
SDiskbasedResultBuf *pResultBuf; // query result buffer based on blocked-wised disk file
SHashObj* pResultRowHashTable; // quick locate the window object for each result
SHashObj* pResultRowListSet; // used to check if current ResultRowInfo has ResultRow object or not
SArray* pResultRowArrayList; // The array list that contains the Result rows
char* keyBuf; // window key buffer
SResultRowPool* pool; // The window result objects pool, all the resultRow Objects are allocated and managed by this object.
STableQueryInfo *current;
} SAggOperatorInfo;
typedef struct SProjectOperatorInfo {
@ -659,6 +660,6 @@ 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 doCreateExecTaskInfo(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, STableGroupInfo* pGroupInfo, void* readerHandle);
int32_t doCreateExecTaskInfo(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, void* readerHandle);
#endif // TDENGINE_EXECUTORIMPL_H

3
source/libs/executor/src/executil.c
View File

@ -170,8 +170,7 @@ void clearResultRow(STaskRuntimeEnv *pRuntimeEnv, SResultRow *pResultRow, int16_
// TODO refactor: use macro
SResultRowEntryInfo* getResultCell(const SResultRow* pRow, int32_t index, int32_t* offset) {
assert(index >= 0 && offset != NULL);
// return (SResultRowEntryInfo*)((char*) pRow->pCellInfo + offset[index]);
return NULL;
return (SResultRowEntryInfo*)((char*) pRow->pEntryInfo + offset[index]);
}
size_t getResultRowSize(SArray* pExprInfo) {

35
source/libs/executor/src/executorMain.c
View File

@ -79,40 +79,9 @@ int32_t qCreateExecTask(void* tsdb, int32_t vgId, SSubplan* pSubplan, qTaskInfo_
int32_t tableType = 0;
SPhyNode* pPhyNode = pSubplan->pNode;
STableGroupInfo groupInfo = {0};
// STableGroupInfo groupInfo = {0};
int32_t type = pPhyNode->info.type;
if (type == OP_TableScan || type == OP_DataBlocksOptScan) {
STableScanPhyNode* pTableScanNode = (STableScanPhyNode*)pPhyNode;
uid = pTableScanNode->scan.uid;
window = pTableScanNode->window;
tableType = pTableScanNode->scan.tableType;
if (tableType == TSDB_SUPER_TABLE) {
code =
tsdbQuerySTableByTagCond(tsdb, uid, window.skey, NULL, 0, 0, NULL, &groupInfo, NULL, 0, pSubplan->id.queryId);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
} else { // Create one table group.
groupInfo.numOfTables = 1;
groupInfo.pGroupList = taosArrayInit(1, POINTER_BYTES);
SArray* pa = taosArrayInit(1, sizeof(STableKeyInfo));
STableKeyInfo info = {.pTable = NULL, .lastKey = 0, .uid = uid};
taosArrayPush(pa, &info);
taosArrayPush(groupInfo.pGroupList, &pa);
}
if (groupInfo.numOfTables == 0) {
code = 0;
// qDebug("no table qualified for query, reqId:0x%"PRIx64, (*pTask)->id.queryId);
goto _error;
}
}
code = doCreateExecTaskInfo(pSubplan, pTask, &groupInfo, tsdb);
code = doCreateExecTaskInfo(pSubplan, pTask, tsdb);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}

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

@ -1196,10 +1196,8 @@ static void doSetInputDataBlock(SOperatorInfo* pOperator, SQLFunctionCtx* pCtx,
}
static void doAggregateImpl(SOperatorInfo* pOperator, TSKEY startTs, SQLFunctionCtx* pCtx, SSDataBlock* pSDataBlock) {
STaskRuntimeEnv* pRuntimeEnv = pOperator->pRuntimeEnv;
for (int32_t k = 0; k < pOperator->numOfOutput; ++k) {
if (functionNeedToExecute(pRuntimeEnv, &pCtx[k])) {
if (functionNeedToExecute(NULL, &pCtx[k])) {
pCtx[k].startTs = startTs;// this can be set during create the struct
pCtx[k].fpSet->addInput(&pCtx[k]);
}
@ -1818,7 +1816,6 @@ static int32_t getGroupbyColumnIndex(SGroupbyExpr *pGroupbyExpr, SSDataBlock* pD
static bool functionNeedToExecute(STaskRuntimeEnv *pRuntimeEnv, SQLFunctionCtx *pCtx) {
struct SResultRowEntryInfo *pResInfo = GET_RES_INFO(pCtx);
STaskAttr* pQueryAttr = pRuntimeEnv->pQueryAttr;
// in case of timestamp column, always generated results.
int32_t functionId = pCtx->functionId;
@ -1831,12 +1828,12 @@ static bool functionNeedToExecute(STaskRuntimeEnv *pRuntimeEnv, SQLFunctionCtx *
}
if (functionId == FUNCTION_FIRST_DST || functionId == FUNCTION_FIRST) {
return QUERY_IS_ASC_QUERY(pQueryAttr);
// return QUERY_IS_ASC_QUERY(pQueryAttr);
}
// denote the order type
if ((functionId == FUNCTION_LAST_DST || functionId == FUNCTION_LAST)) {
return pCtx->param[0].i == pQueryAttr->order.order;
// return pCtx->param[0].i == pQueryAttr->order.order;
}
// in the reverse table scan, only the following functions need to be executed
@ -2050,6 +2047,7 @@ static SQLFunctionCtx* createSqlFunctionCtx_rv(SArray* pExprInfo, int32_t** rowC
// pCtx->inputType = pSqlExpr->colType;
pCtx->ptsOutputBuf = NULL;
pCtx->fpSet = fpSet;
pCtx->resDataInfo.bytes = pSqlExpr->resSchema.bytes;
pCtx->resDataInfo.type = pSqlExpr->resSchema.type;
@ -5742,11 +5740,7 @@ static SSDataBlock* doAggregate(void* param, bool* newgroup) {
SAggOperatorInfo* pAggInfo = pOperator->info;
SOptrBasicInfo* pInfo = &pAggInfo->binfo;
STaskRuntimeEnv* pRuntimeEnv = pOperator->pRuntimeEnv;
STaskAttr* pQueryAttr = pRuntimeEnv->pQueryAttr;
int32_t order = pQueryAttr->order.order;
int32_t order = TSDB_ORDER_ASC;
SOperatorInfo* downstream = pOperator->pDownstream[0];
while(1) {
@ -5758,18 +5752,13 @@ static SSDataBlock* doAggregate(void* param, bool* newgroup) {
break;
}
if (pRuntimeEnv->current != NULL) {
setTagValue(pOperator, pRuntimeEnv->current->pTable, pInfo->pCtx, pOperator->numOfOutput);
if (pAggInfo->current != NULL) {
// setTagValue(pOperator, pAggInfo->current->pTable, pInfo->pCtx, pOperator->numOfOutput);
}
// if (downstream->operatorType == OP_DataBlocksOptScan) {
// STableScanInfo* pScanInfo = downstream->info;
// order = getTableScanOrder(pScanInfo);
// }
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pInfo->pCtx, pBlock, order);
doAggregateImpl(pOperator, pQueryAttr->window.skey, pInfo->pCtx, pBlock);
doAggregateImpl(pOperator, 0, pInfo->pCtx, pBlock);
}
doSetOperatorCompleted(pOperator);
@ -6667,7 +6656,7 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SArray* pE
pInfo->pResultRowHashTable = taosHashInit(10, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
pInfo->pResultRowListSet = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
pInfo->keyBuf = NULL;//malloc(pQueryAttr->maxTableColumnWidth + sizeof(int64_t) + POINTER_BYTES);
pInfo->keyBuf = malloc(1024 + sizeof(int64_t) + POINTER_BYTES); // TODO:
pInfo->pool = initResultRowPool(getResultRowSize(pExprInfo));
pInfo->pResultRowArrayList = taosArrayInit(10, sizeof(SResultRowCell));
@ -7625,16 +7614,25 @@ static SExecTaskInfo* createExecTaskInfo(uint64_t queryId) {
return pTaskInfo;
}
SOperatorInfo* doCreateOperatorTreeNode(SPhyNode* pPhyNode, SExecTaskInfo* pTaskInfo, void* param) {
static tsdbReadHandleT doCreateDataReadHandle(STableScanPhyNode* pTableScanNode, void* readerHandle, uint64_t queryId);
SOperatorInfo* doCreateOperatorTreeNode(SPhyNode* pPhyNode, SExecTaskInfo* pTaskInfo, void* readerHandle, uint64_t queryId) {
if (pPhyNode->pChildren == NULL || taosArrayGetSize(pPhyNode->pChildren) == 0) {
if (pPhyNode->info.type == OP_TableScan) {
SScanPhyNode* pScanPhyNode = (SScanPhyNode*)pPhyNode;
size_t numOfCols = taosArrayGetSize(pPhyNode->pTargets);
return createTableScanOperatorInfo(param, pScanPhyNode->order, numOfCols, pScanPhyNode->count, pTaskInfo);
tsdbReadHandleT tReaderHandle = doCreateDataReadHandle((STableScanPhyNode*) pPhyNode, readerHandle, (uint64_t) queryId);
return createTableScanOperatorInfo(tReaderHandle, pScanPhyNode->order, numOfCols, pScanPhyNode->count, pTaskInfo);
} else if (pPhyNode->info.type == OP_DataBlocksOptScan) {
SScanPhyNode* pScanPhyNode = (SScanPhyNode*)pPhyNode;
size_t numOfCols = taosArrayGetSize(pPhyNode->pTargets);
return createDataBlocksOptScanInfo(param, pScanPhyNode->order, numOfCols, pScanPhyNode->count, pScanPhyNode->reverse, pTaskInfo);
tsdbReadHandleT tReaderHandle = doCreateDataReadHandle((STableScanPhyNode*) pPhyNode, readerHandle, (uint64_t) queryId);
return createDataBlocksOptScanInfo(tReaderHandle, pScanPhyNode->order, numOfCols, pScanPhyNode->count, pScanPhyNode->reverse, pTaskInfo);
} else if (pPhyNode->info.type == OP_Exchange) {
SExchangePhyNode* pEx = (SExchangePhyNode*) pPhyNode;
return createExchangeOperatorInfo(pEx->pSrcEndPoints, pEx->node.pTargets, pTaskInfo);
@ -7646,46 +7644,83 @@ SOperatorInfo* doCreateOperatorTreeNode(SPhyNode* pPhyNode, SExecTaskInfo* pTask
assert(size == 1);
for (int32_t i = 0; i < size; ++i) {
SPhyNode* pChildNode = taosArrayGet(pPhyNode->pChildren, i);
SOperatorInfo* op = doCreateOperatorTreeNode(pChildNode, pTaskInfo, param);
SPhyNode* pChildNode = taosArrayGetP(pPhyNode->pChildren, i);
SOperatorInfo* op = doCreateOperatorTreeNode(pChildNode, pTaskInfo, readerHandle, queryId);
return createAggregateOperatorInfo(op, pPhyNode->pTargets, pTaskInfo);
}
}
}
int32_t doCreateExecTaskInfo(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, STableGroupInfo* pGroupInfo, void* readerHandle) {
static tsdbReadHandleT createDataReadHandle(STableScanPhyNode* pTableScanNode, STableGroupInfo* pGroupInfo, void* readerHandle, uint64_t queryId) {
STsdbQueryCond cond = {.loadExternalRows = false};
tsdbReadHandleT tsdbReadHandle = NULL;
cond.order = pTableScanNode->scan.order;
cond.numOfCols = taosArrayGetSize(pTableScanNode->scan.node.pTargets);
cond.colList = calloc(cond.numOfCols, sizeof(SColumnInfo));
cond.twindow = pTableScanNode->window;
cond.type = BLOCK_LOAD_OFFSET_SEQ_ORDER;
SPhyNode* pPhyNode = pPlan->pNode;
if (pPhyNode->info.type == OP_TableScan || pPhyNode->info.type == OP_DataBlocksOptScan) {
STableScanPhyNode* pTableScanNode = (STableScanPhyNode*)pPhyNode;
cond.order = pTableScanNode->scan.order;
cond.numOfCols = taosArrayGetSize(pTableScanNode->scan.node.pTargets);
cond.colList = calloc(cond.numOfCols, sizeof(SColumnInfo));
cond.twindow = pTableScanNode->window;
cond.type = BLOCK_LOAD_OFFSET_SEQ_ORDER;
for (int32_t i = 0; i < cond.numOfCols; ++i) {
SExprInfo* pExprInfo = taosArrayGetP(pTableScanNode->scan.node.pTargets, i);
assert(pExprInfo->pExpr->nodeType == TEXPR_COL_NODE);
for (int32_t i = 0; i < cond.numOfCols; ++i) {
SExprInfo* pExprInfo = taosArrayGetP(pTableScanNode->scan.node.pTargets, i);
assert(pExprInfo->pExpr->nodeType == TEXPR_COL_NODE);
SSchema* pSchema = pExprInfo->pExpr->pSchema;
cond.colList[i].type = pSchema->type;
cond.colList[i].bytes = pSchema->bytes;
cond.colList[i].colId = pSchema->colId;
}
*pTaskInfo = createExecTaskInfo((uint64_t) pPlan->id.queryId);
tsdbReadHandle = tsdbQueryTables(readerHandle, &cond, pGroupInfo, (*pTaskInfo)->id.queryId, NULL);
} else if (pPhyNode->info.type == OP_Exchange) {
*pTaskInfo = createExecTaskInfo((uint64_t) pPlan->id.queryId);
} else {
assert(0);
SSchema* pSchema = pExprInfo->pExpr->pSchema;
cond.colList[i].type = pSchema->type;
cond.colList[i].bytes = pSchema->bytes;
cond.colList[i].colId = pSchema->colId;
}
(*pTaskInfo)->pRoot = doCreateOperatorTreeNode(pPlan->pNode, *pTaskInfo, tsdbReadHandle);
return tsdbQueryTables(readerHandle, &cond, pGroupInfo, queryId, NULL);
}
static tsdbReadHandleT doCreateDataReadHandle(STableScanPhyNode* pTableScanNode, void* readerHandle, uint64_t queryId) {
int32_t code = 0;
STableGroupInfo groupInfo = {0};
uint64_t uid = pTableScanNode->scan.uid;
STimeWindow window = pTableScanNode->window;
int32_t tableType = pTableScanNode->scan.tableType;
if (tableType == TSDB_SUPER_TABLE) {
code =
tsdbQuerySTableByTagCond(readerHandle, uid, window.skey, NULL, 0, 0, NULL, &groupInfo, NULL, 0, queryId);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
} else { // Create one table group.
groupInfo.numOfTables = 1;
groupInfo.pGroupList = taosArrayInit(1, POINTER_BYTES);
SArray* pa = taosArrayInit(1, sizeof(STableKeyInfo));
STableKeyInfo info = {.pTable = NULL, .lastKey = 0, .uid = uid};
taosArrayPush(pa, &info);
taosArrayPush(groupInfo.pGroupList, &pa);
}
if (groupInfo.numOfTables == 0) {
code = 0;
// qDebug("no table qualified for query, reqId:0x%"PRIx64, (*pTask)->id.queryId);
goto _error;
}
return createDataReadHandle(pTableScanNode, &groupInfo, readerHandle, queryId);
_error:
terrno = code;
return NULL;
}
int32_t doCreateExecTaskInfo(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, void* readerHandle) {
tsdbReadHandleT tReaderHandle = NULL;
int32_t code = 0;
uint64_t queryId = pPlan->id.queryId;
SPhyNode* pPhyNode = pPlan->pNode;
*pTaskInfo = createExecTaskInfo(queryId);
(*pTaskInfo)->pRoot = doCreateOperatorTreeNode(pPlan->pNode, *pTaskInfo, readerHandle, queryId);
if ((*pTaskInfo)->pRoot == NULL) {
return terrno;
}

2
source/libs/function/CMakeLists.txt
View File

@ -8,5 +8,5 @@ target_include_directories(
target_link_libraries(
function
PRIVATE os util common
PRIVATE os util common nodes
)

76
source/libs/function/inc/functionMgtInt.h Normal file
View File

@ -0,0 +1,76 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_FUNCTION_MGT_INT_H_
#define _TD_FUNCTION_MGT_INT_H_
#ifdef __cplusplus
extern "C" {
#endif
#include "functionMgt.h"
#define FUNC_MGT_DATA_TYPE_MASK(n) (1 << n)
#define FUNC_MGT_DATA_TYPE_NULL 0
#define FUNC_MGT_DATA_TYPE_BOOL FUNC_MGT_DATA_TYPE_MASK(0)
#define FUNC_MGT_DATA_TYPE_TINYINT FUNC_MGT_DATA_TYPE_MASK(1)
#define FUNC_MGT_DATA_TYPE_SMALLINT FUNC_MGT_DATA_TYPE_MASK(2)
#define FUNC_MGT_DATA_TYPE_INT FUNC_MGT_DATA_TYPE_MASK(3)
#define FUNC_MGT_DATA_TYPE_BIGINT FUNC_MGT_DATA_TYPE_MASK(4)
#define FUNC_MGT_DATA_TYPE_FLOAT FUNC_MGT_DATA_TYPE_MASK(5)
#define FUNC_MGT_DATA_TYPE_DOUBLE FUNC_MGT_DATA_TYPE_MASK(6)
#define FUNC_MGT_DATA_TYPE_BINARY FUNC_MGT_DATA_TYPE_MASK(7)
#define FUNC_MGT_DATA_TYPE_TIMESTAMP FUNC_MGT_DATA_TYPE_MASK(8)
#define FUNC_MGT_DATA_TYPE_NCHAR FUNC_MGT_DATA_TYPE_MASK(9)
#define FUNC_MGT_DATA_TYPE_UTINYINT FUNC_MGT_DATA_TYPE_MASK(10)
#define FUNC_MGT_DATA_TYPE_USMALLINT FUNC_MGT_DATA_TYPE_MASK(11)
#define FUNC_MGT_DATA_TYPE_UINT FUNC_MGT_DATA_TYPE_MASK(12)
#define FUNC_MGT_DATA_TYPE_UBIGINT FUNC_MGT_DATA_TYPE_MASK(13)
#define FUNC_MGT_DATA_TYPE_VARCHAR FUNC_MGT_DATA_TYPE_MASK(14)
#define FUNC_MGT_DATA_TYPE_VARBINARY FUNC_MGT_DATA_TYPE_MASK(15)
#define FUNC_MGT_DATA_TYPE_JSON FUNC_MGT_DATA_TYPE_MASK(16)
#define FUNC_MGT_DATA_TYPE_DECIMAL FUNC_MGT_DATA_TYPE_MASK(17)
#define FUNC_MGT_DATA_TYPE_BLOB FUNC_MGT_DATA_TYPE_MASK(18)
#define FUNC_MGT_EXACT_NUMERIC_DATA_TYPE \
(FUNC_MGT_DATA_TYPE_TINYINT | FUNC_MGT_DATA_TYPE_SMALLINT | FUNC_MGT_DATA_TYPE_INT | FUNC_MGT_DATA_TYPE_BIGINT \
| FUNC_MGT_DATA_TYPE_UTINYINT | FUNC_MGT_DATA_TYPE_USMALLINT | FUNC_MGT_DATA_TYPE_UINT | FUNC_MGT_DATA_TYPE_UBIGINT)
#define FUNC_MGT_APPRO_NUMERIC_DATA_TYPE (FUNC_MGT_DATA_TYPE_FLOAT | FUNC_MGT_DATA_TYPE_DOUBLE)
#define FUNC_MGT_NUMERIC_DATA_TYPE (FUNC_MGT_EXACT_NUMERIC_DATA_TYPE | FUNC_MGT_APPRO_NUMERIC_DATA_TYPE)
typedef void* FuncDef;
typedef struct SFuncElement {
FuncDef (*defineFunc)();
} SFuncElement;
extern const SFuncElement gBuiltinFuncs[];
FuncDef createFuncDef(const char* name, int32_t maxNumOfParams);
FuncDef setOneParamSignature(FuncDef def, int64_t resDataType, int64_t paramDataType);
FuncDef setTwoParamsSignature(FuncDef def, int64_t resDataType, int64_t p1DataType, int64_t p2DataType);
FuncDef setFollowParamSignature(FuncDef def, int64_t paramDataType);
FuncDef setFollowParamsSignature(FuncDef def, int64_t p1DataType, int64_t p2DataType, int32_t followNo);
FuncDef setExecFuncs(FuncDef def, FExecGetEnv getEnv, FExecInit init, FExecProcess process, FExecFinalize finalize);
#ifdef __cplusplus
}
#endif
#endif // _TD_FUNCTION_MGT_INT_H_

5
source/libs/function/inc/taggfunction.h
View File

@ -28,6 +28,7 @@ extern "C" {
#include "function.h"
#include "tudf.h"
extern SAggFunctionInfo aggFunc[35];
#define FUNCSTATE_SO 0x0u
@ -88,6 +89,10 @@ static FORCE_INLINE void initResultRowEntry(SResultRowEntryInfo *pResInfo, int32
memset(GET_ROWCELL_INTERBUF(pResInfo), 0, bufLen);
}
#include "functionMgtInt.h"
FuncDef defineCount();
#ifdef __cplusplus
}
#endif

21
source/libs/function/src/buildins.c Normal file
View File

@ -0,0 +1,21 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "functionMgtInt.h"
#include "taggfunction.h"
const SFuncElement gBuiltinFuncs[] = {
{.defineFunc = defineCount}
};

80
source/libs/function/src/functionMgt.c Normal file
View File

@ -0,0 +1,80 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "functionMgt.h"
#include "functionMgtInt.h"
#include "taos.h"
#include "taoserror.h"
#include "thash.h"
typedef struct SFuncMgtService {
SHashObj* pFuncNameHashTable;
} SFuncMgtService;
static SFuncMgtService gFunMgtService;
int32_t fmFuncMgtInit() {
gFunMgtService.pFuncNameHashTable = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
if (NULL == gFunMgtService.pFuncNameHashTable) {
return TSDB_CODE_FAILED;
}
return TSDB_CODE_SUCCESS;
}
typedef struct SFuncDef {
char name[TSDB_FUNC_NAME_LEN];
int32_t maxNumOfParams;
SFuncExecFuncs execFuncs;
} SFuncDef ;
FuncDef createFuncDef(const char* name, int32_t maxNumOfParams) {
SFuncDef* pDef = calloc(1, sizeof(SFuncDef));
if (NULL == pDef) {
return NULL;
}
strcpy(pDef->name, name);
pDef->maxNumOfParams = maxNumOfParams;
return pDef;
}
FuncDef setOneParamSignature(FuncDef def, int64_t resDataType, int64_t paramDataType) {
// todo
}
FuncDef setTwoParamsSignature(FuncDef def, int64_t resDataType, int64_t p1DataType, int64_t p2DataType) {
// todo
}
FuncDef setFollowParamSignature(FuncDef def, int64_t paramDataType) {
// todo
}
FuncDef setFollowParamsSignature(FuncDef def, int64_t p1DataType, int64_t p2DataType, int32_t followNo) {
// todo
}
FuncDef setExecFuncs(FuncDef def, FExecGetEnv getEnv, FExecInit init, FExecProcess process, FExecFinalize finalize) {
SFuncDef* pDef = (SFuncDef*)def;
pDef->execFuncs.getEnv = getEnv;
pDef->execFuncs.init = init;
pDef->execFuncs.process = process;
pDef->execFuncs.finalize = finalize;
return def;
}
int32_t registerFunc(FuncDef func) {
}

24
source/libs/function/src/taggfunction.c
View File

@ -4379,6 +4379,24 @@ int32_t functionCompatList[] = {
6, 8, 7,
};
//typedef struct SFunctionFpSet {
// bool (*init)(struct SQLFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo); // setup the execute environment
// void (*addInput)(struct SQLFunctionCtx *pCtx);
//
// // finalizer must be called after all exec has been executed to generated final result.
// void (*finalize)(struct SQLFunctionCtx *pCtx);
// void (*combine)(struct SQLFunctionCtx *pCtx);
//} SFunctionFpSet;
SFunctionFpSet fpSet[1] = {
{
.init = function_setup,
.addInput = count_function,
.finalize = doFinalizer,
.combine = count_func_merge,
}
};
SAggFunctionInfo aggFunc[35] = {{
// 0, count function does not invoke the finalize function
"count",
@ -4835,3 +4853,9 @@ SAggFunctionInfo aggFunc[35] = {{
statisRequired,
}
};
FuncDef defineCount() {
FuncDef def = createFuncDef("count", 1);
// todo define signature
return setExecFuncs(def, NULL, function_setup, count_function, doFinalizer);
}

1
source/libs/parser/inc/queryInfoUtil.h
View File

@ -30,7 +30,6 @@ SSchema *getTableTagSchema(const STableMeta* pTableMeta);
SArray *getCurrentExprList(SQueryStmtInfo* pQueryInfo);
size_t getNumOfExprs(SQueryStmtInfo* pQueryInfo);
SExprInfo* createBinaryExprInfo(struct tExprNode* pNode, SSchema* pResSchema);
void addExprInfo(SArray* pExprList, int32_t index, SExprInfo* pExprInfo, int32_t level);
void updateExprInfo(SExprInfo* pExprInfo, int16_t functionId, int32_t colId, int16_t srcColumnIndex, int16_t resType, int16_t resSize);

11
source/libs/planner/src/logicPlan.c
View File

@ -13,9 +13,10 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <function.h>
#include "function.h"
#include "os.h"
#include "parser.h"
#include "function.h"
#include "plannerInt.h"
typedef struct SFillEssInfo {
@ -198,7 +199,6 @@ static SQueryPlanNode* doAddTableColumnNode(const SQueryStmtInfo* pQueryInfo, SQ
if (!pQueryInfo->info.projectionQuery) {
SArray* p = pQueryInfo->exprList[0];
STableMetaInfo* pTableMetaInfo1 = getMetaInfo(pQueryInfo, 0);
// table source column projection, generate the projection expr
int32_t numOfCols = (int32_t) taosArrayGetSize(tableCols);
@ -209,12 +209,11 @@ static SQueryPlanNode* doAddTableColumnNode(const SQueryStmtInfo* pQueryInfo, SQ
SExprInfo* pExprInfo = taosArrayGetP(p, i);
SColumn* pCol = pExprInfo->base.pColumns;
SSourceParam param = {0};
addIntoSourceParam(&param, NULL, pCol);
SSchema schema = createSchema(pCol->info.type, pCol->info.bytes, pCol->info.colId, pCol->name);
SExprInfo* p = createExprInfo(pTableMetaInfo1, "project", &param, &schema, 0);
taosArrayPush(pNode->pExpr, &p);
tExprNode* pExprNode = pExprInfo->pExpr->_function.pChild[0];
SExprInfo* px = createBinaryExprInfo(pExprNode, &schema);
taosArrayPush(pNode->pExpr, &px);
}
}

32
source/libs/planner/src/physicalPlanJson.c
View File

@ -117,6 +117,26 @@ static bool fromPnode(const cJSON* json, const char* name, FFromJson func, void*
return func(jObj, *obj);
}
static bool fromPnodeArray(const cJSON* json, const char* name, FFromJson func, SArray** array) {
const cJSON* jArray = cJSON_GetObjectItem(json, name);
int32_t size = (NULL == jArray ? 0 : cJSON_GetArraySize(jArray));
if (size > 0) {
*array = taosArrayInit(size, POINTER_BYTES);
if (NULL == *array) {
return false;
}
}
for (int32_t i = 0; i < size; ++i) {
cJSON* jItem = cJSON_GetArrayItem(jArray, i);
void* item = calloc(1, getPnodeTypeSize(jItem));
if (NULL == item || !func(jItem, item)) {
return false;
}
taosArrayPush(*array, &item);
}
return true;
}
static bool addTarray(cJSON* json, const char* name, FToJson func, const SArray* array, bool isPoint) {
size_t size = (NULL == array) ? 0 : taosArrayGetSize(array);
if (size > 0) {
@ -387,8 +407,8 @@ static const char* jkFunctionChild = "Child";
static bool functionToJson(const void* obj, cJSON* jFunc) {
const tExprNode* exprInfo = (const tExprNode*)obj;
bool res = cJSON_AddStringToObject(jFunc, jkFunctionName, exprInfo->_function.functionName);
if (res) {
res = addRawArray(jFunc, jkFunctionChild, exprNodeToJson, exprInfo->_function.pChild, sizeof(tExprNode*), exprInfo->_function.num);
if (res && NULL != exprInfo->_function.pChild) {
res = addRawArray(jFunc, jkFunctionChild, exprNodeToJson, exprInfo->_function.pChild, sizeof(tExprNode*), exprInfo->_function.num);
}
return res;
}
@ -396,6 +416,10 @@ static bool functionToJson(const void* obj, cJSON* jFunc) {
static bool functionFromJson(const cJSON* json, void* obj) {
tExprNode* exprInfo = (tExprNode*)obj;
copyString(json, jkFunctionName, exprInfo->_function.functionName);
exprInfo->_function.pChild = calloc(1, sizeof(tExprNode*));
if (NULL == exprInfo->_function.pChild) {
return false;
}
return fromRawArrayWithAlloc(json, jkFunctionChild, exprNodeFromJson, (void**)exprInfo->_function.pChild, sizeof(tExprNode*), &exprInfo->_function.num);
}
@ -851,7 +875,7 @@ static bool specificPhyNodeFromJson(const cJSON* json, void* obj) {
case OP_SystemTableScan:
return scanNodeFromJson(json, obj);
case OP_Aggregate:
break; // todo
return aggNodeFromJson(json, obj);
case OP_Project:
return true;
// case OP_Groupby:
@ -922,7 +946,7 @@ static bool phyNodeFromJson(const cJSON* json, void* obj) {
res = fromObject(json, jkPnodeSchema, dataBlockSchemaFromJson, &node->targetSchema, true);
}
if (res) {
res = fromArray(json, jkPnodeChildren, phyNodeFromJson, &node->pChildren, sizeof(SSlotSchema));
res = fromPnodeArray(json, jkPnodeChildren, phyNodeFromJson, &node->pChildren);
}
if (res) {
res = fromObject(json, node->info.name, specificPhyNodeFromJson, node, true);

19
source/libs/transport/inc/transComm.h
View File

@ -68,6 +68,25 @@ typedef void* queue[2];
QUEUE_PREV_NEXT(e) = QUEUE_NEXT(e); \
QUEUE_NEXT_PREV(e) = QUEUE_PREV(e); \
}
#define QUEUE_SPLIT(h, q, n) \
do { \
QUEUE_PREV(n) = QUEUE_PREV(h); \
QUEUE_PREV_NEXT(n) = (n); \
QUEUE_NEXT(n) = (q); \
QUEUE_PREV(h) = QUEUE_PREV(q); \
QUEUE_PREV_NEXT(h) = (h); \
QUEUE_PREV(q) = (n); \
} while (0)
#define QUEUE_MOVE(h, n) \
do { \
if (QUEUE_IS_EMPTY(h)) { \
QUEUE_INIT(n); \
} else { \
queue* q = QUEUE_HEAD(h); \
QUEUE_SPLIT(h, q, n); \
} \
} while (0)
/* Return the element at the front of the queue. */
#define QUEUE_HEAD(q) (QUEUE_NEXT(q))

1
source/libs/transport/src/trans.c
View File

@ -35,6 +35,7 @@ void* rpcOpen(const SRpcInit* pInit) {
if (pInit->label) {
tstrncpy(pRpc->label, pInit->label, strlen(pInit->label));
}
pRpc->cfp = pInit->cfp;
pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads;
pRpc->connType = pInit->connType;
pRpc->tcphandle = (*taosHandle[pRpc->connType])(0, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc);

163
source/libs/transport/src/transCli.c
View File

@ -20,12 +20,16 @@
typedef struct SCliConn {
uv_connect_t connReq;
uv_stream_t* stream;
uv_write_t* writeReq;
void* data;
queue conn;
char spi;
char secured;
} SCliConn;
typedef struct SCliMsg {
SRpcReqContext* context;
queue q;
uint64_t st;
} SCliMsg;
typedef struct SCliThrdObj {
@ -45,86 +49,169 @@ typedef struct SClientObj {
SCliThrdObj** pThreadObj;
} SClientObj;
static void clientWriteCb(uv_write_t* req, int status);
// conn pool
static SCliConn* getConnFromCache(void* cache, char* ip, uint32_t port);
static void addConnToCache(void* cache, char* ip, uint32_t port, SCliConn* conn);
static void clientAllocrReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);
static void clientReadCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf);
static void clientConnCb(struct uv_connect_s* req, int status);
static void clientWriteCb(uv_write_t* req, int status);
static void clientConnCb(uv_connect_t* req, int status);
static void clientAsyncCb(uv_async_t* handle);
static void clientDestroy(uv_handle_t* handle);
static void clientConnDestroy(SCliConn* pConn);
static void* clientThread(void* arg);
static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd);
static void clientAllocrReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
// impl later
}
static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) {
// impl later
SCliConn* conn = handle->data;
if (nread > 0) {
return;
}
//
uv_close((uv_handle_t*)handle, clientDestroy);
}
static void clientConnDestroy(SCliConn* conn) {
// impl later
//
}
static void clientDestroy(uv_handle_t* handle) {
SCliConn* conn = handle->data;
clientConnDestroy(conn);
}
static void clientWriteCb(uv_write_t* req, int status) {
// impl later
}
static void clientFailedCb(uv_handle_t* handle) {
// impl later
tDebug("close handle");
}
static void clientReadCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf) {
// impl later
}
static void clientConnCb(struct uv_connect_s* req, int status) {
SCliConn* pConn = req->data;
if (status == 0) {
tDebug("data already was written on stream");
} else {
uv_close((uv_handle_t*)pConn->stream, clientDestroy);
return;
}
uv_read_start((uv_stream_t*)pConn->stream, clientAllocrReadBufferCb, clientReadCb);
// impl later
}
static void clientWrite(SCliConn* pConn) {
SCliMsg* pMsg = pConn->data;
SEpSet* pEpSet = &pMsg->context->epSet;
SRpcHead* pHead = rpcHeadFromCont(pMsg->context->pCont);
int msgLen = rpcMsgLenFromCont(pMsg->context->contLen);
char* msg = (char*)(pHead);
uv_buf_t wb = uv_buf_init(msg, msgLen);
uv_write(pConn->writeReq, (uv_stream_t*)pConn->stream, &wb, 1, clientWriteCb);
}
static void clientConnCb(uv_connect_t* req, int status) {
// impl later
SCliConn* pConn = req->data;
if (status != 0) {
tError("failed to connect %s", uv_err_name(status));
clientConnDestroy(pConn);
return;
}
SCliMsg* pMsg = pConn->data;
SEpSet* pEpSet = &pMsg->context->epSet;
SRpcMsg rpcMsg;
// rpcMsg.ahandle = pMsg->context->ahandle;
// rpcMsg.pCont = NULL;
char* fqdn = pEpSet->fqdn[pEpSet->inUse];
uint32_t port = pEpSet->port[pEpSet->inUse];
if (status != 0) {
// call user fp later
tError("failed to connect server(%s, %d), errmsg: %s", fqdn, port, uv_strerror(status));
uv_close((uv_handle_t*)req->handle, clientFailedCb);
SRpcInfo* pRpc = pMsg->context->pRpc;
(pRpc->cfp)(NULL, &rpcMsg, pEpSet);
uv_close((uv_handle_t*)req->handle, clientDestroy);
return;
}
assert(pConn->stream == req->handle);
// impl later
}
static SCliConn* getConnFromCache(void* cache, char* ip, uint32_t port) {
// impl later
return NULL;
}
static void clientAsyncCb(uv_async_t* handle) {
SCliThrdObj* pThrd = handle->data;
SCliMsg* pMsg = NULL;
pthread_mutex_lock(&pThrd->msgMtx);
if (!QUEUE_IS_EMPTY(&pThrd->msg)) {
queue* head = QUEUE_HEAD(&pThrd->msg);
pMsg = QUEUE_DATA(head, SCliMsg, q);
QUEUE_REMOVE(head);
}
pthread_mutex_unlock(&pThrd->msgMtx);
static void addConnToCache(void* cache, char* ip, uint32_t port, SCliConn* conn) {
// impl later
}
static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
SEpSet* pEpSet = &pMsg->context->epSet;
SEpSet* pEpSet = &pMsg->context->epSet;
char* fqdn = pEpSet->fqdn[pEpSet->inUse];
uint32_t port = pEpSet->port[pEpSet->inUse];
uint64_t el = taosGetTimestampUs() - pMsg->st;
tDebug("msg tran time cost: %" PRIu64 "", el);
SCliConn* conn = getConnFromCache(pThrd->cache, fqdn, port);
if (conn != NULL) {
// impl later
conn->data = pMsg;
conn->writeReq->data = conn;
clientWrite(conn);
// uv_buf_t wb;
// uv_write(conn->writeReq, (uv_stream_t*)conn->stream, &wb, 1, clientWriteCb);
} else {
SCliConn* conn = malloc(sizeof(SCliConn));
conn->stream = (uv_stream_t*)malloc(sizeof(uv_tcp_t));
uv_tcp_init(pThrd->loop, (uv_tcp_t*)(conn->stream));
conn->writeReq = malloc(sizeof(uv_write_t));
conn->connReq.data = conn;
conn->data = pMsg;
struct sockaddr_in addr;
uv_ip4_addr(fqdn, port, &addr);
// handle error in callback if connect error
// handle error in callback if fail to connect
uv_tcp_connect(&conn->connReq, (uv_tcp_t*)(conn->stream), (const struct sockaddr*)&addr, clientConnCb);
// SRpcMsg rpcMsg;
// SEpSet* pEpSet = &pMsg->context->epSet;
// SRpcInfo* pRpc = pMsg->context->pRpc;
//// rpcMsg.ahandle = pMsg->context->ahandle;
// rpcMsg.pCont = NULL;
// rpcMsg.ahandle = pMsg->context->ahandle;
// uint64_t el1 = taosGetTimestampUs() - et;
// tError("msg tran back first: time cost: %" PRIu64 "", el1);
// et = taosGetTimestampUs();
//(pRpc->cfp)(NULL, &rpcMsg, pEpSet);
// uint64_t el2 = taosGetTimestampUs() - et;
// tError("msg tran back second: time cost: %" PRIu64 "", el2);
}
}
static void clientAsyncCb(uv_async_t* handle) {
SCliThrdObj* pThrd = handle->data;
SCliMsg* pMsg = NULL;
queue wq;
// SRpcReqContext* pCxt = pMsg->context;
// batch process to avoid to lock/unlock frequently
pthread_mutex_lock(&pThrd->msgMtx);
QUEUE_MOVE(&pThrd->msg, &wq);
pthread_mutex_unlock(&pThrd->msgMtx);
// SRpcHead* pHead = rpcHeadFromCont(pCtx->pCont);
// char* msg = (char*)pHead;
// int len = rpcMsgLenFromCont(pCtx->contLen);
// tmsg_t msgType = pCtx->msgType;
// impl later
int count = 0;
while (!QUEUE_IS_EMPTY(&wq)) {
queue* h = QUEUE_HEAD(&wq);
QUEUE_REMOVE(h);
pMsg = QUEUE_DATA(h, SCliMsg, q);
clientHandleReq(pMsg, pThrd);
count++;
if (count >= 2) {
tError("send batch size: %d", count);
}
}
}
static void* clientThread(void* arg) {
@ -142,9 +229,6 @@ void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads,
SCliThrdObj* pThrd = (SCliThrdObj*)calloc(1, sizeof(SCliThrdObj));
QUEUE_INIT(&pThrd->msg);
pthread_mutex_init(&pThrd->msgMtx, NULL);
// QUEUE_INIT(&pThrd->clientCache);
pThrd->loop = (uv_loop_t*)malloc(sizeof(uv_loop_t));
uv_loop_init(pThrd->loop);
@ -186,6 +270,7 @@ void rpcSendRequest(void* shandle, const SEpSet* pEpSet, SRpcMsg* pMsg, int64_t*
}
SCliMsg* msg = malloc(sizeof(SCliMsg));
msg->context = pContext;
msg->st = taosGetTimestampUs();
SCliThrdObj* thrd = ((SClientObj*)pRpc->tcphandle)->pThreadObj[index % pRpc->numOfThreads];

30
source/libs/transport/src/transSrv.c
View File

@ -277,10 +277,6 @@ void uvOnReadCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf) {
}
return;
}
if (terrno != 0) {
// handle err code
}
if (nread != UV_EOF) {
tDebug("Read error %s\n", uv_err_name(nread));
}
@ -309,21 +305,23 @@ void uvOnWriteCb(uv_write_t* req, int status) {
void uvWorkerAsyncCb(uv_async_t* handle) {
SWorkThrdObj* pThrd = container_of(handle, SWorkThrdObj, workerAsync);
SConn* conn = NULL;
// opt later
queue wq;
// batch process to avoid to lock/unlock frequently
pthread_mutex_lock(&pThrd->connMtx);
if (!QUEUE_IS_EMPTY(&pThrd->conn)) {
queue* head = QUEUE_HEAD(&pThrd->conn);
conn = QUEUE_DATA(head, SConn, queue);
QUEUE_REMOVE(head);
}
QUEUE_MOVE(&pThrd->conn, &wq);
pthread_mutex_unlock(&pThrd->connMtx);
if (conn == NULL) {
tError("except occurred, do nothing");
return;
while (!QUEUE_IS_EMPTY(&wq)) {
queue* head = QUEUE_HEAD(&wq);
QUEUE_REMOVE(head);
SConn* conn = QUEUE_DATA(head, SConn, queue);
if (conn == NULL) {
tError("except occurred, do nothing");
return;
}
uv_buf_t wb = uv_buf_init(conn->writeBuf.buf, conn->writeBuf.len);
uv_write(conn->pWriter, (uv_stream_t*)conn->pTcp, &wb, 1, uvOnWriteCb);
}
uv_buf_t wb = uv_buf_init(conn->writeBuf.buf, conn->writeBuf.len);
uv_write(conn->pWriter, (uv_stream_t*)conn->pTcp, &wb, 1, uvOnWriteCb);
}
void uvOnAcceptCb(uv_stream_t* stream, int status) {

6
source/libs/transport/test/rclient.c
View File

@ -34,8 +34,8 @@ typedef struct {
static void processResponse(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) {
SInfo *pInfo = (SInfo *)pMsg->ahandle;
tDebug("thread:%d, response is received, type:%d contLen:%d code:0x%x", pInfo->index, pMsg->msgType, pMsg->contLen,
pMsg->code);
// tDebug("thread:%d, response is received, type:%d contLen:%d code:0x%x", pInfo->index, pMsg->msgType, pMsg->contLen,
// pMsg->code);
if (pEpSet) pInfo->epSet = *pEpSet;
@ -57,7 +57,7 @@ static void *sendRequest(void *param) {
rpcMsg.contLen = pInfo->msgSize;
rpcMsg.ahandle = pInfo;
rpcMsg.msgType = 1;
tDebug("thread:%d, send request, contLen:%d num:%d", pInfo->index, pInfo->msgSize, pInfo->num);
// tDebug("thread:%d, send request, contLen:%d num:%d", pInfo->index, pInfo->msgSize, pInfo->num);
rpcSendRequest(pInfo->pRpc, &pInfo->epSet, &rpcMsg, NULL);
if (pInfo->num % 20000 == 0) tInfo("thread:%d, %d requests have been sent", pInfo->index, pInfo->num);
tsem_wait(&pInfo->rspSem);

15
source/nodes/CMakeLists.txt Normal file
View File

@ -0,0 +1,15 @@
aux_source_directory(src NODES_SRC)
add_library(nodes STATIC ${NODES_SRC})
target_include_directories(
nodes
PUBLIC "${CMAKE_SOURCE_DIR}/include/nodes"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(
nodes
PRIVATE os util
)
if(${BUILD_TEST})
ADD_SUBDIRECTORY(test)
endif(${BUILD_TEST})

20
source/nodes/src/nodesClone.c Normal file
View File

@ -0,0 +1,20 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "nodes.h"
void cloneNode(const SNode* pNode) {
}

24
source/nodes/src/nodesCode.c Normal file
View File

@ -0,0 +1,24 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "nodes.h"
int32_t nodeToString(const SNode* pNode, char** pStr, int32_t* pLen) {
}
int32_t stringToNode(const char* pStr, SNode** pNode) {
}

141
source/nodes/src/nodesEqual.c Normal file
View File

@ -0,0 +1,141 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "nodes.h"
#define COMPARE_SCALAR_FIELD(fldname) \
do { \
if (a->fldname != b->fldname) \
return false; \
} while (0)
#define COMPARE_STRING(a, b) \
(((a) != NULL && (b) != NULL) ? (strcmp(a, b) == 0) : (a) == (b))
#define COMPARE_STRING_FIELD(fldname) \
do { \
if (!COMPARE_STRING(a->fldname, b->fldname)) \
return false; \
} while (0)
#define COMPARE_NODE_FIELD(fldname) \
do { \
if (!nodeEqual(a->fldname, b->fldname)) \
return false; \
} while (0)
#define COMPARE_ARRAY_FIELD(fldname) \
do { \
if (!nodeArrayEqual(a->fldname, b->fldname)) \
return false; \
} while (0)
static bool nodeArrayEqual(const SArray* a, const SArray* b) {
if (a == b) {
return true;
}
if (NULL == a || NULL == b) {
return false;
}
if (taosArrayGetSize(a) != taosArrayGetSize(b)) {
return false;
}
size_t size = taosArrayGetSize(a);
for (size_t i = 0; i < size; ++i) {
if (!nodeEqual((SNode*)taosArrayGetP(a, i), (SNode*)taosArrayGetP(b, i))) {
return false;
}
}
return true;
}
static bool columnNodeEqual(const SColumnNode* a, const SColumnNode* b) {
COMPARE_STRING_FIELD(dbName);
COMPARE_STRING_FIELD(tableName);
COMPARE_STRING_FIELD(colName);
return true;
}
static bool valueNodeEqual(const SValueNode* a, const SValueNode* b) {
COMPARE_STRING_FIELD(literal);
return true;
}
static bool operatorNodeEqual(const SOperatorNode* a, const SOperatorNode* b) {
COMPARE_SCALAR_FIELD(opType);
COMPARE_NODE_FIELD(pLeft);
COMPARE_NODE_FIELD(pRight);
return true;
}
static bool logicConditionNodeEqual(const SLogicConditionNode* a, const SLogicConditionNode* b) {
COMPARE_SCALAR_FIELD(condType);
COMPARE_ARRAY_FIELD(pParameterList);
return true;
}
static bool isNullConditionNodeEqual(const SIsNullCondNode* a, const SIsNullCondNode* b) {
COMPARE_NODE_FIELD(pExpr);
COMPARE_SCALAR_FIELD(isNot);
return true;
}
static bool functionNodeEqual(const SFunctionNode* a, const SFunctionNode* b) {
COMPARE_SCALAR_FIELD(funcId);
COMPARE_ARRAY_FIELD(pParameterList);
return true;
}
bool nodeEqual(const SNode* a, const SNode* b) {
if (a == b) {
return true;
}
if (NULL == a || NULL == b) {
return false;
}
if (nodeType(a) != nodeType(b)) {
return false;
}
switch (nodeType(a)) {
case QUERY_NODE_COLUMN:
return columnNodeEqual((const SColumnNode*)a, (const SColumnNode*)b);
case QUERY_NODE_VALUE:
return valueNodeEqual((const SValueNode*)a, (const SValueNode*)b);
case QUERY_NODE_OPERATOR:
return operatorNodeEqual((const SOperatorNode*)a, (const SOperatorNode*)b);
case QUERY_NODE_LOGIC_CONDITION:
return logicConditionNodeEqual((const SLogicConditionNode*)a, (const SLogicConditionNode*)b);
case QUERY_NODE_IS_NULL_CONDITION:
return isNullConditionNodeEqual((const SIsNullCondNode*)a, (const SIsNullCondNode*)b);
case QUERY_NODE_FUNCTION:
return functionNodeEqual((const SFunctionNode*)a, (const SFunctionNode*)b);
case QUERY_NODE_REAL_TABLE:
case QUERY_NODE_TEMP_TABLE:
case QUERY_NODE_JOIN_TABLE:
case QUERY_NODE_GROUPING_SET:
case QUERY_NODE_ORDER_BY_EXPR:
return false; // todo
default:
break;
}
return false;
}

83
source/nodes/src/nodesTraverse.c Normal file
View File

@ -0,0 +1,83 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "nodes.h"
typedef bool (*FQueryNodeWalker)(SNode* pNode, void* pContext);
bool nodeArrayWalker(SArray* pArray, FQueryNodeWalker walker, void* pContext) {
size_t size = taosArrayGetSize(pArray);
for (size_t i = 0; i < size; ++i) {
if (!nodeTreeWalker((SNode*)taosArrayGetP(pArray, i), walker, pContext)) {
return false;
}
}
return true;
}
bool nodeTreeWalker(SNode* pNode, FQueryNodeWalker walker, void* pContext) {
if (NULL == pNode) {
return true;
}
if (!walker(pNode, pContext)) {
return false;
}
switch (nodeType(pNode)) {
case QUERY_NODE_COLUMN:
case QUERY_NODE_VALUE:
// these node types with no subnodes
return true;
case QUERY_NODE_OPERATOR: {
SOperatorNode* pOpNode = (SOperatorNode*)pNode;
if (!nodeTreeWalker(pOpNode->pLeft, walker, pContext)) {
return false;
}
return nodeTreeWalker(pOpNode->pRight, walker, pContext);
}
case QUERY_NODE_LOGIC_CONDITION:
return nodeArrayWalker(((SLogicConditionNode*)pNode)->pParameterList, walker, pContext);
case QUERY_NODE_IS_NULL_CONDITION:
return nodeTreeWalker(((SIsNullCondNode*)pNode)->pExpr, walker, pContext);
case QUERY_NODE_FUNCTION:
return nodeArrayWalker(((SFunctionNode*)pNode)->pParameterList, walker, pContext);
case QUERY_NODE_REAL_TABLE:
case QUERY_NODE_TEMP_TABLE:
return true; // todo
case QUERY_NODE_JOIN_TABLE: {
SJoinTableNode* pJoinTableNode = (SJoinTableNode*)pNode;
if (!nodeTreeWalker(pJoinTableNode->pLeft, walker, pContext)) {
return false;
}
if (!nodeTreeWalker(pJoinTableNode->pRight, walker, pContext)) {
return false;
}
return nodeTreeWalker(pJoinTableNode->pOnCond, walker, pContext);
}
case QUERY_NODE_GROUPING_SET:
return nodeArrayWalker(((SGroupingSetNode*)pNode)->pParameterList, walker, pContext);
case QUERY_NODE_ORDER_BY_EXPR:
return nodeTreeWalker(((SOrderByExprNode*)pNode)->pExpr, walker, pContext);
default:
break;
}
return false;
}
bool stmtWalker(SNode* pNode, FQueryNodeWalker walker, void* pContext) {
}

24
source/nodes/src/nodesUtil.c Normal file
View File

@ -0,0 +1,24 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "nodes.h"
bool isTimeorderQuery(const SNode* pQuery) {
}
bool isTimelineQuery(const SNode* pQuery) {
}

19
source/nodes/test/CMakeLists.txt Normal file
View File

@ -0,0 +1,19 @@
MESSAGE(STATUS "build nodes unit test")
# GoogleTest requires at least C++11
SET(CMAKE_CXX_STANDARD 11)
AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
ADD_EXECUTABLE(nodesTest ${SOURCE_LIST})
TARGET_INCLUDE_DIRECTORIES(
nodesTest
PUBLIC "${CMAKE_SOURCE_DIR}/include/nodes/"
PRIVATE "${CMAKE_SOURCE_DIR}/source/nodes/inc"
)
TARGET_LINK_LIBRARIES(
nodesTest
PUBLIC os util common nodes gtest
)

25
source/nodes/test/nodesTest.cpp Normal file
View File

@ -0,0 +1,25 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <gtest/gtest.h>
TEST(NodesTest, traverseTest) {
// todo
}
int main(int argc, char* argv[]) {
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}

87
tests/test/c/create_table.c
View File

@ -28,9 +28,14 @@ int32_t numOfThreads = 1;
int64_t numOfTables = 200000;
int32_t createTable = 1;
int32_t insertData = 0;
int32_t batchNum = 100;
int32_t batchNumOfTbl = 100;
int32_t batchNumOfRow = 1;
int32_t numOfVgroups = 2;
int32_t showTablesFlag = 0;
int32_t queryFlag = 0;
int64_t startTimestamp = 1640966400000; // 2020-01-01 00:00:00.000
typedef struct {
int64_t tableBeginIndex;
@ -167,7 +172,7 @@ void showTables() {
void *threadFunc(void *param) {
SThreadInfo *pInfo = (SThreadInfo *)param;
char *qstr = malloc(2000 * 1000);
char *qstr = malloc(batchNumOfTbl * batchNumOfRow * 128);
int32_t code = 0;
TAOS *con = taos_connect(NULL, "root", "taosdata", NULL, 0);
@ -192,7 +197,7 @@ void *threadFunc(void *param) {
// batch = MIN(batch, batchNum);
int32_t len = sprintf(qstr, "create table");
for (int32_t i = 0; i < batchNum;) {
for (int32_t i = 0; i < batchNumOfTbl;) {
len += sprintf(qstr + len, " %s_t%" PRId64 " using %s tags(%" PRId64 ")", stbName, t, stbName, t);
t++;
i++;
@ -205,7 +210,7 @@ void *threadFunc(void *param) {
TAOS_RES *pRes = taos_query(con, qstr);
code = taos_errno(pRes);
if ((code != 0) && (code != TSDB_CODE_RPC_AUTH_REQUIRED)) {
pError("failed to create table t%" PRId64 ", code: %d, reason:%s", t, code, tstrerror(code));
pError("failed to create table t%" PRId64 ", reason:%s", t, tstrerror(code));
}
taos_free_result(pRes);
int64_t endTs = taosGetTimestampUs();
@ -227,31 +232,49 @@ void *threadFunc(void *param) {
if (insertData) {
int64_t curMs = 0;
int64_t beginMs = taosGetTimestampMs();
pInfo->startMs = beginMs;
int64_t t = pInfo->tableBeginIndex;
for (; t <= pInfo->tableEndIndex;) {
// int64_t batch = (pInfo->tableEndIndex - t);
// batch = MIN(batch, batchNum);
pInfo->startMs = taosGetTimestampMs();
for (int64_t t = pInfo->tableBeginIndex; t < pInfo->tableEndIndex; ++t) {
int64_t batch = (pInfo->tableEndIndex - t);
batch = MIN(batch, batchNum);
int32_t len = sprintf(qstr, "insert into");
for (int32_t i = 0; i < batch; ++i) {
len += sprintf(qstr + len, " t%" PRId64 " values(now, %" PRId64 ")", t + i, t + i);
}
int32_t len = sprintf(qstr, "insert into ");
for (int32_t i = 0; i < batchNumOfTbl;) {
int64_t ts = startTimestamp;
len += sprintf(qstr + len, "%s_t%" PRId64 " values ", stbName, t);
for (int32_t j = 0; j < batchNumOfRow; j++) {
len += sprintf(qstr + len, "(%" PRId64 ", 6666) ", ts++);
}
t++;
i++;
if (t > pInfo->tableEndIndex) {
break;
}
}
int64_t startTs = taosGetTimestampUs();
TAOS_RES *pRes = taos_query(con, qstr);
code = taos_errno(pRes);
if (code != 0) {
pError("failed to insert table t%" PRId64 ", reason:%s", t, tstrerror(code));
if ((code != 0) && (code != TSDB_CODE_RPC_AUTH_REQUIRED)) {
pError("failed to insert %s_t%" PRId64 ", reason:%s", stbName, t, tstrerror(code));
}
taos_free_result(pRes);
int64_t endTs = taosGetTimestampUs();
int64_t delay = endTs - startTs;
// printf("==== %"PRId64" - %"PRId64", %"PRId64"\n", startTs, endTs, delay);
if (delay > pInfo->maxDelay) pInfo->maxDelay = delay;
if (delay < pInfo->minDelay) pInfo->minDelay = delay;
curMs = taosGetTimestampMs();
if (curMs - beginMs > 10000) {
beginMs = curMs;
// printf("==== tableBeginIndex: %"PRId64", t: %"PRId64"\n", pInfo->tableBeginIndex, t);
printInsertProgress(pInfo, t);
}
t += (batch - 1);
}
printInsertProgress(pInfo, pInfo->tableEndIndex);
printInsertProgress(pInfo, t);
}
taos_close(con);
@ -280,9 +303,13 @@ void printHelp() {
printf("%s%s\n", indent, "-i");
printf("%s%s%s%d\n", indent, indent, "insertData, default is ", insertData);
printf("%s%s\n", indent, "-b");
printf("%s%s%s%d\n", indent, indent, "batchNum, default is ", batchNum);
printf("%s%s%s%d\n", indent, indent, "batchNumOfTbl, default is ", batchNumOfTbl);
printf("%s%s\n", indent, "-w");
printf("%s%s%s%d\n", indent, indent, "showTablesFlag, default is ", showTablesFlag);
printf("%s%s\n", indent, "-q");
printf("%s%s%s%d\n", indent, indent, "queryFlag, default is ", queryFlag);
printf("%s%s\n", indent, "-l");
printf("%s%s%s%d\n", indent, indent, "batchNumOfRow, default is ", batchNumOfRow);
exit(EXIT_SUCCESS);
}
@ -309,10 +336,15 @@ void parseArgument(int32_t argc, char *argv[]) {
} else if (strcmp(argv[i], "-i") == 0) {
insertData = atoi(argv[++i]);
} else if (strcmp(argv[i], "-b") == 0) {
batchNum = atoi(argv[++i]);
batchNumOfTbl = atoi(argv[++i]);
} else if (strcmp(argv[i], "-l") == 0) {
batchNumOfRow = atoi(argv[++i]);
} else if (strcmp(argv[i], "-w") == 0) {
showTablesFlag = atoi(argv[++i]);
} else if (strcmp(argv[i], "-q") == 0) {
queryFlag = atoi(argv[++i]);
} else {
pPrint("%s unknow para: %s %s", GREEN, argv[++i], NC);
}
}
@ -324,8 +356,10 @@ void parseArgument(int32_t argc, char *argv[]) {
pPrint("%s numOfVgroups:%d %s", GREEN, numOfVgroups, NC);
pPrint("%s createTable:%d %s", GREEN, createTable, NC);
pPrint("%s insertData:%d %s", GREEN, insertData, NC);
pPrint("%s batchNum:%d %s", GREEN, batchNum, NC);
pPrint("%s batchNumOfTbl:%d %s", GREEN, batchNumOfTbl, NC);
pPrint("%s batchNumOfRow:%d %s", GREEN, batchNumOfRow, NC);
pPrint("%s showTablesFlag:%d %s", GREEN, showTablesFlag, NC);
pPrint("%s queryFlag:%d %s", GREEN, queryFlag, NC);
pPrint("%s start create table performace test %s", GREEN, NC);
}
@ -338,8 +372,15 @@ int32_t main(int32_t argc, char *argv[]) {
return 0;
}
createDbAndStb();
if (queryFlag) {
//selectRowsFromTable();
return 0;
}
if (createTable) {
createDbAndStb();
}
pPrint("%d threads are spawned to create %" PRId64 " tables", numOfThreads, numOfTables);
pthread_attr_t thattr;
@ -396,9 +437,11 @@ int32_t main(int32_t argc, char *argv[]) {
insertDataSpeed += pInfo[i].insertDataSpeed;
}
pPrint("%s total %" PRId64 " tables, %.1f tables/second, threads:%d, maxDelay: %" PRId64 "us, minDelay: %" PRId64
if (createTable) {
pPrint("%s total %" PRId64 " tables, %.1f tables/second, threads:%d, maxDelay: %" PRId64 "us, minDelay: %" PRId64
"us %s",
GREEN, numOfTables, createTableSpeed, numOfThreads, maxDelay, minDelay, NC);
}
if (insertData) {
pPrint("%s total %" PRId64 " tables, %.1f rows/second, threads:%d %s", GREEN, numOfTables, insertDataSpeed,