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

27

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

148,668✔

28

  int32_t code = 0;

148,668✔

29

  int32_t lino = 0;

148,668✔

30

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

297,336✔

31

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

297,336✔

32

  switch (pReq->type) {

148,668✔

33

    case STREAM_MGMT_REQ_TRIGGER_ORIGTBL_READER: {

142,640✔

34

      if (pReq->cont.pReqs) {

142,640✔

35

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

142,640✔

36

        TAOS_CHECK_EXIT(tEncodeI32(pEncoder, num));

142,640✔

37

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

555,830✔

38

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

413,190✔

39

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

826,380✔

40

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

826,380✔

45

      break;

142,640✔

47

    case STREAM_MGMT_REQ_RUNNER_ORIGTBL_READER: {

6,028✔

48

      if (pReq->cont.pReqs) {

6,028✔

49

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

6,028✔

50

        TAOS_CHECK_EXIT(tEncodeI32(pEncoder, num));

6,028✔

51

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

12,056✔

52

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

6,028✔

53

          int32_t vgIdNum = taosArrayGetSize(pDeploy->vgIds);

6,028✔

54

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

12,056✔

55

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

12,056✔

56

          TAOS_CHECK_EXIT(tEncodeI32(pEncoder, vgIdNum));

6,028✔

57

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

12,056✔

58

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

12,056✔

64

      break;

6,028✔

71

_exit:

148,668✔

73

  return code;

148,668✔

76

void tFreeRunnerOReaderDeployReq(void* param) {

9,042✔

77

  SStreamOReaderDeployReq* pReq = (SStreamOReaderDeployReq*)param;

9,042✔

78

  if (pReq) {

9,042✔

79

    taosArrayDestroy(pReq->vgIds);

9,042✔

81

9,042✔

83

void tFreeSStreamMgmtReq(SStreamMgmtReq* pReq) {

297,336✔

84

  if (NULL == pReq) {

297,336✔

85

    return;

74,334✔

88

  switch (pReq->type) {

223,002✔

89

    case STREAM_MGMT_REQ_TRIGGER_ORIGTBL_READER:

213,960✔

90

      taosArrayDestroy(pReq->cont.pReqs);

213,960✔

91

      break;

213,960✔

92

    case STREAM_MGMT_REQ_RUNNER_ORIGTBL_READER:

9,042✔

93

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

9,042✔

94

      break;

9,042✔

101

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

74,334✔

102

  *ppDst = NULL;

74,334✔

104

  if (NULL == pSrc) {

74,334✔

108

  int32_t code = 0, lino = 0;

74,334✔

109

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

74,334✔

110

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

74,334✔

112

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

74,334✔

113

  if (pSrc->cont.pReqs) {

74,334✔

114

    switch (pSrc->type) {

74,334✔

115

      case STREAM_MGMT_REQ_TRIGGER_ORIGTBL_READER:

71,320✔

116

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

71,320✔

117

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

71,320✔

118

        break;

71,320✔

119

      case STREAM_MGMT_REQ_RUNNER_ORIGTBL_READER: {

3,014✔

120

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

3,014✔

121

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

3,014✔

122

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

3,014✔

123

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

6,028✔

124

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

3,014✔

125

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

3,014✔

126

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

3,014✔

127

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

3,014✔

128

          pNew->execId = pReq->execId;

3,014✔

129

          pNew->uid = pReq->uid;

3,014✔

131

        break;

3,014✔

140

  if (code) {

74,334✔

146

  return code;

74,334✔

150

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

74,334✔

151

  int32_t code = 0;

74,334✔

152

  int32_t lino = 0;

74,334✔

154

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

148,668✔

155

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

148,668✔

156

  switch (pReq->type) {

74,334✔

157

    case STREAM_MGMT_REQ_TRIGGER_ORIGTBL_READER: {

71,320✔

158

      int32_t num = 0;

71,320✔

159

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &num));

71,320✔

160

      if (num > 0) {

71,320✔

161

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

71,320✔

162

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

71,320✔

163

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

277,915✔

164

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

206,595✔

165

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

206,595✔

166

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

206,595✔

169

      break;

71,320✔

171

    case STREAM_MGMT_REQ_RUNNER_ORIGTBL_READER: {

3,014✔

172

      int32_t num = 0, vgIdNum = 0;

3,014✔

173

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &num));

3,014✔

174

      if (num > 0) {

3,014✔

175

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

3,014✔

176

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

3,014✔

177

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

6,028✔

178

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

3,014✔

179

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

6,028✔

180

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

6,028✔

181

          TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgIdNum));

3,014✔

182

          if (vgIdNum > 0) {

3,014✔

183

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

3,014✔

184

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

3,014✔

186

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

6,028✔

187

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

3,014✔

188

            TAOS_CHECK_EXIT(tDecodeI32(pDecoder, vgId));

3,014✔

192

      break;

3,014✔

199

_exit:

74,334✔

201

  return code;

74,334✔

204

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

129,076,440✔

205

  int32_t code = 0;

129,076,440✔

208

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

258,152,880✔

209

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

258,152,880✔

210

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

258,152,880✔

212

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

258,152,880✔

213

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

258,152,880✔

214

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

258,152,880✔

215

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

258,152,880✔

217

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

258,152,880✔

218

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

258,152,880✔

219

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

258,152,880✔

220

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

258,152,880✔

221

  if (pTask->pMgmtReq) {

129,076,440✔

222

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, 1));

148,668✔

223

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

148,668✔

225

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, 0));

128,927,772✔

228

_exit:

128,927,772✔

230

  return code;

129,076,440✔

234

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

62,089,213✔

235

  int32_t code = 0;

62,089,213✔

238

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

124,178,426✔

239

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

124,178,426✔

240

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

124,178,426✔

242

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

124,178,426✔

243

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

124,178,426✔

244

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

124,178,426✔

245

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

124,178,426✔

247

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

124,178,426✔

248

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

124,178,426✔

249

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

124,178,426✔

250

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

124,178,426✔

251

  int32_t req = 0;

62,089,213✔

252

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &req));

62,089,213✔

253

  if (req) {

62,089,213✔

254

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

74,334✔

255

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

74,334✔

256

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

74,334✔

259

_exit:

62,089,213✔

261

  return code;

62,089,213✔

293

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

7,785,518✔

294

  int32_t code = 0;

7,785,518✔

297

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

15,571,036✔

298

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

15,571,036✔

299

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

15,571,036✔

300

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

15,571,036✔

301

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

15,571,036✔

303

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

7,785,518✔

304

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, recalcNum));

7,785,518✔

305

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

7,785,518✔

310

_exit:

7,785,518✔

312

  return code;

7,785,518✔

315

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

3,689,146✔

316

  int32_t code = 0;

3,689,146✔

319

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

7,378,292✔

320

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

7,378,292✔

321

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

7,378,292✔

322

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

7,378,292✔

323

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

7,378,292✔

325

  int32_t recalcNum = 0;

3,689,146✔

326

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &recalcNum));

3,689,146✔

327

  if (recalcNum > 0) {

3,689,146✔

335

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

3,689,146✔

340

_exit:

3,689,146✔

342

  return code;

3,689,146✔

346

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

42,922,596✔

347

  int32_t code = 0;

42,922,596✔

350

  TAOS_CHECK_EXIT(tStartEncode(pEncoder));

42,922,596✔

351

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

85,845,192✔

352

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

85,845,192✔

353

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

85,845,192✔

354

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

85,845,192✔

356

  int32_t vgLeaderNum = taosArrayGetSize(pReq->pVgLeaders);

42,922,596✔

357

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, vgLeaderNum));

42,922,596✔

358

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

174,714,002✔

359

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

131,791,406✔

360

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *vgId));

263,582,812✔

363

  int32_t statusNum = taosArrayGetSize(pReq->pStreamStatus);

42,922,596✔

364

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, statusNum));

42,922,596✔

365

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

166,793,642✔

366

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

123,871,046✔

367

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

123,871,046✔

370

  int32_t reqNum = taosArrayGetSize(pReq->pStreamReq);

42,922,596✔

371

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, reqNum));

42,922,596✔

372

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

43,071,264✔

373

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

148,668✔

374

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *idx));

297,336✔

377

  int32_t triggerNum = taosArrayGetSize(pReq->pTriggerStatus);

42,922,596✔

378

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, triggerNum));

42,922,596✔

379

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

50,708,114✔

380

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

7,785,518✔

381

    TAOS_CHECK_EXIT(tEncodeSSTriggerRuntimeStatus(pEncoder, pTrigger));

7,785,518✔

384

  tEndEncode(pEncoder);

42,922,596✔

386

_exit:

42,922,596✔

387

  if (code) {

42,922,596✔

390

    return pEncoder->pos;

42,922,596✔

394

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

20,614,072✔

395

  int32_t code = 0;

20,614,072✔

398

  TAOS_CHECK_EXIT(tStartDecode(pDecoder));

20,614,072✔

399

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

41,228,144✔

400

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

41,228,144✔

401

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

41,228,144✔

402

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

41,228,144✔

404

  int32_t vgLearderNum = 0;

20,614,072✔

405

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgLearderNum));

20,614,072✔

406

  if (vgLearderNum > 0) {

20,614,072✔

407

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

16,348,791✔

408

    if (NULL == pReq->pVgLeaders) {

16,348,791✔

413

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

85,058,705✔

414

    int32_t vgId = 0;

64,444,633✔

415

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgId));

64,444,633✔

416

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

128,889,266✔

423

  int32_t statusNum = 0;

20,614,072✔

424

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &statusNum));

20,614,072✔

425

  if (statusNum > 0) {

20,614,072✔

426

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

1,812,707✔

427

    if (NULL == pReq->pStreamStatus) {

1,812,707✔

432

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

80,101,946✔

433

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

59,487,874✔

434

    if (NULL == pTask) {

59,487,874✔

438

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

59,487,874✔

442

  int32_t reqNum = 0;

20,614,072✔

443

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &reqNum));

20,614,072✔

444

  if (reqNum > 0) {

20,614,072✔

445

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

20,855✔

446

    if (NULL == pReq->pStreamReq) {

20,855✔

451

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

20,688,406✔

452

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

74,334✔

453

    if (NULL == pIdx) {

74,334✔

457

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, pIdx));

74,334✔

461

  int32_t triggerNum = 0;

20,614,072✔

462

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &triggerNum));

20,614,072✔

463

  if (triggerNum > 0) {

20,614,072✔

464

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

829,079✔

465

    if (NULL == pReq->pTriggerStatus) {

829,079✔

470

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

24,303,218✔

471

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

3,689,146✔

472

    if (NULL == pStatus) {

3,689,146✔

476

    TAOS_CHECK_EXIT(tDecodeSSTriggerRuntimeStatus(pDecoder, pStatus));

3,689,146✔

480

  tEndDecode(pDecoder);

20,614,072✔

482

_exit:

20,614,072✔

483

  return code;

20,614,072✔

486

void tFreeSSTriggerRuntimeStatus(void* param) {

7,581,905✔

487

  SSTriggerRuntimeStatus* pStatus = (SSTriggerRuntimeStatus*)param;

7,581,905✔

488

  if (NULL == pStatus) {

7,581,905✔

491

  taosArrayDestroy(pStatus->userRecalcs);

7,581,905✔

494

void tCleanupStreamHbMsg(SStreamHbMsg* pMsg, bool deepClean) {

123,195,444✔

495

  if (pMsg == NULL) {

123,195,444✔

499

  taosArrayDestroy(pMsg->pVgLeaders);

123,195,444✔

500

  if (deepClean) {

123,195,444✔

501

    int32_t reqNum = taosArrayGetSize(pMsg->pStreamReq);

123,195,444✔

502

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

123,344,112✔

503

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

148,668✔

504

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

148,668✔

505

      if (NULL == pTask) {

148,668✔

509

      tFreeSStreamMgmtReq(pTask->pMgmtReq);

148,668✔

510

      taosMemoryFree(pTask->pMgmtReq);

148,668✔

513

  taosArrayDestroy(pMsg->pStreamReq);

123,195,444✔

514

  taosArrayDestroy(pMsg->pStreamStatus);

123,195,444✔

515

  taosArrayDestroyEx(pMsg->pTriggerStatus, tFreeSSTriggerRuntimeStatus);

123,195,444✔

518

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

768,890✔

519

  int32_t code = 0;

768,890✔

522

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

1,537,780✔

523

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

1,537,780✔

524

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

1,537,780✔

525

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

1,537,780✔

526

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

1,537,780✔

527

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

1,537,780✔

528

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

1,537,780✔

529

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

1,537,780✔

530

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

1,537,780✔

532

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

1,537,780✔

533

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

1,537,780✔

535

_exit:

768,890✔

537

  return code;

768,890✔

540

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

880,236✔

541

  int32_t code = 0;

880,236✔

544

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

1,760,472✔

545

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

1,760,472✔

547

_exit:

880,236✔

549

  return code;

880,236✔

553

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

1,649,126✔

554

  int32_t code = 0;

1,649,126✔

557

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

3,298,252✔

558

  if (pMsg->triggerReader) {

1,649,126✔

559

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

768,890✔

561

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

880,236✔

564

_exit:

880,236✔

566

  return code;

1,649,126✔

569

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

2,314,336✔

570

  int32_t code = 0;

2,314,336✔

573

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

4,628,672✔

574

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

4,628,672✔

575

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

2,314,336✔

577

_exit:

2,314,336✔

579

  return code;

2,314,336✔

582

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

1,510,494✔

583

  int32_t code = 0;

1,510,494✔

586

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

1,510,494✔

587

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

3,020,988✔

589

_exit:

1,510,494✔

591

  return code;

1,510,494✔

595

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

510,698✔

596

  int32_t code = 0;

510,698✔

599

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

1,021,396✔

600

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

1,021,396✔

601

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

1,021,396✔

602

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

1,021,396✔

603

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

1,021,396✔

604

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

1,021,396✔

605

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

1,021,396✔

606

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

1,021,396✔

607

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

1,021,396✔

608

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

1,021,396✔

609

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

510,698✔

610

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

1,021,396✔

612

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

510,698✔

613

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, addrSize));

510,698✔

614

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

633,378✔

615

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

122,680✔

616

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

245,360✔

618

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

1,021,396✔

619

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

1,021,396✔

620

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

1,021,396✔

622

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

1,021,396✔

623

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

1,021,396✔

624

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

1,021,396✔

625

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

1,021,396✔

626

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

1,021,396✔

628

  switch (pMsg->triggerType) {

510,698✔

629

    case WINDOW_TYPE_SESSION: {

18,644✔

631

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

37,288✔

632

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

37,288✔

633

      break;

18,644✔

635

    case WINDOW_TYPE_STATE: {

186,470✔

641

      int32_t slotNum = pMsg->trigger.stateWin.pSlotIds == NULL ?

372,940✔

642

        0 : taosArrayGetSize(pMsg->trigger.stateWin.pSlotIds);

186,470✔

643

      TAOS_CHECK_EXIT(

186,470✔

645

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, slotNum));

186,470✔

646

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

398,110✔

647

        TAOS_CHECK_EXIT(tEncodeI16(

423,280✔

652

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

372,940✔

653

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

372,940✔

654

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

372,940✔

655

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

372,940✔

656

      int32_t stateWindowZerothLen =

186,470✔

657

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

186,470✔

658

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

372,940✔

659

      int32_t stateWindowExprLen =

186,470✔

660

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

186,470✔

661

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

372,940✔

662

      break;

186,470✔

664

    case WINDOW_TYPE_INTERVAL: {

180,716✔

666

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

361,432✔

667

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

361,432✔

668

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

361,432✔

669

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

361,432✔

670

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

361,432✔

671

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

361,432✔

672

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

361,432✔

673

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

361,432✔

674

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

361,432✔

675

      break;

180,716✔

677

    case WINDOW_TYPE_EVENT: {

56,176✔

679

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

56,176✔

680

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

56,176✔

682

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

112,352✔

683

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

112,352✔

684

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

112,352✔

685

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

112,352✔

686

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

112,352✔

687

      break;

56,176✔

689

    case WINDOW_TYPE_COUNT: {

52,254✔

691

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

52,254✔

692

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

104,508✔

694

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

104,508✔

695

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

104,508✔

696

      break;

52,254✔

698

    case WINDOW_TYPE_PERIOD: {

16,438✔

700

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

32,876✔

701

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

32,876✔

702

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

32,876✔

703

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

32,876✔

704

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

32,876✔

705

      break;

16,438✔

712

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

1,021,396✔

713

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

1,021,396✔

714

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

1,021,396✔

715

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

1,021,396✔

716

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

1,021,396✔

717

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

1,021,396✔

718

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

510,698✔

719

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

1,021,396✔

720

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

510,698✔

721

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

1,021,396✔

722

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

510,698✔

723

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

1,021,396✔

725

  int32_t readerNum = taosArrayGetSize(pMsg->readerList);

510,698✔

726

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, readerNum));

510,698✔

727

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

1,142,032✔

728

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

631,334✔

729

    TAOS_CHECK_EXIT(tEncodeSStreamTaskAddr(pEncoder, pAddr));

631,334✔

732

  int32_t runnerNum = taosArrayGetSize(pMsg->runnerList);

510,698✔

733

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, runnerNum));

510,698✔

734

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

2,021,192✔

735

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

1,510,494✔

736

    TAOS_CHECK_EXIT(tEncodeSStreamRunnerTarget(pEncoder, pTarget));

1,510,494✔

739

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

1,021,396✔

740

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

1,021,396✔

741

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

1,021,396✔

742

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

1,021,396✔

744

_exit:

510,698✔

746

  return code;

510,698✔

750

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

8,701,398✔

751

  int32_t code = 0;

8,701,398✔

754

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

17,402,796✔

755

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

17,402,796✔

756

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

17,402,796✔

757

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

17,402,796✔

758

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

17,402,796✔

759

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

17,402,796✔

761

_exit:

8,701,398✔

763

  return code;

8,701,398✔

767

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

1,774,860✔

768

  int32_t code = 0;

1,774,860✔

771

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

3,549,720✔

772

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

3,549,720✔

773

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

3,549,720✔

774

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

3,549,720✔

775

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

3,549,720✔

776

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

3,549,720✔

777

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

3,549,720✔

778

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

3,549,720✔

780

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

1,774,860✔

781

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, addrSize));

1,774,860✔

782

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

2,121,300✔

783

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

346,440✔

784

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

692,880✔

786

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

3,549,720✔

788

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

1,774,860✔

789

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, outColNum));

1,774,860✔

790

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

9,035,892✔

791

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

7,261,032✔

792

    TAOS_CHECK_EXIT(tSerializeSFieldWithOptions(pEncoder, pCol));

7,261,032✔

795

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

1,774,860✔

796

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, outTagNum));

1,774,860✔

797

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

3,215,226✔

798

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

1,440,366✔

799

    TAOS_CHECK_EXIT(tSerializeSFieldWithOptions(pEncoder, pTag));

1,440,366✔

802

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

3,549,720✔

803

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

3,549,720✔

805

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

3,549,720✔

806

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

3,549,720✔

808

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

1,774,860✔

809

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, forceOutColsSize));

1,774,860✔

810

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

1,934,442✔

811

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

159,582✔

812

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

159,582✔

814

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

319,164✔

815

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

319,164✔

816

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

319,164✔

817

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

319,164✔

818

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

319,164✔

821

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

3,549,720✔

824

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

1,774,860✔

825

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, colCidsSize));

1,774,860✔

826

  if (colCidsSize > 0) {

1,774,860✔

827

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

113,634✔

828

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

83,976✔

829

      TAOS_CHECK_EXIT(tEncodeI16(pEncoder, *pCid));

167,952✔

833

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

1,774,860✔

834

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, tagCidsSize));

1,774,860✔

835

  if (tagCidsSize > 0) {

1,774,860✔

836

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

57,030✔

837

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

33,690✔

838

      TAOS_CHECK_EXIT(tEncodeI16(pEncoder, *pCid));

67,380✔

842

_exit:

1,774,860✔

844

  return code;

1,774,860✔

847

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

3,934,684✔

848

  int32_t code = 0;

3,934,684✔

851

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

3,934,684✔

852

  switch (pTask->task.type) {

3,934,684✔

853

    case STREAM_READER_TASK:

1,649,126✔

854

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

1,649,126✔

855

      break;

1,649,126✔

856

    case STREAM_TRIGGER_TASK:

510,698✔

857

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

510,698✔

858

      break;

510,698✔

859

    case STREAM_RUNNER_TASK:

1,774,860✔

860

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

1,774,860✔

861

      break;

1,774,860✔

867

_exit:

3,934,684✔

869

  return code;

3,934,684✔

873

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

668,480✔

874

  int32_t code = 0;

668,480✔

877

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

1,336,960✔

879

  int32_t readerNum = taosArrayGetSize(pStream->readerTasks);

668,480✔

880

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, readerNum));

668,480✔

881

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

2,317,606✔

882

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

1,649,126✔

883

    TAOS_CHECK_EXIT(tEncodeSStmTaskDeploy(pEncoder, pDeploy));

1,649,126✔

886

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

1,336,960✔

887

  if (pStream->triggerTask) {

668,480✔

888

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

510,698✔

891

  int32_t runnerNum = taosArrayGetSize(pStream->runnerTasks);

668,480✔

892

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, runnerNum));

