taosdata / TDengine / #3653

coverage: 22.565% (-41.0%) from 63.596%
#3653

push

travis-ci

web-flow 
feat(keep): support keep on super table level. (#30097)

* Feat: support use keep while create super table.

* Test(keep): add test for create super table with keep option.

* Feat(keep): Add tmsg for create keep.

* Feat(keep): support alter table option keep.

* Fix(keep): Add baisc test for alter table option.

* Fix(keep): memory leek.

* Feat(keep): add keep to metaEntry&metaCache and fix earliestTs with stn keep.

* Test(keep): add some cases for select with stb keep.

* Fix: fix ci core while alter stb.

* Feat(keep): delete expired data in super table level.

* Feat: remove get stb keep while query.

* Fix : build error.

* Revert "Fix : build error."

This reverts commit 0ed66e4e8.

* Revert "Feat(keep): delete expired data in super table level."

This reverts commit 36330f6b4.

* Fix : build errors.

* Feat : support restart taosd.

* Fix : alter table comment problems.

* Test : add tests for super table keep.

* Fix: change sdb stb reserve size.

* Test: add more tests.

* Feat: Disable normal tables and sub tables from setting the keep parameter

* Fix: add more checks to avoid unknown address.

* Docs: Add docs for stable keep.

* Fix: some review changes.

* Fix: review errors.

49248 of 302527 branches covered (16.28%)

Branch coverage included in aggregate %.

53 of 99 new or added lines in 12 files covered. (53.54%)

155872 existing lines in 443 files now uncovered.

87359 of 302857 relevant lines covered (28.84%)

570004.22 hits per line

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

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









16



#include "filter.h"










17



#include "os.h"










18



#include "query.h"










19



#include "taosdef.h"










20



#include "tmsg.h"










21



#include "ttypes.h"










22












23



#include "executorInt.h"










24



#include "streamexecutorInt.h"










25



#include "tcommon.h"










26



#include "thash.h"










27



#include "ttime.h"










28












29



#include "function.h"










30



#include "operator.h"










31



#include "querynodes.h"










32



#include "querytask.h"










33



#include "tdatablock.h"










34



#include "tfill.h"










35












36



#define FILL_POS_INVALID 0










37



#define FILL_POS_START   1










38



#define FILL_POS_MID     2










39



#define FILL_POS_END     3










40












41



typedef struct STimeRange {










42



  TSKEY    skey;










43



  TSKEY    ekey;










44



  uint64_t groupId;










45



} STimeRange;










46













UNCOV

47



TSKEY getNextWindowTs(TSKEY ts, SInterval* pInterval) {




×








UNCOV

48



  STimeWindow win = {.skey = ts, .ekey = ts};




×








UNCOV

49



  getNextTimeWindow(pInterval, &win, TSDB_ORDER_ASC);




×








UNCOV

50



  return win.skey;




×







51



}










52













UNCOV

53



TSKEY getPrevWindowTs(TSKEY ts, SInterval* pInterval) {




×








UNCOV

54



  STimeWindow win = {.skey = ts, .ekey = ts};




×








UNCOV

55



  getNextTimeWindow(pInterval, &win, TSDB_ORDER_DESC);




×








UNCOV

56



  return win.skey;




×







57



}










58













UNCOV

59



int32_t setRowCell(SColumnInfoData* pCol, int32_t rowId, const SResultCellData* pCell) {




×








UNCOV

60



  return colDataSetVal(pCol, rowId, pCell->pData, pCell->isNull);




×







61



}










62













UNCOV

63



SResultCellData* getResultCell(SResultRowData* pRaw, int32_t index) {




×








UNCOV

64



  if (!pRaw || !pRaw->pRowVal) {





×







65



    return NULL;




×







66



  }











UNCOV

67



  char*            pData = (char*)pRaw->pRowVal;




×








UNCOV

68



  SResultCellData* pCell = pRaw->pRowVal;




×








UNCOV

69



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





×








UNCOV

70



    pData += (pCell->bytes + sizeof(SResultCellData));




×








UNCOV

71



    pCell = (SResultCellData*)pData;




×







72



  }











UNCOV

73



  return pCell;




×







74



}










75













UNCOV

76



void* destroyFillColumnInfo(SFillColInfo* pFillCol, int32_t start, int32_t end) {




×








UNCOV

77



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





×








UNCOV

78



    destroyExprInfo(pFillCol[i].pExpr, 1);




×








UNCOV

79



    taosVariantDestroy(&pFillCol[i].fillVal);




×







80



  }











UNCOV

81



  if (start < end) {





×








UNCOV

82



    taosMemoryFreeClear(pFillCol[start].pExpr);





×







83



  }











UNCOV

84



  taosMemoryFree(pFillCol);





×








UNCOV

85



  return NULL;




×







86



}










87













UNCOV

88



void destroyStreamFillSupporter(SStreamFillSupporter* pFillSup) {




×








UNCOV

89



  if (pFillSup == NULL) {





×







90



    return;




×







91



  }











UNCOV

92



  pFillSup->pAllColInfo = destroyFillColumnInfo(pFillSup->pAllColInfo, pFillSup->numOfFillCols, pFillSup->numOfAllCols);




×








UNCOV

93



  tSimpleHashCleanup(pFillSup->pResMap);




×








UNCOV

94



  pFillSup->pResMap = NULL;




×








UNCOV

95



  cleanupExprSupp(&pFillSup->notFillExprSup);




×








UNCOV

96



  if (pFillSup->cur.pRowVal != pFillSup->prev.pRowVal && pFillSup->cur.pRowVal != pFillSup->next.pRowVal) {





×








UNCOV

97



    taosMemoryFree(pFillSup->cur.pRowVal);





×







98



  }











UNCOV

99



  taosMemoryFree(pFillSup->prev.pRowVal);





×








UNCOV

100



  taosMemoryFree(pFillSup->next.pRowVal);





×








UNCOV

101



  taosMemoryFree(pFillSup->nextNext.pRowVal);





×







102













UNCOV

103



  taosMemoryFree(pFillSup->pOffsetInfo);





×







104













UNCOV

105



  taosMemoryFree(pFillSup);





×







106



}










107













UNCOV

108



void destroySPoint(void* ptr) {




×








UNCOV

109



  SPoint* point = (SPoint*)ptr;




×








UNCOV

110



  taosMemoryFreeClear(point->val);





×








UNCOV

111



}




×







112













UNCOV

113



void destroyStreamFillLinearInfo(SStreamFillLinearInfo* pFillLinear) {




×








UNCOV

114



  taosArrayDestroyEx(pFillLinear->pEndPoints, destroySPoint);




×








UNCOV

115



  taosArrayDestroyEx(pFillLinear->pNextEndPoints, destroySPoint);




×








UNCOV

116



  taosMemoryFree(pFillLinear);





×








UNCOV

117



}




×







118













UNCOV

119



void destroyStreamFillInfo(SStreamFillInfo* pFillInfo) {




×








UNCOV

120



  if (pFillInfo == NULL) {





×







121



    return;




×







122



  } 











UNCOV

123



  if (pFillInfo->type == TSDB_FILL_SET_VALUE || pFillInfo->type == TSDB_FILL_SET_VALUE_F ||





×








UNCOV

124



      pFillInfo->type == TSDB_FILL_NULL || pFillInfo->type == TSDB_FILL_NULL_F) {





×








UNCOV

125



    taosMemoryFreeClear(pFillInfo->pResRow->pRowVal);





×








UNCOV

126



    taosMemoryFreeClear(pFillInfo->pResRow);





×








UNCOV

127



    taosMemoryFreeClear(pFillInfo->pNonFillRow->pRowVal);





×








UNCOV

128



    taosMemoryFreeClear(pFillInfo->pNonFillRow);





×







129



  }











UNCOV

130



  destroyStreamFillLinearInfo(pFillInfo->pLinearInfo);




×








UNCOV

131



  pFillInfo->pLinearInfo = NULL;




×







132













UNCOV

133



  taosArrayDestroy(pFillInfo->delRanges);




×








UNCOV

134



  taosMemoryFreeClear(pFillInfo->pTempBuff);





×








UNCOV

135



  taosMemoryFree(pFillInfo);





×







136



}










137













UNCOV

138



static void destroyStreamFillOperatorInfo(void* param) {




×








UNCOV

139



  SStreamFillOperatorInfo* pInfo = (SStreamFillOperatorInfo*)param;




×








UNCOV

140



  destroyStreamFillInfo(pInfo->pFillInfo);




×








UNCOV

141



  destroyStreamFillSupporter(pInfo->pFillSup);




×








UNCOV

142



  blockDataDestroy(pInfo->pRes);




×








UNCOV

143



  pInfo->pRes = NULL;




×








UNCOV

144



  blockDataDestroy(pInfo->pSrcBlock);




×








UNCOV

145



  pInfo->pSrcBlock = NULL;




×








UNCOV

146



  blockDataDestroy(pInfo->pDelRes);




×








UNCOV

147



  pInfo->pDelRes = NULL;




×








UNCOV

148



  taosArrayDestroy(pInfo->matchInfo.pList);




×








UNCOV

149



  pInfo->matchInfo.pList = NULL;




×








UNCOV

150



  taosArrayDestroy(pInfo->pUpdated);




×








UNCOV

151



  clearGroupResInfo(&pInfo->groupResInfo);




×








UNCOV

152



  taosArrayDestroy(pInfo->pCloseTs);




×







153













UNCOV

154



  if (pInfo->stateStore.streamFileStateDestroy != NULL) {





×








UNCOV

155



    pInfo->stateStore.streamFileStateDestroy(pInfo->pState->pFileState);




×







156



  }










157













UNCOV

158



  if (pInfo->pState != NULL) {





×








UNCOV

159



    taosMemoryFreeClear(pInfo->pState);





×







160



  }










161













UNCOV

162



  taosMemoryFree(pInfo);





×








UNCOV

163



}




×







164













UNCOV

165



static void resetFillWindow(SResultRowData* pRowData) {




×








UNCOV

166



  pRowData->key = INT64_MIN;




×








UNCOV

167



  taosMemoryFreeClear(pRowData->pRowVal);





×








UNCOV

168



}




×







169













UNCOV

170



static void resetPrevAndNextWindow(SStreamFillSupporter* pFillSup) {




×








UNCOV

171



  if (pFillSup->cur.pRowVal != pFillSup->prev.pRowVal && pFillSup->cur.pRowVal != pFillSup->next.pRowVal) {





×








UNCOV

172



    resetFillWindow(&pFillSup->cur);




×







173



  } else {











UNCOV

174



    pFillSup->cur.key = INT64_MIN;




×








UNCOV

175



    pFillSup->cur.pRowVal = NULL;




×







176



  }











UNCOV

177



  resetFillWindow(&pFillSup->prev);




×








UNCOV

178



  resetFillWindow(&pFillSup->next);




×








UNCOV

179



  resetFillWindow(&pFillSup->nextNext);




×








UNCOV

180



}




×







181













UNCOV

182



void getWindowFromDiscBuf(SOperatorInfo* pOperator, TSKEY ts, uint64_t groupId, SStreamFillSupporter* pFillSup) {




×








UNCOV

183



  SStorageAPI* pAPI = &pOperator->pTaskInfo->storageAPI;




×








UNCOV

184



  void*        pState = pOperator->pTaskInfo->streamInfo.pState;




×








UNCOV

185



  resetPrevAndNextWindow(pFillSup);




×







186













UNCOV

187



  SWinKey key = {.ts = ts, .groupId = groupId};




×








UNCOV

188



  void*   curVal = NULL;




×








UNCOV

189



  int32_t curVLen = 0;




×








UNCOV

190



  bool    hasCurKey = true;




×








UNCOV

191



  int32_t code = pAPI->stateStore.streamStateFillGet(pState, &key, (void**)&curVal, &curVLen, NULL);




×








UNCOV

192



  if (code == TSDB_CODE_SUCCESS) {





×








UNCOV

193



    pFillSup->cur.key = key.ts;




×








UNCOV

194



    pFillSup->cur.pRowVal = curVal;




×







195



  } else {











UNCOV

196



    qDebug("streamStateFillGet key failed, Data may be deleted. ts:%" PRId64 ", groupId:%" PRId64, ts, groupId);





×








UNCOV

197



    pFillSup->cur.key = ts;




×








UNCOV

198



    pFillSup->cur.pRowVal = NULL;




×








UNCOV

199



    hasCurKey = false;




×







200



  }










201













UNCOV

202



  SStreamStateCur* pCur = pAPI->stateStore.streamStateFillSeekKeyPrev(pState, &key);




×








UNCOV

203



  SWinKey          preKey = {.ts = INT64_MIN, .groupId = groupId};




×








UNCOV

204



  void*            preVal = NULL;




×








UNCOV

205



  int32_t          preVLen = 0;




×








UNCOV

206



  code = pAPI->stateStore.streamStateFillGetGroupKVByCur(pCur, &preKey, (const void**)&preVal, &preVLen);




×







207













UNCOV

208



  if (code == TSDB_CODE_SUCCESS) {





×








UNCOV

209



    pFillSup->prev.key = preKey.ts;




×








UNCOV

210



    pFillSup->prev.pRowVal = preVal;




×







211













UNCOV

212



    if (hasCurKey) {





×








UNCOV

213



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




×







214



    }










215













UNCOV

216



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




×







217



  } else {











UNCOV

218



    pAPI->stateStore.streamStateFreeCur(pCur);




×








UNCOV

219



    pCur = pAPI->stateStore.streamStateFillSeekKeyNext(pState, &key);




×







220



  }










221













UNCOV

222



  SWinKey nextKey = {.ts = INT64_MIN, .groupId = groupId};




×








UNCOV

223



  void*   nextVal = NULL;




×








UNCOV

224



  int32_t nextVLen = 0;




×








UNCOV

225



  code = pAPI->stateStore.streamStateFillGetGroupKVByCur(pCur, &nextKey, (const void**)&nextVal, &nextVLen);




×








UNCOV

226



  if (code == TSDB_CODE_SUCCESS) {





×








UNCOV

227



    pFillSup->next.key = nextKey.ts;




×








UNCOV

228



    pFillSup->next.pRowVal = nextVal;




×








UNCOV

229



    if (pFillSup->type == TSDB_FILL_PREV || pFillSup->type == TSDB_FILL_NEXT) {





×








UNCOV

230



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




×








UNCOV

231



      SWinKey nextNextKey = {.groupId = groupId};




×








UNCOV

232



      void*   nextNextVal = NULL;




×








UNCOV

233



      int32_t nextNextVLen = 0;




×








UNCOV

234



      code = pAPI->stateStore.streamStateFillGetGroupKVByCur(pCur, &nextNextKey, (const void**)&nextNextVal, &nextNextVLen);




×








UNCOV

235



      if (code == TSDB_CODE_SUCCESS) {





×








UNCOV

236



        pFillSup->nextNext.key = nextNextKey.ts;




×








UNCOV

237



        pFillSup->nextNext.pRowVal = nextNextVal;




×







238



      }










239



    }










240



  }











UNCOV

241



  pAPI->stateStore.streamStateFreeCur(pCur);




×








UNCOV

242



}




