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

426

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

622,304✔

427

  int32_t code = 0;

622,304✔

428

  int32_t lino = 0;

622,304✔

429

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

1,244,608✔

430

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

1,244,608✔

431

  switch (pReq->type) {

622,304✔

432

    case STREAM_MGMT_REQ_TRIGGER_ORIGTBL_READER: {

501,612✔

433

      if (pReq->cont.pReqs) {

501,612✔

434

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

501,612✔

435

        TAOS_CHECK_EXIT(tEncodeI32(pEncoder, num));

501,612✔

436

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

1,454,450✔

437

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

952,838✔

438

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

1,905,676✔

439

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

1,905,676✔

444

      break;

501,612✔

446

    case STREAM_MGMT_REQ_RUNNER_ORIGTBL_READER: {

120,692✔

447

      if (pReq->cont.pReqs) {

120,692✔

448

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

120,692✔

449

        TAOS_CHECK_EXIT(tEncodeI32(pEncoder, num));

120,692✔

450

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

241,384✔

451

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

120,692✔

452

          int32_t vgIdNum = taosArrayGetSize(pDeploy->vgIds);

120,692✔

453

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

241,384✔

454

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

241,384✔

455

          TAOS_CHECK_EXIT(tEncodeI32(pEncoder, vgIdNum));

120,692✔

456

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

276,184✔

457

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

310,984✔

463

      break;

120,692✔

470

_exit:

622,304✔

472

  return code;

622,304✔

475

void tFreeRunnerOReaderDeployReq(void* param) {

181,038✔

476

  if (NULL == param) {

181,038✔

480

  SStreamOReaderDeployReq* pReq = (SStreamOReaderDeployReq*)param;

181,038✔

481

  taosArrayDestroy(pReq->vgIds);

181,038✔

484

void tFreeSStreamMgmtReq(SStreamMgmtReq* pReq) {

1,244,608✔

485

  if (NULL == pReq) {

1,244,608✔

486

    return;

311,152✔

489

  switch (pReq->type) {

933,456✔

490

    case STREAM_MGMT_REQ_TRIGGER_ORIGTBL_READER:

752,418✔

491

      taosArrayDestroy(pReq->cont.pReqs);

752,418✔

492

      break;

752,418✔

493

    case STREAM_MGMT_REQ_RUNNER_ORIGTBL_READER:

181,038✔

494

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

181,038✔

495

      break;

181,038✔

502

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

311,152✔

503

  *ppDst = NULL;

311,152✔

505

  if (NULL == pSrc) {

311,152✔

509

  int32_t code = 0, lino = 0;

311,152✔

510

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

311,152✔

511

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

311,152✔

513

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

311,152✔

514

  if (pSrc->cont.pReqs) {

311,152✔

515

    switch (pSrc->type) {

311,152✔

516

      case STREAM_MGMT_REQ_TRIGGER_ORIGTBL_READER:

250,806✔

517

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

250,806✔

518

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

250,806✔

519

        break;

250,806✔

520

      case STREAM_MGMT_REQ_RUNNER_ORIGTBL_READER: {

60,346✔

521

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

60,346✔

522

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

60,346✔

523

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

60,346✔

524

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

120,692✔

525

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

60,346✔

526

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

60,346✔

527

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

60,346✔

528

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

60,346✔

529

          pNew->execId = pReq->execId;

60,346✔

530

          pNew->uid = pReq->uid;

60,346✔

532

        break;

60,346✔

541

  if (code) {

311,152✔

547

  return code;

311,152✔

551

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

311,152✔

552

  int32_t code = 0;

311,152✔

553

  int32_t lino = 0;

311,152✔

555

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

622,304✔

556

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

622,304✔

557

  switch (pReq->type) {

311,152✔

558

    case STREAM_MGMT_REQ_TRIGGER_ORIGTBL_READER: {

250,806✔

559

      int32_t num = 0;

250,806✔

560

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &num));

250,806✔

561

      if (num > 0) {

250,806✔

562

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

250,806✔

563

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

250,806✔

564

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

727,225✔

565

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

476,419✔

566

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

476,419✔

567

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

476,419✔

570

      break;

250,806✔

572

    case STREAM_MGMT_REQ_RUNNER_ORIGTBL_READER: {

60,346✔

573

      int32_t num = 0, vgIdNum = 0;

60,346✔

574

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &num));

60,346✔

575

      if (num > 0) {

60,346✔

576

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

60,346✔

577

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

60,346✔

578

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

120,692✔

579

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

60,346✔

580

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

120,692✔

581

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

120,692✔

582

          TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgIdNum));

60,346✔

583

          if (vgIdNum > 0) {

60,346✔

584

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

60,346✔

585

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

60,346✔

587

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

138,092✔

588

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

77,746✔

589

            TAOS_CHECK_EXIT(tDecodeI32(pDecoder, vgId));

77,746✔

593

      break;

60,346✔

600

_exit:

311,152✔

602

  return code;

311,152✔

605

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

353,358,290✔

606

  int32_t code = 0;

353,358,290✔

609

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

706,716,580✔

610

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

706,716,580✔

611

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

706,716,580✔

613

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

706,716,580✔

614

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

706,716,580✔

615

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

706,716,580✔

616

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

706,716,580✔

618

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

706,716,580✔

619

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

706,716,580✔

620

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

706,716,580✔

621

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

706,716,580✔

622

  if (pTask->pMgmtReq) {

353,358,290✔

623

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, 1));

622,304✔

624

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

622,304✔

626

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, 0));

352,735,986✔

629

_exit:

352,735,986✔

631

  return code;

353,358,290✔

635

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

167,990,545✔

636

  int32_t code = 0;

167,990,545✔

639

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

335,981,090✔

640

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

335,981,090✔

641

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

335,981,090✔

643

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

335,981,090✔

644

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

335,981,090✔

645

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

335,981,090✔

646

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

335,981,090✔

648

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

335,981,090✔

649

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

335,981,090✔

650

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

335,981,090✔

651

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

335,981,090✔

652

  int32_t req = 0;

167,990,545✔

653

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &req));

167,990,545✔

654

  if (req) {

167,990,545✔

655

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

311,152✔

656

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

311,152✔

657

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

311,152✔

660

_exit:

167,990,545✔

662

  return code;

167,990,545✔

694

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

26,705,396✔

695

  int32_t code = 0;

26,705,396✔

698

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

53,410,792✔

699

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

53,410,792✔

700

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

53,410,792✔

701

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

53,410,792✔

702

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

53,410,792✔

704

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

26,705,396✔

705

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, recalcNum));

26,705,396✔

706

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

26,705,396✔

711

_exit:

26,705,396✔

713

  return code;

26,705,396✔

716

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

12,511,574✔

717

  int32_t code = 0;

12,511,574✔

720

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

25,023,148✔

721

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

25,023,148✔

722

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

25,023,148✔

723

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

25,023,148✔

724

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

25,023,148✔

726

  int32_t recalcNum = 0;

12,511,574✔

727

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &recalcNum));

12,511,574✔

728

  if (recalcNum > 0) {

12,511,574✔

736

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

12,511,574✔

741

_exit:

12,511,574✔

743

  return code;

12,511,574✔

747

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

320,810,868✔

748

  int32_t code = 0;

320,810,868✔

751

  TAOS_CHECK_EXIT(tStartEncode(pEncoder));

320,810,868✔

752

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

641,621,736✔

753

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

641,621,736✔

754

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

641,621,736✔

755

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

641,621,736✔

757

  int32_t vgLeaderNum = taosArrayGetSize(pReq->pVgLeaders);

320,810,868✔

758

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, vgLeaderNum));

320,810,868✔

759

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

1,212,078,896✔

760

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

891,268,028✔

761

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *vgId));

1,782,536,056✔

764

  int32_t statusNum = taosArrayGetSize(pReq->pStreamStatus);

320,810,868✔

765

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, statusNum));

320,810,868✔

766

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

652,032,470✔

767

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

331,221,602✔

768

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

331,221,602✔

771

  int32_t reqNum = taosArrayGetSize(pReq->pStreamReq);

320,810,868✔

772

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, reqNum));

320,810,868✔

773

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

321,433,172✔

774

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

622,304✔

775

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *idx));

1,244,608✔

778

  int32_t triggerNum = taosArrayGetSize(pReq->pTriggerStatus);

320,810,868✔

779

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, triggerNum));

320,810,868✔

780

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

347,516,264✔

781

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

26,705,396✔

782

    TAOS_CHECK_EXIT(tEncodeSSTriggerRuntimeStatus(pEncoder, pTrigger));

26,705,396✔

785

  tEndEncode(pEncoder);

320,810,868✔

787

_exit:

320,810,868✔

788

  if (code) {

320,810,868✔

791

    return pEncoder->pos;

320,810,868✔

795

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

152,950,795✔

796

  int32_t code = 0;

152,950,795✔

799

  TAOS_CHECK_EXIT(tStartDecode(pDecoder));

152,950,795✔

800

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

305,901,590✔

801

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

305,901,590✔

802

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

305,901,590✔

803

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

305,901,590✔

805

  int32_t vgLearderNum = 0;

152,950,795✔

806

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgLearderNum));

152,950,795✔

807

  if (vgLearderNum > 0) {

152,950,795✔

808

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

117,612,440✔

809

    if (NULL == pReq->pVgLeaders) {

117,612,440✔

814

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

586,920,623✔

815

    int32_t vgId = 0;

433,969,828✔

816

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgId));

433,969,828✔

817

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

867,939,656✔

824

  int32_t statusNum = 0;

152,950,795✔

825

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &statusNum));

152,950,795✔

826

  if (statusNum > 0) {

152,950,795✔

827

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

11,618,599✔

828

    if (NULL == pReq->pStreamStatus) {

11,618,599✔

833

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

309,902,701✔

834

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

156,951,906✔

835

    if (NULL == pTask) {

156,951,906✔

839

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

156,951,906✔

843

  int32_t reqNum = 0;

152,950,795✔

844

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &reqNum));

152,950,795✔

845

  if (reqNum > 0) {

152,950,795✔

846

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

157,107✔

847

    if (NULL == pReq->pStreamReq) {

157,107✔

852

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

153,261,947✔

853

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

311,152✔

854

    if (NULL == pIdx) {

311,152✔

858

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, pIdx));

311,152✔

862

  int32_t triggerNum = 0;

152,950,795✔

863

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &triggerNum));

152,950,795✔

864

  if (triggerNum > 0) {

152,950,795✔

865

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

4,963,670✔

866

    if (NULL == pReq->pTriggerStatus) {

4,963,670✔

871

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

165,462,369✔

872

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

12,511,574✔

873

    if (NULL == pStatus) {

12,511,574✔

877

    TAOS_CHECK_EXIT(tDecodeSSTriggerRuntimeStatus(pDecoder, pStatus));

12,511,574✔

881

  tEndDecode(pDecoder);

152,950,795✔

883

_exit:

152,950,795✔

884

  return code;

152,950,795✔

887

void tFreeSSTriggerRuntimeStatus(void* param) {

25,864,272✔

888

  SSTriggerRuntimeStatus* pStatus = (SSTriggerRuntimeStatus*)param;

25,864,272✔

889

  if (NULL == pStatus) {

25,864,272✔

892

  taosArrayDestroy(pStatus->userRecalcs);

25,864,272✔

895

void tCleanupStreamHbMsg(SStreamHbMsg* pMsg, bool deepClean) {

928,167,910✔

896

  if (pMsg == NULL) {

928,167,910✔

900

  taosArrayDestroy(pMsg->pVgLeaders);

928,167,910✔

901

  if (deepClean) {

928,167,910✔

902

    int32_t reqNum = taosArrayGetSize(pMsg->pStreamReq);

928,167,910✔

903

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

928,790,214✔

904

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

622,304✔

905

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

622,304✔

906

      if (NULL == pTask) {

622,304✔

910

      tFreeSStreamMgmtReq(pTask->pMgmtReq);

622,304✔

911

      taosMemoryFree(pTask->pMgmtReq);

622,304✔

914

  taosArrayDestroy(pMsg->pStreamReq);

928,167,910✔

915

  taosArrayDestroy(pMsg->pStreamStatus);

928,167,910✔

916

  taosArrayDestroyEx(pMsg->pTriggerStatus, tFreeSSTriggerRuntimeStatus);

928,167,910✔

919

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

3,703,370✔

920

  int32_t code = 0;

3,703,370✔

923

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

7,406,740✔

924

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

7,406,740✔

925

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

7,406,740✔

926

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

7,406,740✔

927

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

7,406,740✔

928

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

7,406,740✔

929

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

7,406,740✔

930

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

7,406,740✔

932

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

7,406,740✔

933

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

7,406,740✔

935

_exit:

3,703,370✔

937

  return code;

3,703,370✔

940

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

2,322,978✔

941

  int32_t code = 0;

2,322,978✔

944

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

4,645,956✔

945

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

4,645,956✔

947

_exit:

2,322,978✔

949

  return code;

2,322,978✔

953

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

6,026,348✔

954

  int32_t code = 0;

6,026,348✔

957

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

12,052,696✔

958

  if (pMsg->triggerReader) {

6,026,348✔

959

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

3,703,370✔

961

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

2,322,978✔

964

_exit:

2,322,978✔

966

  return code;

6,026,348✔

969

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

11,745,324✔

970

  int32_t code = 0;

11,745,324✔

973

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

23,490,648✔

974

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

23,490,648✔

975

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

11,745,324✔

977

_exit:

11,745,324✔

979

  return code;

11,745,324✔

982

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

7,918,050✔

983

  int32_t code = 0;

7,918,050✔

986

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

7,918,050✔

987

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

15,836,100✔

989

_exit:

7,918,050✔

991

  return code;

7,918,050✔

995

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

2,646,600✔

996

  int32_t code = 0;

2,646,600✔

999

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

5,293,200✔

1000

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

5,293,200✔

1001

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

5,293,200✔

1002

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

5,293,200✔

1003

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

5,293,200✔

1004

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

5,293,200✔

1005

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

5,293,200✔

1006

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

5,293,200✔

1007

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

5,293,200✔

1008

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

2,646,600✔

1009

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

5,293,200✔

1011

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

2,646,600✔

1012

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, addrSize));

2,646,600✔

1013

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

2,893,200✔

1014

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

246,600✔

1015

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

493,200✔

1017

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

5,293,200✔

1018

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

5,293,200✔

1019

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

5,293,200✔

1021

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

5,293,200✔

1022

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

5,293,200✔

1023

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

5,293,200✔

1024

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

5,293,200✔

1026

  switch (pMsg->triggerType) {

2,646,600✔

1027

    case WINDOW_TYPE_SESSION: {

58,992✔

1029

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

117,984✔

1030

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

117,984✔

1031

      break;

58,992✔

1033

    case WINDOW_TYPE_STATE: {

1,066,570✔

1035

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

2,133,140✔

1036

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

2,133,140✔

1037

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

2,133,140✔

1038

      int32_t stateWindowZerothLen =

1,066,570✔

1039

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

1,066,570✔

1040

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

2,133,140✔

1041

      int32_t stateWindowExprLen =

1,066,570✔

1042

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

1,066,570✔

1043

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

2,133,140✔

1044

      break;

1,066,570✔

1046

    case WINDOW_TYPE_INTERVAL: {

1,097,852✔

1048

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

2,195,704✔

1049

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

2,195,704✔

1050

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

2,195,704✔

1051

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

2,195,704✔

1052

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

2,195,704✔

1053

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

2,195,704✔

1054

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

2,195,704✔

1055

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

2,195,704✔

1056

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

2,195,704✔

1057

      break;

1,097,852✔

1059

    case WINDOW_TYPE_EVENT: {

233,288✔

1061

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

233,288✔

1062

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

233,288✔

1064

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

466,576✔

1065

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

466,576✔

1067

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

466,576✔

1068

      break;

233,288✔

1070

    case WINDOW_TYPE_COUNT: {

127,300✔

1072

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

127,300✔

1073

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

254,600✔

1075

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

254,600✔

1076

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

254,600✔

1077

      break;

127,300✔

1079

    case WINDOW_TYPE_PERIOD: {

62,598✔

1081

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

125,196✔

1082

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

125,196✔

1083

      break;

62,598✔

1090

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

5,293,200✔

1091

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

5,293,200✔

1092

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

5,293,200✔

1093

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

5,293,200✔

1094

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

2,646,600✔

1095

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

5,293,200✔

1096

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

2,646,600✔

1097

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

5,293,200✔

1098

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

2,646,600✔

1099

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

5,293,200✔

1101

  int32_t readerNum = taosArrayGetSize(pMsg->readerList);

2,646,600✔

1102

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, readerNum));

2,646,600✔

1103

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

6,108,620✔

1104

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

3,462,020✔

1105

    TAOS_CHECK_EXIT(tEncodeSStreamTaskAddr(pEncoder, pAddr));

3,462,020✔

1108

  int32_t runnerNum = taosArrayGetSize(pMsg->runnerList);

2,646,600✔

1109

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, runnerNum));

2,646,600✔

1110

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

10,564,650✔

1111

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

7,918,050✔

1112

    TAOS_CHECK_EXIT(tEncodeSStreamRunnerTarget(pEncoder, pTarget));

7,918,050✔

1115

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

5,293,200✔

1116

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

5,293,200✔

1118

_exit:

2,646,600✔

1120

  return code;

2,646,600✔

1124

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

88,707,298✔

1125

  int32_t code = 0;

88,707,298✔

1128

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

177,414,596✔

1129

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

177,414,596✔

1130

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

177,414,596✔

1131

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

177,414,596✔

1132

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

177,414,596✔

1133

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

177,414,596✔

1135

_exit:

88,707,298✔

1137

  return code;

88,707,298✔

1141

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

8,562,994✔

1142

  int32_t code = 0;

8,562,994✔

1145

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

17,125,988✔

1146

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

17,125,988✔

1147

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

17,125,988✔

1148

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

17,125,988✔

1149

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

17,125,988✔

1150

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

17,125,988✔

1151

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

17,125,988✔

1152

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

17,125,988✔

1154

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

8,562,994✔

1155

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, addrSize));

8,562,994✔

1156

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

9,281,044✔

1157

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

718,050✔

1158

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

1,436,100✔

1160

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

17,125,988✔

1162

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

8,562,994✔

1163

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, outColNum));

8,562,994✔

1164

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

44,985,216✔

1165

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

36,422,222✔

1166

    TAOS_CHECK_EXIT(tSerializeSFieldWithOptions(pEncoder, pCol));

36,422,222✔

1169

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

8,562,994✔

1170

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, outTagNum));

8,562,994✔

1171

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

16,479,754✔

1172

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

7,916,760✔

1173

    TAOS_CHECK_EXIT(tSerializeSFieldWithOptions(pEncoder, pTag));

7,916,760✔

1176

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

17,125,988✔

1177

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

17,125,988✔

1179

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

17,125,988✔

1180

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

17,125,988✔

1182

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

8,562,994✔

1183

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, forceOutColsSize));

8,562,994✔

1184

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

10,141,648✔

1185

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

1,578,654✔

1186

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

1,578,654✔

1188

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

3,157,308✔

1189

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

3,157,308✔

1190

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

3,157,308✔

1191

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

3,157,308✔

1192

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

3,157,308✔

1195

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

17,125,988✔

1197

_exit:

8,562,994✔

1199

  return code;

8,562,994✔

1203

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

17,235,942✔

1204

  int32_t code = 0;

17,235,942✔

1207

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

17,235,942✔

1208

  switch (pTask->task.type) {

17,235,942✔

1209

    case STREAM_READER_TASK:

6,026,348✔

1210

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

6,026,348✔

1211

      break;

6,026,348✔

1212

    case STREAM_TRIGGER_TASK:

2,646,600✔

1213

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

2,646,600✔

1214

      break;

2,646,600✔

1215

    case STREAM_RUNNER_TASK:

8,562,994✔

1216

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

8,562,994✔

1217

      break;

8,562,994✔

1223

_exit:

17,235,942✔

1225

  return code;

17,235,942✔

1229

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

3,321,412✔

1230

  int32_t code = 0;

3,321,412✔

1233

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

6,642,824✔

1235

  int32_t readerNum = taosArrayGetSize(pStream->readerTasks);

3,321,412✔

1236

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, readerNum));

3,321,412✔

1237

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

9,347,760✔

1238

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

6,026,348✔

1239

    TAOS_CHECK_EXIT(tEncodeSStmTaskDeploy(pEncoder, pDeploy));

6,026,348✔

1242

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

6,642,824✔

1243

  if (pStream->triggerTask) {

3,321,412✔

1244

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

2,646,600✔

1247

  int32_t runnerNum = taosArrayGetSize(pStream->runnerTasks);

3,321,412✔

1248

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, runnerNum));

3,321,412✔

1249

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

11,884,406✔

1250

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

8,562,994✔

1251

    TAOS_CHECK_EXIT(tEncodeSStmTaskDeploy(pEncoder, pDeploy));

8,562,994✔

1254

_exit:

3,321,412✔

1256

  return code;

3,321,412✔

1259

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

4,900,746✔

1260

  int32_t code = 0;

4,900,746✔

1261

  int32_t lino = 0;

4,900,746✔

1263

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

9,801,492✔

1265

_exit:

4,900,746✔

1266

  return code;

4,900,746✔

1269

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

2,450,704✔

1270

  int32_t code = 0;

2,450,704✔

1273

  int32_t type = 0;

2,450,704✔

1274

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &type));