668,480✔

893

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

2,443,340✔

894

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

1,774,860✔

895

    TAOS_CHECK_EXIT(tEncodeSStmTaskDeploy(pEncoder, pDeploy));

1,774,860✔

898

_exit:

668,480✔

900

  return code;

668,480✔

903

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

1,270,710✔

904

  int32_t code = 0;

1,270,710✔

905

  int32_t lino = 0;

1,270,710✔

907

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

2,541,420✔

909

_exit:

1,270,710✔

910

  return code;

1,270,710✔

913

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

635,490✔

914

  int32_t code = 0;

635,490✔

917

  int32_t type = 0;

635,490✔

918

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &type));

635,490✔

919

  pMsg->msgType = type;

635,490✔

921

_exit:

635,490✔

922

  return code;

635,490✔

925

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

500,666✔

926

  int32_t code = 0;

500,666✔

929

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

500,666✔

931

_exit:

500,666✔

933

  return code;

500,666✔

936

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

500,666✔

937

  int32_t code = 0;

500,666✔

940

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

500,666✔

941

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

500,666✔

943

_exit:

500,666✔

945

  return code;

500,666✔

948

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

612,880✔

949

  int32_t code = 0;

612,880✔

952

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

612,880✔

953

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

1,225,760✔

954

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

1,225,760✔

956

_exit:

612,880✔

958

  return code;

612,880✔

961

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

612,880✔

962

  int32_t code = 0;

612,880✔

965

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

612,880✔

966

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

612,880✔

968

_exit:

612,880✔

970

  return code;

612,880✔

974

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

8,496✔

975

  int32_t code = 0;

8,496✔

978

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

16,992✔

979

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

16,992✔

980

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

16,992✔

982

_exit:

8,496✔

984

  return code;

8,496✔

987

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

157,164✔

988

  int32_t code = 0;

157,164✔

991

  switch (msgType) {

157,164✔

992

    case STREAM_MSG_ORIGTBL_READER_INFO: {

142,640✔

993

      int32_t vgNum = taosArrayGetSize(pRsp->vgIds);

142,640✔

994

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, vgNum));

142,640✔

996

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

555,830✔

997

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

413,190✔

998

        TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *vgId));

826,380✔

1001

      int32_t readerNum = taosArrayGetSize(pRsp->readerList);

142,640✔

1002

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, readerNum));

142,640✔

1004

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

278,326✔

1005

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

135,686✔

1006

        TAOS_CHECK_EXIT(tEncodeSStreamTaskAddr(pEncoder, addr));

135,686✔

1008

      break;

142,640✔

1020

    case STREAM_MSG_USER_RECALC: {

8,496✔

1021

      int32_t recalcNum = taosArrayGetSize(pRsp->recalcList);

8,496✔

1022

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, recalcNum));

8,496✔

1024

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

16,992✔

1025

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

8,496✔

1026

        TAOS_CHECK_EXIT(tEncodeSStreamRecalcReq(pEncoder, recalc));

8,496✔

1028

      break;

8,496✔

1030

    case STREAM_MSG_RUNNER_ORIGTBL_READER: {

6,028✔

1031

      int32_t rspNum = taosArrayGetSize(pRsp->execRspList);

6,028✔

1032

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, rspNum));

6,028✔

1034

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

12,056✔

1035

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

6,028✔

1036

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

12,056✔

1037

        int32_t vgNum = taosArrayGetSize(pDeployRsp->vgList);

6,028✔

1038

        TAOS_CHECK_EXIT(tEncodeI32(pEncoder, vgNum));

6,028✔

1039

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

12,056✔

1040

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

6,028✔

1043

      break;

6,028✔

1049

_exit:

157,164✔

1051

  return code;

157,164✔

1054

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

157,164✔

1055

  int32_t code = 0;

157,164✔

1058

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

157,164✔

1059

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

314,328✔

1060

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

314,328✔

1061

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

157,164✔

1062

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

157,164✔

1064

_exit:

157,164✔

1066

  return code;

157,164✔

1070

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

40,860,438✔

1071

  int32_t code = 0;

40,860,438✔

1074

  TAOS_CHECK_EXIT(tStartEncode(pEncoder));

40,860,438✔

1075

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

81,720,876✔

1076

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

40,860,438✔

1077

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, deployNum));

40,860,438✔

1078

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

41,528,918✔

1079

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

668,480✔

1080

    TAOS_CHECK_EXIT(tEncodeSStmStreamDeploy(pEncoder, pStream));

668,480✔

1083

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

40,860,438✔

1084

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, startNum));

40,860,438✔

1085

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

41,361,104✔

1086

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

500,666✔

1087

    TAOS_CHECK_EXIT(tEncodeSStreamTaskStart(pEncoder, pTask));

500,666✔

1090

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

81,720,876✔

1091

  if (!pRsp->undeploy.undeployAll) {

40,860,438✔

1092

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

40,860,438✔

1093

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, undeployNum));

40,860,438✔

1094

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

41,473,318✔

1095

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

612,880✔

1096

      TAOS_CHECK_EXIT(tEncodeSStreamTaskUndeploy(pEncoder, pTask));

612,880✔

1100

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

40,860,438✔

1101

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, rspNum));

40,860,438✔

1102

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

41,017,602✔

1103

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

157,164✔

1104

    TAOS_CHECK_EXIT(tEncodeSStreamMgmtRsp(pEncoder, pMgmtRsp));

157,164✔

1107

_exit:

40,860,438✔

1109

  tEndEncode(pEncoder);

40,860,438✔

1111

  return code;

40,860,438✔

1114

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

384,146✔

1115

  int32_t code = 0;

384,146✔

1118

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

768,292✔

1119

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

768,292✔

1120

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

768,292✔

1121

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

768,292✔

1122

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

768,292✔

1123

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

768,292✔

1124

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

768,292✔

1125

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

768,292✔

1126

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

768,292✔

1127

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

768,292✔

1128

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

768,292✔

1130

_exit:

384,146✔

1132

  return code;

384,146✔

1136

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

439,819✔

1137

  int32_t code = 0;

439,819✔

1140

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

879,638✔

1141

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

879,638✔

1143

_exit:

439,819✔

1145

  return code;

439,819✔

1149

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

823,965✔

1150

  int32_t code = 0;

823,965✔

1153

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

1,647,930✔

1154

  if (pMsg->triggerReader) {

823,965✔

1155

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

384,146✔

1157

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

439,819✔

1160

_exit:

439,819✔

1162

  return code;

823,965✔

1166

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

1,156,371✔

1167

  int32_t code = 0;

1,156,371✔

1170

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

2,312,742✔

1171

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

2,312,742✔

1172

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

1,156,371✔

1174

_exit:

1,156,371✔

1176

  return code;

1,156,371✔

1180

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

754,554✔

1181

  int32_t code = 0;

754,554✔

1184

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

754,554✔

1185

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

1,509,108✔

1187

_exit:

754,554✔

1189

  return code;

754,554✔

1193

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

255,118✔

1194

  int32_t code = 0;

255,118✔

1197

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

510,236✔

1198

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

510,236✔

1199

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

510,236✔

1200

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

510,236✔

1201

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

510,236✔

1202

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

510,236✔

1203

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

510,236✔

1204

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

510,236✔

1205

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

510,236✔

1206

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

510,236✔

1207

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

510,236✔

1209

  int32_t addrSize = 0;

255,118✔

1210

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &addrSize));

255,118✔

1211

  if (addrSize > 0) {

255,118✔

1212

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

61,340✔

1213

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

61,340✔

1215

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

316,458✔

1216

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

61,340✔

1217

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

61,340✔

1219

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

510,236✔

1220

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

510,236✔

1221

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

510,236✔

1223

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

510,236✔

1224

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

510,236✔

1225

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

510,236✔

1226

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

510,236✔

1227

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

510,236✔

1229

  switch (pMsg->triggerType) {

255,118✔

1230

    case WINDOW_TYPE_SESSION:

9,322✔

1232

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

18,644✔

1233

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

18,644✔

1234

      break;

9,322✔

1235

    case WINDOW_TYPE_STATE: {

92,932✔

1242

      int16_t firstI16 = 0;

92,932✔

1243

      TAOS_CHECK_EXIT(tDecodeI16(pDecoder, &firstI16));

92,932✔

1244

      if (firstI16 == STATE_WIN_SLOT_SENTINEL_V2) {

92,932✔

1246

        int32_t slotNum = 0;

92,932✔

1247

        TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &slotNum));

92,932✔

1248

        if (slotNum > 0) {

92,932✔

1249

          pMsg->trigger.stateWin.pSlotIds = taosArrayInit(slotNum, sizeof(int16_t));

92,932✔

1250

          TSDB_CHECK_NULL(pMsg->trigger.stateWin.pSlotIds, code, lino, _exit, terrno);

92,932✔

1252

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

198,449✔

1253

          int16_t slotId = -1;

105,517✔

1254

          TAOS_CHECK_EXIT(tDecodeI16(pDecoder, &slotId));

105,517✔

1255

          TSDB_CHECK_NULL(taosArrayPush(pMsg->trigger.stateWin.pSlotIds, &slotId), code, lino, _exit, terrno);

211,034✔

1263

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

185,864✔

1264

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

185,864✔

1265

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

185,864✔

1266

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

185,864✔

1267

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

185,864✔

1268

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

185,864✔

1269

      break;

92,932✔

1272

    case WINDOW_TYPE_INTERVAL:

90,430✔

1274

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

180,860✔

1275

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

180,860✔

1276

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

180,860✔

1277

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

180,860✔

1278

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

180,860✔

1279

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

180,860✔

1280

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

180,860✔

1281

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

180,860✔

1282

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

180,860✔

1283

      break;

90,430✔

1285

    case WINDOW_TYPE_EVENT:

28,088✔

1287

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

56,176✔

1288

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

56,176✔

1289

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

56,176✔

1290

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

56,176✔

1291

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

56,176✔

1292

      break;

28,088✔

1294

    case WINDOW_TYPE_COUNT:

26,127✔

1296

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

52,254✔

1298

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

52,254✔

1299

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

52,254✔

1300

      break;

26,127✔

1302

    case WINDOW_TYPE_PERIOD:

8,219✔

1304

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

16,438✔

1305

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

16,438✔

1306

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

16,438✔

1307

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

16,438✔

1308

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

16,438✔

1309

      break;

8,219✔

1315

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

510,236✔

1316

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

510,236✔

1317

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

510,236✔

1318

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

510,236✔

1319

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

510,236✔

1320

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

510,236✔

1321

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

510,236✔

1322

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

510,236✔

1323

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

510,236✔

1325

  int32_t readerNum = 0;

255,118✔

1326

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &readerNum));

255,118✔

1327

  if (readerNum > 0) {

255,118✔

1328

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

254,003✔

1329

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

254,003✔

1331

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

570,592✔

1332

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

315,474✔

1333

    TAOS_CHECK_EXIT(tDecodeSStreamTaskAddr(pDecoder, pAddr));

315,474✔

1336

  int32_t runnerNum = 0;

255,118✔

1337

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &runnerNum));

255,118✔

1338

  if (runnerNum > 0) {

255,118✔

1339

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

251,518✔

1340

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

251,518✔

1342

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

1,009,672✔

1343

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

754,554✔

1344

    TAOS_CHECK_EXIT(tDecodeSStreamRunnerTarget(pDecoder, pTarget));

754,554✔

1347

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

510,236✔

1348

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

510,236✔

1349

  if (!tDecodeIsEnd(pDecoder)) {

255,118✔

1350

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

510,236✔

1352

  if (!tDecodeIsEnd(pDecoder)) {

255,118✔

1353

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

510,236✔

1356

_exit:

255,118✔

1358

  return code;

255,118✔

1363

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

4,326,760✔

1364

  int32_t code = 0;

4,326,760✔

1367

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

4,326,760✔

1368

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

8,653,520✔

1369

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

8,653,520✔

1370

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

8,653,520✔

1371

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

8,653,520✔

1372

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

8,653,520✔

1374

_exit:

4,326,760✔

1376

  return code;

4,326,760✔

1379

void destroySStreamOutCols(void* p){

79,791✔

1380

  if (p == NULL) return;

79,791✔

1381

  SStreamOutCol* col = (SStreamOutCol*)p;

79,791✔

1382

  taosMemoryFreeClear(col->expr);

79,791✔

1385

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

886,766✔

1386

  int32_t code = 0;

886,766✔

1389

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

1,773,532✔

1390

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

1,773,532✔

1391

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

1,773,532✔

1392

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

1,773,532✔

1393

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

1,773,532✔

1394

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

1,773,532✔

1395

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

1,773,532✔

1396

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

1,773,532✔

1398

  int32_t addrSize = 0;

886,766✔

1399

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &addrSize));

886,766✔

1400

  if (addrSize > 0) {

886,766✔

1401

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

173,220✔

1402

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

173,220✔

1404

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

1,059,986✔

1405

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

173,220✔

1406

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

173,220✔

1408

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

1,773,532✔

1410

  int32_t outColNum = 0;

886,766✔

1411

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &outColNum));

886,766✔

1412

  if (outColNum > 0) {

886,766✔

1413

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

886,766✔

1414

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

886,766✔

1416

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

4,493,229✔

1417

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

3,606,463✔

1418

    TAOS_CHECK_EXIT(tDeserializeSFieldWithOptions(pDecoder, pCol));

3,606,463✔

1421

  int32_t outTagNum = 0;

886,766✔

1422

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &outTagNum));

886,766✔

1423

  if (outTagNum > 0) {

886,766✔

1424

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

370,326✔

1425

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

370,326✔

1427

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

1,607,063✔

1428

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

720,297✔

1429

    TAOS_CHECK_EXIT(tDeserializeSFieldWithOptions(pDecoder, pTag));

720,297✔

1432

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

1,773,532✔

1433

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

1,773,532✔

1435

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

1,773,532✔

1436

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

1,773,532✔

1438

  int32_t forceOutColsSize = 0;

886,766✔

1439

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &forceOutColsSize));

886,766✔

1440

  if (forceOutColsSize > 0) {

886,766✔

1441

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

14,340✔

1442

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

14,340✔

1444

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

966,557✔

1445

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

79,791✔

1447

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

159,582✔

1448

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

159,582✔

1449

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

159,582✔

1450

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

159,582✔

1451

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

159,582✔

1454

  if (!tDecodeIsEnd(pDecoder)) {

886,766✔

1455

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

1,773,532✔

1460

  if (!tDecodeIsEnd(pDecoder)) {

886,766✔

1461

    int32_t colCidsSize = 0;

886,766✔

1462

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &colCidsSize));

886,766✔

1463

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

886,766✔

1464

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

14,829✔

1465

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

14,829✔

1466

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

56,817✔

1467

        int16_t cid = 0;

41,988✔

1468

        TAOS_CHECK_EXIT(tDecodeI16(pDecoder, &cid));

41,988✔

1469

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

83,976✔

1476

  if (!tDecodeIsEnd(pDecoder)) {

886,766✔

1477

    int32_t tagCidsSize = 0;

886,766✔

1478

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &tagCidsSize));

886,766✔

1479

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

886,766✔

1480

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

11,670✔

1481

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

11,670✔

1482

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

28,515✔

1483

        int16_t cid = 0;

16,845✔

1484

        TAOS_CHECK_EXIT(tDecodeI16(pDecoder, &cid));

