taosdata / TDengine / #3621

coverage: 2.037% (-61.5%) from 63.573%
#3621

push

travis-ci

web-flow 
Merge pull request #29874 from taosdata/merge/mainto3.0

merge: from main to 3.0 branch

4357 of 287032 branches covered (1.52%)

Branch coverage included in aggregate %.

0 of 174 new or added lines in 18 files covered. (0.0%)

213359 existing lines in 469 files now uncovered.

7260 of 283369 relevant lines covered (2.56%)

23737.72 hits per line

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

0.0
/source/libs/executor/src/streamcountwindowoperator.c
1 
/*
2 
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3 
 *
4 
 * This program is free software: you can use, redistribute, and/or modify
5 
 * it under the terms of the GNU Affero General Public License, version 3
6 
 * or later ("AGPL"), as published by the Free Software Foundation.
7 
 *
8 
 * This program is distributed in the hope that it will be useful, but WITHOUT
9 
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 
 * FITNESS FOR A PARTICULAR PURPOSE.
11 
 *
12 
 * You should have received a copy of the GNU Affero General Public License
13 
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
14 
 */
15 
#include "executorInt.h"
16 
#include "function.h"
17 
#include "functionMgt.h"
18 
#include "operator.h"
19 
#include "querytask.h"
20 
#include "streamexecutorInt.h"
21 
#include "tchecksum.h"
22 
#include "tcommon.h"
23 
#include "tdatablock.h"
24 
#include "tglobal.h"
25 
#include "tlog.h"
26 
#include "ttime.h"
27 









28



#define STREAM_COUNT_OP_STATE_NAME      "StreamCountHistoryState"










29



#define STREAM_COUNT_OP_CHECKPOINT_NAME "StreamCountOperator_Checkpoint"










30












31



typedef struct SCountWindowInfo {










32



  SResultWindowInfo winInfo;










33



  COUNT_TYPE*       pWindowCount;










34



} SCountWindowInfo;










35












36



typedef enum {










37



  NONE_WINDOW = 0,










38



  CREATE_NEW_WINDOW,










39



  MOVE_NEXT_WINDOW,










40



} BuffOp;










41



typedef struct SBuffInfo {










42



  bool             rebuildWindow;










43



  BuffOp           winBuffOp;










44



  SStreamStateCur* pCur;










45



} SBuffInfo;










46













UNCOV

47



void destroyStreamCountAggOperatorInfo(void* param) {




×








UNCOV

48



  if (param == NULL) {





×







49



    return;




×







50



  }











UNCOV

51



  SStreamCountAggOperatorInfo* pInfo = (SStreamCountAggOperatorInfo*)param;




×








UNCOV

52



  cleanupBasicInfo(&pInfo->binfo);




×








UNCOV

53



  if (pInfo->pOperator) {





×








UNCOV

54



    cleanupResultInfoInStream(pInfo->pOperator->pTaskInfo, pInfo->streamAggSup.pState, &pInfo->pOperator->exprSupp,




×







55



                              &pInfo->groupResInfo);











UNCOV

56



    pInfo->pOperator = NULL;




×







57



  }










58













UNCOV

59



  destroyStreamBasicInfo(&pInfo->basic);




×








UNCOV

60



  destroyStreamAggSupporter(&pInfo->streamAggSup);




×








UNCOV

61



  cleanupExprSupp(&pInfo->scalarSupp);




×








UNCOV

62



  clearGroupResInfo(&pInfo->groupResInfo);




×








UNCOV

63



  taosArrayDestroyP(pInfo->pUpdated, destroyFlusedPos);




×








UNCOV

64



  pInfo->pUpdated = NULL;




×







65













UNCOV

66



  colDataDestroy(&pInfo->twAggSup.timeWindowData);




×








UNCOV

67



  blockDataDestroy(pInfo->pDelRes);




×








UNCOV

68



  tSimpleHashCleanup(pInfo->pStUpdated);




×








UNCOV

69



  tSimpleHashCleanup(pInfo->pStDeleted);




×








UNCOV

70



  cleanupGroupResInfo(&pInfo->groupResInfo);




×







71













UNCOV

72



  taosArrayDestroy(pInfo->historyWins);




×








UNCOV

73



  blockDataDestroy(pInfo->pCheckpointRes);




×







74













UNCOV

75



  tSimpleHashCleanup(pInfo->pPkDeleted);




×







76













UNCOV

77



  taosMemoryFreeClear(param);





×







78



}










79













UNCOV

80



bool isSlidingCountWindow(SStreamAggSupporter* pAggSup) { return pAggSup->windowCount != pAggSup->windowSliding; }




×







81













UNCOV

82



int32_t setCountOutputBuf(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, SCountWindowInfo* pCurWin,




×







83



                          SBuffInfo* pBuffInfo, int32_t* pWinCode) {











UNCOV

84



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

85



  int32_t lino = 0;




×








UNCOV

86



  int32_t size = pAggSup->resultRowSize;




×








UNCOV

87



  pCurWin->winInfo.sessionWin.groupId = groupId;




×








UNCOV

88



  pCurWin->winInfo.sessionWin.win.skey = ts;




×








UNCOV

89



  pCurWin->winInfo.sessionWin.win.ekey = ts;




×







90













UNCOV

91



  if (isSlidingCountWindow(pAggSup)) {





×








UNCOV

92



    if (pBuffInfo->winBuffOp == CREATE_NEW_WINDOW) {





×







93



      code =











UNCOV

94



          pAggSup->stateStore.streamStateCountWinAdd(pAggSup->pState, &pCurWin->winInfo.sessionWin,




×








UNCOV

95



                                                     pAggSup->windowCount, (void**)&pCurWin->winInfo.pStatePos, &size);




×








UNCOV

96



      QUERY_CHECK_CODE(code, lino, _end);





×







97













UNCOV

98



      *pWinCode = TSDB_CODE_FAILED;




×








UNCOV

99



    } else if (pBuffInfo->winBuffOp == MOVE_NEXT_WINDOW) {





×








UNCOV

100



      QUERY_CHECK_NULL(pBuffInfo->pCur, code, lino, _end, terrno);





×








UNCOV

101



      pAggSup->stateStore.streamStateCurNext(pAggSup->pState, pBuffInfo->pCur);




×








UNCOV

102



      *pWinCode = pAggSup->stateStore.streamStateSessionGetKVByCur(pBuffInfo->pCur, &pCurWin->winInfo.sessionWin,




×








UNCOV

103



                                                                   (void**)&pCurWin->winInfo.pStatePos, &size);




×








UNCOV

104



      if (*pWinCode == TSDB_CODE_FAILED) {





×








UNCOV

105



        code = pAggSup->stateStore.streamStateCountWinAdd(pAggSup->pState, &pCurWin->winInfo.sessionWin,




×








UNCOV

106



                                                          pAggSup->windowCount, (void**)&pCurWin->winInfo.pStatePos,




×







107



                                                          &size);











UNCOV

108



        QUERY_CHECK_CODE(code, lino, _end);





×







109



      } else {











UNCOV

110



        reuseOutputBuf(pAggSup->pState, pCurWin->winInfo.pStatePos, &pAggSup->stateStore);




×







111



      }










112



    } else {











UNCOV

113



      pBuffInfo->pCur = pAggSup->stateStore.streamStateCountSeekKeyPrev(pAggSup->pState, &pCurWin->winInfo.sessionWin,




×








UNCOV

114



                                                                        pAggSup->windowCount);




×








UNCOV

115



      *pWinCode = pAggSup->stateStore.streamStateSessionGetKVByCur(pBuffInfo->pCur, &pCurWin->winInfo.sessionWin,




×








UNCOV

116



                                                                   (void**)&pCurWin->winInfo.pStatePos, &size);




×








UNCOV

117



      if (*pWinCode == TSDB_CODE_FAILED) {





×








UNCOV

118



        code = pAggSup->stateStore.streamStateCountWinAdd(pAggSup->pState, &pCurWin->winInfo.sessionWin,




×








UNCOV

119



                                                          pAggSup->windowCount, (void**)&pCurWin->winInfo.pStatePos,




×







120



                                                          &size);











UNCOV

121



        QUERY_CHECK_CODE(code, lino, _end);





×







122



      } else {











UNCOV

123



        reuseOutputBuf(pAggSup->pState, pCurWin->winInfo.pStatePos, &pAggSup->stateStore);




×







124



      }










125



    }











UNCOV

126



    if (ts < pCurWin->winInfo.sessionWin.win.ekey) {





×








UNCOV

127



      pBuffInfo->rebuildWindow = true;




×







128



    }










129



  } else {











UNCOV

130



    code = pAggSup->stateStore.streamStateCountWinAddIfNotExist(pAggSup->pState, &pCurWin->winInfo.sessionWin,




×








UNCOV

131



                                                                pAggSup->windowCount,




×








UNCOV

132



                                                                (void**)&pCurWin->winInfo.pStatePos, &size, pWinCode);




×








UNCOV

133



    QUERY_CHECK_CODE(code, lino, _end);





×







134



  }










135













UNCOV

136



  if (*pWinCode == TSDB_CODE_SUCCESS) {





×








UNCOV

137



    pCurWin->winInfo.isOutput = true;




×







138



  }











UNCOV

139



  pCurWin->pWindowCount =




×








UNCOV

140



      (COUNT_TYPE*)((char*)pCurWin->winInfo.pStatePos->pRowBuff + (pAggSup->resultRowSize - sizeof(COUNT_TYPE)));




×







141













UNCOV

142



  if (*pCurWin->pWindowCount == pAggSup->windowCount) {





×








UNCOV

143



    pBuffInfo->rebuildWindow = true;




×







144



  }










145













UNCOV

146



_end:




×








UNCOV

147



  if (code != TSDB_CODE_SUCCESS) {





×







148



    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







149



  }











UNCOV

150



  return code;




×







151



}










