taosdata / TDengine / #3842

coverage: 62.696% (-0.3%) from 63.027%
#3842

push

travis-ci

web-flow 
merge: from main to 3.0 branch (#30679)

154855 of 315075 branches covered (49.15%)

Branch coverage included in aggregate %.

6 of 8 new or added lines in 5 files covered. (75.0%)

2309 existing lines in 130 files now uncovered.

240176 of 314995 relevant lines covered (76.25%)

19119980.29 hits per line

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

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










19



#include "operator.h"










20



#include "querytask.h"










21



#include "tanalytics.h"










22



#include "tcommon.h"










23



#include "tcompare.h"










24



#include "tdatablock.h"










25



#include "tfill.h"










26



#include "ttime.h"










27












28



#ifdef USE_ANALYTICS










29












30



typedef struct {










31



  char         algoName[TSDB_ANALYTIC_ALGO_NAME_LEN];










32



  char         algoUrl[TSDB_ANALYTIC_ALGO_URL_LEN];










33



  char         algoOpt[TSDB_ANALYTIC_ALGO_OPTION_LEN];










34



  int64_t      maxTs;










35



  int64_t      minTs;










36



  int64_t      numOfRows;










37



  uint64_t     groupId;










38



  int64_t      optRows;










39



  int64_t      cachedRows;










40



  int32_t      numOfBlocks;










41



  int64_t      timeout;










42



  int16_t      resTsSlot;










43



  int16_t      resValSlot;










44



  int16_t      resLowSlot;










45



  int16_t      resHighSlot;










46



  int16_t      inputTsSlot;










47



  int16_t      inputValSlot;










48



  int8_t       inputValType;










49



  int8_t       inputPrecision;










50



  SAnalyticBuf analyBuf;










51



} SForecastSupp;










52












53



typedef struct SForecastOperatorInfo {










54



  SSDataBlock*  pRes;










55



  SExprSupp     scalarSup;  // scalar calculation










56



  SForecastSupp forecastSupp;










57



} SForecastOperatorInfo;










58












59



static void destroyForecastInfo(void* param);










60












61



static FORCE_INLINE int32_t forecastEnsureBlockCapacity(SSDataBlock* pBlock, int32_t newRowsNum) {











UNCOV

62



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





×








UNCOV

63



    return TSDB_CODE_SUCCESS;




×







64



  }










65












66



  int32_t code = blockDataEnsureCapacity(pBlock, newRowsNum);




×







67



  if (code != TSDB_CODE_SUCCESS) {





×







68



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





×







69



    return code;




×







70



  }










71












72



  return TSDB_CODE_SUCCESS;




×







73



}










74













UNCOV

75



static int32_t forecastCacheBlock(SForecastSupp* pSupp, SSDataBlock* pBlock, const char* id) {




×








UNCOV

76



  int32_t       code = TSDB_CODE_SUCCESS;




×








UNCOV

77



  int32_t       lino = 0;




×








UNCOV

78



  SAnalyticBuf* pBuf = &pSupp->analyBuf;




×







79













UNCOV

80



  if (pSupp->cachedRows > ANALY_FORECAST_MAX_ROWS) {





×







81



    code = TSDB_CODE_ANA_ANODE_TOO_MANY_ROWS;




×







82



    qError("%s rows:%" PRId64 " for forecast cache, error happens, code:%s, upper limit:%d", id, pSupp->cachedRows,





×







83



           tstrerror(code), ANALY_FORECAST_MAX_ROWS);










84



    return code;




×







85



  }










86













UNCOV

87



  pSupp->numOfBlocks++;




×








UNCOV

88



  qDebug("%s block:%d, %p rows:%" PRId64, id, pSupp->numOfBlocks, pBlock, pBlock->info.rows);





×







89













UNCOV

90



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





×








UNCOV

91



    SColumnInfoData* pValCol = taosArrayGet(pBlock->pDataBlock, pSupp->inputValSlot);




×








UNCOV

92



    SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, pSupp->inputTsSlot);




×








UNCOV

93



    if (pTsCol == NULL || pValCol == NULL) break;





×







94













UNCOV

95



    int64_t ts = ((TSKEY*)pTsCol->pData)[j];




×








UNCOV

96



    char*   val = colDataGetData(pValCol, j);





×








UNCOV

97



    int16_t valType = pValCol->info.type;




×







98













UNCOV

99



    pSupp->minTs = MIN(pSupp->minTs, ts);




×








UNCOV

100



    pSupp->maxTs = MAX(pSupp->maxTs, ts);




×








UNCOV

101



    pSupp->numOfRows++;




×







102













UNCOV

103



    code = taosAnalyBufWriteColData(pBuf, 0, TSDB_DATA_TYPE_TIMESTAMP, &ts);




×








UNCOV

104



    if (TSDB_CODE_SUCCESS != code) {





×







105



      qError("%s failed to write ts in buf, code:%s", id, tstrerror(code));





×







106



      return code;




×







107



    }










108













UNCOV

109



    code = taosAnalyBufWriteColData(pBuf, 1, valType, val);




×








UNCOV

