taosdata / TDengine / #3621

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

push

travis-ci

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

merge: from main to 3.0 branch

4357 of 287032 branches covered (1.52%)

Branch coverage included in aggregate %.

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

213359 existing lines in 469 files now uncovered.

7260 of 283369 relevant lines covered (2.56%)

23737.72 hits per line

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

0.0
/source/libs/executor/src/anomalywindowoperator.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 "executorInt.h"










17



#include "filter.h"










18



#include "function.h"










19



#include "functionMgt.h"










20



#include "operator.h"










21



#include "querytask.h"










22



#include "tanalytics.h"










23



#include "tcommon.h"










24



#include "tcompare.h"










25



#include "tdatablock.h"










26



#include "tjson.h"










27



#include "ttime.h"










28












29



#ifdef USE_ANALYTICS










30












31



typedef struct {










32



  SArray*     blocks;   // SSDataBlock*










33



  SArray*     windows;  // STimeWindow










34



  uint64_t    groupId;










35



  int64_t     cachedRows;










36



  int32_t     curWinIndex;










37



  STimeWindow curWin;










38



  SResultRow* pResultRow;










39



} SAnomalyWindowSupp;










40












41



typedef struct {










42



  SOptrBasicInfo     binfo;










43



  SAggSupporter      aggSup;










44



  SExprSupp          scalarSup;










45



  int32_t            tsSlotId;










46



  STimeWindowAggSupp twAggSup;










47



  char               algoName[TSDB_ANALYTIC_ALGO_NAME_LEN];










48



  char               algoUrl[TSDB_ANALYTIC_ALGO_URL_LEN];










49



  char               anomalyOpt[TSDB_ANALYTIC_ALGO_OPTION_LEN];










50



  SAnomalyWindowSupp anomalySup;










51



  SWindowRowsSup     anomalyWinRowSup;










52



  SColumn            anomalyCol;










53



  SStateKeys         anomalyKey;










54



} SAnomalyWindowOperatorInfo;










55












56



static void    anomalyDestroyOperatorInfo(void* param);










57



static int32_t anomalyAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes);










58



static int32_t anomalyAggregateBlocks(SOperatorInfo* pOperator);










59



static int32_t anomalyCacheBlock(SAnomalyWindowOperatorInfo* pInfo, SSDataBlock* pBlock);










60













UNCOV

61



int32_t createAnomalywindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode, SExecTaskInfo* pTaskInfo,




×







62



                                        SOperatorInfo** pOptrInfo) {











UNCOV

63



  QRY_PARAM_CHECK(pOptrInfo);





×








UNCOV

64



  int32_t     code = TSDB_CODE_SUCCESS;




×








UNCOV

65



  int32_t     lino = 0;




×








UNCOV

66



  size_t      keyBufSize = 0;




×








UNCOV

67



  int32_t     num = 0;




×








UNCOV

68



  SExprInfo*  pExprInfo = NULL;




×








UNCOV

69



  const char* id = GET_TASKID(pTaskInfo);




×







70













UNCOV

71



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





×








UNCOV

72



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





×








UNCOV

73



  SAnomalyWindowPhysiNode*    pAnomalyNode = (SAnomalyWindowPhysiNode*)physiNode;




×








UNCOV

74



  SColumnNode*                pColNode = (SColumnNode*)(pAnomalyNode->pAnomalyKey);




×








UNCOV

75



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





×







76



    code = terrno;




×







77



    goto _error;




×







78



  }










79













UNCOV

80



  if (!taosAnalGetOptStr(pAnomalyNode->anomalyOpt, "algo", pInfo->algoName, sizeof(pInfo->algoName))) {





×







81



    qError("%s failed to get anomaly_window algorithm name from %s", id, pAnomalyNode->anomalyOpt);





×







82



    code = TSDB_CODE_ANA_ALGO_NOT_FOUND;




×







83



    goto _error;




×







84



  }










85













UNCOV

86



  if (taosAnalGetAlgoUrl(pInfo->algoName, ANAL_ALGO_TYPE_ANOMALY_DETECT, pInfo->algoUrl, sizeof(pInfo->algoUrl)) != 0) {





×








UNCOV

87



    qError("%s failed to get anomaly_window algorithm url from %s", id, pInfo->algoName);





×








UNCOV

88



    code = TSDB_CODE_ANA_ALGO_NOT_LOAD;




×








UNCOV

89



    goto _error;




×







90



  }










91













UNCOV

92



  pOperator->exprSupp.hasWindowOrGroup = true;




×








UNCOV

93



  pInfo->tsSlotId = ((SColumnNode*)pAnomalyNode->window.pTspk)->slotId;




×








UNCOV

94



  tstrncpy(pInfo->anomalyOpt, pAnomalyNode->anomalyOpt, sizeof(pInfo->anomalyOpt));




×







95













UNCOV

96



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





×







97



    int32_t    numOfScalarExpr = 0;




×







98



    SExprInfo* pScalarExprInfo = NULL;




×







99



    code = createExprInfo(pAnomalyNode->window.pExprs, NULL, &pScalarExprInfo, &numOfScalarExpr);




×







100



    QUERY_CHECK_CODE(code, lino, _error);





