taosdata/TDengine | Build 5011 | source/common/src/msg/streamMsg.c | Coveralls

27

int32_t tEncodeSStreamMgmtReq(SEncoder* pEncoder, const SStreamMgmtReq* pReq) {

137,710✔

28

  int32_t code = 0;

137,710✔

29

  int32_t lino = 0;

137,710✔

30

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

275,420✔

31

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

275,420✔

32

  switch (pReq->type) {

137,710✔

33

    case STREAM_MGMT_REQ_TRIGGER_ORIGTBL_READER: {

132,408✔

34

      if (pReq->cont.pReqs) {

132,408✔

35

        int32_t num = taosArrayGetSize(pReq->cont.pReqs);

132,408✔

36

        TAOS_CHECK_EXIT(tEncodeI32(pEncoder, num));

132,408✔

37

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

520,632✔

38

          SStreamDbTableName* pName = taosArrayGet(pReq->cont.pReqs, i);

388,224✔

39

          TAOS_CHECK_EXIT(tEncodeCStrWithLen(pEncoder, pName->dbFName, strlen(pName->dbFName) + 1));

776,448✔

40

          TAOS_CHECK_EXIT(tEncodeCStrWithLen(pEncoder, pName->tbName, strlen(pName->tbName) + 1));

776,448✔

45

      break;

132,408✔

47

    case STREAM_MGMT_REQ_RUNNER_ORIGTBL_READER: {

5,302✔

48

      if (pReq->cont.pReqs) {

5,302✔

49

        int32_t num = taosArrayGetSize(pReq->cont.pReqs);

5,302✔

50

        TAOS_CHECK_EXIT(tEncodeI32(pEncoder, num));

5,302✔

51

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

10,604✔

52

          SStreamOReaderDeployReq* pDeploy = taosArrayGet(pReq->cont.pReqs, i);

5,302✔

53

          int32_t vgIdNum = taosArrayGetSize(pDeploy->vgIds);

5,302✔

54

          TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pDeploy->execId));

10,604✔

55

          TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pDeploy->uid));

10,604✔

56

          TAOS_CHECK_EXIT(tEncodeI32(pEncoder, vgIdNum));

5,302✔

57

          for (int32_t n = 0; n < vgIdNum; ++n) {

10,604✔

58

            TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *(int32_t*)taosArrayGet(pDeploy->vgIds, n)));

10,604✔

64

      break;

5,302✔

71

_exit:

137,710✔

73

  return code;

137,710✔

76

void tFreeRunnerOReaderDeployReq(void* param) {

7,953✔

77

  SStreamOReaderDeployReq* pReq = (SStreamOReaderDeployReq*)param;

7,953✔

78

  if (pReq) {

7,953✔

79

    taosArrayDestroy(pReq->vgIds);

7,953✔

81

7,953✔

83

void tFreeSStreamMgmtReq(SStreamMgmtReq* pReq) {

275,420✔

84

  if (NULL == pReq) {

275,420✔

85

    return;

68,855✔

88

  switch (pReq->type) {

206,565✔

89

    case STREAM_MGMT_REQ_TRIGGER_ORIGTBL_READER:

198,612✔

90

      taosArrayDestroy(pReq->cont.pReqs);

198,612✔

91

      break;

198,612✔

92

    case STREAM_MGMT_REQ_RUNNER_ORIGTBL_READER:

7,953✔

93

      taosArrayDestroyEx(pReq->cont.pReqs, tFreeRunnerOReaderDeployReq);

7,953✔

94

      break;

7,953✔

101

int32_t tCloneSStreamMgmtReq(SStreamMgmtReq* pSrc, SStreamMgmtReq** ppDst) {

68,855✔

102

  *ppDst = NULL;

68,855✔

104

  if (NULL == pSrc) {

68,855✔

108

  int32_t code = 0, lino = 0;

68,855✔

109

  *ppDst = taosMemoryCalloc(1, sizeof(SStreamMgmtReq));

68,855✔

110

  TSDB_CHECK_NULL(*ppDst, code, lino, _exit, terrno);

68,855✔

112

  memcpy(*ppDst, pSrc, sizeof(*pSrc));

68,855✔

113

  if (pSrc->cont.pReqs) {

68,855✔

114

    switch (pSrc->type) {

68,855✔

115

      case STREAM_MGMT_REQ_TRIGGER_ORIGTBL_READER:

66,204✔

116

        (*ppDst)->cont.pReqs = taosArrayDup(pSrc->cont.pReqs, NULL);

66,204✔

117

        TSDB_CHECK_NULL((*ppDst)->cont.pReqs, code, lino, _exit, terrno);

66,204✔

118

        break;

66,204✔

119

      case STREAM_MGMT_REQ_RUNNER_ORIGTBL_READER: {

2,651✔

120

        int32_t reqNum = taosArrayGetSize(pSrc->cont.pReqs);

2,651✔

121

        (*ppDst)->cont.pReqs = taosArrayInit_s(sizeof(SStreamOReaderDeployReq), reqNum);

2,651✔

122

        TSDB_CHECK_NULL((*ppDst)->cont.pReqs, code, lino, _exit, terrno);

2,651✔

123

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

5,302✔

124

          SStreamOReaderDeployReq* pNew = taosArrayGet((*ppDst)->cont.pReqs, i);

2,651✔

125

          SStreamOReaderDeployReq* pReq = taosArrayGet(pSrc->cont.pReqs, i);

2,651✔

126

          pNew->vgIds = taosArrayDup(pReq->vgIds, NULL);

2,651✔

127

          TSDB_CHECK_NULL(pNew->vgIds, code, lino, _exit, terrno);

2,651✔

128

          pNew->execId = pReq->execId;

2,651✔

129

          pNew->uid = pReq->uid;

2,651✔

131

        break;

2,651✔

140

  if (code) {

68,855✔

146

  return code;

68,855✔

150

int32_t tDecodeSStreamMgmtReq(SDecoder* pDecoder, SStreamMgmtReq* pReq) {

68,855✔

151

  int32_t code = 0;

68,855✔

152

  int32_t lino = 0;

68,855✔

154

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

137,710✔

155

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

137,710✔

156

  switch (pReq->type) {

68,855✔

157

    case STREAM_MGMT_REQ_TRIGGER_ORIGTBL_READER: {

66,204✔

158

      int32_t num = 0;

66,204✔

159

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &num));

66,204✔

160

      if (num > 0) {

66,204✔

161

        pReq->cont.pReqs = taosArrayInit(num, sizeof(SStreamDbTableName));

66,204✔

162

        TSDB_CHECK_NULL(pReq->cont.pReqs, code, lino, _exit, terrno);

66,204✔

163

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

260,316✔

164

          SStreamDbTableName* p = taosArrayReserve(pReq->cont.pReqs, 1);

194,112✔

165

          TAOS_CHECK_EXIT(tDecodeCStrTo(pDecoder, p->dbFName));

194,112✔

166

          TAOS_CHECK_EXIT(tDecodeCStrTo(pDecoder, p->tbName));

194,112✔

169

      break;

66,204✔

171

    case STREAM_MGMT_REQ_RUNNER_ORIGTBL_READER: {

2,651✔

172

      int32_t num = 0, vgIdNum = 0;

2,651✔

173

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &num));

2,651✔

174

      if (num > 0) {

2,651✔

175

        pReq->cont.pReqs = taosArrayInit_s(sizeof(SStreamOReaderDeployReq), num);

2,651✔

176

        TSDB_CHECK_NULL(pReq->cont.pReqs, code, lino, _exit, terrno);

2,651✔

177

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

5,302✔

178

          SStreamOReaderDeployReq* p = taosArrayGet(pReq->cont.pReqs, i);

2,651✔

179

          TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &p->execId));

5,302✔

180

          TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &p->uid));

5,302✔

181

          TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgIdNum));

2,651✔

182

          if (vgIdNum > 0) {

2,651✔

183

            p->vgIds = taosArrayInit_s(sizeof(int32_t), vgIdNum);

2,651✔

184

            TSDB_CHECK_NULL(p->vgIds, code, lino, _exit, terrno);

2,651✔

186

          for (int32_t n = 0; n < vgIdNum; ++n) {

5,302✔

187

            int32_t* vgId = taosArrayGet(p->vgIds, n);

2,651✔

188

            TAOS_CHECK_EXIT(tDecodeI32(pDecoder, vgId));

2,651✔

192

      break;

2,651✔

199

_exit:

68,855✔

201

  return code;

68,855✔

204

int32_t tEncodeStreamTask(SEncoder* pEncoder, const SStreamTask* pTask) {

75,971,706✔

205

  int32_t code = 0;

75,971,706✔

208

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

151,943,412✔

209

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

151,943,412✔

210

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pTask->taskId));

151,943,412✔

212

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pTask->flags));

151,943,412✔

213

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pTask->seriousId));

151,943,412✔

214

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pTask->deployId));

151,943,412✔

215

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

151,943,412✔

217

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pTask->taskIdx));

151,943,412✔

218

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

151,943,412✔

219

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pTask->detailStatus));

151,943,412✔

220

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pTask->errorCode));

151,943,412✔

221

  if (pTask->pMgmtReq) {

75,971,706✔

222

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, 1));

137,710✔

223

    TAOS_CHECK_EXIT(tEncodeSStreamMgmtReq(pEncoder, pTask->pMgmtReq));

137,710✔

225

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, 0));

75,833,996✔

228

_exit:

75,833,996✔

230

  return code;

75,971,706✔

234

int32_t tDecodeStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {

36,550,719✔

235

  int32_t code = 0;

36,550,719✔

238

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, (int32_t*)&pTask->type));

73,101,438✔

239

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

73,101,438✔

240

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pTask->taskId));

73,101,438✔

242

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pTask->flags));

73,101,438✔

243

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pTask->seriousId));

73,101,438✔

244

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pTask->deployId));

73,101,438✔

245

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

73,101,438✔

247

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pTask->taskIdx));

73,101,438✔

248

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, (int32_t*)&pTask->status));

73,101,438✔

249

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pTask->detailStatus));

73,101,438✔

250

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pTask->errorCode));

73,101,438✔

251

  int32_t req = 0;

36,550,719✔

252

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &req));

36,550,719✔

253

  if (req) {

36,550,719✔

254

    pTask->pMgmtReq = taosMemoryCalloc(1, sizeof(SStreamMgmtReq));

68,855✔

255

    TSDB_CHECK_NULL(pTask->pMgmtReq, code, lino, _exit, terrno);

68,855✔

256

    TAOS_CHECK_EXIT(tDecodeSStreamMgmtReq(pDecoder, pTask->pMgmtReq));

68,855✔

259

_exit:

36,550,719✔

261

  return code;

36,550,719✔

293

int32_t tEncodeSSTriggerRuntimeStatus(SEncoder* pEncoder, const SSTriggerRuntimeStatus* pStatus) {

4,503,630✔

294

  int32_t code = 0;

4,503,630✔

297

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pStatus->autoRecalcNum));

9,007,260✔

298

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pStatus->realtimeSessionNum));

9,007,260✔

299

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pStatus->historySessionNum));

9,007,260✔

300

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pStatus->recalcSessionNum));

9,007,260✔

301

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pStatus->histroyProgress));

9,007,260✔

303

  int32_t recalcNum = (int32_t)taosArrayGetSize(pStatus->userRecalcs);

4,503,630✔

304

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, recalcNum));

4,503,630✔

305

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

4,503,630✔

310

_exit:

4,503,630✔

312

  return code;

4,503,630✔

315

int32_t tDecodeSSTriggerRuntimeStatus(SDecoder* pDecoder, SSTriggerRuntimeStatus* pStatus) {

2,115,750✔

316

  int32_t code = 0;

2,115,750✔

319

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pStatus->autoRecalcNum));

4,231,500✔

320

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pStatus->realtimeSessionNum));

4,231,500✔

321

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pStatus->historySessionNum));

4,231,500✔

322

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pStatus->realtimeSessionNum));

4,231,500✔

323

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pStatus->histroyProgress));

4,231,500✔

325

  int32_t recalcNum = 0;

2,115,750✔

326

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &recalcNum));

2,115,750✔

327

  if (recalcNum > 0) {

2,115,750✔

335

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

2,115,750✔

340

_exit:

2,115,750✔

342

  return code;

2,115,750✔

346

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

37,425,258✔

347

  int32_t code = 0;

37,425,258✔

350

  TAOS_CHECK_EXIT(tStartEncode(pEncoder));

37,425,258✔

351

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

74,850,516✔

352

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

74,850,516✔

353

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

74,850,516✔

354

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

74,850,516✔

356

  int32_t vgLeaderNum = taosArrayGetSize(pReq->pVgLeaders);

37,425,258✔

357

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, vgLeaderNum));

37,425,258✔

358

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

152,615,538✔

359

    int32_t* vgId = taosArrayGet(pReq->pVgLeaders, i);

115,190,280✔

360

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *vgId));

230,380,560✔

363

  int32_t statusNum = taosArrayGetSize(pReq->pStreamStatus);

37,425,258✔

364

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, statusNum));

37,425,258✔

365

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

109,018,194✔

366

    SStmTaskStatusMsg* pStatus = taosArrayGet(pReq->pStreamStatus, i);

71,592,936✔

367

    TAOS_CHECK_EXIT(tEncodeStreamTask(pEncoder, (SStreamTask*)pStatus));

71,592,936✔

370

  int32_t reqNum = taosArrayGetSize(pReq->pStreamReq);

37,425,258✔

371

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, reqNum));

37,425,258✔

372

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

37,562,968✔

373

    int32_t* idx = taosArrayGet(pReq->pStreamReq, i);

137,710✔

374

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *idx));

275,420✔

377

  int32_t triggerNum = taosArrayGetSize(pReq->pTriggerStatus);

37,425,258✔

378

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, triggerNum));

37,425,258✔

379

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

41,928,888✔

380

    SSTriggerRuntimeStatus* pTrigger = taosArrayGet(pReq->pTriggerStatus, i);

4,503,630✔

381

    TAOS_CHECK_EXIT(tEncodeSSTriggerRuntimeStatus(pEncoder, pTrigger));

4,503,630✔

384

  tEndEncode(pEncoder);

37,425,258✔

386

_exit:

37,425,258✔

387

  if (code) {

37,425,258✔

390

    return pEncoder->pos;

37,425,258✔

394

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

17,952,902✔

395

  int32_t code = 0;

17,952,902✔

398

  TAOS_CHECK_EXIT(tStartDecode(pDecoder));

17,952,902✔

399

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

35,905,804✔

400

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

35,905,804✔

401

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

35,905,804✔

402

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

35,905,804✔

404

  int32_t vgLearderNum = 0;

17,952,902✔

405

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgLearderNum));

17,952,902✔

406

  if (vgLearderNum > 0) {

17,952,902✔

407

    pReq->pVgLeaders = taosArrayInit(vgLearderNum, sizeof(int32_t));

14,246,550✔

408

    if (NULL == pReq->pVgLeaders) {

14,246,550✔

413

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

74,270,788✔

414

    int32_t vgId = 0;

56,317,886✔

415

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgId));

56,317,886✔

416

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

112,635,772✔

423

  int32_t statusNum = 0;

17,952,902✔

424

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &statusNum));

17,952,902✔

425

  if (statusNum > 0) {

17,952,902✔

426

    pReq->pStreamStatus = taosArrayInit_s(sizeof(SStmTaskStatusMsg), statusNum);

1,251,818✔

427

    if (NULL == pReq->pStreamStatus) {

1,251,818✔

432

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

52,313,451✔

433

    SStmTaskStatusMsg* pTask = taosArrayGet(pReq->pStreamStatus, i);

34,360,549✔

434

    if (NULL == pTask) {

34,360,549✔

438

    TAOS_CHECK_EXIT(tDecodeStreamTask(pDecoder, (SStreamTask*)pTask));

34,360,549✔

442

  int32_t reqNum = 0;

17,952,902✔

443

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &reqNum));

17,952,902✔

444

  if (reqNum > 0) {

17,952,902✔

445

    pReq->pStreamReq = taosArrayInit_s(sizeof(int32_t), reqNum);

18,924✔

446

    if (NULL == pReq->pStreamReq) {

18,924✔

451

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

18,021,757✔

452

    int32_t* pIdx = taosArrayGet(pReq->pStreamReq, i);

68,855✔

453

    if (NULL == pIdx) {

68,855✔

457

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, pIdx));

68,855✔

461

  int32_t triggerNum = 0;

17,952,902✔

462

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &triggerNum));

17,952,902✔

463

  if (triggerNum > 0) {

17,952,902✔

464

    pReq->pTriggerStatus = taosArrayInit_s(sizeof(SSTriggerRuntimeStatus), triggerNum);

560,452✔

465

    if (NULL == pReq->pTriggerStatus) {

560,452✔

470

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

20,068,652✔

471

    SSTriggerRuntimeStatus* pStatus = taosArrayGet(pReq->pTriggerStatus, i);

2,115,750✔

472

    if (NULL == pStatus) {

2,115,750✔

476

    TAOS_CHECK_EXIT(tDecodeSSTriggerRuntimeStatus(pDecoder, pStatus));

2,115,750✔

480

  tEndDecode(pDecoder);

17,952,902✔

482

_exit:

17,952,902✔

483

  return code;

17,952,902✔

486

void tFreeSSTriggerRuntimeStatus(void* param) {

4,367,565✔

487

  SSTriggerRuntimeStatus* pStatus = (SSTriggerRuntimeStatus*)param;

4,367,565✔

488

  if (NULL == pStatus) {

4,367,565✔

491

  taosArrayDestroy(pStatus->userRecalcs);

4,367,565✔

494

void tCleanupStreamHbMsg(SStreamHbMsg* pMsg, bool deepClean) {

108,626,763✔

495

  if (pMsg == NULL) {

108,626,763✔

499

  taosArrayDestroy(pMsg->pVgLeaders);

108,626,763✔

500

  if (deepClean) {

108,626,763✔

501

    int32_t reqNum = taosArrayGetSize(pMsg->pStreamReq);

108,626,763✔

502

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

108,764,473✔

503

      int32_t* idx = taosArrayGet(pMsg->pStreamReq, i);

137,710✔

504

      SStmTaskStatusMsg* pTask = taosArrayGet(pMsg->pStreamStatus, *idx);

137,710✔

505

      if (NULL == pTask) {

137,710✔

509

      tFreeSStreamMgmtReq(pTask->pMgmtReq);

137,710✔

510

      taosMemoryFree(pTask->pMgmtReq);

137,710✔

513

  taosArrayDestroy(pMsg->pStreamReq);

108,626,763✔

514

  taosArrayDestroy(pMsg->pStreamStatus);

108,626,763✔

515

  taosArrayDestroyEx(pMsg->pTriggerStatus, tFreeSSTriggerRuntimeStatus);

108,626,763✔

518

int32_t tEncodeSStreamReaderDeployFromTrigger(SEncoder* pEncoder, const SStreamReaderDeployFromTrigger* pMsg) {

646,812✔

519

  int32_t code = 0;

646,812✔

522

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, (const uint8_t*)pMsg->triggerTblName, pMsg->triggerTblName == NULL ? 0 : (int32_t)strlen(pMsg->triggerTblName) + 1));

1,293,624✔

523

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

1,293,624✔

524

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

1,293,624✔

525

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->triggerTblType));

1,293,624✔

526

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->isTriggerTblVirt));

1,293,624✔

527

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->deleteReCalc));

1,293,624✔

528

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->deleteOutTbl));

1,293,624✔

529

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->partitionCols, pMsg->partitionCols == NULL ? 0 : (int32_t)strlen(pMsg->partitionCols) + 1));

1,293,624✔

530

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->triggerCols, pMsg->triggerCols == NULL ? 0 : (int32_t)strlen(pMsg->triggerCols) + 1));

1,293,624✔

532

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->triggerScanPlan, pMsg->triggerScanPlan == NULL ? 0 : (int32_t)strlen(pMsg->triggerScanPlan) + 1));

1,293,624✔

533

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->calcCacheScanPlan, pMsg->calcCacheScanPlan == NULL ? 0 : (int32_t)strlen(pMsg->calcCacheScanPlan) + 1));

1,293,624✔

535

_exit:

646,812✔

537

  return code;

646,812✔

540

int32_t tEncodeSStreamReaderDeployFromCalc(SEncoder* pEncoder, const SStreamReaderDeployFromCalc* pMsg) {

774,182✔

541

  int32_t code = 0;

774,182✔

544

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

1,548,364✔

545

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->calcScanPlan, pMsg->calcScanPlan == NULL ? 0 : (int32_t)strlen(pMsg->calcScanPlan) + 1));

1,548,364✔

547

_exit:

774,182✔

549

  return code;

774,182✔

553

int32_t tEncodeSStreamReaderDeployMsg(SEncoder* pEncoder, const SStreamReaderDeployMsg* pMsg) {

1,420,994✔

554

  int32_t code = 0;

1,420,994✔

557

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->triggerReader));

2,841,988✔

558

  if (pMsg->triggerReader) {

1,420,994✔

559

    TAOS_CHECK_EXIT(tEncodeSStreamReaderDeployFromTrigger(pEncoder, &pMsg->msg.trigger));

646,812✔

561

    TAOS_CHECK_EXIT(tEncodeSStreamReaderDeployFromCalc(pEncoder, &pMsg->msg.calc));

774,182✔

564

_exit:

774,182✔

566

  return code;

1,420,994✔

569

int32_t tEncodeSStreamTaskAddr(SEncoder* pEncoder, const SStreamTaskAddr* pMsg) {

1,903,892✔

570

  int32_t code = 0;

1,903,892✔

573

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

3,807,784✔

574

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

3,807,784✔

575

  TAOS_CHECK_EXIT(tEncodeSEpSet(pEncoder, &pMsg->epset));

1,903,892✔

577

_exit:

1,903,892✔

579

  return code;

1,903,892✔

582

int32_t tEncodeSStreamRunnerTarget(SEncoder* pEncoder, const SStreamRunnerTarget* pMsg) {

1,229,682✔

583

  int32_t code = 0;

1,229,682✔

586

  TAOS_CHECK_EXIT(tEncodeSStreamTaskAddr(pEncoder, &pMsg->addr));

1,229,682✔

587

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

2,459,364✔

589

_exit:

1,229,682✔

591

  return code;

1,229,682✔

595

int32_t tEncodeSStreamTriggerDeployMsg(SEncoder* pEncoder, const SStreamTriggerDeployMsg* pMsg) {

416,754✔

596

  int32_t code = 0;

416,754✔

599

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->triggerType));

833,508✔

600

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->igDisorder));

833,508✔

601

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->fillHistory));

833,508✔

602

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->fillHistoryFirst));

833,508✔

603

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->lowLatencyCalc));

833,508✔

604

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->igNoDataTrigger));

833,508✔

605

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->enableMultiGroupCalc));

833,508✔

606

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->isTriggerTblVirt));

833,508✔

607

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->triggerHasPF));

833,508✔

608

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->isTriggerTblStb));

833,508✔

609

  int32_t partitionColsLen = pMsg->partitionCols == NULL ? 0 : (int32_t)strlen((char*)pMsg->partitionCols) + 1;

416,754✔

610

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->partitionCols, partitionColsLen));

833,508✔

612

  int32_t addrSize = (int32_t)taosArrayGetSize(pMsg->pNotifyAddrUrls);

416,754✔

613

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, addrSize));

416,754✔

614

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

531,606✔

615

    const char *url = taosArrayGetP(pMsg->pNotifyAddrUrls, i);

114,852✔

616

    TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, url, NULL == url ? 0 : (int32_t)strlen(url) + 1));

229,704✔

618

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

833,508✔

619

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

833,508✔

620

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->notifyHistory));

833,508✔

622

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

833,508✔

623

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

833,508✔

624

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

833,508✔

625

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

833,508✔

626

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

833,508✔

628

  switch (pMsg->triggerType) {

416,754✔

629

    case WINDOW_TYPE_SESSION: {

17,482✔

631

      TAOS_CHECK_EXIT(tEncodeI16(pEncoder, pMsg->trigger.session.slotId));

34,964✔

632

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pMsg->trigger.session.sessionVal));

34,964✔

633

      break;

17,482✔