110



    if (TSDB_CODE_SUCCESS != code) {





×







111



      qError("%s failed to write val in buf, code:%s", id, tstrerror(code));





×







112



      return code;




×







113



    }










114



  }










115













UNCOV

116



  return 0;




×







117



}










118













UNCOV

119



static int32_t forecastCloseBuf(SForecastSupp* pSupp, const char* id) {




×








UNCOV

120



  SAnalyticBuf* pBuf = &pSupp->analyBuf;




×








UNCOV

121



  int32_t       code = 0;




×







122













UNCOV

123



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





×








UNCOV

124



    code = taosAnalyBufWriteColEnd(pBuf, i);




×








UNCOV

125



    if (code != 0) return code;





×







126



  }










127













UNCOV

128



  code = taosAnalyBufWriteDataEnd(pBuf);




×








UNCOV

129



  if (code != 0) return code;





×







130













UNCOV

131



  code = taosAnalyBufWriteOptStr(pBuf, "option", pSupp->algoOpt);




×








UNCOV

132



  if (code != 0) return code;





×







133













UNCOV

134



  code = taosAnalyBufWriteOptStr(pBuf, "algo", pSupp->algoName);




×








UNCOV

135



  if (code != 0) return code;





×







136













UNCOV

137



  const char* prec = TSDB_TIME_PRECISION_MILLI_STR;




×








UNCOV

138



  if (pSupp->inputPrecision == TSDB_TIME_PRECISION_MICRO) prec = TSDB_TIME_PRECISION_MICRO_STR;





×








UNCOV

139



  if (pSupp->inputPrecision == TSDB_TIME_PRECISION_NANO) prec = TSDB_TIME_PRECISION_NANO_STR;





×








UNCOV

140



  code = taosAnalyBufWriteOptStr(pBuf, "prec", prec);




×








UNCOV

141



  if (code != 0) return code;





×







142













UNCOV

143



  int64_t wncheck = ANALY_FORECAST_DEFAULT_WNCHECK;




×








UNCOV

144



  bool    hasWncheck = taosAnalyGetOptInt(pSupp->algoOpt, "wncheck", &wncheck);




×








UNCOV

145



  if (!hasWncheck) {





×








UNCOV

146



    qDebug("%s forecast wncheck not found from %s, use default:%" PRId64, id, pSupp->algoOpt, wncheck);





×







147



  }










148













UNCOV

149



  code = taosAnalyBufWriteOptInt(pBuf, "wncheck", wncheck);




×








UNCOV

150



  if (code != 0) return code;





×







151













UNCOV

152



  bool noConf = (pSupp->resHighSlot == -1 && pSupp->resLowSlot == -1);





×








UNCOV

153



  code = taosAnalyBufWriteOptInt(pBuf, "return_conf", !noConf);




×








UNCOV

154



  if (code != 0) return code;





×







155













UNCOV

156



  pSupp->optRows = ANALY_FORECAST_DEFAULT_ROWS;




×








UNCOV

157



  bool hasRows = taosAnalyGetOptInt(pSupp->algoOpt, "rows", &pSupp->optRows);




×








UNCOV

158



  if (!hasRows) {





×








UNCOV

159



    qDebug("%s forecast rows not found from %s, use default:%" PRId64, id, pSupp->algoOpt, pSupp->optRows);





×







160



  }










161













UNCOV

162



  if (pSupp->optRows > ANALY_FORECAST_RES_MAX_ROWS) {





×







163



    qError("%s required too many forecast rows, max allowed:%d, required:%" PRId64, id, ANALY_FORECAST_RES_MAX_ROWS,





×







164



           pSupp->optRows);










165



    return TSDB_CODE_ANA_ANODE_TOO_MANY_ROWS;




×







166



  }










167













UNCOV

168



  if (pSupp->cachedRows < ANALY_FORECAST_MIN_ROWS) {





×







169



    qError("%s required too many forecast rows, max allowed:%d, required:%" PRId64, id, ANALY_FORECAST_RES_MAX_ROWS,





×







170



           pSupp->optRows);










171



    return TSDB_CODE_ANA_ANODE_NOT_ENOUGH_ROWS;




×







172



  }










173













UNCOV

174



  code = taosAnalyBufWriteOptInt(pBuf, "forecast_rows", pSupp->optRows);




×








UNCOV

175



  if (code != 0) return code;





×







176













UNCOV

177



  int64_t conf = ANALY_FORECAST_DEFAULT_CONF;




×








UNCOV

178



  bool    hasConf = taosAnalyGetOptInt(pSupp->algoOpt, "conf", &conf);




×








UNCOV

179



  if (!hasConf) {





×








UNCOV

180



    qDebug("%s forecast conf not found from %s, use default:%" PRId64, id, pSupp->algoOpt, conf);





×







181



  }











UNCOV

182



  code = taosAnalyBufWriteOptInt(pBuf, "conf", conf);




×








UNCOV

183



  if (code != 0) return code;





×







184













UNCOV

185



  int32_t len = strlen(pSupp->algoOpt);




×








UNCOV