×







101












102



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




×







103



    QUERY_CHECK_CODE(code, lino, _error);





×







104



  }










105













UNCOV

106



  code = createExprInfo(pAnomalyNode->window.pFuncs, NULL, &pExprInfo, &num);




×








UNCOV

107



  QUERY_CHECK_CODE(code, lino, _error);





×







108













UNCOV

109



  initResultSizeInfo(&pOperator->resultInfo, 4096);




×







110













UNCOV

111



  code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, keyBufSize, id, pTaskInfo->streamInfo.pState,




×







112



                    &pTaskInfo->storageAPI.functionStore);











UNCOV

113



  QUERY_CHECK_CODE(code, lino, _error);





×







114













UNCOV

115



  SSDataBlock* pResBlock = createDataBlockFromDescNode(pAnomalyNode->window.node.pOutputDataBlockDesc);




×








UNCOV

116



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





×








UNCOV

117



  initBasicInfo(&pInfo->binfo, pResBlock);




×







118













UNCOV

119



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




×








UNCOV

120



  QUERY_CHECK_CODE(code, lino, _error);





×







121













UNCOV

122



  initResultRowInfo(&pInfo->binfo.resultRowInfo);




×








UNCOV

123



  pInfo->binfo.inputTsOrder = pAnomalyNode->window.node.inputTsOrder;




×








UNCOV

124



  pInfo->binfo.outputTsOrder = pAnomalyNode->window.node.outputTsOrder;




×







125













UNCOV

126



  pInfo->anomalyCol = extractColumnFromColumnNode(pColNode);




×








UNCOV

127



  pInfo->anomalyKey.type = pInfo->anomalyCol.type;




×








UNCOV

128



  pInfo->anomalyKey.bytes = pInfo->anomalyCol.bytes;




×







129













UNCOV

130



  pInfo->anomalyKey.pData = taosMemoryCalloc(1, pInfo->anomalyCol.bytes);





×








UNCOV

131



  QUERY_CHECK_NULL(pInfo->anomalyKey.pData, code, lino, _error, terrno)





×







132













UNCOV

133



  int32_t itemSize = sizeof(int32_t) + pInfo->aggSup.resultRowSize + pInfo->anomalyKey.bytes;




×








UNCOV

134



  pInfo->anomalySup.pResultRow = taosMemoryCalloc(1, itemSize);





×








UNCOV

135



  QUERY_CHECK_NULL(pInfo->anomalySup.pResultRow, code, lino, _error, terrno)





×







136













UNCOV

137



  pInfo->anomalySup.blocks = taosArrayInit(16, sizeof(SSDataBlock*));




×








UNCOV

138



  QUERY_CHECK_NULL(pInfo->anomalySup.blocks, code, lino, _error, terrno)





×







139













UNCOV

140



  pInfo->anomalySup.windows = taosArrayInit(16, sizeof(STimeWindow));




×








UNCOV

141



  QUERY_CHECK_NULL(pInfo->anomalySup.windows, code, lino, _error, terrno)





×







142













UNCOV

143



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




×








UNCOV

144



  QUERY_CHECK_CODE(code, lino, _error);





×







145













UNCOV

146



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




×








UNCOV

147



  QUERY_CHECK_CODE(code, lino, _error);





×







148













UNCOV

149



  setOperatorInfo(pOperator, "AnomalyWindowOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_ANOMALY, true, OP_NOT_OPENED,




×







150



                  pInfo, pTaskInfo);











UNCOV

151



  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, anomalyAggregateNext, NULL, anomalyDestroyOperatorInfo,




×







152



                                         optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);










153













UNCOV

154



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




×








UNCOV

155



  QUERY_CHECK_CODE(code, lino, _error);





×







156













UNCOV

157



  *pOptrInfo = pOperator;




×







158













UNCOV

159



  qDebug("%s anomaly_window operator is created, algo:%s url:%s opt:%s", id, pInfo->algoName, pInfo->algoUrl,





×







160



         pInfo->anomalyOpt);











UNCOV

161



  return TSDB_CODE_SUCCESS;




×







162













UNCOV

163



_error:




×








UNCOV

164



  qError("%s failed to create anomaly_window operator, line:%d algo:%s code:%s", id, lino, pAnomalyNode->anomalyOpt,





×







165



         tstrerror(code));










166













UNCOV

167



  if (pInfo != NULL) {





×








UNCOV

168



    anomalyDestroyOperatorInfo(pInfo);




×







169



  }










170













UNCOV

171



  destroyOperatorAndDownstreams(pOperator, &downstream, 1);




×








UNCOV

172



  pTaskInfo->code = code;




×







173













UNCOV

174



  return code;




×







175



}










176













UNCOV

177



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




×








UNCOV

178



  int32_t                     code = TSDB_CODE_SUCCESS;




×








UNCOV

179



  int32_t                     lino = 0;




×








UNCOV

180



  SAnomalyWindowOperatorInfo* pInfo = pOperator->info;




×








UNCOV

181



  SExecTaskInfo*              pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

182



  SOptrBasicInfo*             pBInfo = &pInfo->binfo;




×








UNCOV