2,450,704✔

1275

  pMsg->msgType = type;

2,450,704✔

1277

_exit:

2,450,704✔

1278

  return code;

2,450,704✔

1281

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

3,110,742✔

1282

  int32_t code = 0;

3,110,742✔

1285

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

3,110,742✔

1287

_exit:

3,110,742✔

1289

  return code;

3,110,742✔

1292

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

3,110,742✔

1293

  int32_t code = 0;

3,110,742✔

1296

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

3,110,742✔

1297

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

3,110,742✔

1299

_exit:

3,110,742✔

1301

  return code;

3,110,742✔

1304

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

1,048,748✔

1305

  int32_t code = 0;

1,048,748✔

1308

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

1,048,748✔

1309

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

2,097,496✔

1310

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

2,097,496✔

1312

_exit:

1,048,748✔

1314

  return code;

1,048,748✔

1317

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

1,048,748✔

1318

  int32_t code = 0;

1,048,748✔

1321

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

1,048,748✔

1322

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

1,048,748✔

1324

_exit:

1,048,748✔

1326

  return code;

1,048,748✔

1330

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

118,952✔

1331

  int32_t code = 0;

118,952✔

1334

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

237,904✔

1335

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

237,904✔

1336

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

237,904✔

1338

_exit:

118,952✔

1340

  return code;

118,952✔

1343

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

741,256✔

1344

  int32_t code = 0;

741,256✔

1347

  switch (msgType) {

741,256✔

1348

    case STREAM_MSG_ORIGTBL_READER_INFO: {

501,612✔

1349

      int32_t vgNum = taosArrayGetSize(pRsp->vgIds);

501,612✔

1350

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, vgNum));

501,612✔

1352

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

1,454,450✔

1353

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

952,838✔

1354

        TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *vgId));

1,905,676✔

1357

      int32_t readerNum = taosArrayGetSize(pRsp->readerList);

501,612✔

1358

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, readerNum));

501,612✔

1360

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

711,374✔

1361

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

209,762✔

1362

        TAOS_CHECK_EXIT(tEncodeSStreamTaskAddr(pEncoder, addr));

209,762✔

1364

      break;

501,612✔

1376

    case STREAM_MSG_USER_RECALC: {

118,952✔

1377

      int32_t recalcNum = taosArrayGetSize(pRsp->recalcList);

118,952✔

1378

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, recalcNum));

118,952✔

1380

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

237,904✔

1381

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

118,952✔

1382

        TAOS_CHECK_EXIT(tEncodeSStreamRecalcReq(pEncoder, recalc));

118,952✔

1384

      break;

118,952✔

1386

    case STREAM_MSG_RUNNER_ORIGTBL_READER: {

120,692✔

1387

      int32_t rspNum = taosArrayGetSize(pRsp->execRspList);

120,692✔

1388

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, rspNum));

120,692✔

1390

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

241,384✔

1391

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

120,692✔

1392

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

241,384✔

1393

        int32_t vgNum = taosArrayGetSize(pDeployRsp->vgList);

120,692✔

1394

        TAOS_CHECK_EXIT(tEncodeI32(pEncoder, vgNum));

120,692✔

1395

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

276,184✔

1396

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

155,492✔

1399

      break;

120,692✔

1405

_exit:

741,256✔

1407

  return code;

741,256✔

1410

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

741,256✔

1411

  int32_t code = 0;

741,256✔

1414

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

741,256✔

1415

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

1,482,512✔

1416

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

1,482,512✔

1417

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

741,256✔

1418

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

741,256✔

1420

_exit:

741,256✔

1422

  return code;

741,256✔

1426

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

303,621,258✔

1427

  int32_t code = 0;

303,621,258✔

1430

  TAOS_CHECK_EXIT(tStartEncode(pEncoder));

303,621,258✔

1431

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

607,242,516✔

1432

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

303,621,258✔

1433

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, deployNum));

303,621,258✔

1434

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

306,942,670✔

1435

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

3,321,412✔

1436

    TAOS_CHECK_EXIT(tEncodeSStmStreamDeploy(pEncoder, pStream));

3,321,412✔

1439

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

303,621,258✔

1440

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, startNum));

303,621,258✔

1441

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

306,732,000✔

1442

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

3,110,742✔

1443

    TAOS_CHECK_EXIT(tEncodeSStreamTaskStart(pEncoder, pTask));

3,110,742✔

1446

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

607,242,516✔

1447

  if (!pRsp->undeploy.undeployAll) {

303,621,258✔

1448

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

303,621,258✔

1449

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, undeployNum));

303,621,258✔

1450

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

304,670,006✔

1451

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

1,048,748✔

1452

      TAOS_CHECK_EXIT(tEncodeSStreamTaskUndeploy(pEncoder, pTask));

1,048,748✔

1456

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

303,621,258✔

1457

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, rspNum));

303,621,258✔

1458

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

304,362,514✔

1459

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

741,256✔

1460

    TAOS_CHECK_EXIT(tEncodeSStreamMgmtRsp(pEncoder, pMgmtRsp));

741,256✔

1463

_exit:

303,621,258✔

1465

  tEndEncode(pEncoder);

303,621,258✔

1467

  return code;

303,621,258✔

1470

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

1,847,244✔

1471

  int32_t code = 0;

1,847,244✔

1474

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

3,694,488✔

1475

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

3,694,488✔

1476

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

3,694,488✔

1477

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

3,694,488✔

1478

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

3,694,488✔

1479

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

3,694,488✔

1480

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

3,694,488✔

1481

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

3,694,488✔

1482

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

3,694,488✔

1483

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

3,694,488✔

1485

_exit:

1,847,244✔

1487

  return code;

1,847,244✔

1491

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

1,157,048✔

1492

  int32_t code = 0;

1,157,048✔

1495

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

2,314,096✔

1496

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

2,314,096✔

1498

_exit:

1,157,048✔

1500

  return code;

1,157,048✔

1504

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

3,004,292✔

1505

  int32_t code = 0;

3,004,292✔

1508

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

6,008,584✔

1509

  if (pMsg->triggerReader) {

3,004,292✔

1510

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

1,847,244✔

1512

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

1,157,048✔

1515

_exit:

1,157,048✔

1517

  return code;

3,004,292✔

1521

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

7,323,406✔

1522

  int32_t code = 0;

7,323,406✔

1525

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

14,646,812✔

1526

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

14,646,812✔

1527

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

7,323,406✔

1529

_exit:

7,323,406✔

1531

  return code;

7,323,406✔

1535

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

3,953,226✔

1536

  int32_t code = 0;

3,953,226✔

1539

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

3,953,226✔

1540

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

7,906,452✔

1542

_exit:

3,953,226✔

1544

  return code;

3,953,226✔

1548

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

1,321,367✔

1549

  int32_t code = 0;

1,321,367✔

1552

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

2,642,734✔

1553

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

2,642,734✔

1554

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

2,642,734✔

1555

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

2,642,734✔

1556

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

2,642,734✔

1557

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

2,642,734✔

1558

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

2,642,734✔

1559

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

2,642,734✔

1560

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

2,642,734✔

1561

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

2,642,734✔

1563

  int32_t addrSize = 0;

1,321,367✔

1564

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &addrSize));

1,321,367✔

1565

  if (addrSize > 0) {

1,321,367✔

1566

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

123,300✔

1567

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

123,300✔

1569

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

1,444,667✔

1570

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

123,300✔

1571

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

123,300✔

1573

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

2,642,734✔

1574

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

2,642,734✔

1575

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

2,642,734✔

1577

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

2,642,734✔

1578

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

2,642,734✔

1579

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

2,642,734✔

1580

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

2,642,734✔

1582

  switch (pMsg->triggerType) {

1,321,367✔

1583

    case WINDOW_TYPE_SESSION:

29,496✔

1585

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

58,992✔

1586

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

58,992✔

1587

      break;

29,496✔

1588

    case WINDOW_TYPE_STATE:

533,285✔

1590

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

1,066,570✔

1591

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

1,066,570✔

1592

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

1,066,570✔

1593

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

1,066,570✔

1594

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

1,066,570✔

1595

      break;

533,285✔

1597

    case WINDOW_TYPE_INTERVAL:

549,231✔

1599

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

1,098,462✔

1600

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

1,098,462✔

1601

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

1,098,462✔

1602

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

1,098,462✔

1603

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

1,098,462✔

1604

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

1,098,462✔

1605

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

1,098,462✔

1606

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

1,098,462✔

1607

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

1,098,462✔

1608

      break;

549,231✔

1610

    case WINDOW_TYPE_EVENT:

116,644✔

1612

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

233,288✔

1613

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

233,288✔

1615

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

233,288✔

1616

      break;

116,644✔

1618

    case WINDOW_TYPE_COUNT:

63,650✔

1620

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

127,300✔

1622

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

127,300✔

1623

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

127,300✔

1624

      break;

63,650✔

1626

    case WINDOW_TYPE_PERIOD:

29,061✔

1628

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

58,122✔

1629

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

58,122✔

1630

      break;

29,061✔

1636

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

2,642,734✔

1637

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

2,642,734✔

1638

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

2,642,734✔

1639

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

2,642,734✔

1640

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

2,642,734✔

1641

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

2,642,734✔

1642

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

2,642,734✔

1644

  int32_t readerNum = 0;

1,321,367✔

1645

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &readerNum));

1,321,367✔

1646

  if (readerNum > 0) {

1,321,367✔

1647

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

1,317,700✔

1648

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

1,317,700✔

1650

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

3,048,480✔

1651

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

1,727,113✔

1652

    TAOS_CHECK_EXIT(tDecodeSStreamTaskAddr(pDecoder, pAddr));

1,727,113✔

1655

  int32_t runnerNum = 0;

1,321,367✔

1656

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &runnerNum));

1,321,367✔

1657

  if (runnerNum > 0) {

1,321,367✔

1658

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

1,317,742✔

1659

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

1,317,742✔

1661

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

5,274,593✔

1662

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

3,953,226✔

1663

    TAOS_CHECK_EXIT(tDecodeSStreamRunnerTarget(pDecoder, pTarget));

3,953,226✔

1666

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

2,642,734✔

1667

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

2,642,734✔

1669

_exit:

1,321,367✔

1671

  return code;

1,321,367✔

1676

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

34,705,464✔

1677

  int32_t code = 0;

34,705,464✔

1680

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

34,705,464✔

1681

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

69,410,928✔

1682

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

69,410,928✔

1683

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

69,408,714✔

1684

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

69,406,500✔

1685

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

69,408,714✔

1687

_exit:

34,705,464✔

1689

  return code;

34,705,464✔

1692

void destroySStreamOutCols(void* p){

789,327✔

1693

  if (p == NULL) return;

789,327✔

1694

  SStreamOutCol* col = (SStreamOutCol*)p;

789,327✔

1695

  taosMemoryFreeClear(col->expr);

789,327✔

1698

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

4,262,276✔

1699

  int32_t code = 0;

4,262,276✔

1702

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

8,524,552✔

1703

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

8,524,552✔

1704

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

8,524,552✔

1705

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

8,524,552✔

1706

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

8,524,552✔

1707

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

8,524,552✔

1708

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

8,524,552✔

1709

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

8,524,552✔

1711

  int32_t addrSize = 0;

4,262,276✔

1712

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &addrSize));

4,262,276✔

1713

  if (addrSize > 0) {

4,262,276✔

1714

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

359,025✔

1715

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

359,025✔

1717

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

4,621,301✔

1718

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

359,025✔

1719

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

359,025✔

1721

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

8,524,552✔

1723

  int32_t outColNum = 0;

4,262,276✔

1724

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &outColNum));

4,262,276✔

1725

  if (outColNum > 0) {

4,262,276✔

1726

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

4,262,276✔

1727

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

4,262,276✔

1729

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

22,213,388✔

1730

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

17,951,112✔

1731

    TAOS_CHECK_EXIT(tDeserializeSFieldWithOptions(pDecoder, pCol));

17,951,112✔

1734

  int32_t outTagNum = 0;

4,262,276✔

1735

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &outTagNum));

4,262,276✔

1736

  if (outTagNum > 0) {

4,262,276✔

1737

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

2,539,199✔

1738

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

2,539,199✔

1740

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

8,221,483✔

1741

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

3,959,207✔

1742

    TAOS_CHECK_EXIT(tDeserializeSFieldWithOptions(pDecoder, pTag));

3,959,207✔

1745

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

8,524,552✔

1746

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

8,524,552✔

1748

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

8,524,552✔

1749

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

8,524,552✔

1751

  int32_t forceOutColsSize = 0;

4,262,276✔

1752

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &forceOutColsSize));

4,262,276✔

1753

  if (forceOutColsSize > 0) {

4,262,276✔

1754

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

120,516✔

1755

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

120,516✔

1757

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

5,051,603✔

1758

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

789,327✔

1760

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

1,578,654✔

1761

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

1,578,654✔

1762

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

1,578,654✔

1763

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

1,578,654✔

1764

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

1,578,654✔

1767

  if (!tDecodeIsEnd(pDecoder)) {

4,262,276✔

1768

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

8,524,552✔

1771

_exit:

4,262,276✔

1773

  return code;

4,262,276✔

1776

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

8,587,935✔

1777

  int32_t code = 0;

8,587,935✔

1780

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

8,587,935✔

1781

  switch (pTask->task.type) {

8,587,935✔

1782

    case STREAM_READER_TASK:

3,004,292✔

1783

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

3,004,292✔

1784

      break;

3,004,292✔

1785

    case STREAM_TRIGGER_TASK:

1,321,367✔

1786

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

1,321,367✔

1787

      break;

1,321,367✔

1788

    case STREAM_RUNNER_TASK:

4,262,276✔

1789

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

4,262,276✔

1790

      break;

4,262,276✔

1796

_exit:

8,587,935✔

1798

  return code;

8,587,935✔

1802

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

1,658,861✔

1803

  int32_t code = 0;

1,658,861✔

1806

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

3,317,722✔

1808

  int32_t readerNum = 0;

1,658,861✔

1809

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &readerNum));

1,658,861✔

1810

  if (readerNum > 0) {

1,658,861✔

1811

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

1,478,127✔

1812

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

1,478,127✔

1814

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

4,663,153✔

1815

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

3,004,292✔

1816

    TAOS_CHECK_EXIT(tDecodeSStmTaskDeploy(pDecoder, pTask));

3,004,292✔

1819

  int32_t triggerTask = 0;

1,658,861✔

1820

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &triggerTask));

1,658,861✔

1821

  if (triggerTask) {

1,658,861✔

1822

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

1,321,367✔

1823

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

1,321,367✔

1824

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

1,321,367✔

1827

  int32_t runnerNum = 0;

1,658,861✔

1828

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &runnerNum));

1,658,861✔

1829

  if (runnerNum > 0) {

1,658,861✔

1830

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

1,425,466✔

1831

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

1,425,466✔

1833

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

5,921,137✔

1834

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

4,262,276✔

1835

    TAOS_CHECK_EXIT(tDecodeSStmTaskDeploy(pDecoder, pTask));

4,262,276✔

1838

_exit:

1,658,861✔

1840

  return code;

1,658,861✔

1844

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

1,555,676✔

1845

  int32_t code = 0;

1,555,676✔

1848

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

1,555,676✔

1850

_exit:

1,555,676✔

1852

  return code;

1,555,676✔

1856

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

1,555,676✔

1857

  int32_t code = 0;

1,555,676✔

1860

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

1,555,676✔

1861

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

1,555,676✔

1863

_exit:

1,555,676✔

1865

  return code;

1,555,676✔

1869

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

524,400✔

1870

  int32_t code = 0;

524,400✔

1873

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

524,400✔

1874

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

1,048,800✔

1875

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

1,048,800✔

1877

_exit:

524,400✔

1879

  return code;

524,400✔

1883

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

524,400✔

1884

  int32_t code = 0;

524,400✔

1887

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

524,400✔

1888

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

524,400✔

1890

_exit:

524,400✔

1892

  return code;

524,400✔

1895

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

59,476✔

1896

  int32_t code = 0;

59,476✔

1899

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

118,952✔

1900

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

118,952✔

1901

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

118,952✔

1903

_exit:

59,476✔

1905

  return code;

59,476✔

1908

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

370,628✔

1909

  int32_t code = 0;

370,628✔

1912

  switch (msgType) {

370,628✔

1913

    case STREAM_MSG_ORIGTBL_READER_INFO: {

250,806✔

1914

      int32_t vgNum = 0;

250,806✔

1915

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgNum));

250,806✔

1916

      if (vgNum > 0) {

250,806✔

1917

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

250,806✔

1918

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

250,806✔

1920

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

727,225✔

1921

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

476,419✔

1922

        TAOS_CHECK_EXIT(tDecodeI32(pDecoder, vgId));

476,419✔

1925

      int32_t readerNum = 0;

250,806✔

1926

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &readerNum));

250,806✔

1927

      if (readerNum > 0) {

250,806✔

1928

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

101,262✔

1929

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

101,262✔

1931

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

355,687✔

1932

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

104,881✔

1933

        TAOS_CHECK_EXIT(tDecodeSStreamTaskAddr(pDecoder, addr));

104,881✔

1935

      break;

250,806✔

1950

    case STREAM_MSG_USER_RECALC: {

59,476✔

1951

      int32_t recalcNum = 0;

59,476✔

1952

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &recalcNum));

59,476✔

1953

      if (recalcNum > 0) {

59,476✔

1954

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

59,476✔

1955

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

59,476✔

1957

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

118,952✔

1958

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

59,476✔

1959

        TAOS_CHECK_EXIT(tDecodeSStreamRecalcReq(pDecoder, recalc));

59,476✔

1961

      break;

59,476✔

1963

    case STREAM_MSG_RUNNER_ORIGTBL_READER: {

60,346✔

1964

      int32_t rspNum = 0, vgNum = 0;

60,346✔

1965

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &rspNum));

60,346✔

1966

      if (rspNum > 0) {

60,346✔

1967

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

60,346✔

1968

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

60,346✔

1970

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

120,692✔

1971

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

60,346✔

1972

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

120,692✔

1973

        TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgNum));

60,346✔