635

    case WINDOW_TYPE_STATE: {

136,756✔

637

      TAOS_CHECK_EXIT(tEncodeI16(pEncoder, pMsg->trigger.stateWin.slotId));

273,512✔

638

      TAOS_CHECK_EXIT(tEncodeI16(pEncoder, pMsg->trigger.stateWin.extend));

273,512✔

639

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pMsg->trigger.stateWin.trueForType));

273,512✔

640

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pMsg->trigger.stateWin.trueForCount));

273,512✔

641

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pMsg->trigger.stateWin.trueForDuration));

273,512✔

642

      int32_t stateWindowZerothLen =

136,756✔

643

          pMsg->trigger.stateWin.zeroth == NULL ? 0 : (int32_t)strlen((char*)pMsg->trigger.stateWin.zeroth) + 1;

136,756✔

644

      TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->trigger.stateWin.zeroth, stateWindowZerothLen));

273,512✔

645

      int32_t stateWindowExprLen =

136,756✔

646

          pMsg->trigger.stateWin.expr == NULL ? 0 : (int32_t)strlen((char*)pMsg->trigger.stateWin.expr) + 1;

136,756✔

647

      TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->trigger.stateWin.expr, stateWindowExprLen));

273,512✔

648

      break;

136,756✔

650

    case WINDOW_TYPE_INTERVAL: {

156,350✔

652

      TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->trigger.sliding.intervalUnit));

312,700✔

653

      TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->trigger.sliding.slidingUnit));

312,700✔

654

      TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->trigger.sliding.offsetUnit));

312,700✔

655

      TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->trigger.sliding.soffsetUnit));

312,700✔

656

      TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->trigger.sliding.precision));

312,700✔

657

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pMsg->trigger.sliding.interval));

312,700✔

658

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pMsg->trigger.sliding.offset));

312,700✔

659

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pMsg->trigger.sliding.sliding));

312,700✔

660

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pMsg->trigger.sliding.soffset));

312,700✔

661

      break;

156,350✔

663

    case WINDOW_TYPE_EVENT: {

50,508✔

665

      int32_t eventWindowStartCondLen = pMsg->trigger.event.startCond == NULL ? 0 : (int32_t)strlen((char*)pMsg->trigger.event.startCond) + 1;

50,508✔

666

      int32_t eventWindowEndCondLen = pMsg->trigger.event.endCond == NULL ? 0 : (int32_t)strlen((char*)pMsg->trigger.event.endCond) + 1;

50,508✔

668

      TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->trigger.event.startCond, eventWindowStartCondLen));

101,016✔

669

      TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->trigger.event.endCond, eventWindowEndCondLen));

101,016✔

670

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pMsg->trigger.event.trueForType));

101,016✔

671

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pMsg->trigger.event.trueForCount));

101,016✔

672

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pMsg->trigger.event.trueForDuration));

101,016✔

673

      break;

50,508✔

675

    case WINDOW_TYPE_COUNT: {

40,340✔

677

      int32_t countWindowCondColsLen = pMsg->trigger.count.condCols == NULL ? 0 : (int32_t)strlen((char*)pMsg->trigger.count.condCols) + 1;

40,340✔

678

      TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->trigger.count.condCols, countWindowCondColsLen));

80,680✔

680

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pMsg->trigger.count.countVal));

80,680✔

681

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pMsg->trigger.count.sliding));

80,680✔

682

      break;

40,340✔

684

    case WINDOW_TYPE_PERIOD: {

15,318✔

686

      TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->trigger.period.periodUnit));

30,636✔

687

      TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->trigger.period.offsetUnit));

30,636✔

688

      TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->trigger.period.precision));

30,636✔

689

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pMsg->trigger.period.period));

30,636✔

690

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pMsg->trigger.period.offset));

30,636✔

691

      break;

15,318✔

698

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

833,508✔

699

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

833,508✔

700

  TAOS_CHECK_EXIT(tEncodeI16(pEncoder, pMsg->calcTsSlotId));

833,508✔

701

  TAOS_CHECK_EXIT(tEncodeI16(pEncoder, pMsg->triTsSlotId));

833,508✔

702

  TAOS_CHECK_EXIT(tEncodeI16(pEncoder, pMsg->calcPkSlotId));

833,508✔

703

  TAOS_CHECK_EXIT(tEncodeI16(pEncoder, pMsg->triPkSlotId));

833,508✔

704

  int32_t triggerPrevFilterLen = (pMsg->triggerPrevFilter == NULL) ? 0 : ((int32_t)strlen(pMsg->triggerPrevFilter) + 1);

416,754✔

705

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->triggerPrevFilter, triggerPrevFilterLen));

833,508✔

706

  int32_t triggerScanPlanLen = (pMsg->triggerScanPlan == NULL) ? 0 : ((int32_t)strlen(pMsg->triggerScanPlan) + 1);

416,754✔

707

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->triggerScanPlan, triggerScanPlanLen));

833,508✔

708

  int32_t calcCacheScanPlanLen = (pMsg->calcCacheScanPlan == NULL) ? 0 : ((int32_t)strlen(pMsg->calcCacheScanPlan) + 1);

416,754✔

709

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->calcCacheScanPlan, calcCacheScanPlanLen));

833,508✔

711

  int32_t readerNum = taosArrayGetSize(pMsg->readerList);

416,754✔

712

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, readerNum));

416,754✔

713

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

933,528✔

714

    SStreamTaskAddr* pAddr = (SStreamTaskAddr*)taosArrayGet(pMsg->readerList, i);

516,774✔

715

    TAOS_CHECK_EXIT(tEncodeSStreamTaskAddr(pEncoder, pAddr));

516,774✔

718

  int32_t runnerNum = taosArrayGetSize(pMsg->runnerList);

416,754✔

719

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, runnerNum));

416,754✔

720

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

1,646,436✔

721

    SStreamRunnerTarget* pTarget = (SStreamRunnerTarget*)taosArrayGet(pMsg->runnerList, i);

1,229,682✔

722

    TAOS_CHECK_EXIT(tEncodeSStreamRunnerTarget(pEncoder, pTarget));

1,229,682✔

725

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

833,508✔

726

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->streamName, (int32_t)strlen(pMsg->streamName) + 1));

833,508✔

727

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->precision));

833,508✔

728

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->nodelayCreateSubtable));

833,508✔

730

_exit:

416,754✔

732

  return code;

416,754✔

736

int32_t tSerializeSFieldWithOptions(SEncoder* pEncoder, const SFieldWithOptions *pField) {

7,165,984✔

737

  int32_t code = 0;

7,165,984✔

740

  TAOS_CHECK_EXIT(tEncodeCStr(pEncoder, pField->name));

14,331,968✔

741

  TAOS_CHECK_EXIT(tEncodeU8(pEncoder, pField->type));

14,331,968✔

742

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pField->flags));

14,331,968✔

743

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pField->bytes));

14,331,968✔

744

  TAOS_CHECK_EXIT(tEncodeU32(pEncoder, pField->compress));

14,331,968✔

745

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pField->typeMod));

14,331,968✔

747

_exit:

7,165,984✔

749

  return code;

7,165,984✔

753

int32_t tEncodeSStreamRunnerDeployMsg(SEncoder* pEncoder, const SStreamRunnerDeployMsg* pMsg) {

1,457,350✔

754

  int32_t code = 0;

1,457,350✔

757

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

2,914,700✔

758

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->streamName, (int32_t)strlen(pMsg->streamName) + 1));

2,914,700✔

759

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->pPlan, NULL == pMsg->pPlan ? 0 : (int32_t)strlen(pMsg->pPlan) + 1));

2,914,700✔

760

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->outDBFName, NULL == pMsg->outDBFName ? 0 : (int32_t)strlen(pMsg->outDBFName) + 1));

2,914,700✔

761

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->outTblName, NULL == pMsg->outTblName ? 0 : (int32_t)strlen(pMsg->outTblName) + 1));

2,914,700✔

762

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->outTblType));

2,914,700✔

763

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->calcNotifyOnly));

2,914,700✔

764

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->topPlan));

2,914,700✔

766

  int32_t addrSize = (int32_t)taosArrayGetSize(pMsg->pNotifyAddrUrls);

1,457,350✔

767

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, addrSize));

1,457,350✔

768

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

1,781,326✔

769

    const char *url = taosArrayGetP(pMsg->pNotifyAddrUrls, i);

323,976✔

770

    TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, url, NULL == url ? 0 : (int32_t)strlen(url) + 1));

647,952✔

772

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

2,914,700✔

774

  int32_t outColNum = (int32_t)taosArrayGetSize(pMsg->outCols);

1,457,350✔

775

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, outColNum));

1,457,350✔

776

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

7,415,278✔

777

    SFieldWithOptions *pCol = taosArrayGet(pMsg->outCols, i);

5,957,928✔

778

    TAOS_CHECK_EXIT(tSerializeSFieldWithOptions(pEncoder, pCol));

5,957,928✔

781

  int32_t outTagNum = (int32_t)taosArrayGetSize(pMsg->outTags);

1,457,350✔

782

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, outTagNum));

1,457,350✔

783

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

2,665,406✔

784

    SFieldWithOptions *pTag = taosArrayGet(pMsg->outTags, i);

1,208,056✔

785

    TAOS_CHECK_EXIT(tSerializeSFieldWithOptions(pEncoder, pTag));

1,208,056✔

788

  TAOS_CHECK_EXIT(tEncodeU64(pEncoder, pMsg->outStbUid));

2,914,700✔

789

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

2,914,700✔

791

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->subTblNameExpr, NULL == pMsg->subTblNameExpr ? 0 : (int32_t)strlen(pMsg->subTblNameExpr) + 1));

2,914,700✔

792

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pMsg->tagValueExpr, NULL == pMsg->tagValueExpr ? 0 : (int32_t)strlen(pMsg->tagValueExpr) + 1));

2,914,700✔

794

  int32_t forceOutColsSize = (int32_t)taosArrayGetSize(pMsg->forceOutCols);

1,457,350✔

795

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, forceOutColsSize));

1,457,350✔

796

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

1,597,852✔

797

    SStreamOutCol *pCoutCol = (SStreamOutCol*)taosArrayGet(pMsg->forceOutCols, i);

140,502✔

798

    int32_t        exprLen = pCoutCol->expr == NULL ? 0 : (int32_t)strlen((char*)pCoutCol->expr) + 1;

140,502✔

800

    TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pCoutCol->expr, exprLen));

281,004✔

801

    TAOS_CHECK_EXIT(tEncodeU8(pEncoder, pCoutCol->type.type));

281,004✔

802

    TAOS_CHECK_EXIT(tEncodeU8(pEncoder, pCoutCol->type.precision));

281,004✔

803

    TAOS_CHECK_EXIT(tEncodeU8(pEncoder, pCoutCol->type.scale));

281,004✔

804

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pCoutCol->type.bytes));

281,004✔

807

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pMsg->lowLatencyCalc));

2,914,700✔

810

  int32_t colCidsSize = (int32_t)taosArrayGetSize(pMsg->colCids);

1,457,350✔

811

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, colCidsSize));

1,457,350✔

812

  if (colCidsSize > 0) {

1,457,350✔

813

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

115,920✔

814

      int16_t* pCid = (int16_t*)taosArrayGet(pMsg->colCids, i);

87,414✔

815

      TAOS_CHECK_EXIT(tEncodeI16(pEncoder, *pCid));

174,828✔

819

  int32_t tagCidsSize = (int32_t)taosArrayGetSize(pMsg->tagCids);

1,457,350✔

820

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, tagCidsSize));

1,457,350✔

821

  if (tagCidsSize > 0) {

1,457,350✔

822

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

52,362✔

823

      int16_t* pCid = (int16_t*)taosArrayGet(pMsg->tagCids, i);

30,678✔

824

      TAOS_CHECK_EXIT(tEncodeI16(pEncoder, *pCid));

61,356✔

828

_exit:

1,457,350✔

830

  return code;

1,457,350✔

833

int32_t tEncodeSStmTaskDeploy(SEncoder* pEncoder, const SStmTaskDeploy* pTask) {

3,295,098✔

834

  int32_t code = 0;

3,295,098✔

837

  TAOS_CHECK_EXIT(tEncodeStreamTask(pEncoder, (SStreamTask*)&pTask->task));

3,295,098✔

838

  switch (pTask->task.type) {

3,295,098✔

839

    case STREAM_READER_TASK:

1,420,994✔

840

      TAOS_CHECK_EXIT(tEncodeSStreamReaderDeployMsg(pEncoder, &pTask->msg.reader));

1,420,994✔

841

      break;

1,420,994✔

842

    case STREAM_TRIGGER_TASK:

416,754✔

843

      TAOS_CHECK_EXIT(tEncodeSStreamTriggerDeployMsg(pEncoder, &pTask->msg.trigger));

416,754✔

844

      break;

416,754✔

845

    case STREAM_RUNNER_TASK:

1,457,350✔

846

      TAOS_CHECK_EXIT(tEncodeSStreamRunnerDeployMsg(pEncoder, &pTask->msg.runner));

1,457,350✔

847

      break;

1,457,350✔

853

_exit:

3,295,098✔

855

  return code;

3,295,098✔

859

int32_t tEncodeSStmStreamDeploy(SEncoder* pEncoder, const SStmStreamDeploy* pStream) {

563,042✔

860

  int32_t code = 0;

563,042✔

863

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

1,126,084✔

865

  int32_t readerNum = taosArrayGetSize(pStream->readerTasks);

563,042✔

866

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, readerNum));

563,042✔

867

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

1,984,036✔

868

    SStmTaskDeploy* pDeploy = taosArrayGet(pStream->readerTasks, i);

1,420,994✔

869

    TAOS_CHECK_EXIT(tEncodeSStmTaskDeploy(pEncoder, pDeploy));

1,420,994✔

872

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pStream->triggerTask ? 1 : 0));

1,126,084✔

873

  if (pStream->triggerTask) {

563,042✔

874

    TAOS_CHECK_EXIT(tEncodeSStmTaskDeploy(pEncoder, pStream->triggerTask));

416,754✔

877

  int32_t runnerNum = taosArrayGetSize(pStream->runnerTasks);

563,042✔

878

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, runnerNum));

563,042✔

879

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

2,020,392✔

880

    SStmTaskDeploy* pDeploy = taosArrayGet(pStream->runnerTasks, i);

1,457,350✔

881

    TAOS_CHECK_EXIT(tEncodeSStmTaskDeploy(pEncoder, pDeploy));

1,457,350✔

884

_exit:

563,042✔

886

  return code;

563,042✔

889

int32_t tEncodeSStreamMsg(SEncoder* pEncoder, const SStreamMsg* pMsg) {

1,083,672✔

890

  int32_t code = 0;

1,083,672✔

891

  int32_t lino = 0;

1,083,672✔

893

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

2,167,344✔

895

_exit:

1,083,672✔

896

  return code;

1,083,672✔

899

int32_t tDecodeSStreamMsg(SDecoder* pDecoder, SStreamMsg* pMsg) {

542,048✔

900

  int32_t code = 0;

542,048✔

903

  int32_t type = 0;

542,048✔

904

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &type));

542,048✔

905

  pMsg->msgType = type;

542,048✔

907

_exit:

542,048✔

908

  return code;

542,048✔

911

int32_t tEncodeSStreamStartTaskMsg(SEncoder* pEncoder, const SStreamStartTaskMsg* pStart) {

409,302✔

912

  int32_t code = 0;

409,302✔

915

  TAOS_CHECK_EXIT(tEncodeSStreamMsg(pEncoder, &pStart->header));

409,302✔

917

_exit:

409,302✔

919

  return code;

409,302✔

922

int32_t tEncodeSStreamTaskStart(SEncoder* pEncoder, const SStreamTaskStart* pTask) {

409,302✔

923

  int32_t code = 0;

409,302✔

926

  TAOS_CHECK_EXIT(tEncodeStreamTask(pEncoder, (SStreamTask*)&pTask->task));

409,302✔

927

  TAOS_CHECK_EXIT(tEncodeSStreamStartTaskMsg(pEncoder, (SStreamStartTaskMsg*)&pTask->startMsg));

409,302✔

929

_exit:

409,302✔

931

  return code;

409,302✔

934

int32_t tEncodeSStreamUndeployTaskMsg(SEncoder* pEncoder, const SStreamUndeployTaskMsg* pUndeploy) {

529,316✔

935

  int32_t code = 0;

529,316✔

938

  TAOS_CHECK_EXIT(tEncodeSStreamMsg(pEncoder, &pUndeploy->header));

529,316✔

939

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pUndeploy->doCheckpoint));

1,058,632✔

940

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pUndeploy->doCleanup));

1,058,632✔

942

_exit:

529,316✔

944

  return code;

529,316✔

947

int32_t tEncodeSStreamTaskUndeploy(SEncoder* pEncoder, const SStreamTaskUndeploy* pTask) {

529,316✔

948

  int32_t code = 0;

529,316✔

951

  TAOS_CHECK_EXIT(tEncodeStreamTask(pEncoder, (SStreamTask*)&pTask->task));

529,316✔

952

  TAOS_CHECK_EXIT(tEncodeSStreamUndeployTaskMsg(pEncoder, (SStreamUndeployTaskMsg*)&pTask->undeployMsg));

529,316✔

954

_exit:

529,316✔

956

  return code;

529,316✔

960

int32_t tEncodeSStreamRecalcReq(SEncoder* pEncoder, const SStreamRecalcReq* recalc) {

7,344✔

961

  int32_t code = 0;

7,344✔

964

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, recalc->recalcId));

14,688✔

965

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, recalc->start));

14,688✔

966

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, recalc->end));

14,688✔

968

_exit:

7,344✔

970

  return code;

7,344✔

973

int32_t tEncodeSStreamMgmtRspCont(SEncoder* pEncoder, SStreamMsgType msgType, const SStreamMgmtRspCont* pRsp) {

145,054✔

974

  int32_t code = 0;

145,054✔

977

  switch (msgType) {

145,054✔

978

    case STREAM_MSG_ORIGTBL_READER_INFO: {

132,408✔

979

      int32_t vgNum = taosArrayGetSize(pRsp->vgIds);

132,408✔

980

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, vgNum));

132,408✔

982

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

520,632✔

983

        int32_t* vgId = taosArrayGet(pRsp->vgIds, i);

388,224✔

984

        TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *vgId));

776,448✔

987

      int32_t readerNum = taosArrayGetSize(pRsp->readerList);

132,408✔

988

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, readerNum));

132,408✔

990

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

260,334✔

991

        SStreamTaskAddr* addr = taosArrayGet(pRsp->readerList, i);

127,926✔

992

        TAOS_CHECK_EXIT(tEncodeSStreamTaskAddr(pEncoder, addr));

127,926✔

994

      break;

132,408✔

1006

    case STREAM_MSG_USER_RECALC: {

7,344✔

1007

      int32_t recalcNum = taosArrayGetSize(pRsp->recalcList);

7,344✔

1008

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, recalcNum));

7,344✔

1010

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

14,688✔

1011

        SStreamRecalcReq* recalc = taosArrayGet(pRsp->recalcList, i);

7,344✔

1012

        TAOS_CHECK_EXIT(tEncodeSStreamRecalcReq(pEncoder, recalc));

7,344✔

1014

      break;

7,344✔

1016

    case STREAM_MSG_RUNNER_ORIGTBL_READER: {

5,302✔

1017

      int32_t rspNum = taosArrayGetSize(pRsp->execRspList);

5,302✔

1018

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, rspNum));

5,302✔

1020

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

10,604✔

1021

        SStreamOReaderDeployRsp* pDeployRsp = taosArrayGet(pRsp->execRspList, i);

5,302✔

1022

        TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pDeployRsp->execId));

10,604✔

1023

        int32_t vgNum = taosArrayGetSize(pDeployRsp->vgList);

5,302✔

1024

        TAOS_CHECK_EXIT(tEncodeI32(pEncoder, vgNum));

5,302✔

1025

        for (int32_t n = 0; n < vgNum; ++n) {

10,604✔

1026

          TAOS_CHECK_EXIT(tEncodeSStreamTaskAddr(pEncoder, taosArrayGet(pDeployRsp->vgList, n)));

5,302✔

1029

      break;

5,302✔

1035

_exit:

145,054✔

1037

  return code;

145,054✔

1040

int32_t tEncodeSStreamMgmtRsp(SEncoder* pEncoder, const SStreamMgmtRsp* pRsp) {

145,054✔

1041

  int32_t code = 0;

145,054✔

1044

  TAOS_CHECK_EXIT(tEncodeSStreamMsg(pEncoder, &pRsp->header));

145,054✔

1045

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

290,108✔

1046

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

290,108✔

1047

  TAOS_CHECK_EXIT(tEncodeStreamTask(pEncoder, &pRsp->task));

145,054✔

1048

  TAOS_CHECK_EXIT(tEncodeSStreamMgmtRspCont(pEncoder, pRsp->header.msgType, (SStreamMgmtRspCont*)&pRsp->cont));

145,054✔

1050

_exit:

145,054✔

1052

  return code;

145,054✔

1056

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

35,586,540✔

1057

  int32_t code = 0;

35,586,540✔

1060

  TAOS_CHECK_EXIT(tStartEncode(pEncoder));

35,586,540✔

1061

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

71,173,080✔

1062

  int32_t deployNum = taosArrayGetSize(pRsp->deploy.streamList);

35,586,540✔

1063

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, deployNum));

35,586,540✔

1064

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

36,149,582✔

1065

    SStmStreamDeploy* pStream = (SStmStreamDeploy*)taosArrayGet(pRsp->deploy.streamList, i);

563,042✔

1066

    TAOS_CHECK_EXIT(tEncodeSStmStreamDeploy(pEncoder, pStream));

563,042✔

1069

  int32_t startNum = taosArrayGetSize(pRsp->start.taskList);

35,586,540✔

1070

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, startNum));

35,586,540✔

1071

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

35,995,842✔

1072

    SStreamTaskStart* pTask = (SStreamTaskStart*)taosArrayGet(pRsp->start.taskList, i);

409,302✔

1073

    TAOS_CHECK_EXIT(tEncodeSStreamTaskStart(pEncoder, pTask));

409,302✔

1076

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pRsp->undeploy.undeployAll));

71,173,080✔

1077

  if (!pRsp->undeploy.undeployAll) {

35,586,540✔

1078

    int32_t undeployNum = taosArrayGetSize(pRsp->undeploy.taskList);

35,586,540✔

1079

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, undeployNum));

35,586,540✔

1080

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

36,115,856✔

1081

      SStreamTaskUndeploy* pTask = (SStreamTaskUndeploy*)taosArrayGet(pRsp->undeploy.taskList, i);

529,316✔

1082

      TAOS_CHECK_EXIT(tEncodeSStreamTaskUndeploy(pEncoder, pTask));

529,316✔

1086

  int32_t rspNum = taosArrayGetSize(pRsp->rsps.rspList);

35,586,540✔

1087

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, rspNum));

35,586,540✔

1088

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

35,731,594✔

1089

    SStreamMgmtRsp* pMgmtRsp = (SStreamMgmtRsp*)taosArrayGet(pRsp->rsps.rspList, i);

145,054✔

1090

    TAOS_CHECK_EXIT(tEncodeSStreamMgmtRsp(pEncoder, pMgmtRsp));

145,054✔

1093

_exit:

35,586,540✔

1095

  tEndEncode(pEncoder);

35,586,540✔

1097

  return code;

35,586,540✔

1100

int32_t tDecodeSStreamReaderDeployFromTrigger(SDecoder* pDecoder, SStreamReaderDeployFromTrigger* pMsg) {

323,463✔

1101

  int32_t code = 0;

323,463✔

1104

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->triggerTblName, NULL));

646,926✔

1105

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

646,926✔

1106

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

646,926✔

1107

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->triggerTblType));

646,926✔

1108

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->isTriggerTblVirt));

646,926✔

1109

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->deleteReCalc));

646,926✔

1110

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->deleteOutTbl));

646,926✔

1111

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->partitionCols, NULL));

646,926✔

1112

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->triggerCols, NULL));

646,926✔

1113

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->triggerScanPlan, NULL));

646,926✔

1114

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->calcCacheScanPlan, NULL));

646,926✔

1116

_exit:

323,463✔

1118

  return code;

323,463✔

1122

int32_t tDecodeSStreamReaderDeployFromCalc(SDecoder* pDecoder, SStreamReaderDeployFromCalc* pMsg) {

387,148✔

1123

  int32_t code = 0;

387,148✔

1126

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

774,296✔

1127

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->calcScanPlan, NULL));

774,296✔

1129

_exit:

387,148✔

1131

  return code;

387,148✔

1135

int32_t tDecodeSStreamReaderDeployMsg(SDecoder* pDecoder, SStreamReaderDeployMsg* pMsg) {

710,611✔

1136

  int32_t code = 0;

710,611✔

1139

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->triggerReader));

1,421,222✔

1140

  if (pMsg->triggerReader) {

710,611✔

1141

    TAOS_CHECK_EXIT(tDecodeSStreamReaderDeployFromTrigger(pDecoder, &pMsg->msg.trigger));

323,463✔

1143

    TAOS_CHECK_EXIT(tDecodeSStreamReaderDeployFromCalc(pDecoder, &pMsg->msg.calc));

387,148✔

1146

_exit:

387,148✔

1148

  return code;

710,611✔

1152

int32_t tDecodeSStreamTaskAddr(SDecoder* pDecoder, SStreamTaskAddr* pMsg) {

952,419✔

1153

  int32_t code = 0;

952,419✔

1156

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

1,904,838✔

1157

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

1,904,838✔

1158

  TAOS_CHECK_EXIT(tDecodeSEpSet(pDecoder, &pMsg->epset));

952,419✔

1160

_exit:

952,419✔

1162

  return code;

952,419✔

1166

int32_t tDecodeSStreamRunnerTarget(SDecoder* pDecoder, SStreamRunnerTarget* pMsg) {

615,165✔

1167

  int32_t code = 0;

615,165✔

1170

  TAOS_CHECK_EXIT(tDecodeSStreamTaskAddr(pDecoder, &pMsg->addr));

615,165✔

1171

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

1,230,330✔

1173

_exit:

615,165✔

1175

  return code;

615,165✔

1179

int32_t tDecodeSStreamTriggerDeployMsg(SDecoder* pDecoder, SStreamTriggerDeployMsg* pMsg) {

208,485✔

1180

  int32_t code = 0;

208,485✔

1183

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->triggerType));

416,970✔

1184

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->igDisorder));

416,970✔

1185

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->fillHistory));

416,970✔

1186

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->fillHistoryFirst));

416,970✔

1187

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->lowLatencyCalc));

416,970✔

1188

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->igNoDataTrigger));

416,970✔

1189

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->enableMultiGroupCalc));

416,970✔

1190

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->isTriggerTblVirt));

416,970✔

1191

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->triggerHasPF));

416,970✔

1192

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->isTriggerTblStb));

416,970✔

1193

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->partitionCols, NULL));

416,970✔

1195

  int32_t addrSize = 0;

208,485✔

1196

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &addrSize));

208,485✔

1197

  if (addrSize > 0) {

208,485✔

1198

    pMsg->pNotifyAddrUrls = taosArrayInit_s(POINTER_BYTES, addrSize);

57,426✔

1199

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

57,426✔

1201

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

265,911✔

1202

    const char **url = taosArrayGet(pMsg->pNotifyAddrUrls, i);

57,426✔

1203

    TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)url, NULL));

57,426✔

1205

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

416,970✔

1206

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

416,970✔

1207

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->notifyHistory));

416,970✔

1209

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

416,970✔

1210

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

416,970✔

1211

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

416,970✔

1212

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

416,970✔

1213

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

416,970✔

1215

  switch (pMsg->triggerType) {

208,485✔

1216

    case WINDOW_TYPE_SESSION:

8,741✔

1218

      TAOS_CHECK_EXIT(tDecodeI16(pDecoder, &pMsg->trigger.session.slotId));

17,482✔

1219

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pMsg->trigger.session.sessionVal));

17,482✔

1220

      break;

8,741✔

1221

    case WINDOW_TYPE_STATE:

68,378✔

1223

      TAOS_CHECK_EXIT(tDecodeI16(pDecoder, &pMsg->trigger.stateWin.slotId));

136,756✔

1224

      TAOS_CHECK_EXIT(tDecodeI16(pDecoder, &pMsg->trigger.stateWin.extend));

136,756✔

1225

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pMsg->trigger.stateWin.trueForType));

136,756✔

1226

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pMsg->trigger.stateWin.trueForCount));

136,756✔

1227

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pMsg->trigger.stateWin.trueForDuration));

136,756✔

1228

      TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->trigger.stateWin.zeroth, NULL));

136,756✔

1229

      TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->trigger.stateWin.expr, NULL));

136,756✔

1230

      break;

68,378✔

1232

    case WINDOW_TYPE_INTERVAL:

78,282✔

1234

      TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->trigger.sliding.intervalUnit));

156,564✔

1235

      TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->trigger.sliding.slidingUnit));

156,564✔

1236

      TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->trigger.sliding.offsetUnit));

156,564✔

1237

      TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->trigger.sliding.soffsetUnit));

156,564✔

1238

      TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->trigger.sliding.precision));

156,564✔

1239

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pMsg->trigger.sliding.interval));

156,564✔

1240

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pMsg->trigger.sliding.offset));

156,564✔

1241

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pMsg->trigger.sliding.sliding));

156,564✔

1242

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pMsg->trigger.sliding.soffset));

156,564✔

1243

      break;

78,282✔

1245

    case WINDOW_TYPE_EVENT:

25,254✔

1247

      TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->trigger.event.startCond, NULL));

50,508✔

1248

      TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->trigger.event.endCond, NULL));

50,508✔

1249

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pMsg->trigger.event.trueForType));

50,508✔

1250

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pMsg->trigger.event.trueForCount));

50,508✔

1251

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pMsg->trigger.event.trueForDuration));

50,508✔

1252

      break;

25,254✔

1254

    case WINDOW_TYPE_COUNT:

20,171✔

1256

      TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->trigger.count.condCols, NULL));

40,342✔

1258

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pMsg->trigger.count.countVal));

40,342✔

1259

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pMsg->trigger.count.sliding));

40,342✔

1260

      break;

20,171✔

1262

    case WINDOW_TYPE_PERIOD:

7,659✔

1264

      TAOS_CHECK_EXIT(tDecodeI8(pDecoder, (int8_t*)&pMsg->trigger.period.periodUnit));

15,318✔

1265

      TAOS_CHECK_EXIT(tDecodeI8(pDecoder, (int8_t*)&pMsg->trigger.period.offsetUnit));

15,318✔

1266

      TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->trigger.period.precision));

15,318✔

1267

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pMsg->trigger.period.period));

15,318✔

1268

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pMsg->trigger.period.offset));

15,318✔

1269

      break;

7,659✔

1275

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

416,970✔

1276

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

416,970✔

1277

  TAOS_CHECK_EXIT(tDecodeI16(pDecoder, &pMsg->calcTsSlotId));

416,970✔

1278

  TAOS_CHECK_EXIT(tDecodeI16(pDecoder, &pMsg->triTsSlotId));

416,970✔

1279

  TAOS_CHECK_EXIT(tDecodeI16(pDecoder, &pMsg->calcPkSlotId));

416,970✔

1280

  TAOS_CHECK_EXIT(tDecodeI16(pDecoder, &pMsg->triPkSlotId));

416,970✔

1281

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->triggerPrevFilter, NULL));

416,970✔

1282

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->triggerScanPlan, NULL));

416,970✔

1283

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->calcCacheScanPlan, NULL));

416,970✔

1285

  int32_t readerNum = 0;

208,485✔

1286

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &readerNum));

208,485✔

1287

  if (readerNum > 0) {

208,485✔

1288

    pMsg->readerList = taosArrayInit_s(sizeof(SStreamTaskAddr), readerNum);

207,439✔

1289

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

207,439✔

1291

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

467,021✔

1292

    SStreamTaskAddr* pAddr = (SStreamTaskAddr*)taosArrayGet(pMsg->readerList, i);

258,536✔

1293

    TAOS_CHECK_EXIT(tDecodeSStreamTaskAddr(pDecoder, pAddr));

258,536✔

1296

  int32_t runnerNum = 0;

208,485✔

1297

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &runnerNum));

208,485✔

1298

  if (runnerNum > 0) {

208,485✔

1299

    pMsg->runnerList = taosArrayInit_s(sizeof(SStreamRunnerTarget), runnerNum);

205,055✔

1300

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

205,055✔

1302

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

823,650✔

1303

    SStreamRunnerTarget* pTarget = (SStreamRunnerTarget*)taosArrayGet(pMsg->runnerList, i);

615,165✔

1304

    TAOS_CHECK_EXIT(tDecodeSStreamRunnerTarget(pDecoder, pTarget));

615,165✔

1307

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

416,970✔

1308

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->streamName, NULL));

416,970✔

1309

  if (!tDecodeIsEnd(pDecoder)) {

208,485✔

1310

    TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->precision));

416,970✔

1312

  if (!tDecodeIsEnd(pDecoder)) {

208,485✔

1313

    TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->nodelayCreateSubtable));

416,970✔

1316

_exit:

208,485✔

1318

  return code;

208,485✔

1323

int32_t tDeserializeSFieldWithOptions(SDecoder *pDecoder, SFieldWithOptions *pField) {

3,583,924✔

1324

  int32_t code = 0;

3,583,924✔

1327

  TAOS_CHECK_EXIT(tDecodeCStrTo(pDecoder, pField->name));

3,583,924✔

1328

  TAOS_CHECK_EXIT(tDecodeU8(pDecoder, &pField->type));

7,167,848✔

1329

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pField->flags));

7,167,848✔

1330

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pField->bytes));

7,167,848✔

1331

  TAOS_CHECK_EXIT(tDecodeU32(pDecoder, &pField->compress));

7,167,848✔

1332

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pField->typeMod));

7,167,848✔

1334

_exit:

3,583,924✔

1336

  return code;

3,583,924✔

1339

void destroySStreamOutCols(void* p){

70,251✔

1340

  if (p == NULL) return;

70,251✔

1341

  SStreamOutCol* col = (SStreamOutCol*)p;

70,251✔

1342

  taosMemoryFreeClear(col->expr);

70,251✔

1345

int32_t tDecodeSStreamRunnerDeployMsg(SDecoder* pDecoder, SStreamRunnerDeployMsg* pMsg) {

729,026✔

1346

  int32_t code = 0;

729,026✔

1349

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

1,458,052✔

1350

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->streamName, NULL));

1,458,052✔

1351

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->pPlan, NULL));

1,458,052✔

1352

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->outDBFName, NULL));

1,458,052✔

1353

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->outTblName, NULL));

1,458,052✔

1354

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->outTblType));

1,458,052✔

1355

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->calcNotifyOnly));

1,458,052✔

1356

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->topPlan));

1,458,052✔

1358

  int32_t addrSize = 0;

729,026✔

1359

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &addrSize));

729,026✔

1360

  if (addrSize > 0) {

729,026✔

1361

    pMsg->pNotifyAddrUrls = taosArrayInit_s(POINTER_BYTES, addrSize);

161,988✔

1362

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

161,988✔

1364

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

891,014✔

1365

    const char **url = taosArrayGet(pMsg->pNotifyAddrUrls, i);

161,988✔

1366

    TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)url, NULL));

161,988✔

1368

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

1,458,052✔

1370

  int32_t outColNum = 0;

729,026✔

1371

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &outColNum));

729,026✔

1372

  if (outColNum > 0) {

729,026✔

1373

    pMsg->outCols = taosArrayInit_s(sizeof(SFieldWithOptions), outColNum);

729,026✔

1374

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

729,026✔

1376

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

3,708,753✔

1377

    SFieldWithOptions *pCol = taosArrayGet(pMsg->outCols, i);

2,979,727✔

1378

    TAOS_CHECK_EXIT(tDeserializeSFieldWithOptions(pDecoder, pCol));

2,979,727✔

1381

  int32_t outTagNum = 0;

729,026✔

1382

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &outTagNum));

729,026✔

1383

  if (outTagNum > 0) {

729,026✔

1384

    pMsg->outTags = taosArrayInit_s(sizeof(SFieldWithOptions), outTagNum);

316,479✔

1385

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

316,479✔

1387

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

1,333,223✔

1388

    SFieldWithOptions *pTag = taosArrayGet(pMsg->outTags, i);

604,197✔

1389

    TAOS_CHECK_EXIT(tDeserializeSFieldWithOptions(pDecoder, pTag));

604,197✔

1392

  TAOS_CHECK_EXIT(tDecodeU64(pDecoder, &pMsg->outStbUid));

1,458,052✔

1393

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

1,458,052✔

1395

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->subTblNameExpr, NULL));

1,458,052✔

1396

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pMsg->tagValueExpr, NULL));

1,458,052✔

1398

  int32_t forceOutColsSize = 0;

729,026✔

1399

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &forceOutColsSize));

729,026✔

1400

  if (forceOutColsSize > 0) {

729,026✔

1401

    pMsg->forceOutCols = taosArrayInit_s(sizeof(SStreamOutCol), forceOutColsSize);

12,720✔

1402

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

12,720✔

1404

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

799,277✔

1405

    SStreamOutCol *pCoutCol = (SStreamOutCol*)taosArrayGet(pMsg->forceOutCols, i);

70,251✔

1407

    TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pCoutCol->expr, NULL));

140,502✔

1408

    TAOS_CHECK_EXIT(tDecodeU8(pDecoder, &pCoutCol->type.type));

140,502✔

1409

    TAOS_CHECK_EXIT(tDecodeU8(pDecoder, &pCoutCol->type.precision));

140,502✔

1410

    TAOS_CHECK_EXIT(tDecodeU8(pDecoder, &pCoutCol->type.scale));

140,502✔

1411

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pCoutCol->type.bytes));

140,502✔

1414

  if (!tDecodeIsEnd(pDecoder)) {

729,026✔

1415

    TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pMsg->lowLatencyCalc));

1,458,052✔

1420

  if (!tDecodeIsEnd(pDecoder)) {

729,026✔

1421

    int32_t colCidsSize = 0;

729,026✔

1422

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &colCidsSize));

729,026✔

1423

    if (colCidsSize > 0 && colCidsSize <= TSDB_MAX_COLUMNS) {  // Sanity check

729,026✔

1424

      pMsg->colCids = taosArrayInit(colCidsSize, sizeof(int16_t));

14,253✔

1425

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

14,253✔

1426

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

57,960✔

1427

        int16_t cid = 0;

43,707✔

1428

        TAOS_CHECK_EXIT(tDecodeI16(pDecoder, &cid));

43,707✔

1429

        if (taosArrayPush(pMsg->colCids, &cid) == NULL) {

87,414✔

1436

  if (!tDecodeIsEnd(pDecoder)) {

729,026✔

1437

    int32_t tagCidsSize = 0;

729,026✔

1438

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &tagCidsSize));

729,026✔

1439

    if (tagCidsSize > 0 && tagCidsSize <= TSDB_MAX_TAGS) {  // Sanity check

729,026✔

1440

      pMsg->tagCids = taosArrayInit(tagCidsSize, sizeof(int16_t));

10,842✔

1441

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

10,842✔

1442

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

26,181✔

1443

        int16_t cid = 0;

15,339✔

1444

        TAOS_CHECK_EXIT(tDecodeI16(pDecoder, &cid));

15,339✔

1445

        if (taosArrayPush(pMsg->tagCids, &cid) == NULL) {

30,678✔

1452

_exit:

727,346✔

1454

  return code;

729,026✔

1457

int32_t tDecodeSStmTaskDeploy(SDecoder* pDecoder, SStmTaskDeploy* pTask) {

1,648,122✔

1458

  int32_t code = 0;

1,648,122✔

1461

  TAOS_CHECK_EXIT(tDecodeStreamTask(pDecoder, (SStreamTask*)&pTask->task));

1,648,122✔

1462

  switch (pTask->task.type) {

1,648,122✔

1463

    case STREAM_READER_TASK:

710,611✔

1464

      TAOS_CHECK_EXIT(tDecodeSStreamReaderDeployMsg(pDecoder, &pTask->msg.reader));

710,611✔

1465

      break;

710,611✔

1466

    case STREAM_TRIGGER_TASK:

208,485✔

1467

      TAOS_CHECK_EXIT(tDecodeSStreamTriggerDeployMsg(pDecoder, &pTask->msg.trigger));

208,485✔

1468

      break;

208,485✔

1469

    case STREAM_RUNNER_TASK:

729,026✔

1470

      TAOS_CHECK_EXIT(tDecodeSStreamRunnerDeployMsg(pDecoder, &pTask->msg.runner));

729,026✔

1471

      break;

729,026✔

1477

_exit:

1,648,122✔

1479

  return code;

1,648,122✔

1483

int32_t tDecodeSStmStreamDeploy(SDecoder* pDecoder, SStmStreamDeploy* pStream) {

281,739✔

1484

  int32_t code = 0;

281,739✔

1487

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

563,478✔

1489

  int32_t readerNum = 0;

281,739✔

1490

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &readerNum));

281,739✔

1491

  if (readerNum > 0) {

281,739✔

1492

    pStream->readerTasks = taosArrayInit_s(sizeof(SStmTaskDeploy), readerNum);

270,195✔

1493

    TSDB_CHECK_NULL(pStream->readerTasks, code, lino, _exit, terrno);

270,195✔

1495

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

992,350✔

1496

    SStmTaskDeploy* pTask = taosArrayGet(pStream->readerTasks, i);

710,611✔

1497

    TAOS_CHECK_EXIT(tDecodeSStmTaskDeploy(pDecoder, pTask));

710,611✔

1500

  int32_t triggerTask = 0;

281,739✔

1501

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &triggerTask));

281,739✔

1502

  if (triggerTask) {

281,739✔

1503

    pStream->triggerTask = taosMemoryCalloc(1, sizeof(SStmTaskDeploy));

208,485✔

1504

    TSDB_CHECK_NULL(pStream->triggerTask, code, lino, _exit, terrno);

208,485✔

1505

    TAOS_CHECK_EXIT(tDecodeSStmTaskDeploy(pDecoder, pStream->triggerTask));

208,485✔

1508

  int32_t runnerNum = 0;

281,739✔

1509

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &runnerNum));

281,739✔

1510

  if (runnerNum > 0) {

281,739✔

1511

    pStream->runnerTasks = taosArrayInit_s(sizeof(SStmTaskDeploy), runnerNum);

213,788✔

1512

    TSDB_CHECK_NULL(pStream->runnerTasks, code, lino, _exit, terrno);

213,788✔

1514

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

1,010,765✔

1515

    SStmTaskDeploy* pTask = taosArrayGet(pStream->runnerTasks, i);

729,026✔

1516

    TAOS_CHECK_EXIT(tDecodeSStmTaskDeploy(pDecoder, pTask));

729,026✔

1519

_exit:

281,739✔

1521

  return code;

281,739✔

1525

int32_t tDecodeSStreamStartTaskMsg(SDecoder* pDecoder, SStreamStartTaskMsg* pStart) {

204,745✔

1526

  int32_t code = 0;

204,745✔

1529

  TAOS_CHECK_EXIT(tDecodeSStreamMsg(pDecoder, &pStart->header));

204,745✔

1531

_exit:

204,745✔

1533

  return code;

204,745✔

1537

int32_t tDecodeSStreamTaskStart(SDecoder* pDecoder, SStreamTaskStart* pTask) {

204,745✔

1538

  int32_t code = 0;

204,745✔

1541

  TAOS_CHECK_EXIT(tDecodeStreamTask(pDecoder, (SStreamTask*)&pTask->task));

204,745✔

1542

  TAOS_CHECK_EXIT(tDecodeSStreamStartTaskMsg(pDecoder, (SStreamStartTaskMsg*)&pTask->startMsg));

204,745✔

1544

_exit:

204,745✔

1546

  return code;

204,745✔

1550

int32_t tDecodeSStreamUndeployTaskMsg(SDecoder* pDecoder, SStreamUndeployTaskMsg* pUndeploy) {

264,776✔

1551

  int32_t code = 0;

264,776✔

1554

  TAOS_CHECK_EXIT(tDecodeSStreamMsg(pDecoder, &pUndeploy->header));

264,776✔

1555

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pUndeploy->doCheckpoint));

529,552✔

1556

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pUndeploy->doCleanup));

529,552✔

1558

_exit:

264,776✔

1560

  return code;

264,776✔

1564

int32_t tDecodeSStreamTaskUndeploy(SDecoder* pDecoder, SStreamTaskUndeploy* pTask) {

264,776✔

1565

  int32_t code = 0;

264,776✔

1568

  TAOS_CHECK_EXIT(tDecodeStreamTask(pDecoder, (SStreamTask*)&pTask->task));

264,776✔

1569

  TAOS_CHECK_EXIT(tDecodeSStreamUndeployTaskMsg(pDecoder, (SStreamUndeployTaskMsg*)&pTask->undeployMsg));

264,776✔

1571

_exit:

264,776✔

1573

  return code;

264,776✔

1576

int32_t tDecodeSStreamRecalcReq(SDecoder* pDecoder, SStreamRecalcReq* recalc) {

3,672✔

1577

  int32_t code = 0;

3,672✔

1580

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &recalc->recalcId));

7,344✔

1581

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &recalc->start));

7,344✔

1582

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &recalc->end));

7,344✔

1584

_exit:

3,672✔

1586

  return code;

3,672✔

1589

int32_t tDecodeSStreamMgmtRspCont(SDecoder* pDecoder, SStreamMsgType msgType, SStreamMgmtRspCont* pCont) {

72,527✔

1590

  int32_t code = 0;

72,527✔

1593

  switch (msgType) {

72,527✔

1594

    case STREAM_MSG_ORIGTBL_READER_INFO: {

66,204✔

1595

      int32_t vgNum = 0;

66,204✔

1596

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgNum));

66,204✔

1597

      if (vgNum > 0) {

66,204✔

1598

        pCont->vgIds = taosArrayInit_s(sizeof(int32_t), vgNum);

66,204✔

1599

        TSDB_CHECK_NULL(pCont->vgIds, code, lino, _exit, terrno);

66,204✔

1601

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

260,316✔

1602

        int32_t *vgId = taosArrayGet(pCont->vgIds, i);

194,112✔

1603

        TAOS_CHECK_EXIT(tDecodeI32(pDecoder, vgId));

194,112✔

1606

      int32_t readerNum = 0;

66,204✔

1607

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &readerNum));

66,204✔

1608

      if (readerNum > 0) {

66,204✔

1609

        pCont->readerList = taosArrayInit_s(sizeof(SStreamTaskAddr), readerNum);

55,534✔

1610

        TSDB_CHECK_NULL(pCont->readerList, code, lino, _exit, terrno);

55,534✔

1612

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

130,167✔

1613

        SStreamTaskAddr *addr = taosArrayGet(pCont->readerList, i);

63,963✔

1614

        TAOS_CHECK_EXIT(tDecodeSStreamTaskAddr(pDecoder, addr));

63,963✔

1616

      break;

66,204✔

1631

    case STREAM_MSG_USER_RECALC: {

3,672✔

1632

      int32_t recalcNum = 0;

3,672✔

1633

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &recalcNum));

3,672✔

1634

      if (recalcNum > 0) {

3,672✔

1635

        pCont->recalcList = taosArrayInit_s(sizeof(SStreamRecalcReq), recalcNum);

3,672✔

1636

        TSDB_CHECK_NULL(pCont->recalcList, code, lino, _exit, terrno);

3,672✔

1638

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

7,344✔

1639

        SStreamRecalcReq *recalc = taosArrayGet(pCont->recalcList, i);

3,672✔

1640

        TAOS_CHECK_EXIT(tDecodeSStreamRecalcReq(pDecoder, recalc));

3,672✔

1642

      break;

3,672✔

1644

    case STREAM_MSG_RUNNER_ORIGTBL_READER: {

2,651✔

1645

      int32_t rspNum = 0, vgNum = 0;

2,651✔

1646

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &rspNum));

2,651✔