186



  int64_t every = (pSupp->maxTs - pSupp->minTs) / (pSupp->numOfRows - 1);




×








UNCOV

187



  int64_t start = pSupp->maxTs + every;




×








UNCOV

188



  bool    hasStart = taosAnalyGetOptInt(pSupp->algoOpt, "start", &start);




×








UNCOV

189



  if (!hasStart) {





×








UNCOV

190



    qDebug("%s forecast start not found from %s, use %" PRId64, id, pSupp->algoOpt, start);





×







191



  }











UNCOV

192



  code = taosAnalyBufWriteOptInt(pBuf, "start", start);




×








UNCOV

193



  if (code != 0) return code;





×







194













UNCOV

195



  bool hasEvery = taosAnalyGetOptInt(pSupp->algoOpt, "every", &every);




×








UNCOV

196



  if (!hasEvery) {





×








UNCOV

197



    qDebug("%s forecast every not found from %s, use %" PRId64, id, pSupp->algoOpt, every);





×







198



  }











UNCOV

199



  code = taosAnalyBufWriteOptInt(pBuf, "every", every);




×








UNCOV

200



  if (code != 0) return code;





×







201













UNCOV

202



  code = taosAnalyBufClose(pBuf);




×








UNCOV

203



  return code;




×







204



}










205













UNCOV

206



static int32_t forecastAnalysis(SForecastSupp* pSupp, SSDataBlock* pBlock, const char* pId) {




×








UNCOV

207



  SAnalyticBuf* pBuf = &pSupp->analyBuf;




×








UNCOV

208



  int32_t       resCurRow = pBlock->info.rows;




×








UNCOV

209



  int8_t        tmpI8 = 0;




×








UNCOV

210



  int16_t       tmpI16 = 0;




×








UNCOV

211



  int32_t       tmpI32 = 0;




×








UNCOV

212



  int64_t       tmpI64 = 0;




×








UNCOV

213



  float         tmpFloat = 0;




×








UNCOV

214



  double        tmpDouble = 0;




×








UNCOV

215



  int32_t       code = 0;




×







216













UNCOV

217



  SColumnInfoData* pResValCol = taosArrayGet(pBlock->pDataBlock, pSupp->resValSlot);




×








UNCOV

218



  if (NULL == pResValCol) {





×







219



    return terrno;




×







220



  }










221













UNCOV

222



  SColumnInfoData* pResTsCol = ((pSupp->resTsSlot != -1) ? taosArrayGet(pBlock->pDataBlock, pSupp->resTsSlot) : NULL);





×








UNCOV

223



  SColumnInfoData* pResLowCol =




×








UNCOV

224



      ((pSupp->resLowSlot != -1) ? taosArrayGet(pBlock->pDataBlock, pSupp->resLowSlot) : NULL);





×








UNCOV

225



  SColumnInfoData* pResHighCol =




×








UNCOV

226



      (pSupp->resHighSlot != -1 ? taosArrayGet(pBlock->pDataBlock, pSupp->resHighSlot) : NULL);





×







227













UNCOV

228



  SJson* pJson = taosAnalySendReqRetJson(pSupp->algoUrl, ANALYTICS_HTTP_TYPE_POST, pBuf, pSupp->timeout * 1000);




×








UNCOV

229



  if (pJson == NULL) {





×







230



    return terrno;




×







231



  }










232













UNCOV

233



  int32_t rows = 0;




×








UNCOV

234



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




×








UNCOV

235



  if (rows < 0 && code == 0) {





×








UNCOV

236



    char pMsg[1024] = {0};




×








UNCOV

237



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




×








UNCOV

238



    if (code != 0) {





×







239



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





×







240



    } else {











UNCOV

241



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





×







242



    }










243













UNCOV

244



    tjsonDelete(pJson);




×








UNCOV

245



    return TSDB_CODE_ANA_ANODE_RETURN_ERROR;




×







246



  }










247













UNCOV

248



  if (code < 0) {





×







249



    goto _OVER;




×







250



  }










251













UNCOV

252



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




×








UNCOV

253



  if (res == NULL) goto _OVER;





×








UNCOV

254



  int32_t ressize = tjsonGetArraySize(res);




×








UNCOV

255



  bool    returnConf = (pSupp->resHighSlot != -1 || pSupp->resLowSlot != -1);





×







256













UNCOV

257



  if ((returnConf && (ressize != 4)) || ((!returnConf) && (ressize != 2))) {





×







258



    goto _OVER;




×







259



  }










260













UNCOV

261



  if (pResTsCol != NULL) {





×








UNCOV

262



    resCurRow = pBlock->info.rows;




×








UNCOV

263



    SJson* tsJsonArray = tjsonGetArrayItem(res, 0);




×








UNCOV

264



    if (tsJsonArray == NULL) goto _OVER;





×








UNCOV

265



    int32_t tsSize = tjsonGetArraySize(tsJsonArray);




×








UNCOV

266



    if (tsSize != rows) goto _OVER;





×








UNCOV

267



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





×








UNCOV

268



      SJson* tsJson = tjsonGetArrayItem(tsJsonArray, i);




×








UNCOV

269



      tjsonGetObjectValueBigInt(tsJson, &tmpI64);




×








UNCOV

270



      colDataSetInt64(pResTsCol, resCurRow, &tmpI64);




×








UNCOV

271



      resCurRow++;




×







272



    }










273



  }










