taosdata / TDengine / #3720

coverage: 30.242% (-31.7%) from 61.936%
#3720

push

travis-ci

web-flow 
feat(taosBenchmark): supports decimal data type (#30456)

* feat: taosBenchmark supports decimal data type

* build: decimal script not use pytest.sh

* fix: fix typo for decimal script

* test: insertBasic.py debug

71234 of 313946 branches covered (22.69%)

Branch coverage included in aggregate %.

38 of 423 new or added lines in 8 files covered. (8.98%)

120240 existing lines in 447 files now uncovered.

118188 of 312400 relevant lines covered (37.83%)

1450220.33 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

3.92
/source/common/src/msg/streamMsg.c
1 
/*
2 
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3 
 *
4 
 * This program is free software: you can use, redistribute, and/or modify
5 
 * it under the terms of the GNU Affero General Public License, version 3
6 
 * or later ("AGPL"), as published by the Free Software Foundation.
7 
 *
8 
 * This program is distributed in the hope that it will be useful, but WITHOUT
9 
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 
 * FITNESS FOR A PARTICULAR PURPOSE.
11 
 *
12 
 * You should have received a copy of the GNU Affero General Public License
13 
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
14 
 */
15 









16



#include "streamMsg.h"










17



#include "os.h"










18



#include "tcommon.h"










19












20



typedef struct STaskId {










21



  int64_t streamId;










22



  int64_t taskId;










23



} STaskId;










24












25



typedef struct STaskCkptInfo {










26



  int64_t latestId;          // saved checkpoint id










27



  int64_t latestVer;         // saved checkpoint ver










28



  int64_t latestTime;        // latest checkpoint time










29



  int64_t latestSize;        // latest checkpoint size










30



  int8_t  remoteBackup;      // latest checkpoint backup done










31



  int64_t activeId;          // current active checkpoint id










32



  int32_t activeTransId;     // checkpoint trans id










33



  int8_t  failed;            // denote if the checkpoint is failed or not










34



  int8_t  consensusChkptId;  // required the consensus-checkpointId










35



  int64_t consensusTs;       //










36



} STaskCkptInfo;










37












38



typedef struct STaskStatusEntry {










39



  STaskId       id;










40



  int32_t       status;










41



  int32_t       statusLastDuration;  // to record the last duration of current status










42



  int64_t       stage;










43



  int32_t       nodeId;










44



  SVersionRange verRange;      // start/end version in WAL, only valid for source task










45



  int64_t       processedVer;  // only valid for source task










46



  double        inputQUsed;    // in MiB










47



  double        inputRate;










48



  double        procsThroughput;   // duration between one element put into input queue and being processed.










49



  double        procsTotal;        // duration between one element put into input queue and being processed.










50



  double        outputThroughput;  // the size of dispatched result blocks in bytes










51



  double        outputTotal;       // the size of dispatched result blocks in bytes










52



  double        sinkQuota;         // existed quota size for sink task










53



  double        sinkDataSize;      // sink to dst data size










54



  int64_t       startTime;










55



  int64_t       startCheckpointId;










56



  int64_t       startCheckpointVer;










57



  int64_t       hTaskId;










58



  STaskCkptInfo checkpointInfo;










59



  STaskNotifyEventStat notifyEventStat;










60



} STaskStatusEntry;










61













UNCOV

62



int32_t tEncodeStreamEpInfo(SEncoder* pEncoder, const SStreamUpstreamEpInfo* pInfo) {




×








UNCOV

63



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pInfo->taskId));





×








UNCOV

64



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pInfo->nodeId));





×








UNCOV

65



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pInfo->childId));





×








UNCOV

66



  TAOS_CHECK_RETURN(tEncodeSEpSet(pEncoder, &pInfo->epSet));





×








UNCOV

67



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pInfo->stage));





×








UNCOV

68



  return 0;




×







69



}










70













UNCOV

71



int32_t tDecodeStreamEpInfo(SDecoder* pDecoder, SStreamUpstreamEpInfo* pInfo) {




×








UNCOV

72



  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pInfo->taskId));





×








UNCOV

73



  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pInfo->nodeId));





×








UNCOV

74



  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pInfo->childId));





×








UNCOV

75



  TAOS_CHECK_RETURN(tDecodeSEpSet(pDecoder, &pInfo->epSet));





×








UNCOV

76



  TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pInfo->stage));





×








UNCOV

77



  return 0;




×







78



}










79













UNCOV

80



int32_t tEncodeStreamCheckpointSourceReq(SEncoder* pEncoder, const SStreamCheckpointSourceReq* pReq) {




×








UNCOV

81



  TAOS_CHECK_RETURN(tStartEncode(pEncoder));





×








UNCOV

82



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

83



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pReq->checkpointId));





×








UNCOV

84



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pReq->taskId));





×








UNCOV

85



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pReq->nodeId));





×








UNCOV

86



  TAOS_CHECK_RETURN(tEncodeSEpSet(pEncoder, &pReq->mgmtEps));





×








UNCOV

87



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pReq->mnodeId));





×








UNCOV

88



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pReq->expireTime));





×








UNCOV

89



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pReq->transId));





×








UNCOV

90



  TAOS_CHECK_RETURN(tEncodeI8(pEncoder, pReq->mndTrigger));





×








UNCOV

91



  tEndEncode(pEncoder);




×








UNCOV

92



  return pEncoder->pos;