×







243












244



bool hasCurWindow(SStreamFillSupporter* pFillSup) { return pFillSup->cur.key != INT64_MIN; }




×








UNCOV

245



bool hasPrevWindow(SStreamFillSupporter* pFillSup) { return pFillSup->prev.key != INT64_MIN; }




×








UNCOV

246



bool hasNextWindow(SStreamFillSupporter* pFillSup) { return pFillSup->next.key != INT64_MIN; }




×








UNCOV

247



static bool hasNextNextWindow(SStreamFillSupporter* pFillSup) { return pFillSup->nextNext.key != INT64_MIN; }




×







248













UNCOV

249



static void transBlockToResultRow(const SSDataBlock* pBlock, int32_t rowId, TSKEY ts, SResultRowData* pRowVal) {




×








UNCOV

250



  int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);




×








UNCOV

251



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





×








UNCOV

252



    SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, i);




×








UNCOV

253



    SResultCellData* pCell = getResultCell(pRowVal, i);




×








UNCOV

254



    if (!colDataIsNull_s(pColData, rowId)) {





×








UNCOV

255



      pCell->isNull = false;




×








UNCOV

256



      pCell->type = pColData->info.type;




×








UNCOV

257



      pCell->bytes = pColData->info.bytes;




×








UNCOV

258



      char* val = colDataGetData(pColData, rowId);





×








UNCOV

259



      if (IS_VAR_DATA_TYPE(pCell->type)) {





×








UNCOV

260



        memcpy(pCell->pData, val, varDataTLen(val));




×







261



      } else {











UNCOV

262



        memcpy(pCell->pData, val, pCell->bytes);




×







263



      }










264



    } else {











UNCOV

265



      pCell->isNull = true;




×







266



    }










267



  }











UNCOV

268



  pRowVal->key = ts;




×








UNCOV

269



}




×







270













UNCOV

271



static void calcRowDeltaData(SResultRowData* pEndRow, SArray* pEndPoins, SFillColInfo* pFillCol, int32_t numOfCol) {




×








UNCOV

272



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





×








UNCOV

273



    if (!pFillCol[i].notFillCol) {





×








UNCOV

274



      int32_t          slotId = GET_DEST_SLOT_ID(pFillCol + i);




×








UNCOV

275



      SResultCellData* pECell = getResultCell(pEndRow, slotId);




×








UNCOV

276



      SPoint*          pPoint = taosArrayGet(pEndPoins, slotId);




×








UNCOV

277



      pPoint->key = pEndRow->key;




×








UNCOV

278



      memcpy(pPoint->val, pECell->pData, pECell->bytes);




×







279



    }










280



  }











UNCOV

281



}




×







282













UNCOV

283



static void setFillInfoStart(TSKEY ts, SInterval* pInterval, SStreamFillInfo* pFillInfo) {




×








UNCOV

284



  ts = taosTimeAdd(ts, pInterval->sliding, pInterval->slidingUnit, pInterval->precision, NULL);




×








UNCOV

285



  pFillInfo->start = ts;




×








UNCOV

286



}




×







287













UNCOV

288



static void setFillInfoEnd(TSKEY ts, SInterval* pInterval, SStreamFillInfo* pFillInfo) {




×








UNCOV

289



  ts = taosTimeAdd(ts, pInterval->sliding * -1, pInterval->slidingUnit, pInterval->precision, NULL);




×








UNCOV

290



  pFillInfo->end = ts;




×








UNCOV

291



}




×







292













UNCOV

293



static void setFillKeyInfo(TSKEY start, TSKEY end, SInterval* pInterval, SStreamFillInfo* pFillInfo) {




×








UNCOV

294



  setFillInfoStart(start, pInterval, pFillInfo);




×








UNCOV

295



  pFillInfo->current = pFillInfo->start;




×








UNCOV

296



  setFillInfoEnd(end, pInterval, pFillInfo);




×








UNCOV

297



}




×







298













UNCOV

299



void setDeleteFillValueInfo(TSKEY start, TSKEY end, SStreamFillSupporter* pFillSup, SStreamFillInfo* pFillInfo) {




×








UNCOV

300



  if (!hasPrevWindow(pFillSup) || !hasNextWindow(pFillSup)) {





×








UNCOV

301



    pFillInfo->needFill = false;




×








UNCOV

302



    return;




×







303



  }










304













UNCOV

305



  TSKEY realStart = taosTimeAdd(pFillSup->prev.key, pFillSup->interval.sliding, pFillSup->interval.slidingUnit,




×








UNCOV

306



                                pFillSup->interval.precision, NULL);




×







307













UNCOV

308



  pFillInfo->needFill = true;




×








UNCOV

309



  pFillInfo->start = realStart;




×








UNCOV

310



  pFillInfo->current = pFillInfo->start;




×








UNCOV

311



  pFillInfo->end = end;




×








UNCOV

312



  pFillInfo->pos = FILL_POS_INVALID;




×








UNCOV

313



  switch (pFillInfo->type) {





×








UNCOV

314



    case TSDB_FILL_NULL:




×







315



    case TSDB_FILL_NULL_F:










316



    case TSDB_FILL_SET_VALUE:










317



    case TSDB_FILL_SET_VALUE_F:











UNCOV

318



      break;




×








UNCOV

319



    case TSDB_FILL_PREV:




×








UNCOV

320



      pFillInfo->pResRow = &pFillSup->prev;




×








UNCOV

321



      break;




×








UNCOV

322



    case TSDB_FILL_NEXT:




×








UNCOV

323



      pFillInfo->pResRow = &pFillSup->next;




×








UNCOV

324



      break;




×








UNCOV

325



    case TSDB_FILL_LINEAR: {




×








UNCOV

326



      setFillKeyInfo(pFillSup->prev.key, pFillSup->next.key, &pFillSup->interval, pFillInfo);




×








UNCOV

327



      pFillInfo->pLinearInfo->hasNext = false;




×








UNCOV

328



      pFillInfo->pLinearInfo->nextEnd = INT64_MIN;




×








UNCOV

329



      calcRowDeltaData(&pFillSup->next, pFillInfo->pLinearInfo->pEndPoints, pFillSup->pAllColInfo,




×







330



                       pFillSup->numOfAllCols);











UNCOV

331



      pFillInfo->pResRow = &pFillSup->prev;




×








UNCOV

332



      pFillInfo->pLinearInfo->winIndex = 0;




×








UNCOV

333



    } break;




×







334



    default:




×







335



      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));





×







336



      break;




×







337



  }










338



}










339













UNCOV

340



void copyNotFillExpData(SStreamFillSupporter* pFillSup, SStreamFillInfo* pFillInfo) {




×








UNCOV

341



  for (int32_t i = pFillSup->numOfFillCols; i < pFillSup->numOfAllCols; ++i) {





×








UNCOV

342



    SFillColInfo*    pFillCol = pFillSup->pAllColInfo + i;




×








UNCOV

343



    int32_t          slotId = GET_DEST_SLOT_ID(pFillCol);




×








UNCOV

344



    SResultCellData* pCell = getResultCell(pFillInfo->pResRow, slotId);




×








UNCOV

345



    SResultCellData* pCurCell = getResultCell(&pFillSup->cur, slotId);




×








UNCOV

346



    pCell->isNull = pCurCell->isNull;




×








UNCOV

347



    if (!pCurCell->isNull) {





×








UNCOV

348



      memcpy(pCell->pData, pCurCell->pData, pCell->bytes);




×







349



    }










350



  }











UNCOV

351



}




×







352













UNCOV

353



void setFillValueInfo(SSDataBlock* pBlock, TSKEY ts, int32_t rowId, SStreamFillSupporter* pFillSup,




×







354



                      SStreamFillInfo* pFillInfo) {











UNCOV

355



  pFillInfo->preRowKey = pFillSup->cur.key;




×








UNCOV

356



  if (!hasPrevWindow(pFillSup) && !hasNextWindow(pFillSup)) {





×








UNCOV

357



    pFillInfo->needFill = false;




×








UNCOV

358



    pFillInfo->pos = FILL_POS_START;




×








UNCOV

359



    return;




×







360



  }











UNCOV

361



  TSKEY prevWKey = INT64_MIN;




×








UNCOV

362



  TSKEY nextWKey = INT64_MIN;




×








UNCOV

363



  if (hasPrevWindow(pFillSup)) {





×








UNCOV

364



    prevWKey = pFillSup->prev.key;




×







365



  }











UNCOV

366



  if (hasNextWindow(pFillSup)) {





×








UNCOV

367



    nextWKey = pFillSup->next.key;




×







368



  }










369













UNCOV

370



  pFillInfo->needFill = true;




×








UNCOV

371



  pFillInfo->pos = FILL_POS_INVALID;




×








UNCOV

372



  switch (pFillInfo->type) {





×








UNCOV

373



    case TSDB_FILL_NULL:




×







374



    case TSDB_FILL_NULL_F:










375



    case TSDB_FILL_SET_VALUE:










376



    case TSDB_FILL_SET_VALUE_F: {











UNCOV

377



      if (pFillSup->prev.key == pFillInfo->preRowKey) {





×







378



        resetFillWindow(&pFillSup->prev);




×







379



      }











UNCOV

380



      if (hasPrevWindow(pFillSup) && hasNextWindow(pFillSup)) {





×








UNCOV

381



        if (pFillSup->next.key == pFillInfo->nextRowKey) {





×








UNCOV

382



          pFillInfo->preRowKey = INT64_MIN;




×








UNCOV

383



          setFillKeyInfo(prevWKey, ts, &pFillSup->interval, pFillInfo);




×








UNCOV

384



          pFillInfo->pos = FILL_POS_END;




×







385



        } else {











UNCOV

386



          pFillInfo->needFill = false;




×








UNCOV

387



          pFillInfo->pos = FILL_POS_START;




×







388



        }











UNCOV

389



      } else if (hasPrevWindow(pFillSup)) {





×








UNCOV

390



        setFillKeyInfo(prevWKey, ts, &pFillSup->interval, pFillInfo);




×








UNCOV

391



        pFillInfo->pos = FILL_POS_END;




×







392



      } else {











UNCOV

393



        setFillKeyInfo(ts, nextWKey, &pFillSup->interval, pFillInfo);




×








UNCOV

394



        pFillInfo->pos = FILL_POS_START;




×







395



      }











UNCOV

396



      copyNotFillExpData(pFillSup, pFillInfo);




×








UNCOV

397



    } break;




×








UNCOV

398



    case TSDB_FILL_PREV: {




×








UNCOV

399



      if (hasNextWindow(pFillSup) && ((pFillSup->next.key != pFillInfo->nextRowKey) ||





×








UNCOV

400



                                      (pFillSup->next.key == pFillInfo->nextRowKey && hasNextNextWindow(pFillSup)) ||





×








UNCOV

401



                                      (pFillSup->next.key == pFillInfo->nextRowKey && !hasPrevWindow(pFillSup)))) {





×








UNCOV

402



        setFillKeyInfo(ts, nextWKey, &pFillSup->interval, pFillInfo);




×








UNCOV

403



        pFillInfo->pos = FILL_POS_START;




×








UNCOV

404



        resetFillWindow(&pFillSup->prev);




×








UNCOV

405



        pFillSup->prev.key = pFillSup->cur.key;




×








UNCOV

406



        pFillSup->prev.pRowVal = pFillSup->cur.pRowVal;




×








UNCOV

407



      } else if (hasPrevWindow(pFillSup)) {





×








UNCOV

408



        setFillKeyInfo(prevWKey, ts, &pFillSup->interval, pFillInfo);




×








UNCOV

409



        pFillInfo->pos = FILL_POS_END;




×








UNCOV

410



        pFillInfo->preRowKey = INT64_MIN;




×







411



      }











UNCOV

412



      pFillInfo->pResRow = &pFillSup->prev;




×








UNCOV

413



    } break;




×








UNCOV

414



    case TSDB_FILL_NEXT: {




×








UNCOV

415



      if (hasPrevWindow(pFillSup)) {





×








UNCOV

416



        setFillKeyInfo(prevWKey, ts, &pFillSup->interval, pFillInfo);




×








UNCOV

417



        pFillInfo->pos = FILL_POS_END;




×








UNCOV

418



        resetFillWindow(&pFillSup->next);




×








UNCOV

419



        pFillSup->next.key = pFillSup->cur.key;




×








UNCOV

420



        pFillSup->next.pRowVal = pFillSup->cur.pRowVal;




×








UNCOV

421



        pFillInfo->preRowKey = INT64_MIN;




×







422



      } else {











UNCOV

423



        setFillKeyInfo(ts, nextWKey, &pFillSup->interval, pFillInfo);




×








UNCOV

424



        pFillInfo->pos = FILL_POS_START;




×







425



      }











UNCOV

426



      pFillInfo->pResRow = &pFillSup->next;




×








UNCOV

427



    } break;




×








UNCOV

428



    case TSDB_FILL_LINEAR: {




×








UNCOV

429



      pFillInfo->pLinearInfo->winIndex = 0;




×








UNCOV

430



      if (hasPrevWindow(pFillSup) && hasNextWindow(pFillSup)) {





×








UNCOV

431



        setFillKeyInfo(prevWKey, ts, &pFillSup->interval, pFillInfo);




×








UNCOV

432



        pFillInfo->pos = FILL_POS_MID;




×








UNCOV

433



        pFillInfo->pLinearInfo->nextEnd = nextWKey;




×








UNCOV

434



        calcRowDeltaData(&pFillSup->cur, pFillInfo->pLinearInfo->pEndPoints, pFillSup->pAllColInfo,




×







435



                         pFillSup->numOfAllCols);











UNCOV

436



        pFillInfo->pResRow = &pFillSup->prev;




×







437













UNCOV

438



        calcRowDeltaData(&pFillSup->next, pFillInfo->pLinearInfo->pNextEndPoints, pFillSup->pAllColInfo,




×







439



                         pFillSup->numOfAllCols);











UNCOV

440



        pFillInfo->pLinearInfo->hasNext = true;




×








UNCOV

441



      } else if (hasPrevWindow(pFillSup)) {





×








UNCOV

442



        setFillKeyInfo(prevWKey, ts, &pFillSup->interval, pFillInfo);




×








UNCOV

443



        pFillInfo->pos = FILL_POS_END;




×








UNCOV

444



        pFillInfo->pLinearInfo->nextEnd = INT64_MIN;




×








UNCOV

445



        calcRowDeltaData(&pFillSup->cur, pFillInfo->pLinearInfo->pEndPoints, pFillSup->pAllColInfo,




×







446



                         pFillSup->numOfAllCols);











UNCOV

447



        pFillInfo->pResRow = &pFillSup->prev;




×








UNCOV

448



        pFillInfo->pLinearInfo->hasNext = false;




×







449



      } else {











UNCOV

450



        setFillKeyInfo(ts, nextWKey, &pFillSup->interval, pFillInfo);




×








UNCOV

451



        pFillInfo->pos = FILL_POS_START;




×








UNCOV

452



        pFillInfo->pLinearInfo->nextEnd = INT64_MIN;




×








UNCOV

453



        calcRowDeltaData(&pFillSup->next, pFillInfo->pLinearInfo->pEndPoints, pFillSup->pAllColInfo,




×







454



                         pFillSup->numOfAllCols);











UNCOV

455



        pFillInfo->pResRow = &pFillSup->cur;




×








UNCOV

456



        pFillInfo->pLinearInfo->hasNext = false;




×







457



      }











UNCOV

458



    } break;




×








UNCOV

459



    default:




×








UNCOV

460



      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));