274













UNCOV

275



  if (pResLowCol != NULL) {





×








UNCOV

276



    resCurRow = pBlock->info.rows;




×








UNCOV

277



    SJson* lowJsonArray = tjsonGetArrayItem(res, 2);




×








UNCOV

278



    if (lowJsonArray == NULL) goto _OVER;





×








UNCOV

279



    int32_t lowSize = tjsonGetArraySize(lowJsonArray);




×








UNCOV

280



    if (lowSize != rows) goto _OVER;





×








UNCOV

281



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





×








UNCOV

282



      SJson* lowJson = tjsonGetArrayItem(lowJsonArray, i);




×








UNCOV

283



      tjsonGetObjectValueDouble(lowJson, &tmpDouble);




×








UNCOV

284



      tmpFloat = (float)tmpDouble;




×








UNCOV

285



      colDataSetFloat(pResLowCol, resCurRow, &tmpFloat);




×








UNCOV

286



      resCurRow++;




×







287



    }










288



  }










289













UNCOV

290



  if (pResHighCol != NULL) {





×








UNCOV

291



    resCurRow = pBlock->info.rows;




×








UNCOV

292



    SJson* highJsonArray = tjsonGetArrayItem(res, 3);




×








UNCOV

293



    if (highJsonArray == NULL) goto _OVER;





×








UNCOV

294



    int32_t highSize = tjsonGetArraySize(highJsonArray);




×








UNCOV

295



    if (highSize != rows) goto _OVER;





×








UNCOV

296



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





×








UNCOV

297



      SJson* highJson = tjsonGetArrayItem(highJsonArray, i);




×








UNCOV

298



      tjsonGetObjectValueDouble(highJson, &tmpDouble);




×








UNCOV

299



      tmpFloat = (float)tmpDouble;




×








UNCOV

300



      colDataSetFloat(pResHighCol, resCurRow, &tmpFloat);




×








UNCOV

301



      resCurRow++;




×







302



    }










303



  }










304













UNCOV

305



  resCurRow = pBlock->info.rows;




×








UNCOV

306



  SJson* valJsonArray = tjsonGetArrayItem(res, 1);




×








UNCOV

307



  if (valJsonArray == NULL) goto _OVER;





×








UNCOV

308



  int32_t valSize = tjsonGetArraySize(valJsonArray);




×








UNCOV

309



  if (valSize != rows) goto _OVER;





×








UNCOV

310



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





×








UNCOV

311



    SJson* valJson = tjsonGetArrayItem(valJsonArray, i);




×








UNCOV

312



    tjsonGetObjectValueDouble(valJson, &tmpDouble);




×







313













UNCOV

314



    switch (pSupp->inputValType) {





×








UNCOV

315



      case TSDB_DATA_TYPE_BOOL:




×







316



      case TSDB_DATA_TYPE_UTINYINT:










317



      case TSDB_DATA_TYPE_TINYINT: {











UNCOV

318



        tmpI8 = (int8_t)tmpDouble;




×








UNCOV

319



        colDataSetInt8(pResValCol, resCurRow, &tmpI8);




×








UNCOV

320



        break;




×







321



      }










322



      case TSDB_DATA_TYPE_USMALLINT:




×







323



      case TSDB_DATA_TYPE_SMALLINT: {










324



        tmpI16 = (int16_t)tmpDouble;




×







325



        colDataSetInt16(pResValCol, resCurRow, &tmpI16);




×







326



        break;




×







327



      }











UNCOV

328



      case TSDB_DATA_TYPE_INT:




×







329



      case TSDB_DATA_TYPE_UINT: {











UNCOV

330



        tmpI32 = (int32_t)tmpDouble;




×








UNCOV

331



        colDataSetInt32(pResValCol, resCurRow, &tmpI32);




×








UNCOV

332



        break;




×







333



      }











UNCOV

334



      case TSDB_DATA_TYPE_TIMESTAMP:




×







335



      case TSDB_DATA_TYPE_UBIGINT:










336



      case TSDB_DATA_TYPE_BIGINT: {











UNCOV

337



        tmpI64 = (int64_t)tmpDouble;




×








UNCOV

338



        colDataSetInt64(pResValCol, resCurRow, &tmpI64);




×








UNCOV

339



        break;




×







340



      }











UNCOV

341



      case TSDB_DATA_TYPE_FLOAT: {




×








UNCOV

342



        tmpFloat = (float)tmpDouble;




×








UNCOV

343



        colDataSetFloat(pResValCol, resCurRow, &tmpFloat);




×








UNCOV

344



        break;




×







345



      }











UNCOV

346



      case TSDB_DATA_TYPE_DOUBLE: {




×








UNCOV

347



        colDataSetDouble(pResValCol, resCurRow, &tmpDouble);




×








UNCOV

348



        break;




×







349



      }










350



      default:




×







351



        code = TSDB_CODE_FUNC_FUNTION_PARA_TYPE;




×







352



        goto _OVER;




×







353



    }