×







93



}










94













UNCOV

95



int32_t tDecodeStreamCheckpointSourceReq(SDecoder* pDecoder, SStreamCheckpointSourceReq* pReq) {




×








UNCOV

96



  TAOS_CHECK_RETURN(tStartDecode(pDecoder));





×








UNCOV

97



  TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pReq->streamId));





×








UNCOV

98



  TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pReq->checkpointId));





×








UNCOV

99



  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pReq->taskId));





×








UNCOV

100



  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pReq->nodeId));





×








UNCOV

101



  TAOS_CHECK_RETURN(tDecodeSEpSet(pDecoder, &pReq->mgmtEps));





×








UNCOV

102



  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pReq->mnodeId));





×








UNCOV

103



  TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pReq->expireTime));





×








UNCOV

104



  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pReq->transId));





×








UNCOV

105



  TAOS_CHECK_RETURN(tDecodeI8(pDecoder, &pReq->mndTrigger));





×








UNCOV

106



  tEndDecode(pDecoder);




×








UNCOV

107



  return 0;




×







108



}










109













UNCOV

110



int32_t tEncodeStreamCheckpointSourceRsp(SEncoder* pEncoder, const SStreamCheckpointSourceRsp* pRsp) {




×








UNCOV

111



  TAOS_CHECK_RETURN(tStartEncode(pEncoder));





×








UNCOV

112



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pRsp->streamId));





×








UNCOV

113



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pRsp->checkpointId));





×








UNCOV

114



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pRsp->taskId));





×








UNCOV

115



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pRsp->nodeId));





×








UNCOV

116



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pRsp->expireTime));





×








UNCOV

117



  TAOS_CHECK_RETURN(tEncodeI8(pEncoder, pRsp->success));





×








UNCOV

118



  tEndEncode(pEncoder);




×








UNCOV

119



  return pEncoder->pos;




×







120



}










121













UNCOV

122



int32_t tEncodeStreamTaskUpdateMsg(SEncoder* pEncoder, const SStreamTaskNodeUpdateMsg* pMsg) {




×








UNCOV

123



  int32_t code = 0;




×







124



  int32_t lino;










125













UNCOV

126



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

127



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pMsg->streamId));





×








UNCOV

128



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pMsg->taskId));





×







129













UNCOV

130



  int32_t size = taosArrayGetSize(pMsg->pNodeList);





×








UNCOV

131



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, size));





×







132













UNCOV

133



  for (int32_t i = 0; i < size; ++i) {





×








UNCOV

134



    SNodeUpdateInfo* pInfo = taosArrayGet(pMsg->pNodeList, i);




×








UNCOV

135



    if (pInfo == NULL) {





×







136



      TAOS_CHECK_EXIT(terrno);





×







137



    }










138













UNCOV

139



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pInfo->nodeId));





×








UNCOV

140



    TAOS_CHECK_EXIT(tEncodeSEpSet(pEncoder, &pInfo->prevEp));





×








UNCOV

141



    TAOS_CHECK_EXIT(tEncodeSEpSet(pEncoder, &pInfo->newEp));





×







142



  }










143












144



  // todo this new attribute will be result in being incompatible with previous version











UNCOV

145



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pMsg->transId));





×







146













UNCOV

147



  int32_t numOfTasks = taosArrayGetSize(pMsg->pTaskList);





×








UNCOV

148



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, numOfTasks));





×







149













UNCOV

150



  for (int32_t i = 0; i < numOfTasks; ++i) {





×








UNCOV

151



    int32_t* pId = taosArrayGet(pMsg->pTaskList, i);




×








UNCOV

152



    if (pId == NULL) {





×








UNCOV

153



      TAOS_CHECK_EXIT(terrno);





×







154



    }











UNCOV

155



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *(int32_t*)pId));





×







156



  }










157













UNCOV

158



  tEndEncode(pEncoder);




×








UNCOV

159



_exit:




×








UNCOV

160



  if (code) {





×








UNCOV

161



    return code;




×







162



  } else {











UNCOV

163



    return pEncoder->pos;




×







164



  }










165



}










166












167



int32_t tDecodeStreamTaskUpdateMsg(SDecoder* pDecoder, SStreamTaskNodeUpdateMsg* pMsg) {




×








UNCOV

168



  int32_t code = 0;




×







169



  int32_t lino;










170













UNCOV

171



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

172



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pMsg->streamId));





×








UNCOV

173



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pMsg->taskId));





×







174













UNCOV

175



  int32_t size = 0;




×







176



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &size));





×







177













UNCOV

178



  pMsg->pNodeList = taosArrayInit(size, sizeof(SNodeUpdateInfo));




×








UNCOV

179



  TSDB_CHECK_NULL(pMsg->pNodeList, code, lino, _exit, terrno);





×







180













UNCOV

181



  for (int32_t i = 0; i < size; ++i) {





×








UNCOV

182



    SNodeUpdateInfo info = {0};




×








UNCOV

183



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &info.nodeId));





×








UNCOV

184



    TAOS_CHECK_EXIT(tDecodeSEpSet(pDecoder, &info.prevEp));





×








UNCOV

185



    TAOS_CHECK_EXIT(tDecodeSEpSet(pDecoder, &info.newEp));





×







186













UNCOV

187



    if (taosArrayPush(pMsg->pNodeList, &info) == NULL) {





×








UNCOV

188



      TAOS_CHECK_EXIT(terrno);





×







189



    }