152













UNCOV

153



static void removeCountResult(SSHashObj* pHashMap, SSHashObj* pResMap, SSessionKey* pKey) {




×








UNCOV

154



  SSessionKey key = {0};




×








UNCOV

155



  getSessionHashKey(pKey, &key);




×








UNCOV

156



  int32_t code = tSimpleHashRemove(pHashMap, &key, sizeof(SSessionKey));




×








UNCOV

157



  if (code != TSDB_CODE_SUCCESS) {





×








UNCOV

158



    qInfo("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));





×







159



  }










160













UNCOV

161



  code = tSimpleHashRemove(pResMap, &key, sizeof(SSessionKey));




×








UNCOV

162



  if (code != TSDB_CODE_SUCCESS) {





×








UNCOV

163



    qInfo("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));





×







164



  }











UNCOV

165



}




×







166













UNCOV

167



static int32_t updateCountWindowInfo(SStreamAggSupporter* pAggSup, SCountWindowInfo* pWinInfo, TSKEY* pTs,




×







168



                                     int32_t start, int32_t rows, int32_t maxRows, SSHashObj* pStUpdated,










169



                                     SSHashObj* pStDeleted, bool* pRebuild, int32_t* pWinRows) {











UNCOV

170



  int32_t     code = TSDB_CODE_SUCCESS;




×








UNCOV

171



  int32_t     lino = 0;




×








UNCOV

172



  SSessionKey sWinKey = pWinInfo->winInfo.sessionWin;




×








UNCOV

173



  int32_t     num = 0;




×








UNCOV

174



  for (int32_t i = start; i < rows; i++) {





×








UNCOV

175



    if (pTs[i] < pWinInfo->winInfo.sessionWin.win.ekey) {





×








UNCOV

176



      num++;




×







177



    } else {











UNCOV

178



      break;




×







179



    }










180



  }











UNCOV

181



  int32_t maxNum = TMIN(maxRows - *pWinInfo->pWindowCount, rows - start);




×








UNCOV

182



  if (num > maxNum) {





×







183



    *pRebuild = true;




×







184



  }











UNCOV

185



  *pWinInfo->pWindowCount += maxNum;




×








UNCOV

186



  bool needDelState = false;




×








UNCOV

187



  if (pWinInfo->winInfo.sessionWin.win.skey > pTs[start]) {





×








UNCOV

188



    needDelState = true;




×








UNCOV

189



    if (pStDeleted && pWinInfo->winInfo.isOutput) {





×








UNCOV

190



      code = saveDeleteRes(pStDeleted, pWinInfo->winInfo.sessionWin);




×








UNCOV

191



      QUERY_CHECK_CODE(code, lino, _end);





×







192



    }










193













UNCOV

194



    pWinInfo->winInfo.sessionWin.win.skey = pTs[start];




×







195



  }










196













UNCOV

197



  if (pWinInfo->winInfo.sessionWin.win.ekey < pTs[maxNum + start - 1]) {





×








UNCOV

198



    needDelState = true;




×








UNCOV

199



    pWinInfo->winInfo.sessionWin.win.ekey = pTs[maxNum + start - 1];




×







200



  }










201













UNCOV

202



  if (needDelState) {





×








UNCOV

203



    memcpy(pWinInfo->winInfo.pStatePos->pKey, &pWinInfo->winInfo.sessionWin, sizeof(SSessionKey));




×








UNCOV

204



    removeCountResult(pStUpdated, pAggSup->pResultRows, &sWinKey);




×








UNCOV

205



    if (pWinInfo->winInfo.pStatePos->needFree) {





×








UNCOV

206



      pAggSup->stateStore.streamStateSessionDel(pAggSup->pState, &sWinKey);




×







207



    }










208



  }










209













UNCOV

210



  (*pWinRows) = maxNum;




×







211













UNCOV

212



_end:




×








UNCOV

213



  if (code != TSDB_CODE_SUCCESS) {





×







214



    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







215



  }











UNCOV

216



  return code;




×







217



}










218













UNCOV

219



void getCountWinRange(SStreamAggSupporter* pAggSup, const SSessionKey* pKey, EStreamType mode, SSessionKey* pDelRange) {




×








UNCOV

220



  *pDelRange = *pKey;




×








UNCOV

221



  SStreamStateCur* pCur = NULL;




×








UNCOV

222



  if (isSlidingCountWindow(pAggSup)) {





×








UNCOV

223



    pCur = pAggSup->stateStore.streamStateCountSeekKeyPrev(pAggSup->pState, pKey, pAggSup->windowCount);




×







224



  } else {











UNCOV

225



    pCur = pAggSup->stateStore.streamStateSessionSeekKeyCurrentNext(pAggSup->pState, pKey);




×







226



  }











UNCOV

227



  SSessionKey tmpKey = {.groupId = pKey->groupId, .win.ekey = INT64_MIN, .win.skey = INT64_MIN};




×








UNCOV

228



  int32_t     code = pAggSup->stateStore.streamStateSessionGetKVByCur(pCur, &tmpKey, NULL, 0);




×








UNCOV

229



  if (code != TSDB_CODE_SUCCESS) {





×







230



    pAggSup->stateStore.streamStateFreeCur(pCur);




×







231



    return;




×







232



  }











UNCOV

233



  pDelRange->win = tmpKey.win;




×








UNCOV

234



  while (mode == STREAM_DELETE_DATA || mode == STREAM_PARTITION_DELETE_DATA) {





×








UNCOV

235



    pAggSup->stateStore.streamStateCurNext(pAggSup->pState, pCur);




×








UNCOV

236



    code = pAggSup->stateStore.streamStateSessionGetKVByCur(pCur, &tmpKey, NULL, 0);




×








UNCOV

237



    if (code != TSDB_CODE_SUCCESS) {





×








UNCOV

238



      break;




×







239



    }











UNCOV

240



    pDelRange->win.ekey = TMAX(pDelRange->win.ekey, tmpKey.win.ekey);




×







241



  }











UNCOV

242



  pAggSup->stateStore.streamStateFreeCur(pCur);




×







243



}










244













UNCOV

245



static void destroySBuffInfo(SStreamAggSupporter* pAggSup, SBuffInfo* pBuffInfo) {




×








UNCOV

246



  pAggSup->stateStore.streamStateFreeCur(pBuffInfo->pCur);




×








UNCOV

247



}




×







248













UNCOV

249



bool inCountCalSlidingWindow(SStreamAggSupporter* pAggSup, STimeWindow* pWin, TSKEY sKey, TSKEY eKey) {




×








UNCOV

250



  if (pAggSup->windowCount == pAggSup->windowSliding) {





×








UNCOV

251



    return true;




×







252



  }











UNCOV

253



  if (sKey <= pWin->skey && pWin->ekey <= eKey) {





×








UNCOV

254



    return true;




×







255



  }











UNCOV

256



  return false;




×







257



}










258













UNCOV

259



bool inCountSlidingWindow(SStreamAggSupporter* pAggSup, STimeWindow* pWin, SDataBlockInfo* pBlockInfo) {




×








UNCOV

260



  return inCountCalSlidingWindow(pAggSup, pWin, pBlockInfo->calWin.skey, pBlockInfo->calWin.ekey);




×







261



}










262













UNCOV

263



static void doStreamCountAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBlock, SSHashObj* pStUpdated,




×







264



                                 SSHashObj* pStDeleted) {











UNCOV

265



  int32_t                      code = TSDB_CODE_SUCCESS;




×








UNCOV

266



  int32_t                      lino = 0;




×








UNCOV

267



  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

268



  SStreamCountAggOperatorInfo* pInfo = pOperator->info;




×








UNCOV

269



  int32_t                      numOfOutput = pOperator->exprSupp.numOfExprs;




×








UNCOV

270



  uint64_t                     groupId = pSDataBlock->info.id.groupId;




×








UNCOV

271



  SResultRow*                  pResult = NULL;




×








UNCOV

272



  int32_t                      rows = pSDataBlock->info.rows;




×








UNCOV

273



  int32_t                      winRows = 0;




×








UNCOV

274



  SStreamAggSupporter*         pAggSup = &pInfo->streamAggSup;




×








UNCOV

275



  SBuffInfo                    buffInfo = {.rebuildWindow = false, .winBuffOp = NONE_WINDOW, .pCur = NULL};




×







276













UNCOV

277



  pInfo->dataVersion = TMAX(pInfo->dataVersion, pSDataBlock->info.version);




×








UNCOV

278



  pAggSup->winRange = pTaskInfo->streamInfo.fillHistoryWindow;




×








UNCOV

279



  if (pAggSup->winRange.ekey <= 0) {





×







280



    pAggSup->winRange.ekey = INT64_MAX;




×







281



  }










282













UNCOV

283



  SColumnInfoData* pStartTsCol = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);




×








UNCOV

284



  if (!pStartTsCol) {





×







285



    code = TSDB_CODE_FAILED;




×







286



    QUERY_CHECK_CODE(code, lino, _end);





×







287



  }











UNCOV

288



  TSKEY* startTsCols = (int64_t*)pStartTsCol->pData;




×








UNCOV

289



  code = blockDataEnsureCapacity(pAggSup->pScanBlock, rows * 2);




×








UNCOV

290



  QUERY_CHECK_CODE(code, lino, _end);





×







291













UNCOV

292



  SStreamStateCur* pCur = NULL;




×








UNCOV

293



  COUNT_TYPE       slidingRows = 0;




×







294













UNCOV

295



  for (int32_t i = 0; i < rows;) {





×








UNCOV

296



    if (pInfo->ignoreExpiredData &&





×








UNCOV

297



        checkExpiredData(&pInfo->streamAggSup.stateStore, pInfo->streamAggSup.pUpdateInfo, &pInfo->twAggSup,




×








UNCOV

298



                         pSDataBlock->info.id.uid, startTsCols[i], NULL, 0)) {




×








UNCOV

299



      i++;




×








UNCOV

300



      continue;




×







301



    }











UNCOV

302



    SCountWindowInfo curWin = {0};




×








UNCOV

303



    int32_t          winCode = TSDB_CODE_SUCCESS;




×








UNCOV

304



    buffInfo.rebuildWindow = false;




×








UNCOV

305



    code = setCountOutputBuf(pAggSup, startTsCols[i], groupId, &curWin, &buffInfo, &winCode);




×








UNCOV

306



    QUERY_CHECK_CODE(code, lino, _end);





×







307













UNCOV

308



    if (winCode != TSDB_CODE_SUCCESS &&





×








UNCOV

309



        BIT_FLAG_TEST_MASK(pTaskInfo->streamInfo.eventTypes, SNOTIFY_EVENT_WINDOW_OPEN)) {





×







310



      code = addCountAggNotifyEvent(SNOTIFY_EVENT_WINDOW_OPEN, &curWin.winInfo.sessionWin, &pInfo->basic.notifyEventSup,




×







311



                                    pTaskInfo->streamInfo.pNotifyEventStat);










312



      QUERY_CHECK_CODE(code, lino, _end);





×







313



    }










314













UNCOV

315



    if (!inCountSlidingWindow(pAggSup, &curWin.winInfo.sessionWin.win, &pSDataBlock->info)) {





×








UNCOV

316



      buffInfo.winBuffOp = MOVE_NEXT_WINDOW;




×








UNCOV

317



      continue;




×







318



    }











UNCOV

319



    setSessionWinOutputInfo(pStUpdated, &curWin.winInfo);




×








UNCOV

320



    slidingRows = *curWin.pWindowCount;




×








UNCOV

321



    if (!buffInfo.rebuildWindow) {





×








UNCOV

322



      code = updateCountWindowInfo(pAggSup, &curWin, startTsCols, i, rows, pAggSup->windowCount, pStUpdated, pStDeleted,




×







323



                                   &buffInfo.rebuildWindow, &winRows);











UNCOV

324



      QUERY_CHECK_CODE(code, lino, _end);





×







325



    }











UNCOV

326



    if (buffInfo.rebuildWindow) {





×








UNCOV

327



      SSessionKey range = {0};




×








UNCOV

328



      if (isSlidingCountWindow(pAggSup)) {





×








UNCOV

329



        curWin.winInfo.sessionWin.win.skey = startTsCols[i];




×








UNCOV

330



        curWin.winInfo.sessionWin.win.ekey = startTsCols[i];




×







331



      }











UNCOV

332



      getCountWinRange(pAggSup, &curWin.winInfo.sessionWin, STREAM_DELETE_DATA, &range);




×








UNCOV

333



      range.win.skey = TMIN(startTsCols[i], range.win.skey);




×








UNCOV

334



      range.win.ekey = TMAX(startTsCols[rows - 1], range.win.ekey);




×








UNCOV

335



      uint64_t uid = 0;




×







336



      code =











UNCOV

337



          appendDataToSpecialBlock(pAggSup->pScanBlock, &range.win.skey, &range.win.ekey, &uid, &range.groupId, NULL);




×








UNCOV

338



      QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

339



      break;




×







340



    }











UNCOV

341



    code = doOneWindowAggImpl(&pInfo->twAggSup.timeWindowData, &curWin.winInfo, &pResult, i, winRows, rows, numOfOutput,




×







342



                              pOperator, 0);











UNCOV

343



    QUERY_CHECK_CODE(code, lino, _end);





×







344













UNCOV

345



    code = saveSessionOutputBuf(pAggSup, &curWin.winInfo);




×








UNCOV

346



    QUERY_CHECK_CODE(code, lino, _end);





×







347













UNCOV

348



    if (pInfo->destHasPrimaryKey && curWin.winInfo.isOutput && IS_NORMAL_COUNT_OP(pOperator)) {





×








UNCOV

349



      code = saveDeleteRes(pInfo->pPkDeleted, curWin.winInfo.sessionWin);




×








UNCOV

350



      QUERY_CHECK_CODE(code, lino, _end);





×







351



    }










352













UNCOV

353



    if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE && pStUpdated) {





×








UNCOV

354



      code = saveResult(curWin.winInfo, pStUpdated);




×








UNCOV

355



      QUERY_CHECK_CODE(code, lino, _end);





×







356



    }











UNCOV

357



    if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_WINDOW_CLOSE) {





×








UNCOV

358



      curWin.winInfo.pStatePos->beUpdated = true;




×








UNCOV

359



      SSessionKey key = {0};




×








UNCOV

360



      getSessionHashKey(&curWin.winInfo.sessionWin, &key);




×







361



      code =











UNCOV

362



          tSimpleHashPut(pAggSup->pResultRows, &key, sizeof(SSessionKey), &curWin.winInfo, sizeof(SResultWindowInfo));




×








UNCOV

363



      QUERY_CHECK_CODE(code, lino, _end);





×







364



    }