UNCOV

354



    resCurRow++;




×







355



  }










356













UNCOV

357



  pBlock->info.rows += rows;




×







358













UNCOV

359



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





×








UNCOV

360



  return 0;




×







361












362



_OVER:




×







363



  tjsonDelete(pJson);




×







364



  if (code == 0) {





×







365



    code = TSDB_CODE_INVALID_JSON_FORMAT;




×







366



  }










367












368



  qError("%s failed to perform forecast finalize since %s", pId, tstrerror(code));





×







369



  return code;




×







370



}










371













UNCOV

372



static int32_t forecastAggregateBlocks(SForecastSupp* pSupp, SSDataBlock* pResBlock, const char* pId) {




×








UNCOV

373



  int32_t       code = TSDB_CODE_SUCCESS;




×








UNCOV

374



  int32_t       lino = 0;




×








UNCOV

375



  SAnalyticBuf* pBuf = &pSupp->analyBuf;




×







376













UNCOV

377



  code = forecastCloseBuf(pSupp, pId);




×








UNCOV

378



  QUERY_CHECK_CODE(code, lino, _end);





×







379













UNCOV

380



  code = forecastEnsureBlockCapacity(pResBlock, 1);




×








UNCOV

381



  QUERY_CHECK_CODE(code, lino, _end);





×







382













UNCOV

383



  code = forecastAnalysis(pSupp, pResBlock, pId);




×








UNCOV

384



  QUERY_CHECK_CODE(code, lino, _end);





×







385













UNCOV

386



  uInfo("%s block:%d, forecast finalize", pId, pSupp->numOfBlocks);





×







387












388



_end:




×








UNCOV

389



  pSupp->numOfBlocks = 0;




×








UNCOV

390



  taosAnalyBufDestroy(&pSupp->analyBuf);




×








UNCOV

391



  return code;




×







392



}










393













UNCOV

394



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




×








UNCOV

395



  int32_t                code = TSDB_CODE_SUCCESS;




×








UNCOV

396



  int32_t                lino = 0;




×








UNCOV

397



  SExecTaskInfo*         pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

398



  SForecastOperatorInfo* pInfo = pOperator->info;




×








UNCOV

399



  SSDataBlock*           pResBlock = pInfo->pRes;




×








UNCOV

400



  SForecastSupp*         pSupp = &pInfo->forecastSupp;




×








UNCOV

401



  SAnalyticBuf*          pBuf = &pSupp->analyBuf;




×








UNCOV

402



  int64_t                st = taosGetTimestampUs();




×








UNCOV

403



  int32_t                numOfBlocks = pSupp->numOfBlocks;




×








UNCOV

404



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




×







405













UNCOV

406



  blockDataCleanup(pResBlock);




×







407













UNCOV

408



  while (1) {




×








UNCOV

409



    SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);




×








UNCOV

410



    if (pBlock == NULL) {





×








UNCOV

411



      break;




×







412



    }










413













UNCOV

414



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





×








UNCOV

415



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




×








UNCOV

416



      numOfBlocks++;




×








UNCOV

417



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




×








UNCOV

418



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





×







419



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











UNCOV

420



      code = forecastCacheBlock(pSupp, pBlock, pId);




×








UNCOV

421



      QUERY_CHECK_CODE(code, lino, _end);





×







422



    } else {










423



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





×







424



      code = forecastAggregateBlocks(pSupp, pResBlock, pId);




×







425



      QUERY_CHECK_CODE(code, lino, _end);





×







426



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




×







427



      numOfBlocks = 1;




×







428



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




×







429



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





×







430



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










431



      code = forecastCacheBlock(pSupp, pBlock, pId);




×







432



      QUERY_CHECK_CODE(code, lino, _end);





×







433



    }










434













UNCOV

435



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





×







436



      (*ppRes) = pResBlock;




×







437



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





×







438



      return code;




×







439



    }










440



  }










441













UNCOV

442



  if (numOfBlocks > 0) {





×








UNCOV

443



    qDebug("%s group:%" PRId64 ", read finish, blocks:%d", pId, pSupp->groupId, numOfBlocks);





×








UNCOV

444



    code = forecastAggregateBlocks(pSupp, pResBlock, pId);




×








UNCOV

445



    QUERY_CHECK_CODE(code, lino, _end);





×







446



  }










447













UNCOV

448



  int64_t cost = taosGetTimestampUs() - st;




×








UNCOV

449



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





×







450












451



_end:




×








UNCOV

452



  if (code != TSDB_CODE_SUCCESS) {





×








UNCOV

453



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





×








UNCOV

454



    pTaskInfo->code = code;




×








UNCOV

455



    T_LONG_JMP(pTaskInfo->env, code);





×







456



  }










457













UNCOV

458



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





×








UNCOV

459



  return code;




×







460



}










461













UNCOV

462