190



  }










191













UNCOV

192



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pMsg->transId));





×







193












194



  // number of tasks











UNCOV

195



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &size));





×








UNCOV

196



  pMsg->pTaskList = taosArrayInit(size, sizeof(int32_t));




×








UNCOV

197



  if (pMsg->pTaskList == NULL) {





×








UNCOV

198



    TAOS_CHECK_EXIT(terrno);





×







199



  }










200













UNCOV

201



  for (int32_t i = 0; i < size; ++i) {





×








UNCOV

202



    int32_t id = 0;




×








UNCOV

203



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &id));





×







204



    if (taosArrayPush(pMsg->pTaskList, &id) == NULL) {





×








UNCOV

205



      TAOS_CHECK_EXIT(terrno);





×







206



    }










207



  }










208













UNCOV

209



  tEndDecode(pDecoder);




×








UNCOV

210



_exit:




×








UNCOV

211



  return code;




×







212



}










213













UNCOV

214



void tDestroyNodeUpdateMsg(SStreamTaskNodeUpdateMsg* pMsg) {




×








UNCOV

215



  taosArrayDestroy(pMsg->pNodeList);




×








UNCOV

216



  taosArrayDestroy(pMsg->pTaskList);




×








UNCOV

217



  pMsg->pNodeList = NULL;




×








UNCOV

218



  pMsg->pTaskList = NULL;




×








UNCOV

219



}




×







220













UNCOV

221



int32_t tEncodeStreamTaskCheckReq(SEncoder* pEncoder, const SStreamTaskCheckReq* pReq) {




×








UNCOV

222



  int32_t code = 0;




×







223



  int32_t lino;










224













UNCOV

225



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

226



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->reqId));





×








UNCOV

227



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

228



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamNodeId));





×








UNCOV

229



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamTaskId));





×








UNCOV

230



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->downstreamNodeId));





×








UNCOV

231



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->downstreamTaskId));





×








UNCOV

232



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->childId));





×








UNCOV

233



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->stage));





×








UNCOV

234



  tEndEncode(pEncoder);




×







235













UNCOV

236



_exit:




×








UNCOV

237



  if (code) {





×








UNCOV

238



    return code;




×







239



  } else {











UNCOV

240



    return pEncoder->pos;




×







241



  }










242



}










243













UNCOV

244



int32_t tDecodeStreamTaskCheckReq(SDecoder* pDecoder, SStreamTaskCheckReq* pReq) {




×








UNCOV

245



  int32_t code = 0;




×







246



  int32_t lino;










247













UNCOV

248



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

249



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->reqId));





×








UNCOV

250



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





×








UNCOV

251



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->upstreamNodeId));





×








UNCOV

252



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->upstreamTaskId));





×








UNCOV

253



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->downstreamNodeId));





×








UNCOV

254



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->downstreamTaskId));





×








UNCOV

255



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->childId));





×








UNCOV

256



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->stage));





×








UNCOV

257



  tEndDecode(pDecoder);




×







258













UNCOV

259



_exit:




×








UNCOV

260



  return code;




×







261



}










262













UNCOV

263



int32_t tEncodeStreamTaskCheckRsp(SEncoder* pEncoder, const SStreamTaskCheckRsp* pRsp) {




×








UNCOV

264



  int32_t code = 0;




×







265



  int32_t lino;










266













UNCOV

267



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

268



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pRsp->reqId));





×








UNCOV

269



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pRsp->streamId));





×








UNCOV

270



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->upstreamNodeId));





×








UNCOV

271



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->upstreamTaskId));





×








UNCOV

272



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->downstreamNodeId));





×








UNCOV

273



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->downstreamTaskId));





×








UNCOV

274



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->childId));





×








UNCOV

275



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pRsp->oldStage));





×








UNCOV

276



  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pRsp->status));





×








UNCOV

277



  tEndEncode(pEncoder);




×







278













UNCOV

279



_exit:




×








UNCOV

280



  if (code) {





×








UNCOV

281



    return code;




×







282



  } else {











UNCOV

283



    return pEncoder->pos;




×







284



  }










285



}










286













UNCOV

287



int32_t tDecodeStreamTaskCheckRsp(SDecoder* pDecoder, SStreamTaskCheckRsp* pRsp) {




×








UNCOV

288



  int32_t code = 0;




×







289



  int32_t lino;










290













UNCOV

291



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

292



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pRsp->reqId));





×








UNCOV

293



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pRsp->streamId));





×








UNCOV

294



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->upstreamNodeId));





×








UNCOV

295



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->upstreamTaskId));





×








UNCOV

296



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->downstreamNodeId));





×








UNCOV

297



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->downstreamTaskId));





×








UNCOV

298



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->childId));





×








UNCOV

299



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pRsp->oldStage));





×








UNCOV

300



  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pRsp->status));





×








UNCOV

301



  tEndDecode(pDecoder);




×







302













UNCOV

303



_exit:




×








UNCOV

304



  return code;




×







305



}










306













UNCOV

307



int32_t tEncodeStreamCheckpointReadyMsg(SEncoder* pEncoder, const SStreamCheckpointReadyMsg* pReq) {




×








UNCOV

308



  int32_t code = 0;




×







309



  int32_t lino;










310













UNCOV

311



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

312



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

313



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->checkpointId));





×








UNCOV