183



  SAnomalyWindowSupp*         pSupp = &pInfo->anomalySup;




×








UNCOV

184



  SSDataBlock*                pRes = pInfo->binfo.pRes;




×








UNCOV

185



  int64_t                     st = taosGetTimestampUs();




×








UNCOV

186



  int32_t                     numOfBlocks = taosArrayGetSize(pSupp->blocks);




×







187













UNCOV

188



  blockDataCleanup(pRes);




×







189













UNCOV

190



  while (1) {




×








UNCOV

191



    SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);




×








UNCOV

192



    if (pBlock == NULL) {





×








UNCOV

193



      break;




×







194



    }










195













UNCOV

196



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





×








UNCOV

197



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




×








UNCOV

198



      numOfBlocks++;




×








UNCOV

199



      pSupp->cachedRows += pBlock->info.rows;




×








UNCOV

200



      qDebug("group:%" PRId64 ", blocks:%d, rows:%" PRId64 ", total rows:%" PRId64, pSupp->groupId, numOfBlocks,





×







201



             pBlock->info.rows, pSupp->cachedRows);











UNCOV

202



      code = anomalyCacheBlock(pInfo, pBlock);




×








UNCOV

203



      QUERY_CHECK_CODE(code, lino, _end);





×







204



    } else {










205



      qDebug("group:%" PRId64 ", read finish for new group coming, blocks:%d", pSupp->groupId, numOfBlocks);





×







206



      code = anomalyAggregateBlocks(pOperator);




×







207



      QUERY_CHECK_CODE(code, lino, _end);





×







208












209



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




×







210



      numOfBlocks = 1;




×







211



      pSupp->cachedRows = pBlock->info.rows;




×







212



      qDebug("group:%" PRId64 ", new group, rows:%" PRId64 ", total rows:%" PRId64, pSupp->groupId,





×







213



             pBlock->info.rows, pSupp->cachedRows);










214



      code = anomalyCacheBlock(pInfo, pBlock);




×







215



      QUERY_CHECK_CODE(code, lino, _end);





×







216



    }










217













UNCOV

218



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





×







219



      (*ppRes) = pRes;




×







220



      qDebug("group:%" PRId64 ", return to upstream, blocks:%d", pRes->info.id.groupId, numOfBlocks);





×







221



      return code;




×







222



    }










223



  }










224













UNCOV

225



  if (numOfBlocks > 0) {





×








UNCOV

226



    qDebug("group:%" PRId64 ", read finish, blocks:%d", pInfo->anomalySup.groupId, numOfBlocks);





×








UNCOV

227



    code = anomalyAggregateBlocks(pOperator);




×







228



  }










229













UNCOV

230



  int64_t cost = taosGetTimestampUs() - st;




×








UNCOV

231



  qDebug("all groups finished, cost:%" PRId64 "us", cost);





×







232












233



_end:




×








UNCOV

234



  if (code != TSDB_CODE_SUCCESS) {





×







235



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





×







236



    pTaskInfo->code = code;




×







237



    T_LONG_JMP(pTaskInfo->env, code);





×







238



  }










239













UNCOV

240



  (*ppRes) = (pBInfo->pRes->info.rows == 0) ? NULL : pBInfo->pRes;





×








UNCOV

241



  return code;




×







242



}










243













UNCOV

244



static void anomalyDestroyOperatorInfo(void* param) {




×








UNCOV

245



  SAnomalyWindowOperatorInfo* pInfo = (SAnomalyWindowOperatorInfo*)param;




×








UNCOV

246



  if (pInfo == NULL) return;





×







247













UNCOV

248



  qDebug("anomaly_window operator is destroyed, algo:%s", pInfo->algoName);





×







249













UNCOV

250



  cleanupBasicInfo(&pInfo->binfo);




×








UNCOV

251



  cleanupAggSup(&pInfo->aggSup);




×








UNCOV

252



  cleanupExprSupp(&pInfo->scalarSup);




×








UNCOV

253



  colDataDestroy(&pInfo->twAggSup.timeWindowData);




×







254













UNCOV

255



  for (int32_t i = 0; i < taosArrayGetSize(pInfo->anomalySup.blocks); ++i) {





×







256



    SSDataBlock* pBlock = taosArrayGetP(pInfo->anomalySup.blocks, i);




×







257



    blockDataDestroy(pBlock);




×







258



  }











UNCOV

259



  taosArrayDestroy(pInfo->anomalySup.blocks);




×








UNCOV

260



  taosArrayDestroy(pInfo->anomalySup.windows);




×








UNCOV

261



  taosMemoryFreeClear(pInfo->anomalySup.pResultRow);





×








UNCOV

262



  taosMemoryFreeClear(pInfo->anomalyKey.pData);





×







263













UNCOV

264



  taosMemoryFreeClear(param);





×







265



}










266













UNCOV

267