1647

      if (rspNum > 0) {

2,651✔

1648

        pCont->execRspList = taosArrayInit_s(sizeof(SStreamOReaderDeployRsp), rspNum);

2,651✔

1649

        TSDB_CHECK_NULL(pCont->execRspList, code, lino, _exit, terrno);

2,651✔

1651

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

5,302✔

1652

        SStreamOReaderDeployRsp *pDeployRsp = taosArrayGet(pCont->execRspList, i);

2,651✔

1653

        TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pDeployRsp->execId));

5,302✔

1654

        TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgNum));

2,651✔

1655

        if (vgNum > 0) {

2,651✔

1656

          pDeployRsp->vgList = taosArrayInit_s(sizeof(SStreamTaskAddr), vgNum);

2,651✔

1657

          TSDB_CHECK_NULL(pDeployRsp->vgList, code, lino, _exit, terrno);

2,651✔

1659

        for (int32_t n = 0; n < vgNum; ++n) {

5,302✔

1660

          SStreamTaskAddr* pAddr = taosArrayGet(pDeployRsp->vgList, n);

2,651✔

1661

          TAOS_CHECK_EXIT(tDecodeSStreamTaskAddr(pDecoder, pAddr));

2,651✔

1664

      break;

2,651✔

1670

_exit:

72,527✔

1672

  return code;

72,527✔

1676

int32_t tDecodeSStreamMgmtRsp(SDecoder* pDecoder, SStreamMgmtRsp* pRsp) {

72,527✔

1677

  int32_t code = 0;

72,527✔

1680

  TAOS_CHECK_EXIT(tDecodeSStreamMsg(pDecoder, &pRsp->header));

72,527✔

1681

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

145,054✔

1682

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

145,054✔

1683

  TAOS_CHECK_EXIT(tDecodeStreamTask(pDecoder, &pRsp->task));

72,527✔

1684

  TAOS_CHECK_EXIT(tDecodeSStreamMgmtRspCont(pDecoder, pRsp->header.msgType, &pRsp->cont));

72,527✔

1686

_exit:

72,527✔

1688

  return code;

72,527✔

1691

void tFreeSStreamOReaderDeployRsp(void* param) {

5,302✔

1692

  if (NULL == param) {

5,302✔

1696

  SStreamOReaderDeployRsp* pRsp = (SStreamOReaderDeployRsp*)param;

5,302✔

1697

  taosArrayDestroy(pRsp->vgList);

5,302✔

1700

void tFreeSStreamMgmtRsp(void* param) {

145,054✔

1701

  if (NULL == param) {

145,054✔

1705

  SStreamMgmtRsp* pRsp = (SStreamMgmtRsp*)param;

145,054✔

1707

  taosArrayDestroy(pRsp->cont.vgIds);

145,054✔

1708

  taosArrayDestroy(pRsp->cont.readerList);

145,054✔

1709

  taosArrayDestroy(pRsp->cont.runnerList);

145,054✔

1710

  taosArrayDestroy(pRsp->cont.recalcList);

145,054✔

1711

  taosArrayDestroyEx(pRsp->cont.execRspList, tFreeSStreamOReaderDeployRsp);

145,054✔

1714

void tFreeSStreamReaderDeployMsg(SStreamReaderDeployMsg* pReader) {

710,611✔

1715

  if (NULL == pReader) {

710,611✔

1719

  if (pReader->triggerReader) {

710,611✔

1720

    SStreamReaderDeployFromTrigger* pMsg = (SStreamReaderDeployFromTrigger*)&pReader->msg.trigger;

323,463✔

1721

    taosMemoryFree(pMsg->triggerTblName);

323,463✔

1722

    taosMemoryFree(pMsg->partitionCols);

323,463✔

1723

    taosMemoryFree(pMsg->triggerCols);

323,463✔

1724

    taosMemoryFree(pMsg->triggerScanPlan);

323,463✔

1725

    taosMemoryFree(pMsg->calcCacheScanPlan);

323,463✔

1727

    SStreamReaderDeployFromCalc* pMsg = (SStreamReaderDeployFromCalc*)&pReader->msg.calc;

387,148✔

1728

    taosMemoryFree(pMsg->calcScanPlan);

387,148✔

1740

void tFreeSStreamTriggerDeployMsg(SStreamTriggerDeployMsg* pTrigger) {

208,485✔

1741

  if (NULL == pTrigger) {

208,485✔

1745

  taosArrayDestroyEx(pTrigger->pNotifyAddrUrls, tFreeStreamNotifyUrl);

208,485✔

1746

  switch (pTrigger->triggerType) {

208,485✔

1747

    case WINDOW_TYPE_STATE:

68,378✔

1748

      taosMemoryFree(pTrigger->trigger.stateWin.zeroth);

68,378✔

1749

      taosMemoryFree(pTrigger->trigger.stateWin.expr);

68,378✔

1750

      break;

68,378✔

1751

    case WINDOW_TYPE_EVENT:

25,254✔

1752

      taosMemoryFree(pTrigger->trigger.event.startCond);

25,254✔

1753

      taosMemoryFree(pTrigger->trigger.event.endCond);

25,254✔

1754

      break;

25,254✔

1755

    case WINDOW_TYPE_COUNT:

20,171✔

1756

      taosMemoryFree(pTrigger->trigger.count.condCols);

20,171✔

1757

      break;

20,171✔

1758

    default:

94,682✔

1759

      break;

94,682✔

1762

  taosMemoryFree(pTrigger->partitionCols);

208,485✔

1763

  taosMemoryFree(pTrigger->triggerPrevFilter);

208,485✔

1764

  taosMemoryFree(pTrigger->triggerScanPlan);

208,485✔

1765

  taosMemoryFree(pTrigger->calcCacheScanPlan);

208,485✔

1767

  taosArrayDestroy(pTrigger->readerList);

208,485✔

1768

  taosArrayDestroy(pTrigger->runnerList);

208,485✔

1769

  taosMemoryFree(pTrigger->streamName);

208,485✔

1781

void tFreeSStreamRunnerDeployMsg(SStreamRunnerDeployMsg* pRunner) {

729,026✔

1782

  if (NULL == pRunner) {

729,026✔

1786

  taosMemoryFree(pRunner->streamName);

729,026✔

1787

  taosMemoryFree(pRunner->pPlan);

729,026✔

1788

  taosMemoryFree(pRunner->outDBFName);

729,026✔

1789

  taosMemoryFree(pRunner->outTblName);

729,026✔

1791

  taosArrayDestroyEx(pRunner->pNotifyAddrUrls, tFreeStreamNotifyUrl);

729,026✔

1792

  taosArrayDestroy(pRunner->outCols);

729,026✔

1793

  taosArrayDestroy(pRunner->outTags);

729,026✔

1795

  taosMemoryFree(pRunner->subTblNameExpr);

729,026✔

1796

  taosMemoryFree(pRunner->tagValueExpr);

729,026✔

1797

  taosArrayDestroyEx(pRunner->forceOutCols, tFreeSStreamOutCol);

729,026✔

1800

void tFreeSStmTaskDeploy(void* param) {

2,002,897✔

1801

  if (NULL == param) {

2,002,897✔

1802

    return;

354,775✔

1805

  SStmTaskDeploy* pTask = (SStmTaskDeploy*)param;

1,648,122✔

1806

  switch (pTask->task.type)  {

1,648,122✔

1807

    case STREAM_READER_TASK:

710,611✔

1808

      tFreeSStreamReaderDeployMsg(&pTask->msg.reader);

710,611✔

1809

      break;

710,611✔

1810

    case STREAM_TRIGGER_TASK:

208,485✔

1811

      tFreeSStreamTriggerDeployMsg(&pTask->msg.trigger);

208,485✔

1812

      break;

208,485✔

1813

    case STREAM_RUNNER_TASK:

729,026✔

1814

      tFreeSStreamRunnerDeployMsg(&pTask->msg.runner);

729,026✔

1815

      break;

729,026✔

1822

void tFreeSStmStreamDeploy(void* param) {

281,521✔

1823

  if (NULL == param) {

281,521✔

1827

  SStmStreamDeploy* pDeploy = (SStmStreamDeploy*)param;

281,521✔

1828

  int32_t readerNum = taosArrayGetSize(pDeploy->readerTasks);

281,521✔

1829

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

992,018✔

1830

    SStmTaskDeploy* pReader = taosArrayGet(pDeploy->readerTasks, i);

710,497✔

1831

    if (!pReader->msg.reader.triggerReader && pReader->msg.reader.msg.calc.freeScanPlan) {

710,497✔

1832

      taosMemoryFreeClear(pReader->msg.reader.msg.calc.calcScanPlan);

383,873✔

1835

  taosArrayDestroy(pDeploy->readerTasks);

281,521✔

1837

  if (pDeploy->triggerTask) {

281,521✔

1838

    taosArrayDestroy(pDeploy->triggerTask->msg.trigger.readerList);

208,377✔

1839

    taosArrayDestroy(pDeploy->triggerTask->msg.trigger.runnerList);

208,377✔

1840

    taosMemoryFree(pDeploy->triggerTask);

208,377✔

1843

  int32_t runnerNum = taosArrayGetSize(pDeploy->runnerTasks);

281,521✔

1844

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

1,010,196✔

1845

    SStmTaskDeploy* pRunner = taosArrayGet(pDeploy->runnerTasks, i);

728,675✔

1846

    taosMemoryFree(pRunner->msg.runner.pPlan);

728,675✔

1848

  taosArrayDestroy(pDeploy->runnerTasks);

281,521✔

1851

void tDeepFreeSStmStreamDeploy(void* param) {

563,260✔

1852

  if (NULL == param) {

563,260✔

1856

  SStmStreamDeploy* pDeploy = (SStmStreamDeploy*)param;

563,260✔

1857

  taosArrayDestroyEx(pDeploy->readerTasks, tFreeSStmTaskDeploy);

563,260✔

1858

  tFreeSStmTaskDeploy(pDeploy->triggerTask);

563,260✔

1859

  taosMemoryFree(pDeploy->triggerTask);

563,260✔

1860

  taosArrayDestroyEx(pDeploy->runnerTasks, tFreeSStmTaskDeploy);

563,260✔

1864

void tFreeSMStreamHbRspMsg(SMStreamHbRspMsg* pRsp) {

35,905,804✔

1865

  if (NULL == pRsp) {

35,905,804✔

1868

  taosArrayDestroyEx(pRsp->deploy.streamList, tFreeSStmStreamDeploy);

35,905,804✔

1869

  taosArrayDestroy(pRsp->start.taskList);

35,905,804✔

1870

  taosArrayDestroy(pRsp->undeploy.taskList);

35,905,804✔

1871

  taosArrayDestroyEx(pRsp->rsps.rspList, tFreeSStreamMgmtRsp);

35,905,804✔

1874

void tDeepFreeSMStreamHbRspMsg(SMStreamHbRspMsg* pRsp) {

17,737,394✔

1875

  if (NULL == pRsp) {

17,737,394✔

1878

  taosArrayDestroyEx(pRsp->deploy.streamList, tDeepFreeSStmStreamDeploy);

17,737,394✔

1879

  taosArrayDestroy(pRsp->start.taskList);

17,737,394✔

1880

  taosArrayDestroy(pRsp->undeploy.taskList);

17,737,394✔

1881

  taosArrayDestroyEx(pRsp->rsps.rspList, tFreeSStreamMgmtRsp);

17,737,394✔

1886

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

17,737,394✔

1887

  int32_t code = 0;

17,737,394✔

1890

  TAOS_CHECK_EXIT(tStartDecode(pDecoder));

17,737,394✔

1891

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

35,474,788✔

1892

  int32_t deployNum = 0;

17,737,394✔

1893

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &deployNum));

17,737,394✔

1894

  if (deployNum > 0) {

17,737,394✔

1895

    pRsp->deploy.streamList = taosArrayInit_s(sizeof(SStmStreamDeploy), deployNum);

74,818✔

1896

    TSDB_CHECK_NULL(pRsp->deploy.streamList, code, lino, _exit, terrno);

74,818✔

1898

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

18,019,133✔

1899

    SStmStreamDeploy* pStream = taosArrayGet(pRsp->deploy.streamList, i);

281,739✔

1900

    TAOS_CHECK_EXIT(tDecodeSStmStreamDeploy(pDecoder, pStream));

281,739✔

1903

  int32_t startNum = 0;

17,737,394✔

1904

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &startNum));

17,737,394✔

1905

  if (startNum > 0) {

17,737,394✔

1906

    pRsp->start.taskList = taosArrayInit_s(sizeof(SStreamTaskStart), startNum);

101,089✔

1907

    TSDB_CHECK_NULL(pRsp->start.taskList, code, lino, _exit, terrno);

101,089✔

1909

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

17,942,139✔

1910

    SStreamTaskStart* pTask = (SStreamTaskStart*)taosArrayGet(pRsp->start.taskList, i);

204,745✔

1911

    TAOS_CHECK_EXIT(tDecodeSStreamTaskStart(pDecoder, pTask));

204,745✔

1914

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pRsp->undeploy.undeployAll));

35,474,788✔

1915

  if (!pRsp->undeploy.undeployAll) {

17,737,394✔

1916

    int32_t undeployNum = 0;

17,737,394✔

1917

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &undeployNum));

17,737,394✔

1918

    if (undeployNum > 0) {

17,737,394✔

1919

      pRsp->undeploy.taskList = taosArrayInit_s(sizeof(SStreamTaskUndeploy), undeployNum);

38,885✔

1920

      TSDB_CHECK_NULL(pRsp->undeploy.taskList, code, lino, _exit, terrno);

38,885✔

1922

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

18,002,170✔

1923

      SStreamTaskUndeploy* pTask = (SStreamTaskUndeploy*)taosArrayGet(pRsp->undeploy.taskList, i);

264,776✔

1924

      TAOS_CHECK_EXIT(tDecodeSStreamTaskUndeploy(pDecoder, pTask));

264,776✔

1928

  int32_t rspNum = 0;

17,737,394✔

1929

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &rspNum));

17,737,394✔

1930

  if (rspNum > 0) {

17,737,394✔

1931

    pRsp->rsps.rspList = taosArrayInit_s(sizeof(SStreamMgmtRsp), rspNum);

22,166✔

1932

    TSDB_CHECK_NULL(pRsp->rsps.rspList, code, lino, _exit, terrno);

22,166✔

1933

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

94,693✔

1934

      SStreamMgmtRsp* pMgmtRsp = (SStreamMgmtRsp*)taosArrayGet(pRsp->rsps.rspList, i);

72,527✔

1935

      TAOS_CHECK_EXIT(tDecodeSStreamMgmtRsp(pDecoder, pMgmtRsp));

72,527✔

1939

  tEndDecode(pDecoder);

17,737,394✔

1941

_exit:

17,737,394✔

1942

  return code;

17,737,394✔

1999

int32_t tSerializeSCMCreateStreamReqImpl(SEncoder* pEncoder, const SCMCreateStreamReq *pReq) {

1,613,240✔

2000

  int32_t code = TSDB_CODE_SUCCESS;

1,613,240✔

2001

  int32_t lino = 0;

1,613,240✔

2003

  char*   json = NULL;

1,613,240✔

2004

  int32_t jsonLen = 0;

1,613,240✔

2005

  TAOS_CHECK_EXIT(scmCreateStreamReqToJson(pReq, false, &json, &jsonLen));

1,613,240✔

2006

  TAOS_CHECK_EXIT(tEncodeCStrWithLen(pEncoder, json, jsonLen));

3,226,480✔

2008

_exit:

1,613,240✔

2009

  taosMemoryFreeClear(json);

1,613,240✔

2010

  if (code) {

1,613,240✔

2014

  return 0;

1,613,240✔

2017

int32_t tSerializeSCMCreateStreamReq(void *buf, int32_t bufLen, const SCMCreateStreamReq *pReq) {

766,828✔

2018

  SEncoder encoder = {0};

766,828✔

2019

  tEncoderInit(&encoder, buf, bufLen);

766,828✔

2020

  int32_t code = 0;

766,828✔

2023

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

766,828✔

2025

  TAOS_CHECK_EXIT(tSerializeSCMCreateStreamReqImpl(&encoder, pReq));

766,828✔

2027

  tEndEncode(&encoder);

766,828✔

2029

_exit:

766,828✔

2030

  if (code) {

766,828✔

2034

    int32_t tlen = encoder.pos;

766,828✔

2035

    tEncoderClear(&encoder);

766,828✔

2036

    return tlen;

766,828✔

2334

int32_t tDeserializeSCMCreateStreamReqImpl(SDecoder *pDecoder, SCMCreateStreamReq *pReq) {

492,782✔

2335

  int32_t code = 0;

492,782✔

2338

  char* json = NULL;

492,782✔

2339

  SJson* pJson = NULL;

492,782✔

2340

  TAOS_CHECK_EXIT(tDecodeCStrAlloc(pDecoder, &json));

492,782✔

2341

  pJson = tjsonParse(json);

492,782✔

2342

  if (pJson == NULL) {

492,782✔

2345

  TAOS_CHECK_EXIT(jsonToSCMCreateStreamReq(pJson, pReq));

492,782✔

2347

_exit:

492,782✔

2348

  taosMemoryFreeClear(json);

492,782✔

2349

  if (NULL != pJson) {

492,782✔

2350

    tjsonDelete(pJson);

492,782✔

2353

  return code;

492,782✔

2357

int32_t tDeserializeSCMCreateStreamReq(void *buf, int32_t bufLen, SCMCreateStreamReq *pReq) {

241,980✔

2358

  SDecoder decoder = {0};

241,980✔

2359

  tDecoderInit(&decoder, buf, bufLen);

241,980✔

2360

  int32_t code = 0;

241,980✔

2363

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

241,980✔

2365

  code = tDeserializeSCMCreateStreamReqImpl(&decoder, pReq);

241,980✔

2366

  if (TSDB_CODE_MND_STREAM_INVALID_JSON == code) {

241,980✔

2375

  tEndDecode(&decoder);

241,980✔

2377

_exit:

241,980✔

2379

  tDecoderClear(&decoder);

241,980✔

2380

  return code;

241,980✔

2384

int32_t tSerializeSMDropStreamReq(void *buf, int32_t bufLen, const SMDropStreamReq *pReq) {

55,248✔

2385

  int32_t  code = 0;

55,248✔

2388

  SEncoder encoder = {0};

55,248✔

2389

  tEncoderInit(&encoder, buf, bufLen);

55,248✔

2391

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

55,248✔

2393

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, pReq->count));

110,496✔

2394

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

117,216✔

2395

    int32_t nameLen = pReq->name[i] == NULL ? 0 : (int32_t)strlen(pReq->name[i]) + 1;

61,968✔

2396

    TAOS_CHECK_EXIT(tEncodeBinary(&encoder, pReq->name[i], nameLen));

123,936✔

2398

  TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->igNotExists));

110,496✔

2400

  tEndEncode(&encoder);

55,248✔

2402

_exit:

55,248✔

2403

  if (code) {

55,248✔

2406

    tlen = encoder.pos;

55,248✔

2408

  tEncoderClear(&encoder);

55,248✔

2409

  return tlen;

55,248✔

2412

int32_t tDeserializeSMDropStreamReq(void *buf, int32_t bufLen, SMDropStreamReq *pReq) {

21,691✔

2413

  SDecoder decoder = {0};

21,691✔

2414

  int32_t  code = 0;

21,691✔

2416

  tDecoderInit(&decoder, buf, bufLen);

21,691✔

2418

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

21,691✔

2419

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pReq->count));

43,382✔

2420

  if (pReq->count > 0) {

21,691✔

2421

    pReq->name = taosMemoryCalloc(pReq->count, sizeof(char*));

21,691✔

2422

    if (pReq->name == NULL) {

21,691✔

2426

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

46,742✔

2427

      TAOS_CHECK_EXIT(tDecodeBinaryAlloc(&decoder, (void**)&pReq->name[i], NULL));

50,102✔

2430

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->igNotExists));

43,382✔

2432

  tEndDecode(&decoder);

21,691✔

2434

_exit:

21,691✔

2435

  tDecoderClear(&decoder);

21,691✔

2436

  return code;

21,691✔

2439

void tFreeMDropStreamReq(SMDropStreamReq *pReq) {

50,131✔

2440

  if (NULL == pReq) {

50,131✔

2443

  if (pReq->name) {

50,131✔

2444

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

106,982✔

2445

      taosMemoryFreeClear(pReq->name[i]);

56,851✔

2447

    taosMemoryFreeClear(pReq->name);

50,131✔

2451

static FORCE_INLINE void tFreeStreamCalcScan(void* pScan) {

2,388,988✔

2452

  if (pScan == NULL) {

2,388,988✔

2455

  SStreamCalcScan *pCalcScan = (SStreamCalcScan *)pScan;

2,388,988✔

2456

  taosArrayDestroy(pCalcScan->vgList);

2,388,988✔

2457

  taosMemoryFreeClear(pCalcScan->scanPlan);

2,388,988✔

2460

void tFreeStreamOutCol(void* pCol) {

80,160✔

2461

  if (pCol == NULL) {

80,160✔

2464

  SStreamOutCol *pOutCol = (SStreamOutCol *)pCol;

80,160✔

2465

  taosMemoryFreeClear(pOutCol->expr);

80,160✔

2470

void tFreeSCMCreateStreamReq(SCMCreateStreamReq *pReq) {

1,398,454✔

2471

  if (NULL == pReq) {

1,398,454✔

2472

    return;

215,580✔

2474

  taosMemoryFreeClear(pReq->name);

1,182,874✔

2475

  taosMemoryFreeClear(pReq->sql);

1,182,874✔

2476

  taosMemoryFreeClear(pReq->streamDB);

1,182,874✔

2477

  taosMemoryFreeClear(pReq->triggerDB);

1,182,874✔

2478

  taosMemoryFreeClear(pReq->outDB);

1,182,874✔

2479

  taosMemoryFreeClear(pReq->triggerTblName);

1,182,874✔

2480

  taosMemoryFreeClear(pReq->outTblName);

1,182,874✔

2482

  taosArrayDestroyP(pReq->calcDB, NULL);

1,182,874✔

2483

  pReq->calcDB = NULL;

1,182,874✔

2484

  taosArrayDestroyP(pReq->pNotifyAddrUrls, NULL);

1,182,874✔

2485

  pReq->pNotifyAddrUrls = NULL;

1,182,874✔

2487

  taosMemoryFreeClear(pReq->triggerFilterCols);

1,182,874✔

2488

  taosMemoryFreeClear(pReq->triggerCols);

1,182,874✔

2489

  taosMemoryFreeClear(pReq->partitionCols);

1,182,874✔

2491

  taosArrayDestroy(pReq->outTags);

1,182,874✔

2492

  pReq->outTags = NULL;

1,182,874✔

2493

  taosArrayDestroy(pReq->outCols);

1,182,874✔

2494

  pReq->outCols = NULL;

1,182,874✔

2496

  switch (pReq->triggerType) {

1,182,874✔

2497

    case WINDOW_TYPE_STATE:

316,222✔

2498

      taosMemoryFreeClear(pReq->trigger.stateWin.zeroth);

316,222✔

2499

      taosMemoryFreeClear(pReq->trigger.stateWin.expr);

316,222✔

2500

      break;

316,222✔

2501

    case WINDOW_TYPE_EVENT:

116,992✔

2502

      taosMemoryFreeClear(pReq->trigger.event.startCond);

116,992✔

2503

      taosMemoryFreeClear(pReq->trigger.event.endCond);

116,992✔

2504

      break;

116,992✔

2505

    default:

749,660✔

2506

      break;

749,660✔

2509

  taosMemoryFreeClear(pReq->triggerScanPlan);

1,182,874✔

2510

  taosArrayDestroyEx(pReq->calcScanPlanList, tFreeStreamCalcScan);

1,182,874✔

2511

  pReq->calcScanPlanList = NULL;

1,182,874✔

2512

  taosMemoryFreeClear(pReq->triggerPrevFilter);

1,182,874✔

2514

  taosMemoryFreeClear(pReq->calcPlan);

1,182,874✔

2515

  taosMemoryFreeClear(pReq->subTblNameExpr);

1,182,874✔

2516

  taosMemoryFreeClear(pReq->tagValueExpr);

1,182,874✔

2517

  taosArrayDestroyEx(pReq->forceOutCols, tFreeStreamOutCol);

1,182,874✔

2518

  pReq->forceOutCols = NULL;

1,182,874✔

2519

  taosArrayDestroy(pReq->colCids);

1,182,874✔

2520

  pReq->colCids = NULL;

1,182,874✔

2521

  taosArrayDestroy(pReq->tagCids);

1,182,874✔

2522

  pReq->tagCids = NULL;

1,182,874✔

2525

int32_t tCloneStreamCreateDeployPointers(SCMCreateStreamReq *pSrc, SCMCreateStreamReq** ppDst) {

201,171✔

2526

  int32_t code = 0, lino = 0;

201,171✔

2527

  if (NULL == pSrc) {

201,171✔

2531

  void* p = NULL;

201,171✔

2532

  int32_t num = 0;

201,171✔

2533

  *ppDst = taosMemoryCalloc(1, sizeof(SCMCreateStreamReq));

201,171✔

2534

  TSDB_CHECK_NULL(*ppDst, code, lino, _exit, terrno);

201,171✔

2536

  SCMCreateStreamReq* pDst = *ppDst;

201,171✔

2538

  if (pSrc->outDB) {

201,171✔

2539

    pDst->outDB = COPY_STR(pSrc->outDB);

197,741✔

2540

    TSDB_CHECK_NULL(pDst->outDB, code, lino, _exit, terrno);

197,741✔

2543

  if (pSrc->triggerTblName) {

201,171✔

2544

    pDst->triggerTblName = COPY_STR(pSrc->triggerTblName);

200,125✔

2545

    TSDB_CHECK_NULL(pDst->triggerTblName, code, lino, _exit, terrno);

200,125✔

2548

  if (pSrc->outTblName) {

201,171✔

2549

    pDst->outTblName = COPY_STR(pSrc->outTblName);

197,741✔

2550

    TSDB_CHECK_NULL(pDst->outTblName, code, lino, _exit, terrno);

197,741✔

2553

  if (pSrc->pNotifyAddrUrls) {

201,171✔

2554

    num = taosArrayGetSize(pSrc->pNotifyAddrUrls);

57,426✔

2555

    if (num > 0) {

57,426✔

2556

      pDst->pNotifyAddrUrls = taosArrayInit(num, POINTER_BYTES);

57,426✔

2557

      TSDB_CHECK_NULL(pDst->pNotifyAddrUrls, code, lino, _exit, terrno);

57,426✔

2559

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

114,852✔

2560

      p = taosStrdup(taosArrayGetP(pSrc->pNotifyAddrUrls, i));

57,426✔

2561

      TSDB_CHECK_NULL(p, code, lino, _exit, terrno);

57,426✔

2562

      TSDB_CHECK_NULL(taosArrayPush(pDst->pNotifyAddrUrls, &p), code, lino, _exit, terrno);

114,852✔

2566

  if (pSrc->triggerFilterCols) {

201,171✔

2567

    pDst->triggerFilterCols = COPY_STR(pSrc->triggerFilterCols);

18,206✔

2568

    TSDB_CHECK_NULL(pDst->triggerFilterCols, code, lino, _exit, terrno);

18,206✔

2571

  if (pSrc->triggerCols) {

201,171✔

2572

    pDst->triggerCols = COPY_STR(pSrc->triggerCols);

193,727✔

2573

    TSDB_CHECK_NULL(pDst->triggerCols, code, lino, _exit, terrno);

193,727✔

2576

  if (pSrc->partitionCols) {

201,171✔

2577

    pDst->partitionCols = COPY_STR(pSrc->partitionCols);

76,836✔

2578

    TSDB_CHECK_NULL(pDst->partitionCols, code, lino, _exit, terrno);

76,836✔

2581

  if (pSrc->outCols) {

201,171✔

2582

    pDst->outCols = taosArrayDup(pSrc->outCols, NULL);

197,741✔

2583

    TSDB_CHECK_NULL(pDst->outCols, code, lino, _exit, terrno);

197,741✔

2586

  if (pSrc->outTags) {

201,171✔

2587

    pDst->outTags = taosArrayDup(pSrc->outTags, NULL);

76,836✔

2588

    TSDB_CHECK_NULL(pDst->outTags, code, lino, _exit, terrno);

76,836✔

2591

  pDst->triggerType = pSrc->triggerType;

201,171✔

2593

  switch (pSrc->triggerType) {

201,171✔

2594

    case WINDOW_TYPE_STATE:

67,884✔

2595

      pDst->trigger.stateWin.slotId = pSrc->trigger.stateWin.slotId;

67,884✔

2596

      pDst->trigger.stateWin.extend = pSrc->trigger.stateWin.extend;

67,884✔

2597

      pDst->trigger.stateWin.trueForType = pSrc->trigger.stateWin.trueForType;

67,884✔

2598

      pDst->trigger.stateWin.trueForCount = pSrc->trigger.stateWin.trueForCount;

67,884✔

2599

      pDst->trigger.stateWin.trueForDuration = pSrc->trigger.stateWin.trueForDuration;

67,884✔

2600

      if (pSrc->trigger.stateWin.zeroth) {

67,884✔

2604

      if (pSrc->trigger.stateWin.expr) {

67,884✔

2605

        pDst->trigger.stateWin.expr = COPY_STR(pSrc->trigger.stateWin.expr);

67,884✔

2606

        TSDB_CHECK_NULL(pDst->trigger.stateWin.expr, code, lino, _exit, terrno);

67,884✔

2608

      break;

67,884✔

2609

    case WINDOW_TYPE_EVENT:

24,484✔

2610

      if (pSrc->trigger.event.startCond) {

24,484✔

2611

        pDst->trigger.event.startCond = COPY_STR(pSrc->trigger.event.startCond);

24,484✔

2612

        TSDB_CHECK_NULL(pDst->trigger.event.startCond, code, lino, _exit, terrno);

24,484✔

2615

      if (pSrc->trigger.event.endCond) {

24,484✔

2616

        pDst->trigger.event.endCond = COPY_STR(pSrc->trigger.event.endCond);

22,364✔

2617

        TSDB_CHECK_NULL(pDst->trigger.event.endCond, code, lino, _exit, terrno);

22,364✔

2619

      pDst->trigger.event.trueForType = pSrc->trigger.event.trueForType;

24,484✔

2620

      pDst->trigger.event.trueForCount = pSrc->trigger.event.trueForCount;

24,484✔

2621

      pDst->trigger.event.trueForDuration = pSrc->trigger.event.trueForDuration;

24,484✔

2622

      break;

24,484✔

2623

    case WINDOW_TYPE_COUNT:

20,170✔

2624

      pDst->trigger.count.countVal = pSrc->trigger.count.countVal;

20,170✔

2625

      pDst->trigger.count.sliding = pSrc->trigger.count.sliding;

20,170✔

2626

      if (pSrc->trigger.count.condCols) {

20,170✔

2630

      break;

20,170✔

2631

    default:

88,633✔

2632

      pDst->trigger = pSrc->trigger;

88,633✔

2633

      break;

88,633✔

2637

  if (pSrc->triggerScanPlan) {

201,171✔

2638

    pDst->triggerScanPlan = COPY_STR(pSrc->triggerScanPlan);

200,125✔

2639

    TSDB_CHECK_NULL(pDst->triggerScanPlan, code, lino, _exit, terrno);

200,125✔

2642

  if (pSrc->calcScanPlanList) {

201,171✔

2643

    num = taosArrayGetSize(pSrc->calcScanPlanList);

197,741✔

2644

    if (num > 0) {

197,741✔

2645

      pDst->calcScanPlanList = taosArrayInit(num, sizeof(SStreamCalcScan));

197,741✔

2646

      TSDB_CHECK_NULL(pDst->calcScanPlanList, code, lino, _exit, terrno);

197,741✔

2648

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

658,571✔

2649

      SStreamCalcScan* sscan = taosArrayGet(pSrc->calcScanPlanList, i);

460,830✔

2650

      SStreamCalcScan  dscan = {.readFromCache = sscan->readFromCache};

460,830✔

2652

      dscan.vgList = taosArrayDup(sscan->vgList, NULL);

460,830✔

2653

      TSDB_CHECK_NULL(dscan.vgList, code, lino, _exit, terrno);

460,830✔

2655

      dscan.scanPlan = COPY_STR(sscan->scanPlan);

460,830✔

2656

      TSDB_CHECK_NULL(dscan.scanPlan, code, lino, _exit, terrno);

460,830✔

2658

      TSDB_CHECK_NULL(taosArrayPush(pDst->calcScanPlanList, &dscan), code, lino, _exit, terrno);

921,660✔

2662

  if (pSrc->triggerPrevFilter) {

201,171✔

2663

    pDst->triggerPrevFilter = COPY_STR(pSrc->triggerPrevFilter);

7,960✔

2664

    TSDB_CHECK_NULL(pDst->triggerPrevFilter, code, lino, _exit, terrno);

7,960✔

2667

  if (pSrc->calcPlan) {

201,171✔

2668

    pDst->calcPlan = COPY_STR(pSrc->calcPlan);

197,741✔

2669

    TSDB_CHECK_NULL(pDst->calcPlan, code, lino, _exit, terrno);

197,741✔

2672

  if (pSrc->subTblNameExpr) {

201,171✔

2673

    pDst->subTblNameExpr = COPY_STR(pSrc->subTblNameExpr);

76,836✔

2674

    TSDB_CHECK_NULL(pDst->subTblNameExpr, code, lino, _exit, terrno);

76,836✔

2677

  if (pSrc->tagValueExpr) {

201,171✔

2678

    pDst->tagValueExpr = COPY_STR(pSrc->tagValueExpr);

76,836✔

2679

    TSDB_CHECK_NULL(pDst->tagValueExpr, code, lino, _exit, terrno);

76,836✔

2682

  if (pSrc->forceOutCols) {

201,171✔

2683

    num = taosArrayGetSize(pSrc->forceOutCols);

3,496✔

2684

    if (num > 0) {

3,496✔

2685

      pDst->forceOutCols = taosArrayInit(num, sizeof(SStreamOutCol));

3,496✔

2686

      TSDB_CHECK_NULL(pDst->forceOutCols, code, lino, _exit, terrno);

3,496✔

2688

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

23,193✔

2689

      SStreamOutCol* scol = taosArrayGet(pSrc->forceOutCols, i);

19,697✔

2690

      SStreamOutCol  dcol = {.type = scol->type};

19,697✔

2692

      dcol.expr = COPY_STR(scol->expr);

19,697✔

2693

      TSDB_CHECK_NULL(dcol.expr, code, lino, _exit, terrno);

19,697✔

2695

      TSDB_CHECK_NULL(taosArrayPush(pDst->forceOutCols, &dcol), code, lino, _exit, terrno);

39,394✔

2699

  if (pSrc->colCids) {

201,171✔

2700

    pDst->colCids = taosArrayDup(pSrc->colCids, NULL);

4,321✔

2701

    TSDB_CHECK_NULL(pDst->colCids, code, lino, _exit, terrno);

4,321✔

2704

  if (pSrc->tagCids) {

201,171✔

2705

    pDst->tagCids = taosArrayDup(pSrc->tagCids, NULL);

3,614✔

2706

    TSDB_CHECK_NULL(pDst->tagCids, code, lino, _exit, terrno);

3,614✔

2709

  pDst->triggerTblUid = pSrc->triggerTblUid;

201,171✔

2710

  pDst->triggerTblSuid = pSrc->triggerTblSuid;

201,171✔

2711

  pDst->triggerTblType = pSrc->triggerTblType;

201,171✔

2712

  pDst->triggerPrec = pSrc->triggerPrec;

201,171✔

2713

  pDst->deleteReCalc = pSrc->deleteReCalc;

201,171✔

2714

  pDst->deleteOutTbl = pSrc->deleteOutTbl;

201,171✔

2715

  pDst->flags = pSrc->flags;

201,171✔

2717

_exit:

201,171✔

2719

  if (code) {

201,171✔

2724

  return code;

201,171✔

2728

int32_t tSerializeSMPauseStreamReq(void *buf, int32_t bufLen, const SMPauseStreamReq *pReq) {

6,172✔

2729

  int32_t  code = 0;

6,172✔

2732

  SEncoder encoder = {0};

6,172✔

2733

  tEncoderInit(&encoder, buf, bufLen);

6,172✔

2734

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

6,172✔

2736

  int32_t nameLen = pReq->name == NULL ? 0 : (int32_t)strlen(pReq->name) + 1;

6,172✔

2737

  TAOS_CHECK_EXIT(tEncodeBinary(&encoder, pReq->name, nameLen));

12,344✔

2738

  TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->igNotExists));

12,344✔

2739

  tEndEncode(&encoder);

6,172✔

2741

_exit:

6,172✔

2742

  if (code) {

6,172✔

2745

    tlen = encoder.pos;

6,172✔

2747

  tEncoderClear(&encoder);

6,172✔

2748

  return tlen;

6,172✔

2751

int32_t tDeserializeSMPauseStreamReq(void *buf, int32_t bufLen, SMPauseStreamReq *pReq) {

2,909✔

2752

  SDecoder decoder = {0};

2,909✔

2753

  int32_t  code = 0;

2,909✔

2756

  tDecoderInit(&decoder, buf, bufLen);

2,909✔

2757

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

2,909✔

2758

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(&decoder, (void**)&pReq->name, NULL));

5,818✔

2759

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->igNotExists));

5,818✔

2760

  tEndDecode(&decoder);

2,909✔

2762

_exit:

2,909✔

2763

  tDecoderClear(&decoder);

2,909✔

2764

  return code;

2,909✔

2767

void tFreeMPauseStreamReq(SMPauseStreamReq *pReq) {

3,086✔

2768

  taosMemoryFreeClear(pReq->name);

3,086✔

2769

3,086✔

2771

int32_t tSerializeSMResumeStreamReq(void *buf, int32_t bufLen, const SMResumeStreamReq *pReq) {

5,424✔

2772

  SEncoder encoder = {0};

5,424✔

2773

  int32_t  code = 0;

5,424✔

2776

  tEncoderInit(&encoder, buf, bufLen);

5,424✔

2777

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

5,424✔

2778

  int32_t nameLen = pReq->name == NULL ? 0 : (int32_t)strlen(pReq->name) + 1;

5,424✔

2779

  TAOS_CHECK_EXIT(tEncodeBinary(&encoder, pReq->name, nameLen));

10,848✔

2780

  TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->igNotExists));

10,848✔

2781

  TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->igUntreated));

10,848✔

2782

  tEndEncode(&encoder);

5,424✔

2784

_exit:

5,424✔

2785

  if (code) {

5,424✔

2788

    tlen = encoder.pos;

5,424✔

2790

  tEncoderClear(&encoder);

5,424✔

2791

  return tlen;

5,424✔

2794

int32_t tDeserializeSMResumeStreamReq(void *buf, int32_t bufLen, SMResumeStreamReq *pReq) {

2,537✔

2795

  SDecoder decoder = {0};

2,537✔

2796

  int32_t  code = 0;

2,537✔

2799

  tDecoderInit(&decoder, buf, bufLen);

2,537✔

2800

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

2,537✔

2801

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(&decoder, (void**)&pReq->name, NULL));

5,074✔

2802

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->igNotExists));

5,074✔

2803

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->igUntreated));

5,074✔

2804

  tEndDecode(&decoder);

2,537✔

2806

_exit:

2,537✔

2807

  tDecoderClear(&decoder);

2,537✔

2808

  return code;

2,537✔

2811

void tFreeMResumeStreamReq(SMResumeStreamReq *pReq) {

2,712✔

2812

  taosMemoryFreeClear(pReq->name);

2,712✔

2813

2,712✔

2815

int32_t tSerializeSMRecalcStreamReq(void *buf, int32_t bufLen, const SMRecalcStreamReq *pReq) {

24,888✔

2816

  SEncoder encoder = {0};

24,888✔

2817

  int32_t  code = 0;

24,888✔

2820

  tEncoderInit(&encoder, buf, bufLen);

24,888✔

2821

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

24,888✔

2822

  int32_t nameLen = pReq->name == NULL ? 0 : (int32_t)strlen(pReq->name) + 1;

24,888✔

2823

  TAOS_CHECK_EXIT(tEncodeBinary(&encoder, pReq->name, nameLen));

49,776✔

2824

  TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->calcAll));

49,776✔

2825

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->timeRange.skey));

49,776✔

2826

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->timeRange.ekey));

49,776✔

2827

  tEndEncode(&encoder);

24,888✔

2829

_exit:

24,888✔

2830

  if (code) {

24,888✔

2833

    tlen = encoder.pos;

24,888✔

2835

  tEncoderClear(&encoder);

24,888✔

2836

  return tlen;

24,888✔

2839

int32_t tDeserializeSMRecalcStreamReq(void *buf, int32_t bufLen, SMRecalcStreamReq *pReq) {

12,439✔

2840

  SDecoder decoder = {0};

12,439✔

2841

  int32_t  code = 0;

12,439✔

2844

  tDecoderInit(&decoder, buf, bufLen);

12,439✔

2845

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

12,439✔

2847

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(&decoder, (void**)&pReq->name, NULL));

24,878✔

2848

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->calcAll));

24,878✔

2849

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->timeRange.skey));

24,878✔

2850

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->timeRange.ekey));

24,878✔

2851

  tEndDecode(&decoder);

12,439✔

2853

_exit:

12,439✔

2854

  tDecoderClear(&decoder);

12,439✔

2855

  return code;

12,439✔

2858

void tFreeMRecalcStreamReq(SMRecalcStreamReq *pReq) {

24,883✔

2859

  taosMemoryFreeClear(pReq->name);

24,883✔

2860

24,883✔

2862

static int32_t tEncodeStreamProgressReq(SEncoder *pEncoder, const SStreamProgressReq *pReq) {

12,784✔

2863

  int32_t code = 0;

12,784✔

2866

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

25,568✔

2867

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

25,568✔

2868

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

25,568✔

2870

_exit:

12,784✔

2871

  return code;

12,784✔

2874

int32_t tSerializeStreamProgressReq(void *buf, int32_t bufLen, const SStreamProgressReq *pReq) {

12,784✔

2875

  SEncoder encoder = {0};

12,784✔

2876

  int32_t  code = 0;

12,784✔

2879

  tEncoderInit(&encoder, buf, bufLen);

12,784✔

2881

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

12,784✔

2882

  TAOS_CHECK_EXIT(tEncodeStreamProgressReq(&encoder, pReq));

12,784✔

2884

  tEndEncode(&encoder);

12,784✔

2886

_exit:

12,784✔

2887

  if (code) {

12,784✔

2890

    tlen = encoder.pos;

12,784✔

2892

  tEncoderClear(&encoder);

12,784✔

2893

  return tlen;

12,784✔

2896

static int32_t tDecodeStreamProgressReq(SDecoder *pDecoder, SStreamProgressReq *pReq) {

10,608✔

2897

  int32_t code = 0;

10,608✔

2900

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

21,216✔

2901

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

21,216✔

2902

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

21,216✔

2904

_exit:

10,608✔

2905

  return code;

10,608✔

2908

int32_t tDeserializeStreamProgressReq(void *buf, int32_t bufLen, SStreamProgressReq *pReq) {

10,608✔

2909

  SDecoder decoder = {0};

10,608✔

2910

  int32_t  code = 0;

10,608✔

2913

  tDecoderInit(&decoder, (char *)buf, bufLen);

10,608✔

2915

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

10,608✔

2916

  TAOS_CHECK_EXIT(tDecodeStreamProgressReq(&decoder, pReq));

10,608✔

2918

  tEndDecode(&decoder);

10,608✔

2920

_exit:

10,608✔

2921

  tDecoderClear(&decoder);

10,608✔

2922

  return code;

10,608✔

2925

static int32_t tEncodeStreamProgressRsp(SEncoder *pEncoder, const SStreamProgressRsp *pRsp) {

12,784✔

2926

  int32_t code = 0;

12,784✔

2929

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

25,568✔

2930

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

25,568✔

2931

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

25,568✔

2932

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

25,568✔

2934

_exit:

12,784✔

2935

  return code;

12,784✔

2938

int32_t tSerializeStreamProgressRsp(void *buf, int32_t bufLen, const SStreamProgressRsp *pRsp) {

12,784✔

2939

  SEncoder encoder = {0};

12,784✔

2940

  int32_t  code = 0;

12,784✔

2943

  tEncoderInit(&encoder, buf, bufLen);

12,784✔

2945

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

12,784✔

2946

  TAOS_CHECK_EXIT(tEncodeStreamProgressRsp(&encoder, pRsp));

12,784✔

2948

  tEndEncode(&encoder);

12,784✔

2950

_exit:

12,784✔

2951

  if (code) {

12,784✔

2954

    tlen = encoder.pos;

12,784✔

2956

  tEncoderClear(&encoder);

12,784✔

2957

  return tlen;

12,784✔

2960

static int32_t tDecodeStreamProgressRsp(SDecoder *pDecoder, SStreamProgressRsp *pRsp) {

6,392✔

2961

  int32_t code = 0;

6,392✔

2964

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

12,784✔

2965

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, (int8_t *)&pRsp->fillHisFinished));

12,784✔

2966

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

12,784✔

2967

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

12,784✔

2969

_exit:

6,392✔

2970

  return code;

6,392✔

2973

int32_t tDeserializeSStreamProgressRsp(void *buf, int32_t bufLen, SStreamProgressRsp *pRsp) {

6,392✔

2974

  SDecoder decoder = {0};

6,392✔

2975

  int32_t  code = 0;

6,392✔

2978

  tDecoderInit(&decoder, buf, bufLen);

6,392✔

2980

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

6,392✔

2981

  TAOS_CHECK_EXIT(tDecodeStreamProgressRsp(&decoder, pRsp));

6,392✔

2983

  tEndDecode(&decoder);

6,392✔

2985

_exit:

6,392✔

2986

  tDecoderClear(&decoder);

6,392✔

2987

  return code;

6,392✔

2990

int32_t tSerializeSTriggerOrigTableInfoRsp(void* buf, int32_t bufLen, const SSTriggerOrigTableInfoRsp* pRsp){

278,482✔

2991

  SEncoder encoder = {0};

278,482✔

2992

  int32_t  code = TSDB_CODE_SUCCESS;

278,482✔

2993

  int32_t  lino = 0;

278,482✔

2994

  int32_t  tlen = 0;

278,482✔

2996

  tEncoderInit(&encoder, buf, bufLen);

278,482✔

2997

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

278,482✔

2999

  int32_t size = taosArrayGetSize(pRsp->cols);

278,482✔

3000

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, size));

278,482✔

3001

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

1,026,594✔

3002

    OTableInfoRsp* oInfo = taosArrayGet(pRsp->cols, i);

748,112✔

3003

    if (oInfo == NULL) {

748,112✔

3008

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, oInfo->suid));

1,496,224✔

3009

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, oInfo->uid));

1,496,224✔

3010

    TAOS_CHECK_EXIT(tEncodeI16(&encoder, oInfo->cid));

1,496,224✔

3013

  tEndEncode(&encoder);

278,482✔

3015

_exit:

278,482✔

3016

  if (code != TSDB_CODE_SUCCESS) {

278,482✔

3019

    tlen = encoder.pos;

278,482✔

3021

  tEncoderClear(&encoder);

278,482✔

3022

  return tlen;

278,482✔

3025

int32_t tDserializeSTriggerOrigTableInfoRsp(void* buf, int32_t bufLen, SSTriggerOrigTableInfoRsp* pRsp){

139,241✔

3026

  SDecoder decoder = {0};

139,241✔

3027

  int32_t  code = TSDB_CODE_SUCCESS;

139,241✔

3028

  int32_t  lino = 0;

139,241✔

3030

  tDecoderInit(&decoder, buf, bufLen);

139,241✔

3031

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

139,241✔

3033

  int32_t size = 0;

139,241✔

3034

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &size));

139,241✔

3035

  pRsp->cols = taosArrayInit(size, sizeof(OTableInfoRsp));

139,241✔

3036

  if (pRsp->cols == NULL) {

139,241✔

3041

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

513,297✔

3042

    OTableInfoRsp* oInfo = taosArrayReserve(pRsp->cols, 1);

374,056✔

3043

    if (oInfo == NULL) {

374,056✔

3048

    TAOS_CHECK_RETURN(tDecodeI64(&decoder, &oInfo->suid));

748,112✔

3049

    TAOS_CHECK_RETURN(tDecodeI64(&decoder, &oInfo->uid));

748,112✔

3050

    TAOS_CHECK_RETURN(tDecodeI16(&decoder, &oInfo->cid));

748,112✔

3053

  tEndDecode(&decoder);

139,241✔

3055

_exit:

139,241✔

3056

  tDecoderClear(&decoder);

139,241✔

3057

  return code;

139,241✔

3060

void    tDestroySTriggerOrigTableInfoRsp(SSTriggerOrigTableInfoRsp* pRsp){

27,038,422✔

3061

  taosArrayDestroy(pRsp->cols);

27,038,422✔

3062

27,038,422✔

3064

void tDestroySTriggerPullRequest(SSTriggerPullRequestUnion* pReq) {

36,036,461✔

3065

  if (pReq == NULL) return;

36,036,461✔

3066

  if (pReq->base.type == STRIGGER_PULL_WAL_DATA_NEW || pReq->base.type == STRIGGER_PULL_WAL_CALC_DATA_NEW) {

45,388,912✔

3067

    SSTriggerWalDataNewRequest* pRequest = (SSTriggerWalDataNewRequest*)pReq;

9,353,103✔

3068

    taosArrayDestroy(pRequest->versions);

9,353,103✔

3069

    tSimpleHashCleanup(pRequest->ranges);

9,352,698✔

3070

  } else if (pReq->base.type == STRIGGER_PULL_TSDB_DATA) {

26,685,860✔

3071

    SSTriggerTsdbDataRequest* pRequest = (SSTriggerTsdbDataRequest*)pReq;

496,354✔

3072

    if (pRequest->cids != NULL) {

496,354✔

3073

      taosArrayDestroy(pRequest->cids);

496,354✔

3074

      pRequest->cids = NULL;

496,354✔

3076

  } else if (pReq->base.type == STRIGGER_PULL_VTABLE_INFO) {

26,187,415✔

3077

    SSTriggerVirTableInfoRequest* pRequest = (SSTriggerVirTableInfoRequest*)pReq;

79,349✔

3078

    if (pRequest->cids != NULL) {

79,349✔

3079

      taosArrayDestroy(pRequest->cids);

79,349✔

3080

      pRequest->cids = NULL;

79,349✔

3082

    if (pRequest->uids != NULL) {

79,349✔

3086

  } else if (pReq->base.type == STRIGGER_PULL_VTABLE_PSEUDO_COL) {

26,108,043✔

3087

    SSTriggerVirTablePseudoColRequest *pRequest = (SSTriggerVirTablePseudoColRequest*)pReq;

1,039,304✔

3088

    if (pRequest->cids != NULL) {

1,039,304✔

3089

      taosArrayDestroy(pRequest->cids);

1,039,304✔

3090

      pRequest->cids = NULL;

1,039,304✔

3092

  } else if (pReq->base.type == STRIGGER_PULL_OTABLE_INFO) {

25,068,756✔

3093

    SSTriggerOrigTableInfoRequest* pRequest = (SSTriggerOrigTableInfoRequest*)pReq;

139,241✔

3094

    if (pRequest->cols != NULL) {

139,241✔

3095

      taosArrayDestroy(pRequest->cols);

139,241✔

3096

      pRequest->cols = NULL;

139,241✔

3098

  } else if (pReq->base.type == STRIGGER_PULL_SET_TABLE) {

24,929,515✔

3099

    SSTriggerSetTableRequest* pRequest = (SSTriggerSetTableRequest*)pReq;

139,241✔

3100

    tSimpleHashCleanup(pRequest->uidInfoTrigger);

139,241✔

3101

    tSimpleHashCleanup(pRequest->uidInfoCalc);

139,017✔

3105

int32_t encodePlainArray(SEncoder *encoder, SArray *pArr) {

3,395,086✔

3106

  int32_t  code = TSDB_CODE_SUCCESS;

3,395,086✔

3107

  int32_t  lino = 0;

3,395,086✔

3108

  int32_t  nEle = taosArrayGetSize(pArr);

3,395,086✔

3109

  uint8_t* buf = (nEle > 0) ? TARRAY_DATA(pArr) : NULL;

3,395,086✔

3110

  int32_t  len = (nEle > 0) ? (nEle * pArr->elemSize) : 0;

3,395,086✔

3111

  TAOS_CHECK_EXIT(tEncodeBinary(encoder, buf, len));

6,790,422✔

3113

_exit:

3,395,336✔

3114

  return code;

3,395,336✔

3117

int32_t decodePlainArray(SDecoder* decoder, SArray** ppArr, uint32_t elemSize) {

1,694,356✔

3118

  int32_t  code = TSDB_CODE_SUCCESS;

1,694,356✔

3119

  int32_t  lino = 0;

1,694,356✔

3120

  void*    buf = NULL;

1,694,356✔

3121

  uint64_t len = 0;

1,694,356✔

3122

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(decoder, &buf, &len));

1,694,356✔

3124

  if (len > 0) {

1,694,356✔

3125

    *ppArr = taosArrayInit(0, elemSize);

1,615,007✔

3126

    TSDB_CHECK_NULL(*ppArr, code, lino, _exit, terrno);

1,615,007✔

3127

    TSWAP((*ppArr)->pData, buf);

1,615,007✔

3128

    (*ppArr)->size = (*ppArr)->capacity = len / elemSize;

1,615,007✔

3131

_exit:

1,694,356✔

3132

  if (buf != NULL) {

1,694,356✔

3133

    taosMemoryFree(buf);

1,615,007✔

3135

  return code;

1,694,356✔

3138

static int32_t encodeSetTableMapInfo(SEncoder* encoder, SSHashObj* pInfo) {

556,964✔

3139

  int32_t  code = TSDB_CODE_SUCCESS;

556,964✔

3140

  int32_t  lino = 0;

556,964✔

3141

  int32_t size = tSimpleHashGetSize(pInfo);

556,964✔

3142

  TAOS_CHECK_EXIT(tEncodeI32(encoder, size));

556,964✔

3143

  int32_t iter = 0;

556,964✔

3144

  void*   px = tSimpleHashIterate(pInfo, NULL, &iter);

556,964✔

3145

  while (px != NULL) {

1,248,676✔

3146

    int64_t* uid = tSimpleHashGetKey(px, NULL);

691,712✔

3147

    TAOS_CHECK_EXIT(tEncodeI64(encoder, *uid));

1,383,424✔

3148

    TAOS_CHECK_EXIT(tEncodeI64(encoder, *(uid + 1)));

1,383,424✔

3149

    SSHashObj* info = *(SSHashObj**)px;

691,712✔

3150

    int32_t len = tSimpleHashGetSize(info);

691,712✔

3151

    TAOS_CHECK_EXIT(tEncodeI32(encoder, len));

691,712✔

3152

    int32_t iter1 = 0;

691,712✔

3153

    void*   px1 = tSimpleHashIterate(info, NULL, &iter1);

691,712✔

3154

    while (px1 != NULL) {

2,606,844✔

3155

      int16_t* slot = tSimpleHashGetKey(px1, NULL);

1,915,132✔

3156

      int16_t* cid = (int16_t*)px1;

1,915,132✔

3157

      TAOS_CHECK_EXIT(tEncodeI16(encoder, *slot));

3,830,264✔

3158

      TAOS_CHECK_EXIT(tEncodeI16(encoder, *cid));

3,830,014✔

3160

      px1 = tSimpleHashIterate(info, px1, &iter1);

1,914,882✔

3163

    px = tSimpleHashIterate(pInfo, px, &iter);

691,712✔

3166

_exit:

556,714✔

3167

  return code;

556,964✔

3170

int32_t tSerializeSTriggerPullRequest(void* buf, int32_t bufLen, const SSTriggerPullRequest* pReq) {

72,462,206✔

3171

  SEncoder encoder = {0};

72,462,206✔

3172

  int32_t  code = TSDB_CODE_SUCCESS;

72,466,617✔

3173

  int32_t  lino = 0;

72,466,617✔

3174

  int32_t  tlen = 0;

72,466,617✔

3176

  tEncoderInit(&encoder, buf, bufLen);

72,466,617✔

3177

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

72,467,748✔

3179

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, pReq->type));

144,930,383✔

3180

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->streamId));

144,922,872✔

3181

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->readerTaskId));

144,919,389✔

3182

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->sessionId));

144,922,627✔

3184

  switch (pReq->type) {

72,460,711✔

3185

    case STRIGGER_PULL_SET_TABLE: {

278,482✔

3186

      SSTriggerSetTableRequest* pRequest = (SSTriggerSetTableRequest*)pReq;

278,482✔

3187

      TAOS_CHECK_EXIT(encodeSetTableMapInfo(&encoder, pRequest->uidInfoTrigger));

278,482✔

3188

      TAOS_CHECK_EXIT(encodeSetTableMapInfo(&encoder, pRequest->uidInfoCalc));

278,482✔

3189

      break;

278,482✔

3191

    case STRIGGER_PULL_LAST_TS: {

612,359✔

3192

      break;

612,359✔

3194

    case STRIGGER_PULL_FIRST_TS: {

507,389✔

3195

      SSTriggerFirstTsRequest* pRequest = (SSTriggerFirstTsRequest*)pReq;

507,389✔

3196

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->gid));

1,014,778✔

3197

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->startTime));

1,014,778✔

3198

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

1,014,531✔

3199

      break;

507,142✔

3201

    case STRIGGER_PULL_TSDB_META: {

1,110,090✔

3202

      SSTriggerTsdbMetaRequest* pRequest = (SSTriggerTsdbMetaRequest*)pReq;

1,110,090✔

3203

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->startTime));

2,220,180✔

3204

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->endTime));

2,220,180✔

3205

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->gid));

2,220,180✔

3206

      TAOS_CHECK_EXIT(tEncodeI8(&encoder, pRequest->order));

2,220,180✔

3207

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

2,220,180✔

3208

      break;

1,110,090✔

3213

    case STRIGGER_PULL_TSDB_TS_DATA: {

368,430✔

3214

      SSTriggerTsdbTsDataRequest* pRequest = (SSTriggerTsdbTsDataRequest*)pReq;

368,430✔

3215

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->suid));

736,860✔

3216

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->uid));

736,860✔

3217

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->skey));

736,860✔

3218

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ekey));

736,860✔

3219

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

736,860✔

3220

      break;

368,430✔

3222

    case STRIGGER_PULL_TSDB_TRIGGER_DATA: {

119,418✔

3223

      SSTriggerTsdbTriggerDataRequest* pRequest = (SSTriggerTsdbTriggerDataRequest*)pReq;

119,418✔

3224

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->startTime));

238,837✔

3225

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->gid));

238,837✔

3226

      TAOS_CHECK_EXIT(tEncodeI8(&encoder, pRequest->order));

238,836✔

3227

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

239,084✔

3228

      break;

119,666✔

3230

    case STRIGGER_PULL_TSDB_TRIGGER_DATA_NEXT: {

119,666✔

3231

      break;

119,666✔

3233

    case STRIGGER_PULL_TSDB_CALC_DATA: {

13,466,346✔

3234

      SSTriggerTsdbCalcDataRequest* pRequest = (SSTriggerTsdbCalcDataRequest*)pReq;

13,466,346✔

3235

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->gid));

26,932,692✔

3236

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->skey));

26,932,692✔

3237

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ekey));

26,932,692✔

3238

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

26,932,692✔

3239

      break;

13,466,346✔

3244

    case STRIGGER_PULL_TSDB_DATA: {

995,892✔

3245

      SSTriggerTsdbDataRequest* pRequest = (SSTriggerTsdbDataRequest*)pReq;

995,892✔

3246

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->suid));

1,991,784✔

3247

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->uid));

1,991,784✔

3248

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->skey));

1,991,784✔

3249

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ekey));

1,991,784✔

3250

      TAOS_CHECK_EXIT(encodePlainArray(&encoder, pRequest->cids));

995,892✔

3251

      TAOS_CHECK_EXIT(tEncodeI8(&encoder, pRequest->order));

1,991,784✔

3252

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

1,991,784✔

3253

      break;

995,892✔

3258

    case STRIGGER_PULL_WAL_META_NEW: {

25,856,153✔

3259

      SSTriggerWalMetaNewRequest* pRequest = (SSTriggerWalMetaNewRequest*)pReq;

25,856,153✔

3260

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->lastVer));

51,715,097✔

3261

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ctime));

51,716,769✔

3262

      break;

25,857,825✔

3264

    case STRIGGER_PULL_WAL_DATA_NEW:

18,705,671✔

3266

      SSTriggerWalDataNewRequest* pRequest = (SSTriggerWalDataNewRequest*)pReq;

18,705,671✔

3267

      int32_t                     nVersion = taosArrayGetSize(pRequest->versions);

18,705,671✔

3268

      TAOS_CHECK_EXIT(tEncodeI32(&encoder, nVersion));

18,706,168✔

3269

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

36,626,762✔

3270

        int64_t ver = *(int64_t*)TARRAY_GET_ELEM(pRequest->versions, i);

17,921,100✔

3271

        TAOS_CHECK_EXIT(tEncodeI64(&encoder, ver));

17,920,594✔

3273

      int32_t nRanges = tSimpleHashGetSize(pRequest->ranges);

18,705,662✔

3274

      TAOS_CHECK_EXIT(tEncodeI32(&encoder, nRanges));

18,706,418✔

3275

      int32_t iter = 0;

18,706,418✔

3276

      void*   px = tSimpleHashIterate(pRequest->ranges, NULL, &iter);

18,706,418✔

3277

      while (px != NULL) {

26,527,561✔

3278

        uint64_t* gid = tSimpleHashGetKey(px, NULL);

7,823,488✔

3279

        TAOS_CHECK_EXIT(tEncodeU64(&encoder, *gid));

15,647,224✔

3280

        int64_t* key = (int64_t*)px;

7,823,736✔

3281

        TAOS_CHECK_EXIT(tEncodeI64(&encoder, key[0]));

15,647,472✔

3282

        TAOS_CHECK_EXIT(tEncodeI64(&encoder, key[1]));

15,647,472✔

3284

        px = tSimpleHashIterate(pRequest->ranges, px, &iter);

7,823,736✔

3286

      break;

18,704,073✔

3288

    case STRIGGER_PULL_WAL_META_DATA_NEW: {

7,148,835✔

3289

      SSTriggerWalMetaDataNewRequest* pRequest = (SSTriggerWalMetaDataNewRequest*)pReq;

7,148,835✔

3290

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->lastVer));

14,297,848✔

3291

      break;

7,149,013✔

3293

    case STRIGGER_PULL_GROUP_COL_VALUE: {

653,812✔

3294

      SSTriggerGroupColValueRequest* pRequest = (SSTriggerGroupColValueRequest*)pReq;

653,812✔

3295

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->gid));

1,307,874✔

3296

      break;

654,062✔

3298

    case STRIGGER_PULL_VTABLE_INFO: {

159,622✔

3299

      SSTriggerVirTableInfoRequest* pRequest = (SSTriggerVirTableInfoRequest*)pReq;

159,622✔

3300

      TAOS_CHECK_EXIT(encodePlainArray(&encoder, pRequest->cids));

159,622✔

3301

      TAOS_CHECK_EXIT(encodePlainArray(&encoder, pRequest->uids));

159,622✔

3302

      TAOS_CHECK_EXIT(tEncodeBool(&encoder, pRequest->fetchAllTable));

159,622✔

3303

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

319,244✔

3304

      break;

159,622✔

3306

    case STRIGGER_PULL_VTABLE_PSEUDO_COL: {

2,079,950✔

3307

      SSTriggerVirTablePseudoColRequest* pRequest = (SSTriggerVirTablePseudoColRequest*)pReq;

2,079,950✔

3308

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->uid));

4,159,900✔

3309

      TAOS_CHECK_EXIT(encodePlainArray(&encoder, pRequest->cids));

2,079,950✔

3310

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

4,160,400✔

3311

      break;

2,080,200✔

3313

    case STRIGGER_PULL_OTABLE_INFO: {

278,482✔

3314

      SSTriggerOrigTableInfoRequest* pRequest = (SSTriggerOrigTableInfoRequest*)pReq;

278,482✔

3315

      int32_t size = taosArrayGetSize(pRequest->cols);

278,482✔

3316

      TAOS_CHECK_EXIT(tEncodeI32(&encoder, size));

278,482✔

3317

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

1,026,594✔

3318

        OTableInfo* oInfo = taosArrayGet(pRequest->cols, i);

748,112✔

3319

        if (oInfo == NULL) {

748,112✔

3324

        TAOS_CHECK_EXIT(tEncodeCStr(&encoder, oInfo->refTableName));

1,496,224✔

3325

        TAOS_CHECK_EXIT(tEncodeCStr(&encoder, oInfo->refColName));

1,496,224✔

3327

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

556,964✔

3328

      break;

278,482✔

3330

    default: {

477✔

3331

      uError("unknown pull type %d", pReq->type);

477✔

3337

  tEndEncode(&encoder);

72,462,146✔

3339

_exit:

72,456,579✔

3340

  if (code != TSDB_CODE_SUCCESS) {

72,458,406✔

3343

    tlen = encoder.pos;

72,458,406✔

3345

  tEncoderClear(&encoder);

72,458,406✔

3346

  return tlen;

72,457,445✔

3349

static void destroyHash(void* data){

345,856✔

3350

  if (data){

345,856✔

3351

    SSHashObj* tmp = *(SSHashObj**)data;

345,856✔

3352

    tSimpleHashCleanup(tmp);

345,856✔

3354

345,856✔

3356

static int32_t decodeSetTableMapInfo(SDecoder* decoder, SSHashObj** ppInfo) {

278,482✔

3357

  int32_t  code = TSDB_CODE_SUCCESS;

278,482✔

3358

  int32_t  lino = 0;

278,482✔

3359

  int32_t size = 0;

278,482✔

3360

  TAOS_CHECK_EXIT(tDecodeI32(decoder, &size));

278,482✔

3361

  *ppInfo = tSimpleHashInit(size, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));

278,482✔

3362

  if (*ppInfo == NULL) {

278,482✔

3365

  tSimpleHashSetFreeFp(*ppInfo, destroyHash);

278,482✔

3367

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

624,338✔

3368

    int64_t id[2] = {0};

345,856✔

3369

    TAOS_CHECK_EXIT(tDecodeI64(decoder, id));

345,856✔

3370

    TAOS_CHECK_EXIT(tDecodeI64(decoder, id+1));

691,712✔

3371

    int32_t len = 0;

345,856✔

3372

    TAOS_CHECK_EXIT(tDecodeI32(decoder, &len));

345,856✔

3373

    SSHashObj* tmp = tSimpleHashInit(len, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT));

345,856✔

3374

    if (tmp == NULL) {

345,856✔

3377

    TAOS_CHECK_EXIT(tSimpleHashPut(*ppInfo, id, sizeof(id), &tmp, POINTER_BYTES));

345,856✔

3379

    for (int32_t j = 0; j < len; ++j) {

1,303,422✔

3380

      int16_t slotId = 0;

957,566✔

3381

      int16_t cid = 0;

957,566✔

3382

      TAOS_CHECK_EXIT(tDecodeI16(decoder, &slotId));

957,566✔

3383

      TAOS_CHECK_EXIT(tDecodeI16(decoder, &cid));

957,566✔

3384

      TAOS_CHECK_EXIT(tSimpleHashPut(tmp, &slotId, sizeof(slotId), &cid, sizeof(cid)));

957,566✔

3387

_exit:

278,482✔

3388

  if (code != TSDB_CODE_SUCCESS) {

278,482✔

3392

  return code;

278,482✔

3395

int32_t tDeserializeSTriggerPullRequest(void* buf, int32_t bufLen, SSTriggerPullRequestUnion* pReq) {

36,037,619✔

3396

  SDecoder decoder = {0};

36,037,619✔

3397

  int32_t  code = TSDB_CODE_SUCCESS;

36,037,643✔

3398

  int32_t  lino = 0;

36,037,643✔

3400

  tDecoderInit(&decoder, buf, bufLen);

36,037,643✔

3401

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

36,038,361✔

3403

  int32_t type = 0;

36,041,829✔

3404

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &type));

36,040,043✔

3405

  SSTriggerPullRequest* pBase = &(pReq->base);

36,040,043✔

3406

  pBase->type = type;

36,039,466✔

3407

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pBase->streamId));

72,082,913✔

3408

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pBase->readerTaskId));

72,082,195✔

3409

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pBase->sessionId));

72,081,978✔

3411

  switch (type) {

36,041,362✔

3412

    case STRIGGER_PULL_SET_TABLE: {

139,241✔

3413

      SSTriggerSetTableRequest* pRequest = &(pReq->setTableReq);

139,241✔

3414

      TAOS_CHECK_EXIT(decodeSetTableMapInfo(&decoder, &pRequest->uidInfoTrigger));

139,241✔

3415

      TAOS_CHECK_EXIT(decodeSetTableMapInfo(&decoder, &pRequest->uidInfoCalc));

139,241✔

3416

      break;

139,241✔

3418

    case STRIGGER_PULL_LAST_TS: {

306,227✔

3419

      break;

306,227✔

3421

    case STRIGGER_PULL_FIRST_TS: {

250,912✔

3422

      SSTriggerFirstTsRequest* pRequest = &(pReq->firstTsReq);

250,912✔

3423

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->gid));

501,824✔

3424

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->startTime));

501,574✔

3425

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

501,324✔

3426

      break;

250,662✔

3428

    case STRIGGER_PULL_TSDB_META: {

554,967✔

3429

      SSTriggerTsdbMetaRequest* pRequest = &(pReq->tsdbMetaReq);

554,967✔

3430

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->startTime));

1,109,934✔

3431

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->endTime));

1,109,934✔

3432

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->gid));

1,109,934✔

3433

      TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pRequest->order));

1,109,934✔

3434

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

1,109,934✔

3435

      break;

554,967✔

3440

    case STRIGGER_PULL_TSDB_TS_DATA: {

184,115✔

3441

      SSTriggerTsdbTsDataRequest* pRequest = &(pReq->tsdbTsDataReq);

184,115✔

3442

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->suid));

368,230✔

3443

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->uid));

368,230✔

3444

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->skey));

368,230✔

3445

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ekey));

368,230✔

3446

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

368,230✔

3447

      break;

184,115✔

3449

    case STRIGGER_PULL_TSDB_TRIGGER_DATA: {

59,833✔

3450

      SSTriggerTsdbTriggerDataRequest* pRequest = &(pReq->tsdbTriggerDataReq);

59,833✔

3451

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->startTime));

119,666✔

3452

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->gid));

119,666✔

3453

      TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pRequest->order));

119,666✔

3454

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

119,666✔

3455

      break;

59,833✔

3457

    case STRIGGER_PULL_TSDB_TRIGGER_DATA_NEXT: {

59,833✔

3458

      break;

59,833✔

3460

    case STRIGGER_PULL_TSDB_CALC_DATA: {

6,731,329✔

3461

      SSTriggerTsdbCalcDataRequest* pRequest = &(pReq->tsdbCalcDataReq);

6,731,329✔

3462

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->gid));

13,463,098✔

3463

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->skey));

13,463,538✔

3464

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ekey));

13,463,538✔

3465

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

13,463,758✔

3466

      break;

6,731,989✔

3471

    case STRIGGER_PULL_TSDB_DATA: {

496,354✔

3472

      SSTriggerTsdbDataRequest* pRequest = &(pReq->tsdbDataReq);

496,354✔

3473

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->suid));

992,708✔

3474

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->uid));

992,708✔

3475

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->skey));

992,708✔

3476

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ekey));

992,708✔

3477

      TAOS_CHECK_EXIT(decodePlainArray(&decoder, &pRequest->cids, sizeof(col_id_t)));

496,354✔

3478

      TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pRequest->order));

992,708✔

3479

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

992,708✔

3480

      break;

496,354✔

3485

    case STRIGGER_PULL_WAL_META_NEW: {

12,821,975✔

3486

      SSTriggerWalMetaNewRequest* pRequest = &(pReq->walMetaNewReq);

12,821,975✔

3487

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->lastVer));

25,644,170✔

3488

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ctime));

25,644,390✔

3489

      break;

12,822,195✔

3491

    case STRIGGER_PULL_WAL_DATA_NEW:

9,351,902✔

3493

      SSTriggerWalDataNewRequest* pRequest = &(pReq->walDataNewReq);

9,351,902✔

3494

      int32_t                     nVersion = 0;

9,351,902✔

3495

      TAOS_CHECK_EXIT(tDecodeI32(&decoder, &nVersion));

9,352,698✔

3496

      pRequest->versions = taosArrayInit_s(sizeof(int64_t), nVersion);

9,352,698✔

3497

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

18,301,571✔

3498

        int64_t* pVer = TARRAY_GET_ELEM(pRequest->versions, i);

8,948,873✔

3499

        TAOS_CHECK_EXIT(tDecodeI64(&decoder, pVer));

8,948,873✔

3501

      int32_t nRanges = 0;

9,352,698✔

3502

      TAOS_CHECK_EXIT(tDecodeI32(&decoder, &nRanges));

9,352,698✔

3503

      pRequest->ranges = tSimpleHashInit(nRanges, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT));

9,352,698✔

3504

      if (pRequest->ranges == NULL) {

9,352,698✔

3507

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

13,262,980✔

3508

        uint64_t gid = 0;

3,910,282✔

3509

        int64_t pRange[2] = {0};

3,910,282✔

3510

        TAOS_CHECK_EXIT(tDecodeU64(&decoder, &gid));

3,910,282✔

3511

        TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRange[0]));

3,910,282✔

3512

        TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRange[1]));