365













UNCOV

366



    if (isSlidingCountWindow(pAggSup)) {





×








UNCOV

367



      if (slidingRows <= pAggSup->windowSliding) {





×








UNCOV

368



        if (slidingRows + winRows > pAggSup->windowSliding) {





×








UNCOV

369



          buffInfo.winBuffOp = CREATE_NEW_WINDOW;




×








UNCOV

370



          winRows = pAggSup->windowSliding - slidingRows;




×







371



        }










372



      } else {











UNCOV

373



        buffInfo.winBuffOp = MOVE_NEXT_WINDOW;




×








UNCOV

374



        winRows = 0;




×







375



      }










376



    }











UNCOV

377



    i += winRows;




×







378



  }










379













UNCOV

380



_end:




×








UNCOV

381



  if (code != TSDB_CODE_SUCCESS) {





×







382



    qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));





×







383



  }











UNCOV

384



  destroySBuffInfo(pAggSup, &buffInfo);




×








UNCOV

385



}




×







386













UNCOV

387



static int32_t buildCountResult(SOperatorInfo* pOperator, SSDataBlock** ppRes) {




×








UNCOV

388



  int32_t                      code = TSDB_CODE_SUCCESS;




×








UNCOV

389



  int32_t                      lino = 0;




×








UNCOV

390



  SStreamCountAggOperatorInfo* pInfo = pOperator->info;




×








UNCOV

391



  SStreamAggSupporter*         pAggSup = &pInfo->streamAggSup;




×








UNCOV

392



  SOptrBasicInfo*              pBInfo = &pInfo->binfo;




×








UNCOV

393



  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

394



  SStreamNotifyEventSupp*      pNotifySup = &pInfo->basic.notifyEventSup;




×








UNCOV

395



  STaskNotifyEventStat*        pNotifyEventStat = pTaskInfo->streamInfo.pNotifyEventStat;




×








UNCOV

396



  bool                         addNotifyEvent = false;




×








UNCOV

397



  addNotifyEvent = BIT_FLAG_TEST_MASK(pTaskInfo->streamInfo.eventTypes, SNOTIFY_EVENT_WINDOW_CLOSE);




×








UNCOV

398



  doBuildDeleteDataBlock(pOperator, pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);




×








UNCOV

399



  if (pInfo->pDelRes->info.rows > 0) {





×








UNCOV

400



    printDataBlock(pInfo->pDelRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));




×








UNCOV

401



    if (addNotifyEvent) {





×







402



      code = addAggDeleteNotifyEvent(pInfo->pDelRes, pNotifySup, pNotifyEventStat);




×







403



      QUERY_CHECK_CODE(code, lino, _end);





×







404



    }











UNCOV

405



    (*ppRes) = pInfo->pDelRes;




×








UNCOV

406



    return code;




×







407



  }










