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

Merge pull request #26714 from taosdata/enh/TD-31063 

enh:[TD-31063] Handling return value in qnode
This commit is contained in:
dapan1121 2024-07-23 18:06:10 +08:00 committed by GitHub
parent 57284ca121 6255a7ba10
commit 2220a93b41
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 269 additions and 105 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
include/dnode/qnode/qnode.h
View File

@ -34,9 +34,10 @@ typedef struct {
* @brief Start one Qnode in Dnode.
*
* @param pOption Option of the qnode.
* @return SQnode* The qnode object.
* @param pQnode The qnode object.
* @return int32_t The error code.
*/
SQnode *qndOpen(const SQnodeOpt *pOption);
int32_t qndOpen(const SQnodeOpt *pOption, SQnode **pQnode);
/**
* @brief Stop Qnode in Dnode.

9
source/dnode/mgmt/mgmt_qnode/src/qmInt.c
View File

@ -34,11 +34,8 @@ static void qmClose(SQnodeMgmt *pMgmt) {
}
static int32_t qndOpenWrapper(SQnodeOpt *pOption, SQnode **pQnode) {
*pQnode = qndOpen(pOption);
if (*pQnode == NULL) {
return terrno;
}
return 0;
int32_t code = qndOpen(pOption, pQnode);
return code;
}
static int32_t qmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
int32_t code = 0;
@ -62,7 +59,7 @@ static int32_t qmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
if (code != 0) {
dError("failed to open qnode since %s", tstrerror(code));
qmClose(pMgmt);
return -1;
return code;
}
tmsgReportStartup("qnode-impl", "initialized");

4
source/dnode/mgmt/mgmt_qnode/src/qmWorker.c
View File

@ -48,8 +48,8 @@ static int32_t qmPutNodeMsgToWorker(SSingleWorker *pWorker, SRpcMsg *pMsg) {
}
int32_t qmPutNodeMsgToQueryQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg) {
qndPreprocessQueryMsg(pMgmt->pQnode, pMsg);
int32_t code = qndPreprocessQueryMsg(pMgmt->pQnode, pMsg);
if (code) return code;
return qmPutNodeMsgToWorker(&pMgmt->queryWorker, pMsg);
}

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

@ -19,21 +19,22 @@
#include "query.h"
#include "qworker.h"
SQnode *qndOpen(const SQnodeOpt *pOption) {
SQnode *pQnode = taosMemoryCalloc(1, sizeof(SQnode));
if (NULL == pQnode) {
int32_t qndOpen(const SQnodeOpt *pOption, SQnode **pQnode) {
*pQnode = taosMemoryCalloc(1, sizeof(SQnode));
if (NULL == *pQnode) {
qError("calloc SQnode failed");
return NULL;
return TSDB_CODE_OUT_OF_MEMORY;
}
pQnode->qndId = QNODE_HANDLE;
(*pQnode)->qndId = QNODE_HANDLE;
if (qWorkerInit(NODE_TYPE_QNODE, pQnode->qndId, (void **)&pQnode->pQuery, &pOption->msgCb)) {
int32_t code = qWorkerInit(NODE_TYPE_QNODE, (*pQnode)->qndId, (void **)&(*pQnode)->pQuery, &pOption->msgCb);
if (TSDB_CODE_SUCCESS != code) {
taosMemoryFreeClear(pQnode);
return NULL;
return code;
}
pQnode->msgCb = pOption->msgCb;
return pQnode;
(*pQnode)->msgCb = pOption->msgCb;
return TSDB_CODE_SUCCESS;
}
void qndClose(SQnode *pQnode) {

9
source/dnode/vnode/src/meta/metaQuery.c
View File

@ -1542,9 +1542,14 @@ int32_t metaGetStbStats(void *pVnode, int64_t uid, int64_t *numOfTables, int32_t
// slow path: search TDB
int64_t ctbNum = 0;
int32_t colNum = 0;
vnodeGetCtbNum(pVnode, uid, &ctbNum);
vnodeGetStbColumnNum(pVnode, uid, &colNum);
code = vnodeGetCtbNum(pVnode, uid, &ctbNum);
if (TSDB_CODE_SUCCESS == code) {
code = vnodeGetStbColumnNum(pVnode, uid, &colNum);
}
metaULock(pVnodeObj->pMeta);
if (TSDB_CODE_SUCCESS != code) {
goto _exit;
}
if (numOfTables) *numOfTables = ctbNum;
if (numOfCols) *numOfCols = colNum;

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

@ -47,7 +47,7 @@ int32_t fillTableColCmpr(SMetaReader *reader, SSchemaExt *pExt, int32_t numOfCol
return 0;
}
int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
int32_t vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
STableInfoReq infoReq = {0};
STableMetaRsp metaRsp = {0};
SMetaReader mer1 = {0};
@ -62,15 +62,15 @@ int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
// decode req
if (tDeserializeSTableInfoReq(pMsg->pCont, pMsg->contLen, &infoReq) != 0) {
code = TSDB_CODE_INVALID_MSG;
code = terrno;
goto _exit4;
}
metaRsp.dbId = pVnode->config.dbId;
strcpy(metaRsp.tbName, infoReq.tbName);
memcpy(metaRsp.dbFName, infoReq.dbFName, sizeof(metaRsp.dbFName));
(void)strcpy(metaRsp.tbName, infoReq.tbName);
(void)memcpy(metaRsp.dbFName, infoReq.dbFName, sizeof(metaRsp.dbFName));
sprintf(tableFName, "%s.%s", infoReq.dbFName, infoReq.tbName);
(void)sprintf(tableFName, "%s.%s", infoReq.dbFName, infoReq.tbName);
code = vnodeValidateTableHash(pVnode, tableFName);
if (code) {
goto _exit4;
@ -89,7 +89,7 @@ int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
metaRsp.tuid = mer1.me.uid;
if (mer1.me.type == TSDB_SUPER_TABLE) {
strcpy(metaRsp.stbName, mer1.me.name);
(void)strcpy(metaRsp.stbName, mer1.me.name);
schema = mer1.me.stbEntry.schemaRow;
schemaTag = mer1.me.stbEntry.schemaTag;
metaRsp.suid = mer1.me.uid;
@ -97,7 +97,7 @@ int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
metaReaderDoInit(&mer2, pVnode->pMeta, META_READER_NOLOCK);
if (metaReaderGetTableEntryByUid(&mer2, mer1.me.ctbEntry.suid) < 0) goto _exit2;
strcpy(metaRsp.stbName, mer2.me.name);
(void)strcpy(metaRsp.stbName, mer2.me.name);
metaRsp.suid = mer2.me.uid;
schema = mer2.me.stbEntry.schemaRow;
schemaTag = mer2.me.stbEntry.schemaTag;
@ -114,10 +114,13 @@ int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
metaRsp.tversion = schemaTag.version;
metaRsp.pSchemas = (SSchema *)taosMemoryMalloc(sizeof(SSchema) * (metaRsp.numOfColumns + metaRsp.numOfTags));
metaRsp.pSchemaExt = (SSchemaExt *)taosMemoryCalloc(metaRsp.numOfColumns, sizeof(SSchemaExt));
memcpy(metaRsp.pSchemas, schema.pSchema, sizeof(SSchema) * schema.nCols);
if (NULL == metaRsp.pSchemas || NULL == metaRsp.pSchemaExt) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
(void)memcpy(metaRsp.pSchemas, schema.pSchema, sizeof(SSchema) * schema.nCols);
if (schemaTag.nCols) {
memcpy(metaRsp.pSchemas + schema.nCols, schemaTag.pSchema, sizeof(SSchema) * schemaTag.nCols);
(void)memcpy(metaRsp.pSchemas + schema.nCols, schemaTag.pSchema, sizeof(SSchema) * schemaTag.nCols);
}
if (metaRsp.pSchemaExt) {
SMetaReader *pReader = mer1.me.type == TSDB_CHILD_TABLE ? &mer2 : &mer1;
@ -134,7 +137,7 @@ int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
// encode and send response
rspLen = tSerializeSTableMetaRsp(NULL, 0, &metaRsp);
if (rspLen < 0) {
code = TSDB_CODE_INVALID_MSG;
code = terrno;
goto _exit;
}
@ -148,7 +151,12 @@ int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
tSerializeSTableMetaRsp(pRsp, rspLen, &metaRsp);
rspLen = tSerializeSTableMetaRsp(pRsp, rspLen, &metaRsp);
if (rspLen < 0) {
code = terrno;
goto _exit;
}
_exit:
taosMemoryFree(metaRsp.pSchemas);
@ -174,10 +182,10 @@ _exit4:
*pMsg = rpcMsg;
}
return TSDB_CODE_SUCCESS;
return code;
}
int vnodeGetTableCfg(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
int32_t vnodeGetTableCfg(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
STableCfgReq cfgReq = {0};
STableCfgRsp cfgRsp = {0};
SMetaReader mer1 = {0};
@ -192,14 +200,14 @@ int vnodeGetTableCfg(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
// decode req
if (tDeserializeSTableCfgReq(pMsg->pCont, pMsg->contLen, &cfgReq) != 0) {
code = TSDB_CODE_INVALID_MSG;
code = terrno;
goto _exit;
}
strcpy(cfgRsp.tbName, cfgReq.tbName);
memcpy(cfgRsp.dbFName, cfgReq.dbFName, sizeof(cfgRsp.dbFName));
(void)strcpy(cfgRsp.tbName, cfgReq.tbName);
(void)memcpy(cfgRsp.dbFName, cfgReq.dbFName, sizeof(cfgRsp.dbFName));
sprintf(tableFName, "%s.%s", cfgReq.dbFName, cfgReq.tbName);
(void)sprintf(tableFName, "%s.%s", cfgReq.dbFName, cfgReq.tbName);
code = vnodeValidateTableHash(pVnode, tableFName);
if (code) {
goto _exit;
@ -222,24 +230,36 @@ int vnodeGetTableCfg(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
metaReaderDoInit(&mer2, pVnode->pMeta, META_READER_LOCK);
if (metaReaderGetTableEntryByUid(&mer2, mer1.me.ctbEntry.suid) < 0) goto _exit;
strcpy(cfgRsp.stbName, mer2.me.name);
(void)strcpy(cfgRsp.stbName, mer2.me.name);
schema = mer2.me.stbEntry.schemaRow;
schemaTag = mer2.me.stbEntry.schemaTag;
cfgRsp.ttl = mer1.me.ctbEntry.ttlDays;
cfgRsp.commentLen = mer1.me.ctbEntry.commentLen;
if (mer1.me.ctbEntry.commentLen > 0) {
cfgRsp.pComment = taosStrdup(mer1.me.ctbEntry.comment);
if (NULL == cfgRsp.pComment) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
}
STag *pTag = (STag *)mer1.me.ctbEntry.pTags;
cfgRsp.tagsLen = pTag->len;
cfgRsp.pTags = taosMemoryMalloc(cfgRsp.tagsLen);
memcpy(cfgRsp.pTags, pTag, cfgRsp.tagsLen);
if (NULL == cfgRsp.pTags) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
(void)memcpy(cfgRsp.pTags, pTag, cfgRsp.tagsLen);
} else if (mer1.me.type == TSDB_NORMAL_TABLE) {
schema = mer1.me.ntbEntry.schemaRow;
cfgRsp.ttl = mer1.me.ntbEntry.ttlDays;
cfgRsp.commentLen = mer1.me.ntbEntry.commentLen;
if (mer1.me.ntbEntry.commentLen > 0) {
cfgRsp.pComment = taosStrdup(mer1.me.ntbEntry.comment);
if (NULL == cfgRsp.pComment) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
}
} else {
ASSERT(0);
@ -250,9 +270,13 @@ int vnodeGetTableCfg(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
cfgRsp.pSchemas = (SSchema *)taosMemoryMalloc(sizeof(SSchema) * (cfgRsp.numOfColumns + cfgRsp.numOfTags));
cfgRsp.pSchemaExt = (SSchemaExt *)taosMemoryMalloc(cfgRsp.numOfColumns * sizeof(SSchemaExt));
memcpy(cfgRsp.pSchemas, schema.pSchema, sizeof(SSchema) * schema.nCols);
if (NULL == cfgRsp.pSchemas || NULL == cfgRsp.pSchemaExt) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
(void)memcpy(cfgRsp.pSchemas, schema.pSchema, sizeof(SSchema) * schema.nCols);
if (schemaTag.nCols) {
memcpy(cfgRsp.pSchemas + schema.nCols, schemaTag.pSchema, sizeof(SSchema) * schemaTag.nCols);
(void)memcpy(cfgRsp.pSchemas + schema.nCols, schemaTag.pSchema, sizeof(SSchema) * schemaTag.nCols);
}
// if (useCompress(cfgRsp.tableType)) {
@ -271,7 +295,7 @@ int vnodeGetTableCfg(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
// encode and send response
rspLen = tSerializeSTableCfgRsp(NULL, 0, &cfgRsp);
if (rspLen < 0) {
code = TSDB_CODE_INVALID_MSG;
code = terrno;
goto _exit;
}
@ -285,7 +309,12 @@ int vnodeGetTableCfg(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
tSerializeSTableCfgRsp(pRsp, rspLen, &cfgRsp);
rspLen = tSerializeSTableCfgRsp(pRsp, rspLen, &cfgRsp);
if (rspLen < 0) {
code = terrno;
goto _exit;
}
_exit:
rpcMsg.info = pMsg->info;
@ -307,7 +336,7 @@ _exit:
tFreeSTableCfgRsp(&cfgRsp);
metaReaderClear(&mer2);
metaReaderClear(&mer1);
return TSDB_CODE_SUCCESS;
return code;
}
static FORCE_INLINE void vnodeFreeSBatchRspMsg(void *p) {
@ -331,7 +360,7 @@ int32_t vnodeGetBatchMeta(SVnode *pVnode, SRpcMsg *pMsg) {
void *pRsp = NULL;
if (tDeserializeSBatchReq(pMsg->pCont, pMsg->contLen, &batchReq)) {
code = TSDB_CODE_OUT_OF_MEMORY;
code = terrno;
qError("tDeserializeSBatchReq failed");
goto _exit;
}
@ -352,6 +381,10 @@ int32_t vnodeGetBatchMeta(SVnode *pVnode, SRpcMsg *pMsg) {
for (int32_t i = 0; i < msgNum; ++i) {
req = taosArrayGet(batchReq.pMsgs, i);
if (req == NULL) {
code = TSDB_CODE_OUT_OF_RANGE;
goto _exit;
}
reqMsg.msgType = req->msgType;
reqMsg.pCont = req->msg;
@ -359,13 +392,16 @@ int32_t vnodeGetBatchMeta(SVnode *pVnode, SRpcMsg *pMsg) {
switch (req->msgType) {
case TDMT_VND_TABLE_META:
vnodeGetTableMeta(pVnode, &reqMsg, false);
// error code has been set into reqMsg, no need to handle it here.
(void)vnodeGetTableMeta(pVnode, &reqMsg, false);
break;
case TDMT_VND_TABLE_CFG:
vnodeGetTableCfg(pVnode, &reqMsg, false);
// error code has been set into reqMsg, no need to handle it here.
(void)vnodeGetTableCfg(pVnode, &reqMsg, false);
break;
case TDMT_VND_GET_STREAM_PROGRESS:
vnodeGetStreamProgress(pVnode, &reqMsg, false);
// error code has been set into reqMsg, no need to handle it here.
(void)vnodeGetStreamProgress(pVnode, &reqMsg, false);
break;
default:
qError("invalid req msgType %d", req->msgType);
@ -381,24 +417,28 @@ int32_t vnodeGetBatchMeta(SVnode *pVnode, SRpcMsg *pMsg) {
rsp.rspCode = reqMsg.code;
rsp.msg = reqMsg.pCont;
taosArrayPush(batchRsp.pRsps, &rsp);
if (NULL == taosArrayPush(batchRsp.pRsps, &rsp)) {
qError("taosArrayPush failed");
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
}
rspSize = tSerializeSBatchRsp(NULL, 0, &batchRsp);
if (rspSize < 0) {
qError("tSerializeSBatchRsp failed");
code = TSDB_CODE_OUT_OF_MEMORY;
code = terrno;
goto _exit;
}
pRsp = rpcMallocCont(rspSize);
if (pRsp == NULL) {
qError("rpcMallocCont %d failed", rspSize);
code = TSDB_CODE_OUT_OF_MEMORY;
code = terrno;
goto _exit;
}
if (tSerializeSBatchRsp(pRsp, rspSize, &batchRsp) < 0) {
qError("tSerializeSBatchRsp %d failed", rspSize);
code = TSDB_CODE_OUT_OF_MEMORY;
code = terrno;
goto _exit;
}
@ -501,7 +541,12 @@ int32_t vnodeGetTableList(void *pVnode, int8_t type, SArray *pList) {
}
int32_t vnodeGetAllTableList(SVnode *pVnode, uint64_t uid, SArray *list) {
int32_t code = TSDB_CODE_SUCCESS;
SMCtbCursor *pCur = metaOpenCtbCursor(pVnode, uid, 1);
if (NULL == pCur) {
qError("vnode get all table list failed");
return TSDB_CODE_FAILED;
}
while (1) {
tb_uid_t id = metaCtbCursorNext(pCur);
@ -510,11 +555,15 @@ int32_t vnodeGetAllTableList(SVnode *pVnode, uint64_t uid, SArray *list) {
}
STableKeyInfo info = {uid = id};
taosArrayPush(list, &info);
if (NULL == taosArrayPush(list, &info)) {
qError("taosArrayPush failed");
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
}
_exit:
metaCloseCtbCursor(pCur);
return TSDB_CODE_SUCCESS;
return code;
}
int32_t vnodeGetCtbIdListByFilter(SVnode *pVnode, int64_t suid, SArray *list, bool (*filter)(void *arg), void *arg) {
@ -522,8 +571,13 @@ int32_t vnodeGetCtbIdListByFilter(SVnode *pVnode, int64_t suid, SArray *list, bo
}
int32_t vnodeGetCtbIdList(void *pVnode, int64_t suid, SArray *list) {
int32_t code = TSDB_CODE_SUCCESS;
SVnode *pVnodeObj = pVnode;
SMCtbCursor *pCur = metaOpenCtbCursor(pVnodeObj, suid, 1);
if (NULL == pCur) {
qError("vnode get all table list failed");
return TSDB_CODE_FAILED;
}
while (1) {
tb_uid_t id = metaCtbCursorNext(pCur);
@ -531,14 +585,20 @@ int32_t vnodeGetCtbIdList(void *pVnode, int64_t suid, SArray *list) {
break;
}
taosArrayPush(list, &id);
if (NULL == taosArrayPush(list, &id)) {
qError("taosArrayPush failed");
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
}
_exit:
metaCloseCtbCursor(pCur);
return TSDB_CODE_SUCCESS;
return code;
}
int32_t vnodeGetStbIdList(SVnode *pVnode, int64_t suid, SArray *list) {
int32_t code = TSDB_CODE_SUCCESS;
SMStbCursor *pCur = metaOpenStbCursor(pVnode->pMeta, suid);
if (!pCur) {
return TSDB_CODE_FAILED;
@ -550,15 +610,21 @@ int32_t vnodeGetStbIdList(SVnode *pVnode, int64_t suid, SArray *list) {
break;
}
taosArrayPush(list, &id);
if (NULL == taosArrayPush(list, &id)) {
qError("taosArrayPush failed");
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
}
_exit:
metaCloseStbCursor(pCur);
return TSDB_CODE_SUCCESS;
return code;
}
int32_t vnodeGetStbIdListByFilter(SVnode *pVnode, int64_t suid, SArray *list, bool (*filter)(void *arg, void *arg1),
void *arg) {
int32_t code = TSDB_CODE_SUCCESS;
SMStbCursor *pCur = metaOpenStbCursor(pVnode->pMeta, suid);
if (!pCur) {
return TSDB_CODE_FAILED;
@ -574,11 +640,16 @@ int32_t vnodeGetStbIdListByFilter(SVnode *pVnode, int64_t suid, SArray *list, bo
continue;
}
taosArrayPush(list, &id);
if (NULL == taosArrayPush(list, &id)) {
qError("taosArrayPush failed");
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
}
_exit:
metaCloseStbCursor(pCur);
return TSDB_CODE_SUCCESS;
return code;
}
int32_t vnodeGetCtbNum(SVnode *pVnode, int64_t suid, int64_t *num) {
@ -700,16 +771,20 @@ int32_t vnodeGetTimeSeriesNum(SVnode *pVnode, int64_t *num) {
*num = 0;
int64_t arrSize = taosArrayGetSize(suidList);
int32_t code = TSDB_CODE_SUCCESS;
for (int64_t i = 0; i < arrSize; ++i) {
tb_uid_t suid = *(tb_uid_t *)taosArrayGet(suidList, i);
int64_t ctbNum = 0;
int32_t numOfCols = 0;
metaGetStbStats(pVnode, suid, &ctbNum, &numOfCols);
code = metaGetStbStats(pVnode, suid, &ctbNum, &numOfCols);
if (TSDB_CODE_SUCCESS != code) {
goto _exit;
}
*num += ctbNum * (numOfCols - 1);
}
_exit:
taosArrayDestroy(suidList);
return TSDB_CODE_SUCCESS;
}
@ -728,7 +803,11 @@ int32_t vnodeGetAllCtbNum(SVnode *pVnode, int64_t *num) {
}
int64_t ctbNum = 0;
vnodeGetCtbNum(pVnode, id, &ctbNum);
int32_t code = vnodeGetCtbNum(pVnode, id, &ctbNum);
if (TSDB_CODE_SUCCESS != code) {
metaCloseStbCursor(pCur);
return code;
}
*num += ctbNum;
}
@ -771,14 +850,17 @@ int32_t vnodeGetStreamProgress(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
rsp.vgId = req.vgId;
rsp.streamId = req.streamId;
rspLen = tSerializeStreamProgressRsp(0, 0, &rsp);
if (rspLen < 0) {
code = terrno;
goto _OVER;
}
if (direct) {
buf = rpcMallocCont(rspLen);
} else {
buf = taosMemoryCalloc(1, rspLen);
}
if (!buf) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
code = -1;
code = TSDB_CODE_OUT_OF_MEMORY;
goto _OVER;
}
}
@ -787,7 +869,11 @@ int32_t vnodeGetStreamProgress(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
code = tqGetStreamExecInfo(pVnode, req.streamId, &rsp.progressDelay, &rsp.fillHisFinished);
}
if (code == TSDB_CODE_SUCCESS) {
tSerializeStreamProgressRsp(buf, rspLen, &rsp);
rspLen = tSerializeStreamProgressRsp(buf, rspLen, &rsp);
if (rspLen < 0) {
code = terrno;
goto _OVER;
}
rpcMsg.pCont = buf;
buf = NULL;
rpcMsg.contLen = rspLen;

73
source/libs/scalar/src/filter.c
View File

@ -997,9 +997,15 @@ int32_t filterDetachCnfGroups(SArray *group, SArray *left, SArray *right) {
for (int32_t l = 0; l < leftSize; ++l) {
SFilterGroup *gp1 = taosArrayGet(left, l);
if (NULL == gp1) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_RANGE);
}
for (int32_t r = 0; r < rightSize; ++r) {
SFilterGroup *gp2 = taosArrayGet(right, r);
if (NULL == gp2) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_RANGE);
}
FLT_ERR_RET(filterDetachCnfGroup(gp1, gp2, group));
}
@ -1212,7 +1218,7 @@ int32_t filterAddUnitImpl(SFilterInfo *info, uint8_t optr, SFilterFieldId *left,
FLT_PACKAGE_UNIT_HASH_KEY(&v, optr, optr2, left->idx, (right ? right->idx : -1), (right2 ? right2->idx : -1));
if (taosHashPut(info->pctx.unitHash, v, sizeof(v), uidx, sizeof(*uidx))) {
fltError("taosHashPut to set failed");
FLT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
FLT_ERR_RET(terrno);
}
}
@ -2157,7 +2163,7 @@ int32_t fltInitValFieldData(SFilterInfo *info) {
int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(fi->data), varDataLen(fi->data), varDataVal(newValData));
if (len < 0) {
qError("filterInitValFieldData taosUcs4ToMbs error 1");
return TSDB_CODE_APP_ERROR;
return TSDB_CODE_SCALAR_CONVERT_ERROR;
}
varDataSetLen(newValData, len);
(void)varDataCopy(fi->data, newValData);
@ -2307,6 +2313,9 @@ int32_t filterMergeUnits(SFilterInfo *info, SFilterGroupCtx *gRes, uint32_t colI
for (uint32_t i = 0; i < size; ++i) {
SFilterUnit *u = taosArrayGetP(colArray, i);
if (NULL == u) {
FLT_ERR_JRET(TSDB_CODE_OUT_OF_RANGE);
}
uint8_t optr = FILTER_UNIT_OPTR(u);
FLT_ERR_RET(filterAddRangeOptr(ctx, optr, LOGIC_COND_TYPE_AND, empty, NULL));
@ -2637,6 +2646,9 @@ int32_t filterMergeTwoGroups(SFilterInfo *info, SFilterGroupCtx **gRes1, SFilter
for (int32_t i = 0; i < ctxSize; ++i) {
pctx = taosArrayGet(colCtxs, i);
if (NULL == pctx) {
FLT_ERR_JRET(TSDB_CODE_OUT_OF_RANGE);
}
colInfo = &(*gRes1)->colInfo[pctx->colIdx];
filterFreeColInfo(colInfo);
@ -2763,6 +2775,9 @@ int32_t filterConvertGroupFromArray(SFilterInfo *info, SArray *group) {
for (size_t i = 0; i < groupSize; ++i) {
SFilterGroup *pg = taosArrayGet(group, i);
if (NULL == pg) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_RANGE);
}
pg->unitFlags = taosMemoryCalloc(pg->unitNum, sizeof(*pg->unitFlags));
if (pg->unitFlags == NULL) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
@ -2819,7 +2834,9 @@ int32_t filterRewrite(SFilterInfo *info, SFilterGroupCtx **gRes, int32_t gResNum
for (int32_t n = 0; n < usize; ++n) {
SFilterUnit *u = (SFilterUnit *)taosArrayGetP((SArray *)colInfo->info, n);
if (NULL == u) {
FLT_ERR_JRET(TSDB_CODE_OUT_OF_RANGE);
}
FLT_ERR_JRET(filterAddUnitFromUnit(info, &oinfo, u, &uidx));
FLT_ERR_JRET(filterAddUnitToGroup(&ng, uidx));
}
@ -3425,7 +3442,7 @@ int32_t filterExecuteImplMisc(void *pinfo, int32_t numOfRows, SColumnInfoData *p
if (len < 0) {
qError("castConvert1 taosUcs4ToMbs error");
taosMemoryFreeClear(newColData);
FLT_ERR_RET(TSDB_CODE_APP_ERROR);
FLT_ERR_RET(TSDB_CODE_SCALAR_CONVERT_ERROR);
} else {
varDataSetLen(newColData, len);
p[i] = filterDoCompare(gDataCompare[info->cunits[uidx].func], info->cunits[uidx].optr, newColData,
@ -3500,7 +3517,7 @@ int32_t filterExecuteImpl(void *pinfo, int32_t numOfRows, SColumnInfoData *pRes,
if (len < 0) {
qError("castConvert1 taosUcs4ToMbs error");
taosMemoryFreeClear(newColData);
FLT_ERR_RET(TSDB_CODE_APP_ERROR);
FLT_ERR_RET(TSDB_CODE_SCALAR_CONVERT_ERROR);
} else {
varDataSetLen(newColData, len);
p[i] = filterDoCompare(gDataCompare[cunit->func], cunit->optr, newColData, cunit->valData);
@ -3814,6 +3831,9 @@ int32_t fltSclMergeSort(SArray *pts1, SArray *pts2, SArray *result) {
while (i < len1 && j < len2) {
SFltSclPoint *pt1 = taosArrayGet(pts1, i);
SFltSclPoint *pt2 = taosArrayGet(pts2, j);
if (NULL == pt1 || NULL == pt2) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_RANGE);
}
bool less = fltSclLessPoint(pt1, pt2);
if (less) {
if (NULL == taosArrayPush(result, pt1)) {
@ -3830,6 +3850,9 @@ int32_t fltSclMergeSort(SArray *pts1, SArray *pts2, SArray *result) {
if (i < len1) {
for (; i < len1; ++i) {
SFltSclPoint *pt1 = taosArrayGet(pts1, i);
if (NULL == pt1) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_RANGE);
}
if (NULL == taosArrayPush(result, pt1)) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
@ -3838,6 +3861,9 @@ int32_t fltSclMergeSort(SArray *pts1, SArray *pts2, SArray *result) {
if (j < len2) {
for (; j < len2; ++j) {
SFltSclPoint *pt2 = taosArrayGet(pts2, j);
if (NULL == pt2) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_RANGE);
}
if (NULL == taosArrayPush(result, pt2)) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
@ -3859,6 +3885,9 @@ int32_t fltSclMerge(SArray *pts1, SArray *pts2, bool isUnion, SArray *merged) {
int32_t count = 0;
for (int32_t i = 0; i < taosArrayGetSize(all); ++i) {
SFltSclPoint *pt = taosArrayGet(all, i);
if (NULL == pt) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_RANGE);
}
if (pt->start) {
++count;
if (count == countRequired) {
@ -3893,6 +3922,9 @@ typedef struct {
int32_t fltSclGetOrCreateColumnRange(SColumnNode *colNode, SArray *colRangeList, SFltSclColumnRange **colRange) {
for (int32_t i = 0; i < taosArrayGetSize(colRangeList); ++i) {
*colRange = taosArrayGet(colRangeList, i);
if (NULL == colRange) {
return TSDB_CODE_OUT_OF_RANGE;
}
if (nodesEqualNode((SNode *)(*colRange)->colNode, (SNode *)colNode)) {
return TSDB_CODE_SUCCESS;
}
@ -4044,6 +4076,9 @@ int32_t filterRangeExecute(SFilterInfo *info, SColumnDataAgg *pDataStatis, int32
SArray *colRanges = info->sclCtx.fltSclRange;
for (int32_t i = 0; i < taosArrayGetSize(colRanges); ++i) {
SFltSclColumnRange *colRange = taosArrayGet(colRanges, i);
if (NULL == colRange) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_RANGE);
}
bool foundCol = false;
int32_t j = 0;
for (; j < numOfCols; ++j) {
@ -4332,11 +4367,17 @@ int32_t filterGetTimeRange(SNode *pNode, STimeWindow *win, bool *isStrict) {
SArray *colRanges = info->sclCtx.fltSclRange;
if (taosArrayGetSize(colRanges) == 1) {
SFltSclColumnRange *colRange = taosArrayGet(colRanges, 0);
if (NULL == colRange) {
FLT_ERR_JRET(TSDB_CODE_OUT_OF_RANGE);
}
SArray *points = colRange->points;
if (taosArrayGetSize(points) == 2) {
*win = TSWINDOW_DESC_INITIALIZER;
SFltSclPoint *startPt = taosArrayGet(points, 0);
SFltSclPoint *endPt = taosArrayGet(points, 1);
if (NULL == startPt || NULL == endPt) {
FLT_ERR_JRET(TSDB_CODE_OUT_OF_RANGE);
}
SFltSclDatum start;
SFltSclDatum end;
FLT_ERR_JRET(fltSclGetTimeStampDatum(startPt, &start));
@ -4398,7 +4439,7 @@ int32_t filterConverNcharColumns(SFilterInfo *info, int32_t rows, bool *gotNchar
bool ret = taosMbsToUcs4(varDataVal(src), varDataLen(src), (TdUcs4 *)varDataVal(dst), bufSize, &len);
if (!ret) {
qError("filterConverNcharColumns taosMbsToUcs4 error");
return TSDB_CODE_FAILED;
return TSDB_CODE_SCALAR_CONVERT_ERROR;
}
varDataLen(dst) = len;
}
@ -4432,7 +4473,7 @@ int32_t fltAddValueNodeToConverList(SFltTreeStat *stat, SValueNode *pNode) {
if (NULL == stat->nodeList) {
stat->nodeList = taosArrayInit(10, POINTER_BYTES);
if (NULL == stat->nodeList) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
FLT_ERR_RET(terrno);
}
}
@ -4816,6 +4857,9 @@ int32_t fltSclProcessCNF(SArray *sclOpListCNF, SArray *colRangeList) {
size_t sz = taosArrayGetSize(sclOpListCNF);
for (int32_t i = 0; i < sz; ++i) {
SFltSclOperator *sclOper = taosArrayGet(sclOpListCNF, i);
if (NULL == sclOper) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_RANGE);
}
SFltSclColumnRange *colRange = NULL;
FLT_ERR_RET(fltSclGetOrCreateColumnRange(sclOper->colNode, colRangeList, &colRange));
SArray *points = taosArrayInit(4, sizeof(SFltSclPoint));
@ -4920,22 +4964,27 @@ static int32_t fltSclCollectOperators(SNode *pNode, SArray *sclOpList) {
int32_t fltOptimizeNodes(SFilterInfo *pInfo, SNode **pNode, SFltTreeStat *pStat) {
SArray *sclOpList = taosArrayInit(16, sizeof(SFltSclOperator));
int32_t code = TSDB_CODE_SUCCESS;
if (NULL == sclOpList) {
FLT_ERR_RET(terrno);
}
FLT_ERR_RET(fltSclCollectOperators(*pNode, sclOpList));
FLT_ERR_JRET(fltSclCollectOperators(*pNode, sclOpList));
SArray *colRangeList = taosArrayInit(16, sizeof(SFltSclColumnRange));
if (NULL == colRangeList) {
FLT_ERR_RET(terrno);
}
FLT_ERR_RET(fltSclProcessCNF(sclOpList, colRangeList));
FLT_ERR_JRET(fltSclProcessCNF(sclOpList, colRangeList));
pInfo->sclCtx.fltSclRange = colRangeList;
for (int32_t i = 0; i < taosArrayGetSize(sclOpList); ++i) {
SFltSclOperator *sclOp = taosArrayGet(sclOpList, i);
if (NULL == sclOp) {
FLT_ERR_JRET(TSDB_CODE_OUT_OF_RANGE);
}
nodesDestroyNode((SNode *)sclOp->colNode);
nodesDestroyNode((SNode *)sclOp->valNode);
}
_return:
taosArrayDestroy(sclOpList);
return TSDB_CODE_SUCCESS;
}
@ -4946,6 +4995,9 @@ int32_t fltGetDataFromColId(void *param, int32_t id, void **data) {
for (int32_t j = 0; j < numOfCols; ++j) {
SColumnInfoData *pColInfo = taosArrayGet(pDataBlock, j);
if (NULL == pColInfo) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_RANGE);
}
if (id == pColInfo->info.colId) {
*data = pColInfo;
break;
@ -4965,6 +5017,9 @@ int32_t fltGetDataFromSlotId(void *param, int32_t id, void **data) {
}
SColumnInfoData *pColInfo = taosArrayGet(pDataBlock, id);
if (NULL == pColInfo) {
return TSDB_CODE_OUT_OF_RANGE;
}
*data = pColInfo;
return TSDB_CODE_SUCCESS;

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

@ -90,6 +90,9 @@ int32_t sclConvertValueToSclParam(SValueNode *pValueNode, SScalarParam *out, int
int32_t sclExtendResRows(SScalarParam *pDst, SScalarParam *pSrc, SArray *pBlockList) {
SSDataBlock *pb = taosArrayGetP(pBlockList, 0);
if (NULL == pb) {
SCL_ERR_RET(TSDB_CODE_OUT_OF_RANGE);
}
SScalarParam *pLeft = taosMemoryCalloc(1, sizeof(SScalarParam));
int32_t code = TSDB_CODE_SUCCESS;
if (NULL == pLeft) {
@ -116,7 +119,7 @@ int32_t scalarGenerateSetFromList(void **data, void *pNode, uint32_t type) {
SHashObj *pObj = taosHashInit(256, taosGetDefaultHashFunction(type), true, false);
if (NULL == pObj) {
sclError("taosHashInit failed, size:%d", 256);
SCL_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
SCL_ERR_RET(terrno);
}
taosHashSetEqualFp(pObj, taosGetDefaultEqualFunction(type));
@ -176,7 +179,7 @@ int32_t scalarGenerateSetFromList(void **data, void *pNode, uint32_t type) {
if (taosHashPut(pObj, buf, (size_t)len, NULL, 0)) {
sclError("taosHashPut to set failed");
SCL_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
SCL_ERR_JRET(terrno);
}
colInfoDataCleanup(out.columnData, out.numOfRows);
@ -336,6 +339,9 @@ int32_t sclInitParam(SNode *node, SScalarParam *param, SScalarCtx *ctx, int32_t
switch (nodeType(node)) {
case QUERY_NODE_LEFT_VALUE: {
SSDataBlock *pb = taosArrayGetP(ctx->pBlockList, 0);
if (NULL == pb) {
SCL_ERR_RET(TSDB_CODE_OUT_OF_RANGE);
}
param->numOfRows = pb->info.rows;
break;
}
@ -347,10 +353,7 @@ int32_t sclInitParam(SNode *node, SScalarParam *param, SScalarCtx *ctx, int32_t
SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
}
param->numOfRows = 1;
int32_t code = sclCreateColumnInfoData(&valueNode->node.resType, 1, param);
if (code != TSDB_CODE_SUCCESS) {
SCL_RET(TSDB_CODE_OUT_OF_MEMORY);
}
SCL_ERR_RET(sclCreateColumnInfoData(&valueNode->node.resType, 1, param));
if (TSDB_DATA_TYPE_NULL == valueNode->node.resType.type || valueNode->isNull) {
colDataSetNULL(param->columnData, 0);
} else {
@ -377,7 +380,7 @@ int32_t sclInitParam(SNode *node, SScalarParam *param, SScalarCtx *ctx, int32_t
taosHashCleanup(param->pHashFilter);
param->pHashFilter = NULL;
sclError("taosHashPut nodeList failed, size:%d", (int32_t)sizeof(*param));
return TSDB_CODE_OUT_OF_MEMORY;
return terrno;
}
param->colAlloced = false;
break;
@ -393,6 +396,9 @@ int32_t sclInitParam(SNode *node, SScalarParam *param, SScalarCtx *ctx, int32_t
int32_t index = -1;
for (int32_t i = 0; i < taosArrayGetSize(ctx->pBlockList); ++i) {
SSDataBlock *pb = taosArrayGetP(ctx->pBlockList, i);
if (NULL == pb) {
SCL_ERR_RET(TSDB_CODE_OUT_OF_RANGE);
}
if (pb->info.id.blockId == ref->dataBlockId) {
index = i;
break;
@ -461,6 +467,9 @@ int32_t sclInitParamList(SScalarParam **pParams, SNodeList *pParamList, SScalarC
if (NULL == pParamList) {
if (ctx->pBlockList) {
SSDataBlock *pBlock = taosArrayGetP(ctx->pBlockList, 0);
if (NULL == pBlock) {
SCL_ERR_RET(TSDB_CODE_OUT_OF_RANGE);
}
*rowNum = pBlock->info.rows;
} else {
*rowNum = 1;
@ -919,6 +928,9 @@ int32_t sclExecCaseWhen(SCaseWhenNode *node, SScalarCtx *ctx, SScalarParam *outp
if (ctx->pBlockList) {
SSDataBlock *pb = taosArrayGetP(ctx->pBlockList, 0);
if (NULL == pb) {
SCL_ERR_RET(TSDB_CODE_OUT_OF_RANGE);
}
rowNum = pb->info.rows;
output->numOfRows = pb->info.rows;
}
@ -1532,6 +1544,10 @@ EDealRes sclWalkTarget(SNode *pNode, SScalarCtx *ctx) {
int32_t index = -1;
for (int32_t i = 0; i < taosArrayGetSize(ctx->pBlockList); ++i) {
SSDataBlock *pb = taosArrayGetP(ctx->pBlockList, i);
if (NULL == pb) {
ctx->code = TSDB_CODE_OUT_OF_RANGE;
return DEAL_RES_ERROR;
}
if (pb->info.id.blockId == target->dataBlockId) {
index = i;
break;
@ -1636,7 +1652,7 @@ int32_t sclCalcConstants(SNode *pNode, bool dual, SNode **pRes) {
ctx.pRes = taosHashInit(SCL_DEFAULT_OP_NUM, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
if (NULL == ctx.pRes) {
sclError("taosHashInit failed, num:%d", SCL_DEFAULT_OP_NUM);
SCL_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
SCL_ERR_RET(terrno);
}
nodesRewriteExprPostOrder(&pNode, sclConstantsRewriter, (void *)&ctx);
@ -1770,7 +1786,7 @@ int32_t scalarCalculate(SNode *pNode, SArray *pBlockList, SScalarParam *pDst) {
ctx.pRes = taosHashInit(SCL_DEFAULT_OP_NUM, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
if (NULL == ctx.pRes) {
sclError("taosHashInit failed, num:%d", SCL_DEFAULT_OP_NUM);
SCL_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
SCL_ERR_RET(terrno);
}
nodesWalkExprPostOrder(pNode, sclCalcWalker, (void *)&ctx);
@ -1784,6 +1800,9 @@ int32_t scalarCalculate(SNode *pNode, SArray *pBlockList, SScalarParam *pDst) {
}
SSDataBlock *pb = taosArrayGetP(pBlockList, 0);
if (NULL == pb) {
SCL_ERR_JRET(TSDB_CODE_OUT_OF_RANGE);
}
if (1 == res->numOfRows && pb->info.rows > 0) {
SCL_ERR_JRET(sclExtendResRows(pDst, res, pBlockList));
} else {

32
source/libs/scalar/src/sclfunc.c
View File

@ -399,7 +399,7 @@ static int32_t concatCopyHelper(const char *input, char *output, bool hasNchar,
bool ret = taosMbsToUcs4(varDataVal(input), len, newBuf, (varDataLen(input) + 1) * TSDB_NCHAR_SIZE, &len);
if (!ret) {
taosMemoryFree(newBuf);
return TSDB_CODE_FAILED;
return TSDB_CODE_SCALAR_CONVERT_ERROR;
}
memcpy(varDataVal(output) + *dataLen, newBuf, varDataLen(input) * TSDB_NCHAR_SIZE);
*dataLen += varDataLen(input) * TSDB_NCHAR_SIZE;
@ -818,7 +818,7 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
} else if (inputType == TSDB_DATA_TYPE_NCHAR) {
int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), convBuf);
if (len < 0) {
code = TSDB_CODE_FAILED;
code = TSDB_CODE_SCALAR_CONVERT_ERROR;
goto _end;
}
@ -837,7 +837,7 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
} else if (inputType == TSDB_DATA_TYPE_NCHAR) {
int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), convBuf);
if (len < 0) {
code = TSDB_CODE_FAILED;
code = TSDB_CODE_SCALAR_CONVERT_ERROR;
goto _end;
}
convBuf[len] = 0;
@ -855,7 +855,7 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
} else if (inputType == TSDB_DATA_TYPE_NCHAR) {
int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), convBuf);
if (len < 0) {
code = TSDB_CODE_FAILED;
code = TSDB_CODE_SCALAR_CONVERT_ERROR;
goto _end;
}
@ -874,7 +874,7 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
} else if (inputType == TSDB_DATA_TYPE_NCHAR) {
int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), convBuf);
if (len < 0) {
code = TSDB_CODE_FAILED;
code = TSDB_CODE_SCALAR_CONVERT_ERROR;
goto _end;
}
convBuf[len] = 0;
@ -892,7 +892,7 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
} else if (inputType == TSDB_DATA_TYPE_NCHAR) {
int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), convBuf);
if (len < 0) {
code = TSDB_CODE_FAILED;
code = TSDB_CODE_SCALAR_CONVERT_ERROR;
goto _end;
}
convBuf[len] = 0;
@ -910,7 +910,7 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
} else if (inputType == TSDB_DATA_TYPE_NCHAR) {
int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), convBuf);
if (len < 0) {
code = TSDB_CODE_FAILED;
code = TSDB_CODE_SCALAR_CONVERT_ERROR;
goto _end;
}
convBuf[len] = 0;
@ -928,7 +928,7 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
} else if (inputType == TSDB_DATA_TYPE_NCHAR) {
int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), convBuf);
if (len < 0) {
code = TSDB_CODE_FAILED;
code = TSDB_CODE_SCALAR_CONVERT_ERROR;
goto _end;
}
convBuf[len] = 0;
@ -946,7 +946,7 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
} else if (inputType == TSDB_DATA_TYPE_NCHAR) {
int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), convBuf);
if (len < 0) {
code = TSDB_CODE_FAILED;
code = TSDB_CODE_SCALAR_CONVERT_ERROR;
goto _end;
}
@ -965,7 +965,7 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
} else if (inputType == TSDB_DATA_TYPE_NCHAR) {
int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), convBuf);
if (len < 0) {
code = TSDB_CODE_FAILED;
code = TSDB_CODE_SCALAR_CONVERT_ERROR;
goto _end;
}
convBuf[len] = 0;
@ -983,7 +983,7 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
} else if (inputType == TSDB_DATA_TYPE_NCHAR) {
int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), convBuf);
if (len < 0) {
code = TSDB_CODE_FAILED;
code = TSDB_CODE_SCALAR_CONVERT_ERROR;
goto _end;
}
convBuf[len] = 0;
@ -1001,7 +1001,7 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
} else if (inputType == TSDB_DATA_TYPE_NCHAR) {
int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), convBuf);
if (len < 0) {
code = TSDB_CODE_FAILED;
code = TSDB_CODE_SCALAR_CONVERT_ERROR;
goto _end;
}
convBuf[len] = 0;
@ -1041,7 +1041,7 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
} else if (inputType == TSDB_DATA_TYPE_NCHAR) {
int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), convBuf);
if (len < 0) {
code = TSDB_CODE_FAILED;
code = TSDB_CODE_SCALAR_CONVERT_ERROR;
goto _end;
}
len = TMIN(len, outputLen - VARSTR_HEADER_SIZE);
@ -1075,7 +1075,7 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
len = sprintf(tmp, "%.*s", outputCharLen, *(int8_t *)input ? "true" : "false");
bool ret = taosMbsToUcs4(tmp, len, (TdUcs4 *)varDataVal(output), outputLen - VARSTR_HEADER_SIZE, &len);
if (!ret) {
code = TSDB_CODE_FAILED;
code = TSDB_CODE_SCALAR_CONVERT_ERROR;
goto _end;
}
@ -1085,7 +1085,7 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
bool ret = taosMbsToUcs4(input + VARSTR_HEADER_SIZE, len, (TdUcs4 *)varDataVal(output),
outputLen - VARSTR_HEADER_SIZE, &len);
if (!ret) {
code = TSDB_CODE_FAILED;
code = TSDB_CODE_SCALAR_CONVERT_ERROR;
goto _end;
}
varDataSetLen(output, len);
@ -1099,7 +1099,7 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
len = outputCharLen > len ? len : outputCharLen;
bool ret = taosMbsToUcs4(buf, len, (TdUcs4 *)varDataVal(output), outputLen - VARSTR_HEADER_SIZE, &len);
if (!ret) {
code = TSDB_CODE_FAILED;
code = TSDB_CODE_SCALAR_CONVERT_ERROR;
goto _end;
}
varDataSetLen(output, len);

4
source/libs/scalar/src/sclvector.c
View File

@ -105,7 +105,7 @@ int32_t convertNcharToDouble(const void *inData, void *outData) {
int len = taosUcs4ToMbs((TdUcs4 *)varDataVal(inData), varDataLen(inData), tmp);
if (len < 0) {
sclError("castConvert taosUcs4ToMbs error 1");
SCL_ERR_JRET(TSDB_CODE_FAILED);
SCL_ERR_JRET(TSDB_CODE_SCALAR_CONVERT_ERROR);
}
tmp[len] = 0;
@ -596,7 +596,7 @@ int32_t ncharTobinary(void *buf, void **out) { // todo need to remove , if tobi
sclError("charset:%s to %s. val:%s convert ncharTobinary failed.", DEFAULT_UNICODE_ENCODEC, tsCharset,
(char *)varDataVal(buf));
taosMemoryFree(*out);
SCL_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
SCL_ERR_RET(TSDB_CODE_SCALAR_CONVERT_ERROR);
}
varDataSetLen(*out, len);
SCL_RET(TSDB_CODE_SUCCESS);