3,910,282✔

3513

        TAOS_CHECK_EXIT(tSimpleHashPut(pRequest->ranges, &gid, sizeof(gid), pRange, sizeof(pRange)));

3,910,282✔

3515

      break;

9,352,698✔

3517

    case STRIGGER_PULL_WAL_META_DATA_NEW: {

3,497,648✔

3518

      SSTriggerWalMetaDataNewRequest* pRequest = &(pReq->walMetaDataNewReq);

3,497,648✔

3519

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->lastVer));

6,995,823✔

3520

      break;

3,498,176✔

3522

    case STRIGGER_PULL_GROUP_COL_VALUE: {

326,685✔

3523

      SSTriggerGroupColValueRequest* pRequest = &(pReq->groupColValueReq);

326,685✔

3524

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->gid));

653,370✔

3525

      break;

326,685✔

3527

    case STRIGGER_PULL_VTABLE_INFO: {

79,349✔

3528

      SSTriggerVirTableInfoRequest* pRequest = &(pReq->virTableInfoReq);

79,349✔

3529

      TAOS_CHECK_EXIT(decodePlainArray(&decoder, &pRequest->cids, sizeof(col_id_t)));

79,349✔

3530

      TAOS_CHECK_EXIT(decodePlainArray(&decoder, &pRequest->uids, sizeof(int64_t)));

79,349✔

3531

      TAOS_CHECK_EXIT(tDecodeBool(&decoder, &pRequest->fetchAllTable));

79,349✔

3532

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

158,698✔

3533

      break;

79,349✔

3535

    case STRIGGER_PULL_VTABLE_PSEUDO_COL: {

1,039,304✔

3536

      SSTriggerVirTablePseudoColRequest* pRequest = &(pReq->virTablePseudoColReq);

1,039,304✔

3537

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->uid));