1974

        if (vgNum > 0) {

60,346✔

1975

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

60,346✔

1976

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

60,346✔

1978

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

138,092✔

1979

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

77,746✔

1980

          TAOS_CHECK_EXIT(tDecodeSStreamTaskAddr(pDecoder, pAddr));

77,746✔

1983

      break;

60,346✔

1989

_exit:

370,628✔

1991

  return code;

370,628✔

1995

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

370,628✔

1996

  int32_t code = 0;

370,628✔

1999

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

370,628✔

2000

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

741,256✔

2001

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

741,256✔

2002

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

370,628✔

2003

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

370,628✔

2005

_exit:

370,628✔

2007

  return code;

370,628✔

2010

void tFreeSStreamOReaderDeployRsp(void* param) {

120,692✔

2011

  if (NULL == param) {

120,692✔

2015

  SStreamOReaderDeployRsp* pRsp = (SStreamOReaderDeployRsp*)param;

120,692✔

2016

  taosArrayDestroy(pRsp->vgList);

120,692✔

2019

void tFreeSStreamMgmtRsp(void* param) {

741,256✔

2020

  if (NULL == param) {

741,256✔

2024

  SStreamMgmtRsp* pRsp = (SStreamMgmtRsp*)param;

741,256✔

2026

  taosArrayDestroy(pRsp->cont.vgIds);

741,256✔

2027

  taosArrayDestroy(pRsp->cont.readerList);

741,256✔

2028

  taosArrayDestroy(pRsp->cont.runnerList);

741,256✔

2029

  taosArrayDestroy(pRsp->cont.recalcList);

741,256✔

2030

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

741,256✔

2033

void tFreeSStreamReaderDeployMsg(SStreamReaderDeployMsg* pReader) {

3,004,292✔

2034

  if (NULL == pReader) {

3,004,292✔

2038

  if (pReader->triggerReader) {

3,004,292✔

2039

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

1,847,244✔

2040

    taosMemoryFree(pMsg->triggerTblName);

1,847,244✔

2041

    taosMemoryFree(pMsg->partitionCols);

1,847,244✔

2042

    taosMemoryFree(pMsg->triggerCols);

1,847,244✔

2043

    taosMemoryFree(pMsg->triggerScanPlan);

1,847,244✔

2044

    taosMemoryFree(pMsg->calcCacheScanPlan);

1,847,244✔

2046

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

1,157,048✔

2047

    taosMemoryFree(pMsg->calcScanPlan);

1,157,048✔

2059

void tFreeSStreamTriggerDeployMsg(SStreamTriggerDeployMsg* pTrigger) {

1,321,367✔

2060

  if (NULL == pTrigger) {

1,321,367✔

2064

  taosArrayDestroyEx(pTrigger->pNotifyAddrUrls, tFreeStreamNotifyUrl);

1,321,367✔

2065

  switch (pTrigger->triggerType) {

1,321,367✔

2066

    case WINDOW_TYPE_STATE:

533,285✔

2067

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

533,285✔

2068

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

533,285✔

2069

      break;

533,285✔

2070

    case WINDOW_TYPE_EVENT:

116,644✔

2071

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

116,644✔

2072

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

116,644✔

2073

      break;

116,644✔

2074

    case WINDOW_TYPE_COUNT:

63,650✔

2075

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

63,650✔

2076

      break;

63,650✔

2077

    default:

607,788✔

2078

      break;

607,788✔

2081

  taosMemoryFree(pTrigger->partitionCols);

1,321,367✔

2082

  taosMemoryFree(pTrigger->triggerPrevFilter);

1,321,367✔

2083

  taosMemoryFree(pTrigger->triggerScanPlan);

1,321,367✔

2084

  taosMemoryFree(pTrigger->calcCacheScanPlan);

1,321,367✔

2086

  taosArrayDestroy(pTrigger->readerList);

1,321,367✔

2087

  taosArrayDestroy(pTrigger->runnerList);

1,321,367✔

2088

  taosMemoryFree(pTrigger->streamName);

1,321,367✔

2100

void tFreeSStreamRunnerDeployMsg(SStreamRunnerDeployMsg* pRunner) {

4,262,276✔

2101

  if (NULL == pRunner) {

4,262,276✔

2105

  taosMemoryFree(pRunner->streamName);

4,262,276✔

2106

  taosMemoryFree(pRunner->pPlan);

4,262,276✔

2107

  taosMemoryFree(pRunner->outDBFName);

4,262,276✔

2108

  taosMemoryFree(pRunner->outTblName);

4,262,276✔

2110

  taosArrayDestroyEx(pRunner->pNotifyAddrUrls, tFreeStreamNotifyUrl);

4,262,276✔

2111

  taosArrayDestroy(pRunner->outCols);

4,262,276✔

2112

  taosArrayDestroy(pRunner->outTags);

4,262,276✔

2114

  taosMemoryFree(pRunner->subTblNameExpr);

4,262,276✔

2115

  taosMemoryFree(pRunner->tagValueExpr);

4,262,276✔

2116

  taosArrayDestroyEx(pRunner->forceOutCols, tFreeSStreamOutCol);

4,262,276✔

2119

void tFreeSStmTaskDeploy(void* param) {

10,586,135✔

2120

  if (NULL == param) {

10,586,135✔

2121

    return;

1,998,200✔

2124

  SStmTaskDeploy* pTask = (SStmTaskDeploy*)param;

8,587,935✔

2125

  switch (pTask->task.type)  {

8,587,935✔

2126

    case STREAM_READER_TASK:

3,004,292✔

2127

      tFreeSStreamReaderDeployMsg(&pTask->msg.reader);

3,004,292✔

2128

      break;

3,004,292✔

2129

    case STREAM_TRIGGER_TASK:

1,321,367✔

2130

      tFreeSStreamTriggerDeployMsg(&pTask->msg.trigger);

1,321,367✔

2131

      break;

1,321,367✔

2132

    case STREAM_RUNNER_TASK:

4,262,276✔

2133

      tFreeSStreamRunnerDeployMsg(&pTask->msg.runner);

4,262,276✔

2134

      break;

4,262,276✔

2140

void tFreeSStmStreamDeploy(void* param) {

1,660,706✔

2141

  if (NULL == param) {

1,660,706✔

2145

  SStmStreamDeploy* pDeploy = (SStmStreamDeploy*)param;

1,660,706✔

2146

  taosArrayDestroy(pDeploy->readerTasks);

1,660,706✔

2147

  if (pDeploy->triggerTask) {

1,660,706✔

2148

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

1,323,300✔

2149

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

1,323,300✔

2150

    taosMemoryFree(pDeploy->triggerTask);

1,323,300✔

2153

  int32_t runnerNum = taosArrayGetSize(pDeploy->runnerTasks);

1,660,706✔

2154

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

5,942,203✔

2155

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

4,281,497✔

2156

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

4,281,497✔

2158

  taosArrayDestroy(pDeploy->runnerTasks);

1,660,706✔

2161

void tDeepFreeSStmStreamDeploy(void* param) {

3,319,567✔

2162

  if (NULL == param) {

3,319,567✔

2166

  SStmStreamDeploy* pDeploy = (SStmStreamDeploy*)param;

3,319,567✔

2167

  taosArrayDestroyEx(pDeploy->readerTasks, tFreeSStmTaskDeploy);

3,319,567✔

2168

  tFreeSStmTaskDeploy(pDeploy->triggerTask);

3,319,567✔

2169

  taosMemoryFree(pDeploy->triggerTask);

3,319,567✔

2170

  taosArrayDestroyEx(pDeploy->runnerTasks, tFreeSStmTaskDeploy);

3,319,567✔

2174

void tFreeSMStreamHbRspMsg(SMStreamHbRspMsg* pRsp) {

305,901,590✔

2175

  if (NULL == pRsp) {

305,901,590✔

2178

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

305,901,590✔

2179

  taosArrayDestroy(pRsp->start.taskList);

305,901,590✔

2180

  taosArrayDestroy(pRsp->undeploy.taskList);

305,901,590✔

2181

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

305,901,590✔

2184

void tDeepFreeSMStreamHbRspMsg(SMStreamHbRspMsg* pRsp) {

151,786,934✔

2185

  if (NULL == pRsp) {

151,786,934✔

2188

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

151,786,934✔

2189

  taosArrayDestroy(pRsp->start.taskList);

151,786,934✔

2190

  taosArrayDestroy(pRsp->undeploy.taskList);

151,786,934✔

2191

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

151,786,934✔

2196

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

151,786,934✔

2197

  int32_t code = 0;

151,786,934✔

2200

  TAOS_CHECK_EXIT(tStartDecode(pDecoder));

151,786,934✔

2201

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

303,573,868✔

2202

  int32_t deployNum = 0;

151,786,934✔

2203

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &deployNum));

151,786,934✔

2204

  if (deployNum > 0) {

151,786,934✔

2205

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

524,328✔

2206

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

524,328✔

2208

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

153,445,795✔

2209

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

1,658,861✔

2210

    TAOS_CHECK_EXIT(tDecodeSStmStreamDeploy(pDecoder, pStream));

1,658,861✔

2213

  int32_t startNum = 0;

151,786,934✔

2214

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &startNum));

151,786,934✔

2215

  if (startNum > 0) {

151,786,934✔

2216

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

808,608✔

2217

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

808,608✔

2219

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

153,342,610✔

2220

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

1,555,676✔

2221

    TAOS_CHECK_EXIT(tDecodeSStreamTaskStart(pDecoder, pTask));

1,555,676✔

2224

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

303,573,868✔

2225

  if (!pRsp->undeploy.undeployAll) {

151,786,934✔

2226

    int32_t undeployNum = 0;

151,786,934✔

2227

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &undeployNum));

151,786,934✔

2228

    if (undeployNum > 0) {

151,786,934✔

2229

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

168,891✔

2230

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

168,891✔

2232

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

152,311,334✔

2233

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

524,400✔

2234

      TAOS_CHECK_EXIT(tDecodeSStreamTaskUndeploy(pDecoder, pTask));

524,400✔

2238

  int32_t rspNum = 0;

151,786,934✔

2239

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &rspNum));

151,786,934✔

2240

  if (rspNum > 0) {

151,786,934✔

2241

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

211,541✔

2242

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

211,541✔

2243

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

582,169✔

2244

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

370,628✔

2245

      TAOS_CHECK_EXIT(tDecodeSStreamMgmtRsp(pDecoder, pMgmtRsp));

370,628✔

2249

  tEndDecode(pDecoder);

151,786,934✔

2251

_exit:

151,786,934✔

2252

  return code;

151,786,934✔

2309

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

10,099,674✔

2310

  int32_t code = 0;

10,099,674✔

2314

  int32_t sqlLen = pReq->sql == NULL ? 0 : (int32_t)strlen(pReq->sql) + 1;

10,099,674✔

2315

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

10,099,674✔

2316

  int32_t outDbLen = pReq->outDB == NULL ? 0 : (int32_t)strlen(pReq->outDB) + 1;

10,099,674✔

2317

  int32_t streamDBLen = pReq->streamDB == NULL ? 0 : (int32_t)strlen(pReq->streamDB) + 1;

10,099,674✔

2318

  int32_t triggerDBLen = pReq->triggerDB == NULL ? 0 : (int32_t)strlen(pReq->triggerDB) + 1;

10,099,674✔

2319

  int32_t triggerTblNameLen = pReq->triggerTblName == NULL ? 0 : (int32_t)strlen(pReq->triggerTblName) + 1;

10,099,674✔

2320

  int32_t outTblNameLen = pReq->outTblName == NULL ? 0 : (int32_t)strlen(pReq->outTblName) + 1;

10,099,674✔

2322

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

20,199,348✔

2324

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pReq->name, nameLen));

20,199,348✔

2325

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pReq->sql, sqlLen));

20,199,348✔

2326

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pReq->outDB, outDbLen));

20,199,348✔

2327

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pReq->streamDB, streamDBLen));

20,199,348✔

2328

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pReq->triggerDB, triggerDBLen));

20,199,348✔

2329

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pReq->triggerTblName, triggerTblNameLen));

20,199,348✔

2330

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pReq->outTblName, outTblNameLen));

20,199,348✔

2332

  int32_t calcDbSize = (int32_t)taosArrayGetSize(pReq->calcDB);

10,099,674✔

2333

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, calcDbSize));

10,099,674✔

2334

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

20,143,868✔

2335

    const char *dbName = taosArrayGetP(pReq->calcDB, i);

10,044,194✔

2336

    TAOS_CHECK_EXIT((tEncodeCStr(pEncoder, dbName)));

10,044,194✔

2340

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

20,199,348✔

2341

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

20,199,348✔

2342

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

20,199,348✔

2343

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

20,199,348✔

2344

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

20,199,348✔

2345

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

20,199,348✔

2346

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

20,199,348✔

2347

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

20,199,348✔

2348

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

20,199,348✔

2349

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

20,199,348✔

2352

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

10,099,674✔

2353

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, addrSize));

10,099,674✔

2354

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

11,663,536✔

2355

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

1,563,862✔

2356

    TAOS_CHECK_EXIT((tEncodeCStr(pEncoder, url)));

1,563,862✔

2358

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

20,199,348✔

2359

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

20,199,348✔

2360

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

20,199,348✔

2365

  int32_t filterColsLen = pReq->triggerFilterCols == NULL ? 0 : (int32_t)strlen((char*)pReq->triggerFilterCols) + 1;

10,099,674✔

2366

  int32_t triggerColsLen = pReq->triggerCols == NULL ? 0 : (int32_t)strlen((char*)pReq->triggerCols) + 1;

10,099,674✔

2367

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

10,099,674✔

2368

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pReq->triggerFilterCols, filterColsLen));

20,199,348✔

2369

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pReq->triggerCols, triggerColsLen));

20,199,348✔

2370

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

20,199,348✔

2373

  int32_t outColSize = (int32_t )taosArrayGetSize(pReq->outCols);

10,099,674✔

2374

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, outColSize));

10,099,674✔

2375

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

54,467,990✔

2376

    SFieldWithOptions *pField = taosArrayGet(pReq->outCols, i);

44,368,316✔

2377

    TAOS_CHECK_EXIT(tSerializeSFieldWithOptions(pEncoder, pField));

44,368,316✔

2381

  int32_t outTagSize = (int32_t )taosArrayGetSize(pReq->outTags);

10,099,674✔

2382

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, outTagSize));

10,099,674✔

2383

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

18,683,944✔

2384

    SField *pField = taosArrayGet(pReq->outTags, i);

8,584,270✔

2385

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

17,168,540✔

2386

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

17,168,540✔

2387

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

17,168,540✔

2388

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

17,168,540✔

2391

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

20,199,348✔

2392

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

20,199,348✔

2393

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

20,199,348✔

2394

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

20,199,348✔

2396

  switch (pReq->triggerType) {

10,099,674✔

2397

    case WINDOW_TYPE_SESSION: {

387,760✔

2399

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

775,520✔

2400

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

775,520✔

2401

      break;

387,760✔

2403

    case WINDOW_TYPE_STATE: {

3,426,610✔

2405

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

6,853,220✔

2406

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

6,853,220✔

2407

      break;

3,426,610✔

2409

    case WINDOW_TYPE_INTERVAL: {

4,329,522✔

2411

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

8,659,044✔

2412

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

8,659,044✔

2413

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

8,659,044✔

2414

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

8,659,044✔

2415

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

8,659,044✔

2416

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

8,659,044✔

2417

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

8,659,044✔

2418

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

8,659,044✔

2419

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

8,659,044✔

2420

      break;

4,329,522✔

2422

    case WINDOW_TYPE_EVENT: {

1,017,392✔

2424

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

1,017,392✔

2425

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

1,017,392✔

2427

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

2,034,784✔

2428

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

2,034,784✔

2430

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

2,034,784✔

2431

      break;

1,017,392✔

2433

    case WINDOW_TYPE_COUNT: {

614,962✔

2435

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

614,962✔

2436

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

1,229,924✔

2438

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

1,229,924✔

2439

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

1,229,924✔

2440

      break;

614,962✔

2442

    case WINDOW_TYPE_PERIOD: {

323,428✔

2444

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

646,856✔

2445

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

646,856✔

2446

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

646,856✔

2447

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

646,856✔

2448

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

646,856✔

2449

      break;

323,428✔

2453

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

20,199,348✔

2454

  TAOS_CHECK_EXIT(tEncodeU64(pEncoder, pReq->triggerTblUid));

20,199,348✔

2455

  TAOS_CHECK_EXIT(tEncodeU64(pEncoder, pReq->triggerTblSuid));

20,199,348✔

2456

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

20,199,348✔

2457

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

20,199,348✔

2458

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

20,199,348✔

2459

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

20,199,348✔

2460

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

20,199,348✔

2461

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

20,199,348✔

2462

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

20,199,348✔

2463

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

20,199,348✔

2464

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

20,199,348✔

2465

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

20,199,348✔

2466

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

20,199,348✔

2468

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

20,199,348✔

2469

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

20,199,348✔

2471

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

10,099,674✔

2472

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

20,199,348✔

2474

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

20,199,348✔

2475

  int32_t triggerFilterLen = pReq->triggerPrevFilter == NULL ? 0 : (int32_t)strlen((char*)pReq->triggerPrevFilter) + 1;

10,099,674✔

2476

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pReq->triggerPrevFilter, triggerFilterLen));

20,199,348✔

2478

  int32_t calcScanPlanListSize = (int32_t)taosArrayGetSize(pReq->calcScanPlanList);

10,099,674✔

2479

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, calcScanPlanListSize));

10,099,674✔

2480

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

25,773,146✔

2481

    SStreamCalcScan* pCalcScanPlan = (SStreamCalcScan*)taosArrayGet(pReq->calcScanPlanList, i);

15,673,472✔

2482

    int32_t          vgListSize = (int32_t)taosArrayGetSize(pCalcScanPlan->vgList);

15,673,472✔

2483

    int32_t          scanPlanLen = pCalcScanPlan->scanPlan == NULL ? 0 : (int32_t)strlen((char*)pCalcScanPlan->scanPlan) + 1;

15,673,472✔

2484

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, vgListSize));

15,673,472✔

2485

    for (int32_t j = 0; j < vgListSize; ++j) {

31,346,944✔

2486

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *(int32_t*)taosArrayGet(pCalcScanPlan->vgList, j)));

31,346,944✔

2488

    TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pCalcScanPlan->readFromCache));

31,346,944✔

2489

    TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pCalcScanPlan->scanPlan, scanPlanLen));

31,346,944✔

2492

  int32_t calcPlanLen = pReq->calcPlan == NULL ? 0 : (int32_t)strlen((char*)pReq->calcPlan) + 1;

10,099,674✔

2493

  int32_t subTblNameExprLen = pReq->subTblNameExpr == NULL ? 0 : (int32_t)strlen((char*)pReq->subTblNameExpr) + 1;

10,099,674✔

2494

  int32_t tagValueExprLen = pReq->tagValueExpr == NULL ? 0 : (int32_t)strlen((char*)pReq->tagValueExpr) + 1;

10,099,674✔

2496

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

20,199,348✔

2497

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pReq->calcPlan, calcPlanLen));

20,199,348✔

2498

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pReq->subTblNameExpr, subTblNameExprLen));

20,199,348✔

2499

  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pReq->tagValueExpr, tagValueExprLen));

20,199,348✔

2501

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

10,099,674✔

2502

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, forceOutColsSize));

10,099,674✔

2503

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

11,722,768✔

2504

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

1,623,094✔

2505

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

1,623,094✔

2507

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

3,246,188✔

2508

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

3,246,188✔

2509

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

3,246,188✔

2510

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

3,246,188✔

2511

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

3,246,188✔

2514

  switch (pReq->triggerType) {

10,099,674✔

2515

    case WINDOW_TYPE_STATE: {

3,426,610✔

2517

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

3,426,610✔

2518

      TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pReq->trigger.stateWin.expr, stateExprLen));

6,853,220✔

2519

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

6,853,220✔

2520

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

3,426,610✔

2521

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

6,853,220✔

2522

      break;

3,426,610✔

2524

    case WINDOW_TYPE_INTERVAL: {

4,329,522✔

2525

      TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pReq->trigger.sliding.overlap));

8,659,044✔

2526

      break;

4,329,522✔

2528

    default: {

2,343,542✔

2529

      break;

2,343,542✔

2533

_exit:

10,099,674✔

2535

  if (code) {

10,099,674✔

2539

  return 0;

10,099,674✔

2542

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

4,752,576✔

2543

  SEncoder encoder = {0};

4,752,576✔

2544

  tEncoderInit(&encoder, buf, bufLen);

4,752,576✔

2545

  int32_t code = 0;

4,752,576✔

2548

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

4,752,576✔

2550

  TAOS_CHECK_EXIT(tSerializeSCMCreateStreamReqImpl(&encoder, pReq));

4,752,576✔

2552

  tEndEncode(&encoder);

4,752,576✔

2554

_exit:

4,752,576✔

2555

  if (code) {

4,752,576✔

2559

    int32_t tlen = encoder.pos;

4,752,576✔

2560

    tEncoderClear(&encoder);

4,752,576✔

2561

    return tlen;

4,752,576✔

2567

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

3,040,744✔

2568

  int32_t code = 0;

3,040,744✔

2571

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

6,081,488✔

2573

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

6,081,488✔

2574

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pReq->sql, NULL));

6,081,488✔

2575

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pReq->outDB, NULL));