408













UNCOV

409



  doBuildSessionResult(pOperator, pAggSup->pState, &pInfo->groupResInfo, pBInfo->pRes,





×







410



                       addNotifyEvent ? pNotifySup->pSessionKeys : NULL);











UNCOV

411



  if (pBInfo->pRes->info.rows > 0) {





×








UNCOV

412



    printDataBlock(pBInfo->pRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));




×








UNCOV

413



    if (addNotifyEvent) {





×







414



      code = addAggResultNotifyEvent(pBInfo->pRes, pNotifySup->pSessionKeys, pTaskInfo->streamInfo.notifyResultSchema,




×







415



                                     pNotifySup, pNotifyEventStat);










416



      QUERY_CHECK_CODE(code, lino, _end);





×







417



    }











UNCOV

418



    (*ppRes) = pBInfo->pRes;




×








UNCOV

419



    return code;




×







420



  }










421













UNCOV

422



  code = buildNotifyEventBlock(pTaskInfo, pNotifySup, pNotifyEventStat);




×








UNCOV

423



  QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

424



  if (pNotifySup->pEventBlock && pNotifySup->pEventBlock->info.rows > 0) {





×







425



    printDataBlock(pNotifySup->pEventBlock, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));




×







426



    (*ppRes) = pNotifySup->pEventBlock;




×







427



    return code;




×







428



  }










429













UNCOV

430



  code = removeOutdatedNotifyEvents(&pInfo->twAggSup, pNotifySup, pNotifyEventStat);




×








UNCOV

431



  QUERY_CHECK_CODE(code, lino, _end);





×







432













UNCOV

433



_end:




×








UNCOV

434



  if (code != TSDB_CODE_SUCCESS) {





×







435



    qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));





×







436



  }











UNCOV

437



  (*ppRes) = NULL;




×








UNCOV

438



  return code;




×







439



}










440












441



int32_t doStreamCountEncodeOpState(void** buf, int32_t len, SOperatorInfo* pOperator, bool isParent) {




×







442



  SStreamCountAggOperatorInfo* pInfo = pOperator->info;




×







443



  if (!pInfo) {





×







444



    return 0;




×







445



  }










446












447



  void* pData = (buf == NULL) ? NULL : *buf;





×







448












449



  // 1.streamAggSup.pResultRows










450



  int32_t tlen = 0;




×







451



  int32_t mapSize = tSimpleHashGetSize(pInfo->streamAggSup.pResultRows);




×







452



  tlen += taosEncodeFixedI32(buf, mapSize);




×







453



  void*   pIte = NULL;




×







454



  size_t  keyLen = 0;




×







455



  int32_t iter = 0;




×







456



  while ((pIte = tSimpleHashIterate(pInfo->streamAggSup.pResultRows, pIte, &iter)) != NULL) {





×







457



    void* key = tSimpleHashGetKey(pIte, &keyLen);




×







458



    tlen += encodeSSessionKey(buf, key);




×







459



    tlen += encodeSResultWindowInfo(buf, pIte, pInfo->streamAggSup.resultRowSize);




×







460



  }










461












462



  // 2.twAggSup










463



  tlen += encodeSTimeWindowAggSupp(buf, &pInfo->twAggSup);




×







464












465



  // 3.dataVersion










466



  tlen += taosEncodeFixedI32(buf, pInfo->dataVersion);





×







467












468



  // 4.basicInfo










469



  tlen += encodeStreamBasicInfo(buf, &pInfo->basic);




×







470












471



  // 5.checksum










472



  if (isParent) {





×







473



    if (buf) {





×







474



      uint32_t cksum = taosCalcChecksum(0, pData, len - sizeof(uint32_t));





×







475



      tlen += taosEncodeFixedU32(buf, cksum);




×







476



    } else {










477



      tlen += sizeof(uint32_t);




×







478



    }










479



  }










480












481



  return tlen;




×







482



}










483












484