16,845✔

1485

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

33,690✔

1492

_exit:

885,011✔

1494

  return code;

886,766✔

1497

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

1,965,849✔

1498

  int32_t code = 0;

1,965,849✔

1501

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

1,965,849✔

1502

  switch (pTask->task.type) {

1,965,849✔

1503

    case STREAM_READER_TASK:

823,965✔

1504

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

823,965✔

1505

      break;

823,965✔

1506

    case STREAM_TRIGGER_TASK:

255,118✔

1507

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

255,118✔

1508

      break;

255,118✔

1509

    case STREAM_RUNNER_TASK:

886,766✔

1510

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

886,766✔

1511

      break;

886,766✔

1517

_exit:

1,965,849✔

1519

  return code;

1,965,849✔

1523

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

334,082✔

1524

  int32_t code = 0;

334,082✔

1527

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

668,164✔

1529

  int32_t readerNum = 0;

334,082✔

1530

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &readerNum));

334,082✔

1531

  if (readerNum > 0) {

334,082✔

1532

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

321,281✔

1533

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

321,281✔

1535

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

1,158,047✔

1536

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

823,965✔

1537

    TAOS_CHECK_EXIT(tDecodeSStmTaskDeploy(pDecoder, pTask));

823,965✔

1540

  int32_t triggerTask = 0;

334,082✔

1541

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &triggerTask));

334,082✔

1542

  if (triggerTask) {

334,082✔

1543

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

255,118✔

1544

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

255,118✔

1545

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

255,118✔

1548

  int32_t runnerNum = 0;

334,082✔

1549

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &runnerNum));

334,082✔

1550

  if (runnerNum > 0) {

334,082✔

1551

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

260,575✔

1552

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

260,575✔

1554

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

1,220,848✔

1555

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

886,766✔

1556

    TAOS_CHECK_EXIT(tDecodeSStmTaskDeploy(pDecoder, pTask));

886,766✔

1559

_exit:

334,082✔

1561

  return code;

334,082✔

1565

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

250,405✔

1566

  int32_t code = 0;

250,405✔

1569

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

250,405✔

1571

_exit:

250,405✔

1573

  return code;

250,405✔

1577

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

250,405✔

1578

  int32_t code = 0;

250,405✔

1581

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

250,405✔

1582

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

250,405✔

1584

_exit:

250,405✔

1586

  return code;

250,405✔

1590

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

306,503✔

1591

  int32_t code = 0;

306,503✔

1594

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

306,503✔

1595

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

613,006✔

1596

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

613,006✔

1598

_exit:

306,503✔

1600

  return code;

306,503✔

1604

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

306,503✔

1605

  int32_t code = 0;

306,503✔

1608

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

306,503✔

1609

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

306,503✔

1611

_exit:

306,503✔

1613

  return code;

306,503✔

1616

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

4,248✔

1617

  int32_t code = 0;

4,248✔

1620

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

8,496✔

1621

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

8,496✔

1622

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

8,496✔

1624

_exit:

4,248✔

1626

  return code;

4,248✔

1629

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

78,582✔

1630

  int32_t code = 0;

78,582✔

1633

  switch (msgType) {

78,582✔

1634

    case STREAM_MSG_ORIGTBL_READER_INFO: {

71,320✔

1635

      int32_t vgNum = 0;

71,320✔

1636

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgNum));

71,320✔

1637

      if (vgNum > 0) {

71,320✔

1638

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

71,320✔

1639

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

71,320✔

1641

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

277,915✔

1642

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

206,595✔

1643

        TAOS_CHECK_EXIT(tDecodeI32(pDecoder, vgId));

206,595✔

1646

      int32_t readerNum = 0;

71,320✔

1647

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &readerNum));

71,320✔

1648

      if (readerNum > 0) {

71,320✔

1649

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

59,001✔

1650

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

59,001✔

1652

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

139,163✔

1653

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

67,843✔

1654

        TAOS_CHECK_EXIT(tDecodeSStreamTaskAddr(pDecoder, addr));

67,843✔

1656

      break;

71,320✔

1671

    case STREAM_MSG_USER_RECALC: {

4,248✔

1672

      int32_t recalcNum = 0;

4,248✔

1673

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &recalcNum));

4,248✔

1674

      if (recalcNum > 0) {

4,248✔

1675

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

4,248✔

1676

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

4,248✔

1678

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

8,496✔

1679

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

4,248✔

1680

        TAOS_CHECK_EXIT(tDecodeSStreamRecalcReq(pDecoder, recalc));

4,248✔

1682

      break;

4,248✔

1684

    case STREAM_MSG_RUNNER_ORIGTBL_READER: {

3,014✔

1685

      int32_t rspNum = 0, vgNum = 0;

3,014✔

1686

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &rspNum));

3,014✔

1687

      if (rspNum > 0) {

3,014✔

1688

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

3,014✔

1689

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

3,014✔

1691

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

6,028✔

1692

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

3,014✔

1693

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

6,028✔

1694

        TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgNum));

3,014✔