2,078,608✔

3538

      TAOS_CHECK_EXIT(decodePlainArray(&decoder, &pRequest->cids, sizeof(col_id_t)));

1,039,304✔

3539

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

2,078,608✔

3540

      break;

1,039,304✔

3542

    case STRIGGER_PULL_OTABLE_INFO: {

139,241✔

3543

      SSTriggerOrigTableInfoRequest* pRequest = &(pReq->origTableInfoReq);

139,241✔

3544

      int32_t size = 0;

139,241✔

3545

      TAOS_CHECK_EXIT(tDecodeI32(&decoder, &size));

139,241✔

3546

      pRequest->cols = taosArrayInit(size, sizeof(OTableInfo));

139,241✔

3547

      if (pRequest->cols == NULL) {

138,445✔

3552

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

512,501✔

3553

        OTableInfo* oInfo = taosArrayReserve(pRequest->cols, 1);

374,056✔

3554

        if (oInfo == NULL) {

374,056✔

3559

        TAOS_CHECK_RETURN(tDecodeCStrTo(&decoder, oInfo->refTableName));

374,056✔

3560

        TAOS_CHECK_RETURN(tDecodeCStrTo(&decoder, oInfo->refColName));

374,056✔

3562

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

277,686✔

3564

      break;

139,241✔

3566

    default: {

2,447✔

3567

      uError("unknown pull type %d", type);

2,447✔

3573

  tEndDecode(&decoder);

36,040,869✔

3575

_exit:

36,038,179✔

3576

  tDecoderClear(&decoder);

36,038,263✔

3577

  return code;

36,039,623✔

3580

static int32_t tSerializeSTriggerCalcParam(SEncoder* pEncoder, SArray* pParams, bool ignoreNotificationInfo, bool full) {

39,322,718✔

3581

  int32_t size = full ? taosArrayGetSize(pParams) : 0;

39,322,718✔

3582

  int32_t code = 0;

39,323,571✔

3583

  int32_t lino = 0;

39,323,571✔

3584

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, size));

39,323,460✔

3585

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

2,147,483,647✔

3586

    SSTriggerCalcParam* param = taosArrayGet(pParams, i);

2,147,483,647✔

3587

    if (param == NULL) {

2,147,483,647✔

3590

    int64_t plainFieldSize = offsetof(SSTriggerCalcParam, notifyType);

2,147,483,647✔

3591

    if (pEncoder->data) {

2,147,483,647✔

3592

      TAOS_MEMCPY(pEncoder->data + pEncoder->pos, param, plainFieldSize);

2,147,483,647✔

3594

    pEncoder->pos += plainFieldSize;

2,147,483,647✔

3596

    if (!ignoreNotificationInfo) {

2,147,483,647✔

3597

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, param->notifyType));

1,244,126,235✔

3598

      uint64_t len = (param->extraNotifyContent != NULL) ? strlen(param->extraNotifyContent) + 1 : 0;

622,083,451✔

3599

      TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, (uint8_t*)param->extraNotifyContent, len));

1,186,092,826✔

3602

_exit:

4,439,066✔

3603

  return code;

4,439,066✔

3606

void tDestroySSTriggerCalcParam(void* ptr) {

2,147,483,647✔

3607

  SSTriggerCalcParam* pParam = ptr;

2,147,483,647✔

3608

  if (pParam && pParam->extraNotifyContent != NULL) {

2,147,483,647✔

3609

    taosMemoryFreeClear(pParam->extraNotifyContent);

277,832✔

3611

  if (pParam && pParam->resultNotifyContent != NULL) {

2,147,483,647✔

3614

  if (pParam && pParam->pExternalWindowData != NULL) {

2,147,483,647✔

3615

    taosArrayDestroyEx(pParam->pExternalWindowData, tDestroySStreamGroupValue);

55,910,060✔

3616

    pParam->pExternalWindowData = NULL;

55,860,860✔

3618

2,147,483,647✔

3620

void tDestroySSTriggerGroupCalcInfo(void* ptr) {

152,930✔

3621

  SSTriggerGroupCalcInfo* pCalcInfo = ptr;

152,930✔

3622

  if (pCalcInfo && pCalcInfo->pParams != NULL) {

152,930✔

3623

    taosArrayDestroyEx(pCalcInfo->pParams, tDestroySSTriggerCalcParam);

152,930✔

3624

    pCalcInfo->pParams = NULL;

152,930✔

3626

  if (pCalcInfo && pCalcInfo->pGroupColVals != NULL) {

152,930✔

3630

152,930✔

3648

void tDestroySStreamGroupValue(void* ptr) {

82,486,627✔

3649

  SStreamGroupValue* pValue = ptr;

82,486,627✔

3650

  if ((pValue != NULL) && (IS_VAR_DATA_TYPE(pValue->data.type) || pValue->data.type == TSDB_DATA_TYPE_DECIMAL)) {

82,486,627✔

3651

    taosMemoryFreeClear(pValue->data.pData);

19,773,711✔

3652

    pValue->data.nData = 0;

18,445,721✔

3654

82,535,417✔

3656

static int32_t tDeserializeSTriggerCalcParam(SDecoder* pDecoder, SArray**ppParams, bool ignoreNotificationInfo) {

19,662,126✔

3657

  int32_t size = 0, code = 0, lino = 0;

19,662,126✔

3658

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &size));

19,662,587✔

3659

  if (size <= 0) {

19,662,587✔

3660

    return code;

4,740,871✔

3663

  *ppParams = taosArrayInit(size, sizeof(SSTriggerCalcParam));

14,921,716✔

3664

  if (*ppParams == NULL) {

14,921,479✔

3667

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

2,147,483,647✔

3668

    SSTriggerCalcParam* param = taosArrayReserve(*ppParams, 1);

2,147,483,647✔

3669

    if (param == NULL) {

2,147,483,647✔

3672

    int64_t plainFieldSize = offsetof(SSTriggerCalcParam, notifyType);

2,147,483,647✔

3673

    TAOS_MEMCPY(param, pDecoder->data + pDecoder->pos, plainFieldSize);

2,147,483,647✔

3674

    pDecoder->pos += plainFieldSize;

2,147,483,647✔

3676

    if (!ignoreNotificationInfo) {

2,147,483,647✔

3677

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &param->notifyType));

622,044,127✔

3678

      uint64_t len = 0;

311,021,884✔

3679

      TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&param->extraNotifyContent, &len));

622,046,711✔

3683

_exit:

1,768,251✔

3684

  return code;

14,921,749✔

3687

static int32_t tSerializeStriggerGroupColVals(SEncoder* pEncoder, SArray* pGroupColVals, int32_t vgId) {

39,977,942✔

3688

  int32_t code = TSDB_CODE_SUCCESS;

39,977,942✔

3689

  int32_t lino = 0;

39,977,942✔

3691

  int32_t size = taosArrayGetSize(pGroupColVals);

39,977,942✔

3692

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, size));

39,977,477✔

3693

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

92,449,239✔

3694

    SStreamGroupValue* pValue = taosArrayGet(pGroupColVals, i);

52,472,696✔

3695

    if (pValue == NULL) {

52,473,016✔

3698

    TAOS_CHECK_EXIT(tEncodeBool(pEncoder, pValue->isNull));

52,473,016✔

3699

    if (pValue->isNull) {

52,473,194✔

3700

      continue;

16,650✔

3702

    TAOS_CHECK_EXIT(tEncodeBool(pEncoder, pValue->isTbname));

52,456,274✔

3703

    if (pValue->isTbname) {

52,457,051✔

3704

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pValue->uid));

49,053,135✔

3705

      if (vgId != -1) { pValue->vgId = vgId; }

24,526,973✔

3706

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

49,053,735✔

3708

    TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pValue->data.type));

104,913,643✔

3709

    if (IS_VAR_DATA_TYPE(pValue->data.type)) {

52,455,744✔

3710

      TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pValue->data.pData, pValue->data.nData));

69,950,131✔

3712

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pValue->data.val));

34,961,714✔

3716

_exit:

39,976,543✔

3717

  return code;

39,976,543✔

3720

static int32_t tDeserializeStriggerGroupColVals(SDecoder* pDecoder, SArray** ppGroupColVals) {

19,989,469✔

3721

  int32_t code = TSDB_CODE_SUCCESS;

19,989,469✔

3722

  int32_t lino = 0;

19,989,469✔

3723

  int32_t size = 0;

19,989,469✔

3725

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &size));

19,989,462✔

3726

  taosArrayClearEx(*ppGroupColVals, tDestroySStreamGroupValue);

19,989,462✔

3727

  if (size > 0) {

19,989,059✔

3728

    if (*ppGroupColVals == NULL) {

13,421,190✔

3729

      *ppGroupColVals = taosArrayInit(size, sizeof(SStreamGroupValue));

13,421,190✔

3730

      if (*ppGroupColVals == NULL) {

13,421,190✔

3737

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

46,224,977✔

3738

    SStreamGroupValue* pValue = taosArrayReserve(*ppGroupColVals, 1);

26,235,804✔

3739

    if (pValue == NULL) {

26,235,272✔

3742

    TAOS_CHECK_EXIT(tDecodeBool(pDecoder, &pValue->isNull));

26,235,272✔

3743

    if (pValue->isNull) {

26,235,556✔

3744

      continue;

8,325✔

3746

    TAOS_CHECK_EXIT(tDecodeBool(pDecoder, &pValue->isTbname));

26,227,231✔

3747

    if (pValue->isTbname) {

26,226,912✔

3748

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pValue->uid));

24,524,164✔

3749

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

24,524,300✔

3751

    TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pValue->data.type));

52,454,512✔

3752

    if (IS_VAR_DATA_TYPE(pValue->data.type)) {

43,713,608✔

3753

      uint64_t len = 0;

17,486,011✔

3754

      TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pValue->data.pData, &len));

34,972,158✔

3755

      pValue->data.nData = len;

17,485,878✔

3757

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pValue->data.val));

17,482,898✔

3760

_exit:

19,988,907✔

3761

  return code;

19,988,923✔

3764

int32_t tSerializeSStreamGroupInfo(void* buf, int32_t bufLen, const SStreamGroupInfo* gInfo, int32_t vgId) {

653,370✔

3765

  SEncoder encoder = {0};

653,370✔

3766

  int32_t  code = TSDB_CODE_SUCCESS;

653,370✔

3767

  int32_t  lino = 0;

653,370✔

3768

  int32_t  tlen = 0;

653,370✔

3770

  tEncoderInit(&encoder, buf, bufLen);

653,370✔

3771

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

653,370✔

3773

  TAOS_CHECK_EXIT(tSerializeStriggerGroupColVals(&encoder, gInfo->gInfo, vgId));

653,370✔

3775

  tEndEncode(&encoder);

652,874✔

3777

_exit:

653,122✔

3778

  if (code != TSDB_CODE_SUCCESS) {

653,122✔

3781

    tlen = encoder.pos;

653,122✔

3783

  tEncoderClear(&encoder);

653,122✔

3784

  return tlen;

653,370✔

3787

int32_t tDeserializeSStreamGroupInfo(void* buf, int32_t bufLen, SStreamGroupInfo* gInfo) {

327,031✔

3788

  SDecoder decoder = {0};

327,031✔

3789

  int32_t  code = TSDB_CODE_SUCCESS;

327,031✔

3790

  int32_t  lino = 0;

327,031✔

3791

  int32_t  size = 0;

327,031✔

3793

  tDecoderInit(&decoder, buf, bufLen);

327,031✔

3794

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

327,031✔

3796

  TAOS_CHECK_EXIT(tDeserializeStriggerGroupColVals(&decoder, &gInfo->gInfo));

327,031✔

3798

  tEndDecode(&decoder);

327,031✔

3800

_exit:

327,031✔

3801

  tDecoderClear(&decoder);

327,031✔

3802

  return code;

327,031✔

3805

static int32_t tSerializeSSTriggerGroupCalcInfo(SEncoder* pEncoder, SSTriggerGroupCalcInfo* pInfo) {

22,960✔

3806

  int32_t code = 0;

22,960✔

3807

  int32_t lino = 0;

22,960✔

3809

  TAOS_CHECK_EXIT(tSerializeSTriggerCalcParam(pEncoder, pInfo->pParams, false, true));

22,960✔

3810

  TAOS_CHECK_EXIT(tSerializeStriggerGroupColVals(pEncoder, pInfo->pGroupColVals, -1));

22,960✔

3811

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pInfo->createTable));

45,920✔

3813

_exit:

22,960✔

3814

  return code;

22,960✔

3843

int32_t tSerializeSTriggerCalcRequest(void* buf, int32_t bufLen, const SSTriggerCalcRequest* pReq) {

6,528,665✔

3844

  SEncoder encoder = {0};

6,528,665✔

3845

  int32_t  code = TSDB_CODE_SUCCESS;

6,528,665✔

3846

  int32_t  lino = 0;

6,528,665✔

3847

  int32_t  tlen = 0;

6,528,665✔

3849

  tEncoderInit(&encoder, buf, bufLen);

6,528,665✔

3850

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

6,528,483✔

3852

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->streamId));

13,057,703✔

3853

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->runnerTaskId));

13,058,440✔

3854

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->sessionId));