×







461



      break;




×







462



  }










463



}










464













UNCOV

465



int32_t checkResult(SStreamFillSupporter* pFillSup, TSKEY ts, uint64_t groupId, bool* pRes) {




×








UNCOV

466



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

467



  int32_t lino = 0;




×








UNCOV

468



  SWinKey key = {.groupId = groupId, .ts = ts};




×








UNCOV

469



  if (tSimpleHashGet(pFillSup->pResMap, &key, sizeof(SWinKey)) != NULL) {





×








UNCOV

470



    (*pRes) = false;




×








UNCOV

471



    goto _end;




×







472



  }











UNCOV

473



  code = tSimpleHashPut(pFillSup->pResMap, &key, sizeof(SWinKey), NULL, 0);




×








UNCOV

474



  QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

475



  (*pRes) = true;




×







476













UNCOV

477



_end:




×








UNCOV

478



  if (code != TSDB_CODE_SUCCESS) {





×







479



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





×







480



  }











UNCOV

481



  return code;




×







482



}










483













UNCOV

484



static int32_t buildFillResult(SResultRowData* pResRow, SStreamFillSupporter* pFillSup, TSKEY ts, SSDataBlock* pBlock,




×







485



                               bool* pRes) {











UNCOV

486



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

487



  int32_t lino = 0;




×








UNCOV

488



  if (pBlock->info.rows >= pBlock->info.capacity) {





×








UNCOV

489



    (*pRes) = false;




×








UNCOV

490



    goto _end;




×







491



  }











UNCOV

492



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




×








UNCOV

493



  bool     ckRes = true;




×








UNCOV

494



  code = checkResult(pFillSup, ts, groupId, &ckRes);




×








UNCOV

495



  QUERY_CHECK_CODE(code, lino, _end);





×







496













UNCOV

497



  if (pFillSup->hasDelete && !ckRes) {





×








UNCOV

498



    (*pRes) = true;




×








UNCOV

499



    goto _end;




×







500



  }











UNCOV

501



  for (int32_t i = 0; i < pFillSup->numOfAllCols; ++i) {





×








UNCOV

502



    SFillColInfo*    pFillCol = pFillSup->pAllColInfo + i;




×








UNCOV

503



    int32_t          slotId = GET_DEST_SLOT_ID(pFillCol);




×








UNCOV

504



    SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, slotId);




×








UNCOV

505



    SFillInfo        tmpInfo = {




×







506



               .currentKey = ts,










507



               .order = TSDB_ORDER_ASC,










508



               .interval = pFillSup->interval,










509



    };











UNCOV

510



    bool filled = fillIfWindowPseudoColumn(&tmpInfo, pFillCol, pColData, pBlock->info.rows);




×








UNCOV

511



    if (!filled) {





×








UNCOV

512



      SResultCellData* pCell = getResultCell(pResRow, slotId);




×








UNCOV

513



      code = setRowCell(pColData, pBlock->info.rows, pCell);




×








UNCOV

514



      QUERY_CHECK_CODE(code, lino, _end);





×







515



    }










516



  }











UNCOV

517



  pBlock->info.rows++;




×








UNCOV

518



  (*pRes) = true;




×







519













UNCOV

520



_end:




×








UNCOV

521



  if (code != TSDB_CODE_SUCCESS) {





×







522



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





×







523



  }











UNCOV

524



  return code;




×







525



}










526













UNCOV

527



bool hasRemainCalc(SStreamFillInfo* pFillInfo) {




×








UNCOV

528



  if (pFillInfo->current != INT64_MIN && pFillInfo->current <= pFillInfo->end) {





×








UNCOV

529



    return true;




×







530



  }











UNCOV

531



  return false;




×







532



}










533













UNCOV

534



static void doStreamFillNormal(SStreamFillSupporter* pFillSup, SStreamFillInfo* pFillInfo, SSDataBlock* pBlock) {




×








UNCOV

535



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

536



  int32_t lino = 0;




×








UNCOV

537



  while (hasRemainCalc(pFillInfo) && pBlock->info.rows < pBlock->info.capacity) {





×








UNCOV

538



    STimeWindow st = {.skey = pFillInfo->current, .ekey = pFillInfo->current};




×








UNCOV

539



    if (inWinRange(&pFillSup->winRange, &st)) {





×








UNCOV

540



      bool res = true;




×








UNCOV

541



      code = buildFillResult(pFillInfo->pResRow, pFillSup, pFillInfo->current, pBlock, &res);




×








UNCOV

542



      QUERY_CHECK_CODE(code, lino, _end);





×







543



    }











UNCOV

544



    pFillInfo->current = taosTimeAdd(pFillInfo->current, pFillSup->interval.sliding, pFillSup->interval.slidingUnit,




×








UNCOV

545



                                     pFillSup->interval.precision, NULL);




×







546



  }










547













UNCOV

548



_end:




×








UNCOV

549



  if (code != TSDB_CODE_SUCCESS) {





×







550



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





×







551



  }











UNCOV

552



}




×







553













UNCOV

554



static void doStreamFillLinear(SStreamFillSupporter* pFillSup, SStreamFillInfo* pFillInfo, SSDataBlock* pBlock) {




×








UNCOV

555



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

556



  int32_t lino = 0;




×








UNCOV

557



  while (hasRemainCalc(pFillInfo) && pBlock->info.rows < pBlock->info.capacity) {





×








UNCOV

558



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




×








UNCOV

559



    SWinKey     key = {.groupId = groupId, .ts = pFillInfo->current};




×








UNCOV

560



    STimeWindow st = {.skey = pFillInfo->current, .ekey = pFillInfo->current};




×








UNCOV

561



    bool        ckRes = true;




×








UNCOV

562



    code = checkResult(pFillSup, pFillInfo->current, groupId, &ckRes);




×








UNCOV

563



    QUERY_CHECK_CODE(code, lino, _end);





×







564













UNCOV

565



    if ((pFillSup->hasDelete && !ckRes) || !inWinRange(&pFillSup->winRange, &st)) {





×








UNCOV

566



      pFillInfo->current = taosTimeAdd(pFillInfo->current, pFillSup->interval.sliding, pFillSup->interval.slidingUnit,




×








UNCOV

567



                                       pFillSup->interval.precision, NULL);




×








UNCOV

568



      pFillInfo->pLinearInfo->winIndex++;




×








UNCOV

569



      continue;




×







570



    }











UNCOV

571



    pFillInfo->pLinearInfo->winIndex++;




×








UNCOV

572



    for (int32_t i = 0; i < pFillSup->numOfAllCols; ++i) {





×








UNCOV

573



      SFillColInfo* pFillCol = pFillSup->pAllColInfo + i;




×








UNCOV

574



      SFillInfo     tmp = {




×








UNCOV

575



              .currentKey = pFillInfo->current,




×







576



              .order = TSDB_ORDER_ASC,










577



              .interval = pFillSup->interval,










578



      };










579













UNCOV

580



      int32_t          slotId = GET_DEST_SLOT_ID(pFillCol);




×








UNCOV

581



      SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, slotId);




×








UNCOV

582



      int16_t          type = pColData->info.type;




×








UNCOV

583



      SResultCellData* pCell = getResultCell(pFillInfo->pResRow, slotId);




×








UNCOV

584



      int32_t          index = pBlock->info.rows;




×








UNCOV

585



      if (pFillCol->notFillCol) {





×








UNCOV

586



        bool filled = fillIfWindowPseudoColumn(&tmp, pFillCol, pColData, index);




×








UNCOV

587



        if (!filled) {





×








UNCOV

588



          code = setRowCell(pColData, index, pCell);




×







589



          QUERY_CHECK_CODE(code, lino, _end);





×







590



        }










591



      } else {











UNCOV

592



        if (IS_VAR_DATA_TYPE(type) || type == TSDB_DATA_TYPE_BOOL || pCell->isNull) {





×








UNCOV

593



          colDataSetNULL(pColData, index);





×








UNCOV

594



          continue;




×







595



        }











UNCOV

596



        SPoint* pEnd = taosArrayGet(pFillInfo->pLinearInfo->pEndPoints, slotId);




×








UNCOV

597



        double  vCell = 0;




×








UNCOV

598



        SPoint  start = {0};




×








UNCOV

599



        start.key = pFillInfo->pResRow->key;




×








UNCOV

600



        start.val = pCell->pData;




×







601













UNCOV

602



        SPoint cur = {0};




×








UNCOV

603



        cur.key = pFillInfo->current;




×








UNCOV

604



        cur.val = taosMemoryCalloc(1, pCell->bytes);





×








UNCOV

605



        QUERY_CHECK_NULL(cur.val, code, lino, _end, terrno);





×








UNCOV

606



        taosGetLinearInterpolationVal(&cur, pCell->type, &start, pEnd, pCell->type);




×








UNCOV

607



        code = colDataSetVal(pColData, index, (const char*)cur.val, false);




×








UNCOV

608



        QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

609



        destroySPoint(&cur);




×







610



      }










611



    }











UNCOV

612



    pFillInfo->current = taosTimeAdd(pFillInfo->current, pFillSup->interval.sliding, pFillSup->interval.slidingUnit,




×








UNCOV

613



                                     pFillSup->interval.precision, NULL);




×








UNCOV

614



    pBlock->info.rows++;




×







615



  }










616













UNCOV

617



_end:




×








UNCOV

618



  if (code != TSDB_CODE_SUCCESS) {





×







619



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





×







620



  }











UNCOV

621



}




×







622













UNCOV

623



static void keepResultInDiscBuf(SOperatorInfo* pOperator, uint64_t groupId, SResultRowData* pRow, int32_t len) {




×








UNCOV

624



  SStorageAPI* pAPI = &pOperator->pTaskInfo->storageAPI;




×







625













UNCOV

626



  SWinKey key = {.groupId = groupId, .ts = pRow->key};




×








UNCOV

627



  int32_t code = pAPI->stateStore.streamStateFillPut(pOperator->pTaskInfo->streamInfo.pState, &key, pRow->pRowVal, len);




×








UNCOV

628



  qDebug("===stream===fill operator save key ts:%" PRId64 " group id:%" PRIu64 "  code:%d", key.ts, key.groupId, code);





×








UNCOV

629



  if (code != TSDB_CODE_SUCCESS) {





×







630



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





×







631



  }











UNCOV

632



}




×







633













UNCOV

634



static void doStreamFillRange(SStreamFillInfo* pFillInfo, SStreamFillSupporter* pFillSup, SSDataBlock* pRes) {




×








UNCOV

635



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

636



  int32_t lino = 0;




×








UNCOV

637



  bool    res = false;




×








UNCOV

638



  if (pFillInfo->needFill == false) {





×








UNCOV

639



    code = buildFillResult(&pFillSup->cur, pFillSup, pFillSup->cur.key, pRes, &res);




×








UNCOV

640



    QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

641



    return;




×







642



  }










643













UNCOV

644



  if (pFillInfo->pos == FILL_POS_START) {





×








UNCOV

645



    code = buildFillResult(&pFillSup->cur, pFillSup, pFillSup->cur.key, pRes, &res);




×








UNCOV

646



    QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

647



    if (res) {





×








UNCOV

648



      pFillInfo->pos = FILL_POS_INVALID;




×







649



    }










650



  }











UNCOV

651



  if (pFillInfo->type != TSDB_FILL_LINEAR) {





×








UNCOV

652



    doStreamFillNormal(pFillSup, pFillInfo, pRes);




×







653



  } else {











UNCOV

654



    doStreamFillLinear(pFillSup, pFillInfo, pRes);




×







655













UNCOV

656



    if (pFillInfo->pos == FILL_POS_MID) {





×








UNCOV

657



      code = buildFillResult(&pFillSup->cur, pFillSup, pFillSup->cur.key, pRes, &res);




×








UNCOV

658



      QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

659



      if (res) {





×








UNCOV

660



        pFillInfo->pos = FILL_POS_INVALID;




×







661



      }










662



    }










663













UNCOV

664



    if (pFillInfo->current > pFillInfo->end && pFillInfo->pLinearInfo->hasNext) {





×








UNCOV

665



      pFillInfo->pLinearInfo->hasNext = false;




×








UNCOV

666



      pFillInfo->pLinearInfo->winIndex = 0;




×








UNCOV

667



      taosArraySwap(pFillInfo->pLinearInfo->pEndPoints, pFillInfo->pLinearInfo->pNextEndPoints);




×








UNCOV

668



      pFillInfo->pResRow = &pFillSup->cur;




×








UNCOV

669



      setFillKeyInfo(pFillSup->cur.key, pFillInfo->pLinearInfo->nextEnd, &pFillSup->interval, pFillInfo);




×








UNCOV

670



      doStreamFillLinear(pFillSup, pFillInfo, pRes);




×







671



    }










672



  }











UNCOV

673



  if (pFillInfo->pos == FILL_POS_END) {





×








UNCOV

674



    code = buildFillResult(&pFillSup->cur, pFillSup, pFillSup->cur.key, pRes, &res);




×








UNCOV

675



    QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

676



    if (res) {





×








UNCOV

677



      pFillInfo->pos = FILL_POS_INVALID;




×







678



    }










679



  }










680













UNCOV

681



_end:




×








UNCOV

682



  if (code != TSDB_CODE_SUCCESS) {





×







683



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





×







684



  }










685



}










686













UNCOV

687



int32_t keepBlockRowInDiscBuf(SOperatorInfo* pOperator, SStreamFillInfo* pFillInfo, SSDataBlock* pBlock, TSKEY* tsCol,




×







688



                           int32_t rowId, uint64_t groupId, int32_t rowSize) {











UNCOV

689



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

690



  int32_t lino = 0;




×








UNCOV

691



  TSKEY ts = tsCol[rowId];




×








UNCOV

692



  pFillInfo->nextRowKey = ts;




×








UNCOV

693



  SResultRowData tmpNextRow = {.key = ts};




×








UNCOV

694



  tmpNextRow.pRowVal = taosMemoryCalloc(1, rowSize);





×








UNCOV

695



  QUERY_CHECK_NULL(tmpNextRow.pRowVal, code, lino, _end, terrno);





×








UNCOV

696



  transBlockToResultRow(pBlock, rowId, ts, &tmpNextRow);




×








UNCOV

697



  keepResultInDiscBuf(pOperator, groupId, &tmpNextRow, rowSize);




×








UNCOV

698



  taosMemoryFreeClear(tmpNextRow.pRowVal);





×







699












700



_end:




×








UNCOV

701



  if (code != TSDB_CODE_SUCCESS) {





×







702



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





×







703



  }











UNCOV

704



  return code;




×







705



}