1695

        if (vgNum > 0) {

3,014✔

1696

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

3,014✔

1697

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

3,014✔

1699

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

6,028✔

1700

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

3,014✔

1701

          TAOS_CHECK_EXIT(tDecodeSStreamTaskAddr(pDecoder, pAddr));

3,014✔

1704

      break;

3,014✔

1710

_exit:

78,582✔

1712

  return code;

78,582✔

1716

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

78,582✔

1717

  int32_t code = 0;

78,582✔

1720

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

78,582✔

1721

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

157,164✔

1722

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

157,164✔

1723

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

78,582✔

1724

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

78,582✔

1726

_exit:

78,582✔

1728

  return code;

78,582✔

1731

void tFreeSStreamOReaderDeployRsp(void* param) {

6,028✔

1732

  if (NULL == param) {

6,028✔

1736

  SStreamOReaderDeployRsp* pRsp = (SStreamOReaderDeployRsp*)param;

6,028✔

1737

  taosArrayDestroy(pRsp->vgList);

6,028✔

1740

void tFreeSStreamMgmtRsp(void* param) {

157,164✔

1741

  if (NULL == param) {

157,164✔

1745

  SStreamMgmtRsp* pRsp = (SStreamMgmtRsp*)param;

157,164✔

1747

  taosArrayDestroy(pRsp->cont.vgIds);

157,164✔

1748

  taosArrayDestroy(pRsp->cont.readerList);

157,164✔

1749

  taosArrayDestroy(pRsp->cont.runnerList);

157,164✔

1750

  taosArrayDestroy(pRsp->cont.recalcList);

157,164✔

1751

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

157,164✔

1754

void tFreeSStreamReaderDeployMsg(SStreamReaderDeployMsg* pReader) {

823,965✔

1755

  if (NULL == pReader) {

823,965✔

1759

  if (pReader->triggerReader) {

823,965✔

1760

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

384,146✔

1761

    taosMemoryFree(pMsg->triggerTblName);

384,146✔

1762

    taosMemoryFree(pMsg->partitionCols);

384,146✔

1763

    taosMemoryFree(pMsg->triggerCols);

384,146✔

1764

    taosMemoryFree(pMsg->triggerScanPlan);

384,146✔

1765

    taosMemoryFree(pMsg->calcCacheScanPlan);

384,146✔

1767

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

439,819✔

1768

    taosMemoryFree(pMsg->calcScanPlan);

439,819✔

1780

void tFreeSStreamTriggerDeployMsg(SStreamTriggerDeployMsg* pTrigger) {

255,118✔

1781

  if (NULL == pTrigger) {

255,118✔

1785

  taosArrayDestroyEx(pTrigger->pNotifyAddrUrls, tFreeStreamNotifyUrl);

255,118✔

1786

  switch (pTrigger->triggerType) {

255,118✔

1787

    case WINDOW_TYPE_STATE:

92,932✔

1788

      taosArrayDestroy(pTrigger->trigger.stateWin.pSlotIds);

92,932✔

1789

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

92,932✔

1790

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

92,932✔

1791

      break;

92,932✔

1792

    case WINDOW_TYPE_EVENT:

28,088✔

1793

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

28,088✔

1794

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

28,088✔

1795

      break;

28,088✔

1796

    case WINDOW_TYPE_COUNT:

26,127✔

1797

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

26,127✔

1798

      break;

26,127✔

1799

    default:

107,971✔

1800

      break;

107,971✔

1803

  taosMemoryFree(pTrigger->partitionCols);

255,118✔

1804

  taosMemoryFree(pTrigger->triggerPrevFilter);

255,118✔

1805

  taosMemoryFree(pTrigger->triggerScanPlan);

255,118✔

1806

  taosMemoryFree(pTrigger->calcCacheScanPlan);

255,118✔

1808

  taosArrayDestroy(pTrigger->readerList);

255,118✔

1809

  taosArrayDestroy(pTrigger->runnerList);

255,118✔

1810

  taosMemoryFree(pTrigger->streamName);

255,118✔

1822

void tFreeSStreamRunnerDeployMsg(SStreamRunnerDeployMsg* pRunner) {

886,766✔

1823

  if (NULL == pRunner) {

886,766✔

1827

  taosMemoryFree(pRunner->streamName);

886,766✔

1828

  taosMemoryFree(pRunner->pPlan);

886,766✔

1829

  taosMemoryFree(pRunner->outDBFName);

886,766✔

1830

  taosMemoryFree(pRunner->outTblName);

886,766✔

1832

  taosArrayDestroyEx(pRunner->pNotifyAddrUrls, tFreeStreamNotifyUrl);

886,766✔

1833

  taosArrayDestroy(pRunner->outCols);

886,766✔

1834

  taosArrayDestroy(pRunner->outTags);

886,766✔

1836

  taosMemoryFree(pRunner->subTblNameExpr);

886,766✔

1837

  taosMemoryFree(pRunner->tagValueExpr);

886,766✔

1838

  taosArrayDestroyEx(pRunner->forceOutCols, tFreeSStreamOutCol);

886,766✔

1841

void tFreeSStmTaskDeploy(void* param) {

2,379,053✔

1842

  if (NULL == param) {

2,379,053✔

1843

    return;

413,204✔

1846

  SStmTaskDeploy* pTask = (SStmTaskDeploy*)param;

1,965,849✔

1847

  switch (pTask->task.type)  {

1,965,849✔

1848

    case STREAM_READER_TASK:

823,965✔

1849

      tFreeSStreamReaderDeployMsg(&pTask->msg.reader);

823,965✔

1850

      break;

823,965✔

1851

    case STREAM_TRIGGER_TASK:

255,118✔

1852

      tFreeSStreamTriggerDeployMsg(&pTask->msg.trigger);

255,118✔

1853

      break;

255,118✔

1854

    case STREAM_RUNNER_TASK:

886,766✔

1855

      tFreeSStreamRunnerDeployMsg(&pTask->msg.runner);

886,766✔

1856

      break;

886,766✔

1863

void tFreeSStmStreamDeploy(void* param) {

334,240✔

1864

  if (NULL == param) {

334,240✔

1868

  SStmStreamDeploy* pDeploy = (SStmStreamDeploy*)param;

334,240✔

1869

  int32_t readerNum = taosArrayGetSize(pDeploy->readerTasks);

334,240✔

1870

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

1,158,803✔

1871

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

824,563✔

1872

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

824,563✔

1873

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

440,118✔

1876

  taosArrayDestroy(pDeploy->readerTasks);

334,240✔

1878

  if (pDeploy->triggerTask) {

334,240✔

1879

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

255,349✔

1880

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

255,349✔

1881

    taosMemoryFree(pDeploy->triggerTask);

255,349✔

1884

  int32_t runnerNum = taosArrayGetSize(pDeploy->runnerTasks);

334,240✔

1885

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

1,221,670✔

1886

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

887,430✔

1887

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

887,430✔

1889

  taosArrayDestroy(pDeploy->runnerTasks);

334,240✔

1892

void tDeepFreeSStmStreamDeploy(void* param) {

668,322✔

1893

  if (NULL == param) {

668,322✔

1897

  SStmStreamDeploy* pDeploy = (SStmStreamDeploy*)param;

668,322✔

1898

  taosArrayDestroyEx(pDeploy->readerTasks, tFreeSStmTaskDeploy);

668,322✔

1899

  tFreeSStmTaskDeploy(pDeploy->triggerTask);

668,322✔

1900

  taosMemoryFree(pDeploy->triggerTask);

668,322✔

1901

  taosArrayDestroyEx(pDeploy->runnerTasks, tFreeSStmTaskDeploy);

668,322✔

1905

void tFreeSMStreamHbRspMsg(SMStreamHbRspMsg* pRsp) {

41,228,144✔

1906

  if (NULL == pRsp) {

41,228,144✔

1909

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

41,228,144✔

1910

  taosArrayDestroy(pRsp->start.taskList);

41,228,144✔

1911

  taosArrayDestroy(pRsp->undeploy.taskList);

41,228,144✔

1912

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

41,228,144✔

1915

void tDeepFreeSMStreamHbRspMsg(SMStreamHbRspMsg* pRsp) {

20,355,790✔

1916

  if (NULL == pRsp) {

20,355,790✔

1919

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

20,355,790✔

1920

  taosArrayDestroy(pRsp->start.taskList);

20,355,790✔

1921

  taosArrayDestroy(pRsp->undeploy.taskList);

20,355,790✔

1922

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

20,355,790✔

1927

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

20,355,790✔

1928

  int32_t code = 0;

20,355,790✔

1931

  TAOS_CHECK_EXIT(tStartDecode(pDecoder));

20,355,790✔

1932

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

40,711,580✔

1933

  int32_t deployNum = 0;

20,355,790✔

1934

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &deployNum));

20,355,790✔

1935

  if (deployNum > 0) {

20,355,790✔

1936

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

103,183✔

1937

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

103,183✔

1939

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

20,689,872✔

1940

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

334,082✔

1941

    TAOS_CHECK_EXIT(tDecodeSStmStreamDeploy(pDecoder, pStream));

334,082✔

1944

  int32_t startNum = 0;

20,355,790✔

1945

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &startNum));

20,355,790✔

1946

  if (startNum > 0) {

20,355,790✔

1947

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

131,753✔

1948

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

131,753✔

1950

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

20,606,195✔

1951

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

250,405✔

1952

    TAOS_CHECK_EXIT(tDecodeSStreamTaskStart(pDecoder, pTask));

250,405✔

1955

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

40,711,580✔

1956

  if (!pRsp->undeploy.undeployAll) {

20,355,790✔

1957

    int32_t undeployNum = 0;

20,355,790✔

1958

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &undeployNum));

20,355,790✔

1959

    if (undeployNum > 0) {

20,355,790✔

1960

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

42,677✔

1961

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

42,677✔

1963

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

20,662,293✔

1964

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

306,503✔

1965

      TAOS_CHECK_EXIT(tDecodeSStreamTaskUndeploy(pDecoder, pTask));

306,503✔

1969

  int32_t rspNum = 0;

20,355,790✔

1970

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &rspNum));

20,355,790✔

1971

  if (rspNum > 0) {

20,355,790✔

1972

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

24,846✔

1973

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

24,846✔

1974

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

103,428✔

1975

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

78,582✔

1976

      TAOS_CHECK_EXIT(tDecodeSStreamMgmtRsp(pDecoder, pMgmtRsp));

78,582✔

1980

  tEndDecode(pDecoder);

20,355,790✔

1982

_exit:

20,355,790✔

1983

  return code;

20,355,790✔

2040

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

1,909,034✔

2041

  int32_t code = TSDB_CODE_SUCCESS;

1,909,034✔

2042

  int32_t lino = 0;

1,909,034✔

2044

  char*   json = NULL;

1,909,034✔

2045

  int32_t jsonLen = 0;

1,909,034✔

2046

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

1,909,034✔

2047

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

3,818,068✔

2049

_exit:

1,909,034✔

2050

  taosMemoryFreeClear(json);

1,909,034✔

2051

  if (code) {

1,909,034✔

2055

  return 0;

1,909,034✔

2058

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

872,260✔

2059

  SEncoder encoder = {0};

872,260✔

2060

  tEncoderInit(&encoder, buf, bufLen);

872,260✔

2061

  int32_t code = 0;

872,260✔

2064

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

872,260✔

2066

  TAOS_CHECK_EXIT(tSerializeSCMCreateStreamReqImpl(&encoder, pReq));

872,260✔

2068

  tEndEncode(&encoder);

872,260✔

2070

_exit:

872,260✔

2071

  if (code) {

872,260✔

2075

    int32_t tlen = encoder.pos;

872,260✔

2076

    tEncoderClear(&encoder);

872,260✔

2077

    return tlen;

872,260✔

2379

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

600,608✔

2380

  int32_t code = 0;

600,608✔

2383

  char* json = NULL;

600,608✔

2384

  SJson* pJson = NULL;

600,608✔

2385

  TAOS_CHECK_EXIT(tDecodeCStrAlloc(pDecoder, &json));

600,608✔

2386

  pJson = tjsonParse(json);

600,608✔

2387

  if (pJson == NULL) {

600,608✔

2390

  TAOS_CHECK_EXIT(jsonToSCMCreateStreamReq(pJson, pReq));

600,608✔

2392

_exit:

600,608✔

2393

  taosMemoryFreeClear(json);

600,608✔

2394

  if (NULL != pJson) {

600,608✔

2395

    tjsonDelete(pJson);

600,608✔

2398

  return code;

600,608✔

2402

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

295,031✔

2403

  SDecoder decoder = {0};

295,031✔

2404

  tDecoderInit(&decoder, buf, bufLen);

295,031✔

2405

  int32_t code = 0;

295,031✔

2408

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

295,031✔

2410

  code = tDeserializeSCMCreateStreamReqImpl(&decoder, pReq);

295,031✔

2411

  if (TSDB_CODE_MND_STREAM_INVALID_JSON == code) {

295,031✔

2420

  tEndDecode(&decoder);

295,031✔

2422

_exit:

295,031✔

2424

  tDecoderClear(&decoder);

295,031✔

2425

  return code;

295,031✔

2429

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

64,740✔

2430

  int32_t  code = 0;

64,740✔

2433

  SEncoder encoder = {0};

64,740✔

2434

  tEncoderInit(&encoder, buf, bufLen);

64,740✔

2436

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

64,740✔

2438

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

129,480✔

2439

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

137,112✔

2440

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

72,372✔

2441

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

144,744✔

2443

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

129,480✔

2445

  tEndEncode(&encoder);

64,740✔

2447

_exit:

64,740✔

2448

  if (code) {

64,740✔

2451

    tlen = encoder.pos;

64,740✔

2453

  tEncoderClear(&encoder);

64,740✔

2454

  return tlen;

64,740✔

2457

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

25,272✔

2458

  SDecoder decoder = {0};

25,272✔

2459

  int32_t  code = 0;

25,272✔

2461

  tDecoderInit(&decoder, buf, bufLen);

25,272✔

2463

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

25,272✔

2464

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

50,544✔

2465

  if (pReq->count > 0) {

25,272✔

2466

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

25,272✔

2467

    if (pReq->name == NULL) {

25,272✔

2471

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

54,360✔

2472

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

58,176✔

2475

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

50,544✔

2477

  tEndDecode(&decoder);

25,272✔

2479

_exit:

25,272✔

2480

  tDecoderClear(&decoder);

25,272✔

2481

  return code;

25,272✔

2484

void tFreeMDropStreamReq(SMDropStreamReq *pReq) {

58,686✔

2485

  if (NULL == pReq) {

58,686✔

2488

  if (pReq->name) {

58,686✔

2489

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

125,004✔

2490

      taosMemoryFreeClear(pReq->name[i]);

66,318✔

2492

    taosMemoryFreeClear(pReq->name);

58,686✔

2496

static FORCE_INLINE void tFreeStreamCalcScan(void* pScan) {

2,709,914✔

2497

  if (pScan == NULL) {

2,709,914✔

2500

  SStreamCalcScan *pCalcScan = (SStreamCalcScan *)pScan;

2,709,914✔

2501

  taosArrayDestroy(pCalcScan->vgList);

2,709,914✔

2502

  taosMemoryFreeClear(pCalcScan->scanPlan);

2,709,914✔

2505

void tFreeStreamOutCol(void* pCol) {

90,941✔

2506

  if (pCol == NULL) {

90,941✔

2509

  SStreamOutCol *pOutCol = (SStreamOutCol *)pCol;

90,941✔

2510

  taosMemoryFreeClear(pOutCol->expr);

90,941✔

2515

void tFreeSCMCreateStreamReq(SCMCreateStreamReq *pReq) {

1,664,345✔

2516

  if (NULL == pReq) {

1,664,345✔

2517

    return;

264,227✔

2519

  taosMemoryFreeClear(pReq->name);

1,400,118✔

2520

  taosMemoryFreeClear(pReq->sql);

1,400,118✔

2521

  taosMemoryFreeClear(pReq->streamDB);

1,400,118✔

2522

  taosMemoryFreeClear(pReq->triggerDB);

1,400,118✔

2523

  taosMemoryFreeClear(pReq->outDB);

1,400,118✔

2524

  taosMemoryFreeClear(pReq->triggerTblName);

1,400,118✔

2525

  taosMemoryFreeClear(pReq->outTblName);

1,400,118✔

2527

  taosArrayDestroyP(pReq->calcDB, NULL);

1,400,118✔

2528

  pReq->calcDB = NULL;

1,400,118✔

2529

  taosArrayDestroyP(pReq->pNotifyAddrUrls, NULL);

1,400,118✔

2530

  pReq->pNotifyAddrUrls = NULL;

1,400,118✔

2532

  taosMemoryFreeClear(pReq->triggerFilterCols);

1,400,118✔

2533

  taosMemoryFreeClear(pReq->triggerCols);

1,400,118✔

2534

  taosMemoryFreeClear(pReq->partitionCols);

1,400,118✔

2536

  taosArrayDestroy(pReq->outTags);

1,400,118✔

2537

  pReq->outTags = NULL;

1,400,118✔

2538

  taosArrayDestroy(pReq->outCols);

1,400,118✔

2539

  pReq->outCols = NULL;

1,400,118✔

2541

  switch (pReq->triggerType) {

1,400,118✔

2542

    case WINDOW_TYPE_STATE:

428,987✔

2543

      taosArrayDestroy(pReq->trigger.stateWin.pSlotIds);

428,987✔

2544

      pReq->trigger.stateWin.pSlotIds = NULL;

428,987✔

2545

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

428,987✔

2546

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

428,987✔

2547

      break;

428,987✔

2548

    case WINDOW_TYPE_EVENT:

128,433✔

2549

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

128,433✔

2550

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

128,433✔

2551

      break;

128,433✔

2552

    default:

842,698✔

2553

      break;

842,698✔

2556

  taosMemoryFreeClear(pReq->triggerScanPlan);

1,400,118✔

2557

  taosArrayDestroyEx(pReq->calcScanPlanList, tFreeStreamCalcScan);

1,400,118✔

2558

  pReq->calcScanPlanList = NULL;

1,400,118✔

2559

  taosMemoryFreeClear(pReq->triggerPrevFilter);

1,400,118✔

2561

  taosMemoryFreeClear(pReq->calcPlan);

1,400,118✔

2562

  taosMemoryFreeClear(pReq->subTblNameExpr);

1,400,118✔

2563

  taosMemoryFreeClear(pReq->tagValueExpr);

1,400,118✔

2564

  taosArrayDestroyEx(pReq->forceOutCols, tFreeStreamOutCol);

1,400,118✔

2565

  pReq->forceOutCols = NULL;

1,400,118✔

2566

  taosArrayDestroy(pReq->colCids);

1,400,118✔

2567

  pReq->colCids = NULL;

1,400,118✔

2568

  taosArrayDestroy(pReq->tagCids);

1,400,118✔

2569

  pReq->tagCids = NULL;

1,400,118✔

2572

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

247,292✔

2573

  int32_t code = 0, lino = 0;

247,292✔

2574

  if (NULL == pSrc) {

247,292✔

2578

  void* p = NULL;

247,292✔

2579

  int32_t num = 0;

247,292✔

2580

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

247,292✔

2581

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

247,292✔

2583

  SCMCreateStreamReq* pDst = *ppDst;

247,292✔

2585

  if (pSrc->outDB) {

247,292✔

2586

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

243,692✔

2587

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

243,692✔

2590

  if (pSrc->triggerTblName) {

247,292✔

2591

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

246,177✔

2592

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

246,177✔

2595

  if (pSrc->outTblName) {

247,292✔

2596

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

243,692✔

2597

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

243,692✔

2600

  if (pSrc->pNotifyAddrUrls) {

247,292✔

2601

    num = taosArrayGetSize(pSrc->pNotifyAddrUrls);

61,340✔

2602

    if (num > 0) {

61,340✔

2603

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

61,340✔

2604

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

61,340✔

2606

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

122,680✔

2607

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

61,340✔

2608

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

61,340✔

2609

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

122,680✔

2613

  if (pSrc->triggerFilterCols) {

247,292✔

2614

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

19,705✔

2615

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

19,705✔

2618

  if (pSrc->triggerCols) {

247,292✔

2619

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

239,330✔

2620

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

239,330✔

2623

  if (pSrc->partitionCols) {

247,292✔

2624

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

90,181✔

2625

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

90,181✔

2628

  if (pSrc->outCols) {

247,292✔

2629

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

243,692✔

2630

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

243,692✔

2633

  if (pSrc->outTags) {

247,292✔

2634

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

90,181✔

2635

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

90,181✔

2638

  pDst->triggerType = pSrc->triggerType;

247,292✔

2640

  switch (pSrc->triggerType) {

247,292✔

2641

    case WINDOW_TYPE_STATE:

92,667✔

2642

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

92,667✔

2643

        pDst->trigger.stateWin.pSlotIds = taosArrayDup(pSrc->trigger.stateWin.pSlotIds, NULL);

92,667✔

2644

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

92,667✔

2646

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

92,667✔

2647

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

92,667✔

2648

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

92,667✔

2649

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

92,667✔

2650

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

92,667✔

2651

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

5,397✔

2652

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

5,397✔

2654

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

92,667✔

2655

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

92,667✔

2656

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

92,667✔

2658

      break;

92,667✔

2659

    case WINDOW_TYPE_EVENT:

27,197✔

2660

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

27,197✔

2661

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

27,197✔

2662

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

27,197✔

2665

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

27,197✔

2666

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

24,757✔

2667

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

24,757✔

2669

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

27,197✔

2670

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

27,197✔

2671

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

27,197✔

2672

      break;

27,197✔

2673

    case WINDOW_TYPE_COUNT:

26,127✔

2674

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

26,127✔

2675

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

26,127✔

2676

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

26,127✔

2680

      break;

26,127✔

2681

    default:

101,301✔

2682

      pDst->trigger = pSrc->trigger;

101,301✔

2683

      break;

101,301✔

2687

  if (pSrc->triggerScanPlan) {

247,292✔

2688

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

246,177✔

2689

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

246,177✔

2692

  if (pSrc->calcScanPlanList) {

247,292✔

2693

    num = taosArrayGetSize(pSrc->calcScanPlanList);

243,692✔

2694

    if (num > 0) {

243,692✔

2695

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

243,692✔

2696

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

243,692✔

2698

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

779,773✔

2699

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

536,081✔

2700

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

536,081✔

2702

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

536,081✔

2703

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

536,081✔

2705

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

536,081✔

2706

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

536,081✔

2708

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

1,072,162✔

2712

  if (pSrc->triggerPrevFilter) {

247,292✔

2713

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

8,290✔

2714

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

8,290✔

2717

  if (pSrc->calcPlan) {

247,292✔

2718

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

243,692✔

2719

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

243,692✔

2722

  if (pSrc->subTblNameExpr) {

247,292✔

2723

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

90,181✔

2724

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

90,181✔

2727

  if (pSrc->tagValueExpr) {

247,292✔

2728

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

90,181✔

2729

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

90,181✔

2732

  if (pSrc->forceOutCols) {

247,292✔

2733

    num = taosArrayGetSize(pSrc->forceOutCols);

3,928✔

2734

    if (num > 0) {

3,928✔

2735

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

3,928✔

2736

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

3,928✔

2738

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

26,265✔

2739

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

22,337✔

2740

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

22,337✔

2742

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

22,337✔

2743

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

22,337✔

2745

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

44,674✔

2749

  if (pSrc->colCids) {

247,292✔

2750

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

4,943✔

2751

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

4,943✔

2754

  if (pSrc->tagCids) {

247,292✔

2755

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

3,890✔

2756

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

3,890✔

2759

  pDst->triggerTblUid = pSrc->triggerTblUid;

247,292✔

2760

  pDst->triggerTblSuid = pSrc->triggerTblSuid;

247,292✔

2761

  pDst->triggerTblType = pSrc->triggerTblType;

247,292✔

2762

  pDst->triggerPrec = pSrc->triggerPrec;

247,292✔

2763

  pDst->deleteReCalc = pSrc->deleteReCalc;

247,292✔

2764

  pDst->deleteOutTbl = pSrc->deleteOutTbl;

247,292✔

2765

  pDst->flags = pSrc->flags;

247,292✔

2767

_exit:

247,292✔

2769

  if (code) {

247,292✔

2774

  return code;

247,292✔

2778

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

7,300✔

2779

  int32_t  code = 0;

7,300✔

2782

  SEncoder encoder = {0};

7,300✔

2783

  tEncoderInit(&encoder, buf, bufLen);

7,300✔

2784

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

7,300✔

2786

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

7,300✔

2787

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

14,600✔

2788

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

14,600✔

2789

  tEndEncode(&encoder);

7,300✔

2791

_exit:

7,300✔

2792

  if (code) {

7,300✔

2795

    tlen = encoder.pos;

7,300✔

2797

  tEncoderClear(&encoder);

7,300✔

2798

  return tlen;

7,300✔

2801

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

3,468✔

2802

  SDecoder decoder = {0};

3,468✔

2803

  int32_t  code = 0;

3,468✔

2806

  tDecoderInit(&decoder, buf, bufLen);

3,468✔

2807

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

3,468✔

2808

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

6,936✔

2809

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

6,936✔

2810

  tEndDecode(&decoder);

3,468✔

2812

_exit:

3,468✔

2813

  tDecoderClear(&decoder);

3,468✔

2814

  return code;

3,468✔

2817

void tFreeMPauseStreamReq(SMPauseStreamReq *pReq) {

3,650✔

2818

  taosMemoryFreeClear(pReq->name);

3,650✔

2819

3,650✔

2821

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

6,412✔

2822

  SEncoder encoder = {0};

6,412✔

2823

  int32_t  code = 0;

6,412✔

2826

  tEncoderInit(&encoder, buf, bufLen);

6,412✔

2827

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

6,412✔

2828

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

6,412✔

2829

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

12,824✔

2830

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

12,824✔

2831

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

12,824✔

2832

  tEndEncode(&encoder);

6,412✔

2834

_exit:

6,412✔

2835

  if (code) {

6,412✔

2838

    tlen = encoder.pos;

6,412✔

2840

  tEncoderClear(&encoder);

6,412✔

2841

  return tlen;

6,412✔

2844

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

3,024✔

2845

  SDecoder decoder = {0};

3,024✔

2846

  int32_t  code = 0;

3,024✔

2849

  tDecoderInit(&decoder, buf, bufLen);

3,024✔

2850

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

3,024✔

2851

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

6,048✔

2852

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

6,048✔

2853

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

6,048✔

2854

  tEndDecode(&decoder);

3,024✔

2856

_exit:

3,024✔

2857

  tDecoderClear(&decoder);

3,024✔

2858

  return code;

3,024✔

2861

void tFreeMResumeStreamReq(SMResumeStreamReq *pReq) {

3,206✔

2862

  taosMemoryFreeClear(pReq->name);

3,206✔

2863

3,206✔

2865

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

28,642✔

2866

  SEncoder encoder = {0};

28,642✔

2867

  int32_t  code = 0;

28,642✔

2870

  tEncoderInit(&encoder, buf, bufLen);

28,642✔

2871

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

28,642✔

2872

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

28,642✔

2873

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

57,284✔

2874

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

57,284✔

2875

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

57,284✔

2876

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

57,284✔

2877

  tEndEncode(&encoder);

28,642✔

2879

_exit:

28,642✔

2880

  if (code) {

28,642✔

2883

    tlen = encoder.pos;

28,642✔

2885

  tEncoderClear(&encoder);

28,642✔

2886

  return tlen;

28,642✔

2889

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

14,321✔

2890

  SDecoder decoder = {0};

14,321✔

2891

  int32_t  code = 0;

14,321✔

2894

  tDecoderInit(&decoder, buf, bufLen);

14,321✔

2895

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

14,321✔

2897

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

28,642✔

2898

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

28,642✔

2899

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

28,642✔

2900

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

28,642✔

2901

  tEndDecode(&decoder);

14,321✔

2903

_exit:

14,321✔

2904

  tDecoderClear(&decoder);

14,321✔

2905

  return code;

14,321✔

2908

void tFreeMRecalcStreamReq(SMRecalcStreamReq *pReq) {

28,642✔

2909

  taosMemoryFreeClear(pReq->name);

28,642✔

2910

28,642✔

2912

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

16,356✔

2913

  int32_t code = 0;

16,356✔

2916

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

32,712✔

2917

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

32,712✔

2918

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

32,712✔

2920

_exit:

16,356✔

2921

  return code;

16,356✔

2924

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

16,356✔

2925

  SEncoder encoder = {0};

16,356✔

2926

  int32_t  code = 0;

16,356✔

2929

  tEncoderInit(&encoder, buf, bufLen);

16,356✔

2931

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

16,356✔

2932

  TAOS_CHECK_EXIT(tEncodeStreamProgressReq(&encoder, pReq));

16,356✔

2934

  tEndEncode(&encoder);

16,356✔

2936

_exit:

16,356✔

2937

  if (code) {

16,356✔

2940

    tlen = encoder.pos;

16,356✔

2942

  tEncoderClear(&encoder);

16,356✔

2943

  return tlen;

16,356✔

2946

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

13,494✔

2947

  int32_t code = 0;

13,494✔

2950

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

26,988✔

2951

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

26,988✔

2952

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

26,988✔

2954

_exit:

13,494✔

2955

  return code;

13,494✔

2958

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

13,494✔

2959

  SDecoder decoder = {0};

13,494✔

2960

  int32_t  code = 0;

13,494✔

2963

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

13,494✔

2965

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

13,494✔

2966

  TAOS_CHECK_EXIT(tDecodeStreamProgressReq(&decoder, pReq));

13,494✔

2968

  tEndDecode(&decoder);

13,494✔

2970

_exit:

13,494✔

2971

  tDecoderClear(&decoder);

13,494✔

2972

  return code;

13,494✔

2975

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

16,262✔

2976

  int32_t code = 0;

16,262✔

2979

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

32,524✔

2980

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

32,524✔

2981

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

32,524✔

2982

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

32,524✔

2984

_exit:

16,262✔

2985

  return code;

16,262✔

2988

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

16,262✔

2989

  SEncoder encoder = {0};

16,262✔

2990

  int32_t  code = 0;

16,262✔

2993

  tEncoderInit(&encoder, buf, bufLen);

16,262✔

2995

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

16,262✔

2996

  TAOS_CHECK_EXIT(tEncodeStreamProgressRsp(&encoder, pRsp));

16,262✔

2998

  tEndEncode(&encoder);

16,262✔

3000

_exit:

16,262✔

3001

  if (code) {

16,262✔

3004

    tlen = encoder.pos;

16,262✔

3006

  tEncoderClear(&encoder);

16,262✔

3007

  return tlen;

16,262✔

3010

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

8,178✔

3011

  int32_t code = 0;

8,178✔

3014

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

16,356✔

3015

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

16,356✔

3016

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

16,356✔

3017

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

16,356✔

3019

_exit:

8,178✔

3020

  return code;

8,178✔

3023

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

8,178✔

3024

  SDecoder decoder = {0};

8,178✔

3025

  int32_t  code = 0;

8,178✔

3028

  tDecoderInit(&decoder, buf, bufLen);

8,178✔

3030

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

8,178✔

3031

  TAOS_CHECK_EXIT(tDecodeStreamProgressRsp(&decoder, pRsp));

8,178✔

3033

  tEndDecode(&decoder);

8,178✔

3035

_exit:

8,178✔

3036

  tDecoderClear(&decoder);

8,178✔

3037

  return code;

8,178✔

3040

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

297,552✔

3041

  SEncoder encoder = {0};

297,552✔

3042

  int32_t  code = TSDB_CODE_SUCCESS;

297,552✔

3043

  int32_t  lino = 0;

297,552✔

3044

  int32_t  tlen = 0;

297,552✔

3046

  tEncoderInit(&encoder, buf, bufLen);

297,552✔

3047

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

297,552✔

3049

  int32_t size = taosArrayGetSize(pRsp->cols);

297,552✔

3050

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, size));

297,552✔

3051

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

1,088,060✔

3052

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

790,508✔

3053

    if (oInfo == NULL) {

790,508✔

3058

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

1,580,730✔

3059

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

1,580,730✔

3060

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

1,581,016✔

3063

  tEndEncode(&encoder);

297,552✔

3065

_exit:

297,552✔

3066

  if (code != TSDB_CODE_SUCCESS) {

297,552✔

3069

    tlen = encoder.pos;

297,552✔

3071

  tEncoderClear(&encoder);

297,552✔

3072

  return tlen;

296,694✔

3075

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

148,776✔

3076

  SDecoder decoder = {0};

148,776✔

3077

  int32_t  code = TSDB_CODE_SUCCESS;

148,776✔

3078

  int32_t  lino = 0;

148,776✔

3080

  tDecoderInit(&decoder, buf, bufLen);

148,776✔

3081

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

148,776✔

3083

  int32_t size = 0;

148,776✔

3084

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

148,776✔

3085

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

148,776✔

3086

  if (pRsp->cols == NULL) {

148,776✔

3091

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

544,173✔

3092

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

395,397✔

3093

    if (oInfo == NULL) {

395,397✔

3098

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

790,794✔

3099

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

790,794✔

3100

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

790,794✔

3103

  tEndDecode(&decoder);

148,776✔

3105

_exit:

148,776✔

3106

  tDecoderClear(&decoder);

148,776✔

3107

  return code;

148,776✔

3110

void    tDestroySTriggerOrigTableInfoRsp(SSTriggerOrigTableInfoRsp* pRsp){

46,794,254✔

3111

  taosArrayDestroy(pRsp->cols);

46,794,254✔

3112

46,794,254✔

3114

void tDestroySTriggerPullRequest(SSTriggerPullRequestUnion* pReq) {

57,045,733✔

3115

  if (pReq == NULL) return;

57,045,733✔

3116

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

73,607,248✔

3117

    SSTriggerWalDataNewRequest* pRequest = (SSTriggerWalDataNewRequest*)pReq;

16,561,807✔

3118

    taosArrayDestroy(pRequest->versions);

16,561,807✔

3119

    tSimpleHashCleanup(pRequest->ranges);

16,561,256✔

3120

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

40,486,865✔

3121

    SSTriggerTsdbDataRequest* pRequest = (SSTriggerTsdbDataRequest*)pReq;

579,009✔

3122

    if (pRequest->cids != NULL) {

579,009✔

3123

      taosArrayDestroy(pRequest->cids);

579,009✔

3124

      pRequest->cids = NULL;

579,009✔

3126

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

39,904,301✔

3127

    SSTriggerVirTableInfoRequest* pRequest = (SSTriggerVirTableInfoRequest*)pReq;

85,599✔

3128

    if (pRequest->cids != NULL) {

85,599✔

3129

      taosArrayDestroy(pRequest->cids);

85,599✔

3130

      pRequest->cids = NULL;

85,599✔

3132

    if (pRequest->uids != NULL) {

85,599✔

3136

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

39,819,558✔

3137

    SSTriggerVirTablePseudoColRequest *pRequest = (SSTriggerVirTablePseudoColRequest*)pReq;

1,114,997✔

3138

    if (pRequest->cids != NULL) {

1,114,997✔

3139

      taosArrayDestroy(pRequest->cids);

1,114,997✔

3140

      pRequest->cids = NULL;

1,114,997✔

3142

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

38,702,105✔

3143

    SSTriggerOrigTableInfoRequest* pRequest = (SSTriggerOrigTableInfoRequest*)pReq;

148,776✔

3144

    if (pRequest->cols != NULL) {

148,776✔

3145

      taosArrayDestroy(pRequest->cols);

148,776✔

3146

      pRequest->cols = NULL;

148,776✔

3148

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

38,555,244✔

3149

    SSTriggerSetTableRequest* pRequest = (SSTriggerSetTableRequest*)pReq;

148,776✔

3150

    tSimpleHashCleanup(pRequest->uidInfoTrigger);

148,776✔

3151

    tSimpleHashCleanup(pRequest->uidInfoCalc);

148,776✔

3155

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

3,735,888✔

3156

  int32_t  code = TSDB_CODE_SUCCESS;

3,735,888✔

3157

  int32_t  lino = 0;

3,735,888✔

3158

  int32_t  nEle = taosArrayGetSize(pArr);

3,735,888✔

3159

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

3,735,888✔

3160

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

3,735,888✔

3161

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

7,471,776✔

3163

_exit:

3,735,888✔

3164

  return code;

3,735,888✔

3167

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

1,865,204✔

3168

  int32_t  code = TSDB_CODE_SUCCESS;

1,865,204✔

3169

  int32_t  lino = 0;

1,865,204✔

3170

  void*    buf = NULL;

1,865,204✔

3171

  uint64_t len = 0;

1,865,204✔

3172

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

1,865,204✔

3174

  if (len > 0) {

1,865,204✔

3175

    *ppArr = taosArrayInit(0, elemSize);

1,779,605✔

3176

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

1,779,605✔

3177

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

1,779,605✔

3178

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

1,779,605✔

3181

_exit:

1,865,204✔

3182

  if (buf != NULL) {

1,865,204✔

3183

    taosMemoryFree(buf);

1,779,605✔

3185

  return code;

1,865,204✔

3188

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

595,104✔

3189

  int32_t  code = TSDB_CODE_SUCCESS;

595,104✔

3190

  int32_t  lino = 0;

595,104✔

3191

  int32_t size = tSimpleHashGetSize(pInfo);

595,104✔

3192

  TAOS_CHECK_EXIT(tEncodeI32(encoder, size));

595,104✔

3193

  int32_t iter = 0;

595,104✔

3194

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

595,104✔

3195

  while (px != NULL) {

1,331,882✔

3196

    int64_t* uid = tSimpleHashGetKey(px, NULL);

736,778✔

3197

    TAOS_CHECK_EXIT(tEncodeI64(encoder, *uid));

1,473,556✔

3198

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

1,473,556✔

3199

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

736,778✔

3200

    int32_t len = tSimpleHashGetSize(info);

736,778✔

3201

    TAOS_CHECK_EXIT(tEncodeI32(encoder, len));

736,778✔

3202

    int32_t iter1 = 0;

736,778✔

3203

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

736,778✔

3204

    while (px1 != NULL) {

2,767,282✔

3205

      int16_t* slot = tSimpleHashGetKey(px1, NULL);

2,030,504✔

3206

      int16_t* cid = (int16_t*)px1;

2,030,504✔

3207

      TAOS_CHECK_EXIT(tEncodeI16(encoder, *slot));

4,061,294✔

3208

      TAOS_CHECK_EXIT(tEncodeI16(encoder, *cid));

4,061,580✔

3210

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

2,030,790✔

3213

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

736,778✔

3216

_exit:

595,104✔

3217

  return code;

595,104✔

3220

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

114,577,157✔

3221

  SEncoder encoder = {0};

114,577,157✔

3222

  int32_t  code = TSDB_CODE_SUCCESS;

114,578,251✔

3223

  int32_t  lino = 0;

114,578,251✔

3224

  int32_t  tlen = 0;

114,578,251✔

3226

  tEncoderInit(&encoder, buf, bufLen);

114,578,251✔

3227

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

114,577,642✔

3229

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

229,149,617✔

3230

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

229,146,544✔

3231

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

229,149,379✔

3232

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->sessionId));

229,141,639✔

3234

  switch (pReq->type) {

114,565,534✔

3235

    case STRIGGER_PULL_SET_TABLE: {

297,552✔

3236

      SSTriggerSetTableRequest* pRequest = (SSTriggerSetTableRequest*)pReq;

297,552✔

3237

      TAOS_CHECK_EXIT(encodeSetTableMapInfo(&encoder, pRequest->uidInfoTrigger));

297,552✔

3238

      TAOS_CHECK_EXIT(encodeSetTableMapInfo(&encoder, pRequest->uidInfoCalc));

297,552✔

3239

      break;

297,552✔

3241

    case STRIGGER_PULL_LAST_TS: {

727,986✔

3242

      break;

727,986✔

3244

    case STRIGGER_PULL_FIRST_TS: {

744,809✔

3245

      SSTriggerFirstTsRequest* pRequest = (SSTriggerFirstTsRequest*)pReq;

744,809✔

3246

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->gid));

1,490,209✔

3247

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->startTime));

1,490,209✔

3248

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

1,489,904✔

3249

      break;

745,095✔

3251

    case STRIGGER_PULL_TSDB_META: {

999,228✔

3252

      SSTriggerTsdbMetaRequest* pRequest = (SSTriggerTsdbMetaRequest*)pReq;

999,228✔

3253

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->startTime));

1,998,456✔

3254

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->endTime));

1,998,456✔

3255

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->gid));

1,998,456✔

3256

      TAOS_CHECK_EXIT(tEncodeI8(&encoder, pRequest->order));

1,998,456✔

3257

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

1,998,456✔

3258

      break;

999,228✔

3263

    case STRIGGER_PULL_TSDB_TS_DATA: {

188,536✔

3264

      SSTriggerTsdbTsDataRequest* pRequest = (SSTriggerTsdbTsDataRequest*)pReq;

188,536✔

3265

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->suid));

377,072✔

3266

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->uid));

377,072✔

3267

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->skey));

377,072✔

3268

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ekey));

377,072✔

3269

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

377,072✔

3270

      break;

188,536✔

3272

    case STRIGGER_PULL_TSDB_TRIGGER_DATA: {

205,374✔

3273

      SSTriggerTsdbTriggerDataRequest* pRequest = (SSTriggerTsdbTriggerDataRequest*)pReq;

205,374✔

3274

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->startTime));

410,748✔

3275

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->gid));

410,748✔

3276

      TAOS_CHECK_EXIT(tEncodeI8(&encoder, pRequest->order));

410,748✔

3277

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

410,748✔

3278

      break;

205,374✔

3280

    case STRIGGER_PULL_TSDB_TRIGGER_DATA_NEXT: {

381,498✔

3281

      break;

381,498✔

3283

    case STRIGGER_PULL_TSDB_CALC_DATA: {

15,478,756✔

3284

      SSTriggerTsdbCalcDataRequest* pRequest = (SSTriggerTsdbCalcDataRequest*)pReq;

15,478,756✔

3285

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->gid));

30,957,512✔

3286

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->skey));

30,957,512✔

3287

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ekey));

30,957,512✔

3288

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

30,957,512✔

3289

      break;

15,478,756✔

3294

    case STRIGGER_PULL_TSDB_DATA: {

1,161,334✔

3295

      SSTriggerTsdbDataRequest* pRequest = (SSTriggerTsdbDataRequest*)pReq;

1,161,334✔

3296

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->suid));

2,322,668✔

3297

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->uid));

2,322,668✔

3298

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->skey));

2,322,668✔

3299

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ekey));

2,322,668✔

3300

      TAOS_CHECK_EXIT(encodePlainArray(&encoder, pRequest->cids));

1,161,334✔

3301

      TAOS_CHECK_EXIT(tEncodeI8(&encoder, pRequest->order));

2,322,668✔

3302

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

2,322,668✔

3303

      break;

1,161,334✔

3308

    case STRIGGER_PULL_WAL_META_NEW: {

44,528,407✔

3309

      SSTriggerWalMetaNewRequest* pRequest = (SSTriggerWalMetaNewRequest*)pReq;

44,528,407✔

3310

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->lastVer));

89,059,092✔

3311

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ctime));

89,060,007✔

3312

      break;

44,529,322✔

3314

    case STRIGGER_PULL_WAL_DATA_NEW:

33,124,271✔

3316

      SSTriggerWalDataNewRequest* pRequest = (SSTriggerWalDataNewRequest*)pReq;

33,124,271✔

3317

      int32_t                     nVersion = taosArrayGetSize(pRequest->versions);

33,124,271✔

3318

      TAOS_CHECK_EXIT(tEncodeI32(&encoder, nVersion));

33,124,242✔

3319

      for (int32_t i = 0; i < nVersion; i++) {

52,354,768✔

3320

        int64_t ver = *(int64_t*)TARRAY_GET_ELEM(pRequest->versions, i);

19,230,526✔

3321

        TAOS_CHECK_EXIT(tEncodeI64(&encoder, ver));

19,230,526✔

3323

      int32_t nRanges = tSimpleHashGetSize(pRequest->ranges);

33,124,242✔

3324

      TAOS_CHECK_EXIT(tEncodeI32(&encoder, nRanges));

33,124,528✔

3325

      int32_t iter = 0;

33,124,528✔

3326

      void*   px = tSimpleHashIterate(pRequest->ranges, NULL, &iter);

33,124,814✔

3327

      while (px != NULL) {

42,095,737✔

3328

        uint64_t* gid = tSimpleHashGetKey(px, NULL);

8,971,466✔

3329

        TAOS_CHECK_EXIT(tEncodeU64(&encoder, *gid));

17,942,932✔

3330

        int64_t* key = (int64_t*)px;

8,971,466✔

3331

        TAOS_CHECK_EXIT(tEncodeI64(&encoder, key[0]));

17,942,932✔

3332

        TAOS_CHECK_EXIT(tEncodeI64(&encoder, key[1]));

17,942,932✔

3334

        px = tSimpleHashIterate(pRequest->ranges, px, &iter);

8,971,466✔

3336

      break;

33,124,271✔

3338

    case STRIGGER_PULL_WAL_META_DATA_NEW: {

13,284,409✔

3339

      SSTriggerWalMetaDataNewRequest* pRequest = (SSTriggerWalMetaDataNewRequest*)pReq;

13,284,409✔

3340

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->lastVer));

26,563,739✔

3341

      break;

13,279,330✔

3343

    case STRIGGER_PULL_GROUP_COL_VALUE: {

751,484✔

3344

      SSTriggerGroupColValueRequest* pRequest = (SSTriggerGroupColValueRequest*)pReq;

751,484✔

3345

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->gid));

1,502,968✔

3346

      break;

751,484✔

3348

    case STRIGGER_PULL_VTABLE_INFO: {

172,280✔

3349

      SSTriggerVirTableInfoRequest* pRequest = (SSTriggerVirTableInfoRequest*)pReq;

172,280✔

3350

      TAOS_CHECK_EXIT(encodePlainArray(&encoder, pRequest->cids));

172,280✔

3351

      TAOS_CHECK_EXIT(encodePlainArray(&encoder, pRequest->uids));

172,280✔

3352

      TAOS_CHECK_EXIT(tEncodeBool(&encoder, pRequest->fetchAllTable));

172,280✔

3353

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

344,560✔

3354

      break;

172,280✔

3356

    case STRIGGER_PULL_VTABLE_PSEUDO_COL: {

2,229,994✔

3357

      SSTriggerVirTablePseudoColRequest* pRequest = (SSTriggerVirTablePseudoColRequest*)pReq;

2,229,994✔

3358

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->uid));

4,459,988✔

3359

      TAOS_CHECK_EXIT(encodePlainArray(&encoder, pRequest->cids));

2,229,994✔

3360

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

4,459,988✔

3361

      break;

2,229,994✔

3363

    case STRIGGER_PULL_OTABLE_INFO: {

297,552✔

3364

      SSTriggerOrigTableInfoRequest* pRequest = (SSTriggerOrigTableInfoRequest*)pReq;

297,552✔

3365

      int32_t size = taosArrayGetSize(pRequest->cols);

297,552✔

3366

      TAOS_CHECK_EXIT(tEncodeI32(&encoder, size));

297,552✔

3367

      for (int32_t i = 0; i < size; ++i) {

1,088,346✔

3368

        OTableInfo* oInfo = taosArrayGet(pRequest->cols, i);

790,794✔

3369

        if (oInfo == NULL) {

790,794✔

3374

        TAOS_CHECK_EXIT(tEncodeCStr(&encoder, oInfo->refTableName));

1,581,588✔

3375

        TAOS_CHECK_EXIT(tEncodeCStr(&encoder, oInfo->refColName));

1,581,588✔

3377

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

595,104✔

3378

      break;

297,552✔

3380

    default: {

1,272✔

3381

      uError("unknown pull type %d", pReq->type);

1,272✔

3387

  tEndEncode(&encoder);

114,569,878✔

3389

_exit:

114,569,238✔

3390

  if (code != TSDB_CODE_SUCCESS) {

114,569,522✔

3393

    tlen = encoder.pos;

114,569,522✔

3395

  tEncoderClear(&encoder);

114,569,522✔

3396

  return tlen;

114,566,566✔

3399

static void destroyHash(void* data){

368,389✔

3400

  if (data){

368,389✔

3401

    SSHashObj* tmp = *(SSHashObj**)data;

368,389✔

3402

    tSimpleHashCleanup(tmp);

368,389✔

3404

368,389✔

3406

static int32_t decodeSetTableMapInfo(SDecoder* decoder, SSHashObj** ppInfo) {

297,552✔

3407

  int32_t  code = TSDB_CODE_SUCCESS;

297,552✔

3408

  int32_t  lino = 0;

297,552✔

3409

  int32_t size = 0;

297,552✔

3410

  TAOS_CHECK_EXIT(tDecodeI32(decoder, &size));

297,552✔

3411

  *ppInfo = tSimpleHashInit(size, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));

297,552✔

3412

  if (*ppInfo == NULL) {

297,552✔

3415

  tSimpleHashSetFreeFp(*ppInfo, destroyHash);

297,552✔

3417

  for (int32_t i = 0; i < size; ++i) {

665,941✔

3418

    int64_t id[2] = {0};

368,389✔

3419

    TAOS_CHECK_EXIT(tDecodeI64(decoder, id));

368,389✔

3420

    TAOS_CHECK_EXIT(tDecodeI64(decoder, id+1));

736,778✔

3421

    int32_t len = 0;

368,389✔

3422

    TAOS_CHECK_EXIT(tDecodeI32(decoder, &len));

368,389✔

3423

    SSHashObj* tmp = tSimpleHashInit(len, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT));

368,389✔

3424

    if (tmp == NULL) {

368,389✔

3427

    TAOS_CHECK_EXIT(tSimpleHashPut(*ppInfo, id, sizeof(id), &tmp, POINTER_BYTES));

368,389✔

3429

    for (int32_t j = 0; j < len; ++j) {

1,383,784✔

3430

      int16_t slotId = 0;

1,015,395✔

3431

      int16_t cid = 0;

1,015,395✔

3432

      TAOS_CHECK_EXIT(tDecodeI16(decoder, &slotId));

1,015,395✔

3433

      TAOS_CHECK_EXIT(tDecodeI16(decoder, &cid));

1,015,395✔

3434

      TAOS_CHECK_EXIT(tSimpleHashPut(tmp, &slotId, sizeof(slotId), &cid, sizeof(cid)));

1,015,395✔

3437

_exit:

297,552✔

3438

  if (code != TSDB_CODE_SUCCESS) {

297,552✔

3442

  return code;

297,552✔

3445

int32_t tDeserializeSTriggerPullRequest(void* buf, int32_t bufLen, SSTriggerPullRequestUnion* pReq) {

57,046,806✔

3446

  SDecoder decoder = {0};

57,046,806✔

3447

  int32_t  code = TSDB_CODE_SUCCESS;

57,047,808✔

3448

  int32_t  lino = 0;

57,047,808✔

3450

  tDecoderInit(&decoder, buf, bufLen);

57,047,808✔

3451

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

57,049,528✔

3453

  int32_t type = 0;

57,051,483✔

3454

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &type));

57,050,260✔

3455

  SSTriggerPullRequest* pBase = &(pReq->base);

57,050,260✔

3456

  pBase->type = type;

57,050,517✔

3457

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pBase->streamId));

114,099,027✔

3458

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pBase->readerTaskId));

114,100,511✔

3459

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pBase->sessionId));

114,100,511✔

3461

  switch (type) {

57,050,168✔

3462

    case STRIGGER_PULL_SET_TABLE: {

148,776✔

3463

      SSTriggerSetTableRequest* pRequest = &(pReq->setTableReq);

148,776✔

3464

      TAOS_CHECK_EXIT(decodeSetTableMapInfo(&decoder, &pRequest->uidInfoTrigger));

148,776✔

3465

      TAOS_CHECK_EXIT(decodeSetTableMapInfo(&decoder, &pRequest->uidInfoCalc));

148,776✔

3466

      break;

148,776✔

3468

    case STRIGGER_PULL_LAST_TS: {

363,724✔

3469

      break;

363,724✔

3471

    case STRIGGER_PULL_FIRST_TS: {

367,731✔

3472

      SSTriggerFirstTsRequest* pRequest = &(pReq->firstTsReq);

367,731✔

3473

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->gid));

735,462✔

3474

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->startTime));

735,462✔

3475

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

735,462✔

3476

      break;

367,731✔

3478

    case STRIGGER_PULL_TSDB_META: {

499,462✔

3479

      SSTriggerTsdbMetaRequest* pRequest = &(pReq->tsdbMetaReq);

499,462✔

3480

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->startTime));

998,924✔

3481

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->endTime));

998,924✔

3482

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->gid));

998,924✔

3483

      TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pRequest->order));

998,924✔

3484

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

998,924✔

3485

      break;

499,462✔

3490

    case STRIGGER_PULL_TSDB_TS_DATA: {

94,268✔

3491

      SSTriggerTsdbTsDataRequest* pRequest = &(pReq->tsdbTsDataReq);

94,268✔

3492

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->suid));

188,536✔

3493

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->uid));

188,536✔

3494

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->skey));

188,536✔

3495

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ekey));

188,536✔

3496

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

188,536✔

3497

      break;

94,268✔

3499

    case STRIGGER_PULL_TSDB_TRIGGER_DATA: {

102,687✔

3500

      SSTriggerTsdbTriggerDataRequest* pRequest = &(pReq->tsdbTriggerDataReq);

102,687✔

3501

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->startTime));

205,374✔

3502

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->gid));

205,374✔

3503

      TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pRequest->order));

205,374✔

3504

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

205,374✔

3505

      break;

102,687✔

3507

    case STRIGGER_PULL_TSDB_TRIGGER_DATA_NEXT: {

190,749✔

3508

      break;

190,749✔

3510

    case STRIGGER_PULL_TSDB_CALC_DATA: {

7,737,782✔

3511

      SSTriggerTsdbCalcDataRequest* pRequest = &(pReq->tsdbCalcDataReq);

7,737,782✔

3512

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->gid));

15,475,564✔

3513

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->skey));

15,475,564✔

3514

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ekey));

15,475,564✔

3515

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

15,475,564✔

3516

      break;

7,737,782✔

3521

    case STRIGGER_PULL_TSDB_DATA: {

579,009✔

3522

      SSTriggerTsdbDataRequest* pRequest = &(pReq->tsdbDataReq);

579,009✔

3523

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->suid));

1,158,018✔

3524

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->uid));

1,158,018✔

3525

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->skey));

1,158,018✔

3526

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ekey));

1,158,018✔

3527

      TAOS_CHECK_EXIT(decodePlainArray(&decoder, &pRequest->cids, sizeof(col_id_t)));

579,009✔

3528

      TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pRequest->order));

1,158,018✔

3529

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

1,158,018✔

3530

      break;

579,009✔

3535

    case STRIGGER_PULL_WAL_META_NEW: {

22,141,500✔

3536

      SSTriggerWalMetaNewRequest* pRequest = &(pReq->walMetaNewReq);

22,141,500✔

3537

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->lastVer));

44,288,546✔

3538

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ctime));

44,290,776✔

3539

      break;

22,145,388✔

3541

    case STRIGGER_PULL_WAL_DATA_NEW:

16,561,513✔

3543

      SSTriggerWalDataNewRequest* pRequest = &(pReq->walDataNewReq);

16,561,513✔

3544

      int32_t                     nVersion = 0;

16,561,513✔

3545

      TAOS_CHECK_EXIT(tDecodeI32(&decoder, &nVersion));

16,561,799✔

3546

      pRequest->versions = taosArrayInit_s(sizeof(int64_t), nVersion);

16,561,799✔

3547

      for (int32_t i = 0; i < nVersion; i++) {

26,176,302✔

3548

        int64_t* pVer = TARRAY_GET_ELEM(pRequest->versions, i);

9,614,503✔

3549

        TAOS_CHECK_EXIT(tDecodeI64(&decoder, pVer));

9,614,217✔

3551

      int32_t nRanges = 0;

16,561,799✔

3552

      TAOS_CHECK_EXIT(tDecodeI32(&decoder, &nRanges));

16,562,085✔

3553

      pRequest->ranges = tSimpleHashInit(nRanges, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT));

16,562,085✔

3554

      if (pRequest->ranges == NULL) {

16,560,141✔

3557

      for (int32_t i = 0; i < nRanges; i++) {

21,045,266✔

3558

        uint64_t gid = 0;

4,485,125✔

3559

        int64_t pRange[2] = {0};

4,485,125✔

3560

        TAOS_CHECK_EXIT(tDecodeU64(&decoder, &gid));

4,485,125✔

3561

        TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRange[0]));

4,485,125✔

3562

        TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRange[1]));

4,485,125✔

3563

        TAOS_CHECK_EXIT(tSimpleHashPut(pRequest->ranges, &gid, sizeof(gid), pRange, sizeof(pRange)));

4,485,125✔

3565

      break;

16,560,141✔

3567

    case STRIGGER_PULL_WAL_META_DATA_NEW: {

6,535,049✔

3568

      SSTriggerWalMetaDataNewRequest* pRequest = &(pReq->walMetaDataNewReq);

6,535,049✔

3569

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->lastVer));

13,070,728✔

3570

      break;

6,535,222✔

3572

    case STRIGGER_PULL_GROUP_COL_VALUE: {

375,400✔

3573

      SSTriggerGroupColValueRequest* pRequest = &(pReq->groupColValueReq);

375,400✔

3574

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->gid));

750,800✔

3575

      break;

375,400✔

3577

    case STRIGGER_PULL_VTABLE_INFO: {

85,599✔

3578

      SSTriggerVirTableInfoRequest* pRequest = &(pReq->virTableInfoReq);

85,599✔

3579

      TAOS_CHECK_EXIT(decodePlainArray(&decoder, &pRequest->cids, sizeof(col_id_t)));

85,599✔

3580

      TAOS_CHECK_EXIT(decodePlainArray(&decoder, &pRequest->uids, sizeof(int64_t)));

85,599✔

3581

      TAOS_CHECK_EXIT(tDecodeBool(&decoder, &pRequest->fetchAllTable));

85,599✔

3582

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

171,198✔

3583

      break;

85,599✔

3585

    case STRIGGER_PULL_VTABLE_PSEUDO_COL: {

1,114,997✔

3586

      SSTriggerVirTablePseudoColRequest* pRequest = &(pReq->virTablePseudoColReq);

1,114,997✔

3587

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->uid));

2,229,994✔

3588

      TAOS_CHECK_EXIT(decodePlainArray(&decoder, &pRequest->cids, sizeof(col_id_t)));

1,114,997✔

3589

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

2,229,994✔

3590

      break;

1,114,997✔

3592

    case STRIGGER_PULL_OTABLE_INFO: {

148,776✔

3593

      SSTriggerOrigTableInfoRequest* pRequest = &(pReq->origTableInfoReq);

148,776✔

3594

      int32_t size = 0;

148,776✔

3595

      TAOS_CHECK_EXIT(tDecodeI32(&decoder, &size));

148,776✔

3596

      pRequest->cols = taosArrayInit(size, sizeof(OTableInfo));

148,776✔

3597

      if (pRequest->cols == NULL) {

148,776✔

3602

      for (int32_t i = 0; i < size; ++i) {

544,173✔

3603

        OTableInfo* oInfo = taosArrayReserve(pRequest->cols, 1);

395,397✔

3604

        if (oInfo == NULL) {

395,397✔

3609

        TAOS_CHECK_RETURN(tDecodeCStrTo(&decoder, oInfo->refTableName));

395,397✔

3610

        TAOS_CHECK_RETURN(tDecodeCStrTo(&decoder, oInfo->refColName));

395,397✔

3612

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

297,552✔

3614

      break;

148,776✔

3616

    default: {

3,146✔

3617

      uError("unknown pull type %d", type);

3,146✔

3623

  tEndDecode(&decoder);

57,050,826✔

3625

_exit:

57,049,141✔

3626

  tDecoderClear(&decoder);

57,049,997✔

3627

  return code;

57,047,873✔

3630

static int32_t tSerializeSTriggerCalcParam(SEncoder* pEncoder, SArray* pParams, bool ignoreNotificationInfo, bool full) {

44,162,932✔

3631

  int32_t size = full ? taosArrayGetSize(pParams) : 0;

44,162,932✔

3632

  int32_t code = 0;

44,165,006✔

3633

  int32_t lino = 0;

44,165,006✔

3634

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, size));

44,162,494✔

3635

  for (int32_t i = 0; i < size; ++i) {

2,147,483,647✔

3636

    SSTriggerCalcParam* param = taosArrayGet(pParams, i);

2,147,483,647✔

3637

    if (param == NULL) {

2,147,483,647✔

3640

    int64_t plainFieldSize = offsetof(SSTriggerCalcParam, notifyType);

2,147,483,647✔

3641

    if (pEncoder->data) {

2,147,483,647✔

3642

      TAOS_MEMCPY(pEncoder->data + pEncoder->pos, param, plainFieldSize);

2,147,483,647✔

3644

    pEncoder->pos += plainFieldSize;

2,147,483,647✔

3646

    if (!ignoreNotificationInfo) {

2,147,483,647✔

3647

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, param->notifyType));

1,520,846,722✔

3648

      uint64_t len = (param->extraNotifyContent != NULL) ? strlen(param->extraNotifyContent) + 1 : 0;

760,423,504✔

3649

      TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, (uint8_t*)param->extraNotifyContent, len));

1,400,521,388✔

3652

_exit:

7,595,424✔

3653

  return code;

7,595,424✔

3656

void tDestroySSTriggerCalcParam(void* ptr) {

2,147,483,647✔

3657

  SSTriggerCalcParam* pParam = ptr;

2,147,483,647✔

3658

  if (pParam && pParam->extraNotifyContent != NULL) {

2,147,483,647✔

3659

    taosMemoryFreeClear(pParam->extraNotifyContent);

316,615✔

3661

  if (pParam && pParam->resultNotifyContent != NULL) {

2,147,483,647✔

3664

  if (pParam && pParam->pExternalWindowData != NULL) {

2,147,483,647✔

3665

    taosArrayDestroyEx(pParam->pExternalWindowData, tDestroySStreamGroupValue);

62,599,024✔

3666

    pParam->pExternalWindowData = NULL;

62,587,194✔

3668

2,147,483,647✔

3670

void tDestroySSTriggerGroupCalcInfo(void* ptr) {

169,715✔

3671

  SSTriggerGroupCalcInfo* pCalcInfo = ptr;

169,715✔

3672

  if (pCalcInfo && pCalcInfo->pParams != NULL) {

169,715✔

3673

    taosArrayDestroyEx(pCalcInfo->pParams, tDestroySSTriggerCalcParam);

169,715✔

3674

    pCalcInfo->pParams = NULL;

169,715✔

3676

  if (pCalcInfo && pCalcInfo->pGroupColVals != NULL) {

169,715✔

3680

169,715✔

3698

void tDestroySStreamGroupValue(void* ptr) {

91,823,011✔

3699

  SStreamGroupValue* pValue = ptr;

91,823,011✔

3700

  if ((pValue != NULL) && (IS_VAR_DATA_TYPE(pValue->data.type) || pValue->data.type == TSDB_DATA_TYPE_DECIMAL)) {

91,823,011✔

3701

    taosMemoryFreeClear(pValue->data.pData);

20,513,874✔

3702

    pValue->data.nData = 0;

20,510,860✔

3704

91,823,011✔

3706

static int32_t tDeserializeSTriggerCalcParam(SDecoder* pDecoder, SArray**ppParams, bool ignoreNotificationInfo) {

22,073,106✔

3707

  int32_t size = 0, code = 0, lino = 0;

22,073,106✔

3708

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &size));

22,073,385✔

3709

  if (size <= 0) {

22,073,385✔

3710

    return code;

5,791,147✔

3713

  *ppParams = taosArrayInit(size, sizeof(SSTriggerCalcParam));

16,282,238✔

3714

  if (*ppParams == NULL) {

16,282,238✔

3717

  for (int32_t i = 0; i < size; ++i) {

2,147,483,647✔

3718

    SSTriggerCalcParam* param = taosArrayReserve(*ppParams, 1);

2,147,483,647✔

3719

    if (param == NULL) {

2,147,483,647✔

3722

    int64_t plainFieldSize = offsetof(SSTriggerCalcParam, notifyType);

2,147,483,647✔

3723

    TAOS_MEMCPY(param, pDecoder->data + pDecoder->pos, plainFieldSize);

2,147,483,647✔

3724

    pDecoder->pos += plainFieldSize;

2,147,483,647✔

3726

    if (!ignoreNotificationInfo) {

2,147,483,647✔

3727

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &param->notifyType));

728,557,526✔

3728

      uint64_t len = 0;

364,278,763✔

3729

      TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&param->extraNotifyContent, &len));

728,557,526✔

3733

_exit:

7,954,319✔

3734

  return code;

16,282,238✔

3737

static int32_t tSerializeStriggerGroupColVals(SEncoder* pEncoder, SArray* pGroupColVals, int32_t vgId) {

44,915,555✔

3738

  int32_t code = TSDB_CODE_SUCCESS;

44,915,555✔

3739

  int32_t lino = 0;

44,915,555✔

3741

  int32_t size = taosArrayGetSize(pGroupColVals);

44,915,555✔

3742

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, size));

44,916,116✔

3743

  for (int32_t i = 0; i < size; ++i) {

102,426,648✔

3744

    SStreamGroupValue* pValue = taosArrayGet(pGroupColVals, i);

57,510,075✔

3745

    if (pValue == NULL) {

57,509,889✔

3748

    TAOS_CHECK_EXIT(tEncodeBool(pEncoder, pValue->isNull));

57,509,889✔

3749

    if (pValue->isNull) {

57,509,941✔

3750

      continue;

19,980✔

3752

    TAOS_CHECK_EXIT(tEncodeBool(pEncoder, pValue->isTbname));

57,490,279✔

3753

    if (pValue->isTbname) {

57,490,270✔

3754

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pValue->uid));

52,913,472✔

3755

      if (vgId != -1) { pValue->vgId = vgId; }

26,456,736✔

3756

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pValue->vgId));

52,912,833✔

3758

    TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pValue->data.type));

114,977,061✔

3759

    if (IS_VAR_DATA_TYPE(pValue->data.type)) {

57,490,428✔

3760

      TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pValue->data.pData, pValue->data.nData));

77,602,290✔

3762

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pValue->data.val));

37,378,690✔

3766

_exit:

44,916,573✔

3767

  return code;

44,916,573✔

3770

static int32_t tDeserializeStriggerGroupColVals(SDecoder* pDecoder, SArray** ppGroupColVals) {

22,448,564✔

3771

  int32_t code = TSDB_CODE_SUCCESS;

22,448,564✔

3772

  int32_t lino = 0;

22,448,564✔

3773

  int32_t size = 0;

22,448,564✔

3775

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &size));

22,449,127✔

3776

  taosArrayClearEx(*ppGroupColVals, tDestroySStreamGroupValue);

22,449,127✔

3777

  if (size > 0) {

22,447,953✔

3778

    if (*ppGroupColVals == NULL) {

14,583,786✔

3779

      *ppGroupColVals = taosArrayInit(size, sizeof(SStreamGroupValue));

14,584,098✔

3780

      if (*ppGroupColVals == NULL) {

14,583,465✔

3787

  for (int32_t i = 0; i < size; ++i) {

51,198,228✔

3788

    SStreamGroupValue* pValue = taosArrayReserve(*ppGroupColVals, 1);

28,749,692✔

3789

    if (pValue == NULL) {

28,750,581✔

3792

    TAOS_CHECK_EXIT(tDecodeBool(pDecoder, &pValue->isNull));

28,750,581✔

3793

    if (pValue->isNull) {

28,750,312✔

3794

      continue;

9,990✔

3796

    TAOS_CHECK_EXIT(tDecodeBool(pDecoder, &pValue->isTbname));

28,740,010✔

3797

    if (pValue->isTbname) {

28,740,433✔

3798

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pValue->uid));

26,449,063✔

3799

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pValue->vgId));

26,449,672✔

3801

    TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pValue->data.type));

57,481,633✔

3802

    if (IS_VAR_DATA_TYPE(pValue->data.type)) {

48,136,109✔

3803

      uint64_t len = 0;

19,395,490✔

3804

      TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pValue->data.pData, &len));

38,790,495✔

3805

      pValue->data.nData = len;

19,395,005✔

3807

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pValue->data.val));

18,690,202✔

3810

_exit:

22,448,820✔

3811

  return code;

22,448,536✔

3814

int32_t tSerializeSStreamGroupInfo(void* buf, int32_t bufLen, const SStreamGroupInfo* gInfo, int32_t vgId) {

750,800✔

3815

  SEncoder encoder = {0};

750,800✔

3816

  int32_t  code = TSDB_CODE_SUCCESS;

750,800✔

3817

  int32_t  lino = 0;

750,800✔

3818

  int32_t  tlen = 0;

750,800✔

3820

  tEncoderInit(&encoder, buf, bufLen);

750,800✔

3821

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

750,800✔

3823

  TAOS_CHECK_EXIT(tSerializeStriggerGroupColVals(&encoder, gInfo->gInfo, vgId));

750,800✔

3825

  tEndEncode(&encoder);

750,800✔

3827

_exit:

750,578✔

3828

  if (code != TSDB_CODE_SUCCESS) {

750,578✔

3831

    tlen = encoder.pos;

750,578✔

3833

  tEncoderClear(&encoder);

750,578✔

3834

  return tlen;

750,800✔

3837

int32_t tDeserializeSStreamGroupInfo(void* buf, int32_t bufLen, SStreamGroupInfo* gInfo) {

375,742✔

3838

  SDecoder decoder = {0};

375,742✔

3839

  int32_t  code = TSDB_CODE_SUCCESS;

375,742✔

3840

  int32_t  lino = 0;

375,742✔

3841

  int32_t  size = 0;

375,742✔

3843

  tDecoderInit(&decoder, buf, bufLen);

375,742✔

3844

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

375,742✔

3846

  TAOS_CHECK_EXIT(tDeserializeStriggerGroupColVals(&decoder, &gInfo->gInfo));

375,742✔

3848

  tEndDecode(&decoder);

375,458✔

3850

_exit:

375,742✔

3851

  tDecoderClear(&decoder);

375,742✔

3852

  return code;

375,742✔

3855

static int32_t tSerializeSSTriggerGroupCalcInfo(SEncoder* pEncoder, SSTriggerGroupCalcInfo* pInfo) {

25,480✔

3856

  int32_t code = 0;

25,480✔

3857

  int32_t lino = 0;

25,480✔

3859

  TAOS_CHECK_EXIT(tSerializeSTriggerCalcParam(pEncoder, pInfo->pParams, false, true));

25,480✔

3860

  TAOS_CHECK_EXIT(tSerializeStriggerGroupColVals(pEncoder, pInfo->pGroupColVals, -1));

25,480✔

3861

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pInfo->createTable));

50,960✔

3863

_exit:

25,480✔

3864

  return code;

25,480✔

3893

int32_t tSerializeSTriggerCalcRequest(void* buf, int32_t bufLen, const SSTriggerCalcRequest* pReq) {

7,137,121✔

3894

  SEncoder encoder = {0};

7,137,121✔

3895

  int32_t  code = TSDB_CODE_SUCCESS;

7,137,582✔

3896

  int32_t  lino = 0;

7,137,582✔

3897

  int32_t  tlen = 0;

7,137,582✔

3899

  tEncoderInit(&encoder, buf, bufLen);

7,137,582✔

3900

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

7,137,582✔

3902

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->streamId));

14,274,616✔

3903

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->runnerTaskId));

14,274,703✔

3904

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->sessionId));

14,274,242✔

3905

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, pReq->triggerType));

14,274,703✔

3906

  TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->isMultiGroupCalc));

14,275,164✔

3907

  TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->stbPartByTbname));

14,273,600✔

3909

  if (!pReq->isMultiGroupCalc) {

7,136,018✔

3910

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->gid));

14,274,890✔

3911

    TAOS_CHECK_EXIT(tSerializeSTriggerCalcParam(&encoder, pReq->params, false, true));

7,137,308✔

3912

    TAOS_CHECK_EXIT(tSerializeStriggerGroupColVals(&encoder, pReq->groupColVals, -1));

7,137,034✔

3913

    TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->createTable));

14,274,061✔

3944

  TAOS_CHECK_EXIT(tEncodeBool(&encoder, pReq->isWindowTrigger));

7,137,582✔

3945

  TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->precision));

14,274,890✔

3947

  tEndEncode(&encoder);

7,137,308✔

3949

_exit:

7,137,034✔

3950

  if (code != TSDB_CODE_SUCCESS) {

7,137,034✔

3953

    tlen = encoder.pos;

7,137,034✔

3955

  tEncoderClear(&encoder);

7,137,034✔

3956

  return tlen;

7,136,753✔

3959

static int32_t tDeserializeSSTriggerGroupCalcInfo(SDecoder* pDecoder, SSTriggerGroupCalcInfo* pInfo) {

12,740✔

3960

  int32_t code = 0;

12,740✔

3961

  int32_t lino = 0;

12,740✔

3963

  TAOS_CHECK_EXIT(tDeserializeSTriggerCalcParam(pDecoder, &pInfo->pParams, false));

12,740✔

3964

  TAOS_CHECK_EXIT(tDeserializeStriggerGroupColVals(pDecoder, &pInfo->pGroupColVals));

12,740✔

3965

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pInfo->createTable));

25,480✔

3967

_exit:

12,740✔

3968

  return code;

12,740✔

3995

int32_t tDeserializeSTriggerCalcRequest(void* buf, int32_t bufLen, SSTriggerCalcRequest* pReq) {

3,567,040✔

3996

  SDecoder decoder = {0};

3,567,040✔

3997

  int32_t  code = TSDB_CODE_SUCCESS;

3,567,040✔

3998

  int32_t  lino = 0;

3,567,040✔

4000

  tDecoderInit(&decoder, buf, bufLen);

3,567,040✔

4001

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

3,566,492✔

4003

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->streamId));

7,134,080✔

4004

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->runnerTaskId));

7,134,080✔

4005

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->sessionId));

7,134,080✔

4006

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pReq->triggerType));

7,134,080✔

4007

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->isMultiGroupCalc));