13,058,192✔

3855

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, pReq->triggerType));

13,057,951✔

3856

  TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->isMultiGroupCalc));

13,057,710✔

3857

  TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->stbPartByTbname));

13,057,462✔

3859

  if (!pReq->isMultiGroupCalc) {

6,528,731✔

3860

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->gid));

13,057,469✔

3861

    TAOS_CHECK_EXIT(tSerializeSTriggerCalcParam(&encoder, pReq->params, false, true));

6,528,738✔

3862

    TAOS_CHECK_EXIT(tSerializeStriggerGroupColVals(&encoder, pReq->groupColVals, -1));

6,528,731✔

3863

    TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->createTable));

13,056,973✔

3894

  TAOS_CHECK_EXIT(tEncodeBool(&encoder, pReq->isWindowTrigger));

6,528,242✔

3895

  TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->precision));

13,058,199✔

3897

  tEndEncode(&encoder);

6,528,979✔

3899

_exit:

6,528,490✔

3900

  if (code != TSDB_CODE_SUCCESS) {

6,528,731✔

3903

    tlen = encoder.pos;

6,528,731✔

3905

  tEncoderClear(&encoder);

6,528,731✔

3906

  return tlen;

6,528,497✔

3909

static int32_t tDeserializeSSTriggerGroupCalcInfo(SDecoder* pDecoder, SSTriggerGroupCalcInfo* pInfo) {

11,480✔

3910

  int32_t code = 0;

11,480✔

3911

  int32_t lino = 0;

11,480✔

3913

  TAOS_CHECK_EXIT(tDeserializeSTriggerCalcParam(pDecoder, &pInfo->pParams, false));

11,480✔

3914

  TAOS_CHECK_EXIT(tDeserializeStriggerGroupColVals(pDecoder, &pInfo->pGroupColVals));

11,480✔

3915

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pInfo->createTable));

22,960✔

3917

_exit:

11,480✔

3918

  return code;

11,480✔

3945

int32_t tDeserializeSTriggerCalcRequest(void* buf, int32_t bufLen, SSTriggerCalcRequest* pReq) {

3,263,778✔

3946

  SDecoder decoder = {0};

3,263,778✔

3947

  int32_t  code = TSDB_CODE_SUCCESS;

3,263,778✔

3948

  int32_t  lino = 0;

3,263,778✔

3950

  tDecoderInit(&decoder, buf, bufLen);

3,263,778✔

3951

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

3,263,778✔

3953

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

6,527,556✔

3954

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->runnerTaskId));

6,527,556✔

3955

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->sessionId));

6,527,556✔

3956

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pReq->triggerType));

6,527,556✔

3957

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->isMultiGroupCalc));

6,527,556✔

3958

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->stbPartByTbname));

6,527,556✔

3960

  if (!pReq->isMultiGroupCalc) {

3,263,778✔

3961

    TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->gid));

6,527,556✔

3962

    TAOS_CHECK_EXIT(tDeserializeSTriggerCalcParam(&decoder, &pReq->params, false));

3,263,778✔

3963

    TAOS_CHECK_EXIT(tDeserializeStriggerGroupColVals(&decoder, &pReq->groupColVals));

3,263,778✔

3964

    TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->createTable));

6,527,315✔

4005

  if (!tDecodeIsEnd(&decoder)) {

3,263,778✔

4006

    TAOS_CHECK_EXIT(tDecodeBool(&decoder, &pReq->isWindowTrigger));

3,263,778✔

4007

    TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->precision));

6,527,556✔

4010

  tEndDecode(&decoder);

3,263,778✔

4012

_exit:

3,263,778✔

4013

  tDecoderClear(&decoder);

3,263,778✔

4014

  return code;

3,263,778✔

4017

void tDestroySTriggerCalcRequest(SSTriggerCalcRequest* pReq) {

9,380,689✔

4018

  if (pReq != NULL) {

9,380,689✔

4019

    if (pReq->params != NULL) {

9,380,689✔

4020

      taosArrayDestroyEx(pReq->params, tDestroySSTriggerCalcParam);

6,327,384✔

4021

      pReq->params = NULL;

6,327,143✔

4023

    if (pReq->groupColVals != NULL) {

9,380,448✔

4024

      taosArrayDestroyEx(pReq->groupColVals, tDestroySStreamGroupValue);

3,522,359✔

4025

      pReq->groupColVals = NULL;

3,522,359✔

4027

    if (pReq->pGroupCalcInfos != NULL) {

9,380,448✔

4031

    if (pReq->pGroupReadInfos != NULL) {

9,380,689✔

4035

    blockDataDestroy(pReq->pOutBlock);

9,380,689✔

4036

    pReq->pOutBlock = NULL;

9,380,448✔

4038

9,380,689✔

4099

int32_t tSerializeSTriggerCtrlRequest(void* buf, int32_t bufLen, const SSTriggerCtrlRequest* pReq) {

39,411,260✔

4100

  SEncoder encoder = {0};

39,411,260✔

4101

  int32_t  code = TSDB_CODE_SUCCESS;

39,411,260✔

4102

  int32_t  lino = 0;

39,411,260✔

4103

  int32_t  tlen = 0;

39,411,260✔

4105

  tEncoderInit(&encoder, buf, bufLen);

39,411,260✔

4106

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

39,411,260✔

4108

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, pReq->type));

78,822,520✔

4109

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->streamId));

78,822,520✔

4110

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->taskId));

78,822,520✔

4111

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->sessionId));

78,822,520✔

4113

  tEndEncode(&encoder);

39,411,260✔

4115

_exit:

39,411,260✔

4116

  if (code != TSDB_CODE_SUCCESS) {

39,411,260✔

4119

    tlen = encoder.pos;

39,411,260✔

4121

  tEncoderClear(&encoder);

39,411,260✔

4122

  return tlen;

39,411,260✔

4125

int32_t tDeserializeSTriggerCtrlRequest(void* buf, int32_t bufLen, SSTriggerCtrlRequest* pReq) {

59,090,855✔

4126

  SDecoder decoder = {0};

59,090,855✔

4127

  int32_t  code = TSDB_CODE_SUCCESS;

59,092,328✔

4128

  int32_t  lino = 0;

59,092,328✔

4130

  tDecoderInit(&decoder, buf, bufLen);

59,092,328✔

4131

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

59,087,348✔

4133

  int32_t type = 0;

59,099,148✔

4134

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &type));

59,098,305✔

4135

  pReq->type = type;

59,098,305✔

4136

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

118,200,807✔

4137

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->taskId));

118,207,307✔

4138

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->sessionId));

118,205,388✔

4140

  tEndDecode(&decoder);

59,103,064✔

4142

_exit:

59,100,618✔

4143

  tDecoderClear(&decoder);

59,100,866✔

4144

  return code;

59,093,784✔

4147

int32_t tSerializeStRtFuncInfo(SEncoder* pEncoder, const SStreamRuntimeFuncInfo* pInfo, bool needStreamRtInfo, bool needStreamGrpInfo) {

32,785,266✔

4148

  int32_t code = 0, lino = 0;

32,785,266✔

4149

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pInfo->isMultiGroupCalc));

65,570,566✔

4150

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pInfo->stbPartByTbname));

65,568,871✔

4151

  TAOS_CHECK_EXIT(tEncodeBool(pEncoder, needStreamRtInfo));

32,783,571✔

4152

  TAOS_CHECK_EXIT(tEncodeBool(pEncoder, needStreamGrpInfo));

32,785,134✔