706













UNCOV

707



static void doFillResults(SOperatorInfo* pOperator, SStreamFillSupporter* pFillSup, SStreamFillInfo* pFillInfo,




×







708



                          SSDataBlock* pBlock, TSKEY* tsCol, int32_t rowId, SSDataBlock* pRes) {











UNCOV

709



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




×








UNCOV

710



  getWindowFromDiscBuf(pOperator, tsCol[rowId], groupId, pFillSup);




×








UNCOV

711



  if (pFillSup->prev.key == pFillInfo->preRowKey) {





×








UNCOV

712



    resetFillWindow(&pFillSup->prev);




×







713



  }











UNCOV

714



  setFillValueInfo(pBlock, tsCol[rowId], rowId, pFillSup, pFillInfo);




×








UNCOV

715



  doStreamFillRange(pFillInfo, pFillSup, pRes);




×








UNCOV

716



}




×







717













UNCOV

718



static void doStreamFillImpl(SOperatorInfo* pOperator) {




×








UNCOV

719



  int32_t                  code = TSDB_CODE_SUCCESS;




×








UNCOV

720



  int32_t                  lino = 0;




×








UNCOV

721



  SStreamFillOperatorInfo* pInfo = pOperator->info;




×








UNCOV

722



  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

723



  SStreamFillSupporter*    pFillSup = pInfo->pFillSup;




×








UNCOV

724



  SStreamFillInfo*         pFillInfo = pInfo->pFillInfo;




×








UNCOV

725



  SSDataBlock*             pBlock = pInfo->pSrcBlock;




×








UNCOV

726



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




×








UNCOV

727



  SSDataBlock*             pRes = pInfo->pRes;




×








UNCOV

728



  SColumnInfoData*         pTsCol = taosArrayGet(pInfo->pSrcBlock->pDataBlock, pInfo->primaryTsCol);




×








UNCOV

729



  TSKEY*                   tsCol = (TSKEY*)pTsCol->pData;




×








UNCOV

730



  pRes->info.id.groupId = groupId;




×








UNCOV

731



  pInfo->srcRowIndex++;




×







732













UNCOV

733



  if (pInfo->srcRowIndex == 0) {





×








UNCOV

734



    code = keepBlockRowInDiscBuf(pOperator, pFillInfo, pBlock, tsCol, pInfo->srcRowIndex, groupId, pFillSup->rowSize);




×








UNCOV

735



    QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

736



    pInfo->srcRowIndex++;




×







737



  }










738













UNCOV

739



  while (pInfo->srcRowIndex < pBlock->info.rows) {





×








UNCOV

740



    code = keepBlockRowInDiscBuf(pOperator, pFillInfo, pBlock, tsCol, pInfo->srcRowIndex, groupId, pFillSup->rowSize);




×








UNCOV

741



    QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

742



    doFillResults(pOperator, pFillSup, pFillInfo, pBlock, tsCol, pInfo->srcRowIndex - 1, pRes);




×








UNCOV

743



    if (pInfo->pRes->info.rows == pInfo->pRes->info.capacity) {





×








UNCOV

744



      code = blockDataUpdateTsWindow(pRes, pInfo->primaryTsCol);




×








UNCOV

745



      QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

746



      return;




×







747



    }











UNCOV

748



    pInfo->srcRowIndex++;




×







749



  }











UNCOV

750



  doFillResults(pOperator, pFillSup, pFillInfo, pBlock, tsCol, pInfo->srcRowIndex - 1, pRes);




×








UNCOV

751



  code = blockDataUpdateTsWindow(pRes, pInfo->primaryTsCol);




×








UNCOV

752



  QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

753



  blockDataCleanup(pInfo->pSrcBlock);




×







754













UNCOV

755



_end:




×








UNCOV

756



  if (code != TSDB_CODE_SUCCESS) {





×







757



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





×







758



  }










759



}










760













UNCOV

761



static int32_t buildDeleteRange(SOperatorInfo* pOp, TSKEY start, TSKEY end, uint64_t groupId, SSDataBlock* delRes) {




×








UNCOV

762



  int32_t          code = TSDB_CODE_SUCCESS;




×








UNCOV

763



  int32_t          lino = 0;




×








UNCOV

764



  SStorageAPI*     pAPI = &pOp->pTaskInfo->storageAPI;




×








UNCOV

765



  void*            pState = pOp->pTaskInfo->streamInfo.pState;




×








UNCOV

766



  SExecTaskInfo*   pTaskInfo = pOp->pTaskInfo;




×








UNCOV

767



  SSDataBlock*     pBlock = delRes;




×








UNCOV

768



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




×








UNCOV

769



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




×








UNCOV

770



  SColumnInfoData* pUidCol = taosArrayGet(pBlock->pDataBlock, UID_COLUMN_INDEX);




×








UNCOV

771



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




×








UNCOV

772



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




×








UNCOV

773



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




×








UNCOV

774



  SColumnInfoData* pTbNameCol = taosArrayGet(pBlock->pDataBlock, TABLE_NAME_COLUMN_INDEX);




×








UNCOV

775



  code = colDataSetVal(pStartCol, pBlock->info.rows, (const char*)&start, false);




×








UNCOV

776



  QUERY_CHECK_CODE(code, lino, _end);





×







777













UNCOV

778



  code = colDataSetVal(pEndCol, pBlock->info.rows, (const char*)&end, false);




×








UNCOV

779



  QUERY_CHECK_CODE(code, lino, _end);





×







780













UNCOV

781



  colDataSetNULL(pUidCol, pBlock->info.rows);





×








UNCOV

782



  code = colDataSetVal(pGroupCol, pBlock->info.rows, (const char*)&groupId, false);




×








UNCOV

783



  QUERY_CHECK_CODE(code, lino, _end);





×







784













UNCOV

785



  colDataSetNULL(pCalStartCol, pBlock->info.rows);





×








UNCOV

786



  colDataSetNULL(pCalEndCol, pBlock->info.rows);





×







787













UNCOV

788



  SColumnInfoData* pTableCol = taosArrayGet(pBlock->pDataBlock, TABLE_NAME_COLUMN_INDEX);




×







789













UNCOV

790



  void*   tbname = NULL;




×








UNCOV

791



  int32_t winCode = TSDB_CODE_SUCCESS;




×








UNCOV

792



  code = pAPI->stateStore.streamStateGetParName(pOp->pTaskInfo->streamInfo.pState, groupId, &tbname, false, &winCode);




×








UNCOV

793



  QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

794



  if (winCode != TSDB_CODE_SUCCESS) {





×








UNCOV

795



    colDataSetNULL(pTableCol, pBlock->info.rows);





×







796



  } else {










797



    char parTbName[VARSTR_HEADER_SIZE + TSDB_TABLE_NAME_LEN];











UNCOV

798



    STR_WITH_MAXSIZE_TO_VARSTR(parTbName, tbname, sizeof(parTbName));




×








UNCOV

799



    code = colDataSetVal(pTableCol, pBlock->info.rows, (const char*)parTbName, false);




×








UNCOV

800



    QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

801



    pAPI->stateStore.streamStateFreeVal(tbname);




×







802



  }










803













UNCOV

804



  pBlock->info.rows++;




×







805













UNCOV

806



_end:




×








UNCOV

807



  if (code != TSDB_CODE_SUCCESS) {





×







808



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





×







809



  }











UNCOV

810



  return code;




×







811



}










812













UNCOV

813



static int32_t buildDeleteResult(SOperatorInfo* pOperator, TSKEY startTs, TSKEY endTs, uint64_t groupId,




×







814



                                 SSDataBlock* delRes) {











UNCOV

815



  int32_t                  code = TSDB_CODE_SUCCESS;




×








UNCOV

816



  int32_t                  lino = 0;




×








UNCOV

817



  SStreamFillOperatorInfo* pInfo = pOperator->info;




×








UNCOV

818



  SStreamFillSupporter*    pFillSup = pInfo->pFillSup;




×








UNCOV

819



  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

820



  if (hasPrevWindow(pFillSup)) {





×








UNCOV

821



    TSKEY start = getNextWindowTs(pFillSup->prev.key, &pFillSup->interval);




×








UNCOV

822



    code = buildDeleteRange(pOperator, start, endTs, groupId, delRes);




×








UNCOV

823



    QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

824



  } else if (hasNextWindow(pFillSup)) {





×








UNCOV

825



    TSKEY end = getPrevWindowTs(pFillSup->next.key, &pFillSup->interval);




×








UNCOV

826



    code = buildDeleteRange(pOperator, startTs, end, groupId, delRes);




×








UNCOV

827



    QUERY_CHECK_CODE(code, lino, _end);





×







828



  } else {











UNCOV

829



    code = buildDeleteRange(pOperator, startTs, endTs, groupId, delRes);




×








UNCOV

830



    QUERY_CHECK_CODE(code, lino, _end);





×







831



  }










832













UNCOV

833



_end:




×








UNCOV

834



  if (code != TSDB_CODE_SUCCESS) {





×







835



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





×







836



  }











UNCOV

837



  return code;




×







838



}










839













UNCOV

840



static int32_t doDeleteFillResultImpl(SOperatorInfo* pOperator, TSKEY startTs, TSKEY endTs, uint64_t groupId) {




×








UNCOV

841



  int32_t                  code = TSDB_CODE_SUCCESS;




×








UNCOV

842



  int32_t                  lino = 0;




×








UNCOV

843



  SStorageAPI*             pAPI = &pOperator->pTaskInfo->storageAPI;




×








UNCOV

844



  SStreamFillOperatorInfo* pInfo = pOperator->info;




×








UNCOV

845



  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

846



  getWindowFromDiscBuf(pOperator, startTs, groupId, pInfo->pFillSup);




×








UNCOV

847



  setDeleteFillValueInfo(startTs, endTs, pInfo->pFillSup, pInfo->pFillInfo);




×








UNCOV

848



  SWinKey key = {.ts = startTs, .groupId = groupId};




×








UNCOV

849



  pAPI->stateStore.streamStateFillDel(pOperator->pTaskInfo->streamInfo.pState, &key);




×








UNCOV

850



  if (!pInfo->pFillInfo->needFill) {





×








UNCOV

851



    code = buildDeleteResult(pOperator, startTs, endTs, groupId, pInfo->pDelRes);




×








UNCOV

852



    QUERY_CHECK_CODE(code, lino, _end);





×







853



  } else {











UNCOV

854



    STimeRange tw = {




×







855



        .skey = startTs,










856



        .ekey = endTs,










857



        .groupId = groupId,










858



    };











UNCOV

859



    void* tmp = taosArrayPush(pInfo->pFillInfo->delRanges, &tw);




×








UNCOV

860



    if (!tmp) {





×







861



      code = terrno;




×







862



      QUERY_CHECK_CODE(code, lino, _end);





×







863



    }










864



  }










865













UNCOV

866



_end:




×








UNCOV

867



  if (code != TSDB_CODE_SUCCESS) {





×







868



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





×







869



  }











UNCOV

870



  return code;




×







871



}










872












873



static void getWindowInfoByKey(SStorageAPI* pAPI, void* pState, TSKEY ts, int64_t groupId, SResultRowData* pWinData) {




×







874



  SWinKey key = {.ts = ts, .groupId = groupId};




×







875



  void*   val = NULL;




×







876



  int32_t len = 0;




×







877



  int32_t code = pAPI->stateStore.streamStateFillGet(pState, &key, (void**)&val, &len, NULL);




×







878



  if (code != TSDB_CODE_SUCCESS) {





×







879



    qDebug("get window info by key failed, Data may be deleted, try next window. ts:%" PRId64 ", groupId:%" PRId64, ts,





×







880



           groupId);










881



    SStreamStateCur* pCur = pAPI->stateStore.streamStateFillSeekKeyNext(pState, &key);




×







882



    code = pAPI->stateStore.streamStateFillGetGroupKVByCur(pCur, &key, (const void**)&val, &len);




×







883



    pAPI->stateStore.streamStateFreeCur(pCur);




×







884



    qDebug("get window info by key ts:%" PRId64 ", groupId:%" PRId64 ", res%d", ts, groupId, code);





×







885



  }










886












887



  if (code == TSDB_CODE_SUCCESS) {





×







888



    resetFillWindow(pWinData);




×







889



    pWinData->key = key.ts;




×







890



    pWinData->pRowVal = val;




×







891



  }










892



}




×







893













UNCOV

894



static void doDeleteFillFinalize(SOperatorInfo* pOperator) {




×








UNCOV

895



  SStorageAPI* pAPI = &pOperator->pTaskInfo->storageAPI;




×







896













UNCOV

897



  SStreamFillOperatorInfo* pInfo = pOperator->info;




×








UNCOV

898



  SStreamFillInfo*         pFillInfo = pInfo->pFillInfo;




×








UNCOV

899



  int32_t                  size = taosArrayGetSize(pFillInfo->delRanges);




×








UNCOV

900



  while (pFillInfo->delIndex < size) {





×








UNCOV

901



    STimeRange* range = taosArrayGet(pFillInfo->delRanges, pFillInfo->delIndex);




×








UNCOV

902



    if (pInfo->pRes->info.id.groupId != 0 && pInfo->pRes->info.id.groupId != range->groupId) {





×







903



      return;




×







904



    }











UNCOV

905



    getWindowFromDiscBuf(pOperator, range->skey, range->groupId, pInfo->pFillSup);




×








UNCOV

906



    TSKEY realEnd = range->ekey + 1;




×








UNCOV

907



    if (pInfo->pFillInfo->type == TSDB_FILL_NEXT && pInfo->pFillSup->next.key != realEnd) {





×







908



      getWindowInfoByKey(pAPI, pOperator->pTaskInfo->streamInfo.pState, realEnd, range->groupId,




×







909



                         &pInfo->pFillSup->next);




×







910



    }











UNCOV

911



    setDeleteFillValueInfo(range->skey, range->ekey, pInfo->pFillSup, pInfo->pFillInfo);




×








UNCOV

912



    pFillInfo->delIndex++;




×








UNCOV

913



    if (pInfo->pFillInfo->needFill) {





×








UNCOV

914



      doStreamFillRange(pInfo->pFillInfo, pInfo->pFillSup, pInfo->pRes);




×








UNCOV

915



      pInfo->pRes->info.id.groupId = range->groupId;




×







916



    }










917



  }










918



}










919













UNCOV

920