static int32_t anomalyCacheBlock(SAnomalyWindowOperatorInfo* pInfo, SSDataBlock* pSrc) {




×








UNCOV

268



  if (pInfo->anomalySup.cachedRows > ANAL_ANOMALY_WINDOW_MAX_ROWS) {





×







269



    return TSDB_CODE_ANA_ANODE_TOO_MANY_ROWS;




×







270



  }










271













UNCOV

272



  SSDataBlock* pDst = NULL;




×








UNCOV

273



  int32_t      code = createOneDataBlock(pSrc, true, &pDst);




×







274













UNCOV

275



  if (code != 0) return code;





×








UNCOV

276



  if (pDst == NULL) return code;





×








UNCOV

277



  if (taosArrayPush(pInfo->anomalySup.blocks, &pDst) == NULL) return terrno;





×







278













UNCOV

279



  return 0;




×







280



}










281













UNCOV

282



static int32_t anomalyFindWindow(SAnomalyWindowSupp* pSupp, TSKEY key) {




×








UNCOV

283



  for (int32_t i = pSupp->curWinIndex; i < taosArrayGetSize(pSupp->windows); ++i) {





×








UNCOV

284



    STimeWindow* pWindow = taosArrayGet(pSupp->windows, i);




×








UNCOV

285



    if (key >= pWindow->skey && key < pWindow->ekey) {





×







286



      pSupp->curWin = *pWindow;




×







287



      pSupp->curWinIndex = i;




×







288



      return 0;




×







289



    }










290



  }











UNCOV

291



  return -1;




×







292



}










293













UNCOV

294



static int32_t anomalyParseJson(SJson* pJson, SArray* pWindows, const char* pId) {




×








UNCOV

295



  int32_t     code = 0;




×








UNCOV

296



  int32_t     rows = 0;




×








UNCOV

297



  STimeWindow win = {0};




×







298













UNCOV

299



  taosArrayClear(pWindows);




×







300













UNCOV

301



  tjsonGetInt32ValueFromDouble(pJson, "rows", rows, code);




×








UNCOV

302



  if (code < 0) {





×







303



    return TSDB_CODE_INVALID_JSON_FORMAT;




×







304



  }










305













UNCOV

306



  if (rows < 0) {





×







307



    char pMsg[1024] = {0};




×







308



    code = tjsonGetStringValue(pJson, "msg", pMsg);




×







309



    if (code) {





×







310



      qError("%s failed to get error msg from rsp, unknown error", pId);





×







311



    } else {










312



      qError("%s failed to exec forecast, msg:%s", pId, pMsg);





×







313



    }










314












315



    return TSDB_CODE_ANA_WN_DATA;




×








UNCOV

316



  } else if (rows == 0) {





×








UNCOV

317



    return TSDB_CODE_SUCCESS;




×







318



  }










319













UNCOV

320



  SJson* res = tjsonGetObjectItem(pJson, "res");




×








UNCOV

321



  if (res == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;





×







322













UNCOV

323



  int32_t ressize = tjsonGetArraySize(res);




×








UNCOV

324



  if (ressize != rows) return TSDB_CODE_INVALID_JSON_FORMAT;





×







325













UNCOV

326



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





×








UNCOV

327



    SJson* row = tjsonGetArrayItem(res, i);




×








UNCOV

328



    if (row == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;





×







329













UNCOV

330



    int32_t colsize = tjsonGetArraySize(row);




×








UNCOV

331



    if (colsize != 2) return TSDB_CODE_INVALID_JSON_FORMAT;





×







332













UNCOV

333



    SJson* start = tjsonGetArrayItem(row, 0);




×








UNCOV

334



    SJson* end = tjsonGetArrayItem(row, 1);




×








UNCOV

335



    if (start == NULL || end == NULL) {





×







336



      qError("%s invalid res from analytic sys, code:%s", pId, tstrerror(TSDB_CODE_INVALID_JSON_FORMAT));





×







337



      return TSDB_CODE_INVALID_JSON_FORMAT;




×







338



    }










339













UNCOV

340



    tjsonGetObjectValueBigInt(start, &win.skey);




×








UNCOV

341



    tjsonGetObjectValueBigInt(end, &win.ekey);




×







342













UNCOV

343



    if (win.skey >= win.ekey) {





×








UNCOV

344



      win.ekey = win.skey + 1;




×







345



    }










346













UNCOV

347



    if (taosArrayPush(pWindows, &win) == NULL) {





×







348



      qError("%s out of memory in generating anomaly_window", pId);





×







349



      return TSDB_CODE_OUT_OF_BUFFER;




×







350



    }










351



  }










352













UNCOV

353



  int32_t numOfWins = taosArrayGetSize(pWindows);




×








UNCOV

354



  qDebug("%s anomaly window recevied, total:%d", pId, numOfWins);





×








UNCOV

355



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





×








UNCOV

356



    STimeWindow* pWindow = taosArrayGet(pWindows, i);




×








UNCOV

357



    qDebug("%s anomaly win:%d [%" PRId64 ", %" PRId64 ")", pId, i, pWindow->skey, pWindow->ekey);





×







358



  }










359













UNCOV

360



  return code;




×







361



}










362













UNCOV

363