int32_t doStreamCountDecodeOpState(void* buf, int32_t len, SOperatorInfo* pOperator, bool isParent) {




×







485



  int32_t                      code = TSDB_CODE_SUCCESS;




×







486



  int32_t                      lino = 0;




×







487



  SStreamCountAggOperatorInfo* pInfo = pOperator->info;




×







488



  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;




×







489



  void*                        pDataEnd = POINTER_SHIFT(buf, len);




×







490



  if (!pInfo) {





×







491



    code = TSDB_CODE_FAILED;




×







492



    QUERY_CHECK_CODE(code, lino, _end);





×







493



  }










494












495



  // 5.checksum










496



  if (isParent) {





×







497



    int32_t dataLen = len - sizeof(uint32_t);




×







498



    void*   pCksum = POINTER_SHIFT(buf, dataLen);




×







499



    if (taosCheckChecksum(buf, dataLen, *(uint32_t*)pCksum) != TSDB_CODE_SUCCESS) {





×







500



      code = TSDB_CODE_FAILED;




×







501



      QUERY_CHECK_CODE(code, lino, _end);





×







502



    }










503



    pDataEnd = pCksum;




×







504



  }










505












506



  // 1.streamAggSup.pResultRows










507



  int32_t mapSize = 0;




×







508



  buf = taosDecodeFixedI32(buf, &mapSize);





×







509



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





×







510



    SSessionKey      key = {0};




×







511



    SCountWindowInfo curWin = {0};




×







512



    int32_t          winCode = TSDB_CODE_SUCCESS;




×







513



    buf = decodeSSessionKey(buf, &key);




×







514



    SBuffInfo buffInfo = {.rebuildWindow = false, .winBuffOp = NONE_WINDOW, .pCur = NULL};




×







515



    code = setCountOutputBuf(&pInfo->streamAggSup, key.win.skey, key.groupId, &curWin, &buffInfo, &winCode);




×







516



    QUERY_CHECK_CODE(code, lino, _end);





×







517












518



    buf = decodeSResultWindowInfo(buf, &curWin.winInfo, pInfo->streamAggSup.resultRowSize);




×







519



    code = tSimpleHashPut(pInfo->streamAggSup.pResultRows, &key, sizeof(SSessionKey), &curWin.winInfo,




×







520



                          sizeof(SResultWindowInfo));










521



    QUERY_CHECK_CODE(code, lino, _end);





×







522



  }










523












524



  // 2.twAggSup










525



  buf = decodeSTimeWindowAggSupp(buf, &pInfo->twAggSup);




×







526












527



  // 3.dataVersion










528



  buf = taosDecodeFixedI64(buf, &pInfo->dataVersion);





×







529












530



  // 4.basicInfo










531



  if (buf < pDataEnd) {





×







532



    code = decodeStreamBasicInfo(&buf, &pInfo->basic);




×







533



    QUERY_CHECK_CODE(code, lino, _end);





×







534



  }










535












536



_end:




×







537



  if (code != TSDB_CODE_SUCCESS) {





×







538



    qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));





×







539



  }










540



  return code;




×







541



}










542













UNCOV

543



void doStreamCountSaveCheckpoint(SOperatorInfo* pOperator) {




×








UNCOV

544



  int32_t                      code = TSDB_CODE_SUCCESS;




×








UNCOV

545



  int32_t                      lino = 0;




×








UNCOV

546



  void*                        pBuf = NULL;




×








UNCOV

547



  SStreamCountAggOperatorInfo* pInfo = pOperator->info;




×








UNCOV

548



  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

549



  if (needSaveStreamOperatorInfo(&pInfo->basic)) {





×







550



    int32_t len = doStreamCountEncodeOpState(NULL, 0, pOperator, true);




×







551



    pBuf = taosMemoryCalloc(1, len);





×







552



    if (!pBuf) {





×







553



      code = terrno;




×







554



      QUERY_CHECK_CODE(code, lino, _end);





×







555



    }










556



    void* pTmpBuf = pBuf;




×







557



    len = doStreamCountEncodeOpState(&pTmpBuf, len, pOperator, true);




×







558



    pInfo->streamAggSup.stateStore.streamStateSaveInfo(pInfo->streamAggSup.pState, STREAM_COUNT_OP_CHECKPOINT_NAME,




×







559



                                                       strlen(STREAM_COUNT_OP_CHECKPOINT_NAME), pBuf, len);










560



    saveStreamOperatorStateComplete(&pInfo->basic);




×







561



  }










562













UNCOV

563



_end:




×








UNCOV

564



  taosMemoryFreeClear(pBuf);





×








UNCOV

565



  if (code != TSDB_CODE_SUCCESS) {





×







566



    qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));





×







567



  }











UNCOV

568



}




×







569













UNCOV

570



void doResetCountWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBlock) {




×








UNCOV

571



  SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);




×








UNCOV

572



  TSKEY*           startDatas = (TSKEY*)pStartTsCol->pData;




×








UNCOV

573



  SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);




×








UNCOV

574



  TSKEY*           endDatas = (TSKEY*)pEndTsCol->pData;




×








UNCOV

575



  SColumnInfoData* pCalStartTsCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);




×








UNCOV

576



  TSKEY*           calStartDatas = (TSKEY*)pStartTsCol->pData;




×








UNCOV

577



  SColumnInfoData* pCalEndTsCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);




×








UNCOV

578



  TSKEY*           calEndDatas = (TSKEY*)pEndTsCol->pData;




×








UNCOV

579



  SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);




×








UNCOV

580



  uint64_t*        gpDatas = (uint64_t*)pGroupCol->pData;




×







581













UNCOV

582



  SRowBuffPos* pPos = NULL;




×








UNCOV

583



  int32_t      size = 0;




×








UNCOV

584



  for (int32_t i = 0; i < pBlock->info.rows; i++) {





×








UNCOV

585



    SSessionKey      key = {.groupId = gpDatas[i], .win.skey = startDatas[i], .win.ekey = endDatas[i]};




×








UNCOV

586



    SStreamStateCur* pCur = NULL;




×








UNCOV

587



    if (isSlidingCountWindow(pAggSup)) {





×








UNCOV

588



      pCur = pAggSup->stateStore.streamStateCountSeekKeyPrev(pAggSup->pState, &key, pAggSup->windowCount);




×







589



    } else {











UNCOV

590



      pCur = pAggSup->stateStore.streamStateSessionSeekKeyCurrentNext(pAggSup->pState, &key);




×







591



    }











UNCOV

592



    while (1) {




×








UNCOV

593



      SSessionKey tmpKey = {.groupId = gpDatas[i], .win.skey = INT64_MIN, .win.ekey = INT64_MIN};




×








UNCOV

594



      int32_t     code = pAggSup->stateStore.streamStateSessionGetKVByCur(pCur, &tmpKey, (void**)&pPos, &size);




×








UNCOV

595



      if (code != TSDB_CODE_SUCCESS || tmpKey.win.skey > endDatas[i]) {





×








UNCOV

596



        pAggSup->stateStore.streamStateFreeCur(pCur);




×








UNCOV

597



        break;




×







598



      }











UNCOV

599



      if (!inCountCalSlidingWindow(pAggSup, &tmpKey.win, calStartDatas[i], calEndDatas[i])) {





×








UNCOV

600



        pAggSup->stateStore.streamStateCurNext(pAggSup->pState, pCur);




×








UNCOV

601



        continue;




×







602



      }











UNCOV

603



      pAggSup->stateStore.streamStateSessionReset(pAggSup->pState, pPos->pRowBuff);




×








UNCOV

604



      pAggSup->stateStore.streamStateCurNext(pAggSup->pState, pCur);




×







605



    }










606



  }











UNCOV

607



}




×







608













UNCOV

609



int32_t doDeleteCountWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBlock, SArray* result) {




×








UNCOV

610



  int32_t          code = TSDB_CODE_SUCCESS;




×








UNCOV

611



  int32_t          lino = 0;




×








UNCOV

612



  SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);




×








UNCOV

613



  TSKEY*           startDatas = (TSKEY*)pStartTsCol->pData;




×








UNCOV

614



  SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);




×








UNCOV

615



  TSKEY*           endDatas = (TSKEY*)pEndTsCol->pData;




×