static int32_t doDeleteFillResult(SOperatorInfo* pOperator) {




×








UNCOV

921



  int32_t                  code = TSDB_CODE_SUCCESS;




×








UNCOV

922



  int32_t                  lino = 0;




×








UNCOV

923



  SStorageAPI*             pAPI = &pOperator->pTaskInfo->storageAPI;




×








UNCOV

924



  SStreamFillOperatorInfo* pInfo = pOperator->info;




×








UNCOV

925



  SStreamFillInfo*         pFillInfo = pInfo->pFillInfo;




×








UNCOV

926



  SSDataBlock*             pBlock = pInfo->pSrcDelBlock;




×








UNCOV

927



  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;




×







928













UNCOV

929



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




×








UNCOV

930



  TSKEY*           tsStarts = (TSKEY*)pStartCol->pData;




×








UNCOV

931



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




×








UNCOV

932



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




×








UNCOV

933



  while (pInfo->srcDelRowIndex < pBlock->info.rows) {





×








UNCOV

934



    TSKEY            ts = tsStarts[pInfo->srcDelRowIndex];




×








UNCOV

935



    TSKEY            endTs = ts;




×








UNCOV

936



    uint64_t         groupId = groupIds[pInfo->srcDelRowIndex];




×








UNCOV

937



    SWinKey          key = {.ts = ts, .groupId = groupId};




×








UNCOV

938



    SStreamStateCur* pCur = pAPI->stateStore.streamStateGetAndCheckCur(pOperator->pTaskInfo->streamInfo.pState, &key);




×







939













UNCOV

940



    if (!pCur) {





×








UNCOV

941



      pInfo->srcDelRowIndex++;




×








UNCOV

942



      continue;




×







943



    }










944













UNCOV

945



    SWinKey nextKey = {.groupId = groupId, .ts = ts};




×








UNCOV

946



    while (pInfo->srcDelRowIndex < pBlock->info.rows) {





×








UNCOV

947



      TSKEY    delTs = tsStarts[pInfo->srcDelRowIndex];




×








UNCOV

948



      uint64_t delGroupId = groupIds[pInfo->srcDelRowIndex];




×








UNCOV

949



      int32_t  winCode = TSDB_CODE_SUCCESS;




×








UNCOV

950



      if (groupId != delGroupId) {





×








UNCOV

951



        break;




×







952



      }











UNCOV

953



      if (delTs > nextKey.ts) {





×








UNCOV

954



        break;




×







955



      }










956













UNCOV

957



      SWinKey delKey = {.groupId = delGroupId, .ts = delTs};




×








UNCOV

958



      if (delTs == nextKey.ts) {





×








UNCOV

959



        pAPI->stateStore.streamStateCurNext(pOperator->pTaskInfo->streamInfo.pState, pCur);




×








UNCOV

960



        winCode = pAPI->stateStore.streamStateFillGetGroupKVByCur(pCur, &nextKey, NULL, NULL);




×







961



        // ts will be deleted later











UNCOV

962



        if (delTs != ts) {





×








UNCOV

963



          pAPI->stateStore.streamStateFillDel(pOperator->pTaskInfo->streamInfo.pState, &delKey);




×








UNCOV

964



          pAPI->stateStore.streamStateFreeCur(pCur);




×








UNCOV

965



          pCur = pAPI->stateStore.streamStateGetAndCheckCur(pOperator->pTaskInfo->streamInfo.pState, &nextKey);




×







966



        }











UNCOV

967



        endTs = TMAX(delTs, nextKey.ts - 1);




×








UNCOV

968



        if (winCode != TSDB_CODE_SUCCESS) {





×








UNCOV

969



          break;




×







970



        }










971



      }











UNCOV

972



      pInfo->srcDelRowIndex++;




×







973



    }










974













UNCOV

975



    pAPI->stateStore.streamStateFreeCur(pCur);




×








UNCOV

976



    code = doDeleteFillResultImpl(pOperator, ts, endTs, groupId);




×








UNCOV

977



    QUERY_CHECK_CODE(code, lino, _end);





×







978



  }










979













UNCOV

980



  pFillInfo->current = pFillInfo->end + 1;




×







981













UNCOV

982



_end:




×








UNCOV

983



  if (code != TSDB_CODE_SUCCESS) {





×







984



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





×







985



  }











UNCOV

986



  return code;




×







987



}










988













UNCOV

989



void resetStreamFillSup(SStreamFillSupporter* pFillSup) {




×








UNCOV

990



  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);




×








UNCOV

991



  SSHashObj* pNewMap = tSimpleHashInit(16, hashFn);




×








UNCOV

992



  if (pNewMap != NULL) {





×








UNCOV

993



    tSimpleHashCleanup(pFillSup->pResMap);




×








UNCOV

994



    pFillSup->pResMap = pNewMap;




×







995



  } else {










996



    tSimpleHashClear(pFillSup->pResMap);




×







997



  }











UNCOV

998



  pFillSup->hasDelete = false;




×








UNCOV

999



}




×








UNCOV

1000



void resetStreamFillInfo(SStreamFillOperatorInfo* pInfo) {




×








UNCOV

1001



  resetStreamFillSup(pInfo->pFillSup);




×








UNCOV

1002



  taosArrayClear(pInfo->pFillInfo->delRanges);




×








UNCOV

1003



  pInfo->pFillInfo->delIndex = 0;




×








UNCOV

1004



}




×







1005













UNCOV

1006



static int32_t doApplyStreamScalarCalculation(SOperatorInfo* pOperator, SSDataBlock* pSrcBlock,




×







1007



                                              SSDataBlock* pDstBlock) {











UNCOV

1008



  int32_t                  code = TSDB_CODE_SUCCESS;




×








UNCOV

1009



  int32_t                  lino = 0;




×








UNCOV

1010



  SStreamFillOperatorInfo* pInfo = pOperator->info;




×








UNCOV

1011



  SExprSupp*               pSup = &pOperator->exprSupp;




×








UNCOV

1012



  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;




×







1013













UNCOV

1014



  blockDataCleanup(pDstBlock);




×








UNCOV

1015



  code = blockDataEnsureCapacity(pDstBlock, pSrcBlock->info.rows);




×








UNCOV

1016



  QUERY_CHECK_CODE(code, lino, _end);





×







1017













UNCOV

1018



  code = setInputDataBlock(pSup, pSrcBlock, TSDB_ORDER_ASC, MAIN_SCAN, false);




×








UNCOV

1019



  QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

1020



  code = projectApplyFunctions(pSup->pExprInfo, pDstBlock, pSrcBlock, pSup->pCtx, pSup->numOfExprs, NULL);




×








UNCOV

1021



  QUERY_CHECK_CODE(code, lino, _end);





×







1022













UNCOV

1023



  pDstBlock->info.rows = 0;




×








UNCOV

1024



  pSup = &pInfo->pFillSup->notFillExprSup;




×








UNCOV

1025



  code = setInputDataBlock(pSup, pSrcBlock, TSDB_ORDER_ASC, MAIN_SCAN, false);




×








UNCOV

1026



  QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

1027



  code = projectApplyFunctions(pSup->pExprInfo, pDstBlock, pSrcBlock, pSup->pCtx, pSup->numOfExprs, NULL);




×








UNCOV

1028



  QUERY_CHECK_CODE(code, lino, _end);





×







1029













UNCOV

1030



  pDstBlock->info.id.groupId = pSrcBlock->info.id.groupId;




×







1031













UNCOV

1032



  code = blockDataUpdateTsWindow(pDstBlock, pInfo->primaryTsCol);




×







1033













UNCOV

1034



_end:




×








UNCOV

1035



  if (code != TSDB_CODE_SUCCESS) {





×







1036



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





×







1037



  }











UNCOV

1038



  return code;




×







1039



}










1040













UNCOV

1041



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




×








UNCOV

1042



  int32_t                  code = TSDB_CODE_SUCCESS;




×








UNCOV

1043



  int32_t                  lino = 0;




×








UNCOV

1044



  SStreamFillOperatorInfo* pInfo = pOperator->info;




×








UNCOV

1045



  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;




×







1046













UNCOV

1047



  if (pOperator->status == OP_EXEC_DONE) {





×







1048



    (*ppRes) = NULL;




×







1049



    return code;




×







1050



  }











UNCOV

1051



  blockDataCleanup(pInfo->pRes);




×








UNCOV

1052



  if (hasRemainCalc(pInfo->pFillInfo) ||





×








UNCOV

1053



      (pInfo->pFillInfo->pos != FILL_POS_INVALID && pInfo->pFillInfo->needFill == true)) {





×








UNCOV

1054



    doStreamFillRange(pInfo->pFillInfo, pInfo->pFillSup, pInfo->pRes);




×








UNCOV

1055



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





×








UNCOV

1056



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




×








UNCOV

1057



      (*ppRes) = pInfo->pRes;




×








UNCOV

1058



      return code;




×







1059



    }










1060



  }











UNCOV

1061



  if (pOperator->status == OP_RES_TO_RETURN) {





×








UNCOV

1062



    doDeleteFillFinalize(pOperator);




×








UNCOV

1063



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





×







1064



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




×







1065



      (*ppRes) = pInfo->pRes;




×







1066



      return code;




×







1067



    }











UNCOV

1068



    setOperatorCompleted(pOperator);




×








UNCOV

1069



    resetStreamFillInfo(pInfo);




×








UNCOV

1070



    (*ppRes) = NULL;




×








UNCOV

1071



    return code;




×







1072



  }










1073













UNCOV

1074



  SSDataBlock*   fillResult = NULL;




×








UNCOV

1075



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




×







1076



  while (1) {











UNCOV

1077



    if (pInfo->srcRowIndex >= pInfo->pSrcBlock->info.rows || pInfo->pSrcBlock->info.rows == 0) {





×







1078



      // If there are delete datablocks, we receive  them first.











UNCOV

1079



      SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);




×








UNCOV

1080



      if (pBlock == NULL) {





×








UNCOV

1081



        pOperator->status = OP_RES_TO_RETURN;




×








UNCOV

1082



        pInfo->pFillInfo->preRowKey = INT64_MIN;




×








UNCOV

1083



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





×







1084



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




×







1085



          (*ppRes) = pInfo->pRes;




×







1086



          return code;




×







1087



        }











UNCOV

1088



        break;




×







1089



      }











UNCOV

1090



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




×







1091













UNCOV

1092



      if (pInfo->pFillInfo->curGroupId != pBlock->info.id.groupId) {





×








UNCOV

1093



        pInfo->pFillInfo->curGroupId = pBlock->info.id.groupId;




×








UNCOV

1094



        pInfo->pFillInfo->preRowKey = INT64_MIN;




×







1095



      }










1096













UNCOV

1097



      pInfo->pFillSup->winRange = pTaskInfo->streamInfo.fillHistoryWindow;




×








UNCOV

1098



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





×







1099



        pInfo->pFillSup->winRange.ekey = INT64_MAX;




×







1100



      }










1101













UNCOV

1102



      switch (pBlock->info.type) {





×








UNCOV

1103



        case STREAM_RETRIEVE:




×








UNCOV

1104



          (*ppRes) = pBlock;




×








UNCOV

1105



          return code;




×








UNCOV

1106



        case STREAM_DELETE_RESULT: {




×








UNCOV

1107



          pInfo->pSrcDelBlock = pBlock;




×








UNCOV

1108



          pInfo->srcDelRowIndex = 0;




×








UNCOV

1109



          blockDataCleanup(pInfo->pDelRes);




×








UNCOV

1110



          pInfo->pFillSup->hasDelete = true;




×








UNCOV

1111



          code = doDeleteFillResult(pOperator);




×








UNCOV

1112



          QUERY_CHECK_CODE(code, lino, _end);





×







1113













UNCOV

1114



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





×








UNCOV

1115



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




×








UNCOV

1116



            (*ppRes) = pInfo->pDelRes;




×








UNCOV

1117



            return code;




×







1118



          }











UNCOV

1119



          continue;




×







1120



        } break;











UNCOV

1121



        case STREAM_NORMAL:




×







1122



        case STREAM_INVALID:










1123



        case STREAM_PULL_DATA: {











UNCOV

1124



          code = doApplyStreamScalarCalculation(pOperator, pBlock, pInfo->pSrcBlock);




×








UNCOV

1125



          QUERY_CHECK_CODE(code, lino, _end);





×







1126













UNCOV

1127



          memcpy(pInfo->pSrcBlock->info.parTbName, pBlock->info.parTbName, TSDB_TABLE_NAME_LEN);




×








UNCOV

1128



          pInfo->srcRowIndex = -1;




×








UNCOV

1129



        } break;




×








UNCOV

1130



        case STREAM_CHECKPOINT:




×







1131



        case STREAM_CREATE_CHILD_TABLE: {











UNCOV

1132



          (*ppRes) = pBlock;




×








UNCOV

1133



          return code;




×







1134



        } break;










1135



        default:




×







1136



          return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;




×







1137



      }










1138



    }










1139













UNCOV

1140



    doStreamFillImpl(pOperator);




×








UNCOV

1141



    code = doFilter(pInfo->pRes, pOperator->exprSupp.pFilterInfo, &pInfo->matchInfo);




×








UNCOV

1142



    QUERY_CHECK_CODE(code, lino, _end);





×







1143













UNCOV

1144



    memcpy(pInfo->pRes->info.parTbName, pInfo->pSrcBlock->info.parTbName, TSDB_TABLE_NAME_LEN);




×








UNCOV

1145



    pOperator->resultInfo.totalRows += pInfo->pRes->info.rows;




×








UNCOV

1146



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





×








UNCOV

1147



      break;




×







1148



    }










1149



  }











UNCOV

1150



  if (pOperator->status == OP_RES_TO_RETURN) {





×








UNCOV

1151



    doDeleteFillFinalize(pOperator);




×







1152



  }










1153













UNCOV

1154



  if (pInfo->pRes->info.rows == 0) {





×








UNCOV

1155



    setOperatorCompleted(pOperator);




×








UNCOV

1156



    resetStreamFillInfo(pInfo);




×








UNCOV

1157



    (*ppRes) = NULL;




×








UNCOV

1158



    return code;




×







1159



  }










1160













UNCOV

1161



  pOperator->resultInfo.totalRows += pInfo->pRes->info.rows;




×








UNCOV

1162



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




×








UNCOV

1163



  (*ppRes) = pInfo->pRes;




×








UNCOV

1164



  return code;




×







1165












1166



_end:




×







1167



  if (code != TSDB_CODE_SUCCESS) {





×







1168



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





×







1169



    pTaskInfo->code = code;




×







1170



    T_LONG_JMP(pTaskInfo->env, code);





×







1171



  }










1172



  setOperatorCompleted(pOperator);




×







1173



  resetStreamFillInfo(pInfo);




×







1174



  (*ppRes) = NULL;




×







1175



  return code;




×







1176



}










1177













UNCOV

1178



static void resetForceFillWindow(SResultRowData* pRowData) {




×








UNCOV

1179



  pRowData->key = INT64_MIN;




×








UNCOV

1180



  pRowData->pRowVal = NULL;




×








UNCOV

1181



}




×







1182













UNCOV

1183