314



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->downstreamTaskId));





×








UNCOV

315



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->downstreamNodeId));





×








UNCOV

316



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamTaskId));





×








UNCOV

317



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamNodeId));





×








UNCOV

318



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->childId));





×








UNCOV

319



  tEndEncode(pEncoder);




×







320













UNCOV

321



_exit:




×








UNCOV

322



  if (code) {





×








UNCOV

323



    return code;




×







324



  } else {











UNCOV

325



    return pEncoder->pos;




×







326



  }










327



}










328













UNCOV

329



int32_t tDecodeStreamCheckpointReadyMsg(SDecoder* pDecoder, SStreamCheckpointReadyMsg* pRsp) {




×








UNCOV

330



  int32_t code = 0;




×







331



  int32_t lino;










332












333



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

334



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pRsp->streamId));





×








UNCOV

335



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pRsp->checkpointId));





×








UNCOV

336



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->downstreamTaskId));





×








UNCOV

337



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->downstreamNodeId));





×








UNCOV

338



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->upstreamTaskId));





×








UNCOV

339



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->upstreamNodeId));





×








UNCOV

340



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->childId));





×







341



  tEndDecode(pDecoder);




×







342













UNCOV

343



_exit:




×








UNCOV

344



  return code;




×







345



}










346













UNCOV

347



int32_t tEncodeStreamDispatchReq(SEncoder* pEncoder, const SStreamDispatchReq* pReq) {




×








UNCOV

348



  int32_t code = 0;




×







349



  int32_t lino;










350













UNCOV

351



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

352



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->stage));





×








UNCOV

353



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->msgId));





×








UNCOV

354



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->srcVgId));





×








UNCOV

355



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->type));





×








UNCOV

356



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

357



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->taskId));





×








UNCOV

358



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->type));





×








UNCOV

359



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamTaskId));





×








UNCOV

360



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamChildId));





×








UNCOV

361



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamNodeId));





×








UNCOV

362



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamRelTaskId));





×








UNCOV

363



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->blockNum));





×








UNCOV

364



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->totalLen));





×







365













UNCOV

366



  if (taosArrayGetSize(pReq->data) != pReq->blockNum || taosArrayGetSize(pReq->dataLen) != pReq->blockNum) {





×








UNCOV

367



    uError("invalid dispatch req msg");





×








UNCOV

368



    TAOS_CHECK_EXIT(TSDB_CODE_INVALID_MSG);





×







369



  }










370













UNCOV

371



  for (int32_t i = 0; i < pReq->blockNum; i++) {





×








UNCOV

372



    int32_t* pLen = taosArrayGet(pReq->dataLen, i);




×








UNCOV

373



    void*    data = taosArrayGetP(pReq->data, i);




×








UNCOV

374



    if (data == NULL || pLen == NULL) {





×








UNCOV

375



      TAOS_CHECK_EXIT(terrno);





×







376



    }










377













UNCOV

378



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *pLen));





×








UNCOV

379



    TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, data, *pLen));





×







380



  }











UNCOV

381



  tEndEncode(pEncoder);




×








UNCOV

382



_exit:




×








UNCOV

383



  if (code) {





×








UNCOV

384



    return code;




×







385



  } else {











UNCOV

386



    return pEncoder->pos;




×







387



  }










388



}










389













UNCOV

390



int32_t tDecodeStreamDispatchReq(SDecoder* pDecoder, SStreamDispatchReq* pReq) {




×








UNCOV

391



  int32_t code = 0;




×







392



  int32_t lino;










393













UNCOV

394



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

395



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->stage));





×








UNCOV

396



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->msgId));





×







397



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->srcVgId));





×








UNCOV

398



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->type));





×








UNCOV

399



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





×








UNCOV

400



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->taskId));





×








UNCOV

401



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->type));





×








UNCOV

402



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->upstreamTaskId));





×








UNCOV

403



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->upstreamChildId));





×








UNCOV

404



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->upstreamNodeId));





×








UNCOV

405



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->upstreamRelTaskId));





×








UNCOV

406



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->blockNum));





×








UNCOV

407



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->totalLen));





×







408













UNCOV

409



  if ((pReq->data = taosArrayInit(pReq->blockNum, sizeof(void*))) == NULL) {





×








UNCOV

410



    TAOS_CHECK_EXIT(terrno);





×







411



  }











UNCOV

412



  if ((pReq->dataLen = taosArrayInit(pReq->blockNum, sizeof(int32_t))) == NULL) {





×








UNCOV

413



    TAOS_CHECK_EXIT(terrno);





×







414



  }











UNCOV

415



  for (int32_t i = 0; i < pReq->blockNum; i++) {





×







416



    int32_t  len1;










417



    uint64_t len2;










418



    void*    data;











UNCOV

419



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &len1));





×








UNCOV

420



    TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, &data, &len2));





×







421













UNCOV

422



    if (len1 != len2) {





×








UNCOV

423



      TAOS_CHECK_EXIT(TSDB_CODE_INVALID_MSG);





×







424



    }










425













UNCOV

426



    if (taosArrayPush(pReq->dataLen, &len1) == NULL) {





×







427



      TAOS_CHECK_EXIT(terrno);





×







428



    }










429













UNCOV

430



    if (taosArrayPush(pReq->data, &data) == NULL) {





×








UNCOV

431



      TAOS_CHECK_EXIT(terrno);





×







432



    }