6,081,488✔

2576

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pReq->streamDB, NULL));

6,081,488✔

2577

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pReq->triggerDB, NULL));

6,081,488✔

2578

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

6,081,488✔

2579

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

6,081,488✔

2581

  int32_t calcDbSize = 0;

3,040,744✔

2582

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &calcDbSize));

3,040,744✔

2583

  pReq->calcDB = taosArrayInit(calcDbSize, POINTER_BYTES);

3,040,744✔

2584

  if (pReq->calcDB == NULL) {

3,040,744✔

2587

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

6,074,238✔

2588

    char *calcDb = NULL;

3,033,494✔

2589

    TAOS_CHECK_EXIT(tDecodeCStr(pDecoder, &calcDb));

3,033,494✔

2590

    calcDb = taosStrndup(calcDb, TSDB_DB_FNAME_LEN);

3,033,494✔

2591

    if (calcDb == NULL) {

3,033,494✔

2594

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

6,066,988✔

2600

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

6,081,488✔

2601

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

6,081,488✔

2602

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

6,081,488✔

2603

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

6,081,488✔

2604

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

6,081,488✔

2605

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

6,081,488✔

2606

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

6,081,488✔

2607

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

6,081,488✔

2608

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

6,081,488✔

2609

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

6,081,488✔

2611

  int32_t addrSize = 0;

3,040,744✔

2612

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &addrSize));

3,040,744✔

2613

  if (addrSize > 0) {

3,040,744✔

2614

    pReq->pNotifyAddrUrls = taosArrayInit(addrSize, POINTER_BYTES);

246,600✔

2615

    if (pReq->pNotifyAddrUrls == NULL) {

246,600✔

2619

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

3,287,344✔

2620

    char *url = NULL;

246,600✔

2621

    TAOS_CHECK_EXIT(tDecodeCStr(pDecoder, &url));

246,600✔

2622

    url = taosStrndup(url, TSDB_STREAM_NOTIFY_URL_LEN);

246,600✔

2623

    if (url == NULL) {

246,600✔

2626

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

493,200✔

2631

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

6,081,488✔

2632

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

6,081,488✔

2633

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

6,081,488✔

2635

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pReq->triggerFilterCols, NULL));

6,081,488✔

2636

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

6,081,488✔

2637

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

6,081,488✔

2639

  int32_t outColSize = 0;

3,040,744✔

2640

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &outColSize));

3,040,744✔

2641

  if (outColSize > 0) {

3,040,744✔

2642

    pReq->outCols = taosArrayInit_s(sizeof(SFieldWithOptions), outColSize);

3,033,494✔

2643

    if (pReq->outCols == NULL) {

3,033,494✔

2647

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

15,828,639✔

2648

      SFieldWithOptions* pField = taosArrayGet(pReq->outCols, i);

12,795,145✔

2649

      TAOS_CHECK_EXIT(tDeserializeSFieldWithOptions(pDecoder, pField));

12,795,145✔

2653

  int32_t outTagSize = 0;

3,040,744✔

2654

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &outTagSize));

3,040,744✔

2655

  if (outTagSize > 0) {

3,040,744✔

2656

    pReq->outTags = taosArrayInit(outTagSize, sizeof(SFieldWithOptions));

1,806,622✔

2657

    if (pReq->outTags == NULL) {

1,806,622✔

2661

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

4,519,416✔

2662

      SFieldWithOptions field = {0};

2,712,794✔

2663

      TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &field.type));

2,712,794✔

2664

      TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &field.flags));

2,712,794✔

2665

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &field.bytes));

2,712,794✔

2666

      TAOS_CHECK_EXIT(tDecodeCStrTo(pDecoder, field.name));

2,712,794✔

2667

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

5,425,588✔

2673

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

6,081,488✔

2674

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

6,081,488✔

2675

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

6,081,488✔

2676

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

6,081,488✔

2678

  switch (pReq->triggerType) {

3,040,744✔

2679

    case WINDOW_TYPE_SESSION: {

58,452✔

2681

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

116,904✔

2682

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

116,904✔

2683

      break;

58,452✔

2685

      case WINDOW_TYPE_STATE: {

1,096,518✔

2687

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

2,193,036✔

2688

        pReq->trigger.stateWin.extend = 0;

1,096,518✔

2689

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

2,193,036✔

2690

        break;

1,096,518✔

2692

      case WINDOW_TYPE_INTERVAL: {

1,452,376✔

2694

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

2,904,752✔

2695

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

2,904,752✔

2696

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

2,904,752✔

2697

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

2,904,752✔

2698

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

2,904,752✔

2699

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

2,904,752✔

2700

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

2,904,752✔

2701

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

2,904,752✔

2702

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

2,904,752✔

2703

        break;

1,452,376✔

2705

      case WINDOW_TYPE_EVENT: {

233,288✔

2707

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

466,576✔

2708

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

466,576✔

2709

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

466,576✔

2710

        break;

233,288✔

2712

      case WINDOW_TYPE_COUNT: {

130,909✔

2713

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

261,818✔

2715

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

261,818✔

2716

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

261,818✔

2717

        break;

130,909✔

2719

      case WINDOW_TYPE_PERIOD: {

69,201✔

2721

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

138,402✔

2722

        TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pReq->trigger.period.periodUnit));

138,402✔

2723

        TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pReq->trigger.period.offsetUnit));

138,402✔

2724

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

138,402✔

2725

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

138,402✔

2726

        break;

69,201✔

2732

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

6,081,488✔

2733

  TAOS_CHECK_EXIT(tDecodeU64(pDecoder, &pReq->triggerTblUid));

6,081,488✔

2734

  TAOS_CHECK_EXIT(tDecodeU64(pDecoder, &pReq->triggerTblSuid));

6,081,488✔

2735

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

6,081,488✔

2736

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

6,081,488✔

2737

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

6,081,488✔

2738

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

6,081,488✔

2739

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

6,081,488✔

2740

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

6,081,488✔

2741

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

6,081,488✔

2742

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

6,081,488✔

2743

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

6,081,488✔

2744

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

6,081,488✔

2745

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

6,081,488✔

2747

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

6,081,488✔

2748

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

6,081,488✔

2750

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

6,081,488✔

2752

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

6,081,488✔

2753

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

6,081,488✔

2755

  int32_t calcScanPlanListSize = 0;

3,040,744✔

2756

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &calcScanPlanListSize));

3,040,744✔

2757

  if (calcScanPlanListSize > 0) {

3,040,744✔

2758

    pReq->calcScanPlanList = taosArrayInit(calcScanPlanListSize, sizeof(SStreamCalcScan));

3,033,494✔

2759

    if (pReq->calcScanPlanList == NULL) {

3,033,494✔

2762

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

7,044,090✔

2763

      SStreamCalcScan calcScan = {0};

4,010,596✔

2764

      int32_t         vgListSize = 0;

4,010,596✔

2765

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgListSize));

4,010,596✔

2766

      if (vgListSize > 0) {

4,010,596✔

2767

        calcScan.vgList = taosArrayInit(vgListSize, sizeof(int32_t));

4,010,596✔

2768

        if (calcScan.vgList == NULL) {

4,010,596✔

2771

        for (int32_t j = 0; j < vgListSize; ++j) {

8,021,192✔

2772

          int32_t vgId = 0;

4,010,596✔

2773

          TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgId));

4,010,596✔

2774

          if (taosArrayPush(calcScan.vgList, &vgId) == NULL) {

8,021,192✔

2779

      TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &calcScan.readFromCache));

4,010,596✔

2780

      TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&calcScan.scanPlan, NULL));

4,010,596✔

2781

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

8,021,192✔

2787

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

6,081,488✔

2788

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pReq->calcPlan, NULL));

6,081,488✔

2789

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

6,081,488✔

2790

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

6,081,488✔

2792

  int32_t forceOutColsSize = 0;

3,040,744✔

2793

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &forceOutColsSize));

3,040,744✔

2794

  if (forceOutColsSize > 0) {

3,040,744✔

2795

    pReq->forceOutCols = taosArrayInit(forceOutColsSize, sizeof(SStreamOutCol));

80,344✔

2796

    if (pReq->forceOutCols == NULL) {

80,344✔

2799

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

606,562✔

2800

      SStreamOutCol outCol = {0};

526,218✔

2801

      int64_t       exprLen = 0;

526,218✔

2802

      TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&outCol.expr, &exprLen));

526,218✔

2803

      TAOS_CHECK_EXIT(tDecodeU8(pDecoder, &outCol.type.type));

526,218✔

2804

      TAOS_CHECK_EXIT(tDecodeU8(pDecoder, &outCol.type.precision));

526,218✔

2805

      TAOS_CHECK_EXIT(tDecodeU8(pDecoder, &outCol.type.scale));

526,218✔

2806

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &outCol.type.bytes));

526,218✔

2807

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

1,052,436✔