UNCOV

616



  SColumnInfoData* pCalStartTsCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);




×








UNCOV

617



  TSKEY*           calStartDatas = (TSKEY*)pStartTsCol->pData;




×








UNCOV

618



  SColumnInfoData* pCalEndTsCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);




×








UNCOV

619



  TSKEY*           calEndDatas = (TSKEY*)pEndTsCol->pData;




×








UNCOV

620



  SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);




×








UNCOV

621



  uint64_t*        gpDatas = (uint64_t*)pGroupCol->pData;




×








UNCOV

622



  for (int32_t i = 0; i < pBlock->info.rows; i++) {





×








UNCOV

623



    SSessionKey key = {.win.skey = startDatas[i], .win.ekey = endDatas[i], .groupId = gpDatas[i]};




×








UNCOV

624



    while (1) {




×








UNCOV

625



      SSessionKey curWin = {0};




×








UNCOV

626



      int32_t     winCode = pAggSup->stateStore.streamStateCountGetKeyByRange(pAggSup->pState, &key, &curWin);




×








UNCOV

627



      if (winCode != TSDB_CODE_SUCCESS) {





×








UNCOV

628



        break;




×







629



      }











UNCOV

630



      doDeleteSessionWindow(pAggSup, &curWin);




×








UNCOV

631



      if (result) {





×








UNCOV

632



        code = saveDeleteInfo(result, curWin);




×








UNCOV

633



        QUERY_CHECK_CODE(code, lino, _end);





×







634



      }










635



    }










636



  }










637













UNCOV

638



_end:




×








UNCOV

639



  if (code != TSDB_CODE_SUCCESS) {





×







640



    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







641



  }











UNCOV

642



  return code;




×







643



}










644













UNCOV

645



int32_t deleteCountWinState(SStreamAggSupporter* pAggSup, SSDataBlock* pBlock, SSHashObj* pMapUpdate,




×







646



                            SSHashObj* pMapDelete, SSHashObj* pPkDelete, bool needAdd) {











UNCOV

647



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

648



  int32_t lino = 0;




×








UNCOV

649



  SArray* pWins = taosArrayInit(16, sizeof(SSessionKey));




×








UNCOV

650



  if (!pWins) {





×







651



    code = terrno;




×







652



    QUERY_CHECK_CODE(code, lino, _end);





×







653



  }










654













UNCOV

655



  if (isSlidingCountWindow(pAggSup)) {





×








UNCOV

656



    code = doDeleteCountWindows(pAggSup, pBlock, pWins);




×








UNCOV

657



    QUERY_CHECK_CODE(code, lino, _end);





×







658



  } else {











UNCOV

659



    code = doDeleteTimeWindows(pAggSup, pBlock, pWins);




×








UNCOV

660



    QUERY_CHECK_CODE(code, lino, _end);





×







661



  }











UNCOV

662



  removeSessionResults(pAggSup, pMapUpdate, pWins);




×








UNCOV

663



  code = copyDeleteWindowInfo(pWins, pMapDelete);




×








UNCOV

664



  QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

665



  if (needAdd) {





×







666



    code = copyDeleteWindowInfo(pWins, pPkDelete);




×







667



    QUERY_CHECK_CODE(code, lino, _end);





×







668



  }











UNCOV

669



  taosArrayDestroy(pWins);




×







670













UNCOV

671



_end:




×








UNCOV

672



  if (code != TSDB_CODE_SUCCESS) {





×







673



    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







674



  }











UNCOV

675



  return code;




×







676



}










677













UNCOV

678



static int32_t doStreamCountAggNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {




×








UNCOV

679



  int32_t                      code = TSDB_CODE_SUCCESS;




×








UNCOV

680



  int32_t                      lino = 0;




×








UNCOV

681



  SExprSupp*                   pSup = &pOperator->exprSupp;




×








UNCOV

682



  SStreamCountAggOperatorInfo* pInfo = pOperator->info;




×








UNCOV

683



  SOptrBasicInfo*              pBInfo = &pInfo->binfo;




×








UNCOV

684



  SStreamAggSupporter*         pAggSup = &pInfo->streamAggSup;




×








UNCOV

685



  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

686



  qDebug("stask:%s  %s status: %d", GET_TASKID(pTaskInfo), getStreamOpName(pOperator->operatorType), pOperator->status);





×








UNCOV

687



  if (pOperator->status == OP_EXEC_DONE) {





×







688



    (*ppRes) = NULL;




×







689



    return code;




×








UNCOV

690



  } else if (pOperator->status == OP_RES_TO_RETURN) {





×








UNCOV

691



    SSDataBlock* opRes = NULL;




×








UNCOV

692



    code = buildCountResult(pOperator, &opRes);




×








UNCOV

693



    QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

694



    if (opRes) {





×








UNCOV

695



      (*ppRes) = opRes;




×








UNCOV

696



      return code;




×







697



    }










698













UNCOV

699



    if (pInfo->recvGetAll) {





×







700



      pInfo->recvGetAll = false;




×







701



      resetUnCloseSessionWinInfo(pInfo->streamAggSup.pResultRows);




×







702



    }










703













UNCOV

704



    if (pInfo->reCkBlock) {





×







705



      pInfo->reCkBlock = false;




×







706



      printDataBlock(pInfo->pCheckpointRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));




×







707



      (*ppRes) = pInfo->pCheckpointRes;




×







708



      return code;




×







709



    }










710













UNCOV

711



    setStreamOperatorCompleted(pOperator);




×








UNCOV

712



    (*ppRes) = NULL;




×








UNCOV

713



    return code;




×







714



  }










715













UNCOV

716



  SOperatorInfo* downstream = pOperator->pDownstream[0];




×








UNCOV

717



  if (!pInfo->pUpdated) {





×








UNCOV

718



    pInfo->pUpdated = taosArrayInit(16, sizeof(SResultWindowInfo));




×








UNCOV

719



    QUERY_CHECK_NULL(pInfo->pUpdated, code, lino, _end, terrno);





×







720



  }











UNCOV

721



  if (!pInfo->pStUpdated) {





×








UNCOV

722



    _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);




×








UNCOV

723



    pInfo->pStUpdated = tSimpleHashInit(64, hashFn);




×








UNCOV

724



    QUERY_CHECK_NULL(pInfo->pStUpdated, code, lino, _end, terrno);





×







725



  }











UNCOV

726



  while (1) {




×








UNCOV

727



    SSDataBlock* pBlock = NULL;




×








UNCOV

728



    code = downstream->fpSet.getNextFn(downstream, &pBlock);




×








UNCOV

729



    QUERY_CHECK_CODE(code, lino, _end);





×







730













UNCOV

731



    if (pBlock == NULL) {





×








UNCOV

732



      break;




×







733



    }










734













UNCOV

735



    printSpecDataBlock(pBlock, getStreamOpName(pOperator->operatorType), "recv", GET_TASKID(pTaskInfo));




×








UNCOV

736



    setStreamOperatorState(&pInfo->basic, pBlock->info.type);




×







737













UNCOV

738



    if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT) {





×








UNCOV

739



      bool add = pInfo->destHasPrimaryKey && IS_NORMAL_COUNT_OP(pOperator);





×








UNCOV

740



      code = deleteCountWinState(&pInfo->streamAggSup, pBlock, pInfo->pStUpdated, pInfo->pStDeleted, pInfo->pPkDeleted,




×







741



                                 add);











UNCOV

742



      QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

743



      continue;




×








UNCOV

744



    } else if (pBlock->info.type == STREAM_CLEAR) {





×








UNCOV

745



      doResetCountWindows(&pInfo->streamAggSup, pBlock);




×








UNCOV

746



      continue;




×








UNCOV

747



    } else if (pBlock->info.type == STREAM_GET_ALL) {





×







748



      pInfo->recvGetAll = true;




×







749



      code = getAllSessionWindow(pAggSup->pResultRows, pInfo->pStUpdated);




×







750



      QUERY_CHECK_CODE(code, lino, _end);





×







751



      continue;




×








UNCOV

752



    } else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE) {





×








UNCOV

753



      (*ppRes) = pBlock;




×








UNCOV

754



      return code;




×








UNCOV

755



    } else if (pBlock->info.type == STREAM_CHECKPOINT) {





×








UNCOV

756



      pAggSup->stateStore.streamStateCommit(pAggSup->pState);




×








UNCOV

757



      doStreamCountSaveCheckpoint(pOperator);




×








UNCOV

758



      code = copyDataBlock(pInfo->pCheckpointRes, pBlock);




×








UNCOV

759



      QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

760



      continue;




×







761



    } else {











UNCOV

762



      if (pBlock->info.type != STREAM_NORMAL && pBlock->info.type != STREAM_INVALID) {





×







763



        code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;




×







764



        QUERY_CHECK_CODE(code, lino, _end);





×







765



      }










766



    }