433



  }










434













UNCOV

435



  tEndDecode(pDecoder);




×








UNCOV

436



_exit:




×








UNCOV

437



  return code;




×







438



}










439













UNCOV

440



void tCleanupStreamDispatchReq(SStreamDispatchReq* pReq) {




×








UNCOV

441



  taosArrayDestroyP(pReq->data, NULL);




×








UNCOV

442



  taosArrayDestroy(pReq->dataLen);




×








UNCOV

443



}




×







444













UNCOV

445



int32_t tEncodeStreamRetrieveReq(SEncoder* pEncoder, const SStreamRetrieveReq* pReq) {




×








UNCOV

446



  int32_t code = 0;




×







447



  int32_t lino;










448













UNCOV

449



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

450



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

451



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->reqId));





×








UNCOV

452



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->dstNodeId));





×








UNCOV

453



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->dstTaskId));





×








UNCOV

454



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->srcNodeId));





×








UNCOV

455



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->srcTaskId));





×








UNCOV

456



  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, (const uint8_t*)pReq->pRetrieve, pReq->retrieveLen));





×








UNCOV

457



  tEndEncode(pEncoder);




×







458













UNCOV

459



_exit:




×








UNCOV

460



  if (code) {





×








UNCOV

461



    return code;




×







462



  } else {











UNCOV

463



    return pEncoder->pos;




×







464



  }










465



}










466













UNCOV

467



int32_t tDecodeStreamRetrieveReq(SDecoder* pDecoder, SStreamRetrieveReq* pReq) {




×








UNCOV

468



  int32_t code = 0;




×







469



  int32_t lino;










470













UNCOV

471



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

472



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





×








UNCOV

473



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->reqId));





×








UNCOV

474



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->dstNodeId));





×








UNCOV

475



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->dstTaskId));





×








UNCOV

476



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->srcNodeId));





×








UNCOV

477



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->srcTaskId));





×








UNCOV

478



  uint64_t len = 0;




×








UNCOV

479



  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pReq->pRetrieve, &len));





×








UNCOV

480



  pReq->retrieveLen = (int32_t)len;




×








UNCOV

481



  tEndDecode(pDecoder);




×







482













UNCOV

483



_exit:




×








UNCOV

484



  return code;




×







485



}










486













UNCOV

487



void tCleanupStreamRetrieveReq(SStreamRetrieveReq* pReq) { taosMemoryFree(pReq->pRetrieve); }





×







488













UNCOV

489



int32_t tEncodeStreamTaskCheckpointReq(SEncoder* pEncoder, const SStreamTaskCheckpointReq* pReq) {




×








UNCOV

490



  int32_t code = 0;




×







491



  int32_t lino;










492













UNCOV

493



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







494



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

495



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->taskId));





×








UNCOV

496



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->nodeId));





×








UNCOV

497



  tEndEncode(pEncoder);




×







498













UNCOV

499



_exit:




×








UNCOV

500



  return code;




×







501



}










502













UNCOV

503



int32_t tDecodeStreamTaskCheckpointReq(SDecoder* pDecoder, SStreamTaskCheckpointReq* pReq) {




×








UNCOV

504



  int32_t code = 0;




×







505



  int32_t lino;










506













UNCOV

507



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

508



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





×








UNCOV

509



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->taskId));





×








UNCOV

510



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->nodeId));





×








UNCOV

511



  tEndDecode(pDecoder);




×







512













UNCOV

513



_exit:




×








UNCOV

514



  return code;




×







515



}










516













UNCOV

517



int32_t tEncodeStreamHbMsg(SEncoder* pEncoder, const SStreamHbMsg* pReq) {




×








UNCOV

518



  int32_t code = 0;




×







519



  int32_t lino;










520













UNCOV

521



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

522



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->vgId));





×








UNCOV

523



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->numOfTasks));





×







524













UNCOV

525



  for (int32_t i = 0; i < pReq->numOfTasks; ++i) {





×








UNCOV

526



    STaskStatusEntry* ps = taosArrayGet(pReq->pTaskStatus, i);




×








UNCOV

527



    if (ps == NULL) {





×








UNCOV

528



      TAOS_CHECK_EXIT(terrno);





×







529



    }










530













UNCOV

531



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->id.streamId));





×








UNCOV

532



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, ps->id.taskId));





×








UNCOV

533



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, ps->status));





×








UNCOV

534



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->stage));





×








UNCOV

535



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, ps->nodeId));





×








UNCOV

536



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->inputQUsed));





×








UNCOV

537



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->inputRate));





×








UNCOV

538



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->procsTotal));





×








UNCOV

539



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->procsThroughput));





×








UNCOV

540



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->outputTotal));





×








UNCOV

541



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->outputThroughput));





×








UNCOV

542



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->sinkQuota));





×








UNCOV

543



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->sinkDataSize));





×








UNCOV

544



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->processedVer));





×








UNCOV

545



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->verRange.minVer));





×








UNCOV

546



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->verRange.maxVer));





×








UNCOV

547



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->checkpointInfo.activeId));





×







548



    TAOS_CHECK_EXIT(tEncodeI8(pEncoder, ps->checkpointInfo.failed));





×








UNCOV

549



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, ps->checkpointInfo.activeTransId));





×








UNCOV

550



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->checkpointInfo.latestId));





×








UNCOV

551



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->checkpointInfo.latestVer));