2813

  switch (pReq->triggerType) {

3,040,744✔

2814

    case WINDOW_TYPE_STATE: {

1,096,518✔

2816

      if (!tDecodeIsEnd(pDecoder)) {

1,096,518✔

2817

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

2,193,036✔

2819

      if (!tDecodeIsEnd(pDecoder)) {

1,096,518✔

2820

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

2,193,036✔

2822

      if (!tDecodeIsEnd(pDecoder)) {

1,096,518✔

2823

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

2,193,036✔

2825

      break;

1,096,518✔

2827

    case WINDOW_TYPE_INTERVAL: {

1,452,376✔

2828

      if (!tDecodeIsEnd(pDecoder)) {

1,452,376✔

2829

        TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pReq->trigger.sliding.overlap));

2,904,752✔

2831

      break;

1,452,376✔

2833

    default:

491,850✔

2834

      break;

491,850✔

2838

_exit:

3,040,744✔

2840

  return code;

3,040,744✔

2844

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

1,532,924✔

2845

  SDecoder decoder = {0};

1,532,924✔

2846

  tDecoderInit(&decoder, buf, bufLen);

1,532,924✔

2847

  int32_t code = 0;

1,532,924✔

2850

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

1,532,924✔

2852

  TAOS_CHECK_EXIT(tDeserializeSCMCreateStreamReqImpl(&decoder, pReq));

1,532,924✔

2854

  tEndDecode(&decoder);

1,532,924✔

2856

_exit:

1,532,924✔

2858

  tDecoderClear(&decoder);

1,532,924✔

2859

  return code;

1,532,924✔

2863

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

258,306✔

2864

  int32_t  code = 0;

258,306✔

2867

  SEncoder encoder = {0};

258,306✔

2868

  tEncoderInit(&encoder, buf, bufLen);

258,306✔

2870

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

258,306✔

2872

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

258,306✔

2873

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

516,612✔

2874

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

516,612✔

2876

  tEndEncode(&encoder);

258,306✔

2878

_exit:

258,306✔

2879

  if (code) {

258,306✔

2882

    tlen = encoder.pos;

258,306✔

2884

  tEncoderClear(&encoder);

258,306✔

2885

  return tlen;

258,306✔

2888

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

110,305✔

2889

  SDecoder decoder = {0};

110,305✔

2890

  int32_t  code = 0;

110,305✔

2892

  tDecoderInit(&decoder, buf, bufLen);

110,305✔

2894

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

110,305✔

2895

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

220,610✔

2896

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

220,610✔

2898

  tEndDecode(&decoder);

110,305✔

2900

_exit:

110,305✔

2901

  tDecoderClear(&decoder);

110,305✔

2902

  return code;

110,305✔

2905

void tFreeMDropStreamReq(SMDropStreamReq *pReq) {

239,458✔

2906

  taosMemoryFreeClear(pReq->name);

239,458✔

2907

239,458✔

2909

static FORCE_INLINE void tFreeStreamCalcScan(void* pScan) {

10,245,023✔

2910

  if (pScan == NULL) {

10,245,023✔

2913

  SStreamCalcScan *pCalcScan = (SStreamCalcScan *)pScan;

10,245,023✔

2914

  taosArrayDestroy(pCalcScan->vgList);

10,245,023✔

2915

  taosMemoryFreeClear(pCalcScan->scanPlan);

10,245,023✔

2918

void tFreeStreamOutCol(void* pCol) {

1,074,656✔

2919

  if (pCol == NULL) {

1,074,656✔

2922

  SStreamOutCol *pOutCol = (SStreamOutCol *)pCol;

1,074,656✔

2923

  taosMemoryFreeClear(pOutCol->expr);

1,074,656✔

2928

void tFreeSCMCreateStreamReq(SCMCreateStreamReq *pReq) {

8,824,159✔

2929

  if (NULL == pReq) {

8,824,159✔

2930

    return;

1,532,924✔

2932

  taosMemoryFreeClear(pReq->name);

7,291,235✔

2933

  taosMemoryFreeClear(pReq->sql);

7,291,235✔

2934

  taosMemoryFreeClear(pReq->streamDB);

7,291,235✔

2935

  taosMemoryFreeClear(pReq->triggerDB);

7,291,235✔

2936

  taosMemoryFreeClear(pReq->outDB);

7,291,235✔

2937

  taosMemoryFreeClear(pReq->triggerTblName);

7,291,235✔

2938

  taosMemoryFreeClear(pReq->outTblName);

7,291,235✔

2940

  taosArrayDestroyP(pReq->calcDB, NULL);

7,291,235✔

2941

  pReq->calcDB = NULL;

7,291,235✔

2942

  taosArrayDestroyP(pReq->pNotifyAddrUrls, NULL);

7,291,235✔

2943

  pReq->pNotifyAddrUrls = NULL;

7,291,235✔

2945

  taosMemoryFreeClear(pReq->triggerFilterCols);

7,291,235✔

2946

  taosMemoryFreeClear(pReq->triggerCols);

7,291,235✔

2947

  taosMemoryFreeClear(pReq->partitionCols);

7,291,235✔

2949

  taosArrayDestroy(pReq->outTags);

7,291,235✔

2950

  pReq->outTags = NULL;

7,291,235✔

2951

  taosArrayDestroy(pReq->outCols);

7,291,235✔

2952

  pReq->outCols = NULL;

7,291,235✔

2954

  switch (pReq->triggerType) {

7,291,235✔

2955

    case WINDOW_TYPE_STATE:

2,301,774✔

2956

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

2,301,774✔

2957

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

2,301,774✔

2958

      break;

2,301,774✔

2959

    case WINDOW_TYPE_EVENT:

643,605✔

2960

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

643,605✔

2961

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

643,605✔

2962

      break;

643,605✔

2963

    default:

4,345,856✔

2964

      break;

4,345,856✔

2967

  taosMemoryFreeClear(pReq->triggerScanPlan);

7,291,235✔

2968

  taosArrayDestroyEx(pReq->calcScanPlanList, tFreeStreamCalcScan);

7,291,235✔

2969

  pReq->calcScanPlanList = NULL;

7,291,235✔

2970

  taosMemoryFreeClear(pReq->triggerPrevFilter);

7,291,235✔

2972

  taosMemoryFreeClear(pReq->calcPlan);

7,291,235✔

2973

  taosMemoryFreeClear(pReq->subTblNameExpr);

7,291,235✔

2974

  taosMemoryFreeClear(pReq->tagValueExpr);

7,291,235✔

2975

  taosArrayDestroyEx(pReq->forceOutCols, tFreeStreamOutCol);

7,291,235✔

2976

  pReq->forceOutCols = NULL;

7,291,235✔

2979

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

1,272,444✔

2980

  int32_t code = 0, lino = 0;

1,272,444✔

2981

  if (NULL == pSrc) {

1,272,444✔

2985

  void* p = NULL;

1,272,444✔

2986

  int32_t num = 0;

1,272,444✔

2987

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

1,272,444✔

2988

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

1,272,444✔

2990

  SCMCreateStreamReq* pDst = *ppDst;

1,272,444✔

2992

  if (pSrc->outDB) {

1,272,444✔

2993

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

1,268,819✔

2994

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

1,268,819✔

2997

  if (pSrc->triggerTblName) {

1,272,444✔

2998

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

1,268,777✔

2999

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

1,268,777✔

3002

  if (pSrc->outTblName) {

1,272,444✔

3003

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

1,268,819✔

3004

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

1,268,819✔

3007

  if (pSrc->pNotifyAddrUrls) {

1,272,444✔

3008

    num = taosArrayGetSize(pSrc->pNotifyAddrUrls);

123,300✔

3009

    if (num > 0) {

123,300✔

3010

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

123,300✔

3011

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

123,300✔

3013

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

246,600✔

3014

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

123,300✔

3015

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

123,300✔

3016

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

246,600✔

3020

  if (pSrc->triggerFilterCols) {

1,272,444✔

3021

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

72,798✔

3022

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

72,798✔

3025

  if (pSrc->triggerCols) {

1,272,444✔

3026

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

1,241,145✔

3027

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

1,241,145✔

3030

  if (pSrc->partitionCols) {

1,272,444✔

3031

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

729,468✔

3032

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

729,468✔

3035

  if (pSrc->outCols) {

1,272,444✔

3036

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

1,268,819✔

3037

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

1,268,819✔

3040

  if (pSrc->outTags) {

1,272,444✔

3041

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

729,468✔

3042

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

729,468✔

3045

  pDst->triggerType = pSrc->triggerType;

1,272,444✔

3047

  switch (pSrc->triggerType) {

1,272,444✔

3048

    case WINDOW_TYPE_STATE:

529,643✔

3049

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

529,643✔

3050

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

529,643✔

3051

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

529,643✔

3052

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

529,643✔

3053

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

13,566✔

3054

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

13,566✔

3056

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

529,643✔

3057

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

529,643✔

3058

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

529,643✔

3060

      break;

529,643✔

3061

    case WINDOW_TYPE_EVENT:

116,644✔

3062

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

116,644✔

3063

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

116,644✔

3064

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

116,644✔

3067

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

116,644✔

3068

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

116,644✔

3069

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

116,644✔

3071

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

116,644✔

3072

      break;

116,644✔

3073

    default:

626,157✔

3074

      pDst->trigger = pSrc->trigger;

626,157✔

3075

      break;

626,157✔

3079

  if (pSrc->triggerScanPlan) {

1,272,444✔

3080

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

1,268,777✔

3081

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

1,268,777✔

3084

  if (pSrc->calcScanPlanList) {

1,272,444✔

3085

    num = taosArrayGetSize(pSrc->calcScanPlanList);

1,268,819✔

3086

    if (num > 0) {

1,268,819✔

3087

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

1,268,819✔

3088

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

1,268,819✔

3090

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

2,962,093✔

3091

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

1,693,274✔

3092

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

1,693,274✔

3094

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

1,693,274✔

3095

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

1,693,274✔

3097

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

1,693,274✔

3098

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

1,693,274✔

3100

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

3,386,548✔

3104

  if (pSrc->triggerPrevFilter) {

1,272,444✔

3109

  if (pSrc->calcPlan) {

1,272,444✔

3110

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

1,268,819✔

3111

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

1,268,819✔

3114

  if (pSrc->subTblNameExpr) {

1,272,444✔

3115

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

729,468✔

3116

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

729,468✔

3119

  if (pSrc->tagValueExpr) {

1,272,444✔

3120

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

729,468✔

3121

    TSDB_CHECK_NULL(pDst->tagValueExpr, code, lino, _exit, terrno);

729,468✔

3124

  if (pSrc->forceOutCols) {

1,272,444✔

3125

    num = taosArrayGetSize(pSrc->forceOutCols);

40,172✔

3126

    if (num > 0) {

40,172✔

3127

      pDst->forceOutCols = taosArrayInit(num, sizeof(SStreamOutCol));

40,172✔

3128

      TSDB_CHECK_NULL(pDst->forceOutCols, code, lino, _exit, terrno);

40,172✔

3130

    for (int32_t i = 0; i < num; ++i) {

303,281✔

3131

      SStreamOutCol* scol = taosArrayGet(pSrc->forceOutCols, i);

263,109✔

3132

      SStreamOutCol  dcol = {.type = scol->type};

263,109✔

3134

      dcol.expr = COPY_STR(scol->expr);

263,109✔

3135

      TSDB_CHECK_NULL(dcol.expr, code, lino, _exit, terrno);

263,109✔

3137

      TSDB_CHECK_NULL(taosArrayPush(pDst->forceOutCols, &dcol), code, lino, _exit, terrno);

526,218✔

3141

  pDst->triggerTblUid = pSrc->triggerTblUid;

1,272,444✔

3142

  pDst->triggerTblType = pSrc->triggerTblType;

1,272,444✔

3143

  pDst->deleteReCalc = pSrc->deleteReCalc;

1,272,444✔

3144

  pDst->deleteOutTbl = pSrc->deleteOutTbl;

1,272,444✔

3146

_exit:

1,272,444✔

3148

  if (code) {

1,272,444✔

3153

  return code;

1,272,444✔

3157

int32_t tSerializeSMPauseStreamReq(void *buf, int32_t bufLen, const SMPauseStreamReq *pReq) {

162,670✔

3158

  int32_t  code = 0;

162,670✔

3161

  SEncoder encoder = {0};

162,670✔

3162

  tEncoderInit(&encoder, buf, bufLen);

162,670✔

3163

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

162,670✔

3165

  int32_t nameLen = pReq->name == NULL ? 0 : (int32_t)strlen(pReq->name) + 1;

162,670✔

3166

  TAOS_CHECK_EXIT(tEncodeBinary(&encoder, pReq->name, nameLen));

325,340✔

3167

  TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->igNotExists));

325,340✔

3168

  tEndEncode(&encoder);

162,670✔

3170

_exit:

162,670✔

3171

  if (code) {

162,670✔

3174

    tlen = encoder.pos;

162,670✔

3176

  tEncoderClear(&encoder);

162,670✔

3177

  return tlen;

162,670✔

3180

int32_t tDeserializeSMPauseStreamReq(void *buf, int32_t bufLen, SMPauseStreamReq *pReq) {

79,937✔

3181

  SDecoder decoder = {0};

79,937✔

3182

  int32_t  code = 0;

79,937✔

3185

  tDecoderInit(&decoder, buf, bufLen);

79,937✔

3186

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

79,937✔

3187

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(&decoder, (void**)&pReq->name, NULL));

159,874✔

3188

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->igNotExists));

159,874✔

3189

  tEndDecode(&decoder);

79,937✔

3191

_exit:

79,937✔

3192

  tDecoderClear(&decoder);

79,937✔

3193

  return code;

79,937✔

3196

void tFreeMPauseStreamReq(SMPauseStreamReq *pReq) {

81,335✔

3197

  taosMemoryFreeClear(pReq->name);

81,335✔

3198

81,335✔

3200

int32_t tSerializeSMResumeStreamReq(void *buf, int32_t bufLen, const SMResumeStreamReq *pReq) {

162,670✔

3201

  SEncoder encoder = {0};

162,670✔

3202

  int32_t  code = 0;

162,670✔

3205

  tEncoderInit(&encoder, buf, bufLen);

162,670✔

3206

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

162,670✔

3207

  int32_t nameLen = pReq->name == NULL ? 0 : (int32_t)strlen(pReq->name) + 1;

162,670✔

3208

  TAOS_CHECK_EXIT(tEncodeBinary(&encoder, pReq->name, nameLen));

325,340✔

3209

  TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->igNotExists));

325,340✔

3210

  TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->igUntreated));

325,340✔

3211

  tEndEncode(&encoder);

162,670✔

3213

_exit:

162,670✔

3214

  if (code) {

162,670✔

3217

    tlen = encoder.pos;

162,670✔

3219

  tEncoderClear(&encoder);

162,670✔

3220

  return tlen;

162,670✔

3223

int32_t tDeserializeSMResumeStreamReq(void *buf, int32_t bufLen, SMResumeStreamReq *pReq) {

79,937✔

3224

  SDecoder decoder = {0};

79,937✔

3225

  int32_t  code = 0;

79,937✔

3228

  tDecoderInit(&decoder, buf, bufLen);

79,937✔

3229

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

79,937✔

3230

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(&decoder, (void**)&pReq->name, NULL));

159,874✔

3231

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->igNotExists));

159,874✔

3232

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->igUntreated));

159,874✔

3233

  tEndDecode(&decoder);

79,937✔

3235

_exit:

79,937✔

3236

  tDecoderClear(&decoder);

79,937✔

3237

  return code;

79,937✔

3240

void tFreeMResumeStreamReq(SMResumeStreamReq *pReq) {

81,335✔

3241

  taosMemoryFreeClear(pReq->name);

81,335✔

3242

81,335✔

3244

int32_t tSerializeSMRecalcStreamReq(void *buf, int32_t bufLen, const SMRecalcStreamReq *pReq) {

278,890✔

3245

  SEncoder encoder = {0};

278,890✔

3246

  int32_t  code = 0;

278,890✔

3249

  tEncoderInit(&encoder, buf, bufLen);

278,890✔

3250

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

278,890✔

3251

  int32_t nameLen = pReq->name == NULL ? 0 : (int32_t)strlen(pReq->name) + 1;

278,890✔

3252

  TAOS_CHECK_EXIT(tEncodeBinary(&encoder, pReq->name, nameLen));

557,780✔

3253

  TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->calcAll));

557,780✔

3254

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->timeRange.skey));

557,780✔

3255

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->timeRange.ekey));

557,780✔

3256

  tEndEncode(&encoder);

278,890✔

3258

_exit:

278,890✔

3259

  if (code) {

278,890✔

3262

    tlen = encoder.pos;

278,890✔

3264

  tEncoderClear(&encoder);

278,890✔

3265

  return tlen;

278,890✔

3268

int32_t tDeserializeSMRecalcStreamReq(void *buf, int32_t bufLen, SMRecalcStreamReq *pReq) {

139,422✔

3269

  SDecoder decoder = {0};

139,422✔

3270

  int32_t  code = 0;

139,422✔

3273

  tDecoderInit(&decoder, buf, bufLen);

139,422✔

3274

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

139,422✔

3276

  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(&decoder, (void**)&pReq->name, NULL));

278,844✔

3277

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->calcAll));

278,844✔

3278

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->timeRange.skey));

278,844✔

3279

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->timeRange.ekey));

278,844✔

3280

  tEndDecode(&decoder);

139,422✔

3282

_exit:

139,422✔

3283

  tDecoderClear(&decoder);

139,422✔

3284

  return code;

139,422✔

3287

void tFreeMRecalcStreamReq(SMRecalcStreamReq *pReq) {

278,867✔

3288

  taosMemoryFreeClear(pReq->name);

278,867✔

3289

278,867✔

3291

static int32_t tEncodeStreamProgressReq(SEncoder *pEncoder, const SStreamProgressReq *pReq) {

2,910,808✔

3292

  int32_t code = 0;

2,910,808✔

3295

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));

5,821,616✔

3296

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->taskId));

5,821,616✔

3297

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->fetchIdx));

5,821,616✔

3299

_exit:

2,910,808✔

3300

  return code;

2,910,808✔

3303

int32_t tSerializeStreamProgressReq(void *buf, int32_t bufLen, const SStreamProgressReq *pReq) {

2,910,808✔

3304

  SEncoder encoder = {0};

2,910,808✔

3305

  int32_t  code = 0;

2,910,808✔

3308

  tEncoderInit(&encoder, buf, bufLen);

2,910,808✔

3310

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

2,910,808✔

3311

  TAOS_CHECK_EXIT(tEncodeStreamProgressReq(&encoder, pReq));

2,910,808✔

3313

  tEndEncode(&encoder);

2,910,808✔

3315

_exit:

2,910,808✔

3316

  if (code) {

2,910,808✔

3319

    tlen = encoder.pos;

2,910,808✔

3321

  tEncoderClear(&encoder);

2,910,808✔

3322

  return tlen;

2,910,808✔

3389

static int32_t tDecodeStreamProgressRsp(SDecoder *pDecoder, SStreamProgressRsp *pRsp) {

1,455,404✔

3390

  int32_t code = 0;

1,455,404✔

3393

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pRsp->streamId));

2,910,808✔

3394

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, (int8_t *)&pRsp->fillHisFinished));

2,910,808✔

3395

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pRsp->progressDelay));

2,910,808✔

3396

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->fetchIdx));

2,910,808✔

3398

_exit:

1,455,404✔

3399

  return code;

1,455,404✔

3402

int32_t tDeserializeSStreamProgressRsp(void *buf, int32_t bufLen, SStreamProgressRsp *pRsp) {

1,455,404✔

3403

  SDecoder decoder = {0};

1,455,404✔

3404

  int32_t  code = 0;

1,455,404✔

3407

  tDecoderInit(&decoder, buf, bufLen);

1,455,404✔

3409

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

1,455,404✔

3410

  TAOS_CHECK_EXIT(tDecodeStreamProgressRsp(&decoder, pRsp));

1,455,404✔

3412

  tEndDecode(&decoder);

1,455,404✔

3414

_exit:

1,455,404✔

3415

  tDecoderClear(&decoder);

1,455,404✔

3416

  return code;

1,455,404✔

3419

int32_t tSerializeSTriggerOrigTableInfoRsp(void* buf, int32_t bufLen, const SSTriggerOrigTableInfoRsp* pRsp){

728,704✔

3420

  SEncoder encoder = {0};

728,704✔

3421

  int32_t  code = TSDB_CODE_SUCCESS;

730,152✔

3422

  int32_t  lino = 0;

730,152✔

3423

  int32_t  tlen = 0;

730,152✔

3425

  tEncoderInit(&encoder, buf, bufLen);

730,152✔

3426

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

733,048✔

3428

  int32_t size = taosArrayGetSize(pRsp->cols);

731,625✔

3429

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, size));

733,048✔

3430

  for (int32_t i = 0; i < size; ++i) {

1,953,852✔

3431

    OTableInfoRsp* oInfo = taosArrayGet(pRsp->cols, i);

1,219,356✔

3432

    if (oInfo == NULL) {

1,219,356✔

3437

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, oInfo->suid));

2,436,526✔

3438

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, oInfo->uid));

2,435,103✔

3439

    TAOS_CHECK_EXIT(tEncodeI16(&encoder, oInfo->cid));

2,438,737✔

3442

  tEndEncode(&encoder);

734,496✔

3444

_exit:

730,177✔

3445

  if (code != TSDB_CODE_SUCCESS) {

730,177✔

3448

    tlen = encoder.pos;

730,177✔

3450

  tEncoderClear(&encoder);

730,177✔

3451

  return tlen;

730,177✔

3454

int32_t tDserializeSTriggerOrigTableInfoRsp(void* buf, int32_t bufLen, SSTriggerOrigTableInfoRsp* pRsp){

366,524✔

3455

  SDecoder decoder = {0};

366,524✔

3456

  int32_t  code = TSDB_CODE_SUCCESS;

366,524✔

3457

  int32_t  lino = 0;

366,524✔

3459

  tDecoderInit(&decoder, buf, bufLen);

366,524✔

3460

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

366,524✔

3462

  int32_t size = 0;

366,524✔

3463

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &size));

366,524✔

3464

  pRsp->cols = taosArrayInit(size, sizeof(OTableInfoRsp));

366,524✔

3465

  if (pRsp->cols == NULL) {

366,524✔

3470

  for (int32_t i = 0; i < size; ++i) {

976,926✔

3471

    OTableInfoRsp* oInfo = taosArrayReserve(pRsp->cols, 1);

610,402✔

3472

    if (oInfo == NULL) {

610,402✔

3477

    TAOS_CHECK_RETURN(tDecodeI64(&decoder, &oInfo->suid));

1,220,804✔

3478

    TAOS_CHECK_RETURN(tDecodeI64(&decoder, &oInfo->uid));

1,220,804✔

3479

    TAOS_CHECK_RETURN(tDecodeI16(&decoder, &oInfo->cid));

1,220,804✔

3482

  tEndDecode(&decoder);

366,524✔

3484

_exit:

366,524✔

3485

  tDecoderClear(&decoder);

366,524✔

3486

  return code;

366,524✔

3489

void    tDestroySTriggerOrigTableInfoRsp(SSTriggerOrigTableInfoRsp* pRsp){

105,139,131✔

3490

  taosArrayDestroy(pRsp->cols);

105,139,131✔

3491

105,136,255✔

3493

void tDestroySTriggerPullRequest(SSTriggerPullRequestUnion* pReq) {

214,587,197✔

3494

  if (pReq == NULL) return;

214,587,197✔

3495

  if (pReq->base.type == STRIGGER_PULL_WAL_DATA_NEW || pReq->base.type == STRIGGER_PULL_WAL_CALC_DATA_NEW) {

234,945,266✔

3496

    SSTriggerWalDataNewRequest* pRequest = (SSTriggerWalDataNewRequest*)pReq;

20,390,807✔

3497

    taosArrayDestroy(pRequest->versions);

20,390,807✔

3498

    tSimpleHashCleanup(pRequest->ranges);

20,358,069✔

3499

  } else if (pReq->base.type == STRIGGER_PULL_TSDB_DATA) {

194,228,518✔

3500

    SSTriggerTsdbDataRequest* pRequest = (SSTriggerTsdbDataRequest*)pReq;

591,505✔

3501

    if (pRequest->cids != NULL) {

591,505✔

3502

      taosArrayDestroy(pRequest->cids);

591,505✔

3503

      pRequest->cids = NULL;

591,505✔

3505

  } else if (pReq->base.type == STRIGGER_PULL_VTABLE_INFO) {

193,629,713✔

3506

    SSTriggerVirTableInfoRequest* pRequest = (SSTriggerVirTableInfoRequest*)pReq;

1,950,183✔

3507

    if (pRequest->cids != NULL) {

1,950,183✔

3508

      taosArrayDestroy(pRequest->cids);

1,950,183✔

3509

      pRequest->cids = NULL;

1,950,183✔

3511

  } else if (pReq->base.type == STRIGGER_PULL_VTABLE_PSEUDO_COL) {

191,668,576✔

3512

    SSTriggerVirTablePseudoColRequest *pRequest = (SSTriggerVirTablePseudoColRequest*)pReq;

1,139,634✔

3513

    if (pRequest->cids != NULL) {

1,139,634✔

3514

      taosArrayDestroy(pRequest->cids);

1,139,634✔

3515

      pRequest->cids = NULL;

1,139,634✔

3517

  } else if (pReq->base.type == STRIGGER_PULL_OTABLE_INFO) {

190,516,546✔

3518

    SSTriggerOrigTableInfoRequest* pRequest = (SSTriggerOrigTableInfoRequest*)pReq;

366,524✔

3519

    if (pRequest->cols != NULL) {

366,524✔

3520

      taosArrayDestroy(pRequest->cols);

366,524✔

3521

      pRequest->cols = NULL;

366,524✔

3523

  } else if (pReq->base.type == STRIGGER_PULL_SET_TABLE) {

190,146,414✔

3524

    SSTriggerSetTableRequest* pRequest = (SSTriggerSetTableRequest*)pReq;

366,524✔

3525

    tSimpleHashCleanup(pRequest->uidInfoTrigger);

366,524✔

3526

    tSimpleHashCleanup(pRequest->uidInfoCalc);

366,524✔

3530

int32_t encodeColsArray(SEncoder* encoder, SArray* cids) {

7,360,433✔

3531

  int32_t  code = TSDB_CODE_SUCCESS;

7,360,433✔

3532

  int32_t  lino = 0;

7,360,433✔

3533

  int32_t size = taosArrayGetSize(cids);

7,360,433✔

3534

  TAOS_CHECK_EXIT(tEncodeI32(encoder, size));

7,362,644✔

3535

  for (int32_t i = 0; i < size; ++i) {

22,484,872✔

3536

    col_id_t* pColId = taosArrayGet(cids, i);

15,124,422✔

3537

    if (pColId == NULL) {

15,124,422✔

3542

    TAOS_CHECK_EXIT(tEncodeI16(encoder, *pColId));

30,246,650✔

3544

  _exit:

7,360,450✔

3546

  return code;

7,360,450✔

3549

int32_t decodeColsArray(SDecoder* decoder, SArray** cids) {

3,681,322✔

3550

  int32_t code = TSDB_CODE_SUCCESS;

3,681,322✔

3551

  int32_t lino = 0;

3,681,322✔

3552

  int32_t size = 0;

3,681,322✔

3554

  TAOS_CHECK_EXIT(tDecodeI32(decoder, &size));

3,681,322✔

3555

  if (size > 0){

3,681,322✔

3556

    *cids = taosArrayInit(size, sizeof(col_id_t));

3,681,322✔

3557

    if (*cids == NULL) {

3,681,322✔

3563

    for (int32_t i = 0; i < size; ++i) {

11,243,533✔

3564

      col_id_t* pColId = taosArrayReserve(*cids, 1);

7,562,211✔

3565

      if (pColId == NULL) {

7,562,211✔

3570

      TAOS_CHECK_RETURN(tDecodeI16(decoder, pColId));

7,562,211✔

3574

_exit:

3,681,322✔

3575

  if (code != TSDB_CODE_SUCCESS) {

3,681,322✔

3579

  return code;

3,681,322✔

3582

static int32_t encodeSetTableMapInfo(SEncoder* encoder, SSHashObj* pInfo) {

1,466,096✔

3583

  int32_t  code = TSDB_CODE_SUCCESS;

1,466,096✔

3584

  int32_t  lino = 0;

1,466,096✔

3585

  int32_t size = tSimpleHashGetSize(pInfo);

1,466,096✔

3586

  TAOS_CHECK_EXIT(tEncodeI32(encoder, size));

1,466,096✔

3587

  int32_t iter = 0;

1,466,096✔

3588

  void*   px = tSimpleHashIterate(pInfo, NULL, &iter);

1,466,096✔

3589

  while (px != NULL) {

4,397,862✔

3590

    int64_t* uid = tSimpleHashGetKey(px, NULL);

2,931,766✔

3591

    TAOS_CHECK_EXIT(tEncodeI64(encoder, *uid));

5,863,532✔

3592

    TAOS_CHECK_EXIT(tEncodeI64(encoder, *(uid + 1)));

5,863,532✔

3593

    SSHashObj* info = *(SSHashObj**)px;

2,931,766✔

3594

    int32_t len = tSimpleHashGetSize(info);

2,931,766✔

3595

    TAOS_CHECK_EXIT(tEncodeI32(encoder, len));

2,931,766✔

3596

    int32_t iter1 = 0;

2,931,766✔

3597

    void*   px1 = tSimpleHashIterate(info, NULL, &iter1);

2,931,766✔

3598

    while (px1 != NULL) {

9,870,364✔

3599

      int16_t* slot = tSimpleHashGetKey(px1, NULL);

6,940,023✔

3600

      int16_t* cid = (int16_t*)px1;

6,940,023✔

3601

      TAOS_CHECK_EXIT(tEncodeI16(encoder, *slot));

13,880,046✔

3602

      TAOS_CHECK_EXIT(tEncodeI16(encoder, *cid));

13,880,832✔

3604

      px1 = tSimpleHashIterate(info, px1, &iter1);

6,940,809✔

3607

    px = tSimpleHashIterate(pInfo, px, &iter);

2,930,341✔

3610

_exit:

1,466,096✔

3611

  return code;

1,466,096✔

3614

int32_t tSerializeSTriggerPullRequest(void* buf, int32_t bufLen, const SSTriggerPullRequest* pReq) {

431,704,347✔

3615

  SEncoder encoder = {0};

431,704,347✔

3616

  int32_t  code = TSDB_CODE_SUCCESS;

431,762,185✔

3617

  int32_t  lino = 0;

431,762,185✔

3618

  int32_t  tlen = 0;

431,762,185✔

3620

  tEncoderInit(&encoder, buf, bufLen);

431,762,185✔

3621

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

431,729,785✔

3623

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, pReq->type));

863,353,054✔

3624

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->streamId));

863,288,504✔

3625

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->readerTaskId));

863,386,929✔

3626

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->sessionId));

863,393,299✔

3628

  switch (pReq->type) {

431,681,964✔

3629

    case STRIGGER_PULL_SET_TABLE: {

733,048✔

3630

      SSTriggerSetTableRequest* pRequest = (SSTriggerSetTableRequest*)pReq;

733,048✔

3631

      TAOS_CHECK_EXIT(encodeSetTableMapInfo(&encoder, pRequest->uidInfoTrigger));

733,048✔

3632

      TAOS_CHECK_EXIT(encodeSetTableMapInfo(&encoder, pRequest->uidInfoCalc));

733,048✔

3633

      break;

733,048✔

3635

    case STRIGGER_PULL_LAST_TS: {

3,531,374✔

3636

      break;

3,531,374✔

3638

    case STRIGGER_PULL_FIRST_TS: {

2,806,332✔

3639

      SSTriggerFirstTsRequest* pRequest = (SSTriggerFirstTsRequest*)pReq;

2,806,332✔

3640

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->gid));

5,614,872✔

3641

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->startTime));

5,614,872✔

3642

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

5,614,872✔

3643

      break;

2,808,540✔

3645

    case STRIGGER_PULL_TSDB_META: {

5,551,320✔

3646

      SSTriggerTsdbMetaRequest* pRequest = (SSTriggerTsdbMetaRequest*)pReq;

5,551,320✔

3647

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->startTime));

11,102,640✔

3648

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->endTime));

11,102,640✔

3649

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->gid));

11,102,640✔

3650

      TAOS_CHECK_EXIT(tEncodeI8(&encoder, pRequest->order));

11,102,640✔

3651

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

11,102,640✔

3652

      break;

5,551,320✔

3657

    case STRIGGER_PULL_TSDB_TS_DATA: {

267,994✔

3658

      SSTriggerTsdbTsDataRequest* pRequest = (SSTriggerTsdbTsDataRequest*)pReq;

267,994✔

3659

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->suid));

535,988✔

3660

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->uid));

535,988✔

3661

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->skey));

535,988✔

3662

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ekey));

535,988✔

3663

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

535,988✔

3664

      break;

267,994✔

3666

    case STRIGGER_PULL_TSDB_TRIGGER_DATA: {

1,040,242✔

3667

      SSTriggerTsdbTriggerDataRequest* pRequest = (SSTriggerTsdbTriggerDataRequest*)pReq;

1,040,242✔

3668

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->startTime));

2,080,484✔

3669

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->gid));

2,080,484✔

3670

      TAOS_CHECK_EXIT(tEncodeI8(&encoder, pRequest->order));

2,080,484✔

3671

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

2,080,484✔

3672

      break;

1,040,242✔

3674

    case STRIGGER_PULL_TSDB_TRIGGER_DATA_NEXT: {

1,120,750✔

3675

      break;

1,120,750✔

3677

    case STRIGGER_PULL_TSDB_CALC_DATA: {

204,934,740✔

3678

      SSTriggerTsdbCalcDataRequest* pRequest = (SSTriggerTsdbCalcDataRequest*)pReq;

204,934,740✔

3679

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->gid));

409,868,038✔

3680

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->skey));

409,868,038✔

3681

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ekey));

409,868,038✔

3682

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

409,868,038✔

3683

      break;

204,934,740✔

3688

    case STRIGGER_PULL_TSDB_DATA: {

1,183,010✔

3689

      SSTriggerTsdbDataRequest* pRequest = (SSTriggerTsdbDataRequest*)pReq;

1,183,010✔

3690

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->suid));

2,366,020✔

3691

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->uid));

2,366,020✔

3692

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->skey));

2,366,020✔

3693

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ekey));