static int32_t forecastParseOutput(SForecastSupp* pSupp, SExprSupp* pExprSup) {




×








UNCOV

463



  pSupp->resLowSlot = -1;




×








UNCOV

464



  pSupp->resHighSlot = -1;




×








UNCOV

465



  pSupp->resTsSlot = -1;




×








UNCOV

466



  pSupp->resValSlot = -1;




×







467













UNCOV

468



  for (int32_t j = 0; j < pExprSup->numOfExprs; ++j) {





×








UNCOV

469



    SExprInfo* pExprInfo = &pExprSup->pExprInfo[j];




×








UNCOV

470



    int32_t    dstSlot = pExprInfo->base.resSchema.slotId;




×








UNCOV

471



    if (pExprInfo->pExpr->_function.functionType == FUNCTION_TYPE_FORECAST) {





×








UNCOV

472



      pSupp->resValSlot = dstSlot;




×








UNCOV

473



    } else if (pExprInfo->pExpr->_function.functionType == FUNCTION_TYPE_FORECAST_ROWTS) {





×








UNCOV

474



      pSupp->resTsSlot = dstSlot;




×








UNCOV

475



    } else if (pExprInfo->pExpr->_function.functionType == FUNCTION_TYPE_FORECAST_LOW) {





×








UNCOV

476



      pSupp->resLowSlot = dstSlot;




×








UNCOV

477



    } else if (pExprInfo->pExpr->_function.functionType == FUNCTION_TYPE_FORECAST_HIGH) {





×








UNCOV

478



      pSupp->resHighSlot = dstSlot;




×







479



    } else {










480



    }










481



  }










482













UNCOV

483



  return 0;




×







484



}










485













UNCOV

486



static int32_t forecastParseInput(SForecastSupp* pSupp, SNodeList* pFuncs) {




×








UNCOV

487



  SNode* pNode = NULL;




×







488













UNCOV

489



  pSupp->inputTsSlot = -1;




×








UNCOV

490



  pSupp->inputValSlot = -1;




×








UNCOV

491



  pSupp->inputValType = -1;




×








UNCOV

492



  pSupp->inputPrecision = -1;




×







493













UNCOV

494



  FOREACH(pNode, pFuncs) {





×








UNCOV

495



    if ((nodeType(pNode) == QUERY_NODE_TARGET) && (nodeType(((STargetNode*)pNode)->pExpr) == QUERY_NODE_FUNCTION)) {





×








UNCOV

496



      SFunctionNode* pFunc = (SFunctionNode*)((STargetNode*)pNode)->pExpr;




×








UNCOV

497



      int32_t        numOfParam = LIST_LENGTH(pFunc->pParameterList);





×







498













UNCOV

499



      if (pFunc->funcType == FUNCTION_TYPE_FORECAST) {





×








UNCOV

500



        if (numOfParam == 3) {





×








UNCOV

501



          SNode* p1 = nodesListGetNode(pFunc->pParameterList, 0);




×








UNCOV

502



          SNode* p2 = nodesListGetNode(pFunc->pParameterList, 1);




×








UNCOV

503



          SNode* p3 = nodesListGetNode(pFunc->pParameterList, 2);




×








UNCOV

504



          if (p1 == NULL || p2 == NULL || p3 == NULL) return TSDB_CODE_PLAN_INTERNAL_ERROR;





×








UNCOV

505



          if (p1->type != QUERY_NODE_COLUMN) return TSDB_CODE_PLAN_INTERNAL_ERROR;





×








UNCOV

506



          if (p2->type != QUERY_NODE_VALUE) return TSDB_CODE_PLAN_INTERNAL_ERROR;





×








UNCOV

507



          if (p3->type != QUERY_NODE_COLUMN) return TSDB_CODE_PLAN_INTERNAL_ERROR;





×








UNCOV

508



          SColumnNode* pValNode = (SColumnNode*)p1;




×








UNCOV

509



          SValueNode*  pOptNode = (SValueNode*)p2;




×








UNCOV

510



          SColumnNode* pTsNode = (SColumnNode*)p3;




×








UNCOV

511



          pSupp->inputTsSlot = pTsNode->slotId;




×








UNCOV

512



          pSupp->inputPrecision = pTsNode->node.resType.precision;




×








UNCOV

513



          pSupp->inputValSlot = pValNode->slotId;




×








UNCOV

514



          pSupp->inputValType = pValNode->node.resType.type;




×








UNCOV

515



          tstrncpy(pSupp->algoOpt, pOptNode->literal, sizeof(pSupp->algoOpt));




×







516



        } else if (numOfParam == 2) {





×







517



          SNode* p1 = nodesListGetNode(pFunc->pParameterList, 0);




×







518



          SNode* p2 = nodesListGetNode(pFunc->pParameterList, 1);




×







519



          if (p1 == NULL || p2 == NULL) return TSDB_CODE_PLAN_INTERNAL_ERROR;





×







520



          if (p1->type != QUERY_NODE_COLUMN) return TSDB_CODE_PLAN_INTERNAL_ERROR;





×







521



          if (p2->type != QUERY_NODE_COLUMN) return TSDB_CODE_PLAN_INTERNAL_ERROR;





×







522



          SColumnNode* pValNode = (SColumnNode*)p1;




×







523



          SColumnNode* pTsNode = (SColumnNode*)p2;




×







524



          pSupp->inputTsSlot = pTsNode->slotId;




×







525



          pSupp->inputPrecision = pTsNode->node.resType.precision;




×







526



          pSupp->inputValSlot = pValNode->slotId;




×







527



          pSupp->inputValType = pValNode->node.resType.type;




×







528



          tstrncpy(pSupp->algoOpt, "algo=arima", TSDB_ANALYTIC_ALGO_OPTION_LEN);




×







529



        } else {










530



          return TSDB_CODE_PLAN_INTERNAL_ERROR;




×







531



        }










532



      }










533



    }










534



  }










535













UNCOV

536



  return 0;




×







537



}










