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

enh: add query response message processing

This commit is contained in:
dapan1121 2022-11-21 15:02:05 +08:00
parent 70eb1f445e
commit cba374ae6b
1 changed files with 9 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
source/libs/scheduler/src/schRemote.c
View File

@ -334,17 +334,17 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t execId, SDa
SCH_ERR_JRET(TSDB_CODE_QRY_INVALID_INPUT); SCH_ERR_JRET(TSDB_CODE_QRY_INVALID_INPUT);
} }
SQueryTableRsp *rsp = (SQueryTableRsp *)msg; SQueryTableRsp rsp = {0};
rsp->code = ntohl(rsp->code); if (tDeserializeSQueryTableRsp(msg, msgSize, &rsp) < 0) {
rsp->sversion = ntohl(rsp->sversion); SCH_TASK_ELOG("tDeserializeSQueryTableRsp failed, msgSize:%d", msgSize);
rsp->tversion = ntohl(rsp->tversion); SCH_ERR_JRET(TSDB_CODE_QRY_INVALID_MSG);
rsp->affectedRows = be64toh(rsp->affectedRows); }
SCH_ERR_JRET(rsp->code); SCH_ERR_JRET(rsp.code);
SCH_ERR_JRET(schSaveJobExecRes(pJob, rsp)); SCH_ERR_JRET(schSaveJobExecRes(pJob, &rsp));
atomic_add_fetch_32(&pJob->resNumOfRows, rsp->affectedRows); atomic_add_fetch_32(&pJob->resNumOfRows, rsp.affectedRows);
taosMemoryFreeClear(msg); taosMemoryFreeClear(msg);