2,366,020✔

3694

      TAOS_CHECK_EXIT(encodeColsArray(&encoder, pRequest->cids));

1,183,010✔

3695

      TAOS_CHECK_EXIT(tEncodeI8(&encoder, pRequest->order));

2,366,020✔

3696

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ver));

2,366,020✔

3697

      break;

1,183,010✔

3702

    case STRIGGER_PULL_WAL_META_NEW: {

98,087,140✔

3703

      SSTriggerWalMetaNewRequest* pRequest = (SSTriggerWalMetaNewRequest*)pReq;

98,087,140✔

3704

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->lastVer));

196,151,085✔

3705

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->ctime));

196,134,309✔

3706

      break;

98,070,364✔

3708

    case STRIGGER_PULL_WAL_DATA_NEW:

40,705,339✔

3710

      SSTriggerWalDataNewRequest* pRequest = (SSTriggerWalDataNewRequest*)pReq;

40,705,339✔

3711

      int32_t                     nVersion = taosArrayGetSize(pRequest->versions);

40,705,339✔

3712

      TAOS_CHECK_EXIT(tEncodeI32(&encoder, nVersion));

40,699,375✔

3713

      for (int32_t i = 0; i < nVersion; i++) {

87,253,980✔

3714

        int64_t ver = *(int64_t*)TARRAY_GET_ELEM(pRequest->versions, i);

46,561,162✔

3715

        TAOS_CHECK_EXIT(tEncodeI64(&encoder, ver));

46,554,605✔

3717

      int32_t nRanges = tSimpleHashGetSize(pRequest->ranges);

40,692,818✔

3718

      TAOS_CHECK_EXIT(tEncodeI32(&encoder, nRanges));

40,705,245✔

3719

      int32_t iter = 0;

40,705,245✔

3720

      void*   px = tSimpleHashIterate(pRequest->ranges, NULL, &iter);

40,706,011✔

3721

      while (px != NULL) {

48,472,915✔

3722

        uint64_t* gid = tSimpleHashGetKey(px, NULL);

7,771,210✔

3723

        TAOS_CHECK_EXIT(tEncodeU64(&encoder, *gid));

15,540,974✔

3724

        int64_t* key = (int64_t*)px;

7,769,764✔

3725

        TAOS_CHECK_EXIT(tEncodeI64(&encoder, key[0]));

15,538,763✔

3726

        TAOS_CHECK_EXIT(tEncodeI64(&encoder, key[1]));

15,537,315✔

3728

        px = tSimpleHashIterate(pRequest->ranges, px, &iter);

7,768,316✔

3730

      break;

40,701,705✔

3732

    case STRIGGER_PULL_WAL_META_DATA_NEW: {

59,698,929✔

3733

      SSTriggerWalMetaDataNewRequest* pRequest = (SSTriggerWalMetaDataNewRequest*)pReq;

59,698,929✔

3734

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->lastVer));

119,397,478✔

3735

      break;

59,698,549✔

3737

    case STRIGGER_PULL_GROUP_COL_VALUE: {

5,132,164✔

3738

      SSTriggerGroupColValueRequest* pRequest = (SSTriggerGroupColValueRequest*)pReq;

5,132,164✔

3739

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->gid));

10,264,328✔

3740

      break;

5,132,164✔

3742

    case STRIGGER_PULL_VTABLE_INFO: {

3,900,366✔

3743

      SSTriggerVirTableInfoRequest* pRequest = (SSTriggerVirTableInfoRequest*)pReq;

3,900,366✔

3744

      TAOS_CHECK_EXIT(encodeColsArray(&encoder, pRequest->cids));

3,900,366✔

3745

      break;

3,900,366✔

3747

    case STRIGGER_PULL_VTABLE_PSEUDO_COL: {

2,277,057✔

3748

      SSTriggerVirTablePseudoColRequest* pRequest = (SSTriggerVirTablePseudoColRequest*)pReq;

2,277,057✔

3749

      TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRequest->uid));

4,554,114✔

3750

      TAOS_CHECK_EXIT(encodeColsArray(&encoder, pRequest->cids));

2,277,057✔

3751

      break;

2,279,268✔

3753

    case STRIGGER_PULL_OTABLE_INFO: {

733,048✔

3754

      SSTriggerOrigTableInfoRequest* pRequest = (SSTriggerOrigTableInfoRequest*)pReq;

733,048✔

3755

      int32_t size = taosArrayGetSize(pRequest->cols);

733,048✔

3756

      TAOS_CHECK_EXIT(tEncodeI32(&encoder, size));

733,048✔

3757

      for (int32_t i = 0; i < size; ++i) {

1,953,852✔

3758

        OTableInfo* oInfo = taosArrayGet(pRequest->cols, i);

1,220,804✔

3759

        if (oInfo == NULL) {

1,220,804✔

3764

        TAOS_CHECK_EXIT(tEncodeCStr(&encoder, oInfo->refTableName));

2,441,608✔

3765

        TAOS_CHECK_EXIT(tEncodeCStr(&encoder, oInfo->refColName));

2,441,608✔

3767

      break;

733,048✔

3776

  tEndEncode(&encoder);

431,686,482✔

3778

_exit:

431,673,461✔

3779

  if (code != TSDB_CODE_SUCCESS) {

431,694,638✔

3782

    tlen = encoder.pos;

431,694,638✔

3784

  tEncoderClear(&encoder);

431,694,638✔

3785

  return tlen;

431,715,342✔

3788

static void destroyHash(void* data){

1,464,445✔

3789

  if (data){

1,464,445✔

3790

    SSHashObj* tmp = *(SSHashObj**)data;

1,464,445✔

3791

    tSimpleHashCleanup(tmp);

1,464,445✔

3793

1,465,883✔

3795

static int32_t decodeSetTableMapInfo(SDecoder* decoder, SSHashObj** ppInfo) {

733,048✔

3796

  int32_t  code = TSDB_CODE_SUCCESS;

733,048✔

3797

  int32_t  lino = 0;

733,048✔

3798

  int32_t size = 0;

733,048✔

3799

  TAOS_CHECK_EXIT(tDecodeI32(decoder, &size));

733,048✔

3800

  *ppInfo = tSimpleHashInit(size, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));

733,048✔

3801

  if (*ppInfo == NULL) {

733,048✔

3804

  tSimpleHashSetFreeFp(*ppInfo, destroyHash);

733,048✔

3806

  for (int32_t i = 0; i < size; ++i) {

2,198,931✔

3807

    int64_t id[2] = {0};

1,465,883✔

3808

    TAOS_CHECK_EXIT(tDecodeI64(decoder, id));

1,465,883✔

3809

    TAOS_CHECK_EXIT(tDecodeI64(decoder, id+1));

2,931,766✔

3810

    int32_t len = 0;

1,465,883✔

3811

    TAOS_CHECK_EXIT(tDecodeI32(decoder, &len));

1,465,883✔

3812

    SSHashObj* tmp = tSimpleHashInit(len, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT));

1,465,883✔

3813

    if (tmp == NULL) {

1,465,883✔

3816

    TAOS_CHECK_EXIT(tSimpleHashPut(*ppInfo, id, sizeof(id), &tmp, POINTER_BYTES));

1,465,883✔

3818

    for (int32_t j = 0; j < len; ++j) {

4,937,000✔

3819

      int16_t slotId = 0;

3,471,117✔

3820

      int16_t cid = 0;

3,471,117✔

3821

      TAOS_CHECK_EXIT(tDecodeI16(decoder, &slotId));

3,471,117✔

3822

      TAOS_CHECK_EXIT(tDecodeI16(decoder, &cid));

3,471,117✔

3823

      TAOS_CHECK_EXIT(tSimpleHashPut(tmp, &slotId, sizeof(slotId), &cid, sizeof(cid)));

3,471,117✔

3826

_exit:

733,048✔

3827

  if (code != TSDB_CODE_SUCCESS) {

733,048✔

3831

  return code;

733,048✔

3834

int32_t tDeserializeSTriggerPullRequest(void* buf, int32_t bufLen, SSTriggerPullRequestUnion* pReq) {

214,515,892✔

3835

  SDecoder decoder = {0};

214,515,892✔

3836

  int32_t  code = TSDB_CODE_SUCCESS;

214,515,892✔

3837

  int32_t  lino = 0;

214,515,892✔

3839

  tDecoderInit(&decoder, buf, bufLen);

214,515,892✔

3840

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

214,545,825✔

3842

  int32_t type = 0;

214,625,456✔

3843

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &type));

214,599,809✔

3844

  SSTriggerPullRequest* pBase = &(pReq->base);

214,599,809✔

3845

  pBase->type = type;

214,610,748✔

3846

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pBase->streamId));

429,215,817✔

3847

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pBase->readerTaskId));

429,222,476✔

3848

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pBase->sessionId));

429,237,707✔

3850

  switch (type) {

214,610,051✔

3851

    case STRIGGER_PULL_SET_TABLE: {

366,524✔

3852

      SSTriggerSetTableRequest* pRequest = &(pReq->setTableReq);

366,524✔

3853

      TAOS_CHECK_EXIT(decodeSetTableMapInfo(&decoder, &pRequest->uidInfoTrigger));

366,524✔

3854

      TAOS_CHECK_EXIT(decodeSetTableMapInfo(&decoder, &pRequest->uidInfoCalc));

366,524✔

3855

      break;

366,524✔

3857

    case STRIGGER_PULL_LAST_TS: {

1,765,143✔

3858

      break;

1,765,143✔

3860

    case STRIGGER_PULL_FIRST_TS: {

1,400,738✔

3861

      SSTriggerFirstTsRequest* pRequest = &(pReq->firstTsReq);

1,400,738✔

3862

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->gid));

2,804,368✔

3863

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->startTime));

2,804,368✔

3864

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

2,804,368✔

3865

      break;

1,402,184✔

3867

    case STRIGGER_PULL_TSDB_META: {

2,775,112✔

3868

      SSTriggerTsdbMetaRequest* pRequest = &(pReq->tsdbMetaReq);

2,775,112✔

3869

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->startTime));

5,550,224✔

3870

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->endTime));

5,550,224✔

3871

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->gid));

5,550,224✔

3872

      TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pRequest->order));

5,550,224✔

3873

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

5,550,224✔

3874

      break;

2,775,112✔

3879

    case STRIGGER_PULL_TSDB_TS_DATA: {

133,997✔

3880

      SSTriggerTsdbTsDataRequest* pRequest = &(pReq->tsdbTsDataReq);

133,997✔

3881

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->suid));

267,994✔

3882

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->uid));

267,994✔

3883

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->skey));

267,994✔

3884

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ekey));

267,994✔

3885

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

267,994✔

3886

      break;

133,997✔

3888

    case STRIGGER_PULL_TSDB_TRIGGER_DATA: {

520,121✔

3889

      SSTriggerTsdbTriggerDataRequest* pRequest = &(pReq->tsdbTriggerDataReq);

520,121✔

3890

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->startTime));

1,040,242✔

3891

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->gid));

1,040,242✔

3892

      TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pRequest->order));

1,040,242✔

3893

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

1,040,242✔

3894

      break;

520,121✔

3896

    case STRIGGER_PULL_TSDB_TRIGGER_DATA_NEXT: {

560,375✔

3897

      break;

560,375✔

3899

    case STRIGGER_PULL_TSDB_CALC_DATA: {

102,451,523✔

3900

      SSTriggerTsdbCalcDataRequest* pRequest = &(pReq->tsdbCalcDataReq);

102,451,523✔

3901

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->gid));

204,907,446✔

3902

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->skey));

204,913,273✔

3903

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ekey));

204,914,700✔

3904

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

204,914,700✔

3905

      break;

102,457,350✔

3910

    case STRIGGER_PULL_TSDB_DATA: {

591,505✔

3911

      SSTriggerTsdbDataRequest* pRequest = &(pReq->tsdbDataReq);

591,505✔

3912

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->suid));

1,183,010✔

3913

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->uid));

1,183,010✔

3914

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->skey));

1,183,010✔

3915

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ekey));

1,183,010✔

3916

      TAOS_CHECK_EXIT(decodeColsArray(&decoder, &pRequest->cids));

591,505✔

3917

      TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pRequest->order));

1,183,010✔

3918

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ver));

1,183,010✔

3919

      break;

591,505✔

3924

    case STRIGGER_PULL_WAL_META_NEW: {

48,364,855✔

3925

      SSTriggerWalMetaNewRequest* pRequest = &(pReq->walMetaNewReq);

48,364,855✔

3926

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->lastVer));

96,771,189✔

3927

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->ctime));

96,789,423✔

3928

      break;

48,400,561✔

3930

    case STRIGGER_PULL_WAL_DATA_NEW:

20,365,380✔

3932

      SSTriggerWalDataNewRequest* pRequest = &(pReq->walDataNewReq);

20,365,380✔

3933

      int32_t                     nVersion = 0;

20,365,380✔

3934

      TAOS_CHECK_EXIT(tDecodeI32(&decoder, &nVersion));

20,365,370✔

3935

      pRequest->versions = taosArrayInit_s(sizeof(int64_t), nVersion);

20,365,370✔

3936

      for (int32_t i = 0; i < nVersion; i++) {

43,659,144✔

3937

        int64_t* pVer = TARRAY_GET_ELEM(pRequest->versions, i);

23,293,018✔

3938

        TAOS_CHECK_EXIT(tDecodeI64(&decoder, pVer));

23,293,018✔

3940

      int32_t nRanges = 0;

20,366,126✔

3941

      TAOS_CHECK_EXIT(tDecodeI32(&decoder, &nRanges));

20,366,126✔

3942

      pRequest->ranges = tSimpleHashInit(nRanges, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT));

20,366,126✔

3943

      if (pRequest->ranges == NULL) {

20,355,885✔

3946

      for (int32_t i = 0; i < nRanges; i++) {

24,242,935✔

3947

        uint64_t gid = 0;

3,885,605✔

3948

        int64_t pRange[2] = {0};

3,885,605✔

3949

        TAOS_CHECK_EXIT(tDecodeU64(&decoder, &gid));

3,884,180✔

3950

        TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRange[0]));

3,885,605✔

3951

        TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRange[1]));

3,884,180✔

3952

        TAOS_CHECK_EXIT(tSimpleHashPut(pRequest->ranges, &gid, sizeof(gid), pRange, sizeof(pRange)));

3,884,180✔

3954

      break;

20,357,330✔

3956

    case STRIGGER_PULL_WAL_META_DATA_NEW: {

29,252,377✔

3957

      SSTriggerWalMetaDataNewRequest* pRequest = &(pReq->walMetaDataNewReq);

29,252,377✔

3958

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->lastVer));

58,536,098✔

3959

      break;

29,270,646✔

3961

    case STRIGGER_PULL_GROUP_COL_VALUE: {

2,563,097✔

3962

      SSTriggerGroupColValueRequest* pRequest = &(pReq->groupColValueReq);

2,563,097✔

3963

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->gid));

5,126,194✔

3964

      break;

2,563,097✔

3966

    case STRIGGER_PULL_VTABLE_INFO: {

1,950,183✔

3967

      SSTriggerVirTableInfoRequest* pRequest = &(pReq->virTableInfoReq);

1,950,183✔

3968

      TAOS_CHECK_EXIT(decodeColsArray(&decoder, &pRequest->cids));

1,950,183✔

3969

      break;

1,950,183✔

3971

    case STRIGGER_PULL_VTABLE_PSEUDO_COL: {

1,139,634✔

3972

      SSTriggerVirTablePseudoColRequest* pRequest = &(pReq->virTablePseudoColReq);

1,139,634✔

3973

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRequest->uid));

2,279,268✔

3974

      TAOS_CHECK_EXIT(decodeColsArray(&decoder, &pRequest->cids));

1,139,634✔

3975

      break;

1,139,634✔