7,134,080✔

4008

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->stbPartByTbname));

7,134,080✔

4010

  if (!pReq->isMultiGroupCalc) {

3,567,040✔

4011

    TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->gid));

7,134,080✔

4012

    TAOS_CHECK_EXIT(tDeserializeSTriggerCalcParam(&decoder, &pReq->params, false));

3,567,040✔

4013

    TAOS_CHECK_EXIT(tDeserializeStriggerGroupColVals(&decoder, &pReq->groupColVals));

3,567,040✔

4014

    TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->createTable));

7,134,080✔

4055

  if (!tDecodeIsEnd(&decoder)) {

3,567,040✔

4056

    TAOS_CHECK_EXIT(tDecodeBool(&decoder, &pReq->isWindowTrigger));

3,567,040✔

4057

    TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->precision));

7,134,080✔

4060

  tEndDecode(&decoder);

3,567,040✔

4062

_exit:

3,567,040✔

4063

  tDecoderClear(&decoder);

3,567,040✔

4064

  return code;

3,567,040✔

4067

void tDestroySTriggerCalcRequest(SSTriggerCalcRequest* pReq) {

10,557,824✔

4068

  if (pReq != NULL) {

10,557,824✔

4069

    if (pReq->params != NULL) {

10,558,098✔

4070

      taosArrayDestroyEx(pReq->params, tDestroySSTriggerCalcParam);

6,808,761✔

4071

      pReq->params = NULL;

6,808,761✔

4073

    if (pReq->groupColVals != NULL) {

10,558,098✔

4074

      taosArrayDestroyEx(pReq->groupColVals, tDestroySStreamGroupValue);

3,789,364✔

4075

      pReq->groupColVals = NULL;

3,789,364✔

4077

    if (pReq->pGroupCalcInfos != NULL) {

10,558,098✔

4081

    if (pReq->pGroupReadInfos != NULL) {

10,558,098✔

4085

    blockDataDestroy(pReq->pOutBlock);

10,558,098✔

4086

    pReq->pOutBlock = NULL;

10,558,098✔

4088

10,557,824✔

4149

int32_t tSerializeSTriggerCtrlRequest(void* buf, int32_t bufLen, const SSTriggerCtrlRequest* pReq) {

70,272,180✔

4150

  SEncoder encoder = {0};

70,272,180✔

4151

  int32_t  code = TSDB_CODE_SUCCESS;

70,272,180✔

4152

  int32_t  lino = 0;

70,272,180✔

4153

  int32_t  tlen = 0;

70,272,180✔

4155

  tEncoderInit(&encoder, buf, bufLen);

70,272,180✔

4156

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

70,271,608✔

4158

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, pReq->type));

140,543,788✔

4159

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->streamId));

140,544,074✔

4160

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->taskId));

140,544,360✔

4161

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->sessionId));