767













UNCOV

768



    if (pInfo->scalarSupp.pExprInfo != NULL) {





×







769



      SExprSupp* pExprSup = &pInfo->scalarSupp;




×







770



      code = projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);




×







771



      QUERY_CHECK_CODE(code, lino, _end);





×







772



    }










773



    // the pDataBlock are always the same one, no need to call this again











UNCOV

774



    code = setInputDataBlock(pSup, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);




×








UNCOV

775



    QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

776



    doStreamCountAggImpl(pOperator, pBlock, pInfo->pStUpdated, pInfo->pStDeleted);




×








UNCOV

777



    pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);




×








UNCOV

778



    pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.watermark);




×







779



  }










780



  // restore the value











UNCOV

781



  pOperator->status = OP_RES_TO_RETURN;




×







782













UNCOV

783



  code = closeSessionWindow(pAggSup->pResultRows, &pInfo->twAggSup, pInfo->pStUpdated);




×








UNCOV

784



  QUERY_CHECK_CODE(code, lino, _end);





×







785













UNCOV

786



  code = copyUpdateResult(&pInfo->pStUpdated, pInfo->pUpdated, sessionKeyCompareAsc);




×








UNCOV

787



  QUERY_CHECK_CODE(code, lino, _end);





×







788













UNCOV

789



  removeSessionDeleteResults(pInfo->pStDeleted, pInfo->pUpdated);




×








UNCOV

790



  initGroupResInfoFromArrayList(&pInfo->groupResInfo, pInfo->pUpdated);




×








UNCOV

791



  pInfo->pUpdated = NULL;




×








UNCOV

792



  code = blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);




×








UNCOV

793



  QUERY_CHECK_CODE(code, lino, _end);





×







794













UNCOV

795



  if (pInfo->destHasPrimaryKey && IS_NORMAL_COUNT_OP(pOperator)) {





×








UNCOV

796



    code = copyDeleteSessionKey(pInfo->pPkDeleted, pInfo->pStDeleted);




×








UNCOV

797



    QUERY_CHECK_CODE(code, lino, _end);





×







798



  }










799













UNCOV

800



  SSDataBlock* opRes = NULL;




×








UNCOV

801



  code = buildCountResult(pOperator, &opRes);




×








UNCOV

802



  QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

803



  if (opRes) {





×








UNCOV

804



    (*ppRes) = opRes;




×








UNCOV

805



    return code;




×







806



  }










807













UNCOV

808



_end:




×








UNCOV

809



  if (code != TSDB_CODE_SUCCESS) {





×







810



    qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));





×







811



    pTaskInfo->code = code;




×







812



    T_LONG_JMP(pTaskInfo->env, code);





×







813



  }











UNCOV

814



  setStreamOperatorCompleted(pOperator);




×








UNCOV

815



  (*ppRes) = NULL;




×








UNCOV

816



  return code;




×







817



}










818













UNCOV

819



void streamCountReleaseState(SOperatorInfo* pOperator) {




×








UNCOV

820



  int32_t                      code = TSDB_CODE_SUCCESS;




×








UNCOV

821



  int32_t                      lino = 0;




×








UNCOV

822



  SStreamEventAggOperatorInfo* pInfo = pOperator->info;




×








UNCOV

823



  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

824



  int32_t                      resSize = sizeof(TSKEY);




×








UNCOV

825



  char*                        pBuff = taosMemoryCalloc(1, resSize);





×








UNCOV

826



  QUERY_CHECK_NULL(pBuff, code, lino, _end, terrno);





×







827













UNCOV

828



  memcpy(pBuff, &pInfo->twAggSup.maxTs, sizeof(TSKEY));




×








UNCOV

829



  qDebug("===stream=== count window operator relase state. ");





×








UNCOV

830



  pInfo->streamAggSup.stateStore.streamStateSaveInfo(pInfo->streamAggSup.pState, STREAM_COUNT_OP_STATE_NAME,




×







831



                                                     strlen(STREAM_COUNT_OP_STATE_NAME), pBuff, resSize);











UNCOV

832



  pInfo->streamAggSup.stateStore.streamStateCommit(pInfo->streamAggSup.pState);




×








UNCOV

833



  taosMemoryFreeClear(pBuff);





×








UNCOV

834



  SOperatorInfo* downstream = pOperator->pDownstream[0];




×








UNCOV

835



  if (downstream->fpSet.releaseStreamStateFn) {





×








UNCOV

836



    downstream->fpSet.releaseStreamStateFn(downstream);




×







837



  }










838



_end:




×








UNCOV

839



  if (code != TSDB_CODE_SUCCESS) {





×







840



    terrno = code;




×







841



    qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));





×







842



  }











UNCOV

843



}




×







844













UNCOV

845



void streamCountReloadState(SOperatorInfo* pOperator) {




×








UNCOV

846



  int32_t                      code = TSDB_CODE_SUCCESS;




×








UNCOV

847



  int32_t                      lino = 0;




×








UNCOV

848



  SStreamCountAggOperatorInfo* pInfo = pOperator->info;




×








UNCOV

849



  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

850



  SStreamAggSupporter*         pAggSup = &pInfo->streamAggSup;




×








UNCOV

851



  int32_t                      size = 0;




×








UNCOV

852



  void*                        pBuf = NULL;




×







853













UNCOV

854



  code = pAggSup->stateStore.streamStateGetInfo(pAggSup->pState, STREAM_COUNT_OP_STATE_NAME,




×







855



                                                strlen(STREAM_COUNT_OP_STATE_NAME), &pBuf, &size);











UNCOV

856



  QUERY_CHECK_CODE(code, lino, _end);





×







857













UNCOV

858



  TSKEY ts = *(TSKEY*)pBuf;




×








UNCOV

859



  pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, ts);




×








UNCOV

860



  taosMemoryFree(pBuf);





×







861













UNCOV

862



  SOperatorInfo* downstream = pOperator->pDownstream[0];




×








UNCOV

863



  if (downstream->fpSet.reloadStreamStateFn) {





×








UNCOV

864



    downstream->fpSet.reloadStreamStateFn(downstream);




×







865



  }











UNCOV

866



  reloadAggSupFromDownStream(downstream, &pInfo->streamAggSup);




×







867













UNCOV

868



_end:




×








UNCOV

869



  if (code != TSDB_CODE_SUCCESS) {





×







870



    terrno = code;




×







871



    qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));





×







872



  }











UNCOV

873



}




×







874













UNCOV

875



int32_t createStreamCountAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo,




×







876



                                         SReadHandle* pHandle, SOperatorInfo** pOptrInfo) {











UNCOV

877



  QRY_PARAM_CHECK(pOptrInfo);





×







878













UNCOV

879



  SCountWinodwPhysiNode*       pCountNode = (SCountWinodwPhysiNode*)pPhyNode;




×








UNCOV

880



  int32_t                      numOfCols = 0;




×








UNCOV

881



  int32_t                      code = TSDB_CODE_SUCCESS;




×








UNCOV

882



  int32_t                      lino = 0;




×








UNCOV

883



  SStreamCountAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamCountAggOperatorInfo));