static int32_t anomalyAnalysisWindow(SOperatorInfo* pOperator) {




×








UNCOV

364



  SAnomalyWindowOperatorInfo* pInfo = pOperator->info;




×








UNCOV

365



  SAnomalyWindowSupp*         pSupp = &pInfo->anomalySup;




×








UNCOV

366



  SJson*                      pJson = NULL;




×








UNCOV

367



  SAnalyticBuf                analBuf = {.bufType = ANALYTICS_BUF_TYPE_JSON};




×








UNCOV

368



  char                        dataBuf[64] = {0};




×








UNCOV

369



  int32_t                     code = 0;




×








UNCOV

370



  int64_t                     ts = 0;




×








UNCOV

371



  int32_t                     lino = 0;




×








UNCOV

372



  const char*                 pId = GET_TASKID(pOperator->pTaskInfo);




×







373













UNCOV

374



  snprintf(analBuf.fileName, sizeof(analBuf.fileName), "%s/tdengine-anomaly-%" PRId64 "-%" PRId64, tsTempDir, ts,




×







375



           pSupp->groupId);











UNCOV

376



  code = tsosAnalBufOpen(&analBuf, 2);




×








UNCOV

377



  QUERY_CHECK_CODE(code, lino, _OVER);





×







378













UNCOV

379



  const char* prec = TSDB_TIME_PRECISION_MILLI_STR;




×








UNCOV

380



  if (pInfo->anomalyCol.precision == TSDB_TIME_PRECISION_MICRO) prec = TSDB_TIME_PRECISION_MICRO_STR;





×








UNCOV

381



  if (pInfo->anomalyCol.precision == TSDB_TIME_PRECISION_NANO) prec = TSDB_TIME_PRECISION_NANO_STR;





×







382













UNCOV

383



  code = taosAnalBufWriteColMeta(&analBuf, 0, TSDB_DATA_TYPE_TIMESTAMP, "ts");




×








UNCOV

384



  QUERY_CHECK_CODE(code, lino, _OVER);





×







385













UNCOV

386



  code = taosAnalBufWriteColMeta(&analBuf, 1, pInfo->anomalyCol.type, "val");




×








UNCOV

387



  QUERY_CHECK_CODE(code, lino, _OVER);





×







388













UNCOV

389



  code = taosAnalBufWriteDataBegin(&analBuf);




×








UNCOV

390



  QUERY_CHECK_CODE(code, lino, _OVER);





×







391













UNCOV

392



  int32_t numOfBlocks = (int32_t)taosArrayGetSize(pSupp->blocks);




×







393












394



  // timestamp











UNCOV

395



  code = taosAnalBufWriteColBegin(&analBuf, 0);




×








UNCOV

396



  QUERY_CHECK_CODE(code, lino, _OVER);





×







397













UNCOV

398



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





×








UNCOV

399



    SSDataBlock* pBlock = taosArrayGetP(pSupp->blocks, i);




×








UNCOV

400



    if (pBlock == NULL) break;





×








UNCOV

401



    SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);




×








UNCOV

402



    if (pTsCol == NULL) break;





×








UNCOV

403



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





×








UNCOV

404



      code = taosAnalBufWriteColData(&analBuf, 0, TSDB_DATA_TYPE_TIMESTAMP, &((TSKEY*)pTsCol->pData)[j]);




×








UNCOV

405



      QUERY_CHECK_CODE(code, lino, _OVER);





×







406



    }










407



  }










408













UNCOV

409



  code = taosAnalBufWriteColEnd(&analBuf, 0);




×








UNCOV

410



  QUERY_CHECK_CODE(code, lino, _OVER);





×







411












412



  // data











UNCOV

413



  code = taosAnalBufWriteColBegin(&analBuf, 1);




×








UNCOV

414



  QUERY_CHECK_CODE(code, lino, _OVER);





×







415













UNCOV

416



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





×








UNCOV

417



    SSDataBlock* pBlock = taosArrayGetP(pSupp->blocks, i);




×








UNCOV

418



    if (pBlock == NULL) break;





×








UNCOV

419



    SColumnInfoData* pValCol = taosArrayGet(pBlock->pDataBlock, pInfo->anomalyCol.slotId);




×








UNCOV

420



    if (pValCol == NULL) break;





×







421













UNCOV

422



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





×








UNCOV

423



      code = taosAnalBufWriteColData(&analBuf, 1, pValCol->info.type, colDataGetData(pValCol, j));





×








UNCOV

424



      QUERY_CHECK_CODE(code, lino, _OVER);





×







425



    }










426



  }











UNCOV

427



  code = taosAnalBufWriteColEnd(&analBuf, 1);




×








UNCOV

428



  QUERY_CHECK_CODE(code, lino, _OVER);





×







429













UNCOV

430



  code = taosAnalBufWriteDataEnd(&analBuf);




×








UNCOV

431



  QUERY_CHECK_CODE(code, lino, _OVER);





×







432













UNCOV

433



  code = taosAnalBufWriteOptStr(&analBuf, "option", pInfo->anomalyOpt);




×








UNCOV

434



  QUERY_CHECK_CODE(code, lino, _OVER);





×







435













UNCOV

436



  code = taosAnalBufWriteOptStr(&analBuf, "algo", pInfo->algoName);




×








UNCOV

437



  QUERY_CHECK_CODE(code, lino, _OVER);





×







438













UNCOV