×








UNCOV

552



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->checkpointInfo.latestTime));





×








UNCOV

553



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->checkpointInfo.latestSize));





×








UNCOV

554



    TAOS_CHECK_EXIT(tEncodeI8(pEncoder, ps->checkpointInfo.remoteBackup));





×








UNCOV

555



    TAOS_CHECK_EXIT(tEncodeI8(pEncoder, ps->checkpointInfo.consensusChkptId));





×








UNCOV

556



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->checkpointInfo.consensusTs));





×








UNCOV

557



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->startTime));





×








UNCOV

558



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->startCheckpointId));





×








UNCOV

559



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->startCheckpointVer));





×







560



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->hTaskId));





×








UNCOV

561



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventAddTimes));





×








UNCOV

562



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventAddElems));





×








UNCOV

563



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->notifyEventStat.notifyEventAddCostSec));





×








UNCOV

564



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventPushTimes));





×








UNCOV

565



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventPushElems));





×








UNCOV

566



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->notifyEventStat.notifyEventPushCostSec));





×








UNCOV

567



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventPackTimes));





×








UNCOV

568



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventPackElems));





×








UNCOV

569



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->notifyEventStat.notifyEventPackCostSec));





×








UNCOV

570



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventSendTimes));





×








UNCOV

571



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventSendElems));





×








UNCOV

572



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->notifyEventStat.notifyEventSendCostSec));





×








UNCOV

573



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventHoldElems));





×







574



  }










575













UNCOV

576



  int32_t numOfVgs = taosArrayGetSize(pReq->pUpdateNodes);





×








UNCOV

577



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, numOfVgs));





×







578













UNCOV

579



  for (int j = 0; j < numOfVgs; ++j) {





×








UNCOV

580



    int32_t* pVgId = taosArrayGet(pReq->pUpdateNodes, j);




×








UNCOV

581



    if (pVgId == NULL) {





×








UNCOV

582



      TAOS_CHECK_EXIT(terrno);





×







583



    }










584













UNCOV

585



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *pVgId));





×







586



  }










587













UNCOV

588



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->msgId));





×








UNCOV

589



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->ts));





×








UNCOV

590



  tEndEncode(pEncoder);




×







591













UNCOV

592



_exit:




×








UNCOV

593



  if (code) {





×








UNCOV

594



    return code;




×







595



  } else {











UNCOV

596



    return pEncoder->pos;




×







597



  }










598



}










599













UNCOV

600



int32_t tDecodeStreamHbMsg(SDecoder* pDecoder, SStreamHbMsg* pReq) {




×








UNCOV

601



  int32_t code = 0;




×







602



  int32_t lino;










603













UNCOV

604



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

605



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->vgId));





×








UNCOV

606



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->numOfTasks));





×







607













UNCOV

608



  if ((pReq->pTaskStatus = taosArrayInit(pReq->numOfTasks, sizeof(STaskStatusEntry))) == NULL) {





×








UNCOV

609



    TAOS_CHECK_EXIT(terrno);





×







610



  }











UNCOV

611



  for (int32_t i = 0; i < pReq->numOfTasks; ++i) {





×








UNCOV

612



    int32_t          taskId = 0;




×








UNCOV

613



    STaskStatusEntry entry = {0};




×







614













UNCOV

615



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.id.streamId));





×








UNCOV

616



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &taskId));





×








UNCOV

617



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &entry.status));





×








UNCOV

618



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.stage));





×








UNCOV

619



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &entry.nodeId));





×








UNCOV

620



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.inputQUsed));





×








UNCOV

621



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.inputRate));





×








UNCOV

622



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.procsTotal));





×








UNCOV

623



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.procsThroughput));





×








UNCOV

624



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.outputTotal));





×








UNCOV

625



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.outputThroughput));





×








UNCOV

626



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.sinkQuota));





×








UNCOV

627



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.sinkDataSize));





×








UNCOV

628



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.processedVer));





×







629



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.verRange.minVer));





×








UNCOV

630



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.verRange.maxVer));





×








UNCOV

631



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.checkpointInfo.activeId));





×








UNCOV

632



    TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &entry.checkpointInfo.failed));





×








UNCOV

633



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &entry.checkpointInfo.activeTransId));





×







634













UNCOV

635



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.checkpointInfo.latestId));





×








UNCOV

636



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.checkpointInfo.latestVer));





×







637



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.checkpointInfo.latestTime));





×








UNCOV

638



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.checkpointInfo.latestSize));





×








UNCOV

639



    TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &entry.checkpointInfo.remoteBackup));





×








UNCOV

640



    TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &entry.checkpointInfo.consensusChkptId));





×








UNCOV

641



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.checkpointInfo.consensusTs));





×








UNCOV

642



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.startTime));





×








UNCOV

643



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.startCheckpointId));





×







644



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.startCheckpointVer));





×








UNCOV

645



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.hTaskId));





×







646













UNCOV

647



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventAddTimes));





×








UNCOV

648



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventAddElems));





×








UNCOV

649



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.notifyEventStat.notifyEventAddCostSec));





×








UNCOV

650



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventPushTimes));





×








UNCOV

651



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventPushElems));





×








UNCOV

652



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.notifyEventStat.notifyEventPushCostSec));





×








UNCOV

653



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventPackTimes));





×








UNCOV

654



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventPackElems));





×