140,544,360✔

4163

  tEndEncode(&encoder);

70,272,180✔

4165

_exit:

70,271,894✔

4166

  if (code != TSDB_CODE_SUCCESS) {

70,271,894✔

4169

    tlen = encoder.pos;

70,271,894✔

4171

  tEncoderClear(&encoder);

70,271,894✔

4172

  return tlen;

70,272,180✔

4175

int32_t tDeserializeSTriggerCtrlRequest(void* buf, int32_t bufLen, SSTriggerCtrlRequest* pReq) {

105,374,113✔

4176

  SDecoder decoder = {0};

105,374,113✔

4177

  int32_t  code = TSDB_CODE_SUCCESS;

105,376,465✔

4178

  int32_t  lino = 0;

105,376,465✔

4180

  tDecoderInit(&decoder, buf, bufLen);

105,376,465✔

4181

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

105,367,081✔

4183

  int32_t type = 0;

105,390,947✔

4184

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &type));

105,385,675✔

4185

  pReq->type = type;

105,385,675✔

4186

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->streamId));

210,779,738✔

4187

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->taskId));

210,790,150✔

4188

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->sessionId));

210,786,499✔

4190

  tEndDecode(&decoder);

105,389,941✔

4192

_exit:

105,387,517✔

4193

  tDecoderClear(&decoder);