439



  code = taosAnalBufWriteOptStr(&analBuf, "prec", prec);




×








UNCOV

440



  QUERY_CHECK_CODE(code, lino, _OVER);





×







441













UNCOV

442



  int64_t wncheck = ANAL_FORECAST_DEFAULT_WNCHECK;




×








UNCOV

443



  bool    hasWncheck = taosAnalGetOptInt(pInfo->anomalyOpt, "wncheck", &wncheck);




×








UNCOV

444



  if (!hasWncheck) {





×








UNCOV

445



    qDebug("anomaly_window wncheck not found from %s, use default:%" PRId64, pInfo->anomalyOpt, wncheck);





×







446



  }










447













UNCOV

448



  code = taosAnalBufWriteOptInt(&analBuf, "wncheck", wncheck);




×








UNCOV

449



  QUERY_CHECK_CODE(code, lino, _OVER);





×







450













UNCOV

451



  code = taosAnalBufClose(&analBuf);




×








UNCOV

452



  QUERY_CHECK_CODE(code, lino, _OVER);





×







453













UNCOV

454



  pJson = taosAnalSendReqRetJson(pInfo->algoUrl, ANALYTICS_HTTP_TYPE_POST, &analBuf);




×








UNCOV

455



  if (pJson == NULL) {





×







456



    code = terrno;




×







457



    goto _OVER;




×







458



  }










459













UNCOV

460



  code = anomalyParseJson(pJson, pSupp->windows, pId);




×







461













UNCOV

462



_OVER:




×








UNCOV

463



  if (code != 0) {





×







464



    qError("%s failed to analysis window since %s, lino:%d", pId, tstrerror(code), lino);





×







465



  }










466













UNCOV

467



  taosAnalBufDestroy(&analBuf);




×








UNCOV

468



  if (pJson != NULL) tjsonDelete(pJson);





×








UNCOV

469



  return code;




×







470



}










471













UNCOV

472



static int32_t anomalyAggregateRows(SOperatorInfo* pOperator, SSDataBlock* pBlock) {




×








UNCOV

473



  SAnomalyWindowOperatorInfo* pInfo = pOperator->info;




×








UNCOV

474



  SExecTaskInfo*              pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

475



  SExprSupp*                  pExprSup = &pOperator->exprSupp;




×








UNCOV

476



  SAnomalyWindowSupp*         pSupp = &pInfo->anomalySup;




×








UNCOV

477



  SWindowRowsSup*             pRowSup = &pInfo->anomalyWinRowSup;




×








UNCOV

478



  SResultRow*                 pResRow = pSupp->pResultRow;




×








UNCOV

479



  int32_t                     numOfOutput = pOperator->exprSupp.numOfExprs;




×







480













UNCOV

481



  int32_t code = setResultRowInitCtx(pResRow, pExprSup->pCtx, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);




×








UNCOV

482



  if (code == 0) {





×








UNCOV

483



    updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pSupp->curWin, 0);




×








UNCOV

484



    code = applyAggFunctionOnPartialTuples(pTaskInfo, pExprSup->pCtx, &pInfo->twAggSup.timeWindowData,




×








UNCOV

485



                                           pRowSup->startRowIndex, pRowSup->numOfRows, pBlock->info.rows, numOfOutput);




×







486



  }










487













UNCOV

488



  return code;




×







489



}










490













UNCOV

491



static int32_t anomalyBuildResult(SOperatorInfo* pOperator) {




×








UNCOV

492



  SAnomalyWindowOperatorInfo* pInfo = pOperator->info;




×








UNCOV

493



  SExecTaskInfo*              pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

494



  SExprSupp*                  pExprSup = &pOperator->exprSupp;




×








UNCOV

495



  SSDataBlock*                pRes = pInfo->binfo.pRes;




×








UNCOV

496



  SResultRow*                 pResRow = pInfo->anomalySup.pResultRow;




×







497













UNCOV

498



  doUpdateNumOfRows(pExprSup->pCtx, pResRow, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);




×








UNCOV

499



  int32_t code = copyResultrowToDataBlock(pExprSup->pExprInfo, pExprSup->numOfExprs, pResRow, pExprSup->pCtx, pRes,




×








UNCOV

500



                                          pExprSup->rowEntryInfoOffset, pTaskInfo);




×








UNCOV

501



  if (code == 0) {





×








UNCOV

502



    pRes->info.rows += pResRow->numOfRows;




×







503



  }










504













UNCOV

505



  clearResultRowInitFlag(pExprSup->pCtx, pExprSup->numOfExprs);




×








UNCOV

506



  return code;




×







507



}










508













UNCOV

509



