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

feature/qnode

This commit is contained in:
dapan 2022-04-24 20:11:34 +08:00
parent 7b8c32a3df
commit bb33139572
11 changed files with 161 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -51,7 +51,7 @@ typedef struct SMetaData {
SArray *pTableMeta; // STableMeta array
SArray *pVgroupInfo; // SVgroupInfo list
SArray *pUdfList; // udf info list
SArray *pEpSetList; // qnode epset list, SArray<SEpSet>
SArray *pQnodeList; // qnode list, SArray<SQueryNodeAddr>
} SMetaData;
typedef struct SCatalogCfg {

11
source/common/src/tmsg.c
View File

@ -2091,10 +2091,15 @@ int32_t tDeserializeSQnodeListRsp(void *buf, int32_t bufLen, SQnodeListRsp *pRsp
if (tStartDecode(&decoder) < 0) return -1;
int32_t num = 0;
if (tDecodeI32(&decoder, &num) < 0) return -1;
pRsp->addrsList = taosArrayInit(num, sizeof(SQueryNodeAddr));
if (NULL == pRsp->addrsList) return -1;
if (NULL == pRsp->addrsList) {
pRsp->addrsList = taosArrayInit(num, sizeof(SQueryNodeAddr));
if (NULL == pRsp->addrsList) return -1;
}
for (int32_t i = 0; i < num; ++i) {
if (tDecodeSQueryNodeAddr(&decoder, TARRAY_GET_ELEM(pRsp->addrsList, i)) < 0) return -1;
SQueryNodeAddr addr = {0};
if (tDecodeSQueryNodeAddr(&decoder, &addr) < 0) return -1;
taosArrayPush(pRsp->addrsList, &addr);
}
tEndDecode(&decoder);

1
source/dnode/mgmt/mgmt_mnode/src/mmHandle.c
View File

@ -177,6 +177,7 @@ void mmInitMsgHandle(SMgmtWrapper *pWrapper) {
dmSetMsgHandle(pWrapper, TDMT_MND_DROP_MNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_MND_CREATE_QNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_MND_DROP_QNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_MND_QNODE_LIST, mmProcessReadMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_MND_CREATE_SNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_MND_DROP_SNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_MND_CREATE_BNODE, mmProcessWriteMsg, DEFAULT_HANDLE);

5
source/dnode/mnode/impl/src/mndQnode.c
View File

@ -451,8 +451,9 @@ static int32_t mndProcessQnodeListReq(SNodeMsg *pReq) {
goto _OVER;
}
void *pIter = NULL;
while (1) {
void *pIter = sdbFetch(pSdb, SDB_QNODE, NULL, (void **)&pObj);
pIter = sdbFetch(pSdb, SDB_QNODE, pIter, (void **)&pObj);
if (pIter == NULL) break;
SQueryNodeAddr nodeAddr = {0};
@ -472,7 +473,7 @@ static int32_t mndProcessQnodeListReq(SNodeMsg *pReq) {
}
int32_t rspLen = tSerializeSQnodeListRsp(NULL, 0, &qlistRsp);
void *pRsp = taosMemoryMalloc(rspLen);
void *pRsp = rpcMallocCont(rspLen);
if (pRsp == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto _OVER;

2
source/dnode/qnode/src/qnode.c
View File

@ -43,7 +43,7 @@ void qndClose(SQnode *pQnode) {
int32_t qndGetLoad(SQnode *pQnode, SQnodeLoad *pLoad) { return 0; }
int32_t qndProcessQueryMsg(SQnode *pQnode, SRpcMsg *pMsg) {
qTrace("message in query queue is processing");
qTrace("message in qnode query queue is processing");
SReadHandle handle = {0};
switch (pMsg->msgType) {

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

@ -135,7 +135,7 @@ int vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg
}
int vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
vTrace("message in query queue is processing");
vTrace("message in vnode query queue is processing");
SReadHandle handle = {.reader = pVnode->pTsdb, .meta = pVnode->pMeta, .config = &pVnode->config};
switch (pMsg->msgType) {

9
source/libs/catalog/src/catalog.c
View File

@ -494,7 +494,7 @@ _return:
return TSDB_CODE_SUCCESS;
}
int32_t ctgGetQnodeListFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, SArray **out) {
int32_t ctgGetQnodeListFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, SArray *out) {
char *msg = NULL;
int32_t msgLen = 0;
@ -526,7 +526,7 @@ int32_t ctgGetQnodeListFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmt
CTG_ERR_RET(code);
}
ctgDebug("Got qnode list from mnode, listNum:%d", (int32_t)taosArrayGetSize(*out));
ctgDebug("Got qnode list from mnode, listNum:%d", (int32_t)taosArrayGetSize(out));
return TSDB_CODE_SUCCESS;
}
@ -2778,7 +2778,8 @@ int32_t catalogGetAllMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps,
}
if (pReq->qNodeRequired) {
CTG_ERR_JRET(ctgGetQnodeListFromMnode(pCtg, pTrans, pMgmtEps, &pRsp->pEpSetList));
pRsp->pQnodeList = taosArrayInit(10, sizeof(SQueryNodeAddr));
CTG_ERR_JRET(ctgGetQnodeListFromMnode(pCtg, pTrans, pMgmtEps, pRsp->pQnodeList));
}
CTG_API_LEAVE(TSDB_CODE_SUCCESS);
@ -2807,7 +2808,7 @@ int32_t catalogGetQnodeList(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps,
CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
}
CTG_ERR_JRET(ctgGetQnodeListFromMnode(pCtg, pRpc, pMgmtEps, &pQnodeList));
CTG_ERR_JRET(ctgGetQnodeListFromMnode(pCtg, pRpc, pMgmtEps, pQnodeList));
_return:

16
source/libs/qcom/src/querymsg.c
View File

@ -342,28 +342,20 @@ PROCESS_META_OVER:
int32_t queryProcessQnodeListRsp(void *output, char *msg, int32_t msgSize) {
SQnodeListRsp out = {0};
int32_t code = -1;
int32_t code = 0;
if (NULL == output || NULL == msg || msgSize <= 0) {
code = TSDB_CODE_TSC_INVALID_INPUT;
goto PROCESS_QLIST_OVER;
return code;
}
out.addrsList = (SArray *)output;
if (tDeserializeSQnodeListRsp(msg, msgSize, &out) != 0) {
qError("invalid qnode list rsp msg, msgSize:%d", msgSize);
code = TSDB_CODE_INVALID_MSG;
goto PROCESS_QLIST_OVER;
return code;
}
PROCESS_QLIST_OVER:
if (code != 0) {
tFreeSQnodeListRsp(&out);
out.addrsList = NULL;
}
*(SArray **)output = out.addrsList;
return code;
}

2
source/libs/scalar/inc/sclInt.h
View File

@ -32,7 +32,7 @@ typedef struct SScalarCtx {
#define SCL_DATA_TYPE_DUMMY_HASH 9000
#define SCL_DEFAULT_OP_NUM 10
#define SCL_IS_CONST_NODE(_node) ((NULL == (_node)) || (QUERY_NODE_VALUE == (_node)->type))
#define SCL_IS_CONST_NODE(_node) ((NULL == (_node)) || (QUERY_NODE_VALUE == (_node)->type) || (QUERY_NODE_NODE_LIST == (_node)->type))
#define SCL_IS_CONST_CALC(_ctx) (NULL == (_ctx)->pBlockList)
#define sclFatal(...) qFatal(__VA_ARGS__)

36
source/libs/scalar/src/scalar.c
View File

@ -244,9 +244,8 @@ int32_t sclInitParam(SNode* node, SScalarParam *param, SScalarCtx *ctx, int32_t
return TSDB_CODE_SUCCESS;
}
int32_t sclInitParamList(SScalarParam **pParams, SNodeList* pParamList, SScalarCtx *ctx, int32_t *rowNum) {
int32_t sclInitParamList(SScalarParam **pParams, SNodeList* pParamList, SScalarCtx *ctx, int32_t *paramNum, int32_t *rowNum) {
int32_t code = 0;
int32_t paramNum = 0;
if (NULL == pParamList) {
if (ctx->pBlockList) {
SSDataBlock *pBlock = taosArrayGet(ctx->pBlockList, 0);
@ -255,14 +254,14 @@ int32_t sclInitParamList(SScalarParam **pParams, SNodeList* pParamList, SScalarC
*rowNum = 1;
}
paramNum = 1;
*paramNum = 1;
} else {
paramNum = pParamList->length;
*paramNum = pParamList->length;
}
SScalarParam *paramList = taosMemoryCalloc(paramNum, sizeof(SScalarParam));
SScalarParam *paramList = taosMemoryCalloc(*paramNum, sizeof(SScalarParam));
if (NULL == paramList) {
sclError("calloc %d failed", (int32_t)(paramNum * sizeof(SScalarParam)));
sclError("calloc %d failed", (int32_t)((*paramNum) * sizeof(SScalarParam)));
SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
@ -272,14 +271,13 @@ int32_t sclInitParamList(SScalarParam **pParams, SNodeList* pParamList, SScalarC
if (SCL_IS_CONST_CALC(ctx)) {
WHERE_EACH (tnode, pParamList) {
if (!SCL_IS_CONST_NODE(tnode)) {
continue;
WHERE_NEXT;
} else {
SCL_ERR_JRET(sclInitParam(tnode, &paramList[i], ctx, rowNum));
ERASE_NODE(pParamList);
}
++i;
WHERE_NEXT;
}
} else {
FOREACH(tnode, pParamList) {
@ -340,7 +338,8 @@ int32_t sclExecFunction(SFunctionNode *node, SScalarCtx *ctx, SScalarParam *outp
SScalarParam *params = NULL;
int32_t rowNum = 0;
SCL_ERR_RET(sclInitParamList(&params, node->pParameterList, ctx, &rowNum));
int32_t paramNum = 0;
SCL_ERR_RET(sclInitParamList(&params, node->pParameterList, ctx, &paramNum, &rowNum));
output->columnData = createColumnInfoData(&node->node.resType, rowNum);
if (output->columnData == NULL) {
@ -348,7 +347,7 @@ int32_t sclExecFunction(SFunctionNode *node, SScalarCtx *ctx, SScalarParam *outp
SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
code = (*ffpSet.process)(params, node->pParameterList->length, output);
code = (*ffpSet.process)(params, paramNum, output);
if (code) {
sclError("scalar function exec failed, funcId:%d, code:%s", node->funcId, tstrerror(code));
SCL_ERR_JRET(code);
@ -356,7 +355,7 @@ int32_t sclExecFunction(SFunctionNode *node, SScalarCtx *ctx, SScalarParam *outp
_return:
for (int32_t i = 0; i < node->pParameterList->length; ++i) {
for (int32_t i = 0; i < paramNum; ++i) {
// sclFreeParamNoData(params + i);
}
@ -382,8 +381,9 @@ int32_t sclExecLogic(SLogicConditionNode *node, SScalarCtx *ctx, SScalarParam *o
SScalarParam *params = NULL;
int32_t rowNum = 0;
int32_t paramNum = 0;
int32_t code = 0;
SCL_ERR_RET(sclInitParamList(&params, node->pParameterList, ctx, &rowNum));
SCL_ERR_RET(sclInitParamList(&params, node->pParameterList, ctx, &paramNum, &rowNum));
if (NULL == params) {
output->numOfRows = 0;
return TSDB_CODE_SUCCESS;
@ -400,10 +400,12 @@ int32_t sclExecLogic(SLogicConditionNode *node, SScalarCtx *ctx, SScalarParam *o
}
bool value = false;
bool complete = false;
bool complete = true;
for (int32_t i = 0; i < rowNum; ++i) {
for (int32_t m = 0; m < node->pParameterList->length; ++m) {
complete = true;
for (int32_t m = 0; m < paramNum; ++m) {
if (NULL == params[m].columnData) {
complete = false;
continue;
}
char* p = colDataGetData(params[m].columnData, i);
@ -420,7 +422,9 @@ int32_t sclExecLogic(SLogicConditionNode *node, SScalarCtx *ctx, SScalarParam *o
}
}
colDataAppend(output->columnData, i, (char*) &value, false);
if (complete) {
colDataAppend(output->columnData, i, (char*) &value, false);
}
}
if (SCL_IS_CONST_CALC(ctx) && (false == complete)) {
@ -430,7 +434,7 @@ int32_t sclExecLogic(SLogicConditionNode *node, SScalarCtx *ctx, SScalarParam *o
_return:
for (int32_t i = 0; i < node->pParameterList->length; ++i) {
for (int32_t i = 0; i < paramNum; ++i) {
// sclFreeParamNoData(params + i);
}

116
source/libs/scalar/test/scalar/scalarTests.cpp
View File

@ -137,6 +137,11 @@ void scltMakeColumnNode(SNode **pNode, SSDataBlock **block, int32_t dataType, in
rnode->node.resType.bytes = dataBytes;
rnode->dataBlockId = 0;
if (NULL == block) {
*pNode = (SNode *)rnode;
return;
}
if (NULL == *block) {
SSDataBlock *res = (SSDataBlock *)taosMemoryCalloc(1, sizeof(SSDataBlock));
res->info.numOfCols = 3;
@ -889,6 +894,8 @@ TEST(constantTest, int_greater_int_is_true2) {
}
TEST(constantTest, greater_and_lower) {
scltInitLogFile();
SNode *pval1 = NULL, *pval2 = NULL, *opNode1 = NULL, *opNode2 = NULL, *logicNode = NULL, *res = NULL;
bool eRes[5] = {false, false, true, true, true};
int64_t v1 = 333, v2 = 222, v3 = -10, v4 = 20;
@ -913,6 +920,115 @@ TEST(constantTest, greater_and_lower) {
nodesDestroyNode(res);
}
TEST(constantTest, column_and_value1) {
scltInitLogFile();
SNode *pval1 = NULL, *pval2 = NULL, *opNode1 = NULL, *opNode2 = NULL, *logicNode = NULL, *res = NULL;
bool eRes[5] = {false, false, true, true, true};
int64_t v1 = 333, v2 = 222, v3 = -10, v4 = 20;
SNode *list[2] = {0};
scltMakeValueNode(&pval1, TSDB_DATA_TYPE_BIGINT, &v1);
scltMakeValueNode(&pval2, TSDB_DATA_TYPE_BIGINT, &v2);
scltMakeOpNode(&opNode1, OP_TYPE_GREATER_THAN, TSDB_DATA_TYPE_BOOL, pval1, pval2);
scltMakeValueNode(&pval1, TSDB_DATA_TYPE_BIGINT, &v3);
scltMakeColumnNode(&pval2, NULL, TSDB_DATA_TYPE_BIGINT, sizeof(int64_t), 0, NULL);
scltMakeOpNode(&opNode2, OP_TYPE_LOWER_THAN, TSDB_DATA_TYPE_BOOL, pval1, pval2);
list[0] = opNode1;
list[1] = opNode2;
scltMakeLogicNode(&logicNode, LOGIC_COND_TYPE_AND, list, 2);
int32_t code = scalarCalculateConstants(logicNode, &res);
ASSERT_EQ(code, 0);
ASSERT_TRUE(res);
ASSERT_EQ(nodeType(res), QUERY_NODE_LOGIC_CONDITION);
SLogicConditionNode *v = (SLogicConditionNode *)res;
ASSERT_EQ(v->condType, LOGIC_COND_TYPE_AND);
ASSERT_EQ(v->pParameterList->length, 1);
nodesDestroyNode(res);
}
TEST(constantTest, column_and_value2) {
scltInitLogFile();
SNode *pval1 = NULL, *pval2 = NULL, *opNode1 = NULL, *opNode2 = NULL, *logicNode = NULL, *res = NULL;
bool eRes[5] = {false, false, true, true, true};
int64_t v1 = 333, v2 = 222, v3 = -10, v4 = 20;
SNode *list[2] = {0};
scltMakeValueNode(&pval1, TSDB_DATA_TYPE_BIGINT, &v1);
scltMakeValueNode(&pval2, TSDB_DATA_TYPE_BIGINT, &v2);
scltMakeOpNode(&opNode1, OP_TYPE_LOWER_THAN, TSDB_DATA_TYPE_BOOL, pval1, pval2);
scltMakeValueNode(&pval1, TSDB_DATA_TYPE_BIGINT, &v3);
scltMakeColumnNode(&pval2, NULL, TSDB_DATA_TYPE_BIGINT, sizeof(int64_t), 0, NULL);
scltMakeOpNode(&opNode2, OP_TYPE_LOWER_THAN, TSDB_DATA_TYPE_BOOL, pval1, pval2);
list[0] = opNode1;
list[1] = opNode2;
scltMakeLogicNode(&logicNode, LOGIC_COND_TYPE_AND, list, 2);
int32_t code = scalarCalculateConstants(logicNode, &res);
ASSERT_EQ(code, 0);
ASSERT_TRUE(res);
ASSERT_EQ(nodeType(res), QUERY_NODE_VALUE);
SValueNode *v = (SValueNode *)res;
ASSERT_EQ(v->node.resType.type, TSDB_DATA_TYPE_BOOL);
ASSERT_EQ(v->datum.b, false);
nodesDestroyNode(res);
}
TEST(constantTest, column_and_value3) {
scltInitLogFile();
SNode *pval1 = NULL, *pval2 = NULL, *opNode1 = NULL, *opNode2 = NULL, *logicNode = NULL, *res = NULL;
bool eRes[5] = {false, false, true, true, true};
int64_t v1 = 333, v2 = 222, v3 = -10, v4 = 20;
SNode *list[2] = {0};
scltMakeValueNode(&pval1, TSDB_DATA_TYPE_BIGINT, &v1);
scltMakeValueNode(&pval2, TSDB_DATA_TYPE_BIGINT, &v2);
scltMakeOpNode(&opNode1, OP_TYPE_GREATER_THAN, TSDB_DATA_TYPE_BOOL, pval1, pval2);
scltMakeValueNode(&pval1, TSDB_DATA_TYPE_BIGINT, &v3);
scltMakeColumnNode(&pval2, NULL, TSDB_DATA_TYPE_BIGINT, sizeof(int64_t), 0, NULL);
scltMakeOpNode(&opNode2, OP_TYPE_LOWER_THAN, TSDB_DATA_TYPE_BOOL, pval1, pval2);
list[0] = opNode1;
list[1] = opNode2;
scltMakeLogicNode(&logicNode, LOGIC_COND_TYPE_OR, list, 2);
int32_t code = scalarCalculateConstants(logicNode, &res);
ASSERT_EQ(code, 0);
ASSERT_TRUE(res);
ASSERT_EQ(nodeType(res), QUERY_NODE_VALUE);
SValueNode *v = (SValueNode *)res;
ASSERT_EQ(v->node.resType.type, TSDB_DATA_TYPE_BOOL);
ASSERT_EQ(v->datum.b, true);
nodesDestroyNode(res);
}
TEST(constantTest, column_and_value4) {
scltInitLogFile();
SNode *pval1 = NULL, *pval2 = NULL, *opNode1 = NULL, *opNode2 = NULL, *logicNode = NULL, *res = NULL;
bool eRes[5] = {false, false, true, true, true};
int64_t v1 = 333, v2 = 222, v3 = -10, v4 = 20;
SNode *list[2] = {0};
scltMakeValueNode(&pval1, TSDB_DATA_TYPE_BIGINT, &v1);
scltMakeValueNode(&pval2, TSDB_DATA_TYPE_BIGINT, &v2);
scltMakeOpNode(&opNode1, OP_TYPE_LOWER_THAN, TSDB_DATA_TYPE_BOOL, pval1, pval2);
scltMakeValueNode(&pval1, TSDB_DATA_TYPE_BIGINT, &v3);
scltMakeColumnNode(&pval2, NULL, TSDB_DATA_TYPE_BIGINT, sizeof(int64_t), 0, NULL);
scltMakeOpNode(&opNode2, OP_TYPE_LOWER_THAN, TSDB_DATA_TYPE_BOOL, pval1, pval2);
list[0] = opNode1;
list[1] = opNode2;
scltMakeLogicNode(&logicNode, LOGIC_COND_TYPE_OR, list, 2);
int32_t code = scalarCalculateConstants(logicNode, &res);
ASSERT_EQ(code, 0);
ASSERT_TRUE(res);
ASSERT_EQ(nodeType(res), QUERY_NODE_LOGIC_CONDITION);
SLogicConditionNode *v = (SLogicConditionNode *)res;
ASSERT_EQ(v->condType, LOGIC_COND_TYPE_OR);
ASSERT_EQ(v->pParameterList->length, 1);
nodesDestroyNode(res);
}
void makeJsonArrow(SSDataBlock **src, SNode **opNode, void *json, char *key){
char keyVar[32] = {0};
memcpy(varDataVal(keyVar), key, strlen(key));