105,388,027✔

4194

  return code;

105,383,231✔

4197

int32_t tSerializeStRtFuncInfo(SEncoder* pEncoder, const SStreamRuntimeFuncInfo* pInfo, bool needStreamRtInfo, bool needStreamGrpInfo) {

37,014,106✔

4198

  int32_t code = 0, lino = 0;

37,014,106✔

4199

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pInfo->isMultiGroupCalc));

74,031,265✔

4200

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pInfo->stbPartByTbname));

74,033,248✔

4201

  TAOS_CHECK_EXIT(tEncodeBool(pEncoder, needStreamRtInfo));

37,016,089✔

4202

  TAOS_CHECK_EXIT(tEncodeBool(pEncoder, needStreamGrpInfo));

37,015,827✔

4203

  if (pInfo->isMultiGroupCalc) {

37,015,972✔

4204

    if (needStreamRtInfo) {

14,560✔

4205

      if (needStreamGrpInfo) {

14,560✔

4206

        int32_t nGroups = taosArrayGetSize(pInfo->curGrpRead);

14,560✔

4207

        TAOS_CHECK_EXIT(tEncodeI32(pEncoder, nGroups));

14,560✔

4208

        int8_t withGrpCalcInfo = 1;

14,560✔

4209

        for (int32_t i = 0; i < nGroups; ++i) {

14,560✔

4217

        TAOS_CHECK_EXIT(tEncodeI8(pEncoder, withGrpCalcInfo));

29,120✔

4218

        if (withGrpCalcInfo) {

14,560✔

4219

          int32_t nGroups = tSimpleHashGetSize(pInfo->pGroupCalcInfos);

14,560✔

4220

          TAOS_CHECK_EXIT(tEncodeI32(pEncoder, nGroups));

14,560✔

4221

          int32_t                 iter1 = 0;

14,560✔

4222

          SSTriggerGroupCalcInfo* pCalcInfo = tSimpleHashIterate(pInfo->pGroupCalcInfos, NULL, &iter1);

14,560✔

4223

          while (pCalcInfo != NULL) {

40,040✔

4224

            int64_t* gid = tSimpleHashGetKey(pCalcInfo, NULL);

25,480✔

4225

            TAOS_CHECK_EXIT(tEncodeI64(pEncoder, *gid));

50,960✔

4226

            TAOS_CHECK_EXIT(tSerializeSSTriggerGroupCalcInfo(pEncoder, pCalcInfo));

25,480✔

4227

            pCalcInfo = tSimpleHashIterate(pInfo->pGroupCalcInfos, pCalcInfo, &iter1);

25,480✔

4261

    TAOS_CHECK_EXIT(tSerializeSTriggerCalcParam(pEncoder, pInfo->pStreamPesudoFuncVals, true, needStreamRtInfo));

37,002,315✔

4262

    TAOS_CHECK_EXIT(tSerializeStriggerGroupColVals(pEncoder, pInfo->pStreamPartColVals, -1));

37,000,932✔

4263

    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pInfo->groupId));

74,004,716✔

4266

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pInfo->curWindow.skey));

74,033,088✔

4267

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pInfo->curWindow.ekey));

74,033,939✔

4268

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pInfo->curIdx));

74,033,769✔

4269

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pInfo->sessionId));

74,035,129✔

4270

  TAOS_CHECK_EXIT(tEncodeBool(pEncoder, pInfo->withExternalWindow));

37,018,091✔

4271

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pInfo->triggerType));

74,036,004✔

4272

  TAOS_CHECK_EXIT(tEncodeBool(pEncoder, pInfo->isWindowTrigger));

37,017,918✔

4273

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pInfo->precision));

74,034,736✔

4274

_exit:

37,016,929✔

4275

  return code;

37,017,523✔

4278

int32_t tDeserializeStRtFuncInfo(SDecoder* pDecoder, SStreamRuntimeFuncInfo* pInfo) {

18,500,877✔

4279

  int32_t code = 0, lino = 0;

18,500,877✔

4280

  int32_t size = 0;

18,500,877✔

4281

  bool needStreamRtInfo = false;

18,500,877✔

4282

  bool needStreamGrpInfo = false;

18,500,877✔

4284

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pInfo->isMultiGroupCalc));

37,001,179✔

4285

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pInfo->stbPartByTbname));

37,002,041✔

4286

  TAOS_CHECK_EXIT(tDecodeBool(pDecoder, &needStreamRtInfo));

18,501,169✔

4287

  TAOS_CHECK_EXIT(tDecodeBool(pDecoder, &needStreamGrpInfo));

18,501,169✔

4289

  if (pInfo->isMultiGroupCalc) {

18,501,169✔

4290

    if (needStreamRtInfo) {

7,280✔

4291

      if (needStreamGrpInfo) {

7,280✔

4292

        int32_t nGroups = 0;

7,280✔

4293

        TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &nGroups));

7,280✔

4294

        if (nGroups > 0) {

7,280✔

4298

        for (int32_t j = 0; j < nGroups; ++j) {

7,280✔

4302

        int8_t withGrpCalcInfo = 0;

7,280✔

4303

        TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &withGrpCalcInfo));

7,280✔

4304

        if (withGrpCalcInfo) {

7,280✔

4305

          int32_t nGroups = 0;

7,280✔

4306

          TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &nGroups));

7,280✔

4307

          pInfo->pGroupCalcInfos = tSimpleHashInit(nGroups, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));

7,280✔

4308

          QUERY_CHECK_NULL(pInfo->pGroupCalcInfos, code, lino, _exit, terrno);

7,280✔

4309

          tSimpleHashSetFreeFp(pInfo->pGroupCalcInfos, tDestroySSTriggerGroupCalcInfo);

7,280✔

4311

          for (int32_t i = 0; i < nGroups; i++) {

20,020✔

4312

            int64_t gid = 0;

12,740✔

4313

            TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &gid));

12,740✔

4314

            SSTriggerGroupCalcInfo info = {0};

12,740✔

4315

            TAOS_CHECK_EXIT(tSimpleHashPut(pInfo->pGroupCalcInfos, &gid, sizeof(int64_t), &info, sizeof(info)));

12,740✔

4316

            SSTriggerGroupCalcInfo* pCalcInfo = tSimpleHashGet(pInfo->pGroupCalcInfos, &gid, sizeof(int64_t));

12,740✔

4317

            QUERY_CHECK_NULL(pCalcInfo, code, lino, _exit, TSDB_CODE_INTERNAL_ERROR);

12,740✔

4318

            TAOS_CHECK_EXIT(tDeserializeSSTriggerGroupCalcInfo(pDecoder, pCalcInfo));

12,740✔

4363

    TAOS_CHECK_EXIT(tDeserializeSTriggerCalcParam(pDecoder, &pInfo->pStreamPesudoFuncVals, true));

18,493,889✔

4364

    TAOS_CHECK_EXIT(tDeserializeStriggerGroupColVals(pDecoder, &pInfo->pStreamPartColVals));

18,493,313✔

4365

    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pInfo->groupId));

36,986,333✔

4368

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pInfo->curWindow.skey));

37,001,768✔

4369

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pInfo->curWindow.ekey));

37,002,338✔

4370

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pInfo->curIdx));

37,002,338✔

4371

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pInfo->sessionId));

37,002,338✔

4372

  TAOS_CHECK_EXIT(tDecodeBool(pDecoder, &pInfo->withExternalWindow));

18,501,169✔

4373

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pInfo->triggerType));

37,002,338✔

4374

  if (!tDecodeIsEnd(pDecoder)) {

18,500,131✔

4375

    TAOS_CHECK_EXIT(tDecodeBool(pDecoder, &pInfo->isWindowTrigger));

18,500,131✔

4376

    TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pInfo->precision));