3977

    case STRIGGER_PULL_OTABLE_INFO: {

364,313✔

3978

      SSTriggerOrigTableInfoRequest* pRequest = &(pReq->origTableInfoReq);

364,313✔

3979

      int32_t size = 0;

366,524✔

3980

      TAOS_CHECK_EXIT(tDecodeI32(&decoder, &size));

364,313✔

3981

      pRequest->cols = taosArrayInit(size, sizeof(OTableInfo));

364,313✔

3982

      if (pRequest->cols == NULL) {

366,524✔

3987

      for (int32_t i = 0; i < size; ++i) {

976,926✔

3988

        OTableInfo* oInfo = taosArrayReserve(pRequest->cols, 1);

610,402✔

3989

        if (oInfo == NULL) {

610,402✔

3994

        TAOS_CHECK_RETURN(tDecodeCStrTo(&decoder, oInfo->refTableName));

610,402✔

3995

        TAOS_CHECK_RETURN(tDecodeCStrTo(&decoder, oInfo->refColName));

610,402✔

3997

      break;

366,524✔

3999

    default: {

45,174✔

4000

      uError("unknown pull type %d", type);

45,174✔

4006

  tEndDecode(&decoder);

214,628,326✔

4008

_exit:

214,604,134✔

4009

  tDecoderClear(&decoder);

214,616,533✔

4010

  return code;

214,534,804✔

4013

static int32_t tSerializeSTriggerCalcParam(SEncoder* pEncoder, SArray* pParams, bool ignoreNotificationInfo, bool full) {

302,804,888✔

4014

  int32_t size = full ? taosArrayGetSize(pParams) : 0;

302,804,888✔

4015

  int32_t code = 0;

302,804,392✔

4016

  int32_t lino = 0;

302,804,392✔

4017

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, size));

302,810,375✔

4018

  for (int32_t i = 0; i < size; ++i) {

2,147,483,647✔

4019

    SSTriggerCalcParam* param = taosArrayGet(pParams, i);

2,147,483,647✔

4020

    if (param == NULL) {

2,147,483,647✔

4023

    int64_t plainFieldSize = offsetof(SSTriggerCalcParam, notifyType);

2,147,483,647✔

4024

    if (pEncoder->data) {

2,147,483,647✔

4025

      TAOS_MEMCPY(pEncoder->data + pEncoder->pos, param, plainFieldSize);

2,147,483,647✔

4027

    pEncoder->pos += plainFieldSize;

2,147,483,647✔

4029

    if (!ignoreNotificationInfo) {

2,147,483,647✔

4030

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, param->notifyType));

1,300,692,892✔

4031

      uint64_t len = (param->extraNotifyContent != NULL) ? strlen(param->extraNotifyContent) + 1 : 0;

650,343,274✔

4032

      TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, (uint8_t*)param->extraNotifyContent, len));

1,273,713,091✔

4035

_exit:

288,667,873✔

4036

  return code;

288,667,873✔

4039

void tDestroySSTriggerCalcParam(void* ptr) {

2,147,483,647✔

4040

  SSTriggerCalcParam* pParam = ptr;

2,147,483,647✔

4041

  if (pParam && pParam->extraNotifyContent != NULL) {

2,147,483,647✔

4042

    taosMemoryFreeClear(pParam->extraNotifyContent);

775,659✔

4044

  if (pParam && pParam->resultNotifyContent != NULL) {

2,147,483,647✔

4047

2,147,483,647✔

4049

void tDestroySStreamGroupValue(void* ptr) {

118,265,145✔

4050

  SStreamGroupValue* pValue = ptr;

118,265,145✔

4051

  if ((pValue != NULL) && (IS_VAR_DATA_TYPE(pValue->data.type) || pValue->data.type == TSDB_DATA_TYPE_DECIMAL)) {

118,265,145✔

4052

    taosMemoryFreeClear(pValue->data.pData);

106,466,975✔

4053

    pValue->data.nData = 0;

106,466,185✔

4055

118,264,355✔

4057

static int32_t tDeserializeSTriggerCalcParam(SDecoder* pDecoder, SArray**ppParams, bool ignoreNotificationInfo) {

151,397,614✔

4058

  int32_t size = 0, code = 0, lino = 0;

151,397,614✔

4059

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &size));

151,399,781✔

4060

  *ppParams = taosArrayInit(size, sizeof(SSTriggerCalcParam));

151,399,781✔

4061

  if (*ppParams == NULL) {

151,395,412✔

4064

  for (int32_t i = 0; i < size; ++i) {

2,147,483,647✔

4065

    SSTriggerCalcParam* param = taosArrayReserve(*ppParams, 1);

2,147,483,647✔

4066

    if (param == NULL) {

2,147,483,647✔

4069

    int64_t plainFieldSize = offsetof(SSTriggerCalcParam, notifyType);

2,147,483,647✔

4070

    TAOS_MEMCPY(param, pDecoder->data + pDecoder->pos, plainFieldSize);

2,147,483,647✔

4071

    pDecoder->pos += plainFieldSize;

2,147,483,647✔

4073

    if (!ignoreNotificationInfo) {

2,147,483,647✔

4074

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &param->notifyType));

598,784,320✔

4075

      uint64_t len = 0;

299,366,702✔

4076

      TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&param->extraNotifyContent, &len));

598,737,744✔

4080

_exit:

151,712,288✔

4081

  return code;

151,401,248✔

4084

static int32_t tSerializeStriggerGroupColVals(SEncoder* pEncoder, SArray* pGroupColVals, int32_t vgId) {

307,950,827✔

4085

  int32_t code = TSDB_CODE_SUCCESS;

307,950,827✔

4086

  int32_t lino = 0;

307,950,827✔

4088

  int32_t size = taosArrayGetSize(pGroupColVals);

307,950,827✔

4089

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, size));

307,957,014✔

4090

  for (int32_t i = 0; i < size; ++i) {

525,495,139✔

4091

    SStreamGroupValue* pValue = taosArrayGet(pGroupColVals, i);

217,533,792✔

4092

    if (pValue == NULL) {

217,536,519✔

4095

    TAOS_CHECK_EXIT(tEncodeBool(pEncoder, pValue->isNull));

217,536,519✔

4096

    if (pValue->isNull) {

217,537,594✔

4099

    TAOS_CHECK_EXIT(tEncodeBool(pEncoder, pValue->isTbname));

217,536,140✔

4100

    if (pValue->isTbname) {

217,540,702✔

4101

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pValue->uid));

200,452,638✔

4102

      if (vgId != -1) { pValue->vgId = vgId; }

100,223,511✔

4103

      TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pValue->vgId));

200,458,623✔

4105

    TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pValue->data.type));

435,076,014✔

4106

    if (IS_VAR_DATA_TYPE(pValue->data.type)) {

217,526,730✔

4107

      TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, pValue->data.pData, pValue->data.nData));

398,171,557✔

4109

      TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pValue->data.val));

36,900,711✔

4113

_exit:

307,961,347✔

4114

  return code;

307,961,347✔

4117

static int32_t tDeserializeStriggerGroupColVals(SDecoder* pDecoder, SArray** ppGroupColVals) {

153,955,924✔

4118

  int32_t code = TSDB_CODE_SUCCESS;

153,955,924✔

4119

  int32_t lino = 0;

153,955,924✔

4120

  int32_t size = 0;

153,955,924✔

4122

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &size));

153,969,858✔

4123

  taosArrayClearEx(*ppGroupColVals, tDestroySStreamGroupValue);

153,969,858✔

4124

  if (size > 0) {

153,964,792✔

4125

    if (*ppGroupColVals == NULL) {

68,137,748✔

4126

      *ppGroupColVals = taosArrayInit(size, sizeof(SStreamGroupValue));

65,572,842✔

4127

      if (*ppGroupColVals == NULL) {

65,572,842✔

4131

      TAOS_CHECK_EXIT(taosArrayEnsureCap(*ppGroupColVals, size));

2,564,906✔

4134

  for (int32_t i = 0; i < size; ++i) {

262,725,918✔

4135

    SStreamGroupValue* pValue = taosArrayReserve(*ppGroupColVals, 1);

108,764,727✔

4136

    if (pValue == NULL) {

108,764,727✔

4139

    TAOS_CHECK_EXIT(tDecodeBool(pDecoder, &pValue->isNull));

108,764,727✔

4140

    if (pValue->isNull) {

108,763,259✔

4143

    TAOS_CHECK_EXIT(tDecodeBool(pDecoder, &pValue->isTbname));

108,763,259✔

4144

    if (pValue->isTbname) {

108,761,022✔

4145

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pValue->uid));

100,209,154✔

4146

      TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pValue->vgId));

100,209,154✔

4148

    TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pValue->data.type));

217,524,280✔

4149

    if (IS_VAR_DATA_TYPE(pValue->data.type)) {

208,297,837✔

4150

      uint64_t len = 0;

99,533,902✔

4151

      TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pValue->data.pData, &len));

199,066,248✔

4152

      pValue->data.nData = len;

99,533,138✔

4154

      TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pValue->data.val));

18,455,822✔

4157

_exit:

153,961,191✔

4158

  return code;

153,961,191✔

4161

int32_t tSerializeSStreamGroupInfo(void* buf, int32_t bufLen, const SStreamGroupInfo* gInfo, int32_t vgId) {

5,122,537✔

4162

  SEncoder encoder = {0};

5,122,537✔

4163

  int32_t  code = TSDB_CODE_SUCCESS;

5,124,740✔

4164

  int32_t  lino = 0;

5,124,740✔

4165

  int32_t  tlen = 0;

5,124,740✔

4167

  tEncoderInit(&encoder, buf, bufLen);

5,124,740✔

4168

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

5,126,194✔

4170

  TAOS_CHECK_EXIT(tSerializeStriggerGroupColVals(&encoder, gInfo->gInfo, vgId));

5,126,194✔

4172

  tEndEncode(&encoder);

5,119,521✔

4174

_exit:

5,126,021✔

4175

  if (code != TSDB_CODE_SUCCESS) {

5,123,994✔

4178

    tlen = encoder.pos;

5,123,994✔

4180

  tEncoderClear(&encoder);

5,123,994✔

4181

  return tlen;

5,119,521✔

4184

int32_t tDeserializeSStreamGroupInfo(void* buf, int32_t bufLen, SStreamGroupInfo* gInfo) {

2,564,906✔

4185

  SDecoder decoder = {0};

2,564,906✔

4186

  int32_t  code = TSDB_CODE_SUCCESS;

2,564,906✔

4187

  int32_t  lino = 0;

2,564,906✔

4188

  int32_t  size = 0;

2,564,906✔

4190

  tDecoderInit(&decoder, buf, bufLen);

2,564,906✔

4191

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

2,564,906✔

4193

  TAOS_CHECK_EXIT(tDeserializeStriggerGroupColVals(&decoder, &gInfo->gInfo));

2,564,906✔

4195

  tEndDecode(&decoder);

2,564,906✔

4197

_exit:

2,564,906✔

4198

  tDecoderClear(&decoder);

2,564,906✔

4199

  return code;

2,564,906✔

4202

int32_t tSerializeSTriggerCalcRequest(void* buf, int32_t bufLen, const SSTriggerCalcRequest* pReq) {

128,489,202✔

4203

  SEncoder encoder = {0};

128,489,202✔

4204

  int32_t  code = TSDB_CODE_SUCCESS;

128,489,202✔

4205

  int32_t  lino = 0;

128,489,202✔

4206

  int32_t  tlen = 0;

128,489,202✔

4208

  tEncoderInit(&encoder, buf, bufLen);

128,489,202✔

4209

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

128,484,909✔

4211

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->streamId));

256,966,298✔

4212

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->runnerTaskId));

256,965,744✔

4213

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->sessionId));

256,966,508✔

4214

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, pReq->triggerType));

256,965,630✔

4215

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->gid));

256,961,999✔

4217

  TAOS_CHECK_EXIT(tSerializeSTriggerCalcParam(&encoder, pReq->params, false, true));

128,482,487✔

4218

  TAOS_CHECK_EXIT(tSerializeStriggerGroupColVals(&encoder, pReq->groupColVals, -1));

128,478,846✔

4219

  TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->createTable));

256,971,794✔

4221

  tEndEncode(&encoder);

128,484,800✔

4223

_exit:

128,480,490✔

4224

  if (code != TSDB_CODE_SUCCESS) {

128,481,922✔

4227

    tlen = encoder.pos;

128,481,922✔

4229

  tEncoderClear(&encoder);

128,481,922✔

4230

  return tlen;

128,483,922✔

4233

int32_t tDeserializeSTriggerCalcRequest(void* buf, int32_t bufLen, SSTriggerCalcRequest* pReq) {

64,242,014✔

4234

  SDecoder decoder = {0};

64,242,014✔

4235

  int32_t  code = TSDB_CODE_SUCCESS;

64,242,014✔

4236

  int32_t  lino = 0;

64,242,014✔

4238

  tDecoderInit(&decoder, buf, bufLen);

64,242,014✔

4239

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

64,237,610✔

4241

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->streamId));

128,484,028✔

4242

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->runnerTaskId));

128,484,028✔

4243

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->sessionId));

128,481,129✔

4244

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pReq->triggerType));

128,478,230✔

4245

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->gid));

128,479,697✔

4247

  TAOS_CHECK_EXIT(tDeserializeSTriggerCalcParam(&decoder, &pReq->params, false));

64,240,582✔

4248

  TAOS_CHECK_EXIT(tDeserializeStriggerGroupColVals(&decoder, &pReq->groupColVals));

64,238,380✔

4249

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->createTable));

128,476,098✔

4251

  tEndDecode(&decoder);

64,240,582✔

4253

_exit:

64,238,380✔

4254

  tDecoderClear(&decoder);

64,240,582✔

4255

  return code;

64,238,380✔

4258

void tDestroySTriggerCalcRequest(SSTriggerCalcRequest* pReq) {

84,377,481✔

4259

  if (pReq != NULL) {

84,377,481✔

4260

    if (pReq->params != NULL) {

84,377,481✔

4261

      taosArrayDestroyEx(pReq->params, tDestroySSTriggerCalcParam);

65,879,190✔

4262

      pReq->params = NULL;

65,879,190✔

4264

    if (pReq->groupColVals != NULL) {

84,377,481✔

4265

      taosArrayDestroyEx(pReq->groupColVals, tDestroySStreamGroupValue);

4,985,279✔

4266

      pReq->groupColVals = NULL;

4,985,279✔

4268

    blockDataDestroy(pReq->pOutBlock);

84,376,027✔

4270

84,376,049✔

4272

int32_t tSerializeSTriggerDropTableRequest(void* buf, int32_t bufLen, const SSTriggerDropRequest* pReq) {

7,408✔

4273

  SEncoder encoder = {0};

7,408✔

4274

  int32_t  code = TSDB_CODE_SUCCESS;

7,408✔

4275

  int32_t  lino = 0;

7,408✔

4276

  int32_t  tlen = 0;

7,408✔

4278

  tEncoderInit(&encoder, buf, bufLen);

7,408✔

4279

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

7,408✔

4281

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->streamId));

14,816✔

4282

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->runnerTaskId));

14,816✔

4283

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->sessionId));

14,816✔

4284

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->gid));

14,816✔

4286

  TAOS_CHECK_EXIT(tSerializeStriggerGroupColVals(&encoder, pReq->groupColVals, -1));

7,408✔

4288

  tEndEncode(&encoder);

7,408✔

4290

_exit:

7,408✔

4291

  if (code != TSDB_CODE_SUCCESS) {

7,408✔

4294

    tlen = encoder.pos;

7,408✔

4296

  tEncoderClear(&encoder);

7,408✔

4297

  return tlen;

7,408✔

4300

int32_t tDeserializeSTriggerDropTableRequest(void* buf, int32_t bufLen, SSTriggerDropRequest* pReq) {

3,704✔

4301

  SDecoder decoder = {0};

3,704✔

4302

  int32_t  code = TSDB_CODE_SUCCESS;

3,704✔

4303

  int32_t  lino = 0;

3,704✔

4305

  tDecoderInit(&decoder, buf, bufLen);

3,704✔

4306

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

3,704✔

4308

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->streamId));

7,408✔

4309

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->runnerTaskId));

7,408✔

4310

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->sessionId));

7,408✔

4311

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->gid));

7,408✔

4313

  TAOS_CHECK_EXIT(tDeserializeStriggerGroupColVals(&decoder, &pReq->groupColVals));

3,704✔

4315

  tEndDecode(&decoder);

3,704✔

4317

_exit:

3,704✔

4318

  tDecoderClear(&decoder);

3,704✔

4319

  return code;

3,704✔

4322

void tDestroySSTriggerDropRequest(SSTriggerDropRequest* pReq) {

3,704✔

4323

  if (pReq != NULL) {

3,704✔

4324

    if (pReq->groupColVals != NULL) {

3,704✔

4325

      taosArrayDestroyEx(pReq->groupColVals, tDestroySStreamGroupValue);

3,704✔

4326

      pReq->groupColVals = NULL;

3,704✔

4329

3,704✔

4331

int32_t tSerializeSTriggerCtrlRequest(void* buf, int32_t bufLen, const SSTriggerCtrlRequest* pReq) {

257,655,840✔

4332

  SEncoder encoder = {0};

257,655,840✔

4333

  int32_t  code = TSDB_CODE_SUCCESS;

257,655,840✔

4334

  int32_t  lino = 0;

257,655,840✔

4335

  int32_t  tlen = 0;

257,655,840✔

4337

  tEncoderInit(&encoder, buf, bufLen);

257,655,840✔

4338

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

257,655,840✔

4340

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, pReq->type));

515,311,680✔

4341

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->streamId));

515,311,680✔

4342

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->taskId));

515,311,680✔

4343

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->sessionId));

515,311,680✔

4345

  tEndEncode(&encoder);

257,655,840✔

4347

_exit:

257,655,840✔

4348

  if (code != TSDB_CODE_SUCCESS) {

257,655,840✔

4351

    tlen = encoder.pos;

257,655,840✔

4353

  tEncoderClear(&encoder);

257,655,840✔

4354

  return tlen;

257,655,840✔

4357

int32_t tDeserializeSTriggerCtrlRequest(void* buf, int32_t bufLen, SSTriggerCtrlRequest* pReq) {

386,016,137✔

4358

  SDecoder decoder = {0};

386,016,137✔

4359

  int32_t  code = TSDB_CODE_SUCCESS;

386,027,074✔

4360

  int32_t  lino = 0;

386,027,074✔

4362

  tDecoderInit(&decoder, buf, bufLen);

386,027,074✔

4363

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

385,958,812✔

4365

  int32_t type = 0;

386,212,889✔

4366

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &type));

386,174,947✔

4367

  pReq->type = type;

386,174,947✔

4368

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->streamId));

772,476,287✔

4369

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->taskId));

772,564,681✔

4370

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->sessionId));

772,585,896✔

4372

  tEndDecode(&decoder);

386,294,162✔

4374

_exit:

386,263,192✔

4375

  tDecoderClear(&decoder);

386,275,595✔

4376

  return code;

386,191,781✔

4379

int32_t tSerializeStRtFuncInfo(SEncoder* pEncoder, const SStreamRuntimeFuncInfo* pInfo, bool full) {

174,329,494✔

4380

  int32_t code = 0, lino = 0;

174,329,494✔

4381

  TAOS_CHECK_EXIT(tSerializeSTriggerCalcParam(pEncoder, pInfo->pStreamPesudoFuncVals, true, full));

174,329,494✔

4382

  TAOS_CHECK_EXIT(tSerializeStriggerGroupColVals(pEncoder, pInfo->pStreamPartColVals, -1));

174,335,413✔

4383

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pInfo->curWindow.skey));

348,675,024✔

4384

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pInfo->curWindow.ekey));

348,669,743✔

4385

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pInfo->groupId));

348,655,063✔

4386

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pInfo->curIdx));

348,644,817✔

4387

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pInfo->sessionId));

348,647,903✔

4388

  TAOS_CHECK_EXIT(tEncodeBool(pEncoder, pInfo->withExternalWindow));

174,329,518✔

4389

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pInfo->triggerType));

348,659,525✔

4390

_exit:

174,327,889✔

4391

  return code;

174,327,889✔

4394

int32_t tDeserializeStRtFuncInfo(SDecoder* pDecoder, SStreamRuntimeFuncInfo* pInfo) {

87,159,234✔

4395

  int32_t code = 0, lino = 0;

87,159,234✔

4396

  int32_t size = 0;

87,159,234✔

4397

  TAOS_CHECK_EXIT(tDeserializeSTriggerCalcParam(pDecoder, &pInfo->pStreamPesudoFuncVals, true));

87,159,234✔

4398

  TAOS_CHECK_EXIT(tDeserializeStriggerGroupColVals(pDecoder, &pInfo->pStreamPartColVals));

87,159,234✔

4399

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pInfo->curWindow.skey));

174,311,233✔

4400

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pInfo->curWindow.ekey));

174,312,599✔

4401

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pInfo->groupId));

174,316,233✔

4402

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pInfo->curIdx));

174,318,468✔

4403

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pInfo->sessionId));

174,314,785✔

4404

  TAOS_CHECK_EXIT(tDecodeBool(pDecoder, &pInfo->withExternalWindow));

87,155,551✔

4405

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pInfo->triggerType));

174,314,832✔

4406

_exit:

87,156,997✔

4407

  return code;

87,156,997✔

4410