void doBuildForceFillResultImpl(SOperatorInfo* pOperator, SStreamFillSupporter* pFillSup,




×







1184



                                SStreamFillInfo* pFillInfo, SSDataBlock* pBlock, SGroupResInfo* pGroupResInfo) {











UNCOV

1185



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

1186



  int32_t lino = 0;




×







1187













UNCOV

1188



  SStreamFillOperatorInfo* pInfo = pOperator->info;




×








UNCOV

1189



  bool                     res = false;




×








UNCOV

1190



  int32_t                  numOfRows = getNumOfTotalRes(pGroupResInfo);




×








UNCOV

1191



  for (; pGroupResInfo->index < numOfRows; pGroupResInfo->index++) {





×








UNCOV

1192



    SWinKey* pKey = (SWinKey*)taosArrayGet(pGroupResInfo->pRows, pGroupResInfo->index);




×








UNCOV

1193



    if (pBlock->info.id.groupId == 0) {





×








UNCOV

1194



      pBlock->info.id.groupId = pKey->groupId;




×








UNCOV

1195



    } else if (pBlock->info.id.groupId != pKey->groupId) {





×








UNCOV

1196



      break;




×







1197



    }










1198













UNCOV

1199



    SRowBuffPos* pValPos = NULL;




×








UNCOV

1200



    int32_t      len = 0;




×








UNCOV

1201



    int32_t      winCode = TSDB_CODE_SUCCESS;




×








UNCOV

1202



    code = pInfo->stateStore.streamStateFillGet(pInfo->pState, pKey, (void**)&pValPos, &len, &winCode);




×








UNCOV

1203



    QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

1204



    qDebug("===stream=== build force fill res. key:%" PRId64 ",groupId:%" PRId64".res:%d", pKey->ts, pKey->groupId, winCode);





×








UNCOV

1205



    if (winCode == TSDB_CODE_SUCCESS) {





×








UNCOV

1206



      pFillSup->cur.key = pKey->ts;




×








UNCOV

1207



      pFillSup->cur.pRowVal = pValPos->pRowBuff;




×








UNCOV

1208



      code = buildFillResult(&pFillSup->cur, pFillSup, pKey->ts, pBlock, &res);




×








UNCOV

1209



      QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

1210



      resetForceFillWindow(&pFillSup->cur);




×








UNCOV

1211



      releaseOutputBuf(pInfo->pState, pValPos, &pInfo->stateStore);




×







1212



    } else {











UNCOV

1213



      SWinKey      preKey = {.ts = INT64_MIN, .groupId = pKey->groupId};




×








UNCOV

1214



      SRowBuffPos* prePos = NULL;




×








UNCOV

1215



      int32_t      preVLen = 0;




×








UNCOV

1216



      code = pInfo->stateStore.streamStateFillGetPrev(pInfo->pState, pKey, &preKey,




×







1217



                                                      (void**)&prePos, &preVLen, &winCode);











UNCOV

1218



      QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

1219



      if (winCode == TSDB_CODE_SUCCESS) {





×








UNCOV

1220



        pFillSup->cur.key = pKey->ts;




×








UNCOV

1221



        pFillSup->cur.pRowVal = prePos->pRowBuff;




×








UNCOV

1222



        if (pFillInfo->type == TSDB_FILL_PREV) {





×








UNCOV

1223



          code = buildFillResult(&pFillSup->cur, pFillSup, pKey->ts, pBlock, &res);




×








UNCOV

1224



          QUERY_CHECK_CODE(code, lino, _end);





×







1225



        } else {











UNCOV

1226



          copyNotFillExpData(pFillSup, pFillInfo);




×








UNCOV

1227



          pFillInfo->pResRow->key = pKey->ts;




×








UNCOV

1228



          code = buildFillResult(pFillInfo->pResRow, pFillSup, pKey->ts, pBlock, &res);




×








UNCOV

1229



          QUERY_CHECK_CODE(code, lino, _end);





×







1230



        }











UNCOV

1231



        resetForceFillWindow(&pFillSup->cur);




×







1232



      }











UNCOV

1233



      releaseOutputBuf(pInfo->pState, prePos, &pInfo->stateStore);




×







1234



    }










1235



  }










1236













UNCOV

1237



_end:




×








UNCOV

1238



  if (code != TSDB_CODE_SUCCESS) {





×







1239



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





×







1240



  }











UNCOV

1241



}




×







1242













UNCOV

1243



void doBuildForceFillResult(SOperatorInfo* pOperator, SStreamFillSupporter* pFillSup, SStreamFillInfo* pFillInfo,




×







1244



                            SSDataBlock* pBlock, SGroupResInfo* pGroupResInfo) {











UNCOV

1245



  blockDataCleanup(pBlock);




×








UNCOV

1246



  if (!hasRemainResults(pGroupResInfo)) {





×








UNCOV

1247



    return;




×







1248



  }










1249












1250



  // clear the existed group id











UNCOV

1251



  pBlock->info.id.groupId = 0;




×








UNCOV

1252



  doBuildForceFillResultImpl(pOperator, pFillSup, pFillInfo, pBlock, pGroupResInfo);




×







1253



}










1254













UNCOV

1255



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




×








UNCOV

1256



  int32_t                  code = TSDB_CODE_SUCCESS;




×








UNCOV

1257



  int32_t                  lino = 0;




×








UNCOV

1258



  SStreamFillOperatorInfo* pInfo = pOperator->info;




×








UNCOV

1259



  uint16_t                 opType = pOperator->operatorType;




×








UNCOV

1260



  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;




×







1261













UNCOV

1262



  doBuildForceFillResult(pOperator, pInfo->pFillSup, pInfo->pFillInfo, pInfo->pRes, &pInfo->groupResInfo);




×








UNCOV

1263



  if (pInfo->pRes->info.rows != 0) {





×








UNCOV

1264



    printDataBlock(pInfo->pRes, getStreamOpName(opType), GET_TASKID(pTaskInfo));




×








UNCOV

1265



    (*ppRes) = pInfo->pRes;




×








UNCOV

1266



    goto _end;




×







1267



  }










1268













UNCOV

1269



  (*ppRes) = NULL;




×







1270













UNCOV

1271



_end:




×








UNCOV

1272



  if (code != TSDB_CODE_SUCCESS) {





×







1273



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





×







1274



  }











UNCOV

1275



  return code;




×







1276



}










1277













UNCOV

1278



static void keepResultInStateBuf(SStreamFillOperatorInfo* pInfo, uint64_t groupId, SResultRowData* pRow) {




×








UNCOV

1279



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

1280



  int32_t lino = 0;




×







1281













UNCOV

1282



  SWinKey      key = {.groupId = groupId, .ts = pRow->key};




×








UNCOV

1283



  int32_t      curVLen = 0;




×








UNCOV

1284



  SRowBuffPos* pStatePos = NULL;




×








UNCOV

1285



  int32_t      winCode = TSDB_CODE_SUCCESS;




×








UNCOV

1286



  code = pInfo->stateStore.streamStateFillAddIfNotExist(pInfo->pState, &key, (void**)&pStatePos,




×







1287



                                                        &curVLen, &winCode);











UNCOV

1288



  QUERY_CHECK_CODE(code, lino, _end);





×








UNCOV

1289



  memcpy(pStatePos->pRowBuff, pRow->pRowVal, pInfo->pFillSup->rowSize);




×








UNCOV

1290



  qDebug("===stream===fill operator save key ts:%" PRId64 " group id:%" PRIu64 "  code:%d", key.ts, key.groupId, code);





×







1291












1292



_end:




×








UNCOV

1293



  if (code != TSDB_CODE_SUCCESS) {





×







1294



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





×







1295



  }











UNCOV

1296



}




×







1297













UNCOV

1298



int32_t keepBlockRowInStateBuf(SStreamFillOperatorInfo* pInfo, SStreamFillInfo* pFillInfo, SSDataBlock* pBlock, TSKEY* tsCol,




×







1299



                               int32_t rowId, uint64_t groupId, int32_t rowSize) {











UNCOV

1300



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

1301



  int32_t lino = 0;




×








UNCOV

1302



  TSKEY ts = tsCol[rowId];




×








UNCOV

1303



  pFillInfo->nextRowKey = ts;




×








UNCOV

1304



  TAOS_MEMSET(pFillInfo->pTempBuff, 0, rowSize);




×








UNCOV

1305



  SResultRowData tmpNextRow = {.key = ts, .pRowVal = pFillInfo->pTempBuff};




×







1306













UNCOV

1307



  transBlockToResultRow(pBlock, rowId, ts, &tmpNextRow);




×








UNCOV

1308



  keepResultInStateBuf(pInfo, groupId, &tmpNextRow);




×







1309













UNCOV

1310



_end:




×








UNCOV

1311



  if (code != TSDB_CODE_SUCCESS) {





×







1312



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





×







1313



  }











UNCOV

1314



  return code;




×







1315



}










1316












1317



// force window close impl











UNCOV

1318



static int32_t doStreamForceFillImpl(SOperatorInfo* pOperator) {




×








UNCOV

1319



  int32_t                  code = TSDB_CODE_SUCCESS;




×








UNCOV

1320



  int32_t                  lino = 0;




×








UNCOV

1321



  SStreamFillOperatorInfo* pInfo = pOperator->info;




×








UNCOV

1322



  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

1323



  SStreamFillSupporter*    pFillSup = pInfo->pFillSup;




×








UNCOV

1324



  SStreamFillInfo*         pFillInfo = pInfo->pFillInfo;




×








UNCOV

1325



  SSDataBlock*             pBlock = pInfo->pSrcBlock;




×








UNCOV

1326



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




×








UNCOV

1327



  SColumnInfoData*         pTsCol = taosArrayGet(pInfo->pSrcBlock->pDataBlock, pInfo->primaryTsCol);




×








UNCOV

1328



  TSKEY*                   tsCol = (TSKEY*)pTsCol->pData;




×








UNCOV

1329



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





×








UNCOV

1330



    code = keepBlockRowInStateBuf(pInfo, pFillInfo, pBlock, tsCol, i, groupId, pFillSup->rowSize);




×








UNCOV

1331



    QUERY_CHECK_CODE(code, lino, _end);





×







1332













UNCOV

1333



    int32_t size =  taosArrayGetSize(pInfo->pCloseTs);




×








UNCOV

1334



    if (size > 0) {





×








UNCOV

1335



      TSKEY* pTs = (TSKEY*) taosArrayGet(pInfo->pCloseTs, 0);




×








UNCOV

1336



      TSKEY  resTs = tsCol[i];




×








UNCOV

1337



      while (resTs < (*pTs)) {





×








UNCOV

1338



        SWinKey key = {.groupId = groupId, .ts = resTs};




×








UNCOV

1339



        void* pPushRes = taosArrayPush(pInfo->pUpdated, &key);




×








UNCOV

1340



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





×







1341













UNCOV

1342



        if (IS_FILL_CONST_VALUE(pFillSup->type)) {





×







1343



          break;










1344



        }











UNCOV

1345



        resTs = taosTimeAdd(resTs, pFillSup->interval.sliding, pFillSup->interval.slidingUnit,




×








UNCOV

1346



                            pFillSup->interval.precision, NULL);




×







1347



      }










1348



    }










1349



  }











UNCOV

1350



  code = pInfo->stateStore.streamStateGroupPut(pInfo->pState, groupId, NULL, 0);




×








UNCOV

1351



  QUERY_CHECK_CODE(code, lino, _end);





×







1352













UNCOV

1353



_end:




×








UNCOV

1354



  if (code != TSDB_CODE_SUCCESS) {





×







1355



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





×







1356



  }











UNCOV

1357



  return code;




×







1358



}










1359













UNCOV

1360



int32_t buildAllResultKey(SStateStore* pStateStore, SStreamState* pState, TSKEY ts, SArray* pUpdated) {




×








UNCOV

1361



  int32_t          code = TSDB_CODE_SUCCESS;




×








UNCOV

1362



  int32_t          lino = 0;




×








UNCOV

1363



  int64_t          groupId = 0;




×








UNCOV

1364



  SStreamStateCur* pCur = pStateStore->streamStateGroupGetCur(pState);




×








UNCOV

1365



  while (1) {  




×








UNCOV

1366



    int32_t winCode = pStateStore->streamStateGroupGetKVByCur(pCur, &groupId, NULL, NULL);




×








UNCOV

1367



    if (winCode != TSDB_CODE_SUCCESS) {





×








UNCOV

1368



      break;




×







1369



    }











UNCOV

1370



    SWinKey key = {.ts = ts, .groupId = groupId};




×








UNCOV

1371



    void* pPushRes = taosArrayPush(pUpdated, &key);




×








UNCOV

1372



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





×







1373













UNCOV

1374



    pStateStore->streamStateGroupCurNext(pCur);




×







1375



  }











UNCOV

1376



  pStateStore->streamStateFreeCur(pCur);




×








UNCOV

1377



  pCur = NULL;




×







1378













UNCOV

1379



_end:




×








UNCOV

1380



  if (code != TSDB_CODE_SUCCESS) {





×







1381



    pStateStore->streamStateFreeCur(pCur);




×







1382



    pCur = NULL;




×







1383



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





×







1384



  }











UNCOV

1385



  return code;




×







1386



}










1387













UNCOV

1388



static void removeDuplicateResult(SArray* pTsArrray, __compar_fn_t fn) {




×








UNCOV

1389



  taosArraySort(pTsArrray, fn);




×








UNCOV

1390



  taosArrayRemoveDuplicate(pTsArrray, fn, NULL);




×








UNCOV

1391



}




×







1392












1393



// force window close











UNCOV

1394



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




×








UNCOV

1395



  int32_t                  code = TSDB_CODE_SUCCESS;




×








UNCOV

1396



  int32_t                  lino = 0;




×








UNCOV

1397



  SStreamFillOperatorInfo* pInfo = pOperator->info;




×








UNCOV

1398



  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;




×







1399













UNCOV

1400



  if (pOperator->status == OP_EXEC_DONE) {





×







1401



    (*ppRes) = NULL;




×







1402



    return code;




×







1403



  }










1404













UNCOV

1405



  if (pOperator->status == OP_RES_TO_RETURN) {





×








UNCOV

1406



    SSDataBlock* resBlock = NULL;




×








UNCOV

1407



    code = buildForceFillResult(pOperator, &resBlock);




×








UNCOV

1408



    QUERY_CHECK_CODE(code, lino, _end);





×







1409













UNCOV

1410



    if (resBlock != NULL) {





×








UNCOV

1411



      (*ppRes) = resBlock;




×








UNCOV

1412



      goto _end;




×







1413



    }










1414













UNCOV

1415



    pInfo->stateStore.streamStateClearExpiredState(pInfo->pState);




×








UNCOV

1416



    resetStreamFillInfo(pInfo);




×








UNCOV

1417



    setStreamOperatorCompleted(pOperator);




×








UNCOV

1418



    (*ppRes) = NULL;




×








UNCOV

1419



    goto _end;




×







1420



  }










1421













UNCOV

1422



  SSDataBlock*   fillResult = NULL;




×








UNCOV

1423



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




×








UNCOV

1424



  while (1) {




×








UNCOV

1425



    SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);




×








UNCOV

1426



    if (pBlock == NULL) {





×








UNCOV

1427



      pOperator->status = OP_RES_TO_RETURN;




×








UNCOV

1428



      qDebug("===stream===return data:%s.", getStreamOpName(pOperator->operatorType));





×








UNCOV

1429



      break;




×







1430



    }











UNCOV

1431



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




×