static int32_t anomalyAggregateBlocks(SOperatorInfo* pOperator) {




×








UNCOV

510



  int32_t                     code = TSDB_CODE_SUCCESS;




×








UNCOV

511



  int32_t                     lino = 0;




×








UNCOV

512



  SAnomalyWindowOperatorInfo* pInfo = pOperator->info;




×








UNCOV

513



  SExecTaskInfo*              pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

514



  SExprSupp*                  pExprSup = &pOperator->exprSupp;




×








UNCOV

515



  SSDataBlock*                pRes = pInfo->binfo.pRes;




×








UNCOV

516



  SAnomalyWindowSupp*         pSupp = &pInfo->anomalySup;




×








UNCOV

517



  SWindowRowsSup*             pRowSup = &pInfo->anomalyWinRowSup;




×








UNCOV

518



  SResultRow*                 pResRow = pSupp->pResultRow;




×








UNCOV

519



  int32_t                     numOfOutput = pOperator->exprSupp.numOfExprs;




×








UNCOV

520



  int32_t                     rowsInWin = 0;




×








UNCOV

521



  int32_t                     rowsInBlock = 0;




×








UNCOV

522



  const int64_t               gid = pSupp->groupId;




×








UNCOV

523



  const int32_t               order = pInfo->binfo.inputTsOrder;




×







524













UNCOV

525



  int32_t numOfBlocks = (int32_t)taosArrayGetSize(pSupp->blocks);




×








UNCOV

526



  if (numOfBlocks == 0) goto _OVER;





×







527













UNCOV

528



  qDebug("group:%" PRId64 ", aggregate blocks, blocks:%d", pSupp->groupId, numOfBlocks);





×








UNCOV

529



  pRes->info.id.groupId = pSupp->groupId;




×







530













UNCOV

531



  code = anomalyAnalysisWindow(pOperator);




×








UNCOV

532



  QUERY_CHECK_CODE(code, lino, _OVER);





×







533













UNCOV

534



  int32_t numOfWins = taosArrayGetSize(pSupp->windows);




×








UNCOV

535



  qDebug("group:%" PRId64 ", wins:%d, rows:%" PRId64, pSupp->groupId, numOfWins, pSupp->cachedRows);





×








UNCOV

536



  for (int32_t w = 0; w < numOfWins; ++w) {





×








UNCOV

537



    STimeWindow* pWindow = taosArrayGet(pSupp->windows, w);




×








UNCOV

538



    if (w == 0) {





×








UNCOV

539



      pSupp->curWin = *pWindow;




×








UNCOV

540



      pRowSup->win.skey = pSupp->curWin.skey;




×







541



    }











UNCOV

542



    qDebug("group:%" PRId64 ", win:%d [%" PRId64 ", %" PRId64 ")", pSupp->groupId, w, pWindow->skey, pWindow->ekey);





×







543



  }










544













UNCOV

545



  if (numOfWins <= 0) goto _OVER;





×








UNCOV

546



  if (numOfWins > pRes->info.capacity) {





×







547



    code = blockDataEnsureCapacity(pRes, numOfWins);




×







548



    QUERY_CHECK_CODE(code, lino, _OVER);





×







549



  }










550













UNCOV

551



  for (int32_t b = 0; b < numOfBlocks; ++b) {





×








UNCOV

552



    SSDataBlock* pBlock = taosArrayGetP(pSupp->blocks, b);




×








UNCOV

553



    if (pBlock == NULL) break;





×







554













UNCOV

555



    pRes->info.scanFlag = pBlock->info.scanFlag;




×








UNCOV

556



    code = setInputDataBlock(pExprSup, pBlock, order, MAIN_SCAN, true);




×








UNCOV

557



    if (code != 0) break;





×







558













UNCOV

559



    code = blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);




×








UNCOV

560



    if (code != 0) break;





×







561












562



    // there is an scalar expression that needs to be calculated right before apply the group aggregation.











UNCOV

563



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





×







564



      code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,




×







565



                                   pInfo->scalarSup.numOfExprs, NULL);










566



      if (code != 0) break;





×







567



    }










568













UNCOV

569



    SColumnInfoData* pValCol = taosArrayGet(pBlock->pDataBlock, pInfo->anomalyCol.slotId);




×








UNCOV

570



    if (pValCol == NULL) break;





×








UNCOV

571



    SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);




×








UNCOV

572



    if (pTsCol == NULL) break;





×








UNCOV

573



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




×








UNCOV

574



    bool   lastBlock = (b == numOfBlocks - 1);




×







575













UNCOV

576



    qTrace("group:%" PRId64 ", block:%d win:%d, riwin:%d riblock:%d, rows:%" PRId64, pSupp->groupId, b,





×







577



           pSupp->curWinIndex, rowsInWin, rowsInBlock, pBlock->info.rows);










578













UNCOV

579



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





×








UNCOV

580



      TSKEY key = tsList[r];




×








UNCOV

581



      bool  keyInWin = (key >= pSupp->curWin.skey && key < pSupp->curWin.ekey);





×








UNCOV

582



      bool  lastRow = (r == pBlock->info.rows - 1);




×







583













UNCOV