UNCOV

655



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.notifyEventStat.notifyEventPackCostSec));





×








UNCOV

656



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventSendTimes));





×








UNCOV

657



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventSendElems));





×







658



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.notifyEventStat.notifyEventSendCostSec));





×








UNCOV

659



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventHoldElems));





×







660













UNCOV

661



    entry.id.taskId = taskId;




×








UNCOV

662



    if (taosArrayPush(pReq->pTaskStatus, &entry) == NULL) {





×








UNCOV

663



      TAOS_CHECK_EXIT(terrno);





×







664



    }










665



  }










666













UNCOV

667



  int32_t numOfVgs = 0;




×








UNCOV

668



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &numOfVgs));





×







669













UNCOV

670



  if ((pReq->pUpdateNodes = taosArrayInit(numOfVgs, sizeof(int32_t))) == NULL) {





×








UNCOV

671



    TAOS_CHECK_EXIT(terrno);





×







672



  }










673













UNCOV

674



  for (int j = 0; j < numOfVgs; ++j) {





×








UNCOV

675



    int32_t vgId = 0;




×








UNCOV

676



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgId));





×








UNCOV

677



    if (taosArrayPush(pReq->pUpdateNodes, &vgId) == NULL) {





×








UNCOV

678



      TAOS_CHECK_EXIT(terrno);





×







679



    }










680



  }










681













UNCOV

682



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->msgId));





×








UNCOV

683



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->ts));





×








UNCOV

684



  tEndDecode(pDecoder);




×







685













UNCOV

686



_exit:




×








UNCOV

687



  return code;




×







688



}










689












690



void tCleanupStreamHbMsg(SStreamHbMsg* pMsg) {




865✔





691



  if (pMsg == NULL) {





865!






UNCOV

692



    return;




×







693



  }










694












695



  if (pMsg->pUpdateNodes != NULL) {





865✔





696



    taosArrayDestroy(pMsg->pUpdateNodes);




422✔





697



    pMsg->pUpdateNodes = NULL;




422✔





698



  }










699












700



  if (pMsg->pTaskStatus != NULL) {





865✔





701



    taosArrayDestroy(pMsg->pTaskStatus);




422✔





702



    pMsg->pTaskStatus = NULL;




422✔





703



  }










704












705



  pMsg->msgId = -1;




865✔





706



  pMsg->vgId = -1;




865✔





707



  pMsg->numOfTasks = -1;




865✔





708



}










709













UNCOV

710



int32_t tEncodeStreamHbRsp(SEncoder* pEncoder, const SMStreamHbRspMsg* pRsp) {




×








UNCOV

711



  int32_t code = 0;




×







712



  int32_t lino;










713













UNCOV

714



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

715



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->msgId));





×








UNCOV

716



  TAOS_CHECK_EXIT(tEncodeSEpSet(pEncoder, &pRsp->mndEpset));





×








UNCOV

717



  tEndEncode(pEncoder);




×







718












719



_exit:




×







720



  return code;




×







721



}










722












723



int32_t tDecodeStreamHbRsp(SDecoder* pDecoder, SMStreamHbRspMsg* pRsp) {




×







724



  int32_t code = 0;




×







725



  int32_t lino;










726












727



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







728



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->msgId));





×







729



  TAOS_CHECK_EXIT(tDecodeSEpSet(pDecoder, &pRsp->mndEpset));





×







730



  tEndDecode(pDecoder);




×







731












732



_exit:




×







733



  return code;




×







734



}










735













UNCOV

736



int32_t tEncodeRetrieveChkptTriggerReq(SEncoder* pEncoder, const SRetrieveChkptTriggerReq* pReq) {




×








UNCOV

737



  int32_t code = 0;




×







738



  int32_t lino;










739













UNCOV

740



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

741



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

742



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->checkpointId));





×








UNCOV

743



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamNodeId));





×








UNCOV

744



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamTaskId));





×








UNCOV

745



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->downstreamNodeId));





×








UNCOV

746



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->downstreamTaskId));





×








UNCOV

747



  tEndEncode(pEncoder);




×







748













UNCOV

749



_exit:




×








UNCOV

750



  return code;




×







751



}










752












753



int32_t tDecodeRetrieveChkptTriggerReq(SDecoder* pDecoder, SRetrieveChkptTriggerReq* pReq) {




×







754



  int32_t code = 0;




×







755



  int32_t lino;










756












757



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







758



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





×







759



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->checkpointId));





×







760



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->upstreamNodeId));





×







761



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->upstreamTaskId));





×







762



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->downstreamNodeId));





×







763



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->downstreamTaskId));





×







764



  tEndDecode(pDecoder);




×







765












766



_exit:




×







767



  return code;




×







768



}










769













UNCOV

770



int32_t tEncodeCheckpointTriggerRsp(SEncoder* pEncoder, const SCheckpointTriggerRsp* pRsp) {




×








UNCOV

771



  int32_t code = 0;




×







772



  int32_t lino;










773













UNCOV

774



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

775



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pRsp->streamId));





×








UNCOV

776



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pRsp->checkpointId));





×








UNCOV

777



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->upstreamTaskId));





×








UNCOV

778



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->taskId));





×








UNCOV

779



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->transId));





×








UNCOV

780



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->rspCode));





×








UNCOV

781



  tEndEncode(pEncoder);




×







782













UNCOV

783



_exit:




×