×








UNCOV

884



  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));





×








UNCOV

885



  if (pInfo == NULL || pOperator == NULL) {





×







886



    code = terrno;




×







887



    QUERY_CHECK_CODE(code, lino, _error);





×







888



  }










889













UNCOV

890



  pOperator->pTaskInfo = pTaskInfo;




×







891













UNCOV

892



  initResultSizeInfo(&pOperator->resultInfo, 4096);




×








UNCOV

893



  if (pCountNode->window.pExprs != NULL) {





×







894



    int32_t    numOfScalar = 0;




×







895



    SExprInfo* pScalarExprInfo = NULL;




×







896



    code = createExprInfo(pCountNode->window.pExprs, NULL, &pScalarExprInfo, &numOfScalar);




×







897



    QUERY_CHECK_CODE(code, lino, _error);





×







898












899



    code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar, &pTaskInfo->storageAPI.functionStore);




×







900



    QUERY_CHECK_CODE(code, lino, _error);





×







901



  }











UNCOV

902



  SExprSupp* pExpSup = &pOperator->exprSupp;




×







903













UNCOV

904



  SSDataBlock* pResBlock = createDataBlockFromDescNode(pPhyNode->pOutputDataBlockDesc);




×








UNCOV

905



  QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno);





×








UNCOV

906



  pInfo->binfo.pRes = pResBlock;




×







907













UNCOV

908



  SExprInfo* pExprInfo = NULL;




×








UNCOV

909



  code = createExprInfo(pCountNode->window.pFuncs, NULL, &pExprInfo, &numOfCols);




×








UNCOV

910



  QUERY_CHECK_CODE(code, lino, _error);





×







911













UNCOV

912



  code = initBasicInfoEx(&pInfo->binfo, pExpSup, pExprInfo, numOfCols, pResBlock, &pTaskInfo->storageAPI.functionStore);




×








UNCOV

913



  QUERY_CHECK_CODE(code, lino, _error);





×







914













UNCOV

915



  pInfo->twAggSup = (STimeWindowAggSupp){




×








UNCOV

916



      .waterMark = pCountNode->window.watermark,




×








UNCOV

917



      .calTrigger = pCountNode->window.triggerType,




×







918



      .maxTs = INT64_MIN,










919



      .minTs = INT64_MAX,











UNCOV

920



      .deleteMark = getDeleteMark(&pCountNode->window, 0),




×







921



  };










922













UNCOV

923



  pInfo->primaryTsIndex = ((SColumnNode*)pCountNode->window.pTspk)->slotId;




×








UNCOV

924



  code = initStreamAggSupporter(&pInfo->streamAggSup, pExpSup, numOfCols, 0, pTaskInfo->streamInfo.pState,




×







925



                                sizeof(COUNT_TYPE), 0, &pTaskInfo->storageAPI.stateStore, pHandle, &pInfo->twAggSup,











UNCOV

926



                                GET_TASKID(pTaskInfo), &pTaskInfo->storageAPI, pInfo->primaryTsIndex,




×







927



                                STREAM_STATE_BUFF_SORT, 1);











UNCOV

928



  QUERY_CHECK_CODE(code, lino, _error);





×







929













UNCOV

930



  pInfo->streamAggSup.windowCount = pCountNode->windowCount;




×








UNCOV

931



  pInfo->streamAggSup.windowSliding = pCountNode->windowSliding;




×







932













UNCOV

933



  code = initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);




×








UNCOV

934



  QUERY_CHECK_CODE(code, lino, _error);





×







935













UNCOV

936



  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);




×








UNCOV

937



  pInfo->pStDeleted = tSimpleHashInit(64, hashFn);




×








UNCOV

938



  QUERY_CHECK_NULL(pInfo->pStDeleted, code, lino, _error, terrno);





×








UNCOV

939



  pInfo->pDelIterator = NULL;




×







940













UNCOV

941



  code = createSpecialDataBlock(STREAM_DELETE_RESULT, &pInfo->pDelRes);




×








UNCOV

942



  QUERY_CHECK_CODE(code, lino, _error);





×







943













UNCOV

944



  pInfo->ignoreExpiredData = pCountNode->window.igExpired;




×








UNCOV

945



  pInfo->ignoreExpiredDataSaved = false;




×








UNCOV

946



  pInfo->pUpdated = NULL;




×








UNCOV

947



  pInfo->pStUpdated = NULL;




×








UNCOV

948



  pInfo->dataVersion = 0;




×








UNCOV

949



  pInfo->historyWins = taosArrayInit(4, sizeof(SSessionKey));




×








UNCOV

950



  if (!pInfo->historyWins) {





×







951



    code = terrno;




×







952



    QUERY_CHECK_CODE(code, lino, _error);





×







953



  }










954













UNCOV

955



  code = createSpecialDataBlock(STREAM_CHECKPOINT, &pInfo->pCheckpointRes);




×








UNCOV

956



  QUERY_CHECK_CODE(code, lino, _error);





×







957













UNCOV

958



  pInfo->recvGetAll = false;




×








UNCOV

959



  pInfo->pPkDeleted = tSimpleHashInit(64, hashFn);




×








UNCOV

960



  QUERY_CHECK_NULL(pInfo->pPkDeleted, code, lino, _error, terrno);





×








UNCOV

961



  pInfo->destHasPrimaryKey = pCountNode->window.destHasPrimaryKey;




×







962













UNCOV

963



  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT;




×








UNCOV

964



  setOperatorInfo(pOperator, getStreamOpName(pOperator->operatorType), QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT, true,




×







965



                  OP_NOT_OPENED, pInfo, pTaskInfo);










966



  // for stream











UNCOV

967



  void*   buff = NULL;




×








UNCOV

968



  int32_t len = 0;




×







969



  int32_t res =











UNCOV

970



      pInfo->streamAggSup.stateStore.streamStateGetInfo(pInfo->streamAggSup.pState, STREAM_COUNT_OP_CHECKPOINT_NAME,




×







971



                                                        strlen(STREAM_COUNT_OP_CHECKPOINT_NAME), &buff, &len);











UNCOV

972



  if (res == TSDB_CODE_SUCCESS) {





×







973



    code = doStreamCountDecodeOpState(buff, len, pOperator, true);




×







974



    QUERY_CHECK_CODE(code, lino, _error);





×







975



    taosMemoryFree(buff);





×







976



  }











UNCOV

977



  pInfo->pOperator = pOperator;




×








UNCOV

978



  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamCountAggNext, NULL, destroyStreamCountAggOperatorInfo,




×







979



                                         optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);











UNCOV

980



  setOperatorStreamStateFn(pOperator, streamCountReleaseState, streamCountReloadState);




×







981













UNCOV

982



  code = initStreamBasicInfo(&pInfo->basic, pOperator);




×








UNCOV

983



  QUERY_CHECK_CODE(code, lino, _error);





×







984













UNCOV

985



  if (downstream) {





×








UNCOV

986



    code = initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex,




×








UNCOV

987



                          &pInfo->twAggSup, &pInfo->basic);




×








UNCOV

988



    QUERY_CHECK_CODE(code, lino, _error);





×







989













UNCOV

990



    code = appendDownstream(pOperator, &downstream, 1);




×








UNCOV

991



    QUERY_CHECK_CODE(code, lino, _error);





×







992



  }










993













UNCOV

994



  *pOptrInfo = pOperator;




×








UNCOV

995



  return TSDB_CODE_SUCCESS;




×







996












997



_error:




×







998



  if (pInfo != NULL) {





×







999



    destroyStreamCountAggOperatorInfo(pInfo);




×







1000



  }










1001












1002



  destroyOperatorAndDownstreams(pOperator, &downstream, 1);




×







1003



  pTaskInfo->code = code;




×







1004



  qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







1005



  return code;




×







1006



}
























    

  • 

    •