UNCOV

1432



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




×







1433













UNCOV

1434



    switch (pBlock->info.type) {





×








UNCOV

1435



      case STREAM_NORMAL:




×







1436



      case STREAM_INVALID: {











UNCOV

1437



        code = doApplyStreamScalarCalculation(pOperator, pBlock, pInfo->pSrcBlock);




×








UNCOV

1438



        QUERY_CHECK_CODE(code, lino, _end);





×







1439













UNCOV

1440



        memcpy(pInfo->pSrcBlock->info.parTbName, pBlock->info.parTbName, TSDB_TABLE_NAME_LEN);




×








UNCOV

1441



        pInfo->srcRowIndex = -1;




×








UNCOV

1442



      } break;




×








UNCOV

1443



      case STREAM_CHECKPOINT: {




×








UNCOV

1444



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




×








UNCOV

1445



        (*ppRes) = pBlock;




×








UNCOV

1446



        goto _end;




×







1447



      } break;











UNCOV

1448



      case STREAM_CREATE_CHILD_TABLE: {




×








UNCOV

1449



        (*ppRes) = pBlock;




×








UNCOV

1450



        goto _end;




×







1451



      } break;











UNCOV

1452



      case STREAM_GET_RESULT: {




×








UNCOV

1453



        void* pPushRes = taosArrayPush(pInfo->pCloseTs, &pBlock->info.window.skey);




×








UNCOV

1454



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





×








UNCOV

1455



        continue;




×







1456



      }










1457



      default:




×







1458



        code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;




×







1459



        QUERY_CHECK_CODE(code, lino, _end);





×







1460



    }










1461













UNCOV

1462



    code = doStreamForceFillImpl(pOperator);




×








UNCOV

1463



    QUERY_CHECK_CODE(code, lino, _end);





×







1464



  }










1465













UNCOV

1466



  for (int32_t i = 0; i < taosArrayGetSize(pInfo->pCloseTs); i++) {





×








UNCOV

1467



    TSKEY ts = *(TSKEY*) taosArrayGet(pInfo->pCloseTs, i);




×








UNCOV

1468



    code = buildAllResultKey(&pInfo->stateStore, pInfo->pState, ts, pInfo->pUpdated);




×








UNCOV

1469



    QUERY_CHECK_CODE(code, lino, _end);





×







1470



  }











UNCOV

1471



  taosArrayClear(pInfo->pCloseTs);




×








UNCOV

1472



  removeDuplicateResult(pInfo->pUpdated, winKeyCmprImpl);




×







1473













UNCOV

1474



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




×








UNCOV

1475



  pInfo->groupResInfo.freeItem = false;




×







1476













UNCOV

1477



  pInfo->pUpdated = taosArrayInit(1024, sizeof(SWinKey));




×








UNCOV

1478



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





×







1479













UNCOV

1480



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




×








UNCOV

1481



  QUERY_CHECK_CODE(code, lino, _end);





×







1482













UNCOV

1483



  code = buildForceFillResult(pOperator, ppRes);




×








UNCOV

1484



  QUERY_CHECK_CODE(code, lino, _end);





×







1485













UNCOV

1486



  if ((*ppRes) == NULL) {





×








UNCOV

1487



    pInfo->stateStore.streamStateClearExpiredState(pInfo->pState);




×








UNCOV

1488



    resetStreamFillInfo(pInfo);




×








UNCOV

1489



    setStreamOperatorCompleted(pOperator);




×







1490



  }










1491













UNCOV

1492



_end:




×








UNCOV

1493



  if (code != TSDB_CODE_SUCCESS) {





×







1494



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





×







1495



    pTaskInfo->code = code;




×







1496



  }











UNCOV

1497



  return code;




×







1498



}










1499













UNCOV

1500



static int32_t initResultBuf(SSDataBlock* pInputRes, SStreamFillSupporter* pFillSup) {




×








UNCOV

1501



  int32_t numOfCols = taosArrayGetSize(pInputRes->pDataBlock);




×








UNCOV

1502



  pFillSup->rowSize = sizeof(SResultCellData) * numOfCols;




×








UNCOV

1503



  for (int i = 0; i < numOfCols; i++) {





×








UNCOV

1504



    SColumnInfoData* pCol = taosArrayGet(pInputRes->pDataBlock, i);




×








UNCOV

1505



    pFillSup->rowSize += pCol->info.bytes;




×







1506



  }











UNCOV

1507



  pFillSup->next.key = INT64_MIN;




×








UNCOV

1508



  pFillSup->nextNext.key = INT64_MIN;




×








UNCOV

1509



  pFillSup->prev.key = INT64_MIN;




×








UNCOV

1510



  pFillSup->cur.key = INT64_MIN;




×








UNCOV

1511



  pFillSup->next.pRowVal = NULL;




×








UNCOV

1512



  pFillSup->nextNext.pRowVal = NULL;




×








UNCOV

1513



  pFillSup->prev.pRowVal = NULL;




×








UNCOV

1514



  pFillSup->cur.pRowVal = NULL;




×







1515













UNCOV

1516



  return TSDB_CODE_SUCCESS;




×







1517



}










1518













UNCOV

1519



static SStreamFillSupporter* initStreamFillSup(SStreamFillPhysiNode* pPhyFillNode, SInterval* pInterval,




×







1520



                                               SExprInfo* pFillExprInfo, int32_t numOfFillCols, SStorageAPI* pAPI, SSDataBlock* pInputRes) {











UNCOV

1521



  int32_t               code = TSDB_CODE_SUCCESS;




×








UNCOV

1522



  int32_t               lino = 0;




×








UNCOV

1523



  SStreamFillSupporter* pFillSup = taosMemoryCalloc(1, sizeof(SStreamFillSupporter));





×








UNCOV

1524



  if (!pFillSup) {





×







1525



    code = terrno;




×







1526



    QUERY_CHECK_CODE(code, lino, _end);





×







1527



  }











UNCOV

1528



  pFillSup->numOfFillCols = numOfFillCols;




×








UNCOV

1529



  int32_t    numOfNotFillCols = 0;




×








UNCOV

1530



  SExprInfo* noFillExprInfo = NULL;




×







1531













UNCOV

1532



  code = createExprInfo(pPhyFillNode->pNotFillExprs, NULL, &noFillExprInfo, &numOfNotFillCols);




×








UNCOV

1533



  QUERY_CHECK_CODE(code, lino, _end);





×







1534













UNCOV

1535



  pFillSup->pAllColInfo = createFillColInfo(pFillExprInfo, pFillSup->numOfFillCols, noFillExprInfo, numOfNotFillCols,




×








UNCOV

1536



                                            NULL, 0, (const SNodeListNode*)(pPhyFillNode->pValues));




×








UNCOV

1537



  if (pFillSup->pAllColInfo == NULL) {





×







1538



    code = terrno;




×







1539



    lino = __LINE__;




×







1540



    destroyExprInfo(noFillExprInfo, numOfNotFillCols);




×







1541



    goto _end;




×







1542



  }










1543













UNCOV

1544



  pFillSup->type = convertFillType(pPhyFillNode->mode);




×








UNCOV

1545



  pFillSup->numOfAllCols = pFillSup->numOfFillCols + numOfNotFillCols;




×








UNCOV

1546



  pFillSup->interval = *pInterval;




×








UNCOV

1547



  pFillSup->pAPI = pAPI;




×







1548













UNCOV

1549



  code = initResultBuf(pInputRes, pFillSup);




×








UNCOV

1550



  QUERY_CHECK_CODE(code, lino, _end);





×







1551













UNCOV

1552



  SExprInfo* noFillExpr = NULL;




×








UNCOV

1553



  code = createExprInfo(pPhyFillNode->pNotFillExprs, NULL, &noFillExpr, &numOfNotFillCols);




×








UNCOV

1554



  QUERY_CHECK_CODE(code, lino, _end);





×







1555













UNCOV

1556



  code = initExprSupp(&pFillSup->notFillExprSup, noFillExpr, numOfNotFillCols, &pAPI->functionStore);




×








UNCOV

1557



  QUERY_CHECK_CODE(code, lino, _end);





×







1558













UNCOV

1559



  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);




×








UNCOV

1560



  pFillSup->pResMap = tSimpleHashInit(16, hashFn);




×








UNCOV

1561



  QUERY_CHECK_NULL(pFillSup->pResMap, code, lino, _end, terrno);





×








UNCOV

1562



  pFillSup->hasDelete = false;




×







1563













UNCOV

1564



_end:




×








UNCOV

1565



  if (code != TSDB_CODE_SUCCESS) {





×







1566



    destroyStreamFillSupporter(pFillSup);




×







1567



    pFillSup = NULL;




×







1568



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





×







1569



  }











UNCOV

1570



  return pFillSup;




×







1571



}










1572













UNCOV

1573



SStreamFillInfo* initStreamFillInfo(SStreamFillSupporter* pFillSup, SSDataBlock* pRes) {




×








UNCOV

1574



  int32_t          code = TSDB_CODE_SUCCESS;




×








UNCOV

1575



  int32_t          lino = 0;




×








UNCOV

1576



  SStreamFillInfo* pFillInfo = taosMemoryCalloc(1, sizeof(SStreamFillInfo));





×








UNCOV

1577



  if (!pFillInfo) {





×







1578



    code = terrno;




×







1579



    QUERY_CHECK_CODE(code, lino, _end);





×







1580



  }










1581













UNCOV

1582



  pFillInfo->start = INT64_MIN;




×








UNCOV

1583



  pFillInfo->current = INT64_MIN;




×








UNCOV

1584



  pFillInfo->end = INT64_MIN;




×








UNCOV

1585



  pFillInfo->preRowKey = INT64_MIN;




×








UNCOV

1586



  pFillInfo->needFill = false;




×








UNCOV

1587



  pFillInfo->pLinearInfo = taosMemoryCalloc(1, sizeof(SStreamFillLinearInfo));





×








UNCOV

1588



  if (!pFillInfo) {





×







1589



    code = terrno;




×







1590



    QUERY_CHECK_CODE(code, lino, _end);





×







1591



  }










1592













UNCOV

1593



  pFillInfo->pLinearInfo->hasNext = false;




×








UNCOV

1594



  pFillInfo->pLinearInfo->nextEnd = INT64_MIN;




×








UNCOV

1595



  pFillInfo->pLinearInfo->pEndPoints = NULL;




×








UNCOV

1596



  pFillInfo->pLinearInfo->pNextEndPoints = NULL;




×








UNCOV

1597



  if (pFillSup->type == TSDB_FILL_LINEAR) {





×








UNCOV

1598



    pFillInfo->pLinearInfo->pEndPoints = taosArrayInit(pFillSup->numOfAllCols, sizeof(SPoint));




×








UNCOV

1599



    if (!pFillInfo->pLinearInfo->pEndPoints) {





×







1600



      code = terrno;




×







1601



      QUERY_CHECK_CODE(code, lino, _end);





×







1602



    }










1603













UNCOV

1604



    pFillInfo->pLinearInfo->pNextEndPoints = taosArrayInit(pFillSup->numOfAllCols, sizeof(SPoint));




×








UNCOV

1605



    if (!pFillInfo->pLinearInfo->pNextEndPoints) {





×







1606



      code = terrno;




×







1607



      QUERY_CHECK_CODE(code, lino, _end);





×







1608



    }










1609













UNCOV

1610



    for (int32_t i = 0; i < pFillSup->numOfAllCols; i++) {





×








UNCOV

1611



      SColumnInfoData* pColData = taosArrayGet(pRes->pDataBlock, i);




×








UNCOV

1612



      if (pColData == NULL) {





×








UNCOV

1613



        SPoint dummy = {0};




×








UNCOV

1614



        dummy.val = taosMemoryCalloc(1, 1);





×








UNCOV

1615



        void* tmpRes = taosArrayPush(pFillInfo->pLinearInfo->pEndPoints, &dummy);




×








UNCOV

1616



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





×







1617













UNCOV

1618



        dummy.val = taosMemoryCalloc(1, 1);





×








UNCOV

1619



        tmpRes = taosArrayPush(pFillInfo->pLinearInfo->pNextEndPoints, &dummy);




×








UNCOV

1620



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





×







1621













UNCOV

1622



        continue;




×







1623



      }











UNCOV

1624



      SPoint value = {0};




×








UNCOV

1625



      value.val = taosMemoryCalloc(1, pColData->info.bytes);





×








UNCOV

1626



      QUERY_CHECK_NULL(value.val, code, lino, _end, terrno);





×







1627













UNCOV

1628



      void* tmpRes = taosArrayPush(pFillInfo->pLinearInfo->pEndPoints, &value);




×








UNCOV

1629



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





×







1630













UNCOV

1631



      value.val = taosMemoryCalloc(1, pColData->info.bytes);





×








UNCOV

1632



      QUERY_CHECK_NULL(value.val, code, lino, _end, terrno);





×







1633













UNCOV

1634



      tmpRes = taosArrayPush(pFillInfo->pLinearInfo->pNextEndPoints, &value);




×








UNCOV

1635



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





×







1636



    }










1637



  }











UNCOV

1638



  pFillInfo->pLinearInfo->winIndex = 0;




×







1639













UNCOV

1640



  pFillInfo->pNonFillRow = NULL;




×








UNCOV

1641



  pFillInfo->pResRow = NULL;




×








UNCOV

1642



  if (pFillSup->type == TSDB_FILL_SET_VALUE || pFillSup->type == TSDB_FILL_SET_VALUE_F ||





×








UNCOV

1643



      pFillSup->type == TSDB_FILL_NULL || pFillSup->type == TSDB_FILL_NULL_F) {





×








UNCOV

1644



    pFillInfo->pResRow = taosMemoryCalloc(1, sizeof(SResultRowData));





×








UNCOV

1645



    QUERY_CHECK_NULL(pFillInfo->pResRow, code, lino, _end, terrno);





×







1646













UNCOV

1647



    pFillInfo->pResRow->key = INT64_MIN;




×








UNCOV

1648



    pFillInfo->pResRow->pRowVal = taosMemoryCalloc(1, pFillSup->rowSize);





×








UNCOV

1649



    QUERY_CHECK_NULL(pFillInfo->pResRow->pRowVal, code, lino, _end, terrno);





×







1650













UNCOV

1651



    for (int32_t i = 0; i < pFillSup->numOfAllCols; ++i) {





×








UNCOV

1652



      SColumnInfoData* pColData = taosArrayGet(pRes->pDataBlock, i);




×








UNCOV

1653



      SResultCellData* pCell = getResultCell(pFillInfo->pResRow, i);




×








UNCOV

1654



      if (pColData == NULL) {





×








UNCOV

1655



        pCell->bytes = 1;




×








UNCOV

1656



        pCell->type = 4;




×








UNCOV

1657



        continue;




×







1658



      }











UNCOV

1659



      pCell->bytes = pColData->info.bytes;




×








UNCOV

1660



      pCell->type = pColData->info.type;




×







1661



    }










1662













UNCOV

1663



    int32_t numOfResCol = taosArrayGetSize(pRes->pDataBlock);




×








UNCOV

1664



    if (numOfResCol < pFillSup->numOfAllCols) {





×








UNCOV

1665



      int32_t* pTmpBuf = (int32_t*)taosMemoryRealloc(pFillSup->pOffsetInfo, pFillSup->numOfAllCols * sizeof(int32_t));





×








UNCOV

1666



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





×








UNCOV

1667



      pFillSup->pOffsetInfo = pTmpBuf;




×







1668













UNCOV

1669



      SResultCellData* pCell = getResultCell(pFillInfo->pResRow, numOfResCol - 1);




×








UNCOV

1670



      int32_t preLength = pFillSup->pOffsetInfo[numOfResCol - 1] + pCell->bytes + sizeof(SResultCellData);




×








UNCOV

1671



      for (int32_t i = numOfResCol; i < pFillSup->numOfAllCols; i++) {





×








UNCOV

1672



        pFillSup->pOffsetInfo[i] = preLength;




×








UNCOV

1673



        pCell = getResultCell(pFillInfo->pResRow, i);




×








UNCOV

1674



        preLength += pCell->bytes + sizeof(SResultCellData);




×







1675



      }










1676



    }










1677













UNCOV

1678



    pFillInfo->pNonFillRow = taosMemoryCalloc(1, sizeof(SResultRowData));





×








UNCOV

1679



    QUERY_CHECK_NULL(pFillInfo->pNonFillRow, code, lino, _end, terrno);





×








UNCOV

1680



    pFillInfo->pNonFillRow->key = INT64_MIN;




×








UNCOV

1681



    pFillInfo->pNonFillRow->pRowVal = taosMemoryCalloc(1, pFillSup->rowSize);





×








UNCOV

1682



    memcpy(pFillInfo->pNonFillRow->pRowVal, pFillInfo->pResRow->pRowVal, pFillSup->rowSize);




×







1683



  }










1684













UNCOV

1685



  pFillInfo->type = pFillSup->type;




×








UNCOV

1686



  pFillInfo->delRanges = taosArrayInit(16, sizeof(STimeRange));




×








UNCOV

1687



  if (!pFillInfo->delRanges) {





×







1688



    code = terrno;




×







1689



    QUERY_CHECK_CODE(code, lino, _end);





×







1690



  }










1691













UNCOV

1692



  pFillInfo->delIndex = 0;




×








UNCOV

1693



  pFillInfo->curGroupId = 0;




×








UNCOV

1694



  pFillInfo->hasNext = false;




×








UNCOV

1695



  pFillInfo->pTempBuff = taosMemoryCalloc(1, pFillSup->rowSize);





×








UNCOV

1696



  return pFillInfo;




×







1697












1698



_end:




×







1699



  if (code != TSDB_CODE_SUCCESS) {





×







1700



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





×







1701



  }










1702



  destroyStreamFillInfo(pFillInfo);




×







1703



  return NULL;




×







1704



}