4153

  if (pInfo->isMultiGroupCalc) {

32,784,714✔

4154

    if (needStreamRtInfo) {

13,120✔

4155

      if (needStreamGrpInfo) {

13,120✔

4156

        int32_t nGroups = taosArrayGetSize(pInfo->curGrpRead);

13,120✔

4157

        TAOS_CHECK_EXIT(tEncodeI32(pEncoder, nGroups));

13,120✔

4158

        int8_t withGrpCalcInfo = 1;

13,120✔

4159

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

13,120✔

4167

        TAOS_CHECK_EXIT(tEncodeI8(pEncoder, withGrpCalcInfo));

26,240✔

4168

        if (withGrpCalcInfo) {

13,120✔

4169

          int32_t nGroups = tSimpleHashGetSize(pInfo->pGroupCalcInfos);

13,120✔

4170

          TAOS_CHECK_EXIT(tEncodeI32(pEncoder, nGroups));

13,120✔

4171

          int32_t                 iter1 = 0;

13,120✔

4172

          SSTriggerGroupCalcInfo* pCalcInfo = tSimpleHashIterate(pInfo->pGroupCalcInfos, NULL, &iter1);

13,120✔

4173

          while (pCalcInfo != NULL) {

36,080✔

4174

            int64_t* gid = tSimpleHashGetKey(pCalcInfo, NULL);

22,960✔

4175

            TAOS_CHECK_EXIT(tEncodeI64(pEncoder, *gid));

45,920✔

4176

            TAOS_CHECK_EXIT(tSerializeSSTriggerGroupCalcInfo(pEncoder, pCalcInfo));

22,960✔

4177

            pCalcInfo = tSimpleHashIterate(pInfo->pGroupCalcInfos, pCalcInfo, &iter1);

22,960✔

4211

    TAOS_CHECK_EXIT(tSerializeSTriggerCalcParam(pEncoder, pInfo->pStreamPesudoFuncVals, true, needStreamRtInfo));

32,771,876✔

4212

    TAOS_CHECK_EXIT(tSerializeStriggerGroupColVals(pEncoder, pInfo->pStreamPartColVals, -1));

32,772,833✔

4213

    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pInfo->groupId));

65,541,665✔

4216

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pInfo->curWindow.skey));

65,568,627✔

4217

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pInfo->curWindow.ekey));

65,569,639✔

4218

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

65,568,535✔

4219

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pInfo->sessionId));

65,568,808✔

4220

  TAOS_CHECK_EXIT(tEncodeBool(pEncoder, pInfo->withExternalWindow));

32,784,742✔

4221

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

65,568,919✔

4222

  TAOS_CHECK_EXIT(tEncodeBool(pEncoder, pInfo->isWindowTrigger));

32,783,563✔

4223

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pInfo->precision));

65,566,758✔

4224

_exit:

32,783,270✔

4225

  return code;

32,783,270✔

4228

int32_t tDeserializeStRtFuncInfo(SDecoder* pDecoder, SStreamRuntimeFuncInfo* pInfo) {

16,393,584✔

4229

  int32_t code = 0, lino = 0;

16,393,584✔

4230

  int32_t size = 0;

16,393,584✔

4231

  bool needStreamRtInfo = false;

16,393,584✔

4232

  bool needStreamGrpInfo = false;

16,392,796✔

4234

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pInfo->isMultiGroupCalc));

32,786,380✔

4235

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pInfo->stbPartByTbname));

32,787,754✔

4236

  TAOS_CHECK_EXIT(tDecodeBool(pDecoder, &needStreamRtInfo));

16,394,170✔

4237

  TAOS_CHECK_EXIT(tDecodeBool(pDecoder, &needStreamGrpInfo));

16,394,170✔

4239

  if (pInfo->isMultiGroupCalc) {

16,394,170✔

4240

    if (needStreamRtInfo) {

6,560✔

4241

      if (needStreamGrpInfo) {

6,560✔

4242

        int32_t nGroups = 0;

6,560✔

4243

        TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &nGroups));

6,560✔

4244

        if (nGroups > 0) {

6,560✔

4248

        for (int32_t j = 0; j < nGroups; ++j) {

6,560✔

4252

        int8_t withGrpCalcInfo = 0;

6,560✔

4253

        TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &withGrpCalcInfo));

6,560✔

4254

        if (withGrpCalcInfo) {

6,560✔

4255

          int32_t nGroups = 0;

6,560✔

4256

          TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &nGroups));

6,560✔

4257

          pInfo->pGroupCalcInfos = tSimpleHashInit(nGroups, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));

6,560✔

4258

          QUERY_CHECK_NULL(pInfo->pGroupCalcInfos, code, lino, _exit, terrno);

6,560✔

4259

          tSimpleHashSetFreeFp(pInfo->pGroupCalcInfos, tDestroySSTriggerGroupCalcInfo);

6,560✔

4261

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

18,040✔

4262

            int64_t gid = 0;

11,480✔

4263

            TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &gid));

11,480✔

4264

            SSTriggerGroupCalcInfo info = {0};

11,480✔

4265

            TAOS_CHECK_EXIT(tSimpleHashPut(pInfo->pGroupCalcInfos, &gid, sizeof(int64_t), &info, sizeof(info)));

11,480✔

4266

            SSTriggerGroupCalcInfo* pCalcInfo = tSimpleHashGet(pInfo->pGroupCalcInfos, &gid, sizeof(int64_t));

11,480✔

4267

            QUERY_CHECK_NULL(pCalcInfo, code, lino, _exit, TSDB_CODE_INTERNAL_ERROR);

11,480✔

4268

            TAOS_CHECK_EXIT(tDeserializeSSTriggerGroupCalcInfo(pDecoder, pCalcInfo));

11,480✔

4313

    TAOS_CHECK_EXIT(tDeserializeSTriggerCalcParam(pDecoder, &pInfo->pStreamPesudoFuncVals, true));

16,387,610✔

4314

    TAOS_CHECK_EXIT(tDeserializeStriggerGroupColVals(pDecoder, &pInfo->pStreamPartColVals));

16,387,362✔

4315

    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pInfo->groupId));

32,773,705✔

4318

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pInfo->curWindow.skey));

32,787,416✔

4319

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pInfo->curWindow.ekey));

32,787,930✔

4320

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pInfo->curIdx));

32,788,340✔

4321

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pInfo->sessionId));

32,788,074✔

4322

  TAOS_CHECK_EXIT(tDecodeBool(pDecoder, &pInfo->withExternalWindow));

16,393,904✔

4323

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pInfo->triggerType));

32,788,057✔

4324

  if (!tDecodeIsEnd(pDecoder)) {

16,393,228✔

4325

    TAOS_CHECK_EXIT(tDecodeBool(pDecoder, &pInfo->isWindowTrigger));

16,393,494✔

4326

    TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pInfo->precision));

32,787,930✔

4328

_exit:

16,393,494✔

4329

  return code;

16,393,904✔

4332

void tDestroyStRtFuncInfo(SStreamRuntimeFuncInfo* pInfo){

563,467,686✔

4333

  if (pInfo == NULL) return;

563,467,686✔

4334

  if (pInfo->pStreamPesudoFuncVals != NULL) {

26,222,862✔

4335

    taosArrayDestroyEx(pInfo->pStreamPesudoFuncVals, tDestroySSTriggerCalcParam);

9,445,626✔

4336

    pInfo->pStreamPesudoFuncVals = NULL;

9,446,095✔

4338

  if (pInfo->pStreamPartColVals != NULL) {

26,250,981✔

4339

    taosArrayDestroyEx(pInfo->pStreamPartColVals, tDestroySStreamGroupValue);

10,100,860✔

4340

    pInfo->pStreamPartColVals = NULL;

10,100,860✔

4342

  if (pInfo->pGroupCalcInfos != NULL) {

26,250,275✔

4343

    tSimpleHashCleanup(pInfo->pGroupCalcInfos);

77,900✔

4344

    pInfo->pGroupCalcInfos = NULL;

77,900✔

4346

  if (pInfo->pGroupReadInfos != NULL) {

26,249,995✔

4350

  if (pInfo->outNormalTable != NULL) {

26,250,555✔

4355

int32_t tSerializeSStreamMsgVTableInfo(void* buf, int32_t bufLen, const SStreamMsgVTableInfo* pRsp){

158,698✔

4356

  SEncoder encoder = {0};

158,698✔

4357

  int32_t  code = TSDB_CODE_SUCCESS;

158,698✔

4358

  int32_t  lino = 0;

158,698✔

4359

  int32_t  tlen = 0;

158,698✔

4361

  tEncoderInit(&encoder, buf, bufLen);

158,698✔

4362

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

158,698✔

4364

  int32_t size = taosArrayGetSize(pRsp->infos);

158,698✔

4365

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, size));

158,698✔

4366

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

546,922✔

4367

    VTableInfo* info = taosArrayGet(pRsp->infos, i);

388,224✔

4368

    if (info == NULL) {

388,224✔

4371

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, info->gId));

776,448✔

4372

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, info->uid));

776,448✔

4373

    TAOS_CHECK_EXIT(tEncodeSColRefWrapper(&encoder, &info->cols));

388,224✔

4376

  tEndEncode(&encoder);

158,698✔

4378

_exit:

158,698✔

4379

  if (code != TSDB_CODE_SUCCESS) {

158,698✔

4382

    tlen = encoder.pos;

158,698✔

4384

  tEncoderClear(&encoder);

158,698✔

4385

  return tlen;

158,698✔

4388

int32_t tDeserializeSStreamMsgVTableInfo(void* buf, int32_t bufLen, SStreamMsgVTableInfo *vTableInfo){

79,349✔

4389

  SDecoder decoder = {0};

79,349✔

4390

  int32_t  code = TSDB_CODE_SUCCESS;

79,349✔

4391

  int32_t  lino = 0;

79,349✔

4392

  int32_t  size = 0;

79,349✔

4394

  tDecoderInit(&decoder, buf, bufLen);

79,349✔

4395

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

79,349✔

4397

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &size));

79,349✔

4398

  vTableInfo->infos = taosArrayInit(size, sizeof(VTableInfo));

79,349✔

4399

  if (vTableInfo->infos == NULL) {

79,349✔

4402

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

273,461✔

4403

    VTableInfo* info = taosArrayReserve(vTableInfo->infos, 1);

194,112✔

4404

    if (info == NULL) {

194,112✔

4407

    TAOS_CHECK_EXIT(tDecodeI64(&decoder, &info->gId));

388,224✔

4408

    TAOS_CHECK_EXIT(tDecodeI64(&decoder, &info->uid));

388,224✔

4409

    TAOS_CHECK_EXIT(tDecodeSColRefWrapperEx(&decoder, &info->cols, false));

194,112✔

4412

  tEndDecode(&decoder);

79,349✔

4414

_exit:

79,349✔

4415

  tDecoderClear(&decoder);

79,349✔

4416

  return code;

79,349✔

4420

void tDestroyVTableInfo(void *ptr) {

388,224✔

4421

  if (NULL == ptr) {

388,224✔

4424

  VTableInfo* pTable = (VTableInfo*)ptr;

388,224✔

4425

  taosMemoryFree(pTable->cols.pColRef);

388,224✔

4428

void tDestroySStreamMsgVTableInfo(SStreamMsgVTableInfo *ptr) {

26,977,734✔

4429

  if (ptr == NULL) return;

26,977,734✔

4430

  taosArrayDestroyEx(ptr->infos, tDestroyVTableInfo);

26,977,734✔

4431

  ptr->infos = NULL;

26,979,079✔

4434

int32_t tSerializeSStreamTsResponse(void* buf, int32_t bufLen, const SStreamTsResponse* pRsp) {

1,113,534✔

4435

  SEncoder encoder = {0};

1,113,534✔

4436

  int32_t  code = TSDB_CODE_SUCCESS;

1,113,534✔

4437

  int32_t  lino = 0;

1,113,534✔

4438

  int32_t  tlen = 0;

1,113,534✔

4440

  tEncoderInit(&encoder, buf, bufLen);

1,113,534✔

4441

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

1,113,535✔

4443

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRsp->ver));

2,226,824✔

4444

  int32_t size = taosArrayGetSize(pRsp->tsInfo);

1,113,536✔

4445

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, size));

1,114,278✔

4446

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

2,738,939✔

4447

    STsInfo* tsInfo = taosArrayGet(pRsp->tsInfo, i);

1,625,405✔

4448

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, tsInfo->gId));

3,251,800✔

4449

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, tsInfo->ts));

3,250,561✔

4452

  tEndEncode(&encoder);

1,113,534✔

4454

_exit:

1,114,030✔

4455

  if (code != TSDB_CODE_SUCCESS) {

1,114,031✔

4458

    tlen = encoder.pos;

1,114,031✔

4460

  tEncoderClear(&encoder);

1,114,031✔

4461

  return tlen;

1,112,788✔

4464

int32_t tDeserializeSStreamTsResponse(void* buf, int32_t bufLen, void *pBlock) {

557,279✔

4465

  SDecoder decoder = {0};

557,279✔

4466

  int32_t  code = TSDB_CODE_SUCCESS;

557,279✔

4467

  int32_t  lino = 0;

557,279✔

4468

  SSDataBlock *pResBlock = pBlock;

557,279✔

4470

  tDecoderInit(&decoder, buf, bufLen);

557,279✔

4471

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

557,279✔

4473

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, (int64_t*)&pResBlock->info.id.groupId));

1,114,558✔

4474

  int32_t numOfCols = 2;

557,279✔

4475

  if (pResBlock->pDataBlock == NULL) {

557,279✔

4476

    pResBlock->pDataBlock = taosArrayInit_s(sizeof(SColumnInfoData), numOfCols);

557,279✔

4477

    if (pResBlock->pDataBlock == NULL) {

557,279✔

4480

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

1,671,837✔

4481

      SColumnInfoData *pColInfoData = taosArrayGet(pResBlock->pDataBlock, i);

1,114,558✔

4482

      if (pColInfoData == NULL) {

1,114,558✔

4485

      pColInfoData->info.type = TSDB_DATA_TYPE_BIGINT;

1,114,558✔

4486

      pColInfoData->info.bytes = sizeof(int64_t);

1,114,558✔

4489

  int32_t numOfRows = 0;

557,279✔

4490

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &numOfRows));

557,279✔

4491

  TAOS_CHECK_EXIT(blockDataEnsureCapacity(pResBlock, numOfRows));

557,279✔

4492

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

1,370,733✔

4493

    for (int32_t j = 0; j < numOfCols; ++j) {

2,440,362✔

4494

      SColumnInfoData *pColInfoData = taosArrayGet(pResBlock->pDataBlock, j);

1,626,908✔

4495

      if (pColInfoData == NULL) {

1,626,908✔

4498

      int64_t value = 0;

1,626,908✔

4499

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &value));

1,626,908✔

4500

      colDataSetInt64(pColInfoData, i, &value);

1,626,908✔

4504

  pResBlock->info.dataLoad = 1;

557,279✔

4505

  pResBlock->info.rows = numOfRows;

557,279✔

4507

  tEndDecode(&decoder);

557,279✔

4509

_exit:

557,279✔

4510

  tDecoderClear(&decoder);

557,279✔

4511

  return code;

557,279✔

4514

static int32_t encodeData(SEncoder* encoder, void* pBlock, SSHashObj* indexHash) {

2,833,523✔

4515

  int32_t code = TSDB_CODE_SUCCESS;

2,833,523✔

4516

  int32_t lino = 0;

2,833,523✔

4517

  int32_t len = 0;

2,833,523✔

4518

  if (encoder->data == NULL){

2,833,523✔

4519

    len = blockGetEncodeSize(pBlock);

1,417,012✔

4521

    len = blockEncode(pBlock, (char*)(encoder->data + encoder->pos), encoder->size - encoder->pos, blockDataGetNumOfCols(pBlock));

1,416,764✔

4522

    if (len < 0) {

1,416,764✔

4526

  encoder->pos += len;

2,833,776✔

4528

  if (indexHash == NULL) {

2,834,024✔

4529

    goto _exit;

1,333,234✔

4532

  uint32_t pos = encoder->pos;

1,500,790✔

4533

  encoder->pos += sizeof(uint32_t); // reserve space for tables

1,500,542✔

4534

  int32_t tables = 0;

1,500,790✔

4536

  void*   pe = NULL;

1,500,790✔

4537

  int32_t iter = 0;

1,500,790✔

4538

  while ((pe = tSimpleHashIterate(indexHash, pe, &iter)) != NULL) {

3,246,908✔

4539

    SStreamWalDataSlice* pInfo = (SStreamWalDataSlice*)pe;

1,746,366✔

4540

    if (pInfo->gId == -1){

1,746,366✔

4543

    int64_t uid = *(int64_t*)(tSimpleHashGetKey(pe, NULL));

1,746,366✔

4544

    TAOS_CHECK_EXIT(tEncodeI64(encoder, uid));

1,746,118✔

4545

    TAOS_CHECK_EXIT(tEncodeU64(encoder, pInfo->gId));

3,491,988✔

4546

    TAOS_CHECK_EXIT(tEncodeI32(encoder, pInfo->startRowIdx));

3,491,988✔

4547

    TAOS_CHECK_EXIT(tEncodeI32(encoder, pInfo->numRows));

3,492,236✔

4548

    tables++;

1,746,118✔

4550

  uint32_t tmpPos = encoder->pos;

1,500,790✔

4551

  encoder->pos = pos;

1,500,790✔

4552

  TAOS_CHECK_EXIT(tEncodeI32(encoder, tables));

1,500,790✔

4553

  encoder->pos = tmpPos;

1,500,790✔

4554

_exit:

2,833,776✔

4555

  return code;

2,833,776✔

4558

static int32_t encodeBlock(SEncoder* encoder, void* block, SSHashObj* indexHash) {

10,244,886✔

4559

  int32_t  code = TSDB_CODE_SUCCESS;

10,244,886✔

4560

  int32_t  lino = 0;

10,244,886✔

4561

  if (block != NULL && ((SSDataBlock*)block)->info.rows > 0) {

10,244,886✔

4562

    TAOS_CHECK_EXIT(tEncodeI8(encoder, 1));

2,833,771✔

4563

    TAOS_CHECK_EXIT(encodeData(encoder, block, indexHash));

2,833,771✔

4565

    TAOS_CHECK_EXIT(tEncodeI8(encoder, 0));

7,410,875✔

4568

_exit:

7,410,875✔

4569

  return code;

10,244,651✔

4572

int32_t tSerializeSStreamWalDataResponse(void* buf, int32_t bufLen, SSTriggerWalNewRsp* rsp) {

2,559,641✔

4573

  SEncoder encoder = {0};

2,559,641✔

4574

  int32_t  code = TSDB_CODE_SUCCESS;

2,561,470✔

4575

  int32_t  lino = 0;

2,561,470✔

4576

  int32_t  tlen = 0;

2,561,470✔

4578

  tEncoderInit(&encoder, buf, bufLen);

2,561,470✔

4579

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

2,561,470✔

4581

  TAOS_CHECK_EXIT(encodeBlock(&encoder, rsp->dataBlock, rsp->indexHash));

2,561,470✔

4582

  TAOS_CHECK_EXIT(encodeBlock(&encoder, rsp->metaBlock, NULL));

2,560,974✔

4583

  TAOS_CHECK_EXIT(encodeBlock(&encoder, rsp->deleteBlock, NULL));

2,561,470✔

4584

  TAOS_CHECK_EXIT(encodeBlock(&encoder, rsp->tableBlock, NULL));

2,560,985✔

4586

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, rsp->ver));

5,122,940✔

4587

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, rsp->verTime));

5,122,468✔

4588

  tEndEncode(&encoder);

2,560,998✔

4590

_exit:

2,560,985✔

4591

  if (code != TSDB_CODE_SUCCESS) {

2,560,985✔

4594

    tlen = encoder.pos;

2,560,985✔

4596

  tEncoderClear(&encoder);

2,560,985✔

4597

  return tlen;

2,561,233✔

4600

static int32_t decodeBlock(SDecoder* decoder, void* pBlock) {

3,843,093✔

4601

  int32_t  code = TSDB_CODE_SUCCESS;

3,843,093✔

4602

  int32_t  lino = 0;

3,843,093✔

4604

  int8_t hasData = false;

3,843,093✔

4605

  TAOS_CHECK_EXIT(tDecodeI8(decoder, &hasData));

3,843,093✔

4606

  if (hasData) {

3,843,093✔

4607

    TAOS_CHECK_EXIT(pBlock != NULL ? TSDB_CODE_SUCCESS : TSDB_CODE_INVALID_PARA);

666,765✔

4608

    const char* pEndPos = NULL;

666,765✔

4609

    TAOS_CHECK_EXIT(blockDecode(pBlock, (char*)decoder->data + decoder->pos, &pEndPos));

666,765✔

4610

    decoder->pos = (uint8_t*)pEndPos - decoder->data;

666,765✔

4611

  } else if (pBlock != NULL) {

3,176,328✔

4612

    blockDataEmpty(pBlock);

1,331,523✔

4615

_exit:

3,840,351✔

4616

  return code;

3,843,093✔

4619

int32_t tDeserializeSStreamWalDataResponse(void* buf, int32_t bufLen, SSTriggerWalNewRsp* pRsp, SArray* pSlices){

1,281,031✔

4620

  SDecoder     decoder = {0};

1,281,031✔

4621

  int32_t      code = TSDB_CODE_SUCCESS;

1,281,031✔

4622

  int32_t      lino = 0;

1,281,031✔

4623

  SSDataBlock* pBlock = NULL;

1,281,031✔

4625

  tDecoderInit(&decoder, buf, bufLen);

1,281,031✔

4626

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

1,281,031✔

4629

  int8_t hasData = false;

1,281,031✔

4630

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &hasData));

1,281,031✔

4631

  pBlock = pRsp->dataBlock;

1,281,031✔

4632

  if (hasData) {

1,281,031✔

4633

    TAOS_CHECK_EXIT(pBlock != NULL ? TSDB_CODE_SUCCESS : TSDB_CODE_INVALID_PARA);

750,543✔

4634

    const char* pEndPos = NULL;

750,543✔

4635

    TAOS_CHECK_EXIT(blockDecode(pBlock, (char*)decoder.data + decoder.pos, &pEndPos));

750,543✔

4636

    decoder.pos = (uint8_t*)pEndPos - decoder.data;

750,543✔

4638

    int32_t nSlices = 0;

750,543✔

4639

    TAOS_CHECK_EXIT(tDecodeI32(&decoder, &nSlices));

750,543✔

4640

    TAOS_CHECK_EXIT(taosArrayEnsureCap(pSlices, nSlices));

750,543✔

4641

    taosArrayClear(pSlices);

750,543✔

4642

    int64_t  uid = 0;

750,543✔

4643

    uint64_t gid = 0;

750,543✔

4644

    int32_t  startIdx = 0;

750,543✔

4645

    int32_t  numRows = 0;

750,543✔

4646

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

1,623,874✔

4647

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &uid));

873,331✔

4648

      TAOS_CHECK_EXIT(tDecodeU64(&decoder, &gid));

873,331✔

4649

      TAOS_CHECK_EXIT(tDecodeI32(&decoder, &startIdx));

873,331✔

4650

      TAOS_CHECK_EXIT(tDecodeI32(&decoder, &numRows));

873,331✔

4651

      int32_t endIdx = startIdx + numRows;

873,331✔

4652

      int64_t value[3] = {gid, uid, (int64_t)startIdx << 32 | endIdx};

873,331✔

4653

      void*   px = taosArrayPush(pSlices, value);

873,331✔

4654

      if (px == NULL) {

873,331✔

4659

  } else if (pBlock != NULL) {

530,488✔

4660

    blockDataEmpty(pBlock);

3,505✔

4661

    taosArrayClear(pSlices);

3,505✔

4664

  TAOS_CHECK_EXIT(decodeBlock(&decoder, pRsp->metaBlock));

1,281,031✔

4665

  TAOS_CHECK_EXIT(decodeBlock(&decoder, pRsp->deleteBlock));

1,281,031✔

4666

  TAOS_CHECK_EXIT(decodeBlock(&decoder, pRsp->tableBlock));

1,281,031✔

4668

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRsp->ver));

2,562,062✔

4669

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRsp->verTime));

2,562,062✔

4671

  tEndDecode(&decoder);

1,281,031✔

4673

_exit:

1,281,031✔

4674

  if (code != TSDB_CODE_SUCCESS) {

1,281,031✔

4677

  tDecoderClear(&decoder);

1,281,031✔

4678

  return code;

1,281,031✔

taosdata / TDengine / #5011

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

taosdata / TDengine / #5011

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

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