37,002,026✔

4378

_exit:

18,501,169✔

4379

  return code;

18,501,169✔

4382

void tDestroyStRtFuncInfo(SStreamRuntimeFuncInfo* pInfo){

590,637,677✔

4383

  if (pInfo == NULL) return;

590,637,677✔

4384

  if (pInfo->pStreamPesudoFuncVals != NULL) {

30,089,459✔

4385

    taosArrayDestroyEx(pInfo->pStreamPesudoFuncVals, tDestroySSTriggerCalcParam);

10,586,548✔

4386

    pInfo->pStreamPesudoFuncVals = NULL;

10,586,548✔

4388

  if (pInfo->pStreamPartColVals != NULL) {

30,106,664✔

4389

    taosArrayDestroyEx(pInfo->pStreamPartColVals, tDestroySStreamGroupValue);

11,047,328✔

4390

    pInfo->pStreamPartColVals = NULL;

11,047,044✔

4392

  if (pInfo->pGroupCalcInfos != NULL) {

30,106,419✔

4393

    tSimpleHashCleanup(pInfo->pGroupCalcInfos);

86,450✔

4394

    pInfo->pGroupCalcInfos = NULL;

86,450✔

4396

  if (pInfo->pGroupReadInfos != NULL) {

30,107,248✔

4400

  if (pInfo->outNormalTable != NULL) {

30,107,248✔

4405

int32_t tSerializeSStreamMsgVTableInfo(void* buf, int32_t bufLen, const SStreamMsgVTableInfo* pRsp){

171,198✔

4406

  SEncoder encoder = {0};

171,198✔

4407

  int32_t  code = TSDB_CODE_SUCCESS;

171,198✔

4408

  int32_t  lino = 0;

171,198✔

4409

  int32_t  tlen = 0;

171,198✔

4411

  tEncoderInit(&encoder, buf, bufLen);

171,198✔

4412

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

171,198✔

4414

  int32_t size = taosArrayGetSize(pRsp->infos);

171,198✔

4415

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, size));

171,198✔

4416

  for (int32_t i = 0; i < size; ++i) {

584,388✔

4417

    VTableInfo* info = taosArrayGet(pRsp->infos, i);

413,190✔

4418

    if (info == NULL) {

413,190✔

4421

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, info->gId));

826,380✔

4422

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, info->uid));

826,380✔

4423

    TAOS_CHECK_EXIT(tEncodeSColRefWrapper(&encoder, &info->cols));

413,190✔

4426

  tEndEncode(&encoder);

171,198✔

4428

_exit:

170,931✔

4429

  if (code != TSDB_CODE_SUCCESS) {

170,931✔

4432

    tlen = encoder.pos;

170,931✔

4434

  tEncoderClear(&encoder);

170,931✔

4435

  return tlen;

171,198✔

4438

int32_t tDeserializeSStreamMsgVTableInfo(void* buf, int32_t bufLen, SStreamMsgVTableInfo *vTableInfo){

85,599✔

4439

  SDecoder decoder = {0};

85,599✔

4440

  int32_t  code = TSDB_CODE_SUCCESS;

85,599✔

4441

  int32_t  lino = 0;

85,599✔

4442

  int32_t  size = 0;

85,599✔

4444

  tDecoderInit(&decoder, buf, bufLen);

85,599✔

4445

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

85,599✔

4447

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &size));

85,599✔

4448

  vTableInfo->infos = taosArrayInit(size, sizeof(VTableInfo));

85,599✔

4449

  if (vTableInfo->infos == NULL) {

85,599✔

4452

  for (int32_t i = 0; i < size; ++i) {

292,194✔

4453

    VTableInfo* info = taosArrayReserve(vTableInfo->infos, 1);

206,595✔

4454

    if (info == NULL) {

206,595✔

4457

    TAOS_CHECK_EXIT(tDecodeI64(&decoder, &info->gId));

413,190✔

4458

    TAOS_CHECK_EXIT(tDecodeI64(&decoder, &info->uid));

413,190✔

4459

    TAOS_CHECK_EXIT(tDecodeSColRefWrapperEx(&decoder, &info->cols, false));

206,595✔

4462

  tEndDecode(&decoder);

85,599✔

4464

_exit:

85,599✔

4465

  tDecoderClear(&decoder);

85,599✔

4466

  return code;

85,599✔

4470

void tDestroyVTableInfo(void *ptr) {

413,190✔

4471

  if (NULL == ptr) {

413,190✔

4474

  VTableInfo* pTable = (VTableInfo*)ptr;

413,190✔

4475

  taosMemoryFree(pTable->cols.pColRef);

413,190✔

4478

void tDestroySStreamMsgVTableInfo(SStreamMsgVTableInfo *ptr) {

46,731,392✔

4479

  if (ptr == NULL) return;

46,731,392✔

4480

  taosArrayDestroyEx(ptr->infos, tDestroyVTableInfo);

46,731,392✔

4481

  ptr->infos = NULL;

46,731,079✔

4484

int32_t tSerializeSStreamTsResponse(void* buf, int32_t bufLen, const SStreamTsResponse* pRsp) {

1,462,404✔

4485

  SEncoder encoder = {0};

1,462,404✔

4486

  int32_t  code = TSDB_CODE_SUCCESS;

1,462,404✔

4487

  int32_t  lino = 0;

1,462,404✔

4488

  int32_t  tlen = 0;

1,462,404✔

4490

  tEncoderInit(&encoder, buf, bufLen);

1,462,404✔

4491

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

1,462,688✔

4493

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRsp->ver));

2,925,092✔

4494

  int32_t size = taosArrayGetSize(pRsp->tsInfo);

1,462,688✔

4495

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, size));

1,462,404✔

4496

  for (int32_t i = 0; i < size; ++i) {

3,471,984✔

4497

    STsInfo* tsInfo = taosArrayGet(pRsp->tsInfo, i);

2,009,294✔

4498

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, tsInfo->gId));

4,018,874✔

4499

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, tsInfo->ts));

4,018,876✔

4502

  tEndEncode(&encoder);

1,462,690✔

4504

_exit:

1,462,402✔

4505

  if (code != TSDB_CODE_SUCCESS) {

1,462,402✔

4508

    tlen = encoder.pos;

1,462,402✔

4510

  tEncoderClear(&encoder);

1,462,402✔

4511

  return tlen;

1,460,693✔

4514

int32_t tDeserializeSStreamTsResponse(void* buf, int32_t bufLen, void *pBlock) {

731,896✔

4515

  SDecoder decoder = {0};

731,896✔

4516

  int32_t  code = TSDB_CODE_SUCCESS;

731,896✔

4517

  int32_t  lino = 0;

731,896✔

4518

  SSDataBlock *pResBlock = pBlock;

731,896✔

4520

  tDecoderInit(&decoder, buf, bufLen);

731,896✔

4521

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

731,896✔

4523

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, (int64_t*)&pResBlock->info.id.groupId));

1,463,792✔

4524

  int32_t numOfCols = 2;

731,896✔

4525

  if (pResBlock->pDataBlock == NULL) {

731,896✔

4526

    pResBlock->pDataBlock = taosArrayInit_s(sizeof(SColumnInfoData), numOfCols);

731,896✔

4527

    if (pResBlock->pDataBlock == NULL) {

731,896✔

4530

    for (int32_t i = 0; i< numOfCols; ++i) {

2,195,688✔

4531

      SColumnInfoData *pColInfoData = taosArrayGet(pResBlock->pDataBlock, i);

1,463,792✔

4532

      if (pColInfoData == NULL) {

1,463,792✔

4535

      pColInfoData->info.type = TSDB_DATA_TYPE_BIGINT;

1,463,792✔

4536

      pColInfoData->info.bytes = sizeof(int64_t);

1,463,792✔

4539

  int32_t numOfRows = 0;

731,896✔

4540

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &numOfRows));

731,896✔

4541

  TAOS_CHECK_EXIT(blockDataEnsureCapacity(pResBlock, numOfRows));

731,896✔

4542

  for (int32_t i = 0; i < numOfRows; ++i) {

1,737,796✔

4543

    for (int32_t j = 0; j < numOfCols; ++j) {

3,018,236✔

4544

      SColumnInfoData *pColInfoData = taosArrayGet(pResBlock->pDataBlock, j);

2,012,336✔

4545

      if (pColInfoData == NULL) {

2,012,336✔

4548

      int64_t value = 0;

2,012,336✔

4549

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &value));

2,012,068✔

4550

      colDataSetInt64(pColInfoData, i, &value);

2,012,068✔

4554

  pResBlock->info.dataLoad = 1;

731,628✔

4555

  pResBlock->info.rows = numOfRows;

731,896✔

4557

  tEndDecode(&decoder);

731,896✔

4559

_exit:

731,896✔

4560

  tDecoderClear(&decoder);

731,896✔

4561

  return code;

731,896✔

4564

static int32_t encodeData(SEncoder* encoder, void* pBlock, SSHashObj* indexHash) {

3,308,630✔

4565

  int32_t code = TSDB_CODE_SUCCESS;

3,308,630✔

4566

  int32_t lino = 0;

3,308,630✔

4567

  int32_t len = 0;

3,308,630✔

4568

  if (encoder->data == NULL){

3,308,630✔

4569

    len = blockGetEncodeSize(pBlock);

1,654,030✔

4571

    len = blockEncode(pBlock, (char*)(encoder->data + encoder->pos), encoder->size - encoder->pos, blockDataGetNumOfCols(pBlock));

1,654,600✔

4572

    if (len < 0) {

1,653,539✔

4576

  encoder->pos += len;

3,308,139✔

4578

  if (indexHash == NULL) {

3,308,139✔

4579

    goto _exit;

1,606,961✔

4582

  uint32_t pos = encoder->pos;

1,701,178✔

4583

  encoder->pos += sizeof(uint32_t); // reserve space for tables

1,701,748✔

4584

  int32_t tables = 0;

1,701,178✔

4586

  void*   pe = NULL;

1,701,178✔

4587

  int32_t iter = 0;

1,701,178✔

4588

  while ((pe = tSimpleHashIterate(indexHash, pe, &iter)) != NULL) {

3,667,280✔

4589

    SStreamWalDataSlice* pInfo = (SStreamWalDataSlice*)pe;

1,966,330✔

4590

    if (pInfo->gId == -1){

1,966,330✔

4593

    int64_t uid = *(int64_t*)(tSimpleHashGetKey(pe, NULL));

1,966,330✔

4594

    TAOS_CHECK_EXIT(tEncodeI64(encoder, uid));

1,966,330✔

4595

    TAOS_CHECK_EXIT(tEncodeU64(encoder, pInfo->gId));

3,933,230✔

4596

    TAOS_CHECK_EXIT(tEncodeI32(encoder, pInfo->startRowIdx));

3,933,288✔

4597

    TAOS_CHECK_EXIT(tEncodeI32(encoder, pInfo->numRows));

3,932,490✔

4598

    tables++;

1,966,102✔

4600

  uint32_t tmpPos = encoder->pos;

1,701,464✔

4601

  encoder->pos = pos;

1,701,464✔

4602

  TAOS_CHECK_EXIT(tEncodeI32(encoder, tables));

1,700,610✔

4603

  encoder->pos = tmpPos;

1,700,610✔

4604

_exit:

3,308,425✔

4605

  return code;

3,308,425✔

4608

static int32_t encodeBlock(SEncoder* encoder, void* block, SSHashObj* indexHash) {

11,367,101✔

4609

  int32_t  code = TSDB_CODE_SUCCESS;

11,367,101✔

4610

  int32_t  lino = 0;

11,367,101✔

4611

  if (block != NULL && ((SSDataBlock*)block)->info.rows > 0) {

11,367,101✔

4612

    TAOS_CHECK_EXIT(tEncodeI8(encoder, 1));

3,309,200✔

4613

    TAOS_CHECK_EXIT(encodeData(encoder, block, indexHash));

3,309,200✔

4615

    TAOS_CHECK_EXIT(tEncodeI8(encoder, 0));

8,059,609✔

4618

_exit:

8,059,609✔

4619

  return code;

11,368,034✔

4622

int32_t tSerializeSStreamWalDataResponse(void* buf, int32_t bufLen, SSTriggerWalNewRsp* rsp) {

2,841,820✔

4623

  SEncoder encoder = {0};

2,841,820✔

4624

  int32_t  code = TSDB_CODE_SUCCESS;

2,841,820✔

4625

  int32_t  lino = 0;

2,841,820✔

4626

  int32_t  tlen = 0;

2,841,820✔

4628

  tEncoderInit(&encoder, buf, bufLen);

2,841,820✔

4629

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

2,842,396✔

4631

  TAOS_CHECK_EXIT(encodeBlock(&encoder, rsp->dataBlock, rsp->indexHash));

2,842,396✔

4632

  TAOS_CHECK_EXIT(encodeBlock(&encoder, rsp->metaBlock, NULL));

2,841,258✔

4633

  TAOS_CHECK_EXIT(encodeBlock(&encoder, rsp->deleteBlock, NULL));

2,841,335✔

4634

  TAOS_CHECK_EXIT(encodeBlock(&encoder, rsp->tableBlock, NULL));

2,842,396✔

4636

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, rsp->ver));

5,683,938✔

4637

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, rsp->verTime));

5,684,224✔

4638

  tEndEncode(&encoder);

2,842,112✔

4640

_exit:

2,841,335✔

4641

  if (code != TSDB_CODE_SUCCESS) {

2,841,335✔

4644

    tlen = encoder.pos;

2,841,335✔

4646

  tEncoderClear(&encoder);

2,841,335✔

4647

  return tlen;

2,841,335✔

4650

static int32_t decodeBlock(SDecoder* decoder, void* pBlock) {

4,264,387✔

4651

  int32_t  code = TSDB_CODE_SUCCESS;

4,264,387✔

4652

  int32_t  lino = 0;

4,264,387✔

4654

  int8_t hasData = false;

4,264,387✔

4655

  TAOS_CHECK_EXIT(tDecodeI8(decoder, &hasData));

4,264,962✔

4656

  if (hasData) {

4,264,962✔

4657

    TAOS_CHECK_EXIT(pBlock != NULL ? TSDB_CODE_SUCCESS : TSDB_CODE_INVALID_PARA);

803,878✔

4658

    const char* pEndPos = NULL;

803,878✔

4659

    TAOS_CHECK_EXIT(blockDecode(pBlock, (char*)decoder->data + decoder->pos, &pEndPos));

803,878✔

4660

    decoder->pos = (uint8_t*)pEndPos - decoder->data;

803,878✔

4661

  } else if (pBlock != NULL) {

3,461,084✔

4662

    blockDataEmpty(pBlock);

1,605,392✔

4665

_exit:

4,262,118✔

4666

  return code;

4,264,962✔

4669

int32_t tDeserializeSStreamWalDataResponse(void* buf, int32_t bufLen, SSTriggerWalNewRsp* pRsp, SArray* pSlices){

1,421,654✔

4670

  SDecoder     decoder = {0};

1,421,654✔

4671

  int32_t      code = TSDB_CODE_SUCCESS;

1,421,654✔

4672

  int32_t      lino = 0;

1,421,654✔

4673

  SSDataBlock* pBlock = NULL;

1,421,654✔

4675

  tDecoderInit(&decoder, buf, bufLen);

1,421,654✔

4676

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

1,421,654✔

4679

  int8_t hasData = false;

1,421,654✔

4680

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &hasData));

1,421,654✔

4681

  pBlock = pRsp->dataBlock;

1,421,654✔

4682

  if (hasData) {

1,421,654✔

4683

    TAOS_CHECK_EXIT(pBlock != NULL ? TSDB_CODE_SUCCESS : TSDB_CODE_INVALID_PARA);

851,178✔

4684

    const char* pEndPos = NULL;

851,178✔

4685

    TAOS_CHECK_EXIT(blockDecode(pBlock, (char*)decoder.data + decoder.pos, &pEndPos));

851,178✔

4686

    decoder.pos = (uint8_t*)pEndPos - decoder.data;

851,178✔

4688

    int32_t nSlices = 0;

851,178✔

4689

    TAOS_CHECK_EXIT(tDecodeI32(&decoder, &nSlices));

851,178✔

4690

    TAOS_CHECK_EXIT(taosArrayEnsureCap(pSlices, nSlices));

851,178✔

4691

    taosArrayClear(pSlices);

851,178✔

4692

    int64_t  uid = 0;

851,178✔

4693

    uint64_t gid = 0;

851,178✔

4694

    int32_t  startIdx = 0;

851,178✔

4695

    int32_t  numRows = 0;

851,178✔

4696

    for (int32_t i = 0; i < nSlices; i++) {

1,834,932✔

4697

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &uid));

983,754✔

4698

      TAOS_CHECK_EXIT(tDecodeU64(&decoder, &gid));

983,754✔

4699

      TAOS_CHECK_EXIT(tDecodeI32(&decoder, &startIdx));

983,754✔

4700

      TAOS_CHECK_EXIT(tDecodeI32(&decoder, &numRows));

983,754✔

4701

      int32_t endIdx = startIdx + numRows;

983,754✔

4702

      int64_t value[3] = {gid, uid, (int64_t)startIdx << 32 | endIdx};

983,754✔

4703

      void*   px = taosArrayPush(pSlices, value);

983,754✔

4704

      if (px == NULL) {

983,754✔

4709

  } else if (pBlock != NULL) {

570,476✔

4710

    blockDataEmpty(pBlock);

5,342✔

4711

    taosArrayClear(pSlices);

5,342✔

4714

  TAOS_CHECK_EXIT(decodeBlock(&decoder, pRsp->metaBlock));

1,421,654✔

4715

  TAOS_CHECK_EXIT(decodeBlock(&decoder, pRsp->deleteBlock));

1,421,654✔

4716

  TAOS_CHECK_EXIT(decodeBlock(&decoder, pRsp->tableBlock));

1,421,654✔

4718

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRsp->ver));

2,843,308✔

4719

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRsp->verTime));

2,843,308✔

4721

  tEndDecode(&decoder);

1,421,654✔

4723

_exit:

1,421,654✔

4724

  if (code != TSDB_CODE_SUCCESS) {

1,421,654✔

4727

  tDecoderClear(&decoder);

1,421,654✔

4728

  return code;

1,421,654✔

taosdata / TDengine / #5052

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #5052

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