taosdata/TDengine | Build 4856 | source/libs/executor/src/forecastoperator.c | Coveralls

85

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

×

86

    return TSDB_CODE_SUCCESS;

×

89

  int32_t code = blockDataEnsureCapacity(pBlock, newRowsNum);

×

90

  if (code != TSDB_CODE_SUCCESS) {

×

91

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

×

92

    return code;

×

95

  return TSDB_CODE_SUCCESS;

×

98

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

×

99

  int32_t       code = TSDB_CODE_SUCCESS;

×

100

  int32_t       lino = 0;

×

101

  SAnalyticBuf* pBuf = &pSupp->analyBuf;

×

103

  if (pSupp->cachedRows > ANALY_FORECAST_MAX_ROWS) {

×

104

    code = TSDB_CODE_ANA_ANODE_TOO_MANY_ROWS;

×

105

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

×

107

    return code;

×

110

  pSupp->numOfBlocks++;

×

111

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

×

113

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

×

114

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

×

115

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

×

116

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

×

118

    int32_t index = 0;

×

119

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

×

120

    char*   val = colDataGetData(pValCol, j);

×

121

    int16_t valType = pValCol->info.type;

×

123

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

×

124

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

×

125

    pSupp->numOfRows++;

×

128

    code = taosAnalyBufWriteColData(pBuf, index++, TSDB_DATA_TYPE_TIMESTAMP, &ts);

×

129

    if (TSDB_CODE_SUCCESS != code) {

×

130

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

×

131

      return code;

×

135

    code = taosAnalyBufWriteColData(pBuf, index++, valType, val);

×

136

    if (TSDB_CODE_SUCCESS != code) {

×

137

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

×

138

      return code;

×

142

    for (int32_t i = 0; i < taosArrayGetSize(pSupp->pDynamicRealList); ++i) {

×

143

      SColFutureData*  pCol = taosArrayGet(pSupp->pDynamicRealList, i);

×

144

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

×

146

      char* pVal = colDataGetData(pColData, j);

×

147

      code = taosAnalyBufWriteColData(pBuf, index++, pCol->col.type, pVal);

×

148

      if (TSDB_CODE_SUCCESS != code) {

×

149

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

×

150

        return code;

×

155

    for (int32_t i = 0; i < taosArrayGetSize(pSupp->pCovariateSlotList); ++i) {

×

156

      SColumn*         pCol = taosArrayGet(pSupp->pCovariateSlotList, i);

×

157

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

×

159

      char* pVal = colDataGetData(pColData, j);

×

160

      code = taosAnalyBufWriteColData(pBuf, index++, pCol->type, pVal);

×

161

      if (TSDB_CODE_SUCCESS != code) {

×

162

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

×

163

        return code;

×

168

  return 0;

×

171

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

×

172

  SAnalyticBuf* pBuf = &pSupp->analyBuf;

×

173

  int32_t       code = 0;

×

177

  for (int32_t i = 0; i < pSupp->numOfInputCols; ++i) {

×

180

    if ((i >= 2) && ((i - 2) < taosArrayGetSize(pSupp->pDynamicRealList))) {

×

181

      SColFutureData* pCol = taosArrayGet(pSupp->pDynamicRealList, i - 2);

×

182

      int16_t         valType = pCol->col.type;

×

183

      for (int32_t j = 0; j < pCol->numOfRows; ++j) {

×

184

        char* pVal = colDataGetData(&pCol->data, j);

×

185

        code = taosAnalyBufWriteColData(pBuf, i, valType, pVal);

×

186

        if (code != 0) {

×

187

          return code;

×

192

    code = taosAnalyBufWriteColEnd(pBuf, i);

×

193

    if (code != 0) return code;

×

196

  code = taosAnalyBufWriteDataEnd(pBuf);

×

197

  if (code != 0) return code;

×

199

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

×

200

  if (code != 0) return code;

×

202

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

×

203

  if (code != 0) return code;

×

205

  const char* prec = TSDB_TIME_PRECISION_MILLI_STR;

×

206

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

×

207

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

×

208

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

×

209

  if (code != 0) return code;

×

211

  code = taosAnalyBufWriteOptInt(pBuf, ALGO_OPT_WNCHECK_NAME, pSupp->wncheck);

×

212

  if (code != 0) return code;

×

214

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

×

215

  code = taosAnalyBufWriteOptInt(pBuf, ALGO_OPT_RETCONF_NAME, !noConf);

×

216

  if (code != 0) return code;

×

218

  if (pSupp->cachedRows < ANALY_FORECAST_MIN_ROWS) {

×

219

    qError("%s history rows for forecasting not enough, min required:%d, current:%" PRId64, id, ANALY_FORECAST_MIN_ROWS,

×

221

    return TSDB_CODE_ANA_ANODE_NOT_ENOUGH_ROWS;

×

224

  if (pSupp->cachedRows < ANALY_TDTSFM_FORECAST_MIN_ROWS && strcmp(pSupp->algoName, "tdtsfm_1") == 0) {

×

225

    qError("%s history rows for forecasting when using tdtsfm model not enough, min required:%d, current:%" PRId64, id,

×

227

    return TSDB_CODE_ANA_ANODE_NOT_ENOUGH_ROWS;

×

230

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

×

231

  if (code != 0) return code;

×

233

  code = taosAnalyBufWriteOptFloat(pBuf, "conf", pSupp->conf);

×

234

  if (code != 0) return code;

×

236

  int32_t len = strlen(pSupp->pOptions);

×

237

  int64_t every = (pSupp->setEvery != 0) ? pSupp->every : ((pSupp->maxTs - pSupp->minTs) / (pSupp->numOfRows - 1));

×

238

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

×

239

  if (code != 0) return code;

×

241

  int64_t start = (pSupp->setStart != 0) ? pSupp->startTs : pSupp->maxTs + every;

×

242

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

×

243

  if (code != 0) return code;

×

245

  if (taosArrayGetSize(pSupp->pCovariateSlotList) + taosArrayGetSize(pSupp->pDynamicRealList) > 0) {

×

246

    code = taosAnalyBufWriteOptStr(pBuf, "type", "covariate");

×

247

    if (code != 0) return code;

×

250

  code = taosAnalyBufClose(pBuf);

×

251

  return code;

×

254

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

×

255

  SAnalyticBuf* pBuf = &pSupp->analyBuf;

×

256

  int32_t       resCurRow = pBlock->info.rows;

×

257

  int64_t       tmpI64 = 0;

×

258

  double        tmpDouble = 0;

×

259

  int32_t       code = 0;

×

261

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

×

262

  if (NULL == pResValCol) {

×

263

    return terrno;

×

266

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

×

267

  SColumnInfoData* pResLowCol =

×

268

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

×

269

  SColumnInfoData* pResHighCol =

×

270

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

×

272

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

×

273

  if (pJson == NULL) {

×

274

    return terrno;

×

277

  int32_t rows = 0;

×

278

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

×

279

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

×

NEW

280

    code = parseErrorMsgFromAnalyticServer(pJson, pId);

×

281

    tjsonDelete(pJson);

×

282

    return code;

×

286

  if (code != 0) {

×

287

    goto _OVER;

×

290

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

×

UNCOV

291

  if (res == NULL) goto _OVER;

×

UNCOV

292

  int32_t ressize = tjsonGetArraySize(res);

×

UNCOV

293

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

×

295

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

×

296

    goto _OVER;

×

299

  if (pResTsCol != NULL) {

×

300

    resCurRow = pBlock->info.rows;

×

301

    SJson* tsJsonArray = tjsonGetArrayItem(res, 0);

×

302

    if (tsJsonArray == NULL) goto _OVER;

×

303

    int32_t tsSize = tjsonGetArraySize(tsJsonArray);

×

304

    if (tsSize != rows) goto _OVER;

×

UNCOV

305

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

×

UNCOV

306

      SJson* tsJson = tjsonGetArrayItem(tsJsonArray, i);

×

UNCOV

307

      tjsonGetObjectValueBigInt(tsJson, &tmpI64);

×

308

      colDataSetInt64(pResTsCol, resCurRow, &tmpI64);

×

309

      resCurRow++;

×

313

  if (pResLowCol != NULL) {

×

314

    resCurRow = pBlock->info.rows;

×

315

    SJson* lowJsonArray = tjsonGetArrayItem(res, 2);

×

316

    if (lowJsonArray == NULL) goto _OVER;

×

317

    int32_t lowSize = tjsonGetArraySize(lowJsonArray);

×

318

    if (lowSize != rows) goto _OVER;

×

UNCOV

319

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

×

UNCOV

320

      SJson* lowJson = tjsonGetArrayItem(lowJsonArray, i);

×

UNCOV

321

      tjsonGetObjectValueDouble(lowJson, &tmpDouble);

×

322

      colDataSetDouble(pResLowCol, resCurRow, &tmpDouble);

×

323

      resCurRow++;

×

327

  if (pResHighCol != NULL) {

×

328

    resCurRow = pBlock->info.rows;

×

329

    SJson* highJsonArray = tjsonGetArrayItem(res, 3);

×

UNCOV

330

    if (highJsonArray == NULL) goto _OVER;

×

331

    int32_t highSize = tjsonGetArraySize(highJsonArray);

×

332

    if (highSize != rows) goto _OVER;

×

UNCOV

333

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

×

UNCOV

334

      SJson* highJson = tjsonGetArrayItem(highJsonArray, i);

×

335

      tjsonGetObjectValueDouble(highJson, &tmpDouble);

×

UNCOV

336

      colDataSetDouble(pResHighCol, resCurRow, &tmpDouble);

×

337

      resCurRow++;

×

341

  resCurRow = pBlock->info.rows;

×

342

  SJson* valJsonArray = tjsonGetArrayItem(res, 1);

×

343

  if (valJsonArray == NULL) goto _OVER;

×

UNCOV

344

  int32_t valSize = tjsonGetArraySize(valJsonArray);

×

UNCOV

345

  if (valSize != rows) goto _OVER;

×

346

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

×

347

    SJson* valJson = tjsonGetArrayItem(valJsonArray, i);

×

UNCOV

348

    tjsonGetObjectValueDouble(valJson, &tmpDouble);

×

350

    colDataSetDouble(pResValCol, resCurRow, &tmpDouble);

×

351

    resCurRow++;

×

UNCOV

354

  pBlock->info.rows += rows;

×

356

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

×

UNCOV

357

  return 0;

×

359

_OVER:

×

UNCOV

360

  tjsonDelete(pJson);

×

361

  if (code == 0) {

×

362

    code = TSDB_CODE_INVALID_JSON_FORMAT;

×

UNCOV

365

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

×

366

  return code;

×

369

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

×

UNCOV

370

  int32_t       code = TSDB_CODE_SUCCESS;

×

UNCOV

371

  int32_t       lino = 0;

×

372

  SAnalyticBuf* pBuf = &pSupp->analyBuf;

×

374

  code = forecastCloseBuf(pSupp, pId);

×

375

  QUERY_CHECK_CODE(code, lino, _end);

×

377

  code = forecastEnsureBlockCapacity(pResBlock, 1);

×

378

  QUERY_CHECK_CODE(code, lino, _end);

×

380

  code = forecastAnalysis(pSupp, pResBlock, pId);

×

381

  QUERY_CHECK_CODE(code, lino, _end);

×

383

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

×

385

_end:

×

UNCOV

386

  pSupp->numOfBlocks = 0;

×

387

  taosAnalyBufDestroy(&pSupp->analyBuf);

×

388

  return code;

×

UNCOV

391

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

×

UNCOV

392

  int32_t                code = TSDB_CODE_SUCCESS;

×

393

  int32_t                lino = 0;

×

394

  SExecTaskInfo*         pTaskInfo = pOperator->pTaskInfo;

×

395

  SForecastOperatorInfo* pInfo = pOperator->info;

×

396

  SSDataBlock*           pResBlock = pInfo->pRes;

×

397

  SForecastSupp*         pSupp = &pInfo->forecastSupp;

×

UNCOV

398

  SExprSupp*             pScalarSupp = &pInfo->scalarSup;

×

UNCOV

399

  SAnalyticBuf*          pBuf = &pSupp->analyBuf;

×

UNCOV

400

  int64_t                st = taosGetTimestampUs();

×

401

  int32_t                numOfBlocks = pSupp->numOfBlocks;

×

402

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

×

404

  blockDataCleanup(pResBlock);

×

UNCOV

406

  while (1) {

×

407

    SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);

×

408

    if (pBlock == NULL) {

×

UNCOV

409

      break;

×

412

    if (pScalarSupp->pExprInfo != NULL) {

×

413

      code = projectApplyFunctions(pScalarSupp->pExprInfo, pBlock, pBlock, pScalarSupp->pCtx, pScalarSupp->numOfExprs,

×

414

                                   NULL, GET_STM_RTINFO(pOperator->pTaskInfo));

×

415

      if (code != TSDB_CODE_SUCCESS) {

×

416

        T_LONG_JMP(pTaskInfo->env, code);

×

UNCOV

420

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

×

UNCOV

421

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

×

422

      numOfBlocks++;

×

423

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

×

424

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

×

UNCOV

426

      code = forecastCacheBlock(pSupp, pBlock, pId);

×

UNCOV

427

      QUERY_CHECK_CODE(code, lino, _end);

×

429

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

×

430

      code = forecastAggregateBlocks(pSupp, pResBlock, pId);

×

431

      QUERY_CHECK_CODE(code, lino, _end);

×

432

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

×

UNCOV

433

      numOfBlocks = 1;

×

UNCOV

434

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

×

435

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

×

UNCOV

437

      code = forecastCacheBlock(pSupp, pBlock, pId);

×

438

      QUERY_CHECK_CODE(code, lino, _end);

×

441

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

×

442

      (*ppRes) = pResBlock;

×

UNCOV

443

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

×

UNCOV

444

      return code;

×

UNCOV

448

  if (numOfBlocks > 0) {

×

449

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

×

450

    code = forecastAggregateBlocks(pSupp, pResBlock, pId);

×

451

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

454

  int64_t cost = taosGetTimestampUs() - st;

×

455

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

×

457

_end:

×

458

  if (code != TSDB_CODE_SUCCESS) {

×

459

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

×

460

    pTaskInfo->code = code;

×

461

    T_LONG_JMP(pTaskInfo->env, code);

×

464

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

×

465

  return code;

×

UNCOV

468

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

×

UNCOV

469

  pSupp->resLowSlot = -1;

×

470

  pSupp->resHighSlot = -1;

×

UNCOV

471

  pSupp->resTsSlot = -1;

×

UNCOV

472

  pSupp->resValSlot = -1;

×

474

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

×

475

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

×

UNCOV

476

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

×

477

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

×

478

      pSupp->resValSlot = dstSlot;

×

479

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

×

480

      pSupp->resTsSlot = dstSlot;

×

UNCOV

481

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

×

UNCOV

482

      pSupp->resLowSlot = dstSlot;

×

483

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

×

484

      pSupp->resHighSlot = dstSlot;

×

UNCOV

489

  return 0;

×

492

static int32_t validInputParams(SFunctionNode* pFunc, const char* id) {

×

493

  int32_t code = 0;

×

494

  int32_t num = LIST_LENGTH(pFunc->pParameterList);

×

496

  if (num <= 1) {

×

497

    qError("%s invalid number of parameters:%d", id, num);

×

498

    code = TSDB_CODE_PLAN_INTERNAL_ERROR;

×

499

    goto _end;

×

UNCOV

502

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

×

UNCOV

503

    SNode* p = nodesListGetNode(pFunc->pParameterList, i);

×

UNCOV

504

    if (p == NULL) {

×

505

      code = TSDB_CODE_PLAN_INTERNAL_ERROR;

×

506

      qError("%s %d-th parameter in forecast function is NULL, code:%s", id, i, tstrerror(code));

×

507

      goto _end;

×

UNCOV

511

  if (num == 2) {  // column_name, timestamp_column_name

×

512

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

×

513

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

×

515

    if (nodeType(p1) != QUERY_NODE_COLUMN || nodeType(p2) != QUERY_NODE_COLUMN) {

×

UNCOV

516

      qError("%s invalid column type, column 1:%d, column 2:%d", id, nodeType(p1), nodeType(p2));

×

UNCOV

517

      code = TSDB_CODE_PLAN_INTERNAL_ERROR;

×

UNCOV

518

      goto _end;

×

520

  } else if (num >= 3) {

×

UNCOV

524

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

×

UNCOV

525

    if (nodeType(pTarget) != QUERY_NODE_COLUMN) {

×

526

      code = TSDB_CODE_PLAN_INTERNAL_ERROR;

×

527

      qError("%s first parameter is not valid column in forecast function", id);

×

528

      goto _end;

×

531

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

×

UNCOV

532

    if (nodeType(pNode) != QUERY_NODE_COLUMN) {

×

UNCOV

533

      qError("%s last parameter is not valid column, actual:%d", id, nodeType(pNode));

×

UNCOV

534

      code = TSDB_CODE_PLAN_INTERNAL_ERROR;

×

538

_end:

×

539

  if (code) {

×

540

    qError("%s valid the parameters failed, code:%s", id, tstrerror(code));

×

542

  return code;

×

545

static bool existInList(SForecastSupp* pSupp, int32_t slotId) {

×

UNCOV

546

  for (int32_t j = 0; j < taosArrayGetSize(pSupp->pCovariateSlotList); ++j) {

×

547

    SColumn* pCol = taosArrayGet(pSupp->pCovariateSlotList, j);

×

549

    if (pCol->slotId == slotId) {

×

550

      return true;

×

554

  return false;

×

UNCOV

557

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

×

558

  int32_t code = 0;

×

UNCOV

559

  SNode*  pNode = NULL;

×

UNCOV

561

  pSupp->inputTsSlot = -1;

×

562

  pSupp->targetValSlot = -1;

×

563

  pSupp->targetValType = -1;

×

UNCOV

564

  pSupp->inputPrecision = -1;

×

566

  FOREACH(pNode, pFuncs) {

×

567

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

×

568

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

×

UNCOV

569

      int32_t        numOfParam = LIST_LENGTH(pFunc->pParameterList);

×

571

      if (pFunc->funcType == FUNCTION_TYPE_FORECAST) {

×

572

        code = validInputParams(pFunc, id);

×

573

        if (code) {

×

574

          return code;

×

577

        pSupp->numOfInputCols = 2;

×

579

        if (numOfParam == 2) {

×

581

          SColumnNode* pTarget = (SColumnNode*)nodesListGetNode(pFunc->pParameterList, 0);

×

582

          SColumnNode* pTsNode = (SColumnNode*)nodesListGetNode(pFunc->pParameterList, 1);

×

UNCOV

584

          pSupp->inputTsSlot = pTsNode->slotId;

×

585

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

×

586

          pSupp->targetValSlot = pTarget->slotId;

×

587

          pSupp->targetValType = pTarget->node.resType.type;

×

590

          pSupp->pOptions = taosStrdup("algo=holtwinters");

×

591

          if (pSupp->pOptions == NULL) {

×

592

            qError("%s failed to dup forecast option, code:%s", id, tstrerror(terrno));

×

593

            return terrno;

×

596

          SColumnNode* pTarget = (SColumnNode*)nodesListGetNode(pFunc->pParameterList, 0);

×

597

          bool         assignTs = false;

×

598

          bool         assignOpt = false;

×

600

          pSupp->targetValSlot = pTarget->slotId;

×

UNCOV

601

          pSupp->targetValType = pTarget->node.resType.type;

×

UNCOV

604

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

×

605

            SNode* pNode = nodesListGetNode(pFunc->pParameterList, i);

×

UNCOV

606

            if (nodeType(pNode) == QUERY_NODE_COLUMN) {

×

607

              SColumnNode* pColNode = (SColumnNode*)pNode;

×

608

              if (pColNode->isPrimTs && (!assignTs)) {

×

609

                pSupp->inputTsSlot = pColNode->slotId;

×

610

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

×

UNCOV

611

                assignTs = true;

×

UNCOV

612

                continue;

×

614

            } else if (nodeType(pNode) == QUERY_NODE_VALUE) {

×

615

              if (!assignOpt) {

×

UNCOV

616

                SValueNode* pOptNode = (SValueNode*)pNode;

×

UNCOV

617

                pSupp->pOptions = taosStrdup(pOptNode->literal);

×

618

                assignOpt = true;

×

619

                continue;

×

624

          if (!assignOpt) {

×

UNCOV

626

            pSupp->pOptions = taosStrdup("algo=holtwinters");

×

UNCOV

627

            if (pSupp->pOptions == NULL) {

×

628

              qError("%s failed to dup forecast option, code:%s", id, tstrerror(terrno));

×

629

              return terrno;

×

633

          pSupp->pCovariateSlotList = taosArrayInit(4, sizeof(SColumn));

×

634

          pSupp->pDynamicRealList = taosArrayInit(4, sizeof(SColFutureData));

×

UNCOV

637

          for (int32_t i = 1; i < numOfParam; ++i) {

×

638

            SColumnNode* p = (SColumnNode*)nodesListGetNode(pFunc->pParameterList, i);

×

639

            if ((nodeType(p) != QUERY_NODE_COLUMN) || (nodeType(p) == QUERY_NODE_COLUMN && p->isPrimTs)) {

×

UNCOV

643

            if (p->slotId == pSupp->targetValSlot) {

×

UNCOV

644

              continue; // duplicate the target column, ignore it

×

UNCOV

647

            bool exist = existInList(pSupp, p->slotId);

×

UNCOV

648

            if (exist) {

×

UNCOV

649

              continue;  // duplicate column, ignore it

×

652

            SColumn col = {.slotId = p->slotId,

×

UNCOV

653

                           .colType = p->colType,

×

UNCOV

654

                           .type = p->node.resType.type,

×

655

                           .bytes = p->node.resType.bytes};

×

657

            tstrncpy(col.name, p->colName, tListLen(col.name));

×

658

            void* pRet = taosArrayPush(pSupp->pCovariateSlotList, &col);

×

659

            if (pRet == NULL) {

×

660

              code = terrno;

×

661

              qError("failed to record the covariate slot index, since %s", tstrerror(code));

×

UNCOV

665

          pSupp->numOfInputCols += taosArrayGetSize(pSupp->pCovariateSlotList);

×

671

  return code;

×

UNCOV

674

static void initForecastOpt(SForecastSupp* pSupp) {

×

UNCOV

675

  pSupp->maxTs = 0;

×

UNCOV

676

  pSupp->minTs = INT64_MAX;

×

677

  pSupp->numOfRows = 0;

×

UNCOV

678

  pSupp->wncheck = ANALY_DEFAULT_WNCHECK;

×

UNCOV

679

  pSupp->forecastRows = ANALY_FORECAST_DEFAULT_ROWS;

×

UNCOV

680

  pSupp->conf = ANALY_FORECAST_DEFAULT_CONF;

×

681

  pSupp->setEvery = 0;

×

682

  pSupp->setStart = 0;

×

683

×

UNCOV

685

static int32_t filterNotSupportForecast(SForecastSupp* pSupp) {

×

686

  if (taosArrayGetSize(pSupp->pCovariateSlotList) > 0) {

×

UNCOV

687

    if (taosStrcasecmp(pSupp->algoName, "holtwinters") == 0) {

×

688

      return TSDB_CODE_ANA_NOT_SUPPORT_FORECAST;

×

689

    } else if (taosStrcasecmp(pSupp->algoName, "arima") == 0) {

×

690

      return TSDB_CODE_ANA_NOT_SUPPORT_FORECAST;

×

UNCOV

691

    } else if (taosStrcasecmp(pSupp->algoName, "timemoe-fc") == 0) {

×

UNCOV

692

      return TSDB_CODE_ANA_NOT_SUPPORT_FORECAST;

×

696

  return TSDB_CODE_SUCCESS;

×

700

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

×

701

  int32_t   code = 0;

×

702

  int32_t   lino = 0;

×

UNCOV

703

  SHashObj* pHashMap = NULL;

×

705

  initForecastOpt(pSupp);

×

707

  code = taosAnalyGetOpts(pSupp->pOptions, &pHashMap);

×

UNCOV

708

  if (code != TSDB_CODE_SUCCESS) {

×

709

    return code;

×

712

  if (taosHashGetSize(pHashMap) == 0) {

×

UNCOV

713

    code = TSDB_CODE_INVALID_PARA;

×

UNCOV

714

    qError("%s no valid options for forecast, failed to exec", id);

×

UNCOV

715

    return code;

×

UNCOV

718

  if (taosHashGetSize(pHashMap) == 0) {

×

UNCOV

719

    code = TSDB_CODE_INVALID_PARA;

×

720

    qError("%s no valid options for forecast, failed to exec", id);

×

721

    return code;

×

UNCOV

724

  code = taosAnalysisParseAlgo(pSupp->pOptions, pSupp->algoName, pSupp->algoUrl, ANALY_ALGO_TYPE_FORECAST,

×

726

  TSDB_CHECK_CODE(code, lino, _end);

×

728

  code = filterNotSupportForecast(pSupp);

×

UNCOV

729

  if (code) {

×

UNCOV

730

    qError("%s not support forecast model, %s", id, pSupp->algoName);

×

731

    TSDB_CHECK_CODE(code, lino, _end);

×

735

  pSupp->timeout = taosAnalysisParseTimout(pHashMap, id);

×

UNCOV

736

  pSupp->wncheck = taosAnalysisParseWncheck(pHashMap, id);

×

739

  char* pRows = taosHashGet(pHashMap, ALGO_OPT_FORECASTROWS_NAME, strlen(ALGO_OPT_FORECASTROWS_NAME));

×

740

  if (pRows != NULL) {

×

741

    int64_t v = 0;

×

UNCOV

742

    code = toInteger(pRows, taosHashGetValueSize(pRows), 10, &v);

×

UNCOV

744

    pSupp->forecastRows = v;

×

745

    qDebug("%s forecast rows:%"PRId64, id, pSupp->forecastRows);

×

747

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

×

UNCOV

750

  if (pSupp->forecastRows > ANALY_FORECAST_RES_MAX_ROWS) {

×

751

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

×

753

    code = TSDB_CODE_ANA_ANODE_TOO_MANY_ROWS;

×

UNCOV

754

    goto _end;

×

UNCOV

757

  if (pSupp->forecastRows <= 0) {

×

UNCOV

758

    qError("%s output rows should be greater than 0, input:%" PRId64, id, pSupp->forecastRows);

×

759

    code = TSDB_CODE_INVALID_PARA;

×

UNCOV

760

    goto _end;

×

764

  char* pConf = taosHashGet(pHashMap, ALGO_OPT_CONF_NAME, strlen(ALGO_OPT_CONF_NAME));

×

765

  if (pConf != NULL) {

×

766

    char*  endPtr = NULL;

×

767

    double v = taosStr2Double(pConf, &endPtr);

×

768

    pSupp->conf = v;

×

UNCOV

770

    if (v <= 0 || v > 1.0) {

×

UNCOV

771

      pSupp->conf = ANALY_FORECAST_DEFAULT_CONF;

×

UNCOV

772

      qWarn("%s valid conf range is (0, 1], user specified:%.2f out of range, set the default:%.2f", id, v,

×

775

      qDebug("%s forecast conf:%.2f", id, pSupp->conf);

×

778

    qDebug("%s forecast conf not found:%s, use default:%.2f", id, pSupp->pOptions, pSupp->conf);

×

782

  char* pStart = taosHashGet(pHashMap, ALGO_OPT_START_NAME, strlen(ALGO_OPT_START_NAME));

×

783

  if (pStart != NULL) {

×

784

    int64_t v = 0;

×

785

    code = toInteger(pStart, taosHashGetValueSize(pStart), 10, &v);

×

UNCOV

786

    pSupp->startTs = v;

×

UNCOV

787

    pSupp->setStart = 1;

×

UNCOV

788

    qDebug("%s forecast set start ts:%"PRId64, id, pSupp->startTs);

×

UNCOV

792

  char* pEvery = taosHashGet(pHashMap, ALGO_OPT_EVERY_NAME, strlen(ALGO_OPT_EVERY_NAME));

×

793

  if (pEvery != NULL) {

×

794

    int64_t v = 0;

×

795

    code = toInteger(pEvery, taosHashGetValueSize(pEvery), 10, &v);

×

796

    pSupp->every = v;

×

797

    pSupp->setEvery = 1;

×

UNCOV

798

    qDebug("%s forecast set every ts:%"PRId64, id, pSupp->every);

×

801

  if (pSupp->setEvery && pSupp->every <= 0) {

×

802

    qError("%s period should be greater than 0, user specified:%"PRId64, id, pSupp->every);

×

UNCOV

803

    code = TSDB_CODE_INVALID_PARA;

×

UNCOV

804

    goto _end;

×

UNCOV

808

  void*       pIter = NULL;

×

UNCOV

809

  size_t      keyLen = 0;

×

810

  const char* p = "dynamic_real_";

×

UNCOV

812

  while ((pIter = taosHashIterate(pHashMap, pIter))) {

×

813

    const char* pVal = pIter;

×

814

    char*       pKey = taosHashGetKey((void*)pVal, &keyLen);

×

815

    int32_t     idx = 0;

×

816

    char        nameBuf[512] = {0};

×

818

    if (strncmp(pKey, p, strlen(p)) == 0) {

×

820

      if (strncmp(&pKey[keyLen - 4], "_col", 4) == 0) {

×

UNCOV

821

        continue;

×

824

      int32_t ret = sscanf(pKey, "dynamic_real_%d", &idx);

×

825

      if (ret == 0) {

×

826

        continue;

×

UNCOV

829

      memcpy(nameBuf, pKey, keyLen);

×

830

      strncpy(&nameBuf[keyLen], "_col", strlen("_col"));

×

832

      void* pCol = taosHashGet(pHashMap, nameBuf, strlen(nameBuf));

×

833

      if (pCol == NULL) {

×

834

        char* pTmp = taosStrndupi(pKey, keyLen);

×

835

        qError("%s dynamic real column related:%s column name:%s not specified", id, pTmp, nameBuf);

×

UNCOV

837

        taosMemoryFree(pTmp);

×

UNCOV

838

        code = TSDB_CODE_INVALID_PARA;

×

839

        goto _end;

×

842

        SColFutureData d = {.pName = taosStrndupi(pCol, taosHashGetValueSize(pCol))};

×

843

        if (d.pName == NULL) {

×

UNCOV

844

          qError("%s failed to clone the future dynamic real column name:%s", id, (char*) pCol);

×

UNCOV

845

          code = terrno;

×

846

          goto _end;

×

849

        int32_t index = -1;

×

UNCOV

850

        for (int32_t i = 0; i < taosArrayGetSize(pSupp->pCovariateSlotList); ++i) {

×

851

          SColumn* pColx = taosArrayGet(pSupp->pCovariateSlotList, i);

×

852

          if (strcmp(pColx->name, d.pName) == 0) {

×

853

            index = i;

×

UNCOV

854

            break;

×

858

        if (index == -1) {

×

UNCOV

859

          qError("%s not found the required future dynamic real column:%s", id, d.pName);

×

860

          code = TSDB_CODE_INVALID_PARA;

×

UNCOV

861

          taosMemoryFree(d.pName);

×

862

          goto _end;

×

865

        SColumn* pColx = taosArrayGet(pSupp->pCovariateSlotList, index);

×

UNCOV

866

        d.data.info.slotId = pColx->slotId;

×

UNCOV

867

        d.data.info.type = pColx->type;

×

868

        d.data.info.bytes = pColx->bytes;

×

870

        int32_t len = taosHashGetValueSize((void*)pVal);

×

871

        char*   buf = taosStrndupi(pVal, len);

×

872

        int32_t unused = strdequote((char*)buf);

×

UNCOV

874

        int32_t num = 0;

×

UNCOV

875

        char**  pList = strsplit(buf, " ", &num);

×

876

        if (num != pSupp->forecastRows) {

×

877

          qError("%s the rows:%d of future dynamic real column data is not equalled to the forecasting rows:%" PRId64,

×

879

          code = TSDB_CODE_INVALID_PARA;

×

881

          taosMemoryFree(d.pName);

×

UNCOV

882

          taosMemoryFree(pList);

×

UNCOV

883

          taosMemoryFree(buf);

×

884

          goto _end;

×

887

        d.numOfRows = num;

×

889

        code = colInfoDataEnsureCapacity(&d.data, num, true);

×

890

        if (code != 0) {

×

UNCOV

891

          qError("%s failed to prepare buffer, code:%s", id, tstrerror(code));

×

892

          goto _end;

×

895

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

×

UNCOV

896

          char* ps = NULL;

×

897

          if (j == 0) {

×

898

            ps = strstr(pList[j], "[") + 1;

×

900

            ps = pList[j];

×

903

          code = 0;

×

905

          switch(pColx->type) {

×

UNCOV

906

            case TSDB_DATA_TYPE_TINYINT: {

×

907

              int8_t t1 = taosStr2Int8(ps, NULL, 10);

×

908

              code = colDataSetVal(&d.data, j, (const char*)&t1, false);

×

909

              break;

×

UNCOV

911

            case TSDB_DATA_TYPE_SMALLINT: {

×

912

              int16_t t1 = taosStr2Int16(ps, NULL, 10);

×

913

              code = colDataSetVal(&d.data, j, (const char*)&t1, false);

×

914

              break;

×

UNCOV

916

            case TSDB_DATA_TYPE_INT: {

×

917

              int32_t t1 = taosStr2Int32(ps, NULL, 10);

×

918

              code = colDataSetVal(&d.data, j, (const char*)&t1, false);

×

919

              break;

×

UNCOV

921

            case TSDB_DATA_TYPE_BIGINT: {

×

922

              int64_t t1 = taosStr2Int64(ps, NULL, 10);

×

923

              code = colDataSetVal(&d.data, j, (const char*)&t1, false);

×

924

              break;

×

UNCOV

926

            case TSDB_DATA_TYPE_FLOAT: {

×

927

              float t1 = taosStr2Float(ps, NULL);

×

928

              code = colDataSetVal(&d.data, j, (const char*)&t1, false);

×

929

              break;

×

UNCOV

931

            case TSDB_DATA_TYPE_DOUBLE: {

×

932

              double t1 = taosStr2Double(ps, NULL);

×

933

              code = colDataSetVal(&d.data, j, (const char*)&t1, false);

×

934

              break;

×

UNCOV

936

            case TSDB_DATA_TYPE_UTINYINT: {

×

UNCOV

937

              uint8_t t1 = taosStr2UInt8(ps, NULL, 10);

×

UNCOV

938

              code = colDataSetVal(&d.data, j, (const char*)&t1, false);

×

939

              break;

×

UNCOV

941

            case TSDB_DATA_TYPE_USMALLINT: {

×

UNCOV

942

              uint16_t t1 = taosStr2UInt16(ps, NULL, 10);

×

UNCOV

943

              code = colDataSetVal(&d.data, j, (const char*)&t1, false);

×

944

              break;

×

946

            case TSDB_DATA_TYPE_UINT: {

×

947

              uint32_t t1 = taosStr2UInt32(ps, NULL, 10);

×

UNCOV

948

              code = colDataSetVal(&d.data, j, (const char*)&t1, false);

×

UNCOV

949

              break;

×

951

            case TSDB_DATA_TYPE_UBIGINT: {

×

952

              uint64_t t1 = taosStr2UInt64(ps, NULL, 10);

×

953

              code = colDataSetVal(&d.data, j, (const char*)&t1, false);

×

954

              break;

×

UNCOV

958

          if (code != 0) {

×

UNCOV

959

            break;

×

UNCOV

963

        taosMemoryFree(pList);

×

UNCOV

964

        taosMemoryFree(buf);

×

965

        if (code != 0) {

×

966

          goto _end;

×

UNCOV

969

        void* noret = taosArrayPush(pSupp->pDynamicRealList, &d);

×

970

        if (noret == NULL) {

×

971

          qError("%s failed to add column info in dynamic real column info", id);

×

UNCOV

972

          code = terrno;

×

973

          goto _end;

×

979

_end:

×

UNCOV

980

  taosHashCleanup(pHashMap);

×

981

  return code;

×

984

static int32_t forecastCreateBuf(SForecastSupp* pSupp, const char* pId) {

×

985

  SAnalyticBuf* pBuf = &pSupp->analyBuf;

×

UNCOV

986

  int64_t       ts = taosGetTimestampNs();

×

987

  int32_t       index = 0;

×

989

  pBuf->bufType = ANALYTICS_BUF_TYPE_JSON_COL;

×

UNCOV

990

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

×

992

  int32_t numOfCols = taosArrayGetSize(pSupp->pCovariateSlotList) + 2;

×

994

  int32_t code = tsosAnalyBufOpen(pBuf, numOfCols, pId);

×

995

  if (code != 0) goto _OVER;

×

997

  code = taosAnalyBufWriteColMeta(pBuf, index++, TSDB_DATA_TYPE_TIMESTAMP, "ts");

×

998

  if (code != 0) goto _OVER;

×

UNCOV

1000

  code = taosAnalyBufWriteColMeta(pBuf, index++, pSupp->targetValType, "val");

×

1001

  if (code != 0) goto _OVER;

×

1003

  int32_t numOfDynamicReal = taosArrayGetSize(pSupp->pDynamicRealList);

×

UNCOV

1004

  int32_t numOfPastDynamicReal = taosArrayGetSize(pSupp->pCovariateSlotList);

×

UNCOV

1006

  if (numOfPastDynamicReal >= numOfDynamicReal) {

×

UNCOV

1007

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

×

1008

      SColFutureData* pData = taosArrayGet(pSupp->pDynamicRealList, i);

×

1010

      for(int32_t k = 0; k < taosArrayGetSize(pSupp->pCovariateSlotList); ++k) {

×

UNCOV

1011

        SColumn* pCol = taosArrayGet(pSupp->pCovariateSlotList, k);

×

1012

        if (strcmp(pCol->name, pData->pName) == 0) {

×

1013

          char name[128] = {0};

×

UNCOV

1014

          (void) tsnprintf(name, tListLen(name), "dynamic_real_%d", i + 1);

×

1015

          code = taosAnalyBufWriteColMeta(pBuf, index++, pCol->type, name);

×

1016

          if (code != 0) {

×

1017

            goto _OVER;

×

UNCOV

1020

          memcpy(&pData->col, pCol, sizeof(SColumn));

×

UNCOV

1021

          taosArrayRemove(pSupp->pCovariateSlotList, k);

×

1022

          break;

×

1027

    numOfPastDynamicReal = taosArrayGetSize(pSupp->pCovariateSlotList);

×

UNCOV

1028

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

×

UNCOV

1029

      SColumn* pCol = taosArrayGet(pSupp->pCovariateSlotList, j);

×

1031

      char name[128] = {0};

×

1032

      (void)tsnprintf(name, tListLen(name), "past_dynamic_real_%d", j + 1);

×

UNCOV

1034

      code = taosAnalyBufWriteColMeta(pBuf, index++, pCol->type, name);

×

1035

      if (code) {

×

UNCOV

1036

        goto _OVER;

×

1041

  code = taosAnalyBufWriteDataBegin(pBuf);

×

1042

  if (code != 0) goto _OVER;

×

UNCOV

1044

  for (int32_t i = 0; i < pSupp->numOfInputCols; ++i) {

×

1045

    code = taosAnalyBufWriteColBegin(pBuf, i);

×

UNCOV

1046

    if (code != 0) goto _OVER;

×

1049

_OVER:

×

1050

  if (code != 0) {

×

UNCOV

1051

    (void)taosAnalyBufClose(pBuf);

×

1052

    taosAnalyBufDestroy(pBuf);

×

1054

  return code;

×

1057

static int32_t resetForecastOperState(SOperatorInfo* pOper) {

×

1058

  int32_t code = 0, lino = 0;

×

UNCOV

1059

  SForecastOperatorInfo* pInfo = pOper->info;

×

1060

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

×

UNCOV

1061

  SForecastFuncPhysiNode* pForecastPhyNode = (SForecastFuncPhysiNode*)pOper->pPhyNode;

×

1062

  SExecTaskInfo* pTaskInfo = pOper->pTaskInfo;

×

1064

  pOper->status = OP_NOT_OPENED;

×

UNCOV

1066

  blockDataCleanup(pInfo->pRes);

×

UNCOV

1068

  taosArrayDestroy(pInfo->forecastSupp.pCovariateSlotList);

×

1069

  pInfo->forecastSupp.pCovariateSlotList = NULL;

×

1071

  taosAnalyBufDestroy(&pInfo->forecastSupp.analyBuf);

×

UNCOV

1073

  cleanupExprSupp(&pOper->exprSupp);

×

1074

  cleanupExprSupp(&pInfo->scalarSup);

×

1076

  int32_t                 numOfExprs = 0;

×

1077

  SExprInfo*              pExprInfo = NULL;

×

UNCOV

1079

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

×

1081

  TAOS_CHECK_EXIT(initExprSupp(&pOper->exprSupp, pExprInfo, numOfExprs, &pTaskInfo->storageAPI.functionStore));

×

1083

  TAOS_CHECK_EXIT(filterInitFromNode((SNode*)pForecastPhyNode->node.pConditions, &pOper->exprSupp.pFilterInfo, 0,

×

1086

  TAOS_CHECK_EXIT(forecastParseInput(&pInfo->forecastSupp, pForecastPhyNode->pFuncs, pId));

×

UNCOV

1088

  TAOS_CHECK_EXIT(forecastParseOutput(&pInfo->forecastSupp, &pOper->exprSupp));

×

UNCOV

1090

  TAOS_CHECK_EXIT(forecastParseOpt(&pInfo->forecastSupp, pId));

×

UNCOV

1091

  TAOS_CHECK_EXIT(forecastCreateBuf(&pInfo->forecastSupp, pId));

×

UNCOV

1093

  if (pForecastPhyNode->pExprs != NULL) {

×

UNCOV

1094

    int32_t    num = 0;

×

1095

    SExprInfo* pScalarExprInfo = NULL;

×

UNCOV

1096

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

×

1097

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

×

1100

  initResultSizeInfo(&pOper->resultInfo, 4096);

×

1102

_exit:

×

1104

  if (code) {

×

1105

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

×

1108

  return code;

×

1114

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

×

UNCOV

1116

  QRY_PARAM_CHECK(pOptrInfo);

×

1118

  int32_t                code = 0;

×

UNCOV

1119

  int32_t                lino = 0;

×

1120

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

×

1121

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

×

UNCOV

1122

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

×

1123

    code = terrno;

×

1124

    goto _error;

×

1127

  pOperator->pPhyNode = pPhyNode;

×

1129

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

×

1130

  SForecastSupp*          pSupp = &pInfo->forecastSupp;

×

UNCOV

1131

  SForecastFuncPhysiNode* pForecastPhyNode = (SForecastFuncPhysiNode*)pPhyNode;

×

UNCOV

1132

  SExprSupp*              pExprSup = &pOperator->exprSupp;

×

1133

  int32_t                 numOfExprs = 0;

×

1134

  SExprInfo*              pExprInfo = NULL;

×

UNCOV

1136

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

×

1137

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1139

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

×

1140

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1142

  if (pForecastPhyNode->pExprs != NULL) {

×

1143

    int32_t    num = 0;

×

1144

    SExprInfo* pScalarExprInfo = NULL;

×

UNCOV

1145

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

×

1146

    QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1148

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

×

1149

    QUERY_CHECK_CODE(code, lino, _error);

×

1152

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

×

UNCOV

1153

                            pTaskInfo->pStreamRuntimeInfo);

×

1154

  QUERY_CHECK_CODE(code, lino, _error);

×

1156

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

×

UNCOV

1157

  QUERY_CHECK_CODE(code, lino, _error);

×

1159

  code = forecastParseOutput(pSupp, pExprSup);

×

UNCOV

1160

  QUERY_CHECK_CODE(code, lino, _error);

×

1162

  code = forecastParseOpt(pSupp, pId);

×

UNCOV

1163

  QUERY_CHECK_CODE(code, lino, _error);

×

1165

  code = forecastCreateBuf(pSupp, pId);

×

UNCOV

1166

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1168

  initResultSizeInfo(&pOperator->resultInfo, 4096);

×

1170

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

×

UNCOV

1171

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

×

1173

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

×

UNCOV

1175

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

×

1178

  setOperatorResetStateFn(pOperator, resetForecastOperState);

×

UNCOV

1180

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

×

UNCOV

1181

  QUERY_CHECK_CODE(code, lino, _error);

×

1183

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

×

1184

  QUERY_CHECK_CODE(code, lino, _error);

×

1186

  *pOptrInfo = pOperator;

×

1188

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

×

1189

  return TSDB_CODE_SUCCESS;

×

1191

_error:

×

1192

  if (code != TSDB_CODE_SUCCESS) {

×

UNCOV

1193

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

×

1195

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

×

UNCOV

1196

  destroyOperatorAndDownstreams(pOperator, &downstream, 1);

×

1197

  pTaskInfo->code = code;

×

1198

  return code;

×

UNCOV

1201

static void destroyColFutureData(void* p) {

×

1202

  SColFutureData* pData = p;

×

1203

  taosMemoryFree(pData->pName);

×

1204

  colDataDestroy(&pData->data);

×

1205

×

UNCOV

1207

static void destroyForecastInfo(void* param) {

×

UNCOV

1208

  SForecastOperatorInfo* pInfo = (SForecastOperatorInfo*)param;

×

UNCOV

1210

  blockDataDestroy(pInfo->pRes);

×

UNCOV

1211

  pInfo->pRes = NULL;

×

UNCOV

1213

  taosArrayDestroy(pInfo->forecastSupp.pCovariateSlotList);

×

UNCOV

1214

  pInfo->forecastSupp.pCovariateSlotList = NULL;

×

UNCOV

1216

  taosArrayDestroyEx(pInfo->forecastSupp.pDynamicRealList, destroyColFutureData);

×

UNCOV

1217

  pInfo->forecastSupp.pDynamicRealList = NULL;

×

UNCOV

1219

  taosMemoryFree(pInfo->forecastSupp.pOptions);

×

UNCOV

1221

  cleanupExprSupp(&pInfo->scalarSup);

×

UNCOV

1222

  taosAnalyBufDestroy(&pInfo->forecastSupp.analyBuf);

×

UNCOV

1223

  taosMemoryFreeClear(param);

×

UNCOV

1224

×

taosdata / TDengine / #4856

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

taosdata / TDengine / #4856

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

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