538













UNCOV

539



static int32_t forecastParseAlgo(SForecastSupp* pSupp, const char* id) {




×








UNCOV

540



  pSupp->maxTs = 0;




×








UNCOV

541



  pSupp->minTs = INT64_MAX;




×








UNCOV

542



  pSupp->numOfRows = 0;




×







543













UNCOV

544



  if (!taosAnalyGetOptStr(pSupp->algoOpt, "algo", pSupp->algoName, sizeof(pSupp->algoName))) {





×







545



    qError("%s failed to get forecast algorithm name from %s", id, pSupp->algoOpt);





×







546



    return TSDB_CODE_ANA_ALGO_NOT_FOUND;




×







547



  }










548













UNCOV

549



  bool hasTimeout = taosAnalyGetOptInt(pSupp->algoOpt, "timeout", &pSupp->timeout);




×








UNCOV

550



  if (!hasTimeout) {





×








UNCOV

551



    qDebug("%s not set the timeout val, set default:%d", id, ANALY_DEFAULT_TIMEOUT);





×








UNCOV

552



    pSupp->timeout = ANALY_DEFAULT_TIMEOUT;




×







553



  } else {











UNCOV

554



    if (pSupp->timeout <= 0 || pSupp->timeout > ANALY_MAX_TIMEOUT) {





×








UNCOV

555



      qDebug("%s timeout val:%" PRId64 "s is invalid (greater than 10min or less than 1s), use default:%dms",





×







556



             id, pSupp->timeout, ANALY_DEFAULT_TIMEOUT);











UNCOV

557



      pSupp->timeout = ANALY_DEFAULT_TIMEOUT;




×







558



    } else {











UNCOV

559



      qDebug("%s timeout val is set to: %" PRId64 "s", id, pSupp->timeout);





×







560



    }










561



  }










562













UNCOV

563



  if (taosAnalyGetAlgoUrl(pSupp->algoName, ANALY_ALGO_TYPE_FORECAST, pSupp->algoUrl, sizeof(pSupp->algoUrl)) != 0) {





×








UNCOV

564



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





×








UNCOV

565



    return TSDB_CODE_ANA_ALGO_NOT_LOAD;




×







566



  }










567













UNCOV

568



  return 0;




×







569



}










570













UNCOV

571



static int32_t forecastCreateBuf(SForecastSupp* pSupp) {




×








UNCOV

572



  SAnalyticBuf* pBuf = &pSupp->analyBuf;




×








UNCOV

573



  int64_t       ts = 0;  // taosGetTimestampMs();




×







574













UNCOV

575



  pBuf->bufType = ANALYTICS_BUF_TYPE_JSON_COL;




×








UNCOV

576



  snprintf(pBuf->fileName, sizeof(pBuf->fileName), "%s/tdengine-forecast-%" PRId64, tsTempDir, ts);




×








UNCOV

577



  int32_t code = tsosAnalyBufOpen(pBuf, 2);




×








UNCOV

578



  if (code != 0) goto _OVER;





×







579













UNCOV

580



  code = taosAnalyBufWriteColMeta(pBuf, 0, TSDB_DATA_TYPE_TIMESTAMP, "ts");




×








UNCOV

581



  if (code != 0) goto _OVER;





×







582













UNCOV

583



  code = taosAnalyBufWriteColMeta(pBuf, 1, pSupp->inputValType, "val");




×








UNCOV

584



  if (code != 0) goto _OVER;





×







585













UNCOV

586



  code = taosAnalyBufWriteDataBegin(pBuf);




×








UNCOV

587



  if (code != 0) goto _OVER;





×







588













UNCOV

589



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





×








UNCOV

590



    code = taosAnalyBufWriteColBegin(pBuf, i);




×








UNCOV

591



    if (code != 0) goto _OVER;





×







592



  }










593













UNCOV

594



_OVER:




×








UNCOV

595



  if (code != 0) {





×







596



    (void)taosAnalyBufClose(pBuf);




×







597



    taosAnalyBufDestroy(pBuf);




×







598



  }











UNCOV

599



  return code;




×







600



}










601













UNCOV

602



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




×