584



      if (keyInWin) {





×








UNCOV

585



        if (r < 5) {





×







586



          qTrace("group:%" PRId64 ", block:%d win:%d, row:%d ts:%" PRId64 ", riwin:%d riblock:%d", pSupp->groupId, b,





×







587



                 pSupp->curWinIndex, r, key, rowsInWin, rowsInBlock);










588



        }











UNCOV

589



        if (rowsInBlock == 0) {





×








UNCOV

590



          doKeepNewWindowStartInfo(pRowSup, tsList, r, gid);




×







591



        }











UNCOV

592



        doKeepTuple(pRowSup, tsList[r], gid);




×








UNCOV

593



        rowsInBlock++;




×








UNCOV

594



        rowsInWin++;




×







595



      } else {











UNCOV

596



        if (rowsInBlock > 0) {





×








UNCOV

597



          qTrace("group:%" PRId64 ", block:%d win:%d, row:%d ts:%" PRId64 ", riwin:%d riblock:%d, agg", pSupp->groupId,





×







598



                 b, pSupp->curWinIndex, r, key, rowsInWin, rowsInBlock);











UNCOV

599



          code = anomalyAggregateRows(pOperator, pBlock);




×








UNCOV

600



          QUERY_CHECK_CODE(code, lino, _OVER);





×








UNCOV

601



          rowsInBlock = 0;




×







602



        }











UNCOV

603



        if (rowsInWin > 0) {





×








UNCOV

604



          qTrace("group:%" PRId64 ", block:%d win:%d, row:%d ts:%" PRId64 ", riwin:%d riblock:%d, build result",





×







605



                 pSupp->groupId, b, pSupp->curWinIndex, r, key, rowsInWin, rowsInBlock);











UNCOV

606



          code = anomalyBuildResult(pOperator);




×








UNCOV

607



          QUERY_CHECK_CODE(code, lino, _OVER);





×








UNCOV

608



          rowsInWin = 0;




×







609



        }











UNCOV

610



        if (anomalyFindWindow(pSupp, tsList[r]) == 0) {





×







611



          qTrace("group:%" PRId64 ", block:%d win:%d, row:%d ts:%" PRId64 ", riwin:%d riblock:%d, new window detect",





×







612



                 pSupp->groupId, b, pSupp->curWinIndex, r, key, rowsInWin, rowsInBlock);










613



          doKeepNewWindowStartInfo(pRowSup, tsList, r, gid);




×







614



          doKeepTuple(pRowSup, tsList[r], gid);




×







615



          rowsInBlock = 1;




×







616



          rowsInWin = 1;




×







617



        } else {











UNCOV

618



          qTrace("group:%" PRId64 ", block:%d win:%d, row:%d ts:%" PRId64 ", riwin:%d riblock:%d, window not found",





×







619



                 pSupp->groupId, b, pSupp->curWinIndex, r, key, rowsInWin, rowsInBlock);











UNCOV

620



          rowsInBlock = 0;




×








UNCOV

621



          rowsInWin = 0;




×







622



        }










623



      }










624













UNCOV

625



      if (lastRow && rowsInBlock > 0) {





×








UNCOV

626



        qTrace("group:%" PRId64 ", block:%d win:%d, row:%d ts:%" PRId64 ", riwin:%d riblock:%d, agg since lastrow",





×







627



               pSupp->groupId, b, pSupp->curWinIndex, r, key, rowsInWin, rowsInBlock);











UNCOV

628



        code = anomalyAggregateRows(pOperator, pBlock);




×








UNCOV

629



        QUERY_CHECK_CODE(code, lino, _OVER);





×








UNCOV

630



        rowsInBlock = 0;




×







631



      }










632



    }










633













UNCOV

634



    if (lastBlock && rowsInWin > 0) {





×








UNCOV

635



      qTrace("group:%" PRId64 ", block:%d win:%d, riwin:%d riblock:%d, build result since lastblock", pSupp->groupId, b,





×







636



             pSupp->curWinIndex, rowsInWin, rowsInBlock);











UNCOV

637



      code = anomalyBuildResult(pOperator);




×








UNCOV

638



      QUERY_CHECK_CODE(code, lino, _OVER);





×








UNCOV

639



      rowsInWin = 0;




×







640



    }










641



  }










642













UNCOV

643



  code = doFilter(pRes, pOperator->exprSupp.pFilterInfo, NULL);




×








UNCOV

644



  QUERY_CHECK_CODE(code, lino, _OVER);





×







645













UNCOV

646



_OVER:




×








UNCOV

647



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





×








UNCOV

648



    SSDataBlock* pBlock = taosArrayGetP(pSupp->blocks, i);




×








UNCOV

649



    qDebug("%s, clear block, pBlock:%p pBlock->pDataBlock:%p", __func__, pBlock, pBlock->pDataBlock);





×








UNCOV

650



    blockDataDestroy(pBlock);




×







651



  }










652













UNCOV

653



  taosArrayClear(pSupp->blocks);




×








UNCOV

654



  taosArrayClear(pSupp->windows);




×








UNCOV

655



  pSupp->cachedRows = 0;




×








UNCOV

656



  pSupp->curWin.ekey = 0;




×








UNCOV

657



  pSupp->curWin.skey = 0;




×








UNCOV

658



  pSupp->curWinIndex = 0;




×







659













UNCOV

660



  return code;




×







661



}










662












663



#else










664












665



int32_t createAnomalywindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode, SExecTaskInfo* pTaskInfo,










666



                                        SOperatorInfo** pOptrInfo) {










667



  return TSDB_CODE_OPS_NOT_SUPPORT;










668



}










669



void destroyForecastInfo(void* param) {}










670












671



#endif























    

  • 

    •