UNCOV

784



  return code;




×







785



}










786













UNCOV

787



int32_t tDecodeCheckpointTriggerRsp(SDecoder* pDecoder, SCheckpointTriggerRsp* pRsp) {




×








UNCOV

788



  int32_t code = 0;




×







789



  int32_t lino;










790













UNCOV

791



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

792



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pRsp->streamId));





×








UNCOV

793



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pRsp->checkpointId));





×








UNCOV

794



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->upstreamTaskId));





×








UNCOV

795



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->taskId));





×








UNCOV

796



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->transId));





×








UNCOV

797



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->rspCode));





×








UNCOV

798



  tEndDecode(pDecoder);




×







799













UNCOV

800



_exit:




×








UNCOV

801



  return code;




×







802



}










803













UNCOV

804



int32_t tEncodeStreamTaskChkptReport(SEncoder* pEncoder, const SCheckpointReport* pReq) {




×








UNCOV

805



  int32_t code = 0;




×







806



  int32_t lino;










807













UNCOV

808



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

809



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

810



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->taskId));





×








UNCOV

811



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->nodeId));





×








UNCOV

812



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->checkpointId));





×








UNCOV

813



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->checkpointVer));





×








UNCOV

814



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->checkpointTs));





×








UNCOV

815



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->transId));





×








UNCOV

816



  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pReq->dropHTask));





×








UNCOV

817



  tEndEncode(pEncoder);




×







818













UNCOV

819



_exit:




×








UNCOV

820



  return code;




×







821



}










822













UNCOV

823



int32_t tDecodeStreamTaskChkptReport(SDecoder* pDecoder, SCheckpointReport* pReq) {




×








UNCOV

824



  int32_t code = 0;




×







825



  int32_t lino;










826













UNCOV

827



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

828



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





×








UNCOV

829



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->taskId));





×








UNCOV

830



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->nodeId));





×








UNCOV

831



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->checkpointId));





×








UNCOV

832



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->checkpointVer));





×








UNCOV

833



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->checkpointTs));





×








UNCOV

834



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->transId));





×








UNCOV

835



  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pReq->dropHTask));





×








UNCOV

836



  tEndDecode(pDecoder);




×







837













UNCOV

838



_exit:




×








UNCOV

839



  return code;




×







840



}










841













UNCOV

842



int32_t tEncodeRestoreCheckpointInfo(SEncoder* pEncoder, const SRestoreCheckpointInfo* pReq) {




×








UNCOV

843



  int32_t code = 0;




×







844



  int32_t lino;










845













UNCOV

846



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

847



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->startTs));





×








UNCOV

848



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

849



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->checkpointId));





×








UNCOV

850



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->transId));





×








UNCOV

851



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->taskId));





×








UNCOV

852



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->nodeId));





×








UNCOV

853



  tEndEncode(pEncoder);




×







854













UNCOV

855



_exit:




×








UNCOV

856



  return code;




×







857



}










858












859



// todo: serialized term attributes.











UNCOV

860



int32_t tDecodeRestoreCheckpointInfo(SDecoder* pDecoder, SRestoreCheckpointInfo* pReq) {




×








UNCOV

861



  int32_t code = 0;




×







862



  int32_t lino;










863













UNCOV

864



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

865



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->startTs));





×








UNCOV

866



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





×








UNCOV

867



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->checkpointId));





×








UNCOV

868



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->transId));





×








UNCOV

869



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->taskId));





×








UNCOV

870



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->nodeId));





×








UNCOV

871



  tEndDecode(pDecoder);




×







872













UNCOV

873



_exit:




×








UNCOV

874



  return code;




×







875



}










876












877



int32_t tEncodeStreamTaskRunReq (SEncoder* pEncoder, const SStreamTaskRunReq* pReq) {




1,516✔





878



  int32_t code = 0;




1,516✔





879



  int32_t lino;










880












881



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





1,516!





882



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





3,036!





883



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->taskId));





3,036!





884



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->reqType));





3,036!





885



  tEndEncode(pEncoder);




1,518✔





886












887



_exit:




1,518✔





888



  return code;




1,518✔





889



}










890












891



int32_t tDecodeStreamTaskRunReq(SDecoder* pDecoder, SStreamTaskRunReq* pReq) {




1,518✔





892



  int32_t code = 0;




1,518✔





893



  int32_t lino;










894












895



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





1,518!





896



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





3,036!





897



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->taskId));





3,036!





898



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->reqType));





3,036!





899



  tEndDecode(pDecoder);




1,518✔





900












901



_exit:




1,518✔





902



  return code;




1,518✔





903



}










904












905



int32_t tEncodeStreamTaskStopReq(SEncoder* pEncoder, const SStreamTaskStopReq* pReq) {




32✔





906



  int32_t code = 0;




32✔





907



  int32_t lino;










908












909



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





32!





910



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





64!





911



  tEndEncode(pEncoder);




32✔





912












913



_exit:




32✔





914



  return code;




32✔





915



}










916












917



int32_t tDecodeStreamTaskStopReq(SDecoder* pDecoder, SStreamTaskStopReq* pReq) {




16✔





918



  int32_t code = 0;




16✔





919



  int32_t lino;










920












921



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





16!





922



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





32!





923



  tEndDecode(pDecoder);




16✔





924












925



_exit:




16✔





926



  return code;




16✔





927












928



}
























    

  • 

    •