603



                                   SOperatorInfo** pOptrInfo) {











UNCOV

604



  QRY_PARAM_CHECK(pOptrInfo);





×







605













UNCOV

606



  int32_t                code = 0;




×








UNCOV

607



  int32_t                lino = 0;




×








UNCOV

608



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





×








UNCOV

609



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





×








UNCOV

610



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





×







611



    code = terrno;




×







612



    goto _error;




×







613



  }










614













UNCOV

615



  const char*             pId = pTaskInfo->id.str;




×








UNCOV

616



  SForecastSupp*          pSupp = &pInfo->forecastSupp;




×








UNCOV

617



  SForecastFuncPhysiNode* pForecastPhyNode = (SForecastFuncPhysiNode*)pPhyNode;




×








UNCOV

618



  SExprSupp*              pExprSup = &pOperator->exprSupp;




×








UNCOV

619



  int32_t                 numOfExprs = 0;




×








UNCOV

620



  SExprInfo*              pExprInfo = NULL;




×







621













UNCOV

622



  code = createExprInfo(pForecastPhyNode->pFuncs, NULL, &pExprInfo, &numOfExprs);




×








UNCOV

623



  QUERY_CHECK_CODE(code, lino, _error);





×







624













UNCOV

625



  code = initExprSupp(pExprSup, pExprInfo, numOfExprs, &pTaskInfo->storageAPI.functionStore);




×








UNCOV

626



  QUERY_CHECK_CODE(code, lino, _error);





×







627













UNCOV

628



  if (pForecastPhyNode->pExprs != NULL) {





×







629



    int32_t    num = 0;




×







630



    SExprInfo* pScalarExprInfo = NULL;




×







631



    code = createExprInfo(pForecastPhyNode->pExprs, NULL, &pScalarExprInfo, &num);




×







632



    QUERY_CHECK_CODE(code, lino, _error);





×







633












634



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




×







635



    QUERY_CHECK_CODE(code, lino, _error);





×







636



  }










637













UNCOV

638



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




×








UNCOV

639



  QUERY_CHECK_CODE(code, lino, _error);





×







640













UNCOV

641



  code = forecastParseInput(pSupp, pForecastPhyNode->pFuncs);




×








UNCOV

642



  QUERY_CHECK_CODE(code, lino, _error);





×







643













UNCOV

644



  code = forecastParseOutput(pSupp, pExprSup);




×








UNCOV

645



  QUERY_CHECK_CODE(code, lino, _error);





×







646













UNCOV

647



  code = forecastParseAlgo(pSupp, pId);




×








UNCOV

648



  QUERY_CHECK_CODE(code, lino, _error);





×







649













UNCOV

650



  code = forecastCreateBuf(pSupp);




×








UNCOV

651



  QUERY_CHECK_CODE(code, lino, _error);





×







652













UNCOV

653



  initResultSizeInfo(&pOperator->resultInfo, 4096);




×







654













UNCOV

655



  pInfo->pRes = createDataBlockFromDescNode(pPhyNode->pOutputDataBlockDesc);




×








UNCOV

656



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





×







657













UNCOV

658



  setOperatorInfo(pOperator, "ForecastOperator", QUERY_NODE_PHYSICAL_PLAN_FORECAST_FUNC, false, OP_NOT_OPENED, pInfo,




×







659



                  pTaskInfo);











UNCOV

660



  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, forecastNext, NULL, destroyForecastInfo, optrDefaultBufFn,




×







661



                                         NULL, optrDefaultGetNextExtFn, NULL);










662













UNCOV

663



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




×








UNCOV

664



  QUERY_CHECK_CODE(code, lino, _error);





×







665













UNCOV

666



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




×








UNCOV

667



  QUERY_CHECK_CODE(code, lino, _error);





×







668













UNCOV

669



  *pOptrInfo = pOperator;




×







670













UNCOV

671



  qDebug("%s forecast env is initialized, option:%s", pId, pSupp->algoOpt);





×








UNCOV

672



  return TSDB_CODE_SUCCESS;




×







673













UNCOV

674



_error:




×








UNCOV

675



  if (code != TSDB_CODE_SUCCESS) {





×








UNCOV

676



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





×







677



  }











UNCOV

678



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





×








UNCOV

679



  destroyOperatorAndDownstreams(pOperator, &downstream, 1);




×








UNCOV

680



  pTaskInfo->code = code;




×








UNCOV

681



  return code;




×







682



}










683













UNCOV

684



static void destroyForecastInfo(void* param) {




×








UNCOV

685



  SForecastOperatorInfo* pInfo = (SForecastOperatorInfo*)param;




×







686













UNCOV

687



  blockDataDestroy(pInfo->pRes);




×








UNCOV

688



  pInfo->pRes = NULL;




×








UNCOV

689



  cleanupExprSupp(&pInfo->scalarSup);




×








UNCOV

690



  taosAnalyBufDestroy(&pInfo->forecastSupp.analyBuf);




×








UNCOV

691



  taosMemoryFreeClear(param);





×








UNCOV

692



}




×







693












694



#else










695












696



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










697



                                   SOperatorInfo** pOptrInfo) {










698



  return TSDB_CODE_OPS_NOT_SUPPORT;










699



}










700












701



#endif
























    

  • 

    •