1705













UNCOV

1706



static void setValueForFillInfo(SStreamFillSupporter* pFillSup, SStreamFillInfo* pFillInfo) {




×








UNCOV

1707



  if (pFillInfo->type == TSDB_FILL_SET_VALUE || pFillInfo->type == TSDB_FILL_SET_VALUE_F) {





×








UNCOV

1708



    for (int32_t i = 0; i < pFillSup->numOfAllCols; ++i) {





×








UNCOV

1709



      SFillColInfo*    pFillCol = pFillSup->pAllColInfo + i;




×








UNCOV

1710



      int32_t          slotId = GET_DEST_SLOT_ID(pFillCol);




×








UNCOV

1711



      SResultCellData* pCell = getResultCell(pFillInfo->pResRow, slotId);




×








UNCOV

1712



      SVariant*        pVar = &(pFillCol->fillVal);




×








UNCOV

1713



      if (pCell->type == TSDB_DATA_TYPE_FLOAT) {





×







1714



        float v = 0;




×







1715



        GET_TYPED_DATA(v, float, pVar->nType, &pVar->i);





×







1716



        SET_TYPED_DATA(pCell->pData, pCell->type, v);





×








UNCOV

1717



      } else if (IS_FLOAT_TYPE(pCell->type)) {





×








UNCOV

1718



        double v = 0;




×








UNCOV

1719



        GET_TYPED_DATA(v, double, pVar->nType, &pVar->i);





×








UNCOV

1720



        SET_TYPED_DATA(pCell->pData, pCell->type, v);





×








UNCOV

1721



      } else if (IS_INTEGER_TYPE(pCell->type)) {





×








UNCOV

1722



        int64_t v = 0;




×








UNCOV

1723



        GET_TYPED_DATA(v, int64_t, pVar->nType, &pVar->i);





×








UNCOV

1724



        SET_TYPED_DATA(pCell->pData, pCell->type, v);





×







1725



      } else {











UNCOV

1726



        pCell->isNull = true;




×







1727



      }










1728



    }











UNCOV

1729



  } else if (pFillInfo->type == TSDB_FILL_NULL || pFillInfo->type == TSDB_FILL_NULL_F) {





×








UNCOV

1730



    for (int32_t i = 0; i < pFillSup->numOfAllCols; ++i) {





×








UNCOV

1731



      SFillColInfo*    pFillCol = pFillSup->pAllColInfo + i;




×








UNCOV

1732



      int32_t          slotId = GET_DEST_SLOT_ID(pFillCol);




×








UNCOV

1733



      SResultCellData* pCell = getResultCell(pFillInfo->pResRow, slotId);




×








UNCOV

1734



      pCell->isNull = true;




×







1735



    }










1736



  }











UNCOV

1737



}




×







1738













UNCOV

1739



int32_t getDownStreamInfo(SOperatorInfo* downstream, int8_t* triggerType, SInterval* pInterval) {




×








UNCOV

1740



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

1741



  int32_t lino = 0;




×








UNCOV

1742



  if (IS_NORMAL_INTERVAL_OP(downstream)) {





×








UNCOV

1743



    SStreamIntervalOperatorInfo* pInfo = downstream->info;




×








UNCOV

1744



    *triggerType = pInfo->twAggSup.calTrigger;




×








UNCOV

1745



    *pInterval = pInfo->interval;




×








UNCOV

1746



  } else if (IS_CONTINUE_INTERVAL_OP(downstream)) {





×








UNCOV

1747



    SStreamIntervalSliceOperatorInfo* pInfo = downstream->info;




×








UNCOV

1748



    *triggerType = pInfo->twAggSup.calTrigger;




×








UNCOV

1749



    *pInterval = pInfo->interval;




×








UNCOV

1750



    pInfo->hasFill = true;




×







1751



  } else {










1752



    code = TSDB_CODE_STREAM_INTERNAL_ERROR;




×







1753



  }











UNCOV

1754



  QUERY_CHECK_CODE(code, lino, _end);





×







1755



  











UNCOV

1756



_end:




×








UNCOV

1757



  if (code != TSDB_CODE_SUCCESS) {





×







1758



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





×







1759



  }











UNCOV

1760



  return code;




×







1761



}










1762













UNCOV

1763



int32_t initFillOperatorStateBuff(SStreamFillOperatorInfo* pInfo, SStreamState* pState, SStateStore* pStore,




×







1764



                                  SReadHandle* pHandle, const char* taskIdStr, SStorageAPI* pApi) {











UNCOV

1765



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

1766



  int32_t lino = 0;




×







1767













UNCOV

1768



  pInfo->stateStore = *pStore;




×








UNCOV

1769



  pInfo->pState = taosMemoryCalloc(1, sizeof(SStreamState));





×








UNCOV

1770



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





×







1771













UNCOV

1772



  *(pInfo->pState) = *pState;




×








UNCOV

1773



  pInfo->stateStore.streamStateSetNumber(pInfo->pState, -1, pInfo->primaryTsCol);




×








UNCOV

1774



  code = pInfo->stateStore.streamFileStateInit(tsStreamBufferSize, sizeof(SWinKey), pInfo->pFillSup->rowSize, 0, compareTs,




×








UNCOV

1775



                                               pInfo->pState, INT64_MAX, taskIdStr, pHandle->checkpointId,




×








UNCOV

1776



                                               STREAM_STATE_BUFF_HASH_SORT, &pInfo->pState->pFileState);




×








UNCOV

1777



  QUERY_CHECK_CODE(code, lino, _end);





×







1778













UNCOV

1779



_end:




×








UNCOV

1780



  if (code != TSDB_CODE_SUCCESS) {





×







1781



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





×







1782



  }











UNCOV

1783



  return code;




×







1784



}










1785













UNCOV

1786



int32_t createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFillPhysiNode* pPhyFillNode,




×







1787



                                     SExecTaskInfo* pTaskInfo, SReadHandle* pHandle, SOperatorInfo** pOptrInfo) {











UNCOV

1788



  QRY_PARAM_CHECK(pOptrInfo);





×







1789













UNCOV

1790



  int32_t                  code = TSDB_CODE_SUCCESS;




×








UNCOV

1791



  int32_t                  lino = 0;




×








UNCOV

1792



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





×








UNCOV

1793



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





×








UNCOV

1794



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





×







1795



    code = terrno;




×







1796



    QUERY_CHECK_CODE(code, lino, _error);





×







1797



  }










1798













UNCOV

1799



  int32_t    numOfFillCols = 0;




×








UNCOV

1800



  SExprInfo* pFillExprInfo = NULL;




×







1801













UNCOV

1802



  code = createExprInfo(pPhyFillNode->pFillExprs, NULL, &pFillExprInfo, &numOfFillCols);




×








UNCOV

1803



  QUERY_CHECK_CODE(code, lino, _error);





×







1804













UNCOV

1805



  code = initExprSupp(&pOperator->exprSupp, pFillExprInfo, numOfFillCols, &pTaskInfo->storageAPI.functionStore);




×








UNCOV

1806



  QUERY_CHECK_CODE(code, lino, _error);





×







1807













UNCOV

1808



  pInfo->pSrcBlock = createDataBlockFromDescNode(pPhyFillNode->node.pOutputDataBlockDesc);




×








UNCOV

1809



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





×







1810













UNCOV

1811



  int8_t triggerType = 0;




×








UNCOV

1812



  SInterval interval = {0};




×








UNCOV

1813



  code = getDownStreamInfo(downstream, &triggerType, &interval);




×








UNCOV

1814



  QUERY_CHECK_CODE(code, lino, _error);





×







1815













UNCOV

1816



  pInfo->pFillSup = initStreamFillSup(pPhyFillNode, &interval, pFillExprInfo, numOfFillCols, &pTaskInfo->storageAPI,




×







1817



                                      pInfo->pSrcBlock);











UNCOV

1818



  if (!pInfo->pFillSup) {





×







1819



    code = TSDB_CODE_FAILED;




×







1820



    QUERY_CHECK_CODE(code, lino, _error);





×







1821



  }










1822













UNCOV

1823



  initResultSizeInfo(&pOperator->resultInfo, 4096);




×








UNCOV

1824



  pInfo->pRes = createDataBlockFromDescNode(pPhyFillNode->node.pOutputDataBlockDesc);




×








UNCOV

1825



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





×







1826













UNCOV

1827



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




×








UNCOV

1828



  QUERY_CHECK_CODE(code, lino, _error);





×







1829













UNCOV

1830



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




×








UNCOV

1831



  QUERY_CHECK_CODE(code, lino, _error);





×







1832













UNCOV

1833



  pInfo->pFillInfo = initStreamFillInfo(pInfo->pFillSup, pInfo->pRes);




×








UNCOV

1834



  if (!pInfo->pFillInfo) {





×







1835



    goto _error;




×







1836



  }










1837













UNCOV

1838



  setValueForFillInfo(pInfo->pFillSup, pInfo->pFillInfo);




×







1839













UNCOV

1840



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




×








UNCOV

1841



  QUERY_CHECK_CODE(code, lino, _error);





×







1842













UNCOV

1843



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




×








UNCOV

1844



  QUERY_CHECK_CODE(code, lino, _error);





×







1845













UNCOV

1846



  pInfo->pUpdated = taosArrayInit(1024, sizeof(SWinKey));




×








UNCOV

1847



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





×







1848













UNCOV

1849



  pInfo->pCloseTs = taosArrayInit(1024, sizeof(TSKEY));




×








UNCOV

1850



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





×







1851













UNCOV

1852



  pInfo->primaryTsCol = ((STargetNode*)pPhyFillNode->pWStartTs)->slotId;




×








UNCOV

1853



  pInfo->primarySrcSlotId = ((SColumnNode*)((STargetNode*)pPhyFillNode->pWStartTs)->pExpr)->slotId;




×







1854













UNCOV

1855



  int32_t numOfOutputCols = 0;




×








UNCOV

1856



  code = extractColMatchInfo(pPhyFillNode->pFillExprs, pPhyFillNode->node.pOutputDataBlockDesc, &numOfOutputCols,




×







1857



                             COL_MATCH_FROM_SLOT_ID, &pInfo->matchInfo);











UNCOV

1858



  QUERY_CHECK_CODE(code, lino, _error);





×







1859













UNCOV

1860



  code = filterInitFromNode((SNode*)pPhyFillNode->node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);




×








UNCOV

1861



  QUERY_CHECK_CODE(code, lino, _error);





×







1862













UNCOV

1863



  pInfo->srcRowIndex = -1;




×








UNCOV

1864



  setOperatorInfo(pOperator, "StreamFillOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL, false, OP_NOT_OPENED, pInfo,




×







1865



                  pTaskInfo);










1866













UNCOV

1867



  if (triggerType == STREAM_TRIGGER_FORCE_WINDOW_CLOSE) {





×








UNCOV

1868



    code = initFillOperatorStateBuff(pInfo, pTaskInfo->streamInfo.pState, &pTaskInfo->storageAPI.stateStore, pHandle,




×








UNCOV

1869



                              GET_TASKID(pTaskInfo), &pTaskInfo->storageAPI);




×








UNCOV

1870



    QUERY_CHECK_CODE(code, lino, _error);





×








UNCOV

1871



    pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamForceFillNext, NULL, destroyStreamFillOperatorInfo,




×







1872



                                           optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);










1873



  } else {











UNCOV

1874



    pInfo->pState = NULL;




×








UNCOV

1875



    pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamFillNext, NULL, destroyStreamFillOperatorInfo,




×







1876



                                           optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);










1877



  }











UNCOV

1878



  setOperatorStreamStateFn(pOperator, streamOpReleaseState, streamOpReloadState);




×







1879













UNCOV

1880



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




×








UNCOV

1881



  QUERY_CHECK_CODE(code, lino, _error);





×







1882













UNCOV

1883



  *pOptrInfo = pOperator;




×








UNCOV

1884



  return TSDB_CODE_SUCCESS;




×







1885












1886



_error:




×







1887



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





×







1888












1889



  if (pInfo != NULL) destroyStreamFillOperatorInfo(pInfo);





×







1890



  destroyOperatorAndDownstreams(pOperator, &downstream, 1);




×







1891



  pTaskInfo->code = code;




×







1892



  return code;




×







1893



}
























    

  • 

    •