void tDestroyStRtFuncInfo(SStreamRuntimeFuncInfo* pInfo){

125,816,876✔

4411

  if (pInfo == NULL) return;

125,816,876✔

4412

  if (pInfo->pStreamPesudoFuncVals != NULL) {

125,816,876✔

4413

    taosArrayDestroyEx(pInfo->pStreamPesudoFuncVals, tDestroySSTriggerCalcParam);

89,465,943✔

4414

    pInfo->pStreamPesudoFuncVals = NULL;

89,465,943✔

4416

  if (pInfo->pStreamPartColVals != NULL) {

125,818,308✔

4417

    taosArrayDestroyEx(pInfo->pStreamPartColVals, tDestroySStreamGroupValue);

64,527,744✔

4418

    pInfo->pStreamPartColVals = NULL;

64,525,506✔

4422

int32_t tSerializeSStreamMsgVTableInfo(void* buf, int32_t bufLen, const SStreamMsgVTableInfo* pRsp){

3,900,366✔

4423

  SEncoder encoder = {0};

3,900,366✔

4424

  int32_t  code = TSDB_CODE_SUCCESS;

3,900,366✔

4425

  int32_t  lino = 0;

3,900,366✔

4426

  int32_t  tlen = 0;

3,900,366✔

4428

  tEncoderInit(&encoder, buf, bufLen);

3,900,366✔

4429

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

3,900,366✔

4431

  int32_t size = taosArrayGetSize(pRsp->infos);

3,900,366✔

4432

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, size));

3,900,366✔

4433

  for (int32_t i = 0; i < size; ++i) {

9,712,090✔

4434

    VTableInfo* info = taosArrayGet(pRsp->infos, i);

5,810,286✔

4435

    if (info == NULL) {

5,808,861✔

4438

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, info->gId));

11,620,585✔

4439

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, info->uid));

11,623,448✔

4440

    TAOS_CHECK_EXIT(tEncodeSColRefWrapper(&encoder, &info->cols));

5,811,724✔

4443

  tEndEncode(&encoder);

3,901,804✔

4445

_exit:

3,900,366✔

4446

  if (code != TSDB_CODE_SUCCESS) {

3,900,366✔

4449

    tlen = encoder.pos;

3,900,366✔

4451

  tEncoderClear(&encoder);

3,900,366✔

4452

  return tlen;

3,900,366✔

4455

int32_t tDeserializeSStreamMsgVTableInfo(void* buf, int32_t bufLen, SStreamMsgVTableInfo *vTableInfo){

1,950,183✔

4456

  SDecoder decoder = {0};

1,950,183✔

4457

  int32_t  code = TSDB_CODE_SUCCESS;

1,950,183✔

4458

  int32_t  lino = 0;

1,950,183✔

4459

  int32_t  size = 0;

1,950,183✔

4461

  tDecoderInit(&decoder, buf, bufLen);

1,950,183✔

4462

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

1,950,183✔

4464

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &size));

1,950,183✔

4465

  vTableInfo->infos = taosArrayInit(size, sizeof(VTableInfo));

1,950,183✔

4466

  if (vTableInfo->infos == NULL) {

1,950,183✔

4469

  for (int32_t i = 0; i < size; ++i) {

4,856,045✔

4470

    VTableInfo* info = taosArrayReserve(vTableInfo->infos, 1);

2,905,862✔

4471

    if (info == NULL) {

2,905,862✔

4474

    TAOS_CHECK_EXIT(tDecodeI64(&decoder, &info->gId));

5,811,724✔

4475

    TAOS_CHECK_EXIT(tDecodeI64(&decoder, &info->uid));

5,811,724✔

4476

    TAOS_CHECK_EXIT(tDecodeSColRefWrapperEx(&decoder, &info->cols, false));

2,905,862✔

4479

  tEndDecode(&decoder);

1,950,183✔

4481

_exit:

1,950,183✔

4482

  tDecoderClear(&decoder);

1,950,183✔

4483

  return code;

1,950,183✔

4487

void tDestroyVTableInfo(void *ptr) {

5,811,724✔

4488

  if (NULL == ptr) {

5,811,724✔

4491

  VTableInfo* pTable = (VTableInfo*)ptr;

5,811,724✔

4492

  taosMemoryFree(pTable->cols.pColRef);

5,811,724✔

4495

void tDestroySStreamMsgVTableInfo(SStreamMsgVTableInfo *ptr) {

106,727,229✔

4496

  if (ptr == NULL) return;

106,727,229✔

4497

  taosArrayDestroyEx(ptr->infos, tDestroyVTableInfo);

106,727,229✔

4498

  ptr->infos = NULL;

106,729,434✔

4501

int32_t tSerializeSStreamTsResponse(void* buf, int32_t bufLen, const SStreamTsResponse* pRsp) {

6,275,051✔

4502

  SEncoder encoder = {0};

6,275,051✔

4503

  int32_t  code = TSDB_CODE_SUCCESS;

6,283,114✔

4504

  int32_t  lino = 0;

6,283,114✔

4505

  int32_t  tlen = 0;

6,283,114✔

4507

  tEncoderInit(&encoder, buf, bufLen);

6,283,114✔

4508

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

6,287,576✔

4510

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pRsp->ver));

12,561,961✔

4511

  int32_t size = taosArrayGetSize(pRsp->tsInfo);

6,278,815✔

4512

  TAOS_CHECK_EXIT(tEncodeI32(&encoder, size));

6,280,248✔

4513

  for (int32_t i = 0; i < size; ++i) {

10,751,431✔

4514

    STsInfo* tsInfo = taosArrayGet(pRsp->tsInfo, i);

4,469,741✔

4515

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, tsInfo->gId));

8,943,806✔

4516

    TAOS_CHECK_EXIT(tEncodeI64(&encoder, tsInfo->ts));

8,941,593✔

4519

  tEndEncode(&encoder);

6,281,690✔

4521

_exit:

6,283,134✔

4522

  if (code != TSDB_CODE_SUCCESS) {

6,283,134✔

4525

    tlen = encoder.pos;

6,283,134✔

4527

  tEncoderClear(&encoder);

6,283,134✔

4528

  return tlen;

6,281,009✔

4531

int32_t tDeserializeSStreamTsResponse(void* buf, int32_t bufLen, void *pBlock) {

3,146,703✔

4532

  SDecoder decoder = {0};

3,146,703✔

4533

  int32_t  code = TSDB_CODE_SUCCESS;

3,146,703✔

4534

  int32_t  lino = 0;

3,146,703✔

4535

  SSDataBlock *pResBlock = pBlock;

3,146,703✔

4537

  tDecoderInit(&decoder, buf, bufLen);

3,146,703✔

4538

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

3,146,703✔

4540

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, (int64_t*)&pResBlock->info.id.groupId));

6,293,406✔

4541

  int32_t numOfCols = 2;

3,146,703✔

4542

  if (pResBlock->pDataBlock == NULL) {

3,146,703✔

4543

    pResBlock->pDataBlock = taosArrayInit_s(sizeof(SColumnInfoData), numOfCols);

3,146,703✔

4544

    if (pResBlock->pDataBlock == NULL) {

3,146,703✔

4547

    for (int32_t i = 0; i< numOfCols; ++i) {

9,440,109✔

4548

      SColumnInfoData *pColInfoData = taosArrayGet(pResBlock->pDataBlock, i);

6,293,406✔

4549

      if (pColInfoData == NULL) {

6,293,406✔

4552

      pColInfoData->info.type = TSDB_DATA_TYPE_BIGINT;

6,293,406✔

4553

      pColInfoData->info.bytes = sizeof(int64_t);

6,293,406✔

4556

  int32_t numOfRows = 0;

3,146,703✔

4557

  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &numOfRows));

3,146,703✔

4558

  TAOS_CHECK_EXIT(blockDataEnsureCapacity(pResBlock, numOfRows));

3,146,703✔

4559

  for (int32_t i = 0; i < numOfRows; ++i) {

5,387,202✔

4560

    for (int32_t j = 0; j < numOfCols; ++j) {

6,721,497✔

4561

      SColumnInfoData *pColInfoData = taosArrayGet(pResBlock->pDataBlock, j);

4,480,998✔

4562

      if (pColInfoData == NULL) {

4,480,998✔

4565

      int64_t value = 0;

4,480,998✔

4566

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &value));

4,480,998✔

4567

      colDataSetInt64(pColInfoData, i, &value);

4,480,998✔

4571

  pResBlock->info.dataLoad = 1;

3,146,703✔

4572

  pResBlock->info.rows = numOfRows;

3,146,703✔

4574

  tEndDecode(&decoder);

3,146,703✔

4576

_exit:

3,146,703✔

4577

  tDecoderClear(&decoder);

3,146,703✔

4578

  return code;

3,146,703✔

4581

static int32_t encodeData(SEncoder* encoder, void* pBlock, SSHashObj* indexHash) {

11,445,932✔

4582

  int32_t code = TSDB_CODE_SUCCESS;

11,445,932✔

4583

  int32_t lino = 0;

11,445,932✔

4584

  int32_t len = 0;

11,445,932✔

4585

  if (encoder->data == NULL){

11,445,932✔

4586

    len = blockGetEncodeSize(pBlock);

5,725,172✔

4588

    len = blockEncode(pBlock, (char*)(encoder->data + encoder->pos), encoder->size - encoder->pos, blockDataGetNumOfCols(pBlock));

5,725,172✔

4589

    if (len < 0) {

5,722,278✔

4593

  encoder->pos += len;

11,447,450✔

4595

  if (indexHash == NULL) {

11,447,450✔

4596

    goto _exit;

3,998,146✔

4599

  uint32_t pos = encoder->pos;

7,449,304✔

4600

  encoder->pos += sizeof(uint32_t); // reserve space for tables

7,449,304✔

4601

  int32_t tables = 0;

7,449,304✔

4603

  void*   pe = NULL;

7,449,304✔

4604

  int32_t iter = 0;

7,449,304✔

4605

  while ((pe = tSimpleHashIterate(indexHash, pe, &iter)) != NULL) {

26,731,924✔

4606

    SStreamWalDataSlice* pInfo = (SStreamWalDataSlice*)pe;

19,284,828✔

4607

    if (pInfo->gId == -1){

19,284,828✔

4608

      continue;

10,001,262✔

4610

    int64_t uid = *(int64_t*)(tSimpleHashGetKey(pe, NULL));

9,283,566✔

4611

    TAOS_CHECK_EXIT(tEncodeI64(encoder, uid));

9,283,566✔

4612

    TAOS_CHECK_EXIT(tEncodeU64(encoder, pInfo->gId));

18,565,686✔

4613

    TAOS_CHECK_EXIT(tEncodeI32(encoder, pInfo->startRowIdx));

18,565,686✔

4614

    TAOS_CHECK_EXIT(tEncodeI32(encoder, pInfo->numRows));

18,564,924✔

4615

    tables++;

9,281,358✔

4617

  uint32_t tmpPos = encoder->pos;

7,444,120✔

4618

  encoder->pos = pos;

7,448,536✔

4619

  TAOS_CHECK_EXIT(tEncodeI32(encoder, tables));

7,448,536✔

4620

  encoder->pos = tmpPos;

7,448,536✔

4621

_exit:

11,446,682✔

4622

  return code;

11,446,682✔

4625

int32_t tSerializeSStreamWalDataResponse(void* buf, int32_t bufLen, SSTriggerWalNewRsp* rsp, SSHashObj* indexHash) {

9,253,916✔

4626

  SEncoder encoder = {0};

9,253,916✔

4627

  int32_t  code = TSDB_CODE_SUCCESS;

9,256,110✔

4628

  int32_t  lino = 0;

9,256,110✔

4629

  int32_t  tlen = 0;

9,256,110✔

4631

  tEncoderInit(&encoder, buf, bufLen);

9,256,110✔

4632

  TAOS_CHECK_EXIT(tStartEncode(&encoder));

9,256,110✔

4634

  if (rsp->dataBlock != NULL && ((SSDataBlock*)rsp->dataBlock)->info.rows > 0) {

9,253,902✔

4635

    TAOS_CHECK_EXIT(tEncodeI8(&encoder, 1)); // has real data

7,450,750✔

4636

    TAOS_CHECK_EXIT(encodeData(&encoder, rsp->dataBlock, indexHash));

7,450,750✔

4638

    TAOS_CHECK_EXIT(tEncodeI8(&encoder, 0));  // no real data

1,803,166✔

4641

  if (rsp->metaBlock != NULL && ((SSDataBlock*)rsp->metaBlock)->info.rows > 0) {

9,251,702✔

4642

    TAOS_CHECK_EXIT(tEncodeI8(&encoder, 1)); // has metada

3,836,600✔

4643

    TAOS_CHECK_EXIT(encodeData(&encoder, rsp->metaBlock, NULL));

3,836,600✔

4645

    TAOS_CHECK_EXIT(tEncodeI8(&encoder, 0));  // no meta data

5,413,661✔

4648

  if (rsp->deleteBlock != NULL && ((SSDataBlock*)rsp->deleteBlock)->info.rows > 0) {

9,251,007✔

4649

    TAOS_CHECK_EXIT(tEncodeI8(&encoder, 1)); // has deletedata

153,392✔

4650

    TAOS_CHECK_EXIT(encodeData(&encoder, rsp->deleteBlock, NULL));

153,392✔

4652

    TAOS_CHECK_EXIT(tEncodeI8(&encoder, 0));  // no delete data

9,100,515✔

4655

  if (rsp->dropBlock != NULL && ((SSDataBlock*)rsp->dropBlock)->info.rows > 0) {

9,253,907✔

4656

    TAOS_CHECK_EXIT(tEncodeI8(&encoder, 1)); // has drop table data

7,408✔

4657

    TAOS_CHECK_EXIT(encodeData(&encoder, rsp->dropBlock, NULL));

7,408✔

4659

    TAOS_CHECK_EXIT(tEncodeI8(&encoder, 0));  // no drop table data

9,248,702✔

4662

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, rsp->ver));

18,504,904✔

4663

  TAOS_CHECK_EXIT(tEncodeI64(&encoder, rsp->verTime));

18,502,710✔

4664

  tEndEncode(&encoder);

9,253,916✔

4666

_exit:

9,253,916✔

4667

  if (code != TSDB_CODE_SUCCESS) {

9,253,916✔

4670

    tlen = encoder.pos;

9,253,916✔

4672

  tEncoderClear(&encoder);

9,253,916✔

4673

  return tlen;

9,253,624✔

4676

int32_t tDeserializeSStreamWalDataResponse(void* buf, int32_t bufLen, SSTriggerWalNewRsp* pRsp, SArray* pSlices){

4,626,943✔

4677

  SDecoder     decoder = {0};

4,626,943✔

4678

  int32_t      code = TSDB_CODE_SUCCESS;

4,626,943✔

4679

  int32_t      lino = 0;

4,626,943✔

4680

  SSDataBlock* pBlock = NULL;

4,626,943✔

4682

  tDecoderInit(&decoder, buf, bufLen);

4,626,943✔

4683

  TAOS_CHECK_EXIT(tStartDecode(&decoder));

4,626,943✔

4686

  int8_t hasData = false;

4,629,151✔

4687

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &hasData));

4,629,151✔

4688

  pBlock = pRsp->dataBlock;

4,629,151✔

4689

  if (hasData) {

4,629,151✔

4690

    TAOS_CHECK_EXIT(pBlock != NULL ? TSDB_CODE_SUCCESS : TSDB_CODE_INVALID_PARA);

3,725,375✔

4691

    const char* pEndPos = NULL;

3,725,375✔

4692

    TAOS_CHECK_EXIT(blockDecode(pBlock, (char*)decoder.data + decoder.pos, &pEndPos));

3,725,375✔

4693

    decoder.pos = (uint8_t*)pEndPos - decoder.data;

3,725,375✔

4695

    int32_t nSlices = 0;

3,725,375✔

4696

    TAOS_CHECK_EXIT(tDecodeI32(&decoder, &nSlices));

3,725,375✔

4697

    TAOS_CHECK_EXIT(taosArrayEnsureCap(pSlices, nSlices));

3,725,375✔

4698

    taosArrayClear(pSlices);

3,725,375✔

4699

    int64_t  uid = 0;

3,725,375✔

4700

    uint64_t gid = 0;

3,725,375✔

4701

    int32_t  startIdx = 0;

3,725,375✔

4702

    int32_t  numRows = 0;

3,725,375✔

4703

    for (int32_t i = 0; i < nSlices; i++) {

8,367,158✔

4704

      TAOS_CHECK_EXIT(tDecodeI64(&decoder, &uid));

4,641,783✔

4705

      TAOS_CHECK_EXIT(tDecodeU64(&decoder, &gid));

4,641,783✔

4706

      TAOS_CHECK_EXIT(tDecodeI32(&decoder, &startIdx));

4,641,783✔

4707

      TAOS_CHECK_EXIT(tDecodeI32(&decoder, &numRows));

4,641,783✔

4708

      int32_t endIdx = startIdx + numRows;

4,641,783✔

4709

      int64_t value[3] = {gid, uid, (int64_t)startIdx << 32 | endIdx};

4,641,783✔

4710

      void*   px = taosArrayPush(pSlices, value);

4,641,783✔

4711

      if (px == NULL) {

4,641,783✔

4716

  } else if (pBlock != NULL) {

903,776✔

4717

    blockDataEmpty(pBlock);

47,508✔

4718

    taosArrayClear(pSlices);

47,508✔

4721

  int8_t hasMeta = false;

4,629,151✔

4722

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &hasMeta));

4,629,151✔

4723

  pBlock = pRsp->metaBlock;

4,629,151✔

4724

  if (hasMeta) {

4,629,151✔

4725

    TAOS_CHECK_EXIT(pBlock != NULL ? TSDB_CODE_SUCCESS : TSDB_CODE_INVALID_PARA);

1,920,493✔

4726

    const char* pEndPos = NULL;

1,920,493✔

4727

    TAOS_CHECK_EXIT(blockDecode(pBlock, (char*)decoder.data + decoder.pos, &pEndPos));

1,920,493✔

4728

    decoder.pos = (uint8_t*)pEndPos - decoder.data;

1,920,493✔

4729

  } else if (pBlock != NULL) {

2,708,658✔

4730

    blockDataEmpty(pBlock);

65,822✔

4733

  int8_t hasDel = false;

4,629,151✔

4734

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &hasDel));

4,629,151✔

4735

  pBlock = pRsp->deleteBlock;

4,629,151✔

4736

  if (hasDel) {

4,629,151✔

4737

    TAOS_CHECK_EXIT(pBlock != NULL ? TSDB_CODE_SUCCESS : TSDB_CODE_INVALID_PARA);

76,696✔

4738

    const char* pEndPos = NULL;

76,696✔

4739

    TAOS_CHECK_EXIT(blockDecode(pBlock, (char*)decoder.data + decoder.pos, &pEndPos));

76,696✔

4740

    decoder.pos = (uint8_t*)pEndPos - decoder.data;

76,696✔

4741

  } else if (pBlock != NULL) {

4,552,455✔

4742

    blockDataEmpty(pBlock);

1,909,619✔

4745

  int8_t hasDrop = false;

4,629,151✔

4746

  TAOS_CHECK_EXIT(tDecodeI8(&decoder, &hasDrop));

4,629,151✔

4747

  pBlock = pRsp->dropBlock;

4,629,151✔

4748

  if (hasDrop) {

4,629,151✔

4749

    TAOS_CHECK_EXIT(pBlock != NULL ? TSDB_CODE_SUCCESS : TSDB_CODE_INVALID_PARA);

3,704✔

4750

    const char* pEndPos = NULL;

3,704✔

4751

    TAOS_CHECK_EXIT(blockDecode(pBlock, (char*)decoder.data + decoder.pos, &pEndPos));

3,704✔

4752

    decoder.pos = (uint8_t*)pEndPos - decoder.data;

3,704✔

4753

  } else if (pBlock != NULL) {

4,625,447✔

4754

    blockDataEmpty(pBlock);

1,982,611✔

4756

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRsp->ver));

9,258,302✔

4757

  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pRsp->verTime));

9,258,302✔

4759

  tEndDecode(&decoder);

4,629,151✔

4761

_exit:

4,629,151✔

4762

  if (code != TSDB_CODE_SUCCESS) {

4,629,151✔

4765

  tDecoderClear(&decoder);

4,629,151✔

4766

  return code;

4,629,151✔

taosdata / TDengine / #4838

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #4838

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